ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(tools): compact tirith_security + threat_patterns (docs, layout, defensive collapse)

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:17:04 -07:00
parent 113f04616b
commit 723dddccbe
2 changed files with 123 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/threat_patterns.py
+39 -97
View File
@@ -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"]
tools/tirith_security.py
+84 -187
View File
@@ -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)