mitmproxy Addon 机制详解：事件钩子、类型化选项与命令、脚本热重载及单元测试实践

mitmproxy Addon 机制详解：事件钩子、类型化选项与命令、脚本热重载及单元测试实践

【免费下载链接】mitmproxyAn interactive TLS-capable intercepting HTTP proxy for penetration testers and software developers.项目地址: https://gitcode.com/GitHub_Trending/mi/mitmproxy

本文以 mitmproxy 官方文档中的 Addon 概览（docs/src/content/addons/overview.md）为主线，完整讲解 mitmproxy 插件（Addon）体系的三大交互机制——事件钩子、类型化选项与命令，并基于仓库源码深入剖析脚本加载、-s参数热重载（Live Reloading）的实现原理与测试方式。读完本文，你将能够独立编写类式与模块缩写式两种 addon，理解从mitmdump -s script.py到钩子被调用的完整调用链，并掌握 addon 的单元测试方法。

1. 为什么 Addon 是 mitmproxy 的核心扩展机制

官方文档开宗明义：mitmproxy 的 addon 机制是其"exceptionally powerful"（极其强大）的部分，事实上 mitmproxy 自身大量功能就是以一组内置 addon 的形式实现的，从反缓存（anticache）、粘性 Cookie（sticky cookies）这类流量处理功能，到开箱引导用的 onboarding webapp，全部由 addon 承载。这些内置 addon 的源码集中位于 mitmproxy/addons/ 目录，例如：

• anticache.py：移除缓存相关头部；
• stickycookie.py：实现 sticky cookies 行为；
• view.py、dumper.py、blocklist.py 等。

换言之，"写一个 addon"与"阅读 mitmproxy 内置 addon"是同一套技能，仓库内的 examples/addons/ 与 examples/contrib/ 目录还提供了 30 余个可直接运行的官方示例。

Addon 与 mitmproxy 主体通过三条途径交互：

1. 事件（Events）：addon 通过响应事件钩子切入并改变 mitmproxy 的行为，钩子签名文档见 event-hooks 文档 及 docs/src/scripts/api-events.py 生成的 API 事件列表；
2. 选项（Options）：addon 通过全局选项存储被配置，选项可写在配置文件里、由用户在交互工具中实时修改，或通过命令行传入，详见 options 文档；
3. 命令（Commands）：addon 可以暴露命令，供用户直接调用，或在 mitmproxy 交互工具中绑定快捷键，详见 commands 文档。

1.1 事件钩子的源码基础

事件在源码中由 mitmproxy/hooks.py 定义。所有钩子都是继承自Hook基类的dataclass，其事件名由类名自动推导：

def __init_subclass__(cls, **kwargs): # initialize .name attribute. HttpRequestHook -> http_request if cls.__dict__.get("name", None) is None: name = cls.__name__.replace("Hook", "") cls.name = re.sub("(?!^)([A-Z]+)", r"_\1", name).lower()

也就是说HttpRequestHook类对应的钩子方法名是request还是http_request，取决于类名到 snake_case 的转换规则；每个 dataclass 字段就是钩子的参数，Hook.args()会按字段顺序把参数传给 addon 上同名的可调用对象。生命周期类钩子在 hooks.py 中定义：

钩子类	addon 方法名	触发时机
ConfigureHook(updated)	configure	配置变更时触发，参数为被修改选项键的集合；启动时以全部选项集合调用一次
DoneHook()	done	addon 被移除或 mitmproxy 关闭时调用，是 addon 能收到的最后一个事件（此时日志已关闭）
RunningHook()	running	代理完全启动、所有 addon 加载且选项就绪时调用
UpdateHook(flows)	update	一个或多个 flow 被（通常是其他 addon）修改时调用

而流量类事件（如request、response）则大量接收Flow对象——修改这些对象就能实时改变流量。例如官方事件文档中的示例 http-add-header.py，在response钩子中给每个响应写入一个递增计数头：

