from __future__ import annotations import hashlib import hmac import secrets from abc import ABC, abstractmethod from dataclasses import dataclass from app.core.config import Settings, get_settings @dataclass(frozen=True) class ProviderOrder: provider_order_id: str amount_paise: int currency: str @dataclass(frozen=True) class PaymentVerification: provider_order_id: str provider_payment_id: str amount_paise: int currency: str class PaymentGateway(ABC): @property @abstractmethod def provider_name(self) -> str: ... @property @abstractmethod def public_key_id(self) -> str: ... @abstractmethod def create_provider_order( self, *, amount_paise: int, currency: str, receipt: str, notes: dict[str, str], ) -> ProviderOrder: ... @abstractmethod def verify_checkout_signature( self, *, provider_order_id: str, provider_payment_id: str, signature: str, ) -> None: ... @abstractmethod def verify_webhook_signature(self, body: bytes, signature: str | None) -> None: ... @abstractmethod def parse_webhook_payment(self, payload: dict) -> PaymentVerification | None: ... class MockPaymentGateway(PaymentGateway): """Local/test gateway — signatures derived server-side only (never exposed to clients).""" def __init__(self, settings: Settings) -> None: self._settings = settings self._secret = settings.payment_mock_secret @property def provider_name(self) -> str: return "mock" @property def public_key_id(self) -> str: return "mock_public_key" def create_provider_order( self, *, amount_paise: int, currency: str, receipt: str, notes: dict[str, str], ) -> ProviderOrder: token = secrets.token_hex(6) return ProviderOrder( provider_order_id=f"mock_order_{receipt}_{token}", amount_paise=amount_paise, currency=currency, ) def _expected_signature(self, provider_order_id: str, provider_payment_id: str) -> str: message = f"{provider_order_id}|{provider_payment_id}".encode() return hmac.new(self._secret.encode(), message, hashlib.sha256).hexdigest() def verify_checkout_signature( self, *, provider_order_id: str, provider_payment_id: str, signature: str, ) -> None: expected = self._expected_signature(provider_order_id, provider_payment_id) if not hmac.compare_digest(expected, signature): raise ValueError("Invalid payment signature") def verify_webhook_signature(self, body: bytes, signature: str | None) -> None: if not signature: raise ValueError("Missing webhook signature") expected = hmac.new(self._secret.encode(), body, hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature): raise ValueError("Invalid webhook signature") def parse_webhook_payment(self, payload: dict) -> PaymentVerification | None: event = payload.get("event") if event != "payment.captured": return None entity = payload.get("payload", {}).get("payment", {}).get("entity", {}) if not entity: return None return PaymentVerification( provider_order_id=entity.get("order_id", ""), provider_payment_id=entity.get("id", ""), amount_paise=int(entity.get("amount", 0)), currency=entity.get("currency", "INR"), ) def build_mock_checkout_signature(self, provider_order_id: str, provider_payment_id: str) -> str: return self._expected_signature(provider_order_id, provider_payment_id) class RazorpayPaymentGateway(PaymentGateway): def __init__(self, settings: Settings) -> None: if not settings.razorpay_key_id or not settings.razorpay_key_secret: raise RuntimeError("Razorpay credentials are not configured") import razorpay self._client = razorpay.Client(auth=(settings.razorpay_key_id, settings.razorpay_key_secret)) self._key_id = settings.razorpay_key_id self._webhook_secret = settings.razorpay_webhook_secret or "" @property def provider_name(self) -> str: return "razorpay" @property def public_key_id(self) -> str: return self._key_id def create_provider_order( self, *, amount_paise: int, currency: str, receipt: str, notes: dict[str, str], ) -> ProviderOrder: data = self._client.order.create( { "amount": amount_paise, "currency": currency, "receipt": receipt, "notes": notes, } ) return ProviderOrder( provider_order_id=data["id"], amount_paise=int(data["amount"]), currency=data["currency"], ) def verify_checkout_signature( self, *, provider_order_id: str, provider_payment_id: str, signature: str, ) -> None: self._client.utility.verify_payment_signature( { "razorpay_order_id": provider_order_id, "razorpay_payment_id": provider_payment_id, "razorpay_signature": signature, } ) def verify_webhook_signature(self, body: bytes, signature: str | None) -> None: if not self._webhook_secret: raise RuntimeError("RAZORPAY_WEBHOOK_SECRET is not configured") if not signature: raise ValueError("Missing webhook signature") self._client.utility.verify_webhook_signature(body.decode(), signature, self._webhook_secret) def parse_webhook_payment(self, payload: dict) -> PaymentVerification | None: event = payload.get("event") if event not in ("payment.captured", "order.paid"): return None payment_entity = payload.get("payload", {}).get("payment", {}).get("entity") if not payment_entity and event == "order.paid": payment_entity = payload.get("payload", {}).get("order", {}).get("entity") if not payment_entity: return None order_id = payment_entity.get("order_id") or payment_entity.get("id") payment_id = payment_entity.get("id") if event == "payment.captured" else payment_entity.get("payment_id") if not order_id or not payment_id: return None return PaymentVerification( provider_order_id=order_id, provider_payment_id=payment_id, amount_paise=int(payment_entity.get("amount", 0)), currency=payment_entity.get("currency", "INR"), ) def get_payment_gateway(settings: Settings | None = None) -> PaymentGateway: settings = settings or get_settings() mode = settings.payment_provider.lower() if mode == "razorpay": return RazorpayPaymentGateway(settings) return MockPaymentGateway(settings)