README.md
# paywall-worker
An optional, operator-adjustable paywall for any static single-page app, written in
pure Python for Cloudflare Python Workers.
- Sign in: Google, Facebook, GitHub, Microsoft, Discord, GitLab, any OIDC issuer, email links.
- Pay: Stripe (cards, Apple Pay, Google Pay, Wero / iDEAL, SEPA), PayPal, UPI via Razorpay,
Pix via Mercado Pago, voucher codes.
- Operate: switch between `off`, `login` and `paid`, meter free page loads, edit plans and
prices per currency, comp domains, grant access and mint codes from `/_pw/admin`.
Changes apply within about 10 seconds, with no redeploy.
## Layout
```text
wrangler.jsonc Worker, assets, D1 and KV bindings
pyproject.toml no runtime dependencies; dev: workers-py, pytest
schema.sql D1 tables
.dev.vars.example every secret, with placeholders
spa/dist/ your built SPA (a placeholder index.html ships here)
src/entry.py Workers entry point (the only file touching the runtime)
src/paywall/ app, gate, config, session, store, web, crypto, net, auth/, pay/
tests/ CPython tests with fake KV, D1 and HTTP
```
## Install and test
```bash
uv sync
uv run pytest # 15 tests
```
## Build
No Worker build step. Put your SPA's build output in `spa/dist`. Files that must load
before sign-in go under `spa/dist/public/` or into `public_paths`.
## Local development
```bash
cp .dev.vars.example .dev.vars
npx wrangler d1 execute paywall --local --file=schema.sql
uv run pywrangler dev # http://localhost:8787
```
Register `http://localhost:8787/_pw/callback/{provider}` in each OAuth console. For Stripe
webhooks: `stripe listen --forward-to localhost:8787/_pw/webhook/stripe`. For the other
providers, use a tunnel such as `cloudflared tunnel --url http://localhost:8787`.
## Deploy
```bash
npx wrangler d1 create paywall # paste database_id into wrangler.jsonc
npx wrangler kv namespace create KV # paste id into wrangler.jsonc
npx wrangler d1 execute paywall --remote --file=schema.sql
npx wrangler secret put ADMIN_TOKEN # and ADMIN_EMAILS, provider secrets...
uv run pywrangler deploy
```
Then attach a custom domain, register `https://HOST/_pw/callback/{provider}` with each sign-in
provider and `https://HOST/_pw/webhook/{stripe,paypal,razorpay}` with each payment provider
(Mercado Pago gets its URL per payment), sign in with an `ADMIN_EMAILS` address and open
`/_pw/admin`.
## Operating
- Pause the paywall: `{"mode": "off"}`. Members-only but free: `{"mode": "login"}`.
- Metering: `{"metered_free_loads": 3}`.
- Comp a person: `POST /_pw/admin/grant {"email": ..., "days": 0}` (0 = lifetime).
- Comp an institution: add its domain to `free_email_domains`.
Full tutorial and reference: see the accompanying document.
## License
MIT. Provider names are trademarks of their owners; this project is not affiliated with them.
wrangler.jsonc
{
"$schema": "./node_modules/wrangler/config-schema.json",
"name": "paywall",
"main": "src/entry.py",
"compatibility_date": "2026-11-04",
"compatibility_flags": ["python_workers"],
"assets": {
"directory": "./spa/dist",
"binding": "ASSETS",
"not_found_handling": "single-page-application",
// Every request reaches the Worker first so it can gate the SPA.
"run_worker_first": true
},
"d1_databases": [
{ "binding": "DB", "database_name": "paywall", "database_id": "<from wrangler d1 create>" }
],
"kv_namespaces": [
{ "binding": "KV", "id": "<from wrangler kv namespace create>" }
],
"vars": {
"PAYPAL_ENV": "sandbox",
"FACEBOOK_GRAPH_VERSION": "v23.0",
"EMAIL_API_URL": "https://api.resend.com/emails"
},
"observability": { "enabled": true }
}
pyproject.toml
[project]
name = "paywall-worker"
version = "1.0.0"
description = "An adjustable paywall for static single-page apps on Cloudflare Python Workers"
requires-python = ">=3.12"
dependencies = [] # standard library only; nothing to vendor into the Worker
[dependency-groups]
dev = ["workers-py", "workers-runtime-sdk", "pytest"]
[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]
schema.sql
-- D1 schema. Apply with: npx wrangler d1 execute paywall --remote --file=schema.sql
-- Columns are NOT NULL with '' / 0 defaults because Pyodide turns None
-- into JavaScript undefined, which D1 refuses to bind.
CREATE TABLE IF NOT EXISTS config (
id INTEGER PRIMARY KEY CHECK (id = 1),
json TEXT NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
email TEXT NOT NULL DEFAULT '',
email_verified INTEGER NOT NULL DEFAULT 0,
name TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS users_email ON users (email) WHERE email_verified = 1;
CREATE TABLE IF NOT EXISTS identities (
provider TEXT NOT NULL, -- google, github, email, ...
subject TEXT NOT NULL, -- the provider's stable user id
user_id TEXT NOT NULL REFERENCES users (id),
created_at INTEGER NOT NULL,
PRIMARY KEY (provider, subject)
);
CREATE TABLE IF NOT EXISTS entitlements (
user_id TEXT PRIMARY KEY REFERENCES users (id),
expires_at INTEGER NOT NULL, -- unix seconds; 253402300799 = lifetime
source TEXT NOT NULL, -- e.g. stripe:cs_..., code:ABCD-..., admin
updated_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS payments (
ref TEXT PRIMARY KEY, -- ours; sent to the provider as reference
provider TEXT NOT NULL,
external_id TEXT NOT NULL DEFAULT '', -- checkout session / order / link id
txn_id TEXT NOT NULL DEFAULT '', -- capture / payment id, for refunds
user_id TEXT NOT NULL REFERENCES users (id),
plan_id TEXT NOT NULL,
days INTEGER NOT NULL, -- copied from the plan at checkout time
amount INTEGER NOT NULL, -- minor units
currency TEXT NOT NULL,
status TEXT NOT NULL, -- created, pending, paid, failed, mismatch
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS payments_user ON payments (user_id, created_at);
CREATE INDEX IF NOT EXISTS payments_recent ON payments (created_at);
CREATE TABLE IF NOT EXISTS codes (
code TEXT PRIMARY KEY,
plan_id TEXT NOT NULL,
days INTEGER NOT NULL, -- 0 = lifetime
uses_left INTEGER NOT NULL,
expires_at INTEGER NOT NULL DEFAULT 0 -- 0 = never
);
.dev.vars.example
# Copy to .dev.vars for `uv run pywrangler dev`. Never commit .dev.vars.
# In production, set each with: npx wrangler secret put NAME
# A provider stays hidden until its secrets are filled in.
ADMIN_TOKEN=change-me-to-a-long-random-string
ADMIN_EMAILS=you@example.com
# PUBLIC_ORIGIN=https://app.example.com
GOOGLE_CLIENT_ID=
GOOGLE_CLIENT_SECRET=
FACEBOOK_CLIENT_ID=
FACEBOOK_CLIENT_SECRET=
GITHUB_CLIENT_ID=
GITHUB_CLIENT_SECRET=
MICROSOFT_CLIENT_ID=
MICROSOFT_CLIENT_SECRET=
DISCORD_CLIENT_ID=
DISCORD_CLIENT_SECRET=
GITLAB_CLIENT_ID=
GITLAB_CLIENT_SECRET=
# GITLAB_BASE_URL=https://gitlab.example.org
OIDC_ISSUER=
OIDC_CLIENT_ID=
OIDC_CLIENT_SECRET=
OIDC_LABEL=Company SSO
EMAIL_API_KEY=
EMAIL_FROM=Sign-in <login@example.com>
STRIPE_SECRET_KEY=sk_test_
STRIPE_WEBHOOK_SECRET=whsec_
PAYPAL_CLIENT_ID=
PAYPAL_CLIENT_SECRET=
PAYPAL_WEBHOOK_ID=
RAZORPAY_KEY_ID=rzp_test_
RAZORPAY_KEY_SECRET=
RAZORPAY_WEBHOOK_SECRET=
MERCADOPAGO_ACCESS_TOKEN=TEST-
MERCADOPAGO_WEBHOOK_SECRET=
src/entry.py
"""Workers entry point: translate the runtime's Request/Response to ours."""
from workers import Response, WorkerEntrypoint
from paywall.app import App
from paywall.web import Req, ServeAsset
async def to_req(request) -> Req:
js = getattr(request, "js_object", request) # SDK wrapper or raw JS
headers = {str(k).lower(): str(v) for k, v in js.headers.entries()}
body = b""
if js.method not in ("GET", "HEAD"):
# Read a clone so the original can still be forwarded to ASSETS.
body = (await js.clone().arrayBuffer()).to_bytes()
return Req(str(js.method), str(js.url), headers, body)
class Default(WorkerEntrypoint):
async def fetch(self, request):
result = await App(self.env).handle(await to_req(request))
if isinstance(result, ServeAsset):
js_request = getattr(request, "js_object", request)
asset = await self.env.ASSETS.fetch(js_request)
if not result.headers:
return asset
from js import Response as JsResponse
out = JsResponse.new(asset.body, asset) # mutable copy
for name, value in result.headers.items():
out.headers.set(name, value)
return out
return Response(result.body, status=result.status,
headers=result.headers)
src/paywall/app.py
"""Request router: the gate for SPA assets plus everything under /_pw/."""
import json
import logging
import secrets
import time
from urllib.parse import quote
from . import crypto, gate, pay
from .auth import magic, oauth
from .config import display_price
from .net import workers_fetch
from .session import Session
from .store import LIFETIME, Store
from .web import (Req, Resp, ServeAsset, esc, html, json_resp, redirect,
safe_next)
log = logging.getLogger("paywall")
_DISCOVERY = {} # OIDC discovery documents, cached per isolate
class App:
def __init__(self, env, fetcher=workers_fetch, now=time.time):
self.env, self.http, self.now = env, fetcher, now
self.store = Store(env, now)
self.discovery_cache = _DISCOVERY
self.origin = ""
self.cfg = {}
# ---- helpers ---------------------------------------------------------
def env_str(self, name: str) -> str:
value = getattr(self.env, name, "")
return str(value) if value else ""
@staticmethod
def dumps(obj) -> str:
return json.dumps(obj, separators=(",", ":"))
def same_origin(self, req: Req) -> bool:
origin = req.header("origin")
if origin:
return origin in (self.origin, req.origin)
return req.header("sec-fetch-site") == "same-origin"
def admin_via_token(self, req: Req) -> bool:
expected = self.env_str("ADMIN_TOKEN")
given = req.header("authorization").removeprefix("Bearer ").strip()
return bool(expected and given and crypto.safe_eq(given, expected))
def is_admin(self, req: Req, sess: Session) -> bool:
if self.admin_via_token(req):
return True
user = sess.user()
extra = [e.strip().lower() for e in self.env_str("ADMIN_EMAILS").split(",")]
return bool(user and user["email_verified"] and
user["email"].lower() in self.cfg["admin_emails"] + extra)
def currency_for(self, req: Req) -> str:
chosen = req.query.get("cur", "").upper()
offered = {c for p in self.cfg["plans"] for c in p["prices"]}
if chosen in offered:
return chosen
return self.cfg["country_currency"].get(req.country,
self.cfg["default_currency"])
def payment_options(self, currency: str) -> list:
out = []
for name in self.cfg["payment_providers"]:
module = pay.PROVIDERS[name]
allowed = self.cfg["provider_currencies"].get(name, [])
if name != "codes" and module.configured(self.env) and \
("*" in allowed or currency in allowed):
out.append(name)
return out
def auth_options(self) -> list:
out = []
for name in self.cfg["auth_providers"]:
if name == "magic":
if magic.configured(self.env):
out.append(name)
elif oauth.configured(self.env, name):
out.append(name)
return out
# ---- pages -----------------------------------------------------------
def page(self, title: str, body: str, status: int = 200) -> Resp:
site = esc(self.cfg.get("site_name", "Paywall"))
return html(f"""<!doctype html><html lang="en"><head>
<meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>{esc(title)} · {site}</title><style>
:root{{color-scheme:light dark;--fg:#1d1d1f;--bg:#fafaf8;--card:#fff;--line:#d9d9d4;--accent:#2457c5}}
@media(prefers-color-scheme:dark){{:root{{--fg:#ececec;--bg:#161616;--card:#202020;--line:#3a3a3a;--accent:#7ea2ff}}}}
body{{font:16px/1.5 system-ui,sans-serif;color:var(--fg);background:var(--bg);margin:0;padding:24px 16px}}
main{{max-width:560px;margin:0 auto}}h1{{font-size:1.5rem;margin:0 0 4px}}
.card{{background:var(--card);border:1px solid var(--line);border-radius:10px;padding:16px;margin:16px 0}}
button,.btn{{display:inline-block;font:inherit;padding:8px 14px;margin:4px 4px 0 0;border-radius:8px;
border:1px solid var(--line);background:var(--card);color:var(--fg);cursor:pointer;text-decoration:none}}
button.primary,.btn.primary{{background:var(--accent);border-color:var(--accent);color:#fff}}
input,textarea,select{{font:inherit;padding:8px;border:1px solid var(--line);border-radius:8px;
background:var(--bg);color:var(--fg);box-sizing:border-box;width:100%;margin:4px 0}}
textarea{{font-family:ui-monospace,monospace;font-size:13px}}small,.muted{{opacity:.7}}
table{{border-collapse:collapse;width:100%;font-size:13px}}td,th{{border-bottom:1px solid var(--line);padding:4px;text-align:left}}
</style></head><body><main><h1>{esc(title)}</h1>{body}</main></body></html>""",
status)
def error_page(self, message: str, status: int = 400) -> Resp:
return self.page("Something went wrong", f"""<div class="card">
<p>{esc(message)}</p><a class="btn" href="/_pw/wall">Back</a></div>""",
status)
def poll_script(self, ref: str) -> str:
return f"""<p id="st" class="muted">Waiting for confirmation…</p><script>
(async()=>{{for(let i=0;i<200;i++){{await new Promise(r=>setTimeout(r,3000));
const r=await fetch('/_pw/status/{esc(ref)}');const j=await r.json();
if(j.status==='paid'){{location.href='/';return}}
if(['failed','mismatch'].includes(j.status)){{document.getElementById('st').textContent=
'The payment did not go through. Nothing was charged for access.';return}}}}}})();</script>"""
def pending_page(self, ref: str) -> Resp:
return self.page("Confirming payment", f"""<div class="card">
<p>Thanks. We are confirming your payment with the provider; this
usually takes a few seconds.</p>{self.poll_script(ref)}</div>""")
async def wall_page(self, req: Req, sess: Session) -> Resp:
nxt = safe_next(req.query.get("next"))
user = sess.user()
parts = [f"<p>{esc(self.cfg['wall_message'])}</p>"]
if not user:
buttons = "".join(
f'<a class="btn" href="/_pw/login/{n}?next={quote(nxt)}">'
f'Continue with {esc(oauth.PROVIDERS[n]["label"])}</a>'
for n in self.auth_options() if n != "magic")
if "magic" in self.auth_options():
buttons += f"""<form method="post" action="/_pw/magic">
<input type="hidden" name="next" value="{esc(nxt)}">
<input type="email" name="email" required placeholder="you@example.com">
<button>Email me a sign-in link</button></form>"""
parts.append(f'<div class="card"><h2>Sign in</h2>{buttons}</div>')
else:
until = await sess.entitled_until(force=True)
if self.cfg["mode"] != "paid" or until > self.now() or \
gate.is_comped(self.cfg, user):
parts.append('<p>You have access. '
f'<a class="btn primary" href="{esc(nxt)}">Continue</a></p>')
else:
parts.append(self.plans_html(req))
parts.append(f"""<p class="muted">Signed in as
{esc(user['email'] or user['name'])}.</p>
<form method="post" action="/_pw/logout"><button>Sign out</button></form>""")
return self.page(self.cfg["site_name"], "".join(parts))
def plans_html(self, req: Req) -> str:
cur = self.currency_for(req)
offered = sorted({c for p in self.cfg["plans"] for c in p["prices"]})
switch = " ".join(
f'<a href="?cur={c}">{"<b>" + c + "</b>" if c == cur else c}</a>'
for c in offered)
providers = self.payment_options(cur)
cards = []
for plan in self.cfg["plans"]:
amount = plan["prices"].get(cur)
if not amount:
continue
length = "forever" if plan["days"] == 0 else f"{plan['days']} days"
btns = "".join(f"""<form method="post" action="/_pw/checkout/{p}"
style="display:inline"><input type="hidden" name="plan"
value="{esc(plan['id'])}"><input type="hidden" name="currency"
value="{cur}"><button class="primary">{esc(pay.LABELS[p])}</button>
</form>""" for p in providers) or \
'<p class="muted">No payment method is set up for this currency.</p>'
cards.append(f"""<div class="card"><h2>{esc(plan['name'])}</h2>
<p>{display_price(amount, cur)} · {length}</p>{btns}</div>""")
codes = ""
if "codes" in self.cfg["payment_providers"]:
codes = """<div class="card"><form method="post" action="/_pw/redeem">
<input name="code" placeholder="Voucher code" required>
<button>Redeem</button></form></div>"""
return (f'<p class="muted">Currency: {switch}</p>' + "".join(cards)
+ codes)
# ---- entry point -----------------------------------------------------
async def handle(self, req: Req):
self.origin = self.env_str("PUBLIC_ORIGIN").rstrip("/") or req.origin
self.cfg = await self.store.get_config()
sess = await Session.load(self, req)
try:
if req.path.startswith("/_pw/"):
resp = await self.route(req, sess)
else:
resp = await self.guard(req, sess)
except PermissionError as exc:
resp = self.error_page(str(exc), 403)
except (RuntimeError, ValueError) as exc:
log.warning("request failed: %s", exc)
resp = self.error_page(str(exc), 400)
await sess.commit(resp)
return resp
async def guard(self, req: Req, sess: Session):
user = sess.user()
until = await sess.entitled_until() if user else 0
decision = gate.decide(self.cfg, req.path, user=user,
entitled=until > self.now(),
meter_used=int(sess.data.get("meter", 0)),
is_document=req.is_document)
if decision == gate.METER:
sess.ensure()
sess.data["meter"] = int(sess.data.get("meter", 0)) + 1
sess.dirty = True
if decision in (gate.ALLOW, gate.METER):
if self.cfg["mode"] == "off" or gate.is_public(self.cfg, req.path):
return ServeAsset()
# Gated bytes must never land in a shared cache.
return ServeAsset({"Cache-Control": "private, no-cache",
"Vary": "Cookie"})
target = req.path + (("?" + req.url.split("?", 1)[1])
if "?" in req.url else "")
if req.is_document:
return redirect(f"/_pw/wall?next={quote(target)}")
return json_resp({"error": "payment_required",
"wall": f"/_pw/wall?next={quote(target)}"},
401 if not user else 402)
async def route(self, req: Req, sess: Session):
parts = req.path.strip("/").split("/")[1:] # drop "_pw"
head = parts[0] if parts else ""
arg = parts[1] if len(parts) > 1 else ""
method = req.method.upper()
if head == "webhook" and method == "POST" and arg in pay.PROVIDERS:
return await pay.PROVIDERS[arg].webhook(self, req)
if method not in ("GET", "HEAD") and not (
self.same_origin(req) or self.admin_via_token(req)):
return json_resp({"error": "cross-site request refused"}, 403)
if head == "wall":
return await self.wall_page(req, sess)
if head == "me":
user = sess.user()
until = await sess.entitled_until(force=True) if user else 0
return json_resp({
"user": user and {"name": user["name"], "email": user["email"]},
"entitled": until > self.now(),
"entitled_until": "lifetime" if until >= LIFETIME
else (until or None),
"mode": self.cfg["mode"], "wall": "/_pw/wall"})
if head == "login" and arg in self.auth_options() and arg != "magic":
url = await oauth.start(self, arg, sess.id_hash,
safe_next(req.query.get("next")))
return redirect(url)
if head == "callback" and arg in oauth.PROVIDERS:
profile, nxt = await oauth.finish(self, arg, req.query, sess.id_hash)
return await self.finish_login(sess, arg, profile, nxt)
if head == "magic" and not arg and method == "POST":
form = req.data()
if "magic" in self.auth_options():
await magic.send(self, form.get("email", ""),
safe_next(form.get("next")))
return self.page("Check your inbox", """<div class="card"><p>If that
address can sign in, a link is on its way. It expires in 15
minutes.</p></div>""")
if head == "magic" and arg == "verify":
tok = req.query.get("t") if method == "GET" else req.data().get("t")
if method == "GET":
return self.page("Sign in", f"""<div class="card"><form method="post">
<input type="hidden" name="t" value="{esc(tok or '')}">
<button class="primary">Finish signing in</button></form></div>""")
profile, nxt = await magic.redeem(self, tok or "")
return await self.finish_login(sess, "email", profile, nxt)
if head == "logout" and method == "POST":
sess.logout()
return redirect("/_pw/wall", 303)
if head == "checkout" and method == "POST" and arg in pay.PROVIDERS:
return await self.checkout(req, sess, arg)
if head == "return" and arg in pay.PROVIDERS:
return await pay.PROVIDERS[arg].on_return(self, req)
if head == "status" and arg:
return await self.status(sess, arg)
if head == "redeem" and method == "POST":
user = sess.user()
if not user:
return redirect("/_pw/wall", 303)
until = await pay.codes.redeem(self, user, req.data().get("code", ""))
if not until:
return self.error_page("That code is not valid or has been used.")
await sess.entitled_until(force=True)
return redirect("/", 303)
if head == "admin":
return await self.admin(req, sess, arg, method)
return json_resp({"error": "not found"}, 404)
async def finish_login(self, sess, provider, profile, nxt):
user = await self.store.login_identity(
provider, profile["subject"], profile["email"],
profile["verified"], profile["name"])
sess.login(user)
return redirect(safe_next(nxt))
async def checkout(self, req: Req, sess: Session, provider: str):
user = sess.user()
if not user:
return redirect("/_pw/wall", 303)
form = req.data()
currency = form.get("currency", "").upper()
plan = next((p for p in self.cfg["plans"] if p["id"] == form.get("plan")),
None)
if not plan or provider not in self.payment_options(currency):
return self.error_page("That plan or payment method is unavailable.")
amount = plan["prices"].get(currency)
if not amount:
return self.error_page("That plan is not sold in this currency.")
ref = await self.store.payment_create(provider, user["id"], plan,
amount, currency)
return await pay.PROVIDERS[provider].create(self, ref, plan, amount,
currency, user)
async def status(self, sess: Session, ref: str):
row = await self.store.payment_get(ref)
if not row or row["user_id"] != sess.uid:
return json_resp({"status": "unknown"}, 404)
if row["status"] in ("created", "pending") and \
row["provider"] == "mercadopago":
await pay.mercadopago.check(self, ref)
row = await self.store.payment_get(ref)
if row["status"] == "paid":
await sess.entitled_until(force=True)
return json_resp({"status": row["status"]})
# ---- admin -----------------------------------------------------------
async def admin(self, req: Req, sess: Session, arg: str, method: str):
if not self.is_admin(req, sess):
if req.is_document and not sess.user():
return redirect("/_pw/wall?next=/_pw/admin")
return json_resp({"error": "admin only"}, 403)
if arg == "config" and method == "GET":
return json_resp(self.cfg)
if arg == "config" and method in ("PUT", "POST"):
try:
self.cfg = await self.store.put_config(req.json())
except ValueError as exc:
return json_resp({"error": str(exc)}, 400)
return json_resp(self.cfg)
if arg == "grant" and method == "POST":
data = req.data()
user = await self.store.user_by_email(str(data.get("email", "")))
if not user:
return json_resp({"error": "no verified user with that email; "
"they must sign in once first"}, 404)
until = await self.store.grant(user["id"], int(data.get("days", 0)),
"admin")
return json_resp({"user_id": user["id"], "until": until})
if arg == "codes" and method == "POST":
data = req.data()
count = max(1, min(500, int(data.get("count", 1))))
days = int(data.get("days", 30))
uses = max(1, int(data.get("uses", 1)))
expires = int(data.get("expires_at", 0))
alphabet = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
made = []
for _ in range(count):
raw = "".join(secrets.choice(alphabet) for _ in range(12))
code = f"{raw[:4]}-{raw[4:8]}-{raw[8:]}"
await self.store.code_create(code, "voucher", days, uses, expires)
made.append(code)
return json_resp({"codes": made, "days": days, "uses": uses})
if arg == "payments":
return json_resp(await self.store.payments_recent(
int(req.query.get("limit", 50))))
if not arg:
return self.admin_page()
return json_resp({"error": "not found"}, 404)
def admin_page(self) -> Resp:
cfg = esc(json.dumps(self.cfg, indent=2))
return self.page("Paywall admin", f"""
<div class="card"><h2>Configuration</h2>
<p class="muted">Edit and save; changes reach every edge location within about
10 seconds. Set "mode" to "off" to drop the paywall entirely.</p>
<textarea id="cfg" rows="24">{cfg}</textarea>
<button class="primary" id="save">Save</button> <span id="msg"></span></div>
<div class="card"><h2>Grant access</h2>
<input id="gemail" type="email" placeholder="email of a signed-up user">
<input id="gdays" type="number" value="30" min="0"><small>0 = lifetime</small><br>
<button id="grant">Grant</button> <span id="gmsg"></span></div>
<div class="card"><h2>Voucher codes</h2>
<input id="ccount" type="number" value="10" min="1" max="500"> codes of
<input id="cdays" type="number" value="30" min="0"> days
<button id="mint">Generate</button><textarea id="cout" rows="5" readonly></textarea></div>
<div class="card"><h2>Recent payments</h2><div id="pay">Loading…</div></div>
<script>
const $=id=>document.getElementById(id);
async function call(path,method,body){{const r=await fetch(path,{{method,
headers:{{'Content-Type':'application/json'}},body:body&&JSON.stringify(body)}});
return [r.ok,await r.json()];}}
$('save').onclick=async()=>{{let v;try{{v=JSON.parse($('cfg').value)}}catch(e){{
$('msg').textContent='Invalid JSON: '+e.message;return}}
const[ok,j]=await call('/_pw/admin/config','PUT',v);
$('msg').textContent=ok?'Saved.':'Rejected: '+j.error;
if(ok)$('cfg').value=JSON.stringify(j,null,2);}};
$('grant').onclick=async()=>{{const[ok,j]=await call('/_pw/admin/grant','POST',
{{email:$('gemail').value,days:+$('gdays').value}});
$('gmsg').textContent=ok?'Granted until '+new Date(j.until*1000).toISOString():j.error;}};
$('mint').onclick=async()=>{{const[ok,j]=await call('/_pw/admin/codes','POST',
{{count:+$('ccount').value,days:+$('cdays').value}});
$('cout').value=ok?j.codes.join('\\n'):j.error;}};
call('/_pw/admin/payments?limit=25','GET').then(([ok,rows])=>{{
$('pay').innerHTML=ok&&rows.length?'<table><tr><th>When</th><th>Provider</th>'+
'<th>Amount</th><th>Status</th></tr>'+rows.map(r=>'<tr><td>'+
new Date(r.created_at*1000).toISOString().slice(0,16)+'</td><td>'+r.provider+
'</td><td>'+r.amount+' '+r.currency+'</td><td>'+r.status+'</td></tr>').join('')+
'</table>':'None yet.';}});
</script>""")
src/paywall/gate.py
"""The access decision. Pure function of config + visitor state, no I/O.
This is the one place to change if you want a different paywall policy
(for example, metering by article instead of by page load).
"""
from fnmatch import fnmatchcase
ALLOW = "allow" # serve the asset
METER = "meter" # serve it, and count one free page load
WALL = "wall" # show the paywall (documents) or 402 (everything else)
def is_public(cfg: dict, path: str) -> bool:
return any(fnmatchcase(path, pattern) for pattern in cfg["public_paths"])
def is_comped(cfg: dict, user: dict | None) -> bool:
"""Admins and allow-listed email domains get in free; verified email only."""
if not user or not user.get("email_verified"):
return False
email = user.get("email", "").lower()
domain = email.rpartition("@")[2]
return email in cfg["admin_emails"] or domain in cfg["free_email_domains"]
def decide(cfg: dict, path: str, *, user: dict | None, entitled: bool,
meter_used: int, is_document: bool) -> str:
if cfg["mode"] == "off" or is_public(cfg, path):
return ALLOW
if user and (cfg["mode"] == "login" or entitled or is_comped(cfg, user)):
return ALLOW
limit = cfg["metered_free_loads"]
if limit:
if is_document and meter_used < limit:
return METER
# Let the JS/CSS of a page that was metered in finish loading.
if not is_document and 0 < meter_used <= limit:
return ALLOW
return WALL
src/paywall/config.py
"""Operator-adjustable paywall configuration: defaults and validation.
The live copy is a JSON document in D1 (table `config`), edited through the
admin API or the /_pw/admin page. Every change goes through validate(), so a
typo can never take the site down: invalid config is rejected, not saved.
"""
import copy
import re
MODES = ("off", "login", "paid")
AUTH_PROVIDERS = ("google", "facebook", "github", "microsoft", "discord",
"gitlab", "oidc", "magic")
PAYMENT_PROVIDERS = ("stripe", "paypal", "razorpay", "mercadopago", "codes")
# ISO 4217 currencies whose minor unit is the major unit.
ZERO_DECIMAL = {"BIF", "CLP", "DJF", "GNF", "JPY", "KMF", "KRW", "MGA", "PYG",
"RWF", "UGX", "VND", "VUV", "XAF", "XOF", "XPF"}
# Euro area as of 2026 (Bulgaria joined on 1 January 2026).
EUROZONE = ("AT", "BE", "BG", "CY", "DE", "EE", "ES", "FI", "FR", "GR", "HR",
"IE", "IT", "LT", "LU", "LV", "MT", "NL", "PT", "SI", "SK")
DEFAULTS = {
"mode": "paid",
"site_name": "My App",
"wall_message": "Sign in and pick a pass to continue.",
"public_paths": ["/favicon.ico", "/robots.txt", "/public/*"],
"metered_free_loads": 0,
"admin_emails": [],
"free_email_domains": [],
"auth_providers": ["google", "facebook", "github", "magic"],
"payment_providers": ["stripe", "paypal", "razorpay", "mercadopago",
"codes"],
"provider_currencies": {
"stripe": ["USD", "EUR", "GBP", "CAD", "AUD", "CHF", "SEK", "PLN"],
"paypal": ["USD", "EUR", "GBP", "CAD", "AUD", "CHF", "SEK", "PLN"],
"razorpay": ["INR"],
"mercadopago": ["BRL"],
"codes": ["*"],
},
"default_currency": "USD",
"country_currency": {"IN": "INR", "BR": "BRL", "GB": "GBP",
**{c: "EUR" for c in EUROZONE}},
"plans": [
{"id": "pass30", "name": "30-day pass", "days": 30,
"prices": {"USD": 500, "EUR": 500, "GBP": 400, "INR": 19900,
"BRL": 1500}},
{"id": "lifetime", "name": "Lifetime access", "days": 0,
"prices": {"USD": 4900, "EUR": 4900, "GBP": 3900, "INR": 199900,
"BRL": 14900}},
],
"razorpay_upi_only": False,
"session_days": 30,
}
_SLUG = re.compile(r"^[a-z0-9][a-z0-9_-]{0,31}$")
_CUR = re.compile(r"^[A-Z]{3}$")
def _str_list(value, name):
if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
raise ValueError(f"{name} must be a list of strings")
return value
def validate(update: dict, base: dict | None = None) -> dict:
"""Merge `update` onto `base` (defaults if None) and validate the result."""
if not isinstance(update, dict):
raise ValueError("config must be a JSON object")
cfg = copy.deepcopy(base or DEFAULTS)
unknown = set(update) - set(DEFAULTS)
if unknown:
raise ValueError(f"unknown keys: {sorted(unknown)}")
cfg.update(copy.deepcopy(update))
if cfg["mode"] not in MODES:
raise ValueError(f"mode must be one of {MODES}")
for key in ("site_name", "wall_message", "default_currency"):
if not isinstance(cfg[key], str):
raise ValueError(f"{key} must be a string")
for key in ("public_paths", "admin_emails", "free_email_domains",
"auth_providers", "payment_providers"):
_str_list(cfg[key], key)
cfg["admin_emails"] = [e.lower() for e in cfg["admin_emails"]]
cfg["free_email_domains"] = [d.lower().lstrip("@")
for d in cfg["free_email_domains"]]
if bad := set(cfg["auth_providers"]) - set(AUTH_PROVIDERS):
raise ValueError(f"unknown auth providers: {sorted(bad)}")
if bad := set(cfg["payment_providers"]) - set(PAYMENT_PROVIDERS):
raise ValueError(f"unknown payment providers: {sorted(bad)}")
for key in ("metered_free_loads", "session_days"):
if not isinstance(cfg[key], int) or isinstance(cfg[key], bool) \
or cfg[key] < 0:
raise ValueError(f"{key} must be a non-negative integer")
if cfg["session_days"] < 1:
raise ValueError("session_days must be at least 1")
if not isinstance(cfg["razorpay_upi_only"], bool):
raise ValueError("razorpay_upi_only must be true or false")
if not _CUR.match(cfg["default_currency"]):
raise ValueError("default_currency must be an ISO 4217 code")
if not isinstance(cfg["country_currency"], dict) or not all(
_CUR.match(v) for v in cfg["country_currency"].values()):
raise ValueError("country_currency maps country codes to currencies")
pc = cfg["provider_currencies"]
if not isinstance(pc, dict):
raise ValueError("provider_currencies must be an object")
for prov, curs in pc.items():
if prov not in PAYMENT_PROVIDERS:
raise ValueError(f"provider_currencies: unknown provider {prov}")
_str_list(curs, f"provider_currencies.{prov}")
plans = cfg["plans"]
if not isinstance(plans, list) or not plans:
raise ValueError("plans must be a non-empty list")
seen = set()
for plan in plans:
if not isinstance(plan, dict) or not _SLUG.match(str(plan.get("id"))):
raise ValueError("each plan needs an id like 'pass30'")
if plan["id"] in seen:
raise ValueError(f"duplicate plan id {plan['id']}")
seen.add(plan["id"])
if not isinstance(plan.get("name"), str):
raise ValueError(f"plan {plan['id']}: name must be a string")
days = plan.get("days")
if not isinstance(days, int) or isinstance(days, bool) or days < 0:
raise ValueError(f"plan {plan['id']}: days must be >= 0 (0 = lifetime)")
prices = plan.get("prices")
if not isinstance(prices, dict) or not prices:
raise ValueError(f"plan {plan['id']}: prices must be an object")
for cur, amount in prices.items():
if not _CUR.match(cur) or not isinstance(amount, int) \
or isinstance(amount, bool) or amount <= 0:
raise ValueError(
f"plan {plan['id']}: prices map currency codes to positive "
f"integers in minor units (cents, paise, centavos)")
return cfg
def to_decimal_string(amount: int, currency: str) -> str:
"""Minor units -> '12.34' (or '1234' for zero-decimal currencies)."""
if currency.upper() in ZERO_DECIMAL:
return str(amount)
return f"{amount // 100}.{amount % 100:02d}"
def display_price(amount: int, currency: str) -> str:
return f"{to_decimal_string(amount, currency)} {currency.upper()}"
src/paywall/session.py
"""Server-side sessions in KV behind one opaque, HttpOnly cookie.
The cookie holds a random token; KV is keyed by its SHA-256, so a KV dump
cannot be replayed as cookies. The session also caches the user's
entitlement for ENTITLEMENT_TTL seconds so that serving a static asset
normally costs one KV read and no D1 queries.
"""
from . import crypto
from .web import cookie_header
ENTITLEMENT_TTL = 60
class Session:
def __init__(self, app, token=None, data=None):
self.app = app
self.token = token
self.data = data or {"uid": "", "meter": 0}
self.new = False
self.dirty = False
self.ended = False
self.stale_key = None
@property
def cookie_name(self) -> str:
return self.app.env_str("SESSION_COOKIE") or "__Host-pw"
@classmethod
async def load(cls, app, req):
session = cls(app)
tok = req.cookies.get(session.cookie_name)
if tok:
data = await app.store.kv_get(f"sess:{crypto.sha256_hex(tok)}")
if data:
session.token, session.data = tok, data
return session
@property
def uid(self) -> str:
return self.data.get("uid", "")
@property
def id_hash(self) -> str:
self.ensure()
return crypto.sha256_hex(self.token)
def ensure(self) -> None:
"""Make sure the browser has a session (anonymous if need be)."""
if not self.token:
self.token = crypto.token(32)
self.new = self.dirty = True
def login(self, user: dict) -> None:
# Rotate the token on privilege change (defeats session fixation).
if self.token:
self.stale_key = f"sess:{crypto.sha256_hex(self.token)}"
self.token = crypto.token(32)
self.data = {"uid": user["id"], "email": user["email"],
"email_verified": user["email_verified"],
"name": user["name"], "meter": 0,
"until": 0, "checked": 0}
self.new = self.dirty = True
def logout(self) -> None:
self.ended = True
def user(self) -> dict | None:
if not self.uid:
return None
return {"id": self.uid, "email": self.data.get("email", ""),
"email_verified": self.data.get("email_verified", False),
"name": self.data.get("name", "")}
async def entitled_until(self, force: bool = False) -> int:
if not self.uid:
return 0
now = self.app.now()
if force or now - self.data.get("checked", 0) > ENTITLEMENT_TTL:
self.data["until"] = await self.app.store.entitled_until(self.uid)
self.data["checked"] = int(now)
self.dirty = True
return int(self.data.get("until", 0))
async def commit(self, resp) -> None:
store, name = self.app.store, self.cookie_name
if self.stale_key:
await store.kv_delete(self.stale_key)
if self.ended:
if self.token:
await store.kv_delete(f"sess:{crypto.sha256_hex(self.token)}")
resp.headers["Set-Cookie"] = cookie_header(name, "", 0)
return
if self.dirty:
ttl = self.app.cfg["session_days"] * 86400
await store.kv_put(f"sess:{crypto.sha256_hex(self.token)}",
self.data, ttl)
if self.new:
resp.headers["Set-Cookie"] = cookie_header(name, self.token, ttl)
src/paywall/store.py
"""Persistence: D1 for durable records, KV for short-lived state.
D1 holds users, identities, entitlements, payments, voucher codes and the
config document. KV holds sessions, OAuth state and magic-link tokens, all
with a TTL so they clean themselves up.
Gotcha: Pyodide converts Python None to JavaScript `undefined`, which D1's
bind() rejects. The schema therefore avoids NULL parameters entirely
('' and 0 are used as "absent").
"""
import json
import time
import uuid
from . import config as cfgmod
LIFETIME = 253402300799 # 9999-12-31T23:59:59Z
_CONFIG_TTL = 10 # seconds an isolate may serve a cached config
_config_cache = {"at": 0.0, "value": None}
def _py(value):
"""Convert a JsProxy to Python if needed (no-op under CPython tests)."""
to_py = getattr(value, "to_py", None)
return to_py() if callable(to_py) else value
def _absent(value) -> bool:
"""True for None and for JS null/undefined, however Pyodide surfaces them."""
return value is None or repr(value) in ("null", "undefined", "jsnull")
def _js_opts(options: dict):
"""KV.put options must be a plain JS object, not a JS Map."""
try:
from js import Object
from pyodide.ffi import to_js
except ImportError:
return options
return to_js(options, dict_converter=Object.fromEntries)
class Store:
def __init__(self, env, now=time.time):
self.db = env.DB
self.kv = env.KV
self.now = now
# ---- D1 primitives -------------------------------------------------
async def _first(self, sql, *args):
row = await self.db.prepare(sql).bind(*args).first()
return None if _absent(row) else _py(row)
async def _all(self, sql, *args):
result = _py(await self.db.prepare(sql).bind(*args).all())
return [_py(r) for r in result["results"]]
async def _run(self, sql, *args) -> int:
result = _py(await self.db.prepare(sql).bind(*args).run())
return int(result["meta"]["changes"])
# ---- KV primitives -------------------------------------------------
async def kv_get(self, key):
raw = await self.kv.get(key)
return None if _absent(raw) or not raw else json.loads(str(raw))
async def kv_put(self, key, value, ttl: int):
# KV's minimum expirationTtl is 60 seconds.
await self.kv.put(key, json.dumps(value),
_js_opts({"expirationTtl": max(60, int(ttl))}))
async def kv_delete(self, key):
await self.kv.delete(key)
async def kv_take(self, key):
"""Read and delete: single-use tokens."""
value = await self.kv_get(key)
if value is not None:
await self.kv_delete(key)
return value
# ---- config ---------------------------------------------------------
async def get_config(self) -> dict:
now = self.now()
if _config_cache["value"] and now - _config_cache["at"] < _CONFIG_TTL:
return _config_cache["value"]
row = await self._first("SELECT json FROM config WHERE id = 1")
cfg = cfgmod.DEFAULTS
if row:
try:
cfg = cfgmod.validate(json.loads(row["json"]))
except ValueError:
pass # never brick the site on a bad stored row
_config_cache.update(at=now, value=cfg)
return cfg
async def put_config(self, update: dict) -> dict:
cfg = cfgmod.validate(update, await self.get_config())
await self._run(
"INSERT INTO config (id, json, updated_at) VALUES (1, ?, ?) "
"ON CONFLICT(id) DO UPDATE SET json = excluded.json, "
"updated_at = excluded.updated_at",
json.dumps(cfg), int(self.now()))
_config_cache.update(at=self.now(), value=cfg)
return cfg
# ---- users and identities ------------------------------------------
async def get_user(self, uid: str):
if not uid:
return None
row = await self._first(
"SELECT id, email, email_verified, name FROM users WHERE id = ?",
uid)
if row:
row["email_verified"] = bool(row["email_verified"])
return row
async def user_by_email(self, email: str):
row = await self._first(
"SELECT id FROM users WHERE email = ? AND email_verified = 1",
email.lower())
return await self.get_user(row["id"]) if row else None
async def login_identity(self, provider, subject, email, verified, name):
"""Find or create the user behind (provider, subject).
Links to an existing account only when both sides have a verified
email; linking on unverified email enables account takeover.
"""
email = (email or "").lower()
row = await self._first(
"SELECT user_id FROM identities WHERE provider = ? AND subject = ?",
provider, subject)
if row:
uid = row["user_id"]
else:
existing = await self.user_by_email(email) if verified and email \
else None
if existing:
uid = existing["id"]
else:
uid = uuid.uuid4().hex
await self._run(
"INSERT INTO users (id, email, email_verified, name, "
"created_at) VALUES (?, ?, ?, ?, ?)",
uid, email, 1 if verified else 0, name or "",
int(self.now()))
await self._run(
"INSERT INTO identities (provider, subject, user_id, "
"created_at) VALUES (?, ?, ?, ?)",
provider, subject, uid, int(self.now()))
if verified and email:
# Upgrade a previously unverified address.
await self._run(
"UPDATE users SET email = ?, email_verified = 1 "
"WHERE id = ? AND email_verified = 0", email, uid)
return await self.get_user(uid)
# ---- entitlements ----------------------------------------------------
async def entitled_until(self, uid: str) -> int:
if not uid:
return 0
row = await self._first(
"SELECT expires_at FROM entitlements WHERE user_id = ?", uid)
return int(row["expires_at"]) if row else 0
async def grant(self, uid: str, days: int, source: str) -> int:
"""Extend access by `days` (0 = lifetime). Stacks on unexpired time."""
now = int(self.now())
current = await self.entitled_until(uid)
if days == 0 or current >= LIFETIME:
until = LIFETIME
else:
until = max(now, current) + days * 86400
await self._run(
"INSERT INTO entitlements (user_id, expires_at, source, "
"updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(user_id) DO UPDATE "
"SET expires_at = excluded.expires_at, source = excluded.source, "
"updated_at = excluded.updated_at",
uid, until, source, now)
return until
# ---- payments ------------------------------------------------------
async def payment_create(self, provider, uid, plan, amount, currency):
ref = uuid.uuid4().hex
await self._run(
"INSERT INTO payments (ref, provider, external_id, txn_id, user_id, "
"plan_id, days, amount, currency, status, created_at, updated_at) "
"VALUES (?, ?, '', '', ?, ?, ?, ?, ?, 'created', ?, ?)",
ref, provider, uid, plan["id"], plan["days"], amount,
currency.upper(), int(self.now()), int(self.now()))
return ref
async def payment_get(self, ref: str):
return await self._first("SELECT * FROM payments WHERE ref = ?", ref)
async def payment_set_external(self, ref: str, external_id: str):
await self._run(
"UPDATE payments SET external_id = ?, updated_at = ? WHERE ref = ?",
external_id, int(self.now()), ref)
async def payment_transition(self, ref, status, txn_id="") -> bool:
"""Atomically move a payment out of a non-final state.
Returns True only for the caller that performed the transition, so
duplicate webhooks and browser returns cannot grant access twice.
`txn_id` is the provider's capture/payment id, kept for refunds.
"""
changes = await self._run(
"UPDATE payments SET status = ?, "
"txn_id = CASE WHEN ? = '' THEN txn_id ELSE ? END, "
"updated_at = ? WHERE ref = ? AND status IN ('created', 'pending')",
status, txn_id, txn_id, int(self.now()), ref)
return changes == 1
async def payments_recent(self, limit: int = 50):
return await self._all(
"SELECT ref, provider, external_id, txn_id, user_id, plan_id, amount, "
"currency, status, created_at FROM payments "
"ORDER BY created_at DESC LIMIT ?", int(limit))
# ---- voucher codes ---------------------------------------------------
async def code_create(self, code, plan_id, days, uses, expires_at):
await self._run(
"INSERT INTO codes (code, plan_id, days, uses_left, expires_at) "
"VALUES (?, ?, ?, ?, ?)", code, plan_id, days, uses, expires_at)
async def code_consume(self, code: str):
"""Decrement a code atomically; returns its row or None."""
changes = await self._run(
"UPDATE codes SET uses_left = uses_left - 1 WHERE code = ? AND "
"uses_left > 0 AND (expires_at = 0 OR expires_at > ?)",
code, int(self.now()))
if changes != 1:
return None
return await self._first("SELECT * FROM codes WHERE code = ?", code)
src/paywall/web.py
"""Runtime-neutral request/response types.
Nothing in this module touches Pyodide or the Workers runtime, so the whole
application can be unit-tested under ordinary CPython. src/entry.py converts
the runtime's Request into a Req and our Resp back into a Workers Response.
"""
from __future__ import annotations
import html as _html
import json
from dataclasses import dataclass, field
from http.cookies import SimpleCookie
from urllib.parse import parse_qsl, urlsplit
@dataclass
class Req:
method: str
url: str
headers: dict # keys are lower-case
body: bytes = b""
def __post_init__(self):
parts = urlsplit(self.url)
self.origin = f"{parts.scheme}://{parts.netloc}"
self.path = parts.path or "/"
self.query = dict(parse_qsl(parts.query, keep_blank_values=True))
def header(self, name: str, default: str = "") -> str:
return self.headers.get(name.lower(), default)
@property
def cookies(self) -> dict:
jar = SimpleCookie()
try:
jar.load(self.header("cookie"))
except Exception:
return {}
return {k: m.value for k, m in jar.items()}
@property
def country(self) -> str:
# Set by Cloudflare when IP geolocation is on (the default).
return self.header("cf-ipcountry").upper()
def text(self) -> str:
return self.body.decode("utf-8", "replace")
def json(self) -> dict:
try:
value = json.loads(self.body or b"{}")
except ValueError:
return {}
return value if isinstance(value, dict) else {}
def data(self) -> dict:
"""Body as a dict, whether it was sent as JSON or as a form."""
if "json" in self.header("content-type"):
return self.json()
return dict(parse_qsl(self.text()))
@property
def is_document(self) -> bool:
"""True for top-level page loads (as opposed to JS, CSS, images, XHR)."""
dest = self.header("sec-fetch-dest")
if dest:
return dest in ("document", "iframe")
return "text/html" in self.header("accept")
@dataclass
class Resp:
status: int = 200
body: str | bytes = ""
headers: dict = field(default_factory=dict)
@dataclass
class ServeAsset:
"""Tells entry.py to hand the request to the ASSETS binding."""
headers: dict = field(default_factory=dict)
def esc(value) -> str:
return _html.escape(str(value), quote=True)
def html(body: str, status: int = 200) -> Resp:
return Resp(status, body, {
"Content-Type": "text/html; charset=utf-8",
"Cache-Control": "no-store",
"X-Frame-Options": "DENY",
"Referrer-Policy": "same-origin",
})
def json_resp(obj, status: int = 200) -> Resp:
return Resp(status, json.dumps(obj), {
"Content-Type": "application/json",
"Cache-Control": "no-store",
})
def redirect(location: str, status: int = 302) -> Resp:
return Resp(status, "", {"Location": location, "Cache-Control": "no-store"})
def cookie_header(name: str, value: str, max_age: int) -> str:
# __Host- cookies must be Secure, Path=/ and carry no Domain attribute.
return (f"{name}={value}; Max-Age={max_age}; Path=/; Secure; HttpOnly; "
f"SameSite=Lax")
def safe_next(target: str | None) -> str:
"""Only allow same-origin relative paths as post-login destinations."""
if not target or not target.startswith("/") or target.startswith("//") \
or "\\" in target:
return "/"
return target
src/paywall/crypto.py
"""Small crypto helpers. Pyodide ships hashlib, hmac and secrets from CPython."""
import base64
import hashlib
import hmac
import secrets
def token(nbytes: int = 32) -> str:
return secrets.token_urlsafe(nbytes)
def sha256_hex(value: str) -> str:
return hashlib.sha256(value.encode()).hexdigest()
def hmac_sha256_hex(key: str, message: str | bytes) -> str:
if isinstance(message, str):
message = message.encode()
return hmac.new(key.encode(), message, hashlib.sha256).hexdigest()
def safe_eq(a: str, b: str) -> bool:
return hmac.compare_digest(a.encode(), b.encode())
def pkce_pair() -> tuple[str, str]:
"""Returns (code_verifier, S256 code_challenge) per RFC 7636."""
verifier = token(48)
digest = hashlib.sha256(verifier.encode()).digest()
challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
return verifier, challenge
src/paywall/net.py
"""Outbound HTTP. The App takes a `fetcher` so tests can inject a fake."""
import json
from dataclasses import dataclass
@dataclass
class HttpResult:
status: int
text: str
def json(self):
try:
return json.loads(self.text)
except ValueError:
return {}
@property
def ok(self) -> bool:
return 200 <= self.status < 300
async def workers_fetch(method: str, url: str, headers: dict | None = None,
body: str | None = None) -> HttpResult:
from workers import fetch # only importable inside the Workers runtime
kwargs = {"method": method, "headers": headers or {}}
if body is not None:
kwargs["body"] = body
resp = await fetch(url, **kwargs)
return HttpResult(int(resp.status), await resp.text())
src/paywall/auth/oauth.py
"""OAuth 2.0 / OpenID Connect sign-in, driven by a provider table.
Adding a provider means adding one PROVIDERS entry and, if its profile JSON
is unusual, a branch in normalize(). The flow is the authorization-code
grant with a server-side `state` (bound to the browser's session) and PKCE
where the provider supports it. Profiles come from each provider's userinfo
endpoint over TLS with the fresh access token, so no ID-token signature
checking (and no JWKS fetching) is needed.
"""
from urllib.parse import urlencode
from .. import crypto
FB = "{fb}" # replaced by the FACEBOOK_GRAPH_VERSION var, e.g. v23.0
PROVIDERS = {
"google": {
"label": "Google",
"authorize": "https://accounts.google.com/o/oauth2/v2/auth",
"token": "https://oauth2.googleapis.com/token",
"userinfo": "https://openidconnect.googleapis.com/v1/userinfo",
"scope": "openid email profile",
"pkce": True,
},
"facebook": {
"label": "Facebook",
"authorize": f"https://www.facebook.com/{FB}/dialog/oauth",
"token": f"https://graph.facebook.com/{FB}/oauth/access_token",
"userinfo": f"https://graph.facebook.com/{FB}/me?fields=id,name,email",
"scope": "email,public_profile",
"pkce": False,
},
"github": {
"label": "GitHub",
"authorize": "https://github.com/login/oauth/authorize",
"token": "https://github.com/login/oauth/access_token",
"userinfo": "https://api.github.com/user",
"emails": "https://api.github.com/user/emails",
"scope": "read:user user:email",
"pkce": True,
},
"microsoft": {
"label": "Microsoft",
"authorize": "https://login.microsoftonline.com/common/oauth2/v2.0/authorize",
"token": "https://login.microsoftonline.com/common/oauth2/v2.0/token",
"userinfo": "https://graph.microsoft.com/oidc/userinfo",
"scope": "openid email profile",
"pkce": True,
},
"discord": {
"label": "Discord",
"authorize": "https://discord.com/oauth2/authorize",
"token": "https://discord.com/api/oauth2/token",
"userinfo": "https://discord.com/api/users/@me",
"scope": "identify email",
"pkce": False,
},
"gitlab": {
"label": "GitLab",
"authorize": "{gitlab}/oauth/authorize",
"token": "{gitlab}/oauth/token",
"userinfo": "{gitlab}/oauth/userinfo",
"scope": "openid email profile",
"pkce": True,
},
# Any standards-compliant issuer: Auth0, Okta, Keycloak, Zitadel, ...
# Endpoints come from {OIDC_ISSUER}/.well-known/openid-configuration.
"oidc": {
"label": "Single sign-on",
"discovery": True,
"scope": "openid email profile",
"pkce": True,
},
}
def client_credentials(env, name: str) -> tuple[str, str]:
prefix = name.upper()
return (getattr(env, f"{prefix}_CLIENT_ID", "") or "",
getattr(env, f"{prefix}_CLIENT_SECRET", "") or "")
def configured(env, name: str) -> bool:
cid, secret = client_credentials(env, name)
if name == "oidc" and not getattr(env, "OIDC_ISSUER", ""):
return False
return bool(cid and secret)
async def endpoints(app, name: str) -> dict:
spec = dict(PROVIDERS[name])
if spec.get("discovery"):
issuer = app.env_str("OIDC_ISSUER").rstrip("/")
meta = app.discovery_cache.get(issuer)
if meta is None:
resp = await app.http("GET", f"{issuer}/.well-known/openid-configuration")
meta = resp.json()
app.discovery_cache[issuer] = meta
spec.update(authorize=meta["authorization_endpoint"],
token=meta["token_endpoint"],
userinfo=meta["userinfo_endpoint"])
spec["label"] = app.env_str("OIDC_LABEL") or spec["label"]
subst = {"fb": app.env_str("FACEBOOK_GRAPH_VERSION") or "v23.0",
"gitlab": (app.env_str("GITLAB_BASE_URL")
or "https://gitlab.com").rstrip("/")}
for key in ("authorize", "token", "userinfo"):
spec[key] = spec[key].replace("{fb}", subst["fb"]).replace(
"{gitlab}", subst["gitlab"])
return spec
async def start(app, name: str, session_id_hash: str, next_path: str) -> str:
"""Return the provider URL to redirect the browser to."""
spec = await endpoints(app, name)
client_id, _ = client_credentials(app.env, name)
state = crypto.token(24)
params = {
"response_type": "code",
"client_id": client_id,
"redirect_uri": f"{app.origin}/_pw/callback/{name}",
"scope": spec["scope"],
"state": state,
}
record = {"provider": name, "sid": session_id_hash, "next": next_path}
if spec["pkce"]:
verifier, challenge = crypto.pkce_pair()
params.update(code_challenge=challenge, code_challenge_method="S256")
record["verifier"] = verifier
await app.store.kv_put(f"oauth:{state}", record, ttl=600)
return f"{spec['authorize']}?{urlencode(params)}"
async def finish(app, name: str, query: dict, session_id_hash: str):
"""Validate the callback and return (profile, next_path)."""
if "error" in query:
raise PermissionError(query.get("error_description") or query["error"])
record = await app.store.kv_take(f"oauth:{query.get('state', '')}")
if not record or record["provider"] != name:
raise PermissionError("sign-in expired; please try again")
if not crypto.safe_eq(record["sid"], session_id_hash):
raise PermissionError("sign-in was started in another browser")
spec = await endpoints(app, name)
client_id, client_secret = client_credentials(app.env, name)
form = {
"grant_type": "authorization_code",
"code": query.get("code", ""),
"redirect_uri": f"{app.origin}/_pw/callback/{name}",
"client_id": client_id,
"client_secret": client_secret,
}
if "verifier" in record:
form["code_verifier"] = record["verifier"]
tok = await app.http("POST", spec["token"], {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json",
}, urlencode(form))
access_token = tok.json().get("access_token")
if not tok.ok or not access_token:
raise PermissionError("the provider refused the sign-in code")
headers = {"Authorization": f"Bearer {access_token}",
"Accept": "application/json",
"User-Agent": "paywall-worker"} # GitHub requires a UA
info = (await app.http("GET", spec["userinfo"], headers)).json()
extra = {}
if "emails" in spec:
extra["emails"] = (await app.http("GET", spec["emails"], headers)).json()
return normalize(name, info, extra), record["next"]
def normalize(name: str, info: dict, extra: dict) -> dict:
"""Map each provider's profile JSON onto subject/email/verified/name."""
if name == "github":
primary = next((e for e in extra.get("emails") or []
if isinstance(e, dict) and e.get("primary")), {})
return {"subject": str(info["id"]), "email": primary.get("email", ""),
"verified": bool(primary.get("verified")),
"name": info.get("name") or info.get("login", "")}
if name == "facebook":
# Graph only returns an email the user has confirmed with Facebook.
email = info.get("email", "")
return {"subject": str(info["id"]), "email": email,
"verified": bool(email), "name": info.get("name", "")}
if name == "discord":
return {"subject": str(info["id"]), "email": info.get("email") or "",
"verified": bool(info.get("verified")),
"name": info.get("global_name") or info.get("username", "")}
if name == "microsoft":
# Entra lets tenant admins set arbitrary unverified mail attributes,
# so never trust it for account linking ("nOAuth").
return {"subject": str(info["sub"]), "email": info.get("email", ""),
"verified": False, "name": info.get("name", "")}
verified = info.get("email_verified")
return {"subject": str(info["sub"]), "email": info.get("email", ""),
"verified": verified is True or verified == "true",
"name": info.get("name") or info.get("preferred_username", "")}
src/paywall/auth/magic.py
"""Passwordless email sign-in ("magic links").
Sends through any JSON email API; the defaults match Resend
(POST https://api.resend.com/emails with a Bearer key). Clicking the link
shows a confirm button that POSTs, because corporate mail scanners prefetch
GET links and would otherwise burn the one-time token.
"""
import re
from .. import crypto
from ..web import esc
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
MAX_PER_HOUR = 5
def configured(env) -> bool:
return bool(getattr(env, "EMAIL_API_KEY", "") and
getattr(env, "EMAIL_FROM", ""))
async def send(app, email: str, next_path: str) -> None:
email = email.strip().lower()
if not EMAIL_RE.match(email):
raise ValueError("enter a valid email address")
limit_key = f"magic-rate:{crypto.sha256_hex(email)}"
sent = (await app.store.kv_get(limit_key)) or 0
if sent >= MAX_PER_HOUR:
return # silently drop; the UI message is identical either way
await app.store.kv_put(limit_key, sent + 1, ttl=3600)
tok = crypto.token(32)
await app.store.kv_put(f"magic:{crypto.sha256_hex(tok)}",
{"email": email, "next": next_path}, ttl=900)
link = f"{app.origin}/_pw/magic/verify?t={tok}"
site = esc(app.cfg["site_name"])
await app.http("POST", app.env_str("EMAIL_API_URL")
or "https://api.resend.com/emails", {
"Authorization": f"Bearer {app.env_str('EMAIL_API_KEY')}",
"Content-Type": "application/json",
}, app.dumps({
"from": app.env_str("EMAIL_FROM"),
"to": [email],
"subject": f"Your sign-in link for {app.cfg['site_name']}",
"html": (f"<p>Click to sign in to {site}. The link works once and "
f"expires in 15 minutes.</p><p><a href=\"{esc(link)}\">"
f"Sign in</a></p>"),
}))
async def redeem(app, tok: str):
"""Consume a token; returns (profile, next_path) or raises."""
record = await app.store.kv_take(f"magic:{crypto.sha256_hex(tok)}")
if not record:
raise PermissionError("this sign-in link has expired or was used")
profile = {"subject": record["email"], "email": record["email"],
"verified": True, "name": record["email"].split("@")[0]}
return profile, record["next"]
src/paywall/pay/__init__.py
"""Payment providers share one contract.
configured(env) -> bool
async create(app, ref, plan, amount, currency, user) -> Resp
async on_return(app, req) -> Resp (browser comes back)
async webhook(app, req) -> Resp (provider calls us)
Every provider funnels confirmed money into fulfill(), which checks the
amount and currency against the row written at checkout time (so a
tampered client cannot pay 1 cent for a lifetime pass) and grants access
exactly once.
"""
import logging
from . import codes, mercadopago, paypal, razorpay, stripe
PROVIDERS = {
"stripe": stripe,
"paypal": paypal,
"razorpay": razorpay,
"mercadopago": mercadopago,
"codes": codes,
}
LABELS = {
"stripe": "Card, Apple Pay, Google Pay, Wero/iDEAL, SEPA",
"paypal": "PayPal",
"razorpay": "UPI (Razorpay)",
"mercadopago": "Pix (Mercado Pago)",
"codes": "Voucher code",
}
log = logging.getLogger("paywall")
async def fulfill(app, ref: str, provider: str, amount: int, currency: str,
external_id: str = "") -> bool:
row = await app.store.payment_get(ref)
if not row or row["provider"] != provider:
log.warning("payment %s: unknown ref or provider mismatch", ref)
return False
if row["status"] == "paid":
return True # duplicate notification; already granted
if int(amount) != int(row["amount"]) or \
currency.upper() != row["currency"].upper():
await app.store.payment_transition(ref, "mismatch", external_id)
log.warning("payment %s: expected %s %s, got %s %s", ref,
row["amount"], row["currency"], amount, currency)
return False
if await app.store.payment_transition(ref, "paid", external_id):
await app.store.grant(row["user_id"], int(row["days"]),
f"{provider}:{external_id or ref}")
return True
src/paywall/pay/stripe.py
"""Stripe Checkout: cards, Apple Pay, Google Pay, Link, Wero / iDEAL,
Bancontact, SEPA Direct Debit, Klarna and more.
No payment_method_types is sent, so Stripe shows whatever methods are
enabled in Dashboard > Settings > Payment methods, filtered by the buyer's
country and currency. That is how one integration covers the EU's card and
wallet majority plus local rails such as Wero (which replaces iDEAL in the
Netherlands from late 2026).
"""
import time
from urllib.parse import urlencode
from .. import crypto
from ..web import json_resp, redirect
API = "https://api.stripe.com/v1"
TOLERANCE = 300 # seconds of clock skew accepted on webhook timestamps
def configured(env) -> bool:
return bool(getattr(env, "STRIPE_SECRET_KEY", "") and
getattr(env, "STRIPE_WEBHOOK_SECRET", ""))
def _auth(app):
return {"Authorization": f"Bearer {app.env_str('STRIPE_SECRET_KEY')}",
"Content-Type": "application/x-www-form-urlencoded"}
async def create(app, ref, plan, amount, currency, user):
form = [
("mode", "payment"),
("client_reference_id", ref),
("metadata[ref]", ref),
("payment_intent_data[metadata][ref]", ref),
("line_items[0][quantity]", "1"),
("line_items[0][price_data][currency]", currency.lower()),
("line_items[0][price_data][unit_amount]", str(amount)),
("line_items[0][price_data][product_data][name]", plan["name"]),
("success_url", f"{app.origin}/_pw/return/stripe?ref={ref}"),
("cancel_url", f"{app.origin}/_pw/wall"),
]
if user.get("email"):
form.append(("customer_email", user["email"]))
resp = await app.http("POST", f"{API}/checkout/sessions", _auth(app),
urlencode(form))
session = resp.json()
if not resp.ok or "url" not in session:
raise RuntimeError(session.get("error", {}).get("message", "Stripe error"))
await app.store.payment_set_external(ref, session["id"])
return redirect(session["url"], 303)
async def on_return(app, req):
# Card payments usually settle before the redirect, but the webhook is
# the source of truth; the status page polls until it lands.
return app.pending_page(req.query.get("ref", ""))
def verify_signature(payload: bytes, header: str, secret: str,
now: float | None = None) -> bool:
parts = {}
for item in header.split(","):
key, _, value = item.strip().partition("=")
parts.setdefault(key, []).append(value)
try:
timestamp = int(parts["t"][0])
except (KeyError, ValueError):
return False
if abs((now or time.time()) - timestamp) > TOLERANCE:
return False
expected = crypto.hmac_sha256_hex(secret, f"{timestamp}.".encode() + payload)
return any(crypto.safe_eq(expected, sig) for sig in parts.get("v1", []))
async def webhook(app, req):
if not verify_signature(req.body, req.header("stripe-signature"),
app.env_str("STRIPE_WEBHOOK_SECRET"), app.now()):
return json_resp({"error": "bad signature"}, 400)
event = req.json()
obj = event.get("data", {}).get("object", {})
paid = (event.get("type") == "checkout.session.completed"
and obj.get("payment_status") == "paid") or \
event.get("type") == "checkout.session.async_payment_succeeded"
if paid:
from . import fulfill
await fulfill(app, obj.get("client_reference_id", ""), "stripe",
obj.get("amount_total", 0), obj.get("currency", ""),
obj.get("id", ""))
elif event.get("type") == "checkout.session.async_payment_failed":
await app.store.payment_transition(
obj.get("client_reference_id", ""), "failed", obj.get("id", ""))
return json_resp({"received": True})
src/paywall/pay/paypal.py
"""PayPal Orders v2: create an order, send the buyer to approve it, capture
on return. The PAYMENT.CAPTURE.COMPLETED webhook is a backstop for buyers
who close the tab before the redirect lands; PayPal verifies its own
webhook signatures through the verify-webhook-signature API.
"""
from .. import config as cfgmod
from ..web import json_resp, redirect
LIVE = "https://api-m.paypal.com"
SANDBOX = "https://api-m.sandbox.paypal.com"
def configured(env) -> bool:
return bool(getattr(env, "PAYPAL_CLIENT_ID", "") and
getattr(env, "PAYPAL_CLIENT_SECRET", ""))
def _base(app):
return LIVE if app.env_str("PAYPAL_ENV") == "live" else SANDBOX
async def _token(app) -> str:
import base64
basic = base64.b64encode(
f"{app.env_str('PAYPAL_CLIENT_ID')}:{app.env_str('PAYPAL_CLIENT_SECRET')}"
.encode()).decode()
resp = await app.http("POST", f"{_base(app)}/v1/oauth2/token", {
"Authorization": f"Basic {basic}",
"Content-Type": "application/x-www-form-urlencoded",
}, "grant_type=client_credentials")
if not resp.ok:
raise RuntimeError("PayPal authentication failed")
return resp.json()["access_token"]
def _minor(value: str, currency: str) -> int:
whole, _, frac = value.partition(".")
if currency.upper() in cfgmod.ZERO_DECIMAL:
return int(whole)
return int(whole) * 100 + int((frac + "00")[:2])
async def create(app, ref, plan, amount, currency, user):
tok = await _token(app)
body = {
"intent": "CAPTURE",
"purchase_units": [{
"reference_id": ref,
"custom_id": ref,
"description": plan["name"],
"amount": {"currency_code": currency.upper(),
"value": cfgmod.to_decimal_string(amount, currency)},
}],
"payment_source": {"paypal": {"experience_context": {
"brand_name": app.cfg["site_name"][:127],
"shipping_preference": "NO_SHIPPING",
"user_action": "PAY_NOW",
"return_url": f"{app.origin}/_pw/return/paypal?ref={ref}",
"cancel_url": f"{app.origin}/_pw/wall",
}}},
}
resp = await app.http("POST", f"{_base(app)}/v2/checkout/orders", {
"Authorization": f"Bearer {tok}",
"Content-Type": "application/json",
"PayPal-Request-Id": f"create-{ref}",
}, app.dumps(body))
order = resp.json()
link = next((l["href"] for l in order.get("links", [])
if l.get("rel") in ("payer-action", "approve")), None)
if not resp.ok or not link:
raise RuntimeError("PayPal could not create the order")
await app.store.payment_set_external(ref, order["id"])
return redirect(link, 303)
async def _record_capture(app, capture: dict):
from . import fulfill
amt = capture.get("amount", {})
cur = amt.get("currency_code", "")
await fulfill(app, capture.get("custom_id", ""), "paypal",
_minor(amt.get("value", "0"), cur), cur,
capture.get("id", ""))
async def on_return(app, req):
ref = req.query.get("ref", "")
row = await app.store.payment_get(ref)
order_id = req.query.get("token", "")
if not row or row["external_id"] != order_id:
return app.error_page("That PayPal order does not match this checkout.")
tok = await _token(app)
resp = await app.http(
"POST", f"{_base(app)}/v2/checkout/orders/{order_id}/capture", {
"Authorization": f"Bearer {tok}",
"Content-Type": "application/json",
"PayPal-Request-Id": f"capture-{ref}", # idempotent retries
}, "{}")
order = resp.json()
for unit in order.get("purchase_units", []):
for capture in unit.get("payments", {}).get("captures", []):
if capture.get("status") == "COMPLETED":
capture.setdefault("custom_id", unit.get("custom_id", ref))
await _record_capture(app, capture)
return app.pending_page(ref)
async def webhook(app, req):
h = req.header
tok = await _token(app)
check = await app.http(
"POST", f"{_base(app)}/v1/notifications/verify-webhook-signature", {
"Authorization": f"Bearer {tok}",
"Content-Type": "application/json",
}, app.dumps({
"auth_algo": h("paypal-auth-algo"),
"cert_url": h("paypal-cert-url"),
"transmission_id": h("paypal-transmission-id"),
"transmission_sig": h("paypal-transmission-sig"),
"transmission_time": h("paypal-transmission-time"),
"webhook_id": app.env_str("PAYPAL_WEBHOOK_ID"),
"webhook_event": req.json(),
}))
if check.json().get("verification_status") != "SUCCESS":
return json_resp({"error": "bad signature"}, 400)
event = req.json()
if event.get("event_type") == "PAYMENT.CAPTURE.COMPLETED":
await _record_capture(app, event.get("resource", {}))
return json_resp({"received": True})
src/paywall/pay/razorpay.py
"""Razorpay Payment Links for UPI (India).
A Payment Link is a hosted checkout page, so the Worker never loads
Razorpay's JavaScript. With razorpay_upi_only on, Razorpay creates a
UPI-only link; otherwise the hosted page offers UPI alongside cards,
netbanking and wallets (UPI is listed first for Indian buyers). Check
Razorpay's docs for whether UPI-only links work in test mode on your
account before relying on that flag in staging.
"""
import base64
from .. import crypto
from ..web import json_resp, redirect
API = "https://api.razorpay.com/v1"
def configured(env) -> bool:
return bool(getattr(env, "RAZORPAY_KEY_ID", "") and
getattr(env, "RAZORPAY_KEY_SECRET", "") and
getattr(env, "RAZORPAY_WEBHOOK_SECRET", ""))
def _auth(app):
basic = base64.b64encode(
f"{app.env_str('RAZORPAY_KEY_ID')}:{app.env_str('RAZORPAY_KEY_SECRET')}"
.encode()).decode()
return {"Authorization": f"Basic {basic}",
"Content-Type": "application/json"}
async def create(app, ref, plan, amount, currency, user):
body = {
"amount": amount, # paise
"currency": currency.upper(), # INR
"accept_partial": False,
"reference_id": ref, # max 40 chars; ours is 32
"description": plan["name"][:2048],
"customer": {"email": user.get("email", ""),
"name": user.get("name", "")},
"notify": {"sms": False, "email": False},
"reminder_enable": False,
"callback_url": f"{app.origin}/_pw/return/razorpay?ref={ref}",
"callback_method": "get",
"notes": {"ref": ref},
}
if app.cfg["razorpay_upi_only"]:
body["upi_link"] = True
resp = await app.http("POST", f"{API}/payment_links", _auth(app),
app.dumps(body))
link = resp.json()
if not resp.ok or "short_url" not in link:
raise RuntimeError(link.get("error", {}).get("description",
"Razorpay error"))
await app.store.payment_set_external(ref, link["id"])
return redirect(link["short_url"], 303)
def callback_signature_ok(q: dict, key_secret: str) -> bool:
message = "|".join([q.get("razorpay_payment_link_id", ""),
q.get("razorpay_payment_link_reference_id", ""),
q.get("razorpay_payment_link_status", ""),
q.get("razorpay_payment_id", "")])
expected = crypto.hmac_sha256_hex(key_secret, message)
return crypto.safe_eq(expected, q.get("razorpay_signature", ""))
async def on_return(app, req):
q = req.query
ref = q.get("razorpay_payment_link_reference_id", "")
row = await app.store.payment_get(ref)
if not row or row["external_id"] != q.get("razorpay_payment_link_id") \
or not callback_signature_ok(q, app.env_str("RAZORPAY_KEY_SECRET")):
return app.error_page("We could not verify that UPI payment.")
if q.get("razorpay_payment_link_status") == "paid":
from . import fulfill
# The signed callback proves this link was paid in full, and a link
# can only be created for the amount stored in our row.
await fulfill(app, ref, "razorpay", row["amount"], row["currency"],
q.get("razorpay_payment_id", ""))
return app.pending_page(ref)
async def webhook(app, req):
expected = crypto.hmac_sha256_hex(app.env_str("RAZORPAY_WEBHOOK_SECRET"),
req.body)
if not crypto.safe_eq(expected, req.header("x-razorpay-signature")):
return json_resp({"error": "bad signature"}, 400)
event = req.json()
if event.get("event") == "payment_link.paid":
from . import fulfill
payload = event.get("payload", {})
link = payload.get("payment_link", {}).get("entity", {})
payment = payload.get("payment", {}).get("entity", {})
await fulfill(app, link.get("reference_id", ""), "razorpay",
link.get("amount_paid", 0), link.get("currency", ""),
payment.get("id", ""))
return json_resp({"received": True})
src/paywall/pay/mercadopago.py
"""Pix (Brazil) through Mercado Pago's Payments API.
Pix has no redirect: we create a payment, show its QR code and
copy-and-paste code ("Pix Copia e Cola"), and wait. The page polls
/_pw/status, which also asks Mercado Pago directly in case the webhook is
slow. Webhooks carry no payment data, only an id, so after checking the
x-signature HMAC we fetch the payment and read its status ourselves.
"""
import time
from datetime import datetime, timezone
from .. import config as cfgmod
from .. import crypto
from ..web import esc, json_resp
API = "https://api.mercadopago.com"
def configured(env) -> bool:
return bool(getattr(env, "MERCADOPAGO_ACCESS_TOKEN", "") and
getattr(env, "MERCADOPAGO_WEBHOOK_SECRET", ""))
def _auth(app, idempotency_key=None):
h = {"Authorization": f"Bearer {app.env_str('MERCADOPAGO_ACCESS_TOKEN')}",
"Content-Type": "application/json"}
if idempotency_key:
h["X-Idempotency-Key"] = idempotency_key
return h
async def create(app, ref, plan, amount, currency, user):
if not user.get("email"):
raise RuntimeError("Pix needs an email address on your account")
expires = datetime.fromtimestamp(app.now() + 1800, timezone.utc)
body = {
"transaction_amount": float(cfgmod.to_decimal_string(amount, currency)),
"description": plan["name"],
"payment_method_id": "pix",
"payer": {"email": user["email"]},
"external_reference": ref,
"notification_url": f"{app.origin}/_pw/webhook/mercadopago",
"date_of_expiration": expires.strftime("%Y-%m-%dT%H:%M:%S.000+00:00"),
}
resp = await app.http("POST", f"{API}/v1/payments", _auth(app, ref),
app.dumps(body))
payment = resp.json()
tx = payment.get("point_of_interaction", {}).get("transaction_data", {})
if not resp.ok or not tx.get("qr_code"):
raise RuntimeError(payment.get("message", "Mercado Pago error"))
await app.store.payment_set_external(ref, str(payment["id"]))
await app.store.payment_transition(ref, "pending")
return app.page("Pay with Pix", f"""
<p>Scan with your bank app, or copy the Pix code. This page updates by
itself once the payment clears.</p>
<img alt="Pix QR code" width="240" height="240"
src="data:image/png;base64,{esc(tx.get('qr_code_base64', ''))}">
<textarea readonly rows="4" onclick="this.select()">{esc(tx['qr_code'])}</textarea>
<button onclick="navigator.clipboard.writeText(this.previousElementSibling.value)">
Copy Pix code</button>
{app.poll_script(ref)}""")
async def check(app, ref: str) -> None:
"""Ask Mercado Pago for the payment's status (used by polling)."""
row = await app.store.payment_get(ref)
if not row or not row["external_id"]:
return
resp = await app.http("GET", f"{API}/v1/payments/{row['external_id']}",
_auth(app))
await _apply(app, resp.json())
async def _apply(app, payment: dict) -> None:
if payment.get("status") != "approved":
return
from . import fulfill
cur = payment.get("currency_id", "BRL")
amount = round(float(payment.get("transaction_amount", 0)) * 100)
await fulfill(app, payment.get("external_reference", ""), "mercadopago",
amount, cur, str(payment.get("id", "")))
def signature_ok(data_id: str, request_id: str, header: str, secret: str,
now: float | None = None, tolerance: int = 600) -> bool:
parts = dict(p.strip().split("=", 1) for p in header.split(",") if "=" in p)
ts, v1 = parts.get("ts", ""), parts.get("v1", "")
if not ts or not v1:
return False
try:
# Mercado Pago sends milliseconds; tolerate seconds too.
stamp = int(ts) / (1000 if len(ts) > 10 else 1)
except ValueError:
return False
if abs((now or time.time()) - stamp) > tolerance:
return False
manifest = f"id:{data_id};request-id:{request_id};ts:{ts};"
return crypto.safe_eq(crypto.hmac_sha256_hex(secret, manifest), v1)
async def webhook(app, req):
data_id = req.query.get("data.id") or str(
req.json().get("data", {}).get("id", ""))
if data_id.isalnum():
data_id = data_id.lower()
if not signature_ok(data_id, req.header("x-request-id"),
req.header("x-signature"),
app.env_str("MERCADOPAGO_WEBHOOK_SECRET"), app.now()):
return json_resp({"error": "bad signature"}, 400)
if req.json().get("type") == "payment" and data_id:
resp = await app.http("GET", f"{API}/v1/payments/{data_id}", _auth(app))
await _apply(app, resp.json())
return json_resp({"received": True})
async def on_return(app, req):
return app.pending_page(req.query.get("ref", ""))
src/paywall/pay/codes.py
"""Operator-issued voucher codes: gifts, press passes, scholarships, refunds
handled out of band. Issued through POST /_pw/admin/codes."""
from ..web import json_resp
def configured(env) -> bool:
return True
async def redeem(app, user: dict, code: str):
row = await app.store.code_consume(code.strip().upper())
if not row:
return None
return await app.store.grant(user["id"], int(row["days"]),
f"code:{row['code']}")
async def create(app, ref, plan, amount, currency, user):
raise RuntimeError("voucher codes are redeemed, not purchased")
async def on_return(app, req):
return app.pending_page(req.query.get("ref", ""))
async def webhook(app, req):
return json_resp({"error": "no webhooks for codes"}, 404)
tests/fakes.py
"""In-memory stand-ins for KV, D1 and outbound fetch, for CPython tests."""
import json
import pathlib
import sqlite3
from types import SimpleNamespace
from urllib.parse import parse_qsl
from paywall.net import HttpResult
SCHEMA = (pathlib.Path(__file__).parent.parent / "schema.sql").read_text()
class FakeKV:
def __init__(self):
self.data = {}
async def get(self, key):
return self.data.get(key)
async def put(self, key, value, options=None):
assert options and options["expirationTtl"] >= 60
self.data[key] = value
async def delete(self, key):
self.data.pop(key, None)
class _Stmt:
def __init__(self, conn, sql):
self.conn, self.sql, self.args = conn, sql, ()
def bind(self, *args):
assert None not in args, "D1 cannot bind None (becomes undefined)"
self.args = args
return self
async def first(self):
cur = self.conn.execute(self.sql, self.args)
row = cur.fetchone()
return dict(row) if row else None
async def all(self):
rows = self.conn.execute(self.sql, self.args).fetchall()
return {"results": [dict(r) for r in rows]}
async def run(self):
cur = self.conn.execute(self.sql, self.args)
self.conn.commit()
return {"meta": {"changes": cur.rowcount}}
class FakeD1:
def __init__(self):
self.conn = sqlite3.connect(":memory:")
self.conn.row_factory = sqlite3.Row
self.conn.executescript(SCHEMA)
def prepare(self, sql):
return _Stmt(self.conn, sql)
class FakeHttp:
"""Routes outbound calls to handlers registered by URL prefix."""
def __init__(self):
self.routes = []
self.calls = []
def on(self, method, prefix, handler):
self.routes.append((method, prefix, handler))
async def __call__(self, method, url, headers=None, body=None):
self.calls.append((method, url, headers or {}, body))
# Longest matching prefix wins; among equals, the latest registration.
matches = [(len(p), i, h) for i, (m, p, h) in enumerate(self.routes)
if m == method and url.startswith(p)]
if not matches:
raise AssertionError(f"unexpected outbound call {method} {url}")
status, payload = max(matches)[2](url, headers or {}, body)
return HttpResult(status, json.dumps(payload))
def make_env(**secrets):
return SimpleNamespace(DB=FakeD1(), KV=FakeKV(), **secrets)
def form(body):
return dict(parse_qsl(body or ""))
tests/test_paywall.py
import asyncio
import json
import time
from urllib.parse import parse_qs, urlsplit
import pytest
from fakes import FakeHttp, form, make_env
from paywall import config, crypto, gate, store
from paywall.app import App
from paywall.pay import mercadopago, razorpay, stripe
from paywall.web import Req, ServeAsset
ORIGIN = "https://app.example"
SECRETS = dict(
GITHUB_CLIENT_ID="gh-id", GITHUB_CLIENT_SECRET="gh-secret",
GOOGLE_CLIENT_ID="g-id", GOOGLE_CLIENT_SECRET="g-secret",
STRIPE_SECRET_KEY="sk_test", STRIPE_WEBHOOK_SECRET="whsec_test",
PAYPAL_CLIENT_ID="pp-id", PAYPAL_CLIENT_SECRET="pp-secret",
PAYPAL_WEBHOOK_ID="WH-1",
RAZORPAY_KEY_ID="rzp_test", RAZORPAY_KEY_SECRET="rzp-secret",
RAZORPAY_WEBHOOK_SECRET="rzp-wh",
MERCADOPAGO_ACCESS_TOKEN="mp-token", MERCADOPAGO_WEBHOOK_SECRET="mp-wh",
ADMIN_TOKEN="admintoken",
)
@pytest.fixture(autouse=True)
def fresh_config_cache():
store._config_cache.update(at=0.0, value=None)
class Browser:
"""Keeps one cookie jar across requests, like a real browser."""
def __init__(self, env, http):
self.env, self.http, self.cookie = env, http, ""
def request(self, method, path, body=b"", headers=None, document=True):
h = {"origin": ORIGIN, "cookie": self.cookie,
"sec-fetch-dest": "document" if document else "script"}
h.update(headers or {})
if isinstance(body, dict):
body = json.dumps(body).encode()
h.setdefault("content-type", "application/json")
resp = asyncio.run(App(self.env, self.http).handle(
Req(method, ORIGIN + path, h, body)))
cookie = resp.headers.get("Set-Cookie", "")
if cookie:
value = cookie.split(";")[0]
self.cookie = "" if value.endswith("=") else value
return resp
get = lambda self, p, **kw: self.request("GET", p, **kw)
post = lambda self, p, body=b"", **kw: self.request("POST", p, body, **kw)
def github_login(browser, http, gh_id=42, email="ada@example.org"):
http.on("POST", "https://github.com/login/oauth/access_token",
lambda u, h, b: (200, {"access_token": "gho_x"}))
http.on("GET", "https://api.github.com/user/emails",
lambda u, h, b: (200, [{"email": email, "primary": True,
"verified": True}]))
http.on("GET", "https://api.github.com/user",
lambda u, h, b: (200, {"id": gh_id, "login": "ada"}))
start = browser.get("/_pw/login/github?next=/docs/intro")
q = parse_qs(urlsplit(start.headers["Location"]).query)
assert q["code_challenge_method"] == ["S256"]
done = browser.get(f"/_pw/callback/github?code=abc&state={q['state'][0]}")
assert done.headers["Location"] == "/docs/intro"
# ---- pure logic ------------------------------------------------------------
def test_config_validation_rejects_bad_prices_and_keys():
with pytest.raises(ValueError):
config.validate({"plans": [{"id": "x", "name": "X", "days": 1,
"prices": {"USD": 4.99}}]})
with pytest.raises(ValueError):
config.validate({"moed": "off"})
cfg = config.validate({"mode": "off", "admin_emails": ["Ops@Site.org"]})
assert cfg["mode"] == "off" and cfg["admin_emails"] == ["ops@site.org"]
def test_gate_modes_and_metering():
cfg = config.validate({"metered_free_loads": 2})
d = lambda **kw: gate.decide(cfg, "/article", **{
"user": None, "entitled": False, "meter_used": 0,
"is_document": True, **kw})
assert d() == gate.METER
assert d(meter_used=2) == gate.WALL
assert d(meter_used=1, is_document=False) == gate.ALLOW
assert gate.decide(cfg, "/favicon.ico", user=None, entitled=False,
meter_used=9, is_document=False) == gate.ALLOW
comped = config.validate({"free_email_domains": ["@uni.edu"]})
user = {"email": "s@uni.edu", "email_verified": True}
assert gate.decide(comped, "/x", user=user, entitled=False, meter_used=0,
is_document=True) == gate.ALLOW
user["email_verified"] = False
assert gate.decide(comped, "/x", user=user, entitled=False, meter_used=0,
is_document=True) == gate.WALL
def test_stripe_signature():
body, now = b'{"id":"evt"}', time.time()
sig = crypto.hmac_sha256_hex("whsec", f"{int(now)}.".encode() + body)
assert stripe.verify_signature(body, f"t={int(now)},v1={sig}", "whsec", now)
assert not stripe.verify_signature(body, f"t={int(now)},v1={sig}", "x", now)
assert not stripe.verify_signature(body, f"t={int(now) - 999},v1={sig}",
"whsec", now)
def test_mercadopago_and_razorpay_signatures():
ts = str(int(time.time() * 1000))
v1 = crypto.hmac_sha256_hex("s", f"id:123;request-id:r-1;ts:{ts};")
assert mercadopago.signature_ok("123", "r-1", f"ts={ts},v1={v1}", "s")
assert not mercadopago.signature_ok("124", "r-1", f"ts={ts},v1={v1}", "s")
q = {"razorpay_payment_link_id": "plink_1",
"razorpay_payment_link_reference_id": "ref",
"razorpay_payment_link_status": "paid",
"razorpay_payment_id": "pay_1"}
q["razorpay_signature"] = crypto.hmac_sha256_hex(
"k", "plink_1|ref|paid|pay_1")
assert razorpay.callback_signature_ok(q, "k")
assert not razorpay.callback_signature_ok({**q, "razorpay_payment_link_status":
"partially_paid"}, "k")
# ---- request flows -----------------------------------------------------------
def test_anonymous_document_is_walled_and_assets_get_402():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
resp = b.get("/docs/intro?x=1")
assert resp.status == 302
assert resp.headers["Location"] == "/_pw/wall?next=/docs/intro%3Fx%3D1"
assert b.get("/assets/app.js", document=False).status == 401
wall = b.get("/_pw/wall?next=/docs/intro")
assert "Continue with GitHub" in wall.body and "Continue with Google" in wall.body
assert "Facebook" not in wall.body # enabled in config but not configured
def test_github_login_then_stripe_purchase_grants_once():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
github_login(b, http)
assert b.get("/docs/intro").status == 302 # signed in, not paid
http.on("POST", "https://api.stripe.com/v1/checkout/sessions",
lambda u, h, body: (200, {"id": "cs_1", "url":
"https://checkout.stripe.com/c/cs_1",
"_sent": form(body)}))
resp = b.post("/_pw/checkout/stripe", b"plan=pass30¤cy=EUR",
headers={"content-type": "application/x-www-form-urlencoded"})
assert resp.status == 303 and "checkout.stripe.com" in resp.headers["Location"]
sent = form(http.calls[-1][3])
assert sent["line_items[0][price_data][unit_amount]"] == "500"
assert sent["line_items[0][price_data][currency]"] == "eur"
ref = sent["client_reference_id"]
event = json.dumps({"type": "checkout.session.completed", "data": {"object": {
"id": "cs_1", "client_reference_id": ref, "payment_status": "paid",
"amount_total": 500, "currency": "eur"}}}).encode()
now = int(time.time())
sig = crypto.hmac_sha256_hex("whsec_test", f"{now}.".encode() + event)
hook = lambda: Browser(env, http).post(
"/_pw/webhook/stripe", event,
headers={"stripe-signature": f"t={now},v1={sig}", "origin": ""})
assert hook().status == 200
first = asyncio.run(store.Store(env).entitled_until(_uid(env)))
assert hook().status == 200 # duplicate delivery
assert asyncio.run(store.Store(env).entitled_until(_uid(env))) == first
assert abs(first - (now + 30 * 86400)) < 5
assert json.loads(b.get(f"/_pw/status/{ref}").body)["status"] == "paid"
served = b.get("/docs/intro")
assert isinstance(served, ServeAsset)
assert served.headers["Cache-Control"] == "private, no-cache"
def _uid(env):
return env.DB.conn.execute("SELECT id FROM users").fetchone()[0]
def test_tampered_amount_is_refused():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
github_login(b, http)
ref = asyncio.run(store.Store(env).payment_create(
"stripe", _uid(env), config.DEFAULTS["plans"][1], 4900, "USD"))
event = json.dumps({"type": "checkout.session.completed", "data": {"object": {
"id": "cs_2", "client_reference_id": ref, "payment_status": "paid",
"amount_total": 1, "currency": "usd"}}}).encode()
now = int(time.time())
sig = crypto.hmac_sha256_hex("whsec_test", f"{now}.".encode() + event)
b.post("/_pw/webhook/stripe", event,
headers={"stripe-signature": f"t={now},v1={sig}"})
row = asyncio.run(store.Store(env).payment_get(ref))
assert row["status"] == "mismatch"
assert asyncio.run(store.Store(env).entitled_until(_uid(env))) == 0
def test_paypal_return_captures():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
github_login(b, http)
http.on("POST", "https://api-m.sandbox.paypal.com/v1/oauth2/token",
lambda u, h, body: (200, {"access_token": "A21"}))
http.on("POST", "https://api-m.sandbox.paypal.com/v2/checkout/orders/ORD1/capture",
lambda u, h, body: (201, {"status": "COMPLETED", "purchase_units": [
{"custom_id": REF[0], "payments": {"captures": [{
"id": "CAP1", "status": "COMPLETED",
"amount": {"currency_code": "USD", "value": "5.00"}}]}}]}))
http.on("POST", "https://api-m.sandbox.paypal.com/v2/checkout/orders",
lambda u, h, body: (201, {"id": "ORD1", "links": [
{"rel": "payer-action", "href": "https://www.sandbox.paypal.com/x"}]}))
REF = []
resp = b.post("/_pw/checkout/paypal", {"plan": "pass30", "currency": "USD"})
assert resp.headers["Location"].startswith("https://www.sandbox.paypal.com")
REF.append(json.loads(http.calls[-1][3])["purchase_units"][0]["custom_id"])
b.get(f"/_pw/return/paypal?ref={REF[0]}&token=ORD1&PayerID=P")
assert asyncio.run(store.Store(env).payment_get(REF[0]))["status"] == "paid"
assert isinstance(b.get("/docs/intro"), ServeAsset)
def test_pix_flow_polls_mercadopago():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
github_login(b, http)
state = {"status": "pending"}
http.on("POST", "https://api.mercadopago.com/v1/payments",
lambda u, h, body: (201, {"id": 987, "status": "pending",
"point_of_interaction": {"transaction_data": {
"qr_code": "00020126...", "qr_code_base64": "iVBOR"}}}))
http.on("GET", "https://api.mercadopago.com/v1/payments/987",
lambda u, h, body: (200, {"id": 987, "status": state["status"],
"external_reference": REF[0],
"transaction_amount": 15.0,
"currency_id": "BRL"}))
REF = []
page = b.post("/_pw/checkout/mercadopago", {"plan": "pass30", "currency": "BRL"})
assert "00020126..." in page.body
REF.append(json.loads(http.calls[-1][3])["external_reference"])
assert json.loads(b.get(f"/_pw/status/{REF[0]}").body)["status"] == "pending"
state["status"] = "approved"
assert json.loads(b.get(f"/_pw/status/{REF[0]}").body)["status"] == "paid"
def test_razorpay_currency_routing():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
github_login(b, http)
wall = b.get("/_pw/wall", headers={"cf-ipcountry": "IN"})
assert "UPI (Razorpay)" in wall.body and "199.00 INR" in wall.body
assert "Pix" not in wall.body
refused = b.post("/_pw/checkout/razorpay", {"plan": "pass30", "currency": "USD"})
assert refused.status == 400
def test_cross_site_post_and_open_redirect_are_blocked():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
evil = b.post("/_pw/logout", headers={"origin": "https://evil.example"})
assert evil.status == 403
start = b.get("/_pw/login/github?next=//evil.example")
state = parse_qs(urlsplit(start.headers["Location"]).query)["state"][0]
record = json.loads(env.KV.data[f"oauth:{state}"])
assert record["next"] == "/"
def test_oauth_state_is_bound_to_browser():
env, http = make_env(**SECRETS), FakeHttp()
alice, mallory = Browser(env, http), Browser(env, http)
start = mallory.get("/_pw/login/github")
state = parse_qs(urlsplit(start.headers["Location"]).query)["state"][0]
resp = alice.get(f"/_pw/callback/github?code=x&state={state}")
assert resp.status == 403
def test_admin_can_switch_paywall_off_and_mint_codes():
env, http = make_env(**SECRETS), FakeHttp()
admin = {"authorization": "Bearer admintoken", "origin": ""}
b = Browser(env, http)
assert b.request("PUT", "/_pw/admin/config", {"mode": "off"},
headers={"origin": ""}).status == 403
assert b.request("PUT", "/_pw/admin/config", {"mode": "nope"},
headers=admin).status == 400
assert b.request("PUT", "/_pw/admin/config", {"mode": "off"},
headers=admin).status == 200
store._config_cache.update(at=0.0, value=None) # another isolate
assert isinstance(b.get("/docs/intro"), ServeAsset)
b.request("PUT", "/_pw/admin/config", {"mode": "paid"}, headers=admin)
codes = json.loads(b.post("/_pw/admin/codes", {"count": 1, "days": 7},
headers=admin).body)["codes"]
github_login(b, http)
b.post("/_pw/redeem", b"code=" + codes[0].lower().encode(),
headers={"content-type": "application/x-www-form-urlencoded"})
assert isinstance(b.get("/docs/intro"), ServeAsset)
other = Browser(env, http)
github_login(other, http, gh_id=7, email="bob@example.org")
again = other.post("/_pw/redeem", b"code=" + codes[0].encode(),
headers={"content-type": "application/x-www-form-urlencoded"})
assert again.status == 400
def test_verified_email_links_accounts_but_unverified_does_not():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
github_login(b, http, email="ada@example.org")
http.on("POST", "https://oauth2.googleapis.com/token",
lambda u, h, body: (200, {"access_token": "ya29"}))
profile = {"sub": "g-1", "email": "ada@example.org", "email_verified": True}
http.on("GET", "https://openidconnect.googleapis.com/v1/userinfo",
lambda u, h, body: (200, profile))
g = Browser(env, http)
start = g.get("/_pw/login/google")
state = parse_qs(urlsplit(start.headers["Location"]).query)["state"][0]
g.get(f"/_pw/callback/google?code=c&state={state}")
assert env.DB.conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] == 1
profile.update(sub="g-2", email_verified=False)
h = Browser(env, http)
start = h.get("/_pw/login/google")
state = parse_qs(urlsplit(start.headers["Location"]).query)["state"][0]
h.get(f"/_pw/callback/google?code=c&state={state}")
assert env.DB.conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] == 2
def test_session_cookie_rotates_on_login():
env, http = make_env(**SECRETS), FakeHttp()
b = Browser(env, http)
b.get("/_pw/login/github")
before = b.cookie
github_login(b, http)
assert b.cookie and b.cookie != before
assert f"sess:{crypto.sha256_hex(before.split('=', 1)[1])}" not in env.KV.data
spa/dist/index.html
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Demo SPA</title>
</head>
<body>
<main id="app">Loading…</main>
<script>
// Placeholder SPA: replace spa/dist with your real build output.
fetch("/_pw/me").then(r => r.json()).then(me => {
const app = document.getElementById("app");
if (!me.user) {
app.innerHTML = '<p>Public preview. <a href="/_pw/wall">Sign in</a></p>';
return;
}
app.innerHTML = '<h1></h1><p></p><form method="post" action="/_pw/logout">' +
'<button>Sign out</button></form>';
// textContent, never innerHTML, for names that came from a provider
app.querySelector("h1").textContent = "Welcome, " + (me.user.name || me.user.email);
app.querySelector("p").textContent = "You are reading paid content at " + location.pathname;
});
</script>
</body>
</html>
.gitignore
.dev.vars
.venv/
.wrangler/
node_modules/
python_modules/
__pycache__/
.pytest_cache/