takana/scripts/sh-superficie-cards.py

#!/usr/bin/env python3
"""Mide la superficie REAL de shell de las cards de arje — SDD: docs/plan-botar-busybox.md.

La pregunta que contesta: **si mañana hubiera un `/bin/sh` propio, ¿qué tendría que saber hacer
para que el PRODUCTO arranque?** No «qué dice POSIX», no «qué usa autotools»: qué usan las cards
que arje ejecuta de verdad, contado sobre el árbol.

Poblaciones (las cuatro que emiten cards en este repo):

  1. `recipes/*.toml` · bloques `[[service]]` — el emisor de verdad (`Service::card`,
     `crates/takana-core/src/service.rs`), uno por servicio de paquete.
  2. `crates/takana-bootstrap/src/lib.rs` · las cards empotradas como constantes `r#"…"#`
     (`STAGE1_SEED_CARD` con su génesis, `SSHD_SERVICE_CARD`).
  3. `scripts/servidor/cards/*.json` · cards escritas a mano para el servidor de producción.
  4. `scripts/mudanza/declarar.py` · el envoltorio `sh -c 'cd … && exec …'` que la mudanza le
     pone a todo servicio con `cwd` (SDD 29). No se cuenta por card porque depende del censo:
     se cuenta como PLANTILLA, y su superficie es la de la plantilla.

De cada card se saca el fragmento de shell (el `argv` que sigue a `sh -c`), y de cada fragmento:

  - **qué construcciones del lenguaje usa** (bucles, `case`, sustitución, aritmética, redirecciones,
    grupos, …) — detección por patrón sobre texto tokenizado, no por parseo: aproximada y dicha así;
  - **qué órdenes invoca**, separadas en *builtins* (las tiene que traer el shell) y *externas*
    (las tiene que traer la imagen, y son la lista que necesita el guardián de applets que el plan
    dice que falta).

Uso:
    scripts/sh-superficie-cards.py                 # resumen + TSV
    scripts/sh-superficie-cards.py --list          # además, cada fragmento entero
    scripts/sh-superficie-cards.py --dump-dir DIR  # escribe un .sh por fragmento, para dárselos
                                                   # a sh-banco-diferencial.py --files
"""

from __future__ import annotations

import argparse
import collections
import json
import re
import shlex
import sys
import tomllib
from dataclasses import dataclass, field
from pathlib import Path

RAIZ = Path(__file__).resolve().parent.parent

# Los shells que una card puede invocar, y el multicall que los trae.
SHELLS = {"sh", "ash", "bash", "busybox"}


@dataclass
class Fragmento:
    origen: str       # recipes | bootstrap | servidor | mudanza
    fuente: str       # fichero:cosa
    label: str
    exec_: str
    argv: list[str]
    codigo: str | None = None      # el texto que corre el shell; None = no hay shell
    plantilla: bool = False        # el fragmento es una plantilla, no una card real
    rasgos: set[str] = field(default_factory=set)
    builtins: collections.Counter = field(default_factory=collections.Counter)
    externas: collections.Counter = field(default_factory=collections.Counter)


# ───────────────────────────────────────────────── extracción

def frag_de_payload(origen: str, fuente: str, card: dict) -> Fragmento | None:
    payload = card.get("payload")
    if not isinstance(payload, dict):
        return None
    nativo = payload.get("Native")
    if not isinstance(nativo, dict):
        return None
    ejec = nativo.get("exec", "")
    argv = [str(a) for a in nativo.get("argv", [])]
    f = Fragmento(origen=origen, fuente=fuente, label=str(card.get("label", "?")),
                  exec_=str(ejec), argv=argv)
    f.codigo = codigo_de_argv(str(ejec), argv)
    return f


