takana/scripts/mudanza/formatos/nginx.py

"""Plugin de formato: LECTOR de nginx (SDD 29)."""
import os, sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from modelo import Config, Sitio, Ruta, Accion          # noqa: E402
from formatos import lector                             # noqa: E402

# Bloques cuya semántica depende del orden de evaluación o del contexto: no se traducen, y —esto es
# lo importante— TODO lo de adentro tampoco. Un `return 403` sacado de su `if` deja de ser
# condicional y pasa a aplicar siempre: no se pierde información, se CAMBIA el sentido.
OPACOS = {"map", "if", "stream", "upstream", "geo", "split_clients"}

# Ajustes del proceso nginx que no tienen ni necesitan equivalente.
GLOBALES_IRRELEVANTES = {
    "worker_processes", "worker_connections", "user", "pid", "include", "sendfile",
    "keepalive_timeout", "gzip", "default_type", "log_format", "types_hash_max_size",
    "tcp_nopush", "multi_accept", "server_tokens", "charset",
}


def _tokenizar(texto):
    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


@lector("nginx")
def leer(texto):
    cfg = Config(formato_origen="nginx")
    A = cfg.acta
    pila, sitio, ruta, opaco = [], None, None, 0

    for ln, _niv, txt, tipo in _tokenizar(texto):
        cab = txt.split()[0] if txt else ""

        if tipo == "abre":
            pila.append(cab)
            if cab in OPACOS:
                opaco += 1
                A.anota(ln, "SIN-TRADUCIR", txt + " {",
                        f"`{cab}` depende del orden de evaluación y del contexto; TODO lo de adentro "
                        f"queda sin traducir — sacarlo del bloque le cambiaría el sentido")
            elif cab == "server" and not opaco:
                sitio = Sitio(origen=ln)
            elif cab == "location" and not opaco:
                p = txt.split()
                if len(p) > 2:   # modificador: ~ ~* ^~ =
                    clase = {"~": "regex", "~*": "regex", "=": "exacto", "^~": "prefijo"}.get(p[1], "regex")
                    ruta = Ruta(patron=p[2], clase=clase, origen=ln)
                else:
                    ruta = Ruta(patron=p[1] if len(p) > 1 else "/", clase="prefijo", origen=ln)
            continue

        if tipo == "cierra":
            c = pila.pop() if pila else ""
            if c in OPACOS and opaco:
                opaco -= 1
            elif c == "location" and sitio is not None and ruta is not None:
                sitio.rutas.append(ruta); ruta = None
            elif c == "server" and sitio is not None:
                cfg.sitios.append(sitio); sitio = None
            continue

        if opaco:
            A.anota(ln, "SIN-TRADUCIR", txt,
                    "está dentro de un bloque que no se pudo traducir: sacarla de ahí dejaría de ser "
                    "condicional")
            continue

        p = txt.split()
        destino = ruta if ruta is not None else sitio
        if sitio is None:
            if cab in GLOBALES_IRRELEVANTES or cab.endswith("_log"):
                A.anota(ln, "no-aplica", txt, "ajuste global del proceso nginx, sin equivalente necesario")
            elif txt:
                A.anota(ln, "SIN-TRADUCIR", txt, "directiva global no reconocida")
            continue

        if cab == "server_name":
            sitio.nombres += [x for x in p[1:] if x != "_"]
            A.anota(ln, "traducido", txt)
        elif cab == "listen":
            sitio.puertos.append(p[1].split(":")[-1] if len(p) > 1 else "")
            A.anota(ln, "traducido", txt)
        elif cab == "root":
            if ruta is not None:
                ruta.acciones.append(Accion("archivos", p[1] if len(p) > 1 else ""))
            else:
                sitio.raiz = p[1] if len(p) > 1 else ""
            A.anota(ln, "traducido", txt)
        elif cab == "index":
            sitio.indices = p[1:]
            A.anota(ln, "traducido", txt)
        elif cab in ("ssl_certificate", "ssl_certificate_key"):
            sitio.tls["cert" if cab == "ssl_certificate" else "key"] = p[1] if len(p) > 1 else ""
            A.anota(ln, "traducido", txt)
        elif cab == "proxy_pass":
            destino.acciones.append(Accion("proxy", p[1] if len(p) > 1 else ""))
            A.anota(ln, "traducido", txt)
        elif cab == "try_files":
            destino.acciones.append(Accion("intentar", " ".join(p[1:])))
            A.anota(ln, "traducido", txt)
        elif cab == "return":
            destino.acciones.append(Accion("redir", " ".join(p[1:])))
            A.anota(ln, "traducido", txt)
        elif cab in ("proxy_set_header", "add_header"):
            A.anota(ln, "decision", txt,
                    "las cabeceras de proxy habituales (X-Forwarded-*) las pone el destino por su "
                    "cuenta; si ésta no es de ésas, hay que escribirla")
        elif cab in ("client_max_body_size", "expires", "autoindex", "access_log", "error_log"):
            A.anota(ln, "decision", txt, "tiene equivalente pero con otra forma: revisalo")
        else:
            A.anota(ln, "SIN-TRADUCIR", txt, "directiva no reconocida por este lector")
    return cfg