#!/usr/bin/env python3.10 # *************************************************************************** # IIIIII NNN NNN Copyright (C) 2022 Innovative Networks, Inc. # II NNN N All Rights Reserved. Any redistribution or reproduction # II N NN N of part or all of the content of this program in any form # II N NNN without expressed written consent of the copyright holder # IIIIII NNN NNN is strictly prohibited. Please contact admins@in-kc.com # Be Innovative. for additional information. # *************************************************************************** # check_opnsense.py # Author - Ian Perry # # Purpose: Checks health on a OPNsense device. # # Version History: # 2023.10.17 - Initial Creation # 2025.08.13 - T20250505.0018 - Unable to Retrieve Gateway Crash Reports from Gateways Following Software Update to Version 25.1 # 2025.08.20 - T20250505.0018 - Bug fix to previous patch # 2025.10.15 - T20251015.0012 - Disk Usage Reported By Monitoring Is Inaccurate # *************************************************************************** # pylint: disable=bare-except # pylint: disable=broad-exception-caught # pylint: disable=redefined-outer-name # pylint: disable=missing-module-docstring # pylint: disable=too-many-locals try: import inmon_utils as inmon except Exception as e: import sys print("Failed to import inmon_utils %s", e) sys.exit(3) import argparse import logging import os import json import re from functools import reduce import paramiko import requests import datetime import inmon_quant as quant parser = argparse.ArgumentParser(description="Checks health on OPNsense devices.") parser.add_argument( "-i", "--keyfile", dest="keyfile", help="Specify SSH keyfile" ) parser.add_argument( "-H", "--hostname", dest="hostname", help="Specify device hostname", required=True ) parser.add_argument( "-u", "--username", dest="username", help="Specify SSH user to log into device with" ) parser.add_argument( "-k", "--key", dest="key", help="API key" ) parser.add_argument( "-s", "--secret", dest="secret", help="API Key secret" ) parser.add_argument( "-p", "--parameters", dest="parameters", help="Specify parameters for warning, critical, etc in JSON format", default='{}' ) parser.add_argument( "-d", "--debug", dest="debug", help="Enable debugging output", action="store_true" ) parser.add_argument( "-t", "--test", dest="test", help="Which subprogram to run", choices=[ 'all', 'cpu_usage', 'cpu_load', 'state_table_size', 'memory_usage', 'swap_usage', 'disk_usage', 'gateway_health', 'gateway_crash_health', 'carp_maintenance', 'carp_demotion', 'carp_health', 'service_health', 'unbound_health', 'api', 'ssh' ], nargs='+', required=True ) parser.add_argument( "-w", "--warning", dest="warn", help="Warning threshold in either ## or ##:## format, the latter for ranges." ) parser.add_argument( "-c", "--critical", dest="crit", help="Critical threshold in either ## or ##:## format, the latter for ranges." ) parser.add_argument( "--request-timeout", dest="request_timeout", help="Timeout for any requests module requests made", default=10 ) parser.add_argument( "--insecure", dest="insecure", action="store_true", help="Disable SSL Certificate verification" ) parser.add_argument( "--secondary", dest="is_secondary", action="store_true", help="Disables certain checks for secondary firewalls." ) args = parser.parse_args() if args.parameters is not None: try: PARAMS = json.loads(args.parameters) except Exception as e: print(f"Invalid JSON: {e}") sys.exit(3) else: PARAMS = None logger = logging.getLogger(__name__) if args.debug: LOG_LEVEL = "DEBUG" LOG_LEVEL = getattr(logging, LOG_LEVEL.upper()) logging.basicConfig(level = LOG_LEVEL) logger.debug(args.parameters) logger.debug(PARAMS) if args.insecure: import urllib3 urllib3.disable_warnings() requests.packages.urllib3.disable_warnings() # pylint: disable=no-member VERIFY_SSL = False else: VERIFY_SSL = True api_checks = ['gateway_crash_health', 'carp_maintenance', 'carp_demotion', 'carp_health', 'service_health', 'unbound_health'] ssh_checks = ['cpu_usage', 'cpu_load', 'state_table_size', 'memory_usage', 'swap_usage', 'disk_usage', 'gateway_health'] api_run = bool(any([x if x in api_checks else None for x in args.test]) or 'api' in args.test) ssh_run = bool(any([x if x in ssh_checks else None for x in args.test]) or 'ssh' in args.test) # Checks to see if any api checks are going to be run. if api_run and (args.secret is None or args.key is None): raise ValueError("-s/--secret and -k/--key are required if any api check is used.") if any([x if x in api_checks else None for x in args.test]) and 'api' in args.test: raise ValueError("Cannot specify to check all api checks and individual checks.") # Checks to see if any ssh checks are going to be run if ssh_run and (args.keyfile is None or args.username is None): raise ValueError("-i/--keyfile and -u/--username are required if any ssh check is used.") if any([x if x in ssh_checks else None for x in args.test]) and 'ssh' in args.test: raise ValueError("Cannot specify to check all ssh checks and individual checks.") if 'all' in args.test and len(args.test) > 1: raise ValueError("Cannot specify all checks to be run and individual checks.") # Set up requests session if we're using api checks. if api_run or 'all' in args.test: session = requests.Session() auth_container = requests.auth.HTTPBasicAuth(args.key, args.secret) session.auth = auth_container # Create an SSH tunnel to the device. if ssh_run or 'all' in args.test: keyfile = os.path.expanduser(args.keyfile) client = paramiko.client.SSHClient() client.load_system_host_keys() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( str(args.hostname), username=str(args.username), key_filename=str(args.keyfile), ) def get_func_params(params: dict, name: str) -> dict: """ Provides function parameters given the default parameters, or none if not found Args: params (dict): The program parameters argument converted to a dictionary. name (str): The name of the check. Returns: None|dict: Either a dictionary containing the correct subdictionary, or None. """ if params is not None and name in params: return params[name] return {} def handle_html_err(r: requests.Response) -> str|None: """ Handles API request errors. Args: r (requests.Response): A response from API call Returns: str|None: str of error output or None if ok. """ if r.ok: return None try: j = json.loads(r.text) errmsg = j['message'] if 'message' in j else r.text except Exception as e: errmsg = r.text errmsg2 = e return f"API call returned non-200 status code {r.status_code} with output: {errmsg} - {errmsg2}" def check_cpu_usage() -> inmon.CheckResult: """ Checks CPU percentage Args: client (paramiko.client.SSHClient): A configured SSH session to the host to check Raises: RuntimeError: Raised when the underlying command ran returns non-zero Returns: inmon.CheckResult: A CheckResult with the results of the check. """ # Verify variables default = ("80%", "90%") temp_params = get_func_params(PARAMS, 'cpu_usage') threshold = quant.parse_thresholds(temp_params, args.warn, args.crit, default) # Get information from SSH logger.debug("Performing iostat on remote.") stdin, stdout, stderr = client.exec_command('iostat') # nosec - sanitized exit_status = stdout.channel.recv_exit_status() # Check function call return status if exit_status != 0: if stderr: error = '\n'.join(stderr.read().splitlines()) raise RuntimeError(f"iostat returned a non-zero exit code: {exit_status} - {error}") stdin.close() # Parse return text cpu_usage = stdout.read().splitlines()[2].decode().split()[-5:] cpu_usage = [quant.Quantity(value=float(x), uom="%") for x in cpu_usage] logger.debug(cpu_usage) # Define headers, and then assign them and the values do a dictionary. headers = ['user', 'nice', 'system', 'interrupt', 'idle'] values = dict(zip(headers, cpu_usage)) values["used"] = reduce(lambda x,y: x+y, cpu_usage[0:3]) # Calculate used value logger.debug("Used value: %s", values["used"]) logger.debug("Threshold: %s", threshold) return_val = quant.parse_stat(values["used"], threshold) # Check return value check = inmon.CheckResult(return_val, f"CPU {values['used']} used") check.add_perfdata(inmon.PerfData( 'user', values['user'], minimum=0, maximum=100 )) check.add_perfdata(inmon.PerfData( 'nice', values['nice'], minimum=0, maximum=100 )) check.add_perfdata(inmon.PerfData( 'system', values['system'] , minimum=0, maximum=100 )) check.add_perfdata(inmon.PerfData( 'interrupt', values['interrupt'], minimum=0, maximum=100 )) check.add_perfdata(inmon.PerfData( 'idle', values['idle'], minimum=0, maximum=100 )) check.add_perfdata(inmon.PerfData( 'used', values["used"], thresholds=threshold, minimum=0, maximum=100 )) return check def check_cpu_load() -> inmon.MultiActiveCheck: """ Checks the cpu load on an OPNsense host Args: client (paramiko.client.SSHClient): An open paramiko client to the desired device. Raises: ValueError: Raised if warning values are not provided in correct format. ValueError: Raised if critical values are not provided in correct format. RuntimeError: Raised if the function called on remote returns a non-zero code. RuntimeError: Raised if the function called returns invalid output. Returns: inmon.MultiActiveCheck: A MultiActiveCheck object """ default_1 = ("5", "10") default_5 = ("4", "6") default_15 = ("3", "4") if args.warn is not None and len(args.warn.split('/')) != 3: raise ValueError("Warning value must be provided in format 1/5/15.") if args.crit is not None and len(args.crit.split('/')) != 3: raise ValueError("Critical values must be provided in format 1/5/15.") args_warn = args.warn.split('/') if args.warn is not None else [None, None, None] args_crit = args.crit.split('/') if args.crit is not None else [None, None, None] temp_params = get_func_params(PARAMS, 'cpu_load') if temp_params is None: temp_params = {} params_1 = None if '1' not in temp_params else temp_params['1'] params_5 = None if '5' not in temp_params else temp_params['5'] params_15 = None if '15' not in temp_params else temp_params['15'] thresholds_1 = quant.parse_thresholds(params_1, args_warn[0], args_crit[0], default_1) thresholds_5 = quant.parse_thresholds(params_5, args_warn[1], args_crit[1], default_5) thresholds_15 = quant.parse_thresholds(params_15, args_warn[2], args_crit[2], default_15) logger.debug("cpu_load() - running uptime command on remote") stdin, stdout, stderr = client.exec_command('uptime') # nosec - sanitized exit_status = stdout.channel.recv_exit_status() stdin.close() if exit_status != 0: error = '\n'.join(stderr.read().splitlines()) if stderr else "Program returned no error output." raise RuntimeError(f"uptime returned a non-zero exit code: {exit_status} - {error}") cpu_load = stdout.read().splitlines()[0].decode() logger.debug("cpu_load() - uptime output: %s", cpu_load) match = re.search(r'.*: (.*)$', cpu_load) if match is None: out = '\n'.join(cpu_load) raise RuntimeError(f"uptime returned invalid data: {out}") data = match.group(1).split(', ') data = [quant.Quantity(value=float(x)) for x in data] [retval_1, retval_5, retval_15] = [ quant.parse_stat(data[0], thresholds_1), quant.parse_stat(data[1], thresholds_5), quant.parse_stat(data[2], thresholds_15) ] logger.debug("cpu_load() - data: %s", data) multi = inmon.MultiActiveCheck() check_1 = inmon.CheckResult(retval_1, f"Load 1: {data[0]}") check_5 = inmon.CheckResult(retval_5, f"Load 5: {data[1]}") check_15 = inmon.CheckResult(retval_15, f"Load 15: {data[2]}") check_1.add_perfdata(inmon.PerfData('load_1', data[0], thresholds_1)) check_5.add_perfdata(inmon.PerfData('load_5', data[1], thresholds_5)) check_15.add_perfdata(inmon.PerfData('load_15', data[2], thresholds_15)) multi.append(check_1) multi.append(check_5) multi.append(check_15) return multi def check_state_table_size() -> inmon.CheckResult: """ Checks the state table size agains the maximum defined. Args: client (paramiko.client.SSHClient): An open paramiko client to the desired device. Raises: RuntimeError: Raised if the function called on remote returns a non-zero code. RuntimeError: Raised if the function called on remote returns a non-zero code. Returns: inmon.CheckResult: The check result. """ default = ("75%", "85%") temp_params = get_func_params(PARAMS, 'state_table_size') thresholds = quant.parse_thresholds(temp_params, args.warn, args.crit, default) logger.debug("check_state_table_size() - Running sudo pfctl -s state | wc -l on remote") stdin, stdout, stderr = client.exec_command('sudo pfctl -s state | wc -l') # nosec - sanitized exit_status = stdout.channel.recv_exit_status() if exit_status != 0: error = '\n'.join(stderr.read().decode().splitlines()) if stderr else "Program returned no error output." raise RuntimeError(f"uptime returned a non-zero exit code: {exit_status} - {error}") state_table_size = stdout.read().decode().splitlines() logger.debug("check_state_table_size() - output: %s", state_table_size) state_table_size = int([x.strip() for x in state_table_size if x is not None][0]) stdin, stdout, stderr = client.exec_command("sudo pfctl -sm | grep states | awk '{print $4}'") # nosec - sanitized if exit_status != 0: error = '\n'.join(stderr.read().splitlines()) if stderr else "Program returned no error output." raise RuntimeError(f"uptime returned a non-zero exit code: {exit_status} - {error}") stdin.close() state_table_max = int(stdout.read().decode().splitlines()[0]) logger.debug("check_state_table_size() - output: %s", state_table_max) check_val = quant.Quantity(value=round((state_table_size/state_table_max)*100, 2), uom="%") return_val = quant.parse_stat(check_val, thresholds) check_output = f"State size at {check_val} full, {state_table_size:,d}/{state_table_max:,d}" check = inmon.CheckResult(return_val, check_output) check.add_perfdata(inmon.PerfData('state_table_size_percent', check_val, thresholds=thresholds, minimum=0, maximum=100)) check.add_perfdata(inmon.PerfData('state_table_size', quant.Quantity(value=state_table_size), maximum=state_table_max)) return check def check_memory_usage() -> inmon.CheckResult: """ Checks memory usage on an opnsense firewall. Args: client (paramiko.client.SSHClient): An open paramiko client to the desired device. Raises: RuntimeError: Raised if the function called on remote returns a non-zero code. ValueError: Raised if warning is greater than critical Returns: inmon.CheckResult: A check result. """ default = ("80%", "90%") temp_params = get_func_params(PARAMS, 'memory_usage') thresholds = quant.parse_thresholds(temp_params, args.warn, args.crit, default) stdin, stdout, stderr = client.exec_command('top -b | grep "Mem:"') # nosec - sanitized exit_status = stdout.channel.recv_exit_status() if exit_status != 0: error = '\n'.join(stderr.read().splitlines()) if stderr else "Program returned no error output." raise RuntimeError(f"Command returned a non-zero exit code: {exit_status} - {error}") stdin2, stdout2, stderr2 = client.exec_command('sysctl hw.physmem') # nosec - sanitized exit_status = stdout2.channel.recv_exit_status() if exit_status != 0: error = '\n'.join(stderr.read().splitlines()) if stderr else "Program returned no error output." raise RuntimeError(f"Command returned a non-zero exit code: {exit_status} - {error}") stdin.close() memory_usage = stdout.read().decode().splitlines()[0] memory_usage = memory_usage[5:].split(', ') memory_usage = {y: quant.parse_bytes(x, 'B', True) for x, y in [z.split() for z in memory_usage]} memory_usage['Total'] = int((stdout2.read().decode().splitlines()[0]).split(': ')[1]) memory_usage['Used'] = memory_usage['Total'] - (memory_usage['Free'] + memory_usage['Inact']) # if warn_high is percentage, we calc by percentage. val_abs = val_to_check = quant.Quantity(value=memory_usage['Used'], uom="B") val_pct = quant.Quantity(value=round((memory_usage['Used']/memory_usage['Total'])*100, 2), uom="%") val_to_check = val_pct if thresholds.warn_high.uom == "%" else val_abs return_val = quant.parse_stat(val_to_check, thresholds) tot = quant.format_bytes(memory_usage['Total']) ptot = f"{tot[0]}{tot[1]}B" check = inmon.CheckResult(return_val, f"Memory usage {val_abs.format()}/{ptot} - {val_pct}") for key, value in memory_usage.items(): if key == "Used": check.add_perfdata(inmon.PerfData("used", val_to_check, thresholds)) elif key != "Total": check.add_perfdata(inmon.PerfData(f"{key.lower()}", quant.Quantity(value=value, uom="B"), minimum=0, maximum=memory_usage['Total'])) return check def check_swap_usage() -> inmon.CheckResult: """ Checks swap usage on an opnsense firewall Args: client (paramiko.client.SSHClient): An open paramiko client to the desired device. Raises: RuntimeError: Raised if the function called on remote returns a non-zero code. ValueError: Raised if warning is greater than critical Returns: inmon.CheckResult: A check result. """ default = ("80%", "90%") temp_params = get_func_params(PARAMS, 'swap_usage') thresholds = quant.parse_thresholds(temp_params, args.warn, args.crit, default) stdin, stdout, stderr = client.exec_command('swapinfo') # nosec - sanitized exit_status = stdout.channel.recv_exit_status() if exit_status != 0: error = '\n'.join(stderr.read().splitlines()) if stderr else "Program returned no error output." raise RuntimeError(f"Command returned a non-zero exit code: {exit_status} - {error}") stdin.close() swap_usage = stdout.read().decode().splitlines()[-1].split() used = int(swap_usage[2]) free = int(swap_usage[3]) total = used + free val_abs = quant.Quantity(value=used) val_pct = quant.Quantity(value=(used/total), uom="%") pct = True if thresholds.warn_high.uom == "%" else val_abs val_to_check = val_pct if pct else val_abs return_val = quant.parse_stat(val_to_check, thresholds) check = inmon.CheckResult(return_val, f"Swap usage {used:,d}/{total:,d} blocks - {val_pct}") check.add_perfdata(inmon.PerfData("used", val_to_check, thresholds, minimum=0, maximum=100 if pct else total)) return check def check_disk_usage() -> inmon.CheckResult: """ Checks disk usage Args: client (paramiko.client.SSHClient): An open paramiko client to the desired device. Raises: RuntimeError: Raised if the function called on remote returns a non-zero code. ValueError: Raised if warning is greater than critical Returns: inmon.CheckResult: A check result. """ default = ("80%", "90%") temp_params = get_func_params(PARAMS, 'disk_usage') thresholds = quant.parse_thresholds(temp_params, args.warn, args.crit, default) if temp_params is not None and 'partition' in temp_params: partition = temp_params['partition'] else: partition = '/dev/gpt/rootfs' stdin, stdout, stderr = client.exec_command(f'df -h | grep {partition}') # nosec - sanitized exit_status = stdout.channel.recv_exit_status() if exit_status != 0: error = '\n'.join(stderr.read().splitlines()) if stderr else "Program returned no error output." raise RuntimeError(f"Command returned a non-zero exit code: {exit_status} - {error}") disk_usage = stdout.read().decode().splitlines()[0].split() stdin.close() total = quant.parse_bytes(disk_usage[1], 'B', True) used = quant.parse_bytes(disk_usage[2], 'B', True) val_abs = quant.Quantity(value=used, uom="B") val_pct = quant.Quantity(value=float(disk_usage[4].strip("%")), uom="%") val_to_check = val_pct if thresholds.warn_high.uom == "%" else val_abs return_val = quant.parse_stat(val_to_check, thresholds) uv, uu = quant.format_bytes(used) logger.debug(uv) tv, tu = quant.format_bytes(total) logger.debug(uu) logger.debug(tu) uf = f"{uv}{uu}B" tf = f"{tv}{tu}B" check = inmon.CheckResult(return_val, f"Disk usage {uf}/{tf} - {val_pct}") check.add_perfdata(inmon.PerfData("used", quant.Quantity(value=used, uom="B"), minimum=0, maximum=total)) return check def check_gateway_health() -> inmon.MultiActiveCheck: """ Checks gateway health on an OPNsense firewall Args: client (paramiko.client.SSHClient): An open paramiko client to the desired device. Raises: RuntimeError: Raised if the function called on remote returns a non-zero code. Returns: inmon.MultiActiveCheck: A check result. """ default = ("10%", "20%") temp_params = get_func_params(PARAMS, 'gateway_health') thresholds = quant.parse_thresholds(temp_params, args.warn, args.crit, default) stdin, stdout, stderr = client.exec_command('pluginctl -r return_gateways_status') # nosec - sanitized exit_status = stdout.channel.recv_exit_status() if exit_status != 0: error = '\n'.join(stderr.read().splitlines()) if stderr else "Program returned no error output." raise RuntimeError(f"Command returned a non-zero exit code: {exit_status} - {error}") status = json.loads(''.join(stdout.read().decode().splitlines()))['dpinger'] stdin.close() container = inmon.MultiActiveCheck() for gw_name, gw in status.items(): # Ignore mgmt loopback. if gw_name == "MGMT_Loopback": continue logger.debug(gw) if temp_params is not None and 'alt_px' in temp_params and temp_params['alt_px']: val = quant.Quantity(value=int(gw["loss"].split()[0]), uom="%") res = quant.parse_stat(val, thresholds) text = f'{gw_name} - delay {gw["delay"]}, loss {gw["loss"]}, stddev {gw["stddev"]}' temp = inmon.CheckResult(res, text) elif gw['status'] == 'down': container.append(inmon.CheckResult(2, f"{gw_name} is down")) continue elif gw['status'] != 'none': text = f'{gw_name} is down - delay {gw["delay"]}, loss {gw["loss"]}, stddev {gw["stddev"]}' temp = inmon.CheckResult(2, text) else: temp = inmon.CheckResult(0, f'{gw_name} up') logger.debug("Gateway: %s", gw) stddev, stuom = gw["stddev"].split() std = quant.Quantity(value=float(stddev), uom=stuom) delay, duom = gw["delay"].split() delay = quant.Quantity(value=float(delay), uom=duom) temp.add_perfdata(inmon.PerfData( f"{gw_name.lower()}_loss", quant.Quantity(value=gw["loss"].split()[0], uom='%') )) temp.add_perfdata(inmon.PerfData(f"{gw_name.lower()}_delay", delay)) temp.add_perfdata(inmon.PerfData(f"{gw_name.lower()}_stddev", std)) container.append(temp) return container def check_gateway_crash_health() -> inmon.CheckResult: """ Checks gateway crash health on OPNsense. Raises: ValueError: Raised if API secret is not provided. Returns: inmon.CheckResult: A check result. """ if args.secret is None: raise ValueError("API Secret must be provided when executing API checks.") temp_params = get_func_params(PARAMS, 'gateway_crash_health') timeout = temp_params['timeout'] if temp_params is not None and 'timeout' in temp_params else args.request_timeout result = session.get( f"https://{args.hostname}/api/core/system/status", auth=auth_container, timeout=int(timeout), verify=VERIFY_SSL ) if handle_html_err(result) is not None: return inmon.CheckResult(3, handle_html_err(result)) data = json.loads(result.text) container = inmon.MultiActiveCheck() for top_name, top_item in data.items(): # Search for keys under top level of keys for name, item in top_item.items(): # Search under nested keys if "status" in item: # Does nested key have "status"? if str(item["status"]) == "2" or str(item["status"]) == "OK": # Status is either int or string. Check for both. temp = inmon.CheckResult(0, f"No issues in {name}") else: # Print output depending on if timetamp key exists and has a value if "timestamp" in item: if item["timestamp"] is str: timestamp = datetime.datetime.fromtimestamp(int(item['timestamp'])).strftime('%Y-%m-%d %H:%M:%S') temp = inmon.CheckResult(2, f"Issue in {name} - {item['message']} at time {timestamp}") else: temp = inmon.CheckResult(2, f"Issue in {name} - {item['message']}") else: temp = inmon.CheckResult(2, f"Issue in {name} - {item['message']}") container.append(temp) if temp is None: temp = inmon.CheckResult(3, "No status recorded/reported") container.append(temp) return container def check_carp_health() -> inmon.CheckResult|inmon.MultiActiveCheck: """ Checks CARP health, specifically if any interfaces are backup. Returns: inmon.CheckResult: A check result. """ temp_params = get_func_params(PARAMS, 'carp_health') timeout = temp_params['timeout'] if temp_params is not None and 'timeout' in temp_params else args.request_timeout result = session.get( f"https://{args.hostname}/api/diagnostics/interface/get_vip_status", auth=auth_container, timeout=int(timeout), verify=VERIFY_SSL ) if handle_html_err(result) is not None: return inmon.CheckResult(3, handle_html_err(result)) data = json.loads(result.text) container = inmon.MultiActiveCheck() if data['rowCount'] == 0: return inmon.CheckResult(0, "There are no CARP interfaces on this firewall") for iface in data['rows']: if iface['status'] == "BACKUP": if args.is_secondary: t = inmon.CheckResult(0, f"{iface['interface']} is BACKUP (secondary)") else: t = inmon.CheckResult(2, f"{iface['interface']} is BACKUP") elif iface['status'] == "MASTER": t = inmon.CheckResult(0, f"{iface['interface']} is MASTER") else: t = inmon.CheckResult(3, f"{iface['interface']} is UNKNOWN") container.append(t) logger.debug(container) return container def check_carp_demotion() -> inmon.CheckResult: """ Checks CARP demotion level on an OPNsense firewall. Returns: inmon.CheckResult: A check result. """ temp_params = get_func_params(PARAMS, 'carp_demotion') timeout = temp_params['timeout'] if temp_params is not None and 'timeout' in temp_params else args.request_timeout result = session.get( f"https://{args.hostname}/api/diagnostics/interface/get_vip_status", auth=auth_container, timeout=int(timeout), verify=VERIFY_SSL ) if handle_html_err(result) is not None: return inmon.CheckResult(3, handle_html_err(result)) data = json.loads(result.text) demotion = int(data['carp']['demotion']) if not data['carp']['maintenancemode'] and demotion != 0: return inmon.CheckResult(2, f"CARP demotion {demotion} outside maintenance mode") elif data['carp']['maintenancemode']: return inmon.CheckResult(0, "CARP maintenance mode active") elif not data['carp']['maintenancemode'] and demotion == 0: return inmon.CheckResult(0, "CARP status nominal") else: return inmon.CheckResult(3, "CARP status unknown.") def check_carp_maintenance() -> inmon.CheckResult: """ Checks if an OPNsense firewall is in CARP maintenance mode. Returns: inmon.CheckResult: A check result. """ temp_params = get_func_params(PARAMS, 'carp_maint') timeout = temp_params['timeout'] if temp_params is not None and 'timeout' in temp_params else args.request_timeout result = session.get( f"https://{args.hostname}/api/diagnostics/interface/get_vip_status", auth=auth_container, timeout=int(timeout), verify=VERIFY_SSL ) if handle_html_err(result) is not None: return inmon.CheckResult(3, handle_html_err(result)) data = json.loads(result.text) if data['carp']['maintenancemode']: return inmon.CheckResult(2, "Device in CARP maintenance mode") else: return inmon.CheckResult(0, "Device not in CARP maintenance mode") def check_service_health() -> inmon.CheckResult: """ Checks service health on an OPNsense firewall Returns: inmon.CheckResult: A check result. """ temp_params = get_func_params(PARAMS, 'service_health') timeout = temp_params['timeout'] if temp_params is not None and 'timeout' in temp_params else args.request_timeout logger.debug(timeout) logger.debug(type(timeout)) result = session.get( f"https://{args.hostname}/api/core/service/search", auth=auth_container, timeout=int(timeout), verify=VERIFY_SSL ) if handle_html_err(result) is not None: return inmon.CheckResult(3, handle_html_err(result)) data = json.loads(result.text) container = inmon.MultiActiveCheck() logger.debug(temp_params) for service in data['rows']: if service['running'] == 1: container.append(inmon.CheckResult(0, f"{service['id']} - running")) elif 'ignore' in temp_params and service['id'] in temp_params['ignore']: container.append(inmon.CheckResult(0, f"{service['id']} - stopped (ignore)")) elif args.is_secondary and ( re.match(r'openvpn/\d+', service['id']) is not None or service['id'] == 'mdns-repeater' ): # This is the case for secondary firewalls. We're expecting openVPN and mdns to be off. container.append(inmon.CheckResult(0, f"{service['id']} - stopped (secondary)")) else: container.append(inmon.CheckResult(2, f"{service['id']} - stopped")) return container def check_unbound_health() -> inmon.CheckResult: """ Gets unbound statistics on an OPNsense firewall. Will not alert. Returns: inmon.CheckResult: A check result. """ temp_params = get_func_params(PARAMS, 'unbound_health') timeout = temp_params['timeout'] if 'timeout' in temp_params else args.request_timeout result = session.get( f"https://{args.hostname}/api/unbound/diagnostics/stats", auth=auth_container, timeout=int(timeout), verify=VERIFY_SSL ) if handle_html_err(result) is not None: return inmon.CheckResult(3, handle_html_err(result)) data = json.loads(result.text) data = data['data']['total'] check = inmon.CheckResult(0, "Unbound stats gathered") try: check.add_perfdata( inmon.PerfData('cache_hits', quant.Quantity(value=data['num']['cachehits'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('cache_miss', quant.Quantity(value=data['num']['cachemiss'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('dnscrypt_cert', quant.Quantity(value=data['num']['dnscrypt']['cert'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('dnscrypt_cleartext', quant.Quantity(value=data['num']['dnscrypt']['cleartext'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('dnscrypt_crypted', quant.Quantity(value=data['num']['dnscrypt']['crypted'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('dnscrypt_malformed', quant.Quantity(value=data['num']['dnscrypt']['malformed'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('expired', quant.Quantity(value=data['num']['expired'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('prefetch', quant.Quantity(value=data['num']['prefetch'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('queries', quant.Quantity(value=data['num']['queries'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('queries_cookie_client', quant.Quantity(value=data['num']['queries_cookie_client'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('queries_cookie_invalid', quant.Quantity(value=data['num']['queries_cookie_invalid'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('queries_cookie_valid', quant.Quantity(value=data['num']['queries_cookie_valid'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('queries_ip_ratelimited', quant.Quantity(value=data['num']['queries_ip_ratelimited'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('queries_timed_out', quant.Quantity(value=data['num']['queries_timed_out'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('recursive_replies', quant.Quantity(value=data['num']['recursivereplies'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('query_queue_time_max', quant.Quantity(value=data['query']['queue_time_us']['max'], uom='us')) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('recursion_time_avg', quant.Quantity(value=data['recursion']['time']['avg'], uom='s')) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('recursion_time_median', quant.Quantity(value=data['recursion']['time']['median'], uom='s')) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('request_list_avg', quant.Quantity(value=data['requestlist']['avg'], uom='s')) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('request_list_current_all', quant.Quantity(value=data['requestlist']['current']['all'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('request_list_current_user', quant.Quantity(value=data['requestlist']['current']['user'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('request_list_exceeded', quant.Quantity(value=data['requestlist']['exceeded'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('request_list_max', quant.Quantity(value=data['requestlist']['max'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('request_list_overwritten', quant.Quantity(value=data['requestlist']['overwritten'])) ) except: # noqa: E722 # nosec pass try: check.add_perfdata( inmon.PerfData('tcp_usage', quant.Quantity(value=data['tcpusage'])) ) except: # noqa: E722 # nosec pass return check if 'all' in args.test and len(args.test) > 1: raise ValueError("If all is a test sent, it must be the only test.") checks_to_run = [] if 'all' in args.test: checks_to_run.append(api_checks) checks_to_run.append(ssh_checks) if 'api' in args.test: checks_to_run.append(api_checks) args.test.remove('api') if 'ssh' in args.test: checks_to_run.append(ssh_checks) args.test.remove('ssh') checks_to_run.append(args.test) # Flattens the list. checks_to_run = [x for xs in checks_to_run for x in xs] check_container = inmon.MultiActiveCheck() logger.debug(checks_to_run) for check in checks_to_run: logger.debug(check) match check: case 'cpu_usage': cur_check = check_cpu_usage() case 'cpu_load': cur_check = check_cpu_load() case 'state_table_size': cur_check = check_state_table_size() case 'memory_usage': cur_check = check_memory_usage() case 'swap_usage': cur_check = check_swap_usage() case 'disk_usage': cur_check = check_disk_usage() case 'gateway_health': cur_check = check_gateway_health() case 'gateway_crash_health': cur_check = check_gateway_crash_health() case 'carp_maintenance': cur_check = check_carp_maintenance() case 'carp_health': cur_check = check_carp_health() case 'carp_demotion': cur_check = check_carp_demotion() case 'service_health': cur_check = check_service_health() case 'unbound_health': cur_check = check_unbound_health() #logger.debug("%s", cur_check) check_container.append(cur_check) check_container.process()