AkurAI Build
Menu

BifrOSt

public

Latest change ab72e60225a1ca626906da92e44d0b33b1a35203 - Populate target package keyring 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,
    "log_dir": "/var/log/bifrost-installer",
    "live_log_dir": "/var/log/bifrost-installer",
    "target_log_dir": None,
}, 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"
readonly TARGET_LOG_DIR="$LOG_ROOT/$RUN_ID"
readonly RELEASE_PATH=/usr/share/bifrost/release.json
readonly BIFROST_ALPM_DIR=/usr/share/bifrost/alpm
readonly BIFROST_REPOSITORY_INCLUDE=/etc/pacman.conf.d/bifrost.conf
BIFROST_PACKAGE=

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
TARGET_EVIDENCE_PERSISTED=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" "$GENERATED_CREDS" "$source" "$destination" <<'PY'
import json, re, sys
from pathlib import Path
secrets_path, credentials_path, source_path, destination_path = map(Path, sys.argv[1:])
text = source_path.read_text(errors="replace")

def load(path):
    try:
        return json.loads(path.read_text())
    except Exception:
        return {}

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

sensitive = set(values(load(secrets_path))) | set(values(load(credentials_path)))
for secret in sorted(sensitive, key=len, reverse=True):
    text = text.replace(secret, "[REDACTED]")
text = re.sub(r'(?i)((?:enc_)?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
}

rollback_target_evidence() {
    mountpoint -q "$TARGET_MOUNT" || return 0
    TARGET_ROOT=$TARGET_MOUNT TARGET_RECORD=$TARGET_LOG_DIR RUN_ID_VALUE=$RUN_ID python3 - <<'PY'
import json
import os
import shutil
from pathlib import Path

root = Path(os.environ["TARGET_ROOT"])
evidence = root / Path(os.environ["TARGET_RECORD"]).relative_to("/")
state = root / "etc/bifrost/install-state.json"
try:
    recorded = json.loads(state.read_text(encoding="utf-8"))
except (OSError, ValueError):
    recorded = None
if isinstance(recorded, dict) and recorded.get("run_id") == os.environ["RUN_ID_VALUE"]:
    state.unlink(missing_ok=True)
shutil.rmtree(evidence, ignore_errors=True)
PY
}

on_exit() {
    local rc=$?
    trap - EXIT INT TERM
    set +e
    if (( rc != 0 || FINAL_SUCCESS != 1 )); then
        if (( WIPE_STARTED == 1 )) && [[ -f /var/log/archinstall/install.log ]]; then
            redact_file /var/log/archinstall/install.log "$RUN_DIR/archinstall.log"
        fi
        CURRENT_PHASE=${CURRENT_PHASE:-failed}
        write_status failed "$ERROR_CODE" "$ERROR_MESSAGE"
        redact_file "$RAW_LOG" "$RUN_DIR/backend.log"
        rollback_target_evidence
    fi
    cleanup_target
    if (( MIRRORLIST_CHANGED == 1 )) && [[ -f $MIRRORLIST_BACKUP ]]; then
        install -m 0644 "$MIRRORLIST_BACKUP" "$MIRRORLIST"
    fi
    rm -rf "$RUNTIME_DIR"

    if (( rc == 0 && FINAL_SUCCESS == 1 && TARGET_EVIDENCE_PERSISTED == 1 )); then
        EVENT_LIVE_LOG=$RUN_DIR EVENT_TARGET_LOG=$TARGET_LOG_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, locale state, and persistent evidence were verified.",
    "log_dir": os.environ["EVENT_TARGET_LOG"],
    "live_log_dir": os.environ["EVENT_LIVE_LOG"],
    "target_log_dir": os.environ["EVENT_TARGET_LOG"],
    "wiped": True,
}, ensure_ascii=False), flush=True)
PY
    else
        EVENT_CODE=$ERROR_CODE EVENT_MESSAGE=$ERROR_MESSAGE EVENT_LIVE_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_LIVE_LOG"],
    "live_log_dir": os.environ["EVENT_LIVE_LOG"],
    "target_log_dir": None,
    "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 gpg gpgv 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")
        actual_number = os.stat(target_key).st_rdev
        if [os.major(actual_number), os.minor(actual_number)] != expected.get("device_number"):
            raise RuntimeError("the selected target device number changed")
    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
        target_stat = os.stat(target_key)
        expected["device_number"] = [os.major(target_stat.st_rdev), os.minor(target_stat.st_rdev)]
        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")
        # Archinstall discovers block devices by their canonical kernel paths and
        # silently ignores a /dev/disk/by-id symlink here. Keep the canonical
        # path in its schema; the frozen stable ID, device number, topology, and
        # geometry are revalidated immediately before this configuration runs.
        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

