procurement-ingest/tests/test_ses_auth.py

"""Unit tests for the fail-closed SES sender authentication (INFRA-107).

Fixtures mirror real SES-stamped headers observed on the two ingest
buckets on 2026-07-15: SES prepends a folded Authentication-Results
header with authserv-id amazonses.com and reports passing signers as
``dkim=pass header.i=@<domain>``.
"""

BODY = "\r\n\r\nWork Order 12345 assigned.\r\n"

# Folded exactly like real SES output (continuation lines, header.i form).
WO_SES_HEADER = (
    "Authentication-Results: amazonses.com;\r\n"
    " spf=pass (spfCheck: domain of seahaven.com designates 209.85.219.70 as"
    " permitted sender) client-ip=209.85.219.70;"
    " envelope-from=apm+bnc@seahaven.com; helo=mail-qv1-f70.google.com;\r\n"
    " dkim=pass header.i=@seahaven.com;\r\n"
    " dmarc=none header.from=hxgnsmartcloud.com;\r\n"
)

# Real PO traffic carries two dkim=pass clauses; amazonses.com must not be
# sufficient on its own (every SES customer's mail passes for it).
PO_SES_HEADER = (
    "Authentication-Results: amazonses.com;\r\n"
    " spf=pass (spfCheck: domain of mail.coupahost.com designates"
    " 54.240.41.238 as permitted sender) client-ip=54.240.41.238;\r\n"
    " dkim=pass header.i=@amazonses.com;\r\n"
    " dkim=pass header.i=@amazon.coupahost.com;\r\n"
    " dmarc=pass header.from=amazon.coupahost.com;\r\n"
)

FROM_TO = (
    "From: APM <noreply@hxgnsmartcloud.com>\r\n"
    "To: apm@int.seahaven.com\r\n"
    "Subject: WO 12345\r\n"
)


def raw(*headers: str) -> bytes:
    return ("".join(headers) + FROM_TO + BODY).encode()


