Learning Hub

How to Run Untrusted Plugin Code in a WebAssembly Sandbox From Python With Wasmtime

Learn to run untrusted plugin code from Python inside a WebAssembly sandbox with Wasmtime, using an import allowlist, fuel, epoch deadlines, memory caps and a hardened host function, with every step...

October 2, 2026 37 Min Read
27

Sooner or later, most applications want to run code that somebody else wrote. A billing tool lets customers write their own discount rules. An editor loads community plugins. A coding assistant generates a helper function, and you would like to try it before a person has read every line. In each case the code is untrusted: you cannot assume it will do only what it claims to do.

Table Of Content

Untrusted code can go wrong in four ways. It can reach things it should not, such as a secret file or an environment variable. It can run forever and starve everything else. It can swallow memory until the host falls over. And, the one people forget, it can abuse the helper functions you hand it, so that it reaches the first problem through a side door.

In this tutorial you will build a small plugin runner in Python that deals with all four. The finished function, run_plugin(), takes the bytes of a WebAssembly module, calls one function inside it, and reports exactly one of six outcomes: ok (the plugin returned a value), rejected (the module asked for something we do not allow, or is not valid), out_of_fuel (it used up its work budget), timeout (it ran past its wall-clock deadline), trap (it did something illegal, such as dividing by zero) and host_error (one of our own helper functions raised). Every step has code you can run, the output I got, an explanation of what is happening, and a test at the end.

What WebAssembly and Wasmtime are, in plain language

WebAssembly (abbreviated Wasm) is, in the words of its home page, “a binary instruction format for a stack-based virtual machine”. Its security page says that “Each WebAssembly module executes within a sandboxed environment separated from the host runtime using fault isolation techniques.” The part that matters for this tutorial is what that sandbox leaves out. A Wasm program has no built-in way to open a file, make a network connection or read your environment variables. It can only compute, and it can call functions that the program running it chooses to supply. The Wasmtime documentation states it directly: “WebAssembly is inherently sandboxed by design (must import all functionality, etc).”

Wasmtime is a WebAssembly runtime, and the wasmtime package on PyPI lets Python programs use it. I tested everything below with wasmtime 49.0.0 (the current release on PyPI when I wrote this, uploaded on September 21, 2026), Python 3.13.14 and pytest 9.1.1 on Windows 11.

Key terms

A few words come up in every step, so here they are once, in plain language:

  • The host is your Python program. The guest is the WebAssembly code you are running.
  • A module is compiled guest code. An instance is a module that has been given its own state and is ready to run. The store holds that state for one run: the guest’s memory, its fuel and its limits. The engine is the compiler plus global settings.
  • An import is a function the guest asks the host to provide. An export is a function (or memory) the guest offers to the host.
  • Linear memory is one block of bytes that belongs to the guest. A “pointer” inside the guest is just a number: an offset into that block. Memory is sized in pages of 64 KiB.
  • A trap is a controlled, immediate stop when the guest does something illegal.
  • Fuel and epochs are the two tools Wasmtime gives you for stopping a guest that runs too long.

One honest scoping note. This tutorial is about the safety boundary, not about compiling a language to WebAssembly. Real plugins would come out of a compiler. To keep every step small and reproducible, the plugins here are written by hand in WebAssembly’s text format (WAT), and a helper called wat2wasm turns them into the binary form. You will not need to know WAT beforehand; I explain the first module line by line and keep the rest short.

Prerequisites

You need Python 3.9 or newer (the package’s PyPI page lists >=3.9), a terminal, and basic comfort with Python functions and classes. No WebAssembly knowledge is assumed, and no compiler is needed: on my Windows machine pip installed a prebuilt wheel. The outputs shown are from Windows 11. Two things differ on other systems, and I point them out where they appear: one error class name in Step 8, and timings, which vary on every machine.

Set up the project

Create a folder, a virtual environment, and install the two packages:

mkdir wasm-sandbox
cd wasm-sandbox
python -m venv .venv
# Windows PowerShell:  .venv\Scripts\Activate.ps1
# macOS or Linux:      source .venv/bin/activate
python -m pip install wasmtime pytest
python -m pip list

The package list should include these two lines (the versions I tested):

pytest    9.1.1
wasmtime  49.0.0

Every file in this tutorial goes in that one folder. When you finish, it will contain the step scripts, a few support files, and the test file:

wasm-sandbox/
  step1_exec_problem.py   step2_hello.py          step3_imports.py
  step4_fuel.py           step5_memory.py         step6_traps.py
  step7_wall_clock.py     step7b_guard_cost.py    step8_host_function.py
  step9_hardened_host.py  step9b_signed_args.py   step10_run_plugin.py
  reader_plugin.py        capabilities.py         sandbox.py
  test_sandbox.py

Step 1: See why exec() is not a sandbox

Before reaching for a new tool, it is worth seeing why the obvious one fails. The obvious approach in Python is to take the plugin’s source code and call exec() on it. The Python documentation for exec opens with a warning: “This function executes arbitrary code. Calling it with untrusted user-supplied input will lead to security vulnerabilities.” The next file shows three concrete ways it goes wrong.

Create step1_exec_problem.py. It first writes a fake secret file, then runs a “discount plugin” that quietly reads it. Next it tries the classic fix of taking open() away, and finally it runs a plugin that never returns.

# step1_exec_problem.py
# step1_exec_problem.py
import threading
import time

with open("secret.txt", "w", newline="\n") as handle:
    handle.write("db_password=hunter2\n")

# 1. A "discount plugin" that does more than it promised.
PLUGIN_SOURCE = """
def apply(cents):
    leaked = open("secret.txt").read()
    print("plugin read:", leaked.strip())
    return cents
"""
namespace = {}
exec(PLUGIN_SOURCE, namespace)
print("apply(1000) ->", namespace["apply"](1000))

# 2. Take away open(). Does that help?
restricted = {"__builtins__": {}}
exec(PLUGIN_SOURCE, restricted)
try:
    restricted["apply"](1000)
except NameError as exc:
    print("restricted builtins blocked it:", exc)

ESCAPE = (
    "[c for c in ().__class__.__base__.__subclasses__() "
    "if c.__name__ == 'catch_warnings'][0]()._module.__builtins__['open']"
    "('secret.txt').read()"
)
print("one expression later ->", repr(eval(ESCAPE, {"__builtins__": {}})))

# 3. A plugin that never returns.
LOOP_SOURCE = """
def apply(cents):
    while True:
        pass
"""
loop_namespace = {}
exec(LOOP_SOURCE, loop_namespace)
worker = threading.Thread(target=loop_namespace["apply"], args=(1000,), daemon=True)
worker.start()
worker.join(timeout=1.0)
print("join(timeout=1.0) returned; worker alive:", worker.is_alive())
time.sleep(0.5)
print("0.5 s later, worker alive:", worker.is_alive())

Run it with python step1_exec_problem.py. You should see:

plugin read: db_password=hunter2
apply(1000) -> 1000
restricted builtins blocked it: name 'open' is not defined
one expression later -> 'db_password=hunter2\n'
join(timeout=1.0) returned; worker alive: True
0.5 s later, worker alive: True

Here is what each part shows:

  • The plugin read the secret. Nothing stopped it. Code run through exec() has the same file access, environment variables and network as the rest of your program.
  • Removing builtins did not help. With an empty __builtins__, a direct call to open() fails with NameError, which looks like protection. But one expression, written entirely with attribute lookups, walks from an empty tuple up to object, lists every class Python knows about, picks catch_warnings, and reaches the real builtins through the warnings module. I show it as a reason, not a recipe: the Python documentation says plainly that “Overriding __builtins__ can be used to restrict or change the available names, but this is not a security mechanism: the executed code can still access all builtins.”
  • A runaway plugin cannot be stopped. The join(timeout=1.0) call gave up waiting after one second, but the worker thread is still alive half a second later and still using CPU time inside your process. Python offers no supported way to kill a thread from outside.

The lesson is that a real boundary has to sit outside the interpreter that runs the plugin’s code, and it has to start from “nothing is allowed”. A separate process helps with the runaway problem, since you can kill it, but it still inherits your user’s files and network unless you configure the operating system carefully. WebAssembly offers a different route: run the plugin in a runtime where the only things it can touch are the things you explicitly hand over.

