path: root/alot/db/message.py

# Copyright (C) 2011-2012  Patrick Totzke <patricktotzke@gmail.com>
# This file is released under the GNU GPL, version 3 or a later revision.
# For further details see the COPYING file
import email
import email.charset as charset
import email.policy
import logging
import mailcap
import os
import tempfile

from datetime import datetime

from notmuch import NullPointerError

from .attachment import Attachment
from .. import crypto
from .. import helper
from ..errors import GPGProblem
from ..helper import parse_mailcap_nametemplate
from ..helper import split_commandstring
from ..helper import string_sanitize
from ..settings.const import settings

charset.add_charset('utf-8', charset.QP, charset.QP, 'utf-8')

X_SIGNATURE_VALID_HEADER = 'X-Alot-OpenPGP-Signature-Valid'
X_SIGNATURE_MESSAGE_HEADER = 'X-Alot-OpenPGP-Signature-Message'

_APP_PGP_SIG = 'application/pgp-signature'
_APP_PGP_ENC = 'application/pgp-encrypted'

def _add_signature_headers(mail, sigs, error_msg):
    '''Add pseudo headers to the mail indicating whether the signature
    verification was successful.

    :param mail: :class:`email.message.Message` the message to entitle
    :param sigs: list of :class:`gpg.results.Signature`
    :param error_msg: An error message if there is one, or None
    :type error_msg: :class:`str` or `None`
    '''
    sig_from = ''
    sig_known = True
    uid_trusted = False

    assert error_msg is None or isinstance(error_msg, str)

    if not sigs:
        error_msg = error_msg or 'no signature found'
    elif not error_msg:
        try:
            key = crypto.get_key(sigs[0].fpr)
            for uid in key.uids:
                if crypto.check_uid_validity(key, uid.email):
                    sig_from = uid.uid
                    uid_trusted = True
                    break
            else:
                # No trusted uid found, since we did not break from the loop.
                sig_from = key.uids[0].uid
        except GPGProblem:
            sig_from = sigs[0].fpr
            sig_known = False

    if error_msg:
        msg = 'Invalid: {}'.format(error_msg)
    elif uid_trusted:
        msg = 'Valid: {}'.format(sig_from)
    else:
        msg = 'Untrusted: {}'.format(sig_from)

    mail.add_header(X_SIGNATURE_VALID_HEADER,
                    'False' if (error_msg or not sig_known) else 'True')
    mail.add_header(X_SIGNATURE_MESSAGE_HEADER, msg)

def _handle_signatures(original, message):
    """Shared code for handling message signatures.

    RFC 3156 is quite strict:
    * exactly two messages
    * the second is of type 'application/pgp-signature'
    * the second contains the detached signature

    :param original: The original top-level mail. This is required to attache
        special headers to
    :type original: :class:`email.message.Message`
    :param message: The multipart/signed payload to verify
    :type message: :class:`email.message.Message`
    """
    malformed = None
    payload = message.get_payload()
    if len(payload) != 2:
        malformed = 'expected exactly two messages, got {0}'.format(len(payload))
    else:
        ct = payload[1].get_content_type()
        if ct != _APP_PGP_SIG:
            malformed = 'expected Content-Type: {0}, got: {1}'.format(
                _APP_PGP_SIG, ct)

    # TODO: RFC 3156 says the alg has to be lower case, but I've seen a message
    # with 'PGP-'. maybe we should be more permissive here, or maybe not, this
    # is crypto stuff...
    micalg = message.get_param('micalg', '')
    if not micalg.startswith('pgp-'):
        malformed = 'expected micalg=pgp-..., got: {0}'.format(micalg)

    sigs = []
    if not malformed:
        try:
            sigs = crypto.verify_detached(
                payload[0].as_bytes(policy=email.policy.SMTP),
                payload[1].get_payload(decode=True))
        except GPGProblem as e:
            malformed = str(e)

    _add_signature_headers(original, sigs, malformed)


