""" PNA Assistant — Stripe Webhook Handler POST /webhook → Stripe sends checkout.session.completed events here → We generate a PNA-PRO-XXXXX code and email it to the customer Run standalone: uvicorn webhook:app --host 0.0.0.0 --port 8080 In Docker: separate service in docker-compose.yml """ import hashlib import hmac import json import logging import os import random import string from datetime import datetime, timezone from pathlib import Path import boto3 from botocore.exceptions import ClientError from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse # ─── Config ────────────────────────────────────────────────────────────────── STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET", "") AWS_REGION = os.getenv("AWS_REGION", "eu-west-2") SES_FROM_EMAIL = os.getenv("SES_FROM_EMAIL", "lincoln@clinyqai.com") # Where we persist issued codes (append-only JSONL) CODES_LOG = Path("/app/codes.jsonl") logging.basicConfig(level=logging.INFO) log = logging.getLogger("pna-webhook") app = FastAPI(title="PNA Assistant Webhook", docs_url=None, redoc_url=None) # ─── Helpers ───────────────────────────────────────────────────────────────── def _generate_code(tier: str = "PRO") -> str: """Generate a unique activation code e.g. PNA-PRO-A7K2M9""" suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=6)) return f"PNA-{tier}-{suffix}" def _verify_stripe_signature(payload: bytes, sig_header: str, secret: str) -> bool: """ Validate Stripe-Signature header. Stripe uses HMAC-SHA256 with the raw request body. """ if not secret: log.warning("STRIPE_WEBHOOK_SECRET not set — skipping signature verification") return True # dev mode only; set the secret in production try: parts = {k: v for part in sig_header.split(",") for k, v in [part.split("=", 1)]} timestamp = parts.get("t", "") v1_sig = parts.get("v1", "") signed_payload = f"{timestamp}.".encode() + payload expected = hmac.new( secret.encode(), signed_payload, hashlib.sha256, ).hexdigest() return hmac.compare_digest(expected, v1_sig) except Exception as exc: log.error("Signature verification error: %s", exc) return False def _log_code(email: str, code: str, plan: str, session_id: str) -> None: """Append issued code to local JSONL log for admin audit trail.""" CODES_LOG.parent.mkdir(parents=True, exist_ok=True) record = { "ts": datetime.now(timezone.utc).isoformat(), "email": email, "code": code, "plan": plan, "session_id": session_id, } with CODES_LOG.open("a") as f: f.write(json.dumps(record) + "\n") log.info("Code issued: %s → %s (%s)", email, code, plan) def _send_activation_email(email: str, code: str, plan: str) -> bool: """ Send activation code to customer via Amazon SES. Returns True if sent, False on error. """ label = "Pro" if plan == "PRO" else "Institution" price = "£9.99/month" if plan == "PRO" else "£99/month" body_html = f"""

👨🏾‍⚕️ PNA Assistant

Professional Nurse Advocate AI — {label} Plan

Thank you for subscribing to PNA Assistant {label} ({price}).

Your activation code is:

{code}

How to activate:

  1. Go to pna.nursingcitizendevelopment.com
  2. Open the sidebar and find Subscription
  3. Paste your code in the Activation code box
  4. Click Enter — your {label} features will unlock immediately

Your {label} plan includes:

If you have any questions, reply to this email or contact lincoln@clinyqai.com.

This tool is for educational purposes only and does not replace clinical supervision or your employer's formal governance processes.

Contains public sector information licensed under the Open Government Licence v3.0 — NHS England.

""" body_text = ( f"Thank you for subscribing to PNA Assistant {label} ({price}).\n\n" f"Your activation code is: {code}\n\n" "To activate:\n" "1. Go to pna.nursingcitizendevelopment.com\n" "2. Open the sidebar → Subscription\n" "3. Paste your code in the Activation code box\n\n" "Questions? Email lincoln@clinyqai.com" ) try: ses = boto3.client("ses", region_name=AWS_REGION) ses.send_email( Source = f"PNA Assistant <{SES_FROM_EMAIL}>", Destination = {"ToAddresses": [email]}, Message = { "Subject": {"Data": f"Your PNA Assistant {label} activation code 🎉"}, "Body": { "Text": {"Data": body_text}, "Html": {"Data": body_html}, }, }, ) log.info("Activation email sent to %s", email) return True except ClientError as exc: log.error("SES send failed: %s", exc.response["Error"]["Message"]) return False except Exception as exc: log.error("Email error: %s", exc) return False # ─── Routes ────────────────────────────────────────────────────────────────── @app.get("/health") async def health(): return {"status": "ok"} @app.post("/webhook") async def stripe_webhook(request: Request): payload = await request.body() sig_header = request.headers.get("Stripe-Signature", "") # 1. Verify Stripe signature if not _verify_stripe_signature(payload, sig_header, STRIPE_WEBHOOK_SECRET): log.warning("Invalid Stripe signature — rejected") raise HTTPException(status_code=400, detail="Invalid signature") # 2. Parse event try: event = json.loads(payload) except json.JSONDecodeError: raise HTTPException(status_code=400, detail="Invalid JSON") event_type = event.get("type", "") log.info("Received Stripe event: %s", event_type) # 3. Handle checkout completion if event_type == "checkout.session.completed": session = event["data"]["object"] email = session.get("customer_details", {}).get("email") or session.get("customer_email") amount = session.get("amount_total", 0) # pence/cents session_id = session.get("id", "unknown") if not email: log.warning("No email in session %s — cannot send code", session_id) return JSONResponse({"received": True, "warning": "no_email"}) # Determine tier from amount (999 = £9.99, 9900 = £99.00) tier = "INST" if amount >= 9900 else "PRO" code = _generate_code(tier) # 4. Log the code (always) _log_code(email, code, tier, session_id) # 5. Email the customer sent = _send_activation_email(email, code, tier) if not sent: log.warning( "SES send failed for %s — code %s logged to %s", email, code, CODES_LOG, ) return JSONResponse({"received": True})