add check_hauk

check_hauk.py

@@ -0,0 +1,106 @@
+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('-n', '--number', type=int, default=10, help='Number of points to send.')
+    parser.add_argument('-l', '--length', type=int, default=5, help='Length of time to wait in between points.')
+    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)