def codigo_de_argv(ejecutable: str, argv: list[str]) -> str | None:
    """El texto que el shell va a interpretar, o None si la card no invoca un shell.

    ⚠ En una Card, **`argv` va SIN `argv[0]`**: arje lo pone desde `exec`. Lo dice la card de
    `hammerd`, cuyo `argv` es `["--store", "/store", …]` a secas. Por eso los dos idiomas que
    conviven en el corpus sólo se leen igual si se antepone el `exec`:

        exec=/bin/busybox  argv=["sh","-c",cod]   ← recetas y bootstrap (despacho multicall)
        exec=/bin/sh       argv=["-c",cod]        ← la mudanza (formatos/arje.py:35)

    `getty -n -l /bin/sh` NO cuenta: ahí el shell es interactivo y no ejecuta un fragmento.
    """
    linea = [Path(ejecutable).name] + [str(a) for a in argv]
    if linea[0] == "busybox" and len(linea) > 1:
        linea = linea[1:]
    if linea[0] in SHELLS and "-c" in linea:
        i = linea.index("-c")
        if i + 1 < len(linea):
            return linea[i + 1]
    return None


def de_recetas() -> list[Fragmento]:
    frags = []
    for ruta in sorted(RAIZ.glob("recipes/*.toml")):
        try:
            datos = tomllib.loads(ruta.read_text(encoding="utf-8", errors="replace"))
        except Exception:
            continue
        servicios = datos.get("service")
        if isinstance(servicios, dict):
            servicios = [servicios]
        if not isinstance(servicios, list):
            continue
        for s in servicios:
            if not isinstance(s, dict):
                continue
            argv = [str(a) for a in s.get("argv", [])]
            f = Fragmento(origen="recipes", fuente=ruta.name,
                          label=str(s.get("label", "?")), exec_=str(s.get("exec", "")), argv=argv)
            f.codigo = codigo_de_argv(str(s.get("exec", "")), argv)
            frags.append(f)
    return frags


RE_CONST = re.compile(r'const\s+([A-Z_0-9]+)\s*:\s*&str\s*=\s*r#"(.*?)"#;', re.S)


def de_bootstrap() -> list[Fragmento]:
    """Las cards empotradas como constantes. Una seed trae varias cards en `genesis`."""
    frags = []
    for ruta in sorted(RAIZ.glob("crates/*/src/*.rs")):
        texto = ruta.read_text(encoding="utf-8", errors="replace")
        for nombre, cuerpo in RE_CONST.findall(texto):
            try:
                v = json.loads(cuerpo)
            except Exception:
                continue
            pendientes = [v]
            while pendientes:
                card = pendientes.pop()
                if not isinstance(card, dict):
                    continue
                for hija in card.get("genesis", []) or []:
                    pendientes.append(hija)
                f = frag_de_payload("bootstrap", f"{ruta.name}:{nombre}", card)
                if f:
                    frags.append(f)
    return frags


def de_servidor() -> list[Fragmento]:
    frags = []
    for ruta in sorted(RAIZ.glob("scripts/**/cards/*.json")):
        try:
            v = json.loads(ruta.read_text(encoding="utf-8", errors="replace"))
        except Exception:
            continue
        f = frag_de_payload("servidor", str(ruta.relative_to(RAIZ)), v)
        if f:
            frags.append(f)
    return frags


RE_ENVOLTORIO = re.compile(r'''["']-c["']\s*,\s*f["'](.*?)["']\s*\]''')
RE_HUECO = re.compile(r"\{[^{}]*\}")


