chore: retrigger CI (no run recorded for 7c74ac1)

The SES-stamped Authentication-Results value echoes attacker-controlled
SMTP-session tokens (envelope-from, helo, header.from) as their own
semicolon-delimited property clauses. A naive split(";") tore an RFC 5321
quoted-local-part MAIL FROM apart and manufactured a forged dkim=pass
clause, so a fully spoofed email was accepted on the genuinely
SES-stamped topmost header. Tokenise comment- and quoted-string-aware
(RFC 8601 / RFC 5322): strip CFWS comments, split clauses only on
semicolons outside a quoted-string, and fail closed on unbalanced
quotes/comments so a ';' inside a quoted pvalue can never start a clause.

Rejected mail returns normally (no error, no retry, no DLQ message), so a
signing-domain drift or a wrong allowlist would silently discard 100% of
legitimate mail while every alarm stayed green. Add a CloudWatch Logs
metric filter + alarm on the sender_auth_rejected warning to both stacks
so a false-reject storm pages instead of vanishing. This is also the
safety net for the WO seahaven.com allowlist assumption, which must be
validated against a live SES-stamped header (a plain Gmail auto-forward
re-signs under the sending Workspace domain, not seahaven.com).

Refs: INFRA-107

README.md

@@ -16,12 +16,13 @@ Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed by Claude H
 1. Coupa sends a PO email (new, revision, or cancellation).
 2. SES (`INBOUND_MAIL` rule set) drops the raw MIME into `s3://po-ingest-emails-{AccountId}/inbound/`.
 3. S3 `ObjectCreated` triggers the `po-email-processor` Lambda.
