企百科 Web/H5 混合环境下的 WASM 调试与批处理实现

张 张站长 · 2026-09-14 10:39 · 技术-逆向 · 25 次浏览 · 0 条评论

记录企百科项目中 Python、Node.js、Wasmtime 与 WebAssembly 的混合调试过程，涵盖模块检查、内存模型、授权客户端、分页批处理、重试和排错。

企百科 Web/H5 混合环境下的 WASM 调试与批处理实现

这篇文章记录我在“企百科”项目里做 Web/H5 混合环境调试的完整过程。项目的核心是：使用 Python 管理业务请求和批处理，使用 Node.js 或 Wasmtime 加载前端使用的 WebAssembly 模块，再把结果交给统一的数据处理层。文中代码使用占位配置和本地测试数据，真实账号凭据不写入代码。

一、项目目录

我先把目录里的文件分成三类：

wasm_lib_bg.wasm       WebAssembly 二进制模块
wasm_lib_bg.js         webpack/wasm-bindgen 运行时封装
wasm_lib_bg1.js        精简的模块加载封装
wasm_lib_bg2.js        调试过程中保存的完整运行时代码

aa.py                  Python 调用 Node.js 的桥接示例
qibook混合环境.py       Python + Wasmtime + requests 的混合客户端
aa.js                  Node.js 实例化 WASM 的基础示例
bb.js                  Node.js 读取导出函数的完整示例
3.py                   早期的批量请求脚本
aaa.txt                一组测试输入

先查看文件类型和大小：

file *
ls -lh

在这个项目里，真正参与运行的不是 wasm_lib_bg.js 单独一个文件。JavaScript 文件负责准备导入对象、处理线性内存和封装返回值，.wasm 文件才是实际执行函数的二进制模块。

企百科混合环境结构

二、先检查 WASM 的导入和导出

在调用任何函数之前，我先只读取模块元数据，不执行业务函数。Node.js 自带的 WebAssembly.Module.exports() 和 WebAssembly.Module.imports() 可以完成这一步：

const fs = require("fs");

const buffer = fs.readFileSync("./wasm_lib_bg.wasm");
const module = new WebAssembly.Module(buffer);

console.log("exports:");
for (const item of WebAssembly.Module.exports(module)) {
  console.log(`${item.name}\t${item.kind}`);
}

console.log("imports:");
for (const item of WebAssembly.Module.imports(module)) {
  console.log(`${item.module}::${item.name}\t${item.kind}`);
}

我在本地检查到的导出结构如下：

memory                         memory
get_key                        function
decrypt                        function
__wbindgen_add_to_stack_pointer function
__wbindgen_free                function
__wbindgen_malloc              function
__wbindgen_realloc             function
__webpack_init__               function

导入函数只有三类：

./wasm_lib_bg.js::__wbg_log_6507dcafd840b967
./wasm_lib_bg.js::__wbg_random_26e2d782b541ca6b
./wasm_lib_bg.js::__wbindgen_throw

这个结果说明模块不是完全独立运行的，实例化时需要补齐日志、随机数和异常处理函数。导出函数的存在只能说明模块提供了这些入口，不能说明调用参数、返回值和业务状态可以直接猜测。

三、Node.js 实例化过程

aa.js 里使用的是通用的实例化流程。我把它整理成一个只负责加载模块的版本：

const fs = require("fs/promises");

async function loadWasm(file) {
  const bytes = await fs.readFile(file);
  const module = await WebAssembly.compile(bytes);

  const imports = {
    "./wasm_lib_bg.js": {
      __wbg_log_6507dcafd840b967(ptr, len) {
        console.log("[wasm-log]", { ptr, len });
      },
      __wbg_random_26e2d782b541ca6b() {
        return Math.random();
      },
      __wbindgen_throw(ptr, len) {
        throw new Error(`WASM exception at ${ptr}, length ${len}`);
      }
    }
  };

  return WebAssembly.instantiate(module, imports);
}

async function main() {
  const instance = await loadWasm("./wasm_lib_bg.wasm");
  const exports = Object.keys(instance.exports);
  console.log("loaded exports:", exports);
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});

运行：

node --version
node load_wasm.js

如果出现 Import ... requires a callable，说明导入模块名称或函数名没有对应上。这里的导入模块名必须和 WASM 中记录的名称一致，项目使用的是 ./wasm_lib_bg.js。

四、wasm-bindgen 的内存模型

