agent/redact.py
Rama main · réplica de lectura del repositorio original.
"""Regex-based secret redaction for logs and tool output.
Short tokens (< 18 chars) are fully masked; longer ones keep the first 6 and
last 4 characters for debuggability.
"""
import json
import logging
import os
import re
import shlex
import threading
import unicodedata
from urllib.parse import unquote_plus
# Shared with agent/file_safety's read-block list so the two defenses can't
# drift: a blocked file_tools read that falls back to ``cat`` is still caught.
from agent.file_safety import _BLOCKED_PROJECT_ENV_BASENAMES as _ENV_FILE_BASENAMES
from agent.redact_connstr import _KV_CONNSTR_GATE_RE, _redact_kv_connection_strings
from agent.redact_headers import (
_COOKIE_HEADER_JSON_RE, _COOKIE_HEADER_JSON_UNTERMINATED_RE, _SECRET_HEADER_JSON_RE,
_SECRET_HEADER_JSON_UNTERMINATED_RE, json_header_value_is_credential, redact_cookie_assignments,
redact_header_credentials,
)
logger = logging.getLogger(__name__)
# Browser and CDP results can be a page-controlled JSON tree. The walkers stop
# at this depth and replace the remainder, so a nested payload cannot raise
# RecursionError on the way to the model.
REDACT_WALK_MAX_DEPTH = 320
REDACT_WALK_DEPTH_MARKER = "[redacted: nesting limit]"
# ---------------------------------------------------------------------------
# Vault-value redaction registry (profile-scoped, bounded)
# ---------------------------------------------------------------------------
# Exact secret values that transited a server-side vault fill (browser_vault). Generic
# credential-shaped regexes cannot catch an arbitrary user password, so the fill path registers
# the exact bytes and every browser_* tool result (including browser_cdp Runtime.evaluate
# passthrough) is scrubbed against them before it can reach the model. Memory only: never
# persisted or logged. Keyed by profile home so a multiplex gateway never scrubs profile B's
# output with profile A's passwords (which would also confirm to B that the bytes exist), and
# bounded per profile: a fill-heavy session evicts its oldest entries rather than growing forever.
_VAULT_REDACTION_MAX_PER_PROFILE = 64
_VAULT_REDACTION_VALUES: dict = {} # profile home → ordered {value: None}
_VAULT_REDACTION_LOCK = threading.Lock()
def _vault_scope() -> str:
from hermes_constants import get_hermes_home
return str(get_hermes_home())
def register_vault_redaction_value(value) -> None:
"""Register an exact vault secret value for model-facing redaction.
Called by the vault fill path BEFORE the injection happens, so no later browser tool result
can echo the value back into model context. Also registers the form a text input normalizes
it to (CR/LF stripped), since that is what the page holds.
"""
if not isinstance(value, str) or not value:
return
normalized = value.replace("\r", "").replace("\n", "")
with _VAULT_REDACTION_LOCK:
bucket = _VAULT_REDACTION_VALUES.setdefault(_vault_scope(), {})
for v in (value, normalized):
if v:
bucket.pop(v, None) # re-registering refreshes recency
bucket[v] = None
while len(bucket) > _VAULT_REDACTION_MAX_PER_PROFILE:
del bucket[next(iter(bucket))]
def clear_vault_redaction_values() -> None:
"""Drop the current profile's registered values (profile teardown / explicit lock)."""
with _VAULT_REDACTION_LOCK:
_VAULT_REDACTION_VALUES.pop(_vault_scope(), None)
def redact_registered_vault_values(text: str) -> str:
"""Exact-substring scrub of every vault secret value registered for the current profile."""
if not isinstance(text, str) or not text:
return text
with _VAULT_REDACTION_LOCK:
bucket = _VAULT_REDACTION_VALUES.get(_vault_scope())
values = sorted(bucket, key=len, reverse=True) if bucket else () # longest first: a substring never shadows its superstring
for value in values:
if value in text:
text = text.replace(value, "«redacted-vault-secret»")
return text
# Sensitive query-string param names (case-insensitive): opaque tokens / OAuth
# codes / pre-signed signatures with no vendor prefix.
# Ported from nearai/ironclaw#2529 — catches tokens whose values don't match any known vendor prefix regex
# (e.g. opaque tokens, short OAuth codes).
_SENSITIVE_QUERY_PARAMS = frozenset({
"access_token", "refresh_token", "id_token", "token", "api_key", "apikey",
"client_secret", "password", "auth", "jwt", "session", "secret", "key",
"code", "signature", "x-amz-signature",
"x-goog-signature", "sig", # GCS V4 signed URLs, Azure SAS tokens
})
# Snapshot at import time so runtime env mutations (e.g. an LLM-generated
# `export HERMES_REDACT_SECRETS=false`) cannot disable redaction mid-session.
# ON by default; `security.redact_secrets: false` bridges to this env var.
# ON by default — secure default per issue #17691. Users who need raw credential values in tool output (e.g.
# working on the redactor itself) can opt out via `security.redact_secrets: false` in config.yaml (bridged
# to this env var in hermes_cli/main.py, gateway/run.py, and cli.py) or `HERMES_REDACT_SECRETS=false` in
# ~/.hermes/.env. An opt-out warning is logged at gateway and CLI startup so operators see the downgrade —
# see `_log_redaction_status()` in gateway/run.py and cli.py.
_REDACT_ENABLED = os.getenv("HERMES_REDACT_SECRETS", "true").lower() in {"1", "true", "yes", "on"}
# Routed multiplex profiles: the import-time snapshot above is the LAUNCH profile's policy. A profile
# served under a HERMES_HOME override follows its own ``security.redact_secrets`` (its ``.env`` value
# first, like the standalone bridge in hermes_cli/main.py). The formatter never reads config or files
# for it (it runs under the logging handler's lock, per record): the values are recorded when that
# profile's .env / config is loaded (``note_env_policy`` / ``note_config_policy``) or when log routing
# is set up for it (``resolve_profile_policy``). Until then the profile redacts (the secure default).
_ENV_POLICY_BY_HOME: dict = {} # home key -> raw HERMES_REDACT_SECRETS from its .env, or None
_CONFIG_POLICY_BY_HOME: dict = {} # home key -> raw security.redact_secrets from its config, or None
def note_env_policy(home_key: str, raw) -> None:
_ENV_POLICY_BY_HOME[home_key] = None if raw is None else str(raw)
def note_config_policy(home_key: str, raw) -> None:
_CONFIG_POLICY_BY_HOME[home_key] = None if raw is None else str(raw)
def resolve_profile_policy(home) -> None:
"""Record ``home``'s redaction policy now (log-routing setup), so its records never wait for it."""
from hermes_constants import hermes_home_key, reset_hermes_home_override, set_hermes_home_override
token = set_hermes_home_override(home)
try:
key = hermes_home_key()
from hermes_cli.config import load_config_readonly, load_env
note_env_policy(key, load_env().get("HERMES_REDACT_SECRETS"))
note_config_policy(key, (load_config_readonly().get("security") or {}).get("redact_secrets"))
except (Exception, SystemExit) as exc: # health: allow BLE001 -- unreadable now: stays unknown, which redacts
logging.getLogger(__name__).debug("redaction policy for %s not read: %s", home, exc)
finally:
reset_hermes_home_override(token)
def _redact_enabled() -> bool:
"""Effective redaction switch for the active profile (launch snapshot when no override). Reads
only in-memory state: the live secret scope and the recorded per-profile policy."""
from hermes_constants import get_hermes_home_override, hermes_home_key
if get_hermes_home_override() is None:
return _REDACT_ENABLED
home_key = hermes_home_key()
from agent.secret_scope import current_secret_scope
scope = current_secret_scope()
raw = scope.get("HERMES_REDACT_SECRETS") if scope else None
if raw is None and scope is None:
# No live scope (the log listener thread formats routed records): the profile's own .env, as
# its scope would have it. Not recorded yet: redact, since .env may opt the profile back in.
if home_key not in _ENV_POLICY_BY_HOME:
return True
raw = _ENV_POLICY_BY_HOME[home_key]
if raw is None:
raw = _CONFIG_POLICY_BY_HOME.get(home_key)
if raw is None:
return True
return str(raw).strip().lower() in {"1", "true", "yes", "on"}
# Known API key prefixes -- match the prefix + contiguous token chars.
# Every pattern MUST start with a literal prefix: _PREFIX_SUBSTRINGS (the cheap
# pre-screen gate) is derived from these literals and must stay false-negative-free.
_PREFIX_PATTERNS = [
# Some provider-issued ``sk-`` keys carry dot-delimited body segments (Alibaba
# ``sk-sp-…``/``sk-ws-…``). Each unit is one body char optionally preceded by
# a single dot, so the body ends on its last non-dot char (sentence punctuation
# is never consumed) and can never span ``..``: the ``sk-pro...EFGH`` display
# mask is left alone by a second redaction pass instead of collapsing to
# ``***``. Kept free of nested unbounded repeats so the pattern passes the
# same structural gate plugins must.
r"sk-[A-Za-z0-9_-](?:\.?[A-Za-z0-9_-]){9,}",
# Context7 API keys: a fixed vendor prefix (``ctx7sk-``) plus an opaque body.
# The generic ``sk-`` family cannot catch it: the matcher's lookbehind
# rejects a prefix preceded by a word char, and here the ``sk-`` sits inside
# ``ctx7sk-`` - so a bare Context7 key passed through unmasked (#142576).
r"ctx7sk-[A-Za-z0-9_-]{10,}",
r"ghp_[A-Za-z0-9]{10,}", # GitHub PAT (classic)
r"github_pat_[A-Za-z0-9_]{10,}", # GitHub PAT (fine-grained)
r"gho_[A-Za-z0-9]{10,}", # GitHub OAuth access token
r"ghu_[A-Za-z0-9]{10,}", # GitHub user-to-server token
r"ghs_[A-Za-z0-9]{10,}", # GitHub server-to-server token
r"ghr_[A-Za-z0-9]{10,}", # GitHub refresh token
r"xapp-\d+-[A-Za-z0-9-]{10,}", # Slack app-Level token
r"xox[baprs]-[A-Za-z0-9-]{10,}", # Slack bot/app/user tokens
r"AIza[A-Za-z0-9_-]{30,}", # Google API keys
r"pplx-[A-Za-z0-9]{10,}", # Perplexity
r"fal_[A-Za-z0-9_-]{10,}", # Fal.ai
r"fc-[A-Za-z0-9]{10,}", # Firecrawl
r"bb_live_[A-Za-z0-9_-]{10,}", # BrowserBase
r"gAAAA[A-Za-z0-9_=-]{20,}", # Codex encrypted tokens
r"AKIA[A-Z0-9]{16}", # AWS Access Key ID
r"sk_live_[A-Za-z0-9]{10,}", # Stripe secret key (live)
r"sk_test_[A-Za-z0-9]{10,}", # Stripe secret key (test)
r"rk_live_[A-Za-z0-9]{10,}", # Stripe restricted key
r"SG\.[A-Za-z0-9_-]{10,}", # SendGrid API key
r"hf_[A-Za-z0-9]{10,}", # HuggingFace token
r"r8_[A-Za-z0-9]{10,}", # Replicate API token
r"npm_[A-Za-z0-9]{10,}", # npm access token
r"pypi-[A-Za-z0-9_-]{10,}", # PyPI API token
r"dop_v1_[A-Za-z0-9]{10,}", # DigitalOcean PAT
r"doo_v1_[A-Za-z0-9]{10,}", # DigitalOcean OAuth
# AgentMail API key: ``am_`` / ``am_org_`` + an opaque alphanumeric body. The body has no ``_``/``-``,
# which is what separates it from ``am_example_identifier_123`` (#10983); public docs pin only the
# prefix, so the charset stays broad and the length floor does the discriminating.
r"am_(?:org_)?[A-Za-z0-9]{20,}",
r"sk_[A-Za-z0-9_]{10,}", # ElevenLabs TTS key (sk_ underscore, not sk- dash)
r"tvly-[A-Za-z0-9]{10,}", # Tavily search API key
r"exa_[A-Za-z0-9]{10,}", # Exa search API key
r"gsk_[A-Za-z0-9]{10,}", # Groq Cloud API key
r"syt_[A-Za-z0-9]{10,}", # Matrix access token
r"retaindb_[A-Za-z0-9]{10,}", # RetainDB API key
r"hsk-[A-Za-z0-9]{10,}", # Hindsight API key
r"mem0_[A-Za-z0-9]{10,}", # Mem0 Platform API key
r"brv_[A-Za-z0-9]{10,}", # ByteRover API key
r"xai-[A-Za-z0-9]{30,}", # xAI (Grok) API key
r"ntn_[A-Za-z0-9]{10,}", # Notion internal integration token
r"fw-[A-Za-z0-9]{30,}", # Fireworks AI API key
r"fw_[A-Za-z0-9]{30,}", # Fireworks AI API key
r"fpk_[A-Za-z0-9]{30,}", # Fireworks AI project key
# GitLab token families (each keeps a full literal prefix for the pre-screen).
# Ported from openclaw/openclaw#112954; follow-up invited in #4541.
r"glpat-[A-Za-z0-9_\-]{10,}", # GitLab personal access token
r"gloas-[A-Za-z0-9_\-]{10,}", # GitLab OAuth application secret
r"gldt-[A-Za-z0-9_\-]{10,}", # GitLab deploy token
r"glrt-[A-Za-z0-9_.\-]{10,}", # GitLab runner authentication token (routable tokens are dotted)
r"glrtr-[A-Za-z0-9_.\-]{10,}", # GitLab runner registration token (routable)
r"glcbt-[A-Za-z0-9_\-]{10,}", # GitLab CI/CD job token
r"glptt-[A-Za-z0-9_\-]{10,}", # GitLab pipeline trigger token
r"glft-[A-Za-z0-9_\-]{10,}", # GitLab feed token
r"glimt-[A-Za-z0-9_\-]{10,}", # GitLab incoming mail token
r"glagent-[A-Za-z0-9_\-]{10,}", # GitLab agent (KAS) token
r"glsoat-[A-Za-z0-9_\-]{10,}", # GitLab service-account access token
r"glffct-[A-Za-z0-9_\-]{10,}", # GitLab feature-flags client token
r"glwt-[A-Za-z0-9_\-]{10,}", # GitLab workspace token
r"GR1348941[A-Za-z0-9_\-]{10,}", # GitLab legacy runner registration token
r"pk-lf-[A-Za-z0-9\-]{8,}", # Langfuse public key (sk-lf- already covered by sk- pattern)
]
# ENV assignment: KEY=value where KEY carries a secret-like name. Uppercase keys
# tolerate spaces around "=" and allow the keyword embedded anywhere
# (``MYTOKEN=…``) — an all-caps key is almost never prose. Bare ``KEY``/``PASS``/
# ``PW`` suffixes are included; _key_has_secret_keyword rejects ``KEYBOARD=``.
# The regex is IGNORECASE so lowercase env names (``openai_key=…``) are caught here too. The secret name
# must sit at a word boundary (``_``-delimited or whole-word) so generic prose words (``password=``,
# ``token=``, ``KEYBOARD=``, ``PASSAGE=``) do not match — those are handled by the config/form/URL paths,
# and a bare ``password=…`` in a form body must not be swallowed greedily by ``\S+``. See #77484.
_SECRET_ENV_NAMES = r"(?:API_?KEY|KEY|TOKEN|SECRET|PASSWORD|PASSWD|PASS|PW|CREDENTIAL|AUTH)"
_ENV_ASSIGN_RE = re.compile(rf"([A-Z0-9_]{{0,50}}{_SECRET_ENV_NAMES}[A-Z0-9_]{{0,50}})\s*=\s*(['\"]?)(\S+)\2")
# Command-line credentials. ``--password=…``, ``--token …`` and ``docker login -p …`` reach
# the approval card and the chat egress verbatim because the ENV passes only look at
# ``NAME=value`` (issue #143982). The flag name is kept so the reader still knows what was
# passed; only the value is masked. ``-p`` is scoped to a container ``login`` because
# everywhere else it is a port or a path.
# Each quote style is matched on its own, so ``"Abc'def ghi"`` is consumed whole instead of
# falling back to ``\S+`` at the apostrophe and leaving the tail visible. A double-quoted
# value honours backslash escapes (``"a\"b c"``) and a single-quoted one the shell's
# ``'\''`` splice, and either may span a newline. The body is capped and possessive: an
# unterminated or over-long quote masks through the end of its line instead of only its
# first word, and never rescans the rest of the text from every flag.
_QUOTED_FLAG_VALUE = (r"\"(?:\\[\s\S]|[^\"\\]){1,512}+\"|'(?:'\\''|[^']){1,512}+'"
r"|[\"'][^\n]*+|\S+")
_CLI_SECRET_FLAG_RE = re.compile(
r"(?P<flag>(?<![\w-])--?(?:password|passwd|passphrase|pass|token|api[-_]?key|apikey"
r"|client[-_]?secret|access[-_]?key|account[-_]?key|auth[-_]?token|session[-_]?token|secret|bearer"
r"|credential)s?)"
r"(?P<sep>\s*=|\s+)"
rf"(?P<value>{_QUOTED_FLAG_VALUE})",
re.IGNORECASE,
)
_CLI_LOGIN_P_RE = re.compile(
r"(?P<flag>\b(?:docker|podman|nerdctl)\s+login\b[^\n;|&]*?\s-p\s*)"
rf"(?P<value>{_QUOTED_FLAG_VALUE})"
)
# Azure connection-string keys. They are ``Name=value`` inside a quoted string, so
# every ENV pass declines them: the name is mixed case and carries no underscore boundary.
# The key must open a connection-string segment (start, ``;``, ``,``, a URL ``?``/``&``, a quote or whitespace before
# it, no space around ``=``): ``this.accountKey = x`` / ``dt.PrimaryKey = new …`` in source
# read through read_file or a terminal stays byte-identical, or the agent cannot patch it.
# A spaced ``AccountKey = …`` (ini files, PowerShell) only counts when the value is a
# 16+ char base64 run carrying a digit, which an identifier like ``config.Key`` or
# ``StorageAccountKeyFromVault`` is not.
_CONNSTR_SECRET_KEY_RE = re.compile(
r"(?P<flag>(?<![^;,?&'\"\s])(?:AccountKey|SharedAccessKey|(?P<sas>SharedAccessSignature))"
r"(?:=|[ \t]*+=[ \t]*+(?=[A-Za-z+/]*[0-9])(?=[A-Za-z0-9+/]{16,}={0,2}(?![\w.(]))))"
# A SAS value is itself a query string (``sv=…&sig=…``): it runs through the ``sig=``
# parameter and stops there, or at a ``;`` / ``)`` / ``,`` / ``|`` / backtick delimiter.
r"(?P<value>(?(sas)(?:(?!sig=)[^;\s'\"&),`|]++&)*+(?:sig=)?[^;\s'\"&),`|]*+|[^;\s'\"&]+))",
re.IGNORECASE,
)
# Cheap pre-gates: every flag the regexes above can match contains one of these.
_CLI_SECRET_FLAG_GATE = re.compile(r"-{1,2}[A-Za-z]")
_CLI_LOGIN_GATE = re.compile(r"(?:docker|podman|nerdctl)\s+login\b", re.IGNORECASE)
_CONNSTR_GATE = re.compile(r"(?:AccountKey|SharedAccessKey|SharedAccessSignature)", re.IGNORECASE)
# Lowercase env names: only underscore-boundary forms (``openai_key=``) — NOT
# bare ``password=``/``token=``, which appear in prose, URLs, and form bodies.
# The lookbehind anchors each attempt to the start of an identifier run; without
# it re.sub retries the greedy prefix at every byte of a long opaque payload.
# See #77484.
_ENV_ASSIGN_LOWER_RE = re.compile(
r"(?<![a-z0-9_])([a-z0-9_]+(?:_|^)(?:key|pass|pw|token|secret|password|passwd|credential|auth)(?=[^a-z0-9_]|$))\s*=\s*(['\"]?)(\S+)\2",
re.IGNORECASE,
)
# Lowercase / dotted config-file keys (``spring.datasource.password=x``,
# line-start ``password=x``). Carve-outs vs prose/code/URLs: values stop at
# whitespace AND ``&`` (form bodies go pair-by-pair via _redact_form_body);
# _CFG_DOTTED_RE needs a NAMESPACED key; _CFG_ANCHORED_RE needs line start
# (optionally after ``export``). The ``://`` URL guard lives at the call site.
# The uppercase _ENV_ASSIGN_RE above never matched these, so config-file passwords leaked verbatim (issue
# #16413). These run only in a config-file context, NOT in prose, code, or URLs — three carve-outs preserved
# from the original design (#4367 + the documented web-URL passthrough below): 1. The value is bounded by
# ``[^\s&]`` (stops at whitespace AND ``&``) so form-urlencoded bodies are handled pair-by-pair (by
# _redact_form_body), not greedily swallowed. 2. _CFG_DOTTED_RE only matches when the key is NAMESPACED
# (contains a dot), which is unambiguously a config key — never a prose word. 3. _CFG_ANCHORED_RE matches a
# bare secret-word key only at line start (optionally after ``export``), so conversational ``I have
# password=foo`` mid-sentence is left alone.
_SECRET_CFG_NAMES = r"(?:api[ _.\-]?key|token|secret|passwd|password|credential|auth)"
# Rendered line-number prefix: ``5|line`` (read_file), ``6:line`` (grep -n), ``7-line`` (grep -A/-B/-C
# context lines) and `` 8\tline`` (cat -n / nl: right-aligned number + TAB). Callers put the ONLY
# leading ``[ \t]*`` in front of it — stacking a second whitespace run around an optional gutter made
# the anchored passes quadratic on long indented lines (2s per 5k spaces).
_LINE_NUMBER_GUTTER = r"(?:[0-9]+(?:[|:\-]|\t)[ \t]*)?"
_CFG_VALUE = r"(['\"]?)([^\s&]+?)\2(?=[\s&]|$)"
# Linear pre-gate for the _CFG_*_RE subs: no secret keyword => neither can match.
_CFG_SECRET_WORD_RE = re.compile(_SECRET_CFG_NAMES, re.IGNORECASE)
# Programmatic env lookups (``os.getenv(...)``, ``process.env.X``, ``$ENV{X}``)
# as the VALUE of a KEY=... match name a variable; they are not a leaked secret.
_ENV_LOOKUP_VALUE_RE = re.compile(r"^(?:os\.(?:getenv|environ)|process\.env|\$ENV\{)")
# Namespaced key: the secret word may sit anywhere in a dotted path.
# NOTE(perf): possessive quantifiers (nested ``(?:[...]+\.)+`` backtracked
# exponentially); the ``*`` runs bordering {_SECRET_CFG_NAMES} must stay
# backtrackable (``app.api.key=``). The lookbehind anchors each attempt to a key
# run start so re.sub is not quadratic; the match set is unchanged.
_CFG_DOTTED_RE = re.compile(
rf"(?<![A-Za-z0-9_.\-])"
rf"([A-Za-z0-9_\-]++\.[A-Za-z0-9_.\-]*{_SECRET_CFG_NAMES}[A-Za-z0-9_.\-]*+"
rf"|[A-Za-z0-9_.\-]*{_SECRET_CFG_NAMES}[A-Za-z0-9_.\-]*\.[A-Za-z0-9_.\-]++)"
rf"={_CFG_VALUE}",
re.IGNORECASE,
)
# Line-anchored bare key: ``password=…`` / ``export api_key=…`` at start of line.
# ``{_LINE_NUMBER_GUTTER}``: line-numbered dumps put the key behind a rendered gutter —
# ``read_file`` emits ``5| ADS_API_TOKEN: …``, ``grep -n`` emits ``6: ADS_API_TOKEN: …``
# and ``cat -n`` emits `` 7\tADS_API_TOKEN: …``. Anchored at ``^`` without it, none of those
# matched, so the rendered read of a secret-bearing file leaked what the raw text masked.
_CFG_ANCHORED_RE = re.compile(
rf"(^[ \t]*{_LINE_NUMBER_GUTTER}(?:export[ \t]+)?[A-Za-z0-9_\-]*{_SECRET_CFG_NAMES}[A-Za-z0-9_\-]*)={_CFG_VALUE}",
re.IGNORECASE | re.MULTILINE,
)
# YAML / colon config: bare-key quoted scalars cannot match _JSON_FIELD_RE.
# Match escaped double quotes, doubled single quotes, or an unquoted token.
# An unterminated quote masks to end-of-line; no branch crosses LF or CRLF.
# NOTE(perf): possessive where the successor is disjoint; the leading class
# stays backtrackable (see _CFG_DOTTED_RE).
_YAML_CFG_NAMES = (
r"(?:api[ _.\-]?key|token|secret|passwd|password|credential"
r"|client[ _.\-]?key[ _.\-]?data|(?:private|signing|encryption|master|hmac)[ _.\-]?key)"
)
_YAML_ASSIGN_RE = re.compile(
rf"(^[ \t]*+{_LINE_NUMBER_GUTTER}[A-Za-z0-9_.\-]*{_YAML_CFG_NAMES}[A-Za-z0-9_.\-]*+)(:[ \t]*+)"
r"(\"(?:\\[^\r\n]|[^\"\\\r\n])*\"|'(?:''|[^'\r\n])*'|['\"][^\r\n]*|[^\s&]++)",
re.IGNORECASE | re.MULTILINE,
)
_YAML_CFG_WORD_RE = re.compile(_YAML_CFG_NAMES, re.IGNORECASE)
# Word-boundary validation for the key patterns above: their classes allow
# arbitrary affixes (``client_secret``, ``s3.secret-key``), which also matched
# prose CONTAINING a keyword (``Secretary:``, ``tokenizer:``). A keyword counts
# only at a word boundary: key edge, next to a non-letter, or a camelCase
# transition (``clientSecret``, ``APIToken``); trailing plural ``s`` is part of
# it. Concatenations match via explicit alternatives (``authtoken``, ``apikey``).
# The side effect: ordinary prose/document words that merely CONTAIN a keyword also matched — ``Secretary:
# J.Smith`` (secret), ``tokenizer: cl100k_base`` (token), ``author=Smith`` (auth) — mangling legitimate
# content on the surfaces that run these passes (browser snapshots, log lines, kanban summaries, CLI-echoed
# command output). Ported from nearai/ironclaw#6129, where the same substring false positive ("Secretary of
# the Treasury" matching the ``secret`` marker) scrubbed legitimate tool results from the replayed
# transcript and sent the model into a re-fetch loop. Common concatenated compounds keep matching via
# explicit alternatives (``authtoken`` ngrok, ``authkey`` tailscale, ``secretkey`` minio, ``apikey``).
# Embedded occurrences inside a larger word (``secretary``, ``tokenizer``, ``authored``, ``credentialing``)
# no longer match. ALL-CAPS keys keep the legacy embedded matching (``MYTOKEN=…``) — an all-caps key is
# almost never prose, the same rationale as _ENV_ASSIGN_RE.
_KEY_KEYWORD_RE = re.compile(
r"(?:api|auth|access|refresh|session|secret)[ _.\\-]?(?:key|token)"
r"|token|secret|passwd|password|pass|pw|credential|auth|key",
re.IGNORECASE,
)
# Key names that are credential-specific even when their values are short or
# human-readable. Bare ``token`` / ``key`` are intentionally absent: they also
# describe model limits, tensor names, and cache keys, so those assignments
# are gated on value shape (_looks_like_opaque_credential).
_STRONG_KEY_KEYWORD_RE = re.compile(
r"(?:api|auth|access|refresh|session|id|bearer)[ _.\\-]?(?:key|token)"
r"|key[ _.\\-]?material|secret|passwd|password|pass|pw|credential|auth|bearer",
re.IGNORECASE,
)
# Password-class keys mask any literal value; for other keys a value that starts like ``$HOME/...``,
# ``/usr/...`` or ``~/...`` references a variable or a path, not a credential, even under a strong key
# (``SSH_AUTH_SOCK=$HOME/.ssh/agent.sock``, ``DOCKER_AUTH_CONFIG=/home/u/.docker``).
_PASSWORD_KEY_RE = re.compile(r"passwd|password|pass|pw", re.IGNORECASE)
# Anchored on both ends: the whole value must be a ``$VAR``/``${VAR}`` reference, a ``~/``
# path, or an absolute path — not merely a string whose FIRST character is one of those.
# A 40-char AWS secret key starts with '/' ~1 in 64 times and argon2/bcrypt digests always
# start with '$'; an unanchored class let those secrets skip every check below.
# Further ``$VAR`` interpolations may appear anywhere in the path (``/run/user/$UID/ssh``,
# ``$XDG_RUNTIME_DIR/agent.$USER.sock``, ``$A:$B`` lists); crypt digests never parse as one
# because their ``$`` fields start with a digit or carry ``=``/``,``.
# A leading ``$(`` is a command substitution (``SSH_AUTH_SOCK=$(gpgconf --list-dirs
# agent-ssh-socket)``): the value token stops at whitespace, so only ``$(gpgconf`` is seen.
_SHELL_VAR_REF = r"\$(?:\{[A-Za-z_]\w*[^}]*\}|[A-Za-z_]\w*)"
_PATH_OR_VAR_VALUE_RE = re.compile(rf"^(?:{_SHELL_VAR_REF}|\$\(|~|/)(?:[\w./:-]|{_SHELL_VAR_REF})*$")
# ``$VAR`` / ``$(cmd`` are unambiguous references. A ``/``- or ``~``-led value is a path only
# while every segment reads like one: a 16+ char segment mixing case and digits with no ``.``
# (``/wJalrXUtnFEMIK7MDENG/bPxRf…``) is a secret that happens to start with a path character,
# whereas ``/home/u/.docker`` / ``~/.ssh/id_rsa`` / ``S.gpg-agent.ssh`` never clear that bar.
_OPAQUE_PATH_SEGMENT_RE = re.compile(r"(?=[^.]*[a-z])(?=[^.]*[A-Z])(?=[^.]*[0-9])[^.]{16,}")
def _is_word_start(s: str, i: int) -> bool:
"""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()): # clientSecret
return True
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; 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()): # secretKey
return True
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))
_PASSWORD_WORD_RE = re.compile(r"passwd|password", re.IGNORECASE)
def _has_password_keyword(key: str) -> bool:
"""``password``/``passwd`` ending a word, wherever it starts: no prose word ends in them, so
concatenated keys (``dbpassword``, ``rootpasswd``) are credentials, while ``passwordless``
is not. Requiring a word START too passed ``dbpassword: <v>`` through in clear."""
return any(_is_word_end(key, m.end()) for m in _PASSWORD_WORD_RE.finditer(key))
# Tool-defined env names that glue the keyword onto a prefix with no separator, so the
# word-boundary test can never see it: libpq's ``PGPASSWORD``, MySQL's ``MYSQL_PWD`` (bare
# ``PWD`` is the shell's working directory, so only this exact name), sshpass's ``SSHPASS``.
_GLUED_CREDENTIAL_ENV_RE = re.compile(r"PGPASSWORD|MYSQL_PWD|SSHPASS", re.IGNORECASE)
def _is_glued_credential_env(key: str) -> bool:
return _GLUED_CREDENTIAL_ENV_RE.fullmatch(key.strip("\"'")) is not None
def _key_has_secret_keyword(key: str) -> bool:
"""Post-match key validator: ``API_KEY``/``DB_PW`` count, ``KEYBOARD``/``secretary`` do not."""
return (_has_word_bounded_keyword(key, _KEY_KEYWORD_RE) or _has_password_keyword(key)
or _is_glued_credential_env(key))
def _looks_like_opaque_credential(value: str) -> bool:
"""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):
return True
if len(value) >= 20 and re.fullmatch(r"[A-Za-z0-9_./+=-]+", value):
return True
if len(value) < 12:
return False
return sum(bool(re.search(p, value)) for p in (r"[a-z]", r"[A-Z]", r"[0-9]")) >= 2
def _should_redact_assignment(key: str, value: str, *, check_keyword: bool) -> bool:
"""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."""
# Programmatic env lookups reference variable *names*, not secret values — masking them corrupts code
# snippets in prose/log contexts (issue #2852): ``KEY=os.getenv('X')``.
# Same programmatic-env-lookup exception as _redact_env above (issue #2852): "apiKey": "os.getenv('X')"
# is a code snippet, not a leaked secret value.
# Same programmatic-env-lookup exception as _redact_env above (issue #2852): api_key: os.getenv('X') is
# a code snippet, not a leaked secret value.
if _ENV_LOOKUP_VALUE_RE.match(value):
return False
# An earlier pass already masked this value (``***`` or the ``«redacted:…»`` sentinel). Masking it
# again only erases what the sentinel deliberately kept — the vendor label (``Digest ***`` →
# ``***``, ``«redacted:ghp_…»`` → ``«redacted-secret»``). Same guard _redact_python_repr_fields uses.
if value == "***" or value.startswith("«redacted"):
return False
if check_keyword and not _key_has_secret_keyword(key):
return False
# A shell rc's ``SSH_AUTH_SOCK=$HOME/.ssh/agent.sock`` is configuration the agent must keep
# readable; only password-class keys mask a path/variable reference.
password_key = _has_word_bounded_keyword(key, _PASSWORD_KEY_RE) or _has_password_keyword(key)
if _PATH_OR_VAR_VALUE_RE.match(value) and not password_key:
# ``$VAR`` is an unambiguous reference. A bare ``/...`` or ``~...`` is not:
# ``/home/u/.docker`` and ``/8f3kd9sKd0als...`` have the same shape, so every
# segment has to look like a path (see _OPAQUE_PATH_SEGMENT_RE) before the
# value is treated as configuration.
if value[0] == "$" or not any(_OPAQUE_PATH_SEGMENT_RE.fullmatch(seg) for seg in value.split("/")):
return False
# ``signing_key`` / ``privateKey`` are credentials at any length, the same policy
# is_secret_field_name applies to mappings, so JSON text and dicts agree.
return (password_key or _has_word_bounded_keyword(key, _STRONG_KEY_KEYWORD_RE)
or _is_glued_credential_env(key)
or (_is_qualified_key_name(key) and not _CONFIG_PLACEHOLDER_VALUE_RE.fullmatch(value))
or _looks_like_opaque_credential(value))
# JSON field patterns: "apiKey": "value", "token": "value", etc.
# ``key``-suffixed credential names (``aws_secret_access_key``, ``SECRET_KEY``,
# ``privateKey``, ``signing_key``) and kubeconfig's ``client-key-data`` (a base64 PEM).
# The bare word ``key`` stays out: it names cache keys and model limits. A qualifier
# word has to sit in front of it, which is what the alternation spells out.
_KEY_QUALIFIED_NAME = (
r"[\w.\-]*(?:secret|private|signing|encryption|encrypt|master|hmac|symmetric|access)"
r"[_.\-]?(?:key|Key|KEY)"
)
_JSON_KEY_NAMES = (
r"(?:api_?[Kk]ey|token|secret|password|pgpassword|mysql_pwd|sshpass|access_token|refresh_token|auth_token|bearer"
r"|secret_value|raw_secret|secret_input|key_material|client[_.\-]?key[_.\-]?data"
rf"|{_KEY_QUALIFIED_NAME})"
)
_JSON_FIELD_RE = re.compile(rf'("{_JSON_KEY_NAMES}")\s*:\s*"([^"]+)"', re.IGNORECASE)
# The same field inside a JSON-encoded string (``{"output": "{\\"api_key\\": \\"…\\"}"}``): every tool result
# that wraps a config dump in JSON escapes the quotes, so the rule above never saw the value (#115104).
# The backreference keeps the key's and value's escape depth equal.
_JSON_FIELD_ESCAPED_RE = re.compile(rf'((\\+)"{_JSON_KEY_NAMES}\2")\s*:\s*\2"([^"\\]+)\2"', re.IGNORECASE)
# Any other identifier-shaped key (``"API_TOKEN"``, ``"DB_PASSWORD"``, ``"client_secret"``): the closed
# list above misses affixed names that the ENV and YAML passes already accept (#147240). The callback
# applies the YAML pass's word set at a word boundary, then the shared assignment gate.
# The value may carry JSON escapes (``"s3cret\\value"``): ``\\.`` consumes each escape pair whole.
_JSON_ANY_KEY_FIELD_RE = re.compile(r'("([A-Za-z0-9_.\-]{1,128}+)")\s*+:\s*+"((?:[^"\\\n]|\\.)++)"')
# The same field inside a JSON-encoded string (``{"output": "{\\"DB_PASSWORD\\": \\"…\\"}"}``),
# like _JSON_FIELD_ESCAPED_RE does for the closed key list: group 2 is the escape run.
_JSON_ANY_KEY_FIELD_ESCAPED_RE = re.compile(
r'((\\++)"([A-Za-z0-9_.\-]{1,128}+)\2")\s*+:\s*+\2"([^"\\\n]++)\2"')
_JSON_KEY_NAMES_FULL_RE = re.compile(_JSON_KEY_NAMES, re.IGNORECASE)
# Python ``repr`` uses single-quoted mapping fields, so opaque credentials in
# tracebacks and pytest failure introspection bypass the double-quoted JSON rule
# above: ``{'BRAVE_API_KEY': 'opaque-value'}``. Capture identifier-shaped keys
# here, then apply the canonical high-confidence key policy in the callback.
_PYTHON_REPR_SECRET_KEYS = frozenset({
"access_token",
"refresh_token",
"id_token",
"auth_token",
"token",
"api_key",
"apikey",
"client_secret",
"secret",
"password",
"passwd",
"private_key",
"credential",
"credentials",
"authorization",
"bearer",
"secret_value",
"raw_secret",
"secret_input",
"key_material",
})
_PYTHON_REPR_ENV_SUFFIXES = (
"_API_KEY",
"_TOKEN",
"_SECRET",
"_PASSWORD",
"_PASSWD",
"_CREDENTIAL",
"_CREDENTIALS",
)
# Casefolded credential suffixes for mixed/camel-case key names
# (``UserPassword``, ``sessionToken``, ``clientApiKey``). Suffix-only so
# ``token_count`` / ``password_policy`` metadata keys never match. Widened per
# OpenHands/software-agent-sdk#4508.
_PYTHON_REPR_CREDENTIAL_SUFFIXES = (
"apikey",
"api_key",
"token",
"secret",
"password",
"passwd",
"credential",
"credentials",
)
_PYTHON_REPR_FIELD_RE = re.compile(
r"'(?P<key>[A-Za-z_][A-Za-z0-9_]*)'(?P<sep>\s*:\s*)"
r"(?:"
r"(?P<single_prefix>[bB]?)'(?P<single_value>(?:\\.|[^'\\])+)'"
r"|(?P<double_prefix>[bB]?)\"(?P<double_value>(?:\\.|[^\"\\])+)\""
r")"
)
# Terminal/process output normally uses ``code_file=True`` to preserve source.
# Add repr masking only to high-confidence diagnostic lines: pytest assertion
# introspection (``E ...``) and final Python exception lines.
_PYTEST_DIAGNOSTIC_LINE_RE = re.compile(r"^(?P<prefix>[ \t]*E[ \t]{2,})(?P<body>.*)$")
_PYTHON_EXCEPTION_LINE_RE = re.compile(
r"^(?P<prefix>(?:[A-Za-z_]\w*\.)*[A-Za-z_]\w*"
r"(?:Error|Exception|Warning):[ \t]*)(?P<body>.*)$"
)
# Authorization / Proxy-Authorization, any scheme or bare credential; header
# name and scheme word preserved. The credential class excludes quotes: pulling
# a closing quote into the mask turns value corruption into SYNTAX corruption
# (unterminated quote → shell EOF / SyntaxError).
_AUTH_HEADER_RE = re.compile(r"((?:Proxy-)?Authorization:\s*)([A-Za-z][\w.+-]*\s+)?([^\s\"']+)", re.IGNORECASE)
# Telegram bot tokens: [bot]<digits>:<token>, token >= 30 chars. Start at an
# identifier boundary, never inside a digit run (quadratic retries on dumps).
# Consume only bounded escape spellings so JSON/repr/URL separators survive
# verbatim; their last letter/digit is not the token's logical left neighbor.
# An ANSI CSI control (colored output), raw or JSON/repr-escaped, is a separator too.
_TELEGRAM_RE = re.compile(
r"(?P<boundary>\\[nrtbfv]|\\x[0-9a-fA-F]{2}|\\u[0-9a-fA-F]{4}|"
r"\\U[0-9a-fA-F]{8}|%[0-9a-fA-F]{2}|(?P<csi>(?:\x1b|\\x1b|\\u001[bB]|\\033)\[[0-?]*[ -/]*[@-~])|(?<![A-Za-z_\d]))"
r"(?P<bot>bot)?(?P<id>\d{8,}):(?P<body>[-A-Za-z0-9_]{30,})"
)
def _redact_telegram_token(match: re.Match) -> str:
boundary = match.group("boundary")
# Encoded identifier characters are still identifiers, not separators.
if len(boundary) > 2 and not match.group("csi"):
value = int(boundary[1:] if boundary[0] == "%" else boundary[2:], 16)
if value > 0x10FFFF:
return match.group(0)
char = chr(value)
if char in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz_" or char.isdecimal():
return match.group(0)
return f"{boundary}{match.group('bot') or ''}{match.group('id')}:***"
_PRIVATE_KEY_RE = re.compile(r"-----BEGIN[A-Z ]*PRIVATE KEY-----[\s\S]*?-----END[A-Z ]*PRIVATE KEY-----")
# A base64-encoded PEM private key: kubeconfig's ``client-key-data`` and the ``*_b64``
# fields around it (``kubectl config view --raw``, cloud-init, CI secrets). The name is
# not reliable — kubeconfig writes ``client-key-data`` in YAML and JSON alike, and the
# terminal path deliberately skips the assignment passes on source-shaped output — so the
# rule keys on the payload: base64 that decodes to a ``BEGIN … PRIVATE KEY`` header.
_B64_PEM_RE = re.compile(r"LS0tLS1CRUdJTi[A-Za-z0-9+/=]{60,}")
# Private-key formats that are not PEM. All three reach the model on their own: PGP's
# armour carries a \` BLOCK\` suffix so the PEM pattern never matched it, PuTTY's .ppk is
# base64 under a \`Private-Lines:\` header (the usual SSH key format on Windows), and age
# writes a bech32 AGE-SECRET-KEY-1… string.
_PGP_PRIVATE_KEY_RE = re.compile(
r"-----BEGIN PGP PRIVATE KEY BLOCK-----[\s\S]*?-----END PGP PRIVATE KEY BLOCK-----"
)
# Line ends may be CRLF (Windows writes .ppk that way) or a JSON-escaped ``\\n``.
_PPK_EOL = r"(?:\r?\n|\\r\\n|\\n)"
# Body lines may carry read_file's `` 4|`` line-number gutter.
_PPK_BODY_GUTTER = r"[ \t]*+(?:[0-9]++(?:[|:]|\t)[ \t]*+)?"
_PUTTY_PRIVATE_LINES_RE = re.compile(
rf"(?<![\w-])(Private-Lines:[ \t]*+[0-9]++[ \t]*+(?P<eol>{_PPK_EOL}))"
rf"(?:{_PPK_BODY_GUTTER}[A-Za-z0-9+/=]{{16,}}+[ \t]*+{_PPK_EOL}?)++"
)
_AGE_SECRET_KEY_RE = re.compile(r"AGE-SECRET-KEY-1[0-9A-Z]{20,}")
def _is_b64_pem_key(blob: str) -> bool:
"""True when ``blob`` is base64 of a PEM private key block."""
import base64
import binascii
padded = blob + "=" * (-len(blob) % 4)
try:
decoded = base64.b64decode(padded, validate=False)
except (binascii.Error, ValueError):
return False
text = decoded[:64]
return text.startswith(b"-----BEGIN") and b"PRIVATE KEY" in text
# Database connection strings: protocol://user:PASSWORD@host. The userinfo and
# password groups forbid whitespace so a match can never span a line break (a
# greedy ``[^@]+`` once ran to a decorator's ``@`` on the next code line).
# Database connection strings: protocol://user:PASSWORD@host Catches postgres, mysql, mongodb, redis, amqp
# URLs and redacts the password. A real DSN password never contains whitespace; without this bound the
# greedy [^@]+ would scan past the end of a code line to the next stray "@" (e.g. a Python decorator),
# swallowing intervening lines and corrupting tool OUTPUT for any source containing a postgresql:// f-string
# template. See issue #33801.
_DB_CONNSTR_RE = re.compile(
r"((?:postgres(?:ql)?|mysql|mongodb(?:\+srv)?|rediss?|amqps?)://[^:\s]*+:)((?:[^@\s:]++|:(?!//))++)(@)",
re.IGNORECASE,
)
# Bare-token URL credential ``scheme://TOKEN@host`` (``git remote set-url
# https://PASSWORD@github.com/...``): unambiguously a secret, since round-trip
# URLs (OAuth callbacks, magic links) carry tokens in the QUERY STRING, never
# bare userinfo. ``user:pass@`` passes through (class forbids ``:``); DB schemes
# belong to _DB_CONNSTR_RE. 8+ char floor skips short usernames; the class
# forbids ``/`` so an ``@`` in a path/query (``?q=user@example.com``) never counts.
# The class also forbids ``"``: a double quote is not URL userinfo, and
# ``https://schema.org","@type`` is a JSON string boundary, not a credential
# (src#903). A comma stays legal userinfo.
# This is the ``git remote set-url origin https://PASSWORD@github.com/...`` shape from issue #6396 — a
# single opaque credential in the userinfo position with NO ``user:pass`` colon. The colon form's
# PASSWORD is masked separately by _redact_userinfo_passwords (issue #125664) — #34029's pass-through
# was about query-string round-trip tokens (magic links / OAuth callbacks / pre-signed URLs live in
# the query, never in userinfo), and a userinfo password is a credential, not a workflow token. DB
# schemes are handled by _DB_CONNSTR_RE above and excluded here. Guards against false positives:
_URL_BARE_TOKEN_RE = re.compile(
r"((?:https?|wss?|git|ssh|ftp|ftps|sftp)://)" # scheme
r"([^\s:@/\"]{8,})" # bare token, 8+ chars; no quote
r"(@[^\s]+)", # @host...
re.IGNORECASE,
)
# 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}")
# Links ``hermes webapp`` prints that sign a browser in as the user: the gateway's browser credential
# after ``hermes-session=`` and the one-use ticket of a ``/webapp/launch`` link. Both values are
# URL-safe base64 (43 characters); the 20-character floor leaves placeholders and code (``${token}``).
_WEBAPP_LINK_CREDENTIAL_RE = re.compile(
r"((?<![A-Za-z0-9_-])hermes-session=|/webapp/launch#(?:[^\s#]*?&)?ticket=)[A-Za-z0-9_-]{20,}")
# 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])")
# 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*)?")
_URL_USERINFO_RE = re.compile(r"(https?|wss?|ftp)://([^/\s:@]+):([^/\s@]+)@")
# Strict provider-egress URL redaction: delimiters stay in capture groups so the
# query/fragment layout is preserved byte-for-byte; the key is decoded
# separately for classification. Values stop at ``&``/``;`` (both valid).
_STRICT_URL_PARAM_RE = re.compile(r"([?#&;])([A-Za-z0-9_.~+%\-]+)=([^#&;\s\"'<>]*)")
# Userinfo in absolute and network-path (``//user:pass@host``) references; the
# authority stops at path/query/fragment delimiters. Anchored on the mandatory
# ``//`` — an optional-scheme prefix backtracked O(n²) on long alphanumeric runs
# (~55s per sub() on a 320KB compaction payload).
# No ``"``: ``"https://a.com","u:pw@y"`` is two JSON strings, not ``a.com","u`` userinfo.
_STRICT_URL_USERINFO_RE = re.compile(r"//[^/\s?#@\"]+@")
# The same authority span with the delimiter and userinfo captured, for _redact_userinfo_passwords.
_URL_USERINFO_PAIR_RE = re.compile(r"(//)([^/\s?#@\"]+)@")
# Scheme-less network path (``mount //user:pass@host/share``), masked under force only: ``//`` must
# start a token (comments are ``// note``, protocol-relative URLs carry no ``@``) and the password
# needs 8+ characters, so ``//user:note@x`` stays as written.
_NETWORK_PATH_USERINFO_RE = re.compile(r"(?<![^\s\"'=(`,\[])(//)([^/\s?#@:\"]+):([^/\s?#@\"]{8,})@")
# ``NAME=user:pass`` where NAME carries no secret keyword — an HTTP Basic auth pair is often stored
# this way (#125664). Shape-only on arbitrary names, so it runs force-only (persistence boundary,
# debug dumps, session export all pass force=True): ordinary prose/config keeps today's behaviour.
# Guards against non-secret colon values: the password needs an 8-char floor AND a letter (ports and
# digit-only tags fail it), the password group is a single token (no whitespace, quotes or ``/`` —
# so ``registry.example.com:5000/repo`` keeps its path, and an ISO timestamp's ``:00:00`` tail
# truncates the group to digits), and a version-shaped tail (``24.04-noble``, ``v1.2.3-alpine``)
# is rejected outright. The value is the WHOLE assignment (``user`` cannot contain ``:``/``@``/``/``),
# and the ``:(?!//)`` lookahead keeps ``scheme://`` values (``MY_URL=https://example.com``) out.
# ``user`` must also be at least 2 characters: a single-letter ``user:`` is a Windows drive
# (``SRC=C:\Users\...``), and masking the whole path as ``SRC=C:***`` at a force boundary is
# unrecoverable data loss — no real Basic pair has a one-letter user, and a secret-keyword NAME
# still masks via the ENV pass (#125680 review).
# ``@`` is not a password character here either: a ``user:pass@host``-shaped value belongs to the
# userinfo pass below, which knows the f-string ``{...}`` template exemption; excluding it lets that
# pass see the pure brace expression (``BASIC="user:{pw}@host"`` keeps its template verbatim).
_BASIC_PAIR_ASSIGN_RE = re.compile(
r"(?<![A-Za-z0-9_.\-])([A-Za-z][A-Za-z0-9_.\-]{0,63})=([\"']?)"
r"([A-Za-z0-9._~\-]{2,})(?::(?!//)([^\s'\"/:@]+))\2(?![^\s;,&|)\]}\"'`])"
)
# ``24.04-noble`` / ``v1.2.3-alpine`` image tags: digit-led dotted versions, optionally ``v``-prefixed
# — long and letter-bearing enough to pass the floor, yet never a credential.
_VERSION_TAG_RE = re.compile(r"^[vV]?\d+(\.\d+)+")
# ``nginx:stable-alpine`` / ``myrepo:latest-release``: lower-case words joined by ``-``/``.``
# are image tags and labels; a password carries a digit, a capital or a symbol.
_WORD_TAG_RE = re.compile(r"[a-z]++(?:[.\-][a-z]++)*+")
# ``CHECKSUM=sha256:1234…`` digest labels: the hex body is long and letter-bearing, but a hash
# algorithm name as the ``user`` half marks the pair as an algorithm:digest, not a credential
# (#125680 review). No real Basic pair has one of these as its username.
_HASH_ALGO_NAMES = frozenset({
"md5", "sha1", "sha224", "sha256", "sha384", "sha512", "sha3-224", "sha3-256",
"sha3-384", "sha3-512", "blake2b", "blake2s", "ripemd160", "crc32", "crc32c",
})
# ``NAME=user:pass`` is only a Basic pair when NAME says it holds one: ``IMAGE=redis:7-alpine``,
# ``BRANCH=origin:feature2``, ``LS_COLORS=...:mh=00:...`` and ``Server=tcp:db,1433`` are colon
# values under names that never carry credentials (#149435 review).
_BASIC_PAIR_NAME_RE = re.compile(r"auth|basic|cred|login|proxy|userpass|userinfo", re.IGNORECASE)
def _is_basic_pair_password(password: str) -> bool:
"""8+ chars with a letter and not a version-shaped tail — a secret, not a tag or timestamp."""
if _VERSION_TAG_RE.search(password) or _WORD_TAG_RE.fullmatch(password):
return False
return len(password) >= 8 and any(c.isalpha() for c in password)
# 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``,
# ``ghp_abc\n123``) and escape the contiguous prefix regexes.
_CONTROL_CHARS_RE = re.compile(r"[\x00-\x1f\x7f\u200b-\u200f\u2028-\u202f\u2060\ufeff]")
# Union of every _PREFIX_PATTERNS body class: a control-stripped match may only
# span token-body or control chars. ``=`` is excluded so a KEY=value separator
# never lets a match span unrelated text.
_TOKEN_BODY_CHARS = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-.")
def _compile_prefix_matcher(patterns: list) -> re.Pattern[str]:
return re.compile(r"(?<![A-Za-z0-9_-])(" + "|".join(patterns) + r")(?![A-Za-z0-9_-])")
def _compile_prefix_sub_matcher(patterns: list) -> re.Pattern[str]:
"""_PREFIX_RE for the masking pass, also admitting a key glued to an identifier by ``-``/``_``
(``backup-sk-…``, ``x-ghp_…``), which _PREFIX_RE rejects so ``scikit-learn`` stays whole. One
fixed-width lookbehind keeps main's single scan; _mask_prefix_match masks a glued match only
when its body is opaque, so ``backup-sk-learn`` stays whole too (#147240)."""
# Branches grouped by their literal first char (every pattern opens on one, see
# register_redaction_patterns): admitting ``-``/``_`` before a key makes every separator a
# candidate, and a flat 60-way alternation per candidate doubled the cost of a ``_ghp_`` flood.
# Groups are disjoint by first char, so leftmost-first match order is unchanged.
by_first: dict = {}
for pattern in patterns:
by_first.setdefault(pattern[0], []).append(pattern[1:])
body = "|".join(f"{re.escape(c)}(?:{'|'.join(rest)})" for c, rest in by_first.items())
return re.compile(r"(?<![A-Za-z0-9])(" + body + r")(?![A-Za-z0-9_-])")
_PREFIX_RE = _compile_prefix_matcher(_PREFIX_PATTERNS)
_PREFIX_SUB_RE = _compile_prefix_sub_matcher(_PREFIX_PATTERNS)
# A real key body has a 16+ char run with no separator mixing two of lower/upper/digit;
# ``sk-prod-deployment-01`` and ``fc-feature-flags-v2`` do not.
_LONG_ALNUM_RUN_RE = re.compile(r"[A-Za-z0-9]{16,}+")
def _is_opaque_prefixed_token(token: str) -> bool:
for run in _LONG_ALNUM_RUN_RE.finditer(token):
run = run.group(0)
if (run.isalpha() and (run.islower() or run.isupper())) or run.isdigit():
continue
return True
return False
def _mask_prefix_match(m: "re.Match[str]", mask) -> str:
glued = m.start() > 0 and m.string[m.start() - 1] in "-_"
return m.group(0) if glued and not _is_opaque_prefixed_token(m.group(1)) else mask(m.group(1))
# Zhipu API keys use an unprefixed ``id.secret`` form. Keep this deliberately
# provider-shaped instead of applying a generic high-entropy dotted-token rule:
# the ID is exactly 32 lowercase hex chars and the credential suffix is a run of
# at least 16 alphanumerics, so content-hash filenames (``<sha>.bundle``,
# ``<md5>.sqlite3``) never match.
_ZHIPU_API_KEY_RE = re.compile(
r"(?<![A-Za-z0-9_.-])([0-9a-f]{32}\.[A-Za-z0-9]{16,})(?![A-Za-z0-9_.-])"
)
def _mask_control_split_tokens(text: str, mask_fn) -> str:
"""Mask tokens whose body is split by control/zero-width characters.
Match on a control-stripped copy, then mask the corresponding span in the
ORIGINAL — only when that span holds solely token-body and control chars, so
a match can never cross into another line's unrelated text.
A credential like ``sk-abc\\x1bdef456…`` or ``ghp_abc\\n123def…`` has its token body interrupted, so the
contiguous _PREFIX_RE cannot match it and the secret leaks verbatim (issue #77484). ``EXA_API_KEY=*** is
rejected).
"""
stripped = _CONTROL_CHARS_RE.sub("", text)
if stripped == text:
return text
orig_idx = [i for i, c in enumerate(text) if not _CONTROL_CHARS_RE.match(c)]
out, matches = list(text), []
for m in _PREFIX_RE.finditer(stripped):
start_orig = orig_idx[m.start(1)]
end_orig = orig_idx[m.end(1) - 1] + 1
span = text[start_orig:end_orig]
# Self-matching fragment AND a span crossing a LINE boundary: do NOT join
# (``ghp_<tok>\nbutton`` would mask ``button``; the prefix pass handles the
# fragment). Non-newline controls (ESC, ZWSP) never legitimately sit
# between a token and prose, so there the join proceeds regardless.
if ("\n" in span or "\r" in span) and _PREFIX_RE.search(span):
continue
# Reject spans containing a non-token char (``sk_abc…\nTAVILY_API_KEY=``
# matched across lines) and matches running into a ``KEY=`` name.
if (all(c in _TOKEN_BODY_CHARS or _CONTROL_CHARS_RE.match(c) for c in span)
and (end_orig >= len(text) or text[end_orig] != "=")):
matches.append((start_orig, end_orig, mask_fn(m.group(1))))
for start_orig, end_orig, replacement in reversed(matches):
out[start_orig:end_orig] = list(replacement)
return "".join(out)
# mask_secret strips EVERY control char (incl. \n/\t, C1, DEL, zero-width) so a
# masked secret never emits multiline or invisible bytes into display output.
_DISPLAY_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f\x80-\x9f\u200b-\u200f\u202a-\u202e\u2060-\u2064]")
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``):
``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
return placeholder if len(value) < floor else f"{value[:head]}...{value[-tail:]}"
# Approval-prompt display integrity: the copy a human approves must show everything that will run.
# Code points that render as nothing or as a line break become visible escapes: every control,
# format, private-use, unassigned and surrogate code point (category C*, except tab/newline), line
# and paragraph separators (Zl/Zp), and blank-rendering marks/letters in other categories (variation
# selectors incl. U+E0100-E01EF, CGJ, Khmer/Mongolian selectors, Hangul fillers, Braille blank).
# Ordinary non-ASCII text is untouched. Padding that pushes the tail off screen or past a preview
# cut (space runs, whitespace-only lines) collapses to a marker; real indentation is kept.
_APPROVAL_NON_ASCII_RE = re.compile(r"[^\x20-\x7e\t\n]+")
_APPROVAL_BLANK_GLYPHS = frozenset(
"\u034f\u115f\u1160\u17b4\u17b5\u180b\u180c\u180d\u180f\u2800\u3164\uffa0"
+ "".join(map(chr, range(0xFE00, 0xFE10))) + "".join(map(chr, range(0xE0100, 0xE01F0))))
_APPROVAL_SPACE = r"[ \t\u00a0\u1680\u2000-\u200a\u202f\u205f\u3000]"
# Mid-line runs (after visible text) of 21+, or line-leading runs past any sane indentation.
_APPROVAL_PAD_RE = re.compile(rf"(?<=\S){_APPROVAL_SPACE}{{21,}}|(?m:^){_APPROVAL_SPACE}{{65,}}")
# 3+ consecutive blank or whitespace-only lines; the match ends at a newline so the next line keeps its indent.
_APPROVAL_BLANK_RE = re.compile(rf"\n(?:{_APPROVAL_SPACE}*\n){{3,}}")
_APPROVAL_SAME_CHAR_RUN_RE = re.compile(r"(.)\1*", re.DOTALL)
# Prose mode (questions shown to a human, never executed) surfaces only what can hide or reorder
# text: controls, bidi embeddings/overrides/isolates, line/paragraph separators and padding. ZWJ,
# ZWNJ and variation selectors stay, because emoji and scripts such as Persian need them.
_PROSE_BIDI_CONTROLS = frozenset("\u202a\u202b\u202c\u202d\u202e\u2066\u2067\u2068\u2069")
def _escape_approval_run(run: str, prose: bool = False) -> str:
"""Escape the invisible code points of a non-ASCII run; 4+ repeats of one show once with their count."""
def one(m: re.Match) -> str:
ch, n = m.group(1), len(m.group())
cat = unicodedata.category(ch)
if prose:
hidden = cat == "Cc" or cat in ("Zl", "Zp") or ch in _PROSE_BIDI_CONTROLS
else:
hidden = ch in _APPROVAL_BLANK_GLYPHS or cat[0] == "C" or cat in ("Zl", "Zp")
if not hidden:
return m.group()
esc = ch.encode("unicode_escape").decode("ascii")
return f"⟨{esc} ×{n}⟩" if n >= 4 else esc * n
return _APPROVAL_SAME_CHAR_RUN_RE.sub(one, run)
def sanitize_command_for_display(text: str | None, *, prose: bool = False) -> str:
"""Make hidden content in an approval-prompt string visible; clean text comes back unchanged.
Display-only: the executed command, smart assessment and allowlist keys never see this copy.
``prose=True`` is the lighter pass for question text (see ``_PROSE_BIDI_CONTROLS``)."""
if not text:
return ""
out = _APPROVAL_NON_ASCII_RE.sub(lambda m: _escape_approval_run(m.group(), prose), text)
out = _APPROVAL_BLANK_RE.sub(lambda m: f"\n⟨+{m.group().count(chr(10)) - 1} blank lines⟩\n", out)
return _APPROVAL_PAD_RE.sub(lambda m: f" ⟨+{len(m.group())} whitespace chars⟩ ", out)
def redact_approval_display(text: str | None, *, force: bool = False, prose: bool = False) -> str:
"""Secret redaction plus :func:`sanitize_command_for_display`, for every approval surface."""
return sanitize_command_for_display(redact_sensitive_text(str(text or ""), force=force), prose=prose)
def _mask_token(token: str) -> str:
"""Mask a log token — 18-char floor, preserves 6 prefix / 4 suffix; empty → ``***``."""
if not token:
return "***"
return mask_secret(token, head=6, tail=4, floor=18)
# Output of an earlier pass: masking it again would erase the kind it keeps visible. Only a
# value that IS a marker is skipped; ``***MyRealPass9`` or ``"«redacted:ghp_…» tail"`` still
# carries cleartext. An unquoted ``[REDACTED PRIVATE KEY]`` reaches here cut at its space.
_REDACTION_MARKER_RE = re.compile(r"\*{3}|«redacted(?::[^»\s]*|-secret)»|\[REDACTED(?: PRIVATE KEY\])?")
def _is_redaction_marker(match: "re.Match[str]", value: str) -> bool:
if value == "[REDACTED":
return match.string.startswith(" PRIVATE KEY]", match.end("value"))
return _REDACTION_MARKER_RE.fullmatch(value) is not None
def _mask_flag_value(match: "re.Match[str]", mask=None) -> str:
"""Mask the value of a flag/connection-string key, keeping the flag and any quotes.
``mask`` is the caller's mask function: ``file_read`` passes the non-reusable sentinel,
so a file read never hands back a head/tail value shaped like a usable key (#35519)."""
value = match.group("value")
stripped = value.strip("'\"")
if not stripped or _is_redaction_marker(match, stripped.strip()):
return match.group(0)
# ``--password=$DB_PASS`` / ``--token=${TOKEN}`` name a variable; the reference is
# not the credential and masking it would make the command unreplayable.
if stripped[0] in "$<{%":
return match.group(0)
if stripped.isdigit():
return match.group(0) # ``--token 60`` is a timeout, not a credential
if match.re is _CLI_SECRET_FLAG_RE:
if "=" not in match.group("sep") and _is_flag_placeholder(value):
return match.group(0)
if _is_flag_path_reference(match.group("flag"), stripped):
return match.group(0)
masked = (mask or _mask_token)(stripped)
start = match.start("value") - match.start(0)
end = match.end("value") - match.start(0)
return match.group(0)[:start] + masked + match.group(0)[end:]
# A space-separated flag is followed by whatever comes next, which in ``--help`` output and
# prose is a placeholder or a word (``--token TOKEN``, ``[--password PASSWORD]``, ``use
# --token to pass…``), the next flag (``psql --password -h db``) or a sub-option spec
# (``docker build --secret id=npmrc``). None of those is the credential.
# A metavar is upper case (``TOKEN``, ``DB_PASSWORD``), two short snake-case words
# (``db_name``) or a one-hump camelCase type name (cobra's ``stringArray``); a prose word
# is short and lower case. A passphrase is longer than that: a long lower-case word
# (``correcthorsebatterystaple``), three or more words (``correct_horse_battery_staple``,
# ``correctHorseBatteryStaple``) or anything carrying a digit or symbol is still masked.
_FLAG_PLACEHOLDER_RE = re.compile(
r"(?:[A-Z][A-Z_]*|[a-z]{1,8}|[a-z]{1,8}_[a-z]{1,8}|[a-z]{1,8}[A-Z][a-z]{1,8})[\]).,:;]*"
r"|[-\[(<].*|[a-z][\w-]*=.*"
r"|\S*(?:xxx|XXX|\.\.\.)" # a docs sample: ``--api-key sk-or-v1-xxx``
)
def _is_flag_placeholder(value: str) -> bool:
"""True for an unquoted space-separated flag value that is a word, not a credential."""
return value[0] not in "'\"" and _FLAG_PLACEHOLDER_RE.fullmatch(value) is not None
def _is_flag_path_reference(flag: str, value: str) -> bool:
"""``--credentials ~/.config/gcloud/creds.json`` names a file, same rule as the ENV passes:
only password-class flags mask a path, and only while no segment looks opaque."""
if _PASSWORD_KEY_RE.search(flag) or not value.startswith(("/", "~", "./", "../")):
return False
return not any(_OPAQUE_PATH_SEGMENT_RE.fullmatch(seg) for seg in value.split("/"))
def _redact_cli_secrets(text: str, mask=None) -> str:
"""Mask credentials written as command-line flags and connection-string keys."""
def sub(m: "re.Match[str]") -> str:
return _mask_flag_value(m, mask)
if "=" in text or " " in text:
if _CLI_SECRET_FLAG_GATE.search(text):
text = _CLI_SECRET_FLAG_RE.sub(sub, text)
if _CLI_LOGIN_GATE.search(text):
text = _CLI_LOGIN_P_RE.sub(sub, text)
if _CONNSTR_GATE.search(text):
text = _CONNSTR_SECRET_KEY_RE.sub(sub, text)
return text
def _redact_private_key_blocks(text: str) -> str:
"""Whole private-key blocks (PEM, base64 PEM, PGP, PuTTY, age). Runs before the
flag/connection-string passes: those stop a value at whitespace, so masking
``AccountKey=-----BEGIN`` first would break the PEM header and leave the body visible."""
if "-----BEGIN" in text and "-----END" in text: # no END, no match: skip the quadratic scan
text = _PRIVATE_KEY_RE.sub("[REDACTED PRIVATE KEY]", text)
if "LS0tLS1CRUdJTi" in text:
text = _B64_PEM_RE.sub(
lambda m: "[REDACTED PRIVATE KEY]" if _is_b64_pem_key(m.group(0)) else m.group(0),
text,
)
if "END PGP PRIVATE KEY BLOCK" in text: # no END, no match: skip the quadratic scan
text = _PGP_PRIVATE_KEY_RE.sub("[REDACTED PRIVATE KEY]", text)
if "Private-Lines:" in text:
text = _PUTTY_PRIVATE_LINES_RE.sub(
lambda m: m.group(1) + "[REDACTED PRIVATE KEY]" + m.group("eol"), text
)
if "AGE-SECRET-KEY-1" in text:
text = _AGE_SECRET_KEY_RE.sub("[REDACTED PRIVATE KEY]", text)
return text
def _is_python_repr_secret_key(key: str) -> bool:
"""Return True for exact secret keys or credential-suffixed key names."""
folded = key.casefold()
if folded in _PYTHON_REPR_SECRET_KEYS:
return True
if key.isupper() and key.endswith(_PYTHON_REPR_ENV_SUFFIXES):
return True
# Mixed/camel-case keys ending in a credential word (``UserPassword``,
# ``sessionToken``, ``clientApiKey``) — the exact-set and uppercase-suffix
# rules above miss these. Suffix-only matching keeps metadata names like
# ``TOKEN_COUNT`` / ``PASSWORD_POLICY`` / ``SECRET_NAME`` untouched.
# Class widened per OpenHands/software-agent-sdk#4508 (their dict-entry
# redaction was uppercase-only and leaked mixed-case keys).
if folded.endswith(_PYTHON_REPR_CREDENTIAL_SUFFIXES):
return True
# Names ending in ``key`` where a qualifier says which kind of key it is:
# ``aws_secret_access_key``, ``SECRET_KEY``, ``privateKey``, ``signing_key``,
# ``encryption_key``, ``master_key``, ``hmac_key``. A bare ``key``, ``public_key``,
# ``hotkey`` and ``key_id`` stay untouched — they name cache keys, limits and ids.
return _is_qualified_key_name(key)
# A qualified key masks short values too, except a sample or switch that cannot be the
# credential: ``{"encryptKey": "true"}``, ``master_key: default``, ``"<your key>"``. Only a
# word-shaped bracket is a docs sample; ``<hunter2>`` / ``<Abc9!xyz>`` are still masked.
_CONFIG_PLACEHOLDER_VALUE_RE = re.compile(
r"<[a-z]+(?:[ _-][a-z]+)+>|<(?:key|token|secret|password|value)>|true|false|yes|no|null|none|default",
re.IGNORECASE)
_KEY_SECRET_QUALIFIERS = frozenset({
"secret", "private", "signing", "encryption", "encrypt", "master", "hmac",
"symmetric", "access",
})
def _key_name_words(key: str) -> list[str]:
"""Split a field name into lower-case words: ``aws_secret_access_key`` and
``awsSecretAccessKey`` both give ``[aws, secret, access, key]``."""
spaced = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", key)
return [w for w in re.split(r"[^A-Za-z0-9]+", spaced.lower()) if w]
def _is_qualified_key_name(key: str) -> bool:
"""True for a ``…key`` field name whose qualifier makes it a credential.
``access`` is in the qualifier set because AWS pairs an access-key id with the
secret that signs with it, and the id alone still identifies an account; it is
never a value safe to print in a chat approval card.
"""
words = _key_name_words(key)
if len(words) < 2 or words[-1] != "key":
return False
return any(w in _KEY_SECRET_QUALIFIERS for w in words[:-1])
def is_secret_field_name(key: object) -> bool:
"""True when a mapping field named ``key`` holds a credential — the repr-field policy, for callers
that mask structured config by field instead of by text."""
return isinstance(key, str) and _is_python_repr_secret_key(key)
def _redact_python_repr_fields(text: str) -> str:
"""Fully mask credential fields in Python mapping ``repr`` output."""
def _sub(match: re.Match) -> str:
key = match.group("key")
if not _is_python_repr_secret_key(key):
return match.group(0)
single_value = match.group("single_value")
if single_value is not None:
prefix = match.group("single_prefix") or ""
quote = "'"
value = single_value
else:
prefix = match.group("double_prefix") or ""
quote = '"'
value = match.group("double_value")
# Mapping repr can contain code-shaped fixture values too. Preserve
# programmatic env lookups just like the ENV/JSON/YAML passes do.
if _ENV_LOOKUP_VALUE_RE.match(value):
return match.group(0)
# An upstream pass (MCP probe header scrub, _mask_token) already masked this
# value; re-masking would erase the scheme word it deliberately kept
# (``'Authorization': 'Digest ***'`` → ``'***'``).
if "***" in value or value.startswith("«redacted:"):
return match.group(0)
# Do not retain head/tail characters here: escaped repr atoms can cross
# a slicing boundary and leave an unescaped quote behind. A full mask is
# parseable for both str and bytes values and leaks no opaque bytes.
return f"'{key}'{match.group('sep')}{prefix}{quote}***{quote}"
return _PYTHON_REPR_FIELD_RE.sub(_sub, text)
def _redact_python_diagnostic_repr_fields(text: str) -> str:
"""Mask repr fields only on pytest/error lines in source-preserving output."""
lines = text.splitlines(keepends=True)
for index, line in enumerate(lines):
ending = ""
body_line = line
if line.endswith("\r\n"):
body_line, ending = line[:-2], "\r\n"
elif line.endswith("\n") or line.endswith("\r"):
body_line, ending = line[:-1], line[-1:]
match = _PYTEST_DIAGNOSTIC_LINE_RE.match(body_line)
if match is None:
match = _PYTHON_EXCEPTION_LINE_RE.match(body_line)
if match is not None:
lines[index] = (
match.group("prefix")
+ _redact_python_repr_fields(match.group("body"))
+ ending
)
return "".join(lines)
def _redact_query_string(query: str) -> str:
"""Replace values of sensitive ``k=v&k=v`` params with ``***``; others pass through."""
if not query:
return query
return "&".join(
f"{key}=***" if sep and key.lower() in _SENSITIVE_QUERY_PARAMS else pair
for pair in query.split("&") for key, sep, _ in (pair.partition("="),)
)
def _canonical_url_param_name(name: str) -> str:
"""Decode a URL parameter name (up to 3 unquote rounds) for case-insensitive matching."""
decoded = name
for _ in range(3):
next_value = unquote_plus(decoded)
if next_value == decoded:
break
decoded = next_value
folded = decoded.casefold()
# Preserve policy names that are canonically hyphenated (for example
# x-amz-signature) before accepting underscore-normalized aliases.
return folded if folded in _SENSITIVE_QUERY_PARAMS else folded.replace("-", "_")
def _redact_strict_url_credentials(text: str) -> str:
"""Strict egress-boundary redaction of URL credentials (absolute, relative and
network references); preserves keys, separators, public params, hosts, paths."""
text = _STRICT_URL_PARAM_RE.sub(
lambda m: f"{m.group(1)}{m.group(2)}=***"
if _canonical_url_param_name(m.group(2)) in _SENSITIVE_QUERY_PARAMS else m.group(0), text)
return _STRICT_URL_USERINFO_RE.sub("//***:***@", text)
def redact_cdp_url(value: object) -> str:
"""Mask secrets in a CDP/browser endpoint URL before it is logged.
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:
return text
text = _URL_WITH_QUERY_RE.sub(
lambda m: f"{m.group(1)}://{m.group(2)}{m.group(3)}?{_redact_query_string(m.group(4))}{m.group(5) or ''}",
text,
)
return _URL_USERINFO_RE.sub(lambda m: f"{m.group(1)}://{m.group(2)}:***@", text)
def _redact_form_body(text: str) -> str:
"""Redact sensitive values when the ENTIRE text is a clean ``k=v&k=v`` body."""
if not text or "\n" in text or "&" not in text or not _FORM_BODY_RE.match(text.strip()):
return text
return _redact_query_string(text.strip())
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),
only the vendor prefix label so the credential KIND stays visible.
* cannot be mistaken for a usable-but-truncated key, so an agent that reads it from a config file and
writes it back does NOT corrupt the stored credential into a dead 13-char string (issue #35519); and *
still does not leak the secret material (no head/tail chars).
"""
label = next((sub for sub in _PREFIX_SUBSTRINGS if token.startswith(sub)), "") if token else ""
return f"«redacted:{label}…»" if label else "«redacted-secret»"
def _redact_json_string_value(m: re.Match, *, mask=_mask_token) -> str:
"""Mask a captured JSON string value without breaking its escapes."""
encoded_value = m.group(2)
try:
value = json.loads(f'"{encoded_value}"')
except json.JSONDecodeError:
# The regex also accepts malformed escape pairs. Keep redaction
# fail-safe while emitting a syntactically valid replacement value.
value = encoded_value
if not json_header_value_is_credential(m.group(1), value):
return m.group(0)
masked = json.dumps(mask(value), ensure_ascii=False)[1:-1]
return m.group(1) + masked + m.group(3)
def _redact_unterminated_json_string_value(m: re.Match, *, mask=_mask_token) -> str:
"""Fail closed on a truncated JSON header value through end-of-line."""
encoded_value = m.group(2)
try:
value = json.loads(f'"{encoded_value}"')
except json.JSONDecodeError:
value = encoded_value
masked = json.dumps(mask(value), ensure_ascii=False)[1:-1]
return m.group(1) + masked
def _assignment_sub(render, *, check_keyword: bool):
"""re.sub callback: keep the match unless the key/value pair (groups[0], groups[-1]) needs redaction."""
def _sub(m):
groups = m.groups()
if not _should_redact_assignment(groups[0], groups[-1], check_keyword=check_keyword):
return m.group(0)
return render(groups)
return _sub
_JSON_FLAG_VALUES = frozenset({"true", "false", "yes", "no", "null", "none"})
_JSON_VALUE_SUFFIX_RE = re.compile(r"[_.\-]?(?:value|Value|VALUE)$")
def _json_affixed_key_secret(key: str, value: str) -> bool:
"""Whether ``"key": "value"`` holds a credential under an affixed secret-word key (#147240)."""
if _JSON_KEY_NAMES_FULL_RE.fullmatch(key): # the closed-list pass already decided this field
return False
if not (_has_word_bounded_keyword(key, _YAML_CFG_WORD_RE) or _has_password_keyword(key)):
return False
if _DSN_PW_PLACEHOLDER_RE.fullmatch(value) or not _should_redact_assignment(key, value, check_keyword=True):
return False
# A key that does not END in the credential word is metadata about one: ``passwordChanged``,
# ``password_expires_at``, ``tokenUrl``, ``secret_id``, ``secret_arn``, ``PASSWORD_STORE_DIR``
# carry timestamps, URLs and ids that look opaque, so the value shape cannot decide (#149435
# review). A boolean answers ``has_password``, it does not hold one.
# ``api_key_value`` / ``passwordValue`` hold the credential itself: drop that one suffix first.
return _is_python_repr_secret_key(_JSON_VALUE_SUFFIX_RE.sub("", key)) and value.casefold() not in _JSON_FLAG_VALUES
def _redact_json_affixed_key(m: "re.Match[str]", mask) -> str:
if not _json_affixed_key_secret(m.group(2), m.group(3)):
return m.group(0)
return f'{m.group(1)}: "{mask(m.group(3))}"'
def _redact_json_affixed_key_escaped(m: "re.Match[str]", mask) -> str:
if not _json_affixed_key_secret(m.group(3), m.group(4)):
return m.group(0)
return f'{m.group(1)}: {m.group(2)}"{mask(m.group(4))}{m.group(2)}"'
def _redact_assignments(text: str, *, mask_nonreusable: bool = False) -> 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).
``mask_nonreusable`` masks the assignments with the ``«redacted:…»`` sentinel that
``file_read=True`` already uses for prefix-matched credentials. Without it an agent
that read a secret-bearing file would hold a head/tail mask shaped like a real but
truncated key and could write it back as a dead credential (#35519)."""
mask = _mask_token_nonreusable if mask_nonreusable else _mask_token
if "=" in text:
text = redact_cookie_assignments(text, mask)
_redact_env = _assignment_sub(lambda g: f"{g[0]}={g[1]}{mask(g[2])}{g[1]}", check_keyword=True)
text = _ENV_ASSIGN_RE.sub(_redact_env, text)
if "://" not in text: # lowercase names would match URL params
# Skip URLs — the query string may contain ``token=``/``key=`` params that are intentionally
# passed through (see note near the bottom of this function; _redact_strict_url_credentials
# handles the opt-in case). The uppercase regex above is all-caps-only, so it never matches URL
# params; the lowercase one would (issue #77484).
text = _ENV_ASSIGN_LOWER_RE.sub(_redact_env, text)
# The keyword pre-gate is exact and matters: _CFG_DOTTED_RE backtracks
# quadratically on long unbroken [A-Za-z0-9_.\-] runs.
# Lowercase/dotted config keys (issue #16413). Skip URLs entirely — web-URL query params are
# intentionally passed through (see note near the bottom of this function); _DB_CONNSTR_RE still
# guards connection-string passwords. Extra gate: every _CFG_*_RE match requires a secret keyword in
# the key, so a text without any secret keyword cannot match — skipping is exact. This matters
# because _CFG_DOTTED_RE backtracks quadratically on long unbroken [A-Za-z0-9_.\-] runs (e.g.
# base64/hex blobs in compaction payloads); the linear keyword scan prevents that pathological path
# on secret-free text.
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)
if ":" in text and '"' in text:
for pattern in (_SECRET_HEADER_JSON_RE, _COOKIE_HEADER_JSON_RE):
text = pattern.sub(lambda m: _redact_json_string_value(m, mask=mask), text)
for pattern in (_SECRET_HEADER_JSON_UNTERMINATED_RE, _COOKIE_HEADER_JSON_UNTERMINATED_RE):
text = pattern.sub(lambda m: _redact_unterminated_json_string_value(m, mask=mask), text)
redact_json = _assignment_sub(lambda g: f'{g[0]}: "{mask(g[1])}"', check_keyword=False)
text = _JSON_FIELD_RE.sub(
lambda m: m.group(0) if m.group(1)[1:-1].lower() == "apikey" else redact_json(m), text)
if '\\"' in text:
text = _JSON_FIELD_ESCAPED_RE.sub(
_assignment_sub(lambda g: f'{g[0]}: {g[1]}"{mask(g[2])}{g[1]}"', check_keyword=False), text)
text = _JSON_ANY_KEY_FIELD_RE.sub(lambda m: _redact_json_affixed_key(m, mask), text)
if '\\"' in text:
text = _JSON_ANY_KEY_FIELD_ESCAPED_RE.sub(lambda m: _redact_json_affixed_key_escaped(m, mask), text)
# Python mapping repr fields ({'API_KEY': '…'}): single-quoted, so the JSON rule
# above never sees them — the traceback / pytest-introspection leak shape.
if ":" in text and "'" in text:
text = _redact_python_repr_fields(text)
# YAML after JSON; unwrap scalars before the shared assignment gate so
# quoted programmatic env lookups retain the same exemption as bare ones.
if ":" in text and "://" not in text:
def _redact_yaml(m):
key, sep, value = m.groups()
quote = value[0] if len(value) >= 2 and value[0] in "\"'" and value[-1] == value[0] else ""
inner = value[1:-1] if quote else value
# An empty or docs-placeholder scalar (``""``, ``"<your key>"``, ``"${API_KEY}"``) is not
# a credential; masking it hides what the config expects, as _redact_db already rules.
if quote and (not inner or _DSN_PW_PLACEHOLDER_RE.fullmatch(inner)
or (inner.startswith("{{") and inner.endswith("}}"))): # Helm/Jinja template
return m.group(0)
if not _should_redact_assignment(key, inner, check_keyword=True):
return m.group(0)
return f"{key}{sep}{quote}{mask(inner)}{quote}"
text = _YAML_ASSIGN_RE.sub(_redact_yaml, text)
return text
# A docs placeholder is a lower-case or ALL-CAPS name in angle brackets (``<password>``) or an
# upper-case env var reference (``${REDIS_PASSWORD}``); ``$ecretPassw0rd`` or ``<Zq8x…>`` is
# a real password that happens to start with the same character.
_DSN_PW_PLACEHOLDER_RE = re.compile(r"<(?:[a-z][a-z_ -]*|[A-Z][A-Z_ -]*)>|\$\{?[A-Z_][A-Z0-9_]*\}?|%[A-Z_][A-Z0-9_]*%")
# ``https://user:password@host`` in docs names the field instead of filling it in.
_USERINFO_PLACEHOLDER_WORDS = frozenset({"password", "passwd", "pass", "pw", "pwd", "secret", "token"})
def _is_userinfo_placeholder(password: str) -> bool:
"""A template (``{pw}``, ``{{ pw }}``), docs placeholder or field name, never a literal secret."""
return ((password.startswith("{") and password.endswith("}"))
or _DSN_PW_PLACEHOLDER_RE.fullmatch(password) is not None
or password.casefold() in _USERINFO_PLACEHOLDER_WORDS)
def _redact_userinfo_passwords(text: str) -> str:
"""Mask the password half of ``scheme://user:pass@host`` userinfo (#125664).
#34029's pass-through was about query-string round-trip tokens (magic links, OAuth callbacks,
pre-signed URLs live in the query, never in userinfo); the userinfo password is a credential —
the same reasoning #6396 used to mask the bare-token form. Keeps scheme, username and host
visible. Templates and docs samples (``{pw}``, ``${TOKEN}``, ``password``) stay as written."""
def _sub(m):
user, sep, password = m.group(2).partition(":")
# f-string / Jinja templates and docs samples (``user:{pw}@``, ``user:${TOKEN}@``,
# ``user:password@``) in prose and logs as much as in code files.
if not sep or not password or _is_userinfo_placeholder(password):
return m.group(0)
return f"{m.group(1)}{user}:***@"
return _URL_USERINFO_PAIR_RE.sub(_sub, text)
def _redact_url_credentials(text: str, code_file: bool) -> str:
"""DB connection-string passwords, bare-token and ``user:pass`` URL userinfo (``://`` text only)."""
def _redact_db(m):
# 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)
# Password-only URLs in docs: ``redis://:<password>@`` / ``redis://:${REDIS_PASSWORD}@``.
if m.group(1).endswith("://:") and _DSN_PW_PLACEHOLDER_RE.fullmatch(pw):
return m.group(0)
return f"{m.group(1)}***{m.group(3)}"
text = _DB_CONNSTR_RE.sub(_redact_db, text)
text = _URL_BARE_TOKEN_RE.sub(lambda m: f"{m.group(1)}{_mask_token(m.group(2))}{m.group(3)}", text)
return _redact_userinfo_passwords(text)
def _mask_network_path_userinfo(m) -> str:
"""``//user:pass@host`` without a scheme (#125664 residual); placeholders stay as written."""
if _is_userinfo_placeholder(m.group(3)):
return m.group(0)
return f"{m.group(1)}{m.group(2)}:***@"
def _redact_phone(m):
phone = m.group(1)
keep = 2 if len(phone) <= 8 else 4
return phone[:keep] + "****" + phone[-keep:]
def redact_sensitive_text(text: str, *, force: bool = False, code_file: bool = False,
file_read: bool = False, secret_file: bool = False,
redact_url_credentials: bool = False) -> str:
"""Apply all redaction patterns to a block of text.
Safe on any string. Enabled by default (``security.redact_secrets: false``
disables); ``force=True`` is for safety boundaries that must never return
raw secrets regardless.
``redact_url_credentials=True``: also redact credential-named query params
and the entire URL userinfo (``***:***@``) — off by default because OAuth-callback /
magic-link / pre-signed URLs must survive ordinary tool flows unchanged.
``code_file=True``: skip the ENV/JSON assignment passes for known source
code (``MAX_TOKENS=***``, ``"apiKey": "test"`` fixtures). ``file_read=True``
(implies code_file unless ``secret_file``): prefix-matched credentials become a
non-reusable sentinel (``«redacted:ghp_…»``) instead of a head/tail mask an agent
could write back into config.yaml as a dead credential.
Every regex sits behind a cheap substring gate that its pattern requires,
so the gates are never false-negative.
Set code_file=True to also skip the Python-repr mapping pass (``{'API_KEY': '…'}``
fixtures in source); pytest/exception diagnostic lines get a narrow pass in
redact_terminal_output instead.
Set file_read=True for file *content* returned to the agent (read_file / search_files / cat). The old
mask looked like a real-but-truncated key, so an agent reading it from config.yaml and writing it back
silently corrupted the stored credential into a dead 13-char value → 401 (issue #35519). The sentinel is
syntactically invalid as a token, so it can't be mistaken for a usable key or written back as one.
Set ``secret_file=True`` when the caller has already classified the SOURCE as secret-bearing
(``_is_secret_file_arg``: command reads on the terminal side, resolved tool paths on the
file side). It re-enables the ENV/JSON/YAML assignment passes that ``file_read`` would
otherwise skip, so an opaque prefix-less credential assigned to a credential-shaped key is
masked instead of passed through in cleartext (issue #110567). With ``file_read=True`` those
assignments are masked with the non-reusable sentinel, so the #35519 write-back hazard stays
closed. ``secret_file`` is authoritative: it wins over ``code_file``, so a caller cannot be
fail-open by setting both. Files that are not secret-bearing (any source file, a project's own
``config.yaml``) keep the code_file behaviour: ``MAX_TOKENS: 100`` and ``"apiKey": "test"``
fixtures are untouched.
"""
if text is None:
return None
text = text if isinstance(text, str) else str(text)
if not text:
return text
# Vault secrets are a hard model-egress boundary: scrubbed regardless of the redact_secrets preference.
text = redact_registered_vault_values(text)
if not (force or _redact_enabled()):
return text
# ``secret_file`` is authoritative: a caller that classified the source as secret-bearing must not
# be silently fail-open because another flag (code_file, or file_read implying it) was also set.
code_file = (code_file or file_read) and not secret_file
# 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
# Control/zero-width chars (\\n, \\r, ESC, U+200B, …) split a token body so _PREFIX_RE cannot match
# across them — a secret smuggled as ``sk-abc\\x1bdef…`` leaks verbatim (issue #77484). Mask such
# runs by first matching on a control-stripped copy, then re-masking the corresponding span in the
# original (the stripped copy and the original are aligned 1:1 for non-control chars).
text = _mask_control_split_tokens(text, _prefix_sub)
text = _PREFIX_SUB_RE.sub(lambda m: _mask_prefix_match(m, _prefix_sub), text)
if "." in text:
_zhipu_sub = _mask_token_nonreusable if file_read else _mask_token
text = _ZHIPU_API_KEY_RE.sub(lambda m: _zhipu_sub(m.group(1)), text)
if not code_file:
text = _redact_assignments(text, mask_nonreusable=file_read)
# Flags and connection-string keys carry a credential in command text, so they are not
# gated on code_file: the shape (``--password=`` / ``;AccountKey=``) is unambiguous, and
# the surfaces that leaked them (approval cards, chat status) run this pass anyway.
text = _redact_private_key_blocks(text)
text = _redact_cli_secrets(text, _mask_token_nonreusable if file_read else None)
if force and "=" in text:
# ``NAME=user:pass`` under an innocuous name (#125664): shape-only, so force-only — every
# force=True caller (persistence boundary, debug dumps, session export) is a safety boundary
# that must not return raw secrets; ordinary flows keep today's pass-through.
def _basic_pair(m):
password = m.group(4)
if password is None or not _BASIC_PAIR_NAME_RE.search(m.group(1)) \
or not _is_basic_pair_password(password):
return m.group(0)
# ``CHECKSUM=sha256:<hex>``: an algorithm:digest label, not a user:credential pair.
if m.group(3).lower() in _HASH_ALGO_NAMES:
return m.group(0)
# A template or docs placeholder, not a literal credential — the rule
# _redact_userinfo_passwords applies below.
if _is_userinfo_placeholder(password):
return m.group(0)
return f"{m.group(1)}={m.group(2)}{m.group(3)}:***{m.group(2)}"
text = _BASIC_PAIR_ASSIGN_RE.sub(_basic_pair, 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)
if ":" in text:
text = redact_header_credentials(text)
text = _TELEGRAM_RE.sub(_redact_telegram_token, text)
# Database connection string passwords. With code_file=True, a password group that is a pure ``{...}``
# brace expression is an f-string template reference (e.g. f"postgresql://{user}:{pass}@{host}"), not a
# literal credential — preserve it. Literal passwords are still redacted. The regex forbids whitespace
# in the password group, so a single-line template's group(2) is exactly the brace expression. See issue
# #33801.
if "://" in text:
text = _redact_url_credentials(text, code_file)
if force and "//" in text:
text = _NETWORK_PATH_USERINFO_RE.sub(_mask_network_path_userinfo, text)
# Keyword/value connection strings (``Host=…;Password=…``, libpq ``host=… password=…``):
# the forms the URL pass above never sees. Gate: a ``password=``/``pwd=`` key in any casing;
# the per-line companion-key check in _redact_kv_connection_strings keeps prose out.
if _KV_CONNSTR_GATE_RE.search(text):
text = _redact_kv_connection_strings(text, code_file)
if "eyJ" in text:
text = _JWT_RE.sub(lambda m: _mask_token(m.group(0)), text)
# A Webapp sign-in link signs a browser in as the user, so its credential is masked whatever the
# web-URL policy below.
if "hermes-session=" in text or "/webapp/launch#" in text:
text = _WEBAPP_LINK_CREDENTIAL_RE.sub(lambda m: m.group(1) + "***", text)
if redact_url_credentials: # opt-in; known credential shapes in URLs are caught above
# NOTE: Web-URL redaction (query params + userinfo + HTTP access-log request targets) is
# intentionally OFF. Many legitimate workflows pass opaque tokens through query strings — magic-link
# checkouts, OAuth callbacks the agent is meant to follow, pre-signed share URLs — and
# blanket-redacting param values by name breaks those skills mid-flow. DB connection-string
# passwords are still caught by _DB_CONNSTR_RE. The ONE userinfo case still redacted is the
# colon-less bare-token form ``scheme://TOKEN@host`` (#6396, handled by _URL_BARE_TOKEN_RE in the
# ``://`` block above): a bare credential in userinfo is never a round-trip workflow token (those
# live in the query string), so masking it can't break a skill. The ``user:pass@`` form is left to
# pass through per #34029.
text = _redact_strict_url_credentials(text)
if "&" in text and "=" in text:
text = _redact_form_body(text)
if "+" in text:
text = _SIGNAL_PHONE_RE.sub(_redact_phone, text)
return text
# Commands whose stdout is an env-var dump: terminal redaction runs the
# ENV-assignment pass (code_file=False) for these so opaque tokens with no vendor
# prefix are masked; everything else uses code_file=True (``MAX_TOKENS=100``).
# Commands whose stdout is an environment-variable dump (KEY=value lines), NOT source code.
# ``MY_SERVICE_TOKEN=abc123randomstring``) are still masked. For all other commands, code_file=True is used
# to avoid mangling legitimate source/config dumps (``MAX_TOKENS=100``, ``"apiKey": "x"`` fixtures,
# ``postgresql://{user}`` f-string templates). See issue #43025.
_ENV_DUMP_COMMANDS = frozenset({"env", "printenv", "set", "export", "declare"})
# Commands that read file contents to stdout, plus the filter readers (``grep``/``awk``/``sed``)
# the model reaches for on config files. A secret-bearing target (``.env`` per AGENTS.md,
# a shell rc/profile, Hermes' own ``config.yaml`` where ``hermes mcp add --env`` writes
# tokens) is a credential dump, so the ENV/YAML assignment pass must run. Arbitrary
# ``config.yaml`` / source files stay on the code_file path (``MAX_TOKENS: 100``).
_FILE_READ_COMMANDS = frozenset({
"cat", "head", "tail", "type", "bat", "less", "more", "nl",
"zcat", "tac", "view", "batcat", "grep", "awk", "sed",
})
_SHELL_RC_BASENAMES = frozenset({
".bashrc", ".bash_profile", ".bash_login", ".profile",
".zshrc", ".zprofile", ".zlogin", ".zshenv",
})
# Filter readers take a PATTERN/program as their first positional; only the operands after
# it are files, so ``grep .bashrc app.py`` must not gate on the pattern.
_PATTERN_FIRST_COMMANDS = frozenset({"grep", "awk", "sed"})
_HERMES_HOME_PREFIXES = ("$HERMES_HOME/", "${HERMES_HOME}/")
# ``$HOME/.hermes/config.yaml`` keeps the ``.hermes`` segment, so stripping the prefix is
# enough to gate it; ``~/`` already survives the ``$``-bearing-path bail-out.
_HOME_PREFIXES = ("$HOME/", "${HOME}/")
def _command_segments(command: str) -> list[str]:
"""Pipeline/sequence segments, split only on unquoted ``| ; &`` so an
``awk '{print $1; print $2}'`` program or ``grep 'foo|bar'`` pattern stays one
segment. Backslash is not an escape (Windows ``C:\\Users\\...``)."""
segments: list[str] = []
buf: list[str] = []
quote: str | None = None
for ch in command:
if quote:
buf.append(ch)
if ch == quote:
quote = None
continue
if ch in "'\"":
quote = ch
buf.append(ch)
continue
if ch in "|;&":
seg = "".join(buf).strip()
if seg:
segments.append(seg)
buf = []
continue
buf.append(ch)
seg = "".join(buf).strip()
if seg:
segments.append(seg)
return segments
def _is_under_hermes_home(path: str) -> bool:
"""True when an absolute ``config.yaml`` path sits under the active Hermes home or root.
The default home's basename is an installation detail — ``.hermes`` on POSIX, ``hermes``
under ``AppData/Local`` on Windows — and a resolved path never spells ``$HERMES_HOME``,
so the literal-segment test in ``_is_secret_file_arg`` cannot see a native Windows path.
Compare against the resolved homes instead. Only reached for a ``config.yaml`` basename,
so the resolve cost stays off the per-token command scan.
"""
from agent.file_safety import _hermes_dirs
try:
target = os.path.normcase(os.path.realpath(os.path.expanduser(path)))
except (OSError, ValueError):
return False
for home in _hermes_dirs():
try:
base = os.path.normcase(os.path.realpath(str(home)))
except (OSError, ValueError):
continue
if target == base or target.startswith(base + os.sep):
return True
return False
def _is_secret_file_arg(arg: str) -> bool:
"""``.env``-style or shell rc basename anywhere; ``config.yaml`` only under a
``.hermes`` directory, ``$HERMES_HOME``, or the resolved Hermes home (never arbitrary
YAML). The resolved-home arm is what covers native Windows, where the home directory
is ``%LOCALAPPDATA%\\hermes`` and carries no ``.hermes`` segment."""
path = arg.strip("\"'").replace("\\", "/")
hermes_home = False
for prefix in _HERMES_HOME_PREFIXES:
if path.startswith(prefix):
path = path[len(prefix):]
hermes_home = True
break
for prefix in _HOME_PREFIXES:
if path.startswith(prefix):
path = path[len(prefix):]
break
if "$" in path:
return False
parts = [part.lower() for part in path.split("/") if part]
if not parts:
return False
if parts[-1] in _ENV_FILE_BASENAMES or parts[-1] in _SHELL_RC_BASENAMES:
return True
# ``config.yaml`` plus the ``config.yaml.good.<stamp>`` / ``.corrupt.<stamp>`` copies Hermes
# writes under ``backups/config/`` — same contents, same secrets.
if parts[-1] != "config.yaml" and not parts[-1].startswith(("config.yaml.good.", "config.yaml.corrupt.")):
return False
return hermes_home or ".hermes" in parts[:-1] or _is_under_hermes_home(path)
def _command_reads_secret_file(command: str | None) -> bool:
"""True if ``command`` reads a secret-bearing file (see ``_is_secret_file_arg``) to
stdout. Defense-in-depth, not a boundary: indirect reads (``sudo cat .env``, ``$(cat
.env)``, unresolved variable paths) are not detected, matching ``is_env_dump_command``."""
if not command or not isinstance(command, str):
return False
for seg in _command_segments(command):
tokens = seg.split() # not shlex: it mangles Windows paths (``C:\Users\...\.env``)
if not tokens:
continue
reader = tokens[0].rsplit("/", 1)[-1].lower()
if reader not in _FILE_READ_COMMANDS:
continue
positional = [arg for arg in tokens[1:] if not arg.startswith("-")]
if reader in _PATTERN_FIRST_COMMANDS:
positional = positional[1:]
if any(_is_secret_file_arg(arg) for arg in positional):
return True
return False
def is_env_dump_command(command: str | None) -> bool:
"""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):
try:
tokens = shlex.split(seg)
except ValueError:
tokens = seg.split()
if tokens and tokens[0] in _ENV_DUMP_COMMANDS:
return True
return False
REDACTION_UNAVAILABLE = "[redaction-unavailable]"
# The opaque branch needs a 20-char floor (the floor the gateway/A2A sweeps always had): without it the
# English word "bearer" turns "the bearer of bad news" into "Bearer [redacted] bad news" on every chat
# reply. The bracket branch folds an already-masked residue ("Bearer [redacted-jwt]") to one marker.
_BEARER_RESIDUE_RE = re.compile(r"\bBearer\s+(?:\[[^\]]+\]|[A-Za-z0-9._~+/-]{20,}=*)", re.IGNORECASE)
def redact_for_egress(text: str) -> str:
"""The one scrub for text leaving the process for a remote reader (chat platforms, A2A peers,
telemetry). ``redact_sensitive_text(force=True)`` — the only secret-pattern list — plus a bearer
sweep, because a ``Bearer <opaque>`` value with no vendor prefix carries no shape the prefix
matcher can key on. Fails CLOSED: if the redactor raises, the raw text is never returned."""
text = str(text or "")
try:
text = redact_sensitive_text(text, force=True)
except Exception:
return REDACTION_UNAVAILABLE
if "earer" in text:
text = _BEARER_RESIDUE_RE.sub("Bearer [redacted]", text)
return text
def redact_terminal_output(output: str, command: str | None = None, *, force: bool = False) -> str:
"""Single redaction policy for ALL terminal-output surfaces: the ENV/YAML-assignment
pass runs only when ``command`` is an env dump or reads a secret-bearing file (``.env``,
shell rc, Hermes ``config.yaml``); 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_secret_file(command))
redacted = redact_sensitive_text(output, force=force, code_file=code_file)
# Source-preserving output still gets the Python-repr pass on high-confidence
# diagnostic lines (pytest ``E `` introspection, final exception lines): that is
# where {'BRAVE_API_KEY': '…'} leaks, not in source dumps.
if code_file and (force or _redact_enabled()) and ":" in redacted and "'" in redacted:
redacted = _redact_python_diagnostic_repr_fields(redacted)
return redacted
# --- 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)."""
meta = "[(\\.?*+|{^$"
for i, ch in enumerate(pattern):
if ch in meta:
return pattern[:i]
return pattern
def _skip_char_class(pattern: str, i: int) -> int:
"""Given ``pattern[i] == "["``, return the index just past the closing ``]``."""
i += 2 if pattern[i + 1:i + 2] == "]" else 1 # a leading "]" is literal
while i < len(pattern) and pattern[i] != "]":
i += 2 if pattern[i] == "\\" else 1
return i
def _unbounded_quantifier_follows(pattern: str, j: int) -> bool:
"""True if an open-ended quantifier (``*``, ``+``, ``{m,}``) starts at ``pattern[j]``."""
if j >= len(pattern):
return False
if pattern[j] in "*+":
return True
if pattern[j] == "{":
k = pattern.find("}", j)
body = pattern[j + 1:k] if k != -1 else ""
return body[:-1].isdigit() and body.endswith(",") # {m,} is open-ended; {m} / {m,n} bounded
return False
def _pattern_structure(pattern: str) -> tuple[bool, bool]:
"""One scan → ``(has_top_level_alternation, has_nested_unbounded_repeat)``.
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?
i = 0
while i < len(pattern):
ch = pattern[i]
if ch == "\\":
i += 2
continue
if ch == "[":
i = _skip_char_class(pattern, i)
elif ch == "(":
contains_unbounded.append(False)
elif ch == ")":
inner = contains_unbounded.pop() if len(contains_unbounded) > 1 else False
if inner and _unbounded_quantifier_follows(pattern, i + 1):
nested = True
contains_unbounded[-1] = contains_unbounded[-1] or inner
elif ch == "|" and len(contains_unbounded) == 1:
top_level_alt = True
elif _unbounded_quantifier_follows(pattern, i):
contains_unbounded[-1] = True
if ch == "{":
i = pattern.find("}", i) # skip the {m,} body
i += 1
return top_level_alt, nested
def _has_top_level_alternation(pattern: str) -> bool:
return _pattern_structure(pattern)[0]
def _has_nested_unbounded_repeat(pattern: str) -> bool:
return _pattern_structure(pattern)[1]
_PREFIX_SUBSTRINGS = tuple(_extract_literal_prefix(p) for p in _PREFIX_PATTERNS)
def _has_known_prefix_substring(text: str) -> bool:
"""Cheap pre-check before the expensive ``_PREFIX_RE``."""
return any(p in text for p in _PREFIX_SUBSTRINGS)
# --- Plugin-registered redaction patterns -----------------------------------
# ADDITIVE-ONLY: a plugin can extend what gets masked but cannot weaken a
# built-in, so it can only over-redact. Keyed by registration source so plugin
# unload has a clean seam to drop ONE plugin's patterns.
# There is deliberately no public removal API — additive-only stands; unload is a host-owned lifecycle
# concern. See #64229.
_PLUGIN_PREFIX_PATTERNS: dict = {}
_registry_lock = threading.Lock()
def _plugin_patterns() -> list:
"""All plugin-registered patterns in registration order."""
return [p for patterns in _PLUGIN_PREFIX_PATTERNS.values() for p in patterns]
def _rebuild_prefix_matcher() -> None:
"""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_SUB_RE, _PREFIX_SUBSTRINGS
combined = _PREFIX_PATTERNS + _plugin_patterns()
_PREFIX_RE = _compile_prefix_matcher(combined)
_PREFIX_SUB_RE = _compile_prefix_sub_matcher(combined)
_PREFIX_SUBSTRINGS = tuple(_extract_literal_prefix(p) for p in combined)
# Structural validators for register_redaction_patterns, in check order:
# (predicate -> reject when True, warning message with (source, pattern) args).
_PATTERN_REJECT_RULES = (
(_has_top_level_alternation,
"%s: skipping redaction pattern %r — top-level alternation escapes the literal-prefix "
"guarantee (in 'ab|.*' the prefix binds only the first branch); wrap alternation in "
"a group after the prefix, e.g. 'ab(?:x|y)'"),
(_has_nested_unbounded_repeat,
"%s: skipping redaction pattern %r — nested unbounded quantifiers (e.g. '(a+)+') can "
"backtrack catastrophically, and registered patterns run on every log line and tool output"),
(lambda pattern: len(_extract_literal_prefix(pattern)) < 2,
"%s: skipping redaction pattern %r — must start with at least 2 literal characters "
"(needed for the pre-screen substring gate)"),
)
def register_redaction_patterns(patterns, source: str = "plugin") -> int:
"""Additively register credential-token regexes; returns the count 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 []:
if not isinstance(pattern, str) or not pattern.strip():
logger.warning("%s: skipping empty/non-string redaction pattern", source)
continue
pattern = pattern.strip()
try:
re.compile(pattern)
except re.error as exc:
logger.warning("%s: skipping invalid redaction pattern %r (%s)", source, pattern, exc)
continue
rejected = next((message for reject, message in _PATTERN_REJECT_RULES if reject(pattern)), None)
if rejected:
logger.warning(rejected, source, pattern)
continue
if pattern in _PREFIX_PATTERNS or pattern in _plugin_patterns() or pattern in accepted:
logger.debug("%s: redaction pattern %r already registered", source, pattern)
continue
accepted.append(pattern)
if accepted:
with _registry_lock:
_PLUGIN_PREFIX_PATTERNS.setdefault(source, []).extend(accepted)
_rebuild_prefix_matcher()
logger.info("%s: registered %d redaction pattern(s)", source, len(accepted))
return len(accepted)
def _reset_plugin_redaction_patterns() -> None:
"""Drop all plugin-registered patterns (tests/teardown only)."""
with _registry_lock:
_PLUGIN_PREFIX_PATTERNS.clear()
_rebuild_prefix_matcher()
class RedactingFormatter(logging.Formatter):
"""Log formatter that redacts secrets from all log messages."""
def format(self, record: logging.LogRecord) -> str:
return redact_sensitive_text(super().format(record))