From 702b37b319cbe79764a7456130e0c38c4900fa6f Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Mon, 27 Jul 2026 14:44:15 +0200 Subject: [PATCH] Cap SSE subscribers per client IP instead of only globally (B-38) GET /rounds/stream capped concurrent subscribers with one global counter (MAX_SUBSCRIBERS=500): anyone opening 500 connections degraded every other user to polling. The comment called it a defensive cap; it was actually the vector, since nothing stopped a single source from exhausting it alone. RoundEventBroadcaster now also tracks subscribers per client IP, capped much lower (MAX_SUBSCRIBERS_PER_IP=5). Past that cap, opening one more stream evicts that same IP's own oldest connection (woken via a new EVICTED sentinel so the SSE generator closes it promptly) rather than refusing the new one or letting one abusive IP crowd out unrelated clients under the old global-only cap. The global cap stays as a backstop against overall resource exhaustion regardless of source. Extracted client_ip() (X-Forwarded-For, since Caddy reverse-proxies every request) out of auth/routes.py into app/api/client_ip.py so the login/registration throttles (B-33) and this new per-IP cap share one definition instead of two that could drift apart. Not implemented: enforcing the connection cap at the Caddy layer itself, which the proposed fix also suggested - the standard Caddy image this project uses has no such directive without a third-party module, and building a custom image felt like a bigger, separate change than this fix warranted. Suite grows from 201 to 211 tests. BUGS.md moves B-38 to Previously fixed. --- BUGS.md | 23 +++-------- app/api/client_ip.py | 17 ++++++++ app/api/routes/rounds.py | 14 +++++-- app/auth/routes.py | 11 +----- app/rounds/events.py | 68 +++++++++++++++++++++++++------- tests/unit/test_client_ip.py | 43 ++++++++++++++++++++ tests/unit/test_round_events.py | 70 ++++++++++++++++++++++++++++++++- 7 files changed, 201 insertions(+), 45 deletions(-) create mode 100644 app/api/client_ip.py create mode 100644 tests/unit/test_client_ip.py diff --git a/BUGS.md b/BUGS.md index e90ff3a..ee2101d 100644 --- a/BUGS.md +++ b/BUGS.md @@ -1,11 +1,11 @@ # Known bugs A second full-codebase audit on 2026-07-27 found **25 further issues** (4 critical, 6 high, -7 medium, 8 low), listed below as B-33 … B-49. B-25 through B-37 are fixed (see "Previously -fixed" below) — no Critical-severity finding remains open; the other 12 are High/Medium/Low. +7 medium, 8 low), listed below as B-39 … B-49. B-25 through B-38 are fixed (see "Previously +fixed" below) — no Critical-severity finding remains open; the other 11 are Medium/Low. The 139-test suite was green at the time of the audit, so none of these were caught by existing -coverage — every fix lands with a regression test (the thirteen fixes so far brought the suite -from 139 to 201). +coverage — every fix lands with a regression test (the fourteen fixes so far brought the suite +from 139 to 211). The recurring pattern across the open findings is worth stating once: the code is rigorous about the failure modes that have actually been hit, and silent about the ones that have not. @@ -18,18 +18,6 @@ admin auth, single-process assumptions, no user-facing history, etc.) are docume ## Medium -### B-38 — The 500-subscriber SSE cap is a zero-cost DoS of the realtime feature - -`GET /rounds/stream` requires no authentication and each connection takes a slot on a -**global** counter (`rounds/events.py:33-38`). Anyone opening 500 connections degrades every -real user to polling. The comment describes it as a defensive cap; it is in fact the vector, -not the defence. - -**Proposed fix.** Cap per client IP (and, once available, per authenticated user) rather than -globally, and evict the oldest idle subscriber instead of refusing new ones. The reverse proxy -is the right place for the connection-count limit — Caddy can enforce it before the request -reaches the app. - ### B-39 — SQLite with no WAL, no `busy_timeout`, and five concurrent writer tasks `db/base.py:6` calls `create_async_engine(settings.database_url)` with no `connect_args`, and @@ -156,9 +144,10 @@ already does. - **B-35** — API timestamps round-tripped as naive datetimes, so the frontend parsed them as local time instead of UTC - **B-36** — a stalled draw wait had no timeout, no log, and no audit trail, so a frozen round showed nothing in `/admin` - **B-37** — a withdrawal covered by unconfirmed change answered "insufficient balance" instead of distinguishing it from actually having no funds +- **B-38** — the SSE subscriber cap was global, so one client opening enough connections degraded every other user to polling See git history for the fix-by-fix breakdown (commits `f13f685`, `50a43ae`, `933760e`, and the -B-28/B-29/B-30/B-31/B-32/B-33/B-34/B-35/B-36/B-37 fixes). Suite grew from 139 to 201 tests over the thirteen. +B-28/B-29/B-30/B-31/B-32/B-33/B-34/B-35/B-36/B-37/B-38 fixes). Suite grew from 139 to 211 tests over the fourteen. A full-codebase audit on 2026-07-26 (commit `d4e0974`) found 24 bugs across every Python module under `app/`, both static frontends, and the Docker/Caddy deployment — 5 critical, diff --git a/app/api/client_ip.py b/app/api/client_ip.py new file mode 100644 index 0000000..0d64756 --- /dev/null +++ b/app/api/client_ip.py @@ -0,0 +1,17 @@ +from fastapi import Request + + +def client_ip(request: Request) -> str: + """The caller's real IP, from Caddy's X-Forwarded-For (see Caddyfile) — + request.client.host would otherwise be the reverse proxy's own address, not + the caller's. Falls back to request.client.host only if the header is + somehow missing (e.g. the app container hit directly, bypassing Caddy). + + Shared by the login/registration throttles (B-33) and the SSE per-IP + subscriber cap (B-38) so the two can't drift into different notions of + "the client's IP". + """ + forwarded = request.headers.get("x-forwarded-for") + if forwarded: + return forwarded.split(",")[0].strip() + return request.client.host if request.client else "unknown" diff --git a/app/api/routes/rounds.py b/app/api/routes/rounds.py index eef8727..69caf80 100644 --- a/app/api/routes/rounds.py +++ b/app/api/routes/rounds.py @@ -8,12 +8,13 @@ from pydantic import BaseModel from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession +from app.api.client_ip import client_ip from app.api.timeutil import isoformat_utc from app.auth.dependencies import get_optional_user from app.db.models import RoundParticipant, User from app.db.session import get_session from app.rounds.config import get_round_config -from app.rounds.events import RoundEventCapacityError, broadcaster +from app.rounds.events import EVICTED, RoundEventCapacityError, broadcaster from app.rounds.service import get_active_round router = APIRouter(prefix="/rounds", tags=["rounds"]) @@ -46,9 +47,14 @@ async def round_stream(request: Request) -> Response: That's also what happens past MAX_SUBSCRIBERS (app/rounds/events.py): this returns 503 rather than opening a stream, and the browser's EventSource just retries later while the frontend keeps working off polling meanwhile. + + Concurrent streams are additionally capped per client IP (B-38): past + MAX_SUBSCRIBERS_PER_IP, opening one more evicts that IP's own oldest + connection rather than refusing the new one or letting a single source + exhaust the global cap and degrade every other user. """ try: - queue = broadcaster.subscribe() + queue = broadcaster.subscribe(client_ip(request)) except RoundEventCapacityError: return JSONResponse(status_code=503, content={"detail": "too many concurrent update streams"}) @@ -59,7 +65,9 @@ async def round_stream(request: Request) -> Response: if await request.is_disconnected(): break try: - await asyncio.wait_for(queue.get(), timeout=_SSE_DISCONNECT_CHECK_SECONDS) + item = await asyncio.wait_for(queue.get(), timeout=_SSE_DISCONNECT_CHECK_SECONDS) + if item is EVICTED: + break # this IP opened another stream past its per-IP cap yield "event: update\ndata: {}\n\n".format(json.dumps({})) ticks_since_keepalive = 0 except asyncio.TimeoutError: diff --git a/app/auth/routes.py b/app/auth/routes.py index 8564e29..4fd2cd9 100644 --- a/app/auth/routes.py +++ b/app/auth/routes.py @@ -4,6 +4,7 @@ from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from app.api.client_ip import client_ip as _client_ip from app.api.errors import http_error from app.auth.rate_limit import AuthRateLimiters from app.auth.security import MIN_PASSWORD_LENGTH, create_access_token, hash_password, verify_password @@ -34,16 +35,6 @@ def _rate_limiters(request: Request) -> AuthRateLimiters: return request.app.state.auth_rate_limiters -def _client_ip(request: Request) -> str: - # Caddy (see Caddyfile) reverse-proxies every request, so request.client.host - # is the proxy's address, not the caller's — fall back to it only if the - # header is somehow missing (e.g. hitting the app container directly). - forwarded = request.headers.get("x-forwarded-for") - if forwarded: - return forwarded.split(",")[0].strip() - return request.client.host if request.client else "unknown" - - def _rate_limited_error(retry_after: float): return http_error( status.HTTP_429_TOO_MANY_REQUESTS, diff --git a/app/rounds/events.py b/app/rounds/events.py index d648e4a..c60cbab 100644 --- a/app/rounds/events.py +++ b/app/rounds/events.py @@ -1,16 +1,31 @@ import asyncio +from collections import defaultdict -# Defensive cap on concurrent SSE subscribers. Expected load is on the order of -# ~100 concurrent users; this is set well above that so it never engages under -# normal use — it exists purely so a runaway/DoS-y number of open connections -# degrades (new connections fall back to polling, see round_stream()) instead -# of growing the in-memory subscriber set without bound. Revisit this number if -# expected concurrency grows well past it. +# Defensive backstop on concurrent SSE subscribers overall, regardless of source +# — expected load is on the order of ~100 concurrent users, so this is set well +# above that. The real defense against a single abusive source is the per-IP cap +# below (B-38): a global-only cap was trivially exhausted by one client opening +# MAX_SUBSCRIBERS connections, degrading every other user to polling — the +# comment used to call it "defensive"; it was actually the vector. MAX_SUBSCRIBERS = 500 +# How many concurrent streams a single client IP may hold. Deliberately small — +# a real browser tab needs at most one, occasionally two briefly across a +# reload — since this bounds one source's share of the global capacity, not a +# legitimate per-user concurrency limit. +MAX_SUBSCRIBERS_PER_IP = 5 + +# Put on a to-be-evicted subscriber's queue (B-38) to wake its generator +# (app/api/routes/rounds.py:round_stream) promptly so it closes the connection +# instead of lingering, silently uncounted, until the client's own network +# timeout or the next keep-alive tick. +EVICTED = object() + class RoundEventCapacityError(Exception): - """Raised by subscribe() when MAX_SUBSCRIBERS is already reached.""" + """Raised by subscribe() when MAX_SUBSCRIBERS — the global backstop — is + already reached. The per-IP cap never raises this; it evicts instead (see + subscribe()).""" class RoundEventBroadcaster: @@ -26,22 +41,47 @@ class RoundEventBroadcaster: would need a shared channel (e.g. Redis pub/sub) instead. """ - def __init__(self, max_subscribers: int = MAX_SUBSCRIBERS): - self._subscribers: set[asyncio.Queue] = set() + def __init__(self, max_subscribers: int = MAX_SUBSCRIBERS, max_per_ip: int = MAX_SUBSCRIBERS_PER_IP): + self._ip_by_queue: dict[asyncio.Queue, str] = {} + self._queues_by_ip: dict[str, list[asyncio.Queue]] = defaultdict(list) self.max_subscribers = max_subscribers + self.max_per_ip = max_per_ip - def subscribe(self) -> asyncio.Queue: - if len(self._subscribers) >= self.max_subscribers: + def subscribe(self, client_ip: str = "unknown") -> asyncio.Queue: + if len(self._ip_by_queue) >= self.max_subscribers: raise RoundEventCapacityError(f"already at the {self.max_subscribers}-subscriber cap") + + ip_queues = self._queues_by_ip[client_ip] + if len(ip_queues) >= self.max_per_ip: + # B-38: evict this IP's own oldest connection rather than refusing + # the new one — bounds one source's footprint without turning a + # legitimate reconnect storm (a flaky network retrying EventSource) + # into an outright block, and without letting one abusive IP crowd + # out unrelated clients the way the old global-only cap did. + oldest = ip_queues.pop(0) + self._ip_by_queue.pop(oldest, None) + if not oldest.full(): + oldest.put_nowait(EVICTED) + queue: asyncio.Queue = asyncio.Queue(maxsize=1) - self._subscribers.add(queue) + self._ip_by_queue[queue] = client_ip + ip_queues.append(queue) return queue def unsubscribe(self, queue: asyncio.Queue) -> None: - self._subscribers.discard(queue) + client_ip = self._ip_by_queue.pop(queue, None) + if client_ip is None: + return + ip_queues = self._queues_by_ip.get(client_ip) + if ip_queues is None: + return + if queue in ip_queues: + ip_queues.remove(queue) + if not ip_queues: + self._queues_by_ip.pop(client_ip, None) def publish(self) -> None: - for queue in self._subscribers: + for queue in self._ip_by_queue: if queue.full(): continue # a not-yet-delivered notification already covers this one queue.put_nowait(None) diff --git a/tests/unit/test_client_ip.py b/tests/unit/test_client_ip.py new file mode 100644 index 0000000..acabfa4 --- /dev/null +++ b/tests/unit/test_client_ip.py @@ -0,0 +1,43 @@ +"""app.api.client_ip is shared by the login/registration throttles (B-33) and +the SSE per-IP subscriber cap (B-38) — both depend on it correctly preferring +X-Forwarded-For (Caddy reverse-proxies every request, see Caddyfile) over +request.client.host, which would otherwise be the proxy's own address.""" + +from starlette.requests import Request + +from app.api.client_ip import client_ip + + +def _request(*, forwarded: str | None = None, client_host: str | None = "127.0.0.1") -> Request: + headers = [(b"x-forwarded-for", forwarded.encode())] if forwarded else [] + scope = { + "type": "http", + "headers": headers, + "client": (client_host, 12345) if client_host else None, + } + return Request(scope) + + +def test_client_ip_prefers_x_forwarded_for(): + request = _request(forwarded="5.6.7.8", client_host="10.0.0.1") + assert client_ip(request) == "5.6.7.8" + + +def test_client_ip_takes_the_first_hop_of_a_forwarded_chain(): + request = _request(forwarded="5.6.7.8, 10.0.0.1, 172.17.0.1") + assert client_ip(request) == "5.6.7.8" + + +def test_client_ip_strips_whitespace(): + request = _request(forwarded=" 5.6.7.8 , 10.0.0.1") + assert client_ip(request) == "5.6.7.8" + + +def test_client_ip_falls_back_to_request_client_without_the_header(): + request = _request(forwarded=None, client_host="10.0.0.1") + assert client_ip(request) == "10.0.0.1" + + +def test_client_ip_falls_back_to_unknown_with_neither(): + request = _request(forwarded=None, client_host=None) + assert client_ip(request) == "unknown" diff --git a/tests/unit/test_round_events.py b/tests/unit/test_round_events.py index 6425d06..7607fe8 100644 --- a/tests/unit/test_round_events.py +++ b/tests/unit/test_round_events.py @@ -2,7 +2,7 @@ import asyncio import pytest -from app.rounds.events import RoundEventBroadcaster, RoundEventCapacityError +from app.rounds.events import EVICTED, RoundEventBroadcaster, RoundEventCapacityError async def test_publish_wakes_up_subscriber(): @@ -60,3 +60,71 @@ async def test_unsubscribe_frees_a_capacity_slot(): broadcaster.unsubscribe(queue) broadcaster.subscribe() # no longer at capacity + + +# --- B-38: a single IP must not be able to exhaust the global cap and degrade +# every other user to polling. ---------------------------------------------------- + + +async def test_subscribe_evicts_the_same_ips_oldest_connection_past_its_cap(): + """Past MAX_SUBSCRIBERS_PER_IP, one more stream from the *same* IP evicts + that IP's own oldest connection rather than being refused — bounds one + source's footprint without an outright block.""" + broadcaster = RoundEventBroadcaster(max_subscribers=100, max_per_ip=2) + first = broadcaster.subscribe("1.2.3.4") + second = broadcaster.subscribe("1.2.3.4") + + third = broadcaster.subscribe("1.2.3.4") # past the per-IP cap of 2 + + assert await asyncio.wait_for(first.get(), timeout=1) is EVICTED + assert second.empty() # untouched — only the oldest was evicted + assert third is not None + + +async def test_subscribe_does_not_evict_across_different_ips(): + """A different IP hitting its own cap must never evict an unrelated IP's + connection — that would let one abusive source crowd out real users, which + is exactly what the global-only cap used to allow.""" + broadcaster = RoundEventBroadcaster(max_subscribers=100, max_per_ip=1) + other_ip_queue = broadcaster.subscribe("9.9.9.9") + + broadcaster.subscribe("1.2.3.4") + broadcaster.subscribe("1.2.3.4") # evicts 1.2.3.4's own oldest, not 9.9.9.9's + + assert other_ip_queue.empty() + + +async def test_subscribe_still_enforces_the_global_cap_across_many_ips(): + """The per-IP cap doesn't replace the global backstop — spreading across + enough distinct IPs must still eventually hit MAX_SUBSCRIBERS.""" + broadcaster = RoundEventBroadcaster(max_subscribers=3, max_per_ip=1) + broadcaster.subscribe("1.1.1.1") + broadcaster.subscribe("2.2.2.2") + broadcaster.subscribe("3.3.3.3") + + with pytest.raises(RoundEventCapacityError): + broadcaster.subscribe("4.4.4.4") + + +async def test_unsubscribe_clears_the_per_ip_tracking_too(): + """Regression guard: unsubscribe must forget the queue's IP association, or + a churned-through connection would keep counting against that IP's cap + forever.""" + broadcaster = RoundEventBroadcaster(max_subscribers=100, max_per_ip=1) + queue = broadcaster.subscribe("1.2.3.4") + broadcaster.unsubscribe(queue) + + broadcaster.subscribe("1.2.3.4") # must not evict anything — nothing left to evict + assert queue.empty() + + +async def test_subscribe_defaults_to_a_shared_ip_when_none_given(): + """Existing callers (and most tests) that don't care about IP isolation + still share one implicit bucket rather than needing every call updated.""" + broadcaster = RoundEventBroadcaster(max_subscribers=100, max_per_ip=2) + first = broadcaster.subscribe() + broadcaster.subscribe() + + broadcaster.subscribe() # past the default bucket's cap of 2 — evicts, doesn't raise + + assert await asyncio.wait_for(first.get(), timeout=1) is EVICTED