icinga2-checks/check_hauk.py

import argparse
import sys
import time
import traceback

import requests

from checker import nagios
from checker import print_icinga2_check_status


def quit_check(text_result, exit_code):
    print_icinga2_check_status(text_result, exit_code)
    sys.exit(exit_code)


def main():
    parser = argparse.ArgumentParser(description='Validate Hauk tracking functionality.')
    parser.add_argument('-b', '--base', required=True, help='Base path to your Hauk installation.')
    parser.add_argument('-l', '--length', type=int, default=5, help='Length of time to wait in between points.')
    parser.add_argument('-n', '--number', type=int, default=10, help='Number of points to send.')
    parser.add_argument('-p', '--password', default='', help='Password for Hauk.')
    args = parser.parse_args()

    # Define the parameters for the share
    send_params = {
        "dur": min(100, (args.number * args.length) + 30),  # Duration in seconds
        "int": args.length,  # Interval in seconds
        "mod": 0,  # Share mode (0 for solo share)
        "ado": 0,  # Whether the share is adoptable (0 for not adoptable)
        "e2e": 0,  # End-to-end encryption (0 for disabled)
        "pwd": args.password,
        "salt": "",  # Salt for encryption (empty if encryption is disabled)
    }

    # Create a share
    response = requests.post(args.base + '/api/create.php', data=send_params)
    if response.status_code != 200:
        quit_check(f'failed to create share, received response code {response.status_code}', nagios.CRITICAL)

    lines = response.text.split("\n")
    if lines[0] != "OK":
        r = response.text.strip("\n")
        quit_check(f'failed to create share - "{r}"', nagios.CRITICAL)

    session_id = lines[1]
    view_link = lines[2]
    share_id = lines[3]

    # Simulate location updates
    for i in range(args.number):
        lat = lon = i + 1
        point_params = {
            "sid": session_id,
            "time": time.time(),
            "acc": 1,  # accuracy
            "lat": lat,
            "lon": lon
        }
        # Send a location update
        response = requests.post(args.base + '/api/post.php', data=point_params)
        if response.status_code != 200:
            quit_check(f'failed to update location, received response code {response.status_code}', nagios.CRITICAL)

        # Verify the server's response
        if response.text.split("\n")[0] != "OK":
            r = response.text.strip("\n")
            quit_check(f'failed to update location - "{r}"', nagios.CRITICAL)

        # Fetch the points from the server
        points_response = requests.get(args.base + f'/api/fetch.php?id={share_id}&since=0')
        if points_response.status_code != 200:
            quit_check(f'failed to fetch points, received response code {points_response.status_code}', nagios.CRITICAL)

        # Verify the points
        points_data = []  # make PyCharm happy
        try:
            points_data = points_response.json()['points']
        except requests.exceptions.JSONDecodeError:
            r = points_response.text.strip("\n")
            quit_check(f'Failed to parse points JSON\n{r}', nagios.STATE_CRIT)

        latest_point = points_data[-1]
        if latest_point[0] != lat or latest_point[1] != lon:
            quit_check(f"Point {i} has incorrect coordinates. Expected {lat, lon}, got {latest_point[0], latest_point[1]}.", nagios.CRITICAL)

        time.sleep(args.length)

    # End the share
    stop_response = requests.post(args.base + '/api/stop.php', data={'sid': share_id})
    if stop_response.status_code != 200:
        quit_check(f'failed to stop share, received response code {stop_response.status_code}', nagios.CRITICAL)
    if stop_response.text.split("\n")[0] != "OK":
        r = stop_response.text.strip("\n")
        quit_check(f'failed to stop share - "{r}"', nagios.CRITICAL)

    quit_check(f"Hauk is fully functional", nagios.STATE_OK)


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"UNKNOWN: exception\n{e}")
        print(traceback.format_exc())
        sys.exit(nagios.UNKNOWN)