简介这份资源是面向高校学生与初学者的毕业设计/期末大作业级项目主题为基于监督学习的Web入侵检测系统适合需要完成课程设计、毕设或想入门安全检测与机器学习实战的人群。包内共60个文件以18个ipynb实验笔记、14个txt样本与说明、14个html页面、8个py脚本为主另含pkl/pickle模型文件与md说明压缩包约2.25MB结构清晰便于按模块查阅。项目围绕Web请求日志与payload数据展开涵盖数据爬取、字符处理、去重、参数提取、特征保存与模型训练等环节代码注释较完整新手也能理解整体流程。已有145人学习说明其参考价值获得一定认可。下载后可获得一套可部署运行的完整源码与文档说明便于快速复现实验、理解监督学习在入侵检测中的应用并在此基础上完成自己的论文或答辩展示。1. 基于监督学习的web入侵检测系统从HTTP请求到告警的完整链路很多同学做毕业设计时第一反应是找个现成的WAF规则引擎套壳把ModSecurity的规则文件读一遍就算完事。但真正让答辩老师眼前一亮的是你能说清楚一条HTTP请求进来之后怎么被拆成特征向量、怎么被模型判定为恶意、又怎么把结果落成一条可追溯的告警。这套基于监督学习的web入侵检测系统核心思路就是把标注好的正常请求和攻击请求喂给分类器让模型自己学出判别边界而不是靠人工写正则。它适合有Python基础、想做出可演示可复现系统的本科生也适合想从规则匹配转向机器学习检测的运维安全方向从业者。源码和文档说明的价值不在于代码量多大而在于整条链路是通的数据从哪来、特征怎么提、模型怎么训、服务怎么起、告警怎么存每一步都能跑起来看到结果。2. 数据与特征工程把HTTP请求变成模型能吃的数字2.1 为什么选监督学习而不是规则匹配规则匹配的痛点是写不完。SQL注入的变种、XSS的编码绕过、命令注入的分隔符组合靠正则一条条补补到最后规则文件几千行误报还是压不住。监督学习的优势在于你只需要标注一批样本模型会自动找区分度最高的特征组合。常见做法是用公开数据集打底比如CSIC 2010 HTTP数据集里面有几万条正常请求和攻击请求覆盖SQL注入、XSS、路径遍历、参数篡改等类型。如果觉得公开数据集太旧也可以自己用DVWA或Pikachu靶场生成流量再用Burp Suite抓包导出。我一般会建议两条腿走路公开数据集保证样本量和标注质量自采数据保证场景贴合度。选监督学习还有一个现实原因毕业设计需要可量化的评估指标。准确率、召回率、F1值这些数字比“我写了500条规则”更有说服力。而且监督学习的训练过程本身就是可展示的从特征分布到混淆矩阵每一步都能放进论文里。2.2 请求解析与特征提取的实操步骤拿到原始HTTP请求后第一步是解析。不要直接对原始字符串做哈希那样特征维度爆炸且没有泛化能力。正确的做法是按HTTP协议结构拆开分别处理请求行、请求头、请求体。import re from urllib.parse import unquote def parse_http_request(raw_request): 解析原始HTTP请求文本返回结构化字典 raw_request: 从数据集或抓包工具导出的单条请求字符串 lines raw_request.strip().split(\n) # 请求行METHOD PATH HTTP/VERSION request_line lines[0].strip() method, path, version request_line.split( , 2) # 请求头 headers {} body_start 0 for i, line in enumerate(lines[1:], start1): if line.strip() : body_start i 1 break if : in line: key, value line.split(:, 1) headers[key.strip().lower()] value.strip() # 请求体 body \n.join(lines[body_start:]) if body_start else # URL解码还原编码绕过 decoded_path unquote(path) decoded_body unquote(body) return { method: method, path: decoded_path, version: version, headers: headers, body: decoded_body, raw_path: path, raw_body: body }这段代码的关键点在于unquote解码。攻击者常用%27代替单引号、%20代替空格来绕过简单规则解码后再提特征才能抓住本质。参数说明raw_request是单条请求的原始文本通常以\n分隔返回的字典里同时保留原始和解码后的字段方便后续做对比特征。解析完之后提特征。我一般会分四组第一组是长度类特征URL长度、请求体长度、参数个数、请求头数量。攻击请求往往参数异常多或异常长比如SQL注入的UNION查询会拉长URL。第二组是字符分布特征特殊字符占比。统计、、、、(、)、;、--、/*这些字符在URL和body中的出现次数和比例。正常请求里这些字符很少攻击请求里密集出现。第三组是关键词特征是否包含select、union、script、alert、../、cmd、exec等敏感词。注意不要直接做二值化而是统计出现次数因为正常请求也可能偶然包含某个词。第四组是编码特征URL编码字符的比例、双重编码的比例、非ASCII字符的比例。编码绕过是攻击者的常用手段编码比例异常高往往意味着有人在尝试绕过。import numpy as np def extract_features(parsed): 从解析后的请求中提取数值特征向量 parsed: parse_http_request的返回值 path parsed[path] body parsed[body] full_text path body features [] # 长度特征 features.append(len(path)) features.append(len(body)) features.append(len(parsed[headers])) features.append(path.count() 1) # 参数个数估算 # 特殊字符统计 special_chars [, , , , (, ), ;, --, /*, ../] for char in special_chars: features.append(full_text.count(char)) # 敏感关键词统计 keywords [select, union, script, alert, cmd, exec, drop, insert, update, delete, eval] lower_text full_text.lower() for kw in keywords: features.append(lower_text.count(kw)) # 编码特征 raw_full parsed[raw_path] parsed[raw_body] features.append(raw_full.count(%) / (len(raw_full) 1)) features.append(sum(1 for c in full_text if ord(c) 127) / (len(full_text) 1)) return np.array(features, dtypenp.float32)参数说明special_chars和keywords列表可以根据你的数据集调整但不要加太多否则特征维度太高容易过拟合。我一般控制在30维以内足够区分大部分攻击类型。features最后转成float32是为了后续模型训练时节省内存。提示特征提取函数一旦确定训练和推理必须用同一个函数。很多同学训练时用一套代码部署时又重写一遍结果特征顺序对不上模型输出全是错的。建议把extract_features单独放在一个模块里训练脚本和服务脚本都import它。2.3 数据集划分与类别不平衡处理公开数据集里正常请求通常远多于攻击请求直接训练会导致模型偏向多数类。常见做法是欠采样正常请求或过采样攻击请求。我一般用SMOTE对攻击样本做过采样同时保留原始正常样本的70%。划分比例按7:2:1分训练集、验证集、测试集注意要按时间顺序划分而不是随机打乱因为真实场景中攻击是持续演进的随机打乱会高估模型性能。from sklearn.model_selection import train_test_split from imblearn.over_sampling import SMOTE # X是特征矩阵y是标签0正常1攻击 X_train, X_temp, y_train, y_temp train_test_split( X, y, test_size0.3, random_state42, stratifyy ) X_val, X_test, y_val, y_test train_test_split( X_temp, y_temp, test_size0.33, random_state42, stratifyy_temp ) # 只对训练集做SMOTE验证集和测试集保持原始分布 smote SMOTE(random_state42) X_train_res, y_train_res smote.fit_resample(X_train, y_train)这里有个血泪经验SMOTE只能在训练集上做验证集和测试集绝对不能过采样。否则你评估出来的召回率是虚高的上线后一跑就翻车。3. 模型选型与训练从逻辑回归到集成学习的取舍3.1 三个基线模型的对比与选择理由毕业设计不需要一上来就上深度学习。HTTP请求的特征维度不高树模型和线性模型往往就够了。我一般会跑三个基线逻辑回归、随机森林、XGBoost。逻辑回归作为最简基线用来验证特征是否线性可分随机森林看特征重要性和非线性边界XGBoost作为最终候选通常在准确率和召回率上都是最好的。模型训练速度可解释性小样本表现部署复杂度逻辑回归快强一般低随机森林中等中等较好低XGBoost较慢较弱好中等简单MLP慢弱差高选XGBoost的理由很实际它自带特征重要性输出答辩时可以展示哪些特征对检测贡献最大比如特殊字符占比和敏感词计数通常排在前列。而且XGBoost的scale_pos_weight参数可以直接处理类别不平衡不用额外做采样。import xgboost as xgb from sklearn.metrics import classification_report, confusion_matrix # 计算正负样本比例用于scale_pos_weight scale (y_train_res 0).sum() / (y_train_res 1).sum() model xgb.XGBClassifier( n_estimators200, # 树的数量200在大多数场景够用 max_depth6, # 树深度太深容易过拟合 learning_rate0.1, # 学习率0.1是常用起点 scale_pos_weightscale, # 处理类别不平衡 eval_metriclogloss, random_state42 ) model.fit( X_train_res, y_train_res, eval_set[(X_val, y_val)], verboseFalse ) # 在测试集上评估 y_pred model.predict(X_test) print(classification_report(y_test, y_pred, target_names[正常, 攻击])) print(confusion_matrix(y_test, y_pred))参数说明n_estimators设200是因为再往上增加对F1提升不明显但训练时间线性增长max_depth设6是防止模型记住训练集中的噪声scale_pos_weight让模型更关注攻击样本提高召回率。如果测试集上误报率太高可以把scale_pos_weight调小一点牺牲一点召回换精确率。3.2 训练过程中的监控与早停XGBoost支持早停在验证集损失不再下降时停止训练避免过拟合。把early_stopping_rounds设成20意思是验证集损失连续20轮不下降就停。model xgb.XGBClassifier( n_estimators500, max_depth6, learning_rate0.1, scale_pos_weightscale, eval_metriclogloss, early_stopping_rounds20, random_state42 ) model.fit( X_train_res, y_train_res, eval_set[(X_val, y_val)], verbose10 # 每10轮打印一次验证集指标 )训练过程中要盯着验证集的logloss和error rate。如果训练集loss一直降但验证集loss开始上升说明过拟合了这时候要么减特征要么加正则。我一般会在XGBClassifier里再加reg_alpha0.1和reg_lambda1.0这两个L1和L2正则项对控制过拟合很有效。3.3 模型持久化与推理服务封装训练完的模型要存下来用joblib或XGBoost自带的save_model都行。我习惯用joblib因为可以同时存模型和特征提取的配置。import joblib # 保存模型 joblib.dump(model, web_ids_model.pkl) # 保存特征提取函数所需的配置关键词列表等 config { special_chars: [, , , , (, ), ;, --, /*, ../], keywords: [select, union, script, alert, cmd, exec, drop, insert, update, delete, eval] } joblib.dump(config, feature_config.pkl)推理服务用Flask起一个HTTP接口接收原始请求文本返回是否攻击的判定和置信度。from flask import Flask, request, jsonify import joblib import numpy as np app Flask(__name__) model joblib.load(web_ids_model.pkl) config joblib.load(feature_config.pkl) app.route(/detect, methods[POST]) def detect(): raw request.json.get(request, ) parsed parse_http_request(raw) features extract_features(parsed).reshape(1, -1) pred model.predict(features)[0] prob model.predict_proba(features)[0][1] return jsonify({ is_attack: bool(pred), confidence: float(prob), label: 攻击 if pred else 正常 }) if __name__ __main__: app.run(host0.0.0.0, port5000)参数说明port5000是Flask默认端口如果被占用可以改成5001。request.json.get(request)接收的是原始HTTP请求文本调用方需要把请求序列化成字符串。返回的confidence是模型判定为攻击的概率可以用来做告警分级比如置信度大于0.9的告警直接推送0.7到0.9的进人工审核队列。4. 系统集成与告警落地从单次检测到持续监控4.1 请求采集与流式检测的对接方式单次检测接口只能做演示真实场景需要持续采集流量。常见做法是在Nginx或Apache前面挂一个日志采集器把access log实时推给检测服务。如果不想动生产环境可以用mitmproxy做旁路监听把请求转发到检测接口。# 用mitmproxy的脚本模式把请求转发到检测服务 mitmdump -s forward_to_ids.py --listen-port 8080# forward_to_ids.py import requests from mitmproxy import http def request(flow: http.HTTPFlow): raw flow.request.method flow.request.path HTTP/1.1\n for k, v in flow.request.headers.items(): raw f{k}: {v}\n raw \n flow.request.get_text() try: resp requests.post(http://127.0.0.1:5000/detect, json{request: raw}, timeout1) result resp.json() if result[is_attack]: print(f[告警] {flow.request.url} 置信度: {result[confidence]:.3f}) except Exception as e: print(f检测服务异常: {e})这段脚本的作用是把经过mitmproxy的每个请求转发给检测服务如果判定为攻击就打印告警。timeout1是防止检测服务卡住导致代理阻塞。实际部署时可以把告警写到文件或推送到消息队列方便后续做告警聚合和去重。4.2 告警存储与误报反馈闭环告警不能只打印在控制台要存下来做后续分析。用SQLite就够毕业设计用了建一张表记录时间、请求摘要、判定结果、置信度。CREATE TABLE alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, alert_time DATETIME DEFAULT CURRENT_TIMESTAMP, request_summary TEXT, is_attack INTEGER, confidence REAL, feedback INTEGER DEFAULT -1 -- -1未反馈0误报1确认攻击 );feedback字段是关键。每次告警后人工可以标记是误报还是真实攻击这些反馈数据积累起来可以重新训练模型形成闭环。我一般会在Flask里加一个/feedback接口接收告警ID和反馈标签更新数据库。import sqlite3 app.route(/feedback, methods[POST]) def feedback(): alert_id request.json.get(alert_id) label request.json.get(label) # 0误报1确认攻击 conn sqlite3.connect(alerts.db) conn.execute(UPDATE alerts SET feedback ? WHERE id ?, (label, alert_id)) conn.commit() conn.close() return jsonify({status: ok})这个闭环的价值在于模型上线后不是一成不变的。每周把标记为误报的样本重新提特征加入训练集重新训练模型的误报率会逐步下降。这也是答辩时能讲清楚的工程亮点。4.3 检测服务的性能压测与优化Flask自带的开发服务器性能很差单线程处理并发一高就排队。用gunicorn起多worker能显著提升吞吐。gunicorn -w 4 -b 0.0.0.0:5000 app:app --timeout 30-w 4表示4个worker进程一般设成CPU核数的2倍。--timeout 30是防止某个请求卡住导致worker被挂起。压测用ab或wrk测一下QPS和P99延迟。如果QPS上不去优先看特征提取函数是不是有性能瓶颈比如正则匹配太慢可以预编译正则或改用字符串查找。# 用ab压测-n总请求数-c并发数 ab -n 1000 -c 50 -p request.json -T application/json http://127.0.0.1:5000/detectrequest.json里放一条示例请求的JSON。压测结果关注两个指标Requests per second和Time per request的99%分位。如果P99超过100ms就要考虑把特征提取里的循环改成向量化操作或者把模型推理放到单独的进程池里。5. 避坑与排查那些让模型翻车的细节5.1 特征提取顺序不一致导致推理全错现象训练时准确率95%部署后所有请求都判成正常或都判成攻击。原因训练脚本和服务脚本各自实现了一遍特征提取关键词列表顺序不同或特殊字符列表有差异导致特征向量错位。解决把extract_features和配置字典抽到独立模块训练和服务都import同一个函数配置从同一个pkl文件加载。5.2 数据集泄露导致评估虚高现象测试集F1值0.98上线后误报率极高。原因划分数据集时随机打乱同一条请求的变体同时出现在训练集和测试集模型记住了样本而不是学到了模式。解决按请求来源或时间划分确保训练集和测试集的请求来自不同时间段或不同IP段。如果数据集本身没有时间戳至少按URL路径做分组划分同一路径的请求只出现在一个集合里。5.3 URL解码不彻底导致编码绕过漏检现象%27%20OR%2011这类请求被判定为正常。原因只做了一次unquote攻击者用了双重编码%2527解码一次后还是%27。解决在特征提取前做循环解码直到不再包含%或达到最大解码次数比如3次。同时把解码次数作为特征之一双重编码本身就是可疑信号。def deep_unquote(text, max_rounds3): 循环解码直到没有%或达到最大次数 for _ in range(max_rounds): new_text unquote(text) if new_text text: break text new_text return text5.4 模型文件路径硬编码导致迁移失败现象在本地跑得好好的换台机器就报FileNotFoundError。原因代码里写了绝对路径/home/user/model.pkl。解决用相对路径加os.path.dirname(__file__)拼接或者把模型路径做成环境变量。我一般会在项目根目录放一个config.py所有路径基于项目根目录计算。5.5 告警风暴导致服务不可用现象攻击者用大量请求轰炸检测服务每个请求都写数据库SQLite锁竞争导致响应超时。原因同步写库在并发高时成为瓶颈。解决告警先写内存队列后台线程批量入库或者直接用Redis做缓冲定时刷到SQLite。另外对同一IP的告警做聚合比如60秒内同一IP只记一条告警避免告警表被刷爆。6. 进阶技巧用特征重要性反推攻击类型与模型迭代模型训练完之后model.feature_importances_能告诉你哪些特征对判定贡献最大。我一般会把这个排序打印出来对照特征名称看。如果排第一的是special_chars里的单引号计数说明SQL注入样本占主导如果排前面的是script关键词计数说明XSS检测能力强。这个分析可以写进论文的“实验结果与分析”章节比单纯列准确率有深度。更进一步可以用SHAP值做单条请求的解释。SHAP能告诉你对于某一条判定为攻击的请求每个特征贡献了多少分。这在答辩演示时非常有用拿一条SQL注入请求展示模型是因为检测到union关键词和单引号数量异常才判攻击的而不是黑盒输出。import shap # 用TreeExplainer解释XGBoost模型 explainer shap.TreeExplainer(model) shap_values explainer.shap_values(X_test[:10]) # 打印第一条攻击样本的特征贡献 for i, val in enumerate(shap_values[1][0]): if abs(val) 0.01: print(f特征{i}: 贡献 {val:.4f})shap_values[1]是攻击类的SHAP值正值表示该特征推动判定为攻击负值表示推动判定为正常。特征索引对应extract_features里的顺序可以建一个映射表把索引翻译成特征名。另一个进阶方向是模型迭代。每次收集到新的误报样本不要直接删掉而是标注后加入训练集重新训练。我习惯每积累200条新样本就重训一次用joblib覆盖旧模型文件服务端用watchdog监听模型文件变化自动重载。这样模型能持续适应新的攻击模式而不是上线即固化。import os import time from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class ModelReloadHandler(FileSystemEventHandler): def on_modified(self, event): if event.src_path.endswith(web_ids_model.pkl): global model model joblib.load(web_ids_model.pkl) print(f[{time.strftime(%H:%M:%S)}] 模型已重载) observer Observer() observer.schedule(ModelReloadHandler(), path., recursiveFalse) observer.start()这段代码让服务在模型文件更新后自动加载新模型不用重启服务。watchdog在Linux和Windows上都能用注意文件路径要写对。最后说一个我踩过的坑不要追求一步到位的高准确率。我第一个版本用了一个很复杂的深度网络测试集F1到了0.99但推理一次要200ms而且换一批新数据就崩。后来退回到XGBoost加30维手工特征F1降到0.94但推理只要5ms泛化也稳得多。毕业设计也好实际项目也好可复现、可解释、可迭代比刷指标重要得多。希望帮到你。本文还有配套的精品资源点击获取