简介一套完整的基于机器学习的Web攻击检测系统源码与配套文档面向网络安全学习者和Web开发人员用于解决XSS与SQL注入等常见攻击的自动识别问题。项目拆分为AiWaf-1和AiWaf-2两个WAF实现分别采用聚类与机器学习方法并提供了基于GRU、CNN、KNN、SVM、RF五个检测模型的完整流程涵盖数据加载、urldecode与小写预处理、Word2Vec向量化及padding补齐、模型训练预测与评估等关键环节。压缩包共66个文件以Python脚本和配置文本为主包含17个py源码、6个txt说明、3个pkl模型、3个csv数据、2个pb、2个md文档及模型文件等整体约26.61MB还附带预训练word2vec模型与目录结构清晰的说明文档。目前已有301人学习适合希望了解机器学习在Web安全领域落地方式的读者可直接参考模型构建和检测管线设计也可将代码迁移到其他Web防护场景。从数据预处理到模型评估各步骤均有对应脚本和文档支撑方便复现与二次开发。1. 机器学习做Web攻击检测规则库越堆越厚漏报误报却一起来这套方案值不值做过Web安全的人都有同感WAF规则从几百条堆到几千条SQL注入、XSS的变体还是拦不完。攻击者改一改编码、加一段注释规则就失效业务请求里偶尔带个select字样又触发一串误报。这套基于机器学习的web攻击检测系统源码文档说明走的是另一条路把HTTP请求当作文本样本让机器学习算法学习攻击流量与正常流量的分布差异靠模型判断请求是恶意还是正常。它适合三类人被规则库折磨的安全工程师想在一周内跑通机器学习项目闭环的研发以及拿Web安全当课程设计主题的学生。下文按数据准备、特征工程、模型训练、线上接入、排错验证的顺序拆解每一步都给出可复现代码和可调参数。2. 从HTTP日志到可用样本数据清洗、特征提取与标签制作拿到这套源码第一件事不是跑模型而是解决数据。web攻击检测这个方向模型算法只占一半另一半是数据质量。常见做法是把你自己的Nginx或Apache访问日志拉出来清洗、解析、打标做成训练集。别指望源码包里附带完整的生产数据——日志涉及脱敏和隐私最稳的来源是你自己的网关日志。2.1 日志解析把一行access log拆成结构化字段Nginx combined格式的一行日志里藏着IP、时间、方法、路径、查询串、状态码、UA等字段。攻击流量集中在path和query里POST攻击则在body中。我的做法是先解析出结构化字段再决定样本主体用什么。import re import pandas as pd # Nginx combined 格式示例 # 192.168.1.10 - - [10/Oct/2024:13:55:36 0800] GET /login.php?usera%27or%271%27%3D%271 HTTP/1.1 200 532 https://example.com/login Mozilla/5.0 log_pattern re.compile( r(?Pip[\d.]) - - \[(?Ptime[^\]])\] r(?Pmethod\w) (?Ppath[^?\s])\??(?Pquery[^\s]*) rHTTP/\d\.\d (?Pstatus\d) (?Psize\d) r(?Preferer[^]*) (?Pua[^]*) ) def parse_log_line(line): m log_pattern.match(line.strip()) if not m: return None return m.groupdict() records [] with open(access.log, r, encodingutf-8, errorsignore) as f: for line in f: parsed parse_log_line(line) if parsed: records.append(parsed) df pd.DataFrame(records) df.head()这段正则把path和query拆开了原因很实际query是攻击载荷的主要落脚点path是路由信息混在一起会让模型分不清结构。errorsignore是血泪经验——生产日志里总有几行编码不完整的记录直接让解析抛异常会中断整个任务。status字段先保留后面做样本过滤时有用比如只保留200、403、500把静态资源请求先用简单规则滤掉能显著降低正常样本里的噪音。如果项目里POST请求占大头access log默认不记录body。常见做法有两个一是改Nginx配置加$request_body到日志格式代价是日志体积翻倍二是开发环境里用中间件单独记录body只记录非静态接口。我一般倾向后者因为生产网关日志脱敏工作量大。2.2 特征工程长度、熵、特殊符号密度哪个特征对攻击流量最敏感机器学习模型不能直接吃字符串得先转成数值向量。这个方向的特征大体分三类统计类长度、比例、熵、结构类最长token、符号出现位置、语义类关键字命中、编码特征。这里用一套函数把最常用的特征一次算出来。import math import re from collections import Counter SQL_KEYWORDS [select, union, sleep, benchmark, updatexml, concat] XSS_KEYWORDS [script, onerror, javascript:, alert(, iframe] def extract_features(path, query): text f{path}?{query} if query else path lower text.lower() length max(len(text), 1) features {} features[path_len] len(path) features[query_len] len(query or ) features[digit_ratio] sum(c.isdigit() for c in text) / length features[letter_ratio] sum(c.isalpha() for c in text) / length # 特殊符号密度SQL/XSS 注入都依赖这些符号 features[special_ratio] sum(c in ;\()% for c in text) / length # 字符熵越高说明字符分布越随机混淆型攻击常见 counter Counter(text) entropy -sum((v / length) * math.log2(v / length) for v in counter.values() if v 0) features[char_entropy] entropy # 最长连续 token 长度 tokens re.split(r[^a-zA-Z0-9], text) features[max_token_len] max((len(t) for t in tokens), default0) # 关键字命中数注意这只是弱信号不能当规则用 features[sql_keyword] sum(k in lower for k in SQL_KEYWORDS) features[xss_keyword] sum(k in lower for k in XSS_KEYWORDS) return features字符熵是个容易被忽略但很有价值的特征。正常URL的字符分布有规律域名、路径、参数名都是可读单词混淆型攻击载荷里百分号编码、随机大小写、长数字串堆在一起熵值明显偏高。special_ratio同理SQL注入离不开单引号和分号XSS离不开尖括号这两个特征在两类攻击上都有区分度。关键字命中是我故意保留的弱特征。它单独用等于规则引擎和统计特征组合后却能让模型更快锁定攻击区域。keywords列表要按业务维护比如登录接口常用user这种参数注入特征会集中落在那里。2.3 标签制作与样本切分WAF打标省人力但必须留人工复核监督学习必须有标签。现实里没有现成的标注数据集最可靠的做法是拿现有WAF的拦截记录做初标再抽检复核。# waf_hit.list 是WAF拦截记录中的请求标识比如 request_id hit_set set() with open(waf_hit.list, encodingutf-8) as f: for line in f: hit_set.add(line.strip()) # 被WAF拦过的先标为攻击其余标为正常 df[label] df[request_id].apply(lambda x: 1 if x in hit_set else 0) # 关键按时间排序后再切分防止数据泄漏 df df.sort_values(time).reset_index(dropTrue) split_idx int(len(df) * 0.8) train_df df.iloc[:split_idx] test_df df.iloc[split_idx:]这里有个新手最容易犯的错直接用train_test_split随机切分。随机切分会让同一时段、同一攻击工具的流量同时出现在训练集和测试集里模型记住的是时间片特征而不是攻击本质测试集指标虚高。按时间切分的测试集才是真实线上表现的近似。WAF初标天然带噪音WAF自己就有误报这些误报请求会被标成1模型学到的攻击样本里混着正常请求。我的习惯是训练集和测试集各抽200条人工过一遍把明显标错的改掉或删掉。抽检脚本一行就能做train_df.sample(200).to_csv(check.csv)看完再回填。这个环节花半小时能省后面几天的排错时间。提示WAF初标必须做人工抽检否则模型会把WAF的误报习惯一并学走线上误报直接翻倍。3. 模型选型与训练TF-IDF向量化加机器学习算法为什么不直接上深度学习特征工程做完进入模型环节。标题写的是基于机器学习范围很广——决策树、随机森林、XGBoost、SVM都算深度学习严格说也是机器学习的分支。但在这个方向我的默认选择是先上TF-IDF向量化加传统机器学习算法而不是一上来就堆Transformer。3.1 选型理由样本量几千到几万时传统机器学习比深度学习更稳Web攻击检测的样本规模决定了选型。一个中等站点一天的有效请求几万条攻击流量往往不到1%清洗完能用的训练样本一般就几千到几万条。这个量级下深度学习模型的优势发挥不出来反而容易过拟合TF-IDF加随机森林或XGBoost在CPU上几分钟就能训完一组网格搜索AUC稳定在95%以上。方案适合样本量训练耗时可解释性线上推理SVM TF-IDF几千分钟级中快随机森林 TF-IDF几千到几万分钟级高快XGBoost TF-IDF几万以上分钟级中快LSTM/Transformer十万以上小时级低一般可解释性在这里不是加分项是刚需。安全场景每次误报都要能解释给业务方听为什么拦了这个请求。随机森林能给出特征重要性告诉你决策依据是熵值过高还是关键字命中这比深度学习黑匣子好交代得多。只有当样本量真的到了十万级、传统模型出现明显瓶颈时我才会考虑上序列模型。3.2 训练脚本Pipeline加GridSearchCV一次跑通把向量化和分类器串成一个Pipeline是这里最省心的组织方式。它保证预处理和模型一起保存、一起加载线上推理不会出现训练时做了归一化、推理时忘了做这种事故。from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.ensemble import RandomForestClassifier from sklearn.pipeline import Pipeline from sklearn.model_selection import GridSearchCV # 样本主体path query 拼成文本 train_text train_df[path] ? train_df[query].fillna() test_text test_df[path] ? test_df[query].fillna() pipeline Pipeline([ (tfidf, TfidfVectorizer( analyzerchar_wb, # 按字符切能捕获被符号隔开的片段 ngram_range(2, 4), # 2到4元字符片段 max_features50000, # 限制词典维度 sublinear_tfTrue # 对高频词做对数压缩 )), (clf, RandomForestClassifier( n_estimators200, class_weightbalanced, # 缓解攻击样本占比低的问题 n_jobs-1 )) ]) param_grid { tfidf__max_features: [30000, 50000], clf__max_depth: [None, 20], } search GridSearchCV(pipeline, param_grid, cv5, scoringroc_auc, n_jobs-1) search.fit(train_text, train_df[label]) from sklearn.metrics import classification_report y_pred search.predict(test_text) print(classification_report(test_df[label], y_pred))TfidfVectorizer这里有个关键参数analyzerchar_wb。按单词切分对攻击样本不友好因为SQL注入的载荷常常是11、sleep(5)这种被符号包围的片段单词切分会把它们拆散char_wb按字符窗口滑配合ngram_range(2,4)能稳定捕获or、1、onerr这类攻击指纹。代价是特征维度变高所以max_features必须设上限否则几万样本能撑出上百万维的矩阵直接内存爆炸。sublinear_tfTrue是个容易忽略的小参数。它把词频做对数压缩避免select这种词在某个攻击样本里出现几十次时权重被过度放大。class_weightbalanced处理样本不平衡让模型对少数类攻击更敏感。GridSearchCV的scoring选roc_auc而不是accuracy理由在下一节说。3.3 类别不平衡与阈值准确率是陷阱召回率和误报率才是KPI攻击检测的标签极度不平衡正常请求可能占95%以上。这时候accuracy是陷阱模型全预测正常都能拿95%的准确率看起来漂亮实际一个攻击都拦不住。正确的评估指标是精确率和召回率。import numpy as np proba search.predict_proba(test_text)[:, 1] best_th, best_f1 0.0, 0.0 for th in np.arange(0.3, 0.9, 0.05): pred (proba th).astype(int) tp ((pred 1) (test_df[label] 1)).sum() fp ((pred 1) (test_df[label] 0)).sum() fn ((pred 0) (test_df[label] 1)).sum() precision tp / max(tp fp, 1) recall tp / max(tp fn, 1) f1 2 * precision * recall / max(precision recall, 1e-9) print(fth{th:.2f} 精确率{precision:.3f} 召回率{recall:.3f} F1{f1:.3f}) if f1 best_f1: best_f1, best_th f1, th print(f推荐阈值: {best_th:.2f})默认阈值0.5在不平衡数据上不是最优解。我线上一般把阈值调到0.65到0.8之间宁可牺牲一点召回也要把误报压下去。原因很现实误报会打断真实用户操作一次就足以让业务方对系统失去信任漏报至少还能靠规则引擎兜底。阈值扫描这个脚本我几乎每个项目都跑一遍它把凭感觉定阈值变成了一个可回溯的过程。4. 从离线到在线模型落盘、检测接口与日志回放模型在测试集上指标好看只是第一步真正决定系统价值的是能不能接进线上链路。常见落地方案有三种旁路日志检测、反向代理同步过滤、网关插件。从这套源码的通用性出发最稳的是把模型包装成一个HTTP检测接口让Nginx或业务网关调用。4.1 模型持久化与检测函数训练完先落盘别关终端训练好的Pipeline对象只存在于内存里关掉Jupyter或终端就没了。用joblib把整个Pipeline序列化到磁盘是标准做法。import joblib # 完整Pipeline含TF-IDF和分类器整体保存 joblib.dump(search.best_estimator_, waf_model.pkl) # 推理阶段加载一次进程内复用 model joblib.load(waf_model.pkl) def detect(path, query, body): text f{path}?{query} {body}.strip() proba model.predict_proba([text])[0, 1] label attack if proba 0.7 else normal return label, round(proba, 4)保存整个Pipeline而不是只保存分类器是为了让向量化参数和模型永远保持一致。如果分开保存训练时TF-IDF词典和线上加载的词典出现一丁点偏差推理结果就会整体偏移。detect函数里把path、query、body拼成一条文本格式要和训练时的样本主体一致——这是最容易踩的坑训练时用了query推理时忘了拼特征空间对不上机器学习模型直接降智。4.2 用FastAPI封装检测服务给网关留一个过滤接口检测函数写好后外面包一层HTTP服务。FastAPI轻量、异步、自带交互文档比Flask更适合这种高并发小任务。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class DetectRequest(BaseModel): method: str path: str query: str body: str app.post(/api/detect) async def api_detect(req: DetectRequest): label, proba detect(req.path, req.query, req.body) return { label: label, score: proba, action: block if label attack else pass }接口返回score而不是直接封禁是个重要的设计取舍。拦截动作应该由网关决定Nginx收到actionblock时返回403业务侧也能根据score做二次审核。这样模型误判时业务方还有人工复核的余地而不是直接被拒。性能上随机森林单条推理在毫秒级单机处理每秒几百个检测请求没什么压力高QPS场景必须复用进程内加载的model对象每个请求都load一次pkl会让延迟爆炸。4.3 日志回放验证上线前用历史流量跑一遍而不是直接切生产模型改动上线前我最喜欢做的一件事是回放。拿最近一周的完整日志按时间顺序逐条喂给检测函数观察命中率、延迟和内存变化。import time hit 0 total 0 latency [] for _, row in test_df.iterrows(): start time.time() label, proba detect(row[path], row[query]) latency.append(time.time() - start) total 1 if label attack: hit 1 print(f回放样本数{total} 攻击命中数{hit} 命中率{hit / max(total, 1):.3f}) latency_sorted sorted(latency) print(f平均延迟{sum(latency) / len(latency) * 1000:.2f}ms fp95{latency_sorted[int(len(latency_sorted) * 0.95)] * 1000:.2f}ms)回放脚本同时是回归测试工具。每次改特征、调阈值先回放再决定要不要上线能防止改了一个参数另一个攻击类型全漏了的暗坑。回放时把预测为attack的样本单独落盘人工过一遍再和WAF原记录比对这是验证误报率最直接的办法。回放不是万能的前提是历史日志的流量分布和未来相似但作为上线前的最后一道闸门性价比极高。5. 避坑与常见问题排查真实流量里最常遇到的五个坑这套方案我在不同环境里部署过几次线上流量和干净数据集完全是两回事。下面五条是高频踩坑记录每条按现象、原因、解决的顺序写希望你能少走弯路。5.1 现象URL里的中文参数全部被判成攻击现象中文站点的搜索请求q春节促销回放时几乎全命中attack。原因日志解析时编码不统一部分请求是UTF-8明文部分是%E6%98%A5%E8%8A%82这种百分号编码。百分号编码让entropy和special_ratio两个特征同时飙升模型把高编码密度学成了攻击特征。解决解析阶段统一做一次URL解码解码后再提取特征。from urllib.parse import unquote # 统一解码注意要处理解码抛异常的情况 df[query] df[query].apply(lambda x: unquote(x) if x else x) # 解码后把特征函数重跑一遍解码失败的样本保留原始形式并加一个is_percent_encoded的布尔特征让模型自己学这个特征和攻击的关系而不是让编码密度污染熵特征。5.2 现象训练集准确率97%线上误报率爆炸现象离线评估AUC 0.99切到线上第一天误报率超过20%正常请求大量被拦。原因两件事叠加。一是随机切分造成数据泄漏同一时段的攻击流量同时落在训练集和测试集二是WAF初标里混着误报模型把正常请求的长尾形态学成了攻击。解决改按时间切分测试集严格晚于训练集初标后做人工抽检剔除明显误标样本阈值按3.3的脚本重扫线上先以低误报配置运行一周观察稳定后再逐步提高召回。5.3 现象SQL注入加个--注释和大小写就绕过现象线上出现漏报攻击者把union select改成UnIoN/**/SeLeCt模型判normal。原因TF-IDF按字符n-gram切分大小写变体本身能被统一小写缓解一部分但注释符号改变了字符序列2到4元片段覆盖不到新的组合关键字特征在/**/面前也失效。解决特征提取前做两件事一是统一小写二是把/**/、/*!、--这类注释符号替换成空格让模型看到的序列和攻击语义对齐。这类变异样本要主动构造加入训练集常见做法是拿已知攻击样本做随机变异扩充让模型见过变体而不是只会背模板。5.4 现象模型上线三个月后召回率明显下滑现象监控发现攻击命中率从95%掉到70%业务流量没有明显变化。原因概念漂移。新攻击手法出现、业务页面改版、参数命名变化都会让旧模型的特征分布失真。攻击检测模型不是一次训练终身使用的东西。解决建立两个监控信号。一是预测分值分布正常流量的平均score持续上涨说明分布漂移二是定期用新增的WAF拦截记录做增量训练我一般按周跑一次用近一个月的日志重训保留特征配置不变。模型文件用时间戳命名waf_model_20250101.pkl回滚就是换一个文件的事这是最大的一颗后悔药。阈值调优在线上环境下有时候像玄学但按周回放能把这些不确定因素拉回可控范围。注意模型文件务必带版本号管理否则一次误操作覆盖掉稳定版本想回退都没有后悔药。5.5 现象特征维度爆炸8GB内存直接OOM现象GridSearchCV跑了一会儿内存飙到90%进程被杀。原因char_wb按字符切ngram_range(2,5)加上max_features100000几万条样本能把特征矩阵撑到上亿个非零元素网格搜索并行时每个线程复制一份数据。解决max_features先压到30000ngram_range用(2,4)先跑通再慢慢放开n_jobs从-1降到4避免并行副本撑爆内存如果样本真的很大改用HashingVectorizer做向量化它不维护词典、维度固定且上限可控代价是特征不可解释适合特征数量是瓶颈的场景。6. 效果验证的最后一公里混淆矩阵、回放报告与A/B对比模型上线的效果验证不能只看一行准确率。我每次改版都会先跑一遍完整评估混淆矩阵加classification_report再配合一周的回放报告存档。from sklearn.metrics import confusion_matrix, classification_report y_true test_df[label] y_score search.predict_proba(test_text)[:, 1] y_pred (y_score 0.7).astype(int) print(confusion_matrix(y_true, y_pred)) print(classification_report(y_true, y_pred, target_names[normal, attack]))混淆矩阵的四个象限对应四种业务后果右上角的假正例是误报直接打扰用户左下角的假负例是漏报放走攻击。安全场景里这两类错误的成本不一样我一般把误报成本算成漏报的3到5倍所以阈值偏向保守。classification_report里的precision就是拦下来的里面有多少真攻击recall是真攻击里拦下来多少这两个数比准确率诚实得多。A/B对比是上线前的最后一关。拿最近一周的新日志同一份数据分别跑规则WAF和这套ML模型人工标注抽样500条做基准对比精确率、召回率、误报率。一个典型的对比结果长这样指标规则WAFML模型精确率0.820.91召回率0.650.88误报率3.2%1.1%规则引擎召回率低是常态因为变体太多ML的价值不是完全替代规则而是把规则漏掉的那部分补上。我的习惯是双跑一段时间规则WAF在前面拦ML给每个请求打分会话日志每周对score在0.5到0.8之间的灰区样本做人工复核这些样本就是下一次增量训练最值钱的素材。这个方向值不值得投入我的判断标准很简单如果你们每周要花一个人天去补WAF规则或者线上已经出现过绕过事件那就值得做。机器学习流程跑通之后改特征、换模型、调阈值都是一两天内能完成的事真正的成本在数据维护和标注闭环上。我现在的习惯是每次调参都留一份回放报告存档阈值、特征版本、训练集时间范围一起记。模型不是一锤子买卖数据、特征、阈值要一起迭代线上验证才是真正的考试。希望帮到你。本文还有配套的精品资源点击获取
