Advanced SQLi & Second-Order Defense — 高级注入、二阶与 OOB 防御

当基础正则规则被 HPP、分块传输或编码变体穿透，或载荷潜伏至后台二阶触发与带外外泄时，唯有网关与应用解析对齐、全链路零信任参数化与数据库出口隔离才能形成闭环防线。

1. 高级注入面与防御对照

针对 02-sqli-advanced.md 中的高阶绕过与非典型注入场景，按以下层级布防：

攻击向量	绕过/利用机制	核心防御策略
HTTP 层绕过（HPP / 分块 / Body 切换）	?id=1&id=2 合并差异、chunked 未重组、JSON/multipart 混用	网关统一规范化，拒绝重复参数名，路由锁定单一 Content-Type
语法与函数变体绕过	/*!50000SELECT*/、换行符、等价函数 schema()/mid()	根治手段为预编译参数化；WAF 仅作纵深告警，不替代绑定参数
二阶 SQL 注入	注册昵称/订单备注/文件名入库，后台搜索或导出时拼接触发	对“来自数据库的字段”执行与外部输入同等严格的参数化与白名单约束
OOB 带外数据外泄	MySQL LOAD_FILE(\\\\...)、Oracle UTL_HTTP、MSSQL xp_dirtree	数据库子网禁出网（拦截 UDP/TCP 53、445、139、80/443），回收相关包/权限
INSERT / UPDATE 注入	VALUES 字符串逃逸或 UPDATE SET 越权覆盖 role/status	使用字段白名单 DTO，禁止用户控制列名或整行未过滤字典入库
ORDER BY / LIMIT / JSON 注入	CASE WHEN 布尔盲注、OFFSET 注入、JSON_EXTRACT 路径闭合	列名与排序方向查表映射、LIMIT 强转正整数范围、固定 JSON 路径模板
支付状态机竞态与篡改	利用 UPDATE 或并发窗口绕过扣款直接写入 delivered_at	SELECT ... FOR UPDATE 行锁 + 条件原子更新 + 唯一幂等键约束

2. 分层加固实现

2.1 HTTP 解析一致性与 HPP / Content-Type 防护

消除前置代理、WAF 与后端框架对参数解析的歧义（例如 Express qs、PHP 数组解析与 Go/Python 取首/尾差异）：

# http_canonical_guard.py — HTTP 解析一致性校验中间件示例
from urllib.parse import parse_qsl

ALLOWED_CONTENT_TYPES = {
    "/api/order/create": "application/json",
    "/api/order/callback": "application/x-www-form-urlencoded",
}

def validate_request_envelope(path: str, raw_query: str, content_type: str, transfer_encoding: str = "") -> dict[str, str]:
    # 1. 拒绝非标准或混淆的 Transfer-Encoding
    if transfer_encoding and transfer_encoding.lower().strip() not in ("", "chunked"):
        raise ValueError("ambiguous transfer-encoding rejected")

    # 2. 拒绝同键名重复传参（阻断 HPP 参数污染）
    pairs = parse_qsl(raw_query, keep_blank_values=True)
    seen: dict[str, str] = {}
    for k, v in pairs:
        if k in seen:
            raise ValueError(f"duplicate query parameter rejected (HPP): {k}")
        seen[k] = v

    # 3. 路由绑定唯一 Content-Type，防止表单/JSON/multipart 切换绕过检查
    expected_ct = ALLOWED_CONTENT_TYPES.get(path)
    if expected_ct:
        actual_ct = content_type.split(";", 1)[0].strip().lower()
        if actual_ct != expected_ct:
            raise ValueError(f"unexpected content-type: {actual_ct} != {expected_ct}")
    return seen

Nginx 侧同步收口：禁用异常 Transfer-Encoding 组合，确保反向代理重组请求体后再转发，且对管理后台与 API 路由按 Content-Type 精确放行。

2.2 根治二阶 SQL 注入（后台检索、导出与异步任务）

二阶注入高发于“写入时无感、读取时拼接”的后台模块（如订单备注搜索、收货邮箱队列、回调 raw_body 检索）。原则：数据库不是可信输入源。

import sqlite3

def escape_like_pattern(value: str) -> str:
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

