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>
This commit is contained in:
2026-06-15 15:54:21 -07:00
parent ccb32e7103
commit 122e66b817
9 changed files with 120 additions and 14 deletions

View File

@@ -2,7 +2,7 @@
- [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] **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] **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] **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) ## Words Words Words (Waiting for Fable to come back)

View File

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

View File

@@ -15,7 +15,13 @@ logger = logging.getLogger(__name__)
# --- Scene Setup Operations --- # --- Scene Setup Operations ---
VALID_ROLES = ("pirat", "deep")
def update_scene_role(db: Session, player_id: str, role: str): 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) player = get_player(db, player_id)
if not player: if not player:
return return

View File

@@ -18,6 +18,7 @@ from pathlib import Path
from . import crud from . import crud
from . import maintenance from . import maintenance
from .database import run_migrations, get_session, engine from .database import run_migrations, get_session, engine
from .validation import sanitize_name
from .ws import manager from .ws import manager
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -149,7 +150,7 @@ def create_game_route(
crew_name: str = Form(...), crew_name: str = Form(...),
db: Session = Depends(get_session) 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 # admin_key is only revealed here, to the creator; the state endpoint hides it
return {"id": game.id, "admin_key": game.admin_key} return {"id": game.id, "admin_key": game.admin_key}
@@ -165,7 +166,7 @@ def join_game_submit(
# First player is automatically the creator # First player is automatically the creator
is_creator = len(game.players) == 0 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} return {"id": player.id}
@api.post("/game/{game_id}/player/{player_id}/leave") @api.post("/game/{game_id}/player/{player_id}/leave")

View File

@@ -4,6 +4,7 @@ from fastapi.responses import JSONResponse
from sqlmodel import Session from sqlmodel import Session
from .database import get_session from .database import get_session
from . import crud from . import crud
from .validation import sanitize_text
router = APIRouter() router = APIRouter()
@@ -23,7 +24,8 @@ def create_challenge_route(
assert isinstance(ids, list) assert isinstance(ids, list)
except Exception: except Exception:
return JSONResponse({"error": "Invalid obstacle list."}, status_code=400) 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: if not ok:
return JSONResponse({"error": msg}, status_code=400) return JSONResponse({"error": msg}, status_code=400)
return {"status": "ok"} return {"status": "ok"}

View File

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

View File

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

41
src/pirats/validation.py Normal file
View File

@@ -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())

View File

@@ -2471,3 +2471,57 @@ def test_purge_old_games(session):
assert summary["reason"] == "finished" assert summary["reason"] == "finished"
assert summary["crew_name"] == "DoneOld" assert summary["crew_name"] == "DoneOld"
assert summary["estimated_bytes"] > 0 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("ab") == "ab"
# 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()