WebAssembly调用：模拟调用Wasm中的加密函数。Python爬虫进阶：WebAssembly调用实战——模拟Wasm加密函数突破反爬

写爬虫写久了，总会遇到一些让人头疼的反爬手段。从最初的User-Agent检测，到后来的IP封禁、验证码、字体反爬，再到JavaScript混淆、WebSocket验证……反爬技术在不断进化，爬虫工程师的应对策略也必须与时俱进。

最近半年，我在爬取某电商平台和某短视频网站的数据时，遇到了一种新的反爬屏障——WebAssembly（以下简称Wasm）。对方的核心加密逻辑不再以传统的JavaScript文件形式暴露，而是编译成了Wasm二进制模块，只在运行时通过JS胶水代码调用。传统的AST还原、动态Hook变得困难重重。

这篇文章，我会从一个真实场景出发：模拟调用Wasm中的加密函数，帮助大家理解Wasm模块的加载、实例化、函数调用全过程，并给出Python端完整的调用方案。不绕弯子，直接上代码和踩坑记录。

目录

一、WebAssembly是什么？为什么反爬喜欢它？

二、整体思路：Python如何调用Wasm模块？

三、准备一个示例Wasm加密模块

四、Python调用Wasm的完整代码（可直接运行）

环境准备

基础调用示例：Wasm中的加法函数

进阶：调用加密哈希函数

实战案例：某视频网站Wasm签名还原

五、处理带状态的Wasm模块

六、处理Wasm导出函数的多种参数和返回值类型

1. 纯数值计算

2. 无返回值（void）

3. 返回两个整数（通常编码成一个i64）

4. 通过内存返回数组/缓冲区

七、处理Wasm导入的JS函数

八、内存分配问题的深入探讨

一、WebAssembly是什么？为什么反爬喜欢它？

在开始写代码之前，有必要先搞清楚Wasm的本质。

WebAssembly是一种基于栈式虚拟机的二进制指令格式，设计目标是让C/C++、Rust、Go等语言编写的代码能以接近原生的速度在浏览器中运行。它不是一个编程语言，而是一个编译目标。

反爬为什么青睐Wasm？三个原因：

1. 二进制格式难以阅读。相比JS源码，Wasm的二进制形式直接展开是一堆字节码，没有变量名、没有注释、没有清晰的逻辑结构。

2. 还原成本高。虽然有wabt等工具可以把Wasm转成文本格式（wat），但那已经是低级指令了，控制流平坦化、局部变量重命名是常态。

3. 执行效率高。Wasm在浏览器里的执行速度比纯JS快很多，对于需要高频调用的加密函数（比如每个请求都生成一次签名），性能优势明显。

很多大厂已经开始把核心签名算法编译成Wasm，JS端只保留最小化的加载逻辑。这就意味着，单纯靠分析JS已经拿不到完整算法了。

二、整体思路：Python如何调用Wasm模块？

Python本身没有原生的Wasm运行时，但我们可以借助一些现成的工具和库。目前主流方案有三个：

• 方案一：使用pywasm库。纯Python实现的Wasm解释器，方便调试，但性能差，适合小型模块。

• 方案二：使用wasmer或wasmtime的Python绑定。这两个是高性能的Wasm运行时，接近原生执行速度，适合生产环境。

• 方案三：通过Node.js桥接。Python调用Node子进程，Node负责执行Wasm，数据通过stdout传回。麻烦，但兼容性好。

本文重点讲方案二：wasmtime的Python绑定。wasmtime由Bytecode Alliance维护，稳定性和性能都有保障，而且支持WASI（WebAssembly System Interface），可以让Wasm模块访问文件系统等资源（虽然加密场景一般不需要）。

三、准备一个示例Wasm加密模块

为了演示，我们首先需要有一个Wasm文件。假设加密逻辑是用Rust写的，非常简单：接收一个字符串，返回它的MD5哈希值（十六进制）。当然真实场景的加密会比这个复杂得多，但调用流程完全一致。

Rust代码（src/lib.rs）：

rust

use md5::{Digest, Md5}; use wasm_bindgen::prelude::*; #[wasm_bindgen] pub fn quick_hash(input: &str) -> String { let mut hasher = Md5::new(); hasher.update(input.as_bytes()); format!("{:x}", hasher.finalize()) }

编译命令（需要安装wasm-pack）：

bash

wasm-pack build --target bundler

生成的Wasm文件在pkg/目录下。如果没有Rust环境，也可以直接下载我提供的示例文件（文末会给出备选方案）。当然你也可以用在线Wasm编辑器生成一个简单的加法函数Wasm，原理一样。

为了让读者更好理解，我先不依赖真实文件，直接给出一个Wasm的wat文本描述，你可以用wat2wasm工具转换成二进制：

wat

