"""Tests for the Bubblewrap isolated execution provider."""

from __future__ import annotations

import asyncio
import os
import shutil
import stat
from pathlib import Path

import pytest

from cyberbrain26.execution_plane import (
    AuthorizedExecutionService,
    BubblewrapExecutionProvider,
    ExecutionRefused,
)
from cyberbrain26.product.local_store import ensure_workspaces_root, is_within_workspaces_root
from cyberbrain26.security_plane.actions import (
    ExecutionAuthorization,
    bound_execution_authorization,
    execution_payload_digest,
)
from cyberbrain26.security_plane.approval import ApprovalManager
from cyberbrain26.security_plane.audit import AuditEventType, AuditLog, AuditQuery
from cyberbrain26.security_plane.identity import AgentIdentity, PermissionScope, TrustLevel
from cyberbrain26.security_plane.permission import PermissionGateway, SecurityError
from cyberbrain26.security_plane.policy import PolicyEngine

requires_bubblewrap = pytest.mark.skipif(
    shutil.which("bwrap") is None, reason="Bubblewrap is unavailable"
)


@pytest.fixture
def no_process_launch(monkeypatch: pytest.MonkeyPatch) -> list[tuple[object, ...]]:
    """Fail the test if any subprocess is created.

    The confinement check has to happen before Bubblewrap exists, so a refused
    workspace is asserted by proving that nothing was ever spawned.
    """
    attempts: list[tuple[object, ...]] = []

    async def _record(*args: object, **kwargs: object) -> None:
        attempts.append(args)
        raise AssertionError("a process was spawned even though the workspace was refused")

    monkeypatch.setattr(asyncio, "create_subprocess_exec", _record)
    return attempts


@pytest.mark.asyncio
@requires_bubblewrap
async def test_python_executes_in_isolated_workspace() -> None:
    provider = BubblewrapExecutionProvider()
    result = await provider.execute("print('sandbox-ok')", "python")

    assert result["return_code"] == 0
    assert result["stdout"].strip() == "sandbox-ok"
    assert result["provider"] == "bubblewrap"
    assert result["timed_out"] is False


@pytest.mark.asyncio
@requires_bubblewrap
async def test_sandbox_has_no_external_network_interface() -> None:
    provider = BubblewrapExecutionProvider()
    result = await provider.execute(
        "import socket\nprint([name for _, name in socket.if_nameindex()])",
        "python",
    )

    assert result["return_code"] == 0
    assert result["stdout"].strip() == "['lo']"


@pytest.mark.asyncio
@requires_bubblewrap
async def test_the_network_profile_keeps_its_own_pid_namespace() -> None:
    """Sharing the network must not share PID, IPC or the user namespace.

    An argv assertion cannot catch this: ``--share-net`` on its own is a
    perfectly well-formed vector that bwrap accepts, and it silently drops
    *every* unshare directive because bwrap unshares nothing by default. Run
    with the flag, the sandbox reported ``/proc/1/comm == "systemd"`` - the host
    init - and could list 327 host processes and read ``/proc/<pid>/environ`` of
    a host process, i.e. every API key and token of this user. Only executing
    bwrap shows that.
    """
    provider = BubblewrapExecutionProvider()
    result = await provider.execute(
        "import os\n"
        "print('pid1=' + open('/proc/1/comm').read().strip())\n"
        "print('procs=' + str(len([d for d in os.listdir('/proc') if d.isdigit()])))\n",
        "python",
        allow_network=True,
    )

    assert result["return_code"] == 0, result["stderr"]
    assert not result["stdout"].startswith("pid1=systemd"), (
        "the sandbox adopted the host PID namespace"
    )
    assert "pid1=bwrap" in result["stdout"], result["stdout"]
    # A shared PID namespace would show the host's whole process table; a real
    # namespace shows this sandbox's own few processes and nothing else.
    visible = int(result["stdout"].split("procs=")[1].splitlines()[0])
    assert visible < 50, f"sandbox sees {visible} processes - PID namespace is shared"


