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

Merge remote-tracking branch 'origin/main' into feat/shoc-wo-webhook

Browse source
This commit is contained in:
Adam Moussa 2026-07-24 14:24:20 -04:00
commit 5abc369e02
No known key found for this signature in database
32 changed files with 5058 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
.github/dependabot.yml vendored
View file

@ -72,3 +72,12 @@ updates:
update-types:
- "minor"
- "patch"
- package-ecosystem: "npm"
directory: "/"
schedule:
interval: "weekly"
groups:
minor-and-patch:
update-types:
- "minor"
- "patch"

12
.github/workflows/ci.yaml vendored
View file

@ -14,3 +14,15 @@ jobs:
run-tests: true
run-cdk-synth: true
run-sam-validate: false
# The published OpenAPI contract must stay compliant with redocly.yaml;
# this is the CI mirror of the local `npm run lint:api`.
spec-lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
- uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
with:
node-version: 22
cache: npm
- run: npm ci
- run: npm run lint:api

17
.redocly.lint-ignore.yaml Normal file
View file

@ -0,0 +1,17 @@
# This file instructs Redocly's linter to ignore the rules contained for specific parts of your API.
# See https://redocly.com/docs/cli/ for more information.
#
# operation-2xx-response: the two x-planned phase-2 write endpoints answer
# ONLY 501 until built -- documenting a 2xx they cannot return would lie.
# paths-kebab-case: webhook keys are the shipped SHOC contract event names
# (docs/shoc-webhook-contract.md Rev 2026-07-23, event_type enum) -- not
# renameable without a contract break.
lambdas/api/openapi.json:
operation-2xx-response:
- '#/paths/~1work-orders~1{workOrderId}/patch/responses'
- '#/paths/~1work-orders~1{workOrderId}~1comments/post/responses'
paths-kebab-case:
- '#/webhooks/work_order.created'
- '#/webhooks/work_order.updated'
- '#/webhooks/work_order.cancelled'
- '#/webhooks/work_order.comment_added'

57
README.md
View file

