Fase 2: 04_sanitize.py - scrubbing de secretos de skills/agentes/plans (10 secretos unicos, gate en verde)

scripts/04_sanitize.py

@@ -0,0 +1,190 @@
+"""Fase 2: sanitiza secretos reales en las fuentes de contenido para autoria de seeds
+(5 SKILL.md, 7 agentes de ~/.claude/agents/, 66 planes de ~/.claude/plans/) antes de
+que ningun subagente developer los lea para escribir ejemplos de entrenamiento.
+
+Corre localmente (no requiere GPU ni el modelo). Usa detect-secrets (entropia alta,
+formatos conocidos AWS/JWT/etc.) mas una lista de regex explicitas para los patrones
+ya conocidos de este proyecto (SPARK_PASSWORD, GMAIL_APP_PASSWORD, el bearer token
+reusado, emails reales, IPs/hostnames internos de spark). La sustitucion es estable:
+el mismo secreto real siempre produce el mismo placeholder, via un diccionario hash
+persistido localmente en data/raw/.secrets_map.json (gitignoreado, solo para debug).
+
+Salida: data/raw/sanitized/{skills,agents,plans}/... (misma estructura relativa).
+
+Gate: termina con exit code != 0 si detect-secrets o las regex explicitas encuentran
+algo en la salida YA sanitizada -- falla cerrado, no adivina reemplazos para
+patrones no reconocidos.
+"""
+import hashlib
+import json
+import re
+import sys
+from pathlib import Path
+
+from detect_secrets.core import scan
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+HOME = Path.home()
+
+SOURCES = [
+    (HOME / ".claude" / "skills", "*/SKILL.md", "skills"),
+    (HOME / ".claude" / "agents", "*.md", "agents"),
+    (HOME / ".claude" / "plans", "*.md", "plans"),
+]
+
+OUTPUT_ROOT = REPO_ROOT / "data" / "raw" / "sanitized"
+SECRETS_MAP_PATH = REPO_ROOT / "data" / "raw" / ".secrets_map.json"
+
+# Patrones explicitos conocidos de este proyecto. Cada tupla es
+# (nombre_semantico, regex compilado). El grupo de captura 0 completo es lo que
+# se reemplaza; si el patron tiene grupos, se reemplaza el match completo igual
+# (el placeholder no intenta preservar texto circundante).
+EXPLICIT_PATTERNS = [
+    ("SPARK_PASSWORD", re.compile(r"\b01140102Alb\?")),
+    ("BEARER_TOKEN", re.compile(r"\b7c1f76a62391a47941d7aab8369eb8f20334daf136ba88080815ff3070773a1f\b")),
+    ("DB_PASSWORD", re.compile(r"\bsarh21234\b")),
+    ("SPARK_IP", re.compile(r"\b10\.212\.133\.200\b")),
+    ("INTERNAL_IP", re.compile(r"\b10\.212\.133\.\d{1,3}\b")),
+    ("INTERNAL_SUBNET", re.compile(r"\b10\.212\.133\.0/24\b")),
+    ("EMAIL", re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")),
+    ("GMAIL_APP_PASSWORD_LITERAL", re.compile(r"\bGMAIL_APP_PASSWORD=[^\s\"']+")),
+]
+
+# Nombres de variables de entorno que NUNCA deben aparecer con un valor real
+# asignado literalmente (uso de la variable en si, p.ej. "$SPARK_PASSWORD", esta bien).
+ENV_VAR_NAMES = ["SPARK_PASSWORD", "GMAIL_APP_PASSWORD"]
+
+PLACEHOLDER_COUNTS = {}
+SECRETS_MAP = {}
+
+
+def load_secrets_map():
+    global SECRETS_MAP
+    if SECRETS_MAP_PATH.exists():
+        SECRETS_MAP = json.loads(SECRETS_MAP_PATH.read_text())
+
+
+def save_secrets_map():
+    SECRETS_MAP_PATH.parent.mkdir(parents=True, exist_ok=True)
+    SECRETS_MAP_PATH.write_text(json.dumps(SECRETS_MAP, indent=2, ensure_ascii=False))
+
+
+def stable_placeholder(semantic_name, raw_value):
+    """Mismo secreto real -> mismo placeholder, vía un hash estable persistido."""
+    digest = hashlib.sha256(raw_value.encode("utf-8")).hexdigest()[:8]
+    if digest not in SECRETS_MAP:
+        PLACEHOLDER_COUNTS[semantic_name] = PLACEHOLDER_COUNTS.get(semantic_name, 0) + 1
+        placeholder = f"<<{semantic_name}_{PLACEHOLDER_COUNTS[semantic_name]}>>"
+        SECRETS_MAP[digest] = {"placeholder": placeholder, "semantic_name": semantic_name}
+    return SECRETS_MAP[digest]["placeholder"]
+
+
+def apply_explicit_patterns(text):
+    for semantic_name, pattern in EXPLICIT_PATTERNS:
+        def _sub(match, semantic_name=semantic_name):
+            return stable_placeholder(semantic_name, match.group(0))
+
+        text = pattern.sub(_sub, text)
+    return text
+
+
+def detect_secrets_scan(text):
+    """Corre los plugins default de detect-secrets sobre el texto linea por linea.
+    Devuelve la lista de (line_number, secret_value) encontrados."""
+    findings = []
+    plugins = list(scan.get_plugins())
+    for lineno, line in enumerate(text.splitlines(), start=1):
+        for plugin in plugins:
+            try:
+                results = plugin.analyze_line(filename="<sanitize>", line=line, line_number=lineno)
+            except Exception:
+                continue
+            if results:
+                for secret in results:
+                    raw = getattr(secret, "secret_value", None)
+                    if raw:
+                        findings.append((lineno, raw))
+    return findings
+
+
+def sanitize_text(text):
+    text = apply_explicit_patterns(text)
+    # Segunda pasada: detect-secrets sobre el resultado de las regex explicitas,
+    # para capturar cualquier secreto de alta entropia no cubierto arriba.
+    for lineno, raw in detect_secrets_scan(text):
+        placeholder = stable_placeholder("GENERIC_SECRET", raw)
+        text = text.replace(raw, placeholder)
+    return text
+
+
+def gate_check(text, relpath):
+    """Falla cerrado: si sobrevive un patron conocido o detect-secrets encuentra
+    algo en la salida YA sanitizada, es un error del gate."""
+    problems = []
+    for semantic_name, pattern in EXPLICIT_PATTERNS:
+        if pattern.search(text):
+            problems.append(f"{relpath}: patron explicito '{semantic_name}' sobrevivio la sanitizacion")
+    for env_var in ENV_VAR_NAMES:
+        if re.search(rf"\b{env_var}\s*=\s*[^\s$\"'][^\s]*", text):
+            problems.append(f"{relpath}: posible asignacion literal de {env_var} sobrevivio")
+    residual = detect_secrets_scan(text)
+    if residual:
+        for lineno, raw in residual:
+            problems.append(f"{relpath}:{lineno}: detect-secrets encontro un secreto residual")
+    return problems
+
+
+def collect_files():
+    files = []
+    for base_dir, glob_pattern, bucket in SOURCES:
+        if not base_dir.exists():
+            print(f"[WARN] fuente no encontrada: {base_dir}")
+            continue
+        for path in sorted(base_dir.glob(glob_pattern)):
+            if path.is_file():
+                files.append((path, bucket))
+    return files
+
+
+def main():
+    load_secrets_map()
+    files = collect_files()
+    print(f"[INFO] {len(files)} archivos fuente encontrados")
+
+    all_problems = []
+    written = 0
+    for path, bucket in files:
+        raw_text = path.read_text(encoding="utf-8")
+        sanitized_text = sanitize_text(raw_text)
+
+        if bucket == "skills":
+            relpath = Path(bucket) / path.parent.name / path.name
+        else:
+            relpath = Path(bucket) / path.name
+
+        problems = gate_check(sanitized_text, relpath)
+        all_problems.extend(problems)
+
+        out_path = OUTPUT_ROOT / relpath
+        out_path.parent.mkdir(parents=True, exist_ok=True)
+        out_path.write_text(sanitized_text, encoding="utf-8")
+        written += 1
+
+    save_secrets_map()
+
+    print(f"[INFO] {written} archivos sanitizados escritos en {OUTPUT_ROOT}")
+    total_subs = sum(PLACEHOLDER_COUNTS.values())
+    print(f"[INFO] {total_subs} secretos unicos sustituidos: {PLACEHOLDER_COUNTS}")
+
+    if all_problems:
+        print(f"[GATE FAIL] {len(all_problems)} problemas encontrados en la salida sanitizada:")
+        for problem in all_problems:
+            print(f"  - {problem}")
+        sys.exit(1)
+
+    print("[GATE OK] 0 secretos sobrevivientes detectados en la salida sanitizada")
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()