#!/usr/bin/env python3
"""¿`atuq` congela lo que leés, y lo encuentra después? (SDD 26 §6.3)
python3 scripts/test-atuq-archivo.py # tres visitas y una búsqueda
python3 scripts/test-atuq-archivo.py --negative-control # en VENTANA PRIVADA no se archiva nada
── QUÉ SE EJERCITA ───────────────────────────────────────────────────────────────────────────
Un servidor HTTP sirve dos páginas distintas; el navegador las visita; el guión de contenido manda
al fondo el HTML **ya ejecutado** (lo que la persona vio, no lo que el servidor mandó) y el texto
visible; el host congela el HTML en el CAS y lo deja en un índice buscable.
página → guión de contenido → connectNative → puriy-costura → CAS (BLAKE3) + índice
visita 1 (página A) ⇒ archivada, visitas=1, total=1
visita 2 (página A) ⇒ **dedup**, visitas=2, total=1 ← volver no duplica
visita 3 (página B) ⇒ total=2
búsqueda «masa» ⇒ SÓLO la A ← y la mitad que importa es la que DESCARTA
La búsqueda se le pregunta al binario del host DESDE AFUERA de la jaula: es la prueba de que lo que
el navegador archivó quedó en disco y se puede consultar sin el navegador.
── EL CONTROL NEGATIVO ES LA PROMESA MÁS IMPORTANTE ──────────────────────────────────────────
`--negative-control` abre la misma página en una **ventana privada** y exige que NO se archive nada.
Archivar lo que alguien abrió en privado sería exactamente lo contrario de lo que pidió.
Y distingue POR QUÉ no se archivó, que no es lo mismo: si en el log aparece `PRIVADA no se archiva`,
lo impidió NUESTRO código; si no aparece nada, lo impidió el navegador (las extensiones no corren en
ventanas privadas salvo que se les permita). Las dos protegen, pero sólo una es nuestra, y decir cuál
es la diferencia entre saber y suponer.
REQUISITOS: el rootfs de `scripts/atuq-nested.sh`, y `atuq` + `puriy-costura` sellados.
"""
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/atuq-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"))
PAGINAS = {
"pan.html": ("Cómo hornear pan", "masa madre reposo y horno de leña"),
"cohete.html": ("Motores de cohete", "tobera de campana y empuje específico"),
}
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):
def log_message(self, *a):
pass
def do_GET(self):
nombre = self.path.lstrip("/").split("?")[0]
if nombre not in PAGINAS:
self.send_error(404)
return
titulo, cuerpo = PAGINAS[nombre]
datos = (f'{titulo}'
f"{titulo}
{cuerpo}
").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 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, url, privada=False):
modo = "--private-window" if privada else ""
guion = f"""
mkdir -p /root /salida/hogar
[ -s /etc/machine-id ] || tr -d - < /proc/sys/kernel/random/uuid > /etc/machine-id
export HOME=/salida/hogar
timeout 45 /usr/bin/atuq --profile /salida/perfil --no-remote --headless {modo} '{url}' \
> /salida/consola.log 2>&1
exit 0
"""
subprocess.run([
"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",
"--setenv", "XDG_RUNTIME_DIR", "/run",
"--setenv", "XDG_DATA_DIRS", "/usr/share", "--setenv", "MOZ_HEADLESS", "1",
"--setenv", "MOZ_DISABLE_CONTENT_SANDBOX", "1",
"--setenv", "MOZ_DISABLE_GMP_SANDBOX", "1",
"--setenv", "MOZ_DISABLE_RDD_SANDBOX", "1",
"--setenv", "MOZ_DISABLE_SOCKET_PROCESS_SANDBOX", "1",
"--setenv", "MOZ_DISABLE_UTILITY_SANDBOX", "1",
"sh", "-c", guion,
], check=False)
log = open(os.path.join(salida, "consola.log"), errors="replace").read()
return [l[len("ARCHIVO "):] for l in log.splitlines() if l.startswith("ARCHIVO ")], log
def preguntar(costura, estado, consulta):
"""Le pregunta al binario del host, desde AFUERA, qué encuentra. Es la prueba de que el archivo
quedó en disco y se consulta sin el navegador."""
binario = os.path.join(costura, "usr/bin/puriy-costura")
cuerpo = json.dumps({"verb": "archive.search", "query": consulta}).encode()
entrada = struct.pack("