@pytest.mark.asyncio
@requires_bubblewrap
async def test_the_network_profile_really_does_reach_the_network() -> None:
    """The counterpart to the test above: the grant must actually work.

    Without this, "keep ``--unshare-all``" is satisfied by a sandbox that has no
    network at all - a silent refusal that would read as a pentest tool bug.
    """
    provider = BubblewrapExecutionProvider()
    result = await provider.execute(
        "import socket\n"
        "names = [n for _, n in socket.if_nameindex()]\n"
        "print('ifaces=' + ','.join(names))\n",
        "python",
        allow_network=True,
    )

    assert result["return_code"] == 0, result["stderr"]
    ifaces = result["stdout"].split("ifaces=")[1].strip().split(",")
    assert len(ifaces) > 1, f"network profile still has no egress: {ifaces}"


@pytest.mark.asyncio
@requires_bubblewrap
async def test_shell_executes_inside_the_sandbox() -> None:
    provider = BubblewrapExecutionProvider()
    result = await provider.execute("printf 'shell-ok\\n'", "shell")

    assert result["return_code"] == 0
    assert result["stdout"].strip() == "shell-ok"


@pytest.mark.asyncio
@requires_bubblewrap
async def test_workspace_is_writable_but_neighboring_files_are_hidden(
    sandbox_workspaces_root: Path,
) -> None:
    provider = BubblewrapExecutionProvider()
    workspace = sandbox_workspaces_root / "workspace"
    workspace.mkdir()
    # A host secret next to, but outside of, the sanctioned root: the sandbox
    # must be able to write its own workspace without seeing anything beside it.
    secret = sandbox_workspaces_root.parent / "outside-secret.txt"
    secret.write_text("host-secret", encoding="utf-8")
    result = await provider.execute(
        "from pathlib import Path\n"
        "Path('created.txt').write_text('inside')\n"
        f"print(Path({str(secret)!r}).exists())\n",
        "python",
        workspace=str(workspace),
    )

    assert result["return_code"] == 0
    assert result["stdout"].strip() == "False"
    assert (workspace / "created.txt").read_text(encoding="utf-8") == "inside"
    assert secret.read_text(encoding="utf-8") == "host-secret"


# ----------------------------------------------------------------------
# Workspace confinement
#
# The workspace reaches the provider straight from a step payload and is
# bind-mounted read-write, so an unconfined workspace gives sandboxed code the
# whole host and makes every other control in the provider pointless. Each
# refusal below is asserted together with ``no_process_launch``, which proves
# the refusal happens before Bubblewrap is started.
# ----------------------------------------------------------------------
@pytest.mark.asyncio
async def test_workspace_outside_the_allowed_root_is_refused(
    tmp_path: Path,
    sandbox_workspaces_root: Path,
    no_process_launch: list[tuple[object, ...]],
) -> None:
    provider = BubblewrapExecutionProvider()
    outside = tmp_path / "outside"
    outside.mkdir()

    with pytest.raises(ValueError) as excinfo:
        await provider.execute("print('must not run')", "python", workspace=str(outside))

    message = str(excinfo.value)
    assert "outside the allowed workspaces root" in message
    assert str(outside) in message, "the refusal must name the offending path"
    assert str(sandbox_workspaces_root) in message, "the refusal must name the allowed root"
    assert no_process_launch == []


@pytest.mark.asyncio
async def test_system_directory_workspace_is_refused(
    sandbox_workspaces_root: Path,
    no_process_launch: list[tuple[object, ...]],
) -> None:
    provider = BubblewrapExecutionProvider()

    with pytest.raises(ValueError, match="outside the allowed workspaces root") as excinfo:
        await provider.execute("print('must not run')", "python", workspace="/etc")

    assert "/etc" in str(excinfo.value)
    assert no_process_launch == []


@pytest.mark.asyncio
async def test_symlinked_workspace_escaping_the_root_is_refused(
    tmp_path: Path,
    sandbox_workspaces_root: Path,
    no_process_launch: list[tuple[object, ...]],
) -> None:
    provider = BubblewrapExecutionProvider()
    target = tmp_path / "escaped"
    target.mkdir()
    link = sandbox_workspaces_root / "escape-link"
    link.symlink_to(target)

    with pytest.raises(ValueError) as excinfo:
        await provider.execute("print('must not run')", "python", workspace=str(link))

    # The link itself lives in the root; its target does not. Resolving both
    # sides is what turns this into a refusal.
    assert str(target) in str(excinfo.value)
    assert "outside the allowed workspaces root" in str(excinfo.value)
    assert no_process_launch == []


