Clean release snapshot

lib/diagnostics_redact.py

@@ -0,0 +1,158 @@
+"""Redaction primitives for diagnostic bundles.
+
+A `Redactor` carries the per-bundle salt and substitution caches so that
+identical inputs (e.g. the same song path appearing in 50 log lines)
+produce identical output tokens (`<song:a3f1c2>`). Different bundles get
+different salts so tokens cannot be cross-correlated between exports.
+
+Stable token grammar (see docs/diagnostics-bundle-spec.md):
+    <DLC_DIR>          — DLC root path
+    <HOME>             — user's home directory
+    <CONFIG_DIR>       — slopsmith config dir
+    <song:hash8>       — song filename / basename (8 hex chars)
+    <ip:hash6>          — IPv4 / IPv6 address (6 hex chars)
+    <redacted>          — bearer tokens, key=/token= query strings
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import secrets
+from pathlib import Path
+
+
+_IPV4_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
+_IPV6_RE = re.compile(
+    r"(?<![A-Fa-f0-9:])"
+    r"(?:[A-Fa-f0-9]{1,4}:){2,7}[A-Fa-f0-9]{1,4}"
+    r"(?![A-Fa-f0-9:])"
+)
+_BEARER_RE = re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._\-+/=]+")
+_URL_USERINFO_RE = re.compile(r"(?i)(https?://)[^@/\s]+@")
+_QSTRING_SECRET_RE = re.compile(
+    r"(?i)\b(api[_-]?key|key|token|secret|password|pwd|auth)=([^\s&\"']+)"
+)
+_SONG_FILENAME_RE = re.compile(
+    r"\b[\w()'\-+&,.!?\[\]]+\.(?:psarc|sloppak|wem|ogg|mp3|wav)\b",
+    re.IGNORECASE,
+)
+
+
+class Redactor:
+    def __init__(
+        self,
+        dlc_dir: Path | None = None,
+        home_dir: Path | None = None,
+        config_dir: Path | None = None,
+    ) -> None:
+        self._salt = secrets.token_hex(8)
+        self._dlc_dir = self._normalize(dlc_dir)
+        self._home_dir = self._normalize(home_dir)
+        self._config_dir = self._normalize(config_dir)
+        self._song_cache: dict[str, str] = {}
+        self._ip_cache: dict[str, str] = {}
+        self.counts: dict[str, int] = {
+            "paths_replaced": 0,
+            "ips_replaced": 0,
+            "song_names_replaced": 0,
+            "secrets_replaced": 0,
+        }
+
+    @staticmethod
+    def _normalize(p: Path | None) -> str | None:
+        if p is None:
+            return None
+        # Resolve only when the path exists, so callers can pass a
+        # synthetic prefix (tests, container-mapped paths) without
+        # having Path.resolve() rewrite a missing /dlc/songs to
+        # C:\dlc\songs on Windows.
+        try:
+            if p.exists():
+                s = str(p.resolve())
+            else:
+                s = str(p)
+        except (OSError, RuntimeError):
+            s = str(p)
+        return s if s and s != "." else None
+
+    def _hash(self, value: str, n: int) -> str:
+        h = hashlib.sha256()
+        h.update(self._salt.encode())
+        h.update(value.encode())
+        return h.hexdigest()[:n]
+
+    def _replace_path_prefix(self, text: str, prefix: str | None, token: str) -> str:
+        if not prefix:
+            return text
+        # Match both forward- and backslash variants — Windows paths
+        # appear with backslashes in tracebacks, Linux with slashes.
+        candidates = {prefix, prefix.replace("/", "\\"), prefix.replace("\\", "/")}
+        replaced = text
+        for cand in candidates:
+            if not cand:
+                continue
+            count = replaced.count(cand)
+            if count:
+                replaced = replaced.replace(cand, token)
+                self.counts["paths_replaced"] += count
+        return replaced
+
+    def _redact_song(self, m: re.Match) -> str:
+        name = m.group(0)
+        token = self._song_cache.get(name)
+        if token is None:
+            token = f"<song:{self._hash(name, 8)}>"
+            self._song_cache[name] = token
+        self.counts["song_names_replaced"] += 1
+        return token
+
+    def _redact_ip(self, m: re.Match) -> str:
+        ip = m.group(0)
+        # Skip obvious non-IPs: dotted version numbers, sloppy fragments.
+        if ip.count(".") == 3:
+            try:
+                if not all(0 <= int(p) <= 255 for p in ip.split(".")):
+                    return ip
+            except ValueError:
+                return ip
+        token = self._ip_cache.get(ip)
+        if token is None:
+            token = f"<ip:{self._hash(ip, 6)}>"
+            self._ip_cache[ip] = token
+        self.counts["ips_replaced"] += 1
+        return token
+
+    def _redact_secret_qstring(self, m: re.Match) -> str:
+        self.counts["secrets_replaced"] += 1
+        return f"{m.group(1)}=<redacted>"
+
+    def _redact_bearer(self, _m: re.Match) -> str:
+        self.counts["secrets_replaced"] += 1
+        return "Bearer <redacted>"
+
+    def _redact_url_userinfo(self, m: re.Match) -> str:
+        self.counts["secrets_replaced"] += 1
+        return f"{m.group(1)}<redacted>@"
+
+    def redact_text(self, text: str) -> str:
+        if not isinstance(text, str) or not text:
+            return text
+        # Path prefixes first (longest-match) so song-name regex never
+        # eats a path component.
+        text = self._replace_path_prefix(text, self._dlc_dir, "<DLC_DIR>")
+        text = self._replace_path_prefix(text, self._config_dir, "<CONFIG_DIR>")
+        text = self._replace_path_prefix(text, self._home_dir, "<HOME>")
+        text = _SONG_FILENAME_RE.sub(self._redact_song, text)
+        text = _IPV6_RE.sub(self._redact_ip, text)
+        text = _IPV4_RE.sub(self._redact_ip, text)
+        # URL userinfo before query-string secrets so user:pass@ is caught
+        # even when the URL also has token= in the query string.
+        text = _URL_USERINFO_RE.sub(self._redact_url_userinfo, text)
+        text = _QSTRING_SECRET_RE.sub(self._redact_secret_qstring, text)
+        text = _BEARER_RE.sub(self._redact_bearer, text)
+        return text
+
+    def redact_lines(self, lines):
+        for line in lines:
+            yield self.redact_text(line)