python-mylib/mylib/email.py

""" Email client to forge and send emails """

import base64
import email.utils
import logging
import os
import smtplib
from email.encoders import encode_base64
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

import magic
from mako.template import Template as MakoTemplate

from mylib.config import (
    BooleanOption,
    ConfigurableObject,
    IntegerOption,
    PasswordOption,
    StringOption,
)

log = logging.getLogger(__name__)


def load_image_as_base64(path):
    """Load image file as base64"""
    log.debug("Load image file '%s'", path)
    with open(path, "rb") as file_desc:
        data = file_desc.read()
    return f"data:{magic.from_buffer(data, mime=True)};base64, {base64.b64encode(data).decode()}"


class EmailClient(
    ConfigurableObject
):  # pylint: disable=useless-object-inheritance,too-many-instance-attributes
    """
    Email client

    This class abstract all interactions with the SMTP server.
    """

    _config_name = "email"
    _config_comment = "Email"
    _defaults = {
        "smtp_host": "localhost",
        "smtp_port": 25,
        "smtp_ssl": False,
        "smtp_tls": False,
        "smtp_user": None,
        "smtp_password": None,
        "smtp_debug": False,
        "sender_name": "No reply",
        "sender_email": "noreply@localhost",
        "encoding": "utf-8",
        "catch_all_addr": None,
        "just_try": False,
        "templates_path": None,
    }

    templates = {}

    def __init__(self, templates=None, initialize=False, **kwargs):
        super().__init__(**kwargs)

        assert templates is None or isinstance(templates, dict)
        self.templates = templates if templates else {}
        if initialize:
            self.initialize()

    # pylint: disable=arguments-differ,arguments-renamed
    def configure(self, use_smtp=True, **kwargs):
        """Configure options on registered mylib.Config object"""
        section = super().configure(
            just_try_help=kwargs.pop("just_try_help", "Just-try mode: do not really send emails"),
            **kwargs,
        )

        if use_smtp:
            section.add_option(
                StringOption,
                "smtp_host",
                default=self._defaults["smtp_host"],
                comment="SMTP server hostname/IP address",
            )
            section.add_option(
                IntegerOption,
                "smtp_port",
                default=self._defaults["smtp_port"],
                comment="SMTP server port",
            )
            section.add_option(
                BooleanOption,
                "smtp_ssl",
                default=self._defaults["smtp_ssl"],
                comment="Use SSL on SMTP server connection",
            )
            section.add_option(
                BooleanOption,
                "smtp_tls",
                default=self._defaults["smtp_tls"],
                comment="Use TLS on SMTP server connection",
            )
            section.add_option(
                StringOption,
                "smtp_user",
                default=self._defaults["smtp_user"],
                comment="SMTP authentication username",
            )
            section.add_option(
                PasswordOption,
                "smtp_password",
                default=self._defaults["smtp_password"],
                comment='SMTP authentication password (set to "keyring" to use XDG keyring)',
                username_option="smtp_user",
                keyring_value="keyring",
            )
            section.add_option(
                BooleanOption,
                "smtp_debug",
                default=self._defaults["smtp_debug"],
                comment="Enable SMTP debugging",
            )

        section.add_option(
            StringOption,
            "sender_name",
            default=self._defaults["sender_name"],
            comment="Sender name",
        )
        section.add_option(
            StringOption,
            "sender_email",
            default=self._defaults["sender_email"],
            comment="Sender email address",
        )
        section.add_option(
            StringOption, "encoding", default=self._defaults["encoding"], comment="Email encoding"
        )
        section.add_option(
            StringOption,
            "catch_all_addr",
            default=self._defaults["catch_all_addr"],
            comment="Catch all sent emails to this specified email address",
        )

        section.add_option(
            StringOption,
            "templates_path",
            default=self._defaults["templates_path"],
            comment="Path to templates directory",
        )

        return section

    def initialize(self, *args, **kwargs):  # pylint: disable=arguments-differ
        """Configuration initialized hook"""
        super().initialize(*args, **kwargs)
        self.load_templates_directory()

    def load_templates_directory(self, templates_path=None):
        """Load templates from specified directory"""
        if templates_path is None:
            templates_path = self._get_option("templates_path")
        if not templates_path:
            return
        log.debug("Load email templates from %s directory", templates_path)
        for filename in os.listdir(templates_path):
            filepath = os.path.join(templates_path, filename)
            if not os.path.isfile(filepath):
                continue
            template_name, template_type = os.path.splitext(filename)
            if template_type not in [".html", ".txt", ".subject"]:
                continue
            template_type = "text" if template_type == ".txt" else template_type[1:]
            if template_name not in self.templates:
                self.templates[template_name] = {"path": templates_path}
            log.debug("Load email template %s %s from %s", template_name, template_type, filepath)
            with open(filepath, encoding="utf8") as file_desc:
                self.templates[template_name][template_type] = MakoTemplate(
                    file_desc.read()
                )  # nosec

    def forge_message(
        self,
        recipients,
        subject=None,
        html_body=None,
        text_body=None,  # pylint: disable=too-many-arguments,too-many-locals
        attachment_files=None,
        attachment_payloads=None,
        sender_name=None,
        sender_email=None,
        encoding=None,
        template=None,
        cc=None,
        **template_vars,
    ):
        """
        Forge a message

        :param recipients:          The recipient(s) of the email. List of tuple(name, email) or
                                    just the email of the recipients.
        :param subject:             The subject of the email.
        :param html_body:           The HTML body of the email
        :param text_body:           The plain text body of the email
        :param attachment_files:    List of filepaths to attach
        :param attachment_payloads: List of tuples with filename and payload to attach
        :param sender_name:         Custom sender name (default: as defined on initialization)
        :param sender_email:        Custom sender email (default: as defined on initialization)
        :param encoding:            Email content encoding (default: as defined on initialization)
        :param template:            The name of a template to use to forge this email
        :param cc:                  Optional list of CC recipient addresses.
                                    List of tuple(name, email) or just the email of the recipients.

        All other parameters will be consider as template variables.
        """
        recipients = [recipients] if not isinstance(recipients, list) else recipients
        msg = MIMEMultipart("alternative")
        msg["To"] = ", ".join(
            [
                email.utils.formataddr(recipient) if isinstance(recipient, tuple) else recipient
                for recipient in recipients
            ]
        )

        if cc:
            cc = [cc] if not isinstance(cc, list) else cc
            msg["Cc"] = ", ".join(
                [
                    email.utils.formataddr(recipient) if isinstance(recipient, tuple) else recipient
                    for recipient in cc
                ]
            )

        msg["From"] = email.utils.formataddr(
            (
                sender_name or self._get_option("sender_name"),
                sender_email or self._get_option("sender_email"),
            )
        )
        if subject:
            msg["Subject"] = (
                subject.render(**template_vars)
                if isinstance(subject, MakoTemplate)
                else subject.format(**template_vars)
            )
        msg["Date"] = email.utils.formatdate(None, True)
        encoding = encoding if encoding else self._get_option("encoding")
        if template:
            log.debug("Forge email from template %s", template)
            assert template in self.templates, f"Unknown template {template}"
            # Handle subject from template
            if not subject:
                assert self.templates[template].get(
                    "subject"
                ), f"No subject defined in template {template}"
                msg["Subject"] = (
                    self.templates[template]["subject"].render(**template_vars)
                    if isinstance(self.templates[template]["subject"], MakoTemplate)
                    else self.templates[template]["subject"].format(**template_vars)
                )

            # Put HTML part in last one to preferred it
            parts = []
            if self.templates[template].get("text"):
                if isinstance(self.templates[template]["text"], MakoTemplate):
                    parts.append(
                        (self.templates[template]["text"].render(**template_vars), "plain")
                    )
                else:
                    parts.append(
                        (self.templates[template]["text"].format(**template_vars), "plain")
                    )
            if self.templates[template].get("html"):
                if isinstance(self.templates[template]["html"], MakoTemplate):
                    template_vars["load_image_as_base64"] = self.template_image_loader(
                        self.templates[template].get("path")
                    )
                    parts.append((self.templates[template]["html"].render(**template_vars), "html"))
                else:
                    parts.append((self.templates[template]["html"].format(**template_vars), "html"))

            for body, mime_type in parts:
                msg.attach(MIMEText(body.encode(encoding), mime_type, _charset=encoding))
        else:
            assert subject, "No subject provided"
            if text_body:
                msg.attach(MIMEText(text_body.encode(encoding), "plain", _charset=encoding))
            if html_body:
                msg.attach(MIMEText(html_body.encode(encoding), "html", _charset=encoding))
        if attachment_files:
            for filepath in attachment_files:
                with open(filepath, "rb") as fp:
                    part = MIMEBase("application", "octet-stream")
                    part.set_payload(fp.read())
                    encode_base64(part)
                    part.add_header(
                        "Content-Disposition",
                        f'attachment; filename="{os.path.basename(filepath)}"',
                    )
                    msg.attach(part)
        if attachment_payloads:
            for filename, payload in attachment_payloads:
                part = MIMEBase("application", "octet-stream")
                part.set_payload(payload)
                encode_base64(part)
                part.add_header("Content-Disposition", f'attachment; filename="{filename}"')
                msg.attach(part)
        return msg

    @staticmethod
    def template_image_loader(directory_path):
        """Return wrapper for the load_image_as_base64 function bind on the template directory"""

        def _load_image_as_base64(path):
            return load_image_as_base64(
                os.path.join(directory_path, path)
                if directory_path and not os.path.isabs(path)
                else path
            )

        return _load_image_as_base64

    def send(
        self, recipients, msg=None, subject=None, just_try=None, cc=None, bcc=None, **forge_args
    ):
        """
        Send an email

        :param recipients:  The recipient(s) of the email. List of tuple(name, email) or
                            just the email of the recipients.
        :param msg:         The message of this email (as MIMEBase or derivated classes)
        :param subject:     The subject of the email (only if the message is not provided
                            using msg parameter)
        :param just_try:    Enable just try mode (do not really send email, default: as defined on
                            initialization)
        :param cc:          Optional list of CC recipient addresses. List of tuple(name, email) or
                            just the email of the recipients.
        :param bcc:         Optional list of BCC recipient addresses. List of tuple(name, email) or
                            just the email of the recipients.

        All other parameters will be consider as parameters to forge the message
        (only if the message is not provided using msg parameter).
        """
        recipients = [recipients] if not isinstance(recipients, list) else recipients
        msg = msg if msg else self.forge_message(recipients, subject, cc=cc, **forge_args)
        catch_addr = self._get_option("catch_all_addr")
        if catch_addr:
            log.debug(
                "Catch email originally send to %s (CC:%s, BCC:%s) to %s",
                ", ".join(recipients),
                ", ".join(cc) if isinstance(cc, list) else cc,
                ", ".join(bcc) if isinstance(bcc, list) else bcc,
                catch_addr,
            )
            recipients = catch_addr if isinstance(catch_addr, list) else [catch_addr]
        else:
            if cc:
                recipients.extend(
                    [
                        recipient[1] if isinstance(recipient, tuple) else recipient
                        for recipient in (cc if isinstance(cc, list) else [cc])
                    ]
                )
            if bcc:
                recipients.extend(
                    [
                        recipient[1] if isinstance(recipient, tuple) else recipient
                        for recipient in (bcc if isinstance(bcc, list) else [bcc])
                    ]
                )

        if just_try if just_try is not None else self._just_try:
            log.debug(
                'Just-try mode: do not really send this email to %s (subject="%s")',
                ", ".join(recipients),
                subject or msg.get("subject", "No subject"),
            )
            return True

        smtp_host = self._get_option("smtp_host")
        smtp_port = self._get_option("smtp_port")
        try:
            if self._get_option("smtp_ssl"):
                logging.info("Establish SSL connection to server %s:%s", smtp_host, smtp_port)
                server = smtplib.SMTP_SSL(smtp_host, smtp_port)
            else:
                logging.info("Establish connection to server %s:%s", smtp_host, smtp_port)
                server = smtplib.SMTP(smtp_host, smtp_port)
                if self._get_option("smtp_tls"):
                    logging.info("Start TLS on SMTP connection")
                    server.starttls()
        except smtplib.SMTPException:
            log.error("Error connecting to SMTP server %s:%s", smtp_host, smtp_port, exc_info=True)
            return False

        if self._get_option("smtp_debug"):
            server.set_debuglevel(True)

        smtp_user = self._get_option("smtp_user")
        smtp_password = self._get_option("smtp_password")
        if smtp_user and smtp_password:
            try:
                log.info("Try to authenticate on SMTP connection as %s", smtp_user)
                server.login(smtp_user, smtp_password)
            except smtplib.SMTPException:
                log.error(
                    "Error authenticating on SMTP server %s:%s with user %s",
                    smtp_host,
                    smtp_port,
                    smtp_user,
                    exc_info=True,
                )
                return False

        error = False
        try:
            log.info("Sending email to %s", ", ".join(recipients))
            server.sendmail(
                self._get_option("sender_email"),
                [
                    recipient[1] if isinstance(recipient, tuple) else recipient
                    for recipient in recipients
                ],
                msg.as_string(),
            )
        except smtplib.SMTPException:
            error = True
            log.error("Error sending email to %s", ", ".join(recipients), exc_info=True)
        finally:
            server.quit()

        return not error