def load(path):
    try:
        return json.load(open(path))
    except Exception:
        return {}

values = sorted(set(secret_values(load(sys.argv[1]))) | set(secret_values(load(sys.argv[2]))),
                key=len, reverse=True)
with open(sys.argv[3], "a") as output:
    for line in sys.stdin:
        for secret in values:
            line = line.replace(secret, "[REDACTED]")
        output.write(line)
        output.flush()
' "$SECRETS_INPUT" "$GENERATED_CREDS" "$RAW_LOG"
}

archinstall_command() {
    local mode=$1 wrapper_dir="$RUNTIME_DIR/archinstall-path"
    install -d -m 0700 "$wrapper_dir"
    cat >"$wrapper_dir/pacstrap" <<'SH'
#!/usr/bin/bash
set -Eeuo pipefail
declare -a forwarded=()
for argument in "$@"; do
    [[ $argument == -K ]] || forwarded+=("$argument")
done
# Archinstall creates this directory before pacstrap. Remove it only if it is
# still empty so pacstrap's default host-keyring copy path can run.
rmdir -- /mnt/etc/pacman.d/gnupg 2>/dev/null || true
exec /usr/bin/pacstrap "${forwarded[@]}"
SH
    chmod 0700 "$wrapper_dir/pacstrap"
    local -a command=(archinstall --config "$GENERATED_CONFIG" --creds "$GENERATED_CREDS" --silent)
    [[ $mode == offline ]] && command+=(--offline)
    # Archinstall 3.0 always asks pacstrap for an empty target keyring. With
    # pacman 7 this cannot authenticate the initial package set. The constrained
    # wrapper removes only -K so pacstrap copies the already-populated, signed
    # live Arch keyring into the target; repository/package signatures remain
    # mandatory and are rechecked by pacman.
    PATH="$wrapper_dir:$PATH" "${command[@]}" "${@:2}" 2>&1 | sanitize_archinstall_stream
}

