codenib.sandbox

Provider-neutral isolation for repository execution.

Sandboxing is opt-in. Importing this package never contacts a container runtime; construct :class:DockerSandboxProvider explicitly on a dedicated worker when untrusted code execution is required.

Modules:

Name	Description
docker	Rootless-Docker sandbox backend for untrusted repository commands.
protocol	Backend-neutral sandbox provider and session protocols.
types	Stable value types for isolated repository execution.

Classes:

Name	Description
DockerSandboxProvider	Create digest-pinned, rootless-Docker sandbox sessions.
DockerSandboxSession	A copied repository backed by private baseline/workspace volumes.
SandboxClosedError	Raised when an operation targets a closed session.
SandboxError	Base error for sandbox infrastructure or policy failures.
SandboxPolicyError	Raised when a request would weaken an enforced policy.
SandboxProvider	Factory for backend-specific sandbox sessions.
SandboxSession	One isolated, copied repository workspace.
SandboxUnavailableError	Raised when the configured runtime or image is unavailable.
ArtifactBundle	Controller-owned ZIP export of selected workspace files.
ArtifactMember	One regular file included in an exported artifact bundle.
DiffResult	Canonical Git patch produced from the immutable source snapshot.
ExecRequest	One argv-based command request.
ExecResult	Bounded model-facing output plus audit hashes for one command.
NetworkMode	Container egress policy.
SandboxCapabilities	Auditable guarantees a provider can truthfully claim.
SandboxLimits	Hard resource and output bounds applied to every command.
SandboxMetadata	Non-secret identity recorded in agent traces and job artifacts.
SandboxPolicy	Security policy for one sandbox session.
SandboxSpec	Immutable request for a repository sandbox.

DockerSandboxProvider

DockerSandboxProvider(
    *,
    allowed_images: Collection[str],
    docker_binary: str = "docker",
    docker_host: str | None = None,
    git_binary: str = "git",
    work_root: Path | None = None,
    retain_audit_logs: bool = False,
    run_cli: RunCLI = run,
    popen_factory: PopenFactory = Popen,
    clock: Callable[[], float] = monotonic,
    id_factory: Callable[[], UUID] = uuid4
)

Create digest-pinned, rootless-Docker sandbox sessions.

Parameters:

Name	Type	Description	Default
allowed_images	Collection[str]	Exact digest-pinned image references approved by the service control plane. The model never selects an arbitrary image.	required
docker_binary	str	Docker-compatible client binary. This backend validates Docker semantics and must not be pointed at Podman.	'docker'
work_root	Path | None	Controller-only directory for bounded audit logs.	None
retain_audit_logs	bool	Keep audit logs after session close. Production bot workers should set this and apply their own retention policy.	False

Methods:

Name	Description
create	Preflight runtime/image policy, copy the source, and return a session.

Source code in codenib/sandbox/docker.py

def __init__(
    self,
    *,
    allowed_images: Collection[str],
    docker_binary: str = "docker",
    docker_host: str | None = None,
    git_binary: str = "git",
    work_root: Path | None = None,
    retain_audit_logs: bool = False,
    run_cli: RunCLI = subprocess.run,
    popen_factory: PopenFactory = subprocess.Popen,
    clock: Callable[[], float] = time.monotonic,
    id_factory: Callable[[], uuid.UUID] = uuid.uuid4,
) -> None:
    images = frozenset(allowed_images)
    if not images:
        raise ValueError("allowed_images must contain at least one pinned image")
    self._allowed_images = images
    self._docker_binary_request = docker_binary
    self._docker_binary: str | None = None
    self._git_binary_request = git_binary
    self._git_binary: str | None = None
    self._docker_host = docker_host or f"unix:///run/user/{os.getuid()}/docker.sock"
    self._validate_docker_host(self._docker_host)
    # Every client call uses the same explicit endpoint, absolute binary,
    # and minimal environment. Host Docker contexts/config and credentials
    # cannot redirect a preflighted call or reach a registry helper.
    self._docker_env = {
        "DOCKER_CONFIG": "/nonexistent/codenib-docker-config",
        "DOCKER_CONTEXT": "",
        "DOCKER_HOST": self._docker_host,
        "HOME": "/nonexistent",
        "LANG": "C.UTF-8",
        "PATH": "/usr/bin:/bin",
    }
    self._work_root = Path(work_root) if work_root is not None else None
    self._retain_audit_logs = retain_audit_logs
    self._run_cli = run_cli
    self._popen_factory = popen_factory
    self._clock = clock
    self._id_factory = id_factory

create

create(spec: SandboxSpec) -> 'DockerSandboxSession'

Preflight runtime/image policy, copy the source, and return a session.

Source code in codenib/sandbox/docker.py

