strix/tests/test_mount_free.py

"""Tests for mount-free transport fail-closed semantics and symlink containment."""

from pathlib import Path
from typing import Any

import pytest

import strix.config.loader
from strix.runtime.backends import (
    _BACKENDS,
    _BIND_MOUNT_BACKENDS,
    _MOUNT_FREE_BACKENDS,
    register_backend,
)
from strix.runtime.session_manager import _symlink_safe_dir_entry, create_or_reuse


async def _dummy_backend(*_args: Any, **_kwargs: Any) -> tuple[Any, Any]:
    return None, None


@pytest.mark.asyncio
async def test_mount_free_transport_refusal_raises_error(monkeypatch: pytest.MonkeyPatch) -> None:
    try:
        register_backend(
            "stub_legacy",
            _dummy_backend,
            supports_bind_mounts=True,
            supports_mount_free=False,
        )

        monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "stub_legacy")
        monkeypatch.setenv("STRIX_REQUIRE_MOUNT_FREE", "1")
        monkeypatch.setattr(strix.config.loader, "_cached", None)

        with pytest.raises(
            RuntimeError,
            match="Sandbox backend 'stub_legacy' does not support mount-free transport",
        ):
            await create_or_reuse("scan_123", image="dummy_image", local_sources=[])
    finally:
        _BACKENDS.pop("stub_legacy", None)
        _BIND_MOUNT_BACKENDS.discard("stub_legacy")
        _MOUNT_FREE_BACKENDS.discard("stub_legacy")


@pytest.mark.asyncio
async def test_mount_free_success_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    captured: dict[str, Any] = {}

    async def _stub_backend(
        image: str, manifest: Any, _exposed_ports: Any, bind_mounts: Any
    ) -> tuple[Any, Any]:
        captured["image"] = image
        captured["manifest"] = manifest
        captured["bind_mounts"] = bind_mounts

        class FakeEndpoint:
            host = "127.0.0.1"
            port = 48080
            tls = False

        class FakeSession:
            async def resolve_exposed_port(self, _port: int) -> FakeEndpoint:
                return FakeEndpoint()

        return None, FakeSession()

    async def _fake_bootstrap_caido(*_args: Any, **_kwargs: Any) -> Any:
        return None

    try:
        register_backend(
            "stub_mount_free",
            _stub_backend,
            supports_bind_mounts=True,
            supports_mount_free=True,
        )

        monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "stub_mount_free")
        monkeypatch.setenv("STRIX_REQUIRE_MOUNT_FREE", "1")
        monkeypatch.setattr(strix.config.loader, "_cached", None)
        monkeypatch.setattr("strix.runtime.session_manager.bootstrap_caido", _fake_bootstrap_caido)

        source_dir = tmp_path / "my_project"
        source_dir.mkdir()
        (source_dir / "app.py").write_text("print('test')")

        local_sources = [{"workspace_subdir": "repo", "source_path": str(source_dir)}]
        bundle = await create_or_reuse(
            "scan_success_123", image="dummy_image", local_sources=local_sources
        )

        assert bundle["session"] is not None
        assert captured["bind_mounts"] == []
        assert "repo" in captured["manifest"].entries

        grants = captured["manifest"].extra_path_grants
        assert len(grants) == 1
        assert str(grants[0].path) == str(source_dir.resolve())
        assert grants[0].read_only is True
    finally:
        _BACKENDS.pop("stub_mount_free", None)
        _BIND_MOUNT_BACKENDS.discard("stub_mount_free")
        _MOUNT_FREE_BACKENDS.discard("stub_mount_free")


def test_symlink_safe_dir_entry_skips_out_of_tree_symlinks(tmp_path: Path) -> None:
    outside_dir = tmp_path / "outside"
    outside_dir.mkdir()
    secret_file = outside_dir / "id_rsa"
    secret_file.write_bytes(b"SECRET_KEY_DATA")

    repo_dir = tmp_path / "repo"
    repo_dir.mkdir()
    (repo_dir / "app.py").write_text("print('hello')")

    # Create symlink pointing outside the source root tree
    leak_symlink = repo_dir / "leak_key"
    leak_symlink.symlink_to(secret_file)

    dir_entry = _symlink_safe_dir_entry(repo_dir)

    # Valid files are included, out-of-tree symlinks are skipped!
    assert "app.py" in dir_entry.children
    assert "leak_key" not in dir_entry.children


def test_symlink_safe_dir_entry_prevents_directory_symlink_loops(tmp_path: Path) -> None:
    repo_dir = tmp_path / "repo"
    repo_dir.mkdir()
    (repo_dir / "main.py").write_text("import sys")

    sub_dir = repo_dir / "sub"
    sub_dir.mkdir()

    # Create directory symlink loop (sub/up -> repo)
    (sub_dir / "up").symlink_to(repo_dir, target_is_directory=True)

    # Must complete without RecursionError or infinite loop
    dir_entry = _symlink_safe_dir_entry(repo_dir)

    assert "main.py" in dir_entry.children
    assert "sub" in dir_entry.children