takana/scripts/mudanza/formatos/caddy.py

"""Plugin de formato: Caddy — ESCRITOR y LECTOR (SDD 29)."""
import os, re, sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from formatos import escritor, lector                   # noqa: E402


def _direcciones(s):
    """En nginx dos `server` con el MISMO nombre se distinguen por su `listen`; en Caddy, por el
    ESQUEMA de la dirección. Sin traducir eso, los bloques colisionan y caddy rechaza el fichero
    entero con «ambiguous site definition» — encontrado validando con el binario real, no leyendo."""
    solo_80 = bool(s.puertos) and all(p == "80" for p in s.puertos)
    pref = "http://" if solo_80 else ""
    return ", ".join(pref + n for n in s.nombres) or (":80" if solo_80 else ":443")


def _emitir_acciones(acciones, ind):
    L = []
    for a in acciones:
        if a.tipo == "proxy":
            L.append(f"{ind}reverse_proxy {a.valor.rstrip('/').replace('http://','').replace('https://','')}")
        elif a.tipo == "archivos":
            L.append(f"{ind}root * {a.valor}")
        elif a.tipo == "intentar":
            L.append(f"{ind}try_files {a.valor}")
        elif a.tipo == "redir":
            p = a.valor.split()
            code = p[0] if p and p[0].isdigit() else ""
            dest = " ".join(p[1:]) if code else a.valor
            L.append(f"{ind}redir {dest} {code}".rstrip() if dest else f"{ind}respond {code}")
    return L


@escritor("caddy", familia="web")
def escribir(cfg):
    A = cfg.acta
    L = [f"# Generado por scripts/mudanza/traducir.py desde una configuración de {cfg.formato_origen}.",
         "# ⚠ NO es una traducción completa. Abajo va el ACTA de lo que NO se pudo traducir, con su",
         "#   línea de origen. Desplegar esto sin resolverlo es desplegar un servidor incompleto.",
         ""]
    for s in cfg.sitios:
        L.append(f"{_direcciones(s)} {{")
        L.append(f"\t# desde la línea {s.origen} del {cfg.formato_origen} original")
        if s.tls.get("cert"):
            L.append(f"\ttls {s.tls.get('cert','')} {s.tls.get('key','')}")
        if s.raiz:
            L.append(f"\troot * {s.raiz}")
        L += _emitir_acciones(s.acciones, "\t")
        for r in s.rutas:
            if r.clase == "regex":
                # Caddy no tiene un equivalente directo de `location ~`. Se DECLARA en el acta en vez
                # de emitir algo parecido: un `handle` de prefijo no es lo mismo que una regex, y
                # parecerse es peor que faltar.
                A.anota(r.origen, "SIN-TRADUCIR", f"ruta por regex: {r.patron}",
                        "Caddy no tiene equivalente directo de `location ~`; se escribe con "
                        "`@matcher path_regexp` y hay que revisar el orden de evaluación")
                continue
            cuerpo = _emitir_acciones(r.acciones, "\t\t")
            if cuerpo:
                L.append(f"\thandle {r.patron} {{    # línea {r.origen}")
                L += cuerpo
                L.append("\t}")
        if s.raiz:
            L.append("\tfile_server")
        L.append("}")
        L.append("")

    pend = A.sin_traducir()
    if pend:
        L.append("# ══ ACTA: lo que NO se tradujo — hay que resolverlo a mano ══")
        for e in pend:
            L.append(f"#   línea {e['linea']:>4}  {e['directiva'][:70]}")
            if e["motivo"]:
                L.append(f"#             └─ {e['motivo'][:96]}")
    return "\n".join(L) + "\n"


# ── LECTOR ──────────────────────────────────────────────────────────────────────────────────────
# Caddy se lee además de escribirse por dos motivos: mucha gente se muda DESDE Caddy, y —el que
# empujó a escribirlo— el CENSO necesita saber a qué directorio apunta cada sitio. Un bloque cuyo
# `root` no existe es un fósil que sirve 502, y eso se detecta leyendo el fichero: sin red, sin
# pedirle una página al servidor, sin arriesgar su fail2ban.

_DIRECTIVAS_SITIO = {"root", "reverse_proxy", "file_server", "redir", "respond", "tls", "try_files",
                     "handle", "handle_path", "encode", "header", "log", "php_fastcgi", "rewrite",
                     "route", "import", "basicauth", "abort", "@"}


# Una dirección de sitio de Caddy: `ejemplo.com`, `:8080`, `http://x.com:81`, `*.ejemplo.com`.
_DIR = re.compile(r'^(?:(?:https?|h2c)://)?(?:\*\.)?[A-Za-z0-9_.-]+(?::\d+)?/?$|^:\d+$')


