★ FEATURED ARTICLE 简介Nemea网络测量分析是一套面向网络流量分析与异常检测的模块化开源系统适合网络安全学习者、流量分析开发者及实验教学人员用于理解基于流的检测流程、模块接口与协同工作机制。压缩包共 46 个文件、约 200KB以 Shell 脚本、Markdown 文档、Makefile.am 构建文件及 Vagrantfile 虚拟环境配置为主并含 Dockerfile、配置样例和说明材料目录按检测器、基础模块、框架与用例组织便于搭建实验环境和梳理项目结构。系统围绕检测器、数据处理模块、NEMEA Framework 与 TRAP 流量分析平台展开覆盖 DNS 隧道、DoS、扫描等恶意流量检测场景也包含流数据导出、过滤、聚合、合并等预处理与后处理思路。已有 517 人学习可作为课程实验、项目原型、模块扩展或二次开发的参考材料也适合复现流式检测流程、验证告警规则并理解异常检测系统的工程组织方式。1. 凌晨两点探针给我的是 pcap不是结论凌晨两点接到电话说内网有台机器在扫端口值班的人只丢来一个几百兆的抓包文件。tshark 筛了一轮真正可疑的 SYN 就几百条剩下全是噪声等定位到源 IP 天已经亮了。那一刻我的想法很朴素网络流量分析不该把「从字节到结论」全压在人的手速上。Nemea 就是冲着这件事来的——一套模块化的流量分析与异常检测框架采集、解析、检测、输出各跑各的进程用统一通道串成流水线。它不卖现成规则卖的是把流量变成可编程记录流的能力。适合要自己写检测逻辑的安全开发也适合把自研探针接进链路的运维只想装个开箱即用的告警盒子它反而偏重。2. 架构拆解TRAP 通道、UniRec 记录和一条流水线2.1 为什么把模块拆成独立进程第一次看这套架构会不习惯一个检测器不是一个二进制里的一堆函数而是一排各自独立运行的可执行文件彼此之间靠 socket 说话。采集模块从网卡或 pcap 里取数据解析模块把字节转成结构化记录检测模块做判定输出模块落盘或转发。每个模块单独 fork、单独退出某个检测器写崩了不会把采集端一起带走这在长时间跑在线上的场景里是保命的。拆分带来两个直接好处。第一是语言自由核心模块用 C 写追求吞吐你自己补的检测逻辑完全可以用 Python 挂在同一条链路上只要它守 TRAP 协议。第二是可替换想换一种流聚合方式只换采集那一个二进制上下游不用动。代价是调试变麻烦——单步调试跨不过进程边界只能靠日志和接口状态定位这也是后面专门写排查一节的原因。模块之间的连接靠 TRAP流量分析处理通道完成。每个模块有若干输入接口和输出接口接口有名字、方向和数据类型一个模块的输出接口名与另一个模块的输入接口名对上连接就建立。这个「按名字配对」的机制很朴素但配置写错时非常难查名字差一个字符进程照常启动只是永远收不到数据。2.2 UniRec让所有模块说同一种记录格式TRAP 只负责搬字节搬什么由 UniRec 定义。UniRec 是一种自描述的二进制记录格式每条记录前挂一个模板模板里列出字段名、类型和偏移数据区只放值字段按偏移取。上游模块启动时把模板推给下游下游声明「我只要这几个字段」框架就只把需要的字段拷过去其余丢弃。这个设计对性能的意义比看上去大。流量场景里一条记录动辄几十个字段检测器往往只用五六个逐字段序列化成 JSON 再解析开销能吃满一个核。UniRec 的模板协商把这部分成本压到接近零代价是字段名必须严格对齐SRC_IP写成SRCIP下游就会一直等一个永远不会来的字段。常见字段大致是这几类写模块时基本绕不开字段类型含义SRC_IP/DST_IPIP 地址源、目的地址IPv4 与 IPv6 共用SRC_PORT/DST_PORTuint16传输层端口PROTOCOLuint8协议号TIME_FIRST/TIME_LAST时间戳流的首包、末包时间PACKETS/BYTESuint32/uint64包数与字节数时间戳字段是踩坑高发区后面单独说。2.3 一条流从网卡到告警的完整链路把上面几块拼起来一条流在系统里的路径是这样的采集模块抓包并按五元组聚合成流记录输出接口声明为 UniRec 类型检测模块从同名接口读记录判定命中后往自己的输出接口写一条告警记录通常带上原始五元组加一个告警类型字段输出模块把告警写文件、写系统日志或转成文本。整条链路的形状写在 XML 配置里一个最小的三段式配置长这样?xml version1.0 encodingUTF-8? nemea instance !-- 采集流聚合从网卡读包输出 UniRec 记录到名为 flow_out 的接口 -- module namecapture path/usr/bin/flow_meter params-i eth0 -o u:flow_out/params /module !-- 检测输入接口名必须与上游输出接口一致 -- module namescan_detector path/usr/bin/scan_detector params-i u:flow_out -o u:alert_out/params /module !-- 输出把告警记录写到文件 -- module namewriter path/usr/bin/logger params-i u:alert_out -o /var/log/nemea/alerts.log/params /module /instance /nemea-i与-o后面的u:前缀表示接口承载的是 UniRec 数据冒号后是接口名。capture的-o u:flow_out和scan_detector的-i u:flow_out拼在一起框架据此把两个进程接到同一对 socket 上。这三个模块的路径、接口名、参数名都要按你机器上实际安装的二进制改不要照抄路径。2.4 手动串联两个模块验证连接是否成立在写整份配置之前我习惯先用命令行把两个模块手动串起来跑一次确认接口能通。这一步能省掉大量「配置看着没问题但就是没数据」的排查时间。做法是先起下游接收端再起上游发送端# 终端 A先起接收端接口名 sink等待上游连入 /usr/bin/logger -i u:sink -o /tmp/test_alerts.log # 终端 B再起上游把数据推到同名接口 /usr/bin/flow_meter -i eth0 -o u:sink # 终端 C观察是否真有记录落下 tail -f /tmp/test_alerts.log顺序不能反。TRAP 建立连接的行为是「一端监听、一端连入」多数模块默认在启动时尝试连入下游下游没起来就会直接退出并打印连接失败。判断链路是否真的通了看两点接收端进程是否还活着以及/tmp/test_alerts.log有没有内容增长。文件一直是空的说明模板协商或接口名有问题不用去怀疑检测逻辑。3. 配置与运行XML 流水线加 nemea-cli 交互 shell3.1 XML 配置里到底写了什么整份配置的根是nemea下面挂一个或多个instance每个 instance 是一条独立的流水线。instance 里每个module至少要有name、path两个属性name是这个模块在配置里的唯一标识path指向可执行文件params里放传给该模块的命令行参数接口就在这里面用-i/-o声明。实际项目里配置会比上面的三段式长得多通常还会有一层结构采集端做初步过滤中间挂若干个并行的检测器最后一层做聚合去重。并行检测器的写法就是让多个模块的输入接口指向同一个上游输出接口名框架自动做分发。这种扇出结构是整个框架里最实用的部分——加一个新检测器只需要加一个module块不用改任何已有模块。有一类参数值得单独记住很多模块支持通过参数指定「活跃超时」和「非活跃超时」决定一条流多久没新包就算结束并向上游输出一次。这两个值直接决定检测的实时性设太长告警会延迟设太短长连接会被切碎成多条记录。我一般从默认值起手先在回放数据上跑通再按链路的实际流量特征调。3.2 nemea-cli 的交互式 shell 怎么用直接nemea -c config.xml能把流水线拉起来但线上更常用的是交互式 shell进到 shell 里再决定加载哪份配置、什么时候停。启动后是一个提示符输入help能看到当前版本支持的全部命令这点很重要——不同构建之间命令名有差异别硬记。# 进入交互式 shell nemea-cli # 加载并启动一份流水线配置 nemea start /etc/nemea/pipeline.xml # 看各模块是否都进入运行态 nemea status # 停掉整条流水线 nemea stopstart之后不要急着看告警先看status里每个模块的状态是不是都在跑。有模块显示已退出问题一定在启动阶段去翻它的 stderr 输出通常是路径不对、参数拼错或上游还没起来。这个 shell 的价值在于改配置不用重启服务框架stop掉当前 instance改完 XML 再start一次就行。命令名以help的实际输出为准不同版本增删过。3.3 日志落在哪出问题先看哪模块自己不打统一格式的日志标准错误直接继承给父进程。用 shell 启动时输出会集中到启动终端的 stderr或者被重定向进日志文件用系统服务方式启动时一般由服务管理器收走。我习惯在调试阶段把整条流水线的 stderr 重定向到一个固定文件出问题第一件事就是tail这个文件比在几个终端之间来回切快得多。真正难查的不是崩溃而是「进程都在数据没有」。这种情况先看接口确认上游输出接口名和下游输入接口名逐字符一致再看下游有没有声明必需字段。最后看 socket 文件是否残留——用文件系统 socket 时模块异常退出可能留下一个旧文件新进程连上去对端却早已不在表现就是只挂不报错。清掉残留文件再启动往往就通了。4. 写一个自己的检测模块从模板协商到告警输出4.1 先定检测目标和输入字段拿一个具体场景练手短时间内同一源地址对多个不同目的端口发起连接这是端口扫描最粗的特征。需要的输入字段不多——SRC_IP、DST_IP、DST_PORT、TIME_FIRST加上可选的PROTOCOL做过滤。判定逻辑用一个滑动窗口为每个源 IP 维护最近一个时间窗内出现过的目的端口集合集合大小超过阈值就产出一条告警。这个逻辑用 C 写性能最好用 Python 写改起来最快。两种都给出骨架按你的链路吞吐量选。4.2 C 模块骨架接收、判定、发送#include stdio.h #include string.h #include trap.h #include unirec/unirec.h /* 输入输出各一个接口按命令行 -i / -o 解析 */ int main(int argc, char **argv) { trap_ctx_t *ctx NULL; ur_template_t *in_tmplt NULL, *out_tmplt NULL; /* 初始化 TRAP 上下文1 个输入接口、1 个输出接口 */ trap_ctx_init(ctx, argc, argv, 1, 1, NULL); /* 只声明需要的字段框架会据此裁剪上游模板 */ trap_ctx_set_required_fields(ctx, SRC_IP,DST_IP,DST_PORT,TIME_FIRST,PROTOCOL); /* 取输入模板同时对端会收到我们的字段需求 */ const char *in_fmt trap_ctx_get_data_fmt(ctx, TRAP_IFC_INPUT, 0); in_tmplt ur_define_fields_and_update_template(in_fmt, NULL); /* 输出模板告警记录只带五元组 一个告警类型 */ out_tmplt ur_create_output_template(0, SRC_IP,DST_IP,DST_PORT,ALERT_TYPE, NULL); trap_ctx_set_data_fmt(ctx, TRAP_IFC_OUTPUT, 0, ur_template_string(out_tmplt)); uint16_t alert_type 1; /* 1 疑似端口扫描 */ while (1) { const void *rec NULL; uint16_t rec_size 0; ur_time_t ts; /* 阻塞接收一条记录对端关闭时返回非 0 */ if (trap_ctx_recv(ctx, TRAP_IFC_INPUT, 0, rec, rec_size) ! TRAP_E_OK) break; ip_addr_t src ur_get_ip_addr(in_tmplt, rec, SRC_IP); uint16_t dport ur_get_u16(in_tmplt, rec, DST_PORT); ts ur_get_time(in_tmplt, rec, TIME_FIRST); /* 这里调用你实现的窗口统计函数命中则发告警 */ if (scan_window_update(src, dport, ts)) { char buf[256]; ur_template_t *t out_tmplt; void *out ur_create_record(t, 0); ur_set_ip_addr(t, out, SRC_IP, src); ur_set_u16(t, out, DST_PORT, dport); ur_set_u16(t, out, ALERT_TYPE, alert_type); trap_ctx_send(ctx, TRAP_IFC_OUTPUT, 0, out, ur_rec_size(t, out)); free(out); } } trap_ctx_finalize(ctx); return 0; }这段骨架里几个参数值得逐条说明。trap_ctx_init的第四、五个参数分别是输入、输出接口数量必须和你在命令行里-i/-o的个数一致多一个少一个都会在启动阶段直接失败。trap_ctx_set_required_fields是模板协商的关键它决定了下游能看到哪些字段也决定了上游要为你保留多少数据字段写错的表现是拿到的值是 0 而不是报错非常隐蔽。ur_define_fields_and_update_template用的是上游给你的格式串不要自己硬编码字段顺序输出模板用ur_create_output_template自己定因为告警记录的字段集合和输入本来就不一样。ur_get_time拿到的ur_time_t是框架内部时间表示不要当成秒或毫秒直接用参与比较统一用它自带的比较/差值函数。4.3 Python 版本同一套接口改起来更快验证逻辑阶段用 Python 更划算改一个阈值不用重新编译import sys import pytrap trap pytrap.TrapCtx() # 参数命令行、输入接口数、输出接口数 trap.init(sys.argv, 1, 1) # 声明需要的字段等价于 C 版本的那一行 trap.set_required_fields(SRC_IP,DST_IP,DST_PORT,TIME_FIRST,PROTOCOL) # 输出模板自己定 trap.set_data_fmt(pytrap.UrDataType.UNIREC, SRC_IP,DST_IP,DST_PORT,ALERT_TYPE) windows {} # src_ip - {dport: last_seen} THRESHOLD 20 # 一个窗口内去重后的目的端口数 WINDOW 60 # 窗口长度秒 while True: try: data, _ trap.recv() except pytrap.FormatChanged: # 上游模板变化重新协商后继续 continue except pytrap.Terminated: break src data.SRC_IP dport data.DST_PORT now data.TIME_FIRST.to_sec() # 转成秒再比 bucket windows.setdefault(src, {}) # 清掉窗口外的端口记录 for port, seen in list(bucket.items()): if now - seen WINDOW: del bucket[port] bucket[dport] now if len(bucket) THRESHOLD: alert pytrap.UnirecTemplate() # 构造告警并发送字段顺序与上面 set_data_fmt 一致 trap.send({SRC_IP: src, DST_PORT: dport, ALERT_TYPE: 1})set_required_fields与 C 版对应是下游对上游的字段请求。recv抛出的FormatChanged必须处理上游模块重启或换了模板不重新协商就会一直拿到脏数据。时间戳统一转成秒再参与窗口计算to_sec()这类转换接口在不同绑定里名字略有差异写之前翻一下头文件。4.4 编译与接入流水线C 模块编译时链接 TRAP 和 UniRec 两个库头文件路径按你机器的安装位置调# 编译自定义检测模块 gcc -O2 -Wall -o scan_detector scan_detector.c \ -I/usr/include/nemea \ -ltrap -lunirec-I指向框架头文件目录-ltrap提供通道 API-lunirec提供记录读写。编译通过不代表能跑先手动串一次下游起一个 logger上游用回放工具灌数据看告警有没有按预期出来。确认之后再写进 XML 的module块把path指向刚编出来的二进制。一个容易忽略的点是告警字段的设计。只写「有告警」没意义至少带上源、目的、端口、告警类型和触发时间字段加得越多下游聚合和去重的空间越大。但字段也不能乱加每加一个都要确认下游的输出模块认得否则会在日志模块那一侧静默丢弃。5. 避坑与排查五条真实踩过的坑现象模块刚起来就退出日志里只有一行连接失败。原因TRAP 是「一端监听、一端连入」上游先起而下游还没就绪连入失败直接退出。整份配置同时启动时偶尔也会抢跑。 解决确认启动顺序是先下游后上游如果框架支持自动重连参数把它打开。手动调试时严格按「先起接收端」的顺序来。现象进程都在跑下游一条记录也收不到。原因接口名不一致或者下游声明的必需字段在上游模板里不存在。这两种情况都不报错只是永远没数据。 解决逐字符比对-o与-i后的接口名把下游的必需字段缩减到最小集合再试能收到数据就说明是字段名写错比如把DST_PORT写成了DPORT。现象时间窗口统计完全不准告警要么不来要么刷屏。原因把ur_time_t或 Python 侧的时间对象当成了秒数直接做减法单位对不上。有的模块输出的是流首包时间有的是末包。 解决统一用框架提供的时间转换函数转成秒再参与计算并在写逻辑前用回放数据打一条日志把原始值和转换值都打出来对照一次。现象长连接的检测结果被切成好几段同一目标反复告警。原因流聚合的活跃/非活跃超时设得太短一条持续的连接被反复判结束又重新开始。 解决按链路的实际连接时长分布调这两个参数先用回放数据跑一遍看同一五元组被拆成了几条记录再决定放大多少。现象重启后模块起不来提示地址已被占用。原因上一个进程异常退出文件系统 socket 或共享内存残留新进程连上去发现对端已死。 解决启动脚本里加一步清理残留文件再拉起线上用服务管理器托管让退出后的清理动作走统一入口别手工 kill。6. 二次开发的改造思路先扩字段再动逻辑真正落地时改造顺序比改造内容更重要。我的建议是先扩字段、再调阈值、最后才动检测逻辑因为前两步的验证成本最低出问题也最好定位。扩展字段的做法是找到上游模块的输出模板把新字段加进去然后让下游在必需字段里声明它——注意所有中间模块都要能透传这个字段少一个环节字段就断在那里。上线前用回放工具灌一批构造好的 pcap看新字段在下游是不是真的有值这一步不能省。第二类改造是替换检测算法。模块之间只靠 UniRec 记录耦合意味着你可以把某个检测器整个换掉只要输入输出字段对得上上下游完全无感。这给了我一个很实用的做法新旧两版检测器同时挂在同一条流水线上并行跑输出到不同的日志文件用同一批回放数据比对两边的告警差异确认新版不漏报再摘掉旧版。比起在测试环境里造流量这种做法更接近真实分布。性能上的改造集中在批量处理。逐条recv/send在低流量下没问题流量上来以后系统调用开销会很明显框架提供了批量接收发送的接口把循环改成一次拿一批、一次发一批通常能省下可观的 CPU。改之前先量一下单条模式的瓶颈在哪别一上来就上批量逻辑会复杂不少。调参这块也提一句。阈值类参数端口数、连接数、时间窗没有通用值跟你的网络规模强相关。我一般先在回放数据上跑出一组分布取一个明显偏离正常基线的分位点作为初值上线后观察一周再微调。直接抄别人配置里的数字翻车概率很高。接口升级时还有个后悔药给模块加一个兼容模式参数让它在遇到旧版字段集时降级运行而不是直接退出。线上滚动升级期间上下游版本不可能完全同步没有这个开关就得停机整体替换。从那以后我每接一个新检测模块都先拿回放数据走一遍模板协商对不对、字段单位对不对、告警字段下游认不认三件事全过了才放到真实链路上。这套流程看着笨但省下来的排查时间远超写它的时间。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
