atlas-iac/scripts/ops/hermes_handoff_redaction.py

#!/usr/bin/env python3
"""Credential redaction and output bounding for the handoff acceptance harness.

Every string the harness records — command output, failure reasons, evidence
values — passes through :func:`redact` before it reaches a report. Screening is
a fail-closed accident barrier, not proof that text is secret-free: a short
secret under an innocuous key is not reliably distinguishable from prose. The
harness therefore also never asks for a secret in the first place. Probes read
names, shapes, and status fields; they never read environment values, token
files, private keys, cookies, or Vault responses.

Bounding is part of the same barrier. Screening always precedes truncation so a
credential that crosses the byte boundary cannot leave a useful prefix behind.
The process runner additionally discards all captured bytes when a live stream
exceeds its cap; the fixed truncation marker is the only recorded output.
"""

from __future__ import annotations

import re
from typing import Any

REDACTED = "[redacted]"
DEFAULT_MAX_BYTES = 64 * 1024
TRUNCATION_NOTE = "[truncated]"

# Keys whose value is never safe to record, matched against a normalised
# (lowercased, separator-stripped) form of the key.
SENSITIVE_KEY_SUFFIXES = (
    "accesskey",
    "accesskeyid",
    "apikey",
    "authtoken",
    "clientsecret",
    "connectionstring",
    "cookie",
    "credential",
    "credentials",
    "passphrase",
    "passwd",
    "password",
    "privatekey",
    "secret",
    "secretkey",
    "sessionkey",
    "signature",
    "signingkey",
    "token",
    "webhook",
)
SENSITIVE_KEY_EXACT = frozenset(
    {
        "auth",
        "authorization",
        "bearer",
        "pat",
        "pw",
        "sas",
        "setcookie",
    }
)
# A bare `key:` is a structural field name across Kubernetes and TOML far more
# often than it is a credential, so it is screened by value shape and by the
# token patterns rather than by name. The compound forms above still match.

# Ordinary prose values shorter than this threshold are not independently
# treated as secret-shaped. Values under sensitive keys are screened at every
# length, including numeric values.
MIN_SECRET_LENGTH = 8
NON_SECRET_LITERALS = frozenset(
    {"true", "false", "null", "none", "yes", "no", "0", "1", ""}
)
_KEY_NOISE_RE = re.compile(r"[^a-z0-9]+")

# Ordered: structural matches first so a nested token is not partially rewritten
# by a looser rule.
_PATTERNS: tuple[re.Pattern[str], ...] = (
    re.compile(
        r"-----BEGIN[ A-Z0-9-]{0,64}-----.*?-----END[ A-Z0-9-]{0,64}-----",
        re.DOTALL,
    ),
    re.compile(r"-----BEGIN[ A-Z0-9-]{0,64}-----.*", re.DOTALL),
    re.compile(r"(?i)\b(?:authorization|proxy-authorization)\s*:\s*\S.*"),
    re.compile(r"(?i)\b(?:set-)?cookie\s*:\s*\S.*"),
    re.compile(r"(?i)\b(?:bearer|basic|token)\s+[A-Za-z0-9+/_=.:~-]{8,}"),
    re.compile(r"(?i)\b[a-z][a-z0-9+.-]{1,20}://[^\s/@:]{1,128}:[^\s/@]{1,256}@"),
    re.compile(
        r"\bssh-(?:rsa|ed25519|dss|ecdsa-[A-Za-z0-9-]+)\s+[A-Za-z0-9+/]{16,}={0,3}"
    ),
    re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{4,}"),
    re.compile(r"(?<![A-Za-z0-9])gh[pousr]_[A-Za-z0-9]*"),
    re.compile(r"(?<![A-Za-z0-9])github_pat_[A-Za-z0-9_]*"),
    re.compile(r"(?<![A-Za-z0-9])glpat-[A-Za-z0-9_-]*"),
    re.compile(r"(?<![A-Za-z0-9])xox[baprs]-[A-Za-z0-9-]*"),
    re.compile(r"(?<![A-Za-z0-9])sk-(?:ant-|proj-)?[A-Za-z0-9_-]*"),
    re.compile(r"(?<![A-Za-z0-9])(?:gta|gto|gitea|forgejo)_[A-Za-z0-9_-]*"),
    re.compile(r"(?<![A-Za-z0-9])hvs\.[A-Za-z0-9_-]*"),
    re.compile(r"(?<![A-Za-z0-9])(?:AKIA|ASIA)[0-9A-Z]*"),
    re.compile(r"(?<![A-Za-z0-9])AIza[0-9A-Za-z_-]{35}"),
    re.compile(r"(?<![A-Za-z0-9])npm_[A-Za-z0-9]{16,}"),
)

# `key: value` / `key=value` where the key names a credential.
_ASSIGNMENT_RE = re.compile(
    r"""(?x)
    (?P<quote>["']?)
    (?P<key>[A-Za-z][A-Za-z0-9_.\ -]{0,63})
    (?P=quote)
    (?P<gap>\s{0,8}[:=]\s{0,8})
    (?P<value>"[^"\n]{0,4096}"|'[^'\n]{0,4096}'|[^\s,;}\n][^,;}\n]{0,4096})
    """
)

