1. 整体设计为什么要做一套 mitmproxy 脚本集做客户端开发或者接口联调的朋友肯定都经历过这种痛苦后端接口还没写好前端页面已经等着联调了线上环境出了个偶现问题需要把请求参数改一下复现又或者想验证一下异常响应下 App 的崩溃保护逻辑总不能真把服务器搞挂吧。这些场景里一个能自动化拦截、分析、修改HTTP/HTTPS流量的工具就成了刚需。mitmproxy 就是干这个的。它本质上是一个中间人代理手机和电脑的流量经过它的时候可以用 Python 脚本实时地看、改、放行或者直接丢弃。我以前常用它来做三件事抓包看接口数据、批量改写请求参数做压测预处理、给测试同学做一个流量日记方便回溯问题。这篇文章就是把我在实际工作中沉淀下来的一套脚本集从头到尾拆开讲包括目录怎么组织、addon 怎么设计、常用的内置方法在哪找、哪些坑我踩过了你们别再踩。先说结论这套脚本集不是一股脑把功能都塞进一个文件里而是按采集、改写、拦截、记录四个维度拆成独立插件再通过一个入口统一加载。这样做的好处特别直接——某天你想临时关掉请求改写功能只需要注释掉一行配置而不是去一个三千行的脚本里去翻逻辑。适合谁来参考呢如果你已经在用 mitmproxy 的交互模式抓包想更进一步做自动化处理或者你负责接口测试平台需要给团队搭一套可复用的流量处理工具又或者你是做爬虫的需要批量清洗请求参数——这篇文章都能给你一个可以直接抄作业的框架。后面我会把每个环节的代码结构、设计理由、实际运行效果全写出来。2. 理解 mitmproxy 的插件机制addon 到底是怎么跑起来的2.1 从一次请求的生命周期看 Hook 点mitmproxy 的脚本本质是一组 Hook 函数的集合它内部的事件驱动模型会在流量通过的各个阶段调用你定义的函数。理解这个模型是写脚本的前提不然你根本不知道某个需求应该挂在哪个函数里。拿一个最简单的 HTTPS 请求举例它从客户端发出到响应返回大概会依次经过这些关键节点# 请求阶段 def request(flow): # 客户端请求头已经被 mitmproxy 解析完毕此时可以修改任何字段 def requestheaders(flow): # 仅请求头可用body 还没读取适合做早期过滤 # 响应阶段 def response(flow): # 服务端响应已完整到达可以改写 body、headers def responseheaders(flow): # 响应头已到达body 还在传输中如果想做流式处理就在这里我最初犯过一个错误想在requestheaders里修改请求 body结果发现 flow.request.content 是空的。因为requestheaders触发时机太早body 还没读完。后来我统一规则凡是需要读 body 的逻辑一律放request或response只碰 header 的才放*headers。这条规则后来写进了团队的编码规范。2.2 addon 的注册和加载机制先看一个最精简的插件文件长什么样# scripts/addons/header_injector.py from mitmproxy import http class HeaderInjector: def request(self, flow: http.HTTPFlow) - None: flow.request.headers[X-Debug-Tag] traffic-script-v1 addons [HeaderInjector()]每个插件文件末尾必须定义addons变量它告诉 mitmproxy 这个模块暴露哪些类需要被注册。这个设计我觉得很干净——一个文件可以包含多个类但只有addons列表里的会被加载。用命令行启动的时候有两种姿势一种是直接指定脚本路径mitmdump -s scripts/header_injector.py -p 8080另一种是写了多个脚本后建议用入口文件统一管理mitmdump -s scripts/entrypoint.py -p 8080入口文件里只需要做一个动作把所有 addon 实例收集起来。# scripts/entrypoint.py from addons.header_injector import HeaderInjector from addons.param_rewriter import ParamRewriter from addons.response_mock import ResponseMock from addons.flow_recorder import FlowRecorder addons [ HeaderInjector(), ParamRewriter(), ResponseMock(), FlowRecorder(), ]很多初学者会困惑一个问题mitmdump -s和直接在mitmproxy交互界面里按~加载脚本有什么区别。我的理解是mitmdump是无界面模式适合跑自动化任务mitmproxy是交互模式适合边看边调。实际做自动化时我们用mitmdump就够了配合-w参数还能把流量输出到文件里供后续离线分析。2.3 脚本集的模块划分原则基于工作场景我把脚本集分成了四类。每类解决一类问题互不干扰模块核心职责典型场景采集与记录把请求/响应全量保存到本地联调过后回溯问题、生成接口文档请求改写修改参数、header、body批量测试不同入参、注入测试标识响应模拟拦截请求并返回自定义响应后端未完成时先给前端提供 mock 数据监控与告警统计异常状态码、超时请求回归测试时自动发现接口异常这种划分还有个隐形好处团队协作时可以多人并行开发不同模块互不 conflict。我们有一次两个同事同时改代码一个在param_rewriter里加签名逻辑一个在flow_recorder里加 es 上报最后合并时几乎没产生冲突就是因为模块边界足够清晰。3. 核心细节解析流量修改的关键操作与参数计算3.1 请求头与请求体的修改技巧改 header 是最常见的需求。比如你要给所有请求加上一个时间戳参数或者把某个环境标识替换掉import time from mitmproxy import http class TimestampInjector: def request(self, flow: http.HTTPFlow) - None: flow.request.headers[X-Request-Time] str(int(time.time() * 1000)) # 如果 header 不存在才添加避免覆盖已有字段 if X-Env not in flow.request.headers: flow.request.headers[X-Env] test这里有个细节很容易踩坑flow.request.headers是一个Headers对象它的行为类似于大小写不敏感的字典但赋值时如果你直接flow.request.headers[X-Env] a而原来已经有一个x-env: b结果会把两个都保留下来。所以先判断存在性再决定是覆盖还是新增这是我在实际中得出的经验。修改请求 body 时要注意 mitmproxy 里flow.request.content是bytes类型。如果你拿到的原始数据是 JSON 字符串必须经历一个字符串 → 字典 → 修改 → 字符串 → 字节的转换过程import json from mitmproxy import http class BodyParamRewriter: def request(self, flow: http.HTTPFlow) - None: if application/json not in flow.request.headers.get(Content-Type, ): return try: payload json.loads(flow.request.content.decode(utf-8)) except (UnicodeDecodeError, json.JSONDecodeError): return # 在 body 里加一个追踪字段 payload.setdefault(traceId, auto-gen-20240528) flow.request.content json.dumps(payload).encode(utf-8) # 修改 body 后必须同步更新 Content-Length flow.request.headers[Content-Length] str(len(flow.request.content))很多人会忘记最后一步改了 body 后Content-Length没有同步更新导致服务端读到了一个不完整的请求体。这是 mitmproxy 新手最容易犯的错误之一。我第一次漏掉的时候后端同事排查了两个小时最后发现是 Content-Length 对不上。从那以后凡是改 body我必改 Content-Length已经形成肌肉记忆了。3.2 响应数据替换与 mock 逻辑后端接口还没就绪时用 mitmproxy 做 mock 是最省事的方案。你不需要起一个单独的 mock server只需要在响应回来之前把它替换掉import json from mitmproxy import http class MockResponse: def __init__(self): self.rules { /api/v1/user/info: { code: 0, data: {name: tester, level: vip}, message: success } } def request(self, flow: http.HTTPFlow) - None: if flow.request.pretty_url in self.rules: mock_data self.rules[flow.request.pretty_url] flow.response http.Response.make( 200, json.dumps(mock_data).encode(utf-8), {Content-Type: application/json} )核心代码就几行。但我在实际项目中很快发现一个痛点mock 规则硬编码在 Python 文件里测试同学想改数据还得找我改代码。后来我把规则拆到了一个 JSON 配置文件里脚本启动时读取这样测试同学自己就能改 mock 数据了。import json import os from mitmproxy import http class ConfigurableMock: def __init__(self): config_path os.path.join(os.path.dirname(__file__), mock_rules.json) with open(config_path, r, encodingutf-8) as f: self.rules json.load(f) def request(self, flow: http.HTTPFlow) - None: url flow.request.pretty_url if url in self.rules: rule self.rules[url] flow.response http.Response.make( rule[status], json.dumps(rule[body]).encode(utf-8), {Content-Type: application/json} )JSON 配置文件长这样{ /api/v1/user/info: { status: 200, body: {code: 0, data: {name: tester}, message: success} }, /api/v1/order/list: { status: 200, body: {code: 0, data: [], message: success} } }3.3 按域名过滤避免影响非目标流量如果脚本不经筛选地对所有流量生效你会发现连 IDE 的自动更新请求都被改写了。所以每个规则都必须带上作用范围。我惯用的做法是维护一个白名单列表class DomainFilter: WHITE_LIST [ api.example.com, gateway.example.net, test.internal-service.com, ] staticmethod def match(host: str) - bool: # 支持子域名通配 for domain in DomainFilter.WHITE_LIST: if host domain or host.endswith(. domain): return True return False然后在各插件里这样用class WhitelistedRewriter: def request(self, flow: http.HTTPFlow) - None: if not DomainFilter.match(flow.request.host): return # 只有命中白名单才做改写通配匹配是我后来加的。第一次只写了精确匹配结果发现测试环境的域名是test.api.example.com而不是api.example.com导致一大批流量没被处理。加上endswith(. domain)后这种问题就消失了。3.4 参数替换的批量场景除了单请求修改还有一类需求是批量压测时的参数替换。比如你要对某个接口做 100 次调用每次传入不同的用户 ID手动改太累用脚本自动化就舒服了import random from mitmproxy import http class UserIdRandomizer: def __init__(self): self.counter 0 self.user_ids [1001, 1002, 1003, 1004, 1005] def request(self, flow: http.HTTPFlow) - None: if /api/v1/user/detail not in flow.request.pretty_url: return # 从预设列表里选一个 ID self.counter (self.counter 1) % len(self.user_ids) uid self.user_ids[self.counter] # query 参数替换 flow.request.query[userId] str(uid) # 打印当前替换情况方便追踪 print(f[UserIdRandomizer] #{self.counter} - userId{uid})这种做法的价值在于你不需要改动任何业务代码就可以在流量层模拟不同用户。配合压测工具使用效果尤其好。4. 实操从零到一搭建一套可用的脚本集4.1 环境准备与版本选择先说说环境。mitmproxy 目前有 Python 3.9 的版本都支持我用的是 Python 3.11。安装方式直接 pip 就行pip install mitmproxy安装完后验证一下版本mitmdump --version这里有个版本选择的经验mitmproxy 的 API 在不同版本之间略有变化。我最早用的 8.x后来升级到 10.x发现response的make方法参数顺序有过调整。如果你照着网上的教程写代码一定要确认教程对应的版本不然运行时会报诡异的错误。我的建议是直接用最新稳定版API 文档以官网为准。首次运行时mitmproxy 会在你的用户目录生成一个.mitmproxy文件夹里面是 CA 证书。手机或者客户端要正常抓 HTTPS 包必须先安装并信任这个证书。这一步没有捷径iOS 和 Android 的安装方式略有不同但核心都是两步下载证书文件、在系统设置里信任它。4.2 目录结构与工程化组织工程化是脚本集能不能长期维护的关键。我建议的目录结构如下mitm-scripts/ ├── addons/ │ ├── __init__.py │ ├── domain_filter.py │ ├── header_injector.py │ ├── param_rewriter.py │ ├── response_mock.py │ └── flow_recorder.py ├── config/ │ ├── mock_rules.json │ └── domains.json ├── logs/ │ └── flow_records/ ├── entrypoint.py └── requirements.txtaddons目录放各类插件config目录放规则配置logs目录放抓取到的流量记录。entrypoint.py作为唯一入口每次启动都指向它mitmdump -s entrypoint.py -p 8080 --set flow_detail2--set flow_detail2是让 mitmproxy 输出更详细的日志方便调试。4.3 实践实现一个全量流量记录器记录流量是最实用的功能之一。联调时报个问题如果能把当时的请求和响应日志拉出来看排查效率翻几倍。我的实现思路是把每个 HTTP 流的关键信息时间、URL、状态码、请求头、响应摘要写成一行 JSON 追加到日志文件。import json import time from mitmproxy import http class FlowRecorder: def __init__(self): self.log_file logs/flow_records/access.log def response(self, flow: http.HTTPFlow) - None: record { timestamp: time.strftime(%Y-%m-%d %H:%M:%S), url: flow.request.pretty_url, method: flow.request.method, status: flow.response.status_code if flow.response else N/A, request_headers: dict(flow.request.headers), response_preview: self._truncate( flow.response.get_text() if flow.response else , 200 ), } with open(self.log_file, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n) def _truncate(self, text: str, max_len: int) - str: if len(text) max_len: return text return text[:max_len] ...(truncated)这里我故意把响应内容截断到 200 字符避免日志文件快速膨胀。如果你想保存完整 body可以单独开一个body_recorder把 body 存成独立文件而不是和 header 混在一起。4.4 实践基于规则引擎的请求改写器规则引擎是脚本集的核心模块。我的做法是把常见的改写需求抽象成规则每个规则包含匹配条件和执行动作两部分。import json import re from mitmproxy import http class RuleEngine: def __init__(self): self.rules [ { match: {url: r/api/v1/order, param: status}, action: {set_param: statusPAID} }, { match: {header: X-Env}, action: {set_header: X-Envtest} } ] def request(self, flow: http.HTTPFlow) - None: for rule in self.rules: self._apply_rule(flow, rule) def _apply_rule(self, flow, rule): match rule[match] action rule[action] # 检查 URL 模式 if url in match and not re.search(match[url], flow.request.pretty_url): return # 执行参数设置 if set_param in action: key, value action[set_param].split() flow.request.query[key] value # 执行 header 设置 if set_header in action: key, value action[set_header].split() flow.request.headers[key] value如果你需要动态规则可以把规则放到 JSON 文件里然后定时重载。我后来做的是脚本启动时加载 JSON然后每隔 30 秒检查文件修改时间有变化就重新加载。这样测试同学改完规则不用重启 mitmproxy。4.5 启动脚本的三种模式实际工作中我常用的启动方式有三种分别对应不同场景# 模式一前台运行输出到终端 mitmdump -s entrypoint.py -p 8080 # 模式二后台运行日志写到文件 nohup mitmdump -s entrypoint.py -p 8080 logs/mitmdump.log 21 # 模式三只记录不修改相当于被动抓包 mitmdump -s entrypoint.py -p 8080 --mode regular第三种模式有个细节配置了--mode regular后如果你想让移动设备的流量也经过需要在启动时指定透传参数。不过日常开发调试默认的--mode regular就够了——它开启了一个透明代理端口但需要配合系统代理设置。当你只是想在本地快速测试电脑上某个工具的请求时可以用最简单的方式直接设置系统 HTTP 代理为127.0.0.1:8080。对于手机需要把 Wi-Fi 的代理指到电脑 IP 和 8080 端口。这个方法我每天都用稳定可靠。5. 常见问题与排查技巧实录5.1 证书信任了但抓不到 HTTPS 包这是几乎每个人都会遇到的头号问题。现象是证书已经安装但 mitmproxy 界面上显示 CONNECT 请求之后的 HTTPS 请求一直没有内容。排查思路确认你安装的是 mitmproxy 生成的 CA 证书而不是其他工具的。检查证书有效期。Android 7.0 以上默认不信任用户安装的 CA 证书需要把证书放到系统证书目录或者让 App 配置信任用户证书。iOS 上安装证书后还要到设置 → 通用 → 关于本机 → 证书信任设置里打开开关。这一步很多人会漏掉。我第一次引导同事操作时他在手机上装好了证书但没打开信任开关结果折腾了大半天。后来我写了一份详细的操作文档专门强调这个开关问题就很少再出现了。5.2 脚本改了 body 但服务端收到的还是旧数据这个问题的根源几乎都是Content-Length没有更新。HTTP 协议里body 的长度是靠这个 header 告诉服务端的。你改了 body 字节数但 header 还是旧的服务端按旧长度去读自然读到的就是截断或者错位的数据。解决方案是每次改完 body 后同步更新flow.request.headers[Content-Length] str(len(flow.request.content))还有一个小概率问题是某些框架特别是 Go 写的一些服务是强制要求Content-Length跟实际 body 一致的不一致直接报 400。这种场景下你更要严格保证长度一致。5.3 脚本运行了但没生效可能是什么原因有段时间我发现一个奇怪现象脚本明明启动了某些接口的请求却没有被改写。最后定位到原因脚本里用的flow.request.pretty_url和我想的不一样。pretty_url返回的是形如https://api.example.com/path?query1的完整地址包括 scheme 和 query。如果你用api.example.com/path in flow.request.pretty_url做匹配大概率能命中。但如果你用flow.request.url不处理特殊字符转义的版本在某些场景下会拿到编码后的 URL匹配不到预期内容。我的建议是统一使用pretty_url做匹配并且打印出来看看实际长什么样。多打日志能省去很多排查时间。5.4 处理 WebSocket 流量mitmproxy 对 WebSocket 也有 Hook 支持但和 HTTP 请求的处理逻辑不太一样。WebSocket 的 Hook 是websocket_message它不区分请求和响应阶段而是在每次收到消息时触发。from mitmproxy import http, websocket class WebSocketLogger: def websocket_message(self, flow: websocket.WebSocketFlow, message: websocket.WebSocketMessage) - None: print(fWS message from {client if message.from_client else server}: {message.content})这里你会发现flow的类型是WebSocketFlow不是HTTPFlow。如果只写了import http而没处理 websocket 类型脚本会在处理 WebSocket 流量时静默报错。后来我在入口文件里增加了异常捕获确保某个 addon 报错不会影响其他 addondef safe_call(addon, method_name, *args, **kwargs): try: func getattr(addon, method_name) return func(*args, **kwargs) except AttributeError: pass except Exception as e: print(f[ERROR] {addon.__class__.__name__}.{method_name}: {e})5.5 多个 addon 的执行顺序怎么控制这是脚本集设计里容易被忽略的问题。打个比方响应模拟 addon 生成了一个响应而协议处理 addon 想给所有响应加 CORS 头。如果顺序反了后一个可能覆盖前一个的工作。mitmproxy 的 addon 执行顺序默认按addons列表的顺序。我在入口文件里故意把顺序排成了依赖关系先执行采集记录保证原始数据先落盘再执行请求改写入参改造然后执行响应 mock模拟场景最后执行响应处理加统一 header如果有特殊的顺序需求也可以在 addon 的初始化方法里设置优先级但日常用列表顺序就够了。6. 工程化进阶日志监控与配置热更新6.1 引入日志分级与采样随手用print打日志在脚本量小的时候没问题但当流量大起来终端刷屏会严重影响排查效率。后来我引入了 Python 的logging模块做了分级import logging logger logging.getLogger(mitm-scripts) logger.setLevel(logging.INFO) # 控制台输出 console_handler logging.StreamHandler() console_handler.setLevel(logging.INFO) formatter logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) console_handler.setFormatter(formatter) logger.addHandler(console_handler)同样对流量采样也是必要的。你不希望每个请求都记录完整 body那会迅速把磁盘打满。我常用的策略是默认只记录 URL、状态码、耗时按 1% 比例采样完整 body。这个比例可以根据环境灵活调整。6.2 配置热更新不用重启脚本改规则上面提到过把规则放到 JSON 文件后可以定时检查文件修改时间。代码实现很简单import os import time import json class HotReloadConfig: def __init__(self, config_path): self.config_path config_path self.last_mtime 0 self.config None self.reload() def reload(self): if self.config_path and os.path.exists(self.config_path): mtime os.path.getmtime(self.config_path) if mtime ! self.last_mtime: self.last_mtime mtime with open(self.config_path, r, encodingutf-8) as f: self.config json.load(f) print(f[Config] Reloaded config from {self.config_path}) def tick(self): self.reload()在request或response里调用tick()即可实现热更新。我第一次实现时犹豫过性能问题后来观察了一下os.path.getmtime的开销非常小完全可以在每个请求上调用。6.3 与测试框架结合把 mitmproxy 当作测试工具脚本集稳定后我把它集成到了自动化回归测试流程里。基本思路是跑测试用例之前启动 mitmdump 加载指定 addon 和配置测试过程中通过 mitmproxy 的-w参数把流量导出到文件测试结束后解析流量文件断言某些关键接口是否被调用、响应码是否符合预期。具体来说测试代码里可以通过 subprocess 启动 mitmdumpimport subprocess def start_proxy(script_path, port8080): proc subprocess.Popen( [mitmdump, -s, script_path, -p, str(port), -q], stdoutsubprocess.PIPE, stderrsubprocess.STDOUT, textTrue ) return proc-q参数表示安静模式减少无关日志。这样在 CI 环境里流量分析脚本也能跑起来。7. 我的几点经验和最后的建议这套脚本集我从最初一个 200 行的单文件脚本逐步演进成了现在这个模块化的工程结构。过程中最深的体会是mitmproxy 脚本的能力上限取决于你对 Hook 时机和对象模型的理解程度。当你清楚知道flow.request.content和flow.request.raw_content的区别前者是解码后的字符串字节后者是原始字节很多奇奇怪怪的问题就能一眼看穿。另外给刚开始用 mitmproxy 做自动化的朋友一个最实用的建议先跑通一个只记录不修改的最小脚本再逐步添加改写逻辑。这样做的好处是你可以先确认数据流是通的再来验证修改逻辑是否正确。如果一开始就急着写复杂逻辑出了问题很难定位是环境问题还是代码问题。最后分享一个我一直在用的小技巧调试脚本时在关键节点加一段print日志然后启动 mitmdump 时加上--set console_eventlog_verbositydebug能看到非常详细的内部事件日志。配合日志文件中的时间戳定位问题会快很多。这套东西跑起来之后你会发现自己排查问题的效率翻了一倍不止。
1. 整体设计为什么要做一套 mitmproxy 脚本集做客户端开发或者接口联调的朋友肯定都经历过这种痛苦后端接口还没写好前端页面已经等着联调了线上环境出了个偶现问题需要把请求参数改一下复现又或者想验证一下异常响应下 App 的崩溃保护逻辑总不能真把服务器搞挂吧。这些场景里一个能自动化拦截、分析、修改HTTP/HTTPS流量的工具就成了刚需。mitmproxy 就是干这个的。它本质上是一个中间人代理手机和电脑的流量经过它的时候可以用 Python 脚本实时地看、改、放行或者直接丢弃。我以前常用它来做三件事抓包看接口数据、批量改写请求参数做压测预处理、给测试同学做一个流量日记方便回溯问题。这篇文章就是把我在实际工作中沉淀下来的一套脚本集从头到尾拆开讲包括目录怎么组织、addon 怎么设计、常用的内置方法在哪找、哪些坑我踩过了你们别再踩。先说结论这套脚本集不是一股脑把功能都塞进一个文件里而是按采集、改写、拦截、记录四个维度拆成独立插件再通过一个入口统一加载。这样做的好处特别直接——某天你想临时关掉请求改写功能只需要注释掉一行配置而不是去一个三千行的脚本里去翻逻辑。适合谁来参考呢如果你已经在用 mitmproxy 的交互模式抓包想更进一步做自动化处理或者你负责接口测试平台需要给团队搭一套可复用的流量处理工具又或者你是做爬虫的需要批量清洗请求参数——这篇文章都能给你一个可以直接抄作业的框架。后面我会把每个环节的代码结构、设计理由、实际运行效果全写出来。2. 理解 mitmproxy 的插件机制addon 到底是怎么跑起来的2.1 从一次请求的生命周期看 Hook 点mitmproxy 的脚本本质是一组 Hook 函数的集合它内部的事件驱动模型会在流量通过的各个阶段调用你定义的函数。理解这个模型是写脚本的前提不然你根本不知道某个需求应该挂在哪个函数里。拿一个最简单的 HTTPS 请求举例它从客户端发出到响应返回大概会依次经过这些关键节点# 请求阶段 def request(flow): # 客户端请求头已经被 mitmproxy 解析完毕此时可以修改任何字段 def requestheaders(flow): # 仅请求头可用body 还没读取适合做早期过滤 # 响应阶段 def response(flow): # 服务端响应已完整到达可以改写 body、headers def responseheaders(flow): # 响应头已到达body 还在传输中如果想做流式处理就在这里我最初犯过一个错误想在requestheaders里修改请求 body结果发现 flow.request.content 是空的。因为requestheaders触发时机太早body 还没读完。后来我统一规则凡是需要读 body 的逻辑一律放request或response只碰 header 的才放*headers。这条规则后来写进了团队的编码规范。2.2 addon 的注册和加载机制先看一个最精简的插件文件长什么样# scripts/addons/header_injector.py from mitmproxy import http class HeaderInjector: def request(self, flow: http.HTTPFlow) - None: flow.request.headers[X-Debug-Tag] traffic-script-v1 addons [HeaderInjector()]每个插件文件末尾必须定义addons变量它告诉 mitmproxy 这个模块暴露哪些类需要被注册。这个设计我觉得很干净——一个文件可以包含多个类但只有addons列表里的会被加载。用命令行启动的时候有两种姿势一种是直接指定脚本路径mitmdump -s scripts/header_injector.py -p 8080另一种是写了多个脚本后建议用入口文件统一管理mitmdump -s scripts/entrypoint.py -p 8080入口文件里只需要做一个动作把所有 addon 实例收集起来。# scripts/entrypoint.py from addons.header_injector import HeaderInjector from addons.param_rewriter import ParamRewriter from addons.response_mock import ResponseMock from addons.flow_recorder import FlowRecorder addons [ HeaderInjector(), ParamRewriter(), ResponseMock(), FlowRecorder(), ]很多初学者会困惑一个问题mitmdump -s和直接在mitmproxy交互界面里按~加载脚本有什么区别。我的理解是mitmdump是无界面模式适合跑自动化任务mitmproxy是交互模式适合边看边调。实际做自动化时我们用mitmdump就够了配合-w参数还能把流量输出到文件里供后续离线分析。2.3 脚本集的模块划分原则基于工作场景我把脚本集分成了四类。每类解决一类问题互不干扰模块核心职责典型场景采集与记录把请求/响应全量保存到本地联调过后回溯问题、生成接口文档请求改写修改参数、header、body批量测试不同入参、注入测试标识响应模拟拦截请求并返回自定义响应后端未完成时先给前端提供 mock 数据监控与告警统计异常状态码、超时请求回归测试时自动发现接口异常这种划分还有个隐形好处团队协作时可以多人并行开发不同模块互不 conflict。我们有一次两个同事同时改代码一个在param_rewriter里加签名逻辑一个在flow_recorder里加 es 上报最后合并时几乎没产生冲突就是因为模块边界足够清晰。3. 核心细节解析流量修改的关键操作与参数计算3.1 请求头与请求体的修改技巧改 header 是最常见的需求。比如你要给所有请求加上一个时间戳参数或者把某个环境标识替换掉import time from mitmproxy import http class TimestampInjector: def request(self, flow: http.HTTPFlow) - None: flow.request.headers[X-Request-Time] str(int(time.time() * 1000)) # 如果 header 不存在才添加避免覆盖已有字段 if X-Env not in flow.request.headers: flow.request.headers[X-Env] test这里有个细节很容易踩坑flow.request.headers是一个Headers对象它的行为类似于大小写不敏感的字典但赋值时如果你直接flow.request.headers[X-Env] a而原来已经有一个x-env: b结果会把两个都保留下来。所以先判断存在性再决定是覆盖还是新增这是我在实际中得出的经验。修改请求 body 时要注意 mitmproxy 里flow.request.content是bytes类型。如果你拿到的原始数据是 JSON 字符串必须经历一个字符串 → 字典 → 修改 → 字符串 → 字节的转换过程import json from mitmproxy import http class BodyParamRewriter: def request(self, flow: http.HTTPFlow) - None: if application/json not in flow.request.headers.get(Content-Type, ): return try: payload json.loads(flow.request.content.decode(utf-8)) except (UnicodeDecodeError, json.JSONDecodeError): return # 在 body 里加一个追踪字段 payload.setdefault(traceId, auto-gen-20240528) flow.request.content json.dumps(payload).encode(utf-8) # 修改 body 后必须同步更新 Content-Length flow.request.headers[Content-Length] str(len(flow.request.content))很多人会忘记最后一步改了 body 后Content-Length没有同步更新导致服务端读到了一个不完整的请求体。这是 mitmproxy 新手最容易犯的错误之一。我第一次漏掉的时候后端同事排查了两个小时最后发现是 Content-Length 对不上。从那以后凡是改 body我必改 Content-Length已经形成肌肉记忆了。3.2 响应数据替换与 mock 逻辑后端接口还没就绪时用 mitmproxy 做 mock 是最省事的方案。你不需要起一个单独的 mock server只需要在响应回来之前把它替换掉import json from mitmproxy import http class MockResponse: def __init__(self): self.rules { /api/v1/user/info: { code: 0, data: {name: tester, level: vip}, message: success } } def request(self, flow: http.HTTPFlow) - None: if flow.request.pretty_url in self.rules: mock_data self.rules[flow.request.pretty_url] flow.response http.Response.make( 200, json.dumps(mock_data).encode(utf-8), {Content-Type: application/json} )核心代码就几行。但我在实际项目中很快发现一个痛点mock 规则硬编码在 Python 文件里测试同学想改数据还得找我改代码。后来我把规则拆到了一个 JSON 配置文件里脚本启动时读取这样测试同学自己就能改 mock 数据了。import json import os from mitmproxy import http class ConfigurableMock: def __init__(self): config_path os.path.join(os.path.dirname(__file__), mock_rules.json) with open(config_path, r, encodingutf-8) as f: self.rules json.load(f) def request(self, flow: http.HTTPFlow) - None: url flow.request.pretty_url if url in self.rules: rule self.rules[url] flow.response http.Response.make( rule[status], json.dumps(rule[body]).encode(utf-8), {Content-Type: application/json} )JSON 配置文件长这样{ /api/v1/user/info: { status: 200, body: {code: 0, data: {name: tester}, message: success} }, /api/v1/order/list: { status: 200, body: {code: 0, data: [], message: success} } }3.3 按域名过滤避免影响非目标流量如果脚本不经筛选地对所有流量生效你会发现连 IDE 的自动更新请求都被改写了。所以每个规则都必须带上作用范围。我惯用的做法是维护一个白名单列表class DomainFilter: WHITE_LIST [ api.example.com, gateway.example.net, test.internal-service.com, ] staticmethod def match(host: str) - bool: # 支持子域名通配 for domain in DomainFilter.WHITE_LIST: if host domain or host.endswith(. domain): return True return False然后在各插件里这样用class WhitelistedRewriter: def request(self, flow: http.HTTPFlow) - None: if not DomainFilter.match(flow.request.host): return # 只有命中白名单才做改写通配匹配是我后来加的。第一次只写了精确匹配结果发现测试环境的域名是test.api.example.com而不是api.example.com导致一大批流量没被处理。加上endswith(. domain)后这种问题就消失了。3.4 参数替换的批量场景除了单请求修改还有一类需求是批量压测时的参数替换。比如你要对某个接口做 100 次调用每次传入不同的用户 ID手动改太累用脚本自动化就舒服了import random from mitmproxy import http class UserIdRandomizer: def __init__(self): self.counter 0 self.user_ids [1001, 1002, 1003, 1004, 1005] def request(self, flow: http.HTTPFlow) - None: if /api/v1/user/detail not in flow.request.pretty_url: return # 从预设列表里选一个 ID self.counter (self.counter 1) % len(self.user_ids) uid self.user_ids[self.counter] # query 参数替换 flow.request.query[userId] str(uid) # 打印当前替换情况方便追踪 print(f[UserIdRandomizer] #{self.counter} - userId{uid})这种做法的价值在于你不需要改动任何业务代码就可以在流量层模拟不同用户。配合压测工具使用效果尤其好。4. 实操从零到一搭建一套可用的脚本集4.1 环境准备与版本选择先说说环境。mitmproxy 目前有 Python 3.9 的版本都支持我用的是 Python 3.11。安装方式直接 pip 就行pip install mitmproxy安装完后验证一下版本mitmdump --version这里有个版本选择的经验mitmproxy 的 API 在不同版本之间略有变化。我最早用的 8.x后来升级到 10.x发现response的make方法参数顺序有过调整。如果你照着网上的教程写代码一定要确认教程对应的版本不然运行时会报诡异的错误。我的建议是直接用最新稳定版API 文档以官网为准。首次运行时mitmproxy 会在你的用户目录生成一个.mitmproxy文件夹里面是 CA 证书。手机或者客户端要正常抓 HTTPS 包必须先安装并信任这个证书。这一步没有捷径iOS 和 Android 的安装方式略有不同但核心都是两步下载证书文件、在系统设置里信任它。4.2 目录结构与工程化组织工程化是脚本集能不能长期维护的关键。我建议的目录结构如下mitm-scripts/ ├── addons/ │ ├── __init__.py │ ├── domain_filter.py │ ├── header_injector.py │ ├── param_rewriter.py │ ├── response_mock.py │ └── flow_recorder.py ├── config/ │ ├── mock_rules.json │ └── domains.json ├── logs/ │ └── flow_records/ ├── entrypoint.py └── requirements.txtaddons目录放各类插件config目录放规则配置logs目录放抓取到的流量记录。entrypoint.py作为唯一入口每次启动都指向它mitmdump -s entrypoint.py -p 8080 --set flow_detail2--set flow_detail2是让 mitmproxy 输出更详细的日志方便调试。4.3 实践实现一个全量流量记录器记录流量是最实用的功能之一。联调时报个问题如果能把当时的请求和响应日志拉出来看排查效率翻几倍。我的实现思路是把每个 HTTP 流的关键信息时间、URL、状态码、请求头、响应摘要写成一行 JSON 追加到日志文件。import json import time from mitmproxy import http class FlowRecorder: def __init__(self): self.log_file logs/flow_records/access.log def response(self, flow: http.HTTPFlow) - None: record { timestamp: time.strftime(%Y-%m-%d %H:%M:%S), url: flow.request.pretty_url, method: flow.request.method, status: flow.response.status_code if flow.response else N/A, request_headers: dict(flow.request.headers), response_preview: self._truncate( flow.response.get_text() if flow.response else , 200 ), } with open(self.log_file, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n) def _truncate(self, text: str, max_len: int) - str: if len(text) max_len: return text return text[:max_len] ...(truncated)这里我故意把响应内容截断到 200 字符避免日志文件快速膨胀。如果你想保存完整 body可以单独开一个body_recorder把 body 存成独立文件而不是和 header 混在一起。4.4 实践基于规则引擎的请求改写器规则引擎是脚本集的核心模块。我的做法是把常见的改写需求抽象成规则每个规则包含匹配条件和执行动作两部分。import json import re from mitmproxy import http class RuleEngine: def __init__(self): self.rules [ { match: {url: r/api/v1/order, param: status}, action: {set_param: statusPAID} }, { match: {header: X-Env}, action: {set_header: X-Envtest} } ] def request(self, flow: http.HTTPFlow) - None: for rule in self.rules: self._apply_rule(flow, rule) def _apply_rule(self, flow, rule): match rule[match] action rule[action] # 检查 URL 模式 if url in match and not re.search(match[url], flow.request.pretty_url): return # 执行参数设置 if set_param in action: key, value action[set_param].split() flow.request.query[key] value # 执行 header 设置 if set_header in action: key, value action[set_header].split() flow.request.headers[key] value如果你需要动态规则可以把规则放到 JSON 文件里然后定时重载。我后来做的是脚本启动时加载 JSON然后每隔 30 秒检查文件修改时间有变化就重新加载。这样测试同学改完规则不用重启 mitmproxy。4.5 启动脚本的三种模式实际工作中我常用的启动方式有三种分别对应不同场景# 模式一前台运行输出到终端 mitmdump -s entrypoint.py -p 8080 # 模式二后台运行日志写到文件 nohup mitmdump -s entrypoint.py -p 8080 logs/mitmdump.log 21 # 模式三只记录不修改相当于被动抓包 mitmdump -s entrypoint.py -p 8080 --mode regular第三种模式有个细节配置了--mode regular后如果你想让移动设备的流量也经过需要在启动时指定透传参数。不过日常开发调试默认的--mode regular就够了——它开启了一个透明代理端口但需要配合系统代理设置。当你只是想在本地快速测试电脑上某个工具的请求时可以用最简单的方式直接设置系统 HTTP 代理为127.0.0.1:8080。对于手机需要把 Wi-Fi 的代理指到电脑 IP 和 8080 端口。这个方法我每天都用稳定可靠。5. 常见问题与排查技巧实录5.1 证书信任了但抓不到 HTTPS 包这是几乎每个人都会遇到的头号问题。现象是证书已经安装但 mitmproxy 界面上显示 CONNECT 请求之后的 HTTPS 请求一直没有内容。排查思路确认你安装的是 mitmproxy 生成的 CA 证书而不是其他工具的。检查证书有效期。Android 7.0 以上默认不信任用户安装的 CA 证书需要把证书放到系统证书目录或者让 App 配置信任用户证书。iOS 上安装证书后还要到设置 → 通用 → 关于本机 → 证书信任设置里打开开关。这一步很多人会漏掉。我第一次引导同事操作时他在手机上装好了证书但没打开信任开关结果折腾了大半天。后来我写了一份详细的操作文档专门强调这个开关问题就很少再出现了。5.2 脚本改了 body 但服务端收到的还是旧数据这个问题的根源几乎都是Content-Length没有更新。HTTP 协议里body 的长度是靠这个 header 告诉服务端的。你改了 body 字节数但 header 还是旧的服务端按旧长度去读自然读到的就是截断或者错位的数据。解决方案是每次改完 body 后同步更新flow.request.headers[Content-Length] str(len(flow.request.content))还有一个小概率问题是某些框架特别是 Go 写的一些服务是强制要求Content-Length跟实际 body 一致的不一致直接报 400。这种场景下你更要严格保证长度一致。5.3 脚本运行了但没生效可能是什么原因有段时间我发现一个奇怪现象脚本明明启动了某些接口的请求却没有被改写。最后定位到原因脚本里用的flow.request.pretty_url和我想的不一样。pretty_url返回的是形如https://api.example.com/path?query1的完整地址包括 scheme 和 query。如果你用api.example.com/path in flow.request.pretty_url做匹配大概率能命中。但如果你用flow.request.url不处理特殊字符转义的版本在某些场景下会拿到编码后的 URL匹配不到预期内容。我的建议是统一使用pretty_url做匹配并且打印出来看看实际长什么样。多打日志能省去很多排查时间。5.4 处理 WebSocket 流量mitmproxy 对 WebSocket 也有 Hook 支持但和 HTTP 请求的处理逻辑不太一样。WebSocket 的 Hook 是websocket_message它不区分请求和响应阶段而是在每次收到消息时触发。from mitmproxy import http, websocket class WebSocketLogger: def websocket_message(self, flow: websocket.WebSocketFlow, message: websocket.WebSocketMessage) - None: print(fWS message from {client if message.from_client else server}: {message.content})这里你会发现flow的类型是WebSocketFlow不是HTTPFlow。如果只写了import http而没处理 websocket 类型脚本会在处理 WebSocket 流量时静默报错。后来我在入口文件里增加了异常捕获确保某个 addon 报错不会影响其他 addondef safe_call(addon, method_name, *args, **kwargs): try: func getattr(addon, method_name) return func(*args, **kwargs) except AttributeError: pass except Exception as e: print(f[ERROR] {addon.__class__.__name__}.{method_name}: {e})5.5 多个 addon 的执行顺序怎么控制这是脚本集设计里容易被忽略的问题。打个比方响应模拟 addon 生成了一个响应而协议处理 addon 想给所有响应加 CORS 头。如果顺序反了后一个可能覆盖前一个的工作。mitmproxy 的 addon 执行顺序默认按addons列表的顺序。我在入口文件里故意把顺序排成了依赖关系先执行采集记录保证原始数据先落盘再执行请求改写入参改造然后执行响应 mock模拟场景最后执行响应处理加统一 header如果有特殊的顺序需求也可以在 addon 的初始化方法里设置优先级但日常用列表顺序就够了。6. 工程化进阶日志监控与配置热更新6.1 引入日志分级与采样随手用print打日志在脚本量小的时候没问题但当流量大起来终端刷屏会严重影响排查效率。后来我引入了 Python 的logging模块做了分级import logging logger logging.getLogger(mitm-scripts) logger.setLevel(logging.INFO) # 控制台输出 console_handler logging.StreamHandler() console_handler.setLevel(logging.INFO) formatter logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) console_handler.setFormatter(formatter) logger.addHandler(console_handler)同样对流量采样也是必要的。你不希望每个请求都记录完整 body那会迅速把磁盘打满。我常用的策略是默认只记录 URL、状态码、耗时按 1% 比例采样完整 body。这个比例可以根据环境灵活调整。6.2 配置热更新不用重启脚本改规则上面提到过把规则放到 JSON 文件后可以定时检查文件修改时间。代码实现很简单import os import time import json class HotReloadConfig: def __init__(self, config_path): self.config_path config_path self.last_mtime 0 self.config None self.reload() def reload(self): if self.config_path and os.path.exists(self.config_path): mtime os.path.getmtime(self.config_path) if mtime ! self.last_mtime: self.last_mtime mtime with open(self.config_path, r, encodingutf-8) as f: self.config json.load(f) print(f[Config] Reloaded config from {self.config_path}) def tick(self): self.reload()在request或response里调用tick()即可实现热更新。我第一次实现时犹豫过性能问题后来观察了一下os.path.getmtime的开销非常小完全可以在每个请求上调用。6.3 与测试框架结合把 mitmproxy 当作测试工具脚本集稳定后我把它集成到了自动化回归测试流程里。基本思路是跑测试用例之前启动 mitmdump 加载指定 addon 和配置测试过程中通过 mitmproxy 的-w参数把流量导出到文件测试结束后解析流量文件断言某些关键接口是否被调用、响应码是否符合预期。具体来说测试代码里可以通过 subprocess 启动 mitmdumpimport subprocess def start_proxy(script_path, port8080): proc subprocess.Popen( [mitmdump, -s, script_path, -p, str(port), -q], stdoutsubprocess.PIPE, stderrsubprocess.STDOUT, textTrue ) return proc-q参数表示安静模式减少无关日志。这样在 CI 环境里流量分析脚本也能跑起来。7. 我的几点经验和最后的建议这套脚本集我从最初一个 200 行的单文件脚本逐步演进成了现在这个模块化的工程结构。过程中最深的体会是mitmproxy 脚本的能力上限取决于你对 Hook 时机和对象模型的理解程度。当你清楚知道flow.request.content和flow.request.raw_content的区别前者是解码后的字符串字节后者是原始字节很多奇奇怪怪的问题就能一眼看穿。另外给刚开始用 mitmproxy 做自动化的朋友一个最实用的建议先跑通一个只记录不修改的最小脚本再逐步添加改写逻辑。这样做的好处是你可以先确认数据流是通的再来验证修改逻辑是否正确。如果一开始就急着写复杂逻辑出了问题很难定位是环境问题还是代码问题。最后分享一个我一直在用的小技巧调试脚本时在关键节点加一段print日志然后启动 mitmdump 时加上--set console_eventlog_verbositydebug能看到非常详细的内部事件日志。配合日志文件中的时间戳定位问题会快很多。这套东西跑起来之后你会发现自己排查问题的效率翻了一倍不止。
