#!/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 from http.server import HTTPServer, BaseHTTPRequestHandler 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" install_proc = None install_status = {"running": False, "done": False, "error": None} # ─── 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 run_install(tier, skip_tools): global install_status install_status = {"running": True, "done": False, "error": None} try: # 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 "" 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" 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} except Exception as e: install_status = {"running": False, "done": False, "error": str(e)} # ─── HTML UI ────────────────────────────────────────────── HTML = r""" Nexus One AI — Server Setup
Server Setup Wizard
1 · Network
2 · Select Tier
3 · AI Tools
4 · 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, }) 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/install": global install_proc tier = body.get("tier", "basic") skip = body.get("skip_tools", []) if not install_status["running"]: t = threading.Thread(target=run_install, args=(tier, skip), daemon=True) t.start() self.send_json({"ok": True}) else: self.send_json({"ok": False, "error": "Install already running"}) 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╔══════════════════════════════════════════════════════╗ ║ ║ ║ CEZEN AI SUITE — 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()