(module (func $add (param $a i32) (param $b i32) (result i32) local.get $a local.get $b i32.add) (export "add" (func $add)) )

这是一个简单的加法模块。但我们今天的重点是调用复杂加密函数，所以我下面的代码会假设已经有一个加密Wasm。

四、Python调用Wasm的完整代码（可直接运行）

环境准备

bash

pip install wasmtime

注意：wasmtime的Python包依赖Rust工具链的安装，但pip会自动处理二进制文件，通常直接装就能用。如果报错，去官网下载对应系统的wasmtime可执行文件并加入PATH。

基础调用示例：Wasm中的加法函数

先拿简单的加法热热身：

python

from wasmtime import Store, Module, Instance, FuncType, Func, ValType # 创建store（可以理解为Wasm的运行时上下文） store = Store() # 定义Wasm模块的字节码（这里是wat格式转换后的二进制，实际从文件读） # 为了演示，我直接写wat，再用wat2wasm转，但wasmtime可以直接从wat文件加载 # 更简单的方式：把上面的wat保存成add.wat，然后用wasmtime的Module.from_file # 假设你已经有一个add.wasm文件 module = Module.from_file(store.engine, "add.wasm") # 实例化模块 instance = Instance(store, module, []) # 获取导出的函数 add_func = instance.exports(store)["add"] # 调用函数 result = add_func(store, 10, 20) print(f"10 + 20 = {result}") # 输出30

进阶：调用加密哈希函数

现在重点来了。假设你的Wasm模块导出了一个叫做quick_hash的函数，接收一个字符串（UTF-8编码），返回一个字符串（也是UTF-8）。

Wasm本身只支持i32、i64、f32、f64等数值类型，字符串如何传递？答案是：线性内存。调用方（Python）把字符串写入Wasm模块的线性内存，然后把指针（起始地址）和长度作为i32参数传给函数；函数内部读取内存、计算、再把结果字符串写回内存，返回指针和长度。

流程如下：

1. Python将字符串编码成UTF-8字节序列。

2. 在Wasm的内存中分配一块空间，把字节序列拷贝进去。

3. 调用Wasm函数，传入指针和长度。

4. Wasm函数返回结果指针和长度（通常编码成一个i64，高32位是长度，低32位是指针，或者两个i32）。

5. Python从内存中读取结果字节，解码成字符串。

听起来复杂，但wasmtime的Python绑定提供了Func类型的高级封装，可以自动处理简单类型转换。对于字符串参数，我们需要手动操作内存。

下面给出一个可以直接运行的完整示例。为了不依赖外部文件，我会先用Python动态创建一个简单的Wasm模块（使用wasmtime的底层API来构造），然后调用它。但更贴近实战的做法是加载外部wasm文件，我会两种都写出来。

方案A：加载外部加密Wasm文件（推荐）

python

import hashlib from wasmtime import Store, Module, Instance, Memory, MemoryType, Func, FuncType, ValType, GlobalType, Global def call_wasm_encrypt(wasm_path, func_name, input_str): """ 通用Wasm加密函数调用封装 Args: wasm_path: wasm文件路径 func_name: 要调用的导出函数名 input_str: 输入字符串 Returns: 函数返回的字符串（假设返回值也是字符串） """ store = Store() module = Module.from_file(store.engine, wasm_path) instance = Instance(store, module, []) # 获取Wasm的线性内存（通常第一个memory就是） memory = instance.exports(store)["memory"] # 获取目标函数 target_func = instance.exports(store)[func_name] # 将字符串写入内存 input_bytes = input_str.encode('utf-8') input_len = len(input_bytes) # 在Wasm内存中分配空间（需要调用Wasm的malloc函数，如果没有，就手动找一个空闲区域） # 这里简化处理：假设我们直接写入内存起始地址偏移0x1000处，避开程序使用的低地址 # 实际工程中建议导出_malloc或externref机制 ptr = 0x1000 # 示例偏移量，实际应动态分配 memory.write(store, ptr, input_bytes) # 调用Wasm函数（假设函数签名是 (ptr: i32, len: i32) -> i32, 返回结果指针） # 不同的Wasm模块签名可能不同，这里需要根据实际情况调整 # 真实场景往往返回两个i32或者一个i64包装 # 我们以某电商平台的签名函数为例，假设它接收两个参数：指针和长度，返回结果指针（指向UTF-8字符串） result_ptr = target_func(store, ptr, input_len) # 读取返回值：通常返回值后面4个字节存放长度，或者函数内部另有导出函数获取长度 # 这里简化：假设结果字符串以空字符结尾 result_bytes = bytearray() offset = result_ptr while True: b = memory.read(store, offset, 1)[0] if b == 0: break result_bytes.append(b) offset += 1 return result_bytes.decode('utf-8') # 使用示例（假设已经有了encrypt.wasm） # output = call_wasm_encrypt("encrypt.wasm", "quick_hash", "hello world") # print(output)

