#!/usr/bin/env python3 """ Nexus One AI — First Boot Web Setup Server Serves on port 80. Access from any browser on the same network. """ import os, json, subprocess, threading, time, socket, ipaddress, sys from http.server import HTTPServer, BaseHTTPRequestHandler from pathlib import Path from urllib.parse import parse_qs, urlparse SETUP_DONE_FILE = "/opt/cezen/.setup-done" INSTALL_LOG = "/var/log/cezen-install.log" AIPACKAGE_DIR = "/opt/aipackage" SCRIPT_ROOT = Path(__file__).resolve().parents[2] SCRIPTS_DIR = Path(AIPACKAGE_DIR) / "scripts" if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR if SCRIPTS_DIR.exists() else SCRIPT_ROOT / "scripts")) from cezen_license import build_tier_options, evaluate_license, evaluate_override, normalize_tier, read_json_file, write_json_file install_proc = None install_status = {"running": False, "done": False, "error": None} FEASIBILITY_SCRIPT = f"{AIPACKAGE_DIR}/scripts/cezen-feasibility.sh" FEASIBILITY_JSON = "/opt/cezen/feasibility.json" LICENSE_JSON = "/opt/cezen/license.json" OVERRIDE_JSON = "/opt/cezen/license.override.json" INSTALL_RECORD_JSON = "/opt/cezen/install-record.json" INSTALL_OPERATION_JSON = os.environ.get("CEZEN_SETUP_OPERATION_JSON", "/opt/cezen/setup-operation.json") PUBLIC_KEY_PATH = os.environ.get("CEZEN_LICENSE_PUBLIC_KEY", f"{AIPACKAGE_DIR}/autoinstall/keys/cezen-license-public.pem") def read_install_operation(): return read_json_file(INSTALL_OPERATION_JSON) def write_install_operation(operation): operation = dict(operation or {}) operation["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) write_json_file(INSTALL_OPERATION_JSON, operation) try: os.chmod(INSTALL_OPERATION_JSON, 0o640) except OSError: pass return operation def start_install_operation(tier, skip_tools, idempotency_key): existing = read_install_operation() if idempotency_key and existing.get("idempotency_key") == idempotency_key: existing["duplicate_request"] = True return existing, False if existing.get("state") in {"validating", "running", "awaiting_reboot"}: existing["duplicate_request"] = True return existing, False now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) operation = { "schema": "cezen.setup_operation.v1", "operation_id": str(__import__("uuid").uuid4()), "correlation_id": str(__import__("uuid").uuid4()), "idempotency_key": idempotency_key or str(__import__("uuid").uuid4()), "kind": "install", "state": "validating", "tier": tier, "skip_tools": list(skip_tools or []), "progress_percent": 0, "progress_label": "Validating installation", "acknowledged_at": now, "updated_at": now, "duplicate_request": False, "customer_message": "Installation request accepted.", "recovery_actions": [], } return write_install_operation(operation), True # ─── Helpers ────────────────────────────────────────────── def get_ip(): try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except: return "unknown" def get_interfaces(): try: out = subprocess.check_output(["ip", "-o", "link", "show"], text=True) ifaces = [] for line in out.splitlines(): name = line.split(": ")[1].split("@")[0] if name not in ("lo",) and not name.startswith(("docker","br-","veth","k3s")): ifaces.append(name) return ifaces except: return ["eth0"] def has_nvidia_gpu(): """Detect NVIDIA PCI devices before the driver or nvidia-smi exists.""" try: for root, _, files in os.walk("/sys/bus/pci/devices"): if "vendor" not in files: continue with open(os.path.join(root, "vendor")) as f: if f.read().strip().lower() == "0x10de": return True except Exception: pass return False def validate_static_network(ip, prefix, gateway, dns): ipaddress.ip_address(ip) ipaddress.ip_address(gateway) ipaddress.ip_address(dns) prefix_int = int(prefix) if prefix_int < 1 or prefix_int > 32: raise ValueError("CIDR prefix must be between 1 and 32") return str(prefix_int) def is_ip_in_use(ip): """Best-effort conflict check before taking a static IP.""" try: result = subprocess.run( ["ping", "-c", "1", "-W", "1", ip], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) return result.returncode == 0 except Exception: return False def apply_static_ip(iface, ip, prefix, gateway, dns): prefix = validate_static_network(ip, prefix, gateway, dns) config = f"""network: version: 2 ethernets: {iface}: dhcp4: false addresses: - {ip}/{prefix} routes: - to: default via: {gateway} nameservers: addresses: [{dns}] """ with open("/etc/netplan/99-cezen-static.yaml", "w") as f: f.write(config) subprocess.run(["netplan", "apply"], capture_output=True) time.sleep(3) def save_license_artifacts(license_data): os.makedirs("/opt/cezen", exist_ok=True) for target in (LICENSE_JSON, OVERRIDE_JSON): try: os.remove(target) except FileNotFoundError: pass raw_license = (license_data.get("license_json") or "").strip() raw_override = (license_data.get("override_json") or "").strip() if raw_license: payload = json.loads(raw_license) write_json_file(LICENSE_JSON, payload) subprocess.run(["chown", "root:cezen", LICENSE_JSON], check=False) os.chmod(LICENSE_JSON, 0o640) if raw_override: payload = json.loads(raw_override) write_json_file(OVERRIDE_JSON, payload) subprocess.run(["chown", "root:cezen", OVERRIDE_JSON], check=False) os.chmod(OVERRIDE_JSON, 0o640) def ensure_feasibility(): os.makedirs("/opt/cezen", exist_ok=True) if os.path.exists(FEASIBILITY_SCRIPT): subprocess.run(["bash", FEASIBILITY_SCRIPT, FEASIBILITY_JSON], check=False) return read_json_file(FEASIBILITY_JSON) def evaluate_install_constraints(license_data): save_license_artifacts(license_data or {}) feasibility = ensure_feasibility() hardware_tier = normalize_tier(((feasibility.get("recommendation") or {}).get("recommended_tier")), "basic") license_record = read_json_file(LICENSE_JSON) override_record = read_json_file(OVERRIDE_JSON) license_eval = evaluate_license(license_record, PUBLIC_KEY_PATH) override_eval = evaluate_override(override_record, PUBLIC_KEY_PATH) return { "license": license_eval, "override": override_eval, "hardware": { "recommended_tier": hardware_tier, "recommended_profile": (feasibility.get("recommendation") or {}).get("recommended_profile", ""), "estimated_concurrent_users": (feasibility.get("recommendation") or {}).get("estimated_concurrent_users", ""), "notes": (feasibility.get("recommendation") or {}).get("notes", []), }, "tier_options": build_tier_options(license_eval, hardware_tier, override_eval), } def write_install_record(license_data, tier, skip_tools, preview): payload = { "schema": "cezen.install_record.v1", "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "customer_name": (license_data.get("customer_name") or "").strip(), "customer_id": (license_data.get("customer_id") or "").strip(), "contact_email": (license_data.get("contact_email") or "").strip(), "support_until": (license_data.get("support_until") or "").strip(), "selected_tier": tier, "provisioned_tier": tier, "licensed_tier": ((preview.get("license") or {}).get("allowed_tier") or "basic"), "license_status": ((preview.get("license") or {}).get("status") or "missing"), "hardware_recommended_tier": ((preview.get("hardware") or {}).get("recommended_tier") or "starter"), "hardware_recommended_profile": ((preview.get("hardware") or {}).get("recommended_profile") or "core"), "install_type": ((preview.get("license") or {}).get("install_type") or "field-staging"), "override_active": bool((preview.get("override") or {}).get("allow_hardware_mismatch")), "skip_roles": skip_tools or [], "components": {k: k not in set(skip_tools or []) for k in ["ollama", "jupyterlab", "chromadb", "vllm", "mlflow", "minio", "monitoring", "k3s"]}, } write_json_file(INSTALL_RECORD_JSON, payload) subprocess.run(["chown", "root:cezen", INSTALL_RECORD_JSON], check=False) os.chmod(INSTALL_RECORD_JSON, 0o640) def run_install(tier, skip_tools, license_data=None, preview=None, operation_id=""): global install_status install_status = {"running": True, "done": False, "error": None} try: operation = read_install_operation() if operation.get("operation_id") == operation_id: operation.update({"state": "running", "progress_percent": 5, "progress_label": "Preparing appliance"}) write_install_operation(operation) # Write config so phase 2 (post-reboot) knows what to skip os.makedirs("/opt/cezen", exist_ok=True) skip_str = ",".join(skip_tools) if skip_tools else "" save_license_artifacts(license_data or {}) write_install_record(license_data or {}, tier, skip_tools or [], preview or {}) with open("/opt/cezen/install.conf", "w") as f: f.write(f"TIER={tier}\nSKIP_ROLES={skip_str}\n") # Mark setup done NOW so this web UI doesn't restart after the phase-1 reboot open(SETUP_DONE_FILE, "w").close() env = os.environ.copy() # Fresh NVIDIA servers do not have nvidia-smi yet, so detect the PCI # device and run phase 1 to install drivers before the AI stack. phase = "1" if has_nvidia_gpu() else "2" if operation.get("operation_id") == operation_id and phase == "1": operation.update({"state": "awaiting_reboot", "progress_percent": 40, "progress_label": "Installing drivers before reboot"}) write_install_operation(operation) cmd = ["bash", f"{AIPACKAGE_DIR}/install.sh", f"--phase={phase}", f"--tier={tier}"] with open(INSTALL_LOG, "w") as log: proc = subprocess.Popen(cmd, stdout=log, stderr=log, env=env) proc.wait() # Reaches here only if no reboot happened (e.g. no GPU / drivers already installed) install_status = {"running": False, "done": True, "error": None} operation = read_install_operation() if operation.get("operation_id") == operation_id: operation.update({"state": "succeeded", "progress_percent": 100, "progress_label": "Installation command completed", "customer_message": "Installation completed successfully."}) write_install_operation(operation) except Exception as e: install_status = {"running": False, "done": False, "error": str(e)} operation = read_install_operation() if operation.get("operation_id") == operation_id: operation.update({"state": "recovery_required", "progress_label": "Installation needs attention", "result_code": "install_failed", "customer_message": "Installation did not complete. Review the local log and retry or contact support.", "recovery_actions": ["retry", "view_log", "escalate"]}) write_install_operation(operation) # ─── HTML UI ────────────────────────────────────────────── HTML = r""" Nexus One AI — Server Setup
Powered by Cezen  ·  Server Setup Wizard
1 · Network
2 · License
3 · Select Tier
4 · AI Tools
5 · Install

