devday_mailbot.py

#!/usr/bin/env python3
#
# Copyright Dev Day Dresden e.V.
#
# This program 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 3 of the License, or
# (at your option) any later version.
#
# This program 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 this program.  If not, see <https://www.gnu.org/licenses/>.
import logging
import os
import re
from email.header import decode_header
from email.message import EmailMessage
from email.parser import BytesParser
from email.utils import make_msgid, parseaddr
from imaplib import IMAP4, IMAP4_SSL
from smtplib import SMTP
from urllib.parse import urlparse

from requests import get


def fetch_latest_mails(imap_conn: IMAP4) -> list[bytes]:
    imap_conn.select()
    typ, data = imap_conn.search(None, "NOT SEEN")
    if typ != "OK":
        raise RuntimeError("Error fetching latest mails")
    result = []

    for num in data[0].split():
        typ, message = imap_conn.fetch(num, "(RFC822)")
        if typ != "OK":
            raise RuntimeError(f"Error fetching mail {num}")
        print("Fetched mail", num.decode("utf-8"))
        result.append(message[0][1])

    return result


class AddressSource:
    def get_addresses(self) -> list[str]:
        raise NotImplementedError("must be implemented in subclass")


class FileAddressSource(AddressSource):
    def __init__(self, filename: str):
        self.filename = filename

    def get_addresses(self) -> list[str]:
        with open(self.filename, "r", encoding="utf-8") as sourcefile:
            return sourcefile.readlines()


class BasePretixAddressSource(AddressSource):
    def __init__(self, token, netloc, path):
        if len(path) < 2:
            raise ValueError("path is too short")
        if path.count("/") < 1:
            raise ValueError("invalid path")
        self.token, self.netloc, self.organizer = token, netloc, path[1:].split("/")[0]


class PretixCustomerAddressSource(BasePretixAddressSource):
    def get_addresses(self) -> list[str]:
        addresses = []
        url = f"https://{self.netloc}/api/v1/organizers/{self.organizer}/customers/"
        while True:
            r = get(url, headers={"Authorization": f"Token {self.token}"})
            r.raise_for_status()
            json_data = r.json()
            addresses.extend([
                c["email"].lower()
                for c in json_data["results"]
                if c["is_active"] and c["is_verified"]
            ])
            if not json_data["next"]:
                break
            url = json_data["next"]
        return addresses


class PretixOrderAddressSource(BasePretixAddressSource):
    def __init__(self, token, netloc, path):
        if len(path) < 4:
            raise ValueError("path is too short")
        if path.count("/") != 2:
            raise ValueError("invalid path")
        super().__init__(token, netloc, path)
        self.event = path[1:].split("/")[1]

    def get_addresses(self) -> list[str]:
        addresses = []
        url = f"https://{self.netloc}/api/v1/organizers/{self.organizer}/events/{self.event}/orders/?status=p"
        while True:
            r = get(url, headers={"Authorization": f"Token {self.token}"})
            r.raise_for_status()
            json_data = r.json()
            for order in json_data["results"]:
                for position in order["positions"]:
                    if position["attendee_email"]:
                        addresses.append(position["attendee_email"])
                    addresses.append(order["email"])
            if not json_data["next"]:
                break
            url = json_data["next"]
        return list(sorted(set(addresses)))


def determine_address_source(src_url: str) -> AddressSource:
    url = urlparse(src_url)

    if url.scheme == "file":
        return FileAddressSource(url.path)

    if url.scheme == "pretix":
        return PretixCustomerAddressSource(url.password, url.hostname, url.path)

    if url.scheme == "pretix-orders":
        return PretixOrderAddressSource(url.password, url.hostname, url.path)

    raise ValueError(f"unsupported URL scheme {url.scheme}")


