引言原型阶段为什么要自己造轮子做流量侧检测团队通常有两条现成的路一是上 Suricata / Zeek 这类规则与解析引擎二是直接采购 NDR 产品。前者的问题是规则维护成本高、对加密流量和慢速扫描几乎无感误报率随规则库膨胀而上升后者的问题更直接——贵且特征不可控遇到自己业务的特殊协议只能等厂商排期。而机器学习路线在原型阶段的真实痛点是特征工程链路太长从 pcap 到可训练的特征矩阵中间要跨抓包、协议解析、会话重组、特征聚合四五个环节用 C 写太慢用现成工具又拿不到想要的自定义字段。Scapy 恰好卡在这个位置上纯 Python、协议层可任意下钻、改特征只需要改几行代码。它不适合做线速生产采集纯 Python 解析吞吐通常只有万级 pps但用来在两周内跑通“抓包 → 特征 → 模型 → 告警”的闭环几乎没有更快的选择。本文给出一条可复现的落地路径包含完整的特征提取、训练评估和在线推理代码以及几个我实际踩过的坑。需要先明确原型的边界我们验证的是“这套特征和模型能不能把恶意流量从背景流量里拎出来”而不是“明天就能替换现网 NDR”。因此实验环境建议用离线 pcap 加小流量镜像口数据量控制在几十万到几百万包单机 8C16G 足够。合规上也要提前确认pcap 中可能包含员工隐私和业务敏感信息采集前应脱敏或只在授权网段做特征提取阶段尽量不落盘 payload只保留统计量。另一个容易被忽略的点是基线数据。恶意样本可以找公开数据集或自己用虚拟机打靶场生成但正常流量必须来自你自己的网络。不同企业的正常流量差异极大办公网大量 HTTPS、视频会议生产网大量长连接、固定端口云上环境 DNS 查询密集。拿公开正常流量训练出来的模型换一个网络 PR-AUC 可能掉一半。所以原型第一周不要急着调模型先把正常流量的“长相”摸清楚。核心原理从数据包到特征向量检测粒度包级还是会话级包级特征单包长度、TCP 标志位、载荷熵实现简单但信息量太少单看一个包几乎无法区分正常 HTTPS 与 C2 心跳。会话级flow / bidirectional flow特征把同一五元组的双向流量聚合成一条记录能刻画“连接行为”——扫描器表现为大量短连接、单包、SYN 占比极高数据外传表现为上行字节数远超下行、持续时间长、包长分布集中。这也是 NetFlow / IPFIX 体系的思路工业界验证充分。因此原型阶段统一采用双向流作为样本单位五元组做归一化以(ip, port)字典序较小的一端作为源保证正反方向包落在同一条记录里。这里再补几个工程细节。第一流的老化不能只看空闲超时。TCP 长连接可能持续几小时如果只按空闲超时流表会一直膨胀所以要同时设置ACTIVE_TIMEOUT超过最大存活时间就强制落盘并新建流。第二TCP 的 FIN/RST 可以作为提前结束信号但不要一看到 FIN 就立刻删除流因为 FIN 之后还可能有反向 ACK 和残余数据建议收到 FIN 后再等一个短超时例如 5 秒。第三UDP 没有连接状态只能靠空闲超时DNS、QUIC 这类流量要把超时设短一些否则一个流会混进多个请求。第四IP 分片会影响端口解析非首片没有 TCP/UDP 头原型阶段可以直接跳过或只按 IP 聚合不要强行解析。包级和会话级并不是非此即彼。实际落地中可以用“会话级为主、包级为辅”会话级特征进主模型包级特征例如首包长度、TLS ClientHello 的 SNI 长度作为补充字段。但原型阶段建议先只做会话级减少变量等闭环跑通再叠加。特征设计原则特征要满足三个条件能从单个流内计算、对加密流量依然有效、不依赖具体 payload 内容。据此确定的核心特征集类别特征体量正向/反向包数、字节数、总包数、上下行字节比时序流持续时间、包间隔均值/标准差、每秒包数、每秒字节数包长分布均值、标准差、最大值、最小值协议行为SYN/FIN/RST/ACK 计数、平均载荷长度、载荷熵均值服务目的端口常见端口 one-hot其余归 other载荷熵是个便宜且有效的信号加密或压缩载荷熵接近 8明文协议HTTP、DNS明显更低DNS 隧道类攻击常出现异常高的查询名熵。端口特征要谨慎模型很容易学到“22 端口就是 SSH 爆破”这种伪相关跨网络环境迁移时会崩建议只保留几十个常见服务端口做类别编码。特征计算还有几个原则。一是避免泄漏未来信息所有特征只能使用流结束前已经观察到的包不能在finish时回头补算“整个 pcap 的统计量”。二是保持训练和推理一致离线提取和在线提取必须调用同一个Flow.finish否则特征分布会悄悄漂移。三是数值特征要处理极端值字节数、持续时间长尾很重树模型对单调变换不敏感但标准化后的线性模型和神经网络会受影响原型阶段可以用log1p变换字节数和持续时间或者直接用树模型。四是类别特征要有兜底端口 one-hot 要设置handle_unknownignore否则线上遇到训练集没见过的端口会直接报错。五是特征版本管理每次改特征集都要给模型打版本号离线评估报告里记录特征列表否则过两周你很难解释为什么线上分数和离线对不上。类别不平衡与评估指标恶意流量在真实链路里占比常在 0.1% 以下。此时 accuracy 毫无意义全判正常就有 99.9%必须用PR-AUC和固定误报率下的召回率。阈值也不能拍脑袋定 0.5而应按“负样本分位数”反推例如要求误报率不超过 0.1%就取负样本分数的 99.9 分位作为阈值。补充说明PR-AUC 对正样本比例敏感不同数据集的 PR-AUC 不能直接横比报告时要写清楚正负样本数。ROC-AUC 在不平衡场景下容易虚高可以看但不要作为主指标。评估集切分要按时间切不能随机打乱否则同一个攻击会话的包可能同时出现在训练和测试里指标会好得离谱。如果正样本太少阈值分位数不稳定可以用 PR 曲线上的工作点来选先定可接受的误报率再找对应召回率而不是反过来。上线后还要监控告警量和误报反馈阈值不是一次性的流量结构变化后要定期重算。实战案例从 pcap 到可部署模型第一步用 Scapy 做会话重组与特征提取核心是用PcapReader流式读取避免rdpcap一次性把几 GB 文件读进内存维护一个会话字典用空闲超时和存活上限做老化。importmathfromcollectionsimportdefaultdictfromscapy.allimportPcapReader,IP,TCP,UDP IDLE_TIMEOUT30.0# 会话静默超时秒ACTIVE_TIMEOUT600.0# 会话最大存活时间防止长连接永不落盘defshannon_entropy(b:bytes)-float:ifnotb:return0.0freq[0]*256forxinb:freq[x]1nlen(b)return-sum((c/n)*math.log2(c/n)forcinfreqifc)defflow_key(pkt):返回 (归一化五元组, 方向)。方向 1 表示正向-1 表示反向。ifIPnotinpkt:returnNoneippkt[IP]ifTCPinpkt:sport,dportpkt[TCP].sport,pkt[TCP].dportelifUDPinpkt:sport,dportpkt[UDP].sport,pkt[UDP].dportelse:sportdport0a,b(ip.src,sport),(ip.dst,dport)ifab:return(a,b,ip.proto),1return(b,a,ip.proto),-1classFlow:__slots__(key,ts_first,ts_last,fwd_pkts,bwd_pkts,fwd_bytes,bwd_bytes,lens,syn,fin,rst,ack,payload_bytes,ent_sum,ent_n)def__init__(self,key,ts):self.keykey self.ts_firstself.ts_lastts self.fwd_pktsself.bwd_pkts0self.fwd_bytesself.bwd_bytes0self.lens[]self.synself.finself.rstself.ack0self.payload_bytes0self.ent_sum0.0self.ent_n0defupdate(self,pkt,direction,ts):self.ts_lastts sizelen(pkt)self.lens.append(size)ifdirection0:self.fwd_pkts1self.fwd_bytessizeelse:self.bwd_pkts1self.bwd_bytessizeifTCPinpkt:fint(pkt[TCP].flags)self.syn(f1)1self.finf1self.rst(f2)1self.ack(f4)1payloadbytes(pkt.payload)ifpkt.payloadelsebifpayload:self.payload_byteslen(payload)self.ent_sumshannon_entropy(payload)self.ent_n1deffinish(self,intervalsNone):把流对象转成一行特征。intervals 是该流相邻包时间差列表。durmax(self.ts_last-self.ts_first,1e-6)total_pktsself.fwd_pktsself.bwd_pkts lensself.lensor[0]mean_lensum(lens)/len(lens)var_lensum((x-mean_len)**2forxinlens)/len(lens)intervalsintervalsor[]ifintervals:mean_iatsum(intervals)/len(intervals)var_iatsum((x-mean_iat)**2forxinintervals)/len(intervals)else:mean_iatvar_iat0.0port_a,port_bself.key[0][1],self.key[1][1]# 启发式取两端口中较小值作为服务端口生产环境应记录首包目的端口server_portmin(port_a,port_b)return{flow_key:f{self.key[0][0]}:{port_a}-{self.key[1][0]}:{port_b}/{self.key[2]},fwd_pkts:self.fwd_pkts,bwd_pkts:self.bwd_pkts,fwd_bytes:self.fwd_bytes,bwd_bytes:self.bwd_bytes,total_pkts:total_pkts,bytes_ratio:(self.fwd_bytes1)/(self.bwd_bytes1),duration:dur,mean_iat:mean_iat,std_iat:math.sqrt(var_iat),pps:total_pkts/dur,bps:(self.fwd_bytesself.bwd_bytes)/dur,mean_len:mean_len,std_len:math.sqrt(var_len),max_len:max(lens),min_len:min(lens),syn:self.syn,fin:self.fin,rst:self.rst,ack:self.ack,avg_payload_len:self.payload_bytes/max(total_pkts,1),mean_entropy:self.ent_sum/self.ent_nifself.ent_nelse0.0,server_port:server_port,ts_first:self.ts_first,}defextract_features(pcap_path,pkt_filterNone):流式读取 pcap返回特征行列表。pkt_filter 可传入 lambda 做额外过滤。flows{}intervalsdefaultdict(list)rows[]withPcapReader(pcap_path)asreader:forpktinreader:ifIPnotinpkt:continueifpkt_filterandnotpkt_filter(pkt):continuetsfloat(pkt.time)key_dirflow_key(pkt)ifkey_dirisNone:continuekey,directionkey_dir flowflows.get(key)# 老化空闲超时或存活超时先落盘再新建ifflowisnotNone:idlets-flow.ts_last alivets-flow.ts_firstifidleIDLE_TIMEOUToraliveACTIVE_TIMEOUT:rows.append(flow.finish(intervals.pop(key,[])))delflows[key]flowNoneifflowisNone:flowFlow(key,ts)flows[key]flow prev_tstselse:prev_tsflow.ts_last flow.update(pkt,direction,ts)iftsprev_ts:intervals[key].append(ts-prev_ts)# pcap 读完后把仍然存活的流全部落盘forkey,flowinflows.items():rows.append(flow.finish(intervals.pop(key,[])))returnrows这段代码有几个刻意简化但原型阶段可接受的地方。第一server_port取两端口中较小值对客户端随机端口访问常见服务端口通常成立但对高端口服务例如 8080、8443会误判生产环境应在流创建时记录首包目的端口并结合方向判断。第二Flow.lens保存了每个包长流特别大时会占内存如果担心内存可以改成在线维护均值、方差和最大最小值但原型阶段几十万包问题不大。第三bytes_ratio分母加 1 是为了避免除零同时保持单调性。第四ts_first只用于后续按时间切分和排查不进入训练特征。第五pkt_filter可以用来只保留某个网段或端口减少无关流量干扰。实际跑的时候建议先把特征落成 CSV/Parquet检查几件事总流数是否合理、duration是否有异常大的值、server_port的 top 分布是否符合预期、正负样本的mean_entropy分布是否有明显差异。如果发现某个特征在正负样本上几乎完全重合不要急着删先看它和其他特征的组合是否有效如果发现某个特征在正负样本上完全分离反而要警惕标签泄漏例如用目的端口直接判断了攻击类型。第二步训练与评估拿到特征表后训练本身不复杂关键是评估方式要贴近线上。下面用HistGradientBoostingClassifier做原型模型配合ColumnTransformer处理数值和类别特征。数据表假设包含label列1为恶意0为正常行顺序按 pcap 时间顺序排列因此可以直接按前 80% / 后 20% 做时间切分。importnumpyasnpimportpandasaspdfromsklearn.composeimportColumnTransformerfromsklearn.ensembleimportHistGradientBoostingClassifierfromsklearn.metricsimportaverage_precision_score,precision_recall_curvefromsklearn.pipelineimportPipelinefromsklearn.preprocessingimportOneHotEncoderdeftrain_model(df:pd.DataFrame):# df 需包含 label 列flow_key、ts_first 不参与训练drop_cols[cforcin[flow_key,ts_first,label]ifcindf.columns]Xdf.drop(columnsdrop_cols)ydf[label].astype(int).values# 按时间切分前 80% 训练后 20% 测试避免同一攻击会话泄漏cutint(len(df)*0.8)X_train,X_testX.iloc[:cut],X.iloc[cut:]y_train,y_testy[:cut],y[cut:]cat_cols[server_port]num_cols[cforcinX.columnsifcnotincat_cols]preColumnTransformer(transformers[(num,passthrough,num_cols),(cat,OneHotEncoder(handle_unknownignore,min_frequency20),cat_cols),])clfPipeline([(pre,pre),(model,HistGradientBoostingClassifier(max_iter300,learning_rate0.05,max_leaf_nodes31,random_state42,)),])clf.fit(X_train,y_train)probaclf.predict_proba(X_test)[:,1]apaverage_precision_score(y_test,proba)# 按负样本 99.9 分位选阈值对应约 0.1% 误报率neg_scoresproba[y_test0]thresholdfloat(np.quantile(neg_scores,0.999))iflen(neg_scores)else0.5pred(probathreshold).astype(int)tpint(((pred1)(y_test1)).sum())fpint(((pred1)(y_test0)).sum())fnint(((pred0)(y_test1)).sum())recalltp/max(tpfn,1)precisiontp/max(tpfp,1)# 额外输出 PR 曲线上的工作点便于按业务误报预算调整precisions,recalls,thresholdsprecision_recall_curve(y_test,proba)print(fPR-AUC{ap:.4f}, threshold{threshold:.4f}, fprecision{precision:.4f}, recall{recall:.4f}, ftp{tp}, fp{fp}, fn{fn})returnclf,threshold,(precisions,recalls,thresholds)训练时不要只看一个阈值下的结果。建议把precision_recall_curve的输出保存下来和业务方一起看“每天允许多少条告警”。例如测试集有 10 万条流、其中 50 条恶意误报率 0.1% 意味着约 100 条误报如果每天流量规模是测试集的 100 倍那就是每天约 1 万条误报显然不可接受。所以阈值要结合绝对告警量来定而不是只看百分比。原型阶段可以做一个简单的告警量估算表不同分位数阈值下预计每天告警数、召回率、漏报数让业务方选工作点。另外min_frequency20是为了把长尾端口归并减少 one-hot 维度。如果数据量很小可以把这个值调低如果端口分布非常集中也可以手工维护一个常见端口白名单其余统一映射成other。模型选择上HistGradientBoosting对缺失值和长尾数值比较友好训练速度快适合原型如果后续要解释特征贡献可以换 LightGBM 并输出 feature importance但不要在原型第一版就陷入调参。第三步在线推理与告警离线模型训练好后在线推理要做的是“同一套特征提取 模型加载 阈值判断”。最简原型可以用PcapReader回放 pcap 模拟在线也可以用 Scapysniff接镜像口做小流量实时验证。下面给一个批量推理函数实时场景可以把extract_features换成增量流表但特征计算函数必须复用。importjoblibimportpandasaspddefpredict_pcap(pcap_path,model_path,threshold,pkt_filterNone):rowsextract_features(pcap_path,pkt_filterpkt_filter)ifnotrows:return[]dfpd.DataFrame(rows)drop_cols[cforcin[flow_key,ts_first]ifcindf.columns]Xdf.drop(columnsdrop_cols)modeljoblib.load(model_path)scoresmodel.predict_proba(X)[:,1]alerts[]forrow,scoreinzip(rows,scores):ifscorethreshold:alerts.append({flow_key:row[flow_key],score:float(score),server_port:row[server_port],total_pkts:row[total_pkts],duration:row[duration],bytes_ratio:row[bytes_ratio],})returnalerts实时接入时建议把“抓包线程”和“推理线程”分开抓包线程只负责把包放进有界队列推理线程按时间窗口消费并维护流表避免 Scapy 解析阻塞抓包。队列要有丢弃策略宁可丢包也不要让内存无限增长。告警输出至少包含流五元组、分数、阈值、特征快照和时间戳方便后续复盘。不要把 payload 写进告警日志除非合规明确允许原型阶段保留统计量就够了。还有一个容易踩的坑训练时X的列顺序必须和推理时一致。Pipeline里ColumnTransformer按列名选择所以只要 DataFrame 列名一致就安全如果线上用 numpy 数组直接喂模型列顺序错一位分数就会乱。建议模型保存时同时保存feature_names和特征版本号加载后先校验列名集合是否一致不一致直接拒绝启动而不是静默预测。常见问题 FAQQ1Scapy 太慢能不能用于生产不建议直接用于线速生产采集。纯 Python 解析在普通单机上通常只有万级 pps遇到大包和复杂协议还会更低。生产环境可以用 Suricata/Zeek 做采集和流统计把自定义特征通过日志或 Kafka 接出来再用 Python 做模型推理或者用 DPDK/AF_XDP 做采集Scapy 只保留在离线特征验证环节。Q2加密流量还能做检测吗可以但特征要转向“元数据和行为”。流持续时间、包长分布、上下行字节比、包间隔、TLS 握手阶段可见字段SNI 长度、证书长度、JA3 等都不依赖明文 payload。本文的载荷熵对加密流量区分度有限因为加密载荷熵普遍接近 8此时更要依赖时序和体量特征。Q3正样本太少怎么办优先补充真实恶意样本而不是一味过采样。可以用靶场生成扫描、爆破、C2 心跳、DNS 隧道等样本但要注意生成环境和目标网络的一致性。训练时可以用class_weight或阈值调整但评估必须用原始分布的时间切分测试集不能在看不清真实误报的情况下调参。Q4为什么不能随机切分训练集和测试集同一个攻击会话可能产生多条流随机切分会让同一会话的流同时出现在训练和测试中模型相当于“见过答案”指标会虚高。按时间切分更接近线上“用过去预测未来”的场景虽然指标会低一些但更可信。Q5端口特征会不会导致模型学偏会。模型很容易学到“22 端口就是 SSH 爆破”“3389 就是 RDP 扫描”这类伪相关换一个网络或换一个端口就失效。建议只保留常见服务端口做类别编码其余归other并且不要用目的端口直接构造标签。评估时可以做一次“去掉端口特征”的消融实验看 PR-AUC 掉多少掉太多说明模型过度依赖端口。Q6流老化超时怎么定TCP 可以用空闲超时 最大存活时间UDP 用较短空闲超时。办公网 HTTP/HTTPS 流通常几秒到几十秒DNS 可能毫秒级生产网长连接可能几分钟到几小时。建议先统计自己网络里流持续时间的分布取 95 分位附近做参考再结合内存和告警实时性调整。超时太短会把一个流切成多个样本超时太长会混入多个请求。踩坑与优化建议第一pkt.time的类型。Scapy 返回的可能是EDecimal直接参与运算有时会慢建议尽早float()。第二TCP flags 解析。不同版本的 Scapy 对 flags 的表示有差异用int(pkt[TCP].flags)比直接比较字符串稳。第三payload 取值。bytes(pkt.payload)可能包含填充或上层协议头原型阶段可以接受但如果要做精确载荷熵应先按协议剥离头部。第四IPv6 和分片。本文代码只处理 IPv4IPv6 需要单独取IPv6层非首片没有端口直接跳过或按 IP 聚合。第五内存控制。Flow.lens和intervals会随流数量增长长时间抓包要限制流表大小超过上限时强制老化最旧的流。第六特征漂移。上线后定期用新流量重算特征分布和训练集做 PSI 或分位数对比漂移明显时重新训练不要等告警量爆炸才发现。第七模型版本与回滚。模型文件、阈值、特征列表、训练数据时间范围要一起版本化线上支持一键回滚到上一版否则一次误更新可能带来大量误报。如果原型跑通后要继续优化优先级建议是先补正常流量基线再优化流老化策略然后做特征消融找出真正有用的特征最后才调模型超参。很多团队反着来先上复杂模型结果特征链路不稳定离线指标再高也落不了地。结语Scapy 加机器学习的组合价值不在于“性能”而在于“速度”——把从 pcap 到告警的闭环压缩到几天内让你能快速验证特征是否有效、阈值是否可接受、业务方是否愿意用。更多硬核网安与AI工具包请扫码获取完整源码原型阶段的目标不是做一个完美的检测引擎而是用最小成本回答一个问题这套特征和模型能不能在你的网络里把恶意流量拎出来。如果能再考虑用更工程化的采集链路替换 Scapy如果不能尽早换思路比在错误方向上堆模型更划算。