sand-boxer/tests/test_egress.py

import socket
from unittest.mock import patch

import pytest

from sandboxer.extensions.egress import connect_public, destinations, tunnel


@pytest.mark.parametrize(
    "entry",
    [
        "*",
        "api.anthropic.com",
        "api.anthropic.com:80",
        "127.0.0.1:443",
        "API.anthropic.com:443",
        "api.anthropic.com.evil:443/path",
        "x@:443",
    ],
)
def test_invalid_destination(entry):
    with pytest.raises(ValueError):
        destinations([entry])


@pytest.mark.parametrize(
    "address",
    ["127.0.0.1", "10.0.0.1", "169.254.169.254", "::1", "::ffff:127.0.0.1", "224.0.0.1", "ff02::1"],
)
def test_nonpublic_dns_refused(address):
    with (
        patch("socket.getaddrinfo", return_value=[(socket.AF_INET, 1, 6, "", (address, 443))]),
        patch("socket.socket") as factory,
        pytest.raises(ValueError),
    ):
        connect_public("api.anthropic.com")
    factory.assert_not_called()


@pytest.mark.parametrize(
    "wire_request",
    [
        b"CONNECT evil.example:443 HTTP/1.1\r\n\r\n",
        b"CONNECT api.anthropic.com:80 HTTP/1.1\r\n\r\n",
        b"GET https://api.anthropic.com/ HTTP/1.1\r\n\r\n",
        b"CONNECT api.anthropic.com:443 HTTP/1.1\r\nHost: evil.example:443\r\n\r\n",
        b"CONNECT api.anthropic.com:443 HTTP/1.1\r\nContent-Length: 1\r\n\r\n",
    ],
)
def test_denied_connect_never_dials(wire_request):
    left, right = socket.socketpair()
    with left, right, patch("sandboxer.extensions.egress.connect_public") as connect:
        left.sendall(wire_request)
        with pytest.raises(ValueError):
            tunnel(right, destinations(["api.anthropic.com:443"]))
        connect.assert_not_called()


def test_valid_connect_preserves_tls_bytes():
    left, right = socket.socketpair()
    upstream, peer = socket.socketpair()
    with left, right, upstream, peer:
        left.sendall(
            b"CONNECT api.anthropic.com:443 HTTP/1.1\r\nHost: api.anthropic.com:443\r\n\r\nTLS"
        )
        with (
            patch("sandboxer.extensions.egress.connect_public", return_value=upstream) as connect,
            patch("sandboxer.extensions.egress.relay") as relay,
        ):
            tunnel(right, destinations(["api.anthropic.com:443"]))
            connect.assert_called_once_with("api.anthropic.com")
            relay.assert_called_once_with(right, upstream)
            assert right.recv(3) == b"TLS"
            assert b"200 Connection Established" in left.recv(100)


def test_dns_result_is_used_without_second_resolution():
    with (
        patch(
            "socket.getaddrinfo",
            return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("1.1.1.1", 443))],
        ) as dns,
        patch("socket.socket") as factory,
    ):
        assert connect_public("api.anthropic.com") is factory.return_value
        dns.assert_called_once()
        factory.return_value.connect.assert_called_once_with(("1.1.1.1", 443))


def test_profile_cannot_expand_owner_allowlist(tmp_path):
    from sandboxer.extensions.bwrap import BwrapExtension
    from sandboxer.models import Profile

    ext = BwrapExtension(
        {"base_dir": str(tmp_path / "unused"), "allowed_egress": ["api.anthropic.com:443"]}
    )
    profile = Profile(
        id="test",
        version="1",
        extension="ext.bwrap",
        network={"default": "deny", "egress": ["example.com:443"]},
    )
    with pytest.raises(ValueError, match="owner allowlist"):
        ext.provision(profile, {}, "localhost")
    assert not (tmp_path / "unused").exists()


def test_failed_broker_readiness_removes_egress():
    from sandboxer.extensions.bwrap import BwrapExtension

    ext = BwrapExtension()
    handle = {"egress_pid": "123"}
    with (
        patch.object(ext, "_wait_ready", side_effect=RuntimeError("startup failed")),
        patch.object(ext, "teardown") as cleanup,
    ):
        with pytest.raises(RuntimeError, match="startup failed"):
            ext.wait_ready(handle)
        cleanup.assert_called_once_with(handle)


def test_candidate_claude_profile_declares_exact_destination_and_fails_closed(tmp_path):
    """The committed candidate profile grants nothing without owner allowlisting."""
    from sandboxer.extensions.bwrap import BwrapExtension
    from sandboxer.profiles.loader import load_profile

    profile = load_profile("profile.claude-agent-dev-proof")
    assert profile.version == "1.1.0"
    assert profile.extension == "ext.bwrap"
    assert profile.network.default == "deny"
    assert profile.network.egress == ["api.anthropic.com:443"]
    assert profile.setup.secret_refs == []

    ext = BwrapExtension({"base_dir": str(tmp_path / "unused")})
    with pytest.raises(ValueError, match="owner allowlist"):
        ext.provision(profile, {}, "localhost")
    assert not (tmp_path / "unused").exists()