wehub-resource-sync
1811 lines
56 KiB
Python
1811 lines
56 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import socket
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from urllib.parse import urlsplit
|
|
|
|
import pytest
|
|
|
|
from opensquilla.env import trust_env
|
|
from opensquilla.gateway import rpc_tools
|
|
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
|
|
from opensquilla.gateway.auth import Principal
|
|
from opensquilla.gateway.rpc import RpcContext
|
|
from opensquilla.sandbox import integration as integration_mod
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.sandbox.integration import configure_runtime, reset_runtime, sandboxed
|
|
from opensquilla.sandbox.network_guard import NetworkDecision
|
|
from opensquilla.sandbox.network_proxy import SandboxProxyServer as RealSandboxProxyServer
|
|
from opensquilla.sandbox.run_context import (
|
|
DomainGrant,
|
|
RunContext,
|
|
TemporaryGrant,
|
|
get_run_context,
|
|
)
|
|
from opensquilla.sandbox.run_mode import RunMode
|
|
from opensquilla.sandbox.types import (
|
|
DenialResult,
|
|
NetworkMode,
|
|
ResourceLimits,
|
|
SandboxPolicy,
|
|
SecurityLevel,
|
|
)
|
|
from opensquilla.tools.builtin import web as web_mod
|
|
from opensquilla.tools.builtin import web_fetch as web_fetch_mod
|
|
from opensquilla.tools.types import CallerKind, ToolContext, current_tool_context
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def sandbox_runtime(tmp_path: Path) -> Iterator[None]:
|
|
reset_approval_queue()
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
configure_runtime(
|
|
SandboxSettings(
|
|
run_mode="standard",
|
|
backend="noop",
|
|
allow_legacy_mode=True,
|
|
network_default="proxy_allowlist",
|
|
),
|
|
workspace=workspace,
|
|
)
|
|
try:
|
|
yield
|
|
finally:
|
|
reset_approval_queue()
|
|
reset_runtime()
|
|
|
|
|
|
@pytest.fixture
|
|
def managed_context(tmp_path: Path) -> Iterator[ToolContext]:
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="standard",
|
|
sandbox_run_context=RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
domains=(DomainGrant(domain="allowed.test"),),
|
|
),
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
try:
|
|
yield ctx
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
|
|
async def _send_current_managed_proxy_request(request: bytes) -> bytes:
|
|
proxy_url = integration_mod.current_managed_network_proxy_url()
|
|
assert proxy_url is not None
|
|
parsed = urlsplit(proxy_url)
|
|
assert parsed.hostname is not None
|
|
assert parsed.port is not None
|
|
reader, writer = await asyncio.open_connection(parsed.hostname, parsed.port)
|
|
try:
|
|
writer.write(request)
|
|
await writer.drain()
|
|
return await reader.read(4096)
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
|
|
def _install_trusted_session_handles(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
):
|
|
from opensquilla.tools.builtin import sessions as sessions_mod
|
|
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
manager = SimpleNamespace()
|
|
manager.node = SimpleNamespace(
|
|
session_key="s1",
|
|
agent_id="main",
|
|
origin={
|
|
"sandbox_run_context": RunContext(run_mode=RunMode.TRUSTED).to_origin_payload(),
|
|
},
|
|
)
|
|
|
|
async def _get_session(session_key: str):
|
|
return manager.node if session_key == manager.node.session_key else None
|
|
|
|
async def _update(session_key: str, **fields):
|
|
for key, value in fields.items():
|
|
setattr(manager.node, key, value)
|
|
return manager.node
|
|
|
|
manager.get_session = _get_session
|
|
manager.update = _update
|
|
config = SimpleNamespace(
|
|
workspace_dir=str(workspace),
|
|
agents=[],
|
|
sandbox=SimpleNamespace(
|
|
run_mode="trusted",
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="noop",
|
|
network_default="proxy_allowlist",
|
|
),
|
|
permissions=SimpleNamespace(default_mode="off"),
|
|
)
|
|
monkeypatch.setattr(sessions_mod, "_session_manager", manager)
|
|
monkeypatch.setattr(sessions_mod, "_gateway_config", config)
|
|
return workspace, manager, config
|
|
|
|
|
|
def test_request_with_managed_network_proxy_env_overrides_user_proxy_env(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
policy = SandboxPolicy(
|
|
level=SecurityLevel.STANDARD,
|
|
network=NetworkMode.PROXY_ALLOWLIST,
|
|
mounts=(),
|
|
workspace_rw=True,
|
|
tmp_writable=True,
|
|
limits=ResourceLimits(),
|
|
env_allowlist=("PATH",),
|
|
require_approval=False,
|
|
network_proxy=integration_mod.NetworkProxySpec(host="127.0.0.1", port=18080),
|
|
)
|
|
request = integration_mod.SandboxRequest(
|
|
argv=("sh", "-lc", "curl https://example.com"),
|
|
cwd=tmp_path,
|
|
action_kind="shell.exec",
|
|
policy=policy,
|
|
env={
|
|
"PATH": "/bin",
|
|
"HTTP_PROXY": "http://attacker.invalid:1",
|
|
"http_proxy": "http://attacker.invalid:3",
|
|
"HTTPS_PROXY": "http://attacker.invalid:2",
|
|
"https_proxy": "http://attacker.invalid:4",
|
|
"npm_config_https_proxy": "http://attacker.invalid:5",
|
|
"PIP_PROXY": "http://attacker.invalid:6",
|
|
"WS_PROXY": "http://attacker.invalid:7",
|
|
"NO_PROXY": "*",
|
|
"no_proxy": "*",
|
|
},
|
|
)
|
|
|
|
updated = integration_mod.request_with_managed_network_proxy_env(request)
|
|
|
|
assert updated.env["PATH"] == "/bin"
|
|
for key in (
|
|
"HTTP_PROXY",
|
|
"HTTPS_PROXY",
|
|
"http_proxy",
|
|
"https_proxy",
|
|
"npm_config_https_proxy",
|
|
"PIP_PROXY",
|
|
"WS_PROXY",
|
|
):
|
|
assert updated.env[key] == "http://127.0.0.1:18080"
|
|
for key in ("NO_PROXY", "no_proxy"):
|
|
assert updated.env[key]
|
|
assert "http://attacker.invalid:1" not in updated.env.values()
|
|
assert "http://attacker.invalid:3" not in updated.env.values()
|
|
assert "http://attacker.invalid:5" not in updated.env.values()
|
|
assert "http://attacker.invalid:6" not in updated.env.values()
|
|
assert "http://attacker.invalid:7" not in updated.env.values()
|
|
for key in ("HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "PIP_PROXY"):
|
|
assert key in updated.policy.env_allowlist
|
|
assert request.env["HTTP_PROXY"] == "http://attacker.invalid:1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_url_shaped_inprocess_network_action_sets_context_proxy_without_env_mutation(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
events: list[str] = []
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
events.append("proxy.init")
|
|
|
|
async def start(self) -> None:
|
|
events.append("proxy.start")
|
|
decision = self._decide("allowed.test")
|
|
assert isinstance(decision, NetworkDecision)
|
|
seen["decision"] = decision.status
|
|
|
|
async def stop(self) -> None:
|
|
events.append("proxy.stop")
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
monkeypatch.setenv("HTTP_PROXY", "http://user.invalid:1")
|
|
monkeypatch.delenv("HTTPS_PROXY", raising=False)
|
|
monkeypatch.setenv("http_proxy", "http://user-lower.invalid:1")
|
|
monkeypatch.delenv("https_proxy", raising=False)
|
|
monkeypatch.setenv("ALL_PROXY", "http://all.invalid:1")
|
|
monkeypatch.delenv("all_proxy", raising=False)
|
|
monkeypatch.setenv("NO_PROXY", "*")
|
|
monkeypatch.delenv("no_proxy", raising=False)
|
|
monkeypatch.setenv("OPENSQUILLA_TRUST_ENV", "0")
|
|
expected_env = {
|
|
key: os.environ.get(key)
|
|
for key in (
|
|
"HTTP_PROXY",
|
|
"HTTPS_PROXY",
|
|
"http_proxy",
|
|
"https_proxy",
|
|
"ALL_PROXY",
|
|
"all_proxy",
|
|
"NO_PROXY",
|
|
"no_proxy",
|
|
"OPENSQUILLA_TRUST_ENV",
|
|
)
|
|
}
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
seen["url"] = url
|
|
seen["proxy_url"] = integration_mod.current_managed_network_proxy_url()
|
|
seen["httpx_kwargs"] = integration_mod.managed_network_httpx_kwargs()
|
|
seen["trust_env"] = trust_env()
|
|
return "ok"
|
|
|
|
result = await dummy_http_request("http://allowed.test/path")
|
|
|
|
assert result == "ok"
|
|
assert seen["decision"] == "allow"
|
|
assert seen["proxy_url"] == "http://127.0.0.1:28080"
|
|
assert seen["httpx_kwargs"] == {
|
|
"proxy": "http://127.0.0.1:28080",
|
|
"trust_env": False,
|
|
}
|
|
assert seen["trust_env"] is False
|
|
assert events == ["proxy.init", "proxy.start", "proxy.stop"]
|
|
for key, expected in expected_env.items():
|
|
if expected is None:
|
|
assert key not in os.environ
|
|
else:
|
|
assert os.environ[key] == expected
|
|
assert integration_mod.current_managed_network_proxy_url() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_http_request_uses_explicit_context_proxy_kwargs(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("allowed.test")
|
|
assert isinstance(decision, NetworkDecision)
|
|
assert decision.status == "allow"
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
class FakeAsyncClient:
|
|
def __init__(self, **kwargs: object) -> None:
|
|
seen["client_kwargs"] = kwargs
|
|
|
|
async def __aenter__(self) -> FakeAsyncClient:
|
|
return self
|
|
|
|
async def __aexit__(self, *args: object) -> None:
|
|
return None
|
|
|
|
async def request(self, **kwargs: object) -> object:
|
|
seen["request_kwargs"] = kwargs
|
|
return SimpleNamespace(
|
|
status_code=200,
|
|
url=kwargs["url"],
|
|
headers={"content-type": "text/plain"},
|
|
content=b"ok",
|
|
text="ok",
|
|
)
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
monkeypatch.setattr(web_mod.httpx, "AsyncClient", FakeAsyncClient)
|
|
|
|
payload = json.loads(await web_mod.http_request("http://allowed.test/path"))
|
|
|
|
assert payload["status"] == 200
|
|
assert seen["client_kwargs"] == {
|
|
"timeout": 30.0,
|
|
"proxy": "http://127.0.0.1:28080",
|
|
"trust_env": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_explicit_target_does_not_preflight_without_network_request(
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
|
|
result = await dummy_http_request("http://unknown.test/path")
|
|
|
|
assert result == "ok"
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parallel_unknown_targets_do_not_preflight_without_network_request(
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
|
|
first, second = await asyncio.gather(
|
|
dummy_http_request("http://first-unknown.test/path"),
|
|
dummy_http_request("http://second-unknown.test/path"),
|
|
)
|
|
|
|
assert first == "ok"
|
|
assert second == "ok"
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_temporary_network_grant_allows_retry_for_explicit_target(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("unknown.test")
|
|
assert isinstance(decision, NetworkDecision)
|
|
assert decision.status == "allow"
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
|
|
request = integration_mod.build_request(
|
|
action_kind="network.http",
|
|
argv=("http_request", "GET", "http://unknown.test/path"),
|
|
cwd=Path(managed_context.workspace_dir or "."),
|
|
policy=SandboxPolicy(
|
|
level=SecurityLevel.STANDARD,
|
|
network=NetworkMode.PROXY_ALLOWLIST,
|
|
mounts=(),
|
|
workspace_rw=True,
|
|
tmp_writable=True,
|
|
limits=ResourceLimits(),
|
|
env_allowlist=("PATH",),
|
|
require_approval=False,
|
|
),
|
|
)
|
|
managed_context.sandbox_run_context = RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
temporary_grants=(
|
|
TemporaryGrant(
|
|
kind="domain",
|
|
value="unknown.test",
|
|
fingerprint=integration_mod.action_fingerprint(request),
|
|
),
|
|
),
|
|
)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
result = await dummy_http_request("http://unknown.test/path")
|
|
|
|
assert result == "ok"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_unready_boundary_blocks_decorated_network_tools(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
runtime = integration_mod.get_runtime()
|
|
assert runtime is not None
|
|
runtime.backend = SimpleNamespace(name="windows_default")
|
|
managed_context.sandbox_run_context = RunContext(run_mode=RunMode.TRUSTED)
|
|
|
|
class _UnexpectedClient:
|
|
def __init__(self, *args: object, **kwargs: object) -> None:
|
|
pytest.fail("network tool bypassed Windows managed-network readiness")
|
|
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"_windows_proxy_allowlist_enforced",
|
|
lambda runtime: False,
|
|
)
|
|
async def _repair_failed(runtime) -> bool:
|
|
return False
|
|
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"_ensure_windows_proxy_allowlist_setup",
|
|
_repair_failed,
|
|
)
|
|
monkeypatch.setattr(web_mod.httpx, "AsyncClient", _UnexpectedClient)
|
|
monkeypatch.setattr(web_fetch_mod.httpx, "AsyncClient", _UnexpectedClient)
|
|
monkeypatch.setattr(web_fetch_mod, "_check_ssrf", lambda url: None)
|
|
|
|
http_payload = json.loads(await web_mod.http_request("https://example.com"))
|
|
fetch_payload = json.loads(await web_fetch_mod.web_fetch("https://example.com"))
|
|
|
|
assert http_payload["status"] == "denied"
|
|
assert fetch_payload["status"] == "denied"
|
|
assert "Windows sandbox managed network is unavailable" in http_payload["message"]
|
|
assert "Windows sandbox managed network is unavailable" in fetch_payload["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_unready_boundary_does_not_trigger_elevation_when_not_admin(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
runtime = integration_mod.get_runtime()
|
|
assert runtime is not None
|
|
runtime.backend = SimpleNamespace(name="windows_default")
|
|
managed_context.sandbox_run_context = RunContext(run_mode=RunMode.TRUSTED)
|
|
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"_windows_proxy_allowlist_enforced",
|
|
lambda runtime: False,
|
|
)
|
|
monkeypatch.setattr(integration_mod, "_windows_process_is_admin", lambda: False)
|
|
|
|
async def fail_if_called(settings):
|
|
pytest.fail("Trusted managed-network repair must not trigger elevation")
|
|
|
|
monkeypatch.setattr(integration_mod, "ensure_sandbox_setup", fail_if_called, raising=False)
|
|
|
|
assert await integration_mod._windows_proxy_allowlist_ready_or_repaired(runtime) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_unready_boundary_repairs_when_already_admin(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
runtime = integration_mod.get_runtime()
|
|
assert runtime is not None
|
|
runtime.backend = SimpleNamespace(name="windows_default")
|
|
managed_context.sandbox_run_context = RunContext(run_mode=RunMode.TRUSTED)
|
|
calls: list[object] = []
|
|
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"_windows_proxy_allowlist_enforced",
|
|
lambda runtime: False,
|
|
)
|
|
monkeypatch.setattr(integration_mod, "_windows_process_is_admin", lambda: True)
|
|
|
|
async def _repair(runtime):
|
|
calls.append(runtime)
|
|
return True
|
|
|
|
monkeypatch.setattr(integration_mod, "_ensure_windows_proxy_allowlist_setup", _repair)
|
|
|
|
assert await integration_mod._windows_proxy_allowlist_ready_or_repaired(runtime) is True
|
|
assert calls == [runtime]
|
|
|
|
|
|
def test_windows_unavailable_message_reports_outdated_network_marker(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from opensquilla.sandbox.backend import windows_default_setup as setup_mod
|
|
from opensquilla.sandbox.backend.windows_default_network import (
|
|
WindowsNetworkSetup,
|
|
)
|
|
|
|
marker = tmp_path / "setup_marker.json"
|
|
setup_mod.write_setup_marker(
|
|
marker,
|
|
network=WindowsNetworkSetup(
|
|
offline_user_sid="S-1-5-21-100-200-300-400",
|
|
allowed_proxy_ports=(48123,),
|
|
allow_local_binding=False,
|
|
firewall_rule_version=1,
|
|
wfp_rule_version=1,
|
|
),
|
|
)
|
|
monkeypatch.setattr(setup_mod, "default_setup_marker_path", lambda home=None: marker)
|
|
runtime = SimpleNamespace(backend=SimpleNamespace(name="windows_default"))
|
|
|
|
message = integration_mod._windows_managed_network_unavailable_message(runtime)
|
|
|
|
assert "network marker is out of date" in message
|
|
assert "firewall=1 required=5" in message
|
|
assert "wfp=1 required=2" in message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_unready_boundary_blocks_rpc_network_action(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
runtime = integration_mod.get_runtime()
|
|
assert runtime is not None
|
|
runtime.backend = SimpleNamespace(name="windows_default")
|
|
managed_context.sandbox_run_context = RunContext(run_mode=RunMode.TRUSTED)
|
|
called = False
|
|
|
|
async def _callback() -> str:
|
|
nonlocal called
|
|
called = True
|
|
return "unexpected"
|
|
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"_windows_proxy_allowlist_enforced",
|
|
lambda runtime: False,
|
|
)
|
|
async def _repair_failed(runtime) -> bool:
|
|
return False
|
|
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"_ensure_windows_proxy_allowlist_setup",
|
|
_repair_failed,
|
|
)
|
|
|
|
result = await integration_mod.run_in_process_network_action(
|
|
action_kind="network.http",
|
|
argv=("http_request", "GET", "https://example.com"),
|
|
callback=_callback,
|
|
)
|
|
|
|
assert called is False
|
|
assert isinstance(result, DenialResult)
|
|
assert result.retryable is False
|
|
assert "Windows sandbox managed network is unavailable" in result.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_allow_once_resolve_allows_one_retry_then_expires_for_explicit_target(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
return None
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
|
|
first = await dummy_http_request("http://unknown.test/path")
|
|
second = await dummy_http_request("http://unknown.test/path")
|
|
|
|
assert first == "ok"
|
|
assert second == "ok"
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_allow_same_type_choice_keeps_later_unknown_public_targets_scoped(
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
|
|
result = await dummy_http_request("http://docs.example.com/path")
|
|
|
|
assert result == "ok"
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_persisted_temporary_grant_from_saved_origin_does_not_allow_after_restart(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("unknown.test")
|
|
assert isinstance(decision, NetworkDecision)
|
|
seen["decision"] = decision.status
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
|
|
persisted = RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
temporary_grants=(
|
|
TemporaryGrant(
|
|
kind="domain",
|
|
value="unknown.test",
|
|
fingerprint="legacy-fp",
|
|
),
|
|
),
|
|
source="saved",
|
|
)
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="standard",
|
|
sandbox_run_context=persisted,
|
|
)
|
|
)
|
|
try:
|
|
result = await dummy_http_request("http://unknown.test/path")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result == "ok"
|
|
assert seen["decision"] == "ask"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_explicit_target_does_not_auto_add_before_proxy_upstream(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("api.github.com")
|
|
assert isinstance(decision, NetworkDecision)
|
|
assert decision.status == "allow"
|
|
assert decision.reason == "auto_trusted"
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
workspace, manager, config = _install_trusted_session_handles(monkeypatch, tmp_path)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
sandbox_run_context=RunContext(run_mode=RunMode.TRUSTED, source="route_metadata"),
|
|
)
|
|
)
|
|
try:
|
|
result = await dummy_http_request("https://api.github.com/repos/openai")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result == "ok"
|
|
saved = await get_run_context(manager, "s1", config=config, workspace=str(workspace))
|
|
assert DomainGrant(
|
|
domain="api.github.com",
|
|
scope="chat",
|
|
source="auto_trusted",
|
|
) not in saved.domains
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_inprocess_auto_trust_does_not_persist_private_resolution(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace, manager, config = _install_trusted_session_handles(monkeypatch, tmp_path)
|
|
real_getaddrinfo = socket.getaddrinfo
|
|
|
|
def fake_getaddrinfo(host: str, port: int, *args: object, **kwargs: object) -> list[tuple]:
|
|
if host == "new-public.example":
|
|
return [
|
|
(
|
|
socket.AF_INET,
|
|
socket.SOCK_STREAM,
|
|
socket.IPPROTO_TCP,
|
|
"",
|
|
("127.0.0.1", port),
|
|
)
|
|
]
|
|
return real_getaddrinfo(host, port, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> bytes:
|
|
return await _send_current_managed_proxy_request(
|
|
b"GET http://new-public.example/path HTTP/1.1\r\n\r\n"
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
sandbox_run_context=RunContext(run_mode=RunMode.TRUSTED, source="route_metadata"),
|
|
)
|
|
)
|
|
try:
|
|
response = await dummy_http_request("http://new-public.example/path")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert response.startswith(b"HTTP/1.1 403")
|
|
saved = await get_run_context(manager, "s1", config=config, workspace=str(workspace))
|
|
assert DomainGrant(
|
|
domain="new-public.example",
|
|
scope="chat",
|
|
source="auto_trusted",
|
|
) not in saved.domains
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_inprocess_auto_trust_persists_after_safe_proxy_upstream(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
async def handle_upstream(
|
|
reader: asyncio.StreamReader,
|
|
writer: asyncio.StreamWriter,
|
|
) -> None:
|
|
await reader.readuntil(b"\r\n\r\n")
|
|
writer.write(
|
|
b"HTTP/1.1 200 OK\r\n"
|
|
b"Content-Length: 2\r\n"
|
|
b"Connection: close\r\n"
|
|
b"\r\n"
|
|
b"ok"
|
|
)
|
|
await writer.drain()
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
upstream = await asyncio.start_server(handle_upstream, "127.0.0.1", 0)
|
|
upstream_socket = next(iter(upstream.sockets or ()), None)
|
|
assert upstream_socket is not None
|
|
upstream_host, upstream_port = upstream_socket.getsockname()[:2]
|
|
public_upstream_host = "93.184.216.34"
|
|
real_open_connection = asyncio.open_connection
|
|
|
|
async def fake_open_connection(
|
|
host: str,
|
|
port: int,
|
|
*args: object,
|
|
**kwargs: object,
|
|
) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
|
|
if str(host) == public_upstream_host and int(port) == int(upstream_port):
|
|
return await real_open_connection(
|
|
str(upstream_host),
|
|
int(upstream_port),
|
|
*args,
|
|
**kwargs,
|
|
)
|
|
return await real_open_connection(host, port, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(asyncio, "open_connection", fake_open_connection)
|
|
|
|
def proxy_factory(decide: object, **kwargs: object) -> RealSandboxProxyServer:
|
|
return RealSandboxProxyServer(
|
|
decide,
|
|
resolver=lambda host, port: (public_upstream_host, int(upstream_port)),
|
|
**kwargs,
|
|
)
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", proxy_factory)
|
|
workspace, manager, config = _install_trusted_session_handles(monkeypatch, tmp_path)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> bytes:
|
|
return await _send_current_managed_proxy_request(
|
|
b"GET http://new-public.example/path HTTP/1.1\r\n\r\n"
|
|
)
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
sandbox_run_context=RunContext(run_mode=RunMode.TRUSTED, source="route_metadata"),
|
|
)
|
|
)
|
|
try:
|
|
response = await dummy_http_request("http://new-public.example/path")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
upstream.close()
|
|
await upstream.wait_closed()
|
|
|
|
assert response.startswith(b"HTTP/1.1 200 OK")
|
|
saved = await get_run_context(manager, "s1", config=config, workspace=str(workspace))
|
|
assert DomainGrant(
|
|
domain="new-public.example",
|
|
scope="chat",
|
|
source="auto_trusted",
|
|
) in saved.domains
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_explicit_target_auto_adds_chat_domain_grant_in_production_path(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from opensquilla.tools.builtin import sessions as sessions_mod
|
|
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
self._on_upstream_opened = kwargs.get("on_upstream_opened")
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("api.github.com")
|
|
assert isinstance(decision, NetworkDecision)
|
|
assert decision.status == "allow"
|
|
assert self._on_upstream_opened is not None
|
|
await self._on_upstream_opened(decision)
|
|
seen["callback_called"] = True
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
manager = SimpleNamespace()
|
|
manager.node = SimpleNamespace(
|
|
session_key="s1",
|
|
agent_id="main",
|
|
origin={
|
|
"sandbox_run_context": RunContext(run_mode=RunMode.TRUSTED).to_origin_payload(),
|
|
},
|
|
)
|
|
|
|
async def _get_session(session_key: str):
|
|
return manager.node if session_key == manager.node.session_key else None
|
|
|
|
async def _update(session_key: str, **fields):
|
|
for key, value in fields.items():
|
|
setattr(manager.node, key, value)
|
|
return manager.node
|
|
|
|
manager.get_session = _get_session
|
|
manager.update = _update
|
|
config = SimpleNamespace(
|
|
workspace_dir=str(workspace),
|
|
agents=[],
|
|
sandbox=SimpleNamespace(
|
|
run_mode="trusted",
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="noop",
|
|
network_default="proxy_allowlist",
|
|
),
|
|
permissions=SimpleNamespace(default_mode="off"),
|
|
)
|
|
monkeypatch.setattr(sessions_mod, "_session_manager", manager)
|
|
monkeypatch.setattr(sessions_mod, "_gateway_config", config)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
sandbox_run_context=RunContext(run_mode=RunMode.TRUSTED, source="route_metadata"),
|
|
)
|
|
)
|
|
try:
|
|
result = await dummy_http_request("https://api.github.com/repos/openai")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result == "ok"
|
|
assert seen["callback_called"] is True
|
|
saved = await get_run_context(manager, "s1", config=config, workspace=str(workspace))
|
|
assert DomainGrant(
|
|
domain="api.github.com",
|
|
scope="chat",
|
|
source="auto_trusted",
|
|
) in saved.domains
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standard_explicit_target_does_not_auto_add_recognized_default_host(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from opensquilla.tools.builtin import sessions as sessions_mod
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("api.github.com")
|
|
assert isinstance(decision, NetworkDecision)
|
|
assert decision.status == "allow"
|
|
assert decision.reason == "default_allowlist"
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
manager = SimpleNamespace()
|
|
manager.node = SimpleNamespace(
|
|
session_key="s1",
|
|
agent_id="main",
|
|
origin={
|
|
"sandbox_run_context": RunContext(run_mode=RunMode.STANDARD).to_origin_payload(),
|
|
},
|
|
)
|
|
|
|
async def _get_session(session_key: str):
|
|
return manager.node if session_key == manager.node.session_key else None
|
|
|
|
async def _update(session_key: str, **fields):
|
|
for key, value in fields.items():
|
|
setattr(manager.node, key, value)
|
|
return manager.node
|
|
|
|
manager.get_session = _get_session
|
|
manager.update = _update
|
|
config = SimpleNamespace(
|
|
workspace_dir=str(workspace),
|
|
agents=[],
|
|
sandbox=SimpleNamespace(
|
|
run_mode="standard",
|
|
sandbox=True,
|
|
security_grading=True,
|
|
backend="noop",
|
|
network_default="proxy_allowlist",
|
|
),
|
|
permissions=SimpleNamespace(default_mode="off"),
|
|
)
|
|
monkeypatch.setattr(sessions_mod, "_session_manager", manager)
|
|
monkeypatch.setattr(sessions_mod, "_gateway_config", config)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
return "ok"
|
|
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="standard",
|
|
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD, source="route_metadata"),
|
|
)
|
|
)
|
|
try:
|
|
result = await dummy_http_request("https://api.github.com/repos/openai")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result == "ok"
|
|
saved = await get_run_context(manager, "s1", config=config, workspace=str(workspace))
|
|
assert saved.domains == ()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_with_managed_network_proxy_honors_temporary_domain_grant(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("temp-allowed.test")
|
|
assert isinstance(decision, NetworkDecision)
|
|
seen["decision"] = decision
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
class FakeBackend:
|
|
name = "fake"
|
|
|
|
async def run(self, request):
|
|
seen["policy"] = request.policy
|
|
return SimpleNamespace(
|
|
returncode=0,
|
|
stdout="ok",
|
|
stderr="",
|
|
wall_time_s=0.1,
|
|
backend_used="fake",
|
|
backend_notes=(),
|
|
)
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
policy = SandboxPolicy(
|
|
level=SecurityLevel.STANDARD,
|
|
network=NetworkMode.PROXY_ALLOWLIST,
|
|
mounts=(),
|
|
workspace_rw=True,
|
|
tmp_writable=True,
|
|
limits=ResourceLimits(),
|
|
env_allowlist=("PATH",),
|
|
require_approval=False,
|
|
)
|
|
|
|
request = integration_mod.build_request(
|
|
action_kind="network.http",
|
|
argv=("http_request", "GET", "http://temp-allowed.test/path"),
|
|
cwd=workspace,
|
|
policy=policy,
|
|
)
|
|
runtime = SimpleNamespace(
|
|
backend=FakeBackend(),
|
|
)
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="standard",
|
|
sandbox_run_context=RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
temporary_grants=(
|
|
TemporaryGrant(
|
|
kind="domain",
|
|
value="temp-allowed.test",
|
|
fingerprint=integration_mod.action_fingerprint(request),
|
|
),
|
|
),
|
|
),
|
|
)
|
|
)
|
|
try:
|
|
result = await integration_mod._run_with_managed_network_proxy(request, runtime)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result.stdout == "ok"
|
|
decision = seen["decision"]
|
|
assert isinstance(decision, NetworkDecision)
|
|
assert decision.status == "allow"
|
|
assert decision.reason == "domain_grant"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_windows_fixed_proxy_port_subprocesses_are_serialized(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
runtime = integration_mod.get_runtime()
|
|
assert runtime is not None
|
|
runtime.backend = SimpleNamespace(name="windows_default")
|
|
managed_context.sandbox_run_context = RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
domains=(DomainGrant(domain="example.com"),),
|
|
)
|
|
monkeypatch.setattr(integration_mod, "_windows_allowed_proxy_ports", lambda _rt: (48123,))
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"_windows_proxy_allowlist_enforced",
|
|
lambda _rt, *, proxy_ports=(): True,
|
|
)
|
|
|
|
events: list[str] = []
|
|
active = 0
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
|
|
def __init__(self, _service: object, *, port: int = 0, **_kwargs: object) -> None:
|
|
self.port = port
|
|
|
|
async def start(self) -> None:
|
|
nonlocal active
|
|
if active:
|
|
raise AssertionError("fixed Windows proxy port was started concurrently")
|
|
active += 1
|
|
events.append(f"start:{self.port}")
|
|
|
|
async def stop(self) -> None:
|
|
nonlocal active
|
|
active -= 1
|
|
events.append("stop")
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
first_ready = asyncio.Event()
|
|
release_first = asyncio.Event()
|
|
policy = SandboxPolicy(
|
|
level=SecurityLevel.STANDARD,
|
|
network=NetworkMode.PROXY_ALLOWLIST,
|
|
mounts=(),
|
|
workspace_rw=True,
|
|
tmp_writable=True,
|
|
limits=ResourceLimits(),
|
|
env_allowlist=("PATH",),
|
|
require_approval=False,
|
|
)
|
|
|
|
async def run_managed(name: str) -> None:
|
|
request = integration_mod.build_request(
|
|
action_kind="shell.exec",
|
|
argv=("curl", f"https://example.com/{name}"),
|
|
cwd=tmp_path,
|
|
policy=policy,
|
|
)
|
|
managed = await integration_mod.prepare_subprocess_managed_network_proxy(
|
|
request,
|
|
runtime=runtime,
|
|
)
|
|
events.append(f"ready:{name}")
|
|
try:
|
|
if name == "first":
|
|
first_ready.set()
|
|
await release_first.wait()
|
|
finally:
|
|
await managed.cleanup()
|
|
|
|
first = asyncio.create_task(run_managed("first"))
|
|
await first_ready.wait()
|
|
second = asyncio.create_task(run_managed("second"))
|
|
await asyncio.sleep(0)
|
|
|
|
assert events == ["start:48123", "ready:first"]
|
|
|
|
release_first.set()
|
|
await asyncio.gather(first, second)
|
|
|
|
assert events == [
|
|
"start:48123",
|
|
"ready:first",
|
|
"stop",
|
|
"start:48123",
|
|
"ready:second",
|
|
"stop",
|
|
]
|
|
assert active == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_with_managed_network_proxy_does_not_auto_add_before_proxy_upstream(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
seen["has_callback"] = kwargs.get("on_upstream_opened") is not None
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("api.github.com")
|
|
assert isinstance(decision, NetworkDecision)
|
|
assert decision.status == "allow"
|
|
assert decision.reason == "auto_trusted"
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
class FakeBackend:
|
|
name = "fake"
|
|
|
|
async def run(self, request):
|
|
seen["policy"] = request.policy
|
|
return SimpleNamespace(
|
|
returncode=0,
|
|
stdout="ok",
|
|
stderr="",
|
|
wall_time_s=0.1,
|
|
backend_used="fake",
|
|
backend_notes=(),
|
|
)
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
workspace, manager, config = _install_trusted_session_handles(monkeypatch, tmp_path)
|
|
policy = SandboxPolicy(
|
|
level=SecurityLevel.STANDARD,
|
|
network=NetworkMode.PROXY_ALLOWLIST,
|
|
mounts=(),
|
|
workspace_rw=True,
|
|
tmp_writable=True,
|
|
limits=ResourceLimits(),
|
|
env_allowlist=("PATH",),
|
|
require_approval=False,
|
|
)
|
|
request = integration_mod.build_request(
|
|
action_kind="network.http",
|
|
argv=("http_request", "GET", "https://api.github.com/repos/openai"),
|
|
cwd=workspace,
|
|
policy=policy,
|
|
)
|
|
runtime = SimpleNamespace(
|
|
backend=FakeBackend(),
|
|
)
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(workspace),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
sandbox_run_context=RunContext(run_mode=RunMode.TRUSTED, source="route_metadata"),
|
|
)
|
|
)
|
|
try:
|
|
result = await integration_mod._run_with_managed_network_proxy(request, runtime)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result.stdout == "ok"
|
|
assert seen["has_callback"] is True
|
|
saved = await get_run_context(manager, "s1", config=config, workspace=str(workspace))
|
|
assert DomainGrant(
|
|
domain="api.github.com",
|
|
scope="chat",
|
|
source="auto_trusted",
|
|
) not in saved.domains
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_fetch_cache_hit_requires_current_run_context_grant(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
url = "http://blocked.test/page"
|
|
web_fetch_mod._cache.clear()
|
|
web_fetch_mod._cache[(url, "markdown")] = {
|
|
"url": url,
|
|
"final_url": url,
|
|
"status": 200,
|
|
"content_type": "text/html",
|
|
"title": "",
|
|
"extract_mode": "markdown",
|
|
"extractor": "cache",
|
|
"truncated": False,
|
|
"length": 13,
|
|
"text": "cached secret",
|
|
}
|
|
monkeypatch.setattr(web_fetch_mod, "_check_ssrf", lambda value: None)
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"SandboxProxyServer",
|
|
lambda *args, **kwargs: pytest.fail("proxy should not start for denied target"),
|
|
)
|
|
|
|
result = await web_fetch_mod.web_fetch(url)
|
|
|
|
payload = json.loads(result)
|
|
assert payload["status"] == "approval_required"
|
|
assert payload["approval_id"]
|
|
assert payload["approvalKind"] == "sandbox_network"
|
|
assert payload["host"] == "blocked.test"
|
|
assert "cached secret" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_search_query_allows_search_provider_endpoint_under_managed_network(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
provider_decision = self._decide("api.search.brave.com")
|
|
unknown_decision = self._decide("not-approved.example")
|
|
assert isinstance(provider_decision, NetworkDecision)
|
|
assert isinstance(unknown_decision, NetworkDecision)
|
|
seen["provider_decision"] = provider_decision.status
|
|
seen["provider_reason"] = provider_decision.reason
|
|
seen["unknown_decision"] = unknown_decision.status
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
async def fake_search(*args: object, **kwargs: object) -> dict[str, object]:
|
|
seen["search_called"] = True
|
|
seen["provider_kwargs"] = web_mod._search_provider_kwargs("brave")
|
|
return {
|
|
"ok": True,
|
|
"query": "python packages",
|
|
"provider": "brave",
|
|
"results": [{"title": "PyPI", "url": "https://pypi.org", "snippet": ""}],
|
|
}
|
|
|
|
monkeypatch.setattr(rpc_tools, "run_web_search_payload", fake_search)
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
web_mod.configure_search("brave", api_key="test-key")
|
|
ctx = RpcContext(
|
|
conn_id="c",
|
|
principal=Principal(
|
|
role="operator",
|
|
scopes=frozenset(["operator.write", "operator.read"]),
|
|
is_owner=True,
|
|
authenticated=True,
|
|
),
|
|
)
|
|
|
|
result = await rpc_tools._handle_search_query({"query": "python packages"}, ctx)
|
|
|
|
assert result["ok"] is True
|
|
assert result["provider"] == "brave"
|
|
assert seen == {
|
|
"provider_decision": "allow",
|
|
"provider_reason": "system_domain_grant",
|
|
"unknown_decision": "ask",
|
|
"provider_kwargs": {
|
|
"proxy": "http://127.0.0.1:28080",
|
|
"use_env_proxy": False,
|
|
"api_key": "test-key",
|
|
"diagnostics": False,
|
|
},
|
|
"search_called": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_search_query_without_runtime_uses_provider_path(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
reset_runtime()
|
|
|
|
async def fake_search(*args: object, **kwargs: object) -> dict[str, object]:
|
|
return {
|
|
"ok": True,
|
|
"query": "python packages",
|
|
"provider": "fake",
|
|
"results": [{"title": "PyPI", "url": "https://pypi.org", "snippet": "python"}],
|
|
}
|
|
|
|
monkeypatch.setattr(rpc_tools, "run_web_search_payload", fake_search)
|
|
ctx = RpcContext(
|
|
conn_id="c",
|
|
principal=Principal(
|
|
role="operator",
|
|
scopes=frozenset(["operator.write", "operator.read"]),
|
|
is_owner=True,
|
|
authenticated=True,
|
|
),
|
|
)
|
|
|
|
result = await rpc_tools._handle_search_query({"query": "python packages"}, ctx)
|
|
|
|
assert result["ok"] is True
|
|
assert result["results"] == [
|
|
{"title": "PyPI", "url": "https://pypi.org", "snippet": "python"}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inprocess_network_action_without_run_context_fails_closed(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="standard",
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
monkeypatch.setattr(
|
|
integration_mod,
|
|
"SandboxProxyServer",
|
|
lambda *args, **kwargs: pytest.fail("proxy should not start without context"),
|
|
)
|
|
called = False
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
nonlocal called
|
|
called = True
|
|
return url
|
|
|
|
try:
|
|
result = await dummy_http_request("http://allowed.test/path")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
payload = json.loads(result)
|
|
assert payload["status"] == "denied"
|
|
assert payload["reason"] == "policy_denied"
|
|
assert "Run Context" in payload["message"]
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_search_shaped_inprocess_action_uses_search_provider_endpoint_grant(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
managed_context: ToolContext,
|
|
) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("html.duckduckgo.com")
|
|
assert isinstance(decision, NetworkDecision)
|
|
seen["decision"] = decision.status
|
|
seen["reason"] = decision.reason
|
|
|
|
async def stop(self) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
web_mod.configure_search("duckduckgo")
|
|
|
|
@sandboxed(
|
|
"web.fetch",
|
|
argv_factory=lambda a: (
|
|
"web_search",
|
|
str(a.get("query", "")),
|
|
str(a.get("max_results", "")),
|
|
),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_web_search(query: str, max_results: int | None = None) -> str:
|
|
seen["called"] = True
|
|
return query
|
|
|
|
result = await dummy_web_search("python packages", 5)
|
|
|
|
assert result == "python packages"
|
|
assert seen == {
|
|
"decision": "allow",
|
|
"reason": "system_domain_grant",
|
|
"called": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inprocess_network_action_with_network_none_defers_to_proxy_runtime(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
reset_runtime()
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
configure_runtime(
|
|
SandboxSettings(
|
|
run_mode="standard",
|
|
backend="noop",
|
|
allow_legacy_mode=True,
|
|
network_default="none",
|
|
),
|
|
workspace=workspace,
|
|
)
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="standard",
|
|
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
seen["started"] = True
|
|
|
|
async def stop(self) -> None:
|
|
seen["stopped"] = True
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
called = False
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
nonlocal called
|
|
called = True
|
|
return url
|
|
|
|
try:
|
|
result = await dummy_http_request("http://example.com")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result == "http://example.com"
|
|
assert called is True
|
|
assert seen == {"started": True, "stopped": True}
|
|
assert get_approval_queue().list_pending("exec") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inprocess_network_action_with_network_none_uses_granted_explicit_target(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
reset_runtime()
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
configure_runtime(
|
|
SandboxSettings(
|
|
run_mode="standard",
|
|
backend="noop",
|
|
allow_legacy_mode=True,
|
|
network_default="none",
|
|
),
|
|
workspace=workspace,
|
|
)
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="standard",
|
|
sandbox_run_context=RunContext(
|
|
run_mode=RunMode.STANDARD,
|
|
domains=(
|
|
DomainGrant(
|
|
domain="example.com",
|
|
scope="chat",
|
|
source="user",
|
|
),
|
|
),
|
|
),
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("example.com")
|
|
assert isinstance(decision, NetworkDecision)
|
|
seen["decision"] = decision.status
|
|
|
|
async def stop(self) -> None:
|
|
seen["stopped"] = True
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
seen["called"] = True
|
|
return "ok"
|
|
|
|
try:
|
|
result = await dummy_http_request("http://example.com/path")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result == "ok"
|
|
assert seen == {"decision": "allow", "called": True, "stopped": True}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trusted_network_none_auto_allows_unknown_explicit_public_target(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
reset_runtime()
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir(exist_ok=True)
|
|
configure_runtime(
|
|
SandboxSettings(
|
|
run_mode="trusted",
|
|
backend="noop",
|
|
allow_legacy_mode=True,
|
|
network_default="none",
|
|
),
|
|
workspace=workspace,
|
|
)
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
caller_kind=CallerKind.CLI,
|
|
workspace_dir=str(tmp_path),
|
|
session_key="s1",
|
|
run_mode="trusted",
|
|
sandbox_run_context=RunContext(run_mode=RunMode.TRUSTED),
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeProxy:
|
|
host = "127.0.0.1"
|
|
port = 28080
|
|
|
|
def __init__(self, decide: object, **kwargs: object) -> None:
|
|
self._decide = decide
|
|
|
|
async def start(self) -> None:
|
|
decision = self._decide("new-public.example")
|
|
assert isinstance(decision, NetworkDecision)
|
|
seen["decision"] = decision.status
|
|
seen["reason"] = decision.reason
|
|
|
|
async def stop(self) -> None:
|
|
seen["stopped"] = True
|
|
|
|
monkeypatch.setattr(integration_mod, "SandboxProxyServer", FakeProxy)
|
|
|
|
@sandboxed(
|
|
"network.http",
|
|
argv_factory=lambda a: ("http_request", "GET", str(a["url"])),
|
|
record_payload=False,
|
|
)
|
|
async def dummy_http_request(url: str) -> str:
|
|
seen["called"] = True
|
|
return "ok"
|
|
|
|
try:
|
|
result = await dummy_http_request("http://new-public.example/path")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert result == "ok"
|
|
assert seen == {
|
|
"decision": "allow",
|
|
"reason": "auto_trusted",
|
|
"called": True,
|
|
"stopped": True,
|
|
}
|