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)