import asyncio import json import os import sqlite3 import ipaddress import logging import time from collections import defaultdict from pathlib import Path from typing import Any import httpx import asyncssh from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, Response from pydantic import BaseModel, Field from pydantic_settings import BaseSettings logging.basicConfig(level=logging.INFO) log = logging.getLogger("ome-cockpit") class Settings(BaseSettings): ome_url: str = "https://cov-omeprod01.dell-atc.lan" ome_user: str = "admin" ome_password: str = "" poll_interval: float = 15.0 openwebui_url: str = "http://atc-portal01.dell-atc.lan:3080" cors_origins: str = "*" power_fetch_limit: int = 40 gpu_metrics_url: str = "http://10.0.10.106:9110" vllm_url: str = "http://10.0.10.106:8000/v1" vllm_model: str = "llama3-70b-gptq" vllm_max_model_len: int = 4096 vllm_max_tokens: int = 700 chat_system_chars: int = 5500 openwebui_email: str = "" openwebui_password: str = "" cockpit_data: str = "/data" class Config: env_file = ".env" settings = Settings() app = FastAPI(title="OME Cockpit API", version="1.2.0") app.add_middleware( CORSMiddleware, allow_origins=[o.strip() for o in settings.cors_origins.split(",")], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) STATE: dict[str, Any] = { "updated_at": 0, "ome": {}, "summary": {}, "subnets": [], "devices": [], "groups": [], "models": [], "alerts": [], "events": [], "context": {}, "gpu": {}, "pulse": 0, "openwebui_url": settings.openwebui_url, } CLIENTS: set[WebSocket] = set() SSH_SESSIONS: dict[str, dict] = {} SSH_MAX_SESSIONS = 25 SSH_MAX_PER_CLIENT = 3 _ssh_sem: asyncio.Semaphore | None = None _lock = asyncio.Lock() DETAIL_CACHE: dict[int, dict] = {} PREV_DEVICES: dict[int, dict] = {} EVENT_FEED: list[dict] = [] def subnet_of(ip: str | None) -> str: if not ip or ip.count(".") != 3: return "unknown" try: return str(ipaddress.ip_network(f"{ip}/24", strict=False)) except Exception: parts = ip.split(".") return f"{parts[0]}.{parts[1]}.{parts[2]}.0/24" def pick_ip(device: dict) -> str | None: for m in device.get("DeviceManagement") or []: addr = m.get("NetworkAddress") if addr and addr.count(".") == 3: return addr return None async def fetch_power(client: httpx.AsyncClient, base: str, headers: dict, device_id: int) -> dict: try: r = await client.get(f"{base}/api/DeviceService/Devices({device_id})/Power", headers=headers) if r.status_code != 200: return {} p = r.json() def num(key): v = p.get(key) try: return float(v) if v not in (None, "") else None except Exception: return None return { "watts": num("power"), "avg_watts": num("avgPower"), "peak_watts": num("peakPower"), "min_watts": num("minimumPower"), "energy_kwh": num("systemEnergyConsumption"), "power_unit": p.get("powerUnit") or "watt", } except Exception: return {} async def ome_fetch() -> dict: base = settings.ome_url.rstrip("/") async with httpx.AsyncClient(verify=False, timeout=60.0) as client: r = await client.post( f"{base}/api/SessionService/Sessions", json={ "UserName": settings.ome_user, "Password": settings.ome_password, "SessionType": "API", }, ) r.raise_for_status() token = r.headers.get("X-Auth-Token") sid = r.json().get("Id") headers = {"X-Auth-Token": token, "Accept": "application/json"} try: info = (await client.get(f"{base}/api/ApplicationService/Info", headers=headers)).json() devices_resp = ( await client.get(f"{base}/api/DeviceService/Devices?$top=5000", headers=headers) ).json() groups_resp = ( await client.get(f"{base}/api/GroupService/Groups?$top=200", headers=headers) ).json() alerts_raw = [] alerts_total = None try: alerts_resp = ( await client.get( f"{base}/api/AlertService/Alerts?$top=200", headers=headers, ) ).json() alerts_raw = alerts_resp.get("value") or [] alerts_total = alerts_resp.get("@odata.count") except Exception as e: log.warning("OME alerts fetch failed: %s", e) devices_raw = devices_resp.get("value") or [] # power for connected servers first power_targets = [ d.get("Id") for d in devices_raw if d.get("Type") == 1000 and d.get("ConnectionState") and d.get("Id") ][: settings.power_fetch_limit] # also a few offline servers for comparison offline_ids = [ d.get("Id") for d in devices_raw if d.get("Type") == 1000 and not d.get("ConnectionState") and d.get("Id") ][:5] power_targets = list(dict.fromkeys(power_targets + offline_ids)) sem = asyncio.Semaphore(8) async def limited(did): async with sem: return did, await fetch_power(client, base, headers, did) power_map = {} if power_targets: results = await asyncio.gather(*[limited(i) for i in power_targets]) power_map = {i: p for i, p in results if p} finally: if sid: try: await client.delete( f"{base}/api/SessionService/Sessions('{sid}')", headers=headers, ) except Exception: pass devices = [] subnet_map: dict[str, list] = defaultdict(list) model_counts: dict[str, int] = defaultdict(int) idrac_count = server_count = connected = powered = 0 total_watts = 0.0 watts_samples = 0 status_counts: dict[str, int] = defaultdict(int) for d in devices_raw: ip = pick_ip(d) subnet = subnet_of(ip) dtype = d.get("Type") sub = d.get("SubDeviceType") or "" is_server = dtype == 1000 is_idrac = sub == "iDRAC" conn = bool(d.get("ConnectionState")) power_state = d.get("PowerState") status = str(d.get("Status") or "") status_counts[status] += 1 model = d.get("Model") or "Unknown" if is_server: model_counts[model] += 1 if is_idrac: idrac_count += 1 if is_server: server_count += 1 if conn: connected += 1 if power_state == 17: powered += 1 pwr = power_map.get(d.get("Id")) or {} watts = pwr.get("watts") if watts is not None: total_watts += watts watts_samples += 1 node = { "id": d.get("Id"), "name": d.get("DeviceName") or f"device-{d.get('Id')}", "model": model, "service_tag": d.get("DeviceServiceTag"), "type": dtype, "sub_type": sub, "connected": conn, "power_state": power_state, "powered_on": power_state == 17, "status": status, "ip": ip, "subnet": subnet, "is_server": is_server, "is_idrac": is_idrac, "idrac_url": f"https://{ip}" if ip else None, "watts": watts, "avg_watts": pwr.get("avg_watts"), "peak_watts": pwr.get("peak_watts"), "energy_kwh": pwr.get("energy_kwh"), "last_status_time": d.get("LastStatusTime"), "last_inventory_time": d.get("LastInventoryTime"), } devices.append(node) subnet_map[subnet].append(node["id"]) subnets = [] for cidr, ids in sorted(subnet_map.items(), key=lambda x: -len(x[1])): members = [n for n in devices if n["id"] in ids] sub_watts = sum(n["watts"] or 0 for n in members if n.get("watts") is not None) subnets.append({ "cidr": cidr, "count": len(members), "connected": sum(1 for n in members if n["connected"]), "watts": round(sub_watts, 1) if sub_watts else None, "device_ids": ids, }) groups = [ {"id": g.get("Id"), "name": g.get("Name")} for g in (groups_resp.get("value") or []) if g.get("Name") ][:60] models = [ {"name": m, "count": c} for m, c in sorted(model_counts.items(), key=lambda x: -x[1]) ] # --- realtime context: fleet deltas + alerts --- global PREV_DEVICES, EVENT_FEED now = time.time() new_events: list[dict] = [] first_poll = not PREV_DEVICES current_ids: set[int] = set() def _role(n: dict) -> str: if n.get("is_idrac"): return "iDRAC" if n.get("is_server"): return "server" return "device" for n in devices: did = n.get("id") if did is None: continue current_ids.add(did) prev = PREV_DEVICES.get(did) if not prev: # First poll only seeds baseline; later polls raise new-node alerts if not first_poll and (n.get("is_server") or n.get("is_idrac")): role = _role(n) new_events.append({ "ts": now, "kind": "device_new", "severity": "critical", "device_id": did, "title": f"New {role} discovered", "text": ( f"{n.get('name')} · {n.get('ip') or 'no IP'} · " f"{n.get('model') or 'unknown model'} · {n.get('subnet') or 'unknown subnet'}" ), "notify": True, "role": role, "ip": n.get("ip"), "model": n.get("model"), "subnet": n.get("subnet"), "name": n.get("name"), }) continue if bool(prev.get("connected")) != bool(n.get("connected")): new_events.append({ "ts": now, "kind": "connection", "severity": "info" if n.get("connected") else "warning", "device_id": did, "title": n.get("name"), "text": "Connected" if n.get("connected") else "Went offline", }) if bool(prev.get("powered_on")) != bool(n.get("powered_on")): new_events.append({ "ts": now, "kind": "power_state", "severity": "info" if n.get("powered_on") else "warning", "device_id": did, "title": n.get("name"), "text": "Powered on" if n.get("powered_on") else "Powered off", }) pw, pp = n.get("watts"), prev.get("watts") if pw is not None and pp is not None and abs(pw - pp) >= 40: new_events.append({ "ts": now, "kind": "power_watt", "severity": "info", "device_id": did, "title": n.get("name"), "text": f"Power {round(pp)}W → {round(pw)}W", }) if str(prev.get("status")) != str(n.get("status")): new_events.append({ "ts": now, "kind": "status", "severity": "warning", "device_id": did, "title": n.get("name"), "text": f"Status {prev.get('status')} → {n.get('status')}", }) if not first_poll: for did, prev in PREV_DEVICES.items(): if did in current_ids: continue if not (prev.get("is_server") or prev.get("is_idrac")): continue role = _role(prev) new_events.append({ "ts": now, "kind": "device_removed", "severity": "warning", "device_id": did, "title": f"{role} left inventory", "text": ( f"{prev.get('name')} · {prev.get('ip') or 'no IP'} · " f"{prev.get('model') or 'unknown model'}" ), "notify": True, "role": role, "name": prev.get("name"), "ip": prev.get("ip"), "model": prev.get("model"), }) PREV_DEVICES = {n["id"]: n for n in devices if n.get("id") is not None} if new_events: EVENT_FEED = (new_events + EVENT_FEED)[:120] sev_map = {"Critical": 0, "Warning": 0, "Normal": 0, "Info": 0, "Unknown": 0} alerts = [] for a in alerts_raw: sev = a.get("SeverityName") or "Unknown" if sev not in sev_map: sev_map[sev] = 0 sev_map[sev] = sev_map.get(sev, 0) + 1 alerts.append({ "id": a.get("Id"), "severity": sev, "device_id": a.get("AlertDeviceId") or a.get("AlertEntityId"), "device": a.get("AlertDeviceName") or a.get("AlertEntityName"), "ip": a.get("AlertDeviceIpAddress"), "category": a.get("CategoryName"), "subcategory": a.get("SubCategoryName"), "message": a.get("Message"), "message_id": a.get("AlertMessageId"), "status": a.get("StatusName"), "time": a.get("TimeStamp"), "action": a.get("RecommendedAction"), }) hottest = sorted( [n for n in devices if n.get("watts") is not None], key=lambda x: x.get("watts") or 0, reverse=True, )[:5] recent_status = sorted( [n for n in devices if n.get("last_status_time")], key=lambda x: str(x.get("last_status_time")), reverse=True, )[:5] context = { "refreshed_at": now, "poll_seconds": settings.poll_interval, "alerts_total": alerts_total, "alert_severity": sev_map, "events_new": len(new_events), "notifications": [e for e in new_events if e.get("notify")], "hottest": [ {"id": n["id"], "name": n["name"], "watts": n["watts"], "subnet": n["subnet"]} for n in hottest ], "recent_status": [ { "id": n["id"], "name": n["name"], "time": n.get("last_status_time"), "connected": n.get("connected"), "status": n.get("status"), } for n in recent_status ], "focus_hint": ( f"{connected} connected · {round(total_watts)} W sampled · " f"{sev_map.get('Critical', 0)} critical / {sev_map.get('Warning', 0)} warning alerts in feed" ), } return { "ome": { "name": info.get("Name") or "OpenManage Enterprise", "version": info.get("Version"), "build": info.get("BuildNumber"), "url": base, "fqdn": "cov-omeprod01.dell-atc.lan", "console_url": f"{base}/management-console/ome/", }, "summary": { "total": len(devices), "idracs": idrac_count, "servers": server_count, "connected": connected, "powered_on": powered, "offline": max(0, server_count - connected), "status_counts": dict(status_counts), "total_watts": round(total_watts, 1), "avg_node_watts": round(total_watts / watts_samples, 1) if watts_samples else None, "power_samples": watts_samples, "alerts_critical": sev_map.get("Critical", 0), "alerts_warning": sev_map.get("Warning", 0), "alerts_total": alerts_total, }, "subnets": subnets, "devices": devices, "groups": groups, "models": models, "alerts": alerts, "events": EVENT_FEED[:40], "context": context, "openwebui_url": settings.openwebui_url, "updated_at": now, } async def broadcast(payload: dict): dead = [] for ws in list(CLIENTS): try: await ws.send_json(payload) except Exception: dead.append(ws) for ws in dead: CLIENTS.discard(ws) ADMINS = [ {"id": "jody", "name": "Jody van Dongen", "role": "ATC Datacenter Admin", "email": "jody.van.dongen@dell.com"}, {"id": "laurens", "name": "Laurens Rammers", "role": "ATC Datacenter Admin", "email": "laurens.rammers@dell.com"}, ] DATA_DIR = Path(settings.cockpit_data) DB_PATH = DATA_DIR / "ops.db" def _db(): DATA_DIR.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys=ON") return conn def init_db(): with _db() as conn: conn.executescript( """ CREATE TABLE IF NOT EXISTS tickets ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, body TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'open', priority TEXT NOT NULL DEFAULT 'normal', created_by TEXT NOT NULL, assignee TEXT, accepted_by TEXT, created_at REAL NOT NULL, updated_at REAL NOT NULL ); CREATE TABLE IF NOT EXISTS ticket_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, ticket_id INTEGER NOT NULL REFERENCES tickets(id) ON DELETE CASCADE, author TEXT NOT NULL, body TEXT NOT NULL, created_at REAL NOT NULL ); """ ) cols = {r[1] for r in conn.execute("PRAGMA table_info(tickets)").fetchall()} if "accepted_by" not in cols: conn.execute("ALTER TABLE tickets ADD COLUMN accepted_by TEXT") async def fetch_gpu() -> dict: url = settings.gpu_metrics_url.rstrip("/") + "/api/gpu" try: async with httpx.AsyncClient(timeout=4.0) as client: r = await client.get(url) if r.status_code == 200: return r.json() except Exception as e: log.warning("GPU metrics fetch failed: %s", e) return { "host": "atc-gpu-prod", "gpus": [], "error": "unreachable", "updated_at": time.time(), "model": settings.vllm_model, } async def poll_loop(): while True: try: data = await ome_fetch() gpu = await fetch_gpu() data["gpu"] = gpu async with _lock: STATE.update(data) STATE["pulse"] = int(STATE.get("pulse") or 0) + 1 await broadcast({"type": "snapshot", "data": {**STATE}}) log.info( "OME refresh: %s devices, %s connected, %s W sampled, GPU util avg %s", STATE["summary"].get("total"), STATE["summary"].get("connected"), STATE["summary"].get("total_watts"), (gpu.get("summary") or {}).get("avg_util"), ) except Exception as e: log.exception("OME poll failed: %s", e) await broadcast({"type": "error", "message": str(e)}) await asyncio.sleep(settings.poll_interval) async def gpu_loop(): """Faster GPU pulse so matrix feels live during inference.""" while True: try: gpu = await fetch_gpu() async with _lock: STATE["gpu"] = gpu await broadcast({"type": "gpu", "data": gpu}) except Exception as e: log.warning("gpu loop: %s", e) await asyncio.sleep(2.0) @app.on_event("startup") async def startup(): init_db() asyncio.create_task(poll_loop()) asyncio.create_task(gpu_loop()) @app.get("/api/health") async def health(): return { "status": "ok", "updated_at": STATE.get("updated_at"), "pulse": STATE.get("pulse"), } @app.get("/api/fleet") async def fleet(): return {**STATE} @app.get("/api/devices/{device_id}") async def device_detail(device_id: int): # lightweight detail: return fleet node + optional inventory slices node = next((d for d in STATE.get("devices") or [] if d.get("id") == device_id), None) if not node: raise HTTPException(404, "Device not found in current fleet snapshot") if device_id in DETAIL_CACHE and time.time() - DETAIL_CACHE[device_id].get("_ts", 0) < 300: return DETAIL_CACHE[device_id] base = settings.ome_url.rstrip("/") detail = {"device": node, "inventory": {}, "power": {}} async with httpx.AsyncClient(verify=False, timeout=60.0) as client: r = await client.post( f"{base}/api/SessionService/Sessions", json={ "UserName": settings.ome_user, "Password": settings.ome_password, "SessionType": "API", }, ) r.raise_for_status() token = r.headers.get("X-Auth-Token") sid = r.json().get("Id") headers = {"X-Auth-Token": token, "Accept": "application/json"} try: detail["power"] = await fetch_power(client, base, headers, device_id) # Discover all inventory types OME has for this device (app/firmware landscape) inv_types: list[str] = [] try: tr = await client.get( f"{base}/api/DeviceService/Devices({device_id})/InventoryTypes", headers=headers, ) if tr.status_code == 200: inv_types = list((tr.json() or {}).get("InventoryTypes") or []) except Exception: inv_types = [] # Ensure core + software landscape types are always attempted for required in ( "serverProcessors", "serverMemoryDevices", "serverArrayDisks", "serverPowerSupplies", "serverNetworkInterfaces", "deviceLocation", "deviceManagement", "subsystemRollupStatus", "serverOperatingSystems", "deviceSoftware", "deviceLicense", "deviceFru", "deviceCapabilities", "serverRaidControllers", "serverDeviceCards", "deviceBaseboards", "serverDellVideos", "serverFcCards", "serverVirtualFlashes", "serverStorageEnclosures", "serverSupportedPowerStates", "serverBiosSystemProfileSettings", "deviceInventory", ): if required not in inv_types: inv_types.append(required) detail["inventory_types"] = inv_types for inv_type in inv_types: try: ir = await client.get( f"{base}/api/DeviceService/Devices({device_id})/InventoryDetails('{inv_type}')", headers=headers, ) if ir.status_code == 200: info = ir.json().get("InventoryInfo") or [] if info: detail["inventory"][inv_type] = info except Exception: pass # Normalized application / firmware landscape view software = detail["inventory"].get("deviceSoftware") or [] os_info = detail["inventory"].get("serverOperatingSystems") or [] mgmt = detail["inventory"].get("deviceManagement") or [] licenses = detail["inventory"].get("deviceLicense") or [] def _stype(s: dict) -> str: return str(s.get("SoftwareType") or "").strip().upper() fw_types = {"BIOS", "FRMW", "FIRMWARE", "IDRAC", "USC", "BMC", "CPLD", "DCSM"} drv_types = {"DRVR", "DRIVER", "DRV"} firmware = [s for s in software if _stype(s) in fw_types or "FW" in _stype(s)] drivers = [s for s in software if _stype(s) in drv_types or "DRIVER" in _stype(s)] fw_ids = {id(s) for s in firmware} drv_ids = {id(s) for s in drivers} applications = [s for s in software if id(s) not in fw_ids and id(s) not in drv_ids] detail["landscape"] = { "os": os_info, "software": software, "management": mgmt, "licenses": licenses, "firmware": firmware, "drivers": drivers, "applications": applications, } finally: if sid: try: await client.delete( f"{base}/api/SessionService/Sessions('{sid}')", headers=headers, ) except Exception: pass detail["_ts"] = time.time() DETAIL_CACHE[device_id] = detail return detail @app.websocket("/ws/fleet") async def ws_fleet(ws: WebSocket): await ws.accept() CLIENTS.add(ws) try: await ws.send_json({"type": "snapshot", "data": {**STATE}}) while True: await ws.receive_text() except WebSocketDisconnect: pass finally: CLIENTS.discard(ws) class ChatIn(BaseModel): message: str history: list[dict] = Field(default_factory=list) focus_device_id: int | None = None model: str | None = None class TicketIn(BaseModel): title: str body: str created_by: str assignee: str | None = None priority: str = "normal" class TicketMsgIn(BaseModel): author: str body: str class TicketPatch(BaseModel): status: str | None = None assignee: str | None = None priority: str | None = None accepted_by: str | None = None def build_fleet_context(focus_device_id: int | None = None, max_chars: int | None = None) -> str: summary = STATE.get("summary") or {} gpu = STATE.get("gpu") or {} gsum = gpu.get("summary") or {} devices = STATE.get("devices") or [] alerts = STATE.get("alerts") or [] events = STATE.get("events") or [] subnets = STATE.get("subnets") or [] ctx = STATE.get("context") or {} limit = max_chars if max_chars is not None else settings.chat_system_chars connected = [d for d in devices if d.get("connected")] critical = [a for a in alerts if a.get("severity") == "Critical"][:8] warnings = [a for a in alerts if a.get("severity") == "Warning"][:5] hottest = (ctx.get("hottest") or [])[:5] lines = [ "You are OpenManage Cockpit Copilot for Dell ATC. Be concise and operational.", "Admins: Jody van Dongen, Laurens Rammers, Mohamed El Kadi.", "Use ONLY this live snapshot. Unknown => say unknown.", "", "FLEET: total={t} connected={c} offline={o} watts={w} alerts_total={a}".format( t=summary.get("total"), c=summary.get("connected"), o=summary.get("offline"), w=summary.get("total_watts"), a=summary.get("alerts_total"), ), "CONNECTED:", ] for d in connected[:18]: lines.append( "- {name} ip={ip} model={model} W={watts} st={status}".format( name=(d.get("name") or "")[:36], ip=d.get("ip"), model=(d.get("model") or "")[:28], watts=d.get("watts"), status=d.get("status"), ) ) lines.append("SUBNETS:") for s in subnets[:8]: lines.append( "- {cidr}: {conn}/{cnt} W={watts}".format( cidr=s.get("cidr"), conn=s.get("connected"), cnt=s.get("count"), watts=s.get("watts"), ) ) lines.append("CRITICAL:") if not critical: lines.append("(none)") for a in critical: lines.append( "- {dev} ({ip}): {msg}".format( dev=(a.get("device") or "")[:32], ip=a.get("ip"), msg=(a.get("message") or "")[:110], ) ) lines.append("WARNINGS:") for a in warnings: lines.append( "- {dev} ({ip}): {msg}".format( dev=(a.get("device") or "")[:32], ip=a.get("ip"), msg=(a.get("message") or "")[:90], ) ) lines.append("DELTAS:") if not events: lines.append("(none)") for e in events[:8]: lines.append("- {title}: {text}".format(title=e.get("title"), text=(e.get("text") or "")[:100])) lines.append("HOT POWER:") for h in hottest: lines.append("- {name}: {watts}W".format(name=(h.get("name") or "")[:28], watts=h.get("watts"))) lines.append( "GPU atc-gpu-prod: avg_util={u}% power={p}W mem={mu}/{mt}MB".format( u=gsum.get("avg_util"), p=gsum.get("total_power_w"), mu=gsum.get("total_mem_used_mb"), mt=gsum.get("total_mem_mb"), ) ) for g in (gpu.get("gpus") or [])[:8]: lines.append( "- GPU{i}: {util}% {temp}C {pw}W".format( i=g.get("index"), util=g.get("util_gpu"), temp=g.get("temp_c"), pw=g.get("power_w"), ) ) if focus_device_id is not None: node = next((d for d in devices if d.get("id") == focus_device_id), None) lines.append("FOCUS:") if node: lines.append( "{name} id={id} ip={ip} model={model} W={watts} connected={conn} status={st}".format( name=node.get("name"), id=node.get("id"), ip=node.get("ip"), model=node.get("model"), watts=node.get("watts"), conn=node.get("connected"), st=node.get("status"), ) ) related = [a for a in alerts if a.get("device_id") == focus_device_id][:5] for a in related: lines.append("alert {sev}: {msg}".format(sev=a.get("severity"), msg=(a.get("message") or "")[:100])) else: lines.append("device_id=%s missing" % focus_device_id) lines.append("Reply with concrete names, IPs, and next actions.") out = "\n".join(lines) if len(out) > limit: out = out[: limit - 20] + "\n…[truncated]" return out @app.get("/api/gpu") async def api_gpu(): if not STATE.get("gpu"): STATE["gpu"] = await fetch_gpu() return STATE.get("gpu") or {} @app.get("/api/admins") async def api_admins(): return {"admins": ADMINS} _OWUI_TOKEN: dict[str, Any] = {"token": None, "expires": 0.0} async def _owui_token() -> str | None: email = (settings.openwebui_email or "").strip() password = settings.openwebui_password or "" if not email or not password: return None now = time.time() if _OWUI_TOKEN.get("token") and float(_OWUI_TOKEN.get("expires") or 0) > now: return str(_OWUI_TOKEN["token"]) url = settings.openwebui_url.rstrip("/") + "/api/v1/auths/signin" try: async with httpx.AsyncClient(timeout=20.0) as client: r = await client.post(url, json={"email": email, "password": password}) if r.status_code >= 400: log.warning("Open WebUI signin failed: %s %s", r.status_code, r.text[:200]) return None token = (r.json() or {}).get("token") if token: _OWUI_TOKEN["token"] = token _OWUI_TOKEN["expires"] = now + 3500 return token except Exception as e: log.warning("Open WebUI signin error: %s", e) return None async def list_chat_models() -> list[dict]: models: list[dict] = [] seen: set[str] = set() token = await _owui_token() if token: try: async with httpx.AsyncClient(timeout=15.0) as client: r = await client.get( settings.openwebui_url.rstrip("/") + "/api/models", headers={"Authorization": f"Bearer {token}"}, ) if r.status_code == 200: for m in (r.json() or {}).get("data") or []: mid = m.get("id") if not mid or mid in seen: continue seen.add(mid) models.append( { "id": mid, "name": m.get("name") or mid, "source": "openwebui", "owned_by": m.get("owned_by"), } ) except Exception as e: log.warning("Open WebUI models failed: %s", e) try: async with httpx.AsyncClient(timeout=10.0) as client: r = await client.get(settings.vllm_url.rstrip("/") + "/models") if r.status_code == 200: for m in (r.json() or {}).get("data") or []: mid = m.get("id") if not mid or mid in seen: continue seen.add(mid) models.append( { "id": mid, "name": mid, "source": "vllm", "owned_by": m.get("owned_by") or "vllm", } ) except Exception as e: log.warning("vLLM models failed: %s", e) if not models: models.append( { "id": settings.vllm_model, "name": settings.vllm_model, "source": "vllm", "owned_by": "vllm", } ) return models @app.get("/api/models") async def api_models(): models = await list_chat_models() return {"models": models, "default": settings.vllm_model} @app.post("/api/chat") async def api_chat(payload: ChatIn): gpu = STATE.get("gpu") or {} gsum = gpu.get("summary") or {} model = (payload.model or settings.vllm_model or "").strip() or settings.vllm_model system = build_fleet_context(payload.focus_device_id, max_chars=settings.chat_system_chars) messages = [{"role": "system", "content": system}] for h in (payload.history or [])[-4:]: role = h.get("role") content = h.get("content") if role in ("user", "assistant") and content: messages.append({"role": role, "content": str(content)[:1200]}) messages.append({"role": "user", "content": payload.message[:2500]}) owui_models = {"ome-copilot", "arena-model"} use_owui = model in owui_models try: if use_owui: token = await _owui_token() if not token: raise HTTPException( 502, "Open WebUI auth not configured (set OPENWEBUI_EMAIL / OPENWEBUI_PASSWORD)", ) url = settings.openwebui_url.rstrip("/") + "/api/chat/completions" async with httpx.AsyncClient(timeout=180.0) as client: r = await client.post( url, headers={"Authorization": f"Bearer {token}"}, json={ "model": model, "messages": messages, "temperature": 0.2, "max_tokens": settings.vllm_max_tokens, "stream": False, }, ) if r.status_code >= 400: raise HTTPException(502, f"Open WebUI error {r.status_code}: {r.text[:500]}") data = r.json() backend = "openwebui" else: url = settings.vllm_url.rstrip("/") + "/chat/completions" async with httpx.AsyncClient(timeout=180.0) as client: r = await client.post( url, json={ "model": model, "messages": messages, "temperature": 0.2, "max_tokens": settings.vllm_max_tokens, }, ) if r.status_code >= 400: detail = r.text if r.status_code == 400 and "context length" in detail.lower(): tight = build_fleet_context(payload.focus_device_id, max_chars=3200) messages = [{"role": "system", "content": tight}, messages[-1]] r = await client.post( url, json={ "model": model, "messages": messages, "temperature": 0.2, "max_tokens": 500, }, ) detail = r.text if r.status_code >= 400: raise HTTPException(502, f"vLLM error {r.status_code}: {detail[:600]}") data = r.json() backend = "vllm" text_out = ( ((data.get("choices") or [{}])[0].get("message") or {}).get("content") or "" ) return { "reply": text_out, "model": model, "backend": backend, "gpu": gsum, "context_bytes": len(system), } except HTTPException: raise except Exception as e: raise HTTPException(502, f"Chat backend error: {e}") from e def _admin_name(aid: str) -> str: for a in ADMINS: if a["id"] == aid: return a["name"] return aid @app.get("/api/tickets") async def list_tickets(): with _db() as conn: rows = conn.execute( "SELECT * FROM tickets ORDER BY updated_at DESC LIMIT 200" ).fetchall() return {"tickets": [dict(r) for r in rows], "admins": ADMINS} @app.post("/api/tickets") async def create_ticket(payload: TicketIn): now = time.time() assignee = payload.assignee if not assignee: assignee = "laurens" if payload.created_by == "jody" else "jody" with _db() as conn: cur = conn.execute( """ INSERT INTO tickets(title, body, status, priority, created_by, assignee, created_at, updated_at) VALUES (?, ?, 'open', ?, ?, ?, ?, ?) """, ( payload.title.strip()[:200], payload.body.strip()[:5000], payload.priority, payload.created_by, assignee, now, now, ), ) tid = cur.lastrowid conn.execute( "INSERT INTO ticket_messages(ticket_id, author, body, created_at) VALUES (?, ?, ?, ?)", (tid, payload.created_by, payload.body.strip()[:5000], now), ) row = conn.execute("SELECT * FROM tickets WHERE id=?", (tid,)).fetchone() return dict(row) @app.get("/api/tickets/{ticket_id}") async def get_ticket(ticket_id: int): with _db() as conn: row = conn.execute("SELECT * FROM tickets WHERE id=?", (ticket_id,)).fetchone() if not row: raise HTTPException(404, "Ticket not found") msgs = conn.execute( "SELECT * FROM ticket_messages WHERE ticket_id=? ORDER BY created_at ASC", (ticket_id,), ).fetchall() return {"ticket": dict(row), "messages": [dict(m) for m in msgs], "admins": ADMINS} @app.post("/api/tickets/{ticket_id}/messages") async def add_ticket_message(ticket_id: int, payload: TicketMsgIn): now = time.time() with _db() as conn: row = conn.execute("SELECT id FROM tickets WHERE id=?", (ticket_id,)).fetchone() if not row: raise HTTPException(404, "Ticket not found") conn.execute( "INSERT INTO ticket_messages(ticket_id, author, body, created_at) VALUES (?, ?, ?, ?)", (ticket_id, payload.author, payload.body.strip()[:5000], now), ) conn.execute("UPDATE tickets SET updated_at=? WHERE id=?", (now, ticket_id)) msgs = conn.execute( "SELECT * FROM ticket_messages WHERE ticket_id=? ORDER BY created_at ASC", (ticket_id,), ).fetchall() return {"messages": [dict(m) for m in msgs]} @app.patch("/api/tickets/{ticket_id}") async def patch_ticket(ticket_id: int, payload: TicketPatch): now = time.time() with _db() as conn: row = conn.execute("SELECT * FROM tickets WHERE id=?", (ticket_id,)).fetchone() if not row: raise HTTPException(404, "Ticket not found") status = payload.status or row["status"] assignee = payload.assignee if payload.assignee is not None else row["assignee"] priority = payload.priority or row["priority"] # accepted_by: keep existing unless explicitly set; accepting sets status accepted accepted_by = row["accepted_by"] if "accepted_by" in row.keys() else None if payload.accepted_by is not None: accepted_by = payload.accepted_by if status == "open": status = "accepted" conn.execute( "UPDATE tickets SET status=?, assignee=?, priority=?, accepted_by=?, updated_at=? WHERE id=?", (status, assignee, priority, accepted_by, now, ticket_id), ) if payload.accepted_by: conn.execute( "INSERT INTO ticket_messages(ticket_id, author, body, created_at) VALUES (?, ?, ?, ?)", ( ticket_id, payload.accepted_by, f"Ticket accepted by {_admin_name(payload.accepted_by)}", now, ), ) row = conn.execute("SELECT * FROM tickets WHERE id=?", (ticket_id,)).fetchone() return dict(row) @app.delete("/api/tickets/{ticket_id}") async def delete_ticket(ticket_id: int): with _db() as conn: row = conn.execute("SELECT id FROM tickets WHERE id=?", (ticket_id,)).fetchone() if not row: raise HTTPException(404, "Ticket not found") conn.execute("DELETE FROM ticket_messages WHERE ticket_id=?", (ticket_id,)) conn.execute("DELETE FROM tickets WHERE id=?", (ticket_id,)) return {"ok": True, "deleted": ticket_id} def _get_ssh_sem() -> asyncio.Semaphore: global _ssh_sem if _ssh_sem is None: _ssh_sem = asyncio.Semaphore(SSH_MAX_SESSIONS) return _ssh_sem def _fleet_ssh_targets() -> set[str]: """Allow SSH only to IPs currently known in the OME fleet snapshot.""" allowed: set[str] = set() for d in STATE.get("devices") or []: ip = d.get("ip") if ip and ip.count(".") == 3: allowed.add(ip) return allowed @app.websocket("/ws/ssh") async def ws_ssh(ws: WebSocket): """Browser terminal <-> SSH bridge. Isolated per connection; concurrency-limited.""" await ws.accept() conn = None process = None reader_task = None session_id = f"{id(ws)}-{time.time()}" client_host = "" acquired = False async def send_json(payload: dict): try: await ws.send_text(json.dumps(payload)) except Exception: pass def _client_key() -> str: try: return ws.client.host if ws.client else "unknown" except Exception: return "unknown" try: # Limit concurrent sessions globally and per browser IP ck = _client_key() active_for_client = sum(1 for s in SSH_SESSIONS.values() if s.get("client") == ck) if active_for_client >= SSH_MAX_PER_CLIENT: await send_json({ "type": "error", "message": f"Too many SSH sessions from this client (max {SSH_MAX_PER_CLIENT})", }) await ws.close() return try: await asyncio.wait_for(_get_ssh_sem().acquire(), timeout=8.0) acquired = True except asyncio.TimeoutError: await send_json({ "type": "error", "message": "SSH gateway busy — try again in a moment", }) await ws.close() return SSH_SESSIONS[session_id] = { "client": ck, "started": time.time(), "host": None, "user": None, } raw = await asyncio.wait_for(ws.receive_text(), timeout=60.0) try: msg = json.loads(raw) except Exception: await send_json({"type": "error", "message": "Invalid auth payload"}) await ws.close() return if msg.get("type") != "auth": await send_json({"type": "error", "message": "Expected auth message"}) await ws.close() return host = str(msg.get("host") or "").strip() username = str(msg.get("username") or "").strip() password = str(msg.get("password") or "") try: port = int(msg.get("port") or 22) except Exception: port = 22 if not host or not username: await send_json({"type": "error", "message": "Host and username are required"}) await ws.close() return if port < 1 or port > 65535: await send_json({"type": "error", "message": "Invalid port"}) await ws.close() return async with _lock: allowed = set(_fleet_ssh_targets()) if host not in allowed: await send_json({ "type": "error", "message": f"Host {host} is not in the current OME fleet — SSH blocked", }) await ws.close() return client_host = host SSH_SESSIONS[session_id]["host"] = host SSH_SESSIONS[session_id]["user"] = username await send_json({"type": "status", "message": f"Connecting to {username}@{host}:{port}…"}) try: conn = await asyncio.wait_for( asyncssh.connect( host, port=port, username=username, password=password, known_hosts=None, client_keys=None, preferred_auth=["password", "keyboard-interactive"], keepalive_interval=30, keepalive_count_max=3, ), timeout=25.0, ) except Exception as e: await send_json({"type": "error", "message": f"SSH connect failed: {e}"}) await ws.close() return term = str(msg.get("term") or "xterm-256color") cols = int(msg.get("cols") or 120) rows = int(msg.get("rows") or 36) process = await conn.create_process( term_type=term, term_size=(max(cols, 40), max(rows, 10)), encoding="utf-8", errors="replace", ) await send_json({"type": "ready", "message": f"Connected as {username}@{host}"}) log.info("SSH session start id=%s client=%s target=%s@%s active=%s", session_id, ck, username, host, len(SSH_SESSIONS)) async def pump_ssh_to_ws(): try: while True: data = await process.stdout.read(8192) if not data: try: err = await asyncio.wait_for(process.stderr.read(1024), timeout=0.2) except Exception: err = "" if err: await ws.send_text(err if isinstance(err, str) else err.decode("utf-8", "replace")) break await ws.send_text(data if isinstance(data, str) else data.decode("utf-8", "replace")) except Exception: pass try: await send_json({"type": "status", "message": "SSH session ended"}) except Exception: pass reader_task = asyncio.create_task(pump_ssh_to_ws()) # Idle watchdog: close after 30 min without client input last_input = time.time() IDLE_LIMIT = 1800 while True: try: packet = await asyncio.wait_for(ws.receive(), timeout=30.0) except asyncio.TimeoutError: if time.time() - last_input > IDLE_LIMIT: await send_json({"type": "status", "message": "SSH idle timeout"}) break # keepalive ping to browser await send_json({"type": "pong"}) continue if packet.get("type") == "websocket.disconnect": break text = packet.get("text") if text is None: continue last_input = time.time() if text.startswith("{") and '"type"' in text[:48]: try: ctrl = json.loads(text) except Exception: process.stdin.write(text) await process.stdin.drain() continue ctype = ctrl.get("type") if ctype == "resize": try: process.change_terminal_size(int(ctrl.get("cols") or 80), int(ctrl.get("rows") or 24)) except Exception: pass elif ctype == "data": process.stdin.write(str(ctrl.get("data") or "")) await process.stdin.drain() elif ctype == "ping": await send_json({"type": "pong"}) continue process.stdin.write(text) await process.stdin.drain() except WebSocketDisconnect: pass except Exception as e: log.warning("SSH websocket error: %s", e) try: await send_json({"type": "error", "message": str(e)}) except Exception: pass finally: SSH_SESSIONS.pop(session_id, None) if reader_task: reader_task.cancel() try: await reader_task except Exception: pass try: if process: process.close() await asyncio.wait_for(process.wait_closed(), timeout=3.0) except Exception: pass try: if conn: conn.close() await asyncio.wait_for(conn.wait_closed(), timeout=3.0) except Exception: pass if acquired: try: _get_ssh_sem().release() except Exception: pass try: await ws.close() except Exception: pass log.info("SSH session end id=%s target=%s active=%s", session_id, client_host, len(SSH_SESSIONS)) @app.get("/api/ssh/sessions") async def api_ssh_sessions(): """Ops visibility: active SSH bridges (no secrets).""" now = time.time() sessions = [ { "id": sid, "client": meta.get("client"), "host": meta.get("host"), "user": meta.get("user"), "age_sec": int(now - float(meta.get("started") or now)), } for sid, meta in list(SSH_SESSIONS.items()) ] return {"active": len(sessions), "max": SSH_MAX_SESSIONS, "sessions": sessions} STATIC_DIR = Path("/ui") @app.get("/") async def index(): resp = FileResponse(STATIC_DIR / "index.html") resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate" resp.headers["Pragma"] = "no-cache" return resp @app.get("/styles.css") async def styles(): return FileResponse(STATIC_DIR / "styles.css", media_type="text/css") @app.get("/app.js") async def app_js(): return FileResponse(STATIC_DIR / "app.js", media_type="application/javascript") @app.get("/ops.js") async def ops_js(): return FileResponse(STATIC_DIR / "ops.js", media_type="application/javascript") @app.get("/ssh.js") async def ssh_js(): return FileResponse(STATIC_DIR / "ssh.js", media_type="application/javascript") @app.get("/vendor/xterm/xterm.css") async def vendor_xterm_css(): return FileResponse(STATIC_DIR / "vendor/xterm/xterm.css", media_type="text/css") @app.get("/vendor/xterm/xterm.min.js") async def vendor_xterm_js(): return FileResponse( STATIC_DIR / "vendor/xterm/xterm.min.js", media_type="application/javascript", ) @app.get("/vendor/xterm/xterm-addon-fit.min.js") async def vendor_xterm_fit_js(): return FileResponse( STATIC_DIR / "vendor/xterm/xterm-addon-fit.min.js", media_type="application/javascript", ) @app.get("/dell.png") async def dell_png(): return FileResponse(STATIC_DIR / "dell.png", media_type="image/png") @app.get("/dell-mark.png") async def dell_mark(): return FileResponse(STATIC_DIR / "dell-mark.png", media_type="image/png") @app.get("/dell.svg") async def dell_svg(): return FileResponse(STATIC_DIR / "dell.png", media_type="image/png")