smart-org/apps/api/tests/test_docker_sandbox.py

"""§v2-A3.1 — the docker sandbox adapter is the hardened execution boundary.

The tests here are almost entirely about *arguments*: which flags are passed,
which paths are mounted, which strings appear in ``argv``. That is deliberate.
The adapter's job is not to be clever, it is to be exactly right about a small
number of security-critical decisions, and the only way to hold that line
against future edits is to assert on the literal argument vector.

A fake runner stands in for the container runtime, so the suite needs no docker
installed. The integration tests that *do* need it are marked ``docker`` and
skipped unless ``SANDBOX_DOCKER_IT=1``.
"""

from __future__ import annotations

import os

import pytest
import pytest_asyncio
from httpx import AsyncClient

from app.core.errors import PermissionDeniedError, ValidationError
from app.infrastructure.adapters.docker_sandbox import (
    LABEL_ORG,
    LABEL_PROJECT,
    LABEL_ROLE,
    DockerSandboxAdapter,
    PathEscapeError,
    container_name,
    image_for_language,
    safe_join,
)
from app.infrastructure.adapters.process import ProcessResult, run_process
from tests.conftest import auth_header

# ---------------------------------------------------------------------------
# A fake container runtime
# ---------------------------------------------------------------------------


class FakeRunner:
    """Records every argv and replays scripted results in order.

    Responses are command-aware by default, because the adapter's flow depends
    on the subcommand: ``ps`` decides whether a container already exists,
    ``run`` reports the new container id, everything else is inert. A test that
    needs something else queues an explicit response in ``responses``.
    """

    def __init__(
        self,
        *,
        container_id: str = "cid123",
        stdout: str = "",
        ps_stdout: str = "",
        exit_code: int = 0,
    ) -> None:
        self.calls: list[tuple[tuple[str, ...], dict]] = []
        self.responses: list[ProcessResult] = []
        self.container_id = container_id
        self.stdout = stdout
        self.ps_stdout = ps_stdout
        self.exit_code = exit_code

    async def __call__(self, argv, **kwargs) -> ProcessResult:  # noqa: ANN001, ANN003
        self.calls.append((tuple(argv), kwargs))
        if self.responses:
            return self.responses.pop(0)
        subcommand = argv[1] if len(argv) > 1 else ""
        if subcommand == "ps":
            out = self.ps_stdout
        elif subcommand == "run":
            out = f"{self.container_id}\n"
        else:
            out = self.stdout
        return ProcessResult(argv=tuple(argv), exit_code=self.exit_code, stdout=out)

    @property
    def last_argv(self) -> tuple[str, ...]:
        return self.calls[-1][0]


def _adapter(tmp_path, runner: FakeRunner, **overrides) -> DockerSandboxAdapter:
    kwargs = {
        "provider_bin": "docker",
        "workspaces_root": str(tmp_path),
        "default_cpus": 2.0,
        "default_memory_mb": 512,
    }
    kwargs.update(overrides)
    return DockerSandboxAdapter(runner=runner, **kwargs)


# ---------------------------------------------------------------------------
# Image mapping
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    ("language", "expected"),
    [
        ("node", "node:20-bookworm"),
        ("python", "python:3.13-bookworm"),
        ("php", "php:8.3-cli"),
        ("go", "golang:1.23"),
        ("static", "debian:bookworm"),
    ],
)
def test_image_for_language_maps_the_known_languages(language: str, expected: str):
    assert image_for_language(language) == expected


def test_image_for_an_unknown_language_falls_back():
    assert image_for_language("cobol") == "debian:bookworm"
    assert image_for_language(None) == "debian:bookworm"


def test_container_name_is_deterministic_and_dns_safe():
    name = container_name("workspace", "org123", "proj456")
    assert name == "smartorg-workspace-org123-proj456"
    assert container_name("workspace", "org123", "proj456") == name


# ---------------------------------------------------------------------------
# Path safety
# ---------------------------------------------------------------------------


def test_safe_join_accepts_paths_inside_the_root(tmp_path):
    root = tmp_path / "work"
    root.mkdir()
    assert safe_join(str(root), "src/main.py") == str(root / "src/main.py")
    assert safe_join(str(root), "/src/main.py") == str(root / "src/main.py")


def test_safe_join_refuses_a_parent_escape(tmp_path):
    root = tmp_path / "work"
    root.mkdir()
    with pytest.raises(PathEscapeError):
        safe_join(str(root), "../secrets.env")
    with pytest.raises(PathEscapeError):
        safe_join(str(root), "a/b/../../../etc/passwd")


