JS 逆向天花板姿势：JSRPC 远程调用实战

import asyncio, websockets, json, logging
from typing import Any, Dict, Optional

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger('JSRPC')

class JSRPCServer:
    def __init__(self, host='127.0.0.1', port=8080):
        self.host, self.port = host, port
        self.browser_ws: Optional[websockets.WebSocketServerProtocol] = None
        self.pending: Dict[str, asyncio.Future] = {}
        self._req_id = 0

    async def start(self):
        server = await websockets.serve(self._handler, self.host, self.port)
        logger.info(f"JSRPC 服务就绪: ws://{self.host}:{self.port}")
        await server.wait_closed()

    async def _handler(self, ws, path=None):
        self.browser_ws = ws
        logger.info(f"浏览器已连接: {ws.remote_address}")
        try:
            async for msg in ws:
                await self._on_message(msg)
        except websockets.exceptions.ConnectionClosed:
            logger.warning("浏览器断开")
            self.browser_ws = None

    async def _on_message(self, msg: str):
        data = json.loads(msg)
        req_id = data.get('id')
        if req_id in self.pending:
            fut = self.pending.pop(req_id)
            if data.get('error'):
                fut.set_exception(RuntimeError(data['error']))
            else:
                fut.set_result(data.get('result'))

    async def call(self, func_name: str, *args, timeout=10.0) -> Any:
        """远程调用浏览器中的全局函数"""
        if not self.browser_ws:
            raise RuntimeError("无浏览器连接")

        self._req_id += 1
        req_id = str(self._req_id)
        fut = asyncio.get_event_loop().create_future()
        self.pending[req_id] = fut

        await self.browser_ws.send(json.dumps({
            'action': 'call', 'id': req_id, 'func': func_name, 'args': list(args),
        }))

        try:
            return await asyncio.wait_for(fut, timeout)
        except asyncio.TimeoutError:
            self.pending.pop(req_id, None)
            raise TimeoutError(f"{func_name} 超时")

// jsrpc_inject.js
(function() {
    const ws = new WebSocket('ws://127.0.0.1:8080');

    ws.onopen    = () => console.log('[JSRPC] 已连接');
    ws.onclose   = () => setTimeout(() => location.reload(), 5000);  // 断开则刷新重连
    ws.onmessage = async (event) => {
        const msg = JSON.parse(event.data);
        if (msg.action !== 'call') return;

        try {
            // 按路径取函数：支持 "getSign" 或 "obj.encrypt"
            const path = msg.func.split('.');
            let func = window;
            for (const k of path) func = func[k];

            // 原生环境执行（支持 async 函数）
            const result = await func.apply(null, msg.args || []);
            ws.send(JSON.stringify({ id: msg.id, result }));
        } catch (err) {
            ws.send(JSON.stringify({ id: msg.id, error: err.toString() }));
        }
    };
})();

async def main():
    rpc = JSRPCServer(port=8080)
    server_task = asyncio.create_task(rpc.start())

    # 等待浏览器注入脚本后连接
    while rpc.browser_ws is None:
        await asyncio.sleep(0.5)

    # 远程调用浏览器中的加密函数
    sign = await rpc.call('getSign', 'user_token_12345')
    print(f"签名: {sign}")

    # 带签名发起业务请求
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get('https://target.com/api/data',
                               headers={'X-Sign': sign}) as resp:
            data = await resp.json()
            print(f"业务数据: {data}")

    server_task.cancel()

asyncio.run(main())

from playwright.async_api import async_playwright

async def inject_jsrpc():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        page = await browser.new_page()
        await page.goto('https://target.com/login')

        with open('jsrpc_inject.js') as f:
            await page.evaluate(f.read())

        await page.wait_for_timeout(600000)

asyncio.run(inject_jsrpc())

async def batch_call(self, calls: list) -> list:
    """并发发送多个调用请求，一次性等待全部结果"""
    loop = asyncio.get_event_loop()
    futures, req_ids = [], []

    for spec in calls:
        self._req_id += 1
        rid = str(self._req_id)
        fut = loop.create_future()
        self.pending[rid] = fut
        futures.append(fut)
        await self.browser_ws.send(json.dumps({
            'action': 'call', 'id': rid,
            'func': spec['func'], 'args': spec.get('args', []),
        }))

    return await asyncio.gather(*futures, return_exceptions=True)

import time

class CachedJSRPC(JSRPCServer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._cache = {}

    async def cached_call(self, func_name: str, *args, ttl=300):
        key = f"{func_name}:{hash(args)}"
        if key in self._cache:
            ts, val = self._cache[key]
            if time.time() - ts < ttl:
                return val
        result = await self.call(func_name, *args)
        self._cache[key] = (time.time(), result)
        return result