目录

基于 mitmproxy 的手机 Proto 埋点抓包工具

一、背景

在工作中,我们埋点数据的调试和验证一直是一个痛点。传统的抓包方案(如 Charles、Fiddler)虽然能够抓取 HTTPS 流量,但对于使用 Protocol Buffers 编码的埋点数据,只能看到二进制内容,无法直接阅读和解析。

公司内部已有的埋点测试平台主要面向 JSON 格式的埋点设计,对于 Proto/Protobuf 编码的埋点数据无法解析和展示。这意味着:

  1. 平台无法识别 Proto 埋点的 Event ID、用户信息、业务参数等关键数据
  2. 无法进行埋点准确性验证和回归测试
  3. 测试人员只能依赖手动抓包 + 手动解析的方式,效率极低
  4. 产品验证埋点非常麻烦

因此,我开发了一个基于 mitmproxy 的手机 Proto 埋点抓包工具,能够自动解析 Proto 格式的埋点数据,并通过 Web UI 实时展示,填补了公司埋点测试平台在 PB 格式支持上的空白。由于AI的帮助,不到4小时就开发完成。

二、整体方案

flowchart TD A[手机 App] -->|HTTPS 请求 | B[mitmproxy<br/>8088 端口] B -->|DNS 劫持/Host 重定向 | C{目标环境} C -->|测试环境 | D[10.150.x.x] C -->|生产环境 | E[真实 IP] B -->|请求拦截 | F[EventDetector] F -->|Proto 解析 | G[埋点识别] G -->|分类 | H[pv/click/exposure/player] G -->|提取 | I[Event ID / Log ID] G -->|解析 | J[业务信息] H --> K[Web UI 展示<br/>8089 端口] I --> K J --> K K --> L[实时查看埋点] K --> M[扫码安装证书] K --> N[Host 环境切换]

三、核心实现

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 切换存在局限性:

  1. DNS 劫持可能被绕过:部分 App 会使用 HTTPDNS 或直接使用 IP 连接,绕过系统 DNS 解析
  2. DNS 缓存问题:手机系统或 App 内部可能缓存 DNS 结果,导致切换不生效
  3. HTTPS 证书校验:即使 DNS 解析到目标 IP,如果 SNI(Server Name Indication)不匹配,可能导致证书校验失败

因此,设计了双层保障机制:

flowchart TD A[App 发起请求] --> B{DNS 劫持层} B -->|正常域名解析 | C[返回配置 IP] B -->|HTTPDNS/直连 IP | D[跳过 DNS 劫持] C --> E{TCP 连接层} D --> E E -->|目标域名在配置中 | F[重定向到配置 IP] E -->|目标域名不在配置中 | G[连接原始 IP] F --> H[最终到达目标服务器] G --> H

第一层: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 配置后,工具会自动重启以应用新配置,无需重新配置手机代理。

五、使用流程

flowchart TD A[启动工具] --> B[打开 Web UI] B --> C[扫码安装证书] C --> D[手机配置代理<br/>IP: 本机 IP 端口:8088] D --> E[启用完全信任证书] E --> F[开始使用 App] F --> G[实时查看埋点数据] G --> H{需要切换环境?} H -->|是 | I[Web UI 切换 Host] H -->|否 | G I --> F

1. 启动工具

uv run app.py

工具会自动:

  • 启动 mitmproxy(8088 端口)
  • 启动 Web UI(8089 端口)
  • 打开浏览器

2. 手机配置

  1. 扫描下方二维码或点击链接下载证书
  2. 在手机设置中安装证书并启用完全信任
  3. 配置 WiFi 代理:主机名 = 本机 IP,端口 = 8088

3. 开始抓包

打开 App 后,Web UI 会实时显示埋点数据:

  • 按类型筛选:pv / click / exposure / player / custom
  • 搜索 Event ID
  • 查看埋点详情:用户信息、运行时信息、业务信息

六、总结

这个工具通过 mitmproxy 的插件机制,实现了:

  1. 自动 Proto 解析:无需手动分析二进制数据
  2. 实时展示:埋点数据秒级可见
  3. Host 无感切换:DNS 劫持 + TCP 重定向
  4. 友好 UI:扫码配证书、一键切换环境

相比传统方案,大大降低了埋点测试的成本,将原本需要 10+ 分钟的手动操作变为自动完成。