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

refactor(agent/secrets): hand-compact docstrings and rationale comments (WHY kept)

Browse Source
This commit is contained in:
Teknium
2026-09-02 19:23:12 -07:00
parent 2ac0a6a73d
commit 9d97a255f1
6 changed files with 136 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
agent/redact.py
View File

@@ -1,8 +1,7 @@
"""Regex-based secret redaction for logs and tool output.
Masks API keys, tokens, and credentials before they reach log files, verbose
output, or gateway logs. Short tokens (< 18 chars) are fully masked; longer
tokens keep the first 6 and last 4 characters for debuggability.
Short tokens (< 18 chars) are fully masked; longer ones keep the first 6 and
last 4 characters for debuggability.
"""
import logging
@@ -177,34 +176,28 @@ _STRONG_KEY_KEYWORD_RE = re.compile(
def _is_word_start(s: str, i: int) -> bool:
"""True if position ``i`` in ``s`` begins a word (not mid-word)."""
"""True if position ``i`` in ``s`` begins a word (edge, non-letter before, camelCase/acronym boundary)."""
if i == 0:
return True
prev, cur = s[i - 1], s[i]
if not prev.isalpha() or (cur.isupper() and prev.islower()): # camelCase: clientSecret
if not prev.isalpha() or (cur.isupper() and prev.islower()): # clientSecret
return True
# Acronym run ending (APIToken): 'T' starts a word when followed by lowercase.
return cur.isupper() and prev.isupper() and i + 1 < len(s) and s[i + 1].islower()
return cur.isupper() and prev.isupper() and i + 1 < len(s) and s[i + 1].islower() # APIToken
def _is_word_end(s: str, j: int, *, allow_plural: bool = True) -> bool:
"""True if position ``j`` (exclusive end) in ``s`` ends a word."""
"""True if position ``j`` (exclusive end) in ``s`` ends a word; one trailing ``s`` is absorbed."""
if j >= len(s):
return True
cur = s[j]
if not cur.isalpha() or (cur.isupper() and s[j - 1].islower()): # camelCase: secretKey
if not cur.isalpha() or (cur.isupper() and s[j - 1].islower()): # secretKey
return True
if allow_plural and cur in "sS":
return _is_word_end(s, j + 1, allow_plural=False)
return False
return allow_plural and cur in "sS" and _is_word_end(s, j + 1, allow_plural=False)
def _has_word_bounded_keyword(key: str, keyword_re: "re.Pattern[str]") -> bool:
"""True if ``keyword_re`` matches ``key`` at a word boundary (see _KEY_KEYWORD_RE)."""
return any(
_is_word_start(key, m.start()) and _is_word_end(key, m.end())
for m in keyword_re.finditer(key)
)
return any(_is_word_start(key, m.start()) and _is_word_end(key, m.end()) for m in keyword_re.finditer(key))
def _key_has_secret_keyword(key: str) -> bool:
@@ -213,12 +206,8 @@ def _key_has_secret_keyword(key: str) -> bool:
def _looks_like_opaque_credential(value: str) -> bool:
"""Credential-like shape test for ambiguous ``token``/``key`` values.
Vendor prefixes and JWTs have dedicated redactors; this catches the remaining
opaque family without treating short technical scalars (``CPU``, ``local``)
as secrets merely because their key contains ``token`` or ``key``.
"""
"""Credential-like shape test for ambiguous ``token``/``key`` values, so short
technical scalars (``CPU``, ``local``) are not masked merely for their key name."""
if value == "***" or value.startswith("«redacted:"):
return True
if len(value) >= 16 and re.fullmatch(r"[A-Fa-f0-9]+", value):
@@ -231,12 +220,9 @@ def _looks_like_opaque_credential(value: str) -> bool:
def _should_redact_assignment(key: str, value: str, *, check_keyword: bool) -> bool:
"""Shared gate for the ENV / JSON / YAML assignment passes.
Skips programmatic env lookups used as values (code snippets, not secrets),
optionally requires a word-bounded secret keyword in the key, then redacts
when the key is unambiguously credential-bearing or the value looks opaque.
"""
"""Shared gate for the ENV / JSON / YAML assignment passes: skip programmatic env
lookups used as values, optionally require a word-bounded keyword in the key,
then redact when the key is unambiguously credential-bearing or the value looks opaque."""
if _ENV_LOOKUP_VALUE_RE.match(value):
return False
if check_keyword and not _key_has_secret_keyword(key):
@@ -260,10 +246,9 @@ _AUTH_HEADER_RE = re.compile(r"((?:Proxy-)?Authorization:\s*)([A-Za-z][\w.+-]*\s
_SECRET_HEADER_NAMES = r"(?:x-api-key|x-goog-api-key|api-key|apikey|x-api-token|x-auth-token|x-access-token)"
_SECRET_HEADER_RE = re.compile(rf"({_SECRET_HEADER_NAMES}\s*:\s*)(\S+)", re.IGNORECASE)
# Telegram bot tokens: bot<digits>:<token> or <digits>:<token>, token >= 30 chars.
# Telegram bot tokens: [bot]<digits>:<token>, token >= 30 chars.
_TELEGRAM_RE = re.compile(r"(bot)?(\d{8,}):([-A-Za-z0-9_]{30,})")
# Private key blocks: -----BEGIN RSA PRIVATE KEY----- ... -----END RSA PRIVATE KEY-----
_PRIVATE_KEY_RE = re.compile(r"-----BEGIN[A-Z ]*PRIVATE KEY-----[\s\S]*?-----END[A-Z ]*PRIVATE KEY-----")
# Database connection strings: protocol://user:PASSWORD@host. The userinfo and
@@ -287,19 +272,15 @@ _URL_BARE_TOKEN_RE = re.compile(
re.IGNORECASE,
)
# JWT tokens: header.payload[.signature] — always start with "eyJ" (base64 "{").
# Matches 1-part (header only), 2-part, and full 3-part JWTs.
# JWTs always start with "eyJ" (base64 "{"); 1-, 2- and 3-part forms.
_JWT_RE = re.compile(r"eyJ[A-Za-z0-9_-]{10,}(?:\.[A-Za-z0-9_=-]{4,}){0,2}")
# E.164 phone numbers: +<country><number>, 7-15 digits. The negative lookahead
# prevents matching hex strings or identifiers.
# E.164 phone numbers, 7-15 digits; the lookahead rejects hex strings / identifiers.
_SIGNAL_PHONE_RE = re.compile(r"(\+[1-9]\d{6,14})(?![A-Za-z0-9])")
# URLs containing query strings — `scheme://authority path ?query [#fragment]` (CDP-URL path).
# CDP-URL path: web URLs with a query string / with ``user:password@`` userinfo
# (DB protocols are covered by _DB_CONNSTR_RE).
_URL_WITH_QUERY_RE = re.compile(r"(https?|wss?|ftp)://([^\s/?#]+)([^\s?#]*)\?([^\s#]+)(#\S*)?")
# URLs containing userinfo — `scheme://user:password@host` for ANY web scheme
# (DB protocols are covered by _DB_CONNSTR_RE). CDP-URL path.
_URL_USERINFO_RE = re.compile(r"(https?|wss?|ftp)://([^/\s:@]+):([^/\s@]+)@")
# Strict provider-egress URL redaction: delimiters stay in capture groups so the
@@ -313,8 +294,7 @@ _STRICT_URL_PARAM_RE = re.compile(r"([?#&;])([A-Za-z0-9_.~+%\-]+)=([^#&;\s\"'<>]
# (~55s per sub() on a 320KB compaction payload).
_STRICT_URL_USERINFO_RE = re.compile(r"(//)([^/\s?#@]+)@")
# Form-urlencoded body detection: conservative — only applies when the entire
# text looks like a query string (k=v&k=v pattern with no newlines).
# Form-urlencoded body: only when the ENTIRE text is a k=v&k=v string.
_FORM_BODY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*=[^&\s]*(?:&[A-Za-z_][A-Za-z0-9_.-]*=[^&\s]*)+$")
# Control / zero-width characters that can split a token body (``sk-abc\x1bdef``,
@@ -373,16 +353,9 @@ _DISPLAY_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f\x80-\x9f\u200b-\u200f\u202a-\u
def mask_secret(value: str, *, head: int = 4, tail: int = 4, floor: int = 12,
placeholder: str = "***", empty: str = "") -> str:
"""Mask a secret for display (``hermes config`` / ``status`` / ``dump``).
Values shorter than ``floor`` (after control-byte stripping) return
``placeholder``; falsy input returns ``empty``.
>>> mask_secret("sk-proj-abcdef1234567890")
'sk-p...7890'
>>> mask_secret("short")
'***'
"""
"""Mask a secret for display (``hermes config`` / ``status`` / ``dump``):
``sk-p...7890``; shorter than ``floor`` (after control-byte stripping) →
``placeholder``; falsy → ``empty``."""
value = _DISPLAY_CONTROL_RE.sub("", value) if value else value
if not value:
return empty
@@ -419,11 +392,8 @@ def _canonical_url_param_name(name: str) -> str:
def _redact_strict_url_credentials(text: str) -> str:
"""Redact credentials from absolute, relative, and network URL references.
Stricter than display/log redaction; used only at explicit secret-egress
boundaries. Preserves keys, separators, public params, hosts, and paths.
"""
"""Strict egress-boundary redaction of URL credentials (absolute, relative and
network references); preserves keys, separators, public params, hosts, paths."""
def _redact_param(match: re.Match) -> str:
if _canonical_url_param_name(match.group(2)) not in _SENSITIVE_QUERY_PARAMS:
return match.group(0)
@@ -442,10 +412,9 @@ def _redact_strict_url_credentials(text: str) -> str:
def redact_cdp_url(value: object) -> str:
"""Mask secrets in a CDP/browser endpoint URL before it is logged.
``redact_sensitive_text`` deliberately passes web-URL query params and
``user:pass@`` through (OAuth callbacks, magic links the agent must follow).
CDP discovery tokens are pure credentials, so this opts INTO both URL
redactors. Single source of truth for CDP URLs passed to a log/error.
Unlike ``redact_sensitive_text`` (which passes web-URL query params and
``user:pass@`` through for OAuth callbacks / magic links), CDP discovery
tokens are pure credentials, so this opts INTO both URL redactors.
"""
text = redact_sensitive_text("" if value is None else str(value))
if not text:
@@ -465,12 +434,9 @@ def _redact_form_body(text: str) -> str:
def _mask_token_nonreusable(token: str) -> str:
"""Redact a prefix-matched credential to a NON-REUSABLE sentinel.
No head/tail chars (an agent once wrote a truncated-looking mask back into a
config file, corrupting the credential); only the vendor prefix label is kept
so the credential KIND stays visible.
"""
"""Redact a prefix-matched credential to a NON-REUSABLE sentinel: no head/tail
chars (an agent once wrote a truncated-looking mask back into a config file),
only the vendor prefix label so the credential KIND stays visible."""
label = next((sub for sub in _PREFIX_SUBSTRINGS if token.startswith(sub)), "") if token else ""
return f"«redacted:{label}…»" if label else "«redacted-secret»"
@@ -486,20 +452,16 @@ def _assignment_sub(render, *, check_keyword: bool):
def _redact_assignments(text: str) -> str:
"""ENV / config / JSON / YAML assignment passes (skipped for code files).
Passes that would match ``token=``/``key=`` URL params skip ``://`` text:
web-URL query params are intentionally passed through (see redact_sensitive_text).
"""
"""ENV / config / JSON / YAML assignment passes (skipped for code files). Passes
that would match ``token=``/``key=`` URL params skip ``://`` text (web-URL query
params are intentionally passed through, see redact_sensitive_text)."""
if "=" in text:
_redact_env = _assignment_sub(lambda g: f"{g[0]}={g[1]}{_mask_token(g[2])}{g[1]}", check_keyword=True)
text = _ENV_ASSIGN_RE.sub(_redact_env, text)
# Lowercase env names; unlike the all-caps regex this one would match URL params.
if "://" not in text:
if "://" not in text: # lowercase names would match URL params
text = _ENV_ASSIGN_LOWER_RE.sub(_redact_env, text)
# Lowercase/dotted config keys. The keyword pre-gate is exact (every
# _CFG_*_RE match needs a secret keyword) and matters: _CFG_DOTTED_RE
# backtracks quadratically on long unbroken [A-Za-z0-9_.\-] runs.
# The keyword pre-gate is exact and matters: _CFG_DOTTED_RE backtracks
# quadratically on long unbroken [A-Za-z0-9_.\-] runs.
if "://" not in text and _CFG_SECRET_WORD_RE.search(text):
text = _CFG_DOTTED_RE.sub(_redact_env, text)
text = _CFG_ANCHORED_RE.sub(_redact_env, text)
@@ -508,8 +470,7 @@ def _redact_assignments(text: str) -> str:
text = _JSON_FIELD_RE.sub(
_assignment_sub(lambda g: f'{g[0]}: "{_mask_token(g[1])}"', check_keyword=False), text)
# Unquoted YAML / colon config, after JSON so quoted values are handled
# there (_YAML_ASSIGN_RE's lookahead skips quotes).
# YAML after JSON: quoted values are handled there (_YAML_ASSIGN_RE skips quotes).
if ":" in text and "://" not in text:
text = _YAML_ASSIGN_RE.sub(
_assignment_sub(lambda g: f"{g[0]}{g[1]}{_mask_token(g[2])}", check_keyword=True), text)
@@ -519,18 +480,13 @@ def _redact_assignments(text: str) -> str:
def _redact_url_credentials(text: str, code_file: bool) -> str:
"""DB connection-string passwords and bare-token URL userinfo (``://`` text only)."""
def _redact_db(m):
# With code_file=True a pure ``{...}`` password group is an f-string
# template reference (f"postgresql://{user}:{pass}@{host}"), not a
# literal credential — preserve it. The regex forbids whitespace in the
# password group, so a single-line template's group(2) is exactly the
# brace expression.
# code_file: a pure ``{...}`` password is an f-string template reference
# (f"postgresql://{user}:{pass}@{host}"), not a literal credential.
pw = m.group(2)
if code_file and pw.startswith("{") and pw.endswith("}"):
return m.group(0)
return f"{m.group(1)}***{m.group(3)}"
text = _DB_CONNSTR_RE.sub(_redact_db, text)
# ``scheme://TOKEN@host`` — only the colon-less bare-token form; ``user:pass@``
# and query-string tokens pass through (see the web-URL note below).
return _URL_BARE_TOKEN_RE.sub(lambda m: f"{m.group(1)}{_mask_token(m.group(2))}{m.group(3)}", text)
@@ -568,8 +524,7 @@ def redact_sensitive_text(text: str, *, force: bool = False, code_file: bool = F
return text
code_file = code_file or file_read
# Known prefixes (sk-, ghp_, etc.). Control/zero-width chars can split a
# token body so _PREFIX_RE alone misses it — mask those runs first.
# Control/zero-width chars can split a token body so _PREFIX_RE alone misses it.
if _has_known_prefix_substring(text):
_prefix_sub = _mask_token_nonreusable if file_read else _mask_token
text = _mask_control_split_tokens(text, _prefix_sub)
@@ -578,14 +533,9 @@ def redact_sensitive_text(text: str, *, force: bool = False, code_file: bool = F
if not code_file:
text = _redact_assignments(text)
# Case-insensitive regex, so "uthorization" is the cheapest substring gate
# covering every casing without a casefold().
if "uthorization" in text or "UTHORIZATION" in text:
text = _AUTH_HEADER_RE.sub(
lambda m: m.group(1) + (m.group(2) or "") + _mask_token(m.group(3)), text,
)
if "uthorization" in text or "UTHORIZATION" in text: # cheapest gate over every casing
text = _AUTH_HEADER_RE.sub(lambda m: m.group(1) + (m.group(2) or "") + _mask_token(m.group(3)), text)
# API-key style headers and Telegram bot tokens both require ":".
if ":" in text:
text = _SECRET_HEADER_RE.sub(lambda m: m.group(1) + _mask_token(m.group(2)), text)
text = _TELEGRAM_RE.sub(lambda m: f"{m.group(1) or ''}{m.group(2)}:***", text)
@@ -599,16 +549,12 @@ def redact_sensitive_text(text: str, *, force: bool = False, code_file: bool = F
if "eyJ" in text:
text = _JWT_RE.sub(lambda m: _mask_token(m.group(0)), text)
# Web-URL query/userinfo redaction is opt-in (see docstring); known
# credential shapes inside URLs are still caught by the passes above.
if redact_url_credentials:
if redact_url_credentials: # opt-in; known credential shapes in URLs are caught above
text = _redact_strict_url_credentials(text)
# Form-urlencoded bodies (only triggers on clean k=v&k=v inputs).
if "&" in text and "=" in text:
text = _redact_form_body(text)
# E.164 phone numbers (Signal, WhatsApp)
if "+" in text:
text = _SIGNAL_PHONE_RE.sub(_redact_phone, text)
@@ -635,23 +581,17 @@ def _command_segments(command: str) -> list[str]:
def _command_reads_env_file(command: str | None) -> bool:
"""True if ``command`` reads a ``.env``-style file (by basename) to stdout.
Template files (``.env.example``) are not in the basename list. Defense-in-
depth, not a boundary: indirect reads (``sudo cat .env``, ``$(cat .env)``,
``sed``/``awk``) are not detected, matching ``is_env_dump_command``.
"""
Defense-in-depth, not a boundary: indirect reads (``sudo cat .env``, ``$(cat
.env)``, ``sed``/``awk``) are not detected, matching ``is_env_dump_command``."""
if not command:
return False
for seg in _command_segments(command):
# Plain split() rather than shlex: shlex treats backslashes as escapes
# and mangles Windows paths (``C:\Users\...\.env``).
tokens = seg.split()
tokens = seg.split() # not shlex: it mangles Windows paths (``C:\Users\...\.env``)
if not tokens or tokens[0] not in _FILE_READ_COMMANDS:
continue
for arg in tokens[1:]:
if arg.startswith("-"):
continue
# Strip quotes split() leaves attached, then any / or \ path prefix.
basename = arg.strip("\"'").rsplit("/", 1)[-1].rsplit("\\", 1)[-1]
if basename.lower() in _ENV_FILE_BASENAMES:
return True
@@ -659,11 +599,8 @@ def _command_reads_env_file(command: str | None) -> bool:
def is_env_dump_command(command: str | None) -> bool:
"""True if ``command`` dumps environment variables to stdout.
First token of any pipeline/sequence segment in _ENV_DUMP_COMMANDS.
Conservative: unrecognized → False (callers fall back to code_file=True).
"""
"""True if any pipeline/sequence segment starts with an _ENV_DUMP_COMMANDS
token. Conservative: unrecognized → False (callers fall back to code_file=True)."""
if not command or not isinstance(command, str):
return False
for seg in _command_segments(command):
@@ -677,22 +614,17 @@ def is_env_dump_command(command: str | None) -> bool:
def redact_terminal_output(output: str, command: str | None = None, *, force: bool = False) -> str:
"""Redact terminal/process stdout — the single policy for ALL terminal-output surfaces.
``code_file`` is False (ENV-assignment pass runs) only when ``command`` is an
env dump or reads a ``.env`` file; otherwise True to avoid false positives
on source/config dumps. ``force=True`` bypasses the global opt-out.
"""
"""Single redaction policy for ALL terminal-output surfaces: the ENV-assignment
pass runs only when ``command`` is an env dump or reads a ``.env`` file
(otherwise code_file=True avoids false positives on source/config dumps)."""
if not output:
return output
code_file = not (is_env_dump_command(command) or _command_reads_env_file(command))
return redact_sensitive_text(output, force=force, code_file=code_file)
# --- Prefix pre-screen ------------------------------------------------------
# Derived from _PREFIX_PATTERNS at load time so a new prefix can't silently
# break the gate. No false negatives: every pattern has its literal prefix as a
# substring of any match.
# --- Prefix pre-screen: derived from _PREFIX_PATTERNS so a new prefix can't
# silently break the gate (every match contains its pattern's literal prefix).
def _extract_literal_prefix(pattern: str) -> str:
"""Leading literal chars of a regex (up to the first metacharacter)."""
@@ -731,13 +663,10 @@ def _unbounded_quantifier_follows(pattern: str, j: int) -> bool:
def _pattern_structure(pattern: str) -> tuple[bool, bool]:
"""One scan → ``(has_top_level_alternation, has_nested_unbounded_repeat)``.
Top-level ``|`` (outside any group/class) defeats the literal-prefix
guarantee: for ``ab|.*`` the prefix ``ab`` binds only the first branch;
grouped alternation (``ab(?:x|y)``) stays allowed. An unbounded quantifier
applied to a group containing one — ``(a+)+`` / ``(?:x*)*`` / ``(a{2,})+`` —
is the canonical ReDoS shape, and registered patterns run on every log line.
Structural nesting only; overlapping branches (``(a|aa)+``) are the plugin
author's responsibility.
Top-level ``|`` defeats the literal-prefix guarantee (in ``ab|.*`` the prefix
binds only the first branch; ``ab(?:x|y)`` is fine). An unbounded quantifier
on a group containing one (``(a+)+``, ``(a{2,})+``) is the canonical ReDoS
shape. Structural only; overlapping branches (``(a|aa)+``) are not detected.
"""
top_level_alt = nested = False
contains_unbounded = [False] # per open group: does it contain an unbounded repeat?
@@ -796,11 +725,8 @@ def _plugin_patterns() -> list:
def _rebuild_prefix_matcher() -> None:
"""Recompile the prefix alternation and pre-screen substrings.
Callers look these globals up at call time, so swapping the module
attributes (atomic under the GIL) propagates immediately.
"""
"""Recompile the prefix alternation and pre-screen substrings; callers read the
module globals at call time, so the swap propagates immediately."""
global _PREFIX_RE, _PREFIX_SUBSTRINGS
combined = _PREFIX_PATTERNS + _plugin_patterns()
_PREFIX_RE = _compile_prefix_matcher(combined)
@@ -824,13 +750,11 @@ _PATTERN_REJECT_RULES = (
def register_redaction_patterns(patterns, source: str = "plugin") -> int:
"""Additively register credential-token regexes with the redaction engine.
"""Additively register credential-token regexes; returns the count accepted.
Accepted patterns join the vendor-prefix alternation everywhere built-ins
apply. Invalid entries (non-compiling, top-level alternation, nested
unbounded quantifiers, < 2 literal prefix chars) are warned and skipped,
never raised — a broken plugin must not break startup. Duplicates are
skipped. Returns the number of patterns accepted.
Invalid entries (non-compiling, top-level alternation, nested unbounded
quantifiers, < 2 literal prefix chars) and duplicates are warned/skipped,
never raised — a broken plugin must not break startup.
"""
accepted = []
for pattern in patterns or []:

40
agent/secret_scope.py
View File

@@ -109,16 +109,12 @@ def _environ_or(name: str, default: Optional[str]) -> Optional[str]:
def get_secret(name: str, default: Optional[str] = None) -> Optional[str]:
"""Resolve a credential by env-var name, honoring the active profile scope.
1. Global vars (``_is_global_env``) always read ``os.environ``.
2. Scope installed: read from it. Under multiplexing the scope is
authoritative (a miss returns ``default``, never ``os.environ``, which
may hold another profile's value). With multiplexing OFF a miss falls
through to ``os.environ``: single-profile deployments legitimately
inject credentials via the process env (systemd, ``op run``, shell
exports) and the scope — installed around e.g. every cron job — must
stay a ``.env`` overlay, not a blindfold (otherwise cron 401s).
3. No scope: multiplex INACTIVE reads ``os.environ`` (legacy behavior);
ACTIVE raises ``UnscopedSecretError`` (fail closed).
Global vars always read ``os.environ``. With a scope installed, a miss returns
``default`` under multiplexing (never another profile's ``os.environ`` value)
but falls through to ``os.environ`` otherwise — single-profile deployments
inject credentials via the process env (systemd, ``op run``), so the scope
must stay a ``.env`` overlay, not a blindfold (otherwise cron 401s). With no
scope: multiplex INACTIVE reads ``os.environ``; ACTIVE raises (fail closed).
"""
if _is_global_env(name):
return _environ_or(name, default)
@@ -143,15 +139,10 @@ def get_secret(name: str, default: Optional[str] = None) -> Optional[str]:
def _strip_inline_comment(value: str) -> str:
"""Strip a dotenv-style inline comment from a raw ``.env`` value.
Mirrors python-dotenv semantics: for quoted values scan to the matching
close quote (backslash-escape-aware for double quotes, since
``save_env_value`` writes ``\\"``/``\\\\``) and drop a trailing ``# ...``;
other trailing junk (or an unterminated quote) leaves the value untouched.
Unquoted values truncate only at a ``#`` PRECEDED BY WHITESPACE, so
``foo#bar`` and a leading ``#`` survive while ``value # comment`` → ``value``.
"""
"""Strip a dotenv-style inline comment (python-dotenv semantics): quoted values
scan to the matching close quote (backslash-aware for double quotes) and drop a
trailing ``# ...``, else stay untouched; unquoted values truncate only at a
``#`` PRECEDED BY WHITESPACE (``foo#bar`` survives, ``value # c`` → ``value``)."""
value = value.strip()
if not value:
return value
@@ -171,14 +162,9 @@ def _strip_inline_comment(value: str) -> str:
def load_env_file(env_path: Path) -> Dict[str, str]:
"""Parse a ``.env`` file into a dict WITHOUT touching ``os.environ``.
Handles the subset Hermes writes (``export`` prefix, full-line and inline
``#`` comments, quotes with the writer's escapes reversed via the canonical
``_parse_env_value`` — stripping only outer quotes would corrupt credentials
containing ``"`` or ``\\``). ``utf-8-sig`` so a Windows BOM doesn't prefix
the first key as ``\\ufeffNAME``.
"""
"""Parse a ``.env`` file into a dict WITHOUT touching ``os.environ``: ``export``
prefix, ``#`` comments, and the writer's quote escapes reversed via the canonical
``_parse_env_value``. ``utf-8-sig`` so a BOM doesn't prefix the first key."""
secrets: Dict[str, str] = {}
try:
text = env_path.read_text(encoding="utf-8-sig")

64
agent/secret_sources/base.py
View File

@@ -51,12 +51,9 @@ def get_source_environment() -> MutableMapping[str, str]:
def source_child_env() -> Dict[str, str]:
"""Environment for a helper child that legitimately needs the caller's env.
Single-profile startup keeps the legacy contract (full process env, minus
the terminal blocklist). A profile-local fetch (multiplex) gets ONLY the
per-fetch view so a child can never inherit sibling profiles' secrets.
"""
"""Environment for a helper child that legitimately needs the caller's env:
full process env (minus the terminal blocklist) in single-profile startup;
ONLY the per-fetch view under multiplex, so no sibling profile's secrets leak."""
source_env = get_source_environment()
if source_env is os.environ:
from tools.environments.local import build_subprocess_env
@@ -66,12 +63,8 @@ def source_child_env() -> Dict[str, str]:
class ErrorKind(str, Enum):
"""Machine-readable failure taxonomy for :class:`FetchResult.error`.
A fixed vocabulary keeps startup warnings and ``hermes secrets status``
uniform, and lets the orchestrator apply kind-dependent policy (e.g.
stale-cache fallback on NETWORK/TIMEOUT but never AUTH_FAILED) once.
"""
"""Failure taxonomy for :class:`FetchResult.error`; lets the orchestrator apply
kind-dependent policy once (stale-cache fallback on NETWORK/TIMEOUT, never AUTH_FAILED)."""
NOT_CONFIGURED = "not_configured" # enabled but missing token/project/map
BINARY_MISSING = "binary_missing" # helper CLI not found / not installed
@@ -108,12 +101,9 @@ def coerce_float(value: Any, default: float) -> float:
@dataclass
class FetchResult:
"""Outcome of one source's fetch.
``secrets`` is what the source *would* contribute; whether each var is
applied is the orchestrator's decision. ``applied``/``skipped`` exist for
the legacy fetch-and-apply entry points and stay empty in ``fetch()``.
"""
"""Outcome of one source's fetch. ``secrets`` is what the source *would*
contribute; ``applied``/``skipped`` serve the legacy fetch-and-apply entry
points and stay empty in ``fetch()``."""
secrets: Dict[str, str] = field(default_factory=dict)
applied: List[str] = field(default_factory=list)
@@ -169,20 +159,15 @@ class SecretSource(ABC):
@abstractmethod
def fetch(self, cfg: dict, home_path: Path) -> FetchResult:
"""Resolve this source's secrets. MUST NOT raise or prompt.
``cfg`` is the raw ``secrets.<name>`` section — may be malformed.
"""
"""Resolve this source's secrets. MUST NOT raise or prompt; ``cfg`` is the
raw ``secrets.<name>`` section and may be malformed."""
def is_enabled(self, cfg: dict) -> bool:
return bool(isinstance(cfg, dict) and cfg.get("enabled"))
def override_existing(self, cfg: dict) -> bool:
"""May this source overwrite vars .env / the shell already set?
Never extends to vars claimed by another source in the same pass —
cross-source overrides are a config error the orchestrator warns about.
"""
"""May this source overwrite vars .env / the shell already set? Never extends
to vars claimed by another source (a config error the orchestrator warns about)."""
return bool(isinstance(cfg, dict)
and cfg.get("override_existing", self.override_existing_default))
@@ -207,11 +192,7 @@ class SecretSource(ABC):
return {}
def remediation(self, kind: Optional["ErrorKind"], cfg: dict) -> str:
"""One-line actionable next step for a failed fetch (pure, no I/O).
Shown right after the fetch error by the startup status printer and
``hermes secrets ... status``. Empty string suppresses the hint.
"""
"""One-line actionable next step for a failed fetch (pure); "" suppresses the hint."""
if kind is None:
return ""
template = self.remediation_hints.get(kind) or _GENERIC_REMEDIATION.get(kind, "")
@@ -240,11 +221,8 @@ def scrub_ansi(text: str) -> str:
def run_cli(argv: Sequence[str], *, env: Dict[str, str], timeout: float, label: str,
timeout_message: str, stdin: Any = subprocess.DEVNULL) -> subprocess.CompletedProcess:
"""``subprocess.run`` an argv list (never a shell), capturing utf-8 text.
Timeout and spawn failure become ``RuntimeError`` with messages safe to
surface; callers own returncode interpretation.
"""
"""``subprocess.run`` an argv list (never a shell), capturing utf-8 text; timeout
and spawn failure become ``RuntimeError``. Callers own returncode interpretation."""
try:
return subprocess.run( # noqa: S603 — argv list, no shell
list(argv), env=env, capture_output=True, text=True, encoding="utf-8", errors="replace",
@@ -258,14 +236,10 @@ def run_cli(argv: Sequence[str], *, env: Dict[str, str], timeout: float, label:
def run_secret_cli(argv: Sequence[str], *, allow_env: Sequence[str] = (), extra_env: Optional[Dict[str, str]] = None,
timeout: float = DEFAULT_CLI_TIMEOUT_SECONDS) -> subprocess.CompletedProcess:
"""Run a secret-manager helper CLI with a minimal, allowlisted env.
The child gets PATH/HOME/locale basics plus only ``allow_env`` (auth/session
vars) and ``extra_env`` — never the full post-dotenv ``os.environ``, which
holds every credential Hermes knows. ``NO_COLOR=1`` plus ANSI-scrubbed
stderr keep helper diagnostics out of Hermes output; stdin is /dev/null so
a prompting helper fails fast. Pass user refs AFTER a ``--`` terminator.
"""
"""Run a secret-manager helper CLI with a minimal, allowlisted env (never the
full post-dotenv ``os.environ``): PATH/HOME/locale basics plus ``allow_env``
and ``extra_env``. ``NO_COLOR=1`` + ANSI-scrubbed stderr; stdin is /dev/null so
a prompting helper fails fast. Pass user refs AFTER a ``--`` terminator."""
base_keep = ("PATH", "HOME", "USERPROFILE", "SYSTEMROOT", "TMPDIR", "TEMP",
"LANG", "LC_ALL", "XDG_CONFIG_HOME", "XDG_DATA_HOME")
env = {k: os.environ[k] for k in (*base_keep, *allow_env) if k in os.environ}

59
agent/secret_sources/bitwarden.py
View File

@@ -47,9 +47,8 @@ _BWS_CHECKSUM_NAME = f"bws-sha256-checksums-{_BWS_VERSION}.txt"
_BWS_DOWNLOAD_TIMEOUT = 60
_BWS_RUN_TIMEOUT = 30
# Cache layout: <hermes_home>/cache/bws_cache.json holds only secret VALUES
# (never the access token) — plaintext-equivalent to .env but kept out of it so
# users editing .env don't commit BSM-sourced secrets.
# <hermes_home>/cache/bws_cache.json holds only secret VALUES (never the access
# token); kept out of .env so users editing .env don't commit BSM-sourced secrets.
_CacheKey = Tuple[str, str, str] # (access_token_fingerprint, project_id, server_url)
_DISK_CACHE_BASENAME = "bws_cache.json"
_ENCRYPTED_CACHE_BASENAME = "bws_cache.enc.json"
@@ -129,8 +128,7 @@ def _platform_asset_name() -> str:
if system == "Windows":
return f"bws-{arch}-pc-windows-msvc-{_BWS_VERSION}.zip"
if system == "Linux":
# glibc default; musl only if ldd says so (glibc prints to stderr, musl
# to stdout). A wrong guess surfaces as a loader error we catch.
# glibc default; musl only if ldd says so (a wrong guess surfaces as a loader error).
libc = "gnu"
try:
res = subprocess.run(["ldd", "--version"], capture_output=True, text=True, encoding='utf-8',
@@ -145,11 +143,8 @@ def _platform_asset_name() -> str:
def install_bws(*, force: bool = False) -> Path:
"""Download, verify, and install the pinned ``bws`` binary; raises on any failure.
The auto-install path catches; ``hermes secrets bitwarden setup`` lets the
error propagate so the wizard can show it.
"""
"""Download, verify, and install the pinned ``bws`` binary; raises on any failure
(the auto-install path catches; the setup wizard shows the error)."""
bin_dir = _hermes_bin_dir()
bin_dir.mkdir(parents=True, exist_ok=True)
target = bin_dir / _platform_binary_name()
@@ -222,11 +217,8 @@ def _pick_zip_member(zf: zipfile.ZipFile, binary_name: str) -> str:
def _safe_extract_member(zf: zipfile.ZipFile, member: str, dest_dir: Path) -> Path:
"""Extract one member, refusing zip-slip (``../`` or absolute member names).
``ZipFile.extract`` joins the member onto ``dest_dir`` without verifying the
result stays inside it, so containment is checked here first.
"""
"""Extract one member, refusing zip-slip: ``ZipFile.extract`` never verifies the
joined path stays inside ``dest_dir``, so containment is checked here first."""
dest_root = os.path.realpath(dest_dir)
target = os.path.realpath(os.path.join(dest_root, member))
try: # commonpath raises for e.g. different Windows drives — treat as escape
@@ -248,12 +240,8 @@ def _b64e(raw: bytes) -> str:
def _derive_encrypted_cache_key(access_token: str, salt: bytes) -> bytes:
"""HKDF the local cache key from the bootstrap BWS token.
cryptography is imported lazily: most CLI commands import this module while
building argparse, and eagerly mapping ``_rust.pyd`` on Windows blocks the
updater from replacing that file.
"""
"""HKDF the local cache key from the bootstrap BWS token. cryptography is imported
lazily: eagerly mapping ``_rust.pyd`` on Windows blocks the updater replacing it."""
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
@@ -263,11 +251,8 @@ def _derive_encrypted_cache_key(access_token: str, salt: bytes) -> bytes:
def _write_encrypted_disk_cache(*, cache_key: _CacheKey, access_token: str, entry: _CachedFetch,
home_path: Optional[Path] = None) -> None:
"""Persist an AES-GCM encrypted last-good entry atomically (best-effort).
The raw token is never stored; it only derives the key. A successful write
completes migration, so the legacy plaintext cache is removed.
"""
"""Persist an AES-GCM encrypted last-good entry atomically (best-effort). The raw
token only derives the key; a successful write removes the legacy plaintext cache."""
try:
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
@@ -397,12 +382,8 @@ def fetch_bitwarden_secrets(
def _summarize_bws_stderr(raw: str) -> str:
"""Reduce a bws (Rust color-eyre) error dump to its cause line(s).
Keeps the numbered ``0: ...`` cause lines (joined with ``; ``), drops
everything from ``Location:``/``Backtrace omitted`` on, and falls back to
the stripped raw text when the shape is unrecognized.
"""
"""Reduce a bws (color-eyre) error dump to its numbered cause lines joined with
``; `` (dropping ``Location:``/``Backtrace`` on); raw text if unrecognized."""
text = raw.replace("\x1b", "").strip()
if not text:
return text
@@ -460,12 +441,8 @@ def _run_bws_list(bws: Path, access_token: str, project_id: str, server_url: str
class BitwardenSource(SecretSource):
"""Bitwarden Secrets Manager as a registered **bulk** source.
``fetch()`` only fetches — precedence, overrides and the ``os.environ``
writes are the orchestrator's. Bulk: it injects every secret in the BSM
project, so explicit per-var bindings from mapped sources outrank it.
"""
"""Bitwarden Secrets Manager as a registered **bulk** source (injects every
secret in the project, so explicit mapped bindings outrank it)."""
name = "bitwarden"
label = "Bitwarden Secrets Manager"
@@ -543,11 +520,7 @@ class BitwardenSource(SecretSource):
def clear_caches(home_path: Optional[Path] = None) -> None:
"""Drop in-process AND disk caches (plaintext and encrypted).
Used after a token rotation so the next startup fetches fresh instead of
serving a pull cached under the old token's fingerprint.
"""
"""Drop in-process AND disk caches (plaintext and encrypted), e.g. after a token rotation."""
_STORE.clear(home_path)
try:
_encrypted_disk_cache_path(home_path).unlink()

47
agent/secret_sources/onepassword.py
View File

@@ -97,12 +97,8 @@ def _validate_references(references: Optional[Dict[str, str]]) -> Tuple[Dict[str
def _auth_fingerprint(token_env: str) -> str:
"""SHA-256 prefix over everything `op` would authenticate with.
Folds in the service-account token, OP_ACCOUNT, Connect host/token and all
``OP_SESSION_*`` vars, so signing into a different identity changes the
cache key and a value cached under the old identity is never served.
"""
"""SHA-256 prefix over everything `op` would authenticate with (token, account,
Connect host/token, ``OP_SESSION_*``), so a new identity never sees old cached values."""
source_env = get_source_environment()
parts: List[str] = [f"{label}={source_env.get(var, '')}" for label, var in (
("token", token_env), ("account", "OP_ACCOUNT"),
@@ -116,11 +112,8 @@ def _refs_fingerprint(references: Dict[str, str]) -> str:
def find_op(binary_path: str = "") -> Optional[Path]:
"""Resolve a usable ``op`` binary, or None.
A pinned ``binary_path`` is used verbatim (PATH is NOT consulted) and a
pinned-but-missing path returns None rather than silently falling back.
"""
"""Resolve a usable ``op`` binary, or None. A pinned ``binary_path`` is used
verbatim — pinned-but-missing returns None rather than falling back to PATH."""
if binary_path:
pinned = Path(binary_path)
return pinned if pinned.exists() and os.access(pinned, os.X_OK) else None
@@ -146,11 +139,8 @@ def _op_child_env(token_value: str) -> Dict[str, str]:
def _run_op_read(op: Path, reference: str, *, account: str = "", token_value: str = "") -> str:
"""Resolve one ``op://`` reference; raises ``RuntimeError`` on any failure.
An exit-0 empty/whitespace-only value is a failure too — applying it would
silently clobber a good .env/shell credential with ``""``.
"""
"""Resolve one ``op://`` reference; raises ``RuntimeError`` on any failure, including
an exit-0 empty value (applying it would clobber a good credential with ``""``)."""
cmd: List[str] = [str(op), "read"]
if account:
cmd += ["--account", account]
@@ -179,10 +169,9 @@ def fetch_onepassword_secrets(
) -> Tuple[Dict[str, str], List[str]]:
"""Resolve ``references`` (name → ``op://…``) to ``(secrets, warnings)``.
Raises ``RuntimeError`` only when no ``op`` binary is available. Per-ref
failures become warnings and the ref is dropped, so one bad entry never
sinks the rest. Only a complete, error-free pull is cached, so a transient
auth failure isn't frozen in for the whole TTL window.
Raises ``RuntimeError`` only when no ``op`` binary is available; per-ref
failures become warnings. Only a complete, error-free pull is cached, so a
transient auth failure isn't frozen in for the whole TTL window.
"""
valid, warnings = _validate_references(references)
if not valid:
@@ -232,13 +221,9 @@ def apply_onepassword_secrets(
override_existing: bool = True, cache_ttl_seconds: float = 300, home_path: Optional[Path] = None,
) -> FetchResult:
"""Resolve configured ``op://`` references and set them on ``os.environ``
(``hermes secrets onepassword sync --apply``).
Never raises. References already satisfied by the environment (when
``override_existing`` is false) and the token var itself are skipped
*before* fetching, so ``op`` is never invoked for a value that would be
discarded.
"""
(``hermes secrets onepassword sync --apply``). Never raises. Refs already
satisfied by the env (when ``override_existing`` is false) and the token var
are skipped *before* fetching, so ``op`` never runs for a discarded value."""
result = FetchResult()
if not enabled:
return result
@@ -282,12 +267,8 @@ def apply_onepassword_secrets(
class OnePasswordSource(SecretSource):
"""1Password as a registered **mapped** source.
``fetch()`` only fetches — precedence, overrides and the ``os.environ``
writes are the orchestrator's. Mapped: the user explicitly binds each env
var to an ``op://`` ref, so its claims outrank bulk sources on contested vars.
"""
"""1Password as a registered **mapped** source (explicit per-var bindings, so
its claims outrank bulk sources on contested vars)."""
name = "onepassword"
label = "1Password"

29
agent/secret_sources/registry.py
View File

@@ -101,13 +101,9 @@ def _validate_source(source: SecretSource) -> Optional[str]:
def register_source(source: SecretSource, *, replace: bool = False, builtin: bool = False,
scope: Optional[str] = None) -> bool:
"""Register a secret source. Returns True on success.
Rejections are logged, never raised — a bad plugin must not take down
startup. ``replace`` lets tests / user plugins override a same-named
source (last-writer-wins); scheme collisions across *different* names are
always rejected.
"""
"""Register a secret source; True on success. Rejections are logged, never
raised. ``replace`` allows same-name override (last-writer-wins); scheme
collisions across *different* names are always rejected."""
problem = _validate_source(source)
if problem:
logger.warning(problem)
@@ -188,11 +184,8 @@ def list_plugin_sources() -> List[SecretSource]:
def _ensure_builtin_sources() -> None:
"""Idempotently register the bundled sources.
Lazy so importing this module stays cheap, and per-source guarded so a
broken bundled source can never break registration of the others.
"""
"""Idempotently register the bundled sources (lazy so import stays cheap;
per-source guarded so one broken source can't block the others)."""
global _BUILTINS_LOADED
with _REGISTRY_LOCK:
if _BUILTINS_LOADED:
@@ -222,10 +215,9 @@ def _fetch_with_timeout(source: SecretSource, cfg: dict, home_path: Path,
environ: MutableMapping[str, str]) -> FetchResult:
"""Run source.fetch() under a wall-clock budget; never raises.
A daemon worker thread enforces the budget: a source that blows it is
reported as TIMEOUT and its eventual result discarded. The thread may
linger until process exit — acceptable for a startup-only path, and far
better than an unbounded hang on every ``hermes`` invocation.
A worker thread enforces the budget: a source that blows it is reported as
TIMEOUT and its eventual result discarded (the thread may linger until
process exit — acceptable for a startup-only path).
"""
timeout = source.fetch_timeout_seconds(cfg)
executor = concurrent.futures.ThreadPoolExecutor(max_workers=1, thread_name_prefix=f"secret-src-{source.name}")
@@ -263,9 +255,8 @@ def _section(secrets_cfg: dict, name: str) -> dict:
def _ordered_enabled_sources(secrets_cfg: dict, *, scope: Optional[str] = None) -> List[SecretSource]:
"""Which sources run, in which order: ``secrets.sources`` first, then the
rest in registration order; enabled = the source's own ``is_enabled``.
Mapped-vs-bulk precedence is applied on top by :func:`apply_all`."""
"""Enabled sources: ``secrets.sources`` order first, then registration order
(mapped-vs-bulk precedence is applied on top by :func:`apply_all`)."""
sources = {source.name: source for source in list_sources(scope=scope)}
explicit = secrets_cfg.get("sources")