diff --git a/front/plugins/omada_sdn_openapi/script.py b/front/plugins/omada_sdn_openapi/script.py index 11b17b60..9e8bb749 100755 --- a/front/plugins/omada_sdn_openapi/script.py +++ b/front/plugins/omada_sdn_openapi/script.py @@ -19,6 +19,7 @@ __author__ = "xfilo" __version__ = 0.1 # Initial version __version__ = 0.2 # Rephrased error messages, improved logging and code logic __version__ = 0.3 # Refactored data collection into a class, improved code clarity with comments +__version__ = 0.4 # Fix for https://github.com/netalertx/NetAlertX/issues/1595 - Omada Controller versions >= 6.2.0.0 removed the v1 clients endpoint import os import sys @@ -26,9 +27,11 @@ import urllib3 import requests import time import pytz +import operator from datetime import datetime from typing import Literal, Any, Dict +from packaging.version import Version, InvalidVersion # Define the installation path and extend the system path for plugin imports INSTALL_PATH = os.getenv('NETALERTX_APP', '/app') @@ -220,6 +223,40 @@ class OmadaHelper: msg = f"Failed normalizing {input_type}(s) from site '{site_name}' - error: {str(ex)}" OmadaHelper.verbose(msg) return OmadaHelper.response("error", msg) + + @staticmethod + def version_check(version, base: str, op: str = ">=") -> bool: + """ + Compare versions using PEP 440 semantics. + Supports int and str inputs. + """ + ops = { + "==": operator.eq, + "!=": operator.ne, + ">": operator.gt, + ">=": operator.ge, + "<": operator.lt, + "<=": operator.le, + } + + if op not in ops: + raise ValueError("Unsupported operator") + + def to_version(v): + if isinstance(v, int): + return Version(str(v)) + if isinstance(v, str): + try: + return Version(v) + except InvalidVersion: + # fallback: treat invalid versions as 0 + return Version("0") + raise TypeError("version/base must be int or str") + + v = to_version(version) + b = to_version(base) + + return ops[op](v, b) class OmadaAPI: @@ -259,6 +296,7 @@ class OmadaAPI: self.active_sites_dict = {} self.access_token = None self.refresh_token = None + self.controller_version = None OmadaHelper.verbose("OmadaAPI initialized") @@ -328,11 +366,57 @@ class OmadaAPI: OmadaHelper.debug(f"Authentication response: {response}") return OmadaHelper.response("error", f"Authentication failed - error: {response.get('response_message', 'Not provided')}") + def get_controller_status(self) -> Dict[str, Any]: + """Make an endpoint request to get controller status.""" + OmadaHelper.verbose(f"Retrieving controller status for CID: {getattr(self, 'omada_id')}") + endpoint = f"/openapi/v1/{getattr(self, 'omada_id')}/system/setting/controller-status" + response = self._make_request("GET", endpoint) + + if response.get("response_type") == "success": + response_result = response.get("response_result") or {} + result = response_result.get("result") or {} + self.controller_version = result.get("controllerVersion") + if not self.controller_version: + self.controller_version = None + OmadaHelper.debug(f"Controller status: {response}") + return OmadaHelper.response("error", "Controller status response did not include controllerVersion") + else: + return OmadaHelper.response("success", "Successfully retrieved controller status") + + OmadaHelper.debug(f"Controller status: {response}") + return OmadaHelper.response("error", "Failed to call controller status endpoint") + def get_clients(self, site_id: str) -> Dict[str, Any]: """Make an endpoint request to get all online clients on a site.""" OmadaHelper.verbose(f"Retrieving clients for site: {site_id}") - endpoint = f"/openapi/v1/{getattr(self, 'omada_id')}/sites/{site_id}/clients?page=1&pageSize={getattr(self, 'page_size')}" - return self._make_request("GET", endpoint) + + page_size = getattr(self, 'page_size') + omada_id = getattr(self, 'omada_id') + + def call_v2(): + endpoint = f"/openapi/v2/{omada_id}/sites/{site_id}/clients" + payload = { + "page": 1, + "pageSize": page_size, + "scope": 1 + } + return self._make_request("POST", endpoint, json=payload) + + def call_v1(): + endpoint = f"/openapi/v1/{omada_id}/sites/{site_id}/clients?page=1&pageSize={page_size}" + return self._make_request("GET", endpoint) + + if self.controller_version is None: + OmadaHelper.verbose("Controller version unknown, trying v2 then v1") + resp = call_v2() + if resp and resp.get("response_type") == "success": + return resp + return call_v1() + + if OmadaHelper.version_check(self.controller_version, "6.2.0.0", ">="): + return call_v2() + + return call_v1() def get_devices(self, site_id: str) -> Dict[str, Any]: """Make an endpoint request to get all online devices on a site.""" @@ -454,6 +538,13 @@ class OmadaData: OmadaHelper.minimal("Authentication failed, aborting data collection") OmadaHelper.debug(f"{auth_result['response_message']}") return plugin_objects + + # Controller status + status_result = omada_api.get_controller_status() + if status_result["response_type"] == "error": + OmadaHelper.verbose(f"Controller version lookup failed: {status_result['response_message']}") + else: + OmadaHelper.verbose(f"Controller version: {omada_api.controller_version}") # Populate sites sites_result = omada_api.populate_sites()