利用AI一键开启proto全自动明文时代

文章首发奇安信攻防社区：https://forum.butian.net/ai_security/90

本公众号技术文章仅供参考！

文章仅用于学习交流，请勿利用文章中的技术对任何计算机系统进行入侵操作。利用此文所提供的信息而造成的直接或间接后果和损失，均由使用者本人负责。

这个东西如果你现在没碰上，但是哪一天碰上了，我相信你会回来感谢我的，毕竟之前很长一段时间，我被它折磨够呛

0x01 protobuf介绍

0x011 protobuf概述

先跟没有接触过protobuf的伙伴们介绍一下它，它是Google公司开发的一种数据描述语言，主打的就是轻便高效，唯一可能比较难受一点的就是网络层可读性较差；目前很多大公司微服务之间互相的RPC（远程过程调用）都是基于protobuf，而且一些游戏或者实时音视频聊天用的也比较多。

优点：

数据序列化后体积非常小
序列化反序列化速度很快
支持跨平台
兼容性高

缺点：

可读性差

由于游戏和实时音视频聊天，都是追求0卡顿，所以用protobuf比较多，而且属于它体积小，另一方面还能够省电，所以它经常还跟wss进行搭配

0x012 proto格式

先看一眼如果你抓包遇到什么样子形式的东西，你可以判断它是protobuf，如下图，在burp抓包里面当你看到接口的数据是乱码或者乱码了但是没完全乱码还是能看出点东西的这大概率就是了。

那么既然我们都不认识，客户端或者服务器肯定更不认识，那么客户端如何发出来的，服务器又如何读懂了，这就离不开.proto文件了，如果把上面那个流量看成加密后的，那么这个文件就相当于密码本了，有它你就能读懂了，就这么简单，但是其实在代码层面并没有直接使用它，而是需要不同语言将其转换成对应语言的

比如服务端用的java，服务器用的golang，那么 .proto 文件会被编译器（protoc）分别翻译成 Java 的 .java 类文件和 Go 的 .pb.go 结构体文件。Java 端调用 build() 方法将对象压成二进制流，Go 端则通过 Unmarshal() 按照同样的编号索引将字节还原。虽然两端语言不同，但由于它们共享了同一份 .proto 定义的字段编号（Tag）和数据类型。

格式如所示：

//指定版本
syntax = "proto3";

//导出包名
package packet;

// ============================================================
// 外层包装 (所有 WS 帧的顶层结构)
// ============================================================

message Packet {
  int32  msg_no    = 1;  // MsgNo 枚举值
  bytes  data      = 2;  // 内层 protobuf 序列化后的字节
  int64  unix_milli = 3; // 发送时间戳 (毫秒)
  string user_id   = 4;
}

//定义枚举 (Enum)：用来表示一组固定的选项（可选）
enum RoleType {
  ROLE_UNKNOWN = 0; // 未知职业
  WARRIOR      = 1; // 战士
  MAGE         = 2; // 法师
  ARCHER       = 3; // 射手
}

enum Status {
  OFFLINE = 0; // 离线
  ONLINE  = 1; // 在线
  BUSY    = 2; // 忙碌
}

//定义消息结构 (Message)
message Player {
  int32 id = 1;              // 玩家ID
  string name = 2;           // 玩家昵称
  RoleType role = 3;         // 引用上面的枚举：玩家职业
  Status status = 4;         // 引用上面的枚举：在线状态
  repeated string inventory = 5; // 字符串数组：背包物品
}

注意： protobuf里面数字类型并不是用的16进制直接表示的数据，而是用的varint，所以换算的时候要注意一下，我一开始就以为16进制表示怎么也找不到值，所以这里不细说，仅提醒一下。

0x02 方案结构

0x021 整体方案

说真的在Burp中修改protobuf而且还是Websocket连接，简直是噩梦，正常情况下你需要将16进制的数据利用网站或工具进行手动解析，比如https://protobuf-decoder.netlify.app/ ，解析分析完，你还得对着一堆16进制去修改，非常不方便，那么目标就是为了方便渗透测试能让burp中看到明文的值，以及修改完还能正常发送，我查了关于protobuf的burp插件，用了也没效果而且文章看的我一脸蒙圈