def _handle_encrypted(original, message, session_keys=None):
    """Handle encrypted messages helper.

    RFC 3156 is quite strict:
    * exactly two messages
    * the first is of type 'application/pgp-encrypted'
    * the first contains 'Version: 1'
    * the second is of type 'application/octet-stream'
    * the second contains the encrypted and possibly signed data

    :param original: The original top-level mail. This is required to attache
        special headers to
    :type original: :class:`email.message.Message`
    :param message: The multipart/signed payload to verify
    :type message: :class:`email.message.Message`
    :param session_keys: a list OpenPGP session keys
    :type session_keys: [str]
    """
    malformed = False

    ct = message.get_payload(0).get_content_type()
    if ct != _APP_PGP_ENC:
        malformed = 'expected Content-Type: {0}, got: {1}'.format(
            _APP_PGP_ENC, ct)

    want = 'application/octet-stream'
    ct = message.get_payload(1).get_content_type()
    if ct != want:
        malformed = 'expected Content-Type: {0}, got: {1}'.format(want, ct)

    if not malformed:
        # This should be safe because PGP uses US-ASCII characters only
        payload = message.get_payload(1).get_payload().encode('ascii')
        try:
            sigs, d = crypto.decrypt_verify(payload, session_keys)
        except GPGProblem as e:
            # signature verification failures end up here too if the combined
            # method is used, currently this prevents the interpretation of the
            # recovered plain text mail. maybe that's a feature.
            malformed = str(e)
        else:
            n = _decrypted_message_from_bytes(d, session_keys)

            # add the decrypted message to message. note that n contains all
            # the attachments, no need to walk over n here.
            original.attach(n)

            original.defects.extend(n.defects)

            # there are two methods for both signed and encrypted data, one is
            # called 'RFC 1847 Encapsulation' by RFC 3156, and one is the
            # 'Combined method'.
            if not sigs:
                # 'RFC 1847 Encapsulation', the signature is a detached
                # signature found in the recovered mime message of type
                # multipart/signed.
                if X_SIGNATURE_VALID_HEADER in n:
                    for k in (X_SIGNATURE_VALID_HEADER,
                              X_SIGNATURE_MESSAGE_HEADER):
                        original[k] = n[k]
            else:
                # 'Combined method', the signatures are returned by the
                # decrypt_verify function.

                # note that if we reached this point, we know the signatures
                # are valid. if they were not valid, the else block of the
                # current try would not have been executed
                _add_signature_headers(original, sigs, '')

    if malformed:
        msg = 'Malformed OpenPGP message: {0}'.format(malformed)
        content = email.message_from_string(msg,
                                            _class=email.message.EmailMessage,
                                            policy=email.policy.SMTP)
        content.set_charset('utf-8')
        original.attach(content)

def _decrypted_message_from_bytes(bytestring, session_keys = None):
    '''Detect and decrypt OpenPGP encrypted data in an email object. If this
    succeeds, any mime messages found in the recovered plaintext
    message are added to the returned message object.

    :param session_keys: a list OpenPGP session keys
    :returns: :class:`email.message.Message` possibly augmented with
              decrypted data
    '''
    enc = email.message_from_bytes(bytestring, policy = email.policy.SMTP)

    # make sure no one smuggles a token in (data from enc is untrusted)
    del enc[X_SIGNATURE_VALID_HEADER]
    del enc[X_SIGNATURE_MESSAGE_HEADER]

    if enc.is_multipart():
        # handle OpenPGP signed data
        if (enc.get_content_subtype() == 'signed' and
            enc.get_param('protocol') == _APP_PGP_SIG):
            _handle_signatures(enc, enc)

        # handle OpenPGP encrypted data
        elif (enc.get_content_subtype() == 'encrypted' and
              enc.get_param('protocol') == _APP_PGP_ENC and
              'Version: 1' in enc.get_payload(0).get_payload()):
            _handle_encrypted(enc, enc, session_keys)

        # It is also possible to put either of the abov into a multipart/mixed
        # segment
        elif enc.get_content_subtype() == 'mixed':
            sub = enc.get_payload(0)

            if sub.is_multipart():
                if (sub.get_content_subtype() == 'signed' and
                    sub.get_param('protocol') == _APP_PGP_SIG):
                    _handle_signatures(enc, sub)
                elif (sub.get_content_subtype() == 'encrypted' and
                      sub.get_param('protocol') == _APP_PGP_ENC):
                    _handle_encrypted(enc, sub, session_keys)

    return enc

