matchlock 0.2.16

Python SDK for Matchlock sandbox

Matchlock Python SDK

A Python client for Matchlock — a lightweight micro-VM sandbox for running AI-generated code securely with network interception and secret protection.

Requirements

• Python 3.10+
• The matchlock CLI binary installed and available on $PATH (or specify its path via Config)

Installation

pip install matchlock

Or install from source:

pip install -e sdk/python

Quick Start

from matchlock import Client, Sandbox

sandbox = Sandbox("python:3.12-alpine")

with Client() as client:
    client.launch(sandbox)

    result = client.exec("echo hello from the sandbox")
    print(result.stdout)  # "hello from the sandbox\n"

Usage

Builder API

The Sandbox class provides a fluent builder for configuring sandboxes:

import os
from matchlock import Client, Sandbox

sandbox = (
    Sandbox("python:3.12-alpine")
    .with_privileged()
    .with_cpus(2)
    .with_memory(512)
    .with_disk_size(2048)
    .with_timeout(300)
    .with_workspace("/home/user/code")
    .allow_host("api.openai.com", "pypi.org")
    .with_port_forward(18080, 8080)
    .with_port_forward_addresses("127.0.0.1")
    .with_network_mtu(1200)
    .block_private_ips()
    .add_secret("API_KEY", os.environ["API_KEY"], "api.openai.com")
)

with Client() as client:
    vm_id = client.launch(sandbox)
    result = client.exec("python3 -c 'print(1+1)'")
    print(result.exit_code)  # 0
    print(result.stdout)     # "2\n"

Streaming Output

Use exec_stream for real-time stdout/stderr streaming:

import sys
from matchlock import Client, Sandbox

with Client() as client:
    client.launch(Sandbox("alpine:latest"))

    result = client.exec_stream(
        "for i in 1 2 3; do echo $i; sleep 1; done",
        stdout=sys.stdout,
        stderr=sys.stderr,
    )
    print(f"Exit code: {result.exit_code}")
    print(f"Duration: {result.duration_ms}ms")

File Operations

Read, write, and list files inside the sandbox:

from matchlock import Client, Sandbox

with Client() as client:
    client.launch(Sandbox("alpine:latest"))

    # Write a file
    client.write_file("/workspace/hello.txt", "Hello, world!")
    client.write_file("/workspace/script.sh", "#!/bin/sh\necho hi", mode=0o755)

    # Read a file
    content = client.read_file("/workspace/hello.txt")
    print(content.decode())  # "Hello, world!"

    # List files
    files = client.list_files("/workspace")
    for f in files:
        print(f"{f.name} ({f.size} bytes, dir={f.is_dir})")

Network Policy & Secrets

Control network access and inject secrets securely:

import os
from matchlock import Client, Sandbox

sandbox = (
    Sandbox("python:3.12-alpine")
    # Only allow these hosts
    .allow_host("api.anthropic.com", "pypi.org", "files.pythonhosted.org")
    # Optional MTU override
    .with_network_mtu(1200)
    # Explicitly block or allow private IPs (10.x, 172.16.x, 192.168.x)
    .with_block_private_ips(True)
    # Inject secret — the MITM proxy replaces the placeholder with the real
    # value only when requests go to the specified host
    .add_secret("ANTHROPIC_API_KEY", os.environ["ANTHROPIC_API_KEY"], "api.anthropic.com")
)

with Client() as client:
    client.launch(sandbox)
    result = client.exec("python3 call_api.py")

    # Optional: forward local 18080 -> guest 8080 after VM creation
    bindings = client.port_forward("18080:8080")
    print(bindings[0].address, bindings[0].local_port, bindings[0].remote_port)

Private-IP behavior in the Python SDK:

• Default (unset): private IPs are blocked whenever a network config is sent.
• Explicit block: .block_private_ips() or .with_block_private_ips(True).
• Explicit allow: .allow_private_ips() or .with_block_private_ips(False).
• Reset to default behavior: .unset_block_private_ips().

If you use CreateOptions(...) directly instead of the builder, set both:

• block_private_ips_set=True
• block_private_ips=<True|False>

For fully offline sandboxes (no guest NIC / no egress), use:

• Builder: .with_no_network()
• Direct options: CreateOptions(no_network=True)