@ -86,9 +86,28 @@ Amazon APM work order emails (from Hexagon EAM / HxGN SmartCloud) are received a
- `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
### Procurement API (`procurement-api` stack)
A read-only REST API (API Gateway + the `procurement-api` Lambda, `lambdas/api/`) over both pipelines' tables, plus a token-gated OpenAPI docs page. Primary consumer: the SHOC backend, for which this replaces the retired SyncController cross-account DynamoDB scan as the reconciliation/backfill path (and the initial-history load for the outbound work-order webhook, which starts at LATEST).
| Method + path | Auth | Backing read |
|---|---|---|
| `GET /work-orders` · `/purchase-orders` · `/verified-sites` | IAM SigV4 | unordered paginated Scan (`limit` 1–500, opaque `cursor` = base64 `LastEvaluatedKey`; malformed cursor → 400) |
| `GET /work-orders/{id}` · `/purchase-orders/{id}` · `/verified-sites/{siteCode}` | IAM SigV4 | GetItem (404 on miss) |
| `GET /work-orders/{id}/comments` | IAM SigV4 | Query on the partition key, paginated |
| `POST /work-orders/{id}/comments`, `PATCH /work-orders/{id}` | — | **phase-2 planned** (`x-planned` in the spec); handler answers 501 |
| `GET /docs`, `GET /openapi.json` | shared docs token (`X-Auth-Token` header or `?token=` in a browser) | Redoc reference docs (vendored offline, no CDN) with the spec inlined / the committed spec |
- **Spec is source of truth:** `lambdas/api/openapi.json` (OpenAPI 3.1). Its top-level `webhooks` section documents the outbound SHOC work-order feed, so one page describes both directions (call + be-called). `tests/test_api_spec_drift.py` pins the spec's paths to the router table, so spec and implementation cannot drift.
- **Auth:** data routes use API Gateway `AWS_IAM` (SigV4) plus a resource policy allowing exactly `arn:aws:iam::396287094661:role/shoc-backend-dev` on `GET/*`; same-account admin callers authorize via identity policy (Postman signs SigV4 natively). Docs routes are auth `NONE` at the gateway (resource-policy carve-out for exactly those two GETs) but the handler fails closed on the shared token (`lambdas/shared/web_ui_auth.py`, secret `procurement-ingest/web-ui-auth-token`) — not an unauthenticated data path (INFRA-74 posture).
- **KMS:** `purchase-orders` is CMK-encrypted; the imported-by-name table doesn't carry the key association, so the stack grants `kms:Decrypt`/`DescribeKey` on the CMK from SSM `/seahaven/dynamodb/cmk-arn` explicitly (the INFRA-104 failure class).
- **No access logging in v1** (keeps the `?token=` shim out of any log and avoids the account-level API Gateway CloudWatch role); rotate the docs token before ever enabling it. No CORS (server-to-server + Postman callers).
- **Alarms:** `procurement-api-errors`/`-throttles`/`-duration` (p99 ≥ 22.5 s) + gateway `procurement-api-5xx`, all ALARM-only → `site-alerts`. No 4XX alarm (401/403/404 are expected traffic).
- **Docs page:** `/docs` serves Redoc (read-only reference docs; live calls go through Postman since data routes need SigV4) as ONE token-gated response: the handler inlines the vendored `redoc.standalone.js`, the design-system fonts (`fonts.css`, data-URI `@font-face` — nothing may fetch from Google Fonts, test-pinned), and the spec into `docs.html`, with `</script>`/`</style>` breakout guards on every blob. Theme = SHOC tokens (Montserrat/DM Sans/JetBrains Mono, primary `#1c75bc`, 64px gradient topbar). The right-panel gradient and the topbar's Hide/Show-samples toggle target styled-components class names that are deterministic for the pinned Redoc bundle but change on any bump — re-derive them then (headless probe: elements whose computed background equals the `rightPanel` color); stale selectors degrade to a solid panel / inert toggle, cosmetic only. Tooling: `npm run lint:api` lints the spec against `redocly.yaml` (CI job `spec-lint`; deliberate exceptions live in `.redocly.lint-ignore.yaml`), `npm run docs:preview` renders the real handler output locally.
## Architecture
**IaC:** AWS CDK (Python), two stacks in one app, region `us-east-1`. The `cdk.Environment` is deliberately **account-agnostic** (region-only, no `account=`): the stacks deploy to whichever account the deploy credentials target (`011934824531` seahaven-prod since the 2026-07 account migration; formerly mgmt `328440206208`), and every account-derived template value — bucket names, `Lambda::Permission` source account, the site-alerts SNS action ARN, the Bedrock ARN below — renders as the CloudFormation `AWS::AccountId` pseudo-parameter rather than a literal. Pinning `account=` was evaluated in Phase 4 and rejected: it would resolve those tokens to literals, and against the deployed (account-agnostic) templates CloudFormation flags the `RemovalPolicy.RETAIN` email buckets as requiring replacement — a data-loss risk — for no functional gain.
**IaC:** AWS CDK (Python), three stacks in one app, region `us-east-1`. The `cdk.Environment` is deliberately **account-agnostic** (region-only, no `account=`): the stacks deploy to whichever account the deploy credentials target (`011934824531` seahaven-prod since the 2026-07 account migration; formerly mgmt `328440206208`), and every account-derived template value — bucket names, `Lambda::Permission` source account, the site-alerts SNS action ARN, the Bedrock ARN below — renders as the CloudFormation `AWS::AccountId` pseudo-parameter rather than a literal. Pinning `account=` was evaluated in Phase 4 and rejected: it would resolve those tokens to literals, and against the deployed (account-agnostic) templates CloudFormation flags the `RemovalPolicy.RETAIN` email buckets as requiring replacement — a data-loss risk — for no functional gain.
All Lambdas: Python 3.12, ARM64, 60-day log retention.
@ -179,7 +198,7 @@ if isinstance(event, dict) and event.get("healthcheck") is True:
This placement is deliberate, not incidental: real mail always arrives as an S3 `ObjectCreated` event whose top-level keys (`Records`) AWS controls, so email content can never set a top-level `healthcheck` key — the branch creates no accept path for forged mail. It also emits **no EMF metric and no log line**, so it can never match the `sender_auth_rejected` log-metric-filter pattern that feeds the `<fn>-sender-auth-rejected` alarm (see [CloudWatch alarms](#cloudwatch-alarms)) — that alarm pages at ≥1 match in its window, so repeated healthcheck invokes across deploys (two deploys in ~30 min is routine) must never contribute to it. Unit coverage: `lambdas/po/email_processor/tests/test_po_healthcheck.py` and `lambdas/wo/email_processor/tests/test_healthcheck.py`.
**Post-deploy smoke gate.** `scripts/post-deploy-smoke.sh` is wired into the CD workflow as `cd-cdk.yaml`'s `post-deploy-script` input (see [CI/CD](#cicd)) and runs synchronously after every deploy, before the workflow is considered green. It invokes both `po-email-processor` and `workorder-email-processor` with `aws lambda invoke --invocation-type RequestResponse --payload '{"healthcheck": true}'` (region `us-east-1`) and asserts, per function:
**Post-deploy smoke gate.** `scripts/post-deploy-smoke.sh` is wired into the CD workflow as `cd-cdk.yaml`'s `post-deploy-script` input (see [CI/CD](#cicd)) and runs synchronously after every deploy, before the workflow is considered green. It invokes `po-email-processor`, `workorder-email-processor`, and `procurement-api` with `aws lambda invoke --invocation-type RequestResponse --payload '{"healthcheck": true}'` (region `us-east-1`) and asserts, per function:
1. The invoke response's **`FunctionError` field is absent** — this is the load-bearing check. A broken bundle (e.g. an `ImportError` at module init from a missing sibling module) still returns HTTP 200 from the Lambda Invoke API with `FunctionError=Unhandled`; a bare exit-code check on `aws lambda invoke` would false-pass on exactly the failure this gate exists to catch.
2. The returned payload is **exactly** `{"healthcheck": "ok"}`.
@ -248,13 +267,13 @@ The `purchase-orders` DynamoDB table is **owned by this repo's `po-ingest` stack
**Consumers (read-only):**
| Repo | How it reads | Purpose |
| Consumer | How it reads | Purpose |
|---|---|---|
| `seahaven-slack-bot` | `po-sync` (DynamoDB Streams + daily scan) and `wo-po-lookup` | Daily KB sync + Bedrock agent PO lookups |
| `procurement-api` (this repo) | `Table.from_table_name` + `grant_read_data` | REST reads (`GET /purchase-orders*`) |
The consumer imports the table via `Table.fromTableName(...)` and is granted read-only access (`grantReadData`); it does not own or define it.
`seahaven-slack-bot`, the former external consumer, was decommissioned 2026-07-23 and its cross-account grants removed. External consumers now read through the `procurement-api` REST contract (`lambdas/api/openapi.json`), never by importing the table directly.
**Schema-coordination rule:** Any change to the `purchase-orders` schema (partition key, item shape, attribute names, streams view type) must be coordinated with `seahaven-slack-bot`. The owner here ships the change; the consumer must be updated in lockstep so its readers do not break. Treat schema changes as a cross-repo migration, not a local edit.
**Schema-coordination rule:** Any change to the `purchase-orders` schema (partition key, item shape, attribute names, streams view type) must be reflected in the API's OpenAPI schemas in the same PR (the spec-drift test pins paths, not item shapes — the schema fields are maintained by hand). Treat schema changes as a contract migration, not a local edit.
**Known exception (INFRA-51):** `amazon-po-parser` currently writes directly to `purchase-orders` outside this stack (backfill/enrichment scripts). This second writer is being folded into the `po-ingest` pipeline so this stack is the sole writer; until INFRA-51 closes, coordinate any schema change with `amazon-po-parser` as well.
@ -284,7 +303,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.
**Consumer (read-only) — data contract:** `seahaven-slack-bot` imports both tables via `Table.fromTableName(...)` (`grantReadData`) and reads them from two Lambdas: `workorder-sync` (daily full-table scan into the Bedrock knowledge base) and `wo-po-lookup` (the Bedrock agent's direct WO lookup action group). The bot depends on the PK/SK schema above and these attributes: on `WorkOrders` — `description`, `wo_status`, `customer`, `site_code`, `building`, `address`, `severity`, `priority`, `assigned_to`, `date_reported`, `scheduled_start`, `due_date`, `updated_at`; on `WorkOrderComments` — `created_at` (used to sort comments), `commenter`, `text`. Any change to table name, key schema, these attribute names, or the encryption configuration (e.g. the INFRA-6 CMK migration) must be coordinated with `seahaven-slack-bot` before it ships, or the Bedrock agent breaks at runtime (not at deploy — the tables are imported by name, so there is no compile-time link).
**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.
**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.
@ -292,9 +311,7 @@ Both currently use default DynamoDB encryption — they are **not** yet on the s
Owned by this repo's `po-ingest` stack (`cdk/po_stack.py`). PK `siteCode` (S); default DynamoDB encryption (NOT the shared CMK).
**Consumer (read-only) — data contract:** `seahaven-slack-bot`'s `wo-po-lookup` Lambda imports this table via `Table.fromTableName(...)` for the Bedrock agent's `lookup_site` action. It does point lookups by `siteCode` and reads `address`, `fullAddress`, `city`, `state`, `zip`, `latitude`, `longitude`, `notes`. Coordinate any change to the table name, key schema, or these attribute names with `seahaven-slack-bot`.
> **GSI drift (INFRA-138):** the `by-state` GSI was removed here on 2026-06-03 (audit M-20, "0 reads in 30d"), but `seahaven-slack-bot` still queries `IndexName: 'by-state'` for its state-listing path, so that path fails at runtime today. Restoring the GSI or removing the consumer's state path needs to be reconciled cross-repo. This is the kind of silent owner-side lifecycle change this data-contract note exists to prevent.
**Consumer (read-only) — data contract:** `seahaven-slack-bot` (former reader, incl. the INFRA-138/INFRA-180 `by-state` GSI drift saga) was decommissioned 2026-07-23 — the GSI-drift issue died with it. Current consumer: the `procurement-api` Lambda (`GET /verified-sites*`). The API's `VerifiedSite` schema deliberately promises only pipeline-written fields (`siteCode`, `address`, `city`, `state`, `zip`, `fullAddress`, `locationCode`, `poCount`, `sourcePOs`) — `latitude`/`longitude`/`notes` were manually curated legacy attributes the extractor never writes, and are not part of the contract.
## Documentation
@ -306,7 +323,7 @@ The canonical map of Sea Haven's AWS infrastructure lives in Confluence. This pr
GitHub Actions with reusable workflows from `Sea-Haven-Industries/.github` (all pinned to a commit SHA of `main`):
- **CI** (`ci.yaml`, PR to `main`): linting + `cdk synth` via `ci-python-sam.yaml`. `cdk synth`'s Docker-bundled asset build for `po-email-processor` and `workorder-email-processor` mounts the widened `../lambdas` asset root (Phase 2, see [Deploy-Pipeline Guards](#deploy-pipeline-guards-phase-0)) as build context, not just each function's own subdirectory — the `exclude` list on both `from_asset` calls strips local-only `__pycache__`/`package/` (and `tests/`) cruft from that wider mount's source fingerprint, so CI's asset hash matches a clean local checkout, and each function's scoped `cp` glob copies only its own pipeline's `*.py` into the zip. (The exclude does not, and under `SOURCE` hashing cannot, keep the *other* pipeline's tracked source out of the fingerprint (see the PO bundling note above on `SOURCE` hashing) — but that source is identical in CI and local, so it does not cause hash divergence.)
- **CD** (`deploy.yaml`, push to `main`): CDK deploy via `cd-cdk.yaml` (OIDC auth), followed by the synchronous `post-deploy-script: scripts/post-deploy-smoke.sh` healthcheck gate (see [Deploy-Pipeline Guards](#deploy-pipeline-guards-phase-0)) — `cd-cdk.yaml`'s `stack-name` input only accepts one stack, so the smoke script itself enumerates both `po-email-processor` and `workorder-email-processor`
- **CD** (`deploy.yaml`, push to `main`): CDK deploy via `cd-cdk.yaml` (OIDC auth), followed by the synchronous `post-deploy-script: scripts/post-deploy-smoke.sh` healthcheck gate (see [Deploy-Pipeline Guards](#deploy-pipeline-guards-phase-0)) — `cd-cdk.yaml`'s `stack-name` input only accepts one stack, so the smoke script itself enumerates `po-email-processor`, `workorder-email-processor`, and `procurement-api`
- Plus dependency review and PR labeler workflows on every PR
Branch protection on `main` — all changes through PR.
@ -392,12 +409,13 @@ python scripts/backfill_sites.py
```
cdk/
app.py # Two stacks: po-ingest + WorkorderIngestStack (region-only env)
app.py # Three stacks: po-ingest + WorkorderIngestStack + procurement-api (region-only env)
common.py # Phase 4: shared plain-function CDK helpers (alarms, Bedrock grant,
# 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
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
# Phase 3, both web_ui functions) -- each email-processor command
@ -408,9 +426,20 @@ lambdas/ # Phase 2: shared Code.from_asset("../lambdas") bundling
shared/ # Phase 3: single-sourced first-party modules, landed FLAT (no
# __init__.py -- bare-name imports) into each bundle by the cp above
ses_auth.py # fail-closed SES sender-auth (INFRA-107) -- one copy, one fix
web_ui_auth.py # fail-closed X-Auth-Token gate + token cache (INFRA-74)
web_ui_auth.py # fail-closed X-Auth-Token gate + token cache (INFRA-74; also gates the API /docs routes)
email_parsing.py # parse_raw_email (WO superset; returns cc unconditionally)
emf.py # generic CloudWatch EMF emitter (dimension-sets pinned once)
api/ # procurement-api Lambda (REST reads over both pipelines' tables)
handler.py # healthcheck early-return + docs-token gate + route dispatch + error mapping
router.py # single route table (spec-drift test pins it to openapi.json)
pagination.py # opaque base64(LastEvaluatedKey) cursor encode/validate (hostile cursor -> 400)
serialization.py # Decimal-safe JSON responses
wo_repo.py # WorkOrders/WorkOrderComments reads (paginated Scan / PK Query)
po_repo.py # purchase-orders/verified-sites reads (no VendorReplies -- dead table)
openapi.json # OpenAPI 3.1 source of truth (paths + outbound `webhooks` section)
docs.html # Redoc shell, SHOC design-system theme; handler inlines bundle/fonts/spec at request time (single token-gated request, no CDN)
redoc.standalone.js # vendored Redoc bundle (redoc 2.5.3, MIT), inlined into /docs; read-only docs, no try-it-out
fonts.css # SHOC fonts (DM Sans/Montserrat/JetBrains Mono, @fontsource latin subsets) as data URIs, inlined into /docs
po/ # PO pipeline Lambdas
email_processor/ # Phase 5: God-handler decomposed into flat siblings (bare-name
# imports; the Phase 0/2/3 `cp <pipeline>/email_processor/*.py`
@ -460,7 +489,7 @@ lambdas/ # Phase 2: shared Code.from_asset("../lambdas") bundling
scripts/
reprocess.py
backfill_sites.py
post-deploy-smoke.sh # CD gate: synchronous healthcheck invoke of both processors, checks FunctionError
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 --
# including a standalone `pytest lambdas/po/email_processor/tests` -- before

8
cdk/app.py
View file

@ -1,6 +1,7 @@
#!/usr/bin/env python3
import aws_cdk as cdk
from po_stack import PoIngestStack
from procurement_api_stack import ProcurementApiStack
from wo_stack import WorkorderIngestStack
app = cdk.App()
@ -19,4 +20,11 @@ WorkorderIngestStack(
env=cdk.Environment(region="us-east-1"),
)
ProcurementApiStack(
app,
"procurement-api",
stack_name="procurement-api",
env=cdk.Environment(region="us-east-1"),
)
app.synth()

364
cdk/procurement_api_stack.py Normal file
View file

@ -0,0 +1,364 @@
"""procurement-api stack: read-only REST API over both pipelines' tables.
Third stack in the app. Serves work orders + comments (WO stack tables) and
purchase orders + verified sites (PO stack tables) to SigV4 callers -- the
primary consumer is the SHOC backend, for which this API replaces the retired
SyncController cross-account DynamoDB scan as the reconciliation/backfill
path. Also hosts the token-gated OpenAPI docs page (/docs, /openapi.json).
Tables are imported by fixed physical name (Table.from_table_name), NOT
passed as cross-stack objects: object passing would synthesize CFN Exports
from the owning stacks and lock them against future changes to the tables.
The one thing name-import does NOT carry is the purchase-orders CMK
association -- see the explicit KMS grant below.
"""
import aws_cdk as cdk
from aws_cdk import (
Duration,
Stack,
aws_apigateway as apigateway,
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_secretsmanager as secretsmanager,
aws_sns as sns,
aws_ssm as ssm,
)
from constructs import Construct
import common
# The only cross-account caller. Grants are to this exact role ARN -- future
# shoc-backend-staging/-prod roles are each a deliberate, individually
# cross-reviewed addition (no wildcard/prefix trust).
SHOC_BACKEND_DEV_ROLE_ARN = "arn:aws:iam::396287094661:role/shoc-backend-dev"
STAGE_NAME = "prod"
class ProcurementApiStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
alarm_topic = sns.Topic.from_topic_arn(
self,
"SiteAlertsTopic",
f"arn:aws:sns:{self.region}:{self.account}:site-alerts",
)
work_orders_table = dynamodb.Table.from_table_name(
self, "WorkOrdersTable", "WorkOrders"
)
comments_table = dynamodb.Table.from_table_name(
self, "CommentsTable", "WorkOrderComments"
)
po_table = dynamodb.Table.from_table_name(
self, "PurchaseOrdersTable", "purchase-orders"
)
verified_sites_table = dynamodb.Table.from_table_name(
self, "VerifiedSitesTable", "verified-sites"
)
web_ui_auth_secret = secretsmanager.Secret.from_secret_name_v2(
self,
"WebUiAuthToken",
"procurement-ingest/web-ui-auth-token",
)
# purchase-orders is SSE-KMS encrypted with the org DynamoDB CMK. A
# name-imported Table has no encryption-key association, so
# grant_read_data alone leaves the reader without kms:Decrypt and every
# purchase-orders read AccessDenies at runtime (the INFRA-104 failure
# class). Import the key from the same SSM parameter po_stack uses and
# grant it explicitly.
dynamodb_cmk = kms.Key.from_key_arn(
self,
"DynamoDbCmk",
ssm.StringParameter.value_for_string_parameter(
self, "/seahaven/dynamodb/cmk-arn"
),
)
# RETAIN log group (INFRA-114). This is the only stateful resource in
# this otherwise-stateless stack, so it carries the fresh-deploy
# rollback trap: if the FIRST create fails after the group exists,
# rollback deletes everything else but retains the group, and the retry
# CREATE then collides on "/aws/lambda/procurement-api already exists".
# Recovery: delete that log group before re-running a failed first
# deploy (same class as the RETAIN-orphan recovery in the deploy-role
# runbook / reference_cfn_deploy_role_gotchas).
api_log_group = common.make_function_log_group(
self, "ProcurementApi", "procurement-api"
)
api_fn = lambda_.Function(
self,
"ProcurementApi",
function_name="procurement-api",
runtime=lambda_.Runtime.PYTHON_3_12,
architecture=lambda_.Architecture.ARM_64,
handler="handler.handler",
code=lambda_.Code.from_asset(
"../lambdas",
exclude=["**/__pycache__/**"],
bundling=cdk.BundlingOptions(
image=lambda_.Runtime.PYTHON_3_12.bundling_image,
command=[
"bash",
"-c",
# Non-recursive glob ships every api/ sibling (the
# allowlist-omission trap from PR #105/PR #2); the spec,
# docs page, and the vendored Redoc bundle ride along
# because the handler serves them from its own package
# dir. web_ui_auth.py must land FLAT beside handler.py
# for the bare import.
"cp api/*.py /asset-output/ && "
"cp api/openapi.json /asset-output/ && "
"cp api/docs.html /asset-output/ && "
"cp api/redoc.standalone.js /asset-output/ && "
"cp api/fonts.css /asset-output/ && "
"cp shared/web_ui_auth.py /asset-output/ && "
"rm -rf /asset-output/__pycache__",
],
),
),
timeout=Duration.seconds(30),
memory_size=256,
log_group=api_log_group,
environment={
"WORK_ORDERS_TABLE": work_orders_table.table_name,
"COMMENTS_TABLE": comments_table.table_name,
"PO_TABLE": po_table.table_name,
"VERIFIED_SITES_TABLE": verified_sites_table.table_name,
"WEB_UI_AUTH_TOKEN_SECRET_ARN": web_ui_auth_secret.secret_arn,
},
)
work_orders_table.grant_read_data(api_fn)
comments_table.grant_read_data(api_fn)
po_table.grant_read_data(api_fn)
verified_sites_table.grant_read_data(api_fn)
web_ui_auth_secret.grant_read(api_fn)
api_fn.add_to_role_policy(
iam.PolicyStatement(
actions=["kms:Decrypt", "kms:DescribeKey"],
resources=[dynamodb_cmk.key_arn],
# Scope the grant to the DynamoDB data path only: the role can
# decrypt purchase-orders items via DynamoDB, never call
# kms:Decrypt directly on arbitrary ciphertext under the shared
# org CMK.
conditions={
"StringEquals": {
"kms:ViaService": f"dynamodb.{self.region}.amazonaws.com"
}
},
)
)
# Resource policy: once a REST API has one, anything not explicitly
# allowed is denied -- so the docs routes need their own Allow or the
# NONE-auth methods would be black-holed. The docs routes are still
# token-gated inside the Lambda (fail-closed), so this is not an
# unauthenticated data path (INFRA-74 posture).
#
# The SHOC data grant enumerates the exact GET resources rather than
# GET/*: adding a RESOURCE must be as reviewable as adding a PRINCIPAL,
# so a future GET route can't silently inherit cross-account reach
# without a policy diff. (Note: this resource policy binds CROSS-account
# callers only; a same-account principal holding execute-api:Invoke is
# authorized by its own identity policy under AWS union semantics -- it
# is NOT constrained here, including on the planned PATCH/POST methods,
# which is why those also rely on the 501 handler + absent write grant,
# not on this policy, until phase 2.)
shoc_data_resources = [
f"execute-api:/{STAGE_NAME}/GET/work-orders",
f"execute-api:/{STAGE_NAME}/GET/work-orders/*",
f"execute-api:/{STAGE_NAME}/GET/purchase-orders",
f"execute-api:/{STAGE_NAME}/GET/purchase-orders/*",
f"execute-api:/{STAGE_NAME}/GET/verified-sites",
f"execute-api:/{STAGE_NAME}/GET/verified-sites/*",
]
api_policy = iam.PolicyDocument(
statements=[
iam.PolicyStatement(
sid="ShocBackendDevDataRead",
principals=[iam.ArnPrincipal(SHOC_BACKEND_DEV_ROLE_ARN)],
actions=["execute-api:Invoke"],
resources=shoc_data_resources,
),
# SECURITY INVARIANT: this AnyPrincipal allow is safe only
# while /docs//openapi.json serve static docs (handler
# enforces the shared token, fail-closed). Widening these
# routes to dynamic data, or enabling access logging (which
# would record the docs ?token= shim), requires a security
# re-review + docs-token rotation.
iam.PolicyStatement(
sid="DocsTokenGatedRoutes",
principals=[iam.AnyPrincipal()],
actions=["execute-api:Invoke"],
resources=[
f"execute-api:/{STAGE_NAME}/GET/docs",
f"execute-api:/{STAGE_NAME}/GET/openapi.json",
],
),
]
)
# OPERATIONAL NOTE: API Gateway serves the resource policy from the
# deployed stage snapshot, and CDK's Deployment hash is computed from
# resources/methods, not the RestApi Policy. A later policy-ONLY change
# (e.g. revoking the SHOC role) will UPDATE the RestApi but keep serving
# the old policy until a new Deployment is forced (any method/resource
# change, or a salted deployment). When tightening this policy, force a
# redeploy and verify the effective policy post-deploy.
api = apigateway.RestApi(
self,
"ProcurementRestApi",
rest_api_name="procurement-api",
description=(
"Read API over procurement-ingest work orders + purchase "
"orders; token-gated OpenAPI docs at /docs"
),
endpoint_types=[apigateway.EndpointType.REGIONAL],
policy=api_policy,
deploy_options=apigateway.StageOptions(
stage_name=STAGE_NAME,
# Bound the blast radius of the unauthenticated /docs routes
# (and the whole API) below the 10k rps account default -- this
# is a low-volume reconciliation/backfill API, not a hot path.
throttling_rate_limit=50,
throttling_burst_limit=100,
),
# No access/execution logging in v1: avoids the account-level API
# Gateway CloudWatch role prerequisite AND keeps the docs ?token=
# query shim out of any log. Rotate the docs token before ever
# enabling access logging here.
cloud_watch_role=False,
)
integration = apigateway.LambdaIntegration(api_fn)
iam_auth = {"authorization_type": apigateway.AuthorizationType.IAM}
work_orders = api.root.add_resource("work-orders")
work_orders.add_method("GET", integration, **iam_auth)
wo_by_id = work_orders.add_resource("{workOrderId}")
wo_by_id.add_method("GET", integration, **iam_auth)
# Phase-2 planned write endpoints: deployed but the handler answers 501
# and the role holds no DynamoDB write grant. The resource policy denies
# these to the cross-account SHOC role (GET-only enumeration above); a
# same-account caller is NOT blocked by the resource policy, so the 501
# handler + absent write grant are the real gate until phase 2 lands the
# deliberate policy + handler + write-grant change with its own review.
wo_by_id.add_method("PATCH", integration, **iam_auth)
wo_comments = wo_by_id.add_resource("comments")
wo_comments.add_method("GET", integration, **iam_auth)
wo_comments.add_method("POST", integration, **iam_auth)
purchase_orders = api.root.add_resource("purchase-orders")
purchase_orders.add_method("GET", integration, **iam_auth)
purchase_orders.add_resource("{poNumber}").add_method(
"GET", integration, **iam_auth
)
verified_sites = api.root.add_resource("verified-sites")
verified_sites.add_method("GET", integration, **iam_auth)
verified_sites.add_resource("{siteCode}").add_method(
"GET", integration, **iam_auth
)
api.root.add_resource("docs").add_method(
"GET",
integration,
authorization_type=apigateway.AuthorizationType.NONE,
)
api.root.add_resource("openapi.json").add_method(
"GET",
integration,
authorization_type=apigateway.AuthorizationType.NONE,
)
# --- Alarms (ALARM-only -> site-alerts, NOT_BREACHING) ---
# common.add_standard_lambda_alarms is NOT used here: its duration
# threshold is a fixed 45000 ms (75% of the processors' 60 s timeout),
# which this function's 30 s timeout can never reach. Same idiom,
# right-sized thresholds.
api_fn.metric_errors(period=Duration.minutes(5), statistic="Sum").create_alarm(
self,
"ProcurementApiErrorsAlarm",
alarm_name="procurement-api-errors",
alarm_description=(
"procurement-api Lambda raised (bundle/init failures; the "
"handler catches request errors, so any signal here is "
"structural)"
),
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))
api_fn.metric_throttles(
period=Duration.minutes(5), statistic="Sum"
).create_alarm(
self,
"ProcurementApiThrottlesAlarm",
alarm_name="procurement-api-throttles",
alarm_description="procurement-api Lambda throttled",
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))
api_fn.metric_duration(
period=Duration.minutes(5), statistic="p99"
).create_alarm(
self,
"ProcurementApiDurationAlarm",
alarm_name="procurement-api-duration",
alarm_description=(
"procurement-api p99 duration >= 22.5s (75% of the 30s "
"timeout; scans degrading toward timeout)"
),
threshold=22500,
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))
# Gateway-side 5xx: catches what the Lambda's own Errors metric can't
# (the handler returns clean 500s; integration faults surface here).
# No 4XX alarm -- 401/403/404 are expected traffic.
cloudwatch.Metric(
namespace="AWS/ApiGateway",
metric_name="5XXError",
dimensions_map={"ApiName": "procurement-api", "Stage": STAGE_NAME},
period=Duration.minutes(5),
statistic="Sum",
).create_alarm(
self,
"ProcurementApi5xxAlarm",
alarm_name="procurement-api-5xx",
alarm_description="procurement-api gateway 5XX responses",
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(
self,
"ApiEndpointUrl",
value=api.url,
description="procurement-api invoke URL (stage prod)",
)
cdk.CfnOutput(
self,
"ProcurementApiFunctionArn",
value=api_fn.function_arn,
description="ARN of the procurement-api Lambda",
)

13
infra/deploy-role/README.md
View file

@ -10,19 +10,20 @@ stays untouched until decommission as the emergency mgmt deploy path.
| File | Purpose |
|---|---|
| `trust-policy.json` | OIDC trust: `repo:Sea-Haven-Industries/procurement-ingest:ref:refs/heads/main` only |
| `permissions-policy.json` | `sts:AssumeRole` on the four `cdk-hnb659fds-*` bootstrap roles, `cloudformation:DescribeStacks` scoped to this repo's stacks + `CDKToolkit` (cd-cdk health check), and `lambda:InvokeFunction` on exactly the two email-processor function ARNs (post-deploy smoke gate) |
| `permissions-policy.json` | `sts:AssumeRole` on the four `cdk-hnb659fds-*` bootstrap roles, `cloudformation:DescribeStacks` scoped to this repo's stacks + `CDKToolkit` (cd-cdk health check), and `lambda:InvokeFunction` on exactly the three smoke-gated function ARNs (post-deploy smoke gate) |
| `create-deploy-role.sh` | Idempotent create-or-update from the two JSON files, profile `seahaven-prod` |
> **Maintenance note:** `DescribeStacks` is scoped to `stack/po-ingest/*`, `stack/WorkorderIngestStack/*`, and `stack/CDKToolkit/*`. If a third stack is ever added to this CDK app, add its ARN pattern here and re-run the review-then-apply flow — otherwise the cd-cdk health check on the new stack will `AccessDenied`.
> **Maintenance note:** `DescribeStacks` is scoped to `stack/po-ingest/*`, `stack/WorkorderIngestStack/*`, `stack/procurement-api/*` (added with the procurement-api stack), and `stack/CDKToolkit/*`. If another stack is ever added to this CDK app, add its ARN pattern here and re-run the review-then-apply flow — otherwise the cd-cdk health check on the new stack will `AccessDenied`.
## Why the SmokeInvokeLambda statement exists
`deploy.yaml` runs `scripts/post-deploy-smoke.sh` under the deploy role's own
session, not the assumed `cdk-*` roles. Without `lambda:InvokeFunction` on the
two function ARNs the smoke gate hits AccessDenied and every deploy fails
closed. The mgmt-era grant was applied out-of-band and undocumented; keeping
it in these reviewed artifacts closes that gap. Scope it to exactly the two
ARNs, never `Resource: "*"`.
smoke-gated function ARNs (the two email processors + `procurement-api`) the
smoke gate hits AccessDenied and every deploy fails closed. The mgmt-era grant
was applied out-of-band and undocumented; keeping it in these reviewed
artifacts closes that gap. Scope it to exactly the named ARNs, never
`Resource: "*"`.
## Change process

4
infra/deploy-role/permissions-policy.json
View file

@ -19,6 +19,7 @@
"Resource": [
"arn:aws:cloudformation:us-east-1:011934824531:stack/po-ingest/*",
"arn:aws:cloudformation:us-east-1:011934824531:stack/WorkorderIngestStack/*",
"arn:aws:cloudformation:us-east-1:011934824531:stack/procurement-api/*",
"arn:aws:cloudformation:us-east-1:011934824531:stack/CDKToolkit/*"
],
"Condition": {
@ -33,7 +34,8 @@
"Action": "lambda:InvokeFunction",
"Resource": [
"arn:aws:lambda:us-east-1:011934824531:function:po-email-processor",
"arn:aws:lambda:us-east-1:011934824531:function:workorder-email-processor"
"arn:aws:lambda:us-east-1:011934824531:function:workorder-email-processor",
"arn:aws:lambda:us-east-1:011934824531:function:procurement-api"
]
}
]

162
lambdas/api/docs.html Normal file
View file

@ -0,0 +1,162 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Procurement Ingest API</title>
<link rel="icon" href="data:image/svg+xml,%3Csvg%20xmlns='http://www.w3.org/2000/svg'%20viewBox='0%200%2016%2016'%3E%3Crect%20width='16'%20height='16'%20rx='3'%20fill='%231c75bc'/%3E%3Ctext%20x='8'%20y='12'%20font-size='10'%20font-family='sans-serif'%20font-weight='700'%20text-anchor='middle'%20fill='white'%3ES%3C/text%3E%3C/svg%3E">
<!--
Redoc (redoc.standalone.js, MIT), vendored offline and inlined server-side
so /docs is a single token-gated request with no CDN or follow-up asset
fetch. The bundle, the SHOC design-system fonts (fonts.css, data-URI
@font-face), and the OpenAPI spec are substituted for the placeholders
below by lambdas/api/handler.py::_docs_shell / _render_docs_html. Redoc is
read-only by design -- there is no try-it-out to disable; live calls go
through Postman (Authorization type "AWS Signature") since data routes
require SigV4.
Theme = the SHOC design-system token set (the Sea Haven visual standard):
Montserrat headings / DM Sans body / JetBrains Mono code, primary #1c75bc,
navy #262262, page background #f9fafb, 244px sidebar. Fonts are vendored --
nothing here may fetch from Google Fonts.
-->
<style>__FONTS_CSS__</style>
<style>
html, body { margin: 0; padding: 0; background: #f9fafb; }
/* SHOC shell topbar: 64px, gradient header token. Redoc has no topbar of
its own; scrollYOffset below keeps its sticky sidebar/scroll accounting
clear of this fixed bar. Gradient runs bright-to-dark (reversed from the
SHOC shell) so its dark end (#1b1f52) sits flush over the right panel,
which uses the same color -- one continuous blue family, no seam. */
#topbar {
position: fixed;
top: 0;
left: 0;
right: 0;
height: 64px;
z-index: 10;
display: flex;
align-items: center;
padding: 0 24px;
box-sizing: border-box;
background: linear-gradient(90deg, #1c75bc 0%, #1c4f8f 45%, #1b1f52 100%);
color: #ffffff;
font-family: "Montserrat", system-ui, -apple-system, "Segoe UI", Helvetica, Arial, sans-serif;
}
#topbar .brand {
font-size: 16px;
font-weight: 600;
letter-spacing: 0.02em;
}
#topbar .page {
margin-left: 16px;
font-family: "DM Sans", system-ui, -apple-system, "Segoe UI", Helvetica, Arial, sans-serif;
font-size: 13px;
font-weight: 400;
opacity: 0.85;
}
#samples-toggle {
margin-left: auto;
padding: 5px 14px;
background: transparent;
color: #ffffff;
border: 1px solid rgba(255, 255, 255, 0.45);
border-radius: 6px;
font-family: "DM Sans", system-ui, -apple-system, "Segoe UI", Helvetica, Arial, sans-serif;
font-size: 12px;
cursor: pointer;
}
#samples-toggle:hover { border-color: #ffffff; background: rgba(255, 255, 255, 0.1); }
#redoc { padding-top: 64px; }
/* Collapsible samples column: Redoc CE has no built-in panel toggle, so
the topbar button flips .samples-collapsed on <html>, hiding the
right-panel divs (same bundle-pinned classes as the gradient below)
and letting each section's content half take the full width. If the
pinned classes stop matching after a Redoc bump the toggle goes inert
-- cosmetic only. Choice persists in localStorage. */
html.samples-collapsed .sc-iGgWBj.sc-gsFSXq,
html.samples-collapsed div.sc-dExYaf { display: none; }
html.samples-collapsed [data-section-id] > div:first-child { width: 100%; }
/* Right-panel gradient (#1b3d79 -> #1b1f52, continuing the topbar blend).
Redoc's theme only accepts solid colors (it derives shades from
rightPanel.backgroundColor), so the solid #1b1f52 stays in the theme as
the fallback and this override layers the gradient on top.
BUNDLE-PINNED SELECTORS: .sc-iGgWBj.sc-gsFSXq are the styled-components
classes of the per-section right-panel divs (plus .sc-dExYaf, the
full-height background stub) generated by the vendored
redoc.standalone.js 2.5.3. They are deterministic for this exact bundle
but WILL change on any Redoc bump -- re-derive them then (headless probe:
find elements whose computed background is the rightPanel color). If
they stop matching, the panel silently falls back to solid #1b1f52 --
cosmetic only, nothing breaks. Every stripe shares the same horizontal
geometry, so the per-section gradients read as one continuous column. */
.sc-iGgWBj.sc-gsFSXq,
div.sc-dExYaf {
background-image: linear-gradient(90deg, #1b3d79 0%, #1b3068 45%, #1b1f52 100%);
}
</style>
</head>
<body>
<div id="topbar">
<span class="brand">Sea Haven Industries</span>
<button id="samples-toggle" type="button" aria-pressed="false">Hide samples</button>
<span class="page">Procurement Ingest API</span>
</div>
<div id="redoc"></div>
<script>__REDOC_JS__</script>
<script>
Redoc.init(
__OPENAPI_SPEC_JSON__,
{
sortRequiredPropsFirst: true,
expandResponses: "200",
scrollYOffset: 64,
theme: {
colors: {
primary: { main: "#1c75bc" }
},
typography: {
fontFamily: '"DM Sans", system-ui, -apple-system, "Segoe UI", Helvetica, Arial, sans-serif',
fontWeightBold: "600",
headings: {
fontFamily: '"Montserrat", system-ui, -apple-system, "Segoe UI", Helvetica, Arial, sans-serif',
fontWeight: "600"
},
code: {
fontFamily: '"JetBrains Mono", ui-monospace, SFMono-Regular, Menlo, monospace'
}
},
sidebar: {
width: "244px",
backgroundColor: "#f9fafb",
textColor: "#262262"
},
rightPanel: {
backgroundColor: "#1b1f52"
},
fab: {
backgroundColor: "#1c75bc"
}
}
},
document.getElementById("redoc")
);
(function () {
var KEY = "procurement-docs-samples-collapsed";
var btn = document.getElementById("samples-toggle");
function apply(collapsed) {
document.documentElement.classList.toggle("samples-collapsed", collapsed);
btn.textContent = collapsed ? "Show samples" : "Hide samples";
btn.setAttribute("aria-pressed", String(collapsed));
}
var initial = false;
try { initial = localStorage.getItem(KEY) === "1"; } catch (e) {}
apply(initial);
btn.addEventListener("click", function () {
var collapsed = !document.documentElement.classList.contains("samples-collapsed");
try { localStorage.setItem(KEY, collapsed ? "1" : "0"); } catch (e) {}
apply(collapsed);
});
})();
</script>
</body>
</html>

37
lambdas/api/fonts.css Normal file
View file

File diff suppressed because one or more lines are too long

235
lambdas/api/handler.py Normal file
View file

@ -0,0 +1,235 @@
"""Procurement API Lambda (API Gateway REST proxy integration).
Auth is split by route class and enforced at two layers:
- Data routes: AWS_IAM at the gateway (SigV4; cross-account callers allowed by
the API resource policy). The handler does NOT re-check a token there --
authorization is API Gateway's job on those routes.
- Docs routes (/docs, /openapi.json): reachable at the gateway (auth NONE +
resource-policy carve-out) but the handler fails closed on the shared
header token via web_ui_auth (same secret + constant-time compare as the
web UIs), so they are never an unauthenticated data path (INFRA-74).
"""
import logging
from pathlib import Path
import po_repo
import wo_repo
from botocore.exceptions import ClientError
from pagination import BadCursor, clamp_limit
from router import DATA_ROUTES, DOCS_ROUTES, PLANNED_ROUTES
from serialization import error_response, json_response
from web_ui_auth import is_authenticated
logger = logging.getLogger()
logger.setLevel(logging.INFO)
_MODULE_DIR = Path(__file__).resolve().parent
_SPEC_PATH = _MODULE_DIR / "openapi.json"
_DOCS_PATH = _MODULE_DIR / "docs.html"
_REDOC_JS_PATH = _MODULE_DIR / "redoc.standalone.js"
_FONTS_CSS_PATH = _MODULE_DIR / "fonts.css"
# The docs page template carries these placeholders, filled server-side so
# /docs is a single token-gated request (a browser can't attach the auth
# header to a follow-up asset fetch) with the vendored Redoc bundle, the
# design-system fonts, and the spec inlined, no CDN.
_SPEC_PLACEHOLDER = "__OPENAPI_SPEC_JSON__"
_JS_PLACEHOLDER = "__REDOC_JS__"
_FONTS_PLACEHOLDER = "__FONTS_CSS__"
_spec_cache = None
_docs_shell_cache = None
# The server URL committed in openapi.json; swapped for the live invoke URL
# (derived from the request) when the spec is served, so the docs page shows a
# correct, current endpoint even if the RestApi is recreated.
_COMMITTED_SERVER_URL = "https://mvul1efda2.execute-api.us-east-1.amazonaws.com/prod"
_REPO_FUNCS = {
"list_work_orders": wo_repo.list_work_orders,
"get_work_order": wo_repo.get_work_order,
"list_comments": wo_repo.list_comments,
"list_purchase_orders": po_repo.list_purchase_orders,
"get_purchase_order": po_repo.get_purchase_order,
"list_verified_sites": po_repo.list_verified_sites,
"get_verified_site": po_repo.get_verified_site,
}
_PATH_PARAM_BY_RESOURCE = {
"/work-orders/{workOrderId}": "workOrderId",
"/work-orders/{workOrderId}/comments": "workOrderId",
"/purchase-orders/{poNumber}": "poNumber",
"/verified-sites/{siteCode}": "siteCode",
}
# Headers on the docs responses: the token rides in the ?token= query shim,
# so keep the token-keyed URL and page out of shared/browser caches and out of
# any Referer sent to a followed link.
_DOCS_SECURITY_HEADERS = {
"Cache-Control": "no-store",
"Referrer-Policy": "no-referrer",
}
def _load_spec() -> str:
"""The committed spec text (cached)."""
global _spec_cache
if _spec_cache is None:
_spec_cache = _SPEC_PATH.read_text(encoding="utf-8")
return _spec_cache
def _base_url(event: dict) -> str | None:
rc = event.get("requestContext") or {}
domain, stage = rc.get("domainName"), rc.get("stage")
return f"https://{domain}/{stage}" if domain and stage else None
def _spec_for_request(event: dict) -> str:
"""Committed spec with servers[0].url set to the live invoke URL when the
request context provides it (falls back to the committed literal)."""
raw = _load_spec()
base = _base_url(event)
return raw.replace(_COMMITTED_SERVER_URL, base) if base else raw
def _docs_shell() -> str:
"""The Redoc page with the vendored bundle inlined; the SPEC placeholder
is left intact so the per-request server-injected spec splices in cheaply
(the heavy ~1.1MB bundle is assembled once and cached).
Each blob is spliced into a <style>/<script> block, where the HTML parser
ends the element at the first literal "</style"/"</script" regardless of
quoting, so guard against a breakout a future asset update could add (the
pinned assets have none today): JS "</script" -> "<\\/script" (equivalent
inside a JS string/regex), CSS "</style" -> "<\\/style".
"""
global _docs_shell_cache
if _docs_shell_cache is None:
js = _REDOC_JS_PATH.read_text(encoding="utf-8").replace(
"</script", "<\\/script"
)
fonts = _FONTS_CSS_PATH.read_text(encoding="utf-8").replace(
"</style", "<\\/style"
)
_docs_shell_cache = (
_DOCS_PATH.read_text(encoding="utf-8")
.replace(_JS_PLACEHOLDER, js)
.replace(_FONTS_PLACEHOLDER, fonts)
)
return _docs_shell_cache
def _render_docs_html(event: dict) -> str:
# "<" -> < so a spec string can never close the <script> block (same
# value to JSON.parse); the spec object is inlined into Redoc.init.
spec = _spec_for_request(event).replace("<", "\\u003c")
return _docs_shell().replace(_SPEC_PLACEHOLDER, spec)
def _with_token_shim(event: dict) -> dict:
"""Copy a ?token= query parameter into an x-auth-token header.
Browsers can't set headers on plain navigation, so /docs accepts the
shared token as a query parameter too. The constant-time compare still
happens inside web_ui_auth -- this only synthesizes the header on a
shallow copy. Acceptable only while API Gateway access logging stays off
(nothing at the gateway records the query string); rotate the token if
access logging is ever enabled.
"""
qs = event.get("queryStringParameters") or {}
token = qs.get("token")
if not token:
return event
shimmed = dict(event)
headers = dict(event.get("headers") or {})
headers["x-auth-token"] = token
shimmed["headers"] = headers
return shimmed
def _docs_response(resource: str, event: dict) -> dict:
# SECURITY INVARIANT: these routes serve ONLY the committed spec and the
# static docs page -- never table data. The gateway resource policy allows
# Principal "*" on exactly these two GETs on the strength of that; serving
# anything dynamic here requires a resource-policy + security re-review.
if not is_authenticated(_with_token_shim(event)):
return error_response(401, "unauthorized")
if resource == "/openapi.json":
return {
"statusCode": 200,
"headers": {"Content-Type": "application/json", **_DOCS_SECURITY_HEADERS},
"body": _spec_for_request(event),
}
return {
"statusCode": 200,
"headers": {
"Content-Type": "text/html; charset=utf-8",
**_DOCS_SECURITY_HEADERS,
},
"body": _render_docs_html(event),
}
def _data_response(route_key: tuple, event: dict) -> dict:
method, resource = route_key
func = _REPO_FUNCS[DATA_ROUTES[route_key]]
qs = event.get("queryStringParameters") or {}
path_params = event.get("pathParameters") or {}
if DATA_ROUTES[route_key].startswith("list_"):
limit = clamp_limit(qs.get("limit"))
cursor = qs.get("cursor")
if resource in _PATH_PARAM_BY_RESOURCE:
entity_id = path_params.get(_PATH_PARAM_BY_RESOURCE[resource], "")
items, next_cursor = func(entity_id, limit, cursor)
else:
items, next_cursor = func(limit, cursor)
return json_response(200, {"items": items, "next_cursor": next_cursor})
entity_id = path_params.get(_PATH_PARAM_BY_RESOURCE[resource], "")
item = func(entity_id)
if item is None:
return error_response(404, "not found")
return json_response(200, item)
def _dispatch(route_key: tuple, event: dict) -> dict:
if route_key in PLANNED_ROUTES:
return error_response(501, "planned endpoint - not implemented (phase 2)")
if route_key in DOCS_ROUTES:
return _docs_response(route_key[1], event)
if route_key in DATA_ROUTES:
return _data_response(route_key, event)
return error_response(404, "not found")
def handler(event, context):
# Deploy-guard healthcheck: a direct-invoke {"healthcheck": true} probe
# returns before any routing/auth so the post-deploy smoke gate can verify
# the bundle imports and the runtime boots.
if isinstance(event, dict) and event.get("healthcheck") is True:
return {"healthcheck": "ok"}
method = (event.get("httpMethod") or "").upper()
resource = event.get("resource") or ""
try:
return _dispatch((method, resource), event)
except BadCursor as exc:
return error_response(400, str(exc))
except ClientError as exc:
# A client-supplied cursor that survives validation but is still
# inconsistent at the data layer makes DynamoDB raise
# ValidationException; map it to 400, not 500, so a crafted cursor
# can't drive the 5xx alarm. Any other AWS error is a real 500.
if exc.response.get("Error", {}).get("Code") == "ValidationException":
return error_response(400, "cursor is not valid")
logger.exception("AWS error serving %s %s", method, resource)
return error_response(500, "internal error")
except Exception:
# A raised exception would surface as an opaque 502 from the proxy
# integration; return a clean 500 instead. The API Gateway 5XX alarm
# pages on these; the exception (never the request token) is logged.
logger.exception("Unhandled error serving %s %s", method, resource)
return error_response(500, "internal error")

1097
lambdas/api/openapi.json Normal file
View file

File diff suppressed because it is too large Load diff

62
lambdas/api/pagination.py Normal file
View file

@ -0,0 +1,62 @@
"""Opaque cursor pagination over DynamoDB ``LastEvaluatedKey``.
The cursor is base64url(JSON(LastEvaluatedKey)). It is untrusted client input:
``decode_cursor`` validates shape strictly (a flat dict whose keys are exactly
a subset of the table's key attributes and whose values are non-empty strings)
and raises ``BadCursor`` -- mapped to HTTP 400 by the handler -- on anything
else, so a malformed or tampered cursor can never reach DynamoDB as an
arbitrary ``ExclusiveStartKey`` or surface as a 500.
"""
import base64
import binascii
import json
DEFAULT_LIMIT = 100
MAX_LIMIT = 500
_MAX_CURSOR_CHARS = 2048
class BadCursor(ValueError):
pass
def clamp_limit(raw) -> int:
if raw is None or raw == "":
return DEFAULT_LIMIT
try:
value = int(raw)
except (TypeError, ValueError):
raise BadCursor("limit must be an integer") from None
return max(1, min(MAX_LIMIT, value))
def encode_cursor(last_evaluated_key: dict) -> str:
raw = json.dumps(last_evaluated_key, default=str, sort_keys=True)
return base64.urlsafe_b64encode(raw.encode()).decode()
def decode_cursor(cursor: str, key_attrs: frozenset) -> dict:
"""Decode a cursor and require it to be EXACTLY the table's key attributes.
``key_attrs`` is the full key schema of the operation the cursor feeds
(e.g. ``{work_order_id, comment_id}`` for the comments Query). An exact
match -- not a subset -- is required: a partial composite key or a cursor
minted for a different endpoint would otherwise pass a subset check, reach
DynamoDB as an incomplete/inconsistent ``ExclusiveStartKey``, and raise a
ValidationException that surfaces as a 500 (and pages the 5xx alarm). Here
it fails closed as a 400 instead. Callers additionally pin the
partition-key value to the request path (see ``wo_repo.list_comments``).
"""
if len(cursor) > _MAX_CURSOR_CHARS:
raise BadCursor("cursor too long")
try:
decoded = json.loads(base64.urlsafe_b64decode(cursor.encode()))
except (binascii.Error, UnicodeDecodeError, json.JSONDecodeError, ValueError):
raise BadCursor("cursor is not valid") from None
if not isinstance(decoded, dict) or set(decoded) != key_attrs:
raise BadCursor("cursor is not valid")
for value in decoded.values():
if not isinstance(value, str) or not value:
raise BadCursor("cursor is not valid")
return decoded

53
lambdas/api/po_repo.py Normal file
View file

@ -0,0 +1,53 @@
"""Read access to the purchase-order tables for the procurement API.
Same unordered cursor-paginated Scan pattern as wo_repo (no GSIs; listing
serves reconciliation, not ranked queries). VendorReplies is deliberately
absent: the table is dead (its writer was deleted) and must not gain new
consumers.
"""
import os
import boto3
from pagination import decode_cursor, encode_cursor
dynamodb = boto3.resource("dynamodb")
PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders")
VERIFIED_SITES_TABLE = os.environ.get("VERIFIED_SITES_TABLE", "verified-sites")
PO_KEY_ATTRS = frozenset({"po_number"})
SITE_KEY_ATTRS = frozenset({"siteCode"})
def _page(response) -> tuple[list, str | None]:
items = response.get("Items", [])
lek = response.get("LastEvaluatedKey")
return items, (encode_cursor(lek) if lek else None)
def list_purchase_orders(limit: int, cursor: str | None) -> tuple[list, str | None]:
table = dynamodb.Table(PO_TABLE)
kwargs = {"Limit": limit}
if cursor:
kwargs["ExclusiveStartKey"] = decode_cursor(cursor, PO_KEY_ATTRS)
return _page(table.scan(**kwargs))
def get_purchase_order(po_number: str) -> dict | None:
table = dynamodb.Table(PO_TABLE)
return table.get_item(Key={"po_number": po_number}).get("Item")
def list_verified_sites(limit: int, cursor: str | None) -> tuple[list, str | None]:
table = dynamodb.Table(VERIFIED_SITES_TABLE)
kwargs = {"Limit": limit}
if cursor:
kwargs["ExclusiveStartKey"] = decode_cursor(cursor, SITE_KEY_ATTRS)
return _page(table.scan(**kwargs))
def get_verified_site(site_code: str) -> dict | None:
table = dynamodb.Table(VERIFIED_SITES_TABLE)
return table.get_item(Key={"siteCode": site_code}).get("Item")

1838
lambdas/api/redoc.standalone.js Normal file
View file

File diff suppressed because one or more lines are too long

1
lambdas/api/requirements.txt Normal file
View file

@ -0,0 +1 @@
# Dependabot anchor only. The procurement-api Lambda is stdlib+boto3 (provided by the runtime); nothing is pip-installed into the bundle.

33
lambdas/api/router.py Normal file
View file

@ -0,0 +1,33 @@
"""Route tables for the procurement API.
Single source of truth for what the API implements: the spec-drift test
(tests/test_api_spec_drift.py) asserts these tables match lambdas/api/
openapi.json exactly, so an endpoint can't be added, removed, or renamed on
one side without failing CI. Keys are (httpMethod, resource) as API Gateway's
Lambda-proxy event presents them (resource = the templated path).
DATA_ROUTES values are handler-module attribute names resolved by handler.py.
PLANNED_ROUTES are phase-2 write endpoints: present in the spec (x-planned)
and at the gateway, but the handler answers 501 until they ship with their
own IAM diff + cross-family review.
"""
DATA_ROUTES = {
("GET", "/work-orders"): "list_work_orders",
("GET", "/work-orders/{workOrderId}"): "get_work_order",
("GET", "/work-orders/{workOrderId}/comments"): "list_comments",
("GET", "/purchase-orders"): "list_purchase_orders",
("GET", "/purchase-orders/{poNumber}"): "get_purchase_order",
("GET", "/verified-sites"): "list_verified_sites",
("GET", "/verified-sites/{siteCode}"): "get_verified_site",
}
DOCS_ROUTES = {
("GET", "/docs"),
("GET", "/openapi.json"),
}
PLANNED_ROUTES = {
("POST", "/work-orders/{workOrderId}/comments"),
("PATCH", "/work-orders/{workOrderId}"),
}

43
lambdas/api/serialization.py Normal file
View file

@ -0,0 +1,43 @@
"""JSON response helpers for the procurement API.
DynamoDB items come back through boto3 with numeric attributes as
``decimal.Decimal`` (e.g. purchase-orders ``total_amount``/``line_items``
prices), which ``json.dumps`` rejects. The encoder below renders integral
Decimals as JSON integers (exact) and non-integral Decimals as floats. For the
values this API serves -- currency amounts at PO magnitudes, well under ~15
significant digits -- the shortest-float repr round-trips the original digits
exactly. A Decimal carrying more precision than a float can hold (DynamoDB
Number supports 38 digits) would be truncated silently; the pipeline never
writes such values, but if that ever changes, switch non-integral Decimals to
``str(o)`` and update the OpenAPI number types.
"""
import decimal
import json
class _DecimalSafeEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, decimal.Decimal):
if o == o.to_integral_value():
return int(o)
return float(o)
if isinstance(o, set):
return sorted(o)
return super().default(o)
def dumps(payload) -> str:
return json.dumps(payload, cls=_DecimalSafeEncoder)
def json_response(status_code: int, payload) -> dict:
return {
"statusCode": status_code,
"headers": {"Content-Type": "application/json"},
"body": dumps(payload),
}
def error_response(status_code: int, message: str) -> dict:
return json_response(status_code, {"error": message})

