refactor(redact): one secret-pattern source; a2a, gateway chat and monitoring egress scrub through redact_for_egress

plugins/platforms/a2a/security.py::redact_outbound shipped text to a REMOTE peer
through 8 private regexes (sk-, sk-ant-, ghp_ only, xox[bap] only, AKIA, JWT,
Bearer, email) and never called redact_sensitive_text, so every prefix added to
agent/redact.py (hf_, glpat-, xapp-, npm_, Telegram bot tokens, private keys,
DB URLs, env assignments, auth headers, plugin-registered patterns) was absent
on the A2A path. gateway/run.py::_GATEWAY_SECRET_PATTERNS and
agent/monitoring/redaction.py::_TOKEN_RE/_BEARER_RE were two more parallel
"fallback" lists to maintain.

Now agent/redact.py::redact_for_egress is the one egress scrub:
redact_sensitive_text(force=True) + a bearer sweep for prefix-less opaque
tokens, fail-closed ("[redaction-unavailable]"). Gateway user-facing text,
monitoring export and A2A outbound call it; A2A keeps only its e-mail pass.

Behavior changes: a2a egress now masks the full canonical set; the gateway
chat path returns the fail-closed sentinel instead of a raw string when the
redactor raises; honcho plugin registers hch-at-/hch-rt- with
register_redaction_patterns (masked on every surface; mask shape is the
shared head/tail form instead of "hch-at-[redacted]"); proxy_cli token
display uses mask_secret (4 visible prefix chars instead of 12).

Invariant test: redact_outbound masks a synthesized token for every
registered prefix pattern (fails when reverted to the private list).

agent/monitoring/redaction.py

@@ -1,9 +1,9 @@
 """Redaction applied to monitoring data before egress.
 
 One unconditional scrub, no modes, no knobs. Every string that leaves the process passes
-through ``redact_for_export``: secrets first (``agent/redact.py::redact_sensitive_text(force=True)``
-plus bearer/token shapes, failing CLOSED so a broken redactor never emits the raw string), then
-PII (e-mail, phone, UUID-shaped ids -> ``[email]`` / ``[phone]`` / ``[id]``).
+through ``redact_for_export``: secrets via ``agent/redact.py::redact_for_egress`` (the single
+pattern source; fails CLOSED so a broken redactor never emits the raw string), then PII
+(e-mail, phone, UUID-shaped ids -> ``[email]`` / ``[phone]`` / ``[id]``).
 """
 
 from __future__ import annotations
@@ -11,11 +11,7 @@ from __future__ import annotations
 import re
 from typing import Any, Optional
 
-# ── secret shapes (belt-and-suspenders on top of agent/redact.py) ───────────
-_BEARER_RE = re.compile(r"\bBearer\s+[A-Za-z0-9._~+\-/]+=*", re.IGNORECASE)
-_TOKEN_RE = re.compile(r"\b(xox[baprs]-[A-Za-z0-9-]+|sk-[A-Za-z0-9_-]{8,}|gh[pousr]_[A-Za-z0-9_]{8,})\b")
-_SECRET_LITERAL_RE = re.compile(r"\*{3,}")
-_BEARER_RESIDUE_RE = re.compile(r"\bBearer\s+\[[^\]]+\]", re.IGNORECASE)
+from agent.redact import REDACTION_UNAVAILABLE as UNAVAILABLE, redact_for_egress
 
 # ── PII shapes ───────────────────────────────────────────────────────────────
 _EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}")
@@ -25,27 +21,12 @@ _PHONE_RE = re.compile(
 )
 _UUID_RE = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b")
 
-UNAVAILABLE = "[redaction-unavailable]"
-
-
-def _secret_redact(text: str) -> str:
-    """Always-on secret redaction. force=True so user config can't disable it."""
-    try:
-        from agent.redact import redact_sensitive_text
-        out = redact_sensitive_text(text, force=True)
-    except Exception:
-        # Fail CLOSED: if the redactor can't run, do not emit the raw string.
-        return UNAVAILABLE
-    for pattern in (_BEARER_RE, _TOKEN_RE, _SECRET_LITERAL_RE, _BEARER_RESIDUE_RE):
-        out = pattern.sub("[redacted]", out)
-    return out
-
 
 def redact_for_export(text: Optional[str]) -> Optional[str]:
     """Scrub a string for egress: secrets, then PII. Unconditional."""
     if text is None:
         return None
-    out = _secret_redact(str(text))
+    out = redact_for_egress(str(text))
     out = _EMAIL_RE.sub("[email]", out)
     out = _UUID_RE.sub("[id]", out)
     out = _PHONE_RE.sub("[phone]", out)