-4. Claude extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year).
-5. Conditional write to DynamoDB:
+4. Fail-closed sender authentication (INFRA-107): the SES-stamped `Authentication-Results` header must show `dkim=pass` for `amazon.coupahost.com` (see [Sender authentication](#sender-authentication-infra-107)); otherwise the email is logged and dropped.
+5. Claude extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year).
+6. Conditional write to DynamoDB:
    - `new_po` — idempotent insert (no-op if PO exists)
    - `revision` — unconditional overwrite
    - `cancellation` — marks existing row `Cancelled`
-6. DynamoDB Streams feeds downstream consumers:
+7. DynamoDB Streams feeds downstream consumers:
    - **LedgerFlow** (`seahaven-slack-bot/po-sync`) — daily KB sync
    - **Site extractor** (`po-ingest-site-extractor`) — real-time site address extraction into `verified-sites` table
 
@@ -46,8 +47,9 @@ Amazon APM work order emails (from Hexagon EAM / HxGN SmartCloud) are received a
 2. Gmail filter forwards APM emails to `apm@int.seahaven.com` (SES).
 3. SES drops the raw MIME into `s3://workorder-ingest-emails-{AccountId}/inbound/`.
 4. S3 triggers the `workorder-email-processor` Lambda.
-5. Claude extracts structured JSON (work order ID, site code, severity, priority, dates, assigned technician).
-6. Work order upserted to `WorkOrders`, event/comment appended to `WorkOrderComments`.
+5. Fail-closed sender authentication (INFRA-107): the SES-stamped `Authentication-Results` header must show `dkim=pass` for the domain that re-signs the forward (currently allowlisted as `seahaven.com` — see the validation caveat under [Sender authentication](#sender-authentication-infra-107)); otherwise the email is logged and dropped.
+6. Claude extracts structured JSON (work order ID, site code, severity, priority, dates, assigned technician).
+7. Work order upserted to `WorkOrders`, event/comment appended to `WorkOrderComments`.
 
 **Lambdas** (`lambdas/wo/`):
 | Function | Trigger | Purpose |
@@ -71,6 +73,25 @@ All Lambdas: Python 3.12, ARM64, 60-day log retention.
 
 **SES:** Both stacks add rules to the shared `INBOUND_MAIL` receipt rule set on `int.seahaven.com`.
 
+### Sender authentication (INFRA-107)
+
+The `From` header and any `Authentication-Results` header inside the raw MIME are attacker-forgeable, so neither is trusted. Instead, both email processors (`lambdas/*/email_processor/ses_auth.py`) authenticate the sender against the verdicts SES itself stamps at delivery time, failing closed:
+
+1. Take **only the topmost** `Authentication-Results` header (SES prepends its trace headers; any lower copies arrived inside the message and are ignored).
+2. Require its authserv-id to be `amazonses.com`.
+3. Require a `dkim=pass` clause whose `header.d=`/`header.i=` domain is in the pipeline's allowlist.
+
+The allowlist is the `ALLOWED_DKIM_DOMAINS` Lambda environment variable (comma-separated, set per stack in CDK — no code change needed to adjust):
+
+| Pipeline | `ALLOWED_DKIM_DOMAINS` | Why |
+|---|---|---|
+| Work orders | `seahaven.com` | APM mail reaches `apm@int.seahaven.com` via a forward off `amazon@seahavenind.com`; the allowlist trusts the domain that **re-signs** DKIM on that forward (the original `hxgnsmartcloud.com` signature does not survive it). **Must be validated against a live SES-stamped header** — a plain Gmail auto-forward re-signs under the *sending Workspace* domain (`seahavenind.com` / a `*.gappssmtp.com` key), **not** `seahaven.com`; only a Google Group (or Workspace routing) with "sign as `seahaven.com`" produces `dkim=pass header.i=@seahaven.com`. If the observed re-signing domain differs, update this value (do **not** widen it to a shared key like `*.gappssmtp.com`, which any Google customer's mail would pass). The `workorder-email-processor-sender-auth-rejected` alarm pages if this assumption is wrong instead of silently dropping every work order. |
+| Purchase orders | `amazon.coupahost.com` | Coupa signs as `amazon.coupahost.com`. `amazonses.com` also passes but is deliberately not allowlisted — every SES customer's mail passes for it |
+
+> **Clause-injection hardening:** SES echoes attacker-controlled SMTP-session tokens (`envelope-from`, `helo`, `header.from`) into its own `Authentication-Results` value, and an RFC 5321 quoted-local-part MAIL FROM may legally contain `;` and spaces. The parser therefore tokenises comment- and quoted-string-aware (RFC 8601 / RFC 5322): CFWS comments `(...)` are stripped and clauses are split only on semicolons **outside** a quoted string, so a `;` inside a quoted `envelope-from=` value can never be torn into a forged `dkim=pass` clause.
+
+On any failure (env var unset, header missing/unparseable, verdict fail, unaligned domain) the processor logs a structured `sender_auth_rejected` warning with the reason and S3 key, skips the email, and returns normally — rejected mail never triggers Lambda retries or DLQ messages, but the `<fn>-sender-auth-rejected` CloudWatch alarm (see [CloudWatch alarms](#cloudwatch-alarms)) pages on a rejection spike so a drift-induced outage is not silent. Unit tests live in `tests/test_ses_auth.py`.
+
 **Failure handling (INFRA-41):** Each email-processor is async-invoked (S3 → Lambda). Both have a CDK-managed SQS dead-letter queue (`dead_letter_queue=`, 14-day retention, SSL-enforced) so a failed parse is captured rather than silently dropped after Lambda's retries.
 
 ### CloudWatch alarms
@@ -84,6 +105,9 @@ Every alarm is **ALARM-only** (no OK action), sends to the shared `site-alerts`
 | `<fn>-errors` | `po-email-processor`, `po-ingest-site-extractor`, `workorder-email-processor` | `Errors` Sum, 5 min, `> 0`, eval 1 |
 | `<fn>-throttles` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui`, `workorder-email-processor` | `Throttles` Sum, 5 min, `> 0`, eval 1 |
 | `<fn>-duration` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui` (p99); `workorder-email-processor` (p95) | `Duration` percentile, 5 min, `>= 45000` ms (75% of the 60s timeout), eval 3 / datapoints 2 |
+| `<fn>-sender-auth-rejected` | `po-email-processor`, `workorder-email-processor` | Log-metric-filter count (namespace `Seahaven/ProcurementIngest`, `default_value=0`) on `sender_auth_rejected` warnings, `Sum` 15 min, `>= 3`, eval 1 |
+
+The `<fn>-sender-auth-rejected` alarm closes the silent-drop gap in INFRA-107: a rejected email returns normally (no error, no retry, no DLQ message), so without a log-metric filter a signing-domain drift or a wrong allowlist would discard 100% of legitimate mail while every other alarm stayed green. `>= 3` rejections in 15 minutes tolerates the occasional spoofed probe to the internal ingest address but pages quickly on a genuine false-reject storm; the threshold is easy to tune in the CDK helper.
 
 The `<fn>-duration` and `<fn>-throttles` alarms for `po-email-processor` and `workorder-email-processor` supersede the orphaned, CLI-created `Lambda-Duration-*` / `Lambda-Throttles-*` alarms (deleted post-deploy).
 
@@ -187,4 +211,7 @@ shared/
 scripts/
   reprocess.py
   backfill_sites.py
+tests/
+  conftest.py          # AWS env stubs + per-pipeline module loader
+  test_ses_auth.py     # Sender-authentication parser tests (INFRA-107)
 ```

cdk/po_stack.py

@@ -77,6 +77,67 @@ def _add_ddb_alarms(scope, id_prefix, table, alarm_name_prefix, alarm_topic):
     ).add_alarm_action(cw_actions.SnsAction(alarm_topic))
 
 
+# CloudWatch namespace for the log-derived sender-authentication metrics.
+_SENDER_AUTH_METRIC_NAMESPACE = "Seahaven/ProcurementIngest"
+
+
+def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic):
+    """Metric-filter + alarm on ``sender_auth_rejected`` warnings (INFRA-107).
+
+    A rejected inbound email is skipped without erroring the invocation, so it
+    is invisible to the Errors/Throttles/DLQ alarms. This turns the structured
+    warning log into a CloudWatch metric and pages when rejections spike --
+    catching a silent false-reject storm (allowlist wrong, signing-domain
+    drift, SES header-format change) that would otherwise discard legitimate
+    mail while the pipeline reports healthy.
+
+    ALARM-only SnsAction to site-alerts; no OK action. The metric filter reads
+    the function's own log group (imported by the deterministic
+    ``/aws/lambda/<fn>`` name, created by the function's log_retention). A plain
+    substring pattern is used because Lambda prefixes each line with its own
+    level/timestamp/request-id, so the JSON payload is not a standalone JSON
+    log event a `{$.event=...}` pattern could match.
+    """
+    metric_name = f"{function_name}-sender-auth-rejected"
+    logs.MetricFilter(
+        scope,
+        f"{id_prefix}SenderAuthRejectedFilter",
+        log_group=logs.LogGroup.from_log_group_name(
+            scope,
+            f"{id_prefix}LogGroup",
+            f"/aws/lambda/{function_name}",
+        ),
+        filter_pattern=logs.FilterPattern.literal('"sender_auth_rejected"'),
+        metric_namespace=_SENDER_AUTH_METRIC_NAMESPACE,
+        metric_name=metric_name,
+        metric_value="1",
+        default_value=0,
+    )
+
+    # >=3 rejections in a 15-min window pages: a lone spoofed probe to the
+    # (obscure, internal) ingest address is tolerated, but a genuine
+    # false-reject storm -- where legitimate senders are being dropped -- trips
+    # quickly. Threshold is intentionally conservative and easy to tune.
+    cloudwatch.Metric(
+        namespace=_SENDER_AUTH_METRIC_NAMESPACE,
+        metric_name=metric_name,
+        period=Duration.minutes(15),
+        statistic="Sum",
+    ).create_alarm(
+        scope,
+        f"{id_prefix}SenderAuthRejectedAlarm",
+        alarm_name=f"{function_name}-sender-auth-rejected",
+        alarm_description=(
+            f"{function_name} rejected inbound mail on sender authentication "
+            "(possible allowlist/DKIM-domain drift silently dropping real mail)"
+        ),
+        threshold=3,
+        evaluation_periods=1,
+        comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,
+        treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
+    ).add_alarm_action(cw_actions.SnsAction(alarm_topic))
+
+
 class PoIngestStack(Stack):
     def __init__(self, scope: Construct, construct_id: str, **kwargs):
         super().__init__(scope, construct_id, **kwargs)
@@ -174,7 +235,7 @@ class PoIngestStack(Stack):
                         "-c",
                         "pip install --platform manylinux2014_aarch64 --only-binary=:all: "
                         "-r requirements.txt -t /asset-output && "
-                        "cp handler.py /asset-output/",
+                        "cp handler.py ses_auth.py /asset-output/",
                     ],
                 ),
             ),
@@ -185,6 +246,14 @@ class PoIngestStack(Stack):
             environment={
                 "PO_TABLE": "purchase-orders",
                 "ANTHROPIC_API_KEY_SECRET_ARN": anthropic_secret.secret_arn,
+                # Fail-closed sender auth (INFRA-107): the handler only
+                # accepts mail whose SES-stamped Authentication-Results
+                # header carries dkim=pass for one of these domains.
+                # Observed on live traffic 2026-07-15: Coupa PO mail passes
+                # DKIM for amazon.coupahost.com (and amazonses.com, which is
+                # deliberately NOT allowlisted — every SES customer's mail
+                # passes that). Unset/empty ⇒ the handler rejects all mail.
+                "ALLOWED_DKIM_DOMAINS": "amazon.coupahost.com",
             },
         )
 
@@ -210,6 +279,20 @@ class PoIngestStack(Stack):
             treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
         ).add_alarm_action(cw_actions.SnsAction(alarm_topic))
 
+        # --- Sender-auth rejection alarm (INFRA-107) ---
+        # A rejected email (bad/unaligned DKIM verdict) returns normally, so it
+        # produces NO Lambda error, NO DLQ message and NO retry -- only a
+        # `sender_auth_rejected` warning log. Without this metric filter + alarm a
+        # domain drift (Coupa rotates its signing subdomain, SES changes its
+        # Authentication-Results format, the allowlist is wrong) would silently
+        # discard 100% of legitimate PO mail while every other alarm stays green.
+        # A CloudWatch Logs metric filter turns those warnings into a metric so a
+        # false-reject storm pages instead of vanishing. default_value=0 keeps the
+        # series populated (alarm stays OK, never INSUFFICIENT_DATA) between events.
+        _add_sender_auth_rejected_alarm(
+            self, "EmailProcessor", "po-email-processor", alarm_topic
+        )
+
         # --- Throttles alarm: po-email-processor ---
         # Any throttled invocation (concurrency cap hit) in a 5-min window pages.
         # ALARM-only to site-alerts; no OK action; NOT_BREACHING when no data.

cdk/wo_stack.py

@@ -76,6 +76,70 @@ def _add_ddb_alarms(scope, id_prefix, table, alarm_name_prefix, alarm_topic):
     ).add_alarm_action(cw_actions.SnsAction(alarm_topic))
 
 
+# CloudWatch namespace for the log-derived sender-authentication metrics.
+_SENDER_AUTH_METRIC_NAMESPACE = "Seahaven/ProcurementIngest"
+
+
+def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic):
+    """Metric-filter + alarm on ``sender_auth_rejected`` warnings (INFRA-107).
+
+    A rejected inbound email is skipped without erroring the invocation, so it
+    is invisible to the Errors/Throttles/DLQ alarms. This turns the structured
+    warning log into a CloudWatch metric and pages when rejections spike --
+    catching a silent false-reject storm (allowlist wrong, signing-domain
+    drift, SES header-format change) that would otherwise discard legitimate
+    mail while the pipeline reports healthy. This is the safety net for the WO
+    allowlist domain assumption (seahaven.com) -- if the real Gmail-forward
+    re-signing domain differs, this alarm surfaces it instead of a silent
+    work-order outage.
+
+    ALARM-only SnsAction to site-alerts; no OK action. The metric filter reads
+    the function's own log group (imported by the deterministic
+    ``/aws/lambda/<fn>`` name, created by the function's log_retention). A plain
+    substring pattern is used because Lambda prefixes each line with its own
+    level/timestamp/request-id, so the JSON payload is not a standalone JSON
+    log event a `{$.event=...}` pattern could match.
+    """
+    metric_name = f"{function_name}-sender-auth-rejected"
+    logs.MetricFilter(
+        scope,
+        f"{id_prefix}SenderAuthRejectedFilter",
+        log_group=logs.LogGroup.from_log_group_name(
+            scope,
+            f"{id_prefix}LogGroup",
+            f"/aws/lambda/{function_name}",
+        ),
+        filter_pattern=logs.FilterPattern.literal('"sender_auth_rejected"'),
+        metric_namespace=_SENDER_AUTH_METRIC_NAMESPACE,
+        metric_name=metric_name,
+        metric_value="1",
+        default_value=0,
+    )
+
+    # >=3 rejections in a 15-min window pages: a lone spoofed probe to the
+    # (obscure, internal) ingest address is tolerated, but a genuine
+    # false-reject storm -- where legitimate senders are being dropped -- trips
+    # quickly. Threshold is intentionally conservative and easy to tune.
+    cloudwatch.Metric(
+        namespace=_SENDER_AUTH_METRIC_NAMESPACE,
+        metric_name=metric_name,
+        period=Duration.minutes(15),
+        statistic="Sum",
+    ).create_alarm(
+        scope,
+        f"{id_prefix}SenderAuthRejectedAlarm",
+        alarm_name=f"{function_name}-sender-auth-rejected",
+        alarm_description=(
+            f"{function_name} rejected inbound mail on sender authentication "
+            "(possible allowlist/DKIM-domain drift silently dropping real mail)"
+        ),
+        threshold=3,
+        evaluation_periods=1,
+        comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,
+        treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
+    ).add_alarm_action(cw_actions.SnsAction(alarm_topic))
+
+
 class WorkorderIngestStack(Stack):
     def __init__(self, scope: Construct, construct_id: str, **kwargs):
         super().__init__(scope, construct_id, **kwargs)
@@ -185,6 +249,15 @@ class WorkorderIngestStack(Stack):
                 "WORK_ORDERS_TABLE": work_orders_table.table_name,
                 "COMMENTS_TABLE": comments_table.table_name,
                 "ANTHROPIC_API_KEY_SECRET_ARN": anthropic_secret.secret_arn,
+                # Fail-closed sender auth (INFRA-107): the handler only
+                # accepts mail whose SES-stamped Authentication-Results
+                # header carries dkim=pass for one of these domains. APM
+                # mail arrives via the apm@ Google Groups forward, which
+                # re-signs as seahaven.com (observed on live traffic
+                # 2026-07-15: "dkim=pass header.i=@seahaven.com"; the
+                # original hxgnsmartcloud.com signature does not survive
+                # the forward). Unset/empty ⇒ the handler rejects all mail.
+                "ALLOWED_DKIM_DOMAINS": "seahaven.com",
             },
         )
 
@@ -227,6 +300,19 @@ class WorkorderIngestStack(Stack):
             treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
         ).add_alarm_action(cw_actions.SnsAction(alarm_topic))
 
+        # --- Sender-auth rejection alarm (INFRA-107) ---
+        # A rejected email (bad/unaligned DKIM verdict) returns normally, so it
+        # produces NO Lambda error, NO DLQ message and NO retry -- only a
+        # `sender_auth_rejected` warning log. The WO allowlist trusts dkim=pass
+        # for seahaven.com on the assumption the apm@ forward re-signs there; if
+        # that assumption is wrong (e.g. a Gmail auto-forward re-signs under a
+        # different domain), 100% of legitimate work-order mail is silently
+        # dropped. This metric filter + alarm turns those warnings into a paging
+        # signal so a false-reject storm surfaces instead of a silent outage.
+        _add_sender_auth_rejected_alarm(
+            self, "EmailProcessor", "workorder-email-processor", alarm_topic
+        )
+
         # --- Throttles alarm: workorder-email-processor ---
         # Any throttled invocation (concurrency cap hit) in a 5-min window pages.
         # ALARM-only to site-alerts; no OK action; NOT_BREACHING when no data.

lambdas/po/email_processor/handler.py

@@ -17,6 +17,7 @@ from email import policy
 
 import anthropic
 import boto3
+from ses_auth import authenticate_inbound_email
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
@@ -365,6 +366,13 @@ def handler(event, context):
         response = s3.get_object(Bucket=bucket, Key=key)
         raw_email = response["Body"].read()
 
+        # Fail-closed sender authentication (INFRA-107): only mail with an
+        # SES-stamped dkim=pass verdict for an allowlisted domain may create
+        # or update purchase orders. Rejected mail is logged and skipped
+        # without erroring the invocation (no retries / DLQ spam).
+        if not authenticate_inbound_email(raw_email, s3_key):
+            continue
+
         email_data = parse_raw_email(raw_email)
         logger.info(f"Subject: {email_data['subject']}")
 

lambdas/po/email_processor/ses_auth.py

@@ -0,0 +1,297 @@
+"""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.
+
+**Clause injection defence (INFRA-107 hardening).** SES echoes several
+attacker-controlled SMTP-session tokens into its own Authentication-Results
+value as their own semicolon-delimited property clauses -- notably
+``envelope-from=<MAIL FROM>``, ``helo=<HELO>`` and ``header.from=<From
+domain>``. An RFC 5321 quoted-local-part MAIL FROM may legally contain
+spaces and semicolons, e.g.::
+
+    MAIL FROM:<"x; dkim=pass header.i=@amazon.coupahost.com"@attacker.com>
+
+which SES renders verbatim as ``envelope-from="x; dkim=pass
+header.i=@amazon.coupahost.com"@attacker.com;``. A naive ``split(";")``
+would tear the quoted string apart and manufacture a synthetic
+``dkim=pass header.i=@amazon.coupahost.com`` clause out of attacker input.
+The parser therefore tokenises per RFC 8601 / RFC 5322 structure: CFWS
+comments ``(...)`` are stripped first, and the value is split into clauses
+only on semicolons that sit *outside* a quoted-string. A ``;`` inside a
+quoted ``pvalue`` stays part of that one property clause and can never be
+read as the start of a ``dkim=`` methodspec.
+"""
+
+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 or whitespace so
+# "dkim=pass-anything" can never be read as "pass". (Comments are stripped
+# before matching, so the pre-strip "dkim=pass(comment)" form arrives here
+# as "dkim=pass ..." and still terminates on whitespace.)
+_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 _strip_comments(value: str) -> tuple:
+    """Remove RFC 5322 CFWS comments ``(...)`` from an unfolded header value.
+
+    Comments may nest and may contain quoted pairs (``\\)``). A ``(`` that
+    appears *inside* a quoted-string is literal text, not a comment start,
+    so quoted-strings (which carry attacker-controlled ``pvalue`` content
+    such as a quoted MAIL FROM local part) are passed through untouched.
+    Dropping comments first means comment-embedded ``header.i=`` / ``;``
+    fakes can never influence clause splitting or domain extraction.
+
+    Returns ``(stripped_text, well_formed)``. ``well_formed`` is False when
+    the value ends inside an unterminated comment or quoted-string, i.e. the
+    parens/quotes are unbalanced. Callers reject on ``not well_formed`` so a
+    malformed header (which could otherwise be mis-tokenised) fails closed
+    rather than being partially parsed.
+    """
+    out = []
+    depth = 0  # comment nesting depth
+    in_quote = False  # inside a quoted-string (only tracked at depth 0)
+    i = 0
+    n = len(value)
+    while i < n:
+        c = value[i]
+        if depth > 0:
+            # Inside a comment: only quoted-pairs and nested parens matter.
+            if c == "\\":
+                i += 2
+                continue
+            if c == "(":
+                depth += 1
+            elif c == ")":
+                depth -= 1
+            i += 1
+            continue
+        if in_quote:
+            out.append(c)
+            if c == "\\" and i + 1 < n:
+                out.append(value[i + 1])
+                i += 2
+                continue
+            if c == '"':
+                in_quote = False
+            i += 1
+            continue
+        # Normal context (outside any comment or quoted-string).
+        if c == "(":
+            depth += 1
+            i += 1
+            continue
+        if c == '"':
+            in_quote = True
+        out.append(c)
+        i += 1
+    well_formed = depth == 0 and not in_quote
+    return "".join(out), well_formed
+
+
+def _split_clauses(value: str) -> list:
+    """Split a comment-free header value into clauses on top-level ``;``.
+
+    A semicolon inside a quoted-string is preserved as part of the clause,
+    so an attacker-controlled quoted ``pvalue`` (e.g. a quoted MAIL FROM
+    echoed into ``envelope-from=``) cannot smuggle in a fake ``dkim=pass``
+    clause. Callers must run :func:`_strip_comments` first.
+    """
+    clauses = []
+    buf = []
+    in_quote = False
+    i = 0
+    n = len(value)
+    while i < n:
+        c = value[i]
+        if in_quote:
+            buf.append(c)
+            if c == "\\" and i + 1 < n:
+                buf.append(value[i + 1])
+                i += 2
+                continue
+            if c == '"':
+                in_quote = False
+            i += 1
+            continue
+        if c == '"':
+            in_quote = True
+            buf.append(c)
+            i += 1
+            continue
+        if c == ";":
+            clauses.append("".join(buf))
+            buf = []
+            i += 1
+            continue
+        buf.append(c)
+        i += 1
+    clauses.append("".join(buf))
+    return clauses
+
+
+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.
+
+    Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322)
+    so attacker-controlled tokens SES echoes into its header (envelope-from,
+    helo, header.from) cannot be split into a forged ``dkim=pass`` clause.
+    """
+    text, well_formed = _strip_comments(_unfold(value))
+    if not well_formed:
+        # Unbalanced quotes/comments: refuse to guess how to tokenise it.
+        return "", frozenset()
+    clauses = [c.strip() for c in _split_clauses(text)]
+    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

lambdas/wo/email_processor/handler.py

@@ -16,6 +16,7 @@ from email import policy
 
 import anthropic
 import boto3
+from ses_auth import authenticate_inbound_email
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
@@ -239,13 +240,21 @@ def handler(event, context):
     for record in event.get("Records", []):
         bucket = record["s3"]["bucket"]["name"]
         key = record["s3"]["object"]["key"]
+        s3_key = f"s3://{bucket}/{key}"
 
-        logger.info(f"Processing email: s3://{bucket}/{key}")
+        logger.info(f"Processing email: {s3_key}")
 
         # Fetch raw email from S3
         response = s3.get_object(Bucket=bucket, Key=key)
         raw_email = response["Body"].read()
 
+        # Fail-closed sender authentication (INFRA-107): only mail with an
+        # SES-stamped dkim=pass verdict for an allowlisted domain may create
+        # or update work orders. Rejected mail is logged and skipped without
+        # erroring the invocation (no retries / DLQ spam).
+        if not authenticate_inbound_email(raw_email, s3_key):
+            continue
+
         # Parse the raw email
         email_data = parse_raw_email(raw_email)
         logger.info(f"Subject: {email_data['subject']}")
@@ -260,8 +269,6 @@ def handler(event, context):
             logger.warning(f"No work order ID found in email, skipping: {key}")
             continue
 
-        s3_key = f"s3://{bucket}/{key}"
-
         # Always upsert the work order with any new info
         save_work_order(parsed, s3_key)
 

lambdas/wo/email_processor/ses_auth.py

@@ -0,0 +1,297 @@
+"""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.
+
+**Clause injection defence (INFRA-107 hardening).** SES echoes several
+attacker-controlled SMTP-session tokens into its own Authentication-Results
+value as their own semicolon-delimited property clauses -- notably
+``envelope-from=<MAIL FROM>``, ``helo=<HELO>`` and ``header.from=<From
+domain>``. An RFC 5321 quoted-local-part MAIL FROM may legally contain
+spaces and semicolons, e.g.::
+
+    MAIL FROM:<"x; dkim=pass header.i=@amazon.coupahost.com"@attacker.com>
+
+which SES renders verbatim as ``envelope-from="x; dkim=pass
+header.i=@amazon.coupahost.com"@attacker.com;``. A naive ``split(";")``
+would tear the quoted string apart and manufacture a synthetic
+``dkim=pass header.i=@amazon.coupahost.com`` clause out of attacker input.
+The parser therefore tokenises per RFC 8601 / RFC 5322 structure: CFWS
+comments ``(...)`` are stripped first, and the value is split into clauses
+only on semicolons that sit *outside* a quoted-string. A ``;`` inside a
+quoted ``pvalue`` stays part of that one property clause and can never be
+read as the start of a ``dkim=`` methodspec.
+"""
+
+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 or whitespace so
+# "dkim=pass-anything" can never be read as "pass". (Comments are stripped
+# before matching, so the pre-strip "dkim=pass(comment)" form arrives here
+# as "dkim=pass ..." and still terminates on whitespace.)
+_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 _strip_comments(value: str) -> tuple:
+    """Remove RFC 5322 CFWS comments ``(...)`` from an unfolded header value.
+
+    Comments may nest and may contain quoted pairs (``\\)``). A ``(`` that
+    appears *inside* a quoted-string is literal text, not a comment start,
+    so quoted-strings (which carry attacker-controlled ``pvalue`` content
+    such as a quoted MAIL FROM local part) are passed through untouched.
+    Dropping comments first means comment-embedded ``header.i=`` / ``;``
+    fakes can never influence clause splitting or domain extraction.
+
+    Returns ``(stripped_text, well_formed)``. ``well_formed`` is False when
+    the value ends inside an unterminated comment or quoted-string, i.e. the
+    parens/quotes are unbalanced. Callers reject on ``not well_formed`` so a
+    malformed header (which could otherwise be mis-tokenised) fails closed
+    rather than being partially parsed.
+    """
+    out = []
+    depth = 0  # comment nesting depth
+    in_quote = False  # inside a quoted-string (only tracked at depth 0)
+    i = 0
+    n = len(value)
+    while i < n:
+        c = value[i]
+        if depth > 0:
+            # Inside a comment: only quoted-pairs and nested parens matter.
+            if c == "\\":
+                i += 2
+                continue
+            if c == "(":
+                depth += 1
+            elif c == ")":
+                depth -= 1
+            i += 1
+            continue
+        if in_quote:
+            out.append(c)
+            if c == "\\" and i + 1 < n:
+                out.append(value[i + 1])
+                i += 2
+                continue
+            if c == '"':
+                in_quote = False
+            i += 1
+            continue
+        # Normal context (outside any comment or quoted-string).
+        if c == "(":
+            depth += 1
+            i += 1
+            continue
+        if c == '"':
+            in_quote = True
+        out.append(c)
+        i += 1
+    well_formed = depth == 0 and not in_quote
+    return "".join(out), well_formed
+
+
+def _split_clauses(value: str) -> list:
+    """Split a comment-free header value into clauses on top-level ``;``.
+
+    A semicolon inside a quoted-string is preserved as part of the clause,
+    so an attacker-controlled quoted ``pvalue`` (e.g. a quoted MAIL FROM
+    echoed into ``envelope-from=``) cannot smuggle in a fake ``dkim=pass``
+    clause. Callers must run :func:`_strip_comments` first.
+    """
+    clauses = []
+    buf = []
+    in_quote = False
+    i = 0
+    n = len(value)
+    while i < n:
+        c = value[i]
+        if in_quote:
+            buf.append(c)
+            if c == "\\" and i + 1 < n:
+                buf.append(value[i + 1])
+                i += 2
+                continue
+            if c == '"':
+                in_quote = False
+            i += 1
+            continue
+        if c == '"':
+            in_quote = True
+            buf.append(c)
+            i += 1
+            continue
+        if c == ";":
+            clauses.append("".join(buf))
+            buf = []
+            i += 1
+            continue
+        buf.append(c)
+        i += 1
+    clauses.append("".join(buf))
+    return clauses
+
+
+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.
+
+    Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322)
+    so attacker-controlled tokens SES echoes into its header (envelope-from,
+    helo, header.from) cannot be split into a forged ``dkim=pass`` clause.
+    """
+    text, well_formed = _strip_comments(_unfold(value))
+    if not well_formed:
+        # Unbalanced quotes/comments: refuse to guess how to tokenise it.
+        return "", frozenset()
+    clauses = [c.strip() for c in _split_clauses(text)]
+    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

tests/conftest.py

@@ -0,0 +1,41 @@
+"""Shared test fixtures.
+
+The Lambda handlers create boto3 clients at import time, so a region and
+dummy credentials must be in the environment before any handler module is
+imported. Setting them here (conftest runs before test collection imports
+anything) keeps every test hermetic — no real AWS calls can succeed with
+these values.
+"""
+
+import importlib.util
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
+os.environ.setdefault("AWS_ACCESS_KEY_ID", "testing")
+os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "testing")
+os.environ.setdefault("AWS_SESSION_TOKEN", "testing")
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def load_module(name: str, path: Path):
+    """Import a module from an explicit file path under a unique name."""
+    spec = importlib.util.spec_from_file_location(name, path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture(params=["wo", "po"])
+def ses_auth(request):
+    """The ses_auth module of each pipeline (duplicated file, kept in sync)."""
+    pipeline = request.param
+    return load_module(
+        f"{pipeline}_ses_auth",
+        REPO_ROOT / "lambdas" / pipeline / "email_processor" / "ses_auth.py",
+    )

tests/test_ses_auth.py

@@ -0,0 +1,320 @@
+"""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