diff --git a/backend/app/main.py b/backend/app/main.py index 229aca3..c2c43c1 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -69,6 +69,20 @@ async def lifespan(app: FastAPI): await conn.execute(text( "ALTER TABLE entities ADD COLUMN IF NOT EXISTS at_peace BOOLEAN NOT NULL DEFAULT false" )) + # Defense-in-depth: purchase_unlock() already enforces one row per + # (user, unlock_key) via a row-locked check-then-insert, so this + # constraint should never actually find a conflict on a live DB. + # `ADD CONSTRAINT` has no IF NOT EXISTS form, so the guard is a + # catalog check instead — safe to run on every startup. + await conn.execute(text( + "DO $$ BEGIN " + "IF NOT EXISTS (" + " SELECT 1 FROM pg_constraint WHERE conname = 'uq_unlocks_user_key'" + ") THEN " + " ALTER TABLE unlocks ADD CONSTRAINT uq_unlocks_user_key UNIQUE (user_id, unlock_key); " + "END IF; " + "END $$;" + )) cleanup_task = asyncio.create_task(_session_cleanup_loop()) try: yield diff --git a/backend/app/models/unlock.py b/backend/app/models/unlock.py index 22190fc..e3cf064 100644 --- a/backend/app/models/unlock.py +++ b/backend/app/models/unlock.py @@ -1,7 +1,7 @@ import uuid from datetime import datetime, timezone -from sqlalchemy import DateTime, ForeignKey, String +from sqlalchemy import DateTime, ForeignKey, String, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from app.db import Base @@ -9,9 +9,18 @@ from app.db import Base class UnlockRecord(Base): """A permanent unlock a seeker has purchased with essence (e.g. the - listening tool). One row per (user, unlock_key).""" + listening tool). One row per (user, unlock_key). + + The unique constraint is defense-in-depth: today, idempotency (no double + charge for an already-owned unlock) is enforced entirely by + `app.inventory.purchase_unlock`'s row-locked check-then-insert. This + constraint means any *other* code path that ever inserts an + UnlockRecord without going through that lock still can't create a + duplicate row for the same (user, unlock_key). + """ __tablename__ = "unlocks" + __table_args__ = (UniqueConstraint("user_id", "unlock_key", name="uq_unlocks_user_key"),) id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4) user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"), index=True) diff --git a/backend/app/ws.py b/backend/app/ws.py index 01cc365..adadcc2 100644 --- a/backend/app/ws.py +++ b/backend/app/ws.py @@ -29,6 +29,7 @@ from pathlib import Path from fastapi import APIRouter, WebSocket, WebSocketDisconnect from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from app import judgment from app.config import settings @@ -68,6 +69,15 @@ fragment_limiter = RateLimiter(max_requests=30, window_seconds=60) question_limiter = RateLimiter(max_requests=6, window_seconds=60) summon_limiter = RateLimiter(max_requests=4, window_seconds=60) +# ritual_start/judgment don't call the LLM, but each one is an essence/favor/ +# item-drop reward trigger point same as summon/question — unlike those, +# they previously had no limiter at all, which meant a scripted client could +# credit itself unbounded essence by simply replaying "judgment" (or +# ritual_start -> 4x ritual_step) in a tight loop. These bound that to the +# same modest, human-plausible cadence as the other reward triggers. +ritual_limiter = RateLimiter(max_requests=6, window_seconds=60) +judgment_limiter = RateLimiter(max_requests=10, window_seconds=60) + # Per-IP limiters for the same trigger points (spec §5). Ollama is a shared, # single-instance, CPU-only resource — per-account limits alone don't stop # one compromised/scripted account from hammering it across many source IPs, @@ -77,6 +87,8 @@ summon_limiter = RateLimiter(max_requests=4, window_seconds=60) fragment_ip_limiter = RateLimiter(max_requests=60, window_seconds=60) question_ip_limiter = RateLimiter(max_requests=12, window_seconds=60) summon_ip_limiter = RateLimiter(max_requests=8, window_seconds=60) +ritual_ip_limiter = RateLimiter(max_requests=12, window_seconds=60) +judgment_ip_limiter = RateLimiter(max_requests=20, window_seconds=60) AUDIO_DIR = Path(settings.data_dir) / "audio" @@ -287,55 +299,73 @@ async def _summon(state: SeanceState, channel: str) -> tuple[Entity, bool]: str(state.session_id) ) - async with session_maker() as db: - entity = await db.scalar( - select(Entity).where(Entity.signature == signature, Entity.at_peace.is_(False)) - ) - is_new = entity is None + # Two concurrent sessions can race to mint the same signature (this is + # the whole point of the match-or-mint check above being racy across + # connections), or two mints can independently decide on the same + # "next available" name via _unique_entity_name's read-then-decide + # check — either raises IntegrityError on commit against Entity's + # unique(signature)/unique(name) constraints. Retrying re-runs the + # match against what the winning transaction just committed, so the + # loser finds and reuses that row instead of crashing the session. + last_error: IntegrityError | None = None + for _attempt in range(2): + async with session_maker() as db: + try: + entity = await db.scalar( + select(Entity).where(Entity.signature == signature, Entity.at_peace.is_(False)) + ) + is_new = entity is None - if is_new: - mint_signature = signature - retired = await db.scalar(select(Entity).where(Entity.signature == signature)) - if retired is not None: - mint_signature = f"{signature}:{uuid.uuid4().hex[:8]}" + if is_new: + mint_signature = signature + retired = await db.scalar(select(Entity).where(Entity.signature == signature)) + if retired is not None: + mint_signature = f"{signature}:{uuid.uuid4().hex[:8]}" - profile = await spirit_service.mint_profile( - mint_signature, channel, state.anomalies, state.language - ) - discoverer = await db.get(User, state.user_id) - favor = discoverer.favor if discoverer is not None else 0.0 - entity = Entity( - name=await _unique_entity_name(db, profile["name"]), - epithet=profile["epithet"], - persona=profile["persona"], - rarity_tier=profile["rarity"], - signature=mint_signature, - voice_profile=profile["voice"], - visual_profile=profile["visual"], - sample_quotes=profile["quotes"], - # Workstream B: signature-seeded traits, nudged by the - # discovering user's favor (app.judgment.apply_favor_bias) — - # never derived from/fed into the persona above. - traits=judgment.apply_favor_bias(profile["traits"], favor), - discovered_by=state.user_id, - contact_count=1, - ) - db.add(entity) - else: - entity.contact_count += 1 + profile = await spirit_service.mint_profile( + mint_signature, channel, state.anomalies, state.language + ) + discoverer = await db.get(User, state.user_id) + favor = discoverer.favor if discoverer is not None else 0.0 + entity = Entity( + name=await _unique_entity_name(db, profile["name"]), + epithet=profile["epithet"], + persona=profile["persona"], + rarity_tier=profile["rarity"], + signature=mint_signature, + voice_profile=profile["voice"], + visual_profile=profile["visual"], + sample_quotes=profile["quotes"], + # Workstream B: signature-seeded traits, nudged by the + # discovering user's favor (app.judgment.apply_favor_bias) — + # never derived from/fed into the persona above. + traits=judgment.apply_favor_bias(profile["traits"], favor), + discovered_by=state.user_id, + contact_count=1, + ) + db.add(entity) + else: + entity.contact_count += 1 - await db.flush() - session = await db.get(ContactSession, state.session_id) - if session is not None: - session.entity_id = entity.id - db.add( - EntitySighting( - entity_id=entity.id, session_id=state.session_id, user_id=state.user_id - ) - ) - await db.commit() - await db.refresh(entity) - return entity, is_new + await db.flush() + session = await db.get(ContactSession, state.session_id) + if session is not None: + session.entity_id = entity.id + db.add( + EntitySighting( + entity_id=entity.id, session_id=state.session_id, user_id=state.user_id + ) + ) + await db.commit() + await db.refresh(entity) + return entity, is_new + except IntegrityError as exc: + await db.rollback() + last_error = exc + continue + + assert last_error is not None + raise last_error async def _reward_summon(state: SeanceState) -> None: @@ -353,7 +383,10 @@ async def _reward_summon(state: SeanceState) -> None: rarity = state.entity.get("rarity", "common") async with session_maker() as db: - user = await db.get(User, state.user_id) + # Locked (see purchase_unlock's docstring in app/inventory.py) so + # this read-modify-write on essence can't race a concurrent + # purchase's own locked deduction and silently clobber it. + user = await db.scalar(select(User).where(User.id == state.user_id).with_for_update()) if user is None: return credit_essence(user, SUMMON_ESSENCE_TRICKLE) @@ -480,7 +513,14 @@ async def _handle_question(state: SeanceState, text: str) -> None: if state.entity is None: await _handle_summon(state) - assert state.entity is not None + if state.entity is None: + # _handle_summon() returns without setting state.entity when + # the seeker is rate-limited (it already sent its own + # "rate_limited" error frame in that case) — bail out here + # instead of asserting, which would raise uncaught and crash + # this session's whole WS message loop (only WebSocketDisconnect + # is caught around it in session_socket()). + return text = text.strip()[:500] await _record_event(state.session_id, "question", text=text) @@ -582,6 +622,18 @@ RITUAL_STEPS_REQUIRED = 4 async def _handle_ritual_start(state: SeanceState) -> None: if state.entity is None: return # no presence to focus on — frontend already gates the button + if not ( + ritual_limiter.allow(str(state.user_id)) + and ritual_ip_limiter.allow(state.client_ip) + ): + await state.send_queue.put( + { + "type": "error", + "code": "rate_limited", + "message": "The channel needs a moment to settle before it can be focused again.", + } + ) + return state.ritual_steps = 0 state.ritual_completed = False state.ritual_success = False @@ -592,7 +644,8 @@ async def _reward_ritual_success(state: SeanceState) -> None: ritual milestone trigger point.""" item = None async with session_maker() as db: - user = await db.get(User, state.user_id) + # Locked — see the same comment in _reward_summon above. + user = await db.scalar(select(User).where(User.id == state.user_id).with_for_update()) if user is None: return credit_essence(user, RITUAL_SUCCESS_ESSENCE) @@ -640,6 +693,18 @@ async def _handle_judgment(state: SeanceState, message: dict) -> None: verdict = message.get("verdict") if verdict not in judgment.VERDICTS: return + if not ( + judgment_limiter.allow(str(state.user_id)) + and judgment_ip_limiter.allow(state.client_ip) + ): + await state.send_queue.put( + { + "type": "error", + "code": "rate_limited", + "message": "The veil needs a moment before it can render another verdict.", + } + ) + return traits = state.entity.get("traits", {}) outcome = judgment.judge_verdict( @@ -658,7 +723,9 @@ async def _handle_judgment(state: SeanceState, message: dict) -> None: "crossed_over", ): async with session_maker() as db: - user = await db.get(User, state.user_id) + # Locked — see the same comment in _reward_summon above; this + # path writes essence too, so it's exposed to the same race. + user = await db.scalar(select(User).where(User.id == state.user_id).with_for_update()) if user is not None: if outcome.favor_delta: user.favor = judgment.clamp_favor(user.favor + outcome.favor_delta) diff --git a/backend/tests/test_ws_ritual_judgment.py b/backend/tests/test_ws_ritual_judgment.py index 5f8cbce..a53b863 100644 --- a/backend/tests/test_ws_ritual_judgment.py +++ b/backend/tests/test_ws_ritual_judgment.py @@ -59,6 +59,10 @@ def _fake_spirits(monkeypatch): monkeypatch.setattr(app.ws, "question_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60)) monkeypatch.setattr(app.ws, "fragment_limiter", RateLimiter(max_requests=1000, window_seconds=60)) monkeypatch.setattr(app.ws, "fragment_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60)) + monkeypatch.setattr(app.ws, "ritual_limiter", RateLimiter(max_requests=1000, window_seconds=60)) + monkeypatch.setattr(app.ws, "ritual_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60)) + monkeypatch.setattr(app.ws, "judgment_limiter", RateLimiter(max_requests=1000, window_seconds=60)) + monkeypatch.setattr(app.ws, "judgment_ip_limiter", RateLimiter(max_requests=1000, window_seconds=60)) def _read_until(ws, msg_type, max_frames=60, **match):