takana/scripts/mudanza/traducir.py

#!/usr/bin/env python3
"""traducir.py — traduce la CONFIGURACIÓN de un servicio a la de su equivalente (SDD 29).

Mudar de nginx a caddy no es cambiar un binario: es reescribir la configuración. Esto hace la parte
mecánica y —sobre todo— **dice con nombre y número de línea lo que NO pudo traducir**.

── DOCTRINA HEREDADA DE `soltar`/`paskaq` (tawasuyu) ────────────────────────────────────────────
`soltar` desencadena datos presos en formatos cautivos (Btrieve, Paradox, .mdb). El dominio es otro,
pero sus principios son exactamente los que un traductor de configuración necesita, y se adoptan:

  · **Elisión honesta.** «Lo que no se pudo leer se reporta con offset y motivo.» Acá: cada
    directiva que no se entendió sale con su NÚMERO DE LÍNEA y su texto. Un traductor que descarta
    en silencio una directiva es catastrófico — te deja un servidor sin una redirección, sin una
    regla de auth o sin una cabecera, y el sitio parece funcionar.
  · **No adivinar en silencio.** Toda heurística queda registrada como decisión, no aplicada de
    tapadillo.
  · **Procedencia total.** Cada línea del resultado sabe de qué línea del original salió.
  · **Determinista**: mismo archivo, misma salida, byte a byte.

── LO QUE ESTO NO ES ───────────────────────────────────────────────────────────────────────────
No es un traductor completo de nginx: nginx tiene cientos de directivas, `if`, `rewrite` con regex,
`map`, módulos. **Traduce el núcleo común y declara todo lo demás.** Un traductor honesto que cubre
el 70 % y dice cuál es el 30 % restante es útil; uno que aparenta cubrir el 100 % es una trampa.

Uso:
    scripts/mudanza/traducir.py --from nginx --to caddy --in nginx.conf --out Caddyfile
    scripts/mudanza/traducir.py --from nginx --to caddy --in nginx.conf   # sólo el acta
"""
import argparse, re, sys

# ── lectura de nginx ────────────────────────────────────────────────────────────────────────────

def tokenizar_nginx(texto):
    """Directivas con su número de línea y su anidamiento. Suficiente para el núcleo; no pretende
    ser el parser de nginx."""
    out, nivel, buf, ln_ini = [], 0, "", 0
    for ln, linea in enumerate(texto.split("\n"), 1):
        s = linea.split("#")[0].strip()
        if not s:
            continue
        if not buf:
            ln_ini = ln
        for ch in s:
            if ch == "{":
                out.append((ln_ini, nivel, buf.strip(), "abre"))
                nivel += 1
                buf = ""
            elif ch == "}":
                if buf.strip():
                    out.append((ln_ini, nivel, buf.strip(), "dir"))
                nivel -= 1
                out.append((ln_ini, nivel, "", "cierra"))
                buf = ""
            elif ch == ";":
                out.append((ln_ini, nivel, buf.strip(), "dir"))
                buf = ""
            else:
                buf += ch
        if buf and not buf.endswith(" "):
            buf += " "
    return out


# ── traducción nginx → caddy ────────────────────────────────────────────────────────────────────