def de_mudanza() -> list[Fragmento]:
    """El envoltorio `sh -c 'cd … && exec …'` que la mudanza le pone a todo servicio con `cwd`
    (`scripts/mudanza/formatos/arje.py:35`, SDD 29, regla 2 de `declarar.py`).

    Es una PLANTILLA: el contenido sale del censo de la máquina que se muda, así que se mide la
    forma y no una card concreta. Los `{…}` de la f-string se sustituyen por un marcador para
    poder analizarla como shell — y el marcador es deliberadamente inofensivo, porque lo que la
    plantilla interpole es justo lo que no está medido."""
    frags = []
    for ruta in sorted(RAIZ.glob("scripts/mudanza/**/*.py")):
        texto = ruta.read_text(encoding="utf-8", errors="replace")
        for i, cod in enumerate(RE_ENVOLTORIO.findall(texto)):
            if not cod.strip():
                continue
            concreto = RE_HUECO.sub("VALOR", cod)
            frags.append(Fragmento(
                origen="mudanza", fuente=str(ruta.relative_to(RAIZ)), label=f"plantilla-{i+1}",
                exec_="/bin/sh", argv=["-c", concreto], codigo=concreto, plantilla=True))
    return frags


# ───────────────────────────────────────────────── análisis

def piezas_shell(codigo: str) -> list[tuple[str, str, str]]:
    """Descompone el código en PIEZAS y devuelve, por pieza, `(original, sintaxis, expansiones)`.

    Una pieza es un nivel de shell: el código de la card, y el cuerpo de cada sustitución de
    comando — que es shell OTRA VEZ y casi siempre vive dentro de comillas dobles. Sin esto, el
    `df -h /vvv | awk …` de `montar-trabajo` no existía para la medición: ni su tubería ni sus dos
    órdenes externas.

    Los tres textos, y por qué hacen falta tres:

    - **original**: el texto tal cual, con el cuerpo de cada sustitución retirado (queda en su
      propia pieza, así no se cuenta dos veces). Va a `shlex`, que sí entiende comillas, para
      sacar las ÓRDENES.
    - **sintaxis**: lo de fuera de comillas. Ahí se buscan `;`, `&&`, `|`, `{ }`, `>`, `for`,
      `case`, `*`. Sin separar por comillas, un mensaje de error que dice «lo genera
      `minga init`» se cuenta como sustitución de comando — pasó, y daba 11 backticks falsos de
      23 cards, más una docena de «órdenes externas» que eran palabras sueltas de la prosa.
    - **expansiones**: lo que el shell EXPANDE, o sea lo de fuera de comillas más lo de dentro de
      comillas DOBLES. Ahí se buscan `$( )`, `` ` ``, `$(( ))`, `${…}`, `$1`.

    Escáner de estados, no regex: comillas simples (literal absoluto), dobles (expanden), barra
    invertida y anidamiento de paréntesis.
    """
    piezas: list[tuple[str, str, str]] = []

    def escanear(texto: str) -> None:
        orig, sin, exp = [], [], []
        estado = None  # None | "'" | '"'
        i = 0
        while i < len(texto):
            c = texto[i]
            if estado is None and c == "\\" and i + 1 < len(texto):
                orig.append(texto[i:i + 2]); sin.append(" "); exp.append(" ")
                i += 2
                continue
            vivo = estado != "'"
            if vivo and texto.startswith("$(", i) and not texto.startswith("$((", i):
                prof, j = 1, i + 2
                while j < len(texto) and prof:
                    if texto[j] == "(":
                        prof += 1
                    elif texto[j] == ")":
                        prof -= 1
                    j += 1
                escanear(texto[i + 2:j - 1])
                orig.append(" _SUST_ "); sin.append(" "); exp.append("$(_)")
                i = j
                continue
            if vivo and c == "`":
                j = texto.find("`", i + 1)
                if j < 0:
                    j = len(texto)
                escanear(texto[i + 1:j])
                orig.append(" _SUST_ "); sin.append(" "); exp.append("`_`")
                i = j + 1
                continue
            if estado is None and c in "'\"":
                estado = c
                orig.append(c); sin.append("_")
                i += 1
                continue
            if estado is not None and c == estado:
                estado = None
                orig.append(c)
                i += 1
                continue
            orig.append(c)
            if estado is None:
                sin.append(c); exp.append(c)
            elif estado == '"':
                sin.append("_"); exp.append(c)
            else:
                sin.append("_")
            i += 1
        piezas.append(("".join(orig), "".join(sin), "".join(exp)))

    escanear(codigo)
    return piezas