def _es_cabecera_de_sitio(ln):
    """`ejemplo.com, www.ejemplo.com {` — una cabecera de sitio, no una directiva con bloque.

    ⚠ La distinción no es cosmética, y una lista negra de directivas NO alcanza: probado contra el
    Caddyfile real, `log { output file /var/log/…` y el snippet `(acceso) {` pasaban por sitios, o
    sea DOS dominios inventados que el censo habría puesto a decidir. La regla que sí aguanta es
    positiva: **todos** los tokens de la cabecera tienen que PARECER una dirección."""
    if not ln.endswith("{") or ln.startswith(("}", "@", "#", "(")):
        return False
    cab = ln[:-1].strip()
    if not cab:
        return False
    piezas = [x for x in cab.replace(",", " ").split() if x]
    if not piezas or piezas[0].lstrip("@") in _DIRECTIVAS_SITIO:
        return False
    return all(_DIR.match(x) for x in piezas)


@lector("caddy", familia="web")
def leer(texto, nombre=""):
    from modelo import Config, Sitio, Ruta, Accion
    cfg = Config(formato_origen="caddy")
    acta = cfg.acta
    sitio, prof, anidado = None, 0, ""
    for n, cruda in enumerate(texto.splitlines(), 1):
        ln = cruda.strip()
        if not ln or ln.startswith("#"):
            continue
        if sitio is None:
            if _es_cabecera_de_sitio(ln):
                sitio, prof = Sitio(origen=n), 1
                for d in ln[:-1].replace(",", " ").split():
                    d = d.strip()
                    esquema = d.split("://")[0] if "://" in d else ""
                    host = d.split("://")[-1]
                    host, _, puerto = host.partition(":")
                    if puerto:
                        sitio.puertos.append(puerto)
                    elif esquema:
                        sitio.puertos.append("80" if esquema == "http" else "443")
                    if host:
                        sitio.nombres.append(host)
                acta.anota(n, "traducido", ln)
            elif ln.endswith("{"):
                prof += 1
            elif ln == "}":
                prof = max(0, prof - 1)
            else:
                # Opciones globales, `import`, snippets: no son un sitio y no se pierden en silencio.
                acta.anota(n, "no-aplica", ln, "fuera de todo bloque de sitio")
            continue

        if ln.endswith("{"):
            prof += 1
            anidado = ln[:-1].strip()
            acta.anota(n, "SIN-TRADUCIR", ln,
                       "bloque anidado (`handle`, `route`, matcher…): el pivote no lo modela. De su "
                       "interior sólo se leen `root` y `reverse_proxy`")
            continue
        if ln == "}":
            prof -= 1
            if prof == 0:
                cfg.sitios.append(sitio)
                sitio = None
            continue
        if prof > 1:
            # ⚠ El bloque no se traduce, pero lo que APUNTA A DISCO sí se lee. Medido: el `root` de
            # `git.tawasuyu.net` vive dentro de un `handle`, así que saltar el bloque entero hacía
            # que el censo diera `/var/www/git-tawasuyu` por huérfano estando servido — y ese aviso,
            # aplicado, tira el sitio. Que el pivote no sepa MODELAR algo no es razón para no VERLO.
            pa = ln.split()
            if pa[0] == "root" and len(pa) >= 2:
                r = Ruta(patron=anidado, clase="prefijo", origen=n)
                r.acciones.append(Accion("archivos", pa[-1]))
                sitio.rutas.append(r)
            elif pa[0] == "reverse_proxy" and len(pa) >= 2:
                r = Ruta(patron=anidado, clase="prefijo", origen=n)
                r.acciones.append(Accion("proxy", pa[-1]))
                sitio.rutas.append(r)
            continue

        p = ln.split()
        d = p[0]
        if d == "root" and len(p) >= 2:
            sitio.raiz = p[-1]
            acta.anota(n, "traducido", ln)
        elif d == "reverse_proxy" and len(p) >= 2:
            sitio.acciones.append(Accion("proxy", p[-1]))
            acta.anota(n, "traducido", ln)
        elif d == "file_server":
            sitio.acciones.append(Accion("archivos", sitio.raiz))
            acta.anota(n, "traducido", ln)
        elif d == "redir" and len(p) >= 2:
            sitio.acciones.append(Accion("redir", " ".join(p[1:])))
            acta.anota(n, "traducido", ln)
        elif d == "tls" and len(p) >= 3:
            sitio.tls = {"cert": p[1], "key": p[2]}
            acta.anota(n, "traducido", ln)
        elif d == "try_files":
            sitio.acciones.append(Accion("intentar", " ".join(p[1:])))
            acta.anota(n, "traducido", ln)
        elif d in ("encode", "log", "header", "tls"):
            acta.anota(n, "no-aplica", ln, f"`{d}` es afinado del servidor, no de lo que sirve")
        else:
            acta.anota(n, "SIN-TRADUCIR", ln, f"directiva `{d}` no reconocida")

    if sitio is not None:
        acta.anota(sitio.origen, "SIN-TRADUCIR", f"sitio {', '.join(sitio.nombres)}",
                   "el bloque no se cierra: falta una llave. Se descarta, porque un sitio a medias "
                   "traducido es peor que uno ausente")
    return cfg