4、mitmproxy脚本开发(一):request与response钩子函数、修改请求头与响应体、实战:给所有API请求添加Token
好,咱们今天正式进入 mitmproxy 脚本开发的第一课。
说实话,mitmproxy 这个工具我用了好多年。从最早做接口测试时手动抓包,到后来做自动化测试平台的数据录制与回放,它一直是我的得力助手。你想想看,一个能让你在请求发出前、响应返回后都能插一脚的工具,能干的事情太多了。
4.1 理解钩子函数:脚本的灵魂
mitmproxy 的脚本机制,说白了就是「事件驱动」。你写一个 Python 脚本,定义好某些「钩子函数」,mitmproxy 在运行过程中遇到对应的事件,就会自动调用你的函数。
我个人习惯把钩子函数分成两大类:
- 请求钩子:在请求发出去之前触发。你可以修改请求头、请求体,甚至直接拦截请求返回假数据。
- 响应钩子:在响应返回给客户端之前触发。你可以修改响应体、添加响应头,或者记录响应数据。
嗯,这里要注意:钩子函数的执行顺序是固定的。请求先经过 request 钩子,然后发到服务器,服务器返回后,再经过 response 钩子,最后才到客户端。
核心概念:钩子函数就是你的「拦截点」。你在这些点上写代码,就能控制流量。
4.2 request 钩子:在请求发出前动手
先看一个最简单的例子。假设我想给所有经过代理的请求都打上日志:
# add_log.py
def request(flow):
# flow 对象包含了整个请求-响应的上下文
print(f"请求方法: {flow.request.method}")
print(f"请求URL: {flow.request.pretty_url}")
print(f"请求头: {dict(flow.request.headers)}")
运行方式很简单:
mitmproxy -s add_log.py
或者用 mitmweb(我更喜欢这个,有界面):
mitmweb -s add_log.py
我在项目中遇到过一个问题:有些请求的 URL 特别长,打印出来刷屏。后来我加了个过滤条件,只打印 API 接口的请求。这个后面会讲到。
4.3 response 钩子:在响应返回前动手
response 钩子的用法和 request 类似,只不过操作的对象变成了响应:
# modify_response.py
def response(flow):
# 修改响应头
flow.response.headers["X-Custom-Header"] = "Hello from mitmproxy"
# 修改响应体(注意:需要处理编码)
if "application/json" in flow.response.headers.get("content-type", ""):
text = flow.response.text
text = text.replace("old_value", "new_value")
flow.response.text = text
你可能会问:为什么要修改响应体?
我举个例子。有一次测试一个支付系统,后端返回的订单状态总是「待支付」,我想模拟「已支付」的状态来测试前端展示。用 mitmproxy 脚本,几行代码就搞定了。
小技巧:修改响应体时,记得先检查 content-type。如果是二进制数据(比如图片),用 flow.response.content 而不是 .text。
4.4 实战:给所有API请求添加Token
好,理论说完了,咱们直接上实战。这个场景太常见了:
- 你的被测系统需要 Token 鉴权
- 但录制数据时,Token 已经过期了
- 回放时,请求因为没有 Token 被服务器拒绝
解决方案?用 mitmproxy 脚本,在请求发出前自动加上 Token。
4.4.1 基础版:硬编码Token
# add_token_v1.py
def request(flow):
# 只给 API 请求加 Token
if "/api/" in flow.request.pretty_url:
flow.request.headers["Authorization"] = "Bearer your_token_here"
这个版本简单粗暴,但有个问题:Token 写死在代码里,换一次 Token 就要改脚本。
4.4.2 进阶版:从配置文件读取Token
# add_token_v2.py
import json
import os
def load_token():
config_path = os.path.join(os.path.dirname(__file__), "config.json")
with open(config_path, "r") as f:
config = json.load(f)
return config.get("token", "")
TOKEN = load_token()
def request(flow):
if "/api/" in flow.request.pretty_url:
flow.request.headers["Authorization"] = f"Bearer {TOKEN}"
对应的 config.json:
{
"token": "your_actual_token_here"
}
这样就好多了。换 Token 时改配置文件就行,不用动脚本。
4.4.3 高级版:自动刷新Token
我曾经遇到一个更棘手的情况:Token 有效期只有 15 分钟。手动换 Token 根本来不及。
我的做法是:在脚本里加一个定时器,每隔 10 分钟调用一次登录接口,拿到新 Token 后更新全局变量。
# add_token_v3.py
import threading
import requests
TOKEN = None
TOKEN_URL = "https://your-api.com/auth/login"
USERNAME = "test_user"
PASSWORD = "test_pass"
def refresh_token():
global TOKEN
resp = requests.post(TOKEN_URL, json={
"username": USERNAME,
"password": PASSWORD
})
if resp.status_code == 200:
TOKEN = resp.json().get("token")
print(f"Token refreshed: {TOKEN[:20]}...")
# 每10分钟刷新一次
threading.Timer(600, refresh_token).start()
# 启动时立即刷新一次
refresh_token()
def request(flow):
global TOKEN
if "/api/" in flow.request.pretty_url and TOKEN:
flow.request.headers["Authorization"] = f"Bearer {TOKEN}"
注意:这个方案有个坑——requests.post 本身也会经过 mitmproxy 代理。如果你的脚本里也加了 Token,就会形成死循环。解决办法是在请求登录接口时跳过代理,或者加一个白名单判断。
4.5 本章知识体系
为了让你更直观地理解整个流程,我画了一张图:
4.6 避坑指南
最后,分享几个我踩过的坑:
| 问题 | 现象 | 解决方案 |
|---|---|---|
| 修改请求头不生效 | 服务器收到的还是原始请求头 | 检查是否在 request 钩子中修改,而不是 response |
| 修改响应体乱码 | 中文变成乱码 | 用 flow.response.text 而不是 .content,并指定编码 |
| 脚本报错导致代理崩溃 | mitmproxy 直接退出 | 在钩子函数里加 try...except,至少保证不崩溃 |
| Token 刷新死循环 | 脚本不断调用登录接口 | 登录请求跳过代理,或者加白名单判断 |
嗯,今天就先到这里。这一章的内容其实已经覆盖了 mitmproxy 脚本开发最核心的两个钩子。你先把这几个例子跑通,下一章我们会聊更高级的用法——比如怎么用 mitmproxy 做数据录制。
课后练习:试着写一个脚本,给所有 /api/ 路径的 POST 请求添加一个自定义请求头 X-Source: mitmproxy,并在响应中打印出修改后的请求头。
可可精品资料库