方案B：更通用的处理方式——读取结果长度

许多Wasm模块会在返回值后隐式存储长度，或者通过导出的get_result_len函数获取。更规范的做法是：

python

def call_wasm_encrypt_robust(wasm_path, func_name, input_str, malloc_func_name="malloc", free_func_name=None): store = Store() module = Module.from_file(store.engine, wasm_path) instance = Instance(store, module, []) memory = instance.exports(store)["memory"] # 尝试获取malloc和free（如果需要动态分配内存） try: malloc = instance.exports(store)[malloc_func_name] except: malloc = None encrypt_func = instance.exports(store)[func_name] # 准备输入数据 input_bytes = input_str.encode('utf-8') input_len = len(input_bytes) if malloc: # 动态分配内存 input_ptr = malloc(store, input_len + 1) memory.write(store, input_ptr, input_bytes) # 写入结束符（可选） memory.write(store, input_ptr + input_len, b'\x00') else: # 固定偏移（风险高，仅用于测试） input_ptr = 0x10000 memory.write(store, input_ptr, input_bytes) # 调用加密函数（假设返回结果指针） result_ptr = encrypt_func(store, input_ptr, input_len) # 读取结果长度（假设前4字节是长度） if result_ptr: len_bytes = memory.read(store, result_ptr, 4) result_len = int.from_bytes(len_bytes, 'little') result_bytes = memory.read(store, result_ptr + 4, result_len) result = result_bytes.decode('utf-8') else: result = "" # 释放内存（如果有free） if malloc and free_func_name: free = instance.exports(store)[free_func_name] free(store, input_ptr) free(store, result_ptr) return result

实战案例：某视频网站Wasm签名还原

去年年底，我在分析某视频弹幕网站的请求签名时，发现sign参数的生成逻辑藏在一个sign.wasm里。通过浏览器开发者工具的网络面板，下载到这个wasm文件，大小约45KB。

分析步骤：

1. 在页面JS里找到加载wasm的代码，发现导出了三个函数：init,sign,get_sign_str。

2. sign函数接收两个i32参数（字符串指针和长度），返回一个i32（结果指针）。

3. 使用上面的call_wasm_encrypt_robust，成功生成了签名。

简化版代码（脱敏后）：

python

import requests from wasmtime import Store, Module, Instance def video_sign(data_string): store = Store() module = Module.from_file(store.engine, "video_sign.wasm") instance = Instance(store, module, []) memory = instance.exports(store)["memory"] sign_func = instance.exports(store)["sign"] data_bytes = data_string.encode() ptr = 0x100000 # 临时选择空闲地址 memory.write(store, ptr, data_bytes) # 调用签名函数 result_ptr = sign_func(store, ptr, len(data_bytes)) # 结果字符串以长度+内容存储 res_len = int.from_bytes(memory.read(store, result_ptr, 4), 'little') res_bytes = memory.read(store, result_ptr+4, res_len) return res_bytes.decode() # 使用 sign_str = video_sign("user_id=12345&timestamp=1678900000") print(sign_str) # 输出类似 f5a3b2c1...

后来我发现，每次调用sign之前需要先调用init传入一个动态token，否则签名结果错误。所以我们还需要支持多步骤调用和状态保持。

五、处理带状态的Wasm模块

很多Wasm模块不是纯函数——它们会在内存中维持一个状态（比如加密的上下文、session key）。这就要求我们在Python中保持store和instance的存活，而不是每次调用都重新实例化。

正确做法：封装一个类，复用Store和Instance。

python

class WasmEncryptor: def __init__(self, wasm_path): self.store = Store() self.module = Module.from_file(self.store.engine, wasm_path) self.instance = Instance(self.store, self.module, []) self.memory = self.instance.exports(self.store)["memory"] def call_func(self, func_name, input_str, input_param_count=2, output_type="string"): # 通用调用逻辑（省略内存分配细节） pass def sign(self, data): # 专用于sign函数 data_bytes = data.encode() # 假设我们事先知道一个可用的内存区域 ptr = 0x200000 self.memory.write(self.store, ptr, data_bytes) sign_func = self.instance.exports(self.store)["sign"] res_ptr = sign_func(self.store, ptr, len(data_bytes)) # 读取结果（假设结构: 4字节长度 + 数据） res_len = int.from_bytes(self.memory.read(self.store, res_ptr, 4), 'little') res_data = self.memory.read(self.store, res_ptr+4, res_len) return res_data.decode() def set_init_token(self, token): init_func = self.instance.exports(self.store)["init"] token_bytes = token.encode() ptr = 0x100000 self.memory.write(self.store, ptr, token_bytes) init_func(self.store, ptr, len(token_bytes)) # 使用 enc = WasmEncryptor("video_sign.wasm") enc.set_init_token("server_provided_token_xyz") for i in range(10): sig = enc.sign(f"request_{i}") print(sig)