Step 2: Run your first WebAssembly module

Now the same idea with Wasm. Create step2_hello.py. It defines the smallest useful module (a function that adds two numbers), compiles it, and calls it from Python.

# step2_hello.py
# step2_hello.py
from wasmtime import Engine, Instance, Module, Store, wat2wasm

ADD_WAT = """
(module
  (func (export "add") (param i32 i32) (result i32)
    local.get 0
    local.get 1
    i32.add))
"""

wasm_bytes = bytes(wat2wasm(ADD_WAT))
print("compiled", len(wasm_bytes), "bytes, header:", wasm_bytes[:4])

engine = Engine()
Module.validate(engine, wasm_bytes)  # raises WasmtimeError if the bytes are not valid Wasm
module = Module(engine, wasm_bytes)
store = Store(engine)
instance = Instance(store, module, [])  # this module needs no imports

add = instance.exports(store)["add"]
print("add(2, 40) =", add(store, 2, 40))
print("exports:", [export.name for export in module.exports])
print("imports:", [(item.module, item.name) for item in module.imports])

Run it. The output:

compiled 41 bytes, header: b'\x00asm'
add(2, 40) = 42
exports: ['add']
imports: []

The WAT text is a tiny program. (module ...) is one module. (func (export "add") (param i32 i32) (result i32) ...) defines a function named add, visible to the host as an export, that takes two 32-bit integers and returns one. WebAssembly is a stack machine: local.get 0 pushes the first parameter, local.get 1 pushes the second, and i32.add pops both and pushes their sum.

On the Python side, wat2wasm compiles that text into binary (41 bytes here, beginning with the four-byte header that every Wasm file starts with). Module.validate checks that bytes really are valid WebAssembly before you do anything else with them, and it raises WasmtimeError if they are not. Then four objects appear in order: an Engine, a Module compiled by that engine, a Store to hold the run’s state, and an Instance that ties the module to the store. You call an export with add(store, 2, 40). The store goes in as the first argument every time; remember that, because it comes back in Step 8.

The last two lines print what the module offers and what it asks for. This one exports add and imports nothing, so it cannot do anything except arithmetic. The empty import list is the whole security story in miniature, and the next step builds on it.

Step 3: Treat imports as the only door

A WebAssembly module declares, ahead of time, every outside function it wants to call. The Wasmtime documentation spells out the consequence: “All interaction with the outside world is done through imports and exports. There is no raw access to system calls or other forms of I/O; the only thing a WebAssembly instance can do is what is available through interfaces it has been explicitly linked with.” So the list of imports is the complete list of favors the guest can ask for, and you can read it before running a single instruction.

Create step3_imports.py. It defines two modules: LOGGER_WAT, which asks the host for a function named log, and SNEAKY_WAT, which asks for something called system. It then tries four things.

# step3_imports.py
# step3_imports.py
from wasmtime import Engine, FuncType, Instance, Linker, Module, Store, ValType, WasmtimeError, wat2wasm

LOGGER_WAT = """
(module
  (import "host" "log" (func $log (param i32)))
  (func (export "run") (param i32)
    local.get 0
    call $log))
"""

SNEAKY_WAT = """
(module
  (import "env" "system" (func $system (param i32) (result i32)))
  (memory (export "memory") 1)
  (data (i32.const 0) "calc.exe")
  (func (export "run") (result i32)
    i32.const 0
    call $system))
"""

engine = Engine()
logger = Module(engine, wat2wasm(LOGGER_WAT))
print("logger asks for:", [(item.module, item.name) for item in logger.imports])

# 1. Instantiating without supplying the import fails.
store = Store(engine)
try:
    Instance(store, logger, [])
except WasmtimeError as exc:
    print("no import supplied ->", exc)

# 2. A Linker states, by name, exactly what the host offers.
seen = []
linker = Linker(engine)
linker.define_func("host", "log", FuncType([ValType.i32()], []), lambda value: seen.append(value))
instance = linker.instantiate(store, logger)
instance.exports(store)["run"](store, 42)
print("host saw:", seen)

# 3. A module that asks for something the host never defined.
sneaky = Module(engine, wat2wasm(SNEAKY_WAT))
try:
    linker.instantiate(store, sneaky)
except WasmtimeError as exc:
    print("sneaky module ->", exc)

# 4. Better: read the import list first and compare it with an allowlist.
ALLOWED_IMPORTS = {("host", "log")}


def disallowed_imports(module):
    return [
        (item.module, item.name)
        for item in module.imports
        if (item.module, item.name) not in ALLOWED_IMPORTS
    ]


for name, module in (("logger", logger), ("sneaky", sneaky)):
    print(name, "->", disallowed_imports(module) or "ok")

Run it:

logger asks for: [('host', 'log')]
no import supplied -> expected 1 imports, found 0
host saw: [42]
sneaky module -> unknown import: `env::system` has not been defined
logger -> ok
sneaky -> [('env', 'system')]

Walking through the output:

  1. No import supplied. Instantiating the logger with an empty import list fails with expected 1 imports, found 0. The module cannot start without what it asked for.
  2. A Linker states the offer by name. A linker is a registry of everything the host is willing to provide, keyed by module name and function name. Here the host offers exactly one thing, host.log, which appends to a list. The logger ran, called it with 42, and the host saw [42].
  3. The sneaky module cannot start. It asks for env.system, a function the host never defined, so instantiation fails with unknown import: `env::system` has not been defined. That is default-deny: there is no ambient system() for it to find.
  4. An allowlist check before instantiating. Failing at instantiation is already safe, but it is better to refuse before doing any work and to say why. module.imports lists every import without running anything, so disallowed_imports() can compare it with an allowlist. The logger passes; the sneaky module is reported as asking for ('env', 'system').

Keep that allowlist idea. The final runner uses it as its first gate.

Step 4: Stop runaway code with fuel

Imports control what a plugin can reach, not how long it takes. For that, Wasmtime has fuel: a budget of work. Each instruction the guest executes spends fuel, and when the tank is empty the guest traps. The Wasmtime Config documentation says fuel “can be used to deterministically prevent infinitely-executing WebAssembly code by instrumenting generated code to consume fuel as it executes.”

Two settings are needed. First, turn fuel on in the engine’s configuration with config.consume_fuel = True. Second, put fuel into the store with store.set_fuel(n). A Config is used up by the engine you build from it, so make a fresh one each time (reusing one raised ValueError: already closed when I tried).

Create step4_fuel.py. It runs an infinite loop, measures how much fuel a counting loop uses, and then demonstrates three mistakes people make: one fuel pool shared across calls, a store with no fuel, and a large memory operation.

# step4_fuel.py
# step4_fuel.py
import time

from wasmtime import Config, Engine, Instance, Module, Store, Trap, wat2wasm

SPIN_WAT = '(module (func (export "spin") (loop $forever br $forever)))'

COUNT_WAT = """
(module
  (func (export "count") (param $n i32) (result i32)
    (local $i i32)
    (loop $again
      local.get $i
      i32.const 1
      i32.add
      local.set $i
      local.get $i
      local.get $n
      i32.lt_u
      br_if $again)
    local.get $i))
"""

FILL_WAT = """
(module
  (memory (export "memory") 1024)
  (func (export "fill") (param $bytes i32)
    i32.const 0
    i32.const 255
    local.get $bytes
    memory.fill))
"""


def instantiate(wat, fuel):
    config = Config()
    config.consume_fuel = True  # without this, set_fuel() raises an error
    engine = Engine(config)
    store = Store(engine)
    store.set_fuel(fuel)
    return store, Instance(store, Module(engine, wat2wasm(wat)), [])


def why(trap):
    return trap.trap_code.name + ": " + trap.message.rsplit("wasm trap:", 1)[-1].strip()


# 1. An infinite loop is stopped when the fuel runs out.
store, instance = instantiate(SPIN_WAT, 1_000_000)
try:
    instance.exports(store)["spin"](store)
