Build a GraphQL Security Auditor
Build a Python CLI tool that audits GraphQL endpoints for 6 common vulnerabilities.
Introduction
30 Second Summary
GraphQL APIs give clients the power to ask for exactly the data they need. That same flexibility also gives attackers the power to craft deeply nested queries, enumerate schemas, and bypass rate limits.
In this project, you will build a Python command-line tool that audits GraphQL endpoints for 6 common security vulnerabilities. You'll test it against a deliberately insecure GraphQL application running locally in Docker.
What You'll Build
A command-line GraphQL Security Auditor written in Python that scans any GraphQL endpoint and reports on common misconfigurations and vulnerabilities.
By the end of this project, you'll have:
- A working Python CLI tool that detects 6 common GraphQL security vulnerabilities including introspection exposure, query depth attacks, and CSRF risk.
- A local vulnerable GraphQL target (DVGA) running in Docker for safe, legal security testing.
- A color-coded terminal report showing severity-rated findings with curl reproduction commands.
- Secret Mission: Add circular fragment DoS detection that tests for a real-world CVE pattern.
Are there any prerequisites?
You should be comfortable with basic Python (functions, variables, and working with JSON). A general understanding of how APIs work (sending requests and receiving responses) is helpful but not required.
Before We Start
To build a security auditing tool that probes GraphQL endpoints, you need a few pieces in place.
You need Python to write the scanner, Docker to run a vulnerable test target, and a project directory with the right dependencies installed.
In this step, get ready to:
- Install Python 3.10+ and Docker.
- Launch a vulnerable GraphQL application (DVGA) as your test target.
- Create a project directory with a virtual environment and dependencies.
Install Python and Docker
Python powers the scanner script, and Docker lets you spin up a deliberately vulnerable GraphQL server to test against safely.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Check your Python version by running this command:
python3 --version✔️ I see version 3.10 or higher
You're all set. Your Python version is compatible with this project.
ⓧ I see an older version
You have Python installed, but you need version 3.10 or higher for this project.
- Visit python.org/downloads and download the latest Python 3 release.
- Run the installer and follow the prompts.
- On macOS, the system Python may still resolve to the older version after installing. Use python3.10 or python3.12 (matching the version you installed) to ensure you are using the correct binary.
- Run python3 --version again to confirm you see 3.10 or higher.
Still seeing the old version?
On macOS, the system default python3 may point to an older release even after installing a newer one. Try running the version-specific binary directly (e.g. python3.12 --version). If that works, use that binary for the rest of this project.
Why isn't my Python version updating?
ⓧ Command not found
Python is not installed on your machine yet.
- Visit python.org/downloads and download the latest Python 3 release for your operating system.
- Run the installer and follow the prompts.
- On Windows, make sure to check the Add Python to PATH option during installation.
- Close and reopen your terminal, then run python3 --version again to confirm you see 3.10 or higher.
Still getting command not found?
On Windows, try python --version (without the 3). If that works and shows 3.10+, use python instead of python3 for the rest of this project.
Help me troubleshoot Python installation
- Check that Docker is installed by running this command:
docker --version✔️ I see a version number
Docker is ready to go.
ⓧ Command not found
Docker is not installed yet.
- Visit docs.docker.com/get-started/get-docker and download Docker Desktop for your operating system.
- Run the installer and follow the prompts.
- Once installed, open Docker Desktop and wait for it to finish starting up.
- Close and reopen your terminal, then run docker --version again to confirm you see a version number.
Docker command still not found?
Make sure Docker Desktop is running. On macOS and Windows, the docker CLI only works while Docker Desktop is active.
Help me troubleshoot Docker installation
Pull and run your vulnerable test target
DVGA (Damn Vulnerable GraphQL Application) is a deliberately insecure GraphQL server designed for learning and testing. You will run it locally in a Docker container so you have a safe target to scan.
What is DVGA?
DVGA is an open-source project built specifically for security researchers and learners. It exposes common GraphQL vulnerabilities by design, including unrestricted introspection, no depth limits, and batch query support.
Running it inside Docker means it stays isolated on your machine. No real data is at risk.
- Make sure Docker Desktop is running.
- Pull and start the DVGA container by running this command:
docker run -d -t -p 5013:5013 -e WEB_HOST=0.0.0.0 --name dvga dolevf/dvgaWhat does this command do?
- docker run creates and starts a new container from the dolevf/dvga image.
- -d runs the container in the background (detached mode).
- -t allocates a terminal for the container process.
- -p 5013:5013 maps port 5013 on your machine to port 5013 inside the container.
- -e WEB_HOST=0.0.0.0 tells DVGA to listen on all network interfaces.
- --name dvga gives the container a friendly name so you can manage it easily later.
- Wait for Docker to pull the image and start the container. This may take a minute on first run.
- Open your browser and navigate to http://localhost:5013 to confirm DVGA is running.
You should see the DVGA web interface load in your browser.
Page not loading?
Give the container a few seconds to start up. If the page still does not load, check that Docker Desktop is running and try the URL again.
You can also verify the container is running by executing docker ps in your terminal. You should see a row with the name dvga and status Up.
Help me troubleshoot DVGA not loading
Create your project directory and install dependencies
Now that your test target is running, you need a place to write the auditor code and a virtual environment to keep your Python dependencies isolated.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir graphql-auditor
cd graphql-auditor- Create a Python virtual environment by running:
python3 -m venv venvWhat is a virtual environment?
A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay separate from your system Python, preventing version conflicts between projects.
- Activate the virtual environment:
macOS
source venv/bin/activateWindows
venv\Scripts\activateYou should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.
- Install the requests library by running:
pip install requests==2.34.2Why requests?
The requests library provides a simple HTTP interface for sending custom payloads to a GraphQL endpoint. Since you are testing the raw HTTP layer (not using a GraphQL client), requests gives you full control over how queries are sent.
- Confirm the install succeeded by running:
pip show requestsYou should see output showing Name: requests and Version: 2.34.2.
Seeing a permission error or pip not found?
Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If not, re-run the activation command for your platform.
If pip is not found, try pip3 instead.
Help me fix my pip install error
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code (or Cursor if you use Cursor) and press Enter to open it.
- Click File in the top menu bar.
- Click Open Folder.
- Select the graphql-auditor folder on your Desktop, then click Open.
Your environment is ready. Next up, you will build the first security check and start probing DVGA for vulnerabilities.
Build the Introspection Check
Your environment is ready and DVGA is waiting to be probed. Now it's time to build the core of your security auditor.
The first vulnerability any GraphQL security scanner checks for is introspection. When introspection is enabled, a single query exposes the entire API schema to anyone who can reach the endpoint. An attacker can map every type, field, and mutation without guessing.
In this step, get ready to:
- Create a base GraphQLAuditor class that sends queries to any target endpoint.
- Implement an introspection check that detects exposed schemas.
- Run the check against DVGA and confirm the vulnerability is detected.
Create the auditor file with imports and the class constructor
Your auditor needs a class that manages the target URL, sends HTTP requests, and collects findings. You will build this foundation first and add individual security checks on top.
- In VS Code, create a new file called graphql_auditor.py in your project directory.
- Add the following imports and class constructor:
import requests
import sys
import time
from report import SecurityReport
class GraphQLAuditor:
def __init__(self, target_url, timeout=10):
# Store the target endpoint and configure defaults
self.target_url = target_url
self.timeout = timeout
self.report = SecurityReport(target_url)
self.headers = {"Content-Type": "application/json"}What does this code do?
The imports bring in requests for making HTTP calls, sys for command-line arguments, time for timing checks, and SecurityReport from a report module you will build in Step 5.
The GraphQLAuditor class stores the target URL, a timeout for requests, a report object for collecting findings, and the JSON content-type header that GraphQL endpoints expect.
Why does this import SecurityReport if report.py doesn't exist yet?
You are building the auditor's logic first and the reporting module later. The import line is there so you can wire everything together without editing imports in Step 5. The file will not be runnable on its own until you create report.py in a later step.
Add the send_query method
Every security check needs to send a crafted GraphQL query to the target and inspect the response. This shared method handles POST and GET requests with error handling for timeouts and connection failures.
- Below the __init__ method, add the send_query method:
def send_query(self, query, variables=None, method="POST"):
payload = {"query": query}
if variables:
payload["variables"] = variables
try:
if method == "POST":
response = requests.post(
self.target_url,
json=payload,
headers=self.headers,
timeout=self.timeout,
)
else:
response = requests.get(
self.target_url,
params={"query": query},
timeout=self.timeout,
)
return response
except requests.exceptions.Timeout:
return None
except requests.exceptions.ConnectionError:
return NoneWhat does this method do?
- Wraps the query string in the JSON payload format that GraphQL servers expect.
- Supports both POST (standard) and GET (needed for CSRF testing later).
- Returns None on timeout or connection error instead of crashing. This lets each check gracefully handle unresponsive targets.
Add the get_curl_command helper
Security reports are most useful when they include reproduction steps. This helper generates a ready-to-paste curl command for each finding so anyone can verify the vulnerability independently.
- Below send_query, add the get_curl_command method:
def get_curl_command(self, query, method="POST"):
if method == "POST":
escaped_query = query.replace('"', '\\"').replace("\n", " ")
return (
f'curl -X POST -H "Content-Type: application/json" '
f'-d \'{{"query": "{escaped_query}"}}\' '
f"'{self.target_url}'"
)
else:
import urllib.parse
encoded_query = urllib.parse.quote(query)
return f"curl '{self.target_url}?query={encoded_query}'"What does this method do?
- For POST requests, it escapes quotes and newlines in the query, then wraps it in a valid curl command with the correct headers.
- For GET requests, it URL-encodes the query and appends it as a query parameter.
- The output is a command anyone on the team can paste directly into a terminal to reproduce the finding.
- Save graphql_auditor.py.
Seeing a syntax error on save?
Make sure the send_query and get_curl_command methods are indented one level inside the GraphQLAuditor class (4 spaces). Python uses indentation to determine class membership.
help me fix indentation errors in my GraphQLAuditor class
Implement the introspection check
Introspection is the most fundamental GraphQL security check. The __schema meta-field is part of the GraphQL specification, and every major implementation enables it by default. When left exposed in production, a single query hands attackers a complete map of the API surface.
- Below get_curl_command, add the check_introspection method:
def check_introspection(self):
print("[*] Testing: Introspection Query...")
query = """
query {
__schema {
types {
name
fields {
name
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data and "__schema" in data["data"]:
types = data["data"]["__schema"]["types"]
type_count = len(types)
self.report.add_finding(
title="Introspection Query Enabled",
severity="HIGH",
description=(
f"Full schema introspection is enabled. "
f"Discovered {type_count} types. "
f"Attackers can map the entire API surface."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Introspection Query",
severity="PASS",
description="Introspection is disabled or restricted.",
curl_command=None,
)
return FalseWhat does this check do?
- Sends the standard __schema introspection query asking for all type names and their fields.
- If the server responds with a full schema dump (status 200 with __schema in the response body), it flags a HIGH severity finding.
- The finding includes a count of discovered types and a curl command for reproduction.
- If introspection is blocked, it records a PASS result instead.
Why is introspection a HIGH severity finding?
When introspection is enabled in production, an attacker can discover every queryable type, mutation, input object, and enum with a single request. Tools like InQL can take that output and automatically generate attack payloads.
The schema alone reveals internal architecture, field names, and relationships. Even if data access requires authentication, the schema gives attackers a complete map of what to probe.
- Save graphql_auditor.py.
✔️ Awesome, I've got everything!
Great. Make sure you have saved graphql_auditor.py with all four sections: imports, __init__, send_query, get_curl_command, and check_introspection.
ⓧ I'd like to double check the full code
import requests
import sys
import time
from report import SecurityReport
class GraphQLAuditor:
def __init__(self, target_url, timeout=10):
self.target_url = target_url
self.timeout = timeout
self.report = SecurityReport(target_url)
self.headers = {"Content-Type": "application/json"}
def send_query(self, query, variables=None, method="POST"):
payload = {"query": query}
if variables:
payload["variables"] = variables
try:
if method == "POST":
response = requests.post(
self.target_url,
json=payload,
headers=self.headers,
timeout=self.timeout,
)
else:
response = requests.get(
self.target_url,
params={"query": query},
timeout=self.timeout,
)
return response
except requests.exceptions.Timeout:
return None
except requests.exceptions.ConnectionError:
return None
def get_curl_command(self, query, method="POST"):
if method == "POST":
escaped_query = query.replace('"', '\\"').replace("\n", " ")
return (
f'curl -X POST -H "Content-Type: application/json" '
f'-d \'{{"query": "{escaped_query}"}}\' '
f"'{self.target_url}'"
)
else:
import urllib.parse
encoded_query = urllib.parse.quote(query)
return f"curl '{self.target_url}?query={encoded_query}'"
def check_introspection(self):
print("[*] Testing: Introspection Query...")
query = """
query {
__schema {
types {
name
fields {
name
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data and "__schema" in data["data"]:
types = data["data"]["__schema"]["types"]
type_count = len(types)
self.report.add_finding(
title="Introspection Query Enabled",
severity="HIGH",
description=(
f"Full schema introspection is enabled. "
f"Discovered {type_count} types. "
f"Attackers can map the entire API surface."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Introspection Query",
severity="PASS",
description="Introspection is disabled or restricted.",
curl_command=None,
)
return FalseTest the introspection check against DVGA
Since report.py does not exist yet, you cannot run graphql_auditor.py directly. Instead, you will test the introspection logic interactively in a Python session to confirm DVGA's schema is exposed.
- Open a terminal in your project directory and start a Python session by running:
python3 -c "import requests; r = requests.post('http://localhost:5013/graphql', json={'query': '{ __schema { types { name } } }'}, headers={'Content-Type': 'application/json'}); data = r.json(); print(f'Introspection exposed! Found {len(data[\"data\"][\"__schema\"][\"types\"])} types') if '__schema' in data.get('data', {}) else print('Introspection blocked')"- You should see output similar to: Introspection exposed! Found followed by a number of types (the exact count depends on the DVGA version).
What does this confirm?
This quick test proves that DVGA's GraphQL endpoint at http://localhost:5013/graphql has introspection enabled and will return its full schema to any unauthenticated request. This is exactly what your check_introspection method detects.
Getting a connection error?
- Make sure your DVGA container is still running. Check with docker ps and look for the dvga container. If it is not listed, restart it with docker start dvga.
- Make sure your virtual environment is activated. You should see the environment name in your terminal prompt.
- Verify you are hitting the correct URL. The GraphQL endpoint is at /graphql, not the root path.
help me troubleshoot my DVGA connection error when testing introspection
Your auditor's foundation is in place and you have confirmed that DVGA exposes its schema. Next up, you will add checks that detect denial-of-service vulnerabilities.
Detect DoS Vulnerabilities
Your introspection check from the previous step reveals the shape of a GraphQL API. Now you need to find out if an attacker can overwhelm it.
GraphQL's flexibility is a double-edged sword. A single request can ask for deeply nested data, duplicate the same field hundreds of times, or bundle dozens of operations into one HTTP call. If the server doesn't enforce limits, any of these becomes a denial-of-service vector.
In this step, get ready to:
- Implement a query depth check that tests for unbounded nesting.
- Implement an alias overloading check that tests field repetition limits.
- Implement a batch query check that tests multi-operation abuse.
Implement the query depth check
GraphQL allows clients to request nested relationships to arbitrary depth. If a server has no depth limit, an attacker can craft a single query that expands exponentially and exhausts server memory or CPU.
- In graphql_auditor.py, find the end of the check_introspection() method (the final return False line).
- Add the following method directly below it:
def check_query_depth(self):
print("[*] Testing: Query Depth Limit...")
# Send a 7-level nested query using pastes/owner relationship
query = """
query {
pastes {
owner {
pastes {
owner {
pastes {
owner {
name
}
}
}
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
# Server accepted the deep query without errors
if "data" in data and "errors" not in data:
self.report.add_finding(
title="No Query Depth Limit",
severity="HIGH",
description=(
"Server accepts deeply nested queries (7 levels) "
"without restriction. Vulnerable to depth-based "
"denial of service attacks."
),
curl_command=self.get_curl_command(query),
)
return True
# Server returned partial data with errors (still vulnerable)
elif "errors" in data and "data" in data:
self.report.add_finding(
title="No Query Depth Limit",
severity="HIGH",
description=(
"Server processes deeply nested queries (7 levels) "
"and returns partial data. Vulnerable to depth-based "
"denial of service attacks."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Query Depth Limit",
severity="PASS",
description="Server enforces query depth limits.",
curl_command=None,
)
return FalseWhat does this code do?
DVGA has a pastes type where each paste has an owner, and each owner has pastes. This circular relationship lets you nest 7 levels deep in a single query.
The method handles two vulnerable scenarios: the server returning clean data (no errors at all), or the server returning partial data alongside errors. Both mean the server attempted to resolve the deep query instead of rejecting it outright.
- Save graphql_auditor.py.
- You won't see output yet. The next chunk adds another DoS check that works alongside this one.
Seeing a red underline under check_query_depth?
Make sure the method is indented at the same level as check_introspection() (one level inside the GraphQLAuditor class). Python uses indentation to determine class membership.
help me fix indentation errors in my GraphQLAuditor class
Implement the alias overloading check
GraphQL aliases let you request the same field multiple times under different names in a single query. An attacker can pack 100+ aliases into one operation to amplify server load and bypass per-field rate limiting.
- In graphql_auditor.py, add the following method directly below check_query_depth():
def check_alias_overloading(self):
print("[*] Testing: Alias Overloading...")
# Generate 100 aliases of __typename in a single query
aliases = " ".join(
[f"alias{i}: __typename" for i in range(100)]
)
query = f"query {{ {aliases} }}"
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data:
alias_count = len(data["data"])
# Server resolved all 100 aliases without restriction
if alias_count >= 100:
self.report.add_finding(
title="Alias Overloading Allowed",
severity="HIGH",
description=(
f"Server accepted {alias_count} aliases in a "
f"single query. Attackers can use aliases to "
f"bypass rate limiting and amplify attacks."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Alias Overloading",
severity="PASS",
description="Server restricts alias count.",
curl_command=None,
)
return FalseWhat does this code do?
The method dynamically generates a query containing 100 aliases like alias0: __typename, alias1: __typename, and so on. Each alias forces the server to resolve the same field again.
If the response data contains 100+ keys, the server didn't restrict alias count. In a real attack, an attacker could request expensive computed fields hundreds of times in one request.
Why is __typename used here?
The __typename field is a built-in meta-field available on every GraphQL type. It's cheap to resolve and always present, making it a reliable test target. The vulnerability isn't about what field you alias. It's about how many aliases the server allows.
- Save graphql_auditor.py.
- You won't see output yet. The next chunk adds the final DoS detection method.
Seeing a syntax error on the f-string?
The line query = f"query {{ {aliases} }}" uses double curly braces {{ and }} to produce literal braces in the output. Make sure you have exactly two braces on each side of {aliases}.
help me fix the f-string syntax in check_alias_overloading
Implement the batch query check
Many GraphQL servers accept an array of operations in a single HTTP request. If batching is unrestricted, an attacker can bundle thousands of queries (login attempts, data fetches) into one request, completely bypassing per-request rate limits.
- In graphql_auditor.py, add the following method directly below check_alias_overloading():
def check_batch_queries(self):
print("[*] Testing: Batch Query Support...")
# Send an array of 10 operations in a single HTTP request
batch_payload = [
{"query": "query { __typename }"}
for _ in range(10)
]
try:
response = requests.post(
self.target_url,
json=batch_payload,
headers=self.headers,
timeout=self.timeout,
)
if response and response.status_code == 200:
data = response.json()
# Server returned a list of 10+ results (batching is enabled)
if isinstance(data, list) and len(data) >= 10:
self.report.add_finding(
title="Batch Queries Accepted",
severity="MEDIUM",
description=(
f"Server processes {len(data)} batched operations "
f"in a single HTTP request. Attackers can bypass "
f"per-request rate limiting."
),
curl_command=(
f'curl -X POST -H "Content-Type: application/json" '
f"-d '[{{\"query\": \"query {{ __typename }}\"}},"
f"{{\"query\": \"query {{ __typename }}\"}}]' "
f"'{self.target_url}'"
),
)
return True
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
pass
self.report.add_finding(
title="Batch Queries",
severity="PASS",
description="Server does not accept batched operations.",
curl_command=None,
)
return FalseWhat does this code do?
Instead of sending a single {"query": "..."} object, this method sends a JSON array of 10 operation objects. If the server processes all 10 and returns a list of 10 results, batching is enabled.
This check uses requests.post() directly instead of the send_query() helper because the payload format is different. Batch requests send a raw array, not a single query object.
Why is this rated MEDIUM instead of HIGH?
Batch queries enable rate-limit bypass, but they don't directly crash a server the way unbounded depth or alias overloading can. The impact depends on what operations an attacker batches (login attempts, expensive mutations). It's still a significant finding, but the immediate risk is lower than a direct resource-exhaustion attack.
- Save graphql_auditor.py.
- You won't see output yet. In Step 5, you'll wire all checks together and run the full audit against DVGA.
Unsure about the try/except structure?
The try/except block catches both Timeout and ConnectionError because batch requests can overwhelm fragile servers. Make sure the self.report.add_finding() call for the PASS case sits outside the try block, at the same indentation level as try itself.
help me fix the try/except structure in check_batch_queries
✔️ Awesome, I've got everything!
Great. Double check you've saved graphql_auditor.py with all three new methods in place.
ⓧ I'd like to double check the full code
Here's the complete graphql_auditor.py as it should look at the end of this step:
import requests
import sys
import time
from report import SecurityReport
class GraphQLAuditor:
def __init__(self, target_url, timeout=10):
self.target_url = target_url
self.timeout = timeout
self.report = SecurityReport(target_url)
self.headers = {"Content-Type": "application/json"}
def send_query(self, query, variables=None, method="POST"):
payload = {"query": query}
if variables:
payload["variables"] = variables
try:
if method == "POST":
response = requests.post(
self.target_url,
json=payload,
headers=self.headers,
timeout=self.timeout,
)
else:
response = requests.get(
self.target_url,
params={"query": query},
timeout=self.timeout,
)
return response
except requests.exceptions.Timeout:
return None
except requests.exceptions.ConnectionError:
return None
def get_curl_command(self, query, method="POST"):
if method == "POST":
escaped_query = query.replace('"', '\\"').replace("\n", " ")
return (
f'curl -X POST -H "Content-Type: application/json" '
f'-d \'{{"query": "{escaped_query}"}}\' '
f"'{self.target_url}'"
)
else:
import urllib.parse
encoded_query = urllib.parse.quote(query)
return f"curl '{self.target_url}?query={encoded_query}'"
def check_introspection(self):
print("[*] Testing: Introspection Query...")
query = """
query {
__schema {
types {
name
fields {
name
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data and "__schema" in data["data"]:
types = data["data"]["__schema"]["types"]
type_count = len(types)
self.report.add_finding(
title="Introspection Query Enabled",
severity="HIGH",
description=(
f"Full schema introspection is enabled. "
f"Discovered {type_count} types. "
f"Attackers can map the entire API surface."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Introspection Query",
severity="PASS",
description="Introspection is disabled or restricted.",
curl_command=None,
)
return False
def check_query_depth(self):
print("[*] Testing: Query Depth Limit...")
query = """
query {
pastes {
owner {
pastes {
owner {
pastes {
owner {
name
}
}
}
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data and "errors" not in data:
self.report.add_finding(
title="No Query Depth Limit",
severity="HIGH",
description=(
"Server accepts deeply nested queries (7 levels) "
"without restriction. Vulnerable to depth-based "
"denial of service attacks."
),
curl_command=self.get_curl_command(query),
)
return True
elif "errors" in data and "data" in data:
self.report.add_finding(
title="No Query Depth Limit",
severity="HIGH",
description=(
"Server processes deeply nested queries (7 levels) "
"and returns partial data. Vulnerable to depth-based "
"denial of service attacks."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Query Depth Limit",
severity="PASS",
description="Server enforces query depth limits.",
curl_command=None,
)
return False
def check_alias_overloading(self):
print("[*] Testing: Alias Overloading...")
aliases = " ".join(
[f"alias{i}: __typename" for i in range(100)]
)
query = f"query {{ {aliases} }}"
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data:
alias_count = len(data["data"])
if alias_count >= 100:
self.report.add_finding(
title="Alias Overloading Allowed",
severity="HIGH",
description=(
f"Server accepted {alias_count} aliases in a "
f"single query. Attackers can use aliases to "
f"bypass rate limiting and amplify attacks."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Alias Overloading",
severity="PASS",
description="Server restricts alias count.",
curl_command=None,
)
return False
def check_batch_queries(self):
print("[*] Testing: Batch Query Support...")
batch_payload = [
{"query": "query { __typename }"}
for _ in range(10)
]
try:
response = requests.post(
self.target_url,
json=batch_payload,
headers=self.headers,
timeout=self.timeout,
)
if response and response.status_code == 200:
data = response.json()
if isinstance(data, list) and len(data) >= 10:
self.report.add_finding(
title="Batch Queries Accepted",
severity="MEDIUM",
description=(
f"Server processes {len(data)} batched operations "
f"in a single HTTP request. Attackers can bypass "
f"per-request rate limiting."
),
curl_command=(
f'curl -X POST -H "Content-Type: application/json" '
f"-d '[{{\"query\": \"query {{ __typename }}\"}},"
f"{{\"query\": \"query {{ __typename }}\"}}]' "
f"'{self.target_url}'"
),
)
return True
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
pass
self.report.add_finding(
title="Batch Queries",
severity="PASS",
description="Server does not accept batched operations.",
curl_command=None,
)
return FalseYour auditor can now detect three DoS vectors that exploit GraphQL's flexible query language. Next up, you'll add checks for information leakage and CSRF risk to round out the vulnerability detection.
Detect Information Leakage and CSRF Risk
Your auditor can now detect three types of DoS vulnerabilities. Those checks test whether a server can handle abuse at scale.
In this step, you'll tackle a different class of risk: information leakage and cross-site request forgery. These are subtler. Instead of overwhelming the server, they exploit small misconfigurations that leak data or let attackers trick authenticated users into running queries.
In this step, get ready to:
- Implement a field suggestion detection check that catches schema enumeration leaks.
- Implement a GET-based query check that identifies CSRF risk.
Implement the field suggestion check
Even when introspection is disabled, many GraphQL servers still leak schema information through error messages. When you send a query with a misspelled field name, the server helpfully responds with "Did you mean" suggestions. Attackers use this to enumerate the schema one field at a time.
- In graphql_auditor.py, scroll down to the end of the check_batch_queries() method.
- Add the following check_field_suggestions() method below it:
def check_field_suggestions(self):
print("[*] Testing: Field Suggestions...")
# Send a query with a deliberately misspelled field name
query = "query { __typena }"
response = self.send_query(query)
if response and response.status_code in (200, 400):
try:
data = response.json()
response_text = str(data)
# Check if error message reveals valid field names
if "Did you mean" in response_text or "did you mean" in response_text:
self.report.add_finding(
title="Field Suggestions Enabled",
severity="LOW",
description=(
"Server returns field name suggestions in error "
"messages. Attackers can enumerate the schema even "
"if introspection is disabled."
),
curl_command=self.get_curl_command(query),
)
return True
except ValueError:
pass
self.report.add_finding(
title="Field Suggestions",
severity="PASS",
description="Server does not expose field name suggestions.",
curl_command=None,
)
return FalseWhat does this code do?
This method sends a deliberately broken query using __typena (a misspelling of __typename). If the server responds with hints like "Did you mean __typename?", it confirms that field suggestions are active.
This is rated LOW severity because it only leaks field names, not data. However, it enables schema enumeration even when introspection is disabled, making it a stepping stone for more targeted attacks.
- Save graphql_auditor.py.
- You won't see a change yet. The next step wires all checks together with the reporting module.
Seeing a syntax error?
Make sure the method is indented at the same level as check_batch_queries() above it (one level inside the class body). Python uses indentation to determine class membership.
Also check that response.status_code in (200, 400) uses parentheses, not square brackets. This checks both success and error HTTP codes because some servers return 400 for invalid queries while still including suggestion text.
Help me fix a syntax error in my check_field_suggestions method.
Implement the GET-based query check
GraphQL APIs typically use POST requests. When a server also accepts queries via HTTP GET, it opens the door to CSRF attacks. An attacker can embed a GraphQL query in an image tag or link, and an authenticated user's browser will execute it automatically.
- In graphql_auditor.py, add the following check_get_based_queries() method below check_field_suggestions():
def check_get_based_queries(self):
print("[*] Testing: GET-based Queries (CSRF)...")
# A simple query to test if GET method is accepted
query = "query { __typename }"
# Use the send_query helper with method="GET"
response = self.send_query(query, method="GET")
if response and response.status_code == 200:
try:
data = response.json()
if "data" in data and "__typename" in data["data"]:
self.report.add_finding(
title="GET-based Queries Accepted",
severity="HIGH",
description=(
"Server processes GraphQL queries via HTTP GET. "
"This enables Cross-Site Request Forgery (CSRF) "
"attacks against authenticated users."
),
curl_command=self.get_curl_command(query, method="GET"),
)
return True
except ValueError:
pass
self.report.add_finding(
title="GET-based Queries",
severity="PASS",
description="Server does not accept queries via GET method.",
curl_command=None,
)
return FalseWhat does this code do?
This method reuses the send_query() helper with method="GET" to send a GraphQL query as a URL parameter instead of a POST body. If the server processes it and returns valid data, the endpoint is vulnerable to CSRF.
This is rated HIGH because CSRF attacks can execute mutations (like changing passwords or deleting data) without the user's knowledge. The curl command in the finding uses the GET format so security teams can reproduce the issue.
Why is GET dangerous for GraphQL?
Browsers automatically attach cookies to GET requests for any domain. If a GraphQL server accepts queries via GET, an attacker can craft a URL like https://target.com/graphql?query={deleteAccount} and embed it in a page the victim visits. The victim's browser sends their session cookie along with the malicious query.
- Save graphql_auditor.py.
- You won't see a change yet. The next step creates the reporting module and runs the full audit.
Not sure where to place the method?
The check_get_based_queries() method must be at the same indentation level as all other check_ methods inside the GraphQLAuditor class. It goes directly after the last line of check_field_suggestions().
Help me figure out where to place my check_get_based_queries method.
✔️ Awesome, I've got everything!
Great. Double check you've saved graphql_auditor.py with both new methods.
ⓧ I'd like to double check the full code
Here is the full graphql_auditor.py at this point:
import requests
import sys
import time
from report import SecurityReport
class GraphQLAuditor:
def __init__(self, target_url, timeout=10):
self.target_url = target_url
self.timeout = timeout
self.report = SecurityReport(target_url)
self.headers = {"Content-Type": "application/json"}
def send_query(self, query, variables=None, method="POST"):
payload = {"query": query}
if variables:
payload["variables"] = variables
try:
if method == "POST":
response = requests.post(
self.target_url,
json=payload,
headers=self.headers,
timeout=self.timeout,
)
else:
response = requests.get(
self.target_url,
params={"query": query},
timeout=self.timeout,
)
return response
except requests.exceptions.Timeout:
return None
except requests.exceptions.ConnectionError:
return None
def get_curl_command(self, query, method="POST"):
if method == "POST":
escaped_query = query.replace('"', '\\"').replace("\n", " ")
return (
f'curl -X POST -H "Content-Type: application/json" '
f'-d \'{{"query": "{escaped_query}"}}\' '
f"'{self.target_url}'"
)
else:
import urllib.parse
encoded_query = urllib.parse.quote(query)
return f"curl '{self.target_url}?query={encoded_query}'"
def check_introspection(self):
print("[*] Testing: Introspection Query...")
query = """
query {
__schema {
types {
name
fields {
name
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data and "__schema" in data["data"]:
types = data["data"]["__schema"]["types"]
type_count = len(types)
self.report.add_finding(
title="Introspection Query Enabled",
severity="HIGH",
description=(
f"Full schema introspection is enabled. "
f"Discovered {type_count} types. "
f"Attackers can map the entire API surface."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Introspection Query",
severity="PASS",
description="Introspection is disabled or restricted.",
curl_command=None,
)
return False
def check_query_depth(self):
print("[*] Testing: Query Depth Limit...")
query = """
query {
pastes {
owner {
pastes {
owner {
pastes {
owner {
name
}
}
}
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data and "errors" not in data:
self.report.add_finding(
title="No Query Depth Limit",
severity="HIGH",
description=(
"Server accepts deeply nested queries (7 levels) "
"without restriction. Vulnerable to depth-based "
"denial of service attacks."
),
curl_command=self.get_curl_command(query),
)
return True
elif "errors" in data and "data" in data:
self.report.add_finding(
title="No Query Depth Limit",
severity="HIGH",
description=(
"Server processes deeply nested queries (7 levels) "
"and returns partial data. Vulnerable to depth-based "
"denial of service attacks."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Query Depth Limit",
severity="PASS",
description="Server enforces query depth limits.",
curl_command=None,
)
return False
def check_alias_overloading(self):
print("[*] Testing: Alias Overloading...")
aliases = " ".join(
[f"alias{i}: __typename" for i in range(100)]
)
query = f"query {{ {aliases} }}"
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data:
alias_count = len(data["data"])
if alias_count >= 100:
self.report.add_finding(
title="Alias Overloading Allowed",
severity="HIGH",
description=(
f"Server accepted {alias_count} aliases in a "
f"single query. Attackers can use aliases to "
f"bypass rate limiting and amplify attacks."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Alias Overloading",
severity="PASS",
description="Server restricts alias count.",
curl_command=None,
)
return False
def check_batch_queries(self):
print("[*] Testing: Batch Query Support...")
batch_payload = [
{"query": "query { __typename }"}
for _ in range(10)
]
try:
response = requests.post(
self.target_url,
json=batch_payload,
headers=self.headers,
timeout=self.timeout,
)
if response and response.status_code == 200:
data = response.json()
if isinstance(data, list) and len(data) >= 10:
self.report.add_finding(
title="Batch Queries Accepted",
severity="MEDIUM",
description=(
f"Server processes {len(data)} batched operations "
f"in a single HTTP request. Attackers can bypass "
f"per-request rate limiting."
),
curl_command=(
f'curl -X POST -H "Content-Type: application/json" '
f"-d '[{{\"query\": \"query {{ __typename }}\"}},"
f"{{\"query\": \"query {{ __typename }}\"}}]' "
f"'{self.target_url}'"
),
)
return True
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
pass
self.report.add_finding(
title="Batch Queries",
severity="PASS",
description="Server does not accept batched operations.",
curl_command=None,
)
return False
def check_field_suggestions(self):
print("[*] Testing: Field Suggestions...")
query = "query { __typena }"
response = self.send_query(query)
if response and response.status_code in (200, 400):
try:
data = response.json()
response_text = str(data)
if "Did you mean" in response_text or "did you mean" in response_text:
self.report.add_finding(
title="Field Suggestions Enabled",
severity="LOW",
description=(
"Server returns field name suggestions in error "
"messages. Attackers can enumerate the schema even "
"if introspection is disabled."
),
curl_command=self.get_curl_command(query),
)
return True
except ValueError:
pass
self.report.add_finding(
title="Field Suggestions",
severity="PASS",
description="Server does not expose field name suggestions.",
curl_command=None,
)
return False
def check_get_based_queries(self):
print("[*] Testing: GET-based Queries (CSRF)...")
query = "query { __typename }"
response = self.send_query(query, method="GET")
if response and response.status_code == 200:
try:
data = response.json()
if "data" in data and "__typename" in data["data"]:
self.report.add_finding(
title="GET-based Queries Accepted",
severity="HIGH",
description=(
"Server processes GraphQL queries via HTTP GET. "
"This enables Cross-Site Request Forgery (CSRF) "
"attacks against authenticated users."
),
curl_command=self.get_curl_command(query, method="GET"),
)
return True
except ValueError:
pass
self.report.add_finding(
title="GET-based Queries",
severity="PASS",
description="Server does not accept queries via GET method.",
curl_command=None,
)
return FalseYour auditor now covers six distinct vulnerability classes. Next up, you'll build the color-coded reporting module and run the full audit against DVGA to see all your checks in action.
Build the Report and Run the Full Audit
Your auditor now runs six security checks and collects findings through self.report.add_finding(). The missing piece is the report module itself, the code that stores those findings and turns them into a readable terminal output.
In this step, you will create the reporting module and wire everything together into a single command. By the end, you will run your auditor against DVGA and see a color-coded security report with severity ratings and reproduction commands.
In this step, get ready to:
- Create the color-coded security report module.
- Wire all six checks into a single audit runner.
- Run the full audit against DVGA and verify the findings.
Create the SecurityReport class
Professional security tools don't just dump raw text to the terminal. They use color coding so a reader can instantly distinguish critical findings from passing checks. Your report module will use ANSI escape codes to color findings by severity.
- Create a new file called report.py in your project directory.
- Add the class definition with the ANSI color constants and data storage methods by pasting the following code:
class SecurityReport:
COLORS = {
"HIGH": "\033[91m",
"MEDIUM": "\033[93m",
"LOW": "\033[94m",
"PASS": "\033[92m",
"CRITICAL": "\033[95m",
"RESET": "\033[0m",
"BOLD": "\033[1m",
}
def __init__(self, target_url):
self.target_url = target_url
self.findings = []
def add_finding(self, title, severity, description, curl_command=None):
self.findings.append({
"title": title,
"severity": severity,
"description": description,
"curl_command": curl_command,
})What does this code do?
The COLORS dictionary maps severity levels to ANSI escape sequences. These are special character codes that tell your terminal to render text in a specific color.
The add_finding() method is what your auditor has been calling throughout the previous steps. It appends each check result to the self.findings list as a dictionary.
Now add the display methods. These iterate through the stored findings and print them with color and icons.
- In report.py, add the following methods below the add_finding method, still inside the SecurityReport class:
def print_report(self):
print(f"\n{'='*60}")
print(f"{self.COLORS['BOLD']} SECURITY AUDIT REPORT{self.COLORS['RESET']}")
print(f" Target: {self.target_url}")
print(f"{'='*60}\n")
for finding in self.findings:
severity = finding["severity"]
color = self.COLORS.get(severity, self.COLORS["RESET"])
icon = self._get_icon(severity)
print(f" {icon} {color}[{severity}]{self.COLORS['RESET']} {finding['title']}")
print(f" {finding['description']}")
if finding["curl_command"]:
print(f" Reproduce: {finding['curl_command']}")
print()
self._print_summary()
def _get_icon(self, severity):
icons = {
"CRITICAL": "💀",
"HIGH": "🔴",
"MEDIUM": "🟡",
"LOW": "🔵",
"PASS": "✅",
}
return icons.get(severity, "❓")What does this code do?
The print_report() method loops through all findings and prints each one with its severity color, an emoji icon, and the reproduction curl command. The colored output makes it easy to spot critical issues at a glance.
The _get_icon() helper maps each severity level to a visual emoji indicator.
The final piece is the summary section that counts findings by severity level.
- Add the _print_summary method below _get_icon, still inside the class:
def _print_summary(self):
counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "PASS": 0}
for finding in self.findings:
severity = finding["severity"]
if severity in counts:
counts[severity] += 1
print(f"{'='*60}")
print(f"{self.COLORS['BOLD']} SUMMARY{self.COLORS['RESET']}")
print(f"{'='*60}")
print(f" Total checks: {len(self.findings)}")
vuln_count = counts["CRITICAL"] + counts["HIGH"] + counts["MEDIUM"] + counts["LOW"]
print(f" Vulnerabilities found: {vuln_count}")
print()
if counts["CRITICAL"] > 0:
print(f" {self.COLORS['CRITICAL']}CRITICAL: {counts['CRITICAL']}{self.COLORS['RESET']}")
if counts["HIGH"] > 0:
print(f" {self.COLORS['HIGH']}HIGH: {counts['HIGH']}{self.COLORS['RESET']}")
if counts["MEDIUM"] > 0:
print(f" {self.COLORS['MEDIUM']}MEDIUM: {counts['MEDIUM']}{self.COLORS['RESET']}")
if counts["LOW"] > 0:
print(f" {self.COLORS['LOW']}LOW: {counts['LOW']}{self.COLORS['RESET']}")
if counts["PASS"] > 0:
print(f" {self.COLORS['PASS']}PASS: {counts['PASS']}{self.COLORS['RESET']}")
print(f"\n{'='*60}\n")What does this code do?
The _print_summary() method tallies findings by severity and prints a bottom-line count. It calculates the total vulnerability count by adding up everything except PASS results.
- Save report.py.
Seeing indentation errors?
All methods must be indented one level inside the SecurityReport class. Make sure print_report, _get_icon, and _print_summary all start with exactly 4 spaces of indentation.
Help me fix indentation errors in my report.py file.
✔️ Awesome, I've got everything!
Great. Double check you have saved report.py.
ⓧ I'd like to double check the full code
class SecurityReport:
COLORS = {
"HIGH": "\033[91m",
"MEDIUM": "\033[93m",
"LOW": "\033[94m",
"PASS": "\033[92m",
"CRITICAL": "\033[95m",
"RESET": "\033[0m",
"BOLD": "\033[1m",
}
def __init__(self, target_url):
self.target_url = target_url
self.findings = []
def add_finding(self, title, severity, description, curl_command=None):
self.findings.append({
"title": title,
"severity": severity,
"description": description,
"curl_command": curl_command,
})
def print_report(self):
print(f"\n{'='*60}")
print(f"{self.COLORS['BOLD']} SECURITY AUDIT REPORT{self.COLORS['RESET']}")
print(f" Target: {self.target_url}")
print(f"{'='*60}\n")
for finding in self.findings:
severity = finding["severity"]
color = self.COLORS.get(severity, self.COLORS["RESET"])
icon = self._get_icon(severity)
print(f" {icon} {color}[{severity}]{self.COLORS['RESET']} {finding['title']}")
print(f" {finding['description']}")
if finding["curl_command"]:
print(f" Reproduce: {finding['curl_command']}")
print()
self._print_summary()
def _get_icon(self, severity):
icons = {
"CRITICAL": "💀",
"HIGH": "🔴",
"MEDIUM": "🟡",
"LOW": "🔵",
"PASS": "✅",
}
return icons.get(severity, "❓")
def _print_summary(self):
counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "PASS": 0}
for finding in self.findings:
severity = finding["severity"]
if severity in counts:
counts[severity] += 1
print(f"{'='*60}")
print(f"{self.COLORS['BOLD']} SUMMARY{self.COLORS['RESET']}")
print(f"{'='*60}")
print(f" Total checks: {len(self.findings)}")
vuln_count = counts["CRITICAL"] + counts["HIGH"] + counts["MEDIUM"] + counts["LOW"]
print(f" Vulnerabilities found: {vuln_count}")
print()
if counts["CRITICAL"] > 0:
print(f" {self.COLORS['CRITICAL']}CRITICAL: {counts['CRITICAL']}{self.COLORS['RESET']}")
if counts["HIGH"] > 0:
print(f" {self.COLORS['HIGH']}HIGH: {counts['HIGH']}{self.COLORS['RESET']}")
if counts["MEDIUM"] > 0:
print(f" {self.COLORS['MEDIUM']}MEDIUM: {counts['MEDIUM']}{self.COLORS['RESET']}")
if counts["LOW"] > 0:
print(f" {self.COLORS['LOW']}LOW: {counts['LOW']}{self.COLORS['RESET']}")
if counts["PASS"] > 0:
print(f" {self.COLORS['PASS']}PASS: {counts['PASS']}{self.COLORS['RESET']}")
print(f"\n{'='*60}\n")Wire all checks into the audit runner
Your auditor has six individual check methods, but no way to run them all in sequence. The run_audit() method will orchestrate the full scan and trigger the report at the end.
- In graphql_auditor.py, scroll to the bottom of the GraphQLAuditor class (after the check_get_based_queries method).
- Add the run_audit() method by pasting this code:
def run_audit(self):
print(f"\n{'='*60}")
print(f" GraphQL Security Auditor")
print(f" Target: {self.target_url}")
print(f"{'='*60}\n")
self.check_introspection()
self.check_query_depth()
self.check_alias_overloading()
self.check_batch_queries()
self.check_field_suggestions()
self.check_get_based_queries()
self.report.print_report()What does this code do?
The run_audit() method prints a header banner identifying the target, then calls each of the six check methods in sequence. Each check adds its finding to self.report internally. After all checks finish, it calls self.report.print_report() to output the full color-coded results.
Now add the entry point so you can run the script from the command line.
- Below the class (outside it, at the top level of the file), add the __main__ block:
if __name__ == "__main__":
target = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:5013/graphql"
auditor = GraphQLAuditor(target)
auditor.run_audit()What does this code do?
The if __name__ == "__main__" block makes the script runnable from the terminal. It accepts an optional command-line argument for the target URL. If none is provided, it defaults to http://localhost:5013/graphql (your DVGA instance).
- Save graphql_auditor.py.
Not sure where to paste the code?
The run_audit() method goes inside the class, indented with 4 spaces. The if __name__ block goes outside the class with no indentation. Check the full code tab below to confirm placement.
Help me fix the placement of run_audit and the __main__ block in graphql_auditor.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved graphql_auditor.py.
ⓧ I'd like to double check the full code
import requests
import sys
import time
from report import SecurityReport
class GraphQLAuditor:
def __init__(self, target_url, timeout=10):
self.target_url = target_url
self.timeout = timeout
self.report = SecurityReport(target_url)
self.headers = {"Content-Type": "application/json"}
def send_query(self, query, variables=None, method="POST"):
payload = {"query": query}
if variables:
payload["variables"] = variables
try:
if method == "POST":
response = requests.post(
self.target_url,
json=payload,
headers=self.headers,
timeout=self.timeout,
)
else:
response = requests.get(
self.target_url,
params={"query": query},
timeout=self.timeout,
)
return response
except requests.exceptions.Timeout:
return None
except requests.exceptions.ConnectionError:
return None
def get_curl_command(self, query, method="POST"):
if method == "POST":
escaped_query = query.replace('"', '\\"').replace("\n", " ")
return (
f'curl -X POST -H "Content-Type: application/json" '
f'-d \'{{"query": "{escaped_query}"}}\' '
f"'{self.target_url}'"
)
else:
import urllib.parse
encoded_query = urllib.parse.quote(query)
return f"curl '{self.target_url}?query={encoded_query}'"
def check_introspection(self):
print("[*] Testing: Introspection Query...")
query = """
query {
__schema {
types {
name
fields {
name
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data and "__schema" in data["data"]:
types = data["data"]["__schema"]["types"]
type_count = len(types)
self.report.add_finding(
title="Introspection Query Enabled",
severity="HIGH",
description=(
f"Full schema introspection is enabled. "
f"Discovered {type_count} types. "
f"Attackers can map the entire API surface."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Introspection Query",
severity="PASS",
description="Introspection is disabled or restricted.",
curl_command=None,
)
return False
def check_query_depth(self):
print("[*] Testing: Query Depth Limit...")
query = """
query {
pastes {
owner {
pastes {
owner {
pastes {
owner {
name
}
}
}
}
}
}
}
"""
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data and "errors" not in data:
self.report.add_finding(
title="No Query Depth Limit",
severity="HIGH",
description=(
"Server accepts deeply nested queries (7 levels) "
"without restriction. Vulnerable to depth-based "
"denial of service attacks."
),
curl_command=self.get_curl_command(query),
)
return True
elif "errors" in data and "data" in data:
self.report.add_finding(
title="No Query Depth Limit",
severity="HIGH",
description=(
"Server processes deeply nested queries (7 levels) "
"and returns partial data. Vulnerable to depth-based "
"denial of service attacks."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Query Depth Limit",
severity="PASS",
description="Server enforces query depth limits.",
curl_command=None,
)
return False
def check_alias_overloading(self):
print("[*] Testing: Alias Overloading...")
aliases = " ".join(
[f"alias{i}: __typename" for i in range(100)]
)
query = f"query {{ {aliases} }}"
response = self.send_query(query)
if response and response.status_code == 200:
data = response.json()
if "data" in data:
alias_count = len(data["data"])
if alias_count >= 100:
self.report.add_finding(
title="Alias Overloading Allowed",
severity="HIGH",
description=(
f"Server accepted {alias_count} aliases in a "
f"single query. Attackers can use aliases to "
f"bypass rate limiting and amplify attacks."
),
curl_command=self.get_curl_command(query),
)
return True
self.report.add_finding(
title="Alias Overloading",
severity="PASS",
description="Server restricts alias count.",
curl_command=None,
)
return False
def check_batch_queries(self):
print("[*] Testing: Batch Query Support...")
batch_payload = [
{"query": "query { __typename }"}
for _ in range(10)
]
try:
response = requests.post(
self.target_url,
json=batch_payload,
headers=self.headers,
timeout=self.timeout,
)
if response and response.status_code == 200:
data = response.json()
if isinstance(data, list) and len(data) >= 10:
self.report.add_finding(
title="Batch Queries Accepted",
severity="MEDIUM",
description=(
f"Server processes {len(data)} batched operations "
f"in a single HTTP request. Attackers can bypass "
f"per-request rate limiting."
),
curl_command=(
f'curl -X POST -H "Content-Type: application/json" '
f"-d '[{{\"query\": \"query {{ __typename }}\"}},"
f"{{\"query\": \"query {{ __typename }}\"}}]' "
f"'{self.target_url}'"
),
)
return True
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
pass
self.report.add_finding(
title="Batch Queries",
severity="PASS",
description="Server does not accept batched operations.",
curl_command=None,
)
return False
def check_field_suggestions(self):
print("[*] Testing: Field Suggestions...")
query = "query { __typena }"
response = self.send_query(query)
if response and response.status_code in (200, 400):
try:
data = response.json()
response_text = str(data)
if "Did you mean" in response_text or "did you mean" in response_text:
self.report.add_finding(
title="Field Suggestions Enabled",
severity="LOW",
description=(
"Server returns field name suggestions in error "
"messages. Attackers can enumerate the schema even "
"if introspection is disabled."
),
curl_command=self.get_curl_command(query),
)
return True
except ValueError:
pass
self.report.add_finding(
title="Field Suggestions",
severity="PASS",
description="Server does not expose field name suggestions.",
curl_command=None,
)
return False
def check_get_based_queries(self):
print("[*] Testing: GET-based Queries (CSRF)...")
query = "query { __typename }"
response = self.send_query(query, method="GET")
if response and response.status_code == 200:
try:
data = response.json()
if "data" in data and "__typename" in data["data"]:
self.report.add_finding(
title="GET-based Queries Accepted",
severity="HIGH",
description=(
"Server processes GraphQL queries via HTTP GET. "
"This enables Cross-Site Request Forgery (CSRF) "
"attacks against authenticated users."
),
curl_command=self.get_curl_command(query, method="GET"),
)
return True
except ValueError:
pass
self.report.add_finding(
title="GET-based Queries",
severity="PASS",
description="Server does not accept queries via GET method.",
curl_command=None,
)
return False
def run_audit(self):
print(f"\n{'='*60}")
print(f" GraphQL Security Auditor")
print(f" Target: {self.target_url}")
print(f"{'='*60}\n")
self.check_introspection()
self.check_query_depth()
self.check_alias_overloading()
self.check_batch_queries()
self.check_field_suggestions()
self.check_get_based_queries()
self.report.print_report()
if __name__ == "__main__":
target = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:5013/graphql"
auditor = GraphQLAuditor(target)
auditor.run_audit()Run the full audit against DVGA
Everything is wired up. Time to run your security auditor and see what it finds against DVGA's intentionally vulnerable configuration.
- Make sure your DVGA container is still running. Confirm by visiting http://localhost:5013 in your browser.
- Run the full audit by executing this command in your terminal:
python3 graphql_auditor.pyWhat should I see?
Your terminal should display a color-coded security report with a header, individual findings (each showing severity, title, description, and a curl reproduction command), and a summary section at the bottom.
DVGA is deliberately insecure. You should see at least 4 vulnerabilities detected, including HIGH severity findings for introspection, query depth, alias overloading, and GET-based queries.
- Review the output. Confirm the report shows at least 4 findings with severity levels (HIGH, MEDIUM, or LOW) and that the summary counts match the individual findings listed above it.
Seeing a ModuleNotFoundError for report?
Make sure report.py is saved in the same directory as graphql_auditor.py. Python looks for modules in the current directory.
Also confirm you are running the command from inside your project folder (the folder containing both files).
Help me fix a ModuleNotFoundError when running my graphql_auditor.py script.
Seeing a ConnectionError?
This means DVGA is not reachable. Verify the container is running by executing docker ps and looking for the dvga container. If it is not listed, restart it with docker start dvga.
Help me troubleshoot a ConnectionError when running my GraphQL auditor against DVGA.
Why curl reproduction commands?
Each finding includes a curl command that anyone can paste into a terminal to reproduce the vulnerability. This is standard practice in security reporting. It gives developers proof that the issue exists and a quick way to verify after they apply a fix.
Secret mission
Add Circular Fragment DoS Detection
Your auditor covers the six most common GraphQL misconfigurations. But there's a seventh vulnerability class that exploits how GraphQL resolves fragments. Can you detect it?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally using a Docker container and Python files on your machine. There are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- DVGA Docker container (running at http://localhost:5013).
- Python virtual environment with requests installed.
- Project files: graphql_auditor.py and report.py.
Keep everything running
No action needed. Choose this if you plan to keep practicing GraphQL security testing or want to extend the auditor with additional checks.
- DVGA will continue running on port 5013 until you restart your machine or stop Docker.
- Your project files and virtual environment remain in your project folder on the Desktop.
Pause - I'll come back to this later
Stop the DVGA container to free up resources. Your project files stay intact so you can resume later.
- Stop the DVGA container by running this command:
docker stop dvgaWhen you want to resume, restart the container by running:
docker start dvga- Your project files and virtual environment remain on disk. Reactivate the virtual environment when you return.
Delete - I don't want to use this again
Remove all project resources and start fresh. This deletes the DVGA container and your project files permanently.
- Remove the DVGA container by running this command:
docker rm -f dvga- Deactivate your virtual environment by running:
deactivate- Delete your project folder from the Desktop by running:
macOS
cd ~/Desktop
rm -rf graphql-auditorWindows
cd ~\Desktop
rmdir /s /q graphql-auditorNice Work!
Nice Work!
You just built a GraphQL Security Auditor from scratch. Your Python CLI tool can now scan any GraphQL endpoint and produce a severity-rated vulnerability report complete with reproduction commands.
Here's what you learned:
- Built a Python CLI tool that detects 6 common GraphQL security vulnerabilities including introspection exposure, depth-based DoS, and CSRF risk.
- Set up a local vulnerable target (DVGA) in Docker for safe, legal security testing without touching production systems.
- Generated color-coded security reports with severity ratings and curl reproduction commands that prove each finding to stakeholders.
- Secret Mission: Added circular fragment DoS detection based on a real-world CVE pattern that tests for recursive fragment reference vulnerabilities.
Ready to quiz yourself?