这个模块由 wasm-bindgen 生成 JavaScript 封装，所以字符串参数不是直接传入 WASM 函数，而是经过下面几步：

  1. JavaScript 把字符串编码成 UTF-8；
  2. 调用 __wbindgen_malloc 分配线性内存；
  3. 把 UTF-8 字节写入 memory；
  4. 通过栈指针创建临时返回区域；
  5. 调用导出函数；
  6. 从返回区域取出指针和长度；
  7. 从线性内存读取 UTF-8；
  8. 调用 __wbindgen_free 释放返回缓冲区；
  9. 恢复栈指针。

WASM 字符串调用过程

返回区域通常可以看成两个连续的 32 位整数：

offset + 0     result_ptr
offset + 4     result_len

只读取元数据时，我会使用下面这个辅助函数查看返回布局：

function readReturnPair(memory, stackPtr) {
  const view = new DataView(memory.buffer);
  return {
    ptr: view.getUint32(stackPtr, true),
    len: view.getUint32(stackPtr + 4, true)
  };
}

这里没有把真实业务参数、真实密文或真实返回内容写入日志。调试内存时只记录指针范围、长度和错误码，避免把业务数据带进日志文件。

五、Python 侧用 Wasmtime 加载模块

项目中的 qibook混合环境.py 使用 wasmtime。我把初始化部分单独拆出来，先完成实例化和导出函数检查：

from pathlib import Path
import random
import wasmtime


class WasmRuntime:
    def __init__(self, wasm_path: str):
        self.engine = wasmtime.Engine()
        self.store = wasmtime.Store(self.engine)
        self.module = wasmtime.Module.from_file(
            self.engine, Path(wasm_path)
        )
        self.linker = wasmtime.Linker(self.engine)

        self.linker.define_func(
            "./wasm_lib_bg.js",
            "__wbg_random_26e2d782b541ca6b",
            wasmtime.FuncType([], [wasmtime.ValType.f64()]),
            lambda: random.random(),
        )
        self.linker.define_func(
            "./wasm_lib_bg.js",
            "__wbg_log_6507dcafd840b967",
            wasmtime.FuncType(
                [wasmtime.ValType.i32(), wasmtime.ValType.i32()],
                [],
            ),
            lambda ptr, length: None,
        )
        self.linker.define_func(
            "./wasm_lib_bg.js",
            "__wbindgen_throw",
            wasmtime.FuncType(
                [wasmtime.ValType.i32(), wasmtime.ValType.i32()],
                [],
            ),
            lambda ptr, length: None,
        )

        self.instance = self.linker.instantiate(
            self.store, self.module
        )
        self.exports = self.instance.exports(self.store)

    def export_names(self):
        return [item.name for item in self.module.exports]

运行检查：

runtime = WasmRuntime("./wasm_lib_bg.wasm")
print(runtime.export_names())

如果这里只做模块兼容性检查，到这里就结束。只有在授权的本地测试环境里，才继续实现字符串参数的内存拷贝和返回值解析。

六、为什么 Python 和 Node.js 要分开

我在项目中保留两个运行时，原因是两边的职责不同：

Node.js
  - 更接近浏览器的 WebAssembly 加载方式
  - 方便对照 webpack / wasm-bindgen 封装
  - 适合验证导入对象和导出函数

Python
  - 方便组织请求、分页、重试和输出文件
  - 适合处理 CSV、JSONL 和任务记录
  - 方便做批处理和统计

如果把所有内容都塞进一个脚本里，WASM 内存问题、HTTP 请求问题和批处理问题会混在一起。我的实现方式是把它们拆成三个对象：

class WasmRuntime:
    pass


class AuthorizedClient:
    pass


class BatchRunner:
    pass

WasmRuntime 只负责模块生命周期，AuthorizedClient 只负责经过授权的接口调用，BatchRunner 只负责输入、分页和输出。

七、配置文件不写入账号和 Token

项目早期脚本里把 Cookie、Token、设备标识和请求头直接写在源文件中。正式运行时，我改成从环境变量读取：

import os


def required_env(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"missing environment variable: {name}")
    return value


API_URL = required_env("AUTHORIZED_API_URL")
ACCESS_TOKEN = required_env("AUTHORIZED_ACCESS_TOKEN")

本地运行前：

export AUTHORIZED_API_URL="http://127.0.0.1:3000/mock/search"
export AUTHORIZED_ACCESS_TOKEN="local-test-token"
python3 client.py

测试日志不能输出完整请求头。调试时只输出是否存在：

def safe_headers(headers: dict) -> dict:
    return {
        "content_type": headers.get("Content-Type", ""),
        "has_authorization": bool(headers.get("Authorization")),
        "user_agent": headers.get("User-Agent", ""),
    }

八、授权客户端的请求结构

业务请求只保留通用的分页字段。接口路径、字段名和授权方式以接口所有者提供的文档为准：

