引言：原型阶段为什么要自己造轮子

做流量侧检测，团队通常有两条现成的路：一是上 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 内容。据此确定的核心特征集：

载荷熵是个便宜且有效的信号：加密或压缩载荷熵接近 8，明文协议（HTTP、DNS）明显更低，DNS 隧道类攻击常出现异常高的查询名熵。端口特征要谨慎，模型很容易学到“22 端口就是 SSH 爆破”这种伪相关，跨网络环境迁移时会崩，建议只保留几十个常见服务端口做类别编码。

特征计算还有几个原则。一是避免泄漏未来信息：所有特征只能使用流结束前已经观察到的包，不能在finish时回头补算“整个 pcap 的统计量”。二是保持训练和推理一致：离线提取和在线提取必须调用同一个Flow.finish，否则特征分布会悄悄漂移。三是数值特征要处理极端值：字节数、持续时间长尾很重，树模型对单调变换不敏感，但标准化后的线性模型和神经网络会受影响；原型阶段可以用log1p变换字节数和持续时间，或者直接用树模型。四是类别特征要有兜底：端口 one-hot 要设置handle_unknown="ignore"，否则线上遇到训练集没见过的端口会直接报错。五是特征版本管理：每次改特征集都要给模型打版本号，离线评估报告里记录特征列表，否则过两周你很难解释为什么线上分数和离线对不上。

类别不平衡与评估指标

恶意流量在真实链路里占比常在 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_TIMEOUT=30.0# 会话静默超时（秒）ACTIVE_TIMEOUT=600.0# 会话最大存活时间，防止长连接永不落盘defshannon_entropy(b:bytes)->float:ifnotb:return0.0freq=[0]*256forxinb:freq[x]+=1n=len(b)return-sum((c/n)*math.log2(c/n)forcinfreqifc)defflow_key(pkt):"""返回 (归一化五元组, 方向)。方向 +1 表示正向，-1 表示反向。"""ifIPnotinpkt:returnNoneip=pkt[IP]ifTCPinpkt:sport,dport=pkt[TCP].sport,pkt[TCP].dportelifUDPinpkt:sport,dport=pkt[UDP].sport,pkt[UDP].dportelse:sport=dport=0a,b=(ip.src,sport),(ip.dst,dport)ifa<=b: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.key=key self.ts_first=self.ts_last=ts self.fwd_pkts=self.bwd_pkts=0self.fwd_bytes=self.bwd_bytes=0self.lens=[]self.syn=self.fin=self.rst=self.ack=0self.payload_bytes=0self.ent_sum=0.0self.ent_n=0defupdate(self,pkt,direction,ts):self.ts_last=ts size=len(pkt)self.lens.append(size)ifdirection>0:self.fwd_pkts+=1self.fwd_bytes+=sizeelse:self.bwd_pkts+=1self.bwd_bytes+=sizeifTCPinpkt:f=int(pkt[TCP].flags)self.syn+=(f>>1)&1self.fin+=f&1self.rst+=(f>>2)&1self.ack+=(f>>4)&1payload=bytes(pkt.payload)ifpkt.payloadelseb""ifpayload:self.payload_bytes+=len(payload)self.ent_sum+=shannon_entropy(payload)self.ent_n+=1deffinish(self,intervals=None):"""把流对象转成一行特征。intervals 是该流相邻包时间差列表。"""dur=max(self.ts_last-self.ts_first,1e-6)total_pkts=self.fwd_pkts+self.bwd_pkts lens=self.lensor[0]mean_len=sum(lens)/len(lens)var_len=sum((x-mean_len)**2forxinlens)/len(lens)intervals=intervalsor[]ifintervals:mean_iat=sum(intervals)/len(intervals)var_iat=sum((x-mean_iat)**2forxinintervals)/len(intervals)else:mean_iat=var_iat=0.0port_a,port_b=self.key[0][1],self.key[1][1]# 启发式：取两端口中较小值作为服务端口；生产环境应记录首包目的端口server_port=min(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_bytes+1)/(self.bwd_bytes+1),"duration":dur,"mean_iat":mean_iat,"std_iat":math.sqrt(var_iat),"pps":total_pkts/dur,"bps":(self.fwd_bytes+self.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_filter=None):"""流式读取 pcap，返回特征行列表。pkt_filter 可传入 lambda 做额外过滤。"""flows={}intervals=defaultdict(list)rows=[]withPcapReader(pcap_path)asreader:forpktinreader:ifIPnotinpkt:continueifpkt_filterandnotpkt_filter(pkt):continuets=float(pkt.time)key_dir=flow_key(pkt)ifkey_dirisNone:continuekey,direction=key_dir flow=flows.get(key)# 老化：空闲超时或存活超时，先落盘再新建ifflowisnotNone:idle=ts-flow.ts_last alive=ts-flow.ts_firstifidle>IDLE_TIMEOUToralive>ACTIVE_TIMEOUT:rows.append(flow.finish(intervals.pop(key,[])))delflows[key]flow=NoneifflowisNone:flow=Flow(key,ts)flows[key]=flow prev_ts=tselse:prev_ts=flow.ts_last flow.update(pkt,direction,ts)ifts>prev_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]X=df.drop(columns=drop_cols)y=df["label"].astype(int).values# 按时间切分：前 80% 训练，后 20% 测试，避免同一攻击会话泄漏cut=int(len(df)*0.8)X_train,X_test=X.iloc[:cut],X.iloc[cut:]y_train,y_test=y[:cut],y[cut:]cat_cols=["server_port"]num_cols=[cforcinX.columnsifcnotincat_cols]pre=ColumnTransformer(transformers=[("num","passthrough",num_cols),("cat",OneHotEncoder(handle_unknown="ignore",min_frequency=20),cat_cols),])clf=Pipeline([("pre",pre),("model",HistGradientBoostingClassifier(max_iter=300,learning_rate=0.05,max_leaf_nodes=31,random_state=42,)),])clf.fit(X_train,y_train)proba=clf.predict_proba(X_test)[:,1]ap=average_precision_score(y_test,proba)# 按负样本 99.9 分位选阈值，对应约 0.1% 误报率neg_scores=proba[y_test==0]threshold=float(np.quantile(neg_scores,0.999))iflen(neg_scores)else0.5pred=(proba>=threshold).astype(int)tp=int(((pred==1)&(y_test==1)).sum())fp=int(((pred==1)&(y_test==0)).sum())fn=int(((pred==0)&(y_test==1)).sum())recall=tp/max(tp+fn,1)precision=tp/max(tp+fp,1)# 额外输出 PR 曲线上的工作点，便于按业务误报预算调整precisions,recalls,thresholds=precision_recall_curve(y_test,proba)print(f"PR-AUC={ap:.4f}, threshold={threshold:.4f}, "f"precision={precision:.4f}, recall={recall:.4f}, "f"tp={tp}, fp={fp}, fn={fn}")returnclf,threshold,(precisions,recalls,thresholds)

