diff --git a/README.md b/README.md index 6922d49..0254bd8 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 extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year). -5. Merge write to DynamoDB: +4. Fail-closed sender authentication (INFRA-107): the SES-stamped `Authentication-Results` header must show `dkim=pass` for `amazon.coupahost.com` (see [Sender authentication](#sender-authentication-infra-107)); otherwise the email is logged and dropped. +5. Claude extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year). +6. Merge write to DynamoDB: - `new_po` — merge insert. Creates the PO, or backfills data into a pre-existing `Cancelled` skeleton left by an out-of-order cancellation (preserving the `Cancelled` status). No longer silently dropped when a record already exists. - `revision` — field-level merge (`update_item` SETs only the fields present in the revision). A revision that omits `line_items`/`supplier` no longer deletes them. Will not un-cancel a `Cancelled` PO. - `cancellation` — marks the row `Cancelled` (creating a minimal skeleton if the cancellation arrives before the `new_po`). -6. DynamoDB Streams feeds downstream consumers: +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. Claude extracts structured JSON (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. Claude extracts structured JSON (work order ID, site code, severity, priority, dates, assigned technician). +7. Work order upserted to `WorkOrders`, event/comment appended to `WorkOrderComments`. **Lambdas** (`lambdas/wo/`): | Function | Trigger | Purpose | @@ -73,6 +75,27 @@ The email processors **require** `ANTHROPIC_API_KEY_SECRET_ARN` to be set and re **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). @@ -205,4 +231,11 @@ scripts/ reprocess.py backfill_sites.py test_local.py # Parse sample emails through Claude locally (no AWS) +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 47d8536..4071e01 100644 --- a/cdk/po_stack.py +++ b/cdk/po_stack.py @@ -77,6 +77,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) @@ -174,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/", ], ), ), @@ -185,6 +255,14 @@ class PoIngestStack(Stack): environment={ "PO_TABLE": "purchase-orders", "ANTHROPIC_API_KEY_SECRET_ARN": anthropic_secret.secret_arn, + # 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", }, ) @@ -210,6 +288,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. diff --git a/cdk/wo_stack.py b/cdk/wo_stack.py index dfb83c1..f1fed10 100644 --- a/cdk/wo_stack.py +++ b/cdk/wo_stack.py @@ -74,6 +74,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 +256,15 @@ class WorkorderIngestStack(Stack): "WORK_ORDERS_TABLE": work_orders_table.table_name, "COMMENTS_TABLE": comments_table.table_name, "ANTHROPIC_API_KEY_SECRET_ARN": anthropic_secret.secret_arn, + # 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", }, ) @@ -219,6 +301,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. diff --git a/lambdas/po/email_processor/handler.py b/lambdas/po/email_processor/handler.py index fa193f5..d25d5af 100644 --- a/lambdas/po/email_processor/handler.py +++ b/lambdas/po/email_processor/handler.py @@ -17,6 +17,7 @@ from email import policy import anthropic import boto3 +from ses_auth import authenticate_inbound_email logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -477,6 +478,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/wo/email_processor/handler.py b/lambdas/wo/email_processor/handler.py index f51cc5e..64a9c26 100644 --- a/lambdas/wo/email_processor/handler.py +++ b/lambdas/wo/email_processor/handler.py @@ -16,6 +16,7 @@ from email import policy import anthropic import boto3 +from ses_auth import authenticate_inbound_email logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -246,13 +247,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']}") @@ -267,8 +276,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/tests/conftest.py b/tests/conftest.py index 8bb84e3..de58e0f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,9 +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 + package_dir = str(path.parent) + if package_dir not in sys.path: + sys.path.insert(0, package_dir) spec = importlib.util.spec_from_file_location(module_name, path) module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module @@ -58,3 +69,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/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