Tirith security scanner installation and stability issues

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #23050 fix(tirith): add Windows support for auto-install and command scanning

  merged to main

• #26618 fix(windows): stop spamming cwd-missing + tirith-spawn warnings on every terminal call

  merged to main

  files: tests/tools/test_local_env_windows_msys.py, tests/tools/test_tirith_security.py, tools/environments/local.py, tools/tirith_security.py

  Show the diff

  • tests/tools/test_local_env_windows_msys.py

    @@ -0,0 +1,200 @@
    +"""Tests for the Windows / Git Bash MSYS-path normalization in
    +``LocalEnvironment``.
    +
    +Background
    +----------
    +On Windows, ``pwd -P`` inside Git Bash emits paths like
    +``/c/Users/NVIDIA``. ``subprocess.Popen(..., cwd=...)`` only accepts
    +native Windows paths (``C:\\Users\\NVIDIA``), and the validation done
    +by ``_resolve_safe_cwd`` was also checking the MSYS form against
    +``os.path.isdir``, which returns ``False`` on Windows. The combined
    +effect was a warning logged on every single terminal call:
    +
    +    LocalEnvironment cwd '/c/Users/NVIDIA' is missing on disk;
    +    falling back to '/' so terminal commands keep working.
    +
    +These tests fake the Windows env on Linux CI by patching ``_IS_WINDOWS``
    +and ``os.path.isdir`` so the MSYS path tests as "missing" exactly like
    +on the real OS.
    +"""
    +
    +import os
    +from unittest.mock import patch
    +
    +import pytest
    +
    +from tools.environments import local as local_mod
    +from tools.environments.local import (
    +    LocalEnvironment,
    +    _msys_to_windows_path,
    +    _resolve_safe_cwd,
    +)
    +
    +
    +# ---------------------------------------------------------------------------
    +# _msys_to_windows_path — pure-function unit tests
    +# ---------------------------------------------------------------------------
    +
    +class TestMsysToWindowsPath:
    +    def test_noop_on_non_windows(self, monkeypatch):
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", False)
    +        # On a non-Windows host the function must never rewrite the path
    +        # — POSIX-style paths are real paths there.
    +        assert _msys_to_windows_path("/c/Users/NVIDIA") == "/c/Users/NVIDIA"
    +        assert _msys_to_windows_path("/home/teknium") == "/home/teknium"
    +
    +    def test_translates_drive_path(self, monkeypatch):
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +        assert _msys_to_windows_path("/c/Users/NVIDIA") == r"C:\Users\NVIDIA"
    +        assert _msys_to_windows_path("/d/Projects/foo bar") == r"D:\Projects\foo bar"
    +
    +    def test_translates_bare_drive_root(self, monkeypatch):
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +        # Bare "/c" alone should resolve to the drive root.
    +        assert _msys_to_windows_path("/c") == "C:\\"
    +        # Trailing slash on the drive letter is also a root.
    +        assert _msys_to_windows_path("/c/") == "C:\\"
    +
    +    def test_idempotent_on_already_windows_path(self, monkeypatch):
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +        assert _msys_to_windows_path(r"C:\Users\NVIDIA") == r"C:\Users\NVIDIA"
    +
    +    def test_does_not_translate_multi_char_first_segment(self, monkeypatch):
    +        """``/tmp/foo`` and ``/home/x`` must NOT be misread as drive paths
    +        just because they start with ``/`` and a single letter — the regex
    +        only matches when the first segment is exactly one character."""
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +        assert _msys_to_windows_path("/tmp/foo") == "/tmp/foo"
    +        assert _msys_to_windows_path("/home/x") == "/home/x"
    +
    +    def test_empty_string(self, monkeypatch):
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +        assert _msys_to_windows_path("") == ""
    +
    +
    +# ---------------------------------------------------------------------------
    +# _resolve_safe_cwd — Windows fast path
    +# ---------------------------------------------------------------------------
    +
    +class TestResolveSafeCwdWindows:
    +    def test_msys_path_resolves_to_native_when_native_exists(
    +        self, monkeypatch, tmp_path,
    +    ):
    +        """The whole point of this fix: a Git Bash ``/c/Users/x`` value
    +        should resolve to its native equivalent if that native dir exists,
    +        WITHOUT falling back to the temp dir."""
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +
    +        # tmp_path is a real native dir on the test host. Build a fake
    +        # MSYS form pointing at it and prove the resolver finds it.
    +        native = str(tmp_path)
    +        # Construct a synthetic MSYS form for whatever tmp_path is.
    +        # On Linux CI tmp_path is /tmp/... ; the resolver shouldn't even
    +        # try to translate that (regex won't match), so emulate the
    +        # mapping by pointing the translator at the real native dir.
    +        with patch.object(
    +            local_mod, "_msys_to_windows_path", return_value=native
    +        ):
    +            assert _resolve_safe_cwd("/c/whatever") == native
    +
    +
    +# ---------------------------------------------------------------------------
    +# End-to-end: _update_cwd via marker file (Windows simulation)
    +# ---------------------------------------------------------------------------
    +
    +class TestUpdateCwdWindowsMsys:
    +    def test_marker_file_msys_path_stored_in_native_form(
    +        self, monkeypatch, tmp_path,
    +    ):
    +        """When Git Bash writes ``/c/Users/x`` to the cwd marker file on
    +        Windows, ``_update_cwd`` must translate to native form before
    +        validating and storing — otherwise ``os.path.isdir`` rejects a
    +        perfectly real directory."""
    +        original = tmp_path / "starting"
    +        original.mkdir()
    +
    +        # Fake Windows for the test
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +
    +        with patch.object(
    +            LocalEnvironment, "init_session", autospec=True, return_value=None
    +        ):
    +            env = LocalEnvironment(cwd=str(original), timeout=10)
    +
    +        # Pretend Git Bash wrote an MSYS path that maps to tmp_path/"next"
    +        new_dir = tmp_path / "next"
    +        new_dir.mkdir()
    +
    +        with open(env._cwd_file, "w") as f:
    +            f.write("/c/whatever/from/bash")
    +
    +        # Translate the synthetic MSYS string to the real native dir.
    +        def fake_translate(p):
    +            if p == "/c/whatever/from/bash":
    +                return str(new_dir)
    +            return p
    +
    +        with patch.object(local_mod, "_msys_to_windows_path", side_effect=fake_translate):
    +            env._update_cwd({"output": "", "returncode": 0})
    +
    +        assert env.cwd == str(new_dir)
    +
    +
    +# ---------------------------------------------------------------------------
    +# End-to-end: _extract_cwd_from_output rollback when marker is invalid
    +# ---------------------------------------------------------------------------
    +
    +class TestExtractCwdFromOutputWindowsMsys:
    +    def test_stale_msys_marker_does_not_clobber_cwd(self, monkeypatch, tmp_path):
    +        """When the cwd marker in stdout points at a non-existent path,
    +        ``LocalEnvironment._extract_cwd_from_output`` must roll back to
    +        the previous cwd instead of propagating a bad value."""
    +        original = tmp_path / "starting"
    +        original.mkdir()
    +
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +
    +        with patch.object(
    +            LocalEnvironment, "init_session", autospec=True, return_value=None
    +        ):
    +            env = LocalEnvironment(cwd=str(original), timeout=10)
    +
    +        marker = env._cwd_marker
    +        result = {
    +            "output": f"some command output\n{marker}/c/no/such/path{marker}\n",
    +            "returncode": 0,
    +        }
    +
    +        # Translation produces a path that doesn't exist on disk → rollback.
    +        with patch.object(
    +            local_mod,
    +            "_msys_to_windows_path",
    +            return_value=str(tmp_path / "definitely-does-not-exist"),
    +        ):
    +            env._extract_cwd_from_output(result)
    +
    +        assert env.cwd == str(original)
    +
    +    def test_valid_msys_marker_normalized_to_native(self, monkeypatch, tmp_path):
    +        original = tmp_path / "starting"
    +        original.mkdir()
    +        new_dir = tmp_path / "next"
    +        new_dir.mkdir()
    +
    +        monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
    +
    +        with patch.object(
    +            LocalEnvironment, "init_session", autospec=True, return_value=None
    +        ):
    +            env = LocalEnvironment(cwd=str(original), timeout=10)
    +
    +        marker = env._cwd_marker
    +        result = {
    +            "output": f"x\n{marker}/c/whatever{marker}\n",
    +            "returncode": 0,
    +        }
    +
    +        with patch.object(local_mod, "_msys_to_windows_path", return_value=str(new_dir)):
    +            env._extract_cwd_from_output(result)
    +
    +        assert env.cwd == str(new_dir)

  • tests/tools/test_tirith_security.py

    @@ -1007,3 +1007,120 @@ def test_get_hermes_home_fallback(self):
                 expected = os.path.join(os.path.expanduser("~"), ".hermes")
                 result = _get_hermes_home()
             assert result == expected
    +
    +
    +# ---------------------------------------------------------------------------
    +# Warn-once dedupe (issue: tirith spawn failed spamming on Windows)
    +# ---------------------------------------------------------------------------
    +
    +class TestSpawnWarningDedup:
    +    """When tirith isn't installed yet (background install in flight, or
    +    install marked failed), every terminal command spammed an identical
    +    ``tirith spawn failed: [WinError 2]`` warning to ``errors.log``. The
    +    dedupe set in ``_warn_once`` collapses repeats by ``(exc class, errno)``
    +    while still surfacing the first occurrence so users see the failure.
    +    """
    +
    +    @patch("tools.tirith_security.subprocess.run")
    +    @patch("tools.tirith_security._load_security_config")
    +    def test_repeated_spawn_failure_logs_once(self, mock_cfg, mock_run, caplog):
    +        mock_cfg.return_value = {
    +            "tirith_enabled": True, "tirith_path": "tirith",
    +            "tirith_timeout": 5, "tirith_fail_open": True,
    +        }
    +        mock_run.side_effect = FileNotFoundError("[WinError 2]")
    +        # Fresh dedupe state — clear any keys left by other tests.
    +        _tirith_mod._reset_spawn_warning_state()
    +
    +        with caplog.at_level("WARNING", logger="tools.tirith_security"):
    +            for _ in range(15):
    +                result = check_command_security("echo hi")
    +                # Behavior must remain the same on every call —
    +                # fail-open allow, with the exception captured in summary.
    +                assert result["action"] == "allow"
    +                assert "unavailable" in result["summary"]
    +
    +        spawn_warnings = [
    +            rec for rec in caplog.records
    +            if "tirith spawn failed" in rec.message
    +        ]
    +        assert len(spawn_warnings) == 1, (
    +            f"expected exactly 1 spawn-failed warning across 15 commands, "
    +            f"got {len(spawn_warnings)}: {[r.message for r in spawn_warnings]}"
    +        )
    +
    +    @patch("tools.tirith_security.subprocess.run")
    +    @patch("tools.tirith_security._load_security_config")
    +    def test_distinct_exception_types_each_log_once(self, mock_cfg, mock_run, caplog):
    +        """``FileNotFoundError`` and ``PermissionError`` are distinct
    +        failure modes and each deserves its own first-occurrence log
    +        line; the dedupe key includes the exception class."""
    +        mock_cfg.return_value = {
    +            "tirith_enabled": True, "tirith_path": "tirith",
    +            "tirith_timeout": 5, "tirith_fail_open": True,
    +        }
    +        _tirith_mod._reset_spawn_warning_state()
    +
    +        with caplog.at_level("WARNING", logger="tools.tirith_security"):
    +            mock_run.side_effect = FileNotFoundError("[WinError 2]")
    +            for _ in range(3):
    +                check_command_security("a")
    +            mock_run.side_effect = PermissionError("denied")
    +            for _ in range(3):
    +                check_command_security("b")
    +
    +        spawn_warnings = [
    +            rec for rec in caplog.records
    +            if "tirith spawn failed" in rec.message
    +        ]
    +        assert len(spawn_warnings) == 2, (
    +            f"expected 2 distinct first-occurrence warnings, "
    +            f"got {len(spawn_warnings)}"
    +        )
    +
    +    @patch("tools.tirith_security.subprocess.run")
    +    @patch("tools.tirith_security._load_security_config")
    +    def test_repeated_timeout_logs_once(self, mock_cfg, mock_run, caplog):
    +        mock_cfg.return_value = {
    +            "tirith_enabled": True, "tirith_path": "tirith",
    +            "tirith_timeout": 5, "tirith_fail_open": True,
    +        }
    +        mock_run.side_effect = subprocess.TimeoutExpired(cmd="tirith", timeout=5)
    +        _tirith_mod._reset_spawn_warning_state()
    +
    +        with caplog.at_level("WARNING", logger="tools.tirith_security"):
    +            for _ in range(10):
    +                result = check_command_security("slow")
    +                assert result["action"] == "allow"
    +
    +        timeout_warnings = [
    +            rec for rec in caplog.records
    +            if "tirith timed out" in rec.message
    +        ]
    +        assert len(timeout_warnings) == 1
    +
    +    @patch("tools.tirith_security._load_security_config")
    +    def test_path_none_logs_once(self, mock_cfg, caplog):
    +        """``_resolve_tirith_path`` returning ``None`` (explicit path set
    +        but resolver returned None — unusual) should not spam the log
    +        either."""
    +        mock_cfg.return_value = {
    +            "tirith_enabled": True, "tirith_path": "tirith",
    +            "tirith_timeout": 5, "tirith_fail_open": True,
    +        }
    +        _tirith_mod._reset_spawn_warning_state()
    +
    +        with patch(
    +            "tools.tirith_security._resolve_tirith_path", return_value=None
    +        ):
    +            with caplog.at_level("WARNING", logger="tools.tirith_security"):
    +                for _ in range(10):
    +                    result = check_command_security("echo")
    +                    assert result["action"] == "allow"
    +                    assert "tirith path unavailable" in result["summary"]
    +
    +        none_warnings = [
    +            rec for rec in caplog.records
    +            if "tirith path resolved to None" in rec.message
    +        ]
    +        assert len(none_warnings) == 1

  • tools/environments/local.py

    @@ -18,18 +18,44 @@
     logger = logging.getLogger(__name__)
     
     
    +def _msys_to_windows_path(cwd: str) -> str:
    +    """Translate a Git Bash / MSYS-style POSIX path (``/c/Users/x``) to the
    +    native Windows form (``C:\\Users\\x``) so ``os.path.isdir`` and
    +    ``subprocess.Popen(..., cwd=...)`` can find it.
    +
    +    No-ops on non-Windows hosts or for paths that aren't in MSYS form.
    +    Returns the input unchanged when no translation applies. This is
    +    idempotent — calling it on an already-Windows path returns it as-is.
    +    """
    +    if not _IS_WINDOWS or not cwd:
    +        return cwd
    +    # Match leading "/<single letter>/" or exactly "/<letter>" (bare drive root).
    +    m = re.match(r'^/([a-zA-Z])(/.*)?$', cwd)
    +    if not m:
    +        return cwd
    +    drive = m.group(1).upper()
    +    tail = (m.group(2) or "").replace('/', '\\')
    +    return f"{drive}:{tail or chr(92)}"  # chr(92) = backslash, avoid raw-string escape
    +
    +
     def _resolve_safe_cwd(cwd: str) -> str:
         """Return ``cwd`` if it exists as a directory, else the nearest existing
         ancestor.  Falls back to ``tempfile.gettempdir()`` only if walking up the
         path can't find any existing directory (effectively never on a healthy
         filesystem, but cheap belt-and-braces).
     
    +    On Windows, also normalizes Git Bash / MSYS-style POSIX paths
    +    (``/c/Users/x``) to native Windows form before the isdir check so a
    +    perfectly valid ``pwd -P`` result from bash doesn't get rejected as
    +    "missing" (see ``_msys_to_windows_path``).
    +
         Used by ``_run_bash`` to recover when the configured cwd is gone — most
         commonly because a previous tool call deleted its own working directory
         (issue #17558).  Without this guard, ``subprocess.Popen(..., cwd=...)``
         raises ``FileNotFoundError`` before bash starts, wedging every subsequent
         terminal call until the gateway restarts.
         """
    +    cwd = _msys_to_windows_path(cwd) if _IS_WINDOWS else cwd
         if cwd and os.path.isdir(cwd):
             return cwd
         parent = os.path.dirname(cwd) if cwd else ""
    @@ -455,21 +481,27 @@ def _run_bash(self, cmd_string: str, *, login: bool = False,
             # (issue #17558).  Popen would otherwise raise FileNotFoundError on
             # the cwd before bash starts, wedging every subsequent call until the
             # gateway restarts.
    +        #
    +        # On Windows, ``_resolve_safe_cwd`` also normalises Git Bash-style
    +        # POSIX paths (``/c/Users/...``) to native form so a perfectly valid
    +        # ``pwd -P`` result from bash isn't mistakenly treated as "missing"
    +        # and spammed as a warning on every command.
             safe_cwd = _resolve_safe_cwd(self.cwd)
             if safe_cwd != self.cwd:
    -            logger.warning(
    -                "LocalEnvironment cwd %r is missing on disk; "
    -                "falling back to %r so terminal commands keep working.",
    -                self.cwd,
    -                safe_cwd,
    -            )
    +            # MSYS → Windows translation alone shouldn't surface as a warning
    +            # (it's a benign normalization, not a recovery). Only warn when
    +            # the directory really doesn't exist on disk.
    +            normalized = _msys_to_windows_path(self.cwd) if _IS_WINDOWS else self.cwd
    +            if safe_cwd != normalized:
    +                logger.warning(
    +                    "LocalEnvironment cwd %r is missing on disk; "
    +                    "falling back to %r so terminal commands keep working.",
    +                    self.cwd,
    +                    safe_cwd,
    +                )
                 self.cwd = safe_cwd
     
    -        # On Windows, self.cwd may be a Git Bash-style path (/c/Users/...)
    -        # from pwd output. subprocess.Popen needs a native Windows path.
             _popen_cwd = self.cwd
    -        if _IS_WINDOWS and _popen_cwd and re.match(r'^/[a-zA-Z]/', _popen_cwd):
    -            _popen_cwd = _popen_cwd[1].upper() + ':' + _popen_cwd[2:].replace('/', '\\')
     
             proc = subprocess.Popen(
                 args,
    @@ -571,10 +603,19 @@ def _update_cwd(self, result: dict):
             ``pwd -P`` on a deleted cwd can leave a stale value in the marker
             file, and propagating it would re-wedge the next ``Popen``.  The
             ``_run_bash`` recovery path will resolve a safe fallback if needed.
    +
    +        On Windows, the value written by Git Bash's ``pwd -P`` is in
    +        MSYS form (``/c/Users/x``). Translate it to native Windows form
    +        before validating with ``os.path.isdir`` and before storing on
    +        ``self.cwd``; otherwise the isdir check rejects every valid
    +        result and ``_run_bash`` later prints a misleading "cwd is
    +        missing" warning on every command.
             """
             try:
                 with open(self._cwd_file, encoding="utf-8") as f:
                     cwd_path = f.read().strip()
    +            if _IS_WINDOWS:
    +                cwd_path = _msys_to_windows_path(cwd_path)
                 if cwd_path and os.path.isdir(cwd_path):
                     self.cwd = cwd_path
             except (OSError, FileNotFoundError):
    @@ -583,6 +624,30 @@ def _update_cwd(self, result: dict):
             # Still strip the marker from output so it's not visible
             self._extract_cwd_from_output(result)
     
    +    def _extract_cwd_from_output(self, result: dict):
    +        """Same semantics as the base class, but on Windows the value
    +        emitted by ``pwd -P`` inside Git Bash is in MSYS form
    +        (``/c/Users/x``). Normalize to native Windows form and validate
    +        the directory exists before assigning to ``self.cwd`` — otherwise
    +        ``_run_bash``'s safe-cwd recovery would warn on every subsequent
    +        command.
    +
    +        Always defers to the base class for stripping the marker text from
    +        ``result["output"]`` so output formatting is identical.
    +        """
    +        # Snapshot pre-existing cwd, defer to base for parsing + marker
    +        # stripping, then validate / normalize whatever it assigned.
    +        prev_cwd = self.cwd
    +        super()._extract_cwd_from_output(result)
    +        if self.cwd != prev_cwd:
    +            normalized = _msys_to_windows_path(self.cwd) if _IS_WINDOWS else self.cwd
    +            if normalized and os.path.isdir(normalized):
    +                self.cwd = normalized
    +            else:
    +                # Stale / non-existent path — keep previous cwd; _run_bash
    +                # will resolve a safe fallback on the next call if needed.
    +                self.cwd = prev_cwd
    +
         def cleanup(self):
             """Clean up temp files."""
             for f in (self._snapshot_path, self._cwd_file):

  • tools/tirith_security.py

    @@ -101,6 +101,34 @@ def _load_security_config() -> dict:
     _install_lock = threading.Lock()
     _install_thread: threading.Thread | None = None
     
    +# Warning de-duplication. The spawn/path warnings live in the hot path —
    +# without this dedupe set, a Windows install where ``tirith`` isn't on PATH
    +# (e.g. background install thread still running, or install marked failed)
    +# spams ``tirith spawn failed: [WinError 2]...`` once per terminal command,
    +# easily filling errors.log with hundreds of identical lines.
    +_warned_messages: set[str] = set()
    +_warned_lock = threading.Lock()
    +
    +
    +def _warn_once(key: str, message: str, *args) -> None:
    +    """``logger.warning`` but at-most-once per ``key`` for the process
    +    lifetime. Used to avoid drowning the log when a fail-open tirith
    +    misconfiguration fires on every command."""
    +    with _warned_lock:
    +        if key in _warned_messages:
    +            return
    +        _warned_messages.add(key)
    +    logger.warning(message, *args)
    +
    +
    +def _reset_spawn_warning_state() -> None:
    +    """Clear the warn-once dedupe set. Called when tirith is freshly
    +    (re)installed so a subsequent failure surfaces again — e.g. user
    +    deletes the binary mid-session.
    +    """
    +    with _warned_lock:
    +        _warned_messages.clear()
    +
     # Disk-persistent failure marker — avoids retry across process restarts
     _MARKER_TTL = 86400  # 24 hours
     
    @@ -168,6 +196,10 @@ def _mark_install_failed(reason: str = ""):
     
     def _clear_install_failed():
         """Remove the failure marker after successful install."""
    +    # Reset the warn-once dedupe set so a subsequent failure (e.g. user
    +    # deletes the binary) surfaces in the log again instead of being
    +    # silently suppressed by a stale dedupe key from before the fix.
    +    _reset_spawn_warning_state()
         try:
             os.unlink(_failure_marker_path())
         except OSError:
    @@ -632,7 +664,10 @@ def check_command_security(command: str) -> dict:
         fail_open = cfg["tirith_fail_open"]
     
         if tirith_path is None:
    -        logger.warning("tirith path resolved to None; scanning disabled")
    +        _warn_once(
    +            "tirith_path_none",
    +            "tirith path resolved to None; scanning disabled",
    +        )
             if fail_open:
                 return {"action": "allow", "findings": [], "summary": "tirith path unavailable"}
             return {"action": "block", "findings": [], "summary": "tirith path unavailable (fail-closed)"}
    @@ -646,13 +681,23 @@ def check_command_security(command: str) -> dict:
                 timeout=timeout,
             )
         except OSError as exc:
    -        # Covers FileNotFoundError, PermissionError, exec format error
    -        logger.warning("tirith spawn failed: %s", exc)
    +        # Covers FileNotFoundError, PermissionError, exec format error.
    +        # Dedupe by ``(errno, exc class)`` so a transient failure mode
    +        # surfaces once but doesn't drown the log on every command —
    +        # commonly seen on Windows when the configured path "tirith"
    +        # isn't on PATH yet (background install still running, or
    +        # install marked failed for the day).
    +        spawn_key = f"tirith_spawn_failed:{type(exc).__name__}:{getattr(exc, 'errno', '')}"
    +        _warn_once(spawn_key, "tirith spawn failed: %s", exc)
             if fail_open:
                 return {"action": "allow", "findings": [], "summary": f"tirith unavailable: {exc}"}
             return {"action": "block", "findings": [], "summary": f"tirith spawn failed (fail-closed): {exc}"}
         except subprocess.TimeoutExpired:
    -        logger.warning("tirith timed out after %ds", timeout)
    +        _warn_once(
    +            f"tirith_timeout:{timeout}",
    +            "tirith timed out after %ds",
    +            timeout,
    +        )
             if fail_open:
                 return {"action": "allow", "findings": [], "summary": f"tirith timed out ({timeout}s)"}
             return {"action": "block", "findings": [], "summary": "tirith timed out (fail-closed)"}

• #26718 fix(windows): silence tirith-unavailable CLI banner on platforms with no tirith build

  merged to main

  files: cli.py, tests/tools/test_tirith_security.py, tools/tirith_security.py, website/docs/user-guide/security.md

  Show the diff

  • cli.py

    @@ -11736,11 +11736,13 @@ def run(self):
     
             # Ensure tirith security scanner is available (downloads if needed).
             # Warn the user if tirith is enabled in config but not available,
    -        # so they know command security scanning is degraded.
    +        # so they know command security scanning is degraded.  Suppressed
    +        # on platforms where tirith ships no binary (Windows etc.) — the
    +        # user can't act on it and pattern-matching guards still run.
             try:
    -            from tools.tirith_security import ensure_installed
    +            from tools.tirith_security import ensure_installed, is_platform_supported
                 tirith_path = ensure_installed(log_failures=False)
    -            if tirith_path is None:
    +            if tirith_path is None and is_platform_supported():
                     security_cfg = self.config.get("security", {}) or {}
                     tirith_enabled = security_cfg.get("tirith_enabled", True)
                     if tirith_enabled:

  • tests/tools/test_tirith_security.py

    @@ -333,6 +333,103 @@ def test_startup_prefetch_can_suppress_install_failure_logs(self, mock_cfg):
             _tirith_mod._resolved_path = None
     
     
    +# ---------------------------------------------------------------------------
    +# Unsupported platform (Windows etc.) — silent fast-path everywhere
    +# ---------------------------------------------------------------------------
    +
    +class TestUnsupportedPlatform:
    +    """When _detect_target() returns None (no tirith binary for this OS+arch),
    +    the entire subsystem must stay silent: no PATH probes, no download thread,
    +    no disk failure marker, no spawn attempts, no CLI banner. Pattern-matching
    +    guards still cover the gap; tirith content scanning is just absent."""
    +
    +    def test_is_platform_supported_true_on_linux_x86_64(self):
    +        with patch("tools.tirith_security.platform.system", return_value="Linux"), \
    +             patch("tools.tirith_security.platform.machine", return_value="x86_64"):
    +            assert _tirith_mod.is_platform_supported() is True
    +
    +    def test_is_platform_supported_true_on_darwin_arm64(self):
    +        with patch("tools.tirith_security.platform.system", return_value="Darwin"), \
    +             patch("tools.tirith_security.platform.machine", return_value="arm64"):
    +            assert _tirith_mod.is_platform_supported() is True
    +
    +    def test_is_platform_supported_false_on_windows(self):
    +        with patch("tools.tirith_security.platform.system", return_value="Windows"), \
    +             patch("tools.tirith_security.platform.machine", return_value="AMD64"):
    +            assert _tirith_mod.is_platform_supported() is False
    +
    +    def test_is_platform_supported_false_on_unknown_arch(self):
    +        with patch("tools.tirith_security.platform.system", return_value="Linux"), \
    +             patch("tools.tirith_security.platform.machine", return_value="riscv64"):
    +            assert _tirith_mod.is_platform_supported() is False
    +
    +    @patch("tools.tirith_security._load_security_config")
    +    def test_ensure_installed_unsupported_returns_none_no_thread(self, mock_cfg):
    +        """Windows: don't start a background install thread, don't write a
    +        failure marker — just cache the verdict and return None."""
    +        mock_cfg.return_value = {"tirith_enabled": True, "tirith_path": "tirith",
    +                                 "tirith_timeout": 5, "tirith_fail_open": True}
    +        _tirith_mod._resolved_path = None
    +        with patch("tools.tirith_security.is_platform_supported", return_value=False), \
    +             patch("tools.tirith_security.threading.Thread") as MockThread, \
    +             patch("tools.tirith_security._mark_install_failed") as mock_mark, \
    +             patch("tools.tirith_security.shutil.which") as mock_which:
    +            result = ensure_installed()
    +            assert result is None
    +            MockThread.assert_not_called()
    +            mock_mark.assert_not_called()
    +            mock_which.assert_not_called()
    +            assert _tirith_mod._resolved_path is _tirith_mod._INSTALL_FAILED
    +            assert _tirith_mod._install_failure_reason == "unsupported_platform"
    +
    +    @patch("tools.tirith_security._load_security_config")
    +    def test_check_command_security_unsupported_allows_silently(self, mock_cfg):
    +        """Windows: skip the resolver and spawn entirely — return allow with
    +        an empty summary so callers can't accidentally surface 'tirith
    +        unavailable' messaging to the user."""
    +        mock_cfg.return_value = {"tirith_enabled": True, "tirith_path": "tirith",
    +                                 "tirith_timeout": 5, "tirith_fail_open": True}
    +        with patch("tools.tirith_security.is_platform_supported", return_value=False), \
    +             patch("tools.tirith_security.subprocess.run") as mock_run, \
    +             patch("tools.tirith_security._resolve_tirith_path") as mock_resolve:
    +            result = check_command_security("rm -rf /")
    +            assert result == {"action": "allow", "findings": [], "summary": ""}
    +            mock_run.assert_not_called()
    +            mock_resolve.assert_not_called()
    +
    +    @patch("tools.tirith_security._load_security_config")
    +    def test_resolve_path_unsupported_caches_failure_without_probing(self, mock_cfg):
    +        """The per-command resolver must also short-circuit on Windows so
    +        long-running gateways don't churn through `shutil.which` and disk
    +        I/O for every scanned command."""
    +        mock_cfg.return_value = {"tirith_enabled": True, "tirith_path": "tirith",
    +                                 "tirith_timeout": 5, "tirith_fail_open": True}
    +        _tirith_mod._resolved_path = None
    +        with patch("tools.tirith_security.is_platform_supported", return_value=False), \
    +             patch("tools.tirith_security.shutil.which") as mock_which:
    +            result = _tirith_mod._resolve_tirith_path("tirith")
    +            assert result == "tirith"
    +            mock_which.assert_not_called()
    +            assert _tirith_mod._resolved_path is _tirith_mod._INSTALL_FAILED
    +            assert _tirith_mod._install_failure_reason == "unsupported_platform"
    +
    +    @patch("tools.tirith_security._load_security_config")
    +    def test_explicit_path_still_honored_on_unsupported_platform(self, mock_cfg):
    +        """If a user explicitly configured a tirith_path (e.g. they built it
    +        themselves under WSL), the unsupported-platform short-circuit must
    +        NOT override that — explicit config wins."""
    +        mock_cfg.return_value = {"tirith_enabled": True,
    +                                 "tirith_path": "/opt/custom/tirith",
    +                                 "tirith_timeout": 5, "tirith_fail_open": True}
    +        _tirith_mod._resolved_path = None
    +        with patch("tools.tirith_security.is_platform_supported", return_value=False), \
    +             patch("os.path.isfile", return_value=True), \
    +             patch("os.access", return_value=True):
    +            result = _tirith_mod._resolve_tirith_path("/opt/custom/tirith")
    +            assert result == "/opt/custom/tirith"
    +            assert _tirith_mod._resolved_path == "/opt/custom/tirith"
    +
    +
     # ---------------------------------------------------------------------------
     # Failed download caches the miss (Finding #1)
     # ---------------------------------------------------------------------------

  • tools/tirith_security.py

    @@ -214,7 +214,12 @@ def _hermes_bin_dir() -> str:
     
     
     def _detect_target() -> str | None:
    -    """Return the Rust target triple for the current platform, or None."""
    +    """Return the Rust target triple for the current platform, or None.
    +
    +    Windows is intentionally unsupported — tirith does not ship a Windows
    +    build. Callers should treat `None` as "this platform will never have
    +    tirith" and silently fall back to pattern-matching guards.
    +    """
         system = platform.system()
         machine = platform.machine().lower()
     
    @@ -236,6 +241,16 @@ def _detect_target() -> str | None:
         return f"{arch}-{plat}"
     
     
    +def is_platform_supported() -> bool:
    +    """True when tirith ships a prebuilt binary for this OS+arch.
    +
    +    Used by callers (CLI banner, etc.) to distinguish "tirith failed to
    +    install" from "tirith was never going to install here" — the latter
    +    is silent because there is nothing the user can do about it.
    +    """
    +    return _detect_target() is not None
    +
    +
     def _download_file(url: str, dest: str, timeout: int = 10):
         """Download a URL to a local file."""
         req = urllib.request.Request(url)
    @@ -448,6 +463,15 @@ def _resolve_tirith_path(configured_path: str) -> str:
         explicit = _is_explicit_path(configured_path)
         install_failed = _resolved_path is _INSTALL_FAILED
     
    +    # Platform has no tirith build (Windows etc.). Cache the verdict and
    +    # return the unexpanded configured path — the spawn loop will fail-open
    +    # via the dedupe'd OSError handler, but only after the first call; on
    +    # subsequent calls the fast-path above short-circuits before spawning.
    +    if not explicit and not is_platform_supported():
    +        _resolved_path = _INSTALL_FAILED
    +        _install_failure_reason = "unsupported_platform"
    +        return expanded
    +
         # Explicit path: check it and stop. Never auto-download a replacement.
         if explicit:
             if os.path.isfile(expanded) and os.access(expanded, os.X_OK):
    @@ -574,6 +598,14 @@ def ensure_installed(*, log_failures: bool = True):
                 return path
             return None
     
    +    # Platform has no tirith build (e.g. Windows) — don't probe PATH,
    +    # don't start a download thread, don't write a disk failure marker.
    +    # Pattern-matching guards still run; this path stays silent.
    +    if not is_platform_supported():
    +        _resolved_path = _INSTALL_FAILED
    +        _install_failure_reason = "unsupported_platform"
    +        return None
    +
         configured_path = cfg["tirith_path"]
         explicit = _is_explicit_path(configured_path)
         expanded = os.path.expanduser(configured_path)
    @@ -659,6 +691,12 @@ def check_command_security(command: str) -> dict:
         if not cfg["tirith_enabled"]:
             return {"action": "allow", "findings": [], "summary": ""}
     
    +    # Unsupported platform (Windows etc.) — tirith has no binary here and
    +    # never will. Skip the resolver entirely so we don't even try to spawn.
    +    # Pattern-matching guards still run via the rest of approval.py.
    +    if not is_platform_supported():
    +        return {"action": "allow", "findings": [], "summary": ""}
    +
         tirith_path = _resolve_tirith_path(cfg["tirith_path"])
         timeout = cfg["tirith_timeout"]
         fail_open = cfg["tirith_fail_open"]

  • website/docs/user-guide/security.md

    @@ -537,6 +537,8 @@ security:
     
     When `tirith_fail_open` is `true` (default), commands proceed if tirith is not installed or times out. Set to `false` in high-security environments to block commands when tirith is unavailable.
     
    +Tirith ships prebuilt binaries for Linux (x86_64 / aarch64) and macOS (x86_64 / arm64). On platforms with no prebuilt binary (Windows, etc.), tirith is silently skipped — pattern-matching guards still run, and the CLI does not surface an "unavailable" banner. To use tirith on Windows, run Hermes under WSL.
    +
     Tirith's verdict integrates with the approval flow: safe commands pass through, while both suspicious and blocked commands trigger user approval with the full tirith findings (severity, title, description, safer alternatives). Users can approve or deny — the default choice is deny to keep unattended scenarios secure.
     
     ### Context File Injection Protection

• #52997 fix(security): circuit breaker for tirith crashes to prevent agent hangs (#41400)

  merged to main

  files: tests/tools/test_tirith_security.py, tools/tirith_security.py

  Show the diff

  • tests/tools/test_tirith_security.py

    @@ -17,17 +17,20 @@
     @pytest.fixture(autouse=True)
     def _reset_resolved_path():
         """Pre-set cached path to skip auto-install in scan tests.
    -
         Tests that specifically test ensure_installed / resolve behavior
         reset this to None themselves.
         """
         _tirith_mod._resolved_path = "tirith"
         _tirith_mod._install_thread = None
         _tirith_mod._install_failure_reason = ""
    +    _tirith_mod._crash_count = 0
    +    _tirith_mod._circuit_open = False
         yield
         _tirith_mod._resolved_path = None
         _tirith_mod._install_thread = None
         _tirith_mod._install_failure_reason = ""
    +    _tirith_mod._crash_count = 0
    +    _tirith_mod._circuit_open = False
     
     
     # ---------------------------------------------------------------------------
    @@ -1216,12 +1219,17 @@ def test_repeated_spawn_failure_logs_once(self, mock_cfg, mock_run, caplog):
             _tirith_mod._reset_spawn_warning_state()
     
             with caplog.at_level("WARNING", logger="tools.tirith_security"):
    -            for _ in range(15):
    +            for i in range(15):
                     result = check_command_security("echo hi")
                     # Behavior must remain the same on every call —
                     # fail-open allow, with the exception captured in summary.
                     assert result["action"] == "allow"
    -                assert "unavailable" in result["summary"]
    +                if i < _tirith_mod._CRASH_LIMIT:
    +                    # Before circuit breaker opens, summary has the exception
    +                    assert "unavailable" in result["summary"]
    +                else:
    +                    # After circuit breaker opens, summary is generic
    +                    assert "circuit breaker" in result["summary"]
     
             spawn_warnings = [
                 rec for rec in caplog.records
    @@ -1237,7 +1245,11 @@ def test_repeated_spawn_failure_logs_once(self, mock_cfg, mock_run, caplog):
         def test_distinct_exception_types_each_log_once(self, mock_cfg, mock_run, caplog):
             """``FileNotFoundError`` and ``PermissionError`` are distinct
             failure modes and each deserves its own first-occurrence log
    -        line; the dedupe key includes the exception class."""
    +        line; the dedupe key includes the exception class.
    +        
    +        After _CRASH_LIMIT consecutive failures the circuit breaker opens
    +        and subsequent calls short-circuit without spawning, so we only
    +        see the warnings from the first batch."""
             mock_cfg.return_value = {
                 "tirith_enabled": True, "tirith_path": "tirith",
                 "tirith_timeout": 5, "tirith_fail_open": True,
    @@ -1248,6 +1260,9 @@ def test_distinct_exception_types_each_log_once(self, mock_cfg, mock_run, caplog
                 mock_run.side_effect = FileNotFoundError("[WinError 2]")
                 for _ in range(3):
                     check_command_security("a")
    +            # Circuit breaker is now open — switching to PermissionError
    +            # won't generate a new warning because the function returns
    +            # before reaching subprocess.run.
                 mock_run.side_effect = PermissionError("denied")
                 for _ in range(3):
                     check_command_security("b")
    @@ -1256,8 +1271,8 @@ def test_distinct_exception_types_each_log_once(self, mock_cfg, mock_run, caplog
                 rec for rec in caplog.records
                 if "tirith spawn failed" in rec.message
             ]
    -        assert len(spawn_warnings) == 2, (
    -            f"expected 2 distinct first-occurrence warnings, "
    +        assert len(spawn_warnings) == 1, (
    +            f"expected 1 warning before circuit breaker opens, "
                 f"got {len(spawn_warnings)}"
             )
     

  • tools/tirith_security.py

    @@ -97,6 +97,35 @@ def _load_security_config() -> dict:
     _INSTALL_FAILED = False  # sentinel: distinct from "not yet tried"
     _install_failure_reason: str = ""  # reason tag when _resolved_path is _INSTALL_FAILED
     
    +# Circuit breaker: after _CRASH_LIMIT consecutive spawn/execution failures,
    +# disable tirith for the rest of the process to prevent agent hangs (#41400).
    +# Reset on successful execution (see _record_tirith_crash / check_command_security).
    +#
    +# Thread safety: _crash_count and _circuit_open are module-level globals
    +# mutated without a lock. check_command_security can be called from
    +# concurrent agent threads (gateway multi-session). The race is benign —
    +# at worst two threads both increment past _CRASH_LIMIT and both set
    +# _circuit_open = True, opening the breaker one call early. No data
    +# corruption or security bypass is possible. This intentionally matches
    +# the lock-free style of error counters in mcp_tool.py rather than the
    +# locked _warn_once pattern, because the worst case is harmless.
    +_CRASH_LIMIT = 3
    +_crash_count: int = 0
    +_circuit_open: bool = False
    +
    +
    +def _record_tirith_crash() -> None:
    +    """Increment the crash counter and open the circuit breaker if needed."""
    +    global _crash_count, _circuit_open
    +    _crash_count += 1
    +    if _crash_count >= _CRASH_LIMIT:
    +        _circuit_open = True
    +        logger.warning(
    +            "tirith circuit breaker opened after %d consecutive failures; "
    +            "disabling for the rest of the process",
    +            _crash_count,
    +        )
    +
     # Background install thread coordination
     _install_lock = threading.Lock()
     _install_thread: threading.Thread | None = None
    @@ -708,11 +737,21 @@ def check_command_security(command: str) -> dict:
         Returns:
             {"action": "allow"|"warn"|"block", "findings": [...], "summary": str}
         """
    +    global _crash_count, _circuit_open
    +
         cfg = _load_security_config()
     
         if not cfg["tirith_enabled"]:
             return {"action": "allow", "findings": [], "summary": ""}
     
    +    # Circuit breaker: if tirith has crashed _CRASH_LIMIT times in a row,
    +    # stop trying for the rest of the process.  Without this, a corrupted
    +    # or missing binary causes every tool call to hit the same spawn failure
    +    # → fail-open → agent retry loop, hanging the user for 20+ minutes
    +    # (issue #41400).
    +    if _circuit_open:
    +        return {"action": "allow", "findings": [], "summary": "tirith disabled (circuit breaker)"}
    +
         # Unsupported platform (Windows etc.) — tirith has no binary here and
         # never will. Skip the resolver entirely so we don't even try to spawn.
         # Pattern-matching guards still run via the rest of approval.py.
    @@ -750,6 +789,7 @@ def check_command_security(command: str) -> dict:
             # install marked failed for the day).
             spawn_key = f"tirith_spawn_failed:{type(exc).__name__}:{getattr(exc, 'errno', '')}"
             _warn_once(spawn_key, "tirith spawn failed: %s", exc)
    +        _record_tirith_crash()
             if fail_open:
                 return {"action": "allow", "findings": [], "summary": f"tirith unavailable: {exc}"}
             return {"action": "block", "findings": [], "summary": f"tirith spawn failed (fail-closed): {exc}"}
    @@ -759,6 +799,7 @@ def check_command_security(command: str) -> dict:
                 "tirith timed out after %ds",
                 timeout,
             )
    +        _record_tirith_crash()
             if fail_open:
                 return {"action": "allow", "findings": [], "summary": f"tirith timed out ({timeout}s)"}
             return {"action": "block", "findings": [], "summary": "tirith timed out (fail-closed)"}
    @@ -767,13 +808,17 @@ def check_command_security(command: str) -> dict:
         exit_code = result.returncode
         if exit_code == 0:
             action = "allow"
    +        # Successful execution — reset circuit breaker
    +        _crash_count = 0
         elif exit_code == 1:
             action = "block"
         elif exit_code == 2:
             action = "warn"
         else:
    -        # Unknown exit code — respect fail_open
    +        # Unknown exit code (includes signal-killed processes like -11/SIGSEGV)
    +        # — respect fail_open
             logger.warning("tirith returned unexpected exit code %d", exit_code)
    +        _record_tirith_crash()
             if fail_open:
                 return {"action": "allow", "findings": [], "summary": f"tirith exit code {exit_code} (fail-open)"}
             return {"action": "block", "findings": [], "summary": f"tirith exit code {exit_code} (fail-closed)"}