class TestParseAuthenticationResults:
    def test_ses_wo_header(self, ses_auth):
        value = WO_SES_HEADER.split(":", 1)[1]
        authserv_id, passing = ses_auth.parse_authentication_results(value)
        assert authserv_id == "amazonses.com"
        assert passing == frozenset({"seahaven.com"})

    def test_ses_po_header_multiple_dkim_clauses(self, ses_auth):
        value = PO_SES_HEADER.split(":", 1)[1]
        authserv_id, passing = ses_auth.parse_authentication_results(value)
        assert authserv_id == "amazonses.com"
        assert passing == frozenset({"amazonses.com", "amazon.coupahost.com"})

    def test_header_d_form(self, ses_auth):
        _, passing = ses_auth.parse_authentication_results(
            "amazonses.com; dkim=pass header.d=Example.COM."
        )
        assert passing == frozenset({"example.com"})

    def test_case_insensitive_result(self, ses_auth):
        _, passing = ses_auth.parse_authentication_results(
            "amazonses.com; DKIM=Pass HEADER.I=@SeaHaven.COM"
        )
        assert passing == frozenset({"seahaven.com"})

    def test_dkim_fail_yields_no_domains(self, ses_auth):
        _, passing = ses_auth.parse_authentication_results(
            "amazonses.com; dkim=fail header.i=@seahaven.com"
        )
        assert passing == frozenset()

    def test_authserv_id_version_token(self, ses_auth):
        authserv_id, _ = ses_auth.parse_authentication_results(
            "amazonses.com 1; dkim=pass header.i=@seahaven.com"
        )
        assert authserv_id == "amazonses.com"

    def test_garbage_value(self, ses_auth):
        authserv_id, passing = ses_auth.parse_authentication_results(";;;")
        assert authserv_id == ""
        assert passing == frozenset()

    def test_result_token_boundary(self, ses_auth):
        # "pass-anything" / "passfail" must never be read as "pass".
        for result in ("pass-fake", "passfail"):
            _, passing = ses_auth.parse_authentication_results(
                f"amazonses.com; dkim={result} header.i=@seahaven.com"
            )
            assert passing == frozenset()

    def test_result_followed_by_comment(self, ses_auth):
        _, passing = ses_auth.parse_authentication_results(
            "amazonses.com; dkim=pass(good signature) header.i=@seahaven.com"
        )
        assert passing == frozenset({"seahaven.com"})

    def test_quoted_domain_value(self, ses_auth):
        _, passing = ses_auth.parse_authentication_results(
            'amazonses.com; dkim=pass header.i="@seahaven.com"'
        )
        assert passing == frozenset({"seahaven.com"})

    def test_folding_inside_dkim_clause(self, ses_auth):
        _, passing = ses_auth.parse_authentication_results(
            "amazonses.com;\r\n dkim=pass\r\n header.i=@seahaven.com"
        )
        assert passing == frozenset({"seahaven.com"})

    def test_quoted_semicolon_in_envelope_from_is_not_split(self, ses_auth):
        # RFC 5321 quoted-local-part MAIL FROM can carry ';' and spaces; SES
        # echoes it into envelope-from=. The ';' inside the quotes must stay
        # part of the one envelope-from clause, never a synthetic dkim clause.
        value = (
            "amazonses.com; spf=pass client-ip=1.2.3.4;"
            ' envelope-from="x; dkim=pass header.i=@amazon.coupahost.com"@attacker.com;'
            " helo=mail.attacker.com; dkim=fail header.i=@attacker.com;"
        )
        _, passing = ses_auth.parse_authentication_results(value)
        assert passing == frozenset()

    def test_quoted_pair_in_envelope_from_is_not_split(self, ses_auth):
        # A quoted-pair (\") inside the quoted local part must not prematurely
        # end the quoted-string and re-expose the smuggled tokens.
        value = (
            'amazonses.com; envelope-from="a\\"; dkim=pass header.d=seahaven.com"@evil.com;'
            " dkim=fail header.i=@evil.com;"
        )
        _, passing = ses_auth.parse_authentication_results(value)
        assert passing == frozenset()

    def test_helo_injection_is_not_split(self, ses_auth):
        # helo= is attacker-influenced too; a crafted value quoting a fake
        # methodspec must not manufacture a passing clause.
        value = (
            'amazonses.com; helo="h; dkim=pass header.i=@seahaven.com";'
            " dkim=fail header.i=@attacker.com;"
        )
        _, passing = ses_auth.parse_authentication_results(value)
        assert passing == frozenset()

    def test_comment_embedded_header_i_is_ignored(self, ses_auth):
        # A CFWS comment carrying a fake header.i must be stripped before
        # domain extraction, so only the real (failing) result is considered.
        _, passing = ses_auth.parse_authentication_results(
            "amazonses.com; dkim=fail (header.i=@seahaven.com) header.i=@attacker.com"
        )
        assert passing == frozenset()

    def test_comment_hiding_semicolon_does_not_split(self, ses_auth):
        # A ';' inside a comment is not a clause separator either.
        _, passing = ses_auth.parse_authentication_results(
            "amazonses.com; spf=pass (note: a; b) client-ip=1.2.3.4;"
            " dkim=pass header.i=@seahaven.com"
        )
        assert passing == frozenset({"seahaven.com"})

    def test_unbalanced_quote_fails_closed(self, ses_auth):
        # An unterminated quoted-string is malformed; refuse to parse it so a
        # dkim=pass clause "swallowed" by the runaway quote can't be salvaged
        # (and, conversely, a runaway quote can't be used to mis-tokenise).
        authserv_id, passing = ses_auth.parse_authentication_results(
            'amazonses.com; envelope-from="oops@attacker.com;'
            " dkim=pass header.i=@seahaven.com"
        )
        assert authserv_id == ""
        assert passing == frozenset()

    def test_unbalanced_comment_fails_closed(self, ses_auth):
        # An unterminated comment is malformed and must fail closed.
        authserv_id, passing = ses_auth.parse_authentication_results(
            "amazonses.com; dkim=pass header.i=@seahaven.com (runaway comment"
        )
        assert authserv_id == ""
        assert passing == frozenset()


