furtka/webinstaller/app.py

import json
import os
import re
import subprocess
from pathlib import Path

from drives import list_scored_devices
from flask import Flask, jsonify, redirect, render_template, request, url_for

app = Flask(__name__)

LANGUAGES = {
    "en": {"locale": "en_US.UTF-8", "label": "English"},
    "de": {"locale": "de_DE.UTF-8", "label": "Deutsch"},
    "pl": {"locale": "pl_PL.UTF-8", "label": "Polski"},
}

STATE_DIR = Path(os.environ.get("FURTKA_STATE_DIR", "/tmp/furtka"))
INSTALL_LOG = STATE_DIR / "install.log"
CONFIG_PATH = STATE_DIR / "user_configuration.json"
CREDS_PATH = STATE_DIR / "user_credentials.json"

# Pre-populated with sane defaults so the form has something useful on first
# render. POSTs validate and overwrite.
settings = {
    "hostname": "furtka",
    "username": "",
    "password": "",
    "language": "en",
    "boot_drive": "",
}

HOSTNAME_RE = re.compile(r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$")
USERNAME_RE = re.compile(r"^[a-z_][a-z0-9_-]{0,31}$")

# Ordered phase markers for the install progress bar. Each tuple is
# (substring to search for in the archinstall log, progress percent when
# reached, user-facing label). Pick the furthest phase whose marker is
# present in the log. If archinstall changes its stdout wording the bar
# stalls on the last recognized phase — the install itself keeps going.
PROGRESS_PHASES = [
    ("Wiping partitions",                          8,   "Preparing your disk"),
    ("Creating partitions",                        12,  "Creating partitions"),
    ("Starting installation",                      15,  "Starting installation"),
    ("Waiting for",                                18,  "Syncing time and packages"),
    ("Installing packages: ['base'",               25,  "Installing the base system (this takes a while)"),
    ("Adding bootloader",                          65,  "Setting up boot"),
    ("Installing packages: ['efibootmgr'",         70,  "Setting up boot"),
    ("Installing packages: ['docker'",             80,  "Installing your apps"),
    ("Enabling service",                           90,  "Turning on services"),
    ("Updating /mnt/etc/fstab",                    95,  "Almost done"),
    ("Installation completed without any errors", 100,  "Done!"),
]

PROGRESS_ERROR_MARKERS = ("Traceback (most recent call last)", "archinstall: error:")


def parse_install_progress(log):
    percent = 2
    phase = "Starting up…"
    for marker, pct, label in PROGRESS_PHASES:
        if marker in log:
            percent = pct
            phase = label

    if percent >= 100:
        status = "done"
    elif any(m in log for m in PROGRESS_ERROR_MARKERS):
        status = "error"
        phase = "Installation failed — open Show details below"
    else:
        status = "running"

    return {"percent": percent, "phase": phase, "status": status}


def validate_step1(form):
    errors = []
    values = {
        "hostname": form.get("hostname", "").strip(),
        "username": form.get("username", "").strip(),
        "password": form.get("password", ""),
        "language": form.get("language", ""),
    }
    password2 = form.get("password2", "")

    if not HOSTNAME_RE.match(values["hostname"]):
        errors.append("Hostname must be lowercase letters, digits, hyphens (max 63 chars).")
    if not USERNAME_RE.match(values["username"]):
        errors.append("Username must start with a letter or underscore, lowercase only.")
    if len(values["password"]) < 8:
        errors.append("Password must be at least 8 characters.")
    if values["password"] != password2:
        errors.append("Passwords do not match.")
    if values["language"] not in LANGUAGES:
        errors.append("Pick a language.")
    return errors, values


def build_disk_config(boot_drive):
    # archinstall 4.x dropped the `use_entire_disk` shortcut — `default_layout`
    # now requires fully-specified partitions. We call suggest_single_disk_layout
    # with ext4 + no separate /home, which short-circuits its interactive prompts.
    import asyncio

    from archinstall.lib.disk.device_handler import device_handler
    from archinstall.lib.disk.disk_menu import suggest_single_disk_layout
    from archinstall.lib.models.device import (
        DiskLayoutConfiguration,
        DiskLayoutType,
        FilesystemType,
    )

    device_handler.load_devices()
    device = device_handler.get_device(Path(boot_drive))
    if device is None:
        raise RuntimeError(f"archinstall could not resolve device {boot_drive!r}")

    device_mod = asyncio.run(
        suggest_single_disk_layout(
            device,
            filesystem_type=FilesystemType.Ext4,
            separate_home=False,
        )
    )
    layout = DiskLayoutConfiguration(
        config_type=DiskLayoutType.Default,
        device_modifications=[device_mod],
    )
    return layout.json()


def build_archinstall_config(s):
    return {
        "archinstall-language": "English",
        "timezone": "Europe/Berlin",
        "ntp": True,
        "bootloader": "Systemd-boot",
        "disk_config": build_disk_config(s["boot_drive"]),
        "hostname": s["hostname"],
        "kernels": ["linux"],
        "packages": ["docker", "docker-compose", "vim", "git", "htop", "curl"],
        "profile": {"type": "server"},
        "services": ["docker"],
        # Add user to the docker group post-install. We can't put "docker" in
        # the user's `groups` at create-time because archinstall creates users
        # before pacstrapping the extras, so the docker group doesn't exist
        # yet. custom_commands runs at the very end.
        "custom_commands": [f"gpasswd -a {s['username']} docker"],
        "network_config": {"type": "iso"},
        "ssh": True,
        "audio_config": None,
        "locale_config": {
            "locale": LANGUAGES[s["language"]]["locale"],
            "keyboard_layout": "us",
        },
    }


def build_archinstall_creds(s):
    # archinstall 4.x expects `!root-password` and `!password` (plaintext
    # sentinels). Users with neither `!password` nor `enc_password` are
    # silently dropped by User.parse_arguments — hence login failures.
    return {
        "!root-password": s["password"],
        "users": [
            {
                "username": s["username"],
                "!password": s["password"],
                "sudo": True,
                "groups": [],
            }
        ],
    }


def write_install_files(s, state_dir):
    state_dir.mkdir(parents=True, exist_ok=True)
    config_path = state_dir / "user_configuration.json"
    creds_path = state_dir / "user_credentials.json"
    config_path.write_text(json.dumps(build_archinstall_config(s), indent=2))
    creds_path.write_text(json.dumps(build_archinstall_creds(s), indent=2))
    creds_path.chmod(0o600)
    return config_path, creds_path


def spawn_archinstall(config_path, creds_path, log_path):
    log_fh = open(log_path, "wb")
    return subprocess.Popen(
        [
            "archinstall",
            "--config", str(config_path),
            "--creds", str(creds_path),
            "--silent",
        ],
        stdout=log_fh,
        stderr=subprocess.STDOUT,
        start_new_session=True,
    )


@app.route("/")
def home():
    return redirect(url_for("install_step_1"))


@app.route("/install/step1", methods=["GET", "POST"])
def install_step_1():
    errors = []
    if request.method == "POST":
        errors, values = validate_step1(request.form)
        if not errors:
            settings.update(values)
            return redirect(url_for("install_step_2"))
        form_values = values
    else:
        form_values = {k: settings[k] for k in ("hostname", "username", "language")}
    return render_template(
        "install/step1.html",
        values=form_values,
        languages=LANGUAGES,
        errors=errors,
    )


@app.route("/install/step2", methods=["GET", "POST"])
def install_step_2():
    if request.method == "POST":
        boot_drive = request.form.get("boot_drive", "").strip()
        if boot_drive:
            settings["boot_drive"] = boot_drive
            return redirect(url_for("install_overview"))
    return render_template(
        "install/step2.html",
        drives=list_scored_devices(),
        selected=settings.get("boot_drive", ""),
    )


@app.route("/install/overview")
def install_overview():
    masked = {**settings, "password": "•" * 8 if settings["password"] else ""}
    return render_template("install/overview.html", settings=masked)


@app.route("/install/run", methods=["POST"])
def install_run():
    if not settings["boot_drive"] or not settings["username"] or not settings["password"]:
        return redirect(url_for("install_step_1"))
    config_path, creds_path = write_install_files(settings, STATE_DIR)
    INSTALL_LOG.write_bytes(b"")
    if os.environ.get("FURTKA_DRY_RUN") == "1":
        INSTALL_LOG.write_text(
            f"DRY RUN: would exec archinstall --config {config_path} "
            f"--creds {creds_path} --silent\n"
        )
    else:
        spawn_archinstall(config_path, creds_path, INSTALL_LOG)
    return redirect(url_for("install_log_view"))


@app.route("/install/log")
def install_log_view():
    log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else ""
    return render_template(
        "install/log.html",
        log=log,
        progress=parse_install_progress(log),
    )


@app.route("/install/log.json")
def install_log_json():
    log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else ""
    return jsonify(log=log, progress=parse_install_progress(log))


@app.route("/install/reboot", methods=["POST"])
def install_reboot():
    # Only allow rebooting once the install has actually finished — we don't
    # want a panicked click during install to reboot mid-pacstrap.
    log = INSTALL_LOG.read_text() if INSTALL_LOG.exists() else ""
    if parse_install_progress(log)["status"] != "done":
        return redirect(url_for("install_log_view"))
    subprocess.Popen(
        ["/usr/bin/systemctl", "reboot"],
        start_new_session=True,
    )
    return render_template("install/rebooting.html")


if __name__ == "__main__":
    app.run(debug=True, port=5000)