131 lines
4 KiB
Python
131 lines
4 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Stripe Checkout service for billing credit top-ups.
|
|
Creates Checkout Sessions for redirect-based payment flow.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
|
|
|
from modules.shared.configuration import APP_CONFIG
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Server-side allowed amounts in CHF - never trust client
|
|
ALLOWED_AMOUNTS_CHF = [10, 25, 50, 100, 250, 500]
|
|
|
|
|
|
def _normalizeReturnUrl(returnUrl: str) -> str:
|
|
"""
|
|
Validate and normalize an absolute frontend return URL.
|
|
|
|
Allowed examples:
|
|
- https://nyla.poweron-center.net/billing/transactions
|
|
- https://nyla-int.poweron-center.net/billing/transactions?tab=overview
|
|
"""
|
|
if not returnUrl:
|
|
raise ValueError("returnUrl is required")
|
|
|
|
parsed = urlsplit(returnUrl.strip())
|
|
|
|
if parsed.scheme not in ("http", "https"):
|
|
raise ValueError("returnUrl must use http or https")
|
|
|
|
if not parsed.netloc:
|
|
raise ValueError("returnUrl must contain a host")
|
|
|
|
if parsed.username or parsed.password:
|
|
raise ValueError("returnUrl must not contain credentials")
|
|
|
|
query_items = [
|
|
(key, value)
|
|
for key, value in parse_qsl(parsed.query, keep_blank_values=True)
|
|
if key not in {"success", "canceled", "session_id"}
|
|
]
|
|
normalized_query = urlencode(query_items, doseq=True)
|
|
normalized_path = parsed.path or "/"
|
|
|
|
# Keep scheme + host + path + query, strip fragment for deterministic redirects.
|
|
return urlunsplit((parsed.scheme, parsed.netloc, normalized_path, normalized_query, ""))
|
|
|
|
|
|
def create_checkout_session(
|
|
mandate_id: str,
|
|
user_id: Optional[str],
|
|
amount_chf: float,
|
|
return_url: str
|
|
) -> str:
|
|
"""
|
|
Create a Stripe Checkout Session for credit top-up.
|
|
|
|
Amount and currency are validated server-side. The client-provided amount
|
|
must match an allowed preset.
|
|
|
|
Args:
|
|
mandate_id: Target mandate ID
|
|
user_id: Target user ID (for PREPAY_USER) or None (for mandate pool)
|
|
amount_chf: Amount in CHF (must be in ALLOWED_AMOUNTS_CHF)
|
|
|
|
Returns:
|
|
Stripe Checkout Session URL for redirect
|
|
|
|
Raises:
|
|
ValueError: If amount is invalid
|
|
"""
|
|
import stripe
|
|
|
|
# Validate amount server-side
|
|
if amount_chf not in ALLOWED_AMOUNTS_CHF:
|
|
raise ValueError(
|
|
f"Invalid amount {amount_chf} CHF. Allowed: {ALLOWED_AMOUNTS_CHF}"
|
|
)
|
|
|
|
from modules.shared.stripeClient import getStripeClient
|
|
stripe = getStripeClient()
|
|
|
|
base_return_url = _normalizeReturnUrl(return_url)
|
|
query_separator = "&" if "?" in base_return_url else "?"
|
|
success_url = f"{base_return_url}{query_separator}success=true&session_id={{CHECKOUT_SESSION_ID}}"
|
|
cancel_url = f"{base_return_url}{query_separator}canceled=true"
|
|
|
|
# Amount in cents for Stripe (CHF uses 2 decimal places)
|
|
amount_cents = int(round(amount_chf * 100))
|
|
|
|
metadata = {
|
|
"mandateId": mandate_id,
|
|
"amountChf": str(amount_chf),
|
|
}
|
|
if user_id:
|
|
metadata["userId"] = user_id
|
|
|
|
session = stripe.checkout.Session.create(
|
|
mode="payment",
|
|
line_items=[
|
|
{
|
|
"price_data": {
|
|
"currency": "chf",
|
|
"unit_amount": amount_cents,
|
|
"product_data": {
|
|
"name": "Guthaben aufladen",
|
|
"description": "AI Service Guthaben (CHF)",
|
|
},
|
|
},
|
|
"quantity": 1,
|
|
}
|
|
],
|
|
success_url=success_url,
|
|
cancel_url=cancel_url,
|
|
metadata=metadata,
|
|
)
|
|
|
|
if not session or not session.url:
|
|
raise ValueError("Stripe Checkout Session creation failed")
|
|
|
|
logger.info(
|
|
f"Created Stripe Checkout Session {session.id} for mandate {mandate_id}, "
|
|
f"amount {amount_chf} CHF"
|
|
)
|
|
|
|
return session.url
|