[783] Add GraphQL Query Whitelist for Security

django-backend/soroscan/graphql_views.py

@@ -1,7 +1,9 @@
 """
 Custom GraphQL views with rate limiting and introspection control.
 """
+
 import json
+import logging
 
 from django.conf import settings
 from django.http import JsonResponse
@@ -10,8 +12,11 @@
 from strawberry.django.views import GraphQLView
 
 from soroscan.graphql_complexity import calculate_complexity, complexity_error_message
+from soroscan.graphql_whitelist import hash_graphql_query, normalize_graphql_query
 from soroscan.throttles import IngestRateThrottle
 
+logger = logging.getLogger(__name__)
+
 _INTROSPECTION_FIELDS = {"__schema", "__type", "__typename"}
 
 
@@ -29,6 +34,57 @@ def _is_introspection_query(body: bytes) -> bool:
     return any(field in query for field in _INTROSPECTION_FIELDS)
 
 
+def _client_ip(request) -> str | None:
+    forwarded = request.META.get("HTTP_X_FORWARDED_FOR", "")
+    if forwarded:
+        first = forwarded.split(",", maxsplit=1)[0].strip()
+        if first:
+            return first
+    return request.META.get("REMOTE_ADDR")
+
+
+def _check_query_whitelist(body: bytes, request) -> JsonResponse | None:
+    """Reject unknown queries when production whitelist enforcement is enabled."""
+    if not getattr(settings, "GRAPHQL_QUERY_WHITELIST_ENABLED", False):
+        return None
+
+    query = _parse_request_body(body).get("query", "")
+    if not query:
+        return None
+
+    from soroscan.ingest.models import GraphQLRejectedQueryLog, GraphQLWhitelistedQuery
+
+    query_hash = hash_graphql_query(query)
+    if GraphQLWhitelistedQuery.objects.filter(query_hash=query_hash).exists():
+        return None
+
+    GraphQLRejectedQueryLog.objects.create(
+        query_hash=query_hash,
+        query_preview=normalize_graphql_query(query)[:500],
+        client_ip=_client_ip(request),
+    )
+    logger.warning(
+        "Rejected unknown GraphQL query hash=%s ip=%s",
+        query_hash,
+        _client_ip(request),
+    )
+
+    return JsonResponse(
+        {
+            "errors": [
+                {
+                    "message": (
+                        "GraphQL query is not whitelisted. "
+                        "Register the query hash via the admin endpoint."
+                    ),
+                    "extensions": {"queryHash": query_hash},
+                }
+            ]
+        },
+        status=403,
+    )
+
+
 def _evaluate_query_complexity(body: bytes):
     """
     Return a ComplexityResult for POST bodies with a query, or None otherwise.
@@ -106,6 +162,10 @@ def dispatch(self, request, *args, **kwargs):
                     status=403,
                 )
 
+            whitelist_response = _check_query_whitelist(body, request)
+            if whitelist_response is not None:
+                return whitelist_response
+
             if complexity_result is not None and complexity_result.exceeded:
                 return JsonResponse(
                     {
@@ -126,8 +186,6 @@ def dispatch(self, request, *args, **kwargs):
 
         if complexity_result is not None and hasattr(response, "__setitem__"):
             response["X-GraphQL-Complexity"] = str(complexity_result.score)
-            response["X-GraphQL-Complexity-Limit"] = str(
-                complexity_result.max_allowed
-            )
+            response["X-GraphQL-Complexity-Limit"] = str(complexity_result.max_allowed)
 
         return response

django-backend/soroscan/graphql_whitelist.py

@@ -0,0 +1,24 @@
+"""
+GraphQL trusted query whitelist utilities.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+
+WHITESPACE_RE = re.compile(r"\s+")
+COMMENT_RE = re.compile(r"#.*")
+
+
+def normalize_graphql_query(query: str) -> str:
+    """Normalize a GraphQL query string for stable hashing."""
+    without_comments = COMMENT_RE.sub("", query)
+    collapsed = WHITESPACE_RE.sub(" ", without_comments).strip()
+    return collapsed
+
+
+def hash_graphql_query(query: str) -> str:
+    """Return SHA-256 hex digest for a normalized GraphQL query."""
+    normalized = normalize_graphql_query(query)
+    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()