def admin_search_orders_safe(conn: sqlite3.Connection, remark_keyword: str, sort_field: str = "id"):
    # 1. 即使 remark 来自上游数据库表或管理员输入，LIKE 依然参数化且转义通配符
    allowed_sort = {"id": "id", "created_at": "created_at", "amount": "amount"}
    col = allowed_sort.get(sort_field, "id")
    escaped = f"%{escape_like_pattern(remark_keyword)}%"
    sql = f"SELECT id, remark, status FROM orders WHERE remark LIKE ? ESCAPE '\\' ORDER BY {col} DESC LIMIT 100"
    return conn.execute(sql, (escaped,)).fetchall()

2.3 切断 OOB 带外通道（MySQL / Oracle / MSSQL）

1. 数据库主机出口网络隔离：
   • 数据库所在安全组/防火墙配置默认拒绝全部出站流量（仅允许连接授权内网日志/备份网关），阻断 TCP/UDP 53（DNSLog）、TCP 445/139（SMB NTLM/unc 共享）、TCP 80/443（HTTP 外带）。
2. 引擎高危扩展禁用：
   • MySQL：设置 secure_file_priv = NULL，回收应用账号 FILE 权限。
   • Oracle：REVOKE EXECUTE ON UTL_HTTP FROM PUBLIC; REVOKE EXECUTE ON UTL_INADDR FROM PUBLIC; REVOKE EXECUTE ON UTL_FILE FROM PUBLIC;。
   • MSSQL：禁用 xp_dirtree、xp_fileexist、xp_cmdshell 对非特权角色的执行权限，并关闭未使用的 CLR 集成。

2.4 非 SELECT 注入防护与支付状态机原子锁

针对 INSERT、UPDATE、JSON_EXTRACT 及支付状态竞态：

• 字段级 DTO 更新：UPDATE 绝不允许将用户 JSON 的 key 直接拼为列名，必须显式列出允许修改的业务列（如仅允许更新 shipping_address，禁止更新 status、role、paid_at）。
• 固定 JSON 路径查询：不要把用户输入拼入 JSON_EXTRACT(cfg, '$.' + user_key)；应使用固定路径或在应用层解析 JSON 后查字典键，或在 MySQL 8+ 使用 JSON_VALUE 配合静态路径常量。
• 支付扣款与发货原子事务：

-- 事务内先锁行并做条件余额扣减，防止并发下先发货后扣款或负余额
BEGIN;
SELECT status, amount, user_id FROM orders WHERE id = ? FOR UPDATE;
UPDATE users SET balance = balance - ? WHERE id = ? AND balance >= ?;
-- 应用校验 ROW_COUNT() == 1 后再推进订单状态与发货
UPDATE orders SET status = 'paid', paid_at = CURRENT_TIMESTAMP WHERE id = ? AND status = 'pending';
COMMIT;

同时在支付流水表建立 UNIQUE KEY uk_provider_tx (provider, transaction_id)，从数据库约束层彻底消灭重复回调入账。

3. 可运行防御自检脚本

下面脚本在本地自检（1）HPP 重复参数及非法 Content-Type 拦截；（2）二阶注入载荷入库后再经后台 LIKE 检索与 UPDATE 执行时的免疫验证；（3）支付状态机原子条件更新防超扣：

#!/usr/bin/env python3
"""Self-check for HPP guard, second-order SQLi immunity, and atomic balance transition."""
import sqlite3
from urllib.parse import parse_qsl


def reject_hpp_query(raw_query: str) -> dict[str, str]:
    parsed: dict[str, str] = {}
    for k, v in parse_qsl(raw_query, keep_blank_values=True):
        if k in parsed:
            raise ValueError(f"HPP detected on key: {k}")
        parsed[k] = v
    return parsed


