EvoScientist-Multi/tests/test_native_sandbox.py

from __future__ import annotations

import json
import sys
import types
from pathlib import Path

import pytest

import EvoScientist.native_sandbox as sandbox


def _installation(tmp_path: Path) -> sandbox.NativeSandboxInstallation:
    package = tmp_path / "package"
    package.mkdir()
    srt = tmp_path / "srt"
    srt.touch(mode=0o700)
    return sandbox.NativeSandboxInstallation(
        srt=srt,
        package_root=package,
        path_env="/usr/bin:/bin",
        system_read_paths=("/usr", "/bin", "/dev/null"),
    )


def test_existing_read_paths_resolve_and_deduplicate_symlink_aliases(tmp_path: Path):
    usr = tmp_path / "usr"
    usr.mkdir()
    bin_alias = tmp_path / "bin"
    bin_alias.symlink_to(usr, target_is_directory=True)
    missing = tmp_path / "missing"

    paths = sandbox._existing_resolved_paths((str(usr), str(bin_alias), str(missing)))

    assert paths == (str(usr.resolve()),)


def test_policy_denies_root_and_only_writes_scope_and_command_tmp(tmp_path: Path):
    files = tmp_path / "files"
    command_tmp = tmp_path / "runtime" / "tmp" / "run"
    files.mkdir()
    command_tmp.mkdir(parents=True)

    policy = sandbox._sandbox_settings(_installation(tmp_path), files, command_tmp)

    assert policy["filesystem"]["denyRead"] == ["/"]
    assert str(files) in policy["filesystem"]["allowRead"]
    assert str(command_tmp) in policy["filesystem"]["allowRead"]
    assert policy["filesystem"]["allowWrite"] == [
        str(files),
        str(command_tmp),
        "/dev/null",
    ]
    assert policy["filesystem"]["denyWrite"] == [
        str(files / "uploads"),
        "/tmp/claude",
        "/private/tmp/claude",
        "/dev/tty",
        "/dev/dtracehelper",
        "/dev/autofs_nowait",
    ]
    assert policy["network"]["allowedDomains"] == []
    assert policy["network"]["allowAllUnixSockets"] is False
    assert policy["allowAppleEvents"] is False
    assert "control" not in json.dumps(policy)


def test_weaker_nested_mode_requires_explicit_environment_opt_in(tmp_path: Path, monkeypatch):
    monkeypatch.delenv("EVOSCIENTIST_NATIVE_SANDBOX_WEAKER_NESTED", raising=False)
    files = tmp_path / "files"
    command_tmp = tmp_path / "runtime" / "tmp" / "run"
    files.mkdir()
    command_tmp.mkdir(parents=True)
    installation = _installation(tmp_path)

    assert sandbox._sandbox_settings(installation, files, command_tmp)[
        "enableWeakerNestedSandbox"
    ] is False

    monkeypatch.setenv("EVOSCIENTIST_NATIVE_SANDBOX_WEAKER_NESTED", "true")
    assert sandbox._sandbox_settings(installation, files, command_tmp)[
        "enableWeakerNestedSandbox"
    ] is True


def test_network_preflight_probe_accepts_kernel_denied_unix_socket(monkeypatch):
    monkeypatch.delenv("OPENAI_API_KEY", raising=False)

    class FakeSocket:
        def __init__(self, family=None, *_args):
            if family == 1:
                raise PermissionError("blocked by seccomp")

        def connect_ex(self, _address):
            return 1

        def close(self):
            return None

    fake_socket = types.SimpleNamespace(
        AF_UNIX=1,
        socket=lambda family=None, *args: FakeSocket(family, *args),
    )
    monkeypatch.setitem(sys.modules, "socket", fake_socket)

    namespace: dict[str, object] = {}
    exec(sandbox._network_preflight_probe(1234, Path("/blocked.sock")), namespace)


def test_clean_environment_does_not_inherit_secrets(tmp_path: Path, monkeypatch):
    command_tmp = tmp_path / "tmp"
    (command_tmp / "home").mkdir(parents=True)
    (command_tmp / "tmp").mkdir()
    monkeypatch.setenv("OPENAI_API_KEY", "secret")

    environment = sandbox._clean_environment(_installation(tmp_path), command_tmp)

    assert set(environment) == {"PATH", "HOME", "TMPDIR", "WORKSPACE", "LANG", "LC_ALL"}
    assert "OPENAI_API_KEY" not in environment
    assert environment["WORKSPACE"] == "."


def test_executor_requires_control_directory_outside_files(tmp_path: Path):
    files = tmp_path / "files"
    files.mkdir()
    runtime = files / "runtime"
    runtime.mkdir()

    with pytest.raises(sandbox.NativeSandboxUnavailable):
        sandbox.NativeSandboxExecutor(files, runtime)


def test_readiness_is_cached_and_failure_is_fail_closed(monkeypatch):
    sandbox._reset_native_sandbox_readiness_for_tests()
    calls = {"install": 0, "preflight": 0}

    def install():
        calls["install"] += 1
        return object()

    def preflight(_installation):
        calls["preflight"] += 1

    monkeypatch.setattr(sandbox, "_assert_install_contract", install)
    monkeypatch.setattr(sandbox, "_run_preflight", preflight)
    sandbox.ensure_native_sandbox_ready()
    sandbox.ensure_native_sandbox_ready()
    assert calls == {"install": 1, "preflight": 1}

    sandbox._reset_native_sandbox_readiness_for_tests()
    monkeypatch.setattr(
        sandbox,
        "_run_preflight",
        lambda _installation: (_ for _ in ()).throw(
            sandbox.NativeSandboxUnavailable("failed once")
        ),
    )
    with pytest.raises(sandbox.NativeSandboxUnavailable, match="failed once"):
        sandbox.ensure_native_sandbox_ready()
    with pytest.raises(sandbox.NativeSandboxUnavailable, match="failed once"):
        sandbox.ensure_native_sandbox_ready()
    assert calls["install"] == 2

    sandbox._reset_native_sandbox_readiness_for_tests()