From 97a5886139fd5d106fb0d4bb9039a11e98a9548e Mon Sep 17 00:00:00 2001
From: Adam Moussa <166072409+amoussa1229@users.noreply.github.com>
Date: Wed, 15 Jul 2026 20:17:46 -0400
Subject: [PATCH] Land safe fixes from 2026-06-17 security sweep (#97)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* Remove gratuitous KMS grant on shared DynamoDB CMK

wo-email-processor held grant_encrypt_decrypt on the shared
seahaven-dynamodb CMK, but the WorkOrders/WorkOrderComments tables
are not encrypted with that CMK. The grant was dead weight that
extended the WO processor's decrypt reach to the CMK protecting the
purchase-orders table (cross-stack decrypt). Drop it to restore
least privilege; re-add as part of the table CMK migration (INFRA-6).

Refs: INFRA-6

* Require Secrets Manager key for Anthropic client

Remove the silent fallback to a plaintext ANTHROPIC_API_KEY env var
in both email processors; require ANTHROPIC_API_KEY_SECRET_ARN and
raise if absent so a misconfigured deploy fails loudly instead of
using an unmanaged key.

Adapted from f175323 on security/sweep-2026-06-17. The From-header
sender-domain allowlist from that commit is intentionally dropped:
the From header is spoofable (INFRA-107, confirmed critical) and
sender authentication is being reworked in a separate PR.

Refs: INFRA-107

* Merge PO revisions and handle out-of-order events

save_revision did a full put_item overwrite, so a revision omitting
line_items/supplier permanently deleted them. save_new_po used a
conditional put that silently dropped the PO when an out-of-order
cancellation had already created a skeleton row.

Switch both to field-level merge update_items: a revision now SETs
only the fields it carries, and a new_po backfills data into a
pre-existing Cancelled skeleton while preserving the Cancelled
status. No email can now delete data established by an earlier one.

* Gate web UIs behind auth and escape currency XSS

The po-web-ui and workorder-web-ui handlers had no auth: any
invocation path returned the full PO/WO DB. Add a fail-closed
shared-secret gate (X-Auth-Token / Bearer, constant-time compared to
WEB_UI_AUTH_TOKEN) so a future re-attached Function URL cannot
re-expose the data (URLs removed under INFRA-74). Wire the token from
the SSM String param /procurement-ingest/web-ui-auth-token.

Also fix stored XSS in po-web-ui fmt_currency: the non-numeric
fallback returned str(val) unescaped, so a prompt-injected email
could make Claude emit total_amount as <script>. Escape it.

Refs: INFRA-74

* Document sweep security fixes and merge semantics

Update the README for the 2026-06-17 security sweep: required
Secrets Manager key (no plaintext env fallback), web UI auth gate +
SSM token setup step, output-escaping note, and the new PO
revision/cancellation merge behavior.

Adapted from d91f45e on security/sweep-2026-06-17; the sender
allowlist documentation is dropped along with the allowlist itself
(deferred to the INFRA-107 sender-authentication rework).

Refs: INFRA-107

* fix: resolve web UI auth token from Secrets Manager at runtime

Replace the plaintext SSM String parameter with a Secrets Manager secret
referenced by ARN only. The token is fetched and cached at module level on
first invocation, keeping shared secrets out of CloudFormation templates and
Lambda environment variables.

Refs: PR-97

* Add TTL to web UI auth token cache for rotation

The web-ui handlers cached the Secrets Manager auth token at module
level with no expiry, so a rotated secret was only picked up when the
warm container recycled — an emergency rotation could take hours to
take effect. Cache the fetched value for a 5-minute TTL instead, so a
rotated token propagates within the TTL while still avoiding a Secrets
Manager call on every request. Still fails closed when the secret is
unset or unreadable.

Refs: INFRA-74

* Log Secrets Manager failures in web UI auth token fetch

The web UI auth gate correctly fails closed when the shared token
cannot be read, but _get_auth_token() swallowed every exception
silently. A Secrets Manager permission or config error then made
every request 401 with no operational signal, leaving an outage
indistinguishable from ordinary unauthenticated traffic.

Add a module-level logger to both web_ui handlers and log the
fetch failure with logger.exception() in the except block before
returning None. Behavior is unchanged (still fails closed); the
failure is now visible in CloudWatch. The secret value is never
logged. The two handlers stay byte-consistent in the mirrored
_get_auth_token() region.

The companion finding on the CDK import of the shared
procurement-ingest/web-ui-auth-token secret was evaluated and left
as-is: the token is a single secret shared by both the PO and WO
stacks, so from_secret_name_v2 (which scopes grant_read via the
standard 6-char suffix wildcard) is correct; making it a CDK-managed
Secret in both stacks would collide the two stacks on the same
explicit secret name at deploy time.

Refs: INFRA-74

* Make Cancelled PO status sticky via atomic write

The PO merge path read status with a get_item (_is_cancelled) and then
wrote with an unconditional update_item. Two defects followed from this:

- Race (Issue A): a cancellation landing between the read and the write
  was silently un-cancelled by a revision carrying a non-cancelled
  po_status — a TOCTOU on a table with concurrent email processing.
- Over-broad strip (Issue B): save_revision dropped po_status whenever
  the PO was Cancelled, so legitimate status updates on non-cancelled
  POs and status-less revisions were affected rather than only the true
  un-cancel transition.

Enforce the invariant server-side instead. "Cancelled" is a sticky,
authoritative status: once set, later new_po/revision emails may enrich
other fields but must never move it to a non-cancelled status. When the
payload carries a non-cancelled po_status, _merge_update issues the
update_item guarded by ConditionExpression "attribute_not_exists(po_status)
OR po_status <> :marker", evaluated atomically at write time, so a
cancellation that lands first always wins. On ConditionalCheckFailedException
the same fields are re-written without po_status/cancelled_at, enriching the
record while Cancelled sticks. Payloads with no status change, or an already
-Cancelled status, take a plain merge — the status is only ever suppressed on
a real un-cancel. This removes the non-atomic get_item from the write path;
_is_cancelled is deleted. Key schema and attribute names are unchanged, so the
cross-stack purchase-orders contract (read-only by seahaven-slack-bot) holds.

Add moto-backed tests covering un-cancel suppression with field enrichment,
status-less merge onto a Cancelled PO, legitimate status updates on
non-cancelled POs, new_po backfill of a Cancelled skeleton, fresh
create/merge, and authoritative save_cancellation.

Refs: #97
---
 README.md                             |  32 +++--
 cdk/po_stack.py                       |  18 +++
 cdk/wo_stack.py                       |  44 ++++---
 lambdas/po/email_processor/handler.py | 158 +++++++++++++++++++++----
 lambdas/po/web_ui/handler.py          |  94 ++++++++++++++-
 lambdas/wo/email_processor/handler.py |  23 ++--
 lambdas/wo/web_ui/handler.py          |  88 ++++++++++++++
 tests/requirements.txt                |   1 +
 tests/test_po_merge.py                | 163 ++++++++++++++++++++++++++
 9 files changed, 563 insertions(+), 58 deletions(-)
 create mode 100644 tests/requirements.txt
 create mode 100644 tests/test_po_merge.py

diff --git a/README.md b/README.md
index 84a04e3..6922d49 100644
--- a/README.md
+++ b/README.md
@@ -17,10 +17,10 @@ Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed by Claude H
 2. SES (`INBOUND_MAIL` rule set) drops the raw MIME into `s3://po-ingest-emails-{AccountId}/inbound/`.
 3. S3 `ObjectCreated` triggers the `po-email-processor` Lambda.
 4. Claude extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year).
