Stop the install dying on non-UTF-8 command output (#118)

* 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 <noreply@openai.com>

* 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 <noreply@openai.com>

---------

Co-authored-by: Codex XHigh <noreply@openai.com>

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)

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}")

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
 

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()

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