Remove mail footer

.github/workflows/test.yaml

@@ -42,7 +42,7 @@ jobs:
               run: sudo apt install gnupg2 -y && mkdir ~/.gnupg
             - uses: Gr1N/setup-poetry@v8
             - name: Install packages
-              run: poetry install
+              run: poetry install --no-interaction
             - uses: Gr1N/setup-poetry@v8
             - name: Apply migrations
               run: poetry run alembic upgrade head

Dockerfile

@@ -19,11 +19,10 @@ RUN mkdir ~/.gnupg
 RUN mkdir /app
 RUN mkdir /tutorial
 
-WORKDIR /app
-
 # Install poetry
 RUN pip3 install poetry
 
+WORKDIR /app
 COPY pyproject.toml /app
 
 # Install dependencies

app/controllers/server_statistics.py

@@ -3,7 +3,6 @@
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.orm import Session
 
-from app import life_constants
 from app.controllers import global_settings as settings
 
 from app.models import ServerStatistics

app/models/alias.py

@@ -105,6 +105,11 @@ class EmailAlias(Base, IDMixin, ModelPreference):
             default=None,
             nullable=True,
         )
+        pref_remove_footer = sa.Column(
+            sa.Boolean,
+            default=None,
+            nullable=True,
+        )
 
     def _get_user_preference_prefix(self) -> str:
         return "alias_"
@@ -143,6 +148,10 @@ def proxy_user_agent(self) -> ProxyUserAgentType:
     def expand_url_shorteners(self) -> bool:
         return self.get_preference_value("expand_url_shorteners")
 
