"""Standard Mailman message object.
This is a subclass of mimeo.Message but provides a slightly extended interface
which is more convenient for use inside Mailman.
"""
import re
import email
import email.Message
import email.Utils
from email.Charset import Charset
from email.Header import Header
from types import ListType, StringType
from Mailman import mm_cfg
from Mailman import Utils
COMMASPACE = ', '
mo = re.match(r'([\d.]+)', email.__version__)
VERSION = tuple([int(s) for s in mo.group().split('.')])
class Message(email.Message.Message):
def __init__(self):
self.__version__ = VERSION
email.Message.Message.__init__(self)
def __repr__(self):
return self.__str__()
def __setstate__(self, d):
self.__dict__ = d
version = d.get('__version__', (0, 0, 0))
d['__version__'] = VERSION
if version >= VERSION:
return
if not d.has_key('_charset'):
self._charset = None
if not d.has_key('_default_type'):
self._default_type = 'text/plain'
headers = []
hchanged = 0
for k, v in self._headers:
if isinstance(v, Header):
chunks = []
cchanged = 0
for s, charset in v._chunks:
if isinstance(charset, StringType):
charset = Charset(charset)
cchanged = 1
chunks.append((s, charset))
if cchanged:
v._chunks = chunks
hchanged = 1
headers.append((k, v))
if hchanged:
self._headers = headers
def get_sender(self, use_envelope=None, preserve_case=0):
"""Return the address considered to be the author of the email.
This can return either the From: header, the Sender: header or the
envelope header (a.k.a. the unixfrom header). The first non-empty
header value found is returned. However the search order is
determined by the following:
- If mm_cfg.USE_ENVELOPE_SENDER is true, then the search order is
Sender:, From:, unixfrom
- Otherwise, the search order is From:, Sender:, unixfrom
The optional argument use_envelope, if given overrides the
mm_cfg.USE_ENVELOPE_SENDER setting. It should be set to either 0 or 1
(don't use None since that indicates no-override).
unixfrom should never be empty. The return address is always
lowercased, unless preserve_case is true.
This method differs from get_senders() in that it returns one and only
one address, and uses a different search order.
"""
senderfirst = mm_cfg.USE_ENVELOPE_SENDER
if use_envelope is not None:
senderfirst = use_envelope
if senderfirst:
headers = ('sender', 'from')
else:
headers = ('from', 'sender')
for h in headers:
fieldval = self[h]
if not fieldval:
continue
addrs = email.Utils.getaddresses([fieldval])
try:
realname, address = addrs[0]
except IndexError:
continue
if address:
break
else:
unixfrom = self.get_unixfrom()
if unixfrom:
address = unixfrom.split()[1]
else:
address = ''
if not preserve_case:
return address.lower()
return address
def get_senders(self, preserve_case=0, headers=None):
"""Return a list of addresses representing the author of the email.
The list will contain the following addresses (in order)
depending on availability:
1. From:
2. unixfrom
3. Reply-To:
4. Sender:
The return addresses are always lower cased, unless `preserve_case' is
true. Optional `headers' gives an alternative search order, with None
meaning, search the unixfrom header. Items in `headers' are field
names without the trailing colon.
"""
if headers is None:
headers = mm_cfg.SENDER_HEADERS
pairs = []
for h in headers:
if h is None:
fieldval = self.get_unixfrom() or ''
try:
pairs.append(('', fieldval.split()[1]))
except IndexError:
pass
else:
fieldvals = self.get_all(h)
if fieldvals:
pairs.extend(email.Utils.getaddresses(fieldvals))
authors = []
for pair in pairs:
address = pair[1]
if address is not None and not preserve_case:
address = address.lower()
authors.append(address)
return authors
def get_filename(self, failobj=None):
"""Some MUA have bugs in RFC2231 filename encoding and cause
Mailman to stop delivery in Scrubber.py (called from ToDigest.py).
"""
try:
filename = email.Message.Message.get_filename(self, failobj)
return filename
except (UnicodeError, LookupError, ValueError):
return failobj
class UserNotification(Message):
"""Class for internally crafted messages."""
def __init__(self, recip, sender, subject=None, text=None, lang=None):
Message.__init__(self)
charset = None
if lang is not None:
charset = Charset(Utils.GetCharSet(lang))
if text is not None:
self.set_payload(text, charset)
if subject is None:
subject = '(no subject)'
self['Subject'] = Header(subject, charset, header_name='Subject',
errors='replace')
self['From'] = sender
if isinstance(recip, ListType):
self['To'] = COMMASPACE.join(recip)
self.recips = recip
else:
self['To'] = recip
self.recips = [recip]
def send(self, mlist, **_kws):
"""Sends the message by enqueuing it to the `virgin' queue.
This is used for all internally crafted messages.
"""
if not self.has_key('message-id'):
self['Message-ID'] = Utils.unique_message_id(mlist)
if not self.has_key('date'):
self['Date'] = email.Utils.formatdate(localtime=1)
if not self.has_key('precedence'):
self['Precedence'] = 'bulk'
self._enqueue(mlist, **_kws)
def _enqueue(self, mlist, **_kws):
from Mailman.Queue.sbcache import get_switchboard
virginq = get_switchboard(mm_cfg.VIRGINQUEUE_DIR)
virginq.enqueue(self,
listname = mlist.internal_name(),
recips = self.recips,
nodecorate = 1,
reduced_list_headers = 1,
**_kws)
class OwnerNotification(UserNotification):
"""Like user notifications, but this message goes to the list owners."""
def __init__(self, mlist, subject=None, text=None, tomoderators=1):
recips = mlist.owner[:]
if tomoderators:
recips.extend(mlist.moderator)
sender = Utils.get_site_email(mlist.host_name, 'bounces')
lang = mlist.preferred_language
UserNotification.__init__(self, recips, sender, subject, text, lang)
del self['to']
self['To'] = mlist.GetOwnerEmail()
self._sender = sender
def _enqueue(self, mlist, **_kws):
from Mailman.Queue.sbcache import get_switchboard
virginq = get_switchboard(mm_cfg.VIRGINQUEUE_DIR)
virginq.enqueue(self,
listname = mlist.internal_name(),
recips = self.recips,
nodecorate = 1,
reduced_list_headers = 1,
envsender = self._sender,
**_kws)