Skip to content
Snippets Groups Projects
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