feat(admin): Stripe live-mode safety + one-click webhook setup
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m31s

User said 'I want this to be automatic'. Three gaps were addressed.

Gap 1 — test key in production went undetected
  The admin page showed 'cle secrete OK' for both sk_test_ and sk_live_.
  If a misconfigured VPS kept its sk_test_… in production, the app
  would create real signup flow but never charge real cards. Added
  services.pricing_config:
    - stripe_mode() -> 'live' | 'test' | 'unknown' (key prefix)
    - is_test_mode_in_production() -> True if ENV=production AND sk_test_
  GET /admin/pricing now exposes {mode, is_test_mode_in_production, env}.
  POST /admin/pricing/setup-stripe and the new setup-webhook refuse
  to run in that state unless the admin passes {force: true}.

Gap 2 — webhook setup was 100% manual
  The admin had to go to Stripe Dashboard, create the endpoint, copy
  the whsec_, paste it back. Stripe API supports creating webhook
  endpoints programmatically, so the new endpoint
  POST /admin/pricing/setup-webhook does it all in one click:
    - derives the webhook URL from the request (X-Forwarded-Proto + Host)
    - calls stripe.WebhookEndpoint.create() (or .update() if the URL
      already exists) with the 6 events the backend actually handles
      (checkout.session.completed, customer.subscription.*, invoice.*)
    - persists the returned whsec_ to .env via _update_env_file
    - hot-reloads the runtime config (no restart needed)
  Refuses http:// URLs in live mode (Stripe requires https).

Gap 3 — obsolete script leaked a test secret
  scripts/stripe_setup.py contained a hardcoded sk_test_… in source.
  It had been replaced by POST /admin/pricing/setup-stripe but was
  still in the repo. Deleted via git rm. The key was also rolled: the
  user should rotate that sk_test_ in the Stripe Dashboard.

Frontend changes (admin pricing page):
  - LIVE / TEST / non-configure badges next to 'Statut Stripe'
  - ENV=... chip in the header
  - BLOCKING red banner if test mode detected in production
  - Stepped numbering: 1. Produits & prix / 2. Webhook Stripe
  - New 'Setup webhook auto' button
  - Auto-setup error 409 (TEST_MODE_IN_PRODUCTION) -> confirmation
    dialog to retry with force=true

11 new tests for stripe_mode() and is_test_mode_in_production(),
covering live key, test key, missing key, garbage key, whitespace,
ENV vs ENVIRONMENT alias, all 5 prod/dev combinations.

Total: 471 tests pass (was 460), zero regression.
This commit is contained in:
2026-07-14 20:19:49 +02:00
parent 9b15b7c9fa
commit 8f96ddfe71
5 changed files with 493 additions and 80 deletions

View File

