Skip to content

Latest commit

 

History

History
976 lines (719 loc) · 36.5 KB

README.md

File metadata and controls

976 lines (719 loc) · 36.5 KB

sandlock Python SDK

Python bindings for sandlock, a lightweight process sandbox using Landlock and seccomp. No root, no Docker, no namespaces required.

Requires Linux 6.7+ with Landlock ABI v6.

pip install sandlock

Quick start

from sandlock import Sandbox

sandbox = Sandbox(
    fs_readable=["/usr", "/lib", "/lib64", "/bin", "/etc", "/proc", "/dev"],
    fs_writable=["/tmp"],
)
result = sandbox.run(["echo", "hello"], timeout=10)
assert result.success
print(result.stdout)  # b"hello\n"

API reference

Platform

sandlock.landlock_abi_version() -> int

Return the Landlock ABI version supported by the running kernel. Returns -1 if Landlock is unavailable.

sandlock.min_landlock_abi() -> int

Return the minimum Landlock ABI version required by sandlock (currently 6).

Confine

Apply a Sandbox's Landlock rules to the current process, in place. No fork, no exec.

from sandlock import Sandbox, confine

confine(Sandbox(
    fs_readable=["/usr", "/lib"],
    fs_writable=["/tmp"],
))

sandlock.confine(sandbox) -> None

Set PR_SET_NO_NEW_PRIVS and install the sandbox's Landlock ruleset on the live process. The confinement is irreversible.

Only Landlock fields are honored (fs_readable, fs_writable, fs_denied); IPC and signal scoping are always applied. Sandbox config that requires a supervisor or a fresh child (seccomp, network, resource limits, COW, env, policy_fn, etc.) is rejected rather than silently ignored. Raises ConfinementError on failure.

Sandbox

sandlock.Sandbox(**kwargs)

Sandbox configuration and runtime handle. Holds both the policy (filesystem, network, resource limits, etc.) and runtime state. Construct once, then call run() (blocking) or spawn() + lifecycle methods, or use as a context manager.

All config fields are optional. Unset fields mean "no restriction" unless noted otherwise. Runtime kwargs (name, registration, policy_fn, init_fn, work_fn) are set at construction time alongside config fields.

A single Sandbox instance holds at most one running process at a time. For concurrent execution, create multiple instances.

Runtime kwargs:

  • registration: defaults to True. Set Sandbox(registration=False) to skip this instance's name claim and control socket. It will not appear in sandlock ps or accept sandlock kill by name, and its name will not be reserved. The virtual hostname and direct lifecycle methods still work. Set before creation; other sandboxes and existing running instances are unaffected. This setting is excluded from serialized policy.

  • name -- sandbox name (also its virtual hostname inside the sandbox). Auto-generated as sandbox-{pid} when omitted.

  • policy_fn -- optional callback for dynamic per-event policy decisions (see Dynamic policy).

  • init_fn / work_fn -- callbacks for COW fork mode (see Fork).

Sandbox is a context manager:

with Sandbox(fs_readable=["/usr", "/lib"]) as sb:
    result = sb.run(["echo", "hello"])

Filesystem (Landlock)

Parameter Type Default Description
fs_readable list[str] [] Paths the sandbox can read
fs_writable list[str] [] Paths the sandbox can read and write
fs_denied list[str] [] Paths explicitly denied
workdir str | None None Working directory; enables COW protection
chroot str | None None Path to chroot into before confinement
fs_mount dict[str, str] {} Map virtual paths to host directories inside chroot
fs_mount_ro dict[str, str] {} Like fs_mount, but writes under the virtual path are denied
image Image | None None Run inside a container image from pull_image()
cwd str | None None Child working directory

Network

