procurement-ingest/tests/test_shoc_emitter_delivery.py

354 lines
12 KiB
Python
Raw Normal View History

feat(webhook): SHOC WO webhook emitter - dark-ship streams + HMAC secret/rotation (PR-2) (#137) * docs(webhook): revise SHOC webhook contract and plan for post-migration reality Branch re-cut on main 2026-07-23 (old base carried stale PR #99 commits). Contract Rev 2026-07-23: - Producer account corrected: seahaven-prod (011934824531); mgmt frozen - Reconciliation backstop is the new procurement read API, not SyncController - wo_status "unknown" is real; SHOC must map it (checklist item added) - write_origin forward-compat note for phase-2 write-back echo suppression - SyncVendorReplies retirement flagged (dead table, no vendor_reply event) Plan updates: - Account gate: seahaven-prod only; never enable streams on mgmt tables - Emitter ships DARK (ESMs enabled=False); activation is a deliberate flip after the SHOC receiver passes shared HMAC vectors - Post-refactor conventions: common.py helpers, bundle-consistency AST pins, pytest.ini --cov additions, consolidated test roots - Dedicated-CMK rationale, secret-ARN handooff step, consumer audit refreshed (slack-bot decommissioned), enum golden test, write_origin skip-branch test * feat(webhook): SHOC WO webhook emitter — dark-ship streams, HMAC secret + rotation Implements docs/shoc-webhook-plan.md Phases 1-5 (PR-2 of the SHOC call-and-be-called effort). Everything ships DARK: both DynamoDB event source mappings deploy enabled=False; activation is a deliberate one-line follow-up PR gated on the SHOC receiver passing the shared HMAC test vectors. - Streams: NEW_AND_OLD_IMAGES on WorkOrders + WorkOrderComments (in-place update, RETAIN + logical IDs untouched; no existing consumers — verified live, neither table had a stream). - workorder-shoc-emitter (Py3.12/ARM64): stream -> envelope -> HMAC-signed POST per docs/shoc-webhook-contract.md; strict per-shard ordering (parallelization 1, bisect off, retry until 24h age, ReportBatchItemFailures); 429/5xx/timeout block the shard in order, other 4xx park to workorder-shoc-emitter-rejected; ESM failures -> workorder-shoc-emitter-failures (metadata; replay rebuilds from DynamoDB). Echo guard skips write_origin=shoc-write-api. - Secret workorder-ingest/shoc-webhook-hmac on a dedicated CMK (alias workorder-ingest-shoc-webhook-kms); cross-account GetSecretValue/DescribeSecret + kms:Decrypt granted to exactly arn:aws:iam::396287094661:role/shoc-backend-dev. RemovalPolicy DESTROY deliberately (machine-generated material; avoids the fixed-name RETAIN-orphan deadlock). - workorder-shoc-hmac-rotator: 30-day rotation, dual-key overlap, 64-hex keys, kid = UTC %Y-%m-%dT%H. - Alarms (ALARM-only -> site-alerts): emitter errors/throttles/ duration + iterator-age (>=10 min) + failures/rejected queue depth; rotator standard trio. - scripts/replay_shoc_webhooks.py: dry-run-default operator replay (rebuilds from tables, replay:true envelopes). - Tests: 742 passing, 85.56% aggregate; golden HMAC vectors shared with SHOC in docs/shoc-webhook-test-vectors.json (emitter + replay signing pinned to identical vectors); bundle-consistency AST pins for both new bundles. - README: WO stack + webhook feed section, alarm table, runbooks; removed stale seahaven-slack-bot consumer references. * fix(webhook): kms:ViaService pins, https-only delivery, cross-account principal CI pin GPT-4.1 cross-family review of the policy surface (no BLOCK): FIX applied to the cross-account shoc-backend-dev Decrypt statement and both Lambda role KMS grants (the key is only ever used via Secrets Manager); its invariant-enforcement QUESTION answered durably with tests/test_cross_account_principal_pin.py (any new foreign IAM principal in cdk/ fails CI). Scanner mediums fixed: delivery.py and the replay script now refuse non-https URLs (urllib follows file:// and http://). SQS metadata-action and dynamodb:ListStreams NITs skipped: standard CDK grant shapes; ListStreams has no resource-level scoping. The 4 gitleaks HIGHs on docs/shoc-webhook-test-vectors.json are deliberate non-secrets (shared receiver-verification vectors) suppressed machine-level with justification. * harden(webhook): resolve /sh-security-review findings (1 confirmed medium + cheap fixes) High-recall detector fan-out (injection/authz/secrets-crypto/iac-iam/logic) + proof-or-kill verifier. Gate PASSES: 1 confirmed medium, 0 confirmed critical/high. Confirmed finding fixed; several unverified-but-cheap hardenings applied since the emitter ships dark and activation is weeks out. - CONFIRMED medium (confused deputy): the rotation Lambda's generated invoke permission for secretsmanager.amazonaws.com carried no SourceAccount/SourceArn, so any account's Secrets Manager could invoke the rotator. Patched the generated CfnPermission in place (a second permission would be additive, not restrictive) to pin account + this secret ARN. - delivery + replay: refuse to follow receiver 3xx redirects (no-redirect opener) so live X-SH-* auth headers can't be forwarded to a receiver-chosen Location and an http:// Location can't slip past the https guard. Fixed the "unfollowed 3xx" comment that was factually wrong. - delivery: classify 401/403 as retryable (invalidate key cache + retry in order) instead of parking -- transient auth failures (rotation outran the TTL cache, clock skew) are availability events, not contract bugs. - envelope: build_event now genuinely total (guarded eventID / ApproximateCreationDateTime subscripts) per its own never-raise contract. - handler: catch-all so an unexpected per-record error (e.g. SQS park failure) reports only that record instead of failing the whole batch (which would re-deliver every earlier success for 24h); per-invocation emit/skip batch summary so a systemic silent drop is queryable/alarmable. - rotator: narrow the AWSCURRENT-read except to ResourceNotFound/JSONDecode (transient SM/KMS errors re-raise so the overlap key isn't silently dropped); kid uniqueness checked against ALL retained kids with a random suffix on collision (never reissue a kid for a different secret). - contract: skeleton-upsert required on ANY unknown work_order_id (not just comment-before-create) + monotonicity guard (ignore older updated_at), so a parked created or an out-of-order replay can't corrupt receiver state. Unverified/refuted findings left as-is with rationale: the two "high" logic claims (whole-batch crash triggers, ordering violation) were refuted on reachability (real stream records carry required fields; persistence writes strings only; full-state idempotent upsert absorbs the ordering gap). Signed kid/version binding (AUTHZ-002) declined: coordinated contract change, not cheap, no exploit with one algorithm/key. * fix(webhook): drop kid from rotator test_ok log (CodeQL clear-text-logging FP) GHAS CodeQL flagged py/clear-text-logging-sensitive-data (high) at _test_secret's success log because head["kid"] is subscripted from the same parsed-secret dict that holds head["secret"] — the taint tracker can't tell the non-secret key id from the secret. The secret value is never logged. Rather than dismiss the alert (fragile; re-alerts on line moves), remove the flow: kid is already logged at stage time in _create_secret and version_id correlates the steps, so the test_ok log keeps only event + version_id. Also hardens against a future edit that swaps the logged field.
2026-07-24 18:12:20 -04:00
"""Signing + delivery tests for the SHOC webhook emitter (plan Phase 5).
Golden vectors: docs/shoc-webhook-test-vectors.json is the shared handoff
artifact -- Luby's receiver verifies against the same vectors -- and BOTH
producer-side sign_body implementations (the emitter's delivery module and
the replay script) are pinned here against every vector, so they can never
drift from each other or from the published vectors.
Also covers the contract section 7 response-classification matrix (2xx /
429+5xx / timeout+connection error / other 4xx) with a monkeypatched urllib
opener, and the Secrets Manager key cache: 5-minute TTL via time.monotonic,
keys[0] selection, empty-keys (bootstrap) -> retryable.
"""
import importlib.util
import io
import json
import urllib.error
from pathlib import Path
import pytest
from tests.support import REPO_ROOT, load_lambda_module
_VECTORS_PATH = Path(REPO_ROOT) / "docs" / "shoc-webhook-test-vectors.json"
VECTORS = json.loads(_VECTORS_PATH.read_text(encoding="utf-8"))["vectors"]
TEST_KEYS = [
{"kid": "2026-07-20T00", "secret": "ab" * 32},
{"kid": "2026-06-20T00", "secret": "cd" * 32},
]
ENVELOPE = {
"schema_version": 1,
"delivery_id": "evt-1",
"event_type": "work_order.created",
"occurred_at": "2026-07-16T14:03:22.114208+00:00",
"source": "procurement-ingest/workorder-shoc-emitter",
"replay": False,
"data": {"work_order_id": "11144580730"},
}
@pytest.fixture(scope="module")
def delivery():
return load_lambda_module("wo", "shoc_emitter/delivery")
def _load_replay_script():
# scripts/ is not a package and not on sys.path; load by file path
# (mirrors tests/test_reprocess_contract.py). Import is side-effect-free:
# the script builds its boto3 session inside main().
path = Path(REPO_ROOT) / "scripts" / "replay_shoc_webhooks.py"
spec = importlib.util.spec_from_file_location(
"replay_shoc_webhooks_for_vectors", path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
# --- Golden vectors ----------------------------------------------------------
def test_vector_file_covers_required_shapes():
# The handoff artifact itself must keep its coverage promises: at least
# one non-ASCII UTF-8 body (multi-byte signing) and one empty-object body.
assert len(VECTORS) >= 4
assert any(any(ord(ch) > 127 for ch in v["body"]) for v in VECTORS)
assert any(v["body"] == "{}" for v in VECTORS)
for vector in VECTORS:
assert set(vector) == {
"kid",
"secret_hex",
"timestamp",
"body",
"expected_signature",
}
def test_delivery_sign_body_matches_golden_vectors(delivery):
for vector in VECTORS:
signature = delivery.sign_body(
vector["secret_hex"],
vector["timestamp"],
vector["body"].encode("utf-8"),
)
assert signature == vector["expected_signature"], (
f"delivery.sign_body drifted from golden vector kid="
f"{vector['kid']} ts={vector['timestamp']}"
)
def test_replay_sign_body_matches_golden_vectors():
replay = _load_replay_script()
for vector in VECTORS:
signature = replay.sign_body(
vector["secret_hex"],
vector["timestamp"],
vector["body"].encode("utf-8"),
)
assert signature == vector["expected_signature"], (
f"replay sign_body drifted from golden vector kid="
f"{vector['kid']} ts={vector['timestamp']}"
)
# --- Response classification matrix (contract section 7) ---------------------
class _FakeResponse:
def __init__(self, status):
self.status = status
def __enter__(self):
return self
def __exit__(self, *exc_info):
return False
@pytest.fixture(autouse=True)
def _webhook_url(monkeypatch, delivery):
# SHOC_WEBHOOK_URL has no default now (Open SWE #0): set it for tests that
# exercise deliver(), which fails closed on an unset URL.
monkeypatch.setattr(
delivery, "SHOC_WEBHOOK_URL", "https://shoc.example/api/webhooks/work-orders"
)
@pytest.fixture
def signing_keys(monkeypatch, delivery):
monkeypatch.setattr(delivery, "_get_hmac_keys", lambda: TEST_KEYS)
def _patch_urlopen(monkeypatch, delivery, fn):
# deliver() posts through the no-redirect opener, not the module-level
# urlopen, so patch the opener's open method.
monkeypatch.setattr(delivery._opener, "open", fn)
@pytest.mark.parametrize("status", [200, 204])
def test_2xx_is_delivered(monkeypatch, delivery, signing_keys, status):
_patch_urlopen(
monkeypatch, delivery, lambda request, timeout: _FakeResponse(status)
)
assert delivery.deliver(ENVELOPE) == ("delivered", status)
@pytest.mark.parametrize("url", [None, "", "http://insecure.example/hook"])
def test_unset_or_non_https_url_fails_closed(monkeypatch, delivery, signing_keys, url):
# No hardcoded fallback (Open SWE #0): an unset/empty/non-https URL must
# raise (retryable) and NOT sign or POST anything.
monkeypatch.setattr(delivery, "SHOC_WEBHOOK_URL", url)
called = {"opened": False}
_patch_urlopen(
monkeypatch,
delivery,
lambda request, timeout: called.__setitem__("opened", True),
)
with pytest.raises(delivery.RetryableDeliveryError):
delivery.deliver(ENVELOPE)
assert called["opened"] is False
@pytest.mark.parametrize("status", [429, 500, 503])
def test_429_and_5xx_raise_retryable(monkeypatch, delivery, signing_keys, status):
def _raise(request, timeout):
raise urllib.error.HTTPError(
delivery.SHOC_WEBHOOK_URL, status, "boom", None, io.BytesIO(b"")
)
_patch_urlopen(monkeypatch, delivery, _raise)
with pytest.raises(delivery.RetryableDeliveryError) as exc:
delivery.deliver(ENVELOPE)
assert exc.value.status_code == status
@pytest.mark.parametrize(
"error",
[urllib.error.URLError(OSError("connection refused")), TimeoutError()],
ids=["connection-error", "timeout"],
)
def test_connection_failures_raise_retryable(
monkeypatch, delivery, signing_keys, error
):
def _raise(request, timeout):
raise error
_patch_urlopen(monkeypatch, delivery, _raise)
with pytest.raises(delivery.RetryableDeliveryError) as exc:
delivery.deliver(ENVELOPE)
assert exc.value.status_code is None
@pytest.mark.parametrize("status", [400, 404, 422])
def test_other_4xx_is_rejected_not_raised(monkeypatch, delivery, signing_keys, status):
def _raise(request, timeout):
raise urllib.error.HTTPError(
delivery.SHOC_WEBHOOK_URL, status, "bad", None, io.BytesIO(b"")
)
_patch_urlopen(monkeypatch, delivery, _raise)
assert delivery.deliver(ENVELOPE) == ("rejected", status)
@pytest.mark.parametrize("status", [401, 403])
def test_auth_failures_are_retryable_and_invalidate_cache(
monkeypatch, delivery, signing_keys, status
):
# A transient auth failure (stale cached key mid-rotation, receiver
# secret-fetch blip, clock skew) must retry in order -- NOT park -- and
# drop the key cache so the retry re-signs with the current secret.
invalidated = {"called": False}
def _mark(*_a, **_k):
invalidated["called"] = True
monkeypatch.setattr(delivery, "_invalidate_hmac_keys", _mark)
def _raise(request, timeout):
raise urllib.error.HTTPError(
delivery.SHOC_WEBHOOK_URL, status, "unauthorized", None, io.BytesIO(b"")
)
_patch_urlopen(monkeypatch, delivery, _raise)
with pytest.raises(delivery.RetryableDeliveryError) as exc:
delivery.deliver(ENVELOPE)
assert exc.value.status_code == status
assert invalidated["called"] is True
def test_redirects_are_not_followed():
# The opener must refuse 3xx so auth headers are never forwarded to a
# receiver-chosen Location. redirect_request returning None makes urllib
# raise instead of following.
delivery = load_lambda_module("wo", "shoc_emitter/delivery")
handler = delivery._NoRedirectHandler()
assert handler.redirect_request(None, None, 302, "Found", {}, "http://evil") is None
def test_request_signed_with_keys0_and_contract_headers(
monkeypatch, delivery, signing_keys
):
captured = {}
def _capture(request, timeout):
captured["request"] = request
captured["timeout"] = timeout
return _FakeResponse(200)
_patch_urlopen(monkeypatch, delivery, _capture)
assert delivery.deliver(ENVELOPE) == ("delivered", 200)
request = captured["request"]
assert captured["timeout"] == delivery.POST_TIMEOUT_SECONDS
assert request.get_method() == "POST"
assert request.data == json.dumps(ENVELOPE).encode("utf-8")
assert request.get_header("Content-type") == "application/json; charset=utf-8"
assert request.get_header("User-agent") == "workorder-shoc-emitter/1"
# The producer always signs with keys[0] (contract section 6.1).
assert request.get_header("X-sh-key-id") == TEST_KEYS[0]["kid"]
timestamp = request.get_header("X-sh-timestamp")
assert timestamp.isdigit()
expected = delivery.sign_body(TEST_KEYS[0]["secret"], int(timestamp), request.data)
assert request.get_header("X-sh-signature") == f"v1={expected}"
# --- Secret cache ------------------------------------------------------------
class _FakeSecretsManager:
"""Returns payloads in sequence (last one repeats); counts fetches."""
def __init__(self, payloads):
self.payloads = list(payloads)
self.calls = 0
self.secret_ids = []
def get_secret_value(self, SecretId): # noqa: N803 (boto3 kwarg name)
self.calls += 1
self.secret_ids.append(SecretId)
payload = self.payloads.pop(0) if len(self.payloads) > 1 else self.payloads[0]
return {"SecretString": json.dumps(payload)}
@pytest.fixture
def cache_reset(monkeypatch, delivery):
monkeypatch.setattr(delivery, "_hmac_keys_cache", None)
monkeypatch.setattr(delivery, "_hmac_keys_cached_at", 0.0)
monkeypatch.setattr(
delivery,
"HMAC_SECRET_ARN",
"arn:aws:secretsmanager:us-east-1:011934824531:secret:"
"workorder-ingest/shoc-webhook-hmac-AbCdEf",
)
def _wire_cache(monkeypatch, delivery, fake, clock):
# client() is now called with a config= kwarg (bounded timeouts), so accept
# and ignore it.
monkeypatch.setattr(delivery.boto3, "client", lambda service, **kwargs: fake)
monkeypatch.setattr(delivery.time, "monotonic", lambda: clock["t"])
def test_cache_honors_ttl_and_refreshes_after_expiry(
monkeypatch, delivery, cache_reset
):
rotated = [
{"keys": [{"kid": "2026-08-20T00", "secret": "ef" * 32}] + TEST_KEYS[:1]}
]
fake = _FakeSecretsManager([{"keys": TEST_KEYS}] + rotated)
clock = {"t": 1000.0}
_wire_cache(monkeypatch, delivery, fake, clock)
assert delivery._get_hmac_keys()[0]["kid"] == "2026-07-20T00"
assert fake.calls == 1
# Inside the 300s TTL: served from cache, no refetch.
clock["t"] = 1000.0 + 299.0
assert delivery._get_hmac_keys()[0]["kid"] == "2026-07-20T00"
assert fake.calls == 1
# TTL expired: refetch picks up the rotated keys[0] (cache invalidation
# is what makes 30-day rotation propagate within 5 minutes).
clock["t"] = 1000.0 + 300.5
assert delivery._get_hmac_keys()[0]["kid"] == "2026-08-20T00"
assert fake.calls == 2
assert fake.secret_ids[0] == delivery.HMAC_SECRET_ARN
def test_empty_or_missing_keys_is_retryable_bootstrap_state(
monkeypatch, delivery, cache_reset
):
clock = {"t": 5000.0}
for payload in ({"keys": []}, {"keys": [], "bootstrap_entropy": "seed"}, {}):
monkeypatch.setattr(delivery, "_hmac_keys_cache", None)
monkeypatch.setattr(delivery, "_hmac_keys_cached_at", 0.0)
fake = _FakeSecretsManager([payload])
_wire_cache(monkeypatch, delivery, fake, clock)
with pytest.raises(delivery.RetryableDeliveryError):
delivery._get_hmac_keys()
def test_secret_fetch_failure_is_retryable(monkeypatch, delivery, cache_reset):
class _Boom:
def get_secret_value(self, SecretId): # noqa: N803 (boto3 kwarg name)
raise RuntimeError("throttled")
clock = {"t": 9000.0}
_wire_cache(monkeypatch, delivery, _Boom(), clock)
with pytest.raises(delivery.RetryableDeliveryError):
delivery._get_hmac_keys()