except Trap as trap:
    print("spin ->", why(trap), "| fuel left:", store.get_fuel())

# 2. Fuel use is deterministic: about 8 units per loop iteration.
for n in (10, 1000, 100_000):
    used = []
    for _ in range(2):
        store, instance = instantiate(COUNT_WAT, 10_000_000)
        instance.exports(store)["count"](store, n)
        used.append(10_000_000 - store.get_fuel())
    print(f"count({n}) -> fuel used, two runs: {used}")

# 3. One store, one pool: a budget set once drains across calls.
store, instance = instantiate(COUNT_WAT, 200)
count = instance.exports(store)["count"]
for call in (1, 2, 3):
    try:
        count(store, 10)
        print(f"call {call}: ok, fuel left {store.get_fuel()}")
    except Trap as trap:
        print(f"call {call}:", why(trap))

# 4. A store with no fuel traps immediately.
store, instance = instantiate(COUNT_WAT, 0)
try:
    instance.exports(store)["count"](store, 1)
except Trap as trap:
    print("zero fuel ->", why(trap))

# 5. Fuel is charged per byte for bulk memory operations.
for budget in (10_000_000, 100_000_000):
    store, instance = instantiate(FILL_WAT, budget)
    started = time.perf_counter()
    try:
        instance.exports(store)["fill"](store, 64 * 1024 * 1024)
        millis = (time.perf_counter() - started) * 1000
        print(f"fill 64 MiB with budget {budget:,}: ok, fuel used {budget - store.get_fuel():,}, {millis:.0f} ms")
    except Trap as trap:
        print(f"fill 64 MiB with budget {budget:,}:", why(trap))

Run it. The output:

spin -> OUT_OF_FUEL: all fuel consumed by WebAssembly | fuel left: 0
count(10) -> fuel used, two runs: [82, 82]
count(1000) -> fuel used, two runs: [8002, 8002]
count(100000) -> fuel used, two runs: [800002, 800002]
call 1: ok, fuel left 118
call 2: ok, fuel left 36
call 3: OUT_OF_FUEL: all fuel consumed by WebAssembly
zero fuel -> OUT_OF_FUEL: all fuel consumed by WebAssembly
fill 64 MiB with budget 10,000,000: OUT_OF_FUEL: all fuel consumed by WebAssembly
fill 64 MiB with budget 100,000,000: ok, fuel used 67,108,869, 13 ms

What the lines tell you:

  • The infinite loop stops. The guest ran until fuel hit zero, then trapped with OUT_OF_FUEL. The loop never returned, yet your Python code got control back and a clear reason.
  • Fuel use is deterministic. The counting loop costs 8 fuel per iteration, plus a small constant: 82 for 10 iterations, 8,002 for 1,000, and 800,002 for 100,000, identical on both runs. Eight is no accident: the loop body is eight instructions, and the Wasmtime documentation says “Most WebAssembly instructions consume 1 unit of fuel” (structural markers such as loop are free). The Wasmtime page on interrupting execution explains why this matters: “Fuel-based interruption is completely deterministic: the same program run with the same amount of fuel will always be interrupted at the same location in the program”. The same plugin on the same input gets the same verdict every time.
  • One store means one pool of fuel. With 200 units, the first call cost 82 (118 left), the second cost 82 (36 left), and the third needed 82 but had 36, so it trapped. If you set fuel once and reuse a store for many requests, a plugin will seem to fail at random after enough calls. Either refill with set_fuel before every call or, as the final runner does, create a fresh store per run.
  • No fuel means no execution. The Wasmtime Store documentation says “By default a Store starts with 0 fuel for wasm to execute with (meaning it will immediately trap).” Forgetting set_fuel shows up as an OUT_OF_FUEL trap on the very first call.
  • Big operations cost big fuel. One memory.fill of 64 MiB consumed 67,108,869 units in my run, roughly one per byte, so a 10,000,000 budget refused it and a 100,000,000 budget allowed it (about 13 ms here). The Wasmtime docs describe a configurable cost for operators whose work depends on a runtime operand, with “per-byte, per-element, and per-page costs”; I measured this one operator and did not look further, so treat exact fuel numbers as specific to one Wasmtime version.

Step 5: Cap how much memory a plugin can take

A guest can ask for more memory with the memory.grow instruction. Left alone, it can keep asking. The store’s limits stop that. In the wasmtime-py documentation, Store.set_limits(memory_size=...) is described like this: “Setting this to a lower value will cause instantiation to fail if a module needs more memory. Additionally the memory.grow instruction will return -1 once this threshold is reached.” The same method also takes table_elements, instances, tables and memories limits.

Create step5_memory.py. It runs a module with one exported function, grow(pages), three times: with no limit, with a four-page limit, and with a module that declares more memory than the limit allows.

# step5_memory.py
# step5_memory.py
from wasmtime import Engine, Instance, Module, Store, WasmtimeError, wat2wasm

GROW_WAT = """
(module
  (memory (export "memory") 1)
  (func (export "grow") (param $pages i32) (result i32)
    local.get $pages
    memory.grow))
"""

BIG_WAT = '(module (memory (export "memory") 2000))'

PAGE = 64 * 1024  # a WebAssembly page is 64 KiB
engine = Engine()
grow_module = Module(engine, wat2wasm(GROW_WAT))

# 1. No limit: the guest can simply ask for more.
store = Store(engine)
instance = Instance(store, grow_module, [])
grow = instance.exports(store)["grow"]
memory = instance.exports(store)["memory"]
print("no limit, grow(1000) ->", grow(store, 1000), "(previous size in pages)")
print("memory is now", memory.size(store), "pages =", round(memory.data_len(store) / (1024 * 1024), 1), "MiB")

# 2. A four-page limit: growth past it returns -1 instead of trapping.
store = Store(engine)
store.set_limits(memory_size=4 * PAGE)
instance = Instance(store, grow_module, [])
grow = instance.exports(store)["grow"]
results = [grow(store, 2), grow(store, 2), grow(store, 1), grow(store, 0)]
print("limit 4 pages, grow(2), grow(2), grow(1), grow(0) ->", results)

# 3. A module that declares more memory than the limit allows never starts.
store = Store(engine)
store.set_limits(memory_size=16 * PAGE)
try:
    Instance(store, Module(engine, wat2wasm(BIG_WAT)), [])
except WasmtimeError as exc:
    print("2000-page module under a 16-page limit ->", exc)

Run it:

no limit, grow(1000) -> 1 (previous size in pages)
memory is now 1001 pages = 62.6 MiB
limit 4 pages, grow(2), grow(2), grow(1), grow(0) -> [1, -1, 3, 4]
2000-page module under a 16-page limit -> memory minimum size of 2000 pages exceeds memory limits

Without a limit, the guest grew from 1 page to 1,001 pages (62.6 MiB; one page is 64 KiB) just by asking. grow returns the previous size in pages, which is why it printed 1. With a limit of four pages, the four calls returned [1, -1, 3, 4]: the first grew from 1 to 3 pages and returned 1, the second would have reached 5 pages so it was refused with -1, the third grew from 3 to 4 and returned 3, and grow(0) reported the current size, 4. Note that a refusal is not a trap. The guest gets -1 and keeps running, and well-written code is expected to handle it. The third case shows the other half of the limit: a module that declares 2,000 pages up front never starts, and instantiation fails with memory minimum size of 2000 pages exceeds memory limits.

Step 6: Let every failure be a trap, not a crash

When a guest does something illegal, WebAssembly does not corrupt your process; it stops the guest. The WebAssembly security page explains that “Traps are used to immediately terminate execution and signal abnormal behavior to the execution environment”, and lists what can cause one: for example exceeding the call stack, accessing out-of-bounds memory, or dividing by zero. In wasmtime-py every trap arrives as a Python Trap exception with a trap_code that tells you why.

Create step6_traps.py. It runs four small bad modules and one good one, each in its own store, and prints what happened.

# step6_traps.py
# step6_traps.py
from wasmtime import Engine, Instance, Module, Store, Trap, wat2wasm

