107 lines
4.0 KiB
Python
Executable File
107 lines
4.0 KiB
Python
Executable File
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)
|