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 = 9000 openwebui_email: str = "" openwebui_password: str = "" cockpit_data: str = "/data" reports_cache_ttl: float = 600.0 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] = [] REPORT_CACHE: dict[str, Any] = { "warranties": None, "warranties_ts": 0.0, "compliance": None, "compliance_ts": 0.0, "baselines": None, "baselines_ts": 0.0, "catalogs": None, "catalogs_ts": 0.0, "report_defs": None, "report_defs_ts": 0.0, "jobs": None, "jobs_ts": 0.0, } async def ome_session(client: httpx.AsyncClient) -> tuple[str, dict, str]: """Create an OME API session; returns (base, headers, session_id).""" base = settings.ome_url.rstrip("/") 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 = str((r.json() or {}).get("Id") or "") headers = {"X-Auth-Token": token, "Accept": "application/json"} return base, headers, sid async def ome_session_delete(client: httpx.AsyncClient, base: str, headers: dict, sid: str) -> None: if not sid: return try: await client.delete(f"{base}/api/SessionService/Sessions('{sid}')", headers=headers) except Exception: pass def _cache_get(key: str) -> Any | None: ts = float(REPORT_CACHE.get(f"{key}_ts") or 0) if REPORT_CACHE.get(key) is not None and time.time() - ts < settings.reports_cache_ttl: return REPORT_CACHE.get(key) return None def _cache_set(key: str, value: Any) -> Any: REPORT_CACHE[key] = value REPORT_CACHE[f"{key}_ts"] = time.time() return value def csv_escape(val: Any) -> str: s = "" if val is None else str(val) if any(c in s for c in (",", '"', "\n", "\r")): return '"' + s.replace('"', '""') + '"' return s def rows_to_csv(rows: list[dict], columns: list[str] | None = None) -> str: if not rows and not columns: return "" cols = columns or list(rows[0].keys()) lines = [",".join(cols)] for row in rows: lines.append(",".join(csv_escape(row.get(c)) for c in cols)) return "\n".join(lines) + "\n" 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") or d.get("Identifier") or d.get("ChassisServiceTag"), "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()) asyncio.create_task(warm_reports_cache()) async def warm_reports_cache(): """Background warm of warranty/compliance for chat + inspector.""" await asyncio.sleep(8) try: await fetch_warranties() await fetch_compliance() log.info( "Reports cache warmed: warranties=%s compliance_outdated=%s", (REPORT_CACHE.get("warranties") or {}).get("count"), ((REPORT_CACHE.get("compliance") or {}).get("summary") or {}).get("outdated_devices"), ) except Exception as e: log.warning("Reports cache warm failed: %s", e) @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] compliance = _cache_get("compliance") or REPORT_CACHE.get("compliance") or {} comp_sum = (compliance or {}).get("summary") or {} # SERVICE TAG INDEX first — must survive truncation st_lines = [ "SERVICE TAG INDEX (use these Service Tags in every node answer):", ] for d in sorted(devices, key=lambda x: (x.get("name") or "").lower()): st = (d.get("service_tag") or "").strip() or "NONE" st_lines.append( "- ST={st} | {name} | ip={ip} | model={model} | connected={conn}".format( st=st, name=(d.get("name") or "")[:40], ip=d.get("ip") or "—", model=(d.get("model") or "")[:28], conn="yes" if d.get("connected") else "no", ) ) st_block = "\n".join(st_lines) 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.", "ALWAYS include Service Tag (ST=...), model, and IP when discussing any system.", "If a Service Tag is missing in the index, say so explicitly.", "", "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"), ), ] if comp_sum: lines.append( "FIRMWARE COMPLIANCE (Dell catalog baseline): outdated_devices={od} critical_components={cc} baseline={bn}".format( od=comp_sum.get("outdated_devices"), cc=comp_sum.get("critical_components"), bn=(comp_sum.get("baseline_name") or "")[:40], ) ) lines.append("CONNECTED:") for d in connected[:18]: lines.append( "- ST={st} {name} ip={ip} model={model} W={watts} status={status}".format( st=(d.get("service_tag") or "NONE"), 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( "ST={st} {name} id={id} ip={ip} model={model} W={watts} connected={conn} status={stt}".format( st=(node.get("service_tag") or "NONE"), name=node.get("name"), id=node.get("id"), ip=node.get("ip"), model=node.get("model"), watts=node.get("watts"), conn=node.get("connected"), stt=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 Service Tags, names, IPs, models, and next actions." ) body = "\n".join(lines) budget = max(800, limit - len(st_block) - 40) if len(body) > budget: body = body[: budget - 20] + "\n…[truncated]" out = st_block + "\n\n" + body 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)) # --------------------------------------------------------------------------- # Reports / compliance / warranty / export (read-only OME surfaces) # --------------------------------------------------------------------------- async def fetch_warranties(force: bool = False) -> dict: if not force: cached = _cache_get("warranties") if cached is not None: return cached rows: list[dict] = [] async with httpx.AsyncClient(verify=False, timeout=60.0) as client: base, headers, sid = await ome_session(client) try: skip = 0 while True: r = await client.get( f"{base}/api/WarrantyService/Warranties", headers=headers, params={"$top": 200, "$skip": skip}, ) r.raise_for_status() chunk = (r.json() or {}).get("value") or [] if not chunk: break for w in chunk: rows.append( { "id": w.get("Id"), "device_id": w.get("DeviceId"), "service_tag": w.get("DeviceIdentifier"), "device_name": w.get("DeviceName"), "model": w.get("DeviceModel"), "service_level": w.get("ServiceLevelDescription") or w.get("ServiceLevelCode"), "start_date": w.get("StartDate"), "end_date": w.get("EndDate"), "days_remaining": w.get("DaysRemaining"), "ship_date": w.get("SystemShipDate"), "state": w.get("State"), "country": w.get("CountryLookupCode"), } ) skip += len(chunk) if len(chunk) < 200: break finally: await ome_session_delete(client, base, headers, sid) # Prefer longest-remaining warranty per device by_device: dict[int, dict] = {} for w in rows: did = w.get("device_id") if did is None: continue prev = by_device.get(did) if not prev or int(w.get("days_remaining") or 0) > int(prev.get("days_remaining") or 0): by_device[did] = w elif int(w.get("days_remaining") or 0) == int(prev.get("days_remaining") or 0): # keep later end date if str(w.get("end_date") or "") > str(prev.get("end_date") or ""): by_device[did] = w payload = { "updated_at": time.time(), "count": len(rows), "items": rows, "by_device": {str(k): v for k, v in by_device.items()}, } return _cache_set("warranties", payload) async def fetch_baselines(force: bool = False) -> dict: if not force: cached = _cache_get("baselines") if cached is not None: return cached async with httpx.AsyncClient(verify=False, timeout=60.0) as client: base, headers, sid = await ome_session(client) try: r = await client.get(f"{base}/api/UpdateService/Baselines", headers=headers) r.raise_for_status() items = [] for b in (r.json() or {}).get("value") or []: summary = b.get("ComplianceSummary") or {} items.append( { "id": b.get("Id"), "name": b.get("Name"), "description": b.get("Description"), "catalog_id": b.get("CatalogId"), "catalog_type": b.get("CatalogType"), "last_run": b.get("LastRun"), "task_id": b.get("TaskId"), "compliance_status": summary.get("ComplianceStatus"), "critical": summary.get("NumberOfCritical"), "warning": summary.get("NumberOfWarning"), "normal": summary.get("NumberOfNormal"), "downgrade": summary.get("NumberOfDowngrade"), "unknown": summary.get("NumberOfUnknown"), } ) finally: await ome_session_delete(client, base, headers, sid) return _cache_set("baselines", {"updated_at": time.time(), "items": items}) async def fetch_catalogs(force: bool = False) -> dict: if not force: cached = _cache_get("catalogs") if cached is not None: return cached async with httpx.AsyncClient(verify=False, timeout=60.0) as client: base, headers, sid = await ome_session(client) try: r = await client.get(f"{base}/api/UpdateService/Catalogs", headers=headers) r.raise_for_status() items = [] for c in (r.json() or {}).get("value") or []: repo = c.get("Repository") or {} items.append( { "id": c.get("Id"), "filename": c.get("Filename") or c.get("SourcePath"), "source": repo.get("Source") or c.get("Source"), "repository_name": repo.get("Name"), "repository_type": repo.get("RepositoryType") or c.get("RepositoryType"), "owner": c.get("Owner"), "status": (c.get("Status") or {}).get("Name") if isinstance(c.get("Status"), dict) else c.get("Status"), "last_updated": c.get("LastUpdated") or c.get("BundlesUpdateTime"), } ) finally: await ome_session_delete(client, base, headers, sid) return _cache_set("catalogs", {"updated_at": time.time(), "items": items}) def _pick_primary_baseline(baselines: list[dict]) -> dict | None: if not baselines: return None for b in baselines: name = (b.get("name") or "").lower() if "dell" in name and "online" in name: return b # most critical first return sorted(baselines, key=lambda x: int(x.get("critical") or 0), reverse=True)[0] async def fetch_compliance(force: bool = False, baseline_id: int | None = None) -> dict: if not force and baseline_id is None: cached = _cache_get("compliance") if cached is not None: return cached bl = await fetch_baselines(force=force) primary = None if baseline_id is not None: primary = next((b for b in bl["items"] if b.get("id") == baseline_id), None) if primary is None: primary = _pick_primary_baseline(bl["items"]) if not primary: payload = { "updated_at": time.time(), "summary": {"outdated_devices": 0, "critical_components": 0, "baseline_name": None}, "devices": [], "components": [], "baselines": bl["items"], } return _cache_set("compliance", payload) bid = primary["id"] devices_out: list[dict] = [] components_out: list[dict] = [] fleet_by_id = {d.get("id"): d for d in STATE.get("devices") or []} async with httpx.AsyncClient(verify=False, timeout=120.0) as client: base, headers, sid = await ome_session(client) try: r = await client.get( f"{base}/api/UpdateService/Baselines({bid})/DeviceComplianceReports", headers=headers, ) r.raise_for_status() for dcr in (r.json() or {}).get("value") or []: did = dcr.get("DeviceId") fleet = fleet_by_id.get(did) or {} st = dcr.get("ServiceTag") or fleet.get("service_tag") comps = dcr.get("ComponentComplianceReports") or [] device_row = { "device_id": did, "service_tag": st, "name": fleet.get("name") or dcr.get("DeviceName") or st, "model": dcr.get("DeviceModel") or fleet.get("model"), "ip": fleet.get("ip"), "firmware_status": dcr.get("FirmwareStatus"), "compliance_status": dcr.get("ComplianceStatus"), "reboot_required": dcr.get("RebootRequired"), "component_count": len(comps), "dell_uri": None, } devices_out.append(device_row) for comp in comps: uri = comp.get("Uri") if uri and not device_row["dell_uri"]: device_row["dell_uri"] = uri components_out.append( { "device_id": did, "service_tag": st, "device_name": device_row["name"], "model": device_row["model"], "ip": device_row["ip"], "component": comp.get("Name"), "component_type": comp.get("ComponentType"), "current_version": comp.get("CurrentVersion"), "catalog_version": comp.get("Version"), "update_action": comp.get("UpdateAction"), "criticality": comp.get("Criticality"), "compliance_status": comp.get("ComplianceStatus"), "reboot_required": comp.get("RebootRequired"), "dell_uri": uri, "path": comp.get("Path"), "baseline_id": bid, "baseline_name": primary.get("name"), "status_badge": ( "outdated" if str(comp.get("UpdateAction") or "").upper() == "UPGRADE" or str(comp.get("ComplianceStatus") or "").upper() in ("CRITICAL", "WARNING", "NONCOMPLIANT", "NON-COMPLIANT") else "current" if str(comp.get("UpdateAction") or "").upper() in ("EQUAL", "NONE", "") and str(comp.get("ComplianceStatus") or "").upper() in ("", "OK", "COMPLIANT", "NORMAL", "DOWNGRADE") else "unknown" ), } ) finally: await ome_session_delete(client, base, headers, sid) outdated = [ d for d in devices_out if str(d.get("compliance_status") or "").upper() in ("CRITICAL", "WARNING") or str(d.get("firmware_status") or "").lower() in ("non-compliant", "noncompliant") ] crit_comps = [ c for c in components_out if str(c.get("compliance_status") or "").upper() == "CRITICAL" or str(c.get("update_action") or "").upper() == "UPGRADE" ] payload = { "updated_at": time.time(), "baseline": primary, "baselines": bl["items"], "summary": { "baseline_id": bid, "baseline_name": primary.get("name"), "outdated_devices": len(outdated), "devices_in_report": len(devices_out), "critical_components": len(crit_comps), "components_total": len(components_out), "compliance_status": primary.get("compliance_status"), "last_run": primary.get("last_run"), }, "devices": devices_out, "components": components_out, } if baseline_id is None: return _cache_set("compliance", payload) return payload async def fetch_report_defs(force: bool = False) -> dict: if not force: cached = _cache_get("report_defs") if cached is not None: return cached items = [] async with httpx.AsyncClient(verify=False, timeout=60.0) as client: base, headers, sid = await ome_session(client) try: r = await client.get(f"{base}/api/ReportService/ReportDefs", headers=headers) r.raise_for_status() for d in (r.json() or {}).get("value") or []: cols = [c.get("Name") for c in (d.get("ColumnNames") or []) if c.get("Name")] items.append( { "id": d.get("Id"), "name": d.get("Name"), "description": d.get("Description"), "category": d.get("Category") or d.get("FilterGroupName"), "is_builtin": d.get("IsBuiltIn"), "last_run": d.get("LastRunDate"), "last_run_by": d.get("LastRunBy"), "columns": cols, } ) finally: await ome_session_delete(client, base, headers, sid) items.sort(key=lambda x: ((x.get("category") or ""), (x.get("name") or "").lower())) return _cache_set("report_defs", {"updated_at": time.time(), "items": items}) async def fetch_jobs(force: bool = False, top: int = 40) -> dict: if not force: cached = _cache_get("jobs") if cached is not None: return cached items = [] async with httpx.AsyncClient(verify=False, timeout=60.0) as client: base, headers, sid = await ome_session(client) try: r = await client.get( f"{base}/api/JobService/Jobs", headers=headers, params={"$top": top}, ) r.raise_for_status() for j in (r.json() or {}).get("value") or []: status = j.get("LastRunStatus") or {} items.append( { "id": j.get("Id"), "name": j.get("JobName") or j.get("Name"), "status": status.get("Name") if isinstance(status, dict) else status, "job_type": (j.get("JobType") or {}).get("Name") if isinstance(j.get("JobType"), dict) else j.get("JobType"), "last_run": j.get("LastRunStatus") and (j.get("LastRunDate") or j.get("StartTime")), "progress": j.get("Progress") or j.get("PercentComplete"), "created_by": j.get("CreatedBy"), } ) finally: await ome_session_delete(client, base, headers, sid) return _cache_set("jobs", {"updated_at": time.time(), "items": items}) def build_fleet_report_rows(warranties: dict | None = None) -> list[dict]: by_dev = (warranties or {}).get("by_device") or {} rows = [] for d in STATE.get("devices") or []: w = by_dev.get(str(d.get("id"))) or {} rows.append( { "id": d.get("id"), "name": d.get("name"), "service_tag": d.get("service_tag"), "model": d.get("model"), "ip": d.get("ip"), "subnet": d.get("subnet"), "type": d.get("type"), "sub_type": d.get("sub_type"), "is_server": d.get("is_server"), "is_idrac": d.get("is_idrac"), "connected": d.get("connected"), "powered_on": d.get("powered_on"), "power_state": d.get("power_state"), "status": d.get("status"), "watts": d.get("watts"), "avg_watts": d.get("avg_watts"), "peak_watts": d.get("peak_watts"), "last_status_time": d.get("last_status_time"), "last_inventory_time": d.get("last_inventory_time"), "warranty_end": w.get("end_date"), "warranty_days_remaining": w.get("days_remaining"), "warranty_service_level": w.get("service_level"), } ) rows.sort(key=lambda r: (r.get("name") or "").lower()) return rows @app.get("/api/warranties") async def api_warranties(force: bool = False): return await fetch_warranties(force=force) @app.get("/api/compliance") async def api_compliance(force: bool = False, baseline_id: int | None = None): return await fetch_compliance(force=force, baseline_id=baseline_id) @app.get("/api/baselines") async def api_baselines(force: bool = False): return await fetch_baselines(force=force) @app.get("/api/catalogs") async def api_catalogs(force: bool = False): return await fetch_catalogs(force=force) @app.get("/api/ome/report-defs") async def api_report_defs(force: bool = False): return await fetch_report_defs(force=force) @app.get("/api/ome/jobs") async def api_ome_jobs(force: bool = False): return await fetch_jobs(force=force) class ReportRunIn(BaseModel): report_def_id: int @app.post("/api/ome/reports/run") async def api_ome_report_run(payload: ReportRunIn): async with httpx.AsyncClient(verify=False, timeout=60.0) as client: base, headers, sid = await ome_session(client) try: r = await client.post( f"{base}/api/ReportService/Actions/ReportService.RunReport", headers=headers, json={"ReportDefId": payload.report_def_id}, ) if r.status_code >= 400: raise HTTPException(r.status_code, r.text[:500]) job_id = r.json() if isinstance(r.json(), (int, str)) else (r.json() or {}).get("Id") or r.text return { "job_id": job_id, "report_def_id": payload.report_def_id, "message": "Report job started in OME. Poll results shortly.", "results_url": f"/api/ome/reports/{payload.report_def_id}/results", } finally: await ome_session_delete(client, base, headers, sid) @app.get("/api/ome/reports/{report_def_id}/results") async def api_ome_report_results(report_def_id: int): async with httpx.AsyncClient(verify=False, timeout=60.0) as client: base, headers, sid = await ome_session(client) try: r = await client.get( f"{base}/api/ReportService/ReportDefs({report_def_id})/ReportResults", headers=headers, ) if r.status_code >= 400: raise HTTPException( r.status_code, (r.json().get("error", {}) or {}).get("message") if r.headers.get("content-type", "").startswith("application/json") else r.text[:500], ) return r.json() finally: await ome_session_delete(client, base, headers, sid) @app.get("/api/reports/fleet") async def api_reports_fleet(force: bool = False): warranties = await fetch_warranties(force=force) rows = build_fleet_report_rows(warranties) return { "updated_at": STATE.get("updated_at"), "summary": STATE.get("summary"), "ome": STATE.get("ome"), "count": len(rows), "rows": rows, } @app.get("/api/reports/firmware") async def api_reports_firmware(force: bool = False, baseline_id: int | None = None): return await fetch_compliance(force=force, baseline_id=baseline_id) @app.get("/api/reports/brief") async def api_reports_brief(force: bool = False): warranties = await fetch_warranties(force=force) compliance = await fetch_compliance(force=force) fleet_rows = build_fleet_report_rows(warranties) critical = [a for a in (STATE.get("alerts") or []) if a.get("severity") == "Critical"][:15] expiring = sorted( [ w for w in (warranties.get("items") or []) if w.get("days_remaining") is not None and int(w.get("days_remaining") or 0) <= 90 ], key=lambda x: int(x.get("days_remaining") or 0), )[:25] outdated = [ d for d in (compliance.get("devices") or []) if str(d.get("compliance_status") or "").upper() in ("CRITICAL", "WARNING") or str(d.get("firmware_status") or "").lower() in ("non-compliant", "noncompliant") ] return { "generated_at": time.time(), "ome": STATE.get("ome"), "summary": STATE.get("summary"), "compliance_summary": compliance.get("summary"), "critical_alerts": critical, "outdated_devices": outdated[:40], "warranty_expiring": expiring, "fleet": fleet_rows, "baseline": compliance.get("baseline"), } @app.get("/api/export/{kind}") async def api_export(kind: str, fmt: str = "csv", force: bool = False, baseline_id: int | None = None): kind = kind.lower().strip() fmt = fmt.lower().strip() if fmt not in ("csv", "json"): raise HTTPException(400, "fmt must be csv or json") if kind in ("fleet", "inventory"): data = await api_reports_fleet(force=force) rows = data["rows"] filename = f"ome-fleet-{int(time.time())}" elif kind in ("firmware", "compliance"): data = await fetch_compliance(force=force, baseline_id=baseline_id) rows = data.get("components") or [] filename = f"ome-firmware-compliance-{int(time.time())}" elif kind == "warranty": data = await fetch_warranties(force=force) rows = data.get("items") or [] filename = f"ome-warranties-{int(time.time())}" elif kind == "brief": data = await api_reports_brief(force=force) if fmt == "json": return Response( content=json.dumps(data, indent=2, default=str), media_type="application/json", headers={"Content-Disposition": f'attachment; filename="ome-customer-brief-{int(time.time())}.json"'}, ) # flatten brief as fleet CSV appendix rows = data.get("fleet") or [] filename = f"ome-customer-brief-fleet-{int(time.time())}" else: raise HTTPException(404, "Unknown export kind. Use fleet|firmware|warranty|brief") if fmt == "json": body = json.dumps({"kind": kind, "count": len(rows), "rows": rows}, indent=2, default=str) return Response( content=body, media_type="application/json", headers={"Content-Disposition": f'attachment; filename="{filename}.json"'}, ) csv_body = rows_to_csv(rows) return Response( content=csv_body, media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{filename}.csv"'}, ) @app.get("/api/devices/{device_id}/warranty") async def device_warranty(device_id: int, force: bool = False): warranties = await fetch_warranties(force=force) items = [w for w in warranties.get("items") or [] if w.get("device_id") == device_id] primary = (warranties.get("by_device") or {}).get(str(device_id)) return {"device_id": device_id, "primary": primary, "items": items} @app.get("/api/devices/{device_id}/compliance") async def device_compliance(device_id: int, force: bool = False): compliance = await fetch_compliance(force=force) device = next((d for d in compliance.get("devices") or [] if d.get("device_id") == device_id), None) comps = [c for c in compliance.get("components") or [] if c.get("device_id") == device_id] return { "device_id": device_id, "baseline": compliance.get("baseline"), "device": device, "components": comps, } @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("/reports.js") async def reports_js(): return FileResponse(STATIC_DIR / "reports.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")