+    @property
+    def remove_footer(self) -> bool:
+        return self.get_preference_value("remove_footer")
+
 
 class DeletedEmailAlias(Base, IDMixin, CreationMixin):
     """Store all deleted alias to make sure they will not be reused, so that new owner won't

app/models/user_preferences.py

@@ -27,6 +27,7 @@ class UserPreferences(Base, IDMixin):
         alias_image_proxy_format: ImageProxyFormatType
         alias_proxy_user_agent: ProxyUserAgentType
         alias_expand_url_shorteners: bool
+        alias_remove_footer: bool
     else:
         user_id = sa.Column(
             UUID(as_uuid=True),
@@ -57,3 +58,7 @@ class UserPreferences(Base, IDMixin):
             sa.Boolean,
             default=False,
         )
+        alias_remove_footer = sa.Column(
+            sa.Boolean,
+            default=False,
+        )

docker-compose.yaml

@@ -27,7 +27,6 @@ services:
       context: .
 
     ports:
-      - "8000:80"
       - "25:25"
       - "587:587"
 

email_utils/bounce_messages.py

@@ -20,6 +20,7 @@
     "is_bounce",
     "is_not_deliverable",
     "get_report_from_message",
+    "extract_forward_status_header",
 ]
 
 

email_utils/html_handler.py → email_utils/content_handler.py

@@ -1,11 +1,11 @@
 import lxml.html
 import requests
+import talon
 from lxml.etree import _Element, XMLSyntaxError
 from pyquery import PyQuery as pq
 from sqlalchemy.orm import Session
 
 from app.controllers.image_proxy import create_image_proxy
-from app.controllers.server_statistics import add_removed_trackers
 from app.email_report_data import (
     EmailReportData, EmailReportExpandedURLData, EmailReportProxyImageData,
     EmailReportSinglePixelImageTrackerData,
@@ -17,7 +17,8 @@
 __all__ = [
     "convert_images",
     "remove_single_pixel_image_trackers",
-    "expand_shortened_urls"
+    "expand_shortened_urls",
+    "remove_footer",
 ]
 
 
@@ -119,3 +120,12 @@ def expand_shortened_urls(
         )
 
     return d.outer_html()
+
+
+def remove_footer(
+    text: str,
+    content_type: str
+) -> str:
+    content, signature = talon.quotations.extract_from(text, content_type)
+
+    return content

email_utils/handle_local_to_outside.py

@@ -0,0 +1,98 @@
+from email.message import Message
+
+from aiosmtpd.smtp import Envelope
+from sqlalchemy.exc import NoResultFound
+from sqlalchemy.orm import Session
+
+from app import logger
+from app.controllers import server_statistics
+from app.controllers.alias import get_alias_by_local_and_domain
+from app.controllers.email import get_email_by_address
+from app.controllers.reserved_alias import get_reserved_alias_by_address
+from app.utils.email import normalize_email
+from email_utils import headers
+from email_utils.bounce_messages import generate_forward_status, StatusType
+from email_utils.errors import AliasNotYoursError
+from email_utils.headers import set_header
+from email_utils.send_mail import send_mail
+from email_utils.validators import validate_alias
+
+
+__all__ = [
+    "handle_local_to_outside"
+]
+
+
+async def handle_local_to_outside(
+    db: Session,
+    /,
+    alias_address: str,
+    target: str,
+    envelope: Envelope,
+    message: Message,
+    message_id: str
+) -> None:
+    # LOCALLY saved user wants to send a mail FROM alias TO the outside.
+    from_mail = await normalize_email(envelope.mail_from)
+    logger.info(
+        f"{envelope.rcpt_tos[0]} is an forward alias address (LOCAL wants to send to "
+        f"OUTSIDE). It should be sent to {target} via alias {alias_address} "
+        f"Checking if FROM {from_mail} user owns it."
+    )
+
+    try:
+        email = get_email_by_address(db, from_mail)
+    except NoResultFound:
+        logger.info(f"User does not exist. Raising error.")
+        # Return "AliasNotYoursError" to avoid an alias being leaked
+        raise AliasNotYoursError()
+
+    logger.info(f"Checking if user owns the given alias.")
+    user = email.user
+
+    local, domain = alias_address.split("@")
+
+    if (alias := get_reserved_alias_by_address(db, local, domain)) is not None:
+        logger.info("Alias is a reserved alias.")
+
+        # Reserved alias
+        if user not in alias.users:
+            logger.info("User is not in reserved alias users. Raising error.")
+            raise AliasNotYoursError()
+    else:
+        # Local alias
+        try:
+            alias = get_alias_by_local_and_domain(db, local=local, domain=domain)
+        except NoResultFound:
+            logger.info("Alias does not exist. Raising error.")
+            raise AliasNotYoursError()
+
+        if user != alias.user:
+            logger.info("User does not own the alias. Raising error.")
+            raise AliasNotYoursError()
+
+    logger.info("Checking if alias is valid.")
+    validate_alias(alias)
+    logger.info("Alias is valid.")
+
+    logger.info(
+        f"Local mail {alias.address} should be relayed to outside mail {target}. "
+        f"Sending email now..."
+    )
+
+    set_header(
+        message,
+        headers.KLECK_FORWARD_STATUS,
+        generate_forward_status(
+            StatusType.FORWARD_ALIAS_TO_OUTSIDE,
+            outside_address=target,
+            message_id=message_id,
+        )
+    )
+
+    send_mail(
+        message,
+        from_mail=alias.address,
+        to_mail=target,
+    )
+    server_statistics.add_sent_email(db)

email_utils/handle_outside_to_local.py

@@ -0,0 +1,161 @@
+from email.message import Message
+
+from aiosmtpd.smtp import Envelope
+from sqlalchemy.orm import Session
+
+from app.controllers import server_statistics
+from app.controllers import global_settings as settings
+from app import logger
+from app.controllers.email_report import create_email_report
+from app.email_report_data import EmailReportData
+from app.models import EmailAlias
+from email_utils import headers
+from email_utils.bounce_messages import generate_forward_status, StatusType
+from email_utils.content_handler import (
+    convert_images, expand_shortened_urls,
+    remove_footer, remove_single_pixel_image_trackers,
+)
+from email_utils.headers import set_header
+from email_utils.send_mail import send_mail
+from email_utils.utils import get_header_unicode
+from email_utils.validators import validate_alias
+
+
+__all__ = [
+    "handle_outside_to_local"
+]
+
+
+def handle_outside_to_local(
+    db: Session,
+    /,
+    envelope: Envelope,
+    message: Message,
+    alias: EmailAlias,
+    message_id: str,
+) -> None:
+    logger.info("Mail is an alias mail (OUTSIDE wants to send to LOCAL).")
+
+    logger.info("Checking if alias is valid.")
+
+    # OUTSIDE user wants to send a mail TO a locally saved user's private mail.
+    validate_alias(alias)
+
+    logger.info("Alias is valid.")
+
+    report = EmailReportData(
+        mail_from=envelope.mail_from,
+        mail_to=alias.address,
+        subject=get_header_unicode(message[headers.SUBJECT]),
+        message_id=message[headers.MESSAGE_ID],
+    )
+
+    content = message.get_payload()
+
+    if type(content) is str:
+        content_type = message.get_content_type()
+
+        if content_type == "text/html":
+            content = parse_html(
+                db,
+                alias=alias,
+                report=report,
+                content=content,
+            )
+
+            message.set_payload(content, "utf-8")
+        elif content_type == "text/plain":
+            content = parse_text(
+                alias=alias,
+                content=content,
+            )
+
+            message.set_payload(content, "utf-8")
+    elif type(content) is list:
+        content_maps = {
+            part.get_content_type(): part.get_payload()
+            for part in message.walk()
+        }
+
+        if "text/html" in content_maps:
+            content_maps["text/html"] = parse_html(
+                db,
+                alias=alias,
+                report=report,
+                content=content_maps["text/html"],
+            )
+
+        if "text/plain" in content_maps:
+            content_maps["text/plain"] = parse_text(
+                alias=alias,
+                content=content_maps["text/plain"],
+            )
+
+    if alias.create_mail_report and alias.user.public_key is not None:
+        create_email_report(
+            db,
+            report_data=report,
+            user=alias.user,
+        )
+
+    logger.info(
+        f"Email {envelope.mail_from} is from outside and wants to send to alias "
+        f"{alias.address}. "
+        f"Relaying email to locally saved user {alias.user.email.address}."
+    )
+
+    set_header(
+        message,
+        headers.KLECK_FORWARD_STATUS,
+        generate_forward_status(
+            StatusType.FORWARD_OUTSIDE_TO_ALIAS,
+            outside_address=envelope.mail_from,
+            message_id=message_id,
+        )
+    )
+
+    send_mail(
+        message,
+        from_mail=alias.create_outside_email(envelope.mail_from),
+        from_name=envelope.mail_from,
+        to_mail=alias.user.email.address,
+    )
+    server_statistics.add_sent_email(db)
+
+
+def parse_text(
+    alias: EmailAlias,
+    content: str,
+) -> str:
+    if alias.remove_footer:
+        content = remove_footer(content, "text/html")
+
+    return content
+
+
+def parse_html(
+    db: Session,
+    /,
+    alias: EmailAlias,
+    report: EmailReportData,
+    content: str,
+) -> str:
+    enable_image_proxy = settings.get(db, "ENABLE_IMAGE_PROXY")
+
+    if alias.remove_footer:
+        content = remove_footer(content, "text/html")
+
+    if alias.remove_trackers:
+        content = remove_single_pixel_image_trackers(report, html=content)
+
+    if enable_image_proxy and alias.proxy_images:
+        content = convert_images(db, report, alias=alias, html=content)
+
+    if alias.expand_url_shorteners:
+        content = expand_shortened_urls(report, alias=alias, html=content)
+
+    server_statistics.add_removed_trackers(db, len(report.single_pixel_images))
+    server_statistics.add_proxied_images(db, len(report.proxied_images))
+    server_statistics.add_expanded_urls(db, len(report.expanded_urls))
+
+    return content