API 滥用拒绝服务防御与配额治理解析

防御目标与对抗矩阵

API 滥用型 DoS（对应攻击向量见 10-api-abuse.md）常由持有合法身份 Token 的调用方触发，传统基于异常报文签名的 WAF 难以拦截。防御核心是为所有计算与 I/O 放大型接口建立成本预算（Cost Budget）与资源隔离舱。

攻击向量	放大/消耗机制	核心防御手段	硬性预算上限
深度分页遍历	OFFSET 10,000,000 触发全表扫描或可预测游标尾部扫描	强制 Keyset（Seek）分页 + HMAC 签名不透明游标 + 最大页码硬限	max_page <= 200（若需深查强制走异步导出），per_page <= 100
批量操作与导出 OOM	单请求操作 10,000 对象或全表构建 XLSX 触发 OOM	限制批量数组长度 + 导出改为后台异步队列与流式 CSV 写盘（对象存储签名下载）	单次 bulk <= 100 条，禁止同步全量导出
API 限速绕过	伪造 XFF、跨窗口突发、多端点/多版本并发、HTTP/2 复用	仅信任可信网关写入的真实 IP、采用平滑滑动窗口/漏桶、按 user_id/tenant_id 全局共享配额	跨端点共享总成本配额，禁止仅按连接或未校验 XFF 计费
搜索 API 滥用	q=*、~10 模糊、正则检索与三层 terms 聚合爆炸	关闭通配符前导匹配与脚本查询、限制聚合桶数与分页深度	ES max_buckets <= 1000、max_result_window <= 5000、script.allowed_types: none
Webhook 慢速回调耗尽	目标站点慢速响应拖垮出站连接池或回调 SSRF	独立异步 Worker 队列 + 强制 connect=2s, read=5s 超时 + 熔断器 + 目标 IP 黑名单	单租户并发出站 <= 5，连续失败自动熔断禁用 Hook
文件上传 DoS	1x1 像素声明 1e6 x 1e6 炸弹、大量小文件耗尽 inode、慢速大上传	对象存储直传（Presigned URL）+ 上传前校验像素头尺寸 + 限制文件数与体大小	MAX_IMAGE_PIXELS=25_000_000，单请求文件数 <= 5

---

防御工作流

阶段一：分页、批量操作与搜索成本硬限

1. Keyset 分页替代 OFFSET：通过索引列锚点查询（WHERE id < :last_id ORDER BY id DESC LIMIT :limit）保持恒定 $O(\log n)$ 复杂度，并将游标载荷使用 HMAC-SHA256 签名，阻止伪造极端尾部游标。
2. Elasticsearch / OpenSearch 防搜索爆炸配置：在集群配置与查询网关层禁止高开销查询（search.allow_expensive_queries: false 禁用前导通配符、正则和昂贵脚本），并将 search.max_buckets 与 index.max_result_window 收紧至安全上限：

# elasticsearch.yml — 阻断聚合爆炸与昂贵查询
search.max_buckets: 1000
search.allow_expensive_queries: false
script.allowed_types: none
indices.query.bool.max_clause_count: 256

阶段二：防伪造全局限流与 Webhook 舱壁隔离

1. 消除 X-Forwarded-For 伪造绕过：在 Nginx/Envoy 严格配置 set_real_ip_from 仅信任上游网关 CIDR，real_ip_recursive on;，应用层严禁直接信任原始请求头中的 X-Forwarded-For 作为限流 Key；同时按认证主体 sub / tenant_id 结合滑动窗口实施全局积分扣减，消除时间窗口边界 2x 突发。
2. Webhook 异步队列与熔断隔离：主请求线程只向消息队列投递事件；由隔离的 Webhook Worker 消费，限制出站总并发与单域名并发，设置 timeout=(2.0, 5.0) 与最大响应体读取限制（<= 64KB），并校验目标解析 IP 非内网段。

阶段三：文件上传防 Pixel Flood 与 Inode 耗尽

