rss2email 1:3.9-2 / usr / lib / python3 / dist-packages / rss2email / email.py

# -*- encoding: utf-8 -*-
#
# Copyright (C) 2012-2014 Arun Persaud <apersaud@lbl.gov>
#                         Dmitry Bogatov <KAction@gnu.org>
#                         George Saunders <georgesaunders@gmail.com>
#                         Thiago Coutinho <root@thiagoc.net>
#                         W. Trevor King <wking@tremily.us>
#
# This file is part of rss2email.
#
# rss2email 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 2 of the License, or (at your option) version 3 of
# the License.
#
# rss2email 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
# rss2email.  If not, see <http://www.gnu.org/licenses/>.

"""Email message generation and dispatching
"""

import email as _email
from email.charset import Charset as _Charset
import email.encoders as _email_encoders
from email.generator import BytesGenerator as _BytesGenerator
import email.header as _email_header
from email.header import Header as _Header
from email.mime.text import MIMEText as _MIMEText
from email.utils import formataddr as _formataddr
from email.utils import parseaddr as _parseaddr
import imaplib as _imaplib
import io as _io
import smtplib as _smtplib
import ssl as _ssl
import subprocess as _subprocess
import sys as _sys
import time as _time

from . import LOG as _LOG
from . import config as _config
from . import error as _error


def guess_encoding(string, encodings=('US-ASCII', 'UTF-8')):
    """Find an encoding capable of encoding `string`.

    >>> guess_encoding('alpha', encodings=('US-ASCII', 'UTF-8'))
    'US-ASCII'
    >>> guess_encoding('α', encodings=('US-ASCII', 'UTF-8'))
    'UTF-8'
    >>> guess_encoding('α', encodings=('US-ASCII', 'ISO-8859-1'))
    Traceback (most recent call last):
      ...
    rss2email.error.NoValidEncodingError: no valid encoding for α in ('US-ASCII', 'ISO-8859-1')
    """
    for encoding in encodings:
        try:
            string.encode(encoding)
        except (UnicodeError, LookupError):
            pass
        else:
            return encoding
    raise _error.NoValidEncodingError(string=string, encodings=encodings)

def get_message(sender, recipient, subject, body, content_type,
                extra_headers=None, config=None, section='DEFAULT'):
    """Generate a `Message` instance.

    All arguments should be Unicode strings (plain ASCII works as well).

    Only the real name part of sender and recipient addresses may contain
    non-ASCII characters.

    The email will be properly MIME encoded.

    The charset of the email will be the first one out of the list
    that can represent all the characters occurring in the email.

    >>> message = get_message(
    ...     sender='John <jdoe@a.com>', recipient='Ζεύς <z@olympus.org>',
    ...     subject='Testing',
    ...     body='Hello, world!\\n',
    ...     content_type='plain',
    ...     extra_headers={'Approved': 'joe@bob.org'})
    >>> print(message.as_string())  # doctest: +REPORT_UDIFF
    MIME-Version: 1.0
    Content-Type: text/plain; charset="us-ascii"
    Content-Transfer-Encoding: 7bit
    From: John <jdoe@a.com>
    To: =?utf-8?b?zpbOtc+Nz4I=?= <z@olympus.org>
    Subject: Testing
    Approved: joe@bob.org
    <BLANKLINE>
    Hello, world!
    <BLANKLINE>
    """
    if config is None:
        config = _config.CONFIG
    if section not in config.sections():
        section = 'DEFAULT'
    encodings = [
        x.strip() for x in config.get(section, 'encodings').split(',')]

    # Split real name (which is optional) and email address parts
    sender_name,sender_addr = _parseaddr(sender)
    recipient_name,recipient_addr = _parseaddr(recipient)

    sender_encoding = guess_encoding(sender_name, encodings)
    recipient_encoding = guess_encoding(recipient_name, encodings)
    subject_encoding = guess_encoding(subject, encodings)
    body_encoding = guess_encoding(body, encodings)

    # We must always pass Unicode strings to Header, otherwise it will
    # use RFC 2047 encoding even on plain ASCII strings.
    sender_name = str(_Header(sender_name, sender_encoding).encode())
    recipient_name = str(_Header(recipient_name, recipient_encoding).encode())

    # Make sure email addresses do not contain non-ASCII characters
    sender_addr.encode('ascii')
    recipient_addr.encode('ascii')

    # Create the message ('plain' stands for Content-Type: text/plain)
    message = _MIMEText(body, content_type, body_encoding)
    message['From'] = _formataddr((sender_name, sender_addr))
    message['To'] = _formataddr((recipient_name, recipient_addr))
    message['Subject'] = _Header(subject, subject_encoding)
    if config.getboolean(section, 'use-8bit'):
        del message['Content-Transfer-Encoding']
        charset = _Charset(body_encoding)
        charset.body_encoding = _email_encoders.encode_7or8bit
        message.set_payload(body, charset=charset)
    if extra_headers:
        for key,value in extra_headers.items():
            encoding = guess_encoding(value, encodings)
            message[key] = _Header(value, encoding)
    return message

