mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
fix(webhook): address Open SWE review — require webhook URL, guard seq, bound client timeouts
Triaged all 29 Open SWE findings (6 posted + 23 below-threshold). Fixed the three genuine ones; the rest are refuted/unreachable/intentional (details replied on the PR). - #0/#22 (high): SHOC_WEBHOOK_URL now has NO hardcoded default. A dropped env var previously fell back to the dev endpoint and would ship prod work-order data there; deliver() now fails closed (retryable) on an unset/non-https URL. The CDK env var is the single source. - #9/#26 (high): capture record["dynamodb"]["SequenceNumber"] once, safely, before delivery. The catch-all exists so an exception can't re-deliver the whole batch — it must not itself KeyError on the subscript. A record without a SequenceNumber (unreachable for real streams) is logged + skipped. - #16/#17: bound the Secrets Manager + SQS client timeouts (botocore Config: 3s connect / 5s read / 2 retries) so a slow AWS call can't hang the invocation toward its 60s timeout and re-deliver the batch. Dismissed (confirmed sound): HMAC scope is contract-pinned + golden-vector shared (changing it is a breaking cross-party change); the replay script's hardcoded prod account is a deliberate fail-closed guard, and its required --url + dry-run-default + https-only are its intended operator-runbook design. 753 pytest / 85.7%, ruff, cdk synth green. No IAM/auth/Lambda-signature change, so the mandatory cross-family gates are unaffected.
This commit is contained in:
parent
9a3784c471
commit
5746de001e
4 changed files with 99 additions and 22 deletions
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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": []}
|
||||
|
|
|
|||
|
|
@ -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"])
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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": []}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue