hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(email): read spf/dkim verdicts from their own Authentication-Results clause

Browse Source 
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.
This commit is contained in:
kshitijk4poor
2026-09-27 17:23:05 +05:30
committed by kshitij
parent 517a97cd31
commit 3f4533b9de
2 changed files with 82 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
plugins/platforms/email/adapter.py
View File

@@ -55,14 +55,17 @@ _HTML_SUBS = ((re.compile(r"<br\s*/?>", re.IGNORECASE), "\n"), (re.compile(r"<p[
(re.compile(r"&amp;"), "&"), (re.compile(r"&lt;"), "<"), (re.compile(r"&gt;"), ">"), (re.compile(r"\n{3,}"), "\n\n")) (re.compile(r"&amp;"), "&"), (re.compile(r"&lt;"), "<"), (re.compile(r"&gt;"), ">"), (re.compile(r"\n{3,}"), "\n\n"))
# ``display <bracketed>`` split for the _extract_email_address fallback (linear: neither part can match the other's delimiters). # ``display <bracketed>`` 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*') _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"\([^()]*\)") _COMMENT_RE = re.compile(r"\([^()]*\)")
# Longest From: value we parse. parseaddr is pure Python and superlinear on hostile input (~1s at 100KB, GIL held); # 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). # a real mailbox plus display name stays far below this (RFC 5322 caps a line at 998 chars).
_MAX_FROM_LEN = 2048 _MAX_FROM_LEN = 2048
# "method=result" tokens (``dmarc=pass``) and property values (``header.from=x``) in Authentication-Results. # Authentication-Results clause head (``dmarc=pass``), matched only at the start of a clause.
_AUTH_METHOD_RE = re.compile(r"\b(dmarc|dkim|spf)\s*=\s*([a-z]+)", re.IGNORECASE) _AUTH_METHOD_RE = re.compile(r"\s*(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) # 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: def _esecret_int(name: str, default: int) -> int:
@@ -255,10 +258,10 @@ def _extract_email_address(raw: str) -> str:
``Name <addr>`` values parseaddr rejects (``Doe, John <j@x>``) fall back to their single bracketed address, ``Name <addr>`` values parseaddr rejects (``Doe, John <j@x>``) fall back to their single bracketed address,
but only when the display part (``(comments)`` removed) cannot hold another mailbox or group: no quotes, ``;``, 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 <a@x>``). Values over ``:`` or stray parens, and no ``@`` unless it is exactly the bracketed address (``a@x <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 "")) value = re.sub(r"\r?\n[ \t]+", " ", str(raw or ""))
if len(value) > _MAX_FROM_LEN: if len(value) > _MAX_FROM_LEN or value.count("(") > 64:
return "" # hostile size: take the empty-sender drop instead of a GIL-holding parse return "" # hostile size/nesting: take the empty-sender drop (parseaddr recurses per nested comment)
_, addr = parseaddr(value) _, addr = parseaddr(value)
if not addr and (m := _SINGLE_BRACKET_FROM_RE.fullmatch(value)): if not addr and (m := _SINGLE_BRACKET_FROM_RE.fullmatch(value)):
display, bracketed = _strip_comments(m.group(1)).strip(), m.group(2) 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): while i < len(text):
c = text[i] c = text[i]
if c == "\\" and (quoted or depth): if c == "\\" and (quoted or depth):
if depth == 0: if quoted:
cur.append(text[i:i + 2]) cur.append(text[i:i + 2])
i += 2 i += 2
continue continue
@@ -299,6 +302,8 @@ def _ar_clauses(text: str) -> Optional[List[str]]:
elif c == '"': elif c == '"':
quoted = True quoted = True
cur.append(c) cur.append(c)
elif c == ")":
return None # stray close paren: unbalanced
elif c == ";": elif c == ";":
clauses.append("".join(cur)) clauses.append("".join(cur))
cur = [] cur = []
@@ -309,8 +314,9 @@ def _ar_clauses(text: str) -> Optional[List[str]]:
def _auth_props(text: str) -> List[Tuple[str, str]]: def _auth_props(text: str) -> List[Tuple[str, str]]:
"""Authentication-Results ``(property, value)`` pairs (``header.from=x``), property lowercased, quotes stripped.""" """``(property, value)`` pairs (``header.from=x``) of one comment-free Authentication-Results clause, property
return [(p.lower(), v.strip().strip('"')) for p, v in _AUTH_PROP_RE.findall(text)] 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: def _domain_of(address: str) -> str:
@@ -342,25 +348,31 @@ def _verify_sender_authentication(msg: email_lib.message.Message, from_addr: str
or _domains_aligned(serv, authserv_id)), None) or _domains_aligned(serv, authserv_id)), None)
if trusted is None: if trusted is None:
return False, "no Authentication-Results from trusted authserv-id" return False, "no Authentication-Results from trusted authserv-id"
methods = {m.lower(): r.lower() for m, r in _AUTH_METHOD_RE.findall(trusted)} # Each verdict comes from the head of its own clause (split outside quotes/comments) and its domains only from that
props = dict(_auth_props(trusted)) # clause: a quoted local part or comment can otherwise smuggle ``spf=pass``/``header.d=`` (GHSA-rxqh-5572-8m77).
# 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.
if (clauses := _ar_clauses(trusted)) is None: if (clauses := _ar_clauses(trusted)) is None:
return False, "unbalanced quote or comment in Authentication-Results" 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" return False, "ambiguous dmarc result"
dmarc = dmarcs[0] if dmarcs else "" # every header.from in the dmarc clause must be the From domain we parsed (absent header.from: trust the verdict)
if (m := _DMARC_CLAUSE_RE.match(dmarc)) and m.group(1).lower() == "pass": if any(r == "pass" and aligned(props, ("header.from",), required=False) for r, props in results["dmarc"]):
if all(_domains_aligned(_domain_of(v), from_domain) for p, v in _auth_props(dmarc) if p == "header.from"): return True, "dmarc=pass"
return True, "dmarc=pass" # the verdict must be for the From domain we parsed (absent header.from: trust it) # one SMTP transaction has one MAIL FROM verdict: a second spf clause means the SPF signal is not trusted
if methods.get("spf") == "pass": # envelope/MAIL FROM domain must align with From if len(results["spf"]) == 1 and (spf := results["spf"][0])[0] == "pass" and aligned(
spf_domain = _domain_of(props.get("smtp.mailfrom", "")) or props.get("smtp.from", "") or props.get("envelope-from", "") spf[1], ("smtp.mailfrom", "smtp.from", "envelope-from")):
if _domains_aligned(_domain_of(spf_domain) if "@" in spf_domain else spf_domain, from_domain):
return True, "spf=pass aligned" return True, "spf=pass aligned"
if methods.get("dkim") == "pass": # signing domain header.d must align with From # several dkim clauses are normal (one per signature): any single pass whose own header.d aligns is enough
dkim_domain = props.get("header.d", "") or _domain_of(props.get("header.from", "")) if any(r == "pass" and aligned(props, ("header.d",) if any(p == "header.d" for p, _ in props) else ("header.from",))
if _domains_aligned(dkim_domain, from_domain): for r, props in results["dkim"]):
return True, "dkim=pass aligned" return True, "dkim=pass aligned"
return False, f"authentication failed ({trusted[:120]})" return False, f"authentication failed ({trusted[:120]})"

58
tests/gateway/test_email.py
View File

@@ -13,7 +13,6 @@ Covers:
""" """
import os import os
import time
import unittest import unittest
from email.mime.text import MIMEText from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart from email.mime.multipart import MIMEMultipart
@@ -81,12 +80,16 @@ class TestHelperFunctions(unittest.TestCase):
"attacker@evil.test, <victim@x>", "attacker@evil.test,\r\n <victim@x>", "attacker@evil.test, <victim@x>", "attacker@evil.test,\r\n <victim@x>",
"attacker@evil.test (c) <victim@x>", "attacker@evil.test; <victim@x>", "attacker@evil.test (c) <victim@x>", "attacker@evil.test; <victim@x>",
"Grp: attacker@evil.test; <victim@x>", "undisclosed-recipients:; <victim@x>", "Grp: attacker@evil.test; <victim@x>", "undisclosed-recipients:; <victim@x>",
"John", 'a\\"b <victim@x>'): "John", 'a\\"b <victim@x>',
# over _MAX_FROM_LEN (uncapped parseaddr would return a@example.com)
"x" * 3000 + " <a@example.com>",
# >=500 nested comments make stdlib parseaddr raise RecursionError
"Doe, " + "(" * 500 + ")" * 500 + " <v@example.com>"):
self.assertEqual(_extract_email_address(raw), "", raw) self.assertEqual(_extract_email_address(raw), "", raw)
# Hostile-size From values are refused outright (stdlib parseaddr takes ~1s at 100KB with the GIL held). # A From with no usable address is dropped at parse time, before dispatch.
start = time.monotonic() from plugins.platforms.email.adapter import EmailAdapter
self.assertEqual(_extract_email_address("<" + "a@" * 50_000), "") self.assertIsNone(EmailAdapter._parse_fetched_message(
self.assertLess(time.monotonic() - start, 0.5) 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): def test_extract_email_address_ignores_angle_brackets_in_display_name(self):
from plugins.platforms.email.adapter import _extract_email_address from plugins.platforms.email.adapter import _extract_email_address
@@ -193,8 +196,6 @@ class TestDispatchMessage(unittest.TestCase):
asyncio.run(adapter._dispatch_message(msg_data)) asyncio.run(adapter._dispatch_message(msg_data))
adapter._message_handler.assert_not_called() 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): def test_subject_included_in_text(self):
"""Subject should be prepended to body for non-reply emails.""" """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; ' 'mx.google.com; spf=pass smtp.mailfrom="x;dmarc=pass header.from=example.com x"@evil.test; '
"dmarc=fail header.from=example.com", "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 (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 <admin@example.com>", [ar]) ok, reason = self._verify("Admin <admin@example.com>", [ar])
self.assertFalse(ok, ar) self.assertFalse(ok, ar)
ok, reason = self._verify("Admin <admin@example.com>", [ # Real MTA headers (multi-signature DKIM, comments, quoted values) keep authenticating.
"mx.google.com; arc=pass (dmarc=fail header.from=evil.test); dmarc=pass header.from=example.com"]) for ar in ("mx.google.com; arc=pass (dmarc=fail header.from=evil.test); dmarc=pass header.from=example.com",
self.assertTrue(ok, reason) 'mx.google.com; dmarc=pass reason="a;b" header.from="example.com"',
ok, reason = self._verify("Admin <admin@example.com>", [ 'mx.google.com; dmarc=pass reason="header.from=evil.test" header.from=example.com',
'mx.google.com; dmarc=pass reason="a;b" header.from="example.com"']) "mx.google.com; dkim=pass header.i=@example.com header.s=s1 header.b=AbC; spf=pass (google.com: "
self.assertTrue(ok, reason) "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 <admin@example.com>", [ar])
self.assertTrue(ok, (ar, reason))
def test_dkim_pass_aligned_authenticates(self): def test_dkim_pass_aligned_authenticates(self):