62
lambdas/api/wo_repo.py Normal file
View file

@ -0,0 +1,62 @@
"""Read access to the work-order tables for the procurement API.
Listing is a cursor-paginated Scan -- deliberately: the WO tables carry no
GSIs (removed 2026-06-03 for zero reads), and the API's list consumers
(SHOC reconciliation/backfill) walk the full table anyway, which is exactly
the access pattern SHOC's retired SyncController used. Listings are therefore
UNORDERED across pages; per-work-order comment listing is a cheap Query on
the partition key.
"""
import os
import boto3
from boto3.dynamodb.conditions import Key
from pagination import BadCursor, decode_cursor, encode_cursor
dynamodb = boto3.resource("dynamodb")
WORK_ORDERS_TABLE = os.environ.get("WORK_ORDERS_TABLE", "WorkOrders")
COMMENTS_TABLE = os.environ.get("COMMENTS_TABLE", "WorkOrderComments")
WO_KEY_ATTRS = frozenset({"work_order_id"})
COMMENT_KEY_ATTRS = frozenset({"work_order_id", "comment_id"})
def _page(response) -> tuple[list, str | None]:
items = response.get("Items", [])
lek = response.get("LastEvaluatedKey")
return items, (encode_cursor(lek) if lek else None)
def list_work_orders(limit: int, cursor: str | None) -> tuple[list, str | None]:
table = dynamodb.Table(WORK_ORDERS_TABLE)
kwargs = {"Limit": limit}
if cursor:
kwargs["ExclusiveStartKey"] = decode_cursor(cursor, WO_KEY_ATTRS)
return _page(table.scan(**kwargs))
def get_work_order(work_order_id: str) -> dict | None:
table = dynamodb.Table(WORK_ORDERS_TABLE)
return table.get_item(Key={"work_order_id": work_order_id}).get("Item")
def list_comments(
work_order_id: str, limit: int, cursor: str | None
) -> tuple[list, str | None]:
table = dynamodb.Table(COMMENTS_TABLE)
kwargs = {
"KeyConditionExpression": Key("work_order_id").eq(work_order_id),
"Limit": limit,
}
if cursor:
start_key = decode_cursor(cursor, COMMENT_KEY_ATTRS)
# Pin the cursor's partition to the path entity: a cursor minted for
# WO A must never resume WO B's Query (DynamoDB would reject the
# partition mismatch as a 500; reject it as a 400 here instead).
if start_key["work_order_id"] != work_order_id:
raise BadCursor("cursor is not valid")
kwargs["ExclusiveStartKey"] = start_key
return _page(table.query(**kwargs))