@pytest.mark.asyncio
async def test_relative_traversal_workspace_is_refused(
    tmp_path: Path,
    sandbox_workspaces_root: Path,
    monkeypatch: pytest.MonkeyPatch,
    no_process_launch: list[tuple[object, ...]],
) -> None:
    provider = BubblewrapExecutionProvider()
    outside = tmp_path / "outside"
    outside.mkdir()
    nested = sandbox_workspaces_root / "a" / "b"
    nested.mkdir(parents=True)
    monkeypatch.chdir(nested)

    with pytest.raises(ValueError) as excinfo:
        await provider.execute("print('must not run')", "python", workspace="../../../../outside")

    assert str(outside) in str(excinfo.value)
    assert no_process_launch == []


@pytest.mark.asyncio
async def test_root_itself_is_inside_and_siblings_are_not(
    sandbox_workspaces_root: Path, no_process_launch: list[tuple[object, ...]]
) -> None:
    provider = BubblewrapExecutionProvider()

    assert provider.allowed_workspaces_root == sandbox_workspaces_root
    assert is_within_workspaces_root(sandbox_workspaces_root)
    assert is_within_workspaces_root(sandbox_workspaces_root / "nested" / "deep")
    assert not is_within_workspaces_root(sandbox_workspaces_root.parent)
    assert not is_within_workspaces_root(sandbox_workspaces_root.parent / "workspaces-copy")


@pytest.mark.asyncio
async def test_confinement_is_checked_before_bubblewrap_availability(
    tmp_path: Path,
    sandbox_workspaces_root: Path,
    monkeypatch: pytest.MonkeyPatch,
    no_process_launch: list[tuple[object, ...]],
) -> None:
    # Untrusted input is validated first, so a missing Bubblewrap binary can
    # never be the reason an unconfined workspace gets as far as being used.
    monkeypatch.setattr(
        "cyberbrain26.execution_plane.bubblewrap_provider.shutil.which", lambda _: None
    )
    provider = BubblewrapExecutionProvider()
    outside = tmp_path / "outside"
    outside.mkdir()

    with pytest.raises(ValueError, match="outside the allowed workspaces root"):
        await provider.execute("print('must not run')", "python", workspace=str(outside))

    assert no_process_launch == []


@pytest.mark.asyncio
@requires_bubblewrap
async def test_explicit_workspaces_root_confines_to_that_root(
    tmp_path: Path, sandbox_workspaces_root: Path
) -> None:
    # Migration path: a deployment may widen the sanctioned root, but only
    # explicitly, and the new root then becomes the only accepted one.
    custom_root = tmp_path / "custom-workspaces"
    custom_root.mkdir()
    provider = BubblewrapExecutionProvider(workspaces_root=custom_root)
    workspace = custom_root / "project"
    workspace.mkdir()

    result = await provider.execute(
        "open('made-here.txt', 'w').write('x')", "python", workspace=str(workspace)
    )

    assert result["return_code"] == 0
    assert (workspace / "made-here.txt").read_text(encoding="utf-8") == "x"
    with pytest.raises(ValueError, match="outside the allowed workspaces root"):
        await provider.execute(
            "print('must not run')", "python", workspace=str(sandbox_workspaces_root)
        )


