Prevent subscription execution without websocket

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Added `execute_get_queries` setting to the `GraphQL` apps that controls execution of the GraphQL "query" operations made with GET requests. Defaults to `False`.
 - Added support for the Apollo Federation versions up to 2.6.
 - Fixed deprecation warnings in Apollo Tracing extension.
+- Added a check to prevent `subscription` operation execution when query is made with POST request.
 
 
 ## 0.22 (2024-01-31)

ariadne/graphql.py

@@ -185,6 +185,8 @@ async def graphql(
 
             if require_query:
                 validate_operation_is_query(document, operation_name)
+            else:
+                validate_operation_is_not_subscription(document, operation_name)
 
             if callable(root_value):
                 try:
@@ -358,6 +360,8 @@ def graphql_sync(
 
             if require_query:
                 validate_operation_is_query(document, operation_name)
+            else:
+                validate_operation_is_not_subscription(document, operation_name)
 
             if callable(root_value):
                 try:
@@ -680,3 +684,40 @@ def validate_operation_is_query(
         raise GraphQLError(
             "'operationName' is required if 'query' defines multiple operations."
         )
+
+
+def validate_operation_is_not_subscription(
+    document_ast: DocumentNode, operation_name: Optional[str]
+):
+    if operation_name:
+        validate_named_operation_is_not_subscription(document_ast, operation_name)
+    else:
+        validate_anonymous_operation_is_not_subscription(document_ast)
+
+
+def validate_named_operation_is_not_subscription(
+    document_ast: DocumentNode, operation_name: str
+):
+    for definition in document_ast.definitions:
+        if (
+            isinstance(definition, OperationDefinitionNode)
+            and definition.name
+            and definition.name.value == operation_name
+            and definition.operation.name == "SUBSCRIPTION"
+        ):
+            raise GraphQLError(
+                f"Operation '{operation_name}' is a subscription and can only be "
+                "executed over a WebSocket connection."
+            )
+
+
+def validate_anonymous_operation_is_not_subscription(document_ast: DocumentNode):
+    operations: List[OperationDefinitionNode] = []
+    for definition in document_ast.definitions:
+        if isinstance(definition, OperationDefinitionNode):
+            operations.append(definition)
+
+    if len(operations) == 1 and operations[0].operation.name == "SUBSCRIPTION":
+        raise GraphQLError(
+            "Subscription operations can only be executed over a WebSocket connection."
+        )

tests/asgi/__snapshots__/test_query_execution.ambr

@@ -1,4 +1,13 @@
 # serializer version: 1
+# name: test_attempt_execute_anonymous_subscription_over_post_returns_error_json
+  dict({
+    'errors': list([
+      dict({
+        'message': 'Subscription operations can only be executed over a WebSocket connection.',
+      }),
+    ]),
+  })
+# ---
 # name: test_attempt_execute_complex_query_without_variables_returns_error_json
   dict({
     'data': None,
@@ -61,6 +70,15 @@
     ]),
   })
 # ---
+# name: test_attempt_execute_subscription_over_post_returns_error_json
+  dict({
+    'errors': list([
+      dict({
+        'message': "Operation 'Test' is a subscription and can only be executed over a WebSocket connection.",
+      }),
+    ]),
+  })
+# ---
 # name: test_attempt_execute_subscription_with_invalid_query_returns_error_json
   dict({
     'locations': list([

tests/asgi/test_query_execution.py

@@ -107,6 +107,26 @@ def test_attempt_execute_query_with_invalid_operation_name_type_returns_error_js
     assert snapshot == response.json()
 
 
+def test_attempt_execute_anonymous_subscription_over_post_returns_error_json(
+    client, snapshot
+):
+    response = client.post("/", json={"query": "subscription { ping }"})
+    assert response.status_code == 400
+    assert snapshot == response.json()
+
+
+def test_attempt_execute_subscription_over_post_returns_error_json(client, snapshot):
+    response = client.post(
+        "/",
+        json={
+            "query": "subscription Test { ping }",
+            "operationName": "Test",
+        },
+    )
+    assert response.status_code == 400
+    assert snapshot == response.json()
+
+
 def test_attempt_execute_subscription_with_invalid_query_returns_error_json(
     client, snapshot
 ):

tests/wsgi/__snapshots__/test_query_execution.ambr

@@ -1,4 +1,13 @@
 # serializer version: 1
+# name: test_attempt_execute_anonymous_subscription_over_post_returns_error_json
+  dict({
+    'errors': list([
+      dict({
+        'message': 'Subscription operations can only be executed over a WebSocket connection.',
+      }),
+    ]),
+  })
+# ---
 # name: test_attempt_execute_complex_query_without_variables_returns_error_json
   dict({
     'data': None,
@@ -61,6 +70,15 @@
     ]),
   })
 # ---
+# name: test_attempt_execute_subscription_over_post_returns_error_json
+  dict({
+    'errors': list([
+      dict({
+        'message': "Operation 'Test' is a subscription and can only be executed over a WebSocket connection.",
+      }),
+    ]),
+  })
+# ---
 # name: test_complex_query_is_executed_for_post_json_request
   dict({
     'data': dict({

tests/wsgi/test_query_execution.py

@@ -158,6 +158,38 @@ def test_attempt_execute_query_with_invalid_operation_name_type_returns_error_js
     assert_json_response_equals_snapshot(result)
 
 
+def test_attempt_execute_anonymous_subscription_over_post_returns_error_json(
+    middleware,
+    start_response,
+    graphql_query_request_factory,
+    graphql_response_headers,
+    assert_json_response_equals_snapshot,
+):
+    request = graphql_query_request_factory(query="subscription { ping }")
+    result = middleware(request, start_response)
+    start_response.assert_called_once_with(
+        HTTP_STATUS_400_BAD_REQUEST, graphql_response_headers
+    )
+    assert_json_response_equals_snapshot(result)
+
+
+def test_attempt_execute_subscription_over_post_returns_error_json(
+    middleware,
+    start_response,
+    graphql_query_request_factory,
+    graphql_response_headers,
+    assert_json_response_equals_snapshot,
+):
+    request = graphql_query_request_factory(
+        query="subscription Test { ping }", operationName="Test"
+    )
+    result = middleware(request, start_response)
+    start_response.assert_called_once_with(
+        HTTP_STATUS_400_BAD_REQUEST, graphql_response_headers
+    )
+    assert_json_response_equals_snapshot(result)
+
+
 def test_query_is_executed_for_multipart_form_request_with_file(
     middleware, snapshot, start_response, graphql_response_headers
 ):