import requests


class AuthorizedClient:
    def __init__(self, url: str, token: str):
        self.url = url
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "Content-Type": "application/json",
            "Authorization": f"Bearer {token}",
        })

    def search(self, keyword: str, page: int = 1, size: int = 10):
        payload = {
            "keyword": keyword,
            "page": page,
            "size": size,
        }
        response = self.session.post(
            self.url,
            json=payload,
            timeout=(5, 30),
        )
        response.raise_for_status()
        data = response.json()
        if not isinstance(data, dict):
            raise ValueError("response must be an object")
        return data

测试时使用本地 Mock 服务：

from http.server import BaseHTTPRequestHandler, HTTPServer
import json


class MockHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        size = int(self.headers.get("Content-Length", "0"))
        body = json.loads(self.rfile.read(size) or b"{}")
        result = {
            "items": [{
                "name": body.get("keyword", ""),
                "source": "fixture",
            }],
            "page": body.get("page", 1),
            "size": body.get("size", 10),
        }
        encoded = json.dumps(result).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def log_message(self, *_):
        pass


HTTPServer(("127.0.0.1", 3000), MockHandler).serve_forever()

先启动 Mock，再运行客户端，先确认分页、异常和输出逻辑正确，再连接正式授权环境。

九、批处理输入和去重

项目中有一组企业名称测试数据。批处理前先做清洗和去重：

from collections.abc import Iterable


def normalize_keywords(values: Iterable[str]) -> list[str]:
    result = []
    seen = set()
    for value in values:
        keyword = " ".join(value.split())
        if not keyword or keyword in seen:
            continue
        seen.add(keyword)
        result.append(keyword)
    return result

从文本读取：

from pathlib import Path


keywords = normalize_keywords(
    Path("keywords.txt").read_text(encoding="utf-8").splitlines()
)
print("unique keywords:", len(keywords))

不要把测试文件里的真实企业名单直接写进文章或示例输出。教程只展示字段处理方法，正式数据由任务操作者在本地加载。

十、分页、限速和失败记录

批处理的每一项都要记录状态，不能只 print(response.text)。我使用 JSONL 保存每条任务结果：

import json
import time
from pathlib import Path


def run_batch(client, keywords, output: Path, interval: float = 1.0):
    output.parent.mkdir(parents=True, exist_ok=True)
    with output.open("a", encoding="utf-8") as fp:
        for index, keyword in enumerate(keywords, start=1):
            record = {
                "index": index,
                "keyword_hash": hash(keyword),
                "status": "started",
            }
            try:
                data = client.search(keyword, page=1, size=10)
                record["status"] = "ok"
                record["count"] = len(data.get("items", []))
                record["data"] = data
            except Exception as exc:
                record["status"] = "error"
                record["error"] = type(exc).__name__
            fp.write(json.dumps(record, ensure_ascii=False) + "\n")
            fp.flush()
            time.sleep(interval)

正式任务中，我会把 hash(keyword) 换成稳定的 SHA-256 截断值，并且不保存原始请求头：

import hashlib


def stable_id(value: str) -> str:
    return hashlib.sha256(
        value.encode("utf-8")
    ).hexdigest()[:16]

批处理任务流程

10.1 用游标继续任务

批处理被中断后，应该从成功位置继续，而不是重复全部请求：

def load_completed(output: Path) -> set[str]:
    completed = set()
    if not output.exists():
        return completed
    for line in output.read_text(encoding="utf-8").splitlines():
        row = json.loads(line)
        if row.get("status") == "ok":
            completed.add(row["keyword_hash"])
    return completed

运行前跳过已经成功的条目：

completed = load_completed(Path("work/results.jsonl"))
pending = [
    item for item in keywords
    if stable_id(item) not in completed
]

十一、WASM 实例的生命周期

WASM 模块中的内存、分配器和状态属于一个实例。项目早期代码在两个上下文之间复用一个实例，容易出现状态互相覆盖。我的处理方式是每个业务上下文独立创建运行时：

class RuntimePool:
    def __init__(self, wasm_path: str):
        self.h5 = WasmRuntime(wasm_path)
        self.web = WasmRuntime(wasm_path)

    def get(self, name: str) -> WasmRuntime:
        if name == "h5":
            return self.h5
        if name == "web":
            return self.web
        raise KeyError(name)

这样可以把 H5 和 Web 的请求头、页面路径、日志和 WASM 状态分开。任何一个上下文出现异常，都不会直接污染另一个上下文。

十二、异常处理和重试

我把错误分成三类：

网络错误       连接超时、读取超时、DNS 失败
协议错误       HTTP 状态码异常、JSON 格式异常
业务错误       返回结构存在，但业务状态不是成功

