2026-07-15 19:06:20 -04:00
|
|
|
"""Shared pytest configuration for the procurement-ingest test suite.
|
|
|
|
|
|
|
|
|
|
The Lambda handlers create boto3 clients at module import time, so a
|
|
|
|
|
region and dummy credentials must be present in the environment before
|
|
|
|
|
any handler module is imported. Setting them here at conftest import
|
|
|
|
|
time guarantees they exist before test collection touches a handler.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
import importlib.util
|
|
|
|
|
import os
|
|
|
|
|
import sys
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
|
|
os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
|
|
|
|
|
os.environ.setdefault("AWS_ACCESS_KEY_ID", "testing")
|
|
|
|
|
os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "testing")
|
|
|
|
|
os.environ.setdefault("AWS_SESSION_TOKEN", "testing")
|
|
|
|
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def load_handler(relative_path, module_name):
|
|
|
|
|
"""Load a Lambda handler module by file path under a unique name.
|
|
|
|
|
|
|
|
|
|
The handler files all share the basename ``handler.py`` and are not
|
|
|
|
|
importable as packages, so a plain ``import handler`` would collide
|
Add fail-closed SES sender authentication
The From header and any raw-MIME Authentication-Results copies are
attacker-forgeable, so a forged email to apm@int.seahaven.com or
amazon_po@int.seahaven.com could create or mutate a WO/PO (INFRA-107,
CRITICAL). Both S3-triggered email processors now authenticate the
sender against the Authentication-Results header SES itself prepends
at delivery: only the topmost header is consulted, its authserv-id
must be amazonses.com, and it must carry dkim=pass for a domain in
the per-pipeline ALLOWED_DKIM_DOMAINS env var (comma-separated, set
in CDK so ops can adjust without code changes).
Allowlists come from live traffic observed 2026-07-15 on both ingest
buckets: WO mail arrives via the apm@ Google Groups forward, which
re-signs as seahaven.com (the hxgnsmartcloud.com signature does not
survive the forward); PO mail passes for amazon.coupahost.com.
amazonses.com also passes on PO mail but is deliberately excluded --
every SES customer's outbound mail passes for it.
Every failure path (env var unset, header missing or unparseable,
verdict fail, unaligned domain) rejects the email: a structured
warning with the reason and S3 key is logged and the record skipped
without erroring the invocation, so rejected mail causes no Lambda
retries or DLQ messages. Handler signatures and event sources are
unchanged.
Refs: INFRA-107
2026-07-15 18:53:46 -04:00
|
|
|
across Lambdas. The same loader serves the ``ses_auth.py`` modules,
|
|
|
|
|
which are likewise duplicated per pipeline and not importable as
|
|
|
|
|
packages.
|
2026-07-15 20:18:49 -04:00
|
|
|
|
|
|
|
|
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.
|
2026-07-15 19:06:20 -04:00
|
|
|
"""
|
|
|
|
|
path = REPO_ROOT / relative_path
|
2026-07-15 20:18:49 -04:00
|
|
|
package_dir = str(path.parent)
|
|
|
|
|
if package_dir not in sys.path:
|
|
|
|
|
sys.path.insert(0, package_dir)
|
2026-07-15 19:06:20 -04:00
|
|
|
spec = importlib.util.spec_from_file_location(module_name, path)
|
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
|
|
|
sys.modules[module_name] = module
|
|
|
|
|
spec.loader.exec_module(module)
|
|
|
|
|
return module
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
|
|
|
def po_handler():
|
|
|
|
|
"""The PO email processor handler module."""
|
|
|
|
|
return load_handler(
|
|
|
|
|
"lambdas/po/email_processor/handler.py",
|
|
|
|
|
"po_email_processor_handler",
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
|
|
|
def wo_handler():
|
|
|
|
|
"""The WO email processor handler module."""
|
|
|
|
|
return load_handler(
|
|
|
|
|
"lambdas/wo/email_processor/handler.py",
|
|
|
|
|
"wo_email_processor_handler",
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture(params=["po_handler", "wo_handler"])
|
|
|
|
|
def email_handler(request):
|
|
|
|
|
"""Parametrized fixture yielding each email processor handler module."""
|
|
|
|
|
return request.getfixturevalue(request.param)
|
Add fail-closed SES sender authentication
The From header and any raw-MIME Authentication-Results copies are
attacker-forgeable, so a forged email to apm@int.seahaven.com or
amazon_po@int.seahaven.com could create or mutate a WO/PO (INFRA-107,
CRITICAL). Both S3-triggered email processors now authenticate the
sender against the Authentication-Results header SES itself prepends
at delivery: only the topmost header is consulted, its authserv-id
must be amazonses.com, and it must carry dkim=pass for a domain in
the per-pipeline ALLOWED_DKIM_DOMAINS env var (comma-separated, set
in CDK so ops can adjust without code changes).
Allowlists come from live traffic observed 2026-07-15 on both ingest
buckets: WO mail arrives via the apm@ Google Groups forward, which
re-signs as seahaven.com (the hxgnsmartcloud.com signature does not
survive the forward); PO mail passes for amazon.coupahost.com.
amazonses.com also passes on PO mail but is deliberately excluded --
every SES customer's outbound mail passes for it.
Every failure path (env var unset, header missing or unparseable,
verdict fail, unaligned domain) rejects the email: a structured
warning with the reason and S3 key is logged and the record skipped
without erroring the invocation, so rejected mail causes no Lambda
retries or DLQ messages. Handler signatures and event sources are
unchanged.
Refs: INFRA-107
2026-07-15 18:53:46 -04:00
|
|
|
|
|
|
|
|
|
|
|
|
|
@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",
|
|
|
|
|
)
|