用Python构建Frida RPC网关：Burp与移动端加密流量的深度交互方案
移动应用安全测试中，最令人头疼的莫过于遇到加密流量。当你发现HttpCanary能捕获请求而BurpSuite却束手无策时，传统解决方案往往需要在多个工具间频繁切换，效率低下。本文将带你用Python打造一个轻量级Frida RPC服务器，建立Burp与移动端加密流量之间的双向通信通道，实现实时请求/响应篡改的完整工作流。
1. 为什么需要自定义RPC网关
在金融类App测试中，我们常遇到三种典型场景：
- Burp无法直接捕获加密流量：应用采用自定义SSL证书或非标准端口
- HttpCanary可见但不可改：能查看加密报文但缺乏动态修改能力
- 加解密逻辑在Native层：Java层仅暴露接口，实际算法在so库中
传统解决方案存在明显局限：
| 工具 | 抓包能力 | 改包能力 | 协议支持 | 自动化程度 |
|---|---|---|---|---|
| BurpSuite | ★★★ | ★★★★★ | ★★★★ | ★★★ |
| HttpCanary | ★★★★★ | ★★ | ★★★ | ★ |
| Frida CLI | ★★★★ | ★★★★ | ★★★★★ | ★★ |
我们的Python RPC网关方案融合了三者优势：保留Burp的拦截修改界面，继承Frida的动态插桩能力，同时通过Python实现协议转换和逻辑处理。这个架构的核心价值在于：
- 实时双向通信：Burp界面修改能立即反映到移动端请求
- 加解密黑盒处理：自动完成JSON/二进制等格式转换
- 跨平台支持：同一套代码适配Android/iOS测试环境
2. 核心架构设计
整个系统由三个关键组件构成：
[移动端App] ←Frida→ [RPC网关] ←HTTP→ [BurpSuite]2.1 通信流程详解
拦截阶段：
// Frida脚本示例 encryptParams.implementation = function(json,signdata){ send(json.toString()); // 发送明文到RPC网关 var modified = recv('send'); // 等待Burp返回修改结果 return originalCall(modified, signdata); }传输阶段：
# Python网关路由处理 @app.route('/hook', methods=['POST']) def handle_hook(): raw_data = request.data burp_response = requests.post( 'http://127.0.0.1:8080', data=raw_data, proxies={'http': 'http://127.0.0.1:8080'} ) return burp_response.content修改阶段：
# 端口转发配置 adb forward tcp:27042 tcp:27042 # Frida调试端口 adb forward tcp:8889 tcp:8889 # RPC服务端口
2.2 关键技术实现
类型转换难题的解决方案：
当遇到Java JSONObject与Python dict之间的类型转换问题时，推荐以下处理方式：
import json from jep import Jep # Java嵌入式Python def convert_to_java_json(py_dict): jep = Jep() jep.eval("import org.json.JSONObject;") json_obj = jep.invoke("JSONObject", json.dumps(py_dict)) return json_obj性能优化技巧：
- 使用
gevent实现异步IO处理 - 对加解密方法添加LRU缓存
- 二进制数据采用Base64编码传输
from functools import lru_cache import base64 @lru_cache(maxsize=1024) def cached_decrypt(data): # 实现带缓存的解密逻辑 return decrypt_result def handle_binary(data): return base64.b64encode(data).decode('utf-8')3. 完整实现步骤
3.1 环境准备
所需工具清单：
- Python 3.8+ 安装以下包：
pip install frida-tools flask requests pyjni - BurpSuite Community/Professional
- Frida-server (匹配设备架构)
- ADB工具套件
Android设备配置：
# 启用开发者选项 adb shell settings put global development_settings_enabled 1 # 推送frida-server adb push frida-server /data/local/tmp/ adb shell "chmod 755 /data/local/tmp/frida-server" adb shell "/data/local/tmp/frida-server &"3.2 服务端核心代码
网关服务器(server.py)：
from flask import Flask, request, jsonify import requests app = Flask(__name__) @app.route('/hook', methods=['POST']) def handle_hook(): try: # 将原始数据转发到Burp burp_response = requests.post( 'http://127.0.0.1:8080', data=request.data, headers={'Content-Type': 'application/octet-stream'}, proxies={'http': 'http://127.0.0.1:8080'} ) # 处理二进制响应 if burp_response.status_code == 200: return burp_response.content, 200 else: return jsonify({"error": "Burp processing failed"}), 500 except Exception as e: return jsonify({"error": str(e)}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=8889, threaded=True)消息中转脚本(agent.py)：
import frida import sys import requests def on_message(message, data): if message['type'] == 'send': payload = message['payload'] try: # 转发到本地网关 response = requests.post( 'http://127.0.0.1:8889/hook', data=payload.encode('utf-8') ) script.post({'type': 'send', 'payload': response.text}) except Exception as e: print(f"Forward error: {str(e)}") # 加载Frida脚本 with open("hook.js", "r") as f: hook_code = f.read() device = frida.get_usb_device() session = device.attach("目标包名") script = session.create_script(hook_code) script.on('message', on_message) script.load() sys.stdin.read()3.3 增强型Frida Hook脚本
hook.js：
Java.perform(function() { var encClass = Java.use('com.secure.app.EncryptionManager'); // 加密函数Hook encClass.encrypt.implementation = function(data) { var original = this.encrypt(data); send({type: 'encrypt', input: data, output: original}); var modified = recv(function(value) { return value.payload; }); return this.encrypt(modified); }; // 解密函数Hook encClass.decrypt.implementation = function(data) { var result = this.decrypt(data); send({type: 'decrypt', input: data, output: result}); return result; }; });4. 高级调试技巧
4.1 常见问题排查
SSL Pinning绕过方案对比：
| 方法 | 成功率 | 复杂度 | 适用场景 |
|---|---|---|---|
| Objection插件 | 高 | 低 | 通用证书锁定 |
| Frida脚本替换 | 中 | 中 | 自定义验证逻辑 |
| 修改APK资源文件 | 低 | 高 | 硬编码证书场景 |
| 系统证书信任 | 高 | 中 | 用户证书验证 |
调试日志增强方案：
# 在网关代码中添加详细日志 import logging logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('gateway.log'), logging.StreamHandler() ] )4.2 性能优化实践
流量处理基准测试：
import timeit setup = ''' from server import handle_hook from flask.testing import FlaskClient app = FlaskClient(app) ''' print(timeit.timeit( 'app.post("/hook", data="test")', setup=setup, number=1000 ))内存优化建议：
- 使用
memory_profiler分析内存泄漏 - 对大文件采用流式处理
- 定期清理缓存数据
from memory_profiler import profile @profile def handle_large_data(data): # 处理大内存操作 pass5. 安全测试实战案例
在最新某银行App测试中，我们发现其采用三层加密方案：
- 请求头：RSA加密时间戳
- 请求体：AES-CBC加密业务数据
- 签名：HMAC-SHA256校验
通过我们的RPC网关，成功实现了：
- 实时修改交易金额
- 重放授权令牌
- 绕过业务逻辑校验
关键Hook点示例：
// 拦截签名校验 signClass.verify.implementation = function(data, sign) { // 总是返回验证成功 return true; };Burp插件扩展：
from burp import IBurpExtender, IHttpListener class BurpExtender(IBurpExtender, IHttpListener): def processHttpMessage(self, tool, isRequest, message): if isRequest and tool == IBurpExtender.TOOL_PROXY: # 自动修改特定参数 request = message.getRequest() modified = request.replace(b"amount=100", b"amount=9999") message.setRequest(modified)