# Long, mixed-class runs. Deliberately conservative: prose and hex digests of
# fewer than 44 characters are left alone so commit SHAs stay readable.
_ENTROPY_RE = re.compile(
    r"(?<![A-Za-z0-9+/_=-])[A-Za-z0-9+/_=-]{44,}(?![A-Za-z0-9+/_=-])"
)


def normalise_key(key: str) -> str:
    """Return a key lowercased with separators removed, for suffix matching."""
    return _KEY_NOISE_RE.sub("", key.lower())


def is_sensitive_key(key: str) -> bool:
    """Report whether a key name means its value must never be recorded."""
    compact = normalise_key(key)
    if not compact:
        return False
    if compact in SENSITIVE_KEY_EXACT:
        return True
    return compact.endswith(SENSITIVE_KEY_SUFFIXES)


def _is_high_entropy(candidate: str) -> bool:
    """Report whether a long run mixes enough character classes to be a token."""
    classes = sum(
        (
            any(character.islower() for character in candidate),
            any(character.isupper() for character in candidate),
            any(character.isdigit() for character in candidate),
            any(character in "+/_=-" for character in candidate),
        )
    )
    return classes >= 4


def could_hold_secret(value: str) -> bool:
    """Report whether a raw value is long and opaque enough to be a credential."""
    stripped = value.strip().strip("\"'")
    if stripped.lower() in NON_SECRET_LITERALS or len(stripped) < MIN_SECRET_LENGTH:
        return False
    try:
        float(stripped)
    except ValueError:
        return True
    return False


def _redact_assignment(match: re.Match[str]) -> str:
    value = match.group("value")
    stripped = value.strip().strip("\"'").lower()
    if not is_sensitive_key(match.group("key")) or stripped in NON_SECRET_LITERALS:
        return match.group(0)
    # Preserve the original quoting so screened JSON stays parsable JSON.
    quoted = value.strip()[:1] in {'"', "'"}
    replacement = f'"{REDACTED}"' if quoted else REDACTED
    key_quote = match.group("quote")
    return (
        f"{key_quote}{match.group('key')}{key_quote}{match.group('gap')}{replacement}"
    )


def _redact_entropy(match: re.Match[str]) -> str:
    return REDACTED if _is_high_entropy(match.group(0)) else match.group(0)


def redact(text: str) -> str:
    """Return ``text`` with credential-shaped substrings replaced."""
    if not text:
        return text
    for pattern in _PATTERNS:
        text = pattern.sub(REDACTED, text)
    text = _ASSIGNMENT_RE.sub(_redact_assignment, text)
    return _ENTROPY_RE.sub(_redact_entropy, text)


def bound(text: str, max_bytes: int = DEFAULT_MAX_BYTES) -> tuple[str, bool]:
    """Truncate ``text`` to a UTF-8 byte budget, reporting whether it was cut."""
    if max_bytes <= 0:
        return ("", bool(text))
    encoded = text.encode("utf-8", "replace")
    if len(encoded) <= max_bytes:
        return (text, False)
    marker = TRUNCATION_NOTE.encode()[:max_bytes]
    content_budget = max(
        0, max_bytes - len(marker) - (1 if len(marker) < max_bytes else 0)
    )
    kept = encoded[:content_budget].decode("utf-8", "ignore")
    separator = "\n" if kept and len(marker) < max_bytes else ""
    return (f"{kept}{separator}{marker.decode()}", True)


def safe_text(text: str, max_bytes: int = DEFAULT_MAX_BYTES) -> tuple[str, bool]:
    """Redact then bound, so truncation never exposes a credential prefix."""
    return bound(redact(text), max_bytes)


def scrub(value: Any, max_bytes: int = DEFAULT_MAX_BYTES) -> Any:
    """Recursively bound and redact a JSON-shaped value.

    Mapping keys are themselves screened; values under a sensitive key are
    dropped outright rather than pattern-screened, because a short secret
    would survive screening.
    """
    if isinstance(value, str):
        return safe_text(value, max_bytes)[0]
    if isinstance(value, dict):
        screened: dict[str, Any] = {}
        for key, item in value.items():
            raw_key = str(key)
            safe_key = safe_text(raw_key, max_bytes)[0]
            sensitive_value = (
                is_sensitive_key(raw_key)
                and item is not None
                and not isinstance(item, bool)
                and item != ""
            )
            screened[safe_key] = REDACTED if sensitive_value else scrub(item, max_bytes)
        return screened
    if isinstance(value, (list, tuple)):
        return [scrub(item, max_bytes) for item in value]
    return value


def contains_credential_shape(text: str) -> bool:
    """Report whether screening would rewrite ``text``.

    Used by the harness self-test to prove a rendered report is clean before it
    is written, and by the unit suite as an assertion helper.
    """
    return redact(text) != text