takana/scripts/test-atuq-boveda-metal.py

#!/usr/bin/env python3
"""¿La bóveda de `atuq` entrega una contraseña, y sólo si una persona dice que sí? (SDD 26 §7.sexies)

    python3 scripts/test-atuq-boveda-metal.py                 # las seis etapas, en orden
    python3 scripts/test-atuq-boveda-metal.py --hasta C       # sólo hasta la etapa C (el diálogo)
    python3 scripts/test-atuq-boveda-metal.py --list          # qué afirma cada etapa, sin correr nada

── POR QUÉ ESTE GUARDIÁN, Y POR QUÉ NO ALCANZA CON EL DE COHERENCIA ──────────────────────────
`test-atuq-boveda-coherente.py` mira los SEIS lugares donde una extensión se enchufa y tarda un
segundo. Es lo que hay que correr primero y no reemplaza a nada. Lo que no puede contestar es si del
otro lado del cable hay alguien: el 2026-09-18 los cinco lugares que entonces se miraban estaban
en orden —manifiesto, política, permiso, prefs y el commit del host con sus verbos— y la bóveda
contestaba **cerrada** en las cuatro imágenes, porque el DUEÑO de la base y el DIÁLOGO de
consentimiento no estaban en el corpus. Ningún guardián de FICHEROS podía verlo; éste empieza
justamente por ahí. (El sexto lugar nació de eso: desde el 2026-09-21 aquél mira además quién
DECLARA al dueño en la imagen, que es la mitad del §7.sexies que sí se puede ver sin construir
nada — y la que volvió a morder el 2026-09-21, con las dos recetas selladas y en `perfiles: []`.)

── LAS SEIS ETAPAS, Y QUÉ AFIRMA CADA UNA ────────────────────────────────────────────────────
    A  sin la app `boveda`, el host contesta `locked:true`   ← el control, y es el bug del §7.sexies
    B  con la app corriendo, `vault.status` dice ABIERTA     ← el dueño existe y atiende el socket
    C  `vault.save` abre el DIÁLOGO y se lo puede contestar  ← el consentimiento es real y se ve
    D  el navegador, sobre una página servida por HTTP, encuentra la credencial (`vault.match`)
    E  `vault.fill` contestado que SÍ ⇒ la contraseña llega al campo del formulario
    F  `vault.fill` contestado que NO ⇒ `denied`, y el campo queda VACÍO   ← el control de E

La etapa A no es decoración: **«cerrada» es una respuesta exitosa** (`ok:true`), así que sin ese
control un guardián que midiera sólo B no distinguiría «la bóveda abrió» de «nadie preguntó bien».
Y la F es la que hace que la E signifique algo: una bóveda que entrega siempre se ve idéntica a una
que entrega con permiso, y la diferencia es la única razón por la que esto existe.

── LO QUE HIZO FALTA PARA PODER CONTESTAR UNA PREGUNTA ───────────────────────────────────────
El corpus no tenía **ninguna** forma de meter una tecla en un compositor. Todos los arneses de
`scripts/wlr/` miran; ninguno toca. Entra `wtype` (protocolo `zwp_virtual_keyboard_v1`) como
INSTRUMENTO —no va en ninguna imagen— y con él se contesta el diálogo igual que lo haría una
persona: flechas y `Enter`.

⚠ **Y el render es por SOFTWARE a propósito, con `llvmpipe` y no con `softpipe`.** `shuma-pregunta`
y `boveda` son llimphi/wgpu, o sea clientes de GL; el `mesa` de las imágenes es **iris-only** (Intel)
y esta máquina tiene una virtio-gpu, así que EGL no encontraría driver. Se montan `mesa-llvmpipe` +
`llvm18` como últimas capas. Con `mesa-swrast` (softpipe, GL 3.3) el diálogo **no abre**: wgpu pide
compute shaders y muere con `RequestDevice("Parent device is lost")`, un mensaje que no nombra la
causa. Es el entorno de prueba, no el artefacto: en la imagen de verdad el driver lo pone el metal.

⚠⚠ **DÓNDE ESTÁ HOY (2026-09-18): la etapa A pasa y la B choca contra un muro que no es de la
bóveda.** `boveda` y `shuma-pregunta` están sellados, y ninguno de los dos abre ventana: llimphi
construye la instancia de wgpu **sin display handle** —su propio comentario lo dice: «este camino no
tiene ventana todavía»—, wgpu cae entonces a la plataforma EGL **surfaceless**, que por definición no
tiene configs de ventana, y `surface.get_capabilities()` devuelve **cero formatos**; llimphi hace
`caps.formats[0]` y panica. En una máquina con Vulkan no se nota, porque el backend PRIMARY no
necesita el display — y **ninguna imagen de takana tiene Vulkan**: las tres recetas de mesa se
compilan con `-Dvulkan-drivers=` vacío. Ver SDD 26 §7.septies. Este guardián queda escrito y
CORRIENDO hasta donde se puede: la A en verde es la medición del §7.sexies, vigilada.

REQUISITOS: el rootfs de sway (`scripts/hydrate-profile.py escritorio-sway --into …`) y, sellados,
`atuq`, `puriy-costura`, `boveda`, `shuma-pregunta`, `wtype`, `llvm18` y `mesa-llvmpipe`.
"""
import argparse
import http.server
import json
import os
import shutil
import socket
import struct
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"))

