From 3330801de4c6b7a756832195c11a2554e1b2d3ab Mon Sep 17 00:00:00 2001 From: Ryan Barlow <7389646+ryanbarlow97@users.noreply.github.com> Date: Tue, 6 Oct 2026 09:36:20 +0000 Subject: [PATCH 1/6] Add staff player directory, sessions and activity from CoreProtect Staff (mod and above, new view_players capability) can list every Minecraft player, open a profile, and page through sessions and recent CoreProtect actions. Players are keyed by UUID and merged from CoreProtect's co_user, discord_links and character_roster, so linked players and characters still appear when CoreProtect is unavailable. The live CoreProtect SQLite file is read with guards that keep its writer unblocked: read-only autocommit, one reader per process, a 1.5 s budget per request, statements fetched and closed one at a time, and at most 2,000 rows examined per table per page. Chat, command arguments, sign text and item metadata are never selected. backend/benchmarks/coreprotect_reader.py is the deployment gate; on Main it measured p99 8.6 ms per request. Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/benchmarks/coreprotect_reader.py | 215 ++++++++++++ backend/conftest.py | 17 + backend/src/api/admin_routes.py | 46 ++- backend/src/auth/players.py | 355 ++++++++++++++++++++ backend/src/auth/roles.py | 1 + backend/src/auth/test_admin.py | 4 +- backend/src/auth/test_players.py | 191 +++++++++++ backend/src/coreprotect/README.md | 74 ++++ backend/src/coreprotect/__init__.py | 1 + backend/src/coreprotect/activity.py | 243 ++++++++++++++ backend/src/coreprotect/cursors.py | 31 ++ backend/src/coreprotect/maps.py | 69 ++++ backend/src/coreprotect/reader.py | 167 +++++++++ backend/src/coreprotect/sessions.py | 150 +++++++++ backend/src/coreprotect/test_activity.py | 157 +++++++++ backend/src/coreprotect/test_reader.py | 115 +++++++ backend/src/coreprotect/test_sessions.py | 132 ++++++++ backend/src/coreprotect/testdata/schema.sql | 75 +++++ backend/src/coreprotect/testing.py | 72 ++++ 19 files changed, 2111 insertions(+), 4 deletions(-) create mode 100644 backend/benchmarks/coreprotect_reader.py create mode 100644 backend/src/auth/players.py create mode 100644 backend/src/auth/test_players.py create mode 100644 backend/src/coreprotect/README.md create mode 100644 backend/src/coreprotect/__init__.py create mode 100644 backend/src/coreprotect/activity.py create mode 100644 backend/src/coreprotect/cursors.py create mode 100644 backend/src/coreprotect/maps.py create mode 100644 backend/src/coreprotect/reader.py create mode 100644 backend/src/coreprotect/sessions.py create mode 100644 backend/src/coreprotect/test_activity.py create mode 100644 backend/src/coreprotect/test_reader.py create mode 100644 backend/src/coreprotect/test_sessions.py create mode 100644 backend/src/coreprotect/testdata/schema.sql create mode 100644 backend/src/coreprotect/testing.py diff --git a/backend/benchmarks/coreprotect_reader.py b/backend/benchmarks/coreprotect_reader.py new file mode 100644 index 00000000..136260d1 --- /dev/null +++ b/backend/benchmarks/coreprotect_reader.py @@ -0,0 +1,215 @@ +"""Deployment gate for the staff panel's CoreProtect reads. + +hold + Times every player's profile, sessions and activity reads (default and + sparse kind filters) and the directory build, through the real reader. + On a live server database pass --no-lock: it opens the file immutable, + so CoreProtect is never locked, and an occasional torn-page error is + counted rather than fatal. The time each request spends in CoreProtect + is how long a production request would hold its read lock. + +contend + Runs a CoreProtect-like writer (batched inserts, small cache so pages + spill mid-transaction, a 3 s busy timeout like sqlite-jdbc) in another + process, first alone and then while readers make back-to-back staff + requests. Reports writer transaction latency, throughput and failures. + Never point this at a live database: it writes. + +Acceptance (agreed before running): hold p99 <= 100 ms and max <= 500 ms per +request; contend shows no writer failures, and writer transaction p99 and +max grow by no more than 100 ms and 500 ms respectively with readers on. +""" +from __future__ import annotations + +import argparse +import multiprocessing as mp +import os +import sqlite3 +import statistics +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from src.coreprotect import activity, maps, sessions # noqa: E402 +from src.coreprotect.reader import Budget, CoreProtectConfig, Reader, Unavailable # noqa: E402 + + +def _config(path: str) -> CoreProtectConfig: + return CoreProtectConfig(path=path, server="bench", label="", ping_seconds=60) + + +def _no_lock() -> None: + """Open the reader's connections immutable (benchmark only: never in the site).""" + original = sqlite3.connect + + def connect(target, *args, **kwargs): + if isinstance(target, str) and target.endswith("?mode=ro"): + target += "&immutable=1" + return original(target, *args, **kwargs) + + sqlite3.connect = connect + + +def _ids(config): + with Reader(config, Budget(30)) as r: + rows = r.rows("SELECT id FROM co_user WHERE uuid IS NOT NULL") + return [row["id"] for row in rows] + + +def _requests(config, uid): + """The CoreProtect part of each staff request for one player.""" + def profile(r): + r.rows("SELECT id, user, time FROM co_user WHERE id = ?", (uid,)) + sessions.last_event(r, [uid]) + r.rows("SELECT MIN(time) AS first FROM co_session WHERE user = ?", (uid,)) + + def session_page(r): + sessions.fetch(r, [uid], None, 20) + + def feed(kinds): + def run(r): + activity.fetch(r, maps.get(r), [uid], kinds, None, 20) + return run + + return [("profile", profile), ("sessions", session_page), ("activity", feed(activity.KINDS)), + ("activity:skill", feed(("skill",))), ("activity:sign", feed(("sign",))), + ("activity:kill", feed(("kill",)))] + + +def _timed(config, work) -> tuple[float, str | None]: + started = time.perf_counter() + try: + with Reader(config, Budget()) as r: + work(r) + return time.perf_counter() - started, None + except Unavailable as exc: + return time.perf_counter() - started, exc.code + + +def _summary(label, samples): + if not samples: + return f"{label}: no samples" + ordered = sorted(samples) + p99 = ordered[min(len(ordered) - 1, int(len(ordered) * 0.99))] + return (f"{label}: n={len(ordered)} p50={statistics.median(ordered) * 1000:.1f}ms " + f"p99={p99 * 1000:.1f}ms max={ordered[-1] * 1000:.1f}ms") + + +def hold(args) -> None: + if args.no_lock: + _no_lock() + config = _config(args.database) + ids = _ids(config)[: args.players or None] + by_kind: dict[str, list[float]] = {} + errors: dict[str, int] = {} + for uid in ids: + for name, work in _requests(config, uid): + seconds, error = _timed(config, work) + by_kind.setdefault(name, []).append(seconds) + if error: + errors[f"{name}:{error}"] = errors.get(f"{name}:{error}", 0) + 1 + + def directory(r): + rows = r.rows("SELECT id FROM co_user WHERE uuid IS NOT NULL") + r.rows("SELECT uuid, user FROM co_username_log") + for row in rows: + sessions.last_event(r, [row["id"]]) + + for _ in range(3): + seconds, error = _timed(config, directory) + by_kind.setdefault("directory", []).append(seconds) + if error: + errors[f"directory:{error}"] = errors.get(f"directory:{error}", 0) + 1 + print(f"players: {len(ids)}") + for name, samples in by_kind.items(): + print(_summary(name, samples)) + print(_summary("all requests", [s for samples in by_kind.values() for s in samples])) + print(f"errors: {errors or 'none'}") + + +def _writer(path, seconds, batch, out): + conn = sqlite3.connect(path, timeout=3, isolation_level=None) + conn.execute("PRAGMA cache_size = 64") + user = conn.execute("SELECT MIN(id) FROM co_user").fetchone()[0] or 1 + latencies, failures, rows = [], 0, 0 + stop = time.monotonic() + seconds + t = int(time.time()) + while time.monotonic() < stop: + started = time.perf_counter() + try: + conn.execute("BEGIN") + conn.executemany( + "INSERT INTO co_block (time, user, wid, x, y, z, type, data, meta, blockdata, action, rolled_back) " + "VALUES (?, ?, 1, ?, 64, ?, 1, 0, NULL, NULL, 0, 0)", + [(t, user, i, i) for i in range(batch)]) + conn.executemany("INSERT INTO co_session (time, user, wid, x, y, z, action) VALUES (?, ?, 1, 0, 64, 0, 2)", + [(t, user)] * 5) + conn.execute("COMMIT") + rows += batch + latencies.append(time.perf_counter() - started) + except sqlite3.OperationalError: + failures += 1 + try: + conn.execute("ROLLBACK") + except sqlite3.OperationalError: + pass + t += 1 + time.sleep(0.05) + conn.close() + out.put((latencies, failures, rows)) + + +def _hammer(path, seconds, out): + config = _config(path) + ids = _ids(config) + requests = [work for uid in ids for _, work in _requests(config, uid)] + done, errors, i = 0, {}, 0 + stop = time.monotonic() + seconds + while time.monotonic() < stop: + _, error = _timed(config, requests[i % len(requests)]) + i += 1 + done += 1 + if error: + errors[error] = errors.get(error, 0) + 1 + out.put((os.getpid(), done, errors)) + + +def contend(args) -> None: + for phase, readers in (("writer alone", 0), ("writer + readers", args.readers)): + out, reads = mp.Queue(), mp.Queue() + writer = mp.Process(target=_writer, args=(args.database, args.seconds, args.batch, out)) + hammers = [mp.Process(target=_hammer, args=(args.database, args.seconds, reads)) for _ in range(readers)] + writer.start() + for h in hammers: + h.start() + latencies, failures, rows = out.get() + writer.join() + reader_results = [reads.get() for _ in hammers] + for h in hammers: + h.join() + print(_summary(f"{phase}: write txn", latencies)) + print(f"{phase}: failures={failures} rows/s={rows / args.seconds:.0f}") + for pid, done, errors in reader_results: + print(f"{phase}: reader {pid} requests={done} errors={errors or 'none'}") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + sub = parser.add_subparsers(dest="command", required=True) + h = sub.add_parser("hold") + h.add_argument("database") + h.add_argument("--no-lock", action="store_true") + h.add_argument("--players", type=int, default=0) + c = sub.add_parser("contend") + c.add_argument("database") + c.add_argument("--seconds", type=float, default=20) + c.add_argument("--batch", type=int, default=500) + c.add_argument("--readers", type=int, default=1) + args = parser.parse_args() + hold(args) if args.command == "hold" else contend(args) + + +if __name__ == "__main__": + main() diff --git a/backend/conftest.py b/backend/conftest.py index 788d390e..e53d3dd5 100644 --- a/backend/conftest.py +++ b/backend/conftest.py @@ -37,3 +37,20 @@ def tick(): yield finally: asyncio.DefaultEventLoopPolicy.new_event_loop = original + + +@pytest.fixture +def coreprotect(tmp_path, monkeypatch): + """A CoreProtect database the backend reads through COREPROTECT_DB.""" + from src.coreprotect import maps + from src.coreprotect.testing import CoreProtectDb + + (tmp_path / "coreprotect").mkdir() + db = CoreProtectDb(tmp_path / "coreprotect" / "database.db") + monkeypatch.setenv("COREPROTECT_DB", str(db.path)) + monkeypatch.setenv("COREPROTECT_SERVER", "main") + monkeypatch.setenv("COREPROTECT_SERVER_LABEL", "Vardera") + monkeypatch.setenv("COREPROTECT_PING_SECONDS", "60") + maps.clear() + yield db + maps.clear() diff --git a/backend/src/api/admin_routes.py b/backend/src/api/admin_routes.py index 8cdfd9fa..6e0223e0 100644 --- a/backend/src/api/admin_routes.py +++ b/backend/src/api/admin_routes.py @@ -8,14 +8,15 @@ import re -from fastapi import APIRouter, FastAPI, HTTPException, Request, Response +from fastapi import APIRouter, FastAPI, HTTPException, Query, Request, Response from fastapi.exception_handlers import request_validation_exception_handler from fastapi.exceptions import RequestValidationError from pydantic import BaseModel, Field from starlette.concurrency import run_in_threadpool from src.api.auth_routes import _config, _no_store, current_user, require_same_origin, session_cookie -from src.auth import admin, audit, roles, users +from src.auth import admin, audit, players, roles, users +from src.coreprotect.reader import CoreProtectConfig admin_router = APIRouter(prefix="/admin", tags=["admin"]) @@ -36,6 +37,22 @@ def _staff(request: Request) -> dict: return user +def _capable(request: Request, capability: str) -> dict: + user = _staff(request) + if not roles.can(user["role"], capability): + raise HTTPException(403, detail="forbidden") + return user + + +def _players(request: Request, response: Response, run): + _capable(request, "view_players") + _no_store(response) + try: + return run(CoreProtectConfig.from_env()) + except players.PlayerError as exc: + raise HTTPException(exc.status, detail=exc.code) from None + + def _account_json(row: dict) -> dict: return { "user_id": row["user_id"], @@ -94,6 +111,31 @@ def lookup_accounts(request: Request, response: Response, q: str = ""): return {"accounts": [_account_json(row) for row in admin.lookup(q)]} +@admin_router.get("/players") +def get_players(request: Request, response: Response, q: str = "", sort: str = "last_seen", + page: int = Query(1, ge=1, le=10_000)): + return _players(request, response, lambda config: players.directory(config, q, sort, page)) + + +@admin_router.get("/players/{player_uuid}") +def get_player(player_uuid: str, request: Request, response: Response): + return _players(request, response, lambda config: players.profile(config, player_uuid)) + + +@admin_router.get("/players/{player_uuid}/sessions") +def get_player_sessions(player_uuid: str, request: Request, response: Response, before: str | None = None, + limit: int = Query(20, ge=1, le=100)): + return _players(request, response, + lambda config: players.player_sessions(config, player_uuid, before, limit)) + + +@admin_router.get("/players/{player_uuid}/activity") +def get_player_activity(player_uuid: str, request: Request, response: Response, before: str | None = None, + limit: int = Query(20, ge=1, le=100), kinds: str | None = None): + return _players(request, response, + lambda config: players.player_activity(config, player_uuid, before, limit, kinds)) + + @admin_router.post("/accounts/{user_id}/role") def post_role(user_id: int, body: RoleBody, request: Request, response: Response): _no_store(response) diff --git a/backend/src/auth/players.py b/backend/src/auth/players.py new file mode 100644 index 00000000..62d76229 --- /dev/null +++ b/backend/src/auth/players.py @@ -0,0 +1,355 @@ +"""Staff view of Minecraft players: a directory, profiles, sessions and activity. + +A player is a Minecraft account, keyed by its UUID; names are never merged +on. The directory is the union of everyone CoreProtect has seen join, every +Discord link and every character, so it still lists linked players and +characters when CoreProtect cannot be read. Each section reports whether its +source was available, so "unavailable" never looks like "no history". + +Discord links can change, so nothing here grants account actions; those stay +with the account routes, which check the target afresh. +""" +from __future__ import annotations + +import threading +import time +import uuid as uuidlib + +from src.coreprotect import activity, maps as co_maps, sessions +from src.coreprotect.cursors import BadCursor, decode as decode_cursor, encode as encode_cursor +from src.coreprotect.reader import REQUEST_BUDGET_SECONDS, Budget, CoreProtectConfig, Reader, Unavailable +from src.skins.db import connect + +from . import users + +DIRECTORY_TTL_SECONDS = 60 +PAGE_SIZE = 50 +QUERY_MAX = 64 +SORTS = ("last_seen", "minecraft", "discord") + +_DIRECTORY_LOCK = threading.Lock() +_DIRECTORY: dict[tuple[str, str], tuple[float, list[dict], dict]] = {} + + +class PlayerError(Exception): + def __init__(self, status: int, code: str): + super().__init__(code) + self.status = status + self.code = code + + +def canonical_uuid(text: str | None) -> str | None: + try: + return str(uuidlib.UUID(str(text or "").strip())) + except ValueError: + return None + + +def _status(error: Unavailable | None) -> dict: + return {"status": "unavailable", "reason": error.code} if error else {"status": "available"} + + +# -------------------- +# province.db +# -------------------- + +_LINKS = """ + SELECT dl.player_uuid, dl.discord_user_id, dl.discord_username, dl.minecraft_name, + dl.linked_at, dl.left_guild_at, dl.grace_until, + u.id AS user_id, u.role, u.discord_username AS account_username, + u.discord_global_name, u.discord_avatar, u.created_at AS account_created_at, u.last_login_at + FROM discord_links dl LEFT JOIN users u ON u.discord_user_id = dl.discord_user_id +""" +_ROSTER = """ + SELECT player_uuid, realm_id, character_id, name, status, race, class, updated_at + FROM character_roster +""" + + +def _site_rows(key: str | None = None) -> tuple[list[dict], list[dict]]: + """Discord links (with any website account) and characters, for one player or everyone.""" + with connect() as conn: + if key is None: + links = conn.execute(_LINKS).fetchall() + roster = conn.execute(f"{_ROSTER} ORDER BY realm_id, lower(name)").fetchall() + else: + links = conn.execute(f"{_LINKS} WHERE lower(dl.player_uuid) = ?", (key,)).fetchall() + roster = conn.execute(f"{_ROSTER} WHERE lower(player_uuid) = ? ORDER BY realm_id, lower(name)", + (key,)).fetchall() + return [dict(r) for r in links], [dict(r) for r in roster] + + +def _discord(link: dict | None) -> dict | None: + if not link: + return None + return { + "discord_user_id": link["discord_user_id"], + "discord_username": link["account_username"] or link["discord_username"], + "linked_at": link["linked_at"], + "left_guild_at": link["left_guild_at"], + "grace_until": link["grace_until"], + } + + +def _account(link: dict | None) -> dict | None: + if not link or link.get("user_id") is None: + return None + return { + "user_id": link["user_id"], + "role": link["role"], + "discord_global_name": link["discord_global_name"], + "avatar_url": users.avatar_url(link), + "created_at": link["account_created_at"], + "last_login_at": link["last_login_at"], + } + + +# -------------------- +# Directory +# -------------------- + +def _read_directory(config: CoreProtectConfig) -> tuple[dict[str, dict], Unavailable | None]: + """CoreProtect's players, keyed by UUID. A whole-table read of co_user and co_username_log + (a few hundred rows each), then one indexed seek per player for the last session row.""" + now = int(time.time()) + try: + with Reader(config, Budget()) as reader: + accounts = reader.rows("SELECT id, user, uuid, time FROM co_user WHERE uuid IS NOT NULL") + history = reader.rows("SELECT uuid, user FROM co_username_log") + found: dict[str, dict] = {} + for row in accounts: + key = canonical_uuid(row["uuid"]) + if key is None: + continue + entry = found.setdefault(key, {"ids": [], "minecraft_name": row["user"], "names": set()}) + entry["ids"].append(row["id"]) + entry["names"].add(row["user"]) + for row in history: + key = canonical_uuid(row["uuid"]) + if key in found and row["user"]: + found[key]["names"].add(row["user"]) + for entry in found.values(): + event = sessions.last_event(reader, entry["ids"]) + entry["last_seen"] = event["time"] if event else None + entry["online"] = sessions.is_fresh(event, now, config.ping_seconds) + return found, None + except Unavailable as exc: + return {}, exc + + +def _build_directory(config: CoreProtectConfig) -> tuple[list[dict], dict]: + played, error = _read_directory(config) + links, roster = _site_rows() + people: dict[str, dict] = {} + + def person(key: str) -> dict: + return people.setdefault(key, { + "uuid": key, "minecraft_name": None, "names": set(), "discord_user_id": None, + "discord_username": None, "discord_global_name": None, "site_role": None, + "characters": [], "last_seen": None, "online": False, + }) + + for key, seen in played.items(): + entry = person(key) + entry.update(minecraft_name=seen["minecraft_name"], last_seen=seen["last_seen"], online=seen["online"]) + entry["names"] |= seen["names"] + for link in links: + key = canonical_uuid(link["player_uuid"]) + if key is None: + continue + entry = person(key) + entry["minecraft_name"] = entry["minecraft_name"] or link["minecraft_name"] + if link["minecraft_name"]: + entry["names"].add(link["minecraft_name"]) + entry.update( + discord_user_id=link["discord_user_id"], + discord_username=link["account_username"] or link["discord_username"], + discord_global_name=link["discord_global_name"], + site_role=link["role"], + ) + for character in roster: + key = canonical_uuid(character["player_uuid"]) + if key is not None and character["name"]: + person(key)["characters"].append(character["name"]) + return list(people.values()), _status(error) + + +def _directory(config: CoreProtectConfig) -> tuple[list[dict], dict]: + key = (config.server, config.path or "") + cached = _DIRECTORY.get(key) + if cached and cached[0] > time.monotonic(): + return cached[1], cached[2] + # One build at a time; it holds no CoreProtect connection while others wait. + if not _DIRECTORY_LOCK.acquire(timeout=REQUEST_BUDGET_SECONDS * 2): + raise PlayerError(503, "directory_busy") + try: + cached = _DIRECTORY.get(key) + if cached and cached[0] > time.monotonic(): + return cached[1], cached[2] + entries, status = _build_directory(config) + # A degraded build is not cached, so the next request tries CoreProtect again. + if status["status"] == "available" or not config.path: + _DIRECTORY[key] = (time.monotonic() + DIRECTORY_TTL_SECONDS, entries, status) + return entries, status + finally: + _DIRECTORY_LOCK.release() + + +def _matches(entry: dict, needle: str) -> bool: + if needle in {entry["uuid"], entry["uuid"].replace("-", ""), (entry["discord_user_id"] or "")}: + return True + haystack = [*entry["names"], entry["discord_username"], entry["discord_global_name"], *entry["characters"]] + return any(needle in value.casefold() for value in haystack if value) + + +def _sort_key(sort: str): + if sort == "minecraft": + return lambda e: (e["minecraft_name"] is None, (e["minecraft_name"] or "").casefold(), e["uuid"]) + if sort == "discord": + return lambda e: (e["discord_username"] is None, (e["discord_username"] or "").casefold(), e["uuid"]) + return lambda e: (e["last_seen"] is None, -(e["last_seen"] or 0), (e["minecraft_name"] or "").casefold()) + + +def _directory_json(entry: dict) -> dict: + return { + "uuid": entry["uuid"], + "minecraft_name": entry["minecraft_name"], + "discord_user_id": entry["discord_user_id"], + "discord_username": entry["discord_username"], + "site_role": entry["site_role"], + "characters": entry["characters"], + "last_seen": entry["last_seen"], + "online": entry["online"], + } + + +def directory(config: CoreProtectConfig, query: str = "", sort: str = "last_seen", page: int = 1) -> dict: + if sort not in SORTS: + raise PlayerError(400, "bad_sort") + text = " ".join((query or "").split()).lstrip("@") + if len(text) > QUERY_MAX: + raise PlayerError(400, "query_too_long") + entries, status = _directory(config) + needle = text.casefold() + chosen = [e for e in entries if _matches(e, needle)] if needle else list(entries) + chosen.sort(key=_sort_key(sort)) + page = max(1, page) + start = (page - 1) * PAGE_SIZE + return { + "players": [_directory_json(e) for e in chosen[start:start + PAGE_SIZE]], + "total": len(chosen), + "page": page, + "page_size": PAGE_SIZE, + "coreprotect": {**status, "server_label": config.label or None}, + } + + +def clear_cache() -> None: + _DIRECTORY.clear() + + +# -------------------- +# Profile, sessions, activity +# -------------------- + +def _ids(reader: Reader, key: str) -> list[dict]: + """A player's co_user rows, oldest first. CoreProtect keeps one per UUID, but nothing enforces it.""" + return [dict(r) for r in reader.rows("SELECT id, user, time FROM co_user WHERE uuid = ? ORDER BY time, id", (key,))] + + +def _require(text: str) -> str: + key = canonical_uuid(text) + if key is None: + raise PlayerError(400, "bad_uuid") + return key + + +def profile(config: CoreProtectConfig, text: str) -> dict: + key = _require(text) + now = int(time.time()) + error = None + accounts: list[dict] = [] + names: list[dict] = [] + event = first_seen = None + try: + with Reader(config, Budget()) as reader: + accounts = _ids(reader, key) + if accounts: + ids = [a["id"] for a in accounts] + names = [dict(r) for r in reader.rows( + "SELECT user, time FROM co_username_log WHERE uuid = ? ORDER BY time, id", (key,))] + event = sessions.last_event(reader, ids) + first = [r["first"] for uid in ids for r in reader.rows( + "SELECT MIN(time) AS first FROM co_session WHERE user = ?", (uid,)) if r["first"] is not None] + first_seen = min(first) if first else None + except Unavailable as exc: + error = exc + links, roster = _site_rows(key) + if not accounts and not links and not roster and error is None: + raise PlayerError(404, "player_not_found") + link = links[0] if links else None + current = accounts[-1]["user"] if accounts else (link["minecraft_name"] if link else None) + return { + "uuid": key, + "minecraft_name": current, + "past_names": [{"name": n["user"], "time": n["time"]} for n in names if n["user"] != current], + "first_seen": first_seen, + "last_seen": event["time"] if event else None, + "online": sessions.is_fresh(event, now, config.ping_seconds), + "discord": _discord(link), + "account": _account(link), + "characters": [ + {k: c[k] for k in ("realm_id", "character_id", "name", "status", "race", "class", "updated_at")} + for c in roster + ], + "coreprotect": {**_status(error), "server_label": config.label or None, + "ping_seconds": config.ping_seconds or None}, + } + + +def player_sessions(config: CoreProtectConfig, text: str, before: str | None, limit: int) -> dict: + key = _require(text) + try: + values = decode_cursor(config.server, f"sessions:{key}", before, 2) + except BadCursor: + raise PlayerError(400, "bad_cursor") from None + try: + with Reader(config, Budget()) as reader: + ids = [a["id"] for a in _ids(reader, key)] + maps = co_maps.get(reader) + raw = sessions.fetch(reader, ids, sessions.Key(*values) if values else None, limit) if ids else None + except Unavailable as exc: + return {"sessions": [], "next": None, "coreprotect": _status(exc)} + if raw is None: + return {"sessions": [], "next": None, "coreprotect": _status(None)} + built = sessions.build(raw, maps, int(time.time()), config.ping_seconds) + nxt = raw["next"] + return { + **built, + "next": encode_cursor(config.server, f"sessions:{key}", nxt.time, nxt.rowid) if nxt else None, + "coreprotect": _status(None), + } + + +def player_activity(config: CoreProtectConfig, text: str, before: str | None, limit: int, + kinds: str | None) -> dict: + key = _require(text) + try: + chosen = activity.parse_kinds(kinds) + except ValueError: + raise PlayerError(400, "bad_kinds") from None + try: + cursor = activity.decode_cursor(f"{config.server}:{key}", before) + except BadCursor: + raise PlayerError(400, "bad_cursor") from None + try: + with Reader(config, Budget()) as reader: + ids = [a["id"] for a in _ids(reader, key)] + maps = co_maps.get(reader) + raw = activity.fetch(reader, maps, ids, chosen, cursor, limit) if ids else None + except Unavailable as exc: + return {"entries": [], "next": None, "searched_to": None, "kinds": list(activity.KINDS), "coreprotect": _status(exc)} + if raw is None: + return {"entries": [], "next": None, "searched_to": None, "kinds": list(activity.KINDS), "coreprotect": _status(None)} + return {**activity.build(raw, maps, f"{config.server}:{key}"), "kinds": list(activity.KINDS), + "coreprotect": _status(None)} diff --git a/backend/src/auth/roles.py b/backend/src/auth/roles.py index 06180e9d..c43d8473 100644 --- a/backend/src/auth/roles.py +++ b/backend/src/auth/roles.py @@ -7,6 +7,7 @@ # Minimum role for each capability. CAPABILITIES = { "view_admin": "mod", + "view_players": "mod", "revoke_sessions": "mod", "change_role": "admin", } diff --git a/backend/src/auth/test_admin.py b/backend/src/auth/test_admin.py index 090830b0..0d50630b 100644 --- a/backend/src/auth/test_admin.py +++ b/backend/src/auth/test_admin.py @@ -82,7 +82,7 @@ def test_role_rules(): assert roles.outranks("admin", "mod") and not roles.outranks("admin", "admin") assert not roles.outranks("root", "unknown") and not roles.outranks("unknown", "player") assert not roles.is_staff("player") and not roles.is_staff("wizard") and roles.is_staff("mod") - assert roles.capabilities("mod") == ["view_admin", "revoke_sessions"] + assert roles.capabilities("mod") == ["view_admin", "view_players", "revoke_sessions"] assert roles.capabilities("bogus") == [] @@ -159,7 +159,7 @@ def test_panel_reads_need_staff(app, env): _, mod = account(env, "mona", "mod") me = client(app, mod).get("/admin/me") assert me.status_code == 200 and me.headers["cache-control"] == "no-store" - assert me.json()["capabilities"] == ["view_admin", "revoke_sessions"] + assert me.json()["capabilities"] == ["view_admin", "view_players", "revoke_sessions"] assert me.json()["assignable_roles"] == [] diff --git a/backend/src/auth/test_players.py b/backend/src/auth/test_players.py new file mode 100644 index 00000000..b89dbff8 --- /dev/null +++ b/backend/src/auth/test_players.py @@ -0,0 +1,191 @@ +import pytest +from fastapi import FastAPI + +from src.api import admin_routes, auth_routes +from src.auth import players +from src.auth.test_admin import account, client + +HAZEL = "0615a817-8cb4-4aef-95f7-f6c9bf7611b8" +LINKED_ONLY = "11111111-1111-1111-1111-111111111111" +ROSTER_ONLY = "22222222-2222-2222-2222-222222222222" +NOW_ISH = 2_000_000_000 + + +@pytest.fixture +def app(env, coreprotect): + players.clear_cache() + app = FastAPI() + app.include_router(auth_routes.auth_router) + admin_routes.install(app) + yield app + players.clear_cache() + + +@pytest.fixture +def env(database, monkeypatch): + monkeypatch.setenv("DISCORD_AUTH_ENABLED", "1") + monkeypatch.setenv("DISCORD_CLIENT_ID", "client-id") + monkeypatch.setenv("DISCORD_CLIENT_SECRET", "client-secret") + monkeypatch.setenv("DISCORD_GUILD_ID", "999999999999999999") + monkeypatch.setenv("SITE_PUBLIC_URL", "https://www.tfminecraft.net") + monkeypatch.delenv("PS_PRODUCTION", raising=False) + return database + + +def link(db, player_uuid, discord_id, mc_name, discord_name): + with db.connect() as conn: + conn.execute( + "INSERT INTO discord_links (player_uuid, discord_user_id, minecraft_name, discord_username, linked_at) " + "VALUES (?, ?, ?, ?, '2026-09-20T10:00:00Z')", (player_uuid, discord_id, mc_name, discord_name)) + conn.commit() + + +def character(db, player_uuid, name, realm="main"): + with db.connect() as conn: + conn.execute( + "INSERT INTO character_roster (player_uuid, realm_id, character_id, name, status, race, class, " + "updated_at) VALUES (?, ?, ?, ?, 'alive', 'human', 'smith', '2026-09-21T10:00:00Z')", + (player_uuid, realm, f"c-{name}", name)) + conn.commit() + + +@pytest.fixture +def world(env, coreprotect): + hazel = coreprotect.user("MrEnzo99", HAZEL) + coreprotect.execute("INSERT INTO co_username_log (time, uuid, user) VALUES (1, ?, 'OldEnzo')", (HAZEL,)) + coreprotect.execute("INSERT INTO co_username_log (time, uuid, user) VALUES (2, ?, 'MrEnzo99')", (HAZEL,)) + coreprotect.user("#fire", None) + quiet = coreprotect.user("Quiet", "33333333-3333-3333-3333-333333333333") + coreprotect.session(hazel, 100, 1) + coreprotect.session(hazel, 200, 0) + coreprotect.session(quiet, 50, 1) + coreprotect.block(hazel, 150, 0) + coreprotect.command(hazel, 160, "/msg Bob a private thing") + link(env, HAZEL, "422545450919526411", "MrEnzo99", "hazelstone") + link(env, LINKED_ONLY, "500000000000000001", "LinkedOnly", "linky") + character(env, HAZEL, "Hazel Stonebrook") + character(env, ROSTER_ONLY, "Aldric") + return {"hazel": hazel} + + +def staff(env, role="mod"): + _, token = account(env, f"staff-{role}", role) + return token + + +def test_requires_a_staff_role(app, env, world): + assert client(app).get("/admin/players").status_code == 401 + assert client(app, staff(env, "player")).get("/admin/players").status_code == 403 + assert client(app, staff(env, "player")).get(f"/admin/players/{HAZEL}/activity").status_code == 403 + assert client(app, staff(env, "mod")).get("/admin/players").status_code == 200 + + +def test_directory_merges_every_source(app, env, world): + body = client(app, staff(env)).get("/admin/players").json() + assert body["coreprotect"] == {"status": "available", "server_label": "Vardera"} + by_uuid = {p["uuid"]: p for p in body["players"]} + assert set(by_uuid) == {HAZEL, LINKED_ONLY, ROSTER_ONLY, "33333333-3333-3333-3333-333333333333"} + assert by_uuid[HAZEL]["minecraft_name"] == "MrEnzo99" + assert by_uuid[HAZEL]["discord_username"] == "hazelstone" + assert by_uuid[HAZEL]["characters"] == ["Hazel Stonebrook"] + assert by_uuid[HAZEL]["last_seen"] == 200 + assert by_uuid[LINKED_ONLY]["minecraft_name"] == "LinkedOnly" + assert by_uuid[ROSTER_ONLY]["minecraft_name"] is None + # Last seen first, never-seen last. + assert [p["uuid"] for p in body["players"]][:2] == [HAZEL, "33333333-3333-3333-3333-333333333333"] + + +@pytest.mark.parametrize("query", ["oldenzo", "HAZELSTONE", "@hazelstone", "stonebrook", HAZEL, + HAZEL.replace("-", ""), "422545450919526411"]) +def test_directory_search(app, env, world, query): + body = client(app, staff(env)).get("/admin/players", params={"q": query}).json() + assert [p["uuid"] for p in body["players"]] == [HAZEL] + + +def test_directory_sorts(app, env, world): + c = client(app, staff(env)) + by_mc = [p["minecraft_name"] for p in c.get("/admin/players", params={"sort": "minecraft"}).json()["players"]] + assert by_mc == ["LinkedOnly", "MrEnzo99", "Quiet", None] + by_discord = [p["discord_username"] for p in c.get("/admin/players", params={"sort": "discord"}).json()["players"]] + assert by_discord[:2] == ["hazelstone", "linky"] + assert c.get("/admin/players", params={"sort": "bogus"}).status_code == 400 + assert c.get("/admin/players", params={"q": "x" * 65}).status_code == 400 + + +def test_directory_without_coreprotect(app, env, world, monkeypatch): + monkeypatch.setenv("COREPROTECT_DB", str(env.DATA_DIR / "nowhere.db")) + body = client(app, staff(env)).get("/admin/players").json() + assert body["coreprotect"]["status"] == "unavailable" + assert body["coreprotect"]["reason"] == "missing" + assert {p["uuid"] for p in body["players"]} == {HAZEL, LINKED_ONLY, ROSTER_ONLY} + + +def test_profile(app, env, world): + body = client(app, staff(env)).get(f"/admin/players/{HAZEL.upper()}").json() + assert body["uuid"] == HAZEL + assert body["minecraft_name"] == "MrEnzo99" + assert body["past_names"] == [{"name": "OldEnzo", "time": 1}] + assert body["first_seen"] == 100 + assert body["last_seen"] == 200 + assert body["online"] is False + assert body["discord"]["discord_username"] == "hazelstone" + assert body["account"] is None + assert body["characters"][0]["name"] == "Hazel Stonebrook" + assert body["coreprotect"]["status"] == "available" + + +def test_profile_shows_the_website_account(app, env, world): + token = staff(env, "admin") + with env.connect() as conn: + conn.execute("UPDATE discord_links SET discord_user_id = (SELECT discord_user_id FROM users LIMIT 1) " + "WHERE player_uuid = ?", (HAZEL,)) + conn.commit() + body = client(app, token).get(f"/admin/players/{HAZEL}").json() + assert body["account"]["role"] == "admin" + + +def test_profile_errors(app, env, world, monkeypatch): + c = client(app, staff(env)) + assert c.get("/admin/players/not-a-uuid").status_code == 400 + assert c.get("/admin/players/44444444-4444-4444-4444-444444444444").status_code == 404 + monkeypatch.setenv("COREPROTECT_DB", "") + body = c.get("/admin/players/44444444-4444-4444-4444-444444444444").json() + assert body["coreprotect"] == {"status": "unavailable", "reason": "not_configured", "server_label": "Vardera", + "ping_seconds": 60} + assert c.get(f"/admin/players/{ROSTER_ONLY}").json()["characters"][0]["name"] == "Aldric" + + +def test_sessions_and_activity(app, env, world): + c = client(app, staff(env)) + sessions = c.get(f"/admin/players/{HAZEL}/sessions").json() + assert [(s["start"]["time"], s["end"]["time"], s["end_kind"]) for s in sessions["sessions"]] == [(100, 200, "logout")] + assert sessions["next"] is None + feed = c.get(f"/admin/players/{HAZEL}/activity").json() + assert [e["kind"] for e in feed["entries"]] == ["session", "command", "block", "session"] + assert "private" not in str(feed) + assert "chat" not in feed["kinds"] + only_blocks = c.get(f"/admin/players/{HAZEL}/activity", params={"kinds": "block"}).json() + assert [e["kind"] for e in only_blocks["entries"]] == ["block"] + + +def test_paging_and_bad_input(app, env, world): + c = client(app, staff(env)) + first = c.get(f"/admin/players/{HAZEL}/activity", params={"limit": 1}).json() + assert first["next"] + second = c.get(f"/admin/players/{HAZEL}/activity", params={"limit": 1, "before": first["next"]}).json() + assert second["entries"][0]["id"] != first["entries"][0]["id"] + # A cursor for one player is refused for another. + other = "33333333-3333-3333-3333-333333333333" + assert c.get(f"/admin/players/{other}/activity", params={"before": first["next"]}).status_code == 400 + assert c.get(f"/admin/players/{HAZEL}/activity", params={"kinds": "chat"}).status_code == 400 + assert c.get(f"/admin/players/{HAZEL}/activity", params={"limit": 101}).status_code == 422 + assert c.get(f"/admin/players/{HAZEL}/sessions", params={"before": "junk"}).status_code == 400 + + +def test_unknown_player_sections_are_empty(app, env, world, monkeypatch): + c = client(app, staff(env)) + body = c.get(f"/admin/players/{ROSTER_ONLY}/sessions").json() + assert body == {"sessions": [], "next": None, "coreprotect": {"status": "available"}} + monkeypatch.setenv("COREPROTECT_DB", str(env.DATA_DIR / "nowhere.db")) + body = c.get(f"/admin/players/{HAZEL}/activity").json() + assert body["coreprotect"] == {"status": "unavailable", "reason": "missing"} diff --git a/backend/src/coreprotect/README.md b/backend/src/coreprotect/README.md new file mode 100644 index 00000000..28ec87d7 --- /dev/null +++ b/backend/src/coreprotect/README.md @@ -0,0 +1,74 @@ +# CoreProtect reader + +The staff panel's player pages read the Minecraft server's live CoreProtect +SQLite database: who has played, their sessions, and their recent actions. +`src/auth/players.py` joins that with `discord_links`, `users` and +`character_roster`; the routes are `GET /admin/players`, +`/admin/players/{uuid}`, `/sessions` and `/activity` (`view_players`, mod and +above). + +## Settings + +| Variable | Meaning | +| --- | --- | +| `COREPROTECT_DB` | Path to `database.db` inside the backend container. Unset: the directory still lists linked players and characters, and CoreProtect sections say they are unavailable. | +| `COREPROTECT_SERVER` | Short id that scopes caches and page cursors (default `main`). | +| `COREPROTECT_SERVER_LABEL` | World name shown with the data, for example `Vardera`. | +| `COREPROTECT_PING_SECONDS` | The server's `player-pings` interval (default 60; `0` if pings are off). Only used to guess whether someone is online and to end crashed sessions. | + +Mount the CoreProtect **directory** read-only, so SQLite can see a hot +`-journal`. The paths are host-specific, so they belong in the host's +`docker-compose.override.yml`, not the repository's compose files: + +```yaml +services: + backend: + volumes: + - /home/amp/.ampdata/instances/TFMCMain01/Minecraft/plugins/CoreProtect:/coreprotect:ro + environment: + - COREPROTECT_DB=/coreprotect/database.db + - COREPROTECT_SERVER=main + - COREPROTECT_SERVER_LABEL=Vardera +``` + +Docker resolves the bind as root, so the AMP tree needs no permission +changes. The mount also exposes CoreProtect's `config.yml`, which holds +connection settings for database engines the server does not use. + +## Keeping CoreProtect's writer unblocked + +The database uses a rollback journal, so a reader's lock holds off +CoreProtect's commits. `reader.py` explains the guards: autocommit, one +reader per process, a 1.5 s budget per request, and every statement fully +fetched before the next. The backend runs as one uvicorn process; running +more would need the reader limit shared between them. Never open the file +with `immutable=1`: on Main that read torn pages while CoreProtect wrote. + +Every query is an indexed seek by player, except small whole-table reads of +CoreProtect's name tables, `co_user` and `co_username_log` (a few hundred +to a few thousand rows). Activity pages examine at most `SCAN_LIMIT` rows per +table, so a sparse filter stops early and returns a cursor to continue +(`searched_to`). + +## What is shown + +Chat is never read. Commands are cut to their first word inside SQL, and sign +text, item metadata and NBT are not selected. Sessions are rebuilt from +login, logout and ping rows; see `sessions.py` for how crashed sessions end. + +## Deployment gate + +`backend/benchmarks/coreprotect_reader.py` measures both sides: + +```sh +# Lock hold time per request against a live database (opens it immutable, benchmark only) +python3 benchmarks/coreprotect_reader.py hold --no-lock /path/to/database.db +# Writer impact, on a copy (it writes) +python3 benchmarks/coreprotect_reader.py contend /tmp/copy.db --readers 1 +``` + +Acceptance: `hold` p99 ≤ 100 ms and max ≤ 500 ms; `contend` with no writer +failures and writer transaction p99/max up by no more than 100/500 ms. +On 2026-10-06 Main (7.4 GB, 242 players) gave p99 8.6 ms and max 18.5 ms; a +copy of Dev's database gave writer p99 +15 ms with one reader and +59 ms with +three readers making back-to-back requests, with no failures. diff --git a/backend/src/coreprotect/__init__.py b/backend/src/coreprotect/__init__.py new file mode 100644 index 00000000..82e77efc --- /dev/null +++ b/backend/src/coreprotect/__init__.py @@ -0,0 +1 @@ +"""Read-only access to a Minecraft server's live CoreProtect database.""" diff --git a/backend/src/coreprotect/activity.py b/backend/src/coreprotect/activity.py new file mode 100644 index 00000000..2787d7e1 --- /dev/null +++ b/backend/src/coreprotect/activity.py @@ -0,0 +1,243 @@ +"""A player's recent CoreProtect actions, merged across tables, newest first. + +Each source is read with its own (user, time) index seek and the same strict +boundary, so pages never repeat or skip rows that share a second. Each scan +examines at most SCAN_LIMIT rows, so a sparse filter costs the same as any +other page. The merged +order is (time, source rank, rowid), all descending; the cursor is the last +row's key. + +Only what staff need to see what happened is selected. Chat is never read. +Commands are cut to their first word inside SQL, so arguments (messages, +passwords, link codes) never leave the database. Sign text, item metadata +and NBT are not selected. +""" +from __future__ import annotations + +from dataclasses import dataclass + +from . import cursors +from .maps import Maps +from .reader import Reader + +DEFAULT_LIMIT = 20 +MAX_LIMIT = 100 + +# Rows one source may examine per page. A sparse filter (say, kills by a +# player with years of building) would otherwise walk their whole history; +# instead the page stops where the scan did and "next" carries on from there. +SCAN_LIMIT = 2000 + +_SKILL = "substr(message, 1, 7) = '[skill]'" +_TELEPORT = "instr(message, ' teleported from ')" +# The command's first word, or a skill cast's name without teleport coordinates. +_COMMAND_TEXT = ( + f"substr(CASE WHEN {_SKILL} THEN " + f" CASE WHEN {_TELEPORT} > 0 THEN substr(message, 9, {_TELEPORT} - 9) ELSE substr(message, 9) END " + "ELSE CASE WHEN instr(message, ' ') > 0 THEN substr(message, 1, instr(message, ' ') - 1) ELSE message END " + "END, 1, 80)" +) + + +@dataclass(frozen=True) +class Source: + rank: int + name: str + table: str + columns: str + kinds: dict[str, str] + + +SOURCES = ( + Source(1, "block", "co_block", "type, data, action, rolled_back", + {"block": "action IN (0, 1)", "click": "action = 2", "kill": "action = 3", "spawn": "action = 13"}), + Source(2, "container", "co_container", "type, amount, action, rolled_back", + {"container": "action IN (0, 1)"}), + Source(3, "entity_container", "co_entity_container", "type, amount, action, rolled_back", + {"container": "action IN (0, 1)"}), + Source(4, "item", "co_item", "type, amount, action, rolled_back", + {"item": "action BETWEEN 0 AND 12"}), + Source(5, "entity", "co_entity_interaction", "type, action, rolled_back", + {"entity": "action BETWEEN 0 AND 3"}), + Source(6, "sign", "co_sign", "action", {"sign": "action IN (0, 1, 2)"}), + Source(7, "command", "co_command", + f"{_SKILL} AS is_skill, {_TELEPORT} > 0 AS teleport, {_COMMAND_TEXT} AS text", + {"skill": _SKILL, "command": f"NOT {_SKILL}"}), + Source(8, "session", "co_session", "action", {"session": "action IN (0, 1)"}), +) +KINDS = tuple(dict.fromkeys(kind for source in SOURCES for kind in source.kinds)) +_BY_RANK = {source.rank: source for source in SOURCES} + + +@dataclass(frozen=True, order=True) +class Cursor: + time: int + rank: int + rowid: int + + +def encode_cursor(server: str, cursor: Cursor) -> str: + return cursors.encode(server, "activity", cursor.time, cursor.rank, cursor.rowid) + + +def decode_cursor(server: str, token: str | None) -> Cursor | None: + values = cursors.decode(server, "activity", token, 3) + if values is None: + return None + if values[1] not in _BY_RANK: + raise cursors.BadCursor("bad_cursor") + return Cursor(*values) + + +def parse_kinds(text: str | None) -> tuple[str, ...]: + if not text: + return KINDS + chosen = tuple(dict.fromkeys(part.strip() for part in text.split(",") if part.strip())) + if not chosen or any(kind not in KINDS for kind in chosen): + raise ValueError("bad_kinds") + return chosen + + +def _boundary(source: Source, cursor: Cursor | None) -> tuple[str, tuple]: + if cursor is None: + return "", () + if source.rank < cursor.rank: + return " AND time <= ?", (cursor.time,) + if source.rank > cursor.rank: + return " AND time < ?", (cursor.time,) + return " AND time <= ? AND NOT (time = ? AND rowid >= ?)", (cursor.time, cursor.time, cursor.rowid) + + +def _key(source: Source, row: dict) -> Cursor: + return Cursor(row["time"], source.rank, row["rid"]) + + +def fetch(reader: Reader, maps: Maps, ids: list[int], kinds: tuple[str, ...], cursor: Cursor | None, + limit: int) -> dict: + limit = max(1, min(limit, MAX_LIMIT)) + rows = [] + # The newest point a capped scan reached. Rows older than it may sit + # beside unexamined rows of that source, so this page stops there. + frontier: Cursor | None = None + for source in SOURCES: + if source.table not in maps.tables: + continue + conditions = [cond for kind, cond in source.kinds.items() if kind in kinds] + if not conditions: + continue + bound, bound_params = _boundary(source, cursor) + # The oldest row this page may examine, found from the index alone. + edge_sql = ( + f"SELECT time, rowid AS rid FROM {source.table} WHERE user = ?{bound} " + f"ORDER BY time DESC, rowid DESC LIMIT 1 OFFSET {SCAN_LIMIT - 1}" + ) + for uid in ids: + edge = reader.rows(edge_sql, (uid, *bound_params)) + floor, floor_params = "", () + if edge: + floor = " AND time >= ? AND NOT (time = ? AND rowid < ?)" + floor_params = (edge[0]["time"], edge[0]["time"], edge[0]["rid"]) + found = [dict(row) for row in reader.rows( + f"SELECT rowid AS rid, time, wid, x, y, z, {source.columns} FROM {source.table} " + f"WHERE user = ? AND ({' OR '.join(conditions)}){bound}{floor} " + "ORDER BY time DESC, rowid DESC LIMIT ?", + (uid, *bound_params, *floor_params, limit + 1), + )] + rows += [(source, row) for row in found] + if edge and len(found) <= limit: + reached = _key(source, dict(edge[0])) + frontier = reached if frontier is None else max(frontier, reached) + rows.sort(key=lambda item: _key(*item), reverse=True) + if frontier is not None: + rows = [item for item in rows if _key(*item) >= frontier] + if len(rows) > limit: + rows = rows[:limit] + nxt = _key(*rows[-1]) + else: + nxt = frontier + victims = sorted({row["data"] for source, row in rows + if source.name == "block" and row["action"] == 3 and row["type"] == 0 and row["data"]}) + names = {} + if victims: + marks = ",".join("?" * len(victims)) + names = {r["id"]: {"minecraft_name": r["user"], "uuid": r["uuid"]} + for r in reader.rows(f"SELECT id, user, uuid FROM co_user WHERE id IN ({marks})", tuple(victims))} + return { + "rows": rows, + "victims": names, + "next": nxt, + # Set when the page stopped at a scan limit rather than filling up. + "searched_to": frontier.time if frontier is not None and nxt == frontier else None, + } + + +_ITEM_VERBS = { + 0: "removed", 1: "added", 2: "dropped", 3: "picked up", 4: "took from an ender chest", + 5: "put in an ender chest", 6: "threw", 7: "shot", 8: "broke", 9: "destroyed", 10: "crafted", + 11: "sold", 12: "bought", +} +_ENTITY_VERBS = {0: "clicked", 1: "sheared", 2: "leashed", 3: "unleashed"} +_SIGN_VERBS = {0: "broke", 1: "placed", 2: "edited"} + + +def _rollback(source: Source, state: int | None) -> str | None: + if not state: + return None + if source.name in {"container", "entity_container", "item"}: + return {1: "rolled back", 2: "rolled back (player inventory)", 3: "rolled back (both)"}.get( + state, f"rollback state {state}") + return "rolled back" if state == 1 else f"rollback state {state}" + + +def _entry(source: Source, row: dict, maps: Maps, victims: dict) -> dict: + action = row.get("action") + entry = { + "id": f"{source.name}:{row['rid']}", + "time": row["time"], + "world": maps.world(row["wid"]), + "x": row["x"], "y": row["y"], "z": row["z"], + "amount": None, + "victim": None, + "rolled_back": _rollback(source, row.get("rolled_back")), + } + if source.name == "block": + if action == 3: + if row["type"] == 0: + victim = victims.get(row["data"]) + entry.update(kind="kill", verb="killed", victim=victim, + target=victim["minecraft_name"] if victim else "a player") + else: + entry.update(kind="kill", verb="killed", target=maps.entity(row["type"])) + elif action == 13: + entry.update(kind="spawn", verb="spawned", target=maps.entity(row["type"])) + elif action == 2: + entry.update(kind="click", verb="clicked", target=maps.material(row["type"])) + else: + entry.update(kind="block", verb="placed" if action == 1 else "broke", target=maps.material(row["type"])) + elif source.name in {"container", "entity_container"}: + entry.update(kind="container", verb="added" if action == 1 else "removed", + target=maps.material(row["type"]), amount=row["amount"]) + elif source.name == "item": + entry.update(kind="item", verb=_ITEM_VERBS.get(action, "moved"), + target=maps.material(row["type"]), amount=row["amount"]) + elif source.name == "entity": + entry.update(kind="entity", verb=_ENTITY_VERBS.get(action, "clicked"), target=maps.entity(row["type"])) + elif source.name == "sign": + entry.update(kind="sign", verb=_SIGN_VERBS.get(action, "changed"), target="sign") + elif source.name == "command": + if row["is_skill"]: + entry.update(kind="skill", verb="teleported with" if row["teleport"] else "cast", + target=(row["text"] or "").strip() or "a skill") + else: + entry.update(kind="command", verb="ran", target=row["text"] or "/") + else: + entry.update(kind="session", verb="logged in" if action == 1 else "logged out", target=None) + return entry + + +def build(raw: dict, maps: Maps, server: str) -> dict: + return { + "entries": [_entry(source, row, maps, raw["victims"]) for source, row in raw["rows"]], + "next": encode_cursor(server, raw["next"]) if raw["next"] else None, + "searched_to": raw["searched_to"], + } diff --git a/backend/src/coreprotect/cursors.py b/backend/src/coreprotect/cursors.py new file mode 100644 index 00000000..3546ca2e --- /dev/null +++ b/backend/src/coreprotect/cursors.py @@ -0,0 +1,31 @@ +"""Opaque page cursors, bound to one server and one listing.""" +from __future__ import annotations + +import base64 +import binascii +import json + + +class BadCursor(ValueError): + pass + + +def encode(server: str, listing: str, *values: int) -> str: + raw = json.dumps({"s": server, "l": listing, "v": list(values)}, separators=(",", ":")) + return base64.urlsafe_b64encode(raw.encode()).decode().rstrip("=") + + +def decode(server: str, listing: str, token: str | None, size: int) -> tuple[int, ...] | None: + if not token: + return None + if len(token) > 200: + raise BadCursor("bad_cursor") + try: + data = json.loads(base64.urlsafe_b64decode(token + "=" * (-len(token) % 4))) + values = tuple(data["v"]) + ok = data["s"] == server and data["l"] == listing + except (ValueError, TypeError, KeyError, binascii.Error): + raise BadCursor("bad_cursor") from None + if not ok or len(values) != size or not all(type(v) is int for v in values): + raise BadCursor("bad_cursor") + return values diff --git a/backend/src/coreprotect/maps.py b/backend/src/coreprotect/maps.py new file mode 100644 index 00000000..bf45b33a --- /dev/null +++ b/backend/src/coreprotect/maps.py @@ -0,0 +1,69 @@ +"""CoreProtect's id → name tables and table list, cached per server. + +These are small whole-table reads (about 1,300 materials, a few dozen +entities and worlds on Main), the only queries here that are not seeks by +player. One refresh runs at a time; the others wait within their budget. +""" +from __future__ import annotations + +import threading +import time +from dataclasses import dataclass + +from .reader import Reader + +TTL_SECONDS = 300 + +_LOCK = threading.Lock() +_CACHE: dict[tuple[str, str], tuple[float, "Maps"]] = {} + + +@dataclass(frozen=True) +class Maps: + materials: dict[int, str] + entities: dict[int, str] + worlds: dict[int, str] + tables: frozenset[str] + + def material(self, type_id: int | None) -> str: + return _short(self.materials.get(type_id or 0)) or f"material #{type_id}" + + def entity(self, type_id: int | None) -> str: + return _short(self.entities.get(type_id or 0)) or f"entity #{type_id}" + + def world(self, wid: int | None) -> str | None: + return self.worlds.get(wid or 0) + + +def _short(name: str | None) -> str | None: + if not name: + return None + return name.removeprefix("minecraft:") + + +def get(reader: Reader) -> Maps: + key = (reader.config.server, reader.config.path or "") + cached = _CACHE.get(key) + if cached and cached[0] > time.monotonic(): + return cached[1] + # The reader holds no open statement here, so waiting cannot pin a read lock. + reader.budget.acquire(_LOCK) + try: + cached = _CACHE.get(key) + if cached and cached[0] > time.monotonic(): + return cached[1] + maps = Maps( + materials={r["id"]: r["material"] for r in reader.rows("SELECT id, material FROM co_material_map")}, + entities={r["id"]: r["entity"] for r in reader.rows("SELECT id, entity FROM co_entity_map")}, + worlds={r["id"]: r["world"] for r in reader.rows("SELECT id, world FROM co_world")}, + # Older CoreProtect builds lack the fork's entity tables. + tables=frozenset(r["name"] for r in reader.rows("SELECT name FROM sqlite_master WHERE type = 'table'")), + ) + _CACHE[key] = (time.monotonic() + TTL_SECONDS, maps) + return maps + finally: + _LOCK.release() + + +def clear() -> None: + _CACHE.clear() diff --git a/backend/src/coreprotect/reader.py b/backend/src/coreprotect/reader.py new file mode 100644 index 00000000..ab2070ff --- /dev/null +++ b/backend/src/coreprotect/reader.py @@ -0,0 +1,167 @@ +"""Guarded read-only connections to a live CoreProtect SQLite database. + +CoreProtect writes this file from the Minecraft server while we read it. The +database uses a rollback journal, so any reader's SHARED lock holds off the +writer's commit (and its cache spills) until the reader lets go. CoreProtect +writes from its consumer thread and retries a busy commit, so a short hold +delays its queue rather than the server tick, but a long one would back the +queue up. Every read here is therefore kept short and bounded: + +- autocommit, and every statement is fully fetched and closed before the + next, so no read transaction stays open between statements; +- one reader at a time per process. The backend runs as a single uvicorn + process; a second process or replica would need its own coordination; +- one budget per request covering the wait for the reader slot, cache locks + and every statement. Each statement's busy timeout is capped by what is + left, and a progress handler interrupts a statement that overruns. + The progress handler only runs while SQLite executes VM steps, so the + budget is cooperative: it cannot cut short a stalled disk read; +- the connection is closed before any decoding, province.db access or + response building. + +Never open the file with immutable=1: that skips locking and reads torn +pages while CoreProtect writes (seen on Main as "database disk image is +malformed"). +""" +from __future__ import annotations + +import os +import sqlite3 +import threading +import time +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import quote + +REQUEST_BUDGET_SECONDS = 1.5 +BUSY_TIMEOUT_MS = 200 +PROGRESS_STEPS = 1000 + +_GATE = threading.BoundedSemaphore(1) + + +@dataclass(frozen=True) +class CoreProtectConfig: + path: str | None + # Scopes caches and cursors, so a cursor from one server is never read against another. + server: str + label: str + # CoreProtect's player-pings interval; 0 when pings are off or unknown. + ping_seconds: int + + @classmethod + def from_env(cls) -> CoreProtectConfig: + path = os.getenv("COREPROTECT_DB", "").strip() or None + try: + ping = max(0, int(os.getenv("COREPROTECT_PING_SECONDS", "60").strip() or "0")) + except ValueError: + ping = 0 + return cls( + path=path, + server=os.getenv("COREPROTECT_SERVER", "main").strip() or "main", + label=os.getenv("COREPROTECT_SERVER_LABEL", "").strip(), + ping_seconds=ping, + ) + + +class Unavailable(Exception): + """CoreProtect could not be read: not_configured, missing, busy, timeout or error.""" + + def __init__(self, code: str): + super().__init__(code) + self.code = code + + +class Budget: + """A monotonic deadline shared by everything one request does with CoreProtect.""" + + def __init__(self, seconds: float = REQUEST_BUDGET_SECONDS): + self.deadline = time.monotonic() + seconds + + def remaining(self) -> float: + return self.deadline - time.monotonic() + + def check(self) -> float: + left = self.remaining() + if left <= 0: + raise Unavailable("timeout") + return left + + def acquire(self, lock) -> None: + if not lock.acquire(timeout=self.check()): + raise Unavailable("busy") + + +class Reader: + """One guarded connection: `with Reader(config, budget) as r: r.rows(sql, params)`.""" + + def __init__(self, config: CoreProtectConfig, budget: Budget | None = None): + self.config = config + self.budget = budget or Budget() + self._conn: sqlite3.Connection | None = None + self._gated = False + + def __enter__(self) -> Reader: + if not self.config.path: + raise Unavailable("not_configured") + path = Path(self.config.path) + if not path.is_file(): + raise Unavailable("missing") + self.budget.acquire(_GATE) + self._gated = True + try: + self._conn = sqlite3.connect( + f"file:{quote(str(path))}?mode=ro", uri=True, isolation_level=None, + timeout=0, check_same_thread=True, + ) + self._conn.row_factory = sqlite3.Row + deadline = self.budget.deadline + self._conn.set_progress_handler(lambda: int(time.monotonic() > deadline), PROGRESS_STEPS) + self._statement("PRAGMA query_only = 1", ()) + except BaseException: + self.close() + raise + return self + + def __exit__(self, *exc) -> None: + self.close() + + def close(self) -> None: + try: + if self._conn is not None: + self._conn.close() + finally: + self._conn = None + if self._gated: + self._gated = False + _GATE.release() + + def rows(self, sql: str, params: tuple = ()) -> list[sqlite3.Row]: + return self._statement(sql, params) + + def _statement(self, sql: str, params: tuple) -> list[sqlite3.Row]: + if self._conn is None: + raise Unavailable("error") + left = self.budget.check() + try: + self._conn.execute(f"PRAGMA busy_timeout = {max(1, min(BUSY_TIMEOUT_MS, int(left * 1000)))}") + cursor = self._conn.execute(sql, params) + try: + return cursor.fetchall() + finally: + cursor.close() + except sqlite3.OperationalError as exc: + raise Unavailable(_classify(exc)) from None + except sqlite3.DatabaseError: + raise Unavailable("error") from None + + +def _classify(exc: sqlite3.OperationalError) -> str: + text = str(exc).lower() + if "interrupted" in text: + return "timeout" + if "locked" in text or "busy" in text: + return "busy" + if "unable to open" in text: + return "missing" + return "error" diff --git a/backend/src/coreprotect/sessions.py b/backend/src/coreprotect/sessions.py new file mode 100644 index 00000000..cba3f489 --- /dev/null +++ b/backend/src/coreprotect/sessions.py @@ -0,0 +1,150 @@ +"""Play sessions rebuilt from CoreProtect's co_session rows. + +Rows are 1 login, 0 logout and 2 ping (our fork logs each online player's +position once a minute, per world, unless player-pings is off). A session +runs from a login to the next newer login. Inside that window, strictly +after the login and before the next one by (time, rowid): + +- the first logout is the recorded end; +- otherwise the last ping is the last time we saw them ("last_observed"): + a crash or unclean stop logs no logout, so this is not a proven end; +- the newest session with no logout is "open" while its last sighting is + within two ping intervals (a heuristic), else "last_observed"; +- with pings off or unknown, an unended newest session is "unknown". + +Rows before a player's first login (history began on 19 September 2026 on +Main) are ignored; the response says where the retained history begins. +""" +from __future__ import annotations + +from dataclasses import dataclass + +from .maps import Maps +from .reader import Reader + +ACTION_LOGOUT = 0 +ACTION_LOGIN = 1 +ACTION_PING = 2 +FRESH_GRACE_SECONDS = 30 +MAX_LIMIT = 100 + +_COLUMNS = "rowid AS rid, time, wid, x, y, z" + + +@dataclass(frozen=True) +class Key: + time: int + rowid: int + + +def _after(key: Key) -> tuple[str, tuple]: + return "time >= ? AND NOT (time = ? AND rowid <= ?)", (key.time, key.time, key.rowid) + + +def _before(key: Key) -> tuple[str, tuple]: + return "time <= ? AND NOT (time = ? AND rowid >= ?)", (key.time, key.time, key.rowid) + + +def _key(row) -> Key: + return Key(row["time"], row["rid"]) + + +def _seek(reader: Reader, ids: list[int], action: int, bounds: list[tuple[str, tuple]], newest: bool): + """The first row after the bounds in time order (or the last, if newest), across a player's ids.""" + order = "DESC" if newest else "ASC" + where = " AND ".join(["user = ?", "action = ?", *(b[0] for b in bounds)]) + found = [] + for uid in ids: + params = (uid, action, *(p for b in bounds for p in b[1])) + found += reader.rows( + f"SELECT {_COLUMNS} FROM co_session WHERE {where} ORDER BY time {order}, rowid {order} LIMIT 1", + params, + ) + if not found: + return None + pick = max if newest else min + return pick(found, key=lambda r: (r["time"], r["rid"])) + + +def fetch(reader: Reader, ids: list[int], before: Key | None, limit: int) -> dict: + """Read one page of sessions, newest first. Plain rows only; build() shapes them.""" + limit = max(1, min(limit, MAX_LIMIT)) + logins = [] + for uid in ids: + where, params = "user = ? AND action = ?", (uid, ACTION_LOGIN) + if before is not None: + cond, extra = _before(before) + where, params = f"{where} AND {cond}", params + extra + logins += reader.rows( + f"SELECT {_COLUMNS} FROM co_session WHERE {where} ORDER BY time DESC, rowid DESC LIMIT ?", + params + (limit + 1,), + ) + logins.sort(key=lambda r: (r["time"], r["rid"]), reverse=True) + has_more = len(logins) > limit + page = logins[:limit] + + windows = [] + for index, login in enumerate(page): + upper = _key(page[index - 1]) if index else before + bounds = [_after(_key(login))] + ([_before(upper)] if upper else []) + logout = _seek(reader, ids, ACTION_LOGOUT, bounds, newest=False) + ping = None if logout else _seek(reader, ids, ACTION_PING, bounds, newest=True) + windows.append({"login": login, "logout": logout, "ping": ping, "newest": upper is None}) + + first_seen = [r["first"] for uid in ids + for r in reader.rows("SELECT MIN(time) AS first FROM co_session WHERE user = ?", (uid,)) + if r["first"] is not None] + history = reader.rows("SELECT MIN(time) AS start FROM co_session") + return { + "windows": windows, + "next": _key(page[-1]) if has_more and page else None, + "first_seen": min(first_seen) if first_seen else None, + "history_start": history[0]["start"] if history else None, + } + + +def last_event(reader: Reader, ids: list[int]): + """The newest session row of any kind across a player's ids, or None.""" + rows = [] + for uid in ids: + rows += reader.rows( + "SELECT rowid AS rid, time, action FROM co_session WHERE user = ? ORDER BY time DESC, rowid DESC LIMIT 1", + (uid,), + ) + return max(rows, key=lambda r: (r["time"], r["rid"])) if rows else None + + +def is_fresh(event, now: int, ping_seconds: int) -> bool: + """Whether a player was seen recently enough to call them online. A heuristic.""" + if event is None or ping_seconds <= 0 or event["action"] == ACTION_LOGOUT: + return False + return now - event["time"] <= 2 * ping_seconds + FRESH_GRACE_SECONDS + + +def _point(row, maps: Maps) -> dict: + return {"time": row["time"], "world": maps.world(row["wid"]), "x": row["x"], "y": row["y"], "z": row["z"]} + + +def build(raw: dict, maps: Maps, now: int, ping_seconds: int) -> dict: + sessions = [] + for window in raw["windows"]: + login, logout, ping = window["login"], window["logout"], window["ping"] + end = None + if logout is not None: + end_kind, end = "logout", _point(logout, maps) + elif window["newest"] and ping_seconds <= 0: + end_kind = "unknown" + elif window["newest"] and is_fresh({**(ping or login), "action": ACTION_PING}, now, ping_seconds): + end_kind = "open" + else: + end_kind = "last_observed" + end = _point(ping, maps) if ping is not None else None + if end is not None: + duration = end["time"] - login["time"] + elif end_kind == "open": + duration = max(0, now - login["time"]) + else: + duration = None + sessions.append({"start": _point(login, maps), "end": end, "end_kind": end_kind, + "duration_seconds": duration}) + return {"sessions": sessions, "first_seen": raw["first_seen"], "history_start": raw["history_start"]} diff --git a/backend/src/coreprotect/test_activity.py b/backend/src/coreprotect/test_activity.py new file mode 100644 index 00000000..c3f1604e --- /dev/null +++ b/backend/src/coreprotect/test_activity.py @@ -0,0 +1,157 @@ +import json +import sqlite3 + +import pytest + +from src.coreprotect import activity, cursors, maps +from src.coreprotect.reader import CoreProtectConfig, Reader + +SCOPE = "main:0615a817-8cb4-4aef-95f7-f6c9bf7611b8" + + +def read(db, ids, kinds=activity.KINDS, before=None, limit=20): + config = CoreProtectConfig(path=str(db.path), server="main", label="", ping_seconds=60) + with Reader(config) as r: + names = maps.get(r) + raw = activity.fetch(r, names, ids, kinds, activity.decode_cursor(SCOPE, before), limit) + return activity.build(raw, names, SCOPE) + + +def test_decodes_each_source(coreprotect): + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + victim = coreprotect.user("Bob", "00000000-0000-0000-0000-000000000002") + coreprotect.block(me, 1, 0) + coreprotect.block(me, 2, 1) + coreprotect.block(me, 3, 2, type_id=2) + coreprotect.block(me, 4, 3, type_id=1) + coreprotect.block(me, 5, 3, type_id=0, data=victim) + coreprotect.block(me, 6, 13, type_id=2) + coreprotect.container(me, 7, 1, amount=3, rolled_back=2) + coreprotect.item(me, 8, 3, amount=2) + coreprotect.interaction(me, 9, 1) + coreprotect.sign(me, 10, 2, "my secret sign") + coreprotect.command(me, 11, "/msg Bob meet me at the secret base") + coreprotect.command(me, 12, "[skill] Fireball (fireball)") + coreprotect.command(me, 13, "[skill] Blink (blink) teleported from 1 2 3 to 4 5 6") + coreprotect.chat(me, 14, "chat is never shown") + coreprotect.session(me, 15, 1) + coreprotect.session(me, 16, 2) + + entries = read(coreprotect, [me])["entries"] + seen = [(e["kind"], e["verb"], e["target"]) for e in entries] + assert seen == [ + ("session", "logged in", None), + ("skill", "teleported with", "Blink (blink)"), + ("skill", "cast", "Fireball (fireball)"), + ("command", "ran", "/msg"), + ("sign", "edited", "sign"), + ("entity", "sheared", "sheep"), + ("item", "picked up", "iron_ingot"), + ("container", "added", "iron_ingot"), + ("spawn", "spawned", "sheep"), + ("kill", "killed", "Bob"), + ("kill", "killed", "cow"), + ("click", "clicked", "oak_door"), + ("block", "placed", "stone"), + ("block", "broke", "stone"), + ] + by_kind = {e["kind"]: e for e in entries} + assert by_kind["container"]["amount"] == 3 + assert by_kind["container"]["rolled_back"] == "rolled back (player inventory)" + assert entries[9]["victim"] == {"minecraft_name": "Bob", "uuid": "00000000-0000-0000-0000-000000000002"} + assert by_kind["block"]["world"] == "TFMC_Map" + text = json.dumps(entries) + for secret in ("meet me", "secret base", "secret sign", "chat is never", "4 5 6"): + assert secret not in text + + +def test_pages_through_same_second_rows_across_tables(coreprotect): + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + expected = [] + for _ in range(3): + expected.append(f"block:{coreprotect.block(me, 100, 0)}") + expected.append(f"container:{coreprotect.container(me, 100, 1)}") + expected.append(f"command:{coreprotect.command(me, 100, '/home')}") + coreprotect.block(me, 50, 1) + got, before = [], None + while True: + page = read(coreprotect, [me], before=before, limit=2) + got += [e["id"] for e in page["entries"]] + before = page["next"] + if not before: + break + assert len(got) == len(set(got)) == 10 + assert set(expected) <= set(got) + assert got[-1].startswith("block:") + + +def test_kind_filters_apply_before_the_limit(coreprotect): + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + coreprotect.command(me, 1, "/spawn") + for t in range(2, 40): + coreprotect.block(me, t, 0) + page = read(coreprotect, [me], kinds=("command",), limit=5) + assert [e["target"] for e in page["entries"]] == ["/spawn"] + assert page["next"] is None + + +def test_missing_fork_tables_are_skipped(coreprotect): + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + coreprotect.block(me, 1, 0) + conn = sqlite3.connect(coreprotect.path) + conn.execute("DROP TABLE co_entity_interaction") + conn.execute("DROP TABLE co_entity_container") + conn.commit() + conn.close() + assert [e["kind"] for e in read(coreprotect, [me])["entries"]] == ["block"] + + +def test_other_players_and_unknown_maps(coreprotect): + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + other = coreprotect.user("Bob", "00000000-0000-0000-0000-000000000002") + coreprotect.block(other, 5, 0) + coreprotect.block(me, 4, 0, type_id=999) + entries = read(coreprotect, [me])["entries"] + assert [e["target"] for e in entries] == ["material #999"] + + +def test_cursors_are_scoped_and_validated(): + token = activity.encode_cursor(SCOPE, activity.Cursor(10, 1, 5)) + assert activity.decode_cursor(SCOPE, token) == activity.Cursor(10, 1, 5) + for bad in ("nonsense", activity.encode_cursor("main:someone-else", activity.Cursor(10, 1, 5)), + cursors.encode(SCOPE, "activity", 10, 99, 5), cursors.encode(SCOPE, "sessions", 10, 1, 5), "x" * 300): + with pytest.raises(cursors.BadCursor): + activity.decode_cursor(SCOPE, bad) + + +def test_parse_kinds(): + assert activity.parse_kinds(None) == activity.KINDS + assert activity.parse_kinds("block, command,block") == ("block", "command") + for bad in ("chat", ",", "block,nope"): + with pytest.raises(ValueError): + activity.parse_kinds(bad) + + +def test_sparse_filters_scan_a_bounded_window(coreprotect, monkeypatch): + monkeypatch.setattr(activity, "SCAN_LIMIT", 5) + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + kills = [] + for t in range(1, 41): + if t % 13 == 0: + kills.append(f"block:{coreprotect.block(me, t, 3, type_id=1)}") + else: + coreprotect.block(me, t, 0) + coreprotect.command(me, 41, "/home") + got, before, pages = [], None, 0 + while True: + page = read(coreprotect, [me], kinds=("kill", "command"), before=before, limit=2) + pages += 1 + got += [e["id"] for e in page["entries"] if e["kind"] == "kill"] + if page["entries"] == [] or len(page["entries"]) < 2: + assert page["searched_to"] is not None or page["next"] is None + before = page["next"] + if not before: + break + assert sorted(got) == sorted(kills) + assert len(got) == len(set(got)) + assert pages > 3 diff --git a/backend/src/coreprotect/test_reader.py b/backend/src/coreprotect/test_reader.py new file mode 100644 index 00000000..48c16212 --- /dev/null +++ b/backend/src/coreprotect/test_reader.py @@ -0,0 +1,115 @@ +import sqlite3 +import threading +import time + +import pytest + +from src.coreprotect import reader as reader_mod +from src.coreprotect.reader import Budget, CoreProtectConfig, Reader, Unavailable + + +def config(path, ping=60): + return CoreProtectConfig(path=str(path) if path else None, server="main", label="", ping_seconds=ping) + + +def test_config_from_env(monkeypatch): + monkeypatch.setenv("COREPROTECT_DB", " /coreprotect/main/database.db ") + monkeypatch.setenv("COREPROTECT_PING_SECONDS", "nonsense") + monkeypatch.delenv("COREPROTECT_SERVER", raising=False) + loaded = CoreProtectConfig.from_env() + assert loaded.path == "/coreprotect/main/database.db" + assert loaded.server == "main" + assert loaded.ping_seconds == 0 + + +def test_unconfigured_and_missing(tmp_path): + with pytest.raises(Unavailable) as exc: + Reader(config(None)).__enter__() + assert exc.value.code == "not_configured" + with pytest.raises(Unavailable) as exc: + Reader(config(tmp_path / "nope.db")).__enter__() + assert exc.value.code == "missing" + + +def test_reads_are_read_only(coreprotect): + with Reader(config(coreprotect.path)) as r: + assert r.rows("SELECT world FROM co_world ORDER BY id")[0]["world"] == "TFMC_Map" + with pytest.raises(Unavailable): + r.rows("INSERT INTO co_world (id, world) VALUES (9, 'x')") + + +def test_slot_is_released_after_errors(coreprotect): + for _ in range(3): + with pytest.raises(Unavailable): + with Reader(config(coreprotect.path)) as r: + r.rows("SELECT * FROM no_such_table") + with Reader(config(coreprotect.path), Budget(0.2)) as r: + assert r.rows("SELECT 1 AS one")[0]["one"] == 1 + + +def test_one_reader_at_a_time(coreprotect): + entered, release = threading.Event(), threading.Event() + + def hold(): + with Reader(config(coreprotect.path), Budget(5)): + entered.set() + release.wait(5) + + worker = threading.Thread(target=hold) + worker.start() + try: + assert entered.wait(5) + started = time.monotonic() + with pytest.raises(Unavailable) as exc: + Reader(config(coreprotect.path), Budget(0.1)).__enter__() + assert exc.value.code == "busy" + assert time.monotonic() - started < 0.5 + finally: + release.set() + worker.join() + + +def test_writer_lock_is_bounded_by_budget(coreprotect): + writer = sqlite3.connect(coreprotect.path, isolation_level=None) + writer.execute("BEGIN EXCLUSIVE") + try: + started = time.monotonic() + with pytest.raises(Unavailable) as exc: + with Reader(config(coreprotect.path), Budget(0.5)) as r: + r.rows("SELECT COUNT(*) FROM co_user") + assert exc.value.code in {"busy", "timeout"} + assert time.monotonic() - started < 1.0 + finally: + writer.execute("ROLLBACK") + writer.close() + + +def test_long_statement_is_interrupted(coreprotect): + started = time.monotonic() + with pytest.raises(Unavailable) as exc: + with Reader(config(coreprotect.path), Budget(0.2)) as r: + r.rows("WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n) SELECT MAX(i) FROM n") + assert exc.value.code == "timeout" + assert time.monotonic() - started < 1.0 + + +def test_no_lock_survives_a_reader(coreprotect): + with Reader(config(coreprotect.path)) as r: + r.rows("SELECT * FROM co_world") + writer = sqlite3.connect(coreprotect.path, timeout=0, isolation_level=None) + try: + writer.execute("BEGIN EXCLUSIVE") + writer.execute("ROLLBACK") + finally: + writer.close() + + +def test_spent_budget_stops_before_the_next_statement(coreprotect): + budget = Budget(0.05) + with Reader(config(coreprotect.path), budget) as r: + time.sleep(0.06) + with pytest.raises(Unavailable) as exc: + r.rows("SELECT 1") + assert exc.value.code == "timeout" + assert reader_mod._GATE.acquire(timeout=0.1) + reader_mod._GATE.release() diff --git a/backend/src/coreprotect/test_sessions.py b/backend/src/coreprotect/test_sessions.py new file mode 100644 index 00000000..9835f99c --- /dev/null +++ b/backend/src/coreprotect/test_sessions.py @@ -0,0 +1,132 @@ +from src.coreprotect import maps, sessions +from src.coreprotect.reader import CoreProtectConfig, Reader + +LOGOUT, LOGIN, PING = 0, 1, 2 +NOW = 1_000_000 + + +def read(db, ids, before=None, limit=20, ping=60, now=NOW): + config = CoreProtectConfig(path=str(db.path), server="main", label="", ping_seconds=ping) + with Reader(config) as r: + raw = sessions.fetch(r, ids, before, limit) + names = maps.get(r) + return sessions.build(raw, names, now, ping), raw + + +def test_recorded_and_crashed_sessions(coreprotect): + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + coreprotect.session(me, 100, LOGIN) + coreprotect.session(me, 160, PING) + coreprotect.session(me, 200, LOGOUT, x=7) + # Crash: pings, then no logout before the next login. + coreprotect.session(me, 300, LOGIN) + coreprotect.session(me, 360, PING, x=3) + coreprotect.session(me, 420, PING, x=4) + # Crash before any ping. + coreprotect.session(me, 500, LOGIN) + coreprotect.session(me, 600, LOGIN) + coreprotect.session(me, 700, LOGOUT) + + built, _ = read(coreprotect, [me]) + rows = [(s["start"]["time"], s["end_kind"], s["end"] and s["end"]["time"], s["duration_seconds"]) + for s in built["sessions"]] + assert rows == [ + (600, "logout", 700, 100), + (500, "last_observed", None, None), + (300, "last_observed", 420, 120), + (100, "logout", 200, 100), + ] + assert built["sessions"][3]["end"]["x"] == 7 + assert built["sessions"][2]["end"]["x"] == 4 + assert built["sessions"][0]["start"]["world"] == "TFMC_Map" + assert built["first_seen"] == 100 + assert built["history_start"] == 100 + + +def test_newest_session_open_or_stale(coreprotect): + me = coreprotect.user("A", "00000000-0000-0000-0000-000000000001") + coreprotect.session(me, NOW - 600, LOGIN) + coreprotect.session(me, NOW - 60, PING) + built, _ = read(coreprotect, [me]) + assert built["sessions"][0]["end_kind"] == "open" + assert built["sessions"][0]["duration_seconds"] == 600 + + built, _ = read(coreprotect, [me], now=NOW + 3600) + assert built["sessions"][0]["end_kind"] == "last_observed" + assert built["sessions"][0]["end"]["time"] == NOW - 60 + + built, _ = read(coreprotect, [me], ping=0) + assert built["sessions"][0]["end_kind"] == "unknown" + assert built["sessions"][0]["end"] is None + + +def test_just_joined_counts_as_open(coreprotect): + me = coreprotect.user("A", "00000000-0000-0000-0000-000000000001") + coreprotect.session(me, NOW - 20, LOGIN) + built, _ = read(coreprotect, [me]) + assert built["sessions"][0]["end_kind"] == "open" + + +def test_same_second_ordering_uses_rowid(coreprotect): + me = coreprotect.user("A", "00000000-0000-0000-0000-000000000001") + coreprotect.session(me, 100, LOGIN) + coreprotect.session(me, 100, LOGOUT) + coreprotect.session(me, 100, LOGIN) + coreprotect.session(me, 150, LOGOUT) + built, _ = read(coreprotect, [me]) + assert [(s["start"]["time"], s["end"]["time"]) for s in built["sessions"]] == [(100, 150), (100, 100)] + + +def test_pages_keep_windows_bounded(coreprotect): + me = coreprotect.user("A", "00000000-0000-0000-0000-000000000001") + for start in (100, 300, 500, 700, 900): + coreprotect.session(me, start, LOGIN) + if start != 500: + coreprotect.session(me, start + 50, LOGOUT) + first, raw = read(coreprotect, [me], limit=2) + assert [s["start"]["time"] for s in first["sessions"]] == [900, 700] + second, raw = read(coreprotect, [me], before=raw["next"], limit=2) + # 500 crashed: the 750 logout belongs to the newer session, so it is not borrowed. + assert [(s["start"]["time"], s["end_kind"]) for s in second["sessions"]] == [(500, "last_observed"), (300, "logout")] + third, raw = read(coreprotect, [me], before=raw["next"], limit=2) + assert [s["start"]["time"] for s in third["sessions"]] == [100] + assert raw["next"] is None + + +def test_orphans_before_first_login_are_ignored(coreprotect): + me = coreprotect.user("A", "00000000-0000-0000-0000-000000000001") + coreprotect.session(me, 50, PING) + coreprotect.session(me, 60, LOGOUT) + coreprotect.session(me, 100, LOGIN) + coreprotect.session(me, 200, LOGOUT) + built, _ = read(coreprotect, [me]) + assert len(built["sessions"]) == 1 + assert built["first_seen"] == 50 + + +def test_other_players_rows_never_leak(coreprotect): + me = coreprotect.user("A", "00000000-0000-0000-0000-000000000001") + other = coreprotect.user("B", "00000000-0000-0000-0000-000000000002") + coreprotect.session(me, 100, LOGIN) + coreprotect.session(other, 150, LOGOUT) + coreprotect.session(other, 160, PING) + coreprotect.session(me, 300, LOGIN) + coreprotect.session(me, 400, LOGOUT) + built, _ = read(coreprotect, [me]) + assert [(s["start"]["time"], s["end_kind"], s["end"]) for s in built["sessions"]][1] == (100, "last_observed", None) + + +def test_last_event_and_freshness(coreprotect): + me = coreprotect.user("A", "00000000-0000-0000-0000-000000000001") + config = CoreProtectConfig(path=str(coreprotect.path), server="main", label="", ping_seconds=60) + with Reader(config) as r: + assert sessions.last_event(r, [me]) is None + coreprotect.session(me, NOW - 30, LOGIN) + with Reader(config) as r: + event = sessions.last_event(r, [me]) + assert sessions.is_fresh(event, NOW, 60) + assert not sessions.is_fresh(event, NOW + 1000, 60) + assert not sessions.is_fresh(event, NOW, 0) + coreprotect.session(me, NOW - 10, LOGOUT) + with Reader(config) as r: + assert not sessions.is_fresh(sessions.last_event(r, [me]), NOW, 60) diff --git a/backend/src/coreprotect/testdata/schema.sql b/backend/src/coreprotect/testdata/schema.sql new file mode 100644 index 00000000..a4dad7c2 --- /dev/null +++ b/backend/src/coreprotect/testdata/schema.sql @@ -0,0 +1,75 @@ +-- CoreProtect fork 25.3.0 SQLite schema, dumped from TFMCDev01 on 2026-10-06 for tests. +CREATE TABLE co_art_map (id INTEGER, art TEXT); +CREATE TABLE co_block (time INTEGER, user INTEGER, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, type INTEGER, data INTEGER, meta BLOB, blockdata BLOB, action INTEGER, rolled_back INTEGER); +CREATE TABLE co_chat (time INTEGER, user INTEGER, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, message TEXT); +CREATE TABLE co_command (time INTEGER, user INTEGER, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, message TEXT); +CREATE TABLE co_container (time INTEGER, user INTEGER, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, type INTEGER, data INTEGER, amount INTEGER, metadata BLOB, action INTEGER, rolled_back INTEGER); +CREATE TABLE co_entity_container (time INTEGER, user INTEGER, entity_spawn_rowid INTEGER NOT NULL, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, type INTEGER, data INTEGER, amount INTEGER, metadata BLOB, action INTEGER, rolled_back INTEGER); +CREATE TABLE co_entity_interaction (time INTEGER, user INTEGER, entity_spawn_rowid INTEGER NOT NULL, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, type INTEGER, action INTEGER, metadata BLOB, rolled_back INTEGER); +CREATE TABLE co_item (time INTEGER, user INTEGER, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, type INTEGER, data BLOB, amount INTEGER, action INTEGER, rolled_back INTEGER); +CREATE TABLE co_database_lock (status INTEGER, time INTEGER); +CREATE TABLE co_entity (id INTEGER PRIMARY KEY ASC, time INTEGER, data BLOB); +CREATE TABLE co_entity_spawn (id INTEGER PRIMARY KEY ASC, time INTEGER, block_rowid INTEGER, kill_rowid INTEGER, uuid TEXT UNIQUE, wid INTEGER, current_wid INTEGER, origin_x REAL, origin_y REAL, origin_z REAL, x REAL, y REAL, z REAL, yaw REAL, pitch REAL, data BLOB, removed INTEGER); +CREATE TABLE co_entity_map (id INTEGER, entity TEXT); +CREATE TABLE co_material_map (id INTEGER, material TEXT); +CREATE TABLE co_blockdata_map (id INTEGER, data TEXT); +CREATE TABLE co_session (time INTEGER, user INTEGER, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, action INTEGER); +CREATE TABLE co_sign (time INTEGER, user INTEGER, wid INTEGER, x INTEGER, y INTEGER, z INTEGER, action INTEGER, color INTEGER, color_secondary INTEGER, data INTEGER, waxed INTEGER, face INTEGER, line_1 TEXT, line_2 TEXT, line_3 TEXT, line_4 TEXT, line_5 TEXT, line_6 TEXT, line_7 TEXT, line_8 TEXT); +CREATE TABLE co_skull (id INTEGER PRIMARY KEY ASC, time INTEGER, owner TEXT, skin TEXT); +CREATE TABLE co_user (id INTEGER PRIMARY KEY ASC, time INTEGER, user TEXT, uuid TEXT); +CREATE TABLE co_username_log (id INTEGER PRIMARY KEY ASC, time INTEGER, uuid TEXT, user TEXT); +CREATE TABLE co_version (time INTEGER, version TEXT); +CREATE TABLE co_world (id INTEGER, world TEXT); +CREATE INDEX art_map_id_index ON co_art_map(id); +CREATE INDEX block_index ON co_block(wid,x,z,time); +CREATE INDEX block_user_index ON co_block(user,time); +CREATE INDEX block_type_index ON co_block(type,time); +CREATE INDEX blockdata_map_id_index ON co_blockdata_map(id); +CREATE INDEX chat_index ON co_chat(time); +CREATE INDEX chat_user_index ON co_chat(user,time); +CREATE INDEX chat_wid_index ON co_chat(wid,x,z,time); +CREATE INDEX command_index ON co_command(time); +CREATE INDEX command_user_index ON co_command(user,time); +CREATE INDEX command_wid_index ON co_command(wid,x,z,time); +CREATE INDEX container_index ON co_container(wid,x,z,time); +CREATE INDEX container_user_index ON co_container(user,time); +CREATE INDEX container_type_index ON co_container(type,time); +CREATE INDEX entity_container_index ON co_entity_container(wid,x,z,time); +CREATE INDEX entity_container_spawn_index ON co_entity_container(entity_spawn_rowid,time); +CREATE INDEX entity_container_user_index ON co_entity_container(user,time); +CREATE INDEX entity_container_type_index ON co_entity_container(type,time); +CREATE INDEX entity_interaction_index ON co_entity_interaction(wid,x,z,time); +CREATE INDEX entity_interaction_spawn_index ON co_entity_interaction(entity_spawn_rowid,time); +CREATE INDEX entity_interaction_user_index ON co_entity_interaction(user,time); +CREATE INDEX entity_interaction_type_index ON co_entity_interaction(type,time); +CREATE INDEX entity_interaction_action_index ON co_entity_interaction(action,time); +CREATE INDEX item_index ON co_item(wid,x,z,time); +CREATE INDEX item_user_index ON co_item(user,time); +CREATE INDEX item_type_index ON co_item(type,time); +CREATE INDEX entity_map_id_index ON co_entity_map(id); +CREATE INDEX entity_spawn_time_index ON co_entity_spawn(time); +CREATE INDEX entity_spawn_wid_index ON co_entity_spawn(wid); +CREATE INDEX entity_spawn_current_location_index ON co_entity_spawn(current_wid,x,z); +CREATE UNIQUE INDEX entity_spawn_kill_rowid_index ON co_entity_spawn(kill_rowid); +CREATE INDEX material_map_id_index ON co_material_map(id); +CREATE INDEX session_index ON co_session(wid,x,z,time); +CREATE INDEX session_action_index ON co_session(action,time); +CREATE INDEX session_user_index ON co_session(user,time); +CREATE INDEX session_time_index ON co_session(time); +CREATE INDEX sign_index ON co_sign(wid,x,z,time); +CREATE INDEX sign_user_index ON co_sign(user,time); +CREATE INDEX sign_time_index ON co_sign(time); +CREATE INDEX user_index ON co_user(user); +CREATE INDEX uuid_index ON co_user(uuid); +CREATE INDEX username_log_uuid_index ON co_username_log(uuid,user); +CREATE INDEX world_id_index ON co_world(id); +CREATE INDEX chat_message_prefix_index ON co_chat(substr(message,1,16) COLLATE NOCASE); +CREATE INDEX command_message_prefix_index ON co_command(substr(message,1,16) COLLATE NOCASE); +CREATE INDEX sign_line_1_prefix_index ON co_sign(substr(line_1,1,16) COLLATE NOCASE); +CREATE INDEX sign_line_2_prefix_index ON co_sign(substr(line_2,1,16) COLLATE NOCASE); +CREATE INDEX sign_line_3_prefix_index ON co_sign(substr(line_3,1,16) COLLATE NOCASE); +CREATE INDEX sign_line_4_prefix_index ON co_sign(substr(line_4,1,16) COLLATE NOCASE); +CREATE INDEX sign_line_5_prefix_index ON co_sign(substr(line_5,1,16) COLLATE NOCASE); +CREATE INDEX sign_line_6_prefix_index ON co_sign(substr(line_6,1,16) COLLATE NOCASE); +CREATE INDEX sign_line_7_prefix_index ON co_sign(substr(line_7,1,16) COLLATE NOCASE); +CREATE INDEX sign_line_8_prefix_index ON co_sign(substr(line_8,1,16) COLLATE NOCASE); diff --git a/backend/src/coreprotect/testing.py b/backend/src/coreprotect/testing.py new file mode 100644 index 00000000..866760bf --- /dev/null +++ b/backend/src/coreprotect/testing.py @@ -0,0 +1,72 @@ +"""Test helpers: a CoreProtect-shaped SQLite file with the fork's real schema.""" +import sqlite3 +from pathlib import Path + +SCHEMA = Path(__file__).with_name("testdata") / "schema.sql" + + +class CoreProtectDb: + """A CoreProtect-shaped SQLite file with the fork's real schema.""" + + def __init__(self, path: Path): + self.path = path + conn = sqlite3.connect(path) + conn.executescript(SCHEMA.read_text(encoding="utf-8")) + conn.executemany("INSERT INTO co_world (id, world) VALUES (?, ?)", [(1, "TFMC_Map"), (2, "TFMC_Map_the_end")]) + conn.executemany("INSERT INTO co_material_map (id, material) VALUES (?, ?)", + [(1, "minecraft:stone"), (2, "minecraft:oak_door"), (3, "minecraft:iron_ingot")]) + conn.executemany("INSERT INTO co_entity_map (id, entity) VALUES (?, ?)", [(1, "cow"), (2, "sheep")]) + conn.commit() + conn.close() + + def execute(self, sql: str, params: tuple = ()) -> int: + conn = sqlite3.connect(self.path) + try: + cursor = conn.execute(sql, params) + conn.commit() + return cursor.lastrowid + finally: + conn.close() + + def user(self, name: str, uuid: str | None, time: int = 1_000) -> int: + return self.execute("INSERT INTO co_user (time, user, uuid) VALUES (?, ?, ?)", (time, name, uuid)) + + def session(self, user: int, time: int, action: int, x: int = 0) -> int: + return self.execute( + "INSERT INTO co_session (time, user, wid, x, y, z, action) VALUES (?, ?, 1, ?, 64, 0, ?)", + (time, user, x, action)) + + def block(self, user: int, time: int, action: int, type_id: int = 1, data: int = 0, rolled_back: int = 0) -> int: + return self.execute( + "INSERT INTO co_block (time, user, wid, x, y, z, type, data, meta, blockdata, action, rolled_back) " + "VALUES (?, ?, 1, 10, 64, 20, ?, ?, NULL, NULL, ?, ?)", (time, user, type_id, data, action, rolled_back)) + + def container(self, user: int, time: int, action: int, amount: int = 1, rolled_back: int = 0) -> int: + return self.execute( + "INSERT INTO co_container (time, user, wid, x, y, z, type, data, amount, metadata, action, rolled_back) " + "VALUES (?, ?, 1, 1, 2, 3, 3, 0, ?, X'00', ?, ?)", (time, user, amount, action, rolled_back)) + + def item(self, user: int, time: int, action: int, amount: int = 1) -> int: + return self.execute( + "INSERT INTO co_item (time, user, wid, x, y, z, type, data, amount, action, rolled_back) " + "VALUES (?, ?, 1, 1, 2, 3, 3, X'00', ?, ?, 0)", (time, user, amount, action)) + + def command(self, user: int, time: int, message: str) -> int: + return self.execute( + "INSERT INTO co_command (time, user, wid, x, y, z, message) VALUES (?, ?, 1, 5, 6, 7, ?)", + (time, user, message)) + + def chat(self, user: int, time: int, message: str) -> int: + return self.execute( + "INSERT INTO co_chat (time, user, wid, x, y, z, message) VALUES (?, ?, 1, 5, 6, 7, ?)", + (time, user, message)) + + def sign(self, user: int, time: int, action: int, text: str) -> int: + return self.execute( + "INSERT INTO co_sign (time, user, wid, x, y, z, action, color, color_secondary, data, waxed, face, " + "line_1) VALUES (?, ?, 1, 1, 2, 3, ?, 0, 0, 0, 0, 0, ?)", (time, user, action, text)) + + def interaction(self, user: int, time: int, action: int, type_id: int = 2) -> int: + return self.execute( + "INSERT INTO co_entity_interaction (time, user, entity_spawn_rowid, wid, x, y, z, type, action, " + "metadata, rolled_back) VALUES (?, ?, 1, 1, 1, 2, 3, ?, ?, NULL, 0)", (time, user, type_id, action)) From 34543defc8eb4eb1aa7f68e646babba7860ce6bf Mon Sep 17 00:00:00 2001 From: Ryan Barlow <7389646+ryanbarlow97@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:04:20 +0000 Subject: [PATCH 2/6] Seek same-second bursts, decode rollback flags and report unreadable databases Review fixes: a rowid bound now always sits with its second pinned by equality, so a WorldEdit-sized burst in one second is seeked rather than walked; rolled_back is decoded as its two flags for every source; a CoreProtect file that cannot be opened is reported as unavailable instead of failing the request. The gate script gains a Main-sized synthetic database and covers directory rebuilds and second pages. Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/benchmarks/coreprotect_reader.py | 104 ++++++++++++++++++++-- backend/src/coreprotect/activity.py | 106 +++++++++++++++++------ backend/src/coreprotect/reader.py | 6 ++ backend/src/coreprotect/test_activity.py | 69 +++++++++++++++ backend/src/coreprotect/test_reader.py | 20 +++++ 5 files changed, 268 insertions(+), 37 deletions(-) diff --git a/backend/benchmarks/coreprotect_reader.py b/backend/benchmarks/coreprotect_reader.py index 136260d1..d00c92e0 100644 --- a/backend/benchmarks/coreprotect_reader.py +++ b/backend/benchmarks/coreprotect_reader.py @@ -8,6 +8,12 @@ counted rather than fatal. The time each request spends in CoreProtect is how long a production request would hold its read lock. +synth + Builds a disposable database with the fork's schema at Main's size on + 2026-10-06 (23.6M block rows, 3.0M container, 3.3M item, 242 players, + skewed so a few builders own most rows, with one-second bursts), for + contend. Copying the live file would lock CoreProtect for the whole copy. + contend Runs a CoreProtect-like writer (batched inserts, small cache so pages spill mid-transaction, a 3 s busy timeout like sqlite-jdbc) in another @@ -58,6 +64,73 @@ def _ids(config): return [row["id"] for row in rows] +SYNTH_ROWS = {"co_block": 23_600_000, "co_container": 2_960_000, "co_item": 3_300_000, "co_command": 221_000, + "co_sign": 8_200, "co_entity_interaction": 92_000} +SYNTH_PLAYERS = 242 + + +def synth(args) -> None: + schema = (Path(__file__).resolve().parents[1] / "src/coreprotect/testdata/schema.sql").read_text() + conn = sqlite3.connect(args.database, isolation_level=None) + conn.execute("PRAGMA journal_mode = OFF") + conn.execute("PRAGMA synchronous = OFF") + conn.executescript(schema) + indexes = [row[0] for row in conn.execute("SELECT sql FROM sqlite_master WHERE type = 'index' AND sql IS NOT NULL")] + for (name,) in conn.execute("SELECT name FROM sqlite_master WHERE type = 'index' AND sql IS NOT NULL").fetchall(): + conn.execute(f"DROP INDEX {name}") + conn.executemany("INSERT INTO co_user (id, time, user, uuid) VALUES (?, 0, ?, ?)", + [(i, f"Player{i}", f"00000000-0000-4000-8000-{i:012d}") for i in range(1, SYNTH_PLAYERS + 1)]) + conn.execute("INSERT INTO co_world (id, world) VALUES (1, 'TFMC_Map')") + conn.executemany("INSERT INTO co_material_map (id, material) VALUES (?, ?)", [(i, f"minecraft:m{i}") for i in range(1, 1316)]) + conn.executemany("INSERT INTO co_entity_map (id, entity) VALUES (?, ?)", [(i, f"e{i}") for i in range(1, 36)]) + # A skewed player (the product of two uniforms favours low ids) and a time that advances with the row, + # in bursts of up to 500 rows sharing one second, over 17 days. + user = f"1 + CAST({SYNTH_PLAYERS - 1} * (abs(random()) / 9.3e18) * (abs(random()) / 9.3e18) AS INTEGER)" + t0 = 1_789_836_708 + + def fill(table, columns, values): + n = SYNTH_ROWS[table] + conn.execute( + f"INSERT INTO {table} ({columns}) WITH RECURSIVE c(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM c WHERE i < {n}) " + f"SELECT {values} FROM c") + print(f"{table}: {n} rows", flush=True) + + when = f"{t0} + (i * 1468800 / {{n}}) - (i * 1468800 / {{n}}) % (1 + abs(random()) % 3)" + fill("co_block", "time, user, wid, x, y, z, type, data, meta, blockdata, action, rolled_back", + f"{when.format(n=SYNTH_ROWS['co_block'])}, {user}, 1, abs(random()) % 9000, 64, abs(random()) % 9000, " + "1 + abs(random()) % 1315, 0, NULL, randomblob(150), " + "CASE abs(random()) % 100 WHEN 0 THEN 3 WHEN 1 THEN 13 WHEN 2 THEN 2 ELSE abs(random()) % 2 END, 0") + fill("co_container", "time, user, wid, x, y, z, type, data, amount, metadata, action, rolled_back", + f"{when.format(n=SYNTH_ROWS['co_container'])}, {user}, 1, 1, 64, 1, 1 + abs(random()) % 1315, 0, " + "1 + abs(random()) % 64, randomblob(40), abs(random()) % 2, 0") + fill("co_item", "time, user, wid, x, y, z, type, data, amount, action, rolled_back", + f"{when.format(n=SYNTH_ROWS['co_item'])}, {user}, 1, 1, 64, 1, 1 + abs(random()) % 1315, randomblob(40), " + "1 + abs(random()) % 64, abs(random()) % 13, 0") + fill("co_command", "time, user, wid, x, y, z, message", + f"{when.format(n=SYNTH_ROWS['co_command'])}, {user}, 1, 1, 64, 1, " + "CASE abs(random()) % 10 WHEN 0 THEN '[skill] Blink (blink)' ELSE '/home base ' || hex(randomblob(8)) END") + fill("co_sign", "time, user, wid, x, y, z, action, color, color_secondary, data, waxed, face, line_1", + f"{when.format(n=SYNTH_ROWS['co_sign'])}, {user}, 1, 1, 64, 1, abs(random()) % 3, 0, 0, 0, 0, 0, 'x'") + fill("co_entity_interaction", "time, user, entity_spawn_rowid, wid, x, y, z, type, action, metadata, rolled_back", + f"{when.format(n=SYNTH_ROWS['co_entity_interaction'])}, {user}, 1, 1, 1, 64, 1, 1 + abs(random()) % 35, " + "abs(random()) % 4, NULL, 0") + # Sessions: each player logs in, pings once a minute for a while, and logs out (some never do). + conn.execute( + "INSERT INTO co_session (time, user, wid, x, y, z, action) " + "WITH RECURSIVE s(n, u, t) AS (SELECT 1, 1, 0 UNION ALL SELECT n + 1, 1 + n % 242, " + f"(n / 242) * 21600 + abs(random()) % 3600 FROM s WHERE n < 16000), " + "p(n, u, t, k, len) AS (SELECT n, u, t, 0, 1 + abs(random()) % 90 FROM s UNION ALL " + "SELECT n, u, t, k + 1, len FROM p WHERE k <= len) " + f"SELECT {t0} + t + k * 60, u, 1, 0, 64, 0, CASE WHEN k = 0 THEN 1 WHEN k > len THEN " + "(CASE WHEN n % 7 = 0 THEN 2 ELSE 0 END) ELSE 2 END FROM p") + print(f"co_session: {conn.execute('SELECT COUNT(*) FROM co_session').fetchone()[0]} rows", flush=True) + for sql in indexes: + conn.execute(sql) + print("indexes built", flush=True) + conn.execute("PRAGMA journal_mode = DELETE") + conn.close() + + def _requests(config, uid): """The CoreProtect part of each staff request for one player.""" def profile(r): @@ -73,9 +146,25 @@ def run(r): activity.fetch(r, maps.get(r), [uid], kinds, None, 20) return run + def second_pages(r): + first = sessions.fetch(r, [uid], None, 20) + if first["next"]: + sessions.fetch(r, [uid], first["next"], 20) + names = maps.get(r) + page = activity.fetch(r, names, [uid], activity.KINDS, None, 20) + if page["next"]: + activity.fetch(r, names, [uid], activity.KINDS, page["next"], 20) + return [("profile", profile), ("sessions", session_page), ("activity", feed(activity.KINDS)), ("activity:skill", feed(("skill",))), ("activity:sign", feed(("sign",))), - ("activity:kill", feed(("kill",)))] + ("activity:kill", feed(("kill",))), ("second pages", second_pages)] + + +def _directory(r): + rows = r.rows("SELECT id FROM co_user WHERE uuid IS NOT NULL") + r.rows("SELECT uuid, user FROM co_username_log") + for row in rows: + sessions.last_event(r, [row["id"]]) def _timed(config, work) -> tuple[float, str | None]: @@ -111,14 +200,8 @@ def hold(args) -> None: if error: errors[f"{name}:{error}"] = errors.get(f"{name}:{error}", 0) + 1 - def directory(r): - rows = r.rows("SELECT id FROM co_user WHERE uuid IS NOT NULL") - r.rows("SELECT uuid, user FROM co_username_log") - for row in rows: - sessions.last_event(r, [row["id"]]) - for _ in range(3): - seconds, error = _timed(config, directory) + seconds, error = _timed(config, _directory) by_kind.setdefault("directory", []).append(seconds) if error: errors[f"directory:{error}"] = errors.get(f"directory:{error}", 0) + 1 @@ -165,6 +248,8 @@ def _hammer(path, seconds, out): config = _config(path) ids = _ids(config) requests = [work for uid in ids for _, work in _requests(config, uid)] + # A directory rebuild every so often, as its cache expires. + requests[::50] = [_directory] * len(requests[::50]) done, errors, i = 0, {}, 0 stop = time.monotonic() + seconds while time.monotonic() < stop: @@ -202,13 +287,14 @@ def main() -> None: h.add_argument("database") h.add_argument("--no-lock", action="store_true") h.add_argument("--players", type=int, default=0) + sub.add_parser("synth").add_argument("database") c = sub.add_parser("contend") c.add_argument("database") c.add_argument("--seconds", type=float, default=20) c.add_argument("--batch", type=int, default=500) c.add_argument("--readers", type=int, default=1) args = parser.parse_args() - hold(args) if args.command == "hold" else contend(args) + {"hold": hold, "synth": synth, "contend": contend}[args.command](args) if __name__ == "__main__": diff --git a/backend/src/coreprotect/activity.py b/backend/src/coreprotect/activity.py index 2787d7e1..4f3c94e3 100644 --- a/backend/src/coreprotect/activity.py +++ b/backend/src/coreprotect/activity.py @@ -98,14 +98,67 @@ def parse_kinds(text: str | None) -> tuple[str, ...]: return chosen -def _boundary(source: Source, cursor: Cursor | None) -> tuple[str, tuple]: +def _upper(source: Source, cursor: Cursor | None) -> tuple | None: + """This source's bound below the cursor: ("le", t), ("lt", t) or ("before", t, rowid).""" if cursor is None: - return "", () + return None if source.rank < cursor.rank: - return " AND time <= ?", (cursor.time,) + return ("le", cursor.time) if source.rank > cursor.rank: - return " AND time < ?", (cursor.time,) - return " AND time <= ? AND NOT (time = ? AND rowid >= ?)", (cursor.time, cursor.time, cursor.rowid) + return ("lt", cursor.time) + return ("before", cursor.time, cursor.rowid) + + +def _segments(upper: tuple | None, lower: tuple[int, int] | None) -> list[tuple[str, tuple]]: + """Disjoint WHERE fragments covering lower <= (time, rowid) < upper, newest first. + + A rowid bound only ever appears with the second pinned by equality, which + the (user, time) index seeks. Written as "time <= t AND NOT (time = t AND + rowid >= id)", SQLite would walk every row in that second first, and one + WorldEdit paste can put hundreds of thousands of rows in one second. + """ + # Each top segment: (fragment, params, pinned second, does a second fall inside it). + if upper is None: + tops = [("", (), None, lambda t: True)] + elif upper[0] == "le": + tops = [(" AND time <= ?", (upper[1],), None, lambda t: t <= upper[1])] + elif upper[0] == "lt": + tops = [(" AND time < ?", (upper[1],), None, lambda t: t < upper[1])] + else: + tops = [(" AND time = ? AND rowid < ?", (upper[1], upper[2]), upper[1], None), + (" AND time < ?", (upper[1],), None, lambda t: t < upper[1])] + if lower is None: + return [(sql, params) for sql, params, _, _ in tops] + low_time, low_rowid = lower + out = [] + for sql, params, pinned, contains in tops: + if pinned is None: + out.append((f"{sql} AND time > ?", params + (low_time,))) + # On its own: next to a time range, SQLite would use the range and sort. + if contains(low_time): + out.append((" AND time = ? AND rowid >= ?", (low_time, low_rowid))) + elif pinned == low_time: + out.append((f"{sql} AND rowid >= ?", params + (low_rowid,))) + elif pinned > low_time: + out.append((sql, params)) + return out + + +def _edge(reader: Reader, table: str, uid: int, upper: tuple | None): + """The SCAN_LIMIT-th row below the bound, read from the index alone, or None if there are fewer.""" + skip = SCAN_LIMIT - 1 + segments = _segments(upper, None) + for index, (sql, params) in enumerate(segments): + found = reader.rows( + f"SELECT time, rowid AS rid FROM {table} WHERE user = ?{sql} " + "ORDER BY time DESC, rowid DESC LIMIT 1 OFFSET ?", (uid, *params, skip)) + if found: + return dict(found[0]) + if index + 1 < len(segments): + skip -= reader.rows( + f"SELECT COUNT(*) AS n FROM (SELECT 1 FROM {table} WHERE user = ?{sql} LIMIT ?)", + (uid, *params, skip))[0]["n"] + return None def _key(source: Source, row: dict) -> Cursor: @@ -125,27 +178,23 @@ def fetch(reader: Reader, maps: Maps, ids: list[int], kinds: tuple[str, ...], cu conditions = [cond for kind, cond in source.kinds.items() if kind in kinds] if not conditions: continue - bound, bound_params = _boundary(source, cursor) - # The oldest row this page may examine, found from the index alone. - edge_sql = ( - f"SELECT time, rowid AS rid FROM {source.table} WHERE user = ?{bound} " - f"ORDER BY time DESC, rowid DESC LIMIT 1 OFFSET {SCAN_LIMIT - 1}" + upper = _upper(source, cursor) + select = ( + f"SELECT rowid AS rid, time, wid, x, y, z, {source.columns} FROM {source.table} " + f"WHERE user = ? AND ({' OR '.join(conditions)})" ) for uid in ids: - edge = reader.rows(edge_sql, (uid, *bound_params)) - floor, floor_params = "", () - if edge: - floor = " AND time >= ? AND NOT (time = ? AND rowid < ?)" - floor_params = (edge[0]["time"], edge[0]["time"], edge[0]["rid"]) - found = [dict(row) for row in reader.rows( - f"SELECT rowid AS rid, time, wid, x, y, z, {source.columns} FROM {source.table} " - f"WHERE user = ? AND ({' OR '.join(conditions)}){bound}{floor} " - "ORDER BY time DESC, rowid DESC LIMIT ?", - (uid, *bound_params, *floor_params, limit + 1), - )] + edge = _edge(reader, source.table, uid, upper) + found: list[dict] = [] + for sql, params in _segments(upper, (edge["time"], edge["rid"]) if edge else None): + if len(found) > limit: + break + found += [dict(row) for row in reader.rows( + f"{select}{sql} ORDER BY time DESC, rowid DESC LIMIT ?", + (uid, *params, limit + 1 - len(found)))] rows += [(source, row) for row in found] if edge and len(found) <= limit: - reached = _key(source, dict(edge[0])) + reached = _key(source, edge) frontier = reached if frontier is None else max(frontier, reached) rows.sort(key=lambda item: _key(*item), reverse=True) if frontier is not None: @@ -180,13 +229,14 @@ def fetch(reader: Reader, maps: Maps, ids: list[int], kinds: tuple[str, ...], cu _SIGN_VERBS = {0: "broke", 1: "placed", 2: "edited"} -def _rollback(source: Source, state: int | None) -> str | None: +# CoreProtect's rolled_back is two flags: 1 the world side, 2 the player's inventory. +_ROLLBACK = {1: "rolled back", 2: "rolled back (player inventory)", 3: "rolled back (world and inventory)"} + + +def _rollback(state: int | None) -> str | None: if not state: return None - if source.name in {"container", "entity_container", "item"}: - return {1: "rolled back", 2: "rolled back (player inventory)", 3: "rolled back (both)"}.get( - state, f"rollback state {state}") - return "rolled back" if state == 1 else f"rollback state {state}" + return _ROLLBACK.get(state, f"rollback state {state}") def _entry(source: Source, row: dict, maps: Maps, victims: dict) -> dict: @@ -198,7 +248,7 @@ def _entry(source: Source, row: dict, maps: Maps, victims: dict) -> dict: "x": row["x"], "y": row["y"], "z": row["z"], "amount": None, "victim": None, - "rolled_back": _rollback(source, row.get("rolled_back")), + "rolled_back": _rollback(row.get("rolled_back")), } if source.name == "block": if action == 3: diff --git a/backend/src/coreprotect/reader.py b/backend/src/coreprotect/reader.py index ab2070ff..c2a19244 100644 --- a/backend/src/coreprotect/reader.py +++ b/backend/src/coreprotect/reader.py @@ -118,6 +118,12 @@ def __enter__(self) -> Reader: deadline = self.budget.deadline self._conn.set_progress_handler(lambda: int(time.monotonic() > deadline), PROGRESS_STEPS) self._statement("PRAGMA query_only = 1", ()) + except sqlite3.OperationalError as exc: + self.close() + raise Unavailable(_classify(exc)) from None + except sqlite3.Error: + self.close() + raise Unavailable("error") from None except BaseException: self.close() raise diff --git a/backend/src/coreprotect/test_activity.py b/backend/src/coreprotect/test_activity.py index c3f1604e..c385f96a 100644 --- a/backend/src/coreprotect/test_activity.py +++ b/backend/src/coreprotect/test_activity.py @@ -155,3 +155,72 @@ def test_sparse_filters_scan_a_bounded_window(coreprotect, monkeypatch): assert sorted(got) == sorted(kills) assert len(got) == len(set(got)) assert pages > 3 + + +def _plans(db, monkeypatch, run): + """Run `run` and return the query plan of every statement the reader issued.""" + from src.coreprotect.reader import Reader as R + + issued = [] + original = R.rows + + def rows(self, sql, params=()): + issued.append((sql, params)) + return original(self, sql, params) + + monkeypatch.setattr(R, "rows", rows) + run() + conn = sqlite3.connect(db.path) + try: + return [(sql, " | ".join(r[3] for r in conn.execute(f"EXPLAIN QUERY PLAN {sql}", params))) + for sql, params in issued if sql.lstrip().upper().startswith("SELECT")] + finally: + conn.close() + + +def test_same_second_bursts_are_seeked_not_scanned(coreprotect, monkeypatch): + monkeypatch.setattr(activity, "SCAN_LIMIT", 50) + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + conn = sqlite3.connect(coreprotect.path) + conn.executemany( + "INSERT INTO co_block (time, user, wid, x, y, z, type, data, action, rolled_back) VALUES (500, ?, 1, ?, 64, 0, 1, 0, 1, 0)", + [(me, i) for i in range(3000)]) + conn.commit() + conn.close() + kill = coreprotect.block(me, 400, 3, type_id=1) + + # Page through the burst to a cursor deep inside that one second. + page = read(coreprotect, [me], kinds=("block",), limit=100) + for _ in range(10): + page = read(coreprotect, [me], kinds=("block",), before=page["next"], limit=100) + deep = page["next"] + assert activity.decode_cursor(SCOPE, deep).time == 500 + + plans = _plans(coreprotect, monkeypatch, lambda: read(coreprotect, [me], kinds=("kill",), before=deep, limit=20)) + for sql, plan in plans: + if "co_block" in sql: + assert "SCAN" not in plan and "TEMP B-TREE" not in plan, (sql, plan) + if "rowid <" in sql or "rowid >=" in sql: + assert "rowid" in plan, (sql, plan) + + # Every row in the burst and the kill below it come back exactly once. + got, before = [], None + while True: + page = read(coreprotect, [me], kinds=("block", "kill"), before=before, limit=100) + got += [e["id"] for e in page["entries"]] + before = page["next"] + if not before: + break + assert len(got) == len(set(got)) == 3001 + assert got[-1] == f"block:{kill}" + + +@pytest.mark.parametrize("state, label", [ + (0, None), (1, "rolled back"), (2, "rolled back (player inventory)"), + (3, "rolled back (world and inventory)"), (7, "rollback state 7"), +]) +def test_rollback_states(coreprotect, state, label): + me = coreprotect.user("Hazel", "0615a817-8cb4-4aef-95f7-f6c9bf7611b8") + coreprotect.block(me, 1, 0, rolled_back=state) + coreprotect.container(me, 2, 1, rolled_back=state) + assert [e["rolled_back"] for e in read(coreprotect, [me])["entries"]] == [label, label] diff --git a/backend/src/coreprotect/test_reader.py b/backend/src/coreprotect/test_reader.py index 48c16212..936d56b9 100644 --- a/backend/src/coreprotect/test_reader.py +++ b/backend/src/coreprotect/test_reader.py @@ -113,3 +113,23 @@ def test_spent_budget_stops_before_the_next_statement(coreprotect): assert exc.value.code == "timeout" assert reader_mod._GATE.acquire(timeout=0.1) reader_mod._GATE.release() + + +def test_unreadable_or_corrupt_files_are_unavailable(tmp_path): + locked = tmp_path / "locked.db" + sqlite3.connect(locked).close() + locked.chmod(0) + try: + with pytest.raises(Unavailable) as exc: + Reader(config(locked)).__enter__() + assert exc.value.code == "missing" + finally: + locked.chmod(0o644) + junk = tmp_path / "junk.db" + junk.write_bytes(b"not a database" * 100) + with pytest.raises(Unavailable) as exc: + with Reader(config(junk)) as r: + r.rows("SELECT * FROM co_world") + assert exc.value.code == "error" + assert reader_mod._GATE.acquire(timeout=0.1) + reader_mod._GATE.release() From 1a155baa015d8053aac2a47b35dc9e07f62ddbb5 Mon Sep 17 00:00:00 2001 From: Ryan Barlow <7389646+ryanbarlow97@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:18:33 +0000 Subject: [PATCH 3/6] Let CoreProtect lock waits use the request budget; record the Main-size gate A reader waiting for CoreProtect's lock holds nothing, so capping the wait at 200 ms only turned long writer batches into busy errors (one request in ten under load). Waits now share the 1.5 s request budget. The gate script retries setup and never waits forever on a crashed worker. Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/benchmarks/coreprotect_reader.py | 15 ++++++++++----- backend/src/coreprotect/README.md | 23 ++++++++++++++++------- backend/src/coreprotect/reader.py | 8 ++++---- 3 files changed, 30 insertions(+), 16 deletions(-) diff --git a/backend/benchmarks/coreprotect_reader.py b/backend/benchmarks/coreprotect_reader.py index d00c92e0..008fd0f8 100644 --- a/backend/benchmarks/coreprotect_reader.py +++ b/backend/benchmarks/coreprotect_reader.py @@ -59,9 +59,14 @@ def connect(target, *args, **kwargs): def _ids(config): - with Reader(config, Budget(30)) as r: - rows = r.rows("SELECT id FROM co_user WHERE uuid IS NOT NULL") - return [row["id"] for row in rows] + # Setup only: retry while a writer holds the lock, as a page reload would. + for _ in range(100): + try: + with Reader(config, Budget(30)) as r: + return [row["id"] for row in r.rows("SELECT id FROM co_user WHERE uuid IS NOT NULL")] + except Unavailable: + time.sleep(0.1) + raise SystemExit("CoreProtect stayed locked") SYNTH_ROWS = {"co_block": 23_600_000, "co_container": 2_960_000, "co_item": 3_300_000, "co_command": 221_000, @@ -269,9 +274,9 @@ def contend(args) -> None: writer.start() for h in hammers: h.start() - latencies, failures, rows = out.get() + latencies, failures, rows = out.get(timeout=args.seconds * 10) writer.join() - reader_results = [reads.get() for _ in hammers] + reader_results = [reads.get(timeout=args.seconds * 10) for _ in hammers] for h in hammers: h.join() print(_summary(f"{phase}: write txn", latencies)) diff --git a/backend/src/coreprotect/README.md b/backend/src/coreprotect/README.md index 28ec87d7..7f3aa88d 100644 --- a/backend/src/coreprotect/README.md +++ b/backend/src/coreprotect/README.md @@ -39,8 +39,8 @@ connection settings for database engines the server does not use. The database uses a rollback journal, so a reader's lock holds off CoreProtect's commits. `reader.py` explains the guards: autocommit, one -reader per process, a 1.5 s budget per request, and every statement fully -fetched before the next. The backend runs as one uvicorn process; running +reader per process, a 1.5 s budget per request (waiting for the lock +included), and every statement fully fetched before the next. The backend runs as one uvicorn process; running more would need the reader limit shared between them. Never open the file with `immutable=1`: on Main that read torn pages while CoreProtect wrote. @@ -63,12 +63,21 @@ login, logout and ping rows; see `sessions.py` for how crashed sessions end. ```sh # Lock hold time per request against a live database (opens it immutable, benchmark only) python3 benchmarks/coreprotect_reader.py hold --no-lock /path/to/database.db -# Writer impact, on a copy (it writes) -python3 benchmarks/coreprotect_reader.py contend /tmp/copy.db --readers 1 +# A disposable database at Main's size, then writer impact on it (contend writes) +python3 benchmarks/coreprotect_reader.py synth /tmp/main-synth.db +python3 benchmarks/coreprotect_reader.py contend /tmp/main-synth.db --readers 1 ``` Acceptance: `hold` p99 ≤ 100 ms and max ≤ 500 ms; `contend` with no writer failures and writer transaction p99/max up by no more than 100/500 ms. -On 2026-10-06 Main (7.4 GB, 242 players) gave p99 8.6 ms and max 18.5 ms; a -copy of Dev's database gave writer p99 +15 ms with one reader and +59 ms with -three readers making back-to-back requests, with no failures. + +Results on 2026-10-06: + +- `hold` on live Main (7.4 GB, 242 players, 1,455 requests): p99 8.6 ms, + max 18.5 ms. +- `contend` on a 6.5 GB synthetic Main, one reader (as in production): + writer p99 +10 ms, max +18 ms, no failures, no reader errors. +- Three readers making back-to-back requests: writer p99 +18 ms, max + +14 ms, throughput −5%, no writer failures; 2% of reader requests gave + up as busy or out of budget. The synthetic writer spills its cache on + every batch, so this is harsher than CoreProtect. diff --git a/backend/src/coreprotect/reader.py b/backend/src/coreprotect/reader.py index c2a19244..a806e300 100644 --- a/backend/src/coreprotect/reader.py +++ b/backend/src/coreprotect/reader.py @@ -12,8 +12,9 @@ - one reader at a time per process. The backend runs as a single uvicorn process; a second process or replica would need its own coordination; - one budget per request covering the wait for the reader slot, cache locks - and every statement. Each statement's busy timeout is capped by what is - left, and a progress handler interrupts a statement that overruns. + and every statement. A statement waiting for the lock holds nothing, so + it may wait for whatever budget is left; a progress handler interrupts a + statement that overruns once it has the lock. The progress handler only runs while SQLite executes VM steps, so the budget is cooperative: it cannot cut short a stalled disk read; - the connection is closed before any decoding, province.db access or @@ -34,7 +35,6 @@ from urllib.parse import quote REQUEST_BUDGET_SECONDS = 1.5 -BUSY_TIMEOUT_MS = 200 PROGRESS_STEPS = 1000 _GATE = threading.BoundedSemaphore(1) @@ -150,7 +150,7 @@ def _statement(self, sql: str, params: tuple) -> list[sqlite3.Row]: raise Unavailable("error") left = self.budget.check() try: - self._conn.execute(f"PRAGMA busy_timeout = {max(1, min(BUSY_TIMEOUT_MS, int(left * 1000)))}") + self._conn.execute(f"PRAGMA busy_timeout = {max(1, int(left * 1000))}") cursor = self._conn.execute(sql, params) try: return cursor.fetchall() From c731357801a07bec1a8c7f3f7e9e694da4d31211 Mon Sep 17 00:00:00 2001 From: Ryan Barlow <7389646+ryanbarlow97@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:18:40 +0000 Subject: [PATCH 4/6] Reflow the CoreProtect reader README Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/src/coreprotect/README.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/backend/src/coreprotect/README.md b/backend/src/coreprotect/README.md index 7f3aa88d..9875061f 100644 --- a/backend/src/coreprotect/README.md +++ b/backend/src/coreprotect/README.md @@ -40,9 +40,10 @@ connection settings for database engines the server does not use. The database uses a rollback journal, so a reader's lock holds off CoreProtect's commits. `reader.py` explains the guards: autocommit, one reader per process, a 1.5 s budget per request (waiting for the lock -included), and every statement fully fetched before the next. The backend runs as one uvicorn process; running -more would need the reader limit shared between them. Never open the file -with `immutable=1`: on Main that read torn pages while CoreProtect wrote. +included), and every statement fully fetched before the next. The backend +runs as one uvicorn process; running more would need the reader limit +shared between them. Never open the file with `immutable=1`: on Main that +read torn pages while CoreProtect wrote. Every query is an indexed seek by player, except small whole-table reads of CoreProtect's name tables, `co_user` and `co_username_log` (a few hundred From 756145374a1c3cec07d1e7d130e0de84a2c5a637 Mon Sep 17 00:00:00 2001 From: Ryan Barlow <7389646+ryanbarlow97@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:49:19 +0000 Subject: [PATCH 5/6] Show chat and whole commands to admins, and audit each view view_player_messages (admin and owner) adds a chat kind and full command text, each cut to 512 characters. Every page that shows any is recorded as player.messages.view in admin_audit (player, server, filters and the rows returned, never the text) before the response leaves, and the page is refused if that record cannot be written. Moderators see what they did. Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/src/api/admin_routes.py | 13 +++--- backend/src/auth/players.py | 49 +++++++++++++++++---- backend/src/auth/roles.py | 2 + backend/src/auth/test_players.py | 66 +++++++++++++++++++++++++++++ backend/src/coreprotect/README.md | 8 +++- backend/src/coreprotect/activity.py | 53 ++++++++++++++++------- 6 files changed, 159 insertions(+), 32 deletions(-) diff --git a/backend/src/api/admin_routes.py b/backend/src/api/admin_routes.py index 6e0223e0..faec68c7 100644 --- a/backend/src/api/admin_routes.py +++ b/backend/src/api/admin_routes.py @@ -45,10 +45,10 @@ def _capable(request: Request, capability: str) -> dict: def _players(request: Request, response: Response, run): - _capable(request, "view_players") + viewer = _capable(request, "view_players") _no_store(response) try: - return run(CoreProtectConfig.from_env()) + return run(CoreProtectConfig.from_env(), viewer) except players.PlayerError as exc: raise HTTPException(exc.status, detail=exc.code) from None @@ -114,26 +114,27 @@ def lookup_accounts(request: Request, response: Response, q: str = ""): @admin_router.get("/players") def get_players(request: Request, response: Response, q: str = "", sort: str = "last_seen", page: int = Query(1, ge=1, le=10_000)): - return _players(request, response, lambda config: players.directory(config, q, sort, page)) + return _players(request, response, lambda config, _viewer: players.directory(config, q, sort, page)) @admin_router.get("/players/{player_uuid}") def get_player(player_uuid: str, request: Request, response: Response): - return _players(request, response, lambda config: players.profile(config, player_uuid)) + return _players(request, response, lambda config, _viewer: players.profile(config, player_uuid)) @admin_router.get("/players/{player_uuid}/sessions") def get_player_sessions(player_uuid: str, request: Request, response: Response, before: str | None = None, limit: int = Query(20, ge=1, le=100)): return _players(request, response, - lambda config: players.player_sessions(config, player_uuid, before, limit)) + lambda config, _viewer: players.player_sessions(config, player_uuid, before, limit)) @admin_router.get("/players/{player_uuid}/activity") def get_player_activity(player_uuid: str, request: Request, response: Response, before: str | None = None, limit: int = Query(20, ge=1, le=100), kinds: str | None = None): return _players(request, response, - lambda config: players.player_activity(config, player_uuid, before, limit, kinds)) + lambda config, viewer: players.player_activity(config, player_uuid, before, limit, kinds, + viewer)) @admin_router.post("/accounts/{user_id}/role") diff --git a/backend/src/auth/players.py b/backend/src/auth/players.py index 62d76229..6264fe9d 100644 --- a/backend/src/auth/players.py +++ b/backend/src/auth/players.py @@ -20,7 +20,7 @@ from src.coreprotect.reader import REQUEST_BUDGET_SECONDS, Budget, CoreProtectConfig, Reader, Unavailable from src.skins.db import connect -from . import users +from . import audit, roles, users DIRECTORY_TTL_SECONDS = 60 PAGE_SIZE = 50 @@ -331,11 +331,39 @@ def player_sessions(config: CoreProtectConfig, text: str, before: str | None, li } +def _audit_messages(viewer: dict, config: CoreProtectConfig, key: str, kinds: tuple[str, ...], + before: str | None, entries: list[dict]) -> None: + """Record that a viewer was shown a player's chat or whole commands, before they see it. + + Never stores the text itself: only which rows were returned. + """ + shown = [e for e in entries if e["kind"] in activity.MESSAGE_KINDS] + if not shown: + return + detail = { + "player_uuid": key, + "server": config.server, + "kinds": sorted(kinds), + "paged": before is not None, + "rows": len(shown), + "newest": {"id": shown[0]["id"], "time": shown[0]["time"]}, + "oldest": {"id": shown[-1]["id"], "time": shown[-1]["time"]}, + } + try: + with connect() as conn: + audit.record(conn, actor=viewer, action="player.messages.view", outcome="ok", detail=detail) + conn.commit() + except Exception: + raise PlayerError(503, "audit_unavailable") from None + + def player_activity(config: CoreProtectConfig, text: str, before: str | None, limit: int, - kinds: str | None) -> dict: + kinds: str | None, viewer: dict | None = None) -> dict: key = _require(text) + full = bool(viewer) and roles.can(viewer["role"], "view_player_messages") + allowed = activity.FULL_KINDS if full else activity.KINDS try: - chosen = activity.parse_kinds(kinds) + chosen = activity.parse_kinds(kinds, allowed) except ValueError: raise PlayerError(400, "bad_kinds") from None try: @@ -346,10 +374,15 @@ def player_activity(config: CoreProtectConfig, text: str, before: str | None, li with Reader(config, Budget()) as reader: ids = [a["id"] for a in _ids(reader, key)] maps = co_maps.get(reader) - raw = activity.fetch(reader, maps, ids, chosen, cursor, limit) if ids else None + raw = activity.fetch(reader, maps, ids, chosen, cursor, limit, full_text=full) if ids else None except Unavailable as exc: - return {"entries": [], "next": None, "searched_to": None, "kinds": list(activity.KINDS), "coreprotect": _status(exc)} + return {"entries": [], "next": None, "searched_to": None, "kinds": list(allowed), + "shows_messages": full, "coreprotect": _status(exc)} if raw is None: - return {"entries": [], "next": None, "searched_to": None, "kinds": list(activity.KINDS), "coreprotect": _status(None)} - return {**activity.build(raw, maps, f"{config.server}:{key}"), "kinds": list(activity.KINDS), - "coreprotect": _status(None)} + return {"entries": [], "next": None, "searched_to": None, "kinds": list(allowed), + "shows_messages": full, "coreprotect": _status(None)} + built = activity.build(raw, maps, f"{config.server}:{key}") + # CoreProtect is closed by now; the view is on record before the response leaves. + if full: + _audit_messages(viewer, config, key, chosen, before, built["entries"]) + return {**built, "kinds": list(allowed), "shows_messages": full, "coreprotect": _status(None)} diff --git a/backend/src/auth/roles.py b/backend/src/auth/roles.py index c43d8473..3ca874e8 100644 --- a/backend/src/auth/roles.py +++ b/backend/src/auth/roles.py @@ -8,6 +8,8 @@ CAPABILITIES = { "view_admin": "mod", "view_players": "mod", + # Chat and whole commands in a player's activity; every view is audited. + "view_player_messages": "admin", "revoke_sessions": "mod", "change_role": "admin", } diff --git a/backend/src/auth/test_players.py b/backend/src/auth/test_players.py index b89dbff8..8cd7d2cc 100644 --- a/backend/src/auth/test_players.py +++ b/backend/src/auth/test_players.py @@ -1,3 +1,5 @@ +import json + import pytest from fastapi import FastAPI @@ -189,3 +191,67 @@ def test_unknown_player_sections_are_empty(app, env, world, monkeypatch): monkeypatch.setenv("COREPROTECT_DB", str(env.DATA_DIR / "nowhere.db")) body = c.get(f"/admin/players/{HAZEL}/activity").json() assert body["coreprotect"] == {"status": "unavailable", "reason": "missing"} + + +def message_audits(db): + with db.connect() as conn: + return [dict(r) for r in conn.execute( + "SELECT * FROM admin_audit WHERE action = 'player.messages.view' ORDER BY id")] + + +def test_mods_never_see_messages(app, env, world, coreprotect): + coreprotect.chat(world["hazel"], 170, "a private chat line") + c = client(app, staff(env, "mod")) + feed = c.get(f"/admin/players/{HAZEL}/activity").json() + assert feed["shows_messages"] is False + assert "chat" not in feed["kinds"] + assert "private" not in str(feed) + assert [e["target"] for e in feed["entries"] if e["kind"] == "command"] == ["/msg"] + assert c.get(f"/admin/players/{HAZEL}/activity", params={"kinds": "chat"}).status_code == 400 + assert message_audits(env) == [] + + +@pytest.mark.parametrize("role", ["admin", "root"]) +def test_admins_see_chat_and_whole_commands_and_it_is_audited(app, env, world, coreprotect, role): + coreprotect.chat(world["hazel"], 170, "meet at the docks") + coreprotect.chat(world["hazel"], 180, "x" * 600) + _, token = account(env, f"boss-{role}", role) + feed = client(app, token).get(f"/admin/players/{HAZEL}/activity").json() + assert feed["shows_messages"] is True + assert "chat" in feed["kinds"] + chats = [e for e in feed["entries"] if e["kind"] == "chat"] + assert [(c["verb"], c["message"][:17], c["truncated"]) for c in chats] == [ + ("said", "x" * 17, True), ("said", "meet at the docks", False)] + assert len(chats[0]["message"]) == 512 + assert [e["target"] for e in feed["entries"] if e["kind"] == "command"] == ["/msg Bob a private thing"] + + rows = message_audits(env) + assert len(rows) == 1 + assert rows[0]["actor_role"] == role + assert rows[0]["outcome"] == "ok" + assert rows[0]["reason"] is None + detail = json.loads(rows[0]["detail_json"]) + assert detail["player_uuid"] == HAZEL and detail["rows"] == 3 and detail["paged"] is False + assert detail["newest"]["time"] == 180 and detail["oldest"]["time"] == 160 + # The audit records which rows were shown, never what they said. + assert "docks" not in rows[0]["detail_json"] and "private" not in rows[0]["detail_json"] + + +def test_pages_without_messages_are_not_audited(app, env, world): + _, token = account(env, "boss", "admin") + feed = client(app, token).get(f"/admin/players/{HAZEL}/activity", params={"kinds": "block,session"}).json() + assert feed["entries"] + assert message_audits(env) == [] + + +def test_no_audit_no_messages(app, env, world, monkeypatch): + _, token = account(env, "boss", "admin") + + def broken(*args, **kwargs): + raise RuntimeError("disk full") + + monkeypatch.setattr(players.audit, "record", broken) + response = client(app, token).get(f"/admin/players/{HAZEL}/activity") + assert response.status_code == 503 + assert response.json()["detail"] == "audit_unavailable" + assert "private" not in response.text diff --git a/backend/src/coreprotect/README.md b/backend/src/coreprotect/README.md index 9875061f..91709e6e 100644 --- a/backend/src/coreprotect/README.md +++ b/backend/src/coreprotect/README.md @@ -53,8 +53,12 @@ table, so a sparse filter stops early and returns a cursor to continue ## What is shown -Chat is never read. Commands are cut to their first word inside SQL, and sign -text, item metadata and NBT are not selected. Sessions are rebuilt from +For moderators, chat is never read and commands are cut to their first word +inside SQL. Admins and the owner (`view_player_messages`) also get chat and +whole commands, each cut to 512 characters; every page that shows any is +recorded in `admin_audit` as `player.messages.view` (which rows, never the +text) before it is returned, and the page is refused if that record cannot +be written. Sign text, item metadata and NBT are never selected. Sessions are rebuilt from login, logout and ping rows; see `sessions.py` for how crashed sessions end. ## Deployment gate diff --git a/backend/src/coreprotect/activity.py b/backend/src/coreprotect/activity.py index 4f3c94e3..fd75e0e1 100644 --- a/backend/src/coreprotect/activity.py +++ b/backend/src/coreprotect/activity.py @@ -7,10 +7,11 @@ order is (time, source rank, rowid), all descending; the cursor is the last row's key. -Only what staff need to see what happened is selected. Chat is never read. -Commands are cut to their first word inside SQL, so arguments (messages, -passwords, link codes) never leave the database. Sign text, item metadata -and NBT are not selected. +Only what staff need to see what happened is selected. Unless the caller +may read messages (full_text), chat is never read and commands are cut to +their first word inside SQL, so arguments (messages, passwords, link codes) +never leave the database. Sign text, item metadata and NBT are never +selected. """ from __future__ import annotations @@ -30,13 +31,15 @@ _SKILL = "substr(message, 1, 7) = '[skill]'" _TELEPORT = "instr(message, ' teleported from ')" +MESSAGE_MAX = 512 +_CAST = f"CASE WHEN {_TELEPORT} > 0 THEN substr(message, 9, {_TELEPORT} - 9) ELSE substr(message, 9) END" # The command's first word, or a skill cast's name without teleport coordinates. _COMMAND_TEXT = ( - f"substr(CASE WHEN {_SKILL} THEN " - f" CASE WHEN {_TELEPORT} > 0 THEN substr(message, 9, {_TELEPORT} - 9) ELSE substr(message, 9) END " + f"substr(CASE WHEN {_SKILL} THEN {_CAST} " "ELSE CASE WHEN instr(message, ' ') > 0 THEN substr(message, 1, instr(message, ' ') - 1) ELSE message END " "END, 1, 80)" ) +_FULL_COMMAND_TEXT = f"CASE WHEN {_SKILL} THEN substr({_CAST}, 1, 80) ELSE substr(message, 1, {MESSAGE_MAX}) END" @dataclass(frozen=True) @@ -46,6 +49,10 @@ class Source: table: str columns: str kinds: dict[str, str] + # Columns used instead when the caller may read messages. + full_columns: str | None = None + # Read only when the caller may read messages. + sensitive: bool = False SOURCES = ( @@ -61,11 +68,19 @@ class Source: {"entity": "action BETWEEN 0 AND 3"}), Source(6, "sign", "co_sign", "action", {"sign": "action IN (0, 1, 2)"}), Source(7, "command", "co_command", - f"{_SKILL} AS is_skill, {_TELEPORT} > 0 AS teleport, {_COMMAND_TEXT} AS text", - {"skill": _SKILL, "command": f"NOT {_SKILL}"}), + f"{_SKILL} AS is_skill, {_TELEPORT} > 0 AS teleport, {_COMMAND_TEXT} AS text, 0 AS truncated", + {"skill": _SKILL, "command": f"NOT {_SKILL}"}, + full_columns=(f"{_SKILL} AS is_skill, {_TELEPORT} > 0 AS teleport, {_FULL_COMMAND_TEXT} AS text, " + f"NOT {_SKILL} AND length(message) > {MESSAGE_MAX} AS truncated")), Source(8, "session", "co_session", "action", {"session": "action IN (0, 1)"}), + Source(9, "chat", "co_chat", f"substr(message, 1, {MESSAGE_MAX}) AS text, length(message) > {MESSAGE_MAX} AS truncated", + {"chat": "1"}, sensitive=True), ) -KINDS = tuple(dict.fromkeys(kind for source in SOURCES for kind in source.kinds)) +# Kinds anyone with view_players may ask for, and the extra ones for view_player_messages. +KINDS = tuple(dict.fromkeys(kind for source in SOURCES if not source.sensitive for kind in source.kinds)) +FULL_KINDS = tuple(dict.fromkeys(kind for source in SOURCES for kind in source.kinds)) +# Kinds whose rows carry message text when read with full_text. +MESSAGE_KINDS = frozenset({"chat", "command"}) _BY_RANK = {source.rank: source for source in SOURCES} @@ -89,11 +104,11 @@ def decode_cursor(server: str, token: str | None) -> Cursor | None: return Cursor(*values) -def parse_kinds(text: str | None) -> tuple[str, ...]: +def parse_kinds(text: str | None, allowed: tuple[str, ...] = KINDS) -> tuple[str, ...]: if not text: - return KINDS + return allowed chosen = tuple(dict.fromkeys(part.strip() for part in text.split(",") if part.strip())) - if not chosen or any(kind not in KINDS for kind in chosen): + if not chosen or any(kind not in allowed for kind in chosen): raise ValueError("bad_kinds") return chosen @@ -166,21 +181,22 @@ def _key(source: Source, row: dict) -> Cursor: def fetch(reader: Reader, maps: Maps, ids: list[int], kinds: tuple[str, ...], cursor: Cursor | None, - limit: int) -> dict: + limit: int, full_text: bool = False) -> dict: limit = max(1, min(limit, MAX_LIMIT)) rows = [] # The newest point a capped scan reached. Rows older than it may sit # beside unexamined rows of that source, so this page stops there. frontier: Cursor | None = None for source in SOURCES: - if source.table not in maps.tables: + if source.table not in maps.tables or (source.sensitive and not full_text): continue conditions = [cond for kind, cond in source.kinds.items() if kind in kinds] if not conditions: continue upper = _upper(source, cursor) select = ( - f"SELECT rowid AS rid, time, wid, x, y, z, {source.columns} FROM {source.table} " + f"SELECT rowid AS rid, time, wid, x, y, z, " + f"{(source.full_columns if full_text else None) or source.columns} FROM {source.table} " f"WHERE user = ? AND ({' OR '.join(conditions)})" ) for uid in ids: @@ -248,6 +264,8 @@ def _entry(source: Source, row: dict, maps: Maps, victims: dict) -> dict: "x": row["x"], "y": row["y"], "z": row["z"], "amount": None, "victim": None, + "message": None, + "truncated": False, "rolled_back": _rollback(row.get("rolled_back")), } if source.name == "block": @@ -279,7 +297,10 @@ def _entry(source: Source, row: dict, maps: Maps, victims: dict) -> dict: entry.update(kind="skill", verb="teleported with" if row["teleport"] else "cast", target=(row["text"] or "").strip() or "a skill") else: - entry.update(kind="command", verb="ran", target=row["text"] or "/") + entry.update(kind="command", verb="ran", target=row["text"] or "/", truncated=bool(row["truncated"])) + elif source.name == "chat": + entry.update(kind="chat", verb="said", target=None, message=row["text"] or "", + truncated=bool(row["truncated"])) else: entry.update(kind="session", verb="logged in" if action == 1 else "logged out", target=None) return entry From 2bbf7c017d4ed0b06bafc93f6a489dfcbd767591 Mon Sep 17 00:00:00 2001 From: Ryan Barlow <7389646+ryanbarlow97@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:50:49 +0000 Subject: [PATCH 6/6] Refuse cursor values outside SQLite's integer range Review fixes: an out-of-range cursor value is a bad cursor, not a 500; the unreadable-file test is skipped under root and always releases the reader slot; the synthetic database builder reads its index list once. Co-Authored-By: Claude Opus 5.5 (1M context) --- backend/benchmarks/coreprotect_reader.py | 7 ++++--- backend/src/auth/test_players.py | 8 ++++++++ backend/src/coreprotect/cursors.py | 3 ++- backend/src/coreprotect/test_activity.py | 3 ++- backend/src/coreprotect/test_reader.py | 18 +++++++++++------- 5 files changed, 27 insertions(+), 12 deletions(-) diff --git a/backend/benchmarks/coreprotect_reader.py b/backend/benchmarks/coreprotect_reader.py index 008fd0f8..321b7e31 100644 --- a/backend/benchmarks/coreprotect_reader.py +++ b/backend/benchmarks/coreprotect_reader.py @@ -80,9 +80,10 @@ def synth(args) -> None: conn.execute("PRAGMA journal_mode = OFF") conn.execute("PRAGMA synchronous = OFF") conn.executescript(schema) - indexes = [row[0] for row in conn.execute("SELECT sql FROM sqlite_master WHERE type = 'index' AND sql IS NOT NULL")] - for (name,) in conn.execute("SELECT name FROM sqlite_master WHERE type = 'index' AND sql IS NOT NULL").fetchall(): - conn.execute(f"DROP INDEX {name}") + found = conn.execute("SELECT name, sql FROM sqlite_master WHERE type = 'index' AND sql IS NOT NULL").fetchall() + indexes = [sql for _, sql in found] + for name, _ in found: + conn.execute('DROP INDEX "{}"'.format(name.replace('"', '""'))) conn.executemany("INSERT INTO co_user (id, time, user, uuid) VALUES (?, 0, ?, ?)", [(i, f"Player{i}", f"00000000-0000-4000-8000-{i:012d}") for i in range(1, SYNTH_PLAYERS + 1)]) conn.execute("INSERT INTO co_world (id, world) VALUES (1, 'TFMC_Map')") diff --git a/backend/src/auth/test_players.py b/backend/src/auth/test_players.py index 8cd7d2cc..e56e3a33 100644 --- a/backend/src/auth/test_players.py +++ b/backend/src/auth/test_players.py @@ -255,3 +255,11 @@ def broken(*args, **kwargs): assert response.status_code == 503 assert response.json()["detail"] == "audit_unavailable" assert "private" not in response.text + + +def test_oversized_cursor_values_are_refused(app, env, world): + from src.coreprotect import cursors + + c = client(app, staff(env)) + huge = cursors.encode("main", f"sessions:{HAZEL}", 10 ** 30, 1) + assert c.get(f"/admin/players/{HAZEL}/sessions", params={"before": huge}).status_code == 400 diff --git a/backend/src/coreprotect/cursors.py b/backend/src/coreprotect/cursors.py index 3546ca2e..c5920881 100644 --- a/backend/src/coreprotect/cursors.py +++ b/backend/src/coreprotect/cursors.py @@ -26,6 +26,7 @@ def decode(server: str, listing: str, token: str | None, size: int) -> tuple[int ok = data["s"] == server and data["l"] == listing except (ValueError, TypeError, KeyError, binascii.Error): raise BadCursor("bad_cursor") from None - if not ok or len(values) != size or not all(type(v) is int for v in values): + # SQLite integers are 64-bit; a larger one could not even be bound. + if not ok or len(values) != size or not all(type(v) is int and -(2 ** 63) <= v < 2 ** 63 for v in values): raise BadCursor("bad_cursor") return values diff --git a/backend/src/coreprotect/test_activity.py b/backend/src/coreprotect/test_activity.py index c385f96a..e66b4f43 100644 --- a/backend/src/coreprotect/test_activity.py +++ b/backend/src/coreprotect/test_activity.py @@ -119,7 +119,8 @@ def test_cursors_are_scoped_and_validated(): token = activity.encode_cursor(SCOPE, activity.Cursor(10, 1, 5)) assert activity.decode_cursor(SCOPE, token) == activity.Cursor(10, 1, 5) for bad in ("nonsense", activity.encode_cursor("main:someone-else", activity.Cursor(10, 1, 5)), - cursors.encode(SCOPE, "activity", 10, 99, 5), cursors.encode(SCOPE, "sessions", 10, 1, 5), "x" * 300): + cursors.encode(SCOPE, "activity", 10, 99, 5), cursors.encode(SCOPE, "sessions", 10, 1, 5), "x" * 300, + cursors.encode(SCOPE, "activity", 10 ** 30, 1, 5)): with pytest.raises(cursors.BadCursor): activity.decode_cursor(SCOPE, bad) diff --git a/backend/src/coreprotect/test_reader.py b/backend/src/coreprotect/test_reader.py index 936d56b9..9d9eb0d4 100644 --- a/backend/src/coreprotect/test_reader.py +++ b/backend/src/coreprotect/test_reader.py @@ -1,3 +1,4 @@ +import os import sqlite3 import threading import time @@ -118,13 +119,16 @@ def test_spent_budget_stops_before_the_next_statement(coreprotect): def test_unreadable_or_corrupt_files_are_unavailable(tmp_path): locked = tmp_path / "locked.db" sqlite3.connect(locked).close() - locked.chmod(0) - try: - with pytest.raises(Unavailable) as exc: - Reader(config(locked)).__enter__() - assert exc.value.code == "missing" - finally: - locked.chmod(0o644) + # Root reads the file regardless of its mode. + if os.geteuid() != 0: + locked.chmod(0) + try: + with pytest.raises(Unavailable) as exc: + with Reader(config(locked)): + pass + assert exc.value.code == "missing" + finally: + locked.chmod(0o644) junk = tmp_path / "junk.db" junk.write_bytes(b"not a database" * 100) with pytest.raises(Unavailable) as exc: