CVE-2026-35523

@@ -0,0 +1,28 @@
+Release type: patch
+
+This release fixes two security vulnerabilities in the WebSocket subscription
+handlers (CVE-2026-35526, CVE-2026-35523).
+
+**CVE-2026-35526 - Authentication bypass in `graphql-ws`**: The legacy
+`graphql-ws` protocol handler didn't verify that the `connection_init`
+handshake was completed before accepting `start` messages, allowing clients
+to bypass any authentication logic in `on_ws_connect`. The connection is now
+closed with `4401 Unauthorized` if the handshake hasn't been completed.
+
+**CVE-2026-35523 - Unbounded subscriptions per connection**: Both WebSocket
+protocol handlers allowed unlimited concurrent subscriptions on a single
+connection, making it possible for a malicious client to exhaust server
+resources. A new `max_subscriptions_per_connection` parameter has been added
+to all views (default: `100`). Set it to `None` to disable the limit.
+
+Example:
+
+```python
+import strawberry
+from strawberry.fastapi import GraphQLRouter
+
+schema = strawberry.Schema(query=Query, subscription=Subscription)
+
+# default is 100, set to None to disable the limit
+graphql_app = GraphQLRouter(schema, max_subscriptions_per_connection=50)
+```

@@ -102,6 +102,7 @@ def __init__(
         ),
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ) -> None:
         self.schema = schema
         self.allow_queries_via_get = allow_queries_via_get
@@ -110,6 +111,7 @@ def __init__(
         self.subscription_protocols = subscription_protocols
         self.connection_init_wait_timeout = connection_init_wait_timeout
         self.multipart_uploads_enabled = multipart_uploads_enabled
+        self.max_subscriptions_per_connection = max_subscriptions_per_connection
         self.graphql_ide = graphql_ide
 
     async def render_graphql_ide(self, request: web.Request) -> web.Response:

@@ -114,6 +114,7 @@ def __init__(
         ),
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ) -> None:
         self.schema = schema
         self.allow_queries_via_get = allow_queries_via_get
@@ -122,6 +123,7 @@ def __init__(
         self.protocols = subscription_protocols
         self.connection_init_wait_timeout = connection_init_wait_timeout
         self.multipart_uploads_enabled = multipart_uploads_enabled
+        self.max_subscriptions_per_connection = max_subscriptions_per_connection
         self.graphql_ide = graphql_ide
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:

@@ -112,13 +112,15 @@ def __init__(
             GRAPHQL_WS_PROTOCOL,
         ),
         connection_init_wait_timeout: datetime.timedelta | None = None,
+        max_subscriptions_per_connection: int | None = 100,
     ) -> None:
         if connection_init_wait_timeout is None:
             connection_init_wait_timeout = datetime.timedelta(minutes=1)
         self.connection_init_wait_timeout = connection_init_wait_timeout
         self.schema = schema
         self.keep_alive = keep_alive
         self.keep_alive_interval = keep_alive_interval
+        self.max_subscriptions_per_connection = max_subscriptions_per_connection
         self.protocols = subscription_protocols
         self.message_queue: asyncio.Queue[MessageQueueData] = asyncio.Queue()
         self.run_task: asyncio.Task | None = None

@@ -131,6 +131,7 @@ def __init__(
             GRAPHQL_WS_PROTOCOL,
         ),
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
+        max_subscriptions_per_connection: int | None = 100,
         prefix: str = "",
         tags: list[str | Enum] | None = None,
         dependencies: Sequence[params.Depends] | None = None,
@@ -184,6 +185,7 @@ def __init__(
         )
         self.protocols = subscription_protocols
         self.connection_init_wait_timeout = connection_init_wait_timeout
+        self.max_subscriptions_per_connection = max_subscriptions_per_connection
         self.multipart_uploads_enabled = multipart_uploads_enabled
         self.graphql_ide = graphql_ide
 