class AddHeader: def __init__(self): self.num = 0 def response(self, flow): self.num = self.num + 1 flow.response.headers["count"] = str(self.num) addons = [AddHeader()]

2. Addon 解剖：一个最小可运行的计数示例

文档中的第一个完整示例是 anatomy.py：

""" Basic skeleton of a mitmproxy addon. Run as follows: mitmproxy -s anatomy.py """ import logging class Counter: def __init__(self): self.num = 0 def request(self, flow): self.num = self.num + 1 logging.info("We've seen %d flows" % self.num) addons = [Counter()]

这是一个统计所见 flow（此处更精确地说是 HTTP 请求）数量的 addon：每看到一个新 flow 就自增并记录日志，输出可以在交互工具的事件日志或mitmdump的控制台中看到。用你顺手的 mitmproxy 工具加载它即可验证，各工具使用的加载参数完全一致：

mitmdump -s ./anatomy.py

关于这段代码，官方文档强调三点，这里逐一对应到源码实现：

（1）addons全局列表。mitmproxy 会拾取模块中的addons全局列表，并把它里面的对象逐一加载进 addon 机制。这一点在 Script.addons 属性中得到印证：Script类把脚本模块本身作为 addon 注册，其addons属性返回[self.ns]，随后由AddonManager递归展开。

（2）addon 就是普通对象。本例中 addon 是Counter的实例。对象注册进管理器后，其名称由 addonmanager.py 的_get_name决定：取实例的.name属性，没有则回退为类名小写——所以Counter实例在ctx.master.addons中可以被get("counter")取到。

（3）request方法就是一个事件。addon 只需为自己想处理的每个事件实现一个同名方法即可，各事件及其签名在事件钩子 API 文档中有完整列表。从源码看，事件分发在 AddonManager.trigger_event 中完成：

async def trigger_event(self, event: hooks.Hook): """ Asynchronously trigger an event across all addons. """ for i in self.chain: try: with safecall(): await self.invoke_addon(i, event) except exceptions.AddonHalt: return

管理器沿 addon 链（self.chain）依次遍历，用safecall()上下文包裹调用——safecall 会捕获除AddonHalt/OptionsError外的所有异常并记入错误日志，保证单个 addon 崩溃不会拖垮整个代理。若 addon 想主动终止后续 addon 对该事件的处理，可抛出exceptions.AddonHalt；_iter_hooks（addonmanager.py#L243-L262）还支持异步钩子函数（async def方法会被 await），并特意容忍与钩子同名的模块导入（比如 addon 里from mitmproxy import log与log钩子撞名）。

2.1 注册流程：register、LoadHook与Loader

当Counter实例被注册时，AddonManager.register 依次做了这些事：

1. 用traverse()递归收集 addon 及其子 addon，并对旧版 API（如clientconnect、add_log等已移除/废弃的钩子）打印迁移警告；
2. 名称冲突检测：同名 addon 已存在则抛出AddonManagerError；
3. 构造Loader并向 addon 发送 LoadHook，load(loader)方法在其中通过loader.add_option(...)与loader.add_command(...)声明自己的选项与命令；
4. 把 addon 及其子 addon 登记进self.lookup，并收集其@command.command装饰的命令。

Loader.add_option的签名见 Loader.add_option：name、typespec（类型）、default、help，可选choices。重复声明同签名选项会静默忽略，签名不一致则发出警告后覆盖。

3. 缩写脚本语法：不写类也能写 addon

有时只想快速写个脚本，不想走"建一个类"的完整流程。addon 机制提供了缩写语法：把整个模块当作一个 addon 对象，事件处理函数直接放在模块顶层。官方示例 anatomy2.py 就只有一行逻辑，为每个请求添加一个头部：

"""An addon using the abbreviated scripting syntax.""" def request(flow): flow.request.headers["myheader"] = "value"