def create(self, spec: SandboxSpec) -> "DockerSandboxSession":
    """Preflight runtime/image policy, copy the source, and return a session."""

    if spec.image not in self._allowed_images:
        raise SandboxPolicyError("image is not in the provider allowlist")
    if spec.source_revision is not None:
        self._resolve_git_binary()
    source_request = (
        spec.source_dir
        if spec.source_dir.is_absolute()
        else Path.cwd() / spec.source_dir
    )
    self._validate_host_mount_path(source_request)
    if not source_request.is_dir():
        raise SandboxPolicyError(
            f"source directory does not exist: {spec.source_dir}"
        )
    source = source_request.resolve(strict=True)
    if source == Path(source.anchor):
        raise SandboxPolicyError("filesystem root cannot be a sandbox source")
    if self._work_root is not None:
        audit_root = self._work_root.expanduser().resolve()
        if audit_root == source or source in audit_root.parents:
            raise SandboxPolicyError(
                "controller audit root must be outside the repository source"
            )
    git_common_dir = self._verify_source_revision(source, spec.source_revision)

    runtime = self._inspect_runtime(spec)
    image = self._inspect_image(spec)
    sandbox_token = self._id_factory().hex
    sandbox_id = f"sbx-{sandbox_token}"
    self._verify_runtime_limits(spec, sandbox_id)
    task_id = spec.task_id or sandbox_id
    baseline_volume = f"codenib-{sandbox_token}-base"
    workspace_volume = f"codenib-{sandbox_token}-work"
    audit_dir = self._create_audit_dir(sandbox_token)
    created_volumes: list[str] = []

    try:
        created_volumes.append(baseline_volume)
        self._create_volume(baseline_volume, sandbox_id, "baseline")
        created_volumes.append(workspace_volume)
        self._create_volume(workspace_volume, sandbox_id, "workspace")
        self._populate_baseline(
            spec,
            sandbox_id=sandbox_id,
            source=source,
            git_common_dir=git_common_dir,
            baseline_volume=baseline_volume,
        )
        source_identity: _SourceIdentity = (
            self._fingerprint_baseline(
                spec,
                sandbox_id=sandbox_id,
                baseline_volume=baseline_volume,
            )
            if spec.source_revision is not None
            else {"source_fingerprint": "", "file_count": 0}
        )
        # Detect a controller checkout changing during the snapshot.
        self._verify_source_revision(source, spec.source_revision)
        self._populate_workspace(
            spec,
            sandbox_id=sandbox_id,
            baseline_volume=baseline_volume,
            workspace_volume=workspace_volume,
        )

        metadata = SandboxMetadata(
            sandbox_id=sandbox_id,
            task_id=task_id,
            provider="docker",
            image=spec.image,
            image_id=str(image["Id"]),
            platform=spec.platform,
            source_revision=spec.source_revision,
            network=spec.policy.network.value,
            rootless_runtime=runtime["rootless"],
            source_fingerprint=source_identity["source_fingerprint"],
            policy=_policy_summary(spec),
        )
        session = DockerSandboxSession(
            provider=self,
            spec=spec,
            metadata=metadata,
            baseline_volume=baseline_volume,
            workspace_volume=workspace_volume,
            audit_dir=audit_dir,
        )
        session._record_audit(
            "sandbox_created",
            image_id=metadata.image_id,
            source_revision=metadata.source_revision,
            source_fingerprint=metadata.source_fingerprint,
            source_file_count=source_identity["file_count"],
            platform=metadata.platform,
            network=metadata.network,
            rootless_runtime=metadata.rootless_runtime,
            policy=metadata.policy,
            capabilities=asdict(session.capabilities),
        )
    except Exception as exc:
        failed_cleanup = [
            volume
            for volume in reversed(created_volumes)
            if not self._remove_volume(volume)
        ]
        if failed_cleanup:
            _write_emergency_audit(
                audit_dir,
                event="bootstrap_cleanup_failed",
                sandbox_id=sandbox_id,
                volumes=failed_cleanup,
            )
            raise SandboxUnavailableError(
                "sandbox bootstrap cleanup failed; quarantine and reap the "
                "worker before accepting another job"
            ) from exc
        self._remove_audit_dir(audit_dir)
        raise
    return session

DockerSandboxSession

DockerSandboxSession(
    *,
    provider: DockerSandboxProvider,
    spec: SandboxSpec,
    metadata: SandboxMetadata,
    baseline_volume: str,
    workspace_volume: str,
    audit_dir: Path
)

A copied repository backed by private baseline/workspace volumes.

Attributes:

Name	Type	Description
audit_dir	Path	Controller-only audit directory; never mounted into the sandbox.

Source code in codenib/sandbox/docker.py

def __init__(
    self,
    *,
    provider: DockerSandboxProvider,
    spec: SandboxSpec,
    metadata: SandboxMetadata,
    baseline_volume: str,
    workspace_volume: str,
    audit_dir: Path,
) -> None:
    self._provider = provider
    self._spec = spec
    self._metadata = metadata
    self._baseline_volume = baseline_volume
    self._workspace_volume = workspace_volume
    self._audit_dir = audit_dir
    self._closed = False
    self._poisoned = False
    self._lock = threading.RLock()

