Claude
5f3b2b2874
- Login with registration code, sessions, rate limiting, per-user characters - Share characters with other users (read-only), admin CLI - Rules presets with a Presets tab, applied/pending changes with diff in a side sheet - Equipment as tiles with amounts, starting kit removed - Rename nav tab to "Character", add "New character" button - Stop tracking _to_delete/ Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
437 lines
18 KiB
Python
437 lines
18 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Body, Depends, FastAPI, HTTPException, Request, Response
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from . import auth
|
|
from .export import to_docx, to_pdf
|
|
|
|
BASE = Path(__file__).resolve().parent.parent
|
|
RULES_PATH = Path(os.environ.get("AW_RULES", BASE / "data" / "rules.json"))
|
|
DB_PATH = os.environ.get("AW_DB", str(BASE / "aw.db"))
|
|
STATIC = Path(os.environ.get("AW_STATIC", BASE / "static"))
|
|
MAX_BODY = 200_000 # bytes per character
|
|
MAX_PRESETS = 200
|
|
MAX_CHARACTERS = 300 # per user
|
|
MAX_SHARES = 30 # per character
|
|
|
|
# No public API docs: everything about the API stays behind the login
|
|
app = FastAPI(title="Arcane Wasteland Character Creator", docs_url=None, redoc_url=None, openapi_url=None)
|
|
limiter = auth.Limiter()
|
|
TRUST_PROXY = os.environ.get("AW_TRUST_PROXY", "") == "1" # behind a reverse proxy that sets X-Forwarded-For / -Proto
|
|
|
|
|
|
|
|
def load_rules() -> dict:
|
|
# Re-read on every request so rules.json can be edited without a restart.
|
|
return json.loads(RULES_PATH.read_text(encoding="utf-8"))
|
|
|
|
|
|
def db() -> sqlite3.Connection:
|
|
Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
|
|
con = sqlite3.connect(DB_PATH)
|
|
con.row_factory = sqlite3.Row
|
|
con.execute("""CREATE TABLE IF NOT EXISTS characters (
|
|
id TEXT PRIMARY KEY, token TEXT NOT NULL, data TEXT NOT NULL,
|
|
created REAL NOT NULL, updated REAL NOT NULL)""")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS presets (
|
|
id TEXT PRIMARY KEY, token TEXT NOT NULL, data TEXT NOT NULL,
|
|
created REAL NOT NULL, updated REAL NOT NULL)""")
|
|
auth.init_tables(con)
|
|
con.execute("""CREATE TABLE IF NOT EXISTS shares (
|
|
character_id TEXT NOT NULL, user_id INTEGER NOT NULL, PRIMARY KEY (character_id, user_id))""")
|
|
for table in ("characters", "presets"): # older databases: add the owner column
|
|
if "owner_id" not in [r["name"] for r in con.execute(f"PRAGMA table_info({table})")]:
|
|
con.execute(f"ALTER TABLE {table} ADD COLUMN owner_id INTEGER")
|
|
return con
|
|
|
|
|
|
def custom_presets(con, uid=None) -> list:
|
|
rows = con.execute("""SELECT p.id, p.data, p.owner_id, u.username FROM presets p
|
|
LEFT JOIN users u ON u.id = p.owner_id ORDER BY p.created""").fetchall()
|
|
return [{"id": r["id"], **json.loads(r["data"]), "custom": True, "author": r["username"] or "",
|
|
"mine": uid is not None and r["owner_id"] == uid} for r in rows]
|
|
|
|
|
|
def all_rules(uid=None) -> dict:
|
|
"""Rules from rules.json plus the presets created in the app."""
|
|
r = load_rules()
|
|
with db() as con:
|
|
custom = custom_presets(con, uid)
|
|
r["presets"] = [{**p, "builtin": True} for p in r.get("presets", [])] + custom
|
|
return r
|
|
|
|
|
|
def clean_preset(body: dict, rules: dict) -> dict:
|
|
"""Validates a preset: name, description and only known creation numbers within their limits."""
|
|
name = str(body.get("name", "")).strip()[:60]
|
|
if not name:
|
|
raise HTTPException(422, "A preset needs a name")
|
|
tun = {t["key"]: t for t in rules.get("tunables", [])}
|
|
creation = {}
|
|
for k, v in (body.get("creation") or {}).items():
|
|
if k not in tun:
|
|
raise HTTPException(422, f"Unknown value: {k}")
|
|
if isinstance(v, bool) or not isinstance(v, int) or not tun[k]["min"] <= v <= tun[k]["max"]:
|
|
raise HTTPException(422, f"Value out of range: {tun[k]['label']}")
|
|
if v != rules["creation"].get(k):
|
|
creation[k] = v
|
|
return {"name": name, "desc": str(body.get("desc", "")).strip()[:300], "creation": creation}
|
|
|
|
|
|
def check_size(ch: dict):
|
|
if len(json.dumps(ch)) > MAX_BODY:
|
|
raise HTTPException(413, "Character too large")
|
|
|
|
|
|
# ---------------------------------------------------------------- security
|
|
|
|
def client_ip(request: Request) -> str:
|
|
if TRUST_PROXY:
|
|
fwd = request.headers.get("x-forwarded-for", "")
|
|
if fwd:
|
|
return fwd.split(",")[-1].strip() # the address the reverse proxy saw
|
|
return request.client.host if request.client else "?"
|
|
|
|
|
|
def is_https(request: Request) -> bool:
|
|
if TRUST_PROXY:
|
|
return request.headers.get("x-forwarded-proto", "") == "https"
|
|
return request.url.scheme == "https"
|
|
|
|
|
|
@app.middleware("http")
|
|
async def security(request: Request, call_next):
|
|
# Block cross-site writes (cookies are SameSite=Lax as well)
|
|
if request.method not in ("GET", "HEAD", "OPTIONS"):
|
|
origin = request.headers.get("origin")
|
|
hosts = {request.headers.get("host", "")}
|
|
if TRUST_PROXY:
|
|
hosts.add(request.headers.get("x-forwarded-host", ""))
|
|
if origin and origin.split("://", 1)[-1] not in hosts:
|
|
return JSONResponse({"detail": "Cross-site request refused"}, status_code=403)
|
|
resp = await call_next(request)
|
|
resp.headers["X-Content-Type-Options"] = "nosniff"
|
|
resp.headers["X-Frame-Options"] = "DENY"
|
|
resp.headers["Referrer-Policy"] = "same-origin"
|
|
resp.headers["Content-Security-Policy"] = ("default-src 'self'; style-src 'self' 'unsafe-inline'; "
|
|
"img-src 'self' data:; frame-ancestors 'none'")
|
|
if request.url.path.startswith("/api/"):
|
|
resp.headers["Cache-Control"] = "no-store"
|
|
return resp
|
|
|
|
|
|
def current_user(request: Request) -> dict:
|
|
with db() as con:
|
|
user = auth.user_for_token(con, request.cookies.get(auth.SESSION_COOKIE))
|
|
if not user:
|
|
raise HTTPException(401, "Not logged in")
|
|
return user
|
|
|
|
|
|
def set_cookie(request: Request, resp: Response, token: str):
|
|
resp.set_cookie(auth.SESSION_COOKIE, token, max_age=auth.SESSION_DAYS * 86400, httponly=True,
|
|
samesite="lax", secure=is_https(request), path="/")
|
|
|
|
|
|
def throttle(key: str, limit: int, window: int):
|
|
wait = limiter.blocked(key, limit, window)
|
|
if wait:
|
|
raise HTTPException(429, f"Too many attempts. Try again in {wait // 60 + 1} minute(s).",
|
|
headers={"Retry-After": str(wait)})
|
|
|
|
|
|
def check_password_rules(pw: str):
|
|
if not isinstance(pw, str) or len(pw) < auth.MIN_PASSWORD:
|
|
raise HTTPException(422, f"The password needs at least {auth.MIN_PASSWORD} characters")
|
|
if len(pw) > auth.MAX_PASSWORD:
|
|
raise HTTPException(422, "The password is too long")
|
|
|
|
|
|
# ---------------------------------------------------------------- auth routes (public)
|
|
|
|
@app.get("/api/auth/me")
|
|
def me(request: Request):
|
|
with db() as con:
|
|
user = auth.user_for_token(con, request.cookies.get(auth.SESSION_COOKIE))
|
|
return {"user": user, "registration_open": bool(auth.registration_code())}
|
|
|
|
|
|
@app.post("/api/auth/register")
|
|
def register(request: Request, response: Response, body: dict = Body(...)):
|
|
code_cfg = auth.registration_code()
|
|
if not code_cfg:
|
|
raise HTTPException(403, "Registration is disabled on this server")
|
|
ip = client_ip(request)
|
|
# The shared code may be weak, so wrong guesses are throttled per address and for the whole server.
|
|
throttle(f"reg:{ip}", 5, 900)
|
|
throttle("reg:all", 30, 3600)
|
|
username, password, code = str(body.get("username", "")).strip(), body.get("password", ""), str(body.get("code", ""))
|
|
if not secrets.compare_digest(code.encode(), code_cfg.encode()):
|
|
limiter.fail(f"reg:{ip}"); limiter.fail("reg:all")
|
|
raise HTTPException(403, "Wrong registration code")
|
|
if not auth.USERNAME_RE.match(username):
|
|
raise HTTPException(422, "Username: 3-32 characters, letters, digits and . _ - only")
|
|
check_password_rules(password)
|
|
with db() as con:
|
|
if con.execute("SELECT 1 FROM users WHERE username=?", (username,)).fetchone():
|
|
raise HTTPException(409, "This username is taken")
|
|
cur = con.execute("INSERT INTO users (username, pw_hash, created) VALUES (?,?,?)",
|
|
(username, auth.hash_password(password), time.time()))
|
|
token = auth.create_session(con, cur.lastrowid)
|
|
set_cookie(request, response, token)
|
|
return {"user": {"id": cur.lastrowid, "username": username}}
|
|
|
|
|
|
@app.post("/api/auth/login")
|
|
def login(request: Request, response: Response, body: dict = Body(...)):
|
|
username, password = str(body.get("username", "")).strip(), str(body.get("password", ""))[:auth.MAX_PASSWORD]
|
|
ip = client_ip(request)
|
|
keys = (f"login:{ip}", f"login:u:{username.lower()}")
|
|
throttle(keys[0], 10, 600)
|
|
throttle(keys[1], 8, 600)
|
|
with db() as con:
|
|
row = con.execute("SELECT id, username, pw_hash FROM users WHERE username=?", (username,)).fetchone()
|
|
ok = bool(row) and auth.verify_password(password, row["pw_hash"])
|
|
if not row:
|
|
auth.timing_dummy(password)
|
|
if not ok:
|
|
for k in keys:
|
|
limiter.fail(k)
|
|
raise HTTPException(401, "Wrong username or password")
|
|
token = auth.create_session(con, row["id"])
|
|
limiter.reset(keys[1])
|
|
set_cookie(request, response, token)
|
|
return {"user": {"id": row["id"], "username": row["username"]}}
|
|
|
|
|
|
@app.post("/api/auth/logout")
|
|
def logout(request: Request, response: Response):
|
|
with db() as con:
|
|
auth.end_session(con, request.cookies.get(auth.SESSION_COOKIE))
|
|
response.delete_cookie(auth.SESSION_COOKIE, path="/")
|
|
return {"ok": True}
|
|
|
|
|
|
# ---------------------------------------------------------------- everything below needs a login
|
|
|
|
api = APIRouter(prefix="/api", dependencies=[Depends(current_user)])
|
|
|
|
|
|
@api.post("/auth/password")
|
|
def change_password(request: Request, response: Response, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
throttle(f"pw:{user['id']}", 8, 600)
|
|
old, new = str(body.get("old", "")), body.get("new", "")
|
|
check_password_rules(new)
|
|
with db() as con:
|
|
row = con.execute("SELECT pw_hash FROM users WHERE id=?", (user["id"],)).fetchone()
|
|
if not auth.verify_password(old, row["pw_hash"]):
|
|
limiter.fail(f"pw:{user['id']}")
|
|
raise HTTPException(403, "The current password is wrong")
|
|
con.execute("UPDATE users SET pw_hash=? WHERE id=?", (auth.hash_password(new), user["id"]))
|
|
con.execute("DELETE FROM sessions WHERE user_id=?", (user["id"],)) # log out all other devices
|
|
token = auth.create_session(con, user["id"])
|
|
set_cookie(request, response, token)
|
|
return {"ok": True}
|
|
|
|
|
|
@api.get("/rules")
|
|
def get_rules(user: dict = Depends(current_user)):
|
|
return all_rules(user["id"])
|
|
|
|
|
|
# ---- presets
|
|
|
|
def _owned(con, table: str, rid: str, user: dict):
|
|
row = con.execute(f"SELECT owner_id FROM {table} WHERE id=?", (rid,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Not found")
|
|
if row["owner_id"] != user["id"]:
|
|
raise HTTPException(403, "This belongs to another user")
|
|
|
|
|
|
@api.post("/presets")
|
|
def create_preset(body: dict = Body(...), user: dict = Depends(current_user)):
|
|
data = clean_preset(body, load_rules())
|
|
with db() as con:
|
|
if con.execute("SELECT COUNT(*) FROM presets").fetchone()[0] >= MAX_PRESETS:
|
|
raise HTTPException(409, "Too many presets on this server")
|
|
pid, now = "u-" + uuid.uuid4().hex[:8], time.time()
|
|
con.execute("INSERT INTO presets (id, token, data, created, updated, owner_id) VALUES (?,?,?,?,?,?)",
|
|
(pid, "", json.dumps(data), now, now, user["id"]))
|
|
return {"id": pid}
|
|
|
|
|
|
@api.put("/presets/{pid}")
|
|
def update_preset(pid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
data = clean_preset(body, load_rules())
|
|
with db() as con:
|
|
_owned(con, "presets", pid, user)
|
|
con.execute("UPDATE presets SET data=?, updated=? WHERE id=?", (json.dumps(data), time.time(), pid))
|
|
return {"ok": True}
|
|
|
|
|
|
@api.delete("/presets/{pid}")
|
|
def delete_preset(pid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
_owned(con, "presets", pid, user)
|
|
con.execute("DELETE FROM presets WHERE id=?", (pid,))
|
|
return {"ok": True}
|
|
|
|
|
|
# ---- characters (any logged-in user can open a shared link, only the owner can change or delete)
|
|
|
|
def _name(data: str) -> str:
|
|
return json.loads(data).get("name") or "Nameless"
|
|
|
|
|
|
@api.get("/characters")
|
|
def my_characters(user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
rows = con.execute("SELECT id, data, updated FROM characters WHERE owner_id=? ORDER BY updated DESC",
|
|
(user["id"],)).fetchall()
|
|
shares: dict = {}
|
|
for r in con.execute("""SELECT s.character_id, u.username FROM shares s JOIN users u ON u.id = s.user_id
|
|
JOIN characters c ON c.id = s.character_id WHERE c.owner_id=? ORDER BY u.username""",
|
|
(user["id"],)):
|
|
shares.setdefault(r["character_id"], []).append(r["username"])
|
|
shared = con.execute("""SELECT c.id, c.data, c.updated, o.username AS owner FROM shares s
|
|
JOIN characters c ON c.id = s.character_id LEFT JOIN users o ON o.id = c.owner_id
|
|
WHERE s.user_id=? ORDER BY c.updated DESC""", (user["id"],)).fetchall()
|
|
return {"mine": [{"id": r["id"], "name": _name(r["data"]), "updated": r["updated"], "shared_with": shares.get(r["id"], [])}
|
|
for r in rows],
|
|
"shared": [{"id": r["id"], "name": _name(r["data"]), "updated": r["updated"], "owner": r["owner"] or "?"}
|
|
for r in shared]}
|
|
|
|
|
|
@api.post("/characters")
|
|
def create_character(ch: dict = Body(...), user: dict = Depends(current_user)):
|
|
check_size(ch)
|
|
with db() as con:
|
|
if con.execute("SELECT COUNT(*) FROM characters WHERE owner_id=?", (user["id"],)).fetchone()[0] >= MAX_CHARACTERS:
|
|
raise HTTPException(409, "Character limit reached")
|
|
cid, now = uuid.uuid4().hex[:12], time.time()
|
|
con.execute("INSERT INTO characters (id, token, data, created, updated, owner_id) VALUES (?,?,?,?,?,?)",
|
|
(cid, "", json.dumps(ch), now, now, user["id"]))
|
|
return {"id": cid}
|
|
|
|
|
|
@api.get("/characters/{cid}")
|
|
def read_character(cid: str, user: dict = Depends(current_user)):
|
|
"""Only the owner and the users the character is shared with may read it."""
|
|
with db() as con:
|
|
row = con.execute("SELECT data, owner_id FROM characters WHERE id=?", (cid,)).fetchone()
|
|
allowed = bool(row) and (row["owner_id"] == user["id"] or con.execute(
|
|
"SELECT 1 FROM shares WHERE character_id=? AND user_id=?", (cid, user["id"])).fetchone())
|
|
if not allowed:
|
|
raise HTTPException(404, "Character not found")
|
|
return json.loads(row["data"])
|
|
|
|
|
|
@api.post("/characters/{cid}/shares")
|
|
def add_share(cid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
name = str(body.get("username", "")).strip()
|
|
with db() as con:
|
|
_owned(con, "characters", cid, user)
|
|
target = con.execute("SELECT id, username FROM users WHERE username=?", (name,)).fetchone()
|
|
if not target:
|
|
raise HTTPException(404, "No user with this name")
|
|
if target["id"] == user["id"]:
|
|
raise HTTPException(422, "This is your own character")
|
|
if con.execute("SELECT COUNT(*) FROM shares WHERE character_id=?", (cid,)).fetchone()[0] >= MAX_SHARES:
|
|
raise HTTPException(409, "Shared with too many users")
|
|
con.execute("INSERT OR IGNORE INTO shares VALUES (?,?)", (cid, target["id"]))
|
|
return {"username": target["username"]}
|
|
|
|
|
|
@api.delete("/characters/{cid}/shares/{username}")
|
|
def remove_share(cid: str, username: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
_owned(con, "characters", cid, user)
|
|
con.execute("DELETE FROM shares WHERE character_id=? AND user_id=(SELECT id FROM users WHERE username=?)",
|
|
(cid, username))
|
|
return {"ok": True}
|
|
|
|
|
|
@api.put("/characters/{cid}")
|
|
def update_character(cid: str, ch: dict = Body(...), user: dict = Depends(current_user)):
|
|
check_size(ch)
|
|
with db() as con:
|
|
_owned(con, "characters", cid, user)
|
|
con.execute("UPDATE characters SET data=?, updated=? WHERE id=?", (json.dumps(ch), time.time(), cid))
|
|
return {"ok": True}
|
|
|
|
|
|
@api.delete("/characters/{cid}")
|
|
def delete_character(cid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
_owned(con, "characters", cid, user)
|
|
con.execute("DELETE FROM characters WHERE id=?", (cid,))
|
|
con.execute("DELETE FROM shares WHERE character_id=?", (cid,))
|
|
return {"ok": True}
|
|
|
|
|
|
@api.post("/claim/{kind}/{rid}")
|
|
def claim(kind: str, rid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""Takes over something created before accounts existed, using its old secret key."""
|
|
table = {"characters": "characters", "presets": "presets"}.get(kind)
|
|
if not table:
|
|
raise HTTPException(404, "Unknown kind")
|
|
throttle(f"claim:{user['id']}", 60, 600)
|
|
with db() as con:
|
|
row = con.execute(f"SELECT token, owner_id FROM {table} WHERE id=?", (rid,)).fetchone()
|
|
key = str(body.get("token", ""))
|
|
if not row or row["owner_id"] is not None or not row["token"] or not secrets.compare_digest(row["token"], key):
|
|
limiter.fail(f"claim:{user['id']}")
|
|
raise HTTPException(404, "Nothing to claim")
|
|
con.execute(f"UPDATE {table} SET owner_id=?, token='' WHERE id=?", (user["id"], rid))
|
|
return {"ok": True}
|
|
|
|
|
|
# ---- export
|
|
|
|
def _filename(ch: dict, ext: str) -> str:
|
|
base = re.sub(r"[^A-Za-z0-9_-]+", "_", ch.get("name") or "Character").strip("_") or "Character"
|
|
return f"{base}.{ext}"
|
|
|
|
|
|
@api.post("/export/{fmt}")
|
|
def export(fmt: str, ch: dict = Body(...), user: dict = Depends(current_user)):
|
|
check_size(ch)
|
|
rules = all_rules(user["id"])
|
|
if fmt == "docx":
|
|
data = to_docx(ch, rules)
|
|
mime = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
|
elif fmt == "pdf":
|
|
data = to_pdf(ch, rules)
|
|
mime = "application/pdf"
|
|
else:
|
|
raise HTTPException(404, "Unknown format")
|
|
return Response(data, media_type=mime,
|
|
headers={"Content-Disposition": f'attachment; filename="{_filename(ch, fmt)}"'})
|
|
|
|
|
|
app.include_router(api)
|
|
|
|
|
|
# Serve the built frontend with an SPA fallback (the app shell is public, all data is behind the API login)
|
|
if STATIC.is_dir():
|
|
app.mount("/assets", StaticFiles(directory=STATIC / "assets"), name="assets")
|
|
|
|
@app.get("/{path:path}")
|
|
def spa(path: str):
|
|
if path.startswith("api/"):
|
|
raise HTTPException(404, "Not found")
|
|
f = (STATIC / path).resolve()
|
|
if path and f.is_file() and STATIC in f.parents:
|
|
return FileResponse(f)
|
|
return FileResponse(STATIC / "index.html")
|