From 87dcb369db5862083142fca39e705879fd5d1f9a Mon Sep 17 00:00:00 2001 From: Daniel Maksymilian Syrnicki Date: Mon, 28 Sep 2026 12:53:24 +0200 Subject: [PATCH 1/7] chore(website): make logo-assets.py pass ruff (unused var, E501 noqa, format) Co-Authored-By: Claude Opus 5.5 --- website/tools/logo-assets.py | 23 ++++++++++++++++------- 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/website/tools/logo-assets.py b/website/tools/logo-assets.py index 0797638..66a98c7 100644 --- a/website/tools/logo-assets.py +++ b/website/tools/logo-assets.py @@ -1,4 +1,5 @@ #!/usr/bin/env python3 +# ruff: noqa: E501 — inline SVG markup reads better unwrapped """Generate the logo-poll showcase assets for /logo/ from one geometry source. python3 website/tools/logo-assets.py @@ -10,6 +11,7 @@ the output — the site build itself needs nothing beyond Hugo. Mark M: lowercase f knocked out of a gate, 64-unit grid (even-odd fill). Mark R: window, wheel and F, 104x114 grid, stroke 8 (from Robert's sketch). """ + import base64 import io import json @@ -119,7 +121,9 @@ def raster(svgtxt, w, h): def mark_png(k, tier, height, color=None): _, _, w, h = MARKS[k]["vb"] - return raster(svg(k, tier, color or MARKS[k]["color"], standalone=True), round(height * w / h), height) + return raster( + svg(k, tier, color or MARKS[k]["color"], standalone=True), round(height * w / h), height + ) # ---------------------------------------------------------------- wordmark @@ -141,7 +145,9 @@ def wordmark(text, size, x0=0.0, baseline=0.0, tracking=-0.028): g.draw(pen) d = pen.getCommands() if d: - parts.append(f'') + parts.append( + f'' + ) x += g.width * s + tracking * size return "".join(parts), x - x0 - tracking * size @@ -241,21 +247,24 @@ def vote_card(path): m = mark_png(k, "flat", 330) im.alpha_composite(m, (cx - m.width // 2, 110 - (m.height - 330) // 2)) size = 46 - d = ImageDraw.Draw(im) for text, cx, col in (("M", W // 4, "#6b6b6f"), ("R", 3 * W // 4, "#8a8a90")): paths, tw = wordmark(text, size) # letters via the same outlines: rasterise a tiny SVG for crisp text - svgtxt = (f'' - f'{paths}') + svgtxt = ( + f'' + f'{paths}' + ) t = raster(svgtxt, round(tw), round(size * 1.3)) im.alpha_composite(t, (cx - t.width // 2, 500)) # url line, centred on the seam paths, tw = wordmark("furtka.org/logo", 40) - svgtxt = (f'{paths}') + svgtxt = f'{paths}' t = raster(svgtxt, round(tw), 52) pad = 18 box = Image.new("RGBA", (t.width + 2 * pad, t.height + 2 * pad), "#f7f6f3") - ImageDraw.Draw(box).rounded_rectangle([0, 0, box.width - 1, box.height - 1], 14, outline="#e4e3dc") + ImageDraw.Draw(box).rounded_rectangle( + [0, 0, box.width - 1, box.height - 1], 14, outline="#e4e3dc" + ) box.alpha_composite(t, (pad, pad)) im.alpha_composite(box, (W // 2 - box.width // 2, 40)) im.convert("RGB").save(path, optimize=True) -- 2.45.3 From 8bcbaec6215953645e25292fc6d6a7ce36041ad2 Mon Sep 17 00:00:00 2001 From: Daniel Maksymilian Syrnicki Date: Mon, 28 Sep 2026 12:53:24 +0200 Subject: [PATCH 2/7] 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 --- tests/test_app.py | 105 ++++++++ tests/test_drives.py | 244 +++++++++++------ webinstaller/app.py | 82 +++++- webinstaller/drives.py | 270 ++++++++++++------- webinstaller/static/style.css | 8 + webinstaller/templates/install/overview.html | 3 +- webinstaller/templates/install/step2.html | 28 +- 7 files changed, 542 insertions(+), 198 deletions(-) diff --git a/tests/test_app.py b/tests/test_app.py index b44c173..cc0172a 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -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) assert "next-steps" 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" diff --git a/tests/test_drives.py b/tests/test_drives.py index 3af53c5..5d8501b 100644 --- a/tests/test_drives.py +++ b/tests/test_drives.py @@ -1,47 +1,63 @@ +import json + +import drives +import pytest from drives import ( - get_drive_type_label, + MIN_DISK_GB, + _boot_device_from_cmdline, + classify_drive, + format_size, get_drive_type_score, get_size_score, - parse_lsblk_output, - parse_size_gb, - score_device, + parse_lsblk_json, ) - -def test_parse_size_gb_terabytes(): - assert parse_size_gb("1T") == 1024.0 +GB = 1024**3 -def test_parse_size_gb_gigabytes(): - assert parse_size_gb("500G") == 500.0 +def _lsblk(*entries): + """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(): - assert parse_size_gb("2048M") == 2.0 +@pytest.fixture(autouse=True) +def _smart_passed(monkeypatch): + monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") -def test_parse_size_gb_european_comma_decimal(): - assert parse_size_gb("1,5T") == 1.5 * 1024 +def _names(devices): + return [d["name"] for d in devices] -def test_parse_size_gb_empty_returns_none(): - assert parse_size_gb("") is None +def test_classify_uses_transport_and_rotation(): + 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(): - assert parse_size_gb("500K") is None - - -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_type_scores_rank_internal_over_usb(): + 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_size_score_bands(): @@ -51,67 +67,145 @@ def test_size_score_bands(): assert get_size_score(1024) == 10 -def test_score_device_sums_type_and_size(monkeypatch): - import drives - - monkeypatch.setattr(drives, "get_drive_health", lambda _: 10) - assert score_device("/dev/nvme0n1", 1024) == 15 + 10 + 10 - assert score_device("/dev/sda", 64) == 5 + 10 + 5 +def test_format_size(): + assert format_size(500 * GB) == "500 GB" + assert format_size(1024 * GB) == "1 TB" + assert format_size(int(1.5 * 1024 * GB)) == "1.5 TB" + assert format_size(int(7.5 * GB)) == "7.5 GB" -def test_parse_lsblk_drops_loop_and_rom(monkeypatch): - import drives - - monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") - output = "loop0 2.5G loop\nsr0 1024M rom\nsda 500G disk\nnvme0n1 1T disk\n" - devices = parse_lsblk_output(output) - names = [d["name"] for d in devices] - assert names == ["/dev/nvme0n1", "/dev/sda"] +def test_drops_loop_rom_zram_and_emmc_boot_areas(): + out = _lsblk( + ("loop0", 2.5, "loop"), + ("sr0", 1, "rom"), + ("zram0", 4, "disk"), + ("mmcblk0boot0", 0.004, "disk"), + ("mmcblk0boot1", 0.004, "disk"), + ("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): - import drives - - monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") - output = "nvme0n1 1T disk\n" - [dev] = parse_lsblk_output(output) - assert dev["type_label"] == "NVMe" - assert dev["health_label"] == "Healthy" +def test_drops_empty_card_reader_slots_and_read_only_disks(): + out = _lsblk( + ("sdc", 0, "disk", {"rm": True, "tran": "usb"}), + ("sdd", 500, "disk", {"ro": True}), + ("sda", 500, "disk"), + ) + assert _names(parse_lsblk_json(out)) == ["/dev/sda"] -def test_parse_lsblk_surfaces_smart_warning(monkeypatch): - import drives +def test_ssd_beats_bigger_hdd(): + # 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") - [dev] = parse_lsblk_output("sda 500G disk\n") + [dev] = parse_lsblk_json(_lsblk(("sda", 500, "disk"))) assert dev["health_label"] == "SMART warning" -def test_drive_type_label_nvme_ssd_hdd(): - assert get_drive_type_label("/dev/nvme0n1") == "NVMe" - assert get_drive_type_label("/dev/ssd0") == "SSD" - assert get_drive_type_label("/dev/sda") == "HDD" +def test_accepts_old_util_linux_string_flags(): + out = json.dumps( + { + "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(): - assert parse_lsblk_output("") == [] +def test_handles_empty_and_garbage_output(): + assert parse_lsblk_json("") == [] + assert parse_lsblk_json("not json") == [] -def test_parse_lsblk_drops_boot_usb(monkeypatch): - import drives - - monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") - output = "sda 500G disk\nsdb 16G disk\nnvme0n1 1T disk\n" - devices = parse_lsblk_output(output, boot_disk="sdb") - names = [d["name"] for d in devices] - assert "/dev/sdb" not in names - assert names == ["/dev/nvme0n1", "/dev/sda"] +def test_drops_boot_usb(): + out = _lsblk( + ("sda", 500, "disk"), + ("sdb", 16, "disk", {"tran": "usb"}), + ("nvme0n1", 1024, "disk", {"tran": "nvme", "rota": False}), + ) + assert _names(parse_lsblk_json(out, boot_disk="sdb")) == ["/dev/nvme0n1", "/dev/sda"] -def test_parse_lsblk_no_boot_disk_keeps_all(monkeypatch): - import drives +def test_boot_device_from_cmdline(monkeypatch): + calls = [] - monkeypatch.setattr(drives, "_smart_status", lambda _: "passed") - output = "sda 500G disk\nsdb 16G disk\n" - names = [d["name"] for d in parse_lsblk_output(output, boot_disk=None)] - assert set(names) == {"/dev/sda", "/dev/sdb"} + class R: + returncode = 0 + stdout = "/dev/sdb1\n" + + 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" diff --git a/webinstaller/app.py b/webinstaller/app.py index c0db38f..6658f3f 100644 --- a/webinstaller/app.py +++ b/webinstaller/app.py @@ -8,6 +8,8 @@ import os import re import subprocess import sys +import threading +import traceback from datetime import UTC 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 phase = "Starting up…" for marker, pct, label in PROGRESS_PHASES: @@ -118,7 +126,7 @@ def parse_install_progress(log): if percent >= 100: 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" phase = "Installation failed — open Show details below" for marker, hint in PROGRESS_ERROR_HINTS: @@ -127,6 +135,13 @@ def parse_install_progress(log): break else: 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} @@ -596,15 +611,20 @@ def install_step_1(): @app.route("/install/step2", methods=["GET", "POST"]) def install_step_2(): + drives = list_scored_devices() + error = "" if request.method == "POST": 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 return redirect(url_for("install_overview")) + error = "Please pick one of the disks below." return render_template( "install/step2.html", - drives=list_scored_devices(), + drives=drives, selected=settings.get("boot_drive", ""), + error=error, ) @@ -614,12 +634,50 @@ def install_overview(): 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"]) def install_run(): if not settings["boot_drive"] or not settings["username"] or not settings["password"]: 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"") + 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": INSTALL_LOG.write_text( 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). with INSTALL_LOG.open("a") as fh: 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")) @app.route("/install/log") def install_log_view(): - log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else "" + log, code = _read_install_log() return render_template( "install/log.html", log=log, - progress=parse_install_progress(log), + progress=parse_install_progress(log, code), hostname=settings["hostname"], ) @app.route("/install/log.json") def install_log_json(): - log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else "" - return jsonify(log=log, progress=parse_install_progress(log)) + log, code = _read_install_log() + return jsonify(log=log, progress=parse_install_progress(log, code)) @app.route("/install/reboot", methods=["POST"]) def install_reboot(): # Only allow rebooting once the install has actually finished — we don't # want a panicked click during install to reboot mid-pacstrap. - log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else "" - if parse_install_progress(log)["status"] != "done": + log, code = _read_install_log() + if parse_install_progress(log, code)["status"] != "done": return redirect(url_for("install_log_view")) # 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. diff --git a/webinstaller/drives.py b/webinstaller/drives.py index 66a5686..da9a422 100644 --- a/webinstaller/drives.py +++ b/webinstaller/drives.py @@ -1,56 +1,90 @@ +import json +import re 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(): """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, - leaving the device list untouched. On bare metal booted from USB this is - the stick we booted from — we want to filter it out so the user can't - accidentally pick it as the install target. + On bare metal booted from USB this is the stick we booted from — we want + to filter it out so the user can't accidentally pick it as the install + target. Primary source is the mount behind /run/archiso/bootmnt; when + that's gone (copytoram unmounts it) fall back to the UUID/label archiso + put on the kernel command line. """ - try: - result = subprocess.run( - ["findmnt", "-no", "SOURCE", "/run/archiso/bootmnt"], - capture_output=True, - text=True, - ) - except FileNotFoundError: + 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: + partition = _boot_device_from_cmdline(Path("/proc/cmdline").read_text()) + except OSError: + partition = None + if partition is None: return None - if result.returncode != 0: - return None - partition = result.stdout.strip() - 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 + # A stick written with dd has the ISO filesystem on the whole disk, where + # PKNAME is empty — the device itself is then the disk. + return _parent_disk(partition) or Path(partition).name def _smart_status(device): - try: - result = subprocess.run( - ["smartctl", "-H", device], - capture_output=True, - ) - output = result.stdout.decode() - if "PASSED" in output: - return "passed" - elif "FAILED" in output: - return "failed" - return "unknown" - except Exception as e: - print(f"Error checking SMART status for {device}: {e}") + result = _run(["smartctl", "-H", device]) + if result is None: return "unknown" + if "PASSED" in result.stdout: + return "passed" + if "FAILED" in result.stdout: + return "failed" + return "unknown" _HEALTH_SCORE = {"passed": 10, "failed": 0, "unknown": 5} @@ -60,40 +94,33 @@ _HEALTH_LABEL = { "unknown": "Status unknown", } +_TYPE_SCORE = {"NVMe": 15, "SSD": 10, "eMMC": 8, "HDD": 5, "USB": 0} + def get_drive_health(device): return _HEALTH_SCORE[_smart_status(device)] -def get_drive_type_score(device): - name = device.lower() - if "nvme" in name: - return 15 - if "ssd" in name: - return 10 - return 5 +def classify_drive(name, tran="", rota=None, removable=False): + """Human label for a disk: NVMe, SSD, eMMC, HDD or USB. - -def get_drive_type_label(device): - name = device.lower() - if "nvme" in name: + Uses lsblk's transport and rotational flags — the kernel name alone + can't tell an SSD from a spinning disk (both are sdX). + """ + 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" - if "ssd" in name: + if tran == "mmc" or name.startswith("mmcblk"): + return "eMMC" + if rota is False: return "SSD" return "HDD" -def parse_size_gb(size_str): - size_str = size_str.strip().upper().replace(",", ".") - 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_drive_type_score(type_label): + return _TYPE_SCORE.get(type_label, 5) def get_size_score(size_gb): @@ -106,61 +133,98 @@ def get_size_score(size_gb): return 10 -def score_device(device, size_gb): - return get_drive_type_score(device) + get_drive_health(device) + get_size_score(size_gb) +def format_size(size_bytes): + 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): - """Parse `lsblk -dn -o NAME,SIZE,TYPE` output into scored device dicts. +def _flag(value): + # 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, - that disk is also dropped — it's the USB stick the live ISO booted from - on bare metal, where it appears as TYPE=disk and would otherwise be a - valid-looking install target. + +def parse_lsblk_json(output, boot_disk=None, uefi=True): + """Parse `lsblk -J -d -b -o NAME,SIZE,TYPE,RO,RM,TRAN,ROTA` into device dicts. + + 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 = [] - for line in output.strip().split("\n"): - if not line: + for e in entries: + name = e.get("name") or "" + if e.get("type") != "disk" or not name: continue - parts = line.split() - if len(parts) < 3: + if _PSEUDO_DISK_RE.match(name) or (boot_disk and name == boot_disk): continue - name, size, dev_type = parts[0], parts[1], parts[2] - if dev_type != "disk": - continue - if boot_disk and name == boot_disk: + try: + size_bytes = int(e.get("size") or 0) + except (TypeError, ValueError): + size_bytes = 0 + if size_bytes <= 0 or _flag(e.get("ro", False)): 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}" - size_gb = parse_size_gb(size) - status = _smart_status(device) - score = get_drive_type_score(device) + _HEALTH_SCORE[status] + get_size_score(size_gb) + usable = size_gb >= MIN_DISK_GB + status = _smart_status(device) if usable else "unknown" + 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( { "name": device, - "size": size, - "type_label": get_drive_type_label(device), + "size": format_size(size_bytes), + "size_gb": size_gb, + "type_label": type_label, "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 def list_scored_devices(): - """Return [{name, size, score}, ...] for all physical disks, highest score first.""" - try: - result = subprocess.run( - ["lsblk", "-dn", "-o", "NAME,SIZE,TYPE"], - capture_output=True, - text=True, - check=True, - ) - except subprocess.CalledProcessError as e: - print(f"Error listing devices: {e}") + """Return [{name, size, score, usable, ...}, ...] for all physical disks, best first.""" + result = _run(["lsblk", "-J", "-d", "-b", "-o", "NAME,SIZE,TYPE,RO,RM,TRAN,ROTA"]) + if result is None or result.returncode != 0: + print(f"Error listing devices: {result.stderr if result else 'lsblk unavailable'}") 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(): @@ -168,11 +232,11 @@ def main(): if not devices: print("No storage devices found.") return - print(f"\n{'Device':<20} {'Size':<10} {'Score'}") - print("-" * 40) + print(f"\n{'Device':<20} {'Size':<10} {'Type':<6} {'Score'}") + print("-" * 46) for d in devices: - print(f"{d['name']:<20} {d['size']:<10} {d['score']}") - print(f"\nBest drive for boot: {devices[0]['name']} (score {devices[0]['score']})") + note = "" if d["usable"] else f" ({d['reason']})" + print(f"{d['name']:<20} {d['size']:<10} {d['type_label']:<6} {d['score']}{note}") if __name__ == "__main__": diff --git a/webinstaller/static/style.css b/webinstaller/static/style.css index b667596..c6d4674 100644 --- a/webinstaller/static/style.css +++ b/webinstaller/static/style.css @@ -190,9 +190,17 @@ select:focus { background: var(--bg-card); border: 1px solid var(--border); border-radius: 8px; + flex-wrap: wrap; cursor: pointer; 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 input[type="radio"] { accent-color: var(--accent); diff --git a/webinstaller/templates/install/overview.html b/webinstaller/templates/install/overview.html index 9878d5f..95ea3c0 100644 --- a/webinstaller/templates/install/overview.html +++ b/webinstaller/templates/install/overview.html @@ -22,7 +22,8 @@ -
+
← Start over diff --git a/webinstaller/templates/install/step2.html b/webinstaller/templates/install/step2.html index 0309e02..726f424 100644 --- a/webinstaller/templates/install/step2.html +++ b/webinstaller/templates/install/step2.html @@ -7,33 +7,47 @@