def render_part(part, field_key='copiousoutput'):
    """
    renders a non-multipart email part into displayable plaintext by piping its
    payload through an external script. The handler itself is determined by
    the mailcap entry for this part's ctype.
    """
    ctype = part.get_content_type()
    raw_payload = remove_cte(part)
    rendered_payload = None
    # get mime handler
    _, entry = settings.mailcap_find_match(ctype, key=field_key)
    if entry is not None:
        tempfile_name = None
        stdin = None
        handler_raw_commandstring = entry['view']
        # in case the mailcap defined command contains no '%s',
        # we pipe the files content to the handling command via stdin
        if '%s' in handler_raw_commandstring:
            # open tempfile, respect mailcaps nametemplate
            nametemplate = entry.get('nametemplate', '%s')
            prefix, suffix = parse_mailcap_nametemplate(nametemplate)
            with tempfile.NamedTemporaryFile(
                    delete=False, prefix=prefix, suffix=suffix) \
                    as tmpfile:
                tmpfile.write(raw_payload)
                tempfile_name = tmpfile.name
        else:
            stdin = raw_payload

        # read parameter, create handler command
        parms = tuple('='.join(p) for p in part.get_params())

        # create and call external command
        cmd = mailcap.subst(entry['view'], ctype,
                            filename=tempfile_name, plist=parms)
        logging.debug('command: %s', cmd)
        logging.debug('parms: %s', str(parms))
        cmdlist = split_commandstring(cmd)
        # call handler
        stdout, _, _ = helper.call_cmd(cmdlist, stdin=stdin)
        if stdout:
            rendered_payload = stdout

        # remove tempfile
        if tempfile_name:
            os.unlink(tempfile_name)

    return rendered_payload

def remove_cte(part, as_string=False):
    """Interpret MIME-part according to it's Content-Transfer-Encodings.

    This returns the payload of `part` as string or bytestring for display, or
    to be passed to an external program. In the raw file the payload may be
    encoded, e.g. in base64, quoted-printable, 7bit, or 8bit. This method will
    look for one of the above Content-Transfer-Encoding header and interpret
    the payload accordingly.

    Incorrect header values (common in spam messages) will be interpreted as
    lenient as possible and will result in INFO-level debug messages.

    ..Note:: All this may be depricated in favour of
             `email.contentmanager.raw_data_manager` (v3.6+)

    :param email.message.EmailMessage part: The part to decode
    :param bool as_string: If true return a str, otherwise return bytes
    :returns: The mail with any Content-Transfer-Encoding removed
    :rtype: Union[str, bytes]
    """
    payload = part.get_payload(decode = True)
    if as_string:
        enc = part.get_content_charset('ascii')
        if enc.startswith('windows-'):
            enc = enc.replace('windows-', 'cp', 1)

        try:
            payload = payload.decode(enc, errors = 'backslashreplace')
        except LookupError:
            # enc is unknown;
            # fall back to guessing the correct encoding using libmagic
            payload = helper.try_decode(payload)
        except UnicodeDecodeError as emsg:
            # the mail contains chars that are not enc-encoded.
            # libmagic works better than just ignoring those
            logging.debug('Decoding failure: {}'.format(emsg))
            payload = helper.try_decode(payload)

    return payload

MISSING_HTML_MSG = ("This message contains a text/html part that was not "
                    "rendered due to a missing mailcap entry. "
                    "Please refer to item 5 in our FAQ: "
                    "http://alot.rtfd.io/en/latest/faq.html")

