simple-login/app/email_utils.py

import os
from email.message import Message
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import make_msgid, formatdate, parseaddr, formataddr
from smtplib import SMTP
from typing import Optional

import dkim
from jinja2 import Environment, FileSystemLoader

from app.config import (
    SUPPORT_EMAIL,
    ROOT_DIR,
    POSTFIX_SERVER,
    NOT_SEND_EMAIL,
    DKIM_SELECTOR,
    DKIM_PRIVATE_KEY,
    DKIM_HEADERS,
    ALIAS_DOMAINS,
    SUPPORT_NAME,
    POSTFIX_SUBMISSION_TLS,
)
from app.log import LOG
from app.models import Mailbox, User


def render(template_name, **kwargs) -> str:
    templates_dir = os.path.join(ROOT_DIR, "templates", "emails")
    env = Environment(loader=FileSystemLoader(templates_dir))

    template = env.get_template(template_name)

    return template.render(**kwargs)


def send_welcome_email(user):
    send_email(
        user.email,
        f"Welcome to SimpleLogin {user.name}",
        render("com/welcome.txt", name=user.name, user=user),
        render("com/welcome.html", name=user.name, user=user),
    )


def send_trial_end_soon_email(user):
    send_email(
        user.email,
        f"Your trial will end soon {user.name}",
        render("transactional/trial-end.txt", name=user.name, user=user),
        render("transactional/trial-end.html", name=user.name, user=user),
    )


def send_activation_email(email, name, activation_link):
    send_email(
        email,
        f"Just one more step to join SimpleLogin {name}",
        render(
            "transactional/activation.txt",
            name=name,
            activation_link=activation_link,
            email=email,
        ),
        render(
            "transactional/activation.html",
            name=name,
            activation_link=activation_link,
            email=email,
        ),
    )


def send_reset_password_email(email, name, reset_password_link):
    send_email(
        email,
        f"Reset your password on SimpleLogin",
        render(
            "transactional/reset-password.txt",
            name=name,
            reset_password_link=reset_password_link,
        ),
        render(
            "transactional/reset-password.html",
            name=name,
            reset_password_link=reset_password_link,
        ),
    )


def send_change_email(new_email, current_email, name, link):
    send_email(
        new_email,
        f"Confirm email update on SimpleLogin",
        render(
            "transactional/change-email.txt",
            name=name,
            link=link,
            new_email=new_email,
            current_email=current_email,
        ),
        render(
            "transactional/change-email.html",
            name=name,
            link=link,
            new_email=new_email,
            current_email=current_email,
        ),
    )


def send_new_app_email(email, name):
    send_email(
        email,
        f"Any question/feedback for SimpleLogin {name}?",
        render("com/new-app.txt", name=name),
        render("com/new-app.html", name=name),
    )


def send_test_email_alias(email, name):
    send_email(
        email,
        f"This email is sent to {email}",
        render("transactional/test-email.txt", name=name, alias=email),
        render("transactional/test-email.html", name=name, alias=email),
    )


def send_cannot_create_directory_alias(user, alias, directory):
    """when user cancels their subscription, they cannot create alias on the fly.
    If this happens, send them an email to notify
    """
    send_email(
        user.email,
        f"Alias {alias} cannot be created",
        render(
            "transactional/cannot-create-alias-directory.txt",
            name=user.name,
            alias=alias,
            directory=directory,
        ),
        render(
            "transactional/cannot-create-alias-directory.html",
            name=user.name,
            alias=alias,
            directory=directory,
        ),
    )


def send_cannot_create_domain_alias(user, alias, domain):
    """when user cancels their subscription, they cannot create alias on the fly with custom domain.
    If this happens, send them an email to notify
    """
    send_email(
        user.email,
        f"Alias {alias} cannot be created",
        render(
            "transactional/cannot-create-alias-domain.txt",
            name=user.name,
            alias=alias,
            domain=domain,
        ),
        render(
            "transactional/cannot-create-alias-domain.html",
            name=user.name,
            alias=alias,
            domain=domain,
        ),
    )


