scripts/test-guest-console-vm.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Boot an isolated, network-free Firecracker guest and exercise its real ttyS0."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import pty
import select
import shutil
import socket
import struct
import subprocess
import tempfile
import time
import tty
def main():
parser = argparse.ArgumentParser(description=__doc__)
for name in ("kernel", "helper", "busybox"):
parser.add_argument(f"--{name}", required=True, type=Path)
parser.add_argument("--firecracker", default="firecracker")
parser.add_argument("--expected-release", required=True)
parser.add_argument("--report", type=Path, required=True)
args = parser.parse_args()
args.report.parent.mkdir(parents=True, exist_ok=True)
transcript = bytearray()
results = []
with tempfile.TemporaryDirectory(prefix="fcr471-", dir="/tmp") as temporary:
root = Path(temporary)
tree = root / "tree"
for directory in ("bin", "dev", "proc", "sys", "etc", "tmp", "root"):
(tree / directory).mkdir(parents=True)
shutil.copyfile(args.busybox, tree / "bin/busybox")
shutil.copyfile(args.helper, tree / "resize")
(tree / "bin/busybox").chmod(0o755)
(tree / "resize").chmod(0o755)
(tree / "init").write_text("""#!/bin/busybox sh
export PATH=/bin TERM=xterm PS1='FCR_PROMPT> '
/bin/busybox mount -t proc proc /proc
/bin/busybox mount -t sysfs sysfs /sys
/bin/busybox mount -t devtmpfs devtmpfs /dev
/resize >/dev/null 2>&1 &
exec /bin/busybox init
""")
(tree / "init").chmod(0o755)
(tree / "etc/inittab").write_text("::respawn:-/bin/busybox sh\n")
disk = root / "rootfs.ext4"
subprocess.run(["mkfs.ext4", "-q", "-F", "-d", str(tree), str(disk), "32M"], check=True)
vsock = root / "v.sock"
config = {
"boot-source": {"kernel_image_path": str(args.kernel.resolve()), "boot_args": "console=ttyS0 reboot=k panic=1 pci=off root=/dev/vda rw init=/init"},
"drives": [{"drive_id": "rootfs", "path_on_host": str(disk), "is_root_device": True, "is_read_only": False}],
"machine-config": {"vcpu_count": 1, "mem_size_mib": 256},
"vsock": {"guest_cid": 3, "uds_path": str(vsock)},
}
config_path = root / "fc.json"
config_path.write_text(json.dumps(config))
master, slave = pty.openpty()
tty.setraw(slave)
with (root / "firecracker.log").open("wb") as log:
process = subprocess.Popen([args.firecracker, "--no-api", "--config-file", str(config_path)], stdin=slave, stdout=slave, stderr=log)
os.close(slave)
def wait_for(marker, start=0, timeout=30):
deadline = time.monotonic() + timeout
while marker not in transcript[start:] and time.monotonic() < deadline:
if process.poll() is not None:
raise RuntimeError(f"Firecracker exited {process.returncode}: {(root / 'firecracker.log').read_text()}")
if select.select([master], [], [], 0.1)[0]:
transcript.extend(os.read(master, 65536))
if marker not in transcript[start:]:
raise TimeoutError(f"Guest did not produce {marker!r}; serial tail: {bytes(transcript[-2000:])!r}")
def command(line, marker):
start = len(transcript)
os.write(master, line.encode() + b"\n")
wait_for(marker, start)
def resize(rows, cols):
deadline = time.monotonic() + 15
while True:
try:
with socket.socket(socket.AF_UNIX) as stream:
stream.settimeout(2)
stream.connect(str(vsock))
stream.sendall(b"CONNECT 47100\n")
greeting = bytearray()
while not greeting.endswith(b"\n") and len(greeting) < 64:
byte = stream.recv(1)
if not byte:
raise ConnectionError("Guest listener is not ready")
greeting.extend(byte)
if not greeting.startswith(b"OK "):
raise ConnectionError(f"Bad vsock handshake: {greeting!r}")
stream.sendall(b"FCR1" + struct.pack(">HH", rows, cols))
reply = bytearray()
while len(reply) < 8:
part = stream.recv(8 - len(reply))
if not part:
raise ConnectionError("Missing acknowledgement")
reply.extend(part)
assert reply == b"FCA1" + struct.pack(">HH", rows, cols), reply
return
except (OSError, ConnectionError):
if time.monotonic() >= deadline:
raise
time.sleep(0.1)
try:
wait_for(b"FCR_PROMPT>")
command("printf 'FCR_KERNEL_%s_END\\n' \"$(/bin/busybox uname -r)\"", b"FCR_KERNEL_" + args.expected_release.encode() + b"_END")
resize(30, 100)
command("printf 'FCR_SIZE_'; /bin/busybox stty size", b"FCR_SIZE_30 100")
results.append("ttyS0 is 30 rows by 100 columns after acknowledged resize")
# Reap a short-lived child regularly so ash dispatches pending
# traps without needing terminal input to release a built-in read.
command("/bin/busybox sh -c 'trap \"printf FCR_WINCH_; /bin/busybox stty size; exit\" WINCH; printf \"FCR_TRAP_%s\\n\" READY; while :; do /bin/busybox sleep 1; done'", b"FCR_TRAP_READY")
start = len(transcript)
resize(40, 120)
wait_for(b"FCR_WINCH_40 120", start)
wait_for(b"FCR_PROMPT>", start)
command("printf 'FCR_SIZE_'; /bin/busybox stty size", b"FCR_SIZE_40 120")
results.append("foreground process receives SIGWINCH and observes 40 rows by 120 columns")
command("printf 'FCR_VI_%s\\n' READY > /tmp/size; /bin/busybox vi /tmp/size", b"FCR_VI_READY")
start = len(transcript)
resize(24, 80)
wait_for(b"\x1b[24;1H\x1b[K", start)
start = len(transcript)
os.write(master, b":q!\r")
wait_for(b"FCR_PROMPT>", start)
command("printf 'FCR_AFTER_VI_'; /bin/busybox stty size", b"FCR_AFTER_VI_24 80")
results.append("vi remains usable across resize and exits back to the shell")
command("/bin/busybox top -d 1", b"PID")
start = len(transcript)
resize(35, 110)
wait_for(b"PID", start)
start = len(transcript)
os.write(master, b"q")
wait_for(b"FCR_PROMPT>", start)
command("printf 'FCR_AFTER_TOP_'; /bin/busybox stty size", b"FCR_AFTER_TOP_35 110")
results.append("top remains usable across resize and exits back to the shell")
finally:
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
os.close(master)
args.report.with_suffix(".console.log").write_bytes(transcript)
args.report.write_text(json.dumps({"kernel": str(args.kernel), "kernelSha256": hashlib.sha256(args.kernel.read_bytes()).hexdigest(), "expectedRelease": args.expected_release, "checks": results, "passed": len(results) == 4}, indent=2) + "\n")
for result in results:
print("PASS", result)
if __name__ == "__main__":
main()