app-MAIL-temp/app/handler/unsubscribe_generator.py

import urllib
from email.header import Header
from email.message import Message

from app.email import headers
from app.email_utils import add_or_replace_header, delete_header
from app.handler.unsubscribe_encoder import (
    UnsubscribeEncoder,
    UnsubscribeAction,
    UnsubscribeData,
    UnsubscribeOriginalData,
)
from app.log import LOG
from app.models import Alias, Contact, UnsubscribeBehaviourEnum


class UnsubscribeGenerator:
    def _generate_header_with_original_behaviour(
        self, alias: Alias, message: Message
    ) -> Message:
        """
        Generate a header that will encode the original unsub request. To do so
         1. Look if there's an original List_Unsubscribe headers, otherwise do nothing
         2. Header has the form <method1>, <method2>, .. where each method is either
           - mailto:s@b.c?subject=something
           - http(s)://somewhere.com
        3. Check if there are http unsub requests in the header. If there are, reserve them and remove all mailto
           methods to avoid leaking the real mailbox. We forward the message with only http(s) methods.
        4. If there aren't neither https nor mailto methods, strip the header from the message and that's it.
           It could happen if the header is malformed.
        5. Encode in our unsub request the first original mail and subject to unsub, and use that as our unsub header.
        """
        unsubscribe_data = message[headers.LIST_UNSUBSCRIBE]
        if not unsubscribe_data:
            LOG.info("Email has no unsubscribe header")
            return message
        if isinstance(unsubscribe_data, Header):
            unsubscribe_data = str(unsubscribe_data.encode())
        raw_methods = [method.strip() for method in unsubscribe_data.split(",")]
        mailto_unsubs = None
        other_unsubs = []
        for raw_method in raw_methods:
            start = raw_method.find("<")
            end = raw_method.rfind(">")
            if start == -1 or end == -1 or start >= end:
                continue
            method = raw_method[start + 1 : end]
            url_data = urllib.parse.urlparse(method)
            if url_data.scheme == "mailto":
                query_data = urllib.parse.parse_qs(url_data.query)
                mailto_unsubs = (url_data.path, query_data.get("subject", [""])[0])
                LOG.debug(f"Unsub is mailto to {mailto_unsubs}")
            else:
                LOG.debug(f"Unsub has {url_data.scheme} scheme")
                other_unsubs.append(method)
        # If there are non mailto unsubscribe methods, use those in the header
        if other_unsubs:
            add_or_replace_header(
                message,
                headers.LIST_UNSUBSCRIBE,
                ", ".join([f"<{method}>" for method in other_unsubs]),
            )
            add_or_replace_header(
                message, headers.LIST_UNSUBSCRIBE_POST, "List-Unsubscribe=One-Click"
            )
            LOG.debug(f"Adding click unsub methods to header {other_unsubs}")
            return message
        elif not mailto_unsubs:
            LOG.debug("No unsubs. Deleting all unsub headers")
            delete_header(message, headers.LIST_UNSUBSCRIBE)
            delete_header(message, headers.LIST_UNSUBSCRIBE_POST)
            return message
        unsub_data = UnsubscribeData(
            UnsubscribeAction.OriginalUnsubscribeMailto,
            UnsubscribeOriginalData(alias.id, mailto_unsubs[0], mailto_unsubs[1]),
        )
        LOG.debug(f"Adding unsub data {unsub_data}")
        return self._add_unsubscribe_header(message, unsub_data)

    def _add_unsubscribe_header(
        self, message: Message, unsub: UnsubscribeData
    ) -> Message:
        unsub_link = UnsubscribeEncoder.encode(unsub.action, unsub.data)

        add_or_replace_header(message, headers.LIST_UNSUBSCRIBE, f"<{unsub_link.link}>")
        if not unsub_link.via_email:
            add_or_replace_header(
                message, headers.LIST_UNSUBSCRIBE_POST, "List-Unsubscribe=One-Click"
            )
        return message

    def add_header_to_message(
        self, alias: Alias, contact: Contact, message: Message
    ) -> Message:
        """
        Add List-Unsubscribe header based on the user preference.
        """
        unsub_behaviour = alias.user.unsub_behaviour
        if unsub_behaviour == UnsubscribeBehaviourEnum.PreserveOriginal:
            return self._generate_header_with_original_behaviour(alias, message)
        elif unsub_behaviour == UnsubscribeBehaviourEnum.DisableAlias:
            unsub = UnsubscribeData(UnsubscribeAction.DisableAlias, alias.id)
            return self._add_unsubscribe_header(message, unsub)
        else:
            unsub = UnsubscribeData(UnsubscribeAction.DisableContact, contact.id)
            return self._add_unsubscribe_header(message, unsub)