David Heinemeier HanssonandClaude Fable 5
8aee09cb29
Codex xhigh round 2: - Strip account authentication fields (!users, !root-password, root_enc_password, auth_config account entries) from the archinstall config in OEM mode, not just user_credentials — a combined/legacy config could otherwise have archinstall create a provisioning account before OOBE. - Clear stale OEM inputs (oem marker, credentials, optional files) from /root before copying a cidata drive, so a retried install in the same live session can't force a fresh drive back into OEM mode or reuse an old LUKS passphrase. - The test harness waits for the OEM pending flag to clear before saving the base image, so it never snapshots a half-finalized first boot; it also handles the deliberately skipped mode picker on blank disks built against a pre-OEM runtime. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
441 lines
17 KiB
Python
441 lines
17 KiB
Python
"""Unit tests for OEM install support in the orchestrator.
|
|
|
|
Covers the InstallContext OEM plumbing (marker file, credential-less input,
|
|
throwaway LUKS passphrase injection) and the stage_oem_state /
|
|
create_factory_snapshot / configure_login phases against a temp target with
|
|
subprocess mocked out.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from subprocess import CompletedProcess
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "configs/airootfs/usr/share/omarchy-iso"))
|
|
|
|
sys.modules.setdefault(
|
|
"orchestrator.archinstall_adapter", types.ModuleType("orchestrator.archinstall_adapter")
|
|
)
|
|
|
|
from orchestrator import phases_impl # noqa: E402
|
|
from orchestrator.context import InstallContext # noqa: E402
|
|
|
|
|
|
class ContextOemTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.dir = Path(self.tmp.name)
|
|
self.state_dir = self.dir / "state"
|
|
|
|
self.env = {
|
|
"OMARCHY_INSTALL_CONFIG": str(self.dir / "user_configuration.json"),
|
|
"OMARCHY_INSTALL_CREDS": str(self.dir / "user_credentials.json"),
|
|
"OMARCHY_INSTALL_STATE_DIR": str(self.state_dir),
|
|
}
|
|
|
|
def write_config(self, config):
|
|
(self.dir / "user_configuration.json").write_text(json.dumps(config))
|
|
|
|
def write_creds(self, creds):
|
|
(self.dir / "user_credentials.json").write_text(json.dumps(creds))
|
|
|
|
def from_env(self, **extra_env):
|
|
env = {**self.env, **extra_env}
|
|
with mock.patch.dict(os.environ, env, clear=False):
|
|
for key in ("OMARCHY_INSTALL_OEM_FILE",):
|
|
if key not in env:
|
|
os.environ.pop(key, None)
|
|
return InstallContext.from_env()
|
|
|
|
def base_config(self, oem=None, disk_encryption=None):
|
|
config = {
|
|
"disk_config": {"config_type": "default_layout"},
|
|
"omarchy_install": {"mode": "full_disk", "target_mount": "/mnt"},
|
|
}
|
|
if oem is not None:
|
|
config["omarchy_install"]["oem"] = oem
|
|
if disk_encryption is not None:
|
|
config["disk_config"]["disk_encryption"] = disk_encryption
|
|
return config
|
|
|
|
def test_oem_flag_from_config(self):
|
|
self.write_config(self.base_config(oem=True))
|
|
ctx = self.from_env()
|
|
self.assertTrue(ctx.oem)
|
|
self.assertEqual(ctx.user_credentials, {"users": []})
|
|
self.assertEqual(ctx.username, "")
|
|
|
|
def test_oem_marker_file_arms_oem_without_credentials(self):
|
|
self.write_config(self.base_config())
|
|
marker = self.dir / "oem"
|
|
marker.touch()
|
|
ctx = self.from_env(OMARCHY_INSTALL_OEM_FILE=str(marker))
|
|
self.assertTrue(ctx.oem)
|
|
self.assertTrue(ctx.omarchy_install["oem"])
|
|
|
|
def test_absent_marker_file_is_not_oem(self):
|
|
self.write_config(self.base_config())
|
|
self.write_creds({"users": [{"username": "jeff"}]})
|
|
ctx = self.from_env(OMARCHY_INSTALL_OEM_FILE=str(self.dir / "missing-oem"))
|
|
self.assertFalse(ctx.oem)
|
|
self.assertEqual(ctx.username, "jeff")
|
|
|
|
def test_missing_credentials_without_oem_raises(self):
|
|
self.write_config(self.base_config())
|
|
with self.assertRaisesRegex(RuntimeError, "credentials file missing"):
|
|
self.from_env()
|
|
|
|
def test_oem_generates_throwaway_luks_passphrase(self):
|
|
self.write_config(self.base_config(oem=True, disk_encryption={
|
|
"encryption_type": "luks", "partitions": ["x"],
|
|
}))
|
|
ctx = self.from_env()
|
|
|
|
password = ctx.user_configuration["disk_config"]["disk_encryption"]["encryption_password"]
|
|
self.assertTrue(password)
|
|
self.assertEqual(ctx.user_credentials["encryption_password"], password)
|
|
|
|
# archinstall reads both files from disk; the injected passphrase must
|
|
# be present in each.
|
|
arch_config = json.loads(ctx.arch_config_path.read_text())
|
|
self.assertEqual(
|
|
arch_config["disk_config"]["disk_encryption"]["encryption_password"], password
|
|
)
|
|
synthesized = json.loads(ctx.creds_path.read_text())
|
|
self.assertEqual(synthesized["encryption_password"], password)
|
|
|
|
def test_oem_reuses_rig_supplied_passphrase(self):
|
|
self.write_config(self.base_config(oem=True, disk_encryption={
|
|
"encryption_type": "luks", "partitions": ["x"],
|
|
}))
|
|
self.write_creds({"users": [], "encryption_password": "rig-secret"})
|
|
ctx = self.from_env()
|
|
self.assertEqual(
|
|
ctx.user_configuration["disk_config"]["disk_encryption"]["encryption_password"],
|
|
"rig-secret",
|
|
)
|
|
|
|
def test_oem_discards_account_material_from_supplied_credentials(self):
|
|
# A rig credentials file must not smuggle in users or a root password.
|
|
self.write_config(self.base_config(oem=True))
|
|
self.write_creds({
|
|
"users": [{"username": "backdoor", "enc_password": "x", "sudo": True}],
|
|
"root_enc_password": "x",
|
|
"encryption_password": "rig-secret",
|
|
})
|
|
ctx = self.from_env()
|
|
self.assertEqual(ctx.user_credentials["users"], [])
|
|
self.assertNotIn("root_enc_password", ctx.user_credentials)
|
|
self.assertEqual(ctx.user_credentials["encryption_password"], "rig-secret")
|
|
synthesized = json.loads(ctx.creds_path.read_text())
|
|
self.assertEqual(synthesized["users"], [])
|
|
self.assertNotIn("root_enc_password", synthesized)
|
|
|
|
def test_oem_strips_account_fields_from_arch_config(self):
|
|
config = self.base_config(oem=True)
|
|
config["!users"] = [{"username": "rig"}]
|
|
config["!root-password"] = "hunter2"
|
|
config["root_enc_password"] = "x"
|
|
config["auth_config"] = {"users": [{"username": "rig"}], "root_enc_password": "x"}
|
|
self.write_config(config)
|
|
ctx = self.from_env()
|
|
|
|
arch_config = json.loads(ctx.arch_config_path.read_text())
|
|
for key in ("!users", "!root-password", "root_enc_password", "users"):
|
|
self.assertNotIn(key, arch_config)
|
|
self.assertEqual(arch_config["auth_config"], {})
|
|
|
|
def test_oem_unencrypted_touches_nothing(self):
|
|
self.write_config(self.base_config(oem=True))
|
|
ctx = self.from_env()
|
|
self.assertNotIn("encryption_password", ctx.user_credentials)
|
|
|
|
|
|
def make_ctx(target, **overrides):
|
|
defaults = dict(
|
|
target=target,
|
|
oem=True,
|
|
encrypt=False,
|
|
username="",
|
|
omarchy_install={"mode": "full_disk", "oem": True, "storage": {}},
|
|
user_configuration={"disk_config": {}},
|
|
user_credentials={"users": []},
|
|
state_dir=target / "state",
|
|
authorized_keys_path=None,
|
|
)
|
|
defaults.update(overrides)
|
|
return types.SimpleNamespace(**defaults)
|
|
|
|
|
|
class StageOemStateTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.target = Path(self.tmp.name) / "mnt"
|
|
self.target.mkdir()
|
|
|
|
info_patch = mock.patch.object(phases_impl, "info")
|
|
info_patch.start()
|
|
self.addCleanup(info_patch.stop)
|
|
|
|
# A fake bundled Node tarball on the "live ISO".
|
|
self.packages = Path(self.tmp.name) / "opt-packages"
|
|
self.packages.mkdir()
|
|
(self.packages / "node-v24.0.0-linux-x64.tar.gz").write_bytes(b"node")
|
|
node_patch = mock.patch.object(phases_impl, "NODE_PACKAGES_DIR", self.packages)
|
|
node_patch.start()
|
|
self.addCleanup(node_patch.stop)
|
|
|
|
def install_runtime_oem_support(self):
|
|
service = self.target / "usr/share/omarchy/install/oem/omarchy-oem-setup.service"
|
|
service.parent.mkdir(parents=True, exist_ok=True)
|
|
service.write_text("[Unit]\n")
|
|
setup_bin = self.target / "usr/bin/omarchy-oem-setup"
|
|
setup_bin.parent.mkdir(parents=True, exist_ok=True)
|
|
setup_bin.write_text("#!/bin/bash\n")
|
|
|
|
def oem_dir(self):
|
|
return self.target / "var/lib/omarchy/oem"
|
|
|
|
def test_non_oem_install_stages_only_the_node_tarball(self):
|
|
ctx = make_ctx(self.target, oem=False)
|
|
phases_impl.stage_oem_state(ctx)
|
|
|
|
self.assertTrue((self.oem_dir() / "packages/node-v24.0.0-linux-x64.tar.gz").exists())
|
|
self.assertFalse((self.oem_dir() / "pending").exists())
|
|
self.assertFalse((self.target / "etc/systemd/system/omarchy-oem-setup.service").exists())
|
|
|
|
def test_oem_against_runtime_without_support_fails_clearly(self):
|
|
ctx = make_ctx(self.target)
|
|
with self.assertRaisesRegex(RuntimeError, "does not ship"):
|
|
phases_impl.stage_oem_state(ctx)
|
|
|
|
def test_oem_without_node_tarball_fails(self):
|
|
for tarball in self.packages.glob("*.tar.gz"):
|
|
tarball.unlink()
|
|
self.install_runtime_oem_support()
|
|
ctx = make_ctx(self.target)
|
|
with self.assertRaisesRegex(RuntimeError, "Node tarball"):
|
|
phases_impl.stage_oem_state(ctx)
|
|
|
|
def test_missing_node_tarball_fails_normal_installs_too(self):
|
|
# The stash is what makes a later factory reset work offline.
|
|
for tarball in self.packages.glob("*.tar.gz"):
|
|
tarball.unlink()
|
|
ctx = make_ctx(self.target, oem=False)
|
|
with self.assertRaisesRegex(RuntimeError, "Node tarball"):
|
|
phases_impl.stage_oem_state(ctx)
|
|
|
|
def test_oem_arms_first_boot_setup(self):
|
|
self.install_runtime_oem_support()
|
|
ctx = make_ctx(self.target)
|
|
phases_impl.stage_oem_state(ctx)
|
|
|
|
self.assertTrue((self.oem_dir() / "pending").exists())
|
|
self.assertTrue((self.target / "etc/systemd/system/omarchy-oem-setup.service").exists())
|
|
link = self.target / "etc/systemd/system/multi-user.target.wants/omarchy-oem-setup.service"
|
|
self.assertTrue(link.is_symlink())
|
|
self.assertEqual(
|
|
os.readlink(link), "/etc/systemd/system/omarchy-oem-setup.service"
|
|
)
|
|
# Unencrypted: no LUKS staging.
|
|
self.assertFalse((self.oem_dir() / "luks-key").exists())
|
|
|
|
def test_oem_encrypted_stages_luks_auto_unlock(self):
|
|
self.install_runtime_oem_support()
|
|
ctx = make_ctx(
|
|
self.target,
|
|
user_configuration={"disk_config": {"disk_encryption": {
|
|
"encryption_type": "luks",
|
|
"encryption_password": "throwaway-secret",
|
|
}}},
|
|
)
|
|
phases_impl.stage_oem_state(ctx)
|
|
|
|
# Byte-for-byte the slot passphrase — no trailing newline.
|
|
self.assertEqual((self.oem_dir() / "luks-key").read_text(), "throwaway-secret")
|
|
keyfile = self.target / "etc/omarchy/oem.key"
|
|
self.assertEqual(keyfile.read_text(), "throwaway-secret")
|
|
self.assertEqual(keyfile.stat().st_mode & 0o777, 0o600)
|
|
|
|
cmdline = (self.target / "etc/limine-entry-tool.d/99-omarchy-oem-unlock.conf").read_text()
|
|
self.assertIn("cryptkey=rootfs:/etc/omarchy/oem.key", cmdline)
|
|
files = (self.target / "etc/mkinitcpio.conf.d/99-omarchy-oem-key.conf").read_text()
|
|
self.assertIn("FILES+=(/etc/omarchy/oem.key)", files)
|
|
|
|
def test_oem_pre_encrypted_without_passphrase_fails(self):
|
|
self.install_runtime_oem_support()
|
|
ctx = make_ctx(
|
|
self.target,
|
|
omarchy_install={"mode": "protected", "oem": True, "storage": {"luks_uuid": "abc"}},
|
|
)
|
|
with self.assertRaisesRegex(RuntimeError, "passphrase"):
|
|
phases_impl.stage_oem_state(ctx)
|
|
|
|
|
|
class ConfigureLoginOemTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.target = Path(self.tmp.name)
|
|
self.calls = []
|
|
|
|
run_patch = mock.patch.object(
|
|
phases_impl.subprocess, "run",
|
|
side_effect=lambda cmd, **kw: (self.calls.append(cmd), CompletedProcess(cmd, 0))[1],
|
|
)
|
|
run_patch.start()
|
|
self.addCleanup(run_patch.stop)
|
|
|
|
def test_oem_login_leaves_user_state_to_first_boot(self):
|
|
ctx = make_ctx(self.target, encrypt=True)
|
|
phases_impl.configure_login(ctx)
|
|
|
|
self.assertTrue((self.target / "etc/sddm.conf.d/99-omarchy-login.conf").exists())
|
|
self.assertFalse((self.target / "etc/sddm.conf.d/autologin.conf").exists())
|
|
self.assertFalse((self.target / "var/lib/sddm/state.conf").exists())
|
|
self.assertTrue(any("sddm.service" in cmd for cmd in self.calls))
|
|
|
|
def test_normal_encrypted_login_still_autologs_in(self):
|
|
ctx = make_ctx(self.target, oem=False, encrypt=True, username="jeff")
|
|
phases_impl.configure_login(ctx)
|
|
|
|
autologin = (self.target / "etc/sddm.conf.d/autologin.conf").read_text()
|
|
self.assertIn("User=jeff", autologin)
|
|
state = (self.target / "var/lib/sddm/state.conf").read_text()
|
|
self.assertIn("User=jeff", state)
|
|
|
|
|
|
class ConfigureSshAccessOemTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.target = Path(self.tmp.name) / "mnt"
|
|
self.target.mkdir()
|
|
|
|
info_patch = mock.patch.object(phases_impl, "info")
|
|
info_patch.start()
|
|
self.addCleanup(info_patch.stop)
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
self.calls.append(cmd)
|
|
if "ufw" in cmd:
|
|
rules = self.target / "etc/ufw/user.rules"
|
|
rules.parent.mkdir(parents=True, exist_ok=True)
|
|
rules.write_text("-A ufw-user-input -p tcp --dport 22 -j ACCEPT\n")
|
|
return CompletedProcess(cmd, 0)
|
|
|
|
self.calls = []
|
|
run_patch = mock.patch.object(phases_impl.subprocess, "run", side_effect=fake_run)
|
|
run_patch.start()
|
|
self.addCleanup(run_patch.stop)
|
|
|
|
def test_oem_stages_keys_for_first_boot(self):
|
|
keys = Path(self.tmp.name) / "authorized_keys"
|
|
keys.write_text("ssh-ed25519 AAAA rig@host\n")
|
|
ctx = make_ctx(self.target, authorized_keys_path=keys)
|
|
|
|
phases_impl.configure_ssh_access(ctx)
|
|
|
|
staged = self.target / "var/lib/omarchy/oem/authorized_keys"
|
|
self.assertEqual(staged.read_text(), "ssh-ed25519 AAAA rig@host\n")
|
|
self.assertEqual(staged.stat().st_mode & 0o777, 0o600)
|
|
# No user yet — nothing under /home, no chown.
|
|
self.assertFalse((self.target / "home").exists())
|
|
self.assertFalse(any("chown" in cmd for cmd in self.calls))
|
|
# The door is still opened for first boot.
|
|
self.assertTrue(any("sshd.service" in cmd for cmd in self.calls))
|
|
|
|
|
|
class CreateFactorySnapshotTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.target = Path(self.tmp.name) / "mnt"
|
|
self.target.mkdir()
|
|
self.state_dir = Path(self.tmp.name) / "state"
|
|
self.state_dir.mkdir()
|
|
|
|
info_patch = mock.patch.object(phases_impl, "info")
|
|
info_patch.start()
|
|
self.addCleanup(info_patch.stop)
|
|
|
|
self.findmnt = {
|
|
"FSTYPE": "btrfs",
|
|
"OPTIONS": "rw,noatime,compress=zstd:3,subvol=/@",
|
|
"SOURCE": "/dev/mapper/omarchy_root[/@]",
|
|
}
|
|
self.calls = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
self.calls.append(cmd)
|
|
if cmd[0] == "findmnt":
|
|
return CompletedProcess(cmd, 0, stdout=self.findmnt[cmd[2]] + "\n", stderr="")
|
|
if cmd[0] == "mount":
|
|
# Simulate the top level containing @.
|
|
(Path(cmd[-1]) / "@").mkdir(parents=True, exist_ok=True)
|
|
return CompletedProcess(cmd, 0, stdout="", stderr="")
|
|
|
|
run_patch = mock.patch.object(phases_impl.subprocess, "run", side_effect=fake_run)
|
|
run_patch.start()
|
|
self.addCleanup(run_patch.stop)
|
|
|
|
def ctx(self):
|
|
return make_ctx(self.target, oem=False, state_dir=self.state_dir)
|
|
|
|
def test_snapshots_root_as_factory(self):
|
|
phases_impl.create_factory_snapshot(self.ctx())
|
|
|
|
top = str(self.state_dir / "factory-top")
|
|
self.assertIn(["mount", "-o", "subvolid=5", "/dev/mapper/omarchy_root", top], self.calls)
|
|
self.assertIn(
|
|
["btrfs", "subvolume", "snapshot", f"{top}/@", f"{top}/@factory"],
|
|
self.calls,
|
|
)
|
|
self.assertIn(
|
|
["btrfs", "property", "set", "-ts", f"{top}/@factory", "ro", "true"],
|
|
self.calls,
|
|
)
|
|
self.assertIn(["umount", top], self.calls)
|
|
|
|
def test_factory_snapshot_scrubs_provisioning_credentials(self):
|
|
# Simulate the snapshot carrying deployment credentials; the scrub
|
|
# must drop them before the snapshot goes read-only.
|
|
top = self.state_dir / "factory-top"
|
|
factory = top / "@factory"
|
|
secrets = [
|
|
factory / "var/lib/omarchy/oem/authorized_keys",
|
|
factory / "etc/tailscale/authkey",
|
|
factory / "etc/omarchy/oem.key",
|
|
]
|
|
keep = factory / "var/lib/omarchy/oem/groups"
|
|
for path in [*secrets, keep]:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("secret")
|
|
|
|
phases_impl.create_factory_snapshot(self.ctx())
|
|
|
|
for path in secrets:
|
|
self.assertFalse(path.exists(), path)
|
|
self.assertTrue(keep.exists())
|
|
|
|
def test_non_btrfs_target_skips(self):
|
|
self.findmnt["FSTYPE"] = "ext4"
|
|
phases_impl.create_factory_snapshot(self.ctx())
|
|
self.assertFalse(any(cmd[0] == "mount" for cmd in self.calls))
|
|
|
|
def test_non_subvol_root_skips(self):
|
|
self.findmnt["OPTIONS"] = "rw,noatime"
|
|
phases_impl.create_factory_snapshot(self.ctx())
|
|
self.assertFalse(any(cmd[0] == "mount" for cmd in self.calls))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|