kid-sid/claude-spellbook/auth

# Python — PyJWT
import jwt
from datetime import datetime, timedelta, UTC
SECRET = "..."  # use RS256 with a key pair in production
def issue_token(user_id: str, roles: list[str]) -> str:
    return jwt.encode(
        {
            "sub": user_id,
            "roles": roles,
            "iat": datetime.now(UTC),
            "exp": datetime.now(UTC) + timedelta(minutes=15),
        },
        SECRET,
        algorithm="HS256",
    )
def verify_token(token: str) -> dict:
    return jwt.decode(token, SECRET, algorithms=["HS256"])

// TypeScript — jose
import { SignJWT, jwtVerify } from "jose";
const secret = new TextEncoder().encode(process.env.JWT_SECRET);
async function issueToken(userId: string, roles: string[]): Promise<string> {
  return new SignJWT({ sub: userId, roles })
    .setProtectedHeader({ alg: "HS256" })
    .setIssuedAt()
    .setExpirationTime("15m")
    .sign(secret);
}
async function verifyToken(token: string) {
  const { payload } = await jwtVerify(token, secret);
  return payload;
}

// Go — golang-jwt/jwt
import (
    "github.com/golang-jwt/jwt/v5"
    "time"
)
 
type Claims struct {
    Roles []string `json:"roles"`
    jwt.RegisteredClaims
}
 
func IssueToken(userID string, roles []string, secret []byte) (string, error) {
    claims := Claims{
        Roles: roles,
        RegisteredClaims: jwt.RegisteredClaims{
            Subject:   userID,
            IssuedAt:  jwt.NewNumericDate(time.Now()),
            ExpiresAt: jwt.NewNumericDate(time.Now().Add(15 * time.Minute)),
        },
    }
    return jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString(secret)
}

// 1. Generate PKCE values
function generateCodeVerifier(): string {
  const array = new Uint8Array(32);
  crypto.getRandomValues(array);
  return btoa(String.fromCharCode(...array)).replace(/[+/=]/g, (c) =>
    ({ "+": "-", "/": "_", "=": "" })[c]!
  );
}
async function generateCodeChallenge(verifier: string): Promise<string> {
  const data = new TextEncoder().encode(verifier);
  const digest = await crypto.subtle.digest("SHA-256", data);
  return btoa(String.fromCharCode(...new Uint8Array(digest)))
    .replace(/[+/=]/g, (c) => ({ "+": "-", "/": "_", "=": "" })[c]!);
}
// 2. Redirect to provider
const verifier = generateCodeVerifier();
sessionStorage.setItem("pkce_verifier", verifier);
const challenge = await generateCodeChallenge(verifier);
const params = new URLSearchParams({
  response_type: "code",
  client_id: CLIENT_ID,
  redirect_uri: REDIRECT_URI,
  scope: "openid profile email",
  code_challenge: challenge,
  code_challenge_method: "S256",
  state: crypto.randomUUID(), // store and verify on return
});
window.location.href = `${PROVIDER_URL}/authorize?${params}`;
// 3. Exchange code for tokens (on redirect back)
async function handleCallback(code: string): Promise<void> {
  const verifier = sessionStorage.getItem("pkce_verifier")!;
  const res = await fetch(`${PROVIDER_URL}/token`, {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({
      grant_type: "authorization_code",
      code,
      redirect_uri: REDIRECT_URI,
      client_id: CLIENT_ID,
      code_verifier: verifier,
    }),
  });
  const { access_token, refresh_token, id_token } = await res.json();
  // store access_token in memory, refresh_token in HttpOnly cookie via backend
}

# Python — httpx
import httpx
def get_m2m_token(client_id: str, client_secret: str, token_url: str) -> str:
    r = httpx.post(
        token_url,
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
            "scope": "api:read api:write",
        },
    )
    r.raise_for_status()
    return r.json()["access_token"]

let accessToken: string | null = null;
async function getValidToken(): Promise<string> {
  if (accessToken && !isExpiringSoon(accessToken)) return accessToken;
  const res = await fetch("/auth/refresh", {
    method: "POST",
    credentials: "include", // sends HttpOnly refresh token cookie
  });
  if (!res.ok) {
    // refresh token expired — redirect to login
    window.location.href = "/login";
    throw new Error("Session expired");
  }
  accessToken = (await res.json()).access_token;
  return accessToken;
}
function isExpiringSoon(token: string): boolean {
  const { exp } = JSON.parse(atob(token.split(".")[1]));
  return exp * 1000 - Date.now() < 60_000; // refresh if < 60s left
}

# FastAPI
from functools import wraps
from fastapi import Depends, HTTPException, status
from typing import Callable
def require_roles(*roles: str) -> Callable:
    def dependency(token_data: dict = Depends(get_current_user)):
        user_roles = set(token_data.get("roles", []))
        if not user_roles.intersection(roles):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
        return token_data
    return dependency
@router.delete("/users/{user_id}")
async def delete_user(
    user_id: str,
    _=Depends(require_roles("admin")),
):
    ...

// Express middleware
function requireRoles(...roles: string[]) {
  return (req: Request, res: Response, next: NextFunction) => {
    const userRoles: string[] = res.locals.user?.roles ?? [];
    if (!roles.some((r) => userRoles.includes(r))) {
      return res.status(403).json({ error: "Forbidden" });
    }
    next();
  };
}
router.delete("/users/:id", requireRoles("admin"), deleteUserHandler);

# Simple policy engine
def can(user: dict, action: str, resource: dict) -> bool:
    if "admin" in user["roles"]:
        return True
    if action == "read" and resource["public"]:
        return True
    if action in ("update", "delete") and resource["owner_id"] == user["id"]:
        return True
    return False
# Usage
if not can(current_user, "delete", post):
    raise HTTPException(status_code=403)

# Python — argon2-cffi
from argon2 import PasswordHasher
ph = PasswordHasher()
def hash_password(password: str) -> str:
    return ph.hash(password)
def verify_password(hashed: str, password: str) -> bool:
    try:
        return ph.verify(hashed, password)
    except Exception:
        return False