之所以能这样工作，是因为Script注册的是模块对象本身：AddonManager._iter_hooks用getattr(a, event.name, None)在模块命名空间里找同名函数（addonmanager.py#L248-L252），模块顶层的def request(flow)自然命中。注意此写法没有addons = [...]列表，也不依赖实例状态——跨事件的共享状态需要挂在模块级变量上。

4. Addon 开发实战

4.1 热重载（Live Reloading）

用-s path/to/script.py加载的脚本会被持续监视：文件修改时间（mtime）一旦变化，mitmproxy 就会注销旧模块、重新导入文件并注册新 addon——无需重启代理，也不会丢失其他 addon 的状态或在途 flow。这意味着你在编辑器里保存脚本后，约一秒钟内修改即生效。

这一行为的实现完全位于 mitmproxy/addons/script.py：

• 轮询周期是模块常量ReloadInterval = 1（秒），见 script.py#L74；
• Script.__init__在reload=True时通过asyncio_utils.create_task(self.watcher(), ...)启动一个后台监视任务（script.py#L92-L98）；
• watcher() 每秒os.stat一次文件，mtime > last_mtime时调用loadscript()；若文件被删除，则记录日志并通过ctx.options.update(scripts=...)把该脚本从选项中移除后结束任务。

loadscript()（script.py#L113-L131）的重载语义与文档描述逐条对应：

def loadscript(self): logger.info("Loading script %s" % self.path) if self.ns: ctx.master.addons.remove(self.ns) # 注销旧模块 self.ns = None with addonmanager.safecall(): # 新模块导入/注册，异常被安全捕获 ns = load_script(self.fullpath) ctx.master.addons.register(ns) self.ns = ns if self.ns: try: ctx.master.addons.invoke_addon_sync( self.ns, hooks.ConfigureHook(ctx.options.keys()) ) except Exception as e: script_error_handler(self.fullpath, e) if self.is_running: # 若代理已在运行，补发 running 事件 ctx.master.addons.invoke_addon_sync(self.ns, hooks.RunningHook())

由此得到文档中"错误处理三原则"的源码依据：

错误位置	行为	源码依据
导入期 /configure/running	记录到事件日志，旧版本保持未注册（self.ns为None或旧模块已被 remove）	loadscript()中safecall包裹导入注册，configure异常走script_error_handler
事件处理函数（request、response…）内	仅记录日志，addon 不被卸载	AddonManager.trigger_event的safecall()兜底

修复错误后再次保存文件，watcher 会在下一轮轮询重试加载。对应的行为测试在 test/mitmproxy/addons/test_script.py：test_reload验证 mtime 变化触发重新加载（测试中把script.ReloadInterval降到 0.1 加速），test_exception验证错误脚本在load报错后仍被正确隔离。

脚本加载细节：load_script

模块级函数 load_script(path) 值得单独看一眼，它解释了脚本为何能与包内其他模块隔离：

def load_script(path: str) -> types.ModuleType | None: fullname = "__mitmproxy_script__.{}".format( os.path.splitext(os.path.basename(path))[0] ) # the fullname is not unique among scripts, so if there already is an existing script with said # fullname, remove it. sys.modules.pop(fullname, None) oldpath = sys.path sys.path.insert(0, os.path.dirname(path)) try: loader = importlib.machinery.SourceFileLoader(fullname, path) spec = importlib.util.spec_from_loader(fullname, loader=loader) ...

• 每个脚本被赋予__mitmproxy_script__.<文件名>的唯一模块名，重载前先sys.modules.pop清掉同名旧模块，保证重新执行；
• 脚本所在目录临时插入sys.path头部（finally中还原），使脚本内部可以import同目录的辅助模块；
• 若运行的是 PyInstaller 冻结的二进制，ImportError时会在错误信息中追加提示：二进制自带独立 Python 环境，若 addon 需要额外依赖，请从 PyPI 安装 mitmproxy（见 script.py#L42-L52）。

脚本本身也是一个"元 addon"

Script类与 ScriptLoader 揭示了-s参数背后的完整机制：