CASES = {
    "divide": (
        '(module (func (export "run") (param i32 i32) (result i32)'
        " local.get 0 local.get 1 i32.div_s))",
        (1, 0),
    ),
    "unreachable": ('(module (func (export "run") unreachable))', ()),
    "peek": (
        '(module (memory 1) (func (export "run") (param i32) (result i32)'
        " local.get 0 i32.load))",
        (70000,),
    ),
    "recurse": ('(module (func $again (export "run") (call $again)))', ()),
    "add": (
        '(module (func (export "run") (param i32 i32) (result i32)'
        " local.get 0 local.get 1 i32.add))",
        (2, 3),
    ),
}

engine = Engine()
for label, (wat, args) in CASES.items():
    store = Store(engine)
    instance = Instance(store, Module(engine, wat2wasm(wat)), [])
    try:
        result = instance.exports(store)["run"](store, *args)
        print(f"{label:12} -> returned {result}")
    except Trap as trap:
        reason = trap.message.rsplit("wasm trap:", 1)[-1].strip()
        print(f"{label:12} -> {trap.trap_code.name:26} {reason}")

The output:

divide       -> INTEGER_DIVISION_BY_ZERO   integer divide by zero
unreachable  -> UNREACHABLE                wasm `unreachable` instruction executed
peek         -> MEMORY_OUT_OF_BOUNDS       out of bounds memory access
recurse      -> STACK_OVERFLOW             call stack exhausted
add          -> returned 5

All four failures became ordinary Python exceptions with a machine-readable code: INTEGER_DIVISION_BY_ZERO, UNREACHABLE, MEMORY_OUT_OF_BOUNDS (a read at address 70,000 in a 65,536-byte memory) and STACK_OVERFLOW (a function calling itself until the guest’s call stack ran out). The host survived each one, and the final line shows a normal call still works afterward. That is the property you want from a plugin boundary: the plugin can fail in any way it likes, and the failure stays inside the plugin.

Step 7: Add a wall-clock limit with epochs

Fuel counts work, not seconds. For a time limit, Wasmtime offers epoch interruption. The engine keeps a counter (the epoch). You set a deadline on the store, a number of ticks after which the guest should trap, and you run a timer thread whose only job is to advance the counter when your time is up. Compiled guest code checks the counter at certain points and traps with INTERRUPT once the deadline has passed. The Config documentation notes that “The interruptions are not deterministic”, which is why fuel and epochs suit different jobs.

There is one catch you must understand before relying on it, and the Wasmtime documentation states it plainly: “Epochs (and fuel) do not assist in handling WebAssembly code blocked in a call to the host.” It continues: “Epochs intentionally only affect running WebAssembly code itself and it’s left to the embedder to determine how best to wake up indefinitely blocking code in the host.” In other words, if one of your own helper functions takes a long time, neither guard can interrupt it.

Create step7_wall_clock.py. It runs four cases against a 0.3-second deadline: a spin loop, a guest that makes one 2-second host call, a guest that makes repeated 0.4-second host calls, and the same long call when the host function enforces the deadline itself.

# step7_wall_clock.py
# step7_wall_clock.py
import threading
import time

from wasmtime import Config, Engine, FuncType, Linker, Module, Store, Trap, wat2wasm

SPIN_WAT = '(module (func (export "run") (loop $forever br $forever)))'

ONE_SLOW_CALL_WAT = """
(module
  (import "host" "slow" (func $slow))
  (func (export "run") call $slow))
"""

LOOPING_SLOW_CALL_WAT = """
(module
  (import "host" "slow" (func $slow))
  (func (export "run")
    (loop $again
      call $slow
      br $again)))
"""


def run(wat, deadline_s, slow_seconds=0.0, bounded=False):
    config = Config()
    config.epoch_interruption = True
    engine = Engine(config)
    store = Store(engine)
    store.set_epoch_deadline(1)  # trap once the engine's epoch has advanced one tick
    started = time.perf_counter()

    def slow():
        if bounded:  # the host function enforces the deadline itself
            remaining = started + deadline_s - time.perf_counter()
            time.sleep(max(0.0, min(slow_seconds, remaining)))
        else:
            time.sleep(slow_seconds)

    linker = Linker(engine)
    linker.define_func("host", "slow", FuncType([], []), slow)
    instance = linker.instantiate(store, Module(engine, wat2wasm(wat)))

    timer = threading.Timer(deadline_s, engine.increment_epoch)
    timer.start()
    try:
        instance.exports(store)["run"](store)
        outcome = "returned normally"
    except Trap as trap:
        outcome = trap.trap_code.name
    finally:
        timer.cancel()
    return outcome, time.perf_counter() - started


for label, wat, slow_seconds, bounded in (
    ("spin loop", SPIN_WAT, 0.0, False),
    ("one 2.0 s host call", ONE_SLOW_CALL_WAT, 2.0, False),
    ("loop of 0.4 s host calls", LOOPING_SLOW_CALL_WAT, 0.4, False),
    ("one 2.0 s call, host-bounded", ONE_SLOW_CALL_WAT, 2.0, True),
):
    outcome, elapsed = run(wat, 0.3, slow_seconds, bounded)
    print(f"{label:30} deadline 0.3 s -> {outcome} after {elapsed:.1f} s")

Run it. This one takes a few seconds, because two of the cases genuinely wait:

spin loop                      deadline 0.3 s -> INTERRUPT after 0.3 s
one 2.0 s host call            deadline 0.3 s -> returned normally after 2.0 s
loop of 0.4 s host calls       deadline 0.3 s -> INTERRUPT after 0.4 s
one 2.0 s call, host-bounded   deadline 0.3 s -> returned normally after 0.3 s

The spin loop was interrupted at 0.3 seconds, exactly as designed. The 2-second host call was not: the deadline passed at 0.3 seconds, nothing noticed, and the guest returned normally after 2.0 seconds. In the third case the guest was caught at 0.4 seconds, not 0.3, because the interrupt could only land once the host function returned and WebAssembly code started running again. The fourth case is the fix: the host function looks at the clock and shortens its own wait, so the whole run ended at 0.3 seconds. Any capability that can block, such as a network request, a lock or a sleep, must carry its own timeout.

The same documentation page adds a second limit. Bulk-data-transfer instructions such as memory.copy “only check the epoch once at the start of the operation”, so one very large copy cannot be interrupted partway, and Wasmtime recommends that hosts needing strict time limits “ensure that the store’s allocated heap size (linear memory + GC heap) are bounded with a ResourceLimiter.” The memory cap from Step 5 is how you do that from Python.

What do the guards cost?

Which one should you use? The documentation says that “In general, epoch-based interruption results in faster execution” and that “Fuel, in contrast, should be used when deterministic yielding or trapping is needed”. I wanted a number for my own machine, so step7b_guard_cost.py runs a tight counting loop of 100 million iterations with no guard, with epochs and with fuel, and keeps the best of three runs for each.

# step7b_guard_cost.py
# step7b_guard_cost.py
import time

from wasmtime import Config, Engine, Instance, Module, Store, wat2wasm

COUNT_WAT = """
(module
  (func (export "count") (param $n i32) (result i32)
    (local $i i32)
    (loop $again
      local.get $i
      i32.const 1
      i32.add
      local.set $i
      local.get $i
      local.get $n
      i32.lt_u
      br_if $again)
    local.get $i))
"""
N = 100_000_000


def seconds(guard):
    config = Config()
    if guard == "fuel":
        config.consume_fuel = True
    elif guard == "epoch":
        config.epoch_interruption = True
    engine = Engine(config)
    store = Store(engine)
    if guard == "fuel":
        store.set_fuel(10**15)
    elif guard == "epoch":
        store.set_epoch_deadline(10**9)
    instance = Instance(store, Module(engine, wat2wasm(COUNT_WAT)), [])
    count = instance.exports(store)["count"]
    started = time.perf_counter()
    count(store, N)
    return time.perf_counter() - started


best = {guard: min(seconds(guard) for _ in range(3)) for guard in ("none", "epoch", "fuel")}
for guard, took in best.items():
    print(f"{guard:6} {took:6.3f} s  {took / best['none']:.2f}x")

The output:

none    0.021 s  1.00x
epoch   0.031 s  1.51x
fuel    0.030 s  1.45x

