# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-timestamp-replay-and-idempotent-deduplication-gates-must-an-instance-verify-and-dispatch-webhooks-to-prevent-replay-attacks-and-duplicate-side-effects.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-timestamp-replay-and-idempotent-deduplication-gates-must-an-instance-verify-and-dispatch-webhooks-to-prevent-replay-attacks-and-duplicate-side-effects.warrant.json
# Markdown SHA-256:   4bf78242b112ef25b9bd5b9354cd457e0cb6da1c36d6433fe1411d01ed9118e5
# Code AST SHA-256:   0f259372431c2efeb92bba7ae45b3b0863e4f58e588461a37b2636159109f302
# Verified Classes:   WebhookSecurityError, VerifiedWebhookEvent, ReplaySafeWebhookGuard
# Defensive Guards:   7 fail-closed invariants (zero external deps)

import hashlib
import hmac
import json
import sqlite3
import time
from dataclasses import dataclass
from typing import Any, Callable, Mapping


class WebhookSecurityError(PermissionError):
    '''Raised when a webhook fails timestamp replay bounds or HMAC verification.'''
    pass


@dataclass(frozen=True)
class VerifiedWebhookEvent:
    event_id: str
    event_type: str
    timestamp_epoch: int
    payload: dict[str, Any]
    is_duplicate_delivery: bool


class ReplaySafeWebhookGuard:
    '''
    Enforces end-to-end webhook security and idempotency:
    1. Verifies HMAC-SHA256 over `f"{timestamp}." + raw_body_bytes` in constant time.
    2. Rejects timestamps older or newer than `tolerance_sec` (default 300s).
    3. Atomically deduplicates `event_id` in SQL so retries never double-execute.
    '''

    def __init__(
        self,
        conn: sqlite3.Connection,
        keyring: Mapping[str, bytes],
        *,
        active_kid: str = "v1",
        tolerance_sec: int = 300,
        max_body_bytes: int = 1024 * 1024,
    ) -> None:
        if not keyring or active_kid not in keyring:
            raise ValueError("active_kid must exist in keyring")
        self.conn = conn
        self.keyring = dict(keyring)
        self.active_kid = active_kid
        self.tolerance_sec = tolerance_sec
        self.max_body_bytes = max_body_bytes
        self._init_dedup_table()

    def _init_dedup_table(self) -> None:
        with self.conn:
            self.conn.execute(
                "CREATE TABLE IF NOT EXISTS webhook_dedup_ledger ("
                "  event_id TEXT PRIMARY KEY,"
                "  event_type TEXT NOT NULL,"
                "  received_at_epoch INTEGER NOT NULL"
                ")"
            )

    def sign_outbound_webhook(
        self,
        raw_body: bytes,
        *,
        timestamp_epoch: int | None = None,
    ) -> tuple[str, str]:
        '''
        Signs raw outbound webhook bytes bound to the current UTC epoch timestamp.
        Returns `(timestamp_header_str, signature_header_str)`.
        '''
        ts = int(time.time()) if timestamp_epoch is None else int(timestamp_epoch)
        secret = self.keyring[self.active_kid]
        signing_input = f"{ts}.".encode("ascii") + raw_body
        digest_hex = hmac.new(secret, signing_input, hashlib.sha256).hexdigest()
        return str(ts), f"{self.active_kid}={digest_hex}"

    def verify_and_claim_inbound(
        self,
        raw_body: bytes,
        *,
        timestamp_header: str,
        signature_header: str,
    ) -> VerifiedWebhookEvent:
        '''
        Verifies raw bytes, checks the 300-second replay window, parses JSON,
        and atomically claims `event_id` in the deduplication ledger.
        '''
        # Gate 1: Enforce raw payload byte ceiling before allocation/parsing
        if not isinstance(raw_body, bytes) or len(raw_body) > self.max_body_bytes:
            raise WebhookSecurityError("Webhook payload missing or exceeds byte limit")

        # Gate 2: Parse & validate timestamp replay window (+-300 seconds)
        try:
            ts_int = int(timestamp_header.strip(), 10)
        except (ValueError, TypeError) as exc:
            raise WebhookSecurityError("Malformed webhook timestamp header") from exc

        now_epoch = int(time.time())
        if abs(now_epoch - ts_int) > self.tolerance_sec:
            raise WebhookSecurityError(
                f"Webhook timestamp {ts_int} outside allowed {self.tolerance_sec}s replay window "
                f"(server_epoch={now_epoch})"
            )

        # Gate 3: Constant-time HMAC check across versioned signatures (`v1=<hex>,v2=<hex>`)
        signing_input = f"{ts_int}.".encode("ascii") + raw_body
        matched_signature = False

        for part in signature_header.split(","):
            if "=" not in part:
                continue
            kid, provided_hex = part.strip().split("=", 1)
            secret = self.keyring.get(kid)
            if secret is None:
                continue
            expected_hex = hmac.new(secret, signing_input, hashlib.sha256).hexdigest()
            if hmac.compare_digest(provided_hex.lower(), expected_hex.lower()):
                matched_signature = True
                break

        if not matched_signature:
            raise WebhookSecurityError("Webhook HMAC signature verification failed")

        # Gate 4: Parse JSON and atomically claim `event_id`
        try:
            payload = json.loads(raw_body.decode("utf-8"))
            event_id = str(payload["id"]).strip()
            event_type = str(payload.get("type", "unknown")).strip()
        except Exception as exc:
            raise WebhookSecurityError("Webhook body is not valid JSON or lacks 'id'") from exc

        if not event_id:
            raise WebhookSecurityError("Webhook event 'id' must not be empty")

        with self.conn:
            cursor = self.conn.execute(
                "INSERT OR IGNORE INTO webhook_dedup_ledger "
                "(event_id, event_type, received_at_epoch) VALUES (?, ?, ?)",
                (event_id, event_type, now_epoch),
            )
            is_duplicate = cursor.rowcount == 0

        return VerifiedWebhookEvent(
            event_id=event_id,
            event_type=event_type,
            timestamp_epoch=ts_int,
            payload=payload,
            is_duplicate_delivery=is_duplicate,
        )