fix(installer): safer disk picker, single install at a time, surface silent archinstall failures

- drives: classify by lsblk TRAN/ROTA (SATA SSDs were labelled HDD and lost
  to bigger spinning disks); drop zram, eMMC boot/rpmb areas, read-only and
  0-byte disks; grey out disks under 16 GB; rank USB disks last; warn about
  >2 TB disks on BIOS (MBR); boot-stick fallback via archiso kernel args
  when bootmnt is gone; timeouts on lsblk/smartctl/blkid
- step 2 only accepts a disk that was actually offered
- /install/run refuses a second concurrent install and disables the button
- archinstall's exit code is tracked: OOM kills / signals now show as errors
  instead of a progress bar stuck forever
- disk-probe exceptions land in the install log instead of a bare 500

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daniel Maksymilian Syrnicki 2026-09-28 12:53:24 +02:00
parent 87dcb369db
commit 8bcbaec621
7 changed files with 542 additions and 198 deletions

View file

@ -226,3 +226,108 @@ def test_log_page_running_state_has_no_restart_button(monkeypatch, tmp_path):
html = app_module.app.test_client().get("/install/log").get_data(as_text=True) html = app_module.app.test_client().get("/install/log").get_data(as_text=True)
assert "next-steps" not in html assert "next-steps" not in html
assert "install_reboot" not in html and "restart now" not in html assert "install_reboot" not in html and "restart now" not in html
def test_nonzero_exit_without_traceback_is_an_error():
import app as app_module
# OOM kill / signal: archinstall dies mid-pacstrap and prints nothing.
log = "Installing packages: ['base'\n"
assert app_module.parse_install_progress(log)["status"] == "running"
assert app_module.parse_install_progress(log, exit_code=-9)["status"] == "error"
assert app_module.parse_install_progress(log, exit_code=0)["status"] == "running"
class _FakeProc:
def __init__(self, code=None):
self.code = code
def poll(self):
return self.code
def _install_env(monkeypatch, tmp_path):
import app as app_module
monkeypatch.setattr(app_module, "STATE_DIR", tmp_path)
monkeypatch.setattr(app_module, "INSTALL_LOG", tmp_path / "install.log")
monkeypatch.setattr(app_module, "_install_proc", None)
monkeypatch.delenv("FURTKA_DRY_RUN", raising=False)
monkeypatch.setitem(app_module.settings, "boot_drive", "/dev/sda")
monkeypatch.setitem(app_module.settings, "username", "u")
monkeypatch.setitem(app_module.settings, "password", "pw12345678")
monkeypatch.setattr(app_module, "release_disk", lambda d: [f"released {d}"])
monkeypatch.setattr(app_module, "build_disk_config", lambda d: {"device": d})
return app_module
def test_install_run_refuses_second_concurrent_install(monkeypatch, tmp_path):
app_module = _install_env(monkeypatch, tmp_path)
spawned = []
monkeypatch.setattr(
app_module, "spawn_archinstall", lambda *a: spawned.append(a) or _FakeProc()
)
client = app_module.app.test_client()
assert client.post("/install/run").status_code == 302
assert client.post("/install/run").status_code == 302
assert len(spawned) == 1
def test_install_run_disk_probe_failure_lands_in_log(monkeypatch, tmp_path):
app_module = _install_env(monkeypatch, tmp_path)
def boom(_):
raise RuntimeError("parted could not read /dev/sda")
monkeypatch.setattr(app_module, "build_disk_config", boom)
resp = app_module.app.test_client().post("/install/run")
assert resp.status_code == 302
log = (tmp_path / "install.log").read_text()
assert "Traceback (most recent call last)" in log
assert app_module.parse_install_progress(log)["status"] == "error"
def test_install_log_json_reports_crashed_archinstall(monkeypatch, tmp_path):
app_module = _install_env(monkeypatch, tmp_path)
(tmp_path / "install.log").write_text("Installing packages: ['base'\n")
monkeypatch.setattr(app_module, "_install_proc", _FakeProc(code=137))
data = app_module.app.test_client().get("/install/log.json").get_json()
assert data["progress"]["status"] == "error"
assert "exit code 137" in data["log"]
def test_step2_rejects_disk_not_offered(monkeypatch):
import app as app_module
offered = [
{
"name": "/dev/sda",
"usable": True,
"size": "500 GB",
"type_label": "SSD",
"health_label": "Healthy",
"score": 27,
"reason": "",
"warning": "",
},
{
"name": "/dev/sdb",
"usable": False,
"size": "8 GB",
"type_label": "USB",
"health_label": "Status unknown",
"score": 10,
"reason": "Too small",
"warning": "",
},
]
monkeypatch.setattr(app_module, "list_scored_devices", lambda: offered)
monkeypatch.setitem(app_module.settings, "boot_drive", "")
client = app_module.app.test_client()
for bad in ("/dev/sdb", "/dev/nope", "/etc/passwd"):
resp = client.post("/install/step2", data={"boot_drive": bad})
assert resp.status_code == 200
assert app_module.settings["boot_drive"] == ""
resp = client.post("/install/step2", data={"boot_drive": "/dev/sda"})
assert resp.status_code == 302
assert app_module.settings["boot_drive"] == "/dev/sda"

