feat: enforce owner allowlisted bwrap HTTPS egress

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0726e-5232-73f2-aaca-2c05ceb62efb

tests/test_bwrap_runtime.py

@@ -88,7 +88,7 @@ def test_source_and_runtime_overlap_refuses_before_copy(artifact):
 def test_unsupported_network_refuses_before_provisioning(tmp_path, network):
     profile = Profile(id="profile.test", version="1", extension="ext.bwrap", network=network)
     base = tmp_path / "sandboxes"
-    with pytest.raises(ValueError, match="empty egress"):
+    with pytest.raises(ValueError, match="default-deny|owner allowlist"):
         BwrapExtension({"base_dir": str(base)}).provision(profile, {}, "localhost")
     assert not base.exists()
 

tests/test_egress.py

@@ -0,0 +1,115 @@
+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"]
+)
+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)