django-backend/soroscan/ingest/admin.py

@@ -40,6 +40,8 @@
     IndexerState,
     IngestError,
     EventDeduplicationConfig,
+    GraphQLRejectedQueryLog,
+    GraphQLWhitelistedQuery,
     Organization,
     OrganizationBudget,
     OrganizationCostSnapshot,
@@ -1418,3 +1420,17 @@ def test_dedup_view(self, request, contract_id):
             json.dumps({"dedup_hash": dedup_hash, "material": material}),
             content_type="application/json",
         )
+
+
+@admin.register(GraphQLWhitelistedQuery)
+class GraphQLWhitelistedQueryAdmin(admin.ModelAdmin):
+    list_display = ["name", "query_hash", "registered_by", "created_at"]
+    search_fields = ["name", "query_hash", "query_text"]
+    readonly_fields = ["query_hash", "created_at", "registered_by"]
+
+
+@admin.register(GraphQLRejectedQueryLog)
+class GraphQLRejectedQueryLogAdmin(admin.ModelAdmin):
+    list_display = ["query_hash", "client_ip", "created_at"]
+    search_fields = ["query_hash", "query_preview"]
+    readonly_fields = ["query_hash", "query_preview", "client_ip", "created_at"]

django-backend/soroscan/ingest/migrations/0046_graphql_query_whitelist.py

@@ -0,0 +1,75 @@
+# Generated by Django 5.2.10 on 2026-06-30 04:14
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("ingest", "0045_organization_cors_origins"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="GraphQLRejectedQueryLog",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("query_hash", models.CharField(db_index=True, max_length=64)),
+                ("query_preview", models.TextField(blank=True, default="")),
+                ("client_ip", models.GenericIPAddressField(blank=True, null=True)),
+                ("created_at", models.DateTimeField(auto_now_add=True, db_index=True)),
+            ],
+            options={
+                "ordering": ["-created_at"],
+            },
+        ),
+        migrations.CreateModel(
+            name="GraphQLWhitelistedQuery",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "query_hash",
+                    models.CharField(db_index=True, max_length=64, unique=True),
+                ),
+                (
+                    "query_text",
+                    models.TextField(help_text="Original GraphQL query document"),
+                ),
+                ("name", models.CharField(blank=True, default="", max_length=128)),
+                ("description", models.TextField(blank=True, default="")),
+                ("created_at", models.DateTimeField(auto_now_add=True, db_index=True)),
+                (
+                    "registered_by",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="registered_graphql_queries",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "ordering": ["-created_at"],
+            },
+        ),
+    ]

django-backend/soroscan/ingest/models.py

@@ -2233,3 +2233,42 @@ class Meta:
 
     def __str__(self):
         return f"Blacklisted({self.contract_id[:8]}...)"
+
+
+class GraphQLWhitelistedQuery(models.Model):
+    """Approved GraphQL query stored by normalized hash."""
+
+    query_hash = models.CharField(max_length=64, unique=True, db_index=True)
+    query_text = models.TextField(help_text="Original GraphQL query document")
+    name = models.CharField(max_length=128, blank=True, default="")
+    description = models.TextField(blank=True, default="")
+    registered_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="registered_graphql_queries",
+    )
+    created_at = models.DateTimeField(auto_now_add=True, db_index=True)
+
+    class Meta:
+        ordering = ["-created_at"]
+
+    def __str__(self):
+        label = self.name or self.query_hash[:12]
+        return f"GraphQL query: {label}"
+
+
+class GraphQLRejectedQueryLog(models.Model):
+    """Audit log for GraphQL queries rejected by the production whitelist."""
+
+    query_hash = models.CharField(max_length=64, db_index=True)
+    query_preview = models.TextField(blank=True, default="")
+    client_ip = models.GenericIPAddressField(null=True, blank=True)
+    created_at = models.DateTimeField(auto_now_add=True, db_index=True)
+
+    class Meta:
+        ordering = ["-created_at"]
+
+    def __str__(self):
+        return f"Rejected GraphQL hash {self.query_hash[:12]}..."

django-backend/soroscan/ingest/tests/test_graphql_whitelist.py

