redactor.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/redactor.py
 
"""
Purpose
-------
Output Redactor (Layer 5) for the ``oct-mcp`` six-layer pipeline.
Applies a five-step sanitisation pipeline to every tool output before
it is returned to the MCP client.
 
Responsibilities
----------------
- Step 1: Apply ``redaction_patterns`` from ``debug_config.json``
  (loaded at server startup; falls back to empty list if unavailable).
- Step 2: Apply ``_SECRET_NAME_PATTERNS`` line filtering — any output
  line that looks like a ``KEY=value`` assignment whose key matches a
  secret pattern is replaced with ``[REDACTED]``.
- Step 3: Anonymise absolute paths:
    ``$HOME/...``  → ``[HOME]/...``
    ``/usr/``, ``/bin/``, ``/opt/``, etc. → ``[SYSTEM]/...``
    ``<project_root>/...`` → ``[PROJECT]/...``
    Drive-letter paths on Windows follow the same rules.
- Step 4: Strip ANSI escape sequences.
- Step 5: Truncate output that exceeds ``max_output_bytes``, appending
  a ``[OUTPUT TRUNCATED AT <N> BYTES]`` marker.
 
Also applies git-URL credential redaction (``_redact_git_url``) as
a supplementary step after step 2 to catch any credentials that
survived the key-pattern filter.
 
Diagnostics
-----------
Domain: MCP
Levels:
    L2 — lifecycle: redactor instantiation
    L3 — per-call redaction counts
    L4 — deep trace: individual substitutions
 
Contracts
---------
- This module does not import ``click`` or the MCP SDK.
- :meth:`Redactor.apply_all` is pure: the same input always produces the
  same output given the same configuration.
- :meth:`Redactor.apply_all` never raises.
- ``redactions_applied`` in :class:`RedactorResult` counts *all*
  individual substitutions made across all steps.
"""
 
from __future__ import annotations
 
import re
import sys
from dataclasses import dataclass
from pathlib import Path
 
try:
    from oc_diagnostics import _dbg as _real_dbg
 
    def _dbg(*args, **kwargs) -> None:
        _real_dbg(*args, **kwargs)
except ImportError:  # pragma: no cover
    def _dbg(*args, **kwargs) -> None:
        return None
 
try:
    from oct.linter.oct_lint import _SECRET_NAME_PATTERNS
except ImportError:  # pragma: no cover - graceful degradation
    _SECRET_NAME_PATTERNS: frozenset[str] = frozenset({
        "password", "passwd", "pwd",
        "secret", "secret_key", "private_key",
        "api_key", "apikey", "api_secret",
        "token", "auth_token", "access_token", "refresh_token",
        "connection_string", "conn_str", "dsn", "database_url", "db_url",
        "aws_secret", "aws_key", "aws_access",
        "credentials", "client_secret",
    })
 
try:
    from oct.core.git import _redact_git_url
except ImportError:  # pragma: no cover
    def _redact_git_url(text: str) -> str:
        return text
 
 
# ---------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------
 
#: ANSI escape sequence pattern (covers colour, cursor, SGR sequences).
_ANSI_ESCAPE_RE: re.Pattern[str] = re.compile(
    r"\x1b(?:[@-Z\\-_]|\‍[[0-?]*[ -/]*[@-~])"
)
 
#: Pattern for ``KEY=value`` or ``KEY: value`` assignment lines whose
#: key matches a secret pattern. Captures the key in group 1 and the
#: separator + value in group 2 so we can reconstruct the line.
#:
#: Matches lines like:
#:   API_KEY=abc123
#:   password: hunter2
#:   token = "xyz"
_ASSIGNMENT_LINE_RE: re.Pattern[str] = re.compile(
    r'^(\s*(?:[A-Za-z_][A-Za-z0-9_]*))\s*[:=]\s*(.+)$'
)
 
