← Back to branch summary

~certify-web-dev/twisted/certify-trunk

« back to all changes in this revision

Viewing changes to twisted/mail/tap.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2007-01-17 14:52:35 UTC
  • mfrom: (1.1.5 upstream) (2.1.2 etch)
  • Revision ID: james.westby@ubuntu.com-20070117145235-btmig6qfmqfen0om
Tags: 2.5.0-0ubuntu1
New upstream version, compatible with python2.5.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/mail/tap.py
 
1
# -*- test-case-name: twisted.mail.test.test_options -*-
 
2
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
 
3
# See LICENSE for details.
 
4
 
 
5
 
 
6
"""I am the support module for creating mail servers with 'mktap'
 
7
"""
 
8
 
 
9
import os
 
10
import sys
 
11
 
 
12
from twisted.mail import mail
 
13
from twisted.mail import maildir
 
14
from twisted.mail import relay
 
15
from twisted.mail import relaymanager
 
16
from twisted.mail import alias
 
17
 
 
18
from twisted.python import usage
 
19
 
 
20
from twisted.cred import checkers
 
21
from twisted.application import internet
 
22
 
 
23
 
 
24
class Options(usage.Options):
 
25
    synopsis = "Usage: mktap mail [options]"
 
26
 
 
27
    optParameters = [
 
28
        ["pop3", "p", 8110, "Port to start the POP3 server on (0 to disable)."],
 
29
        ["pop3s", "S", 0, "Port to start the POP3-over-SSL server on (0 to disable)."],
 
30
        ["smtp", "s", 8025, "Port to start the SMTP server on (0 to disable)."],
 
31
        ["certificate", "c", None, "Certificate file to use for SSL connections"],
 
32
        ["relay", "R", None,
 
33
            "Relay messages according to their envelope 'To', using the given"
 
34
            "path as a queue directory."],
 
35
        ["hostname", "H", None, "The hostname by which to identify this server."],
 
36
    ]
 
37
 
 
38
    optFlags = [
 
39
        ["esmtp", "E", "Use RFC 1425/1869 SMTP extensions"],
 
40
        ["disable-anonymous", None, "Disallow non-authenticated SMTP connections"],
 
41
    ]
 
42
    zsh_actions = {"hostname" : "_hosts"}
 
43
 
 
44
    longdesc = "This creates a mail.tap file that can be used by twistd."
 
45
 
 
46
    def __init__(self):
 
47
        usage.Options.__init__(self)
 
48
        self.service = mail.MailService()
 
49
        self.last_domain = None
 
50
 
 
51
    def opt_passwordfile(self, filename):
 
52
        """Specify a file containing username:password login info for authenticated ESMTP connections."""
 
53
        ch = checkers.OnDiskUsernamePasswordDatabase(filename)
 
54
        self.service.smtpPortal.registerChecker(ch)
 
55
    opt_P = opt_passwordfile
 
56
 
 
57
    def opt_default(self):
 
58
        """Make the most recently specified domain the default domain."""
 
59
        if self.last_domain:
 
60
            self.service.addDomain('', self.last_domain)
 
61
        else:
 
62
            raise usage.UsageError("Specify a domain before specifying using --default")
 
63
    opt_D = opt_default
 
64
 
 
65
    def opt_maildirdbmdomain(self, domain):
 
66
        """generate an SMTP/POP3 virtual domain which saves to \"path\"
 
67
        """
 
68
        try:
 
69
            name, path = domain.split('=')
 
70
        except ValueError:
 
71
            raise usage.UsageError("Argument to --maildirdbmdomain must be of the form 'name=path'")
 
72
 
 
73
        self.last_domain = maildir.MaildirDirdbmDomain(self.service, os.path.abspath(path))
 
74
        self.service.addDomain(name, self.last_domain)
 
75
    opt_d = opt_maildirdbmdomain
 
76
 
 
77
    def opt_user(self, user_pass):
 
78
        """add a user/password to the last specified domains
 
79
        """
 
80
        try:
 
81
            user, password = user_pass.split('=', 1)
 
82
        except ValueError:
 
83
            raise usage.UsageError("Argument to --user must be of the form 'user=password'")
 
84
        if self.last_domain:
 
85
            self.last_domain.addUser(user, password)
 
86
        else:
 
87
            raise usage.UsageError("Specify a domain before specifying users")
 
88
    opt_u = opt_user
 
89
 
 
90
    def opt_bounce_to_postmaster(self):
 
91
        """undelivered mails are sent to the postmaster
 
92
        """
 
93
        self.last_domain.postmaster = 1
 
