fix(email): linear, mailbox-safe From fallback; one-clause dmarc verdict

The r1 malformed-From fallback regex ([^<>\s]+@[^<>\s]+) backtracked
catastrophically on `From: <a@a@a@...`: 32KB held the GIL ~23s, and any
remote sender reaches it before auth. Capture <([^<>\s]+)> instead and
check the '@' in Python (same accepted set, 100KB in ~3ms).

The fallback also mapped multi-mailbox / group / comment-prefixed From
values (`attacker@evil.test, <victim@x>`, `Grp: a@evil; <victim@x>`,
`attacker@evil.test (c) <victim@x>`) to the bracketed victim, which a
dmarc=pass without header.from then authenticated. Only fall back when the
display part has no ; : ( ) and no '@' unless it is exactly the bracketed
address; `Doe, John <j@x>` and `j@x <j@x>` still resolve. The rule now lives
only in the docstring (the old inline comment was wrong).

dmarc: strip (comments) before splitting clauses, and take the verdict and
header.from from the same first clause that starts with dmarc=, so
`dmarc=pass (p=none; sp=none) header.from=evil.test`, a later dmarc=pass
clause after dmarc=fail, and duplicate misaligned header.from are rejected,
while `arc=pass (dmarc=fail ...); dmarc=pass header.from=<ours>` passes.
Property clean-up is shared via _auth_props.

The empty-sender drop now runs right after address parsing and gets an
assertion (it was untested). Tests extend existing ones (no new tests).

plugins/platforms/email/adapter.py

@@ -53,8 +53,9 @@ _CHARSET_ALIASES = {"unknown-8bit": "utf-8", "unknown": "utf-8", "x-unknown": "u
 _HTML_SUBS = ((re.compile(r"<br\s*/?>", re.IGNORECASE), "\n"), (re.compile(r"<p[^>]*>", re.IGNORECASE), "\n"),
               (re.compile(r"</p>", re.IGNORECASE), "\n"), (re.compile(r"<[^>]+>"), ""), (re.compile(r"&nbsp;"), " "),
               (re.compile(r"&amp;"), "&"), (re.compile(r"&lt;"), "<"), (re.compile(r"&gt;"), ">"), (re.compile(r"\n{3,}"), "\n\n"))
-# Unquoted From: with exactly one <addr> pair; fallback for real-world values strict parseaddr rejects.
+# ``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]+)>\s*')
+_SINGLE_BRACKET_FROM_RE = re.compile(r'([^"<>]*)<([^<>\s]+)>\s*')
+_DMARC_CLAUSE_RE = re.compile(r"\s*dmarc\s*=\s*([a-z]+)", re.IGNORECASE)
 # "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)
@@ -247,14 +248,24 @@ def _strip_html(html: str) -> str:
 def _extract_email_address(raw: str) -> str:
     """Bare lowercased addr-spec from a From: value. Uses parseaddr, not a first-<...> regex (GHSA-rxqh-5572-8m77);
     RFC 5322 folding is unfolded first because parseaddr misreads a folded quoted display name. Unquoted
-    ``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 cannot hold another mailbox, group or comment: no quotes, ``;``, ``:``,
+    ``(`` or ``)``, and no ``@`` unless it is exactly the bracketed address (``a@x <a@x>``)."""
     value = re.sub(r"\r?\n[ \t]+", " ", str(raw or ""))
     _, addr = parseaddr(value)
     if not addr and (m := _SINGLE_BRACKET_FROM_RE.fullmatch(value)):
-        addr = m.group(1)  # no quotes, exactly one <...> pair: nothing to hide a second mailbox in
+        display, bracketed = m.group(1).strip(), m.group(2)
+        if ("@" in bracketed[1:-1] and not any(c in display for c in ";:()")
+                and ("@" not in display or display.lower() == bracketed.lower())):
+            addr = bracketed
     return addr.strip().lower()
 
 
+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)]
+
+
 def _domain_of(address: str) -> str:
     """Lowercased domain part of an email address, or ''."""
     return address.rpartition("@")[2].strip().lower()
@@ -285,12 +296,12 @@ def _verify_sender_authentication(msg: email_lib.message.Message, from_addr: str
     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 = {p.lower(): v.strip().strip('"') for p, v in _AUTH_PROP_RE.findall(trusted)}
+    props = dict(_auth_props(trusted))
-    # header.from of the dmarc clause only; a later dkim/spf clause's header.from must not stand in for it.
+    # Verdict and header.from come from ONE clause: the first starting with dmarc= once (comments) are gone.
-    dmarc_from = next((v.strip().strip('"') for clause in trusted.split(";") if re.search(r"\bdmarc\s*=", clause, re.I)
+    dmarc = next((c for c in re.sub(r"\([^()]*\)", " ", trusted).split(";") if _DMARC_CLAUSE_RE.match(c)), "")
-                       for p, v in _AUTH_PROP_RE.findall(clause) if p.lower() == "header.from"), "")
+    if (m := _DMARC_CLAUSE_RE.match(dmarc)) and m.group(1).lower() == "pass":
-    if methods.get("dmarc") == "pass" and (not dmarc_from or _domains_aligned(_domain_of(dmarc_from), from_domain)):
+        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
+            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):
@@ -581,12 +592,12 @@ class EmailAdapter(BasePlatformAdapter):
     def _parse_fetched_message(self, uid: bytes, raw_email: "bytes | bytearray") -> Optional[Dict[str, Any]]:
         """Parse one RFC822 payload into a dispatchable dict; ``None`` for automated senders. Raises on pathological input (caller logs + continues)."""
         msg = email_lib.message_from_bytes(raw_email)
