From 70449a49393bb3e0c0a629341c70dec869d5f219 Mon Sep 17 00:00:00 2001 From: MorAlekss Date: Sat, 4 Jul 2026 12:11:12 -0700 Subject: [PATCH] fix(security): add timestamp-bound V2 signature for generic webhook replay protection --- gateway/platforms/webhook.py | 41 +++++++++- tests/gateway/test_webhook_adapter.py | 111 ++++++++++++++++++++++++++ 2 files changed, 151 insertions(+), 1 deletion(-) diff --git a/gateway/platforms/webhook.py b/gateway/platforms/webhook.py index c9acf61b32c..a53697e30e7 100644 --- a/gateway/platforms/webhook.py +++ b/gateway/platforms/webhook.py @@ -23,6 +23,10 @@ Security: - Rate limiting per route (fixed-window, configurable) - Idempotency cache prevents duplicate agent runs on webhook retries - Body size limits checked before reading payload + - Generic HMAC supports a V2 signature (X-Webhook-Signature-V2) that + binds a timestamp into the signed data for replay protection; the + legacy body-only V1 (X-Webhook-Signature) is deprecated but still + accepted with a warning, since it has no replay protection - Set secret to "INSECURE_NO_AUTH" to skip validation (testing only) """ @@ -876,12 +880,47 @@ class WebhookAdapter(BasePlatformAdapter): if gl_token: return hmac.compare_digest(gl_token, secret) - # Generic: X-Webhook-Signature = + # Generic V2: X-Webhook-Signature-V2 = ."> + # X-Webhook-Timestamp = (required for V2) + # Checked independently of (and before) legacy V1 below — a sender + # that only ever sends V2 headers must still validate here; nesting + # this inside `if generic_sig:` would silently skip V2-only senders. + v2_sig = request.headers.get("X-Webhook-Signature-V2", "") + v2_timestamp = request.headers.get("X-Webhook-Timestamp", "") + if v2_sig and v2_timestamp: + try: + ts = int(v2_timestamp) + except (TypeError, ValueError): + return False + if abs(int(time.time()) - ts) > 300: + logger.warning( + "[webhook] Route '%s' generic HMAC V2 timestamp outside replay window", + request.match_info.get("route_name", ""), + ) + return False + signed_content = v2_timestamp.encode() + b"." + body + expected_v2 = hmac.new( + secret.encode(), signed_content, hashlib.sha256 + ).hexdigest() + return hmac.compare_digest(v2_sig, expected_v2) + + # Generic V1 (legacy): X-Webhook-Signature = + # (deprecated — no replay protection, since the signature only + # covers the body: a captured (body, signature) pair replays + # indefinitely with no timestamp binding it to a specific delivery.) generic_sig = request.headers.get("X-Webhook-Signature", "") if generic_sig: expected = hmac.new( secret.encode(), body, hashlib.sha256 ).hexdigest() + logger.warning( + "[webhook] Route '%s' uses legacy body-only HMAC (no " + "timestamp), which is vulnerable to replay attacks. Add " + "an 'X-Webhook-Timestamp' header and switch to " + "'X-Webhook-Signature-V2' (HMAC-SHA256 of " + "'.').", + request.match_info.get("route_name", ""), + ) return hmac.compare_digest(generic_sig, expected) # No recognised signature header but secret is configured → reject diff --git a/tests/gateway/test_webhook_adapter.py b/tests/gateway/test_webhook_adapter.py index 3c5c59a2001..ee5550ba041 100644 --- a/tests/gateway/test_webhook_adapter.py +++ b/tests/gateway/test_webhook_adapter.py @@ -103,6 +103,12 @@ def _generic_signature(body: bytes, secret: str) -> str: return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() +def _generic_v2_signature(body: bytes, secret: str, timestamp: str) -> str: + """Compute X-Webhook-Signature-V2 (HMAC-SHA256 of ".").""" + signed_content = timestamp.encode() + b"." + body + return hmac.new(secret.encode(), signed_content, hashlib.sha256).hexdigest() + + def _svix_signature(body: bytes, secret: str, msg_id: str, timestamp: str) -> str: """Compute a Svix v1 signature header for *body* using *secret*.""" key = ( @@ -185,6 +191,111 @@ class TestValidateSignature: req = _mock_request(headers={"X-Webhook-Signature": sig}) assert adapter._validate_signature(req, body, secret) is True + def test_validate_generic_v2_signature_valid(self): + """Valid X-Webhook-Signature-V2 (timestamp-bound) is accepted.""" + adapter = _make_adapter() + body = b'{"event": "push"}' + secret = "generic-secret" + timestamp = str(int(time.time())) + sig = _generic_v2_signature(body, secret, timestamp) + req = _mock_request(headers={ + "X-Webhook-Signature-V2": sig, + "X-Webhook-Timestamp": timestamp, + }) + assert adapter._validate_signature(req, body, secret) is True + + def test_validate_generic_v2_old_timestamp_rejects(self): + """A V2 signature outside the replay window is rejected even though + the HMAC itself would otherwise be valid for that (stale) timestamp + — this is the actual replay-protection guarantee: an attacker who + captured (body, signature, timestamp) once cannot resubmit it after + the window closes.""" + adapter = _make_adapter() + body = b'{"event": "push"}' + secret = "generic-secret" + timestamp = str(int(time.time()) - 301) + sig = _generic_v2_signature(body, secret, timestamp) + req = _mock_request(headers={ + "X-Webhook-Signature-V2": sig, + "X-Webhook-Timestamp": timestamp, + }) + assert adapter._validate_signature(req, body, secret) is False + + def test_validate_generic_v2_wrong_timestamp_rejects(self): + """The timestamp is cryptographically bound into the V2 signature — + this is the actual fix for the V1 replay hole. An attacker who only + has a captured (body, signature) pair for V1 (no timestamp binding) + cannot forge a valid V2 signature for a fresh timestamp without the + secret, unlike V1 where the signature covers the body alone and a + forged/fresh timestamp would otherwise sail through unverified.""" + adapter = _make_adapter() + body = b'{"event": "push"}' + secret = "generic-secret" + real_timestamp = str(int(time.time())) + sig = _generic_v2_signature(body, secret, real_timestamp) + forged_timestamp = str(int(time.time()) + 1) + req = _mock_request(headers={ + "X-Webhook-Signature-V2": sig, + "X-Webhook-Timestamp": forged_timestamp, + }) + assert adapter._validate_signature(req, body, secret) is False + + def test_validate_generic_v2_malformed_timestamp_rejects(self): + adapter = _make_adapter() + body = b'{"event": "push"}' + secret = "generic-secret" + req = _mock_request(headers={ + "X-Webhook-Signature-V2": "deadbeef", + "X-Webhook-Timestamp": "not-a-number", + }) + assert adapter._validate_signature(req, body, secret) is False + + def test_validate_generic_v1_still_works_without_timestamp(self): + """Legacy V1 (body-only) senders that never send X-Webhook-Timestamp + must keep working — this is the backward-compatibility guarantee for + existing integrations that predate the V2 scheme.""" + adapter = _make_adapter() + body = b'{"event": "push"}' + secret = "generic-secret" + sig = _generic_signature(body, secret) + req = _mock_request(headers={"X-Webhook-Signature": sig}) + assert adapter._validate_signature(req, body, secret) is True + + def test_validate_generic_v2_preferred_when_both_sent(self): + """If a sender sends both V1 and V2 headers (mid-migration), V2 must + win — a stale/wrong V1 must not be able to override a valid V2.""" + adapter = _make_adapter() + body = b'{"event": "push"}' + secret = "generic-secret" + timestamp = str(int(time.time())) + v2_sig = _generic_v2_signature(body, secret, timestamp) + req = _mock_request(headers={ + "X-Webhook-Signature-V2": v2_sig, + "X-Webhook-Timestamp": timestamp, + # Deliberately wrong V1 — must be ignored since V2 is checked first. + "X-Webhook-Signature": "0" * 64, + }) + assert adapter._validate_signature(req, body, secret) is True + + def test_v1_replay_attack_succeeds_demonstrating_the_hole_v2_closes(self): + """Regression/documentation test: a captured (body, signature) V1 + pair replays successfully no matter how much time has passed, + because the V1 signature has no timestamp binding at all. This is + the exact vulnerability V2 fixes — it is not asserting desired + behavior, it is pinning the known, accepted-with-warning legacy + gap so a future change to V1's semantics doesn't silently alter it + without a deliberate decision.""" + adapter = _make_adapter() + body = b'{"event": "push"}' + secret = "generic-secret" + sig = _generic_signature(body, secret) + original_request = _mock_request(headers={"X-Webhook-Signature": sig}) + assert adapter._validate_signature(original_request, body, secret) is True + # "Time passes" — nothing about a V1 signature depends on time, so + # a captured pair replayed much later still validates. + replayed_request = _mock_request(headers={"X-Webhook-Signature": sig}) + assert adapter._validate_signature(replayed_request, body, secret) is True + def test_validate_svix_signature_valid(self): """Valid Svix/AgentMail v1 signature headers are accepted.""" adapter = _make_adapter()