From 8168bb2d5b6d3e0f2163807126a1c543d2290139 Mon Sep 17 00:00:00 2001 From: xfilo Date: Mon, 20 Apr 2026 00:39:29 +0200 Subject: [PATCH 1/3] Fix for https://github.com/netalertx/NetAlertX/issues/1595 - Omada Controller versions >= 6.2.0.0 removed the v1 clients endpoint --- front/plugins/omada_sdn_openapi/script.py | 68 ++++++++++++++++++++++- 1 file changed, 66 insertions(+), 2 deletions(-) diff --git a/front/plugins/omada_sdn_openapi/script.py b/front/plugins/omada_sdn_openapi/script.py index 11b17b60..685413b1 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 @@ -220,6 +221,39 @@ 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: + def to_tuple(v): + if isinstance(v, int): + return (v,) + + if isinstance(v, str): + return tuple(int(x) for x in v.split(".") if x != "") + + raise TypeError("version/base must be int or str") + + v = to_tuple(version) + b = to_tuple(base) + + max_len = max(len(v), len(b)) + v += (0,) * (max_len - len(v)) + b += (0,) * (max_len - len(b)) + + if op == "==": + return v == b + if op == "!=": + return v != b + if op == ">": + return v > b + if op == ">=": + return v >= b + if op == "<": + return v < b + if op == "<=": + return v <= b + + raise ValueError("Unsupported operator") class OmadaAPI: @@ -259,6 +293,7 @@ class OmadaAPI: self.active_sites_dict = {} self.access_token = None self.refresh_token = None + self.controller_version = 0 OmadaHelper.verbose("OmadaAPI initialized") @@ -328,11 +363,36 @@ 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 all online clients on a site.""" + 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") + self.controller_version = response_result.get("result").get("controllerVersion") + OmadaHelper.debug(f"Controller status: {response}") + return OmadaHelper.response("success", "Successfully retrieved controller status") + + OmadaHelper.debug(f"Controller status: {response}") + return OmadaHelper.response("error", "Failed to retrieve controller status") + 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) + + if OmadaHelper.version_check(self.controller_version, "6.2.0.0", ">="): + endpoint = f"/openapi/v2/{getattr(self, 'omada_id')}/sites/{site_id}/clients" + payload = { + "page": 1, + "pageSize": getattr(self, 'page_size'), + "scope": 1 + } + return self._make_request("POST", endpoint, json=payload) + else: + 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) def get_devices(self, site_id: str) -> Dict[str, Any]: """Make an endpoint request to get all online devices on a site.""" @@ -454,6 +514,10 @@ class OmadaData: OmadaHelper.minimal("Authentication failed, aborting data collection") OmadaHelper.debug(f"{auth_result['response_message']}") return plugin_objects + + # Controller status + omada_api.get_controller_status() + OmadaHelper.verbose(f"Controller version: {omada_api.controller_version}") # Populate sites sites_result = omada_api.populate_sites() From 2fd7e5e700a02571f00dbdde8dd74b31433d3eb4 Mon Sep 17 00:00:00 2001 From: xfilo Date: Mon, 20 Apr 2026 12:13:21 +0200 Subject: [PATCH 2/3] Follow-up on PR https://github.com/netalertx/NetAlertX/pull/1622 based on AI review - fallback to v1 clients endpoint if v2 fails. - On failed controller version retrieval, attempt to use the v2 clients endpoint first and automatically fallback to the v1 endpoint if the v2 request fails. - Improves compatibility with controllers where version detection is unreliable or unavailable. --- front/plugins/omada_sdn_openapi/script.py | 62 ++++++++++++++++------- 1 file changed, 43 insertions(+), 19 deletions(-) diff --git a/front/plugins/omada_sdn_openapi/script.py b/front/plugins/omada_sdn_openapi/script.py index 685413b1..57d8c275 100755 --- a/front/plugins/omada_sdn_openapi/script.py +++ b/front/plugins/omada_sdn_openapi/script.py @@ -293,7 +293,7 @@ class OmadaAPI: self.active_sites_dict = {} self.access_token = None self.refresh_token = None - self.controller_version = 0 + self.controller_version = None OmadaHelper.verbose("OmadaAPI initialized") @@ -364,35 +364,56 @@ class OmadaAPI: 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 all online clients on a site.""" + """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") - self.controller_version = response_result.get("result").get("controllerVersion") - OmadaHelper.debug(f"Controller status: {response}") - return OmadaHelper.response("success", "Successfully retrieved controller status") + 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 retrieve controller status") + 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}") - if OmadaHelper.version_check(self.controller_version, "6.2.0.0", ">="): - endpoint = f"/openapi/v2/{getattr(self, 'omada_id')}/sites/{site_id}/clients" - payload = { - "page": 1, - "pageSize": getattr(self, 'page_size'), - "scope": 1 - } - return self._make_request("POST", endpoint, json=payload) - else: - endpoint = f"/openapi/v1/{getattr(self, 'omada_id')}/sites/{site_id}/clients?page=1&pageSize={getattr(self, 'page_size')}" + 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.""" @@ -516,8 +537,11 @@ class OmadaData: return plugin_objects # Controller status - omada_api.get_controller_status() - OmadaHelper.verbose(f"Controller version: {omada_api.controller_version}") + 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() From fb7dab488140ad1d4612f11503dcb58db6abb1cd Mon Sep 17 00:00:00 2001 From: xfilo Date: Mon, 20 Apr 2026 12:53:51 +0200 Subject: [PATCH 3/3] Replace custom version_check with packaging.version.Version (Follow-up based on AI review) - Use PEP 440-compliant parsing and comparison - Fix issues with non-numeric segments (e.g. "beta", "rc") - Add safe fallback for invalid version strings --- front/plugins/omada_sdn_openapi/script.py | 53 ++++++++++++----------- 1 file changed, 28 insertions(+), 25 deletions(-) diff --git a/front/plugins/omada_sdn_openapi/script.py b/front/plugins/omada_sdn_openapi/script.py index 57d8c275..9e8bb749 100755 --- a/front/plugins/omada_sdn_openapi/script.py +++ b/front/plugins/omada_sdn_openapi/script.py @@ -27,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') @@ -224,36 +226,37 @@ class OmadaHelper: @staticmethod def version_check(version, base: str, op: str = ">=") -> bool: - def to_tuple(v): + """ + 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 (v,) - + return Version(str(v)) if isinstance(v, str): - return tuple(int(x) for x in v.split(".") if x != "") - + 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_tuple(version) - b = to_tuple(base) + v = to_version(version) + b = to_version(base) - max_len = max(len(v), len(b)) - v += (0,) * (max_len - len(v)) - b += (0,) * (max_len - len(b)) - - if op == "==": - return v == b - if op == "!=": - return v != b - if op == ">": - return v > b - if op == ">=": - return v >= b - if op == "<": - return v < b - if op == "<=": - return v <= b - - raise ValueError("Unsupported operator") + return ops[op](v, b) class OmadaAPI: