procurement-ingest/lambdas/po/email_processor/ses_auth.py

"""Fail-closed SES sender authentication (INFRA-107).

SES Email Receiving *prepends* its own trace headers -- including an
``Authentication-Results`` header whose authserv-id is ``amazonses.com`` --
to the top of the raw MIME it writes to S3. Everything below those
prepended headers (the From header, any additional Authentication-Results
copies) is attacker-controlled, so ONLY the topmost Authentication-Results
header is trusted, and only when its authserv-id is ``amazonses.com``.

An email is accepted only when that header carries ``dkim=pass`` for a
domain in the ``ALLOWED_DKIM_DOMAINS`` allowlist (a comma-separated Lambda
environment variable set by the CDK stack). Every other outcome fails
closed and the email is rejected:

- ``ALLOWED_DKIM_DOMAINS`` unset or empty
- no Authentication-Results header at all
- topmost header unparseable or from an authserv-id other than SES
- no ``dkim=pass`` clause
- ``dkim=pass`` only for domains outside the allowlist

Observed SES format (2026-07-15, both ingest buckets)::

    Authentication-Results: amazonses.com;
     spf=pass (spfCheck: ...) client-ip=...; envelope-from=...; helo=...;
     dkim=pass header.i=@seahaven.com;
     dmarc=none header.from=hxgnsmartcloud.com;

Note SES reports the passing DKIM identity as ``header.i=@<domain>``
(RFC 6376 AUID), not ``header.d=``; the parser accepts both.
"""

import email.parser
import email.policy
import json
import logging
import os
import re

logger = logging.getLogger()

ALLOWED_DKIM_DOMAINS_ENV = "ALLOWED_DKIM_DOMAINS"
SES_AUTHSERV_ID = "amazonses.com"

# One resinfo clause of an Authentication-Results value, e.g.
# "dkim=pass header.i=@seahaven.com". The clause must START with the
# method=result pair; header.d= / header.i= may appear anywhere after it.
# The result token must be terminated by end-of-clause, whitespace, or a
# comment so "dkim=pass-anything" can never be read as "pass".
_DKIM_RESULT_RE = re.compile(r"^dkim\s*=\s*([a-z0-9]+)(?=$|[\s(])", re.IGNORECASE)
_HEADER_D_RE = re.compile(r"header\.d\s*=\s*\"?([^\s\";]+)", re.IGNORECASE)
_HEADER_I_RE = re.compile(r"header\.i\s*=\s*\"?([^\s\";]+)", re.IGNORECASE)


def get_allowed_dkim_domains() -> frozenset:
    """Read the DKIM-domain allowlist from the environment (may be empty)."""
    raw = os.environ.get(ALLOWED_DKIM_DOMAINS_ENV, "")
    return frozenset(
        d.strip().lower().lstrip("@").rstrip(".") for d in raw.split(",") if d.strip()
    )


def _unfold(value: str) -> str:
    """Collapse RFC 5322 folding whitespace into single spaces."""
    return re.sub(r"[\r\n\t ]+", " ", value).strip()


def parse_authentication_results(value: str) -> tuple:
    """Parse one Authentication-Results header value.

    Returns ``(authserv_id, passing_dkim_domains)`` where the domains are
    the lowercased d=/i= domains of every ``dkim=pass`` clause. Malformed
    input yields ``("", frozenset())``, which callers treat as a rejection.
    """
    text = _unfold(value)
    clauses = [c.strip() for c in text.split(";")]
    if not clauses or not clauses[0]:
        return "", frozenset()

    # First clause is the authserv-id, optionally followed by a version
    # token ("amazonses.com 1"); take only the first token.
    authserv_id = clauses[0].split()[0].strip('"').lower()

    passing = set()
    for clause in clauses[1:]:
        match = _DKIM_RESULT_RE.match(clause)
        if not match or match.group(1).lower() != "pass":
            continue
        d_match = _HEADER_D_RE.search(clause)
        if d_match:
            passing.add(d_match.group(1).lower().rstrip("."))
            continue
        i_match = _HEADER_I_RE.search(clause)
        if i_match:
            # AUID (header.i) is "local-part@domain" or "@domain";
            # the signing domain is everything after the last "@".
            identity = i_match.group(1).lower().rstrip(".")
            domain = identity.rsplit("@", 1)[-1]
            if domain:
                passing.add(domain)
    return authserv_id, frozenset(passing)


def evaluate_sender_authentication(raw_email: bytes, allowed_domains) -> tuple:
    """Evaluate the SES-stamped verdicts in a raw MIME message.

    Returns ``(accepted, reason, detail)``. Pure function of its inputs so
    it can be unit-tested without touching the environment.
    """
    if not allowed_domains:
        return False, "allowlist_not_configured", {}

    try:
        # compat32 keeps header values as raw strings (we unfold ourselves)
        # and never raises on structurally odd headers; headersonly avoids
        # parsing the body at all.
        msg = email.parser.BytesParser(policy=email.policy.compat32).parsebytes(
            raw_email, headersonly=True
        )
    except Exception:
        return False, "unparseable_message", {}

    ar_headers = msg.get_all("Authentication-Results") or []
    if not ar_headers:
        return False, "authentication_results_missing", {}

    # SES prepends its trace headers, so index 0 is the SES-stamped copy.
    # Any Authentication-Results header further down arrived inside the
    # message (attacker-suppliable) and is deliberately ignored.
    authserv_id, passing = parse_authentication_results(str(ar_headers[0]))
    detail = {
        "authserv_id": authserv_id,
        "passing_dkim_domains": sorted(passing),
    }

    if authserv_id != SES_AUTHSERV_ID:
        return False, "untrusted_authserv_id", detail
    if not passing:
        return False, "no_passing_dkim_signature", detail

    matched = passing & set(allowed_domains)
    if not matched:
        return False, "dkim_domain_not_allowlisted", detail

    detail["matched_domains"] = sorted(matched)
    return True, "authenticated", detail


def authenticate_inbound_email(raw_email: bytes, s3_key: str) -> bool:
    """Fail-closed gate used by the S3-triggered handlers.

    On rejection: logs a structured warning with the reason and S3 key and
    returns False. Callers skip the message and return normally, so
    rejected mail never errors the invocation (no retries, no DLQ spam).
    """
    allowed = get_allowed_dkim_domains()
    accepted, reason, detail = evaluate_sender_authentication(raw_email, allowed)
    if accepted:
        logger.info(json.dumps({"event": "sender_auth_ok", "s3_key": s3_key, **detail}))
        return True
    logger.warning(
        json.dumps(
            {
                "event": "sender_auth_rejected",
                "reason": reason,
                "s3_key": s3_key,
                "allowed_dkim_domains": sorted(allowed),
                **detail,
            }
        )
    )
    return False