@pytest.mark.skipif(
    os.name == "nt",
    reason="creating a symlink on Windows requires SeCreateSymbolicLinkPrivilege "
    "(admin or Developer Mode); the path-escape guard is covered on POSIX hosts",
)
def test_safe_join_refuses_a_symlink_that_escapes(tmp_path):
    """A symlink planted inside the work dir must not become a host read."""
    root = tmp_path / "work"
    root.mkdir()
    outside = tmp_path / "outside.txt"
    outside.write_text("secret")
    os.symlink(outside, root / "escape.txt")

    with pytest.raises(PathEscapeError):
        safe_join(str(root), "escape.txt")


# ---------------------------------------------------------------------------
# create_workspace
# ---------------------------------------------------------------------------


async def test_create_workspace_makes_a_private_host_dir(tmp_path):
    runner = FakeRunner()
    adapter = _adapter(tmp_path, runner)

    info = await adapter.create_workspace(
        organization_id="org1", project_id="proj1", template_id="t1", image="python:3.13-bookworm"
    )

    host_dir = tmp_path / "org1" / "proj1" / "work"
    assert host_dir.is_dir(), "the bind-mount source must exist before `docker run`"
    assert info.external_id == "ws-org1-proj1"
    assert runner.calls == [], "creating a workspace must not shell out"


async def test_create_workspace_refuses_a_hostile_id(tmp_path):
    adapter = _adapter(tmp_path, FakeRunner())
    with pytest.raises(ValidationError):
        await adapter.create_workspace(
            organization_id="../../etc", project_id="p", template_id="t", image="i"
        )


# ---------------------------------------------------------------------------
# start_session — the hardening flags are the whole point
# ---------------------------------------------------------------------------


async def test_start_session_passes_every_hardening_flag(tmp_path):
    runner = FakeRunner(container_id="deadbeef")
    adapter = _adapter(tmp_path, runner)

    info = await adapter.create_workspace(
        organization_id="org1", project_id="proj1", template_id="t1", image="node:20-bookworm"
    )
    session = await adapter.start_session(workspace_external_id=info.external_id)

    assert session.external_id == "deadbeef"
    argv = runner.last_argv
    assert argv[0] == "docker"
    assert argv[1] == "run"

    # Isolation
    assert "--cap-drop" in argv and "ALL" in argv
    assert "--security-opt" in argv and "no-new-privileges" in argv
    assert "--user" in argv and "1000:1000" in argv
    assert "--pids-limit" in argv and "512" in argv
    # Resource limits come from the adapter's configuration
    assert "--cpus" in argv and "2.0" in argv
    assert "--memory" in argv and "512m" in argv
    # Immutable root with scratch space
    assert "--read-only" in argv
    assert argv.count("--tmpfs") == 2
    assert "/tmp" in argv and "/home/app" in argv
    # Labels
    assert f"{LABEL_ORG}=org1" in argv
    assert f"{LABEL_PROJECT}=proj1" in argv
    assert f"{LABEL_ROLE}=workspace" in argv


async def test_start_session_mounts_only_the_work_tree(tmp_path):
    """The git dir must never be visible inside the container (D6)."""
    runner = FakeRunner(container_id="cid")
    adapter = _adapter(tmp_path, runner)
    info = await adapter.create_workspace(
        organization_id="org1", project_id="proj1", template_id="t1", image="node:20-bookworm"
    )
    await adapter.start_session(workspace_external_id=info.external_id)

    argv = runner.last_argv
    mounts = [argv[i + 1] for i, token in enumerate(argv) if token == "-v"]
    # The adapter builds the host path with ``os.path.join``, so the expected
    # string is built the same way rather than hard-coded with "/" — otherwise
    # the assertion fails on Windows, where the separator is a backslash. The
    # security property (exactly one bind mount, and it is the work tree) is
    # platform-neutral.
    expected = os.path.join(str(tmp_path), "org1", "proj1", "work")
    assert mounts == [f"{expected}:/workspace"], (
        "exactly one bind mount, and it is the work tree"
    )
    assert not any("git" in mount for mount in mounts)


async def test_start_session_publishes_no_port(tmp_path):
    """A workspace container is reached with `exec`, never over the network."""
    runner = FakeRunner(container_id="cid")
    adapter = _adapter(tmp_path, runner)
    info = await adapter.create_workspace(
        organization_id="org1", project_id="proj1", template_id="t1", image="node:20-bookworm"
    )
    await adapter.start_session(workspace_external_id=info.external_id)

    argv = runner.last_argv
    assert "-p" not in argv
    assert "--publish" not in argv
    assert not any(token.startswith("0.0.0.0") for token in argv)