只对网络错误和明确允许重试的状态码进行有限重试：

import time
import requests


def request_with_retry(session, method, url, **kwargs):
    delays = (1.0, 2.0, 4.0)
    for attempt, delay in enumerate(delays, start=1):
        try:
            response = session.request(
                method, url, timeout=(5, 30), **kwargs
            )
            if response.status_code in {429, 502, 503, 504}:
                if attempt == len(delays):
                    response.raise_for_status()
                time.sleep(delay)
                continue
            response.raise_for_status()
            return response
        except requests.Timeout:
            if attempt == len(delays):
                raise
            time.sleep(delay)
    raise RuntimeError("request failed")

不对 401、403 等鉴权错误无限重试，也不通过改变设备标识、请求头或客户端环境来规避服务端限制。遇到权限错误时，直接停止任务并检查授权配置。

十三、输出文件和审计

输出内容只保留完成任务所需的字段：

def pick_public_fields(item: dict) -> dict:
    allowed = ("name", "credit_code", "status", "registered_at")
    return {key: item.get(key) for key in allowed}

任务日志保存：

{
  "job_id": "local-test-20260914-001",
  "runtime": "wasmtime",
  "input_count": 20,
  "success_count": 18,
  "error_count": 2,
  "started_at": "2026-09-14T10:00:00+08:00",
  "finished_at": "2026-09-14T10:02:14+08:00"
}

不保存：

完整 Cookie
完整 Token
设备指纹
真实响应密文
未脱敏的个人信息

十四、对项目现有文件的整理

我会按下面的方式整理这个目录：

企百科/
├── wasm/
│   ├── wasm_lib_bg.wasm
│   └── wasm_lib_bg.js
├── runtime/
│   ├── node_loader.js
│   └── wasmtime_runtime.py
├── batch/
│   ├── client.py
│   ├── runner.py
│   └── keywords.example.txt
├── fixtures/
│   └── response.example.json
└── work/
    └── .gitkeep

原始调试文件可以保留在单独的本地归档目录，不直接提交到公共仓库。尤其是包含 Cookie、Token、设备标识和完整请求数据的早期脚本，应先撤销凭据，再清理历史提交。

十五、运行顺序

我实际执行时按照下面的顺序：

# 1. 检查 Node.js
node --version

# 2. 检查 WASM 导入和导出
node inspect_wasm.js

# 3. 安装 Python 依赖
python3 -m venv .venv
source .venv/bin/activate
pip install wasmtime requests

# 4. 运行本地模块检查
python3 wasmtime_runtime.py

# 5. 启动本地 Mock 接口
python3 mock_server.py

# 6. 使用本地测试配置运行批处理
export AUTHORIZED_API_URL="http://127.0.0.1:3000/mock/search"
export AUTHORIZED_ACCESS_TOKEN="local-test-token"
python3 batch_runner.py

运行结束后检查：

wc -l work/results.jsonl
grep '"status": "error"' work/results.jsonl
du -sh work/

十六、排错记录

1. WASM 无法实例化

先检查导入模块名和函数签名：

node inspect_wasm.js

导入名必须和模块中记录的一致，f64、i32 等参数类型也必须匹配。

2. 返回内容为空

检查栈指针是否在调用前减小、调用后恢复；检查返回结构的指针和长度是否按 little-endian 读取；最后确认返回指针是否释放。

3. Python 和 Node.js 结果不一致

确认两边使用的是同一个 .wasm 文件、同样的 UTF-8 编码和同样的导入函数。不要把一个实例的内存指针交给另一个实例使用。

4. 批处理重复

检查输入是否去重，输出是否使用稳定任务 ID，重试是否会重复写入。用 job_id + keyword_hash 做幂等键。

5. 请求频率过高

减小任务规模，增加间隔，并按接口文档执行配额和并发限制。不要通过伪造客户端信息来绕过服务端限制。

十七、总结

这个项目最终可以拆成四个稳定层次：

WASM 模块
  -> 运行时桥接
  -> 授权客户端
  -> 批处理与审计

我在实现时先检查 WASM 的导入导出，再处理内存和字符串，最后接入请求与批处理。这样每一层都能单独测试，出了问题也能快速定位到模块加载、内存管理、请求协议或任务调度中的具体一层。

评论 0

还没有评论
登录后发表评论
← 全部文章

找站长

站长微信二维码 扫码加微信
售后群二维码 扫码进售后群
站长收款码 建站不易，感谢打赏
微信 18581469390
本站不支持自助充值；需要充值或开通额度，请直接联系站长。
加微信可以找我重置密码、充值、问工具怎么用