Parameter Type Default Description
net_allow list[str] [] Outbound endpoint rules. Bare host:port is TCP; protocol prefixes opt others in: tcp://host:port, udp://host:port (or udp://*:* for any UDP), icmp://host (or icmp://* for any ICMP echo via the kernel ping socket — gated by net.ipv4.ping_group_range on the host). Empty = deny all. Raw ICMP is not exposed.
net_deny list[str] [] Default-allow outbound denylist for literal IP/CIDR targets. When combined with net_allow, denied destinations win.
net_allow_bind list[int | str] [] TCP ports the sandbox may bind (empty = deny all; ["*"] = allow any port; a listed 0 authorizes only bind(0))
net_deny_bind list[int | str] [] Default-allow TCP bind denylist. When combined with net_allow_bind, denied ports win.
port_remap bool False Transparent TCP port virtualization

HTTP ACL

Enforce method + host + path rules on HTTP traffic via a transparent MITM proxy. When http_allow is set, all non-matching HTTP requests are denied by default. Block rules are checked first and take precedence.

Parameter Type Default Description
http_allow list[str] [] Allow rules in "METHOD host/path" format
http_deny list[str] [] Block rules in "METHOD host/path" format
http_ports list[int] [80] TCP ports to intercept (443 added when http_ca is set)
http_ca str | None None CA certificate for HTTPS MITM
http_key str | None None CA private key for HTTPS MITM
http_inject_ca list[str] [] Trust bundle paths to splice the active MITM CA's public cert into at open time. Without http_ca, generates an ephemeral CA (key in memory only) and intercepts port 443. Requires an http_allow / http_deny rule
http_ca_out str | None None Writes the active CA's public certificate (PEM) to this path; never the private key. Requires an http_allow / http_deny rule

