feat: validate GraphQL query depth

Refs: HP-2349

open_city_profile/settings.py

@@ -52,6 +52,7 @@
     OPEN_CITY_PROFILE_LOG_LEVEL=(str, None),
     ENABLE_GRAPHIQL=(bool, False),
     ENABLE_GRAPHQL_INTROSPECTION=(bool, False),
+    GRAPHQL_QUERY_DEPTH_LIMIT=(int, 12),
     FORCE_SCRIPT_NAME=(str, ""),
     CSRF_COOKIE_NAME=(str, ""),
     CSRF_COOKIE_PATH=(str, ""),
@@ -160,6 +161,7 @@
 ENABLE_GRAPHIQL = env("ENABLE_GRAPHIQL")
 # Enable GraphQL introspection queries, enabled automatically if DEBUG=True
 ENABLE_GRAPHQL_INTROSPECTION = env("ENABLE_GRAPHQL_INTROSPECTION")
+GRAPHQL_QUERY_DEPTH_LIMIT = env("GRAPHQL_QUERY_DEPTH_LIMIT")
 
 INSTALLED_APPS = [
     "helusers.apps.HelusersConfig",

open_city_profile/tests/graphql_test_helpers.py

@@ -74,7 +74,11 @@ def __call__(self, request):
 
 
 def do_graphql_call(
-    live_server, request_auth=None, query=_QUERY, extra_request_args=None
+    live_server,
+    request_auth=None,
+    query=_QUERY,
+    extra_request_args=None,
+    expected_status=200,
 ):
     if extra_request_args is None:
         extra_request_args = {}
@@ -96,7 +100,7 @@ def do_graphql_call(
         response = requests.post(url, json=payload, **request_args)
 
     assert (
-        response.status_code == 200
+        response.status_code == expected_status
     ), f"Status: {response.status_code}. Body:\n{response.text}"
 
     body = response.json()
@@ -113,6 +117,7 @@ def do_graphql_call_as_user(
     extra_claims=None,
     query=_QUERY,
     extra_request_args=None,
+    expected_status=200,
 ):
     if extra_request_args is None:
         extra_request_args = {}
@@ -138,4 +143,5 @@ def do_graphql_call_as_user(
         BearerTokenAuth(extra_claims=claims),
         query=query,
         extra_request_args=extra_request_args,
+        expected_status=expected_status,
     )

open_city_profile/tests/test_graphql_api_schema.py

@@ -1,32 +1,7 @@
-import pytest
-import requests
-from graphql import get_introspection_query, print_schema
+from graphql import print_schema
 
 
 def test_graphql_schema_matches_the_reference(gql_schema, snapshot):
     actual_schema = print_schema(gql_schema)
 
     snapshot.assert_match(actual_schema)
-
-
-@pytest.mark.parametrize("enabled", [True, False])
-def test_graphql_schema_introspection_can_be_disabled(live_server, settings, enabled):
-    settings.ENABLE_GRAPHQL_INTROSPECTION = enabled
-    url = live_server.url + "/graphql/"
-    payload = {
-        "query": get_introspection_query(),
-    }
-
-    response = requests.post(url, json=payload)
-
-    body = response.json()
-    if enabled:
-        assert response.status_code == 200
-        assert "errors" not in body
-        assert "__schema" in body["data"]
-    else:
-        assert response.status_code == 400
-        assert "data" not in body
-        error = body["errors"][0]["message"]
-        assert "__schema" in error
-        assert "introspection is disabled" in error

open_city_profile/tests/test_graphql_api_validation_rules.py

@@ -0,0 +1,68 @@
+import pytest
+from graphql import get_introspection_query
+
+from open_city_profile.tests.graphql_test_helpers import (
+    do_graphql_call,
+    do_graphql_call_as_user,
+)
+
+
+@pytest.mark.parametrize(
+    "enabled, expected_status",
+    [pytest.param(True, 200, id="enabled"), pytest.param(False, 400, id="disabled")],
+)
+def test_graphql_schema_introspection_can_be_disabled(
+    live_server, settings, enabled, expected_status
+):
+    settings.ENABLE_GRAPHQL_INTROSPECTION = enabled
+
+    data, errors = do_graphql_call(
+        live_server, query=get_introspection_query(), expected_status=expected_status
+    )
+
+    if enabled:
+        assert errors is None
+        assert "__schema" in data
+    else:
+        assert data is None
+        error = errors[0]["message"]
+        assert "__schema" in error
+        assert "introspection is disabled" in error
+
+
+@pytest.mark.parametrize(
+    "depth_limit, expected_status, success",
+    [
+        pytest.param(4, 200, True, id="within-limit"),
+        pytest.param(3, 400, False, id="over-limit"),
+    ],
+)
+def test_graphql_query_depth_can_be_limited(
+    live_server, settings, user, depth_limit, expected_status, success
+):
+    settings.GRAPHQL_QUERY_DEPTH_LIMIT = depth_limit
+    query = """
+        {
+            myProfile {
+                emails {
+                    edges {
+                        node {
+                            id
+                        }
+                    }
+                }
+            }
+        }
+    """
+
+    data, errors = do_graphql_call_as_user(
+        live_server, user, query=query, expected_status=expected_status
+    )
+
+    if success:
+        assert errors is None
+        assert "myProfile" in data
+    else:
+        assert data is None
+        error = errors[0]["message"]
+        assert "exceeds maximum operation depth" in error

open_city_profile/views.py

@@ -2,7 +2,7 @@
 import sentry_sdk
 from django.conf import settings
 from django.core.exceptions import ObjectDoesNotExist, PermissionDenied, ValidationError
-from graphene.validation import DisableIntrospection
+from graphene.validation import depth_limit_validator, DisableIntrospection
 from graphene_django.views import GraphQLView as BaseGraphQLView
 from graphql import ExecutionResult, parse, validate
 from helusers.oidc import AuthenticationError
@@ -94,29 +94,36 @@ def _get_error_code(exception):
 
 class GraphQLView(BaseGraphQLView):
 
-    def _run_custom_validator(self, query):
+    def _run_custom_validators(self, query):
         result = None
+
+        validation_rules = [
+            depth_limit_validator(max_depth=settings.GRAPHQL_QUERY_DEPTH_LIMIT)
+        ]
+
         if not settings.ENABLE_GRAPHQL_INTROSPECTION and not settings.DEBUG:
-            try:
-                document = parse(query)
-            except Exception:
-                # Execution will also fail in super().execute_graphql_request()
-                # when parsing the query so no need to do anything here.
-                pass
-            else:
-                validation_errors = validate(
-                    schema=self.schema.graphql_schema,
-                    document_ast=document,
-                    rules=(DisableIntrospection,),
-                )
-                if validation_errors:
-                    result = ExecutionResult(data=None, errors=validation_errors)
+            validation_rules.append(DisableIntrospection)
+
+        try:
+            document = parse(query)
+        except Exception:
+            # Execution will also fail in super().execute_graphql_request()
+            # when parsing the query so no need to do anything here.
+            pass
+        else:
+            validation_errors = validate(
+                schema=self.schema.graphql_schema,
+                document_ast=document,
+                rules=validation_rules,
+            )
+            if validation_errors:
+                result = ExecutionResult(data=None, errors=validation_errors)
 
         return result
 
     def execute_graphql_request(self, request, data, query, *args, **kwargs):
         """Extract any exceptions and send some of them to Sentry"""
-        result = self._run_custom_validator(query)
+        result = self._run_custom_validators(query)
 
         if not result:
             result = super().execute_graphql_request(