takana/scripts/test-atuq-descargas.py

#!/usr/bin/env python3
"""¿Lo que baja `atuq` tiene identidad, y sabe que ya lo tenías? (SDD 26 §6.2)

    python3 scripts/test-atuq-descargas.py                    # dos descargas del MISMO contenido
    python3 scripts/test-atuq-descargas.py --negative-control  # la segunda es OTRO contenido

── QUÉ SE EJERCITA ───────────────────────────────────────────────────────────────────────────
Un servidor HTTP sirve un fichero como adjunto; el navegador lo baja de verdad (no es una llamada a
la API: es la descarga completa, con su `Content-Disposition`); la extensión `descargas@atuq` ve que
terminó, le pasa al host la ruta absoluta, y el host lo ingiere a un CAS BLAKE3 y contesta el hash,
el tamaño y **cuántas veces se bajó ese mismo contenido**.

    servidor HTTP → descarga real → downloads.onChanged → connectNative → puriy-costura
                  → arje-cas (BLAKE3, deduplicado)

── LAS TRES AFIRMACIONES ─────────────────────────────────────────────────────────────────────
    1ª descarga  contenido A, nombre `uno.bin`   ⇒ dedup=false, veces=1
    2ª descarga  contenido A, nombre `dos.bin`   ⇒ **dedup=true, veces=2** y UN solo blob en disco
    el fichero que bajó el usuario sigue donde estaba, byte a byte

La segunda es la del §6.2: «lo bajás dos veces y no lo sabés». El nombre cambia y el objeto es el
mismo, porque la identidad es el contenido.

`--negative-control` baja OTRO contenido la segunda vez y exige `dedup=false` y DOS objetos. Sin él,
una sonda rota que dijera «ya lo tenías» siempre se vería idéntica a una que funciona.

── ⚠ POR QUÉ ESTO CORRE EN UN SWAY DE VERDAD Y NO EN `--headless` ────────────────────────────
Porque en `--headless` **el navegador se cae con SIGSEGV en cuanto una descarga termina**, y eso se
midió acá antes de escribir una línea de este guardián:

    con la extensión de descargas    → baja el fichero, TERMINADA, Segmentation fault (139)
    SIN la extensión (control)       → baja el fichero, Segmentation fault (139)
    con el panel de descargas apagado→ Segmentation fault (139)

O sea que **no es nuestro**: es el modo headless de Gecko. Y no es del producto tampoco: con un
compositor real —sway headless, salida offscreen— la misma descarga completa la cadena entera sin
caerse. La conclusión importa más que el arnés: si esto se hubiera dado por «la función no anda», se
habría perseguido un bug que no existe; y si se hubiera dado por «el navegador no puede descargar»,
se habría reportado un bug de producto que tampoco existe. Queda anotado en el §6.10 del SDD, que es
donde vive lo que atuq NO puede hacer.

REQUISITOS: el rootfs de sway (`scripts/hydrate-profile.py escritorio-sway --into …`), y `atuq` +
`puriy-costura` sellados.
"""
import http.server
import json
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import threading

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
RFS = os.environ.get("RFS", "/mnt/cosecha/escritorios/sway-rootfs")
ALPINE = os.path.join(ROOT, ".dev-fs/alpine")
HAMMER = os.path.join(ROOT, "target/release/takana")
STORE = os.environ.get("HAMMER_STORE", os.path.join(ROOT, "store"))

CONTENIDO_A = b"un objeto con identidad, no un archivo con nombre\n" * 64
CONTENIDO_B = b"otro contenido cualquiera, para el control negativo\n" * 64


def fatal(msg):
    print(f"✗ {msg}", file=sys.stderr)
    raise SystemExit(1)


def artefacto(receta, nombre, escotilla=None):
    """El artefacto VIGENTE, por `takana hash` y nunca por glob. La escotilla grita."""
    forzado = os.environ.get(escotilla) if escotilla else None
    if forzado:
        if not os.path.isdir(forzado):
            fatal(f"{escotilla}={forzado} no existe")
        print(f"⚠ {escotilla}: se mide {os.path.basename(forzado)}, que NO es necesariamente el vigente")
        return forzado
    r = subprocess.run([HAMMER, "--store", STORE, "hash", f"recipes/{receta}.toml"],
                       cwd=ROOT, capture_output=True, text=True)
    if r.returncode != 0:
        fatal(f"no pude hashear {receta}: {r.stderr.strip()}")
    h = r.stdout.strip().splitlines()[-1].removeprefix("b3:")
    d = os.path.join(STORE, f"{h}-{nombre}")
    if not os.path.isdir(d):
        fatal(f"{nombre} b3:{h[:8]} no está sellado — construilo antes")
    return d