agent/redact.py

@@ -797,6 +797,25 @@ def is_env_dump_command(command: str | None) -> bool:
     return False
 
 
+REDACTION_UNAVAILABLE = "[redaction-unavailable]"
+_BEARER_RESIDUE_RE = re.compile(r"\bBearer\s+(?:\[[^\]]+\]|[A-Za-z0-9._~+/-]+=*)", 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-assignment
     pass runs only when ``command`` is an env dump or reads a ``.env`` file

gateway/run.py

@@ -392,14 +392,6 @@ _CONNECTION_ERROR_MARKERS = (
     r"cannot\s+connect", r"failed\s+to\s+establish", r"could\s+not\s+connect")
 _GATEWAY_CONNECTION_ERROR_RE = re.compile("(" + "|".join(_CONNECTION_ERROR_MARKERS) + ")", re.IGNORECASE)
 
-_GATEWAY_SECRET_PATTERNS = (
-    re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_\-]{12,}\b"),
-    re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"), re.compile(r"\bxapp-\d+-[A-Za-z0-9\-]{20,}\b"),
-    re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{20,}\b"), re.compile(r"\bhf_[A-Za-z0-9]{20,}\b"),
-    re.compile(r"\bglpat-[A-Za-z0-9_\-]{20,}\b"),
-    re.compile(r"(?i)\b(Bearer\s+)[A-Za-z0-9._\-]{20,}\b"))
-
-
 def _ensure_windows_gateway_venv_imports() -> None:
     """Make detached Windows gateway runs see the Hermes venv packages.
 
@@ -544,26 +536,11 @@ def _gateway_loop_exception_handler(
 
 
 def _redact_gateway_user_facing_secrets(text: str) -> str:
-    """Secret redaction before text can leave the gateway.
+    """Secret redaction before text can leave the gateway for a chat platform: the shared egress scrub
+    (``force=True`` holds even when ``security.redact_secrets`` is off; fails closed). See #23810."""
+    from agent.redact import redact_for_egress
 