async def test_start_session_reuses_an_existing_container(tmp_path):
    """A restart must not stack duplicate containers for one project."""
    runner = FakeRunner()
    runner.responses = [
        ProcessResult(argv=(), exit_code=0, stdout="existingcid\n"),  # ps -a lookup
        ProcessResult(argv=(), exit_code=0, stdout="existingcid\n"),  # docker start
    ]
    adapter = _adapter(tmp_path, runner)
    info = await adapter.create_workspace(
        organization_id="org1", project_id="proj1", template_id="t1", image="node:20-bookworm"
    )
    session = await adapter.start_session(workspace_external_id=info.external_id)

    assert session.external_id == "existingcid"
    assert runner.calls[1][0][1] == "start"
    assert all(call[0][1] != "run" for call in runner.calls), "must not run a second container"


# ---------------------------------------------------------------------------
# run_command
# ---------------------------------------------------------------------------


async def test_run_command_execs_inside_the_container_with_a_workdir(tmp_path):
    runner = FakeRunner(stdout="ok\n")
    adapter = _adapter(tmp_path, runner)

    result = await adapter.run_command(
        session_external_id="cid",
        command="pytest -q",
        timeout_seconds=120,
        max_output_bytes=10_000,
    )

    assert result.stdout == "ok\n"
    argv = runner.last_argv
    assert argv[:3] == ("docker", "exec", "-w")
    assert argv[3] == "/workspace"
    assert argv[-3:] == ("sh", "-lc", "pytest -q")
    assert runner.calls[-1][1]["timeout_seconds"] == 120.0
    assert runner.calls[-1][1]["max_output_bytes"] == 10_000