def send_email(
    to_email, subject, plaintext, html=None, bounced_email: Optional[Message] = None
):
    if NOT_SEND_EMAIL:
        LOG.d(
            "send email with subject %s to %s, plaintext: %s",
            subject,
            to_email,
            plaintext,
        )
        return

    LOG.d("send email to %s, subject %s", to_email, subject)

    if POSTFIX_SUBMISSION_TLS:
        smtp = SMTP(POSTFIX_SERVER, 587)
        smtp.starttls()
    else:
        smtp = SMTP(POSTFIX_SERVER, 25)

    if bounced_email:
        msg = MIMEMultipart("mixed")

        # add email main body
        body = MIMEMultipart("alternative")
        body.attach(MIMEText(plaintext, "text"))
        if html:
            body.attach(MIMEText(html, "html"))

        msg.attach(body)

        # add attachment
        rfcmessage = MIMEBase("message", "rfc822")
        rfcmessage.attach(bounced_email)
        msg.attach(rfcmessage)
    else:
        msg = MIMEMultipart("alternative")
        msg.attach(MIMEText(plaintext, "text"))
        if html:
            msg.attach(MIMEText(html, "html"))

    msg["Subject"] = subject
    msg["From"] = f"{SUPPORT_NAME} <{SUPPORT_EMAIL}>"
    msg["To"] = to_email

    msg_id_header = make_msgid()
    msg["Message-ID"] = msg_id_header

    date_header = formatdate()
    msg["Date"] = date_header

    # add DKIM
    email_domain = SUPPORT_EMAIL[SUPPORT_EMAIL.find("@") + 1 :]
    add_dkim_signature(msg, email_domain)

    msg_raw = msg.as_string().encode()
    smtp.sendmail(SUPPORT_EMAIL, to_email, msg_raw)


def get_email_local_part(address):
    """
    Get the local part from email
    ab@cd.com -> ab
    """
    return address[: address.find("@")]


def get_email_domain_part(address):
    """
    Get the domain part from email
    ab@cd.com -> cd.com
    """
    return address[address.find("@") + 1 :]


def add_dkim_signature(msg: Message, email_domain: str):
    delete_header(msg, "DKIM-Signature")

    # Specify headers in "byte" form
    # Generate message signature
    sig = dkim.sign(
        msg.as_string().encode(),
        DKIM_SELECTOR,
        email_domain.encode(),
        DKIM_PRIVATE_KEY.encode(),
        include_headers=DKIM_HEADERS,
    )
    sig = sig.decode()

    # remove linebreaks from sig
    sig = sig.replace("\n", " ").replace("\r", "")
    msg["DKIM-Signature"] = sig[len("DKIM-Signature: ") :]


def add_or_replace_header(msg: Message, header: str, value: str):
    """
    Remove all occurrences of `header` and add `header` with `value`.
    """
    delete_header(msg, header)
    msg[header] = value


def delete_header(msg: Message, header: str):
    """a header can appear several times in message."""
    # inspired from https://stackoverflow.com/a/47903323/1428034
    for i in reversed(range(len(msg._headers))):
        header_name = msg._headers[i][0].lower()
        if header_name == header.lower():
            del msg._headers[i]


def delete_all_headers_except(msg: Message, headers: [str]):
    headers = [h.lower() for h in headers]

    for i in reversed(range(len(msg._headers))):
        header_name = msg._headers[i][0].lower()
        if header_name not in headers:
            del msg._headers[i]


def email_belongs_to_alias_domains(address: str) -> bool:
    """return True if an email ends with one of the alias domains provided by SimpleLogin"""
    for domain in ALIAS_DOMAINS:
        if address.endswith("@" + domain):
            return True

    return False


def can_be_used_as_personal_email(email: str) -> bool:
    """return True if an email can be used as a personal email. Currently the only condition is email domain is not
    - one of ALIAS_DOMAINS
    - one of custom domains
    """
    domain = get_email_domain_part(email)
    if not domain:
        return False

    if domain in ALIAS_DOMAINS:
        return False

    from app.models import CustomDomain

    if CustomDomain.get_by(domain=domain, verified=True):
        return False

    return True


def email_already_used(email: str) -> bool:
    """test if an email can be used when:
    - user signs up
    - add a new mailbox
    """
    if User.get_by(email=email):
        return True

    if Mailbox.get_by(email=email):
        return True

    return False


def mailbox_already_used(email: str, user) -> bool:
    if Mailbox.get_by(email=email):
        return True

    # support the case user wants to re-add their real email as mailbox
    # can happen when user changes their root email and wants to add this new email as mailbox
    if email == user.email:
        return False

    if User.get_by(email=email):
        return True

    return False


def get_orig_message_from_bounce(msg: Message) -> Message:
    """parse the original email from Bounce"""
    i = 0
    for part in msg.walk():
        i += 1

        # the original message is the 4th part
        # 1st part is the root part,  multipart/report
        # 2nd is text/plain, Postfix log
        # ...
        # 7th is original message
        if i == 7:
            return part


def new_addr(old_addr, new_email) -> str:
    """replace First Last <first@example.com> by
    first@example.com by SimpleLogin <new_email>

    `new_email` is a special reply address
    """
    name, old_email = parseaddr(old_addr)
    new_name = f"{old_email} via SimpleLogin"
    new_addr = formataddr((new_name, new_email)).strip()

    return new_addr.strip()


def get_addrs_from_header(msg: Message, header) -> [str]:
    """Get all addresses contained in `header`
    Used for To or CC header.
    """
    ret = []
    header_content = msg.get_all(header)
    if not header_content:
        return ret

    for addrs in header_content:
        for addr in addrs.split(","):
            ret.append(addr.strip())

    # do not return empty string
    return [r for r in ret if r]