tests/test_webhook_security.py

"""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()