adam/procurement-ingest
1
0
Fork 0
mirror of https://github.com/Sea-Haven-Industries/procurement-ingest.git synced 2026-09-30 10:43:14 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(webhook): SHOC WO webhook emitter — dark-ship streams, HMAC secret + rotation

Browse source 
Implements docs/shoc-webhook-plan.md Phases 1-5 (PR-2 of the SHOC
call-and-be-called effort). Everything ships DARK: both DynamoDB event
source mappings deploy enabled=False; activation is a deliberate
one-line follow-up PR gated on the SHOC receiver passing the shared
HMAC test vectors.

- Streams: NEW_AND_OLD_IMAGES on WorkOrders + WorkOrderComments
  (in-place update, RETAIN + logical IDs untouched; no existing
  consumers — verified live, neither table had a stream).
- workorder-shoc-emitter (Py3.12/ARM64): stream -> envelope ->
  HMAC-signed POST per docs/shoc-webhook-contract.md; strict per-shard
  ordering (parallelization 1, bisect off, retry until 24h age,
  ReportBatchItemFailures); 429/5xx/timeout block the shard in order,
  other 4xx park to workorder-shoc-emitter-rejected; ESM failures ->
  workorder-shoc-emitter-failures (metadata; replay rebuilds from
  DynamoDB). Echo guard skips write_origin=shoc-write-api.
- Secret workorder-ingest/shoc-webhook-hmac on a dedicated CMK
  (alias workorder-ingest-shoc-webhook-kms); cross-account
  GetSecretValue/DescribeSecret + kms:Decrypt granted to exactly
  arn:aws:iam::396287094661:role/shoc-backend-dev. RemovalPolicy
  DESTROY deliberately (machine-generated material; avoids the
  fixed-name RETAIN-orphan deadlock).
- workorder-shoc-hmac-rotator: 30-day rotation, dual-key overlap,
  64-hex keys, kid = UTC %Y-%m-%dT%H.
- Alarms (ALARM-only -> site-alerts): emitter errors/throttles/
  duration + iterator-age (>=10 min) + failures/rejected queue
  depth; rotator standard trio.
- scripts/replay_shoc_webhooks.py: dry-run-default operator replay
  (rebuilds from tables, replay:true envelopes).
- Tests: 742 passing, 85.56% aggregate; golden HMAC vectors shared
  with SHOC in docs/shoc-webhook-test-vectors.json (emitter + replay
  signing pinned to identical vectors); bundle-consistency AST pins
  for both new bundles.
- README: WO stack + webhook feed section, alarm table, runbooks;
  removed stale seahaven-slack-bot consumer references.
This commit is contained in:
Adam Moussa 2026-07-24 14:54:07 -04:00
parent 5abc369e02
commit eb56d39b93
No known key found for this signature in database
18 changed files with 3019 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
README.md
View file