上面那个结构很难不让我联想到全自动加解密，所以最终的方案也类似，两个mitm脚本A和B，终端-A-Burp-B-Server，如下图，这样就能保证Burp里面的流量是能看懂的了：

0x022 proto获取

不管什么APP（除了微信小程序），页面上的H5其实就是我们正常WEB站的前端那些东西，APP里面要加载多会以一个zip包的形式，比如我这里以一个游戏举例，点击游戏列表在burp的记录中会找到对应请求，响应里面找到对应的zip压缩包

找到包后下载解压就得到了源码

既如此就可以从js代码中提取对应的.proto，不过你会自己手动去找吗，这活我是懒得干，所以直接扔给AI让它给我找到并生成

这里有个坑就是AI其实99%都会找正确，但是可能数据格式上会发生那1%的错误，虽然占比小，但是一旦错了会影响后面的流程，所以这里要用提示词加强一下AI自我检查和反省

0x023 脚本开发

首先脚本还是用的mitmproxy规则的脚本，利用python完全可以有截获Websocket的能力，但首先需要将.proto生成python文件

#pip install grpcio-tools

python -m grpc_tools.protoc -I. --python_out=. .\game.proto

然后就生成了对应的python文件了，这个文件有点乱其实完全不用管，我这里大概提一嘴，它大概得作用就是将 .proto 定义的抽象协议规则“具象化”为 Python 的类与方法，让别的 py 文件引入的时候会在全局内存中动态构建并使用

脚本A：

import json
import base64
from mitmproxy import http, ctx
import game_pb2 as pb
from google.protobuf.json_format import MessageToDict, ParseDict

#这里定义了第一个字典，键是msgid，值是对应的类
MSG_MAP_RAW = {
0: getattr(pb, 'Unknown', None),
17: getattr(pb, 'message1', None),
18: getattr(pb, 'message2', None)
}
#去除值为None的
MSG_MAP = {k: v for k, v in MSG_MAP_RAW.items() if v isnotNone}

#根据msgid和data将二进制数据转成python字典，识别不出来的话base64兜底
defdecode_payload(msg_no, data_bytes):
ifnot data_bytes: return {}
if msg_no in MSG_MAP:
try:
            inner = MSG_MAP[msg_no]()
            inner.ParseFromString(data_bytes)
return MessageToDict(inner, preserving_proto_field_name=True)
except Exception as e:
            ctx.log.error(f"❌ 解析 payload 失败 (MsgNo: {msg_no}): {e}")
return {"__raw_b64__": base64.b64encode(data_bytes).decode('utf-8')}

#将字典转成二进制
defencode_payload(msg_no, payload_dict):
if"__raw_b64__"in payload_dict:
return base64.b64decode(payload_dict["__raw_b64__"])
if msg_no in MSG_MAP:
        inner = MSG_MAP[msg_no]()
        ParseDict(payload_dict, inner)
return inner.SerializeToString()
returnb""

classScriptA:
#拦截websocket
defwebsocket_message(self, flow: http.HTTPFlow):
#判断确保流量是最新的ws
ifnot flow.websocket: return
        message = flow.websocket.messages[-1]
#判断消息是来源于客户端发往服务器的
if message.from_client:
#判断是否是二进制，如果是转成json
ifisinstance(message.content, bytes):
try:
                    outer = pb.Packet()
                    outer.ParseFromString(message.content)
                    burp_json = {
"direction": "request",
"msg_no": outer.msg_no,
"unix_milli": outer.unix_milli,
"user_id": outer.user_id,
"payload": decode_payload(outer.msg_no, outer.data)
                    }
                    message.content = json.dumps(burp_json, ensure_ascii=False, indent=2).encode("utf-8")
                    ctx.log.info(f"✅ [App->Burp] 请求解包成功 | MsgNo: {outer.msg_no}")
except Exception as e:
                    ctx.log.error(f"❌ [App->Burp] 外壳解析崩溃: {e}")
else:
#如果消息是从服务器发往客户端，这里要将json还原成二进制
try:
                burp_json = json.loads(message.content.decode("utf-8"))
                outer = pb.Packet()
                outer.msg_no = burp_json.get("msg_no", 0)
                outer.unix_milli = burp_json.get("unix_milli", 0)
                outer.user_id = burp_json.get("user_id", "")
                outer.data = encode_payload(outer.msg_no, burp_json.get("payload", {}))
                message.content = outer.SerializeToString()
                ctx.log.info(f"✅ [Burp->App] 响应还原成功 | MsgNo: {outer.msg_no}")