@@ -88,6 +88,7 @@ class AsyncBaseHTTPView(
     keep_alive = False
     keep_alive_interval: float | None = None
     connection_init_wait_timeout: timedelta = timedelta(minutes=1)
+    max_subscriptions_per_connection: int | None = 100
     request_adapter_class: Callable[[Request], AsyncHTTPRequestAdapter]
     websocket_adapter_class: Callable[
         [
@@ -322,6 +323,7 @@ async def run(
                     root_value=root_value,
                     schema=self.schema,
                     connection_init_wait_timeout=self.connection_init_wait_timeout,
+                    max_subscriptions_per_connection=self.max_subscriptions_per_connection,
                 ).handle()
             elif websocket_subprotocol == GRAPHQL_WS_PROTOCOL:
                 await self.graphql_ws_handler_class(
@@ -332,6 +334,7 @@ async def run(
                     schema=self.schema,
                     keep_alive=self.keep_alive,
                     keep_alive_interval=self.keep_alive_interval,
+                    max_subscriptions_per_connection=self.max_subscriptions_per_connection,
                 ).handle()
             else:
                 await websocket.close(4406, "Subprotocol not acceptable")

@@ -224,6 +224,7 @@ class GraphQLController(
     )
     keep_alive: bool = False
     keep_alive_interval: float = 1
+    max_subscriptions_per_connection: int | None = 100
 
     def is_websocket_request(
         self, request: Request | WebSocket
@@ -385,6 +386,7 @@ def make_graphql_controller(
     ),
     connection_init_wait_timeout: timedelta = timedelta(minutes=1),
     multipart_uploads_enabled: bool = False,
+    max_subscriptions_per_connection: int | None = 100,
 ) -> type[GraphQLController]:  # sourcery skip: move-assign
     if context_getter is None:
         custom_context_getter_ = _none_custom_context_getter
@@ -421,6 +423,9 @@ class _GraphQLController(GraphQLController):
     _GraphQLController.allow_queries_via_get = allow_queries_via_get_
     _GraphQLController.graphql_ide = graphql_ide_
     _GraphQLController.multipart_uploads_enabled = multipart_uploads_enabled
+    _GraphQLController.max_subscriptions_per_connection = (
+        max_subscriptions_per_connection
+    )
 
     return _GraphQLController
 

@@ -93,6 +93,7 @@ def __init__(
         ),
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ) -> None:
         self.schema = schema
         self.allow_queries_via_get = allow_queries_via_get
@@ -101,6 +102,7 @@ def __init__(
         self.subscription_protocols = subscription_protocols
         self.connection_init_wait_timeout = connection_init_wait_timeout
         self.multipart_uploads_enabled = multipart_uploads_enabled
+        self.max_subscriptions_per_connection = max_subscriptions_per_connection
         self.graphql_ide = graphql_ide
 
     async def render_graphql_ide(self, request: Request) -> Response:

@@ -54,13 +54,15 @@ def __init__(
         root_value: RootValue | None,
         schema: BaseSchema,
         connection_init_wait_timeout: timedelta,
+        max_subscriptions_per_connection: int | None = None,
     ) -> None:
         self.view = view
         self.websocket = websocket
         self.context = context
         self.root_value = root_value
         self.schema = schema
         self.connection_init_wait_timeout = connection_init_wait_timeout
+        self.max_subscriptions_per_connection = max_subscriptions_per_connection
         self.connection_init_timeout_task: asyncio.Task | None = None
         self.connection_init_received = False
         self.connection_acknowledged = False
@@ -233,6 +235,22 @@ async def handle_subscribe(self, message: SubscribeMessage) -> None:
             await self.websocket.close(code=4409, reason=reason)
             return
 
+        # NOTE: this applies to all in-flight operations (queries and mutations
+        # executed over WebSocket included), not only subscriptions.
+        if (
+            self.max_subscriptions_per_connection is not None
+            and len(self.operations) >= self.max_subscriptions_per_connection
+        ):
+            error = GraphQLError("Subscription limit reached")
+            await self.send_message(
+                {
+                    "id": message["id"],
+                    "type": "error",
+                    "payload": [error.formatted],
+                }
+            )
+            return
+
         operation = Operation(
             self,
             message["id"],

@@ -44,6 +44,7 @@ def __init__(
         schema: BaseSchema,
         keep_alive: bool,
         keep_alive_interval: float | None,
+        max_subscriptions_per_connection: int | None = None,
     ) -> None:
         self.view = view
         self.websocket = websocket
@@ -52,9 +53,11 @@ def __init__(
         self.schema = schema
         self.keep_alive = keep_alive
         self.keep_alive_interval = keep_alive_interval
+        self.max_subscriptions_per_connection = max_subscriptions_per_connection
         self.keep_alive_task: asyncio.Task | None = None
         self.subscriptions: dict[str, AsyncGenerator] = {}
         self.tasks: dict[str, asyncio.Task] = {}
+        self.connection_acknowledged: bool = False
 
     async def handle(self) -> None:
         try:
@@ -119,6 +122,8 @@ async def handle_connection_init(self, message: ConnectionInitMessage) -> None:
                 {"type": "connection_ack", "payload": connection_ack_payload}
             )
 
+        self.connection_acknowledged = True
+
         if self.keep_alive:
             keep_alive_handler = self.handle_keep_alive()
             self.keep_alive_task = asyncio.create_task(keep_alive_handler)
@@ -129,7 +134,29 @@ async def handle_connection_terminate(
         await self.websocket.close(code=1000, reason="")
 
     async def handle_start(self, message: StartMessage) -> None:
+        if not self.connection_acknowledged:
+            await self.websocket.close(code=4401, reason="Unauthorized")
+            return
+
         operation_id = message["id"]
+
+        # Clean up existing operation with same ID to prevent task leaks
+        if operation_id in self.tasks:
+            await self.cleanup_operation(operation_id)
+
+        if (
+            self.max_subscriptions_per_connection is not None
+            and len(self.tasks) >= self.max_subscriptions_per_connection
+        ):
+            await self.send_message(
+                ErrorMessage(
+                    type="error",
+                    id=operation_id,
+                    payload={"message": "Subscription limit reached"},
+                )
+            )
+            return
+
         payload = message["payload"]
         query = payload["query"]
         operation_name = payload.get("operationName")

@@ -74,6 +74,7 @@ def __init__(
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         result_override: ResultOverrideFunction = None,
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ):
         view = GraphQLView(
             schema=schema,
@@ -84,6 +85,7 @@ def __init__(
             subscription_protocols=subscription_protocols,
             connection_init_wait_timeout=connection_init_wait_timeout,
             multipart_uploads_enabled=multipart_uploads_enabled,
+            max_subscriptions_per_connection=max_subscriptions_per_connection,
         )
         view.result_override = result_override
 

@@ -78,6 +78,7 @@ def __init__(
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         result_override: ResultOverrideFunction = None,
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ):
         view = GraphQLView(
             schema,
@@ -88,6 +89,7 @@ def __init__(
             subscription_protocols=subscription_protocols,
             connection_init_wait_timeout=connection_init_wait_timeout,
             multipart_uploads_enabled=multipart_uploads_enabled,
+            max_subscriptions_per_connection=max_subscriptions_per_connection,
         )
         view.result_override = result_override
 

@@ -108,6 +108,7 @@ def __init__(
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         result_override: ResultOverrideFunction = None,
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ): ...
 
     @abc.abstractmethod

@@ -159,13 +159,15 @@ def __init__(
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         result_override: ResultOverrideFunction = None,
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ):
         self.ws_app = DebuggableGraphQLWSConsumer.as_asgi(
             schema=schema,
             keep_alive=keep_alive,
             keep_alive_interval=keep_alive_interval,
             subscription_protocols=subscription_protocols,
             connection_init_wait_timeout=connection_init_wait_timeout,
+            max_subscriptions_per_connection=max_subscriptions_per_connection,
         )
 
         self.http_app = DebuggableGraphQLHTTPConsumer.as_asgi(

@@ -89,6 +89,7 @@ def __init__(
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         result_override: ResultOverrideFunction = None,
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ):
         self.app = FastAPI()
 
@@ -101,6 +102,7 @@ def __init__(
             subscription_protocols=subscription_protocols,
             connection_init_wait_timeout=connection_init_wait_timeout,
             multipart_uploads_enabled=multipart_uploads_enabled,
+            max_subscriptions_per_connection=max_subscriptions_per_connection,
             context_getter=fastapi_get_context,
             root_value_getter=get_root_value,
         )

@@ -65,6 +65,7 @@ def __init__(
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         result_override: ResultOverrideFunction = None,
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ):
         BaseGraphQLController = make_graphql_controller(
             schema=schema,
@@ -75,6 +76,7 @@ def __init__(
             subscription_protocols=subscription_protocols,
             connection_init_wait_timeout=connection_init_wait_timeout,
             multipart_uploads_enabled=multipart_uploads_enabled,
+            max_subscriptions_per_connection=max_subscriptions_per_connection,
             path="/graphql",
             context_getter=litestar_get_context,
             root_value_getter=get_root_value,

@@ -99,6 +99,7 @@ def __init__(
         connection_init_wait_timeout: timedelta = timedelta(minutes=1),
         result_override: ResultOverrideFunction = None,
         multipart_uploads_enabled: bool = False,
+        max_subscriptions_per_connection: int | None = 100,
     ):
         self.app = Quart(__name__)
         self.app.debug = True
@@ -114,6 +115,7 @@ def __init__(
             subscription_protocols=subscription_protocols,
             connection_init_wait_timeout=connection_init_wait_timeout,
             multipart_uploads_enabled=multipart_uploads_enabled,
+            max_subscriptions_per_connection=max_subscriptions_per_connection,
         )
 
         self.app.add_url_rule(

@@ -1216,3 +1216,100 @@ async def test_unexpected_client_disconnects_are_gracefully_handled(
 
         assert not process_errors.called
         assert Subscription.active_infinity_subscriptions == 0
+
+
+async def test_max_subscriptions_per_connection(http_client_class: type[HttpClient]):
+    """Test that subscriptions beyond the limit are rejected with an error."""
+    test_client = http_client_class(schema, max_subscriptions_per_connection=2)
+
+    async with test_client.ws_connect(
+        "/graphql", protocols=[GRAPHQL_TRANSPORT_WS_PROTOCOL]
+    ) as ws:
+        await ws.send_message({"type": "connection_init"})
+        connection_ack_message: ConnectionAckMessage = await ws.receive_json()
+        assert connection_ack_message == {"type": "connection_ack"}
+
+        # First two subscriptions should succeed
+        await ws.send_message(
+            {
+                "id": "sub1",
+                "type": "subscribe",
+                "payload": {"query": 'subscription { infinity(message: "Hi") }'},
+            }
+        )
+        next_message: NextMessage = await ws.receive_json()
+        assert next_message["type"] == "next"
+        assert next_message["id"] == "sub1"
+
+        await ws.send_message(
+            {
+                "id": "sub2",
+                "type": "subscribe",
+                "payload": {"query": 'subscription { infinity(message: "Hi") }'},
+            }
+        )
+        next_message = await ws.receive_json()
+        assert next_message["type"] == "next"
+        assert next_message["id"] == "sub2"
+
+        # Third subscription should be rejected
+        await ws.send_message(
+            {
+                "id": "sub3",
+                "type": "subscribe",
+                "payload": {"query": 'subscription { infinity(message: "Hi") }'},
+            }
+        )
+        error_message: ErrorMessage = await ws.receive_json()
+        assert error_message["type"] == "error"
+        assert error_message["id"] == "sub3"
+        assert error_message["payload"] == [{"message": "Subscription limit reached"}]
+
+        # Completing one should free a slot
+        await ws.send_message({"id": "sub1", "type": "complete"})
+
+        await ws.send_message(
+            {
+                "id": "sub4",
+                "type": "subscribe",
+                "payload": {"query": 'subscription { infinity(message: "Hi") }'},
+            }
+        )
+        next_message = await ws.receive_json()
+        assert next_message["type"] == "next"
+        assert next_message["id"] == "sub4"
+
+        await ws.send_message({"id": "sub2", "type": "complete"})
+        await ws.send_message({"id": "sub4", "type": "complete"})
+        await ws.close()
+
+
+async def test_max_subscriptions_per_connection_disabled(
+    http_client_class: type[HttpClient],
+):
+    """Test that setting max_subscriptions_per_connection to None disables the limit."""
+    test_client = http_client_class(schema, max_subscriptions_per_connection=None)
+
+    async with test_client.ws_connect(
+        "/graphql", protocols=[GRAPHQL_TRANSPORT_WS_PROTOCOL]
+    ) as ws:
+        await ws.send_message({"type": "connection_init"})
+        connection_ack_message: ConnectionAckMessage = await ws.receive_json()
+        assert connection_ack_message == {"type": "connection_ack"}
+
+        # Should be able to create many subscriptions without limit
+        for i in range(5):
+            await ws.send_message(
+                {
+                    "id": f"sub{i}",
+                    "type": "subscribe",
+                    "payload": {"query": 'subscription { infinity(message: "Hi") }'},
+                }
+            )
+            next_message: NextMessage = await ws.receive_json()
+            assert next_message["type"] == "next"
+            assert next_message["id"] == f"sub{i}"
+
+        for i in range(5):
+            await ws.send_message({"id": f"sub{i}", "type": "complete"})
+        await ws.close()

@@ -255,6 +255,30 @@ async def test_rejecting_connection_with_custom_connection_error_payload(
     assert not ws_raw.close_reason
 
 
+async def test_start_without_connection_init_is_rejected(
+    ws_raw: WebSocketClient,
+):
+    """Sending a 'start' message without a prior 'connection_init' must be rejected.
+
+    This ensures that the on_ws_connect authentication hook cannot be bypassed
+    by skipping the connection_init handshake.
+    """
+    await ws_raw.send_legacy_message(
+        {
+            "type": "start",
+            "id": "1",
+            "payload": {
+                "query": "subscription { listener }",
+            },
+        }
+    )
+
+    await ws_raw.receive(timeout=2)
+    assert ws_raw.closed
+    assert ws_raw.close_code == 4401
+    assert ws_raw.close_reason == "Unauthorized"
+
+
 async def test_context_can_be_modified_from_within_on_ws_connect(
     ws_raw: WebSocketClient,
 ):
@@ -865,3 +889,175 @@ async def test_unexpected_client_disconnects_are_gracefully_handled(
 
         assert not process_errors.called
         assert Subscription.active_infinity_subscriptions == 0
+
+
+async def test_max_subscriptions_per_connection(http_client_class: type[HttpClient]):
+    """Test that subscriptions beyond the limit are rejected with an error."""
+    test_client = http_client_class(schema, max_subscriptions_per_connection=2)
+
+    async with test_client.ws_connect(
+        "/graphql", protocols=[GRAPHQL_WS_PROTOCOL]
+    ) as ws:
+        await ws.send_legacy_message({"type": "connection_init"})
+        response: ConnectionAckMessage = await ws.receive_json()
+        assert response["type"] == "connection_ack"
+
+        # First two subscriptions should succeed
+        await ws.send_legacy_message(
+            {
+                "type": "start",
+                "id": "sub1",
+                "payload": {
+                    "query": 'subscription { infinity(message: "Hi") }',
+                },
+            }
+        )
+        data_message: DataMessage = await ws.receive_json()
+        assert data_message["type"] == "data"
+        assert data_message["id"] == "sub1"
+
+        await ws.send_legacy_message(
+            {
+                "type": "start",
+                "id": "sub2",
+                "payload": {
+                    "query": 'subscription { infinity(message: "Hi") }',
+                },
+            }
+        )
+        data_message = await ws.receive_json()
+        assert data_message["type"] == "data"
+        assert data_message["id"] == "sub2"
+
+        # Third subscription should be rejected
+        await ws.send_legacy_message(
+            {
+                "type": "start",
+                "id": "sub3",
+                "payload": {
+                    "query": 'subscription { infinity(message: "Hi") }',
+                },
+            }
+        )
+        error_message: ErrorMessage = await ws.receive_json()
+        assert error_message["type"] == "error"
+        assert error_message["id"] == "sub3"
+        assert error_message["payload"] == {"message": "Subscription limit reached"}
+
+        # Completing one should free a slot
+        await ws.send_legacy_message({"type": "stop", "id": "sub1"})
+
+        await ws.send_legacy_message(
+            {
+                "type": "start",
+                "id": "sub4",
+                "payload": {
+                    "query": 'subscription { infinity(message: "Hi") }',
+                },
+            }
+        )
+        # The 'complete' for sub1 is always sent before sub4's first data message
+        # (cleanup_operation awaits the cancelled task, which sends the complete).
+        msg = await ws.receive_json()
+        while msg.get("id") == "sub1":
+            msg = await ws.receive_json()
+        assert msg["type"] == "data"
+        assert msg["id"] == "sub4"
+
+        await ws.close()
+
+
+async def test_reusing_operation_id_does_not_count_against_limit(
+    http_client_class: type[HttpClient],
+):
+    """Reusing an existing operation ID should clean up the old subscription
+    and not count against the subscription limit."""
+    test_client = http_client_class(schema, max_subscriptions_per_connection=2)
+
+    async with test_client.ws_connect(
+        "/graphql", protocols=[GRAPHQL_WS_PROTOCOL]
+    ) as ws:
+        await ws.send_legacy_message({"type": "connection_init"})
+        response: ConnectionAckMessage = await ws.receive_json()
+        assert response["type"] == "connection_ack"
+
+        # Fill up to the limit
+        await ws.send_legacy_message(
+            {
+                "type": "start",
+                "id": "sub1",
+                "payload": {
+                    "query": 'subscription { infinity(message: "Hi") }',
+                },
+            }
+        )
+        data_message: DataMessage = await ws.receive_json()
+        assert data_message["type"] == "data"
+        assert data_message["id"] == "sub1"
+
+        await ws.send_legacy_message(
+            {
+                "type": "start",
+                "id": "sub2",
+                "payload": {
+                    "query": 'subscription { infinity(message: "Hi") }',
+                },
+            }
+        )
+        data_message = await ws.receive_json()
+        assert data_message["type"] == "data"
+        assert data_message["id"] == "sub2"
+
+        # Reuse "sub1" — should clean up the old one, not hit the limit
+        await ws.send_legacy_message(
+            {
+                "type": "start",
+                "id": "sub1",
+                "payload": {
+                    "query": 'subscription { infinity(message: "Hello") }',
+                },
+            }
+        )
+
+        msg = await ws.receive_json()
+        # skip any leftover messages from the old sub1
+        while msg.get("id") == "sub1" and msg.get("type") == "complete":
+            msg = await ws.receive_json()
+
+        assert msg["type"] == "data"
+        assert msg["id"] == "sub1"
+
+        await ws.close()
+
+
+async def test_max_subscriptions_per_connection_disabled(
+    http_client_class: type[HttpClient],
+):
+    """Test that setting max_subscriptions_per_connection to None disables the limit."""
+    test_client = http_client_class(schema, max_subscriptions_per_connection=None)
+
+    async with test_client.ws_connect(
+        "/graphql", protocols=[GRAPHQL_WS_PROTOCOL]
+    ) as ws:
+        await ws.send_legacy_message({"type": "connection_init"})
+        response: ConnectionAckMessage = await ws.receive_json()
+        assert response["type"] == "connection_ack"
+
+        # Should be able to create many subscriptions without limit
+        for i in range(5):
+            await ws.send_legacy_message(
+                {
+                    "type": "start",
+                    "id": f"sub{i}",
+                    "payload": {
+                        "query": 'subscription { infinity(message: "Hi") }',
+                    },
+                }
+            )
+            data_message: DataMessage = await ws.receive_json()
+            assert data_message["type"] == "data"
+            assert data_message["id"] == f"sub{i}"
+
+        for i in range(5):
+            await ws.send_legacy_message({"type": "stop", "id": f"sub{i}"})
+        await ws.close()