Gitea-Tools/webui/console_redaction.py

"""Secret redaction policy for every console surface (#633).

The MVP already redacts MCP-side mutation records through ``gitea_audit``.
This module is the console-facing policy: one redaction pass applied to API
payloads, rendered HTML, log lines, and audit records *before* they leave the
server or reach persistent storage.

Design constraints:

- **Reuse, never fork.** ``gitea_audit.redact`` remains the authority for
  secret-looking dict keys, ``Authorization`` material, and raw URLs. This
  module runs that pass first and then applies console-specific patterns for
  keychain references, key/value assignments, private-key blocks, and JWTs.
- **Never raises.** Redaction is a safety control; a malformed payload must
  degrade to a redacted placeholder rather than propagate an exception.
- **Redact before persist.** ``webui.console_audit`` calls this module before
  writing, so an unredacted record is never durable.
"""

from __future__ import annotations

import json
import re
from typing import Any

import gitea_audit

REDACTED = gitea_audit.REDACTED

# Console-specific patterns applied after the shared ``gitea_audit`` pass.
# Each keeps the identifying key so an operator can still tell *what* was
# removed, and replaces only the secret run itself.
_KEYCHAIN_REF = re.compile(r"(?i)\bkeychain:[\w.\-/@]+")
_KEYCHAIN_CMD = re.compile(
    r"(?i)\bsecurity\s+find-(?:generic|internet)-password\b[^\n]*"
)
_ASSIGNMENT = re.compile(
    r"(?i)\b(token|password|passwd|secret|api[_-]?key|access[_-]?key|"
    r"client[_-]?secret|private[_-]?key)\b(\s*[:=]\s*)"
    r"(\"[^\"]*\"|'[^']*'|\S+)"
)
_ENV_ASSIGNMENT = re.compile(
    r"(?i)\b(GITEA_(?:TOKEN|PASS|PASSWORD)[A-Z0-9_]*)(\s*=\s*)"
    r"(\"[^\"]*\"|'[^']*'|\S+)"
)
_PRIVATE_KEY_BLOCK = re.compile(
    r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----",
    re.S,
)
_JWT = re.compile(
    r"\beyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}\b"
)

# Shapes that mean a payload still carries a secret. ``scan_for_secrets`` uses
# these to assert a surface is clean.
_DETECTORS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("keychain_reference", _KEYCHAIN_REF),
    ("keychain_command", _KEYCHAIN_CMD),
    ("credential_assignment", _ASSIGNMENT),
    ("credential_env_assignment", _ENV_ASSIGNMENT),
    ("private_key_block", _PRIVATE_KEY_BLOCK),
    ("json_web_token", _JWT),
    ("bearer_credential", re.compile(r"(?i)\b(?:bearer|basic)\s+\S{8,}")),
)


def _mask_assignment(match: re.Match[str]) -> str:
    """Keep the key and separator, replace the value."""
    return f"{match.group(1)}{match.group(2)}{REDACTED}"


def redact_text(text: Any) -> Any:
    """Redact secret material from a single string.

    Non-strings are returned unchanged so this is safe to map over mixed
    payloads. Runs the shared ``gitea_audit`` pass first, then the
    console-specific patterns.
    """
    if not isinstance(text, str) or not text:
        return text
    try:
        out = gitea_audit.redact(text)
        if not isinstance(out, str):  # defensive; redact() returns str for str
            return REDACTED
        out = _PRIVATE_KEY_BLOCK.sub(f"{REDACTED}_PRIVATE_KEY", out)
        out = _ENV_ASSIGNMENT.sub(_mask_assignment, out)
        out = _ASSIGNMENT.sub(_mask_assignment, out)
        out = _KEYCHAIN_CMD.sub(f"{REDACTED}_KEYCHAIN_COMMAND", out)
        out = _KEYCHAIN_REF.sub(f"{REDACTED}_KEYCHAIN_REF", out)
        out = _JWT.sub(f"{REDACTED}_JWT", out)
        return out
    except Exception:
        # Fail closed: an unredactable string is dropped rather than emitted raw.
        return REDACTED


def redact_payload(value: Any) -> Any:
    """Recursively redact a JSON-able payload for any console surface.

    Secret-looking dict keys are replaced wholesale by the shared
    ``gitea_audit`` policy; every remaining string is run through
    :func:`redact_text`.
    """
    try:
        shared = gitea_audit.redact(value)
    except Exception:
        return REDACTED
    return _walk(shared)


def _walk(value: Any) -> Any:
    if isinstance(value, dict):
        return {k: _walk(v) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_walk(v) for v in value]
    if isinstance(value, str):
        return redact_text(value)
    return value


def scan_for_secrets(value: Any) -> list[str]:
    """Return detector names that still match *value* after serialization.

    Used to assert an outbound payload or rendered page is clean. An empty
    list means no known secret shape was found. Already-redacted hits are not
    findings.
    """
    if isinstance(value, str):
        text = value
    else:
        try:
            text = json.dumps(value, default=str)
        except Exception:
            text = str(value)
    findings: list[str] = []
    for name, pattern in _DETECTORS:
        for match in pattern.finditer(text):
            if REDACTED in match.group(0):
                continue
            findings.append(name)
            break
    return findings


def redaction_policy() -> dict[str, Any]:
    """Machine-readable statement of the redaction rules (never secrets)."""
    return {
        "policy_version": 1,
        "applies_to": [
            "json_api_responses",
            "rendered_html",
            "server_logs",
            "audit_records",
        ],
        "ordering": "shared gitea_audit pass, then console patterns",
        "redact_before_persist": True,
        "shared_rules": {
            "source": "gitea_audit.redact",
            "secret_key_hints": list(gitea_audit._SECRET_KEY_HINTS),
            "secret_value_prefixes": list(gitea_audit._SECRET_VALUE_PREFIXES),
            "urls": "credentials, secret query parameters, and real hosts redacted",
        },
        "console_rules": [
            {"name": name, "pattern": pattern.pattern}
            for name, pattern in _DETECTORS
        ],
        "placeholder": REDACTED,
        "failure_mode": "fail closed — unredactable values become the placeholder",
    }