Defensive coding: sanitize free text, constrain inputs

Player-authored free text is scrubbed at the API boundary before storage via a
new validation.py: drop control characters and unpaired surrogates, trim
whitespace, and cap length (names 120, other text 2000). Applied to crew/player
names, the character sheet, like/hate answers, techniques, recruit techniques,
the renamed Name, the Gat description, and challenge stakes. Values matched
against stored text (swap offers, face-card assignments) and identifiers/enums
are left untouched so they still compare equal.

Constrained set-role to the two real roles so it can't stash an arbitrary string
in the column. Audited the rest: queries are parameterized by SQLModel (the lone
f-string SQL in database.py is the hardcoded legacy-migration list, no user
input); objective-type and rollback writes already whitelist their keys; no
endpoint accepts a generic (field, value) write.

Tests cover the sanitizer (control chars, surrogates, emoji, caps) and the
create/join HTTP path end-to-end.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

README.md

@@ -61,6 +61,19 @@ uvicorn pirats.main:app --host 0.0.0.0 --port 8000 --reload
 
 Once running, access the web UI at [http://localhost:8000](http://localhost:8000).
 
+#### Configuration (environment variables)
+
+| Variable | Default | Purpose |
+| --- | --- | --- |
+| `DATABASE_URL` | `sqlite:///./rats_with_gats.db` | SQLAlchemy database URL. |
+| `PIRATS_LOG_FILE` | _(unset)_ | If set, also write logs to this file (rotating, 10 MB × 5 backups). Logs always go to stderr regardless. |
+| `PIRATS_LOG_LEVEL` | `INFO` | Minimum log severity (`DEBUG`/`INFO`/`WARNING`/`ERROR`/`CRITICAL`). |
+| `PIRATS_DEV_MODE` | _(unset = on)_ | Default Dev Mode for new games. Unset is treated as a local checkout (on); set to `0`/`false` to disable. |
+| `PIRATS_PURGE_ENABLED` | `true` | Periodically purge old finished/inactive games. |
+| `PIRATS_PURGE_INTERVAL_HOURS` | `24` | How often the purge task runs. |
+| `PIRATS_PURGE_FINISHED_DAYS` | `14` | Purge games that finished more than this many days ago. |
+| `PIRATS_PURGE_INACTIVE_DAYS` | `30` | Purge games with no activity for this many days. |
+
 ---
 
 ### Option 2: Using Nix Flakes
@@ -99,10 +112,22 @@ services.pirats = {
   host = "127.0.0.1";
   port = 8000;
   databasePath = "/var/lib/pirats/rats_with_gats.db";
+  logFile = "/var/log/pirats/pirats.log"; # rotating; stderr also goes to the journal
+  logLevel = "info";
   openFirewall = false; # Set to true to open ports in the firewall
+  purge = {
+    enable = true;       # periodically remove old finished/inactive games
+    intervalHours = 24;
+    finishedDays = 14;   # purge games finished more than 14 days ago
+    inactiveDays = 30;   # ...or with no activity for 30 days
+  };
 };
 ```
 
+The service writes its log to `logFile` (under the systemd-managed `/var/log/pirats`
+`LogsDirectory`, so the sandboxed `DynamicUser` can write to it) and additionally
+to the systemd journal via stderr (`journalctl -u pirats`).
+
 ---
 
 ## Running Tests

TODO.md

@@ -1,8 +1,8 @@
 ## DevOps/Maintenance
 
-- [ ] **Logging**: Add some logging to the backend. The Nix service should log to /var/log/pirats.log by default (but be configurable). At very least, stderr should go there, plus important server-side events, like:
-- [ ] **Purge old finished/inactive games**: Add a periodic task (daily is good enough, but make the interval configurable through an env variable/the nix flake) to purge games that have finished more than two weeks ago, or have had no moves played within the last month. These time windows should also be configurable. Log all purges, including crew names/uuids and how much disk space they recovered.
-- [ ] **Defensive coding.** While we trust our players, this app is exposed on the open internet. I'd like to take basic precautions like making sure we're safe against SQL injection (SQLModel should handle this?), doing some basic sanitizing of free inputs from players (valid unicode only + generous character limit should be enough), and making sure that every game action API call is specific and constrained enough that it can't just do arbitrary database updates.
+- [x] **Logging**: Add some logging to the backend. The Nix service should log to /var/log/pirats.log by default (but be configurable). At very least, stderr should go there, plus important server-side events, like:
+- [x] **Purge old finished/inactive games**: Add a periodic task (daily is good enough, but make the interval configurable through an env variable/the nix flake) to purge games that have finished more than two weeks ago, or have had no moves played within the last month. These time windows should also be configurable. Log all purges, including crew names/uuids and how much disk space they recovered.
+- [x] **Defensive coding.** While we trust our players, this app is exposed on the open internet. I'd like to take basic precautions like making sure we're safe against SQL injection (SQLModel should handle this?), doing some basic sanitizing of free inputs from players (valid unicode only + generous character limit should be enough), and making sure that every game action API call is specific and constrained enough that it can't just do arbitrary database updates.
 
 ## Words Words Words (Waiting for Fable to come back)
 

flake.nix

@@ -112,6 +112,22 @@
               default = "/var/lib/pirats/rats_with_gats.db";
               description = "Path to the SQLite database file.";
             };
+            logFile = lib.mkOption {
+              type = lib.types.str;
+              default = "/var/log/pirats/pirats.log";
+              description = ''
+                Path to the server log file (rotating, 10 MB x 5 backups). stderr is
+                also captured by the systemd journal. The default lives under the
+                service's LogsDirectory (/var/log/pirats), which the sandboxed
+                DynamicUser can write to; a custom path elsewhere must be made
+                writable by the service separately.
+              '';
+            };
+            logLevel = lib.mkOption {
+              type = lib.types.enum [ "debug" "info" "warning" "error" "critical" ];
+              default = "info";
+              description = "Minimum severity of log messages to record.";
+            };
             openFirewall = lib.mkOption {
               type = lib.types.bool;
               default = false;
@@ -122,6 +138,28 @@
               default = false;
               description = "Whether new games start with Dev Mode (testing shortcuts) enabled.";
             };
+            purge = {
+              enable = lib.mkOption {
+                type = lib.types.bool;
+                default = true;
+                description = "Periodically purge old finished/inactive games.";
+              };
+              intervalHours = lib.mkOption {
+                type = lib.types.numbers.positive;
+                default = 24;
+                description = "How often (in hours) the purge task runs.";
+              };
+              finishedDays = lib.mkOption {
+                type = lib.types.numbers.positive;
+                default = 14;
+                description = "Purge games that finished more than this many days ago.";
+              };
+              inactiveDays = lib.mkOption {
+                type = lib.types.numbers.positive;
+                default = 30;
+                description = "Purge games with no activity for this many days.";
+              };
+            };
           };
 
           config = lib.mkIf cfg.enable {
@@ -137,15 +175,25 @@
                 # Unset means "local checkout" and defaults Dev Mode on, so the
                 # service must always set it explicitly.
                 PIRATS_DEV_MODE = lib.boolToString cfg.devModeDefault;
+                PIRATS_LOG_FILE = cfg.logFile;
+                PIRATS_LOG_LEVEL = cfg.logLevel;
+                PIRATS_PURGE_ENABLED = lib.boolToString cfg.purge.enable;
+                PIRATS_PURGE_INTERVAL_HOURS = toString cfg.purge.intervalHours;
+                PIRATS_PURGE_FINISHED_DAYS = toString cfg.purge.finishedDays;
+                PIRATS_PURGE_INACTIVE_DAYS = toString cfg.purge.inactiveDays;
               };
 
               serviceConfig = {
                 ExecStart = "${cfg.package}/bin/pirats --host ${cfg.host} --port ${toString cfg.port}";
                 Restart = "always";
-                
+
                 # Sandboxing and security
                 DynamicUser = true;
                 StateDirectory = "pirats";
+                # Creates /var/log/pirats owned by the dynamic user and adds it to
+                # ReadWritePaths, so the default logFile under it is writable
+                # despite ProtectSystem=strict.
+                LogsDirectory = "pirats";
                 WorkingDirectory = "/var/lib/pirats";
                 
                 ProtectSystem = "strict";

frontend/src/lib/changelog.js

@@ -6,7 +6,7 @@
 //   - Add a CHANGELOG entry only when a commit changes something players can
 //     see. Skip refactors, tests, and tooling. Keep wording player-facing.
 
-export const VERSION = 8;
+export const VERSION = 11;
 
 // Newest first. Each entry: { version, date: 'YYYY-MM-DD', changes: [string, ...] }.
 export const CHANGELOG = [

src/pirats/crud_base.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import os
 import random
 from typing import Any, Dict, List, Optional
@@ -6,6 +7,8 @@ from sqlmodel import Session, select
 from .models import Game, Player, GameEvent
 from . import cards
 
+logger = logging.getLogger(__name__)
+
 FACE_VALUES = ("J", "Q", "K")
 
 # --- Card and Hand Utilities ---
@@ -261,6 +264,7 @@ def create_game(db: Session, crew_name: Optional[str] = None) -> Game:
     db.add(game)
     db.commit()
     db.refresh(game)
+    logger.info("Game %s created (crew_name=%r, dev_mode=%s)", game.id, crew_name, game.dev_mode)
     return game
 
 def get_game(db: Session, game_id: str) -> Optional[Game]:
@@ -308,6 +312,10 @@ def add_player(db: Session, game_id: str, name: str, is_creator: bool = False) -
         msg += " They'll create their Pi-Rat with the crew between scenes."
     add_game_event(db, game_id, msg, kind="join")
 
+    logger.info(
+        "Player %s (%r) joined game %s (phase=%s, creator=%s)",
+        player.id, name, game_id, game.phase if game else "?", is_creator,
+    )
     return player
 
 def add_game_event(db: Session, game_id: str, message: str, kind: str = "info"):

src/pirats/crud_scene.py

@@ -1,4 +1,5 @@
 import json
+import logging
 from typing import Tuple, Dict, Any
 from sqlmodel import Session
 from .models import Game, Player, Obstacle
@@ -10,9 +11,17 @@ from .crud_base import (
     is_eligible_nominee
 )
 
+logger = logging.getLogger(__name__)
+
 # --- Scene Setup Operations ---
 
+VALID_ROLES = ("pirat", "deep")
+
 def update_scene_role(db: Session, player_id: str, role: str):
+    # Constrain to the two real roles so the endpoint can't stash an arbitrary
+    # string in the column (defensive: it only ever drives gameplay branches).
+    if role not in VALID_ROLES:
+        return
     player = get_player(db, player_id)
     if not player:
         return
@@ -409,3 +418,4 @@ def finish_game(db: Session, game_id: str):
     db.add(game)
     db.commit()
     add_game_event(db, game_id, "The story has ended! The Pi-Rats party til they pass out in a pile. 🎉", kind="victory")
+    logger.info("Game %s ended after %s scene(s)", game_id, game.current_scene_number)

src/pirats/crud_upkeep.py

@@ -1,3 +1,4 @@
+import logging
 import random
 from typing import List, Tuple
 from sqlmodel import Session, select
@@ -8,6 +9,8 @@ from .crud_base import (
     calculate_max_hand_size, change_player_rank, set_captain, is_eligible_nominee
 )
 
+logger = logging.getLogger(__name__)
+
 # --- Between Scenes Operations ---
 
 def eligible_vote_nominees(game: Game, voter) -> list:
@@ -252,6 +255,7 @@ def _remove_player(db: Session, game: Game, target: Player, event_message: str,
     phase's completion gate so a departed blocker can't stall the table."""
     from . import crud_character as cc
     target_id = target.id
+    logger.info("Removing player %s (%r) from game %s (%s)", target_id, target.name, game.id, event_kind)
 
     if game.captain_player_id == target_id:
         set_captain(db, game, None, reason=captain_reason)

src/pirats/main.py

@@ -1,3 +1,5 @@
+import asyncio
+import contextlib
 from contextlib import asynccontextmanager
 from typing import Optional
 from fastapi import FastAPI, Form, Depends, HTTPException, APIRouter, WebSocket, WebSocketDisconnect
@@ -6,13 +8,17 @@ from fastapi.responses import FileResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 from sqlmodel import Session
 import logging
+from logging.handlers import RotatingFileHandler
 import re
+import sys
 import uvicorn
 import os
 from pathlib import Path
 
 from . import crud
+from . import maintenance
 from .database import run_migrations, get_session, engine
+from .validation import sanitize_name
 from .ws import manager
 
 logger = logging.getLogger(__name__)
@@ -21,11 +27,72 @@ EVENT_PAGE_SIZE = 50
 
 BASE_DIR = Path(__file__).parent
 
+_logging_configured = False
+
+def configure_logging() -> None:
+    """Set up backend logging: always to stderr, plus a rotating file when
+    PIRATS_LOG_FILE is set (the NixOS service points it at /var/log/pirats/).
+    Level comes from PIRATS_LOG_LEVEL (default INFO). Idempotent so it's safe to
+    call from both the CLI entrypoint and the app's startup lifespan."""
+    global _logging_configured
+    if _logging_configured:
+        return
+    _logging_configured = True
+
+    level_name = os.environ.get("PIRATS_LOG_LEVEL", "INFO").upper()
+    level = getattr(logging, level_name, logging.INFO)
+
+    root = logging.getLogger()
+    root.setLevel(level)
+    fmt = logging.Formatter(
+        "%(asctime)s %(levelname)s [%(name)s] %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    stderr_handler = logging.StreamHandler(sys.stderr)
+    stderr_handler.setFormatter(fmt)
+    root.addHandler(stderr_handler)
+
+    log_file = os.environ.get("PIRATS_LOG_FILE")
+    if log_file:
+        try:
+            Path(log_file).parent.mkdir(parents=True, exist_ok=True)
+            file_handler = RotatingFileHandler(
+                log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8"
+            )
+            file_handler.setFormatter(fmt)
+            root.addHandler(file_handler)
+            logger.info("Logging to %s", log_file)
+        except OSError:
+            # A bad/unwritable path mustn't take the server down — keep stderr.
+            logger.exception("Could not open log file %s; logging to stderr only", log_file)
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
+    configure_logging()
+    logger.info(
+        "Starting Rats with Gats backend (db=%s, dev_mode_default=%s)",
+        engine.url.render_as_string(hide_password=True),
+        crud.dev_mode_default(),
+    )
     run_migrations()
+    logger.info("Database migrations applied")
+
+    purge_cfg = maintenance.purge_config()
+    purge_task = None
+    if purge_cfg.enabled:
+        purge_task = asyncio.create_task(maintenance.purge_loop(purge_cfg))
+    else:
+        logger.info("Game purge task disabled (PIRATS_PURGE_ENABLED)")
+
     yield
 
+    if purge_task is not None:
+        purge_task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await purge_task
+    logger.info("Shutting down")
+
 app = FastAPI(title="Rats with Gats Remote Play API", lifespan=lifespan)
 api = APIRouter()
 
@@ -83,7 +150,7 @@ def create_game_route(
     crew_name: str = Form(...),
     db: Session = Depends(get_session)
 ):
-    game = crud.create_game(db, crew_name=crew_name.strip())
+    game = crud.create_game(db, crew_name=sanitize_name(crew_name))
     # admin_key is only revealed here, to the creator; the state endpoint hides it
     return {"id": game.id, "admin_key": game.admin_key}
 
@@ -99,7 +166,7 @@ def join_game_submit(
         
     # First player is automatically the creator
     is_creator = len(game.players) == 0
-    player = crud.add_player(db, game_id, name.strip(), is_creator=is_creator)
+    player = crud.add_player(db, game_id, sanitize_name(name), is_creator=is_creator)
     return {"id": player.id}
 
 @api.post("/game/{game_id}/player/{player_id}/leave")
@@ -189,6 +256,7 @@ if os.path.exists(static_dir):
     app.mount("/", StaticFiles(directory=static_dir, html=True), name="static")
 
 def main():
+    configure_logging()
     import argparse
     parser = argparse.ArgumentParser(description="Run the Rats with Gats web app")
     parser.add_argument("--host", default="0.0.0.0", help="Host address to bind to")

src/pirats/maintenance.py

@@ -0,0 +1,204 @@
+"""Periodic database maintenance: purge old finished/inactive games.
+
+A game is purged when it either finished more than `finished_days` ago or has
+had no activity for `inactive_days`. "Activity" is the timestamp of a game's
+most recent GameEvent (every meaningful move records one); for an ended game
+that last event is its finish, so the same signal serves both windows. Games
+with no events at all are undatable and left alone.
+
+The purge runs in-process on a background asyncio loop (started from the app
+lifespan); all knobs are configurable via environment variables / the NixOS
+service. Every purge is logged with the crew name, uuid and an estimate of the
+bytes it held, and each run that removes anything VACUUMs the SQLite file and
+logs the disk space actually reclaimed.
+"""
+import asyncio
+import json
+import logging
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List, Optional
+
+from sqlalchemy import func
+from sqlmodel import Session, select
+
+from .database import engine
+from .models import Game, GameEvent
+
+logger = logging.getLogger(__name__)
+
+DAY_SECONDS = 86400.0
+
+
+@dataclass(frozen=True)
+class PurgeConfig:
+    enabled: bool
+    interval_hours: float
+    finished_days: float
+    inactive_days: float
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    val = os.environ.get(name)
+    if val is None:
+        return default
+    return val.strip().lower() in ("1", "true", "yes", "on")
+
+
+def _env_float(name: str, default: float) -> float:
+    val = os.environ.get(name)
+    if val is None:
+        return default
+    try:
+        return float(val)
+    except ValueError:
+        logger.warning("Invalid %s=%r; using default %s", name, val, default)
+        return default
+
+
+def purge_config() -> PurgeConfig:
+    """Read purge settings from the environment (see README for the knobs)."""
+    return PurgeConfig(
+        enabled=_env_bool("PIRATS_PURGE_ENABLED", True),
+        interval_hours=_env_float("PIRATS_PURGE_INTERVAL_HOURS", 24.0),
+        finished_days=_env_float("PIRATS_PURGE_FINISHED_DAYS", 14.0),
+        inactive_days=_env_float("PIRATS_PURGE_INACTIVE_DAYS", 30.0),
+    )
+
+
+def _db_file_path() -> Optional[Path]:
+    """The on-disk SQLite file, or None for in-memory / non-file databases."""
+    name = engine.url.database
+    if not name or name == ":memory:":
+        return None
+    return Path(name)
+
+
+def _estimate_game_bytes(game: Game) -> int:
+    """A rough (uncompressed) byte footprint of a game and all its rows, summed
+    from each row's serialized columns. Reported per purge; the dominant term is
+    any leftover rollback checkpoints' state_json."""
+    total = len(json.dumps(game.model_dump(), default=str))
+    for rows in (game.players, game.obstacles, game.votes, game.events,
+                 game.challenges, game.checkpoints):
+        for row in rows:
+            total += len(json.dumps(row.model_dump(), default=str))
+    return total
+
+
+def purge_old_games(db: Session, *, finished_days: float, inactive_days: float,
+                    now: Optional[float] = None) -> List[dict]:
+    """Delete games that finished more than `finished_days` ago or have been
+    inactive for `inactive_days`. Returns a summary dict per purged game and
+    logs each one. Does not VACUUM (the caller does, once per run)."""
+    import time
+    if now is None:
+        now = time.time()
+    finished_cutoff = now - finished_days * DAY_SECONDS
+    inactive_cutoff = now - inactive_days * DAY_SECONDS
+
+    # Most-recent event timestamp per game, in one grouped query.
+    last_activity = {
+        gid: ts
+        for gid, ts in db.exec(
+            select(GameEvent.game_id, func.max(GameEvent.timestamp)).group_by(GameEvent.game_id)
+        ).all()
+    }
+
+    purged: List[dict] = []
+    for game in db.exec(select(Game)).all():
+        last = last_activity.get(game.id)
+        if last is None:
+            # No events ever recorded: we can't date it, so leave it be.
+            continue
+        finished = game.phase == "ended" and last < finished_cutoff
+        inactive = last < inactive_cutoff
+        if not (finished or inactive):
+            continue
+
+        est_bytes = _estimate_game_bytes(game)
+        age_days = (now - last) / DAY_SECONDS
+        reason = "finished" if finished else "inactive"
+        logger.info(
+            "Purging game %s (crew=%r, phase=%s, reason=%s, idle %.1f days, ~%d bytes)",
+            game.id, game.crew_name, game.phase, reason, age_days, est_bytes,
+        )
+
+        # _estimate_game_bytes loaded every relationship, so the ORM's
+        # cascade_delete (set on each Game relationship) deletes all the
+        # children along with the game — no DB-level FK pragma needed.
+        db.delete(game)
+        db.commit()
+
+        purged.append({
+            "id": game.id,
+            "crew_name": game.crew_name,
+            "phase": game.phase,
+            "reason": reason,
+            "idle_days": age_days,
+            "estimated_bytes": est_bytes,
+        })
+
+    return purged
+
+
+def _vacuum() -> None:
+    """Reclaim the pages freed by the deletes. VACUUM cannot run inside a
+    transaction, hence the autocommit connection."""
+    with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
+        conn.exec_driver_sql("VACUUM")
+
+
+def run_purge(*, finished_days: float, inactive_days: float,
+              now: Optional[float] = None) -> List[dict]:
+    """One full purge cycle: delete stale games, then VACUUM and report the
+    disk space actually reclaimed from the SQLite file."""
+    db_path = _db_file_path()
+    size_before = db_path.stat().st_size if db_path and db_path.exists() else None
+
+    with Session(engine) as db:
+        purged = purge_old_games(
+            db, finished_days=finished_days, inactive_days=inactive_days, now=now
+        )
+
+    if not purged:
+        logger.debug("Game purge: nothing to remove")
+        return purged
+
+    if db_path and db_path.exists():
+        _vacuum()
+        size_after = db_path.stat().st_size
+        reclaimed = (size_before - size_after) if size_before is not None else None
+        logger.info(
+            "Game purge removed %d game(s); reclaimed %s bytes from %s",
+            len(purged),
+            reclaimed if reclaimed is not None else "?",
+            db_path,
+        )
+    else:
+        logger.info("Game purge removed %d game(s)", len(purged))
+    return purged
+
+
+async def purge_loop(config: Optional[PurgeConfig] = None) -> None:
+    """Background task: run a purge at startup, then every `interval_hours`."""
+    cfg = config or purge_config()
+    logger.info(
+        "Game purge task started: every %.1f h, finished > %.0f days, inactive > %.0f days",
+        cfg.interval_hours, cfg.finished_days, cfg.inactive_days,
+    )
+    interval_seconds = max(cfg.interval_hours, 0.0) * 3600.0
+    while True:
+        try:
+            from fastapi.concurrency import run_in_threadpool
+            await run_in_threadpool(
+                run_purge,
+                finished_days=cfg.finished_days,
+                inactive_days=cfg.inactive_days,
+            )
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("Game purge run failed")
+        await asyncio.sleep(interval_seconds)

src/pirats/routes_challenge.py

@@ -4,6 +4,7 @@ from fastapi.responses import JSONResponse
 from sqlmodel import Session
 from .database import get_session
 from . import crud
+from .validation import sanitize_text
 
 router = APIRouter()
 
@@ -23,7 +24,8 @@ def create_challenge_route(
         assert isinstance(ids, list)
     except Exception:
         return JSONResponse({"error": "Invalid obstacle list."}, status_code=400)
-    ok, msg = crud.create_challenge(db, game_id, player_id, target_player_id, ids, stakes_success, stakes_failure)
+    ok, msg = crud.create_challenge(db, game_id, player_id, target_player_id, ids,
+                                    sanitize_text(stakes_success), sanitize_text(stakes_failure))
     if not ok:
         return JSONResponse({"error": msg}, status_code=400)
     return {"status": "ok"}

src/pirats/routes_character.py

@@ -4,6 +4,7 @@ from sqlmodel import Session
 from .database import get_session
 from . import crud
 from .cards import TECHNIQUE_SUGGESTIONS
+from .validation import sanitize_text, sanitize_name
 
 router = APIRouter()
 
@@ -27,10 +28,10 @@ def player_save_basic_details(
     if not player:
         raise HTTPException(status_code=404, detail="Player not found")
         
-    player.avatar_look = avatar_look.strip()
-    player.avatar_smell = avatar_smell.strip()
-    player.first_words = first_words.strip()
-    player.good_at_math = good_at_math.strip()
+    player.avatar_look = sanitize_text(avatar_look)
+    player.avatar_smell = sanitize_name(avatar_smell)  # seeds the "Recruit {smell}" name
+    player.first_words = sanitize_text(first_words)
+    player.good_at_math = sanitize_text(good_at_math)
     # Until a Pi-Rat earns a Name they are identified by their smell
     if not player.completed_personal_2:
         player.name = crud.recruit_name(player)
@@ -98,7 +99,7 @@ def player_submit_delegated_answer(
     answer: str = Form(...),
     db: Session = Depends(get_session)
 ):
-    crud.submit_delegated_answer(db, target_player_id, question_type, answer.strip(), player_id)
+    crud.submit_delegated_answer(db, target_player_id, question_type, sanitize_text(answer), player_id)
     return {"status": "ok"}
 
 # Submit 3 Secret Pirate Techniques
@@ -111,7 +112,7 @@ def player_submit_techniques(
     tech3: str = Form(...),
     db: Session = Depends(get_session)
 ):
-    error = crud.submit_techniques(db, player_id, tech1.strip(), tech2.strip(), tech3.strip())
+    error = crud.submit_techniques(db, player_id, sanitize_text(tech1), sanitize_text(tech2), sanitize_text(tech3))
     if error:
         return JSONResponse({"error": error}, status_code=400)
     return {"status": "ok"}
@@ -234,7 +235,7 @@ def contribute_technique_route(
     text: str = Form(...),
     db: Session = Depends(get_session)
 ):
-    ok, msg = crud.contribute_recruit_technique(db, player_id, recruit_id, slot, text)
+    ok, msg = crud.contribute_recruit_technique(db, player_id, recruit_id, slot, sanitize_text(text))
     if not ok:
         return JSONResponse({"error": msg}, status_code=400)
     return {"status": "ok"}

src/pirats/routes_scene.py

@@ -4,6 +4,7 @@ from sqlmodel import Session
 from .database import get_session
 from . import crud
 from .cards import OBSTACLES_DATA
+from .validation import sanitize_text, sanitize_name
 
 router = APIRouter()
 
@@ -124,7 +125,7 @@ def set_name_route(
 ):
     player = crud.get_player(db, player_id)
     if player and player.needs_name:
-        player.name = new_name
+        player.name = sanitize_name(new_name)
         player.needs_name = False
         db.add(player)
         db.commit()
@@ -140,7 +141,7 @@ def set_gat_description_route(
 ):
     player = crud.get_player(db, player_id)
     if player and player.needs_gat_description:
-        player.gat_description = description
+        player.gat_description = sanitize_text(description)
         player.needs_gat_description = False
         db.add(player)
         db.commit()

src/pirats/validation.py

@@ -0,0 +1,41 @@
+"""Light input hardening for free-text fields submitted by players.
+
+The app is deliberately cooperative and light on auth (see the trust model), but
+it's exposed on the open internet, so player-authored free text gets a basic
+scrub before it is stored: drop control characters and unpaired surrogates that
+could corrupt JSON or the display, trim surrounding whitespace, and cap the
+length. SQL injection is already handled by SQLModel's parameterized queries —
+this is only about content hygiene and bounding payload size.
+
+Only fields whose value is authored-and-stored go through here. Identifiers
+(player/obstacle ids), enums (role, objective type), and values matched against
+existing stored text (a swap offer, a face-card technique assignment) are left
+untouched so they still compare equal.
+"""
+import unicodedata
+
+MAX_NAME_LENGTH = 120     # crew / player / Pi-Rat names and the smell that seeds them
+MAX_TEXT_LENGTH = 2000    # everything else: catchphrases, techniques, stakes, descriptions
+
+
+def sanitize_text(value: str, max_length: int = MAX_TEXT_LENGTH) -> str:
+    """`value` with control characters and unpaired surrogates removed,
+    surrounding whitespace trimmed, and length capped at `max_length`.
+    Tabs and newlines are kept; emoji (incl. ZWJ sequences) survive."""
+    if not value:
+        return ""
+    cleaned = "".join(
+        ch for ch in value
+        if ch in ("\t", "\n")
+        or (unicodedata.category(ch) != "Cc" and not 0xD800 <= ord(ch) <= 0xDFFF)
+    )
+    cleaned = cleaned.strip()
+    if len(cleaned) > max_length:
+        cleaned = cleaned[:max_length].rstrip()
+    return cleaned
+
+
+def sanitize_name(value: str) -> str:
+    """Sanitize a single-line, name-like field: the tighter name cap, plus
+    collapsing any internal whitespace runs to single spaces."""
+    return " ".join(sanitize_text(value, max_length=MAX_NAME_LENGTH).split())

tests/test_game.py

@@ -3,7 +3,7 @@ import json
 from sqlmodel import SQLModel, create_engine, Session, select
 from pirats import cards
 from pirats import crud
-from pirats.models import Obstacle, Player, Challenge, GameEvent, Checkpoint
+from pirats.models import Game, Obstacle, Player, Challenge, GameEvent, Checkpoint
 
 # In-memory database for testing
 @pytest.fixture(name="session")
@@ -2410,3 +2410,118 @@ def test_rollback_route_enforces_permission():
         assert r.json()["status"] == "ok"
     finally:
         app.dependency_overrides.clear()
+
+
+def _seed_game(session, crew_name, phase, last_event_ts):
+    """A game whose most-recent GameEvent is at `last_event_ts`."""
+    game = Game(crew_name=crew_name, phase=phase)
+    session.add(game)
+    session.commit()
+    session.refresh(game)
+    session.add(GameEvent(game_id=game.id, message="move", timestamp=last_event_ts - 10))
+    session.add(GameEvent(game_id=game.id, message="move", timestamp=last_event_ts))
+    session.commit()
+    return game
+
+
+def test_purge_old_games(session):
+    from pirats import maintenance
+
+    now = 2_000_000_000.0
+    DAY = 86400.0
+
+    active = _seed_game(session, "Active", "scene", now - 2 * DAY)
+    finished_old = _seed_game(session, "DoneOld", "ended", now - 20 * DAY)
+    finished_recent = _seed_game(session, "DoneNew", "ended", now - 3 * DAY)
+    inactive = _seed_game(session, "Ghosted", "scene", now - 40 * DAY)
+
+    # A game with no events at all is undatable and must be left alone.
+    empty = Game(crew_name="Empty", phase="lobby")
+    session.add(empty)
+    session.commit()
+    session.refresh(empty)
+
+    # Give the finished-old game a player + checkpoint to prove children are cleared.
+    session.add(Player(game_id=finished_old.id, name="Crewmate"))
+    session.add(Checkpoint(game_id=finished_old.id, state_json="{}"))
+    session.commit()
+
+    purged = maintenance.purge_old_games(session, finished_days=14, inactive_days=30, now=now)
+    purged_ids = {p["id"] for p in purged}
+
+    assert finished_old.id in purged_ids   # ended > 14 days ago
+    assert inactive.id in purged_ids       # no activity for > 30 days
+    assert active.id not in purged_ids     # recently active
+    assert finished_recent.id not in purged_ids  # ended only 3 days ago
+    assert empty.id not in purged_ids      # undatable
+
+    remaining = {g.id for g in session.exec(select(Game)).all()}
+    assert finished_old.id not in remaining
+    assert inactive.id not in remaining
+    assert active.id in remaining
+    assert empty.id in remaining
+
+    # Children of a purged game are gone.
+    assert session.exec(select(GameEvent).where(GameEvent.game_id == finished_old.id)).first() is None
+    assert session.exec(select(Player).where(Player.game_id == finished_old.id)).first() is None
+    assert session.exec(select(Checkpoint).where(Checkpoint.game_id == finished_old.id)).first() is None
+
+    # Summaries carry the reason and a positive byte estimate.
+    summary = next(p for p in purged if p["id"] == finished_old.id)
+    assert summary["reason"] == "finished"
+    assert summary["crew_name"] == "DoneOld"
+    assert summary["estimated_bytes"] > 0
+
+
+def test_sanitize_text_and_name():
+    from pirats.validation import (
+        sanitize_text, sanitize_name, MAX_TEXT_LENGTH, MAX_NAME_LENGTH,
+    )
+
+    # Control characters dropped, surrounding whitespace trimmed.
+    assert sanitize_text("  hi\x00\x07there  ") == "hithere"
+    # Tabs/newlines kept inside the text.
+    assert sanitize_text("a\nb\tc") == "a\nb\tc"
+    # Zero-width joiner (emoji sequences) survives.
+    assert sanitize_text("a‍b") == "a‍b"
+    # Lone surrogate removed without raising.
+    assert sanitize_text("a" + chr(0xD800) + "b") == "ab"
+    # Length capped.
+    assert len(sanitize_text("x" * (MAX_TEXT_LENGTH + 50))) == MAX_TEXT_LENGTH
+    # Empty / falsy in -> empty out.
+    assert sanitize_text("") == ""
+
+    # Names collapse internal whitespace and use the tighter cap.
+    assert sanitize_name("  Captain   Redbeard\n ") == "Captain Redbeard"
+    assert len(sanitize_name("y" * (MAX_NAME_LENGTH + 10))) == MAX_NAME_LENGTH
+
+
+def test_create_and_join_sanitize_names():
+    from fastapi.testclient import TestClient
+    from sqlalchemy.pool import StaticPool
+    from pirats.main import app
+    from pirats.database import get_session
+    from sqlmodel import SQLModel, create_engine, Session
+
+    engine = create_engine(
+        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool,
+    )
+    SQLModel.metadata.create_all(engine)
+    session = Session(engine)
+
+    def override():
+        yield session
+
+    app.dependency_overrides[get_session] = override
+    client = TestClient(app)
+    try:
+        r = client.post("/api/game", data={"crew_name": "  The\x00 Salty\x07  Dogs  "})
+        gid = r.json()["id"]
+        assert session.get(Game, gid).crew_name == "The Salty Dogs"
+
+        r = client.post(f"/api/game/{gid}/join", data={"name": "  Bad\x01name  "})
+        player = session.get(Player, r.json()["id"])
+        assert player.player_name == "Badname"
+        assert "\x01" not in player.name
+    finally:
+        app.dependency_overrides.clear()