简介这份资源是面向计算机、信息安全、数据科学与大数据、人工智能等专业学生及企业员工的机器学习实战项目聚焦恶意加密流量的识别与分类可用于课程设计、大作业、毕业设计或初期项目立项演示。压缩包共140个文件约31.9MB包含32个py源码与49个pyc编译文件、11个txt说明、10个log运行日志、7个csv数据集、4个joblib与2个pkl模型文件以及png图表、xml配置、cfg参数、md文档等覆盖数据预处理、特征提取、模型训练与评估的完整链路。内容预览中可见词向量模型、训练与测试数据划分等痕迹说明项目已具备可复现的实验流程。目前已有197人学习适合希望理解加密流量特征工程与机器学习建模思路的读者参考借鉴。1. 恶意加密流量识别从“看不见”到“抓得住”的那条分界线很多做安全的同行第一次接触恶意加密流量识别都会卡在同一个地方流量全是 TLS 密文端口、载荷、SNI 全被加密或伪装传统 DPI 直接失效。你手里只有一串包长、时间间隔、方向序列却要判断它到底是正常业务还是 C2 回连。这个标题里的“完整源码说明”本质就是一套把机器学习塞进这条缝隙里的工程方案——用流量的统计特征和时序行为替代已经看不见的明文内容。它解决的不是“解密”而是“不解密也能分类”。适合两类人一是做 IDS/NDR、态势感知的工程师需要给现有检测链路补一个加密流量分类模块二是做机器学习课程设计或安全方向毕设的学生想找一个既有真实数据、又能跑通端到端流程的题目。下面我按自己落地时的顺序把特征怎么提、模型怎么选、源码怎么跑、坑在哪一层层拆开。2. 特征工程先立住加密流量到底能提取什么2.1 为什么统计特征比载荷特征更靠谱加密流量识别最核心的认知转变是放弃载荷转向元数据。TLS 握手之后应用层内容不可见但包长序列、到达时间间隔、流持续时间、上下行字节比、包数量这些量仍然保留。恶意流量和正常流量在这些维度上往往有稳定差异——比如心跳型 C2 的包长高度规律、间隔接近固定值数据外传型流量则表现为上行字节远大于下行、包长分布集中在大包区间。常见做法是把一条流五元组相同、超时时间内切成前 N 个包提取两类特征一类是单流统计量如均值、方差、最大最小包长、总字节数另一类是时序序列如包长序列、方向序列1 上行 / -1 下行、时间间隔序列。前者喂给树模型后者喂给序列模型。我一般会先做统计特征因为它在小样本下更稳也更容易解释。提示不要一上来就上深度学习。加密流量公开数据集规模普遍不大树模型加统计特征往往能先拿到一个可用的 baseline再决定要不要上序列模型。2.2 用 Python 提取流级统计特征的最小实现下面这段代码演示从一条已解析的流记录里提取基础统计特征。输入假设是每个包一个字典包含length包长、direction1 上行 / -1 下行、timestamp秒。import numpy as np def extract_flow_features(packets): packets: list of dict, 每个元素形如 {length: 512, direction: 1, timestamp: 1690000000.12} 返回一条流的统计特征字典 lengths np.array([p[length] for p in packets], dtypefloat) dirs np.array([p[direction] for p in packets], dtypefloat) times np.array([p[timestamp] for p in packets], dtypefloat) # 时间间隔首包无前驱补 0 iats np.diff(times) iats np.insert(iats, 0, 0.0) feats { pkt_count: len(packets), byte_total: lengths.sum(), len_mean: lengths.mean(), len_std: lengths.std(), len_max: lengths.max(), len_min: lengths.min(), dir_up_ratio: (dirs 0).mean(), # 上行包占比 iat_mean: iats.mean(), iat_std: iats.std(), iat_max: iats.max(), duration: times[-1] - times[0] if len(times) 1 else 0.0, } # 上下行字节比避免除零 up_bytes lengths[dirs 0].sum() down_bytes lengths[dirs 0].sum() feats[up_down_byte_ratio] up_bytes / (down_bytes 1e-6) return feats逻辑说明先按包长、方向、时间三个维度分别取统计量再补一个上下行字节比。iat用np.diff求相邻时间差首包补 0 保证长度对齐。参数上len_std和iat_std是区分“规律心跳”和“突发业务”的关键up_down_byte_ratio对数据外传类恶意流量很敏感。实际工程里还会加前 10 个包的包长序列作为定长输入这里先不展开。2.3 特征归一化和缺失值处理的两个硬规矩第一训练集和测试集必须用同一套归一化参数。我见过太多人分别对训练集和测试集做fit_transform结果线下 AUC 0.98上线直接崩。正确做法是在训练集上fit一个StandardScaler保存下来测试和推理时只transform。第二缺失值不要无脑填 0。iat_std在单包流里是 NaN填 0 会让模型误以为“间隔极其稳定”。更稳的做法是加一个is_single_packet标志位缺失值填该特征在训练集的中位数。这两条看着简单但在我排查过的翻车案例里一半以上都跟它们有关。3. 模型选型与训练树模型、序列模型怎么分工3.1 先跑通 LightGBM/XGBoost 的 baseline统计特征维度通常在 20 到 80 之间样本量几千到几十万不等这个区间里梯度提升树几乎是最优解训练快、对缺失和异常值鲁棒、能输出特征重要性。我一般先用 LightGBM 跑一版看两个东西——AUC 和特征重要性排序。如果iat_std、len_std、up_down_byte_ratio排在前列说明特征方向对了如果模型主要靠pkt_count这种和流量类型强相关的量就要警惕数据集偏差。import lightgbm as lgb from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler from sklearn.metrics import roc_auc_score # X: 特征矩阵, y: 标签 0 正常 / 1 恶意 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, stratifyy, random_state42 ) scaler StandardScaler().fit(X_train) X_train_s scaler.transform(X_train) X_test_s scaler.transform(X_test) clf lgb.LGBMClassifier( n_estimators500, learning_rate0.05, num_leaves31, min_child_samples20, subsample0.8, colsample_bytree0.8, random_state42, ) clf.fit(X_train_s, y_train) pred clf.predict_proba(X_test_s)[:, 1] print(AUC:, roc_auc_score(y_test, pred))参数说明num_leaves31控制模型复杂度样本少时调到 15 以下防过拟合min_child_samples20是叶子最小样本数恶意样本少的时候可以降到 5 到 10subsample和colsample_bytree做行、列采样提升泛化。stratifyy保证训练测试集类别比例一致类别极不平衡时还要加class_weightbalanced。3.2 什么时候该上 1D-CNN 或 LSTM当统计特征已经榨干、AUC 卡在某个值上不去时再考虑序列模型。做法是把每条流的前 100 个包长不足补 0和方向序列拼成两通道输入喂给 1D-CNN 或双向 LSTM。CNN 擅长抓局部模式比如固定长度的握手包序列LSTM 擅长抓长程依赖比如心跳间隔的周期性。但要有心理准备序列模型对数据量要求高公开数据集往往撑不起来容易过拟合。我的经验是先用树模型定 baseline序列模型只作为 ensemble 的一个分支把它的输出概率和树模型概率加权平均通常比单独用序列模型稳。权重可以按验证集 AUC 调一般树模型占 0.6 到 0.7。3.3 类别不平衡和阈值选择恶意流量识别里恶意样本通常远少于正常样本比例可能到 1:100 甚至更极端。这时候 accuracy 完全没意义要看召回率Recall和误报率FPR。工程上更关心的是在 FPR 控制在 1% 的前提下Recall 能到多少。做法是训练时用scale_pos_weight或class_weight给正样本加权推理时不用默认 0.5 阈值而是在验证集上画 PR 曲线选一个满足 FPR 约束的阈值。这个阈值要写进配置文件不能硬编码在代码里因为不同网络环境的正常流量分布不同阈值需要现场微调。4. 源码工程怎么跑起来目录、依赖与推理链路4.1 典型工程目录和依赖清单拿到一份“完整源码说明”的压缩包先别急着pip install。我一般先看目录结构判断它是脚本堆还是工程化项目。一个能落地的恶意加密流量识别工程通常长这样目录/文件作用关注点data/原始 pcap 或已提取的 csv看字段是否含标签、时间戳features/特征提取脚本是否与训练时特征顺序一致models/训练好的模型文件格式、版本、是否带 scalertrain.py训练入口参数是否可配、随机种子infer.py推理入口是否支持单流/批量requirements.txt依赖版本是否锁死依赖上核心是scapy或dpkt解析 pcap、pandas/numpy特征处理、lightgbm/xgboost/torch模型、scikit-learn预处理和评估。如果requirements.txt里没锁版本建议自己把关键库版本固定下来否则换台机器跑结果可能对不上。4.2 从 pcap 到特征向量的完整命令链假设源码提供了extract_features.py典型跑法是# 1. 安装依赖 pip install -r requirements.txt # 2. 从 pcap 提取流级特征输出 csv python extract_features.py \ --pcap_dir ./data/pcap \ --out_csv ./data/flow_features.csv \ --max_pkts 100 \ --timeout 120 # 3. 训练模型保存模型和 scaler python train.py \ --csv ./data/flow_features.csv \ --model_out ./models/lgbm.txt \ --scaler_out ./models/scaler.pkl # 4. 对新的 pcap 做推理 python infer.py \ --pcap ./data/test.pcap \ --model ./models/lgbm.txt \ --scaler ./models/scaler.pkl \ --threshold 0.85参数说明--max_pkts 100表示每条流最多取前 100 个包超出的截断保证特征长度一致--timeout 120是流超时时间超过 120 秒没新包就认为流结束--threshold 0.85是推理阈值比训练默认 0.5 高目的是压误报。这几个参数是现场最常调的建议都做成命令行参数而不是写死。4.3 推理链路要和训练链路严格对齐这是最容易翻车的地方。训练时特征列的顺序是[pkt_count, byte_total, len_mean, ...]推理时如果pandas读进来的列顺序变了模型输出会完全错乱而且不报错。我的做法是在特征提取脚本里定义一个FEATURE_ORDER列表训练和推理都按这个列表重排并在推理前断言列名和顺序一致。另外scaler必须和模型一起保存、一起加载。只保存模型不保存 scaler推理时要么忘了归一化要么用了错误的均值方差结果就是线上指标和线下对不上。这个坑我踩过不止一次后来直接在infer.py里加了检查如果 scaler 文件不存在直接报错退出不给你“凑合跑”的机会。5. 避坑与排查那些让指标虚高、上线翻车的细节5.1 现象线下 AUC 0.99上线召回率不到 30%原因训练集和测试集来自同一批 pcap做了随机划分导致同一条流的不同片段同时出现在训练和测试里数据泄漏。解决按时间或按源 IP 划分数据集确保测试集的流在训练时完全没见过。如果数据量够最好按天划分模拟真实上线场景。5.2 现象模型把某些正常业务大量误报为恶意原因训练集里缺少这类正常业务的样本模型没见过只能往最近的恶意类靠。解决收集误报样本加入训练集的正常类重新训练同时检查特征里是否有和具体业务强相关的量比如某个固定端口对应的包长模式必要时在特征层面做泛化或剔除。5.3 现象同一份 pcap 跑两次结果不一样原因特征提取时用了字典或集合遍历顺序不确定或者模型训练没固定随机种子。解决所有涉及顺序的地方显式排序train.py里固定random_statenumpy、python的随机种子也一并固定。工程上要求可复现这不是洁癖是排查问题的前提。5.4 现象推理速度慢单条流要几百毫秒原因每次推理都重新加载模型和 scaler或者特征提取里做了全量 pcap 重解析。解决模型和 scaler 在服务启动时加载一次常驻内存特征提取改成流式边解析边出特征不要等整个 pcap 读完。如果用的是深度学习模型考虑转 ONNX 或做量化。5.5 现象换了网络环境误报率飙升原因不同网络的正常流量分布差异很大训练集覆盖不到。解决把阈值做成可配置并提供一个“正常流量基线学习”模式——在新环境先跑一段纯正常流量统计模型输出的分数分布据此调整阈值。这个模式在实际交付里非常实用能省掉大量现场调参时间。6. 进阶技巧用集成和在线更新把系统养起来单模型上线只是开始。我后来习惯做两件事一是多模型集成把 LightGBM、XGBoost 和一个轻量 1D-CNN 的输出概率加权平均权重按验证集 AUC 定通常能把 FPR 再压一截二是在线更新机制把线上推理中模型置信度低、但人工确认为恶意的样本回流到训练集定期增量训练。验证集成是否有效不能只看 AUC要看FPR1% 时的 Recall和FPR0.1% 时的 Recall这两个点更贴近实际运营。下面这个小脚本用来在验证集上扫阈值输出不同 FPR 下的 Recall我每次调完模型都会跑一遍import numpy as np from sklearn.metrics import roc_curve def recall_at_fpr(y_true, y_score, fpr_targets(0.01, 0.001)): fpr, tpr, thr roc_curve(y_true, y_score) result {} for target in fpr_targets: idx np.searchsorted(fpr, target) idx min(idx, len(tpr) - 1) result[fFPR{target}] { recall: tpr[idx], threshold: thr[idx], } return result # 用法 stats recall_at_fpr(y_test, pred) for k, v in stats.items(): print(k, Recall%.4f % v[recall], Thr%.4f % v[threshold])逻辑说明roc_curve返回的fpr是单调递增的用searchsorted找到第一个不小于目标 FPR 的位置取对应的tpr和阈值。参数上fpr_targets按你的运营容忍度设一般 1% 和 0.1% 两档就够。拿到阈值后写进推理配置别再用 0.5。最后说个我自己的习惯每次交付前一定拿一份完全没参与训练的真实 pcap跑一遍端到端从 pcap 到告警看整个链路有没有断点、阈值是否合理、日志是否够排查。模型指标再漂亮链路跑不通都是白搭。这套东西值不值得做取决于你有没有持续回流样本、持续调阈值的耐心——它不是一个训练完就完事的模型而是一个需要养的检测系统。希望帮到你。本文还有配套的精品资源点击获取