takana/scripts/mudanza/formatos/systemd.py

"""Lector de units de systemd → modelo pivote «servicio» (SDD 29).

systemd es el init de la mayoría de los servidores de los que uno se muda, así que éste es el lector
que más entradas abre. Pero un unit dice MUCHO más de lo que cualquier otro init sabe expresar, y
ahí está todo el riesgo: traducir sólo `ExecStart=` y callarse el resto produce un servicio que
arranca y **hace otra cosa**.

⚠ Los tres campos donde una elisión silenciosa no pierde una opción sino que CAMBIA el sistema:

  · `User=` / `Group=` — sin esto el servicio corre como root. No es «una opción menos»: es
    escalación de privilegios en un fichero generado, que nadie vuelve a leer.
  · `EnvironmentFile=` — el fichero no existe en la máquina donde se traduce, así que su contenido
    no se puede ni mirar. Un servicio al que le falta medio entorno arranca y falla tarde.
  · Los especificadores (`%i`, `%h`, `%n`…) y las plantillas `foo@.service` — no se resuelven sin
    saber la instancia. Expandirlos «por lo más probable» es adivinar en silencio.

Los tres van al acta como SIN-TRADUCIR, con su línea.
"""
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from modelo import Servicio, Sistema                       # noqa: E402
from formatos import lector                                # noqa: E402

# `Restart=` de systemd → el vocabulario del pivote. `on-abnormal`/`on-watchdog`/`on-abort` son
# matices de «cuándo cuenta como fallo» que ningún otro init distingue: se acercan a `si-falla` y
# eso se ANOTA, no se hace callado.
_REINICIO = {"no": "no", "always": "siempre", "on-failure": "si-falla", "on-success": "si-falla",
             "on-abnormal": "si-falla", "on-watchdog": "si-falla", "on-abort": "si-falla"}
_MATIZ = {"on-abnormal", "on-watchdog", "on-abort", "on-success"}

# Prefijos de `ExecStart=`. Cada uno cambia la semántica del arranque y ninguno es parte del comando.
_PREFIJOS = {"-": "ignorar el fallo de arranque", "@": "argv[0] distinto del ejecutable",
             "+": "correr con privilegios completos", "!": "saltar la restricción de usuario",
             ":": "no expandir variables de entorno"}


def _seg(val):
    """`RestartSec=` admite `5`, `5s`, `500ms`, `1min`. Devuelve milisegundos."""
    v = val.strip().lower()
    for suf, mult in (("ms", 1), ("min", 60000), ("s", 1000), ("m", 60000)):
        if v.endswith(suf):
            try:
                return int(float(v[:-len(suf)]) * mult)
            except ValueError:
                return 0
    try:
        return int(float(v) * 1000)
    except ValueError:
        return 0


def _entorno(val):
    """`Environment="A=1" B=2` — comillas opcionales, varios pares por línea."""
    import shlex
    d = {}
    try:
        piezas = shlex.split(val)
    except ValueError:
        piezas = val.split()
    for p in piezas:
        if "=" in p:
            k, _, v = p.partition("=")
            d[k] = v
    return d


