Add fail-closed SES sender authentication (INFRA-107) (#98)

* Add fail-closed SES sender authentication

The From header and any raw-MIME Authentication-Results copies are
attacker-forgeable, so a forged email to apm@int.seahaven.com or
amazon_po@int.seahaven.com could create or mutate a WO/PO (INFRA-107,
CRITICAL). Both S3-triggered email processors now authenticate the
sender against the Authentication-Results header SES itself prepends
at delivery: only the topmost header is consulted, its authserv-id
must be amazonses.com, and it must carry dkim=pass for a domain in
the per-pipeline ALLOWED_DKIM_DOMAINS env var (comma-separated, set
in CDK so ops can adjust without code changes).

Allowlists come from live traffic observed 2026-07-15 on both ingest
buckets: WO mail arrives via the apm@ Google Groups forward, which
re-signs as seahaven.com (the hxgnsmartcloud.com signature does not
survive the forward); PO mail passes for amazon.coupahost.com.
amazonses.com also passes on PO mail but is deliberately excluded --
every SES customer's outbound mail passes for it.

Every failure path (env var unset, header missing or unparseable,
verdict fail, unaligned domain) rejects the email: a structured
warning with the reason and S3 key is logged and the record skipped
without erroring the invocation, so rejected mail causes no Lambda
retries or DLQ messages. Handler signatures and event sources are
unchanged.

Refs: INFRA-107

* Harden AR parser per cross-family review

Cross-family (GPT-4.1) review findings: terminate the dkim result
token at end-of-clause, whitespace, or a comment so a value like
"dkim=pass-fake" can never be read as a pass; normalize trailing
dots off allowlist entries so "seahaven.com." matches; make the
compat32 parser policy explicit. Adds tests for result-token
boundaries, comments after the result, quoted domain values, and
folding inside a dkim clause.

Refs: INFRA-107

* Harden AR parsing and alarm on sender-auth rejects

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

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

* Fix quoted-AUID DKIM domain spoof in sender auth

Resolve three confirmed /sh-security-review findings on the fail-closed
SES sender-authentication control.

HIGH: header.i/header.d domain extraction was not quoted-string aware.
An attacker with a valid DKIM key for their own domain could set an
RFC 6376-legal AUID such as i="@seahaven.com"@attacker.com; the naive
extractor stopped at the closing quote and returned seahaven.com,
accepting forged mail. Extraction now tokenises the clause with the same
quoted-string discipline already used for clause splitting: header.d
(the plain signing domain) is authoritative when present, otherwise the
header.i domain is the part after the AUID's LAST top-level "@", so a "@"
inside a quoted local-part is treated as signer-controlled label text and
yields the true signer (attacker.com), not seahaven.com.

LOW: the topmost-header parse ran outside evaluate_sender_authentication's
try/except, so an unexpected parser exception on crafted input could
propagate into the handler and Lambda async retries/DLQ. The parse now
fails CLOSED with an authentication_results_unparseable reason.

MEDIUM: the sender_auth_rejected alarm used Sum>=3 over 15 min, blind to
a low-volume total-reject outage (a trickle that never sums to 3). Both
stacks now alarm on >=1 reject per 5-min period with evaluation_periods=3
/ datapoints_to_alarm=2, so a sustained reject condition pages even at one
reject per period while a lone stray probe self-clears.

Refs: INFRA-107

* Load Lambda function dir on sys.path in tests

Rebasing INFRA-107 onto main folded #95's pytest suite into this
branch's tests. The unified conftest loads the PO/WO handlers by file
path, and handler.py now does `from ses_auth import
authenticate_inbound_email` -- a bare sibling import that resolves in
the Lambda only because the runtime puts each function's own directory
on sys.path. The shared load_handler now adds that directory so the
handler tests import correctly alongside the sender-auth tests.

Refs: INFRA-107

* Note #97 test files in README directory tree

The rebase onto main brought in #97's tests/requirements.txt and
tests/test_po_merge.py. List both in the directory tree so it matches
the tree on disk.

Refs: INFRA-107

* Document INFRA-107 forwarder-binding risk acceptance

Record the accepted risk that WO sender auth binds to the apm@ forward's
re-signing domain (seahaven.com) rather than the Hexagon originator; the
apm@ Google Group's restricted posting policy is the load-bearing control
(escalates to HIGH if the group is opened to external posting). Also
correct the sender-auth-rejected alarm docs to match the shipped config
(>=1 per 5-min, 2-of-3 datapoints, not the superseded >=3/15min) and
note the SES-AR-01/02 parser hardening follow-ups.

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).
+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. Merge write to DynamoDB:
+5. Claude extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year).
+6. Merge write to DynamoDB:
    - `new_po` — merge insert. Creates the PO, or backfills data into a pre-existing `Cancelled` skeleton left by an out-of-order cancellation (preserving the `Cancelled` status). No longer silently dropped when a record already exists.
    - `revision` — field-level merge (`update_item` SETs only the fields present in the revision). A revision that omits `line_items`/`supplier` no longer deletes them. Will not un-cancel a `Cancelled` PO.
    - `cancellation` — marks the row `Cancelled` (creating a minimal skeleton if the cancellation arrives before the `new_po`).
