
简介这是一套面向网络安全学习者的机器学习恶意加密流量监测平台源码采用Flask搭建可视化前端将流量抓包、特征提取与模型训练集成为完整项目适合毕业设计、课程实践或安全方向进阶学习。资源共90个文件压缩包约1.11MB其中包含14个py脚本负责协议解析、特征处理与模型调用、7个pcap流量样本、3个pkl训练模型以及8个HTML和8个CSS页面用于界面展示还有XML配置、CSV数据和SQLite数据库等层次清晰便于按需查阅。项目内附使用说明文档明确环境配置、训练流程与检测步骤并包含抓包协议分析器、特征提取模块、模型训练模块等核心模块可帮助从流量抓取到分类结果完整理解恶意流量检测流程。目前已有332人学习下载适合作为结合机器学习与网络安全的综合性参考项目。1. 加密流量不透明但元数据骗不了人加密流量早已占据网络流量的大部分传统防火墙基于端口和协议规则的检测在TLS加密面前几乎失效。攻击者的C2通信、数据窃取通道同样会加密但加密不代表无特征包长分布、到达间隔、握手证书、流持续时间这些元数据仍然透露出大量行为模式。基于机器学习的恶意加密流量监测平台正是抓住这一点用Python把抓包、特征提取、模型推理串成一条完整链路。平台内置流量切分脚本和特征生成模块训练好的模型以pkl形式保存最终通过Flask提供Web上传检测界面。对蓝队、安全研发和做毕设的学生来说这套代码既能当检测工具也能作为入侵检测研究的起点。2. 从pcap到特征向量特征工程与数据集构建2.1 协议解析器与流量切分原始pcap文件里是一帧帧二进制数据不能直接喂给机器学习模型。第一步要按五元组源IP、目的IP、源端口、目的端口、传输层协议把报文聚合成“流”。项目里抓包协议分析器.py和get_feature.py负责这个环节。切分粒度直接决定特征质量如果一条流里混入两个不同会话统计特征会被平均化导致模型分不清。常见做法是用scapy读取pcap以(src_ip, dst_ip, src_port, dst_port, proto)为key聚包同时按空闲时间阈值把长连接拆成多条流。代码骨架如下from scapy.all import rdpcap, TCP, UDP, IP from collections import defaultdict def split_flows(pcap_file, idle_timeout30): packets rdpcap(pcap_file) flows defaultdict(list) for pkt in packets: if IP in pkt and (TCP in pkt or UDP in pkt): proto TCP if TCP in pkt else UDP key (pkt[IP].src, pkt[IP].dst, pkt[TCP].sport if TCP in pkt else pkt[UDP].sport, pkt[TCP].dport if TCP in pkt else pkt[UDP].dport, proto) flows[key].append(pkt) # 按时间排序并用idle_timeout拆流 return flows逻辑说明rdpcap一次性把pcap读入内存适合单文件几十MB的场景如果pcap很大应改用PcapReader逐包迭代避免内存溢出。idle_timeout默认30秒C2心跳间隔通常在几秒到几十秒之间设太大会把多个会话拼成一条假流设太小又会把正常的长轮询拆得零碎。2.2 特征维度怎么选flows、packets、统计量所谓基于机器学习的流量监测核心是把每条流的原始报文序列压缩成一组数值特征。get_goodx.py和get_badx.py分别从良性流量和恶意流量目录里提取特征矩阵这些脚本的输入就是上一节切分好的流对象。通常每个流提取几十维特征下面这张表覆盖了最常用的一类特征类别具体特征攻击行为上的含义包长统计总包数、总字节数、平均包长、包长方差、25/75分位恶意C2心跳包通常小而均匀数据窃取则出现大包暴增时间统计到达间隔均值、方差、最小/最大间隔周期性心跳会产生低方差的间隔序列正常网页访问间隔混乱方向比例上行下行包数比、字节比下载型恶意软件上行远小于下行而被入侵的内网主机反过来TLS握手证书长度、加密套件数量、SNI是否存在恶意样本常自签名证书或缺失SNI字段标志位SYN/FIN/ACK比例、重复ACK次数扫描和爆破行为在标志位上与正常差异很大提取特征的代码逻辑如下import numpy as np def extract_features(flow_packets): lengths [len(pkt) for pkt in flow_packets] times [float(pkt.time) for pkt in flow_packets] intervals [times[i1] - times[i] for i in range(len(times)-1)] return { packet_count: len(flow_packets), total_bytes: sum(lengths), mean_len: float(np.mean(lengths)), std_len: float(np.std(lengths)), q25_len: float(np.percentile(lengths, 25)), q75_len: float(np.percentile(lengths, 75)), mean_interval: float(np.mean(intervals)) if intervals else 0.0, std_interval: float(np.std(intervals)) if intervals else 0.0, min_interval: float(np.min(intervals)) if intervals else 0.0, max_interval: float(np.max(intervals)) if intervals else 0.0, }分位数比均值更能刻画包长分布形状因为恶意流量经常表现为“头部固定大小尾部突发大包”单单均值会把两种模式混成一个数字。时间特征里min_interval对固定频率的扫描流量非常敏感而std_interval可以区分固定心跳和抖动网络延迟。2.3 构建良性与恶意样本集模型上限由数据质量决定。项目里的dataset目录、get_goodx.py和get_badx.py反映了训练集的构建过程良性流量来自正常网页浏览、视频播放等常规行为恶意流量来自已知恶意软件运行的抓包或公开恶意样本库。这里有两个工程点需要注意第一类别不平衡。恶意流量样本往往远少于良性训练时要在class_weight或sample_weight里给恶意样本更高的权重否则模型会把所有流量都判为良性。第二流量切分后要过滤过短的流比如包数少于3的流没有统计意义直接丢掉。我一般会在特征提取前加一个过滤函数def is_valid_flow(flow_packets, min_packets3): return len(flow_packets) min_packets这样能减少特征矩阵里的空值和极端噪声让后续模型训练更稳定。3. 机器学习模型训练与pkl序列化3.1 模型选型与训练流程流量特征维度不高通常在几十维左右且特征间有较强的非线性关系。这类场景优先考虑树模型随机森林和梯度提升。它们天然处理混合类型特征不需要归一化对缺失值也有一定的鲁棒性。项目最终的产物是model.pkl从文件结构和常见做法推断是用sklearn训练并持久化保存。训练脚本的核心代码如下from sklearn.ensemble import GradientBoostingClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report import joblib # X是特征矩阵y是标签0良性1恶意 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, stratifyy, random_state42 ) clf GradientBoostingClassifier( n_estimators300, max_depth5, learning_rate0.1, subsample0.8 ) clf.fit(X_train, y_train) print(classification_report(y_test, clf.predict(X_test))) joblib.dump(clf, model.pkl)stratifyy保证训练和测试集中正负样本比例一致这对不平衡数据非常重要。max_depth5配合subsample0.8是控制过拟合的关键树太深会记住训练集噪声子采样让每棵树看到不同数据提升泛化能力。如果数据量不到几千条建议把n_estimators降到200以内避免训练时间过长。3.2 评估指标与阈值调节加密流量检测场景里误报率比漏报率更让人头疼。一个正常的网页访问被判定成恶意比漏掉一次C2心跳更容易消耗处置精力。所以训练时不能只盯着准确率要重点看精确率precision和召回率recall的平衡。classification_report里会输出这两个指标但默认临界值是0.5。实际部署时往往要调高阈值只在高置信度时告警probs clf.predict_proba(X_test)[:, 1] preds (probs 0.7).astype(int) # 调高阈值到0.7阈值越高精确率越高、召回率越低。如果是蓝队做告警平台阈值设0.8甚至0.9把结果交给人工研判如果是全流量审计做线索发现阈值回到0.5。这个参数应当在Web平台的配置项里开放出来而不是写死在代码里。3.3 model.pkl的加载与预测接口训练好的pkl文件要封装成可复用接口供main.py和Flask控制器调用。关键问题是特征顺序训练时用的是pandas DataFrame各列顺序就是特征向量顺序预测时必须保持一致否则数值全错位。我一般会把特征名列表保存成json和pkl一起部署import joblib import json model joblib.load(model.pkl) feature_names json.load(open(feature_names.json)) def predict_flow(features_dict): arr [[features_dict[name] for name in feature_names]] prob model.predict_proba(arr)[0, 1] label malicious if prob 0.6 else benign return label, probmodel.pkl加载后是完整的sklearn估计器predict_proba返回形状(n_samples, n_classes)取[:, 1]得到恶意类别的概率。注意sklearn版本兼容问题用joblib在高版本sklearn下加载旧pkl常常报错建议用docker固定环境或者在训练机器上重新导出。4. Flask平台搭建从离线检测到实时Web平台4.1 Flask路由与上传机制Web平台位于web_platform目录controller/runserver.py是启动入口。Flask在这里做三件事接收用户上传的pcap文件、调用后台特征提取和模型预测、把结果渲染到页面上。关键路由示例from flask import Flask, request, render_template import os from feature_utils import extract_pcap_features from model_utils import predict_flow app Flask(__name__) UPLOAD_FOLDER static/uploads/ ALLOWED_EXTENSIONS {pcap, pcapng} def allowed_file(filename): return . in filename and filename.rsplit(., 1)[1].lower() in ALLOWED_EXTENSIONS app.route(/detect, methods[POST]) def detect(): file request.files[pcap] if file and allowed_file(file.filename): save_path os.path.join(UPLOAD_FOLDER, file.filename) file.save(save_path) features extract_pcap_features(save_path) label, prob predict_flow(features) return render_template(result.html, labellabel, probprob, featuresfeatures) return 文件格式不支持, 400 app.route(/) def index(): return render_template(index.html)UPLOAD_FOLDER必须先创建否则file.save会抛FileNotFoundError。文件名要做白名单校验防止上传非pcap格式或带有路径穿越字符的文件。这里还缺少一个安全细节对pcap解析后的流数量做上限判断比如一次上传解析超5000条流就提示用户拆分否则后台会把CPU打满。4.2 前端模板与结果可视化templates目录存放HTML页面static目录放图片和CSS。项目里有PieChart.png和PtSc1.png等截图说明结果页用到了图表展示。Flask渲染时把特征表传给模板前端用表格或ECharts画柱状图、饼图。要避免在控制器里写大段HTML模板继承和宏渲染是更清晰的方式。!-- result.html -- h2检测结果/h2 p标签{{ label }}置信度{{ prob | round(4) }}/p table classtable trth特征/thth值/th/tr {% for name, value in features.items() %} trtd{{ name }}/tdtd{{ value }}/td/tr {% endfor %} /table{{ prob | round(4) }}用模板过滤器格式化浮点数避免页面显示一长串小数。特征字典在控制器里已经构造好模板只需遍历。4.3 runserver.py启动与配置开发环境直接运行python runserver.py默认监听127.0.0.1:5000。如果想让同一局域网其他机器访问改成app.run(host0.0.0.0, port5000, debugFalse)debugFalse是硬性要求Flask调试模式自带Werkzeug调试器暴露在公网等于把服务器交出去。项目里还有个User_Info.sqlite3很可能是登录功能用的。这类系统最好放在内网或者前面加一层nginx反向代理做身份认证不要直接用Flask生产服务器扛高并发。5. 实网流量验证与模型部署避坑5.1 用真实pcap验证检测效果验证模型不能只靠训练集里的测试集。平台提供了pcap_test目录专门放没参与训练的独立pcap。我最常用的验证步骤是先抓10分钟正常网页浏览流量再用虚拟机运行一个已知恶意样本抓取C2会话两个pcap分别上传到平台观察输出标签和置信度。如果正常流量被误报优先检查是不是TLS特征有问题——自签名证书和缺少SNI的流量很容易被分为恶意。5.2 模型失效时的日志定位项目根目录的log.txt记录了平台运行日志。遇到“所有流量都判恶意”或“所有流量都判良性”时按下面的顺序排查# 查看日志尾部 tail -100 log.txt # 检查模型是否存在且可加载 python -c import joblib; mjoblib.load(model.pkl); print(m.n_classes_)最常见的问题是特征序号错位。训练时特征列表是[packet_count, total_bytes, ...]预测时dict的键顺序变了构造出的数组就和模型权重对不上。解决方法是保存特征名json并加断言assert sorted(features.keys()) sorted(feature_names), 特征名不一致另一个隐蔽的点是pcap里有VLAN标签。scapy解析时IP层会偏移如果用pkt[IP].src拿源地址带VLAN的帧会解析失败或取值错误。处理方式是在采集端让交换机去标签或者解析时判断是否有Dot1Q层if Dot1Q in pkt: ip pkt[Dot1Q].payload else: ip pkt[IP]模型部署后还要关注sklearn版本。joblib.dump保存的pkl与sklearn主版本强相关升级环境后常常报ModuleNotFoundError或AttributeError。最省事的做法是训练和推理都跑在固定版本的Docker容器里避免线上环境重装依赖导致模型加载失败。最后一个建议把流量切分、特征提取、模型预测三个步骤拆成独立函数命令行和Web入口共用这同一个函数这样在调试pcap文件时调命令行、在日常检测时调Web两边行为完全一致不会出现Web端和离线脚本结果不一致的怪问题。本文还有配套的精品资源点击获取