Add firstboot onboarding and prep/check tooling

pikit_api/constants.py

@@ -22,6 +22,13 @@ API_PACKAGE_DIR = API_DIR / "pikit_api"
 BACKUP_ROOT = pathlib.Path("/var/backups/pikit")
 TMP_ROOT = pathlib.Path("/var/tmp/pikit-update")
 
+# First-boot state
+FIRSTBOOT_DIR = pathlib.Path("/var/lib/pikit/firstboot")
+FIRSTBOOT_STATE = FIRSTBOOT_DIR / "state.json"
+FIRSTBOOT_LOG = FIRSTBOOT_DIR / "firstboot.log"
+FIRSTBOOT_ERROR = FIRSTBOOT_DIR / "firstboot.error"
+FIRSTBOOT_DONE = FIRSTBOOT_DIR / "firstboot.done"
+
 # Apt / unattended-upgrades
 APT_AUTO_CFG = pathlib.Path("/etc/apt/apt.conf.d/20auto-upgrades")
 APT_UA_BASE = pathlib.Path("/etc/apt/apt.conf.d/50unattended-upgrades")

pikit_api/firstboot.py

@@ -0,0 +1,100 @@
+import json
+import pathlib
+from typing import Any, Dict, List, Optional
+
+from .constants import FIRSTBOOT_DIR, FIRSTBOOT_DONE, FIRSTBOOT_ERROR, FIRSTBOOT_LOG, FIRSTBOOT_STATE, WEB_ROOT
+from .helpers import ensure_dir, sha256_file
+
+DEFAULT_STEPS = [
+    "Preparing system",
+    "Generating security keys",
+    "Securing the dashboard",
+    "Updating software (this can take a while)",
+    "Final checks",
+    "Starting Pi-Kit",
+]
+
+
+def _tail_text(path: pathlib.Path, max_lines: int = 200) -> str:
+    if not path.exists():
+        return ""
+    try:
+        text = path.read_text(errors="ignore")
+    except Exception:
+        return ""
+    lines = text.splitlines()
+    if len(lines) > max_lines:
+        lines = lines[-max_lines:]
+    return "\n".join(lines)
+
+
+def _normalize_steps(raw_steps: Optional[List[Dict[str, Any]]], state: str) -> List[Dict[str, Any]]:
+    steps: List[Dict[str, Any]] = []
+    if raw_steps:
+        for entry in raw_steps:
+            label = (entry or {}).get("label") or (entry or {}).get("name")
+            if not label:
+                continue
+            status = (entry or {}).get("status") or "pending"
+            steps.append({"label": str(label), "status": str(status)})
+
+    if not steps:
+        steps = [{"label": label, "status": "pending"} for label in DEFAULT_STEPS]
+
+    if state == "done":
+        for step in steps:
+            step["status"] = "done"
+    return steps
+
+
+def _current_step(steps: List[Dict[str, Any]]) -> Optional[str]:
+    for step in steps:
+        if step.get("status") in ("current", "running", "error"):
+            return step.get("label")
+    return None
+
+
+def _load_state_file() -> Dict[str, Any]:
+    if FIRSTBOOT_STATE.exists():
+        try:
+            return json.loads(FIRSTBOOT_STATE.read_text())
+        except Exception:
+            return {}
+    return {}
+
+
+def read_firstboot_status() -> Dict[str, Any]:
+    ensure_dir(FIRSTBOOT_DIR)
+    state_file = _load_state_file()
+
+    if FIRSTBOOT_ERROR.exists():
+        state = "error"
+    elif FIRSTBOOT_DONE.exists():
+        state = "done"
+    else:
+        state = state_file.get("state") or "running"
+        if state not in ("running", "done", "error"):
+            state = "running"
+
+    steps = _normalize_steps(state_file.get("steps"), state)
+    current_step = state_file.get("current_step") or _current_step(steps)
+
+    ca_path = WEB_ROOT / "assets" / "pikit-ca.crt"
+    ca_hash = sha256_file(ca_path) if ca_path.exists() else None
+
+    return {
+        "state": state,
+        "steps": steps,
+        "current_step": current_step,
+        "log_tail": _tail_text(FIRSTBOOT_LOG, 200),
+        "error_present": FIRSTBOOT_ERROR.exists(),
+        "error_path": "/api/firstboot/error",
+        "ca_hash": ca_hash,
+        "ca_url": "/assets/pikit-ca.crt",
+    }
+
+
+def read_firstboot_error(max_lines: int = 200) -> Dict[str, Any]:
+    if not FIRSTBOOT_ERROR.exists():
+        return {"present": False, "text": ""}
+    return {"present": True, "text": _tail_text(FIRSTBOOT_ERROR, max_lines)}

pikit_api/http_handlers.py

@@ -5,6 +5,7 @@ from http.server import BaseHTTPRequestHandler
 from .auto_updates import auto_updates_state, read_updates_config, set_auto_updates, set_updates_config
 from .constants import CORE_NAME, CORE_PORTS, DIAG_LOG_FILE
 from .diagnostics import _load_diag_state, _save_diag_state, dbg, diag_log, diag_read
+from .firstboot import read_firstboot_error, read_firstboot_status
 from .helpers import default_host, detect_https, normalize_path
 from .releases import (
     check_for_update,
@@ -48,6 +49,12 @@ class Handler(BaseHTTPRequestHandler):
         if self.path.startswith("/api/status"):
             return self._send(200, collect_status())
 
+        if self.path.startswith("/api/firstboot/error"):
+            return self._send(200, read_firstboot_error())
+
+        if self.path.startswith("/api/firstboot"):
+            return self._send(200, read_firstboot_status())
+
         if self.path.startswith("/api/services"):
             return self._send(200, {"services": list_services_for_ui()})