-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).
+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. Work order upserted to `WorkOrders`, event/comment appended to `WorkOrderComments`.
+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 |
@@ -73,6 +75,27 @@ The email processors **require** `ANTHROPIC_API_KEY_SECRET_ARN` to be set and re
 
 **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). **Validated against live SES-stamped headers (2026-07-16)** — real APM deliveries carry `dkim=pass header.i=@seahaven.com`. Note 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. The DKIM signer domain is read from `header.d=` when present (falling back to `header.i=`, taking the domain after the AUID's last top-level `@` so a quoted local-part cannot smuggle an allowlisted domain). Two latent comment-parsing edge cases (early comment-close, no-separator-on-strip) are tracked as hardening follow-ups — see the SES-AR-01/02 issue; neither is reachable through SES's real header encoding today.
+
+> **Risk acceptance — forwarder-domain binding (INFRA-107, accepted 2026-07-16):** for work orders this control authenticates the domain that *re-signs* the `apm@` forward (`seahaven.com`), not the Hexagon originator (`hxgnsmartcloud.com`, whose signature does not survive the forward). Its strength therefore rests on the `apm@` Google Group's posting policy being restricted to trusted internal senders — that restriction is the **load-bearing control** and is accepted as documented risk. **If the `apm@` group is ever opened to external posting, this finding escalates to HIGH** (anyone able to post to the group could inject a forged work order) and the correct fix is to bind acceptance to the originator via DMARC alignment rather than the forwarder's re-signature. The PO pipeline is unaffected — `amazon.coupahost.com` is an external domain an attacker cannot get SES to sign.
+
+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
@@ -86,6 +109,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` 5 min, `>= 1`, eval 3 / datapoints 2 |
+
+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. It counts `sender_auth_rejected` warnings per 5-minute period (`default_value=0` keeps the series continuous) and pages when 2 of the last 3 periods each see at least one rejection — a lone stray spoof probe to the internal ingest address self-clears, but a sustained false-reject storm pages within ~10–15 minutes even at low mail volume; the config is easy to tune in the CDK helper. (A residual gap remains for a *very* sparse total-reject outage — see the SES-AR-01/02 hardening issue.)
 
 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).
 
@@ -205,4 +231,11 @@ scripts/
   reprocess.py
   backfill_sites.py
 test_local.py          # Parse sample emails through Claude locally (no AWS)
+tests/
+  requirements.txt         # Test-only deps (moto)
+  conftest.py              # AWS env stubs + per-pipeline module loader
+  test_pad_zip.py          # PO zip-code padding tests
+  test_parse_raw_email.py  # MIME parsing tests (PO + WO handlers)
+  test_po_merge.py         # PO merge-write semantics tests (#97)
+  test_ses_auth.py         # Sender-authentication parser tests (INFRA-107)
 ```

cdk/po_stack.py

@@ -77,6 +77,76 @@ 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,
+    )
+
+    # Fire on a *sustained* reject condition rather than a volume spike. The
+    # earlier Sum>=3-over-15-min threshold had a blind spot that is exactly the
+    # failure this alarm exists to catch: a low-traffic pipeline in total
+    # drift outage (allowlist wrong / signing-domain changed) may only produce
+    # a trickle of rejects -- one every few minutes -- that never sums to 3 in
+    # any window, so the outage never pages. Instead: >=1 reject per 5-min
+    # period, alarming when 2 of the last 3 periods breach (evaluation_periods=3
+    # / datapoints_to_alarm=2, the same idiom as the duration alarm). A single
+    # stray spoof probe (one lone period) is tolerated and self-clears, but a
+    # sustained reject condition trips within ~10-15 min even at one reject per
+    # period. default_value=0 on the metric filter keeps the series continuous
+    # so NOT_BREACHING only applies before the first datapoint ever arrives.
+    cloudwatch.Metric(
+        namespace=_SENDER_AUTH_METRIC_NAMESPACE,
+        metric_name=metric_name,
+        period=Duration.minutes(5),
+        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=1,
+        evaluation_periods=3,
+        datapoints_to_alarm=2,
+        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 +244,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 +255,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 +288,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

@@ -74,6 +74,79 @@ 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,
+    )
+
+    # Fire on a *sustained* reject condition rather than a volume spike. The
+    # earlier Sum>=3-over-15-min threshold had a blind spot that is exactly the
+    # failure this alarm exists to catch: a low-traffic pipeline in total
+    # drift outage (allowlist wrong / signing-domain changed) may only produce
+    # a trickle of rejects -- one every few minutes -- that never sums to 3 in
+    # any window, so the outage never pages. Instead: >=1 reject per 5-min
+    # period, alarming when 2 of the last 3 periods breach (evaluation_periods=3
+    # / datapoints_to_alarm=2, the same idiom as the duration alarm). A single
+    # stray spoof probe (one lone period) is tolerated and self-clears, but a
+    # sustained reject condition trips within ~10-15 min even at one reject per
+    # period. default_value=0 on the metric filter keeps the series continuous
+    # so NOT_BREACHING only applies before the first datapoint ever arrives.
+    cloudwatch.Metric(
+        namespace=_SENDER_AUTH_METRIC_NAMESPACE,
+        metric_name=metric_name,
+        period=Duration.minutes(5),
+        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=1,
+        evaluation_periods=3,
+        datapoints_to_alarm=2,
+        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)
@@ -183,6 +256,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",
             },
         )
 
