简介：这份资源是面向高校学生与初学者的毕业设计/期末大作业级项目，主题为基于监督学习的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:], start=1): 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, dtype=np.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_size=0.3, random_state=42, stratify=y ) X_val, X_test, y_val, y_test = train_test_split( X_temp, y_temp, test_size=0.33, random_state=42, stratify=y_temp ) # 只对训练集做SMOTE，验证集和测试集保持原始分布 smote = SMOTE(random_state=42) X_train_res, y_train_res = smote.fit_resample(X_train, y_train)

这里有个血泪经验：SMOTE只能在训练集上做，验证集和测试集绝对不能过采样。否则你评估出来的召回率是虚高的，上线后一跑就翻车。

3. 模型选型与训练：从逻辑回归到集成学习的取舍

3.1 三个基线模型的对比与选择理由

毕业设计不需要一上来就上深度学习。HTTP请求的特征维度不高，树模型和线性模型往往就够了。我一般会跑三个基线：逻辑回归、随机森林、XGBoost。逻辑回归作为最简基线，用来验证特征是否线性可分；随机森林看特征重要性和非线性边界；XGBoost作为最终候选，通常在准确率和召回率上都是最好的。

选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_estimators=200, # 树的数量，200在大多数场景够用 max_depth=6, # 树深度，太深容易过拟合 learning_rate=0.1, # 学习率，0.1是常用起点 scale_pos_weight=scale, # 处理类别不平衡 eval_metric='logloss', random_state=42 ) model.fit( X_train_res, y_train_res, eval_set=[(X_val, y_val)], verbose=False ) # 在测试集上评估 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_estimators=500, max_depth=6, learning_rate=0.1, scale_pos_weight=scale, eval_metric='logloss', early_stopping_rounds=20, random_state=42 ) model.fit( X_train_res, y_train_res, eval_set=[(X_val, y_val)], verbose=10 # 每10轮打印一次验证集指标 )

训练过程中要盯着验证集的logloss和error rate。如果训练集loss一直降但验证集loss开始上升，说明过拟合了，这时候要么减特征要么加正则。我一般会在XGBClassifier里再加reg_alpha=0.1和reg_lambda=1.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(host='0.0.0.0', port=5000)

参数说明：port=5000是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}, timeout=1) result = resp.json() if result['is_attack']: print(f"[告警] {flow.request.url} 置信度: {result['confidence']:.3f}") except Exception as e: print(f"检测服务异常: {e}")

这段脚本的作用是把经过mitmproxy的每个请求转发给检测服务，如果判定为攻击就打印告警。timeout=1是防止检测服务卡住导致代理阻塞。实际部署时可以把告警写到文件或推送到消息队列，方便后续做告警聚合和去重。

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/detect

request.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%201=1这类请求被判定为正常。原因：只做了一次unquote，攻击者用了双重编码%2527，解码一次后还是%27。解决：在特征提取前做循环解码，直到不再包含%或达到最大解码次数（比如3次）。同时把解码次数作为特征之一，双重编码本身就是可疑信号。

def deep_unquote(text, max_rounds=3): """循环解码直到没有%或达到最大次数""" for _ in range(max_rounds): new_text = unquote(text) if new_text == text: break text = new_text return text

5.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='.', recursive=False) observer.start()

这段代码让服务在模型文件更新后自动加载新模型，不用重启服务。watchdog在Linux和Windows上都能用，注意文件路径要写对。

最后说一个我踩过的坑：不要追求一步到位的高准确率。我第一个版本用了一个很复杂的深度网络，测试集F1到了0.99，但推理一次要200ms，而且换一批新数据就崩。后来退回到XGBoost加30维手工特征，F1降到0.94，但推理只要5ms，泛化也稳得多。毕业设计也好，实际项目也好，可复现、可解释、可迭代比刷指标重要得多。希望帮到你。

本文还有配套的精品资源，点击获取