-        sender_addr, sender_name = _extract_email_address(msg.get("From", "")), _decode_header_value(msg.get("From", ""))
+        if not (sender_addr := _extract_email_address(msg.get("From", ""))):  # never dispatch an empty identity
-        if "<" in sender_name:
-            sender_name = sender_name.split("<")[0].strip().strip('"')
-        if not sender_addr:  # malformed/multi-address/group From: never dispatch an empty identity
             logger.debug("[Email] Dropping message with no parseable From address: %r", msg.get("From", ""))
             return None
+        sender_name = _decode_header_value(msg.get("From", ""))
+        if "<" in sender_name:
+            sender_name = sender_name.split("<")[0].strip().strip('"')
         subject = _decode_header_value(msg.get("Subject", "(no subject)"))
         if _is_automated_sender(sender_addr, dict(msg.items())):
             logger.debug("[Email] Skipping automated sender: %s", sender_addr)

tests/gateway/test_email.py

@@ -13,6 +13,7 @@ Covers:
 """
 
 import os
+import time
 import unittest
 from email.mime.text import MIMEText
 from email.mime.multipart import MIMEMultipart
@@ -72,10 +73,19 @@ class TestHelperFunctions(unittest.TestCase):
             "john@example.com"
         )
         # Unquoted forms strict parseaddr rejects still resolve to their single bracketed address.
-        for raw in ("john@example.com <john@example.com>", "Doe, John <John@example.com>", "a@x.test <john@example.com>"):
+        for raw in ("john@example.com <john@example.com>", "Doe, John <John@example.com>"):
             self.assertEqual(_extract_email_address(raw), "john@example.com", raw)
-        for raw in ("a@x.com, b@y.com", "Group: a@x.com, b@y.com;", "<>"):
+        # ...but never when the display part could hold another mailbox, group or comment.
+        for raw in ("a@x.com, b@y.com", "Group: a@x.com, b@y.com;", "<>", "a@x.test <john@example.com>",
+                    "attacker@evil.test, <victim@x>", "attacker@evil.test,\r\n <victim@x>",
+                    "attacker@evil.test (c) <victim@x>", "attacker@evil.test; <victim@x>",
+                    "Grp: attacker@evil.test; <victim@x>", "undisclosed-recipients:; <victim@x>"):
             self.assertEqual(_extract_email_address(raw), "", raw)
+        # The fallback regex stays linear on hostile input (a backtracking one took ~20s at 32KB, GIL held).
+        from plugins.platforms.email.adapter import _SINGLE_BRACKET_FROM_RE
+        start = time.monotonic()
+        _SINGLE_BRACKET_FROM_RE.fullmatch("<" + "a@" * 50_000)
+        self.assertLess(time.monotonic() - start, 0.5)
 
     def test_extract_email_address_ignores_angle_brackets_in_display_name(self):
         from plugins.platforms.email.adapter import _extract_email_address
@@ -182,6 +192,8 @@ 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."""
@@ -1202,10 +1214,16 @@ class TestSenderAuthentication(unittest.TestCase):
         self.assertTrue(ok, reason)
         # A dmarc=pass issued for another domain must not vouch for this From,
         # even when a later dkim clause carries an aligned header.from.
+        # Verdict and header.from are read from the one dmarc clause, with (comments) stripped first.
         for ar in ("mx.google.com; dmarc=pass header.from=evil.test",
-                   "mx.google.com; dmarc=pass header.from=evil.test; dkim=pass header.d=x.test header.from=example.com"):
+                   "mx.google.com; dmarc=pass header.from=evil.test; dkim=pass header.d=x.test header.from=example.com",
+                   "mx.google.com; dmarc=pass (p=none; sp=none) header.from=evil.test",
+                   "mx.google.com; dmarc=fail header.from=example.com; dmarc=pass header.from=evil.test"):
             ok, reason = self._verify("Admin <admin@example.com>", [ar])
             self.assertFalse(ok, ar)
+        ok, reason = self._verify("Admin <admin@example.com>", [
+            "mx.google.com; arc=pass (dmarc=fail header.from=evil.test); dmarc=pass header.from=example.com"])
+        self.assertTrue(ok, reason)
 
 
     def test_dkim_pass_aligned_authenticates(self):