Boot drive

Pick the disk Furtka will install onto. We pre-select the drive that looks fastest and healthiest.

-{% 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 %} +
{{ error }}
+{% endif %} + +{% if not usable %}
- 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 %}
{% endif %}
{% for d in drives %} -
- + ← Back
-- 2.45.3 From 20f6398abf7d195066721411ec3a04890c21dcb7 Mon Sep 17 00:00:00 2001 From: Daniel Maksymilian Syrnicki Date: Mon, 28 Sep 2026 13:04:09 +0200 Subject: [PATCH 3/7] fix(api): refuse path-like app names, validate settings before writing .env, harden the HTTP server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - app names from URL paths and JSON bodies must match APP_NAME_RE; installer remove()/update_env() check it too. Before, {"name": ".."} or "/" sent remove() into shutil.rmtree on /var/lib/furtka or the root filesystem - manifests must carry a valid app name as well - settings are validated in memory (placeholders, system paths, line breaks) before .env is written — a rejected /etc used to be saved anyway - values containing $ are single-quoted so compose doesn't expand them ("pa$word" became "pa"); line breaks are refused (env injection) - ThreadingHTTPServer so status polls keep answering during slow requests; state-changing POSTs still run one at a time - 1 MiB body cap, 30 s socket timeout, bad Content-Length → 400 - POSTs with Sec-Fetch-Site other than same-origin are refused (apps on other ports of the same host count as same-site for SameSite=Strict) - unexpected exceptions return a JSON 500 instead of a dropped connection - UI pages send the browser to /login on a 401 instead of showing an empty app list after an API restart Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Wq1hzUp4ZmCKHDuLKCnu73 --- assets/www/index.html | 9 ++++ assets/www/settings/index.html | 9 ++++ furtka/api.py | 86 +++++++++++++++++++++++++++++--- furtka/installer.py | 82 ++++++++++++++++++++----------- furtka/manifest.py | 6 ++- tests/test_api.py | 90 +++++++++++++++++++++++++++++++--- tests/test_installer.py | 55 +++++++++++++++++++++ 7 files changed, 293 insertions(+), 44 deletions(-) diff --git a/assets/www/index.html b/assets/www/index.html index 66e5e2c..86c1559 100644 --- a/assets/www/index.html +++ b/assets/www/index.html @@ -68,6 +68,15 @@ {% endblock %} diff --git a/webinstaller/templates/install/step2.html b/webinstaller/templates/install/step2.html index 726f424..a82df57 100644 --- a/webinstaller/templates/install/step2.html +++ b/webinstaller/templates/install/step2.html @@ -1,21 +1,21 @@ {% extends "base.html" %} -{% block title %}Step 2 — Boot drive · Furtka Installer{% endblock %} +{% block title %}Step 2 — Disk · Furtka Installer{% endblock %} {% block step_indicator %}Step 2 of 3{% endblock %} {% block content %} -

Boot drive

-

Pick the disk Furtka will install onto. We pre-select the drive that looks fastest and healthiest.

+

Pick a disk

+

Choose the disk Furtka will be installed on. Everything on it will be erased. We pre-select the one that looks fastest and healthiest.

{% 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 %} -
{{ error }}
+ {% endif %} {% if not usable %} -
+