diff --git a/backend/app/rate_limit.py b/backend/app/rate_limit.py index d6dc4c6..918699d 100644 --- a/backend/app/rate_limit.py +++ b/backend/app/rate_limit.py @@ -1,5 +1,24 @@ import time from collections import defaultdict +from typing import Mapping + + +def resolve_client_ip(headers: Mapping[str, str], direct_host: str | None) -> str: + """Resolves the real visitor IP for per-IP rate limiting. + + The App CT sits behind a Cloudflare Tunnel that runs on a separate + machine (see README Architecture) — every internet-facing connection's + raw TCP peer is that tunnel machine, not the visitor, which would + collapse per-IP limiting to a single shared bucket for all remote + traffic. Cloudflare's edge sets `CF-Connecting-IP` itself, stripping any + client-supplied value first, so it's safe to trust here. Direct + LAN/local access (no Cloudflare in front, e.g. local dev) has no such + header and falls back to the raw socket peer. + """ + forwarded = headers.get("cf-connecting-ip") + if forwarded: + return forwarded + return direct_host or "unknown" class RateLimiter: diff --git a/backend/app/ws.py b/backend/app/ws.py index 6c32990..1ea4e36 100644 --- a/backend/app/ws.py +++ b/backend/app/ws.py @@ -36,7 +36,7 @@ from app.models.contact_session import ContactSession from app.models.entity import Entity from app.models.entity_sighting import EntitySighting from app.models.event import Event -from app.rate_limit import RateLimiter +from app.rate_limit import RateLimiter, resolve_client_ip from app.telemetry import detect_wire_spike, sample_network from app.tts.piper import synthesize_spirit_voice from app.tts.voices import pick_voice @@ -53,6 +53,16 @@ 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) +# 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, +# nor do they protect the resource from many accounts sharing one IP. IP +# thresholds are looser than the per-account ones since a single IP can +# legitimately host multiple accounts on a shared network. +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) + AUDIO_DIR = Path(settings.data_dir) / "audio" @@ -65,6 +75,7 @@ def _audio_dir() -> Path: class SeanceState: user_id: uuid.UUID session_id: uuid.UUID + client_ip: str send_queue: asyncio.Queue = field(default_factory=asyncio.Queue) mode: str = "unknown" language: str = "en" @@ -91,6 +102,11 @@ def serialize_entity(entity: Entity) -> dict: } +def _client_ip(websocket: WebSocket) -> str: + host = websocket.client.host if websocket.client else None + return resolve_client_ip(websocket.headers, host) + + async def _authenticate(websocket: WebSocket) -> uuid.UUID | None: raw_token = websocket.cookies.get(SESSION_COOKIE_NAME) if raw_token is None: @@ -227,7 +243,12 @@ async def _summon(state: SeanceState, channel: str) -> tuple[Entity, bool]: async def _handle_summon(state: SeanceState) -> None: - if not summon_limiter.allow(str(state.user_id)): + # Short-circuits: an account already over its own cap never gets far + # enough to spend from the IP budget too. + if not ( + summon_limiter.allow(str(state.user_id)) + and summon_ip_limiter.allow(state.client_ip) + ): await state.send_queue.put( { "type": "error", @@ -266,7 +287,10 @@ async def _handle_anomaly(state: SeanceState, message: dict) -> None: await state.send_queue.put({"type": "status", "state": "attuning"}) return - if not fragment_limiter.allow(str(state.user_id)): + if not ( + fragment_limiter.allow(str(state.user_id)) + and fragment_ip_limiter.allow(state.client_ip) + ): return # anomalies during a crowded veil just pass unheard try: @@ -279,7 +303,10 @@ async def _handle_anomaly(state: SeanceState, message: dict) -> None: async def _handle_question(state: SeanceState, text: str) -> None: - if not question_limiter.allow(str(state.user_id)): + if not ( + question_limiter.allow(str(state.user_id)) + and question_ip_limiter.allow(state.client_ip) + ): await state.send_queue.put( { "type": "error", @@ -390,7 +417,11 @@ async def session_socket(websocket: WebSocket) -> None: await db.commit() await db.refresh(contact_session) - state = SeanceState(user_id=user_id, session_id=contact_session.id) + state = SeanceState( + user_id=user_id, + session_id=contact_session.id, + client_ip=_client_ip(websocket), + ) sender = asyncio.create_task(_sender(state, websocket)) await state.send_queue.put({"type": "session", "id": str(contact_session.id)}) diff --git a/backend/tests/test_rate_limit.py b/backend/tests/test_rate_limit.py index dc331e0..83122fa 100644 --- a/backend/tests/test_rate_limit.py +++ b/backend/tests/test_rate_limit.py @@ -1,6 +1,6 @@ from unittest.mock import patch -from app.rate_limit import RateLimiter +from app.rate_limit import RateLimiter, resolve_client_ip def test_allows_up_to_limit_then_blocks(): @@ -29,3 +29,20 @@ def test_hits_expire_after_window_elapses(): with patch("app.rate_limit.time.monotonic", return_value=111.0): # 11 seconds later, both prior hits (at t=100) are older than window_start (111 - 10 = 101) assert limiter.allow("user-1") is True + + +def test_resolve_client_ip_prefers_cf_connecting_ip_over_socket_peer(): + # The App CT sits behind a Cloudflare Tunnel on a separate machine — the + # raw socket peer is always the tunnel, never the visitor, for every + # internet-facing request. + headers = {"cf-connecting-ip": "203.0.113.7"} + assert resolve_client_ip(headers, "10.30.20.1") == "203.0.113.7" + + +def test_resolve_client_ip_falls_back_to_socket_peer_without_header(): + # Direct LAN/local access (no Cloudflare in front) has no such header. + assert resolve_client_ip({}, "10.30.20.1") == "10.30.20.1" + + +def test_resolve_client_ip_falls_back_to_unknown_with_no_peer_or_header(): + assert resolve_client_ip({}, None) == "unknown" diff --git a/backend/tests/test_ws_session.py b/backend/tests/test_ws_session.py index edc3196..b06a9cc 100644 --- a/backend/tests/test_ws_session.py +++ b/backend/tests/test_ws_session.py @@ -7,6 +7,7 @@ import app.ws from app.entities import fallback_profile from app.models.contact_session import ContactSession from app.models.event import Event +from app.rate_limit import RateLimiter class FakeSpiritService: @@ -175,3 +176,92 @@ async def test_same_signature_recontacts_same_entity(sync_client): assert second_frame["entity"]["name"] == first assert second_frame["is_new"] is False assert second_frame["entity"]["contact_count"] == 2 + + +@pytest.mark.asyncio +async def test_summon_rate_limited_per_account(sync_client, monkeypatch): + # Swap in a tight, test-scoped limiter so this doesn't depend on (or + # pollute) the shared module-level budget other tests draw from. + monkeypatch.setattr( + app.ws, "summon_limiter", RateLimiter(max_requests=2, window_seconds=60) + ) + _login(sync_client, "account-limited") + + with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws: + _read_until(ws, "session") + for _ in range(2): + ws.send_json({"type": "summon"}) + _read_until(ws, "entity") + _read_until(ws, "utterance", kind="greeting") + + ws.send_json({"type": "summon"}) + rejection = _read_until(ws, "error") + assert rejection["code"] == "rate_limited" + assert rejection["message"] == ( + "The veil is crowded. The spirits need a moment before another summoning." + ) + + +@pytest.mark.asyncio +async def test_summon_rate_limited_per_ip_even_with_fresh_account( + sync_client, monkeypatch +): + # Starve only the IP bucket; the per-account limiter stays at its + # production default so each account below has plenty of its own budget + # left. This proves the IP limiter alone can reject a request — the + # gap the per-account-only limiters left open. + monkeypatch.setattr( + app.ws, "summon_ip_limiter", RateLimiter(max_requests=1, window_seconds=60) + ) + + _login(sync_client, "ip-limited-a") + with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws: + _read_until(ws, "session") + ws.send_json({"type": "summon"}) + _read_until(ws, "entity") # spends the single per-IP slot + + # A different account — its own per-account budget is untouched — but + # every connection in this test shares the same (simulated) source IP, + # which is already spent. + _login(sync_client, "ip-limited-b") + with _ws_connect(sync_client, sync_client.cookies.get("qm_session")) as ws: + _read_until(ws, "session") + ws.send_json({"type": "summon"}) + rejection = _read_until(ws, "error") + assert rejection["code"] == "rate_limited" + assert rejection["message"] == ( + "The veil is crowded. The spirits need a moment before another summoning." + ) + + +@pytest.mark.asyncio +async def test_summon_per_ip_bucket_follows_cf_connecting_ip_not_socket_peer( + sync_client, monkeypatch +): + # Every connection in this test suite shares the same simulated socket + # peer (TestClient has no real network). Without preferring + # CF-Connecting-IP, distinct visitors behind the Cloudflare Tunnel would + # collapse into one shared per-IP bucket — this proves they don't. + monkeypatch.setattr( + app.ws, "summon_ip_limiter", RateLimiter(max_requests=1, window_seconds=60) + ) + + token_a = _login(sync_client, "cf-ip-a") + with sync_client.websocket_connect( + "/ws/session", + headers={"cookie": f"qm_session={token_a}", "cf-connecting-ip": "203.0.113.1"}, + ) as ws: + _read_until(ws, "session") + ws.send_json({"type": "summon"}) + _read_until(ws, "entity") # spends visitor A's per-IP slot only + + token_b = _login(sync_client, "cf-ip-b") + with sync_client.websocket_connect( + "/ws/session", + headers={"cookie": f"qm_session={token_b}", "cf-connecting-ip": "203.0.113.2"}, + ) as ws: + _read_until(ws, "session") + ws.send_json({"type": "summon"}) + # A different visitor IP behind the same tunnel — must not be + # rejected by visitor A's already-spent bucket. + _read_until(ws, "entity")