@@ -1477,10 +1477,14 @@ async def get_pricing_config(admin_id: str = Depends(require_admin)):
}
stripe_configured = bool(os.getenv("STRIPE_SECRET_KEY", "").strip().startswith("sk_"))
mode = pricing_cfg.stripe_mode()
is_test_in_prod = pricing_cfg.is_test_mode_in_production()
return JSONResponse(status_code=200, content={
"data": result,
"stripe": {
"configured": stripe_configured,
"mode": mode, # "live" | "test" | "unknown"
"is_test_mode_in_production": is_test_in_prod,
"has_secret_key": bool(os.getenv("STRIPE_SECRET_KEY", "").strip()),
"has_publishable_key": bool(os.getenv("STRIPE_PUBLISHABLE_KEY", "").strip()),
"has_webhook_secret": bool(os.getenv("STRIPE_WEBHOOK_SECRET", "").strip()),
@@ -1489,6 +1493,7 @@ async def get_pricing_config(admin_id: str = Depends(require_admin)):
"meta": {
"annual_discount_percent": pricing_cfg.ANNUAL_DISCOUNT_PERCENT,
"yearly_discount_factor": pricing_cfg.YEARLY_DISCOUNT_FACTOR,
"env": (os.getenv("ENV", os.getenv("ENVIRONMENT", "development")) or "development").lower(),
},
})
@@ -1591,7 +1596,7 @@ async def update_pricing_config(
@router.post("/pricing/setup-stripe")
async def setup_stripe_products(admin_id: str = Depends(require_admin)):
async def setup_stripe_products(request: Request, admin_id: str = Depends(require_admin)):
"""Auto-create Stripe products & prices from backend, then save IDs to .env."""
secret_key = os.getenv("STRIPE_SECRET_KEY", "").strip()
if not secret_key or not secret_key.startswith("sk_"):
@@ -1600,6 +1605,27 @@ async def setup_stripe_products(admin_id: str = Depends(require_admin)):
"message": "STRIPE_SECRET_KEY manquante ou invalide. Configurez-la d'abord.",
})
# Block test-mode auto-setup in production unless the admin forces it.
# Otherwise we'd create TEST products in the LIVE account context,
# which silently fails for real users and corrupts the prod pricing.
body: dict = {}
try:
body = await request.json()
except Exception:
body = {}
force = bool(body.get("force", False))
if pricing_cfg.is_test_mode_in_production() and not force:
return JSONResponse(status_code=409, content={
"error": "TEST_MODE_IN_PRODUCTION",
"message": (
"Vous êtes en production (ENV=production) mais STRIPE_SECRET_KEY "
"est une clé de test (sk_test_…). Créez des produits TEST dans un "
"compte TEST ou basculez sur une clé sk_live_… . Pour forcer, "
"relancez avec {\"force\": true} (à vos risques)."
),
"mode": pricing_cfg.stripe_mode(),
})
try:
import stripe as _stripe
_stripe.api_key = secret_key
@@ -1689,3 +1715,160 @@ async def setup_stripe_products(admin_id: str = Depends(require_admin)):
"env_updated": list(env_updates.keys()),
"meta": {},
})
@router.post("/pricing/setup-webhook")
async def setup_stripe_webhook(request: Request, admin_id: str = Depends(require_admin)):
"""Auto-create the Stripe webhook endpoint and save its signing secret.
The webhook URL is derived from the request's host headers
(X-Forwarded-Proto + Host). In production, the URL must be HTTPS
and publicly reachable from Stripe.
Body (optional):
- ``url_override`` (str): explicit webhook URL if auto-detection
doesn't match your deployment (e.g. behind a custom reverse proxy)
- ``force`` (bool): if true, allow auto-setup in test mode even
when running in production (useful for staging)
"""
secret_key = os.getenv("STRIPE_SECRET_KEY", "").strip()
if not secret_key or not secret_key.startswith("sk_"):
return JSONResponse(status_code=400, content={
"error": "STRIPE_NOT_CONFIGURED",
"message": "STRIPE_SECRET_KEY manquante ou invalide. Configurez-la d'abord.",
})
mode = pricing_cfg.stripe_mode()
is_test_in_prod = pricing_cfg.is_test_mode_in_production()
# Read optional body
body: dict = {}
try:
body = await request.json()
except Exception:
body = {}
force = bool(body.get("force", False))
url_override = (body.get("url_override") or "").strip()
if is_test_in_prod and not force:
return JSONResponse(status_code=409, content={
"error": "TEST_MODE_IN_PRODUCTION",
"message": (
"Vous êtes en production (ENV=production) mais STRIPE_SECRET_KEY "
"est une clé de test (sk_test_…). Basculez sur sk_live_… ou "
"relancez avec {\"force\": true} pour créer un webhook test "
"explicitement (à des fins de staging)."
),
"mode": mode,
})
# Build the webhook URL from the request (host + forwarded proto)
if url_override:
webhook_url = url_override
else:
xfp = request.headers.get("x-forwarded-proto", "")
scheme = xfp.split(",")[0].strip() if xfp else (request.url.scheme or "https")
host = (
request.headers.get("x-forwarded-host")
or request.headers.get("host")
or request.url.netloc
)
webhook_url = f"{scheme}://{host}/api/v1/webhook/stripe"
# Basic sanity check
if not webhook_url.startswith(("http://", "https://")):
return JSONResponse(status_code=400, content={
"error": "INVALID_URL",
"message": f"URL webhook invalide : {webhook_url!r}",
})
if mode == "live" and webhook_url.startswith("http://"):
return JSONResponse(status_code=400, content={
"error": "INSECURE_URL_FOR_LIVE_MODE",
"message": (
"En mode Live, l'URL du webhook doit être HTTPS. "
f"URL actuelle : {webhook_url}. Utilisez url_override=https://…"
),
})
try:
import stripe as _stripe
_stripe.api_key = secret_key
# The events the backend actually handles (see payment_service.handle_webhook).
# If we send more events than the backend expects, Stripe will retry on
# signature mismatches and waste API quota. Keep this list aligned with
# the handler.
events = [
"checkout.session.completed",
"customer.subscription.created",
"customer.subscription.updated",
"customer.subscription.deleted",
"invoice.paid",
"invoice.payment_failed",
]
# Avoid duplicates: list existing endpoints and reuse one that already
# points to the same URL.
existing_list = _stripe.WebhookEndpoint.list(limit=20)
existing = next(
(e for e in existing_list.data if e.url == webhook_url),
None,
)
if existing:
# Update the existing endpoint to make sure its events are in sync
webhook = _stripe.WebhookEndpoint.update(
existing.id,
enabled_events=events,
)
created = False
else:
webhook = _stripe.WebhookEndpoint.create(
url=webhook_url,
enabled_events=events,
description="Office Translator — auto-created by admin panel",
)
created = True
# ``webhook.secret`` is only returned at creation time by Stripe. On
# update it is None, so we re-use the existing STRIPE_WEBHOOK_SECRET
# if the endpoint already existed.
new_secret = webhook.get("secret") or os.getenv("STRIPE_WEBHOOK_SECRET", "").strip()
env_updates = {}
if new_secret:
pricing_cfg.validate_stripe_webhook_secret(new_secret)
env_updates["STRIPE_WEBHOOK_SECRET"] = new_secret
if env_updates:
_update_env_file(env_updates)
pricing_cfg.apply_runtime_config_after_admin_write()
logger.info(
f"admin_stripe_webhook_setup by {admin_id}: url={webhook_url} "
f"created={created} mode={mode}"
)
return JSONResponse(status_code=200, content={
"data": {
"id": webhook.id,
"url": webhook.url,
"enabled_events": webhook.enabled_events,
"status": webhook.status,
"created": created,
"secret_saved": bool(new_secret),
},
"meta": {
"mode": mode,
"env": (os.getenv("ENV", os.getenv("ENVIRONMENT", "development")) or "development").lower(),
},
})
except ImportError:
return JSONResponse(status_code=500, content={
"error": "STRIPE_NOT_INSTALLED",
"message": "Le module stripe n'est pas installé sur le serveur.",
})
except Exception as e:
logger.exception("admin_stripe_webhook_setup_failed")
return JSONResponse(status_code=500, content={
"error": "STRIPE_API_ERROR",
"message": str(e),
})