1. 项目概述为什么我们需要自己的Web Fuzzer如果你做过Web安全测试或者渗透测试Burp Suite的Intruder模块肯定是你工具箱里的常客。它功能强大界面直观点几下鼠标就能发起海量请求对参数进行暴力猜解或模糊测试。但不知道你有没有遇到过这样的场景面对一个需要复杂逻辑判断的输入点比如需要根据上一个请求的响应动态生成下一个Payload或者需要对一个自定义的加密参数进行遍历时Intruder那些预设的Payload类型和攻击模式突然就变得不那么好用了。你开始怀念写代码的那种灵活和精确控制的感觉。这就是为什么作为一个有追求的测试人员或安全研究员你不能只会用现成的工具还得知道怎么自己造“轮子”——哪怕这个轮子一开始很简单。用Python写一个简单的Web Fuzzer正是为了填补这个空白。它不是为了替代Burp SuiteBurp在拦截、爬虫、漏洞扫描集成方面的地位无可撼动。我们自己做Fuzzer的目标是获得极致的灵活性和可编程性。你可以完全控制HTTP请求的每一个字节可以轻松地集成复杂的Payload生成算法可以方便地与你的其他自动化脚本联动。更重要的是这个过程会让你对HTTP协议、Web应用交互以及模糊测试的核心原理有更深刻的理解这是单纯使用图形化工具很难获得的。今天我就带你从零开始用大约一百行Python代码构建一个属于你自己的、功能完整的Web Fuzzer核心引擎。2. 核心思路与架构设计在动手写代码之前我们必须先想清楚一个Fuzzer到底要做什么。抛开那些高级特性一个最基础的Web Fuzzer的核心工作流程可以抽象为三个步骤读取或生成测试载荷Payload - 构建并发送HTTP请求 - 接收并分析响应。我们的设计也将紧紧围绕这个流程展开。2.1 为什么选择Requests库和Concurrent Futures首先我们需要一个可靠的HTTP客户端库。Python世界里requests库是当之无愧的王者。它语法优雅几乎成为了Python发送HTTP请求的事实标准。相比于原生的urllibrequests在易用性上有着压倒性的优势能让我们更专注于Fuzzing逻辑本身。其次Fuzzing往往意味着要发送成千上万个请求效率是关键。如果用一个for循环串行发送测试一个简单的4位数字验证码0000-9999就要花上将近3个小时假设每个请求0.1秒。这显然是不可接受的。因此我们必须引入并发。Python标准库中的concurrent.futures模块提供了高级的线程池/进程池接口是我们实现并发的首选。这里我选择ThreadPoolExecutor线程池因为对于I/O密集型任务网络请求等待响应使用多线程可以在等待时切换执行其他任务从而大幅提升效率且线程间共享内存方便我们汇总结果。如果选择多进程则更适合CPU密集型的Payload生成计算。我们的Fuzzer架构将包含以下几个核心类或函数Payload生成器负责产生需要测试的字符串或数据。最简单的就是从一个文本文件中逐行读取。请求构造器负责将原始HTTP请求或模板与当前Payload结合生成一个完整的、待发送的requests.Request对象。请求引擎负责管理线程池调度请求的发送并处理超时、重试等网络问题。响应分析器负责检查服务器返回的响应根据状态码、响应体长度、关键词匹配等规则判断该Payload是否可能导致了异常行为如服务器错误、内容差异等。结果管理器负责收集、去重、保存和展示测试结果。下面我们就进入具体的实现环节。3. 从零开始构建核心模块我们将采用自底向上的方式先实现最基础的模块最后将它们组装起来。请确保你的Python环境是3.6以上并使用pip install requests安装好依赖库。3.1 基础配置与Payload加载任何Fuzzer都需要一个起点。我们首先定义一个配置类用来存放目标URL、请求头、Cookie等静态信息并实现一个简单的Payload加载器。import requests from concurrent.futures import ThreadPoolExecutor, as_completed import time import argparse from urllib.parse import urlparse, parse_qs, urlencode, urlunparse import json class FuzzerConfig: Fuzzer配置类存储目标请求的模板信息 def __init__(self, url, methodGET, headersNone, dataNone, cookiesNone, paramsNone): self.url url self.method method.upper() self.headers headers if headers else {} self.data data self.cookies cookies if cookies else {} self.params params if params else {} # 确保有User-Agent避免被一些基础WAF直接拦截 if User-Agent not in self.headers: self.headers[User-Agent] Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 def load_payloads_from_file(filename): 从文本文件加载Payload每行一个 try: with open(filename, r, encodingutf-8, errorsignore) as f: # 去除每行首尾空白字符并过滤掉空行 return [line.strip() for line in f if line.strip()] except FileNotFoundError: print(f[错误] Payload文件未找到: {filename}) return [] except Exception as e: print(f[错误] 读取Payload文件时出错: {e}) return []注意errorsignore参数是为了防止Payload文件中存在非UTF-8编码字符导致程序崩溃。在安全测试中我们有时会使用包含特殊二进制字符的Payload这个设置能提高程序的健壮性。3.2 智能请求构造处理GET、POST与JSON一个健壮的Fuzzer必须能灵活处理不同类型的请求。我们的请求构造器需要识别原始请求中的参数位置URL查询参数、POST表单数据、JSON Body并将Payload精准地注入进去。class RequestBuilder: 根据配置和Payload构造实际的请求对象 staticmethod def build_request(config, payload, position, param_name): 构造请求 :param config: FuzzerConfig对象 :param payload: 当前测试的载荷字符串 :param position: 参数位置如 url_params, post_data, json :param param_name: 要替换的参数名 :return: 构造好的 (method, url, kwargs) 元组供requests.request使用 url config.url headers dict(config.headers) # 创建副本避免修改原配置 data config.data params dict(config.params) if config.params else {} json_data None # 根据参数位置将Payload注入到正确的地方 if position url_params: # 注入到URL查询参数中 params[param_name] payload elif position post_data: # 注入到POST表单数据中 if isinstance(data, dict): data dict(data) data[param_name] payload else: # 如果data是字符串我们尝试将其解析为字典再替换。这里简化处理。 # 更复杂的场景可以解析application/x-www-form-urlencoded格式字符串 print(f[警告] POST数据为非字典类型直接替换可能不准。原始数据: {data}) data {param_name: payload} elif position json: # 注入到JSON Body中 try: if isinstance(data, dict): json_data dict(data) elif isinstance(data, str): json_data json.loads(data) else: json_data {} json_data[param_name] payload data None # 使用json参数后不应再使用data参数 except json.JSONDecodeError: print(f[错误] 无法解析JSON数据: {data}) return None else: print(f[错误] 不支持的参数位置: {position}) return None # 准备requests.request的参数 kwargs { headers: headers, cookies: config.cookies, timeout: 10, # 设置一个合理的超时时间 allow_redirects: False, # 禁止自动重定向方便我们观察中间响应 } if params: kwargs[params] params if data is not None: kwargs[data] data if json_data is not None: kwargs[json] json_data return (config.method, url, kwargs)这个RequestBuilder类是我们的核心之一。它通过position和param_name精确控制了Payload的注入点。例如当positionjson且param_nameusername时它会尝试将配置中的data解析为JSON字典然后将其中的username字段的值替换为当前Payload。3.3 并发引擎与响应分析有了请求模板接下来就是发送请求并分析结果。我们将发送器和分析器放在一个FuzzingEngine类中。class FuzzingEngine: Fuzzing引擎负责并发发送请求和分析响应 def __init__(self, max_workers20): self.max_workers max_workers # 并发线程数 self.results [] # 存储测试结果 def send_request(self, method, url, kwargs): 发送单个请求并返回响应对象和异常信息 try: resp requests.request(method, url, **kwargs) return resp, None except requests.exceptions.Timeout: return None, Timeout except requests.exceptions.ConnectionError: return None, ConnectionError except Exception as e: return None, str(e) def analyze_response(self, resp, payload, original_lengthNone, original_statusNone): 分析响应判断是否存在异常 这里实现几个简单的启发式规则 1. 状态码异常如5xx服务器错误 2. 响应体长度与基准长度差异巨大 3. 响应体中包含特定的错误关键词 :return: (is_interesting, reason) if resp is None: return False, No Response is_interesting False reasons [] # 规则1: 检查状态码 if 500 resp.status_code 600: is_interesting True reasons.append(f状态码 {resp.status_code}) # 规则2: 检查响应长度需要基准长度 if original_length is not None: current_length len(resp.content) # 如果长度差异超过20%则认为有趣 if abs(current_length - original_length) / original_length 0.2: is_interesting True reasons.append(f长度变化 {original_length}-{current_length}) # 规则3: 检查错误关键词 error_keywords [error, exception, sql, syntax, warning, undefined] resp_text resp.text.lower() for keyword in error_keywords: if keyword in resp_text: is_interesting True reasons.append(f包含关键词 {keyword}) break # 找到一个即可 reason ; .join(reasons) if reasons else Normal return is_interesting, reason def run(self, config, payloads, position, param_name): 运行Fuzzing测试 :param config: 配置 :param payloads: Payload列表 :param position: 参数位置 :param param_name: 参数名 print(f[*] 开始Fuzzing: {config.url}) print(f[*] 目标参数: {param_name} {position}) print(f[*] Payload数量: {len(payloads)}) print(f[*] 并发数: {self.max_workers}) print(- * 50) # 首先发送一个基准请求使用原始参数值或空值 # 这里我们简单地将param_name的值设为空字符串来获取基准响应 baseline_payload baseline_request RequestBuilder.build_request(config, baseline_payload, position, param_name) if not baseline_request: print([错误] 构建基准请求失败) return baseline_resp, error self.send_request(*baseline_request) if error: print(f[错误] 基准请求失败: {error}) baseline_length 0 baseline_status 0 else: baseline_length len(baseline_resp.content) if baseline_resp else 0 baseline_status baseline_resp.status_code if baseline_resp else 0 print(f[基准] 状态码: {baseline_status}, 响应长度: {baseline_length}) # 使用线程池并发发送请求 with ThreadPoolExecutor(max_workersself.max_workers) as executor: future_to_payload {} for payload in payloads: req_tuple RequestBuilder.build_request(config, payload, position, param_name) if req_tuple: future executor.submit(self.send_request, *req_tuple) future_to_payload[future] payload else: print(f[跳过] 无法为Payload构建请求: {payload}) # 处理完成的任务 for future in as_completed(future_to_payload): payload future_to_payload[future] resp, error future.result() status resp.status_code if resp else N/A length len(resp.content) if resp else 0 # 分析响应 is_interesting, reason self.analyze_response(resp, payload, baseline_length, baseline_status) # 记录结果 result { payload: payload, status: status, length: length, error: error, interesting: is_interesting, reason: reason } self.results.append(result) # 实时输出有趣的结果 if is_interesting: print(f[!] 发现异常: Payload{payload}, 状态码{status}, 长度{length}, 原因{reason}) # 也可以选择输出所有结果但可能会刷屏 # else: # print(f[.] 正常: Payload{payload}, 状态码{status}, 长度{length}) print(- * 50) print(f[*] Fuzzing完成。总请求数: {len(self.results)}) interesting_count sum(1 for r in self.results if r[interesting]) print(f[*] 发现异常结果: {interesting_count} 个)这个引擎类做了几件关键事情获取基准响应在正式Fuzzing前先发送一个“正常”请求参数值为空记录其响应长度和状态码作为后续判断的基准。这是非常实用的一步能有效过滤掉大量正常但长度各异的响应。并发控制使用ThreadPoolExecutor管理线程池将所有请求任务提交后通过as_completed获取完成的任务实现高效的并发。实时反馈在控制台实时打印出被标记为“有趣”的异常结果方便测试者即时观察。简单的分析规则实现了状态码、长度偏差、关键词匹配三种最基本的异常检测规则。你可以根据需要轻松扩展这个analyze_response方法。3.4 结果输出与主程序入口最后我们需要一个方式来保存结果并提供一个友好的命令行接口。def save_results(results, filenamefuzzing_results.txt): 将结果保存到文件 with open(filename, w, encodingutf-8) as f: f.write(Payload\tStatus\tLength\tError\tInteresting\tReason\n) for r in results: f.write(f{r[payload]}\t{r[status]}\t{r[length]}\t{r[error] or }\t{r[interesting]}\t{r[reason]}\n) print(f[*] 结果已保存至: {filename}) def main(): parser argparse.ArgumentParser(description简单的Python Web Fuzzer) parser.add_argument(-u, --url, requiredTrue, help目标URL) parser.add_argument(-w, --wordlist, requiredTrue, helpPayload字典文件路径) parser.add_argument(-p, --param, requiredTrue, help要Fuzz的参数名) parser.add_argument(--position, choices[url_params, post_data, json], defaulturl_params, help参数所在位置 (url_params, post_data, json)默认url_params) parser.add_argument(-m, --method, defaultGET, helpHTTP方法 (GET, POST)默认GET) parser.add_argument(--data, helpPOST数据 (e.g., useradminpasstest) 或JSON字符串) parser.add_argument(--headers, help自定义请求头 (e.g., Content-Type: application/json)多个用分号隔开) parser.add_argument(--workers, typeint, default20, help并发线程数默认20) parser.add_argument(-o, --output, help结果输出文件) args parser.parse_args() # 解析Headers headers {} if args.headers: for h in args.headers.split(;): if : in h: key, value h.split(:, 1) headers[key.strip()] value.strip() # 解析POST数据 data None if args.data: # 简单判断是否为JSON if args.data.strip().startswith({): try: data json.loads(args.data) except: # 如果不是合法JSON则视为字符串表单数据 data args.data else: # 尝试解析为字典表单格式 try: # 这里简化处理实际应使用urllib.parse.parse_qs data {} for item in args.data.split(): if in item: k, v item.split(, 1) data[k] v except: data args.data # 创建配置 config FuzzerConfig( urlargs.url, methodargs.method, headersheaders, datadata, ) # 加载Payload payloads load_payloads_from_file(args.wordlist) if not payloads: print([错误] 未加载到有效Payload程序退出。) return # 创建引擎并运行 engine FuzzingEngine(max_workersargs.workers) engine.run(config, payloads, args.position, args.param) # 保存结果 output_file args.output or fuzzing_results.txt save_results(engine.results, output_file) if __name__ __main__: main()至此一个功能完整的命令行Web Fuzzer就完成了。你可以通过命令行参数灵活指定目标、参数、位置、并发数等。4. 实战演练用我们的Fuzzer测试一个靶场理论说得再多不如实际跑一遍。我们用一个简单的靶场来演示。假设我们有一个登录接口http://test.local/login.php它使用POST方法表单数据为usernameadminpassword123。我们怀疑其username参数存在SQL注入漏洞。第一步准备Payload字典创建一个名为sql_payloads.txt的文件内容如下admin -- admin or 11 admin or 11 -- admin UNION SELECT NULL -- 1 AND SLEEP(5) --第二步运行Fuzzer在命令行中执行python web_fuzzer.py -u http://test.local/login.php -w sql_payloads.txt -p username --position post_data -m POST --data usernameadminpassword123 --workers 10程序输出示例[*] 开始Fuzzing: http://test.local/login.php [*] 目标参数: username post_data [*] Payload数量: 5 [*] 并发数: 10 -------------------------------------------------- [基准] 状态码: 200, 响应长度: 1520 [!] 发现异常: Payloadadmin --, 状态码200, 长度1805, 原因长度变化 1520-1805 [!] 发现异常: Payloadadmin or 11, 状态码500, 长度120, 原因状态码 500; 包含关键词 error [*] Fuzzing完成。总请求数: 5 [*] 发现异常结果: 2 个 [*] 结果已保存至: fuzzing_results.txt从输出可以看到Payloadadmin --导致了响应长度显著变化而admin or 11直接引发了服务器500错误并返回了包含“error”的页面。这两个结果都强烈提示可能存在SQL注入漏洞需要进一步手工验证。5. 进阶技巧与避坑指南一个能跑的Fuzzer只是开始要让它在真实复杂环境中稳定、高效、隐蔽地工作还需要很多技巧。5.1 速率控制与随机延迟无限制地狂发请求很容易触发目标的WAFWeb应用防火墙或速率限制导致IP被封锁。因此必须引入速率控制。import random import time class PoliteFuzzingEngine(FuzzingEngine): 有礼貌的Fuzzing引擎增加随机延迟和速率限制 def __init__(self, max_workers20, requests_per_second5, jitter0.5): super().__init__(max_workers) self.requests_per_second requests_per_second self.jitter jitter # 随机抖动系数避免过于规律的请求 self.last_request_time 0 def send_request(self, method, url, kwargs): 重写发送方法加入延迟控制 # 计算距离上次请求应间隔的时间 interval 1.0 / self.requests_per_second # 加入随机抖动使请求间隔不那么规律 jittered_interval interval * (1 random.uniform(-self.jitter, self.jitter)) elapsed time.time() - self.last_request_time if elapsed jittered_interval: time.sleep(jittered_interval - elapsed) self.last_request_time time.time() return super().send_request(method, url, kwargs)实操心得对于不同的目标速率限制策略应不同。对自家测试环境可以快一些如10-20 RPS对公网未知目标则应保守如2-5 RPS。jitter参数非常重要它能有效避免基于固定时间间隔的简单防护规则。5.2 处理Cookie、Session与动态Token很多Web应用依赖Session或CSRF Token。我们的Fuzzer需要能处理这些动态值。一种常见的模式是“先请求再提取后注入”。import re def extract_csrf_token(html_text): 从HTML中提取CSRF Token的简单示例 # 查找类似 input typehidden namecsrf_token valueabc123 pattern rinput[^]*namecsrf_token[^]*value([^]) match re.search(pattern, html_text, re.IGNORECASE) return match.group(1) if match else None # 在主逻辑中可以先发送一个GET请求获取登录页面提取Token session requests.Session() login_page_resp session.get(login_url) csrf_token extract_csrf_token(login_page_resp.text) # 然后将session和token用于后续的Fuzzing配置 config FuzzerConfig(urllogin_action_url, methodPOST, data{csrf_token: csrf_token, username: FUZZ, password: test}) # 注意这里的session.cookies会被自动用于后续请求5.3 更智能的响应分析基础的响应分析规则漏报和误报率都很高。我们可以引入更复杂的检测逻辑响应时间差异检测是否存在时间盲注。记录每个请求的响应时间如果某个Payload的响应时间显著高于基准例如超过2秒则标记为异常。内容相似度使用difflib库计算响应体与基准响应体的相似度。低相似度可能意味着错误信息或不同的页面内容。正则表达式匹配针对特定漏洞定义正则表达式。例如对于SQL注入可以匹配You have an error in your SQL syntax对于XSS可以匹配scriptalert等。状态码序列不仅看最终状态码如果配置了allow_redirectsTrue可以检查重定向链异常的跳转如跳转到错误页、登录页也是重要信号。5.4 常见问题与排查技巧在实际使用中你肯定会遇到各种问题。这里记录几个我踩过的坑和解决方法问题1程序突然卡住不再输出结果。可能原因线程池中的某个任务发生了未捕获的异常或者遇到了网络黑洞请求无限挂起。排查在send_request函数内部用try...except捕获所有异常并返回错误信息。确保为requests.request设置合理的timeout参数如10秒。也可以考虑使用requests的Session对象并配置适配器设置整个连接池的超时。问题2大量请求返回连接错误或超时。可能原因并发数(max_workers)设置过高把目标服务器或自己的网络打满了。解决降低并发数比如从50降到10。使用上面提到的PoliteFuzzingEngine增加请求间隔。检查本地防火墙或代理设置。问题3Payload似乎没有成功替换到请求中。排查在RequestBuilder.build_request方法中将最终构造好的kwargs字典打印出来。确认params、data或json字段的值是否符合预期。特别注意当method为GET时data和json参数是无效的参数应放在params中。问题4如何Fuzz多个参数方案目前的程序是单参数Fuzzer。要测试多个参数最简单的方法是写一个外层脚本循环调用这个Fuzzer每次更换-p和--position参数。更优雅的方式是修改主程序逻辑支持从配置文件中读取多个参数位置和名称并进行排列组合或笛卡尔积式的测试但这会显著增加请求量需要谨慎控制。问题5Payload文件很大内存占用高。优化不要一次性用load_payloads_from_file把所有Payload读入内存。可以将其改造成一个生成器函数每次yield一行。然后修改FuzzingEngine.run方法使其能接受一个可迭代的Payload对象并分批提交给线程池。写一个自己的Fuzzer就像学游泳时第一次离开游泳圈。一开始可能会手忙脚乱但一旦掌握了平衡你就能游向Burp Suite等工具无法轻易触及的深水区。这个简单的脚本只是一个起点你可以根据自己的需求为它添加代理支持、报告生成、分布式测试、甚至机器学习驱动的Payload生成等高级功能。最重要的是你拥有了完全的控制权和无限的可扩展性。下次当你觉得Burp Suite的Intruder不够用时不妨打开你的代码编辑器自己动手丰衣足食。
