尧图网页设计 
简介网络入侵检测系统项目资源包面向网络安全方向的学生、研究人员及企业安全运维人员聚焦基于深度学习的异常流量分析与恶意攻击识别。项目在NetIntrusionDetection-main框架下围绕UNSW_NB15、NSL_KDD、CICIDS_2017等公开数据集给出了1D CNN-BiLSTM模型的完整实现可检测DDoS、SQL注入、恶意软件传播等典型网络威胁。资源共45个文件、约11MB12个py脚本为训练与评估核心代码4个pth文件为预训练权重17个png为训练过程及结果可视化图表另含4个md文档及docx、txt说明文件目录划分清晰便于对照复现。已有52人学习下载适合希望快速上手深度学习入侵检测实践的开发者可直接基于代码与权重复现实验也可结合不同数据集进一步调优提升网络防护能力。1. 网络入侵检测系统当规则引擎被DDoS和SQL注入绕过之后某天凌晨告警平台突然被刷屏交换机镜像口跑满一个小时内几十个IP轮番发SYN包正常业务全部超时。又或者接入层没有任何异常但数据库慢查询突然变多回看HTTP日志才发现一个月前就有SQL注入流量悄悄穿透了WAF。传统网络入侵检测系统NIDS主要靠特征库匹配签名遇到变种攻击、加密流量和从未见过的DDoS手法往往是事后翻日志才知道。这个项目想解决的就是这件事用深度学习和机器学习算法实时监控网络流量把异常行为、DDoS、SQL注入和恶意软件传播在造成数据泄露之前挡住。适合需要自己搭建安全监测能力的运维、安全工程师也适合准备在企业内网做流量侧安全试点的团队。2. 从旁路流量到特征向量网络入侵检测系统的实时数据管道2.1 采集方案取舍镜像端口、流记录和抓包库怎么选构建一个基于深度学习和机器学习的网络入侵检测系统第一步不是选模型而是先想清楚流量从哪里来。最常见的部署位置是核心交换机镜像口在交换机上配置端口镜像把需要审计的进出流量复制一份到IDS探针。镜像流量通常是高速流量比如千兆口平均每个包1500字节每秒要处理8.3万包如果直接用Python逐包抓取再实时计算大概率会丢包。我一般会把采集方案分成三档采集方式数据粒度优点常见坑交换机镜像 libpcap/Scapy原始数据包特征最丰富能看载荷内容高PPS下丢包严重NetFlow/sFlow流记录采集开销小适合长时间统计没有载荷SQL注入检测做不了eBPF/DPDK抓包原始数据包能支撑10Gbps以上线速开发成本高需要绑核对于这个项目建议先用Scapy在旁路抓小流量把特征计算、模型训练、告警链路跑通再考虑换DPDK。用NetFlow做DDoS检测是够了但SQL注入和恶意软件传播必须看载荷所以纯流记录方案做不了这个项目里的一半任务。抓包机要接在镜像口上不要接在业务链路上否则一个网卡做转发、一个网卡做抓包真出问题时抓包动作本身就会拖垮业务。2.2 特征工程一个TCP会话要变出多少维特征机器学习模型不直接吃原始包它吃的是特征向量。做网络入侵检测系统时我把特征分成四类第一类流量统计特征。包括一个时间窗口内的包数、字节数、平均包长、最大包长、包长标准差、每秒字节速率。这类特征对DDoS特别敏感攻击流量会把包数或字节数拉到正常均值的几十倍。第二类TCP标志位特征。SYN、ACK、FIN、RST各自的出现次数SYN/FIN比例。SYN flood攻击会看到大量只有SYN没有FIN的短连接这个比例会非常高。第三类载荷内容特征。对每个TCP会话拼接前N个字节的载荷统计字符熵、可打印字符占比、URL长度、是否存在SQL关键字、是否包含大量编码字符。这里有一个血泪经验不要试图在原始包层面直接做字符串匹配去识别SQL注入攻击者会做注释符、大小写、URL编码、二次编码特征层面的统计比硬匹配耐用得多。第四类时序行为特征。把连续多个窗口的相同五元组特征叠起来形成时间序列。很多恶意软件传播和DDoS不是单窗口就能看出来的比如C2心跳流量、周期性外传数据都需要时间窗口做前后对比。特征列不是越多越好。项目里我限制核心特征在30维以内否则训练耗时增加线上推理延迟也会上去。前面说的四类特征合并之后做成一张宽表每行代表某个五元组在一个时间窗口内的行为摘要。2.3 可复用的特征提取脚本基于Scapy的流量快照下面这段代码是这个项目里最基本的一个流量特征提取器。我把它放在旁路抓包脚本里每收到一批包就更新对应的流状态凑满一个窗口就输出一条特征记录。# flow_extractor.py import time from collections import defaultdict, deque from scapy.all import IP, TCP, UDP class FlowFeatureExtractor: def __init__(self, window5, sample_interval10): # window统计窗口秒数 # sample_interval每凑够多少个包输出一次快照 self.window window self.sample_interval sample_interval self.flows defaultdict(lambda: deque(maxlen200)) self.records [] def packet_handler(self, pkt): if not pkt.haslayer(IP): return ip pkt[IP] if pkt.haslayer(TCP): key (ip.src, ip.dst, pkt[TCP].sport, pkt[TCP].dport, ip.proto) flags int(pkt[TCP].flags) elif pkt.haslayer(UDP): key (ip.src, ip.dst, pkt[UDP].sport, pkt[UDP].dport, ip.proto) flags 0 else: return pkt_len len(pkt) ts pkt.time self.flows[key].append((ts, pkt_len, flags)) if len(self.flows[key]) % self.sample_interval 0: self._snapshot(key) def _snapshot(self, key): q self.flows[key] if q[-1][0] - q[0][0] self.window: return # 只统计窗口内的包 cutoff q[-1][0] - self.window window_pkts [p for p in q if p[0] cutoff] if len(window_pkts) 2: return lengths [p[1] for p in window_pkts] flags [p[2] for p in window_pkts] bytes_sum sum(lengths) pkt_count len(window_pkts) syn_count sum(1 for f in flags if f 0x02) fin_count sum(1 for f in flags if f 0x01) duration window_pkts[-1][0] - window_pkts[0][0] if duration 0: duration 0.001 self.records.append({ src_ip: key[0], dst_ip: key[1], sport: key[2], dport: key[3], proto: key[4], pkt_count: pkt_count, bytes_sum: bytes_sum, avg_len: bytes_sum / pkt_count, max_len: max(lengths), std_len: int(np.std(lengths)) if len(lengths) 1 else 0, syn_ratio: syn_count / pkt_count, fin_ratio: fin_count / pkt_count, duration: duration, bps: bytes_sum / duration, pps: pkt_count / duration, })这段代码的逻辑是先按五元组把包对应到一条独立的“流”然后在内存里维护最近一段时间的包列表。凑满10个包或者窗口时间够长时计算当前窗口内的包数、字节数、包长标准差、SYN比例和每秒字节速率保存到records列表。这里窗口用的是滑动窗口每个新包都会让旧包过期避免长时间TCP连接占满内存。参数上window设为5秒适合DDoS这类突发流量如果要检测慢速SQL注入扫描建议把窗口拉大到60秒同时把sample_interval调小到5否则短会话可能没机会输出特征。注意这段代码没有做五元组层面的老化清理实际部署时还要额外加一个定时器定期把超过两分钟没有新包的流删掉不然高流量下字典会膨胀到几百万条把内存撑爆。2.4 特征存储与实时投递别在pandas里做在线预测特征计算完成后需要把特征向量送给模型做推理。很多新手会把特征攒到pandas里等到一定数量再批量预测。这个做法在离线训练没有问题但在实时网络入侵检测系统里会引入秒级甚至分钟级延迟DDoS早打完了。正确做法是让采集进程只负责产出特征然后把特征推送到Redis Stream或者Kafka再由独立的模型推理进程消费队列做预测。推送时用json序列化就可以字段顺序固定消费端按同样顺序转成numpy数组。这样采集和预测两个环节解耦预测进程重启不会丢网络数据采集进程遇到区块链的流量也可以单独提升队列容量不会拖垮模型。3. 用机器学习先跑通异常检测隔离森林和一分类SVM谁更省心3.1 为什么先上无监督告警样本永远是黑匣子在实际项目里最大的开局难题是没有标注数据。我们不是没有日志而是日志里到底哪个请求是真正的SQL注入、哪个IP属于恶意扫描器全靠人工追溯一个月的日志标出一千条高质量攻击样本已经是极限。若直接训练有监督模型正负样本比例悬殊模型很容易变成“永远预测正常”精度看似很高一个攻击都抓不到。所以这个项目的第一个模型我建议用无监督异常检测。它不需要标签直接对流量特征建模找出远离绝大多数正常行为的点。隔离森林和一分类SVM都是这个场景的常见选择。隔离森林的实现思路是用随机切割的方式把样本切到叶子节点异常样本通常更容易被单独切出来因此路径短一分类SVM则是把正常样本圈在一个超球体内球外都是异常。两者都不需要攻击样本参与训练适用于攻击方式每天都在变的现实环境。3.2 隔离森林训练与阈值调参下面这段代码读取上一章输出的特征表格训练一个隔离森林模型。# train_iforest.py import pandas as pd import numpy as np from sklearn.ensemble import IsolationForest import joblib df pd.read_csv(traffic_features.csv) feature_cols [ pkt_count, bytes_sum, avg_len, max_len, std_len, syn_ratio, fin_ratio, duration, bps, pps ] X df[feature_cols].fillna(0) model IsolationForest( n_estimators200, max_samples256, contamination0.05, random_state42, n_jobs-1 ) model.fit(X) # 预测结果1为正常-1为异常 pred model.predict(X) print(异常数量:, (pred -1).sum()) print(异常比例:, (pred -1).mean()) joblib.dump(model, iforest_nids.pkl)这段代码把特征表读进来选好特征列实例化隔离森林并训练最后保存模型。预测阶段模型会对每个样本给出1或-1。注意contamination并不是攻击流量的真实占比它只是模型在训练时用于估计异常阈值的比例。如果你不清楚生产环境里的异常比例先设为0.05之后拿到实际流量再调。n_estimators在200左右通常够用不必追求1000棵树模型训练速度和线上推理速度都会翻好几倍效果提升却很有限。max_samples设为256是让每棵树只看一部分样本增加随机性防止模型记死某一段时间内的流量形态。3.3 一分类SVM对比与模型保存一分类SVM在中小数据集上表现也不错但这个项目里我不太推荐。原因一它对特征尺度非常敏感必须在训练前用StandardScaler做标准化原因二训练复杂度随样本量上升很快几十万条流量特征要训练非常久不适合每三十分钟重训一次。它只适合做小流量场景的兜底模型或者和隔离森林做集成时投票用。如果你一定要尝试对照代码可以这样写from sklearn.svm import OneClassSVM from sklearn.preprocessing import StandardScaler scaler StandardScaler() X_scaled scaler.fit_transform(X) ocsvm OneClassSVM(nu0.05, kernelrbf, gammascale) ocsvm.fit(X_scaled) pred ocsvm.predict(X_scaled)nu参数和隔离森林的contamination一样指定异常比例的预期上限。gamma用“scale”会自动根据特征方差调核宽度少一个调参环节。但整体上我建议把隔离森林作为主力一分类SVM只在样本量低于1万时才考虑样本一多它训练就会变成一个漫长的等待过程。3.4 评估异常检测效果用精确率还是分组别无监督模型没有真实标签怎么评估效果这个项目里我采用“已知攻击回放验证”的办法。把真实抓包里确定含攻击的时间段单独切出来把攻击时段的特征打到正常模型上看模型能不能报出异常。由于我们不做全量标注这里评估指标以“漏报率”和“误报率”为主不会去算精确率召回率这样指望全量标签的指标。实操时我会把流量分成三类场景正常工作日高峰、夜间定时任务、攻击演练时段。隔离森林的默认score可以按分位数切一般取score小于第5分位数作为异常然后人工看异常时段对应的流是否确实可疑。这一步完全靠经验也是做机器学习模型的玄学所在。我的习惯是第一次调完阈值后保留厚度两周的原始pcap离线回放直到异常窗口里能看到真实的攻击行为才推到线上。4. 让深度学习识别恶意攻击CNN抓SQL注入、LSTM抓DDoS时序4.1 为什么不能只用统计特征把原始载荷交给CNN隔离森林能从统计特征里发现流量异常但它识别不了SQL注入。SQL注入攻击的载荷通常藏在HTTP请求参数里与正常请求在包长、包速率上几乎没有差别。比如一个正常的登录接口POST参数是usernameadminpassword123攻击者改成usernameadmin or 11包长多了十几个字节而已统计特征完全不会有明显波动。这个项目里我用CNN来做载荷层面的识别。做法是把每个TCP会话HTTP层的前128字节或256字节当作一个字符序列将每个字节编码成0到255的整数再用Embedding把整数映射成稠密向量经过两层一维卷积提取局部特征。CNN能捕捉到“or 11”“union select”“length(benchmark”这类跨字节组合的局部模式比正则匹配更抗干扰。4.2 用PyTorch搭一个SQL注入载荷识别CNN下面是一个最小的字符级CNN模型定义和训练骨架。# payload_cnn.py import torch import torch.nn as nn class PayloadCNN(nn.Module): def __init__(self, vocab_size256, embed_dim32, num_classes2): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim) self.conv1 nn.Conv1d(embed_dim, 64, kernel_size3, padding1) self.conv2 nn.Conv1d(64, 128, kernel_size3, padding1) self.pool nn.AdaptiveMaxPool1d(1) self.fc nn.Linear(128, num_classes) self.dropout nn.Dropout(0.3) def forward(self, x): # x: (batch, seq_len) x self.embedding(x) # (batch, seq_len, embed_dim) x x.permute(0, 2, 1) # (batch, embed_dim, seq_len) x torch.relu(self.conv1(x)) x torch.relu(self.conv2(x)) x self.pool(x).squeeze(-1) # (batch, 128) x self.dropout(x) return self.fc(x)训练时把载荷长度固定为128不够的补0超过的截断。Embedding层把每个字符变成一个32维向量卷积核大小为3意味着每次看三个连续字符的组合这对于识别一段连续的SQL关键字足够。两层卷积把通道数从32升到128让模型有足够容量记住常见注入变体。最后使用自适应最大池化把变长的卷积输出压缩成一个128维向量再接全连接层输出分类结果。这里有一个训练参数很关键类别权重。真实HTTP流量里SQL注入样本可能只占万分之一直接训练网络会全部预测为正常。用PyTorch时可以在损失函数里给正类更高的权重或者把正负样本的比例控制在1:5左右再做有放回抽样。4.3 用LSTM检测DDoS流量序列时间步长和步长的坑DDoS和SQL注入不一样DDoS不是单条请求可以定义的它是一段时间内的流量行为。瞬时突发可以通过统计特征里的PPS、BPS抓到但缓慢DDoS会把速率压到峰值的1.5倍拉长到几十分钟统计模型很难单独判异常。这时候需要让模型看到一段时间序列上的变化轨迹LSTM就是为这种时序建模准备的。我通常把每5秒统计一次的五元组特征序列作为LSTM的输入。每条训练样本由连续30个时间步组成每个时间步包含包数、字节数、SYN比例、平均包长、目的端口数量等。让LSTM学会流量从缓慢上升到持续压满的变化过程比只看一个窗口要可靠得多。# ddos_lstm.py import torch.nn as nn class DDoSLSTM(nn.Module): def __init__(self, input_size8, hidden_size64, num_layers2): super().__init__() self.lstm nn.LSTM( input_size, hidden_size, num_layers, batch_firstTrue ) self.fc nn.Linear(hidden_size, 1) def forward(self, x): # x: (batch, seq_len, input_size) out, _ self.lstm(x) # out: (batch, seq_len, hidden) out out[:, -1, :] # 取最后一个时间步的输出 return torch.sigmoid(self.fc(out)).squeeze(-1)这个模型输入形状是(batch, seq_len, input_size)。batch_firstTrue让数据维度更直观。hidden_size设为64两层LSTM能学到时间步之间的一些长程依赖但不要堆四层以上流量数据没有那么多复杂层次层数多了只增加训练时间推理时反而容易过拟合。最后一个时间步的输出被接上全连接层输出0到1之间的DDoS概率。训练时我把DDoS样本和不含攻击的正常序列按时间顺序切出来序列之间要留一部分重叠重叠比例在50%左右相当于数据增强能减少模型对固定起点位置的依赖。注意不要打乱时间顺序做随机抽样那样会让模型偷看未来信息训练指标很漂亮上线抓不到真实攻击。4.4 恶意软件传播检测DNS和外部连接是突破口恶意软件传播在这个项目里比较特殊。它不像DDoS那样有明显的带宽峰值也不像SQL注入那样藏在业务参数里而是大量短连接、频繁DNS查询、某些进程定期向固定域名发送心跳包。我的做法仍然是先走统计特征再走深度学习。统计特征层面抓“连接外部陌生IP的频次”“DNS查询失败率”“TXT记录占比”“非标准端口外连数”。把这些特征按5分钟窗口聚合喂给前面同一个LSTM模型。恶意软件刚进入内网时内网主机通常会开始扫描网段短时间内出现大量到非业务端口的新连接这是机器学习模型最容易抓到的信号。这一部分不需要单独做新模型只要把特征列替换掉复用前面DDoS LSTM网络结构就可以。5. 网络入侵检测系统落地避坑误报、样本不平衡和实时性翻车记录5.1 现象早上九点一到业务总被误报为DDoS项目上线第三天把流量特征接入隔离森林后每天早上九点开始出现大批量告警被标记为异常的IP都是内网办公网段而不是攻击源。查看特征发现九点是员工集中到岗时间大量电脑同时开机AD域同步、杀毒软件升级、邮件客户端唤醒造成短时间内PPS猛增窗口内的包数、字节数、SYN比例全部被推到边缘。隔离森林把它当成异常于是连续崩溃到十点。原因很明确正常流量的波动其实很大模型没有见过“办公时间开机潮”这种场景把正常的高峰行为打成了异常。解决办法是我在训练数据里加入了至少两周一整天的正常镜像流量包含工作日的开机潮、午休的流量低谷、夜间的定时备份让模型把这些形态也划入正常边界。同时给隔离森林加了一个时间特征例如“小时数”的一周周期编码让模型有机会区分“九点高峰”和“半夜的DDoS高峰”。5.2 现象SQL注入训练样本太少CNN完全学不到东西刚开始训练SQL注入CNN时我只有大约200条手工标注的注入流量其余100万条都是正常请求。模型训练后准确率达到了99.9%但再看召回率只有0.8%等于一个攻击也抓不到。原因是模型把所有样本都预测成了正常类负样本的梯度淹没了正样本的信号。解决这个问题的第一步是换损失函数权重把正常样本的loss乘以1注入样本的loss乘以50。第二步是去公开靶场流量里凑正样本比如从DVWA、SQLi-Labs里模拟注入请求再混合一部分真实环境的URL参数让模型见到不同风格的payload。第三步也是我后来一直坚持的做法在模型上线后把未被识别的真实注入流量人工确认后追加到训练集每天重训一次用两周时间把正样本量从200条扩到8000条以上。深度学习模型这类场景正样本数量过千才有基本的泛化能力。5.3 现象线上推理延迟太高流量一多就积压初版推理进程是一个包进来就跑一次模型一次特征提取就要0.5秒四套模型串联加起来超过2秒赶上镜像口流量高峰时积压越来越多告警比攻击晚到十分钟失去了实时性。原因是我把推理粒度做细了每一条流、每10个包就触发一次模型预测。后面改成了滑动窗口合并推理窗口内所有流共享同一个特征向量只有当一个窗口结束时才跑一次模型把模型调用频率降低了两个数量级。同时还加了一个粗糙的预筛层只有统计特征偏离基线超过2个标准差时才把载荷送给CNN正常流量根本不会走到深度学习模型这一步。这是网络入侵检测系统项目里性价比最高的一步优化延迟从秒级降到毫秒级误报没有明显上升。5.4 现象模型上线一周后效果断崖式下跌隔离森林和CNN模型都出现过这种情况第一周表现很好第二周开始告警数量骤减再回查发现攻击流量真实存在但模型把它们当成了正常流量。原因是流量分布漂移。比如公司上线了一个新视频会议系统大量UDP流量特征和之前的办公网特征完全不同模型擅自从“未见过的流量异常”变成了“新业务流量可接受”于是把新攻击也放进了正常区域。我的对策是每天做一次特征分布漂移检测重点监控PPS、BPS、平均包长这三个维度的PSI指标。当PSI超过0.2时就触发当天夜间定时重训模型用过去7天的流量特征作为训练集替代初始模型。这个做法让项目在有新业务上线时最多只有一天的空窗期而不是等到攻击发生了再补救。6. 把模型输出接到企业安全响应链路上滑动窗口在线预测模型训练完只是开始拿它做实时监控才是重点。我把在线预测做成一个由时间触发的滑动窗口任务每收完一个窗口的流量特征就交给模型推理然后把结果推送到安全事件平台或告警群。这里贴一段核心处理逻辑。# online_alert.py import queue import time def build_window_features(pending): # pending是窗口内的包列表提取成模型需要的特征向量 pkt_count len(pending) bytes_sum sum(p[1] for p in pending) syn_count sum(1 for p in pending if p[2] 0x02) return [[ pkt_count, bytes_sum, bytes_sum / pkt_count if pkt_count else 0, syn_count / pkt_count if pkt_count else 0 ]] def sliding_predict(packet_queue, window_sec5): model load_model(iforest_nids.pkl) pending [] window_start time.time() while True: pkt packet_queue.get() pending.append(pkt) if time.time() - window_start window_sec: feat build_window_features(pending) score model.predict(feat)[0] if score -1: alert(f窗口异常: {window_start}, 包数{len(pending)}) pending [] window_start time.time()这段代码用队列接收采集线程丢过来的包特征每5秒切一个时间窗口窗口内聚合出包数、字节数、平均包长和SYN比例再调用模型。这样做的好处是模型推理次数被固定下来不管上游流量多大一个窗口只预测一次。实际项目里我会把前端抓包进程、特征提取进程、模型推理进程拆成三个独立服务这里只是演示最核心的滑动窗口逻辑。我在每次上线新模型前还有一个习惯保留最近一周的原始pcap新模型先用离线回放打一遍需要的指标只有一个——一个小时内误报窗口数不超过三个同时回放攻击演练数据时至少能抓住八成。如果达不到这个标准就先不推向线上先把阈值调整好再试。这个方向真正麻烦的不是算法有多深而是流量里的噪声远比想象中多每一步都要用手工验证来兜底。希望这些落地经验对正在做网络入侵检测系统的你有所帮助。本文还有配套的精品资源点击获取 机器学习驱动的网络入侵检测系统：特征工程与实时监测实战
简介网络入侵检测系统项目资源包面向网络安全方向的学生、研究人员及企业安全运维人员聚焦基于深度学习的异常流量分析与恶意攻击识别。项目在NetIntrusionDetection-main框架下围绕UNSW_NB15、NSL_KDD、CICIDS_2017等公开数据集给出了1D CNN-BiLSTM模型的完整实现可检测DDoS、SQL注入、恶意软件传播等典型网络威胁。资源共45个文件、约11MB12个py脚本为训练与评估核心代码4个pth文件为预训练权重17个png为训练过程及结果可视化图表另含4个md文档及docx、txt说明文件目录划分清晰便于对照复现。已有52人学习下载适合希望快速上手深度学习入侵检测实践的开发者可直接基于代码与权重复现实验也可结合不同数据集进一步调优提升网络防护能力。1. 网络入侵检测系统当规则引擎被DDoS和SQL注入绕过之后某天凌晨告警平台突然被刷屏交换机镜像口跑满一个小时内几十个IP轮番发SYN包正常业务全部超时。又或者接入层没有任何异常但数据库慢查询突然变多回看HTTP日志才发现一个月前就有SQL注入流量悄悄穿透了WAF。传统网络入侵检测系统NIDS主要靠特征库匹配签名遇到变种攻击、加密流量和从未见过的DDoS手法往往是事后翻日志才知道。这个项目想解决的就是这件事用深度学习和机器学习算法实时监控网络流量把异常行为、DDoS、SQL注入和恶意软件传播在造成数据泄露之前挡住。适合需要自己搭建安全监测能力的运维、安全工程师也适合准备在企业内网做流量侧安全试点的团队。2. 从旁路流量到特征向量网络入侵检测系统的实时数据管道2.1 采集方案取舍镜像端口、流记录和抓包库怎么选构建一个基于深度学习和机器学习的网络入侵检测系统第一步不是选模型而是先想清楚流量从哪里来。最常见的部署位置是核心交换机镜像口在交换机上配置端口镜像把需要审计的进出流量复制一份到IDS探针。镜像流量通常是高速流量比如千兆口平均每个包1500字节每秒要处理8.3万包如果直接用Python逐包抓取再实时计算大概率会丢包。我一般会把采集方案分成三档采集方式数据粒度优点常见坑交换机镜像 libpcap/Scapy原始数据包特征最丰富能看载荷内容高PPS下丢包严重NetFlow/sFlow流记录采集开销小适合长时间统计没有载荷SQL注入检测做不了eBPF/DPDK抓包原始数据包能支撑10Gbps以上线速开发成本高需要绑核对于这个项目建议先用Scapy在旁路抓小流量把特征计算、模型训练、告警链路跑通再考虑换DPDK。用NetFlow做DDoS检测是够了但SQL注入和恶意软件传播必须看载荷所以纯流记录方案做不了这个项目里的一半任务。抓包机要接在镜像口上不要接在业务链路上否则一个网卡做转发、一个网卡做抓包真出问题时抓包动作本身就会拖垮业务。2.2 特征工程一个TCP会话要变出多少维特征机器学习模型不直接吃原始包它吃的是特征向量。做网络入侵检测系统时我把特征分成四类第一类流量统计特征。包括一个时间窗口内的包数、字节数、平均包长、最大包长、包长标准差、每秒字节速率。这类特征对DDoS特别敏感攻击流量会把包数或字节数拉到正常均值的几十倍。第二类TCP标志位特征。SYN、ACK、FIN、RST各自的出现次数SYN/FIN比例。SYN flood攻击会看到大量只有SYN没有FIN的短连接这个比例会非常高。第三类载荷内容特征。对每个TCP会话拼接前N个字节的载荷统计字符熵、可打印字符占比、URL长度、是否存在SQL关键字、是否包含大量编码字符。这里有一个血泪经验不要试图在原始包层面直接做字符串匹配去识别SQL注入攻击者会做注释符、大小写、URL编码、二次编码特征层面的统计比硬匹配耐用得多。第四类时序行为特征。把连续多个窗口的相同五元组特征叠起来形成时间序列。很多恶意软件传播和DDoS不是单窗口就能看出来的比如C2心跳流量、周期性外传数据都需要时间窗口做前后对比。特征列不是越多越好。项目里我限制核心特征在30维以内否则训练耗时增加线上推理延迟也会上去。前面说的四类特征合并之后做成一张宽表每行代表某个五元组在一个时间窗口内的行为摘要。2.3 可复用的特征提取脚本基于Scapy的流量快照下面这段代码是这个项目里最基本的一个流量特征提取器。我把它放在旁路抓包脚本里每收到一批包就更新对应的流状态凑满一个窗口就输出一条特征记录。# flow_extractor.py import time from collections import defaultdict, deque from scapy.all import IP, TCP, UDP class FlowFeatureExtractor: def __init__(self, window5, sample_interval10): # window统计窗口秒数 # sample_interval每凑够多少个包输出一次快照 self.window window self.sample_interval sample_interval self.flows defaultdict(lambda: deque(maxlen200)) self.records [] def packet_handler(self, pkt): if not pkt.haslayer(IP): return ip pkt[IP] if pkt.haslayer(TCP): key (ip.src, ip.dst, pkt[TCP].sport, pkt[TCP].dport, ip.proto) flags int(pkt[TCP].flags) elif pkt.haslayer(UDP): key (ip.src, ip.dst, pkt[UDP].sport, pkt[UDP].dport, ip.proto) flags 0 else: return pkt_len len(pkt) ts pkt.time self.flows[key].append((ts, pkt_len, flags)) if len(self.flows[key]) % self.sample_interval 0: self._snapshot(key) def _snapshot(self, key): q self.flows[key] if q[-1][0] - q[0][0] self.window: return # 只统计窗口内的包 cutoff q[-1][0] - self.window window_pkts [p for p in q if p[0] cutoff] if len(window_pkts) 2: return lengths [p[1] for p in window_pkts] flags [p[2] for p in window_pkts] bytes_sum sum(lengths) pkt_count len(window_pkts) syn_count sum(1 for f in flags if f 0x02) fin_count sum(1 for f in flags if f 0x01) duration window_pkts[-1][0] - window_pkts[0][0] if duration 0: duration 0.001 self.records.append({ src_ip: key[0], dst_ip: key[1], sport: key[2], dport: key[3], proto: key[4], pkt_count: pkt_count, bytes_sum: bytes_sum, avg_len: bytes_sum / pkt_count, max_len: max(lengths), std_len: int(np.std(lengths)) if len(lengths) 1 else 0, syn_ratio: syn_count / pkt_count, fin_ratio: fin_count / pkt_count, duration: duration, bps: bytes_sum / duration, pps: pkt_count / duration, })这段代码的逻辑是先按五元组把包对应到一条独立的“流”然后在内存里维护最近一段时间的包列表。凑满10个包或者窗口时间够长时计算当前窗口内的包数、字节数、包长标准差、SYN比例和每秒字节速率保存到records列表。这里窗口用的是滑动窗口每个新包都会让旧包过期避免长时间TCP连接占满内存。参数上window设为5秒适合DDoS这类突发流量如果要检测慢速SQL注入扫描建议把窗口拉大到60秒同时把sample_interval调小到5否则短会话可能没机会输出特征。注意这段代码没有做五元组层面的老化清理实际部署时还要额外加一个定时器定期把超过两分钟没有新包的流删掉不然高流量下字典会膨胀到几百万条把内存撑爆。2.4 特征存储与实时投递别在pandas里做在线预测特征计算完成后需要把特征向量送给模型做推理。很多新手会把特征攒到pandas里等到一定数量再批量预测。这个做法在离线训练没有问题但在实时网络入侵检测系统里会引入秒级甚至分钟级延迟DDoS早打完了。正确做法是让采集进程只负责产出特征然后把特征推送到Redis Stream或者Kafka再由独立的模型推理进程消费队列做预测。推送时用json序列化就可以字段顺序固定消费端按同样顺序转成numpy数组。这样采集和预测两个环节解耦预测进程重启不会丢网络数据采集进程遇到区块链的流量也可以单独提升队列容量不会拖垮模型。3. 用机器学习先跑通异常检测隔离森林和一分类SVM谁更省心3.1 为什么先上无监督告警样本永远是黑匣子在实际项目里最大的开局难题是没有标注数据。我们不是没有日志而是日志里到底哪个请求是真正的SQL注入、哪个IP属于恶意扫描器全靠人工追溯一个月的日志标出一千条高质量攻击样本已经是极限。若直接训练有监督模型正负样本比例悬殊模型很容易变成“永远预测正常”精度看似很高一个攻击都抓不到。所以这个项目的第一个模型我建议用无监督异常检测。它不需要标签直接对流量特征建模找出远离绝大多数正常行为的点。隔离森林和一分类SVM都是这个场景的常见选择。隔离森林的实现思路是用随机切割的方式把样本切到叶子节点异常样本通常更容易被单独切出来因此路径短一分类SVM则是把正常样本圈在一个超球体内球外都是异常。两者都不需要攻击样本参与训练适用于攻击方式每天都在变的现实环境。3.2 隔离森林训练与阈值调参下面这段代码读取上一章输出的特征表格训练一个隔离森林模型。# train_iforest.py import pandas as pd import numpy as np from sklearn.ensemble import IsolationForest import joblib df pd.read_csv(traffic_features.csv) feature_cols [ pkt_count, bytes_sum, avg_len, max_len, std_len, syn_ratio, fin_ratio, duration, bps, pps ] X df[feature_cols].fillna(0) model IsolationForest( n_estimators200, max_samples256, contamination0.05, random_state42, n_jobs-1 ) model.fit(X) # 预测结果1为正常-1为异常 pred model.predict(X) print(异常数量:, (pred -1).sum()) print(异常比例:, (pred -1).mean()) joblib.dump(model, iforest_nids.pkl)这段代码把特征表读进来选好特征列实例化隔离森林并训练最后保存模型。预测阶段模型会对每个样本给出1或-1。注意contamination并不是攻击流量的真实占比它只是模型在训练时用于估计异常阈值的比例。如果你不清楚生产环境里的异常比例先设为0.05之后拿到实际流量再调。n_estimators在200左右通常够用不必追求1000棵树模型训练速度和线上推理速度都会翻好几倍效果提升却很有限。max_samples设为256是让每棵树只看一部分样本增加随机性防止模型记死某一段时间内的流量形态。3.3 一分类SVM对比与模型保存一分类SVM在中小数据集上表现也不错但这个项目里我不太推荐。原因一它对特征尺度非常敏感必须在训练前用StandardScaler做标准化原因二训练复杂度随样本量上升很快几十万条流量特征要训练非常久不适合每三十分钟重训一次。它只适合做小流量场景的兜底模型或者和隔离森林做集成时投票用。如果你一定要尝试对照代码可以这样写from sklearn.svm import OneClassSVM from sklearn.preprocessing import StandardScaler scaler StandardScaler() X_scaled scaler.fit_transform(X) ocsvm OneClassSVM(nu0.05, kernelrbf, gammascale) ocsvm.fit(X_scaled) pred ocsvm.predict(X_scaled)nu参数和隔离森林的contamination一样指定异常比例的预期上限。gamma用“scale”会自动根据特征方差调核宽度少一个调参环节。但整体上我建议把隔离森林作为主力一分类SVM只在样本量低于1万时才考虑样本一多它训练就会变成一个漫长的等待过程。3.4 评估异常检测效果用精确率还是分组别无监督模型没有真实标签怎么评估效果这个项目里我采用“已知攻击回放验证”的办法。把真实抓包里确定含攻击的时间段单独切出来把攻击时段的特征打到正常模型上看模型能不能报出异常。由于我们不做全量标注这里评估指标以“漏报率”和“误报率”为主不会去算精确率召回率这样指望全量标签的指标。实操时我会把流量分成三类场景正常工作日高峰、夜间定时任务、攻击演练时段。隔离森林的默认score可以按分位数切一般取score小于第5分位数作为异常然后人工看异常时段对应的流是否确实可疑。这一步完全靠经验也是做机器学习模型的玄学所在。我的习惯是第一次调完阈值后保留厚度两周的原始pcap离线回放直到异常窗口里能看到真实的攻击行为才推到线上。4. 让深度学习识别恶意攻击CNN抓SQL注入、LSTM抓DDoS时序4.1 为什么不能只用统计特征把原始载荷交给CNN隔离森林能从统计特征里发现流量异常但它识别不了SQL注入。SQL注入攻击的载荷通常藏在HTTP请求参数里与正常请求在包长、包速率上几乎没有差别。比如一个正常的登录接口POST参数是usernameadminpassword123攻击者改成usernameadmin or 11包长多了十几个字节而已统计特征完全不会有明显波动。这个项目里我用CNN来做载荷层面的识别。做法是把每个TCP会话HTTP层的前128字节或256字节当作一个字符序列将每个字节编码成0到255的整数再用Embedding把整数映射成稠密向量经过两层一维卷积提取局部特征。CNN能捕捉到“or 11”“union select”“length(benchmark”这类跨字节组合的局部模式比正则匹配更抗干扰。4.2 用PyTorch搭一个SQL注入载荷识别CNN下面是一个最小的字符级CNN模型定义和训练骨架。# payload_cnn.py import torch import torch.nn as nn class PayloadCNN(nn.Module): def __init__(self, vocab_size256, embed_dim32, num_classes2): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim) self.conv1 nn.Conv1d(embed_dim, 64, kernel_size3, padding1) self.conv2 nn.Conv1d(64, 128, kernel_size3, padding1) self.pool nn.AdaptiveMaxPool1d(1) self.fc nn.Linear(128, num_classes) self.dropout nn.Dropout(0.3) def forward(self, x): # x: (batch, seq_len) x self.embedding(x) # (batch, seq_len, embed_dim) x x.permute(0, 2, 1) # (batch, embed_dim, seq_len) x torch.relu(self.conv1(x)) x torch.relu(self.conv2(x)) x self.pool(x).squeeze(-1) # (batch, 128) x self.dropout(x) return self.fc(x)训练时把载荷长度固定为128不够的补0超过的截断。Embedding层把每个字符变成一个32维向量卷积核大小为3意味着每次看三个连续字符的组合这对于识别一段连续的SQL关键字足够。两层卷积把通道数从32升到128让模型有足够容量记住常见注入变体。最后使用自适应最大池化把变长的卷积输出压缩成一个128维向量再接全连接层输出分类结果。这里有一个训练参数很关键类别权重。真实HTTP流量里SQL注入样本可能只占万分之一直接训练网络会全部预测为正常。用PyTorch时可以在损失函数里给正类更高的权重或者把正负样本的比例控制在1:5左右再做有放回抽样。4.3 用LSTM检测DDoS流量序列时间步长和步长的坑DDoS和SQL注入不一样DDoS不是单条请求可以定义的它是一段时间内的流量行为。瞬时突发可以通过统计特征里的PPS、BPS抓到但缓慢DDoS会把速率压到峰值的1.5倍拉长到几十分钟统计模型很难单独判异常。这时候需要让模型看到一段时间序列上的变化轨迹LSTM就是为这种时序建模准备的。我通常把每5秒统计一次的五元组特征序列作为LSTM的输入。每条训练样本由连续30个时间步组成每个时间步包含包数、字节数、SYN比例、平均包长、目的端口数量等。让LSTM学会流量从缓慢上升到持续压满的变化过程比只看一个窗口要可靠得多。# ddos_lstm.py import torch.nn as nn class DDoSLSTM(nn.Module): def __init__(self, input_size8, hidden_size64, num_layers2): super().__init__() self.lstm nn.LSTM( input_size, hidden_size, num_layers, batch_firstTrue ) self.fc nn.Linear(hidden_size, 1) def forward(self, x): # x: (batch, seq_len, input_size) out, _ self.lstm(x) # out: (batch, seq_len, hidden) out out[:, -1, :] # 取最后一个时间步的输出 return torch.sigmoid(self.fc(out)).squeeze(-1)这个模型输入形状是(batch, seq_len, input_size)。batch_firstTrue让数据维度更直观。hidden_size设为64两层LSTM能学到时间步之间的一些长程依赖但不要堆四层以上流量数据没有那么多复杂层次层数多了只增加训练时间推理时反而容易过拟合。最后一个时间步的输出被接上全连接层输出0到1之间的DDoS概率。训练时我把DDoS样本和不含攻击的正常序列按时间顺序切出来序列之间要留一部分重叠重叠比例在50%左右相当于数据增强能减少模型对固定起点位置的依赖。注意不要打乱时间顺序做随机抽样那样会让模型偷看未来信息训练指标很漂亮上线抓不到真实攻击。4.4 恶意软件传播检测DNS和外部连接是突破口恶意软件传播在这个项目里比较特殊。它不像DDoS那样有明显的带宽峰值也不像SQL注入那样藏在业务参数里而是大量短连接、频繁DNS查询、某些进程定期向固定域名发送心跳包。我的做法仍然是先走统计特征再走深度学习。统计特征层面抓“连接外部陌生IP的频次”“DNS查询失败率”“TXT记录占比”“非标准端口外连数”。把这些特征按5分钟窗口聚合喂给前面同一个LSTM模型。恶意软件刚进入内网时内网主机通常会开始扫描网段短时间内出现大量到非业务端口的新连接这是机器学习模型最容易抓到的信号。这一部分不需要单独做新模型只要把特征列替换掉复用前面DDoS LSTM网络结构就可以。5. 网络入侵检测系统落地避坑误报、样本不平衡和实时性翻车记录5.1 现象早上九点一到业务总被误报为DDoS项目上线第三天把流量特征接入隔离森林后每天早上九点开始出现大批量告警被标记为异常的IP都是内网办公网段而不是攻击源。查看特征发现九点是员工集中到岗时间大量电脑同时开机AD域同步、杀毒软件升级、邮件客户端唤醒造成短时间内PPS猛增窗口内的包数、字节数、SYN比例全部被推到边缘。隔离森林把它当成异常于是连续崩溃到十点。原因很明确正常流量的波动其实很大模型没有见过“办公时间开机潮”这种场景把正常的高峰行为打成了异常。解决办法是我在训练数据里加入了至少两周一整天的正常镜像流量包含工作日的开机潮、午休的流量低谷、夜间的定时备份让模型把这些形态也划入正常边界。同时给隔离森林加了一个时间特征例如“小时数”的一周周期编码让模型有机会区分“九点高峰”和“半夜的DDoS高峰”。5.2 现象SQL注入训练样本太少CNN完全学不到东西刚开始训练SQL注入CNN时我只有大约200条手工标注的注入流量其余100万条都是正常请求。模型训练后准确率达到了99.9%但再看召回率只有0.8%等于一个攻击也抓不到。原因是模型把所有样本都预测成了正常类负样本的梯度淹没了正样本的信号。解决这个问题的第一步是换损失函数权重把正常样本的loss乘以1注入样本的loss乘以50。第二步是去公开靶场流量里凑正样本比如从DVWA、SQLi-Labs里模拟注入请求再混合一部分真实环境的URL参数让模型见到不同风格的payload。第三步也是我后来一直坚持的做法在模型上线后把未被识别的真实注入流量人工确认后追加到训练集每天重训一次用两周时间把正样本量从200条扩到8000条以上。深度学习模型这类场景正样本数量过千才有基本的泛化能力。5.3 现象线上推理延迟太高流量一多就积压初版推理进程是一个包进来就跑一次模型一次特征提取就要0.5秒四套模型串联加起来超过2秒赶上镜像口流量高峰时积压越来越多告警比攻击晚到十分钟失去了实时性。原因是我把推理粒度做细了每一条流、每10个包就触发一次模型预测。后面改成了滑动窗口合并推理窗口内所有流共享同一个特征向量只有当一个窗口结束时才跑一次模型把模型调用频率降低了两个数量级。同时还加了一个粗糙的预筛层只有统计特征偏离基线超过2个标准差时才把载荷送给CNN正常流量根本不会走到深度学习模型这一步。这是网络入侵检测系统项目里性价比最高的一步优化延迟从秒级降到毫秒级误报没有明显上升。5.4 现象模型上线一周后效果断崖式下跌隔离森林和CNN模型都出现过这种情况第一周表现很好第二周开始告警数量骤减再回查发现攻击流量真实存在但模型把它们当成了正常流量。原因是流量分布漂移。比如公司上线了一个新视频会议系统大量UDP流量特征和之前的办公网特征完全不同模型擅自从“未见过的流量异常”变成了“新业务流量可接受”于是把新攻击也放进了正常区域。我的对策是每天做一次特征分布漂移检测重点监控PPS、BPS、平均包长这三个维度的PSI指标。当PSI超过0.2时就触发当天夜间定时重训模型用过去7天的流量特征作为训练集替代初始模型。这个做法让项目在有新业务上线时最多只有一天的空窗期而不是等到攻击发生了再补救。6. 把模型输出接到企业安全响应链路上滑动窗口在线预测模型训练完只是开始拿它做实时监控才是重点。我把在线预测做成一个由时间触发的滑动窗口任务每收完一个窗口的流量特征就交给模型推理然后把结果推送到安全事件平台或告警群。这里贴一段核心处理逻辑。# online_alert.py import queue import time def build_window_features(pending): # pending是窗口内的包列表提取成模型需要的特征向量 pkt_count len(pending) bytes_sum sum(p[1] for p in pending) syn_count sum(1 for p in pending if p[2] 0x02) return [[ pkt_count, bytes_sum, bytes_sum / pkt_count if pkt_count else 0, syn_count / pkt_count if pkt_count else 0 ]] def sliding_predict(packet_queue, window_sec5): model load_model(iforest_nids.pkl) pending [] window_start time.time() while True: pkt packet_queue.get() pending.append(pkt) if time.time() - window_start window_sec: feat build_window_features(pending) score model.predict(feat)[0] if score -1: alert(f窗口异常: {window_start}, 包数{len(pending)}) pending [] window_start time.time()这段代码用队列接收采集线程丢过来的包特征每5秒切一个时间窗口窗口内聚合出包数、字节数、平均包长和SYN比例再调用模型。这样做的好处是模型推理次数被固定下来不管上游流量多大一个窗口只预测一次。实际项目里我会把前端抓包进程、特征提取进程、模型推理进程拆成三个独立服务这里只是演示最核心的滑动窗口逻辑。我在每次上线新模型前还有一个习惯保留最近一周的原始pcap新模型先用离线回放打一遍需要的指标只有一个——一个小时内误报窗口数不超过三个同时回放攻击演练数据时至少能抓住八成。如果达不到这个标准就先不推向线上先把阈值调整好再试。这个方向真正麻烦的不是算法有多深而是流量里的噪声远比想象中多每一步都要用手工验证来兜底。希望这些落地经验对正在做网络入侵检测系统的你有所帮助。本文还有配套的精品资源点击获取