Build SecurePay — a Stripe webhook processor that validates signatures, scopes events to tenants, gates feature access via feature flags, and emits structured audit trails. Python + FastAPI.

SaaS Billing Webhooks with Tenant Scoping

What you'll build: SecurePay — a webhook processor for a SaaS billing system. When Stripe fires invoice.paid, subscription.upgraded, or payment.failed, SecurePay validates the signature, resolves the tenant, gates access to premium features via feature flags, and writes a tamper-evident audit trail. CoreSDK handles auth, flags, and audit in one place.

The Story

Stripe fires webhooks. You need to:

Validate the Stripe-Signature header (HMAC-SHA256) — reject replays and forgeries
Map customer_id → tenant_id (one customer per tenant)
When invoice.paid: flip pro_features flag on for that tenant
When payment.failed: lock the tenant behind a paywall — pro_features off
When subscription.upgraded: enable advanced_analytics flag
Audit every event with the actor, tenant, and outcome — RFC 9457 errors on failure

Without CoreSDK: Stripe signature validation, tenant lookup, flag mutation, and audit logging are all hand-rolled across different files. With CoreSDK: flags and audit are one import.

Architecture

Stripe ──POST /webhooks/stripe──► FastAPI
                                      │
                          validate Stripe-Signature
                                      │
                          resolve tenant_id from customer_id
                                      │
                    ┌─────────────────┼──────────────────┐
                    │                 │                   │
              invoice.paid    payment.failed   subscription.upgraded
                    │                 │                   │
             enable flag        disable flag         enable flag
          pro_features=on      pro_features=off   advanced_analytics=on
                    │                 │                   │
                    └─────────────────┼──────────────────┘
                                      │
                              CoreSDK Audit Trail
                                      │
                               Sidecar :50051

Prerequisites

pip install "coresdk[fastapi]" uvicorn stripe

Quickstart

git clone https://github.com/coresdk-dev/examples
cd examples/python
CORESDK_USE_MOCK=true python 08_billing_webhooks.py

Output:

SecurePay Webhook Processor — Test Suite
==========================================

▸ Webhook Validation
  [PASS]  Valid Stripe signature accepted
  [PASS]  Invalid signature → 400 (replay/forgery blocked)
  [PASS]  Missing signature header → 400

▸ invoice.paid — Feature Flag Activation
  [PASS]  invoice.paid enables pro_features for acme-corp
  [PASS]  pro_features flag is ON after payment
  [PASS]  Audit event recorded: billing.invoice.paid

▸ payment.failed — Paywall Enforcement
  [PASS]  payment.failed disables pro_features for globex
  [PASS]  pro_features flag is OFF after failure
  [PASS]  Audit event recorded: billing.payment.failed

▸ subscription.upgraded — Tier Promotion
  [PASS]  subscription.upgraded enables advanced_analytics
  [PASS]  /api/analytics returns 200 after upgrade
  [PASS]  /api/analytics returns 403 before upgrade

12/12 passed

Code Walkthrough

Step 1 — Stripe signature validation

import hashlib
import hmac
import time

STRIPE_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_test_secret")
REPLAY_TOLERANCE_SEC = 300  # reject events older than 5 minutes

def verify_stripe_signature(payload: bytes, sig_header: str) -> bool:
    """Validate Stripe-Signature header — blocks replays and forgeries."""
    try:
        parts = dict(p.split("=", 1) for p in sig_header.split(","))
        timestamp = int(parts["t"])
        v1_sig = parts["v1"]
    except (KeyError, ValueError):
        return False

    # Reject replays
    if abs(time.time() - timestamp) > REPLAY_TOLERANCE_SEC:
        return False

    signed_payload = f"{timestamp}.".encode() + payload
    expected = hmac.new(
        STRIPE_SECRET.encode(), signed_payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, v1_sig)

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig = request.headers.get("Stripe-Signature", "")

    if not verify_stripe_signature(payload, sig):
        raise HTTPException(status_code=400, detail={
            "type": "https://securepay.example/errors/invalid-signature",
            "title": "Bad Request",
            "status": 400,
            "detail": "Stripe signature validation failed.",
        })
    ...

Step 2 — Tenant resolution

# customer_id → tenant_id mapping (in production: database lookup)
CUSTOMER_TENANT_MAP = {
    "cus_acme_001": "acme-corp",
    "cus_globex_001": "globex",
    "cus_initech_001": "initech",
}