6
lambdas/shared/web_ui_auth.py
View file

@ -74,4 +74,8 @@ def is_authenticated(event: dict) -> bool:
presented = auth[7:].strip()
if not presented:
return False
return hmac.compare_digest(presented, token)
# Compare as bytes: hmac.compare_digest raises TypeError on non-ASCII str
# operands, which a crafted token (?token=%C3%A9 or a non-ASCII header)
# would otherwise turn into an uncaught 500. Bytes always compare in
# constant time, so a non-matching token fails closed (401) instead.
return hmac.compare_digest(presented.encode("utf-8"), token.encode("utf-8"))

27
package-lock.json generated Normal file
View file

@ -0,0 +1,27 @@
{
"name": "procurement-ingest",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"devDependencies": {
"@redocly/cli": "^2.40.0"
}
},
"node_modules/@redocly/cli": {
"version": "2.40.0",
"resolved": "https://registry.npmjs.org/@redocly/cli/-/cli-2.40.0.tgz",
"integrity": "sha512-1uQ4GeNjhApy9EtypZgp70ZN5GC2JFfst3UkNEXSqkXgVIPGdEAnlz5Xwgax/4cEUGOvaZoM3X25iSQcqbplFg==",
"dev": true,
"license": "MIT",
"bin": {
"openapi": "bin/cli.js",
"redocly": "bin/cli.js"
},
"engines": {
"node": ">=22.12.0 || >=20.19.0 <21.0.0",
"npm": ">=10"
}
}
}
}