On this machine the two guards cost about the same: across ten runs of the script, epochs measured between 1.51 and 1.56 times the unguarded time and fuel between 1.41 and 1.45 times. That does not contradict the documentation, which says the difference is sometimes significant and speaks of “some measurements”. A loop this tight puts the guard’s bookkeeping in the largest possible share of the work, so real plugins should see smaller overhead. The practical advice is to choose by what you need, not by speed: fuel for a reproducible budget, epochs for a wall-clock cap, and measure your own workload before you decide. The final runner turns on both.

Step 8: Understand that host functions are the new attack surface

So far the guest has been boxed in. But a plugin that can only compute is not very useful, so sooner or later you give it a capability: read a file, fetch a page, write a log line. Each capability is a Python function that runs in your process, with your privileges, on behalf of untrusted code. The sandbox protects you from the guest. Nothing protects you from your own host function.

To see this, you need a plugin that calls a host function with a pointer. Remember that a pointer is just a number: an offset into the guest’s memory. The helper below, read_text(path_ptr, path_len, out_ptr, out_cap), asks the host to read path_len bytes at path_ptr, treat them as a relative file path, and copy up to out_cap bytes of that file to out_ptr. Create reader_plugin.py, which holds a small plugin with three exports (read_notes, read_secret and read_wild) and a harness that calls one of them with the host function of your choice:

# reader_plugin.py
# reader_plugin.py
from wasmtime import Engine, FuncType, Linker, Module, Store, ValType, wat2wasm

I32 = ValType.i32()

READER_WAT = """
(module
  (import "host" "read_text" (func $read_text (param i32 i32 i32 i32) (result i32)))
  (memory (export "memory") 1)
  (data (i32.const 0) "notes.txt")
  (data (i32.const 32) "../secret.txt")
  (func (export "read_notes") (result i32)
    i32.const 0
    i32.const 9
    i32.const 128
    i32.const 256
    call $read_text)
  (func (export "read_secret") (result i32)
    i32.const 32
    i32.const 13
    i32.const 128
    i32.const 256
    call $read_text)
  (func (export "read_wild") (param $ptr i32) (result i32)
    local.get $ptr
    i32.const 13
    i32.const 128
    i32.const 256
    call $read_text))
"""


def run_export(read_text_impl, export, *args):
    """Call one export of the reader plugin; return (count, bytes the plugin received)."""
    engine = Engine()
    store = Store(engine)
    linker = Linker(engine)
    linker.define_func(
        "host", "read_text", FuncType([I32, I32, I32, I32], [I32]),
        read_text_impl, access_caller=True,
    )
    instance = linker.instantiate(store, Module(engine, wat2wasm(READER_WAT)))
    count = instance.exports(store)[export](store, *args)
    memory = instance.exports(store)["memory"]
    return count, bytes(memory.read(store, 128, 128 + max(count, 0)))

The host function gets at the guest’s memory through a Caller. Passing access_caller=True when defining the function makes wasmtime-py give it a Caller as its first argument, and caller.get("memory") returns the memory the plugin exported under that name. Now create step8_host_function.py with a first-draft implementation, the kind most of us would write on the first try:

# step8_host_function.py
# step8_host_function.py
from pathlib import Path

from wasmtime import Engine, Instance, Module, Store, wat2wasm

from reader_plugin import READER_WAT, run_export

DATA_DIR = Path("plugin_data")
DATA_DIR.mkdir(exist_ok=True)
with open(DATA_DIR / "notes.txt", "w", newline="\n") as handle:
    handle.write("public notes\n")


def naive_read_text(caller, path_ptr, path_len, out_ptr, out_cap):
    memory = caller.get("memory")
    path = memory.read(caller, path_ptr, path_ptr + path_len).decode()
    data = (DATA_DIR / path).read_bytes()[:out_cap]
    memory.write(caller, data, out_ptr)
    return len(data)


print("read_notes ->", run_export(naive_read_text, "read_notes"))
print("read_secret ->", run_export(naive_read_text, "read_secret"))
try:
    run_export(naive_read_text, "read_wild", 70000)
except Exception as exc:
    print("read_wild(70000) raised in the host:", type(exc).__name__)

# Two facts about the Memory API that host code must not forget.
engine = Engine()
store = Store(engine)
instance = Instance(store, Module(engine, wat2wasm('(module (memory (export "memory") 1))')), [])
memory = instance.exports(store)["memory"]
print("read past the end returns", len(memory.read(store, 70000, 70013)), "bytes, no error")
print("read across the end returns", len(memory.read(store, 65530, 65540)), "bytes, silently shortened")
try:
    memory.write(store, b"x" * 10, 65530)
except IndexError as exc:
    print("write across the end ->", type(exc).__name__ + ":", exc)

Run it:

read_notes -> (13, b'public notes\n')
read_secret -> (20, b'db_password=hunter2\n')
read_wild(70000) raised in the host: PermissionError
read past the end returns 0 bytes, no error
read across the end returns 6 bytes, silently shortened
write across the end -> IndexError: index out of range

Three things happened. The honest request for notes.txt worked: 13 bytes came back. The request for ../secret.txt also worked: the plugin received db_password=hunter2. The sandbox held, because the guest never touched the file system. The host did, because the helper joined an untrusted path onto a folder and opened it. This is the same flaw class as the path traversal tutorial on this site, now with a WebAssembly guest asking.

The third request, read_wild(70000), points past the end of the plugin’s 65,536-byte memory, and it made the host raise. It raised PermissionError here because opening a directory is a permission error on Windows (on Linux and macOS, expect IsADirectoryError instead). The reason is in the last three lines. The wasmtime-py documentation says of Memory.read: “The indexing behavior of this method is similar to list[start:stop] where negative starts can be used to read from the end, for example.” Like a Python slice, reading past the end does not raise: it returned 0 bytes, and a read across the end returned 6 bytes instead of 10. Writing is stricter and raises IndexError. So a pointer to nowhere became an empty path, the empty path became the data folder itself, and opening that failed. An exception raised inside a host function does not become a trap. It travels out through your call as the original Python exception, so a plugin that can make your helper raise can crash code you thought was protected.

Step 9: Harden the capability

The fix is to treat every argument from the guest as hostile input. Create capabilities.py. It collects everything the host is willing to do in one class, so the allowlist from Step 3 and the helper functions live together, and it gives each capability its own limit.

# capabilities.py
# capabilities.py
from pathlib import Path

from wasmtime import FuncType, Linker, ValType

I32 = ValType.i32()
ERR_DENIED, ERR_BAD_POINTER, ERR_NOT_FOUND = -1, -2, -3
MAX_PATH_BYTES = 255


class Capabilities:
    """Everything the host is willing to do for a plugin, each with its own limit."""

    def __init__(self, data_dir=None, max_log_entries=100):
        self.base = Path(data_dir).resolve() if data_dir else None
        self.max_log_entries = max_log_entries
        self.logs = []

    def allowed_imports(self):
        allowed = {("host", "log")}
        if self.base is not None:
            allowed.add(("host", "read_text"))
        return allowed

    def log(self, value):
        if len(self.logs) < self.max_log_entries:  # a plugin may call this a million times
            self.logs.append(value)

    def read_text(self, caller, path_ptr, path_len, out_ptr, out_cap):
        memory = caller.get("memory")
        if memory is None:  # the plugin did not export its memory
            return ERR_BAD_POINTER
        size = memory.data_len(caller)
        if min(path_ptr, path_len, out_ptr, out_cap) < 0 or path_len > MAX_PATH_BYTES:
            return ERR_BAD_POINTER
        if path_ptr + path_len > size or out_ptr + out_cap > size:
            return ERR_BAD_POINTER  # memory.read() would silently shorten these
        try:
            relative = bytes(memory.read(caller, path_ptr, path_ptr + path_len)).decode("utf-8")
        except UnicodeDecodeError:
            return ERR_BAD_POINTER
        target = (self.base / relative).resolve()
        if not target.is_relative_to(self.base):
            return ERR_DENIED
        try:
            with open(target, "rb") as handle:
                data = handle.read(out_cap)  # never read more than the plugin has room for
        except (OSError, ValueError):
            return ERR_NOT_FOUND
        memory.write(caller, data, out_ptr)
        return len(data)

    def link(self, engine):
        linker = Linker(engine)
        linker.define_func("host", "log", FuncType([I32], []), self.log)
        if self.base is not None:
            linker.define_func(
                "host", "read_text", FuncType([I32, I32, I32, I32], [I32]),
                self.read_text, access_caller=True,
            )
        return linker

