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、粘性 Cookiesticky 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 主体通过三条途径交互事件Eventsaddon 通过响应事件钩子切入并改变 mitmproxy 的行为钩子签名文档见 event-hooks 文档 及 docs/src/scripts/api-events.py 生成的 API 事件列表选项Optionsaddon 通过全局选项存储被配置选项可写在配置文件里、由用户在交互工具中实时修改或通过命令行传入详见 options 文档命令Commandsaddon 可以暴露命令供用户直接调用或在 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()doneaddon 被移除或 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(Weve seen %d flows % self.num) addons [Counter()]这是一个统计所见 flow此处更精确地说是 HTTP 请求数量的 addon每看到一个新 flow 就自增并记录日志输出可以在交互工具的事件日志或mitmdump的控制台中看到。用你顺手的 mitmproxy 工具加载它即可验证各工具使用的加载参数完全一致mitmdump -s ./anatomy.py关于这段代码官方文档强调三点这里逐一对应到源码实现1addons全局列表。mitmproxy 会拾取模块中的addons全局列表并把它里面的对象逐一加载进 addon 机制。这一点在 Script.addons 属性中得到印证Script类把脚本模块本身作为 addon 注册其addons属性返回[self.ns]随后由AddonManager递归展开。2addon 就是普通对象。本例中 addon 是Counter的实例。对象注册进管理器后其名称由 addonmanager.py 的_get_name决定取实例的.name属性没有则回退为类名小写——所以Counter实例在ctx.master.addons中可以被get(counter)取到。3request方法就是一个事件。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_hooksaddonmanager.py#L243-L262还支持异步钩子函数async def方法会被 await并特意容忍与钩子同名的模块导入比如 addon 里from mitmproxy import log与log钩子撞名。2.1 注册流程register、LoadHook与Loader当Counter实例被注册时AddonManager.register 依次做了这些事用traverse()递归收集 addon 及其子 addon并对旧版 API如clientconnect、add_log等已移除/废弃的钩子打印迁移警告名称冲突检测同名 addon 已存在则抛出AddonManagerError构造Loader并向 addon 发送 LoadHookload(loader)方法在其中通过loader.add_option(...)与loader.add_command(...)声明自己的选项与命令把 addon 及其子 addon 登记进self.lookup并收集其command.command装饰的命令。Loader.add_option的签名见 Loader.add_optionname、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#L74Script.__init__在reloadTrue时通过asyncio_utils.create_task(self.watcher(), ...)启动一个后台监视任务script.py#L92-L98watcher() 每秒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或旧模块已被 removeloadscript()中safecall包裹导入注册configure异常走script_error_handler事件处理函数request、response…内仅记录日志addon 不被卸载AddonManager.trigger_event的safecall()兜底修复错误后再次保存文件watcher 会在下一轮轮询重试加载。对应的行为测试在 test/mitmproxy/addons/test_script.pytest_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, loaderloader) ...每个脚本被赋予__mitmproxy_script__.文件名的唯一模块名重载前先sys.modules.pop清掉同名旧模块保证重新执行脚本所在目录临时插入sys.path头部finally中还原使脚本内部可以import同目录的辅助模块若运行的是 PyInstaller 冻结的二进制ImportError时会在错误信息中追加提示二进制自带独立 Python 环境若 addon 需要额外依赖请从 PyPI 安装 mitmproxy见 script.py#L42-L52。脚本本身也是一个元 addonScript类与 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, reloadTrue)列表顺序变化只重排、不重建实例避免不必要的重新初始化每个Script实例的addons属性返回被监视的脚本模块形成ScriptLoader → Script → 用户脚本模块的 addon 链事件沿链逐层分发此外还注册了一个 script.run 命令对指定 flows 回放模拟各生命周期事件注意load事件不会被调用方便离线调试脚本。4.2 测试 Addon因为 addon 就是普通 Python 文件最简单的单元测试方式就是官方文档给出的在测试中导入模块、实例化 addon、直接调用事件处理函数。例如针对 2 节的Counterfrom 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.pytaddons.context()提供一个完整的临时 mitmproxy 运行上下文master、addons、options用于验证 addon 注册、选项声明等集成行为tflow.tflow(respTrue)快速构造带响应的模拟 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 addheader1000会得到 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考