速率限制绕过防御 — 可信代理链、规范化键、原子令牌桶与验证码强绑定

威胁模型与防御目标

速率限制被绕过的本质，是限流键（Rate-Limit Key）推导链路与真实业务执行链路不一致。对照 01-rate-limit-bypass.md，蓝队必须在基础设施层、网关层与业务层消除以下五类不一致：

攻击向量	根本成因	防御控制
IP Header 注入（X-Forwarded-For / X-Real-IP / X-Internal-Request）	后端盲信客户端可控头，或信任内网段跳板值	边界剥离非可信代理头；仅对明确 CIDR 白名单自右向左解析 XFF
IPv6 /64 子网轮换	按 /128 单地址计数，攻击者持有整段 /64 地址空间	IPv6 统一按 /64（或 /56）网络前缀聚合后再计算限流键
CDN 源站直连绕过	源站防火墙允许公网直接访问 80/443，且 MX/旧子域暴露源站 IP	源站仅放行 CDN 回源 IP + mTLS Authenticated Origin Pulls + 消除源站 DNS 泄露
参数轮换与密码后缀绕过	限流键使用原始字符串（未去空格/大小写/Unicode 折叠），或仅在密码错误后/按 (user, pass) 计数	先做 NFKC + 去空白 + 大小写折叠 + E.164 号码归一；认证前先扣减 (account, subnet) 额度
并发竞态、端点漂移、多 Key 与验证码重放	GET -> CHECK -> INCR 非原子；/v1 与 /v2 未共享策略；验证码未单次消费	Redis Lua 原子滑动窗口/令牌桶；网关统一限流；租户级配额；GETDEL 原子核销验证码

架构加固配置

1. 边缘与反向代理层（Nginx + Origin Lockdown）

在入口网关层阻断伪造代理头与源站直连，同时对 IPv6 按 /64 掩码聚合：

# /etc/nginx/conf.d/rate_limit_hardening.conf
# 1. 仅信任已知 CDN / 内网反向代理网段，拒绝客户端直传 XFF
real_ip_header X-Forwarded-For;
real_ip_recursive on;
set_real_ip_from 173.245.48.0/20;   # 示例：Cloudflare IPv4 网段（按实际 CDN 列表定期同步）
set_real_ip_from 2400:cb00::/32;    # 示例：Cloudflare IPv6 网段
set_real_ip_from 10.20.0.0/16;      # 内部 ALB / Ingress 网段

# 2. 剥离客户端伪造的内部Bypass头与非标准IP头
proxy_set_header X-Internal-Request "";
proxy_set_header X-Is-Internal "";
proxy_set_header X-Original-URL "";
proxy_set_header X-Rewrite-URL "";
proxy_set_header X-Real-IP $remote_addr;

# 3. 源站 mTLS 强校验（防止绕过 CDN 直连源站 IP）
ssl_client_certificate /etc/nginx/certs/cdn-origin-pull-ca.pem;
ssl_verify_client on;

# 4. 将 IPv6 归并为 /64 前缀作为限流键（IPv4 保持 /32）
map $binary_remote_addr $limit_ip_prefix {
    ~^(?P<v6prefix>.{8}) $v6prefix; # IPv6 binary addr 前 8 字节即 /64
    default $binary_remote_addr;
}

limit_req_zone $limit_ip_prefix zone=auth_ip:20m rate=10r/m;
limit_conn_zone $limit_ip_prefix zone=per_prefix_conn:10m;

2. Redis Lua 原子令牌桶与参数规范化限流中间件

在应用层，必须在进入业务分支前完成原子计数与维度统一。下面的 Python 参考实现同时解决：

• 自右向左剥离可信代理提取真实源 IP，并把 IPv6 聚合到 /64；
• 对用户名/邮箱/手机号进行防绕过规范化（拒绝 \x00 与控制字符、NFKC 折叠、去除空白与同形字符归一、手机号 E.164 化）；
• 把 /api/send-code、/api/sendSms、GET/POST、JSON/Form 统一映射到同一逻辑动作键；
• 使用单次原子操作（对应生产环境 Redis Lua 脚本）消除并发竞态；
• 验证码挑战采用 consume_once 原子销毁机制并绑定 action + subject + client_prefix。

import ipaddress
import secrets
import threading
import time
import unicodedata
from dataclasses import dataclass, field


class RateLimitExceeded(Exception):
    """触发限流时抛出的统一异常。"""

    def __init__(self, retry_after: int, reason: str) -> None:
        super().__init__(reason)
        self.retry_after = retry_after
        self.reason = reason


