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")