#!/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 CINCO 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 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í.
── 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"
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"
PAGINA = """Banco de prueba
Banco de prueba
"""
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)
def do_POST(self):
n = int(self.headers.get("Content-Length") or 0)
self.rfile.read(n)
datos = b"adentro"
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(" 2 else 120
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
t0 = time.time()
visto = None
while time.time() - t0 < ESPERA:
a = arbol()
if a:
vs = ventanas(a)
if vs:
visto = vs
enfocada = [v for v in vs if v.get("focused")]
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))
if enfocada:
# Un respiro después del foco: el primer frame puede no estar pintado todavía y
# la tecla llegaría a una ventana que aún no instaló su manejador.
time.sleep(2)
r = subprocess.run(["/usr/bin/wtype", TECLA], capture_output=True, text=True)
print("WTYPE %s rc=%s %s" % (TECLA, r.returncode, r.stderr.strip()))
raise SystemExit(0)
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}
"""
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, 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, 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" > /salida/contestador-C-$quien.log 2>&1 ) &
/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
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" > /salida/contestador-D.log 2>&1 ) &
python3 /salida/preguntar.py guardar >> /salida/pasos-D.log 2>&1
wait
""" % 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}")
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()