[TEST, do not merge] Sample code for: Python Sandbox: Run Untrusted Code in Docker and MicroVMs

python-sandbox/README.md

@@ -0,0 +1,85 @@
+# Python Sandbox: Run Untrusted Code in Docker and MicroVMs
+
+This folder provides the code examples for the Real Python tutorial [Python Sandbox: Run Untrusted Code in Docker and MicroVMs](https://realpython.com/python-sandbox/).
+
+## Prerequisites
+
+- A Linux host (the hostile snippet targets the Linux `/home` folder, and the shell commands assume a POSIX shell). On macOS or Windows, use a Linux virtual machine or WSL 2.
+- Python 3.15
+- Docker (Podman works too, see the tutorial for the two small changes in `sandbox.py`)
+- [RestrictedPython](https://pypi.org/project/RestrictedPython/) 8.5, for `compare.py`
+- For `microvm.py` only: the [Firecracker](https://firecracker-microvm.github.io/) binary on your `PATH`, a guest kernel saved as `vmlinux`, and read/write access to `/dev/kvm` (hardware virtualization)
+
+If you run Docker or Podman rootless, make sure your host uses cgroups version 2, or the resource limits are ignored.
+
+## Setup
+
+Create and activate a virtual environment, then install the dependencies:
+
+```console
+$ python -m venv venv/
+$ source venv/bin/activate
+(venv) $ python -m pip install -r requirements.txt
+```
+
+Pull the container image up front so that the first sandboxed run doesn't spend its time budget on a download:
+
+```console
+$ docker pull python:3.15-slim
+```
+
+## Files
+
+- `sandbox.py`: Defines `run_untrusted()`, which runs code in a locked-down Docker container with no network, a read-only root file system, resource caps, and a wall-clock timeout.
+- `compare.py`: Runs the same hostile snippet three ways: `exec()` with restricted builtins, RestrictedPython, and the locked-down container.
+- `microvm.py`: Defines `run_in_microvm()`, which runs code inside a Firecracker microVM with its own kernel.
+- `rootfs/sbin/run-job`: The shell script that the microVM guest runs instead of an init system.
+
+## Usage
+
+### Docker Sandbox
+
+Import `run_untrusted()` from `sandbox.py`:
+
+```pycon
+>>> from sandbox import run_untrusted
+>>> run_untrusted("print('Hello from the sandbox')")
+```
+
+### Comparing the Three Tiers
+
+Run `compare.py` from the same folder as `sandbox.py`:
+
+```console
+(venv) $ python compare.py
+```
+
+The first tier leaks your host's `/home` listing, the second refuses to compile the code, and the third shows the container's empty `/home` folder.
+
+### Firecracker MicroVM
+
+First, download a Firecracker release binary and a guest kernel for your CPU architecture by following Firecracker's [getting started guide](https://github.com/firecracker-microvm/firecracker/blob/main/docs/getting-started.md). Save the kernel as `vmlinux` in an empty project folder, copy `microvm.py` into it, and make sure your user can read and write `/dev/kvm`.
+
+Then, build the root file system image from the `python:3.15-slim` image inside that project folder. Export the image first, and copy `rootfs/sbin/run-job` from this folder into the exported tree afterward. In the Debian-based image, `/sbin` is a symbolic link to `/usr/sbin`, so exporting on top of an existing `rootfs/sbin/` folder won't work:
+
+```console
+$ cid=$(docker create python:3.15-slim)
+$ mkdir rootfs/
+$ docker export "$cid" | tar -x -C rootfs/
+$ docker rm "$cid"
+$ mkdir rootfs/job/
+$ cp /path/to/python-sandbox/rootfs/sbin/run-job rootfs/sbin/run-job
+$ chmod +x rootfs/sbin/run-job
+$ truncate -s 300M rootfs.ext4
+$ mkfs.ext4 -q -d rootfs/ rootfs.ext4
+```
+
+On some distributions, `mkfs.ext4` lives in `/usr/sbin/`, which may not be on a regular user's `PATH`.
+
+With `vmlinux` and `rootfs.ext4` next to `microvm.py`, call `run_in_microvm()`:
+
+```pycon
+>>> from microvm import run_in_microvm
+>>> run_in_microvm('print("Hello from the microVM")')
+{'exit_code': 0, 'output': 'Hello from the microVM\n', 'error': ''}
+```

python-sandbox/compare.py

@@ -0,0 +1,31 @@
+import warnings
+
+from RestrictedPython import compile_restricted
+
+from sandbox import run_untrusted
+
+# RestrictedPython warns about print() calls, which don't matter here.
+warnings.filterwarnings("ignore", category=SyntaxWarning)
+
+HOSTILE = (
+    "os = [c for c in ().__class__.__base__.__subclasses__()"
+    " if c.__name__ == '_wrap_close'][0].__init__.__globals__\n"
+    "print(os['listdir']('/home'))\n"
+)
+
+print("1. exec() with only print() in builtins:")
+exec(HOSTILE, {"__builtins__": {"print": print}})
+
+print("2. RestrictedPython refused to compile:")
+try:
+    compile_restricted(HOSTILE, "<untrusted>", "exec")
+except SyntaxError as error:
+    print(error.args[0][0])
+
+print("3. Locked-down container:")
+result = run_untrusted(HOSTILE)
+exit_code, stdout, stderr = result.values()
+if exit_code == 0:
+    print(stdout.strip())
+else:
+    print(f"Sandbox failed ({exit_code}): {stderr}")

python-sandbox/microvm.py

@@ -0,0 +1,59 @@
+import json
+import subprocess
+import tempfile
+from pathlib import Path
+
+KERNEL = Path("vmlinux").resolve()
+ROOTFS = Path("rootfs.ext4").resolve()
+BOOT_ARGS = "console=ttyS0 quiet reboot=k panic=1 init=/sbin/run-job"
+START, END = "===JOB-START===", "===JOB-END==="
+
+
+def drive(drive_id, path, is_root):
+    return {
+        "drive_id": drive_id,
+        "path_on_host": str(path),
+        "is_root_device": is_root,
+        "is_read_only": True,
+    }
+
+
+def run_in_microvm(code, timeout=10):
+    with tempfile.TemporaryDirectory() as folder:
+        folder = Path(folder)
+        (folder / "job").mkdir()
+        (folder / "job" / "main.py").write_text(code)
+        job_image = folder / "job.ext4"
+        subprocess.run(["truncate", "-s", "8M", job_image], check=True)
+        subprocess.run(
+            ["mkfs.ext4", "-q", "-d", folder / "job", job_image], check=True
+        )
+        config = {
+            "boot-source": {
+                "kernel_image_path": str(KERNEL),
+                "boot_args": BOOT_ARGS,
+            },
+            "drives": [
+                drive("rootfs", ROOTFS, is_root=True),
+                drive("job", job_image, is_root=False),
+            ],
+            "machine-config": {"vcpu_count": 1, "mem_size_mib": 256},
+        }
+        config_path = folder / "vm_config.json"
+        config_path.write_text(json.dumps(config))
+        try:
+            result = subprocess.run(
+                ["firecracker", "--no-api", "--config-file", config_path],
+                capture_output=True,
+                text=True,
+                errors="replace",
+                timeout=timeout,
+            )
+        except subprocess.TimeoutExpired:
+            return {"exit_code": None, "output": "", "error": "timed out"}
+    console = result.stdout.replace("\r\n", "\n")
+    if START not in console or END not in console:
+        return {"exit_code": None, "output": console, "error": "no job output"}
+    output = console.split(START + "\n", 1)[1].split(END, 1)[0]
+    exit_code = int(console.split(END, 1)[1].split()[0])
+    return {"exit_code": exit_code, "output": output, "error": ""}

python-sandbox/requirements.txt

@@ -0,0 +1 @@
+RestrictedPython==8.5

python-sandbox/rootfs/sbin/run-job

@@ -0,0 +1,10 @@
+#!/bin/sh
+mount -t proc proc /proc
+mount -t devtmpfs devtmpfs /dev 2>/dev/null
+mount -t tmpfs tmpfs /tmp
+mount -o ro /dev/vdb /job
+echo "===JOB-START==="
+setpriv --reuid=65534 --regid=65534 --clear-groups \
+    /usr/local/bin/python /job/main.py
+echo "===JOB-END=== $?"
+echo b > /proc/sysrq-trigger

python-sandbox/sandbox.py

@@ -0,0 +1,46 @@
+import subprocess
+import uuid
+
+IMAGE = "python:3.15-slim"
+RUN = ["docker", "run"]
+REMOVE = ["docker", "rm", "--force"]
+FLAGS = [
+    "--network=none",
+    "--read-only",
+    "--ipc=none",
+    "--cap-drop=ALL",
+    "--security-opt=no-new-privileges",
+    "--user=65534:65534",
+    "--memory=128m",
+    "--memory-swap=128m",
+    "--cpus=1",
+    "--pids-limit=64",
+    "--log-driver=none",
+]
+
+
+def run_untrusted(code, timeout=5):
+    name = f"sandbox-{uuid.uuid4().hex[:12]}"
+    command = [*RUN, "--rm", "--interactive", f"--name={name}", *FLAGS]
+    try:
+        result = subprocess.run(
+            [*command, IMAGE, "python", "-"],
+            input=code,
+            capture_output=True,
+            text=True,
+            errors="replace",
+            timeout=timeout,
+        )
+    except subprocess.TimeoutExpired:
+        cleanup = subprocess.run(
+            [*REMOVE, name], capture_output=True, text=True, timeout=30
+        )
+        if cleanup.returncode != 0:
+            if "no such container" not in cleanup.stderr.lower():
+                raise RuntimeError(f"Couldn't remove {name}: {cleanup.stderr}")
+        return {"exit_code": None, "stdout": "", "stderr": "timed out"}
+    return {
+        "exit_code": result.returncode,
+        "stdout": result.stdout,
+        "stderr": result.stderr,
+    }