# La raíz de la bóveda DE PRUEBA. Con esta variable puesta, la app deriva sus claves de esta cadena
# en vez de la seed de identidad del llavero del kernel —que en ninguna máquina de desarrollo está
# desbloqueada— y abre `…/tawasuyu/boveda-de-prueba`, que es OTRA bóveda, vacía, y lo dice en
# pantalla. Sin esto este guardián no se podría encender en ningún lado.
RAIZ_DE_PRUEBA = "metal-de-atuq"

APPDIR = "usr/lib/atuq"

USUARIO = "sergio"
CLAVE = "Zx7#qLm2pVnR"
TITULO = "Banco de prueba"

ETAPAS = [
    ("A", "sin la app `boveda`, el host contesta locked:true (el bug del §7.sexies, vigilado)"),
    ("B", "con la app corriendo, vault.status dice ABIERTA: el dueño atiende el socket"),
    ("C", "el DIÁLOGO abre en el compositor y se lo puede CONTESTAR (el instrumento: wtype)"),
    ("D", "vault.save contestado que SÍ guarda, y vault.match la encuentra después"),
    ("E", "el navegador real: vault.fill contestado que SÍ ⇒ la contraseña llega al campo"),
    ("F", "vault.fill contestado que NO ⇒ denied y el campo queda VACÍO (el control de E)"),
]

# El atajo del diálogo: el dígito contesta con esa opción (`atajo_de_digito` en shuma-pregunta),
# así que «1» es la primera opción y «2» la segunda. Se teclea con `wtype`, que es la única forma
# que tiene este corpus de meter una tecla en un compositor.
TECLA_SI = "1"
TECLA_NO = "2"

# La página DELATA lo que le pongan en el campo, y lo delata al SERVIDOR.
#
# Es la forma barata de medir el final de la cadena sin plomería chrome↔contenido: si la contraseña
# llegó al formulario, el servidor la ve; si no llegó, no ve nada. Y mide lo que importa —el campo—
# y no lo que el host contestó, que es la distinción que ya costó una pérdida de datos silenciosa en
# este mismo frente (§7.quater: «el guardián leyó el DISCO, no la respuesta»).
PAGINA = """<!doctype html><html><head><meta charset="utf-8"><title>Banco de prueba</title></head>
<body><h1>Banco de prueba</h1>
<form id="f" method="POST" action="/entrar">
  <input id="u" name="username" type="text" autocomplete="username">
  <input id="p" name="password" type="password" autocomplete="current-password">
  <button id="b" type="submit">Entrar</button>
</form>
<script>
(function () {
  var avisado = false;
  setInterval(function () {
    var p = document.getElementById("p").value;
    if (p && !avisado) {
      avisado = true;
      var x = new XMLHttpRequest();
      x.open("POST", "/relleno", true);
      x.setRequestHeader("Content-Type", "text/plain");
      x.send(document.getElementById("u").value + "\\n" + p);
    }
  }, 300);
})();
</script>
</body></html>
"""



