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()


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"


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):
        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