Refactor: deduplicate Clients SSO logic and fix _relay_base edge cases

- Extract shared RemoteClients base class to eliminate duplicated
  auto-SSO login logic between edge.py and drive.py Clients classes
- Fix potential bug in _relay_base: use startswith() instead of
  substring `in` check for device DNS name matching
- Fix trailing-slash edge case in _relay_base URL construction
- Add comprehensive unit tests for _relay_base (DNS fallback,
  hostname derivation, port handling, trailing slash)
- Extract common test setup into _setup_remote_device_with_sso helper

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: asafn-ctera

cterasdk/core/remote.py

@@ -15,12 +15,13 @@ def _relay_base(Portal, device):
     Host header while still connecting to the same portal endpoint.
     """
     device_dns = getattr(device, 'deviceDnsName', None)
-    if device_dns and device.name in device_dns:
-        portal_hostname = device_dns[len(device.name) + 1:]   # strip "vGateway-7192."
+    if device_dns and device_dns.startswith(f'{device.name}.'):
+        portal_hostname = device_dns[len(device.name) + 1:]
         parsed = urlparse(Portal.ctera.baseurl)
         port = f':{parsed.port}' if parsed.port not in (None, 80, 443) else ''
-        return f'{parsed.scheme}://{portal_hostname}{port}{parsed.path}/devices/{device.name}'
-    return f'{Portal.ctera.baseurl}/devices/{device.name}'
+        base_path = parsed.path.rstrip('/')
+        return f'{parsed.scheme}://{portal_hostname}{port}{base_path}/devices/{device.name}'
+    return f'{Portal.ctera.baseurl.rstrip("/")}/devices/{device.name}'
 
 
 def remote_command(Portal, device):

cterasdk/objects/synchronous/drive.py

@@ -1,42 +1,23 @@
-import logging
-
 import cterasdk.settings
 from ...clients import clients
 from ..services import Management
 from ..endpoints import EndpointBuilder
-from ...common import parse_base_object_ref
 from ...lib.session.edge import Session
 from ...edge import backup, cli, logs, services, support, sync
+from .remote_clients import RemoteClients
 
 
-logger = logging.getLogger('cterasdk.drive')
-
-
-class Clients:
+class Clients(RemoteClients):
 
     def __init__(self, drive, Portal):
-        self._drive = drive
-        self._Portal = Portal
-        self._authenticated = False
         if Portal:
             drive._Portal = Portal
             drive.default.close()
             drive._ctera_session.start_remote_session(Portal.session())
-            self._api = Portal.default.clone(clients.API, EndpointBuilder.new(drive.base, '/admingui/api'), authenticator=lambda *_: True)
+            api_client = Portal.default.clone(clients.API, EndpointBuilder.new(drive.base, '/admingui/api'), authenticator=lambda *_: True)
         else:
-            self._api = drive.default.clone(clients.API, EndpointBuilder.new(drive.base, '/admingui/api'))
-
-    @property
-    def api(self):
-        if self._Portal and not self._authenticated:
-            tenant = parse_base_object_ref(self._drive.portal).name
-            device_name = self._drive.name
-            logger.debug('Auto-SSO login via relay channel. %s', {'tenant': tenant, 'device': device_name})
-            token = self._Portal.api.execute(f'/portals/{tenant}/devices/{device_name}', 'singleSignOn')
-            if token:
-                self._api.get('/ssologin', params={'ticket': token})
-            self._authenticated = True
-        return self._api
+            api_client = drive.default.clone(clients.API, EndpointBuilder.new(drive.base, '/admingui/api'))
+        super().__init__(drive, Portal, api_client)
 
 
 class Drive(Management):

cterasdk/objects/synchronous/edge.py

@@ -1,12 +1,11 @@
-import logging
-
 import cterasdk.settings
 from ...clients import clients
 from ..services import Management
 from ..endpoints import EndpointBuilder
 from .. import authenticators
-from ...common import modules, parse_base_object_ref
+from ...common import modules
 from ...lib.session.edge import Session
+from .remote_clients import RemoteClients
 
 
 from ...edge import (
@@ -18,36 +17,19 @@
 )
 
 
-logger = logging.getLogger('cterasdk.edge')
-
-
-class Clients:
+class Clients(RemoteClients):
 
     def __init__(self, edge, Portal):
-        self._edge = edge
-        self._Portal = Portal
-        self._authenticated = False
         if Portal:
             edge._Portal = Portal
             edge.default.close()
             edge._ctera_session.start_remote_session(Portal.session())
-            self._api = Portal.default.clone(clients.API, EndpointBuilder.new(edge.base, '/admingui/api'), authenticator=lambda *_: True)
+            api_client = Portal.default.clone(clients.API, EndpointBuilder.new(edge.base, '/admingui/api'), authenticator=lambda *_: True)
         else:
             self.migrate = edge.default.clone(clients.Migrate, EndpointBuilder.new(edge.base, '/migration/rest/v1'))
-            self._api = edge.default.clone(clients.API, EndpointBuilder.new(edge.base, '/admingui/api'))
+            api_client = edge.default.clone(clients.API, EndpointBuilder.new(edge.base, '/admingui/api'))
             self.io = IO(edge)
-
-    @property
-    def api(self):
-        if self._Portal and not self._authenticated:
-            tenant = parse_base_object_ref(self._edge.portal).name
-            device_name = self._edge.name
-            logger.debug('Auto-SSO login via relay channel. %s', {'tenant': tenant, 'device': device_name})
-            token = self._Portal.api.execute(f'/portals/{tenant}/devices/{device_name}', 'singleSignOn')
-            if token:
-                self._api.get('/ssologin', params={'ticket': token})
-            self._authenticated = True
-        return self._api
+        super().__init__(edge, Portal, api_client)
 
 
 class IO:

cterasdk/objects/synchronous/remote_clients.py

@@ -0,0 +1,32 @@
+import logging
+
+from ...common import parse_base_object_ref
+
+
+logger = logging.getLogger('cterasdk.remote')
+
+
+class RemoteClients:
+    """
+    Base class for remote device client management with lazy SSO authentication.
+
+    Subclasses must set ``self._device`` before calling ``super().__init__``.
+    """
+
+    def __init__(self, device, Portal, api_client):
+        self._device = device
+        self._Portal = Portal
+        self._authenticated = False
+        self._api = api_client
+
+    @property
+    def api(self):
+        if self._Portal and not self._authenticated:
+            tenant = parse_base_object_ref(self._device.portal).name
+            device_name = self._device.name
+            logger.debug('Auto-SSO login via relay channel. %s', {'tenant': tenant, 'device': device_name})
+            token = self._Portal.api.execute(f'/portals/{tenant}/devices/{device_name}', 'singleSignOn')
+            if token:
+                self._api.get('/ssologin', params={'ticket': token})
+            self._authenticated = True
+        return self._api

tests/ut/core/admin/test_relay_base.py

@@ -0,0 +1,81 @@
+import unittest
+from unittest import mock
+
+from cterasdk.core.remote import _relay_base
+
+
+class TestRelayBase(unittest.TestCase):
+
+    def _make_portal(self, baseurl):
+        portal = mock.MagicMock()
+        portal.ctera.baseurl = baseurl
+        return portal
+
+    def _make_device(self, name, device_dns_name=None):
+        device = mock.MagicMock(spec=['name', 'deviceDnsName'] if device_dns_name is not None else ['name'])
+        device.name = name
+        if device_dns_name is not None:
+            device.deviceDnsName = device_dns_name
+        return device
+
+    def test_fallback_when_device_dns_is_none(self):
+        portal = self._make_portal('https://portal.ctera.me')
+        device = self._make_device('vGateway-7192')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'https://portal.ctera.me/devices/vGateway-7192')
+
+    def test_fallback_when_device_dns_does_not_start_with_name(self):
+        portal = self._make_portal('https://portal.ctera.me')
+        device = self._make_device('vGateway-7192', 'other-device.portal.ctera.me')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'https://portal.ctera.me/devices/vGateway-7192')
+
+    def test_hostname_derivation_from_dns_name(self):
+        portal = self._make_portal('https://10.0.0.1')
+        device = self._make_device('vGateway-7192', 'vGateway-7192.portal.ctera.me')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'https://portal.ctera.me/devices/vGateway-7192')
+
+    def test_non_standard_port_preserved(self):
+        portal = self._make_portal('https://10.0.0.1:8443')
+        device = self._make_device('vGateway-7192', 'vGateway-7192.portal.ctera.me')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'https://portal.ctera.me:8443/devices/vGateway-7192')
+
+    def test_standard_https_port_omitted(self):
+        portal = self._make_portal('https://10.0.0.1:443')
+        device = self._make_device('vGateway-7192', 'vGateway-7192.portal.ctera.me')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'https://portal.ctera.me/devices/vGateway-7192')
+
+    def test_standard_http_port_omitted(self):
+        portal = self._make_portal('http://10.0.0.1:80')
+        device = self._make_device('vGateway-7192', 'vGateway-7192.portal.ctera.me')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'http://portal.ctera.me/devices/vGateway-7192')
+
+    def test_trailing_slash_in_baseurl_no_double_slash(self):
+        portal = self._make_portal('https://portal.ctera.me/')
+        device = self._make_device('vGateway-7192', 'vGateway-7192.portal.ctera.me')
+        result = _relay_base(portal, device)
+        self.assertNotIn('//', result.split('://')[1])
+        self.assertEqual(result, 'https://portal.ctera.me/devices/vGateway-7192')
+
+    def test_baseurl_with_path(self):
+        portal = self._make_portal('https://10.0.0.1/api/v1')
+        device = self._make_device('vGateway-7192', 'vGateway-7192.portal.ctera.me')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'https://portal.ctera.me/api/v1/devices/vGateway-7192')
+
+    def test_substring_device_name_does_not_false_match(self):
+        """A device named 'gw' should NOT match dns 'other-gw.portal.ctera.me'."""
+        portal = self._make_portal('https://portal.ctera.me')
+        device = self._make_device('gw', 'other-gw.portal.ctera.me')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'https://portal.ctera.me/devices/gw')
+
+    def test_fallback_strips_trailing_slash(self):
+        portal = self._make_portal('https://portal.ctera.me/')
+        device = self._make_device('vGateway-7192')
+        result = _relay_base(portal, device)
+        self.assertEqual(result, 'https://portal.ctera.me/devices/vGateway-7192')

tests/ut/core/admin/test_remote.py

@@ -73,8 +73,8 @@ def _create_device_param(name, portal, device_type, remote_access_url):
         param.remoteAccessUrl = remote_access_url
         return param
 
-    def test_auto_sso_on_first_api_access(self):
-        """Verify that first access to edge.api triggers SSO login via relay channel."""
+    def _setup_remote_device_with_sso(self):
+        """Common setup for SSO auto-login tests. Returns the remote device with a mocked _api."""
         remote_session = self.patch_call("cterasdk.lib.session.edge.Session.start_remote_session")
         remote_session.return_value = munch.Munch({'account': munch.Munch({'name': 'mickey', 'tenant': 'tenant'})})
         get_multi_response = TestCoreRemote._create_device_param(self._device_name, self._device_portal,
@@ -83,21 +83,19 @@ def test_auto_sso_on_first_api_access(self):
         self._activate_portal_session()
         device = devices.Devices(self._global_admin).device(self._device_name)
         device._ctera_clients._api = mock.MagicMock()
+        return device
+
+    def test_auto_sso_on_first_api_access(self):
+        """Verify that first access to edge.api triggers SSO login via relay channel."""
+        device = self._setup_remote_device_with_sso()
         _ = device.api
         self._global_admin.api.execute.assert_called_once_with(
             f'/portals/{self._tenant_name}/devices/{self._device_name}', 'singleSignOn')
         device._ctera_clients._api.get.assert_called_once_with('/ssologin', params={'ticket': self._sso_ticket})
 
     def test_auto_sso_not_repeated_on_subsequent_api_access(self):
         """Verify that subsequent api accesses do not re-trigger SSO."""
-        remote_session = self.patch_call("cterasdk.lib.session.edge.Session.start_remote_session")
-        remote_session.return_value = munch.Munch({'account': munch.Munch({'name': 'mickey', 'tenant': 'tenant'})})
-        get_multi_response = TestCoreRemote._create_device_param(self._device_name, self._device_portal,
-                                                                 'vGateway', self._device_remote_access_url)
-        self._init_global_admin(get_multi_response=get_multi_response, execute_response=self._sso_ticket)
-        self._activate_portal_session()
-        device = devices.Devices(self._global_admin).device(self._device_name)
-        device._ctera_clients._api = mock.MagicMock()
+        device = self._setup_remote_device_with_sso()
         _ = device.api
         _ = device.api
         _ = device.api