"""Controlled worker environment (runs on the macOS node): sandbox-exec denies all outbound network except loopback; the
only way out is the AKOS egress proxy, which allows only hosts with a live grant. Tests shell (curl), Python urllib,
subprocess-from-Python and proxy routes. Unmanaged routes (host shell outside the sandbox, remote MCP servers) are NOT
covered and are documented in docs/threat-model.md."""
import asyncio
import os
import subprocess
import sys
import tempfile
import threading
import pytest
from akos.gateway import sandbox

pytestmark = pytest.mark.node
PY = sys.executable


@pytest.fixture(scope="module")
def proxy():
    """Run the egress proxy as a real subprocess on the test port against the test database (env inherited from conftest)."""
    import time
    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__)))))
    time.sleep(2.0)
    assert proc.poll() is None, proc.stdout.read().decode()[-1000:]
    yield 18621
    proc.terminate(); proc.wait(timeout=10)


@pytest.mark.skipif(not sandbox.available(), reason="sandbox-exec not available (macOS only)")
def test_direct_egress_blocked_shell_python_subprocess(tmp_path):
    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)
    assert r.returncode != 0 and r.stdout.strip() in ("", "000"), r.stdout + r.stderr            # shell: blocked
    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)
    assert r.returncode != 0, r.stdout                                                                   # python direct: blocked
    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)
    assert r.returncode == 0 and r.stdout.strip() not in ("0",), r.stdout                                # subprocess inherits the sandbox: blocked
    r = sandbox.run(["/bin/bash", "-c", "curl -s -m 8 --noproxy '*' https://example.com/ >/dev/null; echo rc=$?"], workdir=str(tmp_path), timeout=30)
    assert "rc=0" not in r.stdout


@pytest.mark.skipif(not sandbox.available(), reason="sandbox-exec not available (macOS only)")
async def test_proxy_blocks_ungranted_and_allows_granted(tmp_path, proxy, pool):
    # no grant → 403 from the proxy (visible to the client), logged as block
    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)
    assert r.stdout.strip() in ("403", "000", "56"), r.stdout + r.stderr
    async with pool.acquire() as conn:
        await conn.execute("INSERT INTO egress_grants(execution_id, host, port, expires_at) VALUES (NULL, 'api.frankfurter.dev', 443, now() + interval '2 minutes')")
    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)
    assert r.stdout.strip() == "200", r.stdout + r.stderr                                             # granted host via proxy: allowed
    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)
    assert r.stdout.strip() == "200", r.stdout + r.stderr                                             # python honours HTTPS_PROXY
    r = sandbox.run(["/usr/bin/curl", "-s", "-m", "15", "-o", "/dev/null", "-w", "%{http_code}", "https://example.com/"], workdir=str(tmp_path), timeout=40)
    assert r.stdout.strip() in ("403", "000", "56")                                                  # other host still blocked
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT host, decision FROM egress_log ORDER BY id DESC LIMIT 10")
        assert any(x["host"] == "api.frankfurter.dev" and x["decision"] == "allow" for x in rows)
        assert any(x["decision"] == "block" for x in rows)
        await conn.execute("DELETE FROM egress_grants WHERE execution_id IS NULL")


@pytest.mark.skipif(not sandbox.available(), reason="sandbox-exec not available (macOS only)")
def test_sandbox_denies_writes_outside_workdir(tmp_path):
    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)
    assert "rc=1" in r.stdout and "rc2=0" in r.stdout and not os.path.exists("/Users/Shared/akos-escape-test")