def extract_body(mail):
    """Returns a string view of a Message.

    This consults :ref:`prefer_plaintext <prefer-plaintext>`
    to determine if a "text/plain" alternative is preferred over a "text/html"
    part.

    :param mail: the mail to use
    :type mail: :class:`email.message.EmailMessage`
    :returns: The combined text of any parts to be used
    :rtype: str
    """

    if settings.get('prefer_plaintext'):
        preferencelist = ('plain', 'html')
    else:
        preferencelist = ('html', 'plain')

    body_part = mail.get_body(preferencelist)
    if body_part is None:  # if no part matching preferredlist was found
        return ""

    displaystring = ""

    if body_part.get_content_type() == 'text/plain':
        displaystring = string_sanitize(remove_cte(body_part, as_string=True))
    else:
        rendered_payload = render_part(body_part)
        if rendered_payload:  # handler had output
            displaystring = string_sanitize(rendered_payload)
        else:
            if body_part.get_content_type() == 'text/html':
                displaystring = MISSING_HTML_MSG
    return displaystring

class _MessageHeaders:
    _msg = None

    def __init__(self, msg):
        self._msg = msg

    def __getitem__(self, key):
        if not key in self._msg:
            raise KeyError(key)
        return self._msg.get_all(key)

    def keys(self):
        return self._msg.keys()

    def items(self):
        return self._msg.items()

