fix(mcp): fully redact credential headers in MCP probe errors and test display (salvage #97466)

`hermes mcp test` resolved Authorization headers and printed first4***last4
— still a reusable credential fragment — and probe exceptions that echoed
`Authorization: Bearer <value>` reached the CLI error line and the dashboard
`POST /api/mcp/servers/{name}/test` response verbatim.

Redact once at the `_probe_single_server` raise seam so every consumer
(`mcp add`, `mcp test`, `mcp login`, `mcp configure`, the dashboard probe,
`hermes doctor`, catalog probes) prints already-safe text. Recognized
credential header fields (Authorization/Proxy-Authorization plus
agent.redact._SECRET_HEADER_NAMES) have their complete value replaced with
***; bare Bearer/Basic/Token/Digest spans are covered; the generic redactor
runs force=True as a second pass. CLI header display fails closed: only pure
${ENV} template values print.

Salvaged from PR #97466 by @686f6c61 (base predated the mcp_config/web_routers
decomposition; re-applied onto current main, test seams repointed to the
defining modules tools.mcp_tool_loop / tools.mcp_tool_lifecycle).

Inspired by Claude Code 2.1.268: "Fixed /mcp and /plugin server details,
claude mcp list/get, and MCP login errors showing secrets resolved from
${VAR} placeholders in MCP configs."

Fixes #97460

Co-authored-by: 686f6c61 <github@00b.tech>

hermes_cli/mcp_config.py

@@ -25,6 +25,160 @@ logger = logging.getLogger(__name__)
 
 _ENV_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
 