Each guard in read_text exists because of something you have already seen:

  1. Check that the guest exported a memory. caller.get("memory") returns None if the plugin did not export one. I found this the hard way: the first time I ran the final table in Step 10, a plugin without an exported memory made this function raise AttributeError: 'NoneType' object has no attribute 'data_len'. Now it returns an error code instead.
  2. Reject negative and oversized numbers. WebAssembly integers reach Python as signed values, which the next script demonstrates. Combined with the slice behavior of Memory.read, a negative pointer would not be refused by wasmtime; it would quietly read from the end of the guest’s memory.
  3. Check the range yourself. Because Memory.read shortens instead of raising, compare path_ptr + path_len and out_ptr + out_cap with memory.data_len(caller) first.
  4. Resolve the path, then check it is inside the folder. (self.base / relative).resolve() collapses .. and follows symlinks, and is_relative_to(self.base) then confirms the result is still under the allowed folder. The path traversal tutorial compares five validators and shows why this pair is the one that holds.
  5. Read no more than the guest has room for. handle.read(out_cap) stops at the buffer size. Without that, read_bytes() would load a multi-gigabyte file into the host just to discard most of it.
  6. Return error codes, not exceptions. Like system calls, the function returns small negative numbers (-1 denied, -2 bad pointer, -3 not found). A plugin that gets an error is behaving normally. A host function that raises is a bug.
  7. Give log a budget. A plugin may call it a million times, so it stores at most max_log_entries values and ignores the rest.

Now run the same plugin against the hardened function. Create step9_hardened_host.py:

# step9_hardened_host.py
# step9_hardened_host.py
from pathlib import Path

from capabilities import Capabilities
from reader_plugin import run_export

caps = Capabilities(data_dir=Path("plugin_data"))
print("read_notes ->", run_export(caps.read_text, "read_notes"))
print("read_secret ->", run_export(caps.read_text, "read_secret"))
for pointer in (70000, -1, 4_294_967_295):
    print(f"read_wild({pointer}) ->", run_export(caps.read_text, "read_wild", pointer))

The output:

read_notes -> (13, b'public notes\n')
read_secret -> (-1, b'')
read_wild(70000) -> (-2, b'')
read_wild(-1) -> (-2, b'')
read_wild(4294967295) -> (-2, b'')

The honest request still works (13 bytes). The traversal is refused with -1, and all three wild pointers come back as -2 without raising anything. The last pointer, 4,294,967,295, is worth a closer look. To confirm that integers really arrive signed, create step9b_signed_args.py, a guest that forwards whatever number it is given to a host function that records it:

# step9b_signed_args.py
# step9b_signed_args.py
from wasmtime import Engine, FuncType, Linker, Module, Store, ValType, wat2wasm

WAT = """
(module
  (import "host" "see" (func $see (param i32)))
  (func (export "go") (param i32)
    local.get 0
    call $see))
"""

seen = []
engine = Engine()
store = Store(engine)
linker = Linker(engine)
linker.define_func("host", "see", FuncType([ValType.i32()], []), lambda value: seen.append(value))
instance = linker.instantiate(store, Module(engine, wat2wasm(WAT)))
go = instance.exports(store)["go"]
for value in (70000, 4_294_967_295, 2_147_483_648):
    go(store, value)
print("the guest sent 70000, 4294967295 and 2147483648; the host saw", seen)

The output:

the guest sent 70000, 4294967295 and 2147483648; the host saw [70000, -1, -2147483648]

The guest sent 4,294,967,295 and the host saw -1; it sent 2,147,483,648 and the host saw -2147483648. The check for negative numbers is therefore not optional, and it is the check that catches the “huge” pointers, too.

Step 10: Put it together in run_plugin()

You now have every piece: an import allowlist, fuel, an epoch deadline, memory limits, trap handling and a hardened capability class. Create sandbox.py, which combines them into one function.

# sandbox.py
# sandbox.py
import threading
import time
from dataclasses import dataclass, field

from wasmtime import Config, Engine, Module, Store, Trap, TrapCode, WasmtimeError

from capabilities import Capabilities


@dataclass(frozen=True)
class Limits:
    fuel: int = 10_000_000
    max_memory_bytes: int = 16 * 1024 * 1024
    timeout_s: float = 2.0
    max_log_entries: int = 100


@dataclass
class Result:
    status: str  # ok | rejected | out_of_fuel | timeout | trap | host_error
    value: object = None
    detail: str = ""
    fuel_used: int = 0
    seconds: float = 0.0
    logs: list = field(default_factory=list)


def one_line(exc, width=90):
    return " ".join(str(exc).split()).split(" - ")[0][:width]


def run_plugin(wasm, entry, args=(), *, limits=Limits(), data_dir=None):
    started = time.perf_counter()
    caps = Capabilities(data_dir, limits.max_log_entries)

    def result(status, value=None, detail="", fuel_used=0):
        return Result(status, value, detail, fuel_used, time.perf_counter() - started, caps.logs)

    config = Config()
    config.consume_fuel = True
    config.epoch_interruption = True
    engine = Engine(config)  # one engine per run: the epoch counter lives on the engine

    try:
        Module.validate(engine, bytes(wasm))
        module = Module(engine, bytes(wasm))
    except WasmtimeError as exc:
        return result("rejected", detail="invalid module: " + one_line(exc))

    refused = sorted({(item.module, item.name) for item in module.imports} - caps.allowed_imports())
    if refused:
        names = ", ".join(f"{mod}.{name}" for mod, name in refused)
        return result("rejected", detail="imports not allowed: " + names)
    if entry not in {export.name for export in module.exports}:
        return result("rejected", detail=f"no export named {entry!r}")

    store = Store(engine)
    store.set_limits(memory_size=limits.max_memory_bytes)
    store.set_fuel(limits.fuel)
    store.set_epoch_deadline(1)
    try:
        instance = caps.link(engine).instantiate(store, module)
    except WasmtimeError as exc:
        return result("rejected", detail="could not instantiate: " + one_line(exc))

    timer = threading.Timer(limits.timeout_s, engine.increment_epoch)
    timer.daemon = True
    timer.start()
    value, detail = None, ""
    try:
        value = instance.exports(store)[entry](store, *args)
        status = "ok"
    except Trap as trap:
        code = trap.trap_code
        status = {TrapCode.OUT_OF_FUEL: "out_of_fuel", TrapCode.INTERRUPT: "timeout"}.get(code, "trap")
        detail = code.name if code else one_line(trap)
    except Exception as exc:  # raised by one of our own host functions
        status, detail = "host_error", f"{type(exc).__name__}: {exc}"
    finally:
        timer.cancel()
    return result(status, value, detail, limits.fuel - store.get_fuel())

Read run_plugin() from top to bottom, because the order is the design:

  1. Validate and compile the bytes. Module.validate and Module(...) run first, and any WasmtimeError becomes a rejected result. Garbage input is not an exception your caller has to handle.
  2. Check imports against the allowlist and check the entry point exists. This is Step 3, run before anything executes. The detail string names the refused imports.
  3. Create a fresh store with its limits. Memory cap (Step 5), fuel (Step 4) and an epoch deadline of one tick (Step 7). A new store per run means a fresh fuel pool and no state carried over from the last plugin.
  4. Link and instantiate. A module whose declared memory exceeds the cap fails here and is reported as rejected.
  5. Start a timer that advances the epoch, call the entry point, and map every outcome. OUT_OF_FUEL becomes out_of_fuel, INTERRUPT becomes timeout, any other trap becomes trap with its code, and an exception raised by one of our host functions becomes host_error. The timer is always cancelled, and fuel used is the budget minus what is left.

