feat(admin): add a root-only database studio tab

backend/requirements.txt

@@ -10,4 +10,6 @@ pyyaml
 httpx
 anthropic
 psycopg2-binary
+# The database studio's Postgres reads and edits: one statement per request, enforced by Postgres.
+psycopg[binary]
 pgvector

backend/src/api/admin_routes.py

@@ -398,9 +398,11 @@ async def _validation_handler(request: Request, exc: RequestValidationError):
 
 def install(app: FastAPI) -> None:
     """Mount the staff routes and audit their malformed writes."""
+    from src.api.database_routes import database_router
     from src.api.luckperms_routes import plugin_router, staff_router
 
     app.include_router(admin_router)
+    app.include_router(database_router)
     app.include_router(staff_router)
     app.include_router(plugin_router)
     app.add_exception_handler(RequestValidationError, _validation_handler)

backend/src/api/database_routes.py

@@ -0,0 +1,94 @@
+"""Root's database studio. Every browse, query and edit is audited with its SQL."""
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, Request, Response
+from pydantic import BaseModel, Field, StrictBool
+
+from src.api.auth_routes import _no_store
+from src.api.staff_access import require_staff_account
+from src.auth import audit
+from src.dbstudio import studio
+from src.skins.db import connect
+
+database_router = APIRouter(prefix="/admin/database", tags=["admin"])
+
+# Enough to read back what ran; very long statements are cut.
+AUDIT_SQL = 2000
+
+
+class QueryBody(BaseModel):
+    sql: str = Field(..., max_length=studio.MAX_SQL)
+    write: StrictBool = False
+
+
+class BrowseBody(BaseModel):
+    table: str = Field(..., max_length=200)
+    offset: int = Field(0, ge=0, le=10**12)
+    limit: int = Field(100, ge=1, le=studio.MAX_PAGE)
+    order: str | None = Field(None, max_length=200)
+    descending: StrictBool = False
+    where: str = Field("", max_length=4000)
+
+
+def _refuse(exc: studio.StudioError):
+    raise HTTPException(exc.status, detail=exc.message or exc.code)
+
+
+def _audited(actor: dict, source_id: str, sql: str, write: bool, run, action: str | None = None):
+    detail = {"database": source_id, "sql": sql[:AUDIT_SQL]}
+    action = action or ("database.write" if write else "database.read")
+    if write:
+        # Logged before the edit runs: if the audit log can't be written, nothing changes.
+        audit.record_refusal(actor=actor, action=action, outcome="requested", detail=detail)
+    try:
+        result = run()
+    except studio.StudioError as exc:
+        audit.record_refusal(actor=actor, action=action, outcome="failed", detail={**detail, "error": exc.code})
+        _refuse(exc)
+    rows = result.get("rows")
+    audit.record_refusal(actor=actor, action=action, outcome="ok",
+                         detail={**detail, **({"rows": len(rows), "changes": result.get("changes")} if rows is not None else {})})
+    return result
+
+
+@database_router.get("")
+def database_overview(request: Request, response: Response):
+    actor = require_staff_account(request, "use_database")
+    _no_store(response)
+    with connect() as conn:
+        counts = dict(conn.execute(
+            "SELECT json_extract(detail_json, '$.database'), count(*) FROM admin_audit "
+            "WHERE action IN ('database.read', 'database.write') AND outcome IN ('ok', 'failed') "
+            "GROUP BY 1").fetchall())
+    return _audited(actor, "*", "list databases", False, lambda: studio.overview(counts), "database.list")
+
+
+@database_router.get("/{source_id}/schema")
+def database_schema(source_id: str, request: Request, response: Response):
+    actor = require_staff_account(request, "use_database")
+    _no_store(response)
+    return _audited(actor, source_id, "schema", False,
+                    lambda: {**studio.schema(source_id), "database": _describe(source_id)}, "database.schema")
+
+
+@database_router.post("/{source_id}/browse")
+def database_browse(source_id: str, body: BrowseBody, request: Request, response: Response):
+    actor = require_staff_account(request, "use_database", write=True)
+    _no_store(response)
+    where = f" WHERE {body.where.strip()}" if body.where.strip() else ""
+    return _audited(actor, source_id, f"browse {body.table}{where}", False, lambda: studio.browse(
+        source_id, body.table, offset=body.offset, limit=body.limit, order=body.order,
+        descending=body.descending, where=body.where))
+
+
+@database_router.post("/{source_id}/query")
+def database_query(source_id: str, body: QueryBody, request: Request, response: Response):
+    actor = require_staff_account(request, "use_database", write=True)
+    _no_store(response)
+    return _audited(actor, source_id, body.sql.strip(), body.write,
+                    lambda: studio.run(source_id, body.sql, write=body.write).json())
+
+
+def _describe(source_id: str) -> dict:
+    item = studio.source(source_id)
+    return {"id": item.id, "name": item.name, "engine": item.engine, "writable": item.writable, "detail": item.detail}

