企业官网建设流程全解析

JS 逆向走到深处，最终都会撞上一堵墙：核心加密逻辑被 VMP 虚拟机保护、控制流平坦化、反调试陷阱层层包裹，静态分析几乎无法还原，AST 解析也无从下手。传统"抠代码 + 补环境"策略在此彻底失效。

JSRPC（JavaScript Remote Procedure Call）换了一个思路：不逆向算法本身，把浏览器变成远程加密服务。加密函数在浏览器原环境中正常执行，Python 爬虫通过 WebSocket 远程调用。本文从原理到完整实现，构建一套生产可用的 JSRPC 框架。

一、核心思想：不还原算法，远程调用原函数

数据流：

Python 爬虫 浏览器（真实环境） │ WS: {func:"getSign", args:["token"]} │ │ ──────────────────────────────────────────→ │ │ 执行 window.getSign("token") │ WS: {result:"a8f3b2c1"} │ │ ←────────────────────────────────────────── │

二、Python 端：WebSocket Server

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} 超时")

三、浏览器端：注入脚本

通过控制台或 Playwright 注入，建立 WebSocket 长连接并监听调用指令：

// 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())

配合 Playwright 自动注入

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())

五、进阶优化

批量调用（减少 RTT 开销）

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

六、生产部署与代理配合

容器化部署时，浏览器和 Python 服务端打包在同一镜像中。浏览器侧需配合代理 IP 轮换以规避风控：

from playwright.async_api import async_playwright import requests, random def get_proxy(): """亿牛云 API 代理提取出口 IP""" url = "http://ip.16yun.cn:817/myip/pl/<ORDER_ID>/?s=<SIGN>&u=<USER>&format=json" resp = requests.get(url, timeout=10) if resp.status_code == 429: return None # 提取频率过快 d = resp.json() return f"http://{d[0]['ip']}:{d[0]['port']}" async def launch_with_proxy(): proxy = get_proxy() async with async_playwright() as p: browser = await p.chromium.launch( headless=True, proxy={'server': proxy} if proxy else None, ) page = await browser.new_page() await page.goto('https://target.com') with open('jsrpc_inject.js') as f: await page.evaluate(f.read()) # 保持运行 await page.wait_for_timeout(3600000)

浏览器场景下，登录态维持和加密调用必须在同一会话中完成。亿牛云爬虫代理通过**<font style="color:rgb(0, 206, 185);background-color:rgb(252, 252, 252);">Proxy-Tunnel</font>**头绑定出口 IP——相同随机数保持同一 IP，确保登录与后续 RPC 调用的会话一致性。遇到403检查白名单，遇到429降低采集频率。

七、选型决策

八、踩坑速查

九、总结

JSRPC 的核心价值在于将对不可逆向加密的应对策略从"还原算法"转变为"调用原函数"。

三个结构性优势：零逆向成本（不分析内部实现）、零维护成本（站点更新自动生效）、100% 环境还原（原生浏览器执行）。代价是每次调用约 50~200ms 的 RTT 开销，不适合万级 QPS 场景。

选型原则很简单：加密算法能在 2 小时内逆向还原，走扣代码；超过 2 天搞不定，上 JSRPC。在 VMP 保护日益普及的当下，JSRPC 正从非常规手段变成 JS 逆向的标配技能。