Each run gets its own Engine, because Wasmtime’s epoch counter lives on the engine, so a timer firing for one run could also cut short another run that shares the engine. For small plugins a new engine per run is cheap; a busy service would keep one engine per worker thread and cache compiled modules.

Now try it on a gallery of plugins. Create step10_run_plugin.py. Every plugin exports apply(cents, percent), the shape of a discount rule. One is honest; the rest each break a different rule.

# step10_run_plugin.py
# step10_run_plugin.py
from pathlib import Path

from wasmtime import wat2wasm

from sandbox import Limits, run_plugin


def wasm(wat):
    return bytes(wat2wasm(wat))


HONEST = wasm("""
(module
  (func (export "apply") (param $cents i32) (param $percent i32) (result i32)
    local.get $cents
    i32.const 100
    local.get $percent
    i32.sub
    i32.mul
    i32.const 100
    i32.div_u))
""")

FOREVER = wasm("""
(module
  (func (export "apply") (param i32 i32) (result i32)
    (loop $forever br $forever)
    i32.const 0))
""")

MEMORY_HOG = wasm("""
(module
  (memory 2000)
  (func (export "apply") (param i32 i32) (result i32) local.get 0))
""")

STACK_BOMB = wasm("""
(module
  (func $deeper (export "apply") (param i32 i32) (result i32)
    local.get 0
    local.get 1
    call $deeper))
""")

DIVIDES_BY_ZERO = wasm("""
(module
  (func (export "apply") (param $cents i32) (param $percent i32) (result i32)
    local.get $cents
    local.get $percent
    local.get $percent
    i32.sub
    i32.div_u))
""")

WANTS_SYSTEM = wasm("""
(module
  (import "env" "system" (func $system (param i32) (result i32)))
  (func (export "apply") (param i32 i32) (result i32)
    i32.const 0
    call $system))
""")

READS_SECRET = wasm("""
(module
  (import "host" "read_text" (func $read_text (param i32 i32 i32 i32) (result i32)))
  (memory (export "memory") 1)
  (data (i32.const 0) "../secret.txt")
  (func (export "apply") (param i32 i32) (result i32)
    i32.const 0
    i32.const 13
    i32.const 64
    i32.const 256
    call $read_text))
""")

HIDES_MEMORY = wasm("""
(module
  (import "host" "read_text" (func $read_text (param i32 i32 i32 i32) (result i32)))
  (memory 1)
  (data (i32.const 0) "notes.txt")
  (func (export "apply") (param i32 i32) (result i32)
    i32.const 0
    i32.const 9
    i32.const 64
    i32.const 256
    call $read_text))
""")

CASES = {
    "honest discount": (HONEST, Limits()),
    "infinite loop": (FOREVER, Limits()),
    "slow burn": (FOREVER, Limits(fuel=10**15, timeout_s=0.3)),
    "memory hog": (MEMORY_HOG, Limits()),
    "stack bomb": (STACK_BOMB, Limits()),
    "divides by zero": (DIVIDES_BY_ZERO, Limits()),
    "wants system()": (WANTS_SYSTEM, Limits()),
    "reads ../secret.txt": (READS_SECRET, Limits()),
    "hides its memory": (HIDES_MEMORY, Limits()),
    "not even Wasm": (b"print('hello')", Limits()),
}

for label, (payload, limits) in CASES.items():
    result = run_plugin(payload, "apply", (1000, 15), limits=limits, data_dir=Path("plugin_data"))
    fuel = "n/a" if result.status == "timeout" else f"{result.fuel_used:,}"
    print(f"{label:20} {result.status:12} value={result.value!s:5} fuel={fuel:>10}  {result.detail}")

Run it:

honest discount      ok           value=850   fuel=         8  
infinite loop        out_of_fuel  value=None  fuel=10,000,000  OUT_OF_FUEL
slow burn            timeout      value=None  fuel=       n/a  INTERRUPT
memory hog           rejected     value=None  fuel=         0  could not instantiate: memory minimum size of 2000 pages exceeds memory limits
stack bomb           trap         value=None  fuel=    32,748  STACK_OVERFLOW
divides by zero      trap         value=None  fuel=         0  INTEGER_DIVISION_BY_ZERO
wants system()       rejected     value=None  fuel=         0  imports not allowed: env.system
reads ../secret.txt  ok           value=-1    fuel=     4,103  
hides its memory     ok           value=-2    fuel=     4,103  
not even Wasm        rejected     value=None  fuel=         0  invalid module: magic header not detected: bad magic number

Each row is a different lesson:

  • honest discount returned 850 (1,000 cents with 15 percent off) using 8 fuel.
  • infinite loop ended as out_of_fuel after burning the whole 10,000,000 budget, in a fraction of a second.
  • slow burn is the same loop but with a huge fuel budget and a 0.3-second timeout, so the epoch deadline ended it as timeout. Its fuel use depends on timing, so the table shows n/a.
  • memory hog declared 2,000 pages up front and never started.
  • stack bomb recursed until STACK_OVERFLOW, after about 32,700 units of fuel, long before the budget ran out.
  • divides by zero became a trap with INTEGER_DIVISION_BY_ZERO.
  • wants system() was rejected before it ran, by name.
  • reads ../secret.txt ran, but the capability returned -1: the plugin was told no.
  • hides its memory got -2 from the capability instead of crashing it.
  • not even Wasm was rejected as an invalid module.

Two small notes on the table. The two rows that read files show 4,103 units of fuel against 8 for pure arithmetic; I did not chase down where the extra accounting comes from, so compare fuel within one version rather than treating it as a stable unit. And no row ended as host_error: that status exists for the day a host function of yours has a bug, and the plugin’s only job at that point is to be contained.

Step 11: Lock the behavior in with tests

A sandbox that you do not test is a sandbox you hope works. Create test_sandbox.py. Each test pins one promise made above: the honest plugin runs; fuel use is exactly 8,002 for the counting loop on every run; infinite loops and slow loops are stopped; undeclared imports, garbage bytes and missing entry points are rejected; memory growth is capped; traps are reported, not raised; the log capability is capped; and the file capability refuses traversal, wild pointers and a missing memory export.

# test_sandbox.py
# test_sandbox.py
import pytest
from wasmtime import wat2wasm

from reader_plugin import READER_WAT
from sandbox import Limits, run_plugin


def wasm(wat):
    return bytes(wat2wasm(wat))


APPLY = wasm("""
(module
  (func (export "apply") (param $cents i32) (param $percent i32) (result i32)
    local.get $cents
    i32.const 100
    local.get $percent
    i32.sub
    i32.mul
    i32.const 100
    i32.div_u))
""")

SPIN = wasm('(module (func (export "run") (loop $forever br $forever)))')

COUNT = wasm("""
(module
  (func (export "count") (param $n i32) (result i32)
    (local $i i32)
    (loop $again
      local.get $i
      i32.const 1
      i32.add
      local.set $i
      local.get $i
      local.get $n
      i32.lt_u
      br_if $again)
    local.get $i))
""")

GROW = wasm("""
(module
  (memory 1)
  (func (export "grow") (param $pages i32) (result i32)
    local.get $pages
    memory.grow))
""")

BIG_MEMORY = wasm('(module (memory 2000) (func (export "run")))')

RECURSE = wasm('(module (func $again (export "run") (call $again)))')

DIVIDE = wasm("""
(module
  (func (export "run") (param i32 i32) (result i32)
    local.get 0
    local.get 1
    i32.div_s))
""")

SNEAKY = wasm("""
(module
  (import "env" "system" (func $system (param i32) (result i32)))
  (func (export "run") (result i32)
    i32.const 0
    call $system))
""")

SPAM = wasm("""
(module
  (import "host" "log" (func $log (param i32)))
  (func (export "spam") (param $n i32)
    (local $i i32)
    (loop $again
      local.get $i
      call $log
      local.get $i
      i32.const 1
      i32.add
      local.tee $i
      local.get $n
      i32.lt_u
      br_if $again)))
""")


def test_honest_plugin_runs():
    result = run_plugin(APPLY, "apply", (1000, 15))
    assert (result.status, result.value) == ("ok", 850)