class Servidor(http.server.SimpleHTTPRequestHandler):
    """Sirve cada ruta como ADJUNTO. El `Content-Disposition` es lo que hace que el navegador lo
    baje en vez de pintarlo: sin él esto mediría otra cosa."""

    contenidos = {}

    def log_message(self, *a):
        pass

    def do_GET(self):
        nombre = self.path.lstrip("/").split("?")[0]
        datos = Servidor.contenidos.get(nombre)
        if datos is None:
            self.send_error(404)
            return
        self.send_response(200)
        self.send_header("Content-Type", "application/octet-stream")
        self.send_header("Content-Disposition", f'attachment; filename="{nombre}"')
        self.send_header("Content-Length", str(len(datos)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(datos)


def puerto_libre():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    p = s.getsockname()[1]
    s.close()
    return p


def correr(atuq, costura, salida, urls):
    """Levanta UN sway headless y baja, dentro de él, las URLs en orden (un atuq por descarga).

    Un solo compositor para las dos descargas: levantarlo dos veces costaría el doble y no mediría
    nada más. El perfil y el `$HOME` se reusan, así que la segunda descarga encuentra el estado que
    dejó la primera — que es justo lo que hace falta para que la deduplicación signifique algo.
    """
    bajadas = " ".join(f"'{u}'" for u in urls)
    dentro = f"""#!/bin/sh
export XDG_RUNTIME_DIR=/run/user/0 HOME=/salida/hogar LANG=C XDG_DATA_DIRS=/usr/share
export WLR_BACKENDS=headless WLR_HEADLESS_OUTPUTS=1 WLR_RENDERER=pixman
mkdir -p "$XDG_RUNTIME_DIR" /salida/hogar /salida/bajadas; chmod 700 "$XDG_RUNTIME_DIR"
(
  i=0
  while [ ! -S /run/user/0/wayland-1 ] && [ $i -lt 40 ]; do sleep 1; i=$((i+1)); done
  [ -S /run/user/0/wayland-1 ] || {{ echo "@@ sin socket wayland"; exit 1; }}
  export WAYLAND_DISPLAY=wayland-1
  /usr/bin/swaybg -c '#1a4b8c' & sleep 3
  export MOZ_ENABLE_WAYLAND=1 GDK_BACKEND=wayland LIBGL_DRIVERS_PATH=/usr/lib/dri
  # Los sandboxes de Gecko se apagan porque esto YA corre dentro de una jaula sin userns anidado:
  # con ellos puestos los procesos hijos mueren en bucle. Es el entorno de prueba, no el artefacto.
  export MOZ_DISABLE_CONTENT_SANDBOX=1 MOZ_DISABLE_RDD_SANDBOX=1 MOZ_DISABLE_UTILITY_SANDBOX=1
  for u in {bajadas}; do
    # La marca va AL MISMO log que lee el guardián: si sale por la consola, el
    # parser no tiene con qué separar una descarga de la otra.
    echo "@@ bajando $u" >> /salida/atuq.log
    timeout 60 /usr/bin/atuq --no-remote --profile /salida/perfil "$u" >> /salida/atuq.log 2>&1
    echo "@@ atuq salió con $?" >> /salida/atuq.log
  done
  sleep 2
  # El socket de sway se busca explícitamente: esta subshell arrancó ANTES que sway
  # (es hija del shell que después lo `exec`a), así que no hereda `SWAYSOCK` y sin él
  # `swaymsg exit` no hace nada — la jaula se quedaría viva hasta el timeout externo.
  SWAYSOCK=$(ls /run/user/0/sway-ipc.* 2>/dev/null | head -1)
  export SWAYSOCK
  /usr/bin/swaymsg exit 2>/dev/null
) &
exec /usr/bin/sway -d > /salida/sway.log 2>&1
"""
    guion = os.path.join(salida, "dentro.sh")
    with open(guion, "w") as fh:
        fh.write(dentro)
    orden = [
        "bwrap",
        "--overlay-src", ALPINE, "--overlay-src", RFS,
        "--overlay-src", atuq, "--overlay-src", costura,
        "--tmp-overlay", "/",
        "--dev-bind", "/dev", "/dev", "--proc", "/proc", "--bind", "/sys", "/sys",
        "--tmpfs", "/tmp", "--tmpfs", "/run", "--unshare-pid",
        "--bind", salida, "/salida",
        "--uid", "0", "--gid", "0",
        "--setenv", "PATH", "/usr/bin:/bin:/usr/sbin:/sbin",
        "/bin/sh", "/salida/dentro.sh",
    ]
    try:
        subprocess.run(orden, check=False, timeout=300)
    except subprocess.TimeoutExpired:
        # El compositor no se cerró solo. No es motivo para perder la corrida: las
        # descargas y el log ya están en disco, y las aserciones son sobre eso.
        print("   ⚠ el compositor no se cerró solo; se lo cortó a los 300 s")
    ruta = os.path.join(salida, "atuq.log")
    log = open(ruta, errors="replace").read() if os.path.exists(ruta) else ""
    # Las líneas de cada descarga, separadas por el `@@ bajando` que las precede.
    tandas, actual = [], None
    for l in log.splitlines():
        if l.startswith("@@ bajando"):
            actual = []
            tandas.append(actual)
        elif l.startswith("DESCARGAS ") and actual is not None:
            actual.append(l[len("DESCARGAS "):])
    return tandas, log


def main():
    negativo = "--negative-control" in sys.argv[1:]
    if negativo:
        print("== MODO CONTROL NEGATIVO: la segunda descarga es OTRO contenido; NO debe deduplicar")
    if not os.path.isdir(RFS):
        fatal(f"no está el rootfs en {RFS} — corré antes `scripts/atuq-nested.sh`, que lo hidrata")
    atuq = artefacto("atuq", "atuq", "ATUQ_DIR")
    costura = artefacto("puriy-costura", "puriy-costura")
    print(f"artefactos: {os.path.basename(atuq)}\n            {os.path.basename(costura)}")

    segundo = CONTENIDO_B if negativo else CONTENIDO_A
    Servidor.contenidos = {"uno.bin": CONTENIDO_A, "dos.bin": segundo}
    puerto = puerto_libre()
    servidor = http.server.ThreadingHTTPServer(("127.0.0.1", puerto), Servidor)
    threading.Thread(target=servidor.serve_forever, daemon=True).start()

    tmp = tempfile.mkdtemp(prefix="atuq-desc-")
    salida = os.path.join(tmp, "salida")
    perfil = os.path.join(salida, "perfil")
    os.makedirs(perfil)
    try:
        with open(os.path.join(perfil, "user.js"), "w") as fh:
            fh.write('user_pref("browser.dom.window.dump.enabled", true);\n')
            fh.write('user_pref("browser.shell.checkDefaultBrowser", false);\n')
            # Bajar SIN preguntar y a un directorio conocido. Sin esto el navegador
            # headless se queda esperando un diálogo que nadie va a contestar, y la
            # prueba mediría un timeout en vez de una descarga.
            fh.write('user_pref("browser.download.folderList", 2);\n')
            fh.write('user_pref("browser.download.dir", "/salida/bajadas");\n')
            fh.write('user_pref("browser.download.useDownloadDir", true);\n')
            fh.write('user_pref("browser.download.always_ask_before_handling_new_types", false);\n')
            fh.write('user_pref("browser.helperApps.neverAsk.saveToDisk", "application/octet-stream");\n')
            fh.write('user_pref("browser.download.start_downloads_in_tmp_dir", false);\n')

        urls = [f"http://127.0.0.1:{puerto}/uno.bin", f"http://127.0.0.1:{puerto}/dos.bin"]
        bajadas, log = correr(atuq, costura, salida, urls)
        for i, lineas in enumerate(bajadas):
            print(f"\n-- descarga {('uno.bin', 'dos.bin')[i] if i < 2 else i}")
            for l in lineas:
                print("   " + l)
        if len(bajadas) != 2 or not all(any(l.startswith("INGERIDO ") for l in t) for t in bajadas):
            destino = os.path.join(ROOT, "work", "atuq-descargas.log")
            os.makedirs(os.path.dirname(destino), exist_ok=True)
            open(destino, "w").write(log)
            fatal(f"alguna descarga no llegó al CAS — log en {destino}")

        def ingerido(lineas):
            for l in lineas:
                if l.startswith("INGERIDO "):
                    partes = l.split()
                    campos = dict(p.split("=", 1) for p in partes if "=" in p)
                    campos["hash"] = partes[1]
                    return campos
            return None

        uno, dos = ingerido(bajadas[0]), ingerido(bajadas[1])
        cas = os.path.join(salida, "hogar", ".local", "state", "puriy-costura", "descargas-cas")
        indice = os.path.join(salida, "hogar", ".local", "state", "puriy-costura", "descargas.json")

        if uno["dedup"] != "false" or uno["veces"] != "1":
            fatal(f"la PRIMERA descarga no debía estar en el CAS: {uno}")

        # El fichero del usuario sigue donde estaba, y con sus bytes. Es la promesa
        # explícita del §6.2: el CAS es una COPIA, no una mudanza.
        for nombre, contenido in (("uno.bin", CONTENIDO_A), ("dos.bin", segundo)):
            f = os.path.join(salida, "bajadas", nombre)
            if not os.path.isfile(f):
                fatal(f"el fichero bajado {nombre} ya no está donde el usuario lo dejó")
            if open(f, "rb").read() != contenido:
                fatal(f"el fichero bajado {nombre} no es byte a byte lo que se sirvió")

        blobs = sorted(os.listdir(cas)) if os.path.isdir(cas) else []
        objetos = json.load(open(indice)) if os.path.isfile(indice) else {}

        if negativo:
            if dos["dedup"] != "false":
                fatal(f"con OTRO contenido dijo dedup={dos['dedup']} — esta prueba no distingue nada")
            if dos["hash"] == uno["hash"]:
                fatal("dos contenidos distintos dieron el MISMO hash")
            if len(blobs) != 2 or len(objetos) != 2:
                fatal(f"se esperaban DOS objetos y hay blobs={blobs} índice={list(objetos)}")
            print("\n✓ control negativo: otro contenido es otro objeto, y no dice que ya lo tenías")
            return 0

        if dos["dedup"] != "true":
            fatal(f"la SEGUNDA descarga del mismo contenido dijo dedup={dos['dedup']}"
                  " — que es justo lo que ningún navegador sabe hacer")
        if dos["hash"] != uno["hash"]:
            fatal(f"el mismo contenido dio dos hashes: {uno['hash']} y {dos['hash']}")
        if dos["veces"] != "2":
            fatal(f"no contó las dos veces: {dos}")
        if dos["nombre"] != "uno.bin":
            fatal(f"el nombre tenía que ser el de la PRIMERA vez y es {dos['nombre']}")
        if blobs != [uno["hash"]]:
            fatal(f"el CAS tiene {blobs} y tenía que tener UN blob llamado {uno['hash']}")
        if list(objetos) != [uno["hash"]] or objetos[uno["hash"]]["times"] != 2:
            fatal(f"el índice no cuenta un objeto bajado dos veces: {objetos}")

        # Y el blob es el contenido servido, byte a byte, bajo un nombre que es su hash.
        if open(os.path.join(cas, uno["hash"]), "rb").read() != CONTENIDO_A:
            fatal("el blob del CAS no es el contenido que sirvió el servidor")

        print(f"\nCAS: {blobs[0][:16]}… ({len(CONTENIDO_A)} bytes) · índice: {list(objetos)[0][:16]}…"
              f" veces={objetos[uno['hash']]['times']}")
        print("\n✓ la descarga entra al CAS con su BLAKE3, y el fichero del usuario no se toca")
        print("✓ el MISMO contenido con otro nombre es UN objeto, y el navegador lo dice"
              " («esto ya lo tenías»)")
        print("✓ el índice queda legible en disco: la lista de lo que bajaste sobrevive al binario")
        return 0
    finally:
        servidor.shutdown()
        shutil.rmtree(tmp, ignore_errors=True)


sys.exit(main())