""" Cart Plugin services — all business logic lives here. CRUD functions own only DB operations; this layer orchestrates them. """ from __future__ import annotations import logging import secrets from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional from fastapi import Request from sqlalchemy.orm import Session from src.apps.cart_plugin import crud as cart_crud from src.apps.cart_plugin.models.cart_session import CartSession from src.apps.cart_plugin.schemas.cart_schemas import ( CartSubmitRequest, CreateCartSessionRequest, ValidateDiscountResponse, ) from src.core.exceptions import APIException, BadRequestError, ForbiddenError, NotFoundError logger = logging.getLogger(__name__) # ─── Key / token generation ─────────────────────────────────────────────────── def generate_public_key() -> str: """Return 'hwk_' + URL-safe random 32-byte token.""" return "hwk_" + secrets.token_urlsafe(32) def generate_session_token() -> str: """Return a URL-safe random 32-byte token.""" return secrets.token_urlsafe(32) def generate_webhook_secret() -> str: """Return a 32-byte hex webhook signing secret.""" return secrets.token_hex(32) # ─── Origin validation ──────────────────────────────────────────────────────── def validate_origin(origin: Optional[str], allowed_origins: Optional[list]) -> bool: """ Return True only if origin appears in allowed_origins. Normalise both sides: lowercase, strip trailing slash. Returns False when allowed_origins is empty, None, or origin is missing. """ if not allowed_origins: return False if not origin: return False def _norm(s: str) -> str: return s.lower().rstrip("/") norm_origin = _norm(origin) for allowed in allowed_origins: if _norm(allowed) == norm_origin: return True return False # ─── Total computation ──────────────────────────────────────────────────────── def compute_cart_totals( items: List[CartSessionItem], tip_amount_cents: int, tax_rate: float, discount_code: Optional[str], merchant_id: int, db: Session, ) -> Dict[str, int]: """ Server-side computation of all cart totals in integer cents. Returns: { subtotal, tip_amount, tax_amount, discount_amount, total } Never trusts client-supplied totals. """ subtotal_cents = sum(item.unit_price * item.quantity for item in items) tip_cents = max(0, tip_amount_cents) # Validate and apply discount discount_cents = 0 if discount_code: disc_result = _lookup_discount(discount_code, merchant_id, subtotal_cents, db) if disc_result["valid"]: discount_cents = disc_result["discount_cents"] taxable_base = max(0, subtotal_cents - discount_cents) tax_cents = int(round(taxable_base * tax_rate)) total_cents = max(0, subtotal_cents - discount_cents + tip_cents + tax_cents) return { "subtotal": subtotal_cents, "tip_amount": tip_cents, "tax_amount": tax_cents, "discount_amount": discount_cents, "total": total_cents, } def _lookup_discount( code: str, merchant_id: int, subtotal_cents: int, db: Session ) -> Dict[str, Any]: """Internal helper to look up a discount code for a merchant.""" from sqlalchemy import select from src.apps.merchants.models.merchant_discount import MerchantDiscount stmt = select(MerchantDiscount).where( MerchantDiscount.discount_id == code, MerchantDiscount.merchant_id == merchant_id, MerchantDiscount.is_active == True, MerchantDiscount.deleted_at.is_(None), ) discount = db.execute(stmt).scalar_one_or_none() if discount is None: return {"valid": False, "discount_cents": 0} if discount.discount_type == "percentage": amount = int(round(subtotal_cents * (discount.discount_value or 0) / 100)) else: # Fixed discount in cents (stored as cents in merchant_discount) amount = min(subtotal_cents, int(discount.discount_value or 0)) return {"valid": True, "discount_cents": max(0, amount)} # ─── Session management ─────────────────────────────────────────────────────── async def create_cart_session( public_key: str, origin: Optional[str], request_data: CreateCartSessionRequest, db: Session, request: Request, ) -> Dict[str, Any]: """ Validate the widget key, check origin, create session + items, emit cart.session_created event. """ # Resolve widget key widget_key = cart_crud.get_widget_key_by_public_key(db, public_key) if widget_key is None or not widget_key.is_active: raise ForbiddenError(message="Invalid or inactive widget key.") # Origin check if not validate_origin(origin, widget_key.allowed_origins): raise ForbiddenError( message="Origin not permitted by this widget key. " "Add the origin to the allowed_origins list in Cart Plugin settings." ) # HIGH-02: Validate return_url origin against allowed_origins. # Prevents open-redirect attacks where an attacker supplies a return_url # pointing to a malicious site not in the merchant's allow-list. if request_data.return_url is not None: from urllib.parse import urlparse as _urlparse def _norm_origin(s: str) -> str: return s.lower().rstrip("/") parsed_return = _urlparse(request_data.return_url) # Reconstruct just the origin portion (scheme + host + optional port) port_str = f":{parsed_return.port}" if parsed_return.port else "" return_url_origin = f"{parsed_return.scheme}://{parsed_return.hostname}{port_str}" allowed_norms = [_norm_origin(a) for a in (widget_key.allowed_origins or [])] if _norm_origin(return_url_origin) not in allowed_norms: raise BadRequestError( "return_url origin is not in the widget key's allowed_origins" ) # Idempotency guard if request_data.idempotency_key: existing = cart_crud.get_session_by_idempotency_key( db, widget_key.id, request_data.idempotency_key ) if existing is not None: # Return the existing session rather than creating a duplicate. subtotal = sum(i.unit_price * i.quantity for i in existing.items) return _build_create_response(existing, subtotal, request) token = generate_session_token() ttl = timedelta(minutes=request_data.ttl_minutes or 60) expires_at = datetime.now(timezone.utc) + ttl session = cart_crud.create_cart_session( db, widget_key.merchant_id, widget_key.id, token, request_data, expires_at, origin ) items = cart_crud.create_cart_session_items(db, session.id, request_data.items) # Compute initial subtotal for the response subtotal_cents = sum(i.unit_price * i.quantity for i in items) session.subtotal = subtotal_cents db.flush() db.commit() # Emit event (best-effort) try: from src.events.base import BaseEvent from src.events.dispatcher import EventDispatcher await EventDispatcher.dispatch( BaseEvent( event_type="cart.session_created", data={ "cart_session_id": session.id, "merchant_id": session.merchant_id, "widget_key_id": session.widget_key_id, "subtotal": subtotal_cents, "currency": session.currency, }, ) ) except Exception as exc: logger.warning("cart.session_created event dispatch failed: %s", exc) return _build_create_response(session, subtotal_cents, request) def _build_create_response(session: CartSession, subtotal_cents: int, request: Request) -> Dict[str, Any]: from src.core.config import settings # Checkout page lives in the frontend UI, not the API server frontend_url = settings.FRONTEND_BASE_URL.rstrip("/") checkout_url = f"{frontend_url}/cart/{session.token}" return { "token": session.token, "status": session.status, "currency": session.currency, "expires_at": session.expires_at, "items": session.items, "subtotal": subtotal_cents / 100.0, "checkout_url": checkout_url, } async def get_cart_session_public(token: str, db: Session) -> Dict[str, Any]: """ Load session for the checkout page. Returns 404 if not found, 410 if expired/cancelled. """ session = cart_crud.get_cart_session_with_items(db, token) if session is None: raise NotFoundError(message="Cart session not found.") if session.status in ("EXPIRED", "CANCELLED"): raise APIException(message="This cart session has expired or been cancelled.", status_code=410) if session.status == "PAID": raise APIException(message="This cart session has already been paid.", status_code=410) # Check wall-clock expiry if session.expires_at and datetime.now(timezone.utc) > session.expires_at.replace( tzinfo=timezone.utc if session.expires_at.tzinfo is None else session.expires_at.tzinfo ): raise APIException(message="This cart session has expired.", status_code=410) # Load merchant branding only — provider config is served by the /iframe-config endpoint. merchant_branding = _get_merchant_branding(session.merchant_id, db) return { "token": session.token, "status": session.status, "currency": session.currency, "expires_at": session.expires_at, "checkout_mode": session.checkout_mode, "return_url": session.return_url, "items": session.items, "subtotal": session.subtotal / 100.0, "tip_amount": session.tip_amount / 100.0, "tax_amount": session.tax_amount / 100.0, "discount_amount": session.discount_amount / 100.0, "total": session.total / 100.0, "discount_code": session.discount_code, "merchant_branding": merchant_branding, # MED-04: provider_txn_ref intentionally omitted from the public response. # It is only returned in the merchant-authenticated CartSessionDetailResponse. } def _get_merchant_branding(merchant_id: int, db: Session) -> Optional[Dict[str, Any]]: """Load lightweight branding info for the checkout page.""" try: from sqlalchemy import select from src.apps.merchants.models.merchant import Merchant stmt = select(Merchant).where(Merchant.id == merchant_id) merchant = db.execute(stmt).scalar_one_or_none() if merchant is None: return None return { "name": getattr(merchant, "name", None), "logo": getattr(merchant, "logo", None), "primary_color": getattr(merchant, "primary_color", None), } except Exception as exc: logger.warning("_get_merchant_branding: %s", exc) return None def _get_provider_config(merchant_id: int, db: Session) -> Optional[Dict[str, Any]]: """Load the merchant's active provider iframe config.""" try: from sqlalchemy import select from src.apps.payment_providers.models.merchant_provider_config import MerchantProviderConfig stmt = select(MerchantProviderConfig).where( MerchantProviderConfig.merchant_id == merchant_id, MerchantProviderConfig.is_active == True, ) mpc = db.execute(stmt).scalar_one_or_none() if mpc and mpc.config_data: cfg = mpc.config_data return { "api_key": cfg.get("api_key"), "merchant_id": cfg.get("merchant_id"), "mode": cfg.get("mode"), } except Exception as exc: logger.warning("_get_provider_config: %s", exc) return None async def get_cart_iframe_config(token: str, db: Session) -> Dict[str, Any]: """Return provider iframe config for the payment step.""" session = cart_crud.get_cart_session_by_token(db, token) if session is None: raise NotFoundError(message="Cart session not found.") if session.status != "PENDING": raise APIException(message="Session is not in a payable state.", status_code=410) from src.apps.merchants.models.merchant import Merchant from sqlalchemy import select stmt = select(Merchant).where(Merchant.id == session.merchant_id) merchant = db.execute(stmt).scalar_one_or_none() if merchant is None: raise NotFoundError(message="Merchant not found.") from src.core.providers.factory import get_provider_for_merchant provider, config = await get_provider_for_merchant(merchant, db) iframe_config = provider.get_iframe_config(config) return { "sdk_url": iframe_config.sdk_url, "api_key": iframe_config.api_key, "merchant_id": iframe_config.merchant_id, "mode": iframe_config.mode, "extra": iframe_config.extra, } async def validate_discount( token: str, code: str, db: Session ) -> Dict[str, Any]: """Validate a discount code against the merchant's active discounts.""" session = cart_crud.get_cart_session_by_token(db, token) if session is None: raise NotFoundError(message="Cart session not found.") if session.status != "PENDING": raise APIException(message="Session is not in a payable state.", status_code=410) from src.apps.merchants.models.merchant_discount import MerchantDiscount from sqlalchemy import select stmt = select(MerchantDiscount).where( MerchantDiscount.discount_id == code, MerchantDiscount.merchant_id == session.merchant_id, MerchantDiscount.is_active == True, MerchantDiscount.deleted_at.is_(None), ) discount = db.execute(stmt).scalar_one_or_none() if discount is None: return { "valid": False, "discount_type": None, "discount_value": None, "discount_amount": 0.0, "message": "Invalid or expired discount code.", } # Compute what the discount would save on the current subtotal subtotal_cents = session.subtotal or sum(i.unit_price * i.quantity for i in session.items) if discount.discount_type == "percentage": discount_cents = int(round(subtotal_cents * (discount.discount_value or 0) / 100)) else: discount_cents = min(subtotal_cents, int(discount.discount_value or 0)) return { "valid": True, "discount_type": discount.discount_type, "discount_value": float(discount.discount_value or 0), "discount_amount": discount_cents / 100.0, "message": None, } # ─── Submission ─────────────────────────────────────────────────────────────── async def submit_cart( token: str, request_data: CartSubmitRequest, db: Session, payer: Optional[dict] = None, ) -> Dict[str, Any]: """ Full cart submission flow: 1. Lock session with SELECT FOR UPDATE 2. Verify status == PENDING and not expired 3. Validate payer token merchant scope if provided 4. Recompute all totals server-side (integer cents) 5. Validate discount code if provided 6. Resolve provider for merchant 7. Call provider.submit_charge() with idempotency_key = session.token 8. Create a Transaction record 9. Update CartSession to PAID 10. Dispatch cart.payment_completed event """ # Step 1: Acquire row lock session = cart_crud.get_cart_session_with_items(db, token, for_update=True) if session is None: raise NotFoundError(message="Cart session not found.") # Step 2: Validate status and expiry if session.status != "PENDING": raise APIException( message=f"Cart session is not payable (status: {session.status}).", status_code=409 ) if session.expires_at: exp = session.expires_at if exp.tzinfo is None: exp = exp.replace(tzinfo=timezone.utc) if datetime.now(timezone.utc) > exp: session.status = "EXPIRED" db.commit() raise APIException(message="Cart session has expired.", status_code=410) # Step 3: Validate payer merchant scope if payer is not None: payer_merchant_id = payer.get("merchant_id") if payer_merchant_id and int(payer_merchant_id) != session.merchant_id: raise ForbiddenError(message="Payer token does not match session merchant.") # Step 4: Recompute totals server-side tip_cents = int(round((request_data.tip_amount or 0.0) * 100)) # Use a 0% tax rate here unless the merchant has a billing address zip to look up. # Tax is currently informational at session level; provider charges the total. totals = compute_cart_totals( items=session.items, tip_amount_cents=tip_cents, tax_rate=0.0, discount_code=request_data.discount_code, merchant_id=session.merchant_id, db=db, ) total_cents = totals["total"] total_dollars = total_cents / 100.0 # Step 5: Load merchant from sqlalchemy import select from src.apps.merchants.models.merchant import Merchant stmt = select(Merchant).where(Merchant.id == session.merchant_id) merchant = db.execute(stmt).scalar_one_or_none() if merchant is None: raise NotFoundError(message="Merchant not found.") # Step 6 + 7: Submit charge. # IMPORTANT: Check the dev/test bypass BEFORE resolving the provider so that # test tokens work even when the merchant has no configured provider or # onboarding is incomplete. from src.core.providers.base import ChargeResult from src.core.config import settings as _settings _is_test_token = str(request_data.payment_token).startswith("tok_test_") _is_dev_env = _settings.APP_ENV in ("dev", "test") if _is_dev_env and _is_test_token: # Dev/test bypass: simulate a successful charge without hitting the real provider. charge_result = ChargeResult( transaction_id=f"dev_txn_{session.id}_{int(datetime.now(timezone.utc).timestamp())}", status="succeeded", amount=total_dollars, currency=session.currency, raw_response={"mock": True, "note": "dev-mode simulated charge"}, ) else: from src.core.providers.factory import get_provider_for_merchant provider, config = await get_provider_for_merchant(merchant, db) charge_result = await provider.submit_charge( config=config, amount=total_dollars, currency=session.currency, payment_method_token=request_data.payment_token, payment_method_type=request_data.payment_method_type or "card", capture=True, idempotency_key=session.token, metadata={ "cart_session_token": session.token, "cart_session_id": session.id, "source": "cart_plugin", }, ) if charge_result.status == "failed": raise APIException( message="Payment charge failed. Please try a different payment method.", status_code=402, ) # Step 8: Store provider transaction reference on the session. billing_info_dict = ( request_data.billing_info.model_dump() if request_data.billing_info else None ) provider_txn_ref = charge_result.transaction_id # Extract card details from the raw provider response. # TSYS puts card fields under SaleResponse / PreAuthResponse / AchResponse. # The dev-mode mock response has none of these keys, so all values will be None. _raw = charge_result.raw_response or {} _sale_resp = ( _raw.get("SaleResponse") or _raw.get("PreAuthResponse") or _raw.get("AchResponse") or {} ) _provider_slug = ( "dev_stub" if (_is_dev_env and _is_test_token) else config.provider_slug ) # Step 9: Mark session PAID cart_crud.update_cart_session_paid( db, session, transaction_id=None, # Transaction row is created async in the event listener subtotal_cents=totals["subtotal"], tip_cents=totals["tip_amount"], tax_cents=totals["tax_amount"], discount_cents=totals["discount_amount"], total_cents=total_cents, customer_name=request_data.customer_name, customer_email=str(request_data.customer_email) if request_data.customer_email else None, billing_info=billing_info_dict, discount_code=request_data.discount_code, ) # Persist provider reference and card details. # HIGH-03: Only whitelisted fields are stored — never the full raw_response, # which may contain PAN fragments, CVV echoes, or internal routing data. session.provider_txn_ref = provider_txn_ref safe_provider_data = { "provider_status": charge_result.status, "provider_slug": _provider_slug, "card_type": _sale_resp.get("cardType"), "masked_card_number": _sale_resp.get("maskedCardNumber"), "authorization_code": ( _sale_resp.get("authorizationCode") or _sale_resp.get("authCode") ), "host_reference_number": _sale_resp.get("hostReferenceNumber"), "response_code": ( _sale_resp.get("approvalCode") or _sale_resp.get("responseCode") ), "avs_response_code": _sale_resp.get("AVSResponseCode"), "cvv_response_code": _sale_resp.get("CVVResponseCode"), } session.metadata_ = { **(session.metadata_ or {}), "provider_data": {k: v for k, v in safe_provider_data.items() if v is not None}, } db.flush() db.commit() # Step 10: Dispatch event (best-effort) try: from src.events.base import BaseEvent from src.events.dispatcher import EventDispatcher await EventDispatcher.dispatch( BaseEvent( event_type="cart.payment_completed", data={ "cart_session_id": session.id, "cart_session_token": session.token, "merchant_id": session.merchant_id, "provider_txn_ref": provider_txn_ref, "amount": total_cents, "currency": session.currency, "customer_email": ( str(request_data.customer_email) if request_data.customer_email else None ), # Provider card details — used by the listener to create ProviderTransaction "provider_slug": _provider_slug, "card_type": _sale_resp.get("cardType"), "masked_card_number": _sale_resp.get("maskedCardNumber"), "authorization_code": ( _sale_resp.get("authorizationCode") or _sale_resp.get("authCode") ), "host_reference_number": _sale_resp.get("hostReferenceNumber"), "response_code": ( _sale_resp.get("approvalCode") or _sale_resp.get("responseCode") ), "avs_response_code": _sale_resp.get("AVSResponseCode"), "cvv_response_code": _sale_resp.get("CVVResponseCode"), }, ) ) except Exception as exc: logger.warning("cart.payment_completed event dispatch failed: %s", exc) return { "cart_session_id": session.id, "provider_txn_ref": provider_txn_ref, "amount": total_dollars, "currency": session.currency, "status": charge_result.status, "redirect_url": session.return_url, }