• ScriptLoader通过loader.add_option("scripts", Sequence[str], [], "Execute a script.")注册了scripts选项——命令行-s foo.py与配置文件中的scripts选项指向同一存储；
• 其configure钩子对比新旧scripts列表：被移除的路径对应Script会注销（"Un-loading script" 日志），新增路径创建Script(path, reload=True)；列表顺序变化只重排、不重建实例，避免不必要的重新初始化；
• 每个Script实例的addons属性返回被监视的脚本模块，形成ScriptLoader → Script → 用户脚本模块的 addon 链，事件沿链逐层分发；
• 此外还注册了一个 script.run 命令：对指定 flows 回放模拟各生命周期事件（注意load事件不会被调用），方便离线调试脚本。

4.2 测试 Addon

因为 addon 就是普通 Python 文件，最简单的单元测试方式就是官方文档给出的：在测试中导入模块、实例化 addon、直接调用事件处理函数。例如针对 2 节的Counter：

from examples.addons.anatomy import Counter def test_counter(): c = Counter() flow = mitmproxy.tflow.tflow() # 用仓库自带的测试 helper 构造 flow c.request(flow) assert c.num == 1

更复杂的测试需求可以参考仓库内的 test/mitmproxy/addons 目录（文档特别提醒：内部测试辅助工具没有稳定 API 保证）。该目录使用的主要 helper 来自 mitmproxy/test/taddons.py 与 mitmproxy/test/tflow.py：

• taddons.context()提供一个完整的临时 mitmproxy 运行上下文（master、addons、options），用于验证 addon 注册、选项声明等集成行为；
• tflow.tflow(resp=True)快速构造带响应的模拟 flow，配合ctx.master.addons.trigger(HttpRequestHook(f))可断言钩子被正确调用，见 test_script.py 的 test_simple。

5. 延伸：让 addon 拥有自己的选项与命令

文档"Addon 解剖"部分提到的三要素里，事件只是其一。若想让 addon 更完整，通常还需要：

• 声明选项：在load(loader)中调用loader.add_option(...)，选项即可出现在配置文件、--set命令行参数和交互选项编辑器中。最小示例 options-simple.py 声明了一个addheader布尔选项，response钩子中按ctx.options.addheader决定是否写入计数头；options-configure.py 则展示了在configure钩子中校验取值、抛exceptions.OptionsError触发回滚的完整写法（例如mitmdump -s ... --set addheader=1000会得到 "addheader must be <= 100" 的错误提示）。选项类型系统（str/int/float/bool、typing.Optional、collections.abc.Sequence）的详细规则见 options 文档；
• 暴露命令：用@mitmproxy.command.command("addonname.cmd")装饰器声明带类型注解的命令，即可在交互工具命令提示符（:myaddon.inc）中调用并享有 Tab 补全，还支持@focus、~d google.com等 flow 选择器作为参数，详见 commands 文档 与 mitmproxy/command.py。

6. 小结

• mitmproxy 的 addon 机制是"事件 + 选项 + 命令"三位一体的插件体系，mitmproxy 自身大量功能（内置 addon、onboarding webapp 等）同样构建于其上，源码见 mitmproxy/addons/；
• 最小 addon 只需一个实现事件方法的类加addons = [Counter()]全局列表，或用缩写语法把def request(flow): ...直接写在模块顶层，mitmdump -s ./anatomy.py即可加载验证；
• -s加载的脚本由 mitmproxy/addons/script.py 中的Script.watcher每 1 秒轮询 mtime 实现热重载：导入/configure/running阶段的错误记日志且旧版本保持未注册，事件处理函数内的错误只记日志不卸载；
• 测试 addon 的推荐路径是"导入—实例化—直接调用钩子"，复杂场景借助 test/mitmproxy/addons 中的taddons.context()/tflow.tflow()等 helper（注意其 API 不保证稳定）。

【免费下载链接】mitmproxyAn interactive TLS-capable intercepting HTTP proxy for penetration testers and software developers.项目地址: https://gitcode.com/GitHub_Trending/mi/mitmproxy