except Exception as e:
                ctx.log.error(f"❌ [Burp->App] JSON还原PB崩溃: {e}")

addons = [ScriptA()]

用法：

#端口启动到8888，上游代理到Burp
mitmdump -p 8888 --mode upstream:http://192.168.117.73:8080 --ssl-insecure -s script_a.py

脚本B（跟A类似，逻辑反过来）：

import json
import base64
from mitmproxy import http, ctx
import fishing_game_pb2 as pb
from google.protobuf.json_format import MessageToDict, ParseDict

MSG_MAP_RAW = {
0: getattr(pb, 'Unknown', None),
17: getattr(pb, 'message1', None),
18: getattr(pb, 'message2', None)
}
# 过滤掉 proto 里未定义的类
MSG_MAP = {k: v for k, v in MSG_MAP_RAW.items() if v isnotNone}

defdecode_payload(msg_no, data_bytes):
ifnot data_bytes: return {}
if msg_no in MSG_MAP:
try:
            inner = MSG_MAP[msg_no]()
            inner.ParseFromString(data_bytes)
return MessageToDict(inner, preserving_proto_field_name=True)
except Exception as e:
            ctx.log.error(f"❌ 解析 payload 失败 (MsgNo: {msg_no}): {e}")
return {"__raw_b64__": base64.b64encode(data_bytes).decode('utf-8')}

defencode_payload(msg_no, payload_dict):
if"__raw_b64__"in payload_dict:
return base64.b64decode(payload_dict["__raw_b64__"])
if msg_no in MSG_MAP:
        inner = MSG_MAP[msg_no]()
        ParseDict(payload_dict, inner)
return inner.SerializeToString()
returnb""

classScriptB:
defwebsocket_message(self, flow: http.HTTPFlow):
ifnot flow.websocket: return
        message = flow.websocket.messages[-1]

if message.from_client:
# Burp -> Server
try:
                burp_json = json.loads(message.content.decode("utf-8"))
                outer = pb.Packet()
                outer.msg_no = burp_json.get("msg_no", 0)
                outer.unix_milli = burp_json.get("unix_milli", 0)
                outer.user_id = burp_json.get("user_id", "")
                outer.data = encode_payload(outer.msg_no, burp_json.get("payload", {}))
                message.content = outer.SerializeToString()
                ctx.log.info(f"✅ [Burp->Server] 发往服务器 | MsgNo: {outer.msg_no}")
except Exception as e:
                ctx.log.error(f"❌ [Burp->Server] 还原并发送给服务器崩溃: {e}")
else:
ifisinstance(message.content, bytes):
try:
                    outer = pb.Packet()
                    outer.ParseFromString(message.content)
                    burp_json = {
"direction": "response",
"msg_no": outer.msg_no,
"unix_milli": outer.unix_milli,
"user_id": outer.user_id,
"payload": decode_payload(outer.msg_no, outer.data)
                    }
                    message.content = json.dumps(burp_json, ensure_ascii=False, indent=2).encode("utf-8")
                    ctx.log.info(f"✅ [Server->Burp] 响应解包成功 | MsgNo: {outer.msg_no}")
except Exception as e:
                    ctx.log.error(f"❌ [Server->Burp] 服务器响应解壳崩溃: {e}")

addons = [ScriptB()]

用法：

mitmdump -p 9999 --ssl-insecure -s script_b.py

0x024 全自动脚本开发

上述两个脚本其实完全可以半通用了，比如服务这时候换了，是另个一个游戏了，那么它的.proto肯定不同了，然后只需要根据proto重新生成pb2，然后将两个mitm脚本上方的MSG_MAP_RAW根据实际情况替换就可以了，那既然如此，完全可以写一个底座脚本全自动生成pb2，然后生成两个mitm脚本，最后启动代理了

auto_proxy.py

import os
import sys
import subprocess
import re
import time

# 核心配置
PROTO_FILE = "game.proto"#proto 文件名
BURP_PROXY = "192.168.1.11:8080"#Burp Suite 的代理地址
PORT_A = 8888# 节点A 端口
PORT_B = 9999# 节点B 端口
ENVELOPE = "Packet"#统一的外壳名称