@pytest.mark.asyncio
@requires_bubblewrap
async def test_env_workspaces_root_is_honoured(
    tmp_path: Path, sandbox_workspaces_root: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    env_root = tmp_path / "env-workspaces"
    env_root.mkdir()
    monkeypatch.setenv("CYBERBRAIN_WORKSPACES_ROOT", str(env_root))
    provider = BubblewrapExecutionProvider()
    workspace = env_root / "project"
    workspace.mkdir()

    result = await provider.execute("print('env-ok')", "python", workspace=str(workspace))

    assert result["return_code"] == 0
    assert result["stdout"].strip() == "env-ok"
    # The default root stops being accepted once a root is configured, and the
    # refusal says which root is in force.
    with pytest.raises(ValueError) as excinfo:
        await provider.execute(
            "print('must not run')", "python", workspace=str(sandbox_workspaces_root)
        )
    assert str(env_root) in str(excinfo.value)


def test_ensure_workspaces_root_creates_an_owner_only_directory(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    data_dir = tmp_path / "fresh-data"
    monkeypatch.setenv("CYBERBRAIN_DATA_DIR", str(data_dir))
    monkeypatch.delenv("CYBERBRAIN_WORKSPACES_ROOT", raising=False)

    root = ensure_workspaces_root()

    assert root == data_dir / "workspaces"
    assert root.is_dir()
    assert stat.S_IMODE(os.stat(root).st_mode) == 0o700


@pytest.mark.asyncio
@requires_bubblewrap
async def test_execution_times_out_and_caps_output() -> None:
    provider = BubblewrapExecutionProvider(
        max_timeout_seconds=0.2,
        max_output_bytes=32,
    )
    timed_out = await provider.execute("while True: pass", "python", timeout_seconds=10)
    noisy = await provider.execute("print('x' * 1000)", "python")

    assert timed_out["timed_out"] is True
    assert noisy["return_code"] == 0
    assert len(noisy["stdout"]) <= 32
    assert noisy["stdout_truncated"] is True


@pytest.mark.asyncio
async def test_unsupported_language_is_rejected() -> None:
    provider = BubblewrapExecutionProvider()
    with pytest.raises(ValueError, match="unsupported execution language"):
        await provider.execute("echo hi", "javascript")


@pytest.mark.asyncio
async def test_missing_bubblewrap_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr("cyberbrain26.execution_plane.bubblewrap_provider.shutil.which", lambda _: None)
    provider = BubblewrapExecutionProvider()

    with pytest.raises(RuntimeError, match="refusing to run code without a sandbox"):
        await provider.execute("print('must not run')", "python")


@pytest.mark.asyncio
@requires_bubblewrap
async def test_authorized_execution_waits_for_approval_and_audits_result() -> None:
    audit = AuditLog()
    approvals = ApprovalManager()
    identity = AgentIdentity(
        "execution-agent",
        "coder",
        trust_level=TrustLevel.TRUSTED,
        scopes={PermissionScope.SHELL.value},
    )
    service = AuthorizedExecutionService(
        BubblewrapExecutionProvider(),
        PermissionGateway(PolicyEngine(), audit),
        approval_manager=approvals,
        audit_log=audit,
        approval_timeout_seconds=1.0,
    )

    async def approve_when_requested() -> None:
        while not approvals.list_pending():
            await asyncio.sleep(0.005)
        request = approvals.list_pending()[0]
        approvals.approve(request.request_id, "reviewer", "approved")

    approval_task = asyncio.create_task(approve_when_requested())
    result = await service.execute(
        identity,
        "printf 'authorized\\n'",
        "shell",
        task_id="execution-task-1",
    )
    await approval_task

    assert result["return_code"] == 0
    assert result["stdout"].strip() == "authorized"
    granted = audit.query(
        AuditQuery(event_type=AuditEventType.APPROVAL_GRANTED, task_id="execution-task-1")
    )
    completed = audit.query(
        AuditQuery(event_type=AuditEventType.TOOL_RESULT, task_id="execution-task-1")
    )
    assert len(granted) == 1
    assert len(completed) == 1
    assert completed[0].result == "succeeded"


@pytest.mark.asyncio
async def test_authorized_execution_denies_missing_scope_before_provider_call() -> None:
    class NeverRunProvider:
        provider_name = "test"
        called = False

        async def execute(self, *args: object, **kwargs: object) -> dict[str, object]:
            self.called = True
            return {"return_code": 0}

    provider = NeverRunProvider()
    service = AuthorizedExecutionService(
        provider,  # type: ignore[arg-type]
        PermissionGateway(PolicyEngine()),
    )
    identity = AgentIdentity("no-scope", "coder", trust_level=TrustLevel.TRUSTED)

    with pytest.raises(SecurityError, match="lacks required scope"):
        await service.execute(
            identity,
            "print('no')",
            "python",
            task_id="execution-task-denied",
        )
    assert provider.called is False


# ----------------------------------------------------------------------
# Single-approval gating: the inner gateway re-verifies, it does not re-ask
# ----------------------------------------------------------------------
class _NeverRunProvider:
    """Execution provider that records the payloads it was asked to run."""

    provider_name = "never-run"

    def __init__(self) -> None:
        self.calls: list[tuple[object, ...]] = []

    async def execute(self, *args: object, **kwargs: object) -> dict[str, object]:
        self.calls.append(args)
        return {"return_code": 0}


def _approved_grant(
    approvals: ApprovalManager,
    identity: AgentIdentity,
    action: str,
    *,
    code: object,
    language: object,
    workspace: object = None,
) -> ExecutionAuthorization:
    """Return a grant backed by a genuinely approved approval request."""
    request = approvals.create_request(identity, action, None, context={"task_id": "outer"})
    assert approvals.approve(request.request_id, "reviewer", "approved") is not None
    return ExecutionAuthorization(
        agent_id=identity.agent_id,
        task_id="outer",
        action=action,
        resource=None,
        payload_digest=execution_payload_digest(
            code=code, language=language, workspace=workspace
        ),
        approval_request_id=request.request_id,
    )


@pytest.mark.asyncio
async def test_gateway_refuses_a_payload_that_changed_after_the_outer_approval() -> None:
    """A human approved one piece of code; only that code may reach a provider.

    This is the security-critical half of the single-approval flow. The outer
    gate got one approval and the inner gateway must not ask a second time, but
    it must also not run a payload the operator never saw. The grant is bound to
    a digest of the approved payload, so swapping the code between the two
    checks has to be refused outright - the provider is never reached and no new
    approval is created.
    """
    approvals = ApprovalManager()
    identity = AgentIdentity(
        "execution-agent",
        "coder",
        trust_level=TrustLevel.TRUSTED,
        scopes={PermissionScope.PYTHON.value},
    )
    provider = _NeverRunProvider()
    service = AuthorizedExecutionService(
        provider,  # type: ignore[arg-type]
        PermissionGateway(PolicyEngine()),
        approval_manager=approvals,
        approval_timeout_seconds=0.05,
    )
    grant = _approved_grant(
        approvals, identity, "execute_python", code="print('approved')", language="python"
    )

    with bound_execution_authorization(grant):
        with pytest.raises(ExecutionRefused, match="does not cover this execution"):
            await service.execute(
                identity,
                "print('swapped after approval')",
                "python",
                task_id="payload-swap",
            )

    assert provider.calls == []
    # One approval existed - the outer one. The refusal did not ask for another.
    assert len(approvals.list_records()) == 1


@pytest.mark.asyncio
async def test_approval_for_action_a_does_not_authorise_action_b() -> None:
    """A grant for one action is ignored, not honoured, when another is run.

    An approval for ``execute_python`` says nothing about ``execute_shell``, so
    the gateway must fall back to its own full gate for the action it is really
    performing rather than reusing the grant.
    """
    approvals = ApprovalManager()
    identity = AgentIdentity(
        "execution-agent",
        "coder",
        trust_level=TrustLevel.TRUSTED,
        scopes={PermissionScope.PYTHON.value, PermissionScope.SHELL.value},
    )
    provider = _NeverRunProvider()
    service = AuthorizedExecutionService(
        provider,  # type: ignore[arg-type]
        PermissionGateway(PolicyEngine()),
        approval_manager=approvals,
        approval_timeout_seconds=0.05,
    )
    grant = _approved_grant(
        approvals, identity, "execute_python", code="print('approved')", language="python"
    )

    with bound_execution_authorization(grant):
        with pytest.raises(ExecutionRefused, match="execute_shell"):
            await service.execute(
                identity, "echo approved", "shell", task_id="action-swap"
            )

    assert provider.calls == []
    # A second request was raised for the other action: action A's approval was
    # not spent, and action B got no approval of its own.
    pending_or_decided = [
        r for r in [approvals.get_request(grant.approval_request_id or "")]
        if r is not None
    ]
    assert len(pending_or_decided) == 1
    assert pending_or_decided[0].action == "execute_python"


@pytest.mark.asyncio
async def test_unchanged_payload_reuses_the_outer_approval_and_still_runs() -> None:
    """The happy path: one human decision, both gates, the code actually runs."""
    approvals = ApprovalManager()
    identity = AgentIdentity(
        "execution-agent",
        "coder",
        trust_level=TrustLevel.TRUSTED,
        scopes={PermissionScope.PYTHON.value},
    )
    audit = AuditLog()
    provider = _NeverRunProvider()
    service = AuthorizedExecutionService(
        provider,  # type: ignore[arg-type]
        PermissionGateway(PolicyEngine(), audit),
        approval_manager=approvals,
        audit_log=audit,
        approval_timeout_seconds=0.05,
    )
    grant = _approved_grant(
        approvals, identity, "execute_python", code="print('approved')", language="python"
    )

    with bound_execution_authorization(grant):
        await service.execute(
            identity, "print('approved')", "python", task_id="reuse-happy-path"
        )

    assert len(provider.calls) == 1
    granted = audit.query(
        AuditQuery(event_type=AuditEventType.APPROVAL_GRANTED, task_id="reuse-happy-path")
    )
    assert [event.result for event in granted] == ["reused"]
    # The inner gate really did re-evaluate. The permission gateway records its
    # own scope/policy checks without a task id, so they are asserted by result:
    # reuse replaced the second *request*, not a check.
    assert "scope_check_passed" in [event.result for event in audit.query(
        AuditQuery(event_type=AuditEventType.PERMISSION_CHECK)
    )]


@pytest.mark.asyncio
async def test_grant_is_ignored_once_its_approval_is_no_longer_approved() -> None:
    """A grant is a claim made at dispatch time; it is re-read, not trusted.

    The request behind it was cancelled instead of approved, so the grant no
    longer describes a human decision. The gateway must fall back to its own
    full gate and ask - strictly stronger than honouring an unverifiable claim.
    """
    approvals = ApprovalManager()
    identity = AgentIdentity(
        "execution-agent",
        "coder",
        trust_level=TrustLevel.TRUSTED,
        scopes={PermissionScope.PYTHON.value},
    )
    provider = _NeverRunProvider()
    service = AuthorizedExecutionService(
        provider,  # type: ignore[arg-type]
        PermissionGateway(PolicyEngine()),
        approval_manager=approvals,
        approval_timeout_seconds=0.05,
    )
    request = approvals.create_request(identity, "execute_python", None)
    grant = ExecutionAuthorization(
        agent_id=identity.agent_id,
        task_id="outer",
        action="execute_python",
        resource=None,
        payload_digest=execution_payload_digest(
            code="print('approved')", language="python", workspace=None
        ),
        approval_request_id=request.request_id,
    )
    # The request never got a decision; the operator withdrew it.
    assert approvals.cancel(request.request_id) is not None

    with bound_execution_authorization(grant):
        with pytest.raises(ExecutionRefused, match="execute_python"):
            await service.execute(
                identity, "print('approved')", "python", task_id="withdrawn"
            )

    assert provider.calls == []
    assert approvals.list_records() == []


@pytest.mark.asyncio
async def test_outer_allow_without_a_human_decision_is_never_a_grant() -> None:
    """A policy "allow" carries no human decision and cannot stand in for one."""
    approvals = ApprovalManager()
    identity = AgentIdentity(
        "execution-agent",
        "coder",
        trust_level=TrustLevel.TRUSTED,
        scopes={PermissionScope.PYTHON.value},
    )
    provider = _NeverRunProvider()
    service = AuthorizedExecutionService(
        provider,  # type: ignore[arg-type]
        PermissionGateway(PolicyEngine()),
        approval_manager=approvals,
        approval_timeout_seconds=0.05,
    )
    grant = ExecutionAuthorization(
        agent_id=identity.agent_id,
        task_id="outer",
        action="execute_python",
        resource=None,
        payload_digest=execution_payload_digest(
            code="print('allowed')", language="python", workspace=None
        ),
        approval_request_id=None,
    )

    with bound_execution_authorization(grant):
        with pytest.raises(ExecutionRefused, match="execute_python"):
            await service.execute(
                identity, "print('allowed')", "python", task_id="no-human"
            )

    assert provider.calls == []
    assert approvals.list_records() == []