from __future__ import annotations

import shutil
import subprocess
from pathlib import Path

import pytest

import sandbox

needs_bwrap = pytest.mark.skipif(shutil.which("bwrap") is None, reason="bubblewrap is required")


def _jail(tmp_path: Path, command: str = "ls", **kwargs) -> sandbox.Sandboxed:
    options = {"mode": "workspace-write", "network_access": False, "cwd": tmp_path}
    options.update(kwargs)
    return sandbox.sandbox_command(command, workspace=tmp_path, **options)


def _argv(tmp_path: Path, command: str = "ls", **kwargs) -> list[str]:
    return _jail(tmp_path, command, **kwargs).argv


def _ro_binds(argv: list[str]) -> list[tuple[str, str]]:
    return [(argv[i + 1], argv[i + 2]) for i, a in enumerate(argv) if a == "--ro-bind"]


def test_the_command_runs_inside_bubblewrap(tmp_path: Path) -> None:
    argv = _argv(tmp_path, "make")
    assert Path(argv[0]).name == "bwrap"
    assert argv[-3:] == ["/bin/bash", "-c", "make"]


def test_an_unknown_mode_is_refused(tmp_path: Path) -> None:
    with pytest.raises(ValueError, match="unknown sandbox mode"):
        _argv(tmp_path, mode="yolo")


def test_full_access_is_no_sandbox_at_all(tmp_path: Path) -> None:
    assert _argv(tmp_path, "make", mode="full-access") == ["bash", "-c", "make"]


def test_the_workspace_is_the_only_writable_mount(tmp_path: Path) -> None:
    argv = _argv(tmp_path)
    assert argv[argv.index("--bind") + 1 : argv.index("--bind") + 3] == [str(tmp_path), str(tmp_path)]
    assert argv.count("--bind") == 1


def test_read_only_mode_mounts_the_workspace_read_only(tmp_path: Path) -> None:
    argv = _argv(tmp_path, mode="read-only")
    assert "--bind" not in argv
    assert (str(tmp_path), str(tmp_path)) in _ro_binds(argv)


def test_the_namespaces_that_contain_the_command_are_requested(tmp_path: Path) -> None:
    argv = _argv(tmp_path)
    for flag in (
        "--unshare-pid",
        "--as-pid-1",
        "--unshare-ipc",
        "--unshare-user",
        "--die-with-parent",
        "--new-session",
    ):
        assert flag in argv


def test_the_session_bus_directory_is_replaced_by_a_tmpfs(tmp_path: Path) -> None:
    argv = _argv(tmp_path)
    assert argv[argv.index("--tmpfs") : argv.index("--tmpfs") + 2] == ["--tmpfs", "/tmp"]
    assert "/run" in argv


@pytest.mark.parametrize(("network_access", "unshared"), [(False, True), (True, False)])
def test_the_network_namespace_follows_the_network_setting(
    tmp_path: Path, network_access: bool, unshared: bool
) -> None:
    assert ("--unshare-net" in _argv(tmp_path, network_access=network_access)) is unshared


def test_extra_readable_roots_are_mounted_read_only(tmp_path: Path) -> None:
    extra = tmp_path.parent / "toolchain"
    extra.mkdir(exist_ok=True)
    argv = _argv(tmp_path, readable_roots=(str(extra),))
    assert (str(extra), str(extra)) in _ro_binds(argv)


def test_a_missing_readable_root_is_skipped_rather_than_failing(tmp_path: Path) -> None:
    argv = _argv(tmp_path, readable_roots=(str(tmp_path / "absent"),))
    assert str(tmp_path / "absent") not in argv


def test_the_environment_is_cleared_and_rebuilt(tmp_path: Path) -> None:
    argv = _argv(tmp_path)
    assert "--clearenv" in argv
    assert argv.index("--clearenv") < argv.index("--setenv")