-    Shared ``redact_sensitive_text`` with ``force=True`` (holds even when ``security.redact_secrets`` is off);
-    ``_GATEWAY_SECRET_PATTERNS`` is a second pass so redaction degrades gracefully if that import fails.
-
-    Delegates to the authoritative ``agent.redact.redact_sensitive_text`` — the same Tirith-grade redactor
-    already applied to logs, tool output, and approval-command prompts — so the outbound chat path masks the
-    full credential set the startup banner promises ("chat responses are scrubbed before delivery"), not a
-    divergent subset. See #23810.
-    """
-    redacted = str(text or "")
-    try:
-        from agent.redact import redact_sensitive_text
-
-        redacted = redact_sensitive_text(redacted, force=True)
-    except Exception:
-        pass  # fail-soft: the local pattern pass below still runs rather than leaking raw text to chat
-    for pattern in _GATEWAY_SECRET_PATTERNS:
-        redacted = pattern.sub(lambda m: (m.group(1) if m.lastindex else "") + "[REDACTED]", redacted)
-    return redacted
+    return redact_for_egress(text)
 
 
 def _redact_approval_command(cmd: "str | None") -> str:

hermes_cli/proxy_cli.py

@@ -14,6 +14,7 @@ from rich.panel import Panel
 from rich.table import Table
 
 from agent.proxy_sources import iron_proxy as ip
+from agent.redact import mask_secret
 from hermes_cli.config import load_config, load_env, save_config
 
 
@@ -457,7 +458,7 @@ def format_status_text(*, show_tokens: bool = False) -> str:
     if mappings:
         lines.extend(["", "Token mappings:"])
         for m in mappings:
-            tok = m.proxy_token if show_tokens else _redact_token(m.proxy_token)
+            tok = m.proxy_token if show_tokens else mask_secret(m.proxy_token)
             lines.append(f"  - {m.real_env_name}: {tok} ({', '.join(m.upstream_hosts)})")
     uncovered = ip.discover_uncovered_providers()
     if uncovered:
@@ -594,7 +595,7 @@ def _mappings_table(mappings, env_header: str, hosts_header: str, *, show_tokens
     table.add_column(hosts_header, style="dim")
     table.add_column("Proxy token", style="green")
     for m in mappings:
-        tok = m.proxy_token if show_tokens else _redact_token(m.proxy_token)
+        tok = m.proxy_token if show_tokens else mask_secret(m.proxy_token)
         table.add_row(m.real_env_name, ", ".join(m.upstream_hosts), tok)
     return table
 
@@ -638,9 +639,3 @@ def _status_rows(proxy_cfg: dict, status, *, yn, dim) -> list[tuple[str, str]]:
         ("Credential src", str(proxy_cfg.get("credential_source", "env"))),
         ("Docker enforce", yn(bool(proxy_cfg.get("enforce_on_docker", True)))),
     ]
-
-
-def _redact_token(token: str) -> str:
-    if len(token) < 16:
-        return token
-    return f"{token[:12]}…{token[-4:]}"

plugins/memory/honcho/oauth.py

@@ -21,6 +21,8 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
+from agent.redact import redact_sensitive_text, register_redaction_patterns
+
 logger = logging.getLogger(__name__)
 
 ACCESS_TOKEN_PREFIX = "hch-at-"
@@ -36,12 +38,10 @@ _REFRESH_TOTAL_BUDGET_SECONDS = 20.0
 _REFRESH_FAILURE_COOLDOWN_SECONDS = 30.0
 # OAuth error codes a retry can never fix — the grant itself is dead.
 _PERMANENT_OAUTH_ERRORS = frozenset({"invalid_grant", "invalid_client", "unauthorized_client"})
-# Derived from the canonical prefixes so a prefix change can't silently break redaction.
-_TOKEN_VALUE_RE = re.compile(rf"({re.escape(ACCESS_TOKEN_PREFIX)}|{re.escape(REFRESH_TOKEN_PREFIX)})[A-Za-z0-9._~+/=-]+")
-
-def redact_tokens(text: str) -> str:
-    """Replace any embedded token values with their prefix plus a placeholder."""
-    return _TOKEN_VALUE_RE.sub(lambda m: f"{m.group(1)}[redacted]", text)
+# Registered with the shared redactor so EVERY log/tool-output surface masks Honcho tokens, not only
+# this module's own error strings. Built from the constants so a prefix rename can't split them.
+register_redaction_patterns([f"{p}[A-Za-z0-9._~+/=-]{{8,}}" for p in (ACCESS_TOKEN_PREFIX, REFRESH_TOKEN_PREFIX)],
+                            source="plugin:honcho")
 
 class OAuthRefreshError(Exception):
     """Token endpoint rejected the refresh. ``permanent`` means re-login is required."""
@@ -229,7 +229,7 @@ def _exchange_refresh_token(
     if status >= 400:
         error, description = str(body.get("error") or ""), str(body.get("error_description") or "")
         detail = " — ".join(p for p in (error, description) if p) or "no error body"
-        message = redact_tokens(f"token endpoint returned HTTP {status}: {detail}")
+        message = redact_sensitive_text(f"token endpoint returned HTTP {status}: {detail}", force=True)
         raise OAuthRefreshError(message, error=error, permanent=error in _PERMANENT_OAUTH_ERRORS)
     return OAuthCredential.from_token_response(
         body, now=now, client_id=cred.client_id, token_endpoint=cred.token_endpoint,
@@ -249,7 +249,7 @@ def _exchange_with_retry(cred: OAuthCredential, *, now: float) -> OAuthCredentia
     remaining = deadline - time.monotonic() - _REFRESH_RETRY_DELAY_SECONDS
     if remaining <= 0:
         raise first
-    logger.warning("Honcho OAuth token exchange failed, retrying once: %s", redact_tokens(str(first)))
+    logger.warning("Honcho OAuth token exchange failed, retrying once: %s", redact_sensitive_text(str(first), force=True))
     time.sleep(_REFRESH_RETRY_DELAY_SECONDS)
     return _exchange_refresh_token(cred, now=now, timeout=min(remaining, _REFRESH_TIMEOUT_SECONDS))
 
@@ -267,7 +267,7 @@ def _rotate_and_persist(
                          "run 'hermes honcho setup' to re-authenticate", host, exc)
             return None
         _refresh_failure_at[key] = time.monotonic()
-        logger.warning("Honcho OAuth %s failed for host %s: %s", op_label, host, redact_tokens(str(exc)))
+        logger.warning("Honcho OAuth %s failed for host %s: %s", op_label, host, redact_sensitive_text(str(exc), force=True))
         return None
     _persist_credential(path, host, rotated)
     return rotated

plugins/memory/honcho/session_auth.py

@@ -7,7 +7,7 @@ import re
 from pathlib import Path
 from typing import Any, Callable
 
-from plugins.memory.honcho.oauth import redact_tokens as _redact_tokens
+from agent.redact import redact_sensitive_text as _redact_sensitive_text
 
 logger = logging.getLogger("plugins.memory.honcho.session")
 
@@ -48,7 +48,7 @@ _REAUTH_REQUIRED_MESSAGE = (
 
 
 def _auth_error_message(exc: BaseException) -> str:
-    return (f"Honcho rejected our credentials and a forced token refresh did not recover: {_redact_tokens(str(exc))}. "
+    return (f"Honcho rejected our credentials and a forced token refresh did not recover: {_redact_sensitive_text(str(exc), force=True)}. "
             "Re-authenticate with 'hermes honcho setup'.")
 
 
@@ -56,7 +56,7 @@ class SessionAuthMixin:
     """Auth state + ``_authed_call`` for HonchoSessionManager (state lives in __init__)."""
 
     def _record_auth_failure(self, exc: BaseException) -> None:
-        detail = _redact_tokens(str(exc))
+        detail = _redact_sensitive_text(str(exc), force=True)
         if self._auth_failure is None:
             logger.error("Honcho authentication failed and token refresh did not recover; "
                          "memory sync and recall are paused until the user re-authenticates: %s", detail)
@@ -135,7 +135,7 @@ class SessionAuthMixin:
             if not _is_auth_error(e):
                 raise
             logger.warning("Honcho %s hit an auth error; forcing token refresh and retrying once: %s",
-                           op_name, _redact_tokens(str(e)))
+                           op_name, _redact_sensitive_text(str(e), force=True))
             if not self._force_reauth():
                 self._record_auth_failure(e)
                 raise HonchoAuthError(_auth_error_message(e)) from e

plugins/platforms/a2a/security.py

@@ -139,17 +139,8 @@ PRIVACY_PREFIX = (
     "colleague's request.]\n\n"
 )
 
-# Credential-shaped strings we never want to ship to a peer in a task body.
-_REDACTION_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
-    (re.compile(r"sk-[A-Za-z0-9_\-]{16,}"), "sk-[redacted]"),
-    (re.compile(r"sk-ant-[A-Za-z0-9_\-]{16,}"), "sk-ant-[redacted]"),
-    (re.compile(r"ghp_[A-Za-z0-9]{20,}"), "ghp_[redacted]"),
-    (re.compile(r"xox[bap]-[A-Za-z0-9\-]{10,}"), "xox-[redacted]"),
-    (re.compile(r"AKIA[0-9A-Z]{16}"), "AKIA[redacted]"),
-    (re.compile(r"eyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}"), "[redacted-jwt]"),
-    (re.compile(r"(?i)bearer\s+[A-Za-z0-9._\-]{20,}"), "Bearer [redacted]"),
-    (re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}"), "[redacted-email]"),
-)
+# PII the canonical secret redactor deliberately leaves alone; a peer is a third party.
+_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}")
 
 
 def filter_inbound(text: str) -> str:
@@ -166,10 +157,13 @@ def wrap_inbound(peer: str, text: str) -> str:
 
 
 def redact_outbound(text: str) -> str:
-    """Scrub credential-shaped substrings before sending text to a peer."""
-    for pat, repl in _REDACTION_PATTERNS if text else ():
-        text = pat.sub(repl, text)
-    return text
+    """Scrub credentials (the shared egress scrub — every pattern ``agent/redact.py`` knows, fail-closed)
+    and e-mail addresses before text ships to a remote peer."""
+    if not text:
+        return text
+    from agent.redact import redact_for_egress
+
+    return _EMAIL_RE.sub("[redacted-email]", redact_for_egress(text))
 
 
 # Blocked even in localhost-only mode — a remote peer must not make us probe internal services

tests/agent/test_redact.py

@@ -1185,3 +1185,19 @@ class TestValueAwareGatingCorpus:
         result = redact_sensitive_text(block, force=True)
         assert prose_line in result
         assert "A9f3kZq7Lm2Xw8Rt4Yv6" not in result
+
+
+class TestRedactForEgress:
+    """``redact_for_egress`` is the single scrub every remote-reader surface (gateway chat, A2A, monitoring)
+    calls; there is no second pattern list to keep in sync."""
+
+    def test_opaque_bearer_without_vendor_prefix_is_masked(self):
+        from agent.redact import redact_for_egress
+        out = redact_for_egress("curl -H 'Authorization: Bearer opaque0123456789abcdef' https://x.example")
+        assert "opaque0123456789abcdef" not in out
+        assert "https://x.example" in out
+
+    def test_fails_closed_when_the_redactor_raises(self, monkeypatch):
+        from agent import redact as R
+        monkeypatch.setattr(R, "redact_sensitive_text", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")))
+        assert R.redact_for_egress("sk-live-0123456789abcdef") == R.REDACTION_UNAVAILABLE

tests/honcho_plugin/test_auth_recovery.py

@@ -153,14 +153,16 @@ class TestExchangeRetry:
         assert "invalid_grant" in caplog.text
         assert "grant revoked" in caplog.text
 
-    def test_redaction_strips_token_values(self):
-        redacted = oauth.redact_tokens(
-            "exchange failed for hch-rt-supersecret123 got hch-at-alsosecret456"
+    def test_honcho_token_prefixes_are_registered_with_the_shared_redactor(self):
+        """Importing the plugin registers hch-at-/hch-rt- with agent.redact, so every surface that
+        runs the shared redactor (logs, tool output, chat egress) masks Honcho tokens, not only
+        this module's own error strings."""
+        from agent.redact import redact_sensitive_text
+        redacted = redact_sensitive_text(
+            "exchange failed for hch-rt-supersecret123 got hch-at-alsosecret456", force=True
         )
         assert "supersecret123" not in redacted
         assert "alsosecret456" not in redacted
