#!/usr/bin/env python3 """Banco DIFERENCIAL para candidatos a /bin/sh — SDD: docs/plan-botar-busybox.md. No pregunta «¿este shell hace lo que yo creo que POSIX manda?» (eso es `scripts/sh-banco-posix.sh`, 57 casos escritos a mano: brush lo pasa 56/57 y aun así no construye una sola receta autotools). Pregunta otra cosa: ¿este shell hace LO MISMO que el control, sobre el material que hay? Cada caso se corre bajo N shells en un directorio limpio y se compara la tupla completa contra el control: (rc, stdout, ÁRBOL DE FICHEROS PRODUCIDO) ← divergencia DURA (stderr normalizado) ← AVISO: el texto de error diverge legítimamente El árbol de ficheros es la parte que importa y la que faltaba: el bug que tumbó a brush (POSIX 2.6.3, la barra invertida dentro de `…`) no se ve en el stdout de ningún caso sintético — se ve en el `libtool` que el shell ESCRIBE, 25 líneas con un nivel de escapado de menos. Un banco que sólo mira stdout lo deja pasar. Fuentes de casos (combinables): --cases DIR ficheros .sh sueltos (regresiones conocidas) --gen generador de tortura de comillas: producto cartesiano de forma de sustitución × contexto × profundidad de barras × carácter objetivo. Es el que encuentra el bug de brush SIN saber que existe --phases las fases [build.phases] de recipes/*.toml, en modo `sh -n` --files GLOB ficheros reales (configure, libtool, config.status) en `sh -n` --latency N N invocaciones de `sh -c :` por shell, sin comparar nada Uso: scripts/sh-banco-diferencial.py --gen \\ control=/store/2a2b…-busybox/bin/busybox \\ brush=/store/7152…-brush/usr/bin/brush El PRIMER shell es el control salvo que se nombre uno `control=` o se pase --control NOMBRE. Salida: tabla por shell + TSV. Estado de salida = número de divergencias duras del peor shell (0 = todos coinciden con el control). """ from __future__ import annotations import argparse import concurrent.futures import glob as globmod import hashlib import os import re import shutil import subprocess import sys import tempfile import time from dataclasses import dataclass, field from pathlib import Path RAIZ = Path(__file__).resolve().parent.parent # ─────────────────────────────────────────────────────────── modelo @dataclass class Shell: nombre: str binario: Path dir_sh: Path | None = None # dir con un symlink `sh` -> binario (argv[0]=sh) @property def sh(self) -> Path: return (self.dir_sh / "sh") if self.dir_sh else self.binario @dataclass class Caso: nombre: str origen: str # cases | gen | phases | files codigo: str modo: str = "run" # run | parse (parse = `sh -n`, no ejecuta) @dataclass class Resultado: rc: int out: str err: str arbol: str ms: float timeout: bool = False @dataclass class Divergencia: caso: Caso shell: str clase: str = "dura" # dura | aviso | no-determinista | inestable campos: list[str] = field(default_factory=list) # rc | stdout | arbol avisos: list[str] = field(default_factory=list) # stderr control: Resultado | None = None obtenido: Resultado | None = None # ─────────────────────────────────────────────────────────── casos TORTURA_OBJETIVOS = [ ("var", "$V"), ("backtick", "`"), ("barra", "\\"), ("comilla", '"'), ("literal", "x"), ] def generar_tortura_comillas() -> list[Caso]: """Producto cartesiano de la tortura de comillas. POSIX.1-2024 §2.6.3: dentro de `…` la barra invertida conserva su significado literal SALVO cuando precede a `$`, `` ` `` o `\\`, donde se ELIMINA antes de parsear el texto del comando. Dentro de $( ) la regla es otra: el contenido se parsea como un script completo. brush aplica la de $( ) a las dos — y eso es exactamente lo que este generador barre. """ casos: list[Caso] = [] for forma, abre, cierra in (("bt", "`", "`"), ("dollar", "$(", ")")): for ctx in ("desnudo", "comillado"): for prof in range(0, 5): for etiqueta, objetivo in TORTURA_OBJETIVOS: cuerpo = "\\" * prof + objetivo interior = f"echo {cuerpo}" sust = f"{abre}{interior}{cierra}" linea = f"echo [{sust}]" if ctx == "desnudo" else f'echo "[{sust}]"' casos.append(Caso( nombre=f"tortura/{forma}-{ctx}-b{prof}-{etiqueta}", origen="gen", codigo=f"V=hola\n{linea}\n", )) # Anidamientos: el otro sitio donde los shells se separan. anidados = { "anid/bt-en-dollar": 'V=hola\necho "[$(echo `echo \\\\$V`)]"\n', "anid/dollar-en-bt": 'V=hola\necho "[`echo $(echo \\\\$V)`]"\n', "anid/bt-en-bt": 'V=hola\necho "[`echo \\`echo $V\\``]"\n', "anid/dollar-en-dollar": 'V=hola\necho "[$(echo $(echo $V))]"\n', "anid/comilla-en-bt": 'V=hola\necho "[`echo "$V"`]"\n', } for nombre, codigo in anidados.items(): casos.append(Caso(nombre=nombre, origen="gen", codigo=codigo)) # El bucle REAL de config.status que rompió libtool (plan-botar-busybox.md). # No es un caso sintético: es el material que decide. casos.append(Caso( nombre="libtool/case-de-config-status", origen="gen", codigo=r'''ECHO="printf %s\\n" SED=sed GREP=grep sed_quote_subst='s/\([`"$\\]\)/\\\1/g' for var in ECHO SED GREP; do case `eval \\$ECHO \\""\\$$var"\\"` in *[\\\`\"\$]*) eval "lt_$var=\\\"\`\$ECHO \"\$$var\" | \$SED \"\$sed_quote_subst\"\`\\\"" ;; *) eval "lt_$var=\\\"\$$var\\\"" ;; esac done echo "ECHO=[$lt_ECHO]" echo "SED=[$lt_SED]" echo "GREP=[$lt_GREP]" ''', )) # Un caso que ESCRIBE un fichero: verifica que el árbol se compara de verdad. casos.append(Caso( nombre="libtool/escribe-fichero-escapado", origen="gen", codigo=r'''V='a"b`c\d$e' { echo "directo=$V" echo "bt=`echo \\$V`" echo "dollar=$(echo \\$V)" printf '%s\n' "$V" } > generado.txt ''', )) return casos def cargar_casos_dir(directorio: Path) -> list[Caso]: casos = [] for f in sorted(directorio.rglob("*.sh")): casos.append(Caso( nombre=f"cases/{f.relative_to(directorio)}", origen="cases", codigo=f.read_text(encoding="utf-8", errors="replace"), )) return casos RE_FASES = re.compile(r"\[build\.phases\](.*?)(?=\n\[|\Z)", re.S) RE_CLAVE = re.compile(r'^\s*(\w+)\s*=\s*(.*)$', re.S) def cargar_fases(patron: str) -> list[Caso]: """Extrae las fases de shell de las recetas. Modo `parse`: no se ejecutan (necesitarían el sandbox y sus fuentes), pero el parser sí se mide.""" import tomllib casos = [] for ruta in sorted(globmod.glob(patron)): try: datos = tomllib.loads(Path(ruta).read_text(encoding="utf-8", errors="replace")) except Exception: continue fases = datos.get("build", {}).get("phases") if not isinstance(fases, dict): continue nombre_receta = Path(ruta).stem for fase, cuerpo in fases.items(): if isinstance(cuerpo, list): cuerpo = "\n".join(str(x) for x in cuerpo) if not isinstance(cuerpo, str) or not cuerpo.strip(): continue casos.append(Caso( nombre=f"phases/{nombre_receta}:{fase}", origen="phases", codigo=cuerpo if cuerpo.endswith("\n") else cuerpo + "\n", modo="parse", )) return casos def cargar_ficheros(patrones: list[str]) -> list[Caso]: casos = [] for patron in patrones: for ruta in sorted(globmod.glob(patron, recursive=True)): p = Path(ruta) if not p.is_file(): continue casos.append(Caso( nombre=f"files/{p.name}", origen="files", codigo=p.read_text(encoding="utf-8", errors="replace"), modo="parse", )) return casos # ─────────────────────────────────────────────────────────── ejecución def hash_arbol(raiz: Path) -> str: """Huella del árbol de ficheros producido: ruta + bit de ejecución + sha256 del contenido. Esto es lo que atrapa al libtool mal escapado.""" filas = [] for dirpath, dirnames, filenames in os.walk(raiz): dirnames.sort() for nombre in sorted(filenames): f = Path(dirpath) / nombre rel = f.relative_to(raiz).as_posix() try: if f.is_symlink(): filas.append(f"l {rel} {os.readlink(f)}") continue datos = f.read_bytes() except OSError as e: filas.append(f"e {rel} {e.errno}") continue ejec = "x" if os.access(f, os.X_OK) else "-" filas.append(f"f {rel} {ejec} {hashlib.sha256(datos).hexdigest()[:16]}") if not filas: return "(vacío)" return hashlib.sha256("\n".join(filas).encode()).hexdigest()[:16] + f" [{len(filas)}]" def entorno(shell: Shell, bindir: Path, home: Path) -> dict[str, str]: # PATH: primero el shell bajo prueba (para que un `sh` recursivo caiga en # ÉL), después SIEMPRE el mismo userland — las divergencias tienen que # venir del shell, no de un sed distinto. return { "PATH": f"{shell.dir_sh}:{bindir}" if shell.dir_sh else str(bindir), "HOME": str(home), "LC_ALL": "C", "LANG": "C", "TZ": "UTC", "SHELL": str(shell.sh), "TERM": "dumb", } def ejecutar(shell: Shell, caso: Caso, bindir: Path, timeout: float) -> Resultado: base = Path(tempfile.mkdtemp(prefix="shdif-")) try: prog = base / "prog.sh" prog.write_text(caso.codigo, encoding="utf-8") cwd = base / "cwd" cwd.mkdir() home = base / "home" home.mkdir() argv = [str(shell.sh)] if caso.modo == "parse": argv.append("-n") argv.append(str(prog)) t0 = time.monotonic() try: p = subprocess.run( argv, cwd=cwd, env=entorno(shell, bindir, home), stdin=subprocess.DEVNULL, capture_output=True, timeout=timeout, ) rc, out, err, to = p.returncode, p.stdout, p.stderr, False except subprocess.TimeoutExpired as e: rc, out, err, to = -9, e.stdout or b"", e.stderr or b"", True ms = (time.monotonic() - t0) * 1000 return Resultado( rc=rc, out=out.decode("utf-8", "replace"), err=err.decode("utf-8", "replace"), arbol=hash_arbol(cwd), ms=ms, timeout=to, ) finally: shutil.rmtree(base, ignore_errors=True) RE_RUIDO = [ (re.compile(r"/tmp/shdif-[A-Za-z0-9_]+"), ""), (re.compile(r"^[^\s:]*(?:sh|brush|bash|busybox)[^\s:]*:\s*", re.M), ""), (re.compile(r"\bline \d+\b"), "line N"), (re.compile(r"\blínea \d+\b"), "line N"), (re.compile(r"prog\.sh"), ""), ] def normalizar_err(texto: str) -> str: for rx, rep in RE_RUIDO: texto = rx.sub(rep, texto) return " ".join(texto.split()) def comparar(ctrl: Resultado, otro: Resultado) -> tuple[list[str], list[str]]: duras, avisos = [], [] if ctrl.rc != otro.rc: duras.append("rc") if ctrl.out != otro.out: duras.append("stdout") if ctrl.arbol != otro.arbol: duras.append("arbol") if normalizar_err(ctrl.err) != normalizar_err(otro.err): avisos.append("stderr") return duras, avisos # ─────────────────────────────────────────────────────────── informe def recorta(texto: str, n: int = 240) -> str: texto = texto.replace("\n", "\\n") return texto if len(texto) <= n else texto[:n] + f"…(+{len(texto)-n})" def medir_latencia(shells: list[Shell], n: int, bindir: Path) -> None: print(f"\n── Latencia de arranque · {n} invocaciones de `sh -c :`") base = Path(tempfile.mkdtemp(prefix="shdif-lat-")) try: for sh in shells: env = entorno(sh, bindir, base) subprocess.run([str(sh.sh), "-c", ":"], env=env, cwd=base, stdin=subprocess.DEVNULL, capture_output=True) # calentar t0 = time.monotonic() for _ in range(n): subprocess.run([str(sh.sh), "-c", ":"], env=env, cwd=base, stdin=subprocess.DEVNULL, capture_output=True) total = time.monotonic() - t0 tam = sh.binario.stat().st_size / 1024 / 1024 print(f" {sh.nombre:<12} {total/n*1e6:8.0f} µs/invocación " f"binario {tam:5.2f} MB") finally: shutil.rmtree(base, ignore_errors=True) print(" ⚠ incluye el fork+exec de Python: sirve para COMPARAR, no como cifra absoluta.") def main() -> int: ap = argparse.ArgumentParser( description="Banco diferencial de shells: compara rc, stdout y árbol producido " "contra un shell de control.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="ejemplo:\n %(prog)s --gen control=/store/…-busybox/bin/busybox " "brush=/store/…-brush/usr/bin/brush\n", ) ap.add_argument("shells", nargs="+", metavar="NOMBRE=RUTA", help="shells a comparar; el primero es el control salvo --control") ap.add_argument("--control", metavar="NOMBRE", help="cuál de los shells es el control") ap.add_argument("--cases", metavar="DIR", default=str(RAIZ / "scripts/fixtures/sh-casos"), help="directorio de casos .sh (por defecto: scripts/fixtures/sh-casos)") ap.add_argument("--no-cases", action="store_true", help="no cargar el directorio de casos") ap.add_argument("--gen", action="store_true", help="añadir el generador de tortura de comillas") ap.add_argument("--phases", action="store_true", help="añadir las fases [build.phases] de recipes/*.toml en modo `sh -n`") ap.add_argument("--files", metavar="GLOB", action="append", default=[], help="añadir ficheros reales (configure, libtool…) en modo `sh -n`") ap.add_argument("--latency", metavar="N", type=int, default=0, help="medir latencia de arranque con N invocaciones y salir") ap.add_argument("--timeout", metavar="SEG", type=float, default=10.0) ap.add_argument("--jobs", metavar="N", type=int, default=4) ap.add_argument("--tsv", metavar="RUTA", default=str(RAIZ / "work/sh-diferencial.tsv")) ap.add_argument("--verbose", action="store_true", help="volcar el diff de cada divergencia") ap.add_argument("--max-show", metavar="N", type=int, default=12, help="cuántas divergencias detallar por shell (0 = todas)") args = ap.parse_args() # ── shells tmp_sh = Path(tempfile.mkdtemp(prefix="shdif-bin-")) shells: list[Shell] = [] for spec in args.shells: if "=" not in spec: print(f"✗ shell mal declarado (falta NOMBRE=): {spec}", file=sys.stderr) return 2 nombre, ruta = spec.split("=", 1) binario = Path(ruta).resolve() if not binario.is_file() or not os.access(binario, os.X_OK): print(f"✗ no es un ejecutable: {binario}", file=sys.stderr) return 2 # symlink `sh` -> binario, para que argv[0] sea `sh` en todos por igual # (busybox despacha por argv[0]; brush también cambia de modo). d = tmp_sh / nombre d.mkdir(parents=True) (d / "sh").symlink_to(binario) shells.append(Shell(nombre=nombre, binario=binario, dir_sh=d)) nombre_control = args.control or ("control" if any(s.nombre == "control" for s in shells) else shells[0].nombre) try: control = next(s for s in shells if s.nombre == nombre_control) except StopIteration: print(f"✗ el control «{nombre_control}» no está entre los shells", file=sys.stderr) return 2 otros = [s for s in shells if s is not control] if not otros: print("✗ hace falta al menos un shell además del control", file=sys.stderr) return 2 # ── userland común: el busybox del control si lo es, o el del sistema bindir = tmp_sh / "_bin" bindir.mkdir() bb = None for s in shells: if "busybox" in s.binario.name or "busybox" in str(s.binario): bb = s.binario break if bb: subprocess.run([str(bb), "--install", "-s", str(bindir)], capture_output=True) # `sh` del userland NO: cada shell trae el suyo en su propio dir (bindir / "sh").unlink(missing_ok=True) (bindir / "ash").unlink(missing_ok=True) if not any(bindir.iterdir()): for herr in ("sed", "grep", "awk", "tr", "wc", "cat", "printf", "echo", "ls", "rm", "mkdir", "cp", "mv", "sort", "head", "tail", "cut", "expr", "test", "true", "false", "sleep", "env", "date", "basename", "dirname"): r = shutil.which(herr) if r: (bindir / herr).symlink_to(r) try: if args.latency: medir_latencia(shells, args.latency, bindir) return 0 # ── casos casos: list[Caso] = [] dir_casos = Path(args.cases) if not args.no_cases and dir_casos.is_dir(): casos += cargar_casos_dir(dir_casos) if args.gen: casos += generar_tortura_comillas() if args.phases: casos += cargar_fases(str(RAIZ / "recipes/*.toml")) if args.files: casos += cargar_ficheros(args.files) if not casos: print("✗ ningún caso cargado — pasá --gen, --phases, --files o poblá " f"{dir_casos}", file=sys.stderr) return 2 por_origen: dict[str, int] = {} for c in casos: por_origen[c.origen] = por_origen.get(c.origen, 0) + 1 print(f"Banco diferencial · {len(casos)} casos " f"({', '.join(f'{k}={v}' for k, v in sorted(por_origen.items()))})") print(f"Control: {control.nombre} → {control.binario}") for s in otros: print(f" vs {s.nombre} → {s.binario}") # ── corrida # # Toda divergencia dura se CONFIRMA volviendo a correr control y shell. Un caso que # imprime un PID o una fecha diverge siempre y no dice nada; un shell que diverge de sí # mismo es un hallazgo distinto (y peor). Sin esta confirmación el banco se llena de # ruido y deja de leerse — que es como muere un banco. def tarea(caso: Caso) -> tuple[Caso, dict[str, Resultado], list[Divergencia]]: res = {s.nombre: ejecutar(s, caso, bindir, args.timeout) for s in shells} ctrl = res[control.nombre] hallazgos: list[Divergencia] = [] for s in otros: duras, avisos = comparar(ctrl, res[s.nombre]) if duras: ctrl_bis = ejecutar(control, caso, bindir, args.timeout) otro_bis = ejecutar(s, caso, bindir, args.timeout) if comparar(ctrl, ctrl_bis)[0]: clase = "no-determinista" elif comparar(res[s.nombre], otro_bis)[0]: clase = "inestable" else: clase = "dura" elif avisos: clase = "aviso" else: continue hallazgos.append(Divergencia( caso=caso, shell=s.nombre, clase=clase, campos=duras, avisos=avisos, control=ctrl, obtenido=res[s.nombre], )) return caso, res, hallazgos divergencias: list[Divergencia] = [] cuenta: dict[str, dict[str, int]] = { s.nombre: {"dura": 0, "aviso": 0, "no-determinista": 0, "inestable": 0} for s in otros } timeouts: dict[str, int] = {s.nombre: 0 for s in shells} ms_por_shell: dict[str, float] = {s.nombre: 0.0 for s in shells} with concurrent.futures.ThreadPoolExecutor(max_workers=args.jobs) as ex: for i, (caso, res, hallazgos) in enumerate(ex.map(tarea, casos), 1): for s in shells: ms_por_shell[s.nombre] += res[s.nombre].ms if res[s.nombre].timeout: timeouts[s.nombre] += 1 for d in hallazgos: cuenta[d.shell][d.clase] += 1 divergencias.append(d) if i % 200 == 0: print(f" … {i}/{len(casos)}", file=sys.stderr) # ── informe print("\n── Resumen") for s in shells: if s is control: print(f" {s.nombre:<12} (control) " f"{ms_por_shell[s.nombre]/len(casos):6.1f} ms/caso" + (f" timeouts={timeouts[s.nombre]}" if timeouts[s.nombre] else "")) else: c = cuenta[s.nombre] extra = "" if c["no-determinista"]: extra += f" casos-no-deterministas={c['no-determinista']}" if c["inestable"]: extra += f" INESTABLE={c['inestable']}" if timeouts[s.nombre]: extra += f" timeouts={timeouts[s.nombre]}" print(f" {s.nombre:<12} DIVERGE={c['dura']:<5} avisos(stderr)={c['aviso']:<5} " f"{ms_por_shell[s.nombre]/len(casos):6.1f} ms/caso" + extra) for clase, titulo in (("no-determinista", "Casos NO DETERMINISTAS — el control no se " "repite a sí mismo; el caso está mal escrito, no el shell"), ("inestable", "Shell INESTABLE — no se repite a sí mismo. Es un " "hallazgo peor que una divergencia")): cuales = [d for d in divergencias if d.clase == clase] if cuales: print(f"\n── {titulo} ({len(cuales)})") for d in cuales: print(f" ⚠ [{d.shell}] {d.caso.nombre}") duras = [d for d in divergencias if d.clase == "dura"] if duras: print(f"\n── Divergencias DURAS ({len(duras)})") for s in otros: suyas = [d for d in duras if d.shell == s.nombre] if not suyas: continue mostrar = suyas if args.max_show == 0 else suyas[:args.max_show] for d in mostrar: print(f"\n ✗ [{d.shell}] {d.caso.nombre} ({', '.join(d.campos)})") if args.verbose or True: print(f" código: {recorta(d.caso.codigo)}") if "rc" in d.campos: print(f" rc: control={d.control.rc} " f"{d.shell}={d.obtenido.rc}") if "stdout" in d.campos: print(f" control: {recorta(d.control.out)}") print(f" {d.shell:<9} {recorta(d.obtenido.out)}") if "arbol" in d.campos: print(f" árbol: control={d.control.arbol} " f"{d.shell}={d.obtenido.arbol}") if args.max_show and len(suyas) > args.max_show: print(f"\n … y {len(suyas)-args.max_show} divergencias duras más de " f"{s.nombre} (ver el TSV; --max-show 0 las muestra todas)") # ── TSV tsv = Path(args.tsv) tsv.parent.mkdir(parents=True, exist_ok=True) with tsv.open("w", encoding="utf-8") as f: f.write("shell\tcaso\torigen\tmodo\tclase\tcampos\trc_control\trc_shell\t" "arbol_control\tarbol_shell\n") for d in divergencias: f.write(f"{d.shell}\t{d.caso.nombre}\t{d.caso.origen}\t{d.caso.modo}\t{d.clase}\t" f"{','.join(d.campos + ['stderr'] * bool(d.avisos))}\t" f"{d.control.rc}\t{d.obtenido.rc}\t" f"{d.control.arbol}\t{d.obtenido.arbol}\n") print(f"\nTSV: {tsv}") peor = max((c["dura"] for c in cuenta.values()), default=0) if peor == 0: print("\n✓ ningún shell diverge del control en rc, stdout ni árbol producido.") print(" ⚠ y eso NO es prueba de suficiencia: mide el material que se le dio.") return min(peor, 250) finally: shutil.rmtree(tmp_sh, ignore_errors=True) if __name__ == "__main__": sys.exit(main())