security-alert-center/backend/app/services/host_inventory.py

"""Host hardware/software inventory from agent.inventory ingest."""

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from typing import Any

from app.models import Host

INVENTORY_EVENT_TYPE = "agent.inventory"


def _as_dict(value: Any) -> dict[str, Any]:
    return value if isinstance(value, dict) else {}


def _as_list(value: Any) -> list[Any]:
    return value if isinstance(value, list) else []


def _round_gb(value: Any) -> float | None:
    try:
        return round(float(value), 2)
    except (TypeError, ValueError):
        return None


def extract_inventory_payload(details: dict | None) -> dict[str, Any] | None:
    if not isinstance(details, dict):
        return None
    inv = details.get("inventory")
    return inv if isinstance(inv, dict) else None


def hardware_snapshot(inventory: dict[str, Any]) -> dict[str, Any]:
    processors: list[tuple[str, int | None, int | None]] = []
    for item in _as_list(inventory.get("processor")):
        row = _as_dict(item)
        name = str(row.get("name") or "").strip()
        if not name:
            continue
        cores = row.get("cores")
        logical = row.get("logical_processors")
        try:
            cores_i = int(cores) if cores is not None else None
        except (TypeError, ValueError):
            cores_i = None
        try:
            logical_i = int(logical) if logical is not None else None
        except (TypeError, ValueError):
            logical_i = None
        processors.append((name, cores_i, logical_i))
    processors.sort()

    motherboard = _as_dict(inventory.get("motherboard"))
    mobo_product = str(motherboard.get("product") or "").strip()

    memory_gb = _round_gb(inventory.get("memory_gb"))

    disks: list[tuple[str, float | None, str]] = []
    for item in _as_list(inventory.get("disks")):
        row = _as_dict(item)
        friendly = str(row.get("friendly_name") or "").strip()
        media = str(row.get("media_type") or "").strip()
        size_gb = _round_gb(row.get("size_gb"))
        if friendly:
            disks.append((friendly, size_gb, media))
    disks.sort()

    video: list[str] = sorted(
        {
            str(_as_dict(item).get("name") or "").strip()
            for item in _as_list(inventory.get("video"))
            if str(_as_dict(item).get("name") or "").strip()
        }
    )

    return {
        "processor": processors,
        "motherboard_product": mobo_product,
        "memory_gb": memory_gb,
        "disks": disks,
        "video": video,
    }


def _format_change(field: str, old: Any, new: Any) -> dict[str, Any]:
    return {"field": field, "old": old, "new": new}


def diff_hardware(old: dict[str, Any] | None, new: dict[str, Any]) -> list[dict[str, Any]]:
    if not old:
        return []
    prev = hardware_snapshot(old)
    curr = hardware_snapshot(new)
    changes: list[dict[str, Any]] = []

    if prev["processor"] != curr["processor"]:
        changes.append(_format_change("processor", prev["processor"], curr["processor"]))
    if prev["motherboard_product"] != curr["motherboard_product"]:
        changes.append(
            _format_change("motherboard", prev["motherboard_product"], curr["motherboard_product"])
        )
    if prev["memory_gb"] != curr["memory_gb"]:
        changes.append(_format_change("memory_gb", prev["memory_gb"], curr["memory_gb"]))
    if prev["disks"] != curr["disks"]:
        changes.append(_format_change("disks", prev["disks"], curr["disks"]))
    if prev["video"] != curr["video"]:
        changes.append(_format_change("video", prev["video"], curr["video"]))
    return changes


def inventory_fingerprint(inventory: dict[str, Any]) -> str:
    snap = hardware_snapshot(inventory)
    raw = json.dumps(snap, sort_keys=True, ensure_ascii=False, default=str)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]


def apply_os_version_from_inventory(host: Host, inventory: dict[str, Any]) -> None:
    windows = _as_dict(inventory.get("windows"))
    product = str(windows.get("product_name") or "").strip()
    version = str(windows.get("version") or "").strip()
    if product and version:
        host.os_version = f"{product} ({version})"
    elif product:
        host.os_version = product
    elif version:
        host.os_version = version


def process_inventory_ingest(
    host: Host,
    details: dict | None,
) -> tuple[str, list[dict[str, Any]], dict[str, Any] | None]:
    """
    Update host.inventory from agent.inventory details.

    Returns (severity, hardware_changes, merged_details_patch).
    severity is warning when hardware changed, else info.
    """
    inventory = extract_inventory_payload(details)
    if inventory is None:
        return "info", [], None

    previous = host.inventory if isinstance(host.inventory, dict) else None
    changes = diff_hardware(previous, inventory)
    now = datetime.now(timezone.utc)
    host.inventory = inventory
    host.inventory_updated_at = now
    apply_os_version_from_inventory(host, inventory)

    patch: dict[str, Any] = {
        "inventory_fingerprint": inventory_fingerprint(inventory),
        "first_inventory": previous is None,
    }
    if changes:
        patch["hardware_changes"] = changes

    severity = "warning" if changes else "info"
    return severity, changes, patch