这样就不用每次都重新加载wasm，效率提升明显。

六、处理Wasm导出函数的多种参数和返回值类型

前面的例子都假设了字符串输入输出。真实场景的Wasm导出函数可能接收和返回整数、浮点数，或者直接修改线性内存而没有返回值。

1. 纯数值计算

假设Wasm导出一个乘法函数multiply(a: f64, b: f64) -> f64。调用方式：

python

func = instance.exports(store)["multiply"] result = func(store, 3.14, 2.0) print(result) # 6.28

wasmtime会自动转换Python数值到Wasm类型，前提是你知道函数签名。

2. 无返回值（void）

python

func = instance.exports(store)["process_data"] func(store, ptr, length) # 直接调用，无返回值

3. 返回两个整数（通常编码成一个i64）

有些Wasm模块为了效率，把两个32位整数打包成一个64位整数返回。例如返回{offset: 100, length: 20}，编码成(length << 32) | offset。

解包方式：

python

packed = func(store, ptr, length) # 返回Python int offset = packed & 0xFFFFFFFF length = packed >> 32

4. 通过内存返回数组/缓冲区

这是最常见的。函数不直接返回数据，而是把结果写进调用者提供的内存缓冲区。

python

def call_with_buffer(func, input_bytes, output_size=1024): # 分配输入和输出缓冲区 input_ptr = 0x300000 output_ptr = 0x310000 memory.write(store, input_ptr, input_bytes) # 调用函数，传入输入指针、输出指针、输出缓冲区大小 func(store, input_ptr, len(input_bytes), output_ptr, output_size) # 从output_ptr读取结果 result_bytes = memory.read(store, output_ptr, output_size) # 通常结果以空结尾，截取 result = result_bytes.split(b'\x00')[0] return result

七、处理Wasm导入的JS函数

更复杂的情况：Wasm模块可能导入了一些JS函数（例如console.log、Math.random、或者自定义的get_time）。当我们在Python中实例化这个Wasm模块时，必须提供这些导入函数的Python实现，否则实例化会失败。

步骤：

1. 使用wasmtime的Func包装Python函数。

2. 构造导入对象字典（Linker）。

3. 实例化时传入。

示例：Wasm模块导入了一个log_string(ptr, len)函数，我们在Python中实现它。

python

from wasmtime import Linker, Func, FuncType, ValType store = Store() linker = Linker(store.engine) # 定义导入函数的具体实现（Python函数） def wasm_log(ptr, length): # 这里需要访问memory才能读取字符串 # 但注意Python函数本身拿不到store和memory，需要用闭包 pass # 正确做法：在调用时通过store来读内存 # 实际代码中更复杂，wasmtime的API设计需要你把memory作为额外参数 # 简化版（演示用，实际应该用Func.from_callback并捕获store和memory） linker.define_func("env", "log_string", FuncType([ValType.i32(), ValType.i32()], []), lambda ptr, length: None)

完整正确的做法需要结合Store和Memory的闭包，代码量较大，限于篇幅不展开。但核心思路是：通过linker定义所有导入，再调用Instance时传入linker。

八、内存分配问题的深入探讨

前面很多例子用了固定地址写入字符串，这在真实Wasm中是很危险的——可能会覆盖Wasm模块自身的代码或静态数据。安全的方式是让Wasm模块自己管理内存，也就是调用它导出的malloc和free。

观察Wasm模块（用wasm2wat反编译）可以看到是否有(export "malloc" (func $malloc))。

如果没有，也可以自己向Wasm模块注入一个简易的内存分配器（不推荐，容易破坏模块）。最佳实践是选择那些导出内存管理函数的Wasm模块。

以下是一个完整的动态内存分配调用示例：

python

def dynamic_call(wasm_path, func_name, input_str): store = Store() module = Module.from_file(store.engine, wasm_path) # 实例化时先不传导入（假设无导入） instance = Instance(store, module, []) memory = instance.exports(store)["memory"] malloc = instance.exports(store)["malloc"] free = instance.exports(store)["free"] target = instance.exports(store)[func_name] # 分配输入内存 input_bytes = input_str.encode() input_ptr = malloc(store, len(input_bytes) + 1) memory.write(store, input_ptr, input_bytes) memory.write(store, input_ptr + len(input_bytes), b'\x00') # 调用目标函数（假设返回结果指针） result_ptr = target(store, input_ptr, len(input_bytes)) # 读取结果：假设结果字符串前4字节是长度 res_len = int.from_bytes(memory.read(store, result_ptr, 4), 'little') res_bytes = memory.read(store, result_ptr + 4, res_len) result_str = res_bytes.decode() # 释放内存 free(store, input_ptr) free(store, result_ptr) return result_str