class TestEvaluateSenderAuthentication:
    def test_ses_stamped_pass_accepted(self, ses_auth):
        accepted, reason, detail = ses_auth.evaluate_sender_authentication(
            raw(WO_SES_HEADER), {"seahaven.com"}
        )
        assert accepted
        assert reason == "authenticated"
        assert detail["matched_domains"] == ["seahaven.com"]

    def test_po_pass_accepted_on_coupa_domain(self, ses_auth):
        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
            raw(PO_SES_HEADER), {"amazon.coupahost.com"}
        )
        assert accepted
        assert reason == "authenticated"

    def test_amazonses_identity_alone_is_not_allowlisted(self, ses_auth):
        # Any SES customer's outbound mail passes DKIM for amazonses.com,
        # so a pass for it must not satisfy a coupahost-only allowlist.
        forged_via_ses = (
            "Authentication-Results: amazonses.com;\r\n"
            " spf=pass client-ip=54.240.41.1;\r\n"
            " dkim=pass header.i=@amazonses.com;\r\n"
        )
        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
            raw(forged_via_ses), {"amazon.coupahost.com"}
        )
        assert not accepted
        assert reason == "dkim_domain_not_allowlisted"

    def test_forged_ar_below_failing_ses_header_rejected(self, ses_auth):
        # SES's (topmost) header says dkim=fail; the attacker smuggled a
        # perfect-looking AR header inside the message. Only the topmost
        # header may be consulted.
        ses_fail = (
            "Authentication-Results: amazonses.com;\r\n"
            " spf=fail client-ip=203.0.113.7;\r\n"
            " dkim=fail header.i=@seahaven.com;\r\n"
            " dmarc=fail header.from=hxgnsmartcloud.com;\r\n"
        )
        forged = (
            "Authentication-Results: amazonses.com;\r\n"
            " dkim=pass header.i=@seahaven.com;\r\n"
        )
        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
            raw(ses_fail, forged), {"seahaven.com"}
        )
        assert not accepted
        assert reason == "no_passing_dkim_signature"

    def test_missing_ar_header_rejected(self, ses_auth):
        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
            raw(), {"seahaven.com"}
        )
        assert not accepted
        assert reason == "authentication_results_missing"

    def test_untrusted_authserv_id_rejected(self, ses_auth):
        attacker_ar = (
            "Authentication-Results: mail.attacker.example;\r\n"
            " dkim=pass header.i=@seahaven.com;\r\n"
        )
        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
            raw(attacker_ar), {"seahaven.com"}
        )
        assert not accepted
        assert reason == "untrusted_authserv_id"

    def test_unaligned_domain_rejected(self, ses_auth):
        evil = (
            "Authentication-Results: amazonses.com;\r\n"
            " dkim=pass header.i=@evil.example.com;\r\n"
        )
        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
            raw(evil), {"seahaven.com"}
        )
        assert not accepted
        assert reason == "dkim_domain_not_allowlisted"

    def test_lookalike_domain_rejected(self, ses_auth):
        # Substring containment must not match: notseahaven.com != seahaven.com
        lookalike = (
            "Authentication-Results: amazonses.com;\r\n"
            " dkim=pass header.i=@notseahaven.com;\r\n"
        )
        accepted, _, _ = ses_auth.evaluate_sender_authentication(
            raw(lookalike), {"seahaven.com"}
        )
        assert not accepted

    def test_empty_allowlist_fails_closed(self, ses_auth):
        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
            raw(WO_SES_HEADER), frozenset()
        )
        assert not accepted
        assert reason == "allowlist_not_configured"

    def test_envelope_from_clause_injection_rejected(self, ses_auth):
        # Full forged email: attacker's quoted MAIL FROM is echoed by SES into
        # its own (topmost, genuinely SES-stamped) Authentication-Results as
        # envelope-from=, trying to smuggle a dkim=pass for an allowlisted
        # domain. The real DKIM verdict is fail. Must fail closed.
        injected = (
            "Authentication-Results: amazonses.com;\r\n"
            " spf=pass (spfCheck: domain of attacker.com designates 1.2.3.4 as"
            " permitted sender) client-ip=1.2.3.4;\r\n"
            ' envelope-from="x; dkim=pass header.i=@amazon.coupahost.com"@attacker.com;'
            " helo=mail.attacker.com;\r\n"
            " dkim=fail header.i=@attacker.com;\r\n"
            " dmarc=fail header.from=attacker.com;\r\n"
        )
        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
            raw(injected), {"amazon.coupahost.com", "seahaven.com"}
        )
        assert not accepted
        assert reason == "no_passing_dkim_signature"


class TestAuthenticateInboundEmail:
    S3_KEY = "s3://bucket/inbound/abc123"

    def test_accepts_with_configured_allowlist(self, ses_auth, monkeypatch):
        monkeypatch.setenv("ALLOWED_DKIM_DOMAINS", "seahaven.com")
        assert ses_auth.authenticate_inbound_email(raw(WO_SES_HEADER), self.S3_KEY)

    def test_allowlist_is_comma_separated_and_normalized(self, ses_auth, monkeypatch):
        # Stray spaces, case, a leading @, and a trailing dot all normalize.
        monkeypatch.setenv("ALLOWED_DKIM_DOMAINS", " Other.Example , @SEAHAVEN.com. ,")
        assert ses_auth.authenticate_inbound_email(raw(WO_SES_HEADER), self.S3_KEY)

    def test_env_var_unset_fails_closed(self, ses_auth, monkeypatch, caplog):
        monkeypatch.delenv("ALLOWED_DKIM_DOMAINS", raising=False)
        assert not ses_auth.authenticate_inbound_email(raw(WO_SES_HEADER), self.S3_KEY)
        assert "allowlist_not_configured" in caplog.text
        assert self.S3_KEY in caplog.text

    def test_rejection_logs_reason_and_key(self, ses_auth, monkeypatch, caplog):
        monkeypatch.setenv("ALLOWED_DKIM_DOMAINS", "seahaven.com")
        assert not ses_auth.authenticate_inbound_email(raw(), self.S3_KEY)
        assert "sender_auth_rejected" in caplog.text
        assert "authentication_results_missing" in caplog.text
        assert self.S3_KEY in caplog.text