backend/src/auth/roles.py

@@ -31,6 +31,8 @@
     "inspect_codes": "mod",
     # Reports players send from the /diagnostic page.
     "view_diagnostics": "mod",
+    # Browse, query and edit the site's databases; every statement is audited.
+    "use_database": "root",
 }
 
 

backend/src/coreprotect/reader.py

@@ -70,9 +70,11 @@ def from_env(cls) -> CoreProtectConfig:
 class Unavailable(Exception):
     """CoreProtect could not be read: not_configured, missing, cannot_open, busy, timeout or error."""
 
-    def __init__(self, code: str):
+    def __init__(self, code: str, message: str = ""):
         super().__init__(code)
         self.code = code
+        # SQLite's own words, for the database studio.
+        self.message = message
 
 
 class Budget:
@@ -148,6 +150,32 @@ def close(self) -> None:
     def rows(self, sql: str, params: tuple = ()) -> list[sqlite3.Row]:
         return self._statement(sql, params)
 
+    def query(self, sql: str, limit: int) -> tuple[list[str], list[tuple], bool]:
+        """One statement from the database studio: at most `limit` rows, then the cursor closes.
+
+        Returns the column names, the rows and whether more rows were left unread.
+        """
+        if self._conn is None:
+            raise Unavailable("error")
+        left = self.budget.check()
+        try:
+            self._conn.execute(f"PRAGMA busy_timeout = {max(1, int(left * 1000))}")
+            self._conn.set_authorizer(_no_attach)
+            try:
+                cursor = self._conn.execute(sql)
+                try:
+                    columns = [d[0] for d in cursor.description or ()]
+                    rows = cursor.fetchmany(limit + 1) if cursor.description else []
+                finally:
+                    cursor.close()
+            finally:
+                self._conn.set_authorizer(None)
+        except sqlite3.OperationalError as exc:
+            raise Unavailable(_classify(exc), str(exc)) from None
+        except sqlite3.Error as exc:
+            raise Unavailable("error", str(exc)) from None
+        return columns, [tuple(row) for row in rows[:limit]], len(rows) > limit
+
     def _statement(self, sql: str, params: tuple) -> list[sqlite3.Row]:
         if self._conn is None:
             raise Unavailable("error")
@@ -165,6 +193,11 @@ def _statement(self, sql: str, params: tuple) -> list[sqlite3.Row]:
             raise Unavailable("error") from None
 
 
+def _no_attach(action: int, *_) -> int:
+    # ATTACH would let a studio query open other database files on the host.
+    return sqlite3.SQLITE_DENY if action in (sqlite3.SQLITE_ATTACH, sqlite3.SQLITE_DETACH) else sqlite3.SQLITE_OK
+
+
 def _classify(exc: sqlite3.OperationalError) -> str:
     text = str(exc).lower()
     if "interrupted" in text:

backend/src/dbstudio/conftest.py

@@ -0,0 +1,18 @@
+import sys
+from pathlib import Path
+import pytest
+
+BACKEND = Path(__file__).resolve().parents[2]
+if str(BACKEND) not in sys.path:
+    sys.path.insert(0, str(BACKEND))
+
+
+@pytest.fixture
+def database(tmp_path, monkeypatch):
+    from src.skins import db
+    monkeypatch.setattr(db, "DB_PATH", tmp_path / "province.db")
+    monkeypatch.setattr(db, "DATA_DIR", tmp_path)
+    for name in ("SKINS_DIR", "WARDROBE_DIR", "DRINKS_DIR"):
+        monkeypatch.setattr(db, name, tmp_path / name.lower())
+    db.migrate()
+    return db