Network Configuration

Choose how this server gets its IP address. You can change this later.

Server will get an IP automatically from your network. Current IP: detecting...
We will ping the IP before applying it. If it replies, we will block the change to avoid taking an address that is already in use.
""" # ─── HTTP Handler ───────────────────────────────────────── class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *args): pass # suppress access log def send_json(self, data, code=200): body = json.dumps(data).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", len(body)) self.end_headers() self.wfile.write(body) def do_GET(self): path = urlparse(self.path).path if path == "/" or path == "/index.html": body = HTML.encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", len(body)) self.end_headers() self.wfile.write(body) elif path == "/api/status": self.send_json({ "ip": get_ip(), "interfaces": get_interfaces(), "setup_done": os.path.exists(SETUP_DONE_FILE), "install_status": install_status, "operation": read_install_operation(), }) elif path == "/api/progress": self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.end_headers() try: with open(INSTALL_LOG, "r") as f: f.seek(0, 2) # seek to end while install_status["running"]: line = f.readline() if line: msg = f"data: {line.rstrip()}\n\n" self.wfile.write(msg.encode()) self.wfile.flush() else: time.sleep(0.5) # Stream remaining lines after finish for line in f: msg = f"data: {line.rstrip()}\n\n" self.wfile.write(msg.encode()) self.wfile.write(b"data: === INSTALL COMPLETE ===\n\n") self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass else: self.send_response(404) self.end_headers() def do_POST(self): path = urlparse(self.path).path length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length)) if length else {} if path == "/api/network": try: if body.get("mode") == "static": ifaces = get_interfaces() iface = ifaces[0] if ifaces else "eth0" validate_static_network(body["ip"], body["prefix"], body["gateway"], body["dns"]) if is_ip_in_use(body["ip"]): self.send_json({ "ok": False, "error": f"IP address {body['ip']} is already replying to ping. Choose a different address." }, 409) return apply_static_ip(iface, body["ip"], body["prefix"], body["gateway"], body["dns"]) self.send_json({ "ok": True, "ip": get_ip(), "warning": "Ping check passed before applying the static IP. Note: a non-reply does not guarantee the address is unused if another host blocks ICMP." }) return self.send_json({"ok": True, "ip": get_ip()}) except Exception as e: self.send_json({"ok": False, "error": str(e)}, 500) elif path == "/api/license-preview": try: preview = evaluate_install_constraints(body.get("license", {}) or {}) self.send_json(preview) except Exception as e: self.send_json({"ok": False, "error": str(e)}, 400) elif path == "/api/install": global install_proc tier = body.get("tier", "basic") skip = body.get("skip_tools", []) license_data = body.get("license", {}) idempotency_key = str(body.get("idempotency_key") or "").strip() try: preview = evaluate_install_constraints(license_data or {}) except Exception as e: self.send_json({"ok": False, "error": f"License validation failed: {e}"}, 400) return selected = next((opt for opt in preview.get("tier_options", []) if opt.get("tier") == tier), None) if not selected or not selected.get("selectable"): self.send_json({"ok": False, "error": f"Tier '{tier}' is not allowed for this license/hardware."}, 400) return operation, created = start_install_operation(tier, skip, idempotency_key) if created and not install_status["running"]: t = threading.Thread(target=run_install, args=(tier, skip, license_data, preview, operation["operation_id"]), daemon=True) t.start() self.send_json({"ok": True, "operation_id": operation["operation_id"], "acknowledged_at": operation["acknowledged_at"], "duplicate_request": False}) else: self.send_json({"ok": True, "operation_id": operation.get("operation_id"), "acknowledged_at": operation.get("acknowledged_at"), "duplicate_request": True}) else: self.send_response(404) self.end_headers() # ─── Main ───────────────────────────────────────────────── def show_console_banner(ip): """Write the setup URL banner to /dev/tty1 so it appears on the physical console.""" banner = f""" \033[1;36m╔══════════════════════════════════════════════════════╗ ║ ║ ║ NEXUS ONE AI — SERVER SETUP ║ ║ ║ ║ Open a browser on any computer on this network: ║ ║ ║ ║ \033[1;33m➜ http://{ip:<42}\033[1;36m║ ║ \033[1;33m➜ http://cezenai.local\033[1;36m ║ ║ ║ ║ Complete setup from your browser — no keyboard ║ ║ input needed here. ║ ║ ║ ╚══════════════════════════════════════════════════════╝\033[0m """ # Write to tty1 (physical console) and stdout (journalctl) print(banner) try: with open("/dev/tty1", "w") as tty: tty.write(banner) except Exception: pass # tty1 may not be accessible in all environments # Also update /etc/issue so the URL appears above the login prompt try: with open("/etc/issue", "w") as f: f.write(f"Ubuntu 22.04.5 LTS \\n \\l\n\n") f.write(f" \033[1;36mNexus One AI Setup:\033[0m http://{ip} | http://cezenai.local\n\n") except Exception: pass if __name__ == "__main__": # Ensure log file exists open(INSTALL_LOG, "a").close() ip = get_ip() show_console_banner(ip) server = HTTPServer(("0.0.0.0", 80), Handler) server.serve_forever()