"""The LibreNMS module interacts with the inventory management system of :term:`GAP`.""" import logging from http import HTTPStatus from importlib import metadata from typing import Any import requests from requests import HTTPError from gso.settings import load_oss_params from gso.utils.helpers import SNMPVersion logger = logging.getLogger(__name__) class LibreNMSClient: """The client for LibreNMS that interacts with the inventory management system.""" def __init__(self) -> None: """Initialise a new LibreNMS client with an authentication token.""" config = load_oss_params().MONITORING token = config.LIBRENMS.token self.base_url = config.LIBRENMS.base_url self.snmp_config = config.SNMP self.headers = { "User-Agent": f"geant-service-orchestrator/{metadata.version('geant-service-orchestrator')}", "Accept": "application/json", "Content-Type": "application/json", "X-Auth-Token": token, } def get_device(self, fqdn: str) -> dict[str, Any]: """Get an existing device from LibreNMS. :param str fqdn: The :term:`FQDN` of a device that is retrieved. :return dict[str, Any]: A :term:`JSON` formatted list of devices that match the queried :term:`FQDN`. :raises HTTPError: Raises an HTTP error 404 when the device is not found """ response = requests.get(f"{self.base_url}/devices/{fqdn}", headers=self.headers, timeout=(0.5, 75)) response.raise_for_status() return response.json() def device_exists(self, fqdn: str) -> bool: """Check whether a device exists in LibreNMS. :param str fqdn: The hostname that should be checked for. :return bool: Whether the device exists or not. """ try: device = self.get_device(fqdn) except HTTPError as e: if e.response.status_code == HTTPStatus.NOT_FOUND: return False raise return device["status"] == "ok" def add_device(self, fqdn: str, snmp_version: SNMPVersion) -> dict[str, Any]: """Add a new device to LibreNMS. :param str fqdn: The hostname of the newly added device. :param SNMPVersion snmp_version: The SNMP version of the new device, which decides the authentication parameters that LibreNMS should use to poll the device. """ device_data = { "display": fqdn, "hostname": fqdn, "sysName": fqdn, "snmpver": snmp_version.value, } device_data.update(getattr(self.snmp_config, snmp_version)) device = requests.post(f"{self.base_url}/devices", headers=self.headers, json=device_data, timeout=(0.5, 75)) device.raise_for_status() return device.json() def remove_device(self, fqdn: str) -> dict[str, Any]: """Remove a device from LibreNMS. :param str fqdn: The :term:`FQDN` of the hostname that should get deleted. :return dict[str, Any]: A JSON representation of the device that got removed. :raises HTTPError: Raises an exception if the request did not succeed. """ device = requests.delete(f"{self.base_url}/devices/{fqdn}", headers=self.headers, timeout=(0.5, 75)) device.raise_for_status() return device.json() def validate_device(self, fqdn: str) -> list[str]: """Validate a device in LibreNMS by fetching the record match the queried :term:`FQDN` against its hostname. :param str fqdn: The :term:`FQDN` of the host that is validated. :return list[str]: A list of errors, if empty the device is successfully validated. """ errors = [] try: device = self.get_device(fqdn) if device["devices"][0]["hostname"] != fqdn: errors += ["Device hostname in LibreNMS does not match FQDN."] except HTTPError as e: if e.response.status_code == HTTPStatus.NOT_FOUND: errors += ["Device does not exist in LibreNMS."] else: raise return errors