def extract_client_prefix(
    peer_ip: str,
    xff_header: str | None,
    trusted_proxies: list[ipaddress.IPv4Network | ipaddress.IPv6Network],
) -> str:
    """自右向左剥离可信代理，并将 IPv6 归并为 /64 前缀。"""
    chain: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
    if xff_header:
        for raw in xff_header.split(","):
            candidate = raw.strip()
            if not candidate:
                continue
            try:
                chain.append(ipaddress.ip_address(candidate))
            except ValueError:
                # 含非法 IP 的 XFF 片段直接丢弃，回退到真实 TCP 对端 IP
                chain.clear()
                break

    peer = ipaddress.ip_address(peer_ip)
    chain.append(peer)

    # 仅当当前最右端属于可信代理时才继续向左回溯
    resolved = chain[-1]
    for idx in range(len(chain) - 1, -1, -1):
        addr = chain[idx]
        resolved = addr
        if not any(addr in net for net in trusted_proxies):
            break

    if isinstance(resolved, ipaddress.IPv6Address):
        net64 = ipaddress.ip_network(f"{resolved}/64", strict=False)
        return str(net64)
    return str(resolved)


def canonicalize_identifier(raw: str, kind: str = "username") -> str:
    """在生成限流键之前规范化账号或手机号，阻断大小写/空格/Unicode/区号变体绕过。"""
    if "\x00" in raw or any(ord(ch) < 32 for ch in raw):
        raise ValueError("control character rejected")

    # NFKC 归一化消除全角/组合字符差异，strip 去除首尾空白
    normalized = unicodedata.normalize("NFKC", raw).strip()
    if kind == "phone":
        digits = "".join(ch for ch in normalized if ch.isdigit())
        # 统一国内/国际区号变体（如 086 / +86 / 86 前缀）
        if digits.startswith("086") and len(digits) > 11:
            digits = digits[3:]
        elif digits.startswith("86") and len(digits) > 11:
            digits = digits[2:]
        return f"+86{digits}" if len(digits) == 11 else digits

    # 将常见西里尔混用字母等非 ASCII 账号字符直接拒绝或统一规范
    if not normalized.isascii():
        raise ValueError("non-ascii identifier rejected")
    return normalized.casefold()


@dataclass
class AtomicSlidingWindowLimiter:
    """模拟 Redis Lua 原子滑动窗口限流器（线程安全单步扣减）。"""

    max_requests: int
    window_seconds: float
    _buckets: dict[str, list[float]] = field(default_factory=dict)
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def check_and_consume(self, key: str, now: float | None = None) -> tuple[bool, int]:
        ts = now if now is not None else time.monotonic()
        cutoff = ts - self.window_seconds
        with self._lock:
            entries = [t for t in self._buckets.get(key, []) if t > cutoff]
            if len(entries) >= self.max_requests:
                retry_after = max(1, int(self.window_seconds - (ts - entries[0])) + 1)
                self._buckets[key] = entries
                return False, retry_after
            entries.append(ts)
            self._buckets[key] = entries
            return True, max(0, self.max_requests - len(entries))


@dataclass
class SingleUseCaptchaStore:
    """原子单次消费验证码存储，防止 Token 重放与响应泄密。"""

    ttl_seconds: float = 120.0
    _tokens: dict[str, tuple[str, str, float]] = field(default_factory=dict)
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def issue(self, action: str, answer: str) -> str:
        token = secrets.token_urlsafe(24)
        with self._lock:
            self._tokens[token] = (action, answer.casefold(), time.monotonic() + self.ttl_seconds)
        return token

    def verify_and_burn(self, token: str, action: str, user_answer: str) -> bool:
        with self._lock:
            record = self._tokens.pop(token, None)  # 对应 Redis GETDEL：无论对错立即销毁
        if not record:
            return False
        expected_action, expected_answer, expires_at = record
        if time.monotonic() > expires_at:
            return False
        return (
            secrets.compare_digest(expected_action, action)
            and secrets.compare_digest(expected_answer, user_answer.strip().casefold())
        )

对应的生产级 Redis Lua 脚本（确保无竞态）：

-- atomic_sliding_window.lua
-- KEYS[1] = rate limit key, ARGV[1] = now_ms, ARGV[2] = window_ms, ARGV[3] = limit
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call("ZREMRANGEBYSCORE", key, 0, now - window)
local current = redis.call("ZCARD", key)
if current >= limit then
    local oldest = redis.call("ZRANGE", key, 0, 0, "WITHSCORES")
    local retry_ms = window - (now - tonumber(oldest[2]))
    return {0, math.max(1, math.ceil(retry_ms / 1000))}
end
redis.call("ZADD", key, now, tostring(now) .. ":" .. redis.call("INCR", key .. ":seq"))
redis.call("PEXPIRE", key, window)
redis.call("PEXPIRE", key .. ":seq", window)
return {1, limit - current - 1}

3. 统一错误响应与租户配额规范

• 统一拦截响应：触发限流时固定返回 HTTP 429 Too Many Requests、Retry-After 与 {"error": "rate_limit_exceeded"}，禁止返回 200 软拦截混杂业务 oracle，也禁止在密码校验后再决定是否计数。
• API Key 租户归集：API 限流同时检查 tenant_id 与 api_key_id，防止创建多个 Key 线性放大配额。
• 端点归一化：路由中间件在限流前将 /api/send-code、/api/sendSms、/api/v2/verification/send 映射为统一的逻辑能力标签 action:send_otp。

