# -*- test-case-name: twisted.mail.test.test_options -*- # Copyright (c) 2001-2004 Twisted Matrix Laboratories. # See LICENSE for details. """I am the support module for creating mail servers with 'mktap' """ import os import sys from twisted.mail import mail from twisted.mail import maildir from twisted.mail import relay from twisted.mail import relaymanager from twisted.mail import alias from twisted.python import usage from twisted.cred import checkers from twisted.application import internet class Options(usage.Options): synopsis = "Usage: mktap mail [options]" optParameters = [ ["pop3", "p", 8110, "Port to start the POP3 server on (0 to disable)."], ["pop3s", "S", 0, "Port to start the POP3-over-SSL server on (0 to disable)."], ["smtp", "s", 8025, "Port to start the SMTP server on (0 to disable)."], ["certificate", "c", None, "Certificate file to use for SSL connections"], ["relay", "R", None, "Relay messages according to their envelope 'To', using the given" "path as a queue directory."], ["hostname", "H", None, "The hostname by which to identify this server."], ] optFlags = [ ["esmtp", "E", "Use RFC 1425/1869 SMTP extensions"], ["disable-anonymous", None, "Disallow non-authenticated SMTP connections"], ] zsh_actions = {"hostname" : "_hosts"} longdesc = "This creates a mail.tap file that can be used by twistd." def __init__(self): usage.Options.__init__(self) self.service = mail.MailService() self.last_domain = None def opt_passwordfile(self, filename): """Specify a file containing username:password login info for authenticated ESMTP connections.""" ch = checkers.OnDiskUsernamePasswordDatabase(filename) self.service.smtpPortal.registerChecker(ch) opt_P = opt_passwordfile def opt_default(self): """Make the most recently specified domain the default domain.""" if self.last_domain: self.service.addDomain('', self.last_domain) else: raise usage.UsageError("Specify a domain before specifying using --default") opt_D = opt_default def opt_maildirdbmdomain(self, domain): """generate an SMTP/POP3 virtual domain which saves to \"path\" """ try: name, path = domain.split('=') except ValueError: raise usage.UsageError("Argument to --maildirdbmdomain must be of the form 'name=path'") self.last_domain = maildir.MaildirDirdbmDomain(self.service, os.path.abspath(path)) self.service.addDomain(name, self.last_domain) opt_d = opt_maildirdbmdomain def opt_user(self, user_pass): """add a user/password to the last specified domains """ try: user, password = user_pass.split('=', 1) except ValueError: raise usage.UsageError("Argument to --user must be of the form 'user=password'") if self.last_domain: self.last_domain.addUser(user, password) else: raise usage.UsageError("Specify a domain before specifying users") opt_u = opt_user def opt_bounce_to_postmaster(self): """undelivered mails are sent to the postmaster """ self.last_domain.postmaster = 1 opt_b = opt_bounce_to_postmaster def opt_aliases(self, filename): """Specify an aliases(5) file to use for this domain""" if self.last_domain is not None: if mail.IAliasableDomain.providedBy(self.last_domain): aliases = alias.loadAliasFile(self.service.domains, filename) self.last_domain.setAliasGroup(aliases) self.service.monitor.monitorFile( filename, AliasUpdater(self.service.domains, self.last_domain) ) else: raise usage.UsageError( "%s does not support alias files" % ( self.last_domain.__class__.__name__, ) ) else: raise usage.UsageError("Specify a domain before specifying aliases") opt_A = opt_aliases def postOptions(self): for f in ('pop3', 'smtp', 'pop3s'): try: self[f] = int(self[f]) if not (0 <= self[f] < 2 ** 16): raise ValueError except ValueError: raise usage.UsageError( 'Invalid port specified to --%s: %s' % (f, self[f]) ) if self['pop3s']: if not self['certificate']: raise usage.UsageError("Cannot specify --pop3s without " "--certificate") elif not os.path.exists(self['certificate']): raise usage.UsageError("Certificate file %r does not exist." % self['certificate']) if not self['disable-anonymous']: self.service.smtpPortal.registerChecker(checkers.AllowAnonymousAccess()) if not (self['pop3'] or self['smtp'] or self['pop3s']): raise usage.UsageError("You cannot disable all protocols") class AliasUpdater: def __init__(self, domains, domain): self.domains = domains self.domain = domain def __call__(self, new): self.domain.setAliasGroup(alias.loadAliasFile(self.domains, new)) def makeService(config): if config['esmtp']: rmType = relaymanager.SmartHostESMTPRelayingManager smtpFactory = config.service.getESMTPFactory else: rmType = relaymanager.SmartHostSMTPRelayingManager smtpFactory = config.service.getSMTPFactory if config['relay']: dir = config['relay'] if not os.path.isdir(dir): os.mkdir(dir) config.service.setQueue(relaymanager.Queue(dir)) default = relay.DomainQueuer(config.service) manager = rmType(config.service.queue) if config['esmtp']: manager.fArgs += (None, None) manager.fArgs += (config['hostname'],) helper = relaymanager.RelayStateHelper(manager, 1) helper.setServiceParent(config.service) config.service.domains.setDefaultDomain(default) ctx = None if config['certificate']: from twisted.mail.protocols import SSLContextFactory ctx = SSLContextFactory(config['certificate']) if config['pop3']: s = internet.TCPServer(config['pop3'], config.service.getPOP3Factory()) s.setServiceParent(config.service) if config['pop3s']: s = internet.SSLServer(config['pop3s'], config.service.getPOP3Factory(), ctx) s.setServiceParent(config.service) if config['smtp']: f = smtpFactory() f.context = ctx if config['hostname']: f.domain = config['hostname'] f.fArgs = (f.domain,) if config['esmtp']: f.fArgs = (None, None) + f.fArgs s = internet.TCPServer(config['smtp'], f) s.setServiceParent(config.service) return config.service