vulncheck/app/integrations/igel_client.py

"""IGEL UMS client (IGEL Management Interface, IMI API v3).

Thin httpx wrapper, no DB writes — same contract as the other integration
clients. Three things are read and nothing else:

  * `GET /v3/serverstatus` — the UMS server's own identity (`rmGuiServerVersion`,
    `buildNumber`, `serverUUID`). The only endpoint that answers without a
    login, which is what makes it a usable reachability probe.
  * `GET /v3/firmwares` — every firmware registered with this UMS, as
    {id: product, version, firmwareType}. Small list (one row per firmware
    ever imported), and the ONLY place the version string lives.
  * `GET /v3/thinclients?facets=details` — every endpoint device in one call,
    with its `firmwareID`, unit ID, MAC, last IP and hardware details.

Why the firmware table is fetched separately: the device record carries a
`firmwareID`, not a version — "21", not "11.09.100.01". Joining the two is
what turns an inventory into something a CVE range can be compared against,
and it costs exactly one extra request for the whole estate. The alternative
(`/v3/thinclients/{id}` per device, then `/v3/firmwares/{fwId}` per device) is
two calls per thin client; a 3000-device estate would make 6000.

Auth: HTTP Basic on POST /v3/login, which returns a JSESSIONID cookie valid
for 30 minutes, refreshed by every request. httpx.Client keeps the cookie jar,
so the login is a single call and everything after it just works.

Read-only throughout: a UMS account with the built-in read-only permissions is
enough — this client never POSTs a command to a device. IMI has to be enabled
and licensed on the UMS server first (UMS Console → Network → IMI).

API reference: https://kb.igel.com/en/igel-management-interface/current
"""
from __future__ import annotations

import logging
from typing import Dict, List, Optional

import httpx

logger = logging.getLogger(__name__)


class IgelError(RuntimeError):
    """Connection / authentication failure, with the IMI message kept."""