# ── La sonda del chrome: quien aprieta el botón de la extensión ───────────────────────────────
# Se APENDA al `atuq.cfg` del artefacto (no lo reemplaza), que es como lo hace `test-atuq-chrome.py`:
# así se mide el fichero real de atuq con la sonda encima y no una copia que pudo divergir.
#
# ⚠ Aprieta el botón por la API interna (`browserActionFor(...).triggerAction`) y no por un click en
# el DOM: desde las «extensiones unificadas» el botón vive en un panel que se construye recién al
# abrirlo, así que un `getElementById(...).click()` no encuentra nada y eso se lee como «la bóveda no
# ofreció». Si la API no estuviera, cae al DOM y DICE por cuál de los dos caminos fue — un camino que
# no se nombra es un resultado que no se puede interpretar.
SONDA_CHROME = """
// ── SONDA DE PRUEBA — NO VIAJA EN EL ARTEFACTO (scripts/test-atuq-boveda-metal.py) ────────────
try {
  const S = (typeof Services !== "undefined")
    ? Services
    : ChromeUtils.importESModule("resource://gre/modules/Services.sys.mjs").Services;
  const lineas = [];
  const di = function (t) {
    lineas.push(t);
    try { dump("BOVEDA-METAL " + t + "\\n"); } catch (e) {}
    try { IOUtils.writeUTF8("/salida/chrome.txt", lineas.join("\\n") + "\\n"); } catch (e) {}
  };
  di("VIVO");
  S.obs.addObserver(function observador(win) {
    S.obs.removeObserver(observador, "browser-delayed-startup-finished");
    win.setTimeout(function () {
      try {
        // ⚠ `WebExtensionPolicy` NO es global en el scope de una ventana del chrome: medido, la
        // sonda moría con «ReferenceError: WebExtensionPolicy is not defined» y eso se habría
        // leído como «la extensión no está». Sale del módulo, que además lo reexporta.
        const EP = ChromeUtils.importESModule(
          "resource://gre/modules/ExtensionParent.sys.mjs").ExtensionParent;
        const ext = EP.GlobalManager.getExtension("boveda@atuq.tawasuyu");
        di("EXTENSION " + (ext ? "cargada" : "AUSENTE"));
        if (!ext) { di("FIN"); return; }
        let disparo = "ninguno";
        try {
          const ba = EP.apiManager.global.browserActionFor(ext);
          if (ba) {
            // ⚠ Lo que se mira ANTES de apretar, porque `triggerAction` no devuelve nada y un
            // click que no llega se ve igual que una bóveda que no contesta:
            // `triggerClickOrPopup` sale por la puerta de atrás —devuelve null y no dispara— si
            // la acción NO está mostrada para esa pestaña (`isShownForTab`). Y la INSIGNIA dice
            // si `vault.match` funcionó desde adentro del navegador, que es la otra mitad.
            const t = win.gBrowser.selectedTab;
            try { di("MOSTRADA " + ba.action.isShownForTab(t)); } catch (e) { di("MOSTRADA ? " + e); }
            try { di("INSIGNIA " + JSON.stringify(ba.action.getProperty(t, "badgeText"))); }
            catch (e) { di("INSIGNIA ? " + e); }
            ba.triggerAction(win);
            disparo = "api";
          }
        } catch (e) { di("API " + e); }
        if (disparo === "ninguno") {
          const b = win.document.getElementById("boveda_atuq_tawasuyu-browser-action");
          if (b) { b.click(); disparo = "dom"; }
        }
        di("DISPARO " + disparo);
        win.setTimeout(function () { di("FIN"); }, 8000);
      } catch (e) {
        di("EXCEPCION " + e);
      }
    }, ESPERA_MS);
  }, "browser-delayed-startup-finished");
} catch (e) {
  try { dump("BOVEDA-METAL EXCEPCION-EXTERIOR " + e + "\\n"); } catch (e2) {}
}
"""


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 al usarse: el corpus es compartido y otro frente puede re-hashear una dep
    dejando lo de acá sin sellar durante las horas de su build. Un resultado obtenido con la
    escotilla NO se cita como «el de hoy pasa».
    """
    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:\n"
              f"    flock -o work/.farm-build.lock {HAMMER} --store {STORE} build recipes/{receta}.toml")
    return d


class Servidor(http.server.SimpleHTTPRequestHandler):
    """La página de login, servida de verdad. `atuq` la carga por la red como cualquier otra."""

    def log_message(self, *a):
        pass

    def do_GET(self):
        datos = PAGINA.encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(datos)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(datos)

    #: lo que la página delató que le pusieron en el campo, o None si nadie le puso nada.
    relleno = None

    def do_POST(self):
        n = int(self.headers.get("Content-Length") or 0)
        cuerpo = self.rfile.read(n)
        if self.path.startswith("/relleno"):
            Servidor.relleno = cuerpo.decode(errors="replace")
        datos = b"<!doctype html><html><body>adentro</body></html>"
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(datos)))
        self.end_headers()
        self.wfile.write(datos)


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


# ── El cliente del cable, en Python ───────────────────────────────────────────────────────────
# Marco de 4 bytes little-endian + JSON, que es lo que habla native messaging y lo que
# `shared/foreign-webext` implementa del otro lado. Se escribe acá —y no se llama a la extensión—
# porque las etapas A/B/C miden la CADENA host↔dueño↔diálogo, donde el navegador no participa:
# meterlo sólo agregaría diez minutos y un sospechoso más a cada fallo.
CLIENTE = r'''
import json, os, struct, subprocess, sys
def hablar(binario, estado, mensajes):
    p = subprocess.Popen([binario, "--state", estado],
                         stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    entrada = b""
    for m in mensajes:
        b = json.dumps(m).encode()
        entrada += struct.pack("<I", len(b)) + b
    fuera, err = p.communicate(entrada, timeout=180)
    salida, i = [], 0
    while i + 4 <= len(fuera):
        n = struct.unpack("<I", fuera[i:i+4])[0]; i += 4
        salida.append(json.loads(fuera[i:i+n])); i += n
    return salida, err.decode(errors="replace")
'''



# ── El contestador: quien dice que sí (o que no) donde diría una persona ──────────────────────
# ⚠ Espera a que la ventana esté en el ÁRBOL DEL COMPOSITOR y con el foco, **no** a que el proceso
# exista. Que `shuma-pregunta` esté corriendo no dice que su superficie esté mapeada ni enfocada, y
# una tecla mandada antes no la recibe nadie: el diálogo se queda abierto y el resultado se lee como
# «el consentimiento no contesta», que es la conclusión contraria a la verdadera. El compositor es
# el único que sabe quién tiene el foco, así que se le pregunta a él.
CONTESTADOR = r"""
import json, re, subprocess, sys, time

TECLA = sys.argv[1]
ESPERA = int(sys.argv[2]) if len(sys.argv) > 2 else 120
# ⚠ El patrón NO es opcional cuando hay más de una ventana, y la razón es el único modo de fallo
# peligroso de este arnés: con el navegador abierto, contestar «la ventana enfocada» le teclearía el
# «sí» A LA PÁGINA, y el diálogo se quedaría esperando. El resultado sería «el usuario no contestó»
# —o peor, un sí que nadie dio— y las dos cosas se leen como un producto que no anda.
PATRON = re.compile(sys.argv[3], re.I) if len(sys.argv) > 3 else None


def arbol():
    r = subprocess.run(["/usr/bin/swaymsg", "-t", "get_tree", "-r"],
                       capture_output=True, text=True)
    if r.returncode != 0 or not r.stdout.strip():
        return None
    try:
        return json.loads(r.stdout)
    except Exception:
        return None


def ventanas(n, acc=None):
    acc = [] if acc is None else acc
    if n.get("pid") and n.get("type") in ("con", "floating_con"):
        acc.append(n)
    for hijo in (n.get("nodes") or []) + (n.get("floating_nodes") or []):
        ventanas(hijo, acc)
    return acc


def pega(v):
    if PATRON is None:
        return True
    return bool(PATRON.search((v.get("app_id") or "") + " " + (v.get("name") or "")))