#: System path prefixes to anonymise on POSIX.
_SYSTEM_PATH_PREFIXES_POSIX: tuple[str, ...] = (
    "/usr/", "/bin/", "/sbin/", "/lib/", "/lib64/",
    "/opt/", "/etc/", "/var/", "/tmp/", "/run/",
    "/snap/", "/proc/", "/sys/",
)
 
 
# ---------------------------------------------------------------------
# Result type
# ---------------------------------------------------------------------
 
 
@dataclass
class RedactorResult:
    """Output from :meth:`Redactor.apply_all`.
 
    Attributes
    ----------
    text
        The sanitised output string.
    redactions_applied
        Total number of individual substitutions performed across all
        five pipeline steps.
    truncated
        True if the output was truncated to ``max_output_bytes``.
    original_size_bytes
        Byte length of the input text (before any processing).
    """
 
    text: str
    redactions_applied: int
    truncated: bool
    original_size_bytes: int
 
 
# ---------------------------------------------------------------------
# Redactor
# ---------------------------------------------------------------------
 
 
class Redactor:
    """Five-step output sanitisation pipeline (Layer 5).
 
    Parameters
    ----------
    redaction_patterns
        List of regex patterns loaded from ``debug_config.json``. Each
        entry may be a plain string (treated as a literal) or a dict
        with keys ``pattern`` (regex) and ``replacement`` (substitution,
        default ``"[REDACTED]"``). Patterns that fail to compile are
        silently skipped.
    max_output_bytes
        Hard truncation limit. Default 1 MB. Matches ``McpConfig``
        ``max_output_bytes``.
    """
 
    def __init__(
        self,
        redaction_patterns: list[str | dict] | None = None,
        max_output_bytes: int = 1_048_576,
    ) -> None:
        func = "Redactor.__init__"
        self._max_output_bytes: int = max(1, max_output_bytes)
        self._compiled: list[tuple[re.Pattern[str], str]] = []
 
        for entry in (redaction_patterns or []):
            try:
                if isinstance(entry, str):
                    self._compiled.append((re.compile(entry), "[REDACTED]"))
                elif isinstance(entry, dict):
                    pat = entry.get("pattern", "")
                    rep = entry.get("replacement", "[REDACTED]")
                    if pat:
                        self._compiled.append((re.compile(pat), str(rep)))
            except re.error:
                _dbg("MCP", func, f"invalid redaction pattern (skipped): {entry!r}", 1)
 
        # Build home-dir and system path prefixes once at init time.
        home = str(Path.home())
        self._home_prefix: str = home
 
        _dbg("MCP", func, f"compiled {len(self._compiled)} custom patterns", 2)
 
    # -- pipeline steps --------------------------------------------------
 
    def _step1_custom_patterns(self, text: str) -> tuple[str, int]:
        """Apply ``debug_config.json`` redaction_patterns."""
        count = 0
        for pattern, replacement in self._compiled:
            new_text, n = pattern.subn(replacement, text)
            count += n
            text = new_text
        return text, count
 
    def _step2_secret_name_lines(self, text: str) -> tuple[str, int]:
        """Replace lines that assign a value to a secret-named key."""
        count = 0
        out_lines: list[str] = []
        for line in text.splitlines(keepends=True):
            m = _ASSIGNMENT_LINE_RE.match(line)
            if m:
                key = m.group(1).strip()
                if any(p in key.lower() for p in _SECRET_NAME_PATTERNS):
                    # Preserve the newline suffix if present.
                    suffix = "\n" if line.endswith("\n") else ""
                    out_lines.append(f"{key}=[REDACTED]{suffix}")
                    count += 1
                    continue
            out_lines.append(line)
        return "".join(out_lines), count
 
    def _step2b_git_url_redaction(self, text: str) -> tuple[str, int]:
        """Strip embedded git credentials (http user:token@)."""
        new_text = _redact_git_url(text)
        # Count as 1 redaction if any change was made.
        return new_text, (1 if new_text != text else 0)
 
    def _step3_path_anonymisation(self, text: str, project_root: Path) -> tuple[str, int]:
        """Replace absolute paths with [HOME], [SYSTEM], [PROJECT] tokens.
 
        Processing order matters: project root first (most specific),
        then home, then system prefixes (most general).
        """
        count = 0
        project_str = str(project_root)
 
        # --- project root ---
        if project_str in text:
            count += text.count(project_str)
            text = text.replace(project_str, "[PROJECT]")
 
        # Also handle forward-slash version on Windows.
        project_fwd = project_str.replace("\\", "/")
        if project_fwd != project_str and project_fwd in text:
            count += text.count(project_fwd)
            text = text.replace(project_fwd, "[PROJECT]")
 
        # --- home directory ---
        home_str = self._home_prefix
        if home_str in text:
            count += text.count(home_str)
            text = text.replace(home_str, "[HOME]")
 
        home_fwd = home_str.replace("\\", "/")
        if home_fwd != home_str and home_fwd in text:
            count += text.count(home_fwd)
            text = text.replace(home_fwd, "[HOME]")
 
        # --- system paths ---
        if sys.platform == "win32":
            # Windows: replace common drive-root paths.
            for prefix in ("C:\\Windows\\", "C:\\Program Files\\",
                           "C:\\Program Files (x86)\\", "C:\\ProgramData\\"):
                if prefix in text:
                    count += text.count(prefix)
                    text = text.replace(prefix, "[SYSTEM]/")
        else:
            for prefix in _SYSTEM_PATH_PREFIXES_POSIX:
                if prefix in text:
                    count += text.count(prefix)
                    text = text.replace(prefix, "[SYSTEM]/")
 
        return text, count
 
    def _step4_strip_ansi(self, text: str) -> tuple[str, int]:
        """Remove ANSI escape sequences."""
        new_text, n = _ANSI_ESCAPE_RE.subn("", text)
        return new_text, n
 
    def _step5_truncate(self, text: str) -> tuple[str, bool]:
        """Truncate output to ``max_output_bytes`` if necessary."""
        encoded = text.encode("utf-8", errors="replace")
        if len(encoded) <= self._max_output_bytes:
            return text, False
        truncated = encoded[: self._max_output_bytes].decode("utf-8", errors="replace")
        marker = f"\n[OUTPUT TRUNCATED AT {self._max_output_bytes} BYTES]"
        return truncated + marker, True
 
    # -- public API ------------------------------------------------------
 
    def apply_all(self, text: str, project_root: Path) -> RedactorResult:
        """Run the full five-step pipeline on *text*.
 
        Never raises. Any unexpected exception is caught and the
        original text is returned with a warning prepended.
 
        Parameters
        ----------
        text
            Raw tool output to sanitise.
        project_root
            Absolute path to the project root (used for step 3 path
            anonymisation).
 
        Returns
        -------
        :class:`RedactorResult`
        """
        func = "Redactor.apply_all"
        original_size = len(text.encode("utf-8", errors="replace"))
        total_redactions = 0
 
        try:
            # Step 1: custom patterns from debug_config.json
            text, n1 = self._step1_custom_patterns(text)
            total_redactions += n1
 
            # Step 2: secret-named assignment lines
            text, n2 = self._step2_secret_name_lines(text)
            total_redactions += n2
 
            # Step 2b: git URL credential redaction
            text, n2b = self._step2b_git_url_redaction(text)
            total_redactions += n2b
 
            # Step 3: path anonymisation
            text, n3 = self._step3_path_anonymisation(text, project_root)
            total_redactions += n3
 
            # Step 4: ANSI strip
            text, n4 = self._step4_strip_ansi(text)
            total_redactions += n4
 
            # Step 5: truncation
            text, truncated = self._step5_truncate(text)
 
        except Exception as exc:  # pragma: no cover - defensive catch-all
            _dbg("MCP", func, f"SYSTEM_ERROR: redactor pipeline failed: {exc}", 1)
            truncated = False
 
        _dbg(
            "MCP", func,
            f"redactions={total_redactions} original_bytes={original_size}",
            3,
        )
 
        return RedactorResult(
            text=text,
            redactions_applied=total_redactions,
            truncated=truncated,
            original_size_bytes=original_size,
        )