diff --git a/backend/app/main.py b/backend/app/main.py index 88b9b1b..4711ee4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -129,6 +129,45 @@ async def lifespan(app: FastAPI): "END IF; " "END $$;" )) + + # The durable reward ledger (app/models/award_claim.py). `create_all` + # above already builds this table on a fresh database; these two + # statements are for an install that predates it — the CREATE is a + # no-op there, and the constraint is what actually enforces + # once-per-(seeker, presence, milestone), so it is asserted explicitly + # rather than left to whatever the table happened to be created with. + await conn.execute(text( + """ + CREATE TABLE IF NOT EXISTS award_claims ( + id UUID PRIMARY KEY, + user_id UUID NOT NULL REFERENCES users(id), + entity_id UUID NOT NULL REFERENCES entities(id), + award_key VARCHAR(64) NOT NULL, + claimed_at TIMESTAMPTZ NOT NULL DEFAULT now() + ) + """ + )) + await conn.execute(text( + "CREATE INDEX IF NOT EXISTS ix_award_claims_user_id ON award_claims (user_id)" + )) + await conn.execute(text( + "CREATE INDEX IF NOT EXISTS ix_award_claims_entity_id ON award_claims (entity_id)" + )) + # Same catalog-check shape as uq_unlocks_user_key above: `ADD + # CONSTRAINT` has no IF NOT EXISTS form. Unlike that one this + # constraint is not defense-in-depth — it IS the guard: _claim_award + # relies on the IntegrityError it raises to resolve two concurrent + # connections claiming the same award down to a single payout. + await conn.execute(text( + "DO $$ BEGIN " + "IF NOT EXISTS (" + " SELECT 1 FROM pg_constraint WHERE conname = 'uq_award_claims_user_entity_key'" + ") THEN " + " ALTER TABLE award_claims ADD CONSTRAINT uq_award_claims_user_entity_key " + " UNIQUE (user_id, entity_id, award_key); " + "END IF; " + "END $$;" + )) cleanup_task = asyncio.create_task(_session_cleanup_loop()) try: yield diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index ec024bf..f2507d3 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -1,4 +1,5 @@ from app.models.auth_session import AuthSession +from app.models.award_claim import AwardClaim from app.models.contact_session import ContactSession from app.models.device import Device from app.models.entity import Entity @@ -14,6 +15,7 @@ from app.models.waitlist_entry import WaitlistEntry __all__ = [ "User", "AuthSession", + "AwardClaim", "ContactSession", "Device", "Entity", diff --git a/backend/app/models/award_claim.py b/backend/app/models/award_claim.py new file mode 100644 index 0000000..fb67556 --- /dev/null +++ b/backend/app/models/award_claim.py @@ -0,0 +1,47 @@ +import uuid +from datetime import datetime, timezone + +from sqlalchemy import DateTime, ForeignKey, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.db import Base + + +class AwardClaim(Base): + """One row per milestone a seeker has ACTUALLY been paid for a presence. + + The three reward faucets in `app/ws.py` (the ritual milestone, a verdict, + a Passage beat) each used to be guarded by a set held on the in-memory + `SeanceState`. That guard is per-CONNECTION: a client that disconnects and + reconnects — or simply opens a second websocket — got a fresh, empty guard + and could be paid all over again for the same spirit. `summon_limiter` + bounded the rate of that, never the total. + + This table is the durable form of those guards. The UNIQUE constraint on + (user_id, entity_id, award_key) is the actual enforcement, not a + belt-and-braces afterthought: `app.ws._claim_award` claims a milestone by + INSERTing here and treating an IntegrityError as "somebody already paid + this", which is what makes two connections racing the same award resolve + to exactly one payout instead of two. + + `award_key` is the milestone within a presence, not a currency: + `ritual`, `judgment:`, `passage:`, and `crossing` (claimed + by BOTH roads to a crossing — the Passage's `release` beat and the + `cross_over` verdict — so one spirit crosses once whichever road got there + first). + """ + + __tablename__ = "award_claims" + __table_args__ = ( + UniqueConstraint( + "user_id", "entity_id", "award_key", name="uq_award_claims_user_entity_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) + entity_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("entities.id"), index=True) + award_key: Mapped[str] = mapped_column(String(64)) + claimed_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) + ) diff --git a/backend/app/ws.py b/backend/app/ws.py index 98375f9..b870d62 100644 --- a/backend/app/ws.py +++ b/backend/app/ws.py @@ -55,6 +55,7 @@ from app.inventory import ( ) from app.llm.service import SpiritBusyError, spirit_service from app.models.auth_session import AuthSession, hash_token +from app.models.award_claim import AwardClaim from app.models.contact_session import ContactSession from app.models.entity import Entity from app.models.entity_sighting import EntitySighting @@ -153,31 +154,24 @@ class SeanceState: ritual_steps: int = 0 ritual_completed: bool = False ritual_success: bool = False - # Whether the ritual milestone has ALREADY paid out for the current - # presence. Exactly the same faucet `passage_paid` closes, on the other - # rite: `ritual_start` rewinds `ritual_completed` to False at any time - # (and the UI offers precisely that button — RitualPanel's "attempt - # again", which honest play needs after a FAILED roll), so without this - # a client could walk ritual_start + 4x ritual_step for +15 essence and - # an item roll, restart, and repeat forever off a single summon. The - # limiter caps the cadence, not the total. So the milestone pays the - # FIRST time it succeeds for a given presence and never again; - # re-attempting still rolls, still reveals the traits on success, but - # mints no new essence and rolls no new item. Cleared only on a fresh - # summon, never by `ritual_start` — that is the whole point. - ritual_paid: bool = False - # Which verdicts have ALREADY been applied to the ledger for the current - # presence. Same class of hole: `judgment` had no once-per-presence - # guard at all except for `cross_over`, so replaying {"type": - # "judgment", "verdict": "trust"} against a benevolent spirit credited - # CORRECT_JUDGMENT_ESSENCE *and* +0.05 favor *and* rolled an item on - # every single frame — an unbounded faucet for both currencies (favor - # pins to +1.0, which then biases every future mint) that the 10/60s - # limiter only slowed down. Keyed by verdict rather than a single latch - # so an honest correction (a wrong `trust`, then the right `banish` on - # the same spirit) still resolves normally — what can never repeat is - # the SAME verdict on the SAME presence. Cleared only on a fresh summon. - judged_verdicts: set[str] = field(default_factory=set) + # Awards this connection has already OBSERVED to be claimed, as + # (entity_id, award_key) pairs — a pure read-through cache of rows in + # `award_claims` (see `_claim_award`), never the guard itself. + # + # It exists only to spare the DB a doomed INSERT on the common replay + # path. It is safe precisely because it caches one direction: an entry + # means "definitely already paid" (a fact that can never become false — + # rows are never deleted), and its ABSENCE means nothing at all, so a + # miss always falls through to the database. It is deliberately never + # cleared on a summon; a stale entry would only ever be correct. + # + # What used to live here instead — `ritual_paid: bool`, + # `judged_verdicts: set[str]`, `passage_paid: set[str]` — were the guards + # themselves, and that was the hole: they are per-CONNECTION. Every one + # of the faucets they close (see the comments on the handlers below) + # reopened in full the moment a client dropped the socket and + # reconnected, or simply opened a second one alongside the first. + awards_claimed: set[tuple[uuid.UUID, str]] = field(default_factory=set) # Workstream L (passage-doctrine spec): where the layered crossing rite # stands for the *current* entity. `passage_layer` is the next beat to # attempt, `passage_lied` remembers whether the layer just completed was @@ -188,17 +182,6 @@ class SeanceState: passage_layer: str = passage.FIRST_LAYER passage_lied: bool = False passage_crossed: bool = False - # Which beats have ALREADY paid out for the current entity. Without - # this, essence is unbounded: `passage_start` rewinds the rite to - # `listen` at any time (and the UI offers exactly that button after a - # resisted release), so a client could walk listen/name/unbind/open for - # +28 essence, restart, and repeat forever off a single summon — the - # limiter caps the rate, not the total. A volatility collapse rewinds it - # the same way. So each layer pays the FIRST time it opens for a given - # presence and never again; re-walking it still reveals, still costs - # nothing already banked, but mints no new essence. Cleared only on a - # fresh summon, never by `passage_start` — that is the whole point. - passage_paid: set[str] = field(default_factory=set) # Per-session RNG for `tell` frames — unseeded (a session's tells should # vary run to run), but persistent across calls so the draw sequence # isn't restarted on every single message. @@ -513,6 +496,46 @@ async def _summon(state: SeanceState, channel: str) -> tuple[Entity, bool]: raise last_error +async def _claim_award(state: SeanceState, entity_id: str, award_key: str) -> bool: + """Claim a milestone for (this seeker, this presence). True exactly once. + + This is the durable replacement for the three per-connection guards that + used to live on `SeanceState`. Every reward path in this file that is + meant to pay once per presence asks here first, and pays only on True. + + The `award_claims` UNIQUE constraint is the real guard, not the SELECT + that never happens: the claim is a bare INSERT, and losing the race is an + IntegrityError, which we swallow and report as "already paid". That is + the only shape that holds when two websockets for the same account claim + the same award at the same instant — a check-then-insert would let both + read "unclaimed" and both pay. Same handling as `_summon`'s signature/name + collision above: catch, roll back, and take the winner's outcome as the + answer, rather than letting a 500 kill the séance. + + Rows are never deleted, so a claim is permanent for that spirit — which is + the whole point. A genuinely NEW presence is a different `entity_id` and + therefore a fresh, unclaimed set of milestones; see the note in + `_summon_locked`. + """ + key = (uuid.UUID(entity_id), award_key) + if key in state.awards_claimed: + return False # fast path only; a miss still asks the database + + async with session_maker() as db: + db.add( + AwardClaim(user_id=state.user_id, entity_id=key[0], award_key=award_key) + ) + try: + await db.commit() + except IntegrityError: + await db.rollback() + state.awards_claimed.add(key) + return False + + state.awards_claimed.add(key) + return True + + async def _reward_summon(state: SeanceState) -> None: """Workstream C's essence-trickle + item-drop trigger points that exist in this handler today: a small essence trickle for every successful @@ -608,11 +631,18 @@ async def _summon_locked(state: SeanceState) -> None: state.ritual_steps = 0 state.ritual_completed = False state.ritual_success = False - # Only a genuinely fresh summon re-opens the purse for the other two - # rites, exactly as `_reset_passage(new_entity=True)` does below. - state.ritual_paid = False - state.judged_verdicts = set() - _reset_passage(state, new_entity=True) + # Nothing here re-opens the purse any more, and that is the fix. + # + # A genuinely new presence still gets its own full purse — automatically, + # because the purse is now keyed on `entity.id` in `award_claims` and a + # new presence is a new row. What no longer re-opens it is a summon that + # lands back on the SAME spirit, and that distinction is the entire hole: + # `RETURN_CHANCE` means re-calling a known channel usually returns the + # familiar presence, so the old unconditional reset here let a client + # re-summon its way to an unlimited number of payouts for one spirit, + # bounded only by `summon_limiter`. Reconnecting did the same thing for + # free. Each spirit is worth its own rite — once. + _reset_passage(state) await state.send_queue.put( {"type": "entity", "entity": _public_entity(state.entity), "is_new": is_new} ) @@ -918,6 +948,11 @@ async def _handle_ritual_step(state: SeanceState, message: dict) -> None: return traits = state.entity.get("traits", {}) + # Pinned before any await, for the same reason as in the judgment and + # Passage handlers below: the HTTP device-telemetry path drives the same + # SeanceState and can replace `state.entity` mid-handler, and the claim + # must land against the spirit whose rite this actually was. + entity_id = state.entity["id"] success = judgment.roll_ritual_success(traits) state.ritual_completed = True state.ritual_success = success @@ -925,11 +960,12 @@ async def _handle_ritual_step(state: SeanceState, message: dict) -> None: await state.send_queue.put( {"type": "ritual_complete", "success": success, "revealed": revealed} ) - # The milestone pays once per presence. A re-attempt after a rewind - # (`ritual_start`) still rolls and still reveals on success — it just - # doesn't mint a second payout. See `SeanceState.ritual_paid`. - if success and not state.ritual_paid: - state.ritual_paid = True + # The milestone pays once per presence, for good — across rewinds + # (`ritual_start`, the UI's "attempt again"), across reconnects, and + # across two sockets racing each other. A re-attempt still rolls and + # still reveals on success; it just doesn't mint a second payout. See + # `_claim_award`. + if success and await _claim_award(state, entity_id, "ritual"): await _reward_ritual_success(state) @@ -971,14 +1007,28 @@ async def _handle_judgment(state: SeanceState, message: dict) -> None: ritual_success=state.ritual_success, ) - # A verdict lands on a presence once. Replaying the same one — the UI - # re-arms the panel after any result — re-reports the same reading for - # free rather than paying it out again. See `judged_verdicts`; `test` - # never touches the ledger so it is deliberately exempt and stays - # freely repeatable. - already_judged = verdict != "test" and verdict in state.judged_verdicts - if verdict != "test": - state.judged_verdicts.add(verdict) + # A verdict lands on a presence once, durably. Replaying the same one — + # the UI re-arms the panel after any result, and a scripted client can + # simply reconnect or open a second socket — re-reports the same reading + # for free rather than paying it out again. + # + # Claimed PER VERDICT rather than as a single latch, so an honest + # correction (a wrong `trust`, then the right `banish` on the same + # spirit) still resolves normally; what can never repeat is the SAME + # verdict on the SAME presence. `test` never touches the ledger, so it is + # deliberately exempt and stays freely repeatable. + if verdict == "test": + already_judged = False + elif outcome.consequence == "crossed_over": + # One spirit, one crossing, whichever road got there first: the + # Passage's `release` beat claims this same key. Without the shared + # key, a seeker on two sockets could cross the one spirit down both + # roads and be paid for it twice. + already_judged = not await _claim_award(state, entity_id, "crossing") + else: + already_judged = not await _claim_award( + state, entity_id, f"judgment:{verdict}" + ) # What will ACTUALLY reach the ledger. The frame below reports these, # not `outcome`'s face values: the client sums `essence_delta`/ @@ -1060,15 +1110,13 @@ async def _handle_judgment(state: SeanceState, message: dict) -> None: # uses), the `at_peace` write, and the frames. -def _reset_passage(state: SeanceState, *, new_entity: bool = False) -> None: +def _reset_passage(state: SeanceState) -> None: + """Rewind where the rite STANDS. It no longer touches what has been paid — + that lives in `award_claims` now and is keyed on the entity, so there is + nothing here a `passage_start` rewind could reopen.""" state.passage_layer = passage.FIRST_LAYER state.passage_lied = False state.passage_crossed = False - if new_entity: - # Only a genuinely new presence re-opens the purse. A `passage_start` - # rewind must not, or the rite becomes an essence faucet (see - # `passage_paid` on SeanceState). - state.passage_paid = set() def _passage_frame( @@ -1170,12 +1218,23 @@ async def _handle_passage_layer(state: SeanceState) -> None: if outcome.at_peace: state.passage_crossed = True - # A layer pays once per presence. Replaying it — via `passage_start`, via - # a collapse rewind, or via a hand-rolled client spamming the frame — - # reveals the same thing again for free rather than minting essence again. - award = outcome.essence if layer not in state.passage_paid else 0 - if outcome.result in ("opened", "crossed"): - state.passage_paid.add(layer) + # A layer pays once per presence, durably. Replaying it — via + # `passage_start`, via a collapse rewind, via a reconnect, via a second + # socket, or via a hand-rolled client spamming the frame — reveals the + # same thing again for free rather than minting essence again. + award = 0 + if outcome.result in ("opened", "crossed") and await _claim_award( + state, entity_id, f"passage:{layer}" + ): + award = outcome.essence + + # The crossing itself (favor, `at_peace`, the item roll) is claimed + # separately and under a key the `cross_over` VERDICT claims too, so one + # spirit is paid for crossing exactly once no matter which road reached + # it — including two roads walked concurrently on two sockets. + crossing_paid = bool( + outcome.at_peace and await _claim_award(state, entity_id, "crossing") + ) item = None if award or outcome.at_peace: @@ -1188,7 +1247,7 @@ async def _handle_passage_layer(state: SeanceState) -> None: if user is not None: if award: credit_essence(user, award) - if outcome.at_peace: + if crossing_paid: # Completing the Passage is the most compassionate act # in the game, exactly as a correct cross_over verdict # is — so it moves favor by the same amount. @@ -1196,9 +1255,14 @@ async def _handle_passage_layer(state: SeanceState) -> None: user.favor + judgment.FAVOR_CORRECT_CROSS_OVER ) if outcome.at_peace: + # Not gated on `crossing_paid`: the spirit really did cross, + # so the flag is set either way. It is only the PAYOUT that + # happens once. (When the claim was lost, the row is already + # at_peace and this write is a no-op.) entity_row = await db.get(Entity, uuid.UUID(entity_id)) if entity_row is not None: entity_row.at_peace = True + if crossing_paid: item = roll_item_drop("judgment") if item is not None: db.add( diff --git a/backend/tests/test_ws_passage.py b/backend/tests/test_ws_passage.py index 63a1ded..6410252 100644 --- a/backend/tests/test_ws_passage.py +++ b/backend/tests/test_ws_passage.py @@ -99,6 +99,32 @@ def _no_twists(monkeypatch): monkeypatch.setattr(app.ws, "veil_float", selective) +def _force_new_presence(monkeypatch): + """Pin the summon draw so the NEXT summon mints a genuinely new spirit. + + `_summon` draws against RETURN_CHANCE to decide whether the presence + already on this channel answers again, and re-calling the same channel + usually returns the SAME entity row — measured on this suite, 8 of 11 + consecutive re-summons came back with `is_new: false` and an identical id. + + That matters now that the purse is keyed on the entity (award_claims), + because "summon again" and "the same spirit answers again" are then two + different things. This test is about what a genuinely FRESH presence is + worth, so the draw is pinned rather than left to a coin flip; a draw of + 1.0 is >= any possible `return_chance`, so the familiar presence never + answers. Layered over `_no_twists` (call it after), leaving every other + draw real. + """ + real = app.ws.veil_float + + def selective(entropy, context, *args, **kwargs): + if context == "answers": + return 1.0 + return real(entropy, context, *args, **kwargs) + + monkeypatch.setattr(app.ws, "veil_float", selective) + + # The exploit never reaches `release`. Crossing latches `passage_crossed`, # which correctly blocks replay — the faucet ran on the four beats BEFORE # release, rewound by the "walk it again" button after a resisted release. @@ -260,17 +286,20 @@ async def test_a_genuinely_new_presence_reopens_the_purse( ): """The guard must not overshoot: each spirit is worth its own rite. - A `passage_paid` set that survived a summon would silently make every - spirit after the first worthless, which is a worse bug than the faucet. + A durable claim that survived a summon would silently make every spirit + after the first worthless, which is a worse bug than the faucet. """ _no_twists(monkeypatch) + # The second summon must actually bring a DIFFERENT spirit — see + # `_force_new_presence`. Without this the test asserts a coin flip. + _force_new_presence(monkeypatch) token = _login(sync_client, "second-spirit") user_id = _user_id(sync_client, token) with _ws_connect(sync_client, token) as ws: _read_until(ws, "session") - _summon(ws) + first_entity = _summon(ws) baseline = await _essence(db_session, user_id) ws.send_json({"type": "passage_start"}) @@ -278,7 +307,11 @@ async def test_a_genuinely_new_presence_reopens_the_purse( _walk(ws, BEATS_BEFORE_RELEASE) after_first = await _essence(db_session, user_id) - _summon(ws) + second_entity = _summon(ws) + assert second_entity["entity"]["id"] != first_entity["entity"]["id"], ( + "the second summon returned the same spirit — this test is about " + "a genuinely new presence" + ) ws.send_json({"type": "passage_start"}) _settle(ws) _walk(ws, BEATS_BEFORE_RELEASE) diff --git a/backend/tests/test_ws_reward_durability.py b/backend/tests/test_ws_reward_durability.py new file mode 100644 index 0000000..1f748c1 --- /dev/null +++ b/backend/tests/test_ws_reward_durability.py @@ -0,0 +1,682 @@ +"""WS integration tests for the DURABILITY of the reward guards in app.ws. + +`test_ws_passage.py` and `test_ws_reward_replay.py` closed three replay +faucets — the ritual milestone, a verdict, and a Passage beat — but they +closed them with sets held on the in-memory `SeanceState`, and every one of +those tests replays the exploit down a SINGLE websocket. That is exactly the +shape of the remaining hole: the guards were per-connection, so a client that +dropped the socket and reconnected, or simply opened a second one alongside +the first, got a fresh empty guard and could be paid all over again for the +same spirit. Income stayed rate-bounded by `summon_limiter` (per user id, +across connections) and unbounded in total. + +These tests reach past one connection: + + * walk a rite to completion, DISCONNECT, reconnect on a new socket with the + same account, land back on the same spirit, re-walk it — the ledger must + not grow; + * two sockets open SIMULTANEOUSLY for the same account, both claiming the + same award — paid exactly once; + * the same race driven concurrently rather than interleaved, straight at + `_claim_award`, which is where the UNIQUE constraint on award_claims + actually decides the winner; + * and the anti-overshoot direction in both places: a genuinely fresh + presence still pays in full across a reconnect, and a corrected verdict + still pays. + +LANDING ON THE SAME SPIRIT ACROSS CONNECTIONS is the load-bearing detail. A +spirit is matched by the *channel* signature, and with no anomalies at all +that signature falls back to the CONTACT SESSION id — which is per-connection, +so a bare reconnect+summon mints a brand new spirit and would prove nothing. +So these tests feed the same anomaly stream a browser feeds (which is also +what auto-summons), giving both connections the same channel, and pin the +`answers` draw so the familiar presence reliably answers rather than 72% of +the time. +""" + +import asyncio +import uuid + +import pytest +from sqlalchemy import select + +import app.judgment as judgment_module +import app.ws +from app.entities import MIN_ANOMALIES_FOR_SIGNATURE, fallback_profile +from app.inventory import ( + CORRECT_JUDGMENT_ESSENCE, + RITUAL_SUCCESS_ESSENCE, + SUMMON_ESSENCE_TRICKLE, +) +from app.models.award_claim import AwardClaim +from app.models.inventory_item import InventoryItem +from app.models.user import User +from app.rate_limit import RateLimiter + + +class FakeSpiritService: + async def mint_profile( + self, signature, channel, anomalies, language="en", entropy=None, sky=None + ): + return fallback_profile(signature) + + async def fragment(self, source, anomaly, language="en"): + return "listen" + + async def wire_whisper(self, telemetry, language="en"): + return "the wire hums" + + def chat_stream(self, entity, question, history, language="en"): + async def gen(): + yield "here." + + return gen() + + def ambient_ready(self): + return False + + +async def _fake_synth(text, voice, profile, instability=0.0): + return b"RIFFfake wav bytes" + + +@pytest.fixture(autouse=True) +def _fake_spirits(monkeypatch): + monkeypatch.setattr(app.ws, "spirit_service", FakeSpiritService()) + monkeypatch.setattr(app.ws, "synthesize_spirit_voice", _fake_synth) + # Module-level limiters are shared singletons accumulating real hit counts + # across the whole session (precedent: test_ws_reward_replay.py). These + # tests deliberately summon more than the real 4/60s cap allows, because + # the point is what happens on the SECOND connection — a rate limit there + # would mask the durability question instead of answering it. + for name in ( + "summon_limiter", + "summon_ip_limiter", + "question_limiter", + "question_ip_limiter", + "fragment_limiter", + "fragment_ip_limiter", + "ritual_limiter", + "ritual_ip_limiter", + "judgment_limiter", + "judgment_ip_limiter", + "passage_limiter", + "passage_ip_limiter", + ): + monkeypatch.setattr(app.ws, name, RateLimiter(max_requests=10_000, window_seconds=60)) + + +def _read_until(ws, msg_type, max_frames=120, **match): + for _ in range(max_frames): + frame = ws.receive_json() + if frame.get("type") != msg_type: + continue + if all(frame.get(key) == value for key, value in match.items()): + return frame + raise AssertionError(f"never saw frame of type {msg_type!r} matching {match!r}") + + +def _login(sync_client, username): + sync_client.post("/auth/register", json={"username": username, "password": "spookyspooky"}) + sync_client.post("/auth/login", json={"username": username, "password": "spookyspooky"}) + return sync_client.cookies.get("qm_session") + + +def _user_id(sync_client, token): + return uuid.UUID( + sync_client.get("/auth/me", headers={"cookie": f"qm_session={token}"}).json()["id"] + ) + + +def _ws_connect(sync_client, token): + return sync_client.websocket_connect( + "/ws/session", headers={"cookie": f"qm_session={token}"} + ) + + +def _settle(ws): + """Force a full round trip so any post-frame DB write has landed. The + connection's message loop is sequential, so a pong is a hard guarantee + that the previous handler returned — not a poll-and-hope.""" + ws.send_json({"type": "ping"}) + _read_until(ws, "pong") + + +# One fixed anomaly pattern = one fixed channel signature, shared by every +# connection in a test. `signature_from_anomalies` buckets frequency by digit +# count and sorts magnitudes, so identical frames give an identical channel +# regardless of which contact session sent them. +_CHANNEL = [ + {"type": "anomaly", "source": "emf", "frequency": 121.5, "magnitude": 30.0}, + {"type": "anomaly", "source": "emf", "frequency": 122.5, "magnitude": 31.0}, + {"type": "anomaly", "source": "emf", "frequency": 123.5, "magnitude": 32.0}, +] +assert len(_CHANNEL) >= MIN_ANOMALIES_FOR_SIGNATURE + + +def _familiar_answers(monkeypatch): + """Pin the summon draw so a known channel ALWAYS returns its familiar + presence. + + Untouched, this is a genuine draw against RETURN_CHANCE — roughly 7 times + in 10. These tests need the two connections to be looking at the SAME + spirit for the question ("can it be paid twice?") to mean anything, so a + draw of 0.0 is used, which is below any possible `return_chance`. Every + other draw stays real. + """ + real = app.ws.veil_float + + def selective(entropy, context, *args, **kwargs): + if context == "answers": + return 0.0 + return real(entropy, context, *args, **kwargs) + + monkeypatch.setattr(app.ws, "veil_float", selective) + + +def _open_channel(ws): + """Drive the channel a browser drives: enough anomalies to fingerprint it, + which is itself what triggers the summon (see `_handle_anomaly`).""" + _read_until(ws, "session") + for frame in _CHANNEL: + ws.send_json(frame) + entity = _read_until(ws, "entity") + _settle(ws) + return entity["entity"]["id"] + + +def _run_ritual(ws, steps=4): + """One full attempt: the rewind button, then every step.""" + ws.send_json({"type": "ritual_start"}) + for i in range(1, steps + 1): + ws.send_json({"type": "ritual_step", "step": i}) + result = _read_until(ws, "ritual_complete") + _settle(ws) + return result + + +def _judge(ws, verdict): + ws.send_json({"type": "judgment", "verdict": verdict}) + result = _read_until(ws, "judgment_result") + _settle(ws) + return result + + +async def _ledger(db_session, user_id): + db_session.expire_all() + user = await db_session.get(User, user_id) + items = ( + await db_session.execute( + select(InventoryItem).where(InventoryItem.user_id == user_id) + ) + ).scalars().all() + return user.essence, user.favor, len(items) + + +def _always_wins(monkeypatch): + monkeypatch.setattr( + app.ws.judgment, "roll_ritual_success", lambda traits, rng=None: True + ) + + +def _verdict_outcomes(monkeypatch, table): + monkeypatch.setattr( + app.ws.judgment, "judge_verdict", lambda verdict, traits, **kw: table[verdict] + ) + + +_CORRECT_TRUST = judgment_module.JudgmentOutcome( + True, judgment_module.FAVOR_CORRECT_TRUST, CORRECT_JUDGMENT_ESSENCE, False, "reward" +) +_WRONG_TRUST = judgment_module.JudgmentOutcome( + False, judgment_module.FAVOR_WRONG_TRUST, 0, False, "escalation" +) +_CORRECT_BANISH = judgment_module.JudgmentOutcome( + True, judgment_module.FAVOR_CORRECT_BANISH, CORRECT_JUDGMENT_ESSENCE, False, "reward" +) + + +# --- reconnecting ---------------------------------------------------------- + + +# How many times each reconnect test re-opens the socket. Named because the +# expected summon trickle is derived from it, and a bare 5 in two places that +# must agree is exactly how that arithmetic drifts. +RECONNECTS = 5 + + +@pytest.mark.asyncio +async def test_reconnecting_does_not_reopen_the_ritual_purse( + sync_client, db_session, monkeypatch +): + """The exploit the in-memory guard could not see, executed literally. + + Complete the rite, DROP THE SOCKET, reconnect on a fresh websocket with + the same account, tune back to the same channel so the same spirit + answers, and complete the rite again — five times over. The seeker must + finish with exactly one milestone's worth of essence, because there was + only ever one spirit. + + `ritual_paid` was a bool on `SeanceState`, which dies with the connection, + so before this every reconnect paid RITUAL_SUCCESS_ESSENCE again and + rolled another item. + """ + _always_wins(monkeypatch) + _familiar_answers(monkeypatch) + + token = _login(sync_client, "ritual-reconnector") + user_id = _user_id(sync_client, token) + + with _ws_connect(sync_client, token) as ws: + first_entity = _open_channel(ws) + baseline, _, _ = await _ledger(db_session, user_id) + assert _run_ritual(ws)["success"] is True + after_first, _, items_after_first = await _ledger(db_session, user_id) + + for _ in range(RECONNECTS): + with _ws_connect(sync_client, token) as ws: + entity_id = _open_channel(ws) + assert entity_id == first_entity, ( + "the reconnect landed on a different spirit — this test cannot " + "say anything about paying the same one twice" + ) + assert _run_ritual(ws)["success"] is True, "the replay did not even complete" + + after_reconnects, _, items_after_reconnects = await _ledger(db_session, user_id) + + assert after_first - baseline == RITUAL_SUCCESS_ESSENCE, ( + "the rite paid nothing — the test proves nothing" + ) + # Each reconnect re-opens the channel, and opening a channel is a summon, + # which pays SUMMON_ESSENCE_TRICKLE every time by design (app/inventory.py:42 + # — "every successful summon, any mode"; it is bounded by summon_limiter at + # 4/60s, not by any once-per-presence rule). That trickle is NOT what this + # test is about, so it is expected explicitly rather than folded into the + # comparison: what must not repeat is the 15-essence ritual MILESTONE. + expected_trickle = RECONNECTS * SUMMON_ESSENCE_TRICKLE + assert after_reconnects == after_first + expected_trickle, ( + f"reconnecting minted {after_reconnects - after_first - expected_trickle} extra " + f"essence beyond the {expected_trickle} of expected summon trickle, across " + f"{RECONNECTS} fresh websockets; the guard dies with the connection" + ) + assert items_after_reconnects == items_after_first, ( + f"reconnecting rolled {items_after_reconnects - items_after_first} extra item " + "drops off a single spirit" + ) + + +@pytest.mark.asyncio +async def test_reconnecting_does_not_reopen_the_judgment_purse( + sync_client, db_session, monkeypatch +): + """Same hole, the verdict road. `judged_verdicts` was a set on + `SeanceState`; every reconnect handed the client an empty one, so the same + verdict on the same spirit paid essence AND favor AND an item roll again, + with favor walking toward its +1.0 ceiling (which then biases every future + mint).""" + _verdict_outcomes(monkeypatch, {"trust": _CORRECT_TRUST}) + _familiar_answers(monkeypatch) + + token = _login(sync_client, "judgment-reconnector") + user_id = _user_id(sync_client, token) + + with _ws_connect(sync_client, token) as ws: + first_entity = _open_channel(ws) + baseline_essence, baseline_favor, _ = await _ledger(db_session, user_id) + _judge(ws, "trust") + essence_one, favor_one, items_one = await _ledger(db_session, user_id) + + replays = [] + for _ in range(RECONNECTS): + with _ws_connect(sync_client, token) as ws: + assert _open_channel(ws) == first_entity + replays.append(_judge(ws, "trust")) + + essence_end, favor_end, items_end = await _ledger(db_session, user_id) + + assert essence_one - baseline_essence == CORRECT_JUDGMENT_ESSENCE + assert favor_one - baseline_favor == pytest.approx(judgment_module.FAVOR_CORRECT_TRUST) + # As in the ritual test: re-opening the channel is a summon, and the + # trickle is paid per summon by design. The VERDICT is what must not repeat. + # Favor, checked just below, has no trickle — so it must not move at all. + expected_trickle = RECONNECTS * SUMMON_ESSENCE_TRICKLE + assert essence_end == essence_one + expected_trickle, ( + f"reconnecting minted {essence_end - essence_one - expected_trickle} extra " + f"essence beyond the {expected_trickle} of expected summon trickle, across " + f"{RECONNECTS} fresh websockets; the verdict guard dies with the connection" + ) + assert favor_end == pytest.approx(favor_one), ( + f"reconnecting moved favor by a further {favor_end - favor_one}; a client " + "that reconnects in a loop can pin favor at the +1.0 ceiling" + ) + assert items_end == items_one + for frame in replays: + assert frame["essence_delta"] == 0 and frame["favor_delta"] == 0, ( + "a verdict replayed on a fresh connection advertised essence/favor " + "it did not pay" + ) + + +@pytest.mark.asyncio +async def test_reconnecting_does_not_reopen_the_passage_purse( + sync_client, db_session, monkeypatch +): + """Same hole, the Passage road. `passage_paid` was a set on `SeanceState`, + so every reconnect re-credited the whole ladder of beats.""" + real = app.ws.veil_float + + def selective(entropy, context, *args, **kwargs): + # No collapses, no lies — the totals must depend on the guard, not the + # draw — and the familiar presence always answers. + if context.startswith("passage:"): + return 1.0 + if context == "answers": + return 0.0 + return real(entropy, context, *args, **kwargs) + + monkeypatch.setattr(app.ws, "veil_float", selective) + + # Every beat but `release`: crossing latches `at_peace`, which correctly + # takes the spirit out of reach of any later summon. + beats = len(app.ws.passage.LAYERS) - 1 + + token = _login(sync_client, "passage-reconnector") + user_id = _user_id(sync_client, token) + + def _walk(ws): + for _ in range(beats): + ws.send_json({"type": "passage_layer"}) + _read_until(ws, "passage_result") + _settle(ws) + + with _ws_connect(sync_client, token) as ws: + first_entity = _open_channel(ws) + baseline, _, _ = await _ledger(db_session, user_id) + ws.send_json({"type": "passage_start"}) + _settle(ws) + _walk(ws) + after_first, _, _ = await _ledger(db_session, user_id) + + for _ in range(RECONNECTS): + with _ws_connect(sync_client, token) as ws: + assert _open_channel(ws) == first_entity + ws.send_json({"type": "passage_start"}) + _settle(ws) + _walk(ws) + + after_reconnects, _, _ = await _ledger(db_session, user_id) + + assert after_first - baseline > 0, "the rite paid nothing — the test proves nothing" + # Re-opening the channel summons, and the trickle is per-summon by design; + # the BEATS are what must not pay twice. + expected_trickle = RECONNECTS * SUMMON_ESSENCE_TRICKLE + assert after_reconnects == after_first + expected_trickle, ( + f"reconnecting minted {after_reconnects - after_first - expected_trickle} extra " + f"essence beyond the {expected_trickle} of expected summon trickle, across " + f"{RECONNECTS} fresh websockets; the passage is an unbounded faucet again" + ) + + +# --- two sockets at once --------------------------------------------------- + + +@pytest.mark.asyncio +async def test_two_simultaneous_connections_pay_the_ritual_once( + sync_client, db_session, monkeypatch +): + """Not a reconnect: both sockets are OPEN AT THE SAME TIME. + + This is the cheaper half of the exploit — no disconnect needed, just a + second tab. Each connection has its own `SeanceState` and therefore had + its own empty guard, so the two of them paid the same spirit's ritual + twice while every replay test in the suite watched a single socket. + """ + _always_wins(monkeypatch) + _familiar_answers(monkeypatch) + + token = _login(sync_client, "ritual-two-tabs") + user_id = _user_id(sync_client, token) + + with _ws_connect(sync_client, token) as first, _ws_connect(sync_client, token) as second: + entity_a = _open_channel(first) + entity_b = _open_channel(second) + assert entity_a == entity_b, ( + "the two connections landed on different spirits — this test cannot " + "say anything about paying the same one twice" + ) + baseline, _, items_baseline = await _ledger(db_session, user_id) + + # Interleaved, so neither connection's rite is finished before the + # other's begins. + first.send_json({"type": "ritual_start"}) + second.send_json({"type": "ritual_start"}) + for i in range(1, 5): + first.send_json({"type": "ritual_step", "step": i}) + second.send_json({"type": "ritual_step", "step": i}) + assert _read_until(first, "ritual_complete")["success"] is True + assert _read_until(second, "ritual_complete")["success"] is True + _settle(first) + _settle(second) + + essence_end, _, items_end = await _ledger(db_session, user_id) + + assert essence_end - baseline == RITUAL_SUCCESS_ESSENCE, ( + f"two simultaneous connections paid {essence_end - baseline} for one " + f"spirit's ritual instead of {RITUAL_SUCCESS_ESSENCE}" + ) + assert items_end - items_baseline <= 1, ( + "two simultaneous connections each rolled the ritual's item drop" + ) + + +@pytest.mark.asyncio +async def test_two_simultaneous_connections_pay_a_verdict_once( + sync_client, db_session, monkeypatch +): + _verdict_outcomes(monkeypatch, {"trust": _CORRECT_TRUST}) + _familiar_answers(monkeypatch) + + token = _login(sync_client, "judgment-two-tabs") + user_id = _user_id(sync_client, token) + + with _ws_connect(sync_client, token) as first, _ws_connect(sync_client, token) as second: + assert _open_channel(first) == _open_channel(second) + baseline_essence, baseline_favor, _ = await _ledger(db_session, user_id) + + first.send_json({"type": "judgment", "verdict": "trust"}) + second.send_json({"type": "judgment", "verdict": "trust"}) + frames = [ + _read_until(first, "judgment_result"), + _read_until(second, "judgment_result"), + ] + _settle(first) + _settle(second) + + essence_end, favor_end, _ = await _ledger(db_session, user_id) + + assert essence_end - baseline_essence == CORRECT_JUDGMENT_ESSENCE, ( + f"two simultaneous connections paid {essence_end - baseline_essence} for one " + f"verdict instead of {CORRECT_JUDGMENT_ESSENCE}" + ) + assert favor_end - baseline_favor == pytest.approx( + judgment_module.FAVOR_CORRECT_TRUST + ), "two simultaneous connections moved favor twice for one verdict" + # And the frames must be honest about it: exactly one of them paid. + assert sum(f["essence_delta"] for f in frames) == essence_end - baseline_essence + + +@pytest.mark.asyncio +async def test_a_concurrent_claim_race_resolves_to_one_winner(sync_client, db_session): + """Straight at the constraint, with real concurrency. + + The two-tabs tests above interleave frames but the server still handles + them one at a time, so they prove the guard is DURABLE without ever + forcing the two claims to be in flight together. This one does: twelve + coroutines claim the same (user, entity, award) at once via + `asyncio.gather`. + + Exactly one may win. If `_claim_award` were a check-then-insert, several + would read "unclaimed" and all of them would pay; if it did not catch the + IntegrityError, the losers would raise instead of returning False and + would take down their whole séance with a 500. + """ + token = _login(sync_client, "claim-racer") + user_id = _user_id(sync_client, token) + + with _ws_connect(sync_client, token) as ws: + entity_id = _open_channel(ws) + + state = app.ws.SeanceState( + user_id=user_id, session_id=uuid.uuid4(), client_ip="127.0.0.1" + ) + + async def claim(): + # A private state per racer: sharing one would let the in-memory fast + # path answer some of them, which is exactly what must NOT decide this. + racer = app.ws.SeanceState( + user_id=user_id, session_id=state.session_id, client_ip="127.0.0.1" + ) + return await app.ws._claim_award(racer, entity_id, "ritual") + + results = await asyncio.gather(*(claim() for _ in range(12))) + + assert sum(results) == 1, ( + f"{sum(results)} of 12 concurrent claims won the same award; the " + "UNIQUE constraint is not deciding the race" + ) + db_session.expire_all() + rows = ( + await db_session.execute( + select(AwardClaim).where( + AwardClaim.user_id == user_id, + AwardClaim.entity_id == uuid.UUID(entity_id), + AwardClaim.award_key == "ritual", + ) + ) + ).scalars().all() + assert len(rows) == 1, f"the race left {len(rows)} claim rows for one award" + + +# --- the anti-overshoot direction ------------------------------------------ + + +@pytest.mark.asyncio +async def test_a_fresh_presence_after_a_reconnect_still_pays_in_full( + sync_client, db_session, monkeypatch +): + """The guard must not overshoot across connections either. + + A durable claim that keyed on the user alone — or on anything coarser than + the presence — would silently make every spirit after the first worthless + for the rest of that account's life, which is a far worse bug than the + faucet. Reconnect onto a DIFFERENT channel, meet a different spirit, and + the rite must pay in full again. + """ + _always_wins(monkeypatch) + + token = _login(sync_client, "fresh-after-reconnect") + user_id = _user_id(sync_client, token) + + with _ws_connect(sync_client, token) as ws: + first_entity = _open_channel(ws) + baseline, _, _ = await _ledger(db_session, user_id) + _run_ritual(ws) + after_first, _, _ = await _ledger(db_session, user_id) + + # A different anomaly pattern is a different channel, so a different + # presence answers — the honest version of what the replay tests fake. + with _ws_connect(sync_client, token) as ws: + _read_until(ws, "session") + for i in range(MIN_ANOMALIES_FOR_SIGNATURE): + ws.send_json( + {"type": "anomaly", "source": "radio", "frequency": 9000.0 + i, "magnitude": 77.0 + i} + ) + second_entity = _read_until(ws, "entity")["entity"]["id"] + _settle(ws) + assert second_entity != first_entity, ( + "the second channel produced the same spirit — this test is about a " + "genuinely new presence" + ) + _run_ritual(ws) + after_second, _, _ = await _ledger(db_session, user_id) + + first_rite = after_first - baseline + assert first_rite == RITUAL_SUCCESS_ESSENCE + # The second connection's summon pays its own trickle too, so compare only + # the amount that arrived once the rite completed. + assert after_second - after_first >= RITUAL_SUCCESS_ESSENCE, ( + "a fresh presence met on a new connection did not re-open the purse — " + "every spirit after the first is worth nothing" + ) + + +@pytest.mark.asyncio +async def test_a_correction_still_pays_on_a_second_connection( + sync_client, db_session, monkeypatch +): + """The claim is keyed PER VERDICT, and that must survive the move to the + database. A seeker who calls it wrong on one connection and right on the + next is correcting themselves, not replaying — a blanket "judged once per + spirit" row would make the corrected verdict a no-op.""" + _verdict_outcomes(monkeypatch, {"trust": _WRONG_TRUST, "banish": _CORRECT_BANISH}) + _familiar_answers(monkeypatch) + + token = _login(sync_client, "corrector-reconnect") + user_id = _user_id(sync_client, token) + + with _ws_connect(sync_client, token) as ws: + first_entity = _open_channel(ws) + baseline_essence, baseline_favor, _ = await _ledger(db_session, user_id) + wrong = _judge(ws, "trust") + + with _ws_connect(sync_client, token) as ws: + assert _open_channel(ws) == first_entity + right = _judge(ws, "banish") + + essence_end, favor_end, _ = await _ledger(db_session, user_id) + + assert wrong["consequence"] == "escalation" + assert right["consequence"] == "reward" + # One reconnect here, so exactly one summon trickle rides along with the + # corrected verdict's payout. + assert essence_end - baseline_essence == CORRECT_JUDGMENT_ESSENCE + SUMMON_ESSENCE_TRICKLE, ( + "the corrected verdict paid nothing on the second connection — the " + "durable guard is too broad" + ) + assert favor_end - baseline_favor == pytest.approx( + judgment_module.FAVOR_WRONG_TRUST + judgment_module.FAVOR_CORRECT_BANISH + ) + + +@pytest.mark.asyncio +async def test_the_test_verdict_stays_freely_repeatable_across_connections( + sync_client, db_session, monkeypatch +): + """`test` is a diagnostic pulse that never touches the ledger, so it is + deliberately exempt — it must never write a claim row, or the panel would + stop answering a seeker who pulses the same spirit from a second tab.""" + _familiar_answers(monkeypatch) + + token = _login(sync_client, "tester-reconnect") + user_id = _user_id(sync_client, token) + + results = [] + for _ in range(3): + with _ws_connect(sync_client, token) as ws: + entity_id = _open_channel(ws) + baseline, baseline_favor, _ = await _ledger(db_session, user_id) + results.append(_judge(ws, "test")) + essence_end, favor_end, _ = await _ledger(db_session, user_id) + assert essence_end == baseline and favor_end == baseline_favor + + assert all(r["consequence"] == "neutral" for r in results) + db_session.expire_all() + claims = ( + await db_session.execute( + select(AwardClaim).where( + AwardClaim.user_id == user_id, + AwardClaim.entity_id == uuid.UUID(entity_id), + ) + ) + ).scalars().all() + assert claims == [], "the `test` verdict wrote a claim row and is no longer free" diff --git a/backend/tests/test_ws_reward_replay.py b/backend/tests/test_ws_reward_replay.py index c0699f0..4b2f9cd 100644 --- a/backend/tests/test_ws_reward_replay.py +++ b/backend/tests/test_ws_reward_replay.py @@ -192,6 +192,31 @@ def _verdict_outcomes(monkeypatch, table): ) +def _force_new_presence(monkeypatch): + """Pin the summon draw so the NEXT summon mints a genuinely new spirit. + + `_summon` draws against RETURN_CHANCE to decide whether the presence + already on this channel answers again, and re-calling the same channel + usually returns the SAME entity row — measured on this suite, 8 of 11 + consecutive re-summons came back with `is_new: false` and an identical id. + + That matters now that the purse is keyed on the entity (award_claims), + because "summon again" and "the same spirit answers again" are then two + different things. These tests are about what a genuinely FRESH presence is + worth, so the draw is pinned rather than left to a coin flip; a draw of + 1.0 is >= any possible `return_chance`, so the familiar presence never + answers. Every other draw stays real. + """ + real = app.ws.veil_float + + def selective(entropy, context, *args, **kwargs): + if context == "answers": + return 1.0 + return real(entropy, context, *args, **kwargs) + + monkeypatch.setattr(app.ws, "veil_float", selective) + + # --- the ritual faucet ----------------------------------------------------- @@ -272,22 +297,29 @@ async def test_a_genuinely_new_presence_reopens_the_ritual_purse( ): """The guard must not overshoot: each spirit is worth its own rite. - A `ritual_paid` latch that survived a summon would silently make every - spirit after the first worthless. + A durable claim that survived a summon would silently make every spirit + after the first worthless. """ _always_wins(monkeypatch) + # The second summon must actually bring a DIFFERENT spirit — see + # `_force_new_presence`. Without this the test asserts a coin flip. + _force_new_presence(monkeypatch) token = _login(sync_client, "ritual-second-spirit") user_id = _user_id(sync_client, token) with _ws_connect(sync_client, token) as ws: _read_until(ws, "session") - _summon(ws) + first_entity = _summon(ws) baseline, _, _ = await _ledger(db_session, user_id) _run_ritual(ws) after_first, _, _ = await _ledger(db_session, user_id) - _summon(ws) + second_entity = _summon(ws) + assert second_entity["entity"]["id"] != first_entity["entity"]["id"], ( + "the second summon returned the same spirit — this test is about " + "a genuinely new presence" + ) _run_ritual(ws) after_second, _, _ = await _ledger(db_session, user_id) @@ -425,17 +457,24 @@ async def test_a_genuinely_new_presence_reopens_the_judgment_purse( sync_client, db_session, monkeypatch ): _verdict_outcomes(monkeypatch, {"trust": _CORRECT_TRUST}) + # The second summon must actually bring a DIFFERENT spirit — see + # `_force_new_presence`. Without this the test asserts a coin flip. + _force_new_presence(monkeypatch) token = _login(sync_client, "judgment-second-spirit") user_id = _user_id(sync_client, token) with _ws_connect(sync_client, token) as ws: _read_until(ws, "session") - _summon(ws) + first_entity = _summon(ws) _judge(ws, "trust") after_first, _, _ = await _ledger(db_session, user_id) - _summon(ws) + second_entity = _summon(ws) + assert second_entity["entity"]["id"] != first_entity["entity"]["id"], ( + "the second summon returned the same spirit — this test is about " + "a genuinely new presence" + ) _judge(ws, "trust") after_second, _, _ = await _ledger(db_session, user_id)