94
    opt_b = opt_bounce_to_postmaster
 
95
 
 
96
    def opt_aliases(self, filename):
 
97
        """Specify an aliases(5) file to use for this domain"""
 
98
        if self.last_domain is not None:
 
99
            if mail.IAliasableDomain.providedBy(self.last_domain):
 
100
                aliases = alias.loadAliasFile(self.service.domains, filename)
 
101
                self.last_domain.setAliasGroup(aliases)
 
102
                self.service.monitor.monitorFile(
 
103
                    filename,
 
104
                    AliasUpdater(self.service.domains, self.last_domain)
 
105
                )
 
106
            else:
 
107
                raise usage.UsageError(
 
108
                    "%s does not support alias files" % (
 
109
                        self.last_domain.__class__.__name__,
 
110
                    )
 
111
                )
 
112
        else:
 
113
            raise usage.UsageError("Specify a domain before specifying aliases")
 
114
    opt_A = opt_aliases
 
115
 
 
116
    def postOptions(self):
 
117
        for f in ('pop3', 'smtp', 'pop3s'):
 
118
            try:
 
119
                self[f] = int(self[f])
 
120
                if not (0 <= self[f] < 2 ** 16):
 
121
                    raise ValueError
 
122
            except ValueError:
 
123
                raise usage.UsageError(
 
124
                    'Invalid port specified to --%s: %s' % (f, self[f])
 
125
                )
 
126
        if self['pop3s']:
 
127
            if not self['certificate']:
 
128
                raise usage.UsageError("Cannot specify --pop3s without "
 
129
                                       "--certificate")
 
130
            elif not os.path.exists(self['certificate']):
 
131
                raise usage.UsageError("Certificate file %r does not exist."
 
132
                                       % self['certificate'])
 
133
 
 
134
        if not self['disable-anonymous']:
 
135
            self.service.smtpPortal.registerChecker(checkers.AllowAnonymousAccess())
 
136
 
 
137
        if not (self['pop3'] or self['smtp'] or self['pop3s']):
 
138
            raise usage.UsageError("You cannot disable all protocols")
 
139
 
 
140
class AliasUpdater:
 
141
    def __init__(self, domains, domain):
 
142
        self.domains = domains
 
143
        self.domain = domain
 
144
    def __call__(self, new):
 
145
        self.domain.setAliasGroup(alias.loadAliasFile(self.domains, new))
 
146
 
 
147
def makeService(config):
 
148
    if config['esmtp']:
 
149
        rmType = relaymanager.SmartHostESMTPRelayingManager
 
150
        smtpFactory = config.service.getESMTPFactory
 
151
    else:
 
152
        rmType = relaymanager.SmartHostSMTPRelayingManager
 
153
        smtpFactory = config.service.getSMTPFactory
 
154
 
 
155
    if config['relay']:
 
156
        dir = config['relay']
 
157
        if not os.path.isdir(dir):
 
158
            os.mkdir(dir)
 
159
 
 
160
        config.service.setQueue(relaymanager.Queue(dir))
 
161
        default = relay.DomainQueuer(config.service)
 
162
 
 
163
        manager = rmType(config.service.queue)
 
164
        if config['esmtp']:
 
165
            manager.fArgs += (None, None)
 
166
        manager.fArgs += (config['hostname'],)
 
167
 
 
168
        helper = relaymanager.RelayStateHelper(manager, 1)
 
169
        helper.setServiceParent(config.service)
 
170
        config.service.domains.setDefaultDomain(default)
 
171
 
 
172
    ctx = None
 
173
    if config['certificate']:
 
174
        from twisted.mail.protocols import SSLContextFactory
 
175
        ctx = SSLContextFactory(config['certificate'])
 
176
 
 
177
    if config['pop3']:
 
178
        s = internet.TCPServer(config['pop3'], config.service.getPOP3Factory())
 
179
        s.setServiceParent(config.service)
 
180
    if config['pop3s']:
 
181
        s = internet.SSLServer(config['pop3s'],
 
182
                               config.service.getPOP3Factory(), ctx)
 
183
        s.setServiceParent(config.service)
 
184
    if config['smtp']:
 
185
        f = smtpFactory()
 
186
        f.context = ctx
 
187
        if config['hostname']:
 
188
            f.domain = config['hostname']
 
189
            f.fArgs = (f.domain,)
 
190
        if config['esmtp']:
 
191
            f.fArgs = (None, None) + f.fArgs
 
192
        s = internet.TCPServer(config['smtp'], f)
 
193
        s.setServiceParent(config.service)
 
194
    return config.service