def test_a_secret_in_the_parent_environment_never_reaches_the_child(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("HARNESS_LAND_TOKEN", "super-secret")
    monkeypatch.setenv("OPENAI_API_KEY", "sk-secret")
    jail = _jail(tmp_path)
    rendered = " ".join(jail.argv) + " " + " ".join(f"{k}={v}" for k, v in jail.env.items())
    assert "super-secret" not in rendered
    assert "sk-secret" not in rendered


def test_path_is_fixed_rather_than_inherited(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """The operator's PATH points into ~/.local/bin and ~/.claude/bin, which hold
    privileged helpers such as deck-sudo."""
    monkeypatch.setenv("PATH", f"{Path.home()}/.local/bin:/usr/bin")
    assert sandbox.child_env(tmp_path)["PATH"] == sandbox.CHILD_PATH
    assert str(Path.home()) not in sandbox.child_env(tmp_path)["PATH"]


def test_home_points_at_the_workspace_not_the_operators_home(tmp_path: Path) -> None:
    assert sandbox.child_env(tmp_path)["HOME"] == str(tmp_path)


@needs_bwrap
def test_the_jail_reports_itself_enforced(tmp_path: Path) -> None:
    assert sandbox.enforcement_holds(tmp_path, mode="workspace-write", network_access=False) == (
        True,
        "sandbox enforced",
    )


@needs_bwrap
def test_a_readable_root_that_exposes_the_canary_refuses_to_serve(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The probes must run against the jail as configured. A readable root wide
    enough to expose the canary is a readable root wide enough to expose the
    operator's keys, and the startup verdict has to say so."""
    state = tmp_path / "state"
    monkeypatch.setattr(sandbox, "state_dir", lambda: state)
    workspace = tmp_path / "project"
    workspace.mkdir()
    enforced, detail = sandbox.enforcement_holds(
        workspace, mode="workspace-write", network_access=False, readable_roots=(str(state),)
    )
    assert (enforced, "outside the workspace" in detail) == (False, True)


def _run(tmp_path: Path, command: str, **kwargs) -> subprocess.CompletedProcess[str]:
    jail = _jail(tmp_path, command, **kwargs)
    return subprocess.run(
        jail.argv, env=jail.env, cwd=tmp_path, capture_output=True, text=True, timeout=60
    )


@needs_bwrap
def test_a_credential_outside_the_workspace_cannot_be_read(tmp_path: Path) -> None:
    secret = tmp_path.parent / "id_ed25519"
    secret.write_text("PRIVATE KEY MATERIAL", encoding="utf-8")
    assert "PRIVATE KEY MATERIAL" not in _run(tmp_path, f"cat {secret}").stdout


@needs_bwrap
def test_a_write_outside_the_workspace_never_reaches_the_host(tmp_path: Path) -> None:
    target = tmp_path.parent / "breach"
    _run(tmp_path, f"printf breach > {target}")
    assert not target.exists()


@needs_bwrap
def test_the_session_dbus_socket_is_unreachable(tmp_path: Path) -> None:
    """Reachable, it lets the command ask systemd --user to fork a process the
    jail never contained."""
    assert _run(tmp_path, "test -S /run/user/$(id -u)/bus").returncode != 0


@needs_bwrap
def test_systemd_run_cannot_start_an_unconfined_process(tmp_path: Path) -> None:
    escape = tmp_path.parent / "escaped"
    _run(
        tmp_path,
        f"XDG_RUNTIME_DIR=/run/user/$(id -u) systemd-run --user --pipe --wait /bin/sh -c 'touch {escape}'",
    )
    assert not escape.exists()


@needs_bwrap
def test_privileged_helpers_on_the_operators_path_do_not_resolve(tmp_path: Path) -> None:
    assert _run(tmp_path, "command -v deck-sudo").returncode != 0


@needs_bwrap
def test_the_network_is_unreachable_unless_granted(tmp_path: Path) -> None:
    assert _run(tmp_path, "getent hosts api.openai.com").returncode != 0
    assert _run(tmp_path, "getent hosts api.openai.com", network_access=True).returncode == 0


@needs_bwrap
def test_a_write_inside_the_workspace_works(tmp_path: Path) -> None:
    _run(tmp_path, "printf ok > artifact.txt")
    assert (tmp_path / "artifact.txt").read_text() == "ok"


@needs_bwrap
def test_read_only_mode_refuses_a_write_inside_the_workspace(tmp_path: Path) -> None:
    _run(tmp_path, "printf ok > artifact.txt", mode="read-only")
    assert not (tmp_path / "artifact.txt").exists()


@needs_bwrap
def test_a_symlink_pointing_out_of_the_workspace_leads_nowhere(tmp_path: Path) -> None:
    secret = tmp_path.parent / "outside-secret"
    secret.write_text("SECRET", encoding="utf-8")
    (tmp_path / "link").symlink_to(secret)
    assert "SECRET" not in _run(tmp_path, "cat link").stdout


@needs_bwrap
def test_the_launching_environment_is_not_readable_through_proc(tmp_path: Path) -> None:
    """bwrap's own process lives inside the jail's pid namespace, so anything in
    the environment it was launched with is one `cat /proc/1/environ` away."""
    jail = _jail(tmp_path, "cat /proc/*/environ 2>/dev/null | tr '\\0' '\\n'")
    result = subprocess.run(
        jail.argv,
        env={**jail.env, "GPTBRIDGE_PROC_CANARY": "canary-value"},
        cwd=tmp_path,
        capture_output=True,
        text=True,
        timeout=60,
    )
    assert "canary-value" not in result.stdout