def smtp_send(sender, recipient, message, config=None, section='DEFAULT'):
    if config is None:
        config = _config.CONFIG
    server = config.get(section, 'smtp-server')
    _LOG.debug('sending message to {} via {}'.format(recipient, server))
    ssl = config.getboolean(section, 'smtp-ssl')
    if ssl:
        smtp = _smtplib.SMTP_SSL()
    else:
        smtp = _smtplib.SMTP()
    try:
        smtp.connect(server)
    except KeyboardInterrupt:
        raise
    except Exception as e:
        raise _error.SMTPConnectionError(server=server) from e
    if config.getboolean(section, 'smtp-auth'):
        username = config.get(section, 'smtp-username')
        password = config.get(section, 'smtp-password')
        try:
            if not ssl:
                protocol_name = config.get(section, 'smtp-ssl-protocol')
                if protocol_name == 'SSLv3':
                    _LOG.warning('SSLv3 is not available, defaulting to TLSv1.')
                    protocol_name = 'TLSv1'
                protocol = getattr(_ssl, 'PROTOCOL_{}'.format(protocol_name))
                try:
                    smtp.starttls(context=_ssl.SSLContext(protocol=protocol))
                except TypeError:
                    # Python 3.2 or earlier
                    smtp.starttls()
            smtp.login(username, password)
        except KeyboardInterrupt:
            raise
        except Exception as e:
            raise _error.SMTPAuthenticationError(
                server=server, username=username)
    smtp.send_message(message, sender, [recipient])
    smtp.quit()

def imap_send(message, config=None, section='DEFAULT'):
    if config is None:
        config = _config.CONFIG
    server = config.get(section, 'imap-server')
    port = config.getint(section, 'imap-port')
    _LOG.debug('sending message to {}:{}'.format(server, port))
    ssl = config.getboolean(section, 'imap-ssl')
    if ssl:
        imap = _imaplib.IMAP4_SSL(server, port)
    else:
        imap = _imaplib.IMAP4(server, port)
    try:
        if config.getboolean(section, 'imap-auth'):
            username = config.get(section, 'imap-username')
            password = config.get(section, 'imap-password')
            try:
                if not ssl:
                    imap.starttls()
                imap.login(username, password)
            except KeyboardInterrupt:
                raise
            except Exception as e:
                raise _error.IMAPAuthenticationError(
                    server=server, port=port, username=username)
        mailbox = config.get(section, 'imap-mailbox')
        date = _imaplib.Time2Internaldate(_time.localtime())
        message_bytes = _flatten(message)
        imap.append(mailbox, None, date, message_bytes)
    finally:
        imap.logout()

def _decode_header(header):
    """Decode RFC-2047-encoded headers to Unicode strings

    >>> from email.header import Header
    >>> _decode_header('abc')
    'abc'
    >>> _decode_header('=?utf-8?b?zpbOtc+Nz4I=?= <z@olympus.org>')
    'Ζεύς <z@olympus.org>'
    >>> _decode_header(Header('Ζεύς <z@olympus.org>', 'utf-8'))
    'Ζεύς <z@olympus.org>'
    """
    if isinstance(header, _Header):
        return str(header)
    chunks = []
    for chunk,charset in _email_header.decode_header(header):
        if charset is None:
            if isinstance(chunk, bytes):
                chunk = str(chunk, 'ascii')
            chunks.append(chunk)
        else:
            chunks.append(str(chunk, charset))
    if _sys.version_info < (3, 3):  # Python 3.2 and older
        return ' '.join(chunks)  # http://bugs.python.org/issue1079
    return ''.join(chunks)