1. 采用对象存储客户端直传：大文件不经过应用服务器磁盘与内存，彻底隔绝慢速大文件占用 Worker 与本地磁盘 inode 耗尽。
2. 图片解码前尺寸核验：在调用任何图像缩放/转码库（Pillow/ImageMagick/libvips）前，限制 Image.MAX_IMAGE_PIXELS 并读取文件头确认宽高与文件大小在预算内，同时禁用 ImageMagick 高危格式解码器（如 MVG/SVG/EPHEMERAL）。

---

弹性校验代码（可运行 Python 自检）

以下 Python 代码实现四个关键防御守卫：（1）HMAC-SHA256 签名分页游标与 per_page 上限校验；（2）防窗口边界突发的滑动窗口限流器；（3）Bulk 批量条目与搜索 Query 通配符拦截；（4）防 Pixel Flood 的 PNG IHDR 宽高硬校验（无需完整解码即可阻断千万级像素欺骗）。

#!/usr/bin/env python3
"""API abuse DoS defense verifier: signed keyset cursor, sliding-window limiter, search & PNG pixel-flood guards."""

import base64
import hashlib
import hmac
import struct
import time
from collections import deque


# 示例签名密钥（生产环境从密钥管理服务/环境变量注入）
CURSOR_HMAC_SECRET = b"ephemeral-selfcheck-key-change-in-prod"


class ApiGuardError(ValueError):
    """Raised when an API parameter violates DoS cost boundaries."""


def issue_signed_cursor(last_id: int) -> str:
    payload = str(last_id).encode("ascii")
    sig = hmac.new(CURSOR_HMAC_SECRET, payload, hashlib.sha256).digest()[:16]
    return base64.urlsafe_b64encode(payload + b"." + sig).decode("ascii")


def verify_signed_cursor(token: str, per_page: int, max_per_page: int = 100) -> int:
    if not (1 <= per_page <= max_per_page):
        raise ApiGuardError(f"per_page {per_page} out of [1, {max_per_page}]")
    try:
        raw = base64.urlsafe_b64decode(token.encode("ascii"))
        payload, sig = raw.rsplit(b".", 1)
    except Exception as exc:
        raise ApiGuardError("malformed cursor") from exc
    expected = hmac.new(CURSOR_HMAC_SECRET, payload, hashlib.sha256).digest()[:16]
    if not hmac.compare_digest(sig, expected):
        raise ApiGuardError("cursor signature mismatch")
    return int(payload.decode("ascii"))


class SlidingWindowLimiter:
    """Eliminates fixed-window boundary burst (2x quota) bypasses."""

    def __init__(self, limit: int, window_sec: float) -> None:
        self.limit = limit
        self.window_sec = window_sec
        self.buckets: dict[str, deque[float]] = {}

    def allow(self, subject_id: str, cost: int = 1, now: float | None = None) -> bool:
        ts = time.monotonic() if now is None else now
        q = self.buckets.setdefault(subject_id, deque())
        cutoff = ts - self.window_sec
        while q and q[0] <= cutoff:
            q.popleft()
        if len(q) + cost > self.limit:
            return False
        for _ in range(cost):
            q.append(ts)
        return True


def validate_search_query(q: str, max_len: int = 128) -> None:
    """Reject broad wildcard/regex queries that trigger full-index scans."""
    cleaned = q.strip()
    if not cleaned or len(cleaned) > max_len:
        raise ApiGuardError("query length out of bounds")
    if cleaned in {"*", "?"} or cleaned.startswith(("*", "?", "/")):
        raise ApiGuardError("leading wildcard or regex query forbidden")
    if len(cleaned) < 2:
        raise ApiGuardError("query too broad (<2 chars)")


