Spaces:
Running
Running
File size: 3,951 Bytes
9afc3bb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | """Processor-agnostic payment layer.
Skill: the processor is undecided (Stripe vs Razorpay vs self-built all open), and
that being unresolved is NOT a reason to leave payments unaddressed. So this defines
a swappable PaymentProvider interface + a reference provider, so plugging in a real
processor later is one class, not a rearchitect. It must support the range of
methods real subscribers use (cards, UPI, international) — the interface carries a
`methods` list the chosen processor fills in.
Plan activation is NEVER done here off a client signal — providers only create a
checkout and verify a processor event; activation happens in subscription.py behind
the verified webhook.
"""
from __future__ import annotations
import abc
import hashlib
import hmac
import os
class CheckoutSession:
def __init__(self, order_id: str, redirect_url: str, processor_ref: str):
self.order_id = order_id
self.redirect_url = redirect_url
self.processor_ref = processor_ref
def to_dict(self) -> dict:
return {"order_id": self.order_id, "redirect_url": self.redirect_url,
"processor_ref": self.processor_ref}
class PaymentProvider(abc.ABC):
name: str = "abstract"
# Methods this processor supports for the current subscriber region. Real
# processors populate this; the frontend renders the selection from it.
methods: list[str] = ["card", "upi"]
@abc.abstractmethod
def create_checkout(self, order_id: str, amount_cents: int, currency: str,
plan_id: str) -> CheckoutSession: ...
@abc.abstractmethod
def verify_webhook(self, raw_body: bytes, headers: dict) -> dict | None:
"""Return a normalized event {event_id, order_id, status} if the payload is
a genuine, signature-valid processor event; None if it can't be trusted.
This is the trusted server-side confirmation activation depends on."""
class ReferenceProvider(PaymentProvider):
"""A real, signature-checked reference implementation used until a live
processor is chosen. It signs/verifies with an HMAC secret from the
environment (never hardcoded — credentials-and-secrets.md), so the webhook
path is genuinely authenticated and the idempotent/atomic flow is fully
testable end to end now, per the skill's 'verify live' rule."""
name = "reference"
methods = ["card", "upi", "netbanking", "wallet"]
def _secret(self) -> bytes:
return os.environ.get("PAYMENT_WEBHOOK_SECRET", "dev-only-unset").encode()
def create_checkout(self, order_id, amount_cents, currency, plan_id) -> CheckoutSession:
# A real processor returns a hosted redirect URL; here we mint a deterministic
# ref so the webhook can be correlated and signed.
ref = hashlib.sha256(f"{self.name}:{order_id}".encode()).hexdigest()[:24]
return CheckoutSession(order_id, f"/billing/checkout/{order_id}", ref)
def sign(self, raw_body: bytes) -> str:
return hmac.new(self._secret(), raw_body, hashlib.sha256).hexdigest()
def verify_webhook(self, raw_body: bytes, headers: dict) -> dict | None:
import json
sig = headers.get("x-payment-signature") or headers.get("X-Payment-Signature")
if not sig or not hmac.compare_digest(sig, self.sign(raw_body)):
return None
try:
payload = json.loads(raw_body.decode())
except Exception:
return None
if not payload.get("event_id") or not payload.get("order_id"):
return None
return {"event_id": payload["event_id"], "order_id": payload["order_id"],
"status": payload.get("status", "confirmed")}
_PROVIDERS = {"reference": ReferenceProvider()}
def get_provider(name: str | None = None) -> PaymentProvider:
chosen = (name or os.environ.get("PAYMENT_PROCESSOR", "reference")).lower()
return _PROVIDERS.get(chosen, _PROVIDERS["reference"])
|