Claude
69e53c1520
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
730 lines
34 KiB
Python
730 lines
34 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
|
|
MAX_LOBBIES = 20 # per DM
|
|
MAX_MEMBERS = 20 # per lobby
|
|
MAX_MESSAGE = 500 # characters per chat message
|
|
KEEP_MESSAGES = 1000 # per lobby, older ones are dropped
|
|
CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # no 0/O/1/I
|
|
|
|
# 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))""")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS lobbies (
|
|
id TEXT PRIMARY KEY, name TEXT NOT NULL, owner_id INTEGER NOT NULL, code TEXT NOT NULL UNIQUE, created REAL NOT NULL)""")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS lobby_members (
|
|
lobby_id TEXT NOT NULL, user_id INTEGER NOT NULL, character_id TEXT, joined REAL NOT NULL,
|
|
PRIMARY KEY (lobby_id, user_id))""")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS lobby_messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT, lobby_id TEXT NOT NULL, user_id INTEGER, to_user_id INTEGER,
|
|
kind TEXT NOT NULL, text TEXT NOT NULL, created REAL NOT NULL)""")
|
|
con.execute("CREATE INDEX IF NOT EXISTS lobby_messages_lobby ON lobby_messages (lobby_id, 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)):
|
|
"""Readable by the owner, by users the character is shared with, and by the DM of a lobby it plays in."""
|
|
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() or con.execute(
|
|
"""SELECT 1 FROM lobby_members m JOIN lobbies l ON l.id = m.lobby_id
|
|
WHERE m.character_id=? AND l.owner_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,))
|
|
con.execute("UPDATE lobby_members SET character_id=NULL WHERE character_id=?", (cid,))
|
|
return {"ok": True}
|
|
|
|
|
|
# ---- lobbies: the creator is the DM and sees every member's character; players only see a short card of the others
|
|
|
|
def _lobby_for(con, lid: str, user: dict):
|
|
"""Returns the lobby row if the user is its DM or a member, otherwise 404 (hides that it exists)."""
|
|
row = con.execute("""SELECT l.*, u.username AS dm FROM lobbies l JOIN users u ON u.id = l.owner_id WHERE l.id=?""",
|
|
(lid,)).fetchone()
|
|
if not row or (row["owner_id"] != user["id"] and not con.execute(
|
|
"SELECT 1 FROM lobby_members WHERE lobby_id=? AND user_id=?", (lid, user["id"])).fetchone()):
|
|
raise HTTPException(404, "Lobby not found")
|
|
return row
|
|
|
|
|
|
def _dm_only(lobby, user: dict):
|
|
if lobby["owner_id"] != user["id"]:
|
|
raise HTTPException(403, "Only the DM can do this")
|
|
|
|
|
|
def _new_code(con) -> str:
|
|
while True:
|
|
code = "".join(secrets.choice(CODE_ALPHABET) for _ in range(8))
|
|
if not con.execute("SELECT 1 FROM lobbies WHERE code=?", (code,)).fetchone():
|
|
return code
|
|
|
|
|
|
def _own_character(con, cid, user: dict):
|
|
row = con.execute("SELECT id FROM characters WHERE id=? AND owner_id=?", (cid, user["id"])).fetchone()
|
|
if not row:
|
|
raise HTTPException(422, "Pick one of your own characters")
|
|
|
|
|
|
def _event(con, lid: str, text: str, uid=None):
|
|
"""Writes a line into the lobby's event log (shown in the chat feed to everybody)."""
|
|
con.execute("INSERT INTO lobby_messages (lobby_id, user_id, kind, text, created) VALUES (?,?, 'event', ?, ?)",
|
|
(lid, uid, text, time.time()))
|
|
|
|
|
|
def _char_name(con, cid) -> str:
|
|
row = con.execute("SELECT data FROM characters WHERE id=?", (cid,)).fetchone()
|
|
return _name(row["data"]) if row else "a character"
|
|
|
|
|
|
def _card(data: str, rules: dict) -> dict:
|
|
"""The superficial look at a character that fellow players get: name, concept and class/profession names only."""
|
|
ch = json.loads(data)
|
|
cls = next((c for c in rules["classes"] if c["id"] == ch.get("class_id")), None)
|
|
sub = next((x for x in (cls or {}).get("subclasses", []) if x["id"] == ch.get("subclass_id")), None)
|
|
prof = next((p for p in rules["professions"] if p["id"] == ch.get("profession")), None)
|
|
return {"name": ch.get("name") or "Nameless", "concept": str(ch.get("concept") or "")[:120],
|
|
"class": (cls or {}).get("name", ""), "subclass": (sub or {}).get("name", ""),
|
|
"profession": (prof or {}).get("name", "")}
|
|
|
|
|
|
@api.get("/lobbies")
|
|
def my_lobbies(user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
mine = con.execute("""SELECT l.id, l.name, (SELECT COUNT(*) FROM lobby_members m WHERE m.lobby_id = l.id) AS n
|
|
FROM lobbies l WHERE l.owner_id=? ORDER BY l.created DESC""", (user["id"],)).fetchall()
|
|
joined = con.execute("""SELECT l.id, l.name, u.username AS dm, c.data FROM lobby_members m
|
|
JOIN lobbies l ON l.id = m.lobby_id JOIN users u ON u.id = l.owner_id
|
|
LEFT JOIN characters c ON c.id = m.character_id
|
|
WHERE m.user_id=? ORDER BY m.joined DESC""", (user["id"],)).fetchall()
|
|
return {"mine": [{"id": r["id"], "name": r["name"], "members": r["n"]} for r in mine],
|
|
"joined": [{"id": r["id"], "name": r["name"], "dm": r["dm"], "character": _name(r["data"]) if r["data"] else ""}
|
|
for r in joined]}
|
|
|
|
|
|
@api.post("/lobbies")
|
|
def create_lobby(body: dict = Body(...), user: dict = Depends(current_user)):
|
|
name = str(body.get("name", "")).strip()[:60]
|
|
if not name:
|
|
raise HTTPException(422, "Give the lobby a name")
|
|
with db() as con:
|
|
if con.execute("SELECT COUNT(*) FROM lobbies WHERE owner_id=?", (user["id"],)).fetchone()[0] >= MAX_LOBBIES:
|
|
raise HTTPException(409, "Lobby limit reached")
|
|
lid, code = uuid.uuid4().hex[:12], _new_code(con)
|
|
con.execute("INSERT INTO lobbies (id, name, owner_id, code, created) VALUES (?,?,?,?,?)",
|
|
(lid, name, user["id"], code, time.time()))
|
|
_event(con, lid, f"{user['username']} created the lobby", user["id"])
|
|
return {"id": lid, "code": code}
|
|
|
|
|
|
@api.post("/lobbies/join")
|
|
def join_lobby(request: Request, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
ip = client_ip(request)
|
|
throttle(f"join:{ip}", 10, 600) # only wrong codes count, so guessing is slow
|
|
throttle(f"join:u{user['id']}", 10, 600)
|
|
code = re.sub(r"[^A-Za-z0-9]", "", str(body.get("code", ""))).upper()
|
|
with db() as con:
|
|
lobby = con.execute("SELECT * FROM lobbies WHERE code=?", (code,)).fetchone()
|
|
if not lobby:
|
|
limiter.fail(f"join:{ip}"); limiter.fail(f"join:u{user['id']}")
|
|
raise HTTPException(404, "No lobby with this code")
|
|
if lobby["owner_id"] == user["id"]:
|
|
raise HTTPException(422, "You are the DM of this lobby")
|
|
_own_character(con, body.get("character_id"), user)
|
|
known = con.execute("SELECT 1 FROM lobby_members WHERE lobby_id=? AND user_id=?", (lobby["id"], user["id"])).fetchone()
|
|
if not known and con.execute("SELECT COUNT(*) FROM lobby_members WHERE lobby_id=?", (lobby["id"],)).fetchone()[0] >= MAX_MEMBERS:
|
|
raise HTTPException(409, "This lobby is full")
|
|
con.execute("""INSERT INTO lobby_members (lobby_id, user_id, character_id, joined) VALUES (?,?,?,?)
|
|
ON CONFLICT(lobby_id, user_id) DO UPDATE SET character_id=excluded.character_id""",
|
|
(lobby["id"], user["id"], body["character_id"], time.time()))
|
|
_event(con, lobby["id"], f"{user['username']} {'switched to' if known else 'joined as'} {_char_name(con, body['character_id'])}", user["id"])
|
|
return {"id": lobby["id"]}
|
|
|
|
|
|
@api.get("/lobbies/{lid}")
|
|
def read_lobby(lid: str, user: dict = Depends(current_user)):
|
|
rules = load_rules()
|
|
with db() as con:
|
|
lobby = _lobby_for(con, lid, user)
|
|
is_dm = lobby["owner_id"] == user["id"]
|
|
rows = con.execute("""SELECT m.user_id, m.character_id, u.username, c.data FROM lobby_members m
|
|
JOIN users u ON u.id = m.user_id LEFT JOIN characters c ON c.id = m.character_id
|
|
WHERE m.lobby_id=? ORDER BY m.joined""", (lid,)).fetchall()
|
|
members = []
|
|
for r in rows:
|
|
mine = r["user_id"] == user["id"]
|
|
card = _card(r["data"], rules) if r["data"] else None
|
|
# The id is only given out where the viewer may open the full character (the DM sees all, a player only their own)
|
|
members.append({"username": r["username"], "me": mine, "character": card,
|
|
"character_id": r["character_id"] if (is_dm or mine) and card else None})
|
|
return {"id": lobby["id"], "name": lobby["name"], "dm": lobby["dm"], "is_dm": is_dm,
|
|
"code": lobby["code"] if is_dm else None, "members": members}
|
|
|
|
|
|
def _recipient(con, lobby, user: dict, to):
|
|
"""User id to whisper to (the DM or a member, not yourself), None for everybody."""
|
|
to = str(to or "").strip()
|
|
if not to:
|
|
return None
|
|
target = con.execute("""SELECT u.id FROM users u WHERE u.username=? AND (u.id=? OR EXISTS
|
|
(SELECT 1 FROM lobby_members m WHERE m.lobby_id=? AND m.user_id=u.id))""",
|
|
(to, lobby["owner_id"], lobby["id"])).fetchone()
|
|
if not target or target["id"] == user["id"]:
|
|
raise HTTPException(422, "You can only whisper to other people in this lobby")
|
|
return target["id"]
|
|
|
|
|
|
@api.get("/lobbies/{lid}/messages")
|
|
def read_messages(lid: str, after: int = 0, user: dict = Depends(current_user)):
|
|
"""Chat and event log. Whispers are only returned to the sender and the recipient (not even the DM sees them).
|
|
after=0 returns the latest 100 entries, otherwise everything newer than that id (the client polls)."""
|
|
with db() as con:
|
|
_lobby_for(con, lid, user)
|
|
q = """SELECT m.id, m.kind, m.text, m.created, m.user_id, m.to_user_id, f.username AS sender, t.username AS target
|
|
FROM lobby_messages m LEFT JOIN users f ON f.id = m.user_id LEFT JOIN users t ON t.id = m.to_user_id
|
|
WHERE m.lobby_id=? AND (m.to_user_id IS NULL OR m.to_user_id=? OR m.user_id=?)"""
|
|
if after > 0:
|
|
rows = con.execute(q + " AND m.id>? ORDER BY m.id LIMIT 200", (lid, user["id"], user["id"], after)).fetchall()
|
|
else:
|
|
rows = list(reversed(con.execute(q + " ORDER BY m.id DESC LIMIT 100", (lid, user["id"], user["id"])).fetchall()))
|
|
out = []
|
|
for r in rows:
|
|
item = {"id": r["id"], "kind": r["kind"], "text": r["text"], "time": r["created"], "from": r["sender"] or "",
|
|
"to": r["target"] or "", "mine": r["user_id"] == user["id"] and r["kind"] != "event"}
|
|
if r["kind"] == "roll":
|
|
item["roll"], item["text"] = json.loads(r["text"]), ""
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
@api.post("/lobbies/{lid}/messages")
|
|
def send_message(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
text = str(body.get("text", "")).strip()
|
|
if not text:
|
|
raise HTTPException(422, "Write something first")
|
|
if len(text) > MAX_MESSAGE:
|
|
raise HTTPException(422, f"Messages can have at most {MAX_MESSAGE} characters")
|
|
throttle(f"chat:{user['id']}", 30, 60) # 30 messages per minute
|
|
limiter.fail(f"chat:{user['id']}")
|
|
with db() as con:
|
|
lobby = _lobby_for(con, lid, user)
|
|
to_id = _recipient(con, lobby, user, body.get("to"))
|
|
cur = con.execute("INSERT INTO lobby_messages (lobby_id, user_id, to_user_id, kind, text, created) VALUES (?,?,?,?,?,?)",
|
|
(lid, user["id"], to_id, "whisper" if to_id else "chat", text, time.time()))
|
|
con.execute("""DELETE FROM lobby_messages WHERE lobby_id=? AND id <= COALESCE(
|
|
(SELECT id FROM lobby_messages WHERE lobby_id=? ORDER BY id DESC LIMIT 1 OFFSET ?), 0)""", (lid, lid, KEEP_MESSAGES))
|
|
return {"id": cur.lastrowid}
|
|
|
|
|
|
DICE = (4, 6, 8, 10, 12, 20, 100)
|
|
MAX_DICE = 10
|
|
|
|
|
|
@api.post("/lobbies/{lid}/rolls")
|
|
def roll_dice(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""Rolls on the server (so nobody can cheat) and posts the result in the chat; with `to` it is a private roll.
|
|
Rolls by the DM are only visible to the DM, or to the person picked with `to`."""
|
|
try:
|
|
sides, count = int(body.get("sides")), int(body.get("count", 1))
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "Invalid dice")
|
|
if sides not in DICE or not 1 <= count <= MAX_DICE:
|
|
raise HTTPException(422, f"Roll 1 to {MAX_DICE} dice of: " + ", ".join(f"d{d}" for d in DICE))
|
|
throttle(f"chat:{user['id']}", 30, 60)
|
|
limiter.fail(f"chat:{user['id']}")
|
|
results = [secrets.randbelow(sides) + 1 for _ in range(count)]
|
|
with db() as con:
|
|
lobby = _lobby_for(con, lid, user)
|
|
to_id = _recipient(con, lobby, user, body.get("to"))
|
|
if to_id is None and lobby["owner_id"] == user["id"]:
|
|
to_id = user["id"] # the DM's rolls are hidden from the players (unless the DM picks someone to show them to)
|
|
cur = con.execute("INSERT INTO lobby_messages (lobby_id, user_id, to_user_id, kind, text, created) VALUES (?,?,?,?,?,?)",
|
|
(lid, user["id"], to_id, "roll", json.dumps({"sides": sides, "results": results}), time.time()))
|
|
con.execute("""DELETE FROM lobby_messages WHERE lobby_id=? AND id <= COALESCE(
|
|
(SELECT id FROM lobby_messages WHERE lobby_id=? ORDER BY id DESC LIMIT 1 OFFSET ?), 0)""", (lid, lid, KEEP_MESSAGES))
|
|
return {"id": cur.lastrowid, "sides": sides, "results": results}
|
|
|
|
|
|
@api.put("/lobbies/{lid}")
|
|
def rename_lobby(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
name = str(body.get("name", "")).strip()[:60]
|
|
if not name:
|
|
raise HTTPException(422, "Give the lobby a name")
|
|
with db() as con:
|
|
_dm_only(_lobby_for(con, lid, user), user)
|
|
con.execute("UPDATE lobbies SET name=? WHERE id=?", (name, lid))
|
|
_event(con, lid, f"The lobby was renamed to \"{name}\"", user["id"])
|
|
return {"ok": True}
|
|
|
|
|
|
@api.delete("/lobbies/{lid}")
|
|
def delete_lobby(lid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
_dm_only(_lobby_for(con, lid, user), user)
|
|
con.execute("DELETE FROM lobby_members WHERE lobby_id=?", (lid,))
|
|
con.execute("DELETE FROM lobby_messages WHERE lobby_id=?", (lid,))
|
|
con.execute("DELETE FROM lobbies WHERE id=?", (lid,))
|
|
return {"ok": True}
|
|
|
|
|
|
@api.post("/lobbies/{lid}/code")
|
|
def new_lobby_code(lid: str, user: dict = Depends(current_user)):
|
|
"""New join code, the old one stops working (members stay)."""
|
|
with db() as con:
|
|
_dm_only(_lobby_for(con, lid, user), user)
|
|
code = _new_code(con)
|
|
con.execute("UPDATE lobbies SET code=? WHERE id=?", (code, lid))
|
|
_event(con, lid, "The DM created a new join code", user["id"])
|
|
return {"code": code}
|
|
|
|
|
|
@api.put("/lobbies/{lid}/me")
|
|
def switch_lobby_character(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
_lobby_for(con, lid, user)
|
|
_own_character(con, body.get("character_id"), user)
|
|
cur = con.execute("UPDATE lobby_members SET character_id=? WHERE lobby_id=? AND user_id=?",
|
|
(body["character_id"], lid, user["id"]))
|
|
if not cur.rowcount:
|
|
raise HTTPException(422, "You are the DM of this lobby")
|
|
_event(con, lid, f"{user['username']} switched to {_char_name(con, body['character_id'])}", user["id"])
|
|
return {"ok": True}
|
|
|
|
|
|
@api.delete("/lobbies/{lid}/me")
|
|
def leave_lobby(lid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
_lobby_for(con, lid, user)
|
|
gone = con.execute("DELETE FROM lobby_members WHERE lobby_id=? AND user_id=?", (lid, user["id"])).rowcount
|
|
if gone:
|
|
_event(con, lid, f"{user['username']} left the lobby", user["id"])
|
|
return {"ok": True}
|
|
|
|
|
|
@api.delete("/lobbies/{lid}/members/{username}")
|
|
def kick_member(lid: str, username: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
_dm_only(_lobby_for(con, lid, user), user)
|
|
gone = con.execute("DELETE FROM lobby_members WHERE lobby_id=? AND user_id=(SELECT id FROM users WHERE username=?)",
|
|
(lid, username)).rowcount
|
|
if gone:
|
|
_event(con, lid, f"{username} was removed by the DM", user["id"])
|
|
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")
|