#!/usr/bin/env python3 """ Flask dashboard for CM4 eMMC provisioning. Monitors deployment status, shows device connection steps, backup/restore. Supports USB boot mode (ask Backup/Deploy) and network-booted devices (register, then Backup/Deploy). """ import json import os import time from pathlib import Path from flask import Flask, render_template, jsonify, request, send_file, Response app = Flask(__name__) @app.after_request def no_cache(response): """Prevent browser from caching the dashboard so deploys are visible immediately.""" if request.path == "/" or request.path.startswith("/api/"): response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0" response.headers["Pragma"] = "no-cache" return response BASE_DIR = Path(os.environ.get("CM4_PROVISIONING_DIR", "/var/lib/cm4-provisioning")) STATUS_FILE = os.environ.get("CM4_STATUS_FILE", str(BASE_DIR / "status.json")) LOG_FILE = os.environ.get("CM4_LOG_FILE", str(BASE_DIR / "flash.log")) ACTION_REQUEST_FILE = os.environ.get("CM4_ACTION_REQUEST_FILE", str(BASE_DIR / "action_request")) DEVICE_SOURCE_FILE = os.environ.get("CM4_DEVICE_SOURCE_FILE", str(BASE_DIR / "device_source")) BACKUPS_DIR = Path(os.environ.get("CM4_BACKUPS_DIR", str(BASE_DIR / "backups"))) GOLDEN_IMAGE = Path(os.environ.get("CM4_GOLDEN_IMAGE", str(BASE_DIR / "golden.img"))) NETWORK_DEVICES_FILE = Path(os.environ.get("CM4_NETWORK_DEVICES_FILE", str(BASE_DIR / "network_devices.json"))) DEFAULT_STATUS = { "phase": "idle", "message": "Waiting for reTerminal in boot mode or network.", "progress": None, "updated": None, } def read_status(): try: with open(STATUS_FILE, "r") as f: data = json.load(f) out = {**DEFAULT_STATUS, **data} if out.get("phase") == "waiting_choice": try: with open(DEVICE_SOURCE_FILE, "r") as sf: out["device_source"] = (sf.read() or "").strip() or "usb" except (FileNotFoundError, OSError): out["device_source"] = "usb" return out except (FileNotFoundError, json.JSONDecodeError): return DEFAULT_STATUS def read_log_tail(lines=50): try: with open(LOG_FILE, "r") as f: all_lines = f.readlines() return "".join(all_lines[-lines:]).strip() if all_lines else "" except (FileNotFoundError, PermissionError): return "" def _load_network_devices(): try: if NETWORK_DEVICES_FILE.is_file(): with open(NETWORK_DEVICES_FILE, "r") as f: return json.load(f) except (json.JSONDecodeError, OSError): pass return {"devices": []} def _save_network_devices(data): try: os.makedirs(NETWORK_DEVICES_FILE.parent, exist_ok=True) with open(NETWORK_DEVICES_FILE, "w") as f: json.dump(data, f, indent=2) return True except (PermissionError, OSError): return False def list_backups(): if not BACKUPS_DIR.is_dir(): return [] out = [] for p in sorted(BACKUPS_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True): if p.is_file() and p.suffix in (".img", ".img.gz"): try: st = p.stat() out.append({"name": p.name, "size": st.st_size, "mtime": st.st_mtime}) except OSError: pass return out @app.route("/") def index(): return render_template("index.html") @app.route("/api/status") def api_status(): return jsonify(read_status()) @app.route("/api/log") def api_log(): return jsonify({"log": read_log_tail()}) @app.route("/api/pending-devices") def api_pending_devices(): """Returns USB (if waiting_choice) and registered network devices so the UI can show Backup/Deploy.""" st = read_status() usb = None if st.get("phase") == "waiting_choice": usb = {"source": "usb", "message": st.get("message", "Device connected (USB). Choose action.")} data = _load_network_devices() network = [d for d in data.get("devices", []) if d.get("action") in (None, "wait")] return jsonify({"usb": usb, "network": network}) @app.route("/api/device-action", methods=["POST"]) def api_device_action(): """User chose Backup or Deploy for a device. source=usb | network; for network pass mac=.""" body = request.get_json(force=True, silent=True) or {} source = (body.get("source") or "").strip().lower() action = (body.get("action") or "").strip().lower() if action not in ("backup", "deploy"): return jsonify({"ok": False, "error": "action must be 'backup' or 'deploy'"}), 400 if source == "usb": try: os.makedirs(os.path.dirname(ACTION_REQUEST_FILE) or ".", exist_ok=True) with open(ACTION_REQUEST_FILE, "w") as f: f.write(action) return jsonify({"ok": True}) except (PermissionError, OSError): return jsonify({"ok": False, "error": "Could not write action file"}), 500 if source == "network": mac = (body.get("mac") or "").strip() if not mac: return jsonify({"ok": False, "error": "mac required for network device"}), 400 data = _load_network_devices() for d in data.get("devices", []): if (d.get("mac") or "").lower() == mac.lower(): d["action"] = action d["action_at"] = time.time() _save_network_devices(data) return jsonify({"ok": True}) return jsonify({"ok": False, "error": "Device not found"}), 404 return jsonify({"ok": False, "error": "source must be 'usb' or 'network'"}), 400 @app.route("/api/register-device", methods=["POST"]) def api_register_device(): """Called by a network-booted device to register (mac, ip).""" body = request.get_json(force=True, silent=True) or request.form mac = (body.get("mac") or "").strip() ip = (body.get("ip") or request.remote_addr or "").strip() if not mac: return jsonify({"ok": False, "error": "mac required"}), 400 data = _load_network_devices() devices = data.get("devices", []) for d in devices: if (d.get("mac") or "").lower() == mac.lower(): d["ip"] = ip d["registered_at"] = time.time() d["action"] = d.get("action") or "wait" _save_network_devices(data) return jsonify({"ok": True, "message": "registered"}) devices.append({"mac": mac, "ip": ip, "registered_at": time.time(), "action": "wait"}) data["devices"] = devices _save_network_devices(data) return jsonify({"ok": True, "message": "registered"}) @app.route("/api/device-action-poll") def api_device_action_poll(): """Network device polls this to get its assigned action (deploy/backup) and URL.""" mac = (request.args.get("mac") or "").strip() if not mac: return jsonify({"action": "wait"}), 200 data = _load_network_devices() base = request.host_url.rstrip("/") for d in data.get("devices", []): if (d.get("mac") or "").lower() == mac.lower(): action = d.get("action") or "wait" if action == "deploy": return jsonify({"action": "deploy", "url": f"{base}/api/golden-image"}) if action == "backup": return jsonify({"action": "backup", "upload_url": f"{base}/api/backup-upload?mac={mac}"}) return jsonify({"action": "wait"}) return jsonify({"action": "wait"}) @app.route("/api/golden-image") def api_golden_image(): """Stream the golden image for network deploy (device pulls and writes to eMMC).""" if not GOLDEN_IMAGE.is_file(): return jsonify({"error": "Golden image not found"}), 404 return send_file( GOLDEN_IMAGE, mimetype="application/octet-stream", as_attachment=True, download_name="golden.img", ) @app.route("/api/backup-upload", methods=["POST"]) def api_backup_upload(): """Network device uploads its eMMC backup (raw body).""" mac = (request.args.get("mac") or "").strip().replace(":", "-")[:20] if not mac: return jsonify({"error": "mac query param required"}), 400 BACKUPS_DIR.mkdir(parents=True, exist_ok=True) name = f"backup-net-{mac}-{int(time.time())}.img" path = BACKUPS_DIR / name try: with open(path, "wb") as f: while True: chunk = request.stream.read(1024 * 1024) if not chunk: break f.write(chunk) return jsonify({"ok": True, "file": name}) except (OSError, IOError) as e: if path.exists(): path.unlink(missing_ok=True) return jsonify({"error": str(e)}), 500 @app.route("/api/backups") def api_backups(): return jsonify({"backups": list_backups()}) @app.route("/api/backups/") def api_backup_download(name): if ".." in name or "/" in name or "\\" in name: return jsonify({"error": "invalid name"}), 400 path = BACKUPS_DIR / name if not path.is_file(): return jsonify({"error": "not found"}), 404 return send_file(path, as_attachment=True, download_name=name) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=False)