绝杀慢速扫描：我用 Claude 打造自适应异常流量检测器，让 nftables 自动安全拦截高危端口

Web极客码

178人浏览 · 2026-07-17 15:34:57

Web极客码 · 2026-07-17 15:34:57 发布

一、痛点：死板的防火墙规则，正在被黑客“降频绕过”

作为运营着多个大流量体育和游戏资讯站的独立站长，我的服务器每天都要直面来自全球的恶意端口扫描、SSH/FTP 暴力破解，以及低频的 DDoS 穿透攻击。

在过去，我主要依赖像 CSF（ConfigServer Security & Firewall）或 Fail2ban 这类经典的静态规则防火墙。然而，这种基于固定频次（例如：“10 秒内 TCP 连接超过 5 次则封禁”）的防御机制，在如今复杂的对抗环境中开始显得力不从心。

现在的黑客极其狡猾，他们普遍采用了“慢速扫描（Slow Scanning）”或“分布式低频探测”策略：

攻击者将单个 IP 的请求频次刻意控制在传统防火墙的检测临界值以下（例如每 30 秒仅探测一次）。
这种“温水煮青蛙”的探测方式能够堂而皇之地避开所有静态规则拦截，在不触发任何警报的情况下，默默摸清我服务器的开放端口，并对 client 弱口令进行静默暴力破解。

死板的过滤规则已经沦为防线上的摆设，我急需一套能够识别“行为特征”，而非单纯计算“连接频次”的自适应防御系统。

二、破局：用 Claude 构筑“行为级”流量分析与 nftables 联动流

为了给服务器加装一套聪明的“安全大脑”，我利用 Claude 对系统行为分析与轻量级算法设计的极强推理能力，结对编写了一套超轻量级、无需复杂依赖的自适应流量检测与自动封禁脚本。

这套方案的底层核心由三大防护模块紧密咬合而成：

1. 实时多维度日志流管道化（Log Piping）

我编写了一个高效的后台守护进程，通过 Linux 命名管道（Named Pipe），将系统的 Nginx 访问日志、/var/log/secure（SSH 登录尝试日志）以及系统内核的 nftables 连接跟踪表（ConnTrack） 进行实时聚合流式输出。所有进站连接的源 IP、目标端口、TCP 标志位和请求头特征，都会被不间断地送入分析引擎。

2. 行为特征提取与智能检测算法（轻量级、非大模型在线运行）

如果把庞大的大模型直接放在服务器上做实时流量推理，不仅延迟极高，更会瞬间挤爆 CPU。

为此，我采取了折中且优雅的方案：让 Claude 帮我设计算法，由本地 Python 脚本高效执行。

Claude 帮我提取并实现了一套基于统计特征与行为熵值的轻量级检测算法：

多端口探测熵值分析： 正常用户访问通常高度聚焦在 80/443 等特定业务端口。如果一个 IP 在一段时间内，访问的端口分布呈现出高熵值特征（即随机探测了多个非公开端口或死端口，如 3389、1433、22 等），即使其每次探测间隔长达数分钟，也会瞬间被算法标记。
非标准请求头聚集度： 自动审计 Nginx 日志中畸形的 HTTP 请求头、不合规的 User-Agent 聚集度，识别潜在的自动化扫描探针。

3. 基于 `nftables API` 的毫秒级熔断

一旦本地的 Python 检测脚本确认某个 IP 触发了恶意扫描的行为特征，它不会去调用效率低下的 iptables 规则重载，而是直接调用 nftables API，将该恶意 IP 秒级丢入系统内核层的动态黑名单集合（nftables set）中，强制封禁 24 小时。在成功封禁后，系统会自动触发一个 Webhook，向我的手机发送即时警报。

Plaintext

[ Nginx 日志 / 系统安全日志 / ConnTrack 跟踪表 ] 
                       │ 
                       ▼ (管道化流式输出)
         [ 本地自适应 Python 检测引擎 ] 
                       │ (行为熵值 / 慢速特征分析)
                       ▼ (确认扫描行为)
          [ 调用 nftables API 注入 Set ] 
                       │ 
         ┌─────────────┴─────────────┐
         ▼ (毫秒级系统层阻断)          ▼ (即时警报)
   [ 封禁该 IP 24 小时 ]         [ 手机推送通知 ]

三、核心实现：Claude 辅助生成的自适应拦截逻辑

在构建检测逻辑时，Claude 帮我写出了一个精简且性能极佳的 Python 行为监控核心。它能够高效管理内存中的 IP 访问状态，防止恶意 IP 采用“内存占满攻击（Memory Exhaustion）”撑爆服务器的运行内存：

Python

import time
import math
from collections import defaultdict, deque

# 配置参数
WINDOW_SIZE = 300  # 统计窗口：5分钟（300秒）
ENTROPY_THRESHOLD = 1.5  # 端口访问熵值阈值，超过此值判定为扫描
MAX_TRACKED_IPS = 5000  # 最大追踪 IP 数量，防止内存溢出

# 内存状态追踪：使用 deque 自动淘汰过期记录
ip_port_history = defaultdict(lambda: deque(maxlen=100))

def calculate_entropy(port_counts):
    """
    由 Claude 优化生成的香农熵计算函数，用于评估 IP 访问端口的随机度
    """
    total_scans = sum(port_counts.values())
    if total_scans < 3:  # 样本量太少时不进行熵值计算，防止误判
        return 0.0
    
    entropy = 0.0
    for count in port_counts.values():
        p = count / total_scans
        entropy -= p * math.log2(p)
    return entropy

def process_connection_event(ip, port):
    """
    处理单次连接事件，评估是否需要熔断
    """
    # 简单的滑动窗口清理，防止内存占满
    if len(ip_port_history) > MAX_TRACKED_IPS:
        # 简单淘汰一个过期的 IP 记录
        expired_ip = next(iter(ip_port_history))
        del ip_port_history[expired_ip]

    current_time = time.time()
    ip_port_history[ip].append((port, current_time))

    # 统计当前窗口内该 IP 访问的不同端口分布
    port_counts = defaultdict(int)
    for p, t in list(ip_port_history[ip]):
        if current_time - t <= WINDOW_SIZE:
            port_counts[p] += 1

    # 计算端口扫描熵值
    entropy = calculate_entropy(port_counts)
    
    if entropy > ENTROPY_THRESHOLD:
        return True, entropy  # 触发异常扫描行为
    
    return False, entropy

四、选型思维：为什么“自适应行为拦截”超越了传统防火墙？

在 AI 的辅助下，个人独立站长能够以极低的成本重塑服务器的安全边界：

评估维度	传统静态防火墙 (如 CSF / 经典规则)	自适应检测器 (Claude 辅助设计)
防护机制	依赖频次统计，如果 A 触发频次阈值则 B 动作	依赖行为特征，根据访问多端口的熵值、非标准请求判断
对抗慢速扫描	极其被动。黑客只需拉长扫描间隔就能轻松绕过	降维打击。只要扫描行为存在，即使每小时只测一次也会被识别
底层性能开销	频繁调用 iptables 重载会导致内核 CPU 瞬间飙升	极低。Python 内存滑动窗口分析 + nftables 动态 Set 原生秒级阻断
警报与可观测性	只有冷冰冰的日志，难以关联上下文	智能关联 IP 访问行为，提供富媒体渠道（如消息推送）即时报警