9
package.json Normal file
View file

@ -0,0 +1,9 @@
{
"scripts": {
"lint:api": "redocly lint lambdas/api/openapi.json --config=redocly.yaml",
"docs:preview": ".venv/bin/python scripts/preview_docs.py"
},
"devDependencies": {
"@redocly/cli": "^2.40.0"
}
}

1
pytest.ini
View file

@ -12,6 +12,7 @@ addopts =
--cov=lambdas/po/web_ui
--cov=lambdas/wo/web_ui
--cov=lambdas/po/site_extractor
--cov=lambdas/api
--cov=lambdas/shared
--cov-report=term-missing
--cov-fail-under=80

122
redocly.yaml Normal file
View file

@ -0,0 +1,122 @@
extends:
- recommended
rules:
# Proprietary internal license -- no SPDX identifier or public URL exists.
info-license-strict: off
rule/info-title-api:
subject:
type: Info
property: title
assertions:
pattern: /.*API.*/
rule/info-description:
subject:
type: Info
property: description
assertions:
defined: true
operation-4xx-response: error
# Off: the live contract is {"error": string} as plain application/json
# (serialization.py error_response). Adopting RFC 7807 would be a runtime
# + SHOC-contract change, decided against 2026-07-24.
operation-4xx-problem-details-rfc7807: off
operation-operationId: error
rule/operationId-casing:
subject:
type: Operation
property: operationId
assertions:
casing: kebab-case
rule/operationId-prefix:
subject:
type: Operation
property: operationId
assertions:
pattern: /^GET|PUT|POST|DELETE|OPTIONS|HEAD|PATCH|TRACE/i
rule/operation-summary-period:
subject:
type: Operation
property: summary
assertions:
pattern: /[^.]$/
path-not-include-query: error
# No parameter-casing rule: path parameter names (workOrderId, poNumber,
# siteCode) are camelCase by contract -- they are baked into the API Gateway
# resource paths and read by the handler's pathParameters lookup.
rule/params-must-include-examples:
severity: error
subject:
type: Parameter
assertions:
requireAny:
- example
- examples
no-http-verbs-in-paths: error
no-ambiguous-paths: error
path-segment-plural:
severity: error
exceptions:
- docs
- openapi.json
paths-kebab-case: error
no-invalid-schema-examples: error
# No schema-properties casing rule: property names mirror the DynamoDB
# items and the shipped SHOC webhook contract -- snake_case for WO/PO
# tables, camelCase for verified-sites (legacy, issue #24). Not lintable
# to one casing without a contract break.
# Error bodies must carry the top-level "error" field (the Error schema).
# 403 is exempt: it is emitted by API Gateway's SigV4 layer with AWS's
# {"message"} shape, not by the Lambda.
response-contains-property:
severity: error
names:
'400':
- error
'401':
- error
'404':
- error
'501':
- error
request-mime-type:
severity: error
allowedValues:
- application/json
response-mime-type:
severity: error
allowedValues:
- application/json
- text/html
no-server-example.com: error
rule/no-server-localhost:
subject:
type: Server
property: url
assertions:
notPattern: /(localhost|127.0.0.1)
operation-singular-tag: error
operation-tag-defined: error
rule/tag-description:
subject:
type: Tag
property: description
assertions:
defined: true
rule/description-capitalization:
subject:
type: any
property: description
assertions:
pattern: /^([A-Z]|true|seahaven-prod)/
rule/description-punctuation:
subject:
type: any
property: description
assertions:
pattern: /(\.|server)$/
rule/avoid-words-in-descriptions:
subject:
type: any
property: description
assertions:
notPattern: /(simply|easy|easily|just|obviously|notethat)/i