@lector("systemd", familia="servicio")
def leer(texto, nombre=""):
    sis = Sistema(formato_origen="systemd")
    acta = sis.acta
    base = os.path.basename(nombre or "unidad.service")
    srv = Servicio(nombre=base[:-8] if base.endswith(".service") else base)
    srv.plantilla = "@." in base
    if srv.plantilla:
        acta.anota(0, "SIN-TRADUCIR", base,
                   "es una PLANTILLA (`nombre@.service`): sin saber la instancia, `%i` no se resuelve "
                   "y el servicio traducido arrancaría con otro argumento")

    seccion, arranques, extras = "", [], []
    lineas, i = texto.splitlines(), 0
    while i < len(lineas):
        n = i + 1
        cruda = lineas[i]
        # Continuación de línea: systemd la usa muchísimo en ExecStart largos, y cortar ahí deja un
        # comando TRUNCADO que además parece válido.
        while cruda.rstrip().endswith("\\") and i + 1 < len(lineas):
            i += 1
            cruda = cruda.rstrip()[:-1] + " " + lineas[i].strip()
        i += 1
        ln = cruda.strip()
        if not ln or ln[0] in "#;":
            continue
        if ln.startswith("[") and ln.endswith("]"):
            seccion = ln[1:-1]
            continue
        if "=" not in ln:
            acta.anota(n, "SIN-TRADUCIR", ln, "línea sin `clave=valor` y fuera de toda sección")
            continue
        k, _, v = ln.partition("=")
        k, v = k.strip(), v.strip()

        if k == "Description":
            srv.descripcion = v
            acta.anota(n, "traducido", ln)
        elif k == "Type":
            srv.tipo = v
            srv.lineas["tipo"] = n
            acta.anota(n, "traducido", ln)
        elif k == "ExecStart":
            pref = ""
            while v and v[0] in _PREFIJOS:
                pref += v[0]
                v = v[1:].lstrip()
            if pref:
                acta.anota(n, "SIN-TRADUCIR", ln,
                           "prefijo `%s` en ExecStart (%s): el pivote no lo modela y quitarlo "
                           "cambia el arranque" % (pref, "; ".join(_PREFIJOS[c] for c in pref)))
            arranques.append((n, v))
        elif k in ("ExecStartPre", "ExecStartPost", "ExecStop", "ExecStopPost", "ExecReload",
                   "ExecCondition"):
            extras.append((n, k, v))
        elif k == "WorkingDirectory":
            srv.cwd = v
            srv.lineas["cwd"] = n
            acta.anota(n, "traducido", ln)
        elif k in ("User", "Group"):
            campo = "usuario" if k == "User" else "grupo"
            setattr(srv, campo, v)
            srv.lineas[campo] = n
            # Sin anotación: si el destino sabe bajar de privilegio esto es `traducido`, y si no es
            # SIN-TRADUCIR. Lo sabe el ESCRITOR. Que el lector opine acá duplica la entrada y la
            # contradice — y un acta que se contradice se deja de leer.
        elif k == "Environment":
            srv.entorno.update(_entorno(v))
            acta.anota(n, "traducido", ln)
        elif k == "EnvironmentFile":
            srv.entorno_ficheros.append(v)
            srv.lineas["entorno_ficheros"] = n
            acta.anota(n, "SIN-TRADUCIR", ln,
                       "el fichero de entorno NO está en esta máquina: su contenido no se puede "
                       "leer ni copiar, y el servicio arrancaría con medio entorno")
        elif k == "Restart":
            if v in _MATIZ:
                acta.anota(n, "decision", ln,
                           f"`{v}` es un matiz de «cuándo cuenta como fallo» que ningún otro init "
                           "distingue; se traduce como «si-falla»")
            elif v not in _REINICIO:
                acta.anota(n, "SIN-TRADUCIR", ln, f"valor de Restart no reconocido: `{v}`")
            else:
                acta.anota(n, "traducido", ln)
            srv.reinicio = _REINICIO.get(v, "si-falla")
            srv.lineas["reinicio"] = n
        elif k == "RestartSec":
            srv.reinicio_ms = _seg(v)
            acta.anota(n, "traducido", ln)
        elif k == "After":
            srv.tras += v.split()
            acta.anota(n, "traducido", ln)
        elif k in ("Requires", "BindsTo"):
            srv.necesita += v.split()
            acta.anota(n, "traducido", ln)
        elif k == "WantedBy":
            srv.arranque_auto = True
            acta.anota(n, "traducido", ln)
        elif k in ("Wants", "Before", "Conflicts", "PartOf", "RequiredBy", "Also",
                   "DefaultDependencies", "StartLimitIntervalSec", "StartLimitBurst"):
            acta.anota(n, "no-aplica", ln, "orden/relación que el destino resuelve por el grafo")
        elif k in ("StandardOutput", "StandardError", "SyslogIdentifier"):
            acta.anota(n, "no-aplica", ln, "el destino captura la salida por su propio journal")
        elif k in ("Documentation", "RemainAfterExit", "KillMode", "KillSignal", "TimeoutStopSec",
                   "TimeoutStartSec", "PIDFile", "Alias"):
            acta.anota(n, "no-aplica", ln, f"`{k}` es específico de systemd")
        elif seccion == "Service":
            # Todo el endurecimiento (`ProtectSystem`, `NoNewPrivileges`, `CapabilityBoundingSet`…)
            # cae acá. Callarlo sería entregar un servicio MENOS confinado que el original.
            acta.anota(n, "SIN-TRADUCIR", ln,
                       f"`{k}` no se modela; si es confinamiento, el servicio traducido corre con "
                       "MÁS permisos que el original")
        else:
            acta.anota(n, "SIN-TRADUCIR", ln, f"`{k}` en [{seccion or 'sin sección'}] no reconocida")

    if not arranques:
        acta.anota(0, "SIN-TRADUCIR", base, "el unit no tiene `ExecStart=`: no hay qué arrancar")
    else:
        srv.cmd = arranques[0][1]
        srv.origen = arranques[0][0]
        acta.anota(srv.origen, "traducido", f"ExecStart={srv.cmd}")
        for n, v in arranques[1:]:
            acta.anota(n, "SIN-TRADUCIR", f"ExecStart={v}",
                       "systemd admite varios ExecStart en `Type=oneshot`; el pivote modela UNO")
    for n, k, v in extras:
        acta.anota(n, "SIN-TRADUCIR", f"{k}={v}",
                   f"`{k}` es una fase del ciclo de vida aparte del arranque; hay que portarla a mano")

    for especificador in ("%i", "%I", "%n", "%N", "%h", "%H", "%f", "%p", "%t"):
        if especificador in srv.cmd:
            acta.anota(0, "SIN-TRADUCIR", srv.cmd,
                       f"el comando usa el especificador `{especificador}`, que systemd expande en "
                       "tiempo de arranque; resolverlo acá sería adivinar")
            break

    sis.servicios.append(srv)
    return sis