"""Shared webhook hardening helpers (``API-STANDARD.md`` ยง10).""" from __future__ import annotations import time from typing import Any from infrasynth.shared.exceptions import ValidationAppError __all__ = ["assert_fresh_webhook"] def assert_fresh_webhook(timestamp: Any, *, tolerance_seconds: int = 300) -> None: """Rejects a webhook whose timestamp is older than the tolerance window. Replay protection is independent of signature validity: a correctly signed event replayed after the window is still rejected. """ try: event_ts = float(timestamp) except (TypeError, ValueError) as exc: raise ValidationAppError( "Webhook timestamp is missing or invalid.", code="VALIDATION_WEBHOOK_TIMESTAMP", status=400, ) from exc if abs(time.time() - event_ts) > tolerance_seconds: raise ValidationAppError( "Webhook timestamp is outside the accepted tolerance window.", code="VALIDATION_WEBHOOK_STALE", status=400, details=[{"issue": "stale", "toleranceSeconds": tolerance_seconds}], )