View file

@ -1,47 +1,63 @@
import json
import drives
import pytest
from drives import ( from drives import (
get_drive_type_label, MIN_DISK_GB,
_boot_device_from_cmdline,
classify_drive,
format_size,
get_drive_type_score, get_drive_type_score,
get_size_score, get_size_score,
parse_lsblk_output, parse_lsblk_json,
parse_size_gb,
score_device,
) )
GB = 1024**3
def test_parse_size_gb_terabytes():
assert parse_size_gb("1T") == 1024.0
def test_parse_size_gb_gigabytes(): def _lsblk(*entries):
assert parse_size_gb("500G") == 500.0 """Build lsblk -J output. Each entry: (name, size_gb, type, extra-dict)."""
devs = []
for name, size_gb, dev_type, *extra in entries:
d = {
"name": name,
"size": int(size_gb * GB),
"type": dev_type,
"ro": False,
"rm": False,
"tran": "sata",
"rota": True,
}
if extra:
d.update(extra[0])
devs.append(d)
return json.dumps({"blockdevices": devs})
def test_parse_size_gb_megabytes(): @pytest.fixture(autouse=True)
assert parse_size_gb("2048M") == 2.0 def _smart_passed(monkeypatch):
monkeypatch.setattr(drives, "_smart_status", lambda _: "passed")
def test_parse_size_gb_european_comma_decimal(): def _names(devices):
assert parse_size_gb("1,5T") == 1.5 * 1024 return [d["name"] for d in devices]
def test_parse_size_gb_empty_returns_none(): def test_classify_uses_transport_and_rotation():
assert parse_size_gb("") is None assert classify_drive("nvme0n1", tran="nvme", rota=False) == "NVMe"
assert classify_drive("sda", tran="sata", rota=False) == "SSD"
assert classify_drive("sda", tran="sata", rota=True) == "HDD"
assert classify_drive("mmcblk0", tran="", rota=False) == "eMMC"
assert classify_drive("sdb", tran="usb", rota=False) == "USB"
assert classify_drive("sdc", tran="", rota=True, removable=True) == "USB"
# SD cards in a built-in reader are removable mmc — still eMMC/mmc class.
assert classify_drive("mmcblk1", tran="", rota=False, removable=True) == "eMMC"
def test_parse_size_gb_unknown_unit_returns_none(): def test_type_scores_rank_internal_over_usb():
assert parse_size_gb("500K") is None assert get_drive_type_score("NVMe") > get_drive_type_score("SSD")
assert get_drive_type_score("SSD") > get_drive_type_score("HDD")
assert get_drive_type_score("HDD") > get_drive_type_score("USB")
def test_drive_type_score_nvme():
assert get_drive_type_score("/dev/nvme0n1") == 15
def test_drive_type_score_ssd():
assert get_drive_type_score("/dev/ssd0") == 10
def test_drive_type_score_hdd_fallback():
assert get_drive_type_score("/dev/sda") == 5
def test_size_score_bands(): def test_size_score_bands():
@ -51,67 +67,145 @@ def test_size_score_bands():
assert get_size_score(1024) == 10 assert get_size_score(1024) == 10
def test_score_device_sums_type_and_size(monkeypatch): def test_format_size():
import drives assert format_size(500 * GB) == "500 GB"
assert format_size(1024 * GB) == "1 TB"
monkeypatch.setattr(drives, "get_drive_health", lambda _: 10) assert format_size(int(1.5 * 1024 * GB)) == "1.5 TB"
assert score_device("/dev/nvme0n1", 1024) == 15 + 10 + 10 assert format_size(int(7.5 * GB)) == "7.5 GB"
assert score_device("/dev/sda", 64) == 5 + 10 + 5
def test_parse_lsblk_drops_loop_and_rom(monkeypatch): def test_drops_loop_rom_zram_and_emmc_boot_areas():
import drives out = _lsblk(
("loop0", 2.5, "loop"),
monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") ("sr0", 1, "rom"),
output = "loop0 2.5G loop\nsr0 1024M rom\nsda 500G disk\nnvme0n1 1T disk\n" ("zram0", 4, "disk"),
devices = parse_lsblk_output(output) ("mmcblk0boot0", 0.004, "disk"),
names = [d["name"] for d in devices] ("mmcblk0boot1", 0.004, "disk"),
assert names == ["/dev/nvme0n1", "/dev/sda"] ("mmcblk0rpmb", 0.004, "disk"),
("mmcblk0", 64, "disk", {"tran": "", "rota": False}),
)
assert _names(parse_lsblk_json(out)) == ["/dev/mmcblk0"]
def test_parse_lsblk_attaches_human_labels(monkeypatch): def test_drops_empty_card_reader_slots_and_read_only_disks():
import drives out = _lsblk(
("sdc", 0, "disk", {"rm": True, "tran": "usb"}),
monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") ("sdd", 500, "disk", {"ro": True}),
output = "nvme0n1 1T disk\n" ("sda", 500, "disk"),
[dev] = parse_lsblk_output(output) )
assert dev["type_label"] == "NVMe" assert _names(parse_lsblk_json(out)) == ["/dev/sda"]
assert dev["health_label"] == "Healthy"
def test_parse_lsblk_surfaces_smart_warning(monkeypatch): def test_ssd_beats_bigger_hdd():
import drives # Regression: sdX SSDs used to be labelled HDD, so the 2 TB spinning disk
# won the recommendation over the 256 GB SSD.
out = _lsblk(
("sda", 2048, "disk", {"rota": True}),
("sdb", 256, "disk", {"rota": False}),
)
devices = parse_lsblk_json(out)
assert _names(devices) == ["/dev/sdb", "/dev/sda"]
assert [d["type_label"] for d in devices] == ["SSD", "HDD"]
def test_usb_disk_ranked_after_internal():
out = _lsblk(
("sda", 32, "disk", {"tran": "usb", "rota": False, "rm": True}),
("mmcblk0", 32, "disk", {"tran": "", "rota": False}),
)
devices = parse_lsblk_json(out)
assert _names(devices) == ["/dev/mmcblk0", "/dev/sda"]
assert devices[1]["warning"]
def test_too_small_disk_listed_but_unusable_and_last():
out = _lsblk(
("sda", MIN_DISK_GB - 1, "disk", {"rota": False}),
("sdb", 500, "disk"),
)
devices = parse_lsblk_json(out)
assert _names(devices) == ["/dev/sdb", "/dev/sda"]
assert devices[0]["usable"] is True
assert devices[1]["usable"] is False
assert str(MIN_DISK_GB) in devices[1]["reason"]
def test_bios_warns_about_disks_over_2tb():
out = _lsblk(("sda", 4096, "disk"))
[bios] = parse_lsblk_json(out, uefi=False)
[efi] = parse_lsblk_json(out, uefi=True)
assert "2 TB" in bios["warning"]
assert efi["warning"] == ""
def test_surfaces_smart_warning(monkeypatch):
monkeypatch.setattr(drives, "_smart_status", lambda _: "failed") monkeypatch.setattr(drives, "_smart_status", lambda _: "failed")
[dev] = parse_lsblk_output("sda 500G disk\n") [dev] = parse_lsblk_json(_lsblk(("sda", 500, "disk")))
assert dev["health_label"] == "SMART warning" assert dev["health_label"] == "SMART warning"
def test_drive_type_label_nvme_ssd_hdd(): def test_accepts_old_util_linux_string_flags():
assert get_drive_type_label("/dev/nvme0n1") == "NVMe" out = json.dumps(
assert get_drive_type_label("/dev/ssd0") == "SSD" {
assert get_drive_type_label("/dev/sda") == "HDD" "blockdevices": [
{
"name": "sda",
"size": str(500 * GB),
"type": "disk",
"ro": "0",
"rm": "0",
"tran": "sata",
"rota": "0",
},
]
}
)
[dev] = parse_lsblk_json(out)
assert dev["type_label"] == "SSD"
def test_parse_lsblk_handles_empty_output(): def test_handles_empty_and_garbage_output():
assert parse_lsblk_output("") == [] assert parse_lsblk_json("") == []
assert parse_lsblk_json("not json") == []
def test_parse_lsblk_drops_boot_usb(monkeypatch): def test_drops_boot_usb():
import drives out = _lsblk(
("sda", 500, "disk"),
monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") ("sdb", 16, "disk", {"tran": "usb"}),
output = "sda 500G disk\nsdb 16G disk\nnvme0n1 1T disk\n" ("nvme0n1", 1024, "disk", {"tran": "nvme", "rota": False}),
devices = parse_lsblk_output(output, boot_disk="sdb") )
names = [d["name"] for d in devices] assert _names(parse_lsblk_json(out, boot_disk="sdb")) == ["/dev/nvme0n1", "/dev/sda"]
assert "/dev/sdb" not in names
assert names == ["/dev/nvme0n1", "/dev/sda"]
def test_parse_lsblk_no_boot_disk_keeps_all(monkeypatch): def test_boot_device_from_cmdline(monkeypatch):
import drives calls = []
monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") class R:
output = "sda 500G disk\nsdb 16G disk\n" returncode = 0
names = [d["name"] for d in parse_lsblk_output(output, boot_disk=None)] stdout = "/dev/sdb1\n"
assert set(names) == {"/dev/sda", "/dev/sdb"}
monkeypatch.setattr(drives, "_run", lambda cmd, **_: calls.append(cmd) or R())
cmdline = "BOOT_IMAGE=/vmlinuz archisosearchuuid=2026-08-25-10-00-00-00 quiet"
assert _boot_device_from_cmdline(cmdline) == "/dev/sdb1"
assert calls[-1] == ["blkid", "-U", "2026-08-25-10-00-00-00"]
assert _boot_device_from_cmdline("archisolabel=FURTKA_2620") == "/dev/sdb1"
assert calls[-1] == ["blkid", "-L", "FURTKA_2620"]
assert _boot_device_from_cmdline("archisodevice=/dev/sr0") == "/dev/sr0"
assert _boot_device_from_cmdline("quiet splash") is None
def test_boot_disk_whole_disk_iso(monkeypatch):
# dd-written stick: ISO filesystem on the whole disk, PKNAME empty.
class R:
def __init__(self, out):
self.returncode = 0
self.stdout = out
def fake_run(cmd, **_):
if cmd[0] == "findmnt":
return R("/dev/sdb\n")
return R("\n")
monkeypatch.setattr(drives, "_run", fake_run)
assert drives._boot_disk_name() == "sdb"

