M7: in-repo FAT32 image builder + boot the whole system off a USB stick
The system now boots off a real FAT32 filesystem on a USB mass-storage device instead of QEMU's synthesized VVFAT drive. A new in-repo image builder formats that filesystem from the FHS boot tree, and both QEMU call sites (the run step and the test harness) attach it as a usb-storage device on the xHCI bus, so every boot exercises the full USB path OVMF -> BOOTX64.efi -> kernel. - tools/make-fat-image.py: a Python 3 stdlib-only FAT32 formatter (mirrors tools/make-initial-ramdisk.py — no external host dependencies). It lays down the boot sector + BPB/EBPB32, FSInfo, backup boot sector, two FATs, and the root/subdir/file cluster chains, emitting long-name entries where a name is not 8.3. Packs the four boot inputs (EFI/BOOT/BOOTX64.efi, system/kernel, system/services/init, boot/initial-ramdisk.img) into their boot paths. A --verify subcommand re-checks the 0xAA55 signature, recomputes the cluster count -> FAT32, and resolves EFI/BOOT/BOOTX64.efi, all with no dependencies. - build.zig: a mk_fat step builds zig-out/danos-usb.img from the four boot artifacts (so changing -Dtest-case rebuilds the image with that kernel), a check-fat-image step runs --verify, and run-x86-64 boots the image on a usb-storage device (if=none,id=bootusb + usb-storage,bus=xhci.0,bootindex=0), keeping usb-kbd/usb-mouse on the same controller. - test/qemu_test.py: the default boot config now boots off danos-usb.img on a usb-storage device (xHCI + usb-kbd + usb-mouse + the boot stick). The seven per-case qemu_extra blocks that added their own qemu-xhci/usb-kbd/usb-mouse (or a VVFAT stick) collided on id=xhci and are removed — the default provides the bus and the boot device. usb-storage and fat-mount now exercise the real FAT32 boot image (usb-storage reads its 0x55AA boot sector; fat mounts it at /mnt/usb). A build_case override lets a case reuse another's kernel, used by a new usb-boot case: an explicit, named boot-from-USB regression guard. Verified: zig build, zig build test, and zig build check-fat-image are green (FAT32, 128992 clusters, BOOTX64.efi present); a broad sequential QEMU sweep passes — smoke, init, vfs, input, device-manager, usb-report, usb-hid, usb-storage, fat-mount, device-list, driver-restart, acpi-report, iommu, orderly-shutdown, usb-boot, dma, msi, initial-ramdisk, args, process — proving the boot switch holds across kernel tests, the full init tree, the USB stack, the FAT mount, and orderly shutdown.
This commit is contained in:
Daniel Samson
@@ -531,6 +531,36 @@ pub fn build(b: *std.Build) void {
|
||||
const efi_install = b.addInstallArtifact(efiexe, .{ .dest_dir = .{ .override = .{ .custom = "EFI/BOOT" } } });
|
||||
b.getInstallStep().dependOn(&efi_install.step);
|
||||
|
||||
// --- danos-usb.img: the bootable FAT32 USB image ---
|
||||
// Format a real FAT32 image (the in-repo Python builder, no external tools)
|
||||
// holding exactly what the firmware and bootloader need off the ESP: the EFI
|
||||
// stub, the kernel, init, and the initial-ramdisk. QEMU presents this image as
|
||||
// a USB mass-storage device the guest boots from (see run-x86-64 and the test
|
||||
// harness), and the danos fat driver mounts the same image at /mnt/usb.
|
||||
const mk_fat = b.addSystemCommand(&.{"python3"});
|
||||
mk_fat.addFileArg(b.path("tools/make-fat-image.py"));
|
||||
const fat_image = mk_fat.addOutputFileArg("danos-usb.img");
|
||||
mk_fat.addArg("64"); // MiB
|
||||
mk_fat.addArg("EFI/BOOT/BOOTX64.efi");
|
||||
mk_fat.addFileArg(efiexe.getEmittedBin());
|
||||
mk_fat.addArg("system/kernel");
|
||||
mk_fat.addFileArg(exe.getEmittedBin());
|
||||
mk_fat.addArg("system/services/init");
|
||||
mk_fat.addFileArg(init_exe.getEmittedBin());
|
||||
mk_fat.addArg("boot/initial-ramdisk.img");
|
||||
mk_fat.addFileArg(initial_ramdisk_img);
|
||||
const fat_image_install = b.addInstallFile(fat_image, "danos-usb.img");
|
||||
b.getInstallStep().dependOn(&fat_image_install.step);
|
||||
|
||||
// `zig build check-fat-image` — validate the produced image is a real FAT32
|
||||
// with the EFI stub present (the builder's own --verify, no external tools).
|
||||
const check_fat = b.addSystemCommand(&.{"python3"});
|
||||
check_fat.addFileArg(b.path("tools/make-fat-image.py"));
|
||||
check_fat.addArg("--verify");
|
||||
check_fat.addFileArg(fat_image);
|
||||
const check_fat_step = b.step("check-fat-image", "Verify the FAT32 USB image is valid and bootable");
|
||||
check_fat_step.dependOn(&check_fat.step);
|
||||
|
||||
// --- run-x86-64: boot the x86-64 kernel in QEMU via UEFI/OVMF ---
|
||||
// Firmware lives in different places per OS/distro, so probe the known
|
||||
// layouts (Architecture, Debian/Ubuntu, Fedora, macOS Homebrew) and use the first
|
||||
@@ -591,10 +621,13 @@ pub fn build(b: *std.Build) void {
|
||||
});
|
||||
run_efi.addArg("-drive");
|
||||
run_efi.addPrefixedFileArg("if=pflash,format=raw,file=", vars_out);
|
||||
// Present the FHS zig-out to the guest as a FAT drive — it is the boot volume.
|
||||
// Boot off the FAT32 USB image: a mass-storage device on the same xHCI bus as
|
||||
// the keyboard and mouse. OVMF finds \EFI\BOOT\BOOTX64.efi on it and boots.
|
||||
run_efi.addArg("-drive");
|
||||
run_efi.addPrefixedFileArg("if=none,id=bootusb,format=raw,file=", fat_image);
|
||||
run_efi.addArgs(&.{
|
||||
"-drive",
|
||||
b.fmt("format=raw,file=fat:rw:{s}", .{b.install_path}),
|
||||
"-device",
|
||||
"usb-storage,bus=xhci.0,drive=bootusb,removable=on,bootindex=0",
|
||||
"-net",
|
||||
"none",
|
||||
// Emulated display advertising 1280x720 as its native (EDID preferred)
|
||||
|
||||
+35
-34
@@ -67,7 +67,14 @@ ARCHES = {
|
||||
"-machine", "q35", "-m", "128M",
|
||||
"-drive", f"if=pflash,format=raw,readonly=on,file={a['ovmf_code']}",
|
||||
"-drive", f"if=pflash,format=raw,file={vars_fd}",
|
||||
"-drive", f"format=raw,file=fat:rw:{boot_volume}",
|
||||
# Boot off a FAT USB device: the boot volume is a mass-storage device on
|
||||
# the xHCI bus (usb-kbd/usb-mouse ride the same controller). `boot_volume`
|
||||
# is the FAT image the build produces. bootindex=0 steers OVMF to it.
|
||||
"-device", "qemu-xhci,id=xhci",
|
||||
"-device", "usb-kbd,bus=xhci.0",
|
||||
"-device", "usb-mouse,bus=xhci.0",
|
||||
"-drive", f"if=none,id=bootusb,format=raw,file={boot_volume}",
|
||||
"-device", "usb-storage,bus=xhci.0,drive=bootusb,removable=on,bootindex=0",
|
||||
"-net", "none",
|
||||
"-vga", "none", "-device", "VGA,edid=on,xres=1280,yres=720",
|
||||
"-display", "none",
|
||||
@@ -275,9 +282,8 @@ CASES = [
|
||||
{"name": "usb-report",
|
||||
"smp": 4,
|
||||
"timeout": 150,
|
||||
"qemu_extra": ["-device", "qemu-xhci,id=xhci",
|
||||
"-device", "usb-kbd,bus=xhci.0",
|
||||
"-device", "usb-mouse,bus=xhci.0"],
|
||||
# The xHCI bus + usb-kbd/usb-mouse come from the default boot config now
|
||||
# (every case boots off a usb-storage device on that bus).
|
||||
"expect": r"device-manager: child added[\s\S]*"
|
||||
r"device-manager: child added[\s\S]*"
|
||||
r"device-manager: test mode: killing the reporter[\s\S]*"
|
||||
@@ -292,36 +298,37 @@ CASES = [
|
||||
{"name": "usb-hid",
|
||||
"smp": 4,
|
||||
"timeout": 150,
|
||||
"qemu_extra": ["-device", "qemu-xhci,id=xhci",
|
||||
"-device", "usb-kbd,bus=xhci.0",
|
||||
"-device", "usb-mouse,bus=xhci.0"],
|
||||
# usb-kbd/usb-mouse ride the default boot xHCI bus (see qemu_args).
|
||||
"expect": r"(?=[\s\S]*usb-hid/keyboard: ok)(?=[\s\S]*usb-hid/mouse: ok)",
|
||||
"fail": r"DANOS-TEST-RESULT: FAIL"},
|
||||
# USB mass storage end to end: attach a usb-storage device (a FAT volume via
|
||||
# QEMU's VVFAT, so it has a real boot sector), boot the full tree, and let the
|
||||
# manager spawn usb-storage, which opens the device, runs the Bulk-Only /
|
||||
# SCSI bring-up, reads its capacity, and reads block 0 (the 0x55AA boot sig) —
|
||||
# proof of the bulk transfer path + BOT + SCSI end to end.
|
||||
# USB mass storage end to end: the boot usb-storage device (the FAT32 image,
|
||||
# which has a real 0x55AA boot sector) is enough — the manager spawns
|
||||
# usb-storage, which opens the device, runs the Bulk-Only / SCSI bring-up,
|
||||
# reads its capacity, and reads block 0 (the 0x55AA boot sig). Proof of the
|
||||
# bulk transfer path + BOT + SCSI end to end.
|
||||
{"name": "usb-storage",
|
||||
"smp": 4,
|
||||
"timeout": 150,
|
||||
"qemu_extra": ["-device", "qemu-xhci,id=xhci",
|
||||
"-drive", "if=none,id=stick,format=raw,file=fat:rw:" + os.path.join(REPO, "zig-out"),
|
||||
"-device", "usb-storage,drive=stick,bus=xhci.0"],
|
||||
"expect": r"usb-storage: ready[\s\S]*usb-storage: block 0 signature 0x55aa",
|
||||
"fail": r"DANOS-TEST-RESULT: FAIL"},
|
||||
# FAT mount end to end: attach a usb-storage device (a FAT volume via VVFAT),
|
||||
# boot the full tree, and let the fat server mount it into the VFS at /mnt/usb.
|
||||
# A fat-test client then lists and reads through the mount — proof of the whole
|
||||
# stack: block device -> FAT parse -> VFS routing -> file read.
|
||||
# FAT mount end to end: the fat server mounts the boot usb-storage device (the
|
||||
# FAT32 image) into the VFS at /mnt/usb. A fat-test client then lists and reads
|
||||
# through the mount — proof of the whole stack: block device -> FAT parse ->
|
||||
# VFS routing -> file read.
|
||||
{"name": "fat-mount",
|
||||
"smp": 4,
|
||||
"timeout": 150,
|
||||
"qemu_extra": ["-device", "qemu-xhci,id=xhci",
|
||||
"-drive", "if=none,id=stick,format=raw,file=fat:rw:" + os.path.join(REPO, "zig-out"),
|
||||
"-device", "usb-storage,drive=stick,bus=xhci.0"],
|
||||
"expect": r"fat: mounted /mnt/usb[\s\S]*fat-test: ok",
|
||||
"fail": r"DANOS-TEST-RESULT: FAIL"},
|
||||
# Boot-from-USB smoke: the whole system now boots off the FAT32 image on a
|
||||
# usb-storage device (OVMF -> \EFI\BOOT\BOOTX64.efi -> kernel), so the kernel
|
||||
# reaching its PASS marker at all proves the USB boot path end to end. Reuses
|
||||
# the smoke kernel build; the value is the explicit, named regression guard.
|
||||
{"name": "usb-boot",
|
||||
"build_case": "smoke",
|
||||
"qmp_after": {"delay": 2, "command": "query-status"},
|
||||
"expect": r"DANOS-TEST-RESULT: PASS",
|
||||
"fail": r"DANOS-TEST-RESULT: FAIL"},
|
||||
# M20.1: the ring-3 AML parse (the acpi service maps the blobs and parses
|
||||
# them) finds exactly the Device count the kernel's own parse produced.
|
||||
{"name": "acpi-parse",
|
||||
@@ -367,9 +374,6 @@ CASES = [
|
||||
{"name": "acpi-report",
|
||||
"smp": 4,
|
||||
"timeout": 150,
|
||||
"qemu_extra": ["-device", "qemu-xhci,id=xhci",
|
||||
"-device", "usb-kbd,bus=xhci.0",
|
||||
"-device", "usb-mouse,bus=xhci.0"],
|
||||
"expect": r"acpi: reported PNP0303 \(device \d+, 3 resources\)[\s\S]*"
|
||||
r"acpi: reported PNP0F13 \(device \d+, 1 resources\)",
|
||||
"fail": r"DANOS-TEST-RESULT: FAIL"},
|
||||
@@ -386,9 +390,6 @@ CASES = [
|
||||
{"name": "device-list",
|
||||
"smp": 4,
|
||||
"timeout": 150,
|
||||
"qemu_extra": ["-device", "qemu-xhci,id=xhci",
|
||||
"-device", "usb-kbd,bus=xhci.0",
|
||||
"-device", "usb-mouse,bus=xhci.0"],
|
||||
"expect": r"device-list: \d+ devices[\s\S]*"
|
||||
r"device-list: subscribed[\s\S]*"
|
||||
r"device-manager: test mode: killing the reporter[\s\S]*"
|
||||
@@ -401,9 +402,6 @@ CASES = [
|
||||
{"name": "driver-restart",
|
||||
"smp": 4,
|
||||
"timeout": 150,
|
||||
"qemu_extra": ["-device", "qemu-xhci,id=xhci",
|
||||
"-device", "usb-kbd,bus=xhci.0",
|
||||
"-device", "usb-mouse,bus=xhci.0"],
|
||||
"expect": r"usb-xhci-bus: hello acknowledged[\s\S]*"
|
||||
r"device-manager: restarting crash-test[\s\S]*"
|
||||
r"device-manager: crash-test is failing repeatedly",
|
||||
@@ -509,12 +507,15 @@ def qmp_send(path, command):
|
||||
|
||||
|
||||
def run_case(arch, case):
|
||||
err = build(arch, case["name"])
|
||||
# A case's kernel build defaults to its name; `build_case` decouples the two
|
||||
# so a case can reuse another's kernel (e.g. usb-boot reuses smoke's).
|
||||
err = build(arch, case.get("build_case", case["name"]))
|
||||
if err:
|
||||
return False, "build failed:\n" + err
|
||||
|
||||
# zig-out is the FHS boot volume; hand it to the guest as-is (see qemu_args).
|
||||
boot_volume = os.path.join(REPO, "zig-out")
|
||||
# The bootable FAT32 USB image the build produced (tools/make-fat-image.py),
|
||||
# presented to the guest as a usb-storage device (see qemu_args).
|
||||
boot_volume = os.path.join(REPO, "zig-out", "danos-usb.img")
|
||||
vars_fd = os.path.join(WORK, "vars.fd")
|
||||
shutil.copy(arch["ovmf_vars"], vars_fd)
|
||||
serial = os.path.join(WORK, "serial.log")
|
||||
|
||||
@@ -0,0 +1,362 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Format a real FAT32 image from a set of host files — the danos boot volume.
|
||||
|
||||
Mirrors tools/make-initial-ramdisk.py in spirit: pure Python 3 standard library,
|
||||
no external tools (no mkfs.fat / mtools). It writes a valid FAT32 filesystem — a
|
||||
boot sector + BPB, an FSInfo sector, a backup boot sector, two FATs, and a
|
||||
directory tree of clusters — so UEFI/OVMF boots \\EFI\\BOOT\\BOOTX64.efi off it
|
||||
and the danos FAT driver mounts the same image.
|
||||
|
||||
make-fat-image.py <out.img> <size-MiB> [<dest-path> <host-file>]...
|
||||
make-fat-image.py --verify <out.img>
|
||||
|
||||
Each <dest-path> is a forward-slash path inside the image (e.g.
|
||||
"EFI/BOOT/BOOTX64.efi"); intermediate directories are created. Names that do not
|
||||
fit 8.3 get a mangled short name plus long-file-name (LFN) entries.
|
||||
"""
|
||||
|
||||
import struct
|
||||
import sys
|
||||
|
||||
SECTOR = 512
|
||||
SECTORS_PER_CLUSTER = 1 # 512-byte clusters keep the cluster count high for FAT32
|
||||
RESERVED_SECTORS = 32
|
||||
NUM_FATS = 2
|
||||
CLUSTER_BYTES = SECTOR * SECTORS_PER_CLUSTER
|
||||
|
||||
END_OF_CHAIN = 0x0FFFFFFF
|
||||
BAD_CLUSTER = 0x0FFFFFF7
|
||||
|
||||
ATTR_ARCHIVE = 0x20
|
||||
ATTR_DIRECTORY = 0x10
|
||||
ATTR_LONG_NAME = 0x0F
|
||||
|
||||
VALID_83 = set("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789$%'-_@~!(){}^#& ")
|
||||
|
||||
|
||||
def fat32_geometry(total_sectors):
|
||||
"""Solve for the FAT size (sectors per FAT) and cluster count that fit."""
|
||||
fat_size = 1
|
||||
while True:
|
||||
data_sectors = total_sectors - RESERVED_SECTORS - NUM_FATS * fat_size
|
||||
cluster_count = data_sectors // SECTORS_PER_CLUSTER
|
||||
needed = ((cluster_count + 2) * 4 + SECTOR - 1) // SECTOR
|
||||
if needed <= fat_size:
|
||||
return fat_size, cluster_count
|
||||
fat_size = needed
|
||||
|
||||
|
||||
class Fat32Image:
|
||||
def __init__(self, total_sectors):
|
||||
self.total_sectors = total_sectors
|
||||
self.fat_size, self.cluster_count = fat32_geometry(total_sectors)
|
||||
if self.cluster_count < 65525:
|
||||
sys.exit(f"error: image too small for FAT32 ({self.cluster_count} clusters "
|
||||
f"< 65525); use a larger size")
|
||||
self.first_data_sector = RESERVED_SECTORS + NUM_FATS * self.fat_size
|
||||
# The FAT, in memory: entry 0 media, entry 1 EOC, entry 2 the root dir.
|
||||
self.fat = [0] * (self.cluster_count + 2)
|
||||
self.fat[0] = 0x0FFFFFF8
|
||||
self.fat[1] = END_OF_CHAIN
|
||||
self.fat[2] = END_OF_CHAIN
|
||||
self.next_free = 3
|
||||
self.cluster_data = {} # cluster number -> bytes (one cluster's worth)
|
||||
|
||||
def alloc(self):
|
||||
cluster = self.next_free
|
||||
if cluster >= self.cluster_count + 2:
|
||||
sys.exit("error: image out of clusters")
|
||||
self.next_free += 1
|
||||
self.fat[cluster] = END_OF_CHAIN
|
||||
return cluster
|
||||
|
||||
def store_chain(self, content):
|
||||
"""Allocate a cluster chain holding `content` and return its first cluster."""
|
||||
length = max(1, (len(content) + CLUSTER_BYTES - 1) // CLUSTER_BYTES)
|
||||
clusters = [self.alloc() for _ in range(length)]
|
||||
for i in range(length - 1):
|
||||
self.fat[clusters[i]] = clusters[i + 1]
|
||||
for i, cluster in enumerate(clusters):
|
||||
chunk = content[i * CLUSTER_BYTES:(i + 1) * CLUSTER_BYTES]
|
||||
self.cluster_data[cluster] = chunk + b"\x00" * (CLUSTER_BYTES - len(chunk))
|
||||
return clusters[0]
|
||||
|
||||
def store_directory(self, first_cluster, entries):
|
||||
"""Write directory `entries` (bytes) into `first_cluster`, extending the chain."""
|
||||
length = max(1, (len(entries) + CLUSTER_BYTES - 1) // CLUSTER_BYTES)
|
||||
clusters = [first_cluster]
|
||||
for _ in range(length - 1):
|
||||
clusters.append(self.alloc())
|
||||
for i in range(len(clusters) - 1):
|
||||
self.fat[clusters[i]] = clusters[i + 1]
|
||||
for i, cluster in enumerate(clusters):
|
||||
chunk = entries[i * CLUSTER_BYTES:(i + 1) * CLUSTER_BYTES]
|
||||
self.cluster_data[cluster] = chunk + b"\x00" * (CLUSTER_BYTES - len(chunk))
|
||||
|
||||
def cluster_sector(self, cluster):
|
||||
return self.first_data_sector + (cluster - 2) * SECTORS_PER_CLUSTER
|
||||
|
||||
def serialize(self):
|
||||
image = bytearray(self.total_sectors * SECTOR)
|
||||
image[0:SECTOR] = self.boot_sector()
|
||||
image[SECTOR:2 * SECTOR] = self.fsinfo_sector()
|
||||
image[6 * SECTOR:7 * SECTOR] = self.boot_sector() # backup boot sector
|
||||
# Both FATs.
|
||||
fat_bytes = b"".join(struct.pack("<I", entry & 0x0FFFFFFF) for entry in self.fat)
|
||||
fat_bytes += b"\x00" * (self.fat_size * SECTOR - len(fat_bytes))
|
||||
for copy in range(NUM_FATS):
|
||||
base = (RESERVED_SECTORS + copy * self.fat_size) * SECTOR
|
||||
image[base:base + len(fat_bytes)] = fat_bytes
|
||||
# The data region (clusters).
|
||||
for cluster, data in self.cluster_data.items():
|
||||
base = self.cluster_sector(cluster) * SECTOR
|
||||
image[base:base + len(data)] = data
|
||||
return bytes(image)
|
||||
|
||||
def boot_sector(self):
|
||||
sector = bytearray(SECTOR)
|
||||
# BPB.
|
||||
struct.pack_into(
|
||||
"<3s8sHBHBHHBHHHII", sector, 0,
|
||||
b"\xEB\x58\x90", # jump
|
||||
b"MSWIN4.1", # OEM name (widest firmware compatibility)
|
||||
SECTOR, # bytes per sector
|
||||
SECTORS_PER_CLUSTER, # sectors per cluster
|
||||
RESERVED_SECTORS, # reserved sector count
|
||||
NUM_FATS, # number of FATs
|
||||
0, # root entry count (0 for FAT32)
|
||||
0, # total sectors 16 (0 -> use 32)
|
||||
0xF8, # media descriptor
|
||||
0, # FAT size 16 (0 for FAT32)
|
||||
32, # sectors per track
|
||||
2, # heads
|
||||
0, # hidden sectors
|
||||
self.total_sectors, # total sectors 32
|
||||
)
|
||||
# FAT32 extended BPB (offset 36).
|
||||
struct.pack_into(
|
||||
"<IHHIHH12sBBBI11s8s", sector, 36,
|
||||
self.fat_size, # FAT size 32
|
||||
0, # extended flags
|
||||
0, # filesystem version
|
||||
2, # root cluster
|
||||
1, # FSInfo sector
|
||||
6, # backup boot sector
|
||||
b"\x00" * 12, # reserved
|
||||
0x80, # drive number
|
||||
0, # reserved
|
||||
0x29, # extended boot signature
|
||||
0x12345678, # volume id
|
||||
b"DANOS ", # volume label
|
||||
b"FAT32 ", # filesystem type
|
||||
)
|
||||
sector[510] = 0x55
|
||||
sector[511] = 0xAA
|
||||
return bytes(sector)
|
||||
|
||||
def fsinfo_sector(self):
|
||||
sector = bytearray(SECTOR)
|
||||
struct.pack_into("<I", sector, 0, 0x41615252) # lead signature
|
||||
struct.pack_into("<I", sector, 484, 0x61417272) # struct signature
|
||||
free = self.cluster_count - (self.next_free - 2)
|
||||
struct.pack_into("<I", sector, 488, free) # free count
|
||||
struct.pack_into("<I", sector, 492, self.next_free) # next free hint
|
||||
struct.pack_into("<I", sector, 508, 0xAA550000) # trail signature
|
||||
return bytes(sector)
|
||||
|
||||
|
||||
def lfn_checksum(short_name):
|
||||
checksum = 0
|
||||
for byte in short_name:
|
||||
checksum = (((checksum & 1) << 7) + (checksum >> 1) + byte) & 0xFF
|
||||
return checksum
|
||||
|
||||
|
||||
def short_name_for(name, used):
|
||||
"""Return (raw 11-byte 8.3 name, needs_lfn)."""
|
||||
if "." in name and not name.startswith("."):
|
||||
base, ext = name.rsplit(".", 1)
|
||||
else:
|
||||
base, ext = name, ""
|
||||
upper_base, upper_ext = base.upper(), ext.upper()
|
||||
# A name fits 8.3 if it is short enough and uses valid characters; a lowercase
|
||||
# name is simply stored uppercased (FAT is case-insensitive, so the bootloader
|
||||
# and the danos driver still find it). Only genuinely non-8.3 names (too long,
|
||||
# e.g. initial-ramdisk.img) get a mangled short name plus LFN entries.
|
||||
fits = (1 <= len(base) <= 8 and len(ext) <= 3
|
||||
and all(c in VALID_83 for c in upper_base + upper_ext))
|
||||
if fits:
|
||||
return (upper_base.ljust(8) + upper_ext.ljust(3)).encode("ascii"), False
|
||||
# Mangle to STEM~N.EXT.
|
||||
stem = "".join(c for c in upper_base if c in VALID_83 and c != " ")[:6] or "FILE"
|
||||
index = 1
|
||||
while True:
|
||||
candidate = f"{stem}~{index}".ljust(8)[:8] + upper_ext.ljust(3)[:3]
|
||||
raw = candidate.encode("ascii")
|
||||
if raw not in used:
|
||||
used.add(raw)
|
||||
return raw, True
|
||||
index += 1
|
||||
|
||||
|
||||
def lfn_entries(name, short_raw):
|
||||
checksum = lfn_checksum(short_raw)
|
||||
units = list(name.encode("utf-16-le"))
|
||||
pairs = [bytes(units[i:i + 2]) for i in range(0, len(units), 2)]
|
||||
pairs.append(b"\x00\x00") # null terminator
|
||||
while len(pairs) % 13 != 0:
|
||||
pairs.append(b"\xff\xff")
|
||||
count = len(pairs) // 13
|
||||
out = bytearray()
|
||||
for sequence in range(count, 0, -1): # stored last-logical-first
|
||||
piece = pairs[(sequence - 1) * 13:sequence * 13]
|
||||
entry = bytearray(32)
|
||||
entry[0] = sequence | (0x40 if sequence == count else 0)
|
||||
for i in range(5):
|
||||
entry[1 + i * 2:1 + i * 2 + 2] = piece[i]
|
||||
entry[11] = ATTR_LONG_NAME
|
||||
entry[12] = 0
|
||||
entry[13] = checksum
|
||||
for i in range(6):
|
||||
entry[14 + i * 2:14 + i * 2 + 2] = piece[5 + i]
|
||||
entry[26:28] = b"\x00\x00"
|
||||
for i in range(2):
|
||||
entry[28 + i * 2:28 + i * 2 + 2] = piece[11 + i]
|
||||
out += entry
|
||||
return bytes(out)
|
||||
|
||||
|
||||
def short_entry(raw11, attributes, cluster, size):
|
||||
return struct.pack(
|
||||
"<11sBBBHHHHHHHI",
|
||||
raw11, attributes, 0, 0, 0, 0, 0,
|
||||
(cluster >> 16) & 0xFFFF, 0, 0, cluster & 0xFFFF, size,
|
||||
)
|
||||
|
||||
|
||||
def write_directory(image, cluster, children, parent_cluster, is_root):
|
||||
"""Recursively lay out a directory: allocate child clusters, build entries."""
|
||||
entries = bytearray()
|
||||
if not is_root:
|
||||
entries += short_entry(b". ", ATTR_DIRECTORY, cluster, 0)
|
||||
parent = 0 if parent_cluster == 2 else parent_cluster
|
||||
entries += short_entry(b".. ", ATTR_DIRECTORY, parent, 0)
|
||||
used_short_names = set()
|
||||
for name, child in children.items():
|
||||
raw, needs_lfn = short_name_for(name, used_short_names)
|
||||
used_short_names.add(raw)
|
||||
if child["type"] == "dir":
|
||||
child_cluster = image.alloc()
|
||||
if needs_lfn:
|
||||
entries += lfn_entries(name, raw)
|
||||
entries += short_entry(raw, ATTR_DIRECTORY, child_cluster, 0)
|
||||
write_directory(image, child_cluster, child["children"], cluster, False)
|
||||
else:
|
||||
data = child["data"]
|
||||
first = image.store_chain(data) if data else 0
|
||||
if needs_lfn:
|
||||
entries += lfn_entries(name, raw)
|
||||
entries += short_entry(raw, ATTR_ARCHIVE, first, len(data))
|
||||
image.store_directory(cluster, bytes(entries))
|
||||
|
||||
|
||||
def build_tree(pairs):
|
||||
root = {}
|
||||
for dest, host in pairs:
|
||||
with open(host, "rb") as handle:
|
||||
data = handle.read()
|
||||
parts = [p for p in dest.replace("\\", "/").split("/") if p]
|
||||
node = root
|
||||
for part in parts[:-1]:
|
||||
node = node.setdefault(part, {"type": "dir", "children": {}})["children"]
|
||||
node[parts[-1]] = {"type": "file", "data": data}
|
||||
return root
|
||||
|
||||
|
||||
def build(out_path, size_mib, pairs):
|
||||
total_sectors = size_mib * 1024 * 1024 // SECTOR
|
||||
image = Fat32Image(total_sectors)
|
||||
tree = build_tree(pairs)
|
||||
write_directory(image, 2, tree, 0, True)
|
||||
with open(out_path, "wb") as handle:
|
||||
handle.write(image.serialize())
|
||||
print(f"make-fat-image: wrote {out_path} "
|
||||
f"({size_mib} MiB FAT32, {image.cluster_count} clusters)")
|
||||
|
||||
|
||||
def verify(path):
|
||||
with open(path, "rb") as handle:
|
||||
data = handle.read()
|
||||
if len(data) < SECTOR or data[510] != 0x55 or data[511] != 0xAA:
|
||||
sys.exit("verify: missing 0x55AA boot signature")
|
||||
bytes_per_sector, sectors_per_cluster = struct.unpack_from("<HB", data, 11)
|
||||
reserved, num_fats = struct.unpack_from("<H", data, 14)[0], data[16]
|
||||
fat_size_32, root_cluster = struct.unpack_from("<I", data, 36)[0], struct.unpack_from("<I", data, 44)[0]
|
||||
total_sectors = struct.unpack_from("<I", data, 32)[0]
|
||||
if bytes_per_sector != SECTOR or sectors_per_cluster == 0 or num_fats == 0 or fat_size_32 == 0:
|
||||
sys.exit("verify: implausible BPB")
|
||||
first_data = reserved + num_fats * fat_size_32
|
||||
cluster_count = (total_sectors - first_data) // sectors_per_cluster
|
||||
if cluster_count < 65525:
|
||||
sys.exit(f"verify: not FAT32 ({cluster_count} clusters)")
|
||||
# Resolve EFI/BOOT/BOOTX64.efi through the directory tree to prove it is present.
|
||||
if not _resolve(data, ["EFI", "BOOT", "BOOTX64.EFI"], root_cluster,
|
||||
reserved, num_fats, fat_size_32, first_data, sectors_per_cluster):
|
||||
sys.exit("verify: EFI/BOOT/BOOTX64.efi not found")
|
||||
print(f"verify: {path} is FAT32 ({cluster_count} clusters); EFI/BOOT/BOOTX64.efi present")
|
||||
|
||||
|
||||
def _read_fat(data, cluster, reserved):
|
||||
offset = reserved * SECTOR + cluster * 4
|
||||
return struct.unpack_from("<I", data, offset)[0] & 0x0FFFFFFF
|
||||
|
||||
|
||||
def _resolve(data, parts, cluster, reserved, num_fats, fat_size, first_data, spc):
|
||||
for part in parts:
|
||||
cluster = _find(data, cluster, part, reserved, first_data, spc)
|
||||
if cluster is None:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _find(data, dir_cluster, name, reserved, first_data, spc):
|
||||
target = name.upper()
|
||||
cluster = dir_cluster
|
||||
guard = 0
|
||||
while cluster >= 2 and cluster < BAD_CLUSTER and guard < 100000:
|
||||
sector = first_data + (cluster - 2) * spc
|
||||
for s in range(spc):
|
||||
base = (sector + s) * SECTOR
|
||||
for i in range(SECTOR // 32):
|
||||
entry = data[base + i * 32:base + i * 32 + 32]
|
||||
if entry[0] == 0x00:
|
||||
return None
|
||||
if entry[0] == 0xE5 or (entry[11] & ATTR_LONG_NAME) == ATTR_LONG_NAME:
|
||||
continue
|
||||
raw = entry[0:11]
|
||||
short = (raw[0:8].rstrip().decode("latin1") +
|
||||
("." + raw[8:11].rstrip().decode("latin1") if raw[8:11].strip() else "")).upper()
|
||||
if short == target:
|
||||
return ((entry[20] | (entry[21] << 8)) << 16) | (entry[26] | (entry[27] << 8))
|
||||
cluster = _read_fat(data, cluster, reserved)
|
||||
guard += 1
|
||||
return None
|
||||
|
||||
|
||||
def main(argv):
|
||||
if len(argv) == 3 and argv[1] == "--verify":
|
||||
verify(argv[2])
|
||||
return 0
|
||||
if len(argv) < 3 or (len(argv) - 3) % 2 != 0:
|
||||
sys.exit("usage: make-fat-image.py <out.img> <size-MiB> [<dest> <host>]...\n"
|
||||
" make-fat-image.py --verify <out.img>")
|
||||
out_path = argv[1]
|
||||
size_mib = int(argv[2])
|
||||
rest = argv[3:]
|
||||
pairs = [(rest[i], rest[i + 1]) for i in range(0, len(rest), 2)]
|
||||
build(out_path, size_mib, pairs)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv))
|
||||
Reference in New Issue
Block a user