-5. Conditional write to DynamoDB:
-   - `new_po` — idempotent insert (no-op if PO exists)
-   - `revision` — unconditional overwrite
-   - `cancellation` — marks existing row `Cancelled`
+5. 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:
    - **LedgerFlow** (`seahaven-slack-bot/po-sync`) — daily KB sync
    - **Site extractor** (`po-ingest-site-extractor`) — real-time site address extraction into `verified-sites` table
@@ -69,6 +69,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`.
 
 **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.
@@ -91,11 +93,17 @@ The `<fn>-duration` and `<fn>-throttles` alarms for `po-email-processor` and `wo
 
 **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`, `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 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):**
 
@@ -116,7 +124,7 @@ Both tables are **owned by this repo's `WorkorderIngestStack`** (`cdk/wo_stack.p
 - `WorkOrders` — PK `work_order_id` (S).
 - `WorkOrderComments` — PK `work_order_id` (S), SK `comment_id` (S).
 
-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 already holds a pre-emptive encrypt/decrypt grant on the CMK so the migration won't break it.
+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).
 
@@ -146,18 +154,24 @@ Branch protection on `main` — all changes through PR.
 ## Setup
 
 1. Bootstrap CDK: `cdk bootstrap aws://{AccountId}/us-east-1`
-2. Deploy both stacks (this also creates the two Secrets Manager secrets — they are CDK-managed, so don't `create-secret` them by hand):
+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 procurement-ingest/web-ui-auth-token --secret-string "$(openssl rand -hex 32)"
+   ```
+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
    ```
-3. Set the Anthropic API key values:
+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-..."
    ```
-4. Dashboards: `po-web-ui` and `workorder-web-ui` have no public endpoint (the Function URLs were removed 2026-06-08, INFRA-74). Invoke them manually and render the returned HTML, e.g. `aws lambda invoke --function-name po-web-ui /tmp/out.json`.
+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
 
diff --git a/cdk/po_stack.py b/cdk/po_stack.py
index 4648842..47d8536 100644
--- a/cdk/po_stack.py
+++ b/cdk/po_stack.py
@@ -296,6 +296,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,
@@ -310,10 +321,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(
diff --git a/cdk/wo_stack.py b/cdk/wo_stack.py
index fc8d166..dfb83c1 100644
--- a/cdk/wo_stack.py
+++ b/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
 
@@ -194,21 +192,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
@@ -310,6 +302,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,
@@ -325,11 +328,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
diff --git a/lambdas/po/email_processor/handler.py b/lambdas/po/email_processor/handler.py
index 9c5b620..fa193f5 100644
--- a/lambdas/po/email_processor/handler.py
+++ b/lambdas/po/email_processor/handler.py
@@ -27,6 +27,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
@@ -211,14 +216,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:
@@ -313,32 +325,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(
diff --git a/lambdas/po/web_ui/handler.py b/lambdas/po/web_ui/handler.py
index beaa174..e229a0d 100644
--- a/lambdas/po/web_ui/handler.py
+++ b/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 {}
 
diff --git a/lambdas/wo/email_processor/handler.py b/lambdas/wo/email_processor/handler.py
index 2abdc8f..f51cc5e 100644
--- a/lambdas/wo/email_processor/handler.py
+++ b/lambdas/wo/email_processor/handler.py
@@ -66,14 +66,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:
diff --git a/lambdas/wo/web_ui/handler.py b/lambdas/wo/web_ui/handler.py
index 4af9120..3e3c849 100644
--- a/lambdas/wo/web_ui/handler.py
+++ b/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 {}
 
diff --git a/tests/requirements.txt b/tests/requirements.txt
new file mode 100644
index 0000000..021a495
--- /dev/null
+++ b/tests/requirements.txt
@@ -0,0 +1 @@
+moto>=5.0.0
diff --git a/tests/test_po_merge.py b/tests/test_po_merge.py
new file mode 100644
index 0000000..26201c9
--- /dev/null
+++ b/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