iDRAC not reachable via proxy
Target {idrac_ip} did not accept a connection from Cockpit.
Use an iDRAC / OOB endpoint (VLAN 40/41…), not the OS host IP.
{err}
"""
return Response(content=html, status_code=502, media_type="text/html")
raise HTTPException(502, f"iDRAC proxy unreachable: {e}") from e
raw = upstream.content
enc = (upstream.headers.get("content-encoding") or "").lower()
if "gzip" in enc and raw:
try:
raw = gzip.decompress(raw)
except Exception:
pass
elif "deflate" in enc and raw:
try:
import zlib
raw = zlib.decompress(raw)
except Exception:
try:
import zlib
raw = zlib.decompress(raw, -zlib.MAX_WBITS)
except Exception:
pass
ctype = (upstream.headers.get("content-type") or "").lower()
if "text/html" in ctype and raw:
raw = _idrac_inject_bootstrap(raw, device_id, idrac_ip)
elif "text/css" in ctype and raw:
raw = _idrac_rewrite_css(raw, device_id, idrac_ip)
elif ("javascript" in ctype or "ecmascript" in ctype) and raw:
# Light touch: rewrite absolute iDRAC URLs embedded as strings
try:
t = raw.decode("utf-8")
except UnicodeDecodeError:
t = raw.decode("latin-1")
t = t.replace(f"https://{idrac_ip}", prefix)
t = t.replace(f"wss://{idrac_ip}", f"ws://{request.url.hostname}:{request.url.port or (443 if request.url.scheme == 'https' else 80)}{prefix}")
raw = t.encode("utf-8")
out_headers: list[tuple[str, str]] = []
for k, v in upstream.headers.multi_items():
lk = k.lower()
if lk in _IDRAC_HOP_HEADERS:
continue
if lk == "location":
out_headers.append((k, _idrac_rewrite_location(v, device_id, idrac_ip)))
continue
if lk == "set-cookie":
out_headers.append((k, _idrac_rewrite_cookie(v, device_id)))
continue
out_headers.append((k, v))
# Allow embedding in Cockpit UI
out_headers.append(("Content-Security-Policy", "frame-ancestors *"))
out_headers.append(("Cache-Control", "no-store"))
resp = Response(content=raw, status_code=upstream.status_code)
for k, v in out_headers:
resp.headers.append(k, v)
if not any(k.lower() == "content-type" for k, _ in out_headers):
ct = upstream.headers.get("content-type")
if ct:
resp.headers["content-type"] = ct
return resp
@app.api_route(
"/api/idrac-proxy/{device_id}",
methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"],
)
@app.api_route(
"/api/idrac-proxy/{device_id}/{path:path}",
methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"],
)
async def idrac_http_proxy(device_id: int, request: Request, path: str = ""):
return await _idrac_proxy_http(device_id, path, request)
@app.websocket("/api/idrac-proxy/{device_id}/{path:path}")
async def idrac_ws_proxy(ws: WebSocket, device_id: int, path: str):
"""Bridge browser WebSocket ↔ iDRAC (virtual console / live sessions)."""
import ssl
import websockets
_, idrac_ip = _idrac_resolve(device_id)
await ws.accept()
qs = ws.scope.get("query_string", b"").decode()
target = f"wss://{idrac_ip}/{path}" + (f"?{qs}" if qs else "")
sub = ws.scope.get("subprotocols") or []
ssl_ctx = ssl._create_unverified_context()
try:
async with websockets.connect(
target,
ssl=ssl_ctx,
subprotocols=list(sub) if sub else None,
open_timeout=30,
max_size=8 * 1024 * 1024,
) as upstream:
async def client_to_upstream():
while True:
msg = await ws.receive()
if msg.get("type") == "websocket.disconnect":
break
if "text" in msg and msg["text"] is not None:
await upstream.send(msg["text"])
elif "bytes" in msg and msg["bytes"] is not None:
await upstream.send(msg["bytes"])
async def upstream_to_client():
async for message in upstream:
if isinstance(message, (bytes, bytearray)):
await ws.send_bytes(message)
else:
await ws.send_text(message)
_done, pending = await asyncio.wait(
[
asyncio.create_task(client_to_upstream()),
asyncio.create_task(upstream_to_client()),
],
return_when=asyncio.FIRST_COMPLETED,
)
for t in pending:
t.cancel()
except HTTPException:
try:
await ws.close()
except Exception:
pass
raise
except Exception as e:
log.warning("iDRAC WS proxy error device=%s: %s", device_id, e)
try:
await ws.close(code=1011)
except Exception:
pass
@app.get("/api/devices/{device_id}/expansion")
async def device_expansion(device_id: int, force: bool = False):
if force and device_id in DETAIL_CACHE:
DETAIL_CACHE.pop(device_id, None)
detail = await device_detail(device_id)
return {
"device": detail.get("device"),
"expansion": detail.get("expansion")
or build_expansion_report((detail.get("device") or {}).get("model"), detail.get("inventory") or {}),
}
@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 _device_by_id(device_id: int | None) -> dict | None:
if device_id is None:
return None
return next((d for d in (STATE.get("devices") or []) if d.get("id") == device_id), None)
def _device_by_service_tag(st: str) -> dict | None:
needle = (st or "").strip().upper()
if not needle:
return None
for d in STATE.get("devices") or []:
if (d.get("service_tag") or "").strip().upper() == needle:
return d
return None
def _find_devices_fuzzy(query: str, limit: int = 8) -> list[dict]:
q = (query or "").strip().lower()
if not q:
return []
hits = []
for d in STATE.get("devices") or []:
blob = " ".join(
str(d.get(k) or "").lower()
for k in ("name", "service_tag", "ip", "idrac_ip", "rdp_host", "os_hostname", "model")
)
if q in blob:
hits.append(d)
hits.sort(key=lambda d: (0 if (d.get("service_tag") or "").lower() == q else 1, d.get("name") or ""))
return hits[:limit]
def _format_device_card(d: dict) -> str:
mem = d.get("memory_gb")
mem_s = f"{mem}GB/{d.get('dimm_count') or '?'}DIMM" if mem is not None else "unknown"
return (
"ST={st} | name={name} | model={model} | idrac={idrac} | "
"rdp={rdp} | os_host={osh} | ram={ram} | connected={conn} | status={status} | watts={watts}"
).format(
st=(d.get("service_tag") or "NONE"),
name=(d.get("name") or "")[:48],
model=(d.get("model") or "")[:36],
idrac=d.get("idrac_ip") or d.get("ip") or "—",
rdp=(",".join(d.get("rdp_ips") or []) or d.get("rdp_host") or "unresolved"),
osh=d.get("os_hostname") or "—",
ram=mem_s,
conn="yes" if d.get("connected") else "no",
status=d.get("status"),
watts=d.get("watts"),
)
async def ome_fetch_inventory_types(device_id: int, types: list[str]) -> dict[str, list]:
"""Fast parallel inventory fetch for chat tools (not full device landscape)."""
inventory: dict[str, list] = {}
async with httpx.AsyncClient(verify=False, timeout=45.0) as client:
base, headers, sid = await ome_session(client)
try:
async def one(inv_type: str):
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:
return inv_type, info
except Exception as e:
log.debug("chat inv %s failed: %s", inv_type, e)
return inv_type, []
results = await asyncio.gather(*[one(t) for t in types])
for inv_type, info in results:
if info:
inventory[inv_type] = info
finally:
await ome_session_delete(client, base, headers, sid)
return inventory
def _memory_from_dimms(mem_raw: list) -> dict:
"""Compact CURRENT memory summary from OME serverMemoryDevices."""
dimms = 0
total_gb = 0.0
for dimm in mem_raw or []:
gb = _mem_size_gb(dimm.get("Size"))
if gb is None:
continue
dimms += 1
total_gb += gb
return {
"memory_gb": round(total_gb, 1) if dimms else None,
"dimm_count": dimms or None,
}
async def fetch_fleet_memory(force: bool = False) -> dict:
"""Warm/cache CURRENT installed RAM for all managed servers (OME inventory)."""
cached = None if force else _cache_get("fleet_memory")
if cached is not None:
return cached
devices = [
d
for d in (STATE.get("devices") or [])
if (d.get("is_server") or d.get("is_idrac") or (d.get("type") == 1000))
and d.get("source") != "inventory"
and isinstance(d.get("id"), int)
and d.get("id") > 0
]
# Prefer unique device ids
seen: set[int] = set()
targets: list[dict] = []
for d in devices:
did = d.get("id")
if did is None or did in seen:
continue
seen.add(int(did))
targets.append(d)
rows: list[dict] = []
errors = 0
sem = asyncio.Semaphore(10)
async with httpx.AsyncClient(verify=False, timeout=60.0) as client:
base, headers, sid = await ome_session(client)
try:
async def one(node: dict):
nonlocal errors
did = int(node["id"])
async with sem:
try:
ir = await client.get(
f"{base}/api/DeviceService/Devices({did})/InventoryDetails('serverMemoryDevices')",
headers=headers,
)
mem_raw = []
if ir.status_code == 200:
mem_raw = ir.json().get("InventoryInfo") or []
stats = _memory_from_dimms(mem_raw)
return {
"id": did,
"service_tag": node.get("service_tag"),
"name": node.get("name"),
"model": node.get("model"),
"idrac_ip": node.get("idrac_ip") or node.get("ip"),
"connected": bool(node.get("connected")),
"memory_gb": stats.get("memory_gb"),
"dimm_count": stats.get("dimm_count"),
}
except Exception as e:
errors += 1
log.debug("fleet memory %s: %s", did, e)
return {
"id": did,
"service_tag": node.get("service_tag"),
"name": node.get("name"),
"model": node.get("model"),
"idrac_ip": node.get("idrac_ip") or node.get("ip"),
"connected": bool(node.get("connected")),
"memory_gb": None,
"dimm_count": None,
"error": str(e)[:80],
}
rows = list(await asyncio.gather(*[one(n) for n in targets]))
finally:
await ome_session_delete(client, base, headers, sid)
rows.sort(key=lambda r: ((r.get("name") or "").lower(), r.get("service_tag") or ""))
known = [r for r in rows if r.get("memory_gb") is not None]
payload = {
"source": "OME DeviceService InventoryDetails(serverMemoryDevices)",
"count": len(rows),
"with_memory": len(known),
"errors": errors,
"total_memory_gb": round(sum(r.get("memory_gb") or 0 for r in known), 1),
"servers": rows,
"updated_at": time.time(),
}
_cache_set("fleet_memory", payload)
# also stamp onto fleet nodes for UI/context
by_id = {r["id"]: r for r in rows if r.get("id") is not None}
for d in STATE.get("devices") or []:
hit = by_id.get(d.get("id"))
if hit:
d["memory_gb"] = hit.get("memory_gb")
d["dimm_count"] = hit.get("dimm_count")
log.info(
"Fleet memory cache: %s/%s servers with RAM (%.0f GB total)",
len(known),
len(rows),
payload["total_memory_gb"],
)
return payload
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 {}
by_id = {d.get("id"): d for d in devices if d.get("id") is not None}
st_lines = [
"SERVICE TAG INDEX (canonical). idrac=BMC management IP. rdp=OS/Windows RDP candidates (DNS). Never call idrac the RDP IP.",
]
for d in sorted(devices, key=lambda x: (x.get("name") or "").lower()):
st = (d.get("service_tag") or "").strip() or "NONE"
rdp = ",".join((d.get("rdp_ips") or [])[:2]) or (d.get("rdp_host") or "—")
mem = d.get("memory_gb")
ram = f"{mem}GB" if mem is not None else "?"
st_lines.append(
"- ST={st} | {name} | idrac={idrac} | rdp={rdp} | model={model} | ram={ram} | connected={conn}".format(
st=st,
name=(d.get("name") or "")[:36],
idrac=d.get("idrac_ip") or d.get("ip") or "—",
rdp=rdp[:40],
model=(d.get("model") or "")[:26],
ram=ram,
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.",
"ATC Datacenter Admins (escalate here when facts are missing):",
" - Jody van Dongen