async def test_run_command_refuses_a_secret_in_the_environment(tmp_path):
    """argv is world-readable via `ps`; credentials must never travel there."""
    adapter = _adapter(tmp_path, FakeRunner())
    with pytest.raises(ValidationError):
        await adapter.run_command(
            session_external_id="cid",
            command="env",
            timeout_seconds=30,
            max_output_bytes=1_000,
            env={"GIT_TOKEN": "ghp_aaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
        )


async def test_run_command_allows_a_harmless_environment_variable(tmp_path):
    runner = FakeRunner()
    adapter = _adapter(tmp_path, runner)
    await adapter.run_command(
        session_external_id="cid",
        command="node -v",
        timeout_seconds=30,
        max_output_bytes=1_000,
        env={"NODE_ENV": "test"},
    )
    argv = runner.last_argv
    assert "-e" in argv and "NODE_ENV=test" in argv


async def test_run_command_refuses_an_empty_command(tmp_path):
    adapter = _adapter(tmp_path, FakeRunner())
    with pytest.raises(ValidationError):
        await adapter.run_command(
            session_external_id="cid", command="   ", timeout_seconds=10, max_output_bytes=100
        )


# ---------------------------------------------------------------------------
# files
# ---------------------------------------------------------------------------


async def test_write_and_read_round_trip_through_the_bind_mount(tmp_path):
    runner = FakeRunner(container_id="cid")
    adapter = _adapter(tmp_path, runner)
    info = await adapter.create_workspace(
        organization_id="org1", project_id="proj1", template_id="t1", image="node:20-bookworm"
    )
    await adapter.start_session(workspace_external_id=info.external_id)

    await adapter.write_file(session_external_id="cid", path="/workspace/src/a.ts", content="x")
    assert await adapter.read_file(session_external_id="cid", path="/workspace/src/a.ts") == "x"
    assert (tmp_path / "org1" / "proj1" / "work" / "src" / "a.ts").read_text() == "x"


async def test_read_file_refuses_an_escape(tmp_path):
    runner = FakeRunner(container_id="cid")
    adapter = _adapter(tmp_path, runner)
    info = await adapter.create_workspace(
        organization_id="org1", project_id="proj1", template_id="t1", image="node:20-bookworm"
    )
    await adapter.start_session(workspace_external_id=info.external_id)

    with pytest.raises(PathEscapeError):
        await adapter.read_file(session_external_id="cid", path="../../../../etc/passwd")


# ---------------------------------------------------------------------------
# lifecycle
# ---------------------------------------------------------------------------


async def test_destroy_removes_the_container_and_its_volumes(tmp_path):
    runner = FakeRunner(container_id="cid")
    adapter = _adapter(tmp_path, runner)
    await adapter.destroy(session_external_id="cid")

    argv = runner.last_argv
    assert argv[:3] == ("docker", "rm", "-f")
    assert "-v" in argv
    assert argv[-1] == "cid"


async def test_pause_and_resume_map_to_stop_and_start(tmp_path):
    runner = FakeRunner()
    adapter = _adapter(tmp_path, runner)
    await adapter.pause(session_external_id="cid")
    assert runner.last_argv[1] == "stop"
    await adapter.resume(session_external_id="cid")
    assert runner.last_argv[1] == "start"


async def test_preview_url_is_loopback_only(tmp_path):
    adapter = _adapter(tmp_path, FakeRunner())
    assert await adapter.get_preview_url(session_external_id="cid", port=8123) == (
        "http://127.0.0.1:8123"
    )


async def test_preview_url_refuses_a_protected_platform_port(tmp_path):
    from app.domain.platform.rules import RuleViolation

    adapter = _adapter(tmp_path, FakeRunner())
    with pytest.raises(RuleViolation):
        await adapter.get_preview_url(session_external_id="cid", port=6379)


# ---------------------------------------------------------------------------
# orphan reaping
# ---------------------------------------------------------------------------


async def test_list_managed_containers_parses_labels(tmp_path):
    runner = FakeRunner(
        ps_stdout=(
            "abc123\tctr-a\tUp 2 hours\t"
            "smart-org.org=org1,smart-org.project=proj1,smart-org.role=workspace\n"
            "def456\tctr-b\tExited (0)\t"
            "smart-org.org=org2,smart-org.project=proj2,smart-org.role=demo\n"
        )
    )
    adapter = _adapter(tmp_path, runner)
    containers = await adapter.list_managed_containers()

    assert [c.container_id for c in containers] == ["abc123", "def456"]
    assert containers[0].organization_id == "org1"
    assert containers[1].role == "demo"
    assert runner.last_argv[1] == "ps"
    assert f"label={LABEL_ROLE}" in runner.last_argv


async def test_list_managed_containers_is_empty_when_the_cli_fails(tmp_path):
    runner = FakeRunner(exit_code=1)
    adapter = _adapter(tmp_path, runner)
    assert await adapter.list_managed_containers() == []


# ---------------------------------------------------------------------------
# the shared process runner (real subprocesses, no container needed)
# ---------------------------------------------------------------------------


async def test_run_process_caps_output_and_reports_truncation():
    result = await run_process(
        ["sh", "-c", "yes abcdefgh | head -c 200000"],
        timeout_seconds=30,
        max_output_bytes=1_000,
    )
    assert result.exit_code == 0
    assert result.truncated is True
    assert len(result.stdout.encode()) <= 1_000


async def test_run_process_kills_the_process_group_on_timeout():
    result = await run_process(["sh", "-c", "sleep 30"], timeout_seconds=1)
    assert result.timed_out is True
    assert result.ok is False


async def test_run_process_reports_a_missing_binary_instead_of_raising():
    result = await run_process(["definitely-not-a-real-binary-xyz"], timeout_seconds=5)
    assert result.spawn_failed is True
    assert result.exit_code == 127
    assert result.spawn_error == "FileNotFoundError"


async def test_run_process_does_not_inherit_the_parent_environment(monkeypatch):
    """A child must not see the platform's secrets (PLAN-v2 §7.5)."""
    monkeypatch.setenv("JWT_SECRET", "super-secret-value-not-for-children")
    monkeypatch.setenv("DATABASE_URL", "postgres://user:pw@host/db")

    result = await run_process(
        ["sh", "-c", "env"], timeout_seconds=10, max_output_bytes=50_000
    )
    assert "super-secret-value-not-for-children" not in result.stdout
    assert "postgres://user:pw@host/db" not in result.stdout


async def test_run_process_streams_output_to_the_callback():
    chunks: list[str] = []

    async def collect(chunk: str) -> None:
        chunks.append(chunk)

    await run_process(
        ["sh", "-c", "printf 'hello'; printf 'world'"],
        timeout_seconds=10,
        on_output=collect,
    )
    assert "".join(chunks) == "helloworld"


# ---------------------------------------------------------------------------
# HTTP surface: infrastructure is owner-only
# ---------------------------------------------------------------------------


async def _owner_org(client: AsyncClient) -> tuple[str, dict]:
    from app.domain.identity.models import PlatformRole
    from tests.conftest import signup

    token = (
        await signup(client, "sandbox-owner@example.com", platform_role=PlatformRole.OWNER)
    )["access_token"]
    headers = auth_header(token)
    org_id = (
        await client.post("/api/organizations", json={"name": "سازمان زیرساخت"}, headers=headers)
    ).json()["id"]
    return org_id, headers


async def _manager_org(client: AsyncClient) -> tuple[str, dict]:
    from tests.conftest import signup

    token = (await signup(client, "sandbox-manager@example.com"))["access_token"]
    headers = auth_header(token)
    org_id = (
        await client.post("/api/organizations", json={"name": "سازمان مدیر"}, headers=headers)
    ).json()["id"]
    return org_id, headers


async def test_org_member_cannot_run_a_sandbox_command(client: AsyncClient):
    """The org's own owner is still not allowed to touch infrastructure."""
    org_id, headers = await _manager_org(client)
    template_id = (
        await client.get(f"/api/organizations/{org_id}/sandbox/templates", headers=headers)
    ).json()[0]["id"]

    response = await client.post(
        f"/api/organizations/{org_id}/sandbox/workspaces",
        json={"project_id": "whatever", "template_id": template_id},
        headers=headers,
    )
    assert response.status_code == 403, response.text


async def test_org_member_may_still_read_sandbox_status(client: AsyncClient):
    """Templates and limits are status, not infrastructure."""
    org_id, headers = await _manager_org(client)
    templates = await client.get(
        f"/api/organizations/{org_id}/sandbox/templates", headers=headers
    )
    assert templates.status_code == 200
    assert templates.json()

    limits = await client.get(
        f"/api/organizations/{org_id}/sandbox/limits", headers=headers
    )
    assert limits.status_code == 200
    assert limits.json()["timeout_seconds"] >= 1


async def test_owner_may_create_a_workspace(client: AsyncClient):
    org_id, headers = await _owner_org(client)
    project_id = (
        await client.post(
            f"/api/organizations/{org_id}/projects",
            json={"name": "پروژهٔ زیرساخت"},
            headers=headers,
        )
    ).json()["id"]
    template_id = (
        await client.get(f"/api/organizations/{org_id}/sandbox/templates", headers=headers)
    ).json()[0]["id"]

    response = await client.post(
        f"/api/organizations/{org_id}/sandbox/workspaces",
        json={"project_id": project_id, "template_id": template_id},
        headers=headers,
    )
    assert response.status_code in (200, 201), response.text


async def test_the_owner_check_helper_rejects_a_manager():
    from app.domain.identity.models import Membership, PlatformRole, Role, User
    from app.interfaces.http.routers.sandbox import _require_platform_owner

    class _Ctx:
        def __init__(self, user):
            self.user = user

    manager = User(email="m@x.com", password_hash="x", platform_role=PlatformRole.MANAGER)
    with pytest.raises(PermissionDeniedError):
        _require_platform_owner(_Ctx(manager))

    owner = User(email="o@x.com", password_hash="x", platform_role=PlatformRole.OWNER)
    _require_platform_owner(_Ctx(owner))  # must not raise

    _ = (Membership, Role)


# ---------------------------------------------------------------------------
# HTTP surface: the owner's sandbox console
# ---------------------------------------------------------------------------


@pytest_asyncio.fixture(autouse=True)
async def _platform_seed(seeded):
    return seeded


async def _owner_org(client: AsyncClient) -> tuple[str, dict]:
    from app.domain.identity.models import PlatformRole
    from tests.conftest import signup

    token = (
        await signup(
            client,
            "sandbox-owner@example.com",
            name="مالک سامانه",
            platform_role=PlatformRole.OWNER,
        )
    )["access_token"]
    headers = auth_header(token)
    org_id = (
        await client.post("/api/organizations", json={"name": "سازمان سندباکس"}, headers=headers)
    ).json()["id"]
    return org_id, headers


async def test_a_non_owner_cannot_list_containers(client: AsyncClient):
    from tests.conftest import signup

    token = (await signup(client, "not-owner-sandbox@example.com"))["access_token"]
    response = await client.get("/api/admin/sandbox/containers", headers=auth_header(token))
    assert response.status_code == 403, response.text


async def test_the_owner_sees_an_empty_list_with_a_mock_provider(
    client: AsyncClient,
):
    """The console must not invent containers it cannot see."""
    _org, headers = await _owner_org(client)
    response = await client.get("/api/admin/sandbox/containers", headers=headers)

    assert response.status_code == 200, response.text
    body = response.json()
    assert body["containers"] == []
    # The image mapping is always available: it is configuration, not state.
    assert "python" in body["images"]


async def test_stopping_a_container_requires_the_id_typed_back(client: AsyncClient):
    _org, headers = await _owner_org(client)
    response = await client.post(
        "/api/admin/sandbox/containers/abc123/stop",
        json={"confirm": "something-else"},
        headers=headers,
    )
    assert response.status_code == 422, response.text
    assert "مطابقت" in response.json()["error"]["message"]