Menu
BifrOSt
publicLatest change 9cfe2016fbf1878ce207be2c8033c38d11ec899e - Release BifrOSt 0.2.0 by Ólafur Búi Ólafsson
#!/usr/bin/env bash
set -Eeuo pipefail
umask 077
readonly TARGET_MOUNT=/mnt
readonly TEMPLATE=/usr/share/bifrost/archinstall.json
readonly INSTALLED_ROOT=/usr/share/bifrost/installed-root
readonly PROFILE_DIR="$INSTALLED_ROOT/usr/share/bifrost/profiles"
readonly OFFLINE_DIR=/usr/share/bifrost/offline
readonly LOG_ROOT=/var/log/bifrost-installer
readonly RUNTIME_ROOT=/run/bifrost-installer
readonly MINIMUM_BYTES=$((16 * 1024 * 1024 * 1024))
emit_early_error() {
EVENT_MESSAGE=$1 python3 - <<'PY'
import json, os
print(json.dumps({"schema_version": 2, "event": "error", "code": "invalid_invocation", "message": os.environ["EVENT_MESSAGE"], "wiped": False}, ensure_ascii=False), flush=True)
PY
}
if (( EUID != 0 )); then
emit_early_error "BifrOSt installer backend must run as root."
exit 1
fi
if (( $# != 2 )); then
emit_early_error "Usage: bifrost-installer-backend INTENT_JSON SECRETS_JSON"
exit 2
fi
readonly INTENT_INPUT=$1
readonly SECRETS_INPUT=$2
IFS= read -r run_uuid </proc/sys/kernel/random/uuid
readonly RUN_ID="$(date -u +%Y%m%dT%H%M%SZ)-${run_uuid}"
readonly RUN_DIR="$LOG_ROOT/$RUN_ID"
readonly RUNTIME_DIR="$RUNTIME_ROOT/$RUN_ID"
readonly RAW_LOG="$RUNTIME_DIR/backend.raw.log"
readonly PLAN="$RUN_DIR/plan.json"
readonly GENERATED_CONFIG="$RUN_DIR/archinstall.json"
readonly GENERATED_CREDS="$RUNTIME_DIR/credentials.json"
readonly STATUS="$RUN_DIR/status.json"
readonly MIRRORLIST=/etc/pacman.d/mirrorlist
readonly MIRRORLIST_BACKUP="$RUNTIME_DIR/mirrorlist.original"
install -d -m 0700 "$RUN_DIR" "$RUNTIME_DIR"
exec 3>&1
exec >>"$RAW_LOG" 2>&1
ERROR_CODE=internal_error
ERROR_MESSAGE="The installer stopped unexpectedly."
WIPE_STARTED=0
MIRRORLIST_CHANGED=0
FINAL_SUCCESS=0
CURRENT_PHASE=initializing
write_status() {
STATUS_PATH=$STATUS STATUS_STATE=$1 STATUS_PHASE=$CURRENT_PHASE STATUS_CODE=${2:-} STATUS_MESSAGE=${3:-} STATUS_WIPED=$WIPE_STARTED RUN_ID_VALUE=$RUN_ID python3 - <<'PY'
import json, os
from pathlib import Path
p = Path(os.environ["STATUS_PATH"])
data = {
"schema_version": 2,
"run_id": os.environ["RUN_ID_VALUE"],
"state": os.environ["STATUS_STATE"],
"phase": os.environ["STATUS_PHASE"],
"wiped": os.environ["STATUS_WIPED"] == "1",
}
if os.environ["STATUS_CODE"]:
data["code"] = os.environ["STATUS_CODE"]
if os.environ["STATUS_MESSAGE"]:
data["message"] = os.environ["STATUS_MESSAGE"]
tmp = p.with_suffix(".tmp")
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
tmp.replace(p)
PY
}
emit_phase() {
EVENT_PHASE=$1 EVENT_MESSAGE=$2 RUN_ID_VALUE=$RUN_ID python3 - >&3 <<'PY'
import json, os
print(json.dumps({
"schema_version": 2,
"run_id": os.environ["RUN_ID_VALUE"],
"event": "phase",
"phase": os.environ["EVENT_PHASE"],
"message": os.environ["EVENT_MESSAGE"],
}, ensure_ascii=False), flush=True)
PY
}
phase() {
CURRENT_PHASE=$1
write_status running "" "$2"
emit_phase "$1" "$2"
}
fail() {
ERROR_CODE=$1
ERROR_MESSAGE=$2
exit 1
}
redact_file() {
local source=$1 destination=$2
[[ -f $source ]] || return 0
python3 - "$SECRETS_INPUT" "$source" "$destination" <<'PY'
import json, re, sys
from pathlib import Path
secrets_path, source_path, destination_path = map(Path, sys.argv[1:])
text = source_path.read_text(errors="replace")
try:
secrets = json.loads(secrets_path.read_text())
except Exception:
secrets = {}
def values(value, key=""):
if isinstance(value, dict):
for k, v in value.items():
yield from values(v, k)
elif isinstance(value, list):
for item in value:
yield from values(item, key)
elif isinstance(value, str) and "password" in key.lower() and value:
yield value
for secret in sorted(set(values(secrets)), key=len, reverse=True):
text = text.replace(secret, "[REDACTED]")
text = re.sub(r'(?i)(password[^:=\n]{0,24}[:=]\s*)\S+', r'\1[REDACTED]', text)
destination_path.write_text(text)
PY
chmod 0600 "$destination"
}
cleanup_target() {
(( WIPE_STARTED == 1 )) || return 0
if mountpoint -q "$TARGET_MOUNT"; then
umount -R "$TARGET_MOUNT" || true
fi
if [[ -f $PLAN ]]; then
local device kind rest
while read -r device kind rest; do
if [[ $kind == crypt ]]; then
cryptsetup close "${device##*/}" || true
fi
done < <(lsblk -nrpo PATH,TYPE,FSTYPE "$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["target"]["path"])' "$PLAN")" 2>/dev/null || true)
fi
}
on_exit() {
local rc=$?
trap - EXIT INT TERM
set +e
if (( WIPE_STARTED == 1 )) && [[ -f /var/log/archinstall/install.log ]]; then
redact_file /var/log/archinstall/install.log "$RUN_DIR/archinstall.log"
fi
cleanup_target
if (( MIRRORLIST_CHANGED == 1 )) && [[ -f $MIRRORLIST_BACKUP ]]; then
install -m 0644 "$MIRRORLIST_BACKUP" "$MIRRORLIST"
fi
redact_file "$RAW_LOG" "$RUN_DIR/backend.log"
rm -rf "$RUNTIME_DIR"
if (( rc == 0 && FINAL_SUCCESS == 1 )); then
CURRENT_PHASE=complete
write_status success "" "Installation completed and boot artifacts were verified."
EVENT_LOG=$RUN_DIR RUN_ID_VALUE=$RUN_ID python3 - >&3 <<'PY'
import json, os
print(json.dumps({"schema_version": 2, "run_id": os.environ["RUN_ID_VALUE"], "event": "success", "message": "Installation completed and boot artifacts were verified.", "log_dir": os.environ["EVENT_LOG"], "wiped": True}, ensure_ascii=False), flush=True)
PY
else
CURRENT_PHASE=${CURRENT_PHASE:-failed}
write_status failed "$ERROR_CODE" "$ERROR_MESSAGE"
EVENT_CODE=$ERROR_CODE EVENT_MESSAGE=$ERROR_MESSAGE EVENT_LOG=$RUN_DIR EVENT_WIPED=$WIPE_STARTED RUN_ID_VALUE=$RUN_ID python3 - >&3 <<'PY'
import json, os
print(json.dumps({"schema_version": 2, "run_id": os.environ["RUN_ID_VALUE"], "event": "error", "code": os.environ["EVENT_CODE"], "message": os.environ["EVENT_MESSAGE"], "log_dir": os.environ["EVENT_LOG"], "wiped": os.environ["EVENT_WIPED"] == "1"}, ensure_ascii=False), flush=True)
PY
fi
exec 3>&-
exit "$rc"
}
trap on_exit EXIT
trap 'ERROR_CODE=interrupted; ERROR_MESSAGE="Installation was interrupted."; exit 130' INT TERM
validate_input_file() {
local path=$1 label=$2 expected_owner=${SUDO_UID:-0} mode owner links
[[ -f $path && ! -L $path ]] || fail invalid_input "$label must be a regular, non-symlink file."
mode=$(stat -c %a "$path")
owner=$(stat -c %u "$path")
links=$(stat -c %h "$path")
[[ $mode == 600 ]] || fail insecure_input "$label must have mode 0600."
[[ $links == 1 ]] || fail insecure_input "$label must not be hard-linked."
[[ $owner == 0 || $owner == "$expected_owner" ]] || fail insecure_input "$label has an unexpected owner."
}
validate_environment() {
[[ -d /sys/firmware/efi/efivars ]] || fail uefi_required "BifrOSt requires a UEFI booted installer; legacy BIOS mode is not supported."
python3 - <<'PY' || fail secure_boot_enabled "Secure Boot is enabled, but this release does not install signed kernels. Disable Secure Boot before installing."
from pathlib import Path
matches = list(Path("/sys/firmware/efi/efivars").glob("SecureBoot-*"))
if matches:
data = matches[0].read_bytes()
if len(data) >= 5 and data[4] == 1:
raise SystemExit(1)
PY
local command
for command in archinstall arch-chroot bootctl cryptsetup findmnt lsblk mountpoint openssl pacman python3 stat udevadm umount; do
command -v "$command" >/dev/null || fail missing_dependency "Required installer command is unavailable: $command"
done
if findmnt --mountpoint "$TARGET_MOUNT" >/dev/null 2>&1; then
fail target_mount_busy "$TARGET_MOUNT is already mounted; unmount it before installing."
fi
}
build_plan_and_configs() {
python3 - "$INTENT_INPUT" "$SECRETS_INPUT" "$TEMPLATE" "$PROFILE_DIR" "$PLAN" "$GENERATED_CONFIG" "$GENERATED_CREDS" "$MINIMUM_BYTES" <<'PY'
import hashlib
import json
import os
import re
import subprocess
import sys
from pathlib import Path
intent_path, secrets_path, template_path, profile_dir, plan_path, config_path, creds_path = map(Path, sys.argv[1:8])
minimum = int(sys.argv[8])
class Invalid(Exception):
pass
def obj(value, name):
if not isinstance(value, dict):
raise Invalid(f"{name} must be a JSON object")
return value
def exact_keys(value, required, optional, name):
missing = required - value.keys()
unknown = value.keys() - required - optional
if missing:
raise Invalid(f"{name} is missing: {', '.join(sorted(missing))}")
if unknown:
raise Invalid(f"{name} has unknown fields: {', '.join(sorted(unknown))}")
def text(value, name, maximum=256):
if not isinstance(value, str) or not value or len(value) > maximum or "\0" in value or "\n" in value or "\r" in value:
raise Invalid(f"{name} must be a non-empty single-line string")
return value
try:
intent = obj(json.loads(intent_path.read_text()), "intent")
secrets = obj(json.loads(secrets_path.read_text()), "secrets")
exact_keys(intent, {"schema_version", "target", "options"}, set(), "intent")
if intent["schema_version"] != 2:
raise Invalid("intent schema_version must be 2")
target = obj(intent["target"], "target")
exact_keys(target, {"path", "model", "serial", "wwn", "size", "logical_sector"}, set(), "target")
path = text(target["path"], "target.path")
if not path.startswith("/dev/"):
raise Invalid("target.path must be an absolute /dev path")
for field in ("model", "serial", "wwn"):
if not isinstance(target[field], str) or len(target[field]) > 256 or any(c in target[field] for c in "\0\n\r"):
raise Invalid(f"target.{field} must be a single-line string")
if not target["serial"].strip() and not target["wwn"].strip():
raise Invalid("target must have a non-empty serial or WWN")
if type(target["size"]) is not int or target["size"] < minimum:
raise Invalid(f"target.size must be at least {minimum} bytes")
sector = target["logical_sector"]
if type(sector) is not int or sector < 512 or sector > 4096 or sector & (sector - 1):
raise Invalid("target.logical_sector must be a power of two from 512 through 4096")
if target["size"] % sector:
raise Invalid("target.size is not a whole number of logical sectors")
options = obj(intent["options"], "options")
exact_keys(options, {"encryption", "profiles", "installer_language", "source_mode", "system_defaults"}, set(), "options")
if type(options["encryption"]) is not bool:
raise Invalid("options.encryption must be boolean")
if options["installer_language"] not in ("is", "en"):
raise Invalid("options.installer_language must be is or en")
if options["source_mode"] not in ("online", "offline"):
raise Invalid("options.source_mode must be online or offline")
allowed_profiles = ("base", "dev-rust", "dev-containers", "dev-web", "dev-python")
profiles = options["profiles"]
if not isinstance(profiles, list) or not profiles or any(type(p) is not str for p in profiles):
raise Invalid("options.profiles must be a non-empty string array")
if len(profiles) != len(set(profiles)) or any(p not in allowed_profiles for p in profiles):
raise Invalid("options.profiles contains a duplicate or unsupported profile ID")
if "base" not in profiles:
raise Invalid("the base profile must always be selected")
profiles = [p for p in allowed_profiles if p in profiles]
defaults = obj(options["system_defaults"], "system_defaults")
exact_keys(defaults, {"hostname", "timezone", "locale", "keyboard_layout", "username", "full_name"}, set(), "system_defaults")
hostname = text(defaults["hostname"], "system_defaults.hostname", 63)
username = text(defaults["username"], "system_defaults.username", 32)
full_name = defaults["full_name"]
if not isinstance(full_name, str) or len(full_name) > 64 or any(c in full_name for c in ":\0\n\r"):
raise Invalid("system_defaults.full_name is invalid")
if not re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9.-]{0,61}[A-Za-z0-9])?", hostname):
raise Invalid("system_defaults.hostname is invalid")
if not re.fullmatch(r"[a-z_][a-z0-9_-]{0,31}", username):
raise Invalid("system_defaults.username is invalid")
if defaults["timezone"] not in ("Atlantic/Reykjavik", "UTC"):
raise Invalid("system_defaults.timezone is unsupported")
if defaults["locale"] not in ("is_IS.UTF-8", "en_US.UTF-8"):
raise Invalid("system_defaults.locale is unsupported")
if defaults["keyboard_layout"] not in ("is-latin1", "us"):
raise Invalid("system_defaults.keyboard_layout is unsupported")
exact_keys(secrets, {"schema_version", "user_password"} | ({"encryption_password"} if options["encryption"] else set()), set(), "secrets")
if secrets["schema_version"] != 2:
raise Invalid("secrets schema_version must be 2")
user_password = text(secrets["user_password"], "secrets.user_password", 1024)
if len(user_password) < 8:
raise Invalid("user password must contain at least 8 characters")
encryption_password = None
if options["encryption"]:
encryption_password = text(secrets["encryption_password"], "secrets.encryption_password", 1024)
if len(encryption_password) < 8:
raise Invalid("encryption password must contain at least 8 characters")
package_re = re.compile(r"[a-z0-9][a-z0-9@._+:-]*\Z")
packages = []
seen = set()
for profile_id in profiles:
manifest_path = profile_dir / f"{profile_id}.json"
manifest = obj(json.loads(manifest_path.read_text()), f"profile {profile_id}")
if manifest.get("schema_version") != 2 or manifest.get("id") != profile_id:
raise Invalid(f"profile manifest {profile_id} has the wrong schema or ID")
listed = manifest.get("packages")
if not isinstance(listed, list) or any(type(p) is not str or not package_re.fullmatch(p) for p in listed):
raise Invalid(f"profile manifest {profile_id} has an invalid packages list")
for package in listed:
if package not in seen:
packages.append(package)
seen.add(package)
alignment = 1024 * 1024 // sector
esp_start = alignment
esp_length = 1024 * 1024 * 1024 // sector
root_start = esp_start + esp_length
root_start = ((root_start + alignment - 1) // alignment) * alignment
usable_end = (target["size"] - 1024 * 1024) // sector
usable_end = (usable_end // alignment) * alignment
root_length = usable_end - root_start
if root_length <= 0:
raise Invalid("target has no usable aligned root partition space")
config = obj(json.loads(template_path.read_text()), "archinstall template")
for forbidden in ("disk_config", "users", "auth_config", "encryption_password", "root_enc_password"):
if forbidden in config:
raise Invalid(f"archinstall template must not contain dynamic or secret field {forbidden}")
config["hostname"] = hostname
config["timezone"] = defaults["timezone"]
config["packages"] = packages
config["locale_config"] = {
"console_font": "default8x16",
"kb_layout": "",
"sys_enc": "UTF-8",
"sys_lang": defaults["locale"],
}
size = lambda value: {"unit": "sectors", "value": value, "sector_size": {"unit": "B", "value": sector}}
esp = {
"btrfs": [], "dev_path": None, "flags": ["boot", "esp"], "fs_type": "fat32",
"mount_options": [], "mountpoint": "/boot", "obj_id": "bifrost-efi",
"size": size(esp_length), "start": size(esp_start), "status": "create", "type": "primary",
}
root = {
"btrfs": [
{"mountpoint": "/", "name": "@"},
{"mountpoint": "/home", "name": "@home"},
{"mountpoint": "/var/log", "name": "@log"},
],
"dev_path": None, "flags": [], "fs_type": "btrfs", "mount_options": ["compress=zstd"],
"mountpoint": None, "obj_id": "bifrost-root", "size": size(root_length),
"start": size(root_start), "status": "create", "type": "primary",
}
disk_config = {"config_type": "default_layout", "device_modifications": [{"device": path, "wipe": True, "partitions": [esp, root]}]}
if options["encryption"]:
disk_config["disk_encryption"] = {"encryption_type": "luks", "partitions": ["bifrost-root"], "lvm_volumes": []}
config["disk_config"] = disk_config
password_hash = subprocess.run(["openssl", "passwd", "-6", "-stdin"], input=(user_password + "\n").encode(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True).stdout.decode().strip()
credentials = {"users": [{"username": username, "enc_password": password_hash, "sudo": True, "groups": []}]}
if encryption_password is not None:
credentials["encryption_password"] = encryption_password
required_source_packages = []
def require_package(package):
if isinstance(package, str) and package and package not in required_source_packages:
required_source_packages.append(package)
for package in (
"base", "sudo", "linux-firmware", "mkinitcpio", "btrfs-progs", "networkmanager",
"cosmic", "xdg-user-dirs", "cosmic-greeter", "amd-ucode", "intel-ucode",
"zram-generator", *config.get("kernels", []), *packages,
):
require_package(package)
app_config = config.get("app_config", {})
require_package(app_config.get("audio_config", {}).get("audio"))
for package in app_config.get("fonts_config", {}).get("fonts", []):
require_package(package)
require_package(app_config.get("power_management_config", {}).get("power_management"))
config_text = json.dumps(config, indent=2, ensure_ascii=False) + "\n"
plan = {
"schema_version": 2,
"target": {**target, "esp_start_sector": esp_start, "esp_length_sectors": esp_length, "root_start_sector": root_start, "root_length_sectors": root_length},
"options": {**options, "profiles": profiles, "system_defaults": {**defaults}},
"packages": packages,
"required_source_packages": required_source_packages,
"config_sha256": hashlib.sha256(config_text.encode()).hexdigest(),
}
plan_path.write_text(json.dumps(plan, indent=2, ensure_ascii=False) + "\n")
config_path.write_text(config_text)
creds_path.write_text(json.dumps(credentials, ensure_ascii=False) + "\n")
os.chmod(plan_path, 0o600)
os.chmod(config_path, 0o600)
os.chmod(creds_path, 0o600)
except (OSError, json.JSONDecodeError, subprocess.SubprocessError, Invalid) as error:
print(str(error), file=sys.stderr)
raise SystemExit(2)
PY
}
validate_target() {
udevadm settle
python3 - "$PLAN" "$GENERATED_CONFIG" "$MINIMUM_BYTES" <<'PY'
import hashlib
import json
import os
import re
import subprocess
import sys
from pathlib import Path
plan_path, config_path = map(Path, sys.argv[1:3])
plan = json.loads(plan_path.read_text())
expected = plan["target"]
minimum = int(sys.argv[3])
def run(*args):
return subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True).stdout
def canonical(path):
if not path:
return ""
value = str(path)
if not value.startswith("/"):
value = "/dev/" + value
return os.path.realpath(value)
def clean(value):
return "" if value is None else str(value).strip()
def flatten(items):
result = []
for item in items:
result.append(item)
result.extend(flatten(item.get("children", [])))
return result
def atomic_json(path, value):
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(json.dumps(value, indent=2, ensure_ascii=False) + "\n")
os.chmod(temporary, 0o600)
temporary.replace(path)
try:
data = json.loads(run("lsblk", "--json", "--bytes", "--paths", "--output", "PATH,KNAME,TYPE,SIZE,MODEL,SERIAL,WWN,LOG-SEC,RO,RM,PKNAME,MOUNTPOINTS"))
nodes = flatten(data.get("blockdevices", []))
by_path = {canonical(node["path"]): node for node in nodes if node.get("path")}
stable_path = expected.get("stable_path")
if stable_path:
stable = Path(stable_path)
if not stable.is_symlink() or not stable.exists():
raise RuntimeError("the selected stable hardware path no longer exists")
target_key = canonical(stable)
if target_key != expected.get("kernel_path"):
raise RuntimeError("the selected stable hardware path was retargeted")
else:
target_key = canonical(expected["path"])
target = by_path.get(target_key)
if not target or target.get("type") != "disk":
raise RuntimeError("selected target is no longer an installable whole disk")
comparisons = {
"path": clean(target.get("path")),
"model": clean(target.get("model")),
"serial": clean(target.get("serial")),
"wwn": clean(target.get("wwn")),
"size": int(target.get("size") or 0),
"logical_sector": int(target.get("log-sec") or 0),
}
expected_values = {
"path": expected["path"], "model": expected["model"].strip(), "serial": expected["serial"].strip(),
"wwn": expected["wwn"].strip(), "size": expected["size"], "logical_sector": expected["logical_sector"],
}
mismatches = [key for key in expected_values if comparisons[key] != expected_values[key]]
if mismatches:
raise RuntimeError("target identity changed: " + ", ".join(mismatches))
if not comparisons["serial"] and not comparisons["wwn"]:
raise RuntimeError("target has no stable serial or WWN")
if comparisons["size"] < minimum:
raise RuntimeError("target is below the 16 GiB minimum")
if bool(target.get("ro")):
raise RuntimeError("target is read-only")
if bool(target.get("rm")):
raise RuntimeError("removable targets are not supported")
def parent_key(node):
return canonical(node.get("pkname"))
def ancestors(key):
result = set()
while key and key not in result:
result.add(key)
node = by_path.get(key)
key = parent_key(node) if node else None
return result
descendants = {key for key in by_path if target_key in ancestors(key)}
topology = sorted(
(
clean(by_path[key].get("path")),
clean(by_path[key].get("type")),
int(by_path[key].get("size") or 0),
clean(by_path[key].get("pkname")),
)
for key in descendants
)
if "topology" in expected and topology != [tuple(item) for item in expected["topology"]]:
raise RuntimeError("target topology changed")
for key in descendants:
node = by_path[key]
if bool(node.get("ro")):
raise RuntimeError(f"target descendant is read-only: {node.get('path')}")
if any(mountpoint for mountpoint in (node.get("mountpoints") or [])):
raise RuntimeError(f"target or descendant is mounted: {node.get('path')}")
kname = os.path.basename(canonical(node.get("kname") or node.get("path") or ""))
holders = Path("/sys/class/block") / kname / "holders"
if holders.is_dir() and any(holders.iterdir()):
raise RuntimeError(f"target or descendant has active holders: {node.get('path')}")
swaps = Path("/proc/swaps").read_text().splitlines()[1:]
for line in swaps:
source = line.split()[0]
source_key = canonical(source)
if target_key in ancestors(source_key) or source_key in descendants:
raise RuntimeError(f"target backs active swap: {source}")
findmnt = json.loads(run("findmnt", "--json", "--output", "SOURCE,TARGET"))
filesystems = flatten(findmnt.get("filesystems", []))
forbidden_sources = []
for entry in filesystems:
mountpoint = entry.get("target") or ""
if mountpoint == "/" or mountpoint == "/mnt" or mountpoint.startswith(("/mnt/", "/run/archiso", "/run/miso", "/cdrom")):
source = (entry.get("source") or "").split("[", 1)[0]
if source.startswith("/dev/"):
forbidden_sources.append((source, mountpoint))
cmdline = Path("/proc/cmdline").read_text(errors="replace").split()
for token in cmdline:
if token.startswith(("archisodevice=", "img_dev=", "misobasedir=")) and "=" in token:
source = token.split("=", 1)[1]
if source.startswith("/dev/"):
forbidden_sources.append((source, "kernel command line"))
for source, use in forbidden_sources:
source_key = canonical(source)
source_ancestors = ancestors(source_key)
if target_key in source_ancestors or source_key in descendants:
raise RuntimeError(f"target backs the running system or live media ({use})")
node = by_path.get(source_key)
if node and node.get("type") == "loop":
backing = Path("/sys/class/block") / os.path.basename(source_key) / "loop/backing_file"
if backing.is_file():
backing_path = "/" + backing.read_text().strip().lstrip("/")
try:
backing_source = run("findmnt", "-n", "-o", "SOURCE", "--target", backing_path).strip().split("[", 1)[0]
if target_key in ancestors(canonical(backing_source)):
raise RuntimeError(f"target backs live-media loop data ({use})")
except subprocess.CalledProcessError:
pass
if not stable_path:
by_id = Path("/dev/disk/by-id")
if not by_id.is_dir():
raise RuntimeError("stable hardware-ID paths are unavailable")
def normalized(value):
return re.sub(r"[^a-z0-9]", "", value.lower())
candidates = []
for candidate in by_id.iterdir():
if re.search(r"-part[0-9]+$", candidate.name) or not candidate.is_symlink():
continue
if canonical(candidate) == target_key:
candidates.append(candidate)
wwn = comparisons["wwn"]
wwn_key = normalized(wwn).removeprefix("0x")
wwn_candidates = [
candidate for candidate in candidates
if candidate.name.lower().startswith("wwn-")
and normalized(candidate.name[4:]).removeprefix("0x") == wwn_key
] if wwn else []
if wwn and not wwn_candidates:
wwn_candidates = [
candidate for candidate in candidates
if candidate.name.lower().startswith(("nvme-eui.", "scsi-3"))
and wwn_key in normalized(candidate.name)
]
serial_key = normalized(comparisons["serial"])
serial_candidates = [
candidate for candidate in candidates
if serial_key and not candidate.name.lower().startswith("wwn-")
and serial_key in normalized(candidate.name)
]
matches = wwn_candidates if wwn_candidates else serial_candidates
if len(matches) != 1:
raise RuntimeError("target has no unique stable hardware-ID path consistent with its serial or WWN")
stable_path = str(matches[0])
expected["stable_path"] = stable_path
expected["kernel_path"] = target_key
expected["topology"] = topology
config = json.loads(config_path.read_text())
modifications = config.get("disk_config", {}).get("device_modifications", [])
if len(modifications) != 1 or modifications[0].get("device") != expected["path"]:
raise RuntimeError("generated disk configuration target is invalid")
modifications[0]["device"] = target_key
config_text = json.dumps(config, indent=2, ensure_ascii=False) + "\n"
temporary = config_path.with_suffix(config_path.suffix + ".tmp")
temporary.write_text(config_text)
os.chmod(temporary, 0o600)
temporary.replace(config_path)
plan["config_sha256"] = hashlib.sha256(config_text.encode()).hexdigest()
atomic_json(plan_path, plan)
except (OSError, ValueError, KeyError, subprocess.CalledProcessError, RuntimeError) as error:
print(str(error), file=sys.stderr)
raise SystemExit(1)
PY
}
required_packages() {
python3 - "$PLAN" <<'PY'
import json, sys
for package in json.load(open(sys.argv[1]))["required_source_packages"]:
print(package)
PY
}
source_mode() {
python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["options"]["source_mode"])' "$PLAN"
}
prepare_offline_mirror() {
if (( MIRRORLIST_CHANGED == 0 )); then
install -m 0644 "$MIRRORLIST" "$MIRRORLIST_BACKUP"
printf 'Server = file://%s/repo/$repo/os/$arch\n' "$OFFLINE_DIR" >"$MIRRORLIST"
chmod 0644 "$MIRRORLIST"
MIRRORLIST_CHANGED=1
fi
}
validate_offline_source() {
local manifest="$OFFLINE_DIR/manifest.json" signature="$OFFLINE_DIR/manifest.json.sig" keyring="$OFFLINE_DIR/trustedkeys.gpg"
[[ -f $manifest && -f $signature && -f $keyring ]] || fail offline_source_unavailable "Offline installation is unavailable: no signed complete local source manifest is installed."
command -v gpgv >/dev/null || fail missing_dependency "Offline verification requires gpgv."
gpgv --keyring "$keyring" "$signature" "$manifest" || fail offline_source_unverified "Offline source manifest signature verification failed."
if ! python3 - "$manifest" "$OFFLINE_DIR/repo" "$PLAN" <<'PY'
import hashlib, json, os, re, sys
from pathlib import Path
manifest_path, expected_root, plan_path = map(Path, sys.argv[1:])
data = json.loads(manifest_path.read_text())
if data.get("schema_version") != 2 or data.get("complete") is not True:
raise SystemExit("offline manifest is not schema v2 and complete")
root = Path(data.get("repository_root", ""))
if root != expected_root or not root.is_dir():
raise SystemExit("offline repository root is invalid")
entries = data.get("packages")
if not isinstance(entries, list) or not entries:
raise SystemExit("offline manifest package inventory is empty")
name_re = re.compile(r"[a-z0-9][a-z0-9@._+:-]*\Z")
sha_re = re.compile(r"[0-9a-f]{64}\Z")
names = set()
for entry in entries:
if not isinstance(entry, dict) or set(entry) != {"name", "version", "path", "sha256"}:
raise SystemExit("offline manifest package entry is invalid")
if not name_re.fullmatch(entry["name"]) or not isinstance(entry["version"], str) or not entry["version"]:
raise SystemExit("offline manifest package identity is invalid")
relative = Path(entry["path"])
if relative.is_absolute() or ".." in relative.parts or not sha_re.fullmatch(entry["sha256"]):
raise SystemExit("offline manifest package path or checksum is invalid")
package_path = root / relative
if not package_path.is_file() or os.path.commonpath((root.resolve(), package_path.resolve())) != str(root.resolve()):
raise SystemExit(f"offline package is missing: {relative}")
digest = hashlib.file_digest(package_path.open("rb"), "sha256").hexdigest()
if digest != entry["sha256"]:
raise SystemExit(f"offline package checksum failed: {relative}")
names.add(entry["name"])
required = set(json.loads(plan_path.read_text())["required_source_packages"])
missing = sorted(required - names)
if missing:
raise SystemExit("offline source is incomplete for this selection: " + ", ".join(missing))
for repo in ("core", "extra"):
directory = root / repo / "os" / "x86_64"
if not directory.is_dir() or not any(directory.glob("*.db")):
raise SystemExit(f"offline repository database is missing: {repo}")
PY
then
fail offline_source_unverified "Offline source verification failed; see the retained backend log."
fi
prepare_offline_mirror
}
validate_package_source() {
local mode=$1
local pacman_db="$RUNTIME_DIR/pacman-db"
mapfile -t packages < <(required_packages)
if [[ $mode == offline ]]; then
validate_offline_source
fi
chmod 0711 "$RUNTIME_DIR"
install -d -o alpm -g alpm -m 0700 "$pacman_db" "$RUNTIME_DIR/pacman-cache"
pacman --dbpath "$pacman_db" --cachedir "$RUNTIME_DIR/pacman-cache" --logfile "$RUNTIME_DIR/pacman.log" -Sy --noconfirm || {
if [[ $mode == offline ]]; then
fail offline_source_unverified "The verified local package databases could not be synchronized."
else
fail online_source_unavailable "Arch package repositories are not reachable."
fi
}
pacman --dbpath "$pacman_db" --cachedir "$RUNTIME_DIR/pacman-cache" --logfile "$RUNTIME_DIR/pacman.log" -Sp --noconfirm --print-format '%l' "${packages[@]}" >"$RUNTIME_DIR/package-urls" || {
if [[ $mode == offline ]]; then
fail offline_source_incomplete "The verified local source cannot resolve every package needed by the uninstalled target."
else
fail online_source_incomplete "The online Arch sources cannot resolve every package needed by the uninstalled target."
fi
}
python3 - "$RUNTIME_DIR/package-urls" "$mode" <<'PY' || {
from concurrent.futures import ThreadPoolExecutor
import sys
from pathlib import Path
from urllib.request import Request, urlopen
urls = list(dict.fromkeys(line.strip() for line in Path(sys.argv[1]).read_text().splitlines() if "://" in line))
if not urls:
raise SystemExit("package resolution returned no source URLs")
if sys.argv[2] == "offline":
for url in urls:
if not url.startswith("file://"):
raise SystemExit("offline package resolution attempted a network URL")
if not Path(url[7:]).is_file():
raise SystemExit("an offline package URL is missing")
else:
if any(not url.startswith(("https://", "http://")) for url in urls):
raise SystemExit("online package resolution returned a non-network URL")
def probe(url):
request = Request(url, headers={"Range": "bytes=0-0", "User-Agent": "BifrOSt-installer/0.2"})
with urlopen(request, timeout=20) as response:
if response.status >= 400 or not response.read(1):
raise RuntimeError("online package source probe failed")
with ThreadPoolExecutor(max_workers=min(8, len(urls))) as executor:
list(executor.map(probe, urls))
PY
if [[ $mode == offline ]]; then
fail offline_source_incomplete "The verified local source failed its complete package URL check."
else
fail online_source_unavailable "At least one online package URL was unavailable."
fi
}
chmod 0700 "$RUNTIME_DIR"
}
sanitize_archinstall_stream() {
python3 -c '
import json
import sys
def secret_values(value, key=""):
if isinstance(value, dict):
for child_key, child in value.items():
yield from secret_values(child, child_key)
elif isinstance(value, list):
for child in value:
yield from secret_values(child, key)
elif isinstance(value, str) and "password" in key.lower() and value:
yield value
try:
secrets = json.load(open(sys.argv[1]))
except Exception:
secrets = {}
values = sorted(set(secret_values(secrets)), key=len, reverse=True)
with open(sys.argv[2], "a") as output:
for line in sys.stdin:
for secret in values:
line = line.replace(secret, "[REDACTED]")
output.write(line)
output.flush()
' "$SECRETS_INPUT" "$RAW_LOG"
}
archinstall_command() {
local mode=$1
local -a command=(archinstall --config "$GENERATED_CONFIG" --creds "$GENERATED_CREDS" --silent)
[[ $mode == offline ]] && command+=(--offline)
"${command[@]}" "${@:2}" 2>&1 | sanitize_archinstall_stream
}
apply_installed_experience() {
[[ -d $INSTALLED_ROOT && -d $INSTALLED_ROOT/usr ]] || fail installed_assets_missing "Installed-system assets are missing from the live image."
cp -a "$INSTALLED_ROOT/." "$TARGET_MOUNT/"
install -Dm644 /usr/share/bifrost/os-release "$TARGET_MOUNT/usr/lib/os-release"
ln -sfn ../usr/lib/os-release "$TARGET_MOUNT/etc/os-release"
install -Dm644 /usr/share/backgrounds/bifrost/bifrost-aurora.png \
"$TARGET_MOUNT/usr/share/backgrounds/bifrost/bifrost-aurora.png"
install -Dm644 /usr/share/bifrost/branding/bifrost-mark.svg \
"$TARGET_MOUNT/usr/share/bifrost/branding/bifrost-mark.svg"
install -Dm644 /usr/share/icons/hicolor/scalable/apps/bifrost.svg \
"$TARGET_MOUNT/usr/share/icons/hicolor/scalable/apps/bifrost.svg"
install -Dm644 /usr/share/bifrost/cosmic-background.ron \
"$TARGET_MOUNT/usr/share/cosmic/com.system76.CosmicBackground/v1/all"
install -d -m 0755 "$TARGET_MOUNT/usr/share/plymouth/themes"
cp -a /usr/share/plymouth/themes/bifrost "$TARGET_MOUNT/usr/share/plymouth/themes/"
install -Dm644 /etc/plymouth/plymouthd.conf "$TARGET_MOUNT/etc/plymouth/plymouthd.conf"
install -Dm644 /etc/systemd/system/plymouth-quit.service.d/bifrost.conf \
"$TARGET_MOUNT/etc/systemd/system/plymouth-quit.service.d/bifrost.conf"
install -d -m 0755 "$TARGET_MOUNT/etc/bifrost"
python3 - "$PLAN" "$TARGET_MOUNT/etc/bifrost/install-state.json" <<'PY'
import json, sys
from pathlib import Path
plan = json.loads(Path(sys.argv[1]).read_text())
state = {"schema_version": 2, "encryption": plan["options"]["encryption"], "profiles": plan["options"]["profiles"]}
Path(sys.argv[2]).write_text(json.dumps(state, indent=2, ensure_ascii=False) + "\n")
PY
chmod 0644 "$TARGET_MOUNT/etc/bifrost/install-state.json"
[[ -f $TARGET_MOUNT/etc/os-release ]] || fail branding_missing "Installed BifrOSt identity is missing."
grep -qx 'ID=bifrost' "$TARGET_MOUNT/etc/os-release" || fail branding_invalid "Installed BifrOSt identity is invalid."
local username full_name keyboard_layout
username=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["options"]["system_defaults"]["username"])' "$PLAN")
full_name=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["options"]["system_defaults"]["full_name"])' "$PLAN")
keyboard_layout=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["options"]["system_defaults"]["keyboard_layout"])' "$PLAN")
printf 'KEYMAP=%s\nFONT=default8x16\n' "$keyboard_layout" >"$TARGET_MOUNT/etc/vconsole.conf"
if [[ -n $full_name ]]; then
arch-chroot "$TARGET_MOUNT" chfn -f "$full_name" "$username"
fi
if (( MIRRORLIST_CHANGED == 1 )); then
install -m 0644 "$MIRRORLIST_BACKUP" "$TARGET_MOUNT/etc/pacman.d/mirrorlist"
fi
python3 - "$TARGET_MOUNT/etc/mkinitcpio.conf" <<'PY'
import re, sys
from pathlib import Path
path = Path(sys.argv[1])
text = path.read_text()
match = re.search(r"(?m)^HOOKS=\(([^)]*)\)", text)
if not match:
raise SystemExit("mkinitcpio HOOKS is missing")
hooks = match.group(1).split()
if "plymouth" not in hooks:
for anchor in ("encrypt", "sd-encrypt", "filesystems"):
if anchor in hooks:
hooks.insert(hooks.index(anchor), "plymouth")
break
else:
hooks.append("plymouth")
text = text[:match.start()] + "HOOKS=(" + " ".join(hooks) + ")" + text[match.end():]
path.write_text(text)
PY
arch-chroot "$TARGET_MOUNT" mkinitcpio -P
}
verify_installation() {
mountpoint -q "$TARGET_MOUNT" || fail target_not_mounted "The installed system is not mounted for verification."
[[ -d $TARGET_MOUNT/etc && -d $TARGET_MOUNT/boot ]] || fail target_incomplete "The installed system layout is incomplete."
bootctl --root="$TARGET_MOUNT" --esp-path=/boot is-installed >/dev/null || fail bootloader_missing "systemd-boot is not installed on the EFI system partition."
[[ -f $TARGET_MOUNT/boot/EFI/systemd/systemd-bootx64.efi || -f $TARGET_MOUNT/boot/EFI/BOOT/BOOTX64.EFI ]] || fail bootloader_missing "No systemd-boot EFI binary was found."
local kernel
for kernel in linux linux-lts; do
[[ -s $TARGET_MOUNT/boot/vmlinuz-$kernel ]] || fail kernel_missing "Installed kernel is missing: $kernel"
[[ -s $TARGET_MOUNT/boot/initramfs-$kernel.img ]] || fail initramfs_missing "Installed initramfs is missing: $kernel"
done
shopt -s nullglob
local entries=("$TARGET_MOUNT"/boot/loader/entries/*.conf)
(( ${#entries[@]} > 0 )) || fail boot_entry_missing "No systemd-boot loader entries were generated."
grep -qE '^linux[[:space:]]+/vmlinuz-linux([[:space:]]|$)' "${entries[@]}" || fail boot_entry_invalid "The linux loader entry is missing its kernel."
grep -qE '^initrd[[:space:]]+/initramfs-linux\.img([[:space:]]|$)' "${entries[@]}" || fail boot_entry_invalid "The linux loader entry is missing its initramfs."
grep -qE '^linux[[:space:]]+/vmlinuz-linux-lts([[:space:]]|$)' "${entries[@]}" || fail boot_entry_invalid "The linux-lts loader entry is missing its kernel."
grep -qE '^initrd[[:space:]]+/initramfs-linux-lts\.img([[:space:]]|$)' "${entries[@]}" || fail boot_entry_invalid "The linux-lts loader entry is missing its initramfs."
local encrypted
encrypted=$(python3 -c 'import json,sys; print("1" if json.load(open(sys.argv[1]))["options"]["encryption"] else "0")' "$PLAN")
if [[ $encrypted == 1 ]]; then
grep -Eq '^HOOKS=.*(^|[[:space:]])encrypt([[:space:]]|\))' "$TARGET_MOUNT/etc/mkinitcpio.conf" || fail encryption_initramfs_invalid "Encrypted install initramfs lacks the encrypt hook."
grep -Eq '^options .*cryptdevice=(UUID|PARTUUID)=[^[:space:]]+:root.*root=/dev/mapper/root' "${entries[@]}" || fail encryption_boot_invalid "Encrypted root kernel parameters are missing or invalid."
local target_path luks_found=0 device kind fstype rest
target_path=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["target"]["path"])' "$PLAN")
while read -r device kind fstype rest; do
if [[ $kind == part && $fstype == crypto_LUKS ]] \
&& cryptsetup isLuks "$device" \
&& cryptsetup luksDump "$device" | grep -qE '^Version:[[:space:]]*2$'; then
luks_found=1
fi
done < <(lsblk -nrpo PATH,TYPE,FSTYPE "$target_path")
(( luks_found == 1 )) || fail encryption_layout_invalid "The root partition is not a verified LUKS2 container."
else
local target_path btrfs_found=0 device kind fstype rest
target_path=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["target"]["path"])' "$PLAN")
while read -r device kind fstype rest; do
if [[ $kind == part && $fstype == btrfs ]]; then
btrfs_found=1
fi
done < <(lsblk -nrpo PATH,TYPE,FSTYPE "$target_path")
(( btrfs_found == 1 )) || fail filesystem_layout_invalid "The standard root partition is not Btrfs."
fi
}
phase preflight "Validating request, UEFI environment, secrets, and selected profiles."
validate_input_file "$INTENT_INPUT" "Intent file"
validate_input_file "$SECRETS_INPUT" "Secrets file"
validate_environment
build_plan_and_configs || fail invalid_request "Schema-v2 intent, secrets, template, or profile validation failed; see the retained backend log."
validate_target || fail target_refused "The selected disk failed identity or safety validation; no disk was wiped."
readonly SOURCE_MODE=$(source_mode)
phase source "Verifying the complete package source before disk changes."
validate_package_source "$SOURCE_MODE"
phase plan "Validating the generated archinstall disk and encryption configuration."
archinstall_command "$SOURCE_MODE" --dry-run || fail archinstall_config_invalid "archinstall rejected the generated configuration before disk changes."
phase apply "Revalidating the target immediately before the whole-disk install."
validate_environment
validate_package_source "$SOURCE_MODE"
validate_target || fail target_changed "The selected disk changed or became active; no disk was wiped."
write_status applying "" "All preflight checks passed; whole-disk installation is starting."
WIPE_STARTED=1
archinstall_command "$SOURCE_MODE" || fail archinstall_failed "archinstall failed after disk operations began; the disk may contain a partial installation."
if [[ -f /var/log/archinstall/install.log ]]; then
redact_file /var/log/archinstall/install.log "$RUN_DIR/archinstall.log"
fi
mountpoint -q "$TARGET_MOUNT" || fail target_not_mounted "archinstall finished without a mounted installed system."
phase postinstall "Copying installed-system assets and applying BifrOSt identity."
apply_installed_experience
phase verify "Verifying systemd-boot, kernels, initramfs, and encryption state."
verify_installation
FINAL_SUCCESS=1
exit 0