2
scripts/post-deploy-smoke.sh
View file

@ -14,7 +14,7 @@
set -euo pipefail
REGION="us-east-1"
FUNCTION_NAMES=("po-email-processor" "workorder-email-processor")
FUNCTION_NAMES=("po-email-processor" "workorder-email-processor" "procurement-api")
HEALTHCHECK_PAYLOAD='{"healthcheck": true}'
work_dir="$(mktemp -d)"

34
scripts/preview_docs.py Normal file
View file

@ -0,0 +1,34 @@
"""Render the /docs page locally and open it in the default browser.
Goes through the real handler code path (vendored Redoc bundle + fonts
inlined, breakout guards applied, spec spliced), so the preview is
byte-identical to what the Lambda serves, minus the token gate. No AWS
credentials or network needed; without a request context the spec keeps
its committed fallback server URL.
Run from the repo root: `npm run docs:preview` (or
`.venv/bin/python scripts/preview_docs.py`).
"""
import sys
import tempfile
import webbrowser
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT))
from tests.support import load_lambda_module # noqa: E402
def main() -> None:
handler = load_lambda_module("api", "handler")
html = handler._render_docs_html({})
out = Path(tempfile.gettempdir()) / "procurement-docs-preview.html"
out.write_text(html, encoding="utf-8")
print(f"wrote {out} ({len(html)} bytes)")
webbrowser.open(out.as_uri())
if __name__ == "__main__":
main()

9
tests/support/loader.py
View file

