Mitmproxy进阶玩法除了抓包我这样用它做自动化测试和数据Mock当你还在用Mitmproxy简单地抓包调试时一些技术团队已经将它变成了研发流程中的瑞士军刀。想象一下凌晨三点的CI/CD流水线自动触发了数百个接口测试用例Mitmproxy实时修改返回数据模拟服务器崩溃场景前端开发者在后端API尚未完成时直接拦截请求返回本地设计好的JSON数据线上真实流量被精准引流到测试环境进行压力测试——这些场景正在重新定义我们对代理工具的认知。1. 动态流量改写让自动化测试拥有超能力传统自动化测试最大的瓶颈在于难以模拟真实世界的异常场景。通过Mitmproxy的response钩子我们可以像手术刀般精准修改流量内容。下面这段代码展示了如何随机注入错误状态码from mitmproxy import http import random class ChaosInjector: def response(self, flow: http.HTTPFlow): if api/v1/payments in flow.request.url: # 10%概率触发异常模拟 if random.random() 0.1: flow.response.status_code 503 flow.response.content b{error: service_unavailable} addons [ChaosInjector()]典型应用场景对比表测试类型传统方法Mitmproxy方案优势对比超时测试修改服务端代码动态添加延迟无需部署环境错误码覆盖准备多套测试数据实时修改状态码用例维护成本降低70%数据边界测试人工构造异常参数自动篡改响应字段覆盖率达到100%提示在Kubernetes环境中可以将Mitmproxy部署为Sidecar容器通过--set upstream_certfalse参数避免证书验证问题。2. 智能Mock系统前后端并行的加速器现代研发流程中前后端并行开发时最耗时的往往是接口联调。我们构建了一套基于文件系统的智能Mock方案import json from pathlib import Path class ApiMock: def __init__(self): self.mock_dir Path(__file__).parent / mock_data def request(self, flow: http.HTTPFlow): mock_file self.mock_dir / f{flow.request.path.replace(/, _)}.json if mock_file.exists(): flow.response http.Response.make( 200, mock_file.read_bytes(), {Content-Type: application/json} ) addons [ApiMock()]文件结构示例mock_data/ ├── api_v1_users.json ├── api_v1_products.json └── api_v2_orders.json这套系统在实际项目中带来了三个显著改进前端启动时间从平均2小时缩短到5分钟接口变更时只需更新对应JSON文件支持历史版本快速回滚测试3. 流量录制回放真实场景的完美复刻线上问题排查最可靠的方式就是复现真实流量。我们开发了流量归档系统关键代码片段如下import time from datetime import datetime class TrafficRecorder: def __init__(self): self.archive_dir Path(ftraffic_{datetime.now().strftime(%Y%m%d)}) self.archive_dir.mkdir(exist_okTrue) def response(self, flow: http.HTTPFlow): if flow.request.method POST: timestamp int(time.time() * 1000) with open(self.archive_dir / f{timestamp}.har, wb) as f: f.write(json.dumps({ request: { url: flow.request.url, headers: dict(flow.request.headers), content: flow.request.content.decode() }, response: { status: flow.response.status_code, content: flow.response.content.decode() } }).encode()) addons [TrafficRecorder()]回放流程优化技巧使用mitmdump -S archive.har直接回放历史流量结合--ignore-hosts参数过滤非目标域名通过--set flow_detail0提升回放性能4. 安全测试实战加解密接口的自动化方案面对加密接口传统测试工具往往束手无策。我们通过Mitmproxy搭建了透明的加解密中间层from Crypto.Cipher import AES import base64 class CryptoMiddleware: def __init__(self): self.key byour_32byte_secret_key def request(self, flow: http.HTTPFlow): if flow.request.content: cipher AES.new(self.key, AES.MODE_EAX) nonce cipher.nonce ciphertext, tag cipher.encrypt_and_digest(flow.request.content) flow.request.content base64.b64encode(nonce tag ciphertext) def response(self, flow: http.HTTPFlow): if flow.response.content: data base64.b64decode(flow.response.content) nonce, tag, ciphertext data[:16], data[16:32], data[32:] cipher AES.new(self.key, AES.MODE_EAX, noncenonce) flow.response.content cipher.decrypt_and_verify(ciphertext, tag) addons [CryptoMiddleware()]加密接口测试的最佳实践将加解密逻辑封装为独立Python模块通过环境变量管理密钥等敏感信息在测试报告中保留原始加密数据和解密后内容5. 性能监控与瓶颈定位Mitmproxy的流量分析能力可以转化为实时性能监控工具。以下代码实现了API耗时统计import statistics from collections import defaultdict class PerformanceMonitor: def __init__(self): self.stats defaultdict(list) def response(self, flow: http.HTTPFlow): latency flow.response.timestamp_end - flow.request.timestamp_start self.stats[flow.request.path].append(latency) if len(self.stats[flow.request.path]) % 10 0: ctx.log.info( f{flow.request.path} | fAvg: {statistics.mean(self.stats[flow.request.path]):.3f}s | fMax: {max(self.stats[flow.request.path]):.3f}s ) addons [PerformanceMonitor()]性能优化案例 在某电商项目中这套系统帮助发现了三个关键问题商品详情API的99分位响应时间达到2.3秒支付接口在流量高峰时错误率飙升推荐服务存在N1查询问题实际部署时建议结合--set termlog_verbosityinfo参数实时输出监控日志并通过Telegraf等工具接入现有监控体系。
Mitmproxy进阶玩法：除了抓包，我这样用它做自动化测试和数据Mock
Mitmproxy进阶玩法除了抓包我这样用它做自动化测试和数据Mock当你还在用Mitmproxy简单地抓包调试时一些技术团队已经将它变成了研发流程中的瑞士军刀。想象一下凌晨三点的CI/CD流水线自动触发了数百个接口测试用例Mitmproxy实时修改返回数据模拟服务器崩溃场景前端开发者在后端API尚未完成时直接拦截请求返回本地设计好的JSON数据线上真实流量被精准引流到测试环境进行压力测试——这些场景正在重新定义我们对代理工具的认知。1. 动态流量改写让自动化测试拥有超能力传统自动化测试最大的瓶颈在于难以模拟真实世界的异常场景。通过Mitmproxy的response钩子我们可以像手术刀般精准修改流量内容。下面这段代码展示了如何随机注入错误状态码from mitmproxy import http import random class ChaosInjector: def response(self, flow: http.HTTPFlow): if api/v1/payments in flow.request.url: # 10%概率触发异常模拟 if random.random() 0.1: flow.response.status_code 503 flow.response.content b{error: service_unavailable} addons [ChaosInjector()]典型应用场景对比表测试类型传统方法Mitmproxy方案优势对比超时测试修改服务端代码动态添加延迟无需部署环境错误码覆盖准备多套测试数据实时修改状态码用例维护成本降低70%数据边界测试人工构造异常参数自动篡改响应字段覆盖率达到100%提示在Kubernetes环境中可以将Mitmproxy部署为Sidecar容器通过--set upstream_certfalse参数避免证书验证问题。2. 智能Mock系统前后端并行的加速器现代研发流程中前后端并行开发时最耗时的往往是接口联调。我们构建了一套基于文件系统的智能Mock方案import json from pathlib import Path class ApiMock: def __init__(self): self.mock_dir Path(__file__).parent / mock_data def request(self, flow: http.HTTPFlow): mock_file self.mock_dir / f{flow.request.path.replace(/, _)}.json if mock_file.exists(): flow.response http.Response.make( 200, mock_file.read_bytes(), {Content-Type: application/json} ) addons [ApiMock()]文件结构示例mock_data/ ├── api_v1_users.json ├── api_v1_products.json └── api_v2_orders.json这套系统在实际项目中带来了三个显著改进前端启动时间从平均2小时缩短到5分钟接口变更时只需更新对应JSON文件支持历史版本快速回滚测试3. 流量录制回放真实场景的完美复刻线上问题排查最可靠的方式就是复现真实流量。我们开发了流量归档系统关键代码片段如下import time from datetime import datetime class TrafficRecorder: def __init__(self): self.archive_dir Path(ftraffic_{datetime.now().strftime(%Y%m%d)}) self.archive_dir.mkdir(exist_okTrue) def response(self, flow: http.HTTPFlow): if flow.request.method POST: timestamp int(time.time() * 1000) with open(self.archive_dir / f{timestamp}.har, wb) as f: f.write(json.dumps({ request: { url: flow.request.url, headers: dict(flow.request.headers), content: flow.request.content.decode() }, response: { status: flow.response.status_code, content: flow.response.content.decode() } }).encode()) addons [TrafficRecorder()]回放流程优化技巧使用mitmdump -S archive.har直接回放历史流量结合--ignore-hosts参数过滤非目标域名通过--set flow_detail0提升回放性能4. 安全测试实战加解密接口的自动化方案面对加密接口传统测试工具往往束手无策。我们通过Mitmproxy搭建了透明的加解密中间层from Crypto.Cipher import AES import base64 class CryptoMiddleware: def __init__(self): self.key byour_32byte_secret_key def request(self, flow: http.HTTPFlow): if flow.request.content: cipher AES.new(self.key, AES.MODE_EAX) nonce cipher.nonce ciphertext, tag cipher.encrypt_and_digest(flow.request.content) flow.request.content base64.b64encode(nonce tag ciphertext) def response(self, flow: http.HTTPFlow): if flow.response.content: data base64.b64decode(flow.response.content) nonce, tag, ciphertext data[:16], data[16:32], data[32:] cipher AES.new(self.key, AES.MODE_EAX, noncenonce) flow.response.content cipher.decrypt_and_verify(ciphertext, tag) addons [CryptoMiddleware()]加密接口测试的最佳实践将加解密逻辑封装为独立Python模块通过环境变量管理密钥等敏感信息在测试报告中保留原始加密数据和解密后内容5. 性能监控与瓶颈定位Mitmproxy的流量分析能力可以转化为实时性能监控工具。以下代码实现了API耗时统计import statistics from collections import defaultdict class PerformanceMonitor: def __init__(self): self.stats defaultdict(list) def response(self, flow: http.HTTPFlow): latency flow.response.timestamp_end - flow.request.timestamp_start self.stats[flow.request.path].append(latency) if len(self.stats[flow.request.path]) % 10 0: ctx.log.info( f{flow.request.path} | fAvg: {statistics.mean(self.stats[flow.request.path]):.3f}s | fMax: {max(self.stats[flow.request.path]):.3f}s ) addons [PerformanceMonitor()]性能优化案例 在某电商项目中这套系统帮助发现了三个关键问题商品详情API的99分位响应时间达到2.3秒支付接口在流量高峰时错误率飙升推荐服务存在N1查询问题实际部署时建议结合--set termlog_verbosityinfo参数实时输出监控日志并通过Telegraf等工具接入现有监控体系。