class Message:
    """
    a persistent notmuch message object.
    It it uses a :class:`~alot.db.DBManager` for cached manipulation
    and lazy lookups.
    """

    """the :class:`~alot.db.Thread` this Message belongs to"""
    thread = None

    """value of the Date header value as :class:`~datetime.datetime`"""
    date = None

    """value of the Message-Id header (str)"""
    id = None

    """Paths to all files corresponding to this message"""
    filenames = None

    """this message's depth in the thread tree"""
    depth = None

    """A list of replies to this message"""
    replies = None

    """
    This message parent in the list (i.e. the message this message is a reply
    to). None when this message is top-level.
    """
    parent = None

    """
    The object providing access to the email's headers.
    """
    headers = None

    def __init__(self, dbman, thread, msg, depth):
        """
        :param dbman: db manager that is used for further lookups
        :type dbman: alot.db.DBManager
        :param thread: this messages thread
        :type thread: :class:`~alot.db.Thread`
        :param msg: the wrapped message
        :type msg: notmuch.database.Message
        :param depth: depth of this message in the thread tree (0 for toplevel
                      messages, 1 for their replies etc.)
        :type depth int
        """
        self._dbman = dbman
        self.id = msg.get_message_id()
        self.thread = thread
        self.depth   = depth
        try:
            self.date = datetime.fromtimestamp(msg.get_date())
        except ValueError:
            self.date = None

        filenames = []
        for f in msg.get_filenames():
            filenames.append(f[:])
        if len(filenames) == 0:
            raise ValueError('No filenames for a message returned')
        self.filenames = filenames

        session_keys = []
        for name, value in msg.get_properties("session-key", exact=True):
            if name == "session-key":
                session_keys.append(value)

        self._email = self._load_email(session_keys)

        self.headers = _MessageHeaders(self._email)

        self._attachments = None  # will be read upon first use
        self._tags = set(msg.get_tags())

        sender = self._email.get('From')
        if sender is None:
            sender = self._email.get('Sender')

        if sender:
            self._from = sender
        elif 'draft' in self._tags:
            acc = settings.get_accounts()[0]
            self._from = '"{}" <{}>'.format(acc.realname, str(acc.address))
        else:
            self._from = '"Unknown" <>'

    def __str__(self):
        """prettyprint the message"""
        aname, aaddress = self.get_author()
        if not aname:
            aname = aaddress
        return "%s (%s)" % (aname, self.get_datestring())

    def __hash__(self):
        """needed for sets of Messages"""
        return hash(self.id)

    def __eq__(self, other):
        if isinstance(other, type(self)):
            return self.id == other.id
        return NotImplemented

    @property
    def filename(self):
        return self.filenames[0]

    def _load_email(self, session_keys):
        warning = "Subject: Caution!\n"\
                  "Message file is no longer accessible:\n%s" % self.filename
        try:
            with open(self.filename, 'rb') as f:
                mail = _decrypted_message_from_bytes(f.read(), session_keys)
        except IOError:
            mail = email.message_from_string(
                warning, policy=email.policy.SMTP)

        return mail

    def get_email(self):
        """returns :class:`email.email.EmailMessage` for this message"""
        return self._email

    def get_message_parts(self):
        """yield all body parts of this message"""
        for msg in self.get_email().walk():
            if not msg.is_multipart():
                yield msg

    def get_tags(self):
        """returns tags attached to this message as list of strings"""
        return sorted(self._tags)

    def get_datestring(self):
        """
        returns reformated datestring for this message.

        It uses :meth:`SettingsManager.represent_datetime` to represent
        this messages `Date` header

        :rtype: str
        """
        if self.date is None:
            return None
        return settings.represent_datetime(self.date)

    def get_author(self):
        """
        returns realname and address of this messages author

        :rtype: (str,str)
        """
        return email.utils.parseaddr(self._from)

    def add_tags(self, tags, afterwards=None, remove_rest=False):
        """
        adds tags to message

        .. note::

            This only adds the requested operation to this objects
            :class:`DBManager's <alot.db.DBManager>` write queue.
            You need to call :meth:`~alot.db.DBManager.flush` to write out.

        :param tags: a list of tags to be added
        :type tags: list of str
        :param afterwards: callback that gets called after successful
                           application of this tagging operation
        :type afterwards: callable
        :param remove_rest: remove all other tags
        :type remove_rest: bool
        """
        def myafterwards():
            if remove_rest:
                self._tags = set(tags)
            else:
                self._tags = self._tags.union(tags)
            if callable(afterwards):
                afterwards()

        self._dbman.tag('id:' + self.id, tags, afterwards=myafterwards,
                        remove_rest=remove_rest)
        self._tags = self._tags.union(tags)

    def remove_tags(self, tags, afterwards=None):
        """remove tags from message

        .. note::

            This only adds the requested operation to this objects
            :class:`DBManager's <alot.db.DBManager>` write queue.
            You need to call :meth:`~alot.db.DBManager.flush` to actually out.

        :param tags: a list of tags to be added
        :type tags: list of str
        :param afterwards: callback that gets called after successful
                           application of this tagging operation
        :type afterwards: callable
        """
        def myafterwards():
            self._tags = self._tags.difference(tags)
            if callable(afterwards):
                afterwards()

        self._dbman.untag('id:' + self.id, tags, myafterwards)

    def get_attachments(self):
        """
        returns messages attachments

        Derived from the leaves of the email mime tree
        that and are not part of :rfc:`2015` syntax for encrypted/signed mails
        and either have :mailheader:`Content-Disposition` `attachment`
        or have :mailheader:`Content-Disposition` `inline` but specify
        a filename (as parameter to `Content-Disposition`).

        :rtype: list of :class:`Attachment`
        """
        if not self._attachments:
            self._attachments = []
            for part in self.get_message_parts():
                cd = part.get('Content-Disposition', '')
                filename = part.get_filename()
                ct = part.get_content_type()
                # replace underspecified mime description by a better guess
                if ct in ['octet/stream', 'application/octet-stream']:
                    content = part.get_payload(decode=True)
                    ct = helper.guess_mimetype(content)
                    if (self._attachments and
                            self._attachments[-1].get_content_type() ==
                            'application/pgp-encrypted'):
                        self._attachments.pop()

                if cd.lower().startswith('attachment'):
                    if ct.lower() not in ['application/pgp-signature']:
                        self._attachments.append(Attachment(part))
                elif cd.lower().startswith('inline'):
                    if (filename is not None and
                            ct.lower() != 'application/pgp'):
                        self._attachments.append(Attachment(part))
        return self._attachments

    def get_body_text(self):
        """ returns bodystring extracted from this mail """
        # TODO: allow toggle commands to decide which part is considered body
        return extract_body(self.get_email())

    def matches(self, querystring):
        """tests if this messages is in the resultset for `querystring`"""
        searchfor = '( {} ) AND id:{}'.format(querystring, self.id)
        return self._dbman.count_messages(searchfor) > 0

    def parents(self):
        """
        A generator iterating over this message's parents up to the topmost
        level.
        """
        m = self.parent
        while m:
            yield m
            m = m.parent