Spaces:
Running
Running
File size: 5,462 Bytes
871f883 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | """Contratti di sicurezza del webhook HMAC e della sua idempotenza."""
from __future__ import annotations
import hashlib
import hmac
import os
import sys
import unittest
from unittest.mock import patch
from fastapi import HTTPException
_BACKEND = os.path.join(os.path.dirname(__file__), "..")
if _BACKEND not in sys.path:
sys.path.insert(0, _BACKEND)
from api.webhook_security import ( # noqa: E402
WebhookDeliveryStore,
_signing_bytes,
verify_webhook_request,
)
_SECRET = "s" * 32
_EVENT_ID = "evt-security-0001"
_NOW = 1_700_000_000
def _headers(raw: bytes, *, event_id: str = _EVENT_ID, timestamp: int = _NOW) -> dict[str, str]:
timestamp_raw = str(timestamp)
digest = hmac.new(
_SECRET.encode("utf-8"),
_signing_bytes(timestamp_raw, event_id, raw),
hashlib.sha256,
).hexdigest()
return {
"x-webhook-id": event_id,
"x-webhook-timestamp": timestamp_raw,
"x-webhook-signature": f"sha256={digest}",
}
class WebhookSecurityTests(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self) -> None:
await WebhookDeliveryStore.reset_memory_for_test()
def test_signature_binds_the_exact_raw_bytes_and_fresh_timestamp(self) -> None:
raw = b'{"goal":"safe", "context":[]}'
with patch.dict(os.environ, {"WEBHOOK_HMAC_SECRET": _SECRET}, clear=False):
verified = verify_webhook_request(raw, _headers(raw), now=_NOW)
self.assertEqual(verified.event_id, _EVENT_ID)
self.assertEqual(verified.payload_sha256, hashlib.sha256(raw).hexdigest())
with self.assertRaises(HTTPException) as altered:
verify_webhook_request(b'{"goal":"safe","context":[]}', _headers(raw), now=_NOW)
with self.assertRaises(HTTPException) as future:
verify_webhook_request(raw, _headers(raw, timestamp=_NOW + 301), now=_NOW)
self.assertEqual(altered.exception.status_code, 401)
self.assertEqual(future.exception.status_code, 401)
def test_required_headers_and_secret_fail_closed(self) -> None:
raw = b'{"goal":"safe"}'
with patch.dict(os.environ, {"WEBHOOK_HMAC_SECRET": "too-short"}, clear=False):
with self.assertRaises(HTTPException) as missing_secret:
verify_webhook_request(raw, _headers(raw), now=_NOW)
with patch.dict(os.environ, {"WEBHOOK_HMAC_SECRET": _SECRET}, clear=False):
malformed = _headers(raw)
malformed["x-webhook-id"] = "short"
with self.assertRaises(HTTPException) as invalid_id:
verify_webhook_request(raw, malformed, now=_NOW)
missing_signature = _headers(raw)
missing_signature.pop("x-webhook-signature")
with self.assertRaises(HTTPException) as unsigned:
verify_webhook_request(raw, missing_signature, now=_NOW)
self.assertEqual(missing_secret.exception.status_code, 503)
self.assertEqual(invalid_id.exception.status_code, 400)
self.assertEqual(unsigned.exception.status_code, 401)
async def test_memory_state_machine_caches_replay_conflicts_and_releases_pre_dispatch(self) -> None:
raw = b'{"goal":"safe"}'
other_raw = b'{"goal":"other"}'
with patch.dict(
os.environ,
{"WEBHOOK_HMAC_SECRET": _SECRET, "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false"},
clear=False,
):
verified = verify_webhook_request(raw, _headers(raw), now=_NOW)
conflicting = verify_webhook_request(other_raw, _headers(other_raw), now=_NOW)
store = WebhookDeliveryStore(None)
self.assertEqual((await store.claim(verified)).decision, "claimed")
self.assertEqual((await store.claim(verified)).decision, "in_progress")
self.assertEqual((await store.claim(conflicting)).decision, "conflict")
await store.complete(verified, {"ok": True, "output": "cached"})
replay = await store.claim(verified)
self.assertEqual(replay.decision, "replay")
self.assertEqual(replay.response, {"ok": True, "output": "cached"})
release_id = "evt-release-00001"
releasing = verify_webhook_request(raw, _headers(raw, event_id=release_id), now=_NOW)
self.assertEqual((await store.claim(releasing)).decision, "claimed")
await store.release(releasing)
self.assertEqual((await store.claim(releasing)).decision, "claimed")
async def test_durable_store_is_required_by_default_and_rpc_errors_are_unavailable(self) -> None:
raw = b'{"goal":"safe"}'
with patch.dict(os.environ, {"WEBHOOK_HMAC_SECRET": _SECRET, "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "true"}, clear=False):
verified = verify_webhook_request(raw, _headers(raw), now=_NOW)
with self.assertRaises(HTTPException) as absent_store:
await WebhookDeliveryStore(None).claim(verified)
with self.assertRaises(HTTPException) as broken_store:
await WebhookDeliveryStore(_BrokenSupabase()).claim(verified)
self.assertEqual(absent_store.exception.status_code, 503)
self.assertEqual(broken_store.exception.status_code, 503)
class _BrokenSupabase:
def rpc(self, *_args: object, **_kwargs: object) -> object:
raise RuntimeError("database unavailable")
if __name__ == "__main__":
unittest.main()
|