-        assert "hch-rt-[redacted]" in redacted
-        assert "hch-at-[redacted]" in redacted
 
 
 class TestForceRefreshToken:
@@ -537,7 +539,6 @@ class TestAuthNotice:
         mgr._record_auth_failure(Exception("rejected token hch-at-secretvalue99"))
         notice = mgr.pop_auth_notice()
         assert "secretvalue99" not in notice
-        assert "hch-at-[redacted]" in notice
 
     def test_provider_prefetch_injects_notice_once(self):
         class _FakeManager:

tests/plugins/test_a2a_plugin.py

@@ -13,6 +13,7 @@ import asyncio
 import hashlib
 import hmac
 import json
+import re
 import os
 import socket
 import threading
@@ -176,17 +177,31 @@ class TestInjectionFilter:
 
 
 class TestOutboundRedaction:
-    def test_openai_key_redacted(self):
-        out = security.redact_outbound("my key is sk-abcdefghij1234567890XYZ")
-        assert "sk-abcdefghij" not in out
-        assert "[redacted]" in out
+    def test_every_canonical_credential_class_is_scrubbed(self):
+        """Invariant: redact_outbound masks everything redact_sensitive_text masks. A2A ships text to a
+        REMOTE peer, so a private subset here silently drops every prefix later added to agent/redact.py.
+        Corpus: one synthetic token per registered prefix pattern, built from the pattern's literal prefix."""
+        from agent import redact as R
 