class IgelClient:
    def __init__(self, host: str, username: str, password: str,
                 port: int = 8443, verify_ssl: bool = True):
        self.host = (host or "").strip().replace("https://", "").replace("http://", "").rstrip("/")
        self.username = username
        self.password = password
        self.port = int(port or 8443)
        self.verify_ssl = verify_ssl
        self.base_url = f"https://{self.host}:{self.port}/umsapi/v3"
        # Short connect timeout, generous read: the same reasoning as the
        # Nessus client — an unreachable UMS must fail fast instead of holding
        # a worker, while `?facets=details` over a large estate is genuinely
        # slow to serialise.
        self._client = httpx.Client(
            base_url=self.base_url,
            verify=verify_ssl,
            timeout=httpx.Timeout(connect=8.0, read=60.0, write=60.0, pool=5.0),
            headers={"Accept": "application/json"},
        )
        self._logged_in = False

    # ---------- connection ----------
    def login(self) -> None:
        if self._logged_in:
            return
        try:
            r = self._client.post("/login", auth=(self.username, self.password))
        except httpx.HTTPError as e:
            raise IgelError(f"IMI login failed for {self.host}: {e}") from e
        if r.status_code == 401:
            raise IgelError("IMI login rejected the credentials (HTTP 401). "
                            "Check the UMS account, and that it is allowed to "
                            "use IMI (UMS Console → Network → IMI).")
        if r.status_code >= 400:
            raise IgelError(f"IMI login failed (HTTP {r.status_code}): {_imi_error(r)}")
        self._logged_in = True

    def close(self) -> None:
        if self._logged_in:
            try:
                self._client.post("/logout")
            except Exception:
                pass
            self._logged_in = False
        self._client.close()

    def _get(self, path: str) -> object:
        self.login()
        try:
            r = self._client.get(path)
        except httpx.HTTPError as e:
            raise IgelError(f"IMI GET {path} failed: {e}") from e
        if r.status_code >= 400:
            raise IgelError(f"IMI GET {path} failed (HTTP {r.status_code}): {_imi_error(r)}")
        try:
            return r.json()
        except ValueError as e:
            raise IgelError(f"IMI GET {path} returned no JSON") from e

    # ---------- reads ----------
    def get_server_status(self) -> dict:
        """The UMS server's own identity. No login required (by design)."""
        try:
            r = self._client.get("/serverstatus")
        except httpx.HTTPError as e:
            raise IgelError(f"IMI unreachable at {self.base_url}: {e}") from e
        if r.status_code >= 400:
            raise IgelError(f"IMI serverstatus failed (HTTP {r.status_code}): {_imi_error(r)}")
        d = r.json() if r.content else {}
        out = {
            "version": _first(d, _STATUS_VERSION_KEYS),
            "build": _first(d, _STATUS_BUILD_KEYS),
            "server_uuid": _first(d, ("serverUUID", "serverUuid", "uuid")),
            "server": d.get("server"),
        }
        if not out["version"]:
            # Said out loud, with the keys the server actually sent. The field
            # names below are IMI v3's documented ones and UMS answers 200
            # either way, so a rename shows up as an asset with no version and
            # a Test button reading "UMS  (build ?)" — nothing failing, just
            # nothing there. The next rename should be one log line to find.
            logger.warning("IMI serverstatus carried no version; keys were: %s",
                           sorted(d) if isinstance(d, dict) else type(d).__name__)
        return out

    def get_firmwares(self) -> Dict[str, dict]:
        """{firmware id: {product, version, type}}.

        Returned verbatim — deciding which firmware IS an IGEL OS is the
        service's job (igel_service.os_from_firmware), not this client's.

        UMS wraps the list in `FwResource`; a UMS that returns a bare list is
        accepted too, because the same endpoint is documented both ways across
        IMI versions and an unwrapped answer must not read as "no firmwares".
        """
        # No trailing slash. UMS 12 routes "/firmwares/" to nothing and answers
        # HTTP 404 "No endpoint GET /umsapi/v3/firmwares/" — the documented URL
        # is /v3/firmwares, and the slash was silently costing the whole
        # firmware table, which is the only place a version string lives.
        data = self._get("/firmwares")
        rows = data.get("FwResource", []) if isinstance(data, dict) else (data or [])
        out: Dict[str, dict] = {}
        for row in rows:
            if not isinstance(row, dict) or not row.get("id"):
                continue
            out[str(row["id"])] = {
                "product": (row.get("product") or "").strip(),
                "version": (row.get("version") or "").strip(),
                "type": (row.get("firmwareType") or "").strip(),
            }
        return out

    def get_devices(self) -> List[dict]:
        """Every endpoint device, one call, with details.

        Recycle-bin entries (`movedToBin`) are dropped here rather than by the
        caller: UMS keeps a deleted device in the listing indefinitely, and a
        device somebody deleted is exactly the one that should stop counting as
        inventory.
        """
        data = self._get("/thinclients?facets=details")
        rows = data if isinstance(data, list) else (data or {}).get("TcResource", [])
        return [self._device_dict(r) for r in rows
                if isinstance(r, dict) and not r.get("movedToBin")]

    def test_connection(self) -> dict:
        """Credential + reachability probe for the settings UI.

        Server status first because it needs no login: that separates "cannot
        reach the IMI service" from "reached it, credentials rejected", which
        are two very different things to put in front of an operator.
        """
        try:
            status = self.get_server_status()
        except IgelError as e:
            return {"ok": False, "step": "connect", "error": str(e)}
        try:
            self.login()
        except IgelError as e:
            return {"ok": False, "step": "login", "error": str(e), "ums": status}
        try:
            devices = self.get_devices()
            firmwares = self.get_firmwares()
        except IgelError as e:
            return {"ok": True, "step": "inventory", "error": str(e),
                    "ums": status, "device_count": None}
        return {"ok": True, "ums": status, "device_count": len(devices),
                "firmware_count": len(firmwares)}

    @staticmethod
    def _device_dict(r: dict) -> dict:
        # Only what the sync writes onto an asset. `?facets=details` returns
        # forty-odd fields per device (monitors, BIOS dates, uptime counters);
        # carrying the ones nothing reads would be forty columns of estate
        # inventory travelling through a vulnerability scanner for no purpose.
        return {
            "unit_id": (r.get("unitID") or "").strip(),
            "firmware_id": str(r.get("firmwareID") or ""),
            "name": (r.get("name") or "").strip(),
            "network_name": (r.get("networkName") or "").strip(),
            "ip_address": (r.get("lastIP") or "").strip() or None,
            "device_type": (r.get("deviceType") or "").strip(),
            "product_id": (r.get("productId") or "").strip(),
            "os_type": (r.get("osType") or "").strip(),
        }


# Documented IMI v3 spelling first, then the ones UMS 12 has been seen to use.
# https://kb.igel.com/en/igel-management-interface/current/get-v3-serverstatus
_STATUS_VERSION_KEYS = ("rmGuiServerVersion", "umsVersion", "serverVersion",
                        "version")
_STATUS_BUILD_KEYS = ("buildNumber", "build", "buildNo", "buildnumber")


def _first(d: object, keys) -> Optional[str]:
    """First key present with a non-empty value, stringified. Empty is missing:
    an UMS that answers "" for its version must not read as a version of ""."""
    if not isinstance(d, dict):
        return None
    for k in keys:
        v = d.get(k)
        if v is not None and str(v).strip():
            return str(v).strip()
    return None


def _imi_error(r: httpx.Response) -> str:
    """The IMI error message, which lives in the body, not the status line."""
    try:
        body = r.json()
        if isinstance(body, dict):
            return str(body.get("message") or body.get("error") or body)[:300]
    except Exception:
        pass
    return (r.text or "")[:300]