event = json.loads(payload)
customer_id = event.get("data", {}).get("object", {}).get("customer")
tenant_id = CUSTOMER_TENANT_MAP.get(customer_id)

if not tenant_id:
    raise HTTPException(status_code=422, detail={
        "type": "https://securepay.example/errors/unknown-customer",
        "title": "Unprocessable Entity",
        "status": 422,
        "detail": f"No tenant mapped for customer '{customer_id}'.",
    })

Step 3 — Feature flags via CoreSDK

from coresdk import SDK

sdk = SDK.from_env()

# In-memory flag store (replace with CoreSDK control plane)
_flags: dict[str, dict[str, bool]] = {}

def set_flag(tenant_id: str, flag: str, value: bool):
    _flags.setdefault(tenant_id, {})[flag] = value

def get_flag(tenant_id: str, flag: str) -> bool:
    return _flags.get(tenant_id, {}).get(flag, False)

# invoice.paid handler
if event_type == "invoice.paid":
    set_flag(tenant_id, "pro_features", True)
    set_flag(tenant_id, "paywall_active", False)

# payment.failed handler
elif event_type == "payment.failed":
    set_flag(tenant_id, "pro_features", False)
    set_flag(tenant_id, "paywall_active", True)

# subscription.upgraded handler
elif event_type == "customer.subscription.updated":
    new_plan = event["data"]["object"].get("plan", {}).get("nickname", "")
    if "enterprise" in new_plan.lower() or "pro" in new_plan.lower():
        set_flag(tenant_id, "advanced_analytics", True)

Step 4 — Gating API routes with flags

@app.get("/api/analytics")
async def analytics(request: Request):
    claims = get_claims(request)
    tenant_id = claims.get("tenant_id", "")

    if not get_flag(tenant_id, "advanced_analytics"):
        raise HTTPException(status_code=403, detail={
            "type": "https://securepay.example/errors/feature-locked",
            "title": "Feature Unavailable",
            "status": 403,
            "detail": "Advanced analytics requires an Enterprise subscription.",
            "upgrade_url": "https://securepay.example/pricing",
        })

    return {"tenant": tenant_id, "data": generate_analytics(tenant_id)}

@app.get("/api/dashboard")
async def dashboard(request: Request):
    claims = get_claims(request)
    tenant_id = claims.get("tenant_id", "")

    if get_flag(tenant_id, "paywall_active"):
        raise HTTPException(status_code=402, detail={
            "type": "https://securepay.example/errors/payment-required",
            "title": "Payment Required",
            "status": 402,
            "detail": "Your subscription has a payment failure. Update billing to continue.",
            "billing_url": "https://securepay.example/billing",
        })

    return {"tenant": tenant_id, "status": "active"}

Step 5 — Audit trail

audit_log: list[dict] = []

def audit(tenant_id: str, event_type: str, outcome: str, metadata: dict):
    audit_log.append({
        "timestamp": datetime.utcnow().isoformat(),
        "tenant_id": tenant_id,
        "event": event_type,
        "outcome": outcome,
        "actor": "stripe-webhook",
        **metadata,
    })

# Called after each flag mutation
audit(tenant_id, "billing.invoice.paid", "success",
      {"invoice_id": event["data"]["object"]["id"], "flag": "pro_features=on"})

Testing Locally with Stripe CLI

# Install Stripe CLI
brew install stripe/stripe-cli/stripe

# Forward live Stripe events to local server
stripe listen --forward-to localhost:8000/webhooks/stripe

# Trigger test events
stripe trigger invoice.paid
stripe trigger payment_intent.payment_failed
stripe trigger customer.subscription.updated

Environment Variables

Variable	Default	Description
`CORESDK_SIDECAR_ADDR`	`localhost:50051`	Sidecar gRPC address
`CORESDK_FAIL_MODE`	`open`	Fail behavior on sidecar error
`STRIPE_WEBHOOK_SECRET`	—	`whsec_...` from Stripe dashboard
`CORESDK_USE_MOCK`	`false`	`true` for local dev without sidecar

Full Source

→ examples/python/08_billing_webhooks.py

CORESDK_USE_MOCK=true python 08_billing_webhooks.py

SaaS Billing Webhooks with Tenant Scoping

On this page