🟡 unaiverse.utils.sandbox

What this module does 🟡

Provides Docker-based sandboxing utilities that build an isolated image from project requirements and run agent code inside a container with mounted volumes and resource limits.

sandbox

█████ █████ ██████ █████ █████ █████ █████ ██████████ ███████████ █████████ ██████████ ░░███ ░░███ ░░██████ ░░███ ░░███ ░░███ ░░███ ░░███░░░░░█░░███░░░░░███ ███░░░░░███░░███░░░░░█ ░███ ░███ ░███░███ ░███ ██████ ░███ ░███ ░███ ░███ █ ░ ░███ ░███ ░███ ░░░ ░███ █ ░ ░███ ░███ ░███░░███░███ ░░░░░███ ░███ ░███ ░███ ░██████ ░██████████ ░░█████████ ░██████
░███ ░███ ░███ ░░██████ ███████ ░███ ░░███ ███ ░███░░█ ░███░░░░░███ ░░░░░░░░███ ░███░░█
░███ ░███ ░███ ░░█████ ███░░███ ░███ ░░░█████░ ░███ ░ █ ░███ ░███ ███ ░███ ░███ ░ █ ░░████████ █████ ░░█████░░████████ █████ ░░███ ██████████ █████ █████░░█████████ ██████████ ░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░ ░░░░░ ░░░ ░░░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░░ ░░░░░░░░░░ A Collectionless AI Project (https://collectionless.ai) Registration/Login: https://unaiverse.io Code Repositories: https://github.com/collectionlessai/ Main Developers: Stefano Melacci (Project Leader), Christian Di Maio, Tommaso Guidi

DOCKER_IMAGE_NAME module-attribute

DOCKER_IMAGE_NAME = 'unaiverse-sandbox'

CONTAINER_NAME_BASE module-attribute

CONTAINER_NAME_BASE = 'unaiverse-sandbox-container'

CONTAINER_NAME module-attribute

CONTAINER_NAME = f'{CONTAINER_NAME_BASE}-{hex[:8]}'

DOCKERFILE_CONTENT module-attribute

DOCKERFILE_CONTENT = '\n\n# Debian image, automatically guessed architecture\nFROM python:3.12-slim-bookworm\n\n# Installing Go compiler\nRUN apt-get update && apt-get install -y --no-install-recommends build-essential curl git\nRUN rm -rf /var/lib/apt/lists/*\nRUN ARCH=$(dpkg --print-architecture) && curl -LO https://go.dev/dl/go1.24.5.linux-${ARCH}.tar.gz\nRUN ARCH=$(dpkg --print-architecture) && tar -C /usr/local -xzf go1.24.5.linux-${ARCH}.tar.gz\nRUN ARCH=$(dpkg --print-architecture) && rm go1.24.5.linux-${ARCH}.tar.gz\n\n# Set Go environment variables\nENV PATH="/usr/local/go/bin:${PATH}"\nENV GOPATH="/go"\nRUN mkdir -p /go/bin /go/src /go/pkg\n\n# Setting the working directory inside the container\nWORKDIR /unaiverse\n\n# Dependencies\nRUN <create_requirements.txt>\nRUN pip install --no-cache-dir -r requirements.txt --break-system-packages\n'

parser module-attribute

parser = ArgumentParser(description='Run a Python script adding customizable read-only and writable paths.', formatter_class=RawTextHelpFormatter, epilog='\n    Examples:\n      python utils/sandbox.py my_script.py -r /home/user/data:/opt/app/data -p 1234\n      python utils/sandbox.py another_script.py -w /tmp/output:/mnt/results\n      python utils/sandbox.py script_with_both.py -r /input:/app/in -w /output:/app/out -p 8082\n    ')

args module-attribute

args = parse_args()

script_to_run module-attribute

script_to_run = script_to_run

read_only_folders module-attribute

read_only_folders = None

writable_folders module-attribute

writable_folders = None

sandbox

sandbox(file_to_run: str, read_only_paths: tuple[str] | list[str] | None = None, writable_paths: tuple[str] | list[str] | None = None) -> None

Run a Python script inside an isolated Docker sandbox.

Builds a fresh Docker image from an embedded Dockerfile (DOCKERFILE_CONTENT), mounts the UNaIVERSE source tree as read-only together with any caller-supplied paths, mounts writable paths, and executes file_to_run with python3 inside the resulting container. Temporary Docker artifacts (the Dockerfile on disk, the image, and the container) are created at the start and removed at the end, and also cleaned up before the build starts so that stale leftovers from a previous failed run cannot interfere.

The UNaIVERSE root directory is always mounted read-only. In addition, three subdirectories (runners/, unaiverse/library/, and unaiverse/networking/p2p/) are always mounted read-write to allow the sandboxed script to write runtime state back to the host. The caller may extend both sets via read_only_paths and writable_paths.

The path layout of this file (.../<pkg>/unaiverse/utils/sandbox.py) is asserted at startup to detect tampering before any Docker command is executed.

Parameters:

Name	Type	Description	Default
file_to_run	str	Absolute or relative path to the Python script to execute inside the container. It is converted to an absolute path before being passed to Docker.	required
read_only_paths	tuple[str] | list[str] | None	Additional host filesystem paths to mount inside the container as read-only. Each path must be an existing directory. Defaults to None, meaning no extra read-only mounts are added.	None
writable_paths	tuple[str] | list[str] | None	Additional host filesystem paths to mount inside the container with read-write access. Each path must be an existing directory. Defaults to None, meaning no extra writable mounts are added.	None

Raises:

Type	Description
AssertionError	If the file's own path does not match the expected ...unaiverse/utils/sandbox.py structure, indicating a potential security issue.
SystemExit	If the Docker image build fails (exit code 1) or if the Docker container run fails (exit code 1).

Source code in unaiverse/utils/sandbox.py

def sandbox(file_to_run: str,
            read_only_paths: tuple[str] | list[str] | None = None,
            writable_paths: tuple[str] | list[str] | None = None) -> None:
    """Run a Python script inside an isolated Docker sandbox.

    Builds a fresh Docker image from an embedded Dockerfile (``DOCKERFILE_CONTENT``),
    mounts the UNaIVERSE source tree as read-only together with any caller-supplied
    paths, mounts writable paths, and executes ``file_to_run`` with ``python3`` inside
    the resulting container. Temporary Docker artifacts (the Dockerfile on disk, the
    image, and the container) are created at the start and removed at the end, and also
    cleaned up before the build starts so that stale leftovers from a previous failed
    run cannot interfere.

    The UNaIVERSE root directory is always mounted read-only. In addition, three
    subdirectories (``runners/``, ``unaiverse/library/``, and
    ``unaiverse/networking/p2p/``) are always mounted read-write to allow the sandboxed
    script to write runtime state back to the host. The caller may extend both sets via
    ``read_only_paths`` and ``writable_paths``.

    The path layout of this file (``.../<pkg>/unaiverse/utils/sandbox.py``) is asserted
    at startup to detect tampering before any Docker command is executed.

    Args:
        file_to_run: Absolute or relative path to the Python script to execute inside
            the container. It is converted to an absolute path before being passed to
            Docker.
        read_only_paths: Additional host filesystem paths to mount inside the container
            as read-only. Each path must be an existing directory. Defaults to None,
            meaning no extra read-only mounts are added.
        writable_paths: Additional host filesystem paths to mount inside the container
            with read-write access. Each path must be an existing directory. Defaults to
            None, meaning no extra writable mounts are added.

    Raises:
        AssertionError: If the file's own path does not match the expected
            ``...unaiverse/utils/sandbox.py`` structure, indicating a potential
            security issue.
        SystemExit: If the Docker image build fails (exit code 1) or if the Docker
            container run fails (exit code 1).
    """

    # Path of this file
    absolute_path_of_this_file = os.path.abspath(__file__)

    # Folders composing the path (and file name at the end)
    path_components = list(Path(absolute_path_of_this_file).parts)

    # Ensuring the folder/file structure was not manipulated
    assert path_components[-1] == 'sandbox.py', "Major security issue, stopping."
    assert path_components[-2] == 'utils', "Major security issue, stopping."
    assert path_components[-3] == 'unaiverse', "Major security issue, stopping."

    # Main folder of UNaIVERSE
    abspath_of_unaiverse_code = str(Path(*path_components[0:-3]))

    # Clean up any remnants from previous runs first (safety)
    cleanup_docker_artifacts(where=abspath_of_unaiverse_code)

    # Requirements
    echoed_contents_of_requirements = 'printf "'
    with open(os.path.join(abspath_of_unaiverse_code, "requirements.txt"), 'r') as req_file:
        req_lines = req_file.readlines()
    for i, req_line in enumerate(req_lines):
        if i != (len(req_lines) - 1) and len(req_line.strip()) > 0:
            echoed_contents_of_requirements += req_line.strip() + "\\n"
        else:
            echoed_contents_of_requirements += req_line.strip() + "\\n\" > requirements.txt"

    # Create Dockerfile
    print("Creating Dockerfile...")
    with open(os.path.join(abspath_of_unaiverse_code, "Dockerfile"), "w") as f:
        f.write(DOCKERFILE_CONTENT.replace('<create_requirements.txt>', echoed_contents_of_requirements))

    # Building Docker image
    if not build_docker_image(where=abspath_of_unaiverse_code):
        print("Exiting due to Docker image build failure")
        cleanup_docker_artifacts(where=abspath_of_unaiverse_code)  # Try to clean up what was created (if any)
        sys.exit(1)

    # Read only folders from the host machine
    read_only_mount_paths = ([abspath_of_unaiverse_code] +
                             (list(read_only_paths) if read_only_paths is not None else []))

    # Writable folders in host machine
    writable_mount_paths = ([os.path.join(abspath_of_unaiverse_code, 'runners'),
                             os.path.join(abspath_of_unaiverse_code, 'unaiverse', 'library'),
                             os.path.join(abspath_of_unaiverse_code, 'unaiverse', 'networking', 'p2p')] +
                            (list(writable_paths) if writable_paths is not None else []))

    # Running
    if not run_in_docker(file_to_run=os.path.abspath(file_to_run),
                         read_only_host_paths=read_only_mount_paths,
                         writable_host_paths=writable_mount_paths):
        print("Exiting due to Docker container run failure")
        sys.exit(1)

    # Final cleanup
    cleanup_docker_artifacts(where=abspath_of_unaiverse_code)

build_docker_image

build_docker_image(where: str)

Build the Docker image from the Dockerfile located in where.

Invokes docker build -t <DOCKER_IMAGE_NAME> <where> as a subprocess and waits for it to complete. Progress and errors from the Docker build are printed to stdout by the subprocess itself.

Parameters:

Name	Type	Description	Default
where	str	Absolute path to the directory containing the Dockerfile. This is typically the UNaIVERSE source root, where the Dockerfile is written by sandbox before this function is called.	required

Returns:

Type	Description
	True if the build completes successfully, False if
	subprocess.CalledProcessError is raised (i.e., if Docker exits with a
	non-zero status code).

Source code in unaiverse/utils/sandbox.py

def build_docker_image(where: str):
    """Build the Docker image from the Dockerfile located in ``where``.

    Invokes ``docker build -t <DOCKER_IMAGE_NAME> <where>`` as a subprocess and waits
    for it to complete. Progress and errors from the Docker build are printed to stdout
    by the subprocess itself.

    Args:
        where: Absolute path to the directory containing the Dockerfile. This is
            typically the UNaIVERSE source root, where the Dockerfile is written by
            ``sandbox`` before this function is called.

    Returns:
        ``True`` if the build completes successfully, ``False`` if
        ``subprocess.CalledProcessError`` is raised (i.e., if Docker exits with a
        non-zero status code).
    """
    print(f"Building Docker image '{DOCKER_IMAGE_NAME}'...")

    try:

        # The '.' at the end means build from the current directory
        subprocess.run(["docker", "build", "-t", DOCKER_IMAGE_NAME, where], check=True)
        print(f"Docker image '{DOCKER_IMAGE_NAME}' built successfully.")
        return True
    except subprocess.CalledProcessError as e:
        print(f"Error building Docker image: {e}")
        return False

cleanup_docker_artifacts

cleanup_docker_artifacts(where: str)

Remove the generated Dockerfile, Docker image, and any running container.

Attempts, in order: stopping and removing the container whose name matches CONTAINER_NAME, removing the DOCKER_IMAGE_NAME image, and deleting the Dockerfile from where. Each step is performed independently so that a failure in one step does not prevent the others from running. Errors encountered while stopping or removing the container are caught and printed rather than raised. Errors raised by docker rmi (a subprocess.CalledProcessError) are also caught and printed.

This function is called both before the build (to remove stale artifacts from a previously interrupted run) and after a successful or failed run (final cleanup).

Parameters:

Name	Type	Description	Default
where	str	Absolute path to the directory from which the Dockerfile should be removed. This is typically the UNaIVERSE source root.	required

Source code in unaiverse/utils/sandbox.py

def cleanup_docker_artifacts(where: str):
    """Remove the generated Dockerfile, Docker image, and any running container.

    Attempts, in order: stopping and removing the container whose name matches
    ``CONTAINER_NAME``, removing the ``DOCKER_IMAGE_NAME`` image, and deleting the
    ``Dockerfile`` from ``where``. Each step is performed independently so that a
    failure in one step does not prevent the others from running. Errors encountered
    while stopping or removing the container are caught and printed rather than raised.
    Errors raised by ``docker rmi`` (a ``subprocess.CalledProcessError``) are also
    caught and printed.

    This function is called both before the build (to remove stale artifacts from a
    previously interrupted run) and after a successful or failed run (final cleanup).

    Args:
        where: Absolute path to the directory from which the Dockerfile should be
            removed. This is typically the UNaIVERSE source root.
    """
    print("Cleaning...")

    # Stop and remove container if it's still running (e.g., if previous run failed)
    try:
        print(f"Attempting to stop and remove container '{CONTAINER_NAME}' (if running)...")
        subprocess.run(["docker", "stop", CONTAINER_NAME],
                       check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        subprocess.run(["docker", "rm", CONTAINER_NAME],
                       check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    except Exception as e:
        print(f"Error during preliminary container cleanup: {e}")

    # Remove the Docker image
    try:
        print(f"Removing Docker image '{DOCKER_IMAGE_NAME}'...")
        subprocess.run(["docker", "rmi", DOCKER_IMAGE_NAME], check=True)
        print("Docker image removed.")
    except subprocess.CalledProcessError as e:
        print(f"Error removing Docker image '{DOCKER_IMAGE_NAME}': {e}")

    # Remove the generated Dockerfile
    if os.path.exists(os.path.join(where, "Dockerfile")):
        os.remove(os.path.join(where, "Dockerfile"))
        print("Removed Dockerfile.")

run_in_docker

run_in_docker(file_to_run: str, read_only_host_paths: list[str] = None, writable_host_paths: list[str] = None)

Run a Python script inside a Docker container with configurable volume mounts.

Builds a docker run command from the supplied parameters and executes it. The container is started with --rm so it is automatically removed on exit. PYTHONUNBUFFERED=1 and NODE_STARTING_PORT are forwarded from the host environment so the subprocess behaves identically to a direct invocation.

Network configuration is platform-sensitive: on Linux, --net host is used so the container shares the host network stack directly. On other platforms (macOS, Windows), the four consecutive ports starting at NODE_STARTING_PORT are published explicitly (TCP for the even offsets, UDP for the odd ones).

Every path in read_only_host_paths is mounted at the same absolute location inside the container with :ro. Every path in writable_host_paths is mounted at the same absolute location with read-write access. A path that does not exist or is not a directory causes the function to return False immediately without starting the container.

The container's stdout and stderr are streamed line-by-line to the caller's stdout in real time. A KeyboardInterrupt during streaming is silently swallowed so the caller can handle it at a higher level.

Parameters:

Name	Type	Description	Default
file_to_run	str	Absolute path to the Python script to execute inside the container. It is passed directly as the argument to python3.	required
read_only_host_paths	list[str]	Host filesystem paths to mount read-only. Each path must be an existing directory; the mount point inside the container mirrors the host path exactly. Defaults to None.	None
writable_host_paths	list[str]	Host filesystem paths to mount with read-write access. Each path must be an existing directory; the mount point inside the container mirrors the host path exactly. Defaults to None.	None

Returns:

Type	Description
	True if the container starts and finishes (regardless of the script's exit
	code), False if Docker is not found (FileNotFoundError), if Docker
	returns a non-zero status (subprocess.CalledProcessError), or if a
	required host path does not exist.

Source code in unaiverse/utils/sandbox.py

def run_in_docker(file_to_run: str, read_only_host_paths: list[str] = None,
                  writable_host_paths: list[str] = None):
    """Run a Python script inside a Docker container with configurable volume mounts.

    Builds a ``docker run`` command from the supplied parameters and executes it. The
    container is started with ``--rm`` so it is automatically removed on exit.
    ``PYTHONUNBUFFERED=1`` and ``NODE_STARTING_PORT`` are forwarded from the host
    environment so the subprocess behaves identically to a direct invocation.

    Network configuration is platform-sensitive: on Linux, ``--net host`` is used so
    the container shares the host network stack directly. On other platforms (macOS,
    Windows), the four consecutive ports starting at ``NODE_STARTING_PORT`` are
    published explicitly (TCP for the even offsets, UDP for the odd ones).

    Every path in ``read_only_host_paths`` is mounted at the same absolute location
    inside the container with ``:ro``. Every path in ``writable_host_paths`` is mounted
    at the same absolute location with read-write access. A path that does not exist or
    is not a directory causes the function to return ``False`` immediately without
    starting the container.

    The container's stdout and stderr are streamed line-by-line to the caller's stdout
    in real time. A ``KeyboardInterrupt`` during streaming is silently swallowed so the
    caller can handle it at a higher level.

    Args:
        file_to_run: Absolute path to the Python script to execute inside the
            container. It is passed directly as the argument to ``python3``.
        read_only_host_paths: Host filesystem paths to mount read-only. Each path must
            be an existing directory; the mount point inside the container mirrors the
            host path exactly. Defaults to None.
        writable_host_paths: Host filesystem paths to mount with read-write access.
            Each path must be an existing directory; the mount point inside the
            container mirrors the host path exactly. Defaults to None.

    Returns:
        ``True`` if the container starts and finishes (regardless of the script's exit
        code), ``False`` if Docker is not found (``FileNotFoundError``), if Docker
        returns a non-zero status (``subprocess.CalledProcessError``), or if a
        required host path does not exist.
    """
    print(f"\nRunning code in Docker container '{CONTAINER_NAME}'...")

    # Building command (it will continue below...)
    command = ["docker", "run",
               "--rm",  # Automatically remove the container when it exits
               "-e", "PYTHONUNBUFFERED=1",  # Ensure Python output is unbuffered
               "-e", "NODE_STARTING_PORT",
               "--name", CONTAINER_NAME]

    if sys.platform.startswith('linux'):

        # Linux
        command.extend(["--net", "host"]),  # Expose the host network (in macOS and Windows it is still a virtual host)
    else:

        # Not-linux: check ports (adding -p port:port)
        port_int = int(os.getenv("NODE_STARTING_PORT", "0"))
        if port_int > 0:
            command.extend(["-p", str(port_int + 0) + ":" + str(port_int + 0)])
            command.extend(["-p", str(port_int + 1) + ":" + str(port_int + 1) + "/udp"])
            command.extend(["-p", str(port_int + 2) + ":" + str(port_int + 2)])
            command.extend(["-p", str(port_int + 3) + ":" + str(port_int + 3) + "/udp"])

    # Add read-only mount if path is provided
    if read_only_host_paths is not None and len(read_only_host_paths) > 0:
        for path in read_only_host_paths:

            # Ensure the host path exists and is a directory
            if not os.path.isdir(path):
                print(
                    f"Error: Read-only host path '{path}' does not exist or is not a directory. Cannot mount.")
                return False
            else:

                # Augmenting command
                path = os.path.abspath(path)
                command.extend(["-v", f"{path}:{path}:ro"])
                print(f"Mounted host '{path}' as read-only to container")

    # Add writable mount if path is provided
    if writable_host_paths is not None and len(writable_host_paths) > 0:
        for path in writable_host_paths:

            # Ensure the host path exists and is a directory
            if not os.path.isdir(path):
                print(
                    f"Error: Writable host path '{path}' does not exist or is not a directory. Cannot mount.")
                return False
            else:

                # Augmenting command
                path = os.path.abspath(path)
                command.extend(["-v", f"{path}:{path}"])
                print(f"Mounted host '{path}' as writable to container")

    # Completing command
    command.append(DOCKER_IMAGE_NAME)

    try:

        # Running the prepared command... (using Popen to stream output in real-time)
        try:
            command.extend(["python3", file_to_run])
            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
            for line in iter(process.stdout.readline, ''):
                sys.stdout.write(line)
            process.wait()  # Wait for the process to finish
            if process.returncode != 0:
                print(f"Container exited with non-zero status code: {process.returncode}")
        except KeyboardInterrupt:
            pass

        print(f"\nContainer '{CONTAINER_NAME}' finished execution.")
        return True
    except FileNotFoundError:
        print("Error: Docker command not found. Is Docker installed and in your PATH?")
        print("Please ensure Docker is installed and running.")
        return False
    except subprocess.CalledProcessError as e:
        print(f"Error running Docker container: {e}")
        return False