# (rasgo, patrón, sobre-qué): "s" = sintaxis (fuera de comillas), "e" = expansiones.
RASGOS = [
    ("bucle for",           r"\bfor\b\s+\w+\s+in\b", "s"),
    ("bucle while",         r"\bwhile\b", "s"),
    ("bucle until",         r"\buntil\b", "s"),
    ("condicional if",      r"\bif\b", "s"),
    ("case",                r"\bcase\b", "s"),
    ("función",             r"\w+\s*\(\)\s*\{", "s"),
    ("sustitución $( )",    r"\$\((?!\()", "e"),
    ("sustitución `…`",     r"`", "e"),
    ("aritmética $(( ))",   r"\$\(\(", "e"),
    ("expansión ${…}",      r"\$\{", "e"),
    ("variable $x",         r"\$\w", "e"),
    ("posicionales $1 $@",  r"\$[0-9@*#]", "e"),
    ("estado $?",           r"\$\?", "e"),
    ("y/o && ||",           r"&&|\|\|", "s"),
    ("tubería |",           r"(?<![|&])\|(?!\|)", "s"),
    ("grupo { }",           r"\{[^{}]*;\s*\}", "s"),
    ("subshell ( )",        r"(?<![$(])\((?!\()", "s"),
    ("redirección >",       r"(?<![0-9<>&])>", "s"),
    ("redirección 2>",      r"\d>", "s"),
    ("aquí-doc <<",         r"<<", "s"),
    ("glob *",              r"\*", "s"),
    ("negación !",          r"(?:^|[;&|(])\s*!\s", "s"),
    ("secuencia ;",         r";", "s"),
    ("segundo plano &",     r"(?<![&>])&(?![&>])", "s"),
]

BUILTINS_POSIX = {
    ":", ".", "break", "continue", "eval", "exec", "exit", "export", "readonly", "return",
    "set", "shift", "times", "trap", "unset", "cd", "echo", "printf", "pwd", "read", "test",
    "[", "command", "getopts", "hash", "umask", "wait", "alias", "unalias", "type", "kill",
    "jobs", "fg", "bg", "ulimit", "local", "true", "false",
}

PALABRAS_CLAVE = {
    "if", "then", "elif", "else", "fi", "for", "while", "until", "do", "done", "case", "esac",
    "in", "{", "}", "(", ")", "!", "time", "function", "select",
}

SEPARADORES = {";", "&&", "||", "|", "&", "(", ")", "{", "}", "do", "then", "else", "elif"}


def ordenes(codigo: str) -> list[str]:
    """Palabras en posición de ORDEN. Tokeniza respetando comillas (shlex) y toma la primera
    palabra de cada comando simple. Aproximado: `shlex` no entiende la gramática del shell, sólo
    las comillas. Sirve para inventariar, no para decidir.

    Casos que sí se tratan porque ensuciaban la cuenta medida: `for h in …` daba `h` como orden,
    y el `!` de `[ ! -S … ]` daba `-S`.
    """
    try:
        lex = shlex.shlex(codigo, punctuation_chars=True, posix=True)
        lex.whitespace_split = True
        toks = list(lex)
    except ValueError:
        return []
    salida, esperando, saltar_hasta = [], True, None
    for t in toks:
        if saltar_hasta:
            if t in saltar_hasta:
                saltar_hasta = None
                esperando = t in {"do", ";"}
            continue
        if t in ("for", "case", "select"):
            saltar_hasta = {"do", ";", "in"} if t == "for" else {"in"}
            esperando = False
            continue
        if t in SEPARADORES:
            esperando = True
            continue
        if not esperando:
            continue
        if t in PALABRAS_CLAVE or t.startswith("-") or t == "_SUST_":
            continue
        if re.fullmatch(r"\w+=.*", t):       # asignación previa al comando
            continue
        salida.append(t)
        esperando = False
    return salida


