1"""Controlled worker environment (runs on the macOS node): sandbox-exec denies all outbound network except loopback; the2only way out is the AKOS egress proxy, which allows only hosts with a live grant. Tests shell (curl), Python urllib,3subprocess-from-Python and proxy routes. Unmanaged routes (host shell outside the sandbox, remote MCP servers) are NOT4covered and are documented in docs/threat-model.md."""5import asyncio6import os7import subprocess8import sys9import tempfile10import threading11import pytest12from akos.gateway import sandbox1314pytestmark = pytest.mark.node15PY = sys.executable161718@pytest.fixture(scope="module")19def proxy():20 """Run the egress proxy as a real subprocess on the test port against the test database (env inherited from conftest)."""21 import time22 proc = subprocess.Popen([PY, "-m", "akos", "proxy"], env={**os.environ, "AKOS_PROXY_PORT": "18621"}, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, cwd=os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))23 time.sleep(2.0)24 assert proc.poll() is None, proc.stdout.read().decode()[-1000:]25 yield 1862126 proc.terminate(); proc.wait(timeout=10)272829@pytest.mark.skipif(not sandbox.available(), reason="sandbox-exec not available (macOS only)")30def test_direct_egress_blocked_shell_python_subprocess(tmp_path):31 r = sandbox.run(["/usr/bin/curl", "-s", "-m", "8", "-o", "/dev/null", "-w", "%{http_code}", "--noproxy", "*", "https://example.com/"], workdir=str(tmp_path), timeout=30)32 assert r.returncode != 0 and r.stdout.strip() in ("", "000"), r.stdout + r.stderr # shell: blocked33 r = sandbox.run([PY, "-c", "import os;os.environ.pop('HTTPS_PROXY');os.environ.pop('https_proxy');import urllib.request;print(urllib.request.urlopen('https://example.com',timeout=8).status)"], workdir=str(tmp_path), timeout=30)34 assert r.returncode != 0, r.stdout # python direct: blocked35 r = sandbox.run([PY, "-c", "import subprocess;print(subprocess.run(['/usr/bin/curl','-s','-m','8','--noproxy','*','https://example.com/'],capture_output=True).returncode)"], workdir=str(tmp_path), timeout=30)36 assert r.returncode == 0 and r.stdout.strip() not in ("0",), r.stdout # subprocess inherits the sandbox: blocked37 r = sandbox.run(["/bin/bash", "-c", "curl -s -m 8 --noproxy '*' https://example.com/ >/dev/null; echo rc=$?"], workdir=str(tmp_path), timeout=30)38 assert "rc=0" not in r.stdout394041@pytest.mark.skipif(not sandbox.available(), reason="sandbox-exec not available (macOS only)")42async def test_proxy_blocks_ungranted_and_allows_granted(tmp_path, proxy, pool):43 # no grant → 403 from the proxy (visible to the client), logged as block44 r = sandbox.run(["/usr/bin/curl", "-s", "-m", "15", "-o", "/dev/null", "-w", "%{http_code}", "https://api.frankfurter.dev/v1/latest"], workdir=str(tmp_path), timeout=40)45 assert r.stdout.strip() in ("403", "000", "56"), r.stdout + r.stderr46 async with pool.acquire() as conn:47 await conn.execute("INSERT INTO egress_grants(execution_id, host, port, expires_at) VALUES (NULL, 'api.frankfurter.dev', 443, now() + interval '2 minutes')")48 r = sandbox.run(["/usr/bin/curl", "-s", "-m", "20", "-o", "/dev/null", "-w", "%{http_code}", "https://api.frankfurter.dev/v1/latest"], workdir=str(tmp_path), timeout=40)49 assert r.stdout.strip() == "200", r.stdout + r.stderr # granted host via proxy: allowed50 r = sandbox.run([PY, "-c", "import urllib.request;print(urllib.request.urlopen('https://api.frankfurter.dev/v1/latest',timeout=20).status)"], workdir=str(tmp_path), timeout=40)51 assert r.stdout.strip() == "200", r.stdout + r.stderr # python honours HTTPS_PROXY52 r = sandbox.run(["/usr/bin/curl", "-s", "-m", "15", "-o", "/dev/null", "-w", "%{http_code}", "https://example.com/"], workdir=str(tmp_path), timeout=40)53 assert r.stdout.strip() in ("403", "000", "56") # other host still blocked54 async with pool.acquire() as conn:55 rows = await conn.fetch("SELECT host, decision FROM egress_log ORDER BY id DESC LIMIT 10")56 assert any(x["host"] == "api.frankfurter.dev" and x["decision"] == "allow" for x in rows)57 assert any(x["decision"] == "block" for x in rows)58 await conn.execute("DELETE FROM egress_grants WHERE execution_id IS NULL")596061@pytest.mark.skipif(not sandbox.available(), reason="sandbox-exec not available (macOS only)")62def test_sandbox_denies_writes_outside_workdir(tmp_path):63 r = sandbox.run(["/bin/bash", "-c", "echo x > /Users/Shared/akos-escape-test 2>/dev/null; echo rc=$?; echo y > ./ok.txt; echo rc2=$?"], workdir=str(tmp_path), timeout=20)64 assert "rc=1" in r.stdout and "rc2=0" in r.stdout and not os.path.exists("/Users/Shared/akos-escape-test")65