def _flatten(message):
    r"""Flatten an email.message.Message to bytes

    >>> import rss2email.config
    >>> config = rss2email.config.Config()
    >>> config.read_dict(rss2email.config.CONFIG)

    Here's a 7-bit, base64 version:

    >>> message = get_message(
    ...     sender='John <jdoe@a.com>', recipient='Ζεύς <z@olympus.org>',
    ...     subject='Homage',
    ...     body="You're great, Ζεύς!\n",
    ...     content_type='plain',
    ...     config=config)
    >>> for line in _flatten(message).split(b'\n'):
    ...     print(line)  # doctest: +REPORT_UDIFF
    b'MIME-Version: 1.0'
    b'Content-Type: text/plain; charset="utf-8"'
    b'Content-Transfer-Encoding: base64'
    b'From: John <jdoe@a.com>'
    b'To: =?utf-8?b?zpbOtc+Nz4I=?= <z@olympus.org>'
    b'Subject: Homage'
    b''
    b'WW91J3JlIGdyZWF0LCDOls61z43PgiEK'
    b''

    Here's an 8-bit version:

    >>> config.set('DEFAULT', 'use-8bit', str(True))
    >>> message = get_message(
    ...     sender='John <jdoe@a.com>', recipient='Ζεύς <z@olympus.org>',
    ...     subject='Homage',
    ...     body="You're great, Ζεύς!\n",
    ...     content_type='plain',
    ...     config=config)
    >>> for line in _flatten(message).split(b'\n'):
    ...     print(line)  # doctest: +REPORT_UDIFF
    b'MIME-Version: 1.0'
    b'Content-Type: text/plain; charset="utf-8"'
    b'From: John <jdoe@a.com>'
    b'To: =?utf-8?b?zpbOtc+Nz4I=?= <z@olympus.org>'
    b'Subject: Homage'
    b'Content-Transfer-Encoding: 8bit'
    b''
    b"You're great, \xce\x96\xce\xb5\xcf\x8d\xcf\x82!"
    b''

    Here's an 8-bit version in UTF-16:

    >>> config.set('DEFAULT', 'encodings', 'US-ASCII, UTF-16-LE')
    >>> message = get_message(
    ...     sender='John <jdoe@a.com>', recipient='Ζεύς <z@olympus.org>',
    ...     subject='Homage',
    ...     body="You're great, Ζεύς!\n",
    ...     content_type='plain',
    ...     config=config)
    >>> for line in _flatten(message).split(b'\n'):
    ...     print(line)  # doctest: +REPORT_UDIFF
    b'MIME-Version: 1.0'
    b'Content-Type: text/plain; charset="utf-16-le"'
    b'From: John <jdoe@a.com>'
    b'To: =?utf-8?b?zpbOtc+Nz4I=?= <z@olympus.org>'
    b'Subject: Homage'
    b'Content-Transfer-Encoding: 8bit'
    b''
    b"\x00Y\x00o\x00u\x00'\x00r\x00e\x00 \x00g\x00r\x00e\x00a\x00t\x00,\x00 \x00\x96\x03\xb5\x03\xcd\x03\xc2\x03!\x00\n\x00"
    """
    bytesio = _io.BytesIO()
    generator = _BytesGenerator(bytesio)  # use policies for Python >=3.3
    try:
        generator.flatten(message)
    except UnicodeEncodeError as e:
        # HACK: work around deficiencies in BytesGenerator
        _LOG.warning(e)
        b = message.as_string().encode(str(message.get_charset()))
        m = _email.message_from_bytes(b)
        if not m:
            raise
        h = {k:_decode_header(v) for k,v in m.items()}
        head = {k:_decode_header(v) for k,v in message.items()}
        body = str(m.get_payload(decode=True), str(m.get_charsets()[0]))
        if (h == head and body == message.get_payload()):
            return b
        raise
    else:
        return bytesio.getvalue()

def sendmail_send(sender, recipient, message, config=None, section='DEFAULT'):
    if config is None:
        config = _config.CONFIG
    message_bytes = _flatten(message)
    sendmail = config.get(section, 'sendmail')
    sender_name,sender_addr = _parseaddr(sender)
    _LOG.debug(
        'sending message to {} via {}'.format(recipient, sendmail))
    try:
        p = _subprocess.Popen(
            [sendmail, '-F', sender_name, '-f', sender_addr, recipient],
            stdin=_subprocess.PIPE, stdout=_subprocess.PIPE,
            stderr=_subprocess.PIPE)
        stdout,stderr = p.communicate(message_bytes)
        status = p.wait()
        if status:
            raise _error.SendmailError(
                status=status, stdout=stdout, stderr=stderr)
    except Exception as e:
        raise _error.SendmailError() from e

def send(sender, recipient, message, config=None, section='DEFAULT'):
    protocol = config.get(section, 'email-protocol')
    if protocol == 'smtp':
        smtp_send(
            sender=sender, recipient=recipient, message=message,
            config=config, section=section)
    elif protocol == 'imap':
        imap_send(message=message, config=config, section=section)
    else:
        sendmail_send(
            sender=sender, recipient=recipient, message=message,
            config=config, section=section)