t0 = time.time()
visto = None
while time.time() - t0 < ESPERA:
    a = arbol()
    if a:
        vs = ventanas(a)
        if vs:
            visto = vs
            print("VENTANAS " + json.dumps(
                [{"app_id": v.get("app_id"), "name": v.get("name"),
                  "focused": bool(v.get("focused")),
                  "rect": v.get("rect")} for v in vs], ensure_ascii=False))
            candidatas = [v for v in vs if pega(v)]
            if candidatas:
                objetivo = candidatas[0]
                if not objetivo.get("focused"):
                    # Se le PIDE el foco al compositor en vez de esperar a que lo tenga: con el
                    # navegador abierto el diálogo puede salir detrás, y una tecla al aire no
                    # se distingue de un diálogo que no contesta.
                    con = objetivo.get("app_id") or objetivo.get("name") or ""
                    llave = "app_id" if objetivo.get("app_id") else "title"
                    subprocess.run(["/usr/bin/swaymsg", '[%s="%s"] focus' % (llave, con)],
                                   capture_output=True, text=True)
                    print("FOCO pedido para %s=%s" % (llave, con))
                    time.sleep(1)
                # ⚠ SE INSISTE hasta que la ventana se vaya, y no se teclea una sola vez.
                # Medido: con el mismo diálogo y el mismo `wtype rc=0`, una corrida contestaba y
                # otra volvía `denied`. La ventana aparece en el árbol del compositor ANTES de que
                # llimphi tenga su teclado enganchado —bajo llvmpipe el primer frame tarda—, así
                # que la primera tecla puede caer en el vacío. Un `sleep` más largo sólo mueve el
                # borde; lo que cierra el caso es mirar si la ventana SIGUE ahí. Teclear de más es
                # inofensivo: si el diálogo ya cerró, el dígito va a una página web.
                for intento in range(12):
                    time.sleep(2)
                    r = subprocess.run(["/usr/bin/wtype", TECLA], capture_output=True, text=True)
                    print("WTYPE %s intento=%s rc=%s %s"
                          % (TECLA, intento + 1, r.returncode, r.stderr.strip()))
                    time.sleep(1)
                    a2 = arbol()
                    if a2 is None or not [v for v in ventanas(a2) if pega(v)]:
                        print("CERRO tras %s intento(s)" % (intento + 1))
                        raise SystemExit(0)
                print("SIGUE ABIERTO tras 12 intentos: la tecla no lo cierra")
                raise SystemExit(1)
    time.sleep(1)
print("SIN VENTANA tras %ss; lo último que vi: %s" % (ESPERA, json.dumps(visto, ensure_ascii=False)[:2000]))
raise SystemExit(1)
"""


def guion_dentro(cuerpo, con_boveda, log="pasos.log", salida_rel="/salida"):
    """El guión que corre DENTRO de la jaula, con sway headless levantado.

    El compositor se levanta siempre —aunque la etapa no lo necesite— porque el diálogo es un
    cliente wayland y `boveda` también: sin compositor no se distingue «la app no abrió la bóveda»
    de «la app no encontró dónde pintar», que son dos causas muy distintas con el mismo síntoma.
    """
    arranque_boveda = ""
    if con_boveda:
        arranque_boveda = f"""
  # El DUEÑO de la bóveda. Se lanza con la raíz de prueba puesta, así que abre
  # `…/tawasuyu/boveda-de-prueba` y no la de nadie. Levanta el socket del navegador en un hilo.
  BOVEDA_TEST_ROOT={RAIZ_DE_PRUEBA} /usr/bin/boveda >> {salida_rel}/boveda.log 2>&1 &
  echo "@@ boveda lanzada pid $!" >> {salida_rel}/{log}
  # Se espera al SOCKET, no a un `sleep`: un sleep que alcanza hoy no alcanza en una máquina
  # cargada, y el fallo aparece como «la bóveda estaba cerrada» — la conclusión contraria.
  i=0
  while [ ! -S "$XDG_RUNTIME_DIR/boveda.sock" ] && [ $i -lt 60 ]; do sleep 1; i=$((i+1)); done
  [ -S "$XDG_RUNTIME_DIR/boveda.sock" ] \\
    && echo "@@ socket de la boveda LISTO en ${{i}}s" >> {salida_rel}/{log} \\
    || echo "@@ SIN socket de la boveda tras ${{i}}s" >> {salida_rel}/{log}
  # ⚠ Y se espera además a que la app PINTE, no sólo a que atienda. El socket lo levanta un hilo
  # antes de que haya ventana, así que seguir acá deja a la app inicializando su GPU al mismo
  # tiempo que el diálogo inicializa la suya — y con render por software en cuatro núcleos eso se
  # paga en el diálogo: aparece en el árbol del compositor, toma el foco y **no responde al
  # teclado**. Medido: doce teclas seguidas sin efecto y `vault.save` volviendo `denied` por
  # timeout, con la ventana todavía abierta.
  j=0
  while [ $j -lt 90 ]; do
    /usr/bin/swaymsg -t get_tree -r 2>/dev/null | grep -q '"app_id": *"boveda"' && break
    sleep 1; j=$((j+1))
  done
  echo "@@ ventana de la boveda a los ${{j}}s" >> {salida_rel}/{log}
"""
    return f"""#!/bin/sh
