Merge pull request #1622 from xfilo/main addressing #1595

OMDSDNOPENAPI - Handle removal of v1 clients endpoint in Omada Controller >= 6.2.0.0

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()