Network Interception Rules

Use typed network hook rules for host-side request/response mutation:

from matchlock import (
    NetworkBodyTransform,
    NetworkHookRule,
    NetworkInterceptionConfig,
    Sandbox,
)

sandbox = Sandbox("alpine:latest").with_network_interception(
    NetworkInterceptionConfig(
        rules=[
            NetworkHookRule(
                phase="after",
                hosts=["api.example.com"],
                action="mutate",
                set_response_headers={"X-Intercepted": "true"},
                body_replacements=[NetworkBodyTransform(find="foo", replace="bar")],
            )
        ]
    )
)

Callback-based interception:

from matchlock import (
    NetworkHookRequest,
    NetworkHookResult,
    NetworkHookResponseMutation,
    NetworkHookRule,
    NetworkInterceptionConfig,
    Sandbox,
)

def after_hook(req: NetworkHookRequest) -> NetworkHookResult | None:
    if req.status_code != 200:
        return None
    return NetworkHookResult(
        action="mutate",
        response=NetworkHookResponseMutation(
            headers={"X-Intercepted": ["callback"]},
            set_body=b'{"msg":"from-callback"}',
        ),
    )

sandbox = Sandbox("alpine:latest").with_network_interception(
    NetworkInterceptionConfig(
        rules=[
            NetworkHookRule(
                phase="after",
                hosts=["api.example.com"],
                hook=after_hook,
                timeout_ms=1500,
            )
        ]
    )
)

Filesystem Mounts

Mount host directories or use in-memory/snapshot filesystems:

from matchlock import Client, Sandbox, MountConfig

sandbox = (
    Sandbox("alpine:latest")
    .mount_host_dir("/workspace/src", "/home/user/project/src")
    .mount_host_dir_readonly("/workspace/config", "/home/user/project/config")
    .mount_memory("/workspace/tmp")
    .mount_overlay("/workspace/data", "/home/user/project/data")
    # Or use the generic mount() method:
    .mount("/workspace/custom", MountConfig(type="host_fs", host_path="/tmp/custom"))
)

Pass owner_uid and/or owner_gid to have all files in the mount appear owned by a specific user inside the VM. This is useful when the sandbox runs as a non-root user (via .with_user()). Both values must be integers in [0, 4294967295]; a ValueError is raised otherwise. Only host_fs mounts support this option:

sandbox = (
    Sandbox("alpine:latest")
    .with_user("1000:1000")
    .mount_host_dir("/workspace/src", "/home/user/project/src", owner_uid=1000, owner_gid=1000)
    .mount_host_dir_readonly("/workspace/config", "/home/user/project/config", owner_uid=1000)
)

Custom Configuration

from matchlock import Client, Config

config = Config(
    binary_path="/usr/local/bin/matchlock",
    use_sudo=True,  # Required for TAP devices on Linux
)

with Client(config) as client:
    ...

The binary path can also be set via the MATCHLOCK_BIN environment variable.

Lifecycle Management

from matchlock import Client, Sandbox

client = Client()
client.start()

client.launch(Sandbox("alpine:latest"))
print(client.vm_id)  # e.g., "vm-abc12345"

result = client.exec("echo hello")

# Shut down the sandbox VM
client.close()

# Remove the stopped VM's state directory
client.remove()

Error Handling

from matchlock import Client, Sandbox, MatchlockError, RPCError

with Client() as client:
    try:
        client.launch(Sandbox("alpine:latest"))
        result = client.exec("exit 1")
        if result.exit_code != 0:
            print(f"Command failed: {result.stderr}")
    except RPCError as e:
        print(f"RPC error [{e.code}]: {e.message}")
        if e.is_vm_error():
            print("VM-level failure")
        elif e.is_exec_error():
            print("Execution failure")
        elif e.is_file_error():
            print("File operation failure")
    except MatchlockError as e:
        print(f"Matchlock error: {e}")

API Reference

Sandbox(image: str)

Fluent builder for sandbox configuration.