-    def test_github_token_redacted(self):
-        out = security.redact_outbound("token ghp_0123456789abcdefghij0123")
-        assert "ghp_0123456789" not in out
+        bodies = ("Qq7zP2mX9vLk4nRt8wYb1cDf6gHj3sA0", "QQ7ZP2MX9VLK4NRT", "b-Qq7zP2mX9vLk4nRt8wYb1cDf6gHj3sA0",
+                  ".Qq7zP2mX9vLk4nRt8wYb1cDf6gHj3sA0", "1-Qq7zP2mX9vLk4nRt8wYb1cDf6gHj3sA0",
+                  "Qq7zP2mX9vLk4nRt8wYb1cDf6gHj3sA0.Qq7zP2mX9vLk4nRt8wYb1cDf6gHj3sA0")
+        tokens = []
+        for pattern in R._PREFIX_PATTERNS + R._plugin_patterns():
+            prefix = R._extract_literal_prefix(pattern)
+            token = next((prefix + body for body in bodies if re.fullmatch(pattern, prefix + body)), None)
+            assert token, f"could not synthesize a token for {pattern!r}"
+            tokens.append(token)
+        for token in tokens:
+            leaked = R.redact_sensitive_text(f"peer, here: {token}", force=True)
+            if token in leaked:
+                continue  # the canonical redactor itself passes it (word-boundary/shape rule); not our contract
+            assert token not in security.redact_outbound(f"peer, here: {token}"), token
+        assert len(tokens) >= 50
 
-    def test_email_redacted(self):
-        out = security.redact_outbound("contact me at alice@example.com")
+    def test_bearer_and_email_redacted(self):
+        out = security.redact_outbound("Authorization: Bearer opaque0123456789abcdef; contact me at alice@example.com")
+        assert "opaque0123456789abcdef" not in out
         assert "alice@example.com" not in out
         assert "[redacted-email]" in out