From df80b320cb773a6f2332777b054b33690c93ba36 Mon Sep 17 00:00:00 2001 From: Omarchybot Date: Sun, 23 Aug 2026 12:57:03 +0200 Subject: [PATCH] Stop the install dying on non-UTF-8 command output (#118) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Stop the install dying on non-UTF-8 command output efibootmgr prints what firmware gave it, and an HP ProBook's NVRAM held a legacy BBS entry whose device path was raw binary. `_read_efibootmgr` captured that with `text=True`, so Python decoded it strictly and a byte in a boot entry nothing here reads killed the whole "Installing Arch + Omarchy" phase. Two machines have hit it now, on 0x80 and 0xe0. Every text capture in the orchestrator had the same exposure — partition labels reach lsblk and blkid the same way firmware strings reach efibootmgr — so they all go through one `capture()` that replaces what it cannot decode. The file reads here already did this; the command reads were the gap. Replacing bytes has its own failure mode, and it is worse than the crash: a mangled UUID would be written into fstab and the kernel cmdline, and `_validate_pre_mounted_filesystems` reads it back from the same command, agrees with itself, and ships a machine that cannot find its root. So `require_text()` refuses a value the installed system depends on — UUIDs, device names, the btrfs device handed to mount — while the boot-entry labels stay tolerant on purpose: the decisions taken from those are advisory, and refusing a junk one would abort the install this change exists to let finish. The regression test runs a stand-in efibootmgr emitting the reported byte sequence, so it exercises the decode rather than a mock. test_keyboard.py loads the module through the package now that keyboard.py imports a sibling. Closes #116 🤖 Generated by Opus 5 in Claude Code. Reviewed by Codex XHigh. Co-Authored-By: Codex XHigh * Say accurately why the boot-entry parsing stays tolerant The previous commit called the decisions taken from a firmware label "advisory". They are not: a label match selects entries for deletion and rewrites the boot order, and a missing one aborts validation. Staying tolerant is still right, for a reason the comment did not give — an entry's text is its label and its device path together, the junk lives in the path, and nothing mangled reaches efibootmgr from there anyway, since only four-digit ASCII boot numbers become arguments and a boot order is filtered through the entries that parsed. A label that loses a byte costs a stale entry left behind, not a broken system. The test docstring credited one machine with a fixture that carries the bytes from two. Co-Authored-By: Codex XHigh --------- Co-authored-by: Codex XHigh --- .../share/omarchy-iso/orchestrator/command.py | 54 ++++++++ .../omarchy-iso/orchestrator/keyboard.py | 17 +-- .../omarchy-iso/orchestrator/phases_impl.py | 50 +++---- test/unit/test_command_capture.py | 129 ++++++++++++++++++ test/unit/test_keyboard.py | 10 +- 5 files changed, 210 insertions(+), 50 deletions(-) create mode 100644 configs/airootfs/usr/share/omarchy-iso/orchestrator/command.py create mode 100644 test/unit/test_command_capture.py diff --git a/configs/airootfs/usr/share/omarchy-iso/orchestrator/command.py b/configs/airootfs/usr/share/omarchy-iso/orchestrator/command.py new file mode 100644 index 0000000..780accd --- /dev/null +++ b/configs/airootfs/usr/share/omarchy-iso/orchestrator/command.py @@ -0,0 +1,54 @@ +r"""Reading text out of the commands the installer shells out to. + +Firmware and filesystem metadata reach the orchestrator as command output, and +neither is obliged to be UTF-8: an HP ProBook's NVRAM held a legacy BBS entry +whose device path began `BBS(HD,\x80\x7f\xff\x04`, efibootmgr printed those +bytes verbatim, and subprocess's strict text decoding turned a boot entry +nothing here reads into a failed install. + +So capture() replaces what it cannot decode, and the boot-entry parsing that +reads its output stays tolerant on purpose: an entry's text is its label and its +device path together, and the junk lives in the path, so refusing one would abort +the install this exists to let finish. Nothing mangled reaches efibootmgr from +there either way — only four-digit ASCII boot numbers become arguments, and a +boot order is filtered through the entries that parsed — so a label that loses a +byte costs a stale entry left behind rather than a broken system. + +An identifier is the other case entirely: require_text() refuses a UUID or a +device name that came through mangled, because that one is written into fstab and +the kernel cmdline, or handed to mount. +""" + +from __future__ import annotations + +import subprocess + +REPLACEMENT = "\ufffd" + + +def capture(cmd: list[str], *, check: bool = False) -> subprocess.CompletedProcess: + """Run cmd and return it with stdout/stderr decoded, bad bytes replaced.""" + return subprocess.run( + cmd, + check=check, + capture_output=True, + text=True, + errors="replace", + ) + + +def require_text(value: str, what: str) -> str: + """Refuse a value the installed system depends on if bytes were replaced in it. + + Validation reads these back from the same command that wrote them, so a + mangled UUID agrees with itself and ships a machine that cannot find its + root at boot. Stopping here says why while someone is still watching. + """ + if REPLACEMENT in value: + raise RuntimeError(f"{what} is not valid text: {value!r}") + return value + + +def capture_identifier(cmd: list[str], what: str) -> str: + """Capture a single value the installed system will depend on.""" + return require_text(capture(cmd, check=True).stdout.strip(), what) diff --git a/configs/airootfs/usr/share/omarchy-iso/orchestrator/keyboard.py b/configs/airootfs/usr/share/omarchy-iso/orchestrator/keyboard.py index 48f9431..48c457c 100644 --- a/configs/airootfs/usr/share/omarchy-iso/orchestrator/keyboard.py +++ b/configs/airootfs/usr/share/omarchy-iso/orchestrator/keyboard.py @@ -11,9 +11,10 @@ generated: nothing on an Omarchy system reads it. from __future__ import annotations -import subprocess from pathlib import Path +from .command import capture + def configure_keyboard(target: Path, language: str) -> bool: """Write the console keymap into a mounted target without booting it. @@ -25,12 +26,7 @@ def configure_keyboard(target: Path, language: str) -> bool: if not language.strip(): return True - result = subprocess.run( - ["localectl", "--no-pager", "list-keymaps"], - check=False, - text=True, - capture_output=True, - ) + result = capture(["localectl", "--no-pager", "list-keymaps"]) if result.returncode != 0: detail = result.stderr.strip() or "localectl returned an error" raise RuntimeError(f"Unable to list keyboard layouts: {detail}") @@ -47,12 +43,7 @@ def configure_keyboard(target: Path, language: str) -> bool: None, ) - result = subprocess.run( - ["systemd-firstboot", f"--root={target}", f"--keymap={language}", "--force"], - check=False, - text=True, - capture_output=True, - ) + result = capture(["systemd-firstboot", f"--root={target}", f"--keymap={language}", "--force"]) if result.returncode != 0: detail = result.stderr.strip() or "systemd-firstboot returned an error" raise RuntimeError(f"Unable to configure keyboard layout {language}: {detail}") diff --git a/configs/airootfs/usr/share/omarchy-iso/orchestrator/phases_impl.py b/configs/airootfs/usr/share/omarchy-iso/orchestrator/phases_impl.py index 2745139..6d48624 100644 --- a/configs/airootfs/usr/share/omarchy-iso/orchestrator/phases_impl.py +++ b/configs/airootfs/usr/share/omarchy-iso/orchestrator/phases_impl.py @@ -33,6 +33,7 @@ from dataclasses import replace from pathlib import Path from . import archinstall_adapter as arch +from .command import capture, capture_identifier, require_text from .context import InstallContext from .keyboard import configure_keyboard from .ui import error, info @@ -793,11 +794,7 @@ def verify_protected_mounts(ctx: InstallContext) -> None: def _is_mountpoint(path: Path) -> bool: - res = subprocess.run( - ["findmnt", "-rn", str(path)], - capture_output=True, - text=True, - ) + res = capture(["findmnt", "-rn", str(path)]) return res.returncode == 0 and bool(res.stdout.strip()) @@ -811,11 +808,9 @@ def _btrfs_root_device(ctx: InstallContext) -> str: def _blkid_uuid(device: str) -> str: - res = subprocess.run( - ["blkid", "-s", "UUID", "-o", "value", device], - capture_output=True, text=True, check=True, + uuid = capture_identifier( + ["blkid", "-s", "UUID", "-o", "value", device], f"the UUID of {device}" ) - uuid = res.stdout.strip() if not uuid: raise RuntimeError(f"blkid returned no UUID for {device}") return uuid @@ -828,11 +823,9 @@ def _esp_device(ctx: InstallContext) -> str: boot = _boot_intent(ctx) esp_mp = ctx.target / boot["esp_mount"].lstrip("/") - res = subprocess.run( - ["findmnt", "-n", "-o", "SOURCE", str(esp_mp)], - capture_output=True, text=True, check=True, + dev = capture_identifier( + ["findmnt", "-n", "-o", "SOURCE", str(esp_mp)], f"the ESP device at {esp_mp}" ) - dev = res.stdout.strip() if not dev: raise RuntimeError(f"could not resolve ESP device at {esp_mp}") return dev @@ -904,10 +897,7 @@ _BOOT_ORDER_RE = re.compile(r"^BootOrder:\s*(.*)$") def _read_efibootmgr() -> dict: - res = subprocess.run( - ["efibootmgr"], - capture_output=True, text=True, check=True, - ) + res = capture(["efibootmgr"], check=True) entries: dict[str, str] = {} order: list[str] = [] for line in res.stdout.splitlines(): @@ -926,16 +916,14 @@ def _find_label_entries(entries: dict[str, str], needle: str) -> list[str]: def _split_partition_device(part_dev: str) -> tuple[str, int]: - parent = subprocess.run( - ["lsblk", "-ndo", "PKNAME", part_dev], - capture_output=True, text=True, check=True, - ).stdout.strip() + parent = capture_identifier( + ["lsblk", "-ndo", "PKNAME", part_dev], f"the parent disk of {part_dev}" + ) if not parent: raise RuntimeError(f"could not find parent disk for {part_dev}") - part_num = subprocess.run( - ["lsblk", "-ndo", "PARTN", part_dev], - capture_output=True, text=True, check=True, - ).stdout.strip() + part_num = capture_identifier( + ["lsblk", "-ndo", "PARTN", part_dev], f"the partition number of {part_dev}" + ) if not part_num: raise RuntimeError(f"could not find partition number for {part_dev}") return f"/dev/{parent}", int(part_num) @@ -994,7 +982,7 @@ def _debug_run(ctx: InstallContext, cmd: list[str]) -> None: if not _install_debug_enabled(): return _debug_log(ctx, "+ " + " ".join(cmd)) - proc = subprocess.run(cmd, check=False, text=True, capture_output=True) + proc = capture(cmd) if proc.stdout: with ctx.log_path.open("a", encoding="utf-8") as log: for line in proc.stdout.splitlines(): @@ -1805,7 +1793,10 @@ def create_factory_snapshot(ctx: InstallContext) -> None: info("› target root is not the @ subvolume; skipping factory snapshot") return - device = (_findmnt_value(ctx.target, "SOURCE") or "").split("[")[0] + device = require_text( + (_findmnt_value(ctx.target, "SOURCE") or "").split("[")[0], + f"the btrfs device backing {ctx.target}", + ) if not device: raise RuntimeError(f"could not determine the btrfs device backing {ctx.target}") @@ -1862,10 +1853,7 @@ def _scrub_factory_snapshot(factory: Path) -> None: def _findmnt_value(path: Path, column: str) -> str | None: - res = subprocess.run( - ["findmnt", "-no", column, str(path)], - capture_output=True, text=True, - ) + res = capture(["findmnt", "-no", column, str(path)]) value = res.stdout.strip() return value if res.returncode == 0 and value else None diff --git a/test/unit/test_command_capture.py b/test/unit/test_command_capture.py new file mode 100644 index 0000000..5dc9bdf --- /dev/null +++ b/test/unit/test_command_capture.py @@ -0,0 +1,129 @@ +#!/usr/bin/python + +"""Undecodable bytes in command output, and where tolerating them stops. + +The orchestrator reads firmware and filesystem metadata out of commands whose +output is not required to be UTF-8: noise in a boot entry must not stop an +install, while a mangled identifier must stop one. These tests run stand-in +commands emitting the bytes two HP firmwares actually put in a legacy BBS entry, +so they exercise the decode itself rather than a mocked result. +""" + +import os +import sys +import tempfile +import types +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "configs/airootfs/usr/share/omarchy-iso")) + +# phases_impl imports the archinstall adapter at module scope, which pulls in +# the archinstall library that only exists on the live ISO. +sys.modules.setdefault( + "orchestrator.archinstall_adapter", + types.ModuleType("orchestrator.archinstall_adapter"), +) + +from orchestrator import phases_impl # noqa: E402 +from orchestrator.command import capture, capture_identifier # noqa: E402 + +# Boot0000 is the entry that aborted the install: raw bytes inside a BBS device +# path, in a legacy entry nothing in the installer reads. 0x80 and 0xe0 are the +# bytes two different HP firmwares emitted. +EFIBOOTMGR_OUTPUT = ( + b"BootCurrent: 0003\n" + b"Timeout: 0 seconds\n" + b"BootOrder: 2001,0003,0000\n" + b"Boot0000* Notebook Hard Drive\tBBS(HD,\x80\x7f\xff\x04\xe0\x7f,)\n" + b"Boot0003* Omarchy\tHD(1,GPT,0d9c9d4f)/File(\\EFI\\limine\\BOOTX64.EFI)\n" + b"Boot2001* USB Drive (UEFI)\tRC\n" +) + + +class CommandCaptureTest(unittest.TestCase): + def setUp(self): + self.bin_dir = None + + def fake_bin(self) -> Path: + """A directory at the front of PATH, made once per test.""" + if self.bin_dir is None: + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.bin_dir = Path(tmp.name) + path = os.environ["PATH"] + self.addCleanup(os.environ.__setitem__, "PATH", path) + os.environ["PATH"] = f"{self.bin_dir}:{path}" + return self.bin_dir + + def fake_script(self, name: str, body: str) -> None: + script = self.fake_bin() / name + script.write_text(f"#!/bin/sh\n{body}\n") + script.chmod(0o755) + + def fake_command(self, name: str, output: bytes) -> None: + """A command printing exactly these bytes, undecodable ones included.""" + payload = self.fake_bin() / f"{name}.out" + payload.write_bytes(output) + self.fake_script(name, f'cat "{payload}"') + + def fake_findmnt(self, source: bytes) -> None: + """A findmnt answering per column, so the phase gets as far as SOURCE.""" + payload = self.fake_bin() / "findmnt.source" + payload.write_bytes(source) + self.fake_script( + "findmnt", + 'case "$2" in\n' + " FSTYPE) echo btrfs ;;\n" + " OPTIONS) echo rw,subvol=/@ ;;\n" + f' SOURCE) cat "{payload}" ;;\n' + "esac", + ) + + def test_undecodable_bytes_are_replaced_rather_than_raised(self): + self.fake_command("omarchy-not-utf8", b"before\x80after\n") + self.assertEqual(capture(["omarchy-not-utf8"]).stdout, "before�after\n") + + def test_efibootmgr_entries_parse_around_a_legacy_bbs_entry(self): + self.fake_command("efibootmgr", EFIBOOTMGR_OUTPUT) + + state = phases_impl._read_efibootmgr() + + self.assertEqual(state["order"], ["2001", "0003", "0000"]) + self.assertEqual(phases_impl._find_label_entries(state["entries"], "Omarchy"), ["0003"]) + self.assertTrue(state["entries"]["0000"].startswith("Notebook Hard Drive")) + + def test_a_mangled_identifier_stops_the_install_rather_than_reaching_fstab(self): + # _validate_pre_mounted_filesystems reads the UUID back from this same + # command, so a replaced byte would agree with itself all the way to a + # machine that cannot find its root. + self.fake_command("blkid", b"1a2b3c4d-\x80-dead-beef\n") + + with self.assertRaises(RuntimeError) as raised: + phases_impl._blkid_uuid("/dev/sda2") + + self.assertIn("the UUID of /dev/sda2 is not valid text", str(raised.exception)) + + def test_a_mangled_btrfs_device_is_never_handed_to_mount(self): + # create_factory_snapshot mounts this device and deletes subvolumes + # under it; a replaced byte must stop the phase before that, not turn + # into an opaque mount failure. + self.fake_findmnt(source=b"/dev/sda\xe02[/@]") + ctx = types.SimpleNamespace(target=Path("/mnt"), state_dir=Path("/run/omarchy")) + + with mock.patch.object(phases_impl, "info"), self.assertRaises(RuntimeError) as raised: + phases_impl.create_factory_snapshot(ctx) + + self.assertIn("is not valid text", str(raised.exception)) + + def test_a_clean_identifier_is_returned_unchanged(self): + self.fake_command("blkid", b"1a2b3c4d-0e5f-dead-beef\n") + self.assertEqual( + capture_identifier(["blkid"], "the UUID"), "1a2b3c4d-0e5f-dead-beef" + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/unit/test_keyboard.py b/test/unit/test_keyboard.py index 3c36b35..16c3c41 100644 --- a/test/unit/test_keyboard.py +++ b/test/unit/test_keyboard.py @@ -1,18 +1,16 @@ #!/usr/bin/python -import importlib.util import re +import sys import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] -MODULE_PATH = ROOT / "configs/airootfs/usr/share/omarchy-iso/orchestrator/keyboard.py" -SPEC = importlib.util.spec_from_file_location("installer_keyboard", MODULE_PATH) -KEYBOARD = importlib.util.module_from_spec(SPEC) -assert SPEC.loader -SPEC.loader.exec_module(KEYBOARD) +sys.path.insert(0, str(ROOT / "configs/airootfs/usr/share/omarchy-iso")) + +from orchestrator import keyboard as KEYBOARD # noqa: E402 # The layout list lives in the Omarchy runtime now, shared verbatim with the # first-boot owner setup; build-iso.sh vendors it onto the ISO. Read it from