"""Signed and idempotent provider webhook contract.""" from __future__ import annotations import hashlib import hmac import json import os import re from datetime import datetime, timedelta, timezone from typing import Any from sqlalchemy import or_, select from sqlalchemy.exc import IntegrityError from config.control_plane import get_control_plane_session from bridge_platform.tenants.models import ( BillingProductMapping, BillingWebhookEvent, Tenant, TenantSubscription, ) SIGNATURE_TOLERANCE_SECONDS = 300 SUPPORTED_STATUSES = { "active", "trialing", "past_due", "canceled", "paused", "unpaid", "incomplete", "incomplete_expired", } class BillingWebhookError(ValueError): pass def verify_signature(provider: str, raw_body: bytes, timestamp: str, signature: str) -> None: secret = os.getenv(f"BILLING_WEBHOOK_SECRET_{_env_name(provider)}") if not secret: raise BillingWebhookError("Billing webhook is not configured") try: sent_at = datetime.fromtimestamp(int(timestamp), tz=timezone.utc) except (TypeError, ValueError, OSError): raise BillingWebhookError("Invalid webhook timestamp") from None if abs((datetime.now(timezone.utc) - sent_at).total_seconds()) > SIGNATURE_TOLERANCE_SECONDS: raise BillingWebhookError("Webhook timestamp is outside the allowed window") expected = hmac.new( secret.encode("utf-8"), timestamp.encode("ascii") + b"." + raw_body, hashlib.sha256, ).hexdigest() supplied = signature.removeprefix("sha256=").strip().lower() if not hmac.compare_digest(expected, supplied): raise BillingWebhookError("Invalid webhook signature") def process_event(provider: str, payload: dict[str, Any], raw_body: bytes) -> dict[str, Any]: event_id = str(payload.get("id") or "").strip() event_type = str(payload.get("type") or "").strip() data = payload.get("data") or {} if not event_id or not event_type or not isinstance(data, dict): raise BillingWebhookError("Webhook requires id, type, and data") payload_hash = hashlib.sha256(raw_body).hexdigest() with get_control_plane_session() as session: existing = session.execute( select(BillingWebhookEvent).where( BillingWebhookEvent.provider == provider, BillingWebhookEvent.provider_event_id == event_id, ) ).scalar_one_or_none() if existing: if existing.payload_hash != payload_hash: raise BillingWebhookError("Event ID was reused with a different payload") return {"event_id": event_id, "status": existing.status, "duplicate": True} receipt = BillingWebhookEvent( provider=provider, provider_event_id=event_id, event_type=event_type, payload_hash=payload_hash, status="received", ) session.add(receipt) try: session.flush() _apply_subscription_event(session, provider, event_type, data) receipt.status = "processed" receipt.processed_at = datetime.utcnow() session.commit() except Exception as exc: session.rollback() _persist_failed_receipt(provider, event_id, event_type, payload_hash, str(exc)) if isinstance(exc, BillingWebhookError): raise raise BillingWebhookError("Billing event could not be processed") from exc return {"event_id": event_id, "status": "processed", "duplicate": False} def _apply_subscription_event(session, provider: str, event_type: str, data: dict[str, Any]) -> None: if not event_type.startswith("subscription."): return status = str(data.get("status") or event_type.split(".", 1)[1]).lower() if status not in SUPPORTED_STATUSES: raise BillingWebhookError(f"Unsupported subscription status: {status}") tenant_id = str(data.get("tenant_id") or "").strip() customer_id = str(data.get("customer_id") or "").strip() or None subscription_id = str(data.get("subscription_id") or "").strip() or None subscription = session.execute( select(TenantSubscription).where( or_( TenantSubscription.tenant_id == tenant_id if tenant_id else False, TenantSubscription.provider_customer_id == customer_id if customer_id else False, TenantSubscription.provider_subscription_id == subscription_id if subscription_id else False, ) ).limit(1) ).scalar_one_or_none() if subscription is None: if not tenant_id or session.get(Tenant, tenant_id) is None: raise BillingWebhookError("Webhook cannot resolve a registered tenant") price_id = str(data.get("price_id") or "").strip() mapping = _price_mapping(session, provider, price_id) subscription = TenantSubscription( tenant_id=tenant_id, plan_id=mapping.plan_id, provider=provider, ) session.add(subscription) elif data.get("price_id"): subscription.plan_id = _price_mapping(session, provider, str(data["price_id"])).plan_id subscription.provider = provider subscription.provider_customer_id = customer_id or subscription.provider_customer_id subscription.provider_subscription_id = subscription_id or subscription.provider_subscription_id subscription.status = status subscription.current_period_start = _provider_datetime(data.get("current_period_start")) subscription.current_period_end = _provider_datetime(data.get("current_period_end")) subscription.cancel_at_period_end = bool(data.get("cancel_at_period_end", False)) metadata = dict(subscription.metadata_json or {}) if status == "past_due": grace_days = max(0, int(os.getenv("BILLING_GRACE_PERIOD_DAYS", "3"))) metadata["grace_until"] = (datetime.now(timezone.utc) + timedelta(days=grace_days)).isoformat() else: metadata.pop("grace_until", None) metadata["last_provider_event_type"] = event_type subscription.metadata_json = metadata def _price_mapping(session, provider: str, price_id: str) -> BillingProductMapping: mapping = session.execute( select(BillingProductMapping).where( BillingProductMapping.provider == provider, BillingProductMapping.price_id == price_id, BillingProductMapping.is_active.is_(True), ) ).scalar_one_or_none() if mapping is None: raise BillingWebhookError(f"No active plan mapping for price: {price_id}") return mapping def _persist_failed_receipt(provider: str, event_id: str, event_type: str, payload_hash: str, message: str) -> None: with get_control_plane_session() as session: session.add(BillingWebhookEvent( provider=provider, provider_event_id=event_id, event_type=event_type, payload_hash=payload_hash, status="failed", error_message=message[:2000], processed_at=datetime.utcnow(), )) try: session.commit() except IntegrityError: session.rollback() def _provider_datetime(value: Any) -> datetime | None: if value in (None, ""): return None if isinstance(value, (int, float)): return datetime.utcfromtimestamp(value) try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) except ValueError: raise BillingWebhookError("Invalid subscription period timestamp") from None return parsed.astimezone(timezone.utc).replace(tzinfo=None) if parsed.tzinfo else parsed def _env_name(provider: str) -> str: return re.sub(r"[^A-Z0-9]+", "_", provider.upper()).strip("_")