sergioandClaude Opus 4.8
6886f810bd
netup = binario hammer-propio (sync, sólo libc, sin tokio) que configura la red: levanta el link (netlink RTM_NEWLINK), negocia un lease DHCPv4 a mano sobre UDP (DISCOVER/OFFER/REQUEST/ACK con flag broadcast), y aplica IP + ruta default + /etc/resolv.conf (RTM_NEWADDR/RTM_NEWROUTE). Hand-roll de netlink y DHCP porque no hay cliente DHCP Rust "maduro" para adoptar al estilo ripgrep, y el workspace es 100% sync (rtnetlink arrastraría tokio). Reemplaza el `ip` estático de busybox. Validado in-VM contra el DHCP de QEMU slirp: ✓ lease 10.0.2.15 + ruta + DNS + egress TCP. Autodetecta la NIC (primera no-loopback) y espera carrier por sysfs. Infra de prueba: - scripts/drive-netup.py: bootea la VM, corre netup y verifica lease/NAT/DNS. SLIRP=1 ⇒ red user-mode (cero setup de host); si no, tap del lab. - scripts/labnet.sh: lab de LAN real opcional (tap directo + dnsmasq + NAT) para fidelidad de hardware; no requerido para validar netup. - scripts/netdiag.sh: diagnóstico del camino DHCP del lab (tcpdump+nft+dnsmasq). Lección: para validar el CÓDIGO conviene slirp primero (cero plomería de host); la LAN real (firewall INPUT, sutilezas del bridge, perms) es fidelidad posterior. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
119 lines
4.6 KiB
Python
119 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
# drive-netup.py — bootea la VM con la NIC conectada al tap del lab (scripts/labnet.sh) y corre `netup`
|
|
# + chequeos de conectividad real (DHCP de dnsmasq, NAT, DNS). Driver no-interactivo: espera la shell
|
|
# de arje-zero, manda el test, captura hasta el marcador y emite veredicto.
|
|
#
|
|
# Prerequisito: sudo sh scripts/labnet.sh up (bridge hmbr0 + tap hmtap0 + dnsmasq + NAT)
|
|
#
|
|
# Variables: KERNEL (def store/*-linux/boot/bzImage) DISK (def work/hammer-disk.img)
|
|
# TAP (def hmtap0) NICDEV (def e1000) MEM (def 4096) KVM (1 si /dev/kvm)
|
|
# DEADLINE (def 180s) LOG (def work/netup-run.log)
|
|
import os, pty, select, subprocess, sys, time, re, glob
|
|
|
|
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
|
|
def absp(p):
|
|
return p if os.path.isabs(p) else os.path.join(ROOT, p)
|
|
|
|
|
|
KERNEL = os.environ.get("KERNEL") or (glob.glob(os.path.join(ROOT, "store/*-linux/boot/bzImage")) + [""])[0]
|
|
DISK = absp(os.environ.get("DISK", "work/hammer-disk.img"))
|
|
TAP = os.environ.get("TAP", "hmtap0")
|
|
NICDEV = os.environ.get("NICDEV", "e1000")
|
|
MEM = os.environ.get("MEM", "4096")
|
|
DEADLINE = int(os.environ.get("DEADLINE", "180"))
|
|
LOG = absp(os.environ.get("LOG", "work/netup-run.log"))
|
|
WANT_KVM = os.environ.get("KVM", "1") == "1"
|
|
|
|
for p, what in [(KERNEL, "kernel"), (DISK, "imagen de disco")]:
|
|
if not p or not os.path.exists(p):
|
|
sys.exit(f"no existe {what}: {p}")
|
|
|
|
accel = ["-cpu", "Broadwell"]
|
|
if WANT_KVM and os.access("/dev/kvm", os.W_OK):
|
|
accel = ["-enable-kvm", "-cpu", "host"]
|
|
|
|
# SLIRP=1 ⇒ red user-mode de QEMU (DHCP/DNS/NAT virtuales, cero setup de host). Si no, tap del lab.
|
|
if os.environ.get("SLIRP", "0") == "1":
|
|
netdev = ["-netdev", "user,id=n0"]
|
|
else:
|
|
netdev = ["-netdev", f"tap,id=n0,ifname={TAP},script=no,downscript=no"]
|
|
QEMU = (["qemu-system-x86_64", "-m", MEM, "-no-reboot", "-nographic"] + accel +
|
|
["-kernel", KERNEL,
|
|
"-drive", f"file={DISK},if=virtio,format=raw",
|
|
"-append", "console=ttyS0 root=/dev/vda1 rw rdinit=/sbin/init"] +
|
|
netdev + ["-device", f"{NICDEV},netdev=n0"])
|
|
|
|
# El test que corre dentro de la VM: netup (autodetecta la NIC) + conectividad NAT + DNS.
|
|
# Los marcadores se CONSTRUYEN en runtime (printf '%s') para que NO aparezcan literales en el eco del
|
|
# comando tecleado — si no, el driver los matchearía en el eco y saldría antes de que el test corra.
|
|
TEST = os.environ.get("TESTCMD") or (
|
|
"netup; "
|
|
"ping -c2 -W2 8.8.8.8 >/dev/null 2>&1 && printf 'PING_%s\\n' OK || printf 'PING_%s\\n' FAIL; "
|
|
"nslookup google.com >/dev/null 2>&1 && printf 'DNS_%s\\n' OK || printf 'DNS_%s\\n' FAIL; "
|
|
"printf 'TEST_%s\\n' DONE"
|
|
)
|
|
TEST = TEST + "\n"
|
|
|
|
print(f"==> kernel {KERNEL}\n==> disco {DISK}\n==> tap {TAP} ({NICDEV}) mem {MEM} kvm {'sí' if accel[0]=='-enable-kvm' else 'no'}")
|
|
ansi = re.compile(rb'\x1b\[[0-9;?]*[a-zA-Z]')
|
|
master, slave = pty.openpty()
|
|
p = subprocess.Popen(QEMU, stdin=slave, stdout=slave, stderr=slave, close_fds=True)
|
|
os.close(slave)
|
|
log = open(LOG, "wb")
|
|
buf = b""
|
|
sent = False
|
|
start = time.time()
|
|
|
|
while True:
|
|
if time.time() - start > DEADLINE:
|
|
log.write(b"\n[DRIVER] DEADLINE\n")
|
|
break
|
|
r, _, _ = select.select([master], [], [], 5)
|
|
if r:
|
|
try:
|
|
data = os.read(master, 8192)
|
|
except OSError:
|
|
break
|
|
if not data:
|
|
break
|
|
log.write(data)
|
|
log.flush()
|
|
buf = ansi.sub(b'', buf + data)[-20000:]
|
|
if not sent and (b"\n~ #" in buf or b"\r~ #" in buf):
|
|
time.sleep(2)
|
|
os.write(master, TEST.encode())
|
|
sent = True
|
|
log.write(b"\n[DRIVER] enviado test netup\n")
|
|
if sent and b"TEST_DONE" in buf:
|
|
time.sleep(1)
|
|
try:
|
|
log.write(os.read(master, 8192))
|
|
except Exception:
|
|
pass
|
|
break
|
|
elif p.poll() is not None:
|
|
break
|
|
|
|
for fn in (p.terminate, p.kill):
|
|
try:
|
|
fn(); time.sleep(1)
|
|
except Exception:
|
|
pass
|
|
log.close()
|
|
|
|
blob = ansi.sub(b'', open(LOG, "rb").read())
|
|
configured = b"red configurada" in blob # netup imprime esto sólo si todo salió bien
|
|
ping = b"PING_OK" in blob
|
|
dns = b"DNS_OK" in blob
|
|
lease = re.search(rb"netup: lease (\d+\.\d+\.\d+\.\d+)", blob)
|
|
print(f"\nDRIVER done sent={sent} elapsed={int(time.time()-start)}s")
|
|
print(f" netup ok : {configured}")
|
|
print(f" lease : {lease.group(1).decode() if lease else '—'}")
|
|
print(f" ping NAT : {ping}")
|
|
print(f" DNS : {dns}")
|
|
ok = configured and lease and ping and dns
|
|
print("RESULTADO:", "✓ RED OK (lease + NAT + DNS)" if ok else "✗ revisá el log: " + LOG)
|
|
sys.exit(0 if ok else 1)
|