# ==========================================
#将.proto转成python
defcompile_proto(proto_filename):
print(f"[*] 正在编译 Protobuf 文件: {proto_filename}")
ifnot os.path.exists(proto_filename):
print(f"[!] 找不到文件 {proto_filename}，请检查路径！")
        sys.exit(1)

    cmd = [sys.executable, "-m", "grpc_tools.protoc", "-I.", f"--python_out=.", proto_filename]
    result = subprocess.run(cmd, capture_output=True, text=True)

if result.returncode != 0:
print("[!] Protoc 编译失败:\n", result.stderr)
        sys.exit(1)

    pb2_name = proto_filename.replace(".proto", "_pb2")
print(f"[+] 编译成功，生成 {pb2_name}.py")
return pb2_name

#利用正则匹配消息号和对应类名，用于生成MSG_MAP_RAW
defextract_mappings(proto_filename):
withopen(proto_filename, "r", encoding="utf-8") as f:
        content = f.read()

    enum_match = re.search(r'enum\s+MsgNo\s*\{([^}]+)\}', content)
ifnot enum_match: return"{}"

    msg_entries = []
for line in enum_match.group(1).split('\n'):
# 匹配: MsgNo_FishingLoginC2S = 655386;
match = re.search(r'(MsgNo_([a-zA-Z0-9_]+))\s*=\s*(\d+)', line)
ifmatch:
            _, msg_name, msg_id = match.groups()
            msg_entries.append(f"    {msg_id}: getattr(pb, '{msg_name}', None)")
return"{\n" + ",\n".join(msg_entries) + "\n}"


defgenerate_scripts(pb2_name, map_str):
print("[*] 正在生成新版代理脚本 c_burp.py 和 s_server.py...")

    base_template = f"""import json
import base64
from mitmproxy import http, ctx
import {pb2_name} as pb
from google.protobuf.json_format import MessageToDict, ParseDict

MSG_MAP_RAW = {map_str}
# 过滤掉 proto 里未定义的类
MSG_MAP = {{k: v for k, v in MSG_MAP_RAW.items() if v is not None}}

def decode_payload(msg_no, data_bytes):
    if not data_bytes: return {{}}
    if msg_no in MSG_MAP:
        try:
            inner = MSG_MAP[msg_no]()
            inner.ParseFromString(data_bytes)
            return MessageToDict(inner, preserving_proto_field_name=True)
        except Exception as e:
            ctx.log.error(f"❌ 解析 payload 失败 (MsgNo: {{msg_no}}): {{e}}")
    return {{"__raw_b64__": base64.b64encode(data_bytes).decode('utf-8')}}

def encode_payload(msg_no, payload_dict):
    if "__raw_b64__" in payload_dict:
        return base64.b64decode(payload_dict["__raw_b64__"])
    if msg_no in MSG_MAP:
        inner = MSG_MAP[msg_no]()
        ParseDict(payload_dict, inner)
        return inner.SerializeToString()
    return b""
"""

    script_a = base_template + f"""
class ScriptA:
    def websocket_message(self, flow: http.HTTPFlow):
        if not flow.websocket: return
        message = flow.websocket.messages[-1]

        if message.from_client:
            # App -> Burp
            if isinstance(message.content, bytes):
                try:
                    print('分析')
                    outer = pb.{ENVELOPE}()
                    outer.ParseFromString(message.content)
                    burp_json = {{
                        "direction": "request",
                        "msg_no": outer.msg_no,
                        "unix_milli": outer.unix_milli,
                        "user_id": outer.user_id,
                        "payload": decode_payload(outer.msg_no, outer.data)
                    }}
                    message.content = json.dumps(burp_json, ensure_ascii=False, indent=2).encode("utf-8")
                    ctx.log.info(f"✅ [App->Burp] 请求解包成功 | MsgNo: {{outer.msg_no}}")
                except Exception as e:
                    ctx.log.error(f"❌ [App->Burp] 外壳解析崩溃: {{e}}")
        else:
            # Burp -> App
            try:
                burp_json = json.loads(message.content.decode("utf-8"))
                outer = pb.{ENVELOPE}()
                outer.msg_no = burp_json.get("msg_no", 0)
                outer.unix_milli = burp_json.get("unix_milli", 0)
                outer.user_id = burp_json.get("user_id", "")
                outer.data = encode_payload(outer.msg_no, burp_json.get("payload", {{}}))
                message.content = outer.SerializeToString()
                ctx.log.info(f"✅ [Burp->App] 响应还原成功 | MsgNo: {{outer.msg_no}}")
            except Exception as e:
                ctx.log.error(f"❌ [Burp->App] JSON还原PB崩溃: {{e}}")

addons = [ScriptA()]
"""

    script_b = base_template + f"""
class ScriptB:
    def websocket_message(self, flow: http.HTTPFlow):
        if not flow.websocket: return
        message = flow.websocket.messages[-1]

        if message.from_client:
            # Burp -> Server
            try:
                burp_json = json.loads(message.content.decode("utf-8"))
                outer = pb.{ENVELOPE}()
                outer.msg_no = burp_json.get("msg_no", 0)
                outer.unix_milli = burp_json.get("unix_milli", 0)
                outer.user_id = burp_json.get("user_id", "")
                outer.data = encode_payload(outer.msg_no, burp_json.get("payload", {{}}))
                message.content = outer.SerializeToString()
                ctx.log.info(f"✅ [Burp->Server] 发往服务器 | MsgNo: {{outer.msg_no}}")
            except Exception as e:
                ctx.log.error(f"❌ [Burp->Server] 还原并发送给服务器崩溃: {{e}}")
        else:
            # Server -> Burp
            if isinstance(message.content, bytes):
                try:
                    outer = pb.{ENVELOPE}()
                    outer.ParseFromString(message.content)
                    burp_json = {{
                        "direction": "response",
                        "msg_no": outer.msg_no,
                        "unix_milli": outer.unix_milli,
                        "user_id": outer.user_id,
                        "payload": decode_payload(outer.msg_no, outer.data)
                    }}
                    message.content = json.dumps(burp_json, ensure_ascii=False, indent=2).encode("utf-8")
                    ctx.log.info(f"✅ [Server->Burp] 响应解包成功 | MsgNo: {{outer.msg_no}}")
                except Exception as e:
                    ctx.log.error(f"❌ [Server->Burp] 服务器响应解壳崩溃: {{e}}")

addons = [ScriptB()]
"""
withopen("c_burp.py", "w", encoding="utf-8") as f: f.write(script_a)
withopen("s_server.py", "w", encoding="utf-8") as f: f.write(script_b)
print("[+] 脚本生成完成。")


