Skip to content

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_atSELECT ... FOR UPDATE 行锁 + 条件原子更新 + 唯一幂等键约束

2. 分层加固实现 ​

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

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

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 检索)。原则:数据库不是可信输入源。

python
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 配合静态路径常量。
  • 支付扣款与发货原子事务:
sql
-- 事务内先锁行并做条件余额扣减,防止并发下先发货后扣款或负余额
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)支付状态机原子条件更新防超扣:

python
#!/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()

防御工作流 ​

text
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 校验策略

关联技术 ​

Apache-2.0 · For authorized learning and defensive research only · Disclaimer