From 69c49e4616bc3375964f90aae408376b20484d8f Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Tue, 8 Sep 2026 13:37:17 +0200 Subject: feat: extract sender domains and auth verdicts Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01KphFXTc2QajxXsHWyvGJ4R --- abusectl/parse.py | 69 ++++++++++++++++++++++++++++++++++++++++++++++++++++- tests/test_parse.py | 32 +++++++++++++++++++++++++ 2 files changed, 100 insertions(+), 1 deletion(-) 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() -- cgit v1.2.3