def process_mail(
        smtp_conn: SMTP,
        address_source: AddressSource,
        sender_address: str,
        default_to: str,
        default_reply_to: str,
        valid_sender_patterns: list[str],
        mail: bytes,
):
    email_parser = BytesParser()
    mail_data = email_parser.parsebytes(mail)

    if "From" not in mail_data:
        return

    from_address = mail_data.get("From")
    _, from_email = parseaddr(from_address)

    for pattern in valid_sender_patterns:
        if re.match(pattern, from_email):
            logging.info(
                "mail From address value '%s' matched valid recipient pattern",
                from_email,
            )

            break
    else:
        logging.warning("skipping mail from invalid from address %s", from_email)

        return

    _, sender_email = parseaddr(sender_address)
    sender_domain = sender_email.split("@")[1]

    new_message = EmailMessage()

    for h in ("Subject", "Date", "Reply-To"):
        if h in mail_data:
            original_header = decode_header(mail_data.get(h))
            for value, encoding in original_header:
                if encoding is None:
                    new_message.add_header(h, value)
                else:
                    new_message.add_header(h, value.decode(encoding))

    new_message.add_header("From", sender_address)
    new_message.add_header("To", default_to)
    new_message.add_header("Content-Type", mail_data.get_content_type())

    if "Reply-To" in mail_data:
        logging.debug("set reply-to address to %s", mail_data.get("Reply-To"))
    else:
        new_message.add_header("Reply-To", default_reply_to)

    new_message.set_payload(mail_data.get_payload())
    new_message["Message-Id"] = make_msgid(domain=sender_domain)

    recipient_addresses = address_source.get_addresses()

    logging.info(
        "distributing mail with subject '%s' and message id %s to %d recipients",
        new_message.get("Subject"),
        new_message.get("Message-Id"),
        len(recipient_addresses),
    )

    smtp_conn.send_message(new_message, sender_address, recipient_addresses)


def main():
    logging.basicConfig(
        level=logging.getLevelNamesMapping().get(
            os.getenv("MAILBOT_LOG_LEVEL", "INFO")
        ),
        format="%(asctime)s %(levelname)s %(message)s",
    )

    imap_host = os.getenv("MAILBOT_IMAP_HOST", "localhost")
    imap_port = int(os.getenv("MAILBOT_IMAP_PORT", "143"))
    imap_user = os.getenv("MAILBOT_IMAP_USER")
    imap_password = os.getenv("MAILBOT_IMAP_PASSWORD")

    smtp_host = os.getenv("MAILBOT_SMTP_HOST", "localhost")
    smtp_port = int(os.getenv("MAILBOT_SMTP_PORT", "25"))
    smtp_user = os.getenv("MAILBOT_SMTP_USER")
    smtp_password = os.getenv("MAILBOT_SMTP_PASSWORD")

    sender_address = os.getenv("MAILBOT_SENDER_ADDRESS", "Mailbot <info@example.org")
    address_source_url = os.getenv("MAILBOT_ADDRESS_SOURCE")
    default_to = os.getenv("MAILBOT_DEFAULT_TO")
    default_reply_to = os.getenv("MAILBOT_DEFAULT_REPLY_TO")
    valid_sender_patterns = os.getenv("MAILBOT_VALID_SENDER_PATTERNS").split(",")
    address_source = determine_address_source(address_source_url)

    logging.info("devday_mailbot started")

    with IMAP4_SSL(imap_host, imap_port) as imap_conn:
        try:
            imap_conn.login(imap_user, imap_password)
            logging.debug("successfully logged in to IMAP server")
            mails = fetch_latest_mails(imap_conn)
        finally:
            imap_conn.close()
            imap_conn.logout()

    if not mails:
        logging.info("no mails to distribute")

        return

    logging.info("fetched %d mail(s) from IMAP mailbox", len(mails))

    with SMTP(smtp_host, smtp_port) as smtp_conn:
        smtp_conn.ehlo_or_helo_if_needed()
        smtp_conn.starttls()
        smtp_conn.login(smtp_user, smtp_password)

        logging.debug("successfully logged in to SMTP server")

        for mail in mails:
            process_mail(
                smtp_conn,
                address_source,
                sender_address,
                default_to,
                default_reply_to,
                valid_sender_patterns,
                mail,
            )


if __name__ == "__main__":
    main()