@@ -0,0 +1,112 @@
+"""
+Tests for GraphQL query whitelist enforcement and admin registration.
+"""
+
+import json
+from unittest.mock import MagicMock
+
+import pytest
+from django.test import RequestFactory, override_settings
+from django.urls import reverse
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from soroscan.graphql_views import ThrottledGraphQLView, _check_query_whitelist
+from soroscan.graphql_whitelist import hash_graphql_query, normalize_graphql_query
+from soroscan.ingest.models import GraphQLRejectedQueryLog, GraphQLWhitelistedQuery
+from soroscan.ingest.tests.factories import UserFactory
+
+KNOWN_QUERY = "{ contracts { id contractId } }"
+
+
+@pytest.mark.django_db
+class TestGraphQLQueryHashing:
+    def test_normalize_collapses_whitespace(self):
+        assert (
+            normalize_graphql_query("  { contracts { id } }  ")
+            == "{ contracts { id } }"
+        )
+
+    def test_hash_is_stable(self):
+        h1 = hash_graphql_query("{ contracts { id } }")
+        h2 = hash_graphql_query("{\n  contracts { id }\n}")
+        assert h1 == h2
+
+
+@pytest.mark.django_db
+class TestGraphQLWhitelistEnforcement:
+    def _make_request(self, query: str):
+        factory = RequestFactory()
+        body = json.dumps({"query": query}).encode()
+        return factory.post("/graphql/", data=body, content_type="application/json")
+
+    def _make_view(self):
+        view = ThrottledGraphQLView(schema=MagicMock())
+        view.check_throttles = lambda r: None
+        return view
+
+    @override_settings(GRAPHQL_QUERY_WHITELIST_ENABLED=True)
+    def test_unknown_query_rejected(self):
+        request = self._make_request("{ unknownField }")
+        response = _check_query_whitelist(request.body, request)
+        assert response.status_code == 403
+
+    @override_settings(GRAPHQL_QUERY_WHITELIST_ENABLED=True)
+    def test_known_query_accepted(self):
+        GraphQLWhitelistedQuery.objects.create(
+            query_hash=hash_graphql_query(KNOWN_QUERY),
+            query_text=KNOWN_QUERY,
+            name="contracts-list",
+        )
+        request = self._make_request(KNOWN_QUERY)
+        assert _check_query_whitelist(request.body, request) is None
+
+    @override_settings(GRAPHQL_QUERY_WHITELIST_ENABLED=True)
+    def test_unknown_query_logged(self):
+        request = self._make_request("{ notRegistered }")
+        _check_query_whitelist(request.body, request)
+        assert GraphQLRejectedQueryLog.objects.count() == 1
+
+    @override_settings(GRAPHQL_QUERY_WHITELIST_ENABLED=False)
+    def test_developer_mode_allows_unknown(self):
+        request = self._make_request("{ anything }")
+        assert _check_query_whitelist(request.body, request) is None
+
+
+@pytest.mark.django_db
+class TestGraphQLWhitelistAdminEndpoint:
+    def test_staff_can_register_query(self):
+        admin = UserFactory(is_staff=True, is_superuser=True)
+        client = APIClient()
+        client.force_authenticate(user=admin)
+        url = reverse("admin-graphql-whitelist")
+
+        response = client.post(
+            url,
+            {"query": KNOWN_QUERY, "name": "contracts-list"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.data["created"] is True
+        assert GraphQLWhitelistedQuery.objects.filter(name="contracts-list").exists()
+
+    def test_non_staff_rejected(self):
+        user = UserFactory(is_staff=False)
+        client = APIClient()
+        client.force_authenticate(user=user)
+        url = reverse("admin-graphql-whitelist")
+
+        response = client.post(url, {"query": KNOWN_QUERY}, format="json")
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_register_is_idempotent(self):
+        admin = UserFactory(is_staff=True, is_superuser=True)
+        client = APIClient()
+        client.force_authenticate(user=admin)
+        url = reverse("admin-graphql-whitelist")
+
+        first = client.post(url, {"query": KNOWN_QUERY}, format="json")
+        second = client.post(url, {"query": KNOWN_QUERY}, format="json")
+        assert first.status_code == status.HTTP_201_CREATED
+        assert second.status_code == status.HTTP_200_OK
+        assert GraphQLWhitelistedQuery.objects.count() == 1