mirror of
https://github.com/bunny-lab-io/Borealis.git
synced 2025-10-26 15:41:58 -06:00
576 lines
23 KiB
Python
576 lines
23 KiB
Python
"""Mirrors the legacy device inventory HTTP behaviour."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import sqlite3
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from collections.abc import Mapping
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from Data.Engine.repositories.sqlite.device_inventory_repository import (
|
|
SQLiteDeviceInventoryRepository,
|
|
)
|
|
from Data.Engine.domain.device_auth import DeviceAuthContext, normalize_guid
|
|
from Data.Engine.domain.devices import clean_device_str, coerce_int, ts_to_human
|
|
|
|
__all__ = [
|
|
"DeviceInventoryService",
|
|
"RemoteDeviceError",
|
|
"DeviceHeartbeatError",
|
|
"DeviceDetailsError",
|
|
"DeviceDescriptionError",
|
|
]
|
|
|
|
|
|
class RemoteDeviceError(Exception):
|
|
def __init__(self, code: str, message: Optional[str] = None) -> None:
|
|
super().__init__(message or code)
|
|
self.code = code
|
|
|
|
|
|
class DeviceHeartbeatError(Exception):
|
|
def __init__(self, code: str, message: Optional[str] = None) -> None:
|
|
super().__init__(message or code)
|
|
self.code = code
|
|
|
|
|
|
class DeviceDetailsError(Exception):
|
|
def __init__(self, code: str, message: Optional[str] = None) -> None:
|
|
super().__init__(message or code)
|
|
self.code = code
|
|
|
|
|
|
class DeviceDescriptionError(Exception):
|
|
def __init__(self, code: str, message: Optional[str] = None) -> None:
|
|
super().__init__(message or code)
|
|
self.code = code
|
|
|
|
|
|
class DeviceInventoryService:
|
|
def __init__(
|
|
self,
|
|
repository: SQLiteDeviceInventoryRepository,
|
|
*,
|
|
logger: Optional[logging.Logger] = None,
|
|
) -> None:
|
|
self._repo = repository
|
|
self._log = logger or logging.getLogger("borealis.engine.services.devices")
|
|
|
|
def list_devices(self) -> List[Dict[str, object]]:
|
|
return self._repo.fetch_devices()
|
|
|
|
def list_agent_devices(self) -> List[Dict[str, object]]:
|
|
return self._repo.fetch_devices(only_agents=True)
|
|
|
|
def list_remote_devices(self, connection_type: str) -> List[Dict[str, object]]:
|
|
return self._repo.fetch_devices(connection_type=connection_type)
|
|
|
|
def get_device_by_guid(self, guid: str) -> Optional[Dict[str, object]]:
|
|
snapshot = self._repo.load_snapshot(guid=guid)
|
|
if not snapshot:
|
|
return None
|
|
devices = self._repo.fetch_devices(hostname=snapshot.get("hostname"))
|
|
return devices[0] if devices else None
|
|
|
|
def get_device_details(self, hostname: str) -> Dict[str, object]:
|
|
normalized_host = clean_device_str(hostname)
|
|
if not normalized_host:
|
|
return {}
|
|
|
|
snapshot = self._repo.load_snapshot(hostname=normalized_host)
|
|
if not snapshot:
|
|
return {}
|
|
|
|
summary = dict(snapshot.get("summary") or {})
|
|
|
|
payload: Dict[str, Any] = {
|
|
"details": snapshot.get("details", {}),
|
|
"summary": summary,
|
|
"description": snapshot.get("description")
|
|
or summary.get("description")
|
|
or "",
|
|
"created_at": snapshot.get("created_at") or 0,
|
|
"agent_hash": snapshot.get("agent_hash")
|
|
or summary.get("agent_hash")
|
|
or "",
|
|
"agent_guid": snapshot.get("agent_guid")
|
|
or summary.get("agent_guid")
|
|
or "",
|
|
"memory": snapshot.get("memory", []),
|
|
"network": snapshot.get("network", []),
|
|
"software": snapshot.get("software", []),
|
|
"storage": snapshot.get("storage", []),
|
|
"cpu": snapshot.get("cpu", {}),
|
|
"device_type": snapshot.get("device_type")
|
|
or summary.get("device_type")
|
|
or "",
|
|
"domain": snapshot.get("domain")
|
|
or summary.get("domain")
|
|
or "",
|
|
"external_ip": snapshot.get("external_ip")
|
|
or summary.get("external_ip")
|
|
or "",
|
|
"internal_ip": snapshot.get("internal_ip")
|
|
or summary.get("internal_ip")
|
|
or "",
|
|
"last_reboot": snapshot.get("last_reboot")
|
|
or summary.get("last_reboot")
|
|
or "",
|
|
"last_seen": snapshot.get("last_seen")
|
|
or summary.get("last_seen")
|
|
or 0,
|
|
"last_user": snapshot.get("last_user")
|
|
or summary.get("last_user")
|
|
or "",
|
|
"operating_system": snapshot.get("operating_system")
|
|
or summary.get("operating_system")
|
|
or summary.get("agent_operating_system")
|
|
or "",
|
|
"uptime": snapshot.get("uptime")
|
|
or summary.get("uptime")
|
|
or 0,
|
|
"agent_id": snapshot.get("agent_id")
|
|
or summary.get("agent_id")
|
|
or "",
|
|
}
|
|
|
|
return payload
|
|
|
|
def collect_agent_hash_records(self) -> List[Dict[str, object]]:
|
|
records: List[Dict[str, object]] = []
|
|
key_to_index: Dict[str, int] = {}
|
|
|
|
for device in self._repo.fetch_devices():
|
|
summary = device.get("summary", {}) if isinstance(device, dict) else {}
|
|
agent_id = (summary.get("agent_id") or "").strip()
|
|
agent_guid = (summary.get("agent_guid") or "").strip()
|
|
hostname = (summary.get("hostname") or device.get("hostname") or "").strip()
|
|
agent_hash = (summary.get("agent_hash") or device.get("agent_hash") or "").strip()
|
|
|
|
keys: List[str] = []
|
|
if agent_id:
|
|
keys.append(f"id:{agent_id.lower()}")
|
|
if agent_guid:
|
|
keys.append(f"guid:{agent_guid.lower()}")
|
|
if hostname:
|
|
keys.append(f"host:{hostname.lower()}")
|
|
|
|
payload = {
|
|
"agent_id": agent_id or None,
|
|
"agent_guid": agent_guid or None,
|
|
"hostname": hostname or None,
|
|
"agent_hash": agent_hash or None,
|
|
"source": "database",
|
|
}
|
|
|
|
if not keys:
|
|
records.append(payload)
|
|
continue
|
|
|
|
existing_index = None
|
|
for key in keys:
|
|
if key in key_to_index:
|
|
existing_index = key_to_index[key]
|
|
break
|
|
|
|
if existing_index is None:
|
|
existing_index = len(records)
|
|
records.append(payload)
|
|
for key in keys:
|
|
key_to_index[key] = existing_index
|
|
continue
|
|
|
|
merged = records[existing_index]
|
|
for key in ("agent_id", "agent_guid", "hostname", "agent_hash"):
|
|
if not merged.get(key) and payload.get(key):
|
|
merged[key] = payload[key]
|
|
|
|
return records
|
|
|
|
def upsert_remote_device(
|
|
self,
|
|
connection_type: str,
|
|
hostname: str,
|
|
address: Optional[str],
|
|
description: Optional[str],
|
|
os_hint: Optional[str],
|
|
*,
|
|
ensure_existing_type: Optional[str],
|
|
) -> Dict[str, object]:
|
|
normalized_type = (connection_type or "").strip().lower()
|
|
if not normalized_type:
|
|
raise RemoteDeviceError("invalid_type", "connection type required")
|
|
normalized_host = (hostname or "").strip()
|
|
if not normalized_host:
|
|
raise RemoteDeviceError("invalid_hostname", "hostname is required")
|
|
|
|
existing = self._repo.load_snapshot(hostname=normalized_host)
|
|
existing_type = (existing or {}).get("summary", {}).get("connection_type") or ""
|
|
existing_type = existing_type.strip().lower()
|
|
|
|
if ensure_existing_type and existing_type != ensure_existing_type.lower():
|
|
raise RemoteDeviceError("not_found", "device not found")
|
|
if ensure_existing_type is None and existing_type and existing_type != normalized_type:
|
|
raise RemoteDeviceError("conflict", "device already exists with different connection type")
|
|
|
|
created_ts = None
|
|
if existing:
|
|
created_ts = existing.get("summary", {}).get("created_at")
|
|
|
|
endpoint = (address or "").strip() or (existing or {}).get("summary", {}).get("connection_endpoint") or ""
|
|
if not endpoint:
|
|
raise RemoteDeviceError("address_required", "address is required")
|
|
|
|
description_val = description if description is not None else (existing or {}).get("summary", {}).get("description")
|
|
os_value = os_hint or (existing or {}).get("summary", {}).get("operating_system")
|
|
os_value = (os_value or "").strip()
|
|
|
|
device_type_label = "SSH Remote" if normalized_type == "ssh" else "WinRM Remote"
|
|
|
|
summary_payload = {
|
|
"connection_type": normalized_type,
|
|
"connection_endpoint": endpoint,
|
|
"internal_ip": endpoint,
|
|
"external_ip": endpoint,
|
|
"device_type": device_type_label,
|
|
"operating_system": os_value or "",
|
|
"last_seen": 0,
|
|
"description": (description_val or ""),
|
|
}
|
|
|
|
try:
|
|
self._repo.upsert_device(
|
|
normalized_host,
|
|
description_val,
|
|
{"summary": summary_payload},
|
|
created_ts,
|
|
)
|
|
except sqlite3.DatabaseError as exc: # type: ignore[name-defined]
|
|
raise RemoteDeviceError("storage_error", str(exc)) from exc
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
raise RemoteDeviceError("storage_error", str(exc)) from exc
|
|
|
|
devices = self._repo.fetch_devices(hostname=normalized_host)
|
|
if not devices:
|
|
raise RemoteDeviceError("reload_failed", "failed to load device after upsert")
|
|
return devices[0]
|
|
|
|
def delete_remote_device(self, connection_type: str, hostname: str) -> None:
|
|
normalized_host = (hostname or "").strip()
|
|
if not normalized_host:
|
|
raise RemoteDeviceError("invalid_hostname", "invalid hostname")
|
|
existing = self._repo.load_snapshot(hostname=normalized_host)
|
|
if not existing:
|
|
raise RemoteDeviceError("not_found", "device not found")
|
|
existing_type = (existing.get("summary", {}) or {}).get("connection_type") or ""
|
|
if (existing_type or "").strip().lower() != (connection_type or "").strip().lower():
|
|
raise RemoteDeviceError("not_found", "device not found")
|
|
self._repo.delete_device_by_hostname(normalized_host)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Agent heartbeats
|
|
# ------------------------------------------------------------------
|
|
def record_heartbeat(
|
|
self,
|
|
*,
|
|
context: DeviceAuthContext,
|
|
payload: Mapping[str, Any],
|
|
) -> None:
|
|
guid = context.identity.guid.value
|
|
snapshot = self._repo.load_snapshot(guid=guid)
|
|
if not snapshot:
|
|
raise DeviceHeartbeatError("device_not_registered", "device not registered")
|
|
|
|
summary = dict(snapshot.get("summary") or {})
|
|
details = dict(snapshot.get("details") or {})
|
|
|
|
now_ts = int(time.time())
|
|
summary["last_seen"] = now_ts
|
|
summary["agent_guid"] = guid
|
|
|
|
existing_hostname = clean_device_str(summary.get("hostname")) or clean_device_str(
|
|
snapshot.get("hostname")
|
|
)
|
|
incoming_hostname = clean_device_str(payload.get("hostname"))
|
|
raw_metrics = payload.get("metrics")
|
|
metrics = raw_metrics if isinstance(raw_metrics, Mapping) else {}
|
|
metrics_hostname = clean_device_str(metrics.get("hostname")) if metrics else None
|
|
hostname = incoming_hostname or metrics_hostname or existing_hostname
|
|
if not hostname:
|
|
hostname = f"RECOVERED-{guid[:12]}"
|
|
summary["hostname"] = hostname
|
|
|
|
if metrics:
|
|
last_user = metrics.get("last_user") or metrics.get("username") or metrics.get("user")
|
|
if last_user:
|
|
cleaned_user = clean_device_str(last_user)
|
|
if cleaned_user:
|
|
summary["last_user"] = cleaned_user
|
|
operating_system = metrics.get("operating_system")
|
|
if operating_system:
|
|
cleaned_os = clean_device_str(operating_system)
|
|
if cleaned_os:
|
|
summary["operating_system"] = cleaned_os
|
|
uptime = metrics.get("uptime")
|
|
if uptime is not None:
|
|
coerced = coerce_int(uptime)
|
|
if coerced is not None:
|
|
summary["uptime"] = coerced
|
|
agent_id = metrics.get("agent_id")
|
|
if agent_id:
|
|
cleaned_agent = clean_device_str(agent_id)
|
|
if cleaned_agent:
|
|
summary["agent_id"] = cleaned_agent
|
|
|
|
for field in ("external_ip", "internal_ip", "device_type"):
|
|
value = payload.get(field)
|
|
cleaned = clean_device_str(value)
|
|
if cleaned:
|
|
summary[field] = cleaned
|
|
|
|
summary.setdefault("description", summary.get("description") or "")
|
|
created_at = coerce_int(summary.get("created_at"))
|
|
if created_at is None:
|
|
created_at = coerce_int(snapshot.get("created_at"))
|
|
if created_at is None:
|
|
created_at = now_ts
|
|
summary["created_at"] = created_at
|
|
|
|
raw_inventory = payload.get("inventory")
|
|
inventory = raw_inventory if isinstance(raw_inventory, Mapping) else {}
|
|
memory = inventory.get("memory") if isinstance(inventory.get("memory"), list) else details.get("memory")
|
|
network = inventory.get("network") if isinstance(inventory.get("network"), list) else details.get("network")
|
|
software = (
|
|
inventory.get("software") if isinstance(inventory.get("software"), list) else details.get("software")
|
|
)
|
|
storage = inventory.get("storage") if isinstance(inventory.get("storage"), list) else details.get("storage")
|
|
cpu = inventory.get("cpu") if isinstance(inventory.get("cpu"), Mapping) else details.get("cpu")
|
|
|
|
merged_details: Dict[str, Any] = {
|
|
"summary": summary,
|
|
"memory": memory,
|
|
"network": network,
|
|
"software": software,
|
|
"storage": storage,
|
|
"cpu": cpu,
|
|
}
|
|
|
|
try:
|
|
self._repo.upsert_device(
|
|
summary["hostname"],
|
|
summary.get("description"),
|
|
merged_details,
|
|
summary.get("created_at"),
|
|
agent_hash=clean_device_str(summary.get("agent_hash")),
|
|
guid=guid,
|
|
)
|
|
except sqlite3.IntegrityError as exc:
|
|
self._log.warning(
|
|
"device-heartbeat-conflict guid=%s hostname=%s error=%s",
|
|
guid,
|
|
summary["hostname"],
|
|
exc,
|
|
)
|
|
raise DeviceHeartbeatError("storage_conflict", str(exc)) from exc
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
self._log.exception(
|
|
"device-heartbeat-failure guid=%s hostname=%s",
|
|
guid,
|
|
summary["hostname"],
|
|
exc_info=exc,
|
|
)
|
|
raise DeviceHeartbeatError("storage_error", "failed to persist heartbeat") from exc
|
|
|
|
# ------------------------------------------------------------------
|
|
# Agent details
|
|
# ------------------------------------------------------------------
|
|
@staticmethod
|
|
def _is_empty(value: Any) -> bool:
|
|
return value in (None, "", [], {})
|
|
|
|
@classmethod
|
|
def _deep_merge_preserve(cls, prev: Dict[str, Any], incoming: Dict[str, Any]) -> Dict[str, Any]:
|
|
merged: Dict[str, Any] = dict(prev or {})
|
|
for key, value in (incoming or {}).items():
|
|
if isinstance(value, Mapping):
|
|
existing = merged.get(key)
|
|
if not isinstance(existing, Mapping):
|
|
existing = {}
|
|
merged[key] = cls._deep_merge_preserve(dict(existing), dict(value))
|
|
elif isinstance(value, list):
|
|
if value:
|
|
merged[key] = value
|
|
else:
|
|
if cls._is_empty(value):
|
|
continue
|
|
merged[key] = value
|
|
return merged
|
|
|
|
def save_agent_details(
|
|
self,
|
|
*,
|
|
context: DeviceAuthContext,
|
|
payload: Mapping[str, Any],
|
|
) -> None:
|
|
hostname = clean_device_str(payload.get("hostname"))
|
|
details_raw = payload.get("details")
|
|
agent_id = clean_device_str(payload.get("agent_id"))
|
|
agent_hash = clean_device_str(payload.get("agent_hash"))
|
|
|
|
if not isinstance(details_raw, Mapping):
|
|
raise DeviceDetailsError("invalid_payload", "details object required")
|
|
|
|
details_dict: Dict[str, Any]
|
|
try:
|
|
details_dict = json.loads(json.dumps(details_raw))
|
|
except Exception:
|
|
details_dict = dict(details_raw)
|
|
|
|
incoming_summary = dict(details_dict.get("summary") or {})
|
|
if not hostname:
|
|
hostname = clean_device_str(incoming_summary.get("hostname"))
|
|
if not hostname:
|
|
raise DeviceDetailsError("invalid_payload", "hostname required")
|
|
|
|
snapshot = self._repo.load_snapshot(hostname=hostname)
|
|
if not snapshot:
|
|
snapshot = {}
|
|
|
|
previous_details = snapshot.get("details")
|
|
if isinstance(previous_details, Mapping):
|
|
try:
|
|
prev_details = json.loads(json.dumps(previous_details))
|
|
except Exception:
|
|
prev_details = dict(previous_details)
|
|
else:
|
|
prev_details = {}
|
|
|
|
prev_summary = dict(prev_details.get("summary") or {})
|
|
|
|
existing_guid = clean_device_str(snapshot.get("guid") or snapshot.get("summary", {}).get("agent_guid"))
|
|
normalized_existing_guid = normalize_guid(existing_guid)
|
|
auth_guid = context.identity.guid.value
|
|
|
|
if normalized_existing_guid and normalized_existing_guid != auth_guid:
|
|
raise DeviceDetailsError("guid_mismatch", "device guid mismatch")
|
|
|
|
fingerprint = context.identity.fingerprint.value.lower()
|
|
stored_fp = clean_device_str(snapshot.get("summary", {}).get("ssl_key_fingerprint"))
|
|
if stored_fp and stored_fp.lower() != fingerprint:
|
|
raise DeviceDetailsError("fingerprint_mismatch", "device fingerprint mismatch")
|
|
|
|
incoming_summary.setdefault("hostname", hostname)
|
|
if agent_id and not incoming_summary.get("agent_id"):
|
|
incoming_summary["agent_id"] = agent_id
|
|
if agent_hash:
|
|
incoming_summary["agent_hash"] = agent_hash
|
|
incoming_summary["agent_guid"] = auth_guid
|
|
if fingerprint:
|
|
incoming_summary["ssl_key_fingerprint"] = fingerprint
|
|
if not incoming_summary.get("last_seen") and prev_summary.get("last_seen"):
|
|
incoming_summary["last_seen"] = prev_summary.get("last_seen")
|
|
|
|
details_dict["summary"] = incoming_summary
|
|
merged_details = self._deep_merge_preserve(prev_details, details_dict)
|
|
merged_summary = merged_details.setdefault("summary", {})
|
|
|
|
if not merged_summary.get("last_user") and prev_summary.get("last_user"):
|
|
merged_summary["last_user"] = prev_summary.get("last_user")
|
|
|
|
created_at = coerce_int(merged_summary.get("created_at"))
|
|
if created_at is None:
|
|
created_at = coerce_int(snapshot.get("created_at"))
|
|
if created_at is None:
|
|
created_at = int(time.time())
|
|
merged_summary["created_at"] = created_at
|
|
if not merged_summary.get("created"):
|
|
merged_summary["created"] = ts_to_human(created_at)
|
|
|
|
if fingerprint:
|
|
merged_summary["ssl_key_fingerprint"] = fingerprint
|
|
if not merged_summary.get("key_added_at"):
|
|
merged_summary["key_added_at"] = datetime.now(timezone.utc).isoformat()
|
|
if merged_summary.get("token_version") is None:
|
|
merged_summary["token_version"] = 1
|
|
if not merged_summary.get("status") and snapshot.get("summary", {}).get("status"):
|
|
merged_summary["status"] = snapshot.get("summary", {}).get("status")
|
|
uptime_val = merged_summary.get("uptime")
|
|
if merged_summary.get("uptime_sec") is None and uptime_val is not None:
|
|
coerced = coerce_int(uptime_val)
|
|
if coerced is not None:
|
|
merged_summary["uptime_sec"] = coerced
|
|
merged_summary.setdefault("uptime_seconds", coerced)
|
|
if merged_summary.get("uptime_seconds") is None and merged_summary.get("uptime_sec") is not None:
|
|
merged_summary["uptime_seconds"] = merged_summary.get("uptime_sec")
|
|
|
|
description = clean_device_str(merged_summary.get("description"))
|
|
existing_description = snapshot.get("description") if snapshot else ""
|
|
description_to_store = description if description is not None else (existing_description or "")
|
|
|
|
existing_hash = clean_device_str(snapshot.get("agent_hash") or snapshot.get("summary", {}).get("agent_hash"))
|
|
effective_hash = agent_hash or existing_hash
|
|
|
|
try:
|
|
self._repo.upsert_device(
|
|
hostname,
|
|
description_to_store,
|
|
merged_details,
|
|
created_at,
|
|
agent_hash=effective_hash,
|
|
guid=auth_guid,
|
|
)
|
|
except sqlite3.DatabaseError as exc:
|
|
raise DeviceDetailsError("storage_error", str(exc)) from exc
|
|
|
|
added_at = merged_summary.get("key_added_at") or datetime.now(timezone.utc).isoformat()
|
|
self._repo.record_device_fingerprint(auth_guid, fingerprint, added_at)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Description management
|
|
# ------------------------------------------------------------------
|
|
def update_device_description(self, hostname: str, description: Optional[str]) -> None:
|
|
normalized_host = clean_device_str(hostname)
|
|
if not normalized_host:
|
|
raise DeviceDescriptionError("invalid_hostname", "invalid hostname")
|
|
|
|
snapshot = self._repo.load_snapshot(hostname=normalized_host)
|
|
if not snapshot:
|
|
raise DeviceDescriptionError("not_found", "device not found")
|
|
|
|
details = snapshot.get("details")
|
|
if isinstance(details, Mapping):
|
|
try:
|
|
existing = json.loads(json.dumps(details))
|
|
except Exception:
|
|
existing = dict(details)
|
|
else:
|
|
existing = {}
|
|
|
|
summary = dict(existing.get("summary") or {})
|
|
summary["description"] = description or ""
|
|
existing["summary"] = summary
|
|
|
|
created_at = coerce_int(summary.get("created_at"))
|
|
if created_at is None:
|
|
created_at = coerce_int(snapshot.get("created_at"))
|
|
if created_at is None:
|
|
created_at = int(time.time())
|
|
|
|
agent_hash = clean_device_str(summary.get("agent_hash") or snapshot.get("agent_hash"))
|
|
guid = clean_device_str(summary.get("agent_guid") or snapshot.get("guid"))
|
|
|
|
try:
|
|
self._repo.upsert_device(
|
|
normalized_host,
|
|
description or (snapshot.get("description") or ""),
|
|
existing,
|
|
created_at,
|
|
agent_hash=agent_hash,
|
|
guid=guid,
|
|
)
|
|
except sqlite3.DatabaseError as exc:
|
|
raise DeviceDescriptionError("storage_error", str(exc)) from exc
|