test_safety.py

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from dev_harness import safety


REPO_ROOT = Path(__file__).resolve().parents[3]


def test_state_root_layout_and_sentinel(tmp_path: Path) -> None:
    env = {"OMI_LOCAL_STATE_ROOT": str(tmp_path / "state")}

    layout = safety.create_state_layout(REPO_ROOT, "default", env)

    assert layout.state_root == (tmp_path / "state" / "default").resolve()
    assert layout.sentinel_path.is_file()
    assert layout.process_manifest.parent.is_dir()
    assert layout.port_manifest.parent.is_dir()
    assert layout.config_digest_path.parent == layout.state_root
    assert layout.logs_dir.is_dir()
    assert layout.reports_dir.is_dir()
    assert (layout.services_dir / "firestore").is_dir()
    assert (layout.services_dir / "auth").is_dir()
    assert (layout.services_dir / "redis").is_dir()
    assert (layout.services_dir / "typesense").is_dir()

    sentinel = safety.read_and_validate_sentinel(layout.state_root, repo_root=REPO_ROOT, instance="default")
    assert sentinel["project_id"] == safety.DEFAULT_LOCAL_FIREBASE_PROJECT_ID
    assert sentinel["database_id"] == safety.DEFAULT_FIRESTORE_DATABASE_ID

    layout.sentinel_path.write_text("{}\n", encoding="utf-8")
    with pytest.raises(safety.SafetyError, match="Invalid harness sentinel"):
        safety.read_and_validate_sentinel(layout.state_root, repo_root=REPO_ROOT, instance="default")


def test_project_database_and_loopback_validation() -> None:
    assert safety.validate_project_id("demo-omi-local", require_canonical=True) == "demo-omi-local"
    assert safety.validate_project_id("demo-other") == "demo-other"
    assert safety.validate_database_id("(default)") == "(default)"

    for host in ("127.0.0.1:8085", "localhost:9099", "http://[::1]:9099", "127.12.0.3:1234"):
        assert safety.validate_loopback_emulator_host(host) == host

    with pytest.raises(safety.SafetyError, match="non-demo"):
        safety.validate_project_id("omi-prod")
    with pytest.raises(safety.SafetyError, match="demo-omi-local"):
        safety.validate_project_id("demo-memory", require_canonical=True)
    with pytest.raises(safety.SafetyError, match="database"):
        safety.validate_database_id("customer-data")
    with pytest.raises(safety.SafetyError, match="loopback"):
        safety.validate_loopback_emulator_host("0.0.0.0:8085")
    with pytest.raises(safety.SafetyError, match="loopback"):
        safety.validate_loopback_emulator_host("firestore.googleapis.com:443")


@pytest.mark.parametrize(
    "host",
    (
        "127.attacker.example:8085",
        "127.0.0.1.example:8085",
        "127.0.0.256:8085",
    ),
)
def test_loopback_validation_rejects_ipv4_hostname_lookalikes(host: str) -> None:
    assert safety.is_loopback_host(host) is False
    with pytest.raises(safety.SafetyError, match="loopback"):
        safety.validate_loopback_emulator_host(host)


def test_private_dev_host_accepts_loopback_lan_and_tailnet_rejects_public() -> None:
    # #11774: a physical device reaches the harness over a LAN or Tailscale
    # (CGNAT, 100.64.0.0/10) address. This guard is deliberately separate from
    # validate_loopback_emulator_host above, which protects a different thing
    # (the backend's own connection to its co-located emulators) and must stay
    # loopback-only regardless of this setting.
    for host in ("127.0.0.1", "localhost", "10.1.2.3", "172.16.0.5", "192.168.1.42", "100.105.2.5"):
        assert safety.is_private_dev_host(host) is True
        assert safety.validate_dev_bind_host(host) == host

    for host in ("8.8.8.8", "203.0.113.5", "1.1.1.1", "omi-relay.example.com"):
        assert safety.is_private_dev_host(host) is False
        with pytest.raises(safety.SafetyError, match="private"):
            safety.validate_dev_bind_host(host)


