sand-boxer/tests/test_messages_route.py

"""Trusted binding refuses authority expansion before provisioning."""

import socket
from dataclasses import replace
from types import SimpleNamespace

import pytest

from sandboxer.extensions.bwrap import BwrapExtension
from sandboxer.extensions.messages_route import OwnerMessagesRoute
from sandboxer.models import Consumer
from sandboxer.profiles.loader import load_profile


@pytest.fixture
def binding(tmp_path):
    private = tmp_path / "private"
    private.mkdir(mode=0o700)
    path = private / "route.sock"
    with socket.socket(socket.AF_UNIX) as listener:
        listener.bind(str(path))
        path.chmod(0o600)
        yield OwnerMessagesRoute(path, "a" * 43, "profile.bwrap-local", "agt", "fixture", "run-1")


def test_exact_route_has_only_fixed_socket_mount(binding, tmp_path):
    backend = BwrapExtension({"base_dir": str(tmp_path / "workspaces")})
    profile = load_profile("profile.bwrap-local")
    consumer = Consumer(actor="agt", project="fixture", run_id="run-1")
    binding.validate(profile, consumer, backend, {})
    backend.messages_route = binding
    argv = backend._bwrap_argv(str(tmp_path / "workspaces" / "one"))
    assert "--unshare-net" in argv and "--messages" in argv
    assert str(binding.socket_path) in argv and binding.token not in argv
    assert "--egress" not in argv
    assert binding.token not in repr(binding)


@pytest.mark.parametrize("change", ["actor", "project", "run_id", "profile", "network",
                                    "allowlist", "credentials", "mount", "private", "backend"])
def test_route_refuses_expansion(binding, tmp_path, change):
    backend = BwrapExtension({"base_dir": str(tmp_path / "workspaces")})
    profile = load_profile("profile.bwrap-local")
    consumer = Consumer(actor="agt", project="fixture", run_id="run-1")
    if change in {"actor", "project", "run_id"}:
        consumer = consumer.model_copy(update={change: "adm" if change == "actor" else "other"})
        if change == "actor":
            consumer = Consumer(actor="adm", project="fixture", run_id="run-1")
    elif change == "profile":
        binding = replace(binding, profile_id="profile.other")
    elif change == "network":
        profile.network.egress = ["api.anthropic.com:443"]
    elif change == "allowlist":
        backend.config["allowed_egress"] = ["api.anthropic.com:443"]
    elif change == "credentials":
        backend.config["credential_routes"] = {"route": {}}
    elif change == "mount":
        backend.ro_binds.append("/home")
    elif change == "private":
        backend.base_dir = str(binding.socket_path.parent)
    elif change == "backend":
        backend = SimpleNamespace()
    with pytest.raises(ValueError):
        binding.validate(profile, consumer, backend, {})


def test_trusted_runtime_pin_is_injected_and_overlap_refused(binding, tmp_path):
    runtime = tmp_path / "runtime"
    binding = replace(binding, runtime_path=runtime, runtime_sha256="b" * 64)
    backend = BwrapExtension({"base_dir": str(tmp_path / "workspaces")})
    profile = load_profile("profile.bwrap-local")
    consumer = Consumer(actor="agt", project="fixture", run_id="run-1")
    binding.validate(profile, consumer, backend, {})
    assert backend.config["runtime"] == {"path": str(runtime), "sha256": "b" * 64}
    with pytest.raises(ValueError, match="overlaps"):
        replace(binding, runtime_path=binding.socket_path.parent).validate(
            profile, consumer, backend, {}
        )
    with pytest.raises(ValueError, match="exact path"):
        replace(binding, runtime_sha256=None).validate(profile, consumer, backend, {})