ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

用监督学习在Jupyter中实现Web入侵检测与模型部署

用监督学习在Jupyter中实现Web入侵检测与模型部署 简介基于Jupyter监督学习的Web入侵检测系统完整项目资料面向毕业设计、课程设计及安全方向的项目开发场景。系统定位为0day样本收集与检测工具采用scikit-learn中的支持向量机作为核心分类器并记录了从K近邻切换至支持向量机的完整调优思路。数据侧涵盖SQL注入样本收集、浏览器代理日志正则匹配与清洗、字符特征构造、去重和词处理等环节配套脚本覆盖数据收集、特征构建、模型训练与持久化保存的完整流程。压缩包共61个文件包括18个ipynb交互式笔记、8个Python脚本、用于训练和测试的文本数据集、已经保存的模型文件以及README说明文档整体仅2.36MB目录层级清晰明了便于按模块复现实验。目前已有116人学习浏览适合希望快速搭建入侵检测原型、认真理解监督学习数据链路并在此基础上扩展功能的学生或开发者。1. Web入侵检测为什么要用监督学习而不是规则库如果你是做运维或安全的大概率遇到过这种场景WAF或IDS的告警面板里堆着几千条可疑请求点开一看一半是扫描器在碰运气另一半是某个内部系统正常的API调用真正需要人工介入的只有个位数。规则库在这个环节的最大问题是它只能回答像不像回答不了概率多高。而监督学习擅长的恰恰是后者——用一批已经打上标签的流量样本训练分类器让模型输出一个置信度分数然后把告警优先级从这个分数上排出来。基于Jupyter来做这件事是因为整个工作流里最重的不是模型训练而是反复看分布、调特征、验证误报。Notebook的单元格执行模式和可视化输出天然适合安全数据分析这个过程。本文就顺着一条完整链路走一遍从原始Web日志到结构化特征再在Jupyter里完成监督学习模型的训练与评估最后把模型导出成可被业务系统调用的检测服务。文中所用的数据集、脚本和文档结构也对应了毕设或课程设计中常见的那套交付物组织方式。2. 从原始Web日志到训练集入侵检测的特征提取2.1 公开数据集与自采日志怎么选做Web入侵检测的监督学习第一步其实是数据问题。公开数据集里CSIC 2010、CICIDS2017这类经典集合都包含HTTP流量样本它们的好处是标签干净坏处是和真实业务流量差距大——生产环境里哪有那么多规规矩矩的POST表单。另一条路是自己采集把Nginx或Apache的access.log加上WAF拦截记录合并用拦截记录当攻击标签。这个做法的缺点是标签噪声大拦截的不一定是攻击漏拦截的攻击又出现在正常样本里。我一般的建议是毕设或课程设计优先用公开数据集做主体再用自采日志做补充验证。以CSIC 2010为例它的请求样本是文本格式每行是完整HTTP请求。处理时先按空格和换行把请求行、Header、Body拆出来再逐个字段做特征。这个过程在Jupyter里做非常舒服。import pandas as pd from urllib.parse import urlparse, unquote def parse_http_request(raw: str) - dict: lines raw.strip().split(\r\n) request_line lines[0].split( ) method request_line[0] url request_line[1] parsed urlparse(url) user_agent content_type for line in lines[1:]: if line.lower().startswith(user-agent:): user_agent line.split(:, 1)[1].strip() if line.lower().startswith(content-type:): content_type line.split(:, 1)[1].strip() return { method: method, path: parsed.path, query: parsed.query, user_agent: user_agent, content_type: content_type } # 示例解析一条原始请求 raw GET /product.php?id1 ORDER BY 3-- HTTP/1.1\r\nHost: test.com\r\nUser-Agent: Mozilla/5.0 parsed parse_http_request(raw) print(parsed)这段代码做了三件事拆分请求行拿到方法和URL用urlparse把URL拆成路径和查询串再遍历Header提取UA和Content-Type。注意unquote还没有用到等做特征时再处理URL编码因为攻击载荷经常做双重编码绕过。2.2 特征分为五组每组都有明确的入侵检测含义特征设计决定了模型的上限。我习惯把Web请求特征分成五组每一组对应一类攻击行为的痕迹。基础统计特征URL长度、参数个数、参数名最大长度。SQL注入和XSS的payload通常长而扭曲正常请求往往短而规整。字符分布特征特殊字符计数占比——单引号、双引号、分号、尖括号、百分号。这是区分SQL注入、XSS和路径遍历的核心信号。结构特征路径深度两级还是五级、是否含.、是否含..。路径穿越攻击几乎必有..。编码特征URL编码出现次数、十六进制字符占比、是否含%00等危险编码。语义特征把payload分词后匹配SQL关键字、JS关键字、系统命令关键字。这里不追求穷举能覆盖最典型的几类攻击即可。import re from urllib.parse import unquote SQL_KEYWORDS [select, union, insert, update, order by, sleep, benchmark] JS_KEYWORDS [alert, onerror, script, iframe, javascript] CMD_KEYWORDS [cat , id , ls , whoami, /etc/passwd, ;ls] def extract_features(parsed: dict) - dict: raw_query parsed.get(query, ) decoded unquote(raw_query) # 解码一次否则 %27 这类编码会绕过统计 full_payload f{parsed.get(path, )}?{raw_query}.lower() features {} features[url_len] len(full_payload) features[param_count] len(raw_query.split()) if raw_query else 0 features[single_quote_count] decoded.count() features[double_quote_count] decoded.count() features[angle_bracket_count] decoded.count() decoded.count() features[semicolon_count] decoded.count(;) features[percent_count] raw_query.count(%) features[has_encoded_char] 1 if re.search(r%(?:[0-9A-Fa-f]{2}), raw_query) else 0 def keyword_hit(keywords): return sum(1 for kw in keywords if kw in decoded.lower()) features[sql_kw_count] keyword_hit(SQL_KEYWORDS) features[js_kw_count] keyword_hit(JS_KEYWORDS) features[cmd_kw_count] keyword_hit(CMD_KEYWORDS) return features这里有一个关键设计unquote只做一次解码。攻击者可能做双重编码但一次解码已足以让大多数自动化攻击的统计特征暴露。如果你解码两次反而会把正常请求中合法的%25这种字符本来就是百分号编码还原成噪音。关键字计数没有用in硬匹配而是用sum(1 for ...)因为一个payload里出现select和sleep同时命中比只命中一个的信号更强。2.3 类别特征做编码数值特征做分箱Method、UA、Content-Type这些文本字段不能直接丢给RandomForestClassifier。一个保守做法是Method这种取值有限的字段做one-hotUA这种无限取值的字段只抽一个UA长度和一个是否包含常见UA关键字的二元特征。后者不做必然匹配——因为正常浏览器UA也经常升级、增减字段。数值特征中像url_len这种建议直接保留原始值就行树模型对数值尺度不敏感。但像单引号个数这类稀疏的计数特征有一个问题19个单引号和20个单引号在逻辑上几乎没区别。可以做一步截断处理超过10就在特征里仍记为10防止离群样本过度拉动节点分裂。3. 在Jupyter里跑通监督学习训练流程选型、切分与调参3.1 数据集结构设计与标签定义把所有请求样本解析、提取特征后组装成DataFrame。标签列建议定义为is_attack0表示正常1表示攻击。如果数据集本身带分类标签比如CSIC里区分了SQL注入、XSS、路径遍历等子类型建议先做二分类保留原始类别在另一列attack_type。这样做的理由是二分类的样本量更充足模型更稳而攻击细分类型可以在后期用另一个多分类模型来做两个模型解耦调试边界清晰。import pandas as pd from sklearn.model_selection import train_test_split df pd.read_csv(web_attack_features.csv) X df.drop([label, attack_type, raw_request], axis1, errorsignore) y df[label] X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.3, stratifyy, random_state42 ) print(X_train.shape, X_test.shape) print(y_train.value_counts())stratifyy是必须加的。Web入侵检测数据集天然不平衡正常样本往往远多于攻击样本不加分层抽样的话测试集里可能一条攻击样本都没有。3.2 三个模型打底逻辑回归、随机森林、XGBoostWeb入侵检测场景建议同时跑三个模型对比逻辑回归做基线随机森林看特征重要性XGBoost刷新AUC。逻辑回归的作用不是争最优分数而是验证特征的线性可分性特征做得好不好一眼就能看出来。随机森林的feature_importances_能告诉你哪些特征在贡献判断这对写论文或毕设文档很重要。from sklearn.ensemble import RandomForestClassifier from sklearn.linear_model import LogisticRegression from sklearn.metrics import classification_report, roc_auc_score import time models { lr: LogisticRegression(max_iter1000, class_weightbalanced), rf: RandomForestClassifier(n_estimators200, max_depth12, random_state42, n_jobs-1) } try: from xgboost import XGBClassifier models[xgb] XGBClassifier(n_estimators200, max_depth6, learning_rate0.1, eval_metriclogloss) except ImportError: print(xgboost not installed, skip) for name, model in models.items(): t0 time.time() model.fit(X_train, y_train) y_pred model.predict(X_test) y_proba model.predict_proba(X_test)[:, 1] print(f{name}: auc{roc_auc_score(y_test, y_proba):.4f}, acc{model.score(X_test, y_test):.4f}, time{time.time()-t0:.2f}s)class_weightbalanced让少数类在计算损失时权重更高。对于Web入侵检测里那种攻击样本只占5%的场景不设这个参数模型会无脑预测全部为正常。3.3 Jupyter里怎么看训练结果混淆矩阵与阈值准确率在入侵检测里基本是骗人的指标。样本中正常流量占95%模型全部预测为正常就能拿到95%准确率。要看的是召回率攻击样本中被识别出的比例和精确率预测为攻击的样本里真正是攻击的比例。这两个指标此消彼长靠一个东西来平衡——决策阈值。import numpy as np import matplotlib.pyplot as plt from sklearn.metrics import confusion_matrix best_model models[rf] y_proba best_model.predict_proba(X_test)[:, 1] threshold 0.3 # 调低阈值提升召回 y_pred_th (y_proba threshold).astype(int) cm confusion_matrix(y_test, y_pred_th) print(TN, FP, FN, TP:, cm.ravel()) print(TPR (召回率):, cm[1][1] / (cm[1][0] cm[1][1])) print(FPR (误报率):, cm[0][1] / (cm[0][0] cm[0][1]))调低阈值能提高召回代价是误报上升。实际部署时可以把阈值定在0.3到0.5之间然后让线上系统对0.3~0.7区间的告警做二次筛选而不是直接拦截。这是安全场景里比较务实的策略模型负责缩小范围人负责最终决策。4. Web入侵检测模型训练中必踩的四个坑4.1 时间穿越用未来数据训练模型Web流量最大的特性是随时间漂移。新框架上线、新攻击手法出现都会改变流量分布。如果你的数据集是按时间排列的直接train_test_split随机切分会发生时间穿越——训练集里混入了后面的攻击样本模型等于提前看到了答案。正确的做法是按时间顺序切分用前70%的时间窗口做训练后30%做测试。4.2 特征泄漏UA里藏着标签信息某些公开数据集在生成时有一个隐蔽问题攻击样本是由固定扫描器产生的它的UA是固定的sqlmap/1.0或nikto/1.0。如果特征里包含了UA的完整原文模型学到的其实是看到这个UA就是攻击而不是这个请求的语义是攻击。换一个UA绕过模型立刻失效。解决办法是要么干脆去掉UA原始字符串只保留UA特征值比如长度、是否含浏览器内核关键字要么在数据预处理时对UA做匿名化。4.3 类别不平衡引发的全零预测攻击样本占比过低时模型会倾向把所有样本预测为正常。除了前面提到的class_weight还可以用SMOTE做合成采样。但注意SMOTE对文本特征合成的效果不稳定可能会生成select select 1 2 3这种不存在的组合。另一种更稳妥的办法是调低正类的判定阈值利用的是模型输出的概率分布而不是强行造样本。from imblearn.over_sampling import SMOTE smote SMOTE(random_state42, k_neighbors3) X_resampled, y_resampled smote.fit_resample(X_train, y_train)SMOTE的k_neighbors默认是5如果某一类样本太少5近邻里可能混入大量异类。调成3会让合成样本更保守降低生成异常样本的概率。4.4 Jupyter运行环境本身的坑Jupyter做机器学习的频率一高最常见的一个报错是内核连接不上或pandas装不上。前者通常是因为Notebook服务和内核进程不在同一个Python环境比如用系统Python启动了Jupyter但pip install pandas装到了conda环境里。建议用jupyter kernelspec list检查当前内核路径确保和sys.executable一致。后者在pip install pandas时报编译错误大概率是Python版本过新旧版本pandas没有对应wheel包先升级pip再装即可。5. 把训练好的模型做成在线检测接口5.1 模型导出与特征一致性校验训练完成不等于工作完成。模型要上线至少要导出两个文件模型本体.pkl或.json格式和特征列表.json记录特征顺序。后者往往被忽略但它才是线上的老大难问题——训练和推理之间特征顺序不一致模型还能跑只是结果全是垃圾。import json import joblib joblib.dump(best_model, web_attack_model.pkl) feature_names X_train.columns.tolist() with open(feature_schema.json, w) as f: json.dump({ feature_names: feature_names, threshold: 0.3, trained_at: 2024-06-01 }, f, indent2)不需要把训练时用到的原始DataFrame也存下来。模型要的是每一列的特征值不是原始请求。线上推理时请求进来先走特征提取函数然后把特征按feature_schema.json里的顺序拼成数组。5.2 用FastAPI包装检测服务from fastapi import FastAPI, Request import joblib import json app FastAPI() model joblib.load(web_attack_model.pkl) with open(feature_schema.json) as f: schema json.load(f) FEATURE_NAMES schema[feature_names] THRESHOLD schema[threshold] app.post(/detect) async def detect(raw_request: Request): body await raw_request.text() parsed parse_http_request(body) feats extract_features(parsed) # 确保特征顺序和训练时完全一致 feature_vector [feats.get(name, 0) for name in FEATURE_NAMES] prob model.predict_proba([feature_vector])[0][1] return { score: float(prob), is_attack: bool(prob THRESHOLD), action: block if prob 0.7 else review if prob THRESHOLD else pass }接口里设了两档阈值超过0.7直接拦截0.3到0.7进人工复核队列。这样比单一阈值的方案更好落地——既避免漏报也避免把正常业务请求误杀。部署时不要用FastAPI自带的uvicorn单进程扛生产流量前面套一层Nginx反代做负载均衡这是Web服务的基础操作。5.3 验证模型上线后没有衰减的一个技巧上线后不要只看告警量。建议每天在Jupyter里用当天流量重新算一遍AUC和时间序列上的前值对比。如果AUC持续下滑超过5个百分点说明流量分布发生了变化需要增量训练。增量训练时不要全量重训用前几周的数据加当天的数据混合训练效果通常比只用当天数据好。数据累积的时间越长这个监控表的价值越大——它可以成为你论文里模型漂移分析这一章的核心素材。本文还有配套的精品资源点击获取
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进