def validate_png_dimensions(png_bytes: bytes, max_pixels: int = 25_000_000, max_dim: int = 8192) -> tuple[int, int]:
    """Inspect PNG IHDR header before image decoding to stop Pixel Flood OOM bombs."""
    if len(png_bytes) < 24 or png_bytes[:8] != b"\x89PNG\r\n\x1a\n" or png_bytes[12:16] != b"IHDR":
        raise ApiGuardError("invalid PNG header")
    width, height = struct.unpack(">II", png_bytes[16:24])
    if width == 0 or height == 0 or width > max_dim or height > max_dim:
        raise ApiGuardError(f"PNG dimension {width}x{height} exceeds {max_dim}")
    if width * height > max_pixels:
        raise ApiGuardError(f"PNG pixel count {width * height} exceeds {max_pixels}")
    return width, height


def run_self_checks() -> None:
    # 1. 签名游标防篡改与分页大小校验
    cur = issue_signed_cursor(1024)
    assert verify_signed_cursor(cur, per_page=20) == 1024
    for bad_call in (
        lambda: verify_signed_cursor(cur, per_page=5000),
        lambda: verify_signed_cursor(base64.urlsafe_b64encode(b"9999999.bad").decode(), per_page=20),
    ):
        try:
            bad_call()
            raise AssertionError("Cursor guard failed to block invalid pagination")
        except ApiGuardError:
            pass

    # 2. 滑动窗口限流防边界 2x 突发校验
    limiter = SlidingWindowLimiter(limit=5, window_sec=10.0)
    for i in range(5):
        assert limiter.allow("user-1", cost=1, now=9.9 + i * 0.01)
    assert not limiter.allow("user-1", cost=1, now=10.1), "Boundary burst bypassed sliding window"

    # 3. 宽泛搜索拦截校验
    for bad_q in ("*", "*admin", "a", "/(a+)+/"):
        try:
            validate_search_query(bad_q)
            raise AssertionError(f"Broad query {bad_q} was not blocked")
        except ApiGuardError:
            pass

    # 4. PNG Pixel Flood 头校验（声明 1,000,000 x 1,000,000）
    evil_ihdr = b"\x89PNG\r\n\x1a\n" + struct.pack(">I", 13) + b"IHDR" + struct.pack(">II", 1_000_000, 1_000_000) + b"\x08\x02\x00\x00\x00"
    try:
        validate_png_dimensions(evil_ihdr)
        raise AssertionError("Pixel flood PNG was not blocked")
    except ApiGuardError:
        pass
    good_ihdr = b"\x89PNG\r\n\x1a\n" + struct.pack(">I", 13) + b"IHDR" + struct.pack(">II", 1920, 1080) + b"\x08\x02\x00\x00\x00"
    assert validate_png_dimensions(good_ihdr) == (1920, 1080)
    print("[PASS] All API abuse DoS defense checks passed.")


if __name__ == "__main__":
    run_self_checks()

---

验证与判定标准

验证项	达标判定标准	可观测指标
分页与批量上限	page > max_page、篡改游标或 bulk 超出 100 项直接返回 400	数据库查询计划维持 Index Scan，无 SeqScan 尖峰
全局限流不可绕过	伪造 X-Forwarded-For 或多版本端点并发时仍触发同一主体 429 限流	跨窗口突发量不超过单窗口配额
搜索查询拦截	q=*、单字符或深度嵌套聚合返回 400 且未触发 ES OOM	ES 堆内存稳定 < 65%
Webhook 隔离	回调慢速目标超过 5s 自动超时，主业务 API 延迟不受影响	主接口 P99 延迟波动 < 10ms
Pixel Flood 阻断	篡改 IHDR 的恶意图片在解码前直接返回 400	图片处理 Worker 内存稳定，零 OOM 重启

---

MCP 工具映射

防御阶段	MCP 工具	说明
API 限流与参数边界验证	http_probe	验证分页上限、429 限流响应及宽泛搜索拦截
蓝队技术搜索	kb_router	以 track="blue" 搜索 api / rate_limiting / webhook 防御
攻防文档对照	kb_read_file	读取本文件及配对攻击文档 ctf-website/22-dos/10-api-abuse
分类清单查看	kb_catalog	列举 22-dos 下所有蓝队加固条目