diff --git a/README.md b/README.md index 5af2777..0c33014 100644 --- a/README.md +++ b/README.md @@ -16,12 +16,13 @@ Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed by Claude H 1. Coupa sends a PO email (new, revision, or cancellation). 2. SES (`INBOUND_MAIL` rule set) drops the raw MIME into `s3://po-ingest-emails-{AccountId}/inbound/`. 3. S3 `ObjectCreated` triggers the `po-email-processor` Lambda. -4. Claude (via Bedrock `InvokeModel`) extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year). Numeric amounts are decoded with `parse_float=Decimal` (DynamoDB rejects floats). -5. Conditional write to DynamoDB: - - `new_po` — idempotent insert (no-op if PO exists) - - `revision` — unconditional overwrite - - `cancellation` — marks existing row `Cancelled` -6. DynamoDB Streams feeds downstream consumers: +4. Fail-closed sender authentication (INFRA-107): the SES-stamped `Authentication-Results` header must show `dkim=pass` for `amazon.coupahost.com` (see [Sender authentication](#sender-authentication-infra-107)); otherwise the email is logged and dropped. +5. Claude (via Bedrock `InvokeModel`) extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year). Numeric amounts are decoded with `parse_float=Decimal` (DynamoDB rejects floats). +6. Merge write to DynamoDB: + - `new_po` — merge insert. Creates the PO, or backfills data into a pre-existing `Cancelled` skeleton left by an out-of-order cancellation (preserving the `Cancelled` status). No longer silently dropped when a record already exists. + - `revision` — field-level merge (`update_item` SETs only the fields present in the revision). A revision that omits `line_items`/`supplier` no longer deletes them. Will not un-cancel a `Cancelled` PO. + - `cancellation` — marks the row `Cancelled` (creating a minimal skeleton if the cancellation arrives before the `new_po`). +7. DynamoDB Streams feeds downstream consumers: - **LedgerFlow** (`seahaven-slack-bot/po-sync`) — daily KB sync - **Site extractor** (`po-ingest-site-extractor`) — real-time site address extraction into `verified-sites` table @@ -46,8 +47,9 @@ Amazon APM work order emails (from Hexagon EAM / HxGN SmartCloud) are received a 2. Gmail filter forwards APM emails to `apm@int.seahaven.com` (SES). 3. SES drops the raw MIME into `s3://workorder-ingest-emails-{AccountId}/inbound/`. 4. S3 triggers the `workorder-email-processor` Lambda. -5. **Parse:** a pure, offline template parser (`template_parser.py`) tries the two known Hexagon templates first, behind a strict fail-closed validation gate. Only on a miss/invalid result does the Lambda fall back to the Claude-on-Bedrock AI extractor. Both paths emit the identical structured-JSON contract (work order ID, site code, severity, priority, dates, assigned technician). -6. Work order upserted to `WorkOrders`, event/comment appended to `WorkOrderComments`. +5. Fail-closed sender authentication (INFRA-107): the SES-stamped `Authentication-Results` header must show `dkim=pass` for the domain that re-signs the forward (currently allowlisted as `seahaven.com` — see the validation caveat under [Sender authentication](#sender-authentication-infra-107)); otherwise the email is logged and dropped. +6. **Parse:** a pure, offline template parser (`template_parser.py`) tries the two known Hexagon templates first, behind a strict fail-closed validation gate. Only on a miss/invalid result does the Lambda fall back to the Claude-on-Bedrock AI extractor. Both paths emit the identical structured-JSON contract (work order ID, site code, severity, priority, dates, assigned technician). +7. Work order upserted to `WorkOrders`, event/comment appended to `WorkOrderComments`. **Deterministic template parser.** ~93.6% of WO traffic is the plain-text "AMAZON UPDATE WO DETAILS \" comment template (T1) and ~6.4% is the HTML "AMAZON assign Work Order \ on building \" assignment template (T2). `template_parser.try_deterministic_parse()` classifies by subject, extracts the contract fields, and returns a parsed result **only if** it passes a strict validation gate (exact contract-key set; `work_order_id` matches the subject and is all-digits; the T1 `Work Order: ` double-space is literally present; `email_type` matches the template; `site_code` shape; per-type required fields; and a label-bleed guard so a value that over-ran into another field fails). Anything that fails — the rare update/cancellation shapes, Hexagon template drift, or an extractor exception — falls back to the AI extractor. **Data is never corrupted; only the fallback rate rises.** Every record emits one CloudWatch EMF metric (see below). @@ -69,10 +71,31 @@ All Lambdas: Python 3.12, ARM64, 60-day log retention. **LLM provider — Amazon Bedrock.** Both email processors call Claude Haiku 4.5 through the Bedrock inference profile `us.anthropic.claude-haiku-4-5-20251001-v1:0` (`bedrock-runtime` `InvokeModel`), configured via the `BEDROCK_MODEL_ID` env var. There is **no Anthropic API key and no Secrets Manager secret** any more. Each processor role is granted `bedrock:InvokeModel` + `bedrock:InvokeModelWithResponseStream` on **both** the inference-profile ARN **and** the per-region foundation-model ARNs for `us-east-1` / `us-east-2` / `us-west-2` (empty-account foundation-model ARNs) — the `us.*` profile can route cross-region, so a profile-only grant would `AccessDenied` at runtime under load. -> **Retired secrets (manual cleanup outstanding):** the former secrets `po-ingest/anthropic-api-key` and `workorder-ingest/anthropic-api-key` had `RemovalPolicy.RETAIN`, so removing them from the CDK stacks **orphans** them rather than deleting them. Delete both by hand post-deploy and revoke the stored keys at the provider. +> **Retired secrets (manual cleanup outstanding):** the former secrets `po-ingest/anthropic-api-key` and `workorder-ingest/anthropic-api-key` had `RemovalPolicy.RETAIN`, so removing them from the CDK stacks **orphans** them rather than deleting them. Delete both by hand post-deploy and revoke the stored keys at the provider. The processors no longer read any `ANTHROPIC_API_KEY_SECRET_ARN` — authentication to Bedrock is via the Lambda execution-role IAM grant, so there is no provider API key or Secrets Manager fetch on the parse path. **SES:** Both stacks add rules to the shared `INBOUND_MAIL` receipt rule set on `int.seahaven.com`. +### Sender authentication (INFRA-107) + +The `From` header and any `Authentication-Results` header inside the raw MIME are attacker-forgeable, so neither is trusted. Instead, both email processors (`lambdas/*/email_processor/ses_auth.py`) authenticate the sender against the verdicts SES itself stamps at delivery time, failing closed: + +1. Take **only the topmost** `Authentication-Results` header (SES prepends its trace headers; any lower copies arrived inside the message and are ignored). +2. Require its authserv-id to be `amazonses.com`. +3. Require a `dkim=pass` clause whose `header.d=`/`header.i=` domain is in the pipeline's allowlist. + +The allowlist is the `ALLOWED_DKIM_DOMAINS` Lambda environment variable (comma-separated, set per stack in CDK — no code change needed to adjust): + +| Pipeline | `ALLOWED_DKIM_DOMAINS` | Why | +|---|---|---| +| Work orders | `seahaven.com` | APM mail reaches `apm@int.seahaven.com` via a forward off `amazon@seahavenind.com`; the allowlist trusts the domain that **re-signs** DKIM on that forward (the original `hxgnsmartcloud.com` signature does not survive it). **Validated against live SES-stamped headers (2026-07-16)** — real APM deliveries carry `dkim=pass header.i=@seahaven.com`. Note a plain Gmail auto-forward re-signs under the *sending Workspace* domain (`seahavenind.com` / a `*.gappssmtp.com` key), **not** `seahaven.com`; only a Google Group (or Workspace routing) with "sign as `seahaven.com`" produces `dkim=pass header.i=@seahaven.com`. If the observed re-signing domain differs, update this value (do **not** widen it to a shared key like `*.gappssmtp.com`, which any Google customer's mail would pass). The `workorder-email-processor-sender-auth-rejected` alarm pages if this assumption is wrong instead of silently dropping every work order. | +| Purchase orders | `amazon.coupahost.com` | Coupa signs as `amazon.coupahost.com`. `amazonses.com` also passes but is deliberately not allowlisted — every SES customer's mail passes for it | + +> **Clause-injection hardening:** SES echoes attacker-controlled SMTP-session tokens (`envelope-from`, `helo`, `header.from`) into its own `Authentication-Results` value, and an RFC 5321 quoted-local-part MAIL FROM may legally contain `;` and spaces. The parser therefore tokenises comment- and quoted-string-aware (RFC 8601 / RFC 5322): CFWS comments `(...)` are stripped and clauses are split only on semicolons **outside** a quoted string, so a `;` inside a quoted `envelope-from=` value can never be torn into a forged `dkim=pass` clause. The DKIM signer domain is read from `header.d=` when present (falling back to `header.i=`, taking the domain after the AUID's last top-level `@` so a quoted local-part cannot smuggle an allowlisted domain). Two latent comment-parsing edge cases (early comment-close, no-separator-on-strip) are tracked as hardening follow-ups — see the SES-AR-01/02 issue; neither is reachable through SES's real header encoding today. + +> **Risk acceptance — forwarder-domain binding (INFRA-107, accepted 2026-07-16):** for work orders this control authenticates the domain that *re-signs* the `apm@` forward (`seahaven.com`), not the Hexagon originator (`hxgnsmartcloud.com`, whose signature does not survive the forward). Its strength therefore rests on the `apm@` Google Group's posting policy being restricted to trusted internal senders — that restriction is the **load-bearing control** and is accepted as documented risk. **If the `apm@` group is ever opened to external posting, this finding escalates to HIGH** (anyone able to post to the group could inject a forged work order) and the correct fix is to bind acceptance to the originator via DMARC alignment rather than the forwarder's re-signature. The PO pipeline is unaffected — `amazon.coupahost.com` is an external domain an attacker cannot get SES to sign. + +On any failure (env var unset, header missing/unparseable, verdict fail, unaligned domain) the processor logs a structured `sender_auth_rejected` warning with the reason and S3 key, skips the email, and returns normally — rejected mail never triggers Lambda retries or DLQ messages, but the `-sender-auth-rejected` CloudWatch alarm (see [CloudWatch alarms](#cloudwatch-alarms)) pages on a rejection spike so a drift-induced outage is not silent. Unit tests live in `tests/test_ses_auth.py`. + **Failure handling (INFRA-41):** Each email-processor is async-invoked (S3 → Lambda). Both have a CDK-managed SQS dead-letter queue (`dead_letter_queue=`, 14-day retention, SSL-enforced) so a failed parse is captured rather than silently dropped after Lambda's retries. ### CloudWatch alarms @@ -86,6 +109,9 @@ Every alarm is **ALARM-only** (no OK action), sends to the shared `site-alerts` | `-errors` | `po-email-processor`, `po-ingest-site-extractor`, `workorder-email-processor` | `Errors` Sum, 5 min, `> 0`, eval 1 | | `-throttles` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui`, `workorder-email-processor` | `Throttles` Sum, 5 min, `> 0`, eval 1 | | `-duration` | `po-email-processor`, `po-ingest-site-extractor`, `po-web-ui` (p99); `workorder-email-processor` (p95) | `Duration` percentile, 5 min, `>= 45000` ms (75% of the 60s timeout), eval 3 / datapoints 2 | +| `-sender-auth-rejected` | `po-email-processor`, `workorder-email-processor` | Log-metric-filter count (namespace `Seahaven/ProcurementIngest`, `default_value=0`) on `sender_auth_rejected` warnings, `Sum` 5 min, `>= 1`, eval 3 / datapoints 2 | + +The `-sender-auth-rejected` alarm closes the silent-drop gap in INFRA-107: a rejected email returns normally (no error, no retry, no DLQ message), so without a log-metric filter a signing-domain drift or a wrong allowlist would discard 100% of legitimate mail while every other alarm stayed green. It counts `sender_auth_rejected` warnings per 5-minute period (`default_value=0` keeps the series continuous) and pages when 2 of the last 3 periods each see at least one rejection — a lone stray spoof probe to the internal ingest address self-clears, but a sustained false-reject storm pages within ~10–15 minutes even at low mail volume; the config is easy to tune in the CDK helper. (A residual gap remains for a *very* sparse total-reject outage — see the SES-AR-01/02 hardening issue.) The `-duration` and `-throttles` alarms for `po-email-processor` and `workorder-email-processor` supersede the orphaned, CLI-created `Lambda-Duration-*` / `Lambda-Throttles-*` alarms (deleted post-deploy). @@ -95,11 +121,17 @@ The `-duration` and `-throttles` alarms for `po-email-processor` and `wo **DynamoDB alarms** (`AWS/DynamoDB`): each owned table gets `-throttles` (`ThrottledRequests`) and `
-system-errors` (`SystemErrors`). These metrics emit only at the `TableName` + `Operation` dimension set, so each alarm is a `Sum` math expression across the operations the table uses (Get/BatchGet/Query/Scan/Put/Update/Delete/BatchWrite). Tables covered: `purchase-orders`, `verified-sites`, `pending-site-review` (po-ingest); `WorkOrders`, `WorkOrderComments` (workorder-ingest). +## Security + +**Web UI auth (defense-in-depth).** The `po-web-ui` / `workorder-web-ui` handlers refuse unauthenticated requests even though their public Function URLs were removed (INFRA-74). Each requires a shared secret in the `X-Auth-Token` header (or `Authorization: Bearer `), compared in constant time against the configured token. The handler **fails closed** if the token is unset or unreadable (denies all). The token lives in the Secrets Manager secret `procurement-ingest/web-ui-auth-token`; only its ARN is passed to the Lambda (`WEB_UI_AUTH_TOKEN_SECRET_ARN`), and the value is fetched at runtime — never embedded in the CloudFormation template or Lambda env vars. The fetched value is cached in the warm container with a short TTL (5 min) so a rotated secret propagates without waiting for the execution environment to recycle. This is a defense-in-depth floor for a detached URL, not primary auth. + +**Output escaping.** All caller-influenced values (including prompt-injectable strings Claude may return for `total_amount`/line-item amounts) are HTML-escaped before interpolation to prevent stored XSS. + ## Shared Resources ### `purchase-orders` table (owned here) -The `purchase-orders` DynamoDB table is **owned by this repo's `po-ingest` stack** (defined in `cdk/po_stack.py` with `RemovalPolicy.RETAIN`, `StreamViewType.NEW_IMAGE`, and SSE-KMS encryption with the shared customer-managed CMK `alias/seahaven-dynamodb`, INFRA-95 / M-3). The `po-email-processor` Lambda is the authoritative writer — it performs the conditional inserts, revision overwrites, and cancellation updates described above. +The `purchase-orders` DynamoDB table is **owned by this repo's `po-ingest` stack** (defined in `cdk/po_stack.py` with `RemovalPolicy.RETAIN`, `StreamViewType.NEW_IMAGE`, and SSE-KMS encryption with the shared customer-managed CMK `alias/seahaven-dynamodb`, INFRA-95 / M-3). The `po-email-processor` Lambda is the authoritative writer — it performs the merge inserts, field-level revision merges, and cancellation updates described above. **Consumers (read-only):** @@ -137,7 +169,7 @@ Both tables are **owned by this repo's `WorkorderIngestStack`** (`cdk/wo_stack.p > (e.g. `2026-07-15T12:00:00+00:00`). Downstream parsers that assumed a naive/no-offset string > must accept the offset. -Both currently use default DynamoDB encryption — they are **not** yet on the shared customer-managed CMK (`alias/seahaven-dynamodb`, INFRA-95 / M-3); that migration is tracked in INFRA-6. The `workorder-email-processor` role already holds a pre-emptive encrypt/decrypt grant on the CMK so the migration won't break it. +Both currently use default DynamoDB encryption — they are **not** yet on the shared customer-managed CMK (`alias/seahaven-dynamodb`, INFRA-95 / M-3); that migration is tracked in INFRA-6. The `workorder-email-processor` role no longer holds a pre-emptive encrypt/decrypt grant on that CMK (removed in the 2026-06-17 security sweep — it was unused while the tables are unencrypted and extended the role's decrypt reach to the CMK protecting `purchase-orders`). Re-add the grant as part of the INFRA-6 migration, at which point `grant_read_write_data` on the then-encrypted tables propagates the needed key permissions automatically. **Consumer (read-only) — data contract:** `seahaven-slack-bot` imports both tables via `Table.fromTableName(...)` (`grantReadData`) and reads them from two Lambdas: `workorder-sync` (daily full-table scan into the Bedrock knowledge base) and `wo-po-lookup` (the Bedrock agent's direct WO lookup action group). The bot depends on the PK/SK schema above and these attributes: on `WorkOrders` — `description`, `wo_status`, `customer`, `site_code`, `building`, `address`, `severity`, `priority`, `assigned_to`, `date_reported`, `scheduled_start`, `due_date`, `updated_at`; on `WorkOrderComments` — `created_at` (used to sort comments), `commenter`, `text`. Any change to table name, key schema, these attribute names, or the encryption configuration (e.g. the INFRA-6 CMK migration) must be coordinated with `seahaven-slack-bot` before it ships, or the Bedrock agent breaks at runtime (not at deploy — the tables are imported by name, so there is no compile-time link). @@ -168,18 +200,24 @@ Branch protection on `main` — all changes through PR. 1. Bootstrap CDK: `cdk bootstrap aws://{AccountId}/us-east-1` 2. Ensure the Bedrock inference profile `us.anthropic.claude-haiku-4-5-20251001-v1:0` is enabled in `us-east-1` (it is; the CDK grants cover cross-region routing to `us-east-2`/`us-west-2`). No API key or secret to set — the processors authenticate to Bedrock via their IAM roles. -3. Deploy both stacks: +3. Create the web UI auth-gate shared secret. This secret is **imported by name** + (`Secret.from_secret_name_v2`), not CDK-managed, so it must exist before deploy + or the web-ui Lambdas fail closed: + ```bash + aws secretsmanager create-secret --name procurement-ingest/web-ui-auth-token --secret-string "$(openssl rand -hex 32)" + ``` +4. Deploy both stacks: ```bash cd cdk pip install -r requirements.txt cdk deploy --all ``` -4. **Post-deploy cleanup (one-time):** the retired `RETAIN`-policy secrets `po-ingest/anthropic-api-key` and `workorder-ingest/anthropic-api-key` are orphaned by this deploy, not deleted. Remove them and revoke the keys at the provider: +5. **Post-deploy cleanup (one-time):** the retired `RETAIN`-policy secrets `po-ingest/anthropic-api-key` and `workorder-ingest/anthropic-api-key` are orphaned by this deploy, not deleted. Remove them and revoke the keys at the provider: ```bash aws secretsmanager delete-secret --secret-id po-ingest/anthropic-api-key --force-delete-without-recovery aws secretsmanager delete-secret --secret-id workorder-ingest/anthropic-api-key --force-delete-without-recovery ``` -5. Dashboards: `po-web-ui` and `workorder-web-ui` have no public endpoint (the Function URLs were removed 2026-06-08, INFRA-74). Invoke them manually and render the returned HTML, e.g. `aws lambda invoke --function-name po-web-ui /tmp/out.json`. +6. Dashboards: `po-web-ui` and `workorder-web-ui` have no public endpoint (the Function URLs were removed 2026-06-08, INFRA-74). Invoke them through an authenticated path that forwards the `X-Auth-Token` header, e.g. `aws lambda invoke --function-name po-web-ui /tmp/out.json`. ## Tests @@ -191,7 +229,7 @@ pytest ``` Coverage: -- `tests/` — shared handler tests (`parse_raw_email`, `pad_zip`). +- `tests/` — shared handler + cross-pipeline tests: `parse_raw_email` and `pad_zip`, the PO merge-write semantics (`test_po_merge.py`, #97, moto-backed), and the fail-closed sender-authentication parser (`test_ses_auth.py`, INFRA-107). - `lambdas/wo/email_processor/tests/` — the deterministic WO parser suite: golden-file tests over 55 real scrubbed `.eml` fixtures (`test_parser.py`), fail-closed validation-gate rules and adversarial/injection cases (`test_validation_gate.py`), the issue #23 `comment_id` idempotency invariants (`test_comment_id.py`), and the Bedrock-fallback dispatch/EMF-metric behavior with a mocked `invoke_model` (`test_bedrock_fallback.py`). Golden JSON lives under `tests/fixtures/expected/`. Both roots are discovered by `pytest.ini` (`testpaths`). @@ -230,6 +268,12 @@ lambdas/ scripts/ reprocess.py backfill_sites.py -tests/ # shared handler tests (discovered with the WO suite) test_local.py # Parse sample emails through Bedrock locally (no AWS mutation) +tests/ + requirements.txt # Test-only deps (moto) + conftest.py # AWS env stubs + per-pipeline module loader + test_pad_zip.py # PO zip-code padding tests + test_parse_raw_email.py # MIME parsing tests (PO + WO handlers) + test_po_merge.py # PO merge-write semantics tests (#97) + test_ses_auth.py # Sender-authentication parser tests (INFRA-107) ``` diff --git a/cdk/po_stack.py b/cdk/po_stack.py index 2e9061a..18b50c2 100644 --- a/cdk/po_stack.py +++ b/cdk/po_stack.py @@ -17,6 +17,7 @@ from aws_cdk import ( aws_s3_notifications as s3n, aws_ses as ses, aws_ses_actions as ses_actions, + aws_secretsmanager as secretsmanager, aws_sns as sns, aws_sqs as sqs, aws_ssm as ssm, @@ -77,6 +78,76 @@ def _add_ddb_alarms(scope, id_prefix, table, alarm_name_prefix, alarm_topic): ).add_alarm_action(cw_actions.SnsAction(alarm_topic)) +# CloudWatch namespace for the log-derived sender-authentication metrics. +_SENDER_AUTH_METRIC_NAMESPACE = "Seahaven/ProcurementIngest" + + +def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic): + """Metric-filter + alarm on ``sender_auth_rejected`` warnings (INFRA-107). + + A rejected inbound email is skipped without erroring the invocation, so it + is invisible to the Errors/Throttles/DLQ alarms. This turns the structured + warning log into a CloudWatch metric and pages when rejections spike -- + catching a silent false-reject storm (allowlist wrong, signing-domain + drift, SES header-format change) that would otherwise discard legitimate + mail while the pipeline reports healthy. + + ALARM-only SnsAction to site-alerts; no OK action. The metric filter reads + the function's own log group (imported by the deterministic + ``/aws/lambda/`` name, created by the function's log_retention). A plain + substring pattern is used because Lambda prefixes each line with its own + level/timestamp/request-id, so the JSON payload is not a standalone JSON + log event a `{$.event=...}` pattern could match. + """ + metric_name = f"{function_name}-sender-auth-rejected" + logs.MetricFilter( + scope, + f"{id_prefix}SenderAuthRejectedFilter", + log_group=logs.LogGroup.from_log_group_name( + scope, + f"{id_prefix}LogGroup", + f"/aws/lambda/{function_name}", + ), + filter_pattern=logs.FilterPattern.literal('"sender_auth_rejected"'), + metric_namespace=_SENDER_AUTH_METRIC_NAMESPACE, + metric_name=metric_name, + metric_value="1", + default_value=0, + ) + + # Fire on a *sustained* reject condition rather than a volume spike. The + # earlier Sum>=3-over-15-min threshold had a blind spot that is exactly the + # failure this alarm exists to catch: a low-traffic pipeline in total + # drift outage (allowlist wrong / signing-domain changed) may only produce + # a trickle of rejects -- one every few minutes -- that never sums to 3 in + # any window, so the outage never pages. Instead: >=1 reject per 5-min + # period, alarming when 2 of the last 3 periods breach (evaluation_periods=3 + # / datapoints_to_alarm=2, the same idiom as the duration alarm). A single + # stray spoof probe (one lone period) is tolerated and self-clears, but a + # sustained reject condition trips within ~10-15 min even at one reject per + # period. default_value=0 on the metric filter keeps the series continuous + # so NOT_BREACHING only applies before the first datapoint ever arrives. + cloudwatch.Metric( + namespace=_SENDER_AUTH_METRIC_NAMESPACE, + metric_name=metric_name, + period=Duration.minutes(5), + statistic="Sum", + ).create_alarm( + scope, + f"{id_prefix}SenderAuthRejectedAlarm", + alarm_name=f"{function_name}-sender-auth-rejected", + alarm_description=( + f"{function_name} rejected inbound mail on sender authentication " + "(possible allowlist/DKIM-domain drift silently dropping real mail)" + ), + threshold=1, + evaluation_periods=3, + datapoints_to_alarm=2, + comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD, + treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING, + ).add_alarm_action(cw_actions.SnsAction(alarm_topic)) + + class PoIngestStack(Stack): def __init__(self, scope: Construct, construct_id: str, **kwargs): super().__init__(scope, construct_id, **kwargs) @@ -173,7 +244,7 @@ class PoIngestStack(Stack): "-c", "pip install --platform manylinux2014_aarch64 --only-binary=:all: " "-r requirements.txt -t /asset-output && " - "cp handler.py /asset-output/", + "cp handler.py ses_auth.py /asset-output/", ], ), ), @@ -184,6 +255,14 @@ class PoIngestStack(Stack): environment={ "PO_TABLE": "purchase-orders", "BEDROCK_MODEL_ID": "us.anthropic.claude-haiku-4-5-20251001-v1:0", + # Fail-closed sender auth (INFRA-107): the handler only + # accepts mail whose SES-stamped Authentication-Results + # header carries dkim=pass for one of these domains. + # Observed on live traffic 2026-07-15: Coupa PO mail passes + # DKIM for amazon.coupahost.com (and amazonses.com, which is + # deliberately NOT allowlisted — every SES customer's mail + # passes that). Unset/empty ⇒ the handler rejects all mail. + "ALLOWED_DKIM_DOMAINS": "amazon.coupahost.com", }, ) @@ -230,6 +309,20 @@ class PoIngestStack(Stack): treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING, ).add_alarm_action(cw_actions.SnsAction(alarm_topic)) + # --- Sender-auth rejection alarm (INFRA-107) --- + # A rejected email (bad/unaligned DKIM verdict) returns normally, so it + # produces NO Lambda error, NO DLQ message and NO retry -- only a + # `sender_auth_rejected` warning log. Without this metric filter + alarm a + # domain drift (Coupa rotates its signing subdomain, SES changes its + # Authentication-Results format, the allowlist is wrong) would silently + # discard 100% of legitimate PO mail while every other alarm stays green. + # A CloudWatch Logs metric filter turns those warnings into a metric so a + # false-reject storm pages instead of vanishing. default_value=0 keeps the + # series populated (alarm stays OK, never INSUFFICIENT_DATA) between events. + _add_sender_auth_rejected_alarm( + self, "EmailProcessor", "po-email-processor", alarm_topic + ) + # --- Throttles alarm: po-email-processor --- # Any throttled invocation (concurrency cap hit) in a 5-min window pages. # ALARM-only to site-alerts; no OK action; NOT_BREACHING when no data. @@ -316,6 +409,17 @@ class PoIngestStack(Stack): ], ) + # --- Web UI auth token secret --- + # Shared secret for the web UI auth gate, stored in Secrets Manager and + # resolved at runtime so the token never appears in CloudFormation templates + # or Lambda environment variables. Create this secret before deploying + # either stack; both PO and WO stacks reference it by name. + web_ui_auth_secret = secretsmanager.Secret.from_secret_name_v2( + self, + "WebUiAuthToken", + "procurement-ingest/web-ui-auth-token", + ) + # --- Web UI Lambda --- web_ui = lambda_.Function( self, @@ -330,10 +434,17 @@ class PoIngestStack(Stack): log_retention=logs.RetentionDays.TWO_MONTHS, environment={ "PO_TABLE": "purchase-orders", + # Defense-in-depth shared secret for the web UI handler. The + # handler fails closed if this ARN is unset or the secret is + # missing, so any future invocation path cannot re-expose the + # PO DB unauthenticated. The secret value is fetched at runtime + # from Secrets Manager (not embedded in env vars or template). + "WEB_UI_AUTH_TOKEN_SECRET_ARN": web_ui_auth_secret.secret_arn, }, ) po_table.grant_read_data(web_ui) + web_ui_auth_secret.grant_read(web_ui) # --- Throttles alarm: po-web-ui --- web_ui.metric_throttles( diff --git a/cdk/wo_stack.py b/cdk/wo_stack.py index 3081193..def3964 100644 --- a/cdk/wo_stack.py +++ b/cdk/wo_stack.py @@ -9,16 +9,15 @@ from aws_cdk import ( aws_cloudwatch_actions as cw_actions, aws_dynamodb as dynamodb, aws_iam as iam, - aws_kms as kms, aws_lambda as lambda_, aws_logs as logs, aws_s3 as s3, aws_s3_notifications as s3n, aws_ses as ses, aws_ses_actions as ses_actions, + aws_secretsmanager as secretsmanager, aws_sns as sns, aws_sqs as sqs, - aws_ssm as ssm, ) from constructs import Construct @@ -76,6 +75,79 @@ def _add_ddb_alarms(scope, id_prefix, table, alarm_name_prefix, alarm_topic): ).add_alarm_action(cw_actions.SnsAction(alarm_topic)) +# CloudWatch namespace for the log-derived sender-authentication metrics. +_SENDER_AUTH_METRIC_NAMESPACE = "Seahaven/ProcurementIngest" + + +def _add_sender_auth_rejected_alarm(scope, id_prefix, function_name, alarm_topic): + """Metric-filter + alarm on ``sender_auth_rejected`` warnings (INFRA-107). + + A rejected inbound email is skipped without erroring the invocation, so it + is invisible to the Errors/Throttles/DLQ alarms. This turns the structured + warning log into a CloudWatch metric and pages when rejections spike -- + catching a silent false-reject storm (allowlist wrong, signing-domain + drift, SES header-format change) that would otherwise discard legitimate + mail while the pipeline reports healthy. This is the safety net for the WO + allowlist domain assumption (seahaven.com) -- if the real Gmail-forward + re-signing domain differs, this alarm surfaces it instead of a silent + work-order outage. + + ALARM-only SnsAction to site-alerts; no OK action. The metric filter reads + the function's own log group (imported by the deterministic + ``/aws/lambda/`` name, created by the function's log_retention). A plain + substring pattern is used because Lambda prefixes each line with its own + level/timestamp/request-id, so the JSON payload is not a standalone JSON + log event a `{$.event=...}` pattern could match. + """ + metric_name = f"{function_name}-sender-auth-rejected" + logs.MetricFilter( + scope, + f"{id_prefix}SenderAuthRejectedFilter", + log_group=logs.LogGroup.from_log_group_name( + scope, + f"{id_prefix}LogGroup", + f"/aws/lambda/{function_name}", + ), + filter_pattern=logs.FilterPattern.literal('"sender_auth_rejected"'), + metric_namespace=_SENDER_AUTH_METRIC_NAMESPACE, + metric_name=metric_name, + metric_value="1", + default_value=0, + ) + + # Fire on a *sustained* reject condition rather than a volume spike. The + # earlier Sum>=3-over-15-min threshold had a blind spot that is exactly the + # failure this alarm exists to catch: a low-traffic pipeline in total + # drift outage (allowlist wrong / signing-domain changed) may only produce + # a trickle of rejects -- one every few minutes -- that never sums to 3 in + # any window, so the outage never pages. Instead: >=1 reject per 5-min + # period, alarming when 2 of the last 3 periods breach (evaluation_periods=3 + # / datapoints_to_alarm=2, the same idiom as the duration alarm). A single + # stray spoof probe (one lone period) is tolerated and self-clears, but a + # sustained reject condition trips within ~10-15 min even at one reject per + # period. default_value=0 on the metric filter keeps the series continuous + # so NOT_BREACHING only applies before the first datapoint ever arrives. + cloudwatch.Metric( + namespace=_SENDER_AUTH_METRIC_NAMESPACE, + metric_name=metric_name, + period=Duration.minutes(5), + statistic="Sum", + ).create_alarm( + scope, + f"{id_prefix}SenderAuthRejectedAlarm", + alarm_name=f"{function_name}-sender-auth-rejected", + alarm_description=( + f"{function_name} rejected inbound mail on sender authentication " + "(possible allowlist/DKIM-domain drift silently dropping real mail)" + ), + threshold=1, + evaluation_periods=3, + datapoints_to_alarm=2, + comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD, + treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING, + ).add_alarm_action(cw_actions.SnsAction(alarm_topic)) + + class WorkorderIngestStack(Stack): def __init__(self, scope: Construct, construct_id: str, **kwargs): super().__init__(scope, construct_id, **kwargs) @@ -183,6 +255,15 @@ class WorkorderIngestStack(Stack): "WORK_ORDERS_TABLE": work_orders_table.table_name, "COMMENTS_TABLE": comments_table.table_name, "BEDROCK_MODEL_ID": "us.anthropic.claude-haiku-4-5-20251001-v1:0", + # Fail-closed sender auth (INFRA-107): the handler only + # accepts mail whose SES-stamped Authentication-Results + # header carries dkim=pass for one of these domains. APM + # mail arrives via the apm@ Google Groups forward, which + # re-signs as seahaven.com (observed on live traffic + # 2026-07-15: "dkim=pass header.i=@seahaven.com"; the + # original hxgnsmartcloud.com signature does not survive + # the forward). Unset/empty ⇒ the handler rejects all mail. + "ALLOWED_DKIM_DOMAINS": "seahaven.com", }, ) @@ -213,21 +294,15 @@ class WorkorderIngestStack(Stack): ) ) - # Pre-emptive KMS grant on the shared DynamoDB CMK (alias/seahaven-dynamodb, - # /seahaven/dynamodb/cmk-arn). The WorkOrders/WorkOrderComments tables are - # NOT yet SSE-KMS encrypted, so this grant is currently unused; it is added - # ahead of the CMK migration so the processor role does not hit AccessDenied - # the moment those tables are migrated. The actual table migration + kebab - # rename is tracked in INFRA-6 (and the set-aside wo_stack CMK WIP); fold the - # web-ui read grant in there. - dynamodb_cmk = kms.Key.from_key_arn( - self, - "DynamoDbCmk", - ssm.StringParameter.value_for_string_parameter( - self, "/seahaven/dynamodb/cmk-arn" - ), - ) - dynamodb_cmk.grant_encrypt_decrypt(email_processor) + # NOTE: The pre-emptive grant_encrypt_decrypt on the shared DynamoDB CMK + # (alias/seahaven-dynamodb) was removed (security sweep 2026-06-17). The + # WorkOrders/WorkOrderComments tables are NOT SSE-KMS encrypted with that + # CMK, so the grant was unused for these tables yet handed + # wo-email-processor kms:Decrypt on the CMK that also protects the + # purchase-orders table (cross-stack decrypt reach). Re-add this grant only + # as part of the actual CMK migration of these tables (INFRA-6), at which + # point grant_read_write_data on the (then encrypted) tables would propagate + # the needed key permissions automatically. # --- Errors alarm (INFRA-41 / audit H-8) --- # ALARM-only (no OK action, per the CloudWatch-alarm preference) to the @@ -246,6 +321,19 @@ class WorkorderIngestStack(Stack): treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING, ).add_alarm_action(cw_actions.SnsAction(alarm_topic)) + # --- Sender-auth rejection alarm (INFRA-107) --- + # A rejected email (bad/unaligned DKIM verdict) returns normally, so it + # produces NO Lambda error, NO DLQ message and NO retry -- only a + # `sender_auth_rejected` warning log. The WO allowlist trusts dkim=pass + # for seahaven.com on the assumption the apm@ forward re-signs there; if + # that assumption is wrong (e.g. a Gmail auto-forward re-signs under a + # different domain), 100% of legitimate work-order mail is silently + # dropped. This metric filter + alarm turns those warnings into a paging + # signal so a false-reject storm surfaces instead of a silent outage. + _add_sender_auth_rejected_alarm( + self, "EmailProcessor", "workorder-email-processor", alarm_topic + ) + # --- Throttles alarm: workorder-email-processor --- # Any throttled invocation (concurrency cap hit) in a 5-min window pages. # ALARM-only to site-alerts; no OK action; NOT_BREACHING when no data. @@ -377,6 +465,17 @@ class WorkorderIngestStack(Stack): ], ) + # --- Web UI auth token secret --- + # Shared secret for the web UI auth gate, stored in Secrets Manager and + # resolved at runtime so the token never appears in CloudFormation templates + # or Lambda environment variables. Create this secret before deploying + # either stack; both PO and WO stacks reference it by name. + web_ui_auth_secret = secretsmanager.Secret.from_secret_name_v2( + self, + "WebUiAuthToken", + "procurement-ingest/web-ui-auth-token", + ) + # --- Web UI Lambda --- web_ui = lambda_.Function( self, @@ -392,11 +491,18 @@ class WorkorderIngestStack(Stack): environment={ "WORK_ORDERS_TABLE": work_orders_table.table_name, "COMMENTS_TABLE": comments_table.table_name, + # Defense-in-depth shared secret for the web UI handler. The + # handler fails closed if this ARN is unset or the secret is + # missing, so any future invocation path cannot re-expose the + # WO DB unauthenticated. The secret value is fetched at runtime + # from Secrets Manager (not embedded in env vars or template). + "WEB_UI_AUTH_TOKEN_SECRET_ARN": web_ui_auth_secret.secret_arn, }, ) work_orders_table.grant_read_data(web_ui) comments_table.grant_read_data(web_ui) + web_ui_auth_secret.grant_read(web_ui) # --- DynamoDB throttle + system-error alarms --- # ThrottledRequests / SystemErrors emit at TableName + Operation only diff --git a/lambdas/po/email_processor/handler.py b/lambdas/po/email_processor/handler.py index 169cadb..c8a85f8 100644 --- a/lambdas/po/email_processor/handler.py +++ b/lambdas/po/email_processor/handler.py @@ -16,6 +16,7 @@ from decimal import Decimal from email import policy import boto3 +from ses_auth import authenticate_inbound_email logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -29,6 +30,11 @@ BEDROCK_MODEL_ID = os.environ.get( "BEDROCK_MODEL_ID", "us.anthropic.claude-haiku-4-5-20251001-v1:0" ) +# "Cancelled" is a sticky, authoritative status: once a PO reaches it, a later +# new_po/revision may enrich other fields but must never move it back to a +# non-cancelled status. +CANCELLED_STATUS = "Cancelled" + EXTRACTION_PROMPT = """\ You are an email parser for a purchase order ingest pipeline. The emails are Coupa procurement platform notifications containing purchase order @@ -307,32 +313,132 @@ def enrich_parsed(parsed: dict, s3_key: str, email_subject: str): return parsed -def save_new_po(parsed: dict): - """Insert a new PO into DynamoDB. Skips if po_number already exists.""" +def _write_fields(po_number: str, fields: dict, *, guard_cancelled: bool): + """SET the given non-null fields on a PO record via update_item. + + Only the fields supplied are written; absent fields are left untouched, so a + partial payload can never delete data that an earlier email established. The + record is created if it does not exist (DynamoDB update_item upsert). + + When ``guard_cancelled`` is True the write carries a ConditionExpression that + only permits it while the record is not already Cancelled. The condition is + evaluated atomically by DynamoDB at write time, so a cancellation that lands + first always wins — there is no read-then-write TOCTOU window. A failed guard + raises ConditionalCheckFailedException for the caller to handle. + """ table = dynamodb.Table(PO_TABLE) - item = {k: v for k, v in parsed.items() if v is not None} + + set_parts = [] + attr_names = {} + attr_values = {} + for key, value in fields.items(): + if value is None or key == "po_number": + continue + name_ph = f"#{key}" + val_ph = f":{key}" + attr_names[name_ph] = key + attr_values[val_ph] = value + set_parts.append(f"{name_ph} = {val_ph}") + + if not set_parts: + return + + params = { + "Key": {"po_number": po_number}, + "UpdateExpression": "SET " + ", ".join(set_parts), + "ExpressionAttributeNames": attr_names, + "ExpressionAttributeValues": attr_values, + } + if guard_cancelled: + params["ExpressionAttributeValues"][":__cancelled_marker"] = CANCELLED_STATUS + params["ConditionExpression"] = ( + "attribute_not_exists(po_status) OR po_status <> :__cancelled_marker" + ) + + table.update_item(**params) + + +def _merge_update(po_number: str, fields: dict): + """Merge (SET-only) the given fields onto a PO record, keeping Cancelled sticky. + + Only the fields supplied are written; absent fields are left untouched. The + record is created if it does not exist (DynamoDB update_item upsert). + + "Cancelled" is a sticky, authoritative status. When the incoming payload + carries a non-cancelled ``po_status``, the write is guarded by a + ConditionExpression so the status is only applied while the record is not + already Cancelled — enforced atomically at write time, eliminating the + read-then-write TOCTOU where a concurrently-landing cancellation could be + silently un-cancelled. If the guard fails (the PO is already Cancelled), the + same fields are re-written WITHOUT po_status/cancelled_at and + unconditionally, so the other fields still merge while the Cancelled status + stays intact. + + A payload with no ``po_status``, or one whose status is already "Cancelled", + needs no guard — a plain merge is correct. This is what keeps legitimate + status updates (non-cancelled PO) and status-less revisions from ever being + dropped: the status is only ever suppressed on a true un-cancel transition. + """ + incoming_status = fields.get("po_status") + if incoming_status is None or incoming_status == CANCELLED_STATUS: + _write_fields(po_number, fields, guard_cancelled=False) + return try: - table.put_item( - Item=item, - ConditionExpression="attribute_not_exists(po_number)", - ) - logger.info(f"Created PO {parsed['po_number']}") + _write_fields(po_number, fields, guard_cancelled=True) except dynamodb.meta.client.exceptions.ConditionalCheckFailedException: - logger.info(f"PO {parsed['po_number']} already exists, skipping insert") + logger.info( + f"PO {po_number} is Cancelled; suppressing incoming " + f"po_status={incoming_status!r} and merging remaining fields" + ) + enrich_fields = { + k: v for k, v in fields.items() if k not in ("po_status", "cancelled_at") + } + _write_fields(po_number, enrich_fields, guard_cancelled=False) + + +def save_new_po(parsed: dict): + """Create a PO, merging into any pre-existing record. + + Uses a merge update rather than a conditional put so that an out-of-order + cancellation (which leaves a Cancelled skeleton) is filled in with the full + PO data instead of the new_po being silently dropped. "Cancelled" is a sticky + status enforced atomically inside _merge_update: if the PO was already + cancelled, the new_po backfills its remaining fields (supplier, line_items, + amounts) but never un-cancels it. + """ + po_number = parsed["po_number"] + fields = {k: v for k, v in parsed.items() if v is not None} + _merge_update(po_number, fields) + logger.info(f"Created/merged PO {po_number}") def save_revision(parsed: dict): - """Update an existing PO with revised data, or insert if it doesn't exist yet.""" - table = dynamodb.Table(PO_TABLE) - item = {k: v for k, v in parsed.items() if v is not None} + """Merge revised data into an existing PO without deleting omitted fields. - table.put_item(Item=item) - logger.info(f"Revised PO {parsed['po_number']}") + A revision email often omits unchanged sections (line_items, supplier). The + previous full-overwrite put_item permanently dropped those. This SETs only the + fields present in the revision, leaving everything else intact. + + "Cancelled" is a sticky status: a revision may enrich a cancelled PO's fields + but must never move it to a non-cancelled status. That invariant is enforced + atomically inside _merge_update and applies ONLY to the un-cancel transition — + a revision that carries no status change, or one targeting a non-cancelled PO, + updates po_status normally. + """ + po_number = parsed["po_number"] + fields = {k: v for k, v in parsed.items() if v is not None} + _merge_update(po_number, fields) + logger.info(f"Revised PO {po_number}") def save_cancellation(parsed: dict): - """Update an existing PO's status to Cancelled.""" + """Mark a PO Cancelled, creating a minimal skeleton if it doesn't exist yet. + + If the cancellation arrives before the new_po, the skeleton it creates is + later backfilled by save_new_po (which preserves this Cancelled status), so no + PO data is lost on out-of-order delivery. + """ table = dynamodb.Table(PO_TABLE) table.update_item( @@ -361,6 +467,13 @@ def handler(event, context): response = s3.get_object(Bucket=bucket, Key=key) raw_email = response["Body"].read() + # Fail-closed sender authentication (INFRA-107): only mail with an + # SES-stamped dkim=pass verdict for an allowlisted domain may create + # or update purchase orders. Rejected mail is logged and skipped + # without erroring the invocation (no retries / DLQ spam). + if not authenticate_inbound_email(raw_email, s3_key): + continue + email_data = parse_raw_email(raw_email) logger.info(f"Subject: {email_data['subject']}") diff --git a/lambdas/po/email_processor/ses_auth.py b/lambdas/po/email_processor/ses_auth.py new file mode 100644 index 0000000..db5445f --- /dev/null +++ b/lambdas/po/email_processor/ses_auth.py @@ -0,0 +1,429 @@ +"""Fail-closed SES sender authentication (INFRA-107). + +SES Email Receiving *prepends* its own trace headers -- including an +``Authentication-Results`` header whose authserv-id is ``amazonses.com`` -- +to the top of the raw MIME it writes to S3. Everything below those +prepended headers (the From header, any additional Authentication-Results +copies) is attacker-controlled, so ONLY the topmost Authentication-Results +header is trusted, and only when its authserv-id is ``amazonses.com``. + +An email is accepted only when that header carries ``dkim=pass`` for a +domain in the ``ALLOWED_DKIM_DOMAINS`` allowlist (a comma-separated Lambda +environment variable set by the CDK stack). Every other outcome fails +closed and the email is rejected: + +- ``ALLOWED_DKIM_DOMAINS`` unset or empty +- no Authentication-Results header at all +- topmost header unparseable or from an authserv-id other than SES +- no ``dkim=pass`` clause +- ``dkim=pass`` only for domains outside the allowlist + +Observed SES format (2026-07-15, both ingest buckets):: + + Authentication-Results: amazonses.com; + spf=pass (spfCheck: ...) client-ip=...; envelope-from=...; helo=...; + dkim=pass header.i=@seahaven.com; + dmarc=none header.from=hxgnsmartcloud.com; + +Note SES reports the passing DKIM identity as ``header.i=@`` +(RFC 6376 AUID), not ``header.d=``; the parser accepts both, and treats +``header.d`` (the plain signing domain) as authoritative over ``header.i`` +when both are present. The ``header.i`` domain is derived per RFC 6376 as +the part after the AUID's *last* ``@`` -- a ``@`` inside a quoted +local-part (e.g. ``i="@seahaven.com"@attacker.com``) is signer-controlled +label text, never the identity domain, and yields the true signer +(``attacker.com``). + +**Clause injection defence (INFRA-107 hardening).** SES echoes several +attacker-controlled SMTP-session tokens into its own Authentication-Results +value as their own semicolon-delimited property clauses -- notably +``envelope-from=``, ``helo=`` and ``header.from=``. An RFC 5321 quoted-local-part MAIL FROM may legally contain +spaces and semicolons, e.g.:: + + MAIL FROM:<"x; dkim=pass header.i=@amazon.coupahost.com"@attacker.com> + +which SES renders verbatim as ``envelope-from="x; dkim=pass +header.i=@amazon.coupahost.com"@attacker.com;``. A naive ``split(";")`` +would tear the quoted string apart and manufacture a synthetic +``dkim=pass header.i=@amazon.coupahost.com`` clause out of attacker input. +The parser therefore tokenises per RFC 8601 / RFC 5322 structure: CFWS +comments ``(...)`` are stripped first, and the value is split into clauses +only on semicolons that sit *outside* a quoted-string. A ``;`` inside a +quoted ``pvalue`` stays part of that one property clause and can never be +read as the start of a ``dkim=`` methodspec. +""" + +import email.parser +import email.policy +import json +import logging +import os +import re + +logger = logging.getLogger() + +ALLOWED_DKIM_DOMAINS_ENV = "ALLOWED_DKIM_DOMAINS" +SES_AUTHSERV_ID = "amazonses.com" + +# One resinfo clause of an Authentication-Results value, e.g. +# "dkim=pass header.i=@seahaven.com". The clause must START with the +# method=result pair; header.d= / header.i= may appear anywhere after it. +# The result token must be terminated by end-of-clause or whitespace so +# "dkim=pass-anything" can never be read as "pass". (Comments are stripped +# before matching, so the pre-strip "dkim=pass(comment)" form arrives here +# as "dkim=pass ..." and still terminates on whitespace.) +_DKIM_RESULT_RE = re.compile(r"^dkim\s*=\s*([a-z0-9]+)(?=$|\s)", re.IGNORECASE) + + +def get_allowed_dkim_domains() -> frozenset: + """Read the DKIM-domain allowlist from the environment (may be empty).""" + raw = os.environ.get(ALLOWED_DKIM_DOMAINS_ENV, "") + return frozenset( + d.strip().lower().lstrip("@").rstrip(".") for d in raw.split(",") if d.strip() + ) + + +def _unfold(value: str) -> str: + """Collapse RFC 5322 folding whitespace into single spaces.""" + return re.sub(r"[\r\n\t ]+", " ", value).strip() + + +def _strip_comments(value: str) -> tuple: + """Remove RFC 5322 CFWS comments ``(...)`` from an unfolded header value. + + Comments may nest and may contain quoted pairs (``\\)``). A ``(`` that + appears *inside* a quoted-string is literal text, not a comment start, + so quoted-strings (which carry attacker-controlled ``pvalue`` content + such as a quoted MAIL FROM local part) are passed through untouched. + Dropping comments first means comment-embedded ``header.i=`` / ``;`` + fakes can never influence clause splitting or domain extraction. + + Returns ``(stripped_text, well_formed)``. ``well_formed`` is False when + the value ends inside an unterminated comment or quoted-string, i.e. the + parens/quotes are unbalanced. Callers reject on ``not well_formed`` so a + malformed header (which could otherwise be mis-tokenised) fails closed + rather than being partially parsed. + """ + out = [] + depth = 0 # comment nesting depth + in_quote = False # inside a quoted-string (only tracked at depth 0) + i = 0 + n = len(value) + while i < n: + c = value[i] + if depth > 0: + # Inside a comment: only quoted-pairs and nested parens matter. + if c == "\\": + i += 2 + continue + if c == "(": + depth += 1 + elif c == ")": + depth -= 1 + i += 1 + continue + if in_quote: + out.append(c) + if c == "\\" and i + 1 < n: + out.append(value[i + 1]) + i += 2 + continue + if c == '"': + in_quote = False + i += 1 + continue + # Normal context (outside any comment or quoted-string). + if c == "(": + depth += 1 + i += 1 + continue + if c == '"': + in_quote = True + out.append(c) + i += 1 + well_formed = depth == 0 and not in_quote + return "".join(out), well_formed + + +def _split_clauses(value: str) -> list: + """Split a comment-free header value into clauses on top-level ``;``. + + A semicolon inside a quoted-string is preserved as part of the clause, + so an attacker-controlled quoted ``pvalue`` (e.g. a quoted MAIL FROM + echoed into ``envelope-from=``) cannot smuggle in a fake ``dkim=pass`` + clause. Callers must run :func:`_strip_comments` first. + """ + clauses = [] + buf = [] + in_quote = False + i = 0 + n = len(value) + while i < n: + c = value[i] + if in_quote: + buf.append(c) + if c == "\\" and i + 1 < n: + buf.append(value[i + 1]) + i += 2 + continue + if c == '"': + in_quote = False + i += 1 + continue + if c == '"': + in_quote = True + buf.append(c) + i += 1 + continue + if c == ";": + clauses.append("".join(buf)) + buf = [] + i += 1 + continue + buf.append(c) + i += 1 + clauses.append("".join(buf)) + return clauses + + +def _split_properties(clause: str) -> list: + """Split a clause into whitespace-separated ``name=pvalue`` tokens. + + Whitespace *inside* a quoted-string does not split, so an RFC 8601 pvalue + that embeds a quoted-string (e.g. a DKIM AUID with a quoted local-part that + legally contains spaces) survives as a single token. This is the same + quoted-string discipline :func:`_split_clauses` applies at the ``;`` level, + reused here at the token level so ``header.d=`` / ``header.i=`` extraction + is quoted-string-aware rather than a naive regex grab. + """ + tokens = [] + buf = [] + in_quote = False + i = 0 + n = len(clause) + while i < n: + c = clause[i] + if in_quote: + buf.append(c) + if c == "\\" and i + 1 < n: + buf.append(clause[i + 1]) + i += 2 + continue + if c == '"': + in_quote = False + i += 1 + continue + if c == '"': + in_quote = True + buf.append(c) + i += 1 + continue + if c.isspace(): + if buf: + tokens.append("".join(buf)) + buf = [] + i += 1 + continue + buf.append(c) + i += 1 + if buf: + tokens.append("".join(buf)) + return tokens + + +def _auid_domain(pvalue: str) -> str: + """Domain of a DKIM AUID (``header.i``) per RFC 6376. + + The identity domain is the part after the *last* ``@`` of the AUID -- but a + ``@`` inside a quoted local-part is NOT the identity separator. So + ``"@seahaven.com"@attacker.com`` yields ``attacker.com`` (the real signer), + not ``seahaven.com``: the ``@seahaven.com`` sits inside the quoted + local-part and is signer-controlled label text, never the domain. A bare + unquoted ``@seahaven.com`` still yields ``seahaven.com``. Returns "" when + there is no top-level ``@`` (no valid domain) or the domain looks malformed. + """ + last_at = -1 + in_quote = False + i = 0 + n = len(pvalue) + while i < n: + c = pvalue[i] + if in_quote: + if c == "\\": + i += 2 + continue + if c == '"': + in_quote = False + i += 1 + continue + if c == '"': + in_quote = True + elif c == "@": + last_at = i + i += 1 + if last_at < 0: + return "" + domain = pvalue[last_at + 1 :].lower().rstrip(".") + # A DKIM domain-name is a plain dot-atom; anything with a residual quote is + # malformed (or a smuggling attempt) and must not be trusted. + if not domain or '"' in domain: + return "" + return domain + + +def _plain_domain(pvalue: str) -> str: + """Domain of ``header.d`` -- the DKIM signing domain, a plain dot-atom. + + SES writes ``header.d`` verbatim from the signature's ``d=`` tag, which is + never a quoted-string. A residual quote means malformed/smuggled input and + fails closed. + """ + domain = pvalue.lower().rstrip(".") + if not domain or '"' in domain: + return "" + return domain + + +def _clause_signer_domain(clause: str) -> str: + """Authoritative DKIM signer domain of a ``dkim=pass`` clause, or "". + + ``header.d`` (the signing domain) is authoritative and is preferred when + present; only when it is absent does this fall back to ``header.i`` and + derive the domain from the AUID's post-final-``@`` part. Both lookups run + over :func:`_split_properties` tokens, so a ``header.d=``/``header.i=`` + literal smuggled *inside* another property's quoted pvalue is confined to + that one token and can never be read as a top-level property. + """ + header_d = None + header_i = None + for token in _split_properties(clause): + name, sep, val = token.partition("=") + if not sep: + continue + key = name.strip().lower() + if key == "header.d" and header_d is None: + header_d = val + elif key == "header.i" and header_i is None: + header_i = val + if header_d is not None: + return _plain_domain(header_d) + if header_i is not None: + return _auid_domain(header_i) + return "" + + +def parse_authentication_results(value: str) -> tuple: + """Parse one Authentication-Results header value. + + Returns ``(authserv_id, passing_dkim_domains)`` where the domains are the + authoritative signer domains of every ``dkim=pass`` clause -- ``header.d`` + when present, else the ``header.i`` AUID's post-final-``@`` domain (see + :func:`_clause_signer_domain`). Malformed input yields ``("", frozenset())``, + which callers treat as a rejection. + + Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322) at + the ``;`` (clause), whitespace (property) and ``@`` (AUID domain) levels, so + attacker-controlled tokens SES echoes into its header (envelope-from, helo, + header.from) cannot be split into a forged ``dkim=pass`` clause and a quoted + AUID local-part cannot masquerade as the identity domain. + """ + text, well_formed = _strip_comments(_unfold(value)) + if not well_formed: + # Unbalanced quotes/comments: refuse to guess how to tokenise it. + return "", frozenset() + clauses = [c.strip() for c in _split_clauses(text)] + if not clauses or not clauses[0]: + return "", frozenset() + + # First clause is the authserv-id, optionally followed by a version + # token ("amazonses.com 1"); take only the first token. + authserv_id = clauses[0].split()[0].strip('"').lower() + + passing = set() + for clause in clauses[1:]: + match = _DKIM_RESULT_RE.match(clause) + if not match or match.group(1).lower() != "pass": + continue + domain = _clause_signer_domain(clause) + if domain: + passing.add(domain) + return authserv_id, frozenset(passing) + + +def evaluate_sender_authentication(raw_email: bytes, allowed_domains) -> tuple: + """Evaluate the SES-stamped verdicts in a raw MIME message. + + Returns ``(accepted, reason, detail)``. Pure function of its inputs so + it can be unit-tested without touching the environment. + """ + if not allowed_domains: + return False, "allowlist_not_configured", {} + + try: + # compat32 keeps header values as raw strings (we unfold ourselves) + # and never raises on structurally odd headers; headersonly avoids + # parsing the body at all. + msg = email.parser.BytesParser(policy=email.policy.compat32).parsebytes( + raw_email, headersonly=True + ) + except Exception: + return False, "unparseable_message", {} + + ar_headers = msg.get_all("Authentication-Results") or [] + if not ar_headers: + return False, "authentication_results_missing", {} + + # SES prepends its trace headers, so index 0 is the SES-stamped copy. + # Any Authentication-Results header further down arrived inside the + # message (attacker-suppliable) and is deliberately ignored. + # + # The parse is wrapped so an unexpected parser exception fails CLOSED + # (rejected, structured reason) instead of propagating out of the handler + # into Lambda's async retries / DLQ on attacker-crafted input. + try: + authserv_id, passing = parse_authentication_results(str(ar_headers[0])) + except Exception: + return False, "authentication_results_unparseable", {} + detail = { + "authserv_id": authserv_id, + "passing_dkim_domains": sorted(passing), + } + + if authserv_id != SES_AUTHSERV_ID: + return False, "untrusted_authserv_id", detail + if not passing: + return False, "no_passing_dkim_signature", detail + + matched = passing & set(allowed_domains) + if not matched: + return False, "dkim_domain_not_allowlisted", detail + + detail["matched_domains"] = sorted(matched) + return True, "authenticated", detail + + +def authenticate_inbound_email(raw_email: bytes, s3_key: str) -> bool: + """Fail-closed gate used by the S3-triggered handlers. + + On rejection: logs a structured warning with the reason and S3 key and + returns False. Callers skip the message and return normally, so + rejected mail never errors the invocation (no retries, no DLQ spam). + """ + allowed = get_allowed_dkim_domains() + accepted, reason, detail = evaluate_sender_authentication(raw_email, allowed) + if accepted: + logger.info(json.dumps({"event": "sender_auth_ok", "s3_key": s3_key, **detail})) + return True + logger.warning( + json.dumps( + { + "event": "sender_auth_rejected", + "reason": reason, + "s3_key": s3_key, + "allowed_dkim_domains": sorted(allowed), + **detail, + } + ) + ) + return False diff --git a/lambdas/po/web_ui/handler.py b/lambdas/po/web_ui/handler.py index beaa174..e229a0d 100644 --- a/lambdas/po/web_ui/handler.py +++ b/lambdas/po/web_ui/handler.py @@ -5,17 +5,99 @@ Serves a simple HTML dashboard for viewing purchase orders. Accessed via Lambda Function URL. """ +import hmac import json +import logging import os +import time from decimal import Decimal from html import escape as esc import boto3 +logger = logging.getLogger() +logger.setLevel(logging.INFO) + dynamodb = boto3.resource("dynamodb") PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders") +# Defense-in-depth auth gate. The public Function URL was removed (INFRA-74), but +# the handler must still refuse unauthenticated requests so any future invocation +# path (re-attached Function URL, API Gateway, etc.) does not re-expose the whole +# PO DB. Callers must present the shared secret in the X-Auth-Token header (or +# Authorization: Bearer ). The secret is fetched at runtime from Secrets +# Manager to keep it out of CloudFormation templates and Lambda env vars. If the +# ARN is unset or the secret is missing, the handler fails closed and denies every +# request. +_WEB_UI_AUTH_TOKEN_SECRET_ARN = os.environ.get("WEB_UI_AUTH_TOKEN_SECRET_ARN") +# Refresh the cached token this often so a rotated secret propagates without +# waiting for the execution environment to recycle (emergency-rotation path). +_AUTH_TOKEN_CACHE_TTL_SECONDS = 300 +_auth_token_cache = None +_auth_token_cached_at = 0.0 + + +def _get_auth_token() -> str | None: + """Fetch the shared web UI auth token from Secrets Manager. + + Cached in the warm container for a short TTL so we don't hit Secrets Manager + on every request, while still picking up a rotated secret within the TTL + rather than only when the execution environment recycles. Returns None when + not configured or unreadable (the caller then fails closed). + """ + global _auth_token_cache, _auth_token_cached_at + now = time.monotonic() + if ( + _auth_token_cache is not None + and now - _auth_token_cached_at < _AUTH_TOKEN_CACHE_TTL_SECONDS + ): + return _auth_token_cache + if not _WEB_UI_AUTH_TOKEN_SECRET_ARN: + return None + secrets = boto3.client("secretsmanager") + try: + secret = secrets.get_secret_value(SecretId=_WEB_UI_AUTH_TOKEN_SECRET_ARN) + _auth_token_cache = secret["SecretString"] + _auth_token_cached_at = now + return _auth_token_cache + except Exception: + # Fail closed (return None -> caller 401s) but surface the failure: a + # Secrets Manager permission/config error would otherwise make every + # request 401 with no operational signal. The secret value is never + # logged. + logger.exception( + "Failed to fetch web UI auth token from Secrets Manager; " + "denying request (failing closed)" + ) + return None + + +def _header(event: dict, name: str) -> str: + """Case-insensitive header lookup from a Lambda Function URL / APIGW event.""" + headers = event.get("headers") or {} + name_lower = name.lower() + for key, value in headers.items(): + if key.lower() == name_lower: + return value or "" + return "" + + +def is_authenticated(event: dict) -> bool: + """Constant-time check of the request's shared secret against the configured + token. Fails closed when no token is configured.""" + token = _get_auth_token() + if not token: + return False + presented = _header(event, "x-auth-token") + if not presented: + auth = _header(event, "authorization") + if auth.lower().startswith("bearer "): + presented = auth[7:].strip() + if not presented: + return False + return hmac.compare_digest(presented, token) + def get_purchase_orders(limit=500): table = dynamodb.Table(PO_TABLE) @@ -60,7 +142,10 @@ def fmt_currency(val): val = float(val) if isinstance(val, (int, float)): return f"${val:,.2f}" - return str(val) + # Non-numeric fallback: a prompt-injected email can make Claude return + # total_amount/amount as an arbitrary string. Escape it before it is + # interpolated raw into the HTML to prevent stored XSS. + return esc(str(val)) def render_po_detail(po): @@ -250,6 +335,13 @@ def render_po_list(purchase_orders): def handler(event, context): + if not is_authenticated(event): + return { + "statusCode": 401, + "headers": {"Content-Type": "text/html"}, + "body": "

401 Unauthorized

", + } + path = event.get("rawPath", "/") qs = event.get("queryStringParameters") or {} diff --git a/lambdas/wo/email_processor/handler.py b/lambdas/wo/email_processor/handler.py index 4da7abd..de543ee 100644 --- a/lambdas/wo/email_processor/handler.py +++ b/lambdas/wo/email_processor/handler.py @@ -16,6 +16,7 @@ from datetime import datetime, timezone from email import policy import boto3 +from ses_auth import authenticate_inbound_email from template_parser import try_deterministic_parse @@ -285,13 +286,21 @@ def handler(event, context): for record in event.get("Records", []): bucket = record["s3"]["bucket"]["name"] key = record["s3"]["object"]["key"] + s3_key = f"s3://{bucket}/{key}" - logger.info(f"Processing email: s3://{bucket}/{key}") + logger.info(f"Processing email: {s3_key}") # Fetch raw email from S3 response = s3.get_object(Bucket=bucket, Key=key) raw_email = response["Body"].read() + # Fail-closed sender authentication (INFRA-107): only mail with an + # SES-stamped dkim=pass verdict for an allowlisted domain may create + # or update work orders. Rejected mail is logged and skipped without + # erroring the invocation (no retries / DLQ spam). + if not authenticate_inbound_email(raw_email, s3_key): + continue + # Parse the raw email email_data = parse_raw_email(raw_email) logger.info(f"Subject: {email_data['subject']}") @@ -313,8 +322,6 @@ def handler(event, context): logger.warning(f"No work order ID found in email, skipping: {key}") continue - s3_key = f"s3://{bucket}/{key}" - # Always upsert the work order with any new info save_work_order(parsed, s3_key) diff --git a/lambdas/wo/email_processor/ses_auth.py b/lambdas/wo/email_processor/ses_auth.py new file mode 100644 index 0000000..db5445f --- /dev/null +++ b/lambdas/wo/email_processor/ses_auth.py @@ -0,0 +1,429 @@ +"""Fail-closed SES sender authentication (INFRA-107). + +SES Email Receiving *prepends* its own trace headers -- including an +``Authentication-Results`` header whose authserv-id is ``amazonses.com`` -- +to the top of the raw MIME it writes to S3. Everything below those +prepended headers (the From header, any additional Authentication-Results +copies) is attacker-controlled, so ONLY the topmost Authentication-Results +header is trusted, and only when its authserv-id is ``amazonses.com``. + +An email is accepted only when that header carries ``dkim=pass`` for a +domain in the ``ALLOWED_DKIM_DOMAINS`` allowlist (a comma-separated Lambda +environment variable set by the CDK stack). Every other outcome fails +closed and the email is rejected: + +- ``ALLOWED_DKIM_DOMAINS`` unset or empty +- no Authentication-Results header at all +- topmost header unparseable or from an authserv-id other than SES +- no ``dkim=pass`` clause +- ``dkim=pass`` only for domains outside the allowlist + +Observed SES format (2026-07-15, both ingest buckets):: + + Authentication-Results: amazonses.com; + spf=pass (spfCheck: ...) client-ip=...; envelope-from=...; helo=...; + dkim=pass header.i=@seahaven.com; + dmarc=none header.from=hxgnsmartcloud.com; + +Note SES reports the passing DKIM identity as ``header.i=@`` +(RFC 6376 AUID), not ``header.d=``; the parser accepts both, and treats +``header.d`` (the plain signing domain) as authoritative over ``header.i`` +when both are present. The ``header.i`` domain is derived per RFC 6376 as +the part after the AUID's *last* ``@`` -- a ``@`` inside a quoted +local-part (e.g. ``i="@seahaven.com"@attacker.com``) is signer-controlled +label text, never the identity domain, and yields the true signer +(``attacker.com``). + +**Clause injection defence (INFRA-107 hardening).** SES echoes several +attacker-controlled SMTP-session tokens into its own Authentication-Results +value as their own semicolon-delimited property clauses -- notably +``envelope-from=``, ``helo=`` and ``header.from=``. An RFC 5321 quoted-local-part MAIL FROM may legally contain +spaces and semicolons, e.g.:: + + MAIL FROM:<"x; dkim=pass header.i=@amazon.coupahost.com"@attacker.com> + +which SES renders verbatim as ``envelope-from="x; dkim=pass +header.i=@amazon.coupahost.com"@attacker.com;``. A naive ``split(";")`` +would tear the quoted string apart and manufacture a synthetic +``dkim=pass header.i=@amazon.coupahost.com`` clause out of attacker input. +The parser therefore tokenises per RFC 8601 / RFC 5322 structure: CFWS +comments ``(...)`` are stripped first, and the value is split into clauses +only on semicolons that sit *outside* a quoted-string. A ``;`` inside a +quoted ``pvalue`` stays part of that one property clause and can never be +read as the start of a ``dkim=`` methodspec. +""" + +import email.parser +import email.policy +import json +import logging +import os +import re + +logger = logging.getLogger() + +ALLOWED_DKIM_DOMAINS_ENV = "ALLOWED_DKIM_DOMAINS" +SES_AUTHSERV_ID = "amazonses.com" + +# One resinfo clause of an Authentication-Results value, e.g. +# "dkim=pass header.i=@seahaven.com". The clause must START with the +# method=result pair; header.d= / header.i= may appear anywhere after it. +# The result token must be terminated by end-of-clause or whitespace so +# "dkim=pass-anything" can never be read as "pass". (Comments are stripped +# before matching, so the pre-strip "dkim=pass(comment)" form arrives here +# as "dkim=pass ..." and still terminates on whitespace.) +_DKIM_RESULT_RE = re.compile(r"^dkim\s*=\s*([a-z0-9]+)(?=$|\s)", re.IGNORECASE) + + +def get_allowed_dkim_domains() -> frozenset: + """Read the DKIM-domain allowlist from the environment (may be empty).""" + raw = os.environ.get(ALLOWED_DKIM_DOMAINS_ENV, "") + return frozenset( + d.strip().lower().lstrip("@").rstrip(".") for d in raw.split(",") if d.strip() + ) + + +def _unfold(value: str) -> str: + """Collapse RFC 5322 folding whitespace into single spaces.""" + return re.sub(r"[\r\n\t ]+", " ", value).strip() + + +def _strip_comments(value: str) -> tuple: + """Remove RFC 5322 CFWS comments ``(...)`` from an unfolded header value. + + Comments may nest and may contain quoted pairs (``\\)``). A ``(`` that + appears *inside* a quoted-string is literal text, not a comment start, + so quoted-strings (which carry attacker-controlled ``pvalue`` content + such as a quoted MAIL FROM local part) are passed through untouched. + Dropping comments first means comment-embedded ``header.i=`` / ``;`` + fakes can never influence clause splitting or domain extraction. + + Returns ``(stripped_text, well_formed)``. ``well_formed`` is False when + the value ends inside an unterminated comment or quoted-string, i.e. the + parens/quotes are unbalanced. Callers reject on ``not well_formed`` so a + malformed header (which could otherwise be mis-tokenised) fails closed + rather than being partially parsed. + """ + out = [] + depth = 0 # comment nesting depth + in_quote = False # inside a quoted-string (only tracked at depth 0) + i = 0 + n = len(value) + while i < n: + c = value[i] + if depth > 0: + # Inside a comment: only quoted-pairs and nested parens matter. + if c == "\\": + i += 2 + continue + if c == "(": + depth += 1 + elif c == ")": + depth -= 1 + i += 1 + continue + if in_quote: + out.append(c) + if c == "\\" and i + 1 < n: + out.append(value[i + 1]) + i += 2 + continue + if c == '"': + in_quote = False + i += 1 + continue + # Normal context (outside any comment or quoted-string). + if c == "(": + depth += 1 + i += 1 + continue + if c == '"': + in_quote = True + out.append(c) + i += 1 + well_formed = depth == 0 and not in_quote + return "".join(out), well_formed + + +def _split_clauses(value: str) -> list: + """Split a comment-free header value into clauses on top-level ``;``. + + A semicolon inside a quoted-string is preserved as part of the clause, + so an attacker-controlled quoted ``pvalue`` (e.g. a quoted MAIL FROM + echoed into ``envelope-from=``) cannot smuggle in a fake ``dkim=pass`` + clause. Callers must run :func:`_strip_comments` first. + """ + clauses = [] + buf = [] + in_quote = False + i = 0 + n = len(value) + while i < n: + c = value[i] + if in_quote: + buf.append(c) + if c == "\\" and i + 1 < n: + buf.append(value[i + 1]) + i += 2 + continue + if c == '"': + in_quote = False + i += 1 + continue + if c == '"': + in_quote = True + buf.append(c) + i += 1 + continue + if c == ";": + clauses.append("".join(buf)) + buf = [] + i += 1 + continue + buf.append(c) + i += 1 + clauses.append("".join(buf)) + return clauses + + +def _split_properties(clause: str) -> list: + """Split a clause into whitespace-separated ``name=pvalue`` tokens. + + Whitespace *inside* a quoted-string does not split, so an RFC 8601 pvalue + that embeds a quoted-string (e.g. a DKIM AUID with a quoted local-part that + legally contains spaces) survives as a single token. This is the same + quoted-string discipline :func:`_split_clauses` applies at the ``;`` level, + reused here at the token level so ``header.d=`` / ``header.i=`` extraction + is quoted-string-aware rather than a naive regex grab. + """ + tokens = [] + buf = [] + in_quote = False + i = 0 + n = len(clause) + while i < n: + c = clause[i] + if in_quote: + buf.append(c) + if c == "\\" and i + 1 < n: + buf.append(clause[i + 1]) + i += 2 + continue + if c == '"': + in_quote = False + i += 1 + continue + if c == '"': + in_quote = True + buf.append(c) + i += 1 + continue + if c.isspace(): + if buf: + tokens.append("".join(buf)) + buf = [] + i += 1 + continue + buf.append(c) + i += 1 + if buf: + tokens.append("".join(buf)) + return tokens + + +def _auid_domain(pvalue: str) -> str: + """Domain of a DKIM AUID (``header.i``) per RFC 6376. + + The identity domain is the part after the *last* ``@`` of the AUID -- but a + ``@`` inside a quoted local-part is NOT the identity separator. So + ``"@seahaven.com"@attacker.com`` yields ``attacker.com`` (the real signer), + not ``seahaven.com``: the ``@seahaven.com`` sits inside the quoted + local-part and is signer-controlled label text, never the domain. A bare + unquoted ``@seahaven.com`` still yields ``seahaven.com``. Returns "" when + there is no top-level ``@`` (no valid domain) or the domain looks malformed. + """ + last_at = -1 + in_quote = False + i = 0 + n = len(pvalue) + while i < n: + c = pvalue[i] + if in_quote: + if c == "\\": + i += 2 + continue + if c == '"': + in_quote = False + i += 1 + continue + if c == '"': + in_quote = True + elif c == "@": + last_at = i + i += 1 + if last_at < 0: + return "" + domain = pvalue[last_at + 1 :].lower().rstrip(".") + # A DKIM domain-name is a plain dot-atom; anything with a residual quote is + # malformed (or a smuggling attempt) and must not be trusted. + if not domain or '"' in domain: + return "" + return domain + + +def _plain_domain(pvalue: str) -> str: + """Domain of ``header.d`` -- the DKIM signing domain, a plain dot-atom. + + SES writes ``header.d`` verbatim from the signature's ``d=`` tag, which is + never a quoted-string. A residual quote means malformed/smuggled input and + fails closed. + """ + domain = pvalue.lower().rstrip(".") + if not domain or '"' in domain: + return "" + return domain + + +def _clause_signer_domain(clause: str) -> str: + """Authoritative DKIM signer domain of a ``dkim=pass`` clause, or "". + + ``header.d`` (the signing domain) is authoritative and is preferred when + present; only when it is absent does this fall back to ``header.i`` and + derive the domain from the AUID's post-final-``@`` part. Both lookups run + over :func:`_split_properties` tokens, so a ``header.d=``/``header.i=`` + literal smuggled *inside* another property's quoted pvalue is confined to + that one token and can never be read as a top-level property. + """ + header_d = None + header_i = None + for token in _split_properties(clause): + name, sep, val = token.partition("=") + if not sep: + continue + key = name.strip().lower() + if key == "header.d" and header_d is None: + header_d = val + elif key == "header.i" and header_i is None: + header_i = val + if header_d is not None: + return _plain_domain(header_d) + if header_i is not None: + return _auid_domain(header_i) + return "" + + +def parse_authentication_results(value: str) -> tuple: + """Parse one Authentication-Results header value. + + Returns ``(authserv_id, passing_dkim_domains)`` where the domains are the + authoritative signer domains of every ``dkim=pass`` clause -- ``header.d`` + when present, else the ``header.i`` AUID's post-final-``@`` domain (see + :func:`_clause_signer_domain`). Malformed input yields ``("", frozenset())``, + which callers treat as a rejection. + + Tokenisation is comment- and quoted-string-aware (RFC 8601 / RFC 5322) at + the ``;`` (clause), whitespace (property) and ``@`` (AUID domain) levels, so + attacker-controlled tokens SES echoes into its header (envelope-from, helo, + header.from) cannot be split into a forged ``dkim=pass`` clause and a quoted + AUID local-part cannot masquerade as the identity domain. + """ + text, well_formed = _strip_comments(_unfold(value)) + if not well_formed: + # Unbalanced quotes/comments: refuse to guess how to tokenise it. + return "", frozenset() + clauses = [c.strip() for c in _split_clauses(text)] + if not clauses or not clauses[0]: + return "", frozenset() + + # First clause is the authserv-id, optionally followed by a version + # token ("amazonses.com 1"); take only the first token. + authserv_id = clauses[0].split()[0].strip('"').lower() + + passing = set() + for clause in clauses[1:]: + match = _DKIM_RESULT_RE.match(clause) + if not match or match.group(1).lower() != "pass": + continue + domain = _clause_signer_domain(clause) + if domain: + passing.add(domain) + return authserv_id, frozenset(passing) + + +def evaluate_sender_authentication(raw_email: bytes, allowed_domains) -> tuple: + """Evaluate the SES-stamped verdicts in a raw MIME message. + + Returns ``(accepted, reason, detail)``. Pure function of its inputs so + it can be unit-tested without touching the environment. + """ + if not allowed_domains: + return False, "allowlist_not_configured", {} + + try: + # compat32 keeps header values as raw strings (we unfold ourselves) + # and never raises on structurally odd headers; headersonly avoids + # parsing the body at all. + msg = email.parser.BytesParser(policy=email.policy.compat32).parsebytes( + raw_email, headersonly=True + ) + except Exception: + return False, "unparseable_message", {} + + ar_headers = msg.get_all("Authentication-Results") or [] + if not ar_headers: + return False, "authentication_results_missing", {} + + # SES prepends its trace headers, so index 0 is the SES-stamped copy. + # Any Authentication-Results header further down arrived inside the + # message (attacker-suppliable) and is deliberately ignored. + # + # The parse is wrapped so an unexpected parser exception fails CLOSED + # (rejected, structured reason) instead of propagating out of the handler + # into Lambda's async retries / DLQ on attacker-crafted input. + try: + authserv_id, passing = parse_authentication_results(str(ar_headers[0])) + except Exception: + return False, "authentication_results_unparseable", {} + detail = { + "authserv_id": authserv_id, + "passing_dkim_domains": sorted(passing), + } + + if authserv_id != SES_AUTHSERV_ID: + return False, "untrusted_authserv_id", detail + if not passing: + return False, "no_passing_dkim_signature", detail + + matched = passing & set(allowed_domains) + if not matched: + return False, "dkim_domain_not_allowlisted", detail + + detail["matched_domains"] = sorted(matched) + return True, "authenticated", detail + + +def authenticate_inbound_email(raw_email: bytes, s3_key: str) -> bool: + """Fail-closed gate used by the S3-triggered handlers. + + On rejection: logs a structured warning with the reason and S3 key and + returns False. Callers skip the message and return normally, so + rejected mail never errors the invocation (no retries, no DLQ spam). + """ + allowed = get_allowed_dkim_domains() + accepted, reason, detail = evaluate_sender_authentication(raw_email, allowed) + if accepted: + logger.info(json.dumps({"event": "sender_auth_ok", "s3_key": s3_key, **detail})) + return True + logger.warning( + json.dumps( + { + "event": "sender_auth_rejected", + "reason": reason, + "s3_key": s3_key, + "allowed_dkim_domains": sorted(allowed), + **detail, + } + ) + ) + return False diff --git a/lambdas/wo/email_processor/tests/test_bedrock_fallback.py b/lambdas/wo/email_processor/tests/test_bedrock_fallback.py index 864bcdc..dfc7650 100644 --- a/lambdas/wo/email_processor/tests/test_bedrock_fallback.py +++ b/lambdas/wo/email_processor/tests/test_bedrock_fallback.py @@ -93,6 +93,11 @@ def test_template_path_skips_bedrock(fake_dynamo, metric_spy, monkeypatch): handler, "s3", FakeS3(_raw("update-plaintext", "update-plaintext-01")) ) monkeypatch.setattr(handler, "bedrock", fake_bedrock) + # This suite exercises parse dispatch, not the fail-closed SES sender-auth + # gate (INFRA-107) that now runs first in handler(); the scrubbed .eml + # fixtures carry no SES-stamped Authentication-Results header, so bypass it + # here. Authentication itself is covered by tests/test_ses_auth.py. + monkeypatch.setattr(handler, "authenticate_inbound_email", lambda *a: True) handler.handler(_event(), None) @@ -108,6 +113,9 @@ def test_fallback_path_invokes_bedrock(fake_dynamo, metric_spy, monkeypatch): fake_bedrock = FakeBedrock(AI_17_KEY) monkeypatch.setattr(handler, "s3", FakeS3(_raw("ai-fallback", "unknown-subject"))) monkeypatch.setattr(handler, "bedrock", fake_bedrock) + # See note in test_template_path_skips_bedrock: bypass the INFRA-107 sender + # auth gate so this dispatch test reaches the parse path. + monkeypatch.setattr(handler, "authenticate_inbound_email", lambda *a: True) handler.handler(_event(), None) diff --git a/lambdas/wo/web_ui/handler.py b/lambdas/wo/web_ui/handler.py index 4af9120..3e3c849 100644 --- a/lambdas/wo/web_ui/handler.py +++ b/lambdas/wo/web_ui/handler.py @@ -5,17 +5,98 @@ Serves a simple HTML dashboard for viewing work orders and comments. Accessed via Lambda Function URL. """ +import hmac import json +import logging import os +import time from html import escape as esc import boto3 +logger = logging.getLogger() +logger.setLevel(logging.INFO) + dynamodb = boto3.resource("dynamodb") WORK_ORDERS_TABLE = os.environ.get("WORK_ORDERS_TABLE", "WorkOrders") COMMENTS_TABLE = os.environ.get("COMMENTS_TABLE", "WorkOrderComments") +# Defense-in-depth auth gate. The public Function URL was removed (INFRA-74), but +# the handler must still refuse unauthenticated requests so any future invocation +# path does not re-expose the whole WO DB. Callers must present the shared secret +# in the X-Auth-Token header (or Authorization: Bearer ). The secret is +# fetched at runtime from Secrets Manager to keep it out of CloudFormation +# templates and Lambda env vars. If the ARN is unset or the secret is missing, the +# handler fails closed and denies every request. +_WEB_UI_AUTH_TOKEN_SECRET_ARN = os.environ.get("WEB_UI_AUTH_TOKEN_SECRET_ARN") +# Refresh the cached token this often so a rotated secret propagates without +# waiting for the execution environment to recycle (emergency-rotation path). +_AUTH_TOKEN_CACHE_TTL_SECONDS = 300 +_auth_token_cache = None +_auth_token_cached_at = 0.0 + + +def _get_auth_token() -> str | None: + """Fetch the shared web UI auth token from Secrets Manager. + + Cached in the warm container for a short TTL so we don't hit Secrets Manager + on every request, while still picking up a rotated secret within the TTL + rather than only when the execution environment recycles. Returns None when + not configured or unreadable (the caller then fails closed). + """ + global _auth_token_cache, _auth_token_cached_at + now = time.monotonic() + if ( + _auth_token_cache is not None + and now - _auth_token_cached_at < _AUTH_TOKEN_CACHE_TTL_SECONDS + ): + return _auth_token_cache + if not _WEB_UI_AUTH_TOKEN_SECRET_ARN: + return None + secrets = boto3.client("secretsmanager") + try: + secret = secrets.get_secret_value(SecretId=_WEB_UI_AUTH_TOKEN_SECRET_ARN) + _auth_token_cache = secret["SecretString"] + _auth_token_cached_at = now + return _auth_token_cache + except Exception: + # Fail closed (return None -> caller 401s) but surface the failure: a + # Secrets Manager permission/config error would otherwise make every + # request 401 with no operational signal. The secret value is never + # logged. + logger.exception( + "Failed to fetch web UI auth token from Secrets Manager; " + "denying request (failing closed)" + ) + return None + + +def _header(event: dict, name: str) -> str: + """Case-insensitive header lookup from a Lambda Function URL / APIGW event.""" + headers = event.get("headers") or {} + name_lower = name.lower() + for key, value in headers.items(): + if key.lower() == name_lower: + return value or "" + return "" + + +def is_authenticated(event: dict) -> bool: + """Constant-time check of the request's shared secret against the configured + token. Fails closed when no token is configured.""" + token = _get_auth_token() + if not token: + return False + presented = _header(event, "x-auth-token") + if not presented: + auth = _header(event, "authorization") + if auth.lower().startswith("bearer "): + presented = auth[7:].strip() + if not presented: + return False + return hmac.compare_digest(presented, token) + def get_work_orders(limit=500): table = dynamodb.Table(WORK_ORDERS_TABLE) @@ -228,6 +309,13 @@ def render_work_orders_list(work_orders): def handler(event, context): + if not is_authenticated(event): + return { + "statusCode": 401, + "headers": {"Content-Type": "text/html"}, + "body": "

401 Unauthorized

", + } + path = event.get("rawPath", "/") qs = event.get("queryStringParameters") or {} diff --git a/tests/conftest.py b/tests/conftest.py index 5dfc8e2..00674bc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,11 +26,20 @@ def load_handler(relative_path, module_name): The handler files all share the basename ``handler.py`` and are not importable as packages, so a plain ``import handler`` would collide - across Lambdas. + across Lambdas. The same loader serves the ``ses_auth.py`` modules, + which are likewise duplicated per pipeline and not importable as + packages. + + The Lambda runtime puts each function's own directory on ``sys.path``, + so handler modules import their siblings by bare name (e.g. + ``from ses_auth import authenticate_inbound_email``). The function + directory is added to ``sys.path`` here so the module executes the + same way under test as it does in the Lambda. """ path = REPO_ROOT / relative_path # Put the handler's own directory on sys.path so its sibling-module imports - # (e.g. the WO handler's ``from template_parser import ...``) resolve. + # (e.g. the WO handler's ``from template_parser import ...`` and the + # ``from ses_auth import ...`` in both handlers) resolve. handler_dir = str(path.parent) if handler_dir not in sys.path: sys.path.insert(0, handler_dir) @@ -63,3 +72,13 @@ def wo_handler(): def email_handler(request): """Parametrized fixture yielding each email processor handler module.""" return request.getfixturevalue(request.param) + + +@pytest.fixture(params=["wo", "po"]) +def ses_auth(request): + """The ses_auth module of each pipeline (duplicated file, kept in sync).""" + pipeline = request.param + return load_handler( + f"lambdas/{pipeline}/email_processor/ses_auth.py", + f"{pipeline}_ses_auth", + ) diff --git a/tests/requirements.txt b/tests/requirements.txt new file mode 100644 index 0000000..021a495 --- /dev/null +++ b/tests/requirements.txt @@ -0,0 +1 @@ +moto>=5.0.0 diff --git a/tests/test_po_merge.py b/tests/test_po_merge.py new file mode 100644 index 0000000..26201c9 --- /dev/null +++ b/tests/test_po_merge.py @@ -0,0 +1,163 @@ +"""Tests for the sticky-Cancelled merge semantics in the PO email processor. + +These exercise the atomic conditional write in ``_merge_update`` that makes +"Cancelled" a sticky, authoritative status: once a PO is Cancelled, a later +new_po or revision may enrich its other fields but can never move it back to a +non-cancelled status. The invariant is enforced server-side with a DynamoDB +ConditionExpression, so it holds even under concurrent email processing — there +is no read-then-write TOCTOU window. + +A real in-memory purchase-orders table is stood up with moto so the tests +exercise the actual DynamoDB conditional-write and ConditionalCheckFailedException +fallback path, not a stub. +""" + +import boto3 +import pytest +from moto import mock_aws + +TABLE_NAME = "purchase-orders" + + +@pytest.fixture +def po_table(): + """Stand up a mock purchase-orders table matching the real key schema.""" + with mock_aws(): + resource = boto3.resource("dynamodb", region_name="us-east-1") + table = resource.create_table( + TableName=TABLE_NAME, + KeySchema=[{"AttributeName": "po_number", "KeyType": "HASH"}], + AttributeDefinitions=[ + {"AttributeName": "po_number", "AttributeType": "S"}, + ], + BillingMode="PAY_PER_REQUEST", + ) + table.wait_until_exists() + yield table + + +def _item(table, po_number): + return table.get_item(Key={"po_number": po_number}).get("Item") + + +def test_revision_cannot_uncancel_but_enriches(po_handler, po_table): + """(a) Revision carrying a live status onto a Cancelled PO keeps it Cancelled + while still writing the other revision fields (no resurrection).""" + po_table.put_item( + Item={ + "po_number": "PO-A", + "po_status": "Cancelled", + "cancelled_at": "2026-07-01T00:00:00", + } + ) + + po_handler.save_revision( + { + "po_number": "PO-A", + "po_status": "Issued", + "total_amount": 500, + "supplier": {"name": "Acme"}, + } + ) + + item = _item(po_table, "PO-A") + assert item["po_status"] == "Cancelled" # not resurrected + assert item["cancelled_at"] == "2026-07-01T00:00:00" # marker preserved + assert item["total_amount"] == 500 # other fields still enriched + assert item["supplier"] == {"name": "Acme"} + + +def test_revision_without_status_merges_on_cancelled(po_handler, po_table): + """(b) Revision with NO po_status onto a Cancelled PO merges cleanly, no + exception, status stays Cancelled.""" + po_table.put_item(Item={"po_number": "PO-B", "po_status": "Cancelled"}) + + po_handler.save_revision({"po_number": "PO-B", "total_amount": 250}) + + item = _item(po_table, "PO-B") + assert item["po_status"] == "Cancelled" + assert item["total_amount"] == 250 + + +def test_revision_updates_status_on_non_cancelled(po_handler, po_table): + """(c) Issue B: a legitimate status update on a NON-cancelled PO is applied, + never dropped.""" + po_table.put_item(Item={"po_number": "PO-C", "po_status": "Issued"}) + + po_handler.save_revision( + {"po_number": "PO-C", "po_status": "Revised", "total_amount": 999} + ) + + item = _item(po_table, "PO-C") + assert item["po_status"] == "Revised" + assert item["total_amount"] == 999 + + +def test_new_po_backfills_cancelled_skeleton(po_handler, po_table): + """(d) new_po onto a Cancelled skeleton backfills its fields but keeps the + Cancelled status and cancellation marker.""" + po_table.put_item( + Item={ + "po_number": "PO-D", + "po_status": "Cancelled", + "cancelled_at": "2026-07-02T00:00:00", + } + ) + + po_handler.save_new_po( + { + "po_number": "PO-D", + "po_status": "Issued", + "supplier": {"name": "Beta"}, + "total_amount": 1200, + } + ) + + item = _item(po_table, "PO-D") + assert item["po_status"] == "Cancelled" + assert item["cancelled_at"] == "2026-07-02T00:00:00" + assert item["supplier"] == {"name": "Beta"} + assert item["total_amount"] == 1200 + + +def test_new_po_creates_fresh(po_handler, po_table): + """(e) new_po on a fresh PO creates it normally with its status.""" + po_handler.save_new_po( + {"po_number": "PO-E", "po_status": "Issued", "total_amount": 42} + ) + + item = _item(po_table, "PO-E") + assert item["po_status"] == "Issued" + assert item["total_amount"] == 42 + + +def test_revision_creates_fresh_merge(po_handler, po_table): + """(e) revision on a fresh PO merges/creates normally with its status.""" + po_handler.save_revision( + {"po_number": "PO-F", "po_status": "Revised", "total_amount": 77} + ) + + item = _item(po_table, "PO-F") + assert item["po_status"] == "Revised" + assert item["total_amount"] == 77 + + +def test_save_cancellation_is_authoritative(po_handler, po_table): + """Cancellation always wins: save_cancellation writes Cancelled over a live + status while leaving previously-established data intact.""" + po_table.put_item( + Item={"po_number": "PO-G", "po_status": "Issued", "total_amount": 300} + ) + + po_handler.save_cancellation( + { + "po_number": "PO-G", + "processed_at": "2026-07-03T00:00:00", + "raw_s3_key": "s3://bucket/key", + } + ) + + item = _item(po_table, "PO-G") + assert item["po_status"] == "Cancelled" + assert item["cancelled_at"] == "2026-07-03T00:00:00" + assert item["total_amount"] == 300 # pre-existing data retained diff --git a/tests/test_ses_auth.py b/tests/test_ses_auth.py new file mode 100644 index 0000000..69bfa73 --- /dev/null +++ b/tests/test_ses_auth.py @@ -0,0 +1,391 @@ +"""Unit tests for the fail-closed SES sender authentication (INFRA-107). + +Fixtures mirror real SES-stamped headers observed on the two ingest +buckets on 2026-07-15: SES prepends a folded Authentication-Results +header with authserv-id amazonses.com and reports passing signers as +``dkim=pass header.i=@``. +""" + +BODY = "\r\n\r\nWork Order 12345 assigned.\r\n" + +# Folded exactly like real SES output (continuation lines, header.i form). +WO_SES_HEADER = ( + "Authentication-Results: amazonses.com;\r\n" + " spf=pass (spfCheck: domain of seahaven.com designates 209.85.219.70 as" + " permitted sender) client-ip=209.85.219.70;" + " envelope-from=apm+bnc@seahaven.com; helo=mail-qv1-f70.google.com;\r\n" + " dkim=pass header.i=@seahaven.com;\r\n" + " dmarc=none header.from=hxgnsmartcloud.com;\r\n" +) + +# Real PO traffic carries two dkim=pass clauses; amazonses.com must not be +# sufficient on its own (every SES customer's mail passes for it). +PO_SES_HEADER = ( + "Authentication-Results: amazonses.com;\r\n" + " spf=pass (spfCheck: domain of mail.coupahost.com designates" + " 54.240.41.238 as permitted sender) client-ip=54.240.41.238;\r\n" + " dkim=pass header.i=@amazonses.com;\r\n" + " dkim=pass header.i=@amazon.coupahost.com;\r\n" + " dmarc=pass header.from=amazon.coupahost.com;\r\n" +) + +FROM_TO = ( + "From: APM \r\n" + "To: apm@int.seahaven.com\r\n" + "Subject: WO 12345\r\n" +) + + +def raw(*headers: str) -> bytes: + return ("".join(headers) + FROM_TO + BODY).encode() + + +class TestParseAuthenticationResults: + def test_ses_wo_header(self, ses_auth): + value = WO_SES_HEADER.split(":", 1)[1] + authserv_id, passing = ses_auth.parse_authentication_results(value) + assert authserv_id == "amazonses.com" + assert passing == frozenset({"seahaven.com"}) + + def test_ses_po_header_multiple_dkim_clauses(self, ses_auth): + value = PO_SES_HEADER.split(":", 1)[1] + authserv_id, passing = ses_auth.parse_authentication_results(value) + assert authserv_id == "amazonses.com" + assert passing == frozenset({"amazonses.com", "amazon.coupahost.com"}) + + def test_header_d_form(self, ses_auth): + _, passing = ses_auth.parse_authentication_results( + "amazonses.com; dkim=pass header.d=Example.COM." + ) + assert passing == frozenset({"example.com"}) + + def test_case_insensitive_result(self, ses_auth): + _, passing = ses_auth.parse_authentication_results( + "amazonses.com; DKIM=Pass HEADER.I=@SeaHaven.COM" + ) + assert passing == frozenset({"seahaven.com"}) + + def test_dkim_fail_yields_no_domains(self, ses_auth): + _, passing = ses_auth.parse_authentication_results( + "amazonses.com; dkim=fail header.i=@seahaven.com" + ) + assert passing == frozenset() + + def test_authserv_id_version_token(self, ses_auth): + authserv_id, _ = ses_auth.parse_authentication_results( + "amazonses.com 1; dkim=pass header.i=@seahaven.com" + ) + assert authserv_id == "amazonses.com" + + def test_garbage_value(self, ses_auth): + authserv_id, passing = ses_auth.parse_authentication_results(";;;") + assert authserv_id == "" + assert passing == frozenset() + + def test_result_token_boundary(self, ses_auth): + # "pass-anything" / "passfail" must never be read as "pass". + for result in ("pass-fake", "passfail"): + _, passing = ses_auth.parse_authentication_results( + f"amazonses.com; dkim={result} header.i=@seahaven.com" + ) + assert passing == frozenset() + + def test_result_followed_by_comment(self, ses_auth): + _, passing = ses_auth.parse_authentication_results( + "amazonses.com; dkim=pass(good signature) header.i=@seahaven.com" + ) + assert passing == frozenset({"seahaven.com"}) + + def test_quoted_auid_localpart_is_not_the_domain(self, ses_auth): + # header.i="@seahaven.com" is a *quoted local-part* with no top-level + # "@domain" -- per RFC 6376 there is no identity domain, so it must + # yield nothing rather than mistaking the quoted label for the domain. + _, passing = ses_auth.parse_authentication_results( + 'amazonses.com; dkim=pass header.i="@seahaven.com"' + ) + assert passing == frozenset() + + def test_quoted_auid_localpart_attack_yields_true_signer(self, ses_auth): + # The core INFRA-107 defect: a signer with their own valid DKIM key for + # attacker.com sets AUID i="@seahaven.com"@attacker.com (RFC 6376-legal: + # the quoted local-part contains an "@"). The identity domain is the + # part after the LAST top-level "@" -> attacker.com, NOT seahaven.com. + _, passing = ses_auth.parse_authentication_results( + 'amazonses.com; dkim=pass header.i="@seahaven.com"@attacker.com' + ) + assert passing == frozenset({"attacker.com"}) + assert "seahaven.com" not in passing + + def test_quoted_auid_attack_rejected_end_to_end(self, ses_auth): + # Same attack through the full evaluator against a seahaven.com + # allowlist: the genuine dkim=pass binds to attacker.com, so the forged + # mail must fail closed, not be accepted for seahaven.com. + forged = ( + "Authentication-Results: amazonses.com;\r\n" + ' dkim=pass header.i="@seahaven.com"@attacker.com;\r\n' + ) + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(forged), {"seahaven.com"} + ) + assert not accepted + assert reason == "dkim_domain_not_allowlisted" + + def test_header_d_takes_precedence_over_mismatched_header_i(self, ses_auth): + # header.d is the authoritative signing domain; when both appear it + # wins over header.i, whichever order SES emits them in. + for clause in ( + "dkim=pass header.i=@evil.com header.d=seahaven.com", + "dkim=pass header.d=seahaven.com header.i=@evil.com", + ): + _, passing = ses_auth.parse_authentication_results( + f"amazonses.com; {clause}" + ) + assert passing == frozenset({"seahaven.com"}) + + def test_header_d_smuggled_in_quoted_header_i_is_not_top_level(self, ses_auth): + # A "header.d=seahaven.com" literal hidden inside header.i's quoted + # local-part must not be read as a top-level header.d property; the real + # (top-level) header.d=attacker.com is authoritative. + _, passing = ses_auth.parse_authentication_results( + 'amazonses.com; dkim=pass header.i="header.d=seahaven.com x"@attacker.com' + " header.d=attacker.com" + ) + assert passing == frozenset({"attacker.com"}) + + def test_legitimate_unquoted_header_i_still_passes(self, ses_auth): + _, passing = ses_auth.parse_authentication_results( + "amazonses.com; dkim=pass header.i=@seahaven.com" + ) + assert passing == frozenset({"seahaven.com"}) + + def test_folding_inside_dkim_clause(self, ses_auth): + _, passing = ses_auth.parse_authentication_results( + "amazonses.com;\r\n dkim=pass\r\n header.i=@seahaven.com" + ) + assert passing == frozenset({"seahaven.com"}) + + def test_quoted_semicolon_in_envelope_from_is_not_split(self, ses_auth): + # RFC 5321 quoted-local-part MAIL FROM can carry ';' and spaces; SES + # echoes it into envelope-from=. The ';' inside the quotes must stay + # part of the one envelope-from clause, never a synthetic dkim clause. + value = ( + "amazonses.com; spf=pass client-ip=1.2.3.4;" + ' envelope-from="x; dkim=pass header.i=@amazon.coupahost.com"@attacker.com;' + " helo=mail.attacker.com; dkim=fail header.i=@attacker.com;" + ) + _, passing = ses_auth.parse_authentication_results(value) + assert passing == frozenset() + + def test_quoted_pair_in_envelope_from_is_not_split(self, ses_auth): + # A quoted-pair (\") inside the quoted local part must not prematurely + # end the quoted-string and re-expose the smuggled tokens. + value = ( + 'amazonses.com; envelope-from="a\\"; dkim=pass header.d=seahaven.com"@evil.com;' + " dkim=fail header.i=@evil.com;" + ) + _, passing = ses_auth.parse_authentication_results(value) + assert passing == frozenset() + + def test_helo_injection_is_not_split(self, ses_auth): + # helo= is attacker-influenced too; a crafted value quoting a fake + # methodspec must not manufacture a passing clause. + value = ( + 'amazonses.com; helo="h; dkim=pass header.i=@seahaven.com";' + " dkim=fail header.i=@attacker.com;" + ) + _, passing = ses_auth.parse_authentication_results(value) + assert passing == frozenset() + + def test_comment_embedded_header_i_is_ignored(self, ses_auth): + # A CFWS comment carrying a fake header.i must be stripped before + # domain extraction, so only the real (failing) result is considered. + _, passing = ses_auth.parse_authentication_results( + "amazonses.com; dkim=fail (header.i=@seahaven.com) header.i=@attacker.com" + ) + assert passing == frozenset() + + def test_comment_hiding_semicolon_does_not_split(self, ses_auth): + # A ';' inside a comment is not a clause separator either. + _, passing = ses_auth.parse_authentication_results( + "amazonses.com; spf=pass (note: a; b) client-ip=1.2.3.4;" + " dkim=pass header.i=@seahaven.com" + ) + assert passing == frozenset({"seahaven.com"}) + + def test_unbalanced_quote_fails_closed(self, ses_auth): + # An unterminated quoted-string is malformed; refuse to parse it so a + # dkim=pass clause "swallowed" by the runaway quote can't be salvaged + # (and, conversely, a runaway quote can't be used to mis-tokenise). + authserv_id, passing = ses_auth.parse_authentication_results( + 'amazonses.com; envelope-from="oops@attacker.com;' + " dkim=pass header.i=@seahaven.com" + ) + assert authserv_id == "" + assert passing == frozenset() + + def test_unbalanced_comment_fails_closed(self, ses_auth): + # An unterminated comment is malformed and must fail closed. + authserv_id, passing = ses_auth.parse_authentication_results( + "amazonses.com; dkim=pass header.i=@seahaven.com (runaway comment" + ) + assert authserv_id == "" + assert passing == frozenset() + + +class TestEvaluateSenderAuthentication: + def test_ses_stamped_pass_accepted(self, ses_auth): + accepted, reason, detail = ses_auth.evaluate_sender_authentication( + raw(WO_SES_HEADER), {"seahaven.com"} + ) + assert accepted + assert reason == "authenticated" + assert detail["matched_domains"] == ["seahaven.com"] + + def test_po_pass_accepted_on_coupa_domain(self, ses_auth): + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(PO_SES_HEADER), {"amazon.coupahost.com"} + ) + assert accepted + assert reason == "authenticated" + + def test_amazonses_identity_alone_is_not_allowlisted(self, ses_auth): + # Any SES customer's outbound mail passes DKIM for amazonses.com, + # so a pass for it must not satisfy a coupahost-only allowlist. + forged_via_ses = ( + "Authentication-Results: amazonses.com;\r\n" + " spf=pass client-ip=54.240.41.1;\r\n" + " dkim=pass header.i=@amazonses.com;\r\n" + ) + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(forged_via_ses), {"amazon.coupahost.com"} + ) + assert not accepted + assert reason == "dkim_domain_not_allowlisted" + + def test_forged_ar_below_failing_ses_header_rejected(self, ses_auth): + # SES's (topmost) header says dkim=fail; the attacker smuggled a + # perfect-looking AR header inside the message. Only the topmost + # header may be consulted. + ses_fail = ( + "Authentication-Results: amazonses.com;\r\n" + " spf=fail client-ip=203.0.113.7;\r\n" + " dkim=fail header.i=@seahaven.com;\r\n" + " dmarc=fail header.from=hxgnsmartcloud.com;\r\n" + ) + forged = ( + "Authentication-Results: amazonses.com;\r\n" + " dkim=pass header.i=@seahaven.com;\r\n" + ) + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(ses_fail, forged), {"seahaven.com"} + ) + assert not accepted + assert reason == "no_passing_dkim_signature" + + def test_missing_ar_header_rejected(self, ses_auth): + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(), {"seahaven.com"} + ) + assert not accepted + assert reason == "authentication_results_missing" + + def test_untrusted_authserv_id_rejected(self, ses_auth): + attacker_ar = ( + "Authentication-Results: mail.attacker.example;\r\n" + " dkim=pass header.i=@seahaven.com;\r\n" + ) + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(attacker_ar), {"seahaven.com"} + ) + assert not accepted + assert reason == "untrusted_authserv_id" + + def test_unaligned_domain_rejected(self, ses_auth): + evil = ( + "Authentication-Results: amazonses.com;\r\n" + " dkim=pass header.i=@evil.example.com;\r\n" + ) + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(evil), {"seahaven.com"} + ) + assert not accepted + assert reason == "dkim_domain_not_allowlisted" + + def test_lookalike_domain_rejected(self, ses_auth): + # Substring containment must not match: notseahaven.com != seahaven.com + lookalike = ( + "Authentication-Results: amazonses.com;\r\n" + " dkim=pass header.i=@notseahaven.com;\r\n" + ) + accepted, _, _ = ses_auth.evaluate_sender_authentication( + raw(lookalike), {"seahaven.com"} + ) + assert not accepted + + def test_empty_allowlist_fails_closed(self, ses_auth): + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(WO_SES_HEADER), frozenset() + ) + assert not accepted + assert reason == "allowlist_not_configured" + + def test_parser_exception_fails_closed(self, ses_auth, monkeypatch): + # The parse of the topmost AR header runs outside the BytesParser + # try/except; an unexpected parser exception on crafted input must fail + # CLOSED (rejected, structured reason) rather than propagate out of the + # handler into Lambda's async retries / DLQ. + def boom(_value): + raise ValueError("simulated parser blow-up") + + monkeypatch.setattr(ses_auth, "parse_authentication_results", boom) + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(WO_SES_HEADER), {"seahaven.com"} + ) + assert not accepted + assert reason == "authentication_results_unparseable" + + def test_envelope_from_clause_injection_rejected(self, ses_auth): + # Full forged email: attacker's quoted MAIL FROM is echoed by SES into + # its own (topmost, genuinely SES-stamped) Authentication-Results as + # envelope-from=, trying to smuggle a dkim=pass for an allowlisted + # domain. The real DKIM verdict is fail. Must fail closed. + injected = ( + "Authentication-Results: amazonses.com;\r\n" + " spf=pass (spfCheck: domain of attacker.com designates 1.2.3.4 as" + " permitted sender) client-ip=1.2.3.4;\r\n" + ' envelope-from="x; dkim=pass header.i=@amazon.coupahost.com"@attacker.com;' + " helo=mail.attacker.com;\r\n" + " dkim=fail header.i=@attacker.com;\r\n" + " dmarc=fail header.from=attacker.com;\r\n" + ) + accepted, reason, _ = ses_auth.evaluate_sender_authentication( + raw(injected), {"amazon.coupahost.com", "seahaven.com"} + ) + assert not accepted + assert reason == "no_passing_dkim_signature" + + +class TestAuthenticateInboundEmail: + S3_KEY = "s3://bucket/inbound/abc123" + + def test_accepts_with_configured_allowlist(self, ses_auth, monkeypatch): + monkeypatch.setenv("ALLOWED_DKIM_DOMAINS", "seahaven.com") + assert ses_auth.authenticate_inbound_email(raw(WO_SES_HEADER), self.S3_KEY) + + def test_allowlist_is_comma_separated_and_normalized(self, ses_auth, monkeypatch): + # Stray spaces, case, a leading @, and a trailing dot all normalize. + monkeypatch.setenv("ALLOWED_DKIM_DOMAINS", " Other.Example , @SEAHAVEN.com. ,") + assert ses_auth.authenticate_inbound_email(raw(WO_SES_HEADER), self.S3_KEY) + + def test_env_var_unset_fails_closed(self, ses_auth, monkeypatch, caplog): + monkeypatch.delenv("ALLOWED_DKIM_DOMAINS", raising=False) + assert not ses_auth.authenticate_inbound_email(raw(WO_SES_HEADER), self.S3_KEY) + assert "allowlist_not_configured" in caplog.text + assert self.S3_KEY in caplog.text + + def test_rejection_logs_reason_and_key(self, ses_auth, monkeypatch, caplog): + monkeypatch.setenv("ALLOWED_DKIM_DOMAINS", "seahaven.com") + assert not ses_auth.authenticate_inbound_email(raw(), self.S3_KEY) + assert "sender_auth_rejected" in caplog.text + assert "authentication_results_missing" in caplog.text + assert self.S3_KEY in caplog.text