initial forged-from protection

chatmaild/src/chatmaild/dictproxy.py

@@ -31,7 +31,9 @@ def encrypt_password(password: str):
 
 def create_user(db, user, password):
     if os.path.exists(NOCREATE_FILE):
-        logging.warning(f"Didn't create account: {NOCREATE_FILE} exists. Delete the file to enable account creation.")
+        logging.warning(
+            f"Didn't create account: {NOCREATE_FILE} exists. Delete the file to enable account creation."
+        )
         return
     with db.write_transaction() as conn:
         conn.create_user(user, password)

chatmaild/src/chatmaild/filtermail.py

@@ -3,6 +3,7 @@ import asyncio
 import logging
 from email.parser import BytesParser
 from email import policy
+from email.utils import parseaddr
 
 from aiosmtpd.lmtp import LMTP
 from aiosmtpd.controller import UnixSocketController
@@ -42,50 +43,14 @@ class ExampleHandler:
         return "250 OK"
 
     async def handle_DATA(self, server, session, envelope):
-        logging.info("Processing DATA message from %s", envelope.mail_from)
-
-        valid_recipients = []
-
-        message = BytesParser(policy=policy.default).parsebytes(envelope.content)
-        mail_encrypted = check_encrypted(message)
-
-        res = []
-        for recipient in envelope.rcpt_tos:
-            my_local_domain = envelope.mail_from.split("@")
-            if len(my_local_domain) != 2:
-                res += [f"500 Invalid from address <{envelope.mail_from}>"]
-                continue
-
-            if envelope.mail_from == recipient:
-                # Always allow sending emails to self.
-                valid_recipients += [recipient]
-                res += ["250 OK"]
-                continue
-
-            recipient_local_domain = recipient.split("@")
-            if len(recipient_local_domain) != 2:
-                res += [f"500 Invalid address <{recipient}>"]
-                continue
-
-            is_outgoing = recipient_local_domain[1] != my_local_domain[1]
-
-            if (
-                is_outgoing
-                and not mail_encrypted
-                and message.get("secure-join") != "vc-request"
-                and message.get("secure-join") != "vg-request"
-            ):
-                res += ["500 Outgoing mail must be encrypted"]
-                continue
-
-            valid_recipients += [recipient]
-            res += ["250 OK"]
-
+        valid_recipients, res = lmtp_handle_DATA(envelope)
         # Reinject the mail back into Postfix.
         if valid_recipients:
             logging.info("Reinjecting the mail")
             client = SMTPClient("localhost", "10026")
             client.sendmail(envelope.mail_from, valid_recipients, envelope.content)
+        else:
+            logging.info("no valid recipients, ignoring mail")
 
         return "\r\n".join(res)
 
@@ -97,6 +62,55 @@ async def asyncmain(loop):
     controller.start()
 
 
+def lmtp_handle_DATA(envelope):
+    """the central filtering function for e-mails."""
+    logging.info(f"Processing DATA message from {envelope.mail_from}")
+
+    message = BytesParser(policy=policy.default).parsebytes(envelope.content)
+    mail_encrypted = check_encrypted(message)
+
+    valid_recipients = []
+    res = []
+    for recipient in envelope.rcpt_tos:
+        my_local_domain = envelope.mail_from.split("@")
+        if len(my_local_domain) != 2:
+            res += [f"500 Invalid from address <{envelope.mail_from}>"]
+            continue
+
+        _, from_addr = parseaddr(message.get("from").strip().lower())
+        logging.info(f"mime-from: {from_addr} envelope-from: {envelope.mail_from}")
+        if envelope.mail_from != from_addr:
+            res += [f"500 Invalid FROM <{envelope.mail_from}>"]
+            continue
+
+        if envelope.mail_from == recipient:
+            # Always allow sending emails to self.
+            valid_recipients += [recipient]
+            res += ["250 OK"]
+            continue
+
+        recipient_local_domain = recipient.split("@")
+        if len(recipient_local_domain) != 2:
+            res += [f"500 Invalid address <{recipient}>"]
+            continue
+
+        is_outgoing = recipient_local_domain[1] != my_local_domain[1]
+
+        if (
+            is_outgoing
+            and not mail_encrypted
+            and message.get("secure-join") != "vc-request"
+            and message.get("secure-join") != "vg-request"
+        ):
+            res += ["500 Outgoing mail must be encrypted"]
+            continue
+
+        valid_recipients += [recipient]
+        res += ["250 OK"]
+
+    return valid_recipients, res
+
+
 def main():
     logging.basicConfig(level=logging.INFO)
     loop = asyncio.new_event_loop()

chatmaild/src/chatmaild/test_filtermail.py

@@ -1,8 +1,45 @@
-from .filtermail import check_encrypted
+from .filtermail import check_encrypted, lmtp_handle_DATA
 from email.parser import BytesParser
 from email import policy
 
 
+def test_reject_forged_from():
+    def makemail(from_addr):
+        return BytesParser(policy=policy.default).parsebytes(
+            "\r\n".join(
+                [
+                    f"From: <{from_addr}",
+                    "To: <barbaz@c3.testrun.org>",
+                    "Date: Sun, 15 Oct 2023 16:41:44 +0000",
+                    "Message-ID: <Mr.3gckbNy5bch.uK3Hd2Ws6-w@c2.testrun.org>",
+                    "References: <Mr.3gckbNy5bch.uK3Hd2Ws6-w@c2.testrun.org>",
+                    "Chat-Version: 1.0",
+                    "MIME-Version: 1.0",
+                    "Content-Type: text/plain; charset=utf-8; format=flowed; delsp=no",
+                    "",
+                    "Hi!",
+                    "",
+                    "",
+                ]
+            ).encode()
+        )
+
+    class envelope:
+        mail_from = "bob@c3.testrun.org"
+        rcpt_tos = ["somebody@c3.testrun.org"]
+
+    envelope.content = makemail(envelope.mail_from).as_bytes()
+
+    valid_recipients, res = lmtp_handle_DATA(envelope=envelope)
+    assert valid_recipients == envelope.rcpt_tos
+    assert len(res) == 1 and "250" in res[0]
+
+    envelope.content = makemail("forged@c3.testrun.org").as_bytes()
+    valid_recipients, res = lmtp_handle_DATA(envelope=envelope)
+    assert not valid_recipients
+    assert len(res) == 1 and "500" in res[0]
+
+
 def test_filtermail():
     def check_encrypted_bstr(content):
         message = BytesParser(policy=policy.default).parsebytes(content)