基于 mitmproxy 的手机 Proto 埋点抓包工具
一、背景
在工作中,我们埋点数据的调试和验证一直是一个痛点。传统的抓包方案(如 Charles、Fiddler)虽然能够抓取 HTTPS 流量,但对于使用 Protocol Buffers 编码的埋点数据,只能看到二进制内容,无法直接阅读和解析。
公司内部已有的埋点测试平台主要面向 JSON 格式的埋点设计,对于 Proto/Protobuf 编码的埋点数据无法解析和展示。这意味着:
- 平台无法识别 Proto 埋点的 Event ID、用户信息、业务参数等关键数据
- 无法进行埋点准确性验证和回归测试
- 测试人员只能依赖手动抓包 + 手动解析的方式,效率极低
- 产品验证埋点非常麻烦
因此,我开发了一个基于 mitmproxy 的手机 Proto 埋点抓包工具,能够自动解析 Proto 格式的埋点数据,并通过 Web UI 实时展示,填补了公司埋点测试平台在 PB 格式支持上的空白。由于AI的帮助,不到4小时就开发完成。
二、整体方案
三、核心实现
1. mitmproxy 插件架构
mitmproxy 是一个强大的中间人代理工具,支持通过 Python 插件扩展功能。插件可以通过 hook 不同的事件点来拦截和修改流量:
class EventDetector:
def load(self, loader):
"""加载时配置 host 映射规则"""
self.host_map = ui.get_host_config()
def dns_request(self, flow: dns.DNSFlow):
"""DNS 请求劫持 - 第一层 host 切换"""
for q in flow.request.questions:
if q.name in self.host_map:
target_ip = self.host_map[q.name]
flow.response = flow.request.succeed(
answers=[
dns.ResourceRecord(
name=q.name,
type=1,
class_=1,
ttl=15,
data=dns.IPv4Address(target_ip).packed,
)
]
)
def server_connect(self, data: server_hooks.ServerConnectionHookData):
"""TCP 连接时重定向 - 第二层 host 切换"""
host = data.server.address[0]
if host in self.host_map:
new_ip = self.host_map[host]
data.server.address = (new_ip, data.server.address[1])
def request(self, flow: http.HTTPFlow):
"""HTTP 请求拦截 - 核心逻辑"""
if self.is_event(flow.request.pretty_url):
raw_body = flow.request.get_content()
parsed = parse_report(raw_body)
# 处理解析后的埋点数据...2. Proto 消息解析
埋点数据使用 Protocol Buffers 编码,由于没有 .proto 文件,需要通过逆向分析手动解析。我直接让AI阅读h5代码,经过约30分钟的调试,成功逆向分析,将PB数据解析出来。
3. Web UI 实时展示
Web UI 使用简单的 HTTP Server 实现,提供两个接口:
/events- 返回最新 100 条埋点数据/cert- 提供证书下载
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/events":
self.send_json(EVENTS[-100:])
elif self.path == "/cert":
# 返回 mitmproxy CA 证书
path = os.path.expanduser("~/.mitmproxy/mitmproxy-ca-cert.cer")
with open(path, "rb") as f:
content = f.read()
self.send_response(200)
self.send_header("Content-Type", "application/x-x509-ca-cert")
self.end_headers()
self.wfile.write(content)前端使用原生 JavaScript 实现实时刷新和筛选:
async function load() {
let res = await fetch('/events');
allEvents = await res.json();
renderList();
}
// 每秒刷新一次
setInterval(load, 1000);4. Host 切换核心实现
工具支持测试环境和默认两种 Host 配置,通过DNS 劫持和TCP 重定向两层机制实现流量导向。
为什么需要两层?
在实际使用中,发现单一层的 Host 切换存在局限性:
- DNS 劫持可能被绕过:部分 App 会使用 HTTPDNS 或直接使用 IP 连接,绕过系统 DNS 解析
- DNS 缓存问题:手机系统或 App 内部可能缓存 DNS 结果,导致切换不生效
- HTTPS 证书校验:即使 DNS 解析到目标 IP,如果 SNI(Server Name Indication)不匹配,可能导致证书校验失败
因此,设计了双层保障机制:
第一层:DNS 劫持 - 在 DNS 请求阶段直接返回配置的 IP 地址,适用于大多数正常域名解析的场景。
第二层:TCP 重定向 - 在 TCP 连接建立时,检查目标域名是否在配置中,如果在则重定向到配置的 IP。这一层可以捕获:
- 使用 HTTPDNS 的应用
- DNS 已缓存的情况
- 直接使用 IP 连接但需要重定向的场景
两层结合,确保 Host 切换在各种场景下都能生效。
第一层实现:DNS 劫持
def dns_request(self, flow: dns.DNSFlow):
"""DNS 请求劫持 - 第一层 host 切换"""
if not self.host_map:
return
for q in flow.request.questions:
if q.name in self.host_map:
target_ip = self.host_map[q.name]
# 构造虚假 DNS 响应,直接返回配置的 IP
flow.response = flow.request.succeed(
answers=[
dns.ResourceRecord(
name=q.name,
type=1, # A 记录
class_=1,
ttl=15, # 短 TTL,便于快速生效
data=dns.IPv4Address(target_ip).packed,
)
]
)
print(f"🌐 DNS 劫持:{q.name} -> {target_ip}")第二层实现:TCP 重定向
def server_connect(self, data: server_hooks.ServerConnectionHookData):
"""TCP 连接时重定向 - 第二层 host 切换"""
if not self.host_map:
return
host = data.server.address[0]
port = data.server.address[1]
# 如果已经是 IP 地址,跳过
if self._is_ipaddress(host):
return
if host in self.host_map:
new_ip = self.host_map[host]
print(f"🔀 Host 重定向:{host} -> {new_ip}:{port}")
# 修改连接地址,重定向到配置的 IP
data.server.address = (new_ip, port)Host 配置管理
# Host 配置示例(以通用域名为例)
HOST_CONFIGS = {
"uat": {
"grpc.example.com": "10.150.10.11",
"api.example.com": "10.150.10.22",
# ... 更多域名映射
},
"default": {}
}用户在 Web UI 切换 Host 配置后,工具会自动重启以应用新配置,无需重新配置手机代理。
五、使用流程
1. 启动工具
uv run app.py工具会自动:
- 启动 mitmproxy(8088 端口)
- 启动 Web UI(8089 端口)
- 打开浏览器
2. 手机配置
- 扫描下方二维码或点击链接下载证书
- 在手机设置中安装证书并启用完全信任
- 配置 WiFi 代理:主机名 = 本机 IP,端口 = 8088
3. 开始抓包
打开 App 后,Web UI 会实时显示埋点数据:
- 按类型筛选:pv / click / exposure / player / custom
- 搜索 Event ID
- 查看埋点详情:用户信息、运行时信息、业务信息
六、总结
这个工具通过 mitmproxy 的插件机制,实现了:
- 自动 Proto 解析:无需手动分析二进制数据
- 实时展示:埋点数据秒级可见
- Host 无感切换:DNS 劫持 + TCP 重定向
- 友好 UI:扫码配证书、一键切换环境
相比传统方案,大大降低了埋点测试的成本,将原本需要 10+ 分钟的手动操作变为自动完成。