Menu
BifrOSt
publicLatest change 995290fa505c425e8021890af463ee44e0d76fae - Restore live networking and add WiFi setup by Ólafur Búi Ólafsson
#!/usr/bin/env python3
import hashlib
import json
import os
import re
import socket
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
MIN_DISK_BYTES = 20 * 1024**3
BACKEND = "/usr/local/lib/bifrost-installer-backend"
OFFLINE_ROOT = Path("/usr/share/bifrost/offline")
LIVE_MOUNTS = {"/", "/run/archiso/bootmnt", "/run/archiso/cowspace", "/run/archiso/airootfs"}
PROFILE_IDS = ("base", "dev-rust", "dev-containers", "dev-web", "dev-python")
KEYBOARD_LAYOUT = "is-latin1"
def human_size(value):
size = float(value)
for unit in ("B", "KB", "MB", "GB", "TB"):
if size < 1000 or unit == "TB":
return f"{size:.1f} {unit}"
size /= 1000
def _walk_device(device):
yield device
for child in device.get("children") or ():
yield from _walk_device(child)
def inventory_from_snapshot(snapshot, swap_paths=()):
"""Return physical whole disks with machine-readable rejection reasons."""
swaps = {os.path.realpath(path) for path in swap_paths}
inventory = []
for device in snapshot.get("blockdevices", ()):
if device.get("type") != "disk" or str(device.get("path", "")).startswith("/dev/zram"):
continue
reasons = []
serial = str(device.get("serial") or "").strip()
wwn = str(device.get("wwn") or "").strip()
if not (serial or wwn):
reasons.append({"code": "no_stable_id", "detail": ""})
descendants = list(_walk_device(device))
for node in descendants:
path = node.get("path") or node.get("name") or "?"
mounts = [mount for mount in (node.get("mountpoints") or ()) if mount]
if os.path.realpath(str(path)) in swaps or "[SWAP]" in mounts:
reasons.append({"code": "swap", "detail": str(path)})
for mount in mounts:
if mount == "[SWAP]":
continue
if mount in LIVE_MOUNTS or str(mount).startswith("/run/archiso/"):
reasons.append({"code": "live", "detail": f"{path} → {mount}"})
else:
reasons.append({"code": "mounted", "detail": f"{path} → {mount}"})
if bool(device.get("ro")):
reasons.append({"code": "readonly", "detail": ""})
if bool(device.get("rm")):
reasons.append({"code": "removable", "detail": ""})
size = int(device.get("size") or 0)
if size < MIN_DISK_BYTES:
reasons.append({"code": "small", "detail": human_size(MIN_DISK_BYTES)})
logical_sector = int(device.get("log-sec") or 0)
if logical_sector <= 0:
reasons.append({"code": "sector", "detail": ""})
unique = []
seen = set()
for reason in reasons:
marker = (reason["code"], reason["detail"])
if marker not in seen:
seen.add(marker)
unique.append(reason)
inventory.append({
"path": str(device.get("path") or ""),
"name": str(device.get("name") or ""),
"model": str(device.get("model") or "").strip(),
"serial": serial,
"wwn": wwn,
"size": size,
"logical_sector": logical_sector,
"physical_sector": int(device.get("phy-sec") or 0),
"reasons": unique,
"eligible": not unique,
})
return inventory
def read_inventory():
result = subprocess.run(
[
"lsblk", "--bytes", "--json", "--tree", "--output",
"NAME,PATH,SIZE,MODEL,SERIAL,WWN,TYPE,RO,RM,LOG-SEC,PHY-SEC,MOUNTPOINTS",
],
check=True, capture_output=True, text=True, timeout=15,
)
swaps = []
try:
for line in Path("/proc/swaps").read_text().splitlines()[1:]:
fields = line.split()
if fields:
swaps.append(fields[0])
except OSError:
pass
return inventory_from_snapshot(json.loads(result.stdout), swaps)
def target_fingerprint(target):
return (
target.get("path", ""), target.get("model", ""), target.get("serial", ""),
target.get("wwn", ""), int(target.get("size") or 0),
int(target.get("logical_sector") or 0),
)
def preserved_selection_index(disks, previous_fingerprint):
if previous_fingerprint is None:
return None
return next((index for index, disk in enumerate(disks)
if target_fingerprint(disk) == previous_fingerprint), None)
def valid_terminal_event(event):
if not isinstance(event, dict) or event.get("schema_version") != 2:
return False
kind = event.get("event")
live_log = event.get("live_log_dir")
target_log = event.get("target_log_dir")
log_dir = event.get("log_dir")
if not isinstance(live_log, str) or not live_log or not isinstance(log_dir, str) or not log_dir:
return False
if kind == "success":
return (event.get("wiped") is True
and isinstance(event.get("message"), str) and bool(event["message"])
and isinstance(target_log, str) and bool(target_log)
and log_dir == target_log)
return (kind in ("error", "failure")
and isinstance(event.get("code"), str) and bool(event["code"])
and isinstance(event.get("message"), str) and bool(event["message"])
and isinstance(event.get("wiped"), bool)
and target_log is None and log_dir == live_log)
def terminal_log_dir(event):
if not isinstance(event, dict):
return None
return event.get("target_log_dir") or event.get("live_log_dir") or event.get("log_dir")
def failure_recovery_key(event):
if not valid_terminal_event(event) or event.get("event") not in ("error", "failure"):
return "recovery_unknown"
if event.get("wiped") is True:
return "recovery_wiped"
if event.get("wiped") is False:
return "recovery_safe"
return "recovery_unknown"
ERROR_CODE_CATEGORIES = {
"invalid_invocation": "input", "invalid_input": "input", "insecure_input": "input",
"invalid_request": "input", "uefi_required": "environment",
"secure_boot_enabled": "environment", "missing_dependency": "environment",
"target_mount_busy": "environment", "target_refused": "target",
"target_changed": "target", "offline_source_unavailable": "source",
"offline_source_unverified": "source", "offline_source_incomplete": "source",
"online_source_unavailable": "source", "online_source_incomplete": "source",
"package_source_unverified": "source",
"archinstall_config_invalid": "plan", "archinstall_failed": "install",
"package_trust_failed": "install", "package_install_failed": "install",
"installed_assets_missing": "verify", "branding_missing": "verify",
"branding_invalid": "verify", "target_not_mounted": "verify",
"target_incomplete": "verify", "bootloader_missing": "verify",
"kernel_missing": "verify", "initramfs_missing": "verify",
"boot_entry_missing": "verify", "boot_entry_invalid": "verify",
"encryption_initramfs_invalid": "verify", "encryption_boot_invalid": "verify",
"encryption_layout_invalid": "verify", "filesystem_layout_invalid": "verify",
"locale_state_invalid": "verify", "evidence_persistence_failed": "verify",
"repository_config_missing": "verify", "repository_config_invalid": "verify",
"package_ownership_invalid": "verify", "ui_backend_start": "backend",
}
def error_category(code):
return ERROR_CODE_CATEGORIES.get(code, "backend" if code.startswith("backend_exit_")
else "unknown")
def validate_identity(username, hostname, user_password, confirmation,
encryption=False, encryption_password="", encryption_confirmation="",
full_name=""):
if not re.fullmatch(r"[a-z_][a-z0-9_-]{0,31}", username):
return "username"
if not re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9.-]{0,61}[A-Za-z0-9])?", hostname):
return "hostname"
if len(full_name) > 64 or any(character in full_name for character in ":\0\n\r"):
return "full_name"
if len(user_password) < 8:
return "password_short"
if user_password != confirmation:
return "password_mismatch"
if encryption and len(encryption_password) < 8:
return "encryption_short"
if encryption and encryption_password != encryption_confirmation:
return "encryption_mismatch"
return None
def _file_sha256(path):
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def source_readiness():
states = {"online": {"ready": False, "detail": "network"},
"offline": {"ready": False, "detail": "missing"}}
try:
with socket.create_connection(("geo.mirror.pkgbuild.com", 443), timeout=4):
states["online"] = {"ready": True, "detail": "reachable"}
except OSError:
pass
manifest_path = OFFLINE_ROOT / "manifest.json"
signature_path = OFFLINE_ROOT / "manifest.json.sig"
keyring_path = OFFLINE_ROOT / "trustedkeys.gpg"
repository = (OFFLINE_ROOT / "repo").resolve()
try:
manifest = json.loads(manifest_path.read_text())
packages = manifest.get("packages")
if (manifest.get("schema_version") != 2 or manifest.get("complete") is not True
or Path(manifest.get("repository_root", "")).resolve() != repository
or not isinstance(packages, list) or not packages
or not signature_path.is_file() or not keyring_path.is_file()):
raise ValueError("incomplete manifest")
subprocess.run(
["gpgv", "--keyring", str(keyring_path), str(signature_path), str(manifest_path)],
check=True, capture_output=True, text=True, timeout=30,
)
for package in packages:
if not all(isinstance(package.get(key), str) and package[key]
for key in ("name", "version", "path", "sha256")):
raise ValueError("invalid package record")
package_path = Path(package["path"])
if not package_path.is_absolute():
package_path = repository / package_path
package_path = package_path.resolve()
if repository not in package_path.parents or not package_path.is_file():
raise ValueError("package outside repository")
if not re.fullmatch(r"[0-9a-fA-F]{64}", package["sha256"]):
raise ValueError("invalid package hash")
if _file_sha256(package_path).lower() != package["sha256"].lower():
raise ValueError("package hash mismatch")
states["offline"] = {"ready": True, "detail": "verified"}
except (OSError, ValueError, KeyError, json.JSONDecodeError, subprocess.SubprocessError):
pass
return states
def _self_check():
fixture = {"blockdevices": [
{"name": "safe", "path": "/dev/sda", "type": "disk", "size": MIN_DISK_BYTES,
"model": "Safe", "serial": "A", "wwn": "W", "ro": False, "rm": False,
"log-sec": 512, "phy-sec": 4096, "mountpoints": []},
{"name": "anonymous", "path": "/dev/sdb", "type": "disk", "size": MIN_DISK_BYTES,
"model": "Anonymous", "serial": "", "wwn": "", "ro": False, "rm": False,
"log-sec": 512, "phy-sec": 4096, "mountpoints": []},
{"name": "live", "path": "/dev/sdc", "type": "disk", "size": MIN_DISK_BYTES,
"serial": "C", "ro": False, "rm": False, "log-sec": 512, "mountpoints": [],
"children": [{"name": "live1", "path": "/dev/sdc1", "type": "part",
"mountpoints": ["/"]}]},
{"name": "tiny", "path": "/dev/sdd", "type": "disk", "size": 1024,
"serial": "D", "ro": True, "rm": True, "log-sec": 512, "mountpoints": []},
]}
records = inventory_from_snapshot(fixture)
assert records[0]["eligible"]
assert not records[1]["eligible"]
assert "no_stable_id" in {reason["code"] for reason in records[1]["reasons"]}
assert {reason["code"] for reason in records[2]["reasons"]} == {"live"}
assert {"readonly", "removable", "small"} <= {
reason["code"] for reason in records[3]["reasons"]}
assert preserved_selection_index([records[0]], None) is None
assert preserved_selection_index([records[0]], target_fingerprint(records[0])) == 0
changed = dict(records[0], path="/dev/sdz")
assert preserved_selection_index([changed], target_fingerprint(records[0])) is None
assert KEYBOARD_LAYOUT == "is-latin1"
assert failure_recovery_key(None) == "recovery_unknown"
assert failure_recovery_key({"schema_version": 2, "event": "error"}) == "recovery_unknown"
failure = {
"schema_version": 2, "event": "error", "code": "target_refused",
"message": "refused", "wiped": False,
"log_dir": "/var/log/bifrost-installer/run",
"live_log_dir": "/var/log/bifrost-installer/run", "target_log_dir": None,
}
assert failure_recovery_key(failure) == "recovery_safe"
assert terminal_log_dir(failure) == failure["live_log_dir"]
success = {
"schema_version": 2, "event": "success", "message": "done", "wiped": True,
"log_dir": "/var/log/bifrost-installer/run",
"live_log_dir": "/var/log/bifrost-installer/live",
"target_log_dir": "/var/log/bifrost-installer/run",
}
assert valid_terminal_event(success)
assert terminal_log_dir(success) == success["target_log_dir"]
assert not valid_terminal_event({**success, "target_log_dir": None})
assert validate_identity("user", "bifrost", "password", "password") is None
assert validate_identity("Bad User", "bifrost", "password", "password") == "username"
assert validate_identity("user", "bifrost-", "password", "password") == "hostname"
assert validate_identity("user", "bifrost", "password", "password",
full_name="invalid:name") == "full_name"
print("installer self-check passed")
if __name__ == "__main__" and "--self-check" in sys.argv:
_self_check()
raise SystemExit(0)
import gi
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from gi.repository import Adw, Gdk, GLib, Gtk
CSS = b"""
window { background: #07111f; color: #eef7ff; }
.hero { background: #0a1728; padding: 14px 18px; }
.hero-title { font-size: 28px; font-weight: 800; }
.hero-subtitle, .step-copy { font-size: 14px; color: #b7cada; }
.step-title { font-size: 23px; font-weight: 750; }
.card { background: #0e1d30; border: 1px solid #29445d; border-radius: 14px; padding: 14px; }
.warning { background: #39231f; border: 1px solid #b6694f; border-radius: 10px; padding: 12px; color: #ffe2d7; }
.success { background: #133225; border: 1px solid #3e8060; border-radius: 10px; padding: 12px; color: #d8f8e7; }
.failure { background: #39231f; border: 1px solid #b6694f; border-radius: 10px; padding: 12px; color: #ffe2d7; }
.install-button { background: #e76f51; color: #08111e; font-weight: 800; }
.install-button:hover { background: #f58a6b; }
.accent { color: #72dcff; }
.rejected { color: #efb7a5; }
progressbar trough { min-height: 8px; background: #182c42; }
progressbar progress { background: #65d8ff; }
"""
STRINGS = {
"is": {
"title": "Setja upp BifrOSt", "tagline": "Íslenskt, frjálst og byggt fyrir þig.",
"step": "Skref {current} af 6", "back": "Til baka", "next": "Halda áfram",
"install": "Setja upp BifrOSt", "language": "Tungumál uppsetningar",
"welcome_title": "Velkomin í BifrOSt", "welcome_copy": "Uppsetningin notar allan valinn disk. Veldu tungumál og farðu yfir hvert skref áður en gögn eru eydd.",
"feature_locale": "Íslenskar sjálfgefnar stillingar", "feature_locale_copy": "Kerfistungumál, lyklaborð og tímabelti eru stillt fyrir Ísland.",
"feature_arch": "Hreinn Arch grunnur", "feature_arch_copy": "Opinberir pakkar, Arch kjarninn og venjulegar Arch uppfærslur.",
"disk_title": "Veldu heilan disk", "disk_copy": "Aðeins öruggir, ónotaðir diskar eru valanlegir. Endurskannaðu eftir að diskum er tengt eða aftengt.",
"rescan": "Endurskanna diska", "scanning": "Leita að diskum…", "no_disks": "Enginn hæfur uppsetningardiskur fannst.",
"scan_error": "Ekki tókst að lesa diska: {error}", "eligible_count": "{count} hæfur diskur fannst.",
"rejected_title": "Diskar sem ekki má nota", "none_rejected": "Engir útilokaðir diskar.",
"stable_id": "varanlegt auðkenni", "unavailable": "ekki tiltækt",
"reason_no_stable_id": "vantar raðnúmer og WWN; ekki er hægt að staðfesta eininguna örugglega",
"reason_live": "inniheldur ræst kerfi eða uppsetningarmiðil ({detail})", "reason_mounted": "inniheldur tengt skráarkerfi ({detail})",
"reason_swap": "notað sem swap ({detail})", "reason_readonly": "skrifvarinn", "reason_removable": "færanlegur miðill",
"reason_small": "minni en {detail}", "reason_sector": "óþekkt geirastærð",
"disk_warning": "Varúð: Öllum gögnum á valda disknum verður eytt. Afritaðu mikilvæg gögn.",
"identity_title": "Stofnaðu notanda", "identity_copy": "Notandinn fær stjórnandaréttindi. Lykilorðið er aðeins geymt í varinni tímabundinni skrá meðan uppsetning keyrir.",
"full_name": "Nafn", "username": "Notandanafn", "user_password": "Lykilorð", "confirm_password": "Staðfesta lykilorð", "hostname": "Heiti tölvu",
"wifi_title": "Tengjast neti", "wifi_copy": "Veldu Wi-Fi net áður en pakkar og þróunarsnið eru valin. Þú getur haldið áfram með Ethernet eða staðfestan ónettengdan pakkagjafa.",
"wifi_network": "Wi-Fi net", "wifi_password": "Wi-Fi lykilorð", "wifi_rescan": "Leita aftur", "wifi_connect": "Tengjast",
"wifi_scanning": "Leita að Wi-Fi netum…", "wifi_none": "Engin Wi-Fi net fundust. Athugaðu að kveikt sé á þráðlausa kortinu.",
"wifi_connected": "Tengt við {ssid}.", "wifi_failed": "Ekki tókst að tengjast: {error}", "wifi_open": "Opið net",
"options_title": "Öryggi, snið og pakkagjafi", "options_copy": "Grunnkerfið er alltaf sett upp. Veldu dulkóðun og aðeins þau þróunarsnið sem þú þarft.",
"encryption": "Dulkóða kerfisdisk með LUKS2", "encryption_password": "Dulkóðunarlykilorð", "confirm_encryption": "Staðfesta dulkóðunarlykilorð",
"profiles": "Valfrjáls þróunarsnið", "profile_base": "Grunnkerfi (alltaf)", "profile_dev-rust": "Rust þróun", "profile_dev-containers": "Gámar (Podman)", "profile_dev-web": "Vefþróun", "profile_dev-python": "Python þróun",
"source": "Pakkagjafi", "online": "Netið — opinberir Arch speglar", "offline": "Ónettengt — staðfestur staðbundinn pakkagjafi", "refresh_source": "Athuga pakkagjafa",
"source_checking": "Athuga net og staðbundna pakka…", "source_online_ready": "Netið er aðgengilegt. Bakendinn staðfestir pakkagjafa aftur áður en disknum er eytt.",
"source_online_unready": "Engin nettenging við opinberan Arch spegil. Tengdu net eða notaðu fullstaðfestan ónettengdan gjafa.",
"source_offline_ready": "Undirritaður heildstæður staðbundinn pakkagjafi er staðfestur.", "source_offline_unready": "Enginn fullstaðfestur staðbundinn pakkagjafi er tiltækur.",
"review_title": "Farðu yfir uppsetninguna", "review_copy": "Diskurinn er endurskannaður áður en lokastaðfesting birtist. Samþykki gildir aðeins fyrir nákvæmlega þessa auðkenndu einingu.",
"review_empty": "Veldu disk og ljúktu fyrri skrefum.", "summary": "Diskur\n{disk}\n\nNotandi\n{full_name} ({username})\n\nTölva\n{hostname}\n\nDulkóðun\n{encryption}\n\nSnið\n{profiles}\n\nPakkagjafi\n{source}",
"enabled": "LUKS2 virkt", "disabled": "Ekki dulkóðað", "confirm_title": "Eyða öllum gögnum á þessum diski?",
"confirm_body": "Þessi aðgerð eyðir endanlega öllum skiptingum og gögnum á:\n\n{disk}\n\nSláðu inn nákvæma slóð disksins til staðfestingar: {path}",
"typed_path": "Slóð disks", "cancel": "Hætta við", "erase_install": "Eyða disknum og setja upp",
"preflight": "Endurstaðfesti disk og pakkagjafa…", "target_changed": "Diskurinn er ekki lengur sá sami eða er ekki lengur öruggur. Veldu hann aftur.",
"source_changed": "Valinn pakkagjafi er ekki tilbúinn. Athugaðu tengingu eða staðbundinn miðil.",
"username_error": "Notandanafn má aðeins innihalda lágstafi, tölur, _ og -.", "hostname_error": "Ógilt tölvuheiti.", "full_name_error": "Nafn má ekki innihalda tvípunkt eða línuskil og má vera mest 64 stafir.",
"password_short": "Lykilorð þarf að vera að minnsta kosti 8 stafir.", "password_mismatch": "Lykilorðin passa ekki saman.",
"encryption_short": "Dulkóðunarlykilorð þarf að vera að minnsta kosti 8 stafir.", "encryption_mismatch": "Dulkóðunarlykilorðin passa ekki saman.",
"source_required": "Valinn pakkagjafi er ekki tilbúinn.", "disk_required": "Veldu hæfan disk.",
"progress_title": "Set BifrOSt upp", "progress_copy": "Bakendinn sannreynir auðkenni disks og pakkagjafa áður en nokkru er eytt.",
"progress_start": "Undirbý uppsetningu…", "phase_preflight": "Sannreyni disk og öryggisskilyrði…", "phase_source": "Sannreyni pakkagjafa…", "phase_plan": "Útbý örugga diskskipan…",
"phase_apply": "Set upp Arch grunnkerfið…", "phase_postinstall": "Set upp BifrOSt upplifun…", "phase_verify": "Sannreyni uppsett kerfi…", "phase_evidence": "Vista hreinsaða uppsetningarskrá varanlega…",
"success": "Uppsetningu lokið", "success_detail": "BifrOSt er tilbúið. Fjarlægðu uppsetningarmiðilinn og endurræstu tölvuna.",
"failure": "Uppsetning mistókst",
"failure_detail": "{summary}\nVillukóði: {code}\nTæknilegar upplýsingar (óþýddar): {message}",
"error_input": "Inntak uppsetningar er ógilt eða ótryggt.",
"error_environment": "Uppsetningarumhverfið uppfyllir ekki öryggiskröfur.",
"error_target": "Ekki var hægt að staðfesta valinn disk á öruggan hátt.",
"error_source": "Ekki var hægt að staðfesta allan pakkagjafann.",
"error_plan": "Örugg uppsetningaráætlun var ekki samþykkt.",
"error_install": "Uppsetningin stöðvaðist eftir að diskskrif gætu hafa hafist.",
"error_verify": "Staðfesting uppsetta kerfisins mistókst.",
"error_backend": "Samskipti við bakendann rofnuðu.",
"error_unknown": "Óþekkt uppsetningarvilla kom upp.",
"log_path": "Varanlegar uppsetningarskrár: {path}",
"recovery_wiped": "Staða disks: gæti verið breytt. Ekki er lofað afturköllun. Haltu uppsetningarkerfinu keyrandi og notaðu varðveittu skrána til greiningar eða enduruppsetningar.",
"recovery_safe": "Staða disks: óbreytt samkvæmt gildri lokatilkynningu bakendans. Þú getur leiðrétt vandann og reynt aftur.",
"recovery_unknown": "Staða disks: óþekkt. Lokatilkynning vantar eða er ógild, svo ekki er hægt að fullyrða hvort disknum hafi verið breytt. Haltu uppsetningarkerfinu keyrandi og varðveittu uppsetningarskrárnar til greiningar.",
"close_blocked": "Ekki er hægt að loka glugganum meðan uppsetning keyrir. Uppsetning heldur áfram; bíddu eftir lokaðurstöðu.",
"retry": "Fara aftur og endurskanna", "backend_protocol": "Bakendinn lauk án gildrar lokatilkynningar.",
},
"en": {
"title": "Install BifrOSt", "tagline": "Icelandic, free, and built for you.",
"step": "Step {current} of 6", "back": "Back", "next": "Continue", "install": "Install BifrOSt", "language": "Installer language",
"welcome_title": "Welcome to BifrOSt", "welcome_copy": "Installation uses the entire selected disk. Choose a language and review every step before data is erased.",
"feature_locale": "Icelandic system defaults", "feature_locale_copy": "System locale, keyboard, and time zone are configured for Iceland.",
"feature_arch": "A clean Arch base", "feature_arch_copy": "Official packages, the Arch kernel, and normal Arch updates.",
"disk_title": "Choose a whole disk", "disk_copy": "Only safe, unused disks can be selected. Rescan after connecting or disconnecting disks.",
"rescan": "Rescan disks", "scanning": "Scanning disks…", "no_disks": "No eligible installation disk was found.",
"scan_error": "Could not enumerate disks: {error}", "eligible_count": "Found {count} eligible disk(s).", "rejected_title": "Disks that cannot be used", "none_rejected": "No rejected disks.",
"stable_id": "persistent identifier", "unavailable": "unavailable",
"reason_no_stable_id": "has neither a serial number nor WWN; the device cannot be safely identified",
"reason_live": "contains the running system or install media ({detail})", "reason_mounted": "contains a mounted filesystem ({detail})",
"reason_swap": "is used as swap ({detail})", "reason_readonly": "is read-only", "reason_removable": "is removable media", "reason_small": "is smaller than {detail}", "reason_sector": "has an unknown sector size",
"disk_warning": "Warning: All data on the selected disk will be erased. Back up important data.",
"identity_title": "Create your user", "identity_copy": "The user receives administrator rights. The password is stored only in a protected temporary file while installation runs.",
"full_name": "Full name", "username": "Username", "user_password": "Password", "confirm_password": "Confirm password", "hostname": "Computer name",
"wifi_title": "Connect to a network", "wifi_copy": "Choose Wi-Fi before selecting packages and development profiles. You may continue with Ethernet or a verified offline package source.",
"wifi_network": "Wi-Fi network", "wifi_password": "Wi-Fi password", "wifi_rescan": "Rescan", "wifi_connect": "Connect",
"wifi_scanning": "Scanning for Wi-Fi networks…", "wifi_none": "No Wi-Fi networks were found. Check that the wireless adapter is enabled.",
"wifi_connected": "Connected to {ssid}.", "wifi_failed": "Could not connect: {error}", "wifi_open": "Open network",
"options_title": "Security, profiles, and package source", "options_copy": "The base system is always installed. Choose encryption and only the development profiles you need.",
"encryption": "Encrypt the system disk with LUKS2", "encryption_password": "Encryption passphrase", "confirm_encryption": "Confirm encryption passphrase",
"profiles": "Optional development profiles", "profile_base": "Base system (always)", "profile_dev-rust": "Rust development", "profile_dev-containers": "Containers (Podman)", "profile_dev-web": "Web development", "profile_dev-python": "Python development",
"source": "Package source", "online": "Online — official Arch mirrors", "offline": "Offline — verified local package source", "refresh_source": "Check package source",
"source_checking": "Checking network and local packages…", "source_online_ready": "The network is reachable. The backend verifies package sources again before erasing the disk.",
"source_online_unready": "No connection to an official Arch mirror. Connect to a network or use a fully verified offline source.",
"source_offline_ready": "A signed, complete local package source is verified.", "source_offline_unready": "No fully verified local package source is available.",
"review_title": "Review the installation", "review_copy": "The disk is rescanned before final confirmation. Approval applies only to this exact identified device.",
"review_empty": "Choose a disk and complete the previous steps.", "summary": "Disk\n{disk}\n\nUser\n{full_name} ({username})\n\nComputer\n{hostname}\n\nEncryption\n{encryption}\n\nProfiles\n{profiles}\n\nPackage source\n{source}",
"enabled": "LUKS2 enabled", "disabled": "Not encrypted", "confirm_title": "Erase all data on this disk?",
"confirm_body": "This permanently erases all partitions and data on:\n\n{disk}\n\nType the exact disk path to confirm: {path}",
"typed_path": "Disk path", "cancel": "Cancel", "erase_install": "Erase disk and install", "preflight": "Rechecking disk and package source…",
"target_changed": "The disk is no longer the same device or is no longer safe. Select it again.", "source_changed": "The selected package source is not ready. Check the connection or local media.",
"username_error": "Username may contain only lowercase letters, numbers, _ and -.", "hostname_error": "Invalid computer name.", "full_name_error": "Full name must be at most 64 characters and cannot contain a colon or line break.",
"password_short": "Password must be at least 8 characters.", "password_mismatch": "Passwords do not match.",
"encryption_short": "Encryption passphrase must be at least 8 characters.", "encryption_mismatch": "Encryption passphrases do not match.",
"source_required": "The selected package source is not ready.", "disk_required": "Select an eligible disk.",
"progress_title": "Installing BifrOSt", "progress_copy": "The backend verifies disk identity and package sources before anything is erased.", "progress_start": "Preparing installation…",
"phase_preflight": "Verifying disk and safety conditions…", "phase_source": "Verifying package sources…", "phase_plan": "Building a safe disk plan…", "phase_apply": "Installing the Arch base system…",
"phase_postinstall": "Installing the BifrOSt experience…", "phase_verify": "Verifying the installed system…", "phase_evidence": "Persisting sanitized installation evidence…", "success": "Installation complete",
"success_detail": "BifrOSt is ready. Remove the installation media and restart the computer.",
"failure": "Installation failed",
"failure_detail": "{summary}\nError code: {code}\nTechnical details: {message}",
"error_input": "The installer input is invalid or insecure.",
"error_environment": "The live environment does not meet the safety requirements.",
"error_target": "The selected disk could not be safely verified.",
"error_source": "The complete package source could not be verified.",
"error_plan": "The safe installation plan was not accepted.",
"error_install": "Installation stopped after disk writes may have begun.",
"error_verify": "Verification of the installed system failed.",
"error_backend": "Communication with the installer backend was interrupted.",
"error_unknown": "An unknown installer error occurred.",
"log_path": "Retained installer logs: {path}",
"recovery_wiped": "Disk state: possibly modified. No rollback is claimed. Keep the live environment running and use the retained log for diagnosis or reinstalling.",
"recovery_safe": "Disk state: unchanged according to a valid backend terminal event. You can correct the problem and try again.",
"recovery_unknown": "Disk state: unknown. The terminal event is missing or invalid, so the installer cannot claim whether the disk was modified. Keep the live environment running and retain the installer logs for diagnosis.",
"close_blocked": "This window cannot close while installation is running. Installation continues; wait for a terminal result.",
"retry": "Go back and rescan", "backend_protocol": "The backend exited without a valid terminal event.",
},
}
class InstallerWindow(Adw.ApplicationWindow):
def __init__(self, app):
super().__init__(application=app, title=STRINGS["is"]["title"])
self.set_default_size(720, 520)
self.set_size_request(640, 480)
self.connect("close-request", self._close_requested)
self.language = "is"
self.page_index = 0
self.installing = False
self.inventory = []
self.eligible = []
self.source_states = {"online": {"ready": False}, "offline": {"ready": False}}
self.source_mode_ids = ["online"]
self.wifi_networks = []
self.wifi_generation = 0
self.scan_generation = 0
self.source_generation = 0
self.approved_fingerprint = None
self.change_revision = 0
self.text_widgets = []
self.headings = []
self.toast_overlay = Adw.ToastOverlay()
self.set_content(self.toast_overlay)
shell = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
self.toast_overlay.set_child(shell)
hero = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=12)
hero.add_css_class("hero")
shell.append(hero)
mark = Gtk.Image.new_from_file("/usr/share/bifrost/branding/bifrost-mark.svg")
mark.set_pixel_size(48)
hero.append(mark)
brand = Gtk.Label(label="BifrOSt", xalign=0)
brand.add_css_class("hero-title")
hero.append(brand)
self.tagline = self._tracked_label("tagline", xalign=0, wrap=True)
self.tagline.add_css_class("hero-subtitle")
self.tagline.set_hexpand(True)
hero.append(self.tagline)
self.step_label = Gtk.Label(xalign=1)
self.step_label.add_css_class("accent")
hero.append(self.step_label)
content = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
content.set_margin_top(12)
content.set_margin_bottom(12)
content.set_margin_start(18)
content.set_margin_end(18)
content.set_hexpand(True)
content.set_vexpand(True)
shell.append(content)
self.stack = Gtk.Stack(transition_type=Gtk.StackTransitionType.SLIDE_LEFT_RIGHT,
transition_duration=220)
self.stack.set_vexpand(True)
content.append(self.stack)
self._welcome_page()
self._disk_page()
self._identity_page()
self._wifi_page()
self._options_page()
self._review_page()
self._progress_page()
self.nav = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=12)
content.append(self.nav)
self.back = self._tracked_button("back")
self.back.connect("clicked", self._go_back)
self.nav.append(self.back)
self.nav.append(Gtk.Box(hexpand=True))
self.next = self._tracked_button("next")
self.next.add_css_class("suggested-action")
self.next.connect("clicked", self._go_next)
self.nav.append(self.next)
self._refresh_text()
self._show_page(0)
GLib.idle_add(self._rescan)
GLib.idle_add(self._refresh_sources)
GLib.idle_add(self._scan_wifi)
def t(self, key, **values):
return STRINGS[self.language][key].format(**values)
def _tracked_label(self, key, **kwargs):
label = Gtk.Label(label=self.t(key), **kwargs)
self.text_widgets.append((label, key))
return label
def _tracked_button(self, key):
button = Gtk.Button(label=self.t(key))
self.text_widgets.append((button, key))
return button
def _field_label(self, key, widget):
label = self._tracked_label(key, xalign=0)
label.set_mnemonic_widget(widget)
return label
def _heading(self, title_key, copy_key):
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=9)
title = self._tracked_label(title_key, xalign=0, wrap=True)
title.add_css_class("step-title")
copy = self._tracked_label(copy_key, xalign=0, wrap=True)
copy.add_css_class("step-copy")
box.append(title)
box.append(copy)
self.headings.append((title, title_key, copy, copy_key))
return box
def _add_page(self, body, name):
viewport = Gtk.ScrolledWindow(hscrollbar_policy=Gtk.PolicyType.NEVER,
vscrollbar_policy=Gtk.PolicyType.AUTOMATIC)
viewport.set_child(body)
self.stack.add_named(viewport, name)
def _entry_row(self, key, entry):
row = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
label = self._field_label(key, entry)
row.append(label)
row.append(entry)
return row
def _welcome_page(self):
page = self._heading("welcome_title", "welcome_copy")
language_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
language_box.add_css_class("card")
language_box.set_margin_top(20)
self.language_combo = Gtk.DropDown()
language_box.append(self._field_label("language", self.language_combo))
self.language_combo.set_model(Gtk.StringList.new(["Íslenska", "English"]))
self.language_combo.connect("notify::selected", self._language_changed)
language_box.append(self.language_combo)
page.append(language_box)
features = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=13)
features.add_css_class("card")
features.set_margin_top(14)
for icon, title_key, copy_key in [
("preferences-desktop-keyboard-symbolic", "feature_locale", "feature_locale_copy"),
("security-high-symbolic", "feature_arch", "feature_arch_copy"),
]:
row = Gtk.Box(spacing=13)
row.append(Gtk.Image.new_from_icon_name(icon))
labels = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
title = self._tracked_label(title_key, xalign=0)
title.add_css_class("heading")
labels.append(title)
labels.append(self._tracked_label(copy_key, xalign=0, wrap=True))
row.append(labels)
features.append(row)
page.append(features)
self._add_page(page, "welcome")
def _disk_page(self):
page = self._heading("disk_title", "disk_copy")
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=11)
card.add_css_class("card")
card.set_margin_top(18)
self.disk_combo = Gtk.DropDown()
card.append(self._field_label("disk_title", self.disk_combo))
card.append(self.disk_combo)
self.disk_combo.set_model(Gtk.StringList.new([self.t("no_disks")]))
self.disk_combo.set_sensitive(False)
self.disk_combo.connect("notify::selected", self._relevant_changed)
controls = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
self.rescan_button = self._tracked_button("rescan")
self.rescan_button.connect("clicked", self._rescan)
controls.append(self.rescan_button)
self.disk_status = Gtk.Label(label=self.t("scanning"), xalign=0, wrap=True, hexpand=True)
controls.append(self.disk_status)
card.append(controls)
page.append(card)
rejected = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=7)
rejected.add_css_class("card")
rejected.set_margin_top(12)
heading = self._tracked_label("rejected_title", xalign=0)
heading.add_css_class("heading")
rejected.append(heading)
self.rejected_label = Gtk.Label(label=self.t("none_rejected"), xalign=0, wrap=True,
selectable=True)
self.rejected_label.add_css_class("rejected")
rejected.append(self.rejected_label)
page.append(rejected)
warning = self._tracked_label("disk_warning", xalign=0, wrap=True)
warning.add_css_class("warning")
warning.set_margin_top(12)
page.append(warning)
self._add_page(page, "disk")
def _identity_page(self):
page = self._heading("identity_title", "identity_copy")
grid = Gtk.Grid(column_spacing=16, row_spacing=13)
grid.add_css_class("card")
grid.set_margin_top(18)
self.fullname = Gtk.Entry()
self.username = Gtk.Entry()
self.hostname = Gtk.Entry(text="bifrost")
self.user_password = Gtk.PasswordEntry(show_peek_icon=True)
self.user_password2 = Gtk.PasswordEntry(show_peek_icon=True)
grid.attach(self._entry_row("full_name", self.fullname), 0, 0, 1, 1)
grid.attach(self._entry_row("username", self.username), 1, 0, 1, 1)
grid.attach(self._entry_row("user_password", self.user_password), 0, 1, 1, 1)
grid.attach(self._entry_row("confirm_password", self.user_password2), 1, 1, 1, 1)
grid.attach(self._entry_row("hostname", self.hostname), 0, 2, 2, 1)
for widget in (self.fullname, self.username, self.hostname, self.user_password,
self.user_password2):
widget.connect("changed", self._relevant_changed)
page.append(grid)
self._add_page(page, "identity")
def _wifi_page(self):
page = self._heading("wifi_title", "wifi_copy")
card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
card.add_css_class("card")
card.set_margin_top(18)
self.wifi_combo = Gtk.DropDown()
card.append(self._field_label("wifi_network", self.wifi_combo))
card.append(self.wifi_combo)
self.wifi_password = Gtk.PasswordEntry(show_peek_icon=True)
card.append(self._entry_row("wifi_password", self.wifi_password))
controls = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
self.wifi_rescan = self._tracked_button("wifi_rescan")
self.wifi_rescan.connect("clicked", self._scan_wifi)
controls.append(self.wifi_rescan)
self.wifi_connect = self._tracked_button("wifi_connect")
self.wifi_connect.add_css_class("suggested-action")
self.wifi_connect.connect("clicked", self._connect_wifi)
controls.append(self.wifi_connect)
card.append(controls)
self.wifi_status = Gtk.Label(label=self.t("wifi_scanning"), xalign=0, wrap=True)
card.append(self.wifi_status)
page.append(card)
self._add_page(page, "wifi")
def _options_page(self):
page = self._heading("options_title", "options_copy")
storage = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
storage.add_css_class("card")
storage.set_margin_top(18)
self.encrypt = Gtk.CheckButton(label=self.t("encryption"))
self.text_widgets.append((self.encrypt, "encryption"))
self.encrypt.connect("toggled", self._encryption_changed)
storage.append(self.encrypt)
self.encryption_fields = Gtk.Grid(column_spacing=16, row_spacing=10)
self.encryption_password = Gtk.PasswordEntry(show_peek_icon=True)
self.encryption_password2 = Gtk.PasswordEntry(show_peek_icon=True)
self.encryption_fields.attach(self._entry_row("encryption_password", self.encryption_password), 0, 0, 1, 1)
self.encryption_fields.attach(self._entry_row("confirm_encryption", self.encryption_password2), 1, 0, 1, 1)
self.encryption_fields.set_visible(False)
self.encryption_password.connect("changed", self._relevant_changed)
self.encryption_password2.connect("changed", self._relevant_changed)
storage.append(self.encryption_fields)
page.append(storage)
profiles = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=7)
profiles.add_css_class("card")
profiles.set_margin_top(12)
title = self._tracked_label("profiles", xalign=0)
title.add_css_class("heading")
profiles.append(title)
self.profile_buttons = {}
for profile_id in PROFILE_IDS:
button = Gtk.CheckButton(label=self.t(f"profile_{profile_id}"))
self.text_widgets.append((button, f"profile_{profile_id}"))
button.set_active(profile_id == "base")
button.set_sensitive(profile_id != "base")
button.connect("toggled", self._relevant_changed)
profiles.append(button)
self.profile_buttons[profile_id] = button
page.append(profiles)
source = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=9)
source.add_css_class("card")
source.set_margin_top(12)
self.source_combo = Gtk.DropDown()
title = self._field_label("source", self.source_combo)
title.add_css_class("heading")
source.append(title)
self.source_combo.connect("notify::selected", self._source_changed)
source.append(self.source_combo)
controls = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
self.source_refresh = self._tracked_button("refresh_source")
self.source_refresh.connect("clicked", self._refresh_sources)
controls.append(self.source_refresh)
self.source_status = Gtk.Label(label=self.t("source_checking"), xalign=0, wrap=True,
hexpand=True)
controls.append(self.source_status)
source.append(controls)
page.append(source)
self._rebuild_source_combo("online")
self._add_page(page, "options")
def _review_page(self):
page = self._heading("review_title", "review_copy")
self.summary = Gtk.Label(label=self.t("review_empty"), xalign=0, yalign=0, wrap=True,
selectable=True)
self.summary.add_css_class("card")
self.summary.set_margin_top(18)
page.append(self.summary)
self._add_page(page, "review")
def _progress_page(self):
page = self._heading("progress_title", "progress_copy")
self.progress = Gtk.ProgressBar(show_text=True)
self.progress.set_text(self.t("progress_start"))
self.progress.set_margin_top(34)
page.append(self.progress)
self.status = Gtk.Label(label="", xalign=0, wrap=True, selectable=True)
self.status.add_css_class("step-copy")
self.status.set_margin_top(14)
page.append(self.status)
self.log_state = Gtk.Label(label="", xalign=0, wrap=True, selectable=True)
self.log_state.set_margin_top(10)
page.append(self.log_state)
self.recovery_state = Gtk.Label(label="", xalign=0, wrap=True, selectable=True)
self.recovery_state.set_margin_top(10)
page.append(self.recovery_state)
self.close_state = Gtk.Label(label="", xalign=0, wrap=True)
self.close_state.add_css_class("warning")
self.close_state.set_margin_top(10)
self.close_state.set_visible(False)
page.append(self.close_state)
self.retry_button = self._tracked_button("retry")
self.retry_button.connect("clicked", self._retry_after_failure)
self.retry_button.set_halign(Gtk.Align.START)
self.retry_button.set_margin_top(14)
self.retry_button.set_visible(False)
page.append(self.retry_button)
self._add_page(page, "progress")
def _toast(self, text):
self.toast_overlay.add_toast(Adw.Toast(title=text, timeout=5))
def _refresh_text(self):
self.set_title(self.t("title"))
for widget, key in self.text_widgets:
if isinstance(widget, Gtk.Button) or isinstance(widget, Gtk.CheckButton):
widget.set_label(self.t(key))
else:
widget.set_label(self.t(key))
self._show_page(self.page_index)
self._render_inventory()
selected_source = self._selected_source_mode()
self._rebuild_source_combo(selected_source)
self._render_source_state()
if self.page_index == 5:
self._update_summary()
def _language_changed(self, combo, _param):
self.language = "en" if combo.get_selected() == 1 else "is"
self._invalidate_approval()
self._refresh_text()
def _show_page(self, index):
names = ["welcome", "disk", "identity", "wifi", "options", "review", "progress"]
self.page_index = index
self.stack.set_visible_child_name(names[index])
self.step_label.set_label(self.t("step", current=min(index + 1, 6)))
self.back.set_visible(index > 0 and index < 6)
self.next.set_visible(index < 6)
self.next.set_label(self.t("install") if index == 5 else self.t("next"))
self.next.remove_css_class("install-button")
if index == 5:
self.next.add_css_class("install-button")
self.next.set_sensitive(not self.installing)
def _selected_disk(self):
selected = self.disk_combo.get_selected()
if selected == Gtk.INVALID_LIST_POSITION or selected >= len(self.eligible):
return None
return self.eligible[selected]
def _selected_source_mode(self):
if not hasattr(self, "source_combo"):
return "online"
selected = self.source_combo.get_selected()
if selected == Gtk.INVALID_LIST_POSITION or selected >= len(self.source_mode_ids):
return "online"
return self.source_mode_ids[selected]
def _disk_description(self, disk):
stable_id = disk["wwn"] or disk["serial"] or self.t("unavailable")
model = disk["model"] or "Unknown disk"
return (f"{model} - {human_size(disk['size'])}\n"
f"{disk['path']} · {self.t('stable_id')}: {stable_id}")
def _render_inventory(self):
if not hasattr(self, "disk_combo"):
return
previous_disk = self._selected_disk()
previous = target_fingerprint(previous_disk) if previous_disk else None
self.eligible = [disk for disk in self.inventory if disk["eligible"]]
labels = [self._disk_description(disk).replace("\n", " · ") for disk in self.eligible]
self.disk_combo.set_model(Gtk.StringList.new(labels or [self.t("no_disks")]))
self.disk_combo.set_sensitive(bool(self.eligible))
selected = preserved_selection_index(self.eligible, previous)
self.disk_combo.set_selected(
selected if selected is not None else Gtk.INVALID_LIST_POSITION)
rejected_lines = []
for disk in (item for item in self.inventory if not item["eligible"]):
reasons = []
for reason in disk["reasons"]:
key = f"reason_{reason['code']}"
reasons.append(self.t(key, detail=reason["detail"]) if "{detail}" in STRINGS[self.language][key]
else self.t(key))
rejected_lines.append(f"{self._disk_description(disk)}\n — " + "; ".join(reasons))
self.rejected_label.set_text("\n\n".join(rejected_lines) or self.t("none_rejected"))
def _rescan(self, _button=None):
self.scan_generation += 1
generation = self.scan_generation
self.rescan_button.set_sensitive(False)
self.disk_status.set_text(self.t("scanning"))
threading.Thread(target=self._inventory_worker, args=(generation,), daemon=True).start()
return GLib.SOURCE_REMOVE
def _inventory_worker(self, generation):
try:
result, error = read_inventory(), None
except Exception as exc:
result, error = [], str(exc)
GLib.idle_add(self._inventory_finished, generation, result, error)
def _inventory_finished(self, generation, result, error):
if generation != self.scan_generation:
return
self.rescan_button.set_sensitive(True)
self.inventory = result
self._render_inventory()
self._invalidate_approval()
if error:
self.disk_status.set_text(self.t("scan_error", error=error))
self.disk_status.add_css_class("rejected")
elif self.eligible:
self.disk_status.set_text(self.t("eligible_count", count=len(self.eligible)))
self.disk_status.remove_css_class("rejected")
else:
self.disk_status.set_text(self.t("no_disks"))
self.disk_status.add_css_class("rejected")
def _refresh_sources(self, _button=None):
self.source_generation += 1
generation = self.source_generation
self.source_refresh.set_sensitive(False)
self.source_status.set_text(self.t("source_checking"))
threading.Thread(target=self._source_worker, args=(generation,), daemon=True).start()
return GLib.SOURCE_REMOVE
def _scan_wifi(self, _button=None):
self.wifi_generation += 1
generation = self.wifi_generation
self.wifi_rescan.set_sensitive(False)
self.wifi_connect.set_sensitive(False)
self.wifi_status.set_text(self.t("wifi_scanning"))
threading.Thread(target=self._wifi_scan_worker, args=(generation,), daemon=True).start()
return GLib.SOURCE_REMOVE
def _wifi_scan_worker(self, generation):
try:
result = subprocess.run(
["nmcli", "-t", "-f", "SSID,SIGNAL,SECURITY", "device", "wifi", "list",
"--rescan", "yes"],
check=True, capture_output=True, text=True, timeout=20,
)
networks = []
seen = set()
for line in result.stdout.splitlines():
fields = re.split(r"(?<!\\\\):", line)
if len(fields) < 3:
continue
ssid = fields[0].replace("\\\\:", ":").replace("\\\\\\\\", "\\\\").strip()
if not ssid or ssid in seen:
continue
seen.add(ssid)
networks.append((ssid, fields[1], ":".join(fields[2:])))
networks.sort(key=lambda item: int(item[1] or 0), reverse=True)
GLib.idle_add(self._wifi_scan_finished, generation, networks, None)
except (OSError, subprocess.SubprocessError) as error:
GLib.idle_add(self._wifi_scan_finished, generation, [], str(error))
def _wifi_scan_finished(self, generation, networks, error):
if generation != self.wifi_generation:
return GLib.SOURCE_REMOVE
self.wifi_networks = networks
labels = [f"{ssid} · {signal}% · {security or self.t('wifi_open')}"
for ssid, signal, security in networks]
self.wifi_combo.set_model(Gtk.StringList.new(labels or [self.t("wifi_none")]))
self.wifi_combo.set_selected(0)
self.wifi_combo.set_sensitive(bool(networks))
self.wifi_connect.set_sensitive(bool(networks))
self.wifi_rescan.set_sensitive(True)
self.wifi_status.set_text(self.t("wifi_failed", error=error) if error else
("" if networks else self.t("wifi_none")))
return GLib.SOURCE_REMOVE
def _connect_wifi(self, _button):
selected = self.wifi_combo.get_selected()
if selected == Gtk.INVALID_LIST_POSITION or selected >= len(self.wifi_networks):
return
ssid, _signal, security = self.wifi_networks[selected]
password = self.wifi_password.get_text()
self.wifi_connect.set_sensitive(False)
self.wifi_status.set_text(self.t("wifi_scanning"))
threading.Thread(target=self._wifi_connect_worker,
args=(ssid, security, password), daemon=True).start()
def _wifi_connect_worker(self, ssid, security, password):
command = ["nmcli", "device", "wifi", "connect", ssid]
if security:
command.extend(["password", password])
try:
subprocess.run(command, check=True, capture_output=True, text=True, timeout=30)
GLib.idle_add(self._wifi_connect_finished, ssid, None)
except (OSError, subprocess.SubprocessError) as error:
detail = getattr(error, "stderr", None) or str(error)
GLib.idle_add(self._wifi_connect_finished, ssid, detail.strip())
def _wifi_connect_finished(self, ssid, error):
self.wifi_connect.set_sensitive(bool(self.wifi_networks))
self.wifi_status.set_text(self.t("wifi_failed", error=error) if error
else self.t("wifi_connected", ssid=ssid))
if not error:
self._refresh_sources()
return GLib.SOURCE_REMOVE
def _source_worker(self, generation):
states = source_readiness()
GLib.idle_add(self._source_finished, generation, states)
def _source_finished(self, generation, states):
if generation != self.source_generation:
return
selected = self._selected_source_mode()
self.source_states = states
self.source_refresh.set_sensitive(True)
self._rebuild_source_combo(selected)
self._render_source_state()
self._invalidate_approval()
def _rebuild_source_combo(self, preferred):
if not hasattr(self, "source_combo"):
return
self.source_mode_ids = ["online"]
if self.source_states.get("offline", {}).get("ready"):
self.source_mode_ids.append("offline")
self.source_combo.set_model(Gtk.StringList.new([self.t(mode) for mode in self.source_mode_ids]))
self.source_combo.set_selected(self.source_mode_ids.index(preferred)
if preferred in self.source_mode_ids else 0)
def _render_source_state(self):
if not hasattr(self, "source_status"):
return
mode = self._selected_source_mode()
ready = bool(self.source_states.get(mode, {}).get("ready"))
self.source_status.set_text(self.t(f"source_{mode}_{'ready' if ready else 'unready'}"))
self.source_status.remove_css_class("success")
self.source_status.remove_css_class("rejected")
self.source_status.add_css_class("success" if ready else "rejected")
def _source_changed(self, _combo, _param):
self._render_source_state()
self._invalidate_approval()
def _encryption_changed(self, button):
self.encryption_fields.set_visible(button.get_active())
self._invalidate_approval()
def _relevant_changed(self, *_args):
self._invalidate_approval()
def _invalidate_approval(self):
self.approved_fingerprint = None
self.change_revision += 1
def _validate_widgets(self, require_source=True):
selected = self._selected_disk()
if not selected:
self._toast(self.t("disk_required"))
return False
mode = self._selected_source_mode()
if require_source and not self.source_states.get(mode, {}).get("ready"):
self._toast(self.t("source_required"))
return False
error = validate_identity(
self.username.get_text().strip(), self.hostname.get_text().strip(),
self.user_password.get_text(), self.user_password2.get_text(),
self.encrypt.get_active(), self.encryption_password.get_text(),
self.encryption_password2.get_text(), self.fullname.get_text().strip(),
)
if error:
key = {"username": "username_error", "hostname": "hostname_error",
"full_name": "full_name_error"}.get(error, error)
self._toast(self.t(key))
return False
return True
def _update_summary(self):
disk = self._selected_disk()
if not disk:
self.summary.set_text(self.t("review_empty"))
return
profiles = [self.t(f"profile_{profile_id}") for profile_id in PROFILE_IDS
if self.profile_buttons[profile_id].get_active()]
self.summary.set_text(self.t(
"summary", disk=self._disk_description(disk),
full_name=self.fullname.get_text().strip() or self.username.get_text().strip(),
username=self.username.get_text().strip(), hostname=self.hostname.get_text().strip(),
encryption=self.t("enabled" if self.encrypt.get_active() else "disabled"),
profiles=", ".join(profiles), source=self.t(self._selected_source_mode()),
))
def _go_back(self, _button):
if not self.installing and self.page_index > 0:
self._show_page(self.page_index - 1)
def _close_requested(self, *_args):
if not self.installing:
return False
message = self.t("close_blocked")
self.close_state.set_text(message)
self.close_state.set_visible(True)
self._toast(message)
return True
def _go_next(self, _button):
if self.page_index == 1 and not self._selected_disk():
self._toast(self.t("disk_required"))
return
if self.page_index == 2 and not self._validate_widgets(require_source=False):
return
if self.page_index == 4 and not self._validate_widgets():
return
if self.page_index == 4:
self._update_summary()
if self.page_index == 5:
if self._validate_widgets():
self._begin_confirmation()
return
self._show_page(self.page_index + 1)
def _freeze_request(self):
disk = dict(self._selected_disk())
profiles = [profile_id for profile_id in PROFILE_IDS
if self.profile_buttons[profile_id].get_active()]
intent = {
"schema_version": 2,
"target": {
"path": disk["path"], "model": disk["model"], "serial": disk["serial"],
"wwn": disk["wwn"], "size": disk["size"],
"logical_sector": disk["logical_sector"],
},
"options": {
"encryption": self.encrypt.get_active(), "profiles": profiles,
"installer_language": self.language,
"source_mode": self._selected_source_mode(),
"system_defaults": {
"hostname": self.hostname.get_text().strip(),
"timezone": "Atlantic/Reykjavik", "locale": "is_IS.UTF-8",
"keyboard_layout": KEYBOARD_LAYOUT, "username": self.username.get_text().strip(),
"full_name": self.fullname.get_text().strip(),
},
},
}
secrets_data = {"schema_version": 2, "user_password": self.user_password.get_text()}
if self.encrypt.get_active():
secrets_data["encryption_password"] = self.encryption_password.get_text()
return {"intent": intent, "secrets": secrets_data, "disk": disk,
"source_mode": self._selected_source_mode(), "language": self.language,
"revision": self.change_revision}
def _begin_confirmation(self):
frozen = self._freeze_request()
self.next.set_sensitive(False)
self._toast(self.t("preflight"))
threading.Thread(target=self._confirmation_preflight_worker, args=(frozen,),
daemon=True).start()
def _confirmation_preflight_worker(self, frozen):
try:
inventory = read_inventory()
sources = source_readiness()
error = None
except Exception as exc:
inventory, sources, error = [], {}, str(exc)
GLib.idle_add(self._confirmation_preflight_finished, frozen, inventory, sources, error)
def _confirmation_preflight_finished(self, frozen, inventory, sources, error):
self.next.set_sensitive(True)
if frozen["revision"] != self.change_revision:
self.inventory = inventory
self.source_states = sources or self.source_states
self._render_inventory()
self._rebuild_source_combo(self._selected_source_mode())
self._render_source_state()
self._toast(self.t("target_changed"))
return
self.inventory = inventory
self.source_states = sources or self.source_states
wanted = target_fingerprint(frozen["disk"])
matched = next((disk for disk in inventory
if disk["eligible"] and target_fingerprint(disk) == wanted), None)
self._render_inventory()
self._rebuild_source_combo(frozen["source_mode"])
self._render_source_state()
if error or not matched:
self._invalidate_approval()
self._toast(self.t("target_changed"))
return
if not sources.get(frozen["source_mode"], {}).get("ready"):
self._invalidate_approval()
self._toast(self.t("source_changed"))
return
self._show_confirmation_dialog(frozen)
def _show_confirmation_dialog(self, frozen):
disk = frozen["disk"]
self.approved_fingerprint = target_fingerprint(disk)
dialog = Adw.MessageDialog(
transient_for=self, modal=True, heading=self.t("confirm_title"),
body=self.t("confirm_body", disk=self._disk_description(disk), path=disk["path"]),
)
entry_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
typed_path = Gtk.Entry(activates_default=True)
entry_box.append(self._field_label("typed_path", typed_path))
entry_box.append(typed_path)
dialog.set_extra_child(entry_box)
dialog.add_response("cancel", self.t("cancel"))
dialog.add_response("install", self.t("erase_install"))
dialog.set_response_appearance("install", Adw.ResponseAppearance.DESTRUCTIVE)
dialog.set_default_response("cancel")
dialog.set_close_response("cancel")
dialog.set_response_enabled("install", False)
typed_path.connect("changed", lambda entry: dialog.set_response_enabled(
"install", entry.get_text() == disk["path"]))
def responded(_dialog, response):
approved = (response == "install" and typed_path.get_text() == disk["path"]
and self.approved_fingerprint == target_fingerprint(disk))
self._invalidate_approval()
if approved:
self._start_install(frozen)
dialog.connect("response", responded)
dialog.present()
typed_path.grab_focus()
def _start_install(self, frozen):
self.installing = True
self._show_page(6)
self.nav.set_visible(False)
self.progress.set_fraction(0.02)
self.progress.set_text(self.t("progress_start"))
self.status.set_text(self.t("phase_preflight"))
self.log_state.set_text("")
self.recovery_state.set_text("")
self.close_state.set_text("")
self.close_state.set_visible(False)
self.retry_button.set_visible(False)
threading.Thread(target=self._install_worker, args=(frozen["intent"], frozen["secrets"]),
daemon=True).start()
@staticmethod
def _secure_json(path, payload):
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
os.fchmod(descriptor, 0o600)
with os.fdopen(descriptor, "w") as stream:
json.dump(payload, stream, separators=(",", ":"))
stream.flush()
os.fsync(stream.fileno())
def _install_worker(self, intent, secrets_data):
terminal = None
diagnostics = []
try:
with tempfile.TemporaryDirectory(prefix="bifrost-installer-") as tmp:
intent_path = Path(tmp) / "intent.json"
secrets_path = Path(tmp) / "secrets.json"
self._secure_json(intent_path, intent)
self._secure_json(secrets_path, secrets_data)
process = subprocess.Popen(
["sudo", BACKEND, str(intent_path), str(secrets_path)],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
)
for raw_line in process.stdout:
line = raw_line.strip()
if not line:
continue
try:
event = json.loads(line)
except json.JSONDecodeError:
diagnostics.append(line)
continue
if isinstance(event, dict):
GLib.idle_add(self._backend_event, event)
if valid_terminal_event(event):
terminal = event
returncode = process.wait()
if terminal and terminal.get("event") == "success" and returncode == 0:
GLib.idle_add(self._install_done, terminal)
else:
failure = terminal or {
"schema_version": 2, "event": "error", "code": f"backend_exit_{returncode}",
"message": diagnostics[-1] if diagnostics else STRINGS["en"]["backend_protocol"],
"wiped": False, "log_dir": "/var/log/bifrost-installer",
"live_log_dir": "/var/log/bifrost-installer", "target_log_dir": None,
}
GLib.idle_add(self._install_failed, failure)
except Exception as exc:
GLib.idle_add(self._install_failed, {
"schema_version": 2, "event": "error", "code": "ui_backend_start",
"message": str(exc), "wiped": False,
"log_dir": "/var/log/bifrost-installer",
"live_log_dir": "/var/log/bifrost-installer", "target_log_dir": None,
})
def _backend_event(self, event):
if event.get("schema_version") != 2 or event.get("event") != "phase":
return
phase = event.get("phase")
fractions = {"preflight": 0.06, "source": 0.12, "plan": 0.2,
"apply": 0.42, "postinstall": 0.74, "verify": 0.9, "evidence": 0.97}
if phase in fractions:
self.progress.set_fraction(fractions[phase])
self.progress.set_text(self.t(f"phase_{phase}"))
self.status.set_text(self.t(f"phase_{phase}"))
def _install_done(self, event):
self.installing = False
self.close_state.set_visible(False)
self.progress.set_fraction(1)
self.progress.set_text(self.t("success"))
self.status.set_text(self.t("success_detail"))
self.status.add_css_class("success")
log_dir = terminal_log_dir(event)
if log_dir:
self.log_state.set_text(self.t("log_path", path=log_dir))
def _install_failed(self, event):
self.installing = False
self.close_state.set_visible(False)
self.progress.set_fraction(0)
self.progress.set_text(self.t("failure"))
event = event if isinstance(event, dict) else {}
code = str(event.get("code") or "unknown")
message = str(event.get("message") or self.t("backend_protocol"))
summary = self.t(f"error_{error_category(code)}")
self.status.set_text(self.t(
"failure_detail", summary=summary, code=code, message=message))
self.status.add_css_class("failure")
log_dir = terminal_log_dir(event) or "/var/log/bifrost-installer"
self.log_state.set_text(self.t("log_path", path=log_dir))
self.recovery_state.set_text(self.t(failure_recovery_key(event)))
self.retry_button.set_visible(True)
self._toast(self.t("failure"))
def _retry_after_failure(self, _button):
self.status.remove_css_class("failure")
self.status.remove_css_class("success")
self.nav.set_visible(True)
self._show_page(5)
self._invalidate_approval()
self._rescan()
self._refresh_sources()
class InstallerApp(Adw.Application):
def __init__(self):
super().__init__(application_id="is.bifrost.Installer")
def do_startup(self):
Adw.Application.do_startup(self)
provider = Gtk.CssProvider()
provider.load_from_data(CSS)
Gtk.StyleContext.add_provider_for_display(
Gdk.Display.get_default(), provider, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION,
)
def do_activate(self):
window = self.props.active_window or InstallerWindow(self)
window.present()
if __name__ == "__main__":
raise SystemExit(InstallerApp().run())