Rule format: "METHOD host/path" where method and host can be * for wildcard, and path supports trailing * for prefix matching. Paths are normalized (percent-decoding, .. resolution, // collapsing) before matching to prevent bypasses.

sandbox = Sandbox(
    fs_readable=["/usr", "/lib", "/etc"],
    http_allow=[
        "GET docs.python.org/*",
        "POST api.openai.com/v1/chat/completions",
    ],
    http_deny=["* */admin/*"],
)
result = sandbox.run(["python3", "agent.py"])

Chroot with mount mapping

Map host directories into a chroot — like Docker's -v /host:/container but without kernel bind mounts or root privileges. Each sandbox gets its own persistent workspace while sharing a read-only rootfs.

sandbox = Sandbox(
    chroot="/opt/rootfs",
    fs_mount={"/work": "/tmp/sandbox-1/work"},
    fs_readable=["/usr", "/bin", "/lib", "/etc"],
    cwd="/work",
)
result = sandbox.run(["python3", "task.py"])

Combine with workdir + max_disk for quota-enforced writes:

sandbox = Sandbox(
    chroot="/opt/rootfs",
    fs_mount={"/work": "/tmp/sandbox-1/work"},
    workdir="/tmp/sandbox-1/work",
    fs_storage="/tmp/sandbox-1/cow",
    max_disk="100M",
    on_exit="commit",
    fs_readable=["/usr", "/bin", "/lib", "/etc"],
)

Container images

sandlock.pull_image(reference, cache_dir=None) -> Image fetches and unpacks a container image without a Docker daemon or root. Sandbox(image=...) runs inside it: the image's rootfs becomes the chroot, read access to / inside it is granted, and its env and working directory fill only what env and cwd leave unset.

Like a container's writable layer, every write lands in a copy-on-write branch that is discarded when the run ends, so the cached image never changes. To keep output, mount a host directory with fs_mount and grant it in fs_writable; setting workdir, or an on_error other than "abort", is rejected, and on_exit is always abort. The cache belongs to the invoking user, so it is only as protected as that user's other files.

Reference Source
docker://python:3.12, docker://ghcr.io/org/img@sha256:... registry (Docker Hub by default)
oci:<dir>[:tag] OCI image layout directory
oci-archive:<file>[:tag] tar of an OCI image layout
docker-daemon:<ref>, docker-daemon:sha256:<id> local Docker daemon (Docker 25+)
from sandlock import Sandbox, pull_image

image = pull_image("docker://python:3.12-slim")
result = Sandbox(image=image, max_memory="512M").run(["python3", "-c", "print('hello')"])

# Keep output on the host; writes elsewhere are discarded.
Sandbox(image=image, fs_mount={"/out": "/srv/job-1"}, fs_writable=["/out"]).run(
    ["python3", "-c", "open('/out/result.txt', 'w').write('done')"]
)

# Or the image's own command:
Sandbox(image=image).run(image.config.default_cmd())

Image has rootfs and config; ImageConfig has entrypoint, cmd, env (KEY=VALUE strings), working_dir, and default_cmd(). Both are frozen dataclasses. References use skopeo's transport syntax (containers-transports(5)): the transport prefix is required, and a tag together with a digest is rejected. Registry credentials come from docker login's ~/.docker/config.json (or $DOCKER_CONFIG); credential helpers are not run. Every blob is verified against its digest, and each image is unpacked once into cache_dir, or $XDG_CACHE_HOME/sandlock/images by default. A cached image named by digest starts without network access. Failures raise SandlockError.

Resource limits

Parameter Type Default Description
max_memory str | int | None None Memory limit, e.g. "512M" or int bytes
max_processes int | None None Peak concurrent process limit; None is unlimited
max_open_files int | None None Max file descriptors (RLIMIT_NOFILE)
max_cpu int | None None CPU throttle as percentage of one core (1-100)
cpu_cores list[int] | None None CPU cores to pin sandbox to
num_cpus int | None None Visible CPU count in /proc/cpuinfo

Syscall filtering (seccomp)

Parameter Type Default Description
extra_deny_syscalls list[str] [] Extra syscall or syscall-group names to block in addition to Sandlock defaults (groups expand to their member syscalls)
extra_allow_syscalls list[str] [] Syscall groups to allow that are blocked by default (e.g. "sysv_ipc" to enable SysV shared memory, semaphores, and message queues); unknown groups and individual syscall names are rejected

Sandlock always applies its default syscall blocklist.

Deterministic execution

Parameter Type Default Description
random_seed int | None None Seed for deterministic getrandom()
time_start datetime | str | None None Start time for time virtualization: RFC 3339 with an offset, or an aware datetime
no_randomize_memory bool False Disable ASLR
no_huge_pages bool False Disable Transparent Huge Pages
deterministic_dirs bool False Sort directory entries lexicographically

Environment

Parameter Type Default Description
clean_env bool False Start with minimal environment
env dict[str, str] {} Variables to set/override in the child

GPU access

Parameter Type Default Description
gpu_devices list[int] | None None GPU device indices to expose ([] = all)

Misc

Parameter Type Default Description
uid int | None None Map to given UID inside a user namespace (e.g. 0 for fake root). Set together with gid
gid int | None None Map to given GID inside the user namespace. Must be set together with uid (both or neither)
no_coredump bool False Disable core dumps

COW filesystem isolation

Parameter Type Default Description
fs_storage str | None None Storage directory for the seccomp COW upper layer / deltas
max_disk str | int | None None Disk quota for COW storage, e.g. "1G" or int bytes
on_exit BranchAction COMMIT COMMIT, ABORT, KEEP, or DEFER
on_error BranchAction ABORT COMMIT, ABORT, KEEP, or DEFER

Protection opt-out

By default sandlock enforces every Landlock protection the host kernel supports and refuses to start when a required protection is unavailable. Two keyword arguments on Sandbox opt out of the strict default on a per-protection basis:

Parameter Type Default Description
allow_degraded list[Protection] [] Enforce each listed protection where the host kernel supports it, silently skip it where it does not.
disable list[Protection] [] Never enforce each listed protection, even on a kernel that supports it.
from sandlock import Sandbox, Protection

sb = Sandbox(
    fs_readable=["/data"],
    fs_writable=["/tmp"],
    allow_degraded=[Protection.SIGNAL_SCOPE, Protection.ABSTRACT_UNIX_SOCKET_SCOPE],
)

The two allow_degraded entries let the sandbox build on Linux kernels below 6.12, where the v6 IPC scopes are unavailable; on a capable kernel the scopes remain enforced. The protection policy is persisted with a checkpoint, so a restored sandbox keeps the exact posture it was built with. See the "Protection opt-out" section of ../docs/sandbox-reference.md for the per-protection ABI floors and the full semantics.

sandbox.run(cmd, timeout=None) -> Result

Run a command, capturing stdout and stderr.

  • cmd -- list of strings (command and arguments).
  • timeout -- max execution time in seconds (float). None = no timeout.
result = sandbox.run(["python3", "-c", "print(42)"], timeout=10.0)

sandbox.spawn(cmd) -> None

Spawn cmd without waiting. Use pid, pause(), resume(), kill(), wait(), join() and shutdown() to manage the session.

Raises RuntimeError if a process is already running.

Sugar for create(cmd) + start(); use those directly when you need the fork-park-exec split (e.g. starting several sandboxes in lockstep, or attaching external tracing to the parked PID before the child execs).

sandbox.create(cmd) -> None

Fork the sandboxed child and install policy. The child is parked between policy install and execve; call start() to release it. pid is available after this call but the child is not yet running user code.

Raises RuntimeError if a process is already running.

sandbox.start() -> None

Release a previously create()d child to execve.

Raises RuntimeError if no child has been created.

sandbox.wait(timeout=None) -> ExitStatus

Wait for the main process and return its ExitStatus. Background jobs it started keep running under supervision. A timeout raises TimeoutError and leaves the session running.

sandbox.join(timeout=None) -> ExitStatus

Like wait(), but also wait until every background job has exited, so shutdown() has nothing left to kill. Returns main's ExitStatus. A timeout raises TimeoutError and leaves the session running.

sandbox.shutdown() -> Result

Kill whatever is left of the session, wait until it has no process, and return the Result with captured output and filesystem changes.

kill(), pause(), resume() and process_groups() work from any thread, including while another thread is blocked in wait() or join(). A popen() Process has the same wait(), join(), shutdown() and process_groups().

Inspecting and deferring COW changes

Every Result from a sandbox with a workdir carries changes, the list of files and directories the run added, modified, or deleted in its COW branch. A dry run is a run whose branch action is ABORT:

sandbox = Sandbox(workdir=wd, on_exit=BranchAction.ABORT, ...)
result = sandbox.run(["sh", "-c", "echo hi > out.txt"])
for change in result.changes:
    print(change.kind, change.path)  # "A out.txt"

BranchAction.DEFER leaves the branch in the sandbox after the run exits so the caller can inspect it and decide later. While pending, upper_dir holds the new bytes of every added or modified file, laid out like workdir:

sandbox = Sandbox(workdir=wd, on_exit=BranchAction.DEFER, ...)  # on_error stays ABORT
result = sandbox.run(["python3", "tool.py"])
if sandbox.pending:
    if approve(result.changes, sandbox.upper_dir):
        sandbox.commit()
    else:
        sandbox.abort()
  • sandbox.pending -> bool: True between a DEFER run's exit and commit() / abort().
  • sandbox.upper_dir -> str | None: the pending branch's upper directory.
  • sandbox.commit(): merge into workdir. Blocks up to 5s on a workdir another sandbox is merging into; raises BranchError if the merge fails, leaving the branch preserved on disk. Last-writer-wins against anything that changed the workdir since the run.
  • sandbox.abort(): discard the branch.

A pending sandbox refuses another run() until it is decided. Leaving a with block, or letting the sandbox go away, preserves an undecided branch on disk the same way KEEP does; nothing is ever merged without a decision.

sandbox.run_interactive(cmd) -> int

Run with inherited stdio (no capture). Returns the exit code.

sandbox.name -> str | None

The sandbox name.

sandbox.pid -> int | None

Main's PID until it exits, None otherwise.

sandbox.is_running -> bool

True if a process is currently running in this sandbox.

sandbox.ports() -> dict[int, int]

Current port mappings {virtual_port: real_port} while running. Only contains entries where port remapping occurred. Requires port_remap=True.

sandbox.pause() / sandbox.resume() / sandbox.kill()

Send SIGSTOP / SIGCONT / SIGKILL to every process group of the session. Raises RuntimeError if the sandbox is not running.

sandbox.checkpoint(save_fn=None) -> Checkpoint

Capture a checkpoint of the running sandbox. See Checkpoint.

sandbox.restore_interactive(cp) -> None

Restore a checkpoint into a fresh, fully-sandboxed process. The checkpoint image is injected over a parked child, which resumes at the saved program counter under the full confinement; the sandbox is running when this returns (no start() needed). Manage it with pid, pause(), resume(), kill(), and wait() as usual.

x86_64 only. The image is rebuilt by an execve'd restore stub in a fresh, already-confined process, and the kernel vDSO is relocated onto the checkpoint-recorded base, so ordinary libc programs resume. Assumes the restore runs on the same kernel that took the checkpoint.

Raises RuntimeError if a process is already running in this sandbox or the restore fails. See Checkpoint.

sandbox.restore_skipped -> list[SkippedFd]

Fds the last restore_interactive() could not transparently recreate (sockets, pipes, memfds, pseudo-filesystem paths); the restored process runs without them. Each entry is a SkippedFd named tuple with fd (the fd number in the checkpointed process) and path (the resource it pointed at, e.g. pipe:[12345]). Empty if this sandbox never restored a checkpoint or every fd was restored.

Result

Returned by sandbox.run().

Attribute Type Description
success bool True if exit code is 0
exit_code int Process exit code
stdout bytes Captured standard output
stderr bytes Captured standard error
error str | None Error message on failure
changes list[Change] Filesystem changes the run made to its COW branch (empty without workdir)

Change

Attribute Type Description
path str Path relative to workdir
before Entry | None The workdir entry when the run first touched the path; None if absent
after Entry | None The branch entry when the change set was read; None if removed
kind str Derived: "A" (no before), "M" (both sides), "D" (no after)
content_unchanged bool Both sides present with the same kind and digest or target
type_changed bool Both sides present with different kinds

renames(changes) pairs each deleted file with the added file carrying the same digest, as (old_path, new_path) tuples.

Entry

Attribute Type Description
kind str "file", "dir", "symlink", or "other"
mode int Permission bits
size int Byte length for a file; 0 otherwise
digest bytes | None SHA-256 of the bytes; files only
target str | None Link target; symlinks only

Stage and Pipeline

Chain sandboxed commands with pipes using the | operator:

result = (
    sandbox_a.cmd(["echo", "hello"])
    | sandbox_b.cmd(["tr", "a-z", "A-Z"])
).run()
assert result.stdout == b"HELLO\n"

sandbox.cmd(args) -> Stage

Create a lazy Stage bound to this sandbox.

pipeline.run(stdout=None, timeout=None) -> Result

Run the pipeline. Each stage's stdout feeds the next stage's stdin.

Prompt guard

PromptGuard, Rule, ScanReport, Finding, ScanError, StatisticalClassifier, and TransformersClassifier are available from sandlock_guard and sandlock.guard; only the latter adds stage().

PromptGuard(threshold="medium", max_bytes=1_048_576, scan_timeout=2.0, rules=(), model=None, classifier=None)

threshold: low, medium, high, or critical. max_bytes: positive UTF-8 input byte limit. scan_timeout: positive finite stage scanning deadline in seconds, starting after EOF. rules: iterable of custom Rule objects, copied to an immutable tuple and added to the built-in rules. Duplicate IDs, including built-in IDs, raise ValueError; non-Rule entries raise TypeError. classifier: optional StatisticalClassifier or TransformersClassifier configuration. model: compatibility alias for classifier=StatisticalClassifier(model). Supplying both raises ValueError; unsupported classifiers raise TypeError. A classifier score meeting its own threshold adds a high severity finding. The IDs statistical-injection and transformers-injection are reserved.

StatisticalClassifier(path, threshold=None)

Load an existing classifier JSON file. threshold=None uses the value in the file; an explicit threshold must be strictly between 0 and 1. Invalid models raise ValueError; file access errors raise OSError.

TransformersClassifier(path, positive_labels, threshold=0.5, max_length=512, stride=256)

Configure local CPU inference; requires the optional transformers extra. positive_labels: nonempty sequence of exact class names whose softmax scores are summed. Supports mutually exclusive classification with at least two classes. max_length: 4 to 8192 tokens, within the model's limit. stride: overlapping tokens, less than the window's content capacity. threshold: strictly between 0 and 1. path names a directory with regular model and fast-tokenizer files, fingerprinted at construction. Runtime loading is lazy; scan failures raise ScanError.

Rule(id: str, pattern: str, severity: str, message: str)

Immutable custom rule. id starts with an ASCII letter and contains up to 64 letters, digits, underscores, dots, or hyphens. pattern is a nonempty Python regex compiled case-insensitively; severity uses the same levels as threshold. message is a nonblank finding description. Invalid fields raise ValueError.

guard.scan(text: str) -> ScanReport

Scan synchronously without a wall-clock timeout. Raises TypeError for non-string input or ScanError if inspection cannot complete.

guard.stage(*, max_memory=None) -> Stage

Return a sandboxed stage forwarding original UTF-8 bytes only after approval. Exit codes: 0 accepted, 1 rejected, 2 input or scanning error. max_memory overrides the default memory limit: 256 MiB for rules or statistics, 2 GiB for Transformers. Changed model files fail closed at worker startup.

Report and error types

  • ScanReport: flagged: bool (any finding meets the threshold), findings: tuple[Finding, ...], input_bytes: int, ruleset_version: str, model_score: float | None (maximum window score), model_digest: str | None (SHA-256 of the JSON file or neural model file manifest).
  • Finding: rule_id: str, severity: str, message: str.
  • ScanError.code: str: inspection failure identifier.

See Prompt injection defenses for usage and limits.

Gather

Fan multiple producers into one consumer via named pipes. Each producer's stdout is delivered to the consumer under a label the consumer reads from a sandlock.inputs dict. This is the structural primitive behind the XOA pattern: producers and consumer are independent sandboxes, and the consumer never executes producer code inside its own LLM call.

from sandlock import Sandbox

planner  = Sandbox(...)   # writes code
searcher = Sandbox(...)   # produces data
executor = Sandbox(...)   # consumes both

result = (
    searcher.cmd(["python3", "-c", "..."]).as_("data")
    + planner.cmd(["python3", "-c", "..."]).as_("code")
    | executor.cmd(["python3", "consume.py"])
).run()

Inside consume.py:

from sandlock import inputs

code = inputs["code"]
data = inputs["data"]
exec(compile(code, "<planner>", "exec"), {"data": data})

stage.as_(name) -> NamedStage

Label a Stage's stdout so the consumer can address it by name.

named_stage + named_stage_or_gather -> Gather

Combine two or more NamedStage values into a Gather. Repeated + extends the gather:

g = a.as_("x") + b.as_("y") + c.as_("z")

gather | consumer_stage -> GatherPipeline

Compose a Gather with a consumer Stage to form the runnable pipeline.

gather_pipeline.run(timeout=None) -> Result

Run all producers in parallel; each producer's stdout is wired to a pipe the consumer reads via inputs[name]. Returns the consumer's Result.

sandlock.inputs

Lazy dict-like accessor available inside the consumer process. Reads each producer's pipe on first access and caches the value.

from sandlock import inputs

inputs["code"]        # str: the producer's full stdout, decoded as utf-8
"data" in inputs      # bool
list(inputs.keys())   # ["data", "code"]

The pipe fds are passed via the _SANDLOCK_GATHER env var (name:fd,name:fd,...); the inputs object parses it on first access. Users do not interact with the env var directly.

Dynamic policy

Configure PolicyFn(callback, verdict_timeout=...) with a number of seconds (default 5.0) or None for unlimited verdict waits. Approval callbacks must check ctx.is_cancelled() between bounded waits and return on shutdown. See verdict timeout and shutdown. Pass deferred=True to receive an Approval for each held syscall and answer it later from any thread; see deferred approval.

Use policy_fn to make per-syscall decisions at runtime:

from sandlock import Sandbox, PolicyFn, SyscallEvent, PolicyContext, Allow, Deny

def my_policy(event: SyscallEvent, ctx: PolicyContext):
    if event.category == "network" and event.host == "evil.com":
        return Deny()
    if event.category == "filesystem" and "/secrets" in (event.path or ""):
        ctx.deny_path("/secrets")
        return Deny()
    return Allow()

sb = Sandbox(..., policy_fn=PolicyFn(my_policy))

SyscallEvent

Attribute Type Description
syscall str Syscall name (e.g. "openat", "connect")
category str "filesystem", "network", or "process"
pid int Process ID
parent_pid int Parent process ID
path str | None File path (for filesystem events)
host str | None Hostname (for network events)
port int Port number (for network events)
argv tuple[str, ...] | None Command arguments (for execve)
denied bool Whether this event was already denied by static policy

Helper methods:

  • event.path_contains(s) -- True if path contains substring s
  • event.argv_contains(s) -- True if any argv element contains s

PolicyContext

Methods available inside policy_fn:

Method Description
ctx.restrict(net_allow=specs) Narrow the live and approvable network rules
ctx.restrict_max_memory(bytes) Lower memory limit
ctx.restrict_max_processes(n) Lower process limit
ctx.restrict_pid(pid, net_allow=specs) Narrow one process below the live rules
ctx.deny_path(path) Deny access to a path
ctx.allow_path(path) Remove a previously denied path

Runtime specs take IPs and CIDRs only. Widening the network happens by answering with Grant(net_allow=specs) within PolicyFn(approvable=Permissions(...)); see approve network on demand. The static net_deny layer is checked after the allow verdict and always wins.

Callback return values:

Return Meaning
Allow(), None Allow
Deny() Deny (EPERM)
Deny(errno) Deny with that errno
Audit() Allow but flag for audit
Grant(net_allow=specs) Widen the network within approvable, then allow

Any other return value raises TypeError and denies the syscall.

Fork

COW fork for parallel execution with shared initialization:

sb = Sandbox(
    fs_readable=[...],
    init_fn=lambda: load_model(),
    work_fn=lambda clone_id: process(clone_id),
)
clones = sb.fork(4)  # returns ForkResult with .pids

sandbox.reduce(cmd, fork_result) -> Result

Pipe combined clone output into a reducer command:

result = sandbox.reduce(["python3", "sum.py"], clones)

Checkpoint

Save and restore sandbox state:

sb = Sandbox(...)
# ... start a long-running process ...
cp = sb.checkpoint(save_fn=lambda: my_state_bytes())
cp.save("my-snapshot")

# Later:
cp2 = Checkpoint.load("my-snapshot", restore_fn=lambda data: rebuild(data))

# Or restore the OS-level process image into a fresh sandbox
# (x86_64, vDSO-free programs):
sb2 = Sandbox(...)   # same policy
sb2.restore_interactive(cp2)
for f in sb2.restore_skipped:
    print(f"fd {f.fd} not restored: {f.path}")
Method Description
cp.save(name, store=None) Persist checkpoint to disk
Checkpoint.load(name, store=None, restore_fn=None) Load from disk; if the checkpoint carries app_state and restore_fn is given, call restore_fn with it. Neither save_fn nor restore_fn is mandatory, and they need not be paired
Checkpoint.list(store=None) List saved checkpoint names
Checkpoint.delete(name, store=None) Delete a saved checkpoint
sb.restore_interactive(cp) Restore the process image into a fresh sandbox; see Sandbox
sb.restore_skipped Fds the last restore could not recreate, as SkippedFd(fd, path)

Properties: cp.name (str), cp.app_state (bytes or None).

Default store: ~/.sandlock/checkpoints/.

Profiles

Load sandbox configuration from TOML files in ~/.config/sandlock/profiles/. They are parsed by the same core parser as the CLI, so a profile means the same thing to both. Profiles contain sandbox config only; pass the sandbox name at construction: Sandbox(..., name=...).

from sandlock import load_profile, list_profiles

sandbox = load_profile("web-scraper")
names = list_profiles()

Exceptions

SandlockError (base)
  +-- SandboxError         invalid sandbox configuration
  +-- SandboxRuntimeError  sandbox lifecycle errors
        +-- ForkError          fork failed
        +-- ChildError         child exited abnormally
        +-- BranchError        COW branch operation failed
        |     +-- BranchConflictError   sibling branch committed (ESTALE)
        +-- ConfinementError   Landlock/seccomp setup failed
              +-- LandlockUnavailableError   no Landlock support
              +-- SeccompError               seccomp filter failed
                    +-- NotifError           notif supervisor error
  +-- MemoryProtectError   mprotect failed

All exceptions are importable from sandlock.exceptions or directly from sandlock:

from sandlock import SandlockError, SandboxError, SandboxRuntimeError

Enums

BranchAction

  • BranchAction.COMMIT -- merge writes on exit
  • BranchAction.ABORT -- discard writes
  • BranchAction.KEEP -- leave the branch on disk for recovery tooling
  • BranchAction.DEFER -- hold the branch for commit() / abort()

MCP integration

Sandboxed tool execution for AI agents. Each tool runs in a per-call sandbox with deny-by-default permissions.

pip install 'sandlock[mcp]'

McpSandbox

from sandlock.mcp import McpSandbox

mcp = McpSandbox(workspace="/tmp/agent", timeout=30.0)
  • workspace -- directory the sandbox can read (default: "/tmp/sandlock").
  • timeout -- default timeout in seconds per tool call (default: 30.0).

mcp.add_tool(name, func, *, description="", capabilities=None, input_schema=None)

Register a local tool. func must be a top-level function in an import-safe module: the worker imports that module by name and calls the function in a fresh per-call sandbox. Module-level imports, helpers, constants, and state are all fine; lambdas, methods, and nested functions are rejected. Guard any module startup logic under if __name__ == "__main__":.

A tool that declares a parameter named workspace receives the sandbox's workspace path automatically (injected at call time, hidden from the LLM schema, and not overridable by the model). No env wiring needed.

# tools.py  (an importable module)
import os

def read_file(path: str, *, workspace: str) -> str:
    with open(os.path.join(workspace, path)) as f:
        return f.read()
import tools

mcp.add_tool("read_file", tools.read_file,
    description="Read a file from the workspace",
)

No capabilities = read-only, clean environment, no network. Grant permissions explicitly:

Capability Example Description
fs_writable ["/tmp/agent"] Paths the tool can read and write
net_allow ["api.example.com:443", "udp://1.1.1.1:53"] Outbound endpoints. Bare host:port is TCP; udp://... / icmp://... schemes opt UDP / ICMP echo in.
env {"KEY": "val"} Environment variables to pass
max_memory "256M" Memory limit

Any Sandbox field name is accepted as a capability key.

await mcp.add_mcp_session(session)

Discover tools from a remote MCP server. Capabilities are read from sandlock:* keys in the tool's annotations or meta dict.

await mcp.call_tool(name, arguments=None, *, timeout=None) -> str

Call a tool by name. Local tools run in a per-call sandbox. MCP tools are forwarded to their server session.

result = await mcp.call_tool("read_file", {"path": "data.txt"})

mcp.get_policy(tool_name) -> Sandbox

Return the computed Sandbox for a registered tool.

mcp.tool_definitions_openai() -> list[dict]

Tool definitions in OpenAI function-calling format, for use with chat completion APIs.

mcp.tools -> dict[str, Any]

All registered tools (local and MCP).

MCP server

A standalone MCP server with built-in sandboxed tools (shell, python, read_file, write_file, list_files):

# stdio (for Claude Desktop / Cursor)
sandlock-mcp --workspace /tmp/sandbox

# SSE (remote)
pip install 'sandlock[mcp-remote]'
sandlock-mcp --transport sse --host 0.0.0.0 --port 8080 --workspace /tmp/sandbox

Claude Desktop configuration:

{
  "mcpServers": {
    "sandlock": {
      "command": "sandlock-mcp",
      "args": ["--workspace", "/tmp/sandbox"]
    }
  }
}

policy_for_tool(*, workspace, capabilities=None) -> Sandbox

Build a deny-by-default Sandbox from explicit capabilities. Used internally by McpSandbox but available for direct use.

capabilities_from_mcp_tool(tool) -> dict

Extract sandlock:* capabilities from an MCP tool's annotations/meta.

License

Apache-2.0