ottoandClaude Sonnet 5.5
b31936235d
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
1232 lines
59 KiB
Python
1232 lines
59 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import asyncio
|
|
from fastapi import APIRouter, Body, Depends, FastAPI, HTTPException, Request, Response, WebSocket, WebSocketDisconnect
|
|
from fastapi.responses import FileResponse, JSONResponse, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from . import auth
|
|
from .derive import compute
|
|
from . import rules_edit
|
|
from . import maps as maps_mod
|
|
from . import sounds as sounds_mod
|
|
from . import portrait as portrait_mod
|
|
from . import combat as combat_mod
|
|
from .realtime import hub
|
|
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
|
|
PALETTE = [ # player colours (dice, cards, chat names); every player in a lobby has a different one
|
|
("red", "#e0524d"), ("orange", "#ea8a3b"), ("yellow", "#e6c52e"), ("green", "#6fc24b"), ("teal", "#2fb8a6"),
|
|
("blue", "#4f93e6"), ("purple", "#9d6fda"), ("pink", "#e870b1"), ("white", "#e9e6da"), ("brown", "#a8764f")]
|
|
PALETTE_HEX = dict(PALETTE)
|
|
DM_COLOR = "#d4af37"
|
|
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()
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def _attach_hub():
|
|
hub.attach(asyncio.get_running_loop())
|
|
TRUST_PROXY = os.environ.get("AW_TRUST_PROXY", "") == "1" # behind a reverse proxy that sets X-Forwarded-For / -Proto
|
|
|
|
|
|
|
|
# Changes made in the rules editor live next to the database (writable), the shipped rules.json stays untouched.
|
|
OVERRIDES_PATH = Path(DB_PATH).parent / "rules_overrides.json"
|
|
|
|
|
|
def base_rules() -> dict:
|
|
return json.loads(RULES_PATH.read_text(encoding="utf-8"))
|
|
|
|
|
|
def read_overrides() -> dict:
|
|
try:
|
|
return json.loads(OVERRIDES_PATH.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
def load_rules() -> dict:
|
|
# Re-read on every request so rules.json can be edited without a restart; editor changes replace whole sections.
|
|
r = base_rules()
|
|
for sec, items in read_overrides().items():
|
|
if sec in rules_edit.SECTIONS:
|
|
r[sec] = items
|
|
return r
|
|
|
|
|
|
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, col, ddl in (("lobbies", "diff_mod", "INTEGER NOT NULL DEFAULT 0"), ("lobbies", "diff_label", "TEXT NOT NULL DEFAULT 'Challenging'"),
|
|
("lobbies", "diff_hidden", "INTEGER NOT NULL DEFAULT 0"), ("lobby_members", "color", "TEXT"),
|
|
("lobby_members", "char_mod", "INTEGER NOT NULL DEFAULT 0")):
|
|
if col not in [r["name"] for r in con.execute(f"PRAGMA table_info({table})")]:
|
|
con.execute(f"ALTER TABLE {table} ADD COLUMN {col} {ddl}")
|
|
if "is_admin" not in [r["name"] for r in con.execute("PRAGMA table_info(users)")]:
|
|
con.execute("ALTER TABLE users ADD COLUMN is_admin INTEGER NOT NULL DEFAULT 0")
|
|
if "npc" not in [r["name"] for r in con.execute("PRAGMA table_info(characters)")]:
|
|
con.execute("ALTER TABLE characters ADD COLUMN npc INTEGER NOT NULL DEFAULT 0")
|
|
maps_mod.schema(con)
|
|
sounds_mod.schema(con)
|
|
combat_mod.schema(con)
|
|
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"
|
|
host = request.headers.get("x-forwarded-host", "") if TRUST_PROXY else ""
|
|
host = re.sub(r"[^A-Za-z0-9.:\-\[\]]", "", host or request.headers.get("host", ""))
|
|
resp.headers["Content-Security-Policy"] = ("default-src 'self'; style-src 'self' 'unsafe-inline'; "
|
|
f"img-src 'self' data: blob:; connect-src 'self' ws://{host} wss://{host}; frame-ancestors 'none'")
|
|
if request.url.path.startswith("/api/"):
|
|
resp.headers["Cache-Control"] = "no-store"
|
|
return resp
|
|
|
|
|
|
def admin_ids(con) -> set:
|
|
"""Admins: the names in AW_ADMINS, accounts an admin promoted, and (when there is neither) the very first account."""
|
|
names = {n.strip().lower() for n in os.environ.get("AW_ADMINS", "").split(",") if n.strip()}
|
|
ids = {r["id"] for r in con.execute("SELECT id, username FROM users") if r["username"].lower() in names}
|
|
ids |= {r["id"] for r in con.execute("SELECT id FROM users WHERE is_admin=1")}
|
|
if not ids:
|
|
first = con.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
|
|
if first:
|
|
ids.add(first["id"])
|
|
return ids
|
|
|
|
|
|
def is_admin(user: dict) -> bool:
|
|
with db() as con:
|
|
return user["id"] in admin_ids(con)
|
|
|
|
|
|
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))
|
|
if user:
|
|
user = {**user, "admin": is_admin(user)}
|
|
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)
|
|
u = {"id": cur.lastrowid, "username": username}
|
|
return {"user": {**u, "admin": is_admin(u)}}
|
|
|
|
|
|
@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)
|
|
u = {"id": row["id"], "username": row["username"]}
|
|
return {"user": {**u, "admin": is_admin(u)}}
|
|
|
|
|
|
@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"
|
|
|
|
|
|
def _entry(r, shares=None, owner=None) -> dict:
|
|
ch = json.loads(r["data"])
|
|
e = {"id": r["id"], "name": ch.get("name") or "Nameless", "updated": r["updated"], "portrait": portrait_mod.tag(ch.get("portrait"))}
|
|
if r["npc"]:
|
|
e["kind"] = str(ch.get("npc_kind") or "")
|
|
e["size"] = ch.get("token_size", 1)
|
|
e["hp"] = _health(r["data"])
|
|
if shares is not None:
|
|
e["shared_with"] = shares.get(r["id"], [])
|
|
if owner is not None:
|
|
e["owner"] = owner or "?"
|
|
return e
|
|
|
|
|
|
@api.get("/characters")
|
|
def my_characters(npc: int = 0, user: dict = Depends(current_user)):
|
|
"""Your characters and the ones shared with you. With ?npc=1 the NPC library (NPCs, enemies, creatures) instead."""
|
|
flag = 1 if npc else 0
|
|
with db() as con:
|
|
rows = con.execute("SELECT id, data, updated, npc FROM characters WHERE owner_id=? AND npc=? ORDER BY updated DESC",
|
|
(user["id"], flag)).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=? AND c.npc=? ORDER BY u.username""",
|
|
(user["id"], flag)):
|
|
shares.setdefault(r["character_id"], []).append(r["username"])
|
|
shared = con.execute("""SELECT c.id, c.data, c.updated, c.npc, 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=? AND c.npc=? ORDER BY c.updated DESC""", (user["id"], flag)).fetchall()
|
|
return {"mine": [_entry(r, shares=shares) for r in rows], "shared": [_entry(r, owner=r["owner"]) for r in shared]}
|
|
|
|
|
|
@api.post("/characters")
|
|
def create_character(ch: dict = Body(...), user: dict = Depends(current_user)):
|
|
portrait_mod.clean(ch)
|
|
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, npc) VALUES (?,?,?,?,?,?,?)",
|
|
(cid, "", json.dumps(ch), now, now, user["id"], 1 if ch.get("npc") else 0))
|
|
return {"id": cid}
|
|
|
|
|
|
def _can_read_character(con, cid: str, user: dict) -> bool:
|
|
"""The owner, users the character is shared with, the DM of a lobby it plays in, and the DM of a lobby with a token of this NPC."""
|
|
row = con.execute("SELECT owner_id FROM characters WHERE id=?", (cid,)).fetchone()
|
|
return bool(row) and bool(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() or con.execute(
|
|
"""SELECT 1 FROM lobby_tokens t JOIN lobbies l ON l.id = t.lobby_id
|
|
WHERE t.npc_id=? AND l.owner_id=?""", (cid, user["id"])).fetchone())
|
|
|
|
|
|
@api.get("/characters/{cid}/portrait")
|
|
def character_portrait(cid: str, user: dict = Depends(current_user)):
|
|
with db() as con:
|
|
row = con.execute("SELECT data FROM characters WHERE id=?", (cid,)).fetchone()
|
|
allowed = _can_read_character(con, cid, user)
|
|
img = portrait_mod.image(json.loads(row["data"]).get("portrait")) if row and allowed else None
|
|
if not img:
|
|
raise HTTPException(404, "No portrait")
|
|
return Response(img, media_type="image/webp", headers={"Cache-Control": "private, max-age=86400"})
|
|
|
|
|
|
@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 = _can_read_character(con, cid, user)
|
|
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)):
|
|
portrait_mod.clean(ch)
|
|
check_size(ch)
|
|
with db() as con:
|
|
_owned(con, "characters", cid, user)
|
|
con.execute("UPDATE characters SET data=?, updated=?, npc=? WHERE id=?", (json.dumps(ch), time.time(), 1 if ch.get("npc") else 0, cid))
|
|
lobbies = [r["lobby_id"] for r in con.execute("SELECT lobby_id FROM lobby_members WHERE character_id=?", (cid,))]
|
|
for lid in lobbies:
|
|
hub.notify(lid, "sheet", cid=cid)
|
|
return {"ok": True}
|
|
|
|
|
|
@api.put("/characters/{cid}/notes")
|
|
def update_notes(cid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""Only the two note fields, so typing in the lobby never overwrites other changes made elsewhere."""
|
|
with db() as con:
|
|
_owned(con, "characters", cid, user)
|
|
ch = json.loads(con.execute("SELECT data FROM characters WHERE id=?", (cid,)).fetchone()["data"])
|
|
for k in ("notes", "conditions"):
|
|
if k in body:
|
|
ch[k] = str(body[k] or "")[:5000]
|
|
con.execute("UPDATE characters SET data=?, updated=? WHERE id=?", (json.dumps(ch), time.time(), cid))
|
|
lobbies = [r["lobby_id"] for r in con.execute("SELECT lobby_id FROM lobby_members WHERE character_id=?", (cid,))]
|
|
for lid in lobbies:
|
|
hub.notify(lid, "sheet", cid=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=? AND npc=0", (cid, user["id"])).fetchone()
|
|
if not row:
|
|
raise HTTPException(422, "Pick one of your own characters")
|
|
|
|
|
|
def _free_color(con, lid: str):
|
|
used = {r["color"] for r in con.execute("SELECT color FROM lobby_members WHERE lobby_id=?", (lid,))}
|
|
return next((cid for cid, _ in PALETTE if cid not in used), PALETTE[0][0])
|
|
|
|
|
|
def _ensure_colors(con, lid: str):
|
|
"""Members that joined before colours existed get the next free one."""
|
|
for r in con.execute("SELECT user_id FROM lobby_members WHERE lobby_id=? AND color IS NULL ORDER BY joined", (lid,)).fetchall():
|
|
con.execute("UPDATE lobby_members SET color=? WHERE lobby_id=? AND user_id=?", (_free_color(con, lid), lid, r["user_id"]))
|
|
|
|
|
|
def _levels(rules: dict) -> list:
|
|
"""Difficulty levels from the rules text (section 'difficulty', first table): [{name, mod}]."""
|
|
out = []
|
|
for sec in rules.get("sections", []):
|
|
if sec.get("id") != "difficulty":
|
|
continue
|
|
for block in sec.get("blocks", []):
|
|
for row in (block.get("table") or {}).get("rows", []):
|
|
try:
|
|
out.append({"name": row[0], "mod": int(row[1].replace("\u2212", "-").replace("\u00b1", "").replace("+", "").strip() or 0)})
|
|
except (ValueError, IndexError):
|
|
pass
|
|
return out
|
|
|
|
|
|
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()))
|
|
hub.notify(lid, "messages")
|
|
|
|
|
|
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", ""), "portrait": portrait_mod.tag(ch.get("portrait"))}
|
|
|
|
|
|
@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, color) VALUES (?,?,?,?,?)
|
|
ON CONFLICT(lobby_id, user_id) DO UPDATE SET character_id=excluded.character_id""",
|
|
(lobby["id"], user["id"], body["character_id"], time.time(), _free_color(con, lobby["id"])))
|
|
hub.notify(lobby["id"], "map")
|
|
_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}/portrait/{username}")
|
|
def lobby_portrait(lid: str, username: str, user: dict = Depends(current_user)):
|
|
"""The portrait of a player's character in this lobby (icon on the map and in the player bar). Members only."""
|
|
with db() as con:
|
|
_lobby_for(con, lid, user)
|
|
row = con.execute("""SELECT c.data FROM lobby_members m JOIN users u ON u.id = m.user_id
|
|
JOIN characters c ON c.id = m.character_id WHERE m.lobby_id=? AND u.username=?""", (lid, username)).fetchone()
|
|
img = portrait_mod.image(json.loads(row["data"]).get("portrait")) if row else None
|
|
if not img:
|
|
raise HTTPException(404, "No portrait")
|
|
return Response(img, media_type="image/webp", headers={"Cache-Control": "private, max-age=86400"})
|
|
|
|
|
|
@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"]
|
|
_ensure_colors(con, lid)
|
|
rows = con.execute("""SELECT m.user_id, m.character_id, m.color, m.char_mod, m.vgroup, 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()
|
|
combat = combat_view(con, lobby, is_dm)
|
|
members = []
|
|
online = hub.online(lid)
|
|
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, "color": r["color"],
|
|
"mod": r["char_mod"] if (is_dm or mine) else None, "online": r["user_id"] in online, "vgroup": r["vgroup"] if is_dm else None,
|
|
"character_id": r["character_id"] if (is_dm or mine) and card else None})
|
|
hidden = bool(lobby["diff_hidden"])
|
|
diff = {"label": lobby["diff_label"], "mod": lobby["diff_mod"], "hidden": hidden}
|
|
if hidden and not is_dm:
|
|
diff = {"label": "", "mod": None, "hidden": True}
|
|
return {"id": lobby["id"], "name": lobby["name"], "dm": lobby["dm"], "is_dm": is_dm,
|
|
"dm_online": lobby["owner_id"] in online, "combat": combat, "code": lobby["code"] if is_dm else None, "members": members, "difficulty": diff,
|
|
"sounds": sounds_mod.settings(lobby["sounds"]), "levels": _levels(rules), "palette": [{"id": i, "hex": h} for i, h in PALETTE], "dm_color": DM_COLOR}
|
|
|
|
|
|
@api.put("/lobbies/{lid}/difficulty")
|
|
def set_difficulty(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""The DM's current difficulty: a level from the rules table or a custom modifier. It applies to every check."""
|
|
try:
|
|
mod = int(body.get("mod", 0))
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "The modifier must be a whole number")
|
|
if not -100 <= mod <= 100:
|
|
raise HTTPException(422, "The modifier must be between -100 and 100")
|
|
label = (str(body.get("label") or "").strip() or "Custom")[:24]
|
|
with db() as con:
|
|
_dm_only(_lobby_for(con, lid, user), user)
|
|
con.execute("UPDATE lobbies SET diff_mod=?, diff_label=?, diff_hidden=? WHERE id=?", (mod, label, 1 if body.get("hidden") else 0, lid))
|
|
hub.notify(lid, "lobby")
|
|
return {"ok": True}
|
|
|
|
|
|
@api.put("/lobbies/{lid}/members/{username}/mod")
|
|
def set_member_mod(lid: str, username: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""A personal bonus or malus of one character (talent, condition ...), added on top of the lobby difficulty."""
|
|
try:
|
|
mod = int(body.get("mod", 0))
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "The modifier must be a whole number")
|
|
if not -100 <= mod <= 100:
|
|
raise HTTPException(422, "The modifier must be between -100 and 100")
|
|
with db() as con:
|
|
_dm_only(_lobby_for(con, lid, user), user)
|
|
cur = con.execute("UPDATE lobby_members SET char_mod=? WHERE lobby_id=? AND user_id=(SELECT id FROM users WHERE username=?)",
|
|
(mod, lid, username))
|
|
if not cur.rowcount:
|
|
raise HTTPException(404, "No such player in this lobby")
|
|
hub.notify(lid, "lobby")
|
|
return {"ok": True}
|
|
|
|
|
|
def _recipient(con, lobby, user: dict, to):
|
|
"""User id to whisper to (the DM, a member or yourself for private rolls and notes), 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:
|
|
raise HTTPException(422, "You can only whisper to 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, event log, dice rolls and checks. Whispers are only returned to the sender and the recipient (not even the DM
|
|
sees them). Mystery rolls of the DM come without results, hidden-difficulty checks without the numbers behind them.
|
|
after=0 returns the latest 100 entries, otherwise everything newer than that id (the client polls)."""
|
|
with db() as con:
|
|
lobby = _lobby_for(con, lid, user)
|
|
is_dm = lobby["owner_id"] == user["id"]
|
|
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,
|
|
lm.color AS color
|
|
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
|
|
LEFT JOIN lobby_members lm ON lm.lobby_id = m.lobby_id AND lm.user_id = m.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:
|
|
mine = r["user_id"] == user["id"]
|
|
item = {"id": r["id"], "kind": r["kind"], "text": r["text"], "time": r["created"], "from": r["sender"] or "",
|
|
"to": r["target"] or "", "mine": mine and r["kind"] != "event",
|
|
"color": DM_COLOR if r["user_id"] == lobby["owner_id"] else PALETTE_HEX.get(r["color"], "")}
|
|
if r["kind"] == "roll":
|
|
roll = json.loads(r["text"])
|
|
roll["n"] = len(roll["results"])
|
|
if roll.get("mystery") and not mine:
|
|
roll["results"] = None # players see that the DM rolled, not what
|
|
item["roll"], item["text"] = roll, ""
|
|
elif r["kind"] == "check":
|
|
chk = json.loads(r["text"])
|
|
if chk.get("hidden") and not is_dm:
|
|
for k in ("base", "mod", "cmod", "diff", "target", "margin"):
|
|
chk.pop(k, None)
|
|
item["check"], item["text"] = chk, ""
|
|
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))
|
|
hub.notify(lid, "messages", None if to_id is None else {user["id"], to_id})
|
|
return {"id": cur.lastrowid}
|
|
|
|
|
|
DICE = (4, 6, 8, 10, 12, 20, 100)
|
|
MAX_DICE = 10
|
|
|
|
|
|
def _manual_results(raw, sides: int, count: int):
|
|
"""Results typed in by a player who rolled real dice. None when nothing was typed."""
|
|
if raw in (None, [], ""):
|
|
return None
|
|
try:
|
|
vals = [int(v) for v in raw]
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(422, "Enter whole numbers")
|
|
if len(vals) != count or any(not 1 <= v <= sides for v in vals):
|
|
raise HTTPException(422, f"Enter {count} number{'s' if count > 1 else ''} from 1 to {sides}")
|
|
return vals
|
|
|
|
|
|
def _post(con, lid: str, uid: int, to_id, kind: str, payload: dict) -> int:
|
|
cur = con.execute("INSERT INTO lobby_messages (lobby_id, user_id, to_user_id, kind, text, created) VALUES (?,?,?,?,?,?)",
|
|
(lid, uid, to_id, kind, json.dumps(payload), 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))
|
|
hub.notify(lid, "messages", None if to_id is None else {uid, to_id})
|
|
return cur.lastrowid
|
|
|
|
|
|
@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.
|
|
Players with real dice can send their own results in `manual` (marked as such in the chat).
|
|
The DM picks `mode`: hidden (only the DM sees it), mystery (players see that dice were rolled, not the result) or public."""
|
|
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']}")
|
|
manual = _manual_results(body.get("manual"), sides, count)
|
|
results = manual or [secrets.randbelow(sides) + 1 for _ in range(count)]
|
|
with db() as con:
|
|
lobby = _lobby_for(con, lid, user)
|
|
is_dm = lobby["owner_id"] == user["id"]
|
|
to_id = _recipient(con, lobby, user, body.get("to"))
|
|
mystery = False
|
|
if is_dm and to_id is None:
|
|
mode = body.get("mode") or "hidden"
|
|
if mode not in ("hidden", "mystery", "public"):
|
|
raise HTTPException(422, "Unknown roll mode")
|
|
if mode == "hidden":
|
|
to_id = user["id"] # only the DM sees it
|
|
mystery = mode == "mystery"
|
|
mid = _post(con, lid, user["id"], to_id, "roll", {"sides": sides, "results": results, "mystery": mystery, "manual": bool(manual)})
|
|
return {"id": mid, "sides": sides, "results": results}
|
|
|
|
|
|
@api.post("/lobbies/{lid}/checks")
|
|
def roll_check(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""A d100 roll-under check for a skill or attribute of the player's current lobby character.
|
|
Target = attribute + skill + the DM's difficulty + the character's personal modifier. 01-05 always succeeds, 96-100 always fails."""
|
|
rules = load_rules()
|
|
kind, key = str(body.get("kind", "")), str(body.get("id", ""))
|
|
throttle(f"chat:{user['id']}", 30, 60)
|
|
limiter.fail(f"chat:{user['id']}")
|
|
with db() as con:
|
|
lobby = _lobby_for(con, lid, user)
|
|
member = con.execute("""SELECT m.char_mod, c.data FROM lobby_members m LEFT JOIN characters c ON c.id = m.character_id
|
|
WHERE m.lobby_id=? AND m.user_id=?""", (lid, user["id"])).fetchone()
|
|
npc_name = ""
|
|
if body.get("npc_token"): # the DM rolls for an NPC on the table
|
|
_dm_only(lobby, user)
|
|
t = con.execute("""SELECT t.name, c.data FROM lobby_tokens t JOIN characters c ON c.id = t.npc_id
|
|
WHERE t.id=? AND t.lobby_id=?""", (str(body["npc_token"]), lid)).fetchone()
|
|
if not t:
|
|
raise HTTPException(404, "This token has no NPC sheet")
|
|
member, npc_name = {"char_mod": 0, "data": t["data"]}, t["name"]
|
|
if not member or not member["data"]:
|
|
raise HTTPException(422, "Only players with a character in this lobby can roll checks")
|
|
c = compute(json.loads(member["data"]), rules)
|
|
if kind == "skill":
|
|
s = next((x for x in c["skills"] if x["id"] == key), None)
|
|
if not s:
|
|
raise HTTPException(404, "Unknown skill")
|
|
if not s["usable"]:
|
|
raise HTTPException(422, "This skill is still locked")
|
|
what, base = (f"{npc_name}: " if npc_name else "") + s["name"], s["total"]
|
|
elif kind == "attr":
|
|
a = next((x for x in rules["attributes"] if x["id"] == key), None)
|
|
if not a:
|
|
raise HTTPException(404, "Unknown attribute")
|
|
what, base = (f"{npc_name}: " if npc_name else "") + a["name"], c["attrs"][key]
|
|
else:
|
|
raise HTTPException(422, "Roll a skill or an attribute")
|
|
manual = _manual_results(None if body.get("manual") in (None, "") else [body.get("manual")], 100, 1)
|
|
roll = manual[0] if manual else secrets.randbelow(100) + 1
|
|
target = base + lobby["diff_mod"] + member["char_mod"]
|
|
outcome = "crit_success" if roll <= 5 else "crit_failure" if roll >= 96 else "success" if roll <= target else "failure"
|
|
payload = {"kind": kind, "what": what, "base": base, "mod": lobby["diff_mod"], "cmod": member["char_mod"], "diff": lobby["diff_label"],
|
|
"target": target, "roll": roll, "margin": target - roll, "outcome": outcome, "hidden": bool(lobby["diff_hidden"]),
|
|
"manual": bool(manual)}
|
|
private = str(body.get("to") or "") == user["username"] # a private check is only seen by the roller
|
|
mid = _post(con, lid, user["id"], user["id"] if private else None, "check", payload)
|
|
return {"id": mid, "roll": roll, "outcome": outcome}
|
|
|
|
|
|
@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,))
|
|
for tb in ("lobby_tokens", "lobby_explored", "lobby_reveals", "lobby_scenes", "lobby_templates", "lobby_marks"):
|
|
con.execute(f"DELETE FROM {tb} WHERE lobby_id=?", (lid,))
|
|
con.execute("DELETE FROM lobbies WHERE id=?", (lid,))
|
|
purge_sounds(lid)
|
|
hub.close_lobby(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 update_my_membership(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
|
"""Switch your character and/or pick your colour (no two players share one)."""
|
|
with db() as con:
|
|
_lobby_for(con, lid, user)
|
|
if not con.execute("SELECT 1 FROM lobby_members WHERE lobby_id=? AND user_id=?", (lid, user["id"])).fetchone():
|
|
raise HTTPException(422, "You are the DM of this lobby")
|
|
if "character_id" not in body and "color" not in body:
|
|
raise HTTPException(422, "Nothing to change")
|
|
if "color" in body:
|
|
color = body["color"]
|
|
if color not in PALETTE_HEX:
|
|
raise HTTPException(422, "Unknown colour")
|
|
if con.execute("SELECT 1 FROM lobby_members WHERE lobby_id=? AND color=? AND user_id!=?", (lid, color, user["id"])).fetchone():
|
|
raise HTTPException(409, "Another player already uses this colour")
|
|
con.execute("UPDATE lobby_members SET color=? WHERE lobby_id=? AND user_id=?", (color, lid, user["id"]))
|
|
hub.notify(lid, "lobby")
|
|
if "character_id" in body:
|
|
_own_character(con, body.get("character_id"), user)
|
|
con.execute("UPDATE lobby_members SET character_id=? WHERE lobby_id=? AND user_id=?", (body["character_id"], lid, user["id"]))
|
|
_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"])
|
|
if gone:
|
|
hub.notify(lid, "map")
|
|
hub.close_user(lid, 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"])
|
|
gone_id = con.execute("SELECT id FROM users WHERE username=?", (username,)).fetchone()["id"]
|
|
if gone:
|
|
hub.notify(lid, "map")
|
|
hub.close_user(lid, gone_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)}"'})
|
|
|
|
|
|
# ---- rules editor (admins only)
|
|
|
|
def admin_user(user: dict = Depends(current_user)) -> dict:
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Only admins can edit the rules")
|
|
return user
|
|
|
|
|
|
@api.get("/admin/users")
|
|
def admin_users(user: dict = Depends(admin_user)):
|
|
names = {n.strip().lower() for n in os.environ.get("AW_ADMINS", "").split(",") if n.strip()}
|
|
with db() as con:
|
|
ids = admin_ids(con)
|
|
rows = con.execute("SELECT id, username, created FROM users ORDER BY username COLLATE NOCASE").fetchall()
|
|
return [{"username": r["username"], "admin": r["id"] in ids, "fixed": r["username"].lower() in names, "me": r["id"] == user["id"],
|
|
"created": r["created"]} for r in rows]
|
|
|
|
|
|
@api.put("/admin/users/{username}")
|
|
def set_admin(username: str, body: dict = Body(...), user: dict = Depends(admin_user)):
|
|
"""Make a user an admin or take it away again. Names listed in AW_ADMINS can only be changed in the server settings."""
|
|
want = bool(body.get("admin"))
|
|
names = {n.strip().lower() for n in os.environ.get("AW_ADMINS", "").split(",") if n.strip()}
|
|
with db() as con:
|
|
target = con.execute("SELECT id, username FROM users WHERE username=?", (username,)).fetchone()
|
|
if not target:
|
|
raise HTTPException(404, "No user with this name")
|
|
ids = admin_ids(con)
|
|
if want:
|
|
if not any(r["is_admin"] for r in con.execute("SELECT is_admin FROM users")) and not names: # the implicit first-account admin becomes explicit
|
|
con.execute("UPDATE users SET is_admin=1 WHERE id IN (%s)" % ",".join(str(i) for i in ids))
|
|
con.execute("UPDATE users SET is_admin=1 WHERE id=?", (target["id"],))
|
|
else:
|
|
if target["username"].lower() in names:
|
|
raise HTTPException(409, "This admin is set in AW_ADMINS on the server and can only be removed there")
|
|
if ids - {target["id"]} == set():
|
|
raise HTTPException(409, "There must be at least one admin")
|
|
if target["id"] not in ids:
|
|
return {"ok": True}
|
|
if not any(r["is_admin"] for r in con.execute("SELECT is_admin FROM users")): # implicit first account: make it explicit first
|
|
con.execute("UPDATE users SET is_admin=1 WHERE id IN (%s)" % ",".join(str(i) for i in ids))
|
|
con.execute("UPDATE users SET is_admin=0 WHERE id=?", (target["id"],))
|
|
return {"ok": True}
|
|
|
|
|
|
@api.get("/admin/rules")
|
|
def editor_state(user: dict = Depends(admin_user)):
|
|
"""The editable sections as they are now, and which of them differ from the shipped rules.json."""
|
|
r = load_rules()
|
|
return {"sections": {k: r[k] for k in rules_edit.SECTIONS}, "customized": [k for k in read_overrides() if k in rules_edit.SECTIONS]}
|
|
|
|
|
|
def _all_characters(con) -> list:
|
|
return [json.loads(x["data"]) for x in con.execute("SELECT data FROM characters")]
|
|
|
|
|
|
def _save_overrides(data: dict):
|
|
OVERRIDES_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
if OVERRIDES_PATH.exists():
|
|
OVERRIDES_PATH.with_suffix(".json.bak").write_bytes(OVERRIDES_PATH.read_bytes())
|
|
tmp = OVERRIDES_PATH.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
|
|
os.replace(tmp, OVERRIDES_PATH)
|
|
|
|
|
|
def _apply(section: str, new: list, confirm: bool):
|
|
r = load_rules()
|
|
old_ids = {x["id"] for x in r[section]}
|
|
removed = old_ids - {x["id"] for x in new}
|
|
bad = rules_edit.dangling(section, new, r)
|
|
if bad:
|
|
raise HTTPException(422, "Cannot save: " + "; ".join(bad[:5]) + (" ..." if len(bad) > 5 else ""))
|
|
with db() as con:
|
|
used = rules_edit.usage(_all_characters(con), section, removed)
|
|
if used and not confirm:
|
|
raise HTTPException(409, f"{used} saved character{'s' if used != 1 else ''} use{'' if used != 1 else 's'} entries you are removing. "
|
|
"Their data stays in the characters but the removed entries disappear from the sheets.")
|
|
return used
|
|
|
|
|
|
@api.put("/admin/rules/{section}")
|
|
def editor_save(section: str, body: dict = Body(...), user: dict = Depends(admin_user)):
|
|
try:
|
|
new = rules_edit.clean(section, body.get("items"), load_rules())
|
|
except rules_edit.RuleError as e:
|
|
raise HTTPException(422, str(e))
|
|
_apply(section, new, bool(body.get("confirm")))
|
|
data = read_overrides()
|
|
data[section] = new
|
|
_save_overrides(data)
|
|
return {"ok": True, "count": len(new)}
|
|
|
|
|
|
@api.delete("/admin/rules/{section}")
|
|
def editor_reset(section: str, confirm: bool = False, user: dict = Depends(admin_user)):
|
|
"""Back to the shipped rules.json for this section."""
|
|
if section not in rules_edit.SECTIONS:
|
|
raise HTTPException(404, "Unknown section")
|
|
data = read_overrides()
|
|
if section in data:
|
|
_apply(section, base_rules()[section], confirm)
|
|
del data[section]
|
|
_save_overrides(data)
|
|
return {"ok": True}
|
|
|
|
|
|
def _health(data) -> int | None:
|
|
try:
|
|
return int(compute(json.loads(data), load_rules())["derived"].get("health"))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
maps_mod.install(api, db, current_user, _lobby_for, _dm_only, hub, _event, Path(DB_PATH).parent / "maps", _can_read_character, _health)
|
|
combat_view = combat_mod.install(api, db, current_user, _lobby_for, _dm_only, hub, _event, load_rules)
|
|
purge_sounds = sounds_mod.install(api, db, current_user, _lobby_for, _dm_only, hub, Path(DB_PATH).parent / "sounds")
|
|
|
|
|
|
@app.websocket("/api/lobbies/{lid}/ws")
|
|
async def lobby_socket(websocket: WebSocket, lid: str):
|
|
"""One socket per open lobby page. Only notes go out ("messages", "lobby", "sheet"), the data is fetched through the API."""
|
|
origin = websocket.headers.get("origin")
|
|
hosts = {websocket.headers.get("host", "")}
|
|
if TRUST_PROXY:
|
|
hosts.add(websocket.headers.get("x-forwarded-host", ""))
|
|
if origin and origin.split("://", 1)[-1] not in hosts:
|
|
await websocket.close(code=4403)
|
|
return
|
|
|
|
def check():
|
|
with db() as con:
|
|
user = auth.user_for_token(con, websocket.cookies.get(auth.SESSION_COOKIE))
|
|
if not user:
|
|
return None
|
|
lobby = con.execute("SELECT owner_id FROM lobbies WHERE id=?", (lid,)).fetchone()
|
|
ok = lobby and (lobby["owner_id"] == user["id"] or con.execute(
|
|
"SELECT 1 FROM lobby_members WHERE lobby_id=? AND user_id=?", (lid, user["id"])).fetchone())
|
|
return user if ok else None
|
|
|
|
user = await asyncio.to_thread(check)
|
|
if not user:
|
|
await websocket.close(code=4401)
|
|
return
|
|
await websocket.accept()
|
|
was_online = user["id"] in hub.online(lid)
|
|
if not hub.add(lid, websocket, user["id"]):
|
|
await websocket.close(code=4429)
|
|
return
|
|
if not was_online:
|
|
hub.notify(lid, "lobby") # presence changed
|
|
try:
|
|
last_check = time.time()
|
|
while True:
|
|
try:
|
|
text = await asyncio.wait_for(websocket.receive_text(), timeout=70)
|
|
except asyncio.TimeoutError:
|
|
break # the client pings every 25 seconds, silence means it is gone
|
|
if text == "ping":
|
|
await websocket.send_text("pong")
|
|
if time.time() - last_check > 300: # session or membership may have ended
|
|
last_check = time.time()
|
|
if not await asyncio.to_thread(check):
|
|
break
|
|
except WebSocketDisconnect:
|
|
pass
|
|
finally:
|
|
hub.remove(lid, websocket)
|
|
if user["id"] not in hub.online(lid):
|
|
hub.notify(lid, "lobby")
|
|
try:
|
|
await websocket.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
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")
|