vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2025-12-07 01:26:11 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

OMDSDNOPENAPI - Rephrased error messages, improved logging and code logic

Browse Source
This commit is contained in:
xfilo
2025-02-24 16:03:07 +01:00
parent 52de3ae872
commit bf3fdd2766
1 changed files with 76 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
front/plugins/omada_sdn_openapi/script.py
View File

@@ -17,6 +17,7 @@ Author: https://github.com/xfilo
__author__ = "xfilo" __author__ = "xfilo"
__version__ = 0.1 # Initial version __version__ = 0.1 # Initial version
__version__ = 0.2 # Rephrased error messages, improved logging and code logic
import os import os
import sys import sys
@@ -59,7 +60,7 @@ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class OmadaHelper: class OmadaHelper:
@staticmethod @staticmethod
def log(message: str, level: Literal["minimal", "verbose", "debug", "trace"] = "minimal") -> None: def log(message: str, level: Literal["minimal", "verbose", "debug", "trace"] = "minimal") -> None:
mylog(level, [f"[{pluginName}] {message}"]) mylog(level, [f"[{pluginName}] [{level[:1].upper()}] {message}"])
@staticmethod @staticmethod
def debug(message: str) -> None: def debug(message: str) -> None:
@@ -75,8 +76,7 @@ class OmadaHelper:
@staticmethod @staticmethod
def response(response_type: str, response_message: str, response_result: Any = None) -> Dict[str, Any]: def response(response_type: str, response_message: str, response_result: Any = None) -> Dict[str, Any]:
return {"response_type": response_type, "response_message": response_message, return {"response_type": response_type, "response_message": response_message, "response_result": response_result}
"response_result": response_result}
@staticmethod @staticmethod
def timestamp_to_datetime(ms: int, timezone: str) -> Dict[str, Any]: def timestamp_to_datetime(ms: int, timezone: str) -> Dict[str, Any]:
@@ -127,7 +127,7 @@ class OmadaHelper:
if not isinstance(input_data, list): if not isinstance(input_data, list):
raise Exception(f"Expected a list, but got '{type(input_data)}'.") raise Exception(f"Expected a list, but got '{type(input_data)}'.")
OmadaHelper.verbose(f"Starting normalization of {len(input_data)} {input_type}(s) from site {site_name}") OmadaHelper.verbose(f"Starting normalization of {len(input_data)} {input_type}(s) from site: {site_name}")
default_entry = { default_entry = {
"mac_address": "", "mac_address": "",
"ip_address": "", "ip_address": "",
@@ -158,45 +158,35 @@ class OmadaHelper:
entry["name"] = data.get("name", "") entry["name"] = data.get("name", "")
last_seen = OmadaHelper.timestamp_to_datetime(data.get("lastSeen", 0), timezone) last_seen = OmadaHelper.timestamp_to_datetime(data.get("lastSeen", 0), timezone)
entry["last_seen"] = last_seen.get("response_result") if isinstance(last_seen, dict) and last_seen.get( entry["last_seen"] = last_seen.get("response_result") if isinstance(last_seen, dict) and last_seen.get("response_type") == "success" else ""
"response_type") == "success" else ""
if input_type == "device": if input_type == "device":
entry["device_type"] = data.get("type") entry["device_type"] = data.get("type")
if data.get("type", "") != "gateway": if data.get("type", "") != "gateway":
parent_mac = OmadaHelper.normalize_mac(data.get("uplinkDeviceMac")) parent_mac = OmadaHelper.normalize_mac(data.get("uplinkDeviceMac"))
parent_mac = parent_mac.get("response_result") if isinstance(parent_mac, entry["parent_node_mac_address"] = parent_mac.get("response_result") if isinstance(parent_mac, dict) and parent_mac.get("response_type") == "success" else ""
dict) and parent_mac.get(
"response_type") == "success" else ""
entry["parent_node_mac_address"] = parent_mac
if input_type == "client": if input_type == "client":
entry["vlan_id"] = data.get("vid") entry["vlan_id"] = data.get("vid")
entry["device_type"] = data.get("deviceType") entry["device_type"] = data.get("deviceType")
if data.get("connectDevType", "") == "gateway": if data.get("connectDevType", "") == "gateway":
parent_mac = OmadaHelper.normalize_mac(data.get("gatewayMac")) parent_mac = OmadaHelper.normalize_mac(data.get("gatewayMac"))
entry["parent_node_mac_address"] = parent_mac.get("response_result") if isinstance(parent_mac, entry["parent_node_mac_address"] = parent_mac.get("response_result") if isinstance(parent_mac, dict) and parent_mac.get("response_type") == "success" else ""
dict) and parent_mac.get(
"response_type") == "success" else ""
entry["parent_node_port"] = data.get("port", "") entry["parent_node_port"] = data.get("port", "")
elif data.get("connectDevType", "") == "switch": elif data.get("connectDevType", "") == "switch":
parent_mac = OmadaHelper.normalize_mac(data.get("switchMac")) parent_mac = OmadaHelper.normalize_mac(data.get("switchMac"))
entry["parent_node_mac_address"] = parent_mac.get("response_result") if isinstance(parent_mac, entry["parent_node_mac_address"] = parent_mac.get("response_result") if isinstance(parent_mac, dict) and parent_mac.get("response_type") == "success" else ""
dict) and parent_mac.get(
"response_type") == "success" else ""
entry["parent_node_port"] = data.get("port", "") entry["parent_node_port"] = data.get("port", "")
elif data.get("connectDevType", "") == "ap": elif data.get("connectDevType", "") == "ap":
parent_mac = OmadaHelper.normalize_mac(data.get("apMac")) parent_mac = OmadaHelper.normalize_mac(data.get("apMac"))
entry["parent_node_mac_address"] = parent_mac.get("response_result") if isinstance(parent_mac, entry["parent_node_mac_address"] = parent_mac.get("response_result") if isinstance(parent_mac, dict) and parent_mac.get("response_type") == "success" else ""
dict) and parent_mac.get(
"response_type") == "success" else ""
entry["parent_node_ssid"] = data.get("ssid", "") entry["parent_node_ssid"] = data.get("ssid", "")
result.append(entry) result.append(entry)
OmadaHelper.debug(f"Processed {input_type} entry: {entry}") OmadaHelper.debug(f"Processed {input_type} entry: {entry}")
msg = f"Successfully normalized {len(result)} {input_type}(s) from site: {site_name}" msg = f"Successfully normalized {len(result)} {input_type}(s) from site: {site_name}"
OmadaHelper.verbose(msg) OmadaHelper.minimal(msg)
final_result = OmadaHelper.response("success", msg, result) final_result = OmadaHelper.response("success", msg, result)
except Exception as ex: except Exception as ex:
msg = f"Failed normalizing {input_type}(s) from site '{site_name}' - error: {str(ex)}" msg = f"Failed normalizing {input_type}(s) from site '{site_name}' - error: {str(ex)}"
@@ -255,7 +245,7 @@ class OmadaAPI:
def _make_request(self, method: str, endpoint: str, **kwargs: Any) -> Dict[str, Any]: def _make_request(self, method: str, endpoint: str, **kwargs: Any) -> Dict[str, Any]:
time.sleep(1) # Sleep before making any request so it does not rate-limited time.sleep(1) # Sleep before making any request so it does not rate-limited
OmadaHelper.verbose(f"{method} request to endpoint: {endpoint}") OmadaHelper.debug(f"{method} request to endpoint: {endpoint}")
url = f"{getattr(self, 'host')}{endpoint}" url = f"{getattr(self, 'host')}{endpoint}"
headers = self._get_headers(kwargs.pop('include_auth', True)) headers = self._get_headers(kwargs.pop('include_auth', True))
try: try:
@@ -264,37 +254,42 @@ class OmadaAPI:
data = response.json() data = response.json()
response_type = "error" if data.get("errorCode", 0) != 0 else "success" response_type = "error" if data.get("errorCode", 0) != 0 else "success"
msg = f"{method} request completed: {endpoint}" msg = f"{method} request completed: {endpoint}"
OmadaHelper.minimal(msg) OmadaHelper.verbose(msg)
return OmadaHelper.response(response_type, msg, data) return OmadaHelper.response(response_type, msg, data)
except requests.exceptions.RequestException as ex: except requests.exceptions.RequestException as ex:
msg = f"{method} request failed: {str(ex)}" OmadaHelper.minimal(f"{method} request failed: {url}")
OmadaHelper.minimal(f"{method} request to {endpoint} failed") OmadaHelper.verbose(f"{method} request error: {str(ex)}")
OmadaHelper.verbose(msg) return OmadaHelper.response("error", f"{method} request failed to endpoint '{endpoint}' with error: {str(ex)}")
return OmadaHelper.response("error", msg)
def authenticate(self) -> Dict[str, any]: def authenticate(self) -> Dict[str, any]:
"""Make an endpoint request to get access token.""" """Make an endpoint request to get access token."""
OmadaHelper.verbose("Starting authentication process") OmadaHelper.verbose("Starting authentication process")
# Endpoint request
endpoint = "/openapi/authorize/token?grant_type=client_credentials" endpoint = "/openapi/authorize/token?grant_type=client_credentials"
payload = { payload = {
"omadacId": getattr(self, 'omada_id'), "omadacId": getattr(self, 'omada_id'),
"client_id": getattr(self, 'client_id'), "client_id": getattr(self, 'client_id'),
"client_secret": getattr(self, 'client_secret') "client_secret": getattr(self, 'client_secret')
} }
response = self._make_request("POST", endpoint, json=payload, include_auth=False) response = self._make_request("POST", endpoint, json=payload, include_auth=False)
if response["response_type"] == "success":
token_data = response["response_result"]
if token_data.get("errorCode") == 0:
self.access_token = token_data["result"]["accessToken"]
self.refresh_token = token_data["result"]["refreshToken"]
OmadaHelper.minimal("Authentication successful")
return OmadaHelper.response("success", "Authenticated successfully")
OmadaHelper.minimal("Authentication failed") # Successful endpoint response
if response.get("response_type") == "success":
response_result = response.get("response_result")
error_code = response_result.get("errorCode")
access_token = response_result.get("result").get("accessToken")
refresh_token = response_result.get("result").get("refreshToken")
if error_code == 0 and access_token and refresh_token:
self.access_token = access_token
self.refresh_token = refresh_token
msg = "Successfully authenticated"
OmadaHelper.minimal(msg)
return OmadaHelper.response("success", msg)
# Failed authentication
OmadaHelper.debug(f"Authentication response: {response}") OmadaHelper.debug(f"Authentication response: {response}")
return OmadaHelper.response("error", return OmadaHelper.response("error", f"Authentication failed - error: {response.get('response_message', 'Not provided')}")
f"Authentication failed - error: {response.get('response_result').get('msg')}")
def get_clients(self, site_id: str) -> Dict[str, Any]: def get_clients(self, site_id: str) -> Dict[str, Any]:
"""Make an endpoint request to get all online clients on a site.""" """Make an endpoint request to get all online clients on a site."""
@@ -308,52 +303,48 @@ class OmadaAPI:
endpoint = f"/openapi/v1/{getattr(self, 'omada_id')}/sites/{site_id}/devices?page=1&pageSize={getattr(self, 'page_size')}" endpoint = f"/openapi/v1/{getattr(self, 'omada_id')}/sites/{site_id}/devices?page=1&pageSize={getattr(self, 'page_size')}"
return self._make_request("GET", endpoint) return self._make_request("GET", endpoint)
def get_sites(self) -> Dict[str, Any]:
"""Make an endpoint request to populate all accesible sites."""
OmadaHelper.verbose("Retrieving all accessible sites")
endpoint = f"/openapi/v1/{getattr(self, 'omada_id')}/sites?page=1&pageSize={getattr(self, 'page_size')}"
return self._make_request("GET", endpoint)
def populate_sites(self) -> Dict[str, Any]: def populate_sites(self) -> Dict[str, Any]:
"""Make an endpoint request to populate sites.""" """Make an endpoint request to populate all accessible sites."""
try: OmadaHelper.verbose("Starting site population process")
OmadaHelper.verbose("Starting site population process")
# All allowed sites for credentials # Endpoint request
all_sites = self.get_sites()["response_result"].get("result").get("data", []) endpoint = f"/openapi/v1/{getattr(self, 'omada_id')}/sites?page=1&pageSize={getattr(self, 'page_size')}"
OmadaHelper.debug(f"Retrieved {len(all_sites)} sites in total") response = self._make_request("GET", endpoint)
# All available sites # Successful endpoint response
self.available_sites_dict = {site["siteId"]: site["name"] for site in all_sites} if response.get("response_type") == "success":
OmadaHelper.debug(f"Available sites: {self.available_sites_dict}") response_result = response.get("response_result")
if response_result.get("errorCode") == 0:
# All allowed sites for credentials
all_sites = response_result.get("result", "").get("data", [])
OmadaHelper.debug(f"Retrieved {len(all_sites)} sites in total")
# All valid sites from input # All available sites
active_sites_by_id = {site["siteId"]: site["name"] for site in all_sites if self.available_sites_dict = {site["siteId"]: site["name"] for site in all_sites}
site["siteId"] in self.requested_sites()} OmadaHelper.debug(f"Available sites: {self.available_sites_dict}")
active_sites_by_name = {site["siteId"]: site["name"] for site in all_sites if
site["name"] in self.requested_sites()}
self.active_sites_dict = active_sites_by_id | active_sites_by_name
OmadaHelper.debug(f"Active sites after filtering: {self.active_sites_dict}")
# If none of the input sites is valid/accessible, default to the first available site # All valid sites from input
if not self.active_sites_dict: active_sites_by_id = {site["siteId"]: site["name"] for site in all_sites if site["siteId"] in self.requested_sites()}
OmadaHelper.verbose( active_sites_by_name = {site["siteId"]: site["name"] for site in all_sites if site["name"] in self.requested_sites()}
"No valid site requested by configuration options, defaulting to first available site") self.active_sites_dict = active_sites_by_id | active_sites_by_name
first_available_site = next(iter(self.available_sites_dict.items()), (None, None)) OmadaHelper.debug(f"Active sites after filtering: {self.active_sites_dict}")
if first_available_site[0]: # Check if there's an available site
self.active_sites_dict = {first_available_site[0]: first_available_site[1]}
OmadaHelper.debug(f"Using first available site: {first_available_site}")
msg = f"Populated {len(self.active_sites_dict)} active sites" # If none of the input sites is valid/accessible, default to the first available site
OmadaHelper.verbose(msg) if not self.active_sites_dict:
result = OmadaHelper.response("success", msg) OmadaHelper.verbose("No valid site requested by configuration options, defaulting to first available site")
except Exception as ex: first_available_site = next(iter(self.available_sites_dict.items()), (None, None))
OmadaHelper.minimal("Failed to populate sites") if first_available_site[0]: # Check if there's an available site
msg = f"Site population error: {str(ex)}" self.active_sites_dict = {first_available_site[0]: first_available_site[1]}
OmadaHelper.verbose(msg) OmadaHelper.debug(f"Using first available site: {first_available_site}")
result = OmadaHelper.response("error", msg)
return result # Successful site population
msg = f"Successfully populated {len(self.active_sites_dict)} site(s)"
OmadaHelper.minimal(msg)
return OmadaHelper.response("success", msg)
# Failed site population
OmadaHelper.debug(f"Site population response: {response}")
return OmadaHelper.response("error", f"Site population failed - error: {response.get('response_message', 'Not provided')}")
def requested_sites(self) -> list: def requested_sites(self) -> list:
"""Returns sites requested by user.""" """Returns sites requested by user."""
@@ -370,8 +361,7 @@ class OmadaAPI:
def make_entries(plugin_objects: Plugin_Objects, normalized_input_data: dict) -> None: def make_entries(plugin_objects: Plugin_Objects, normalized_input_data: dict) -> None:
if normalized_input_data.get("response_type", "error") != "success": if normalized_input_data.get("response_type", "error") != "success":
OmadaHelper.minimal( OmadaHelper.minimal(f"Unable to make entries - error: {normalized_input_data.get('response_message', 'Not provided')}")
f"Unable to make entries - error: {normalized_input_data.get('response_message', 'Not provided.')}")
return return
response_result = normalized_input_data.get("response_result", {}) response_result = normalized_input_data.get("response_result", {})
@@ -407,19 +397,18 @@ def make_entries(plugin_objects: Plugin_Objects, normalized_input_data: dict) ->
def get_entries(plugin_objects: Plugin_Objects) -> Plugin_Objects: def get_entries(plugin_objects: Plugin_Objects) -> Plugin_Objects:
OmadaHelper.minimal("Starting data collection process")
omada_api = OmadaAPI(OPTIONS) omada_api = OmadaAPI(OPTIONS)
auth_result = omada_api.authenticate() auth_result = omada_api.authenticate()
if auth_result["response_type"] == "error": if auth_result["response_type"] == "error":
OmadaHelper.minimal("Authentication failed, aborting data collection") OmadaHelper.minimal("Authentication failed, aborting data collection")
OmadaHelper.debug(f"Authentication error - {auth_result['response_message']}") OmadaHelper.debug(f"{auth_result['response_message']}")
return plugin_objects return plugin_objects
sites_result = omada_api.populate_sites() sites_result = omada_api.populate_sites()
if sites_result["response_type"] == "error": if sites_result["response_type"] == "error":
OmadaHelper.minimal("Site population failed, aborting data collection") OmadaHelper.minimal("Site population failed, aborting data collection")
OmadaHelper.debug(f"Site population error - {auth_result['response_message']}") OmadaHelper.debug(f"{sites_result['response_message']}")
return plugin_objects return plugin_objects
requested_sites = omada_api.requested_sites() requested_sites = omada_api.requested_sites()
@@ -428,13 +417,15 @@ def get_entries(plugin_objects: Plugin_Objects) -> Plugin_Objects:
OmadaHelper.verbose(f"Requested sites: {requested_sites}") OmadaHelper.verbose(f"Requested sites: {requested_sites}")
OmadaHelper.verbose(f"Available sites: {available_sites}") OmadaHelper.verbose(f"Available sites: {available_sites}")
OmadaHelper.minimal(f"Active sites: {active_sites}") OmadaHelper.verbose(f"Active sites: {active_sites}")
OmadaHelper.minimal("Starting data collection process")
for site_id, site_name in active_sites.items(): for site_id, site_name in active_sites.items():
OmadaHelper.verbose(f"Processing site: {site_name} ({site_id})") OmadaHelper.verbose(f"Processing site: {site_name} ({site_id})")
devices_response = omada_api.get_devices(site_id) devices_response = omada_api.get_devices(site_id)
if devices_response["response_type"] == "error": if devices_response["response_type"] != "success":
OmadaHelper.minimal(f"Failed to retrieve devices for site: {site_name}") OmadaHelper.minimal(f"Failed to retrieve devices for site: {site_name}")
else: else:
devices = devices_response["response_result"].get("result").get("data", []) devices = devices_response["response_result"].get("result").get("data", [])
@@ -443,7 +434,7 @@ def get_entries(plugin_objects: Plugin_Objects) -> Plugin_Objects:
make_entries(plugin_objects, devices) make_entries(plugin_objects, devices)
clients_response = omada_api.get_clients(site_id) clients_response = omada_api.get_clients(site_id)
if clients_response["response_type"] == "error": if clients_response["response_type"] != "success":
OmadaHelper.minimal(f"Failed to retrieve clients for site {site_name}") OmadaHelper.minimal(f"Failed to retrieve clients for site {site_name}")
else: else:
clients = clients_response["response_result"].get("result").get("data", []) clients = clients_response["response_result"].get("result").get("data", [])
@@ -453,7 +444,7 @@ def get_entries(plugin_objects: Plugin_Objects) -> Plugin_Objects:
OmadaHelper.verbose(f"Site complete: {site_name} ({site_id})") OmadaHelper.verbose(f"Site complete: {site_name} ({site_id})")
OmadaHelper.minimal("Data collection process completed") OmadaHelper.minimal("Completed data collection process")
return plugin_objects return plugin_objects