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

.github/workflows/ci.yaml

@@ -7,6 +7,7 @@ jobs:
   ci:
     uses: Sea-Haven-Industries/.github/.github/workflows/ci-python-sam.yaml@fd60e4c9041784f666ac0fdefb9bec3c7fbf5143 # main
     with:
-      source-dirs: "lambdas cdk"
+      source-dirs: "lambdas cdk tests"
+      run-tests: true
       run-cdk-synth: true
       run-sam-validate: false

README.md

@@ -18,10 +18,10 @@ Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed by Claude H
 3. S3 `ObjectCreated` triggers the `po-email-processor` Lambda.
 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. 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`).
 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
@@ -35,7 +35,7 @@ Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed by Claude H
 
 **Tables:**
 - `purchase-orders` (PK: `po_number`, Streams: NEW_IMAGE) — shared with seahaven-slack-bot (read-only; see Shared Resources)
-- `verified-sites` (PK: `siteCode`, GSI: `by-state`) — ~1,100 unique Amazon facility sites
+- `verified-sites` (PK: `siteCode`) — ~1,100 unique Amazon facility sites (`by-state` GSI removed 2026-06-03, audit M-20)
 - `pending-site-review` (PK: `po_number`) — unresolvable POs for manual Payee Central verification
 
 ### Work Orders (`WorkorderIngestStack` stack)
@@ -58,7 +58,7 @@ Amazon APM work order emails (from Hexagon EAM / HxGN SmartCloud) are received a
 | `workorder-web-ui` | Manual invoke | HTML dashboard (public Function URL removed 2026-06-08, INFRA-74) |
 
 **Tables:**
-- `WorkOrders` (PK: `work_order_id`, GSIs: `site-code-index`, `status-index`)
+- `WorkOrders` (PK: `work_order_id`) — `site-code-index` and `status-index` GSIs removed 2026-06-03 (audit M-20)
 - `WorkOrderComments` (PK: `work_order_id`, SK: `comment_id`)
 
 ## Architecture
@@ -71,6 +71,8 @@ All Lambdas: Python 3.12, ARM64, 60-day log retention.
 - `po-ingest/anthropic-api-key` — Anthropic API key for PO parsing
 - `workorder-ingest/anthropic-api-key` — Anthropic API key for WO parsing
 
+The email processors **require** `ANTHROPIC_API_KEY_SECRET_ARN` to be set and resolve the key from Secrets Manager at runtime. There is no plaintext `ANTHROPIC_API_KEY` env fallback — a deploy missing the ARN fails loudly instead of silently running on an unmanaged key.
+
 **SES:** Both stacks add rules to the shared `INBOUND_MAIL` receipt rule set on `int.seahaven.com`.
 
 ### Sender authentication (INFRA-107)
@@ -111,13 +113,21 @@ The `<fn>-sender-auth-rejected` alarm closes the silent-drop gap in INFRA-107: a
 
 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).
 
+**DLQ alarms** (`AWS/SQS`): `po-email-processor-dlq-messages` and `workorder-email-processor-dlq-messages` fire when any message is visible on an email-processor DLQ (`ApproximateNumberOfMessagesVisible` Maximum, 5 min, `> 0`, eval 1) — a message there means an email was dropped after Lambda exhausted its async retries.
+
 **DynamoDB alarms** (`AWS/DynamoDB`): each owned table gets `<table>-throttles` (`ThrottledRequests`) and `<table>-system-errors` (`SystemErrors`). These metrics emit only at the `TableName` + `Operation` dimension set, so each alarm is a `Sum` math expression across the operations the table uses (Get/BatchGet/Query/Scan/Put/Update/Delete/BatchWrite). Tables covered: `purchase-orders`, `verified-sites`, `pending-site-review` (po-ingest); `WorkOrders`, `WorkOrderComments` (workorder-ingest).
 
+## Security
+
+**Web UI auth (defense-in-depth).** The `po-web-ui` / `workorder-web-ui` handlers refuse unauthenticated requests even though their public Function URLs were removed (INFRA-74). Each requires a shared secret in the `X-Auth-Token` header (or `Authorization: Bearer <token>`), compared in constant time against the configured token. The handler **fails closed** if the token is unset or unreadable (denies all). The token lives in the Secrets Manager secret `procurement-ingest/web-ui-auth-token`; only its ARN is passed to the Lambda (`WEB_UI_AUTH_TOKEN_SECRET_ARN`), and the value is fetched at runtime — never embedded in the CloudFormation template or Lambda env vars. The fetched value is cached in the warm container with a short TTL (5 min) so a rotated secret propagates without waiting for the execution environment to recycle. This is a defense-in-depth floor for a detached URL, not primary auth.
+
+**Output escaping.** All caller-influenced values (including prompt-injectable strings Claude may return for `total_amount`/line-item amounts) are HTML-escaped before interpolation to prevent stored XSS.
+
 ## Shared Resources
 
 ### `purchase-orders` table (owned here)
 
-The `purchase-orders` DynamoDB table is **owned by this repo's `po-ingest` stack** (defined in `cdk/po_stack.py` with `RemovalPolicy.RETAIN` and `StreamViewType.NEW_IMAGE`). The `po-email-processor` Lambda is the authoritative writer — it performs the conditional inserts, revision overwrites, and cancellation updates described above.
+The `purchase-orders` DynamoDB table is **owned by this repo's `po-ingest` stack** (defined in `cdk/po_stack.py` with `RemovalPolicy.RETAIN`, `StreamViewType.NEW_IMAGE`, and SSE-KMS encryption with the shared customer-managed CMK `alias/seahaven-dynamodb`, INFRA-95 / M-3). The `po-email-processor` Lambda is the authoritative writer — it performs the merge inserts, field-level revision merges, and cancellation updates described above.
 
 **Consumers (read-only):**
 
@@ -133,16 +143,18 @@ The consumer imports the table via `Table.fromTableName(...)` and is granted rea
 
 ### `WorkOrders` and `WorkOrderComments` tables (owned here)
 
-Both tables are **owned by this repo's `WorkorderIngestStack`** (`cdk/wo_stack.py`, `RemovalPolicy.RETAIN`, shared customer-managed CMK per INFRA-95 / M-3):
+Both tables are **owned by this repo's `WorkorderIngestStack`** (`cdk/wo_stack.py`, `RemovalPolicy.RETAIN`):
 
 - `WorkOrders` — PK `work_order_id` (S).
 - `WorkOrderComments` — PK `work_order_id` (S), SK `comment_id` (S).
 
-**Consumer (read-only) — data contract:** `seahaven-slack-bot` imports both tables via `Table.fromTableName(...)` (`grantReadData` plus an explicit `kms:Decrypt` grant on the shared CMK) and reads them from two Lambdas: `workorder-sync` (daily full-table scan into the Bedrock knowledge base) and `wo-po-lookup` (the Bedrock agent's direct WO lookup action group). The bot depends on the PK/SK schema above, the CMK encryption, and these attributes: on `WorkOrders` — `description`, `wo_status`, `customer`, `site_code`, `building`, `address`, `severity`, `priority`, `assigned_to`, `date_reported`, `scheduled_start`, `due_date`, `updated_at`; on `WorkOrderComments` — `created_at` (used to sort comments), `commenter`, `text`. Any change to table name, key schema, these attribute names, or the encryption key must be coordinated with `seahaven-slack-bot` before it ships, or the Bedrock agent breaks at runtime (not at deploy — the tables are imported by name, so there is no compile-time link).
+Both currently use default DynamoDB encryption — they are **not** yet on the shared customer-managed CMK (`alias/seahaven-dynamodb`, INFRA-95 / M-3); that migration is tracked in INFRA-6. The `workorder-email-processor` role no longer holds a pre-emptive encrypt/decrypt grant on that CMK (removed in the 2026-06-17 security sweep — it was unused while the tables are unencrypted and extended the role's decrypt reach to the CMK protecting `purchase-orders`). Re-add the grant as part of the INFRA-6 migration, at which point `grant_read_write_data` on the then-encrypted tables propagates the needed key permissions automatically.
+
+**Consumer (read-only) — data contract:** `seahaven-slack-bot` imports both tables via `Table.fromTableName(...)` (`grantReadData`) and reads them from two Lambdas: `workorder-sync` (daily full-table scan into the Bedrock knowledge base) and `wo-po-lookup` (the Bedrock agent's direct WO lookup action group). The bot depends on the PK/SK schema above and these attributes: on `WorkOrders` — `description`, `wo_status`, `customer`, `site_code`, `building`, `address`, `severity`, `priority`, `assigned_to`, `date_reported`, `scheduled_start`, `due_date`, `updated_at`; on `WorkOrderComments` — `created_at` (used to sort comments), `commenter`, `text`. Any change to table name, key schema, these attribute names, or the encryption configuration (e.g. the INFRA-6 CMK migration) must be coordinated with `seahaven-slack-bot` before it ships, or the Bedrock agent breaks at runtime (not at deploy — the tables are imported by name, so there is no compile-time link).
 
 ### `verified-sites` table (owned here)
 
-Owned by this repo's `po-ingest` stack (`cdk/po_stack.py`). PK `siteCode` (S); AWS-managed encryption (NOT the shared CMK).
+Owned by this repo's `po-ingest` stack (`cdk/po_stack.py`). PK `siteCode` (S); default DynamoDB encryption (NOT the shared CMK).
 
 **Consumer (read-only) — data contract:** `seahaven-slack-bot`'s `wo-po-lookup` Lambda imports this table via `Table.fromTableName(...)` for the Bedrock agent's `lookup_site` action. It does point lookups by `siteCode` and reads `address`, `fullAddress`, `city`, `state`, `zip`, `latitude`, `longitude`, `notes`. Coordinate any change to the table name, key schema, or these attribute names with `seahaven-slack-bot`.
 
@@ -150,33 +162,40 @@ Owned by this repo's `po-ingest` stack (`cdk/po_stack.py`). PK `siteCode` (S); A
 
 ## Documentation
 
-The canonical map of Sea Haven's AWS infrastructure lives in Confluence. This project's `po-ingest` and `workorder-ingest` stacks are represented there as Mermaid subgraphs.
+The canonical map of Sea Haven's AWS infrastructure lives in Confluence. This project's `po-ingest` and `WorkorderIngestStack` stacks are represented there as Mermaid subgraphs.
 
 - **[AWS Architecture Map](https://seahaven.atlassian.net/wiki/spaces/IT/pages/1540098)** (Confluence, IT space, page 1540098)
 
 ## CI/CD
 
-GitHub Actions with reusable workflows from `Sea-Haven-Industries/.github`:
-- **CI** (PR to `main`): linting + `cdk synth` via `ci-python-sam.yaml@main`
-- **CD** (push to `main`): `cdk deploy --all` via `cd-cdk.yaml@main` (OIDC auth)
+GitHub Actions with reusable workflows from `Sea-Haven-Industries/.github` (all pinned to a commit SHA of `main`):
+- **CI** (`ci.yaml`, PR to `main`): linting + `cdk synth` via `ci-python-sam.yaml`
+- **CD** (`deploy.yaml`, push to `main`): CDK deploy via `cd-cdk.yaml` (OIDC auth)
+- Plus dependency review and PR labeler workflows on every PR
 
 Branch protection on `main` — all changes through PR.
 
 ## Setup
 
 1. Bootstrap CDK: `cdk bootstrap aws://{AccountId}/us-east-1`
