Fix smtp listener to reject email addresses with unexpected domain

cc: #1077

CHANGELOG.md

@@ -4,7 +4,7 @@ All notable changes to this project will be documented in this file.
 ## v3.8-dev - Unreleased
 
 ### Bug Fixes
-- Fix the smtp listener to reject email addresses with non-UUID local parts (#1077)
+- Improve recipient address validation in the smtp listener (#1077)
 
 ## v3.7 - 2024-10-21
 

hc/api/management/commands/smtpd.py

@@ -11,6 +11,7 @@ from typing import Any, Protocol
 from aiosmtpd.controller import Controller
 from aiosmtpd.smtp import SMTP, Envelope, Session
 from asgiref.sync import sync_to_async
+from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.db import connection
 
@@ -107,10 +108,11 @@ class PingHandler:
         address: str,
         rcpt_options: list[str],
     ) -> str:
-        to_parts = address.split("@")
-        code = to_parts[0]
+        code, domain = address.split("@", maxsplit=1)
+        if domain != settings.PING_EMAIL_DOMAIN:
+            return "550 5.1.1 Recipient rejected"
         if not RE_UUID.match(code):
-            return "550 5.1.0 Requested action not taken: mailbox unavailable"
+            return "550 5.1.1 Invalid mailbox"
 
         envelope.rcpt_tos.append(address)
         return "250 OK"

hc/api/tests/test_smtpd.py

@@ -40,12 +40,17 @@ Content-Transfer-Encoding: 8bit
 """.strip()
 
 
-@override_settings(S3_BUCKET=None)
+class NullSink:
+    def write(self, text: str) -> None:
+        pass
+
+
+@override_settings(S3_BUCKET=None, PING_EMAIL_DOMAIN="hc.example.com")
 class SmtpdTestCase(BaseTestCase):
     def setUp(self) -> None:
         super().setUp()
         self.check = Check.objects.create(project=self.project)
-        self.email = "%s@does.not.matter" % self.check.code
+        self.email = f"{self.check.code}@hc.example.com"
 
     def test_it_works(self) -> None:
         _process_message("1.2.3.4", "foo@example.org", self.email, b"hello world")
@@ -210,10 +215,6 @@ class SmtpdTestCase(BaseTestCase):
         envelope.rcpt_tos = ["bar@example.org", self.email]
         envelope.content = b"hello world"
 
-        class NullSink:
-            def write(self, text: str) -> None:
-                pass
-
         handler = PingHandler(NullSink())
         await handler.handle_DATA(Mock(), session, envelope)
 
@@ -226,12 +227,14 @@ class SmtpdTestCase(BaseTestCase):
 
     async def test_it_rejects_non_uuid_mailboxes(self) -> None:
         session = Session(loop=Mock())
-
-        class NullSink:
-            def write(self, text: str) -> None:
-                pass
-
         handler = PingHandler(NullSink())
         address = "foo@example.com"
         result = await handler.handle_RCPT(Mock(), session, Envelope(), address, [])
         self.assertTrue(result.startswith("550"))
+
+    async def test_it_rejects_wrong_domain(self) -> None:
+        session = Session(loop=Mock())
+        handler = PingHandler(NullSink())
+        address = f"{self.check.code}@bad.domain"
+        result = await handler.handle_RCPT(Mock(), session, Envelope(), address, [])
+        self.assertTrue(result.startswith("550"))