tegwick
e45e3e6401
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0726e-5232-73f2-aaca-2c05ceb62efb
119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
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)
|