WEBDIENSTE/lasuite-messages
1
0
Fork 0
mirror of https://github.com/suitenumerique/messages.git synced 2026-09-17 23:37:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

rebase & update config shape

Browse Source
This commit is contained in:
Sylvain Zimmer
2026-07-23 12:42:05 +02:00
parent 0506cb0d66
commit 2c68aac9fa
10 changed files with 856 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/spam-config.md
-104
View File
@@ -1,104 +0,0 @@
# `SPAM_CONFIG`
Inbound spam filtering **and** sender-authentication config. Set globally via the
`SPAM_CONFIG` env var (JSON), overridable per mail domain through
`MailDomain.custom_settings["SPAM_CONFIG"]` (merged by
`MailDomain.get_spam_config()`).
> Naming note: these keys cover both spam scoring and sender auth/trust. The
> auth keys live here because `inbound_auth` already did — not because they are
> "spam". A future split into a dedicated auth/trust config is possible but out
> of scope here.
## Keys
| Key | Type | Default | Purpose |
| --- | --- | --- | --- |
| `rspamd_url` | string | — | rspamd `/checkv2` endpoint. Absent = no rspamd scan. |
| `rspamd_auth` | string | — | Optional `Authorization` header value for rspamd. |
| `rules` | list | `[]` | Hardcoded header-match spam rules (see below). |
| `trusted_relays` | int | `0` | How many upstream `Received` blocks to trust for the header-match `rules`. Block 0 = the block our own MTA prepends. Not used by `inbound_auth: "arc"`. |
| `inbound_auth` | string\|null | `null` | Sender-auth backend for the verdict/banner (see below). |
| `trusted_arc_sealers` | list | `[]` | ARC sealer `d=` allowlist. `[]` = accept any valid seal. Used by `inbound_auth: "arc"` and `arc_gate`. |
| `arc_gate` | string | `"off"` | Action when a message is **not** sealed by a trusted sealer (see below). |
## `rules`
Each rule matches one header, within the `trusted_relays` block window:
```jsonc
{ "header_match": "X-Foo: exact-value", "action": "spam" } // literal, case-insensitive
{ "header_match_regex": "X-Spam-Level:\\*{5,}", "action": "spam" } // regex, IGNORECASE
```
`action`: `"spam"` / `"reject"` → `is_spam=True`; `"ham"` / `"no action"` →
`is_spam=False`. Default `"spam"`. First matching rule wins. `Return-Path` is
never eligible (spoofable envelope value).
## `inbound_auth` — the verdict / banner
Produces `postmark["auth"]`: absent = verified, `"none"` = unverified,
`"fail"` = likely forged (DMARC disavowal).
| Value | Source |
| --- | --- |
| `"native"` | Local DKIM verify + strict `From`/`d=` alignment. |
| `"rspamd"` | dkim/dmarc from the rspamd result. |
| `"arc"` | dkim/dmarc from a trusted sealer's **sealed** `ARC-Authentication-Results` only. Plaintext headers are never read. Untrusted/unsealed → unverified. |
| `"authentication-results"` | Parse `dkim=`/`dmarc=` from the top-level `Authentication-Results`, trusted by `trusted_relays` position. |
| `null` / absent | Disabled. |
## `arc_gate` — relay-trust enforcement
Verifies the ARC chain (dkimpy) and applies an action when the message is **not**
sealed by a trusted sealer (`cv=pass` **and** outermost sealer ∈
`trusted_arc_sealers`, or any `cv=pass` when the allowlist is empty).
| Value | Effect |
| --- | --- |
| `"off"` | No gating (default). |
| `"spam"` | Not trusted-sealed → `is_spam=True` (Junk). |
| `"drop"` | Not trusted-sealed → silently discarded. |
Runs first among the spam steps, so an untrusted verdict is authoritative. A DNS
/ verification failure never spams or drops (can't verify ≠ forged). `quarantine`
and `reject` are planned for a follow-up.
> **Public-MX warning:** with an empty `trusted_arc_sealers`, "any valid seal"
> is bypassable — an attacker self-seals their own domain (`cv=pass`,
> `d=attacker.example`). On a publicly reachable MX, **populate the allowlist**
> so the ARC seal is a real trust anchor. `trusted_arc_sealers` may list several
> sealers (e.g. an external relay plus an internal gateway).
## Examples
Public MX, accept only trusted-ARC-sealed mail (mark the rest as spam):
```jsonc
{
"inbound_auth": "arc",
"trusted_arc_sealers": ["relay.example"],
"arc_gate": "spam",
"rspamd_url": "http://rspamd:11334/checkv2",
"trusted_relays": 99,
"rules": [ { "header_match_regex": "X-Spam-Level:\\*{5,}", "action": "spam" } ]
}
```
Minimal ARC gate, no rspamd:
```jsonc
{ "inbound_auth": "arc", "trusted_arc_sealers": ["relay.example"], "arc_gate": "spam" }
```
Multiple sealers (external relay + internal gateway):
```jsonc
{ "inbound_auth": "arc", "trusted_arc_sealers": ["relay.example", "gateway.internal.example"], "arc_gate": "spam" }
```
Legacy header-based auth (no ARC):
```jsonc
{ "inbound_auth": "authentication-results", "trusted_relays": 1 }
```
docs/spam.md
+150 -15
View File
@@ -13,10 +13,14 @@ mail, runs these steps in order:
1. **Before-spam webhooks** (`message.inbound`) — user/integration webhooks that
may drop, defer, or pre-decide the spam verdict.
2. **Hardcoded header rules** — deterministic `header_match` rules from config.
3. **rspamd** — native `/checkv2` scan.
4. **Inbound authentication** — DKIM/DMARC verdict (SPF indirectly).
5. **After-spam webhooks** (`message.delivering`, then `message.delivered`).
2. **ARC** — holds (`RETRY`) a message whose *trusted* ARC seal couldn't be
verified because of a DNS failure (only when `trusted_arc_sealers` is set).
See [ARC relay-trust](#arc-relay-trust).
3. **Hardcoded rules** — deterministic `header_match` rules **and** `arc_verdict`
relay-trust rules from config.
4. **rspamd** — native `/checkv2` scan.
5. **Inbound authentication** — DKIM/DMARC verdict (SPF indirectly).
6. **After-spam webhooks** (`message.delivering`, then `message.delivered`).
Each step returns a `Decision` (`CONTINUE` / `RETRY` / `DROP`) and may set the
spam verdict. The verdict is a tri-state `ctx.is_spam`: `None` (undecided) until
@@ -106,9 +110,10 @@ global → mail domain. (A future enhancement could add a per-mailbox layer.)
|------------------|--------|---------|
| `rspamd_url` | string | Base URL of the rspamd HTTP endpoint (`/checkv2` is appended). Omit to disable rspamd. |
| `rspamd_auth` | string | Optional value for the `Authorization` header sent to rspamd. |
| `inbound_auth` | string | Sender-auth backend: `native`, `rspamd`, or `authentication-results`. Omit/empty to disable DKIM/DMARC checks. |
| `trusted_relays` | int | Number of sender-side `Received`/`Authentication-Results` blocks to trust, counting from the boundary our own MTA prepends. Default `0` (trust only our own hop). Raise this when a fixed upstream gateway sits in front. |
| `rules` | list | Ordered hardcoded header-match rules (see below). |
| `inbound_auth` | string | Sender-auth backend: `native`, `rspamd`, `arc`, or `authentication-results`. Omit/empty to disable DKIM/DMARC checks. |
| `trusted_relays` | int | Number of sender-side `Received`/`Authentication-Results` blocks to trust, counting from the boundary our own MTA prepends. Default `0` (trust only our own hop). Raise this when a fixed upstream gateway sits in front. Not used by `inbound_auth: "arc"`. |
| `trusted_arc_sealers` | list | ARC sealer `d=` allowlist. **Fail closed: `[]` (or absent) trusts nothing** — you must list your sealers. Used by `inbound_auth: "arc"` and by `arc_verdict` rules, and enables the ARC `RETRY`-on-DNS-failure hold (see [ARC relay-trust](#arc-relay-trust)). |
| `rules` | list | Ordered rules — `header_match` / `header_match_regex` **or** `arc_verdict` trust conditions — with action `spam` / `ham` / `drop` (see below). |
### Related settings
@@ -137,6 +142,10 @@ The backend is selected by `SPAM_CONFIG["inbound_auth"]`:
returns `fail` (no DMARC policy lookup); worst case is `none`.
- **`rspamd`** — read DKIM/DMARC **symbols** from the rspamd `/checkv2` result
(reusing the spam-step scan). Verdict precedence: `fail` > `pass` > `none`.
- **`arc`** — read `dkim=`/`dmarc=` from the `ARC-Authentication-Results` that a
**trusted sealer** cryptographically sealed (RFC 8617). Plaintext headers are
never read; an unsealed or untrusted-sealed message is `none`. See
[ARC relay-trust](#arc-relay-trust).
- **`authentication-results`** — parse `dkim=`/`dmarc=` from the
`Authentication-Results` header(s) added by trusted upstream relays (bounded
by `trusted_relays`). Use this when an upstream MX gateway already does
@@ -156,22 +165,148 @@ is not `pass`, **unverified** (`auth = "none"`); otherwise verified.
> classification indirectly through rspamd scoring (the envelope is forwarded to
> rspamd). The user-facing auth verdict is DKIM + DMARC only.
## Hardcoded header rules
## ARC relay-trust
`SPAM_CONFIG["rules"]` is an ordered list of deterministic header-match rules,
evaluated before rspamd. The first matching rule decides the verdict. Each rule:
[ARC](https://datatracker.ietf.org/doc/html/rfc8617) (Authenticated Received
Chain) lets an intermediary that observed a message's original authentication
**cryptographically seal** that observation, so a downstream receiver can trust
it even after forwarding breaks SPF/DKIM. Messages uses ARC two ways, both keyed
off one allowlist, `trusted_arc_sealers`:
- **`trusted_arc_sealers`** — the `d=` domains whose seals we trust (subdomains
match). **Fail closed:** `[]` (or absent) trusts *nothing* — anyone can
produce a valid ARC seal, so you must list the sealers you trust. Trust is
granted only when the chain is `cv=pass` **and** the outermost sealer is on
the allowlist.
The result is a **binary verdict** (`core/mda/arc.py`):
| `arc_verdict` | Meaning |
|---|---|
| `trusted` | `cv=pass` **and** sealed by an allowlisted sealer |
| `untrusted` | everything else — no ARC chain, a chain from an unlisted sealer, or a chain that fails to validate |
> **We only verify seals we could trust.** If the allowlist is empty, or the
> message's outermost sealer is not on it, the chain is `untrusted` regardless of
> validity, so we skip crypto + DNS entirely. Attacker-controlled mail (which
> never names a trusted sealer) therefore triggers **zero** DNS traffic, and a
> forged chain claiming a trusted sealer is capped at 20 instances before we
> refuse to verify.
**1. As a sender-auth verdict** (`inbound_auth: "arc"`) — `dkim`/`dmarc` are read
from the trusted sealer's sealed `ARC-Authentication-Results`; untrusted/unsealed
→ `none`. See [Sender authentication](#sender-authentication-dkim--dmarc).
**2. As a gating rule** — an `arc_verdict` rule acts on the verdict:
```jsonc
{ "arc_verdict": "untrusted", "action": "drop" } // discard (no Message)
{ "arc_verdict": "untrusted", "action": "spam" } // route to Junk
{ "arc_verdict": "trusted", "action": "ham" } // allowlist trusted mail
```
Rules are evaluated in list order (first match wins), so an `arc_verdict` rule
composes with the header rules in the same `rules` list. A `dnsfail` (below) is
indeterminate and matches **neither** verdict.
### DNS failures hold, they don't fail open
`dnsfail` is an **internal, transient** signal — never an `arc_verdict` value. A
key-record lookup that doesn't complete (timeout / SERVFAIL / NXDOMAIN / empty)
is **indeterminate**, not a forgery — NXDOMAIN in particular can be transient
(negative caching after a fresh publish, a zone mid-reload). Because we only
verify seals from a listed sealer, a `dnsfail` only ever arises for a message
**claiming one of your trusted sealers**. Such a message is held for retry
(`Decision.RETRY`) by the `arc` pipeline step — never delivered unverified,
never dropped. The hold is bounded by `MESSAGES_INBOUND_DEFERRAL_MAX_AGE` (48h
default). **Past the window** the seal is deemed unresolvable (our own relay's
DNS works, so a key that never resolved for 48h is treated as bogus) and
reclassified to a definite `untrusted` verdict — so the `arc_verdict` rules then
apply (an `untrusted` → `drop`/`spam` rule fires) rather than force-delivering it.
> **Widget submissions are exempt.** Messages from a widget channel's web form
> carry no seal by construction, so the arc step and `arc_verdict` rules skip
> them — an `untrusted` → `drop` rule never discards first-party form traffic.
> (rspamd and `header_match` rules still apply.) If no widget channel exists,
> no widget-origin mail exists in the first place.
> **Fail closed:** an empty `trusted_arc_sealers` trusts nothing — with
> `inbound_auth: "arc"` every message is then `none` (unverified), and an
> `arc_verdict: "untrusted"` rule matches everything. **Populate the allowlist**
> (it may list several sealers, e.g. an external relay plus an internal gateway)
> to actually trust anything.
> **Single-relay assumption:** the sealed `ARC-Authentication-Results` is read
> from the **outermost** ARC instance — correct for one trusted relay in front of
> you. With two or more sealing hops the outermost AAR reflects the *last* hop's
> re-evaluation (which may show `dkim=fail` for legitimately forwarded mail),
> collapsing to `none` rather than recovering the origin verdict from an inner
> instance. This fails safe (never a false "verified") and is a deliberate
> limitation, not a bug.
### Examples
Public MX, accept only trusted-ARC-sealed mail, junk the rest:
```jsonc
{
"inbound_auth": "arc",
"trusted_arc_sealers": ["relay.example"],
"rules": [{ "arc_verdict": "untrusted", "action": "spam" }],
"rspamd_url": "http://rspamd:11334/checkv2"
}
```
Third-party MX relay, ARC mandatory + honor the relay's `X-Spam` verdict. The
relay is your published MX: it scans mail, ARC-seals it (`d=relay.thirdparty.example`),
stamps `X-Spam-*` headers, then forwards to your MTA.
```jsonc
{
// Sender-auth banner comes from the relay's sealed results.
"inbound_auth": "arc",
"trusted_arc_sealers": ["relay.thirdparty.example"],
// The relay adds one hop in front of our MTA, so its X-Spam / Received
// headers land in block 1. Trust block 0+1 (raise if it chains more hops).
"trusted_relays": 1,
"rules": [
// 1. ARC mandatory: discard anything the relay didn't seal. First, so
// unsealed mail dies before any of its headers are trusted.
{ "arc_verdict": "untrusted", "action": "drop" },
// 2. Honor the relay's verdict — safe because rule 1 guarantees every
// surviving message came through the relay.
{ "header_match": "X-Spam-Flag: YES", "action": "spam" }
]
// No rspamd_url: the relay already scans.
}
```
Because `trusted_arc_sealers` is non-empty, a DNS failure verifying the relay's
seal **holds** the message (RETRY) instead of dropping it — a relay-DNS outage
never silently discards legitimate mail. Prefer `"action": "spam"` over `"drop"`
in rule 1 while validating the setup, then tighten to `drop`.
## Hardcoded rules
`SPAM_CONFIG["rules"]` is an ordered list of deterministic rules, evaluated
before rspamd. The first matching rule decides the verdict. Each rule has exactly
one condition plus an `action`:
| Field | Meaning |
|----------------------|---------|
| `header_match` | Literal `Header-Name: value` (case-insensitive). Must contain a colon. |
| `header_match_regex` | Regex alternative, full-match, case-insensitive. |
| `action` | `spam` / `reject` → mark spam; `ham` / `no action` → mark not-spam. Default `spam`. |
| `arc_verdict` | ARC relay-trust condition — `trusted` / `untrusted` (see [ARC relay-trust](#arc-relay-trust)). |
| `action` | `spam` / `reject` → mark spam; `ham` / `no action` → mark not-spam; `drop` → discard the message (no `Message` row). Default `spam`. |
Rules honor `trusted_relays`: only headers within the trusted window (the most
recent `trusted_relays + 1` header blocks, newest first) are considered, so a
spammer can't forge a header that an upstream you trust would have stripped or
Header rules honor `trusted_relays`: only headers within the trusted window (the
most recent `trusted_relays + 1` header blocks, newest first) are considered, so
a spammer can't forge a header that an upstream you trust would have stripped or
overwritten. The `Return-Path` header is always ignored (spoofable envelope
value).
value). `arc_verdict` conditions ignore `trusted_relays` — they use the
cryptographic chain, not header position.
This is the primary mechanism for **honoring the verdict of an upstream filter**
(next section).
src/backend/core/mda/arc.py
+136 -21
View File
@@ -1,13 +1,36 @@
"""ARC chain verification for inbound relay-trust (RFC 8617)."""
"""ARC chain verification for inbound relay-trust (RFC 8617).
The rule-facing outcome is a **binary verdict**: a message is either ``trusted``
(a valid ARC chain sealed by an allowlisted sealer) or ``untrusted``
(everything else — no chain, a chain from a sealer we don't list, or a chain
that fails to validate). ``dnsfail`` is an *internal, transient* signal, not a
verdict: a key lookup that didn't complete for a message claiming one of our
trusted sealers is held for retry by the pipeline (never dropped, never
delivered as verified), so it must never reach a gating rule.
Performance / attack surface: we only pay the crypto + DNS cost of full chain
verification when the message's outermost sealer is one we could actually trust.
An unlisted sealer is ``untrusted`` regardless of whether its chain is valid, so
validating it buys nothing — and skipping it means attacker-controlled mail
(which never names a trusted sealer) triggers zero DNS traffic.
"""
import logging
from typing import Any, Dict, Optional, Set
from typing import Any, Dict, Optional, Set, Tuple
from dkim import CV_Pass, arc_verify
from dkim import ARC, CV_Pass, arc_verify, get_txt
from dkim.util import parse_tag_value
logger = logging.getLogger(__name__)
# Upper bound on ARC instances we are willing to cryptographically verify. A
# forged chain claiming a trusted sealer could otherwise force one DNS lookup
# per instance; real chains are a handful of hops, so cap well below that.
# Beyond this the message is simply ``untrusted`` (never verified).
_MAX_ARC_INSTANCES = 20
def _sealer_trusted(sealer: Optional[str], trusted: Set[str]) -> bool:
"""True if sealer equals or is a subdomain of a trusted sealer."""
if not sealer:
@@ -17,8 +40,62 @@ def _sealer_trusted(sealer: Optional[str], trusted: Set[str]) -> bool:
return any(sealer.endswith("." + t) for t in trusted)
def _normalize_domain(raw: Any) -> Optional[str]:
"""Lowercase, strip surrounding whitespace and a trailing dot from a d=."""
if isinstance(raw, (bytes, bytearray)):
raw = raw.decode("ascii", "replace")
if not isinstance(raw, str):
return None
return raw.strip().rstrip(".").lower() or None
def _outermost_sealer(raw_data: bytes) -> Tuple[Optional[str], int]:
"""Cheaply (no crypto, no DNS) find the outermost ARC sealer.
Returns ``(sealer_domain, max_instance)`` where ``sealer_domain`` is the
``d=`` of the ``ARC-Message-Signature`` at the highest instance, and
``max_instance`` is ``0`` when the message carries no (parseable) ARC chain.
Uses dkimpy's own header sorting so the sealer we gate on is exactly the one
``arc_verify`` would treat as outermost — no second parser to diverge from.
Any parse error is reported as "no chain" (``0``); a malformed ARC structure
would fail full verification anyway, and reporting no-chain fails *safe*
(the message is treated as untrusted, never over-trusted).
"""
try:
max_instance, arc_headers = ARC(raw_data).sorted_arc_headers()
except Exception as exc: # pylint: disable=broad-exception-caught
logger.debug("ARC header parse failed (treating as no chain): %s", exc)
return None, 0
if max_instance == 0:
return None, 0
for instance, (name, value) in arc_headers:
if instance == max_instance and name.lower() == b"arc-message-signature":
try:
domain = parse_tag_value(value).get(b"d")
except Exception: # pylint: disable=broad-exception-caught
return None, max_instance
return _normalize_domain(domain), max_instance
return None, max_instance
def arc_result(raw_data: bytes, trusted_sealers: Set[str]) -> Dict[str, Any]:
"""Verify the ARC chain; empty trusted_sealers accepts any valid seal."""
"""Classify a message's ARC relay-trust.
**Fail closed:** an empty ``trusted_sealers`` trusts *nothing* — you must
list the sealers you trust. (Anyone can produce a valid ARC seal, so trusting
"any valid seal" would let a spammer self-seal into "verified".)
Returns a dict:
- ``trusted``: ``cv=pass`` AND sealed by an allowlisted sealer.
- ``sealer``: the outermost ARC-Message-Signature ``d=`` (``None`` if no
chain). Used to scope the retry-on-DNS-failure hold.
- ``aar``: the outermost ARC-Authentication-Results value, but ONLY
when ``trusted``; ``None`` otherwise.
- ``dnsfail``: internal — a key lookup for a *claimed-trusted* sealer did
not complete, so the result is indeterminate. Callers hold (retry), they
never gate on it. See the module docstring.
"""
result: Dict[str, Any] = {
"trusted": False,
"sealer": None,
@@ -26,32 +103,70 @@ def arc_result(raw_data: bytes, trusted_sealers: Set[str]) -> Dict[str, Any]:
"dnsfail": False,
}
# Fail closed: with no allowlist nothing is trusted, so there is nothing to
# parse or verify.
if not trusted_sealers:
return result
# --- Cheap gate: decide whether full verification is even worth it. ---
sealer, max_instance = _outermost_sealer(raw_data)
result["sealer"] = sealer
if max_instance == 0:
# No ARC chain -> untrusted. No crypto, no DNS.
return result
if max_instance > _MAX_ARC_INSTANCES:
# Implausibly long chain — refuse to verify (amplification guard).
logger.info("ARC chain too long (%d instances) — untrusted", max_instance)
return result
if not _sealer_trusted(sealer, trusted_sealers):
# The outermost sealer is not one we trust, so the chain's validity is
# irrelevant — untrusted without spending any crypto/DNS on it.
return result
# --- The outermost sealer could be trusted: verify the chain (crypto+DNS).
dns_incomplete = False
def _tracking_dnsfunc(name, timeout=5):
nonlocal dns_incomplete
try:
txt = get_txt(name, timeout=timeout)
except Exception: # pylint: disable=broad-exception-caught
dns_incomplete = True
raise
if not txt:
dns_incomplete = True
return txt
try:
cv, results, _reason = arc_verify(raw_data)
cv, results, _reason = arc_verify(raw_data, dnsfunc=_tracking_dnsfunc)
except Exception as exc: # pylint: disable=broad-exception-caught
logger.warning("ARC verify errored (treating as untrusted): %s", exc)
result["dnsfail"] = dns_incomplete
return result
if not results:
# Structurally broken chain; if a lookup was also outstanding, treat it
# as indeterminate (hold) rather than a definite untrusted verdict.
result["dnsfail"] = dns_incomplete
return result
# Prefer dkimpy's own parsed domain for the trust decision (authoritative).
outer = results[0]
sealer_raw = outer.get("ams-domain")
if isinstance(sealer_raw, (bytes, bytearray)):
sealer = sealer_raw.decode("ascii", "replace")
else:
sealer = sealer_raw or None
if sealer:
sealer = sealer.strip().rstrip(".").lower() or None
result["sealer"] = sealer
verified_sealer = _normalize_domain(outer.get("ams-domain"))
if verified_sealer:
result["sealer"] = verified_sealer
allowed = not trusted_sealers or _sealer_trusted(sealer, trusted_sealers)
if cv == CV_Pass and allowed:
result["trusted"] = True
aar_raw = outer.get("aar-value")
if isinstance(aar_raw, (bytes, bytearray)):
result["aar"] = aar_raw.decode("utf-8", "replace")
elif isinstance(aar_raw, str):
result["aar"] = aar_raw
if cv == CV_Pass:
if _sealer_trusted(result["sealer"], trusted_sealers):
result["trusted"] = True
aar_raw = outer.get("aar-value")
if isinstance(aar_raw, (bytes, bytearray)):
result["aar"] = aar_raw.decode("utf-8", "replace")
elif isinstance(aar_raw, str):
result["aar"] = aar_raw
elif dns_incomplete:
# Claimed-trusted sealer whose key we couldn't fetch: indeterminate, not
# forged — held for retry by the arc pipeline step.
result["dnsfail"] = True
return result
src/backend/core/mda/inbound_auth.py
+1 -1
View File
@@ -24,7 +24,7 @@ The backend is picked by ``SPAM_CONFIG["inbound_auth"]``:
- ``"rspamd"``: read DKIM / DMARC symbols from the rspamd /checkv2 result
(reused from the spam check, or fetched on demand by the caller).
- ``"arc"``: dkim/dmarc from the ``ARC-Authentication-Results`` sealed by a
trusted sealer (``trusted_arc_sealers``; empty = any valid seal). Plaintext
trusted sealer (``trusted_arc_sealers``; empty = trust nothing). Plaintext
headers are never read. Unsealed/untrusted -> unverified.
- ``"authentication-results"``: parse ``dkim=`` / ``dmarc=`` from the
top-level ``Authentication-Results`` sliced by ``trusted_relays``.
src/backend/core/mda/inbound_pipeline.py
+112 -27
View File
@@ -37,7 +37,7 @@ from jmap_email import JmapEmail
from core import enums, models
from core.mda import spam
from core.mda.arc import arc_result
from core.mda.arc import _sealer_trusted, arc_result
from core.mda.inbound_auth import (
check_inbound_authentication,
get_inbound_auth_mode,
@@ -134,7 +134,9 @@ class InboundContext: # pylint: disable=too-many-instance-attributes
# the symbols (DKIM/DMARC verdicts) without a second HTTP call.
rspamd_result: Optional[Dict[str, Any]] = None
# Populated by ``arc_gate_step`` so ``inbound_auth_step`` can reuse it.
# ARC relay-trust result, computed at most once by ``ensure_arc`` and
# shared by the arc step (RETRY-on-dnsfail), the ``arc`` spam rules, and
# the ``inbound_auth: "arc"`` verdict.
arc: Optional[Dict[str, Any]] = None
# Memoised results of blocking webhook steps, keyed by
@@ -179,37 +181,109 @@ DEFERRAL_MAX_AGE = timedelta(seconds=settings.MESSAGES_INBOUND_DEFERRAL_MAX_AGE)
# ---------------------------------------------------------------------------
def _make_arc_gate_step(spam_config: Dict[str, Any]) -> Step:
action = str(spam_config.get("arc_gate") or "off").strip().lower()
def ensure_arc(ctx: InboundContext, spam_config: Dict[str, Any]) -> Dict[str, Any]:
"""Compute the ARC relay-trust result once and cache it on the context.
def arc_gate(ctx: InboundContext) -> Decision:
if action == "off":
return Decision.CONTINUE
The arc step, the ``arc`` spam rules and the ``inbound_auth: "arc"`` verdict
all read the same ``ctx.arc``, so the crypto + DNS work happens at most once
per message.
"""
if ctx.arc is None:
ctx.arc = arc_result(ctx.raw_data, trusted_arc_sealers(spam_config))
if ctx.arc["trusted"] or ctx.arc["dnsfail"]:
return ctx.arc
def _is_widget(ctx: InboundContext) -> bool:
"""True for messages submitted through a widget channel's web form.
Widget submissions are unauthenticated by construction (a web form, no SMTP
peer, no DKIM/ARC), so ARC relay-trust does not apply to them — gating them
on a seal they can never carry would just drop legitimate first-party form
traffic.
"""
return (ctx.inbound_message.envelope or {}).get(
"origin"
) == enums.InboundOrigin.WIDGET
def _make_arc_step(spam_config: Dict[str, Any]) -> Step:
"""Hold (RETRY) a message whose *trusted* ARC seal we couldn't verify.
Only a populated ``trusted_arc_sealers`` allowlist is an enforcement
posture: there, a DNS lookup that didn't complete (timeout / SERVFAIL /
NXDOMAIN / empty) for a message *claiming* one of our trusted sealers is
indeterminate, not forged — so we hold it for retry rather than fail open
(deliver it unverified) or fail closed (drop it).
Once the hold exceeds ``DEFERRAL_MAX_AGE`` we stop giving the message the
benefit of the doubt: a seal claiming one of our sealers whose key never
resolved for the whole window — while our own relay's DNS works — is treated
as a *definite* ``untrusted`` verdict, so the ``arc_verdict`` rules apply (an
``untrusted`` → ``drop``/``spam`` rule fires). This is deliberately unlike
the generic "force-deliver flagged" deferral fallback: an unresolvable ARC
seal is evidence about the message, not a transient processing outage.
With no allowlist there is nothing to hold against (and "any valid seal"
would make holds pathological), so we skip entirely — and skip the ARC
verification cost too. Widget submissions never carry ARC and are skipped.
Untrusted messages (a *definite* verdict) are handled by the ``arc_verdict``
spam rules, not held here.
"""
trusted = trusted_arc_sealers(spam_config)
def arc(ctx: InboundContext) -> Decision:
if not trusted or _is_widget(ctx):
return Decision.CONTINUE
logger.info(
"ARC gate: untrusted message (sealer=%s) -> %s",
ctx.arc["sealer"],
action,
)
if action == "drop":
return Decision.DROP
if action == "spam" and ctx.is_spam is None:
ctx.is_spam = True
result = ensure_arc(ctx, spam_config)
if result["dnsfail"] and _sealer_trusted(result["sealer"], trusted):
age = timezone.now() - ctx.inbound_message.created_at
if age <= DEFERRAL_MAX_AGE:
logger.info(
"ARC: key lookup for claimed-trusted sealer %s did not "
"complete on inbound message %s — holding for retry",
result["sealer"],
ctx.inbound_message.id,
)
return Decision.RETRY
# Gave up: reclassify as a definite untrusted verdict so the
# ``arc_verdict`` rules decide (drop / spam / deliver-unverified).
logger.warning(
"ARC: sealer %s unresolved past the deferral window on inbound "
"message %s — treating as untrusted",
result["sealer"],
ctx.inbound_message.id,
)
result["dnsfail"] = False
return Decision.CONTINUE
arc_gate.name = "arc_gate"
return arc_gate
arc.name = "arc"
return arc
def _make_hardcoded_rules_step(spam_config: Dict[str, Any]) -> Step:
rules = spam_config.get("rules") or []
# Only pay the ARC verification cost when a rule actually needs it.
needs_arc = any(isinstance(r, dict) and r.get("arc_verdict") for r in rules)
def hardcoded_rules(ctx: InboundContext) -> Decision:
if ctx.is_spam is not None:
return Decision.CONTINUE
verdict = spam.check_hardcoded_rules(ctx.parsed_email, spam_config)
if verdict is not None:
ctx.is_spam = verdict
# Never compute an ARC verdict for widget submissions — they carry no
# seal, so an ``arc_verdict`` rule must not gate them (ctx.arc stays None
# and matches nothing).
if needs_arc and ctx.arc is None and not _is_widget(ctx):
ensure_arc(ctx, spam_config)
verdict = spam.check_hardcoded_rules(ctx.parsed_email, spam_config, arc=ctx.arc)
if verdict == "drop":
logger.info(
"Spam rule 'drop' on inbound message %s — discarding",
ctx.inbound_message.id,
)
return Decision.DROP
if verdict == "spam":
ctx.is_spam = True
elif verdict == "ham":
ctx.is_spam = False
return Decision.CONTINUE
hardcoded_rules.name = "hardcoded_rules"
@@ -350,11 +424,13 @@ def build_inbound_pipeline(ctx: InboundContext) -> List[Step]:
Order matters:
1. Before-spam user webhooks — may DROP, RETRY, or set is_spam.
2. ``hardcoded_rules`` — header-match rules per domain config.
3. ``rspamd`` — fills the gap if nothing decided spam yet, and
2. ``arc`` — holds (RETRY) a message whose *trusted* ARC seal couldn't
be verified due to a DNS failure (only when an allowlist is set).
3. ``hardcoded_rules`` — header-match and ``arc`` trust rules per config.
4. ``rspamd`` — fills the gap if nothing decided spam yet, and
caches symbols for the next step.
4. ``inbound_auth`` — DKIM / DMARC verdict, may mutate parsed_email.
5. After-spam user webhooks — see the verdict, may override it,
5. ``inbound_auth`` — DKIM / DMARC verdict, may mutate parsed_email.
6. After-spam user webhooks — see the verdict, may override it,
may add labels, may DROP/RETRY.
"""
# Imported here to avoid the inbound_pipeline ↔ dispatch_webhooks
@@ -385,9 +461,18 @@ def build_inbound_pipeline(ctx: InboundContext) -> List[Step]:
),
]
if get_inbound_auth_mode(ctx.spam_config) == "arc" and not trusted_arc_sealers(
ctx.spam_config
):
logger.warning(
"inbound_auth='arc' with empty trusted_arc_sealers: no sealer is "
"trusted, so every message is treated as unverified. Populate "
"trusted_arc_sealers."
)
return [
*webhook_steps_for_mailbox(ctx.mailbox, phase="before_spam", channels=channels),
_make_arc_gate_step(ctx.spam_config),
_make_arc_step(ctx.spam_config),
_make_hardcoded_rules_step(ctx.spam_config),
_make_rspamd_step(ctx.spam_config),
_make_inbound_auth_step(ctx.spam_config),
src/backend/core/mda/spam.py
+86 -10
View File
@@ -22,14 +22,91 @@ from core.mda.utils import headers_blocks
logger = logging.getLogger(__name__)
# Rule ``action`` -> normalized verdict token. ``spam``/``reject`` route to
# Junk; ``ham``/``no action`` force-deliver; ``drop`` discards the message
# with no Message row (maps to ``Decision.DROP`` in the pipeline).
_RULE_ACTION_ALIASES = {
"spam": "spam",
"reject": "spam",
"ham": "ham",
"no action": "ham",
"drop": "drop",
}
def _normalize_rule_action(action: Any) -> Optional[str]:
"""Normalize a rule's ``action`` to ``"spam"`` / ``"ham"`` / ``"drop"``.
An empty/absent action defaults to ``"spam"``; an unrecognized action
yields ``None`` so the caller skips the rule rather than guessing.
"""
if action is None or (isinstance(action, str) and not action.strip()):
return "spam"
if not isinstance(action, str):
return None
return _RULE_ACTION_ALIASES.get(action.strip().lower())
def _arc_verdict_matches(
condition: Any, arc: Optional[Dict[str, Any]], idx: int
) -> bool:
"""Whether an ``arc_verdict`` rule condition holds.
The verdict is binary: ``"trusted"`` (valid chain sealed by an allowlisted
sealer) or ``"untrusted"`` (everything else — no chain, an unlisted sealer,
or a chain that failed to validate).
A ``dnsfail`` result (a key lookup that didn't complete) is *indeterminate*
and matches **neither** verdict: such a message claiming a trusted sealer is
held for retry by the arc pipeline step before the rules run, and in
best-effort mode (empty allowlist) it falls through unverified rather than
being gated on a transient DNS error. An absent ``arc`` result (couldn't
compute) likewise never matches — we never gate on what we couldn't decide.
"""
cond = str(condition).strip().lower()
if cond not in ("trusted", "untrusted"):
logger.warning(
"Unknown arc_verdict %r in spam rule #%d — skipping", condition, idx
)
return False
if arc is None or arc.get("dnsfail"):
return False
trusted = bool(arc.get("trusted"))
return trusted if cond == "trusted" else not trusted
def check_hardcoded_rules(
parsed_email: JmapEmail, spam_config: Dict[str, Any]
) -> Optional[bool]:
"""Apply the per-domain hardcoded ``rules`` list, header-matched
only against headers from trusted relay blocks. Returns ``True`` /
``False`` on first matching rule, ``None`` if no rule matched."""
parsed_email: JmapEmail,
spam_config: Dict[str, Any],
arc: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
"""Apply the per-domain ``rules`` list; return the first matching rule's
normalized action — ``"spam"`` (Junk), ``"ham"`` (deliver) or ``"drop"``
(discard) — or ``None`` if nothing matched.
Rules are evaluated in list order (first match wins) and are of two kinds:
- ``header_match`` / ``header_match_regex``: matched only against headers
from trusted relay blocks (see ``trusted_relays``).
- ``arc_verdict``: matched against the pre-computed ``arc`` relay-trust
result (``trusted`` / ``untrusted``). The caller must pass ``arc``; a
rule referencing it when none was computed simply never matches.
"""
rules = spam_config.get("rules", [])
for idx, rule in enumerate(rules):
if not isinstance(rule, dict):
continue
# ARC relay-trust condition — evaluated against the pre-computed
# crypto/DNS result, not the header blocks.
arc_condition = rule.get("arc_verdict")
if arc_condition:
if _arc_verdict_matches(arc_condition, arc, idx):
verdict = _normalize_rule_action(rule.get("action"))
if verdict is not None:
return verdict
logger.warning("Unknown action in spam rule #%d — skipping", idx)
continue
header_match = rule.get("header_match") or rule.get("header_match_regex")
if not header_match:
continue
@@ -105,11 +182,10 @@ def check_hardcoded_rules(
logger.warning("Invalid regex in spam rule #%d — skipping", idx)
continue
if is_match:
action = rule.get("action") or "spam"
if action in ("spam", "reject"):
return True
if action in ("ham", "no action"):
return False
verdict = _normalize_rule_action(rule.get("action"))
if verdict is not None:
return verdict
logger.warning("Unknown action in spam rule #%d — skipping", idx)
return None
src/backend/core/tests/mda/test_arc.py
+146 -26
View File
@@ -26,54 +26,107 @@ class TestSealerTrusted:
class TestArcResult:
def test_empty_allowlist_accepts_any_valid(self):
results = [
{"ams-domain": b"whoever.example", "aar-value": b"i=1; mx; dkim=pass"}
]
with patch(
"core.mda.arc.arc_verify", return_value=(arc.CV_Pass, results, "ok")
):
"""Decision logic — parser and crypto both mocked."""
@staticmethod
def _outer(sealer, max_i=1):
return patch("core.mda.arc._outermost_sealer", return_value=(sealer, max_i))
@staticmethod
def _verify(cv, results):
return patch("core.mda.arc.arc_verify", return_value=(cv, results, "ok"))
def test_empty_allowlist_trusts_nothing(self):
# Fail closed: no allowlist -> nothing trusted, and no parse/verify.
with patch("core.mda.arc._outermost_sealer") as mock_outer, patch(
"core.mda.arc.arc_verify"
) as mock_verify:
out = arc.arc_result(b"raw", set())
assert out["trusted"] is True
assert out["sealer"] == "whoever.example"
assert out == {"trusted": False, "sealer": None, "aar": None, "dnsfail": False}
mock_outer.assert_not_called()
mock_verify.assert_not_called()
def test_verify_exception_untrusted(self):
with patch("core.mda.arc.arc_verify", side_effect=Exception("boom")):
with self._outer("relay.example"), patch(
"core.mda.arc.arc_verify", side_effect=Exception("boom")
):
out = arc.arc_result(b"raw", {"relay.example"})
assert out["dnsfail"] is False
assert out["trusted"] is False
def test_trusted_seal_exposes_aar(self):
results = [{"ams-domain": b"relay.example", "aar-value": b"i=2; mx; dkim=pass"}]
with patch(
"core.mda.arc.arc_verify", return_value=(arc.CV_Pass, results, "ok")
):
with self._outer("relay.example"), self._verify(arc.CV_Pass, results):
out = arc.arc_result(b"raw", {"relay.example"})
assert out["trusted"] is True
assert out["sealer"] == "relay.example"
assert out["aar"] == "i=2; mx; dkim=pass"
def test_untrusted_sealer_no_aar(self):
results = [{"ams-domain": b"evil.net", "aar-value": b"i=2; mx; dkim=pass"}]
with patch(
"core.mda.arc.arc_verify", return_value=(arc.CV_Pass, results, "ok")
):
out = arc.arc_result(b"raw", {"relay.example"})
assert out["trusted"] is False
assert out["sealer"] == "evil.net"
assert out["aar"] is None
def test_cv_fail_untrusted(self):
# Outermost sealer IS listed, so we verify — but the chain fails.
results = [{"ams-domain": b"relay.example", "aar-value": b"i=2; mx; dkim=pass"}]
with patch("core.mda.arc.arc_verify", return_value=(b"fail", results, "bad")):
with self._outer("relay.example"), self._verify(b"fail", results):
out = arc.arc_result(b"raw", {"relay.example"})
assert out["trusted"] is False
assert out["aar"] is None
def test_no_arc_set(self):
with patch("core.mda.arc.arc_verify", return_value=(b"none", [], "no arc")):
# Real b"raw" has no chain: the cheap gate short-circuits before verify.
with patch("core.mda.arc.arc_verify") as mock_verify:
out = arc.arc_result(b"raw", {"relay.example"})
assert out == {"trusted": False, "sealer": None, "aar": None, "dnsfail": False}
mock_verify.assert_not_called()
class TestArcFastPath:
"""Q1: full crypto/DNS verification runs ONLY for a claimed-trusted sealer."""
def test_unlisted_sealer_skips_verification(self):
with patch(
"core.mda.arc._outermost_sealer", return_value=("evil.net", 1)
), patch("core.mda.arc.arc_verify") as mock_verify:
out = arc.arc_result(b"raw", {"relay.example"})
assert out["trusted"] is False
assert out["sealer"] == "evil.net"
assert out["dnsfail"] is False
mock_verify.assert_not_called()
def test_no_chain_skips_verification(self):
with patch(
"core.mda.arc._outermost_sealer", return_value=(None, 0)
), patch("core.mda.arc.arc_verify") as mock_verify:
out = arc.arc_result(b"raw", {"relay.example"})
assert out["trusted"] is False
assert out["sealer"] is None
mock_verify.assert_not_called()
def test_overlong_chain_skips_verification(self):
# One past the cap (_MAX_ARC_INSTANCES = 20) is refused without verifying.
with patch(
"core.mda.arc._outermost_sealer", return_value=("relay.example", 21)
), patch("core.mda.arc.arc_verify") as mock_verify:
out = arc.arc_result(b"raw", {"relay.example"})
assert out["trusted"] is False
mock_verify.assert_not_called()
def test_at_cap_still_verifies(self):
# Exactly at the cap is still verified.
results = [{"ams-domain": b"relay.example", "aar-value": b"x"}]
with patch(
"core.mda.arc._outermost_sealer", return_value=("relay.example", 20)
), patch(
"core.mda.arc.arc_verify", return_value=(arc.CV_Pass, results, "ok")
) as mock_verify:
out = arc.arc_result(b"raw", {"relay.example"})
assert out["trusted"] is True
mock_verify.assert_called_once()
def test_empty_allowlist_never_verifies(self):
# Fail closed: an empty allowlist trusts nothing and does no crypto/DNS.
with patch("core.mda.arc.arc_verify") as mock_verify:
out = arc.arc_result(b"raw", set())
assert out["trusted"] is False
mock_verify.assert_not_called()
# A message ARC-sealed once with a throwaway 2048-bit RSA key (domain
@@ -117,14 +170,48 @@ _SEALED_B64 = (
)
# A *different* throwaway 2048-bit RSA public key — used to prove that a seal
# which fails to validate against a well-resolved key is a definite failure
# (``dnsfail=False``), not confused with an unreachable resolver.
_MISMATCHED_PUBKEY_P = (
"MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAsOMjeA+6E9jY54/PEF1q4gDbR1ai"
"ZSEbyrPvb1zDyDulEzaACYVxxQHK3iJICMK8IW+ewZH/59/WSfAoxKGwv/ua5Ad7rWhW7INk"
"L7v98eLYGDE4B6PYjtiw+xIquKoL2PUVTXXkkUDsPny5TjPk8pfpRG94Wz1dE7E1CMglEW/R2"
"MV3E6UQVBg0sTtBA/OF+PPWiKL5+5YcuSN/fuCEwwdzZ9O3x3UnLrz5GGLxwrWJsg75K4UCVj"
"OLO138VYRt9fN8qtLFs3NZ6gsphMGkZXVPy9FjC+G76PtBRfbN9m2lEMLQCeDuIBDlZ5/Mzsi"
"RKbkzdfAUrNfj1TfTqA6iuQIDAQAB"
)
def _stub_dns(name, timeout=5):
return _PUBKEY_TXT
def _verify_with_stub_dns(raw):
def _verify_with_stub_dns(raw, dnsfunc=None):
# arc_result now passes its own DNS-tracking dnsfunc; ignore it and use the
# offline stub so these tests exercise the real crypto without a network.
return dkim.arc_verify(raw, dnsfunc=_stub_dns)
class TestOutermostSealer:
"""The cheap (no crypto/DNS) outermost-sealer parse."""
def test_real_sealed_message(self):
sealer, max_i = arc._outermost_sealer(base64.b64decode(_SEALED_B64))
assert sealer == "relay.example"
assert max_i == 1
def test_no_chain(self):
assert arc._outermost_sealer(b"From: a@b\r\nSubject: x\r\n\r\nbody") == (
None,
0,
)
def test_garbage_is_no_chain(self):
sealer, max_i = arc._outermost_sealer(b"\x00\xff not a real message")
assert max_i == 0
class TestArcResultRealCrypto:
"""Exercise the real dkimpy chain verification (stub DNS, no network)."""
@@ -143,3 +230,36 @@ class TestArcResultRealCrypto:
assert out["trusted"] is False
assert out["sealer"] == "relay.example"
assert out["aar"] is None
def test_dns_lookup_raises_is_dnsfail(self):
# Real arc_verify runs, but the key lookup blows up (timeout/SERVFAIL/
# NXDOMAIN all land here): indeterminate, not forged -> dnsfail.
def _boom(name, timeout=5):
raise Exception("dns down")
with patch("core.mda.arc.get_txt", _boom):
out = arc.arc_result(self.SEALED, {"relay.example"})
assert out["trusted"] is False
assert out["dnsfail"] is True
def test_dns_lookup_empty_is_dnsfail(self):
# An empty/absent key record is treated the same way — we can't tell a
# genuinely-retired selector from a fresh-publish negative cache.
def _empty(name, timeout=5):
return ""
with patch("core.mda.arc.get_txt", _empty):
out = arc.arc_result(self.SEALED, {"relay.example"})
assert out["trusted"] is False
assert out["dnsfail"] is True
def test_bad_signature_with_dns_ok_is_not_dnsfail(self):
# DNS resolves fine but the seal doesn't validate against a DIFFERENT
# key: a definite failure, must NOT be masked as dnsfail.
def _wrong_key(name, timeout=5):
return "v=DKIM1; k=rsa; p=" + _MISMATCHED_PUBKEY_P
with patch("core.mda.arc.get_txt", _wrong_key):
out = arc.arc_result(self.SEALED, {"relay.example"})
assert out["trusted"] is False
assert out["dnsfail"] is False
src/backend/core/tests/mda/test_inbound_auth.py
+74 -48
View File
@@ -2,21 +2,23 @@
# pylint: disable=missing-function-docstring,too-many-public-methods
from datetime import timedelta
from types import SimpleNamespace
from unittest.mock import Mock, patch
from django.test import override_settings
from django.utils import timezone
import pytest
from jmap_email import parse_email
from core import factories, models
from core import enums, factories, models
from core.mda.inbound_auth import (
VERDICT_FORGED,
VERDICT_UNVERIFIED,
check_inbound_authentication,
)
from core.mda.inbound_pipeline import Decision, _make_arc_gate_step
from core.mda.inbound_pipeline import DEFERRAL_MAX_AGE, Decision, _make_arc_step
from core.mda.inbound_tasks import process_inbound_message_task
RAW_EMAIL = (
@@ -847,63 +849,87 @@ class TestProcessInboundMessageAuthIntegration:
assert values == ["fail"]
class TestArcGateStep:
"""The arc_gate pipeline step marks/drops messages lacking a trusted seal."""
class TestArcStep:
"""The ``arc`` pipeline step holds (RETRY) a message whose *trusted* seal
couldn't be verified due to a DNS failure. Untrusted-message gating now
lives in the ``arc`` spam rules (see TestArcRules), not this step."""
@staticmethod
def _ctx():
return SimpleNamespace(raw_data=b"raw", is_spam=None, arc=None)
def _ctx(created_at=None, origin=None):
return SimpleNamespace(
raw_data=b"raw",
is_spam=None,
arc=None,
inbound_message=SimpleNamespace(
id="msg1",
created_at=created_at or timezone.now(),
envelope={"origin": origin} if origin else None,
),
)
@staticmethod
def _arc(trusted=False, dnsfail=False, sealer=None):
return {"trusted": trusted, "dnsfail": dnsfail, "sealer": sealer, "aar": None}
@patch("core.mda.inbound_pipeline.arc_result")
def test_off_is_noop(self, mock_arc):
def test_no_allowlist_is_noop(self, mock_arc):
# Empty allowlist = not an enforcement posture: skip, and don't even
# pay the ARC verification cost.
ctx = self._ctx()
assert _make_arc_gate_step({})(ctx) == Decision.CONTINUE
assert ctx.is_spam is None
assert _make_arc_step({})(ctx) == Decision.CONTINUE
mock_arc.assert_not_called()
assert ctx.arc is None
@patch("core.mda.inbound_pipeline.arc_result")
def test_trusted_passes(self, mock_arc):
mock_arc.return_value = {
"trusted": True,
"dnsfail": False,
"sealer": "r",
"aar": "x",
}
def test_trusted_continues(self, mock_arc):
mock_arc.return_value = self._arc(trusted=True, sealer="relay.example")
cfg = {"trusted_arc_sealers": ["relay.example"]}
assert _make_arc_step(cfg)(self._ctx()) == Decision.CONTINUE
@patch("core.mda.inbound_pipeline.arc_result")
def test_dnsfail_for_claimed_trusted_sealer_retries(self, mock_arc):
mock_arc.return_value = self._arc(dnsfail=True, sealer="relay.example")
cfg = {"trusted_arc_sealers": ["relay.example"]}
ctx = self._ctx()
step = _make_arc_gate_step({"arc_gate": "spam"})
assert step(ctx) == Decision.CONTINUE
assert _make_arc_step(cfg)(ctx) == Decision.RETRY
# Cached for downstream reuse (rules / inbound_auth).
assert ctx.arc is mock_arc.return_value
@patch("core.mda.inbound_pipeline.arc_result")
def test_dnsfail_for_untrusted_sealer_does_not_retry(self, mock_arc):
# A DNS blip against a sealer we don't list is not our problem to hold —
# it's a definite untrusted verdict handled by the arc rules.
mock_arc.return_value = self._arc(dnsfail=True, sealer="evil.net")
cfg = {"trusted_arc_sealers": ["relay.example"]}
assert _make_arc_step(cfg)(self._ctx()) == Decision.CONTINUE
@patch("core.mda.inbound_pipeline.arc_result")
def test_definite_untrusted_does_not_retry(self, mock_arc):
# The step never gates untrusted messages (that's the rules' job); it
# only holds on dnsfail.
mock_arc.return_value = self._arc(trusted=False, sealer="evil.net")
cfg = {"trusted_arc_sealers": ["relay.example"]}
ctx = self._ctx()
assert _make_arc_step(cfg)(ctx) == Decision.CONTINUE
assert ctx.is_spam is None
@patch("core.mda.inbound_pipeline.arc_result")
def test_spam_marks_untrusted(self, mock_arc):
mock_arc.return_value = {
"trusted": False,
"dnsfail": False,
"sealer": None,
"aar": None,
}
ctx = self._ctx()
assert _make_arc_gate_step({"arc_gate": "spam"})(ctx) == Decision.CONTINUE
assert ctx.is_spam is True
def test_dnsfail_past_deferral_window_reclassified_untrusted(self, mock_arc):
# After the retry window, an unresolvable claimed-trusted seal stops
# being held: dnsfail is cleared so the arc_verdict rules see "untrusted"
# (and an untrusted->drop rule can fire) instead of force-delivery.
mock_arc.return_value = self._arc(dnsfail=True, sealer="relay.example")
cfg = {"trusted_arc_sealers": ["relay.example"]}
old = timezone.now() - DEFERRAL_MAX_AGE - timedelta(hours=1)
ctx = self._ctx(created_at=old)
assert _make_arc_step(cfg)(ctx) == Decision.CONTINUE
assert ctx.arc["dnsfail"] is False
@patch("core.mda.inbound_pipeline.arc_result")
def test_drop_untrusted(self, mock_arc):
mock_arc.return_value = {
"trusted": False,
"dnsfail": False,
"sealer": None,
"aar": None,
}
assert _make_arc_gate_step({"arc_gate": "drop"})(self._ctx()) == Decision.DROP
@patch("core.mda.inbound_pipeline.arc_result")
def test_dnsfail_no_action(self, mock_arc):
mock_arc.return_value = {
"trusted": False,
"dnsfail": True,
"sealer": None,
"aar": None,
}
ctx = self._ctx()
assert _make_arc_gate_step({"arc_gate": "drop"})(ctx) == Decision.CONTINUE
assert ctx.is_spam is None
def test_widget_origin_skips_arc(self, mock_arc):
# Widget submissions carry no seal — ARC gating must not touch them.
cfg = {"trusted_arc_sealers": ["relay.example"]}
ctx = self._ctx(origin=enums.InboundOrigin.WIDGET)
assert _make_arc_step(cfg)(ctx) == Decision.CONTINUE
mock_arc.assert_not_called()
assert ctx.arc is None
src/backend/core/tests/mda/test_spam_processing.py
+141 -20
View File
@@ -11,12 +11,14 @@ import pytest
import requests
from jmap_email import parse_email
from core import factories, models
from core import enums, factories, models
from core.mda.inbound import deliver_inbound_message
from core.mda.inbound_pipeline import (
Decision,
InboundContext,
_make_rspamd_step,
build_inbound_pipeline,
run_inbound_pipeline,
)
from core.mda.inbound_tasks import (
process_inbound_message_task,
@@ -319,7 +321,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_ham(self):
"""Test that ham messages are correctly identified by hardcoded rules."""
@@ -335,7 +337,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is False
assert result == "ham"
def test_check_spam_with_hardcoded_rules_no_match(self):
"""Test that messages without matching rules return None."""
@@ -388,7 +390,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is False
assert result == "ham"
def test_check_spam_with_hardcoded_rules_case_insensitive(self):
"""Test that header matching is case-insensitive."""
@@ -404,7 +406,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_value_with_colon(self):
"""Test that header values containing colons are handled correctly."""
@@ -422,7 +424,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_header_match_regex_spam(self):
"""Test that spam messages are correctly identified by header_match_regex."""
@@ -440,7 +442,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_header_match_regex_spam_no_fullmatch(self):
"""Test that spam messages are correctly identified by header_match_regex."""
@@ -476,7 +478,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_header_match_regex_pattern(self):
"""Test that regex patterns work correctly with header_match_regex."""
@@ -494,7 +496,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_regex_uppercase_metacharacter_preserved(self):
"""Regression: the regex pattern must NOT be lowercased. Lowercasing
@@ -516,7 +518,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_regex_invalid_pattern_is_skipped(self):
"""Regression: a malformed regex must be skipped (logged), not raise
@@ -541,7 +543,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_default_action(self):
"""Test that default action is spam when not specified."""
@@ -561,7 +563,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_reject_action(self):
"""Test that reject action is treated as spam."""
@@ -577,7 +579,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_no_action(self):
"""Test that no action is treated as ham."""
@@ -593,7 +595,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is False
assert result == "ham"
def test_check_spam_with_hardcoded_rules_multiple_rules_order(self):
"""Test that multiple rules are evaluated in order and first match wins."""
@@ -621,7 +623,7 @@ This is a test email body.
# Should return False (ham) because second rule matched first
# Third rule should not be evaluated
assert result is False
assert result == "ham"
def test_check_spam_with_hardcoded_rules_multiple_rules_first_match_wins(self):
"""Test that the first matching rule stops evaluation."""
@@ -646,7 +648,7 @@ This is a test email body.
# Should return True (spam) because first rule matched
# Second rule should not be evaluated
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_x_spam_single_relay(self):
"""Test that X-Spam header from relay is trusted when relay adds its own header."""
@@ -672,7 +674,7 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is True
assert result == "spam"
def test_check_spam_with_hardcoded_rules_x_spam_raw_email_relay_no_header(self):
"""Test X-Spam header is ignored with raw email when relay doesn't add header.
@@ -756,10 +758,10 @@ This is a test email body.
result = check_hardcoded_rules(parsed_email, spam_config)
# Should match the first X-Spam header (No from last relay), not the sender's (Yes)
assert result is False # ham = False (not spam)
assert result == "ham" # matched the trusted X-Spam: No -> ham
@pytest.mark.parametrize(
"trusted_relays_setting, expected_result", [(0, None), (1, False), (2, False)]
"trusted_relays_setting, expected_result", [(0, None), (1, "ham"), (2, "ham")]
)
def test_check_spam_with_hardcoded_rules_trusted_relays(
self, trusted_relays_setting, expected_result
@@ -813,7 +815,7 @@ This is a test email body.
}
result = check_hardcoded_rules(parsed_email, spam_config)
assert result is expected_result
assert result == expected_result
def test_default_ignores_sender_injected_ham_header(self):
"""By default (no trusted_relays) a sender cannot whitelist itself.
@@ -850,6 +852,125 @@ This is a test email body.
assert result is None # forged ham not honoured
class TestArcRules:
"""The ``arc_verdict`` rule condition (binary trusted/untrusted) + ``drop``."""
RAW = b"From: a@sender.example\r\nTo: b@rcpt.example\r\nSubject: hi\r\n\r\nbody"
@staticmethod
def _arc(trusted=False, dnsfail=False, sealer=None, aar=None):
return {"trusted": trusted, "dnsfail": dnsfail, "sealer": sealer, "aar": aar}
def test_untrusted_drops(self):
parsed = parse_email(self.RAW)
cfg = {"rules": [{"arc_verdict": "untrusted", "action": "drop"}]}
assert check_hardcoded_rules(parsed, cfg, arc=self._arc()) == "drop"
def test_untrusted_marks_spam(self):
parsed = parse_email(self.RAW)
cfg = {"rules": [{"arc_verdict": "untrusted", "action": "spam"}]}
assert check_hardcoded_rules(parsed, cfg, arc=self._arc()) == "spam"
def test_trusted_matches_trusted(self):
parsed = parse_email(self.RAW)
cfg = {"rules": [{"arc_verdict": "trusted", "action": "ham"}]}
arc = self._arc(trusted=True, sealer="relay.example")
assert check_hardcoded_rules(parsed, cfg, arc=arc) == "ham"
def test_trusted_does_not_match_untrusted_rule(self):
parsed = parse_email(self.RAW)
cfg = {"rules": [{"arc_verdict": "untrusted", "action": "drop"}]}
arc = self._arc(trusted=True, sealer="relay.example")
assert check_hardcoded_rules(parsed, cfg, arc=arc) is None
def test_dnsfail_matches_no_verdict(self):
# A DNS blip is indeterminate — it matches NEITHER trusted nor untrusted
# (the arc pipeline step holds a claimed-trusted one for retry instead).
parsed = parse_email(self.RAW)
arc = self._arc(dnsfail=True, sealer="relay.example")
for verdict in ("untrusted", "trusted"):
cfg = {"rules": [{"arc_verdict": verdict, "action": "drop"}]}
assert check_hardcoded_rules(parsed, cfg, arc=arc) is None
def test_no_arc_computed_never_matches(self):
parsed = parse_email(self.RAW)
cfg = {"rules": [{"arc_verdict": "untrusted", "action": "drop"}]}
assert check_hardcoded_rules(parsed, cfg, arc=None) is None
def test_unknown_verdict_skipped(self):
parsed = parse_email(self.RAW)
cfg = {"rules": [{"arc_verdict": "missing", "action": "drop"}]}
assert check_hardcoded_rules(parsed, cfg, arc=self._arc()) is None
def test_rules_evaluated_in_order(self):
parsed = parse_email(self.RAW)
cfg = {
"rules": [
{"arc_verdict": "untrusted", "action": "drop"},
{"header_match": "Subject:hi", "action": "spam"},
]
}
# trusted -> arc rule doesn't match, falls through to the header rule.
arc = self._arc(trusted=True, sealer="relay.example")
assert check_hardcoded_rules(parsed, cfg, arc=arc) == "spam"
# untrusted -> arc rule fires first, header rule never reached.
assert check_hardcoded_rules(parsed, cfg, arc=self._arc()) == "drop"
def test_drop_action_on_header_rule(self):
# ``drop`` is a first-class action for any rule, not only arc rules.
parsed = parse_email(self.RAW)
cfg = {"rules": [{"header_match": "Subject:hi", "action": "drop"}]}
assert check_hardcoded_rules(parsed, cfg) == "drop"
@pytest.mark.django_db
class TestArcPipelineIntegration:
"""End-to-end: an ``arc_verdict`` rule drops an unsealed message through the
real ``build_inbound_pipeline`` / ``run_inbound_pipeline``."""
# No ARC headers -> untrusted verdict, no crypto/DNS needed.
RAW = b"From: a@evil.example\r\nTo: b@rcpt.example\r\nSubject: hi\r\n\r\nbody"
def _ctx(self, spam_config, origin=None):
return InboundContext(
mailbox=factories.MailboxFactory(),
inbound_message=Mock(
id="i1",
created_at=timezone.now(),
is_internal=False,
envelope={"origin": origin} if origin else None,
),
recipient_email="b@rcpt.example",
raw_data=self.RAW,
parsed_email=parse_email(self.RAW),
spam_config=spam_config,
)
def test_untrusted_arc_rule_drops_through_full_pipeline(self):
ctx = self._ctx(
{
"trusted_arc_sealers": ["relay.example"],
"rules": [{"arc_verdict": "untrusted", "action": "drop"}],
}
)
decision, step = run_inbound_pipeline(build_inbound_pipeline(ctx), ctx)
assert decision == Decision.DROP
assert step == "hardcoded_rules"
def test_widget_submission_not_dropped_by_arc_rule(self):
# The same drop rule must NOT discard a widget-origin submission.
ctx = self._ctx(
{
"trusted_arc_sealers": ["relay.example"],
"rules": [{"arc_verdict": "untrusted", "action": "drop"}],
},
origin=enums.InboundOrigin.WIDGET,
)
decision, _step = run_inbound_pipeline(build_inbound_pipeline(ctx), ctx)
assert decision == Decision.CONTINUE
assert ctx.arc is None # ARC never computed for widget
@pytest.mark.django_db
class TestProcessInboundMessageTask:
"""Test the process_inbound_message_task."""
src/backend/messages/settings.py
+10 -6
View File
@@ -552,17 +552,21 @@ class Base(Configuration):
# own MTA prepends; a sender can forge any block
# above that, so only raise this to the number of
# relay hops you actually operate)
# rules : list of hardcoded header-match spam rules
# rules : list of spam rules; each is a header_match /
# header_match_regex OR an "arc_verdict" trust
# condition — "trusted" / "untrusted" — e.g.
# {"arc_verdict": "untrusted", "action": "drop"},
# with action spam / ham / drop.
# inbound_auth : sender authentication backend — one of
# "native", "rspamd", "arc",
# "authentication-results", or None to disable.
# See core.mda.inbound_auth for semantics.
# trusted_arc_sealers : list of trusted ARC sealer d= domains
# (empty = any valid seal). Used by inbound_auth
# "arc" and by arc_gate.
# arc_gate : action when a message is not sealed by a
# trusted sealer — "off" (default), "spam", or
# "drop".
# (empty = trust nothing). Used by inbound_auth
# "arc" and by "arc" spam rules. When non-empty,
# a DNS failure verifying a claimed-trusted
# sealer holds the message for retry rather than
# failing open or closed.
SPAM_CONFIG = values.DictValue({}, environ_name="SPAM_CONFIG", environ_prefix=None)
# MTA settings