check_ntp_time

#!/usr/bin/env python3
#-------------------------------------------------------------------------------

import socket
import os
import time
import sys
import ntplib
import datetime
import argparse

#-------------------------------------------------------------------------------

_candebug = False
def candebug():
    global _candebug
    return _candebug
def setcandebug(value):
    global _candebug
    _candebug = value

def infomsg(msg):
    if candebug() == True:
        print(msg, flush=True)

def mapstatustoexitcode(status):
    if status=="OK":
        exitcode = 0
    elif status=="WARNING":
        exitcode = 1
    elif status=="CRITICAL":
        exitcode = 2
    elif status=="UNKNOWN":
        exitcode = 3
    else:
        exitcode = 4
    return exitcode

def exitnagios(status,message):
    exitcode = mapstatustoexitcode(status)
    print(status+": "+message, flush=True)
    sys.exit(exitcode)

#-------------------------------------------------------------------------------

original_getaddrinfo = socket.getaddrinfo

def forced_ipv6_gai_family(*args, **kwargs):
    global original_getaddrinfo
    responses = original_getaddrinfo(*args, **kwargs)
    return [response
            for response in responses
            if response[0] == socket.AF_INET6]

def forced_ipv4_gai_family(*args, **kwargs):
    global original_getaddrinfo
    responses = original_getaddrinfo(*args, **kwargs)
    return [response
            for response in responses
            if response[0] == socket.AF_INET]

#-------------------------------------------------------------------------------

def dontptimecheck(protocol,hostname,warning,critical,timeoutvalue,timeoutstatus):
    client = ntplib.NTPClient()
    counter = 0
    retries = 3
    while True:
        infomsg("executing retry "+str(counter))
        counter = counter+1
        try:
            response = client.request(hostname, version=4)
            offset = response.offset
            infomsg(f"NTP offset: {offset:.3f} seconds")

            if critical is not None and abs(offset) >= float(critical):
                status = "CRITICAL"
            elif warning is not None and abs(offset) >= float(warning):
                status = "WARNING"
            else:
                status = "OK"

            now = datetime.datetime.now(datetime.UTC)
            exitnagios(status, f"NTP offset {offset:.3f}s at {now.isoformat()} UTC")

        except Exception as e:
            infomsg(f"NTP request failed: {e}")
            if counter > retries:
                exitnagios(timeoutstatus, str(e))
            else:
                time.sleep(5)

#-------------------------------------------------------------------------------

def main(forcedargs=None):
    parser = argparse.ArgumentParser()
    parser.add_argument("-4", "--ipv4",              action="store_true",  dest="ipv4",             default=False,                                help="use ipv4 (exclusive with ipv6)")
    parser.add_argument("-6", "--ipv6",              action="store_true",  dest="ipv6",             default=False,                                help="use ipv6 (exclusive with ipv4)")
    parser.add_argument("-0", "--ipv0",              action="store_true",  dest="ipv0",             default=False,                                help="do not specify ip protocol version")
    parser.add_argument("-H", "--hostname",                                dest="hostname",         default="",                                   help="server to check")
    parser.add_argument("-w", "--warning",                                 dest="warning",          default=None,                                 help="Warning level")
    parser.add_argument("-c", "--critical",                                dest="critical",         default=None,                                 help="Critical level")
    parser.add_argument("-t", "--timeout",                                 dest="timeout",          default="30:CRITICAL",                        help="timeout value:STATUS")
    parser.add_argument("-®", "--debug",             action="store_true",  dest="debug",            default=False,                                help="be more verbose")
    args = parser.parse_args(forcedargs)

    setcandebug(args.debug)

    protocol = None
    exclist = 0
    if args.ipv0 == True :
        protocol = ""
        exclist = exclist+1
    if args.ipv4 == True :
        protocol = "-4"
        exclist = exclist+1
        socket.getaddrinfo = forced_ipv4_gai_family
    if args.ipv6 == True :
        protocol = "-6"
        exclist = exclist+1
        socket.getaddrinfo = forced_ipv6_gai_family
    if (exclist>1) :
        exitnagios("CRITICAL","select ip protocol version 4 or 6, or 0 to relay on OS handling")
    elif (exclist<1) :
        exitnagios("CRITICAL","no protocol selected")

    if (args.hostname == ""):
        exitnagios("CRITICAL","hostname not defined")

    parts = args.timeout.split(":")
    timeoutvalue = int(parts[0])
    if len(parts)>1:
        timeoutstatus = parts[1]
    else:
        timeoutstatus = "CRITICAL"
    if timeoutstatus in ["OK", "WARNING", "CRITICAL", "UNKNOWN"]:
        dontptimecheck(protocol,args.hostname,args.warning,args.critical,timeoutvalue,timeoutstatus)
    else:
        exitnagios("CRITICAL","timeout status is not valid")

#-------------------------------------------------------------------------------

if __name__ == "__main__":
    main()

#-------------------------------------------------------------------------------