Method	Description
.with_cpus(n)	Set number of vCPUs (supports fractional values, e.g. 0.5)
.with_privileged()	Enable privileged mode (skip in-guest seccomp/cap-drop/no_new_privs)
.with_memory(mb)	Set memory in MB
.with_disk_size(mb)	Set disk size in MB
.with_timeout(seconds)	Set max execution time
.with_workspace(path)	Set guest VFS mount point (default: /workspace)
.allow_host(*hosts)	Add allowed network hosts (supports wildcards)
.block_private_ips()	Block access to private IP ranges
.with_block_private_ips(enabled)	Explicitly set private IP blocking true/false
.allow_private_ips()	Explicitly allow private IP ranges
.unset_block_private_ips()	Reset private IP behavior to SDK default semantics
.with_network_mtu(mtu)	Override guest network stack/interface MTU
.with_no_network()	Disable guest network egress entirely
.with_port_forward(local_port, remote_port)	Add a host-to-guest port mapping
.with_port_forward_addresses(*addresses)	Set bind addresses for configured port mappings
.with_network_interception(config=None)	Force interception and optionally apply typed network hook rules
.add_secret(name, value, *hosts)	Inject a secret for specific hosts
.mount(guest_path, config)	Add a VFS mount with custom MountConfig
.mount_host_dir(guest, host, *, owner_uid=None, owner_gid=None)	Mount a host directory (read-write); optionally override file ownership inside the VM
.mount_host_dir_readonly(guest, host, *, owner_uid=None, owner_gid=None)	Mount a host directory (read-only); optionally override file ownership inside the VM
.mount_memory(guest_path)	Mount an in-memory filesystem
.mount_overlay(guest, host)	Mount an isolated snapshot of host path
.options()	Return the built CreateOptions

Client(config: Config | None = None)

JSON-RPC client for interacting with Matchlock sandboxes. All public methods are thread-safe.

Method	Description
.start()	Start the matchlock RPC subprocess
.launch(sandbox)	Create a VM and start image ENTRYPOINT/CMD in detached mode — returns VM ID
.create(opts)	Create a VM from CreateOptions (does not auto-start ENTRYPOINT unless launch_entrypoint=True) — returns VM ID
.exec(command, working_dir="")	Execute a command, returns ExecResult
.exec_stream(command, stdout, stderr, working_dir)	Stream command output, returns ExecStreamResult
.log()	Return the current buffered VM log as str
.log_stream(stdout=None)	Stream VM log output until cancelled
.exec_pipe(command, stdin=None, stdout=None, stderr=None, working_dir="")	Bidirectional pipe-mode exec (no PTY), returns ExecPipeResult
.exec_interactive(command, stdin=None, stdout=None, working_dir="", rows=24, cols=80, resize=None)	Interactive PTY exec, returns ExecInteractiveResult
.write_file(path, content, mode=0o644)	Write a file into the sandbox
.read_file(path)	Read a file from the sandbox — returns bytes
.list_files(path)	List directory contents — returns list[FileInfo]
.port_forward(*specs)	Apply one or more [LOCAL_PORT:]REMOTE_PORT mappings
.port_forward_with_addresses(addresses, *specs)	Apply port mappings bound on specific host addresses
.close(timeout=0)	Shut down the sandbox VM. timeout in seconds; 0 = kill immediately
.remove()	Remove the stopped VM's state directory
.vm_id	The current VM ID (property)

Types

Type	Fields
Config	binary_path: str, use_sudo: bool
CreateOptions	image, privileged, cpus, memory_mb, disk_size_mb, timeout_seconds, allowed_hosts, block_private_ips, block_private_ips_set, no_network, force_interception, network_interception, mounts, env, vfs_interception, secrets, workspace, dns_servers, network_mtu, port_forwards, port_forward_addresses, image_config, launch_entrypoint
ExecResult	exit_code: int, stdout: str, stderr: str, duration_ms: int
ExecStreamResult	exit_code: int, duration_ms: int
ExecPipeResult	exit_code: int, duration_ms: int
ExecInteractiveResult	exit_code: int, duration_ms: int
FileInfo	name: str, size: int, mode: int, is_dir: bool
MountConfig	type: str, host_path: str, readonly: bool
PortForward	local_port: int, remote_port: int
PortForwardBinding	address: str, local_port: int, remote_port: int
Secret	name: str, value: str, hosts: list[str]
MatchlockError	Base exception for all Matchlock errors
RPCError	RPC error with code: int and message: str

License

MIT