@ -50,6 +50,15 @@ _SIBLING_MODULES = (
"email_parsing",
"emf",
"web_ui_auth",
# procurement-api siblings (lambdas/api/): pagination/serialization/router
# have no sibling deps; wo_repo/po_repo import pagination, so they follow
# it. These names exist only under lambdas/api/, so the po/wo handler
# loads skip them via the exists() check.
"pagination",
"serialization",
"router",
"wo_repo",
"po_repo",
"prompts",
"template_parser",
"derived_fields",

280
tests/test_api_handlers.py Normal file
View file

@ -0,0 +1,280 @@
"""procurement-api handler tests (offline, fake Dynamo).
Pins the auth seam both ways: docs routes fail closed on the shared token
(including the browser ?token= shim), while data routes must NOT consult the
token at all -- SigV4 authorization is API Gateway's job upstream, and a
handler-side token check there would break every legitimate SigV4 caller.
Also pins routing, 404/501/400 mapping, Decimal-safe serialization, and limit
clamping.
"""
import json
from decimal import Decimal
import pytest
from tests.support import load_lambda_module
class _FakeTable:
def __init__(self, items):
self._items = items
self.last_kwargs = None
def scan(self, **kwargs):
self.last_kwargs = kwargs
return {"Items": list(self._items)}
def query(self, **kwargs):
self.last_kwargs = kwargs
return {"Items": list(self._items)}
def get_item(self, Key): # noqa: N803 (boto3 kwarg name)
pk_attr, pk_val = next(iter(Key.items()))
for item in self._items:
if item.get(pk_attr) == pk_val:
return {"Item": item}
return {}
class _FakeDynamo:
def __init__(self, tables):
self.tables = tables
def Table(self, name): # noqa: N802 (boto3 method name)
return self.tables[name]
def _event(method, resource, path_params=None, qs=None, headers=None):
return {
"httpMethod": method,
"resource": resource,
"pathParameters": path_params or {},
"queryStringParameters": qs or {},
"headers": headers or {},
}
@pytest.fixture()
def api(monkeypatch):
mod = load_lambda_module("api", "handler")
wo_items = [
{
"work_order_id": "11144580730",
"wo_status": "assigned",
"description": "Dock door 14",
}
]
comment_items = [
{
"work_order_id": "11144580730",
"comment_id": "11144580730#2026-01-01T00:00:00#abc123def456",
"text": "Vendor dispatched",
}
]
po_items = [
{
"po_number": "2D-22030794",
"total_amount": Decimal("123.45"),
"line_items": [{"quantity": Decimal("5"), "price": Decimal("24.69")}],
}
]
site_items = [{"siteCode": "JFK8", "state": "NY", "poCount": Decimal("7")}]
tables = {
mod.wo_repo.WORK_ORDERS_TABLE: _FakeTable(wo_items),
mod.wo_repo.COMMENTS_TABLE: _FakeTable(comment_items),
mod.po_repo.PO_TABLE: _FakeTable(po_items),
mod.po_repo.VERIFIED_SITES_TABLE: _FakeTable(site_items),
}
fake = _FakeDynamo(tables)
monkeypatch.setattr(mod.wo_repo, "dynamodb", fake)
monkeypatch.setattr(mod.po_repo, "dynamodb", fake)
return mod, fake
def test_healthcheck_short_circuits(api):
mod, _ = api
assert mod.handler({"healthcheck": True}, None) == {"healthcheck": "ok"}
def test_docs_fail_closed_without_token(api):
# No WEB_UI_AUTH_TOKEN_SECRET_ARN configured -> web_ui_auth returns None
# -> the docs routes must 401, not serve the spec.
mod, _ = api
for resource in ("/docs", "/openapi.json"):
response = mod.handler(_event("GET", resource), None)
assert response["statusCode"] == 401
def test_docs_served_when_authenticated(api, monkeypatch):
mod, _ = api
monkeypatch.setattr(mod, "is_authenticated", lambda event: True)
docs = mod.handler(_event("GET", "/docs"), None)
assert docs["statusCode"] == 200
assert docs["headers"]["Content-Type"].startswith("text/html")
# The vendored Redoc bundle is inlined and the spec (carrying the title)
# with it.
assert "Redoc.init" in docs["body"]
assert "Procurement Ingest API" in docs["body"]
# Every placeholder must be substituted, or the page renders broken.
for placeholder in (
mod._SPEC_PLACEHOLDER,
mod._JS_PLACEHOLDER,
mod._FONTS_PLACEHOLDER,
):
assert placeholder not in docs["body"]
spec = mod.handler(_event("GET", "/openapi.json"), None)
assert spec["statusCode"] == 200
assert json.loads(spec["body"])["openapi"] == "3.1.0"
def test_token_query_shim_synthesizes_header(api, monkeypatch):
mod, _ = api
seen = {}
def capture(event):
seen.update(event.get("headers") or {})
return False
monkeypatch.setattr(mod, "is_authenticated", capture)
original = _event("GET", "/docs", qs={"token": "sekrit"})
mod.handler(original, None)
assert seen.get("x-auth-token") == "sekrit"
# The shim must not mutate the caller's event in place.
assert "x-auth-token" not in original["headers"]
def test_non_ascii_token_fails_closed_not_500(api):
# A non-ASCII presented token must 401 (fail closed), not crash
# hmac.compare_digest into a 500 that pages the 5xx alarm. Reach the
# web_ui_auth module namespace through the imported is_authenticated's
# globals and seed a valid cached token.
mod, _ = api
auth_globals = mod.is_authenticated.__globals__
auth_globals["_auth_token_cache"] = "correct-token"
auth_globals["_auth_token_cached_at"] = float("inf")
try:
response = mod.handler(
_event("GET", "/docs", headers={"x-auth-token": "é"}), None
)
assert response["statusCode"] == 401
finally:
auth_globals["_auth_token_cache"] = None
auth_globals["_auth_token_cached_at"] = 0.0
def test_data_routes_never_consult_docs_token(api, monkeypatch):
mod, _ = api
def explode(event):
raise AssertionError("data routes must not consult the docs token")
monkeypatch.setattr(mod, "is_authenticated", explode)
response = mod.handler(_event("GET", "/work-orders"), None)
assert response["statusCode"] == 200
def test_list_and_get_routing(api):
mod, _ = api
listing = mod.handler(_event("GET", "/work-orders"), None)
body = json.loads(listing["body"])
assert body["items"][0]["work_order_id"] == "11144580730"
assert body["next_cursor"] is None
hit = mod.handler(
_event(
"GET",
"/work-orders/{workOrderId}",
path_params={"workOrderId": "11144580730"},
),
None,
)
assert json.loads(hit["body"])["wo_status"] == "assigned"
miss = mod.handler(
_event(
"GET",
"/work-orders/{workOrderId}",
path_params={"workOrderId": "999"},
),
None,
)
assert miss["statusCode"] == 404
comments = mod.handler(
_event(
"GET",
"/work-orders/{workOrderId}/comments",
path_params={"workOrderId": "11144580730"},
),
None,
)
assert json.loads(comments["body"])["items"][0]["text"] == "Vendor dispatched"
sites = mod.handler(
_event(
"GET",
"/verified-sites/{siteCode}",
path_params={"siteCode": "JFK8"},
),
None,
)
assert json.loads(sites["body"])["state"] == "NY"
def test_decimal_serialization_round_trips(api):
mod, _ = api
response = mod.handler(
_event(
"GET",
"/purchase-orders/{poNumber}",
path_params={"poNumber": "2D-22030794"},
),
None,
)
body = json.loads(response["body"])
assert body["total_amount"] == 123.45
assert body["line_items"][0]["quantity"] == 5
assert isinstance(body["line_items"][0]["quantity"], int)
assert "Decimal" not in response["body"]
def test_limit_clamped_to_bounds(api):
mod, fake = api
mod.handler(_event("GET", "/work-orders", qs={"limit": "9999"}), None)
assert fake.tables[mod.wo_repo.WORK_ORDERS_TABLE].last_kwargs["Limit"] == 500
mod.handler(_event("GET", "/work-orders", qs={"limit": "0"}), None)
assert fake.tables[mod.wo_repo.WORK_ORDERS_TABLE].last_kwargs["Limit"] == 1
def test_non_integer_limit_is_400(api):
mod, _ = api
response = mod.handler(_event("GET", "/work-orders", qs={"limit": "abc"}), None)
assert response["statusCode"] == 400
def test_planned_routes_return_501(api):
mod, _ = api
for method, resource in mod.PLANNED_ROUTES:
response = mod.handler(_event(method, resource), None)
assert response["statusCode"] == 501, (method, resource)
def test_unknown_route_is_404(api):
mod, _ = api
response = mod.handler(_event("GET", "/nope"), None)
assert response["statusCode"] == 404
def test_unexpected_error_maps_to_clean_500(api, monkeypatch):
mod, fake = api
def boom(**kwargs):
raise RuntimeError("dynamo fell over")
monkeypatch.setattr(fake.tables[mod.wo_repo.WORK_ORDERS_TABLE], "scan", boom)
response = mod.handler(_event("GET", "/work-orders"), None)
assert response["statusCode"] == 500
assert json.loads(response["body"]) == {"error": "internal error"}

232
tests/test_api_pagination_moto.py Normal file
View file

@ -0,0 +1,232 @@
"""Cursor pagination against real DynamoDB semantics (moto).
The offline handler tests fake Dynamo, which can't exercise real
LastEvaluatedKey behavior. Here moto-backed tables with the production key
schemas prove: pages are disjoint and complete, the final page's next_cursor
is null, comment Query pagination respects the composite key, and hostile
cursors (garbage, wrong key attrs, huge) map to 400 -- never 500, never a raw
ExclusiveStartKey error.
The repo modules build their boto3 resource at import time; a resource created
before mock_aws starts is NOT intercepted, so each test rebinds the modules'
``dynamodb`` to a fresh resource inside the mock (same trap as test_po_merge's
moto-before-handler ordering).
"""
import base64
import json
import boto3
import pytest
from moto import mock_aws
from tests.support import load_lambda_module
def _event(method, resource, path_params=None, qs=None):
return {
"httpMethod": method,
"resource": resource,
"pathParameters": path_params or {},
"queryStringParameters": qs or {},
"headers": {},
}
@pytest.fixture()
def api(monkeypatch):
with mock_aws():
mod = load_lambda_module("api", "handler")
resource = boto3.resource("dynamodb", region_name="us-east-1")
monkeypatch.setattr(mod.wo_repo, "dynamodb", resource)
monkeypatch.setattr(mod.po_repo, "dynamodb", resource)
resource.create_table(
TableName=mod.wo_repo.WORK_ORDERS_TABLE,
KeySchema=[{"AttributeName": "work_order_id", "KeyType": "HASH"}],
AttributeDefinitions=[
{"AttributeName": "work_order_id", "AttributeType": "S"}
],
BillingMode="PAY_PER_REQUEST",
)
resource.create_table(
TableName=mod.wo_repo.COMMENTS_TABLE,
KeySchema=[
{"AttributeName": "work_order_id", "KeyType": "HASH"},
{"AttributeName": "comment_id", "KeyType": "RANGE"},
],
AttributeDefinitions=[
{"AttributeName": "work_order_id", "AttributeType": "S"},
{"AttributeName": "comment_id", "AttributeType": "S"},
],
BillingMode="PAY_PER_REQUEST",
)
resource.create_table(
TableName=mod.po_repo.PO_TABLE,
KeySchema=[{"AttributeName": "po_number", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "po_number", "AttributeType": "S"}],
BillingMode="PAY_PER_REQUEST",
)
resource.create_table(
TableName=mod.po_repo.VERIFIED_SITES_TABLE,
KeySchema=[{"AttributeName": "siteCode", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "siteCode", "AttributeType": "S"}],
BillingMode="PAY_PER_REQUEST",
)
yield mod, resource
def _walk_pages(mod, resource_path, qs_extra=None, path_params=None, limit=5):
seen, pages = [], 0
cursor = None
while True:
qs = {"limit": str(limit), **(qs_extra or {})}
if cursor:
qs["cursor"] = cursor
response = mod.handler(
_event("GET", resource_path, path_params=path_params, qs=qs), None
)
assert response["statusCode"] == 200
body = json.loads(response["body"])
seen.extend(body["items"])
pages += 1
cursor = body["next_cursor"]
assert pages < 50, "pagination failed to terminate"
if cursor is None:
return seen, pages
def test_work_order_scan_pages_are_disjoint_and_complete(api):
mod, resource = api
table = resource.Table(mod.wo_repo.WORK_ORDERS_TABLE)
for i in range(12):
table.put_item(Item={"work_order_id": str(10000 + i), "wo_status": "new"})
items, pages = _walk_pages(mod, "/work-orders", limit=5)
ids = [item["work_order_id"] for item in items]
assert len(ids) == 12
assert len(set(ids)) == 12
assert pages >= 3
def test_comment_query_pagination_composite_key(api):
mod, resource = api
table = resource.Table(mod.wo_repo.COMMENTS_TABLE)
for i in range(7):
table.put_item(
Item={
"work_order_id": "555",
"comment_id": f"555#2026-01-0{i + 1}T00:00:00#{i:012d}",
"text": f"comment {i}",
}
)
# A second work order's comments must never bleed into the page.
table.put_item(
Item={"work_order_id": "666", "comment_id": "666#x#0", "text": "other"}
)
items, _ = _walk_pages(
mod,
"/work-orders/{workOrderId}/comments",
path_params={"workOrderId": "555"},
limit=3,
)
assert len(items) == 7
assert {item["work_order_id"] for item in items} == {"555"}
def test_purchase_order_and_site_pagination(api):
mod, resource = api
po_table = resource.Table(mod.po_repo.PO_TABLE)
for i in range(6):
po_table.put_item(Item={"po_number": f"2D-{i:08d}"})
sites_table = resource.Table(mod.po_repo.VERIFIED_SITES_TABLE)
for code in ("JFK8", "WNY2", "UVA5"):
sites_table.put_item(Item={"siteCode": code})
po_items, _ = _walk_pages(mod, "/purchase-orders", limit=4)
assert len(po_items) == 6
site_items, _ = _walk_pages(mod, "/verified-sites", limit=2)
assert {item["siteCode"] for item in site_items} == {"JFK8", "WNY2", "UVA5"}
@pytest.mark.parametrize(
"cursor",
[
"not-base64!!!",
"aGVsbG8=", # base64("hello") -- not JSON-dict
"e30=", # base64("{}") -- empty dict
# base64 of {"evil_attr": "x"} -- key attr not allowed for this table
"eyJldmlsX2F0dHIiOiAieCJ9",
"A" * 3000, # oversized
],
)
def test_hostile_cursor_is_400_not_500(api, cursor):
mod, resource = api
resource.Table(mod.wo_repo.WORK_ORDERS_TABLE).put_item(Item={"work_order_id": "1"})
response = mod.handler(_event("GET", "/work-orders", qs={"cursor": cursor}), None)
assert response["statusCode"] == 400
assert "error" in json.loads(response["body"])
def _cursor(payload):
return base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
def test_partial_composite_cursor_is_400_not_500(api):
# A work-orders-shaped cursor {work_order_id} is a partial key for the
# comments Query (needs work_order_id+comment_id). Exact-key match rejects
# it as 400 rather than letting DynamoDB ValidationException -> 500.
mod, resource = api
resource.Table(mod.wo_repo.COMMENTS_TABLE).put_item(
Item={"work_order_id": "555", "comment_id": "555#x#0"}
)
response = mod.handler(
_event(
"GET",
"/work-orders/{workOrderId}/comments",
path_params={"workOrderId": "555"},
qs={"cursor": _cursor({"work_order_id": "555"})},
),
None,
)
assert response["statusCode"] == 400
def test_cross_work_order_comment_cursor_is_400(api):
# A full comments cursor minted for WO A must not resume WO B's Query.
mod, resource = api
response = mod.handler(
_event(
"GET",
"/work-orders/{workOrderId}/comments",
path_params={"workOrderId": "B"},
qs={"cursor": _cursor({"work_order_id": "A", "comment_id": "A#x#0"})},
),
None,
)
assert response["statusCode"] == 400
def test_dynamodb_validation_error_maps_to_400(api, monkeypatch):
# Defense in depth: even if some crafted-but-valid cursor reached DynamoDB
# and raised ValidationException, the handler maps it to 400, not a 500
# that would page the zero-threshold 5xx alarm.
from botocore.exceptions import ClientError
mod, _ = api
class _RaisingTable:
def scan(self, **kwargs):
raise ClientError(
{"Error": {"Code": "ValidationException", "Message": "bad key"}},
"Scan",
)
monkeypatch.setattr(
mod.wo_repo,
"dynamodb",
type("D", (), {"Table": lambda self, n: _RaisingTable()})(),
)
response = mod.handler(_event("GET", "/work-orders"), None)
assert response["statusCode"] == 400

131
tests/test_api_spec_drift.py Normal file
View file

@ -0,0 +1,131 @@
"""Spec <-> implementation drift gate.
lambdas/api/openapi.json is the published contract; lambdas/api/router.py is
what the Lambda actually serves. This test makes them the SAME set: an
endpoint added/removed/renamed on one side without the other fails CI here,
so the docs page can never silently lie. Also pins the phase-2 x-planned
markers, the four outbound webhook events, and the enum values the spec
promises against the WO extraction contract (prompts.py).
"""
import json
from pathlib import Path
from tests.support import REPO_ROOT, load_lambda_module
_SPEC_PATH = Path(REPO_ROOT) / "lambdas" / "api" / "openapi.json"
_HTTP_METHODS = {"get", "put", "post", "patch", "delete", "head", "options"}
def _spec():
return json.loads(_SPEC_PATH.read_text(encoding="utf-8"))
def _spec_routes(spec):
implemented, planned = set(), set()
for path, ops in spec["paths"].items():
for method, op in ops.items():
if method not in _HTTP_METHODS:
continue
key = (method.upper(), path)
if op.get("x-planned"):
planned.add(key)
else:
implemented.add(key)
return implemented, planned
def test_spec_is_openapi_31():
assert _spec()["openapi"] == "3.1.0"
def test_implemented_routes_match_router_exactly():
router = load_lambda_module("api", "router")
implemented, planned = _spec_routes(_spec())
assert implemented == set(router.DATA_ROUTES) | set(router.DOCS_ROUTES)
assert planned == set(router.PLANNED_ROUTES)
def test_every_data_route_has_a_repo_function():
router = load_lambda_module("api", "router")
handler = load_lambda_module("api", "handler")
assert set(router.DATA_ROUTES.values()) == set(handler._REPO_FUNCS)
def test_webhooks_section_documents_all_four_events():
spec = _spec()
assert set(spec["webhooks"]) == {
"work_order.created",
"work_order.updated",
"work_order.cancelled",
"work_order.comment_added",
}
for ops in spec["webhooks"].values():
assert "post" in ops
def test_spec_enums_match_wo_extraction_contract():
# The WO pipeline's prompts.py is the enum source of truth (the webhook
# contract pins it too). If the pipeline ever widens wo_status or
# record_type, the published spec must move in the same PR.
prompts = load_lambda_module("wo", "email_processor/prompts")
prompt_text = prompts.EXTRACTION_PROMPT
spec = _spec()
schemas = spec["components"]["schemas"]
wo_status_enum = {
value
for value in schemas["WorkOrder"]["properties"]["wo_status"]["enum"]
if value is not None
}
record_type_enum = {
value
for value in schemas["WorkOrder"]["properties"]["record_type"]["enum"]
if value is not None
}
for value in wo_status_enum - {"unknown"}:
assert value in prompt_text, f"wo_status {value!r} not in extraction prompt"
for value in record_type_enum:
assert value in prompt_text, f"record_type {value!r} not in extraction prompt"
event_data = schemas["WorkOrderEventData"]["properties"]
assert set(event_data["wo_status"]["enum"]) == set(
schemas["WorkOrder"]["properties"]["wo_status"]["enum"]
)
def test_spec_has_no_script_breakout_sequence():
# The spec is inlined into a <script type="application/json"> block on the
# docs page. The handler escapes "<" -> < defensively, but keep the
# committed spec itself clean so the raw /openapi.json is also breakout-safe
# and a reviewer sees the invariant here.
raw = _SPEC_PATH.read_text(encoding="utf-8")
assert "</" not in raw and "<!--" not in raw
def test_docs_and_spec_files_ship_with_the_handler():
# The handler serves these from its own package dir; if any file is missing
# from lambdas/api/ the bundle would 500 at runtime. Redoc is vendored
# offline (no CDN) and inlined into the single token-gated response.
api_dir = Path(REPO_ROOT) / "lambdas" / "api"
assert (api_dir / "openapi.json").is_file()
assert (api_dir / "docs.html").is_file()
assert (api_dir / "redoc.standalone.js").is_file()
assert (api_dir / "fonts.css").is_file()
docs = (api_dir / "docs.html").read_text(encoding="utf-8")
assert "__OPENAPI_SPEC_JSON__" in docs
assert "__REDOC_JS__" in docs
assert "__FONTS_CSS__" in docs
assert "Redoc.init" in docs
# The docs page must never fetch fonts (or anything else) off-box: the
# design-system fonts ride inline as data URIs in fonts.css.
assert "fonts.googleapis.com" not in docs
# The vendored blobs must carry no raw </script>/</style>: each is inlined
# into a <script>/<style> block a literal terminator would break out of.
# (The handler also escapes them defensively, but keeping the pinned
# assets clean is the load-bearing guarantee and catches a bad bump here.)
assert "</script" not in (api_dir / "redoc.standalone.js").read_text(
encoding="utf-8"
)
assert "</style" not in (api_dir / "fonts.css").read_text(encoding="utf-8")

116
tests/test_bundle_consistency.py
View file

@ -21,8 +21,10 @@ 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"
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"
# Phase 3: ses_auth/web_ui_auth/email_parsing/emf are single-sourced here and
# shipped into the email-processor bundles via `cp shared/*.py`.
SHARED_DIR = REPO_ROOT / "lambdas" / "shared"
@ -90,6 +92,30 @@ WO_PIPELINE_MODULES = frozenset(
PO_EXPECTED_TOP_LEVEL_MODULES = PO_PIPELINE_MODULES | SHARED_MODULES
WO_EXPECTED_TOP_LEVEL_MODULES = WO_PIPELINE_MODULES | SHARED_MODULES
# procurement-api (lambdas/api/): a fourth bundled function, in its own stack.
# Same exact-set discipline as the pipeline dirs -- `cp api/*.py` ships every
# top-level .py here (untracked strays included), so any new module is a
# deliberate addition to this pin.
API_PIPELINE_MODULES = frozenset(
{
"handler",
"router",
"pagination",
"serialization",
"wo_repo",
"po_repo",
}
)
# Non-.py files the api bundle must also EXECUTE cps for: the handler serves
# the spec, docs page, and the vendored Redoc bundle from its own package
# dir, so dropping any cp 500s /docs at runtime with green CI.
API_DATA_FILES_CP_RES = (
r"(?:^|\s)api/openapi\.json(?:\s|$)",
r"(?:^|\s)api/docs\.html(?:\s|$)",
r"(?:^|\s)api/redoc\.standalone\.js(?:\s|$)",
r"(?:^|\s)api/fonts\.css(?:\s|$)",
)
# 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
@ -652,3 +678,93 @@ def test_detection_logic_rejects_commented_out_web_ui_auth_cp():
)
executed = _executed_cp_commands(commented_out_command)
assert not any(re.search(WEB_UI_AUTH_CP_RE, cp) for cp in executed)
def _extract_api_command(stack_path: Path) -> str:
"""Extract the procurement-api function's bash -c bundling command string.
Mirrors _extract_bundling_command but selects the command whose text
mentions ``api/`` (its first cp is ``cp api/*.py``). procurement_api_stack
has exactly one bundled function today; the exactly-one demand makes any
future second bundle in that stack surface 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)
api_cmds = [c for c in commands if "api/" in c]
if len(api_cmds) != 1:
raise AssertionError(
f"expected exactly one api bundling command=[...] list in "
f"{stack_path}, found {len(api_cmds)} (of {len(commands)} total "
"command lists) — update this test to select the intended bundling "
"command explicitly"
)
return api_cmds[0]
def test_api_bundling_ships_all_first_party_siblings():
"""The procurement-api bundle must ship every sibling handler.py imports.
Same PR #105 ImportError class as the email processors: the api handler
imports router/pagination/serialization/wo_repo/po_repo (next to it) and
web_ui_auth (lambdas/shared/); a narrowed glob or dropped shared cp would
cold-start ImportError with green CI.
"""
required = _first_party_sibling_imports(API_HANDLER)
assert required, "expected api/handler.py to import first-party siblings"
command = _extract_api_command(API_STACK)
assert _bundling_ships_all(command, required), (
f"cdk/procurement_api_stack.py bundling does not ship all first-party "
f"siblings {sorted(required)}. Executed cp commands: "
f"{_executed_cp_commands(command)}"
)
def test_api_dir_ships_no_unexpected_top_level_modules():
"""Exact-set pin on lambdas/api/*.py -- both bounds.
`cp api/*.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 API_PIPELINE_MODULES.
"""
api_dir = REPO_ROOT / "lambdas" / "api"
top_level = {p.stem for p in api_dir.glob("*.py")}
assert top_level == set(API_PIPELINE_MODULES), (
f"lambdas/api/ top-level modules {sorted(top_level)} != pinned "
f"{sorted(API_PIPELINE_MODULES)}. If a new module is intended, add it "
"to API_PIPELINE_MODULES."
)
def test_api_bundle_stages_shared_auth_module():
"""The api bundle must EXECUTE `cp shared/web_ui_auth.py` (docs-token gate)."""
command = _extract_api_command(API_STACK)
executed_cps = _executed_cp_commands(command)
assert any(re.search(WEB_UI_AUTH_CP_RE, cp) for cp in executed_cps), (
"cdk/procurement_api_stack.py bundling command must EXECUTE "
"'cp shared/web_ui_auth.py /asset-output/' so the docs-token gate "
f"resolves at cold start. Executed cp commands: {executed_cps}"
)
def test_api_bundle_stages_spec_and_docs_page():
"""The api bundle must EXECUTE cps for openapi.json and docs.html.
The handler serves both from its package dir; the .py glob does not cover
them, so each needs its own executed cp or /docs 500s at runtime.
"""
command = _extract_api_command(API_STACK)
executed_cps = _executed_cp_commands(command)
for pattern in API_DATA_FILES_CP_RES:
assert any(re.search(pattern, cp) for cp in executed_cps), (
f"cdk/procurement_api_stack.py bundling command must EXECUTE a cp "
f"matching {pattern!r}. Executed cp commands: {executed_cps}"
)

14
tests/test_web_ui_auth.py
View file

@ -105,16 +105,12 @@ def test_header_matrix(load_auth, monkeypatch, headers, expected):
assert mod.is_authenticated(_evt(**headers)) is expected
@pytest.mark.xfail(
strict=True,
reason="web_ui_auth is frozen this phase: is_authenticated raises TypeError on "
"a non-ASCII presented token (hmac.compare_digest is ASCII-only) and surfaces "
"as a 500 instead of failing closed to False. Asserting the DESIRED behavior "
"under xfail(strict) documents the intended fix and auto-fails (xpass) the moment "
"the module is corrected, forcing the marker's removal. Tracked as a follow-up.",
raises=TypeError,
)
def test_non_ascii_token_should_fail_closed(load_auth, monkeypatch):
# Fixed 2026-07-23 (procurement-api security review): is_authenticated now
# compares tokens as bytes, so a non-ASCII presented token fails closed to
# False instead of raising TypeError (which surfaced as a 500 that would
# page the API's zero-threshold 5xx alarm). Was xfail(strict) while
# web_ui_auth was frozen; the fix landed with the shared bytes-compare.
mod = load_auth()
monkeypatch.setattr(mod, "_get_auth_token", lambda: "SECRET")
assert mod.is_authenticated(_evt(**{"x-auth-token": "SÉCRET"})) is False