validate_bifrost_package() {
    python3 - "$BIFROST_ALPM_DIR" <<'PY'
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys

root = Path(sys.argv[1])
manifest_path = root / "manifest.json"
manifest_signature = root / "manifest.json.sig"
keyring = root / "alpm-repository-key.gpg"
public_key = root / "alpm-repository-key.asc"
fingerprint_path = root / "alpm-repository-key.fingerprint"
for required in (manifest_path, manifest_signature, keyring, public_key, fingerprint_path):
    if not required.is_file():
        raise SystemExit(f"signed bifrost-system bootstrap input is missing: {required.name}")

manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
fingerprint = fingerprint_path.read_text(encoding="ascii").strip().upper()
live_identity = {}
for line in Path("/usr/share/bifrost/os-release").read_text(encoding="utf-8").splitlines():
    if "=" in line:
        key, value = line.split("=", 1)
        live_identity[key] = value.strip().strip('"')
if (
    manifest.get("schema_version") != 1
    or manifest.get("repository") != "bifrost-system"
    or manifest.get("architecture") != "any"
    or manifest.get("signing_fingerprint") != fingerprint
    or not re.fullmatch(r"[0-9A-F]{40}", fingerprint)
    or not re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+", str(manifest.get("version", "")))
    or manifest.get("version") != live_identity.get("VERSION_ID")
    or not re.fullmatch(
        rf"{re.escape(str(manifest.get('version', '')))}-[1-9][0-9]*",
        str(manifest.get("package_version", "")),
    )
):
    raise SystemExit("bifrost-system bootstrap manifest identity is invalid")

def valid_signature(path, signature):
    result = subprocess.run(
        ["gpgv", "--keyring", str(keyring), "--status-fd", "1", str(signature), str(path)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
        text=True,
    )
    def signing_identity(line):
        fields = line.split()
        if not line.startswith("[GNUPG:] VALIDSIG ") or len(fields) <= 2:
            return None
        primary = fields[-1].upper()
        if len(fields) > 11 and re.fullmatch(r"[0-9A-F]{40}", primary):
            return primary
        return fields[2].upper()

    valid = {
        identity
        for line in result.stdout.splitlines()
        if (identity := signing_identity(line)) is not None
    }
    if result.returncode or valid != {fingerprint}:
        raise SystemExit(f"signature verification failed: {signature.name}")

valid_signature(manifest_path, manifest_signature)
shown = subprocess.run(
    ["gpg", "--batch", "--with-colons", "--show-keys", str(public_key)],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    check=False,
    text=True,
)
primary = []
waiting = False
for line in shown.stdout.splitlines():
    fields = line.split(":")
    kind = fields[0] if fields else ""
    if kind == "pub":
        waiting = True
    elif waiting and kind == "fpr" and len(fields) > 9:
        primary.append(fields[9].upper())
        waiting = False
    elif kind in {"pub", "sub"}:
        waiting = False
if shown.returncode or primary != [fingerprint]:
    raise SystemExit("armored bootstrap key does not match the pinned signing fingerprint")

packages = sorted(root.glob("bifrost-system-*.pkg.tar.zst"))
if len(packages) != 1:
    raise SystemExit("bootstrap must contain exactly one bifrost-system package")
package = packages[0]
package_signature = package.with_name(package.name + ".sig")
files = manifest.get("files")
if not isinstance(files, dict):
    raise SystemExit("bootstrap manifest file inventory is invalid")
for path in (package, package_signature):
    record = files.get(path.name)
    if (
        not path.is_file()
        or not isinstance(record, dict)
        or set(record) != {"sha256", "size"}
        or record["size"] != path.stat().st_size
        or record["sha256"] != hashlib.file_digest(path.open("rb"), "sha256").hexdigest()
    ):
        raise SystemExit(f"bootstrap package inventory failed: {path.name}")
valid_signature(package, package_signature)

metadata = subprocess.run(
    ["bsdtar", "-xOf", str(package), ".PKGINFO"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    check=False,
    text=True,
)
values = {}
for line in metadata.stdout.splitlines():
    key, separator, value = line.partition(" = ")
    if separator and key in {"pkgname", "pkgver"}:
        if key in values:
            raise SystemExit(f"bootstrap package metadata repeats {key}")
        values[key] = value
if (
    metadata.returncode
    or values.get("pkgname") != "bifrost-system"
    or values.get("pkgver") != manifest["package_version"]
):
    raise SystemExit("bootstrap package identity does not match its signed manifest")

release_result = subprocess.run(
    ["bsdtar", "-xOf", str(package), "usr/share/bifrost/release.json"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    check=False,
    text=True,
)
try:
    release = json.loads(release_result.stdout)
except json.JSONDecodeError as error:
    raise SystemExit("bootstrap package has invalid installed release provenance") from error
revision = release.get("source_revision")
epoch = release.get("source_date_epoch")
expected_build_id = (
    f"bifrost-{manifest['version']}-{revision[:12]}-{epoch}"
    if isinstance(revision, str) and isinstance(epoch, int) and not isinstance(epoch, bool)
    else None
)
iso = release.get("iso")
if (
    release_result.returncode
    or release.get("schema_version") != 1
    or release.get("provenance_status") != "build-input"
    or release.get("version") != manifest["version"]
    or not isinstance(revision, str)
    or not re.fullmatch(r"[0-9a-f]{40}", revision)
    or not isinstance(epoch, int)
    or isinstance(epoch, bool)
    or epoch <= 0
    or release.get("build_id") != expected_build_id
    or not re.fullmatch(r"[0-9a-f]{64}", str(release.get("profile_sha256", "")))
    or release.get("profile_digest_excludes") != [
        "airootfs/usr/share/bifrost/installed-root/usr/share/bifrost/release.json",
        "airootfs/usr/share/bifrost/alpm/**",
    ]
    or not isinstance(iso, dict)
    or iso.get("file") != f"bifrost-{manifest['version']}-x86_64.iso"
    or iso.get("bytes") is not None
    or iso.get("sha256") is not None
):
    raise SystemExit("bootstrap package release provenance is not a prepared build input")
print(package)
PY
}

bootstrap_bifrost_system() {
    [[ -n $BIFROST_PACKAGE && -f $BIFROST_PACKAGE ]] \
        || fail installed_assets_missing "The verified bifrost-system bootstrap package is unavailable."
    local bootstrap_dir="$TARGET_MOUNT/run/bifrost-bootstrap"
    local package_name fingerprint
    package_name=${BIFROST_PACKAGE##*/}
    fingerprint=$(<"$BIFROST_ALPM_DIR/alpm-repository-key.fingerprint")
    install -d -m 0700 "$bootstrap_dir"
    install -m 0644 "$BIFROST_PACKAGE" "$bootstrap_dir/$package_name"
    install -m 0644 "$BIFROST_PACKAGE.sig" "$bootstrap_dir/$package_name.sig"
    install -m 0644 "$BIFROST_ALPM_DIR/alpm-repository-key.asc" "$bootstrap_dir/repository-key.asc"

    arch-chroot "$TARGET_MOUNT" pacman-key --add /run/bifrost-bootstrap/repository-key.asc \
        || fail package_trust_failed "The pinned BifrOSt repository key could not be imported."
    arch-chroot "$TARGET_MOUNT" pacman-key --lsign-key "$fingerprint" \
        || fail package_trust_failed "The pinned BifrOSt repository key could not be locally trusted."

    TARGET_ROOT=$TARGET_MOUNT python3 - <<'PY'
import os
from pathlib import Path

root = Path(os.environ["TARGET_ROOT"])
source = root / "etc/pacman.conf"
destination = root / "run/bifrost-bootstrap/pacman.conf"
lines = source.read_text(encoding="utf-8").splitlines()
result = []
in_options = False
set_local_level = False
for line in lines:
    stripped = line.strip()
    if stripped.startswith("[") and stripped.endswith("]"):
        if in_options and not set_local_level:
            result.append("LocalFileSigLevel = Required")
            set_local_level = True
        in_options = stripped == "[options]"
    if in_options and stripped.startswith("LocalFileSigLevel"):
        if not set_local_level:
            result.append("LocalFileSigLevel = Required")
            set_local_level = True
        continue
    result.append(line)
if in_options and not set_local_level:
    result.append("LocalFileSigLevel = Required")
if not set_local_level:
    raise SystemExit("[options] is missing from the installed pacman configuration")
destination.write_text("\n".join(result) + "\n", encoding="utf-8")
PY
    arch-chroot "$TARGET_MOUNT" pacman --config /run/bifrost-bootstrap/pacman.conf \
        -U --noconfirm "/run/bifrost-bootstrap/$package_name" \
        || fail package_install_failed "The signed bifrost-system bootstrap package could not be registered."

    [[ -f $TARGET_MOUNT$BIFROST_REPOSITORY_INCLUDE ]] \
        || fail repository_config_missing "The bifrost-system package did not install its repository configuration."
    grep -qx 'SigLevel = PackageRequired DatabaseRequired' "$TARGET_MOUNT$BIFROST_REPOSITORY_INCLUDE" \
        || fail repository_config_invalid "The BifrOSt repository does not require package and database signatures."
    grep -qx '\[bifrost\]' "$TARGET_MOUNT$BIFROST_REPOSITORY_INCLUDE" \
        || fail repository_config_invalid "The installed package repository is not narrowly named bifrost."
    grep -qx 'Usage = Sync Search Upgrade' "$TARGET_MOUNT$BIFROST_REPOSITORY_INCLUDE" \
        || fail repository_config_invalid "The BifrOSt repository is not restricted to synchronization, search, and upgrades."
    grep -Fqx 'Server = https://olibuijr.github.io/BifrOSt/alpm/$arch' "$TARGET_MOUNT$BIFROST_REPOSITORY_INCLUDE" \
        || fail repository_config_invalid "The BifrOSt repository URL is not the pinned HTTPS endpoint."
    TARGET_ROOT=$TARGET_MOUNT python3 - <<'PY'
import os
from pathlib import Path

path = Path(os.environ["TARGET_ROOT"]) / "etc/pacman.conf"
include = "Include = /etc/pacman.conf.d/bifrost.conf"
lines = path.read_text(encoding="utf-8").splitlines()
if include not in lines:
    lines.extend(("", "# Signed BifrOSt system package repository; official Arch repositories remain first.", include))
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
PY
    rm -rf -- "$bootstrap_dir"
}

apply_installed_experience() {
    arch-chroot "$TARGET_MOUNT" pacman -Q bifrost-system >/dev/null \
        || fail installed_assets_missing "The installed bifrost-system package is missing."
    local packaged_asset
    for packaged_asset in \
        /usr/share/bifrost/os-release \
        /usr/share/backgrounds/bifrost/bifrost-aurora.png \
        /usr/share/bifrost/branding/bifrost-mark.svg \
        /usr/share/icons/hicolor/scalable/apps/bifrost.svg \
        /usr/share/cosmic/com.system76.CosmicBackground/v1/all \
        /usr/share/plymouth/themes/bifrost/bifrost.plymouth \
        /etc/plymouth/plymouthd.conf \
        /etc/systemd/system/plymouth-quit.service.d/bifrost.conf
    do
        [[ -e $TARGET_MOUNT$packaged_asset ]] \
            || fail installed_assets_missing "The bifrost-system package omitted $packaged_asset."
    done
    ln -snf ../usr/share/bifrost/os-release "$TARGET_MOUNT/etc/os-release"
    install -d -m 0755 "$TARGET_MOUNT/etc/bifrost"

    [[ -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 selected_locale
    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")
    selected_locale=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["options"]["system_defaults"]["locale"])' "$PLAN")
    SELECTED_LOCALE=$selected_locale TARGET_ROOT=$TARGET_MOUNT python3 - <<'PY'
import os
from pathlib import Path

root = Path(os.environ["TARGET_ROOT"])
selected = os.environ["SELECTED_LOCALE"]
locales = list(dict.fromkeys(("is_IS.UTF-8", selected)))
(root / "etc/locale.gen").write_text("".join(f"{item} UTF-8\n" for item in locales), encoding="utf-8")
languages = list(dict.fromkeys((selected.split("_", 1)[0], "is", "en")))
(root / "etc/locale.conf").write_text(
    f"LANG={selected}\nLANGUAGE={':'.join(languages)}\n", encoding="utf-8"
)
PY
    chmod 0644 "$TARGET_MOUNT/etc/locale.gen" "$TARGET_MOUNT/etc/locale.conf"
    printf 'KEYMAP=%s\nFONT=default8x16\n' "$keyboard_layout" >"$TARGET_MOUNT/etc/vconsole.conf"
    chmod 0644 "$TARGET_MOUNT/etc/vconsole.conf"
    arch-chroot "$TARGET_MOUNT" locale-gen
    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."
    arch-chroot "$TARGET_MOUNT" pacman -Qk bifrost-system >/dev/null \
        || fail package_ownership_invalid "The installed bifrost-system payload is incomplete."
    arch-chroot "$TARGET_MOUNT" pacman -Qo \
        /usr/bin/bifrost-recovery-info \
        /usr/bin/bifrost-welcome \
        /usr/share/applications/org.bifrost.RecoveryInfo.desktop \
        /etc/xdg/autostart/org.bifrost.Welcome.desktop \
        /usr/share/bifrost/os-release \
        /usr/share/backgrounds/bifrost/bifrost-aurora.png \
        /usr/share/bifrost/branding/bifrost-mark.svg \
        /usr/share/icons/hicolor/scalable/apps/bifrost.svg \
        /usr/share/cosmic/com.system76.CosmicBackground/v1/all \
        /usr/share/plymouth/themes/bifrost/bifrost.plymouth \
        /etc/plymouth/plymouthd.conf \
        /etc/systemd/system/plymouth-quit.service.d/bifrost.conf >/dev/null \
        || fail package_ownership_invalid "Critical BifrOSt executables or desktop assets are not package-owned."
    arch-chroot "$TARGET_MOUNT" pacman -Qlq bifrost-system >"$RUNTIME_DIR/bifrost-owned-paths" \
        || fail package_ownership_invalid "The bifrost-system ownership inventory is unavailable."
    INSTALLED_SOURCE=$INSTALLED_ROOT OWNED_PATHS="$RUNTIME_DIR/bifrost-owned-paths" python3 - <<'PY' \
        || fail package_ownership_invalid "Not every BifrOSt installed-root payload file is owned by bifrost-system."
import os
from pathlib import Path

source = Path(os.environ["INSTALLED_SOURCE"])
owned = {
    line.rstrip("/")
    for line in Path(os.environ["OWNED_PATHS"]).read_text(encoding="utf-8").splitlines()
}
required = {
    "/" + path.relative_to(source).as_posix()
    for path in source.rglob("*")
    if (path.is_file() or path.is_symlink())
    and "__pycache__" not in path.parts
    and path.suffix not in {".pyc", ".pyo"}
}
missing = sorted(required - owned)
if missing:
    raise SystemExit("unowned installed payload: " + ", ".join(missing))
PY

    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."
    local linux_entry=0 linux_lts_entry=0 entry
    for entry in "${entries[@]}"; do
        if grep -qE '^linux[[:space:]]+/vmlinuz-linux([[:space:]]|$)' "$entry" \
            && grep -qE '^initrd[[:space:]]+/initramfs-linux\.img([[:space:]]|$)' "$entry"; then
            linux_entry=1
        fi
        if grep -qE '^linux[[:space:]]+/vmlinuz-linux-lts([[:space:]]|$)' "$entry" \
            && grep -qE '^initrd[[:space:]]+/initramfs-linux-lts\.img([[:space:]]|$)' "$entry"; then
            linux_lts_entry=1
        fi
    done
    (( linux_entry == 1 )) || fail boot_entry_invalid "No single linux loader entry contains both its kernel and initramfs."
    (( linux_lts_entry == 1 )) || fail boot_entry_invalid "No single linux-lts loader entry contains both its kernel and 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
    local expected_locale expected_keymap locale_inventory
    expected_locale=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["options"]["system_defaults"]["locale"])' "$PLAN")
    expected_keymap=$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["options"]["system_defaults"]["keyboard_layout"])' "$PLAN")
    EXPECTED_LOCALE=$expected_locale EXPECTED_KEYMAP=$expected_keymap TARGET_ROOT=$TARGET_MOUNT python3 - <<'PY' \
        || fail locale_state_invalid "Installed locale or virtual-console configuration is not deterministic."
import os
from pathlib import Path

root = Path(os.environ["TARGET_ROOT"])
expected = os.environ["EXPECTED_LOCALE"]
keymap = os.environ["EXPECTED_KEYMAP"]
active = {
    line.split()[0]
    for line in (root / "etc/locale.gen").read_text(encoding="utf-8").splitlines()
    if line and not line.lstrip().startswith("#")
}
if "is_IS.UTF-8" not in active or expected not in active:
    raise SystemExit("required generated locale declarations are missing")
locale_conf = dict(
    line.split("=", 1)
    for line in (root / "etc/locale.conf").read_text(encoding="utf-8").splitlines()
    if "=" in line
)
if locale_conf.get("LANG") != expected:
    raise SystemExit("LANG does not match the selected locale")
vconsole = dict(
    line.split("=", 1)
    for line in (root / "etc/vconsole.conf").read_text(encoding="utf-8").splitlines()
    if "=" in line
)
if vconsole != {"KEYMAP": keymap, "FONT": "default8x16"}:
    raise SystemExit("vconsole state does not match the selected keyboard")
PY
    locale_inventory=$(arch-chroot "$TARGET_MOUNT" locale -a) \
        || fail locale_state_invalid "Installed generated locales cannot be enumerated."
    LOCALE_INVENTORY=$locale_inventory EXPECTED_LOCALE=$expected_locale python3 - <<'PY' \
        || fail locale_state_invalid "Installed locale archive lacks Icelandic or the selected locale."
import os

normalize = lambda value: "".join(character for character in value.lower() if character.isalnum())
available = {normalize(line) for line in os.environ["LOCALE_INVENTORY"].splitlines()}
required = {normalize("is_IS.UTF-8"), normalize(os.environ["EXPECTED_LOCALE"])}
if not required <= available:
    raise SystemExit("generated locale inventory is incomplete")
PY
}


persist_success_evidence() {
    [[ -f $TARGET_MOUNT$RELEASE_PATH ]] || return 1
    if [[ -f /var/log/archinstall/install.log ]]; then
        redact_file /var/log/archinstall/install.log "$RUN_DIR/archinstall.log" || return 1
    fi
    CURRENT_PHASE=complete
    write_status success "" "Installation completed and boot artifacts, locale state, and persistent evidence were verified." || return 1
    redact_file "$RAW_LOG" "$RUN_DIR/backend.log" || return 1
    python3 - "$PLAN" "$RUN_DIR" "$TARGET_MOUNT" "$TARGET_LOG_DIR" "$TARGET_MOUNT$RELEASE_PATH" "$RUN_ID" <<'PY' || return 1
import json
import os
import re
import hashlib
import shutil
import sys
from pathlib import Path
plan_path, live_run, target_root, relative_log, release_path = map(Path, sys.argv[1:6])
run_id = sys.argv[6]
plan = json.loads(plan_path.read_text(encoding="utf-8"))
release_bytes = release_path.read_bytes()
release = json.loads(release_bytes)
required_release = ("version", "source_revision", "source_date_epoch", "build_id")
revision = release.get("source_revision")
epoch = release.get("source_date_epoch")
expected_build_id = (
    f"bifrost-{release.get('version')}-{revision[:12]}-{epoch}"
    if isinstance(revision, str) and isinstance(epoch, int) and not isinstance(epoch, bool)
    else None
)
if (
    release.get("schema_version") != 1
    or release.get("provenance_status") != "build-input"
    or not isinstance(revision, str)
    or not re.fullmatch(r"[0-9a-f]{40}", revision)
    or not isinstance(epoch, int)
    or isinstance(epoch, bool)
    or epoch <= 0
    or release.get("build_id") != expected_build_id
    or not re.fullmatch(r"[0-9a-f]{64}", str(release.get("profile_sha256", "")))
):
    raise SystemExit("installed release provenance is not a prepared build input")

log_root = target_root / relative_log.relative_to("/")
final = log_root
staging = log_root.parent / f".{run_id}.tmp"
state_path = target_root / "etc/bifrost/install-state.json"
state_tmp = state_path.with_name(f".{state_path.name}.{run_id}.tmp")
renamed = False
state_replaced = False
try:
    log_root.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    os.chmod(log_root.parent, 0o700)
    if final.exists() or staging.exists():
        raise FileExistsError("installer evidence destination already exists")
    staging.mkdir(mode=0o700)
    names = ("plan.json", "archinstall.json", "status.json", "backend.log", "archinstall.log")
    for name in names:
        source = live_run / name
        if source.is_file():
            destination = staging / name
            with source.open("rb") as input_file, destination.open("xb") as output_file:
                shutil.copyfileobj(input_file, output_file)
                output_file.flush()
                os.fsync(output_file.fileno())
            os.chmod(destination, 0o600)
    for required in ("plan.json", "archinstall.json", "status.json", "backend.log", "archinstall.log"):
        if not (staging / required).is_file():
            raise FileNotFoundError(f"final installer evidence lacks {required}")
    os.rename(staging, final)
    renamed = True
    directory_fd = os.open(final.parent, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(directory_fd)
    finally:
        os.close(directory_fd)
    if (final.stat().st_mode & 0o777) != 0o700 or final.stat().st_uid != 0:
        raise PermissionError("persistent installer evidence directory is not root-only")
    for required in ("plan.json", "archinstall.json", "status.json", "backend.log", "archinstall.log"):
        metadata = (final / required).stat()
        if (metadata.st_mode & 0o777) != 0o600 or metadata.st_uid != 0:
            raise PermissionError(f"persistent installer evidence has unsafe permissions: {required}")

    defaults = plan["options"]["system_defaults"]
    state = {
        "schema_version": 2,
        "encryption": plan["options"]["encryption"],
        "profiles": plan["options"]["profiles"],
        "run_id": run_id,
        "source_mode": plan["options"]["source_mode"],
        "installer_language": plan["options"]["installer_language"],
        "system": {
            "locale": defaults["locale"],
            "keyboard_layout": defaults["keyboard_layout"],
            "timezone": defaults["timezone"],
        },
        "installer_evidence": relative_log.as_posix(),
        "release": {
            "path": "/usr/share/bifrost/release.json",
            "sha256": hashlib.sha256(release_bytes).hexdigest(),
            **{field: release[field] for field in required_release},
            "provenance_status": release["provenance_status"],
        },
    }
    state_path.parent.mkdir(parents=True, exist_ok=True, mode=0o755)
    with state_tmp.open("x", encoding="utf-8") as output:
        json.dump(state, output, indent=2, ensure_ascii=False)
        output.write("\n")
        output.flush()
        os.fsync(output.fileno())
    os.chmod(state_tmp, 0o644)
    os.replace(state_tmp, state_path)
    state_replaced = True
    state_directory_fd = os.open(state_path.parent, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(state_directory_fd)
    finally:
        os.close(state_directory_fd)
    state_metadata = state_path.stat()
    if ((state_metadata.st_mode & 0o777) != 0o644 or state_metadata.st_uid != 0
            or json.loads(state_path.read_text(encoding="utf-8")) != state):
        raise RuntimeError("persistent install state verification failed")
except Exception:
    state_tmp.unlink(missing_ok=True)
    if state_replaced:
        state_path.unlink(missing_ok=True)
    if renamed:
        shutil.rmtree(final, ignore_errors=True)
    else:
        shutil.rmtree(staging, ignore_errors=True)
    raise
PY
    TARGET_EVIDENCE_PERSISTED=1 FINAL_SUCCESS=1
}

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"
BIFROST_PACKAGE=$(validate_bifrost_package) \
    || fail package_source_unverified "The signed bifrost-system bootstrap package failed verification."

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"
BIFROST_PACKAGE=$(validate_bifrost_package) \
    || fail package_source_unverified "The signed bifrost-system bootstrap package changed or failed verification."
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 "Registering the signed BifrOSt system package and applying system identity."
bootstrap_bifrost_system
apply_installed_experience

phase verify "Verifying systemd-boot, kernels, initramfs, encryption, locale, and console state."
verify_installation

phase evidence "Atomically retaining sanitized installer evidence in the installed system."
persist_success_evidence || fail evidence_persistence_failed "The installed target could not retain and verify sanitized installer evidence."

exit 0