NetFlow 流量基线分析实战：异常流量溯源

作者

流量基线分析是排查带宽突增、主机外联异常、内部数据外传时性价比最高的一步：它不依赖全量抓包，只靠路由器或交换机导出的采样流记录，就能在几分钟内回答「谁在占带宽、流量去了哪里、和昨天相比异常多少」。本文用 softflowd + nfcapd + nfdump 搭一套轻量流采集与基线比对平台，给出可直接复制的采集器配置、交换机侧导出示例、基线计算脚本，以及把异常源 IP 自动同步到 nftables/ipset 封禁的联动命令，适用于只有一两台 Linux 服务器、却要覆盖整个出口流量的运维场景。

适用场景

  • 带宽被占满但抓包看不出目标：需要先定位是哪几个 IP、哪个端口在跑流量；
  • 怀疑内网主机被植入挖矿或远控木马，存在周期性外联心跳；
  • 需要给「网站被刷」「被当成代理跳板」提供可留存、可导出报表的流量证据；
  • 想在 DDoS 处置后复盘，确认清洗前后流量结构的变化。

前置条件

  • 一台 Linux 采集主机（Debian/Ubuntu 或 RHEL 系均可），建议独立部署，避免与业务机抢资源；
  • 可导出 sFlow 或 NetFlow v5/v9 的交换机、路由器或防火墙；若网关本身就是 Linux，可用 softflowd 直接采集本机网卡流量；
  • 采集主机开放 UDP 9995（NetFlow 接收）与 UDP 6343（sFlow 接收）端口，仅对设备管理网段放行；
  • 磁盘按流量规模预留，一般业务出口每天 200MB～1GB 流记录足够。

原理说明

采样与流缓存的取舍

sFlow 采用包采样：设备按固定比例（例如 1:1000）随机抽取报文连同接口计数器一起上报，优点是对设备 CPU 压力极小、能看到接口利用率，缺点是流量越小误差越大。NetFlow 则是维护一张流缓存表，五元组相同的报文聚合为一条流，流超时或结束时整条上报，精度更高但设备开销更大。两者都可以被 nfdump 工具链消费，所以采集侧统一用 nfcapd，查询侧统一用 nfdump。

基线为什么比阈值有效

固定阈值（如「单 IP 超过 100Mbps 就告警」）在业务有正常波动时必然误报。基线做法是把「同一时段的历史分布」作为参照：工作日 10 点的出口流量天然高于凌晨 3 点，把当前小时的 Top 源 IP 流量与过去 7 天同一小时的中位数比较，超过 3 倍才告警。这样既不误伤业务高峰，也能抓住异常外联。

操作步骤

步骤一：部署采集器

# Debian / Ubuntu
apt-get update && apt-get install -y nfdump softflowd

# RHEL / CentOS / Alma
dnf install -y nfdump softflowd

# 建立流记录存储目录，按天分目录存放
mkdir -p /var/cache/nfdump
chown nfdump:nfdump /var/cache/nfdump

用 systemd 托管接收进程，关键是 -t 300 表示每 5 分钟切一个流文件，便于后续按小时聚合：

cat > /etc/systemd/system/nfcapd.service <<'EOF'
[Unit]
Description=NetFlow/sFlow collector (nfcapd)
After=network-online.target

[Service]
Type=simple
ExecStart=/usr/bin/nfcapd -l /var/cache/nfdump -p 9995 -t 300 -S 1 -D -P /var/run/nfcapd.pid
Restart=always
User=nfdump

[Install]
WantedBy=multi-user.target
EOF

systemctl daemon-reload
systemctl enable --now nfcapd

若出口没有支持 sFlow 的设备，就在网关主机上直接跑 softflowd 采集本机网卡：

softflowd -i eth0 -n 127.0.0.1:9995 -v 9 -t 300 -d

参数含义：-i 指定监听网卡，-n 指定导出目标，-v 9 使用 NetFlow v9 格式，-t 300 为流超时时间（秒），-d 表示前台运行、交给 systemd 守护。

步骤二：交换机侧开启导出

以华为、H3C 等主流国产交换机的 sFlow 配置为例（不同型号命令略有差异，请以设备文档为准）：

# 全局视图：指定采样代理与采集器地址
sflow agent ip 10.0.0.1
sflow collector 1 ip 10.0.0.10 port 6343

# 接口视图：设置报文采样率与计数器采样率
interface GigabitEthernet1/0/1
 sflow sampling-rate 1000
 sflow counter-sampling-rate 1000

采样率 1:1000 是通用起点。千兆链路建议 1:1000～1:2000，万兆链路可放宽到 1:4000，既能还原流量结构，又不会把设备 CPU 打满。

步骤三：建立流量基线

# 查看最近一小时的 Top 源 IP（按字节排序）
nfdump -R /var/cache/nfdump -t 2026-10-08T09:00:00-2026-10-08T10:00:00 -s srcip/bytes -n 20

# 查看目的端口分布，快速判断是 Web、DNS 还是数据库外泄
nfdump -R /var/cache/nfdump -t 2026-10-08T09:00:00-2026-10-08T10:00:00 -s dstport/bytes -n 20

# 只看上行流量（源内网、目的外网），排查外传
nfdump -R /var/cache/nfdump -t 2026-10-08T09:00:00-2026-10-08T10:00:00 \
  -s srcip/bytes -n 20 -o 'fmt:%sa %byt %pkt %da %dp'

