-
Karel van Klink authoredKarel van Klink authored
librenms_client.py 4.78 KiB
"""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, Literal
import requests
from requests import HTTPError, Response
from requests.adapters import HTTPAdapter
from gso.settings import load_oss_params
from gso.utils.shared_enums 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.session = requests.Session()
self.session.mount("https://", HTTPAdapter(max_retries=5))
self.session.headers.update({
"User-Agent": f"geant-service-orchestrator/{metadata.version("geant-service-orchestrator")}",
"Accept": "application/json",
"Content-Type": "application/json",
"X-Auth-Token": token,
})
def _send_request(
self, method: Literal["GET", "POST", "PUT", "DELETE"], endpoint: str, data: dict[str, Any] | None = None
) -> Response:
url = self.base_url + endpoint
logger.debug("LibreNMS - Sending request", extra={"method": method, "endpoint": url, "form_data": data})
result = self.session.request(method, url, json=data, timeout=(10, 75))
logger.debug("LibreNMS - Received response", extra=result.__dict__)
return result
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 = self._send_request("GET", f"/devices/{fqdn}")
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 = self._send_request("POST", "/devices", device_data)
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 = self._send_request("DELETE", f"/devices/{fqdn}")
device.raise_for_status()
return device.json()
def validate_device(self, fqdn: str) -> str | None:
"""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.
"""
error = None
try:
device = self.get_device(fqdn)
received_hostname = device["devices"][0]["hostname"]
if received_hostname != fqdn:
error = (
f"Device hostname in LibreNMS does not match FQDN.\n"
f"Expected '{fqdn}' but got '{received_hostname}'."
)
except HTTPError as e:
if e.response.status_code == HTTPStatus.NOT_FOUND:
error = "Device does not exist in LibreNMS."
else:
raise
return error