simonw/micropython-wasm

GitHub: simonw/micropython-wasm

利用 Wasmtime 在 WebAssembly 沙盒中运行 MicroPython 代码片段的 Python 库,提供资源限制和会话持久化能力。

Stars: 152 | Forks: 12

# micropython-wasm [![PyPI](https://img.shields.io/pypi/v/micropython-wasm.svg)](https://pypi.org/project/micropython-wasm/) [![测试](https://static.pigsec.cn/wp-content/uploads/repos/2026/06/8e4835de60122743.svg)](https://github.com/simonw/micropython-wasm/actions/workflows/test.yml) [![更新日志](https://img.shields.io/github/v/release/simonw/micropython-wasm?include_prereleases&label=changelog)](https://github.com/simonw/micropython-wasm/releases) [![许可证](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](https://github.com/simonw/micropython-wasm/blob/main/LICENSE) MicroPython 被打包为 WASI WebAssembly 模块,并使用 Wasmtime 从 Python 中执行。 有关此项目的背景,请参阅[使用 MicroPython 和 WASM 在沙盒中运行 Python 代码](https://simonwillison.net/2026/Jun/6/micropython-in-a-sandbox/)。 本项目是一个实验性的 Python 包,用于在全新的 WebAssembly 沙盒中运行小片段的 MicroPython。它的设计围绕: - 一个自定义的 MicroPython WASI 产物,而不是 Emscripten 浏览器/Node 构建。 - 官方的 `wasmtime` Python 包。 - 用于一次性执行的全新 Wasmtime 实例,以及由后台线程支持的可选持久会话 API。 - 除非预先打开了显式的只读目录,否则不进行主机文件系统访问。 - 没有网络访问能力。 - 可配置的 WebAssembly 内存、fuel 和挂钟时间控制。 ## 安装 从 PyPI 安装: ``` pip install micropython-wasm ``` 对于本地开发,请使用 `uv`: ``` git clone https://github.com/simonw/micropython-wasm cd micropython-wasm uv run pytest ``` ## 快速开始 你可以从命令行在 WASM 中运行 MicroPython: ``` micropython-wasm -c "print(1 + 1)" micropython-wasm script.py micropython-wasm ``` 你也可以使用 `uvx` 在不预先安装的情况下运行 `micropython-wasm`: ``` uvx micropython-wasm --help ``` 无参数形式会启动一个简单的 REPL,并在提示符之间保持持久状态。使用 `--memory` 以字节为单位设置 WebAssembly 内存限制,使用 `--fuel` 设置 Wasmtime fuel 预算: ``` micropython-wasm --memory 33554432 --fuel 20000000 -c "print('hello')" ``` 要在 Python 中使用它,请从 `micropython_wasm` 包中导入代码: ``` from micropython_wasm import run result = run("print(1 + 1)") print(result.stdout) ``` 输出: ``` 2 ``` `run()` 返回一个 `RunResult`: ``` from micropython_wasm import run result = run("print('hello')") print(result.stdout) # "hello\n" print(result.stderr) # "" print(result.fuel_remaining) # integer Wasmtime fuel count ``` 每次调用都会创建一个新的 engine、store、WASI 配置、module 实例和 MicroPython 进程。全局变量和导入在调用之间不会持久化。 对于具有真正驻留 MicroPython 状态的有状态用法,请使用 `MicroPythonSession`: ``` from micropython_wasm import MicroPythonSession with MicroPythonSession() as session: print(session.run("x = 10\nprint(x)").stdout) print(session.run("x += 5\nprint(x)").stdout) print(session.run("print(x * 2)").stdout) ``` 输出: ``` 10 15 30 ``` 你也可以在没有上下文管理器的情况下使用同一个对象,这在交互式 Python REPL 中非常方便: ``` from micropython_wasm import MicroPythonSession session = MicroPythonSession() session.run("x = 10") session.run("print(x)") session.close() ``` ## API ### `run(code, ...)` 使用内置的产物运行 MicroPython 源代码: ``` from micropython_wasm import run result = run( "print(sum(range(10)))", memory_bytes=16 * 1024 * 1024, fuel=20_000_000, wall_timeout_seconds=1.0, host_result_bytes=256 * 1024, ) ``` 参数: - `code`:作为 `micropython -c ` 传递的 MicroPython 源代码。 - `wasm_path`:自定义 WASI MicroPython 产物的可选路径。如果省略,则使用 `micropython_wasm/artifacts/micropython-wasi.wasm`。 - `memory_bytes`:store 的最大 WebAssembly 线性内存。 - `fuel`:Wasmtime fuel 预算。当耗尽时,Guest 执行将发生 trap。 - `wall_timeout_seconds`:挂钟超时。传入 `None` 以禁用 epoch 中断。 - `readonly_dir`:要在 guest 内部作为 `/input` 公开的可选主机目录,具有只读的 WASI 目录和文件权限。 - `host_functions`:主机函数名到 Python 可调用对象的可选映射。这启用了底层的 `host.call(name, payload_json)` 桥接。 - `host_result_bytes`:最大序列化主机回调响应大小。默认值为 `256 * 1024`。 ### `run_micropython_wasi(code, wasm_path, ...)` 针对显式的 `.wasm` 产物运行代码: ``` from micropython_wasm import run_micropython_wasi result = run_micropython_wasi( "print(2 ** 8)", "micropython_wasm/artifacts/micropython-wasi.wasm", ) ``` 这在将本地重新构建的 MicroPython 产物复制到包中之前对其进行测试时非常有用。 ### `MicroPythonSession(...)` 创建一个在后台 Python 线程中运行的持久 MicroPython VM: ``` from micropython_wasm import MicroPythonSession session = MicroPythonSession() session.run("x = 10") session.run("x += 5") result = session.run("print(x)") print(result.stdout) session.close() ``` `MicroPythonSession` 在第一次 `run()` 时延迟启动。一个引导循环在 MicroPython 内部运行,并反复回调 Python 主机以获取下一个代码片段。每个片段都在同一个 MicroPython VM 中使用 `exec(..., globals())` 执行,因此变量、导入、函数、类和活跃对象确实会在调用之间保持驻留。 `MicroPythonSession` 接受与 `run()` 相同的资源、文件系统和主机函数参数: ``` session = MicroPythonSession( memory_bytes=16 * 1024 * 1024, fuel=20_000_000, readonly_dir="fixtures", host_functions={"add": lambda a, b: a + b}, host_result_bytes=256 * 1024, ) ``` 方法和属性: - `session.run(code)`:在驻留的 VM 中运行代码并返回一个 `RunResult`。 - `session.register_function(func, name=None)`:将一个 Python 函数暴露给 MicroPython 代码,可选择使用自定义名称。 - `session.close()`:向 guest 循环发送关闭消息并拒绝进一步的运行。 - `session.closed`:`close()` 后为 `True`。 - `session.host_functions`:已注册主机函数的副本。 - 上下文管理器支持:`with MicroPythonSession() as session: ...`。 每次 `session.run()` 请求都会刷新 fuel。如果某个片段耗尽了 fuel 或以其他方式导致 guest trap,后台 VM 将停止,该会话应被丢弃。 ### `MicroPythonReplaySession(...)` 创建一个由记录支持的 session 对象,其基本的 `run()` 和 `close()` 形状与 `MicroPythonSession` 相同,但不会在后台线程中保持 MicroPython VM 运行。 `MicroPythonReplaySession` 不会运行后台线程,也不会在调用之间保持活跃的 MicroPython VM。每次 `run()` 调用都会执行一个全新的 Wasmtime 实例,并在该命令式执行完成后返回。为了从调用者的角度保留变量、函数、类和导入,它会在每个新片段之前通过重播之前成功的片段来重建状态。 ``` from micropython_wasm import MicroPythonReplaySession session = MicroPythonReplaySession() session.run(""" import math def hypotenuse(a, b): return math.sqrt(a * a + b * b) """) result = session.run("print(hypotenuse(3, 4))") print(result.stdout) session.close() ``` `MicroPythonReplaySession` 接受与 `run()` 相同的资源 和文件系统参数: ``` session = MicroPythonReplaySession( memory_bytes=16 * 1024 * 1024, fuel=20_000_000, wall_timeout_seconds=1.0, readonly_dir="fixtures", host_result_bytes=256 * 1024, ) ``` 方法和属性: - `session.run(code)`:运行代码并返回一个 `RunResult`。 - `session.close()`:清除记录并拒绝进一步的运行。 - `session.closed`:`close()` 后为 `True`。 - `session.snippets`:当前保留的成功片段元组。 - 上下文管理器支持:`with MicroPythonReplaySession() as session: ...`。 仅保留成功的片段。如果某个片段非零退出或发生 trap,则不会将其添加到 session 记录中: ``` from micropython_wasm import MicroPythonReplaySession, MicroPythonWasmError session = MicroPythonReplaySession() session.run("x = 1") try: session.run("x = 2\nraise ValueError('boom')") except MicroPythonWasmError: pass print(session.run("print(x)").stdout) # "1\n" ``` 对于 `MicroPythonReplaySession`,每次 `session.run()` 调用都会创建一个新的 guest 实例,重播之前成功的片段,发出一个内部标记,然后运行新片段并仅返回该标记之后的输出。这从调用者的角度保留了普通的 Python 状态,包括变量、函数、类和导入,但之前的片段会在每次调用时在内部重新执行。 如果之前的片段执行了副作用(例如写入文件、进行依赖于时间的计算、消耗随机性或改变外部主机状态),这种重播行为就非常重要。当你想要真正的驻留 VM 内持久性时,请使用 `MicroPythonSession`。 例如,这会在第二次 `session.run()` 期间再次调用 Python `record()` 函数,因为第一个片段会在执行 `print(count)` 之前被重播: ``` from micropython_wasm import MicroPythonReplaySession calls = [] def record(value): calls.append(value) return len(calls) session = MicroPythonReplaySession(host_functions={"record": record}) session.run("count = record('once')") session.run("print(count)") session.close() print(calls) # ["once", "once"] ``` ### 主机函数 `MicroPythonSession` 和 `MicroPythonReplaySession` 可以将常规的 Python 函数暴露给 MicroPython 代码。注册一个函数,然后在 guest 内部按名称调用它: ``` from micropython_wasm import MicroPythonSession def add(a, b): return a + b session = MicroPythonSession() session.register_function(add) result = session.run("print(add(2, 3))") print(result.stdout) ``` 输出: ``` 5 ``` 如果 Python 可调用对象已经具有你想要暴露的名称,请直接传递它: ``` def shout(value): return value.upper() + "!" session = MicroPythonSession() session.register_function(shout) print(session.run("print(shout('hello'))").stdout) ``` 要使用不同的 MicroPython 名称暴露函数,请传递 `name=`: ``` def add(a, b): return a + b session = MicroPythonSession() session.register_function(add, name="plus") print(session.run("print(plus(2, 3))").stdout) ``` 你还可以在构造 session 时提供函数: ``` def format_name(first, last, uppercase=False): result = f"{first} {last}" if uppercase: result = result.upper() return result session = MicroPythonSession( host_functions={"format_name": format_name}, ) print(session.run("print(format_name('Ada', last='Lovelace', uppercase=True))").stdout) ``` 参数和返回值以 JSON 形式跨越 WebAssembly 边界。因此,支持的值是兼容 JSON 的值:`None`、布尔值、数字、字符串、列表以及带有字符串键的字典。 Python 端的异常会被返回给 MicroPython 包装器,并作为 `RuntimeError` 抛出,因此 guest 代码可以捕获它们: ``` def fail(): raise ValueError("bad host value") session = MicroPythonSession(host_functions={"fail": fail}) result = session.run(""" try: fail() except RuntimeError as ex: print(str(ex)) """) print(result.stdout) ``` 输出: ``` ValueError: bad host value ``` 在底层,内置的 MicroPython 产物包含一个名为 `host` 的微型内置模块。该模块从 Wasmtime 导入 `micropython_wasm.host_call`,并暴露了一个底层的 `host.call(name, payload_json)` 函数。session API 在那个底层桥接之上构建了友好的 MicroPython 包装器。 一次性执行的 `run()` 和 `run_micropython_wasi()` 也接受 `host_functions` 映射,但它们不会自动定义友好的包装器。它们暴露底层的 `host` 模块: ``` from micropython_wasm import run def add(a, b): return a + b result = run( """ import host print(host.call("add", '{"args": [2, 3], "kwargs": {}}')) """, host_functions={"add": add}, ) print(result.stdout) ``` ### `default_wasm_path()` 返回包预期的产物路径: ``` from micropython_wasm import default_wasm_path print(default_wasm_path()) ``` ### 异常 该包会抛出: - `MicroPythonWasmArtifactNotFound` 如果配置的产物不存在。 - `MicroPythonSessionClosed` 如果在 `session.close()` 之后调用 `session.run()`。 - `MicroPythonWasmError` 用于 guest trap、非零的 guest 退出、无效的产物、缺少 Wasmtime 支持或无效的预打开目录。 - `ValueError` 用于无效的主机端资源限制。 例如: ``` from micropython_wasm import MicroPythonWasmError, run try: run('raise ValueError("boom")') except MicroPythonWasmError as ex: print(ex) ``` ## 文件系统访问 默认情况下,guest 没有预打开的主机目录: ``` from micropython_wasm import run run("print('no files by default')") ``` 要公开输入文件,请将它们放在一个目录中并传递 `readonly_dir`: ``` from pathlib import Path from micropython_wasm import run fixtures = Path("fixtures") fixtures.mkdir(exist_ok=True) (fixtures / "example.txt").write_text("hello from the host\n") result = run( "print(open('/input/example.txt').read())", readonly_dir=fixtures, ) print(result.stdout) ``` 该目录被挂载到 WASI guest 的 `/input` 下。该包会向 Wasmtime 请求只读目录和文件权限。尝试在 `/input` 内部写入将会失败。 在运行不受信任的代码时,请勿预打开你的项目根目录、主目录、`/` 或共享的临时目录。 ## 资源控制 主机为每次执行配置这些 Wasmtime 控制: - `Store.set_limits(memory_size=...)` 限制 WebAssembly 线性内存。 - `Store.set_fuel(...)` 限制类 CPU 指令的进度。 - 当 `wall_timeout_seconds` 不为 `None` 时,将启用 epoch 中断。 - `Config.max_wasm_stack` 被设置为 `512 * 1024`。 示例: ``` from micropython_wasm import MicroPythonWasmError, run try: run( "while True:\n pass", fuel=50_000, wall_timeout_seconds=None, ) except MicroPythonWasmError as ex: print("stopped:", ex) ``` `memory_bytes` 限制的是 guest 线性内存,而不是主机进程的总 RSS。Wasmtime runtime 内存、编译后的代码、Python 进程内存和主机回调都不受该限制。对于高风险的多租户工作负载,请在单独的工作进程中运行每次执行,并使用操作系统级别的 CPU、内存和挂钟时间限制。 ## 网络访问 此包不公开网络导入或主机 socket 函数。当前的 MicroPython WASI 产物在 WASI 变体配置中也禁用了 socket 和 SSL 支持。 如果将来添加了网络访问,请优先使用范围狭窄的、主机中介的 API,例如带有显式允许列表、超时、重定向限制和最大响应大小的 `http_get(url)`。请勿向不受信任的代码公开原始 socket。 ## 支持的 Python 行为 MicroPython 不是 CPython。它实现了 Python 的一个重要子集,但在语法支持、标准库覆盖、对象行为和平台细节方面存在差异。 测试套件验证了有用的行为,包括: - 算术和大型整数。 - 字符串和字节。 - 列表、元组、字典和集合。 - 列表、字典、集合和生成器推导式。 - 函数、默认参数、仅限关键字参数、lambda、闭包和递归。 - 类、继承、`property`、`isinstance`。 - `try`/`except`/`finally` 和上下文管理器。 - `math`、`json`、`re`、`binascii`、`sys` 和 `os.listdir('/input')`。 - 调用之间的全新执行状态。 - 跨越 `MicroPythonReplaySession.run()` 调用的、由记录支持的 session 状态。 - 跨越 `MicroPythonSession.run()` 调用的真正驻留 VM 状态。 - 通过 session `register_function()` 方法进行的主机函数回调。 - 只读文件预打开。 - Fuel 耗尽。 该产物的已知观察结果: - `sys.platform` 报告为 `linux`。 - `sys.argv` 在 guest 内部为 `['-c']`。 - 内置产物中不提供 `hashlib.sha256`。 - 内置产物中不提供 `zlib`。 ### 列出可用模块 要查看 MicroPython 导入路径和内置产物可用的模块,请运行: ``` from micropython_wasm import run result = run( """ import sys print("sys.path:") for path in sys.path: print(" ", path) print() print("modules:") help("modules") """ ) print(result.stdout) ``` `help("modules")` 是最有用的 MicroPython 原生列表,因为它包含内置和冻结的模块,以及文件系统上可用的模块。对该产物中的 `sys.path` 进行普通的 `os.listdir()` 扫描会遗漏冻结的模块。 ## 重建 WASI 产物 构建辅助工具是: ``` scripts/build_micropython_wasi.py ``` 它将: 1. 将 MicroPython 克隆到 `/tmp/micropython-wasm-build/micropy`。 2. 检出请求的 ref,包括 GitHub PR refs,如 `pull/13676/head`。 3. 构建 `mpy-cross`。 4. 为 `ports/unix` 运行 `make submodules`。 5. 构建 `ports/unix VARIANT=wasi`。 6. 默认包含内置的 `host` 用户 C 模块。 7. 查找最佳的 wasm 产物。 8. 将其复制到 `micropython_wasm/artifacts/micropython-wasi.wasm`。 ### macOS ARM64 设置 安装 Binaryen: ``` brew install binaryen ``` 下载 `wasi-sdk` 25.0 到 `/tmp`: ``` curl -L -o /tmp/wasi-sdk-25.0-arm64-macos.tar.gz \ https://github.com/WebAssembly/wasi-sdk/releases/download/wasi-sdk-25/wasi-sdk-25.0-arm64-macos.tar.gz tar -xzf /tmp/wasi-sdk-25.0-arm64-macos.tar.gz -C /tmp ``` 构建产物: ``` uv run python scripts/build_micropython_wasi.py \ --ref pull/13676/head \ --wasi-sdk /tmp/wasi-sdk-25.0-arm64-macos ``` 使用 `--clean` 丢弃现有的 `/tmp` 检出: ``` uv run python scripts/build_micropython_wasi.py \ --clean \ --ref pull/13676/head \ --wasi-sdk /tmp/wasi-sdk-25.0-arm64-macos ``` 使用 `--skip-build` 从检出中重新复制已构建的产物: ``` uv run python scripts/build_micropython_wasi.py \ --skip-build \ --ref pull/13676/head \ --wasi-sdk /tmp/wasi-sdk-25.0-arm64-macos ``` ### 有用的构建选项 - `--repo-url`:备用 MicroPython 仓库。 - `--ref`:要构建的 git ref。支持像 `pull//head` 这样的 PR refs。 - `--work-dir`:备用构建检出目录。 - `--output`:复制产物的备用目标位置。 - `--variant`:备用 Unix 变体,默认为 `wasi`。 - `--wasi-sdk`:`wasi-sdk` 目录的路径。 - `--user-c-modules`:MicroPython `USER_C_MODULES` 目录的路径。 - `--jobs`:并行构建任务。 - `--extra-make-arg`:转发给 Unix make 命令的附加参数。 - `--clean`:在克隆之前移除检出。 - `--skip-build`:查找并复制现有产物而无需重新构建。 在 macOS 上,脚本在构建 `mpy-cross` 时传递了 `CFLAGS_EXTRA=-Wno-error=gnu-folding-constant`,因为实验性分支目前在使用 `-Werror` 时触发了此 Apple Clang 警告。 ### 产物选择 脚本按以下顺序优先选择产物: 1. `build-wasi/micropython.spilled.exnref` 2. `build-wasi/micropython.exnref` 3. `build-wasi/micropython.wasm` 4. `build-wasi/micropython` 当前的本地构建产生了: ``` /tmp/micropython-wasm-build/micropython/ports/unix/build-wasi/micropython /tmp/micropython-wasm-build/micropython/ports/unix/build-wasi/micropython.exnref micropython_wasm/artifacts/micropython-wasi.wasm ``` 复制的包产物是 `micropython.exnref` 输出。 ## 测试 运行完整的测试套件: ``` uv run pytest ``` 该套件包括包测试和针对内置 wasm 产物的运行时集成测试。如果产物缺失,运行时集成测试将被跳过,但包/构建脚本测试仍会运行。 手动测试自定义产物: ``` uv run python - <<'PY' from micropython_wasm import run_micropython_wasi result = run_micropython_wasi( "print(1 + 1)", "/path/to/micropython-wasi.wasm", ) print(result.stdout) PY ``` ## 安全说明 此包是 MicroPython WASI 命令模块的 Wasmtime 嵌入。它是一个有用的沙盒层,但就其本身而言,对于高风险的生产用途,它并不是一个完整的安全边界。 此包中合理的默认设置: - 每次运行都是全新实例。 - 不继承主机环境。 - 默认情况下没有预打开的主机目录。 - 仅限可选的只读预打开目录。 - 没有网络主机函数。 - Fuel 和内存限制。 - 可选的挂钟时间超时。 需要考虑的额外保护措施: - 在单独的工作进程中运行执行。 - 应用操作系统级别的内存和 CPU 限制。 - 将 worker 放入容器或其他隔离边界中。 - 如果大规模运行不受信任的代码,需限制 stdout/stderr 捕获大小。 - 避免暴露环境权限的主机回调。 - 保持 Wasmtime 和 MicroPython 产物固定并定期测试。 ## 实现状态和注意事项 该仓库目前包含一个可用的内置产物,位于: ``` micropython_wasm/artifacts/micropython-wasi.wasm ``` 该产物是由 MicroPython PR `#13676` 构建的,使用了 PR ref `pull/13676/head`。MicroPython 的 WASI Unix 变体在上游仍处于实验阶段,因此此包也应被视为实验性的。 测试套件针对内置产物验证了算术、字符串、字节、集合、推导式、函数、闭包、递归、类、异常、上下文管理器、一小部分标准库、全新实例隔离、只读文件访问和 fuel 中断。它还验证了两种有状态的 session API:由记录支持的 `MicroPythonReplaySession` 和由持久后台线程支持的 `MicroPythonSession`。 一个重要的构建注意事项:该 PR 的完整后链接 Binaryen 流程目前在 Binaryen 130 的 `wasm-opt --spill-pointers` 处失败。此仓库中的产物使用了成功的 `wasm-opt --translate-to-exnref` 后处理。简单和中等广度的 Python 执行可以在 Wasmtime 下运行,但在依赖它来处理恶意或长期运行的代码之前,应对其进行压力测试。 ## 许可证 Apache-2.0
标签:AI工具, MicroPython, Python, SOC Prime, Wasmtime, WebAssembly, 开发工具, 无后门, 沙箱, 逆向工具