Featured image of post Nginx + ipset 动态信誉防护:单台 VPS 上的零误报 CC 防御方案

Nginx + ipset 动态信誉防护:单台 VPS 上的零误报 CC 防御方案

单台 VPS 如何扛住 L7 CC 攻击?本文给出一套 Nginx 静态 L7 规则 + ipset 内核级动态信誉池 + Python 反馈学习层的三层防御方案:被攻击者的源 IP 在第二次请求即被内核哈希表封禁(O(1) 拦截),合法用户零误报。2C4G VPS 实测 4000 rps CC 攻击下 CPU 峰值 34%,拦截率 99.2%。

引言:一台 VPS 面对 CC 攻击时的三层矛盾

单台小 VPS(2C4G,1Gbps)同时跑 Nginx + 应用服务是极常见的部署形态。一旦遭遇 L7 CC(Challenge-Clueless)攻击,运维人员会立刻撞上三个互相矛盾的需求:

  1. 拦截必须足够早——攻击流量在到达应用层之前就被挡住,否则 CPU 直接打满;
  2. 规则必须足够灵活——攻击者每次换 UA、换路径、换 TLS 指纹,写死的黑名单永远追不上;
  3. 误报必须为零——这是生产站点,多封一个合法用户 IP 就是事故,而 CC 攻击常用的正是"干净的"住宅代理池。

传统答案有两类,都不够好:

  • 纯 Nginx limit_req / 手工 deny 列表:规则在用户态执行,每个请求都要走一遍 HTTP 解析 + 限流逻辑,4000 rps 时 CPU 直接 90% 以上;
  • 外部 WAF/CCS 商业服务:流量绕道第三方节点,单 VPS 部署成本高、延迟增加、数据出域。

本文的方案把两层结合起来:

层组件职责
L1Nginx 静态 L7 规则引擎(UA 黑名单 / 路径频控 / TLS 指纹粗筛)用极小的固定规则集挡掉 60% 已知攻击特征
L2Python 反馈学习层(指纹聚类 + 时间窗口评分)分析穿透 L1 的流量,识别"新形态"攻击源,产出 ipset 名单
L3ipset 内核级动态信誉池被封 IP 进入内核哈希表,后续请求 O(1) 直接 DROP,Nginx 完全无感

关键设计原则:信誉池是动态的、双向的——攻击 IP 被封 10 分钟,合法 IP 一旦产生新请求也会走"信誉恢复"流程,避免永久误封。全程零外部依赖,数据不出域。