+
+# MCP test/dashboard surfaces must not fingerprint credential header values
+# (firstN/lastN is a reusable fragment). Redact by field identity: once a
+# recognized credential header key is found, replace its complete associated
+# value regardless of scheme, quoting, parameter order, or wire/JSON/Python
+# mapping serialization. Header names come from agent.redact so the lists
+# cannot drift. The generic redactor then runs with force=True.
+_AUTH_SCHEME_PREFIX_RE = re.compile(
+    r"^(?:Bearer|Basic|Token|Digest)\s+",
+    re.IGNORECASE,
+)
+_DIGEST_PARAM_NAMES = (
+    "username|response|opaque|cnonce|nonce|uri|realm|qop|nc|algorithm"
+)
+_DIGEST_PARAM = (
+    rf"(?:{_DIGEST_PARAM_NAMES})\s*=\s*(?:\"[^\"]*\"|'[^']*'|[^\s,]+)"
+)
+_DIGEST_PARAMS = rf"{_DIGEST_PARAM}(?:\s*,\s*{_DIGEST_PARAM})*"
+_PROBE_REDACTION_RES: Optional[Tuple[re.Pattern[str], ...]] = None
+
+
+def _credential_header_names() -> str:
+    from agent.redact import _SECRET_HEADER_NAMES
+
+    return rf"(?:(?:Proxy-)?Authorization|{_SECRET_HEADER_NAMES})"
+
+
+def _probe_redaction_res() -> Tuple[re.Pattern[str], ...]:
+    global _PROBE_REDACTION_RES
+    if _PROBE_REDACTION_RES is not None:
+        return _PROBE_REDACTION_RES
+    header = _credential_header_names()
+    # Quoted-key mapping/JSON: {'X-Api-Key': '…'} / {"Authorization": "…"}
+    mapping = re.compile(
+        rf"""(['\"])({header})\1(\s*:\s*)(['\"])((?:\\.|(?!\4).)*)\4""",
+        re.IGNORECASE,
+    )
+    unquoted_mapping = re.compile(
+        rf"""(['\"])({header})\1(\s*:\s*)(?!['\"])([^\s,}}\]]+)""",
+        re.IGNORECASE,
+    )
+    # Bare wire header: Authorization: Digest … / X-Api-Key: token
+    wire = re.compile(
+        rf"({header})(\s*:\s*)([^\n\r]+)",
+        re.IGNORECASE,
+    )
+    bare_scheme = re.compile(
+        rf"\b(Bearer|Basic|Token)(\s+)([^\s\"']+)"
+        rf"|\b(Digest)(\s+)({_DIGEST_PARAMS})",
+        re.IGNORECASE,
+    )
+    _PROBE_REDACTION_RES = (mapping, unquoted_mapping, wire, bare_scheme)
+    return _PROBE_REDACTION_RES
+
+
+def _mask_header_field_value(value: str) -> str:
+    """Replace a credential header's complete associated value with ``***``.
+
+    A leading auth scheme word is kept for debugging; Digest parameters are
+    part of the field value and are not tokenized.
+    """
+    scheme = _AUTH_SCHEME_PREFIX_RE.match(value)
+    if scheme:
+        return f"{scheme.group(0)}***"
+    return "***"
+
+
+def _sub_mapping_header(match: re.Match[str]) -> str:
+    return (
+        f"{match.group(1)}{match.group(2)}{match.group(1)}"
+        f"{match.group(3)}{match.group(4)}"
+        f"{_mask_header_field_value(match.group(5))}{match.group(4)}"
+    )
+
+
+def _sub_unquoted_mapping_header(match: re.Match[str]) -> str:
+    return (
+        f"{match.group(1)}{match.group(2)}{match.group(1)}"
+        f"{match.group(3)}{_mask_header_field_value(match.group(4))}"
+    )
+
+
+def _sub_wire_header(match: re.Match[str]) -> str:
+    return f"{match.group(1)}{match.group(2)}{_mask_header_field_value(match.group(3))}"
+
+
+def _sub_bare_scheme(match: re.Match[str]) -> str:
+    if match.group(1):
+        return f"{match.group(1)}{match.group(2)}***"
+    return f"{match.group(4)}{match.group(5)}***"
+
+
+def redact_mcp_probe_text(text: object) -> str:
+    """Fully redact MCP probe/display strings before they leave the process.
+
+    Recognized credential header fields (Authorization / Proxy-Authorization
+    and ``agent.redact._SECRET_HEADER_NAMES``) have their complete associated
+    value replaced with ``***``. Bare Bearer/Basic/Token/Digest spans are
+    covered the same way. The generic secret redactor then runs with
+    ``force=True`` as defense in depth.
+    """
+    raw = "" if text is None else str(text)
+    if not raw:
+        return raw
+    mapping, unquoted_mapping, wire, bare_scheme = _probe_redaction_res()
+    redacted = mapping.sub(_sub_mapping_header, raw)
+    redacted = unquoted_mapping.sub(_sub_unquoted_mapping_header, redacted)
+    redacted = wire.sub(_sub_wire_header, redacted)
+    redacted = bare_scheme.sub(_sub_bare_scheme, redacted)
+    from agent.redact import redact_sensitive_text
+
+    return redact_sensitive_text(redacted, force=True)
+
+
+def _header_value_is_only_env_refs(value: str) -> bool:
+    """True when every non-scheme span in *value* is a ``${VAR}`` reference."""
+    leftover = _ENV_VAR_PATTERN.sub("", value)
+    leftover = _AUTH_SCHEME_PREFIX_RE.sub("", leftover)
+    return leftover.strip(" \t;,") == ""
+
+
+def redact_mcp_header_display(name: str, value: object) -> str:
+    """Fail-closed CLI display for a credential-shaped MCP header.
+
+    A value that is only ``${ENV}`` references (optionally with an auth scheme
+    word) may be shown — it carries no secret. Anything else, including opaque
+    API keys and mixed template+literal strings, is replaced with ``***``.
+    ``name`` stays paired with the value so callers cannot drop the header
+    identity before this policy runs.
+    """
+    raw = "" if value is None else str(value)
+    if raw and _header_value_is_only_env_refs(raw):
+        return raw
+    # Pair name+value for the generic redactor, then still fail closed — an
+    # opaque token with no recognized scheme must not print.
+    redact_mcp_probe_text(f"{name}: {raw}")
+    return "***"
+
+
+def _redact_probe_exception(exc: BaseException) -> Exception:
+    """Return a raise-able exception whose ``str()`` is safe to print."""
+    root = _unwrap_exception_group(exc)
+    safe = redact_mcp_probe_text(root)
+    if safe == str(root) and isinstance(root, Exception):
+        return root
+    try:
+        rebuilt = type(root)(safe)
+        if str(rebuilt) == safe and isinstance(rebuilt, Exception):
+            return rebuilt
+    except Exception:
+        pass
+    return RuntimeError(safe)
+
+
 _MCP_PRESETS: Dict[str, Dict[str, Any]] = {
     "codex": {"command": "codex", "args": ["mcp-server"]},
 }
