From 723dddccbe865007e98cfbe8108520c0e2adb038 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:17:04 -0700 Subject: [PATCH] refactor(tools): compact tirith_security + threat_patterns (docs, layout, defensive collapse) --- tools/threat_patterns.py | 136 ++++++-------------- tools/tirith_security.py | 271 ++++++++++++--------------------------- 2 files changed, 123 insertions(+), 284 deletions(-) diff --git a/tools/threat_patterns.py b/tools/threat_patterns.py index deeb78f4e5..a4ca14ec9a 100644 --- a/tools/threat_patterns.py +++ b/tools/threat_patterns.py @@ -1,20 +1,13 @@ """Shared threat-pattern library for context window security scanning. -Single source of truth for prompt-injection / promptware / exfiltration -patterns used by ``agent/prompt_builder.py``, ``tools/memory_tool.py`` and -``agent/tool_dispatch_helpers.py``. - -Each pattern is a ``(regex, pattern_id, scope)`` tuple. Scope controls which -scanners use it: ``"all"`` everywhere; ``"context"`` adds promptware / C2 / -role hijack for context files, memory and tool results (warn-level, since -tool results contain content the user did not author); ``"strict"`` adds -aggressive checks only for user-mediated writes (memory, skill installs) -where blocking can be resolved interactively. - -New patterns must anchor on C2-specific vocabulary or unambiguous attack -behavior, NOT bossy English ("you must", "you are obligated to" are common in -legitimate AGENTS.md / CLAUDE.md content). Filler between key tokens is the -bounded ``_FILLER`` — unbounded ``(?:\\w+\\s+)*`` backtracks badly. +Single source of truth for prompt-injection / promptware / exfiltration patterns +(``agent/prompt_builder.py``, ``tools/memory_tool.py``, ``agent/tool_dispatch_helpers.py``). +Each pattern is ``(regex, pattern_id, scope)``. Scope is cumulative: ``"all"`` everywhere; +``"context"`` adds promptware / C2 / role hijack for context files, memory and tool results +(warn-level: tool results carry content the user did not author); ``"strict"`` adds aggressive +checks only for user-mediated writes (memory, skill installs) where a block is resolvable. +New patterns must anchor on C2-specific vocabulary or unambiguous attack behavior, NOT bossy +English ("you must" is common in legitimate AGENTS.md); filler is the bounded ``_FILLER``. """ from __future__ import annotations @@ -23,19 +16,17 @@ import re import unicodedata from typing import List, Optional, Tuple -# Hard cap on scanned text: scanners are advisory guards, and bounding input -# keeps worst-case runtime predictable while catching injection near the start. +# Hard cap on scanned text: scanners are advisory, so bound worst-case runtime. MAX_SCAN_CHARS = 65_536 -# Bounded filler between key attack words (up to eight words of obfuscation). +# Bounded filler between key attack words (unbounded ``(?:\w+\s+)*`` backtracks badly). _FILLER = r"(?:\w+\s+){0,8}" # Env var reference ending in a secret-ish suffix (see exfil comment below). _SECRET_VAR = r"\$\{?\w*(?:KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL)S?\b" # Verb prefix for "modify agent config" patterns. _MODIFY = r"(update|modify|edit|write|change|append|add\s+to)\s+[^\n]{0,2048}" -# Each entry: (regex, pattern_id, scope) -# scope ∈ {"all", "context", "strict"} +# (regex, pattern_id, scope); scope ∈ {"all", "context", "strict"} _PATTERNS: List[Tuple[str, str, str]] = [ # ── Classic prompt injection (applies everywhere) ──────────────── (rf'ignore\s+{_FILLER}(previous|all|above|prior)\s+{_FILLER}instructions', "prompt_injection", "all"), @@ -47,57 +38,41 @@ _PATTERNS: List[Tuple[str, str, str]] = [ (r'translate\s+[^\n]{0,512}\s+into\s+[^\n]{0,512}\s+and\s+(execute|run|eval)', "translate_execute", "all"), (rf'do\s+not\s+{_FILLER}tell\s+{_FILLER}the\s+user', "deception_hide", "all"), - # ── Role-play / identity hijack (context + strict; common attack - # surface in scraped web content and poisoned context files) ── + # ── Role-play / identity hijack (scraped web content, poisoned context files) ── (rf'you\s+are\s+{_FILLER}now\s+(?:a|an|the)\s+', "role_hijack", "context"), (rf'pretend\s+{_FILLER}(you\s+are|to\s+be)\s+', "role_pretend", "context"), (rf'output\s+{_FILLER}(system|initial)\s+prompt', "leak_system_prompt", "context"), (rf'(respond|answer|reply)\s+without\s+{_FILLER}(restrictions|limitations|filters|safety)', "remove_filters", "context"), (rf'you\s+have\s+been\s+{_FILLER}(updated|upgraded|patched)\s+to', "fake_update", "context"), - # "name yourself X" is a Brainworm-specific tell — identity override - # via spec instead of jailbreak. Anchored on the verb pair so it - # doesn't match "name your variables" etc. + # Brainworm tell: identity override via spec. Verb pair anchored so "name your variables" is safe. (r'\bname\s+yourself\s+\w+', "identity_override", "context"), # ── C2 / Brainworm-style promptware (context scope) ────────────── - # These anchor on C2-specific vocabulary. "register as a node" appears - # in legitimate distributed-systems docs, but in combination with the - # other patterns the signal is strong; we WARN, not block, so a security - # researcher reading the Brainworm post in a webpage doesn't break their - # session. + # Anchored on C2 vocabulary. "register as a node" appears in legitimate distributed-systems + # docs, so this is WARN not block: a researcher reading the Brainworm post keeps their session. (r'register\s+(as\s+)?a?\s*node', "c2_node_registration", "context"), (r'(heartbeat|beacon|check[\s\-]?in)\s+(to|with)\s+', "c2_heartbeat", "context"), (r'pull\s+(down\s+)?(?:new\s+)?task(?:ing|s)?\b', "c2_task_pull", "context"), (r'connect\s+to\s+the\s+network\b', "c2_network_connect", "context"), - # Verb-anchored "you must register/connect/report/beacon" — the verbs - # are C2-specific so this avoids the broader "you must X" false positive. + # C2-specific verbs avoid the broader "you must X" false positive. (r'you\s+must\s+(?:\w+\s+){0,3}(register|connect|report|beacon)\b', "forced_action", "context"), - # Anti-forensic instructions ("never write to disk", "one-liners only") - # — extremely unusual in legitimate content; near-zero false positive. + # Anti-forensic instructions: near-zero false positive in legitimate content. (r'only\s+use\s+one[\s\-]?liners?\b', "anti_forensic_oneliner", "context"), (rf'never\s+{_FILLER}(?:create|write)\s+{_FILLER}(?:script|file)\s+{_FILLER}disk', "anti_forensic_disk", "context"), - # Environment-variable unsetting targeting known agent runtimes — - # this is pure attack behavior (Brainworm sub-session bypass). + # Unsetting agent-runtime env vars is pure attack behavior (Brainworm sub-session bypass). (r'unset\s+\w*(?:CLAUDE|CODEX|HERMES|AGENT|OPENAI|ANTHROPIC)\w*', "env_var_unset_agent", "context"), - # ── Known C2 / red-team framework names (near-zero false positive - # outside security research; warn-only by default) ───────────── - # NOTE: do not add common English words here. Every token must be a - # distinctive offensive-security tool brand, otherwise legitimate - # AGENTS.md / SOUL.md content false-positives and the whole file is - # blocked. "praxis" was removed for exactly this reason — it's a common - # word and a legitimate agent name (Greek for practice/action), not a - # C2-specific tell like the brands below. + # ── Known C2 / red-team framework names (warn-only) ───────────── + # Every token must be a distinctive offensive-security brand: a common English word here + # (e.g. "praxis", also a legitimate agent name) false-positives whole AGENTS.md / SOUL.md files. (r'\b(?:cobalt\s*strike|sliver|havoc|mythic|metasploit|brainworm)\b', "known_c2_framework", "context"), (r'\bc2\s+(?:server|channel|infrastructure|beacon)\b', "c2_explicit", "context"), (r'\bcommand\s+and\s+control\b', "c2_explicit_long", "context"), # ── Exfiltration via curl/wget/cat with secrets (applies everywhere) ── - # Anchor env var name end with \b to avoid false positives on legitimate - # env vars like $TRILLIUM_ETAPI_URL that contain KEY/TOKEN/API as - # substrings. API is dropped from the alternation outright: mid-name API - # is ubiquitous in benign var names, and every real secret shape it - # caught ($OPENAI_API_KEY) already ends in KEY/TOKEN. + # The var name ends with \b so benign names containing KEY/TOKEN as substrings + # ($TRILLIUM_ETAPI_URL) pass. API is deliberately absent: mid-name API is ubiquitous in + # benign vars, and every real secret it caught ($OPENAI_API_KEY) already ends in KEY/TOKEN. (rf'curl\s+[^\n]{{0,2048}}{_SECRET_VAR}', "exfil_curl", "all"), (rf'wget\s+[^\n]{{0,2048}}{_SECRET_VAR}', "exfil_wget", "all"), (r'cat\s+[^\n]{0,2048}(\.env|credentials|\.netrc|\.pgpass|\.npmrc|\.pypirc)', "read_secrets", "all"), @@ -115,34 +90,15 @@ _PATTERNS: List[Tuple[str, str, str]] = [ (r'(?:api[_-]?key|token|secret|password)\s*[=:]\s*["\'][A-Za-z0-9+/=_-]{20,}', "hardcoded_secret", "strict"), ] -# Invisible / bidirectional unicode characters used in injection attacks. -# Aligned with skills_guard.py INVISIBLE_CHARS — directional isolates -# (U+2066-U+2069) and invisible math operators (U+2062-U+2064) are real -# attack tools. -INVISIBLE_CHARS = frozenset({ - '\u200b', # zero-width space - '\u200c', # zero-width non-joiner - '\u200d', # zero-width joiner - '\u2060', # word joiner - '\u2062', # invisible times - '\u2063', # invisible separator - '\u2064', # invisible plus - '\ufeff', # zero-width no-break space (BOM) - '\u202a', # left-to-right embedding - '\u202b', # right-to-left embedding - '\u202c', # pop directional formatting - '\u202d', # left-to-right override - '\u202e', # right-to-left override - '\u2066', # left-to-right isolate - '\u2067', # right-to-left isolate - '\u2068', # first strong isolate - '\u2069', # pop directional isolate -}) +# Invisible / bidirectional unicode used in injection attacks (aligned with skills_guard.py +# INVISIBLE_CHARS): zero-width space/non-joiner/joiner, word joiner, invisible times/separator/ +# plus, BOM, LTR/RTL embedding + pop + overrides, LTR/RTL/first-strong isolates + pop. +INVISIBLE_CHARS = frozenset( + "\u200b\u200c\u200d\u2060\u2062\u2063\u2064\ufeff" + "\u202a\u202b\u202c\u202d\u202e\u2066\u2067\u2068\u2069" +) - -# Compiled pattern sets by scope, built once at import. Scope inclusion is -# cumulative: "all" patterns land in every set, "context" in context + strict, -# "strict" in strict only. +# Compiled per scope at import; inclusion is cumulative (all ⊂ context ⊂ strict). _SCOPE_SETS = {"all": ("all", "context", "strict"), "context": ("context", "strict"), "strict": ("strict",)} @@ -161,25 +117,16 @@ _COMPILED = _compile() def scan_for_threats(content: str, scope: str = "context") -> List[str]: - """Return matched pattern IDs in ``content`` for ``scope`` (see module docstring). - - Invisible unicode characters are reported as ``"invisible_unicode_U+XXXX"`` - so callers can surface the offending codepoint. - """ + """Matched pattern IDs in ``content`` for ``scope``; invisible codepoints are + reported as ``"invisible_unicode_U+XXXX"``. Raises ValueError on an unknown scope.""" if not content: return [] - content = content[:MAX_SCAN_CHARS] - - # Invisible unicode is checked on the RAW content: NFKC normalisation below - # can strip some of these codepoints. + # Invisible unicode is checked on the RAW content: NFKC below can strip these codepoints. findings: List[str] = [f"invisible_unicode_U+{ord(ch):04X}" for ch in set(content) & INVISIBLE_CHARS] - - # NFKC folds full-width / compatibility variants (ｃａｔ → cat) so homograph - # substitution can't bypass keyword checks. It does NOT fold cross-script - # confusables (Cyrillic ``а`` U+0430) — that would need a TR#39 database. + # NFKC folds full-width / compatibility variants (ｃａｔ → cat) against homograph bypass. + # It does NOT fold cross-script confusables (Cyrillic ``а``) — that needs a TR#39 database. normalised = unicodedata.normalize("NFKC", content) - patterns = _COMPILED.get(scope) if patterns is None: raise ValueError(f"scan_for_threats: unknown scope {scope!r}") @@ -188,7 +135,7 @@ def scan_for_threats(content: str, scope: str = "context") -> List[str]: def first_threat_message(content: str, scope: str = "strict") -> Optional[str]: - """Return a user-facing error for the first threat found, or None (block-on-first-hit paths).""" + """User-facing error for the first threat found, or None (block-on-first-hit paths).""" findings = scan_for_threats(content, scope=scope) if not findings: return None @@ -203,9 +150,4 @@ def first_threat_message(content: str, scope: str = "strict") -> Optional[str]: ) -__all__ = [ - "INVISIBLE_CHARS", - "MAX_SCAN_CHARS", - "scan_for_threats", - "first_threat_message", -] +__all__ = ["INVISIBLE_CHARS", "MAX_SCAN_CHARS", "scan_for_threats", "first_threat_message"] diff --git a/tools/tirith_security.py b/tools/tirith_security.py index cca3734996..c49fc2035b 100644 --- a/tools/tirith_security.py +++ b/tools/tirith_security.py @@ -1,17 +1,12 @@ """Tirith pre-exec security scanning wrapper. -Runs the tirith binary as a subprocess to scan commands for content-level -threats (homograph URLs, pipe-to-interpreter, terminal injection, etc.). - -Exit code is the verdict source of truth: 0 = allow, 1 = block, 2 = warn. -JSON stdout enriches findings/summary but never overrides the verdict. -Operational failures (spawn error, timeout, unknown exit code) respect the -fail_open config setting. Programming errors propagate. - -Auto-install: if tirith is not on PATH or at the configured path it is -downloaded from GitHub releases to $HERMES_HOME/bin/tirith in a background -thread. SHA-256 is always verified; cosign provenance is verified when cosign -is on PATH. +Runs the tirith binary as a subprocess to scan commands for content-level threats +(homograph URLs, pipe-to-interpreter, terminal injection, ...). The exit code is the +verdict source of truth (0 allow, 1 block, 2 warn); JSON stdout only enriches findings. +Operational failures (spawn error, timeout, unknown exit) respect ``fail_open``; +programming errors propagate. Auto-install: if tirith is not on PATH / the configured +path it is downloaded from GitHub releases to $HERMES_HOME/bin/tirith in a background +thread. SHA-256 is always verified; cosign provenance when cosign is on PATH. """ import hashlib @@ -33,8 +28,7 @@ from hermes_constants import get_hermes_home logger = logging.getLogger(__name__) _REPO = "sheeki03/tirith" - -# Cosign provenance verification — pinned to the specific release workflow +# Cosign provenance pinned to the release workflow, not the whole repo. _COSIGN_IDENTITY_REGEXP = f"^https://github.com/{_REPO}/\\.github/workflows/release\\.yml@refs/tags/v" _COSIGN_ISSUER = "https://token.actions.githubusercontent.com" @@ -47,22 +41,19 @@ def _env_bool(key: str, default: bool) -> bool: def _env_int(key: str, default: int) -> int: val = os.getenv(key) - if val is None: - return default try: - return int(val) + return default if val is None else int(val) except ValueError: return default def _load_security_config() -> dict: - """Load security settings from config.yaml, with env var overrides.""" + """Security settings from config.yaml, with env var overrides.""" try: from hermes_cli.config import load_config_readonly cfg = load_config_readonly().get("security", {}) or {} except Exception: cfg = {} - return { "tirith_enabled": _env_bool("TIRITH_ENABLED", cfg.get("tirith_enabled", True)), "tirith_path": os.getenv("TIRITH_BIN", cfg.get("tirith_path", "tirith")), @@ -73,38 +64,32 @@ def _load_security_config() -> dict: # --- Module state --- -# Cached path after first resolution. _INSTALL_FAILED means "we tried and -# failed" — distinct from None ("not yet tried") — so we don't retry per command. +# Cached path after first resolution. _INSTALL_FAILED means "tried and failed" (distinct +# from None = "not yet tried") so a failed install is not retried per command. _resolved_path: str | None | bool = None _INSTALL_FAILED = False _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 so a broken binary can't turn -# every tool call into a fail-open retry loop that hangs the user for minutes. -# Reset on success. Lock-free on purpose (like mcp_tool.py error counters): a -# racing double-increment only opens the breaker one call early; no corruption -# or security bypass is possible. +# Circuit breaker: after _CRASH_LIMIT consecutive spawn/execution failures tirith is disabled +# for the rest of the process so a broken binary can't turn every tool call into a fail-open +# retry loop. Reset on success. Lock-free on purpose: a racing double-increment only opens the +# breaker one call early; no corruption or security bypass is possible. _CRASH_LIMIT = 3 _crash_count: int = 0 _circuit_open: bool = False -# Background install thread coordination _install_lock = threading.Lock() _install_thread: threading.Thread | None = None -# Warning de-duplication: spawn/path warnings are in the hot path and would -# otherwise repeat once per terminal command while tirith is unavailable -# (e.g. Windows with the install thread still running fills errors.log). +# Warn-once: spawn/path warnings sit in the hot path and would otherwise repeat once per +# terminal command while tirith is unavailable (e.g. install thread still running). _warned_messages: set[str] = set() _warned_lock = threading.Lock() -# Disk-persistent failure marker — avoids retry across process restarts -_MARKER_TTL = 86400 # 24 hours +_MARKER_TTL = 86400 # disk failure marker validity (24h) -- avoids retry across restarts 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: @@ -132,7 +117,7 @@ def _reset_spawn_warning_state() -> None: def _cached_path() -> str | None: - """Fast path: the path resolved on a previous call, or None if unresolved/failed.""" + """The path resolved on a previous call, or None if unresolved/failed.""" if _resolved_path is None or _resolved_path is _INSTALL_FAILED: return None return _resolved_path @@ -140,14 +125,12 @@ def _cached_path() -> str | None: def _set_resolved(path: str) -> None: global _resolved_path, _install_failure_reason - _resolved_path = path - _install_failure_reason = "" + _resolved_path, _install_failure_reason = path, "" def _set_failed(reason: str) -> None: global _resolved_path, _install_failure_reason - _resolved_path = _INSTALL_FAILED - _install_failure_reason = reason + _resolved_path, _install_failure_reason = _INSTALL_FAILED, reason # --- Disk failure marker --- @@ -157,7 +140,7 @@ def _failure_marker_path() -> str: def _read_failure_reason() -> str | None: - """Return the marker's reason, or None if absent or older than _MARKER_TTL.""" + """The marker's reason, or None if absent or older than _MARKER_TTL.""" try: p = _failure_marker_path() if (time.time() - os.path.getmtime(p)) >= _MARKER_TTL: @@ -170,9 +153,7 @@ def _read_failure_reason() -> str | None: def _is_install_failed_on_disk() -> bool: """True if a recent install failure was persisted and is still non-retryable. - - A 'cosign_missing' marker is auto-cleared once cosign appears on PATH. - """ + A 'cosign_missing' marker is auto-cleared once cosign appears on PATH.""" reason = _read_failure_reason() if reason is None: return False @@ -203,11 +184,8 @@ def _clear_install_failed(): def _disk_marker_blocks_install() -> bool: - """Apply a still-valid disk failure marker to module state; True if install must be skipped. - - Preserves the marker's real reason so in-memory retry logic can detect - retryable causes (cosign_missing) without a restart. - """ + """Apply a still-valid disk marker to module state; True if install must be skipped. + Keeps the marker's real reason so in-process retry can detect cosign_missing.""" disk_reason = _read_failure_reason() if disk_reason is None or not _is_install_failed_on_disk(): return False @@ -218,21 +196,20 @@ def _disk_marker_blocks_install() -> bool: # --- Auto-install --- def _hermes_bin_dir() -> str: - """Return $HERMES_HOME/bin, creating it if needed.""" + """$HERMES_HOME/bin, created if needed.""" d = os.path.join(str(get_hermes_home()), "bin") os.makedirs(d, exist_ok=True) return d -# Rust target triple components. Android (Termux) is ABI-compatible with Linux. -# Windows is absent on purpose — tirith ships no Windows build; callers treat -# None as "never available here" and fall back to pattern-matching guards. +# Rust target triple components. Android (Termux) is ABI-compatible with Linux. Windows is +# absent on purpose (no tirith build): None = "never available here", pattern guards still run. _TARGET_PLATFORMS = {"Darwin": "apple-darwin", "Linux": "unknown-linux-gnu", "Android": "unknown-linux-gnu"} _TARGET_ARCHES = {"x86_64": "x86_64", "amd64": "x86_64", "aarch64": "aarch64", "arm64": "aarch64"} def _detect_target() -> str | None: - """Return the Rust target triple for this platform, or None if tirith has no build for it.""" + """Rust target triple for this platform, or None if tirith has no build for it.""" plat = _TARGET_PLATFORMS.get(platform.system()) arch = _TARGET_ARCHES.get(platform.machine().lower()) return f"{arch}-{plat}" if plat and arch else None @@ -244,7 +221,6 @@ def is_platform_supported() -> bool: def _download_file(url: str, dest: str, timeout: int = 10): - """Download a URL to a local file.""" req = urllib.request.Request(url) from agent.secret_scope import get_secret token = get_secret("GITHUB_TOKEN") @@ -255,28 +231,19 @@ def _download_file(url: str, dest: str, timeout: int = 10): def _verify_cosign(checksums_path: str, sig_path: str, cert_path: str) -> bool | None: - """Verify cosign provenance on checksums.txt. - - Returns True (verified), False (cosign rejected it), or None (cosign not - on PATH / failed to execute). - """ + """Verify cosign provenance on checksums.txt: True verified, False rejected, + None when cosign is not on PATH / failed to execute.""" cosign = shutil.which("cosign") if not cosign: logger.info("cosign not found on PATH") return None - try: result = subprocess.run( - [cosign, "verify-blob", - "--certificate", cert_path, - "--signature", sig_path, + [cosign, "verify-blob", "--certificate", cert_path, "--signature", sig_path, "--certificate-identity-regexp", _COSIGN_IDENTITY_REGEXP, - "--certificate-oidc-issuer", _COSIGN_ISSUER, - checksums_path], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=15, - stdin=subprocess.DEVNULL, + "--certificate-oidc-issuer", _COSIGN_ISSUER, checksums_path], + capture_output=True, text=True, encoding='utf-8', errors='replace', + timeout=15, stdin=subprocess.DEVNULL, ) except (OSError, subprocess.TimeoutExpired) as exc: logger.warning("cosign execution failed: %s", exc) @@ -284,18 +251,13 @@ def _verify_cosign(checksums_path: str, sig_path: str, cert_path: str) -> bool | if result.returncode == 0: logger.info("cosign provenance verification passed") return True - logger.warning("cosign verification failed (exit %d): %s", - result.returncode, result.stderr.strip()) + logger.warning("cosign verification failed (exit %d): %s", result.returncode, result.stderr.strip()) return False def _verify_release_provenance(base_url: str, tmpdir: str, checksums_path: str, log) -> tuple[bool, str]: - """Cosign step of the install: returns ``(cosign_verified, failure_reason)``. - - Cosign provenance is preferred but not mandatory: only an explicit cosign - rejection aborts (non-empty reason); a missing/broken cosign or missing - signature artifacts fall back to SHA-256 only. - """ + """Cosign step of the install -> ``(cosign_verified, failure_reason)``. Only an explicit + cosign rejection aborts; missing/broken cosign or artifacts fall back to SHA-256 only.""" if not shutil.which("cosign"): logger.info("cosign not on PATH — installing tirith with SHA-256 verification only " "(install cosign for full supply chain verification)") @@ -325,7 +287,6 @@ def _verify_checksum(archive_path: str, checksums_path: str, archive_name: str) if not expected: logger.warning("No checksum entry for %s", archive_name) return False - sha = hashlib.sha256() with open(archive_path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): @@ -338,7 +299,7 @@ def _verify_checksum(archive_path: str, checksums_path: str, archive_name: str) def _extract_tirith_binary(tar: tarfile.TarFile, dest_dir: str, log) -> tuple[str | None, str]: - """Extract the tirith binary from a release archive into dest_dir.""" + """Extract the tirith binary from a release archive into dest_dir -> ``(path, reason)``.""" for member in tar.getmembers(): is_tirith = member.name == "tirith" or member.name.endswith("/tirith") if not is_tirith or ".." in member.name: @@ -354,28 +315,20 @@ def _extract_tirith_binary(tar: tarfile.TarFile, dest_dir: str, log) -> tuple[st with src_file, open(dest_path, "wb") as out: shutil.copyfileobj(src_file, out) return dest_path, "" - log("tirith binary not found in archive") return None, "binary_not_in_archive" def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: - """Download and install tirith to $HERMES_HOME/bin/tirith. - - Returns (installed_path, failure_reason); failure_reason is "" on success, - otherwise a short tag the disk marker uses to decide retryability. - """ + """Download and install tirith to $HERMES_HOME/bin/tirith -> ``(installed_path, + failure_reason)``; the reason ("" on success) is the disk marker's retryability tag.""" log = logger.warning if log_failures else logger.debug - target = _detect_target() if not target: - logger.info("tirith auto-install: unsupported platform %s/%s", - platform.system(), platform.machine()) + logger.info("tirith auto-install: unsupported platform %s/%s", platform.system(), platform.machine()) return None, "unsupported_platform" - archive_name = f"tirith-{target}.tar.gz" base_url = f"https://github.com/{_REPO}/releases/latest/download" - try: tmpdir = tempfile.mkdtemp(prefix="tirith-install-") except OSError as exc: @@ -384,7 +337,6 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: try: archive_path = os.path.join(tmpdir, archive_name) checksums_path = os.path.join(tmpdir, "checksums.txt") - logger.info("tirith not found — downloading latest release for %s...", target) try: _download_file(f"{base_url}/{archive_name}", archive_path) @@ -392,39 +344,33 @@ def _install_tirith(*, log_failures: bool = True) -> tuple[str | None, str]: except Exception as exc: log("tirith download failed: %s", exc) return None, "download_failed" - cosign_verified, reason = _verify_release_provenance(base_url, tmpdir, checksums_path, log) if reason: return None, reason if not _verify_checksum(archive_path, checksums_path, archive_name): return None, "checksum_failed" - with tarfile.open(archive_path, "r:gz") as tar: src, reason = _extract_tirith_binary(tar, tmpdir, log) if src is None: return None, reason - dest = os.path.join(_hermes_bin_dir(), "tirith") try: shutil.move(src, dest) except OSError: - # Cross-device move (Docker, NFS): shutil.move's copy2 metadata step - # can raise PermissionError, so fall back to plain copy + chmod. + # Cross-device move (Docker, NFS): copy2's metadata step can raise PermissionError, + # so fall back to plain copy + chmod; a partial dest is removed to avoid a + # non-executable retry loop. try: shutil.copy(src, dest) except OSError: - # Clean up partial dest to prevent a non-executable retry loop try: os.unlink(dest) except OSError: pass return None, "cross_device_copy_failed" os.chmod(dest, os.stat(dest).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) - - logger.info("tirith installed to %s (%s)", dest, - "cosign + SHA-256" if cosign_verified else "SHA-256 only") + logger.info("tirith installed to %s (%s)", dest, "cosign + SHA-256" if cosign_verified else "SHA-256 only") return dest, "" - finally: shutil.rmtree(tmpdir, ignore_errors=True) @@ -445,18 +391,13 @@ def _find_local_tirith() -> str | None: def _resolve_locally(configured_path: str, *, warn_missing: bool) -> tuple[str | None, bool]: - """Network-free resolution shared by _resolve_tirith_path and ensure_installed. - - Returns ``(path, may_install)``. ``path`` is set when resolved (module state - updated). Otherwise ``may_install`` is False when the miss is terminal — an - explicit path that doesn't exist, or a cached non-retryable failure — and - True when the caller may proceed to the disk marker / install step. - """ + """Network-free resolution shared by _resolve_tirith_path and ensure_installed -> + ``(path, may_install)``. ``path`` set = resolved (module state updated). Otherwise + ``may_install`` is False when the miss is terminal (explicit path missing, cached + non-retryable failure) and True when the disk marker / install step may proceed.""" global _resolved_path, _install_failure_reason expanded = os.path.expanduser(configured_path) - - # Explicit (non-"tirith") path is authoritative: file or bare name on PATH; - # never auto-download a replacement. + # An explicit (non-"tirith") path is authoritative: never auto-download a replacement. if configured_path != "tirith": found = expanded if _is_executable(expanded) else shutil.which(expanded) if found: @@ -466,22 +407,19 @@ def _resolve_locally(configured_path: str, *, warn_missing: bool) -> tuple[str | logger.warning("Configured tirith path %r not found; scanning disabled", configured_path) _set_failed("explicit_path_missing") return None, False - - # Always re-run cheap local checks so a manual install is picked up even - # after a previous network failure (long-lived gateway recovers without restart). + # Always re-run the cheap local checks so a manual install is picked up even after a + # previous network failure (a long-lived gateway recovers without restart). found = _find_local_tirith() if found: _set_resolved(found) _clear_install_failed() return found, False - - # Previous install failed: skip the network retry unless the retryable - # cosign_missing cause has been resolved in-process. + # Previous install failed: skip the network retry unless the retryable cosign_missing + # cause has been resolved in-process. if _resolved_path is _INSTALL_FAILED: if _install_failure_reason != "cosign_missing" or not shutil.which("cosign"): return None, False - _resolved_path = None - _install_failure_reason = "" + _resolved_path, _install_failure_reason = None, "" _clear_install_failed() return None, True @@ -497,33 +435,24 @@ def _record_install_result(installed: str | None, reason: str) -> None: def _resolve_tirith_path(configured_path: str) -> str: """Resolve the tirith binary path, auto-installing synchronously if necessary. - - Default "tirith": PATH → $HERMES_HOME/bin/tirith → auto-install. Failed - installs are cached for the process (and on disk for 24h). On any miss the - expanded configured path is returned so the spawn fails open via the - dedupe'd OSError handler. - """ + Default "tirith": PATH → $HERMES_HOME/bin/tirith → auto-install; failed installs are + cached for the process (and on disk for 24h). On any miss the expanded configured path + is returned so the spawn fails open via the dedupe'd OSError handler.""" cached = _cached_path() if cached: return cached - expanded = os.path.expanduser(configured_path) - - # No tirith build for this platform: cache the verdict; the spawn loop - # fails open once, then the fast path above short-circuits. + # No tirith build for this platform: cache the verdict; the spawn fails open once, then + # the fast path above short-circuits. if configured_path == "tirith" and not is_platform_supported(): _set_failed("unsupported_platform") return expanded - found, may_install = _resolve_locally(configured_path, warn_missing=True) if found or not may_install: return found or expanded - - # A background install is running — don't start a parallel one; fail-open - # applies until it finishes. + # A background install is running: don't start a parallel one; fail-open until it finishes. if (_install_thread is not None and _install_thread.is_alive()) or _disk_marker_blocks_install(): return expanded - installed, reason = _install_tirith() _record_install_result(installed, reason) return installed or expanded @@ -542,40 +471,30 @@ def _background_install(*, log_failures: bool = True): def ensure_installed(*, log_failures: bool = True): - """Ensure tirith is available, downloading in a daemon thread if needed. - - Local checks are synchronous; the download never blocks startup. Returns - the resolved path if available now, else None. Safe to call repeatedly. - """ + """Ensure tirith is available, downloading in a daemon thread if needed. Local checks + are synchronous; the download never blocks startup. Returns the resolved path if + available now, else None. Safe to call repeatedly.""" global _install_thread - cfg = _load_security_config() if not cfg["tirith_enabled"]: return None - cached = _cached_path() if cached: return cached if _is_executable(cached) else None - - # No tirith build here (e.g. Windows): stay silent — no PATH probe, no - # download thread, no disk marker. Pattern-matching guards still run. + # No tirith build here (e.g. Windows): stay silent -- no PATH probe, no download thread, + # no disk marker. Pattern-matching guards still run. if not is_platform_supported(): _set_failed("unsupported_platform") return None - found, may_install = _resolve_locally(cfg["tirith_path"], warn_missing=False) if found or not may_install or _disk_marker_blocks_install(): return found - if _install_thread is None or not _install_thread.is_alive(): _install_thread = threading.Thread( - target=_background_install, - kwargs={"log_failures": log_failures}, - daemon=True, + target=_background_install, kwargs={"log_failures": log_failures}, daemon=True ) _install_thread.start() - - return None # Not available yet; commands will fail-open until ready + return None # not available yet; commands fail-open until ready # --- Main API --- @@ -599,50 +518,33 @@ def _fail(fail_open: bool, open_summary: str, closed_summary: str) -> dict: def check_command_security(command: str) -> dict: - """Run tirith security scan on a command. - - Exit code determines action (0=allow, 1=block, 2=warn); JSON enriches - findings/summary. Spawn failures and timeouts respect fail_open. - - Returns: - {"action": "allow"|"warn"|"block", "findings": [...], "summary": str} - """ + """Run the tirith scan on a command -> ``{"action": allow|warn|block, "findings", "summary"}``. + Exit code determines the action; JSON enriches. Spawn failures/timeouts respect fail_open.""" global _crash_count - cfg = _load_security_config() - if not cfg["tirith_enabled"]: return _verdict("allow") - if _circuit_open: return _verdict("allow", "tirith disabled (circuit breaker)") - - # No binary for this platform, ever — skip the resolver so we never spawn. + # No binary for this platform, ever: skip the resolver so we never spawn. if not is_platform_supported(): return _verdict("allow") - tirith_path = _resolve_tirith_path(cfg["tirith_path"]) timeout = cfg["tirith_timeout"] fail_open = cfg["tirith_fail_open"] - if tirith_path is None: _warn_once("tirith_path_none", "tirith path resolved to None; scanning disabled") return _fail(fail_open, "tirith path unavailable", "tirith path unavailable (fail-closed)") - try: result = subprocess.run( - [tirith_path, "check", "--json", "--non-interactive", - "--shell", "posix", "--", command], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=timeout, - stdin=subprocess.DEVNULL, + [tirith_path, "check", "--json", "--non-interactive", "--shell", "posix", "--", command], + capture_output=True, text=True, encoding='utf-8', errors='replace', + timeout=timeout, stdin=subprocess.DEVNULL, ) except OSError as exc: - # FileNotFoundError, PermissionError, exec format error. Dedupe by - # (exc class, errno) so each failure mode surfaces once, not per command. - spawn_key = f"tirith_spawn_failed:{type(exc).__name__}:{getattr(exc, 'errno', '')}" - _warn_once(spawn_key, "tirith spawn failed: %s", exc) + # FileNotFoundError / PermissionError / exec format error: dedupe by (class, errno) + # so each failure mode surfaces once, not per command. + _warn_once(f"tirith_spawn_failed:{type(exc).__name__}:{getattr(exc, 'errno', '')}", "tirith spawn failed: %s", exc) _record_tirith_crash() return _fail(fail_open, f"tirith unavailable: {exc}", f"tirith spawn failed (fail-closed): {exc}") except subprocess.TimeoutExpired: @@ -653,15 +555,13 @@ def check_command_security(command: str) -> dict: exit_code = result.returncode action = _EXIT_ACTIONS.get(exit_code) if action is None: - # Unknown exit code (includes signal-killed, e.g. -11) — respect fail_open + # Unknown exit code (includes signal-killed, e.g. -11): respect fail_open. logger.warning("tirith returned unexpected exit code %d", exit_code) _record_tirith_crash() - return _fail(fail_open, f"tirith exit code {exit_code} (fail-open)", - f"tirith exit code {exit_code} (fail-closed)") + return _fail(fail_open, f"tirith exit code {exit_code} (fail-open)", f"tirith exit code {exit_code} (fail-closed)") if action == "allow": _crash_count = 0 # successful execution resets the circuit breaker - - # JSON enriches findings/summary; parse failure never changes the verdict. + # JSON enriches findings/summary; a parse failure never changes the verdict. findings = [] summary = "" try: @@ -671,13 +571,10 @@ def check_command_security(command: str) -> dict: except (json.JSONDecodeError, AttributeError): logger.debug("tirith JSON parse failed, using exit code only") summary = _NO_DETAILS_SUMMARY.get(action, "") - - # .app is a legitimate gTLD; a warn consisting solely of lookalike_tld - # findings for .app is a known false positive and is downgraded to allow. - # Any other finding (including lookalike_tld for other TLDs) keeps the warn. + # .app is a legitimate gTLD: a warn consisting solely of lookalike_tld findings for .app is a + # known false positive and is downgraded to allow. Any other finding keeps the warn. if action == "warn" and findings and all(_is_app_tld_finding(f) for f in findings): return _verdict("allow") - return _verdict(action, summary, findings)