diff options
Diffstat (limited to 'src/mailman/queue')
| -rw-r--r-- | src/mailman/queue/__init__.py | 466 | ||||
| -rw-r--r-- | src/mailman/queue/archive.py | 91 | ||||
| -rw-r--r-- | src/mailman/queue/bounce.py | 316 | ||||
| -rw-r--r-- | src/mailman/queue/command.py | 214 | ||||
| -rw-r--r-- | src/mailman/queue/docs/OVERVIEW.txt | 78 | ||||
| -rw-r--r-- | src/mailman/queue/docs/archiver.txt | 34 | ||||
| -rw-r--r-- | src/mailman/queue/docs/command.txt | 170 | ||||
| -rw-r--r-- | src/mailman/queue/docs/incoming.txt | 200 | ||||
| -rw-r--r-- | src/mailman/queue/docs/lmtp.txt | 103 | ||||
| -rw-r--r-- | src/mailman/queue/docs/news.txt | 157 | ||||
| -rw-r--r-- | src/mailman/queue/docs/outgoing.txt | 75 | ||||
| -rw-r--r-- | src/mailman/queue/docs/runner.txt | 72 | ||||
| -rw-r--r-- | src/mailman/queue/docs/switchboard.txt | 182 | ||||
| -rw-r--r-- | src/mailman/queue/http.py | 73 | ||||
| -rw-r--r-- | src/mailman/queue/incoming.py | 43 | ||||
| -rw-r--r-- | src/mailman/queue/lmtp.py | 218 | ||||
| -rw-r--r-- | src/mailman/queue/maildir.py | 190 | ||||
| -rw-r--r-- | src/mailman/queue/news.py | 166 | ||||
| -rw-r--r-- | src/mailman/queue/outgoing.py | 130 | ||||
| -rw-r--r-- | src/mailman/queue/pipeline.py | 35 | ||||
| -rw-r--r-- | src/mailman/queue/retry.py | 37 | ||||
| -rw-r--r-- | src/mailman/queue/virgin.py | 39 |
22 files changed, 3089 insertions, 0 deletions
diff --git a/src/mailman/queue/__init__.py b/src/mailman/queue/__init__.py new file mode 100644 index 000000000..6094bda9e --- /dev/null +++ b/src/mailman/queue/__init__.py @@ -0,0 +1,466 @@ +# Copyright (C) 2001-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Queing and dequeuing message/metadata pickle files. + +Messages are represented as email.message.Message objects (or an instance ofa +subclass). Metadata is represented as a Python dictionary. For every +message/metadata pair in a queue, a single file containing two pickles is +written. First, the message is written to the pickle, then the metadata +dictionary is written. +""" + +from __future__ import absolute_import, unicode_literals + +__metaclass__ = type +__all__ = [ + 'Runner', + 'Switchboard', + ] + + +import os +import sys +import time +import email +import errno +import pickle +import cPickle +import hashlib +import logging +import marshal +import traceback + +from cStringIO import StringIO +from lazr.config import as_boolean, as_timedelta +from zope.interface import implements + +from mailman import Message +from mailman import i18n +from mailman.config import config +from mailman.interfaces.runner import IRunner +from mailman.interfaces.switchboard import ISwitchboard +from mailman.utilities.filesystem import makedirs +from mailman.utilities.string import expand + + +# 20 bytes of all bits set, maximum hashlib.sha.digest() value +shamax = 0xffffffffffffffffffffffffffffffffffffffffL + +# Small increment to add to time in case two entries have the same time. This +# prevents skipping one of two entries with the same time until the next pass. +DELTA = .0001 +DOT = '.' +# We count the number of times a file has been moved to .bak and recovered. +# In order to prevent loops and a message flood, when the count reaches this +# value, we move the file to the bad queue as a .psv. +MAX_BAK_COUNT = 3 + +elog = logging.getLogger('mailman.error') +dlog = logging.getLogger('mailman.debug') + + + +class Switchboard: + implements(ISwitchboard) + + @staticmethod + def initialize(): + """Initialize the global switchboards for input/output.""" + for conf in config.qrunner_configs: + name = conf.name.split('.')[-1] + assert name not in config.switchboards, ( + 'Duplicate qrunner name: {0}'.format(name)) + substitutions = config.paths + substitutions['name'] = name + path = expand(conf.path, substitutions) + config.switchboards[name] = Switchboard(path) + + def __init__(self, queue_directory, + slice=None, numslices=1, recover=False): + """Create a switchboard object. + + :param queue_directory: The queue directory. + :type queue_directory: str + :param slice: The slice number for this switchboard, or None. If not + None, it must be [0..`numslices`). + :type slice: int or None + :param numslices: The total number of slices to split this queue + directory into. It must be a power of 2. + :type numslices: int + :param recover: True if backup files should be recovered. + :type recover: bool + """ + assert (numslices & (numslices - 1)) == 0, ( + 'Not a power of 2: {0}'.format(numslices)) + self.queue_directory = queue_directory + # Create the directory if it doesn't yet exist. + makedirs(self.queue_directory, 0770) + # Fast track for no slices + self._lower = None + self._upper = None + # BAW: test performance and end-cases of this algorithm + if numslices <> 1: + self._lower = ((shamax + 1) * slice) / numslices + self._upper = (((shamax + 1) * (slice + 1)) / numslices) - 1 + if recover: + self.recover_backup_files() + + def enqueue(self, _msg, _metadata=None, **_kws): + """See `ISwitchboard`.""" + if _metadata is None: + _metadata = {} + # Calculate the SHA hexdigest of the message to get a unique base + # filename. We're also going to use the digest as a hash into the set + # of parallel qrunner processes. + data = _metadata.copy() + data.update(_kws) + listname = data.get('listname', '--nolist--') + # Get some data for the input to the sha hash. + now = time.time() + if data.get('_plaintext'): + protocol = 0 + msgsave = cPickle.dumps(str(_msg), protocol) + else: + protocol = pickle.HIGHEST_PROTOCOL + msgsave = cPickle.dumps(_msg, protocol) + # listname is unicode but the input to the hash function must be an + # 8-bit string (eventually, a bytes object). + hashfood = msgsave + listname.encode('utf-8') + repr(now) + # Encode the current time into the file name for FIFO sorting. The + # file name consists of two parts separated by a '+': the received + # time for this message (i.e. when it first showed up on this system) + # and the sha hex digest. + rcvtime = data.setdefault('received_time', now) + filebase = repr(rcvtime) + '+' + hashlib.sha1(hashfood).hexdigest() + filename = os.path.join(self.queue_directory, filebase + '.pck') + tmpfile = filename + '.tmp' + # Always add the metadata schema version number + data['version'] = config.QFILE_SCHEMA_VERSION + # Filter out volatile entries. Use .keys() so that we can mutate the + # dictionary during the iteration. + for k in data.keys(): + if k.startswith('_'): + del data[k] + # We have to tell the dequeue() method whether to parse the message + # object or not. + data['_parsemsg'] = (protocol == 0) + # Write to the pickle file the message object and metadata. + with open(tmpfile, 'w') as fp: + fp.write(msgsave) + cPickle.dump(data, fp, protocol) + fp.flush() + os.fsync(fp.fileno()) + os.rename(tmpfile, filename) + return filebase + + def dequeue(self, filebase): + """See `ISwitchboard`.""" + # Calculate the filename from the given filebase. + filename = os.path.join(self.queue_directory, filebase + '.pck') + backfile = os.path.join(self.queue_directory, filebase + '.bak') + # Read the message object and metadata. + with open(filename) as fp: + # Move the file to the backup file name for processing. If this + # process crashes uncleanly the .bak file will be used to + # re-instate the .pck file in order to try again. + os.rename(filename, backfile) + msg = cPickle.load(fp) + data = cPickle.load(fp) + if data.get('_parsemsg'): + # Calculate the original size of the text now so that we won't + # have to generate the message later when we do size restriction + # checking. + original_size = len(msg) + msg = email.message_from_string(msg, Message.Message) + msg.original_size = original_size + data['original_size'] = original_size + return msg, data + + def finish(self, filebase, preserve=False): + """See `ISwitchboard`.""" + bakfile = os.path.join(self.queue_directory, filebase + '.bak') + try: + if preserve: + bad_dir = config.switchboards['bad'].queue_directory + psvfile = os.path.join(bad_dir, filebase + '.psv') + os.rename(bakfile, psvfile) + else: + os.unlink(bakfile) + except EnvironmentError: + elog.exception( + 'Failed to unlink/preserve backup file: %s', bakfile) + + @property + def files(self): + """See `ISwitchboard`.""" + return self.get_files() + + def get_files(self, extension='.pck'): + """See `ISwitchboard`.""" + times = {} + lower = self._lower + upper = self._upper + for f in os.listdir(self.queue_directory): + # By ignoring anything that doesn't end in .pck, we ignore + # tempfiles and avoid a race condition. + filebase, ext = os.path.splitext(f) + if ext <> extension: + continue + when, digest = filebase.split('+', 1) + # Throw out any files which don't match our bitrange. BAW: test + # performance and end-cases of this algorithm. MAS: both + # comparisons need to be <= to get complete range. + if lower is None or (lower <= long(digest, 16) <= upper): + key = float(when) + while key in times: + key += DELTA + times[key] = filebase + # FIFO sort + return [times[key] for key in sorted(times)] + + def recover_backup_files(self): + """See `ISwitchboard`.""" + # Move all .bak files in our slice to .pck. It's impossible for both + # to exist at the same time, so the move is enough to ensure that our + # normal dequeuing process will handle them. We keep count in + # _bak_count in the metadata of the number of times we recover this + # file. When the count reaches MAX_BAK_COUNT, we move the .bak file + # to a .psv file in the bad queue. + for filebase in self.get_files('.bak'): + src = os.path.join(self.queue_directory, filebase + '.bak') + dst = os.path.join(self.queue_directory, filebase + '.pck') + fp = open(src, 'rb+') + try: + try: + msg = cPickle.load(fp) + data_pos = fp.tell() + data = cPickle.load(fp) + except Exception as error: + # If unpickling throws any exception, just log and + # preserve this entry + elog.error('Unpickling .bak exception: %s\n' + 'Preserving file: %s', error, filebase) + self.finish(filebase, preserve=True) + else: + data['_bak_count'] = data.get('_bak_count', 0) + 1 + fp.seek(data_pos) + if data.get('_parsemsg'): + protocol = 0 + else: + protocol = 1 + cPickle.dump(data, fp, protocol) + fp.truncate() + fp.flush() + os.fsync(fp.fileno()) + if data['_bak_count'] >= MAX_BAK_COUNT: + elog.error('.bak file max count, preserving file: %s', + filebase) + self.finish(filebase, preserve=True) + else: + os.rename(src, dst) + finally: + fp.close() + + + +class Runner: + implements(IRunner) + + def __init__(self, name, slice=None): + """Create a queue runner. + + :param slice: The slice number for this queue runner. This is passed + directly to the underlying `ISwitchboard` object. + :type slice: int or None + """ + # Grab the configuration section. + self.name = name + section = getattr(config, 'qrunner.' + name) + substitutions = config.paths + substitutions['name'] = name + self.queue_directory = expand(section.path, substitutions) + numslices = int(section.instances) + self.switchboard = Switchboard( + self.queue_directory, slice, numslices, True) + self.sleep_time = as_timedelta(section.sleep_time) + # sleep_time is a timedelta; turn it into a float for time.sleep(). + self.sleep_float = (86400 * self.sleep_time.days + + self.sleep_time.seconds + + self.sleep_time.microseconds / 1.0e6) + self.max_restarts = int(section.max_restarts) + self.start = as_boolean(section.start) + self._stop = False + + def __repr__(self): + return '<{0} at {1:#x}>'.format(self.__class__.__name__, id(self)) + + def stop(self): + """See `IRunner`.""" + self._stop = True + + def run(self): + """See `IRunner`.""" + # Start the main loop for this queue runner. + try: + while True: + # Once through the loop that processes all the files in the + # queue directory. + filecnt = self._one_iteration() + # Do the periodic work for the subclass. + self._do_periodic() + # If the stop flag is set, we're done. + if self._stop: + break + # Give the runner an opportunity to snooze for a while, but + # pass it the file count so it can decide whether to do more + # work now or not. + self._snooze(filecnt) + except KeyboardInterrupt: + pass + finally: + self._clean_up() + + def _one_iteration(self): + """See `IRunner`.""" + me = self.__class__.__name__ + dlog.debug('[%s] starting oneloop', me) + # List all the files in our queue directory. The switchboard is + # guaranteed to hand us the files in FIFO order. + files = self.switchboard.files + for filebase in files: + dlog.debug('[%s] processing filebase: %s', me, filebase) + try: + # Ask the switchboard for the message and metadata objects + # associated with this queue file. + msg, msgdata = self.switchboard.dequeue(filebase) + except Exception as error: + # This used to just catch email.Errors.MessageParseError, but + # other problems can occur in message parsing, e.g. + # ValueError, and exceptions can occur in unpickling too. We + # don't want the runner to die, so we just log and skip this + # entry, but preserve it for analysis. + self._log(error) + elog.error('Skipping and preserving unparseable message: %s', + filebase) + self.switchboard.finish(filebase, preserve=True) + config.db.abort() + continue + try: + dlog.debug('[%s] processing onefile', me) + self._process_one_file(msg, msgdata) + dlog.debug('[%s] finishing filebase: %s', me, filebase) + self.switchboard.finish(filebase) + except Exception as error: + # All runners that implement _dispose() must guarantee that + # exceptions are caught and dealt with properly. Still, there + # may be a bug in the infrastructure, and we do not want those + # to cause messages to be lost. Any uncaught exceptions will + # cause the message to be stored in the shunt queue for human + # intervention. + self._log(error) + # Put a marker in the metadata for unshunting. + msgdata['whichq'] = self.switchboard.queue_directory + # It is possible that shunting can throw an exception, e.g. a + # permissions problem or a MemoryError due to a really large + # message. Try to be graceful. + try: + shunt = config.switchboards['shunt'] + new_filebase = shunt.enqueue(msg, msgdata) + elog.error('SHUNTING: %s', new_filebase) + self.switchboard.finish(filebase) + except Exception as error: + # The message wasn't successfully shunted. Log the + # exception and try to preserve the original queue entry + # for possible analysis. + self._log(error) + elog.error( + 'SHUNTING FAILED, preserving original entry: %s', + filebase) + self.switchboard.finish(filebase, preserve=True) + config.db.abort() + # Other work we want to do each time through the loop. + dlog.debug('[%s] doing periodic', me) + self._do_periodic() + dlog.debug('[%s] checking short circuit', me) + if self._short_circuit(): + dlog.debug('[%s] short circuiting', me) + break + dlog.debug('[%s] commiting', me) + config.db.commit() + dlog.debug('[%s] ending oneloop: %s', me, len(files)) + return len(files) + + def _process_one_file(self, msg, msgdata): + """See `IRunner`.""" + # Do some common sanity checking on the message metadata. It's got to + # be destined for a particular mailing list. This switchboard is used + # to shunt off badly formatted messages. We don't want to just trash + # them because they may be fixable with human intervention. Just get + # them out of our sight. + # + # Find out which mailing list this message is destined for. + listname = unicode(msgdata.get('listname')) + mlist = config.db.list_manager.get(listname) + if mlist is None: + elog.error('Dequeuing message destined for missing list: %s', + listname) + config.switchboards['shunt'].enqueue(msg, msgdata) + return + # Now process this message. We also want to set up the language + # context for this message. The context will be the preferred + # language for the user if the sender is a member of the list, or it + # will be the list's preferred language. However, we must take + # special care to reset the defaults, otherwise subsequent messages + # may be translated incorrectly. + sender = msg.get_sender() + member = mlist.members.get_member(sender) + language = (member.preferred_language + if member is not None + else mlist.preferred_language) + with i18n.using_language(language): + msgdata['lang'] = language + keepqueued = self._dispose(mlist, msg, msgdata) + if keepqueued: + self.switchboard.enqueue(msg, msgdata) + + def _log(self, exc): + elog.error('Uncaught runner exception: %s', exc) + s = StringIO() + traceback.print_exc(file=s) + elog.error('%s', s.getvalue()) + + def _clean_up(self): + """See `IRunner`.""" + + def _dispose(self, mlist, msg, msgdata): + """See `IRunner`.""" + raise NotImplementedError + + def _do_periodic(self): + """See `IRunner`.""" + pass + + def _snooze(self, filecnt): + """See `IRunner`.""" + if filecnt or self.sleep_float <= 0: + return + time.sleep(self.sleep_float) + + def _short_circuit(self): + """See `IRunner`.""" + return self._stop diff --git a/src/mailman/queue/archive.py b/src/mailman/queue/archive.py new file mode 100644 index 000000000..75e8569e0 --- /dev/null +++ b/src/mailman/queue/archive.py @@ -0,0 +1,91 @@ +# Copyright (C) 2000-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Archive queue runner.""" + +__metaclass__ = type +__all__ = [ + 'ArchiveRunner', + ] + + +import os +import sys +import time +import logging + +from datetime import datetime +from email.Utils import parsedate_tz, mktime_tz, formatdate +from lazr.config import as_boolean, as_timedelta +from locknix.lockfile import Lock + +from mailman.config import config +from mailman.queue import Runner + +log = logging.getLogger('mailman.error') + + + +class ArchiveRunner(Runner): + """The archive runner.""" + + def _dispose(self, mlist, msg, msgdata): + # Support clobber_date, i.e. setting the date in the archive to the + # received date, not the (potentially bogus) Date: header of the + # original message. + clobber = False + original_date = msg.get('date') + received_time = formatdate(msgdata['received_time']) + if not original_date: + clobber = True + elif int(config.archiver.pipermail.clobber_date_policy) == 1: + clobber = True + elif int(config.archiver.pipermail.clobber_date_policy) == 2: + # What's the timestamp on the original message? + timetup = parsedate_tz(original_date) + now = datetime.now() + try: + if not timetup: + clobber = True + else: + utc_timestamp = datetime.fromtimestamp(mktime_tz(timetup)) + date_skew = as_timedelta( + config.archiver.pipermail.allowable_sane_date_skew) + clobber = (abs(now - utc_timestamp) > date_skew) + except (ValueError, OverflowError): + # The likely cause of this is that the year in the Date: field + # is horribly incorrect, e.g. (from SF bug # 571634): + # Date: Tue, 18 Jun 0102 05:12:09 +0500 + # Obviously clobber such dates. + clobber = True + if clobber: + del msg['date'] + del msg['x-original-date'] + msg['Date'] = received_time + if original_date: + msg['X-Original-Date'] = original_date + # Always put an indication of when we received the message. + msg['X-List-Received-Date'] = received_time + # While a list archiving lock is acquired, archive the message. + with Lock(os.path.join(mlist.data_path, 'archive.lck')): + for archiver in config.archivers: + # A problem in one archiver should not prevent other archivers + # from running. + try: + archiver.archive_message(mlist, msg) + except Exception: + log.exception('Broken archiver: %s' % archiver.name) diff --git a/src/mailman/queue/bounce.py b/src/mailman/queue/bounce.py new file mode 100644 index 000000000..ced731d6d --- /dev/null +++ b/src/mailman/queue/bounce.py @@ -0,0 +1,316 @@ +# Copyright (C) 2001-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Bounce queue runner.""" + +import os +import re +import cPickle +import logging +import datetime + +from email.Utils import parseaddr +from lazr.config import as_timedelta + +from mailman import Utils +from mailman.Bouncers import BouncerAPI +from mailman.config import config +from mailman.i18n import _ +from mailman.queue import Runner + +COMMASPACE = ', ' + +log = logging.getLogger('mailman.bounce') +elog = logging.getLogger('mailman.error') + + + +class BounceMixin: + def __init__(self): + # Registering a bounce means acquiring the list lock, and it would be + # too expensive to do this for each message. Instead, each bounce + # runner maintains an event log which is essentially a file with + # multiple pickles. Each bounce we receive gets appended to this file + # as a 4-tuple record: (listname, addr, today, msg) + # + # today is itself a 3-tuple of (year, month, day) + # + # Every once in a while (see _do_periodic()), the bounce runner cracks + # open the file, reads all the records and registers all the bounces. + # Then it truncates the file and continues on. We don't need to lock + # the bounce event file because bounce qrunners are single threaded + # and each creates a uniquely named file to contain the events. + # + # XXX When Python 2.3 is minimal require, we can use the new + # tempfile.TemporaryFile() function. + # + # XXX We used to classify bounces to the site list as bounce events + # for every list, but this caused severe problems. Here's the + # scenario: aperson@example.com is a member of 4 lists, and a list + # owner of the foo list. example.com has an aggressive spam filter + # which rejects any message that is spam or contains spam as an + # attachment. Now, a spambot sends a piece of spam to the foo list, + # but since that spambot is not a member, the list holds the message + # for approval, and sends a notification to aperson@example.com as + # list owner. That notification contains a copy of the spam. Now + # example.com rejects the message, causing a bounce to be sent to the + # site list's bounce address. The bounce runner would then dutifully + # register a bounce for all 4 lists that aperson@example.com was a + # member of, and eventually that person would get disabled on all + # their lists. So now we ignore site list bounces. Ce La Vie for + # password reminder bounces. + self._bounce_events_file = os.path.join( + config.DATA_DIR, 'bounce-events-%05d.pck' % os.getpid()) + self._bounce_events_fp = None + self._bouncecnt = 0 + self._nextaction = ( + datetime.datetime.now() + + as_timedelta(config.bounces.register_bounces_every)) + + def _queue_bounces(self, listname, addrs, msg): + today = datetime.date.today() + if self._bounce_events_fp is None: + self._bounce_events_fp = open(self._bounce_events_file, 'a+b') + for addr in addrs: + cPickle.dump((listname, addr, today, msg), + self._bounce_events_fp, 1) + self._bounce_events_fp.flush() + os.fsync(self._bounce_events_fp.fileno()) + self._bouncecnt += len(addrs) + + def _register_bounces(self): + log.info('%s processing %s queued bounces', self, self._bouncecnt) + # Read all the records from the bounce file, then unlink it. Sort the + # records by listname for more efficient processing. + events = {} + self._bounce_events_fp.seek(0) + while True: + try: + listname, addr, day, msg = cPickle.load(self._bounce_events_fp) + except ValueError, e: + log.error('Error reading bounce events: %s', e) + except EOFError: + break + events.setdefault(listname, []).append((addr, day, msg)) + # Now register all events sorted by list + for listname in events.keys(): + mlist = self._open_list(listname) + mlist.Lock() + try: + for addr, day, msg in events[listname]: + mlist.registerBounce(addr, msg, day=day) + mlist.Save() + finally: + mlist.Unlock() + # Reset and free all the cached memory + self._bounce_events_fp.close() + self._bounce_events_fp = None + os.unlink(self._bounce_events_file) + self._bouncecnt = 0 + + def _clean_up(self): + if self._bouncecnt > 0: + self._register_bounces() + + def _do_periodic(self): + now = datetime.datetime.now() + if self._nextaction > now or self._bouncecnt == 0: + return + # Let's go ahead and register the bounces we've got stored up + self._nextaction = now + as_timedelta( + config.bounces.register_bounces_every) + self._register_bounces() + + def _probe_bounce(self, mlist, token): + locked = mlist.Locked() + if not locked: + mlist.Lock() + try: + op, addr, bmsg = mlist.pend_confirm(token) + info = mlist.getBounceInfo(addr) + mlist.disableBouncingMember(addr, info, bmsg) + # Only save the list if we're unlocking it + if not locked: + mlist.Save() + finally: + if not locked: + mlist.Unlock() + + + +class BounceRunner(Runner, BounceMixin): + """The bounce runner.""" + + def __init__(self, slice=None, numslices=1): + Runner.__init__(self, slice, numslices) + BounceMixin.__init__(self) + + def _dispose(self, mlist, msg, msgdata): + # Make sure we have the most up-to-date state + mlist.Load() + # There are a few possibilities here: + # + # - the message could have been VERP'd in which case, we know exactly + # who the message was destined for. That make our job easy. + # - the message could have been originally destined for a list owner, + # but a list owner address itself bounced. That's bad, and for now + # we'll simply log the problem and attempt to deliver the message to + # the site owner. + # + # All messages sent to list owners have their sender set to the site + # owner address. That way, if a list owner address bounces, at least + # some human has a chance to deal with it. Is this a bounce for a + # message to a list owner, coming to the site owner? + if msg.get('to', '') == config.mailman.site_owner: + # Send it on to the site owners, but craft the envelope sender to + # be the noreply address, so if the site owner bounce, we won't + # get stuck in a bounce loop. + config.switchboards['out'].enqueue( + msg, msgdata, + recips=[config.mailman.site_owner], + envsender=config.mailman.noreply_address, + ) + # List isn't doing bounce processing? + if not mlist.bounce_processing: + return + # Try VERP detection first, since it's quick and easy + addrs = verp_bounce(mlist, msg) + if addrs: + # We have an address, but check if the message is non-fatal. + if BouncerAPI.ScanMessages(mlist, msg) is BouncerAPI.Stop: + return + else: + # See if this was a probe message. + token = verp_probe(mlist, msg) + if token: + self._probe_bounce(mlist, token) + return + # That didn't give us anything useful, so try the old fashion + # bounce matching modules. + addrs = BouncerAPI.ScanMessages(mlist, msg) + if addrs is BouncerAPI.Stop: + # This is a recognized, non-fatal notice. Ignore it. + return + # If that still didn't return us any useful addresses, then send it on + # or discard it. + if not addrs: + log.info('bounce message w/no discernable addresses: %s', + msg.get('message-id')) + maybe_forward(mlist, msg) + return + # BAW: It's possible that there are None's in the list of addresses, + # although I'm unsure how that could happen. Possibly ScanMessages() + # can let None's sneak through. In any event, this will kill them. + addrs = filter(None, addrs) + self._queue_bounces(mlist.fqdn_listname, addrs, msg) + + _do_periodic = BounceMixin._do_periodic + + def _clean_up(self): + BounceMixin._clean_up(self) + Runner._clean_up(self) + + + +def verp_bounce(mlist, msg): + bmailbox, bdomain = Utils.ParseEmail(mlist.GetBouncesEmail()) + # Sadly not every MTA bounces VERP messages correctly, or consistently. + # Fall back to Delivered-To: (Postfix), Envelope-To: (Exim) and + # Apparently-To:, and then short-circuit if we still don't have anything + # to work with. Note that there can be multiple Delivered-To: headers so + # we need to search them all (and we don't worry about false positives for + # forwarded email, because only one should match VERP_REGEXP). + vals = [] + for header in ('to', 'delivered-to', 'envelope-to', 'apparently-to'): + vals.extend(msg.get_all(header, [])) + for field in vals: + to = parseaddr(field)[1] + if not to: + continue # empty header + mo = re.search(config.mta.verp_regexp, to) + if not mo: + continue # no match of regexp + try: + if bmailbox <> mo.group('bounces'): + continue # not a bounce to our list + # All is good + addr = '%s@%s' % mo.group('mailbox', 'host') + except IndexError: + elog.error("verp_regexp doesn't yield the right match groups: %s", + config.mta.verp_regexp) + return [] + return [addr] + + + +def verp_probe(mlist, msg): + bmailbox, bdomain = Utils.ParseEmail(mlist.GetBouncesEmail()) + # Sadly not every MTA bounces VERP messages correctly, or consistently. + # Fall back to Delivered-To: (Postfix), Envelope-To: (Exim) and + # Apparently-To:, and then short-circuit if we still don't have anything + # to work with. Note that there can be multiple Delivered-To: headers so + # we need to search them all (and we don't worry about false positives for + # forwarded email, because only one should match VERP_REGEXP). + vals = [] + for header in ('to', 'delivered-to', 'envelope-to', 'apparently-to'): + vals.extend(msg.get_all(header, [])) + for field in vals: + to = parseaddr(field)[1] + if not to: + continue # empty header + mo = re.search(config.mta.verp_probe_regexp, to) + if not mo: + continue # no match of regexp + try: + if bmailbox <> mo.group('bounces'): + continue # not a bounce to our list + # Extract the token and see if there's an entry + token = mo.group('token') + data = mlist.pend_confirm(token, expunge=False) + if data is not None: + return token + except IndexError: + elog.error( + "verp_probe_regexp doesn't yield the right match groups: %s", + config.mta.verp_probe_regexp) + return None + + + +def maybe_forward(mlist, msg): + # Does the list owner want to get non-matching bounce messages? + # If not, simply discard it. + if mlist.bounce_unrecognized_goes_to_list_owner: + adminurl = mlist.GetScriptURL('admin', absolute=1) + '/bounce' + mlist.ForwardMessage(msg, + text=_("""\ +The attached message was received as a bounce, but either the bounce format +was not recognized, or no member addresses could be extracted from it. This +mailing list has been configured to send all unrecognized bounce messages to +the list administrator(s). + +For more information see: +%(adminurl)s + +"""), + subject=_('Uncaught bounce notification'), + tomoderators=0) + log.error('forwarding unrecognized, message-id: %s', + msg.get('message-id', 'n/a')) + else: + log.error('discarding unrecognized, message-id: %s', + msg.get('message-id', 'n/a')) diff --git a/src/mailman/queue/command.py b/src/mailman/queue/command.py new file mode 100644 index 000000000..d2be7c9fd --- /dev/null +++ b/src/mailman/queue/command.py @@ -0,0 +1,214 @@ +# Copyright (C) 1998-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""-request robot command queue runner.""" + +__metaclass__ = type +__all__ = [ + 'CommandRunner', + 'Results', + ] + +# See the delivery diagram in IncomingRunner.py. This module handles all +# email destined for mylist-request, -join, and -leave. It no longer handles +# bounce messages (i.e. -admin or -bounces), nor does it handle mail to +# -owner. + +import re +import logging + +from StringIO import StringIO +from email.Errors import HeaderParseError +from email.Header import decode_header, make_header +from email.Iterators import typed_subpart_iterator +from zope.interface import implements + +from mailman import Message +from mailman.config import config +from mailman.i18n import _ +from mailman.interfaces.command import ContinueProcessing, IEmailResults +from mailman.queue import Runner + +NL = '\n' + +log = logging.getLogger('mailman.vette') + + + +class CommandFinder: + """Generate commands from the content of a message.""" + + def __init__(self, msg, msgdata, results): + self.command_lines = [] + self.ignored_lines = [] + self.processed_lines = [] + # Depending on where the message was destined to, add some implicit + # commands. For example, if this was sent to the -join or -leave + # addresses, it's the same as if 'join' or 'leave' commands were sent + # to the -request address. + if msgdata.get('tojoin'): + self.command_lines.append('join') + elif msgdata.get('toleave'): + self.command_lines.append('leave') + elif msgdata.get('toconfirm'): + mo = re.match(config.mta.verp_confirm_regexp, msg.get('to', '')) + if mo: + self.command_lines.append('confirm ' + mo.group('cookie')) + # Extract the subject header and do RFC 2047 decoding. + raw_subject = msg.get('subject', '') + try: + subject = unicode(make_header(decode_header(raw_subject))) + # Mail commands must be ASCII. + self.command_lines.append(subject.encode('us-ascii')) + except (HeaderParseError, UnicodeError, LookupError): + # The Subject header was unparseable or not ASCII, so just ignore + # it. + pass + # Find the first text/plain part of the message. + part = None + for part in typed_subpart_iterator(msg, 'text', 'plain'): + break + if part is None or part is not msg: + # Either there was no text/plain part or we ignored some + # non-text/plain parts. + print >> results, _('Ignoring non-text/plain MIME parts') + if part is None: + # There was no text/plain part to be found. + return + body = part.get_payload(decode=True) + # text/plain parts better have string payloads. + assert isinstance(body, basestring), 'Non-string decoded payload' + lines = body.splitlines() + # Use no more lines than specified + max_lines = int(config.mailman.email_commands_max_lines) + self.command_lines.extend(lines[:max_lines]) + self.ignored_lines.extend(lines[max_lines:]) + + def __iter__(self): + """Return each command line, split into commands and arguments. + + :return: 2-tuples where the first element is the command and the + second element is a tuple of the arguments. + """ + while self.command_lines: + line = self.command_lines.pop(0) + self.processed_lines.append(line) + parts = line.strip().split() + if len(parts) == 0: + continue + command = parts.pop(0) + yield command, tuple(parts) + + + +class Results: + """The email command results.""" + + implements(IEmailResults) + + def __init__(self): + self._output = StringIO() + print >> self._output, _("""\ +The results of your email command are provided below. +""") + + def write(self, text): + self._output.write(text) + + def __unicode__(self): + value = self._output.getvalue() + assert isinstance(value, unicode), 'Not a unicode: %r' % value + return value + + + +class CommandRunner(Runner): + """The email command runner.""" + + def _dispose(self, mlist, msg, msgdata): + message_id = msg.get('message-id', 'n/a') + # The policy here is similar to the Replybot policy. If a message has + # "Precedence: bulk|junk|list" and no "X-Ack: yes" header, we discard + # the command message. + precedence = msg.get('precedence', '').lower() + ack = msg.get('x-ack', '').lower() + if ack <> 'yes' and precedence in ('bulk', 'junk', 'list'): + log.info('%s Precedence: %s message discarded by: %s', + message_id, precedence, mlist.request_address) + return False + # Do replybot for commands. + replybot = config.handlers['replybot'] + replybot.process(mlist, msg, msgdata) + if mlist.autorespond_requests == 1: + # Respond and discard. + log.info('%s -request message replied and discard', message_id) + return False + # Now craft the response and process the command lines. + results = Results() + # Include just a few key pieces of information from the original: the + # sender, date, and message id. + print >> results, _('- Original message details:') + subject = msg.get('subject', 'n/a') + date = msg.get('date', 'n/a') + from_ = msg.get('from', 'n/a') + print >> results, _(' From: $from_') + print >> results, _(' Subject: $subject') + print >> results, _(' Date: $date') + print >> results, _(' Message-ID: $message_id') + print >> results, _('\n- Results:') + finder = CommandFinder(msg, msgdata, results) + for command_name, arguments in finder: + command = config.commands.get(command_name) + if command is None: + print >> results, _('No such command: $command_name') + else: + status = command.process( + mlist, msg, msgdata, arguments, results) + assert status in ContinueProcessing, ( + 'Invalid status: %s' % status) + if status == ContinueProcessing.no: + break + # All done, send the response. + if len(finder.command_lines) > 0: + print >> results, _('\n- Unprocessed:') + for line in finder.command_lines: + print >> results, line + if len(finder.ignored_lines) > 0: + print >> results, _('\n- Ignored:') + for line in finder.ignored_lines: + print >> results, line + print >> results, _('\n- Done.') + # Send a reply, but do not attach the original message. This is a + # compromise because the original message is often helpful in tracking + # down problems, but it's also a vector for backscatter spam. + reply = Message.UserNotification( + msg.get_sender(), mlist.bounces_address, + _('The results of your email commands'), + lang=msgdata['lang']) + # Find a charset for the response body. Try ascii first, then + # latin-1 and finally falling back to utf-8. + reply_body = unicode(results) + for charset in ('us-ascii', 'latin-1'): + try: + reply_body.encode(charset) + break + except UnicodeError: + pass + else: + charset = 'utf-8' + reply.set_payload(reply_body, charset=charset) + reply.send(mlist) diff --git a/src/mailman/queue/docs/OVERVIEW.txt b/src/mailman/queue/docs/OVERVIEW.txt new file mode 100644 index 000000000..643fa8a5c --- /dev/null +++ b/src/mailman/queue/docs/OVERVIEW.txt @@ -0,0 +1,78 @@ +Alias overview +============== + +A typical Mailman list exposes nine aliases which point to seven different +wrapped scripts. E.g. for a list named `mylist', you'd have: + + mylist-bounces -> bounces + mylist-confirm -> confirm + mylist-join -> join (-subscribe is an alias) + mylist-leave -> leave (-unsubscribe is an alias) + mylist-owner -> owner + mylist -> post + mylist-request -> request + +-request, -join, and -leave are a robot addresses; their sole purpose is to +process emailed commands, although the latter two are hardcoded to +subscription and unsubscription requests. -bounces is the automated bounce +processor, and all messages to list members have their return address set to +-bounces. If the bounce processor fails to extract a bouncing member address, +it can optionally forward the message on to the list owners. + +-owner is for reaching a human operator with minimal list interaction (i.e. no +bounce processing). -confirm is another robot address which processes replies +to VERP-like confirmation notices. + +So delivery flow of messages look like this: + + joerandom ---> mylist ---> list members + | | + | |[bounces] + | mylist-bounces <---+ <-------------------------------+ + | | | + | +--->[internal bounce processing] | + | ^ | | + | | | [bounce found] | + | [bounces *] +--->[register and discard] | + | | | | | + | | | |[*] | + | [list owners] |[no bounce found] | | + | ^ | | | + | | | | | + +-------> mylist-owner <--------+ | | + | | | + | data/owner-bounces.mbox <--[site list] <---+ | + | | + +-------> mylist-join--+ | + | | | + +------> mylist-leave--+ | + | | | + | v | + +-------> mylist-request | + | | | + | +---> [command processor] | + | | | + +-----> mylist-confirm ----> +---> joerandom | + | | + |[bounces] | + +----------------------+ + +A person can send an email to the list address (for posting), the -owner +address (to reach the human operator), or the -confirm, -join, -leave, and +-request mailbots. Message to the list address are then forwarded on to the +list membership, with bounces directed to the -bounces address. + +[*] Messages sent to the -owner address are forwarded on to the list +owner/moderators. All -owner destined messages have their bounces directed to +the site list -bounces address, regardless of whether a human sent the message +or the message was crafted internally. The intention here is that the site +owners want to be notified when one of their list owners' addresses starts +bouncing (yes, the will be automated in a future release). + +Any messages to site owners has their bounces directed to a special +"loop-killer" address, which just dumps the message into +data/owners-bounces.mbox. + +Finally, message to any of the mailbots causes the requested action to be +performed. Results notifications are sent to the author of the message, which +all bounces pointing back to the -bounces address. diff --git a/src/mailman/queue/docs/archiver.txt b/src/mailman/queue/docs/archiver.txt new file mode 100644 index 000000000..601857cd9 --- /dev/null +++ b/src/mailman/queue/docs/archiver.txt @@ -0,0 +1,34 @@ +Archiving +========= + +Mailman can archive to any number of archivers that adhere to the IArchiver +interface. By default, there's a Pipermail archiver. + + >>> from mailman.app.lifecycle import create_list + >>> mlist = create_list(u'test@example.com') + >>> commit() + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: test@example.com + ... Subject: My first post + ... Message-ID: <first> + ... + ... First post! + ... """) + + >>> archiver_queue = config.switchboards['archive'] + >>> ignore = archiver_queue.enqueue(msg, {}, listname=mlist.fqdn_listname) + + >>> from mailman.queue.archive import ArchiveRunner + >>> from mailman.testing.helpers import make_testable_runner + >>> runner = make_testable_runner(ArchiveRunner) + >>> runner.run() + + # The best we can do is verify some landmark exists. Let's use the + # Pipermail pickle file exists. + >>> listname = mlist.fqdn_listname + >>> import os + >>> os.path.exists(os.path.join( + ... config.PUBLIC_ARCHIVE_FILE_DIR, listname, 'pipermail.pck')) + True diff --git a/src/mailman/queue/docs/command.txt b/src/mailman/queue/docs/command.txt new file mode 100644 index 000000000..0b384de01 --- /dev/null +++ b/src/mailman/queue/docs/command.txt @@ -0,0 +1,170 @@ +The command queue runner +======================== + +This queue runner's purpose is to process and respond to email commands. +Commands are extensible using the Mailman plugin system, but Mailman comes +with a number of email commands out of the box. These are processed when a +message is sent to the list's -request address. + + >>> from mailman.app.lifecycle import create_list + >>> mlist = create_list(u'test@example.com') + + +A command in the Subject +------------------------ + +For example, the 'echo' command simply echoes the original command back to the +sender. The command can be in the Subject header. + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: test-request@example.com + ... Subject: echo hello + ... Message-ID: <aardvark> + ... + ... """) + + >>> from mailman.inject import inject_message + >>> inject_message(mlist, msg, switchboard='command') + >>> from mailman.queue.command import CommandRunner + >>> from mailman.testing.helpers import make_testable_runner + >>> command = make_testable_runner(CommandRunner) + >>> command.run() + +And now the response is in the virgin queue. + + >>> from mailman.queue import Switchboard + >>> virgin_queue = config.switchboards['virgin'] + >>> len(virgin_queue.files) + 1 + >>> from mailman.testing.helpers import get_queue_messages + >>> item = get_queue_messages('virgin')[0] + >>> print item.msg.as_string() + Subject: The results of your email commands + From: test-bounces@example.com + To: aperson@example.com + ... + <BLANKLINE> + The results of your email command are provided below. + <BLANKLINE> + - Original message details: + From: aperson@example.com + Subject: echo hello + Date: ... + Message-ID: <aardvark> + <BLANKLINE> + - Results: + echo hello + <BLANKLINE> + - Done. + <BLANKLINE> + >>> sorted(item.msgdata.items()) + [..., ('listname', u'test@example.com'), ..., + ('recips', [u'aperson@example.com']), + ...] + + +A command in the body +--------------------- + +The command can also be found in the body of the message, as long as the +message is plain text. + + >>> msg = message_from_string("""\ + ... From: bperson@example.com + ... To: test-request@example.com + ... Message-ID: <bobcat> + ... + ... echo foo bar + ... """) + + >>> inject_message(mlist, msg, switchboard='command') + >>> command.run() + >>> len(virgin_queue.files) + 1 + >>> item = get_queue_messages('virgin')[0] + >>> print item.msg.as_string() + Subject: The results of your email commands + From: test-bounces@example.com + To: bperson@example.com + ... + Precedence: bulk + <BLANKLINE> + The results of your email command are provided below. + <BLANKLINE> + - Original message details: + From: bperson@example.com + Subject: n/a + Date: ... + Message-ID: <bobcat> + <BLANKLINE> + - Results: + echo foo bar + <BLANKLINE> + - Done. + <BLANKLINE> + + +Stopping command processing +--------------------------- + +The 'end' command stops email processing, so that nothing following is looked +at by the command queue. + + >>> msg = message_from_string("""\ + ... From: cperson@example.com + ... To: test-request@example.com + ... Message-ID: <caribou> + ... + ... echo foo bar + ... end ignored + ... echo baz qux + ... """) + + >>> inject_message(mlist, msg, switchboard='command') + >>> command.run() + >>> len(virgin_queue.files) + 1 + >>> item = get_queue_messages('virgin')[0] + >>> print item.msg.as_string() + Subject: The results of your email commands + ... + <BLANKLINE> + - Results: + echo foo bar + <BLANKLINE> + - Unprocessed: + echo baz qux + <BLANKLINE> + - Done. + <BLANKLINE> + +The 'stop' command is an alias for 'end'. + + >>> msg = message_from_string("""\ + ... From: cperson@example.com + ... To: test-request@example.com + ... Message-ID: <caribou> + ... + ... echo foo bar + ... stop ignored + ... echo baz qux + ... """) + + >>> inject_message(mlist, msg, switchboard='command') + >>> command.run() + >>> len(virgin_queue.files) + 1 + >>> item = get_queue_messages('virgin')[0] + >>> print item.msg.as_string() + Subject: The results of your email commands + ... + <BLANKLINE> + - Results: + echo foo bar + <BLANKLINE> + - Unprocessed: + echo baz qux + <BLANKLINE> + - Done. + <BLANKLINE> diff --git a/src/mailman/queue/docs/incoming.txt b/src/mailman/queue/docs/incoming.txt new file mode 100644 index 000000000..deb340e71 --- /dev/null +++ b/src/mailman/queue/docs/incoming.txt @@ -0,0 +1,200 @@ +The incoming queue runner +========================= + +This runner's sole purpose in life is to decide the disposition of the +message. It can either be accepted for delivery, rejected (i.e. bounced), +held for moderator approval, or discarded. + +The runner operates by processing chains on a message/metadata pair in the +context of a mailing list. Each mailing list may have a 'start chain' where +processing begins, with a global default. This chain is processed with the +message eventually ending up in one of the four disposition states described +above. + + >>> from mailman.app.lifecycle import create_list + >>> mlist = create_list(u'_xtest@example.com') + >>> mlist.start_chain + u'built-in' + + +Accepted messages +----------------- + +We have a message that is going to be sent to the mailing list. This message +is so perfectly fine for posting that it will be accepted and forward to the +pipeline queue. + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... Subject: My first post + ... Message-ID: <first> + ... + ... First post! + ... """) + +Normally, the upstream mail server would drop the message in the incoming +queue, but this is an effective simulation. + + >>> from mailman.inject import inject_message + >>> inject_message(mlist, msg) + +The incoming queue runner runs until it is empty. + + >>> from mailman.queue.incoming import IncomingRunner + >>> from mailman.testing.helpers import make_testable_runner + >>> incoming = make_testable_runner(IncomingRunner, 'in') + >>> incoming.run() + +And now the message is in the pipeline queue. + + >>> pipeline_queue = config.switchboards['pipeline'] + >>> len(pipeline_queue.files) + 1 + >>> incoming_queue = config.switchboards['in'] + >>> len(incoming_queue.files) + 0 + >>> from mailman.testing.helpers import get_queue_messages + >>> item = get_queue_messages('pipeline')[0] + >>> print item.msg.as_string() + From: aperson@example.com + To: _xtest@example.com + Subject: My first post + Message-ID: <first> + Date: ... + X-Mailman-Rule-Misses: approved; emergency; loop; administrivia; + implicit-dest; + max-recipients; max-size; news-moderation; no-subject; + suspicious-header + <BLANKLINE> + First post! + <BLANKLINE> + >>> sorted(item.msgdata.items()) + [...('envsender', u'noreply@example.com')...('tolist', True)...] + + +Held messages +------------- + +The list moderator sets the emergency flag on the mailing list. The built-in +chain will now hold all posted messages, so nothing will show up in the +pipeline queue. + + # XXX This checks the vette log file because there is no other evidence + # that this chain has done anything. + >>> import os + >>> fp = open(os.path.join(config.LOG_DIR, 'vette')) + >>> fp.seek(0, 2) + + >>> mlist.emergency = True + >>> inject_message(mlist, msg) + >>> file_pos = fp.tell() + >>> incoming.run() + >>> len(pipeline_queue.files) + 0 + >>> len(incoming_queue.files) + 0 + >>> fp.seek(file_pos) + >>> print 'LOG:', fp.read() + LOG: ... HOLD: _xtest@example.com post from aperson@example.com held, + message-id=<first>: n/a + <BLANKLINE> + + >>> mlist.emergency = False + + +Discarded messages +------------------ + +Another possibility is that the message would get immediately discarded. The +built-in chain does not have such a disposition by default, so let's craft a +new chain and set it as the mailing list's start chain. + + >>> from mailman.chains.base import Chain, Link + >>> from mailman.interfaces.chain import LinkAction + >>> truth_rule = config.rules['truth'] + >>> discard_chain = config.chains['discard'] + >>> test_chain = Chain('always-discard', u'Testing discards') + >>> link = Link(truth_rule, LinkAction.jump, discard_chain) + >>> test_chain.append_link(link) + >>> mlist.start_chain = u'always-discard' + + >>> inject_message(mlist, msg) + >>> file_pos = fp.tell() + >>> incoming.run() + >>> len(pipeline_queue.files) + 0 + >>> len(incoming_queue.files) + 0 + >>> fp.seek(file_pos) + >>> print 'LOG:', fp.read() + LOG: ... DISCARD: <first> + <BLANKLINE> + + >>> del config.chains['always-discard'] + + +Rejected messages +----------------- + +Similar to discarded messages, a message can be rejected, or bounced back to +the original sender. Again, the built-in chain doesn't support this so we'll +just create a new chain that does. + + >>> reject_chain = config.chains['reject'] + >>> test_chain = Chain('always-reject', u'Testing rejections') + >>> link = Link(truth_rule, LinkAction.jump, reject_chain) + >>> test_chain.append_link(link) + >>> mlist.start_chain = u'always-reject' + +The virgin queue needs to be cleared out due to artifacts from the previous +tests above. + + >>> virgin_queue = config.switchboards['virgin'] + >>> ignore = get_queue_messages('virgin') + + >>> inject_message(mlist, msg) + >>> file_pos = fp.tell() + >>> incoming.run() + >>> len(pipeline_queue.files) + 0 + >>> len(incoming_queue.files) + 0 + + >>> len(virgin_queue.files) + 1 + >>> item = get_queue_messages('virgin')[0] + >>> print item.msg.as_string() + Subject: My first post + From: _xtest-owner@example.com + To: aperson@example.com + ... + <BLANKLINE> + --===============... + Content-Type: text/plain; charset="us-ascii" + MIME-Version: 1.0 + Content-Transfer-Encoding: 7bit + <BLANKLINE> + [No bounce details are available] + --===============... + Content-Type: message/rfc822 + MIME-Version: 1.0 + <BLANKLINE> + From: aperson@example.com + To: _xtest@example.com + Subject: My first post + Message-ID: <first> + Date: ... + <BLANKLINE> + First post! + <BLANKLINE> + --===============... + + >>> sorted(item.msgdata.items()) + [...('recips', [u'aperson@example.com'])...] + >>> fp.seek(file_pos) + >>> print 'LOG:', fp.read() + LOG: ... REJECT: <first> + <BLANKLINE> + + >>> del config.chains['always-reject'] diff --git a/src/mailman/queue/docs/lmtp.txt b/src/mailman/queue/docs/lmtp.txt new file mode 100644 index 000000000..75e91fd4e --- /dev/null +++ b/src/mailman/queue/docs/lmtp.txt @@ -0,0 +1,103 @@ +LTMP server +=========== + +Mailman can accept messages via LMTP (RFC 2033). Most modern mail servers +support LMTP local delivery, so this is a very portable way to connect Mailman +with your mail server. + +Our LMTP server is fairly simple though; all it does is make sure that the +message is destined for a valid endpoint, e.g. mylist-join@example.com. + +Let's start a testable LMTP queue runner. + + >>> from mailman.testing import helpers + >>> master = helpers.TestableMaster() + >>> master.start('lmtp') + +It also helps to have a nice LMTP client. + + >>> lmtp = helpers.get_lmtp_client() + (220, '... Python LMTP queue runner 1.0') + >>> lmtp.lhlo('remote.example.org') + (250, ...) + + +Posting address +--------------- + +If the mail server tries to send a message to a nonexistent mailing list, it +will get a 550 error. + + >>> lmtp.sendmail( + ... 'anne.person@example.com', + ... ['mylist@example.com'], """\ + ... From: anne.person@example.com + ... To: mylist@example.com + ... Subject: An interesting message + ... Message-ID: <aardvark> + ... + ... This is an interesting message. + ... """) + Traceback (most recent call last): + ... + SMTPDataError: (550, 'Requested action not taken: mailbox unavailable') + +Once the mailing list is created, the posting address is valid. + + >>> from mailman.app.lifecycle import create_list + >>> create_list(u'mylist@example.com') + <mailing list "mylist@example.com" at ...> + >>> commit() + >>> lmtp.sendmail( + ... 'anne.person@example.com', + ... ['mylist@example.com'], """\ + ... From: anne.person@example.com + ... To: mylist@example.com + ... Subject: An interesting message + ... Message-ID: <badger> + ... + ... This is an interesting message. + ... """) + {} + + +Sub-addresses +------------- + +The LMTP server understands each of the list's sub-addreses, such as -join, +-leave, -request and so on. If the message is posted to an invalid +sub-address though, it is rejected. + + >>> lmtp.sendmail( + ... 'anne.person@example.com', + ... ['mylist-bogus@example.com'], """\ + ... From: anne.person@example.com + ... To: mylist-bogus@example.com + ... Subject: Help + ... Message-ID: <cow> + ... + ... Please help me. + ... """) + Traceback (most recent call last): + ... + SMTPDataError: (550, 'Requested action not taken: mailbox unavailable') + +But the message is accepted if posted to a valid sub-address. + + >>> lmtp.sendmail( + ... 'anne.person@example.com', + ... ['mylist-request@example.com'], """\ + ... From: anne.person@example.com + ... To: mylist-request@example.com + ... Subject: Help + ... Message-ID: <dog> + ... + ... Please help me. + ... """) + {} + + +Clean up +-------- + + >>> master.stop() diff --git a/src/mailman/queue/docs/news.txt b/src/mailman/queue/docs/news.txt new file mode 100644 index 000000000..3375b3d54 --- /dev/null +++ b/src/mailman/queue/docs/news.txt @@ -0,0 +1,157 @@ +The news runner +=============== + +The news runner is the queue runner that gateways mailing list messages to an +NNTP newsgroup. One of the most important things this runner does is prepare +the message for Usenet (yes, I know that NNTP is not Usenet, but this runner +was originally written to gate to Usenet, which has its own rules). + + >>> from mailman.queue.news import prepare_message + >>> mlist = config.db.list_manager.create(u'_xtest@example.com') + >>> mlist.linked_newsgroup = u'comp.lang.python' + +Some NNTP servers such as INN reject messages containing a set of prohibited +headers, so one of the things that the news runner does is remove these +prohibited headers. + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... NNTP-Posting-Host: news.example.com + ... NNTP-Posting-Date: today + ... X-Trace: blah blah + ... X-Complaints-To: abuse@dom.ain + ... Xref: blah blah + ... Xref: blah blah + ... Date-Received: yesterday + ... Posted: tomorrow + ... Posting-Version: 99.99 + ... Relay-Version: 88.88 + ... Received: blah blah + ... + ... A message + ... """) + >>> msgdata = {} + >>> prepare_message(mlist, msg, msgdata) + >>> msgdata['prepped'] + True + >>> print msg.as_string() + From: aperson@example.com + To: _xtest@example.com + Newsgroups: comp.lang.python + Message-ID: ... + Lines: 1 + <BLANKLINE> + A message + <BLANKLINE> + +Some NNTP servers will reject messages where certain headers are duplicated, +so the news runner must collapse or move these duplicate headers to an +X-Original-* header that the news server doesn't care about. + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... To: two@example.com + ... Cc: three@example.com + ... Cc: four@example.com + ... Cc: five@example.com + ... Content-Transfer-Encoding: yes + ... Content-Transfer-Encoding: no + ... Content-Transfer-Encoding: maybe + ... + ... A message + ... """) + >>> msgdata = {} + >>> prepare_message(mlist, msg, msgdata) + >>> msgdata['prepped'] + True + >>> print msg.as_string() + From: aperson@example.com + Newsgroups: comp.lang.python + Message-ID: ... + Lines: 1 + To: _xtest@example.com + X-Original-To: two@example.com + CC: three@example.com + X-Original-CC: four@example.com + X-Original-CC: five@example.com + Content-Transfer-Encoding: yes + X-Original-Content-Transfer-Encoding: no + X-Original-Content-Transfer-Encoding: maybe + <BLANKLINE> + A message + <BLANKLINE> + +But if no headers are duplicated, then the news runner doesn't need to modify +the message. + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... Cc: someother@example.com + ... Content-Transfer-Encoding: yes + ... + ... A message + ... """) + >>> msgdata = {} + >>> prepare_message(mlist, msg, msgdata) + >>> msgdata['prepped'] + True + >>> print msg.as_string() + From: aperson@example.com + To: _xtest@example.com + Cc: someother@example.com + Content-Transfer-Encoding: yes + Newsgroups: comp.lang.python + Message-ID: ... + Lines: 1 + <BLANKLINE> + A message + <BLANKLINE> + + +Newsgroup moderation +-------------------- + +When the newsgroup is moderated, an Approved: header with the list's posting +address is added for the benefit of the Usenet system. + + >>> from mailman.interfaces import NewsModeration + >>> mlist.news_moderation = NewsModeration.open_moderated + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... Approved: this gets deleted + ... + ... """) + >>> prepare_message(mlist, msg, {}) + >>> msg['approved'] + u'_xtest@example.com' + + >>> mlist.news_moderation = NewsModeration.moderated + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... Approved: this gets deleted + ... + ... """) + >>> prepare_message(mlist, msg, {}) + >>> msg['approved'] + u'_xtest@example.com' + +But if the newsgroup is not moderated, the Approved: header is not chnaged. + + >>> mlist.news_moderation = NewsModeration.none + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... Approved: this doesn't get deleted + ... + ... """) + >>> prepare_message(mlist, msg, {}) + >>> msg['approved'] + u"this doesn't get deleted" + + +XXX More of the NewsRunner should be tested. diff --git a/src/mailman/queue/docs/outgoing.txt b/src/mailman/queue/docs/outgoing.txt new file mode 100644 index 000000000..6722dee84 --- /dev/null +++ b/src/mailman/queue/docs/outgoing.txt @@ -0,0 +1,75 @@ +Outgoing queue runner +===================== + +The outgoing queue runner is the process that delivers messages to the +directly upstream SMTP server. It is this external SMTP server that performs +final delivery to the intended recipients. + +Messages that appear in the outgoing queue are processed individually through +a 'delivery module', essentially a pluggable interface for determining how the +recipient set will be batched, whether messages will be personalized and +VERP'd, etc. The outgoing runner doesn't itself support retrying but it can +move messages to the 'retry queue' for handling delivery failures. + + >>> from mailman.app.lifecycle import create_list + >>> mlist = create_list(u'test@example.com') + + >>> from mailman.app.membership import add_member + >>> from mailman.interfaces.member import DeliveryMode + >>> add_member(mlist, u'aperson@example.com', u'Anne Person', + ... u'password', DeliveryMode.regular, u'en') + >>> add_member(mlist, u'bperson@example.com', u'Bart Person', + ... u'password', DeliveryMode.regular, u'en') + >>> add_member(mlist, u'cperson@example.com', u'Cris Person', + ... u'password', DeliveryMode.regular, u'en') + +By setting the mailing list to personalize messages, each recipient will get a +unique copy of the message, with certain headers tailored for that recipient. + + >>> from mailman.interfaces.mailinglist import Personalization + >>> mlist.personalize = Personalization.individual + >>> commit() + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: test@example.com + ... Subject: My first post + ... Message-ID: <first> + ... + ... First post! + ... """) + +Normally, messages would show up in the outgoing queue after the message has +been processed by the rule set and pipeline. But we can simulate that here by +injecting a message directly into the outgoing queue. + + >>> msgdata = {} + >>> handler = config.handlers['calculate-recipients'] + >>> handler.process(mlist, msg, msgdata) + + >>> outgoing_queue = config.switchboards['out'] + >>> ignore = outgoing_queue.enqueue( + ... msg, msgdata, + ... verp=True, listname=mlist.fqdn_listname, tolist=True, + ... _plaintext=True) + +Running the outgoing queue runner processes the message, delivering it to the +upstream SMTP, which happens to be our test server. + + >>> from mailman.queue.outgoing import OutgoingRunner + >>> from mailman.testing.helpers import make_testable_runner + >>> outgoing = make_testable_runner(OutgoingRunner, 'out') + >>> outgoing.run() + +Three messages have been delivered to our SMTP server, one for each recipient. + + >>> from operator import itemgetter + >>> messages = sorted(smtpd.messages, key=itemgetter('sender')) + >>> len(messages) + 3 + + >>> for message in messages: + ... print message['sender'] + test-bounces+aperson=example.com@example.com + test-bounces+bperson=example.com@example.com + test-bounces+cperson=example.com@example.com diff --git a/src/mailman/queue/docs/runner.txt b/src/mailman/queue/docs/runner.txt new file mode 100644 index 000000000..d24a8334c --- /dev/null +++ b/src/mailman/queue/docs/runner.txt @@ -0,0 +1,72 @@ +Queue runners +============= + +The queue runners (qrunner) are the processes that move messages around the +Mailman system. Each qrunner is responsible for a slice of the hash space in +a queue directory. It processes all the files in its slice, sleeps a little +while, then wakes up and runs through its queue files again. + + +Basic architecture +------------------ + +The basic architecture of qrunner is implemented in the base class that all +runners inherit from. This base class implements a .run() method that runs +continuously in a loop until the .stop() method is called. + + >>> mlist = config.db.list_manager.create(u'_xtest@example.com') + >>> mlist.preferred_language = u'en' + +Here is a very simple derived qrunner class. Queue runners use a +configuration section in the configuration files to determine run +characteristics, such as the queue directory to use. Here we push a +configuration section for the test runner. + + >>> config.push('test-runner', """ + ... [qrunner.test] + ... max_restarts: 1 + ... """) + + >>> from mailman.queue import Runner + >>> class TestableRunner(Runner): + ... def _dispose(self, mlist, msg, msgdata): + ... self.msg = msg + ... self.msgdata = msgdata + ... return False + ... + ... def _do_periodic(self): + ... self.stop() + ... + ... def _snooze(self, filecnt): + ... return + + >>> runner = TestableRunner('test') + +This qrunner doesn't do much except run once, storing the message and metadata +on instance variables. + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... + ... A test message. + ... """) + >>> switchboard = config.switchboards['test'] + >>> filebase = switchboard.enqueue(msg, listname=mlist.fqdn_listname, + ... foo='yes', bar='no') + >>> runner.run() + >>> print runner.msg.as_string() + From: aperson@example.com + To: _xtest@example.com + <BLANKLINE> + A test message. + <BLANKLINE> + >>> dump_msgdata(runner.msgdata) + _parsemsg: False + bar : no + foo : yes + lang : en + listname : _xtest@example.com + version : 3 + +XXX More of the Runner API should be tested. diff --git a/src/mailman/queue/docs/switchboard.txt b/src/mailman/queue/docs/switchboard.txt new file mode 100644 index 000000000..88ab6ea93 --- /dev/null +++ b/src/mailman/queue/docs/switchboard.txt @@ -0,0 +1,182 @@ +The switchboard +=============== + +The switchboard is subsystem that moves messages between queues. Each +instance of a switchboard is responsible for one queue directory. + + >>> msg = message_from_string("""\ + ... From: aperson@example.com + ... To: _xtest@example.com + ... + ... A test message. + ... """) + +Create a switchboard by giving its queue directory. + + >>> import os + >>> queue_directory = os.path.join(config.QUEUE_DIR, 'test') + >>> from mailman.queue import Switchboard + >>> switchboard = Switchboard(queue_directory) + >>> switchboard.queue_directory == queue_directory + True + +Here's a helper function for ensuring things work correctly. + + >>> def check_qfiles(directory=None): + ... if directory is None: + ... directory = queue_directory + ... files = {} + ... for qfile in os.listdir(directory): + ... root, ext = os.path.splitext(qfile) + ... files[ext] = files.get(ext, 0) + 1 + ... if len(files) == 0: + ... print 'empty' + ... for ext in sorted(files): + ... print '{0}: {1}'.format(ext, files[ext]) + + +Enqueing and dequeing +--------------------- + +The message can be enqueued with metadata specified in the passed in +dictionary. + + >>> filebase = switchboard.enqueue(msg) + >>> check_qfiles() + .pck: 1 + +To read the contents of a queue file, dequeue it. + + >>> msg, msgdata = switchboard.dequeue(filebase) + >>> print msg.as_string() + From: aperson@example.com + To: _xtest@example.com + <BLANKLINE> + A test message. + <BLANKLINE> + >>> dump_msgdata(msgdata) + _parsemsg: False + version : 3 + >>> check_qfiles() + .bak: 1 + +To complete the dequeing process, removing all traces of the message file, +finish it (without preservation). + + >>> switchboard.finish(filebase) + >>> check_qfiles() + empty + +When enqueing a file, you can provide additional metadata keys by using +keyword arguments. + + >>> filebase = switchboard.enqueue(msg, {'foo': 1}, bar=2) + >>> msg, msgdata = switchboard.dequeue(filebase) + >>> switchboard.finish(filebase) + >>> dump_msgdata(msgdata) + _parsemsg: False + bar : 2 + foo : 1 + version : 3 + +Keyword arguments override keys from the metadata dictionary. + + >>> filebase = switchboard.enqueue(msg, {'foo': 1}, foo=2) + >>> msg, msgdata = switchboard.dequeue(filebase) + >>> switchboard.finish(filebase) + >>> dump_msgdata(msgdata) + _parsemsg: False + foo : 2 + version : 3 + + +Iterating over files +-------------------- + +There are two ways to iterate over all the files in a switchboard's queue. +Normally, queue files end in .pck (for 'pickle') and the easiest way to +iterate over just these files is to use the .files attribute. + + >>> filebase_1 = switchboard.enqueue(msg, foo=1) + >>> filebase_2 = switchboard.enqueue(msg, foo=2) + >>> filebase_3 = switchboard.enqueue(msg, foo=3) + >>> filebases = sorted((filebase_1, filebase_2, filebase_3)) + >>> sorted(switchboard.files) == filebases + True + >>> check_qfiles() + .pck: 3 + +You can also use the .get_files() method if you want to iterate over all the +file bases for some other extension. + + >>> for filebase in switchboard.get_files(): + ... msg, msgdata = switchboard.dequeue(filebase) + >>> bakfiles = sorted(switchboard.get_files('.bak')) + >>> bakfiles == filebases + True + >>> check_qfiles() + .bak: 3 + >>> for filebase in switchboard.get_files('.bak'): + ... switchboard.finish(filebase) + >>> check_qfiles() + empty + + +Recovering files +---------------- + +Calling .dequeue() without calling .finish() leaves .bak backup files in +place. These can be recovered when the switchboard is instantiated. + + >>> filebase_1 = switchboard.enqueue(msg, foo=1) + >>> filebase_2 = switchboard.enqueue(msg, foo=2) + >>> filebase_3 = switchboard.enqueue(msg, foo=3) + >>> for filebase in switchboard.files: + ... msg, msgdata = switchboard.dequeue(filebase) + ... # Don't call .finish() + >>> check_qfiles() + .bak: 3 + >>> switchboard_2 = Switchboard(queue_directory, recover=True) + >>> check_qfiles() + .pck: 3 + +The files can be recovered explicitly. + + >>> for filebase in switchboard.files: + ... msg, msgdata = switchboard.dequeue(filebase) + ... # Don't call .finish() + >>> check_qfiles() + .bak: 3 + >>> switchboard.recover_backup_files() + >>> check_qfiles() + .pck: 3 + +But the files will only be recovered at most three times before they are +considered defective. In order to prevent mail bombs and loops, once this +maximum is reached, the files will be preserved in the 'bad' queue. + + >>> for filebase in switchboard.files: + ... msg, msgdata = switchboard.dequeue(filebase) + ... # Don't call .finish() + >>> check_qfiles() + .bak: 3 + >>> switchboard.recover_backup_files() + >>> check_qfiles() + empty + + >>> bad = config.switchboards['bad'] + >>> check_qfiles(bad.queue_directory) + .psv: 3 + +Clean up + + >>> for file in os.listdir(bad.queue_directory): + ... os.remove(os.path.join(bad.queue_directory, file)) + >>> check_qfiles(bad.queue_directory) + empty + + +Queue slices +------------ + +XXX Add tests for queue slices. diff --git a/src/mailman/queue/http.py b/src/mailman/queue/http.py new file mode 100644 index 000000000..941b6d131 --- /dev/null +++ b/src/mailman/queue/http.py @@ -0,0 +1,73 @@ +# Copyright (C) 2006-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Mailman HTTP runner (server).""" + +import sys +import signal +import logging + +from cStringIO import StringIO +from wsgiref.simple_server import make_server, WSGIRequestHandler + +from mailman.Cgi.wsgi_app import mailman_app +from mailman.config import config +from mailman.queue import Runner + +hlog = logging.getLogger('mailman.http') +qlog = logging.getLogger('mailman.qrunner') + + + +class HTTPRunner(Runner): + def __init__(self, slice=None, numslices=1): + pass + + def _clean_up(self): + pass + + + +class MailmanWSGIRequestHandler(WSGIRequestHandler): + def handle(self): + """Handle a single HTTP request with error output to elog""" + stderr = StringIO() + saved_stderr = sys.stderr + sys.stderr = stderr + try: + WSGIRequestHandler.handle(self) + finally: + sys.stderr = saved_stderr + hlog.info(stderr.getvalue().strip()) + + + +server = make_server(config.HTTP_HOST, config.HTTP_PORT, + mailman_app, + handler_class=MailmanWSGIRequestHandler) + + +qlog.info('HTTPRunner qrunner started.') +hlog.info('HTTPRunner listening on %s:%s', config.HTTP_HOST, config.HTTP_PORT) +try: + server.serve_forever() +except KeyboardInterrupt: + qlog.exception('HTTPRunner qrunner exiting.') + sys.exit(signal.SIGTERM) +except: + qlog.exception('HTTPRunner qrunner exiting.') + raise diff --git a/src/mailman/queue/incoming.py b/src/mailman/queue/incoming.py new file mode 100644 index 000000000..1adda6629 --- /dev/null +++ b/src/mailman/queue/incoming.py @@ -0,0 +1,43 @@ +# Copyright (C) 1998-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Incoming queue runner. + +This runner's sole purpose in life is to decide the disposition of the +message. It can either be accepted for delivery, rejected (i.e. bounced), +held for moderator approval, or discarded. + +When accepted, the message is forwarded on to the `prep queue` where it is +prepared for delivery. Rejections, discards, and holds are processed +immediately. +""" + +from mailman.core.chains import process +from mailman.queue import Runner + + + +class IncomingRunner(Runner): + """The incoming queue runner.""" + + def _dispose(self, mlist, msg, msgdata): + if msgdata.get('envsender') is None: + msgdata['envsender'] = mlist.no_reply_address + # Process the message through the mailing list's start chain. + process(mlist, msg, msgdata, mlist.start_chain) + # Do not keep this message queued. + return False diff --git a/src/mailman/queue/lmtp.py b/src/mailman/queue/lmtp.py new file mode 100644 index 000000000..3ac8796ca --- /dev/null +++ b/src/mailman/queue/lmtp.py @@ -0,0 +1,218 @@ +# Copyright (C) 2006-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Mailman LMTP runner (server). + +Most mail servers can be configured to deliver local messages via 'LMTP'[1]. +This module is actually an LMTP server rather than a standard queue runner. + +The LMTP runner opens a local TCP port and waits for the mail server to +connect to it. The messages it receives over LMTP are very minimally parsed +for sanity and if they look okay, they are accepted and injected into +Mailman's incoming queue for normal processing. If they don't look good, or +are destined for a bogus sub-address, they are rejected right away, hopefully +so that the peer mail server can provide better diagnostics. + +[1] RFC 2033 Local Mail Transport Protocol + http://www.faqs.org/rfcs/rfc2033.html +""" + +import email +import smtpd +import logging +import asyncore + +from email.utils import parseaddr + +from mailman.Message import Message +from mailman.config import config +from mailman.database.transaction import txn +from mailman.queue import Runner + +elog = logging.getLogger('mailman.error') +qlog = logging.getLogger('mailman.qrunner') + + +# We only care about the listname and the sub-addresses as in listname@ or +# listname-request@ +SUBADDRESS_NAMES = ( + 'bounces', 'confirm', 'join', ' leave', + 'owner', 'request', 'subscribe', 'unsubscribe', + ) + +DASH = '-' +CRLF = '\r\n' +ERR_451 = '451 Requested action aborted: error in processing' +ERR_501 = '501 Message has defects' +ERR_502 = '502 Error: command HELO not implemented' +ERR_550 = '550 Requested action not taken: mailbox unavailable' + +# XXX Blech +smtpd.__version__ = 'Python LMTP queue runner 1.0' + + + +def split_recipient(address): + """Split an address into listname, subaddress and domain parts. + + For example: + + >>> split_recipient('mylist@example.com') + ('mylist', None, 'example.com') + + >>> split_recipient('mylist-request@example.com') + ('mylist', 'request', 'example.com') + + :param address: The destination address. + :return: A 3-tuple of the form (list-shortname, subaddress, domain). + subaddress may be None if this is the list's posting address. + """ + localpart, domain = address.split('@', 1) + localpart = localpart.split(config.mta.verp_delimiter, 1)[0] + parts = localpart.split(DASH) + if parts[-1] in SUBADDRESS_NAMES: + listname = DASH.join(parts[:-1]) + subaddress = parts[-1] + else: + listname = localpart + subaddress = None + return listname, subaddress, domain + + + +class Channel(smtpd.SMTPChannel): + """An LMTP channel.""" + + def __init__(self, server, conn, addr): + smtpd.SMTPChannel.__init__(self, server, conn, addr) + # Stash this here since the subclass uses private attributes. :( + self._server = server + + def smtp_LHLO(self, arg): + """The LMTP greeting, used instead of HELO/EHLO.""" + smtpd.SMTPChannel.smtp_HELO(self, arg) + + def smtp_HELO(self, arg): + """HELO is not a valid LMTP command.""" + self.push(ERR_502) + + + +class LMTPRunner(Runner, smtpd.SMTPServer): + # Only __init__ is called on startup. Asyncore is responsible for later + # connections from the MTA. slice and numslices are ignored and are + # necessary only to satisfy the API. + def __init__(self, slice=None, numslices=1): + localaddr = config.mta.lmtp_host, int(config.mta.lmtp_port) + # Do not call Runner's constructor because there's no QDIR to create + smtpd.SMTPServer.__init__(self, localaddr, remoteaddr=None) + qlog.debug('LMTP server listening on %s:%s', + localaddr[0], localaddr[1]) + + def handle_accept(self): + conn, addr = self.accept() + channel = Channel(self, conn, addr) + qlog.debug('LMTP accept from %s', addr) + + @txn + def process_message(self, peer, mailfrom, rcpttos, data): + try: + # Refresh the list of list names every time we process a message + # since the set of mailing lists could have changed. + listnames = set(config.db.list_manager.names) + qlog.debug('listnames: %s', listnames) + # Parse the message data. If there are any defects in the + # message, reject it right away; it's probably spam. + msg = email.message_from_string(data, Message) + msg.original_size = len(data) + if msg.defects: + return ERR_501 + msg['X-MailFrom'] = mailfrom + except Exception, e: + elog.exception('LMTP message parsing') + config.db.abort() + return CRLF.join(ERR_451 for to in rcpttos) + # RFC 2033 requires us to return a status code for every recipient. + status = [] + # Now for each address in the recipients, parse the address to first + # see if it's destined for a valid mailing list. If so, then queue + # the message to the appropriate place and record a 250 status for + # that recipient. If not, record a failure status for that recipient. + for to in rcpttos: + try: + to = parseaddr(to)[1].lower() + listname, subaddress, domain = split_recipient(to) + qlog.debug('to: %s, list: %s, sub: %s, dom: %s', + to, listname, subaddress, domain) + listname += '@' + domain + if listname not in listnames: + status.append(ERR_550) + continue + # The recipient is a valid mailing list; see if it's a valid + # sub-address, and if so, enqueue it. + queue = None + msgdata = dict(listname=listname, + original_size=msg.original_size) + if subaddress in ('bounces', 'admin'): + queue = 'bounce' + elif subaddress == 'confirm': + msgdata['toconfirm'] = True + queue = 'command' + elif subaddress in ('join', 'subscribe'): + msgdata['tojoin'] = True + queue = 'command' + elif subaddress in ('leave', 'unsubscribe'): + msgdata['toleave'] = True + queue = 'command' + elif subaddress == 'owner': + msgdata.update(dict( + toowner=True, + envsender=config.mailman.site_owner, + )) + queue = 'in' + elif subaddress is None: + msgdata['tolist'] = True + queue = 'in' + elif subaddress == 'request': + msgdata['torequest'] = True + queue = 'command' + else: + elog.error('Unknown sub-address: %s', subaddress) + status.append(ERR_550) + continue + # If we found a valid subaddress, enqueue the message and add + # a success status for this recipient. + if queue is not None: + config.switchboards[queue].enqueue(msg, msgdata) + status.append('250 Ok') + except Exception, e: + elog.exception('Queue detection: %s', msg['message-id']) + config.db.abort() + status.append(ERR_550) + # All done; returning this big status string should give the expected + # response to the LMTP client. + return CRLF.join(status) + + def run(self): + """See `IRunner`.""" + asyncore.loop() + + def stop(self): + """See `IRunner`.""" + asyncore.socket_map.clear() + asyncore.close_all() + self.close() diff --git a/src/mailman/queue/maildir.py b/src/mailman/queue/maildir.py new file mode 100644 index 000000000..e8d454d39 --- /dev/null +++ b/src/mailman/queue/maildir.py @@ -0,0 +1,190 @@ +# Copyright (C) 2002-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Maildir pre-queue runner. + +Most MTAs can be configured to deliver messages to a `Maildir'[1]. This +runner will read messages from a maildir's new/ directory and inject them into +Mailman's qfiles/in directory for processing in the normal pipeline. This +delivery mechanism contrasts with mail program delivery, where incoming +messages end up in qfiles/in via the MTA executing the scripts/post script +(and likewise for the other -aliases for each mailing list). + +The advantage to Maildir delivery is that it is more efficient; there's no +need to fork an intervening program just to take the message from the MTA's +standard output, to the qfiles/in directory. + +[1] http://cr.yp.to/proto/maildir.html + +We're going to use the :info flag == 1, experimental status flag for our own +purposes. The :1 can be followed by one of these letters: + +- P means that MaildirRunner's in the process of parsing and enqueuing the + message. If successful, it will delete the file. + +- X means something failed during the parse/enqueue phase. An error message + will be logged to log/error and the file will be renamed <filename>:1,X. + MaildirRunner will never automatically return to this file, but once the + problem is fixed, you can manually move the file back to the new/ directory + and MaildirRunner will attempt to re-process it. At some point we may do + this automatically. + +See the variable USE_MAILDIR in Defaults.py.in for enabling this delivery +mechanism. +""" + +# NOTE: Maildir delivery is experimental in Mailman 2.1. + +import os +import errno +import logging + +from email.Parser import Parser +from email.Utils import parseaddr + +from mailman.Message import Message +from mailman.config import config +from mailman.queue import Runner + +log = logging.getLogger('mailman.error') + +# We only care about the listname and the subq as in listname@ or +# listname-request@ +subqnames = ('admin', 'bounces', 'confirm', 'join', 'leave', + 'owner', 'request', 'subscribe', 'unsubscribe') + +def getlistq(address): + localpart, domain = address.split('@', 1) + # TK: FIXME I only know configs of Postfix. + if config.POSTFIX_STYLE_VIRTUAL_DOMAINS: + p = localpart.split(config.POSTFIX_VIRTUAL_SEPARATOR, 1) + if len(p) == 2: + localpart, domain = p + l = localpart.split('-') + if l[-1] in subqnames: + listname = '-'.join(l[:-1]) + subq = l[-1] + else: + listname = localpart + subq = None + return listname, subq, domain + + +class MaildirRunner(Runner): + # This class is much different than most runners because it pulls files + # of a different format than what scripts/post and friends leaves. The + # files this runner reads are just single message files as dropped into + # the directory by the MTA. This runner will read the file, and enqueue + # it in the expected qfiles directory for normal processing. + def __init__(self, slice=None, numslices=1): + # Don't call the base class constructor, but build enough of the + # underlying attributes to use the base class's implementation. + self._stop = 0 + self._dir = os.path.join(config.MAILDIR_DIR, 'new') + self._cur = os.path.join(config.MAILDIR_DIR, 'cur') + self._parser = Parser(Message) + + def _one_iteration(self): + # Refresh this each time through the list. + listnames = list(config.list_manager.names) + # Cruise through all the files currently in the new/ directory + try: + files = os.listdir(self._dir) + except OSError, e: + if e.errno <> errno.ENOENT: + raise + # Nothing's been delivered yet + return 0 + for file in files: + srcname = os.path.join(self._dir, file) + dstname = os.path.join(self._cur, file + ':1,P') + xdstname = os.path.join(self._cur, file + ':1,X') + try: + os.rename(srcname, dstname) + except OSError, e: + if e.errno == errno.ENOENT: + # Some other MaildirRunner beat us to it + continue + log.error('Could not rename maildir file: %s', srcname) + raise + # Now open, read, parse, and enqueue this message + try: + fp = open(dstname) + try: + msg = self._parser.parse(fp) + finally: + fp.close() + # Now we need to figure out which queue of which list this + # message was destined for. See verp_bounce() in + # BounceRunner.py for why we do things this way. + vals = [] + for header in ('delivered-to', 'envelope-to', 'apparently-to'): + vals.extend(msg.get_all(header, [])) + for field in vals: + to = parseaddr(field)[1].lower() + if not to: + continue + listname, subq, domain = getlistq(to) + listname = listname + '@' + domain + if listname in listnames: + break + else: + # As far as we can tell, this message isn't destined for + # any list on the system. What to do? + log.error('Message apparently not for any list: %s', + xdstname) + os.rename(dstname, xdstname) + continue + # BAW: blech, hardcoded + msgdata = {'listname': listname} + # -admin is deprecated + if subq in ('bounces', 'admin'): + queue = Switchboard(config.BOUNCEQUEUE_DIR) + elif subq == 'confirm': + msgdata['toconfirm'] = 1 + queue = Switchboard(config.CMDQUEUE_DIR) + elif subq in ('join', 'subscribe'): + msgdata['tojoin'] = 1 + queue = Switchboard(config.CMDQUEUE_DIR) + elif subq in ('leave', 'unsubscribe'): + msgdata['toleave'] = 1 + queue = Switchboard(config.CMDQUEUE_DIR) + elif subq == 'owner': + msgdata.update({ + 'toowner': True, + 'envsender': config.SITE_OWNER_ADDRESS, + 'pipeline': config.OWNER_PIPELINE, + }) + queue = Switchboard(config.INQUEUE_DIR) + elif subq is None: + msgdata['tolist'] = 1 + queue = Switchboard(config.INQUEUE_DIR) + elif subq == 'request': + msgdata['torequest'] = 1 + queue = Switchboard(config.CMDQUEUE_DIR) + else: + log.error('Unknown sub-queue: %s', subq) + os.rename(dstname, xdstname) + continue + queue.enqueue(msg, msgdata) + os.unlink(dstname) + except Exception, e: + os.rename(dstname, xdstname) + log.error('%s', e) + + def _clean_up(self): + pass diff --git a/src/mailman/queue/news.py b/src/mailman/queue/news.py new file mode 100644 index 000000000..ed408c72e --- /dev/null +++ b/src/mailman/queue/news.py @@ -0,0 +1,166 @@ +# Copyright (C) 2000-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""NNTP queue runner.""" + +import re +import email +import socket +import logging +import nntplib + +from cStringIO import StringIO + +from mailman import Utils +from mailman.config import config +from mailman.interfaces import NewsModeration +from mailman.queue import Runner + +COMMASPACE = ', ' +log = logging.getLogger('mailman.error') + + +# Matches our Mailman crafted Message-IDs. See Utils.unique_message_id() +# XXX The move to email.utils.make_msgid() breaks this. +mcre = re.compile(r""" + <mailman. # match the prefix + \d+. # serial number + \d+. # time in seconds since epoch + \d+. # pid + (?P<listname>[^@]+) # list's internal_name() + @ # localpart@dom.ain + (?P<hostname>[^>]+) # list's host_name + > # trailer + """, re.VERBOSE) + + + +class NewsRunner(Runner): + def _dispose(self, mlist, msg, msgdata): + # Make sure we have the most up-to-date state + mlist.Load() + if not msgdata.get('prepped'): + prepare_message(mlist, msg, msgdata) + try: + # Flatten the message object, sticking it in a StringIO object + fp = StringIO(msg.as_string()) + conn = None + try: + try: + nntp_host, nntp_port = Utils.nntpsplit(mlist.nntp_host) + conn = nntplib.NNTP(nntp_host, nntp_port, + readermode=True, + user=config.nntp.username, + password=config.nntp.password) + conn.post(fp) + except nntplib.error_temp, e: + log.error('(NNTPDirect) NNTP error for list "%s": %s', + mlist.internal_name(), e) + except socket.error, e: + log.error('(NNTPDirect) socket error for list "%s": %s', + mlist.internal_name(), e) + finally: + if conn: + conn.quit() + except Exception, e: + # Some other exception occurred, which we definitely did not + # expect, so set this message up for requeuing. + self._log(e) + return True + return False + + + +def prepare_message(mlist, msg, msgdata): + # If the newsgroup is moderated, we need to add this header for the Usenet + # software to accept the posting, and not forward it on to the n.g.'s + # moderation address. The posting would not have gotten here if it hadn't + # already been approved. 1 == open list, mod n.g., 2 == moderated + if mlist.news_moderation in (NewsModeration.open_moderated, + NewsModeration.moderated): + del msg['approved'] + msg['Approved'] = mlist.posting_address + # Should we restore the original, non-prefixed subject for gatewayed + # messages? TK: We use stripped_subject (prefix stripped) which was + # crafted in CookHeaders.py to ensure prefix was stripped from the subject + # came from mailing list user. + stripped_subject = msgdata.get('stripped_subject') \ + or msgdata.get('origsubj') + if not mlist.news_prefix_subject_too and stripped_subject is not None: + del msg['subject'] + msg['subject'] = stripped_subject + # Add the appropriate Newsgroups: header + ngheader = msg['newsgroups'] + if ngheader is not None: + # See if the Newsgroups: header already contains our linked_newsgroup. + # If so, don't add it again. If not, append our linked_newsgroup to + # the end of the header list + ngroups = [s.strip() for s in ngheader.split(',')] + if mlist.linked_newsgroup not in ngroups: + ngroups.append(mlist.linked_newsgroup) + # Subtitute our new header for the old one. + del msg['newsgroups'] + msg['Newsgroups'] = COMMASPACE.join(ngroups) + else: + # Newsgroups: isn't in the message + msg['Newsgroups'] = mlist.linked_newsgroup + # Note: We need to be sure two messages aren't ever sent to the same list + # in the same process, since message ids need to be unique. Further, if + # messages are crossposted to two Usenet-gated mailing lists, they each + # need to have unique message ids or the nntpd will only accept one of + # them. The solution here is to substitute any existing message-id that + # isn't ours with one of ours, so we need to parse it to be sure we're not + # looping. + # + # Our Message-ID format is <mailman.secs.pid.listname@hostname> + msgid = msg['message-id'] + hackmsgid = True + if msgid: + mo = mcre.search(msgid) + if mo: + lname, hname = mo.group('listname', 'hostname') + if lname == mlist.internal_name() and hname == mlist.host_name: + hackmsgid = False + if hackmsgid: + del msg['message-id'] + msg['Message-ID'] = email.utils.make_msgid() + # Lines: is useful + if msg['Lines'] is None: + # BAW: is there a better way? + count = len(list(email.Iterators.body_line_iterator(msg))) + msg['Lines'] = str(count) + # Massage the message headers by remove some and rewriting others. This + # woon't completely sanitize the message, but it will eliminate the bulk + # of the rejections based on message headers. The NNTP server may still + # reject the message because of other problems. + for header in config.nntp.remove_headers.split(): + del msg[header] + for rewrite_pairs in config.nntp.rewrite_duplicate_headers.splitlines(): + if len(rewrite_pairs.strip()) == 0: + continue + header, rewrite = rewrite_pairs.split() + values = msg.get_all(header, []) + if len(values) < 2: + # We only care about duplicates + continue + del msg[header] + # But keep the first one... + msg[header] = values[0] + for v in values[1:]: + msg[rewrite] = v + # Mark this message as prepared in case it has to be requeued + msgdata['prepped'] = True diff --git a/src/mailman/queue/outgoing.py b/src/mailman/queue/outgoing.py new file mode 100644 index 000000000..fdb1289fd --- /dev/null +++ b/src/mailman/queue/outgoing.py @@ -0,0 +1,130 @@ +# Copyright (C) 2000-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Outgoing queue runner.""" + +import os +import sys +import socket +import logging + +from datetime import datetime +from lazr.config import as_timedelta + +from mailman.config import config +from mailman.core import errors +from mailman.queue import Runner +from mailman.queue.bounce import BounceMixin + +# This controls how often _do_periodic() will try to deal with deferred +# permanent failures. It is a count of calls to _do_periodic() +DEAL_WITH_PERMFAILURES_EVERY = 10 + +log = logging.getLogger('mailman.error') + + + +class OutgoingRunner(Runner, BounceMixin): + """The outgoing queue runner.""" + + def __init__(self, slice=None, numslices=1): + Runner.__init__(self, slice, numslices) + BounceMixin.__init__(self) + # We look this function up only at startup time. + module_name, callable_name = config.mta.outgoing.rsplit('.', 1) + __import__(module_name) + self._func = getattr(sys.modules[module_name], callable_name) + # This prevents smtp server connection problems from filling up the + # error log. It gets reset if the message was successfully sent, and + # set if there was a socket.error. + self._logged = False + self._retryq = config.switchboards['retry'] + + def _dispose(self, mlist, msg, msgdata): + # See if we should retry delivery of this message again. + deliver_after = msgdata.get('deliver_after', datetime.fromtimestamp(0)) + if datetime.now() < deliver_after: + return True + # Make sure we have the most up-to-date state + try: + pid = os.getpid() + self._func(mlist, msg, msgdata) + # Failsafe -- a child may have leaked through. + if pid <> os.getpid(): + log.error('child process leaked thru: %s', pid) + os._exit(1) + self._logged = False + except socket.error: + # There was a problem connecting to the SMTP server. Log this + # once, but crank up our sleep time so we don't fill the error + # log. + port = int(config.mta.port) + if port == 0: + port = 'smtp' + # Log this just once. + if not self._logged: + log.error('Cannot connect to SMTP server %s on port %s', + config.mta.host, port) + self._logged = True + return True + except errors.SomeRecipientsFailed, e: + # Handle local rejects of probe messages differently. + if msgdata.get('probe_token') and e.permfailures: + self._probe_bounce(mlist, msgdata['probe_token']) + else: + # Delivery failed at SMTP time for some or all of the + # recipients. Permanent failures are registered as bounces, + # but temporary failures are retried for later. + # + # BAW: msg is going to be the original message that failed + # delivery, not a bounce message. This may be confusing if + # this is what's sent to the user in the probe message. Maybe + # we should craft a bounce-like message containing information + # about the permanent SMTP failure? + if e.permfailures: + self._queue_bounces(mlist.fqdn_listname, e.permfailures, + msg) + # Move temporary failures to the qfiles/retry queue which will + # occasionally move them back here for another shot at + # delivery. + if e.tempfailures: + now = datetime.now() + recips = e.tempfailures + last_recip_count = msgdata.get('last_recip_count', 0) + deliver_until = msgdata.get('deliver_until', now) + if len(recips) == last_recip_count: + # We didn't make any progress, so don't attempt + # delivery any longer. BAW: is this the best + # disposition? + if now > deliver_until: + return False + else: + # Keep trying to delivery this message for a while + deliver_until = now + as_timedelta( + config.mta.delivery_retry_period) + msgdata['last_recip_count'] = len(recips) + msgdata['deliver_until'] = deliver_until + msgdata['recips'] = recips + self._retryq.enqueue(msg, msgdata) + # We've successfully completed handling of this message + return False + + _do_periodic = BounceMixin._do_periodic + + def _clean_up(self): + BounceMixin._clean_up(self) + Runner._clean_up(self) diff --git a/src/mailman/queue/pipeline.py b/src/mailman/queue/pipeline.py new file mode 100644 index 000000000..6566b75df --- /dev/null +++ b/src/mailman/queue/pipeline.py @@ -0,0 +1,35 @@ +# Copyright (C) 2008-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""The pipeline queue runner. + +This runner's purpose is to take messages that have been approved for posting +through the 'preparation pipeline'. This pipeline adds, deletes and modifies +headers, calculates message recipients, and more. +""" + +from mailman.core.pipelines import process +from mailman.queue import Runner + + + +class PipelineRunner(Runner): + def _dispose(self, mlist, msg, msgdata): + # Process the message through the mailing list's pipeline. + process(mlist, msg, msgdata, mlist.pipeline) + # Do not keep this message queued. + return False diff --git a/src/mailman/queue/retry.py b/src/mailman/queue/retry.py new file mode 100644 index 000000000..2b5a6afad --- /dev/null +++ b/src/mailman/queue/retry.py @@ -0,0 +1,37 @@ +# Copyright (C) 2003-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +import time + +from mailman.config import config +from mailman.queue import Runner + + + +class RetryRunner(Runner): + def __init__(self, slice=None, numslices=1): + Runner.__init__(self, slice, numslices) + self._outq = config.switchboards['out'] + + def _dispose(self, mlist, msg, msgdata): + # Move it to the out queue for another retry + self._outq.enqueue(msg, msgdata) + return False + + def _snooze(self, filecnt): + # We always want to snooze + time.sleep(self.sleep_float) diff --git a/src/mailman/queue/virgin.py b/src/mailman/queue/virgin.py new file mode 100644 index 000000000..b163d3ea2 --- /dev/null +++ b/src/mailman/queue/virgin.py @@ -0,0 +1,39 @@ +# Copyright (C) 1998-2009 by the Free Software Foundation, Inc. +# +# This file is part of GNU Mailman. +# +# GNU Mailman is free software: you can redistribute it and/or modify it under +# the terms of the GNU General Public License as published by the Free +# Software Foundation, either version 3 of the License, or (at your option) +# any later version. +# +# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for +# more details. +# +# You should have received a copy of the GNU General Public License along with +# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. + +"""Virgin message queue runner. + +This qrunner handles messages that the Mailman system gives virgin birth to. +E.g. acknowledgement responses to user posts or Replybot messages. They need +to go through some minimal processing before they can be sent out to the +recipient. +""" + +from mailman.core.pipelines import process +from mailman.queue import Runner + + + +class VirginRunner(Runner): + def _dispose(self, mlist, msg, msgdata): + # We need to fast track this message through any pipeline handlers + # that touch it, e.g. especially cook-headers. + msgdata['_fasttrack'] = True + # Use the 'virgin' pipeline. + process(mlist, msg, msgdata, 'virgin') + # Do not keep this message queued. + return False |