qwen3-6-lora/scripts/04_sanitize.py

"""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()