audit_dir property

audit_dir: Path

Controller-only audit directory; never mounted into the sandbox.

SandboxClosedError

Bases: SandboxError

Raised when an operation targets a closed session.

SandboxError

Bases: RuntimeError

Base error for sandbox infrastructure or policy failures.

SandboxPolicyError

Bases: SandboxError

Raised when a request would weaken an enforced policy.

SandboxProvider

Bases: Protocol

Factory for backend-specific sandbox sessions.

SandboxSession

Bases: Protocol

One isolated, copied repository workspace.

SandboxUnavailableError

Bases: SandboxError

Raised when the configured runtime or image is unavailable.

ArtifactBundle dataclass

ArtifactBundle(path: Path, size: int, sha256: str, members: tuple[ArtifactMember, ...])

Controller-owned ZIP export of selected workspace files.

ArtifactMember dataclass

ArtifactMember(path: str, size: int, sha256: str)

One regular file included in an exported artifact bundle.

DiffResult dataclass

DiffResult(patch: str, sha256: str, bytes: int, truncated: bool = False)

Canonical Git patch produced from the immutable source snapshot.

ExecRequest dataclass

ExecRequest(
    argv: Tuple[str, ...] | Sequence[str],
    cwd: PurePosixPath | str = PurePosixPath("."),
    stdin: bytes | None = None,
    environment: Mapping[str, str] = dict(),
    timeout_seconds: float | None = None,
)

One argv-based command request.

No host shell is involved. Callers that deliberately need shell syntax can request ('/bin/sh', '-lc', command); those three argv elements are still passed after the pinned container image.

ExecResult dataclass

ExecResult(
    command_id: str,
    argv: tuple[str, ...],
    exit_code: int | None,
    stdout: str,
    stderr: str,
    duration_ms: float,
    timed_out: bool = False,
    output_limited: bool = False,
    stdout_truncated: bool = False,
    stderr_truncated: bool = False,
    stdout_sha256: str = "",
    stderr_sha256: str = "",
    stdout_bytes: int = 0,
    stderr_bytes: int = 0,
)

Bounded model-facing output plus audit hashes for one command.

NetworkMode

Bases: str, Enum

Container egress policy.

BRIDGE is intentionally explicit. Docker cannot implement a reliable hostname allowlist by itself, so public issue-bot jobs should remain on NONE and use a separate, policy-enforcing bootstrap service when they need dependencies.

SandboxCapabilities dataclass

SandboxCapabilities(
    provider: str,
    isolation: str,
    non_root_user: bool,
    network_isolation: bool,
    read_only_rootfs: bool,
    resource_limits: bool,
    process_tree_cleanup: bool,
    disk_quota: bool,
    rootless_runtime: bool | None = None,
)

Auditable guarantees a provider can truthfully claim.

SandboxLimits dataclass

SandboxLimits(
    cpus: float = 2.0,
    memory_bytes: int = 2 * 1024**3,
    pids: int = 256,
    command_timeout_seconds: float = 300.0,
    output_bytes: int = 64 * 1024,
    audit_log_bytes: int = 8 * 1024**2,
    stdin_bytes: int = 1024**2,
    tmpfs_bytes: int = 256 * 1024**2,
    artifact_bytes: int = 64 * 1024**2,
)

Hard resource and output bounds applied to every command.

SandboxMetadata dataclass

SandboxMetadata(
    sandbox_id: str,
    task_id: str,
    provider: str,
    image: str,
    image_id: str,
    platform: str,
    source_revision: str | None,
    network: str,
    rootless_runtime: bool | None,
    source_fingerprint: str = "",
    policy: Mapping[str, object] = dict(),
)

Non-secret identity recorded in agent traces and job artifacts.

SandboxPolicy dataclass

SandboxPolicy(
    network: NetworkMode = NONE,
    read_only_rootfs: bool = True,
    require_rootless_runtime: bool = True,
    require_source_revision: bool = True,
    allow_unpinned_image: bool = False,
    seccomp_profile: str | None = None,
    runtime: str | None = None,
    limits: SandboxLimits = SandboxLimits(),
)

Security policy for one sandbox session.

The safe defaults are fail-closed: no network, a read-only container root, no Linux capabilities, no privilege escalation, and a rootless daemon requirement. require_rootless_runtime=False is intended only for explicitly trusted repositories or a separately isolated worker VM.

SandboxSpec dataclass

SandboxSpec(
    source_dir: Path,
    image: str,
    platform: str,
    source_revision: str | None = None,
    task_id: str | None = None,
    policy: SandboxPolicy = SandboxPolicy(),
    source_selection: RepositorySourceSelection = RepositorySourceSelection(),
)

Immutable request for a repository sandbox.

source_revision is optional at the library boundary for generated or non-Git fixtures. GitHub automation should always provide the exact 40-character commit and let the provider verify it before copying files.