defstart_proxies():
print(f"[*] 启动节点 A (监听 {PORT_A}, 上游 -> {BURP_PROXY}) ...")
    proc_a = subprocess.Popen(
        ["mitmdump", "-p", str(PORT_A), "--mode", f"upstream:http://{BURP_PROXY}", "--ssl-insecure", "-s", "c_burp.py"])
    time.sleep(1)
print(f"[*] 启动节点 B (监听 {PORT_B}, 直接连接外网) ...")
    proc_b = subprocess.Popen(["mitmdump", "-p", str(PORT_B), "--ssl-insecure", "-s", "s_server.py"])
return proc_a, proc_b


if __name__ == "__main__":
    os.system('cls'if os.name == 'nt'else'clear')
print("🚀 Protobuf V2 (单字典外壳版) 主控系统启动")

    pb2_name = compile_proto(PROTO_FILE)
    map_str = extract_mappings(PROTO_FILE)
    generate_scripts(pb2_name, map_str)

    proc_a, proc_b = start_proxies()

try:
        proc_a.wait()
        proc_b.wait()
except KeyboardInterrupt:
print("\n[*] 正在安全终止进程...")
        proc_a.terminate()
        proc_b.terminate()

最后的效果如下：

如此这般后，测试就非常简单了

0x03 AI加持全自动

0x031 现有的问题

最后效果已经实现了，实际体验已经非常好了，但是其实前面的脚本并不能通用，而且前面提取protobuf还是得手动提取，那么我们的目标是做成通用的形式，所以一个非常好的方案就是做成skill

上面的内容有三点欠缺之处