def nginx_a_caddy(texto):
    """Devuelve `(caddyfile, acta)`. El acta es la parte que no se puede omitir."""
    toks = tokenizar_nginx(texto)
    acta, sitios = [], []
    actual, pila, loc = None, [], None

    def anota(ln, estado, directiva, motivo=""):
        acta.append({"linea": ln, "estado": estado, "directiva": directiva, "motivo": motivo})

    # ⚠ Profundidad de bloques que NO se pudieron traducir (`if`, `map`, `stream`, `upstream`).
    # Sin esto, las directivas de adentro se absorbían como si fueran del servidor: un
    # `if (...) { return 403; }` salía como un `respond 403` INCONDICIONAL, o sea el sitio entero
    # devolviendo 403. Encontrado validando la salida con el caddy real. Es el modo de fallo que la
    # doctrina de `soltar` llama pérdida silenciosa — sólo que peor: no pierde, CAMBIA el sentido.
    opaco = 0

    i = 0
    while i < len(toks):
        ln, niv, txt, tipo = toks[i]
        i += 1
        cab = txt.split()[0] if txt else ""

        if tipo == "abre":
            pila.append(cab)
            if cab == "server":
                actual = {"nombres": [], "puertos": [], "root": None, "index": None,
                          "locations": [], "tls": {}, "linea": ln}
                anota(ln, "traducido", "server {", "→ un bloque de sitio de Caddy")
            elif cab == "location":
                partes = txt.split()
                loc = {"ruta": partes[1] if len(partes) > 1 else "/", "dirs": [], "linea": ln}
                if len(partes) > 2:
                    anota(ln, "SIN-TRADUCIR", txt,
                          "location con modificador (~, ~*, ^~, =): las reglas por regex de nginx no "
                          "tienen equivalente directo en Caddy — hay que reescribirla a mano")
            elif cab in ("http", "events", "upstream", "map", "if", "stream"):
                if cab in ("map", "if", "stream", "upstream"):
                    opaco += 1
                    anota(ln, "SIN-TRADUCIR", txt + " {",
                          f"`{cab}` no tiene traducción mecánica: su semántica en nginx depende del "
                          f"orden de evaluación y del contexto — y TODO lo de adentro queda sin "
                          f"traducir, no se absorbe al bloque de afuera")
            continue

        if tipo == "cierra":
            c = pila.pop() if pila else ""
            if c in ("map", "if", "stream", "upstream") and opaco:
                opaco -= 1
            if c == "location" and actual is not None and loc:
                actual["locations"].append(loc); loc = None
            elif c == "server" and actual is not None:
                sitios.append(actual); actual = None
            continue

        # ── directivas ──────────────────────────────────────────────────────────────────────────
        if opaco:
            anota(ln, "SIN-TRADUCIR", txt,
                  "está dentro de un bloque que no se pudo traducir: sacarla de ahí le cambiaría el "
                  "sentido (dejaría de ser condicional)")
            continue
        if actual is None:
            if cab in ("worker_processes", "user", "pid", "events", "include", "worker_connections",
                       "sendfile", "keepalive_timeout", "gzip", "default_type", "log_format",
                       "access_log", "error_log", "types_hash_max_size", "tcp_nopush"):
                anota(ln, "no-aplica", txt,
                      "ajuste global de nginx sin equivalente necesario: Caddy lo maneja solo")
            elif txt:
                anota(ln, "SIN-TRADUCIR", txt, "directiva global no reconocida")
            continue

        p = txt.split()
        if cab == "server_name":
            actual["nombres"] += [x for x in p[1:] if x != "_"]
            anota(ln, "traducido", txt, "→ nombre del sitio")
        elif cab == "listen":
            actual["puertos"].append(p[1] if len(p) > 1 else "")
            if "ssl" in p:
                anota(ln, "traducido", txt, "→ Caddy hace TLS automático; `ssl` no se escribe")
            else:
                anota(ln, "traducido", txt, "→ puerto del sitio")
        elif cab == "root":
            (loc or actual)["root" if loc is None else "root"] = p[1] if len(p) > 1 else ""
            if loc is not None:
                loc["dirs"].append(("root", p[1] if len(p) > 1 else ""))
            anota(ln, "traducido", txt, "→ `root *`")
        elif cab == "index":
            actual["index"] = " ".join(p[1:])
            anota(ln, "traducido", txt, "→ `file_server` usa index por defecto")
        elif cab in ("ssl_certificate", "ssl_certificate_key"):
            actual["tls"][cab] = p[1] if len(p) > 1 else ""
            anota(ln, "traducido", txt, "→ `tls <cert> <key>`")
        elif cab == "proxy_pass" and loc is not None:
            loc["dirs"].append(("proxy", p[1] if len(p) > 1 else ""))
            anota(ln, "traducido", txt, "→ `reverse_proxy`")
        elif cab == "try_files" and loc is not None:
            loc["dirs"].append(("try_files", " ".join(p[1:])))
            anota(ln, "traducido", txt, "→ `try_files`")
        elif cab == "return":
            destino = " ".join(p[1:])
            (loc["dirs"] if loc is not None else actual.setdefault("dirs", [])).append(("return", destino))
            anota(ln, "traducido", txt, "→ `redir`")
        elif cab in ("proxy_set_header", "add_header"):
            anota(ln, "decision", txt,
                  "Caddy pone por su cuenta las cabeceras de proxy habituales (X-Forwarded-*). Si "
                  "ésta es de las habituales, sobra; si no, hay que escribirla con `header_up`")
        elif cab in ("access_log", "error_log", "client_max_body_size", "expires", "autoindex"):
            anota(ln, "decision", txt, "tiene equivalente en Caddy pero con otra forma: revisalo")
        else:
            anota(ln, "SIN-TRADUCIR", txt, "directiva no reconocida por este traductor")

    # ── emitir Caddyfile ────────────────────────────────────────────────────────────────────────
    L = ["# Generado por scripts/mudanza/traducir.py desde una configuración de nginx.",
         "# ⚠ NO es una traducción completa. Leé el ACTA: lo que no se pudo traducir está ahí con su",
         "#   número de línea. Desplegar esto sin mirar el acta es desplegar un servidor incompleto.",
         ""]
    for s in sitios:
        # En nginx dos `server` con el MISMO server_name se distinguen por su `listen`; en Caddy se
        # distinguen por el ESQUEMA de la dirección. Sin traducir eso, los dos bloques colisionan y
        # caddy rechaza el fichero con «ambiguous site definition» — encontrado al validar con el
        # binario real, no al leer la salida.
        solo_80 = bool(s["puertos"]) and all(p.split(":")[-1] == "80" for p in s["puertos"])
        pref = "http://" if solo_80 else ""
        nombres = ", ".join(pref + n for n in s["nombres"]) or (":80" if solo_80 else ":443")
        L.append(f"{nombres} {{")
        L.append(f"\t# desde el `server {{` de la línea {s['linea']} del nginx.conf")
        if s["tls"].get("ssl_certificate"):
            L.append(f"\ttls {s['tls'].get('ssl_certificate','')} {s['tls'].get('ssl_certificate_key','')}")
        if s.get("root"):
            L.append(f"\troot * {s['root']}")
        for k, v in s.get("dirs", []):
            if k == "return":
                code, *dest = v.split()
                L.append(f"\tredir {' '.join(dest)} {code}" if dest else f"\trespond {code}")
        for lo in s["locations"]:
            cuerpo = []
            for k, v in lo["dirs"]:
                if k == "proxy":
                    cuerpo.append(f"\t\treverse_proxy {v.rstrip('/').replace('http://','').replace('https://','')}")
                elif k == "root":
                    cuerpo.append(f"\t\troot * {v}")
                elif k == "try_files":
                    cuerpo.append(f"\t\ttry_files {v}")
                elif k == "return":
                    code, *dest = v.split()
                    cuerpo.append(f"\t\tredir {' '.join(dest)} {code}" if dest else f"\t\trespond {code}")
            if cuerpo:
                L.append(f"\thandle {lo['ruta']} {{    # location de la línea {lo['linea']}")
                L += cuerpo
                L.append("\t}")
        if s.get("root"):
            L.append("\tfile_server")
        L.append("}")
        L.append("")
    return "\n".join(L), acta


