"""把 Dart 发出的明文 HTTP 显示在本机页面。
Charles 看不到这些请求：它们不走系统代理，在进系统套接字之前就已经加密。
这里直接从进程内存里取出加密前的报文。
"""
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import frida
DEVICE = "emulator-5556"
PORT = 8791
APP_NAMES = ("RaptDrama",)
JS = r"""
function extract(addr) {
var back = addr;
for (var i = 0; i < 160; i++) {
var b = 0;
try { b = back.sub(1).readU8(); } catch (e) { break; }
if (b !== 0x0a && b !== 0x0d && (b < 0x20 || b > 0x7e)) break;
back = back.sub(1);
}
var s = "";
for (var j = 0; j < 2500; j++) {
var c = 0;
try { c = back.add(j).readU8(); } catch (e) { break; }
if (c === 0x0d) continue;
if (c === 0x0a) { s += "\n"; continue; }
if (c < 0x20 || c > 0x7e) break;
s += String.fromCharCode(c);
}
return s;
}
var seen = {};
var scanning = false;
var lastScan = 0;
function scan() {
var now = Date.now();
if (scanning || now - lastScan < 1200) return;
scanning = true;
lastScan = now;
var ranges = Process.enumerateRanges("rw-");
var fresh = [];
for (var i = 0; i < ranges.length; i++) {
var r = ranges[i];
if (r.size > 24 * 1024 * 1024) continue;
var matches;
try { matches = Memory.scanSync(r.base, r.size, "48 54 54 50 2f 31 2e 31"); }
catch (e) { continue; }
for (var j = 0; j < matches.length; j++) {
var text = extract(matches[j].address);
if (!text || text.length < 16 || seen[text]) continue;
if (text.indexOf("package:") === 0) continue;
seen[text] = true;
fresh.push(text);
}
}
if (fresh.length) send(JSON.stringify(fresh));
scanning = false;
}
var writePtr = Process.getModuleByName("libc.so").getExportByName("write");
Interceptor.attach(writePtr, {
onEnter: function (args) {
try {
if (args[1].readU8() === 0x17) scan();
} catch (e) {}
}
});
scan();
send("ready");
"""
FLOWS = []
LOCK = threading.Lock()
def on_message(message, _data):
if message.get("type") != "send":
print(message, flush=True)
return
payload = message["payload"]
if payload == "ready":
print("已挂上进程，等待请求", flush=True)
return
items = json.loads(payload)
with LOCK:
for text in items:
first = text.strip().split("\n", 1)[0][:180]
FLOWS.append({"id": len(FLOWS) + 1, "summary": first, "text": text})
print(first, flush=True)
def attach():
device = frida.get_device(DEVICE)
pid = None
for proc in device.enumerate_processes():
if proc.name in APP_NAMES or "Rapt" in proc.name:
pid = proc.pid
break
if pid is None:
raise SystemExit("RaptDrama 没在运行。先打开 App，再启动这个页面。")
print(f"attach {pid}", flush=True)
session = device.attach(pid)
script = session.create_script(JS)
script.on("message", on_message)
script.load()
return session, script
PAGE = """
明文请求
打开 App 并滑动页面，新的请求会出现在左边。
"""
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path.startswith("/api/flows"):
with LOCK:
payload = json.dumps(FLOWS, ensure_ascii=False).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
return
body = PAGE.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, fmt, *args):
return
def main():
attach()
server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
print(f"打开 http://127.0.0.1:{PORT}", flush=True)
server.serve_forever()
if __name__ == "__main__":
main()