简介这是一套基于机器学习的入侵检测系统完整实现包含源代码与配套文档适合计算机相关专业的在校学生用于毕业设计、课程设计或项目初期立项演示也适合初学者进阶学习。系统采用Python编写代码涵盖数据预处理、SVM分类算法以及网络抓包模块包括数据读取、标准化处理等环节读者可从中理解从数据清洗、特征提取到模型训练与入侵检测的完整流程。资源共22个文件以Python脚本为核心辅以XML配置、Markdown说明文档及许可证文件整体压缩包仅18KB代码精简、目录结构清晰便于快速阅读与二次修改。目前已有457人学习下载项目说明显示源码已经运行测试答辩评审平均分达到96分完成度较高。通过阅读源码与配套文档不仅能快速复现基础的入侵检测流程还可在此基础上替换算法或接入真实网络数据适合作为进一步研究、课程作业参考或项目拓展的起点。1. 基于机器学习的入侵检测系统源码包拆解与复现路径这套「基于机器学习的入侵检测系统」源码本质是一条完整的批处理检测管线Sniffer.py负责抓流量DataProcessor.py把流量解析成特征向量SVM.py负责训练和判别最终输出正常 / 异常的判断结果。它对应的是计算机相关专业里典型的毕设课题也适合作为课程设计或项目初期立项演示。与大部分用深度学习做 IDS 的演示项目不同这套代码的核心分类器是 SVM在小样本、高维流特征的场景下反而比神经网络更可控训练快、调参透明答辩时也好解释。适合想快速拿到一套能跑通、能复现、能改的入侵检测系统的从业者或学生我下面把建模思路、运行流程、参数设置和踩坑点逐一拆开。2. 为什么选 SVM 加流特征入侵检测建模思路与特征工程2.1 入侵检测先分清检测对象网络侧与数据侧入侵检测系统Intrusion Detection System, IDS按数据源大致分两类。一类是主机侧HIDS看系统日志、进程行为、文件完整性另一类是网络侧NIDS抓网络流量分析。这套源码走的是网络侧因为网络侧的数据获取路径最短而且不依赖目标主机的权限只需在网络接入层做镜像或旁路抓包。网络侧 IDS 的检测对象是「流」flow而非单个包。单个包能提供的信息非常有限只有 IP、端口、TCP 标志位和少量载荷但把一段时间内属于同一个 TCP 连接的数据包聚合起来就能计算会话时长、包长分布、包到达间隔等统计特征这些特征才能暴露端口扫描、DDoS、慢速爆破等行为模式。WebPackageSniffer这个目录名说明这套代码里有针对 Web 流量的抓取组件它配合Sniffer.py的工作方式是先抓原始报文再做五元组源 IP、目的 IP、源端口、目的端口、协议聚合按会话切分最后交给DataProcessor.py做特征统计。这个流程与 Suricata 等传统 IDS 的「抓包→解码→规则匹配」思路不一样它跳过了规则匹配直接让机器学习算法从特征里学规律。2.2 SVM 在入侵检测里凭什么合适常见的 IDS 分类器有决策树、随机森林、朴素贝叶斯、SVM以及后起的深度学习。这套源码选 SVM有几个现实原因。第一流特征通常是高维的。DataProcessor.py里对每个会话提取的特征一般有三十到几十个维度而样本量受限于标注数据集的规模往往只有几万条。高维小样本正是 SVM 的舒适区它通过核函数把原始特征映射到高维空间在映射后的空间里找最大间隔超平面不容易像深度网络那样在小数据上过拟合。第二SVM 的决策边界只由支持向量决定对冗余特征相对不敏感。网络流量特征里经常存在强相关变量比如包长均值与包长中位数决策树类模型也可以处理但 SVM 配合标准化后收敛更稳定。第三毕设场景需要「能讲清楚」。SVM 的数学原理最大间隔、核函数、支持向量是机器学习课程的核心内容答辩时解释成本低。不过要留意一个点SVM 的原生输出是样本到超平面的距离而不是概率不能直接把predict结果当成置信度来展示。下面是典型的训练流程代码骨架对应SVM.py里的核心逻辑from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler from sklearn.svm import SVC from sklearn.metrics import classification_report, confusion_matrix # 加载经过 DataProcessor 处理后的特征表 # 假设特征列是 feats标签列是 label0 正常1 异常 X df[feats].values y df[label].values # 先划分再标准化避免测试集信息泄漏到训练过程 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.3, random_state42, stratifyy ) scaler StandardScaler() X_train scaler.fit_transform(X_train) X_test scaler.transform(X_test) # RBF 核C 控制误分类惩罚gamma 控制单个样本的影响半径 clf SVC(kernelrbf, C10.0, gammascale, class_weightbalanced) clf.fit(X_train, y_train) y_pred clf.predict(X_test) print(classification_report(y_test, y_pred))逻辑说明先划分训练集和测试集是为了防止标准化时测试集的统计量参与计算这是初学者最容易犯的错误。stratifyy保证正常和异常样本在训练集与测试集里的比例一致避免某个分类在测试集里一个样本都没有。class_weightbalanced在样本不平衡时代价敏感地调整误分类权重后面会展开说。参数说明C10.0的语义是误分类惩罚强度C 越大边界越严格但也更容易过拟合gammascale表示根据特征数量自动计算 gamma 初值比手动填一个固定数值更稳健我通常从scale起步再微调。2.3 特征工程重点协议头、报文分布与时间窗统计DataProcessor.py在整条管线里承担的是特征提取这是影响检测效果最关键的一环。模型再强喂进去的特征区分度不够结果也是空转。常见做法是从原始 pcap 里按会话聚合计算三类特征。第一类是报文长度分布特征包括每个方向上包长的均值、方差、最大最小值、中位数。端口扫描和 DDoS 流量的包长往往集中在很小的范围而正常 Web 流量的包长波动很大。第二类是时间相关特征如会话持续时间、到达间隔的均值与标准差。暴力破解会表现为短时间内大量短连接持续时间极短。第三类是协议行为特征比如 TCP 标志位的统计、SYN 包占比、ACK 包占比。SYN flood 的 SYN 包比例会异常高。实际操作时DataProcessor.py一般先从抓包文件里提取每条流的五元组再按流 ID 聚合。伪代码如下import pandas as pd def extract_session_features(flows): 对切分好的会话列表计算统计特征 records [] for flow_id, packets in flows.items(): lengths [p[length] for p in packets] timestamps [p[timestamp] for p in packets] intervals [timestamps[i1] - timestamps[i] for i in range(len(timestamps)-1)] records.append({ flow_id: flow_id, pkts: len(packets), bytes_total: sum(lengths), len_mean: sum(lengths) / len(packets), len_std: (sum((x - sum(lengths)/len(packets))**2 for x in lengths) / len(packets)) ** 0.5, duration: timestamps[-1] - timestamps[0], iat_mean: sum(intervals) / len(intervals) if intervals else 0, }) return pd.DataFrame(records)逻辑说明interval计算的是包与包之间的到达间隔iat_mean反映连接建立的频率特征len_std在扫描攻击时会明显偏小。这些特征都不依赖 IP 和端口因此泛化性比直接输入 IP 好得多。这里有一个关键边界不要把源 IP、目的 IP、源端口、目的端口直接作为特征喂给 SVM否则模型学到的是「某个 IP 是坏的」而非「某种行为是坏的」换一个 IP 就失效。我在实际项目中见过有人把五元组作为特征直接训练训练集准确率接近 100%测试集一到真实网络就崩原因就在标签泄漏。3. 把代码跑起来环境准备、训练流程与关键参数解读3.1 环境搭建与依赖确认打开压缩包先看目录结构核心是MLAlgorithms目录放算法实现以及根目录的Sniffer.py、DataProcessor.py、SVM.py。WebPackageSniffer目录负责 Web 流量抓取模块。在跑之前先把依赖装齐典型环境为 Python 3.7 到 3.10 之间需要的库如下pip install scikit-learn pandas numpy scapy pyshark matplotlib逻辑说明scapy用于抓包和解析报文pyshark是对 Wireshark 的 tshark 命令的封装scikit-learn提供 SVM 实现matplotlib用于后面可视化输出。pyshark依赖系统安装 Wireshark环境变量里要能找到tshark。如果提示找不到tshark属于环境变量问题。Windows 下把 Wireshark 安装目录默认C:\Program Files\Wireshark加入PATHLinux 下检查which tshark没装则执行sudo apt install tshark注意安装过程中会询问是否允许非 root 用户抓包选是否则每次都要用 sudo 执行。3.2 数据集准备与标签理解训练一个监督分类器必须有标注数据。最常用的毕设数据集是 NSL-KDD修正版 KDDCUP99它已经处理好了特征列和标签列省去了大量预处理工作。典型列结构包含 41 个特征加 1 个标签列标签有normal和各类攻击类型。使用时的处理惯例是把多分类任务折叠成二分类正常流量记为 0所有攻击类型统一记为 1。折叠后对模型的压力会小很多也贴合这个项目的「正常 / 异常」二分类定位。读取与折叠的代码一般写在DataProcessor.py里import pandas as pd df pd.read_csv(KDDTrain.txt, headerNone) label_col df.columns[-1] df[label] df[label_col].apply(lambda x: 0 if x normal else 1) # 丢弃原始协议类型列和 label 列保留数值特征 feats [c for c in df.columns if c not in [label_col, label]] X df[feats].values y df[label].values逻辑说明KDD 数据集的原始标签是字符串需要一次性映射成整数。代码里apply仅针对最后一列做判断normal映射为 0其他所有攻击标签映射为 1这是最常见的做法。参数说明feats列表排除了原始标签列但要注意 KDD 里还有两列是符号型特征协议类型、服务类型直接喂给 SVM 前要做编码或直接丢弃。我的习惯是先丢掉协议和服务列只保留连续数值特征因为这两个符号字段在真实流量里分布差异极大容易造成过拟合。3.3 训练流程与 SVM 参数调优SVM.py里提供的训练流程一般分为四步标准化 → 划分数据集 → 训练 → 输出评估指标。完整流程可以写成一个可复用的函数def train_svm(X, y, test_size0.3, kernelrbf, C10.0, gammascale): X_train, X_test, y_train, y_test train_test_split( X, y, test_sizetest_size, random_state42, stratifyy ) scaler StandardScaler().fit(X_train) X_train_scaled scaler.transform(X_train) X_test_scaled scaler.transform(X_test) clf SVC(kernelkernel, CC, gammagamma, class_weightbalanced) clf.fit(X_train_scaled, y_train) y_pred clf.predict(X_test_scaled) acc (y_pred y_test).mean() print(fAccuracy: {acc:.4f}) print(confusion_matrix(y_test, y_pred)) return clf, scaler逻辑说明scaler必须在训练集上fit再对测试集只做transform这是为了保持特征缩放的一致性。返回值里clf和scaler都要保留后续实时检测时需要先用scaler处理在线提取的特征再交给clf预测。参数说明random_state42固定随机种子让每次运行结果可复现这是答辩前必须养成的习惯test_size0.3表示测试集占 30%如果数据量小可以调到 0.2但不宜更小否则评估结果不稳定。调参时可以先用网格搜索找一组较优的C和gamma再人工微调from sklearn.model_selection import GridSearchCV param_grid { C: [1, 10, 100], gamma: [scale, 0.01, 0.001] } grid GridSearchCV(SVC(kernelrbf, class_weightbalanced), param_grid, cv5) grid.fit(X_train_scaled, y_train) print(grid.best_params_, grid.best_score_)逻辑说明网格搜索会在每个参数组合上做 5 折交叉验证选出平均分数最高的一组。这样比手动试参更系统缺点是计算量大KDD 数据集规模不大通常一分钟内能跑完。这里有个经验C从 1 到 100 之间调整gamma如果设得太大比如 0.1 以上SVM 会把每个样本周围都画得很紧训练集分数很高测试集却明显下跌这是过拟合的信号。4. 入侵检测应用中的常见问题与排查流量丢了、样本偏了、误报炸了4.1 抓包长度不足导致特征算出来全是 0现象用Sniffer.py跑实时抓包喂给DataProcessor.py后会话特征的包长度、字节数全部为 0SVM 预测结果清一色是正常。原因抓包时网卡没有开启混杂模式或者抓包长度被限制。在交换网络环境下非混杂模式只能收到发往本机的报文大量旁路流量根本进不来抓包长度snaplen设置过短则只能拿到报文头部载荷被截断虽然长度字段还在但重组和统计会出问题。解决在Sniffer.py里显式设置网卡为混杂模式并把 snaplen 调大import pyshark # 开启混杂模式抓完整包timeout 表示窗口时长秒 capture pyshark.LiveCapture( interfaceeth0, use_jsonTrue, bpf_filtertcp or udp, monitor_modeFalse ) capture.sniff(timeout60)逻辑说明bpf_filter限制只抓 TCP 和 UDP 报文减少不必要的处理开销timeout60表示持续抓取 60 秒然后统一输出这批流量做分析。如果是在虚拟机上跑注意虚拟机网卡默认可能是 NAT 模式只能看到虚拟网段里的流量需要改成桥接模式才能看到真实局域网流量。还有一个容易被忽略的细节在 WSL 环境下跑pyshark抓包经常拿不到接口列表这不是代码问题是 WSL 的网卡架构不支持原生抓包建议直接换到原生 Linux 环境。4.2 样本不平衡导致全判成正常流量现象训练完成后准确率 99%但看混淆矩阵发现异常样本的召回率接近于 0模型把所有流量都判成正常。原因真实网络里正常流量占绝对多数异常流量可能不足 1%。SVM 在默认设置下追求整体准确率少数类的误分对损失函数贡献太小因此边界会偏向多数类一侧。解决在创建SVC实例时加class_weightbalanced它的作用是按类别样本数量的反比自动调整权重少数类分错时付出更高的代价。代码前面已经出现过这里列一下调整前和调整后的效果差异# 调整前异常样本几乎全部漏报 clf SVC(kernelrbf, C10, gammascale) # 调整后少数类召回率明显上升但误报也会增加 clf SVC(kernelrbf, C10, gammascale, class_weightbalanced)逻辑说明class_weightbalanced等于给样本量少的类别更高的误分类惩罚代价是正常流量的误报率会上升这是安全场景下的合理取舍——漏掉一次攻击的代价通常比多告警一次高得多。4.3 离线训练与在线检测的特征不一致现象训练集上用 NSL-KDD 评估效果不错但把Sniffer.py抓到的新流量喂给同一个模型预测结果完全失真。原因离线训练的特征来自已经切分好的会话记录而实时抓包的流量是连续不断的字节流切分方式不同导致每条流的特征计算口径不一致模型输入分布完全变了。解决把特征提取逻辑封装成一个统一函数在线和离线共用同一套特征计算代码。我一般这样做def build_flow_features(packets): 统一的流特征提取入口离线和实时都走这里 flows sessionize_packets(packets) features extract_session_features(flows) features drop_ip_and_port(features) return features逻辑说明sessionize_packets负责按五元组和超时阈值切分会话extract_session_features计算统计量drop_ip_and_port保证特征与离线训练时一致。只要离线训练的数据集预处理也调用了这个函数特征口径就不会错位。4.4 SVM 预测结果全是同一个类别现象对一批新流量预测输出全部为 0 或全部为 1。原因predict_proba需要SVC(probabilityTrue)才能计算概率默认的 SVM 模型只能输出类别标签和决策距离。如果代码里调用了predict_proba而实例化时没加probabilityTrue会直接报错不报错但全判一类的情况通常是特征标准化参数不匹配比如用训练集的scaler处理在线特征但由于在线流量统计量异常标准化后特征值全部落在边界同一侧。解决在SVC实例化时开启probabilityTrue并检查在线特征的值域范围是否与训练集一致clf SVC(kernelrbf, C10, gammascale, class_weightbalanced, probabilityTrue) distances clf.decision_function(X_test_scaled)逻辑说明decision_function返回的是样本到超平面的带符号距离距离越大说明越像正常类为负、越大说明越像异常类为正。用这个距离做阈值比直接用predict更灵活可以在告警敏感度上做调节。4.5 抓包程序长时间运行内存持续上涨现象Sniffer.py连续跑几小时内存占用涨到几个 GB最后进程被杀。原因实时抓包时把报文全部缓存在列表里会话超时后没有及时回收数据结构。解决对 TCP 流做主动老化超过 120 秒没有新包的会话强制关闭并释放内存# flow_table 是五元组 id - 会话对象 for fid in list(flow_table.keys()): if time.time() - flow_table[fid].last_seen 120: processed extract_session_features({fid: flow_table[fid]}) flow_table.pop(fid)逻辑说明这段代码在每次抓取新包时顺带扫描一次流表超时会话先提取特征再删除引用。判断条件里的 120 秒要大于你的会话超时阈值不然正常的慢连接会被提前切断。5. 从离线检测到实时检测把 Sniffer 接入 SVM 判别的重构思路5.1 实时检测的整体结构WebPackageSniffer目录配合Sniffer.py可以做实时检测改造。整体结构是三条线并行抓包线程持续接收报文会话管理模块做流聚合检测模块按时间窗口触发 SVM 判定。三个模块之间用队列解耦避免抓包阻塞影响判别。核心工作流是抓包线程把每个包解析成(五元组, 长度, 时间戳)的元组入队会话管理模块从队列取包更新对应的流对象每隔 N 秒把当前活跃会话的特征做一次批量提取送入SVM.py训练好的模型预测。import queue import threading import time pkt_queue queue.Queue() session_table {} def packet_callback(packet): # 解析包信息并放入队列 pkt_queue.put({ src_ip: packet.ip.src, dst_ip: packet.ip.dst, src_port: packet[packet.transport_layer].srcport, dst_port: packet[packet.transport_layer].dstport, length: int(packet.length), timestamp: float(packet.frame_info.time_epoch) }) def detector_loop(clf, scaler, window60): while True: time.sleep(window) pending [] for fid, flow in session_table.items(): if time.time() - flow[last_seen] window: pending.append((fid, flow)) features extract_session_features(dict(pending)) scaled scaler.transform(features.drop(columns[flow_id])) preds clf.predict(scaled) for fid, pred in zip(features[flow_id], preds): if pred 1: print(f[ALERT] 异常流 {fid})逻辑说明抓包回调只是快速解析入队不做任何统计避免在高流量下阻塞抓包线程。检测循环按固定时间窗口触发窗口内活跃的会话统一推理这样能批量复用模型预测接口效率比每来一个包预测一次高得多。参数说明window60表示 60 秒触发一次全量检测窗口大小需要权衡实时性和特征稳定性。窗口太短5 秒内统计量方差大模型输出不稳定太长300 秒以上异常行为已经被放大到影响业务告警失去意义。实际部署时我会在 30 到 90 秒之间试跑观察误报率变化再定。5.2 窗口切分与特征对齐实时检测里最容易翻车的点是「一个会话跨窗口」。假设窗口是 60 秒某条 TCP 连接在 0 到 120 秒持续传输第一次检测窗口只算了它前半段流量第二次窗口只算了后半段两次的特征分布必然不同模型的判断就无法稳定。处理策略是双窗口会话在第一次被检测后继续累积统计信息但打上「已检测」标记如果后续窗口里它的特征发生突变比如包长方差突然缩小再触发一次告警。这个逻辑比简单按窗口切分要复杂但更接近真实攻击的渐变特征。5.3 告警输出与可视化方式实时检测的告警要有数据支撑不该只是打印一行文字。实践中我会把告警信息落成结构化日志import json import datetime alert { time: datetime.datetime.now().isoformat(), flow_id: fid, confidence: float(clf.decision_function(sample)[0]), src_ip: flow[src_ip], dst_ip: flow[dst_ip], } with open(alerts.jsonl, a) as f: f.write(json.dumps(alert) \n)逻辑说明decision_function输出的距离可以作为置信度参考距离越大说明越偏离正常边界。把告警按 JSON 行格式写入文件后续可以直接导入 Elasticsearch 或做成表格展示毕设答辩时也方便给评审看真实输出样例。6. 检测结果的验证与进阶混淆矩阵、阈值调节与特征扩展完成基础训练后先不要急着上实时检测把评估环节做到位更重要。很多人在答辩时只会报一个准确率评审追问「那误报率多少、漏报率多少」就答不上来。这里建议至少做两件事。第一件事是输出混淆矩阵的标准化视图和召回率曲线。对于入侵检测场景异常类的召回率真正例率才是核心指标准确率反而容易被不平衡样本带偏。可以在SVM.py基础上加一段import matplotlib.pyplot as plt from sklearn.metrics import ConfusionMatrixDisplay ConfusionMatrixDisplay.from_estimator( clf, X_test_scaled, y_test, display_labels[normal, attack], cmapBlues ) plt.savefig(confusion_matrix.png)第二件事是调低决策阈值以提高异常召回率。SVM 在默认情况下以决策距离 0 为分界线但你可以把分界线往正常方向偏移让更多模糊样本被判为异常distances clf.decision_function(X_test_scaled) # 默认阈值是 0改为 -0.5 后更多样本会被判为异常 y_pred_custom (distances -0.5).astype(int)阈值调整这是一种权衡阈值越小召回率越高误报也会变多。具体设多少取决于你对漏报的容忍度没有标准答案应该跑一组不同阈值的对比数据画出召回率与误报率的曲线这组图拿到答辩现场比任何口头解释都有说服力。进阶方向上可以进一步在现有 SV‌M 基础上做特征扩展比如把时间窗内流量的目的端口熵值加入特征向量对端口扫描类攻击会更敏感或者把分类器替换为梯度提升树做对比实验常见的做法是把SVC换为GradientBoostingClassifier复用同一套特征和评估逻辑对比两者在召回率和训练时间上的差异这组对比数据能直接扩充成毕设的一个章节。这套代码我用下来最大的感受是作为毕设源码它的目录划分足够干净Sniffer、DataProcessor、SVM三层职责边界分明改起来不费劲但距离真正的产品化检测还有明显距离需要把会话老化机制、特征口径统一和告警结构化补上。从那以后我做任何流量检测项目都会强制走一遍「统一特征函数 不平衡处理 阈值验证」这条流程至少能少踩一半的坑。希望帮到你。本文还有配套的精品资源点击获取