"""Regex-based secret redaction for logs and tool output. Applies pattern matching to mask API keys, tokens, and credentials before they reach log files, verbose output, or gateway logs. Short tokens (< 18 chars) are fully masked. Longer tokens preserve the first 6 and last 4 characters for debuggability. """ import logging import os import re import shlex import threading from urllib.parse import unquote_plus # Basenames treated as ``.env`` files by _command_reads_env_file. Imported # from agent/file_safety (the read-block list) so the two defenses can't # drift: if file_tools blocks a read and the agent falls back to ``cat``, # the terminal redactor still catches it. file_safety matches # case-insensitively (``resolved.name.lower()``); the lookup mirrors that. from agent.file_safety import _BLOCKED_PROJECT_ENV_BASENAMES as _ENV_FILE_BASENAMES logger = logging.getLogger(__name__) # Sensitive query-string parameter names (case-insensitive exact match). # 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", # OAuth authorization codes "signature", # pre-signed URL signatures "x-amz-signature", }) # Sensitive form-urlencoded / JSON body key names (case-insensitive exact match). # Exact match, NOT substring — "token_count" and "session_id" must NOT match. # Ported from nearai/ironclaw#2529. _SENSITIVE_BODY_KEYS = frozenset({ "access_token", "refresh_token", "id_token", "token", "api_key", "apikey", "client_secret", "password", "auth", "jwt", "secret", "private_key", "authorization", "key", }) # Snapshot at import time so runtime env mutations (e.g. LLM-generated # `export HERMES_REDACT_SECRETS=false`) cannot disable redaction # mid-session. 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"} # Known API key prefixes -- match the prefix + contiguous token chars _PREFIX_PATTERNS = [ r"sk-[A-Za-z0-9_-]{10,}", # OpenAI / OpenRouter / Anthropic (sk-ant-*) 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 r"am_[A-Za-z0-9_-]{10,}", # AgentMail API key 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 pattern keeps a full literal prefix so the # _PREFIX_SUBSTRINGS pre-screen stays false-negative-free). 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 patterns: KEY=value where KEY contains a secret-like name. # Uppercase keys tolerate spaces around "=" (e.g. ``FOO_SECRET = bar``) because # an all-caps key is almost never prose/code. # Bare ``KEY`` / ``PASS`` / ``PW`` suffixes are included (``FAL_KEY=…``, # ``MYSQL_PASS=…``, ``DB_PW=…``) — issue #77484. 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+``. _SECRET_ENV_NAMES = r"(?:API_?KEY|KEY|TOKEN|SECRET|PASSWORD|PASSWD|PASS|PW|CREDENTIAL|AUTH)" # Uppercase keys keep the legacy embedded match (``MYTOKEN=…``, ``FOO_SECRET``) # — an all-caps key is almost never prose. _ENV_ASSIGN_RE = re.compile( rf"([A-Z0-9_]{{0,50}}{_SECRET_ENV_NAMES}[A-Z0-9_]{{0,50}})\s*=\s*(['\"]?)(\S+)\2", ) # Lowercase env names: only underscore-boundary forms (``openai_key=…``, # ``FAL_KEY=…``, ``db_pw=…``) — NOT bare ``password=``/``token=``/``secret=``, # which appear in prose, URLs, and form bodies (issue #77484). # Anchor each attempt to the start of an identifier run. Without the # lookbehind, ``re.sub`` retries the greedy ``[a-z0-9_]+`` prefix at every byte # of a long non-matching opaque payload, making strict compaction redaction # quadratic while holding the GIL (#99255). _ENV_ASSIGN_LOWER_RE = re.compile( rf"(? bool: """True if position ``i`` in ``s`` begins a word (not mid-word).""" if i == 0: return True prev, cur = s[i - 1], s[i] if not prev.isalpha(): return True if cur.isupper() and prev.islower(): return True # camelCase: clientSecret # Acronym run ending: APIToken — the 'T' begins a new word when it is # followed by lowercase while the preceding run is uppercase. if cur.isupper() and prev.isupper() and i + 1 < len(s) and s[i + 1].islower(): return True return False def _is_word_end(s: str, j: int, *, allow_plural: bool = True) -> bool: """True if position ``j`` (exclusive end) in ``s`` ends a word.""" if j >= len(s): return True cur = s[j] if not cur.isalpha(): return True if cur.isupper() and s[j - 1].islower(): return True # camelCase continuation: secretKey if allow_plural and cur in "sS": return _is_word_end(s, j + 1, allow_plural=False) return False def _key_has_secret_keyword(key: str) -> bool: """True if ``key`` contains a secret keyword at a word boundary. Post-match validator for _CFG_DOTTED_RE / _CFG_ANCHORED_RE / _YAML_ASSIGN_RE hits — rejects prose words that merely embed a keyword (``secretary``, ``tokenizer``, ``authored``). Safe to call with the _ENV_ASSIGN_RE key too: all-caps keys short-circuit to the legacy embedded-match behavior. """ letters = [c for c in key if c.isalpha()] if letters and all(c.isupper() for c in letters): # Legacy all-caps behavior (MYTOKEN=…): an all-caps key is almost # never prose. Exception: a bare ``KEY``/``PASS``/``PW`` embedded in # a longer all-caps word (``KEYBOARD``, ``PASSAGE``) is prose, not a # credential — only a word-bounded compound (``API_KEY``, # ``MYSQL_PASSWORD``, ``FAL_KEY``, ``DB_PW``) counts (issue #77484). for m in _KEY_KEYWORD_RE.finditer(key): if _is_word_start(key, m.start()) and _is_word_end(key, m.end()): return True return False for m in _KEY_KEYWORD_RE.finditer(key): if _is_word_start(key, m.start()) and _is_word_end(key, m.end()): return True return False def _key_has_strong_secret_keyword(key: str) -> bool: """Return whether ``key`` names an unambiguously credential-bearing field.""" for match in _STRONG_KEY_KEYWORD_RE.finditer(key): if _is_word_start(key, match.start()) and _is_word_end(key, match.end()): return True return False def _looks_like_opaque_credential(value: str) -> bool: """Return whether an ambiguous token/key value has credential-like shape. Known vendor prefixes and JWTs have dedicated redactors. This catches the remaining opaque family without treating short technical scalars such as ``CPU``, ``local``, or training captions as secrets merely because their key contains ``token`` or ``key``. """ 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 classes = sum( bool(re.search(pattern, value)) for pattern in (r"[a-z]", r"[A-Z]", r"[0-9]") ) return classes >= 2 def _assignment_value_requires_redaction(key: str, value: str) -> bool: """Apply value-aware gating to key-name-only assignment matches.""" return _key_has_strong_secret_keyword(key) or _looks_like_opaque_credential(value) # JSON field patterns: "apiKey": "value", "token": "value", etc. _JSON_KEY_NAMES = r"(?:api_?[Kk]ey|token|secret|password|access_token|refresh_token|auth_token|bearer|secret_value|raw_secret|secret_input|key_material)" _JSON_FIELD_RE = re.compile( rf'("{_JSON_KEY_NAMES}")\s*:\s*"([^"]+)"', re.IGNORECASE, ) # Authorization headers — any scheme (Bearer, Basic, Token, Digest, …) plus the # bare-credential form, and Proxy-Authorization. The credential token is masked # while the header name and scheme word are preserved for debuggability. The # previous rule only matched ``Bearer``, so ``Basic `` and # ``token `` leaked verbatim into logs/transcripts. # # The credential class excludes quote characters (``"`` / ``'``): a token sitting # flush against a closing quote (``"Authorization: Bearer sk-..."``) must not pull # that quote into the match, or masking turns value corruption into *syntax* # corruption — the closing quote vanishes and the command/string no longer parses # (unterminated quote → shell EOF / Python SyntaxError). Real credentials never # contain ``"`` or ``'``, so excluding them is safe. See #43083. _AUTH_HEADER_RE = re.compile( r"((?:Proxy-)?Authorization:\s*)([A-Za-z][\w.+-]*\s+)?([^\s\"']+)", re.IGNORECASE, ) # API-key style auth headers carrying a single opaque value (no scheme word). # Anthropic and many providers authenticate with ``x-api-key``; values without # a known vendor prefix (custom/local backends) would otherwise leak when a # request or curl command is logged or echoed into tool output / transcripts. _SECRET_HEADER_NAMES = ( r"(?:x-api-key|x-goog-api-key|api-key|apikey|x-api-token|x-auth-token|x-access-token)" ) _SECRET_HEADER_RE = re.compile( rf"({_SECRET_HEADER_NAMES}\s*:\s*)(\S+)", re.IGNORECASE, ) # Telegram bot tokens: bot: or :, # where token part is restricted to [-A-Za-z0-9_] and length >= 30 _TELEGRAM_RE = re.compile( r"(bot)?(\d{8,}):([-A-Za-z0-9_]{30,})", ) # Private key blocks: -----BEGIN RSA PRIVATE KEY----- ... -----END RSA PRIVATE KEY----- _PRIVATE_KEY_RE = re.compile( r"-----BEGIN[A-Z ]*PRIVATE KEY-----[\s\S]*?-----END[A-Z ]*PRIVATE KEY-----" ) # Database connection strings: protocol://user:PASSWORD@host # Catches postgres, mysql, mongodb, redis, amqp URLs and redacts the password. # The userinfo and password groups forbid whitespace ([^:\s]+ / [^@\s]+) so the # match can never span a line break. 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)?|redis|amqp)://[^:\s]+:)([^@\s]+)(@)", re.IGNORECASE, ) # Bare-token credential in a web/transport URL: ``scheme://TOKEN@host``. # 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. It is unambiguously a secret: legitimate # round-trip URLs (OAuth callbacks, magic links, pre-signed shares — see the # "Web-URL redaction is intentionally OFF" note in redact_sensitive_text) carry # their tokens in the QUERY STRING, never in bare userinfo. The colon form # ``user:pass@`` is deliberately left to pass through (commit "pass web URLs # through unchanged", #34029) and is NOT matched here — the token class forbids # ``:``. DB schemes are handled by _DB_CONNSTR_RE above and excluded here. # # Guards against false positives: # - 8+ char floor skips short usernames (git, admin, root, deploy, ubuntu). # - The token class ``[^\s:@/]`` cannot cross ``/``, so an ``@`` sitting in a # path or query (e.g. ``?q=user@example.com``) is never treated as userinfo. _URL_BARE_TOKEN_RE = re.compile( r"((?:https?|wss?|git|ssh|ftp|ftps|sftp)://)" # scheme r"([^\s:@/]{8,})" # bare token (no colon/slash/@), 8+ chars r"(@[^\s]+)", # @host... re.IGNORECASE, ) # JWT tokens: header.payload[.signature] — always start with "eyJ" (base64 for "{") # Matches 1-part (header only), 2-part (header.payload), and full 3-part JWTs. _JWT_RE = re.compile( r"eyJ[A-Za-z0-9_-]{10,}" # Header (always starts with eyJ) r"(?:\.[A-Za-z0-9_=-]{4,}){0,2}" # Optional payload and/or signature ) # E.164 phone numbers: +, 7-15 digits # Negative lookahead prevents matching hex strings or identifiers _SIGNAL_PHONE_RE = re.compile(r"(\+[1-9]\d{6,14})(?![A-Za-z0-9])") # URLs containing query strings — matches `scheme://...?...[# or end]`. # Used to scan text for URLs whose query params may contain secrets. # Ported from nearai/ironclaw#2529. _URL_WITH_QUERY_RE = re.compile( r"(https?|wss?|ftp)://" # scheme r"([^\s/?#]+)" # authority (may include userinfo) r"([^\s?#]*)" # path r"\?([^\s#]+)" # query (required) r"(#\S*)?", # optional fragment ) # URLs containing userinfo — `scheme://user:password@host` for ANY scheme # (not just DB protocols already covered by _DB_CONNSTR_RE above). # Catches things like `https://user:token@api.example.com/v1/foo`. _URL_USERINFO_RE = re.compile( r"(https?|wss?|ftp)://([^/\s:@]+):([^/\s@]+)@", ) # Strict provider-egress URL redaction accepts more URL-reference forms than # the display/log helpers above. Parameter delimiters stay in capture groups so # redaction preserves the original query/fragment layout byte-for-byte, while # the key is decoded separately for classification. Values stop at query or # fragment pair separators; both ``&`` and ``;`` are valid in deployed URLs. _STRICT_URL_PARAM_RE = re.compile( r"([?#&;])([A-Za-z0-9_.~+%\-]+)=([^#&;\s\"'<>]*)" ) # Match userinfo in both absolute (``scheme://user:pass@host``) and # network-path (``//user:pass@host``) references. The authority boundary stops # at path/query/fragment delimiters so an ``@`` elsewhere in a URL is ignored. # # Anchored on the mandatory ``//`` rather than an optional scheme prefix: the # scheme sits outside the match either way (replacement callbacks re-emit # group(1), so ``https:`` stays untouched in the surrounding text), and the # old optional-scheme prefix ``(?:[A-Za-z][A-Za-z0-9+.-]*:)?`` backtracked # catastrophically (O(n²)) on long unbroken alphanumeric runs — a 320KB # synthetic compaction payload spent ~55s inside this pattern per sub() call. # Output-equivalence to the old pattern was fuzz-verified (20k random strings # plus targeted URL forms). _STRICT_URL_USERINFO_RE = re.compile( r"(//)([^/\s?#@]+)@" ) # HTTP access logs often use a relative request target rather than a full URL: # `"POST /webhook?password=... HTTP/1.1"`. The full-URL redactor above only # sees strings containing `://`, so handle request-target query strings too. _HTTP_REQUEST_TARGET_QUERY_RE = re.compile( r"\b((?:GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS|TRACE|CONNECT)\s+[^ \t\r\n\"']*?)" r"\?([^ \t\r\n\"']+)", re.IGNORECASE, ) # Form-urlencoded body detection: conservative — only applies when the entire # text looks like a query string (k=v&k=v pattern with no newlines). _FORM_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: a secret # smuggled as ``sk-abc\x1bdef…`` or ``ghp_abc\n123…`` escapes the contiguous # prefix regexes (issue #77484). Used by _mask_control_split_tokens. _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 original chars that are token-body or control chars (see # _mask_control_split_tokens). ``=`` is deliberately excluded: a KEY=value # assignment separator must never let a match span across unrelated text. _TOKEN_BODY_CHARS = frozenset( "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-." ) # Compile known prefix patterns into one alternation _PREFIX_RE = re.compile( r"(? str: """Mask tokens whose body is split by control/zero-width characters. 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). Strategy: build a copy with all control chars removed (the token is contiguous again, matching even when each fragment alone is too short), match on that, then mask the corresponding span in the *original* — but only when the original span contains solely token-body and control chars (a match that crosses into a different line's unrelated text, e.g. ``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 = list(text) matches = [] for m in _PREFIX_RE.finditer(stripped): body = m.group(1) start_orig = orig_idx[m.start(1)] end_orig = orig_idx[m.end(1) - 1] + 1 # If a fragment inside the span already matches _PREFIX_RE on its # own AND the span crosses a LINE boundary (\n / \r), do NOT join. # A complete token at end-of-line followed by a word line # (``ghp_\nbutton [ref=e3]``) joins into one stripped-copy # match and the mask eats ``button``. Line structure is legitimate; # the self-matching fragment is handled by the ordinary prefix pass # (any remainder past the newline is left unmasked — accepted # residual to preserve line structure). # For NON-newline controls (ESC, ZWSP, ...) the join proceeds even # when a fragment self-matches: those bytes never legitimately sit # between a token and adjacent prose, and skipping there let the # non-matching remainder of a split token leak # (``sk-\x1b`` masked only the head). span = text[start_orig:end_orig] if ("\n" in span or "\r" in span) and _PREFIX_RE.search(span): continue # Reject matches whose original span crosses a non-token char # (e.g. ``sk_abc…\nTAVILY_API_KEY=…`` — the ``=`` is not part of a # token body, so the regex matched across unrelated lines). Also # reject when the match runs into a ``KEY=`` name: a real token value # is followed by a newline/space/end, not ``=``. 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(body))) for start_orig, end_orig, replacement in reversed(matches): out[start_orig:end_orig] = list(replacement) return "".join(out) # Display-mask strip for mask_secret: EVERY control char incl. \n/\t, C1, # DEL, and zero-width/format chars — a masked secret must never emit # multiline, tabbed, or invisible bytes into config/status/dump display # output (#55319, #55321). _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, preserving ``head`` and ``tail`` characters. Canonical helper for display-time redaction across Hermes — used by ``hermes config``, ``hermes status``, ``hermes dump``, and anywhere a secret needs to be shown truncated for debuggability while still keeping the bulk hidden. Args: value: The secret to mask. ``None``/empty returns ``empty``. head: Leading characters to preserve. Default 4. tail: Trailing characters to preserve. Default 4. floor: Values shorter than ``head + tail + floor_margin`` are fully masked (returns ``placeholder``). Default 12 — matches the existing config/status/dump convention. placeholder: Value returned for too-short inputs. Default ``"***"``. empty: Value returned when ``value`` is falsy (None, ""). The caller can override this to e.g. ``color("(not set)", Colors.DIM)`` for user-facing display. Examples: >>> mask_secret("sk-proj-abcdef1234567890") 'sk-p...7890' >>> mask_secret("short") # fully masked '***' >>> mask_secret("") # empty default '' >>> mask_secret("", empty="(not set)") # empty override '(not set)' >>> mask_secret("long-token", head=6, tail=4, floor=18) '***' """ if not value: return empty # Visible head/tail must not carry control bytes (newline, NUL, DEL, C1) # into config/status/dump output (#55319, #55321). Strip them before # slicing — the length check below then sees the displayable length. value = _DISPLAY_CONTROL_RE.sub("", value) if not value: return empty if len(value) < floor: return placeholder return f"{value[:head]}...{value[-tail:]}" def _mask_token(token: str) -> str: """Mask a log token — conservative 18-char floor, preserves 6 prefix / 4 suffix.""" # Empty input: historically this returned "***" rather than "". Preserve. if not token: return "***" return mask_secret(token, head=6, tail=4, floor=18) def _redact_query_string(query: str) -> str: """Redact sensitive parameter values in a URL query string. Handles `k=v&k=v` format. Sensitive keys (case-insensitive) have values replaced with `***`. Non-sensitive keys pass through unchanged. Empty or malformed pairs are preserved as-is. """ if not query: return query parts = [] for pair in query.split("&"): if "=" not in pair: parts.append(pair) continue key, _, value = pair.partition("=") if key.lower() in _SENSITIVE_QUERY_PARAMS: parts.append(f"{key}=***") else: parts.append(pair) return "&".join(parts) def _redact_url_query_params(text: str) -> str: """Scan text for URLs with query strings and redact sensitive params. Catches opaque tokens that don't match vendor prefix regexes, e.g. `https://example.com/cb?code=ABC123&state=xyz` → `...?code=***&state=xyz`. """ def _sub(m: re.Match) -> str: scheme = m.group(1) authority = m.group(2) path = m.group(3) query = _redact_query_string(m.group(4)) fragment = m.group(5) or "" return f"{scheme}://{authority}{path}?{query}{fragment}" return _URL_WITH_QUERY_RE.sub(_sub, text) def _redact_url_userinfo(text: str) -> str: """Strip `user:password@` from HTTP/WS/FTP URLs. DB protocols (postgres, mysql, mongodb, redis, amqp) are handled separately by `_DB_CONNSTR_RE`. """ return _URL_USERINFO_RE.sub( lambda m: f"{m.group(1)}://{m.group(2)}:***@", text, ) def _canonical_url_param_name(name: str) -> str: """Decode a URL parameter name for bounded, case-insensitive matching.""" decoded = name for _ in range(3): next_value = unquote_plus(decoded) if next_value == decoded: break decoded = next_value return decoded.casefold().replace("-", "_") def _redact_strict_url_credentials(text: str) -> str: """Redact credentials from absolute, relative, and network URL references. This is intentionally stricter than display/log redaction and is used only at explicit secret-egress boundaries. It preserves original keys, separators, public parameters, hosts, and paths while masking sensitive values and URL userinfo. """ def _redact_param(match: re.Match) -> str: if _canonical_url_param_name(match.group(2)) not in _SENSITIVE_QUERY_PARAMS: return match.group(0) return f"{match.group(1)}{match.group(2)}=***" def _redact_userinfo(match: re.Match) -> str: userinfo = match.group(2) if ":" in userinfo: username, _, _password = userinfo.partition(":") return f"{match.group(1)}{username}:***@" return f"{match.group(1)}***@" text = _STRICT_URL_PARAM_RE.sub(_redact_param, text) return _STRICT_URL_USERINFO_RE.sub(_redact_userinfo, text) def redact_cdp_url(value: object) -> str: """Mask secrets in a CDP/browser endpoint URL before it is logged. The global ``redact_sensitive_text`` deliberately passes web-URL query params and ``user:pass@`` userinfo through unmasked (OAuth callbacks, magic-link / pre-signed URLs the agent is meant to follow -- see the web-URL note above). CDP discovery endpoints are NOT such a workflow: their query-string tokens and userinfo passwords are pure credentials that must never reach the logs. So for CDP URLs we opt INTO the two URL redactors that the global pass leaves off. This is the single source of truth for redacting a CDP URL that is passed *directly* to a log or error message. Callers that instead need to redact an exception whose text embeds the URL (e.g. a ``websockets`` connect error) should route that through their own error-text helper, which delegates here -- see ``tools.browser_supervisor._redact_cdp_error_text``. """ text = redact_sensitive_text("" if value is None else str(value)) if not text: return text text = _redact_url_query_params(text) text = _redact_url_userinfo(text) return text def _redact_http_request_target_query_params(text: str) -> str: """Redact sensitive query params in HTTP access-log request targets.""" def _sub(m: re.Match) -> str: prefix = m.group(1) query = _redact_query_string(m.group(2)) return f"{prefix}?{query}" return _HTTP_REQUEST_TARGET_QUERY_RE.sub(_sub, text) def _redact_form_body(text: str) -> str: """Redact sensitive values in a form-urlencoded body. Only applies when the entire input looks like a pure form body (k=v&k=v with no newlines, no other text). Single-line non-form text passes through unchanged. This is a conservative pass — the `_redact_url_query_params` function handles embedded query strings. """ if not text or "\n" in text or "&" not in text: return text # The body-body form check is strict: only trigger on clean k=v&k=v. if 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. Unlike :func:`_mask_token` (which keeps head/tail chars — fine for logs that are never fed back into a config), this emits a marker that: * 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). The vendor prefix label is preserved for debuggability so the agent can still tell *which* credential is present (e.g. a GitHub PAT vs an OpenAI key) without seeing any of its bytes. """ if not token: return "«redacted-secret»" # Preserve only the recognizable vendor prefix label (e.g. "ghp_", "sk-"), # never any of the random secret body. label = "" for sub in _PREFIX_SUBSTRINGS: if token.startswith(sub): label = sub break return f"«redacted:{label}…»" if label else "«redacted-secret»" def redact_sensitive_text( text: str, *, force: bool = False, code_file: bool = False, file_read: bool = False, redact_url_credentials: bool = False, ) -> str: """Apply all redaction patterns to a block of text. Safe to call on any string -- non-matching text passes through unchanged. Enabled by default. Disable via security.redact_secrets: false in config.yaml. Set force=True for safety boundaries that must never return raw secrets regardless of the user's global logging redaction preference. Set redact_url_credentials=True at non-navigation egress boundaries to additionally redact credential-named query parameters and ``user:pass@`` URL userinfo. The default remains False because actionable OAuth callback, magic-link, and pre-signed URLs must survive ordinary tool flows unchanged. Set code_file=True to skip the ENV-assignment and JSON-field regex patterns when the text is known to be source code (e.g. MAX_TOKENS=*** constants, "apiKey": "test" fixtures). Prefix patterns, auth headers, private keys, DB connstrings, JWTs, and URL secrets are still redacted. Set file_read=True for file *content* returned to the agent (read_file / search_files / cat). Secrets are STILL redacted — they are never exposed — but prefix-matched credentials are replaced with a non-reusable sentinel (``«redacted:ghp_…»``) instead of a head/tail-preserving mask (``ghp_S1...Pn2T``). 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. Implies code_file=True (config/data files shouldn't trigger the source-code ENV/JSON false-positive paths). Performance: each regex pattern is gated behind a cheap substring pre-check (e.g. ``"=" in text`` for ENV assignments, ``"://" in text`` for URLs, ``"eyJ" in text`` for JWTs). On a typical hermes log line (no secrets) this drops the 13-pattern scan from ~5.6us to ~1.8us per record (-68%). The pre-checks are conservative — false positives still run the full regex, which then doesn't match. False negatives are impossible because every regex requires the gated substring to match. """ if text is None: return None if not isinstance(text, str): text = str(text) if not text: return text if not (force or _REDACT_ENABLED): return text # file_read content shouldn't hit the source-code ENV/JSON false-positive # paths either (it's config/data, not log lines). if file_read: code_file = True # Known prefixes (sk-, ghp_, etc.) — gate on substring presence 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_RE.sub(lambda m: _prefix_sub(m.group(1)), text) # ENV assignments: OPENAI_API_KEY=*** (skip for code files — false positives) if not code_file: if "=" in text: def _redact_env(m): name, quote, value = m.group(1), m.group(2), m.group(3) # Programmatic env lookups reference variable *names*, not # secret values — masking them corrupts code snippets in # prose/log contexts (issue #2852): ``KEY=os.getenv('X')``. if _ENV_LOOKUP_VALUE_RE.match(value): return m.group(0) # Keyword must sit at a word boundary within the key — # ``author=Smith`` / ``press.secretary=…`` are prose, not # credentials (ported from nearai/ironclaw#6129). All-caps # keys (the _ENV_ASSIGN_RE shape) short-circuit to legacy # embedded matching inside the helper. if not _key_has_secret_keyword(name): return m.group(0) if not _assignment_value_requires_redaction(name, value): return m.group(0) return f"{name}={quote}{_mask_token(value)}{quote}" text = _ENV_ASSIGN_RE.sub(_redact_env, text) # Lowercase env names (``openai_key=…``). 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). if "://" not in text: text = _ENV_ASSIGN_LOWER_RE.sub(_redact_env, text) # 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) # JSON fields: "apiKey": "***" (skip for code files — false positives) if ":" in text and '"' in text: def _redact_json(m): key, value = m.group(1), m.group(2) # Same programmatic-env-lookup exception as _redact_env above # (issue #2852): "apiKey": "os.getenv('X')" is a code snippet, # not a leaked secret value. if _ENV_LOOKUP_VALUE_RE.match(value): return m.group(0) if not _assignment_value_requires_redaction(key, value): return m.group(0) return f'{key}: "{_mask_token(value)}"' text = _JSON_FIELD_RE.sub(_redact_json, text) # Unquoted YAML / colon config: password: *** (after JSON so quoted # values are handled there; the lookahead in _YAML_ASSIGN_RE skips # quotes). Skip URLs — web-URL query params pass through by design. if ":" in text and "://" not in text: def _redact_yaml(m): key, sep, value = m.group(1), m.group(2), m.group(3) # 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 m.group(0) # Keyword must sit at a word boundary within the key — # ``Secretary: J.Smith`` / ``tokenizer: cl100k_base`` are # document text, not credentials (nearai/ironclaw#6129). if not _key_has_secret_keyword(key): return m.group(0) if not _assignment_value_requires_redaction(key, value): return m.group(0) return f"{key}{sep}{_mask_token(value)}" text = _YAML_ASSIGN_RE.sub(_redact_yaml, text) # Authorization headers — _AUTH_HEADER_RE matches any scheme after # "[Proxy-]Authorization:" case-insensitively, so "uthorization" is the # cheapest substring gate that covers every casing without a casefold(). if "uthorization" in text or "UTHORIZATION" in text: text = _AUTH_HEADER_RE.sub( lambda m: m.group(1) + (m.group(2) or "") + _mask_token(m.group(3)), text, ) # API-key style headers (x-api-key, api-key, …). Header values are # colon-separated, so gate on ":" — the regex itself is the precise filter. if ":" in text: text = _SECRET_HEADER_RE.sub( lambda m: m.group(1) + _mask_token(m.group(2)), text, ) # Telegram bot tokens — pattern requires ":" with digits prefix if ":" in text: def _redact_telegram(m): prefix = m.group(1) or "" digits = m.group(2) return f"{prefix}{digits}:***" text = _TELEGRAM_RE.sub(_redact_telegram, text) # Private key blocks if "BEGIN" in text and "-----" in text: text = _PRIVATE_KEY_RE.sub("[REDACTED PRIVATE KEY]", 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: if code_file: def _redact_db(m): pw = m.group(2) if pw.startswith("{") and pw.endswith("}"): return m.group(0) return f"{m.group(1)}***{m.group(3)}" text = _DB_CONNSTR_RE.sub(_redact_db, text) else: text = _DB_CONNSTR_RE.sub(lambda m: f"{m.group(1)}***{m.group(3)}", text) # Bare-token userinfo in web/transport URLs: ``scheme://TOKEN@host``. # The git-remote-with-embedded-password shape from #6396. Only the # colon-less bare-token form is redacted — ``user:pass@`` and # query-string tokens are left to pass through (see the web-URL note # below). See _URL_BARE_TOKEN_RE for the false-positive guards. text = _URL_BARE_TOKEN_RE.sub( lambda m: f"{m.group(1)}{_mask_token(m.group(2))}{m.group(3)}", text, ) # JWT tokens (eyJ... — base64-encoded JSON headers) if "eyJ" in text: text = _JWT_RE.sub(lambda m: _mask_token(m.group(0)), text) # 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. # Known credential shapes (sk-, ghp_, JWTs, etc.) inside URLs are still # caught by _PREFIX_RE and _JWT_RE above. 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. if redact_url_credentials: text = _redact_strict_url_credentials(text) # Form-urlencoded bodies (only triggers on clean k=v&k=v inputs). if "&" in text and "=" in text: text = _redact_form_body(text) # E.164 phone numbers (Signal, WhatsApp) if "+" in text: def _redact_phone(m): phone = m.group(1) if len(phone) <= 8: return phone[:2] + "****" + phone[-2:] return phone[:4] + "****" + phone[-4:] text = _SIGNAL_PHONE_RE.sub(_redact_phone, text) return text # Commands whose stdout is an environment-variable dump (KEY=value lines), # NOT source code. For these, terminal-output redaction must run the # ENV-assignment pass (code_file=False) so opaque tokens with no recognized # vendor prefix (e.g. ``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. When the target is a ``.env`` # file, the output is a credential dump — the same as ``printenv`` — so the # ENV-assignment pass must run (code_file=False). Per AGENTS.md, ``.env`` is # for secrets only; behavioral settings belong in config.yaml, so running # the generic ENV redactor on ``.env`` content is the correct behavior. _FILE_READ_COMMANDS = frozenset({ "cat", "head", "tail", "type", "bat", "less", "more", "nl", "zcat", "tac", "view", "batcat", }) # Basenames that are treated as ``.env`` files for redaction purposes are # imported at module top as ``_ENV_FILE_BASENAMES`` (see the # ``agent.file_safety`` import). def _command_reads_env_file(command: str | None) -> bool: """Return True if ``command`` reads a ``.env`` file to stdout. Detects file-read commands (``cat``, ``head``, ``tail``, etc.) where any argument's basename is a ``.env``-style file. Template files (``.env.example``, ``.env.sample``, ...) are not in the basename list and therefore never match. Handles pipelines and command sequences. Conservative defense-in-depth, not a boundary — indirect reads (``sudo cat .env``, ``/bin/cat .env``, ``$(cat .env)``, redirection, ``sed``/``awk``/``xxd`` readers) are not detected, matching the precedent of ``is_env_dump_command`` below. """ if not command: return False segments = re.split(r"[|;&]+", command) for seg in segments: seg = seg.strip() if not seg: continue # Use plain split() instead of shlex.split — shlex treats backslashes # as escape chars, which mangles Windows paths (``C:\Users\...\.env``). # We only need the command name and filename, so shell quoting is not # a concern here. tokens = seg.split() if not tokens or tokens[0] not in _FILE_READ_COMMANDS: continue # Check all arguments (skip flags like -n, -A, etc.) for arg in tokens[1:]: if arg.startswith("-"): continue # Strip shell quotes that plain split() leaves attached # (``cat ".env"`` / ``cat '.env'``), then any leading path to # get the basename. Handle both / and \. arg = arg.strip("\"'") basename = arg.rsplit("/", 1)[-1].rsplit("\\", 1)[-1] if basename.lower() in _ENV_FILE_BASENAMES: return True return False def is_env_dump_command(command: str | None) -> bool: """Return True if ``command`` dumps environment variables to stdout. Detects ``env`` / ``printenv`` / ``set`` / ``export`` / ``declare`` as the first token of any segment in a pipeline or sequence (``;`` / ``&&`` / ``||`` / ``|``). Conservative: a parse failure or anything unrecognized returns False (callers then fall back to the safer code_file=True path, which still masks prefix-shaped keys). """ if not command or not isinstance(command, str): return False # Split on shell separators, then inspect the first token of each segment. segments = re.split(r"[|;&]+", command) for seg in segments: seg = seg.strip() if not seg: continue try: tokens = shlex.split(seg) except ValueError: tokens = seg.split() if tokens and tokens[0] in _ENV_DUMP_COMMANDS: return True return False def redact_terminal_output( output: str, command: str | None = None, *, force: bool = False ) -> str: """Redact secrets from terminal/process stdout. Single redaction policy for ALL terminal-output surfaces — foreground ``terminal`` results AND background ``process(action=poll/log/wait)`` output — so they can't diverge. Picks ``code_file`` based on whether ``command`` is an environment dump or reads a ``.env`` file: - env-dump command (``env``/``printenv``/``set``/``export``/``declare``) → ``code_file=False`` so the ENV-assignment pass masks opaque tokens. - file-read command targeting a ``.env`` file (``cat .env``, ``head .env.local``, etc.) → ``code_file=False`` for the same reason. Per AGENTS.md, ``.env`` files contain only secrets, so the generic ENV pass is the right one (keys whose names carry no secret keyword can still slip through it — same limit as the env-dump path). - anything else (or unknown command) → ``code_file=True`` to avoid false positives on source/config dumps. ``force=True`` bypasses the global ``security.redact_secrets`` preference for safety boundaries that must never emit raw credentials. """ if not output: return output cmd = command or "" code_file = not (is_env_dump_command(cmd) or _command_reads_env_file(cmd)) return redact_sensitive_text(output, force=force, code_file=code_file) # Substrings used to gate ``_PREFIX_RE`` execution. If none of these appear in # the input string, the prefix regex cannot match anything, so we skip it. # False positives are fine (they just run the regex, which then matches # nothing) — the bound is "no false negatives" and that holds because every # pattern in ``_PREFIX_PATTERNS`` has at least one of these as a literal # substring of its leading characters. # # Derived automatically from ``_PREFIX_PATTERNS`` at module load time so a # future PR that adds a new prefix to the regex list can't silently break # the screen. def _extract_literal_prefix(pattern: str) -> str: """Return the leading literal characters of a regex pattern. Stops at the first regex metacharacter (``[``, ``(``, ``\\``, ``.``, ``?``, ``*``, ``+``, ``|``, ``{``, ``^``, ``$``). Returns the literal that any match of the pattern MUST contain as a substring, so the pre-screen never produces false negatives. """ meta = "[(\\.?*+|{^$" for i, ch in enumerate(pattern): if ch in meta: return pattern[:i] return pattern def _has_top_level_alternation(pattern: str) -> bool: """True if ``pattern`` contains a ``|`` outside any group or class. A top-level alternation defeats the literal-prefix guarantee: ``_extract_literal_prefix`` stops at ``|``, so for ``ab|.*`` it returns ``ab`` even though the ``.*`` branch is not bound by that prefix and matches anything. Grouped alternation after the prefix (``ab(?:x|y)``) keeps the guarantee and stays allowed. """ depth = 0 i = 0 while i < len(pattern): ch = pattern[i] if ch == "\\": i += 2 continue if ch == "[": i += 1 if i < len(pattern) and pattern[i] == "]": i += 1 while i < len(pattern) and pattern[i] != "]": if pattern[i] == "\\": i += 1 i += 1 elif ch == "(": depth += 1 elif ch == ")": depth = max(0, depth - 1) elif ch == "|" and depth == 0: return True i += 1 return False def _has_nested_unbounded_repeat(pattern: str) -> bool: """True if an unbounded quantifier applies to a group containing one. ``(a+)+``, ``(?:x*)*``, ``(a{2,})+`` — the canonical catastrophic- backtracking (ReDoS) shape. Registered patterns run against every log line, tool output, and transcript chunk, so a pathological pattern from a buggy plugin would stall the host process, not just the plugin. Detects structural nesting only; ambiguity between overlapping alternation branches (``(a|aa)+``) is not statically detected and remains the plugin author's responsibility. """ def _unbounded_quantifier_follows(j: int) -> bool: # Is pattern[j:] an unbounded quantifier (* + {m,}) for the atom # that just ended at 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 "" # {m,} is open-ended; {m} and {m,n} are bounded. return body[:-1].isdigit() and body.endswith(",") return False # Stack of flags: does the group at this depth contain an unbounded # repeat? Index 0 is the top level. contains_unbounded = [False] i = 0 while i < len(pattern): ch = pattern[i] if ch == "\\": i += 2 continue if ch == "[": i += 1 if i < len(pattern) and pattern[i] == "]": i += 1 while i < len(pattern) and pattern[i] != "]": if pattern[i] == "\\": i += 1 i += 1 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(i + 1): return True contains_unbounded[-1] = contains_unbounded[-1] or inner elif _unbounded_quantifier_follows(i): contains_unbounded[-1] = True if ch == "{": i = pattern.find("}", i) # skip the {m,} body i += 1 return False _PREFIX_SUBSTRINGS = tuple( _extract_literal_prefix(p) for p in _PREFIX_PATTERNS ) def _has_known_prefix_substring(text: str) -> bool: """Return True if ``text`` contains any known credential prefix substring. Used as a cheap pre-check before invoking the expensive ``_PREFIX_RE``. """ return any(p in text for p in _PREFIX_SUBSTRINGS) # --------------------------------------------------------------------------- # Plugin-registered redaction patterns # --------------------------------------------------------------------------- # # Every new vendor token format has historically required a core PR appending # to ``_PREFIX_PATTERNS`` above (fw_, retaindb_, hsk-, mem0_, brv_, ...). # This registry lets plugins add their provider's format instead. It is # ADDITIVE-ONLY by design: a plugin can extend what gets masked but has no # API to remove or weaken a built-in pattern, so a plugin can only ever # over-redact, never expose. The operator's global opt-out # (``security.redact_secrets: false`` / HERMES_REDACT_SECRETS) applies to # plugin patterns exactly as it does to built-ins. # Keyed by registration source (e.g. "plugin:my-plugin") so the plugin # lifecycle/ownership-ledger work (#64229) has a clean seam to drop ONE # plugin's patterns on unload. There is deliberately no public removal # API — additive-only stands; unload is a host-owned lifecycle concern. _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. ``redact_sensitive_text`` and ``_mask_token_nonreusable`` look these globals up at call time, so swapping the module attributes (atomic under the GIL) propagates immediately to every caller. """ global _PREFIX_RE, _PREFIX_SUBSTRINGS combined = _PREFIX_PATTERNS + _plugin_patterns() _PREFIX_RE = re.compile( r"(? int: """Additively register credential-token regexes with the redaction engine. Each accepted pattern joins the vendor-prefix alternation used by ``redact_sensitive_text`` (same masking, same head/tail rules, same non-reusable sentinel on ``file_read``) — everywhere built-in patterns apply: logs, terminal output, transport errors, transcripts. Per-pattern validation (invalid entries are warned and skipped, never raised — a broken plugin must not break startup): * must be a non-empty string that compiles as a regex; * must not contain a top-level alternation (``ab|.*`` would escape the literal-prefix guarantee below through its unprefixed branch; grouped alternation after the prefix, ``ab(?:x|y)``, is allowed); * must not nest unbounded quantifiers (``(a+)+``-style patterns can backtrack catastrophically, and registered patterns run against every log line and tool output — see ``_has_nested_unbounded_repeat``); * must start with at least 2 literal characters (the pre-screen substring gate in ``_has_known_prefix_substring`` needs a literal anchor; it also structurally rules out redact-everything patterns like ``.*``); * duplicates of built-in or already-registered patterns are skipped. Args: patterns: iterable of regex strings (e.g. ``[r"nvapi-[A-Za-z0-9_-]{20,}"]``). source: attribution label for log lines (e.g. ``"plugin:my-plugin"``). Returns: The number of patterns actually accepted. """ 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 if _has_top_level_alternation(pattern): logger.warning( "%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)'", source, pattern, ) continue if _has_nested_unbounded_repeat(pattern): logger.warning( "%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", source, pattern, ) continue if len(_extract_literal_prefix(pattern)) < 2: logger.warning( "%s: skipping redaction pattern %r — must start with at " "least 2 literal characters (needed for the pre-screen " "substring gate)", 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() _HTTP_METHOD_SUBSTRINGS = ( "GET ", "POST ", "PUT ", "PATCH ", "DELETE ", "HEAD ", "OPTIONS ", "TRACE ", "CONNECT ", ) def _has_http_method_substring(text: str) -> bool: """Cheap pre-check before scanning for access-log request targets.""" upper = text.upper() return any(method in upper for method in _HTTP_METHOD_SUBSTRINGS) class RedactingFormatter(logging.Formatter): """Log formatter that redacts secrets from all log messages.""" def __init__(self, fmt=None, datefmt=None, style='%', **kwargs): super().__init__(fmt, datefmt, style, **kwargs) def format(self, record: logging.LogRecord) -> str: original = super().format(record) return redact_sensitive_text(original)