def analizar(f: Fragmento) -> None:
    if not f.codigo:
        return
    piezas = piezas_shell(f.codigo)
    sintaxis = "\n".join(s for _, s, _ in piezas)
    expansiones = "\n".join(e for _, _, e in piezas)
    for nombre, patron, donde in RASGOS:
        if re.search(patron, sintaxis if donde == "s" else expansiones, re.M):
            f.rasgos.add(nombre)
    todas = []
    for original, _, _ in piezas:
        todas.extend(ordenes(original))
    for o in todas:
        base = Path(o).name if o.startswith("/") else o
        if base in BUILTINS_POSIX:
            f.builtins[base] += 1
        elif re.fullmatch(r"[\w.+-]+", base):
            f.externas[base] += 1


RUTA_APPLETS = RAIZ / "docs/state/busybox-applets.tsv"


def applets_conocidos() -> dict[str, tuple[str, str]]:
    """`applet → (destino, sigue|RETIRADO)`, del inventario del recorte 401 → 277.

    Cruzarlo con lo que invocan las cards ES el guardián que `plan-botar-busybox.md` dice que
    falta: hoy, una card que llame a un applet ya retirado del `defconfig` no falla en ningún
    test — falla **en el arranque de otra máquina, semanas después**.
    """
    if not RUTA_APPLETS.is_file():
        return {}
    filas = {}
    with RUTA_APPLETS.open(encoding="utf-8") as fh:
        cabecera = fh.readline().rstrip("\n").split("\t")
        try:
            i_dest = cabecera.index("destino")
            i_rec = [i for i, c in enumerate(cabecera) if c.startswith("recorte")][0]
        except (ValueError, IndexError):
            return {}
        for linea in fh:
            campos = linea.rstrip("\n").split("\t")
            if len(campos) > max(i_dest, i_rec):
                filas[campos[0]] = (campos[i_dest], campos[i_rec])
    return filas


# ───────────────────────────────────────────────── informe

