1#@!#!123s

��JFIF�� 1#@!#!123s
: / opt / saltstack / salt / extras-3.10 / mtrlib /
Filename : nscaweb.py
back
"""Interact with an nscaweb server"""
import time
import sys
import platform
import random
from typing import Union
import requests
import dns.resolver
from .nagios import OK


def send_command(cmd: str, nscauser: str, nscapass: str, **kwargs):
    """Send an external Nagios command via nscaweb

    Args:
        cmd: A valid Nagios external command. See
          http://old.nagios.org/developerinfo/externalcommands/commandlist.php
          for further details
        nscauser: user name to use with nscaweb
        nscapass: password for nscaweb
        timestamp (int, optional): timestamp to supply with the external
          command. time.time() is used as a default
        nscahost (str, optional): host where nscaweb is running
    """
    timestamp = kwargs.get('timestamp', int(time.time()))
    nscahost = kwargs.get('nscaserver', None)

    if nscahost is None:
        if platform.node().startswith('lax-'):
            nscahost = 'lax-checkers.imhadmin.net'
        else:
            nscahost = 'ash-checkers.imhadmin.net'

    ips = _get_ips(nscahost)

    for nscahost in ips:
        url = 'https://{!s}:5668/queue'.format(nscahost)
        try:
            return requests.post(
                url,
                verify=False,
                timeout=10.0,
                data={
                    'username': nscauser,
                    'password': nscapass,
                    'input': '[{!s}] {!s}'.format(timestamp, cmd),
                },
            )
        except (requests.RequestException, OSError) as exc:
            print(exc, file=sys.stderr)
            if nscahost == ips[-1]:
                return False
    return None


def _get_ips(nscahost: str):
    """Function to resolve IP addresses from a hostname"""
    res = dns.resolver.Resolver()
    ips = [str(x) for x in res.query(nscahost.split(':')[0], "A")]
    random.shuffle(ips)
    return ips


def service_check(service: str, status: int = OK, message: str = '', **kwargs):
    """Send a service check result to Nagios via nscaweb

    Args:
        service: The service description to be updated
        host (str, optional): The host to be updated. Defaults to the local
            server's short hostname
        status: The status of the service. Defaults to ``mtrlib.nagios.OK``
        message: Check result string

    Additional kwargs understood by ``send_command`` may be passed in
    as kwargs to ``service_check``

    For more information on this external command see
    http://old.nagios.org/developerinfo/externalcommands/commandinfo.php?command_id=114
    """
    host = kwargs.get('host', platform.node().split('.')[0])

    command = 'PROCESS_SERVICE_CHECK_RESULT;{!s};{!s};{!s};{!s}'.format(
        host, service, status, message
    )

    send_command(cmd=command, **kwargs)


def force_service_recheck(service: Union[str, None], **kwargs):
    """Force recheck of a service via nscaweb, or all services on a host if
    the `service` kwarg is not supplied

    Args:
        service: The service description to be updated. If None,
            all services are rechecked
        host (str, optional): The host to be updated. Defaults to
            the local server's short hostname

    Additional kwargs understood by ``send_command`` may be passed in
    as kwargs to ``force_service_recheck``

    For more information on these external commands, see
    http://old.nagios.org/developerinfo/externalcommands/commandinfo.php?command_id=30
    http://old.nagios.org/developerinfo/externalcommands/commandinfo.php?command_id=129
    """
    host = kwargs.get('host', platform.node().split('.')[0])
    timestamp = int(time.time())

    if service is None:
        command = 'SCHEDULE_HOST_SVC_CHECKS;{!s};{!s}'.format(host, timestamp)
    else:
        command = 'SCHEDULE_FORCED_SVC_CHECK;{!s};{!s};{!s}'.format(
            host, service, timestamp
        )

    send_command(cmd=command, timestamp=timestamp, **kwargs)