fix(debug): redact support egress before truncation

(cherry picked from commit 8129ba3e87381b410c664117a102771a1a53fad4)
2026-10-01 03:50:22 +03:00
parent db45b44ab7
commit b5b6050c81
9 changed files with 568 additions and 73 deletions
--- a/gateway/slash_commands.py
+++ b/gateway/slash_commands.py
@@ -1216,6 +1216,7 @@ class GatewaySlashCommandsMixin(
        from hermes_cli.debug import (_GATEWAY_PRIVACY_NOTICE, _best_effort_sweep_expired_pastes,
                                      _capture_dump, _is_dpaste_url, _schedule_auto_delete,
                                      collect_debug_report, upload_to_pastebin)
+        from hermes_cli.debug_redaction import redact_debug_support_error

        def _collect_and_upload():  # blocking I/O (dump capture, log reads, uploads) -> thread
            _best_effort_sweep_expired_pastes()
@@ -1223,7 +1224,7 @@ class GatewaySlashCommandsMixin(
            try:
                urls = {t("gateway.debug.report_label"): upload_to_pastebin(report)}
            except Exception as exc:
-                return t("gateway.debug.upload_failed", error=exc)
+                return t("gateway.debug.upload_failed", error=redact_debug_support_error(exc))
            _schedule_auto_delete(list(urls.values()))  # paste.rs only; dpaste.com has no delete
            label_width = max(len(k) for k in urls)
            # The 6-hour line is only true for paste.rs; the privacy notice above already states
--- a/hermes_cli/debug.py
+++ b/hermes_cli/debug.py
@@ -6,7 +6,6 @@ import gzip
 import io
 import json
 import logging
-import re
 import sys
 import time
 import urllib.request
@@ -16,6 +15,11 @@ from types import SimpleNamespace
 from typing import Optional

 from hermes_constants import get_hermes_home
+from hermes_cli.debug_redaction import (
+    redact_debug_support_error,
+    redact_debug_support_text,
+    redact_debug_support_value,
+)
 from utils import atomic_json_write

 logger = logging.getLogger(__name__)
@@ -24,10 +28,6 @@ logger = logging.getLogger(__name__)
 _REDACTION_BANNER = (
    "[hermes debug share: log content redacted at upload time. "
    "run with --no-redact to disable]\n")
-_EMAIL_ADDRESS_RE = re.compile(
-    r"(?<![A-Za-z0-9._%+-])"
-    r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"
-    r"(?![A-Za-z0-9._%+-])")
 _PASTE_RS_URL = "https://paste.rs/"  # primary; dpaste.com is the fallback
 _DPASTE_COM_URL = "https://dpaste.com/api/"
 _USER_AGENT = "hermes-agent/debug-share"
@@ -177,7 +177,8 @@ def _post_paste(service: str, endpoint: str, body: bytes, content_type: str) ->
    with urllib.request.urlopen(req, timeout=30) as resp:
        url = resp.read().decode("utf-8").strip()
    if not url.startswith("http"):
-        raise ValueError(f"Unexpected response from {service}: {url[:200]}")
+        safe_response = redact_debug_support_text(url, max_chars=200)
+        raise ValueError(f"Unexpected response from {service}: {safe_response}")
    return url


@@ -208,7 +209,7 @@ def upload_to_pastebin(content: str, expiry_days: int = 1) -> str:
        try:
            return upload()
        except Exception as exc:
-            errors.append(f"{service}: {exc}")
+            errors.append(f"{service}: {redact_debug_support_error(exc)}")
    raise RuntimeError("Failed to upload to any paste service:\n  " + "\n  ".join(errors))


@@ -257,27 +258,21 @@ def _resolve_log_path(log_name: str) -> Optional[Path]:


 def _redact_log_text(text: str, *, redact_url_credentials: bool = True) -> str:
-    """``redact_sensitive_text(force=True)`` + email scrub — fires regardless of the operator's
-    ``security.redact_secrets`` setting; only the in-memory upload copy is sanitized. URL
-    credentials (``?token=``, ``user:pass@``) are masked too by default: log lines keep them, since
-    default redaction spares OAuth/magic-link URLs, but nothing follows a link out of an uploaded
-    report."""
-    if not text:
-        return text
-    from agent.redact import redact_sensitive_text
-    text = redact_sensitive_text(
-        text, force=True, redact_url_credentials=redact_url_credentials
-    )
-    return _EMAIL_ADDRESS_RE.sub("[REDACTED_EMAIL]", text)
+    """Strict support-egress scrub; the compatibility flag is intentionally ignored.
+
+    Debug/support payloads always use strict URL handling. Ordinary tool output
+    retains its separate, less destructive URL policy in ``agent.redact``.
+    """
+    return redact_debug_support_text(text)


 def _read_tail_bytes(
-    log_path: Path, size: int, max_bytes: int, tail_lines: int) -> tuple[bytes, bool]:
-    """Whole file, or (oversized) a backwards read holding ``max_bytes`` for the full upload AND
-    enough newlines for the summary tail from the same snapshot → (raw, truncated)."""
+    log_path: Path, size: int, max_bytes: int, tail_lines: int
+) -> tuple[bytes, int, bool]:
+    """Return a bounded tail, its file offset, and whether it starts on a line boundary."""
    with open(log_path, "rb") as f:
        if size <= max_bytes:
-            return f.read(), False
+            return f.read(), 0, True
        chunk_size = 8192
        pos = size
        chunks: list[bytes] = []
@@ -293,7 +288,24 @@ def _read_tail_bytes(
            total += len(chunk)
            newline_count += chunk.count(b"\n")
            chunk_size = min(chunk_size * 2, 65536)
-        return b"".join(chunks), pos > 0
+        starts_on_boundary = True
+        if pos > 0:
+            f.seek(pos - 1)
+            starts_on_boundary = f.read(1) == b"\n"
+        return b"".join(chunks), pos, starts_on_boundary
+
+
+def _truncate_complete_lines(text: str, max_bytes: int) -> tuple[str, bool]:
+    """Keep at most the last ``max_bytes`` without publishing a partial first line."""
+    raw = text.encode("utf-8")
+    if len(raw) <= max_bytes:
+        return text, False
+    cut = len(raw) - max_bytes
+    on_boundary = raw[cut - 1:cut] == b"\n"
+    kept = raw[cut:]
+    if not on_boundary:
+        kept = kept.split(b"\n", 1)[1] if b"\n" in kept else b""
+    return kept.decode("utf-8", errors="replace"), True


 def _capture_log_snapshot(
@@ -311,26 +323,25 @@ def _capture_log_snapshot(
        size = log_path.stat().st_size
        if size == 0:  # truncated between _resolve_log_path and stat
            return LogSnapshot(path=log_path, tail_text="(file empty)", full_text=None)
-        raw, truncated = _read_tail_bytes(log_path, size, max_bytes, tail_lines)
-        full_raw = raw
-        if truncated and len(full_raw) > max_bytes:
-            cut = len(full_raw) - max_bytes
-            # Drop a partial first line only when the cut lands genuinely mid-line.
-            on_boundary = cut > 0 and full_raw[cut - 1 : cut] == b"\n"
-            full_raw = full_raw[cut:]
-            if not on_boundary and b"\n" in full_raw:
-                full_raw = full_raw.split(b"\n", 1)[1]
+        raw, start_offset, starts_on_boundary = _read_tail_bytes(
+            log_path, size, max_bytes, tail_lines
+        )
        all_text = raw.decode("utf-8", errors="replace")
+        if redact:
+            # The support scrub needs the complete logical field before either
+            # the summary-line cap or full-log byte cap can remove its key.
+            all_text = _redact_log_text(all_text)
+        if start_offset and not starts_on_boundary:
+            all_text = all_text.split("\n", 1)[1] if "\n" in all_text else ""
        tail_text = "".join(all_text.splitlines(keepends=True)[-tail_lines:]).rstrip("\n")
-        full_text = full_raw.decode("utf-8", errors="replace")
+        full_text, capped = _truncate_complete_lines(all_text, max_bytes)
+        truncated = bool(start_offset or capped or size > max_bytes)
        if truncated:
            full_text = f"[... truncated — showing last ~{max_bytes // 1024}KB ...]\n{full_text}"
-        if redact:
-            tail_text = _redact_log_text(tail_text)
-            full_text = _redact_log_text(full_text)
        return LogSnapshot(path=log_path, tail_text=tail_text, full_text=full_text)
    except Exception as exc:
-        return LogSnapshot(path=log_path, tail_text=f"(error reading: {exc})", full_text=None)
+        error = redact_debug_support_error(exc)
+        return LogSnapshot(path=log_path, tail_text=f"(error reading: {error})", full_text=None)


 # Logs the debug report tails, in output order. ``agent`` gets the full ``--lines`` budget;
@@ -418,15 +429,20 @@ def collect_share_bundle(log_lines: int = 200, redact: bool = True) -> dict[str,
        if full := log_snapshots[name].full_text:
            filename = LOG_FILES.get(name, f"{name}.log")
            bundle[filename] = banner + dump_text + f"\n\n--- full {filename} ---\n" + full
+    if redact:
+        # Final textual boundary: future bundle members cannot bypass the
+        # shared support scrub merely because their collector forgot it.
+        bundle = {label: _redact_log_text(text) for label, text in bundle.items()}
    return bundle


 def build_nous_bundle(bundle: dict[str, str], redact: bool = True) -> bytes:
    """Gzip a :func:`collect_share_bundle` mapping into the Nous envelope (shape parsed by the
    discord-support viewer — keep it stable)."""
+    files = redact_debug_support_value(bundle) if redact else bundle
    envelope = {"format": _NOUS_BUNDLE_FORMAT, "redacted": bool(redact),
                "created": datetime.datetime.now(datetime.timezone.utc).isoformat(),
-                "files": bundle}
+                "files": files}
    return gzip.compress(json.dumps(envelope).encode("utf-8"))


@@ -475,7 +491,7 @@ def build_debug_share(
        try:
            urls[label] = upload_to_pastebin(content, expiry_days=expiry)
        except Exception as exc:
-            failures.append(f"{label}: {exc}")
+            failures.append(f"{label}: {redact_debug_support_error(exc)}")
    _schedule_auto_delete(list(urls.values()))
    return DebugShareResult(urls=urls, failures=failures, redacted=redact,
                            auto_delete_seconds=_AUTO_DELETE_SECONDS, report=report)
@@ -537,7 +553,7 @@ def run_debug_share(args):
    try:
        result = build_debug_share(log_lines=log_lines, expiry=expiry, redact=redact)
    except RuntimeError as exc:
-        print(f"\nUpload failed: {exc}", file=sys.stderr)
+        print(f"\nUpload failed: {redact_debug_support_error(exc)}", file=sys.stderr)
        print("\nRun `hermes debug share --local` to print the report instead.\n")
        sys.exit(1)
    label_width = max(len(k) for k in result.urls)
@@ -594,7 +610,8 @@ def _run_debug_share_nous(args, *, log_lines: int, redact: bool) -> None:
    try:
        res = share_to_nous(build_nous_bundle(bundle, redact=redact))
    except Exception as exc:
-        print(f"\nNous upload failed: {exc}\n"
+        error = redact_debug_support_error(exc)
+        print(f"\nNous upload failed: {error}\n"
              "\nThe Nous diagnostics service may be unavailable or not yet provisioned.\n"
              "Run `hermes debug share --local` to print the report instead, "
              "or `hermes debug share` to upload to a public paste service.\n", file=sys.stderr)
@@ -626,11 +643,15 @@ def run_debug_delete(args):
            if delete_paste(url):
                print(f"  ✓ Deleted: {url}")
            else:
-                print(f"  ✗ Failed to delete: {url} (unexpected response)")
+                print(
+                    "  ✗ Failed to delete: "
+                    f"{redact_debug_support_text(url)} (unexpected response)"
+                )
        except ValueError as exc:
-            print(f"  ✗ {exc}")
+            print(f"  ✗ {redact_debug_support_error(exc)}")
        except Exception as exc:
-            print(f"  ✗ Could not delete {url}: {exc}")
+            safe_url = redact_debug_support_text(url)
+            print(f"  ✗ Could not delete {safe_url}: {redact_debug_support_error(exc)}")


 def run_debug(args):
--- a/hermes_cli/debug_redaction.py
+++ b/hermes_cli/debug_redaction.py
@@ -0,0 +1,339 @@
+"""Fail-closed redaction for text and structures leaving debug/support flows."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping, Sequence
+from typing import Any
+from urllib.parse import unquote_plus
+
+_REDACTED = "[REDACTED]"
+_EMAIL_ADDRESS_RE = re.compile(
+    r"(?<![A-Za-z0-9._%+-])"
+    r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"
+    r"(?![A-Za-z0-9._%+-])"
+)
+
+# Support uploads are a strict opt-in surface. Compact names cover snake_case,
+# kebab-case and camelCase without widening ordinary tool-output redaction.
+_SENSITIVE_URL_PARAM_NAMES = frozenset(
+    {
+        "accesstoken",
+        "refreshtoken",
+        "idtoken",
+        "authtoken",
+        "token",
+        "apikey",
+        "clientsecret",
+        "password",
+        "auth",
+        "jwt",
+        "session",
+        "secret",
+        "key",
+        "code",
+        "signature",
+        "xamzsignature",
+        "xgoogsignature",
+        "sig",
+        "privatekey",
+        "secretkey",
+        "xamzsecuritytoken",
+        "securitytoken",
+    }
+)
+_RAW_URL_PARAM_RE = re.compile(
+    r"(?P<sep>[?#&;])(?P<key>[A-Za-z0-9_.~+%\-]+)="
+    r"(?P<value>[^?#&;\s\"'<>]*)"
+)
+_ENCODED_URL_PARAM_RE = re.compile(
+    r"(?P<sep>%3[fF]|%23|%26|%3[bB])"
+    r"(?P<key>(?:(?!%3[dD])[^\s\"'<>]){1,160}?)"
+    r"(?P<eq>%3[dD]|=)"
+    r"(?P<value>.*?)"
+    r"(?=(?:%3[fF]|%26|%23|%3[bB])|[\s\"'<>]|$)",
+    re.IGNORECASE,
+)
+
+_SENSITIVE_HEADER_NAMES = (
+    "authorization",
+    "proxy-authorization",
+    "cookie",
+    "set-cookie",
+    "x-api-key",
+    "x-goog-api-key",
+    "api-key",
+    "apikey",
+    "x-api-token",
+    "x-auth-token",
+    "x-access-token",
+)
+_SENSITIVE_HEADER_COMPACT_NAMES = frozenset(
+    "".join(ch for ch in name.casefold() if ch.isalnum())
+    for name in _SENSITIVE_HEADER_NAMES
+)
+_HEADER_NAME_PATTERN = "|".join(re.escape(name) for name in _SENSITIVE_HEADER_NAMES)
+_QUOTED_HEADER_RE = re.compile(
+    rf"(?P<prefix>['\"]?(?:{_HEADER_NAME_PATTERN})['\"]?\s*[:=]\s*)"
+    r"(?P<quote>['\"])(?P<value>.*?)(?P=quote)",
+    re.IGNORECASE,
+)
+_BARE_HEADER_RE = re.compile(
+    rf"(?P<prefix>\b(?:{_HEADER_NAME_PATTERN})\b\s*[:=]\s*)"
+    r"(?P<value>(?:(?:Bearer|Basic|Digest)\s+)?[^\s,}\]]+)",
+    re.IGNORECASE,
+)
+
+_SECRET_ARG_FLAG = (
+    r"--(?:api[-_]?key|access[-_]?token|refresh[-_]?token|id[-_]?token|"
+    r"auth[-_]?token|client[-_]?secret|private[-_]?key|secret[-_]?key|"
+    r"password|passwd|credential|token|secret)"
+)
+_QUOTED_ARGV_RE = re.compile(
+    rf"(?P<flag_quote>['\"])(?P<flag>{_SECRET_ARG_FLAG})(?P=flag_quote)"
+    r"(?P<sep>\s*,\s*)"
+    r"(?P<value_quote>['\"])(?P<value>.*?)(?P=value_quote)",
+    re.IGNORECASE,
+)
+_ARG_EQUALS_RE = re.compile(
+    rf"(?P<flag>{_SECRET_ARG_FLAG})(?P<sep>\s*=\s*)"
+    r"(?P<quote>['\"]?)(?P<value>[^\s,}\]'\"]+)(?P=quote)",
+    re.IGNORECASE,
+)
+_ARG_OPERAND_RE = re.compile(
+    rf"(?P<flag>{_SECRET_ARG_FLAG})(?P<sep>\s+)"
+    r"(?P<quote>['\"]?)(?P<value>[^\s,}\]'\"]+)(?P=quote)",
+    re.IGNORECASE,
+)
+_HEADER_ARGV_RE = re.compile(
+    r"(?P<head_quote>['\"])--header(?P=head_quote)\s*,\s*"
+    rf"(?P<name_quote>['\"])(?P<name>{_HEADER_NAME_PATTERN})(?P=name_quote)\s*,\s*"
+    r"(?:(?P<scheme_quote>['\"])(?P<scheme>Bearer|Basic|Digest)(?P=scheme_quote)\s*,\s*)?"
+    r"(?P<value_quote>['\"])(?P<value>.*?)(?P=value_quote)",
+    re.IGNORECASE,
+)
+_PLAIN_HEADER_ARG_RE = re.compile(
+    rf"(?P<prefix>--header\s+(?:{_HEADER_NAME_PATTERN})\s+)"
+    r"(?:(?:Bearer|Basic|Digest)\s+)?(?P<value>[^\s,}\]]+)",
+    re.IGNORECASE,
+)
+_SUPPORT_BEARER_RE = re.compile(
+    r"(?P<prefix>\bBearer\s+)[A-Za-z0-9._~+/-]{8,}=*",
+    re.IGNORECASE,
+)
+
+_SECRET_FIELD_COMPACT_NAMES = frozenset(
+    {
+        "apikey",
+        "token",
+        "accesstoken",
+        "refreshtoken",
+        "idtoken",
+        "authtoken",
+        "clientsecret",
+        "privatekey",
+        "secretkey",
+        "password",
+        "passwd",
+        "credential",
+        "credentials",
+        "authorization",
+        "bearer",
+        "keymaterial",
+        "rawsecret",
+        "secretvalue",
+        "secretinput",
+    }
+)
+_SECRET_ENV_PARTS = (
+    "apikey",
+    "token",
+    "secret",
+    "password",
+    "passwd",
+    "credential",
+    "authorization",
+    "privatekey",
+    "accesskey",
+)
+_SECRET_ARG_FLAGS = frozenset(
+    {
+        "apikey",
+        "accesstoken",
+        "refreshtoken",
+        "idtoken",
+        "authtoken",
+        "clientsecret",
+        "privatekey",
+        "secretkey",
+        "password",
+        "passwd",
+        "credential",
+        "token",
+        "secret",
+    }
+)
+
+
+def _decoded_compact_name(value: object) -> str:
+    text = str(value or "")
+    for _ in range(3):
+        decoded = unquote_plus(text)
+        if decoded == text:
+            break
+        text = decoded
+    return "".join(ch for ch in text.casefold() if ch.isalnum())
+
+
+def _redact_url_params(text: str) -> str:
+    def _raw_sub(match: re.Match[str]) -> str:
+        if _decoded_compact_name(match.group("key")) not in _SENSITIVE_URL_PARAM_NAMES:
+            return match.group(0)
+        return f"{match.group('sep')}{match.group('key')}=***"
+
+    def _encoded_sub(match: re.Match[str]) -> str:
+        if _decoded_compact_name(match.group("key")) not in _SENSITIVE_URL_PARAM_NAMES:
+            return match.group(0)
+        replacement = "%2A%2A%2A" if match.group("eq").lower() == "%3d" else "***"
+        return f"{match.group('sep')}{match.group('key')}{match.group('eq')}{replacement}"
+
+    # Running until stable handles a raw URL nested inside another query value
+    # without recursive parsing. The pass cap keeps malformed input bounded.
+    for _ in range(4):
+        redacted = _ENCODED_URL_PARAM_RE.sub(_encoded_sub, _RAW_URL_PARAM_RE.sub(_raw_sub, text))
+        if redacted == text:
+            break
+        text = redacted
+    return text
+
+
+def _redact_headers_and_argv(text: str) -> str:
+    text = _QUOTED_HEADER_RE.sub(
+        lambda match: f"{match.group('prefix')}{match.group('quote')}{_REDACTED}{match.group('quote')}",
+        text,
+    )
+    text = _BARE_HEADER_RE.sub(lambda match: f"{match.group('prefix')}{_REDACTED}", text)
+    text = _HEADER_ARGV_RE.sub(
+        lambda match: (
+            f"{match.group('head_quote')}--header{match.group('head_quote')}, "
+            f"{match.group('name_quote')}{match.group('name')}{match.group('name_quote')}, "
+            + (
+                f"{match.group('scheme_quote')}{match.group('scheme')}{match.group('scheme_quote')}, "
+                if match.group("scheme")
+                else ""
+            )
+            + f"{match.group('value_quote')}{_REDACTED}{match.group('value_quote')}"
+        ),
+        text,
+    )
+    text = _QUOTED_ARGV_RE.sub(
+        lambda match: (
+            f"{match.group('flag_quote')}{match.group('flag')}{match.group('flag_quote')}"
+            f"{match.group('sep')}{match.group('value_quote')}{_REDACTED}{match.group('value_quote')}"
+        ),
+        text,
+    )
+    text = _ARG_EQUALS_RE.sub(
+        lambda match: f"{match.group('flag')}{match.group('sep')}{match.group('quote')}{_REDACTED}{match.group('quote')}",
+        text,
+    )
+    text = _ARG_OPERAND_RE.sub(
+        lambda match: f"{match.group('flag')}{match.group('sep')}{match.group('quote')}{_REDACTED}{match.group('quote')}",
+        text,
+    )
+    text = _PLAIN_HEADER_ARG_RE.sub(
+        lambda match: f"{match.group('prefix')}{_REDACTED}", text
+    )
+    return _SUPPORT_BEARER_RE.sub(
+        lambda match: f"{match.group('prefix')}{_REDACTED}", text
+    )
+
+
+def redact_debug_support_text(value: object, *, max_chars: int | None = None) -> str:
+    """Strictly scrub one debug/support value, then apply an optional character cap.
+
+    This is deliberately stricter than ordinary model/tool output: support text is
+    leaving the process for a human or remote diagnostics store. Any redactor
+    failure returns the shared fail-closed sentinel rather than the original.
+    """
+    text = "" if value is None else str(value)
+    if not text:
+        return text
+    try:
+        from agent.redact import REDACTION_UNAVAILABLE, redact_for_egress, redact_sensitive_text
+
+        text = redact_for_egress(text)
+        if text == REDACTION_UNAVAILABLE:
+            return text
+        text = redact_sensitive_text(
+            text, force=True, redact_url_credentials=True
+        )
+        text = _redact_url_params(text)
+        text = _redact_headers_and_argv(text)
+        text = _EMAIL_ADDRESS_RE.sub("[REDACTED_EMAIL]", text)
+    except Exception:
+        return "[redaction-unavailable]"
+    return text[:max_chars] if max_chars is not None else text
+
+
+def redact_debug_support_error(error: object) -> str:
+    """Render an outward-facing debug/support failure without raw exception data."""
+    return redact_debug_support_text(error)
+
+
+def _is_secret_field(key: object, *, parent: str) -> bool:
+    compact = _decoded_compact_name(key)
+    if compact in _SECRET_FIELD_COMPACT_NAMES:
+        return True
+    if parent in {"headers", "extraheaders"}:
+        return compact in _SENSITIVE_HEADER_COMPACT_NAMES
+    if parent == "env":
+        return any(part in compact for part in _SECRET_ENV_PARTS)
+    try:
+        from agent.redact import is_secret_field_name
+
+        return is_secret_field_name(key)
+    except Exception:
+        return True
+
+
+def _redact_sequence(value: Sequence[Any]) -> list[Any]:
+    items = list(value)
+    output = [redact_debug_support_value(item) for item in items]
+    for index, item in enumerate(items):
+        compact = _decoded_compact_name(item) if isinstance(item, str) else ""
+        if compact in _SECRET_ARG_FLAGS and index + 1 < len(output):
+            output[index + 1] = _REDACTED
+        if compact != "header" or not str(item).lstrip().startswith("--") or index + 1 >= len(items):
+            continue
+        header = str(items[index + 1])
+        if ":" in header:
+            output[index + 1] = redact_debug_support_text(header)
+            continue
+        if _decoded_compact_name(header) not in _SENSITIVE_HEADER_COMPACT_NAMES:
+            continue
+        operand = index + 2
+        if operand < len(items) and str(items[operand]).casefold() in {"bearer", "basic", "digest"}:
+            operand += 1
+        if operand < len(output):
+            output[operand] = _REDACTED
+    return output
+
+
+def redact_debug_support_value(value: Any, *, _parent: str = "") -> Any:
+    """Recursively scrub mappings and argv-like sequences for support serialization."""
+    if isinstance(value, Mapping):
+        output: dict[Any, Any] = {}
+        for key, child in value.items():
+            if _is_secret_field(key, parent=_parent):
+                output[key] = _REDACTED
+            else:
+                output[key] = redact_debug_support_value(
+                    child, _parent=_decoded_compact_name(key)
+                )
+        return output
+    if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
+        return _redact_sequence(value)
+    if isinstance(value, str):
+        return redact_debug_support_text(value)
+    return value
--- a/hermes_cli/diagnostics_upload.py
+++ b/hermes_cli/diagnostics_upload.py
@@ -7,6 +7,8 @@ import json
 import os
 import urllib.request

+from hermes_cli.debug_redaction import redact_debug_support_error, redact_debug_support_text
+
 # Overridable via env so the feature can be pointed at staging / a local dev NAS instance.
 NAS_BASE = os.environ.get("HERMES_DIAGNOSTICS_BASE_URL", "https://portal.nousresearch.com")
 _REQUEST_TIMEOUT = 30
@@ -38,9 +40,11 @@ def request_upload_url(content_type: str = "application/gzip", size_bytes: int |
    try:
        result = json.loads(body)
    except (ValueError, json.JSONDecodeError) as exc:
-        raise RuntimeError(f"diagnostics upload-url returned non-JSON response: {body[:200]}") from exc
+        safe_body = redact_debug_support_text(body, max_chars=200)
+        raise RuntimeError(f"diagnostics upload-url returned non-JSON response: {safe_body}") from exc
    if not isinstance(result, dict) or not result.get("uploadUrl"):
-        raise RuntimeError(f"diagnostics upload-url response missing 'uploadUrl': {body[:200]}")
+        safe_body = redact_debug_support_text(body, max_chars=200)
+        raise RuntimeError(f"diagnostics upload-url response missing 'uploadUrl': {safe_body}")
    return result


@@ -53,6 +57,9 @@ def put_bundle(upload_url: str, data: bytes, content_type: str = "application/gz

 def share_to_nous(report_bundle: bytes) -> dict:
    """Mint a presigned PUT URL (with the exact ``sizeBytes`` NAS signs), then PUT *report_bundle*."""
-    info = request_upload_url(content_type="application/gzip", size_bytes=len(report_bundle))
-    put_bundle(info["uploadUrl"], report_bundle, content_type="application/gzip")
-    return info
+    try:
+        info = request_upload_url(content_type="application/gzip", size_bytes=len(report_bundle))
+        put_bundle(info["uploadUrl"], report_bundle, content_type="application/gzip")
+        return info
+    except Exception as exc:
+        raise RuntimeError(redact_debug_support_error(exc)) from exc
--- a/hermes_cli/dump.py
+++ b/hermes_cli/dump.py
@@ -180,25 +180,50 @@ def _config_overrides(config: dict) -> dict[str, str]:
        overrides["toolsets"] = str(user_toolsets)
    fallbacks = config.get("fallback_providers", [])
    if fallbacks:
-        # Entries carry api_key; the dump is made to be pasted, so it never prints a secret.
-        from agent.redact import redact_sensitive_text
-        overrides["fallback_providers"] = redact_sensitive_text(
-            str(_mask_fallback_keys(fallbacks)), force=True, redact_url_credentials=True
+        from hermes_cli.debug_redaction import redact_debug_support_text
+
+        overrides["fallback_providers"] = redact_debug_support_text(
+            str(_fallback_debug_projection(fallbacks))
        )
    return overrides


-def _mask_fallback_keys(value):
-    """Mask secret fields (``api_key``, ``token``, ``password``, ... per ``agent.redact``) by field, not
-    by text: the runtime ``str()``s whatever value is there (a YAML int, a mapping, a key containing
-    ``***``), and text redaction only masks quoted strings that don't already look masked."""
-    from agent.redact import is_secret_field_name
+_FALLBACK_DEBUG_FIELDS = (
+    "provider",
+    "model",
+    "base_url",
+    "api_mode",
+    "key_env",
+    "api_key_env",
+    "max_tokens",
+)

-    if isinstance(value, dict):
-        return {k: "***" if v and is_secret_field_name(k) else _mask_fallback_keys(v) for k, v in value.items()}
-    if isinstance(value, list):
-        return [_mask_fallback_keys(v) for v in value]
-    return value
+
+def _fallback_debug_projection(value) -> list[dict]:
+    """Render only route metadata useful to support; omit arbitrary containers.
+
+    Fallback entries are open mappings consumed by provider-specific code, so a
+    denylist cannot make their ``headers``, ``env``, command/argv, or future
+    fields safe to serialize. Credential pointer names remain useful and carry
+    no values; an inline key is represented only as a source status.
+    """
+    entries = [value] if isinstance(value, dict) else value if isinstance(value, list) else []
+    projected: list[dict] = []
+    for entry in entries:
+        if not isinstance(entry, dict):
+            continue
+        item: dict = {}
+        for key in _FALLBACK_DEBUG_FIELDS:
+            child = entry.get(key)
+            if key == "max_tokens":
+                if isinstance(child, (int, float)) and not isinstance(child, bool):
+                    item[key] = child
+            elif isinstance(child, str):
+                item[key] = child
+        if entry.get("api_key"):
+            item["credential_source"] = "inline"
+        projected.append(item)
+    return projected


 # (env var, dump label) in display order.
--- a/hermes_cli/web_routers/status.py
+++ b/hermes_cli/web_routers/status.py
@@ -796,6 +796,7 @@ async def run_debug_share_endpoint(body: DebugShareRequest | None = None,
    unlike the other diagnostics actions: the point is the shareable URLs, returned as a
    structured payload the dashboard renders as copyable links."""
    from hermes_cli.debug import build_debug_share
+    from hermes_cli.debug_redaction import redact_debug_support_error
    req = body or DebugShareRequest()
    try:
        result = await config_scoped_to_thread(profile, lambda: build_debug_share(
@@ -804,10 +805,12 @@ async def run_debug_share_endpoint(body: DebugShareRequest | None = None,
        raise  # an unknown ?profile= is the scope's 404, not a failed share
    except RuntimeError as exc:
        # Required summary-report upload failed (offline / paste service down).
-        raise HTTPException(status_code=502, detail=f"Upload failed: {exc}")
+        error = redact_debug_support_error(exc)
+        raise HTTPException(status_code=502, detail=f"Upload failed: {error}")
    except Exception as exc:
        _log.exception("debug share failed")
-        raise HTTPException(status_code=500, detail=f"Failed: {exc}")
+        error = redact_debug_support_error(exc)
+        raise HTTPException(status_code=500, detail=f"Failed: {error}")

    return {"ok": True, "urls": result.urls, "failures": result.failures,
            "redacted": result.redacted, "auto_delete_seconds": result.auto_delete_seconds}
--- a/tests/hermes_cli/test_debug.py
+++ b/tests/hermes_cli/test_debug.py
@@ -287,6 +287,38 @@ class TestCaptureLogSnapshotRedaction:
        assert _REDACT_FIXTURE_TOKEN not in snaps["agent"].tail_text
        assert _REDACT_FIXTURE_TOKEN not in (snaps["agent"].full_text or "")

+    def test_redacts_complete_log_lines_before_applying_byte_limits(
+        self, hermes_home_with_secret
+    ):
+        from hermes_cli.debug import _capture_log_snapshot
+
+        max_bytes = 512
+        raw_window = 8192
+        tail_canary = "r36_REDACTED_tail_boundary_canary_012345"
+        full_canary = "r36_REDACTED_full_boundary_canary_678901"
+        payload = bytearray(b"x" * 12_000)
+
+        # The backwards-read window and the final full-log window each begin
+        # inside a credential field. Truncating first drops ``accessToken=``
+        # while retaining its complete value, which the redactor cannot then
+        # classify. A redacted share must either scrub the complete logical
+        # line before either cap or omit that partial line.
+        for boundary, canary in (
+            (len(payload) - raw_window, tail_canary),
+            (len(payload) - max_bytes, full_canary),
+        ):
+            marker = f"accessToken={canary}".encode()
+            start = boundary - len("accessToken=")
+            payload[start:start + len(marker)] = marker
+
+        (hermes_home_with_secret / "logs" / "agent.log").write_bytes(payload)
+
+        snap = _capture_log_snapshot("agent", tail_lines=1, max_bytes=max_bytes)
+
+        assert tail_canary not in snap.tail_text
+        assert snap.full_text is not None
+        assert full_canary not in snap.full_text
+


 # ---------------------------------------------------------------------------
--- a/tests/tui_gateway/test_diagnostics_share_nous.py
+++ b/tests/tui_gateway/test_diagnostics_share_nous.py
@@ -97,6 +97,67 @@ def test_share_nous_client_text_gets_upload_safe_log_redaction(captured_upload):
    assert secret not in files["client/desktop.log"]


+def test_redacted_support_egress_scrubs_structured_values_and_errors(
+    captured_upload, monkeypatch
+):
+    from hermes_constants import get_hermes_home
+
+    canaries = {
+        "url": "r36_REDACTED_url_canary_0123456789",
+        "header": "r36_REDACTED_header_canary_0123456789",
+        "env": "r36_REDACTED_env_canary_0123456789",
+        "argv": "r36_REDACTED_argv_canary_0123456789",
+        "bearer": "r36_REDACTED_bearer_canary_0123456789",
+        "error": "r36_REDACTED_error_canary_0123456789",
+    }
+    home = get_hermes_home()
+    home.mkdir(parents=True, exist_ok=True)
+    (home / "config.yaml").write_text(
+        "fallback_providers:\n"
+        "  - provider: custom\n"
+        "    model: support-fixture\n"
+        "    base_url: 'https://outer.invalid/?next=https://inner.invalid/"
+        f"?accessToken={canaries['url']}'\n"
+        "    extra_headers:\n"
+        f"      X-API-Key: {canaries['header']}\n"
+        "    env:\n"
+        f"      AWS_SECRET_ACCESS_KEY: {canaries['env']}\n"
+        f"    args: ['--api-key', '{canaries['argv']}']\n",
+        encoding="utf-8",
+    )
+    structured = (
+        f"headers={{'Proxy-Authorization': 'Basic {canaries['header']}'}} "
+        f"Command ['provider', '--api-key', '{canaries['argv']}'] "
+        f"Bearer {canaries['bearer']}"
+    )
+
+    result = _handler()(
+        "rid-structured",
+        {"error_context": structured, "extra_files": {"desktop.log": structured}},
+    )
+
+    assert result["result"]["ok"] is True
+    envelope = _envelope(captured_upload["blob"])
+    assert envelope["redacted"] is True
+    rendered = json.dumps(envelope, sort_keys=True)
+    assert "support-fixture" in rendered
+    assert not [canary for canary in canaries.values() if canary in rendered]
+
+    import hermes_cli.diagnostics_upload as du
+
+    def _fail(_blob: bytes) -> dict:
+        raise RuntimeError(
+            "PUT https://upload.invalid/?X-Amz-Security-Token="
+            f"{canaries['error']} headers={{'X-API-Key': '{canaries['header']}'}}"
+        )
+
+    monkeypatch.setattr(du, "share_to_nous", _fail)
+    failure = _handler()("rid-structured-error", {})["result"]
+    assert failure["ok"] is False
+    assert canaries["error"] not in failure["error"]
+    assert canaries["header"] not in failure["error"]
+
+
 def test_share_nous_linkless_success_is_a_failure(monkeypatch):
    """ok:true with neither view_url nor id would strand the user with an
    unreferencable upload — surface it as a structured failure instead."""
--- a/tui_gateway/methods_config.py
+++ b/tui_gateway/methods_config.py
@@ -497,23 +497,27 @@ def _(rid, params: dict) -> dict:
    upload failures render inline. Optional: ``error_context`` (-> ``error-context.txt``),
    ``extra_files`` ({label -> text}), ``log_lines`` (default 200); all force-redacted."""
    try:
-        from hermes_cli.debug import _redact_log_text, build_nous_bundle, collect_share_bundle
+        from hermes_cli.debug import build_nous_bundle, collect_share_bundle
+        from hermes_cli.debug_redaction import redact_debug_support_text
        from hermes_cli.diagnostics_upload import share_to_nous
        log_lines = params.get("log_lines")
        if not isinstance(log_lines, int) or not (10 <= log_lines <= 2000):
            log_lines = 200
        bundle = collect_share_bundle(log_lines=log_lines, redact=True)
-        # Client text goes through the SAME upload-safe redactor as backend logs (force secret
-        # redaction + email masking), never the weaker bare secret pass.
+        # Redact complete client values before applying their support-upload caps.
        error_context = params.get("error_context")
        if isinstance(error_context, str) and error_context.strip():
-            bundle["error-context.txt"] = _redact_log_text(error_context.strip()[:8_000])
+            bundle["error-context.txt"] = redact_debug_support_text(
+                error_context.strip(), max_chars=8_000
+            )
        # Bounded: at most 4 files, 512KB each, sanitized labels — not an arbitrary upload surface.
        extra_files = params.get("extra_files")
        for label, text in list(extra_files.items())[:4] if isinstance(extra_files, dict) else ():
            safe_label = _safe_client_label(label) if isinstance(label, str) else ""
            if safe_label and isinstance(text, str) and text.strip():
-                bundle[f"client/{safe_label}"] = _redact_log_text(text[:524_288])
+                bundle[f"client/{safe_label}"] = redact_debug_support_text(
+                    text, max_chars=524_288
+                )
        res = share_to_nous(build_nous_bundle(bundle, redact=True))
        view_url = res.get("viewUrl") or res.get("view_url")
        upload_id = res.get("id")
@@ -522,7 +526,9 @@ def _(rid, params: dict) -> dict:
        return _ok(rid, {"ok": True, "view_url": view_url, "upload_id": upload_id,
                         "expires_at": res.get("expiresAt") or res.get("expires_at")})
    except Exception as e:
-        return _ok(rid, {"ok": False, "error": str(e)})
+        from hermes_cli.debug_redaction import redact_debug_support_error
+
+        return _ok(rid, {"ok": False, "error": redact_debug_support_error(e)})


 def register(server) -> None: