feat: extract sender domains and auth verdicts

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KphFXTc2QajxXsHWyvGJ4R

2 files changed, 100 insertions, 1 deletions

diff --git a/abusectl/parse.py b/abusectl/parse.py
index f243e0c..33e1226 100644
--- a/abusectl/parse.py
+++ b/abusectl/parse.py
@@ -39,6 +39,8 @@ from email import policy
 from email.parser import BytesParser
 
 _BRACKETED_IP = re.compile(r"\[([0-9a-fA-F:.]+)\]")
+_ADDR_DOMAIN = re.compile(r"@([A-Za-z0-9.-]+)")
+_AUTH_VERDICT = re.compile(r"(spf|dkim|dmarc)=(\w+)", re.IGNORECASE)
 
 
 @dataclass(frozen=True)
@@ -61,6 +63,10 @@ class NoTrustBoundary(Exception):
         super().__init__("no trusted_relays configured: run `abusectl init`")
 
 
+def _message(raw: bytes):
+    return BytesParser(policy=policy.default).parsebytes(raw)
+
+
 def _extract_ip(received_value: str) -> str | None:
     # The bracketed literal after the connecting hostname is the only part
     # of a Received header the accepting server itself wrote; everything
@@ -78,7 +84,7 @@ def _extract_ip(received_value: str) -> str | None:
 
 def received_hops(raw: bytes) -> list[Hop]:
     """Return every Received hop that names an IP, outermost first."""
-    message = BytesParser(policy=policy.default).parsebytes(raw)
+    message = _message(raw)
     hops = []
     for value in message.get_all("received") or []:
         ip = _extract_ip(str(value))
@@ -112,3 +118,64 @@ def sending_ip(raw: bytes, trusted: list[str]) -> str | None:
         if not _in_any(hop.ip, trusted):
             return hop.ip
     return None
+
+
+def _domain_of(header_value: str | None) -> str | None:
+    if header_value is None:
+        return None
+    match = _ADDR_DOMAIN.search(header_value)
+    return match.group(1).lower() if match else None
+
+
+def sender_domains(raw: bytes) -> dict[str, str]:
+    """Return the domain of Return-Path, From, and Reply-To.
+
+    Recipient headers (To, Cc, Delivered-To, X-Original-To, the user's own
+    Message-IDs, maildir paths) are never read here, or anywhere in this
+    module. That is a structural guarantee rather than a filtering step: the
+    module cannot disclose a recipient identifier it was never given.
+
+    reply_to is included only when it differs from from: a Reply-To that
+    repeats From is noise, while one pointing somewhere else is a real
+    indicator, often the actual drop address behind a phishing attempt.
+    """
+    message = _message(raw)
+    domains = {}
+
+    return_path = _domain_of(message.get("Return-Path"))
+    if return_path is not None:
+        domains["return_path"] = return_path
+
+    from_domain = _domain_of(message.get("From"))
+    if from_domain is not None:
+        domains["from"] = from_domain
+
+    reply_to = _domain_of(message.get("Reply-To"))
+    if reply_to is not None and reply_to != from_domain:
+        domains["reply_to"] = reply_to
+
+    return domains
+
+
+def auth_results(raw: bytes) -> dict[str, str]:
+    """Return the SPF, DKIM and DMARC verdicts from Authentication-Results.
+
+    These are read, never recomputed. Recomputing would need DNS lookups,
+    and this module resolves nothing (see the module docstring). More to
+    the point, the receiving server's own verdict is the honest one: it is
+    what the server actually did with the message at delivery time, not a
+    check performed after the fact against records that may have changed.
+    """
+    message = _message(raw)
+    header = message.get("Authentication-Results")
+    if header is None:
+        return {}
+
+    verdicts: dict[str, str] = {}
+    for mechanism, verdict in _AUTH_VERDICT.findall(str(header)):
+        # setdefault: the first mention of a mechanism is the receiving
+        # server's own conclusion; a header can restate one later in a
+        # sub-clause and that repeat must not overwrite the real verdict.
+        verdicts.setdefault(mechanism.lower(), verdict.lower())
+
+    return verdicts
diff --git a/tests/test_parse.py b/tests/test_parse.py
index 92095fd..1083f1d 100644
--- a/tests/test_parse.py
+++ b/tests/test_parse.py
@@ -64,5 +64,37 @@ class TestReceivedChain(unittest.TestCase):
         self.assertIsNone(ip)
 
 
+class TestSenderDomains(unittest.TestCase):
+    def test_the_three_sender_headers_are_collected(self):
+        domains = parse.sender_domains(load("simple.eml"))
+        self.assertEqual(
+            domains,
+            {
+                "return_path": "sender.example.invalid",
+                "from": "bank.example.invalid",
+                "reply_to": "drop.example.invalid",
+            },
+        )
+
+    def test_reply_to_is_absent_when_it_matches_from(self):
+        # Only a DIFFERING Reply-To is an indicator; repeating From adds noise.
+        domains = parse.sender_domains(load("with-attachment.eml"))
+        self.assertNotIn("reply_to", domains)
+
+    def test_recipient_headers_are_never_returned(self):
+        # The safety property, asserted rather than assumed.
+        domains = parse.sender_domains(load("simple.eml"))
+        self.assertNotIn("example.org", domains.values())
+
+
+class TestAuthResults(unittest.TestCase):
+    def test_verdicts_are_read_as_the_server_recorded_them(self):
+        auth = parse.auth_results(load("simple.eml"))
+        self.assertEqual(auth, {"spf": "fail", "dkim": "none", "dmarc": "fail"})
+
+    def test_a_message_with_no_auth_header_reports_nothing(self):
+        self.assertEqual(parse.auth_results(load("with-attachment.eml")), {})
+
+
 if __name__ == "__main__":
     unittest.main()