librenms_client.py

"""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