Update field suggestion detector

Author: omar2535

.github/workflows/integration_tests.yml

@@ -51,6 +51,10 @@ jobs:
       run: |
         uv run pytest tests/integration/test_api_security_api.py --exitfirst --verbose --failed-first --cov=. --cov-report html
 
+    - name: Run integration test for NoSQL and time-based SQL injection API
+      run: |
+        uv run pytest tests/integration/test_nosql_time_sql_api.py --exitfirst --verbose --failed-first --cov=. --cov-report html
+
     - name: Run tests for core
       run: |
         uv run pytest tests/integration/test_core.py --exitfirst --verbose --failed-first --cov=. --cov-report html

graphqler/fuzzer/engine/detectors/field_fuzzing/id_enumeration_detector.py

@@ -34,6 +34,7 @@
 from graphqler.utils.api import API
 from graphqler.utils.objects_bucket import ObjectsBucket
 from graphqler.utils.stats import Stats
+from graphqler.utils.response_utils import is_non_empty_result
 from graphqler.fuzzer.engine.materializers.getter import Getter
 from graphqler.fuzzer.engine.materializers.regular_payload_materializer import RegularPayloadMaterializer
 from graphqler.fuzzer.engine.detectors.detector import Detector
@@ -220,9 +221,16 @@ def _probe_ids(self, field_name: str) -> tuple[int, list[str]]:
                     self.api.url, payload
                 )
                 Stats().add_http_status_code(self.name, request_response.status_code)
-
                 if request_response.status_code == 200 and isinstance(graphql_response.get("data"), dict):
-                    if any(v is not None for v in graphql_response["data"].values()):
+                    data = graphql_response["data"]
+                    field_result = data.get(self.name)
+
+                    is_hit = (
+                        is_non_empty_result(field_result)
+                        if self.name in data
+                        else any(is_non_empty_result(v) for v in data.values())
+                    )
+                    if is_hit:
                         success_count += 1
             except Exception:
                 pass

graphqler/fuzzer/engine/detectors/field_suggestion/field_suggestion_detector.py

@@ -1,14 +1,23 @@
+import random
 from typing import Type
 
 import requests
 
+from graphqler.utils import plugins_handler
+from graphqler.utils.stats import Stats
 from .field_suggestion_materializer import FieldSuggestionMaterializer
 from ..detector import Detector
 
 
 class FieldSuggestionsDetector(Detector):
     """Field Suggestions Detector
+
+    Iterates over all known query names (shuffled) and sends each as a
+    misspelled field name.  Detection succeeds as soon as any response
+    contains a "did you mean" hint, making the result independent of which
+    individual query name happens to be blocked by a deny-list.
     """
+
     @property
     def DETECTION_NAME(self) -> str:
         return "Field Suggestions Enabled"
@@ -25,9 +34,63 @@ def detect_only_once_for_node(self) -> bool:
     def materializer(self) -> Type[FieldSuggestionMaterializer]:
         return FieldSuggestionMaterializer
 
+    # ── multi-query detect override ──────────────────────────────────────────
+
+    def detect(self) -> tuple[bool, bool]:
+        query_names = list(self.api.queries.keys())
+        random.shuffle(query_names)
+
+        for query_name in query_names:
+            misspelled = query_name + "abc"
+            payload = f"query {{\n  {misspelled} {{\n    id\n  }}\n}}"
+
+            graphql_response, request_response = (
+                plugins_handler.get_request_utils().send_graphql_request(
+                    self.api.url, payload
+                )
+            )
+            Stats().add_http_status_code(self.name, request_response.status_code)
+
+            if self._is_vulnerable(graphql_response, request_response):
+                self.payload = payload
+                self.confirmed_vulnerable = True
+                self.potentially_vulnerable = True
+                evidence = self._get_evidence(graphql_response, request_response)
+                Stats().add_vulnerability(
+                    self.DETECTION_NAME,
+                    self.name,
+                    self.confirmed_vulnerable,
+                    self.potentially_vulnerable,
+                    payload=payload,
+                    evidence=evidence,
+                )
+                self.detector_logger.info(
+                    f"Detector {self.DETECTION_NAME} finished detecting - "
+                    f"is_vulnerable: True - potentially_vulnerable: True"
+                )
+                return (True, True)
+
+        self.detector_logger.info(
+            f"Detector {self.DETECTION_NAME} finished detecting - "
+            f"is_vulnerable: False - potentially_vulnerable: False"
+        )
+        Stats().add_vulnerability(
+            self.DETECTION_NAME,
+            self.name,
+            False,
+            False,
+            payload="",
+            evidence="",
+        )
+        return (False, False)
+
+    # ── helpers ──────────────────────────────────────────────────────────────
+
     def _is_vulnerable(self, graphql_response: dict, request_response: requests.Response) -> bool:
-        return ("did you mean" in str(graphql_response['errors'][0]['message']).lower()
-                or "did you mean" in request_response.text.lower())
+        errors = graphql_response.get("errors") or []
+        if errors and "did you mean" in str(errors[0].get("message", "")).lower():
+            return True
+        return "did you mean" in request_response.text.lower()
 
     def _is_potentially_vulnerable(self, graphql_response: dict, request_response: requests.Response) -> bool:
         return self._is_vulnerable(graphql_response, request_response)

graphqler/utils/response_utils.py

@@ -0,0 +1,30 @@
+from typing import Any
+
+def is_non_empty_result(value: Any) -> bool:
+    """Checks if a GraphQL result contains any non-empty fields recursively.
+
+    Args:
+        value (Any): The value to check (dict, list, str, int, float, bool, etc.)
+
+    Returns:
+        bool: True if the result contains actual data, False if entirely empty/null.
+    """
+    if value is None:
+        return False
+
+    if isinstance(value, str):
+        # bool("") is False, bool("text") is True. 
+        # Note: use bool(value.strip()) if you want spaces like "   " to count as empty.
+        return bool(value)
+
+    if isinstance(value, dict):
+        # any() automatically returns False if the dict is empty
+        return any(is_non_empty_result(v) for v in value.values())
+
+    if isinstance(value, list):
+        # any() automatically returns False if the list is empty
+        return any(is_non_empty_result(item) for item in value)
+
+    # Fallback for ints, floats, and booleans (e.g., 0, 0.0, False).
+    # In GraphQL/JSON, these represent actual data points, so they are non-empty.
+    return True

tests/integration/test_nosql_time_sql_api.py

@@ -66,10 +66,14 @@ def test_fuzz_generates_stats_file(self):
 
     # ── Injection detection ───────────────────────────────────────────────────
 
+    _cached_vulns = None
+
     def _run_and_get_vulns(self):
-        __main__.run_compile_mode(self.PATH, self.URL)
-        __main__.run_fuzz_mode(self.PATH, self.URL)
-        return get_vulnerabilities_from_stats(self.PATH)
+        if self.__class__._cached_vulns is None:
+            __main__.run_compile_mode(self.PATH, self.URL)
+            __main__.run_fuzz_mode(self.PATH, self.URL)
+            self.__class__._cached_vulns = get_vulnerabilities_from_stats(self.PATH)
+        return self.__class__._cached_vulns
 
     def test_nosql_injection_detected(self):
         vulns = self._run_and_get_vulns()