mitmproxy脚本集实战：从插件机制到流量改写与Mock自动化

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 脚本集的模块划分原则

基于工作场景，我把脚本集分成了四类。每类解决一类问题，互不干扰：

这种划分还有个隐形好处：团队协作时可以多人并行开发不同模块，互不 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", encoding="utf-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.txt

addons目录放各类插件，config目录放规则配置，logs目录放抓取到的流量记录。entrypoint.py作为唯一入口，每次启动都指向它：

mitmdump -s entrypoint.py -p 8080 --set flow_detail=2

--set flow_detail=2是让 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", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\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": "status=PAID"} }, { "match": {"header": "X-Env"}, "action": {"set_header": "X-Env=test"} } ] 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 2>&1 & # 模式三：只记录不修改（相当于被动抓包） 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 请求一直没有内容。

排查思路：

1. 确认你安装的是 mitmproxy 生成的 CA 证书，而不是其他工具的。检查证书有效期。
2. Android 7.0 以上默认不信任用户安装的 CA 证书，需要把证书放到系统证书目录，或者让 App 配置信任用户证书。
3. 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?query=1的完整地址，包括 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(f"WS message from {'client' if message.from_client else 'server'}: {message.content}")

这里你会发现，flow的类型是WebSocketFlow，不是HTTPFlow。如果只写了import http而没处理 websocket 类型，脚本会在处理 WebSocket 流量时静默报错。后来我在入口文件里增加了异常捕获，确保某个 addon 报错不会影响其他 addon：

def 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列表的顺序。我在入口文件里故意把顺序排成了依赖关系：

1. 先执行采集记录（保证原始数据先落盘）
2. 再执行请求改写（入参改造）
3. 然后执行响应 mock（模拟场景）
4. 最后执行响应处理（加统一 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", encoding="utf-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 启动 mitmdump：

import subprocess def start_proxy(script_path, port=8080): proc = subprocess.Popen( ["mitmdump", "-s", script_path, "-p", str(port), "-q"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True ) return proc

-q参数表示安静模式，减少无关日志。这样在 CI 环境里，流量分析脚本也能跑起来。

7. 我的几点经验和最后的建议

这套脚本集我从最初一个 200 行的单文件脚本，逐步演进成了现在这个模块化的工程结构。过程中最深的体会是：mitmproxy 脚本的能力上限，取决于你对 Hook 时机和对象模型的理解程度。当你清楚知道flow.request.content和flow.request.raw_content的区别（前者是解码后的字符串字节，后者是原始字节），很多奇奇怪怪的问题就能一眼看穿。

另外，给刚开始用 mitmproxy 做自动化的朋友一个最实用的建议：先跑通一个"只记录不修改"的最小脚本，再逐步添加改写逻辑。这样做的好处是，你可以先确认数据流是通的，再来验证修改逻辑是否正确。如果一开始就急着写复杂逻辑，出了问题很难定位是环境问题还是代码问题。

最后分享一个我一直在用的小技巧：调试脚本时，在关键节点加一段print日志，然后启动 mitmdump 时加上--set console_eventlog_verbosity=debug，能看到非常详细的内部事件日志。配合日志文件中的时间戳，定位问题会快很多。这套东西跑起来之后，你会发现自己排查问题的效率翻了一倍不止。