def test_fuel_use_is_deterministic():
    runs = [run_plugin(COUNT, "count", (1000,)).fuel_used for _ in range(3)]
    assert runs == [8002, 8002, 8002]


def test_infinite_loop_runs_out_of_fuel():
    result = run_plugin(SPIN, "run", limits=Limits(fuel=100_000))
    assert result.status == "out_of_fuel"
    assert result.fuel_used == 100_000


def test_wall_clock_timeout():
    result = run_plugin(SPIN, "run", limits=Limits(fuel=10**15, timeout_s=0.2))
    assert result.status == "timeout"
    assert result.seconds < 1.5


def test_undeclared_import_is_rejected_before_it_runs():
    result = run_plugin(SNEAKY, "run")
    assert result.status == "rejected"
    assert "env.system" in result.detail


def test_garbage_bytes_are_rejected():
    assert run_plugin(b"print('hello')", "run").status == "rejected"


def test_missing_entry_point_is_rejected():
    result = run_plugin(APPLY, "nope")
    assert result.status == "rejected"
    assert "no export" in result.detail


def test_declared_memory_over_the_limit_is_rejected():
    result = run_plugin(BIG_MEMORY, "run", limits=Limits(max_memory_bytes=16 * 65536))
    assert result.status == "rejected"
    assert "exceeds memory limits" in result.detail


@pytest.mark.parametrize("pages, expected", [(2, 1), (10, -1)])
def test_memory_growth_is_capped(pages, expected):
    result = run_plugin(GROW, "grow", (pages,), limits=Limits(max_memory_bytes=4 * 65536))
    assert (result.status, result.value) == ("ok", expected)


@pytest.mark.parametrize("wasm_bytes, args, code", [
    (RECURSE, (), "STACK_OVERFLOW"),
    (DIVIDE, (1, 0), "INTEGER_DIVISION_BY_ZERO"),
])
def test_traps_are_reported_not_raised(wasm_bytes, args, code):
    result = run_plugin(wasm_bytes, "run", args)
    assert (result.status, result.detail) == ("trap", code)


def test_log_capability_is_capped():
    result = run_plugin(SPAM, "spam", (1000,), limits=Limits(max_log_entries=100))
    assert result.status == "ok"
    assert len(result.logs) == 100


@pytest.fixture
def data_dir(tmp_path):
    folder = tmp_path / "data"
    folder.mkdir()
    (folder / "notes.txt").write_bytes(b"public notes\n")
    (tmp_path / "secret.txt").write_bytes(b"db_password=hunter2\n")
    return folder


def test_capability_reads_inside_the_data_dir(data_dir):
    result = run_plugin(wasm(READER_WAT), "read_notes", data_dir=data_dir)
    assert (result.status, result.value) == ("ok", 13)


def test_capability_refuses_paths_that_climb_out(data_dir):
    result = run_plugin(wasm(READER_WAT), "read_secret", data_dir=data_dir)
    assert (result.status, result.value) == ("ok", -1)


@pytest.mark.parametrize("pointer", [70000, -1, 4_294_967_295])
def test_capability_rejects_wild_pointers(data_dir, pointer):
    result = run_plugin(wasm(READER_WAT), "read_wild", (pointer,), data_dir=data_dir)
    assert (result.status, result.value) == ("ok", -2)


def test_capability_needs_an_exported_memory(data_dir):
    hidden = wasm("""
(module
  (import "host" "read_text" (func $read_text (param i32 i32 i32 i32) (result i32)))
  (memory 1)
  (data (i32.const 0) "notes.txt")
  (func (export "run") (result i32)
    i32.const 0
    i32.const 9
    i32.const 64
    i32.const 256
    call $read_text))
""")
    result = run_plugin(hidden, "run", data_dir=data_dir)
    assert (result.status, result.value) == ("ok", -2)


def test_capability_is_not_offered_without_a_data_dir():
    result = run_plugin(wasm(READER_WAT), "read_notes")
    assert result.status == "rejected"
    assert "host.read_text" in result.detail

Run it with python -m pytest -q:

....................                                                     [100%]
20 passed in 0.35s

All 20 tests pass in well under a second. The test for the wall-clock timeout allows up to 1.5 seconds, so it will not flake on a slow machine, and the fuel-determinism test asserts the exact number 8,002, which only holds if the same plugin really does cost the same every time.

Confirm it works end to end

To check the whole thing, run the step scripts in order from the project folder and compare with the outputs above: step1_exec_problem.py through step10_run_plugin.py, including step7b_guard_cost.py and step9b_signed_args.py. Expect the same results with different timings, and the one platform difference noted in Step 8. Then run python -m pytest -q and look for 20 passed. Finally, try to break it yourself, which is the best way to learn where the limits are. Run the honest plugin with Limits(fuel=5) and confirm it ends as out_of_fuel, since it needs 8. Write a plugin that calls log a million times and confirm that result.logs never holds more than 100 entries, whatever the final status. Edit the path in the reader plugin to ../../secret.txt or to an absolute path (and update its length) and confirm the answer is always -1.

Common mistakes

Forgetting to put fuel in the store

Turning on consume_fuel without calling set_fuel gives a store with zero fuel, so even the first instruction traps. If a trivial plugin reports OUT_OF_FUEL, check this first.

Reusing one store for many requests

Fuel, memory growth and any state inside the instance all carry over between calls on the same store. Step 4 showed the budget draining across calls. A fresh store per run is the simplest cure.

Mixing objects from different stores

While writing this lab I once passed a Memory that belonged to one store into a call that used another. wasmtime-py did not raise a Python exception: the process died with a Rust panic that read “object used with the wrong store”. Keep one store per run and do not cache exports between runs.

Trusting Memory.read and signed arguments

Memory.read slices instead of raising, and integer arguments can be negative. Validate both before you use a pointer, as in Step 9.

Forgetting that host functions block

Fuel and epochs only measure WebAssembly code. A slow capability needs its own timeout (Step 7).

Accepting text from an untrusted source

The runner accepts binary .wasm bytes and validates them first. Pass untrusted input as bytes, not as WAT text, and keep wat2wasm for your own test fixtures.

What a WebAssembly sandbox does not do

Be clear about the limits. A sandbox is a strong boundary, not a magic one. Even the Wasmtime security page, which is confident about the design, acknowledges that “While WebAssembly is designed to be sandboxed bugs or issues inevitably arise”, and it describes extra mitigations that exist to limit the damage if a bug is found. A runtime is software, and software has bugs. For code from strangers, add layers: run the runner in a separate process under a low-privilege account, keep a kill timer outside the process, and apply operating-system limits. The same caution appears elsewhere on this site: my article on Cloudflare’s EmDash plugin registry quotes the workerd README’s warning that it is not a hardened sandbox on its own and should run inside something like a virtual machine when running possibly malicious code.

The second limit is the one this tutorial spent Steps 8 and 9 on: the sandbox only protects what the host does not hand over. Every capability you add is code you must secure like any other input handler. A plugin that can fetch URLs inherits the risks in the SSRF tutorial, and one that can use regular expressions you supply inherits those in the ReDoS tutorial. Third, the numbers here (fuel per instruction, which traps fire where) are properties of the Wasmtime version I tested, so pin your version and rerun the tests when you upgrade.

Where to go next

You now have a working pattern: default-deny imports, fuel, epochs, memory limits, hardened capabilities and tests. Natural next steps:

  • Compile a real plugin. Take a small function written in a language that targets WebAssembly, compile it to a .wasm file, and run it through run_plugin(). Anything that expects a system interface (WASI) will appear in the import list and be rejected until you decide to provide it.
  • Read the Wasmtime book. Its table of contents includes a chapter titled “An Application with Plugins” that shows the standardized component-model approach to the same problem, and a chapter on interrupting execution.
  • Study agent sandboxes. This site has covered the same boundary-first idea from other angles, in NVIDIA’s OpenShell launch and Docker’s Sandbox Kit specification.

Tags:

AI Coding AgentsApplication SecurityPythonSandboxingWebAssembly

Share

No Comment! Be the first one.

Leave a Reply

Your email address will not be published. Required fields are marked *

Related Posts