Fix broadcasting private-channel auth bypass and /broadcasting/auth 404

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)

fastapi_startkit/src/fastapi_startkit/broadcasting/provider.py

@@ -2,13 +2,19 @@
 
 import importlib
 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 is_presence, requires_auth, sign, valid_channel, valid_socket_id
+
+UNSAFE_SECRETS = ("", "secret")
+INSECURE_ENVIRONMENTS = ("local", "testing")
 
 
 class ReverbProvider(Provider):
@@ -36,7 +42,14 @@ 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", {})
+        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)
@@ -81,91 +94,75 @@ 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", "")
+            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:
-                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)
+                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)
+
+            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)
 
-            # 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)
+            if not await manager.channel_registry.authorize(channel_name, user):
+                return JSONResponse({"error": "Forbidden"}, status_code=403)
+
+            if not is_presence(channel_name):
+                return JSONResponse({"auth": sign(server.key, server.secret, socket_id, channel_name)})
 
-            if not allowed:
+            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})
 
-            return JSONResponse({"auth": True, "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

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()))

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)

fastapi_startkit/src/fastapi_startkit/broadcasting/reverb/server.py

@@ -1,48 +1,108 @@
 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 .connection import ConnectionManager
+from .signature import requires_auth, valid_channel, verify
+
+MESSAGE_TOO_BIG = 1009
 
 
 class ReverbServer:
-    def __init__(self):
+    max_message_size = 10_000
+
+    def __init__(self, key: str = "local", secret: str = ""):
+        self.key = key
+        self.secret = secret
         self.connections = ConnectionManager()
         self.channels = ChannelManager()
 
     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":
-            channel = message.get("data", {}).get("channel", "")
-            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)