infrasynth-backend-kit/infrasynth/api/webhooks.py
jcv-dev 551b42eab5 feat: production-hardening pass across the kit
Close the gaps between the documented contract (API-STANDARD, TENANCY,
ENTITLEMENTS) and the implementation, and remove committed build artifacts.

Security:
- verify + process inbound webhooks (HMAC/handler verify, size limit,
  timestamp tolerance, idempotency via InboundEvent.external_id)
- real 2FA login flow (pre-auth challenge; tokens only after verify/recovery)
- wire HybridPermission into security/audit views; add API-key rotate and
  users/<id>/permissions|roles endpoints
- tenant-scoped throttling on by default; webhook replay protection
- verify MercadoPago webhook signatures
- login brute-force guard, configurable password policy, real ALTCHA PoW

Correctness:
- apply verified billing webhooks idempotently (subscription/entitlement/
  invoice/PaymentTransaction); scheduled payment lifecycle jobs
- capture audit update diffs automatically; add audit retention purge
- working notification retries, per-channel rate limits, log retention
- pluggable virus scanner, upload-size limit, pipeline toggle
- feature rollout %/environment targeting; settings-driven registrations
- workflow guards (instance cap, route depth, self-assignment, clone on re-entry)
- wire every previously-dead INFRASYNTH_* setting; drop truly dead ones

Delivery:
- README + CHANGELOG; CI format check + coverage gate
- keep test media out of the tree; untrack .coverage, __pycache__,
  egg-info, docs/ and invoice artifacts
2026-09-24 10:41:21 -05:00

34 lines
1.1 KiB
Python

"""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}],
)