""" app/billing/routes.py --------------------- MT-8 Stripe billing routes. Routes: GET /billing/subscribe → create Stripe Checkout session → redirect GET /billing/portal → create Stripe Customer Portal session → redirect POST /billing/webhook → verify Stripe signature → handle event GET /billing/suspended → suspension landing page (no auth required) GET /billing/success → post-checkout success landing GET /billing/cancel → post-checkout cancel (redirects to plan page) All routes are CSRF-exempt (billing.bp is exempted in app/__init__.py). The webhook route is the primary reason — Stripe cannot send CSRF tokens. """ import logging from flask import ( current_app, redirect, url_for, render_template, request, flash, g, abort, ) from flask_login import login_required, current_user from app.billing import bp from app.utils.decorators import admin_required logger = logging.getLogger(__name__) def _billing_enabled(): return current_app.config.get('BILLING_ENABLED', False) def _mt_enabled(): return current_app.config.get('MULTI_TENANT_ENABLED', False) # ── Subscribe: initiate Stripe Checkout ─────────────────────────────────────── @bp.route('/subscribe') @login_required @admin_required def subscribe(): if not _billing_enabled(): flash('Billing is not enabled on this deployment.', 'info') return redirect(url_for('tenant_settings.plan')) if not _mt_enabled(): flash('Billing is only available in multi-tenant mode.', 'info') return redirect(url_for('tenant_settings.plan')) tenant = getattr(g, 'tenant', None) if tenant is None: abort(400) from control.base import control_session from control.models import Tenant as ControlTenant, Plan from app.billing.stripe_client import create_stripe_customer, create_checkout_session # Allow tenant to pick a plan via ?plan= when on Free. requested_plan_code = request.args.get('plan') with control_session() as s: t = s.get(ControlTenant, tenant.id) if t is None: flash('Tenant not found.', 'danger') return redirect(url_for('tenant_settings.plan')) plan = t.plan # If a specific plan was requested (e.g. from the plan picker), use it. if requested_plan_code: chosen = s.query(Plan).filter_by(code=requested_plan_code).first() if chosen and chosen.stripe_price_id: plan = chosen # If still on Free (no price ID), show the plan picker instead. if plan is None or not plan.stripe_price_id: paid_plans = (s.query(Plan) .filter(Plan.stripe_price_id.isnot(None)) .order_by(Plan.price_cents) .all()) return render_template('billing/plan_picker.html', plans=paid_plans) # Get or create Stripe customer. Write back immediately within the # control_session so the customer_id is persisted before the redirect. if not t.stripe_customer_id: try: customer = create_stripe_customer( email=current_user.email, name=current_user.display_name, tenant_id=tenant.id, slug=tenant.slug, ) t.stripe_customer_id = customer['id'] t.billing_email = current_user.email except Exception as exc: logger.error('BILLING | create_customer_failed | tenant=%s err=%s', tenant.slug, exc) flash('Could not connect to Stripe. Please try again or contact support.', 'danger') return redirect(url_for('tenant_settings.plan')) stripe_customer_id = t.stripe_customer_id price_id = plan.stripe_price_id # Create Checkout Session AFTER the control_session commits customer_id. base_url = (current_app.config.get('APP_BASE_URL') or request.host_url).rstrip('/') try: session = create_checkout_session( customer_id=stripe_customer_id, price_id=price_id, tenant_id=tenant.id, success_url=f'{base_url}/billing/success?session_id={{CHECKOUT_SESSION_ID}}', cancel_url=f'{base_url}/billing/cancel', ) except Exception as exc: logger.error('BILLING | create_checkout_failed | tenant=%s err=%s', tenant.slug, exc) flash('Could not create a Stripe Checkout session. Please try again.', 'danger') return redirect(url_for('tenant_settings.plan')) return redirect(session['url'], 303) # ── Portal: Stripe Customer Portal for managing subscriptions ───────────────── @bp.route('/portal') @login_required @admin_required def portal(): if not _billing_enabled(): flash('Billing is not enabled on this deployment.', 'info') return redirect(url_for('tenant_settings.plan')) tenant = getattr(g, 'tenant', None) if tenant is None: abort(400) from control.base import control_session from control.models import Tenant as ControlTenant from app.billing.stripe_client import create_portal_session with control_session() as s: t = s.get(ControlTenant, tenant.id) if t is None or not t.stripe_customer_id: flash('No active subscription found. Please subscribe first.', 'warning') return redirect(url_for('tenant_settings.plan')) stripe_customer_id = t.stripe_customer_id base_url = (current_app.config.get('APP_BASE_URL') or request.host_url).rstrip('/') return_url = f'{base_url}/settings/plan' try: session = create_portal_session( customer_id=stripe_customer_id, return_url=return_url, ) except Exception as exc: logger.error('BILLING | create_portal_failed | tenant=%s err=%s', tenant.slug, exc) flash('Could not open the billing portal. Please try again.', 'danger') return redirect(url_for('tenant_settings.plan')) return redirect(session['url'], 303) # ── Webhook: receives all Stripe events ─────────────────────────────────────── @bp.route('/webhook', methods=['POST']) def webhook(): """Stripe webhook endpoint. Verifies signature then dispatches to webhooks.py.""" webhook_secret = current_app.config.get('STRIPE_WEBHOOK_SECRET') if not webhook_secret: logger.error('BILLING WEBHOOK | STRIPE_WEBHOOK_SECRET not configured') return {'error': 'Webhook secret not configured'}, 500 payload = request.get_data() sig_header = request.headers.get('Stripe-Signature', '') try: import stripe stripe.api_key = current_app.config.get('STRIPE_SECRET_KEY', '') event = stripe.Webhook.construct_event(payload, sig_header, webhook_secret) except ValueError: logger.warning('BILLING WEBHOOK | invalid_payload') return {'error': 'Invalid payload'}, 400 except stripe.error.SignatureVerificationError: logger.warning('BILLING WEBHOOK | invalid_signature') return {'error': 'Invalid signature'}, 400 from app.billing.webhooks import handle_event handle_event(event) return {'ok': True}, 200 # ── Suspended: shown when subscription is cancelled ─────────────────────────── @bp.route('/suspended') def suspended(): """Billing-suspended landing page. No authentication required.""" return render_template('billing/suspended.html') # ── Post-checkout success / cancel ──────────────────────────────────────────── @bp.route('/success') @login_required def success(): flash('Your subscription is now active. Welcome aboard!', 'success') return render_template('billing/success.html') @bp.route('/cancel') @login_required def cancel(): flash('Checkout was cancelled. Your subscription has not changed.', 'info') return redirect(url_for('tenant_settings.plan'))