From 3f4533b9dedb526e5891354164b07b59a8557cc5 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sun, 27 Sep 2026 17:23:05 +0530 Subject: [PATCH] fix(email): read spf/dkim verdicts from their own Authentication-Results clause The r3 fold scoped only the dmarc verdict to its clause. SPF and DKIM still came from a whole-string, last-match-wins regex scan, so the same quoted/comment smuggle closed for dmarc still authenticated a spoofed From (GHSA-rxqh-5572-8m77), e.g. spf=fail smtp.mailfrom="x spf=pass smtp.mailfrom=example.com "@evil.test spf=fail (spf=pass) smtp.mailfrom=a@example.com spf=fail smtp.mailfrom=a.spf=pass@example.com dkim=pass header.d=evil.test header.i="x header.d=example.com y"@evil.test Every verdict now comes from the leading method=result token of its own clause (from _ar_clauses, comments dropped), and its domains only from that clause. Properties are read by a token scanner that consumes quoted-strings and other key=value tokens whole, so quoted contents are never read as properties while a quoted value still is (header.from="example.com"). SPF fails closed on more than one spf clause; DKIM accepts any single dkim=pass clause whose own header.d aligns (multi-signature mail is normal), never mixing clauses. The whole-string methods/props (methods["dmarc"] was dead) are gone. Also: a stray ')' at depth 0 is now unbalanced (it split header.from out of the dmarc clause); a From with more than 64 '(' takes the silent empty-sender drop instead of a parseaddr RecursionError logged as an error; the cap test asserts the cap directly instead of wall-clock timing; the empty-From drop assertion moves next to the other _extract_email_address rejects; and the untested >1-dmarc, unbalanced and backslash-escape rules get reject strings. --- plugins/platforms/email/adapter.py | 66 ++++++++++++++++++------------ tests/gateway/test_email.py | 58 +++++++++++++++++++------- 2 files changed, 82 insertions(+), 42 deletions(-) diff --git a/plugins/platforms/email/adapter.py b/plugins/platforms/email/adapter.py index 0a8510f4ef..29c22674fe 100644 --- a/plugins/platforms/email/adapter.py +++ b/plugins/platforms/email/adapter.py @@ -55,14 +55,17 @@ _HTML_SUBS = ((re.compile(r"", re.IGNORECASE), "\n"), (re.compile(r""), (re.compile(r"\n{3,}"), "\n\n")) # ``display `` split for the _extract_email_address fallback (linear: neither part can match the other's delimiters). _SINGLE_BRACKET_FROM_RE = re.compile(r'([^"<>]*)<([^<>\s]+)>\s*') -_DMARC_CLAUSE_RE = re.compile(r"\s*dmarc\s*=\s*([a-z]+)", re.IGNORECASE) _COMMENT_RE = re.compile(r"\([^()]*\)") # Longest From: value we parse. parseaddr is pure Python and superlinear on hostile input (~1s at 100KB, GIL held); # a real mailbox plus display name stays far below this (RFC 5322 caps a line at 998 chars). _MAX_FROM_LEN = 2048 -# "method=result" tokens (``dmarc=pass``) and property values (``header.from=x``) in Authentication-Results. -_AUTH_METHOD_RE = re.compile(r"\b(dmarc|dkim|spf)\s*=\s*([a-z]+)", re.IGNORECASE) -_AUTH_PROP_RE = re.compile(r"\b(header\.from|header\.d|smtp\.mailfrom|smtp\.from|envelope-from)\s*=\s*([^\s;]+)", re.IGNORECASE) +# Authentication-Results clause head (``dmarc=pass``), matched only at the start of a clause. +_AUTH_METHOD_RE = re.compile(r"\s*(dmarc|dkim|spf)\s*=\s*([a-z]+)", re.IGNORECASE) +# One token of a clause: a property we read (``header.from=x``; the value may be or contain a quoted-string), or +# any other whitespace-delimited token consumed whole, so text inside quotes or other values is never read as a prop. +_QUOTED = r'"(?:[^"\\]|\\.)*"' +_AUTH_PROP_RE = re.compile(r'(header\.from|header\.d|smtp\.mailfrom|smtp\.from|envelope-from)\s*=\s*((?:%s|[^\s";])+)' + r'|(?:%s|[^\s"])+' % (_QUOTED, _QUOTED), re.IGNORECASE) def _esecret_int(name: str, default: int) -> int: @@ -255,10 +258,10 @@ def _extract_email_address(raw: str) -> str: ``Name `` values parseaddr rejects (``Doe, John ``) fall back to their single bracketed address, but only when the display part (``(comments)`` removed) cannot hold another mailbox or group: no quotes, ``;``, ``:`` or stray parens, and no ``@`` unless it is exactly the bracketed address (``a@x ``). Values over - ``_MAX_FROM_LEN`` and results without ``@`` return ``""`` so the caller drops the message.""" + ``_MAX_FROM_LEN`` (or with more than 64 ``(``) and results without ``@`` return ``""`` so the caller drops the message.""" value = re.sub(r"\r?\n[ \t]+", " ", str(raw or "")) - if len(value) > _MAX_FROM_LEN: - return "" # hostile size: take the empty-sender drop instead of a GIL-holding parse + if len(value) > _MAX_FROM_LEN or value.count("(") > 64: + return "" # hostile size/nesting: take the empty-sender drop (parseaddr recurses per nested comment) _, addr = parseaddr(value) if not addr and (m := _SINGLE_BRACKET_FROM_RE.fullmatch(value)): display, bracketed = _strip_comments(m.group(1)).strip(), m.group(2) @@ -283,7 +286,7 @@ def _ar_clauses(text: str) -> Optional[List[str]]: while i < len(text): c = text[i] if c == "\\" and (quoted or depth): - if depth == 0: + if quoted: cur.append(text[i:i + 2]) i += 2 continue @@ -299,6 +302,8 @@ def _ar_clauses(text: str) -> Optional[List[str]]: elif c == '"': quoted = True cur.append(c) + elif c == ")": + return None # stray close paren: unbalanced elif c == ";": clauses.append("".join(cur)) cur = [] @@ -309,8 +314,9 @@ def _ar_clauses(text: str) -> Optional[List[str]]: def _auth_props(text: str) -> List[Tuple[str, str]]: - """Authentication-Results ``(property, value)`` pairs (``header.from=x``), property lowercased, quotes stripped.""" - return [(p.lower(), v.strip().strip('"')) for p, v in _AUTH_PROP_RE.findall(text)] + """``(property, value)`` pairs (``header.from=x``) of one comment-free Authentication-Results clause, property + lowercased, surrounding quotes stripped. Quoted-string contents are never scanned for properties.""" + return [(p.lower(), v.strip('"')) for p, v in _AUTH_PROP_RE.findall(text) if p] def _domain_of(address: str) -> str: @@ -342,26 +348,32 @@ def _verify_sender_authentication(msg: email_lib.message.Message, from_addr: str or _domains_aligned(serv, authserv_id)), None) if trusted is None: return False, "no Authentication-Results from trusted authserv-id" - methods = {m.lower(): r.lower() for m, r in _AUTH_METHOD_RE.findall(trusted)} - props = dict(_auth_props(trusted)) - # Verdict and header.from come from the ONE clause starting with dmarc= (split outside quotes/comments). - # A quoted smtp.mailfrom can smuggle a fake clause, so an unbalanced value or a second dmarc clause fails closed. + # Each verdict comes from the head of its own clause (split outside quotes/comments) and its domains only from that + # clause: a quoted local part or comment can otherwise smuggle ``spf=pass``/``header.d=`` (GHSA-rxqh-5572-8m77). if (clauses := _ar_clauses(trusted)) is None: return False, "unbalanced quote or comment in Authentication-Results" - if len(dmarcs := [c for c in clauses if _DMARC_CLAUSE_RE.match(c)]) > 1: + results: Dict[str, List[Tuple[str, List[Tuple[str, str]]]]] = {"dmarc": [], "spf": [], "dkim": []} + for clause in clauses: + if m := _AUTH_METHOD_RE.match(clause): + results[m.group(1).lower()].append((m.group(2).lower(), _auth_props(clause))) + + def aligned(props: List[Tuple[str, str]], names: Tuple[str, ...], *, required: bool = True) -> bool: + domains = [_domain_of(v) for p, v in props if p in names] + return (bool(domains) or not required) and all(_domains_aligned(d, from_domain) for d in domains) + + if len(results["dmarc"]) > 1: return False, "ambiguous dmarc result" - dmarc = dmarcs[0] if dmarcs else "" - if (m := _DMARC_CLAUSE_RE.match(dmarc)) and m.group(1).lower() == "pass": - if all(_domains_aligned(_domain_of(v), from_domain) for p, v in _auth_props(dmarc) if p == "header.from"): - return True, "dmarc=pass" # the verdict must be for the From domain we parsed (absent header.from: trust it) - if methods.get("spf") == "pass": # envelope/MAIL FROM domain must align with From - spf_domain = _domain_of(props.get("smtp.mailfrom", "")) or props.get("smtp.from", "") or props.get("envelope-from", "") - if _domains_aligned(_domain_of(spf_domain) if "@" in spf_domain else spf_domain, from_domain): - return True, "spf=pass aligned" - if methods.get("dkim") == "pass": # signing domain header.d must align with From - dkim_domain = props.get("header.d", "") or _domain_of(props.get("header.from", "")) - if _domains_aligned(dkim_domain, from_domain): - return True, "dkim=pass aligned" + # every header.from in the dmarc clause must be the From domain we parsed (absent header.from: trust the verdict) + if any(r == "pass" and aligned(props, ("header.from",), required=False) for r, props in results["dmarc"]): + return True, "dmarc=pass" + # one SMTP transaction has one MAIL FROM verdict: a second spf clause means the SPF signal is not trusted + if len(results["spf"]) == 1 and (spf := results["spf"][0])[0] == "pass" and aligned( + spf[1], ("smtp.mailfrom", "smtp.from", "envelope-from")): + return True, "spf=pass aligned" + # several dkim clauses are normal (one per signature): any single pass whose own header.d aligns is enough + if any(r == "pass" and aligned(props, ("header.d",) if any(p == "header.d" for p, _ in props) else ("header.from",)) + for r, props in results["dkim"]): + return True, "dkim=pass aligned" return False, f"authentication failed ({trusted[:120]})" diff --git a/tests/gateway/test_email.py b/tests/gateway/test_email.py index a923fc7619..7de4c97a31 100644 --- a/tests/gateway/test_email.py +++ b/tests/gateway/test_email.py @@ -13,7 +13,6 @@ Covers: """ import os -import time import unittest from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart @@ -81,12 +80,16 @@ class TestHelperFunctions(unittest.TestCase): "attacker@evil.test, ", "attacker@evil.test,\r\n ", "attacker@evil.test (c) ", "attacker@evil.test; ", "Grp: attacker@evil.test; ", "undisclosed-recipients:; ", - "John", 'a\\"b '): + "John", 'a\\"b ', + # over _MAX_FROM_LEN (uncapped parseaddr would return a@example.com) + "x" * 3000 + " ", + # >=500 nested comments make stdlib parseaddr raise RecursionError + "Doe, " + "(" * 500 + ")" * 500 + " "): self.assertEqual(_extract_email_address(raw), "", raw) - # Hostile-size From values are refused outright (stdlib parseaddr takes ~1s at 100KB with the GIL held). - start = time.monotonic() - self.assertEqual(_extract_email_address("<" + "a@" * 50_000), "") - self.assertLess(time.monotonic() - start, 0.5) + # A From with no usable address is dropped at parse time, before dispatch. + from plugins.platforms.email.adapter import EmailAdapter + self.assertIsNone(EmailAdapter._parse_fetched_message( + object.__new__(EmailAdapter), b"2", b"From: a@x.com, b@y.com\r\nSubject: x\r\n\r\nbody")) def test_extract_email_address_ignores_angle_brackets_in_display_name(self): from plugins.platforms.email.adapter import _extract_email_address @@ -193,8 +196,6 @@ class TestDispatchMessage(unittest.TestCase): asyncio.run(adapter._dispatch_message(msg_data)) adapter._message_handler.assert_not_called() - # A From with no usable address is dropped at parse time, before dispatch. - self.assertIsNone(adapter._parse_fetched_message(b"2", b"From: a@x.com, b@y.com\r\nSubject: x\r\n\r\nbody")) def test_subject_included_in_text(self): """Subject should be prepended to body for non-reply emails.""" @@ -1226,15 +1227,42 @@ class TestSenderAuthentication(unittest.TestCase): 'mx.google.com; spf=pass smtp.mailfrom="x;dmarc=pass header.from=example.com x"@evil.test; ' "dmarc=fail header.from=example.com", "mx.google.com; dmarc=pass (a (b) ; header.from=example.com) header.from=evil.test", - 'mx.google.com; dmarc=pass reason="a;b" header.from=evil.test'): + 'mx.google.com; dmarc=pass reason="a;b" header.from=evil.test', + "mx.google.com; dmarc=pass a) ; header.from=evil.test", # stray ')' is unbalanced + "mx.google.com; dmarc=pass header.from=example.com; dmarc=pass header.from=evil.test", + "mx.google.com; dmarc=pass (a ; header.from=evil.test", + r'mx.google.com; spf=pass smtp.mailfrom="x\\";dmarc=pass header.from=example.com;x="y"; ' + "dmarc=fail header.from=example.com", + # spf/dkim verdicts and domains come only from their own clause, never quoted text or comments + 'mx.google.com; spf=fail smtp.mailfrom="x spf=pass smtp.mailfrom=example.com "@evil.test; ' + "dmarc=fail header.from=example.com", + "mx.google.com; spf=fail (spf=pass) smtp.mailfrom=a@example.com", + "mx.google.com; spf=fail smtp.mailfrom=a.spf=pass@example.com; dmarc=fail header.from=example.com", + 'mx.google.com; dkim=pass header.d=evil.test header.i="x header.d=example.com y"@evil.test', + "mx.google.com; spf=pass smtp.mailfrom=example.com; spf=fail smtp.mailfrom=evil.test", + "mx.google.com; dkim=pass header.d=evil.test; dkim=fail header.d=example.com", + 'mx.google.com; dkim=pass header.i="x header.d=example.com"@evil.test', + # an escaped quote keeps the quoted-string open, so no dmarc clause is smuggled out of it + r'mx.google.com; spf=fail smtp.mailfrom="a\";dmarc=pass header.from=example.com;x=\""@evil.test'): ok, reason = self._verify("Admin ", [ar]) self.assertFalse(ok, ar) - ok, reason = self._verify("Admin ", [ - "mx.google.com; arc=pass (dmarc=fail header.from=evil.test); dmarc=pass header.from=example.com"]) - self.assertTrue(ok, reason) - ok, reason = self._verify("Admin ", [ - 'mx.google.com; dmarc=pass reason="a;b" header.from="example.com"']) - self.assertTrue(ok, reason) + # Real MTA headers (multi-signature DKIM, comments, quoted values) keep authenticating. + for ar in ("mx.google.com; arc=pass (dmarc=fail header.from=evil.test); dmarc=pass header.from=example.com", + 'mx.google.com; dmarc=pass reason="a;b" header.from="example.com"', + 'mx.google.com; dmarc=pass reason="header.from=evil.test" header.from=example.com', + "mx.google.com; dkim=pass header.i=@example.com header.s=s1 header.b=AbC; spf=pass (google.com: " + "domain of admin@example.com designates 1.2.3.4 as permitted sender) smtp.mailfrom=admin@example.com; " + "dmarc=pass (p=REJECT sp=REJECT dis=NONE) header.from=example.com", + "spf=pass (sender IP is 1.2.3.4) smtp.mailfrom=example.com; dkim=pass (signature was verified) " + "header.d=example.com;dmarc=pass action=none header.from=example.com;compauth=pass reason=100", + "mail.example.org; dmarc=pass (p=none dis=none) header.from=example.com", + 'mail.example.org; dkim=pass (2048-bit key; unprotected) header.d=example.com header.i=@example.com ' + 'header.b="AbC+/1"; spf=pass smtp.mailfrom=example.com', + "mx.example.org; dkim=pass (1024-bit key) header.d=esp.test header.i=@esp.test; " + "dkim=pass (2048-bit key) header.d=example.com header.i=@example.com; spf=softfail " + "smtp.mailfrom=bounce@esp.test"): + ok, reason = self._verify("Admin ", [ar]) + self.assertTrue(ok, (ar, reason)) def test_dkim_pass_aligned_authenticates(self):