def escape_like(val: str) -> str:
    return val.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def run_self_check() -> None:
    # 1. HPP 拦截测试
    try:
        reject_hpp_query("id=1&id=2%20UNION%20SELECT%201,2,3--")
        raise AssertionError("HPP query should be rejected")
    except ValueError:
        pass
    assert reject_hpp_query("id=101&sort=created") == {"id": "101", "sort": "created"}

    # 2. 二阶注入免疫测试（写入含恶意 SQL 片段的用户名与备注，再通过检索与更新验证隔离）
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, role TEXT, balance INTEGER)")
    conn.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, user_id INTEGER, remark TEXT, status TEXT)")

    malicious_user = "admin' OR '1'='1'--"
    malicious_remark = "%' UNION SELECT 1,2,3,4--"
    conn.execute("INSERT INTO users (username, role, balance) VALUES (?, 'member', 100)", (malicious_user,))
    conn.execute("INSERT INTO users (username, role, balance) VALUES ('alice', 'member', 100)")
    conn.execute("INSERT INTO orders (user_id, remark, status) VALUES (1, ?, 'pending')", (malicious_remark,))

    # 后台按用户名更新密码/角色时使用参数化，不会影响 alice
    conn.execute("UPDATE users SET role = 'vip' WHERE username = ?", (malicious_user,))
    alice_role = conn.execute("SELECT role FROM users WHERE username = 'alice'").fetchone()[0]
    assert alice_role == "member", "Second-order UPDATE leaked into other row"

    # 后台按存储的 remark 做 LIKE 检索（带 ESCAPE），不会扩展匹配
    rows = conn.execute(
        "SELECT id, remark FROM orders WHERE remark LIKE ? ESCAPE '\\'",
        (f"%{escape_like(malicious_remark)}%",),
    ).fetchall()
    assert len(rows) == 1

    # 3. 原子条件扣款测试（余额不足时受影响行数为 0）
    cur = conn.execute("UPDATE users SET balance = balance - ? WHERE id = 2 AND balance >= ?", (150, 150))
    assert cur.rowcount == 0
    assert conn.execute("SELECT balance FROM users WHERE id = 2").fetchone()[0] == 100
    print("[PASS] Advanced SQLi & second-order defense self-check succeeded.")


if __name__ == "__main__":
    run_self_check()

防御工作流

1. 在网关与应用层对齐参数解析：禁止重复参数名（HPP），按路由强制校验 Content-Type
2. 对内部后台检索、CSV 导出、消息队列消费与回调归档代码排查二阶 SQL 拼接并全部参数化
3. 为所有非 SELECT 写操作建立显式字段白名单与 ORDER BY / LIMIT / JSON 路径映射表
4. 在数据库宿主机与安全组层面阻断出网流量（DNS 53 / SMB 445 / HTTP），回收 FILE/UTL_HTTP/xp_dirtree 权限
5. 在订单与账户表上实施行级锁事务（FOR UPDATE）、原子余额条件更新与回调幂等唯一索引
6. 运行 HPP 与二阶注入自检脚本，确认无越权更新或带外触发

验证与判定标准

防御目标	判定手段	合格标准	异常信号
HTTP 解析一致性	重复键/混用 Content-Type 请求测试	网关与后端同时拒绝重复参数及非预期 body 格式	?id=1&id=2 或表单/JSON 切换可绕过前置校验
二阶注入免疫	写入 ' OR '1'='1 至昵称/备注后触发后台查询/导出	后台查询结果与更新影响行数仅限精确匹配行	触发报错、全表更新或排序逻辑改变
OOB 信道阻断	数据库容器/主机出网连通性检查与权限查询	出网连接（53/80/443/445）被拒绝，高危函数无执行权限	数据库可发起外部 DNS 解析或 HTTP/SMB 连接
非 SELECT 结构约束	INSERT/UPDATE/ORDER BY/JSON 边界输入测试	拒绝未在白名单内的列名/路径，限制影响行与权限字段	任意字段可随请求覆盖 role/status
支付事务与幂等性	并发扣款与重复回调重放压测	余额永不为负，重复 transaction_id 触发唯一索引拒绝	出现并发超扣、先发货后扣款或重复入账

MCP 工具映射

防御步骤	MCP 工具	说明
检索高阶防御规则	kb_router	按 second-order、OOB、HPP、transaction lock 检索技术文件
对照攻击样本	kb_read_file	查阅 02-sqli-advanced.md 核对绕过层与非 SELECT 触发点
浏览数据库防御集	kb_catalog	获取 24-database 全部防御手册索引
网关与解析测试	http_probe	在 red/both 模式下验证 HPP 拒绝码与 Content-Type 校验策略

关联技术

• 02-sqli-advanced.md — 配对攻击面：高级 SQL 注入与 WAF 绕过
• 01-sqli-fundamentals-defense.md — 基础参数化与字符集统一防御
• 03-nosql-injection-defense.md — NoSQL 解析器与操作符注入防御