1
Fork 0
blog-python-postgres/extras/auth-clerk/clerk_auth.py
Leonardo Devai 80f2a6dbb0 Review and modernize all 42 projects to the updated standard
Pinned 2026 toolchains (Go 1.26, Rust 1.98/edition 2024, Python 3.14 + uv, Node 24,
Zig 0.16, NixOS 26.05), postgres 18 / mongo 8, lockfiles built from, non-root
runtimes, .dockerignore, per-project LICENSE, READMEs with the git.devai.io clone
line, checkout@v7 CI. Security fixes in the legacy Rust APIs (any-password login,
self-assigned admin, hard-coded JWT secret), JWT alg/exp/sub enforcement across the
blog series, safe markdown links in the frontends, and many smaller bugs — every
project was built, run and exercised end to end.

Adds scripts/publish.sh + a CI publish job that splits every folder into its own
repo at git.devai.io/templates/<folder>.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01128fhuZbgivaSJvtMf4s1G
2026-09-27 21:10:38 +02:00

34 lines
1.3 KiB
Python

import os
import jwt
from fastapi import HTTPException, Request
ISSUER = os.environ["CLERK_ISSUER"].rstrip("/")
# Origins allowed in the token's azp claim, comma-separated; unset skips the check.
AUTHORIZED_PARTIES = [p for p in os.environ.get("CLERK_AUTHORIZED_PARTIES", "").split(",") if p]
# Caches the key set; an unknown key id (a rotation) triggers a refetch, at most every 30s.
_jwks = jwt.PyJWKClient(f"{ISSUER}/.well-known/jwks.json")
def current_user_id(request: Request) -> str:
"""FastAPI dependency: the Clerk user id (`sub`) from a valid session token."""
header = request.headers.get("authorization", "")
if not header.lower().startswith("bearer "):
raise HTTPException(401, "missing bearer token")
token = header[7:]
try:
key = _jwks.get_signing_key_from_jwt(token).key
# Clerk session tokens live for 60s, so allow a little clock skew.
claims = jwt.decode(
token,
key,
algorithms=["RS256"],
issuer=ISSUER,
leeway=5,
options={"require": ["exp", "iss", "sub"]},
)
except jwt.PyJWTError:
raise HTTPException(401, "invalid or expired token")
if AUTHORIZED_PARTIES and claims.get("azp") not in AUTHORIZED_PARTIES:
raise HTTPException(401, "token issued for an unknown origin")
return claims["sub"]