def main() -> int:
    ap = argparse.ArgumentParser(description="Mide la superficie de shell de las cards de arje.")
    ap.add_argument("--list", action="store_true", help="volcar cada fragmento entero")
    ap.add_argument("--dump-dir", metavar="DIR",
                    help="escribir un .sh por fragmento (para sh-banco-diferencial.py --files)")
    ap.add_argument("--tsv", metavar="RUTA", default=str(RAIZ / "work/sh-superficie-cards.tsv"))
    args = ap.parse_args()

    frags = de_recetas() + de_bootstrap() + de_servidor() + de_mudanza()
    for f in frags:
        analizar(f)

    con_shell = [f for f in frags if f.codigo and not f.plantilla]
    sin_shell = [f for f in frags if not f.codigo]
    plantillas = [f for f in frags if f.plantilla]

    print(f"Cards analizadas: {len(frags) - len(plantillas)}   "
          f"(+{len(plantillas)} plantilla(s) de la mudanza)")
    por_origen = collections.Counter(f.origen for f in frags if not f.plantilla)
    for o, n in sorted(por_origen.items()):
        con = sum(1 for f in frags if f.origen == o and f.codigo and not f.plantilla)
        print(f"   {o:<10} {n:3d} cards · {con:3d} invocan un shell")
    print(f"\n   → **{len(con_shell)} de {len(frags)-len(plantillas)} cards "
          f"({100*len(con_shell)//max(1,len(frags)-len(plantillas))} %) ejecutan shell**; "
          f"{len(sin_shell)} son exec directo.")

    # ── rasgos
    print("\n── Construcciones usadas (sobre los fragmentos reales, sin plantillas)")
    hist = collections.Counter()
    for f in con_shell:
        hist.update(f.rasgos)
    ancho = max((len(r) for r in hist), default=10)
    for rasgo, n in hist.most_common():
        barra = "█" * n
        print(f"   {rasgo:<{ancho}}  {n:3d}  {barra}")
    no_usados = [r for r, _, _ in RASGOS if r not in hist]
    if no_usados:
        print(f"\n   NO aparece ninguna vez: {', '.join(no_usados)}")

    # ── órdenes
    bi, ex = collections.Counter(), collections.Counter()
    for f in con_shell:
        bi.update(f.builtins)
        ex.update(f.externas)
    print(f"\n── Builtins que el shell tiene que traer ({len(bi)})")
    print("   " + "  ".join(f"{k}×{v}" for k, v in bi.most_common()))
    print(f"\n── Órdenes EXTERNAS que la imagen tiene que traer ({len(ex)})")
    print("   " + "  ".join(f"{k}×{v}" for k, v in ex.most_common()))

    # ── el cruce con el recorte de busybox: acá vive el guardián que el plan dice que falta
    applets = applets_conocidos()
    if applets:
        retirados, siguen, ajenas = [], [], []
        for orden, n in ex.most_common():
            fila = applets.get(orden)
            if fila is None:
                ajenas.append(orden)
            elif fila[1] == "RETIRADO":
                retirados.append((orden, n, fila[0]))
            else:
                siguen.append(orden)
        print(f"\n── Cruce con el recorte del defconfig ({RUTA_APPLETS.name})")
        print(f"   siguen en busybox: {len(siguen)}  ·  no son applets (otro paquete): "
              f"{len(ajenas)}  ·  RETIRADAS por el recorte: {len(retirados)}")
        if ajenas:
            print(f"   no-applets: {', '.join(sorted(ajenas))}")
        if retirados:
            print("\n   ⛔ CARDS QUE INVOCAN UN APPLET QUE YA NO ESTÁ EN LA IMAGEN:")
            for orden, n, destino in retirados:
                quien = [f.label for f in con_shell if orden in f.externas]
                print(f"      {orden} (×{n}, dueño previsto: {destino}) ← {', '.join(quien)}")
        else:
            print("   ✓ ninguna card invoca un applet retirado por el recorte 401 → 277.")

    if plantillas:
        print("\n── Plantillas (la mudanza; la card real depende del censo)")
        for f in plantillas:
            print(f"   {f.fuente}:{f.label}  rasgos={sorted(f.rasgos)}")
            print(f"      {f.codigo}")

    if args.list:
        print("\n── Fragmentos")
        for f in sorted(con_shell, key=lambda x: (x.origen, x.label)):
            print(f"\n  [{f.origen}] {f.label}   ({f.fuente})")
            print(f"      exec={f.exec_}")
            print(f"      {f.codigo}")

    if args.dump_dir:
        d = Path(args.dump_dir)
        d.mkdir(parents=True, exist_ok=True)
        for f in con_shell:
            nombre = re.sub(r"[^\w.-]", "_", f"{f.origen}-{f.label}") + ".sh"
            (d / nombre).write_text(f.codigo + "\n", encoding="utf-8")
        print(f"\nFragmentos escritos en {d} ({len(con_shell)} ficheros)")

    tsv = Path(args.tsv)
    tsv.parent.mkdir(parents=True, exist_ok=True)
    with tsv.open("w", encoding="utf-8") as fh:
        fh.write("origen\tfuente\tlabel\tshell\tplantilla\trasgos\tbuiltins\texternas\tcodigo\n")
        for f in frags:
            fh.write(f"{f.origen}\t{f.fuente}\t{f.label}\t{'si' if f.codigo else 'no'}\t"
                     f"{'si' if f.plantilla else 'no'}\t{'|'.join(sorted(f.rasgos))}\t"
                     f"{'|'.join(sorted(f.builtins))}\t{'|'.join(sorted(f.externas))}\t"
                     f"{(f.codigo or '').replace(chr(9), ' ').replace(chr(10), ' ')}\n")
    print(f"\nTSV: {tsv}")
    return 0


if __name__ == "__main__":
    sys.exit(main())