diff --git a/lambdas/wo/shoc_emitter/delivery.py b/lambdas/wo/shoc_emitter/delivery.py index bce3414..961abd8 100644 --- a/lambdas/wo/shoc_emitter/delivery.py +++ b/lambdas/wo/shoc_emitter/delivery.py @@ -28,19 +28,30 @@ import urllib.request from http import HTTPStatus import boto3 +from botocore.config import Config logger = logging.getLogger() logger.setLevel(logging.INFO) -# Endpoint path is TBD by SHOC; the activation PR confirms the final URL. -SHOC_WEBHOOK_URL = os.environ.get( - "SHOC_WEBHOOK_URL", "https://api.dev.seahaven.com/api/webhooks/work-orders" -) +# Required, NO default (Open SWE #0): a hardcoded fallback URL would silently +# ship production work-order state to that endpoint if the CDK-set env var were +# ever dropped. Unset -> deliver() fails closed (retries in order) rather than +# defaulting anywhere. The single source of the URL is the Lambda environment +# (CDK), confirmed by the activation PR. +SHOC_WEBHOOK_URL = os.environ.get("SHOC_WEBHOOK_URL") HMAC_SECRET_ARN = os.environ.get("HMAC_SECRET_ARN") POST_TIMEOUT_SECONDS = 10 USER_AGENT = "workorder-shoc-emitter/1" +# Bound the Secrets Manager client's timeouts (Open SWE #16): default botocore +# timeouts can run to ~60s, which combined with the 10s POST could exhaust the +# Lambda budget and make a slow AWS call re-deliver the whole batch. Fast, few +# retries -- the key is 300s-cached so this fetch is rare. +_BOTO_CONFIG = Config( + connect_timeout=3, read_timeout=5, retries={"max_attempts": 2, "mode": "standard"} +) + # 2xx = delivered; 429 and 5xx retry; everything else parks (contract sec. 7). _HTTP_SUCCESS_RANGE = range(HTTPStatus.OK, HTTPStatus.MULTIPLE_CHOICES) _HTTP_SERVER_ERROR_MIN = HTTPStatus.INTERNAL_SERVER_ERROR @@ -117,7 +128,7 @@ def _refresh_hmac_keys() -> list[dict]: """Force a fresh Secrets Manager fetch, bypassing the TTL cache.""" global _hmac_keys_cache, _hmac_keys_cached_at now = time.monotonic() - secrets = boto3.client("secretsmanager") + secrets = boto3.client("secretsmanager", config=_BOTO_CONFIG) try: secret = secrets.get_secret_value(SecretId=HMAC_SECRET_ARN) keys = json.loads(secret["SecretString"]).get("keys") or [] @@ -150,12 +161,14 @@ def deliver(envelope: dict) -> tuple[str, int]: Raises RetryableDeliveryError for 429/5xx/timeout/connection failures so the caller can block the shard (in-order retry, contract section 7). """ - if not SHOC_WEBHOOK_URL.startswith("https://"): - # Fail closed on any non-HTTPS scheme (file://, http://, ...): the - # URL is operator-set env config, but urllib would happily follow - # other schemes and the HMAC only protects an HTTPS transport. + if not SHOC_WEBHOOK_URL or not SHOC_WEBHOOK_URL.startswith("https://"): + # Fail closed if the URL is unset (no hardcoded fallback -- Open SWE + # #0) or is any non-HTTPS scheme (file://, http://, ...): urllib would + # otherwise follow it, and the HMAC only protects an HTTPS transport. + # Retryable, so a config gap blocks the shard (visible via iterator-age) + # rather than shipping data somewhere unintended. raise RetryableDeliveryError( - "SHOC_WEBHOOK_URL is not an https:// URL; refusing to deliver" + "SHOC_WEBHOOK_URL is unset or not an https:// URL; refusing to deliver" ) raw_body = json.dumps(envelope).encode("utf-8") timestamp = int(time.time()) diff --git a/lambdas/wo/shoc_emitter/handler.py b/lambdas/wo/shoc_emitter/handler.py index 1f3a920..9f88c17 100644 --- a/lambdas/wo/shoc_emitter/handler.py +++ b/lambdas/wo/shoc_emitter/handler.py @@ -36,12 +36,20 @@ import time import boto3 import delivery import envelope +from botocore.config import Config logger = logging.getLogger() logger.setLevel(logging.INFO) REJECTED_QUEUE_URL = os.environ.get("REJECTED_QUEUE_URL") +# Bound the SQS client's timeouts (Open SWE #17): a slow SQS response while +# parking a rejected envelope must not hang the invocation toward its 60s +# timeout and re-deliver the whole batch. +_BOTO_CONFIG = Config( + connect_timeout=3, read_timeout=5, retries={"max_attempts": 2, "mode": "standard"} +) + # Lazy cached SQS client. Keeps the public attribute name ``sqs`` so the test # monkeypatch target changes module only, not attribute name. sqs = None @@ -50,7 +58,7 @@ sqs = None def _get_sqs(): global sqs if sqs is None: - sqs = boto3.client("sqs") + sqs = boto3.client("sqs", config=_BOTO_CONFIG) return sqs @@ -117,6 +125,24 @@ def handler(event, context): if webhook_event is None: skipped += 1 continue + # Capture the sequence number BEFORE any delivery attempt so the + # failure paths below can never KeyError on the subscript (Open SWE + # #9/#26 -- the catch-all exists to keep an exception from re-delivering + # the whole batch, so it must not itself raise). SequenceNumber is + # present on every real DynamoDB stream record; a record lacking it is + # unmappable-to-a-checkpoint, so log and skip rather than crash. + seq = record.get("dynamodb", {}).get("SequenceNumber") + if seq is None: + logger.warning( + json.dumps( + { + "event": "shoc_missing_sequence_number", + "delivery_id": webhook_event["delivery_id"], + } + ) + ) + skipped += 1 + continue emitted += 1 start = time.monotonic() try: @@ -133,11 +159,7 @@ def handler(event, context): logger.info(_batch_summary(len(records), emitted, skipped)) # Stop here: reporting this record's sequence number makes the ESM # retry from it in order; earlier successes are not re-delivered. - return { - "batchItemFailures": [ - {"itemIdentifier": record["dynamodb"]["SequenceNumber"]} - ] - } + return {"batchItemFailures": [{"itemIdentifier": seq}]} except Exception: # Catch-all so an unexpected error (e.g. an SQS park failure) # cannot escape and fail the WHOLE invocation -- that would make @@ -153,10 +175,6 @@ def handler(event, context): ) ) logger.info(_batch_summary(len(records), emitted, skipped)) - return { - "batchItemFailures": [ - {"itemIdentifier": record["dynamodb"]["SequenceNumber"]} - ] - } + return {"batchItemFailures": [{"itemIdentifier": seq}]} logger.info(_batch_summary(len(records), emitted, skipped)) return {"batchItemFailures": []} diff --git a/tests/test_shoc_emitter_delivery.py b/tests/test_shoc_emitter_delivery.py index 16f0cc3..a6649b3 100644 --- a/tests/test_shoc_emitter_delivery.py +++ b/tests/test_shoc_emitter_delivery.py @@ -119,6 +119,15 @@ class _FakeResponse: 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) @@ -138,6 +147,22 @@ def test_2xx_is_delivered(monkeypatch, delivery, signing_keys, 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): @@ -272,7 +297,9 @@ def cache_reset(monkeypatch, delivery): def _wire_cache(monkeypatch, delivery, fake, clock): - monkeypatch.setattr(delivery.boto3, "client", lambda service: fake) + # 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"]) diff --git a/tests/test_shoc_emitter_handler.py b/tests/test_shoc_emitter_handler.py index 1a4a713..a5f1c15 100644 --- a/tests/test_shoc_emitter_handler.py +++ b/tests/test_shoc_emitter_handler.py @@ -205,6 +205,25 @@ def test_unexpected_error_reports_record_not_whole_batch(monkeypatch, handler_mo assert attempted == ["evt-1", "evt-2"] +def test_record_missing_sequence_number_is_skipped_not_crashed( + monkeypatch, handler_mod +): + # A record that maps to an envelope but lacks SequenceNumber (unreachable + # for real streams) must be skipped, not crash the failure paths (Open SWE + # #9/#26). Craft one that build_event accepts but strip SequenceNumber. + rec = _wo_record("evt-1", "999") + del rec["dynamodb"]["SequenceNumber"] + attempted, _ = _wire( + monkeypatch, handler_mod, {"evt-1": "retryable", "evt-2": ("delivered", 200)} + ) + good = _wo_record("evt-2", "1000") + result = handler_mod.handler({"Records": [rec, good]}, None) + # The seq-less record is skipped (never delivered); the good one delivers. + assert result == {"batchItemFailures": []} + assert "evt-1" not in attempted + assert "evt-2" in attempted + + def test_empty_batch_returns_no_failures(monkeypatch, handler_mod): attempted, _ = _wire(monkeypatch, handler_mod, {}) assert handler_mod.handler({"Records": []}, None) == {"batchItemFailures": []}