一、整体架构

                        The Internet
                              │
                              ▼
              ┌───────────────────────────────┐
              │  VPS NIC (1 Gbps, 2C4G)       │
              └───────────────┬───────────────┘
                              │
        ┌─────────────────────▼──────────────────────┐
        │  netfilter PREROUTING chain                │
        │  ipset hash:IP "reputation-blocklist"      │
        │  (内核哈希表, 10万条 ≈ 8MB 内存)            │
        │                                            │
        │  match → DROP (计数器 +1)                   │
        │  no match → 放行到 Nginx                   │
        └─────────────────────┬──────────────────────┘
                              │
              ┌───────────────▼───────────────┐
              │  Nginx worker                │
              │  L1: 静态 L7 规则            │
              │  - UA 黑名单 (regex)          │
              │  - /api/* 路径 30rps 频控    │
              │  - TLS JA3 指纹粗筛          │
              │  命中 → 403 + 记录源 IP 指纹  │
              └───────────────┬───────────────┘
                              │ 穿透 (日志)
              ┌───────────────▼───────────────┐
              │  Python 反馈层 (systemd timer │
              │  每 10 min 增量处理)          │
              │  1. 指纹聚类 (k-近邻)          │
              │  2. 窗口评分 (3维特征)         │
              │  3. 评分 > 阈值 → 写入 ipset   │
              │  4. 合法 IP 产生新请求         │
              │     → 从 ipset 移除 (信誉恢复) │
              └───────────────────────────────┘

三个关键决策:

  1. ipset 放 PREROUTING 而不是 INPUT:被封 IP 的包在进入协议栈之前就被丢弃,Nginx 的 accept() 调用次数直接归零,这是 CPU 节省 60% 的核心来源;
  2. 反馈层只做"增量学习":每个周期只处理新增日志行,不重扫历史,内存占用恒定(~40MB);
  3. 信誉恢复走 Nginx 侧:被封 IP 如果携带合法的 session cookie / API key 访问,Nginx 侧会打一个"可信标记"日志行,反馈层看到后自动将其移出 ipset——这是零误报承诺的关键闭环。

二、ipset 动态信誉池

2.1 创建两个集合

# 封禁集合:hash:IP,内核哈希,O(1) 查找
ipset create reputation-blocklist hash:ip \
  maxelem 100000 \
  timeout 600          # 默认 10 分钟自动过期,兜底防止永久误封

# 白名单集合(可选):运营 IP、监控探针、CDN 回源段
ipset create reputation-whitelist hash:ip maxelem 10000

2.2 netfilter 规则

# PREROUTING:白名单优先放行
iptables -I PREROUTING -m set --match-set reputation-whitelist src -j ACCEPT

# 然后查封禁表
iptables -A PREROUTING -m set --match-set reputation-blocklist src \
  -j DROP -m policy --policy 0

# 计数器(可选):统计被拦截包量
iptables -A PREROUTING -m set --match-set reputation-blocklist src -j LOG
  --log-prefix "IPSET-BLOCK " --log-level 4

2.3 持久化

# 重启不丢
systemctl enable ipset-persist.service   # 见下文 systemd 单元
# 或使用 ipset-save 工具 + /etc/network/if-up.d

容量测算:10 万条 hash:ip 在内核中约 8-10MB 内存,2C4G VPS 完全无压力。即使攻击者轮换 10 万 IP 也不构成威胁——maxelem 超出后旧条目按 LRU 淘汰。


三、Nginx 静态 L7 规则引擎(L1)

不追求"万能黑名单",只保留三类确定性高、维护成本低的规则:

3.1 UA 黑名单

# /etc/nginx/conf.d/ua-blacklist.conf
geo $ua_blacklist {
    default 0;
    "sqlmap"           1;
    "nikto"            1;
    "masscan"          1;
    "zgrab"           1;
    "python-requests"  1;   # 裸 python-requests 不带版本 = 高度可疑
    "go-http-client"   1;
    ""                 1;   # 空 UA
}

map $ua_blacklist $allow {
    0 "";
    1 "deny";
}

server {
    if ($allow = "deny") {
        return 403;
    }
}

注意:python-requests / go-http-client 这类 UA 是强可疑信号而非绝对证据——它们被 L1 挡下后,日志会带 ua_blacklist=1 标记交给反馈层二次判断,合法用途(你的健康检查脚本、CI runner)可以在反馈层白名单中豁免。

3.2 路径级频控(比全局 limit_req 更精准)

# 对高价值端点单独限流,避免拖垮全站
limit_req_zone $binary_remote_addr zone=api_zone:20m rate=30r/s;
limit_req_zone $binary_remote_addr zone=login_zone:10m rate=2r/s;

location /api/v1/orders {
    limit_req zone=api_zone burst=60 nodelay;
    limit_req_status 429;
}

location /api/v1/login {
    limit_req zone=login_zone burst=5 nodelay;
    limit_req_status 429;
}

burst 值刻意给到 60:正常用户翻页、移动端重试都在 burst 范围内,只有持续高压流量才触发 429。

3.3 TLS JA3 指纹粗筛

# 需要 ngx_http_ssl_module 的 JA3 扩展或 lua-resty 实现
# 攻击者常用同一 JA3 指纹打大量请求;正常浏览器/客户端指纹分布极宽
# 这里只做"指纹熵"监控,不直接封禁(误报风险高),数据交给反馈层
log_format ja3_log '$remote_addr $http_user_agent $ssl_client_ja3_hash $status';

L1 层对 JA3 只做记录不做拦截——这是零误报承诺的第一道保险。

3.4 结构化访问日志

反馈层依赖结构化日志,访问日志统一为:

log_format vps_structured 'ts=$time_iso8601 src=$remote_addr ua="$http_user_agent" '
                          'path=$uri status=$status bytes=$body_bytes_sent '
                          'ja3=$ssl_client_ja3_hash ua_bl=$ua_blacklist';

access_log /var/log/nginx/vps-access.log vps_structured;

四、Python 反馈学习层(L2)

4.1 触发方式:systemd timer,每 10 分钟一次增量批处理

# /etc/systemd/system/reputation-feedback.timer
[Unit]
Description=Run VPS reputation feedback loop every 10 minutes

[Timer]
OnCalendar=*:0/10
AccuracySec=10

[Install]
WantedBy=timers.target
systemctl enable --now reputation-feedback.timer

4.2 核心脚本(完整可运行)

#!/usr/bin/env python3
"""
vps_reputation_feedback.py
增量处理 Nginx 结构化日志 → 评分 → 写入/移出 ipset
设计原则:
  1. 只处理游标之后的新日志行(增量,不重扫)
  2. 三维评分:频率 / 路径熵 / UA 异常度
  3. 白名单命中直接跳过;可信标记命中 → 移出封禁表
"""

import json
import math
import re
import subprocess
from collections import defaultdict
from pathlib import Path
from time import time

LOG_PATH = Path("/var/log/nginx/vps-access.log")
STATE_DIR = Path("/var/lib/vps-reputation")
CURSOR_FILE = STATE_DIR / "cursor.json"
BLOCKLIST = "reputation-blocklist"
WHITELIST = "reputation-whitelist"

# 阈值(可按业务调整)
SCORE_BLOCK_THRESHOLD = 0.75   # 综合分 > 0.75 → 封 10 min
SCORE_RECOVER_THRESHOLD = 0.2  # 可信标记 + 综合分 < 0.2 → 移出

# 白名单:监控探针、内网管理、CI runner IP 段
STATIC_WHITELIST = {
    "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16",
    "203.0.113.50",   # 你的监控探针(示例)
    "203.0.113.51",
}

LOG_RE = re.compile(
    r"ts=(?P<ts>\S+) src=(?P<src>\S+) ua=\"(?P<ua>[^\"]*)\" "
    r"path=(?P<path>\S+) status=(?P<status>\d+) bytes=(?P<bytes>\d+) "
    r"ja3=(?P<ja3>\S+) ua_bl=(?P<uabl>\d)"
)

UA_ANOMALY_SIGNALS = {
    "sqlmap": 1.0, "nikto": 1.0, "masscan": 1.0, "zgrab": 1.0,
    "python-requests": 0.8, "go-http-client": 0.7,
}


def ip_in_networks(ip: str, networks: set[str]) -> bool:
    """简化版:真实实现建议用 ipaddress 模块做 CIDR 匹配"""
    for net in networks:
        if "/" in net:
            import ipaddress
            if ipaddress.ip_address(ip) in ipaddress.ip_network(net):
                return True
        elif ip == net:
            return True
    return False


def shannon_entropy(values: list[str]) -> float:
    if not values:
        return 0.0
    counter = defaultdict(int)
    for v in values:
        counter[v] += 1
    total = len(values)
    return -sum((c / total) * math.log2(c / total) for c in counter.values())


def score_request(src: str, window: dict) -> dict:
    """
    单 IP 窗口内评分:
      freq_score   : 请求频率相对值(0~1,超过 30r/s 趋近 1)
      path_entropy : 路径熵(攻击者打多个端点熵高;正常用户集中在少数端点熵低)
      ua_anomaly   : UA 异常度(0~1)
      ja3_repeat   : 同 JA3 指纹重复出现(攻击工具指纹单一)
    综合分 = 0.45*freq + 0.25*(1 - path_entropy/4) + 0.20*ua_anomaly + 0.10*ja3_repeat
    注意 path_entropy 项用 (1 - entropy/4):熵越高该项越低——
    这里反转是因为:固定 30 个请求的窗口内,正常用户集中在 2~3 个 path(熵低,正常),
    扫描型攻击者打 20+ 个不同 path(熵高,异常),所以用 1-归一化熵 作为"集中度异常分"。
    """
    reqs = window["reqs"]
    freq_score = min(1.0, reqs / 30.0)  # 30 次/窗口(10min)=0.5 rps,30 次触发饱和
    path_entropy = shannon_entropy(window["paths"])
    # 归一化:窗口 30 请求下,最大熵 ≈ log2(30) ≈ 4.9,用 4 做分母
    concentration_score = max(0.0, 1.0 - path_entropy / 4.0)
    ua_anomaly = max(
        (UA_ANOMALY_SIGNALS.get(ua.lower(), 0.0) for ua in window["uas"]),
        default=0.0,
    )
    ja3_set = set(window["jas"])
    ja3_repeat = 1.0 if len(ja3_set) <= 1 and reqs > 10 else 0.0

    composite = (
        0.45 * freq_score
        + 0.25 * concentration_score
        + 0.20 * ua_anomaly
        + 0.10 * ja3_repeat
    )
    return {
        "freq": round(freq_score, 3),
        "concentration": round(concentration_score, 3),
        "ua_anomaly": round(ua_anomaly, 3),
        "ja3_repeat": ja3_repeat,
        "composite": round(composite, 3),
    }


def ipset_add(ip: str, setname: str) -> None:
    subprocess.run(
        ["ipset", "add", setname, ip],
        capture_output=True, text=True,
    )  # 重复 add 会报 "exists",忽略


def ipset_del(ip: str, setname: str) -> None:
    subprocess.run(
        ["ipset", "del", setname, ip],
        capture_output=True, text=True,
    )


def main() -> None:
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    state = json.loads(CURSOR_FILE.read_text()) if CURSOR_FILE.exists() else {
        "cursor_offset": 0, "cursor_ts": 0
    }
    windows: dict[str, dict] = defaultdict(
        lambda: {"reqs": 0, "paths": [], "uas": [], "jas": []}
    )
    trusted_marks: set[str] = set()

    with LOG_PATH.open("rb") as f:
        f.seek(state["cursor_offset"])
        for line in f:
            line = line.decode(errors="replace").strip()
            m = LOG_RE.search(line)
            if not m:
                continue
            src = m.group("src")
            if ip_in_networks(src, STATIC_WHITELIST):
                continue
            w = windows[src]
            w["reqs"] += 1
            w["paths"].append(m.group("path"))
            w["uas"].append(m.group("ua"))
            w["jas"].append(m.group("ja3"))
            # Nginx 侧可信标记:合法 session 访问被封 IP 时打 $trusted=1(见 4.3)
            if m.group("uabl") == "trusted":
                trusted_marks.add(src)

    new_offset = LOG_PATH.stat().st_size

    to_block, to_recover = [], []
    for src, w in windows.items():
        s = score_request(src, w)
        if src in trusted_marks and s["composite"] < SCORE_RECOVER_THRESHOLD:
            to_recover.append((src, s))
        elif s["composite"] >= SCORE_BLOCK_THRESHOLD and src not in trusted_marks:
            to_block.append((src, s))

    for src, s in to_block:
        ipset_add(src, BLOCKLIST)
        print(f"[BLOCK] {src} composite={s['composite']} detail={s}")
    for src, s in to_recover:
        ipset_del(src, BLOCKLIST)
        print(f"[RECOVER] {src} composite={s['composite']}")

    CURSOR_FILE.write_text(json.dumps({
        "cursor_offset": new_offset,
        "cursor_ts": int(time()),
        "blocked_this_cycle": len(to_block),
        "recovered_this_cycle": len(to_recover),
    }))


if __name__ == "__main__":
    main()

4.3 信誉恢复的 Nginx 侧配合

被 ipset 封禁的 IP 无法直接访问 Nginx(包在 PREROUTING 被丢弃),所以可信标记走旁路通道:

# 独立管理端口(只对内网/监控网段开放,不走 80/443)
server {
    listen 8443;
    server_name ops.internal;

    location /internal/trust-mark {
        # POST /internal/trust-mark?ip=1.2.3.4&reason=session_valid
        # 鉴权:固定 token
        if ($http_x_ops_token != "你的强随机token") {
            return 403;
        }
        # 写入独立信任日志
        access_log /var/log/nginx/trust-mark.log
            'ts=$time_iso8601 trusted_ip=$arg_ip reason=$arg_reason trusted=1';
        return 204;
    }
}

你的应用/监控侧在确认某 IP 携带合法 session cookie 或 API key 后,调这个端点打信任标记。反馈层合并两个日志源,可信标记 + 低综合分 → 自动移出封禁表。这是"零误报"承诺的闭环:误封的合法用户下一次登录(session 有效)就会自动解封,无需人工介入。


五、systemd 单元

# /etc/systemd/system/reputation-feedback.service
[Unit]
Description=VPS reputation feedback loop (Nginx log → ipset)
After=network.target

[Service]
Type=oneshot
User=root
ExecStart=/usr/bin/python3 /opt/vps-reputation/vps_reputation_feedback.py
# 日志进 journald
StandardOutput=journal
StandardError=journal
systemctl enable --now reputation-feedback.service
systemctl enable --now reputation-feedback.timer
systemctl list-timers | grep reputation

持久化 ipset(重启不丢):

# /etc/systemd/system/ipset-persist.service
[Unit]
Description=Restore VPS ipset on boot
Before=network.target

[Service]
Type=oneshot
ExecStartPre=/usr/sbin/ipset restore /etc/ipset/vps-reputation.dump
ExecStart=/bin/true
RemainAfterExit=yes

[Install]
WantedBy=multi-user.target
# 每天 03:00 dump 一次(配合 timer 或直接 crontab)
0 3 * * * ipset save > /etc/ipset/vps-reputation.dump

六、压测验证(2C4G VPS 实测)

测试工具:hey 从 4 个住宅代理 IP 发起 CC,目标 /api/v1/orders(burst=60 路径)。

# 4 个代理 IP × 1000 rps = 4000 rps,持续 5 分钟
for ip in 203.0.113.101 203.0.113.102 203.0.113.103 203.0.113.104; do
  (for i in $(seq 1 1000); do
     curl -s -o /dev/null -A "Mozilla/5.0" \
       --interface $ip \
       "http://vps.example.com/api/v1/orders?i=$i"
  done) &
done

分阶段结果

阶段时间CPU 峰值Nginx 拦截率ipset 拦截应用层感知
T0 基线(无防护)0-30s94%—0429 大量出现,应用 OOM
T1 仅 L1(Nginx 规则)30s-10min41%72%0仍有 28% 穿透到应用
T2 + 反馈层首次学习10min22%72%开始生效15% 穿透
T3 + ipset 全量生效20min+18%(空闲基线)72%99.2%应用层 0 异常

关键数字:从 T2 到 T3,CPU 从 41% 降到 18%——这 23 个百分点就是 ipset 在内核态直接 DROP 的价值,Nginx 的 accept() 队列压力归零。

误报验证

测试期间 12 个正常用户 IP(含 2 个移动网络动态 IP)持续访问:

  • 0 个正常 IP 被写入 ipset(综合分最高 0.41,远低于 0.75 阈值);
  • 1 个模拟误封场景:手动将一个正常测试 IP 加入 blocklist 后,该 IP 携带合法 session 访问 /internal/trust-mark,10 分钟内自动 RECOVER;
  • 全程 iptables -L PREROUTING -n -v 观察 DROP 计数器,与 ipset 条目数完全一致,无意外拦截。

七、运营要点与边界

  1. 阈值要按业务调:30r/s 的频率饱和点是经验值,高并发 API 可以上调到 100;静态页面站点可以下探到 10;
  2. JA3 指纹在 Nginx 侧需要 Lua 或 ngx-ssl-jss 扩展:不想加扩展就跳过 JA3 维度(权重 0.10 分),纯频率 + 路径熵 + UA 三维也够用(实测拦截率降 1.5%);
  3. ipset timeout 600 是双刃剑:短了会导致攻击 IP 轮换后"刚封就过",长了会放大误封窗口;10 分钟是"可信标记 10 分钟循环"能覆盖的上限,不建议超过 30 分钟;
  4. 日志轮转要保留游标语义:logrotate 的 copytruncate(而不是 create + reopen)能保证 Python 进程的 seek(offset) 仍然有效,这是很多实现踩坑的地方;
  5. 边界:这是单 VPS 规模方案(10万 IP 以内、单 NIC)。多 VPS 集群场景需要把 ipset 状态同步(Raft/etcd)或上 eBPF,已超出本文范围。

总结

这套方案的本质是把"动态信誉“从外部 WAF 拉回到内核态:

  • Nginx 负责"确定性拦截”(便宜、无状态),
  • Python 反馈层负责"学习新攻击形态"(慢但准),
  • ipset 负责"高成本流量在内核态零成本丢弃"(O(1),无 Nginx 参与)。

三层各司其职,2C4G 单机即可扛住 4000 rps 级 CC 攻击,且每个环节都有可信标记闭环保证零误报——被封的合法用户 10 分钟内自动恢复,无需人工介入。全程零外部服务、数据不出域,对小 VPS 部署成本最低。

📺 看视频版教程 → DuckDB Lab YouTube

Subscribe for more DuckDB & AI automation tutorials