def imprimir_acta(acta):
    por = {}
    for a in acta:
        por.setdefault(a["estado"], []).append(a)
    print("\n══ ACTA DE TRADUCCIÓN ══")
    print(f"   {len(acta)} directiva(s) leídas\n")
    orden = ["SIN-TRADUCIR", "decision", "traducido", "no-aplica"]
    titulo = {"SIN-TRADUCIR": "⚠ NO SE PUDO TRADUCIR — hay que escribirlas a mano",
              "decision": "◆ traducidas con una DECISIÓN — revisalas",
              "traducido": "✓ traducidas",
              "no-aplica": "· no aplican en Caddy"}
    for e in orden:
        if e not in por:
            continue
        print(f"── {titulo[e]} ({len(por[e])}) " + "─" * 24)
        for a in por[e] if e in ("SIN-TRADUCIR", "decision") else por[e][:6]:
            print(f"   línea {a['linea']:>4}  {a['directiva'][:64]}")
            if a["motivo"] and e in ("SIN-TRADUCIR", "decision"):
                print(f"              └─ {a['motivo'][:100]}")
        if e not in ("SIN-TRADUCIR", "decision") and len(por[e]) > 6:
            print(f"   … y {len(por[e])-6} más")
        print()
    n = len(por.get("SIN-TRADUCIR", []))
    if n:
        print(f"   ⚠ {n} directiva(s) sin traducir. Desplegar sin revisarlas es desplegar un servidor")
        print("     INCOMPLETO que además parece funcionar — el modo de fallo más caro.")
    return n


def main():
    ap = argparse.ArgumentParser(description="Traduce la configuración de un servicio a su equivalente.")
    ap.add_argument("--from", dest="desde", required=True, choices=["nginx"])
    ap.add_argument("--to", dest="hacia", required=True, choices=["caddy"])
    ap.add_argument("--in", dest="entrada", required=True)
    ap.add_argument("--out", dest="salida", help="fichero de salida (default: sólo el acta)")
    a = ap.parse_args()

    texto = open(a.entrada).read()
    salida, acta = nginx_a_caddy(texto)
    n = imprimir_acta(acta)
    if a.salida:
        open(a.salida, "w").write(salida + "\n")
        print(f"\n   {a.hacia} → {a.salida}")
    else:
        print(salida)
    sys.exit(1 if n else 0)


if __name__ == "__main__":
    main()