训练时不要只看一个阈值下的结果。建议把precision_recall_curve的输出保存下来，和业务方一起看“每天允许多少条告警”。例如测试集有 10 万条流、其中 50 条恶意，误报率 0.1% 意味着约 100 条误报，如果每天流量规模是测试集的 100 倍，那就是每天约 1 万条误报，显然不可接受。所以阈值要结合绝对告警量来定，而不是只看百分比。原型阶段可以做一个简单的告警量估算表：不同分位数阈值下，预计每天告警数、召回率、漏报数，让业务方选工作点。

另外，min_frequency=20是为了把长尾端口归并，减少 one-hot 维度。如果数据量很小，可以把这个值调低；如果端口分布非常集中，也可以手工维护一个常见端口白名单，其余统一映射成other。模型选择上，HistGradientBoosting对缺失值和长尾数值比较友好，训练速度快，适合原型；如果后续要解释特征贡献，可以换 LightGBM 并输出 feature importance，但不要在原型第一版就陷入调参。

第三步：在线推理与告警

离线模型训练好后，在线推理要做的是“同一套特征提取 + 模型加载 + 阈值判断”。最简原型可以用PcapReader回放 pcap 模拟在线，也可以用 Scapysniff接镜像口做小流量实时验证。下面给一个批量推理函数，实时场景可以把extract_features换成增量流表，但特征计算函数必须复用。

importjoblibimportpandasaspddefpredict_pcap(pcap_path,model_path,threshold,pkt_filter=None):rows=extract_features(pcap_path,pkt_filter=pkt_filter)ifnotrows:return[]df=pd.DataFrame(rows)drop_cols=[cforcin["flow_key","ts_first"]ifcindf.columns]X=df.drop(columns=drop_cols)model=joblib.load(model_path)scores=model.predict_proba(X)[:,1]alerts=[]forrow,scoreinzip(rows,scores):ifscore>=threshold: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和特征版本号，加载后先校验列名集合是否一致，不一致直接拒绝启动，而不是静默预测。

常见问题 FAQ

Q1：Scapy 太慢，能不能用于生产？
不建议直接用于线速生产采集。纯 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 和分片。本文代码只处理 IPv4，IPv6 需要单独取IPv6层；非首片没有端口，直接跳过或按 IP 聚合。第五，内存控制。Flow.lens和intervals会随流数量增长，长时间抓包要限制流表大小，超过上限时强制老化最旧的流。第六，特征漂移。上线后定期用新流量重算特征分布，和训练集做 PSI 或分位数对比，漂移明显时重新训练，不要等告警量爆炸才发现。第七，模型版本与回滚。模型文件、阈值、特征列表、训练数据时间范围要一起版本化，线上支持一键回滚到上一版，否则一次误更新可能带来大量误报。

如果原型跑通后要继续优化，优先级建议是：先补正常流量基线，再优化流老化策略，然后做特征消融找出真正有用的特征，最后才调模型超参。很多团队反着来，先上复杂模型，结果特征链路不稳定，离线指标再高也落不了地。

结语

Scapy 加机器学习的组合，价值不在于“性能”，而在于“速度”——把从 pcap 到告警的闭环压缩到几天内，让你能快速验证特征是否有效、阈值是否可接受、业务方是否愿意用。

更多硬核网安与AI工具包，请扫码获取完整源码！
原型阶段的目标不是做一个完美的检测引擎，而是用最小成本回答一个问题：这套特征和模型，能不能在你的网络里把恶意流量拎出来。如果能，再考虑用更工程化的采集链路替换 Scapy；如果不能，尽早换思路，比在错误方向上堆模型更划算。