@@ -219,6 +301,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)
@@ -477,6 +478,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,429 @@
+"""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, and treats
+``header.d`` (the plain signing domain) as authoritative over ``header.i``
+when both are present. The ``header.i`` domain is derived per RFC 6376 as
+the part after the AUID's *last* ``@`` -- a ``@`` inside a quoted
+local-part (e.g. ``i="@seahaven.com"@attacker.com``) is signer-controlled
+label text, never the identity domain, and yields the true signer
+(``attacker.com``).
+
+**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)
+
+
+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 _split_properties(clause: str) -> list:
+    """Split a clause into whitespace-separated ``name=pvalue`` tokens.
+
+    Whitespace *inside* a quoted-string does not split, so an RFC 8601 pvalue
+    that embeds a quoted-string (e.g. a DKIM AUID with a quoted local-part that
+    legally contains spaces) survives as a single token. This is the same
+    quoted-string discipline :func:`_split_clauses` applies at the ``;`` level,
+    reused here at the token level so ``header.d=`` / ``header.i=`` extraction
+    is quoted-string-aware rather than a naive regex grab.
+    """
+    tokens = []
+    buf = []
+    in_quote = False
+    i = 0
+    n = len(clause)
+    while i < n:
+        c = clause[i]
+        if in_quote:
+            buf.append(c)
+            if c == "\\" and i + 1 < n:
+                buf.append(clause[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.isspace():
+            if buf:
+                tokens.append("".join(buf))
+                buf = []
+            i += 1
+            continue
+        buf.append(c)
+        i += 1
+    if buf:
+        tokens.append("".join(buf))
+    return tokens
+
+
+def _auid_domain(pvalue: str) -> str:
+    """Domain of a DKIM AUID (``header.i``) per RFC 6376.
+
+    The identity domain is the part after the *last* ``@`` of the AUID -- but a
+    ``@`` inside a quoted local-part is NOT the identity separator. So
+    ``"@seahaven.com"@attacker.com`` yields ``attacker.com`` (the real signer),
+    not ``seahaven.com``: the ``@seahaven.com`` sits inside the quoted
+    local-part and is signer-controlled label text, never the domain. A bare
+    unquoted ``@seahaven.com`` still yields ``seahaven.com``. Returns "" when
+    there is no top-level ``@`` (no valid domain) or the domain looks malformed.
+    """
+    last_at = -1
+    in_quote = False
+    i = 0
+    n = len(pvalue)
+    while i < n:
+        c = pvalue[i]
+        if in_quote:
+            if c == "\\":
+                i += 2
+                continue
+            if c == '"':
+                in_quote = False
+            i += 1
+            continue
+        if c == '"':
+            in_quote = True
+        elif c == "@":
+            last_at = i
+        i += 1
+    if last_at < 0:
+        return ""
+    domain = pvalue[last_at + 1 :].lower().rstrip(".")
+    # A DKIM domain-name is a plain dot-atom; anything with a residual quote is
+    # malformed (or a smuggling attempt) and must not be trusted.
+    if not domain or '"' in domain:
+        return ""
+    return domain
+
+
+def _plain_domain(pvalue: str) -> str:
+    """Domain of ``header.d`` -- the DKIM signing domain, a plain dot-atom.
+
+    SES writes ``header.d`` verbatim from the signature's ``d=`` tag, which is
+    never a quoted-string. A residual quote means malformed/smuggled input and
+    fails closed.
+    """
+    domain = pvalue.lower().rstrip(".")
+    if not domain or '"' in domain:
+        return ""
+    return domain
+
+
+def _clause_signer_domain(clause: str) -> str:
+    """Authoritative DKIM signer domain of a ``dkim=pass`` clause, or "".
+
+    ``header.d`` (the signing domain) is authoritative and is preferred when
+    present; only when it is absent does this fall back to ``header.i`` and
+    derive the domain from the AUID's post-final-``@`` part. Both lookups run
+    over :func:`_split_properties` tokens, so a ``header.d=``/``header.i=``
+    literal smuggled *inside* another property's quoted pvalue is confined to
+    that one token and can never be read as a top-level property.
+    """
+    header_d = None
+    header_i = None
+    for token in _split_properties(clause):
+        name, sep, val = token.partition("=")
+        if not sep:
+            continue
+        key = name.strip().lower()
+        if key == "header.d" and header_d is None:
+            header_d = val
+        elif key == "header.i" and header_i is None:
+            header_i = val
+    if header_d is not None:
+        return _plain_domain(header_d)
+    if header_i is not None:
+        return _auid_domain(header_i)
+    return ""
+
+
+def parse_authentication_results(value: str) -> tuple:
+    """Parse one Authentication-Results header value.
+
+    Returns ``(authserv_id, passing_dkim_domains)`` where the domains are the
+    authoritative signer domains of every ``dkim=pass`` clause -- ``header.d``
+    when present, else the ``header.i`` AUID's post-final-``@`` domain (see
+    :func:`_clause_signer_domain`). Malformed input yields ``("", frozenset())``,
+    which callers treat as a rejection.
+
+    Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322) at
+    the ``;`` (clause), whitespace (property) and ``@`` (AUID domain) levels, so
+    attacker-controlled tokens SES echoes into its header (envelope-from, helo,
+    header.from) cannot be split into a forged ``dkim=pass`` clause and a quoted
+    AUID local-part cannot masquerade as the identity domain.
+    """
+    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
+        domain = _clause_signer_domain(clause)
+        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.
+    #
+    # The parse is wrapped so an unexpected parser exception fails CLOSED
+    # (rejected, structured reason) instead of propagating out of the handler
+    # into Lambda's async retries / DLQ on attacker-crafted input.
+    try:
+        authserv_id, passing = parse_authentication_results(str(ar_headers[0]))
+    except Exception:
+        return False, "authentication_results_unparseable", {}
+    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)
@@ -246,13 +247,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']}")
@@ -267,8 +276,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,429 @@
+"""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, and treats
+``header.d`` (the plain signing domain) as authoritative over ``header.i``
+when both are present. The ``header.i`` domain is derived per RFC 6376 as
+the part after the AUID's *last* ``@`` -- a ``@`` inside a quoted
+local-part (e.g. ``i="@seahaven.com"@attacker.com``) is signer-controlled
+label text, never the identity domain, and yields the true signer
+(``attacker.com``).
+
+**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)
+
+
+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 _split_properties(clause: str) -> list:
+    """Split a clause into whitespace-separated ``name=pvalue`` tokens.
+
+    Whitespace *inside* a quoted-string does not split, so an RFC 8601 pvalue
+    that embeds a quoted-string (e.g. a DKIM AUID with a quoted local-part that
+    legally contains spaces) survives as a single token. This is the same
+    quoted-string discipline :func:`_split_clauses` applies at the ``;`` level,
+    reused here at the token level so ``header.d=`` / ``header.i=`` extraction
+    is quoted-string-aware rather than a naive regex grab.
+    """
+    tokens = []
+    buf = []
+    in_quote = False
+    i = 0
+    n = len(clause)
+    while i < n:
+        c = clause[i]
+        if in_quote:
+            buf.append(c)
+            if c == "\\" and i + 1 < n:
+                buf.append(clause[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.isspace():
+            if buf:
+                tokens.append("".join(buf))
+                buf = []
+            i += 1
+            continue
+        buf.append(c)
+        i += 1
+    if buf:
+        tokens.append("".join(buf))
+    return tokens
+
+
+def _auid_domain(pvalue: str) -> str:
+    """Domain of a DKIM AUID (``header.i``) per RFC 6376.
+
+    The identity domain is the part after the *last* ``@`` of the AUID -- but a
+    ``@`` inside a quoted local-part is NOT the identity separator. So
+    ``"@seahaven.com"@attacker.com`` yields ``attacker.com`` (the real signer),
+    not ``seahaven.com``: the ``@seahaven.com`` sits inside the quoted
+    local-part and is signer-controlled label text, never the domain. A bare
+    unquoted ``@seahaven.com`` still yields ``seahaven.com``. Returns "" when
+    there is no top-level ``@`` (no valid domain) or the domain looks malformed.
+    """
+    last_at = -1
+    in_quote = False
+    i = 0
+    n = len(pvalue)
+    while i < n:
+        c = pvalue[i]
+        if in_quote:
+            if c == "\\":
+                i += 2
+                continue
+            if c == '"':
+                in_quote = False
+            i += 1
+            continue
+        if c == '"':
+            in_quote = True
+        elif c == "@":
+            last_at = i
+        i += 1
+    if last_at < 0:
+        return ""
+    domain = pvalue[last_at + 1 :].lower().rstrip(".")
+    # A DKIM domain-name is a plain dot-atom; anything with a residual quote is
+    # malformed (or a smuggling attempt) and must not be trusted.
+    if not domain or '"' in domain:
+        return ""
+    return domain
+
+
+def _plain_domain(pvalue: str) -> str:
+    """Domain of ``header.d`` -- the DKIM signing domain, a plain dot-atom.
+
+    SES writes ``header.d`` verbatim from the signature's ``d=`` tag, which is
+    never a quoted-string. A residual quote means malformed/smuggled input and
+    fails closed.
+    """
+    domain = pvalue.lower().rstrip(".")
+    if not domain or '"' in domain:
+        return ""
+    return domain
+
+
+def _clause_signer_domain(clause: str) -> str:
+    """Authoritative DKIM signer domain of a ``dkim=pass`` clause, or "".
+
+    ``header.d`` (the signing domain) is authoritative and is preferred when
+    present; only when it is absent does this fall back to ``header.i`` and
+    derive the domain from the AUID's post-final-``@`` part. Both lookups run
+    over :func:`_split_properties` tokens, so a ``header.d=``/``header.i=``
+    literal smuggled *inside* another property's quoted pvalue is confined to
+    that one token and can never be read as a top-level property.
+    """
+    header_d = None
+    header_i = None
+    for token in _split_properties(clause):
+        name, sep, val = token.partition("=")
+        if not sep:
+            continue
+        key = name.strip().lower()
+        if key == "header.d" and header_d is None:
+            header_d = val
+        elif key == "header.i" and header_i is None:
+            header_i = val
+    if header_d is not None:
+        return _plain_domain(header_d)
+    if header_i is not None:
+        return _auid_domain(header_i)
+    return ""
+
+
+def parse_authentication_results(value: str) -> tuple:
+    """Parse one Authentication-Results header value.
+
+    Returns ``(authserv_id, passing_dkim_domains)`` where the domains are the
+    authoritative signer domains of every ``dkim=pass`` clause -- ``header.d``
+    when present, else the ``header.i`` AUID's post-final-``@`` domain (see
+    :func:`_clause_signer_domain`). Malformed input yields ``("", frozenset())``,
+    which callers treat as a rejection.
+
+    Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322) at
+    the ``;`` (clause), whitespace (property) and ``@`` (AUID domain) levels, so
+    attacker-controlled tokens SES echoes into its header (envelope-from, helo,
+    header.from) cannot be split into a forged ``dkim=pass`` clause and a quoted
+    AUID local-part cannot masquerade as the identity domain.
+    """
+    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
+        domain = _clause_signer_domain(clause)
+        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.
+    #
+    # The parse is wrapped so an unexpected parser exception fails CLOSED
+    # (rejected, structured reason) instead of propagating out of the handler
+    # into Lambda's async retries / DLQ on attacker-crafted input.
+    try:
+        authserv_id, passing = parse_authentication_results(str(ar_headers[0]))
+    except Exception:
+        return False, "authentication_results_unparseable", {}
+    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

@@ -26,9 +26,20 @@ def load_handler(relative_path, module_name):
 
     The handler files all share the basename ``handler.py`` and are not
     importable as packages, so a plain ``import handler`` would collide
-    across Lambdas.
+    across Lambdas. The same loader serves the ``ses_auth.py`` modules,
+    which are likewise duplicated per pipeline and not importable as
+    packages.
+
+    The Lambda runtime puts each function's own directory on ``sys.path``,
+    so handler modules import their siblings by bare name (e.g.
+    ``from ses_auth import authenticate_inbound_email``). The function
+    directory is added to ``sys.path`` here so the module executes the
+    same way under test as it does in the Lambda.
     """
     path = REPO_ROOT / relative_path
+    package_dir = str(path.parent)
+    if package_dir not in sys.path:
+        sys.path.insert(0, package_dir)
     spec = importlib.util.spec_from_file_location(module_name, path)
     module = importlib.util.module_from_spec(spec)
     sys.modules[module_name] = module
@@ -58,3 +69,13 @@ def wo_handler():
 def email_handler(request):
     """Parametrized fixture yielding each email processor handler module."""
     return request.getfixturevalue(request.param)
+
+
+@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_handler(
+        f"lambdas/{pipeline}/email_processor/ses_auth.py",
+        f"{pipeline}_ses_auth",
+    )

tests/test_ses_auth.py

@@ -0,0 +1,391 @@
+"""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_auid_localpart_is_not_the_domain(self, ses_auth):
+        # header.i="@seahaven.com" is a *quoted local-part* with no top-level
+        # "@domain" -- per RFC 6376 there is no identity domain, so it must
+        # yield nothing rather than mistaking the quoted label for the domain.
+        _, passing = ses_auth.parse_authentication_results(
+            'amazonses.com; dkim=pass header.i="@seahaven.com"'
+        )
+        assert passing == frozenset()
+
+    def test_quoted_auid_localpart_attack_yields_true_signer(self, ses_auth):
+        # The core INFRA-107 defect: a signer with their own valid DKIM key for
+        # attacker.com sets AUID i="@seahaven.com"@attacker.com (RFC 6376-legal:
+        # the quoted local-part contains an "@"). The identity domain is the
+        # part after the LAST top-level "@" -> attacker.com, NOT seahaven.com.
+        _, passing = ses_auth.parse_authentication_results(
+            'amazonses.com; dkim=pass header.i="@seahaven.com"@attacker.com'
+        )
+        assert passing == frozenset({"attacker.com"})
+        assert "seahaven.com" not in passing
+
+    def test_quoted_auid_attack_rejected_end_to_end(self, ses_auth):
+        # Same attack through the full evaluator against a seahaven.com
+        # allowlist: the genuine dkim=pass binds to attacker.com, so the forged
+        # mail must fail closed, not be accepted for seahaven.com.
+        forged = (
+            "Authentication-Results: amazonses.com;\r\n"
+            ' dkim=pass header.i="@seahaven.com"@attacker.com;\r\n'
+        )
+        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
+            raw(forged), {"seahaven.com"}
+        )
+        assert not accepted
+        assert reason == "dkim_domain_not_allowlisted"
+
+    def test_header_d_takes_precedence_over_mismatched_header_i(self, ses_auth):
+        # header.d is the authoritative signing domain; when both appear it
+        # wins over header.i, whichever order SES emits them in.
+        for clause in (
+            "dkim=pass header.i=@evil.com header.d=seahaven.com",
+            "dkim=pass header.d=seahaven.com header.i=@evil.com",
+        ):
+            _, passing = ses_auth.parse_authentication_results(
+                f"amazonses.com; {clause}"
+            )
+            assert passing == frozenset({"seahaven.com"})
+
+    def test_header_d_smuggled_in_quoted_header_i_is_not_top_level(self, ses_auth):
+        # A "header.d=seahaven.com" literal hidden inside header.i's quoted
+        # local-part must not be read as a top-level header.d property; the real
+        # (top-level) header.d=attacker.com is authoritative.
+        _, passing = ses_auth.parse_authentication_results(
+            'amazonses.com; dkim=pass header.i="header.d=seahaven.com x"@attacker.com'
+            " header.d=attacker.com"
+        )
+        assert passing == frozenset({"attacker.com"})
+
+    def test_legitimate_unquoted_header_i_still_passes(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_parser_exception_fails_closed(self, ses_auth, monkeypatch):
+        # The parse of the topmost AR header runs outside the BytesParser
+        # try/except; an unexpected parser exception on crafted input must fail
+        # CLOSED (rejected, structured reason) rather than propagate out of the
+        # handler into Lambda's async retries / DLQ.
+        def boom(_value):
+            raise ValueError("simulated parser blow-up")
+
+        monkeypatch.setattr(ses_auth, "parse_authentication_results", boom)
+        accepted, reason, _ = ses_auth.evaluate_sender_authentication(
+            raw(WO_SES_HEADER), {"seahaven.com"}
+        )
+        assert not accepted
+        assert reason == "authentication_results_unparseable"
+
+    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