@ -22,9 +22,7 @@ Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed **determini
- `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
7. DynamoDB Streams feeds the **site extractor** (`po-ingest-site-extractor`) — real-time site address extraction into the `verified-sites` table. (LedgerFlow / `seahaven-slack-bot/po-sync`, the former second stream consumer, was decommissioned 2026-07-23.)
**Deterministic template parser.** `template_parser.try_deterministic_parse()` classifies by exact subject regex, extracts the nested contract (`supplier{}`, `ship_to{}` — 8 keys, `line_items[]` — 10 keys per item), and returns a parsed result **only if** it passes a fail-closed validation gate: recursive exact key-set at every nesting level; `email_type` emitted **only** on the exact new-PO subject *and* a confirmed-safe `Status` (never defaulted — a cancellation misrouted as `new_po` would defeat the sticky-`Cancelled` guard); `po_number` shape + byte-equality with the subject, the body `PO ID`, the `Amazon Purchase Order #` heading, and the `orders/<id>` URL; duplicate-label anchor integrity (`Supplier`/`Shipping`/`Total` each appear twice, the first `Shipping` must be the literal `None` placeholder); money fidelity (every `Decimal` re-serializes byte-identically to its source token with a digit/comma border check — the thousands-separator-truncation kill switch — plus `sum(line amounts) == total`, proven against **both** `Total` blocks); bullet-metadata label discipline (closed label set, assigned by leading label, never ordinal — immune to the optional `Part Number` segment); USPS address shape on the raw pre-enrichment value; and rejection of any unparseable sentinel or residual `\r`/`\xa0` artifact. Multi-line-item (0.18%) and non-USD (0 observed) new-POs, comment emails, and anything else falls back to the AI extractor. The AI path is gated too: the untrusted email reaches Bedrock inside a neutralized `<email>` data block (forged tag lookalikes in the body are defanged) with `temperature=0`, and the raw model output must pass the fail-closed `validate_ai_fallback()` gate — the PO-specific nested contract (exact key-set at every level, with missing keys normalized rather than rejected), a `po_number` shape check hardened against fullwidth-digit and trailing-artifact injection (the same regex family protecting the DynamoDB partition key the handler builds from it), an `email_type` allow-list enforced *before* dispatch so a miss can never fall into the `new_po` default branch, and `Decimal`/`int`/`None` money typing (PO decodes with `parse_float=Decimal`) — before any DynamoDB write. **The template path is gate-enforced end-to-end; the AI-fallback path is validated and fail-closed — output that fails the gate is skipped, never written (see `ai_fallback_rejected` below), so a malformed or injected email raises the fallback rate rather than corrupting a record.** Every record emits one CloudWatch EMF metric (see below).
@ -57,7 +55,7 @@ Or aggregate overall agreement per field: `| filter ispresent(DerivedFieldAgreem
| `po-web-ui` | Manual invoke (authenticated — see Setup §6) | HTML dashboard (public Function URL removed 2026-06-08, INFRA-74) |
**Tables:**
- `purchase-orders` (PK: `po_number`, Streams: NEW_IMAGE) — shared with seahaven-slack-bot (read-only; see Shared Resources)
- `purchase-orders` (PK: `po_number`, Streams: NEW_IMAGE) — read by `procurement-api` (see Shared Resources; the former `seahaven-slack-bot` reader was decommissioned 2026-07-23)
- `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
@ -81,10 +79,23 @@ Amazon APM work order emails (from Hexagon EAM / HxGN SmartCloud) are received a
|---|---|---|
| `workorder-email-processor` | S3 ObjectCreated | Claude extraction + DynamoDB write |
| `workorder-web-ui` | Manual invoke (authenticated — see Setup §6) | HTML dashboard (public Function URL removed 2026-06-08, INFRA-74) |
| `workorder-shoc-emitter` | DynamoDB Streams, both WO tables (**ESMs ship disabled** — see [SHOC webhook feed](#shoc-webhook-feed-workorder-shoc-emitter)) | HMAC-signed webhook push of every WO mutation to the SHOC backend |
| `workorder-shoc-hmac-rotator` | Secrets Manager rotation schedule (30 days) | Rotates the webhook HMAC signing keys (dual-key overlap) |
**Tables:**
- `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`) — see the `comment_id` format note below
- `WorkOrders` (PK: `work_order_id`, Streams: NEW_AND_OLD_IMAGES) — `site-code-index` and `status-index` GSIs removed 2026-06-03 (audit M-20)
- `WorkOrderComments` (PK: `work_order_id`, SK: `comment_id`, Streams: NEW_AND_OLD_IMAGES) — see the `comment_id` format note below
### SHOC webhook feed (`workorder-shoc-emitter`)
**Flow:** `WorkOrders` / `WorkOrderComments` DynamoDB Streams (`NEW_AND_OLD_IMAGES` — the OLD image is what lets the emitter detect the `wo_status → cancelled` transition) → `workorder-shoc-emitter` → HMAC-signed HTTPS POST → SHOC backend. Every WO mutation becomes one webhook event (`work_order.created` / `.updated` / `.cancelled` / `.comment_added`) within seconds of the DynamoDB commit; DynamoDB stays the source of truth.
- **Contract:** `docs/shoc-webhook-contract.md` (Rev 2026-07-23) is the producer/consumer contract SHOC builds its receiver against, and `docs/shoc-webhook-test-vectors.json` is the shared receiver-verification vector set — both sides pin their HMAC implementation against the same vectors (producer-side via the golden-vector tests over `delivery.sign_body`).
- **Ships DARK.** Both event-source mappings deploy with `enabled=False`: the full stack (Lambda, queues, alarms, secret, rotation) deploys and is testable with zero deliveries while SHOC has no receiver, so this repo's merge cadence never depends on SHOC's. Activation is a deliberate one-line `enabled=True` PR, gated on the SHOC receiver passing the shared test vectors. The ESMs start at `LATEST` — no historical flood; SHOC loads history through the `procurement-api` read API at activation instead.
- **Ordering/retry semantics.** `parallelization_factor=1`, `bisect_batch_on_error=False`, `retry_attempts=-1`, `maximum_record_age=24h`: a retryable failure (429/5xx/timeout/connection error) blocks the shard and retries from the failed record — per-work-order commit order is the guarantee, and blocking is the intended behavior when SHOC is down. `report_batch_item_failures` keeps earlier in-batch successes from being re-delivered. Records that exhaust the 24h age are parked as ESM **failure metadata** (not full records) on `workorder-shoc-emitter-failures` (`on_failure` destination); a non-retryable 4xx (a contract bug, never worth blocking the shard for 24h) parks the **full `{envelope, response_status}` payload** on `workorder-shoc-emitter-rejected` and the loop continues. Both queues: 14-day retention, SSL-enforced, alarmed (see [CloudWatch alarms](#cloudwatch-alarms)); recovery is `scripts/replay_shoc_webhooks.py` (see [Scripts](#scripts)).
- **Secret + KMS.** The HMAC signing keys live in Secrets Manager secret `workorder-ingest/shoc-webhook-hmac` (value `{"keys": [{"kid", "secret"}, ...]}`, newest first, max 2), encrypted with the dedicated CMK `workorder-ingest-shoc-webhook-kms` — deliberately **not** `alias/seahaven-dynamodb`, so the SHOC cross-account grant's decrypt reach covers exactly this one secret and nothing else. The secret's removal policy is **`DESTROY`, deliberately not `RETAIN`**: the value is machine-generated HMAC material, fully regenerable by a single rotation, so `RETAIN` buys nothing and would expose the fixed-name RETAIN-orphan deadlock (a failed create orphans an empty shell holding the global name; every later create fails `AlreadyExists`). **Accidental-deletion recovery runbook:** redeploy to recreate the secret, force a rotation (`aws secretsmanager rotate-secret --secret-id workorder-ingest/shoc-webhook-hmac`), notify the SHOC team — receivers re-fetch within their ≤5-minute cache TTL, so no coordination window is needed — then watch the `-failures` queue and replay the gap with the replay script.
- **Rotation.** `workorder-shoc-hmac-rotator` runs on a 30-day schedule: it prepends a fresh 64-hex-char key as `keys[0]` and truncates the list to 2 entries (one overlap cycle). `kid` format is `YYYY-MM-DDTHH`. The emitter always signs with `keys[0]` behind a 5-minute TTL cache; the receiver accepts any listed `kid` and re-fetches on an unknown one — there is no delivery window in which signatures can't verify.
- **Cross-account grants (exact ARN only):** `arn:aws:iam::396287094661:role/shoc-backend-dev` is granted `secretsmanager:GetSecretValue` on the secret's resource policy **and** `kms:Decrypt` on the CMK's key policy — both halves are required; either one alone fails silently at the receiver. Future staging/prod receiver roles are each a deliberate, individually-reviewed policy addition — no wildcard/prefix trust.
### Procurement API (`procurement-api` stack)
@ -160,9 +171,9 @@ Every alarm is **ALARM-only** (no OK action), sends to the shared `site-alerts`
| Alarm | Functions | Metric / config |
|---|---|---|
| `<fn>-errors` | `po-email-processor`, `po-ingest-site-extractor`, `workorder-email-processor` | `Errors` Sum, 5 min, `> 0`, eval 1 |
| `<fn>-throttles` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui`, `workorder-email-processor` | `Throttles` Sum, 5 min, `> 0`, eval 1 |
| `<fn>-duration` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui` (p99); `workorder-email-processor` (p95) | `Duration` percentile, 5 min, `>= 45000` ms (75% of the 60s timeout), eval 3 / datapoints 2 |
| `<fn>-errors` | `po-email-processor`, `po-ingest-site-extractor`, `workorder-email-processor`, `workorder-shoc-emitter`, `workorder-shoc-hmac-rotator` | `Errors` Sum, 5 min, `> 0`, eval 1 |
| `<fn>-throttles` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui`, `workorder-email-processor`, `workorder-shoc-emitter`, `workorder-shoc-hmac-rotator` | `Throttles` Sum, 5 min, `> 0`, eval 1 |
| `<fn>-duration` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui`, `workorder-shoc-emitter`, `workorder-shoc-hmac-rotator` (p99); `workorder-email-processor` (p95) | `Duration` percentile, 5 min, `>= 45000` ms (75% of the 60s timeout), eval 3 / datapoints 2 |
| `<fn>-sender-auth-rejected` | `po-email-processor`, `workorder-email-processor` | Log-metric-filter count (namespace `Seahaven/ProcurementIngest`, `default_value=0`) on `sender_auth_rejected` warnings, `Sum` 5 min, `>= 1`, eval 3 / datapoints 2 |
The `<fn>-sender-auth-rejected` alarm closes the silent-drop gap in INFRA-107: a rejected email returns normally (no error, no retry, no DLQ message), so without a log-metric filter a signing-domain drift or a wrong allowlist would discard 100% of legitimate mail while every other alarm stayed green. It counts `sender_auth_rejected` warnings per 5-minute period (`default_value=0` keeps the series continuous) and pages when 2 of the last 3 periods each see at least one rejection — a lone stray spoof probe to the internal ingest address self-clears, but a sustained false-reject storm pages within ~10–15 minutes even at low mail volume; the config is easy to tune in the CDK helper. (A residual gap remains for a *very* sparse total-reject outage — see the SES-AR-01/02 hardening issue.)
@ -171,6 +182,8 @@ The `<fn>-duration` and `<fn>-throttles` alarms for `po-email-processor` and `wo
**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. Recovery from a DLQ message (no console redrive) is documented in the [DLQ recovery runbook](docs/runbook-dlq-recovery.md).
**SHOC emitter alarms** (bespoke — these don't fit the standard-Lambda-alarm helper's shape): `workorder-shoc-emitter-iterator-age` (`IteratorAge` Maximum, 5 min, `>= 600000` ms, eval 3 / datapoints 2) fires when the stream lags ≥ 10 minutes — SHOC is likely down and the shard is blocking on retries, which is exactly the ordered-backpressure design working, but an operator should know. `workorder-shoc-emitter-failures-messages` and `workorder-shoc-emitter-rejected-messages` (`ApproximateNumberOfMessagesVisible` Maximum, 5 min, `> 0`, eval 1) page the replay runbook: the `-failures` queue receives ESM failure **metadata** for retry-exhausted records, the `-rejected` queue receives the **full parked payloads** of non-retryable 4xx deliveries (see the [SHOC webhook feed](#shoc-webhook-feed-workorder-shoc-emitter) section and `scripts/replay_shoc_webhooks.py`). All three: ALARM-only → `site-alerts`, `NOT_BREACHING`, per the house rules above.
**Parse-outcome metric + fallback-rate alarm (workorder-ingest):** the WO processor writes one CloudWatch **EMF** line per email to namespace `Seahaven/WorkorderIngest`, metric `ParseOutcome` (Unit Count, value 1), dimensioned by `ParseMethod` (`template` | `ai_fallback` | `ai_fallback_rejected`) and `TemplateId` (`update_plaintext` | `assign_html` | `unknown`). `ai_fallback_rejected` counts AI-fallback output that failed the fail-closed `validate_ai_fallback()` gate (schema/enum/date contract on raw Bedrock output — prompt-injection defence) and was dropped without a DynamoDB write. Non-dimension EMF properties `ReasonCode` and `work_order_id` are queryable in Logs Insights but not promoted to metrics (kept low-cardinality). EMF is used instead of `PutMetricData` so there is no extra sync call / latency / IAM grant on the async hot path (the role already has `logs:PutLogEvents`). The alarm `workorder-email-processor-template-fallback-rate` fires when the AI-fallback share of parses — rejected fallback parses included, so a drift outage whose AI output also fails the gate cannot lower the observed rate while dropping mail — exceeds **15%** sustained (a `MathExpression` with `FILL(...,0)` and a ≥10-sample volume floor over 15-minute periods, eval 3 / datapoints 2) — catching Hexagon template-drift coverage collapse while the volume floor + `FILL` prevent low-volume false pages / `INSUFFICIENT_DATA`. ALARM-only `SnsAction` to `site-alerts`, no OK action, `NOT_BREACHING`. The 15-minute period is a deliberate deviation from the 5-minute house style to accumulate a stable denominator at the low ~760/day volume. A second alarm, `workorder-email-processor-ai-fallback-rejected`, pages on the rejected series itself (≥1 rejection per 5-min period, 2 of the last 6 periods — the sender-auth-rejected sparse-arrival idiom) because a gate rejection drops mail without error/retry/DLQ and would otherwise be silent.
**WO Bedrock transport-error metric (Phase 8).** A Bedrock-side transport error (throttling, malformed response, non-JSON model text) during the AI-fallback attempt previously emitted **zero** `ParseOutcome` datapoints — the only emit sites were post-gate. `handler.py` now wraps the `extract_with_bedrock` call in a try/except that emits exactly one `ParseMethod=ai_fallback` / `ReasonCode=bedrock_error` datapoint and then re-raises (the exception still propagates into the errors alarm / DLQ path unchanged). This is an except-and-reraise, not a reorder: a gate-rejected email (Bedrock *returns* successfully, `validate_ai_fallback()` then rejects it) still emits only the single `ai_fallback_rejected` datapoint and nothing else — the except branch never fires because Bedrock did not raise — so the `workorder-email-processor-ai-fallback-rejected` "a rejected email emits nothing else" alarm contract holds with no double-count.
@ -303,7 +316,7 @@ Both tables are **owned by this repo's `WorkorderIngestStack`** (`cdk/wo_stack.p
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.
**Consumers (read-only) — data contract:** `seahaven-slack-bot` (the former external reader) was decommissioned 2026-07-23; its grants are gone. Current consumers: the `procurement-api` Lambda (this repo, `Table.from_table_name` + `grant_read_data`, serving `GET /work-orders*`), and — once the SHOC webhook ships — the `workorder-shoc-emitter` stream consumer. External readers (SHOC) consume through the REST contract (`lambdas/api/openapi.json`) and the webhook contract (`docs/shoc-webhook-contract.md`); both documents' field lists mirror `lambdas/wo/email_processor/persistence.py`. Any change to table name, key schema, attribute names, or encryption configuration (e.g. the INFRA-6 CMK migration) must update those two contracts in the same PR — the tables are imported by name, so there is no compile-time link and breakage surfaces at runtime.
**Consumers (read-only) — data contract:** `seahaven-slack-bot` (the former external reader) was decommissioned 2026-07-23; its grants are gone. Current consumers: the `procurement-api` Lambda (this repo, `Table.from_table_name` + `grant_read_data`, serving `GET /work-orders*`), and the `workorder-shoc-emitter` stream consumer (this repo) — both tables now stream `NEW_AND_OLD_IMAGES`, and the emitter is a live consumer of those streams, dark (ESMs disabled) until the SHOC activation PR. External readers (SHOC) consume through the REST contract (`lambdas/api/openapi.json`) and the webhook contract (`docs/shoc-webhook-contract.md`); both documents' field lists mirror `lambdas/wo/email_processor/persistence.py`. Any change to table name, key schema, attribute names, or encryption configuration (e.g. the INFRA-6 CMK migration) must update those two contracts in the same PR — the tables are imported by name, so there is no compile-time link and breakage surfaces at runtime.
**Stream field contract (`site_code`).** Three definitions of "is this a valid site code" have existed in this repo at once. The canonical shape is `derived_fields._STRICT_CODE_RE` (`[A-Z][A-Z0-9]{2,4}`, `fullmatch`) plus its skip-list semantics (a code-shaped token is only a real site code if it is *not* skip-listed, e.g. `LLC`/`INC`/`CORP`/`LTD`/`ATTN`), used by `derive_site_code()` (the `enrich_parsed()` classifier, see the **Derived fields** note in the Purchase Orders flow above). Honestly noted, not papered over: `lambdas/po/site_extractor/handler.py`'s own `SITE_CODE_PATTERN` (`[A-Z]{2,4}\d{1,2}`, prefix-anchored `.match`, digit-requiring) still diverges from the canonical shape as of this phase — it rejects valid all-letter codes like `KLAL` (which surface as permanent `pending-site-review` rows) and accepts overlong junk like `DLI6X`/`SNY55` that the canonical `fullmatch` would not. Reconciling `po-ingest-site-extractor`'s direct-field validation onto the canonical `derived_fields` shape is Phase 6 scope, tracked separately — this paragraph will be updated when it lands.
@ -405,6 +418,13 @@ python scripts/reprocess.py --pipeline wo --all --execute
python scripts/backfill_sites.py
```
**Replay SHOC webhooks** (rebuild work-order webhook events from DynamoDB and re-POST them, marked `"replay": true`). `replay_shoc_webhooks.py` is the operator runbook for the `workorder-shoc-emitter-failures` / `-rejected` alarms: when deliveries were parked (SHOC down past the 24h retry window, or a contract-bug 4xx), replay re-sends the affected work orders from **current table state** — receivers dedupe on the deterministic `delivery_id`, so overlapping or repeated runs are harmless. Selection is exactly one of `--work-order-id` (repeatable, targeted) or `--since` (a full table Scan — a count banner prints per table); `--events` narrows to state rows, comments, or both. `--url` is required with no default — replay must be a deliberate act against a known receiver. Every mode is dry-run unless `--execute`.
```bash
python scripts/replay_shoc_webhooks.py --url https://... --work-order-id 11144580730 # targeted, dry-run first
python scripts/replay_shoc_webhooks.py --url https://... --work-order-id 11144580730 --execute # then re-POST
python scripts/replay_shoc_webhooks.py --url https://... --since 2026-07-24T02:00:00Z --execute # everything touched since (full Scan)
```
## Directory Structure
```
@ -414,7 +434,8 @@ cdk/
# email bucket, processor DLQ, fallback-rate alarm) -- called with
# each stack's own scope + literal construct ids, logical-ID-safe
po_stack.py # Purchase order pipeline resources
wo_stack.py # Work order pipeline resources
wo_stack.py # Work order pipeline resources + the SHOC webhook feed (secret/CMK,
# rotator, emitter, disabled ESMs, queues, alarms)
procurement_api_stack.py # REST API (IAM SigV4 + resource policy) over both pipelines' tables + token-gated /docs
lambdas/ # Phase 2: shared Code.from_asset("../lambdas") bundling root for
# BOTH po-email-processor and workorder-email-processor (and, since
@ -486,9 +507,19 @@ lambdas/ # Phase 2: shared Code.from_asset("../lambdas") bundling
ses-stamped/
auth-pass-01.eml # Phase 8: the one new fixture allowed this phase -- a synthesized Authentication-Results header block (from WO_SES_HEADER in test_ses_auth.py), not scraped mail
web_ui/ # handler.py imports `from web_ui_auth import is_authenticated` (shared)
shoc_emitter/ # SHOC webhook emitter (ships dark -- ESMs disabled; see the SHOC
# webhook feed section). Flat siblings, bare-name imports
handler.py # thin per-record event loop + partial-batch failure report + rejected-queue parking
envelope.py # pure stream-record -> envelope mapping + event classification (incl. echo guard)
delivery.py # TTL-cached secret fetch + HMAC signing (sign_body) + POST + response classification
shoc_hmac_rotator/
handler.py # 30-day Secrets Manager rotation (single-user: createSecret/finishSecret;
# dual-key overlap, kid YYYY-MM-DDTHH)
scripts/
reprocess.py
backfill_sites.py
replay_shoc_webhooks.py # rebuild + re-POST webhook events ("replay": true) -- the
# -failures/-rejected alarm runbook; dry-run unless --execute
post-deploy-smoke.sh # CD gate: synchronous healthcheck invoke of the three smoke-gated functions, checks FunctionError
conftest.py # Phase 8: THE repo-root session-invariant conftest. rootdir is pinned by
# pytest.ini at repo root, so this loads for every pytest invocation shape --

396
cdk/wo_stack.py
View file

@ -8,13 +8,17 @@ from aws_cdk import (
aws_cloudwatch as cloudwatch,
aws_cloudwatch_actions as cw_actions,
aws_dynamodb as dynamodb,
aws_iam as iam,
aws_kms as kms,
aws_lambda as lambda_,
aws_lambda_event_sources as lambda_event_sources,
aws_s3 as s3,
aws_s3_notifications as s3n,
aws_ses as ses,
aws_ses_actions as ses_actions,
aws_secretsmanager as secretsmanager,
aws_sns as sns,
aws_sqs as sqs,
)
from constructs import Construct
@ -51,6 +55,10 @@ class WorkorderIngestStack(Stack):
type=dynamodb.AttributeType.STRING,
),
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
# NEW_AND_OLD_IMAGES: the SHOC emitter needs OLD.wo_status to
# classify the cancelled transition (docs/shoc-webhook-plan.md
# Phase 2). In-place CFN update -- no table replacement.
stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
removal_policy=RemovalPolicy.RETAIN,
)
# site-code-index and status-index GSIs removed 2026-06-03 (audit M-20):
@ -70,6 +78,8 @@ class WorkorderIngestStack(Stack):
type=dynamodb.AttributeType.STRING,
),
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
# Streamed for the SHOC emitter (docs/shoc-webhook-plan.md Phase 2).
stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
removal_policy=RemovalPolicy.RETAIN,
)
@ -410,3 +420,389 @@ class WorkorderIngestStack(Stack):
# unauthenticated FunctionUrlAuthType.NONE URL was deleted out-of-band
# via CLI. Removing the construct (and its auto-generated Principal:*
# invoke permission) reconciles IaC with the live state.
# =====================================================================
# --- SHOC webhook emitter (docs/shoc-webhook-plan.md) ---
# Realtime work-order feed to the SHOC backend: DynamoDB Streams on the
# two WO tables -> workorder-shoc-emitter -> HMAC-signed HTTPS POST.
# Contract: docs/shoc-webhook-contract.md (Rev 2026-07-23). Built in
# _add_shoc_webhook_emitter (module helper, common.py plain-helper
# style) to keep __init__ under the PLR0915 statement ceiling; the
# stack stays the construct scope, so extraction does not move any
# logical ID.
# =====================================================================
_add_shoc_webhook_emitter(self, work_orders_table, comments_table, alarm_topic)
def _add_shoc_webhook_emitter(stack, work_orders_table, comments_table, alarm_topic):
"""SHOC webhook emitter (docs/shoc-webhook-plan.md Phases 1-4).
Dedicated HMAC CMK + secret with cross-account SHOC read grants, the
30-day rotation Lambda, the two failure queues, the stream-driven emitter
Lambda with its two (dark, enabled=False) event source mappings, and the
full alarm set. `stack` is the construct scope for every child, exactly as
if this code were inline in __init__.
"""
# SHOC consumer principal for the cross-account read grants. EXACT role
# ARN only -- future shoc-backend-staging/-prod roles are each a
# deliberate, individually-reviewed policy addition (no wildcard or
# prefix trust).
shoc_consumer_principal = iam.ArnPrincipal(
"arn:aws:iam::396287094661:role/shoc-backend-dev"
)
# --- Dedicated CMK for the HMAC secret ---
# Dedicated key, NOT alias/seahaven-dynamodb: reusing the DynamoDB CMK
# would hand the SHOC cross-account grant decrypt reach over the PO
# table's encryption key -- the dedicated key scopes the grant to
# exactly this secret (plan Phase 1).
shoc_webhook_key = kms.Key(
stack,
"ShocWebhookHmacKey",
alias="workorder-ingest-shoc-webhook-kms",
description=(
"Dedicated CMK for the workorder-ingest/shoc-webhook-hmac "
"secret (cross-account readable by the SHOC backend)"
),
enable_key_rotation=True,
# GOTCHA: DESTROY is deliberate -- do not "harden" this to RETAIN.
# The key protects only machine-generated HMAC material that is
# fully regenerable by one rotation, and DESTROY avoids the
# fixed-name RETAIN-orphan deadlock on the alias (mirrors the
# secret's rationale below).
removal_policy=RemovalPolicy.DESTROY,
pending_window=Duration.days(7),
)
# Key-policy half of the cross-account read grant (the secret resource
# policy below is the other half; either one alone fails silently at
# the receiver). resources=["*"] is key-scoped, not account-wide --
# KMS key policies only ever apply to this key.
shoc_webhook_key.add_to_resource_policy(
iam.PolicyStatement(
actions=["kms:Decrypt"],
principals=[shoc_consumer_principal],
resources=["*"],
)
)
# --- HMAC signing secret ---
# Value shape (contract section 6.1):
# {"keys": [{"kid": "<YYYY-MM-DDTHH>", "secret": "<64 hex>"}, ...]},
# newest first, max 2; the producer signs with keys[0]. The
# generate_secret_string below is BOOTSTRAP shape only ({"keys": []}
# plus throwaway entropy the rotator ignores); the first rotation
# (rotate_immediately default) populates the real keys.
shoc_hmac_secret = secretsmanager.Secret(
stack,
"ShocWebhookHmacSecret",
secret_name="workorder-ingest/shoc-webhook-hmac",
encryption_key=shoc_webhook_key,
description=(
"HMAC signing keys for the SHOC work-order webhook "
"(docs/shoc-webhook-contract.md section 6)"
),
# GOTCHA: DESTROY is deliberate -- do not "harden" this to RETAIN.
# The value is machine-generated HMAC material with no operator-set
# content, fully regenerable by one rotation, so RETAIN buys
# nothing and would expose the fixed-name RETAIN orphan deadlock
# (a failed first create orphans an empty shell holding the global
# name; see reference_secret_retain_orphan_deadlock).
removal_policy=RemovalPolicy.DESTROY,
generate_secret_string=secretsmanager.SecretStringGenerator(
secret_string_template='{"keys": []}',
generate_string_key="bootstrap_entropy",
password_length=32,
exclude_punctuation=True,
),
)
cdk.Tags.of(shoc_hmac_secret).add("Purpose", "shoc-webhook-hmac")
cdk.Tags.of(shoc_hmac_secret).add("ManagedBy", "procurement-ingest-cdk")
# Secret-resource-policy half of the cross-account read grant (the key
# policy above is the other half). DescribeSecret lets the receiver
# resolve secret metadata without any broader list permission.
shoc_hmac_secret.add_to_resource_policy(
iam.PolicyStatement(
actions=[
"secretsmanager:GetSecretValue",
"secretsmanager:DescribeSecret",
],
principals=[shoc_consumer_principal],
resources=["*"],
)
)
# --- HMAC rotation Lambda ---
# 30-day schedule: generates a new key, prepends as keys[0], truncates
# to 2 entries. Single-user rotation (receivers re-fetch on a <=5-min
# TTL), so the standard 4-step rotation collapses to
# createSecret/finishSecret.
shoc_hmac_rotator_log_group = common.make_function_log_group(
stack, "ShocHmacRotator", "workorder-shoc-hmac-rotator"
)
shoc_hmac_rotator = lambda_.Function(
stack,
"ShocHmacRotator",
function_name="workorder-shoc-hmac-rotator",
runtime=lambda_.Runtime.PYTHON_3_12,
architecture=lambda_.Architecture.ARM_64,
handler="handler.handler",
code=lambda_.Code.from_asset(
"../lambdas",
exclude=["**/__pycache__/**", "**/tests/**", "**/package/**"],
bundling=cdk.BundlingOptions(
image=lambda_.Runtime.PYTHON_3_12.bundling_image,
command=[
"bash",
"-c",
# Stdlib + boto3-from-runtime only; nothing installed.
"cp wo/shoc_hmac_rotator/*.py /asset-output/",
],
),
),
timeout=Duration.seconds(60),
memory_size=128,
log_group=shoc_hmac_rotator_log_group,
)
# Rotation permissions, scoped to the one secret. CDK's secret_arn
# token resolves to the full ARN including the -?????? suffix wildcard,
# so no separate "*"-suffixed resource variant is needed.
shoc_hmac_rotator.add_to_role_policy(
iam.PolicyStatement(
actions=[
"secretsmanager:DescribeSecret",
"secretsmanager:GetSecretValue",
"secretsmanager:PutSecretValue",
"secretsmanager:UpdateSecretVersionStage",
],
resources=[shoc_hmac_secret.secret_arn],
)
)
shoc_webhook_key.grant_encrypt_decrypt(shoc_hmac_rotator)
shoc_hmac_secret.add_rotation_schedule(
"Rotation",
rotation_lambda=shoc_hmac_rotator,
automatically_after=Duration.days(30),
)
# --- Standard per-Lambda alarms: workorder-shoc-hmac-rotator ---
# errors + throttles + p99 duration. No DLQ alarm: rotation is invoked
# synchronously by Secrets Manager (dlq=None); a failed rotation
# surfaces as an invocation error.
common.add_standard_lambda_alarms(
stack,
"ShocHmacRotator",
shoc_hmac_rotator,
"workorder-shoc-hmac-rotator",
alarm_topic,
duration_statistic="p99",
errors=True,
dlq=None,
descriptions={
"errors": "workorder-shoc-hmac-rotator invocation errors",
"throttles": "workorder-shoc-hmac-rotator invocation throttles",
"duration": (
"workorder-shoc-hmac-rotator p99 duration approaching the 60s timeout"
),
},
)
# --- Emitter failure queues ---
# Failures queue: ESM on_failure destination. It receives ESM failure
# METADATA (shard/sequence pointers), not full payloads -- replay
# rebuilds events from DynamoDB (contract section 8).
shoc_emitter_failures_queue = sqs.Queue(
stack,
"ShocEmitterFailuresQueue",
queue_name="workorder-shoc-emitter-failures",
retention_period=Duration.days(14),
enforce_ssl=True,
)
# Rejected queue: full {envelope, response_status} payloads parked by
# the handler on non-retryable 4xx responses (contract section 7).
shoc_emitter_rejected_queue = sqs.Queue(
stack,
"ShocEmitterRejectedQueue",
queue_name="workorder-shoc-emitter-rejected",
retention_period=Duration.days(14),
enforce_ssl=True,
)
# --- Emitter Lambda ---
shoc_emitter_log_group = common.make_function_log_group(
stack, "ShocEmitter", "workorder-shoc-emitter"
)
shoc_emitter = lambda_.Function(
stack,
"ShocEmitter",
function_name="workorder-shoc-emitter",
runtime=lambda_.Runtime.PYTHON_3_12,
architecture=lambda_.Architecture.ARM_64,
handler="handler.handler",
code=lambda_.Code.from_asset(
"../lambdas",
exclude=["**/__pycache__/**", "**/tests/**", "**/package/**"],
bundling=cdk.BundlingOptions(
image=lambda_.Runtime.PYTHON_3_12.bundling_image,
command=[
"bash",
"-c",
# Stdlib HTTP (urllib.request) + boto3-from-runtime
# only; nothing installed.
"cp wo/shoc_emitter/*.py /asset-output/",
],
),
),
timeout=Duration.seconds(60),
memory_size=256,
log_group=shoc_emitter_log_group,
environment={
# Non-sensitive endpoint URL (HMAC is the auth, the URL is
# not). path TBD by SHOC -- confirmed in the activation PR.
"SHOC_WEBHOOK_URL": (
"https://api.dev.seahaven.com/api/webhooks/work-orders"
),
"HMAC_SECRET_ARN": shoc_hmac_secret.secret_arn,
"REJECTED_QUEUE_URL": shoc_emitter_rejected_queue.queue_url,
},
)
work_orders_table.grant_stream_read(shoc_emitter)
comments_table.grant_stream_read(shoc_emitter)
shoc_hmac_secret.grant_read(shoc_emitter)
shoc_webhook_key.grant_decrypt(shoc_emitter)
shoc_emitter_rejected_queue.grant_send_messages(shoc_emitter)
# ------------------------------------------------------------------
# SHIPS DARK: both event source mappings deploy with enabled=False,
# deliberately. The full stack (Lambda, queues, alarms, secret,
# rotation) deploys and is testable with zero deliveries while SHOC
# has no receiver, so our merge cadence never depends on Luby's.
# Activation is a one-line enabled=True PR gated on the SHOC receiver
# passing the shared HMAC test vectors (plan Phase 3). LATEST start
# position => no historical flood at activation; SHOC loads history
# via the procurement read API instead.
# Ordering knobs: parallelization_factor=1, bisect_batch_on_error=
# False and retry_attempts=-1 (retry until the 24h record age) are
# REQUIRED for strict per-work-order in-order delivery -- a retryable
# failure blocks the shard rather than skipping ahead, and
# report_batch_item_failures keeps earlier in-batch successes from
# being re-delivered.
# ------------------------------------------------------------------
shoc_emitter.add_event_source(
lambda_event_sources.DynamoEventSource(
work_orders_table,
starting_position=lambda_.StartingPosition.LATEST,
batch_size=10,
bisect_batch_on_error=False,
retry_attempts=-1,
max_record_age=Duration.hours(24),
parallelization_factor=1,
report_batch_item_failures=True,
enabled=False,
on_failure=lambda_event_sources.SqsDlq(shoc_emitter_failures_queue),
)
)
shoc_emitter.add_event_source(
lambda_event_sources.DynamoEventSource(
comments_table,
starting_position=lambda_.StartingPosition.LATEST,
batch_size=10,
bisect_batch_on_error=False,
retry_attempts=-1,
max_record_age=Duration.hours(24),
parallelization_factor=1,
report_batch_item_failures=True,
enabled=False,
on_failure=lambda_event_sources.SqsDlq(shoc_emitter_failures_queue),
)
)
# --- Standard per-Lambda alarms: workorder-shoc-emitter ---
# errors + throttles + p99 duration. No DLQ alarm here: the emitter is
# a stream consumer with no async DLQ (dlq=None); its failure surfaces
# are the two SQS queues alarmed bespoke below.
common.add_standard_lambda_alarms(
stack,
"ShocEmitter",
shoc_emitter,
"workorder-shoc-emitter",
alarm_topic,
duration_statistic="p99",
errors=True,
dlq=None,
descriptions={
"errors": "workorder-shoc-emitter invocation errors",
"throttles": "workorder-shoc-emitter invocation throttles",
"duration": (
"workorder-shoc-emitter p99 duration approaching the 60s timeout"
),
},
)
# --- Bespoke emitter alarms (plan Phase 4) ---
# These don't fit add_standard_lambda_alarms' shape and stay bespoke.
# Iterator age >= 10 min sustained means SHOC is likely down and the
# shard is blocking (exactly the ordered-backpressure design working);
# the two SQS-visible alarms page the operator replay runbook.
shoc_emitter.metric(
"IteratorAge",
statistic="Maximum",
period=Duration.minutes(5),
).create_alarm(
stack,
"ShocEmitterIteratorAgeAlarm",
alarm_name="workorder-shoc-emitter-iterator-age",
alarm_description=(
"workorder-shoc-emitter stream lag >= 10 min "
"(SHOC receiver likely down; shard blocking on retries)"
),
threshold=600000,
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))
shoc_emitter_failures_queue.metric_approximate_number_of_messages_visible(
period=Duration.minutes(5),
statistic="Maximum",
).create_alarm(
stack,
"ShocEmitterFailuresMessagesAlarm",
alarm_name="workorder-shoc-emitter-failures-messages",
alarm_description=(
"workorder-shoc-emitter retry-exhausted stream records parked "
"(ESM failure metadata; replay rebuilds from DynamoDB)"
),
threshold=0,
evaluation_periods=1,
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
shoc_emitter_rejected_queue.metric_approximate_number_of_messages_visible(
period=Duration.minutes(5),
statistic="Maximum",
).create_alarm(
stack,
"ShocEmitterRejectedMessagesAlarm",
alarm_name="workorder-shoc-emitter-rejected-messages",
alarm_description=(
"workorder-shoc-emitter parked non-retryable 4xx deliveries "
"(contract bug; inspect payloads and replay)"
),
threshold=0,
evaluation_periods=1,
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
cdk.CfnOutput(
stack,
"ShocWebhookHmacSecretArn",
value=shoc_hmac_secret.secret_arn,
description=(
"SHOC webhook HMAC secret ARN -- hand off to Luby (SHOC team) "
"for the cross-account receiver fetch"
),
)

33
docs/shoc-webhook-test-vectors.json Normal file
View file

@ -0,0 +1,33 @@
{
"_readme": "Shared HMAC signing test vectors for the SHOC work-order webhook (docs/shoc-webhook-contract.md section 6). string_to_sign = \"{timestamp}.{raw_body}\" computed over the RAW UTF-8 body bytes; signature = lowercase hex of HMAC-SHA256(secret, string_to_sign), sent as header X-SH-Signature: \"v1=<hex>\" alongside X-SH-Timestamp: <unix seconds> and X-SH-Key-Id: <kid>. The HMAC key is the UTF-8 bytes of the 64-hex 'secret' string exactly as stored in the workorder-ingest/shoc-webhook-hmac secret (NO hex-decoding on either side). 'body' is the exact raw JSON string to sign, byte-for-byte: vector 2 contains non-ASCII UTF-8 (multi-byte characters must be signed as their UTF-8 bytes), vector 3 is an empty JSON object. Producer pins: lambdas/wo/shoc_emitter/delivery.py sign_body and scripts/replay_shoc_webhooks.py sign_body, both enforced by tests/test_shoc_emitter_delivery.py. The SHOC receiver should verify its implementation against every vector before activation.",
"vectors": [
{
"kid": "2026-07-20T00",
"secret_hex": "3afc6cf9cc5782b304fda7efa7f0a77c4b67ab7336536daa228960dae34aa2fe",
"timestamp": 1784642602,
"body": "{\"schema_version\": 1, \"delivery_id\": \"f2a9c1de-7b34-4d5c-9e01-8a6b5c4d3e2f\", \"event_type\": \"work_order.created\", \"occurred_at\": \"2026-07-16T14:03:22.114208+00:00\", \"source\": \"procurement-ingest/workorder-shoc-emitter\", \"replay\": false, \"data\": {\"work_order_id\": \"11144580730\", \"wo_status\": \"new\", \"description\": \"Dock door 14 won't close\", \"customer\": \"AMAZON\", \"site_code\": \"JFK8\", \"building\": \"JFK8\", \"address\": \"546 Gulf Ave, Staten Island, NY 10314\", \"severity\": \"3-Normal\", \"priority\": \"Medium\", \"assigned_to\": \"Sea Haven Industries\", \"date_reported\": \"2026-07-14T09:12:00\", \"scheduled_start\": null, \"due_date\": null, \"record_type\": \"new_work_order\", \"created_at\": \"2026-07-16T14:03:22.114208+00:00\", \"updated_at\": \"2026-07-16T14:03:22.114208+00:00\"}}",
"expected_signature": "4e6e171480e80eed333c966743c9da46595df86b4b65b76926e0781d2d3b0b46"
},
{
"kid": "2026-07-20T00",
"secret_hex": "3afc6cf9cc5782b304fda7efa7f0a77c4b67ab7336536daa228960dae34aa2fe",
"timestamp": 1784642700,
"body": "{\"schema_version\": 1, \"delivery_id\": \"0d1e2f3a-4b5c-6d7e-8f90-a1b2c3d4e5f6\", \"event_type\": \"work_order.comment_added\", \"occurred_at\": \"2026-07-16T14:05:00+00:00\", \"source\": \"procurement-ingest/workorder-shoc-emitter\", \"replay\": true, \"data\": {\"work_order_id\": \"11144580730\", \"comment_id\": \"11144580730#2026-04-27T23:51:48#a1b2c3d4e5f6\", \"record_type\": \"comment\", \"commenter\": \"APM Technician\", \"text\": \"Vendor dispatched — café access via süd door ✓\", \"created_at\": \"2026-04-27T23:51:48\", \"ingested_at\": \"2026-07-16T14:05:00+00:00\"}}",
"expected_signature": "6b61d5ac09bbf032b1a9c9b651bd7d5ea2ec925f94a857eeade52995801479e5"
},
{
"kid": "2026-06-20T00",
"secret_hex": "737719b2c437aa252a0db5f65411f828f244d403f7e625f7336da10ee985b2e4",
"timestamp": 1784000000,
"body": "{}",
"expected_signature": "ebe65980194b494c8de8d40cd6b8a42ff64c26749cf40fc6f36f59426dafa3c7"
},
{
"kid": "2026-06-20T00",
"secret_hex": "737719b2c437aa252a0db5f65411f828f244d403f7e625f7336da10ee985b2e4",
"timestamp": 1784650000,
"body": "{\"schema_version\": 1, \"event_type\": \"work_order.updated\", \"data\": {\"work_order_id\": \"999\"}}",
"expected_signature": "626c5d241887c6186fe021907b1a67a0f14ff2286d19df97785a33994c56ca54"
}
]
}

0
lambdas/wo/shoc_emitter/__init__.py Normal file
View file

152
lambdas/wo/shoc_emitter/delivery.py Normal file
View file

@ -0,0 +1,152 @@
"""HMAC-signed webhook delivery to the SHOC receiver (contract sections 6-7).
Owns the signed POST and its response classification. The signing key comes
from Secrets Manager (``workorder-ingest/shoc-webhook-hmac``, dual-key shape
``{"keys": [{"kid", "secret"}, ...]}``), cached in the warm container for a
short TTL so 30-day rotation propagates without waiting for the execution
environment to recycle -- the producer always signs with ``keys[0]``.
Response contract (section 7): 2xx -> delivered; 429/5xx/timeout/connection
error -> RetryableDeliveryError (the handler surfaces it as a batch item
failure so the ESM blocks the shard and retries in order); any other status ->
("rejected", code) for the handler to park (a contract bug must not block the
shard for 24 hours). Secret material and signatures are never logged.
``sign_body`` is a module-level pure function on purpose: the golden-vector
test suite and the replay script both pin against it, and it is the shared
definition Luby's receiver verifies with.
"""
import hashlib
import hmac
import json
import logging
import os
import time
import urllib.error
import urllib.request
from http import HTTPStatus
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# Endpoint path is TBD by SHOC; the activation PR confirms the final URL.
SHOC_WEBHOOK_URL = os.environ.get(
"SHOC_WEBHOOK_URL", "https://api.dev.seahaven.com/api/webhooks/work-orders"
)
HMAC_SECRET_ARN = os.environ.get("HMAC_SECRET_ARN")
POST_TIMEOUT_SECONDS = 10
USER_AGENT = "workorder-shoc-emitter/1"
# 2xx = delivered; 429 and 5xx retry; everything else parks (contract sec. 7).
_HTTP_SUCCESS_RANGE = range(HTTPStatus.OK, HTTPStatus.MULTIPLE_CHOICES)
_HTTP_SERVER_ERROR_MIN = HTTPStatus.INTERNAL_SERVER_ERROR
# Refresh the cached key material this often so a rotated secret propagates
# without waiting for the execution environment to recycle (contract requires
# receiver-side TTL <= 300s; the producer matches it).
_HMAC_KEYS_CACHE_TTL_SECONDS = 300
_hmac_keys_cache = None
_hmac_keys_cached_at = 0.0
class RetryableDeliveryError(Exception):
"""Delivery failed in a way the ESM should retry in order (shard-blocking).
``status_code`` carries the HTTP status when one exists (429/5xx); it is
None for timeouts, connection errors, and secret-fetch failures.
"""
def __init__(self, message: str, status_code: int | None = None):
super().__init__(message)
self.status_code = status_code
def _get_hmac_keys() -> list[dict]:
"""Fetch the signing-key list from Secrets Manager, TTL-cached.
An empty/missing ``keys`` list is the bootstrap state before the first
rotation has run -- retryable, not a crash: the shard blocks until the
rotator populates the secret. Fetch failures (throttle, transient IAM/KMS
denial) are likewise retryable so a Secrets Manager blip blocks in order
instead of failing the whole batch. Key material is never logged.
"""
global _hmac_keys_cache, _hmac_keys_cached_at
now = time.monotonic()
if (
_hmac_keys_cache is not None
and now - _hmac_keys_cached_at < _HMAC_KEYS_CACHE_TTL_SECONDS
):
return _hmac_keys_cache
secrets = boto3.client("secretsmanager")
try:
secret = secrets.get_secret_value(SecretId=HMAC_SECRET_ARN)
keys = json.loads(secret["SecretString"]).get("keys") or []
except Exception as exc:
raise RetryableDeliveryError(f"hmac secret fetch failed: {exc}") from exc
if not keys:
raise RetryableDeliveryError("hmac secret not yet rotated")
_hmac_keys_cache = keys
_hmac_keys_cached_at = now
return keys
def sign_body(secret_hex: str, timestamp: int, raw_body: bytes) -> str:
"""HMAC-SHA256 hex digest over ``f"{timestamp}.{raw_body}"`` (raw bytes).
The key is the UTF-8 bytes of the secret string exactly as stored in the
secret's ``secret`` field (the receiver reads the same JSON field -- no
hex-decoding on either side). Pure function; the golden-vector tests and
Luby's receiver both pin against this definition.
"""
string_to_sign = f"{timestamp}.".encode() + raw_body
return hmac.new(
secret_hex.encode("utf-8"), string_to_sign, hashlib.sha256
).hexdigest()
def deliver(envelope: dict) -> tuple[str, int]:
"""POST one envelope to SHOC. Returns ("delivered"|"rejected", status).
Raises RetryableDeliveryError for 429/5xx/timeout/connection failures so
the caller can block the shard (in-order retry, contract section 7).
"""
raw_body = json.dumps(envelope).encode("utf-8")
timestamp = int(time.time())
signing_key = _get_hmac_keys()[0]
signature = sign_body(signing_key["secret"], timestamp, raw_body)
request = urllib.request.Request(
SHOC_WEBHOOK_URL,
data=raw_body,
headers={
"Content-Type": "application/json; charset=utf-8",
"User-Agent": USER_AGENT,
"X-SH-Timestamp": str(timestamp),
"X-SH-Key-Id": signing_key["kid"],
"X-SH-Signature": f"v1={signature}",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=POST_TIMEOUT_SECONDS) as response:
status_code = response.status
except urllib.error.HTTPError as exc:
status_code = exc.code
if (
status_code == HTTPStatus.TOO_MANY_REQUESTS
or status_code >= _HTTP_SERVER_ERROR_MIN
):
raise RetryableDeliveryError(
f"receiver returned {status_code}", status_code=status_code
) from exc
return ("rejected", status_code)
except (TimeoutError, urllib.error.URLError, OSError) as exc:
raise RetryableDeliveryError(f"connection error: {exc}") from exc
if status_code in _HTTP_SUCCESS_RANGE:
return ("delivered", status_code)
# A non-2xx that urlopen did not raise for (e.g. an unfollowed 3xx):
# contract bug on someone's side -- park it, never block the shard.
return ("rejected", status_code)

156
lambdas/wo/shoc_emitter/envelope.py Normal file
View file

@ -0,0 +1,156 @@
"""Stream-record -> SHOC webhook envelope mapping (contract sections 3-4).
Pure and total: no boto3 clients, no environment reads, no network. Every
function either returns a value or returns None (skip) -- a malformed stream
record must never raise here; classification gaps fall through to None so the
shard is never blocked by an unmappable record. ``build_event`` is the single
entry point the handler calls per record.
Classification (docs/shoc-webhook-contract.md section 4):
WorkOrders INSERT -> work_order.created
WorkOrders MODIFY -> work_order.cancelled iff OLD wo_status was not
"cancelled" AND NEW wo_status is "cancelled",
else work_order.updated
WorkOrderComments INSERT -> work_order.comment_added
REMOVE (either table) -> skip (no deletes in this pipeline)
NEW write_origin == "shoc-write-api" -> skip (echo guard; a no-op branch
today -- nothing writes that attribute yet -- kept so the phase-2
write-back API never echoes SHOC's own writes back at it)
Anything else (unknown table, comment MODIFYs) -> skip.
"""
from datetime import datetime, timezone
from decimal import Decimal
from boto3.dynamodb.types import TypeDeserializer
SCHEMA_VERSION = 1
SOURCE = "procurement-ingest/workorder-shoc-emitter"
CANCELLED_STATUS = "cancelled"
SHOC_WRITE_ORIGIN = "shoc-write-api"
WORK_ORDERS_TABLE = "WorkOrders"
COMMENTS_TABLE = "WorkOrderComments"
# data payload field lists (contract sections 4.1 / 4.2). Absent attributes are
# null-filled; source_email_s3_key is deliberately EXCLUDED (internal key, not
# part of the contract).
WO_DATA_FIELDS = (
"work_order_id",
"wo_status",
"description",
"customer",
"site_code",
"building",
"address",
"severity",
"priority",
"assigned_to",
"date_reported",
"scheduled_start",
"due_date",
"record_type",
"created_at",
"updated_at",
)
COMMENT_DATA_FIELDS = (
"work_order_id",
"comment_id",
"record_type",
"commenter",
"text",
"created_at",
"ingested_at",
)
_deserializer = TypeDeserializer()
# arn:aws:dynamodb:...:table/NAME[/stream/LABEL] -- slash-split needs at least
# the ":table" head plus the name segment for the parse to be meaningful.
_MIN_ARN_SLASH_SEGMENTS = 2
def _plain(value):
"""Convert TypeDeserializer output to json.dumps-safe plain Python.
Mirrors lambdas/api/serialization.py: integral Decimals become JSON
integers (exact), non-integral Decimals become floats, sets become sorted
lists. Local copy on purpose -- the emitter bundles flat and must not grow
a cross-package import for twelve lines of logic.
"""
if isinstance(value, Decimal):
if value == value.to_integral_value():
return int(value)
return float(value)
if isinstance(value, dict):
return {key: _plain(item) for key, item in value.items()}
if isinstance(value, list):
return [_plain(item) for item in value]
if isinstance(value, set):
return sorted(_plain(item) for item in value)
return value
def _deserialize_image(image: dict) -> dict:
"""DynamoDB stream image (attribute-value encoded) -> plain dict."""
return {key: _plain(_deserializer.deserialize(av)) for key, av in image.items()}
def _table_name(event_source_arn: str) -> str | None:
"""Extract the exact table-name segment from a stream eventSourceARN.
ARN format: arn:aws:dynamodb:region:acct:table/NAME/stream/LABEL. Parsing
the segment exactly (not a loose substring) matters: "WorkOrders" is a
prefix of nothing, but a substring test for it WOULD match
"WorkOrderComments"-adjacent names -- segment equality can't.
"""
parts = event_source_arn.split("/")
if len(parts) >= _MIN_ARN_SLASH_SEGMENTS and parts[0].endswith(":table"):
return parts[1]
return None
def _classify(table, event_name, new_image, old_image) -> str | None:
if table == WORK_ORDERS_TABLE:
if event_name == "INSERT":
return "work_order.created"
if event_name == "MODIFY":
was_cancelled = old_image.get("wo_status") == CANCELLED_STATUS
if not was_cancelled and new_image.get("wo_status") == CANCELLED_STATUS:
return "work_order.cancelled"
return "work_order.updated"
return None
if table == COMMENTS_TABLE and event_name == "INSERT":
return "work_order.comment_added"
return None
def build_event(record: dict) -> dict | None:
"""Map one DynamoDB stream record to a webhook envelope, or None to skip."""
event_name = record.get("eventName")
if event_name == "REMOVE":
return None
stream = record.get("dynamodb") or {}
new_image = _deserialize_image(stream.get("NewImage") or {})
if new_image.get("write_origin") == SHOC_WRITE_ORIGIN:
return None # echo guard (forward-compat, contract section 3)
table = _table_name(record.get("eventSourceARN") or "")
old_image = _deserialize_image(stream.get("OldImage") or {})
event_type = _classify(table, event_name, new_image, old_image)
if event_type is None:
return None
if event_type == "work_order.comment_added":
fields = COMMENT_DATA_FIELDS
else:
fields = WO_DATA_FIELDS
# ApproximateCreationDateTime arrives as epoch seconds (float/Decimal).
occurred_at = datetime.fromtimestamp(
float(stream["ApproximateCreationDateTime"]), tz=timezone.utc
).isoformat()
return {
"schema_version": SCHEMA_VERSION,
"delivery_id": record["eventID"],
"event_type": event_type,
"occurred_at": occurred_at,
"source": SOURCE,
"replay": False,
"data": {field: new_image.get(field) for field in fields},
}

116
lambdas/wo/shoc_emitter/handler.py Normal file
View file

@ -0,0 +1,116 @@
"""
SHOC work-order webhook emitter Lambda.
Consumes the WorkOrders/WorkOrderComments DynamoDB streams and pushes each
mutation to SHOC as an HMAC-signed HTTPS POST (docs/shoc-webhook-contract.md).
This handler is the thin event loop; the work lives in flat sibling modules
(bare-name imports resolve via the same flat-landing bundling as the email
processor's siblings):
envelope.py -- stream record -> envelope mapping + event classification
delivery.py -- secret cache, signing, POST, response classification
Ships DARK: both event-source mappings deploy with enabled=False, so this code
runs zero deliveries until the activation PR flips them on after SHOC's
receiver passes the shared HMAC test vectors.
Ordering semantics: the ESMs run parallelization_factor=1 with bisect-on-error
off, and this loop processes records strictly in order. A retryable failure
(429/5xx/timeout/connection error) stops the batch immediately and reports
that record via report_batch_item_failures -- earlier successes are not
re-delivered, and the ESM blocks the shard and retries from the failed record,
preserving per-work-order commit order. Non-retryable 4xx responses are a
contract bug, not an availability blip: the full envelope is parked on the
rejected queue (alarmed) and the loop continues, so a bad payload can never
block the shard for 24 hours.
No healthcheck branch on purpose: stream consumers are not smoke-gated
(po-ingest-site-extractor precedent) -- there is no direct-invoke path to
probe, and a synthetic stream record would be a real delivery.
"""
import json
import logging
import os
import time
import boto3
import delivery
import envelope
logger = logging.getLogger()
logger.setLevel(logging.INFO)
REJECTED_QUEUE_URL = os.environ.get("REJECTED_QUEUE_URL")
# Lazy cached SQS client. Keeps the public attribute name ``sqs`` so the test
# monkeypatch target changes module only, not attribute name.
sqs = None
def _get_sqs():
global sqs
if sqs is None:
sqs = boto3.client("sqs")
return sqs
def _delivery_log(event: dict, status_code, latency_ms: int, outcome: str) -> str:
"""One structured record per delivery attempt (Logs-Insights-queryable)."""
return json.dumps(
{
"event": "shoc_delivery",
"delivery_id": event["delivery_id"],
"event_type": event["event_type"],
"work_order_id": event["data"].get("work_order_id"),
"status_code": status_code,
"latency_ms": latency_ms,
"outcome": outcome,
}
)
def _park_rejected(event: dict, status_code: int):
"""Send the full envelope to the rejected queue for operator replay."""
_get_sqs().send_message(
QueueUrl=REJECTED_QUEUE_URL,
MessageBody=json.dumps({"envelope": event, "response_status": status_code}),
)
logger.warning(
json.dumps(
{
"event": "shoc_delivery_rejected",
"delivery_id": event["delivery_id"],
"event_type": event["event_type"],
"work_order_id": event["data"].get("work_order_id"),
"response_status": status_code,
}
)
)
def handler(event, context):
"""Lambda entry point. Triggered by the two WO-table stream ESMs."""
for record in event.get("Records", []):
webhook_event = envelope.build_event(record)
if webhook_event is None:
continue
start = time.monotonic()
try:
outcome, status_code = delivery.deliver(webhook_event)
except delivery.RetryableDeliveryError as exc:
latency_ms = int((time.monotonic() - start) * 1000)
logger.warning(
_delivery_log(webhook_event, exc.status_code, latency_ms, "retryable")
)
# Stop here: reporting this record's sequence number makes the ESM
# retry from it in order; earlier successes are not re-delivered.
return {
"batchItemFailures": [
{"itemIdentifier": record["dynamodb"]["SequenceNumber"]}
]
}
latency_ms = int((time.monotonic() - start) * 1000)
logger.info(_delivery_log(webhook_event, status_code, latency_ms, outcome))
if outcome == "rejected":
_park_rejected(webhook_event, status_code)
return {"batchItemFailures": []}

0
lambdas/wo/shoc_hmac_rotator/__init__.py Normal file
View file

226
lambdas/wo/shoc_hmac_rotator/handler.py Normal file
View file

@ -0,0 +1,226 @@
"""Rotation Lambda for the SHOC webhook HMAC secret.
Secrets Manager invokes this on the 30-day rotation schedule for
``workorder-ingest/shoc-webhook-hmac`` (docs/shoc-webhook-contract.md section
6.1). Secret value shape::
{"keys": [{"kid": "<YYYY-MM-DDTHH>", "secret": "<64 hex chars>"}, ...]}
newest first, truncated to 2 entries (one overlap cycle). The emitter always
signs with ``keys[0]``; the SHOC receiver accepts any listed ``kid``.
This is *single-user* rotation: no external system holds the value -- both
sides re-fetch from Secrets Manager on their own <=5-minute cache TTLs -- so
the standard 4-step protocol collapses to createSecret/finishSecret, with
setSecret a logged no-op and testSecret validating the pending value's shape.
The secret material is machine-generated HMAC key bytes and is NEVER logged;
structured logs carry only step names, version ids, and ``kid`` values.
"""
import json
import logging
import re
import secrets
from datetime import datetime, timezone
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# One overlap cycle: the new key plus the previous key (contract section 6.1).
MAX_KEYS = 2
# 32 random bytes -> secrets.token_hex emits 64 lowercase hex chars.
SECRET_NUM_BYTES = 32
SECRET_HEX_PATTERN = re.compile(r"[0-9a-f]{64}")
KID_FORMAT = "%Y-%m-%dT%H"
# Forced re-rotation within the same hour needs a distinct kid.
KID_FORMAT_INTRA_HOUR = "%Y-%m-%dT%H%M"
# Lazy cached Secrets Manager client. Keeps the public attribute name
# ``secretsmanager_client`` so the test monkeypatch target is stable.
secretsmanager_client = None
def _get_client():
global secretsmanager_client
if secretsmanager_client is None:
secretsmanager_client = boto3.client("secretsmanager")
return secretsmanager_client
def _current_keys(client, secret_id: str) -> list:
"""Read the AWSCURRENT ``keys`` list, tolerating malformed prior values.
The CDK-seeded bootstrap value is ``{"keys": [], "bootstrap_entropy": ...}``;
any unparseable or wrong-shape prior value likewise falls back to ``[]``
(rotation then simply starts a fresh key list -- the material is fully
regenerable, receivers refresh within their TTL).
"""
try:
raw = client.get_secret_value(SecretId=secret_id, VersionStage="AWSCURRENT")
parsed = json.loads(raw["SecretString"])
except Exception:
logger.warning(
json.dumps(
{"event": "hmac_rotation_current_unreadable", "fallback": "empty"}
)
)
return []
keys = parsed.get("keys") if isinstance(parsed, dict) else None
if not isinstance(keys, list) or not all(
isinstance(key, dict)
and isinstance(key.get("kid"), str)
and isinstance(key.get("secret"), str)
for key in keys
):
logger.warning(
json.dumps(
{"event": "hmac_rotation_current_malformed", "fallback": "empty"}
)
)
return []
return keys
def _create_secret(client, secret_id: str, token: str) -> None:
"""Stage a new AWSPENDING value: fresh key prepended, list truncated to 2."""
versions = client.describe_secret(SecretId=secret_id).get("VersionIdsToStages", {})
if "AWSCURRENT" in versions.get(token, []):
logger.info(
json.dumps(
{"event": "hmac_rotation_create_already_current", "version_id": token}
)
)
return
try:
client.get_secret_value(
SecretId=secret_id, VersionId=token, VersionStage="AWSPENDING"
)
# This token's version already holds a value: retried createSecret
# invocation, nothing to do.
logger.info(
json.dumps(
{"event": "hmac_rotation_create_idempotent", "version_id": token}
)
)
return
except client.exceptions.ResourceNotFoundException:
pass
current_keys = _current_keys(client, secret_id)
now = datetime.now(timezone.utc)
new_kid = now.strftime(KID_FORMAT)
if current_keys and current_keys[0]["kid"] == new_kid:
new_kid = now.strftime(KID_FORMAT_INTRA_HOUR)
new_key = {"kid": new_kid, "secret": secrets.token_hex(SECRET_NUM_BYTES)}
client.put_secret_value(
SecretId=secret_id,
ClientRequestToken=token,
SecretString=json.dumps({"keys": [new_key] + current_keys[: MAX_KEYS - 1]}),
VersionStages=["AWSPENDING"],
)
logger.info(
json.dumps(
{
"event": "hmac_rotation_pending_staged",
"version_id": token,
"kid": new_kid,
"key_count": 1 + len(current_keys[: MAX_KEYS - 1]),
}
)
)
def _set_secret(client, secret_id: str, token: str) -> None:
"""No-op: no external system holds the value (single-user rotation)."""
logger.info(
json.dumps({"event": "hmac_rotation_set_secret_noop", "version_id": token})
)
def _test_secret(client, secret_id: str, token: str) -> None:
"""Validate the AWSPENDING value's shape; raise ValueError to fail rotation."""
pending = client.get_secret_value(
SecretId=secret_id, VersionId=token, VersionStage="AWSPENDING"
)
# json.JSONDecodeError is a ValueError subclass: bad JSON fails the rotation.
parsed = json.loads(pending["SecretString"])
keys = parsed.get("keys") if isinstance(parsed, dict) else None
if not isinstance(keys, list) or not keys:
raise ValueError("AWSPENDING value has no keys list")
head = keys[0]
if not isinstance(head, dict) or not isinstance(head.get("kid"), str):
raise ValueError("AWSPENDING keys[0] has no string kid")
secret_value = head.get("secret")
if not isinstance(secret_value, str) or not SECRET_HEX_PATTERN.fullmatch(
secret_value
):
raise ValueError("AWSPENDING keys[0].secret is not 64 lowercase hex chars")
logger.info(
json.dumps(
{
"event": "hmac_rotation_test_ok",
"version_id": token,
"kid": head["kid"],
}
)
)
def _finish_secret(client, secret_id: str, token: str) -> None:
"""Move AWSCURRENT to the pending version (idempotent on retry)."""
versions = client.describe_secret(SecretId=secret_id).get("VersionIdsToStages", {})
current_version_id = None
for version_id, stages in versions.items():
if "AWSCURRENT" in stages:
current_version_id = version_id
break
if current_version_id == token:
logger.info(
json.dumps(
{"event": "hmac_rotation_finish_idempotent", "version_id": token}
)
)
return
stage_move = {
"SecretId": secret_id,
"VersionStage": "AWSCURRENT",
"MoveToVersionId": token,
}
if current_version_id is not None:
stage_move["RemoveFromVersionId"] = current_version_id
client.update_secret_version_stage(**stage_move)
logger.info(
json.dumps(
{
"event": "hmac_rotation_finished",
"moved_to": token,
"removed_from": current_version_id,
}
)
)
def handler(event, context):
"""Secrets Manager rotation entry point.
``event`` = ``{"SecretId", "ClientRequestToken", "Step"}``; dispatches the
standard rotation steps. Unknown steps raise (rotation marked failed).
"""
secret_id = event["SecretId"]
token = event["ClientRequestToken"]
step = event["Step"]
logger.info(
json.dumps({"event": "hmac_rotation_step", "step": step, "version_id": token})
)
client = _get_client()
steps = {
"createSecret": _create_secret,
"setSecret": _set_secret,
"testSecret": _test_secret,
"finishSecret": _finish_secret,
}
if step not in steps:
raise ValueError(f"Unknown rotation step: {step}")
steps[step](client, secret_id, token)

2
pytest.ini
View file

@ -14,6 +14,8 @@ addopts =
--cov=lambdas/po/site_extractor
--cov=lambdas/api
--cov=lambdas/shared
--cov=lambdas/wo/shoc_emitter
--cov=lambdas/wo/shoc_hmac_rotator
--cov-report=term-missing
--cov-fail-under=80
; The 80% floor is an aggregate whole-suite gate, enforced in CI by the

425
scripts/replay_shoc_webhooks.py Normal file
View file

@ -0,0 +1,425 @@
"""
Rebuild SHOC work-order webhook events from DynamoDB and re-POST them to a
receiver endpoint (contract section 8 replay backstop, for the
workorder-shoc-emitter-failures / -rejected alarm runbook).
Dry-run is the DEFAULT in every mode; nothing is POSTed unless --execute is
passed. Validations (account gate, --since parsing, secret shape, unknown
--work-order-id) run on dry-run too, so the rehearsal surfaces the same
failures the real run would.
Semantics (deliberate divergences from the live emitter):
* Every envelope carries "replay": true.
* Every WorkOrders state row is emitted as work_order.updated -- a replay
reads current table state, not the stream, so it cannot distinguish the
original created (or cancelled transition). The receiver upserts
idempotently, so updated is always safe.
* delivery_id is deterministic: "replay-" + sha256 over
table#work_order_id#comment_id#occurred_at, truncated. Re-running the
same replay yields the SAME ids -- receivers dedupe on delivery_id, so
overlapping replay runs are harmless.
* occurred_at is the row's updated_at (WO) / ingested_at (comment), falling
back to now-UTC when absent.
Selection is exactly one of --work-order-id (repeatable; per-item get_item /
Query) or --since (ISO-8601 UTC; a FULL TABLE Scan on WorkOrders filtered on
updated_at >= since and on WorkOrderComments filtered on ingested_at >= since
-- a count banner is printed per table). --events narrows to state rows,
comments, or both (default).
Usage:
# dry-run two work orders (state + comments)
python scripts/replay_shoc_webhooks.py \
--url https://api.dev.seahaven.com/api/webhooks/work-orders \
--work-order-id 11144580730 --work-order-id 11144580731
# replay everything touched since a timestamp (full table Scan)
python scripts/replay_shoc_webhooks.py --url https://... \
--since 2026-07-24T02:00:00Z --execute
# comments only, for one work order
python scripts/replay_shoc_webhooks.py --url https://... \
--work-order-id 11144580730 --events comments --execute
--url is required with no default: replay must be a deliberate act against a
known receiver. Signing mirrors the emitter's delivery module byte-for-byte
(same string-to-sign, same header names; only the User-Agent gains a
"-replay" suffix). Non-2xx responses are counted as failures; the script
keeps going and exits 1 if any event failed.
"""
import argparse
import hashlib
import hmac
import json
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timedelta, timezone
from decimal import Decimal
import boto3
from boto3.dynamodb.conditions import Key
EXPECTED_ACCOUNT = "011934824531"
REGION = "us-east-1"
WO_TABLE = "WorkOrders"
COMMENTS_TABLE = "WorkOrderComments"
SECRET_NAME = "workorder-ingest/shoc-webhook-hmac"
SCHEMA_VERSION = 1
SOURCE = "procurement-ingest/workorder-shoc-emitter"
USER_AGENT = "workorder-shoc-emitter/1-replay"
POST_TIMEOUT_SECONDS = 10
DELIVERY_ID_HASH_CHARS = 32
HTTP_2XX_MIN = 200
HTTP_2XX_MAX = 300
HTTP_TOO_MANY_REQUESTS = 429
HTTP_5XX_MIN = 500
# data payload field lists -- pinned to the emitter's envelope module. Absent
# attributes are emitted as null. source_email_s3_key is deliberately
# EXCLUDED (internal S3 pointer, never leaves the account).
WO_DATA_FIELDS = [
"work_order_id",
"wo_status",
"description",
"customer",
"site_code",
"building",
"address",
"severity",
"priority",
"assigned_to",
"date_reported",
"scheduled_start",
"due_date",
"record_type",
"created_at",
"updated_at",
]
COMMENT_DATA_FIELDS = [
"work_order_id",
"comment_id",
"record_type",
"commenter",
"text",
"created_at",
"ingested_at",
]
def parse_since(value):
"""Parse the operator-supplied --since and re-emit the canonical
zero-padded ``+00:00`` isoformat string that ``updated_at`` /
``ingested_at`` are lexicographically comparable against (both are
``datetime.now(timezone.utc).isoformat()`` strings). A raw operator
string would compare WRONG under DynamoDB's byte-wise ``>=``, so parse
strictly and require an aware UTC instant."""
raw = value[:-1] + "+00:00" if value.endswith("Z") else value
try:
dt = datetime.fromisoformat(raw)
except ValueError:
sys.exit(
f"ERROR: --since {value!r} is not ISO-8601 (e.g. 2026-07-24T02:00:00Z)."
)
if dt.tzinfo is None or dt.utcoffset() != timedelta(0):
sys.exit(f"ERROR: --since {value!r} must be UTC (trailing 'Z' or '+00:00').")
return dt.isoformat()
def _json_safe(value):
"""boto3's DynamoDB resource deserializes numbers as Decimal, which
json.dumps rejects; convert recursively (defensive -- these fields are
string-typed today)."""
if isinstance(value, Decimal):
return int(value) if value == value.to_integral_value() else float(value)
if isinstance(value, list):
return [_json_safe(v) for v in value]
if isinstance(value, dict):
return {k: _json_safe(v) for k, v in value.items()}
return value
def sign_body(secret_hex, timestamp, raw_body_bytes):
"""HMAC-SHA256 over ``f"{timestamp}.{raw_body}"`` (contract section 6),
computed over the raw UTF-8 body bytes exactly as the emitter's delivery
module does -- the shared test vectors pin both implementations to
identical output. Never log the secret."""
string_to_sign = f"{timestamp}.".encode("utf-8") + raw_body_bytes
mac = hmac.new(secret_hex.encode("utf-8"), string_to_sign, hashlib.sha256)
return mac.hexdigest()
def build_headers(kid, signature, timestamp):
return {
"Content-Type": "application/json; charset=utf-8",
"User-Agent": USER_AGENT,
"X-SH-Timestamp": str(timestamp),
"X-SH-Key-Id": kid,
"X-SH-Signature": f"v1={signature}",
}
def post_event(url, raw_body_bytes, kid, secret_hex):
"""POST one signed envelope. Returns the HTTP status code, or None on
timeout / connection error."""
timestamp = int(time.time())
signature = sign_body(secret_hex, timestamp, raw_body_bytes)
request = urllib.request.Request(
url,
data=raw_body_bytes,
headers=build_headers(kid, signature, timestamp),
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=POST_TIMEOUT_SECONDS) as resp:
return resp.status
except urllib.error.HTTPError as exc:
return exc.code
except (urllib.error.URLError, TimeoutError, OSError):
return None
def classify_response(status):
"""Contract section 7 classification, printed per event for the operator
(the replay tool itself never retries -- re-run it instead)."""
if status is not None and HTTP_2XX_MIN <= status < HTTP_2XX_MAX:
return "delivered"
if status is None or status == HTTP_TOO_MANY_REQUESTS or status >= HTTP_5XX_MIN:
return "retryable"
return "non-retryable"
def fetch_signing_key(session, secret_id):
"""Fetch the HMAC secret once and return (kid, secret_hex) from keys[0]
(the producer signing key). Secret material is never printed."""
client = session.client("secretsmanager")
value = json.loads(client.get_secret_value(SecretId=secret_id)["SecretString"])
keys = value.get("keys") or []
if not keys:
sys.exit(
f"ERROR: secret {secret_id!r} has no keys (bootstrap state); run the "
"workorder-shoc-hmac-rotator rotation first."
)
return keys[0]["kid"], keys[0]["secret"]
def replay_delivery_id(table, work_order_id, comment_id, occurred_at):
"""Deterministic delivery_id, stable across replay runs, so receivers
dedupe repeated replays on it."""
seed = f"{table}#{work_order_id}#{comment_id}#{occurred_at}"
digest = hashlib.sha256(seed.encode("utf-8")).hexdigest()
return f"replay-{digest[:DELIVERY_ID_HASH_CHARS]}"
def _envelope(event_type, delivery_id, occurred_at, data):
return {
"schema_version": SCHEMA_VERSION,
"delivery_id": delivery_id,
"event_type": event_type,
"occurred_at": occurred_at,
"source": SOURCE,
"replay": True,
"data": data,
}
def build_state_event(item):
# Always work_order.updated: current-state replay cannot distinguish the
# original created/cancelled, and the receiver upserts idempotently.
occurred_at = item.get("updated_at") or datetime.now(timezone.utc).isoformat()
return _envelope(
"work_order.updated",
replay_delivery_id(WO_TABLE, item["work_order_id"], "", occurred_at),
occurred_at,
{f: _json_safe(item.get(f)) for f in WO_DATA_FIELDS},
)
def build_comment_event(item):
occurred_at = item.get("ingested_at") or datetime.now(timezone.utc).isoformat()
delivery_id = replay_delivery_id(
COMMENTS_TABLE, item["work_order_id"], item["comment_id"], occurred_at
)
return _envelope(
"work_order.comment_added",
delivery_id,
occurred_at,
{f: _json_safe(item.get(f)) for f in COMMENT_DATA_FIELDS},
)
def query_comments(table, work_order_id):
kwargs = {"KeyConditionExpression": Key("work_order_id").eq(work_order_id)}
while True:
page = table.query(**kwargs)
yield from page.get("Items", [])
lek = page.get("LastEvaluatedKey")
if not lek:
return
kwargs["ExclusiveStartKey"] = lek
def scan_since(table, attr, since):
kwargs = {
"FilterExpression": "#a >= :since",
"ExpressionAttributeNames": {"#a": attr},
"ExpressionAttributeValues": {":since": since},
}
while True:
page = table.scan(**kwargs)
yield from page.get("Items", [])
lek = page.get("LastEvaluatedKey")
if not lek:
return
kwargs["ExclusiveStartKey"] = lek
def select_by_ids(dynamodb, work_order_ids, want_state, want_comments):
envelopes = []
if want_state:
table = dynamodb.Table(WO_TABLE)
for wid in work_order_ids:
item = table.get_item(Key={"work_order_id": wid}).get("Item")
if item is None:
sys.exit(f"ERROR: work order {wid!r} not found in {WO_TABLE}.")
envelopes.append(build_state_event(item))
if want_comments:
table = dynamodb.Table(COMMENTS_TABLE)
for wid in work_order_ids:
envelopes.extend(build_comment_event(i) for i in query_comments(table, wid))
return envelopes
def select_since(dynamodb, since, want_state, want_comments):
envelopes = []
if want_state:
table = dynamodb.Table(WO_TABLE)
items = list(scan_since(table, "updated_at", since))
print(
f"[{WO_TABLE}] full table Scan (updated_at >= {since}): {len(items)} row(s)"
)
envelopes.extend(build_state_event(i) for i in items)
if want_comments:
table = dynamodb.Table(COMMENTS_TABLE)
items = list(scan_since(table, "ingested_at", since))
print(
f"[{COMMENTS_TABLE}] full table Scan (ingested_at >= {since}): "
f"{len(items)} row(s)"
)
envelopes.extend(build_comment_event(i) for i in items)
return envelopes
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument(
"--work-order-id",
action="append",
help="Replay this work order (repeatable). Exactly one of "
"--work-order-id or --since is required.",
)
ap.add_argument(
"--since",
help="Replay every row with updated_at/ingested_at >= this ISO-8601 "
"UTC timestamp (FULL TABLE Scan on both tables).",
)
ap.add_argument(
"--events",
choices=["state", "comments", "both"],
default="both",
help="Which event families to replay (default: both).",
)
ap.add_argument(
"--url",
required=True,
help="Receiver endpoint URL. Required, no default -- replay must be "
"deliberate.",
)
ap.add_argument(
"--secret-arn",
default=SECRET_NAME,
help=f"HMAC secret to sign with (default: resolve {SECRET_NAME!r} by name).",
)
ap.add_argument("--profile", help="AWS profile (default: seahaven-prod).")
ap.add_argument(
"--execute",
action="store_true",
help="Actually POST. Default is dry-run (list only).",
)
args = ap.parse_args()
if bool(args.work_order_id) == bool(args.since):
sys.exit("ERROR: pass exactly one of --work-order-id or --since.")
session = boto3.Session(
profile_name=args.profile or "seahaven-prod", region_name=REGION
)
acct = session.client("sts").get_caller_identity()["Account"]
if acct != EXPECTED_ACCOUNT:
sys.exit(
f"ERROR: profile resolves to account {acct}, expected "
f"{EXPECTED_ACCOUNT}. Aborting."
)
want_state = args.events in ("state", "both")
want_comments = args.events in ("comments", "both")
dynamodb = session.resource("dynamodb")
if args.work_order_id:
events = select_by_ids(dynamodb, args.work_order_id, want_state, want_comments)
else:
since = parse_since(args.since)
events = select_since(dynamodb, since, want_state, want_comments)
# Fetch on dry-run too: an empty/missing secret should surface on the
# rehearsal, not first appear when the operator adds --execute.
kid, secret_hex = fetch_signing_key(session, args.secret_arn)
if not events:
print("No matching events -- nothing to replay.")
return
# In-order per work order: state and comments interleave by occurred_at
# (receiver upserts + skeleton-upserts, so cross-family order is a
# nicety, not a requirement).
events.sort(
key=lambda e: (e["data"]["work_order_id"], e["occurred_at"], e["event_type"])
)
state_count = sum(1 for e in events if e["event_type"] == "work_order.updated")
print(
f"{len(events)} event(s) selected ({state_count} state, "
f"{len(events) - state_count} comment) -> {args.url}\n"
)
failed = 0
for event in events:
wid = event["data"]["work_order_id"]
if not args.execute:
print(
f"[{event['delivery_id']}] DRY-RUN: would POST {event['event_type']} wo={wid}"
)
continue
raw_body = json.dumps(event).encode("utf-8")
status = post_event(args.url, raw_body, kid, secret_hex)
outcome = classify_response(status)
status_str = "conn-error/timeout" if status is None else str(status)
print(
f"[{event['delivery_id']}] POST {event['event_type']} wo={wid} -> "
f"{status_str} ({outcome})"
)
if outcome != "delivered":
failed += 1
if not args.execute:
print("\nDry run complete. Re-run with --execute to POST.")
return
print(f"\n{len(events) - failed} delivered, {failed} failed.")
if failed:
sys.exit(1)
if __name__ == "__main__":
main()

7
tests/support/loader.py
View file

@ -66,6 +66,13 @@ _SIBLING_MODULES = (
"extraction",
"enrichment",
"persistence",
# SHOC emitter siblings (lambdas/wo/shoc_emitter/): envelope and delivery
# are both standalone (no sibling deps, stdlib + boto3 only), so appending
# them at the end keeps the tuple dependency-topological; the order between
# the two is irrelevant. They exist only under wo/shoc_emitter/, so every
# other handler load skips them via the exists() check.
"envelope",
"delivery",
)

166
tests/test_bundle_consistency.py
View file

@ -22,6 +22,8 @@ REPO_ROOT = Path(__file__).resolve().parents[1]
PO_HANDLER = REPO_ROOT / "lambdas" / "po" / "email_processor" / "handler.py"
WO_HANDLER = REPO_ROOT / "lambdas" / "wo" / "email_processor" / "handler.py"
API_HANDLER = REPO_ROOT / "lambdas" / "api" / "handler.py"
SHOC_EMITTER_HANDLER = REPO_ROOT / "lambdas" / "wo" / "shoc_emitter" / "handler.py"
SHOC_ROTATOR_HANDLER = REPO_ROOT / "lambdas" / "wo" / "shoc_hmac_rotator" / "handler.py"
PO_STACK = REPO_ROOT / "cdk" / "po_stack.py"
WO_STACK = REPO_ROOT / "cdk" / "wo_stack.py"
API_STACK = REPO_ROOT / "cdk" / "procurement_api_stack.py"
@ -116,6 +118,15 @@ API_DATA_FILES_CP_RES = (
r"(?:^|\s)api/fonts\.css(?:\s|$)",
)
# SHOC webhook emitter + HMAC rotator (lambdas/wo/shoc_emitter|shoc_hmac_rotator,
# bundled by cdk/wo_stack.py's _add_shoc_webhook_emitter). Same exact-set
# discipline as the other bundles: `cp wo/shoc_emitter/*.py` /
# `cp wo/shoc_hmac_rotator/*.py` ship every top-level .py in those dirs
# (untracked strays included -- Code.from_asset does not honor .gitignore), so
# any new module is a deliberate addition to these pins.
SHOC_EMITTER_MODULES = frozenset({"__init__", "handler", "envelope", "delivery"})
SHOC_ROTATOR_MODULES = frozenset({"__init__", "handler"})
# Pipeline-scoped glob shapes the per-stack ships-all pins accept. Phase 2
# widened the bundling cwd to the shared ../lambdas asset root, so the executed
# glob carries its own pipeline's path prefix (`po/email_processor/*.py`); a
@ -768,3 +779,158 @@ def test_api_bundle_stages_spec_and_docs_page():
f"cdk/procurement_api_stack.py bundling command must EXECUTE a cp "
f"matching {pattern!r}. Executed cp commands: {executed_cps}"
)
def _extract_shoc_emitter_command(stack_path: Path) -> str:
"""Extract the SHOC emitter's bash -c bundling command string.
Mirrors _extract_api_command but selects the command whose text mentions
``shoc_emitter`` (its cp is ``cp wo/shoc_emitter/*.py``). The plan's CI
note (docs/shoc-webhook-plan.md Phase 3) is explicit that an unrecognized
bundle ships unchecked -- the email/web_ui selectors deliberately do not
match the emitter command, so it needs its own selector. Demands exactly
one match so an ambiguity surfaces loudly instead of silently checking the
wrong command.
"""
tree = ast.parse(stack_path.read_text())
commands: list[str] = []
for node in ast.walk(tree):
if isinstance(node, ast.keyword) and node.arg == "command":
list_node = node.value
if isinstance(list_node, ast.List) and list_node.elts:
last = list_node.elts[-1]
if isinstance(last, ast.Constant) and isinstance(last.value, str):
commands.append(last.value)
shoc_cmds = [c for c in commands if "shoc_emitter" in c]
if len(shoc_cmds) != 1:
raise AssertionError(
f"expected exactly one shoc_emitter bundling command=[...] list in "
f"{stack_path}, found {len(shoc_cmds)} (of {len(commands)} total "
"command lists) — update this test to select the intended bundling "
"command explicitly"
)
return shoc_cmds[0]
def _extract_shoc_rotator_command(stack_path: Path) -> str:
"""Extract the SHOC HMAC rotator's bash -c bundling command string.
Mirrors _extract_shoc_emitter_command with the ``shoc_hmac_rotator``
selector (its cp is ``cp wo/shoc_hmac_rotator/*.py``); the emitter command
does not contain that substring, so exactly-one holds.
"""
tree = ast.parse(stack_path.read_text())
commands: list[str] = []
for node in ast.walk(tree):
if isinstance(node, ast.keyword) and node.arg == "command":
list_node = node.value
if isinstance(list_node, ast.List) and list_node.elts:
last = list_node.elts[-1]
if isinstance(last, ast.Constant) and isinstance(last.value, str):
commands.append(last.value)
rotator_cmds = [c for c in commands if "shoc_hmac_rotator" in c]
if len(rotator_cmds) != 1:
raise AssertionError(
f"expected exactly one shoc_hmac_rotator bundling command=[...] list "
f"in {stack_path}, found {len(rotator_cmds)} (of {len(commands)} "
"total command lists) — update this test to select the intended "
"bundling command explicitly"
)
return rotator_cmds[0]
def test_shoc_emitter_bundling_ships_all_first_party_siblings():
"""The SHOC emitter bundle must ship every sibling handler.py imports.
Same PR #105 ImportError class as the other bundles: the emitter handler
imports envelope and delivery as bare-name flat siblings
(lambdas/wo/shoc_emitter/); a narrowed cp would cold-start ImportError on
the first stream invocation with green CI.
"""
required = _first_party_sibling_imports(SHOC_EMITTER_HANDLER)
# Sanity: the emitter handler is known to import envelope + delivery. If
# this ever collapses to empty, ships-all would vacuously pass.
assert required == {"envelope", "delivery"}, (
f"expected the emitter handler's first-party siblings to be envelope + "
f"delivery, found {sorted(required)} — update this pin deliberately"
)
command = _extract_shoc_emitter_command(WO_STACK)
assert _bundling_ships_all(command, required), (
f"cdk/wo_stack.py shoc_emitter bundling does not ship all first-party "
f"siblings {sorted(required)}. Executed cp commands: "
f"{_executed_cp_commands(command)}"
)
def test_shoc_rotator_bundling_ships_handler():
"""The rotator bundle must ship handler.py (it has no first-party siblings).
The rotator is stdlib + boto3-from-runtime only, so its required sibling
set is empty -- pin that fact (a future sibling import must extend this
test), then prove the executed glob actually ships handler.py itself.
"""
required = _first_party_sibling_imports(SHOC_ROTATOR_HANDLER)
assert required == set(), (
f"the rotator handler grew first-party sibling imports "
f"{sorted(required)} — extend this ships-all test to require them"
)
command = _extract_shoc_rotator_command(WO_STACK)
assert _bundling_ships_all(command, {"handler"}), (
f"cdk/wo_stack.py shoc_hmac_rotator bundling does not ship handler.py. "
f"Executed cp commands: {_executed_cp_commands(command)}"
)
def test_shoc_emitter_dir_ships_no_unexpected_top_level_modules():
"""Exact-set pin on lambdas/wo/shoc_emitter/*.py -- both bounds.
`cp wo/shoc_emitter/*.py` ships every top-level .py in the dir (untracked
strays included, since Code.from_asset does not honor .gitignore), so a
stray scratch/secrets module would silently ship into the production zip.
Any new module must be deliberately added to SHOC_EMITTER_MODULES.
"""
top_level = {p.stem for p in SHOC_EMITTER_HANDLER.parent.glob("*.py")}
assert top_level == set(SHOC_EMITTER_MODULES), (
f"lambdas/wo/shoc_emitter/ top-level modules {sorted(top_level)} != "
f"pinned {sorted(SHOC_EMITTER_MODULES)}. If a new module is intended, "
"add it to SHOC_EMITTER_MODULES."
)
def test_shoc_rotator_dir_ships_no_unexpected_top_level_modules():
"""Exact-set pin on lambdas/wo/shoc_hmac_rotator/*.py -- both bounds.
Same rationale as the emitter pin: `cp wo/shoc_hmac_rotator/*.py` ships
every top-level .py in the dir, strays included.
"""
top_level = {p.stem for p in SHOC_ROTATOR_HANDLER.parent.glob("*.py")}
assert top_level == set(SHOC_ROTATOR_MODULES), (
f"lambdas/wo/shoc_hmac_rotator/ top-level modules {sorted(top_level)} "
f"!= pinned {sorted(SHOC_ROTATOR_MODULES)}. If a new module is "
"intended, add it to SHOC_ROTATOR_MODULES."
)
def test_shoc_commands_do_not_collide_with_existing_selectors():
"""The SHOC bundling commands must not trip the other exactly-one selectors.
_extract_bundling_command selects on the substring ``email_processor`` and
_extract_web_ui_command on ``web_ui``, each demanding exactly one match in
wo_stack.py. If either SHOC command ever grew one of those substrings
(e.g. a copy-pasted comment folded into the command string), those
selectors would find two commands and every email/web_ui pin would error.
Assert the invariant here so the failure names the actual cause.
"""
for command in (
_extract_shoc_emitter_command(WO_STACK),
_extract_shoc_rotator_command(WO_STACK),
):
assert "email_processor" not in command, (
f"SHOC bundling command contains 'email_processor', which would "
f"break _extract_bundling_command's exactly-one selection: "
f"{command!r}"
)
assert "web_ui" not in command, (
f"SHOC bundling command contains 'web_ui', which would break "
f"_extract_web_ui_command's exactly-one selection: {command!r}"
)

222
tests/test_replay_shoc_webhooks_contract.py Normal file
View file

@ -0,0 +1,222 @@
"""Contract tests for scripts/replay_shoc_webhooks.py (plan Phase 5).
Pins the replay tool's operator-safety and envelope contracts: dry-run is
the default and performs NO HTTP, the sts account gate rejects any profile
that does not resolve to seahaven-prod, the envelope data field lists stay
byte-identical to the emitter's (the receiver sees one schema regardless of
path), every replay envelope carries "replay": true, and delivery_id is
deterministic across runs (receivers dedupe overlapping replays on it).
The script is loaded by file path (scripts/ is not a package -- mirrors
tests/test_reprocess_contract.py); boto3.Session is monkeypatched to a plain
fake so main() runs fully offline.
"""
import importlib.util
import json
import pathlib
import sys
import pytest
from tests.support import load_lambda_module
# replay builds its boto3 session inside main(), so import is side-effect-free.
_p = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "replay_shoc_webhooks.py"
_spec = importlib.util.spec_from_file_location("replay_shoc_webhooks", _p)
replay = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(replay)
WO_ITEM = {
"work_order_id": "11144580730",
"wo_status": "assigned",
"record_type": "new_work_order",
"updated_at": "2026-07-01T00:00:00+00:00",
"source_email_s3_key": "inbound/2026/07/abc.eml",
}
COMMENT_ITEM = {
"work_order_id": "11144580730",
"comment_id": "11144580730#2026-04-27T23:51:48#a1b2c3d4e5f6",
"record_type": "comment",
"commenter": "APM Technician",
"text": "Vendor dispatched.",
"created_at": "2026-04-27T23:51:48",
"ingested_at": "2026-07-02T00:00:00+00:00",
}
# --- Offline fakes for main() ------------------------------------------------
class _FakeSTS:
def __init__(self, account):
self.account = account
def get_caller_identity(self):
return {"Account": self.account}
class _FakeSecrets:
def get_secret_value(self, SecretId): # noqa: N803 (boto3 kwarg name)
return {
"SecretString": json.dumps(
{"keys": [{"kid": "2026-07-20T00", "secret": "a" * 64}]}
)
}
class _FakeWOTable:
def get_item(self, Key): # noqa: N803 (boto3 kwarg name)
return {"Item": dict(WO_ITEM, work_order_id=Key["work_order_id"])}
class _FakeCommentsTable:
def query(self, **kwargs):
return {"Items": [dict(COMMENT_ITEM)]}
class _FakeDynamo:
def Table(self, name): # noqa: N802 (boto3 method name)
if name == replay.WO_TABLE:
return _FakeWOTable()
return _FakeCommentsTable()
class _FakeSession:
def __init__(self, account):
self.account = account
def client(self, name):
if name == "sts":
return _FakeSTS(self.account)
return _FakeSecrets()
def resource(self, name):
return _FakeDynamo()
def _run_main(monkeypatch, account, argv):
monkeypatch.setattr(replay.boto3, "Session", lambda **kwargs: _FakeSession(account))
def _no_http(*args, **kwargs):
raise AssertionError("urlopen must not be called")
monkeypatch.setattr(replay.urllib.request, "urlopen", _no_http)
monkeypatch.setattr(sys, "argv", ["replay_shoc_webhooks.py", *argv])
replay.main()
# --- Dry-run default + account gate ------------------------------------------
def test_dry_run_is_default_and_performs_no_http(monkeypatch, capsys):
# No --execute: main() must complete without ever touching urlopen (the
# patched opener raises if reached).
_run_main(
monkeypatch,
replay.EXPECTED_ACCOUNT,
["--url", "https://receiver.invalid/webhook", "--work-order-id", "wo-1"],
)
out = capsys.readouterr().out
assert "DRY-RUN: would POST" in out
assert "Dry run complete" in out
def test_account_gate_rejects_wrong_account(monkeypatch):
assert replay.EXPECTED_ACCOUNT == "011934824531"
with pytest.raises(SystemExit) as exc:
_run_main(
monkeypatch,
"999999999999",
[
"--url",
"https://receiver.invalid/webhook",
"--work-order-id",
"wo-1",
],
)
assert "999999999999" in str(exc.value)
assert replay.EXPECTED_ACCOUNT in str(exc.value)
def test_selector_is_exactly_one_of_id_or_since(monkeypatch):
with pytest.raises(SystemExit, match="exactly one"):
_run_main(
monkeypatch,
replay.EXPECTED_ACCOUNT,
["--url", "https://receiver.invalid/webhook"],
)
with pytest.raises(SystemExit, match="exactly one"):
_run_main(
monkeypatch,
replay.EXPECTED_ACCOUNT,
[
"--url",
"https://receiver.invalid/webhook",
"--work-order-id",
"wo-1",
"--since",
"2026-07-24T02:00:00Z",
],
)
# --- Envelope contract vs the emitter ----------------------------------------
def test_data_field_lists_match_emitter_exactly():
envelope = load_lambda_module("wo", "shoc_emitter/envelope")
# Same fields, same order -- the receiver sees one schema whether an
# event arrived live or via replay.
assert list(replay.WO_DATA_FIELDS) == list(envelope.WO_DATA_FIELDS)
assert list(replay.COMMENT_DATA_FIELDS) == list(envelope.COMMENT_DATA_FIELDS)
assert "source_email_s3_key" not in replay.WO_DATA_FIELDS
assert replay.SCHEMA_VERSION == envelope.SCHEMA_VERSION
assert replay.SOURCE == envelope.SOURCE
def test_replay_envelopes_carry_replay_true_and_exclude_s3_key():
state = replay.build_state_event(dict(WO_ITEM))
comment = replay.build_comment_event(dict(COMMENT_ITEM))
assert state["replay"] is True
assert comment["replay"] is True
# State replays are always work_order.updated: current-state rebuilds
# cannot distinguish the original created/cancelled, and the receiver
# upserts idempotently.
assert state["event_type"] == "work_order.updated"
assert comment["event_type"] == "work_order.comment_added"
assert "source_email_s3_key" not in state["data"]
assert set(state["data"]) == set(replay.WO_DATA_FIELDS)
assert set(comment["data"]) == set(replay.COMMENT_DATA_FIELDS)
def test_delivery_id_is_stable_across_builds():
first = replay.build_comment_event(dict(COMMENT_ITEM))
second = replay.build_comment_event(dict(COMMENT_ITEM))
assert first["delivery_id"] == second["delivery_id"]
assert first["delivery_id"].startswith("replay-")
state_first = replay.build_state_event(dict(WO_ITEM))
state_second = replay.build_state_event(dict(WO_ITEM))
assert state_first["delivery_id"] == state_second["delivery_id"]
# State and comment ids never collide (different table seeds).
assert first["delivery_id"] != state_first["delivery_id"]
# --- --since parsing ---------------------------------------------------------
def test_parse_since_canonicalizes_utc():
assert replay.parse_since("2026-07-24T02:00:00Z") == "2026-07-24T02:00:00+00:00"
assert (
replay.parse_since("2026-07-24T02:00:00+00:00") == "2026-07-24T02:00:00+00:00"
)
@pytest.mark.parametrize(
"value",
["2026-07-24T02:00:00", "2026-07-24T02:00:00+02:00", "yesterday"],
ids=["naive", "non-utc-offset", "garbage"],
)
def test_parse_since_rejects_non_utc_or_garbage(value):
with pytest.raises(SystemExit):
replay.parse_since(value)

289
tests/test_shoc_emitter_delivery.py Normal file
View file

@ -0,0 +1,289 @@
"""Signing + delivery tests for the SHOC webhook emitter (plan Phase 5).
Golden vectors: docs/shoc-webhook-test-vectors.json is the shared handoff
artifact -- Luby's receiver verifies against the same vectors -- and BOTH
producer-side sign_body implementations (the emitter's delivery module and
the replay script) are pinned here against every vector, so they can never
drift from each other or from the published vectors.
Also covers the contract section 7 response-classification matrix (2xx /
429+5xx / timeout+connection error / other 4xx) with a monkeypatched urllib
opener, and the Secrets Manager key cache: 5-minute TTL via time.monotonic,
keys[0] selection, empty-keys (bootstrap) -> retryable.
"""
import importlib.util
import io
import json
import urllib.error
from pathlib import Path
import pytest
from tests.support import REPO_ROOT, load_lambda_module
_VECTORS_PATH = Path(REPO_ROOT) / "docs" / "shoc-webhook-test-vectors.json"
VECTORS = json.loads(_VECTORS_PATH.read_text(encoding="utf-8"))["vectors"]
TEST_KEYS = [
{"kid": "2026-07-20T00", "secret": "ab" * 32},
{"kid": "2026-06-20T00", "secret": "cd" * 32},
]
ENVELOPE = {
"schema_version": 1,
"delivery_id": "evt-1",
"event_type": "work_order.created",
"occurred_at": "2026-07-16T14:03:22.114208+00:00",
"source": "procurement-ingest/workorder-shoc-emitter",
"replay": False,
"data": {"work_order_id": "11144580730"},
}
@pytest.fixture(scope="module")
def delivery():
return load_lambda_module("wo", "shoc_emitter/delivery")
def _load_replay_script():
# scripts/ is not a package and not on sys.path; load by file path
# (mirrors tests/test_reprocess_contract.py). Import is side-effect-free:
# the script builds its boto3 session inside main().
path = Path(REPO_ROOT) / "scripts" / "replay_shoc_webhooks.py"
spec = importlib.util.spec_from_file_location(
"replay_shoc_webhooks_for_vectors", path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
# --- Golden vectors ----------------------------------------------------------
def test_vector_file_covers_required_shapes():
# The handoff artifact itself must keep its coverage promises: at least
# one non-ASCII UTF-8 body (multi-byte signing) and one empty-object body.
assert len(VECTORS) >= 4
assert any(any(ord(ch) > 127 for ch in v["body"]) for v in VECTORS)
assert any(v["body"] == "{}" for v in VECTORS)
for vector in VECTORS:
assert set(vector) == {
"kid",
"secret_hex",
"timestamp",
"body",
"expected_signature",
}
def test_delivery_sign_body_matches_golden_vectors(delivery):
for vector in VECTORS:
signature = delivery.sign_body(
vector["secret_hex"],
vector["timestamp"],
vector["body"].encode("utf-8"),
)
assert signature == vector["expected_signature"], (
f"delivery.sign_body drifted from golden vector kid="
f"{vector['kid']} ts={vector['timestamp']}"
)
def test_replay_sign_body_matches_golden_vectors():
replay = _load_replay_script()
for vector in VECTORS:
signature = replay.sign_body(
vector["secret_hex"],
vector["timestamp"],
vector["body"].encode("utf-8"),
)
assert signature == vector["expected_signature"], (
f"replay sign_body drifted from golden vector kid="
f"{vector['kid']} ts={vector['timestamp']}"
)
# --- Response classification matrix (contract section 7) ---------------------
class _FakeResponse:
def __init__(self, status):
self.status = status
def __enter__(self):
return self
def __exit__(self, *exc_info):
return False
@pytest.fixture
def signing_keys(monkeypatch, delivery):
monkeypatch.setattr(delivery, "_get_hmac_keys", lambda: TEST_KEYS)
def _patch_urlopen(monkeypatch, delivery, fn):
monkeypatch.setattr(delivery.urllib.request, "urlopen", fn)
@pytest.mark.parametrize("status", [200, 204])
def test_2xx_is_delivered(monkeypatch, delivery, signing_keys, status):
_patch_urlopen(
monkeypatch, delivery, lambda request, timeout: _FakeResponse(status)
)
assert delivery.deliver(ENVELOPE) == ("delivered", status)
@pytest.mark.parametrize("status", [429, 500, 503])
def test_429_and_5xx_raise_retryable(monkeypatch, delivery, signing_keys, status):
def _raise(request, timeout):
raise urllib.error.HTTPError(
delivery.SHOC_WEBHOOK_URL, status, "boom", None, io.BytesIO(b"")
)
_patch_urlopen(monkeypatch, delivery, _raise)
with pytest.raises(delivery.RetryableDeliveryError) as exc:
delivery.deliver(ENVELOPE)
assert exc.value.status_code == status
@pytest.mark.parametrize(
"error",
[urllib.error.URLError(OSError("connection refused")), TimeoutError()],
ids=["connection-error", "timeout"],
)
def test_connection_failures_raise_retryable(
monkeypatch, delivery, signing_keys, error
):
def _raise(request, timeout):
raise error
_patch_urlopen(monkeypatch, delivery, _raise)
with pytest.raises(delivery.RetryableDeliveryError) as exc:
delivery.deliver(ENVELOPE)
assert exc.value.status_code is None
@pytest.mark.parametrize("status", [400, 404, 422])
def test_other_4xx_is_rejected_not_raised(monkeypatch, delivery, signing_keys, status):
def _raise(request, timeout):
raise urllib.error.HTTPError(
delivery.SHOC_WEBHOOK_URL, status, "bad", None, io.BytesIO(b"")
)
_patch_urlopen(monkeypatch, delivery, _raise)
assert delivery.deliver(ENVELOPE) == ("rejected", status)
def test_request_signed_with_keys0_and_contract_headers(
monkeypatch, delivery, signing_keys
):
captured = {}
def _capture(request, timeout):
captured["request"] = request
captured["timeout"] = timeout
return _FakeResponse(200)
_patch_urlopen(monkeypatch, delivery, _capture)
assert delivery.deliver(ENVELOPE) == ("delivered", 200)
request = captured["request"]
assert captured["timeout"] == delivery.POST_TIMEOUT_SECONDS
assert request.get_method() == "POST"
assert request.data == json.dumps(ENVELOPE).encode("utf-8")
assert request.get_header("Content-type") == "application/json; charset=utf-8"
assert request.get_header("User-agent") == "workorder-shoc-emitter/1"
# The producer always signs with keys[0] (contract section 6.1).
assert request.get_header("X-sh-key-id") == TEST_KEYS[0]["kid"]
timestamp = request.get_header("X-sh-timestamp")
assert timestamp.isdigit()
expected = delivery.sign_body(TEST_KEYS[0]["secret"], int(timestamp), request.data)
assert request.get_header("X-sh-signature") == f"v1={expected}"
# --- Secret cache ------------------------------------------------------------
class _FakeSecretsManager:
"""Returns payloads in sequence (last one repeats); counts fetches."""
def __init__(self, payloads):
self.payloads = list(payloads)
self.calls = 0
self.secret_ids = []
def get_secret_value(self, SecretId): # noqa: N803 (boto3 kwarg name)
self.calls += 1
self.secret_ids.append(SecretId)
payload = self.payloads.pop(0) if len(self.payloads) > 1 else self.payloads[0]
return {"SecretString": json.dumps(payload)}
@pytest.fixture
def cache_reset(monkeypatch, delivery):
monkeypatch.setattr(delivery, "_hmac_keys_cache", None)
monkeypatch.setattr(delivery, "_hmac_keys_cached_at", 0.0)
monkeypatch.setattr(
delivery,
"HMAC_SECRET_ARN",
"arn:aws:secretsmanager:us-east-1:011934824531:secret:"
"workorder-ingest/shoc-webhook-hmac-AbCdEf",
)
def _wire_cache(monkeypatch, delivery, fake, clock):
monkeypatch.setattr(delivery.boto3, "client", lambda service: fake)
monkeypatch.setattr(delivery.time, "monotonic", lambda: clock["t"])
def test_cache_honors_ttl_and_refreshes_after_expiry(
monkeypatch, delivery, cache_reset
):
rotated = [
{"keys": [{"kid": "2026-08-20T00", "secret": "ef" * 32}] + TEST_KEYS[:1]}
]
fake = _FakeSecretsManager([{"keys": TEST_KEYS}] + rotated)
clock = {"t": 1000.0}
_wire_cache(monkeypatch, delivery, fake, clock)
assert delivery._get_hmac_keys()[0]["kid"] == "2026-07-20T00"
assert fake.calls == 1
# Inside the 300s TTL: served from cache, no refetch.
clock["t"] = 1000.0 + 299.0
assert delivery._get_hmac_keys()[0]["kid"] == "2026-07-20T00"
assert fake.calls == 1
# TTL expired: refetch picks up the rotated keys[0] (cache invalidation
# is what makes 30-day rotation propagate within 5 minutes).
clock["t"] = 1000.0 + 300.5
assert delivery._get_hmac_keys()[0]["kid"] == "2026-08-20T00"
assert fake.calls == 2
assert fake.secret_ids[0] == delivery.HMAC_SECRET_ARN
def test_empty_or_missing_keys_is_retryable_bootstrap_state(
monkeypatch, delivery, cache_reset
):
clock = {"t": 5000.0}
for payload in ({"keys": []}, {"keys": [], "bootstrap_entropy": "seed"}, {}):
monkeypatch.setattr(delivery, "_hmac_keys_cache", None)
monkeypatch.setattr(delivery, "_hmac_keys_cached_at", 0.0)
fake = _FakeSecretsManager([payload])
_wire_cache(monkeypatch, delivery, fake, clock)
with pytest.raises(delivery.RetryableDeliveryError):
delivery._get_hmac_keys()
def test_secret_fetch_failure_is_retryable(monkeypatch, delivery, cache_reset):
class _Boom:
def get_secret_value(self, SecretId): # noqa: N803 (boto3 kwarg name)
raise RuntimeError("throttled")
clock = {"t": 9000.0}
_wire_cache(monkeypatch, delivery, _Boom(), clock)
with pytest.raises(delivery.RetryableDeliveryError):
delivery._get_hmac_keys()

304
tests/test_shoc_emitter_envelope.py Normal file
View file

@ -0,0 +1,304 @@
"""Stream-record -> SHOC webhook envelope mapping tests (plan Phase 5).
Pins the emitter's classification and envelope-building contract
(docs/shoc-webhook-contract.md sections 3-4) against realistic DynamoDB
stream records: event-type classification incl. the cancelled transition and
the already-cancelled no-false-cancel case, the write_origin echo guard,
exact eventSourceARN table discrimination, delivery_id/occurred_at mapping,
data null-fill + source_email_s3_key exclusion, and Decimal JSON-safety.
Also the enum golden pin: the emitted event_type set must equal the four
contract names AND the published spec's webhooks keys (lambdas/api/
openapi.json), and the classification's status/record_type value space must
stay anchored to the WO pipeline's template_parser enums.
Offline and pure -- envelope.py has no boto3 clients, no env reads.
"""
import json
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path
import pytest
from tests.support import REPO_ROOT, load_lambda_module
# Realistic stream ARNs for BOTH tables: "WorkOrders" is a leading substring
# of "WorkOrderComments", which is exactly the trap the exact-segment parse
# in envelope._table_name exists to avoid.
WO_STREAM_ARN = (
"arn:aws:dynamodb:us-east-1:011934824531:table/WorkOrders"
"/stream/2026-07-23T00:00:00.000"
)
COMMENTS_STREAM_ARN = (
"arn:aws:dynamodb:us-east-1:011934824531:table/WorkOrderComments"
"/stream/2026-07-23T00:00:00.000"
)
CONTRACT_EVENT_TYPES = {
"work_order.created",
"work_order.updated",
"work_order.cancelled",
"work_order.comment_added",
}
@pytest.fixture(scope="module")
def envelope():
return load_lambda_module("wo", "shoc_emitter/envelope")
def _record(arn, event_name, new_image=None, old_image=None, **meta):
"""Build a stream record; ``meta`` overrides event_id / creation."""
stream = {
"ApproximateCreationDateTime": meta.get("creation", 1784642602.0),
"SequenceNumber": meta.get("sequence_number", "100"),
"StreamViewType": "NEW_AND_OLD_IMAGES",
}
if new_image is not None:
stream["NewImage"] = new_image
if old_image is not None:
stream["OldImage"] = old_image
return {
"eventID": meta.get("event_id", "4b7c2f0e-0001"),
"eventName": event_name,
"eventSource": "aws:dynamodb",
"eventSourceARN": arn,
"dynamodb": stream,
}
def _wo_image(**overrides):
image = {
"work_order_id": {"S": "11144580730"},
"wo_status": {"S": "new"},
"description": {"S": "Dock door 14 won't close"},
"record_type": {"S": "new_work_order"},
"created_at": {"S": "2026-07-16T14:03:22.114208+00:00"},
"updated_at": {"S": "2026-07-16T14:03:22.114208+00:00"},
}
image.update(overrides)
return image
def _comment_image(**overrides):
image = {
"work_order_id": {"S": "11144580730"},
"comment_id": {"S": "11144580730#2026-04-27T23:51:48#a1b2c3d4e5f6"},
"record_type": {"S": "comment"},
"commenter": {"S": "APM Technician"},
"text": {"S": "Vendor dispatched, ETA tomorrow AM."},
"created_at": {"S": "2026-04-27T23:51:48"},
"ingested_at": {"S": "2026-07-16T14:05:00+00:00"},
}
image.update(overrides)
return image
# --- Classification ----------------------------------------------------------
def test_workorders_insert_is_created(envelope):
event = envelope.build_event(_record(WO_STREAM_ARN, "INSERT", _wo_image()))
assert event["event_type"] == "work_order.created"
def test_workorders_modify_is_updated(envelope):
event = envelope.build_event(
_record(
WO_STREAM_ARN,
"MODIFY",
_wo_image(wo_status={"S": "assigned"}),
_wo_image(wo_status={"S": "new"}),
)
)
assert event["event_type"] == "work_order.updated"
def test_cancelled_transition_is_cancelled(envelope):
event = envelope.build_event(
_record(
WO_STREAM_ARN,
"MODIFY",
_wo_image(wo_status={"S": "cancelled"}),
_wo_image(wo_status={"S": "assigned"}),
)
)
assert event["event_type"] == "work_order.cancelled"
def test_already_cancelled_modify_is_updated_not_cancelled(envelope):
# OLD already cancelled + NEW cancelled: any later touch to a cancelled
# WO must NOT re-emit a cancelled event (no false cancels on retries).
event = envelope.build_event(
_record(
WO_STREAM_ARN,
"MODIFY",
_wo_image(wo_status={"S": "cancelled"}),
_wo_image(wo_status={"S": "cancelled"}),
)
)
assert event["event_type"] == "work_order.updated"
def test_comment_insert_is_comment_added(envelope):
event = envelope.build_event(
_record(COMMENTS_STREAM_ARN, "INSERT", _comment_image())
)
assert event["event_type"] == "work_order.comment_added"
assert set(event["data"]) == set(envelope.COMMENT_DATA_FIELDS)
def test_remove_is_skipped(envelope):
assert envelope.build_event(_record(WO_STREAM_ARN, "REMOVE")) is None
assert envelope.build_event(_record(COMMENTS_STREAM_ARN, "REMOVE")) is None
def test_comment_modify_is_skipped(envelope):
record = _record(COMMENTS_STREAM_ARN, "MODIFY", _comment_image(), _comment_image())
assert envelope.build_event(record) is None
def test_shoc_write_origin_echo_guard_skips(envelope):
record = _record(
WO_STREAM_ARN,
"INSERT",
_wo_image(write_origin={"S": "shoc-write-api"}),
)
assert envelope.build_event(record) is None
# Any OTHER write_origin value is not an echo and must still deliver.
record = _record(
WO_STREAM_ARN,
"INSERT",
_wo_image(write_origin={"S": "email-processor"}),
)
assert envelope.build_event(record) is not None
# --- eventSourceARN table discrimination -------------------------------------
def test_table_discrimination_is_exact_segment_not_substring(envelope):
# The WorkOrders stream must classify as a WO state event and the
# WorkOrderComments stream as a comment event -- never cross-parsed even
# though "WorkOrders" is a leading substring of "WorkOrderComments".
wo_event = envelope.build_event(_record(WO_STREAM_ARN, "INSERT", _wo_image()))
assert wo_event["event_type"] == "work_order.created"
comment_event = envelope.build_event(
_record(COMMENTS_STREAM_ARN, "INSERT", _comment_image())
)
assert comment_event["event_type"] == "work_order.comment_added"
# Superstring table names must not substring-match either known table.
for table in ("WorkOrdersLegacy", "WorkOrderCommentsArchive", "XWorkOrders"):
arn = (
f"arn:aws:dynamodb:us-east-1:011934824531:table/{table}"
"/stream/2026-07-23T00:00:00.000"
)
assert envelope.build_event(_record(arn, "INSERT", _wo_image())) is None
# A malformed / missing ARN skips rather than raises (total function).
assert envelope.build_event(_record("", "INSERT", _wo_image())) is None
# --- Envelope fields ---------------------------------------------------------
def test_envelope_constant_fields_and_delivery_id(envelope):
record = _record(WO_STREAM_ARN, "INSERT", _wo_image(), event_id="evt-abc-123")
event = envelope.build_event(record)
assert event["schema_version"] == 1
assert event["source"] == "procurement-ingest/workorder-shoc-emitter"
assert event["replay"] is False
# delivery_id is the stream record eventID: unique per source event and
# stable across ESM retries (the receiver's idempotency key).
assert event["delivery_id"] == "evt-abc-123"
def test_occurred_at_from_decimal_creation_time(envelope):
# ApproximateCreationDateTime arrives as epoch seconds; via the low-level
# stream API it deserializes as Decimal -- must still map to an aware
# ISO-8601 +00:00 instant.
creation = Decimal("1784642602")
record = _record(WO_STREAM_ARN, "INSERT", _wo_image(), creation=creation)
event = envelope.build_event(record)
expected = datetime.fromtimestamp(1784642602.0, tz=timezone.utc).isoformat()
assert event["occurred_at"] == expected
assert event["occurred_at"].endswith("+00:00")
def test_data_null_fills_absent_fields_and_excludes_s3_key(envelope):
image = {
"work_order_id": {"S": "11144580730"},
"wo_status": {"S": "unknown"},
# Internal pointer: must never leave the account via the webhook.
"source_email_s3_key": {"S": "inbound/2026/07/abc.eml"},
}
event = envelope.build_event(_record(WO_STREAM_ARN, "INSERT", image))
assert set(event["data"]) == set(envelope.WO_DATA_FIELDS)
assert "source_email_s3_key" not in event["data"]
assert event["data"]["work_order_id"] == "11144580730"
for field in set(envelope.WO_DATA_FIELDS) - {"work_order_id", "wo_status"}:
assert event["data"][field] is None
def test_decimal_image_values_become_json_safe(envelope):
image = _wo_image(
severity={"N": "3"},
priority={"N": "2.5"},
)
event = envelope.build_event(_record(WO_STREAM_ARN, "INSERT", image))
assert event["data"]["severity"] == 3
assert isinstance(event["data"]["severity"], int)
assert event["data"]["priority"] == 2.5
assert isinstance(event["data"]["priority"], float)
# The whole envelope must serialize -- this is the raw body that gets
# signed and POSTed.
json.dumps(event)
# --- Enum golden pin ---------------------------------------------------------
def test_event_type_set_matches_contract_and_openapi_webhooks(envelope):
scenarios = [
_record(WO_STREAM_ARN, "INSERT", _wo_image()),
_record(
WO_STREAM_ARN,
"MODIFY",
_wo_image(wo_status={"S": "assigned"}),
_wo_image(wo_status={"S": "new"}),
),
_record(
WO_STREAM_ARN,
"MODIFY",
_wo_image(wo_status={"S": "cancelled"}),
_wo_image(wo_status={"S": "assigned"}),
),
_record(COMMENTS_STREAM_ARN, "INSERT", _comment_image()),
]
emitted = {envelope.build_event(record)["event_type"] for record in scenarios}
assert emitted == CONTRACT_EVENT_TYPES
# The published read-API spec documents the same four outbound webhook
# events; emitter and spec must move together.
spec_path = Path(REPO_ROOT) / "lambdas" / "api" / "openapi.json"
spec = json.loads(spec_path.read_text(encoding="utf-8"))
assert set(spec["webhooks"]) == emitted
def test_status_and_record_type_value_space_anchored_to_template_parser(envelope):
# The WO pipeline's template_parser enums are the wo_status/record_type
# value space the stream carries (prompts.py mirrors them for the AI
# path). The cancelled-transition trigger value must be a real status,
# and the contract's record_type enum must be exactly the pipeline's
# email-type enum -- widen either side only in a PR that updates both.
template_parser = load_lambda_module("wo", "email_processor/template_parser")
assert envelope.CANCELLED_STATUS in template_parser.VALID_STATUSES
assert template_parser.VALID_EMAIL_TYPES == {
"new_work_order",
"update",
"comment",
"cancellation",
}

177
tests/test_shoc_emitter_handler.py Normal file
View file

@ -0,0 +1,177 @@
"""Batch-loop tests for the SHOC webhook emitter handler (plan Phase 5).
Pins the partial-batch ordering contract: on a retryable failure at record i
the loop STOPS -- record i's DynamoDB SequenceNumber is reported via
report_batch_item_failures (so the ESM retries from it, in order) and later
records are never attempted; earlier in-batch successes are not re-delivered.
Non-retryable rejections park the full envelope on the rejected queue and the
loop CONTINUES (a contract bug must never block the shard). Skip records
(REMOVE / echo guard) produce no delivery attempts at all.
delivery.deliver is monkeypatched on the handler's own sibling instance (the
handler imports it by bare name); envelope.build_event runs for real, so
these batches exercise the true record -> envelope -> deliver path. SQS is a
plain fake on the handler's public ``sqs`` module global.
"""
import json
import pytest
from tests.support import load_lambda_module
WO_STREAM_ARN = (
"arn:aws:dynamodb:us-east-1:011934824531:table/WorkOrders"
"/stream/2026-07-23T00:00:00.000"
)
REJECTED_QUEUE_URL = (
"https://sqs.us-east-1.amazonaws.com/011934824531/workorder-shoc-emitter-rejected"
)
@pytest.fixture(scope="module")
def handler_mod():
return load_lambda_module("wo", "shoc_emitter/handler")
class _FakeSQS:
def __init__(self):
self.sent = []
def send_message(self, QueueUrl, MessageBody): # noqa: N803 (boto3 kwargs)
self.sent.append({"QueueUrl": QueueUrl, "MessageBody": MessageBody})
def _wo_record(event_id, sequence_number, event_name="INSERT", **image_overrides):
image = {
"work_order_id": {"S": f"wo-{event_id}"},
"wo_status": {"S": "new"},
"record_type": {"S": "new_work_order"},
}
image.update(image_overrides)
stream = {
"ApproximateCreationDateTime": 1784642602.0,
"SequenceNumber": sequence_number,
"StreamViewType": "NEW_AND_OLD_IMAGES",
}
if event_name != "REMOVE":
stream["NewImage"] = image
return {
"eventID": event_id,
"eventName": event_name,
"eventSource": "aws:dynamodb",
"eventSourceARN": WO_STREAM_ARN,
"dynamodb": stream,
}
def _wire(monkeypatch, handler_mod, outcomes):
"""Patch delivery.deliver with a per-delivery_id outcome table.
``outcomes`` maps delivery_id (the record eventID) to either a
("delivered"|"rejected", status) tuple or the string "retryable" (raise).
Returns the ordered list of attempted delivery_ids and the fake SQS.
"""
attempted = []
def _fake_deliver(webhook_event):
attempted.append(webhook_event["delivery_id"])
outcome = outcomes[webhook_event["delivery_id"]]
if outcome == "retryable":
raise handler_mod.delivery.RetryableDeliveryError(
"receiver returned 503", status_code=503
)
return outcome
monkeypatch.setattr(handler_mod.delivery, "deliver", _fake_deliver)
fake_sqs = _FakeSQS()
monkeypatch.setattr(handler_mod, "sqs", fake_sqs)
monkeypatch.setattr(handler_mod, "REJECTED_QUEUE_URL", REJECTED_QUEUE_URL)
return attempted, fake_sqs
def test_retryable_middle_record_stops_batch_and_reports_its_sequence(
monkeypatch, handler_mod
):
event = {
"Records": [
_wo_record("evt-1", "101"),
_wo_record("evt-2", "102"),
_wo_record("evt-3", "103"),
]
}
attempted, fake_sqs = _wire(
monkeypatch,
handler_mod,
{"evt-1": ("delivered", 200), "evt-2": "retryable"},
)
result = handler_mod.handler(event, None)
# EXACTLY the failed record's SequenceNumber: the ESM retries from it in
# order, and evt-1 (already delivered) is not re-delivered.
assert result == {"batchItemFailures": [{"itemIdentifier": "102"}]}
# The loop stopped at the failure: record 3 was never attempted.
assert attempted == ["evt-1", "evt-2"]
assert fake_sqs.sent == []
def test_rejected_record_parks_to_sqs_and_processing_continues(
monkeypatch, handler_mod
):
event = {
"Records": [
_wo_record("evt-1", "201"),
_wo_record("evt-2", "202"),
_wo_record("evt-3", "203"),
]
}
attempted, fake_sqs = _wire(
monkeypatch,
handler_mod,
{
"evt-1": ("delivered", 200),
"evt-2": ("rejected", 422),
"evt-3": ("delivered", 204),
},
)
result = handler_mod.handler(event, None)
# A 4xx rejection must NOT block the shard: no batch item failures, and
# the records after the rejection were still attempted.
assert result == {"batchItemFailures": []}
assert attempted == ["evt-1", "evt-2", "evt-3"]
# The full envelope was parked for operator replay.
assert len(fake_sqs.sent) == 1
assert fake_sqs.sent[0]["QueueUrl"] == REJECTED_QUEUE_URL
parked = json.loads(fake_sqs.sent[0]["MessageBody"])
assert set(parked) == {"envelope", "response_status"}
assert parked["response_status"] == 422
assert parked["envelope"]["delivery_id"] == "evt-2"
assert parked["envelope"]["event_type"] == "work_order.created"
assert parked["envelope"]["data"]["work_order_id"] == "wo-evt-2"
def test_skip_records_produce_no_delivery_calls(monkeypatch, handler_mod):
event = {
"Records": [
_wo_record("evt-1", "301", event_name="REMOVE"),
_wo_record("evt-2", "302", write_origin={"S": "shoc-write-api"}),
]
}
attempted, fake_sqs = _wire(monkeypatch, handler_mod, {})
result = handler_mod.handler(event, None)
assert result == {"batchItemFailures": []}
assert attempted == []
assert fake_sqs.sent == []
def test_empty_batch_returns_no_failures(monkeypatch, handler_mod):
attempted, _ = _wire(monkeypatch, handler_mod, {})
assert handler_mod.handler({"Records": []}, None) == {"batchItemFailures": []}
assert attempted == []

306
tests/test_shoc_hmac_rotator.py Normal file
View file

@ -0,0 +1,306 @@
"""Rotation-step tests for workorder-shoc-hmac-rotator (plan Phase 5).
Exercises the Secrets Manager rotation protocol against a dict-backed fake
client (no moto, no AWS): createSecret from the CDK bootstrap value and from
a populated list (newest-first, truncated to 2), createSecret idempotency on
retried tokens, the intra-hour kid-collision fallback, testSecret's
shape/hex validation, finishSecret's stage move + idempotency, the setSecret
no-op, and the unknown-step failure.
datetime is monkeypatched to a fixed instant so kid assertions are exact and
the collision case is deterministic (no wall-clock hour-boundary flake).
"""
import json
import re
from datetime import datetime, timezone
import pytest
from tests.support import load_lambda_module
SECRET_ID = "workorder-ingest/shoc-webhook-hmac"
HEX_64_RE = re.compile(r"^[0-9a-f]{64}$")
# Fixed rotation instant: kid "2026-07-24T15" (intra-hour fallback
# "2026-07-24T1507").
FIXED_NOW = datetime(2026, 7, 24, 15, 7, 42, tzinfo=timezone.utc)
KEY_A = {"kid": "2026-06-20T00", "secret": "a" * 64}
KEY_B = {"kid": "2026-05-21T00", "secret": "b" * 64}
class _ResourceNotFound(Exception):
pass
class _Exceptions:
ResourceNotFoundException = _ResourceNotFound
class FakeSecretsManager:
"""Dict-backed Secrets Manager stub recording every mutating call.
``versions`` maps version_id -> {"stages": set[str], "value": str}.
"""
exceptions = _Exceptions
def __init__(self, versions=None):
self.versions = {
vid: {"stages": set(v["stages"]), "value": v["value"]}
for vid, v in (versions or {}).items()
}
self.put_calls = []
self.stage_calls = []
def describe_secret(self, SecretId): # noqa: N803 (boto3 kwarg names)
return {
"VersionIdsToStages": {
vid: sorted(v["stages"]) for vid, v in self.versions.items()
}
}
def get_secret_value( # noqa: N803 (boto3 kwarg names)
self, SecretId, VersionId=None, VersionStage=None
):
if VersionId is not None:
version = self.versions.get(VersionId)
if version is None or (
VersionStage is not None and VersionStage not in version["stages"]
):
raise _ResourceNotFound(f"{VersionId} / {VersionStage}")
return {"SecretString": version["value"]}
stage = VersionStage or "AWSCURRENT"
for version in self.versions.values():
if stage in version["stages"]:
return {"SecretString": version["value"]}
raise _ResourceNotFound(stage)
def put_secret_value( # noqa: N803 (boto3 kwarg names)
self, SecretId, ClientRequestToken, SecretString, VersionStages
):
self.put_calls.append(
{
"SecretId": SecretId,
"ClientRequestToken": ClientRequestToken,
"SecretString": SecretString,
"VersionStages": list(VersionStages),
}
)
self.versions[ClientRequestToken] = {
"stages": set(VersionStages),
"value": SecretString,
}
def update_secret_version_stage(self, **kwargs):
self.stage_calls.append(kwargs)
stage = kwargs["VersionStage"]
removed = kwargs.get("RemoveFromVersionId")
if removed is not None:
self.versions[removed]["stages"].discard(stage)
self.versions[kwargs["MoveToVersionId"]]["stages"].add(stage)
class _FixedDatetime:
@classmethod
def now(cls, tz=None):
return FIXED_NOW
@pytest.fixture(scope="module")
def rotator():
return load_lambda_module("wo", "shoc_hmac_rotator/handler")
@pytest.fixture
def fixed_now(monkeypatch, rotator):
monkeypatch.setattr(rotator, "datetime", _FixedDatetime)
def _run(monkeypatch, rotator, fake, step, token="tok-1"):
monkeypatch.setattr(rotator, "secretsmanager_client", fake)
rotator.handler(
{"SecretId": SECRET_ID, "ClientRequestToken": token, "Step": step}, None
)
def _bootstrap_fake(value=None):
current = value if value is not None else {"keys": [], "bootstrap_entropy": "x"}
return FakeSecretsManager(
{"cur-1": {"stages": {"AWSCURRENT"}, "value": json.dumps(current)}}
)
# --- createSecret ------------------------------------------------------------
def test_create_from_bootstrap_stages_one_fresh_key(monkeypatch, rotator, fixed_now):
fake = _bootstrap_fake()
_run(monkeypatch, rotator, fake, "createSecret")
assert len(fake.put_calls) == 1
put = fake.put_calls[0]
assert put["SecretId"] == SECRET_ID
assert put["ClientRequestToken"] == "tok-1"
assert put["VersionStages"] == ["AWSPENDING"]
staged = json.loads(put["SecretString"])
assert list(staged) == ["keys"]
assert len(staged["keys"]) == 1
assert staged["keys"][0]["kid"] == "2026-07-24T15"
assert HEX_64_RE.fullmatch(staged["keys"][0]["secret"])
def test_create_prepends_and_truncates_to_two_newest_first(
monkeypatch, rotator, fixed_now
):
fake = _bootstrap_fake({"keys": [KEY_A, KEY_B]})
_run(monkeypatch, rotator, fake, "createSecret")
staged = json.loads(fake.put_calls[0]["SecretString"])
assert len(staged["keys"]) == 2
# Newest first: the fresh key leads, the previous head is retained for
# one overlap cycle, the oldest key falls off.
assert staged["keys"][0]["kid"] == "2026-07-24T15"
assert staged["keys"][1] == KEY_A
assert KEY_B["secret"] not in fake.put_calls[0]["SecretString"]
def test_create_is_idempotent_when_pending_already_staged(
monkeypatch, rotator, fixed_now
):
fake = _bootstrap_fake({"keys": [KEY_A]})
fake.versions["tok-1"] = {
"stages": {"AWSPENDING"},
"value": json.dumps({"keys": [KEY_A]}),
}
_run(monkeypatch, rotator, fake, "createSecret")
assert fake.put_calls == []
def test_create_is_noop_when_token_already_current(monkeypatch, rotator, fixed_now):
fake = FakeSecretsManager(
{
"tok-1": {
"stages": {"AWSCURRENT"},
"value": json.dumps({"keys": [KEY_A]}),
}
}
)
_run(monkeypatch, rotator, fake, "createSecret")
assert fake.put_calls == []
def test_create_kid_collision_same_hour_gets_minute_suffix(
monkeypatch, rotator, fixed_now
):
# Forced re-rotation within the same hour: keys[0].kid already equals the
# hour-format kid, so the new kid extends to minutes for uniqueness.
fake = _bootstrap_fake({"keys": [{"kid": "2026-07-24T15", "secret": "c" * 64}]})
_run(monkeypatch, rotator, fake, "createSecret")
staged = json.loads(fake.put_calls[0]["SecretString"])
assert staged["keys"][0]["kid"] == "2026-07-24T1507"
assert staged["keys"][1]["kid"] == "2026-07-24T15"
def test_create_malformed_current_value_starts_fresh_list(
monkeypatch, rotator, fixed_now
):
fake = FakeSecretsManager(
{"cur-1": {"stages": {"AWSCURRENT"}, "value": "not json {"}}
)
_run(monkeypatch, rotator, fake, "createSecret")
staged = json.loads(fake.put_calls[0]["SecretString"])
assert len(staged["keys"]) == 1
assert HEX_64_RE.fullmatch(staged["keys"][0]["secret"])
# --- testSecret --------------------------------------------------------------
def _pending_fake(value):
return FakeSecretsManager(
{"tok-1": {"stages": {"AWSPENDING"}, "value": json.dumps(value)}}
)
def test_test_secret_accepts_valid_pending_value(monkeypatch, rotator):
fake = _pending_fake({"keys": [{"kid": "2026-07-24T15", "secret": "d" * 64}]})
_run(monkeypatch, rotator, fake, "testSecret") # no raise
@pytest.mark.parametrize(
"pending_value",
[
{"keys": []},
{"keys": [{"kid": "2026-07-24T15", "secret": "d" * 63}]},
{"keys": [{"kid": "2026-07-24T15", "secret": "z" * 64}]},
{"keys": [{"kid": "2026-07-24T15", "secret": "D" * 64}]},
{"keys": [{"secret": "d" * 64}]},
],
ids=["empty-keys", "short-secret", "non-hex", "uppercase-hex", "missing-kid"],
)
def test_test_secret_rejects_bad_pending_values(monkeypatch, rotator, pending_value):
fake = _pending_fake(pending_value)
with pytest.raises(ValueError):
_run(monkeypatch, rotator, fake, "testSecret")
# --- finishSecret ------------------------------------------------------------
def test_finish_moves_current_stage_from_old_version(monkeypatch, rotator):
fake = FakeSecretsManager(
{
"old-1": {
"stages": {"AWSCURRENT"},
"value": json.dumps({"keys": [KEY_A]}),
},
"tok-1": {
"stages": {"AWSPENDING"},
"value": json.dumps({"keys": [KEY_B, KEY_A]}),
},
}
)
_run(monkeypatch, rotator, fake, "finishSecret")
assert fake.stage_calls == [
{
"SecretId": SECRET_ID,
"VersionStage": "AWSCURRENT",
"MoveToVersionId": "tok-1",
"RemoveFromVersionId": "old-1",
}
]
assert "AWSCURRENT" in fake.versions["tok-1"]["stages"]
assert "AWSCURRENT" not in fake.versions["old-1"]["stages"]
def test_finish_is_idempotent_when_token_already_current(monkeypatch, rotator):
fake = FakeSecretsManager(
{
"tok-1": {
"stages": {"AWSCURRENT", "AWSPENDING"},
"value": json.dumps({"keys": [KEY_A]}),
}
}
)
_run(monkeypatch, rotator, fake, "finishSecret")
assert fake.stage_calls == []
# --- Dispatch ----------------------------------------------------------------
def test_set_secret_is_a_noop(monkeypatch, rotator):
fake = _bootstrap_fake()
_run(monkeypatch, rotator, fake, "setSecret")
assert fake.put_calls == []
assert fake.stage_calls == []
def test_unknown_step_raises(monkeypatch, rotator):
fake = _bootstrap_fake()
with pytest.raises(ValueError, match="Unknown rotation step"):
_run(monkeypatch, rotator, fake, "rotateHarder")