227 lines
8.2 KiB
Python
227 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Protocol
|
|
|
|
|
|
class PaymentWebhookVerificationError(Exception):
|
|
def __init__(self, signature_status: str, message: str, *, status_code: int = 400) -> None:
|
|
super().__init__(message)
|
|
self.signature_status = signature_status
|
|
self.message = message
|
|
self.status_code = status_code
|
|
|
|
|
|
class PaymentProviderAdapter(Protocol):
|
|
def get_status(self, external_payment_id: str, *, context: dict[str, Any] | None = None) -> dict[str, Any] | str:
|
|
...
|
|
|
|
|
|
_PROVIDER_ADAPTERS: dict[str, PaymentProviderAdapter] = {}
|
|
|
|
|
|
def register_payment_provider_adapter(provider: str, adapter: PaymentProviderAdapter) -> None:
|
|
normalized = normalize_provider(provider)
|
|
if normalized:
|
|
_PROVIDER_ADAPTERS[normalized] = adapter
|
|
|
|
|
|
def get_payment_provider_adapter(provider: str | None) -> PaymentProviderAdapter | None:
|
|
return _PROVIDER_ADAPTERS.get(normalize_provider(provider))
|
|
|
|
|
|
def normalize_provider(provider: str | None) -> str:
|
|
return str(provider or "").strip().lower() or "manual"
|
|
|
|
|
|
def raw_payload_hash(raw_body: bytes) -> str:
|
|
return hashlib.sha256(raw_body).hexdigest()
|
|
|
|
|
|
def safe_json_loads(raw_body: bytes) -> dict[str, Any]:
|
|
if not raw_body:
|
|
return {}
|
|
parsed = json.loads(raw_body.decode("utf-8"))
|
|
if not isinstance(parsed, dict):
|
|
raise ValueError("Webhook payload must be a JSON object")
|
|
return parsed
|
|
|
|
|
|
def normalize_headers(headers: Any) -> dict[str, str]:
|
|
return {str(key).lower(): str(value) for key, value in dict(headers).items()}
|
|
|
|
|
|
def header_value(headers: dict[str, str], *names: str) -> str | None:
|
|
for name in names:
|
|
value = headers.get(name.lower())
|
|
normalized = str(value or "").strip()
|
|
if normalized:
|
|
return normalized
|
|
return None
|
|
|
|
|
|
def metadata_value(payload: dict[str, Any] | None, *keys: str) -> str | None:
|
|
data = payload or {}
|
|
metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
|
|
candidates = [data, metadata]
|
|
payment = data.get("payment") if isinstance(data.get("payment"), dict) else {}
|
|
candidates.append(payment)
|
|
nested_metadata = payment.get("metadata") if isinstance(payment.get("metadata"), dict) else {}
|
|
candidates.append(nested_metadata)
|
|
data_object = data.get("data", {}).get("object") if isinstance(data.get("data"), dict) else {}
|
|
if isinstance(data_object, dict):
|
|
candidates.append(data_object)
|
|
|
|
for source in candidates:
|
|
for key in keys:
|
|
value = source.get(key)
|
|
normalized = str(value or "").strip()
|
|
if normalized:
|
|
return normalized
|
|
return None
|
|
|
|
|
|
def extract_payment_provider(payload: dict[str, Any], headers: dict[str, str]) -> str:
|
|
return normalize_provider(
|
|
header_value(headers, "x-payment-provider", "x-provider")
|
|
or metadata_value(payload, "payment_provider", "provider")
|
|
)
|
|
|
|
|
|
def extract_provider_account_id(payload: dict[str, Any], headers: dict[str, str]) -> str | None:
|
|
return header_value(
|
|
headers,
|
|
"x-provider-account-id",
|
|
"x-merchant-id",
|
|
"x-terminal-id",
|
|
"x-integration-id",
|
|
) or metadata_value(
|
|
payload,
|
|
"provider_account_id",
|
|
"payment_provider_account_id",
|
|
"merchant_id",
|
|
"terminal_id",
|
|
"account_id",
|
|
"integration_id",
|
|
)
|
|
|
|
|
|
def extract_external_event_id(payload: dict[str, Any], headers: dict[str, str]) -> str | None:
|
|
return header_value(headers, "x-webhook-event-id", "x-event-id") or metadata_value(
|
|
payload,
|
|
"external_event_id",
|
|
"webhook_event_id",
|
|
"event_id",
|
|
)
|
|
|
|
|
|
def extract_external_payment_id(payload: dict[str, Any], headers: dict[str, str]) -> str | None:
|
|
return header_value(headers, "x-payment-id", "x-external-payment-id") or metadata_value(
|
|
payload,
|
|
"external_payment_id",
|
|
"external_payment_ref",
|
|
"transaction_id",
|
|
"provider_payment_id",
|
|
"payment_id",
|
|
"id",
|
|
)
|
|
|
|
|
|
def extract_event_type(payload: dict[str, Any], headers: dict[str, str]) -> str:
|
|
return (
|
|
header_value(headers, "x-webhook-event-type", "x-event-type")
|
|
or metadata_value(payload, "event_type", "type")
|
|
or normalize_payment_event_type(None, metadata_value(payload, "status", "provider_status"))
|
|
)
|
|
|
|
|
|
def normalize_payment_status(provider: str | None, provider_status: str | None) -> str:
|
|
normalized = str(provider_status or "").strip().lower()
|
|
normalized = normalized.replace("-", "_").replace(" ", "_")
|
|
if normalized in {"succeeded", "success", "paid", "captured", "settled", "completed"}:
|
|
return "success"
|
|
if normalized in {"authorized", "authorised", "processing", "pending", "created", "new"}:
|
|
return "pending"
|
|
if normalized in {"failed", "declined", "error", "rejected"}:
|
|
return "failed"
|
|
if normalized in {"canceled", "cancelled", "voided", "expired"}:
|
|
return "canceled"
|
|
if normalized in {"partial", "partially_paid", "partial_paid"}:
|
|
return "partial"
|
|
return normalized if normalized in {"pending", "success", "failed", "canceled", "partial"} else "pending"
|
|
|
|
|
|
def normalize_payment_event_type(provider: str | None, provider_status: str | None, event_type: str | None = None) -> str:
|
|
raw_event_type = str(event_type or "").strip().lower()
|
|
if raw_event_type.startswith("payment."):
|
|
if raw_event_type in {"payment.succeeded", "payment.captured", "payment.paid"}:
|
|
return "payment.received"
|
|
if raw_event_type in {"payment.cancelled", "payment.canceled"}:
|
|
return "payment.canceled"
|
|
return raw_event_type
|
|
|
|
status = normalize_payment_status(provider, provider_status)
|
|
if status in {"success", "partial"}:
|
|
return "payment.received"
|
|
if status == "failed":
|
|
return "payment.failed"
|
|
if status == "canceled":
|
|
return "payment.canceled"
|
|
return "payment.pending"
|
|
|
|
|
|
def _signature_candidates(signature_header: str) -> set[str]:
|
|
raw = str(signature_header or "").strip()
|
|
if not raw:
|
|
return set()
|
|
candidates = {raw}
|
|
if raw.startswith("sha256="):
|
|
candidates.add(raw.split("=", 1)[1])
|
|
for part in raw.split(","):
|
|
key, sep, value = part.partition("=")
|
|
if sep and key.strip() in {"v1", "sha256"} and value.strip():
|
|
candidates.add(value.strip())
|
|
return candidates
|
|
|
|
|
|
def verify_payment_webhook_signature(
|
|
*,
|
|
raw_body: bytes,
|
|
headers: dict[str, str],
|
|
secret: str | None,
|
|
replay_window_seconds: int = 600,
|
|
now: datetime | None = None,
|
|
required: bool = True,
|
|
) -> str:
|
|
normalized_secret = str(secret or "").strip()
|
|
if not required and not normalized_secret:
|
|
return "not_required"
|
|
if not normalized_secret:
|
|
raise PaymentWebhookVerificationError("missing_secret", "Payment webhook secret is not configured", status_code=403)
|
|
|
|
timestamp = header_value(headers, "x-webhook-timestamp", "x-provider-timestamp", "x-timestamp")
|
|
if timestamp:
|
|
try:
|
|
event_ts = int(float(timestamp))
|
|
except ValueError as exc:
|
|
raise PaymentWebhookVerificationError("invalid_timestamp", "Invalid webhook timestamp") from exc
|
|
now_ts = int((now or datetime.now(timezone.utc)).timestamp())
|
|
if abs(now_ts - event_ts) > max(int(replay_window_seconds), 1):
|
|
raise PaymentWebhookVerificationError("expired", "Webhook timestamp is outside the replay window")
|
|
|
|
signature_header = header_value(headers, "x-webhook-signature", "x-signature", "stripe-signature")
|
|
if not signature_header:
|
|
raise PaymentWebhookVerificationError("missing", "Payment webhook signature is missing", status_code=403)
|
|
|
|
body_to_sign = raw_body
|
|
if timestamp:
|
|
body_to_sign = f"{timestamp}.".encode("utf-8") + raw_body
|
|
expected = hmac.new(normalized_secret.encode("utf-8"), body_to_sign, hashlib.sha256).hexdigest()
|
|
for candidate in _signature_candidates(signature_header):
|
|
if hmac.compare_digest(candidate, expected):
|
|
return "valid"
|
|
raise PaymentWebhookVerificationError("invalid", "Payment webhook signature is invalid", status_code=403)
|