diff --git a/netx_api/dsh_alarm_hub.py b/netx_api/dsh_alarm_hub.py new file mode 100644 index 0000000..233f9c9 --- /dev/null +++ b/netx_api/dsh_alarm_hub.py @@ -0,0 +1,212 @@ +"""DSH / netxops key-alarm subscribe hub (netx is the fixed-IP WebSocket server). + +Clients (netxops on each DSH host) dial out to: + ws[s]://:/v1/integrations/dsh-alarm/ws +authenticate with the same API token used for REST, then receive `netx.alarm` events. +""" + +from __future__ import annotations + +import asyncio +import logging +import threading +from datetime import datetime, timezone +from typing import Any + +from fastapi import WebSocket, WebSocketDisconnect +from starlette.websockets import WebSocketState + +from .auth_deps import resolve_user_from_token +from .auth_scopes import SCOPE_ALARMS_READ, has_scope +from .config import settings +from .db import SessionLocal + +_log = logging.getLogger("netx.dsh.alarm_hub") + +_LOCK = threading.Lock() +_CLIENTS: set[WebSocket] = set() +_LOOP: asyncio.AbstractEventLoop | None = None +_STATS = { + "published": 0, + "deliver_ok": 0, + "deliver_fail": 0, + "subscribers": 0, +} + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def bind_event_loop(loop: asyncio.AbstractEventLoop | None) -> None: + """Remember the API process loop so sync publishers can schedule sends.""" + global _LOOP + _LOOP = loop + + +def subscriber_count() -> int: + with _LOCK: + return len(_CLIENTS) + + +def hub_status() -> dict[str, Any]: + with _LOCK: + clients = len(_CLIENTS) + stats = dict(_STATS) + stats["subscribers"] = clients + return { + "enabled": True, + "path": "/v1/integrations/dsh-alarm/ws", + "subscribers": clients, + "published": int(stats.get("published") or 0), + "deliver_ok": int(stats.get("deliver_ok") or 0), + "deliver_fail": int(stats.get("deliver_fail") or 0), + } + + +def _authorize_token(token: str) -> tuple[bool, str]: + raw = str(token or "").strip() + if not bool(settings.auth_enabled): + return True, "auth_disabled" + if not raw: + return False, "token_required" + db = SessionLocal() + try: + hit = resolve_user_from_token(db, raw) + if hit is None: + return False, "invalid_token" + user, _via, scopes, _token_id, _jti = hit + if has_scope(scopes, SCOPE_ALARMS_READ) or str(user.role or "").lower() == "admin": + return True, str(user.username or user.id or "user") + return False, "missing_alarms_read_scope" + except Exception as exc: # noqa: BLE001 + _log.warning("dsh alarm hub auth error: %s", exc) + return False, "auth_error" + finally: + db.close() + + +async def _send_json(ws: WebSocket, payload: dict[str, Any]) -> bool: + if ws.client_state != WebSocketState.CONNECTED: + return False + try: + await ws.send_json(payload) + return True + except Exception: + return False + + +async def _broadcast(payload: dict[str, Any]) -> int: + with _LOCK: + clients = list(_CLIENTS) + if not clients: + return 0 + envelope = { + "type": "event", + "event": "netx.alarm", + "ts": _utc_now_iso(), + "payload": payload, + } + dead: list[WebSocket] = [] + ok = 0 + for ws in clients: + if await _send_json(ws, envelope): + ok += 1 + else: + dead.append(ws) + if dead: + with _LOCK: + for ws in dead: + _CLIENTS.discard(ws) + _STATS["subscribers"] = len(_CLIENTS) + _STATS["deliver_fail"] += len(dead) + for ws in dead: + try: + await ws.close() + except Exception: + pass + with _LOCK: + _STATS["deliver_ok"] += ok + return ok + + +def publish_alarm(payload: dict[str, Any]) -> int: + """Fan-out a matched key-alert payload to all connected DSH subscribers. + + Safe to call from sync UME/alarm threads. Returns the number of clients that + accepted the message (0 when nobody is subscribed). + """ + if not isinstance(payload, dict): + return 0 + with _LOCK: + _STATS["published"] += 1 + clients = len(_CLIENTS) + loop = _LOOP + if clients <= 0: + return 0 + if loop is None or not loop.is_running(): + _log.warning("dsh alarm hub has subscribers but no running event loop") + return 0 + + future = asyncio.run_coroutine_threadsafe(_broadcast(dict(payload)), loop) + try: + return int(future.result(timeout=5)) + except Exception as exc: # noqa: BLE001 + _log.warning("dsh alarm hub publish failed: %s", exc) + return 0 + + +async def dsh_alarm_ws_loop(websocket: WebSocket) -> None: + """Accept one netxops subscriber: auth → ping/pong → receive netx.alarm pushes.""" + await websocket.accept() + bind_event_loop(asyncio.get_running_loop()) + authed = False + try: + while True: + raw = await websocket.receive_text() + try: + import json + + msg = json.loads(raw) + except Exception: + await _send_json(websocket, {"type": "error", "error": "invalid_json"}) + continue + if not isinstance(msg, dict): + await _send_json(websocket, {"type": "error", "error": "invalid_message"}) + continue + mtype = str(msg.get("type") or "").strip().lower() + if not authed: + if mtype != "auth": + await _send_json(websocket, {"type": "auth-fail", "error": "auth_required"}) + await websocket.close(code=4401) + return + ok, detail = _authorize_token(str(msg.get("token") or "")) + if not ok: + await _send_json(websocket, {"type": "auth-fail", "error": detail}) + await websocket.close(code=4401) + return + authed = True + with _LOCK: + _CLIENTS.add(websocket) + _STATS["subscribers"] = len(_CLIENTS) + await _send_json( + websocket, + { + "type": "auth-ok", + "user": detail, + "ts": _utc_now_iso(), + }, + ) + _log.info("dsh alarm hub subscriber connected (%s)", detail) + continue + if mtype == "ping": + await _send_json(websocket, {"type": "pong", "ts": _utc_now_iso()}) + continue + await _send_json(websocket, {"type": "error", "error": f"unknown_type:{mtype}"}) + except WebSocketDisconnect: + return + finally: + with _LOCK: + _CLIENTS.discard(websocket) + _STATS["subscribers"] = len(_CLIENTS) + _log.info("dsh alarm hub subscriber disconnected") diff --git a/netx_api/integrations_router.py b/netx_api/integrations_router.py index 120454f..360aad9 100644 --- a/netx_api/integrations_router.py +++ b/netx_api/integrations_router.py @@ -5,17 +5,24 @@ from __future__ import annotations import time from typing import Any -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, WebSocket from sqlalchemy import text as sql_text from sqlalchemy.orm import Session from .config import settings from .db import get_db +from .dsh_alarm_hub import dsh_alarm_ws_loop, hub_status from .oclaw_alarm_forwarder import forwarder_status router = APIRouter(tags=["health"]) +@router.websocket("/v1/integrations/dsh-alarm/ws") +async def dsh_alarm_subscribe(websocket: WebSocket) -> None: + """netxops dials out here to receive matched key-alert pushes.""" + await dsh_alarm_ws_loop(websocket) + + @router.get("/health/live", status_code=200) def health_live() -> dict[str, str]: """Process liveness — no DB or upstream checks.""" @@ -130,4 +137,9 @@ def integrations_status(db: Session = Depends(get_db)) -> dict: "forwarder": fwd, } - return {"netx_api": netx_api, "db": db_status, "oclaw_bridge": oclaw_status} + return { + "netx_api": netx_api, + "db": db_status, + "oclaw_bridge": oclaw_status, + "dsh_alarm_hub": hub_status(), + } diff --git a/netx_api/key_alert_forward.py b/netx_api/key_alert_forward.py index ab991b1..ce3ecab 100644 --- a/netx_api/key_alert_forward.py +++ b/netx_api/key_alert_forward.py @@ -10,6 +10,7 @@ from sqlalchemy.orm import Session from .key_alert_matcher import match_key_alert_rule from .models import UmeInventoryNE, UmeKeyAlertForwardLog from .oclaw_alarm_forwarder import enqueue_alarm_forward, is_forwarder_operational +from .dsh_alarm_hub import publish_alarm from .ume_sync_service import ( _derive_ne_id_from_alarm, _pick, @@ -69,8 +70,6 @@ def maybe_forward_key_alert( alarm_key: str, action: str, ) -> bool: - if not is_forwarder_operational(): - return False rule = match_key_alert_rule(db, norm=norm, action=action) if rule is None: return False @@ -96,10 +95,23 @@ def maybe_forward_key_alert( payload["ne"] = _ne_payload(db, ne_id) payload["rule_key"] = str(rule.notification_id or "") - queued = enqueue_alarm_forward(payload) - if not queued: + hub_sent = publish_alarm(payload) + oclaw_queued = False + if is_forwarder_operational(): + oclaw_queued = bool(enqueue_alarm_forward(payload)) + if hub_sent <= 0 and not oclaw_queued: return False + # Hub-only delivery already reached DSH clients — mark ok for dedup. + # Oclaw path stays pending until the bridge records a result. + delivered_ok = hub_sent > 0 and not oclaw_queued + status = [] + if hub_sent > 0: + status.append(f"dsh_hub:{hub_sent}") + if oclaw_queued: + status.append("oclaw_queued") + status_text = ",".join(status) if status else "queued" + row = existing if row is None: row = UmeKeyAlertForwardLog( @@ -108,16 +120,16 @@ def maybe_forward_key_alert( rule_key=str(rule.notification_id or ""), notification_id=notification_id_from_norm(norm), forwarded_at=_utc_now_naive(), - oclaw_ok=0, - error="queued", + oclaw_ok=1 if delivered_ok else 0, + error="" if delivered_ok else status_text, ) db.add(row) else: row.notification_id = notification_id_from_norm(norm) row.rule_key = str(rule.notification_id or "") row.forwarded_at = _utc_now_naive() - row.oclaw_ok = 0 - row.error = "queued" + row.oclaw_ok = 1 if delivered_ok else 0 + row.error = "" if delivered_ok else status_text try: db.commit() except IntegrityError: diff --git a/netx_api/main.py b/netx_api/main.py index 74946a1..b365699 100644 --- a/netx_api/main.py +++ b/netx_api/main.py @@ -45,10 +45,14 @@ _BOOT_MONO = time.monotonic() @asynccontextmanager async def lifespan(_app: FastAPI) -> AsyncIterator[None]: + import asyncio + from .app_startup import run_api_startup from .app_shutdown import shutdown_runtime + from .dsh_alarm_hub import bind_event_loop run_api_startup() + bind_event_loop(asyncio.get_running_loop()) try: yield finally: