wehub/opensquilla--opensquilla
1
0
Fork 0
Code Issues Pull Requests Actions 1 Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
opensquilla--opensquilla/tests/test_sandbox/test_shell_code_network_hints.py
T
wehub-resource-sync aacb60a4af chore: import upstream snapshot with attribution
2026-07-13 13:12:33 +08:00

3365 lines
107 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import json
import os
import shutil
import socket
import subprocess
import sys
import threading
from collections.abc import Iterator
from pathlib import Path
from types import SimpleNamespace
import pytest
from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
from opensquilla.sandbox.config import SandboxSettings
from opensquilla.sandbox.integration import configure_runtime, get_runtime, reset_runtime
from opensquilla.sandbox.network_runtime import NetworkPolicyRequest, NetworkProtocol
from opensquilla.sandbox.run_context import (
DomainGrant,
PackageBundleGrant,
RunContext,
TemporaryGrant,
)
from opensquilla.sandbox.run_mode import RunMode
from opensquilla.tools.types import CallerKind, ToolContext, current_tool_context
@pytest.fixture
def managed_runtime(tmp_path: Path) -> Iterator[Path]:
reset_approval_queue()
configure_runtime(
SandboxSettings(
run_mode="standard",
backend="noop",
allow_legacy_mode=True,
network_default="proxy_allowlist",
),
workspace=tmp_path,
)
try:
yield tmp_path
finally:
reset_runtime()
reset_approval_queue()
@pytest.fixture
def standard_runtime_no_preflight(tmp_path: Path) -> Iterator[Path]:
reset_approval_queue()
configure_runtime(
SandboxSettings(
run_mode="standard",
backend="noop",
allow_legacy_mode=True,
network_default="none",
),
workspace=tmp_path,
)
try:
yield tmp_path
finally:
reset_runtime()
reset_approval_queue()
@pytest.mark.asyncio
async def test_shell_backend_request_preserves_resolved_run_mode(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import SandboxRequest
from opensquilla.tools.builtin import shell
reset_approval_queue()
configure_runtime(
SandboxSettings(
run_mode="trusted",
backend="noop",
allow_legacy_mode=True,
network_default="none",
),
workspace=tmp_path,
)
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
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,
workspace=str(tmp_path),
),
)
)
try:
result = await shell.exec_command("echo ok", workdir=str(tmp_path))
finally:
current_tool_context.reset(token)
reset_runtime()
reset_approval_queue()
assert "ok" in result
assert backend_calls
assert backend_calls[0].run_mode == "trusted"
@pytest.mark.asyncio
async def test_trusted_windows_shell_receives_managed_proxy_without_network_hint(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import NetworkMode, SandboxRequest
from opensquilla.tools.builtin import shell
reset_approval_queue()
configure_runtime(
SandboxSettings(
run_mode="trusted",
backend="noop",
allow_legacy_mode=True,
network_default="proxy_allowlist",
),
workspace=tmp_path,
)
runtime = get_runtime()
assert runtime is not None
runtime.backend = SimpleNamespace(name="windows_default")
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=0,
stdout=request.env["HTTP_PROXY"],
stderr="",
backend_notes=(),
)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
async def _fake_preflight_subprocess_managed_network(request, runtime, **kwargs):
return None
monkeypatch.setattr(
shell,
"preflight_subprocess_managed_network",
_fake_preflight_subprocess_managed_network,
)
async def _fake_cleanup() -> None:
return None
async def _fake_prepare_subprocess_managed_network_proxy(request, *, runtime=None):
from dataclasses import replace
proxy_policy = replace(request.policy, network=NetworkMode.PROXY_ALLOWLIST)
proxy_request = request.with_policy(proxy_policy)
proxy_request.env["HTTP_PROXY"] = "http://127.0.0.1:48123"
proxy_request.env["HTTPS_PROXY"] = "http://127.0.0.1:48123"
proxy_request.env["NO_PROXY"] = ""
return SimpleNamespace(request=proxy_request, cleanup=_fake_cleanup)
monkeypatch.setattr(
shell,
"prepare_subprocess_managed_network_proxy",
_fake_prepare_subprocess_managed_network_proxy,
)
token = current_tool_context.set(
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,
workspace=str(tmp_path),
),
)
)
try:
result = await shell.exec_command(
"powershell -NoProfile -Command \"Write-Output $env:HTTP_PROXY\"",
workdir=str(tmp_path),
)
finally:
current_tool_context.reset(token)
reset_runtime()
reset_approval_queue()
assert result.startswith("exit_code=0")
assert backend_calls
assert backend_calls[0].policy.network is NetworkMode.PROXY_ALLOWLIST
assert backend_calls[0].env["HTTP_PROXY"].startswith("http://127.0.0.1:")
@pytest.mark.asyncio
async def test_trusted_linux_shell_receives_managed_proxy_without_network_hint(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import NetworkMode, SandboxRequest
from opensquilla.tools.builtin import shell
reset_approval_queue()
configure_runtime(
SandboxSettings(
run_mode="trusted",
backend="noop",
allow_legacy_mode=True,
network_default="proxy_allowlist",
),
workspace=tmp_path,
)
runtime = get_runtime()
assert runtime is not None
runtime.backend = SimpleNamespace(name="bubblewrap")
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=0,
stdout=(
request.env["HTTP_PROXY"]
+ "\n"
+ request.env["OPENSQUILLA_SANDBOX_NETWORK"]
+ "\n"
),
stderr="",
backend_notes=(),
)
async def _windows_ready_should_not_run(runtime):
raise AssertionError("linux managed-network preflight used windows readiness")
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
monkeypatch.setattr(
"opensquilla.sandbox.integration._windows_proxy_allowlist_ready_or_repaired",
_windows_ready_should_not_run,
)
token = current_tool_context.set(
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,
workspace=str(tmp_path),
),
)
)
try:
result = await shell.exec_command(
"sh -lc 'printf \"HTTP_PROXY=%s\\nOPENSQUILLA_SANDBOX_NETWORK=%s\\n\" "
"\"$HTTP_PROXY\" \"$OPENSQUILLA_SANDBOX_NETWORK\"'",
workdir=str(tmp_path),
)
finally:
current_tool_context.reset(token)
reset_runtime()
reset_approval_queue()
assert result.startswith("exit_code=0")
assert backend_calls
assert backend_calls[0].policy.network is NetworkMode.PROXY_ALLOWLIST
assert backend_calls[0].env["HTTP_PROXY"].startswith("http://127.0.0.1:")
assert backend_calls[0].env["OPENSQUILLA_SANDBOX_NETWORK"] == "proxy_allowlist"
@pytest.mark.asyncio
async def test_trusted_windows_code_exec_receives_managed_proxy_without_network_hint(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import NetworkMode, NetworkProxySpec
from opensquilla.tools.builtin import code_exec, shell
reset_approval_queue()
configure_runtime(
SandboxSettings(
run_mode="trusted",
backend="noop",
allow_legacy_mode=True,
network_default="proxy_allowlist",
),
workspace=tmp_path,
)
runtime = get_runtime()
assert runtime is not None
runtime.backend = SimpleNamespace(name="windows_default")
seen: dict[str, object] = {}
prepare_calls: list[object] = []
async def _fake_preflight_subprocess_managed_network(request, runtime):
return None
async def _fake_cleanup() -> None:
return None
async def _fake_prepare_subprocess_managed_network_proxy(request, *, runtime=None):
from dataclasses import replace
prepare_calls.append(request)
proxy_policy = replace(
request.policy,
network=NetworkMode.PROXY_ALLOWLIST,
network_proxy=NetworkProxySpec(host="127.0.0.1", port=48123),
)
proxy_request = request.with_policy(proxy_policy)
proxy_request.env["HTTP_PROXY"] = "http://127.0.0.1:48123"
proxy_request.env["HTTPS_PROXY"] = "http://127.0.0.1:48123"
return SimpleNamespace(request=proxy_request, cleanup=_fake_cleanup)
async def _fake_run_under_backend(request, *, runtime=None):
seen["policy"] = request.policy
seen["env"] = request.env
return SimpleNamespace(
returncode=0,
stdout="ok\n",
stderr="",
timed_out=False,
backend_notes=(),
)
monkeypatch.setattr(code_exec, "_resolve_python_bin", lambda *, sandbox_enabled: sys.executable)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
monkeypatch.setattr(
code_exec,
"preflight_subprocess_managed_network",
_fake_preflight_subprocess_managed_network,
)
monkeypatch.setattr(
code_exec,
"prepare_subprocess_managed_network_proxy",
_fake_prepare_subprocess_managed_network_proxy,
)
monkeypatch.setattr(code_exec, "run_under_backend", _fake_run_under_backend)
token = current_tool_context.set(
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,
workspace=str(tmp_path),
),
)
)
try:
result = json.loads(await code_exec.execute_code("print('no network token here')"))
finally:
current_tool_context.reset(token)
reset_runtime()
reset_approval_queue()
assert result["exit_code"] == 0, result
assert prepare_calls
assert prepare_calls[0].policy.network is NetworkMode.PROXY_ALLOWLIST
assert seen["policy"].network is NetworkMode.PROXY_ALLOWLIST
env = seen["env"]
assert isinstance(env, dict)
assert env["HTTP_PROXY"] == "http://127.0.0.1:48123"
def test_trusted_windows_tools_collapse_host_network_to_managed_proxy(
tmp_path: Path,
) -> None:
from opensquilla.sandbox.types import (
NetworkMode,
ResourceLimits,
SandboxPolicy,
SecurityLevel,
)
from opensquilla.tools.builtin import code_exec, shell
policy = SandboxPolicy(
level=SecurityLevel.STANDARD,
network=NetworkMode.HOST,
mounts=(),
workspace_rw=True,
tmp_writable=True,
limits=ResourceLimits(),
env_allowlist=("PATH",),
require_approval=False,
)
runtime = SimpleNamespace(
backend=SimpleNamespace(name="windows_default"),
settings=SimpleNamespace(network_default="proxy_allowlist"),
)
token = current_tool_context.set(
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),
)
)
try:
shell_policy = shell._trusted_windows_managed_network_policy(policy, runtime)
code_policy = code_exec._trusted_windows_managed_network_policy(policy, runtime)
finally:
current_tool_context.reset(token)
assert shell_policy.network is NetworkMode.PROXY_ALLOWLIST
assert code_policy.network is NetworkMode.PROXY_ALLOWLIST
def test_windows_direct_powershell_argv_injects_proxy_defaults() -> None:
from opensquilla.tools.builtin import shell
argv = shell._windows_direct_powershell_argv(
"Invoke-WebRequest -UseBasicParsing https://example.com"
)
command = argv[-1]
assert "$PSDefaultParameterValues['Invoke-WebRequest:Proxy']" in command
assert "$PSDefaultParameterValues['Invoke-RestMethod:Proxy']" in command
assert "[System.Net.WebRequest]::DefaultWebProxy" in command
assert "Invoke-WebRequest -UseBasicParsing https://example.com" in command
assert "System.Net.Sockets.TcpClient" not in command
assert "Invoke-OpenSquillaProxyNetworkFallback" not in command
def test_windows_shell_host_handles_invoke_webrequest_status_via_managed_proxy(
tmp_path: Path,
) -> None:
from opensquilla.tools.builtin import shell
with _SingleResponseHttpProxy(
b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nexample"
) as proxy_port:
env = {
**os.environ,
"HTTP_PROXY": f"http://127.0.0.1:{proxy_port}",
"HTTPS_PROXY": f"http://127.0.0.1:{proxy_port}",
"NO_PROXY": "",
}
result = subprocess.run(
(
sys.executable,
"-c",
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
"powershell.exe",
(
"Invoke-WebRequest -UseBasicParsing http://example.test "
"| Select-Object -ExpandProperty StatusCode"
),
str(tmp_path),
str(tmp_path),
),
env=env,
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
assert result.stdout.strip() == "200"
def test_windows_shell_host_handles_try_wrapped_invoke_webrequest_status_via_managed_proxy(
tmp_path: Path,
) -> None:
from opensquilla.tools.builtin import shell
with _SingleResponseHttpProxy(
b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nexample"
) as proxy_port:
env = {
**os.environ,
"HTTP_PROXY": f"http://127.0.0.1:{proxy_port}",
"HTTPS_PROXY": f"http://127.0.0.1:{proxy_port}",
"NO_PROXY": "",
}
result = subprocess.run(
(
sys.executable,
"-c",
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
"missing-powershell.exe",
(
"try { Invoke-WebRequest -UseBasicParsing http://example.test "
"-TimeoutSec 15 | Select-Object -ExpandProperty StatusCode } "
"catch { Write-Output ($_.Exception.GetType().Name + ': ' + "
"$_.Exception.Message) }"
),
str(tmp_path),
str(tmp_path),
),
env=env,
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
assert result.stdout.strip() == "200"
def test_windows_shell_host_handles_assigned_invoke_webrequest_status_via_managed_proxy(
tmp_path: Path,
) -> None:
from opensquilla.tools.builtin import shell
with _SingleResponseHttpProxy(
b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nexample"
) as proxy_port:
env = {
**os.environ,
"HTTP_PROXY": f"http://127.0.0.1:{proxy_port}",
"HTTPS_PROXY": f"http://127.0.0.1:{proxy_port}",
"NO_PROXY": "",
}
result = subprocess.run(
(
sys.executable,
"-c",
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
"missing-powershell.exe",
(
"try { $r = Invoke-WebRequest -UseBasicParsing http://example.test "
"-TimeoutSec 15; Write-Output ('StatusCode=' + $r.StatusCode) } "
"catch { Write-Output ($_.Exception.GetType().Name + ': ' + "
"$_.Exception.Message) }"
),
str(tmp_path),
str(tmp_path),
),
env=env,
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
assert result.stdout.strip() == "StatusCode=200"
def test_windows_shell_host_handles_assigned_curl_head_status_via_managed_proxy(
tmp_path: Path,
) -> None:
from opensquilla.tools.builtin import shell
with _SingleResponseHttpProxy(
b"HTTP/1.1 200 OK\r\nX-Test: yes\r\n\r\n"
) as proxy_port:
env = {
**os.environ,
"HTTP_PROXY": f"http://127.0.0.1:{proxy_port}",
"HTTPS_PROXY": f"http://127.0.0.1:{proxy_port}",
"NO_PROXY": "",
}
result = subprocess.run(
(
sys.executable,
"-c",
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
"missing-powershell.exe",
(
"try { $r = curl.exe -I http://example.test 2>&1 | "
"Select-String 'HTTP/'; Write-Output $r } "
"catch { Write-Output ('ERROR: ' + $_.Exception.Message) }"
),
str(tmp_path),
str(tmp_path),
),
env=env,
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
assert result.stdout.strip() == "HTTP/1.1 200 OK"
def test_windows_shell_host_handles_curl_head_via_managed_proxy(
tmp_path: Path,
) -> None:
from opensquilla.tools.builtin import shell
with _SingleResponseHttpProxy(
b"HTTP/1.1 204 No Content\r\nX-Test: yes\r\n\r\n"
) as proxy_port:
env = {
**os.environ,
"HTTP_PROXY": f"http://127.0.0.1:{proxy_port}",
"HTTPS_PROXY": f"http://127.0.0.1:{proxy_port}",
"NO_PROXY": "",
}
result = subprocess.run(
(
sys.executable,
"-c",
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
"powershell.exe",
"curl.exe -I http://example.test",
str(tmp_path),
str(tmp_path),
),
env=env,
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
assert "HTTP/1.1 204 No Content" in result.stdout
assert "X-Test: yes" in result.stdout
def test_windows_shell_host_handles_remove_item_then_test_path_without_powershell(
tmp_path: Path,
) -> None:
from opensquilla.tools.builtin import shell
target = tmp_path / "delete-me.txt"
target.write_text("x", encoding="utf-8")
result = subprocess.run(
(
sys.executable,
"-c",
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
"missing-powershell.exe",
f"Remove-Item '{target}' -Force; Test-Path '{target}'",
str(tmp_path),
str(tmp_path),
),
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
assert result.stdout.strip() == "False"
assert not target.exists()
@pytest.mark.skipif(
not sys.platform.startswith("win") or shutil.which("powershell.exe") is None,
reason="requires native Windows PowerShell",
)
def test_windows_shell_host_leaves_output_expressions_to_powershell(
tmp_path: Path,
) -> None:
from opensquilla.tools.builtin import shell
result = subprocess.run(
(
sys.executable,
"-c",
shell._WINDOWS_SANDBOX_SHELL_HOST_CODE,
"powershell.exe",
"Write-Output ('HTTP_PROXY=' + $env:HTTP_PROXY)",
str(tmp_path),
str(tmp_path),
),
env={**os.environ, "HTTP_PROXY": "http://127.0.0.1:48123"},
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
assert result.stdout.strip() == "HTTP_PROXY=http://127.0.0.1:48123"
class _SingleResponseHttpProxy:
def __init__(self, response: bytes) -> None:
self._response = response
self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._socket.bind(("127.0.0.1", 0))
self._socket.listen(1)
self._socket.settimeout(5)
self.port = int(self._socket.getsockname()[1])
self._thread = threading.Thread(target=self._serve_once, daemon=True)
def __enter__(self) -> int:
self._thread.start()
return self.port
def __exit__(self, *args: object) -> None:
self._thread.join(timeout=5)
self._socket.close()
def _serve_once(self) -> None:
try:
conn, _addr = self._socket.accept()
except OSError:
return
with conn:
conn.settimeout(5)
data = b""
while b"\r\n\r\n" not in data:
chunk = conn.recv(4096)
if not chunk:
break
data += chunk
conn.sendall(self._response)
def test_windows_direct_powershell_argv_does_not_install_socket_fallbacks() -> None:
from opensquilla.tools.builtin import shell
argv = shell._windows_direct_powershell_argv(
"curl.exe -I https://example.com"
)
command = argv[-1]
assert "function Invoke-WebRequest" not in command
assert "function curl.exe" not in command
assert "Invoke-OpenSquillaProxyNetworkFallback" not in command
assert "System.Net.Sockets.TcpClient" not in command
@pytest.mark.asyncio
async def test_code_exec_backend_request_preserves_resolved_run_mode(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import SandboxRequest
from opensquilla.tools.builtin import code_exec, shell
reset_approval_queue()
configure_runtime(
SandboxSettings(
run_mode="trusted",
backend="noop",
allow_legacy_mode=True,
network_default="none",
),
workspace=tmp_path,
)
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=0,
stdout="ok\n",
stderr="",
timed_out=False,
backend_notes=(),
)
monkeypatch.setattr(code_exec, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
token = current_tool_context.set(
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,
workspace=str(tmp_path),
),
)
)
try:
result = json.loads(await code_exec.execute_code("print('ok')"))
finally:
current_tool_context.reset(token)
reset_runtime()
reset_approval_queue()
assert result["stdout"] == "ok\n"
assert backend_calls
assert backend_calls[0].run_mode == "trusted"
@pytest.mark.asyncio
async def test_shell_network_command_passes_network_hint(monkeypatch: pytest.MonkeyPatch) -> None:
from opensquilla.tools.builtin import shell
calls: list[dict[str, object]] = []
class _Runtime:
effective = SimpleNamespace(sandbox_enabled=True)
async def _fake_gate_action(**kwargs):
calls.append(kwargs)
policy = SimpleNamespace()
request = SimpleNamespace(cwd="/tmp", action_kind="shell.exec", policy=policy)
return object(), policy, request
async def _fake_run_under_backend(request, *, runtime=None):
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
monkeypatch.setattr(shell, "get_runtime", lambda: _Runtime())
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
result = await shell.exec_command("curl https://example.com")
assert "ok" in result
assert len(calls) == 1
hints = calls[0]["hints"]
assert hints.needs_network is True
assert hints.high_impact is False
@pytest.mark.asyncio
async def test_powershell_invoke_webrequest_passes_network_hint(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
calls: list[dict[str, object]] = []
class _Runtime:
effective = SimpleNamespace(sandbox_enabled=True)
async def _fake_gate_action(**kwargs):
calls.append(kwargs)
policy = SimpleNamespace()
request = SimpleNamespace(cwd="/tmp", action_kind="shell.exec", policy=policy)
return object(), policy, request
async def _fake_run_under_backend(request, *, runtime=None):
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
monkeypatch.setattr(shell, "get_runtime", lambda: _Runtime())
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
result = await shell.exec_command(
'powershell -NoProfile -Command "Invoke-WebRequest -UseBasicParsing https://example.com"'
)
assert "ok" in result
assert len(calls) == 1
hints = calls[0]["hints"]
assert hints.needs_network is True
assert hints.high_impact is False
@pytest.mark.asyncio
async def test_shell_url_text_does_not_pass_network_hint(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
calls: list[dict[str, object]] = []
class _Runtime:
effective = SimpleNamespace(sandbox_enabled=True)
async def _fake_gate_action(**kwargs):
calls.append(kwargs)
policy = SimpleNamespace()
request = SimpleNamespace(cwd="/tmp", action_kind="shell.exec", policy=policy)
return object(), policy, request
async def _fake_run_under_backend(request, *, runtime=None):
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
monkeypatch.setattr(shell, "get_runtime", lambda: _Runtime())
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
result = await shell.exec_command("echo https://example.com")
assert "ok" in result
assert len(calls) == 1
hints = calls[0]["hints"]
assert hints.needs_network is False
assert hints.high_impact is False
@pytest.mark.asyncio
async def test_code_with_url_literal_passes_network_hint(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import code_exec, shell
calls: list[dict[str, object]] = []
class _Runtime:
effective = SimpleNamespace(sandbox_enabled=True)
workspace = tmp_path
async def _fake_gate_action(**kwargs):
calls.append(kwargs)
policy = SimpleNamespace()
request = SimpleNamespace(cwd="/tmp", action_kind="code.exec", policy=policy)
return object(), policy, request
async def _fake_run_under_backend(request, *, runtime=None):
return SimpleNamespace(
returncode=0,
stdout="ok\n",
stderr="",
timed_out=False,
backend_notes=(),
)
monkeypatch.setattr(code_exec, "get_runtime", lambda: _Runtime())
monkeypatch.setattr(code_exec, "gate_action", _fake_gate_action)
monkeypatch.setattr(code_exec, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(code_exec, "_resolve_python_bin", lambda *, sandbox_enabled: sys.executable)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
result = json.loads(
await code_exec.execute_code('import requests\nrequests.get("https://example.com")')
)
assert result["stdout"] == "ok\n"
assert len(calls) == 1
hints = calls[0]["hints"]
assert hints.needs_network is True
assert hints.high_impact is False
@pytest.mark.asyncio
async def test_code_plain_url_literal_does_not_pass_network_hint(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import code_exec, shell
calls: list[dict[str, object]] = []
class _Runtime:
effective = SimpleNamespace(sandbox_enabled=True)
workspace = tmp_path
async def _fake_gate_action(**kwargs):
calls.append(kwargs)
policy = SimpleNamespace()
request = SimpleNamespace(cwd="/tmp", action_kind="code.exec", policy=policy)
return object(), policy, request
async def _fake_run_under_backend(request, *, runtime=None):
return SimpleNamespace(
returncode=0,
stdout="ok\n",
stderr="",
timed_out=False,
backend_notes=(),
)
monkeypatch.setattr(code_exec, "get_runtime", lambda: _Runtime())
monkeypatch.setattr(code_exec, "gate_action", _fake_gate_action)
monkeypatch.setattr(code_exec, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(code_exec, "_resolve_python_bin", lambda *, sandbox_enabled: sys.executable)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
result = json.loads(await code_exec.execute_code('print("https://example.com")'))
assert result["stdout"] == "ok\n"
assert len(calls) == 1
hints = calls[0]["hints"]
assert hints.needs_network is False
assert hints.high_impact is False
@pytest.mark.asyncio
async def test_shell_unknown_explicit_url_runs_with_managed_proxy(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
backend_requests: list[object] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_requests.append(request)
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
result = await shell.exec_command(
"curl https://unknown.test/path",
workdir=str(managed_runtime),
)
finally:
current_tool_context.reset(token)
assert "exit_code=0" in result
assert "ok" in result
assert len(backend_requests) == 1
request = backend_requests[0]
assert request.policy.network.value == "proxy_allowlist"
assert request.policy.network_proxy is not None
assert request.env["HTTP_PROXY"].startswith("http://127.0.0.1:")
assert get_approval_queue().list_pending("exec") == []
@pytest.mark.asyncio
async def test_windows_proxy_allowlist_runtime_skips_platform_network_boundary(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import (
NetworkMode,
NetworkProxySpec,
ResourceLimits,
SandboxPolicy,
SandboxRequest,
SecurityLevel,
)
events = []
async def fake_prepare_boundary(request, runtime):
events.append(("prepare", request.policy.network_proxy.port))
return "ctx"
async def fake_cleanup_boundary(ctx):
events.append(("cleanup", ctx))
class Backend:
name = "windows_default"
async def run(self, request):
events.append(("run", request.policy.network_proxy.port))
return SimpleNamespace(returncode=0, stdout="ok", stderr="", backend_notes=())
policy = SandboxPolicy(
level=SecurityLevel.STANDARD,
network=NetworkMode.PROXY_ALLOWLIST,
network_proxy=NetworkProxySpec(host="127.0.0.1", port=48123),
mounts=(),
workspace_rw=True,
tmp_writable=True,
limits=ResourceLimits(wall_timeout_s=30),
env_allowlist=("PATH",),
require_approval=False,
)
request = SandboxRequest(
argv=("python", "-m", "pip", "install", "humanize"),
cwd=tmp_path,
action_kind="shell.exec",
policy=policy,
env={"PATH": "x"},
)
runtime = SimpleNamespace(backend=Backend())
monkeypatch.setattr(
integration_mod,
"_prepare_platform_network_boundary",
fake_prepare_boundary,
)
monkeypatch.setattr(
integration_mod,
"_cleanup_platform_network_boundary",
fake_cleanup_boundary,
)
result = await integration_mod.run_under_backend(request, runtime=runtime)
assert result.stdout == "ok"
assert events == [("run", 48123)]
@pytest.mark.asyncio
async def test_windows_unready_proxy_allowlist_blocks_network_workarounds(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import (
NetworkMode,
ResourceLimits,
SandboxPolicy,
SandboxRequest,
SecurityLevel,
)
class _Ledger:
async def record_denial(self, *args: object, **kwargs: object) -> None:
return None
policy = SandboxPolicy(
level=SecurityLevel.STANDARD,
network=NetworkMode.PROXY_ALLOWLIST,
mounts=(),
workspace_rw=True,
tmp_writable=True,
limits=ResourceLimits(wall_timeout_s=30),
env_allowlist=("PATH",),
require_approval=False,
)
request = SandboxRequest(
argv=("powershell", "-Command", "python -m pip install humanize"),
cwd=tmp_path,
action_kind="shell.exec",
policy=policy,
env={"PATH": r"C:\Windows\System32"},
)
runtime = SimpleNamespace(
backend=SimpleNamespace(name="windows_default"),
workspace=tmp_path,
ledger=_Ledger(),
)
monkeypatch.setattr(
integration_mod,
"_windows_proxy_allowlist_enforced",
lambda runtime: False,
)
result = await integration_mod.preflight_subprocess_managed_network(
request,
runtime,
)
assert result is not None
assert not isinstance(result, dict)
assert result.retryable is False
assert "Windows sandbox managed network is unavailable" in result.message
assert "PROXY_ALLOWLIST" in result.message
assert "Do not retry with http_request" in result.message
assert "Do not retry with web_fetch" in result.message
assert "Do not retry with offline wheel downloads" in result.message
assert "Do not retry with host Python" in result.message
@pytest.mark.asyncio
async def test_windows_proxy_allowlist_preflight_does_not_repair_during_command(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import (
NetworkMode,
ResourceLimits,
SandboxPolicy,
SandboxRequest,
SecurityLevel,
)
class _Ledger:
def __init__(self) -> None:
self.denials: list[tuple[object, ...]] = []
async def record_denial(self, *args: object, **kwargs: object) -> None:
self.denials.append(args)
policy = SandboxPolicy(
level=SecurityLevel.STANDARD,
network=NetworkMode.PROXY_ALLOWLIST,
mounts=(),
workspace_rw=True,
tmp_writable=True,
limits=ResourceLimits(wall_timeout_s=30),
env_allowlist=("PATH",),
require_approval=False,
)
request = SandboxRequest(
argv=("powershell", "-Command", "curl -I https://example.com"),
cwd=tmp_path,
action_kind="shell.exec",
policy=policy,
env={"PATH": r"C:\Windows\System32"},
)
runtime = SimpleNamespace(
backend=SimpleNamespace(name="windows_default"),
workspace=tmp_path,
settings=SandboxSettings(run_mode="trusted", backend="windows_default"),
ledger=_Ledger(),
)
ledger = _Ledger()
runtime.ledger = ledger
monkeypatch.setattr(
integration_mod,
"_windows_proxy_allowlist_enforced",
lambda runtime: False,
)
async def _fake_ensure_windows_proxy_allowlist_setup(runtime):
pytest.fail("command preflight must not repair Windows setup or trigger elevation")
monkeypatch.setattr(
integration_mod,
"_ensure_windows_proxy_allowlist_setup",
_fake_ensure_windows_proxy_allowlist_setup,
raising=False,
)
token = current_tool_context.set(
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,
workspace=str(tmp_path),
),
)
)
try:
result = await integration_mod.preflight_subprocess_managed_network(
request,
runtime,
)
finally:
current_tool_context.reset(token)
assert result is not None
assert not isinstance(result, dict)
assert result.retryable is False
assert "Windows sandbox managed network is unavailable" in result.message
assert ledger.denials
@pytest.mark.asyncio
async def test_windows_ready_proxy_allowlist_preflight_continues_to_proxy_runtime(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import (
NetworkMode,
ResourceLimits,
SandboxPolicy,
SandboxRequest,
SecurityLevel,
)
class _Ledger:
async def record_denial(self, *args: object, **kwargs: object) -> None:
pytest.fail("ready Windows network backend should not record denial")
policy = SandboxPolicy(
level=SecurityLevel.STANDARD,
network=NetworkMode.PROXY_ALLOWLIST,
mounts=(),
workspace_rw=True,
tmp_writable=True,
limits=ResourceLimits(wall_timeout_s=30),
env_allowlist=("PATH",),
require_approval=False,
)
request = SandboxRequest(
argv=("powershell", "-Command", "curl -I https://example.com"),
cwd=tmp_path,
action_kind="shell.exec",
policy=policy,
env={"PATH": r"C:\Windows\System32"},
)
runtime = SimpleNamespace(
backend=SimpleNamespace(name="windows_default"),
workspace=tmp_path,
ledger=_Ledger(),
)
monkeypatch.setattr(
integration_mod,
"_windows_proxy_allowlist_enforced",
lambda runtime: True,
)
token = current_tool_context.set(
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,
workspace=str(tmp_path),
),
)
)
try:
result = await integration_mod.preflight_subprocess_managed_network(
request,
runtime,
)
finally:
current_tool_context.reset(token)
assert result is None
@pytest.mark.asyncio
async def test_shell_package_install_queues_bundle_approval_before_proxy_run(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
async def _fail_run_under_backend(request, *, runtime=None):
pytest.fail("package bundle approval should run before proxy execution")
monkeypatch.setattr(shell, "run_under_backend", _fail_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
payload = json.loads(
await shell.exec_command(
"pip install requests",
workdir=str(managed_runtime),
)
)
finally:
current_tool_context.reset(token)
assert payload["status"] == "approval_required"
assert payload["approvalKind"] == "sandbox_network"
assert payload["bundle_id"] == "python-package-install"
pending = get_approval_queue().list_pending("exec")
assert len(pending) == 1
assert pending[0]["params"]["bundle_id"] == "python-package-install"
@pytest.mark.asyncio
async def test_uv_pip_install_queues_bundle_approval_before_proxy_run(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
async def _fail_run_under_backend(request, *, runtime=None):
pytest.fail("uv pip package bundle approval should run before proxy execution")
monkeypatch.setattr(shell, "run_under_backend", _fail_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
payload = json.loads(
await shell.exec_command(
"uv pip install --no-cache-dir httpx[http2] pendulum",
workdir=str(managed_runtime),
)
)
finally:
current_tool_context.reset(token)
assert payload["status"] == "approval_required"
assert payload["approvalKind"] == "sandbox_network"
assert payload["bundle_id"] == "python-package-install"
pending = get_approval_queue().list_pending("exec")
assert len(pending) == 1
assert pending[0]["params"]["bundle_id"] == "python-package-install"
@pytest.mark.asyncio
async def test_poetry_install_queues_python_bundle_before_proxy_run(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.tools.builtin import shell
profile_calls: list[tuple[str, ...]] = []
async def _fail_run_under_backend(request, *, runtime=None):
pytest.fail("poetry package bundle approval should run before proxy execution")
def _fake_capability_profile_for_command(argv):
profile_calls.append(tuple(argv))
return SimpleNamespace(package_bundles=("python-package-install",))
monkeypatch.setattr(
integration_mod,
"capability_profile_for_command",
_fake_capability_profile_for_command,
)
monkeypatch.setattr(shell, "run_under_backend", _fail_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
payload = json.loads(
await shell.exec_command(
"poetry install",
workdir=str(managed_runtime),
)
)
finally:
current_tool_context.reset(token)
assert profile_calls == [("sh", "-lc", "poetry install")]
assert payload["status"] == "approval_required"
assert payload["approvalKind"] == "sandbox_network"
assert payload["bundle_id"] == "python-package-install"
pending = get_approval_queue().list_pending("exec")
assert len(pending) == 1
assert pending[0]["params"]["bundle_id"] == "python-package-install"
@pytest.mark.asyncio
async def test_composer_install_queues_php_bundle_before_proxy_run(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.tools.builtin import shell
profile_calls: list[tuple[str, ...]] = []
async def _fail_run_under_backend(request, *, runtime=None):
pytest.fail("composer package bundle approval should run before proxy execution")
def _fake_capability_profile_for_command(argv):
profile_calls.append(tuple(argv))
return SimpleNamespace(package_bundles=("php-package-install",))
monkeypatch.setattr(
integration_mod,
"capability_profile_for_command",
_fake_capability_profile_for_command,
)
monkeypatch.setattr(shell, "run_under_backend", _fail_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
payload = json.loads(
await shell.exec_command(
"composer install",
workdir=str(managed_runtime),
)
)
finally:
current_tool_context.reset(token)
assert profile_calls == [("sh", "-lc", "composer install")]
assert payload["status"] == "approval_required"
assert payload["approvalKind"] == "sandbox_network"
assert payload["bundle_id"] == "php-package-install"
pending = get_approval_queue().list_pending("exec")
assert len(pending) == 1
assert pending[0]["params"]["bundle_id"] == "php-package-install"
@pytest.mark.asyncio
async def test_trusted_uv_pip_install_receives_managed_proxy_without_prompt(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.tools.builtin import shell
class _FakeProxyServer:
host = "127.0.0.1"
port = 48123
def __init__(self, *args, **kwargs) -> None:
return None
async def start(self) -> None:
return None
async def stop(self) -> None:
return None
seen: dict[str, object] = {}
async def _fake_run_under_backend(request, *, runtime=None):
managed = await integration_mod.prepare_subprocess_managed_network_proxy(
request,
runtime=runtime,
)
try:
seen["env"] = managed.request.env
seen["policy"] = managed.request.policy
return SimpleNamespace(returncode=0, stdout="installed\n", stderr="", backend_notes=())
finally:
await managed.cleanup()
monkeypatch.setattr(integration_mod, "SandboxProxyServer", _FakeProxyServer)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command(
"uv pip install --no-cache-dir httpx[http2] pendulum",
workdir=str(managed_runtime),
)
finally:
current_tool_context.reset(token)
assert "installed" in result
assert get_approval_queue().list_pending("exec") == []
env = seen["env"]
assert isinstance(env, dict)
assert env["HTTP_PROXY"] == "http://127.0.0.1:48123"
assert env["HTTPS_PROXY"] == env["HTTP_PROXY"]
assert env["npm_config_proxy"] == env["HTTP_PROXY"]
assert env["NODE_USE_ENV_PROXY"] == "1"
assert env["CODEX_NETWORK_PROXY_ACTIVE"] == "1"
assert env["CODEX_NETWORK_ALLOW_LOCAL_BINDING"] == "0"
assert env["OPENSQUILLA_SANDBOX_NETWORK"] == "proxy_allowlist"
assert "GIT_CONFIG_KEY_0" not in env
assert "GIT_CONFIG_VALUE_0" not in env
@pytest.mark.asyncio
async def test_trusted_python_explicit_url_uses_managed_proxy_before_execution(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import NetworkMode, SandboxRequest
from opensquilla.tools.builtin import shell
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="", backend_notes=())
async def _fake_cleanup() -> None:
return None
async def _fake_prepare_subprocess_managed_network_proxy(request, *, runtime=None):
return SimpleNamespace(request=request, cleanup=_fake_cleanup)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"prepare_subprocess_managed_network_proxy",
_fake_prepare_subprocess_managed_network_proxy,
)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command(
"python - <<'PY'\n"
"import urllib.request\n"
"urllib.request.urlopen('https://example.com')\n"
"PY",
workdir=str(managed_runtime),
)
finally:
current_tool_context.reset(token)
assert "ok" in result
assert len(backend_calls) == 1
assert backend_calls[0].policy.network is NetworkMode.PROXY_ALLOWLIST
@pytest.mark.asyncio
async def test_trusted_python_caught_network_error_still_uses_proxy_before_execution(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import NetworkMode, SandboxRequest
from opensquilla.tools.builtin import shell
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(returncode=0, stdout="handled\n", stderr="", backend_notes=())
async def _fake_cleanup() -> None:
return None
async def _fake_prepare_subprocess_managed_network_proxy(request, *, runtime=None):
return SimpleNamespace(request=request, cleanup=_fake_cleanup)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"prepare_subprocess_managed_network_proxy",
_fake_prepare_subprocess_managed_network_proxy,
)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command(
"python - <<'PY'\n"
"import urllib.request\n"
"try:\n"
" urllib.request.urlopen('https://httpbin.org/get', timeout=10)\n"
"except Exception as e:\n"
" print(type(e).__name__ + ': ' + str(e))\n"
"PY",
workdir=str(managed_runtime),
)
finally:
current_tool_context.reset(token)
assert "handled" in result
assert len(backend_calls) == 1
assert backend_calls[0].policy.network is NetworkMode.PROXY_ALLOWLIST
@pytest.mark.asyncio
async def test_trusted_npm_view_uses_managed_proxy_before_execution(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import NetworkMode, SandboxRequest
from opensquilla.tools.builtin import shell
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(returncode=0, stdout="4.17.21\n", stderr="", backend_notes=())
async def _fake_cleanup() -> None:
return None
async def _fake_prepare_subprocess_managed_network_proxy(request, *, runtime=None):
return SimpleNamespace(request=request, cleanup=_fake_cleanup)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"prepare_subprocess_managed_network_proxy",
_fake_prepare_subprocess_managed_network_proxy,
)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("npm view lodash version", workdir=str(managed_runtime))
finally:
current_tool_context.reset(token)
assert "4.17.21" in result
assert len(backend_calls) == 1
assert backend_calls[0].policy.network is NetworkMode.PROXY_ALLOWLIST
@pytest.mark.asyncio
async def test_trusted_unknown_install_uses_managed_proxy_without_redundant_retry(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import SandboxRequest
from opensquilla.tools.builtin import shell
backend_calls: list[SandboxRequest] = []
cleanup_calls = 0
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
if len(backend_calls) == 1:
return SimpleNamespace(
returncode=1,
stdout="",
stderr="curl: (6) Could not resolve host: pypi.org\n",
backend_notes=(),
)
return SimpleNamespace(
returncode=0,
stdout="installed\n",
stderr="",
backend_notes=(),
)
async def _fake_prepare_subprocess_managed_network_proxy(request, *, runtime=None):
managed_env = dict(request.env)
managed_env["HTTP_PROXY"] = "http://127.0.0.1:48123"
managed_env["HTTPS_PROXY"] = managed_env["HTTP_PROXY"]
managed_request = SandboxRequest(
argv=request.argv,
cwd=request.cwd,
action_kind=request.action_kind,
policy=request.policy,
stdin=request.stdin,
env=managed_env,
reason=request.reason,
)
async def _cleanup() -> None:
nonlocal cleanup_calls
cleanup_calls += 1
return SimpleNamespace(request=managed_request, cleanup=_cleanup)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"prepare_subprocess_managed_network_proxy",
_fake_prepare_subprocess_managed_network_proxy,
)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("pip install demo", workdir=str(managed_runtime))
finally:
current_tool_context.reset(token)
assert "Could not resolve host: pypi.org" in result
assert len(backend_calls) == 1
assert backend_calls[0].env["HTTP_PROXY"] == "http://127.0.0.1:48123"
assert cleanup_calls == 1
@pytest.mark.asyncio
async def test_standard_network_failure_does_not_return_package_bundle_recovery_approval(
standard_runtime_no_preflight: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
backend_calls = 0
async def _fake_run_under_backend(request, *, runtime=None):
nonlocal backend_calls
backend_calls += 1
return SimpleNamespace(
returncode=1,
stdout="",
stderr="curl: (6) Could not resolve host: pypi.org\n",
backend_notes=(),
)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(standard_runtime_no_preflight),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
result = await shell.exec_command(
"pip install demo",
workdir=str(standard_runtime_no_preflight),
)
finally:
current_tool_context.reset(token)
assert backend_calls == 1
assert "exit_code=1" in result
assert "Could not resolve host: pypi.org" in result
assert get_approval_queue().list_pending("exec") == []
@pytest.mark.asyncio
async def test_standard_network_failure_does_not_retry_explicit_url(
standard_runtime_no_preflight: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
backend_calls = 0
async def _fake_run_under_backend(request, *, runtime=None):
nonlocal backend_calls
backend_calls += 1
return SimpleNamespace(
returncode=1,
stdout="",
stderr="curl: (6) Could not resolve host: unknown.test\n",
backend_notes=(),
)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(standard_runtime_no_preflight),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
result = await shell.exec_command(
"curl https://unknown.test/path",
workdir=str(standard_runtime_no_preflight),
)
finally:
current_tool_context.reset(token)
assert backend_calls == 1
assert "exit_code=1" in result
assert "Could not resolve host: unknown.test" in result
assert get_approval_queue().list_pending("exec") == []
@pytest.mark.asyncio
async def test_standard_network_failure_does_not_retry_with_approved_bundle(
standard_runtime_no_preflight: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import NetworkMode, SandboxRequest
from opensquilla.tools.builtin import shell
backend_calls: list[SandboxRequest] = []
cleanup_calls = 0
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
if len(backend_calls) == 1:
return SimpleNamespace(
returncode=1,
stdout="",
stderr="curl: (6) Could not resolve host: pypi.org\n",
backend_notes=(),
)
return SimpleNamespace(returncode=0, stdout="installed\n", stderr="", backend_notes=())
async def _fake_prepare_subprocess_managed_network_proxy(request, *, runtime=None):
managed_env = dict(request.env)
managed_env["HTTP_PROXY"] = "http://127.0.0.1:48123"
managed_env["HTTPS_PROXY"] = managed_env["HTTP_PROXY"]
managed_request = SandboxRequest(
argv=request.argv,
cwd=request.cwd,
action_kind=request.action_kind,
policy=request.policy,
stdin=request.stdin,
env=managed_env,
reason=request.reason,
)
async def _cleanup() -> None:
nonlocal cleanup_calls
cleanup_calls += 1
return SimpleNamespace(request=managed_request, cleanup=_cleanup)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"prepare_subprocess_managed_network_proxy",
_fake_prepare_subprocess_managed_network_proxy,
)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(standard_runtime_no_preflight),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(
run_mode=RunMode.STANDARD,
workspace=str(standard_runtime_no_preflight),
bundles=(PackageBundleGrant(bundle_id="python-package-install"),),
),
)
)
try:
result = await shell.exec_command(
"pip install demo",
workdir=str(standard_runtime_no_preflight),
)
finally:
current_tool_context.reset(token)
assert "Could not resolve host: pypi.org" in result
assert len(backend_calls) == 1
assert cleanup_calls == 0
assert backend_calls[0].policy.network is NetworkMode.NONE
assert get_approval_queue().list_pending("exec") == []
@pytest.mark.asyncio
async def test_standard_network_failure_does_not_consume_allow_once_grant(
standard_runtime_no_preflight: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import (
NetworkMode,
ResourceLimits,
SandboxPolicy,
SandboxRequest,
SecurityLevel,
)
from opensquilla.tools.builtin import shell
command = "curl https://unknown.test/path"
workdir = standard_runtime_no_preflight.resolve(strict=False)
base_policy = SandboxPolicy(
level=SecurityLevel.STANDARD,
network=NetworkMode.NONE,
mounts=(),
workspace_rw=True,
tmp_writable=True,
limits=ResourceLimits(),
env_allowlist=("PATH",),
require_approval=False,
)
approval_policy = SandboxPolicy(
level=base_policy.level,
network=NetworkMode.PROXY_ALLOWLIST,
mounts=base_policy.mounts,
workspace_rw=base_policy.workspace_rw,
tmp_writable=base_policy.tmp_writable,
limits=base_policy.limits,
env_allowlist=base_policy.env_allowlist,
require_approval=base_policy.require_approval,
description=base_policy.description,
network_proxy=base_policy.network_proxy,
)
approval_request = SandboxRequest(
argv=("sh", "-lc", command),
cwd=workdir,
action_kind="shell.exec",
policy=approval_policy,
env=dict(os.environ),
)
class _FakeProxyServer:
host = "127.0.0.1"
port = 48123
def __init__(self, decide, *args, **kwargs) -> None:
self._decide = decide
async def start(self) -> None:
decision = await self._decide.decide(
NetworkPolicyRequest(
protocol=NetworkProtocol.HTTPS_CONNECT,
host="unknown.test",
port=443,
method="CONNECT",
)
)
assert decision.status == "allow"
async def stop(self) -> None:
return None
async def _fake_gate_action(**kwargs):
request = SimpleNamespace(
cwd=workdir,
action_kind="shell.exec",
policy=base_policy,
reason="",
)
return object(), base_policy, request
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
if len(backend_calls) == 1:
return SimpleNamespace(
returncode=1,
stdout="",
stderr="curl: (6) Could not resolve host: unknown.test\n",
backend_notes=(),
)
return SimpleNamespace(returncode=0, stdout="downloaded\n", stderr="", backend_notes=())
monkeypatch.setattr(integration_mod, "SandboxProxyServer", _FakeProxyServer)
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
grant = TemporaryGrant(
kind="domain",
value="unknown.test",
fingerprint=integration_mod.action_fingerprint(approval_request),
)
run_context = RunContext(
run_mode=RunMode.STANDARD,
workspace=str(standard_runtime_no_preflight),
temporary_grants=(grant,),
)
tool_context = ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(standard_runtime_no_preflight),
session_key="s1",
run_mode="standard",
sandbox_run_context=run_context,
)
token = current_tool_context.set(tool_context)
try:
result = await shell.exec_command(command, workdir=str(standard_runtime_no_preflight))
finally:
current_tool_context.reset(token)
assert "Could not resolve host: unknown.test" in result
assert len(backend_calls) == 1
assert backend_calls[0].policy.network is NetworkMode.NONE
assert isinstance(tool_context.sandbox_run_context, RunContext)
assert tool_context.sandbox_run_context.temporary_grants == (grant,)
assert get_approval_queue().list_pending("exec") == []
@pytest.mark.asyncio
async def test_trusted_hostless_private_network_failure_does_not_retry(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
backend_calls = 0
async def _fake_run_under_backend(request, *, runtime=None):
nonlocal backend_calls
backend_calls += 1
return SimpleNamespace(
returncode=1,
stdout="",
stderr="Network is unreachable\n",
backend_notes=(),
)
async def _fake_preflight_subprocess_managed_network(request, runtime):
return None
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_sensitive_shell_block", lambda *args, **kwargs: None)
monkeypatch.setattr(
shell,
"preflight_subprocess_managed_network",
_fake_preflight_subprocess_managed_network,
)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command(
"curl http://127.0.0.1:8000/",
workdir=str(managed_runtime),
)
finally:
current_tool_context.reset(token)
assert backend_calls == 1
assert "Network is unreachable" in result
@pytest.mark.asyncio
async def test_trusted_metadata_target_is_not_auto_retried(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
backend_calls = 0
async def _fake_run_under_backend(request, *, runtime=None):
nonlocal backend_calls
backend_calls += 1
return SimpleNamespace(
returncode=1,
stdout="",
stderr="Network is unreachable\n",
backend_notes=(),
)
async def _fake_preflight_subprocess_managed_network(request, runtime):
return None
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_sensitive_shell_block", lambda *args, **kwargs: None)
monkeypatch.setattr(
shell,
"preflight_subprocess_managed_network",
_fake_preflight_subprocess_managed_network,
)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command(
"curl http://169.254.169.254/latest/meta-data/",
workdir=str(managed_runtime),
)
finally:
current_tool_context.reset(token)
assert backend_calls == 1
assert "exit_code=1" in result
assert "Network is unreachable" in result
assert "approval_required" not in result
@pytest.mark.asyncio
async def test_trusted_network_failure_does_not_retry_after_managed_proxy_execution(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import NetworkMode
from opensquilla.tools.builtin import shell
class _FakeProxyServer:
host = "127.0.0.1"
port = 48123
def __init__(self, *args, **kwargs) -> None:
return None
async def start(self) -> None:
return None
async def stop(self) -> None:
return None
async def _fake_gate_action(**kwargs):
_, policy, request = await integration_mod.gate_action(**kwargs)
policy = policy.__class__(
level=policy.level,
network=NetworkMode.NONE,
mounts=policy.mounts,
workspace_rw=policy.workspace_rw,
tmp_writable=policy.tmp_writable,
limits=policy.limits,
env_allowlist=policy.env_allowlist,
require_approval=policy.require_approval,
description=policy.description,
network_proxy=policy.network_proxy,
)
request = SimpleNamespace(
cwd=request.cwd,
action_kind=request.action_kind,
policy=policy,
reason=getattr(request, "reason", ""),
)
return object(), policy, request
backend_calls = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
if len(backend_calls) == 1:
return SimpleNamespace(
returncode=1,
stdout="",
stderr="curl: (6) Could not resolve host: pypi.org\n",
backend_notes=(),
)
return SimpleNamespace(returncode=0, stdout="installed\n", stderr="", backend_notes=())
monkeypatch.setattr(integration_mod, "SandboxProxyServer", _FakeProxyServer)
monkeypatch.setattr(shell, "gate_action", _fake_gate_action)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("pip install demo", workdir=str(managed_runtime))
finally:
current_tool_context.reset(token)
assert "Could not resolve host: pypi.org" in result
assert len(backend_calls) == 1
assert backend_calls[0].policy.network is NetworkMode.PROXY_ALLOWLIST
assert backend_calls[0].policy.network_proxy is not None
@pytest.mark.asyncio
async def test_trusted_normal_user_path_denial_escalates_without_retry(
managed_runtime: Path,
tmp_path_factory: pytest.TempPathFactory,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import SandboxRequest
from opensquilla.tools.builtin import shell
outside = tmp_path_factory.mktemp("outside-project")
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=1,
stdout="",
stderr="",
backend_notes=(f"mount denied: {outside}",),
)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_sensitive_shell_block", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_workdir_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_read_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_write_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("python -m pip install -e .", workdir=str(outside))
finally:
current_tool_context.reset(token)
assert json.loads(result)["status"] == "denied"
assert len(backend_calls) == 1
@pytest.mark.asyncio
async def test_trusted_read_path_denial_escalates_without_retry(
managed_runtime: Path,
tmp_path_factory: pytest.TempPathFactory,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import SandboxRequest
from opensquilla.tools.builtin import shell
outside = tmp_path_factory.mktemp("outside-read")
target = outside / "data.txt"
target.write_text("data\n", encoding="utf-8")
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=1,
stdout="",
stderr="",
backend_notes=(f"filesystem.read.denied: {target}",),
)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_sensitive_shell_block", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_read_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_write_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("pip install demo", workdir=str(managed_runtime))
finally:
current_tool_context.reset(token)
assert json.loads(result)["status"] == "denied"
assert len(backend_calls) == 1
@pytest.mark.asyncio
async def test_trusted_execve_path_denial_escalates_without_retry(
managed_runtime: Path,
tmp_path_factory: pytest.TempPathFactory,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import SandboxRequest
from opensquilla.tools.builtin import shell
outside = tmp_path_factory.mktemp("outside-exec")
target = outside / "tool"
target.write_text("#!/bin/sh\n", encoding="utf-8")
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=1,
stdout="",
stderr="",
backend_notes=(f"execve.denied: {target}",),
)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "_sensitive_shell_block", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_read_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_write_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("pip install demo", workdir=str(managed_runtime))
finally:
current_tool_context.reset(token)
assert json.loads(result)["status"] == "denied"
assert len(backend_calls) == 1
@pytest.mark.asyncio
async def test_trusted_managed_network_denial_escalates_without_retry(
managed_runtime: Path,
tmp_path_factory: pytest.TempPathFactory,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import SandboxRequest
from opensquilla.tools.builtin import shell
outside = tmp_path_factory.mktemp("outside-path-network")
backend_calls: list[SandboxRequest] = []
cleanup_calls = 0
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=1,
stdout="",
stderr="",
backend_notes=(f"mount denied: {outside}",),
)
async def _fake_prepare_subprocess_managed_network_proxy(request, *, runtime=None):
managed_env = dict(request.env)
managed_env["HTTP_PROXY"] = "http://127.0.0.1:48123"
managed_env["HTTPS_PROXY"] = managed_env["HTTP_PROXY"]
managed_request = SandboxRequest(
argv=request.argv,
cwd=request.cwd,
action_kind=request.action_kind,
policy=request.policy,
stdin=request.stdin,
env=managed_env,
reason=request.reason,
)
async def _cleanup() -> None:
nonlocal cleanup_calls
cleanup_calls += 1
return SimpleNamespace(request=managed_request, cleanup=_cleanup)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"prepare_subprocess_managed_network_proxy",
_fake_prepare_subprocess_managed_network_proxy,
)
monkeypatch.setattr(shell, "_sensitive_shell_block", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_workdir_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_read_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_write_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("pip install demo", workdir=str(outside))
finally:
current_tool_context.reset(token)
assert json.loads(result)["status"] == "denied"
assert len(backend_calls) == 1
assert cleanup_calls == 1
assert backend_calls[0].env["HTTP_PROXY"] == "http://127.0.0.1:48123"
@pytest.mark.asyncio
async def test_trusted_sensitive_path_denial_does_not_auto_retry(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import (
DenialReason,
DenialResult,
SandboxRequest,
SecurityLevel,
SuggestedNextStep,
)
from opensquilla.tools.builtin import shell
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=1,
stdout="",
stderr="",
backend_notes=("mount denied: /etc/passwd",),
)
async def _fake_escalate_backend_denial(*args, **kwargs):
return DenialResult(
reason=DenialReason.SEATBELT_DENIED,
suggested_next_step=SuggestedNextStep.ASK_USER,
level=SecurityLevel.STANDARD,
action_fingerprint="test",
message="denied",
retryable=False,
)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(shell, "escalate_backend_denial", _fake_escalate_backend_denial)
monkeypatch.setattr(shell, "_sensitive_shell_block", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_read_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(shell, "_sandbox_write_path_access_envelope", lambda *args, **kwargs: None)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("cat /etc/passwd", workdir=str(managed_runtime))
finally:
current_tool_context.reset(token)
assert json.loads(result)["status"] == "denied"
assert len(backend_calls) == 1
@pytest.mark.asyncio
async def test_trusted_successful_network_failure_text_does_not_retry(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
backend_calls = 0
async def _fake_run_under_backend(request, *, runtime=None):
nonlocal backend_calls
backend_calls += 1
return SimpleNamespace(
returncode=0,
stdout="Could not resolve host: pypi.org\n",
stderr="",
backend_notes=(),
)
monkeypatch.setattr(shell, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = await shell.exec_command("pip install demo", workdir=str(managed_runtime))
finally:
current_tool_context.reset(token)
assert backend_calls == 1
assert "Could not resolve host: pypi.org" in result
@pytest.mark.asyncio
async def test_timeout_wrapped_node_install_queues_bundle_approval_before_proxy_run(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.tools.builtin import shell
async def _fail_run_under_backend(request, *, runtime=None):
pytest.fail("node package bundle approval should run before proxy execution")
monkeypatch.setattr(shell, "run_under_backend", _fail_run_under_backend)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
payload = json.loads(
await shell.exec_command(
"timeout 30 npm install lodash",
workdir=str(managed_runtime),
)
)
finally:
current_tool_context.reset(token)
assert payload["status"] == "approval_required"
assert payload["approvalKind"] == "sandbox_network"
assert payload["bundle_id"] == "node-package-install"
pending = get_approval_queue().list_pending("exec")
assert len(pending) == 1
assert pending[0]["params"]["bundle_id"] == "node-package-install"
@pytest.mark.asyncio
async def test_subprocess_network_approval_uses_session_workspace_for_external_cwd(
managed_runtime: Path,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import (
NetworkMode,
ResourceLimits,
SandboxPolicy,
SandboxRequest,
SecurityLevel,
)
external = managed_runtime.parent / f"{managed_runtime.name}-external"
external.mkdir()
runtime = get_runtime()
assert runtime is not None
request = SandboxRequest(
argv=("sh", "-lc", "curl https://unknown.test/path"),
cwd=external,
action_kind="shell.exec",
policy=SandboxPolicy(
level=SecurityLevel.STANDARD,
network=NetworkMode.PROXY_ALLOWLIST,
mounts=(),
workspace_rw=True,
tmp_writable=True,
limits=ResourceLimits(),
env_allowlist=("PATH",),
require_approval=False,
),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(
run_mode=RunMode.STANDARD,
workspace=str(managed_runtime),
),
)
)
try:
payload = await integration_mod.preflight_subprocess_managed_network(
request,
runtime,
)
finally:
current_tool_context.reset(token)
assert payload is None
assert get_approval_queue().list_pending("exec") == []
@pytest.mark.asyncio
async def test_subprocess_network_once_grant_consumes_from_session_workspace(
managed_runtime: Path,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import (
NetworkMode,
ResourceLimits,
SandboxPolicy,
SandboxRequest,
SecurityLevel,
)
external = managed_runtime.parent / f"{managed_runtime.name}-external"
external.mkdir()
runtime = get_runtime()
assert runtime is not None
request = SandboxRequest(
argv=("sh", "-lc", "curl https://unknown.test/path"),
cwd=external,
action_kind="shell.exec",
policy=SandboxPolicy(
level=SecurityLevel.STANDARD,
network=NetworkMode.PROXY_ALLOWLIST,
mounts=(),
workspace_rw=True,
tmp_writable=True,
limits=ResourceLimits(),
env_allowlist=("PATH",),
require_approval=False,
),
)
grant = TemporaryGrant(
kind="domain",
value="unknown.test",
fingerprint=integration_mod.action_fingerprint(request),
)
ctx = ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(
run_mode=RunMode.STANDARD,
workspace=str(managed_runtime),
temporary_grants=(grant,),
),
)
token = current_tool_context.set(ctx)
try:
payload = await integration_mod.preflight_subprocess_managed_network(
request,
runtime,
)
finally:
current_tool_context.reset(token)
assert payload is None
assert isinstance(ctx.sandbox_run_context, RunContext)
assert ctx.sandbox_run_context.temporary_grants == (grant,)
@pytest.mark.asyncio
async def test_background_shell_network_spawn_receives_managed_proxy(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.tools.builtin import shell
class _FakeProxyServer:
host = "127.0.0.1"
port = 48123
def __init__(self, *args, **kwargs) -> None:
return None
async def start(self) -> None:
return None
async def stop(self) -> None:
return None
class _FakeStream:
async def read(self, size: int) -> bytes:
return b""
class _FakeProcess:
stdout = _FakeStream()
stdin = None
returncode = 0
async def wait(self) -> int:
return 0
seen: dict[str, object] = {}
async def _fake_spawn(*, runtime: object, request: object) -> object:
seen["policy"] = request.policy
seen["env"] = request.env
assert request.policy.network_proxy is not None
return shell._SpawnedBackgroundProcess(process=_FakeProcess()) # type: ignore[arg-type]
monkeypatch.setattr(shell, "_spawn_sandboxed_background_process", _fake_spawn)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
monkeypatch.setattr(integration_mod, "SandboxProxyServer", _FakeProxyServer)
monkeypatch.setattr(
shell,
"check_safe_bin",
lambda command: SimpleNamespace(allowed=True, needs_approval=False, reason=""),
)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(
run_mode=RunMode.STANDARD,
domains=(DomainGrant(domain="example.com"),),
),
)
)
try:
result = await shell.background_process(
"curl https://example.com",
workdir=str(managed_runtime),
timeout=5,
)
session_id = result.splitlines()[0].split("=", 1)[1]
session = shell._bg_sessions[session_id]
assert session.collector_task is not None
await session.collector_task
finally:
current_tool_context.reset(token)
assert "policy" in seen
env = seen["env"]
assert isinstance(env, dict)
assert env["HTTP_PROXY"].startswith("http://127.0.0.1:")
assert env["HTTPS_PROXY"] == env["HTTP_PROXY"]
assert "127.0.0.1" in env["NO_PROXY"]
assert env["PIP_PROXY"] == env["HTTP_PROXY"]
@pytest.mark.asyncio
async def test_code_network_subprocess_receives_managed_proxy_env(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.tools.builtin import code_exec, shell
class _FakeProxyServer:
host = "127.0.0.1"
port = 48123
def __init__(self, *args, **kwargs) -> None:
return None
async def start(self) -> None:
return None
async def stop(self) -> None:
return None
monkeypatch.setattr(code_exec, "_resolve_python_bin", lambda *, sandbox_enabled: sys.executable)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
monkeypatch.setattr(integration_mod, "SandboxProxyServer", _FakeProxyServer)
seen: dict[str, object] = {}
async def _fake_run_under_backend(request, *, runtime=None):
managed = await integration_mod.prepare_subprocess_managed_network_proxy(
request,
runtime=runtime,
)
try:
seen["env"] = managed.request.env
seen["policy"] = managed.request.policy
return SimpleNamespace(
returncode=0,
stdout="ok\n",
stderr="",
timed_out=False,
backend_notes=(),
)
finally:
await managed.cleanup()
monkeypatch.setattr(code_exec, "run_under_backend", _fake_run_under_backend)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(
run_mode=RunMode.STANDARD,
domains=(DomainGrant(domain="example.com"),),
),
)
)
try:
result = json.loads(
await code_exec.execute_code(
"\n".join(
(
"import os, socket",
"url = 'https://example.com/path'",
"socket.gethostname()",
"print(os.environ.get('HTTP_PROXY', ''))",
"print(os.environ.get('HTTPS_PROXY', ''))",
"print(os.environ.get('NO_PROXY', '<missing>'))",
)
)
)
)
finally:
current_tool_context.reset(token)
assert result["exit_code"] == 0, result
assert result["stdout"] == "ok\n"
env = seen["env"]
assert isinstance(env, dict)
assert env["HTTP_PROXY"].startswith("http://127.0.0.1:")
assert env["HTTPS_PROXY"] == env["HTTP_PROXY"]
assert "127.0.0.1" in env["NO_PROXY"]
assert env["npm_config_https_proxy"] == env["HTTP_PROXY"]
@pytest.mark.asyncio
async def test_code_exec_prepares_managed_proxy_before_backend_run(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
import dataclasses
from opensquilla.sandbox.types import NetworkProxySpec
from opensquilla.tools.builtin import code_exec, shell
class _Managed:
def __init__(self, request) -> None:
self.request = request
self.cleaned = False
async def cleanup(self) -> None:
self.cleaned = True
monkeypatch.setattr(code_exec, "_resolve_python_bin", lambda *, sandbox_enabled: sys.executable)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
seen: dict[str, object] = {}
managed_objects: list[_Managed] = []
async def _fake_preflight(request, runtime):
return None
async def _fake_prepare(request, *, runtime=None):
policy = dataclasses.replace(
request.policy,
network_proxy=NetworkProxySpec(host="127.0.0.1", port=48123),
)
managed = _Managed(
request.with_policy(policy).with_policy(policy)
)
managed.request.env["HTTP_PROXY"] = "http://127.0.0.1:48123"
managed.request.env["HTTPS_PROXY"] = "http://127.0.0.1:48123"
managed.request.env["NO_PROXY"] = ""
managed_objects.append(managed)
return managed
async def _fake_run_under_backend(request, *, runtime=None):
seen["env"] = request.env
seen["policy"] = request.policy
return SimpleNamespace(
returncode=0,
stdout="ok\n",
stderr="",
timed_out=False,
backend_notes=(),
)
monkeypatch.setattr(code_exec, "preflight_subprocess_managed_network", _fake_preflight)
monkeypatch.setattr(
code_exec,
"prepare_subprocess_managed_network_proxy",
_fake_prepare,
raising=False,
)
monkeypatch.setattr(code_exec, "run_under_backend", _fake_run_under_backend)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
domains=(DomainGrant(domain="example.com"),),
),
)
)
try:
result = json.loads(
await code_exec.execute_code(
"import urllib.request\nurllib.request.urlopen('https://example.com')"
)
)
finally:
current_tool_context.reset(token)
assert result["exit_code"] == 0, result
env = seen["env"]
assert isinstance(env, dict)
assert env["HTTP_PROXY"] == "http://127.0.0.1:48123"
assert env["HTTPS_PROXY"] == env["HTTP_PROXY"]
assert env["NO_PROXY"] == ""
assert managed_objects
assert managed_objects[0].cleaned
@pytest.mark.asyncio
async def test_trusted_code_exec_path_denial_escalates_without_retry(
managed_runtime: Path,
tmp_path_factory: pytest.TempPathFactory,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox.types import (
DenialReason,
DenialResult,
SandboxRequest,
SecurityLevel,
SuggestedNextStep,
)
from opensquilla.tools.builtin import code_exec, shell
outside = tmp_path_factory.mktemp("outside-code")
backend_calls: list[SandboxRequest] = []
async def _fake_run_under_backend(request, *, runtime=None):
backend_calls.append(request)
return SimpleNamespace(
returncode=1,
stdout="",
stderr="",
timed_out=False,
backend_notes=(f"filesystem.read.denied: {outside}",),
)
async def _fake_escalate_backend_denial(*args, **kwargs):
return DenialResult(
reason=DenialReason.SEATBELT_DENIED,
suggested_next_step=SuggestedNextStep.ASK_USER,
level=SecurityLevel.STANDARD,
action_fingerprint="test",
message="denied",
retryable=False,
)
monkeypatch.setattr(code_exec, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(code_exec, "escalate_backend_denial", _fake_escalate_backend_denial)
monkeypatch.setattr(code_exec, "_resolve_python_bin", lambda *, sandbox_enabled: sys.executable)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="trusted",
sandbox_run_context=RunContext(
run_mode=RunMode.TRUSTED,
workspace=str(managed_runtime),
),
)
)
try:
result = json.loads(await code_exec.execute_code("print('ok')"))
finally:
current_tool_context.reset(token)
assert result["status"] == "denied"
assert len(backend_calls) == 1
@pytest.mark.asyncio
async def test_code_unknown_explicit_url_runs_with_managed_proxy(
managed_runtime: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
from opensquilla.sandbox import integration as integration_mod
from opensquilla.sandbox.types import SandboxRequest
from opensquilla.tools.builtin import code_exec, shell
class _FakeProxyServer:
host = "127.0.0.1"
port = 48123
def __init__(self, *args: object, **kwargs: object) -> None:
pass
async def start(self) -> None:
return None
async def stop(self) -> None:
return None
monkeypatch.setattr(integration_mod, "SandboxProxyServer", _FakeProxyServer)
seen: dict[str, object] = {}
async def _fake_run_under_backend(request, *, runtime=None):
managed = await integration_mod.prepare_subprocess_managed_network_proxy(
request,
runtime=runtime,
)
try:
seen["request"] = managed.request
return SimpleNamespace(
returncode=0,
stdout="ok\n",
stderr="",
backend_notes=(),
timed_out=False,
)
finally:
await managed.cleanup()
monkeypatch.setattr(code_exec, "run_under_backend", _fake_run_under_backend)
monkeypatch.setattr(code_exec, "_resolve_python_bin", lambda *, sandbox_enabled: sys.executable)
monkeypatch.setattr(shell, "_host_execution_allowed", lambda: False)
token = current_tool_context.set(
ToolContext(
is_owner=True,
caller_kind=CallerKind.CLI,
workspace_dir=str(managed_runtime),
session_key="s1",
run_mode="standard",
sandbox_run_context=RunContext(run_mode=RunMode.STANDARD),
)
)
try:
payload = json.loads(
await code_exec.execute_code(
"import urllib.request\n"
"import os\n"
"print(os.environ.get('HTTP_PROXY'))\n"
"print(os.environ.get('HTTPS_PROXY'))\n"
"print(os.environ.get('NO_PROXY'))\n"
"print('https://unknown.test/path')\n"
)
)
finally:
current_tool_context.reset(token)
assert payload["stdout"] == "ok\n"
request = seen["request"]
assert isinstance(request, SandboxRequest)
assert request.policy.network.value == "proxy_allowlist"
assert request.policy.network_proxy is not None
assert request.env["HTTP_PROXY"] == "http://127.0.0.1:48123"
assert request.env["HTTPS_PROXY"] == "http://127.0.0.1:48123"
assert "127.0.0.1" in request.env["NO_PROXY"]
assert get_approval_queue().list_pending("exec") == []
Reference in New Issue View Git Blame Copy Permalink