防御工作流

1. 边界代理与源站封堵：在 CDN/WAF 与源站间启用 mTLS 与回源 IP 白名单；核查 DNS 历史、MX 与子域是否暴露源站 IP，必要时轮换源站 IP 并剥离客户端注入的代理与内网标识头。
2. 身份与维度规范化：所有限流键在计算前先做 IPv6 /64 聚合、标识符 NFKC/大小写/E.164 规范化，并强制拒绝控制字符与混用同形字符；所有等价端点与 HTTP 方法收敛到同一逻辑 action 键。
3. 原子计数与校验顺序：将读取、判断与写入合并为单条 Redis Lua 脚本，在密码校验与业务逻辑之前执行配额扣减；API 密钥同时受所属租户总配额约束。
4. 单次验证码强绑定：高风险阈值后强制要求服务端验证码，使用 GETDEL 语义原子核销 token，并绑定具体 action 与超时窗口，禁用客户端 JS 判定与降级算术验证码。
5. 回归测试与监控告警：对代理头伪造、IPv6 子网轮换、参数变体、高并发脉冲与验证码重放运行自动化回归断言，并监控 429 突变与单一租户/前缀高频触发告警。

验证与判定标准

运行下方自检脚本，验证五类核心防御控制均按预期拦截绕过：

import concurrent.futures
import ipaddress


def run_self_checks() -> None:
    trusted = [ipaddress.ip_network("10.20.0.0/16")]

    # 1. 伪造 X-Forwarded-For 无法绕过（非可信直连时回退到 peer_ip）
    attacker_xff = "1.2.3.4, 127.0.0.1"
    assert extract_client_prefix("198.51.100.25", attacker_xff, trusted) == "198.51.100.25"

    # 经过可信代理时取可信链左侧第一跳真实客户端 IP
    assert extract_client_prefix("10.20.1.5", "203.0.113.9, 10.20.0.8", trusted) == "203.0.113.9"

    # 2. IPv6 /64 子网聚合：同 /64 下不同 /128 映射到同一限流前缀
    p1 = extract_client_prefix("2001:db8:abcd:0001::1", None, trusted)
    p2 = extract_client_prefix("2001:db8:abcd:0001:ffff:ffff:ffff:9999", None, trusted)
    assert p1 == p2 == "2001:db8:abcd:1::/64"

    # 3. 用户名与手机号规范化：大小写/空格/区号变体命中同一键
    assert canonicalize_identifier(" Admin ", "username") == canonicalize_identifier("admin", "username")
    assert canonicalize_identifier("+8613800138000", "phone") == canonicalize_identifier("08613800138000", "phone")

    # 控制字符与西里尔同形字符被拒绝
    for bad in ("admin\x00", "аdmin"):
        try:
            canonicalize_identifier(bad, "username")
            raise AssertionError(f"Expected ValueError for {bad!r}")
        except ValueError:
            pass

    # 4. 高并发下原子限流严格不超过 max_requests（无竞态超发）
    limiter = AtomicSlidingWindowLimiter(max_requests=5, window_seconds=60.0)
    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as pool:
        results = list(pool.map(lambda _: limiter.check_and_consume("login:admin:198.51.100.25")[0], range(30)))
    assert sum(1 for ok in results if ok) == 5

    # 5. 验证码单次消费且绑定 action：重放或跨端点必然失败
    store = SingleUseCaptchaStore()
    tok = store.issue("login", "X9K2")
    assert store.verify_and_burn(tok, "send_otp", "X9K2") is False  # action 不匹配且已销毁
    assert store.verify_and_burn(tok, "login", "X9K2") is False     # 重放失败

    tok2 = store.issue("login", "M4P7")
    assert store.verify_and_burn(tok2, "login", "m4p7") is True
    assert store.verify_and_burn(tok2, "login", "m4p7") is False    # 二次重放失败
    print("[PASS] All rate-limit defense checks passed.")


if __name__ == "__main__":
    run_self_checks()

判定通过标准：

• X-Forwarded-For 与 X-Internal-Request 等伪造头无法改变限流计数键，仅当对端 IP 位于可信代理 CIDR 内时才解析代理链；
• 同一 IPv6 /64 内任意地址请求命中同一限流桶；
• 用户名空白/大小写/Unicode 变体与手机号区号变体归并为相同规范键，含空字节或同形异体字符直接返回 400；
• 30 并发请求下成功放行数严格等于阈值 5，无超发；
• 验证码 Token 无论验证成功或失败均在单次核销后失效。

MCP 工具映射

防御阶段	MCP 工具	用途说明
边界与限流响应基线检查	http_probe	验证各端点在伪造代理头、重复请求与高并发下的 429/Retry-After 一致性
关联防御知识检索	kb_router	按 rate limit defense、brute force、CAPTCHA 检索关联防御文档
读取技术细节	kb_read_file	读取本文档或对应攻击面文档 01-rate-limit-bypass.md 进行对照
知识目录定位	kb_catalog	列出 16-rate-limit 与认证防御板块的完整文档清单