View file

@ -8,6 +8,8 @@ import os
import re import re
import subprocess import subprocess
import sys import sys
import threading
import traceback
from datetime import UTC from datetime import UTC
from pathlib import Path from pathlib import Path
@ -108,7 +110,13 @@ PROGRESS_ERROR_HINTS = [
] ]
def parse_install_progress(log): def parse_install_progress(log, exit_code=None):
"""Progress dict from the install log.
`exit_code` is archinstall's return code once the process has ended
(None while running or unknown). A non-zero exit is an error even when
the log has no Traceback — OOM kills and signals print nothing.
"""
percent = 2 percent = 2
phase = "Starting up…" phase = "Starting up…"
for marker, pct, label in PROGRESS_PHASES: for marker, pct, label in PROGRESS_PHASES:
@ -118,7 +126,7 @@ def parse_install_progress(log):
if percent >= 100: if percent >= 100:
status = "done" status = "done"
elif any(m in log for m in PROGRESS_ERROR_MARKERS): elif any(m in log for m in PROGRESS_ERROR_MARKERS) or exit_code not in (None, 0):
status = "error" status = "error"
phase = "Installation failed — open Show details below" phase = "Installation failed — open Show details below"
for marker, hint in PROGRESS_ERROR_HINTS: for marker, hint in PROGRESS_ERROR_HINTS:
@ -127,6 +135,13 @@ def parse_install_progress(log):
break break
else: else:
status = "running" status = "running"
if percent == 18:
# archinstall waits for NTP time sync with no timeout. Only a hint:
# it usually clears on its own within a minute.
phase = (
"Syncing time and packages — if this takes more than a few "
"minutes, check that the network cable is plugged in"
)
return {"percent": percent, "phase": phase, "status": status} return {"percent": percent, "phase": phase, "status": status}
@ -596,15 +611,20 @@ def install_step_1():
@app.route("/install/step2", methods=["GET", "POST"]) @app.route("/install/step2", methods=["GET", "POST"])
def install_step_2(): def install_step_2():
drives = list_scored_devices()
error = ""
if request.method == "POST": if request.method == "POST":
boot_drive = request.form.get("boot_drive", "").strip() boot_drive = request.form.get("boot_drive", "").strip()
if boot_drive: # Only accept a disk we actually offered — never a free-form path.
if any(d["name"] == boot_drive and d["usable"] for d in drives):
settings["boot_drive"] = boot_drive settings["boot_drive"] = boot_drive
return redirect(url_for("install_overview")) return redirect(url_for("install_overview"))
error = "Please pick one of the disks below."
return render_template( return render_template(
"install/step2.html", "install/step2.html",
drives=list_scored_devices(), drives=drives,
selected=settings.get("boot_drive", ""), selected=settings.get("boot_drive", ""),
error=error,
) )
@ -614,12 +634,50 @@ def install_overview():
return render_template("install/overview.html", settings=masked) return render_template("install/overview.html", settings=masked)
# The running archinstall process, so a second click (double-submit, Back +
# Install again, a second browser) can't start another install on the same
# disk, and so the progress page can see an exit without a Traceback.
_install_proc = None
_install_lock = threading.Lock()
def _install_exit_code():
return None if _install_proc is None else _install_proc.poll()
def _read_install_log():
log = INSTALL_LOG.read_text(errors="replace") if INSTALL_LOG.exists() else ""
code = _install_exit_code()
if code not in (None, 0):
log += f"\n\narchinstall stopped unexpectedly (exit code {code}).\n"
return log, code
@app.route("/install/run", methods=["POST"]) @app.route("/install/run", methods=["POST"])
def install_run(): def install_run():
if not settings["boot_drive"] or not settings["username"] or not settings["password"]: if not settings["boot_drive"] or not settings["username"] or not settings["password"]:
return redirect(url_for("install_step_1")) return redirect(url_for("install_step_1"))
config_path, creds_path = write_install_files(settings, STATE_DIR) with _install_lock:
if _install_proc is not None and _install_proc.poll() is None:
return redirect(url_for("install_log_view"))
return _start_install()
def _start_install():
global _install_proc
STATE_DIR.mkdir(parents=True, exist_ok=True)
INSTALL_LOG.write_bytes(b"") INSTALL_LOG.write_bytes(b"")
try:
config_path, creds_path = write_install_files(settings, STATE_DIR)
except Exception:
# archinstall's disk probe raises on disks it can't parse (broken
# GPT, odd card readers). Put the traceback where the progress page
# finds it instead of a bare "Internal Server Error".
with INSTALL_LOG.open("a") as fh:
fh.write(
f"Could not read the layout of {settings['boot_drive']}.\n" + traceback.format_exc()
)
return redirect(url_for("install_log_view"))
if os.environ.get("FURTKA_DRY_RUN") == "1": if os.environ.get("FURTKA_DRY_RUN") == "1":
INSTALL_LOG.write_text( INSTALL_LOG.write_text(
f"DRY RUN: would exec archinstall --config {config_path} " f"DRY RUN: would exec archinstall --config {config_path} "
@ -631,33 +689,33 @@ def install_run():
# while those hold the partitions open (first hardware bench, 26.19). # while those hold the partitions open (first hardware bench, 26.19).
with INSTALL_LOG.open("a") as fh: with INSTALL_LOG.open("a") as fh:
fh.write("\n".join(release_disk(settings["boot_drive"])) + "\n\n") fh.write("\n".join(release_disk(settings["boot_drive"])) + "\n\n")
spawn_archinstall(config_path, creds_path, INSTALL_LOG) _install_proc = spawn_archinstall(config_path, creds_path, INSTALL_LOG)
return redirect(url_for("install_log_view")) return redirect(url_for("install_log_view"))
@app.route("/install/log") @app.route("/install/log")
def install_log_view(): def install_log_view():
log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else "" log, code = _read_install_log()
return render_template( return render_template(
"install/log.html", "install/log.html",
log=log, log=log,
progress=parse_install_progress(log), progress=parse_install_progress(log, code),
hostname=settings["hostname"], hostname=settings["hostname"],
) )
@app.route("/install/log.json") @app.route("/install/log.json")
def install_log_json(): def install_log_json():
log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else "" log, code = _read_install_log()
return jsonify(log=log, progress=parse_install_progress(log)) return jsonify(log=log, progress=parse_install_progress(log, code))
@app.route("/install/reboot", methods=["POST"]) @app.route("/install/reboot", methods=["POST"])
def install_reboot(): def install_reboot():
# Only allow rebooting once the install has actually finished — we don't # Only allow rebooting once the install has actually finished — we don't
# want a panicked click during install to reboot mid-pacstrap. # want a panicked click during install to reboot mid-pacstrap.
log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else "" log, code = _read_install_log()
if parse_install_progress(log)["status"] != "done": if parse_install_progress(log, code)["status"] != "done":
return redirect(url_for("install_log_view")) return redirect(url_for("install_log_view"))
# Delay reboot a few seconds so the browser can finish fetching CSS / assets # Delay reboot a few seconds so the browser can finish fetching CSS / assets
# for the rebooting page before the Flask server (and network) go away. # for the rebooting page before the Flask server (and network) go away.

View file

@ -1,55 +1,89 @@
import json
import re
import subprocess import subprocess
from pathlib import Path
# Smallest disk we offer as an install target. The base system plus Docker
# needs ~6 GB; the rest is headroom for images and app data.
MIN_DISK_GB = 16
# MBR (what archinstall lays out on BIOS boxes) can't address past 2 TiB.
MBR_LIMIT_GB = 2048
# Block devices lsblk reports as TYPE=disk that are never install targets:
# compressed RAM swap and the eMMC hardware boot / RPMB areas (4 MB, usually
# read-only) that show up next to mmcblk0 on every eMMC mini-PC and laptop.
_PSEUDO_DISK_RE = re.compile(r"^(zram\d+|mmcblk\d+boot\d+|mmcblk\d+rpmb)$")
_SUBPROCESS_TIMEOUT = 15
def _run(cmd, timeout=_SUBPROCESS_TIMEOUT):
"""subprocess.run with a timeout; None when the tool is missing or hangs."""
try:
return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
def _parent_disk(device):
result = _run(["lsblk", "-no", "PKNAME", device])
if result is None or result.returncode != 0:
return None
lines = [line.strip() for line in result.stdout.splitlines() if line.strip()]
return lines[0] if lines else None
def _boot_device_from_cmdline(cmdline):
"""Resolve the boot partition from archiso's kernel args, or None."""
args = dict(arg.split("=", 1) for arg in cmdline.split() if "=" in arg)
if args.get("archisodevice", "").startswith("/dev/"):
return args["archisodevice"]
if args.get("archisosearchuuid"):
cmd = ["blkid", "-U", args["archisosearchuuid"]]
elif args.get("archisolabel"):
cmd = ["blkid", "-L", args["archisolabel"]]
else:
return None
result = _run(cmd)
if result is None or result.returncode != 0:
return None
return result.stdout.strip() or None
def _boot_disk_name(): def _boot_disk_name():
"""Return the parent disk name of the live-ISO boot media (e.g. "sdb"), or None. """Return the parent disk name of the live-ISO boot media (e.g. "sdb"), or None.
On a normal box `/run/archiso/bootmnt` does not exist and we return None, On bare metal booted from USB this is the stick we booted from — we want
leaving the device list untouched. On bare metal booted from USB this is to filter it out so the user can't accidentally pick it as the install
the stick we booted from — we want to filter it out so the user can't target. Primary source is the mount behind /run/archiso/bootmnt; when
accidentally pick it as the install target. that's gone (copytoram unmounts it) fall back to the UUID/label archiso
put on the kernel command line.
""" """
partition = None
result = _run(["findmnt", "-no", "SOURCE", "/run/archiso/bootmnt"])
if result is not None and result.returncode == 0:
partition = result.stdout.strip() or None
if partition is None:
try: try:
result = subprocess.run( partition = _boot_device_from_cmdline(Path("/proc/cmdline").read_text())
["findmnt", "-no", "SOURCE", "/run/archiso/bootmnt"], except OSError:
capture_output=True, partition = None
text=True, if partition is None:
)
except FileNotFoundError:
return None return None
if result.returncode != 0: # A stick written with dd has the ISO filesystem on the whole disk, where
return None # PKNAME is empty — the device itself is then the disk.
partition = result.stdout.strip() return _parent_disk(partition) or Path(partition).name
if not partition:
return None
try:
parent = subprocess.run(
["lsblk", "-no", "PKNAME", partition],
capture_output=True,
text=True,
)
except FileNotFoundError:
return None
if parent.returncode != 0:
return None
name = parent.stdout.strip().splitlines()[0] if parent.stdout.strip() else ""
return name or None
def _smart_status(device): def _smart_status(device):
try: result = _run(["smartctl", "-H", device])
result = subprocess.run( if result is None:
["smartctl", "-H", device],
capture_output=True,
)
output = result.stdout.decode()
if "PASSED" in output:
return "passed"
elif "FAILED" in output:
return "failed"
return "unknown" return "unknown"
except Exception as e: if "PASSED" in result.stdout:
print(f"Error checking SMART status for {device}: {e}") return "passed"
if "FAILED" in result.stdout:
return "failed"
return "unknown" return "unknown"
@ -60,40 +94,33 @@ _HEALTH_LABEL = {
"unknown": "Status unknown", "unknown": "Status unknown",
} }
_TYPE_SCORE = {"NVMe": 15, "SSD": 10, "eMMC": 8, "HDD": 5, "USB": 0}
def get_drive_health(device): def get_drive_health(device):
return _HEALTH_SCORE[_smart_status(device)] return _HEALTH_SCORE[_smart_status(device)]
def get_drive_type_score(device): def classify_drive(name, tran="", rota=None, removable=False):
name = device.lower() """Human label for a disk: NVMe, SSD, eMMC, HDD or USB.
if "nvme" in name:
return 15
if "ssd" in name:
return 10
return 5
Uses lsblk's transport and rotational flags — the kernel name alone
def get_drive_type_label(device): can't tell an SSD from a spinning disk (both are sdX).
name = device.lower() """
if "nvme" in name: tran = (tran or "").lower()
if tran == "usb" or (removable and not name.startswith("mmcblk")):
return "USB"
if tran == "nvme" or name.startswith("nvme"):
return "NVMe" return "NVMe"
if "ssd" in name: if tran == "mmc" or name.startswith("mmcblk"):
return "eMMC"
if rota is False:
return "SSD" return "SSD"
return "HDD" return "HDD"
def parse_size_gb(size_str): def get_drive_type_score(type_label):
size_str = size_str.strip().upper().replace(",", ".") return _TYPE_SCORE.get(type_label, 5)
if not size_str:
return None
if size_str.endswith("T"):
return float(size_str[:-1]) * 1024
if size_str.endswith("G"):
return float(size_str[:-1])
if size_str.endswith("M"):
return float(size_str[:-1]) / 1024
return None
def get_size_score(size_gb): def get_size_score(size_gb):
@ -106,61 +133,98 @@ def get_size_score(size_gb):
return 10 return 10
def score_device(device, size_gb): def format_size(size_bytes):
return get_drive_type_score(device) + get_drive_health(device) + get_size_score(size_gb) gb = size_bytes / 1024**3
if gb >= 1024:
return f"{gb / 1024:.1f} TB".replace(".0 ", " ")
if gb >= 10:
return f"{gb:.0f} GB"
return f"{gb:.1f} GB"
def parse_lsblk_output(output, boot_disk=None): def _flag(value):
"""Parse `lsblk -dn -o NAME,SIZE,TYPE` output into scored device dicts. # util-linux ≥2.37 emits JSON booleans; older builds emit "0"/"1".
if isinstance(value, bool):
return value
return str(value).strip() in ("1", "true")
Keeps only TYPE=disk so the live ISO's own squashfs (loop) and the boot
CD-ROM (rom) don't show up as install targets. If `boot_disk` is given, def parse_lsblk_json(output, boot_disk=None, uefi=True):
that disk is also dropped — it's the USB stick the live ISO booted from """Parse `lsblk -J -d -b -o NAME,SIZE,TYPE,RO,RM,TRAN,ROTA` into device dicts.
on bare metal, where it appears as TYPE=disk and would otherwise be a
valid-looking install target. Keeps only real, writable, non-empty TYPE=disk devices (drops the live
ISO's loop/rom, zram, eMMC boot areas, empty card-reader slots) and the
USB stick we booted from. Disks too small to install on stay in the list
with `usable=False` and a reason so the user sees why they're greyed out.
Sorted best-first; unusable disks last.
""" """
try:
entries = json.loads(output or "{}").get("blockdevices", [])
except json.JSONDecodeError:
return []
devices = [] devices = []
for line in output.strip().split("\n"): for e in entries:
if not line: name = e.get("name") or ""
if e.get("type") != "disk" or not name:
continue continue
parts = line.split() if _PSEUDO_DISK_RE.match(name) or (boot_disk and name == boot_disk):
if len(parts) < 3:
continue continue
name, size, dev_type = parts[0], parts[1], parts[2] try:
if dev_type != "disk": size_bytes = int(e.get("size") or 0)
continue except (TypeError, ValueError):
if boot_disk and name == boot_disk: size_bytes = 0
if size_bytes <= 0 or _flag(e.get("ro", False)):
continue continue
size_gb = size_bytes / 1024**3
type_label = classify_drive(
name,
tran=e.get("tran") or "",
rota=_flag(e.get("rota")) if e.get("rota") is not None else None,
removable=_flag(e.get("rm", False)),
)
device = f"/dev/{name}" device = f"/dev/{name}"
size_gb = parse_size_gb(size) usable = size_gb >= MIN_DISK_GB
status = _smart_status(device) status = _smart_status(device) if usable else "unknown"
score = get_drive_type_score(device) + _HEALTH_SCORE[status] + get_size_score(size_gb) warning = ""
if not uefi and size_gb > MBR_LIMIT_GB:
warning = (
"This computer starts in legacy (BIOS) mode, which can only use "
"the first 2 TB of this disk. Switch the firmware to UEFI to use all of it."
)
elif type_label == "USB":
warning = (
"This is a USB drive — it works, but an internal disk is faster and more reliable."
)
devices.append( devices.append(
{ {
"name": device, "name": device,
"size": size, "size": format_size(size_bytes),
"type_label": get_drive_type_label(device), "size_gb": size_gb,
"type_label": type_label,
"health_label": _HEALTH_LABEL[status], "health_label": _HEALTH_LABEL[status],
"score": score, "score": get_drive_type_score(type_label)
+ _HEALTH_SCORE[status]
+ get_size_score(size_gb),
"usable": usable,
"reason": "" if usable else f"Too small — Furtka needs at least {MIN_DISK_GB} GB",
"warning": warning,
} }
) )
devices.sort(key=lambda d: d["score"], reverse=True) devices.sort(key=lambda d: (d["usable"], d["score"]), reverse=True)
return devices return devices
def list_scored_devices(): def list_scored_devices():
"""Return [{name, size, score}, ...] for all physical disks, highest score first.""" """Return [{name, size, score, usable, ...}, ...] for all physical disks, best first."""
try: result = _run(["lsblk", "-J", "-d", "-b", "-o", "NAME,SIZE,TYPE,RO,RM,TRAN,ROTA"])
result = subprocess.run( if result is None or result.returncode != 0:
["lsblk", "-dn", "-o", "NAME,SIZE,TYPE"], print(f"Error listing devices: {result.stderr if result else 'lsblk unavailable'}")
capture_output=True,
text=True,
check=True,
)
except subprocess.CalledProcessError as e:
print(f"Error listing devices: {e}")
return [] return []
return parse_lsblk_output(result.stdout, boot_disk=_boot_disk_name()) return parse_lsblk_json(
result.stdout,
boot_disk=_boot_disk_name(),
uefi=Path("/sys/firmware/efi").exists(),
)
def main(): def main():
@ -168,11 +232,11 @@ def main():
if not devices: if not devices:
print("No storage devices found.") print("No storage devices found.")
return return
print(f"\n{'Device':<20} {'Size':<10} {'Score'}") print(f"\n{'Device':<20} {'Size':<10} {'Type':<6} {'Score'}")
print("-" * 40) print("-" * 46)
for d in devices: for d in devices:
print(f"{d['name']:<20} {d['size']:<10} {d['score']}") note = "" if d["usable"] else f" ({d['reason']})"
print(f"\nBest drive for boot: {devices[0]['name']} (score {devices[0]['score']})") print(f"{d['name']:<20} {d['size']:<10} {d['type_label']:<6} {d['score']}{note}")
if __name__ == "__main__": if __name__ == "__main__":

View file

@ -190,9 +190,17 @@ select:focus {
background: var(--bg-card); background: var(--bg-card);
border: 1px solid var(--border); border: 1px solid var(--border);
border-radius: 8px; border-radius: 8px;
flex-wrap: wrap;
cursor: pointer; cursor: pointer;
transition: border-color 120ms, background 120ms; transition: border-color 120ms, background 120ms;
} }
.drive-disabled { cursor: not-allowed; opacity: 0.6; }
.drive-disabled:hover { border-color: var(--border); }
.drive .drive-note {
flex-basis: 100%;
font-size: 0.82rem;
color: var(--fg-muted);
}
.drive:hover { border-color: color-mix(in srgb, var(--accent) 50%, var(--border)); } .drive:hover { border-color: color-mix(in srgb, var(--accent) 50%, var(--border)); }
.drive input[type="radio"] { .drive input[type="radio"] {
accent-color: var(--accent); accent-color: var(--accent);

View file

@ -22,7 +22,8 @@
</table> </table>
</div> </div>
<form method="post" action="{{ url_for('install_run') }}"> <form method="post" action="{{ url_for('install_run') }}"
onsubmit="this.querySelector('button').disabled = true; this.querySelector('button').textContent = 'Starting…';">
<div class="actions"> <div class="actions">
<button type="submit" class="btn btn-danger">Install — wipe drive and proceed</button> <button type="submit" class="btn btn-danger">Install — wipe drive and proceed</button>
<a href="{{ url_for('install_step_1') }}" class="btn btn-link">← Start over</a> <a href="{{ url_for('install_step_1') }}" class="btn btn-link">← Start over</a>

View file

@ -7,33 +7,47 @@
<h1>Boot drive</h1> <h1>Boot drive</h1>
<p class="lede">Pick the disk Furtka will install onto. We pre-select the drive that looks fastest and healthiest.</p> <p class="lede">Pick the disk Furtka will install onto. We pre-select the drive that looks fastest and healthiest.</p>
{% if not drives %} {% set usable = drives | selectattr("usable") | list %}
{% set pick = selected if (selected and usable | selectattr("name", "equalto", selected) | list) else (usable[0].name if usable else "") %}
{% if error %}
<div class="alert alert-error">{{ error }}</div>
{% endif %}
{% if not usable %}
<div class="alert alert-error"> <div class="alert alert-error">
No physical disks detected. The live ISO can't see any install targets — check your BIOS / SATA controller. {% if drives %}
None of the disks in this computer is big enough for Furtka (it needs at least 16 GB).
{% else %}
No disk found to install onto. Check that the disk is connected, and in the BIOS/UEFI settings
that the SATA mode is "AHCI" rather than "RAID" or "Intel RST" — Linux can't see disks in RAID mode.
{% endif %}
</div> </div>
{% endif %} {% endif %}
<form method="post" action="{{ url_for('install_step_2') }}"> <form method="post" action="{{ url_for('install_step_2') }}">
<div class="drive-list"> <div class="drive-list">
{% for d in drives %} {% for d in drives %}
<label class="drive"> <label class="drive{% if not d.usable %} drive-disabled{% endif %}">
<input type="radio" name="boot_drive" value="{{ d.name }}" <input type="radio" name="boot_drive" value="{{ d.name }}"
{% if (selected and d.name == selected) or (not selected and loop.first) %}checked{% endif %} /> {% if not d.usable %}disabled{% elif d.name == pick %}checked{% endif %} />
<span class="name">{{ d.name }}</span> <span class="name">{{ d.name }}</span>
{% if not selected and loop.first %} {% if not selected and d.usable and loop.first %}
<span class="badge-recommended">Recommended</span> <span class="badge-recommended">Recommended</span>
{% endif %} {% endif %}
<span class="meta"> <span class="meta">
<span class="chip">{{ d.size }}</span> <span class="chip">{{ d.size }}</span>
<span class="chip">{{ d.type_label }}</span> <span class="chip">{{ d.type_label }}</span>
<span class="chip">{{ d.health_label }}</span> {% if d.usable %}<span class="chip">{{ d.health_label }}</span>{% endif %}
</span> </span>
{% if d.reason %}<span class="drive-note">{{ d.reason }}</span>{% endif %}
{% if d.warning %}<span class="drive-note">{{ d.warning }}</span>{% endif %}
</label> </label>
{% endfor %} {% endfor %}
</div> </div>
<div class="actions"> <div class="actions">
<button type="submit" class="btn btn-primary" {% if not drives %}disabled{% endif %}>Next — review</button> <button type="submit" class="btn btn-primary" {% if not usable %}disabled{% endif %}>Next — review</button>
<a href="{{ url_for('install_step_1') }}" class="btn btn-link">← Back</a> <a href="{{ url_for('install_step_1') }}" class="btn btn-link">← Back</a>
</div> </div>
</form> </form>