def test_child_environment_strips_cloud_defaults_and_offline_provider_secrets() -> None:
    parent = {
        "PATH": "/usr/bin",
        "HOME": "/home/dev",
        "GOOGLE_APPLICATION_CREDENTIALS": "/prod/service-account.json",
        "GOOGLE_CLOUD_PROJECT": "prod-project",
        "GCLOUD_PROJECT": "prod-project",
        "FIREBASE_CONFIG": "prod-config",
        "SERVICE_ACCOUNT_JSON": "secret",
        "OPENAI_API_KEY": "secret",
        "OMI_VISIBLE": "1",
    }

    env = safety.build_child_env(parent, provider_mode="offline", extra={"FIRESTORE_EMULATOR_HOST": "127.0.0.1:8085"})

    assert env["PATH"] == "/usr/bin"
    assert env["OMI_VISIBLE"] == "1"
    assert env["FIREBASE_PROJECT_ID"] == "demo-omi-local"
    assert env["FIRESTORE_DATABASE_ID"] == "(default)"
    assert env["FIRESTORE_EMULATOR_HOST"] == "127.0.0.1:8085"
    for stripped in (
        "GOOGLE_APPLICATION_CREDENTIALS",
        "GOOGLE_CLOUD_PROJECT",
        "GCLOUD_PROJECT",
        "FIREBASE_CONFIG",
        "SERVICE_ACCOUNT_JSON",
        "OPENAI_API_KEY",
    ):
        assert stripped not in env

    with pytest.raises(safety.SafetyError, match="unsafe child environment"):
        safety.build_child_env(parent, extra={"GOOGLE_APPLICATION_CREDENTIALS": "/tmp/key.json"})
    env_with_backend_secret = safety.build_child_env(
        parent,
        provider_mode="offline",
        extra={
            "ENCRYPTION_SECRET": "local-only-test-secret",
            "ADMIN_KEY": "local-admin",
            "TYPESENSE_API_KEY": "local-typesense",
        },
    )
    assert env_with_backend_secret["ENCRYPTION_SECRET"] == "local-only-test-secret"
    assert env_with_backend_secret["ADMIN_KEY"] == "local-admin"
    assert env_with_backend_secret["TYPESENSE_API_KEY"] == "local-typesense"

    with pytest.raises(safety.SafetyError, match="provider credential"):
        safety.build_child_env(parent, provider_mode="offline", extra={"DEEPGRAM_API_KEY": "x"})


def test_destructive_path_guard_rejects_dangerous_paths(tmp_path: Path) -> None:
    env = {"OMI_LOCAL_STATE_ROOT": str(tmp_path / "state")}
    layout = safety.create_state_layout(REPO_ROOT, "default", env)
    owned_child = layout.state_root / "services" / "firestore"

    assert (
        safety.validate_destructive_target(owned_child, state_root=layout.state_root, repo_root=REPO_ROOT)
        == owned_child.resolve()
    )

    dangerous = [Path("/"), Path.home(), REPO_ROOT, tmp_path / "outside"]
    for target in dangerous:
        with pytest.raises(safety.SafetyError):
            safety.validate_destructive_target(target, state_root=layout.state_root, repo_root=REPO_ROOT)

    (layout.state_root / safety.HARNESS_SENTINEL_FILENAME).unlink()
    with pytest.raises(safety.SafetyError, match="Missing harness ownership sentinel"):
        safety.validate_destructive_target(owned_child, state_root=layout.state_root, repo_root=REPO_ROOT)


def _write_manifests(layout: safety.HarnessLayout, process: dict[str, object], port: dict[str, object]) -> None:
    layout.process_manifest.parent.mkdir(parents=True, exist_ok=True)
    layout.process_manifest.write_text(json.dumps({"processes": [process]}) + "\n", encoding="utf-8")
    layout.port_manifest.write_text(json.dumps({"ports": [port]}) + "\n", encoding="utf-8")


def test_process_exists_observes_live_and_exited_processes() -> None:
    proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
    try:
        assert safety.process_exists(os.getpid())
        assert safety.process_exists(proc.pid)
    finally:
        proc.terminate()
        proc.wait(timeout=5)

    assert not safety.process_exists(proc.pid)


def test_windows_process_probe_closes_each_open_handle(monkeypatch: pytest.MonkeyPatch) -> None:
    if os.name != "nt":
        pytest.skip("Windows process handles are not used on this platform")

    closed: list[int] = []
    monkeypatch.setattr(safety, "_open_process", lambda _access, _inherit, _pid: 123)
    monkeypatch.setattr(safety, "_close_handle", lambda handle: closed.append(handle) or True)
    for wait_result, expected in ((safety._WAIT_OBJECT_0, False), (safety._WAIT_TIMEOUT, True)):
        monkeypatch.setattr(safety, "_wait_for_single_object", lambda _handle, _timeout, result=wait_result: result)
        assert safety.process_exists(456) is expected

    assert closed == [123, 123]


