From 1cf2f01a9a21a0e8e3c1f29f972e7be10331e955 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sun, 25 Oct 2026 01:28:24 -0700 Subject: [PATCH 1/5] Verify private-channel signatures and fix /broadcasting/auth 404 The Reverb server subscribed any socket to private and presence channels without checking auth, and the empty-path WebSocket mount shadowed /broadcasting/auth and every route added after boot. The auth endpoint now returns a Pusher HMAC signature over socket_id and channel, the server verifies it on subscribe, and both endpoints are plain router routes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DCf9PvS7udMZZFFz55EhPo --- docs/adr/035-broadcasting-channel-auth.md | 34 ++++ docs/adr/AGENTS.md | 1 + .../fastapi_startkit/broadcasting/provider.py | 100 ++++------ .../broadcasting/reverb/server.py | 23 ++- .../broadcasting/reverb/signature.py | 19 ++ .../broadcasting/test_private_channel_auth.py | 163 ++++++++++++++++ .../tests/broadcasting/test_provider.py | 178 +++++++++--------- 7 files changed, 371 insertions(+), 147 deletions(-) create mode 100644 docs/adr/035-broadcasting-channel-auth.md create mode 100644 fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py create mode 100644 fastapi_startkit/tests/broadcasting/test_private_channel_auth.py diff --git a/docs/adr/035-broadcasting-channel-auth.md b/docs/adr/035-broadcasting-channel-auth.md new file mode 100644 index 00000000..e094b76e --- /dev/null +++ b/docs/adr/035-broadcasting-channel-auth.md @@ -0,0 +1,34 @@ +# 035. Signed private-channel subscriptions and a reachable /broadcasting/auth + +Date: 2026-10-25 + +## Problem + +Two defects in `fastapi_startkit.broadcasting`: + +- **F4, auth bypass.** `ReverbServer` subscribed every socket to any channel named in `pusher:subscribe`, including `private-*` and `presence-*`. The `auth` field was never checked, so any WebSocket client received private broadcasts without calling `/broadcasting/auth`. The endpoint also returned `{"auth": true}` rather than a Pusher signature, so a real Echo client could not have been verified even with a check in place. +- **F3, 404.** `ReverbProvider` mounted the WebSocket app with `fastapi_app.mount("", ...)` before adding `POST /broadcasting/auth`. Starlette tries routes in order and an empty-path `Mount` matches every path, so the auth route, and every route registered after the provider boots, returned the inner app's 404. The endpoint also used a function-local `Request` import under `from __future__ import annotations`, so FastAPI could not resolve the annotation and would have treated `request` as a query parameter. + +## Alternatives + +1. Track authorised `(socket_id, channel)` pairs in server memory when `/broadcasting/auth` succeeds and check them on subscribe. This is not the Pusher protocol, so it breaks Echo, and it fails with more than one worker. +2. Use the Pusher protocol: `/broadcasting/auth` returns `key:HMAC-SHA256(secret, "socket_id:channel[:channel_data]")` and the server recomputes the signature on subscribe. This is stateless, works across workers and matches what Laravel Echo sends. + +For F3, keep the mount and register it last, or register plain routes on the FastAPI router. Ordering cannot be guaranteed, because application routes can be added after any provider boots. + +## Decision + +Option 2 for F4, and plain routes for F3. + +## Implementation + +- `broadcasting/reverb/signature.py`: `requires_auth(channel)`, `sign(key, secret, socket_id, channel, channel_data=None)` and `verify(...)`, which compares with `hmac.compare_digest` and rejects non-string `auth` or `channel_data`. +- `ReverbServer(key, secret)`: on a `private-`/`presence-` subscribe with a missing or invalid signature it sends `pusher:subscription_error` (`status: 401`) and does not subscribe the socket. Public channels are unchanged. +- `ReverbProvider.register` builds the server from `broadcasting.connections.reverb.key`/`secret`. +- `ReverbProvider` registers `/app/{key}` with `router.add_websocket_route` and `/broadcasting/auth` with `router.add_route`, a Starlette endpoint, so neither depends on route order or annotation resolution. +- `/broadcasting/auth` requires string `channel_name` and `socket_id` (422 otherwise) and returns 403 when the channel callback denies. On success it returns `{"auth": signature}`. Presence channels also return `channel_data` (`{"user_id": ""}`), which is included in the signature. + +## Validation + +- Tests drive the fully booted app through `TestClient`: the auth route and routes added after boot are reachable whatever the provider order. A private subscribe without a signature is rejected and receives no broadcasts. Forged, wrong-socket, wrong-secret and non-string signatures are rejected. A signature obtained from `/broadcasting/auth` subscribes and receives broadcasts. Tampered presence `channel_data` is rejected. +- Full suite: `uv run pytest --ignore=tests/masoniteorm/postgres`. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index c13f7f1b..53d84e32 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -35,3 +35,4 @@ Read this index first, then open the records relevant to the change. Before impl | [029](029-cache-module.md) | 2026-10-21 | Cache module with remember, memo, locks and tags | New `fastapi_startkit.cache`: async `Cache.get/put/has/forget/flush`, `get` with a callable default, and `remember(key, ttl, factory)` that runs a sync or async factory only on a miss; `remember(key, ttl)` without a factory is a decorator keyed by `key` plus a hash of the call arguments. Stores implement an async contract with a `MISSING` sentinel (cached `None` is a hit): array, file (pickled per-key files), database (masoniteorm table) and redis (named connection from the existing `RedisManager`, `SET PX` TTL, flush by `SCAN` of prefixed keys, never `FLUSHDB`). TTL in seconds or `timedelta`, `None` forever, <=0 forgets. `CacheConfig` via `resolve_config`; `CacheProvider` binds `cache`; facade `remember` resolves lazily. `Cache.memo()` memoizes reads per asyncio task through a `ContextVar` owned by the task, so requests never share memo state. `Cache.lock(name, seconds, owner)` gives atomic, owner-checked, non-re-entrant locks with `get`, `block` (250 ms polling, `LockTimeoutError`), `release`, `force_release` and `async with`, per store: event-loop dict, `fcntl.flock` file, `lock:` row with update-if-expired, Redis `SET NX PX` plus a Lua owner-checked delete. `Cache.tags([...])` uses Laravel's namespaced keys: random tag ids at `:tag::key`, entries at `:sha1(ids):`, flush rewrites the ids; tags are sorted and deduplicated. Shipped in slices into `feature/caches`; the former memo, lock and tags ADRs (030, 031, 034) are folded in. | | [032](032-migrate-seed-option.md) | 2026-10-21 | db:migrate:fresh and db:migrate:refresh --seed run seeders in the same event loop | `--seed` called the unregistered `seed:run` through Cleo `call` (which also drops the first argument) inside a running event loop, where `db:seed` would call `asyncio.run` again. The seeder selection moves into async `DBSeedCommand.run_seeders`, which `db:seed`, `db:migrate:fresh` and `db:migrate:refresh` all await. `db:migrate:refresh` is now registered. | | [033](033-async-seeder-stub.md) | 2026-10-21 | Generated seeders have an async run | `create_seed.stub` generated a synchronous `run` with docstrings, but `Seeder` awaits `run()`, so every generated table seeder raised `TypeError`. The stub now generates `async def run(self) -> None` with no docstrings; `Seeder` is unchanged. | +| [035](035-broadcasting-channel-auth.md) | 2026-10-25 | Signed private-channel subscriptions and a reachable /broadcasting/auth | `ReverbServer` subscribed any socket to `private-`/`presence-` channels without checking `auth`, and the empty-path WebSocket `Mount` shadowed `/broadcasting/auth` and later routes (404). The auth endpoint now requires `socket_id` and returns a Pusher signature `key:HMAC-SHA256(secret, socket_id:channel[:channel_data])` (presence adds signed `channel_data` with `user_id`). The server verifies it with `compare_digest` and replies `pusher:subscription_error` 401 on failure. Both endpoints are registered as plain router routes, so route order does not matter. | diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py index 63f0ae25..5507b476 100644 --- a/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py @@ -2,6 +2,7 @@ import importlib import importlib.util +import json import sys from pathlib import Path @@ -9,6 +10,7 @@ from .config import BroadcastingConfig from .manager import BroadcastManager from .reverb.server import ReverbServer +from .reverb.signature import sign class ReverbProvider(Provider): @@ -36,7 +38,8 @@ def register(self) -> None: config_data = self.resolve_config(BroadcastingConfig) self.merge_config_from(config_data, "broadcasting") - server = ReverbServer() + reverb = config_data.get("connections", {}).get("reverb", {}) + server = ReverbServer(key=reverb.get("key", "local"), secret=reverb.get("secret", "secret")) manager = BroadcastManager(config_data, server) self.app.bind("broadcasting", manager) @@ -81,91 +84,66 @@ def _load_channels_file(self) -> None: spec.loader.exec_module(module) # type: ignore[union-attr] def _mount_websocket(self) -> None: - """Mount the Reverb WebSocket endpoint on the FastAPI application. - - The endpoint path is ``/app/{app_key}``, matching the Pusher protocol - expected by Laravel Echo. The ``app_key`` is taken from the - ``reverb.connections.reverb.key`` config value (default: ``"local"``). - """ try: fastapi_app = self.app.fastapi except RuntimeError: - # FastAPI not installed or app not yet booted — skip return - config = self.app.make("config").get("broadcasting") or {} - connections = config.get("connections", {}) if isinstance(config, dict) else {} - reverb_conn = connections.get("reverb", {}) if isinstance(connections, dict) else {} - app_key = reverb_conn.get("key", "local") if isinstance(reverb_conn, dict) else "local" - server: ReverbServer = self.app.make("reverb.server") - from starlette.routing import WebSocketRoute - from starlette.applications import Starlette - async def ws_endpoint(websocket): await server.handle(websocket) - ws_app = Starlette(routes=[WebSocketRoute(f"/app/{app_key}", ws_endpoint)]) - fastapi_app.mount("", ws_app) + fastapi_app.router.add_websocket_route(f"/app/{server.key}", ws_endpoint) def _mount_auth_endpoint(self) -> None: - """Mount the ``/broadcasting/auth`` HTTP endpoint. - - The endpoint handles the Laravel Echo authentication handshake for - private and presence channels. It: - - 1. Reads ``channel_name`` from the POST body (form or JSON). - 2. Resolves the current user via the container's ``auth`` binding (if - available), falling back to ``request.state.user``. - 3. Calls :meth:`~.auth.ChannelAuthRegistry.authorize` with the channel - name and the resolved user. - 4. Returns **200** on success or **403** on denial. - """ try: fastapi_app = self.app.fastapi except RuntimeError: return - from fastapi import Request - from fastapi.responses import JSONResponse + from starlette.requests import Request + from starlette.responses import JSONResponse application = self.app - @fastapi_app.post("/broadcasting/auth") async def broadcasting_auth(request: Request): - # Resolve channel name from form data or JSON body content_type = request.headers.get("content-type", "") if "application/json" in content_type: body = await request.json() - channel_name = body.get("channel_name", "") else: - form = await request.form() - channel_name = form.get("channel_name", "") - - if not isinstance(channel_name, str): - return JSONResponse({"error": "channel_name must be a string"}, status_code=422) - - # Resolve authenticated user. - # Prefer request.state.user (set by auth middleware), then fall - # back to the container's "auth" service. - user = getattr(request.state, "user", None) - if user is None: - try: - auth_service = application.make("auth") - user_attr = getattr(auth_service, "user", None) - if callable(user_attr): - user = user_attr() - else: - user = user_attr - except Exception: - user = None - - # Authorise - manager: BroadcastManager = application.make("broadcasting") - allowed = await manager.channel_registry.authorize(channel_name, user) + body = await request.form() + channel_name = body.get("channel_name", "") + socket_id = body.get("socket_id", "") - if not allowed: + if not isinstance(channel_name, str) or not isinstance(socket_id, str): + return JSONResponse({"error": "channel_name and socket_id must be strings"}, status_code=422) + if not socket_id: + return JSONResponse({"error": "socket_id is required"}, status_code=422) + + user = self._resolve_user(application, request) + + manager: BroadcastManager = application.make("broadcasting") + if not await manager.channel_registry.authorize(channel_name, user): return JSONResponse({"error": "Forbidden"}, status_code=403) - return JSONResponse({"auth": True, "channel_data": {}}) + server: ReverbServer = application.make("reverb.server") + if not channel_name.startswith("presence-"): + return JSONResponse({"auth": sign(server.key, server.secret, socket_id, channel_name)}) + + channel_data = json.dumps({"user_id": str(getattr(user, "id", ""))}) + auth = sign(server.key, server.secret, socket_id, channel_name, channel_data) + return JSONResponse({"auth": auth, "channel_data": channel_data}) + + fastapi_app.router.add_route("/broadcasting/auth", broadcasting_auth, methods=["POST"]) + + @staticmethod + def _resolve_user(application, request): + user = getattr(request.state, "user", None) + if user is not None: + return user + try: + user_attr = getattr(application.make("auth"), "user", None) + except Exception: + return None + return user_attr() if callable(user_attr) else user_attr diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py index aff7b65c..50cd5fce 100644 --- a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py @@ -4,10 +4,13 @@ from starlette.applications import Starlette from .connection import ConnectionManager from .channel_manager import ChannelManager +from .signature import requires_auth, verify class ReverbServer: - def __init__(self): + def __init__(self, key: str = "local", secret: str = "secret"): + self.key = key + self.secret = secret self.connections = ConnectionManager() self.channels = ChannelManager() @@ -36,7 +39,23 @@ async def _handle_message(self, socket_id: str, websocket: WebSocket, message: d if event == "pusher:ping": await websocket.send_text(json.dumps({"event": "pusher:pong", "data": {}})) elif event == "pusher:subscribe": - channel = message.get("data", {}).get("channel", "") + data = message.get("data") or {} + channel = data.get("channel", "") + if not isinstance(channel, str) or not channel: + return + if requires_auth(channel) and not verify( + self.key, self.secret, socket_id, channel, data.get("auth"), data.get("channel_data") + ): + await websocket.send_text( + json.dumps( + { + "event": "pusher:subscription_error", + "channel": channel, + "data": {"type": "AuthError", "error": "Invalid signature", "status": 401}, + } + ) + ) + return self.channels.subscribe(channel, socket_id) await websocket.send_text( json.dumps({"event": "pusher_internal:subscription_succeeded", "channel": channel, "data": {}}) diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py new file mode 100644 index 00000000..3084ecad --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py @@ -0,0 +1,19 @@ +import hashlib +import hmac + + +def requires_auth(channel: object) -> bool: + return isinstance(channel, str) and channel.startswith(("private-", "presence-")) + + +def sign(key: str, secret: str, socket_id: str, channel: str, channel_data: str | None = None) -> str: + parts = [socket_id, channel] if channel_data is None else [socket_id, channel, channel_data] + digest = hmac.new(secret.encode(), ":".join(parts).encode(), hashlib.sha256).hexdigest() + return f"{key}:{digest}" + + +def verify(key: str, secret: str, socket_id: str, channel: str, auth: object, channel_data: object = None) -> bool: + if not isinstance(auth, str) or not (channel_data is None or isinstance(channel_data, str)): + return False + expected = sign(key, secret, socket_id, channel, channel_data) + return hmac.compare_digest(auth.encode(), expected.encode()) diff --git a/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py b/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py new file mode 100644 index 00000000..47fb8cba --- /dev/null +++ b/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py @@ -0,0 +1,163 @@ +import asyncio +import json + +import pytest +from starlette.testclient import TestClient + +from fastapi_startkit.application import Application +from fastapi_startkit.broadcasting.provider import ReverbProvider +from fastapi_startkit.support import Provider + +KEY = "local" + + +class RouteProvider(Provider): + def register(self) -> None: + pass + + def boot(self) -> None: + @self.app.fastapi.get("/ping") + async def ping(): + return {"pong": True} + + +class User: + def __init__(self, id): + self.id = id + + +@pytest.fixture +def base_path(tmp_path): + routes = tmp_path / "routes" + routes.mkdir() + (routes / "channels.py").write_text( + "from fastapi_startkit.broadcasting import channel\n" + "\n" + "@channel('private-orders.{order_id}')\n" + "async def authorize_orders(user, order_id: int):\n" + " return user is not None and user.id == order_id\n" + "\n" + "@channel('presence-room.{room}')\n" + "async def authorize_room(user, room):\n" + " return user is not None\n" + ) + return tmp_path + + +def make_client(base_path, providers, user=None): + app = Application(base_path=base_path, env="testing", providers=providers) + + @app.fastapi.middleware("http") + async def attach_user(request, call_next): + request.state.user = user + return await call_next(request) + + @app.fastapi.get("/after") + async def after(): + return {"after": True} + + return app, TestClient(app.fastapi) + + +def connect(ws): + return json.loads(json.loads(ws.receive_text())["data"])["socket_id"] + + +def subscribe(ws, channel, auth=None, channel_data=None): + data = {"channel": channel} + if auth is not None: + data["auth"] = auth + if channel_data is not None: + data["channel_data"] = channel_data + ws.send_text(json.dumps({"event": "pusher:subscribe", "data": data})) + return json.loads(ws.receive_text()) + + +def ping(ws): + ws.send_text(json.dumps({"event": "pusher:ping", "data": {}})) + return json.loads(ws.receive_text()) + + +class TestRouteOrder: + @pytest.mark.parametrize("reverb_first", [True, False]) + def test_routes_and_auth_endpoint_reachable(self, base_path, reverb_first): + providers = [ReverbProvider, RouteProvider] if reverb_first else [RouteProvider, ReverbProvider] + _, client = make_client(base_path, providers, user=User(7)) + + assert client.get("/ping").json() == {"pong": True} + assert client.get("/after").json() == {"after": True} + response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": "private-orders.7"}) + assert response.status_code == 200 + + with client.websocket_connect(f"/app/{KEY}") as ws: + assert connect(ws) + + +class TestPrivateSubscription: + def test_unauthenticated_subscribe_is_rejected_and_receives_nothing(self, base_path): + app, client = make_client(base_path, [ReverbProvider]) + server = app.make("reverb.server") + + with client.websocket_connect(f"/app/{KEY}") as ws: + socket_id = connect(ws) + reply = subscribe(ws, "private-orders.7") + assert reply["event"] == "pusher:subscription_error" + assert reply["channel"] == "private-orders.7" + assert reply["data"]["status"] == 401 + assert socket_id not in server.channels.subscribers("private-orders.7") + + asyncio.run(server.broadcast_to_channel("private-orders.7", "OrderShipped", {"secret": 1})) + assert ping(ws)["event"] == "pusher:pong" + + def test_forged_signature_is_rejected(self, base_path): + _, client = make_client(base_path, [ReverbProvider]) + with client.websocket_connect(f"/app/{KEY}") as ws: + connect(ws) + assert subscribe(ws, "private-orders.7", auth=f"{KEY}:{'0' * 64}")["event"] == "pusher:subscription_error" + assert subscribe(ws, "private-orders.7", auth="ünïcode")["event"] == "pusher:subscription_error" + assert subscribe(ws, "private-orders.7", auth=123)["event"] == "pusher:subscription_error" + + def test_signature_for_another_socket_is_rejected(self, base_path): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + auth = client.post( + "/broadcasting/auth", data={"socket_id": "someone-else", "channel_name": "private-orders.7"} + ).json()["auth"] + with client.websocket_connect(f"/app/{KEY}") as ws: + connect(ws) + assert subscribe(ws, "private-orders.7", auth=auth)["event"] == "pusher:subscription_error" + + def test_unauthorized_user_cannot_obtain_signature(self, base_path): + _, client = make_client(base_path, [ReverbProvider], user=User(8)) + response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": "private-orders.7"}) + assert response.status_code == 403 + + def test_authorized_subscribe_receives_broadcast(self, base_path): + app, client = make_client(base_path, [ReverbProvider], user=User(7)) + server = app.make("reverb.server") + with client.websocket_connect(f"/app/{KEY}") as ws: + socket_id = connect(ws) + auth = client.post( + "/broadcasting/auth", json={"socket_id": socket_id, "channel_name": "private-orders.7"} + ).json()["auth"] + assert subscribe(ws, "private-orders.7", auth=auth)["event"] == "pusher_internal:subscription_succeeded" + + asyncio.run(server.broadcast_to_channel("private-orders.7", "OrderShipped", {"order_id": 7})) + message = json.loads(ws.receive_text()) + assert message["event"] == "OrderShipped" + assert message["data"] == {"order_id": 7} + + +class TestPresenceSubscription: + def test_presence_requires_signed_channel_data(self, base_path): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + with client.websocket_connect(f"/app/{KEY}") as ws: + socket_id = connect(ws) + body = client.post( + "/broadcasting/auth", json={"socket_id": socket_id, "channel_name": "presence-room.1"} + ).json() + assert json.loads(body["channel_data"])["user_id"] == "7" + + tampered = json.dumps({"user_id": "1", "user_info": {}}) + assert subscribe(ws, "presence-room.1", body["auth"], tampered)["event"] == "pusher:subscription_error" + reply = subscribe(ws, "presence-room.1", body["auth"], body["channel_data"]) + assert reply["event"] == "pusher_internal:subscription_succeeded" diff --git a/fastapi_startkit/tests/broadcasting/test_provider.py b/fastapi_startkit/tests/broadcasting/test_provider.py index 8b3f0c10..61971d44 100644 --- a/fastapi_startkit/tests/broadcasting/test_provider.py +++ b/fastapi_startkit/tests/broadcasting/test_provider.py @@ -1,60 +1,29 @@ +import json from types import SimpleNamespace +from starlette.testclient import TestClient + from fastapi_startkit.application import Application from fastapi_startkit.broadcasting.manager import BroadcastManager from fastapi_startkit.broadcasting.provider import ReverbProvider from fastapi_startkit.broadcasting.reverb.server import ReverbServer - - -_MISSING = object() +from fastapi_startkit.broadcasting.reverb.signature import sign def make_app(tmp_path): return Application(base_path=tmp_path, env="testing", providers=[ReverbProvider]) -def mount_auth(tmp_path): - """Register broadcasting and mount only the auth endpoint. - - The full ``boot()`` mounts a catch-all WebSocket sub-app at ``""`` which - shadows the HTTP auth route, so exercise the auth handler on its own. - """ - app = Application(base_path=tmp_path, env="testing", providers=[]) - provider = ReverbProvider(app) - provider.register() - provider._mount_auth_endpoint() - return app - +def client_for(app): + return TestClient(app.fastapi) -def auth_handler(app): - for route in app.fastapi.routes: - if getattr(route, "path", None) == "/broadcasting/auth": - return route.endpoint - raise AssertionError("auth route not mounted") +def register_rule(app, pattern, callback): + app.make("broadcasting").channel_registry.register(pattern, callback) -class FakeRequest: - """Stand-in for a Starlette Request. - ``from __future__ import annotations`` in the provider turns the handler's - ``Request`` annotation into an unresolved forward-ref, so FastAPI cannot - inject a real request through the test client. Calling the handler with - this fake exercises the same body directly. - """ - - def __init__(self, content_type, *, json_body=None, form_body=None, user=_MISSING): - self.headers = {"content-type": content_type} - self._json = json_body or {} - self._form = form_body or {} - self.state = SimpleNamespace() - if user is not _MISSING: - self.state.user = user - - async def json(self): - return self._json - - async def form(self): - return self._form +def post_auth(app, **body): + return client_for(app).post("/broadcasting/auth", data=body) class TestRegister: @@ -63,6 +32,10 @@ def test_binds_manager_and_server(self, tmp_path): assert isinstance(app.make("broadcasting"), BroadcastManager) assert isinstance(app.make("reverb.server"), ReverbServer) + def test_server_uses_configured_credentials(self, tmp_path): + server = make_app(tmp_path).make("reverb.server") + assert (server.key, server.secret) == ("local", "secret") + def test_publishes_channels_stub(self, tmp_path): app = make_app(tmp_path) published = app.published_resources.get("broadcasting", {}) @@ -70,53 +43,96 @@ def test_publishes_channels_stub(self, tmp_path): class TestAuthEndpoint: - async def test_public_channel_allowed_json(self, tmp_path): - handler = auth_handler(mount_auth(tmp_path)) - response = await handler(FakeRequest("application/json", json_body={"channel_name": "orders"})) - assert response.status_code == 200 - - async def test_private_channel_denied_without_rule(self, tmp_path): - handler = auth_handler(mount_auth(tmp_path)) - response = await handler(FakeRequest("application/json", json_body={"channel_name": "private-secret"})) + def test_auth_route_is_reachable_after_full_boot(self, tmp_path): + response = post_auth(make_app(tmp_path), channel_name="private-secret", socket_id="1.1") assert response.status_code == 403 - async def test_accepts_form_encoded_body(self, tmp_path): - handler = auth_handler(mount_auth(tmp_path)) - response = await handler( - FakeRequest("application/x-www-form-urlencoded", form_body={"channel_name": "public-room"}) - ) + def test_routes_added_after_boot_are_not_shadowed(self, tmp_path): + app = make_app(tmp_path) + app.fastapi.get("/ping")(lambda: {"ok": True}) + assert client_for(app).get("/ping").status_code == 200 + + def test_private_channel_returns_signed_auth(self, tmp_path): + app = make_app(tmp_path) + register_rule(app, "private-room.{room}", lambda user, room: True) + response = post_auth(app, channel_name="private-room.7", socket_id="123.456") assert response.status_code == 200 + assert response.json() == {"auth": sign("local", "secret", "123.456", "private-room.7")} + + def test_accepts_json_body(self, tmp_path): + app = make_app(tmp_path) + register_rule(app, "private-room.{room}", lambda user, room: True) + response = client_for(app).post( + "/broadcasting/auth", json={"channel_name": "private-room.7", "socket_id": "123.456"} + ) + assert response.json()["auth"] == sign("local", "secret", "123.456", "private-room.7") + + def test_presence_channel_signs_channel_data(self, tmp_path): + app = make_app(tmp_path) + register_rule(app, "presence-room", lambda user: True) + app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=9))) + body = post_auth(app, channel_name="presence-room", socket_id="1.2").json() + assert body["channel_data"] == '{"user_id": "9"}' + assert body["auth"] == sign("local", "secret", "1.2", "presence-room", body["channel_data"]) + + def test_denied_callback_returns_403(self, tmp_path): + app = make_app(tmp_path) + register_rule(app, "private-room.{room}", lambda user, room: False) + assert post_auth(app, channel_name="private-room.7", socket_id="1.1").status_code == 403 - async def test_rejects_non_string_json_channel_name(self, tmp_path): - handler = auth_handler(mount_auth(tmp_path)) - response = await handler(FakeRequest("application/json", json_body={"channel_name": 42})) + def test_requires_socket_id(self, tmp_path): + assert post_auth(make_app(tmp_path), channel_name="orders").status_code == 422 + + def test_rejects_non_string_json_channel_name(self, tmp_path): + response = client_for(make_app(tmp_path)).post( + "/broadcasting/auth", json={"channel_name": 42, "socket_id": "1.1"} + ) assert response.status_code == 422 - async def test_rejects_file_upload_as_channel_name(self, tmp_path): - handler = auth_handler(mount_auth(tmp_path)) - response = await handler( - FakeRequest("multipart/form-data", form_body={"channel_name": SimpleNamespace(filename="x.txt")}) + def test_rejects_file_upload_as_channel_name(self, tmp_path): + response = client_for(make_app(tmp_path)).post( + "/broadcasting/auth", data={"socket_id": "1.1"}, files={"channel_name": ("x.txt", b"x")} ) assert response.status_code == 422 - async def test_uses_request_state_user(self, tmp_path): - handler = auth_handler(mount_auth(tmp_path)) - response = await handler(FakeRequest("application/json", json_body={"channel_name": "orders"}, user=object())) - assert response.status_code == 200 + def test_uses_request_state_user(self, tmp_path): + app = make_app(tmp_path) + register_rule(app, "private-me", lambda user: user.id == 3) + + @app.fastapi.middleware("http") + async def authenticate(request, call_next): + request.state.user = SimpleNamespace(id=3) + return await call_next(request) - async def test_resolves_user_from_callable_auth_service(self, tmp_path): - app = mount_auth(tmp_path) + assert post_auth(app, channel_name="private-me", socket_id="1.1").status_code == 200 + + def test_resolves_user_from_callable_auth_service(self, tmp_path): + app = make_app(tmp_path) + register_rule(app, "private-me", lambda user: user.id == 1) app.bind("auth", SimpleNamespace(user=lambda: SimpleNamespace(id=1))) - handler = auth_handler(app) - response = await handler(FakeRequest("application/json", json_body={"channel_name": "orders"})) - assert response.status_code == 200 + assert post_auth(app, channel_name="private-me", socket_id="1.1").status_code == 200 - async def test_resolves_user_from_attribute_auth_service(self, tmp_path): - app = mount_auth(tmp_path) + def test_resolves_user_from_attribute_auth_service(self, tmp_path): + app = make_app(tmp_path) + register_rule(app, "private-me", lambda user: user.id == 2) app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=2))) - handler = auth_handler(app) - response = await handler(FakeRequest("application/json", json_body={"channel_name": "orders"})) - assert response.status_code == 200 + assert post_auth(app, channel_name="private-me", socket_id="1.1").status_code == 200 + + +class TestWebsocketSubscription: + def test_private_subscribe_requires_signature_from_auth_endpoint(self, tmp_path): + app = make_app(tmp_path) + register_rule(app, "private-room.{room}", lambda user, room: True) + client = client_for(app) + with client.websocket_connect("/app/local") as ws: + socket_id = json.loads(json.loads(ws.receive_text())["data"])["socket_id"] + + ws.send_text(json.dumps({"event": "pusher:subscribe", "data": {"channel": "private-room.7"}})) + assert json.loads(ws.receive_text())["event"] == "pusher:subscription_error" + + auth = post_auth(app, channel_name="private-room.7", socket_id=socket_id).json()["auth"] + ws.send_text(json.dumps({"event": "pusher:subscribe", "data": {"channel": "private-room.7", "auth": auth}})) + assert json.loads(ws.receive_text())["event"] == "pusher_internal:subscription_succeeded" class TestChannelsFileAutoloading: @@ -136,14 +152,11 @@ async def test_registered_channel_authorizer_is_used(self, tmp_path): assert await registry.authorize("private-room.7", None) is True def test_missing_channels_file_is_ignored(self, tmp_path): - # No routes/channels.py — boot must still succeed. app = make_app(tmp_path) assert isinstance(app.make("broadcasting"), BroadcastManager) class _RaisingFastapiApp: - """Minimal app whose ``.fastapi`` raises, mimicking FastAPI not installed.""" - base_path = "/nonexistent-base-path" @property @@ -153,13 +166,10 @@ def fastapi(self): class TestMountingWithoutFastapi: def test_mount_websocket_skips_when_fastapi_unavailable(self): - provider = ReverbProvider(_RaisingFastapiApp()) - provider._mount_websocket() # returns without raising + ReverbProvider(_RaisingFastapiApp())._mount_websocket() def test_mount_auth_endpoint_skips_when_fastapi_unavailable(self): - provider = ReverbProvider(_RaisingFastapiApp()) - provider._mount_auth_endpoint() # returns without raising + ReverbProvider(_RaisingFastapiApp())._mount_auth_endpoint() def test_load_channels_file_skips_when_absent(self): - provider = ReverbProvider(_RaisingFastapiApp()) - provider._load_channels_file() # base_path has no routes/channels.py + ReverbProvider(_RaisingFastapiApp())._load_channels_file() From 55594c074ba0b1a583bd8e8a3028e57894ceb76d Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sun, 25 Oct 2026 13:11:00 -0700 Subject: [PATCH 2/5] Renumber broadcasting channel auth ADR to 036 ADR 035 is taken by the config subclass ADR in another open PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QgHs6PRFJehnb1BbCNtLww --- ...adcasting-channel-auth.md => 036-broadcasting-channel-auth.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename docs/adr/{035-broadcasting-channel-auth.md => 036-broadcasting-channel-auth.md} (100%) diff --git a/docs/adr/035-broadcasting-channel-auth.md b/docs/adr/036-broadcasting-channel-auth.md similarity index 100% rename from docs/adr/035-broadcasting-channel-auth.md rename to docs/adr/036-broadcasting-channel-auth.md From 9d6d0ef306c0592c058996218c9bee33a720278e Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sun, 25 Oct 2026 13:41:20 -0700 Subject: [PATCH 3/5] Harden broadcasting auth against signature confusion and malformed frames - Socket ids are Pusher digits.digits; /broadcasting/auth and verify reject anything else, and channel names must use the Pusher charset (no ':') - channel_data is signed and verified only for presence- channels - /broadcasting/auth returns 403 for public channels, anonymous users and presence users without an id; 400/422 for malformed bodies - REVERB_APP_SECRET has no default; the provider refuses to boot outside local/testing with an empty or "secret" value, and an empty secret rejects every protected subscription - Malformed or binary frames get pusher:error, oversized frames close with 1009, and connections are always cleaned up - ADR 036 updated with the threat model; index row fixed Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QgHs6PRFJehnb1BbCNtLww --- docs/adr/036-broadcasting-channel-auth.md | 45 +++++-- docs/adr/AGENTS.md | 2 +- .../fastapi_startkit/broadcasting/config.py | 2 +- .../fastapi_startkit/broadcasting/provider.py | 45 +++++-- .../broadcasting/reverb/channel_manager.py | 16 ++- .../broadcasting/reverb/connection.py | 13 +- .../broadcasting/reverb/server.py | 123 ++++++++++++------ .../broadcasting/reverb/signature.py | 26 +++- .../tests/broadcasting/conftest.py | 9 ++ .../broadcasting/test_private_channel_auth.py | 107 ++++++++++++++- .../tests/broadcasting/test_provider.py | 59 ++++++++- .../tests/broadcasting/test_reverb_server.py | 72 ++++++++++ 12 files changed, 439 insertions(+), 80 deletions(-) create mode 100644 fastapi_startkit/tests/broadcasting/conftest.py diff --git a/docs/adr/036-broadcasting-channel-auth.md b/docs/adr/036-broadcasting-channel-auth.md index e094b76e..44c18136 100644 --- a/docs/adr/036-broadcasting-channel-auth.md +++ b/docs/adr/036-broadcasting-channel-auth.md @@ -1,4 +1,4 @@ -# 035. Signed private-channel subscriptions and a reachable /broadcasting/auth +# 036. Signed private-channel subscriptions and a reachable /broadcasting/auth Date: 2026-10-25 @@ -9,26 +9,55 @@ Two defects in `fastapi_startkit.broadcasting`: - **F4, auth bypass.** `ReverbServer` subscribed every socket to any channel named in `pusher:subscribe`, including `private-*` and `presence-*`. The `auth` field was never checked, so any WebSocket client received private broadcasts without calling `/broadcasting/auth`. The endpoint also returned `{"auth": true}` rather than a Pusher signature, so a real Echo client could not have been verified even with a check in place. - **F3, 404.** `ReverbProvider` mounted the WebSocket app with `fastapi_app.mount("", ...)` before adding `POST /broadcasting/auth`. Starlette tries routes in order and an empty-path `Mount` matches every path, so the auth route, and every route registered after the provider boots, returned the inner app's 404. The endpoint also used a function-local `Request` import under `from __future__ import annotations`, so FastAPI could not resolve the annotation and would have treated `request` as a query parameter. +The security review of the first fix found more problems: + +- **Signature confusion (critical).** The signed string is `socket_id:channel[:channel_data]`, joined with `:`. The endpoint accepted any `socket_id`, signed public channels for anonymous users, and the server verified client `channel_data` for `private-` channels as well. An anonymous client with socket id `R` could request a signature for channel `news` with `socket_id=R:private-orders.1`, which is the HMAC of `R:private-orders.1:news`, then subscribe to `private-orders.1` with `channel_data: "news"`. `presence-` channels fell the same way. +- **Default secret (high).** `REVERB_APP_SECRET` defaulted to `"secret"`, so anyone could mint signatures for an unconfigured deployment. +- **Malformed frames (medium).** A non-JSON frame or a non-object `data` raised inside the handler. Only `WebSocketDisconnect` triggered cleanup, so each crashed socket stayed in `ConnectionManager` and `ChannelManager` forever. The frame size limit was the ASGI server's default (16 MB in uvicorn). +- **Low.** A JSON body that is not an object returned 500. Presence auth signed `user_id: ""` when the callback allowed a `None` user. + +## Threat model + +The attacker is any network client. It can open WebSockets, call `/broadcasting/auth` with or without a session, and send arbitrary frames. It does not know `REVERB_APP_SECRET`. Two things must hold. A socket can join a `private-`/`presence-` channel only with a signature that the server issued for that socket, that channel and, for presence, that `channel_data`. Malformed traffic must not crash the handler or grow server memory. + ## Alternatives 1. Track authorised `(socket_id, channel)` pairs in server memory when `/broadcasting/auth` succeeds and check them on subscribe. This is not the Pusher protocol, so it breaks Echo, and it fails with more than one worker. 2. Use the Pusher protocol: `/broadcasting/auth` returns `key:HMAC-SHA256(secret, "socket_id:channel[:channel_data]")` and the server recomputes the signature on subscribe. This is stateless, works across workers and matches what Laravel Echo sends. +For the confusion, a length-prefixed or JSON-encoded signing string would remove the ambiguity, but pusher-js and Laravel's broadcasters compute the colon-joined form, so changing it breaks compatibility. Instead, constrain each part so that `:` can only be a separator. + For F3, keep the mount and register it last, or register plain routes on the FastAPI router. Ordering cannot be guaranteed, because application routes can be added after any provider boots. ## Decision -Option 2 for F4, and plain routes for F3. +Option 2 for F4, plain routes for F3, and the Pusher wire format kept as is with every signed part validated. ## Implementation -- `broadcasting/reverb/signature.py`: `requires_auth(channel)`, `sign(key, secret, socket_id, channel, channel_data=None)` and `verify(...)`, which compares with `hmac.compare_digest` and rejects non-string `auth` or `channel_data`. -- `ReverbServer(key, secret)`: on a `private-`/`presence-` subscribe with a missing or invalid signature it sends `pusher:subscription_error` (`status: 401`) and does not subscribe the socket. Public channels are unchanged. -- `ReverbProvider.register` builds the server from `broadcasting.connections.reverb.key`/`secret`. +- `broadcasting/reverb/signature.py`: + - `valid_socket_id` accepts only ASCII `^[0-9]+\.[0-9]+$`, Pusher's format. + - `valid_channel` accepts only Pusher channel names: 1 to 200 characters from `A-Za-z0-9_-=@,.;`. A `:` is never allowed. + - `verify` returns false when the secret is empty, when the socket id or channel is invalid, or when the channel is not `private-`/`presence-`. It includes `channel_data` only for `presence-` channels, where it is required, and ignores it for `private-`. It compares with `hmac.compare_digest`. +- `ConnectionManager` issues socket ids as `.` from `secrets.randbelow`. The server always verifies against the subscribing connection's own id. +- `ReverbServer`: + - On a `private-`/`presence-` subscribe with a missing or invalid signature it sends `pusher:subscription_error` (`AuthError`, 401). An invalid channel name gets `InvalidChannel`, 400. In both cases the socket is not subscribed. + - Frames that are binary, not JSON, not an object, or that have a non-string `event` or a malformed subscribe `data` get `pusher:error` (code 4200), and the connection stays open. + - A text frame larger than `max_message_size` (10 000 characters, Pusher's message limit) gets `pusher:error` and the connection is closed with 1009. Operators should also set the ASGI server's limit (`uvicorn --ws-max-size`) so oversized frames are rejected before they are buffered. + - The connection and its subscriptions are removed in a `finally`, whatever ends the loop. `ChannelManager` deletes channels with no subscribers. +- `/broadcasting/auth`: + - Malformed JSON returns 400. A body that is not an object returns 422. An invalid `socket_id` or `channel_name` returns 422. + - Returns 403 when the channel is not `private-`/`presence-` (as Laravel does), when there is no authenticated user, when the server has no secret, or when the channel callback denies. + - Presence returns 403 if the user has no `id`, and otherwise signs `channel_data` `{"user_id": ""}`. +- Secret: `REVERB_APP_SECRET` has no default. `ReverbProvider.register` raises `RuntimeError` when the secret is empty or `"secret"` and `APP_ENV` is not `local` or `testing`. In local and testing an empty secret boots, but every `private-`/`presence-` subscription and auth request is rejected. - `ReverbProvider` registers `/app/{key}` with `router.add_websocket_route` and `/broadcasting/auth` with `router.add_route`, a Starlette endpoint, so neither depends on route order or annotation resolution. -- `/broadcasting/auth` requires string `channel_name` and `socket_id` (422 otherwise) and returns 403 when the channel callback denies. On success it returns `{"auth": signature}`. Presence channels also return `channel_data` (`{"user_id": ""}`), which is included in the signature. + +The endpoint accepts form posts. A cross-site page can send one, but it cannot read the signature unless CORS is permissive, so no CSRF token is required. `pusher:unsubscribe` and `client-*` events remain unimplemented (unchanged); `client-*` events are dropped, so nothing leaks. ## Validation -- Tests drive the fully booted app through `TestClient`: the auth route and routes added after boot are reachable whatever the provider order. A private subscribe without a signature is rejected and receives no broadcasts. Forged, wrong-socket, wrong-secret and non-string signatures are rejected. A signature obtained from `/broadcasting/auth` subscribes and receives broadcasts. Tampered presence `channel_data` is rejected. -- Full suite: `uv run pytest --ignore=tests/masoniteorm/postgres`. +- Tests drive the fully booted app through `TestClient`. The auth route and routes added after boot are reachable whatever the provider order. A private subscribe without a signature is rejected and receives no broadcasts. Forged, wrong-socket, wrong-secret and non-string signatures are rejected, as is tampered presence `channel_data`. +- Regression tests replay the reviewer's attack against `private-orders.1` and `presence-room.vip`. Before this change they ended in `subscription_succeeded`. Further tests cover a self-signed private signature over `socket:channel:data`, malformed socket ids and channel names on both auth and subscribe, 403 for public channels and anonymous users, and 400/422 bodies. +- Secret tests cover the empty default, refusing to boot in `production` with an unset, empty or `"secret"` value, booting in local and testing, and rejecting every protected subscription when the secret is empty. +- Server tests cover each malformed frame shape (error reply, connection kept), oversized frames (close 1009), and cleanup after malformed frames and after a handler exception. +- Full suite: `uv run pytest --ignore=tests/masoniteorm/postgres --cov`. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index 53d84e32..254af36f 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -35,4 +35,4 @@ Read this index first, then open the records relevant to the change. Before impl | [029](029-cache-module.md) | 2026-10-21 | Cache module with remember, memo, locks and tags | New `fastapi_startkit.cache`: async `Cache.get/put/has/forget/flush`, `get` with a callable default, and `remember(key, ttl, factory)` that runs a sync or async factory only on a miss; `remember(key, ttl)` without a factory is a decorator keyed by `key` plus a hash of the call arguments. Stores implement an async contract with a `MISSING` sentinel (cached `None` is a hit): array, file (pickled per-key files), database (masoniteorm table) and redis (named connection from the existing `RedisManager`, `SET PX` TTL, flush by `SCAN` of prefixed keys, never `FLUSHDB`). TTL in seconds or `timedelta`, `None` forever, <=0 forgets. `CacheConfig` via `resolve_config`; `CacheProvider` binds `cache`; facade `remember` resolves lazily. `Cache.memo()` memoizes reads per asyncio task through a `ContextVar` owned by the task, so requests never share memo state. `Cache.lock(name, seconds, owner)` gives atomic, owner-checked, non-re-entrant locks with `get`, `block` (250 ms polling, `LockTimeoutError`), `release`, `force_release` and `async with`, per store: event-loop dict, `fcntl.flock` file, `lock:` row with update-if-expired, Redis `SET NX PX` plus a Lua owner-checked delete. `Cache.tags([...])` uses Laravel's namespaced keys: random tag ids at `:tag::key`, entries at `:sha1(ids):`, flush rewrites the ids; tags are sorted and deduplicated. Shipped in slices into `feature/caches`; the former memo, lock and tags ADRs (030, 031, 034) are folded in. | | [032](032-migrate-seed-option.md) | 2026-10-21 | db:migrate:fresh and db:migrate:refresh --seed run seeders in the same event loop | `--seed` called the unregistered `seed:run` through Cleo `call` (which also drops the first argument) inside a running event loop, where `db:seed` would call `asyncio.run` again. The seeder selection moves into async `DBSeedCommand.run_seeders`, which `db:seed`, `db:migrate:fresh` and `db:migrate:refresh` all await. `db:migrate:refresh` is now registered. | | [033](033-async-seeder-stub.md) | 2026-10-21 | Generated seeders have an async run | `create_seed.stub` generated a synchronous `run` with docstrings, but `Seeder` awaits `run()`, so every generated table seeder raised `TypeError`. The stub now generates `async def run(self) -> None` with no docstrings; `Seeder` is unchanged. | -| [035](035-broadcasting-channel-auth.md) | 2026-10-25 | Signed private-channel subscriptions and a reachable /broadcasting/auth | `ReverbServer` subscribed any socket to `private-`/`presence-` channels without checking `auth`, and the empty-path WebSocket `Mount` shadowed `/broadcasting/auth` and later routes (404). The auth endpoint now requires `socket_id` and returns a Pusher signature `key:HMAC-SHA256(secret, socket_id:channel[:channel_data])` (presence adds signed `channel_data` with `user_id`). The server verifies it with `compare_digest` and replies `pusher:subscription_error` 401 on failure. Both endpoints are registered as plain router routes, so route order does not matter. | +| [036](036-broadcasting-channel-auth.md) | 2026-10-25 | Signed private-channel subscriptions and a reachable /broadcasting/auth | `ReverbServer` subscribed any socket to `private-`/`presence-` channels without checking `auth`, and the empty-path WebSocket `Mount` shadowed `/broadcasting/auth` and later routes (404). The endpoint returns a Pusher signature `key:HMAC-SHA256(secret, socket_id:channel[:channel_data])` and the server verifies it with `compare_digest` against the connection's own socket id (`pusher:subscription_error` 401 on failure). The colon-joined wire format is kept for Echo compatibility, so ambiguity is closed by validation: socket ids are Pusher `digits.digits`, channel names are Pusher charset with no `:`, and `channel_data` is signed only for `presence-`. The endpoint returns 403 for public channels, anonymous users and presence users without an id, 400/422 for malformed bodies. `REVERB_APP_SECRET` has no default; the provider refuses to boot outside local/testing with an empty or `"secret"` value, and an empty secret rejects every protected subscription. Malformed frames get `pusher:error`, frames over 10 000 chars close with 1009, and connections are always cleaned up in `finally`. Both endpoints are plain router routes, so route order does not matter. | diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/config.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/config.py index d91fc094..6667de55 100644 --- a/fastapi_startkit/src/fastapi_startkit/broadcasting/config.py +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/config.py @@ -8,7 +8,7 @@ class ReverbConnectionConfig(Config): driver: str = "reverb" key: str = env("REVERB_APP_KEY", "local") - secret: str = env("REVERB_APP_SECRET", "secret") + secret: str = env("REVERB_APP_SECRET", "") app_id: str = env("REVERB_APP_ID", "1") host: str = env("REVERB_HOST", "0.0.0.0") port: int = env("REVERB_PORT", 8080) diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py index 5507b476..ecc58326 100644 --- a/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py @@ -4,13 +4,17 @@ import importlib.util import json import sys +from collections.abc import Mapping from pathlib import Path from fastapi_startkit.support import Provider from .config import BroadcastingConfig from .manager import BroadcastManager from .reverb.server import ReverbServer -from .reverb.signature import sign +from .reverb.signature import is_presence, requires_auth, sign, valid_channel, valid_socket_id + +UNSAFE_SECRETS = ("", "secret") +INSECURE_ENVIRONMENTS = ("local", "testing") class ReverbProvider(Provider): @@ -39,7 +43,13 @@ def register(self) -> None: self.merge_config_from(config_data, "broadcasting") reverb = config_data.get("connections", {}).get("reverb", {}) - server = ReverbServer(key=reverb.get("key", "local"), secret=reverb.get("secret", "secret")) + secret = reverb.get("secret") or "" + if secret in UNSAFE_SECRETS and self.app.env not in INSECURE_ENVIRONMENTS: + raise RuntimeError( + f"REVERB_APP_SECRET must be set to a private value when APP_ENV is {self.app.env!r}; " + "private and presence channel signatures would otherwise be forgeable." + ) + server = ReverbServer(key=reverb.get("key", "local"), secret=secret) manager = BroadcastManager(config_data, server) self.app.bind("broadcasting", manager) @@ -108,30 +118,39 @@ def _mount_auth_endpoint(self) -> None: application = self.app async def broadcasting_auth(request: Request): - content_type = request.headers.get("content-type", "") - if "application/json" in content_type: - body = await request.json() + if "application/json" in request.headers.get("content-type", ""): + try: + body = await request.json() + except ValueError: + return JSONResponse({"error": "Malformed JSON body"}, status_code=400) else: body = await request.form() + if not isinstance(body, Mapping): + return JSONResponse({"error": "Request body must be an object"}, status_code=422) + channel_name = body.get("channel_name", "") socket_id = body.get("socket_id", "") + if not valid_socket_id(socket_id): + return JSONResponse({"error": "socket_id is invalid"}, status_code=422) + if not valid_channel(channel_name): + return JSONResponse({"error": "channel_name is invalid"}, status_code=422) - if not isinstance(channel_name, str) or not isinstance(socket_id, str): - return JSONResponse({"error": "channel_name and socket_id must be strings"}, status_code=422) - if not socket_id: - return JSONResponse({"error": "socket_id is required"}, status_code=422) - + server: ReverbServer = application.make("reverb.server") user = self._resolve_user(application, request) + if not server.secret or not requires_auth(channel_name) or user is None: + return JSONResponse({"error": "Forbidden"}, status_code=403) manager: BroadcastManager = application.make("broadcasting") if not await manager.channel_registry.authorize(channel_name, user): return JSONResponse({"error": "Forbidden"}, status_code=403) - server: ReverbServer = application.make("reverb.server") - if not channel_name.startswith("presence-"): + if not is_presence(channel_name): return JSONResponse({"auth": sign(server.key, server.secret, socket_id, channel_name)}) - channel_data = json.dumps({"user_id": str(getattr(user, "id", ""))}) + user_id = getattr(user, "id", None) + if user_id is None: + return JSONResponse({"error": "Forbidden"}, status_code=403) + channel_data = json.dumps({"user_id": str(user_id)}) auth = sign(server.key, server.secret, socket_id, channel_name, channel_data) return JSONResponse({"auth": auth, "channel_data": channel_data}) diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/channel_manager.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/channel_manager.py index d92f9446..611167c1 100644 --- a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/channel_manager.py +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/channel_manager.py @@ -1,17 +1,21 @@ class ChannelManager: def __init__(self): - self._subscriptions: dict[str, set] = {} + self._subscriptions: dict[str, set[str]] = {} def subscribe(self, channel: str, socket_id: str) -> None: self._subscriptions.setdefault(channel, set()).add(socket_id) def unsubscribe(self, channel: str, socket_id: str) -> None: - if channel in self._subscriptions: - self._subscriptions[channel].discard(socket_id) + subscribers = self._subscriptions.get(channel) + if subscribers is None: + return + subscribers.discard(socket_id) + if not subscribers: + del self._subscriptions[channel] def remove_socket(self, socket_id: str) -> None: - for subs in self._subscriptions.values(): - subs.discard(socket_id) + for channel in list(self._subscriptions): + self.unsubscribe(channel, socket_id) - def subscribers(self, channel: str) -> set: + def subscribers(self, channel: str) -> set[str]: return set(self._subscriptions.get(channel, set())) diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/connection.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/connection.py index 945a8ea2..b25da052 100644 --- a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/connection.py +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/connection.py @@ -1,4 +1,5 @@ -import uuid +import secrets + from starlette.websockets import WebSocket @@ -7,15 +8,21 @@ def __init__(self): self._connections: dict[str, WebSocket] = {} def add(self, websocket: WebSocket) -> str: - socket_id = str(uuid.uuid4()) + socket_id = self._new_socket_id() + while socket_id in self._connections: + socket_id = self._new_socket_id() self._connections[socket_id] = websocket return socket_id + @staticmethod + def _new_socket_id() -> str: + return f"{secrets.randbelow(10**10)}.{secrets.randbelow(10**10)}" + def remove(self, socket_id: str) -> None: self._connections.pop(socket_id, None) def get(self, socket_id: str): return self._connections.get(socket_id) - def all(self) -> dict: + def all(self) -> dict[str, WebSocket]: return dict(self._connections) diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py index 50cd5fce..70e20f30 100644 --- a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py @@ -1,14 +1,20 @@ import json -from starlette.websockets import WebSocket, WebSocketDisconnect -from starlette.routing import WebSocketRoute + from starlette.applications import Starlette -from .connection import ConnectionManager +from starlette.routing import WebSocketRoute +from starlette.websockets import WebSocket, WebSocketDisconnect + from .channel_manager import ChannelManager -from .signature import requires_auth, verify +from .connection import ConnectionManager +from .signature import requires_auth, valid_channel, verify + +MESSAGE_TOO_BIG = 1009 class ReverbServer: - def __init__(self, key: str = "local", secret: str = "secret"): + max_message_size = 10_000 + + def __init__(self, key: str = "local", secret: str = ""): self.key = key self.secret = secret self.connections = ConnectionManager() @@ -17,51 +23,86 @@ def __init__(self, key: str = "local", secret: str = "secret"): async def handle(self, websocket: WebSocket): await websocket.accept() socket_id = self.connections.add(websocket) - await websocket.send_text( - json.dumps( - { - "event": "pusher:connection_established", - "data": json.dumps({"socket_id": socket_id, "activity_timeout": 120}), - } - ) - ) try: - while True: - data = await websocket.receive_text() - message = json.loads(data) - await self._handle_message(socket_id, websocket, message) + await self._send( + websocket, + "pusher:connection_established", + json.dumps({"socket_id": socket_id, "activity_timeout": 120}), + ) + await self._receive_loop(socket_id, websocket) except WebSocketDisconnect: + pass + finally: self.connections.remove(socket_id) self.channels.remove_socket(socket_id) - async def _handle_message(self, socket_id: str, websocket: WebSocket, message: dict): + async def _receive_loop(self, socket_id: str, websocket: WebSocket): + while True: + frame = await websocket.receive() + if frame["type"] == "websocket.disconnect": + return + text = frame.get("text") + if text is not None and len(text) > self.max_message_size: + await self._error(websocket, "Maximum message size exceeded") + await websocket.close(code=MESSAGE_TOO_BIG) + return + message = self._parse(text) + if message is None: + await self._error(websocket, "Invalid message format") + continue + await self._handle_message(socket_id, websocket, message) + + @staticmethod + def _parse(text: str | None) -> dict[str, object] | None: + if text is None: + return None + try: + message = json.loads(text) + except ValueError: + return None + if not isinstance(message, dict) or not isinstance(message.get("event", ""), str): + return None + return message + + async def _handle_message(self, socket_id: str, websocket: WebSocket, message: dict[str, object]): event = message.get("event", "") if event == "pusher:ping": - await websocket.send_text(json.dumps({"event": "pusher:pong", "data": {}})) + await self._send(websocket, "pusher:pong", {}) elif event == "pusher:subscribe": - data = message.get("data") or {} - channel = data.get("channel", "") - if not isinstance(channel, str) or not channel: - return - if requires_auth(channel) and not verify( - self.key, self.secret, socket_id, channel, data.get("auth"), data.get("channel_data") - ): - await websocket.send_text( - json.dumps( - { - "event": "pusher:subscription_error", - "channel": channel, - "data": {"type": "AuthError", "error": "Invalid signature", "status": 401}, - } - ) - ) - return - self.channels.subscribe(channel, socket_id) - await websocket.send_text( - json.dumps({"event": "pusher_internal:subscription_succeeded", "channel": channel, "data": {}}) - ) + await self._subscribe(socket_id, websocket, message.get("data")) + + async def _subscribe(self, socket_id: str, websocket: WebSocket, data: object): + if not isinstance(data, dict) or not isinstance(data.get("channel"), str) or not data["channel"]: + await self._error(websocket, "Invalid subscription") + return + channel = data["channel"] + if not valid_channel(channel): + await self._subscription_error(websocket, channel, "InvalidChannel", "Invalid channel name", 400) + return + if requires_auth(channel) and not verify( + self.key, self.secret, socket_id, channel, data.get("auth"), data.get("channel_data") + ): + await self._subscription_error(websocket, channel, "AuthError", "Invalid signature", 401) + return + self.channels.subscribe(channel, socket_id) + await self._send(websocket, "pusher_internal:subscription_succeeded", {}, channel) + + async def _subscription_error(self, websocket: WebSocket, channel: str, kind: str, error: str, status: int): + await self._send( + websocket, "pusher:subscription_error", {"type": kind, "error": error, "status": status}, channel + ) + + async def _error(self, websocket: WebSocket, message: str): + await self._send(websocket, "pusher:error", {"code": 4200, "message": message}) + + @staticmethod + async def _send(websocket: WebSocket, event: str, data: object, channel: str | None = None): + payload: dict[str, object] = {"event": event, "data": data} + if channel is not None: + payload["channel"] = channel + await websocket.send_text(json.dumps(payload)) - async def broadcast_to_channel(self, channel_name: str, event_name: str, data: dict): + async def broadcast_to_channel(self, channel_name: str, event_name: str, data: dict[str, object]): payload = json.dumps({"event": event_name, "channel": channel_name, "data": data}) for socket_id in self.channels.subscribers(channel_name): ws = self.connections.get(socket_id) diff --git a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py index 3084ecad..dcf71ef9 100644 --- a/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py +++ b/fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/signature.py @@ -1,11 +1,29 @@ import hashlib import hmac +import re + +from typing_extensions import TypeIs + +SOCKET_ID = re.compile(r"[0-9]+\.[0-9]+", re.ASCII) +CHANNEL_NAME = re.compile(r"[A-Za-z0-9_\-=@,.;]{1,200}", re.ASCII) + + +def valid_socket_id(socket_id: object) -> TypeIs[str]: + return isinstance(socket_id, str) and SOCKET_ID.fullmatch(socket_id) is not None + + +def valid_channel(channel: object) -> TypeIs[str]: + return isinstance(channel, str) and CHANNEL_NAME.fullmatch(channel) is not None def requires_auth(channel: object) -> bool: return isinstance(channel, str) and channel.startswith(("private-", "presence-")) +def is_presence(channel: str) -> bool: + return channel.startswith("presence-") + + def sign(key: str, secret: str, socket_id: str, channel: str, channel_data: str | None = None) -> str: parts = [socket_id, channel] if channel_data is None else [socket_id, channel, channel_data] digest = hmac.new(secret.encode(), ":".join(parts).encode(), hashlib.sha256).hexdigest() @@ -13,7 +31,13 @@ def sign(key: str, secret: str, socket_id: str, channel: str, channel_data: str def verify(key: str, secret: str, socket_id: str, channel: str, auth: object, channel_data: object = None) -> bool: - if not isinstance(auth, str) or not (channel_data is None or isinstance(channel_data, str)): + if not secret or not isinstance(auth, str): + return False + if not valid_socket_id(socket_id) or not valid_channel(channel) or not requires_auth(channel): + return False + if not is_presence(channel): + channel_data = None + elif not isinstance(channel_data, str): return False expected = sign(key, secret, socket_id, channel, channel_data) return hmac.compare_digest(auth.encode(), expected.encode()) diff --git a/fastapi_startkit/tests/broadcasting/conftest.py b/fastapi_startkit/tests/broadcasting/conftest.py new file mode 100644 index 00000000..fb1da947 --- /dev/null +++ b/fastapi_startkit/tests/broadcasting/conftest.py @@ -0,0 +1,9 @@ +import pytest + +SECRET = "test-reverb-secret" + + +@pytest.fixture(autouse=True) +def reverb_secret(monkeypatch): + monkeypatch.setenv("REVERB_APP_SECRET", SECRET) + return SECRET diff --git a/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py b/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py index 47fb8cba..13833b94 100644 --- a/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py +++ b/fastapi_startkit/tests/broadcasting/test_private_channel_auth.py @@ -1,11 +1,13 @@ import asyncio import json +import re import pytest from starlette.testclient import TestClient from fastapi_startkit.application import Application from fastapi_startkit.broadcasting.provider import ReverbProvider +from fastapi_startkit.broadcasting.reverb.signature import sign from fastapi_startkit.support import Provider KEY = "local" @@ -120,7 +122,7 @@ def test_forged_signature_is_rejected(self, base_path): def test_signature_for_another_socket_is_rejected(self, base_path): _, client = make_client(base_path, [ReverbProvider], user=User(7)) auth = client.post( - "/broadcasting/auth", data={"socket_id": "someone-else", "channel_name": "private-orders.7"} + "/broadcasting/auth", data={"socket_id": "999.999", "channel_name": "private-orders.7"} ).json()["auth"] with client.websocket_connect(f"/app/{KEY}") as ws: connect(ws) @@ -161,3 +163,106 @@ def test_presence_requires_signed_channel_data(self, base_path): assert subscribe(ws, "presence-room.1", body["auth"], tampered)["event"] == "pusher:subscription_error" reply = subscribe(ws, "presence-room.1", body["auth"], body["channel_data"]) assert reply["event"] == "pusher_internal:subscription_succeeded" + + def test_presence_requires_channel_data(self, base_path): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + with client.websocket_connect(f"/app/{KEY}") as ws: + socket_id = connect(ws) + auth = client.post( + "/broadcasting/auth", data={"socket_id": socket_id, "channel_name": "presence-room.1"} + ).json()["auth"] + assert subscribe(ws, "presence-room.1", auth=auth)["event"] == "pusher:subscription_error" + + +class TestSignatureConfusion: + @pytest.mark.parametrize("target", ["private-orders.1", "presence-room.vip"]) + def test_anonymous_client_cannot_reuse_public_signature_for_protected_channel(self, base_path, target): + app, client = make_client(base_path, [ReverbProvider]) + server = app.make("reverb.server") + with client.websocket_connect(f"/app/{KEY}") as ws: + socket_id = connect(ws) + response = client.post( + "/broadcasting/auth", data={"channel_name": "news", "socket_id": f"{socket_id}:{target}"} + ) + auth = response.json().get("auth", f"{KEY}:{'0' * 64}") + + assert subscribe(ws, target, auth=auth, channel_data="news")["event"] == "pusher:subscription_error" + assert socket_id not in server.channels.subscribers(target) + assert response.status_code in (403, 422) + + def test_private_verification_ignores_client_channel_data(self, base_path, reverb_secret): + _, client = make_client(base_path, [ReverbProvider]) + with client.websocket_connect(f"/app/{KEY}") as ws: + socket_id = connect(ws) + auth = sign(KEY, reverb_secret, socket_id, "private-orders.1", "news") + reply = subscribe(ws, "private-orders.1", auth=auth, channel_data="news") + assert reply["event"] == "pusher:subscription_error" + + def test_private_signature_accepts_and_ignores_extra_channel_data(self, base_path): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + with client.websocket_connect(f"/app/{KEY}") as ws: + socket_id = connect(ws) + body = client.post( + "/broadcasting/auth", data={"socket_id": socket_id, "channel_name": "private-orders.7"} + ).json() + assert "channel_data" not in body + reply = subscribe(ws, "private-orders.7", auth=body["auth"], channel_data="anything") + assert reply["event"] == "pusher_internal:subscription_succeeded" + + def test_socket_ids_use_pusher_format(self, base_path): + _, client = make_client(base_path, [ReverbProvider]) + with client.websocket_connect(f"/app/{KEY}") as ws: + assert re.fullmatch(r"\d+\.\d+", connect(ws)) + + @pytest.mark.parametrize("socket_id", ["1.1:private-orders.7", "abc", "1", "1.1\n", "1.1.1", " 1.1", "١.١"]) + def test_auth_rejects_malformed_socket_id(self, base_path, socket_id): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + response = client.post("/broadcasting/auth", data={"socket_id": socket_id, "channel_name": "private-orders.7"}) + assert response.status_code == 422 + + @pytest.mark.parametrize("channel", ["private-orders.7:x", "private-orders 7", "private-" + "a" * 200, ""]) + def test_auth_rejects_invalid_channel_name(self, base_path, channel): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": channel}) + assert response.status_code == 422 + + @pytest.mark.parametrize("channel", ["news:private-orders.1", "private-orders.1:news", "a b"]) + def test_subscribe_rejects_invalid_channel_name(self, base_path, channel): + app, client = make_client(base_path, [ReverbProvider]) + server = app.make("reverb.server") + with client.websocket_connect(f"/app/{KEY}") as ws: + socket_id = connect(ws) + reply = subscribe(ws, channel) + assert reply["event"] == "pusher:subscription_error" + assert reply["data"]["status"] == 400 + assert socket_id not in server.channels.subscribers(channel) + + def test_auth_refuses_public_channels(self, base_path): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": "news"}) + assert response.status_code == 403 + + @pytest.mark.parametrize("channel", ["private-open", "presence-open"]) + def test_auth_requires_user_even_when_callback_allows(self, base_path, channel): + app, client = make_client(base_path, [ReverbProvider]) + app.make("broadcasting").channel_registry.register(channel, lambda user: True) + response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": channel}) + assert response.status_code == 403 + + def test_presence_refuses_user_without_id(self, base_path): + _, client = make_client(base_path, [ReverbProvider], user=User(None)) + response = client.post("/broadcasting/auth", data={"socket_id": "1.1", "channel_name": "presence-room.1"}) + assert response.status_code == 403 + + +class TestAuthRequestBody: + @pytest.mark.parametrize("body", [[1], "x", 1, None]) + def test_rejects_non_object_json_body(self, base_path, body): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + response = client.post("/broadcasting/auth", json=body) + assert response.status_code == 422 + + def test_rejects_invalid_json(self, base_path): + _, client = make_client(base_path, [ReverbProvider], user=User(7)) + response = client.post("/broadcasting/auth", content=b"{not json", headers={"content-type": "application/json"}) + assert response.status_code == 400 diff --git a/fastapi_startkit/tests/broadcasting/test_provider.py b/fastapi_startkit/tests/broadcasting/test_provider.py index 61971d44..d685f7fa 100644 --- a/fastapi_startkit/tests/broadcasting/test_provider.py +++ b/fastapi_startkit/tests/broadcasting/test_provider.py @@ -1,14 +1,60 @@ import json from types import SimpleNamespace +import pytest from starlette.testclient import TestClient from fastapi_startkit.application import Application +from fastapi_startkit.broadcasting.config import ReverbConnectionConfig from fastapi_startkit.broadcasting.manager import BroadcastManager from fastapi_startkit.broadcasting.provider import ReverbProvider from fastapi_startkit.broadcasting.reverb.server import ReverbServer from fastapi_startkit.broadcasting.reverb.signature import sign +from .conftest import SECRET + + +class TestSecret: + def test_has_no_default_secret(self, monkeypatch): + monkeypatch.delenv("REVERB_APP_SECRET") + assert ReverbConnectionConfig().secret == "" + + @pytest.mark.parametrize("secret", [None, "", "secret"]) + def test_refuses_to_boot_outside_local_and_testing_without_real_secret(self, monkeypatch, tmp_path, secret): + monkeypatch.delenv("PYTEST_CURRENT_TEST") + if secret is None: + monkeypatch.delenv("REVERB_APP_SECRET") + else: + monkeypatch.setenv("REVERB_APP_SECRET", secret) + with pytest.raises(RuntimeError, match="REVERB_APP_SECRET"): + Application(base_path=tmp_path, env="production", providers=[ReverbProvider]) + + def test_boots_in_production_with_real_secret(self, monkeypatch, tmp_path): + monkeypatch.delenv("PYTEST_CURRENT_TEST") + app = Application(base_path=tmp_path, env="production", providers=[ReverbProvider]) + assert app.make("reverb.server").secret == SECRET + + @pytest.mark.parametrize("env", ["local", "testing"]) + def test_boots_in_local_and_testing_without_secret(self, monkeypatch, tmp_path, env): + monkeypatch.delenv("PYTEST_CURRENT_TEST") + monkeypatch.delenv("REVERB_APP_SECRET") + app = Application(base_path=tmp_path, env=env, providers=[ReverbProvider]) + assert app.make("reverb.server").secret == "" + + def test_empty_secret_rejects_every_protected_subscription(self, monkeypatch, tmp_path): + monkeypatch.delenv("REVERB_APP_SECRET") + app = make_app(tmp_path) + register_rule(app, "private-room.{room}", lambda user, room: True) + app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=1))) + client = client_for(app) + with client.websocket_connect("/app/local") as ws: + socket_id = json.loads(json.loads(ws.receive_text())["data"])["socket_id"] + assert post_auth(app, channel_name="private-room.7", socket_id=socket_id).status_code == 403 + + auth = sign("local", "", socket_id, "private-room.7") + ws.send_text(json.dumps({"event": "pusher:subscribe", "data": {"channel": "private-room.7", "auth": auth}})) + assert json.loads(ws.receive_text())["event"] == "pusher:subscription_error" + def make_app(tmp_path): return Application(base_path=tmp_path, env="testing", providers=[ReverbProvider]) @@ -32,9 +78,9 @@ def test_binds_manager_and_server(self, tmp_path): assert isinstance(app.make("broadcasting"), BroadcastManager) assert isinstance(app.make("reverb.server"), ReverbServer) - def test_server_uses_configured_credentials(self, tmp_path): + def test_server_uses_configured_credentials(self, tmp_path, reverb_secret): server = make_app(tmp_path).make("reverb.server") - assert (server.key, server.secret) == ("local", "secret") + assert (server.key, server.secret) == ("local", reverb_secret) def test_publishes_channels_stub(self, tmp_path): app = make_app(tmp_path) @@ -55,17 +101,19 @@ def test_routes_added_after_boot_are_not_shadowed(self, tmp_path): def test_private_channel_returns_signed_auth(self, tmp_path): app = make_app(tmp_path) register_rule(app, "private-room.{room}", lambda user, room: True) + app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=1))) response = post_auth(app, channel_name="private-room.7", socket_id="123.456") assert response.status_code == 200 - assert response.json() == {"auth": sign("local", "secret", "123.456", "private-room.7")} + assert response.json() == {"auth": sign("local", SECRET, "123.456", "private-room.7")} def test_accepts_json_body(self, tmp_path): app = make_app(tmp_path) register_rule(app, "private-room.{room}", lambda user, room: True) + app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=1))) response = client_for(app).post( "/broadcasting/auth", json={"channel_name": "private-room.7", "socket_id": "123.456"} ) - assert response.json()["auth"] == sign("local", "secret", "123.456", "private-room.7") + assert response.json()["auth"] == sign("local", SECRET, "123.456", "private-room.7") def test_presence_channel_signs_channel_data(self, tmp_path): app = make_app(tmp_path) @@ -73,7 +121,7 @@ def test_presence_channel_signs_channel_data(self, tmp_path): app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=9))) body = post_auth(app, channel_name="presence-room", socket_id="1.2").json() assert body["channel_data"] == '{"user_id": "9"}' - assert body["auth"] == sign("local", "secret", "1.2", "presence-room", body["channel_data"]) + assert body["auth"] == sign("local", SECRET, "1.2", "presence-room", body["channel_data"]) def test_denied_callback_returns_403(self, tmp_path): app = make_app(tmp_path) @@ -123,6 +171,7 @@ class TestWebsocketSubscription: def test_private_subscribe_requires_signature_from_auth_endpoint(self, tmp_path): app = make_app(tmp_path) register_rule(app, "private-room.{room}", lambda user, room: True) + app.bind("auth", SimpleNamespace(user=SimpleNamespace(id=1))) client = client_for(app) with client.websocket_connect("/app/local") as ws: socket_id = json.loads(json.loads(ws.receive_text())["data"])["socket_id"] diff --git a/fastapi_startkit/tests/broadcasting/test_reverb_server.py b/fastapi_startkit/tests/broadcasting/test_reverb_server.py index c0c95377..6d8f1b50 100644 --- a/fastapi_startkit/tests/broadcasting/test_reverb_server.py +++ b/fastapi_startkit/tests/broadcasting/test_reverb_server.py @@ -1,5 +1,6 @@ import json import pytest +from starlette.websockets import WebSocketDisconnect from starlette.testclient import TestClient from fastapi_startkit.broadcasting.reverb.server import ReverbServer @@ -71,3 +72,74 @@ def test_socket_id_is_unique(client): data = json.loads(msg["data"]) socket_ids.append(data["socket_id"]) assert len(set(socket_ids)) == 3 + + +def _connect(ws): + return json.loads(json.loads(ws.receive_text())["data"])["socket_id"] + + +def _ping(ws): + ws.send_text(json.dumps({"event": "pusher:ping", "data": {}})) + return json.loads(ws.receive_text())["event"] + + +@pytest.mark.parametrize( + "frame", + [ + "not json", + "[1, 2]", + "42", + json.dumps({"event": "pusher:subscribe", "data": "x"}), + json.dumps({"event": "pusher:subscribe", "data": {"channel": 1}}), + json.dumps({"event": 5}), + ], +) +def test_malformed_frame_replies_with_error_and_keeps_connection(client, frame): + with client.websocket_connect("/app/local") as ws: + _connect(ws) + ws.send_text(frame) + assert json.loads(ws.receive_text())["event"] == "pusher:error" + assert _ping(ws) == "pusher:pong" + + +def test_binary_frame_replies_with_error(client): + with client.websocket_connect("/app/local") as ws: + _connect(ws) + ws.send_bytes(b"\x00\x01") + assert json.loads(ws.receive_text())["event"] == "pusher:error" + assert _ping(ws) == "pusher:pong" + + +def test_oversized_frame_closes_connection(server, client): + with client.websocket_connect("/app/local") as ws: + _connect(ws) + ws.send_text("x" * (server.max_message_size + 1)) + assert json.loads(ws.receive_text())["event"] == "pusher:error" + with pytest.raises(WebSocketDisconnect) as closed: + ws.receive_text() + assert closed.value.code == 1009 + assert server.connections.all() == {} + + +def test_connection_is_cleaned_up_after_malformed_frames(server, client): + with client.websocket_connect("/app/local") as ws: + socket_id = _connect(ws) + ws.send_text(json.dumps({"event": "pusher:subscribe", "data": {"channel": "orders.1"}})) + ws.receive_text() + ws.send_text("not json") + ws.receive_text() + assert socket_id not in server.connections.all() + assert socket_id not in server.channels.subscribers("orders.1") + + +def test_connection_is_cleaned_up_when_handler_fails(server, client, monkeypatch): + async def explode(*args): + raise RuntimeError("boom") + + monkeypatch.setattr(server, "_handle_message", explode) + with pytest.raises(RuntimeError): + with client.websocket_connect("/app/local") as ws: + _connect(ws) + ws.send_text(json.dumps({"event": "pusher:ping"})) + ws.receive_text() + assert server.connections.all() == {} From dbf74950ed38e1d70867f3d1d4e77e01eecd5c07 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Mon, 26 Oct 2026 14:38:09 -0700 Subject: [PATCH 4/5] Drop ADR for broadcasting auth bug fix Co-Authored-By: Claude Sonnet 5.5 --- docs/adr/036-broadcasting-channel-auth.md | 63 ----------------------- docs/adr/AGENTS.md | 1 - 2 files changed, 64 deletions(-) delete mode 100644 docs/adr/036-broadcasting-channel-auth.md diff --git a/docs/adr/036-broadcasting-channel-auth.md b/docs/adr/036-broadcasting-channel-auth.md deleted file mode 100644 index 44c18136..00000000 --- a/docs/adr/036-broadcasting-channel-auth.md +++ /dev/null @@ -1,63 +0,0 @@ -# 036. Signed private-channel subscriptions and a reachable /broadcasting/auth - -Date: 2026-10-25 - -## Problem - -Two defects in `fastapi_startkit.broadcasting`: - -- **F4, auth bypass.** `ReverbServer` subscribed every socket to any channel named in `pusher:subscribe`, including `private-*` and `presence-*`. The `auth` field was never checked, so any WebSocket client received private broadcasts without calling `/broadcasting/auth`. The endpoint also returned `{"auth": true}` rather than a Pusher signature, so a real Echo client could not have been verified even with a check in place. -- **F3, 404.** `ReverbProvider` mounted the WebSocket app with `fastapi_app.mount("", ...)` before adding `POST /broadcasting/auth`. Starlette tries routes in order and an empty-path `Mount` matches every path, so the auth route, and every route registered after the provider boots, returned the inner app's 404. The endpoint also used a function-local `Request` import under `from __future__ import annotations`, so FastAPI could not resolve the annotation and would have treated `request` as a query parameter. - -The security review of the first fix found more problems: - -- **Signature confusion (critical).** The signed string is `socket_id:channel[:channel_data]`, joined with `:`. The endpoint accepted any `socket_id`, signed public channels for anonymous users, and the server verified client `channel_data` for `private-` channels as well. An anonymous client with socket id `R` could request a signature for channel `news` with `socket_id=R:private-orders.1`, which is the HMAC of `R:private-orders.1:news`, then subscribe to `private-orders.1` with `channel_data: "news"`. `presence-` channels fell the same way. -- **Default secret (high).** `REVERB_APP_SECRET` defaulted to `"secret"`, so anyone could mint signatures for an unconfigured deployment. -- **Malformed frames (medium).** A non-JSON frame or a non-object `data` raised inside the handler. Only `WebSocketDisconnect` triggered cleanup, so each crashed socket stayed in `ConnectionManager` and `ChannelManager` forever. The frame size limit was the ASGI server's default (16 MB in uvicorn). -- **Low.** A JSON body that is not an object returned 500. Presence auth signed `user_id: ""` when the callback allowed a `None` user. - -## Threat model - -The attacker is any network client. It can open WebSockets, call `/broadcasting/auth` with or without a session, and send arbitrary frames. It does not know `REVERB_APP_SECRET`. Two things must hold. A socket can join a `private-`/`presence-` channel only with a signature that the server issued for that socket, that channel and, for presence, that `channel_data`. Malformed traffic must not crash the handler or grow server memory. - -## Alternatives - -1. Track authorised `(socket_id, channel)` pairs in server memory when `/broadcasting/auth` succeeds and check them on subscribe. This is not the Pusher protocol, so it breaks Echo, and it fails with more than one worker. -2. Use the Pusher protocol: `/broadcasting/auth` returns `key:HMAC-SHA256(secret, "socket_id:channel[:channel_data]")` and the server recomputes the signature on subscribe. This is stateless, works across workers and matches what Laravel Echo sends. - -For the confusion, a length-prefixed or JSON-encoded signing string would remove the ambiguity, but pusher-js and Laravel's broadcasters compute the colon-joined form, so changing it breaks compatibility. Instead, constrain each part so that `:` can only be a separator. - -For F3, keep the mount and register it last, or register plain routes on the FastAPI router. Ordering cannot be guaranteed, because application routes can be added after any provider boots. - -## Decision - -Option 2 for F4, plain routes for F3, and the Pusher wire format kept as is with every signed part validated. - -## Implementation - -- `broadcasting/reverb/signature.py`: - - `valid_socket_id` accepts only ASCII `^[0-9]+\.[0-9]+$`, Pusher's format. - - `valid_channel` accepts only Pusher channel names: 1 to 200 characters from `A-Za-z0-9_-=@,.;`. A `:` is never allowed. - - `verify` returns false when the secret is empty, when the socket id or channel is invalid, or when the channel is not `private-`/`presence-`. It includes `channel_data` only for `presence-` channels, where it is required, and ignores it for `private-`. It compares with `hmac.compare_digest`. -- `ConnectionManager` issues socket ids as `.` from `secrets.randbelow`. The server always verifies against the subscribing connection's own id. -- `ReverbServer`: - - On a `private-`/`presence-` subscribe with a missing or invalid signature it sends `pusher:subscription_error` (`AuthError`, 401). An invalid channel name gets `InvalidChannel`, 400. In both cases the socket is not subscribed. - - Frames that are binary, not JSON, not an object, or that have a non-string `event` or a malformed subscribe `data` get `pusher:error` (code 4200), and the connection stays open. - - A text frame larger than `max_message_size` (10 000 characters, Pusher's message limit) gets `pusher:error` and the connection is closed with 1009. Operators should also set the ASGI server's limit (`uvicorn --ws-max-size`) so oversized frames are rejected before they are buffered. - - The connection and its subscriptions are removed in a `finally`, whatever ends the loop. `ChannelManager` deletes channels with no subscribers. -- `/broadcasting/auth`: - - Malformed JSON returns 400. A body that is not an object returns 422. An invalid `socket_id` or `channel_name` returns 422. - - Returns 403 when the channel is not `private-`/`presence-` (as Laravel does), when there is no authenticated user, when the server has no secret, or when the channel callback denies. - - Presence returns 403 if the user has no `id`, and otherwise signs `channel_data` `{"user_id": ""}`. -- Secret: `REVERB_APP_SECRET` has no default. `ReverbProvider.register` raises `RuntimeError` when the secret is empty or `"secret"` and `APP_ENV` is not `local` or `testing`. In local and testing an empty secret boots, but every `private-`/`presence-` subscription and auth request is rejected. -- `ReverbProvider` registers `/app/{key}` with `router.add_websocket_route` and `/broadcasting/auth` with `router.add_route`, a Starlette endpoint, so neither depends on route order or annotation resolution. - -The endpoint accepts form posts. A cross-site page can send one, but it cannot read the signature unless CORS is permissive, so no CSRF token is required. `pusher:unsubscribe` and `client-*` events remain unimplemented (unchanged); `client-*` events are dropped, so nothing leaks. - -## Validation - -- Tests drive the fully booted app through `TestClient`. The auth route and routes added after boot are reachable whatever the provider order. A private subscribe without a signature is rejected and receives no broadcasts. Forged, wrong-socket, wrong-secret and non-string signatures are rejected, as is tampered presence `channel_data`. -- Regression tests replay the reviewer's attack against `private-orders.1` and `presence-room.vip`. Before this change they ended in `subscription_succeeded`. Further tests cover a self-signed private signature over `socket:channel:data`, malformed socket ids and channel names on both auth and subscribe, 403 for public channels and anonymous users, and 400/422 bodies. -- Secret tests cover the empty default, refusing to boot in `production` with an unset, empty or `"secret"` value, booting in local and testing, and rejecting every protected subscription when the secret is empty. -- Server tests cover each malformed frame shape (error reply, connection kept), oversized frames (close 1009), and cleanup after malformed frames and after a handler exception. -- Full suite: `uv run pytest --ignore=tests/masoniteorm/postgres --cov`. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index 254af36f..c13f7f1b 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -35,4 +35,3 @@ Read this index first, then open the records relevant to the change. Before impl | [029](029-cache-module.md) | 2026-10-21 | Cache module with remember, memo, locks and tags | New `fastapi_startkit.cache`: async `Cache.get/put/has/forget/flush`, `get` with a callable default, and `remember(key, ttl, factory)` that runs a sync or async factory only on a miss; `remember(key, ttl)` without a factory is a decorator keyed by `key` plus a hash of the call arguments. Stores implement an async contract with a `MISSING` sentinel (cached `None` is a hit): array, file (pickled per-key files), database (masoniteorm table) and redis (named connection from the existing `RedisManager`, `SET PX` TTL, flush by `SCAN` of prefixed keys, never `FLUSHDB`). TTL in seconds or `timedelta`, `None` forever, <=0 forgets. `CacheConfig` via `resolve_config`; `CacheProvider` binds `cache`; facade `remember` resolves lazily. `Cache.memo()` memoizes reads per asyncio task through a `ContextVar` owned by the task, so requests never share memo state. `Cache.lock(name, seconds, owner)` gives atomic, owner-checked, non-re-entrant locks with `get`, `block` (250 ms polling, `LockTimeoutError`), `release`, `force_release` and `async with`, per store: event-loop dict, `fcntl.flock` file, `lock:` row with update-if-expired, Redis `SET NX PX` plus a Lua owner-checked delete. `Cache.tags([...])` uses Laravel's namespaced keys: random tag ids at `:tag::key`, entries at `:sha1(ids):`, flush rewrites the ids; tags are sorted and deduplicated. Shipped in slices into `feature/caches`; the former memo, lock and tags ADRs (030, 031, 034) are folded in. | | [032](032-migrate-seed-option.md) | 2026-10-21 | db:migrate:fresh and db:migrate:refresh --seed run seeders in the same event loop | `--seed` called the unregistered `seed:run` through Cleo `call` (which also drops the first argument) inside a running event loop, where `db:seed` would call `asyncio.run` again. The seeder selection moves into async `DBSeedCommand.run_seeders`, which `db:seed`, `db:migrate:fresh` and `db:migrate:refresh` all await. `db:migrate:refresh` is now registered. | | [033](033-async-seeder-stub.md) | 2026-10-21 | Generated seeders have an async run | `create_seed.stub` generated a synchronous `run` with docstrings, but `Seeder` awaits `run()`, so every generated table seeder raised `TypeError`. The stub now generates `async def run(self) -> None` with no docstrings; `Seeder` is unchanged. | -| [036](036-broadcasting-channel-auth.md) | 2026-10-25 | Signed private-channel subscriptions and a reachable /broadcasting/auth | `ReverbServer` subscribed any socket to `private-`/`presence-` channels without checking `auth`, and the empty-path WebSocket `Mount` shadowed `/broadcasting/auth` and later routes (404). The endpoint returns a Pusher signature `key:HMAC-SHA256(secret, socket_id:channel[:channel_data])` and the server verifies it with `compare_digest` against the connection's own socket id (`pusher:subscription_error` 401 on failure). The colon-joined wire format is kept for Echo compatibility, so ambiguity is closed by validation: socket ids are Pusher `digits.digits`, channel names are Pusher charset with no `:`, and `channel_data` is signed only for `presence-`. The endpoint returns 403 for public channels, anonymous users and presence users without an id, 400/422 for malformed bodies. `REVERB_APP_SECRET` has no default; the provider refuses to boot outside local/testing with an empty or `"secret"` value, and an empty secret rejects every protected subscription. Malformed frames get `pusher:error`, frames over 10 000 chars close with 1009, and connections are always cleaned up in `finally`. Both endpoints are plain router routes, so route order does not matter. | From 1146464685fdb6f23f910949a29a3d24b4e31c96 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Mon, 26 Oct 2026 14:46:14 -0700 Subject: [PATCH 5/5] Cover remaining reverb patch lines with behavior tests Co-Authored-By: Claude Sonnet 5.5 --- .../broadcasting/test_reverb_internals.py | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) create mode 100644 fastapi_startkit/tests/broadcasting/test_reverb_internals.py diff --git a/fastapi_startkit/tests/broadcasting/test_reverb_internals.py b/fastapi_startkit/tests/broadcasting/test_reverb_internals.py new file mode 100644 index 00000000..2d559c63 --- /dev/null +++ b/fastapi_startkit/tests/broadcasting/test_reverb_internals.py @@ -0,0 +1,84 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest +from starlette.websockets import WebSocketDisconnect + +from fastapi_startkit.broadcasting.reverb.channel_manager import ChannelManager +from fastapi_startkit.broadcasting.reverb.connection import ConnectionManager +from fastapi_startkit.broadcasting.reverb.server import ReverbServer +from fastapi_startkit.broadcasting.reverb.signature import sign, verify + +KEY = "local" +SECRET = "unit-secret" +SOCKET_ID = "123.456" + + +def test_unsubscribe_from_unknown_channel_keeps_other_subscriptions(): + channels = ChannelManager() + channels.subscribe("orders.1", "1.1") + + channels.unsubscribe("missing", "1.1") + + assert channels.subscribers("orders.1") == {"1.1"} + assert channels.subscribers("missing") == set() + + +def test_connection_manager_retries_when_socket_id_is_taken(monkeypatch): + connections = ConnectionManager() + first, second = MagicMock(), MagicMock() + ids = iter(["1.1", "1.1", "2.2"]) + monkeypatch.setattr(ConnectionManager, "_new_socket_id", staticmethod(lambda: next(ids))) + + first_id = connections.add(first) + second_id = connections.add(second) + + assert (first_id, second_id) == ("1.1", "2.2") + assert connections.get("1.1") is first + assert connections.get("2.2") is second + + +@pytest.mark.parametrize( + ("socket_id", "channel"), + [ + ("not-a-socket-id", "private-orders.1"), + (SOCKET_ID, "private-orders:1"), + (SOCKET_ID, "orders.1"), + ], +) +def test_verify_rejects_invalid_socket_id_or_unprotected_channel(socket_id, channel): + auth = sign(KEY, SECRET, socket_id, channel) + + assert verify(KEY, SECRET, socket_id, channel, auth) is False + + +def test_verify_accepts_valid_signature_for_private_channel(): + channel = "private-orders.1" + auth = sign(KEY, SECRET, SOCKET_ID, channel) + + assert verify(KEY, SECRET, SOCKET_ID, channel, auth) is True + + +async def test_disconnect_while_handshaking_releases_connection(): + server = ReverbServer(KEY, SECRET) + websocket = MagicMock() + websocket.accept = AsyncMock() + websocket.send_text = AsyncMock(side_effect=WebSocketDisconnect(1006)) + + await server.handle(websocket) + + assert server.connections.all() == {} + + +async def test_broadcast_skips_failing_socket_and_reaches_the_rest(): + server = ReverbServer(KEY, SECRET) + broken, healthy = MagicMock(), MagicMock() + broken.send_text = AsyncMock(side_effect=RuntimeError("closed")) + healthy.send_text = AsyncMock() + for websocket in (broken, healthy): + socket_id = server.connections.add(websocket) + server.channels.subscribe("orders.1", socket_id) + + await server.broadcast_to_channel("orders.1", "OrderShipped", {"order_id": 1}) + + broken.send_text.assert_awaited_once() + healthy.send_text.assert_awaited_once()