export XDG_RUNTIME_DIR=/run/user/0 HOME={salida_rel}/hogar LANG=C XDG_DATA_DIRS=/usr/share
export XDG_STATE_HOME={salida_rel}/hogar/.local/state
export WLR_BACKENDS=headless WLR_HEADLESS_OUTPUTS=1 WLR_RENDERER=pixman
# El cliente de GL va por software: ver la cabecera. `LIBGL_ALWAYS_SOFTWARE` hace que EGL elija
# `swrast` en vez de buscar un driver de la GPU que esta máquina no tiene, y `GALLIUM_DRIVER`
# elige CUÁL de los dos software.
#
# ⚠ **Y tiene que ser `llvmpipe`, no `softpipe`, y no es una preferencia.** Medido el 2026-09-18:
# con softpipe el diálogo no abre nunca y muere con `RequestDevice("Parent device is lost")`, que no
# nombra la causa. La causa está una línea más arriba en el log, y sólo aparece con
# `RUST_LOG=wgpu_core=info`:
#
#     Renderer: softpipe · Version: 3.3 (Core Profile)
#     indirect-validation error: ComputePipeline(Internal("The selected version doesn't support
#         Features(BUFFER_STORAGE | COMPUTE_SHADER | DYNAMIC_ARRAY_SIZE)"))
#
# O sea: **wgpu exige compute shaders** y softpipe se queda en GL 3.3. `llvmpipe` da GL 4.5 y pasa.
# Las dos son «mesa por software» y en el store son dos recetas hermanas (`mesa-swrast` es softpipe,
# `mesa-llvmpipe` es llvmpipe + `llvm18`), así que la diferencia se ve como un detalle de nombre y
# decide si el arnés existe o no.
export LIBGL_ALWAYS_SOFTWARE=1 LIBGL_DRIVERS_PATH=/usr/lib/dri GALLIUM_DRIVER=llvmpipe
mkdir -p "$XDG_RUNTIME_DIR" {salida_rel}/hogar "$XDG_STATE_HOME"; 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" >> {salida_rel}/{log}; exit 1; }}
  export WAYLAND_DISPLAY=wayland-1
  export SWAYSOCK=$(ls /run/user/0/sway-ipc.* 2>/dev/null | head -1)
  echo "@@ wayland listo" >> {salida_rel}/{log}
{arranque_boveda}
{cuerpo}
  sleep 1
  /usr/bin/swaymsg exit 2>/dev/null
) &
exec /usr/bin/sway -d > {salida_rel}/sway.log 2>&1
"""


def correr_jaula(capas, salida, guion, segundos=420):
    """Una corrida de bwrap con las capas dadas. La ÚLTIMA capa gana."""
    ruta = os.path.join(salida, "dentro.sh")
    with open(ruta, "w") as fh:
        fh.write(guion)
    orden = ["bwrap"]
    for c in capas:
        orden += ["--overlay-src", c]
    orden += [
        "--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=segundos)
        return True
    except subprocess.TimeoutExpired:
        print(f"   ⚠ la jaula no se cerró sola; se la cortó a los {segundos} s")
        return False


def leer(salida, nombre):
    r = os.path.join(salida, nombre)
    return open(r, errors="replace").read() if os.path.exists(r) else ""


def main():
    ap = argparse.ArgumentParser(add_help=False)
    ap.add_argument("--hasta", default="F")
    ap.add_argument("--list", action="store_true")
    ap.add_argument("-h", "--help", action="store_true")
    args, _ = ap.parse_known_args()
    if args.help:
        print(__doc__)
        return
    if args.list:
        for clave, que in ETAPAS:
            print(f"  {clave}  {que}")
        return

    hasta = args.hasta.upper()
    if hasta not in [c for c, _ in ETAPAS]:
        fatal(f"--hasta {hasta}: las etapas son " + ", ".join(c for c, _ in ETAPAS))

    if not os.path.isdir(RFS):
        fatal(f"no está el rootfs de sway en {RFS} — hidratalo con "
              "`scripts/hydrate-profile.py escritorio-sway --into …`")

    # Los artefactos se exigen POR ETAPA y no todos de entrada. No es comodidad: pedir los seis para
    # correr la A haría que este guardián no se pudiera encender mientras se construye una de las
    # piezas que él mismo destrabó —y la A es justamente la que mide el estado SIN esas piezas—.
    necesita = {"A": [("puriy-costura", "puriy-costura", None)]}
    necesita["B"] = necesita["A"] + [("boveda", "boveda", None),
                                 ("llvm18", "llvm18", None),
                                 ("mesa-llvmpipe", "mesa-llvmpipe", None)]
    necesita["C"] = necesita["B"] + [("shuma-pregunta", "shuma-pregunta", None), ("wtype", "wtype", None)]
    necesita["D"] = necesita["C"] + [("atuq", "atuq", "ATUQ_DIR")]
    necesita["E"] = necesita["D"]
    necesita["F"] = necesita["D"]

    dirs = {}
    for receta, nombre, escotilla in necesita[hasta]:
        dirs[nombre] = artefacto(receta, nombre, escotilla)
        print(f"artefacto: {os.path.basename(dirs[nombre])}")

    # El orden de las capas ES la semántica: **la última gana**. El mesa por software va al final
    # para pisar el `iris`-only del rootfs; si fuera antes, el rootfs le devolvería el driver de una
    # GPU Intel que esta máquina no tiene. `llvm18` va antes porque `swrast_dri.so` lo NECESITA:
    # sin él, el loader dice «failed to open swrast: libLLVM.so.18.1» y wgpu reporta «NoAdapter»,
    # que es el mismo síntoma que no tener driver.
    capas_base = [ALPINE, RFS]
    for n in ("atuq", "puriy-costura", "boveda", "shuma-pregunta", "wtype", "llvm18", "mesa-llvmpipe"):
        if n in dirs:
            capas_base.append(dirs[n])

    # El servidor de la página de login. Se levanta aunque la etapa no lo use: su PUERTO es lo que
    # define la URL con la que se guarda la credencial, y esa URL tiene que ser la MISMA en la
    # etapa que guarda y en la que el navegador visita — si no, `vault.match` no encontraría nada
    # y el fallo hablaría de la bóveda cuando el que cambió fue el puerto.
    puerto = puerto_libre()
    servidor = http.server.ThreadingHTTPServer(("0.0.0.0", puerto), Servidor)
    threading.Thread(target=servidor.serve_forever, daemon=True).start()
    url = f"http://127.0.0.1:{puerto}/"
    print(f"la página de login se sirve en {url}")

    tmp = tempfile.mkdtemp(prefix="atuq-boveda-metal-")
    salida = os.path.join(tmp, "salida")
    os.makedirs(salida)
    fallos = []
    try:
        correr_etapas(capas_base, dirs, salida, hasta, url, fallos)
    finally:
        guardado = os.path.join(ROOT, "work", "atuq-boveda-metal")
        os.makedirs(guardado, exist_ok=True)
        for n in os.listdir(salida):
            o = os.path.join(salida, n)
            if os.path.isfile(o):
                shutil.copy(o, os.path.join(guardado, n))
        print(f"\nlogs en {guardado}")
        shutil.rmtree(tmp, ignore_errors=True)

    if fallos:
        for f in fallos:
            print(f"✗ {f}", file=sys.stderr)
        raise SystemExit(1)
    print("\n✓ todo lo que se corrió, en verde")


def respuesta(salida, fichero, verbo):
    """La respuesta a un verbo, leída del JSON que dejó el preguntador dentro de la jaula."""
    crudo = leer(salida, fichero)
    if not crudo.strip():
        return None
    for r in json.loads(crudo):
        if r.get("verb") == verbo:
            return r
    return None


def correr_etapas(capas, dirs, salida, hasta, url, fallos):
    """Las etapas, en orden y cada una en su propia jaula.

    Cada etapa levanta su propio compositor. Es más lento que reusar uno y es a propósito: el
    estado que importa —la bóveda de prueba, el registro del host— vive en `/salida`, que se
    comparte, y así una etapa no hereda procesos de la anterior. Dos corridas sobre el mismo
    log se MEZCLAN y el resultado parece un pase; eso ya pasó en este frente (§6.1.bis).
    """
    orden = [c for c, _ in ETAPAS]
    tope = orden.index(hasta)
    escribir_preguntador(salida, url)
    with open(os.path.join(salida, "contestar.py"), "w") as fh:
        fh.write(CONTESTADOR)

    # ── A: SIN la app. Es el estado en que estuvo el corpus entero hasta el 2026-09-18. ────────
    print("\n== A · sin la app `boveda`, ¿qué contesta el host?")
    cuerpo = "  python3 /salida/preguntar.py sin-boveda >> /salida/pasos-A.log 2>&1\n"
    correr_jaula(capas, salida, guion_dentro(cuerpo, con_boveda=False, log="pasos-A.log"), segundos=240)
    print(indentar(leer(salida, "pasos-A.log")))
    estado = respuesta(salida, "sin-boveda.json", "vault.status")
    if estado is None:
        fallos.append("A: el preguntador no dejó respuesta — mirá pasos-A.log")
    elif estado.get("locked") is not True:
        fallos.append(f"A: sin la app, vault.status tenía que decir locked:true y dijo {estado}")
    else:
        print("   ✓ A: sin dueño, la bóveda contesta CERRADA (y con ok:true, que es la trampa)")
    if tope < orden.index("B"):
        return

    # ── B: CON la app. El dueño abre la base y atiende el socket. ──────────────────────────────
    print("\n== B · con la app `boveda` corriendo, ¿la bóveda está abierta?")
    cuerpo = "  python3 /salida/preguntar.py con-boveda >> /salida/pasos-B.log 2>&1\n"
    correr_jaula(capas, salida, guion_dentro(cuerpo, con_boveda=True, log="pasos-B.log"), segundos=300)
    print(indentar(leer(salida, "pasos-B.log")))
    estado = respuesta(salida, "con-boveda.json", "vault.status")
    if estado is None:
        fallos.append("B: el preguntador no dejó respuesta — mirá pasos-B.log y boveda.log")
    elif estado.get("locked") is not False:
        fallos.append(f"B: con la app corriendo, vault.status tenía que decir locked:false y dijo "
                      f"{estado}.\n"
                      "     Mirá `boveda.log`. Si dice «index out of bounds: the len is 0 but the index "
                      "is 0» en `llimphi-hal/src/lib.rs`, NO es de la bóveda: es el muro del §7.septies "
                      "—llimphi construye la instancia de wgpu SIN display handle, wgpu cae a la "
                      "plataforma EGL surfaceless, esa plataforma no tiene configs de ventana, y la "
                      "lista de formatos de la surface sale vacía—. Se destraba en llimphi, no acá, y "
                      "hasta entonces la app no abre ventana ni levanta el socket del navegador.")
    else:
        print("   ✓ B: el dueño atiende — la bóveda de prueba está ABIERTA")
    if tope < orden.index("C"):
        return

    # ── C: el instrumento. Antes de pedirle nada a la bóveda, ¿se puede contestar un diálogo? ──
    # Va ANTES de D a propósito: si el diálogo no se puede contestar, `vault.save` va a decir
    # `denied` — que es la respuesta CORRECTA a un consentimiento que no llegó— y sin esta etapa
    # eso se leería como «la bóveda no guarda». Separar el instrumento del producto es lo que
    # hace que el fallo diga cuál de los dos se rompió.
    print("\n== C · ¿abre el diálogo en el compositor, y se lo puede contestar?")
    cuerpo = """
  for par in si:%s:1 no:%s:2; do
    quien=${par%%%%:*}; resto=${par#*:}; tecla=${resto%%%%:*}; n=${resto#*:}
    ( python3 /salida/contestar.py "$tecla" 120 "pregunta|shuma" > /salida/contestador-C-$quien.log 2>&1 ) &
    contestador=$!
    /usr/bin/shuma-pregunta --title "Prueba del instrumento" --detail "esto no toca la boveda" \
        --option "yes=Si, contesta" --option "no=No" \
        > /salida/dialogo-C-$quien.json 2> /salida/dialogo-C-$quien.err
    echo "@@ dialogo $quien salio con $? y dijo: $(cat /salida/dialogo-C-$quien.json)" >> /salida/pasos-C.log
    wait "$contestador"
  done
""" % (TECLA_SI, TECLA_NO)
    correr_jaula(capas, salida, guion_dentro(cuerpo, con_boveda=False, log="pasos-C.log"), segundos=420)
    print(indentar(leer(salida, "pasos-C.log")))
    for quien in ("si", "no"):
        print(indentar(leer(salida, f"contestador-C-{quien}.log")))
    dijo_si = leer(salida, "dialogo-C-si.json").strip()
    dijo_no = leer(salida, "dialogo-C-no.json").strip()
    if '"choice":"yes"' not in dijo_si.replace(" ", ""):
        fallos.append(f"C: al teclear «{TECLA_SI}» el diálogo tenía que contestar yes y dijo {dijo_si!r} "
                      "— si no hay VENTANAS en el log del contestador, el cliente de GL no pintó "
                      "(mirá dialogo-C-si.err: es donde llimphi dice por qué)")
    elif '"choice":"no"' not in dijo_no.replace(" ", ""):
        # El control: sin él, un diálogo que contestara «yes» pase lo que pase se vería igual.
        fallos.append(f"C: el control falló — al teclear «{TECLA_NO}» tenía que contestar no y dijo {dijo_no!r}")
    else:
        print("   ✓ C: el diálogo abre, la tecla llega, y contesta lo que se le tecleó (sí Y no)")
    if tope < orden.index("D"):
        return

    # ── D: la bóveda guarda, y sólo con permiso. ───────────────────────────────────────────────
    print("\n== D · vault.save con el consentimiento dado, ¿guarda? ¿y después la encuentra?")
    cuerpo = """
  ( python3 /salida/contestar.py "%s" 150 "pregunta|shuma" > /salida/contestador-D.log 2>&1 ) &
  contestador=$!
  python3 /salida/preguntar.py guardar >> /salida/pasos-D.log 2>&1
  # ⚠ `wait "$contestador"` y NO `wait` a secas: la app de la bóveda TAMBIÉN es un trabajo en
  # segundo plano, así que un `wait` pelado no vuelve nunca, la jaula se come su timeout y a la
  # app la mata un KILL. Y lo que se pierde ahí es lo que se acaba de guardar: `sled` no alcanza a
  # volcar, la etapa siguiente no encuentra la credencial, y el cuadro que queda es «la bóveda no
  # guardó» — el diagnóstico contrario al verdadero.
  wait "$contestador"
""" % TECLA_SI
    correr_jaula(capas, salida, guion_dentro(cuerpo, con_boveda=True, log="pasos-D.log"), segundos=420)
    print(indentar(leer(salida, "pasos-D.log")))
    print(indentar(leer(salida, "contestador-D.log")))
    guardo = respuesta(salida, "guardar.json", "vault.save")
    encontro = respuesta(salida, "guardar.json", "vault.match")
    if guardo is None:
        fallos.append("D: el preguntador no dejó respuesta de vault.save — mirá pasos-D.log")
    elif guardo.get("denied") is True:
        fallos.append("D: vault.save salió DENEGADO. Si la etapa C está en verde, el diálogo se puede "
                      "contestar y esto es del producto; si C falló, es el instrumento y no la bóveda")
    elif not guardo.get("item"):
        fallos.append(f"D: vault.save no devolvió el identificador de lo guardado: {guardo}")
    elif not (encontro or {}).get("items"):
        fallos.append(f"D: guardó pero vault.match no la encuentra para {url}: {encontro}")
    else:
        item = encontro["items"][0]
        if item.get("username") != USUARIO:
            fallos.append(f"D: la credencial encontrada no es la guardada: {item}")
        else:
            print(f"   ✓ D: guardada con permiso y encontrada de nuevo ({item.get('title')!r}, "
                  f"usuario {item.get('username')!r})")
            # Y la mitad que importa: `vault.match` NO puede traer la contraseña. El tipo del
            # otro lado no tiene dónde ponerla, y esto lo comprueba desde afuera.
            if any(k in item for k in ("password", "clave")):
                fallos.append(f"D: ⚠ vault.match trajo una CONTRASEÑA, y no debe: {item}")

    if tope < orden.index("E"):
        return

    # ── E y F: el navegador de verdad, y el par que hace que E signifique algo. ────────────────
    # La E sola no prueba nada: una bóveda que entrega SIEMPRE se ve idéntica a una que entrega con
    # permiso. Lo único que las separa es contestar que NO y ver que no llega nada.
    for etapa, tecla, espera_relleno in (("E", TECLA_SI, True), ("F", TECLA_NO, False)):
        if tope < orden.index(etapa):
            return
        que = "SÍ" if espera_relleno else "NO"
        print(f"\n== {etapa} · el navegador pide la contraseña y se contesta que {que}")
        Servidor.relleno = None
        correr_navegador(capas, dirs, salida, url, tecla, etapa)
        print(indentar(leer(salida, f"pasos-{etapa}.log")))
        print(indentar(leer(salida, f"contestador-{etapa}.log")))
        print(indentar(leer(salida, "chrome.txt")))
        lineas = [l for l in leer(salida, f"atuq-{etapa}.log").splitlines() if l.startswith("BOVEDA ")]
        for l in lineas[:12]:
            print("   " + l)
        llego = Servidor.relleno
        if etapa == "E":
            if "CONECTADO" not in " ".join(lineas):
                fallos.append("E: la extensión de la bóveda no llegó a conectar con el host "
                              "(no hay línea «BOVEDA CONECTADO») — mirá atuq-E.log")
            if llego is None:
                fallos.append("E: la contraseña NUNCA llegó al campo. Si el contestador no vio la "
                              "ventana del diálogo, el muro es el del §7.septies; si la vio y tecleó, "
                              "mirá si el DISPARO del chrome fue «ninguno» (nadie apretó el botón)")
            elif CLAVE not in llego:
                fallos.append(f"E: al campo llegó algo que no es la contraseña guardada: {llego!r}")
            else:
                print("   ✓ E: la contraseña llegó al formulario, y sólo después del sí")
        else:
            if llego is not None:
                fallos.append(f"F: ⚠⚠ contestando que NO igual llegó algo al campo: {llego!r}. "
                              "Es la falla que este guardián existe para atrapar: la bóveda entrega "
                              "sin permiso")
            else:
                print("   ✓ F: con el no, al campo no llegó nada (y es el control que sostiene la E)")


def correr_navegador(capas, dirs, salida, url, tecla, etapa):
    """Una corrida con `atuq` de verdad: la página servida, el botón de la extensión, el diálogo.

    La sonda del chrome se apenda al `atuq.cfg` del artefacto en una capa propia — se mide el
    fichero real con la sonda encima, no una copia.
    """
    capa = os.path.join(salida, f"capa-{etapa}")
    app = os.path.join(capa, APPDIR)
    os.makedirs(app, exist_ok=True)
    original = open(os.path.join(dirs["atuq"], APPDIR, "atuq.cfg"), encoding="utf8").read()
    with open(os.path.join(app, "atuq.cfg"), "w", encoding="utf8") as fh:
        fh.write(original)
        # La espera antes de apretar: la extensión tiene que haber conectado con el host y haber
        # mirado la pestaña. Apretar antes es preguntarle a una bóveda que todavía no se enteró.
        fh.write(SONDA_CHROME.replace("ESPERA_MS", "9000"))
    perfil = os.path.join(salida, f"perfil-{etapa}")
    os.makedirs(perfil, exist_ok=True)
    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')
    cuerpo = """
  # El patrón del contestador NO es opcional acá: con el navegador abierto, teclear «la ventana
  # enfocada» le contestaría a la PÁGINA. Se busca la ventana del diálogo por app_id/título.
  ( python3 /salida/contestar.py "%s" 150 "pregunta|shuma" > /salida/contestador-%s.log 2>&1 ) &
  contestador=$!
  export MOZ_ENABLE_WAYLAND=1 GDK_BACKEND=wayland
  # 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
  timeout 150 /usr/bin/atuq --no-remote --profile /salida/perfil-%s '%s' >> /salida/atuq-%s.log 2>&1
  echo "@@ atuq salio con $?" >> /salida/pasos-%s.log
  wait "$contestador"
""" % (tecla, etapa, etapa, url, etapa, etapa)
    correr_jaula(capas + [capa], salida, guion_dentro(cuerpo, con_boveda=True, log=f"pasos-{etapa}.log"),
                 segundos=420)


def indentar(t):
    return "\n".join("   " + l for l in t.splitlines() if l.strip())


def escribir_preguntador(salida, url):
    """El cliente del cable que corre DENTRO de la jaula.

    Habla con el host EXACTAMENTE como lo hace la extensión —marco de 4 bytes + JSON— y no con la
    bóveda directamente: lo que este guardián tiene que medir es la cadena entera, y el host es el
    único que sabe encontrar al dueño.
    """
    with open(os.path.join(salida, "preguntar.py"), "w") as fh:
        fh.write(CLIENTE)
        fh.write(f'''
QUE = sys.argv[1]
URL = {url!r}
BIN = "/usr/bin/puriy-costura"
ESTADO = "/salida/hogar/estado-costura"
os.makedirs(ESTADO, exist_ok=True)
# `ping` va SIEMPRE y es el control de la corrida: separa «el host no contestó» de «el host
# contestó que la bóveda está cerrada», que es la distinción que costó tres semanas.
mensajes = [
    {{"id": 1, "verb": "ping"}},
    {{"id": 2, "verb": "vault.status"}},
]
if QUE == "guardar":
    mensajes.append({{"id": 3, "verb": "vault.save", "url": URL,
                      "username": {USUARIO!r}, "password": {CLAVE!r}, "title": {TITULO!r}}})
if QUE in ("match", "guardar"):
    mensajes.append({{"id": 4, "verb": "vault.match", "url": URL}})
respuestas, err = hablar(BIN, ESTADO, mensajes)
open("/salida/%s.json" % QUE, "w").write(json.dumps(respuestas, ensure_ascii=False, indent=1))
open("/salida/%s.err" % QUE, "w").write(err)
for r in respuestas:
    print("   <-", json.dumps(r, ensure_ascii=False))
''')


if __name__ == "__main__":
    main()