def test_windows_process_probe_reports_close_failure(monkeypatch: pytest.MonkeyPatch) -> None:
    if os.name != "nt":
        pytest.skip("Windows process handles are not used on this platform")

    monkeypatch.setattr(safety, "_open_process", lambda _access, _inherit, _pid: 123)
    monkeypatch.setattr(safety, "_wait_for_single_object", lambda _handle, _timeout: safety._WAIT_TIMEOUT)
    monkeypatch.setattr(safety, "_close_handle", lambda _handle: False)
    monkeypatch.setattr(safety.ctypes, "get_last_error", lambda: 6)

    with pytest.raises(OSError):
        safety.process_exists(456)


def test_windows_command_line_probe_ignores_path_shadowing(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    if os.name != "nt":
        pytest.skip("Windows PowerShell lookup is not used on this platform")

    marker = "omi-harness-system-powershell-marker"
    proc = subprocess.Popen([sys.executable, "-c", f"import time; time.sleep(60)  # {marker}"])
    fake_bin = tmp_path / "fake-bin"
    fake_bin.mkdir()
    shutil.copy2(sys.executable, fake_bin / "powershell.exe")
    monkeypatch.setenv("PATH", str(fake_bin) + os.pathsep + os.environ.get("PATH", ""))
    try:
        assert marker in safety.command_line_for_pid(proc.pid)
    finally:
        proc.terminate()
        proc.wait(timeout=5)


def test_foreign_pid_and_port_are_rejected(tmp_path: Path) -> None:
    env = {"OMI_LOCAL_STATE_ROOT": str(tmp_path / "state")}
    layout = safety.create_state_layout(REPO_ROOT, "default", env)
    marker = "omi-harness-test-owned-marker"
    proc = subprocess.Popen([sys.executable, "-c", f"import time; time.sleep(60)  # {marker}"])
    foreign_proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
    try:
        time.sleep(0.1)
        _write_manifests(
            layout,
            {"service": "backend", "pid": proc.pid, "ownership_marker": marker},
            {"service": "backend", "port": 49152, "pid": proc.pid},
        )

        assert (
            safety.validate_owned_pid(proc.pid, process_manifest=layout.process_manifest, service="backend")["pid"]
            == proc.pid
        )
        assert (
            safety.validate_port_owner(
                49152,
                pid=proc.pid,
                port_manifest=layout.port_manifest,
                process_manifest=layout.process_manifest,
                service="backend",
            )["port"]
            == 49152
        )

        with pytest.raises(safety.SafetyError, match="foreign PID"):
            safety.validate_owned_pid(foreign_proc.pid, process_manifest=layout.process_manifest)
        with pytest.raises(safety.SafetyError, match="Foreign process owns port"):
            safety.validate_port_owner(49152, pid=foreign_proc.pid, port_manifest=layout.port_manifest)

        _write_manifests(
            layout,
            {"service": "backend", "pid": foreign_proc.pid, "ownership_marker": "not-in-command-line"},
            {"service": "backend", "port": 49153, "pid": foreign_proc.pid},
        )
        with pytest.raises(safety.SafetyError, match="ownership marker"):
            safety.validate_owned_pid(foreign_proc.pid, process_manifest=layout.process_manifest)
    finally:
        for running in (proc, foreign_proc):
            running.terminate()
            try:
                running.wait(timeout=5)
            except subprocess.TimeoutExpired:
                running.kill()
                running.wait(timeout=5)


def test_redis_reset_guard_refuses_shared_redis(tmp_path: Path) -> None:
    layout = safety.create_state_layout(REPO_ROOT, "default", {"OMI_LOCAL_STATE_ROOT": str(tmp_path / "state")})

    assert (
        safety.validate_redis_reset_target(
            "redis://127.0.0.1:6379/0?omi_instance=default", state_root=layout.state_root, expected_instance="default"
        )
        == "redis://127.0.0.1:6379/0?omi_instance=default"
    )
    with pytest.raises(safety.SafetyError, match="shared Redis"):
        safety.validate_redis_reset_target(
            "redis://127.0.0.1:6379/0", state_root=layout.state_root, expected_instance="default"
        )
    with pytest.raises(safety.SafetyError, match="loopback"):
        safety.validate_redis_reset_target(
            "redis://redis.internal:6379/0?omi_instance=default",
            state_root=layout.state_root,
            expected_instance="default",
        )