第一点就是需要手动下载zip，然后提取.proto文件，这点其实实现起来很简单
主要是第二点，上面我没有提到，针对我上述实际遇到的例子，它的.proto有一个外层包装，就是下面这个，这是一个标准的信封模式，这里的Packet就是信封，由于 Proto 里的 Message 种类太多，通常情况我们不能为每个请求都维护一套独立的传输逻辑，所以引入了这个信封模式，当客户端与服务器传递消息时，先将具体的业务 Message序列化为二进制并塞进 Packet 的 data 字段（信封），再填好 msg_no 等通用元数据（面单），从而实现一套传输逻辑兼容成百上千种业务数据的消息分发机制

message Packet {
  int32  msg_no    = 1;  // MsgNo 枚举值
  bytes  data      = 2;  // 内层 protobuf 序列化后的字节
  int64  unix_milli = 3; // 发送时间戳 (毫秒)
  string user_id   = 4;
}

我们的代码中也用到了这个名称

虽然我们目前只用一个 Packet 来承载发送和接收的所有逻辑，但将信封拆分为 RequestPacket 和 ResponsePacket，这种模式也很常见，比如客户端->服务器用RequestPacket，服务器->客户端用ResponsePacket，这样就能很好实现发送与接收逻辑的精细化解耦，所以一旦遇到这种模式当前版本的代码就显得力不从心了，虽然还是可以通过代码的继续优化来实现两种模式区分，然后根据不同模式采取不用的编写子脚本的规则。

但是由于我比较懒，而且AI时代，我打算放弃部分老手艺活。

还有就是第三点了，正则的提取，extract_mappings函数里面正则提取也属于是一个变量，这个我们也要依靠AI的能力（感谢这个盛世）

0x032 第一版skill

大体思路就是，用户需要给出两个参数，一个是burp地址，另一个就是zip的链接，然后大模型会下载zip然后解压分析，提取出.proto文件，然后我预先留了一个auto_proxy.py的脚本在skill的资源目录，大模型会将这个文件拷贝到当前目录根据对应.proto进行分析修改（以一个成功的代码让它改绝对比让它从头写少很多麻烦事，就类似于文生图和图生图的区别），并且还会将整体逻辑总结到一个md文档用于辅助理解。

第一版skill我觉得已经将条件限制的很严格了，比如用户必须给出两个参数及burp地址和zip链接，以及为了防止它发散所以限制当它遇到错误的时候立即停止，不要胡乱猜测和尝试，同时不要改动skill的脚本而是拷贝到一份到当前目录下再修改等等，然后如下图所示AI总会给你一些“惊喜”

首先第一点就是这货我一开始告诉它参数了，然后竟然还让我提供？
第二点用于下载、提取、分析的脚本创建了一堆，最后产出眼花缭乱的
最坑的一点就是证书问题，问题来源于它直接帮我启动了最后的脚本，然后我的mitmproxy证书被更新了，导致原来手机中的不好用了，我还得重新配置

0x033 最终版skill

图就不贴了改动如下，针对参数问题，括号里是新增的

用户必须提供两个参数，缺一不可。如果缺少任何一个（分析用户是否给了两个参数值，用户可能不会非常严谨的说明，可能用户就给了个代理地址和zip链接过来，这个判断你不要太死板），立即中断并提醒用户补充

针对一堆过程文件

如果在分析过程中你需要依赖一些脚本文件以及下载一些zip包和解压文件，那么你需要在now目录下创建一个cache目录将那些工作中产生的文件放在里面（放进去就可以，工作完成不需要清理），保持now目录下干净整洁，最后now目录下的文件仅仅只有3个：proto文件、md说明文档、auto_proxy.py

针对自动运行增加如下限制

当你生成脚本后，一定不要自己去运行，这个交给用户自己在客户端运行

最后成果如下，一个新的游戏，用了这个skill，全部变成明文

0x04 使用方案

下载地址：https://github.com/likeNlong/ws-protobuf-proxy/blob/main/SKILL.md

这里在啰嗦一下使用方案

首先第一点证书！证书！证书！你要用python装好mitmproxy然后将生成的证书以及burp的证书在电脑和手机端都装上，当然也别忘记处理证书信任问题。
git clone到对应skills目录
引用skills告诉它两个参数，zip链接和burp地址
然后如果你自动授权那几乎就不用管了，最后会给你产出3个文件：MD说明文档、proto文件、最重要的auto_proxy.py
然后你自己在本地用python运行这个py脚本，会开启两个代理8888和9999
APP->8888端口-->（自动）Burp-->9999端口即可