curl -s -X POST "https://target.com/graphql" \
-H "Content-Type: application/json" \
-d '{"query": "{ __schema { types { name fields { name type { name } } } } }"}' | jq '.'
curl -s -X POST "https://target.com/graphql" \
-H "Content-Type: application/json" \
-d '{"query": "{ __schema { queryType { fields { name } } mutationType { fields { name } } } }"}' | jq '.'
import requests
import json
class GraphQLTester:
def __init__(self, url, headers=None):
self.url = url
self.session = requests.Session()
if headers:
self.session.headers.update(headers)
self.findings = []
def query(self, query, variables=None):
"""Execute GraphQL query"""
payload = {"query": query}
if variables:
payload["variables"] = variables
response = self.session.post(
self.url,
json=payload,
headers={"Content-Type": "application/json"}
)
return response.json()
def test_introspection(self):
"""Test if introspection is enabled"""
print("[*] Testing introspection...")
query = """
{
__schema {
types { name }
queryType { fields { name } }
mutationType { fields { name } }
}
}
"""
result = self.query(query)
if 'errors' not in result or '__schema' in str(result):
print("[VULN] Introspection enabled!")
self.findings.append({
"issue": "GraphQL introspection enabled",
"severity": "Medium"
})
if 'data' in result and result['data']:
schema = result['data'].get('__schema', {})
types = schema.get('types', [])
print(f" Found {len(types)} types")
query_type = schema.get('queryType', {})
if query_type:
queries = [f['name'] for f in query_type.get('fields', [])]
print(f" Queries: {queries[:10]}")
else:
print("[OK] Introspection disabled")
def test_authorization(self, queries):
"""Test authorization on queries"""
print("\n[*] Testing authorization...")
for query_name, query in queries.items():
result = self.query(query)
if 'data' in result and result['data']:
print(f"[POTENTIAL] Accessible: {query_name}")
elif 'errors' in result:
error_msg = str(result['errors'])
if 'unauthorized' in error_msg.lower() or 'forbidden' in error_msg.lower():
print(f"[OK] Protected: {query_name}")
else:
print(f"[CHECK] Error on {query_name}: {error_msg[:50]}")
def test_depth_limit(self, max_depth=10):
"""Test query depth limits"""
print("\n[*] Testing depth limits...")
nested = "{ users "
for i in range(max_depth):
nested += "{ friends "
nested += "{ id } " + "} " * max_depth + "}"
result = self.query(nested)
if 'errors' in result:
error = str(result['errors'])
if 'depth' in error.lower() or 'complexity' in error.lower():
print(f"[OK] Depth limit enforced")
else:
print(f"[WARN] Query failed but not due to depth limit")
else:
print(f"[VULN] No depth limit (tested {max_depth} levels)")
self.findings.append({
"issue": "No query depth limit",
"severity": "Medium"
})
def test_batching(self):
"""Test batch query vulnerabilities"""
print("\n[*] Testing batching...")
batch_query = """
query {
q1: login(email: "test1@test.com", password: "pass1") { token }
q2: login(email: "test2@test.com", password: "pass2") { token }
q3: login(email: "test3@test.com", password: "pass3") { token }
}
"""
result = self.query(batch_query)
if 'errors' not in result or 'data' in result:
print("[WARN] Batching allowed - potential for brute force")
self.findings.append({
"issue": "Batching allows brute force",
"severity": "Medium"
})
def test_injection(self):
"""Test for injection vulnerabilities"""
print("\n[*] Testing injection...")
payloads = [
'{ user(id: "1 OR 1=1") { id } }',
'{ user(id: "1\'; DROP TABLE users--") { id } }',
'{ user(id: "${7*7}") { id } }',
'{ user(id: "{{7*7}}") { id } }',
]
for payload in payloads:
result = self.query(payload)
if 'errors' in result:
error = str(result['errors'])
if 'sql' in error.lower() or 'syntax' in error.lower():
print(f"[VULN] SQL error disclosure")
self.findings.append({
"issue": "SQL injection possible",
"severity": "Critical"
})
def generate_report(self):
print("\n" + "="*50)
print("GRAPHQL SECURITY REPORT")
print("="*50)
if not self.findings:
print("\nNo issues found.")
return
print(f"\nFindings: {len(self.findings)}")
for f in self.findings:
print(f"\n [{f['severity']}] {f['issue']}")
tester = GraphQLTester(
"https://target.com/graphql",
{"Authorization": "Bearer token"}
)
tester.test_introspection()
tester.test_authorization({
"users": "{ users { id email } }",
"admin": "{ admin { secrets } }",
})
tester.test_depth_limit()
tester.test_batching()
tester.test_injection()
tester.generate_report()
from graphene import Schema
schema = Schema(query=Query, mutation=Mutation)
if not DEBUG:
schema.introspection = False
from graphene.validation import depth_limit_validator
schema.execute(query, validation_rules=[depth_limit_validator(5)])