把上面的查询做成定时脚本，每小时把 Top 20 结果写入 CSV，连续跑满 7 天后即可得到基线数据。下面这段脚本完成「当前小时 vs 7 天同一小时中位数」的比值比对：

#!/usr/bin/env python3
import csv, statistics, subprocess, datetime, collections

CACHE = "/var/cache/nfdump"
TOP = 20

def hour_top(day, hour):
    win = "%sT%02d:00:00-%sT%02d:59:59" % (day, hour, day, hour)
    out = subprocess.run(
        ["nfdump", "-R", CACHE, "-t", win, "-s", "srcip/bytes", "-n", str(TOP),
         "-o", "fmt:%sa %byt"],
        capture_output=True, text=True, check=False).stdout
    res = {}
    for line in out.splitlines():
        parts = line.split()
        if len(parts) == 2 and parts[1].isdigit():
            res[parts[0]] = int(parts[1])
    return res

now = datetime.datetime.now()
today = now.strftime("%Y-%m-%d")
cur = hour_top(today, now.hour)
hist = collections.defaultdict(list)
for d in range(1, 8):
    day = (now - datetime.timedelta(days=d)).strftime("%Y-%m-%d")
    for ip, b in hour_top(day, now.hour).items():
        hist[ip].append(b)

for ip, b in sorted(cur.items(), key=lambda x: -x[1])[:5]:
    base = hist.get(ip, [])
    med = statistics.median(base) if base else 0
    ratio = (b / med) if med else float("inf")
    if ratio >= 3 or (med == 0 and b > 50 * 1024 * 1024):
        print("[异常] %s 当前 %.1fMB 基线 %.1fMB 倍数 %.1fx"
              % (ip, b / 1048576, med / 1048576, ratio))

步骤四：异常源自动封禁

# 导出 Top 源 IP
nfdump -R /var/cache/nfdump -s srcip/bytes -n 20 -o 'fmt:%sa' \
  | grep -Eo '^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+' > /tmp/bad.txt

# nftables 方案：建集合 + 引用集合的丢弃规则
nft add set inet filter blackhole '{ type ipv4_addr; timeout 1h; }'
while read -r ip; do nft add element inet filter blackhole "{ $ip }"; done < /tmp/bad.txt
nft add rule inet filter input ip saddr @blackhole drop

# iptables 方案
ipset create blackhole hash:ip timeout 3600 2>/dev/null
while read -r ip; do ipset add blackhole "$ip"; done < /tmp/bad.txt
iptables -I INPUT -m set --match-set blackhole src -j DROP

封禁前务必比对白名单：CDN 回源节点、监控探针、大客户专线出口都可能出现在 Top 列表里，误封会直接造成业务中断。建议先用 nfdump 查该 IP 的完整会话再决定。

配置验证

# 1. 确认流数据已到达（设备侧发包后应能看到持续计数）
tcpdump -i any -n udp port 9995 -c 20

# 2. 查看流文件元信息：起止时间、流数、包数、是否有序列丢失
nfdump -R /var/cache/nfdump -I

# 3. 确认目录按时段正常切分
ls -lh /var/cache/nfdump | tail -n 10

# 4. 查询总流量，与交换机接口计数对账（采样误差通常在 ±5% 内）
nfdump -R /var/cache/nfdump -t 2026-10-08T09:00:00-2026-10-08T10:00:00 -s bytes -n 1

若第 2 步的「Sequence failures」持续增长，说明 UDP 丢包，应降低采样率或改用 TCP 传输（部分设备支持）；若流数长期为 0，先确认防火墙是否放行了 9995/6343，再检查设备的导出源地址是否与采集主机同一路由。

常见问题

FAQ 1：交换机不支持 sFlow/NetFlow，能采集吗？

可以退而求其次：把采集主机放在出口网关位置，用 softflowd 采集本机转发流量；或者用「端口镜像 + 采集主机」的方式，在镜像口上跑 softflowd。代价是会增加一次流量拷贝，镜像口带宽必须足够。若只需定位内网横向流量，也可以在各业务主机上部署 softflowd，把流导出到同一采集器。

FAQ 2：采样 1:1000 会不会漏掉真正的攻击？

会。采样对大流量型攻击（DDoS、带宽盗刷、大文件外传）非常敏感，但对短连接高频扫描、低频 C2 心跳、单包漏洞利用这类「流量小、特征强」的行为几乎看不见。正确做法是分层部署：流采样负责「看宏观、定位异常主机」，包级 IDS（Suricata/Zeek）负责「看细节、判定攻击类型」，两者用同一套告警平台关联。

FAQ 3：流文件无限增长，磁盘被写满怎么办？

用 nfexpire 设置保留策略，例如只保留 30 天：nfexpire -l /var/cache/nfdump -t 30。更稳妥的方式是加一条 cron，每天清理超期目录：find /var/cache/nfdump -type d -mtime +30 -exec rm -rf {} +。同时用 -S 1 让文件按天分目录，便于按时间整体归档或删除，避免单目录下堆积上万个小文件。

总结

流采集方案的价值在于用极低的成本换来「看得见」：一台普通服务器、两个软件包、几条交换机命令，就能覆盖整个出口的流量视图。落地时抓住三点即可——采集侧用 nfcapd 按时段切文件保证可回溯，分析侧用「同小时中位数」做基线避免误报，处置侧把封禁动作收敛到集合类规则（nftables set / ipset）以便随时解封。后续若规模扩大，可把 nfdump 换成 ntopng 做可视化，或把流数据接入 SIEM 与现有 WAF 日志做关联分析。