# This file is released under the GNU GPL, version 3 or a later revision. # For further details see the COPYING file import email.policy from email.utils import getaddresses from enum import Enum, auto import logging from . import headers as HDR from ..helper import formataddr from ..settings.const import settings ACT_REPLY = 'reply' ACT_BOUNCE = 'bounce' ACT_FWD = 'forward' def determine_account(headers, action = ACT_REPLY): """ Determine the local account to use for acting on a message with the given headers. :param headers: headers of the email to inspect :type headers: `db.message._MessageHeaders` :param action: intended use case: one of the ACT_* constants :type action: str :returns: 2-tuple of (from address, account) """ # get accounts my_accounts = settings.get_accounts() assert my_accounts, 'no accounts set!' # extract list of addresses to check for my address # X-Envelope-To and Envelope-To are used to store the recipient address # if not included in other fields # Process the headers in order of importance: if a mail was sent with # account X, with account Y in e.g. CC or delivered-to, make sure that # account X is the one selected and not account Y. for candidate_header in settings.get("reply_account_header_priority"): candidate_addresses = getaddresses(headers.get_all(candidate_header)) logging.debug('candidate addresses: %s', candidate_addresses) # pick the most important account that has an address in candidates # and use that account's realname and the address found here for account in my_accounts: for seen_name, seen_address in candidate_addresses: if account.matches_address(seen_address): if settings.get(action + '_force_realname'): realname = account.realname else: realname = seen_name if settings.get(action + '_force_address'): address = str(account.address) else: address = seen_address logging.debug('using realname: "%s"', realname) logging.debug('using address: %s', address) from_value = formataddr((realname, address)) return from_value, account # revert to default account if nothing found account = my_accounts[0] realname = account.realname address = account.address logging.debug('using realname: "%s"', realname) logging.debug('using address: %s', address) from_value = formataddr((realname, str(address))) return from_value, account class ReplyMode(Enum): AUTHOR = auto() GROUP = auto() LIST = auto() class AddrList: """ An ordered set of unique addresses. """ # list of Address objects _addrs = None # mapping of addr-spec -> Address _addr_specs = None def __init__(self, addresses): self._addrs = [] self._addr_specs = {} for addr in addresses: self.add(addr) def __str__(self): return ', '.join((str(a) for a in self._addrs)) def __len__(self): return len(self._addrs) def __iter__(self): return iter(self._addrs) def _addr_to_addr_spec(self, addr): if hasattr(addr, 'addr_spec'): # headerregistry.Address return addr.addr_spec # account.Address return str(addr) def __contains__(self, addr): return self._addr_to_addr_spec(addr) in self._addr_specs def __delitem__(self, addr): addr_spec = self._addr_to_addr_spec(addr) idx = self._addrs.index(addr_spec) del self._addrs[idx] del self._addr_specs[addr_spec] def __add__(self, other): ret = self.__class__(self) for a in other: ret.add(a) return ret def add(self, addr): if not addr.addr_spec: raise ValueError('Missing addr-spec') if addr in self: idx = self._addrs.index(self._addr_specs[addr.addr_spec]) # if the address we already have is missing display name, # replace it with the new one if not self._addrs[idx].display_name and addr.display_name: self._addrs[idx] = addr self._addr_specs[addr.addr_spec] = addr return self._addrs.append(addr) self._addr_specs[addr.addr_spec] = addr def _reply_list(message): # To choose the target of the reply --list # Reply-To is standart reply target RFC 2822:, RFC 1036: 2.2.1 # X-BeenThere is needed by sourceforge ML also winehq # X-Mailing-List is also standart and is used by git-send-mail # Some mail server (gmail) will not resend you own mail, so we # fall back to 'To' as last resort to = message.headers.pick_addrlist_header( HDR.X_BEEN_THERE, HDR.X_MAILING_LIST, HDR.REPLY_TO, HDR.TO) # copy original Cc cc = message.headers.pick_addrlist_header(HDR.CC) return to, cc def _reply_group(message, senders): # make sure that our own address is not included # if the message was self-sent, then our address is not included MFT = message.headers.pick_addrlist_header(HDR.MAIL_FOLLOWUP_TO) if MFT and settings.get('honor_followup_to'): logging.debug('honor followup to: %s', MFT) to = MFT cc = () else: to = message.headers.pick_addrlist_header(HDR.TO) + senders cc = message.headers.pick_addrlist_header(HDR.CC) return to, cc def determine_recipients(message, account, mode): # TODO: detect reply-to munging by ML here? senders = message.headers.pick_addrlist_header(HDR.MAIL_REPLY_TO, HDR.REPLY_TO, HDR.FROM) self_reply = len(senders) == 1 and account.matches_address(senders[0].addr_spec) if self_reply: senders = message.headers.get(HDR.TO) senders = senders.addresses if senders else () logging.debug('Replying to own message, invert senders to: %s', senders) if mode == ReplyMode.LIST: to, cc = _reply_list(message) elif mode == ReplyMode.GROUP: to, cc = _reply_group(message, senders) else: # mode == ReplyMode.AUTHOR to = senders cc = () to = AddrList(to) cc = AddrList(cc) # remove duplicates across to/cc for a in to: if a in cc: del cc[a] # remove sending account from the address lists for a in [account.address] + account.aliases: # keep own address if it's the only recipient # i.e. the user deliberately wants to self-send the reply if a in to and len(to) > 1: del to[a] if a in cc: del cc[a] logging.debug('%s reply to: %s %s', mode, str(to), str(cc)) return to, cc def subject(message): subject = message.headers.get(HDR.SUBJECT) or '' reply_subject_hook = settings.get_hook('reply_subject') if reply_subject_hook: subject = reply_subject_hook(subject) else: rsp = settings.get('reply_subject_prefix') if not subject.lower().startswith(('re:', rsp.lower())): subject = rsp + subject return subject def body_text(message, ui): name, address = message.get_author() timestamp = message.date qf = settings.get_hook('reply_prefix') if qf: mail = email.message_from_bytes(message.as_bytes(), policy = email.policy.SMTP) quotestring = qf(name, address, timestamp, message=mail, ui=ui, dbm=ui.dbman) else: quotestring = 'Quoting %s (%s)\n' % (name or address, timestamp) mailcontent = quotestring quotehook = settings.get_hook('text_quote') if quotehook: mailcontent += quotehook(message.get_body_text()) else: quote_prefix = settings.get('quote_prefix') for line in message.get_body_text().splitlines(): mailcontent += quote_prefix + line + '\n' return mailcontent def references(message): old_references = message.headers.get(HDR.REFERENCES) references = [] if old_references: references = old_references.split() # limit to 16 references, including the one we add del references[1:-14] references.append('<%s>' % message.id) return ' '.join(references) def mail_followup_to(recipients): if not settings.get('followup_to'): return None lists = settings.get('mailinglists') # check if any recipient address matches a known mailing list if any(a.addr_spec in lists for a in recipients): logging.debug('mail followup to: %s', recipients) return recipients