Added logic for dynamic netconf forwarding

connector/docs/changelog/undistributed/changelog_dynamic_netconf_port_20261007080051.rst

@@ -0,0 +1,6 @@
+--------------------------------------------------------------------------------
+New
+--------------------------------------------------------------------------------
+* NETCONF Connector
+    * Modified Netconf:
+        * Validate SSH tunnel targets against the runtime endpoint, replace stale tunnels, preserve configured bind addresses, and clean up tunnels created by the connector

connector/src/yang/connector/netconf.py

@@ -349,6 +349,10 @@ def __init__(self, *args, **kwargs):
             timeout=self.timeout)
 
         self.active_notifications = {}
+        self._tunnel_owned = False
+        self._tunnel_port = None
+        self._tunnel_proxy_host = None
+        self._tunnel_type = None
 
     @property
     def session(self):
@@ -461,6 +465,148 @@ def configure_session_logging(self):
         self.session.logger = NetconfSessionLoggerAdapter(
             logger, extra, self.log)
 
+    def _connection_block(self):
+        """Return the pyATS connection profile used by this connector."""
+        connections = getattr(self.device, 'connections', {})
+        try:
+            return connections[self.via]
+        except (KeyError, TypeError):
+            return {}
+
+    def _tunnel_config(self):
+        """Return this connection's Unicon SSH-tunnel configuration."""
+        connection = self._connection_block()
+        return connection.get('sshtunnel') if hasattr(connection, 'get') else {}
+
+    def _default_tunnel_config(self):
+        connections = getattr(self.device, 'connections', {})
+        defaults = connections.get('defaults', {}) if hasattr(connections, 'get') else {}
+        return defaults.get('sshtunnel') or {} if hasattr(defaults, 'get') else {}
+
+    def _add_netconf_tunnel(self, sshtunnel, target_address, target_port):
+        """Reuse only a tunnel known to forward to this NETCONF endpoint."""
+        sshtunnel.lock.acquire()
+        try:
+            return self._add_netconf_tunnel_locked(
+                sshtunnel, target_address, target_port)
+        finally:
+            sshtunnel.lock.release()
+
+    def _add_netconf_tunnel_locked(self, sshtunnel, target_address, target_port):
+        connection = self._connection_block()
+        tunnel_config = self._tunnel_config()
+        default_tunnel_config = self._default_tunnel_config()
+        if tunnel_config is None:
+            tunnel_config = dict(default_tunnel_config)
+            connection['sshtunnel'] = tunnel_config
+        if hasattr(tunnel_config, 'setdefault'):
+            for key, value in default_tunnel_config.items():
+                tunnel_config.setdefault(key, value)
+        proxy_host = (tunnel_config.get('host')
+                      or default_tunnel_config.get('host'))
+        if not proxy_host:
+            raise RuntimeError('Cannot add NETCONF SSH tunnel: proxy host is not configured')
+
+        target_matches = (
+            tunnel_config.get('target_address') == target_address
+            and str(tunnel_config.get('target_port')) == str(target_port)
+        )
+        tunnel_port = tunnel_config.get('tunnel_port')
+        tunnel_type = tunnel_config.get('tunnel_type', 'local')
+        if 'tunnel_bind_ip' in tunnel_config:
+            tunnel_bind_ip = tunnel_config.get('tunnel_bind_ip')
+        else:
+            tunnel_bind_ip = tunnel_config.get('tunnel_ip')
+        tunnel_ip = ('127.0.0.1' if tunnel_bind_ip is None
+                     else tunnel_bind_ip)
+
+        if tunnel_config.get('connected') and target_matches and tunnel_port:
+            return tunnel_ip, tunnel_port
+
+        # A connected flag alone does not identify the forwarding target.
+        # Recreate unknown or mismatched tunnels so NETCONF cannot silently
+        # reach the previous target.
+        sshtunnel.connect_proxy_host(
+            testbed=self.device.testbed, proxy_host=proxy_host)
+        proxy_connection = sshtunnel.proxy_connections.get(proxy_host)
+        if proxy_connection is None:
+            raise RuntimeError(
+                "Cannot add NETCONF SSH tunnel: proxy {!r} is not connected".format(
+                    proxy_host)
+            )
+
+        if tunnel_config.get('connected'):
+            if not tunnel_port:
+                raise RuntimeError(
+                    'Cannot replace NETCONF SSH tunnel without its local port')
+            sshtunnel.remove_tunnel(
+                proxy_conn=proxy_connection,
+                tunnel_type=tunnel_type,
+                tunnel_port=tunnel_port,
+                tunnel_ip=tunnel_bind_ip,
+            )
+            tunnel_config['connected'] = False
+
+        tunnel_port = sshtunnel.add_tunnel(
+            proxy_conn=proxy_connection,
+            tunnel_port=tunnel_port or 0,
+            tunnel_type=tunnel_type,
+            tunnel_protocol=tunnel_config.get('tunnel_protocol', 'tcp'),
+            target_address=target_address,
+            target_port=target_port,
+            tunnel_ip=tunnel_bind_ip,
+        )
+        tunnel_config.update({
+            'connected': True,
+            'tunnel_ip': tunnel_ip,
+            'tunnel_bind_ip': tunnel_bind_ip,
+            'tunnel_port': tunnel_port,
+            'tunnel_type': tunnel_type,
+            'target_address': target_address,
+            'target_port': target_port,
+        })
+        self._tunnel_owned = True
+        self._tunnel_port = tunnel_port
+        self._tunnel_proxy_host = proxy_host
+        self._tunnel_type = tunnel_type
+        return tunnel_ip, tunnel_port
+
+    def _cleanup_owned_tunnel(self):
+        """Remove only a tunnel created by this NETCONF connection."""
+        if not self._tunnel_owned:
+            return
+
+        from unicon.sshutils import sshtunnel
+
+        tunnel_config = self._tunnel_config()
+        proxy_host = (self._tunnel_proxy_host
+                      or tunnel_config.get('host')
+                      or self._default_tunnel_config().get('host'))
+        proxy_connection = sshtunnel.proxy_connections.get(proxy_host)
+        if proxy_connection is None:
+            raise RuntimeError(
+                "Cannot remove NETCONF SSH tunnel: proxy {!r} is not connected".format(
+                    proxy_host)
+            )
+
+        tunnel_bind_ip = (
+            tunnel_config.get('tunnel_bind_ip')
+            if 'tunnel_bind_ip' in tunnel_config
+            else tunnel_config.get('tunnel_ip')
+        )
+        sshtunnel.remove_tunnel(
+            proxy_conn=proxy_connection,
+            tunnel_type=self._tunnel_type or tunnel_config.get('tunnel_type', 'local'),
+            tunnel_port=self._tunnel_port or tunnel_config.get('tunnel_port'),
+            tunnel_ip=tunnel_bind_ip,
+        )
+        if hasattr(tunnel_config, '__setitem__'):
+            tunnel_config['connected'] = False
+        self._tunnel_owned = False
+        self._tunnel_port = None
+        self._tunnel_proxy_host = None
+        self._tunnel_type = None
+
     def connect(self):
         '''connect
 
@@ -589,7 +735,8 @@ def connect(self):
 
         # remove items
         disregards = ['class', 'model', 'protocol',
-                      'async_mode', 'raise_mode', 'credentials']
+                      'async_mode', 'raise_mode', 'credentials',
+                      'tunnel_target', 'qemu_usernet']
         defaults = {k: v for k, v in defaults.items() if k not in disregards}
 
         # rename ip -> host, cast to str type
@@ -613,7 +760,13 @@ def connect(self):
             except Exception:
                 pass
 
-        # support sshtunnel
+        # Preserve the API-selected remote endpoint before replacing it with
+        # the local end of an optional SSH tunnel.
+        effective_port = getattr(self, 'port', defaults.get('port', 830))
+
+        # Support an optional SSH tunnel. `effective_port` is the remote
+        # service/forward port; ncclient receives only the local tunnel port.
+        tunnel_endpoint = None
         if 'sshtunnel' in defaults:
             from unicon.sshutils import sshtunnel
             tunnel_logger = logging.getLogger('unicon.sshutils')
@@ -623,11 +776,24 @@ def connect(self):
             if tunnel_logger.getEffectiveLevel() > logging.INFO:
                 tunnel_logger.setLevel(logging.INFO)
             try:
-                tunnel_port = sshtunnel.auto_tunnel_add(self.device, self.via)
+                tunnel_target = self.connection_info.get('tunnel_target') or {}
+                connection = self._connection_block()
+                target_address = (
+                    tunnel_target.get('address')
+                    if hasattr(tunnel_target, 'get') else None
+                ) or connection.get('ip') or connection.get('host')
+                if target_address is None:
+                    raise AttributeError('NETCONF SSH tunnel target address is not configured')
+                target_address = str(target_address)
+                if ':' in target_address and not target_address.startswith('['):
+                    target_address = '[{}]'.format(target_address)
+
+                tunnel_ip, tunnel_port = self._add_netconf_tunnel(
+                    sshtunnel, target_address, effective_port)
                 if tunnel_port:
-                    defaults['host'] = self.device.connections[self.via] \
-                                           .sshtunnel.tunnel_ip
+                    defaults['host'] = tunnel_ip
                     defaults['port'] = tunnel_port
+                    tunnel_endpoint = (defaults['host'], tunnel_port)
             except AttributeError as err:
                 raise AttributeError("Cannot add ssh tunnel. \
                 Connection %s may not have ip/host or port.\n%s"
@@ -638,27 +804,50 @@ def connect(self):
             del defaults['sshtunnel']
 
         defaults = {k: getattr(self, k, v) for k, v in defaults.items()}
+        if tunnel_endpoint:
+            # Runtime connection attributes contain the remote effective port;
+            # keep the local endpoint returned by Unicon for ncclient.
+            defaults['host'], defaults['port'] = tunnel_endpoint
 
         try:
             self.session.connect(**defaults)
             self.log.info('NETCONF CONNECTED')
         except Exception:
             if self.session.transport:
                 self.session.close()
+            try:
+                self._cleanup_owned_tunnel()
+            except Exception:
+                self.log.exception('Failed to roll back NETCONF SSH tunnel')
             raise
 
         @atexit.register
         def cleanup():
-            if self.session.transport:
-                self.session.close()
+            try:
+                if self.session.transport:
+                    self.session.close()
+            except Exception:
+                self.log.exception('Failed to close NETCONF session at exit')
+            try:
+                self._cleanup_owned_tunnel()
+            except Exception:
+                self.log.exception('Failed to remove NETCONF SSH tunnel at exit')
 
     def disconnect(self):
         '''disconnect
 
         High-level api: closes the NetConf connection.
         '''
 
-        self.session.close()
+        try:
+            self.session.close()
+        except Exception:
+            try:
+                self._cleanup_owned_tunnel()
+            except Exception:
+                self.log.exception('Failed to remove NETCONF SSH tunnel after disconnect error')
+            raise
+        self._cleanup_owned_tunnel()
 
     def subscribe(self, request):
         """ Creates a notification listener and mark it as active """