@@ -334,7 +488,7 @@ def _probe_single_server(
     try:
         _run_on_mcp_loop(_probe(), timeout=connect_timeout + 10)
     except BaseException as exc:
-        raise _unwrap_exception_group(exc) from None
+        raise _redact_probe_exception(exc) from None
     finally:
         _stop_mcp_loop_if_idle()
     return tools_found
@@ -490,7 +644,7 @@ def cmd_mcp_add(args):
     try:
         tools = _probe_single_server(name, server_config)
     except Exception as exc:
-        _error(f"Failed to connect: {exc}")
+        _error(f"Failed to connect: {redact_mcp_probe_text(exc)}")
         if _confirm("Save config anyway (you can test later)?", default=False):
             server_config["enabled"] = False
             if _save_mcp_server(name, server_config):
@@ -603,10 +757,9 @@ def cmd_mcp_test(args):
     elif headers:
         for k, v in headers.items():
             if isinstance(v, str) and ("key" in k.lower() or "auth" in k.lower()):
-                # Mask the value (accepts ${VAR} and Cursor-style ${env:VAR})
-                resolved = _ENV_VAR_PATTERN.sub(lambda m: os.getenv(_env_ref_name(m.group(1)), ""), v)
-                masked = resolved[:4] + "***" + resolved[-4:] if len(resolved) > 8 else "***"
-                print(f"    {k}: {masked}")
+                # Keep header identity with the value. A ${ENV} substring is
+                # not proof the rest of the header is non-secret.
+                print(f"    {k}: {redact_mcp_header_display(k, v)}")
     else:
         _info("Auth: none")
 
@@ -614,7 +767,7 @@ def cmd_mcp_test(args):
     try:
         tools = _probe_single_server(name, cfg)
     except Exception as exc:
-        _error(f"Connection failed ({(time.monotonic() - start) * 1000:.0f}ms): {exc}")
+        _error(f"Connection failed ({(time.monotonic() - start) * 1000:.0f}ms): {redact_mcp_probe_text(exc)}")
         return
     _success(f"Connected ({(time.monotonic() - start) * 1000:.0f}ms)")
     _success(f"Tools discovered: {len(tools)}")
@@ -706,7 +859,7 @@ def _reauth_oauth_server(name: str, server_config: dict, *, flow: str | None = N
             humanized = humanize_oauth_registration_error(name, exc, server_url=url)
         except Exception:
             humanized = None
-        _error(f"Authentication failed: {humanized or exc}")
+        _error(f"Authentication failed: {redact_mcp_probe_text(humanized or exc)}")
         return False
 
 
@@ -801,7 +954,7 @@ def cmd_mcp_configure(args):
     try:
         all_tools = _probe_single_server(name, cfg)
     except Exception as exc:
-        _error(f"Failed to connect: {exc}")
+        _error(f"Failed to connect: {redact_mcp_probe_text(exc)}")
         return
     if not all_tools:
         _warning("Server reports no tools.")

hermes_cli/web_routers/mcp.py

@@ -170,7 +170,9 @@ async def test_mcp_server(name: str, profile: Optional[str] = None):
     try:  # probe blocks on a dedicated MCP event loop — keep it off the FastAPI loop
         tools, token_present = await asyncio.to_thread(_probe_scoped)
     except Exception as exc:
-        return {"ok": False, "error": str(exc), "tools": []}
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        return {"ok": False, "error": redact_mcp_probe_text(exc), "tools": []}
     if not token_present:
         return {"ok": False, "error": "OAuth authentication required — no token found.", "tools": []}
     # Optional per-tool schema size (chars) for the desktop's cost overlay;

tests/hermes_cli/test_mcp_probe_redaction.py

@@ -0,0 +1,476 @@
+"""MCP test/dashboard must not fingerprint Authorization credentials (#97460)."""
+
+import argparse
+import itertools
+import json
+
+import pytest
+import yaml
+
+
+SYNTHETIC = "SYNTHETIC_MCP_BEARER_NOT_A_SECRET_123456"
+SYNTHETIC_PREFIX = "SYNTHETIC_MCP_BEARER"
+SYNTHETIC_SUFFIX = "SECRET_123456"
+HEADER = f"Authorization: Bearer {SYNTHETIC}"
+OPAQUE_API_KEY = "opaquecredential1234567890ABCDEF"
+OPAQUE_PREFIX = OPAQUE_API_KEY[:6]
+OPAQUE_SUFFIX = OPAQUE_API_KEY[-4:]
+
+
+def _assert_fully_redacted(text: str) -> None:
+    assert SYNTHETIC not in text
+    assert SYNTHETIC_PREFIX not in text
+    assert SYNTHETIC_SUFFIX not in text
+    assert OPAQUE_API_KEY not in text
+    assert OPAQUE_PREFIX not in text
+    assert OPAQUE_SUFFIX not in text
+
+
+def _make_args(**kwargs):
+    defaults = {
+        "name": "test-server",
+        "url": None,
+        "mcp_command": None,
+        "args": None,
+        "auth": None,
+        "preset": None,
+        "env": None,
+        "mcp_action": None,
+        "connect_timeout": None,
+    }
+    defaults.update(kwargs)
+    return argparse.Namespace(**defaults)
+
+
+@pytest.fixture(autouse=True)
+def _isolate_config(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr("hermes_cli.config.get_hermes_home", lambda: tmp_path)
+    config_path = tmp_path / "config.yaml"
+    env_path = tmp_path / ".env"
+    monkeypatch.setattr("hermes_cli.config.get_config_path", lambda: config_path)
+    monkeypatch.setattr("hermes_cli.config.get_env_path", lambda: env_path)
+    return tmp_path
+
+
+def _seed_config(tmp_path, mcp_servers):
+    config_path = tmp_path / "config.yaml"
+    with open(config_path, "w", encoding="utf-8") as f:
+        yaml.safe_dump({"mcp_servers": mcp_servers, "_config_version": 9}, f)
+
+
+class TestRedactMcpProbeText:
+    def test_authorization_header_is_fully_replaced(self):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        out = redact_mcp_probe_text(f"401 {HEADER}")
+        _assert_fully_redacted(out)
+        assert "Authorization:" in out
+        assert "Bearer ***" in out
+
+    def test_bare_bearer_scheme_is_fully_replaced(self):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        out = redact_mcp_probe_text(f"probe Bearer {SYNTHETIC} failed")
+        _assert_fully_redacted(out)
+        assert "Bearer ***" in out
+
+    def test_opaque_api_key_header_is_fully_replaced(self):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        out = redact_mcp_probe_text(f"connect failed: X-Api-Key: {OPAQUE_API_KEY}")
+        _assert_fully_redacted(out)
+        assert "X-Api-Key: ***" in out
+
+    def test_digest_authorization_params_are_fully_replaced(self):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        out = redact_mcp_probe_text(
+            f'Authorization: Digest username="u", response="{OPAQUE_API_KEY}"'
+        )
+        _assert_fully_redacted(out)
+        assert "Digest ***" in out
+
+    def test_digest_response_first_does_not_orphan_quoted_value(self):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        out = redact_mcp_probe_text(
+            f'Authorization: Digest response="{OPAQUE_API_KEY}", username="u"'
+        )
+        _assert_fully_redacted(out)
+        assert "Digest ***" in out
+        assert "response=" not in out
+
+    def test_python_mapping_api_key_is_fully_replaced(self):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        payload = {"X-Api-Key": OPAQUE_API_KEY}
+        out = redact_mcp_probe_text(f"headers={payload!r}")
+        _assert_fully_redacted(out)
+        assert "***" in out
+
+    def test_json_mapping_api_key_is_fully_replaced(self):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        out = redact_mcp_probe_text(json.dumps({"X-Api-Key": OPAQUE_API_KEY}))
+        _assert_fully_redacted(out)
+        assert "***" in out
+
+    @pytest.mark.parametrize(
+        "params",
+        list(
+            itertools.permutations(
+                [
+                    ("username", "u"),
+                    ("response", OPAQUE_API_KEY),
+                    ("opaque", OPAQUE_API_KEY),
+                ]
+            )
+        ),
+    )
+    @pytest.mark.parametrize("quoted_values", [True, False])
+    def test_digest_parameter_order_and_quoting(self, params, quoted_values):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        parts = []
+        for key, value in params:
+            parts.append(f'{key}="{value}"' if quoted_values else f"{key}={value}")
+        wire = "Authorization: Digest " + ", ".join(parts)
+        out = redact_mcp_probe_text(wire)
+        _assert_fully_redacted(out)
+        assert "Digest ***" in out
+
+    @pytest.mark.parametrize("key_quote,value_quote", [
+        ("'", "'"),
+        ('"', '"'),
+        ("'", '"'),
+        ('"', "'"),
+    ])
+    def test_mapping_quotes_on_api_key(self, key_quote, value_quote):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        text = (
+            "headers={"
+            f"{key_quote}X-Api-Key{key_quote}: "
+            f"{value_quote}{OPAQUE_API_KEY}{value_quote}"
+            "}"
+        )
+        out = redact_mcp_probe_text(text)
+        _assert_fully_redacted(out)
+        assert "***" in out
+
+    def test_mapping_digest_authorization_is_fully_replaced(self):
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        value = f'Digest response="{OPAQUE_API_KEY}", username="u"'
+        out = redact_mcp_probe_text(f"headers={{'Authorization': {value!r}}}")
+        _assert_fully_redacted(out)
+        assert "Digest ***" in out
+
+    def test_shared_secret_header_vocabulary_is_fully_replaced(self):
+        from agent.redact import _SECRET_HEADER_NAMES
+        from hermes_cli.mcp_config import redact_mcp_probe_text
+
+        inner = _SECRET_HEADER_NAMES
+        if inner.startswith("(?:") and inner.endswith(")"):
+            inner = inner[3:-1]
+        for name in inner.split("|"):
+            out = redact_mcp_probe_text({name: OPAQUE_API_KEY}.__repr__())
+            _assert_fully_redacted(out)
+
+    def test_header_display_masks_opaque_api_key(self):
+        from hermes_cli.mcp_config import redact_mcp_header_display
+
+        assert redact_mcp_header_display("X-Api-Key", OPAQUE_API_KEY) == "***"
+        assert redact_mcp_header_display(
+            "Authorization", f"Bearer ${{MCP_TEST_TOKEN}}; backup={SYNTHETIC}"
+        ) == "***"
+
+    def test_header_display_keeps_pure_env_template(self):
+        from hermes_cli.mcp_config import redact_mcp_header_display
+
+        assert redact_mcp_header_display(
+            "Authorization", "Bearer ${MCP_TEST_TOKEN}"
+        ) == "Bearer ${MCP_TEST_TOKEN}"
+
+
+class TestProbeHelperRedactsBeforeRaise:
+    def test_probe_exception_leaving_helper_is_already_safe(self, monkeypatch):
+        import tools.mcp_tool_lifecycle as mcp_lifecycle
+        import tools.mcp_tool_loop as mcp_loop
+        from hermes_cli.mcp_config import _probe_single_server
+
+        monkeypatch.setattr(mcp_loop, "_ensure_mcp_loop", lambda: None)
+        monkeypatch.setattr(mcp_lifecycle, "_stop_mcp_loop_if_idle", lambda: None)
+
+        def boom(coro, timeout):
+            coro.close()
+            raise RuntimeError(f"connect failed: {HEADER}")
+
+        monkeypatch.setattr(mcp_loop, "_run_on_mcp_loop", boom)
+
+        with pytest.raises(Exception) as caught:
+            _probe_single_server("ink", {"url": "https://mcp.example/mcp"})
+        _assert_fully_redacted(str(caught.value))
+        assert "Bearer ***" in str(caught.value)
+
+    def test_probe_exception_redacts_opaque_api_key(self, monkeypatch):
+        import tools.mcp_tool_lifecycle as mcp_lifecycle
+        import tools.mcp_tool_loop as mcp_loop
+        from hermes_cli.mcp_config import _probe_single_server
+
+        monkeypatch.setattr(mcp_loop, "_ensure_mcp_loop", lambda: None)
+        monkeypatch.setattr(mcp_lifecycle, "_stop_mcp_loop_if_idle", lambda: None)
+
+        def boom(coro, timeout):
+            coro.close()
+            raise RuntimeError(f"connect failed: X-Api-Key: {OPAQUE_API_KEY}")
+
+        monkeypatch.setattr(mcp_loop, "_run_on_mcp_loop", boom)
+
+        with pytest.raises(Exception) as caught:
+            _probe_single_server("ink", {"url": "https://mcp.example/mcp"})
+        _assert_fully_redacted(str(caught.value))
+        assert "X-Api-Key: ***" in str(caught.value)
+
+    def test_probe_exception_redacts_digest_response_first(self, monkeypatch):
+        import tools.mcp_tool_lifecycle as mcp_lifecycle
+        import tools.mcp_tool_loop as mcp_loop
+        from hermes_cli.mcp_config import _probe_single_server
+
+        monkeypatch.setattr(mcp_loop, "_ensure_mcp_loop", lambda: None)
+        monkeypatch.setattr(mcp_lifecycle, "_stop_mcp_loop_if_idle", lambda: None)
+
+        def boom(coro, timeout):
+            coro.close()
+            raise RuntimeError(
+                f'connect failed: Authorization: Digest '
+                f'response="{OPAQUE_API_KEY}", username="u"'
+            )
+
+        monkeypatch.setattr(mcp_loop, "_run_on_mcp_loop", boom)
+
+        with pytest.raises(Exception) as caught:
+            _probe_single_server("ink", {"url": "https://mcp.example/mcp"})
+        _assert_fully_redacted(str(caught.value))
+        assert "Digest ***" in str(caught.value)
+
+    def test_probe_exception_redacts_python_mapping_api_key(self, monkeypatch):
+        import tools.mcp_tool_lifecycle as mcp_lifecycle
+        import tools.mcp_tool_loop as mcp_loop
+        from hermes_cli.mcp_config import _probe_single_server
+
+        monkeypatch.setattr(mcp_loop, "_ensure_mcp_loop", lambda: None)
+        monkeypatch.setattr(mcp_lifecycle, "_stop_mcp_loop_if_idle", lambda: None)
+
+        def boom(coro, timeout):
+            coro.close()
+            raise RuntimeError(f"headers={{'X-Api-Key': '{OPAQUE_API_KEY}'}}")
+
+        monkeypatch.setattr(mcp_loop, "_run_on_mcp_loop", boom)
+
+        with pytest.raises(Exception) as caught:
+            _probe_single_server("ink", {"url": "https://mcp.example/mcp"})
+        _assert_fully_redacted(str(caught.value))
+        assert "***" in str(caught.value)
+
+
+class TestCmdMcpTestRedaction:
+    def test_success_display_keeps_env_template(self, tmp_path, capsys, monkeypatch):
+        monkeypatch.setenv("MCP_TEST_TOKEN", SYNTHETIC)
+        _seed_config(tmp_path, {
+            "ink": {
+                "url": "https://mcp.example/mcp",
+                "headers": {"Authorization": "Bearer ${MCP_TEST_TOKEN}"},
+            },
+        })
+        monkeypatch.setattr(
+            "hermes_cli.mcp_config._probe_single_server",
+            lambda *a, **k: [("ping", "Ping")],
+        )
+        from hermes_cli.mcp_config import cmd_mcp_test
+
+        cmd_mcp_test(argparse.Namespace(name="ink"))
+        out = capsys.readouterr().out
+        _assert_fully_redacted(out)
+        assert "Connected" in out
+        assert "${MCP_TEST_TOKEN}" in (tmp_path / "config.yaml").read_text(encoding="utf-8")
+
+    def test_success_display_masks_literal_header(self, tmp_path, capsys, monkeypatch):
+        _seed_config(tmp_path, {
+            "ink": {
+                "url": "https://mcp.example/mcp",
+                "headers": {"Authorization": f"Bearer {SYNTHETIC}"},
+            },
+        })
+        monkeypatch.setattr(
+            "hermes_cli.mcp_config._probe_single_server",
+            lambda *a, **k: [("ping", "Ping")],
+        )
+        from hermes_cli.mcp_config import cmd_mcp_test
+
+        cmd_mcp_test(argparse.Namespace(name="ink"))
+        out = capsys.readouterr().out
+        _assert_fully_redacted(out)
+        assert "Authorization:" in out
+
+    def test_success_display_masks_opaque_api_key_header(self, tmp_path, capsys, monkeypatch):
+        _seed_config(tmp_path, {
+            "ink": {
+                "url": "https://mcp.example/mcp",
+                "headers": {"X-Api-Key": OPAQUE_API_KEY},
+            },
+        })
+        monkeypatch.setattr(
+            "hermes_cli.mcp_config._probe_single_server",
+            lambda *a, **k: [("ping", "Ping")],
+        )
+        from hermes_cli.mcp_config import cmd_mcp_test
+
+        cmd_mcp_test(argparse.Namespace(name="ink"))
+        out = capsys.readouterr().out
+        _assert_fully_redacted(out)
+        assert "X-Api-Key:" in out
+        assert "***" in out
+
+    def test_success_display_masks_mixed_env_and_literal(self, tmp_path, capsys, monkeypatch):
+        _seed_config(tmp_path, {
+            "ink": {
+                "url": "https://mcp.example/mcp",
+                "headers": {
+                    "Authorization": f"Bearer ${{MCP_TEST_TOKEN}}; backup={SYNTHETIC}",
+                },
+            },
+        })
+        monkeypatch.setattr(
+            "hermes_cli.mcp_config._probe_single_server",
+            lambda *a, **k: [("ping", "Ping")],
+        )
+        from hermes_cli.mcp_config import cmd_mcp_test
+
+        cmd_mcp_test(argparse.Namespace(name="ink"))
+        out = capsys.readouterr().out
+        _assert_fully_redacted(out)
+        assert "***" in out
+
+    def test_probe_exception_is_redacted(self, tmp_path, capsys, monkeypatch):
+        _seed_config(tmp_path, {
+            "ink": {"url": "https://mcp.example/mcp"},
+        })
+
+        def boom(*a, **k):
+            raise RuntimeError(f"connect failed: {HEADER}")
+
+        monkeypatch.setattr("hermes_cli.mcp_config._probe_single_server", boom)
+        from hermes_cli.mcp_config import cmd_mcp_test
+
+        cmd_mcp_test(argparse.Namespace(name="ink"))
+        out = capsys.readouterr().out
+        _assert_fully_redacted(out)
+        assert "Connection failed" in out
+        assert "Bearer ***" in out
+
+
+class TestDashboardMcpTestRedaction:
+    def test_probe_error_json_is_redacted(self, tmp_path, monkeypatch):
+        try:
+            from starlette.testclient import TestClient
+        except ImportError:
+            pytest.skip("fastapi/starlette not installed")
+
+        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
+        import hermes_cli.mcp_config as mcp_config
+
+        _seed_config(tmp_path, {
+            "ink": {"url": "https://mcp.example/mcp"},
+        })
+
+        def boom(name, config, connect_timeout=30, details=None):
+            raise RuntimeError(f"connect failed: {HEADER}")
+
+        monkeypatch.setattr(mcp_config, "_probe_single_server", boom)
+        monkeypatch.setattr(mcp_config, "_get_mcp_servers", lambda: {
+            "ink": {"url": "https://mcp.example/mcp"},
+        })
+
+        client = TestClient(app)
+        client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
+        resp = client.post("/api/mcp/servers/ink/test")
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["ok"] is False
+        _assert_fully_redacted(body["error"])
+        assert "Bearer ***" in body["error"]
+        assert SYNTHETIC not in resp.text
+
+    def test_probe_error_json_redacts_digest_and_mapping(self, tmp_path, monkeypatch):
+        try:
+            from starlette.testclient import TestClient
+        except ImportError:
+            pytest.skip("fastapi/starlette not installed")
+
+        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
+        import hermes_cli.mcp_config as mcp_config
+
+        _seed_config(tmp_path, {
+            "ink": {"url": "https://mcp.example/mcp"},
+        })
+
+        def boom(name, config, connect_timeout=30, details=None):
+            raise RuntimeError(
+                f"headers={{'X-Api-Key': '{OPAQUE_API_KEY}'}}; "
+                f'Authorization: Digest response="{OPAQUE_API_KEY}", username="u"'
+            )
+
+        monkeypatch.setattr(mcp_config, "_probe_single_server", boom)
+        monkeypatch.setattr(mcp_config, "_get_mcp_servers", lambda: {
+            "ink": {"url": "https://mcp.example/mcp"},
+        })
+
+        client = TestClient(app)
+        client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
+        resp = client.post("/api/mcp/servers/ink/test")
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["ok"] is False
+        _assert_fully_redacted(body["error"])
+        assert "***" in body["error"]
+        assert SYNTHETIC not in resp.text
+        assert OPAQUE_API_KEY not in resp.text
+
+
+class TestSiblingProbeConsumersRedact:
+    def test_mcp_add_redacts_probe_exception(self, tmp_path, capsys, monkeypatch):
+        def boom(*a, **k):
+            raise RuntimeError(f"connect failed: {HEADER}")
+
+        monkeypatch.setattr("hermes_cli.mcp_config._probe_single_server", boom)
+        monkeypatch.setattr("hermes_cli.mcp_config._confirm", lambda *a, **k: False)
+        from hermes_cli.mcp_config import cmd_mcp_add
+
+        cmd_mcp_add(_make_args(name="ink", url="https://mcp.example/mcp"))
+        out = capsys.readouterr().out
+        _assert_fully_redacted(out)
+        assert "Failed to connect" in out
+        assert "Bearer ***" in out
+
+    def test_mcp_login_redacts_probe_exception(self, tmp_path, capsys, monkeypatch):
+        _seed_config(tmp_path, {
+            "ink": {"url": "https://mcp.example/mcp", "auth": "oauth"},
+        })
+
+        def boom(*a, **k):
+            raise RuntimeError(f"connect failed: {HEADER}")
+
+        monkeypatch.setattr("hermes_cli.mcp_config._probe_single_server", boom)
+        monkeypatch.setattr(
+            "tools.mcp_oauth.humanize_oauth_registration_error",
+            lambda *a, **k: None,
+        )
+        from hermes_cli.mcp_config import cmd_mcp_login
+
+        cmd_mcp_login(_make_args(name="ink"))
+        out = capsys.readouterr().out
+        _assert_fully_redacted(out)
+        assert "Authentication failed" in out
+        assert "Bearer ***" in out