-2. Store Anthropic API keys:
+2. Create the web UI auth-gate shared secret. Unlike the Anthropic keys, this
+   secret is **imported by name** (`Secret.from_secret_name_v2`), not CDK-managed,
+   so it must exist before deploy or the web-ui Lambdas fail closed:
    ```bash
-   aws secretsmanager create-secret --name po-ingest/anthropic-api-key --secret-string "sk-ant-..."
-   aws secretsmanager create-secret --name workorder-ingest/anthropic-api-key --secret-string "sk-ant-..."
+   aws secretsmanager create-secret --name procurement-ingest/web-ui-auth-token --secret-string "$(openssl rand -hex 32)"
    ```
-3. Deploy both stacks:
+3. Deploy both stacks (this also creates the two Anthropic Secrets Manager secrets — they are CDK-managed, so don't `create-secret` them by hand):
    ```bash
    cd cdk
    pip install -r requirements.txt
    cdk deploy --all
    ```
-4. CloudFormation outputs include `WebUIUrl` for each stack's dashboard.
+4. Set the Anthropic API key values:
+   ```bash
+   aws secretsmanager put-secret-value --secret-id po-ingest/anthropic-api-key --secret-string "sk-ant-..."
+   aws secretsmanager put-secret-value --secret-id workorder-ingest/anthropic-api-key --secret-string "sk-ant-..."
+   ```
+5. Dashboards: `po-web-ui` and `workorder-web-ui` have no public endpoint (the Function URLs were removed 2026-06-08, INFRA-74). Invoke them through an authenticated path that forwards the `X-Auth-Token` header, e.g. `aws lambda invoke --function-name po-web-ui /tmp/out.json`.
 
 ## Scripts
 
@@ -195,7 +214,7 @@ python scripts/backfill_sites.py
 
 ```
 cdk/
-  app.py              # Two stacks: po-ingest + WorkorderIngestStack
+  app.py               # Two stacks: po-ingest + WorkorderIngestStack
   po_stack.py          # Purchase order pipeline resources
   wo_stack.py          # Work order pipeline resources
 lambdas/
@@ -206,12 +225,15 @@ lambdas/
   wo/                  # WO pipeline Lambdas
     email_processor/
     web_ui/
-shared/
-  models.py            # Work order dataclasses/enums
 scripts/
   reprocess.py
   backfill_sites.py
+test_local.py          # Parse sample emails through Claude locally (no AWS)
 tests/
-  conftest.py          # AWS env stubs + per-pipeline module loader
-  test_ses_auth.py     # Sender-authentication parser tests (INFRA-107)
+  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

@@ -114,14 +114,22 @@ def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic
         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.
+    # 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(15),
+        period=Duration.minutes(5),
         statistic="Sum",
     ).create_alarm(
         scope,
@@ -131,8 +139,9 @@ def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic
             f"{function_name} rejected inbound mail on sender authentication "
             "(possible allowlist/DKIM-domain drift silently dropping real mail)"
         ),
-        threshold=3,
-        evaluation_periods=1,
+        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))
@@ -379,6 +388,17 @@ class PoIngestStack(Stack):
             ],
         )
 
+        # --- Web UI auth token secret ---
+        # Shared secret for the web UI auth gate, stored in Secrets Manager and
+        # resolved at runtime so the token never appears in CloudFormation templates
+        # or Lambda environment variables. Create this secret before deploying
+        # either stack; both PO and WO stacks reference it by name.
+        web_ui_auth_secret = secretsmanager.Secret.from_secret_name_v2(
+            self,
+            "WebUiAuthToken",
+            "procurement-ingest/web-ui-auth-token",
+        )
+
         # --- Web UI Lambda ---
         web_ui = lambda_.Function(
             self,
@@ -393,10 +413,17 @@ class PoIngestStack(Stack):
             log_retention=logs.RetentionDays.TWO_MONTHS,
             environment={
                 "PO_TABLE": "purchase-orders",
+                # Defense-in-depth shared secret for the web UI handler. The
+                # handler fails closed if this ARN is unset or the secret is
+                # missing, so any future invocation path cannot re-expose the
+                # PO DB unauthenticated. The secret value is fetched at runtime
+                # from Secrets Manager (not embedded in env vars or template).
+                "WEB_UI_AUTH_TOKEN_SECRET_ARN": web_ui_auth_secret.secret_arn,
             },
         )
 
         po_table.grant_read_data(web_ui)
+        web_ui_auth_secret.grant_read(web_ui)
 
         # --- Throttles alarm: po-web-ui ---
         web_ui.metric_throttles(

cdk/wo_stack.py

@@ -8,7 +8,6 @@ from aws_cdk import (
     aws_cloudwatch as cloudwatch,
     aws_cloudwatch_actions as cw_actions,
     aws_dynamodb as dynamodb,
-    aws_kms as kms,
     aws_lambda as lambda_,
     aws_logs as logs,
     aws_s3 as s3,
@@ -18,7 +17,6 @@ from aws_cdk import (
     aws_secretsmanager as secretsmanager,
     aws_sns as sns,
     aws_sqs as sqs,
-    aws_ssm as ssm,
 )
 from constructs import Construct
 
@@ -116,14 +114,22 @@ def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic
         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.
+    # 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(15),
+        period=Duration.minutes(5),
         statistic="Sum",
     ).create_alarm(
         scope,
@@ -133,8 +139,9 @@ def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic
             f"{function_name} rejected inbound mail on sender authentication "
             "(possible allowlist/DKIM-domain drift silently dropping real mail)"
         ),
-        threshold=3,
-        evaluation_periods=1,
+        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))
@@ -267,21 +274,15 @@ class WorkorderIngestStack(Stack):
         comments_table.grant_read_write_data(email_processor)
         anthropic_secret.grant_read(email_processor)
 
-        # Pre-emptive KMS grant on the shared DynamoDB CMK (alias/seahaven-dynamodb,
-        # /seahaven/dynamodb/cmk-arn). The WorkOrders/WorkOrderComments tables are
-        # NOT yet SSE-KMS encrypted, so this grant is currently unused; it is added
-        # ahead of the CMK migration so the processor role does not hit AccessDenied
-        # the moment those tables are migrated. The actual table migration + kebab
-        # rename is tracked in INFRA-6 (and the set-aside wo_stack CMK WIP); fold the
-        # web-ui read grant in there.
-        dynamodb_cmk = kms.Key.from_key_arn(
-            self,
-            "DynamoDbCmk",
-            ssm.StringParameter.value_for_string_parameter(
-                self, "/seahaven/dynamodb/cmk-arn"
-            ),
-        )
-        dynamodb_cmk.grant_encrypt_decrypt(email_processor)
+        # NOTE: The pre-emptive grant_encrypt_decrypt on the shared DynamoDB CMK
+        # (alias/seahaven-dynamodb) was removed (security sweep 2026-06-17). The
+        # WorkOrders/WorkOrderComments tables are NOT SSE-KMS encrypted with that
+        # CMK, so the grant was unused for these tables yet handed
+        # wo-email-processor kms:Decrypt on the CMK that also protects the
+        # purchase-orders table (cross-stack decrypt reach). Re-add this grant only
+        # as part of the actual CMK migration of these tables (INFRA-6), at which
+        # point grant_read_write_data on the (then encrypted) tables would propagate
+        # the needed key permissions automatically.
 
         # --- Errors alarm (INFRA-41 / audit H-8) ---
         # ALARM-only (no OK action, per the CloudWatch-alarm preference) to the
@@ -396,6 +397,17 @@ class WorkorderIngestStack(Stack):
             ],
         )
 
+        # --- Web UI auth token secret ---
+        # Shared secret for the web UI auth gate, stored in Secrets Manager and
+        # resolved at runtime so the token never appears in CloudFormation templates
+        # or Lambda environment variables. Create this secret before deploying
+        # either stack; both PO and WO stacks reference it by name.
+        web_ui_auth_secret = secretsmanager.Secret.from_secret_name_v2(
+            self,
+            "WebUiAuthToken",
+            "procurement-ingest/web-ui-auth-token",
+        )
+
         # --- Web UI Lambda ---
         web_ui = lambda_.Function(
             self,
@@ -411,11 +423,18 @@ class WorkorderIngestStack(Stack):
             environment={
                 "WORK_ORDERS_TABLE": work_orders_table.table_name,
                 "COMMENTS_TABLE": comments_table.table_name,
+                # Defense-in-depth shared secret for the web UI handler. The
+                # handler fails closed if this ARN is unset or the secret is
+                # missing, so any future invocation path cannot re-expose the
+                # WO DB unauthenticated. The secret value is fetched at runtime
+                # from Secrets Manager (not embedded in env vars or template).
+                "WEB_UI_AUTH_TOKEN_SECRET_ARN": web_ui_auth_secret.secret_arn,
             },
         )
 
         work_orders_table.grant_read_data(web_ui)
         comments_table.grant_read_data(web_ui)
+        web_ui_auth_secret.grant_read(web_ui)
 
         # --- DynamoDB throttle + system-error alarms ---
         # ThrottledRequests / SystemErrors emit at TableName + Operation only

lambdas/po/email_processor/handler.py

@@ -28,6 +28,11 @@ dynamodb = boto3.resource("dynamodb")
 PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders")
 ANTHROPIC_API_KEY_SECRET_ARN = os.environ.get("ANTHROPIC_API_KEY_SECRET_ARN")
 
+# "Cancelled" is a sticky, authoritative status: once a PO reaches it, a later
+# new_po/revision may enrich other fields but must never move it back to a
+# non-cancelled status.
+CANCELLED_STATUS = "Cancelled"
+
 EXTRACTION_PROMPT = """\
 You are an email parser for a purchase order ingest pipeline.
 The emails are Coupa procurement platform notifications containing purchase order
@@ -212,14 +217,21 @@ The Coupa commodity/category field if present in the email (e.g., "Maintenance -
 
 
 def get_anthropic_client() -> anthropic.Anthropic:
-    """Create Anthropic client, fetching API key from Secrets Manager."""
-    if ANTHROPIC_API_KEY_SECRET_ARN:
-        secrets = boto3.client("secretsmanager")
-        secret = secrets.get_secret_value(SecretId=ANTHROPIC_API_KEY_SECRET_ARN)
-        api_key = secret["SecretString"]
-        return anthropic.Anthropic(api_key=api_key)
-    # Fall back to ANTHROPIC_API_KEY env var (for local testing)
-    return anthropic.Anthropic()
+    """Create Anthropic client, fetching API key from Secrets Manager.
+
+    Requires ANTHROPIC_API_KEY_SECRET_ARN to be set. The previous silent
+    fallback to a plaintext ANTHROPIC_API_KEY env var is removed: a misconfigured
+    deploy must fail loudly rather than quietly run on an unmanaged key.
+    """
+    if not ANTHROPIC_API_KEY_SECRET_ARN:
+        raise RuntimeError(
+            "ANTHROPIC_API_KEY_SECRET_ARN is not set; refusing to fall back to a "
+            "plaintext API key. Configure the Secrets Manager ARN."
+        )
+    secrets = boto3.client("secretsmanager")
+    secret = secrets.get_secret_value(SecretId=ANTHROPIC_API_KEY_SECRET_ARN)
+    api_key = secret["SecretString"]
+    return anthropic.Anthropic(api_key=api_key)
 
 
 def parse_raw_email(raw_bytes: bytes) -> dict:
@@ -314,32 +326,132 @@ def enrich_parsed(parsed: dict, s3_key: str, email_subject: str):
     return parsed
 
 
-def save_new_po(parsed: dict):
-    """Insert a new PO into DynamoDB. Skips if po_number already exists."""
+def _write_fields(po_number: str, fields: dict, *, guard_cancelled: bool):
+    """SET the given non-null fields on a PO record via update_item.
+
+    Only the fields supplied are written; absent fields are left untouched, so a
+    partial payload can never delete data that an earlier email established. The
+    record is created if it does not exist (DynamoDB update_item upsert).
+
+    When ``guard_cancelled`` is True the write carries a ConditionExpression that
+    only permits it while the record is not already Cancelled. The condition is
+    evaluated atomically by DynamoDB at write time, so a cancellation that lands
+    first always wins — there is no read-then-write TOCTOU window. A failed guard
+    raises ConditionalCheckFailedException for the caller to handle.
+    """
     table = dynamodb.Table(PO_TABLE)
-    item = {k: v for k, v in parsed.items() if v is not None}
+
+    set_parts = []
+    attr_names = {}
+    attr_values = {}
+    for key, value in fields.items():
+        if value is None or key == "po_number":
+            continue
+        name_ph = f"#{key}"
+        val_ph = f":{key}"
+        attr_names[name_ph] = key
+        attr_values[val_ph] = value
+        set_parts.append(f"{name_ph} = {val_ph}")
+
+    if not set_parts:
+        return
+
+    params = {
+        "Key": {"po_number": po_number},
+        "UpdateExpression": "SET " + ", ".join(set_parts),
+        "ExpressionAttributeNames": attr_names,
+        "ExpressionAttributeValues": attr_values,
+    }
+    if guard_cancelled:
+        params["ExpressionAttributeValues"][":__cancelled_marker"] = CANCELLED_STATUS
+        params["ConditionExpression"] = (
+            "attribute_not_exists(po_status) OR po_status <> :__cancelled_marker"
+        )
+
+    table.update_item(**params)
+
+
+def _merge_update(po_number: str, fields: dict):
+    """Merge (SET-only) the given fields onto a PO record, keeping Cancelled sticky.
+
+    Only the fields supplied are written; absent fields are left untouched. The
+    record is created if it does not exist (DynamoDB update_item upsert).
+
+    "Cancelled" is a sticky, authoritative status. When the incoming payload
+    carries a non-cancelled ``po_status``, the write is guarded by a
+    ConditionExpression so the status is only applied while the record is not
+    already Cancelled — enforced atomically at write time, eliminating the
+    read-then-write TOCTOU where a concurrently-landing cancellation could be
+    silently un-cancelled. If the guard fails (the PO is already Cancelled), the
+    same fields are re-written WITHOUT po_status/cancelled_at and
+    unconditionally, so the other fields still merge while the Cancelled status
+    stays intact.
+
+    A payload with no ``po_status``, or one whose status is already "Cancelled",
+    needs no guard — a plain merge is correct. This is what keeps legitimate
+    status updates (non-cancelled PO) and status-less revisions from ever being
+    dropped: the status is only ever suppressed on a true un-cancel transition.
+    """
+    incoming_status = fields.get("po_status")
+    if incoming_status is None or incoming_status == CANCELLED_STATUS:
+        _write_fields(po_number, fields, guard_cancelled=False)
+        return
 
     try:
-        table.put_item(
-            Item=item,
-            ConditionExpression="attribute_not_exists(po_number)",
-        )
-        logger.info(f"Created PO {parsed['po_number']}")
+        _write_fields(po_number, fields, guard_cancelled=True)
     except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
-        logger.info(f"PO {parsed['po_number']} already exists, skipping insert")
+        logger.info(
+            f"PO {po_number} is Cancelled; suppressing incoming "
+            f"po_status={incoming_status!r} and merging remaining fields"
+        )
+        enrich_fields = {
+            k: v for k, v in fields.items() if k not in ("po_status", "cancelled_at")
+        }
+        _write_fields(po_number, enrich_fields, guard_cancelled=False)
+
+
+def save_new_po(parsed: dict):
+    """Create a PO, merging into any pre-existing record.
+
+    Uses a merge update rather than a conditional put so that an out-of-order
+    cancellation (which leaves a Cancelled skeleton) is filled in with the full
+    PO data instead of the new_po being silently dropped. "Cancelled" is a sticky
+    status enforced atomically inside _merge_update: if the PO was already
+    cancelled, the new_po backfills its remaining fields (supplier, line_items,
+    amounts) but never un-cancels it.
+    """
+    po_number = parsed["po_number"]
+    fields = {k: v for k, v in parsed.items() if v is not None}
+    _merge_update(po_number, fields)
+    logger.info(f"Created/merged PO {po_number}")
 
 
 def save_revision(parsed: dict):
-    """Update an existing PO with revised data, or insert if it doesn't exist yet."""
-    table = dynamodb.Table(PO_TABLE)
-    item = {k: v for k, v in parsed.items() if v is not None}
+    """Merge revised data into an existing PO without deleting omitted fields.
 
-    table.put_item(Item=item)
-    logger.info(f"Revised PO {parsed['po_number']}")
+    A revision email often omits unchanged sections (line_items, supplier). The
+    previous full-overwrite put_item permanently dropped those. This SETs only the
+    fields present in the revision, leaving everything else intact.
+
+    "Cancelled" is a sticky status: a revision may enrich a cancelled PO's fields
+    but must never move it to a non-cancelled status. That invariant is enforced
+    atomically inside _merge_update and applies ONLY to the un-cancel transition —
+    a revision that carries no status change, or one targeting a non-cancelled PO,
+    updates po_status normally.
+    """
+    po_number = parsed["po_number"]
+    fields = {k: v for k, v in parsed.items() if v is not None}
+    _merge_update(po_number, fields)
+    logger.info(f"Revised PO {po_number}")
 
 
 def save_cancellation(parsed: dict):
-    """Update an existing PO's status to Cancelled."""
+    """Mark a PO Cancelled, creating a minimal skeleton if it doesn't exist yet.
+
+    If the cancellation arrives before the new_po, the skeleton it creates is
+    later backfilled by save_new_po (which preserves this Cancelled status), so no
+    PO data is lost on out-of-order delivery.
+    """
     table = dynamodb.Table(PO_TABLE)
 
     table.update_item(

lambdas/po/email_processor/ses_auth.py

@@ -26,7 +26,13 @@ Observed SES format (2026-07-15, both ingest buckets)::
      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.
+(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
@@ -68,8 +74,6 @@ SES_AUTHSERV_ID = "amazonses.com"
 # 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:
@@ -183,16 +187,146 @@ def _split_clauses(value: str) -> list:
     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 lowercased d=/i= domains of every ``dkim=pass`` clause. Malformed
-    input yields ``("", frozenset())``, which callers treat as a rejection.
+    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)
-    so attacker-controlled tokens SES echoes into its header (envelope-from,
-    helo, header.from) cannot be split into a forged ``dkim=pass`` clause.
+    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:
@@ -211,18 +345,9 @@ def parse_authentication_results(value: str) -> tuple:
         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)
+        domain = _clause_signer_domain(clause)
+        if domain:
+            passing.add(domain)
     return authserv_id, frozenset(passing)
 
 
@@ -252,7 +377,14 @@ def evaluate_sender_authentication(raw_email: bytes, allowed_domains) -> tuple:
     # 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]))
+    #
+    # 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),

lambdas/po/web_ui/handler.py

@@ -5,17 +5,99 @@ Serves a simple HTML dashboard for viewing purchase orders.
 Accessed via Lambda Function URL.
 """
 
+import hmac
 import json
+import logging
 import os
+import time
 from decimal import Decimal
 from html import escape as esc
 
 import boto3
 
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
 dynamodb = boto3.resource("dynamodb")
 
 PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders")
 
+# Defense-in-depth auth gate. The public Function URL was removed (INFRA-74), but
+# the handler must still refuse unauthenticated requests so any future invocation
+# path (re-attached Function URL, API Gateway, etc.) does not re-expose the whole
+# PO DB. Callers must present the shared secret in the X-Auth-Token header (or
+# Authorization: Bearer <token>). The secret is fetched at runtime from Secrets
+# Manager to keep it out of CloudFormation templates and Lambda env vars. If the
+# ARN is unset or the secret is missing, the handler fails closed and denies every
+# request.
+_WEB_UI_AUTH_TOKEN_SECRET_ARN = os.environ.get("WEB_UI_AUTH_TOKEN_SECRET_ARN")
+# Refresh the cached token this often so a rotated secret propagates without
+# waiting for the execution environment to recycle (emergency-rotation path).
+_AUTH_TOKEN_CACHE_TTL_SECONDS = 300
+_auth_token_cache = None
+_auth_token_cached_at = 0.0
+
+
+def _get_auth_token() -> str | None:
+    """Fetch the shared web UI auth token from Secrets Manager.
+
+    Cached in the warm container for a short TTL so we don't hit Secrets Manager
+    on every request, while still picking up a rotated secret within the TTL
+    rather than only when the execution environment recycles. Returns None when
+    not configured or unreadable (the caller then fails closed).
+    """
+    global _auth_token_cache, _auth_token_cached_at
+    now = time.monotonic()
+    if (
+        _auth_token_cache is not None
+        and now - _auth_token_cached_at < _AUTH_TOKEN_CACHE_TTL_SECONDS
+    ):
+        return _auth_token_cache
+    if not _WEB_UI_AUTH_TOKEN_SECRET_ARN:
+        return None
+    secrets = boto3.client("secretsmanager")
+    try:
+        secret = secrets.get_secret_value(SecretId=_WEB_UI_AUTH_TOKEN_SECRET_ARN)
+        _auth_token_cache = secret["SecretString"]
+        _auth_token_cached_at = now
+        return _auth_token_cache
+    except Exception:
+        # Fail closed (return None -> caller 401s) but surface the failure: a
+        # Secrets Manager permission/config error would otherwise make every
+        # request 401 with no operational signal. The secret value is never
+        # logged.
+        logger.exception(
+            "Failed to fetch web UI auth token from Secrets Manager; "
+            "denying request (failing closed)"
+        )
+        return None
+
+
+def _header(event: dict, name: str) -> str:
+    """Case-insensitive header lookup from a Lambda Function URL / APIGW event."""
+    headers = event.get("headers") or {}
+    name_lower = name.lower()
+    for key, value in headers.items():
+        if key.lower() == name_lower:
+            return value or ""
+    return ""
+
+
+def is_authenticated(event: dict) -> bool:
+    """Constant-time check of the request's shared secret against the configured
+    token. Fails closed when no token is configured."""
+    token = _get_auth_token()
+    if not token:
+        return False
+    presented = _header(event, "x-auth-token")
+    if not presented:
+        auth = _header(event, "authorization")
+        if auth.lower().startswith("bearer "):
+            presented = auth[7:].strip()
+    if not presented:
+        return False
+    return hmac.compare_digest(presented, token)
+
 
 def get_purchase_orders(limit=500):
     table = dynamodb.Table(PO_TABLE)
@@ -60,7 +142,10 @@ def fmt_currency(val):
         val = float(val)
     if isinstance(val, (int, float)):
         return f"${val:,.2f}"
-    return str(val)
+    # Non-numeric fallback: a prompt-injected email can make Claude return
+    # total_amount/amount as an arbitrary string. Escape it before it is
+    # interpolated raw into the HTML to prevent stored XSS.
+    return esc(str(val))
 
 
 def render_po_detail(po):
@@ -250,6 +335,13 @@ def render_po_list(purchase_orders):
 
 
 def handler(event, context):
+    if not is_authenticated(event):
+        return {
+            "statusCode": 401,
+            "headers": {"Content-Type": "text/html"},
+            "body": "<h1>401 Unauthorized</h1>",
+        }
+
     path = event.get("rawPath", "/")
     qs = event.get("queryStringParameters") or {}
 

lambdas/wo/email_processor/handler.py

@@ -67,14 +67,21 @@ Rules:
 
 
 def get_anthropic_client() -> anthropic.Anthropic:
-    """Create Anthropic client, fetching API key from Secrets Manager if configured."""
-    if ANTHROPIC_API_KEY_SECRET_ARN:
-        secrets = boto3.client("secretsmanager")
-        secret = secrets.get_secret_value(SecretId=ANTHROPIC_API_KEY_SECRET_ARN)
-        api_key = secret["SecretString"]
-        return anthropic.Anthropic(api_key=api_key)
-    # Fall back to ANTHROPIC_API_KEY env var (for local testing)
-    return anthropic.Anthropic()
+    """Create Anthropic client, fetching API key from Secrets Manager.
+
+    Requires ANTHROPIC_API_KEY_SECRET_ARN to be set. The previous silent
+    fallback to a plaintext ANTHROPIC_API_KEY env var is removed: a misconfigured
+    deploy must fail loudly rather than quietly run on an unmanaged key.
+    """
+    if not ANTHROPIC_API_KEY_SECRET_ARN:
+        raise RuntimeError(
+            "ANTHROPIC_API_KEY_SECRET_ARN is not set; refusing to fall back to a "
+            "plaintext API key. Configure the Secrets Manager ARN."
+        )
+    secrets = boto3.client("secretsmanager")
+    secret = secrets.get_secret_value(SecretId=ANTHROPIC_API_KEY_SECRET_ARN)
+    api_key = secret["SecretString"]
+    return anthropic.Anthropic(api_key=api_key)
 
 
 def parse_raw_email(raw_bytes: bytes) -> dict:

lambdas/wo/email_processor/ses_auth.py

@@ -26,7 +26,13 @@ Observed SES format (2026-07-15, both ingest buckets)::
      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.
+(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
@@ -68,8 +74,6 @@ SES_AUTHSERV_ID = "amazonses.com"
 # 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:
@@ -183,16 +187,146 @@ def _split_clauses(value: str) -> list:
     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 lowercased d=/i= domains of every ``dkim=pass`` clause. Malformed
-    input yields ``("", frozenset())``, which callers treat as a rejection.
+    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)
-    so attacker-controlled tokens SES echoes into its header (envelope-from,
-    helo, header.from) cannot be split into a forged ``dkim=pass`` clause.
+    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:
@@ -211,18 +345,9 @@ def parse_authentication_results(value: str) -> tuple:
         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)
+        domain = _clause_signer_domain(clause)
+        if domain:
+            passing.add(domain)
     return authserv_id, frozenset(passing)
 
 
@@ -252,7 +377,14 @@ def evaluate_sender_authentication(raw_email: bytes, allowed_domains) -> tuple:
     # 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]))
+    #
+    # 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),

lambdas/wo/web_ui/handler.py

@@ -5,17 +5,98 @@ Serves a simple HTML dashboard for viewing work orders and comments.
 Accessed via Lambda Function URL.
 """
 
+import hmac
 import json
+import logging
 import os
+import time
 from html import escape as esc
 
 import boto3
 
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
 dynamodb = boto3.resource("dynamodb")
 
 WORK_ORDERS_TABLE = os.environ.get("WORK_ORDERS_TABLE", "WorkOrders")
 COMMENTS_TABLE = os.environ.get("COMMENTS_TABLE", "WorkOrderComments")
 
+# Defense-in-depth auth gate. The public Function URL was removed (INFRA-74), but
+# the handler must still refuse unauthenticated requests so any future invocation
+# path does not re-expose the whole WO DB. Callers must present the shared secret
+# in the X-Auth-Token header (or Authorization: Bearer <token>). The secret is
+# fetched at runtime from Secrets Manager to keep it out of CloudFormation
+# templates and Lambda env vars. If the ARN is unset or the secret is missing, the
+# handler fails closed and denies every request.
+_WEB_UI_AUTH_TOKEN_SECRET_ARN = os.environ.get("WEB_UI_AUTH_TOKEN_SECRET_ARN")
+# Refresh the cached token this often so a rotated secret propagates without
+# waiting for the execution environment to recycle (emergency-rotation path).
+_AUTH_TOKEN_CACHE_TTL_SECONDS = 300
+_auth_token_cache = None
+_auth_token_cached_at = 0.0
+
+
+def _get_auth_token() -> str | None:
+    """Fetch the shared web UI auth token from Secrets Manager.
+
+    Cached in the warm container for a short TTL so we don't hit Secrets Manager
+    on every request, while still picking up a rotated secret within the TTL
+    rather than only when the execution environment recycles. Returns None when
+    not configured or unreadable (the caller then fails closed).
+    """
+    global _auth_token_cache, _auth_token_cached_at
+    now = time.monotonic()
+    if (
+        _auth_token_cache is not None
+        and now - _auth_token_cached_at < _AUTH_TOKEN_CACHE_TTL_SECONDS
+    ):
+        return _auth_token_cache
+    if not _WEB_UI_AUTH_TOKEN_SECRET_ARN:
+        return None
+    secrets = boto3.client("secretsmanager")
+    try:
+        secret = secrets.get_secret_value(SecretId=_WEB_UI_AUTH_TOKEN_SECRET_ARN)
+        _auth_token_cache = secret["SecretString"]
+        _auth_token_cached_at = now
+        return _auth_token_cache
+    except Exception:
+        # Fail closed (return None -> caller 401s) but surface the failure: a
+        # Secrets Manager permission/config error would otherwise make every
+        # request 401 with no operational signal. The secret value is never
+        # logged.
+        logger.exception(
+            "Failed to fetch web UI auth token from Secrets Manager; "
+            "denying request (failing closed)"
+        )
+        return None
+
+
+def _header(event: dict, name: str) -> str:
+    """Case-insensitive header lookup from a Lambda Function URL / APIGW event."""
+    headers = event.get("headers") or {}
+    name_lower = name.lower()
+    for key, value in headers.items():
+        if key.lower() == name_lower:
+            return value or ""
+    return ""
+
+
+def is_authenticated(event: dict) -> bool:
+    """Constant-time check of the request's shared secret against the configured
+    token. Fails closed when no token is configured."""
+    token = _get_auth_token()
+    if not token:
+        return False
+    presented = _header(event, "x-auth-token")
+    if not presented:
+        auth = _header(event, "authorization")
+        if auth.lower().startswith("bearer "):
+            presented = auth[7:].strip()
+    if not presented:
+        return False
+    return hmac.compare_digest(presented, token)
+
 
 def get_work_orders(limit=500):
     table = dynamodb.Table(WORK_ORDERS_TABLE)
@@ -228,6 +309,13 @@ def render_work_orders_list(work_orders):
 
 
 def handler(event, context):
+    if not is_authenticated(event):
+        return {
+            "statusCode": 401,
+            "headers": {"Content-Type": "text/html"},
+            "body": "<h1>401 Unauthorized</h1>",
+        }
+
     path = event.get("rawPath", "/")
     qs = event.get("queryStringParameters") or {}
 

pytest.ini

@@ -0,0 +1,5 @@
+[pytest]
+; Restrict discovery to tests/. test_local.py at the repo root matches the
+; default test_*.py glob but is a manual script that imports a handler (and
+; therefore creates boto3 clients) at collection time.
+testpaths = tests

tests/conftest.py

@@ -1,10 +1,9 @@
-"""Shared test fixtures.
+"""Shared pytest configuration for the procurement-ingest test suite.
 
-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.
+The Lambda handlers create boto3 clients at module import time, so a
+region and dummy credentials must be present in the environment before
+any handler module is imported. Setting them here at conftest import
+time guarantees they exist before test collection touches a handler.
 """
 
 import importlib.util
@@ -22,20 +21,61 @@ 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)
+def load_handler(relative_path, module_name):
+    """Load a Lambda handler module by file path under a unique 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. 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[name] = module
+    sys.modules[module_name] = module
     spec.loader.exec_module(module)
     return module
 
 
+@pytest.fixture(scope="session")
+def po_handler():
+    """The PO email processor handler module."""
+    return load_handler(
+        "lambdas/po/email_processor/handler.py",
+        "po_email_processor_handler",
+    )
+
+
+@pytest.fixture(scope="session")
+def wo_handler():
+    """The WO email processor handler module."""
+    return load_handler(
+        "lambdas/wo/email_processor/handler.py",
+        "wo_email_processor_handler",
+    )
+
+
+@pytest.fixture(params=["po_handler", "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_module(
+    return load_handler(
+        f"lambdas/{pipeline}/email_processor/ses_auth.py",
         f"{pipeline}_ses_auth",
-        REPO_ROOT / "lambdas" / pipeline / "email_processor" / "ses_auth.py",
     )

tests/requirements.txt

@@ -0,0 +1 @@
+moto>=5.0.0

tests/test_pad_zip.py

@@ -0,0 +1,29 @@
+"""Tests for pad_zip in the PO email processor."""
+
+import pytest
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        # Missing values pass through untouched
+        (None, None),
+        ("", ""),
+        # Leading zeros restored (Coupa drops them: "MA 2149" -> "02149")
+        ("2149", "02149"),
+        ("7001", "07001"),
+        ("501", "00501"),
+        # Already-valid 5-digit zips are unchanged
+        ("85704", "85704"),
+        ("02149", "02149"),
+        # Zip+4: leading zeros restored on the base, suffix preserved
+        ("2149-1234", "02149-1234"),
+        ("85704-1234", "85704-1234"),
+        # Surrounding whitespace is handled
+        (" 2149 ", "02149"),
+        # Non-numeric values are left alone
+        ("ABC12", "ABC12"),
+    ],
+)
+def test_pad_zip(po_handler, raw, expected):
+    assert po_handler.pad_zip(raw) == expected

tests/test_parse_raw_email.py

@@ -0,0 +1,91 @@
+"""Tests for parse_raw_email in the PO and WO email processors.
+
+Synthetic MIME messages are built with email.message.EmailMessage so the
+tests exercise real multipart parsing without shipping .eml fixtures.
+"""
+
+from email.message import EmailMessage
+
+SUBJECT = "Amazon.com Services LLC Purchase Order 2D-18206023"
+SENDER = "do_not_reply@coupahost.com"
+TO = "orders@seahavenind.com"
+CC = "ap@seahavenind.com"
+DATE = "Tue, 14 Jul 2026 09:30:00 -0400"
+
+PLAIN_BODY = "PO 2D-18206023 issued to Sea Haven Industries."
+HTML_BODY = "<html><body><p>PO 2D-18206023 issued.</p></body></html>"
+
+
+def _base_message():
+    msg = EmailMessage()
+    msg["Subject"] = SUBJECT
+    msg["From"] = SENDER
+    msg["To"] = TO
+    msg["Cc"] = CC
+    msg["Date"] = DATE
+    return msg
+
+
+def make_plain_email():
+    msg = _base_message()
+    msg.set_content(PLAIN_BODY)
+    return msg.as_bytes()
+
+
+def make_multipart_alternative_email():
+    msg = _base_message()
+    msg.set_content(PLAIN_BODY)
+    msg.add_alternative(HTML_BODY, subtype="html")
+    return msg.as_bytes()
+
+
+def make_multipart_html_only_email():
+    msg = _base_message()
+    msg.set_content(HTML_BODY, subtype="html")
+    # An attachment forces the message to multipart/mixed
+    msg.add_attachment(
+        b"%PDF-1.4 fake",
+        maintype="application",
+        subtype="pdf",
+        filename="po.pdf",
+    )
+    return msg.as_bytes()
+
+
+def test_headers_extracted(email_handler):
+    result = email_handler.parse_raw_email(make_plain_email())
+    assert result["subject"] == SUBJECT
+    assert result["sender"] == SENDER
+    assert result["to"] == TO
+    assert result["date"] == DATE
+
+
+def test_simple_plain_body(email_handler):
+    result = email_handler.parse_raw_email(make_plain_email())
+    assert result["body"].strip() == PLAIN_BODY
+
+
+def test_multipart_prefers_text_plain(email_handler):
+    result = email_handler.parse_raw_email(make_multipart_alternative_email())
+    assert result["body"].strip() == PLAIN_BODY
+    assert "<html>" not in result["body"]
+
+
+def test_multipart_falls_back_to_html(email_handler):
+    result = email_handler.parse_raw_email(make_multipart_html_only_email())
+    assert "PO 2D-18206023 issued." in result["body"]
+    assert "<html>" in result["body"]
+
+
+def test_wo_handler_extracts_cc(wo_handler):
+    result = wo_handler.parse_raw_email(make_plain_email())
+    assert result["cc"] == CC
+
+
+def test_missing_headers_default_to_empty(email_handler):
+    msg = EmailMessage()
+    msg.set_content(PLAIN_BODY)
+    result = email_handler.parse_raw_email(msg.as_bytes())
+    assert result["subject"] == ""
+    assert result["sender"] == ""
+    assert result["to"] == ""

tests/test_po_merge.py

@@ -0,0 +1,163 @@
+"""Tests for the sticky-Cancelled merge semantics in the PO email processor.
+
+These exercise the atomic conditional write in ``_merge_update`` that makes
+"Cancelled" a sticky, authoritative status: once a PO is Cancelled, a later
+new_po or revision may enrich its other fields but can never move it back to a
+non-cancelled status. The invariant is enforced server-side with a DynamoDB
+ConditionExpression, so it holds even under concurrent email processing — there
+is no read-then-write TOCTOU window.
+
+A real in-memory purchase-orders table is stood up with moto so the tests
+exercise the actual DynamoDB conditional-write and ConditionalCheckFailedException
+fallback path, not a stub.
+"""
+
+import boto3
+import pytest
+from moto import mock_aws
+
+TABLE_NAME = "purchase-orders"
+
+
+@pytest.fixture
+def po_table():
+    """Stand up a mock purchase-orders table matching the real key schema."""
+    with mock_aws():
+        resource = boto3.resource("dynamodb", region_name="us-east-1")
+        table = resource.create_table(
+            TableName=TABLE_NAME,
+            KeySchema=[{"AttributeName": "po_number", "KeyType": "HASH"}],
+            AttributeDefinitions=[
+                {"AttributeName": "po_number", "AttributeType": "S"},
+            ],
+            BillingMode="PAY_PER_REQUEST",
+        )
+        table.wait_until_exists()
+        yield table
+
+
+def _item(table, po_number):
+    return table.get_item(Key={"po_number": po_number}).get("Item")
+
+
+def test_revision_cannot_uncancel_but_enriches(po_handler, po_table):
+    """(a) Revision carrying a live status onto a Cancelled PO keeps it Cancelled
+    while still writing the other revision fields (no resurrection)."""
+    po_table.put_item(
+        Item={
+            "po_number": "PO-A",
+            "po_status": "Cancelled",
+            "cancelled_at": "2026-07-01T00:00:00",
+        }
+    )
+
+    po_handler.save_revision(
+        {
+            "po_number": "PO-A",
+            "po_status": "Issued",
+            "total_amount": 500,
+            "supplier": {"name": "Acme"},
+        }
+    )
+
+    item = _item(po_table, "PO-A")
+    assert item["po_status"] == "Cancelled"  # not resurrected
+    assert item["cancelled_at"] == "2026-07-01T00:00:00"  # marker preserved
+    assert item["total_amount"] == 500  # other fields still enriched
+    assert item["supplier"] == {"name": "Acme"}
+
+
+def test_revision_without_status_merges_on_cancelled(po_handler, po_table):
+    """(b) Revision with NO po_status onto a Cancelled PO merges cleanly, no
+    exception, status stays Cancelled."""
+    po_table.put_item(Item={"po_number": "PO-B", "po_status": "Cancelled"})
+
+    po_handler.save_revision({"po_number": "PO-B", "total_amount": 250})
+
+    item = _item(po_table, "PO-B")
+    assert item["po_status"] == "Cancelled"
+    assert item["total_amount"] == 250
+
+
+def test_revision_updates_status_on_non_cancelled(po_handler, po_table):
+    """(c) Issue B: a legitimate status update on a NON-cancelled PO is applied,
+    never dropped."""
+    po_table.put_item(Item={"po_number": "PO-C", "po_status": "Issued"})
+
+    po_handler.save_revision(
+        {"po_number": "PO-C", "po_status": "Revised", "total_amount": 999}
+    )
+
+    item = _item(po_table, "PO-C")
+    assert item["po_status"] == "Revised"
+    assert item["total_amount"] == 999
+
+
+def test_new_po_backfills_cancelled_skeleton(po_handler, po_table):
+    """(d) new_po onto a Cancelled skeleton backfills its fields but keeps the
+    Cancelled status and cancellation marker."""
+    po_table.put_item(
+        Item={
+            "po_number": "PO-D",
+            "po_status": "Cancelled",
+            "cancelled_at": "2026-07-02T00:00:00",
+        }
+    )
+
+    po_handler.save_new_po(
+        {
+            "po_number": "PO-D",
+            "po_status": "Issued",
+            "supplier": {"name": "Beta"},
+            "total_amount": 1200,
+        }
+    )
+
+    item = _item(po_table, "PO-D")
+    assert item["po_status"] == "Cancelled"
+    assert item["cancelled_at"] == "2026-07-02T00:00:00"
+    assert item["supplier"] == {"name": "Beta"}
+    assert item["total_amount"] == 1200
+
+
+def test_new_po_creates_fresh(po_handler, po_table):
+    """(e) new_po on a fresh PO creates it normally with its status."""
+    po_handler.save_new_po(
+        {"po_number": "PO-E", "po_status": "Issued", "total_amount": 42}
+    )
+
+    item = _item(po_table, "PO-E")
+    assert item["po_status"] == "Issued"
+    assert item["total_amount"] == 42
+
+
+def test_revision_creates_fresh_merge(po_handler, po_table):
+    """(e) revision on a fresh PO merges/creates normally with its status."""
+    po_handler.save_revision(
+        {"po_number": "PO-F", "po_status": "Revised", "total_amount": 77}
+    )
+
+    item = _item(po_table, "PO-F")
+    assert item["po_status"] == "Revised"
+    assert item["total_amount"] == 77
+
+
+def test_save_cancellation_is_authoritative(po_handler, po_table):
+    """Cancellation always wins: save_cancellation writes Cancelled over a live
+    status while leaving previously-established data intact."""
+    po_table.put_item(
+        Item={"po_number": "PO-G", "po_status": "Issued", "total_amount": 300}
+    )
+
+    po_handler.save_cancellation(
+        {
+            "po_number": "PO-G",
+            "processed_at": "2026-07-03T00:00:00",
+            "raw_s3_key": "s3://bucket/key",
+        }
+    )
+
+    item = _item(po_table, "PO-G")
+    assert item["po_status"] == "Cancelled"
+    assert item["cancelled_at"] == "2026-07-03T00:00:00"
+    assert item["total_amount"] == 300  # pre-existing data retained

tests/test_ses_auth.py

@@ -96,10 +96,66 @@ class TestParseAuthenticationResults:
         )
         assert passing == frozenset({"seahaven.com"})
 
-    def test_quoted_domain_value(self, ses_auth):
+    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):
@@ -273,6 +329,21 @@ class TestEvaluateSenderAuthentication:
         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