Start wrapping asyncssh

For auto discovery, banner extraction, and interactive ssh, refactor to commen sshclient class.

This hooks the pubkeys.ssh in a manner compatible with confluent 3.x

confluent_server/confluent/discovery/core.py

@@ -62,7 +62,7 @@
 #     - Apply defined configuration to endpoint
 
 import asyncio
-import asyncssh
+import confluent.sshclient as sshclient
 import base64
 import confluent.config.configmanager as cfm
 import confluent.collective.manager as collective
@@ -225,53 +225,6 @@ unknown_info = {}
 pending_nodes = {}
 pending_by_uuid = {}
 
-
-
-class CancelSsh(Exception):
-    pass
-
-class MyClient(asyncssh.SSHClient):
-    def validate_host_public_key(self, host, addr, port, key):
-        #print(repr(key))
-        return True
-
-    def auth_banner_received(self, msg, lang):
-        if hasattr(self, 'confluent_custom_ctx'):
-            self.confluent_custom_ctx['banner'] = msg
-
-    def password_auth_requested(self):
-        raise CancelSsh("noauth")
-
-    def password_change_requested(self, prompt, lang):
-        print(repr(prompt))
-        print(repr(lang))
-
-    def password_change_failed(self):
-        print("pcf")
-
-    def password_changed(self):
-        print("pc")
-
-    def confluent_set_context(self, ctx):
-        self.confluent_custom_ctx = ctx
-
-async def get_ssh_banner(target):
-    mycontext = {}
-    def make_client():
-        client = MyClient()
-        client.confluent_set_context(mycontext)
-        return client
-    sco = asyncssh.SSHClientConnectionOptions(client_factory=make_client, x509_trusted_cert_paths=None, known_hosts=None)
-    try:
-        async with asyncssh.connect(target, options=sco):
-            pass
-    except CancelSsh:
-        pass
-    return mycontext.get('banner')
-
-
-
-
 def register_affluent(affluenthdl):
     global affluent
     affluent = affluenthdl
@@ -1804,7 +1757,7 @@ async def generic_eval(address, hwaddr):
         peerdata['services'] = ['generic-https']
     if 22 in ports:
         sockaddr = (sockaddr[0], 22) + tuple(sockaddr[2:])
-        banner = await get_ssh_banner(address)
+        banner = await sshclient.get_ssh_banner(address)
         if 'addresses' not in peerdata:
             peerdata['addresses'] = [sockaddr]
         if banner and banner.strip() == 'NVOS switch':

confluent_server/confluent/plugins/shell/ssh.py

@@ -25,7 +25,7 @@ import confluent.tasks as tasks
 
 import sys
 sys.modules['gssapi'] = None
-import asyncssh
+import confluent.sshclient as sshclient
 
 
 
@@ -81,18 +81,10 @@ class SshShell(conapi.Console):
         tasks.spawn(self.do_logon())
     
     async def do_logon(self):
-        sco = asyncssh.SSHClientConnectionOptions()
-        #The below would be to support the confluent db, and only fallback if the SSH CA do not work
-        # have to catch the valueerror and use ssh-keyscan to trigger this, asyncssh host key handling
-        # is a bit more limited compared to paramiko
-
-        #but... leverage /etc/ssh/ssh_known_hosts, we can try that way, and if it fails, fallback to our
-        #confluent db based handler
-        #sco.client_fatory = SSHKnownHostsLookup
         try:
             await self.datacallback('\r\nConnecting to {}...'.format(self.node))
             try:
-                self.ssh = await asyncssh.connect(self.node, username=self.username.decode(), password=self.password.decode(), known_hosts='/etc/ssh/ssh_known_hosts')
+                self.ssh = await sshclient.connect(self.node, username=self.username.decode(), password=self.password.decode(), configmanager=self.nodeconfig, nodename=self.node)
             except ValueError:
                 #TODO: non-cert ssh targets
                 raise

confluent_server/confluent/sshclient.py

@@ -0,0 +1,131 @@
+#!/usr/bin/python3
+
+# Copyright 2026 Lenovo
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import asyncssh
+import hashlib
+
+_hashes = {'sha256': hashlib.sha256, 'sha384': hashlib.sha384, 'sha512': hashlib.sha512}
+
+class _CancelSsh(Exception):
+    pass
+
+class _MyClient(asyncssh.SSHClient):
+    def set_keyattrib(self, keyattrib):
+        self.confluent_keyattrib = keyattrib
+
+    def set_nodename(self, nodename):
+        self.confluent_nodename = nodename
+
+    def set_configmanager(self, configmanager):
+        self.confluent_configmanager = configmanager
+
+    def validate_host_public_key(self, host, addr, port, key):
+        if hasattr(self, 'confluent_validate_hostkey'):
+            if not self.confluent_validate_hostkey:
+                return True
+        if not hasattr(self, 'confluent_nodename') or not hasattr(self, 'confluent_configmanager'):
+            return False
+        cfg = self.confluent_configmanager
+        nodename = self.confluent_nodename
+        cfi = cfg.get_node_attributes(nodename, [self.confluent_keyattrib, 'pubkeys.addpolicy'])
+        fprint = cfi.get(nodename, {}).get(self.confluent_keyattrib, {}).get('value', None)
+        policy = cfi.get(nodename, {}).get('pubkeys.addpolicy', {}).get('value', 'tofu')
+
+        if fprint:
+            algo, expectedfingerprint = fprint.split('$', 1)
+            if algo not in _hashes:
+                return False
+            keyfingerprint = _hashes[algo](key.public_data).hexdigest()
+            if keyfingerprint == expectedfingerprint:
+                return True
+            return False  
+        elif policy == 'tofu':
+            keyfingerprint = hashlib.sha512(key.public_data).hexdigest()
+            fprint = 'sha512$' + keyfingerprint
+            cfg.set_node_attributes(nodename, {self.confluent_keyattrib: {'value': fprint}})
+            return True
+        return False
+
+    def disable_host_key_validation(self):
+        self.confluent_validate_hostkey = False
+
+    def auth_banner_received(self, msg, lang):
+        if hasattr(self, 'confluent_custom_ctx'):
+            self.confluent_custom_ctx['banner'] = msg
+
+    def password_auth_requested(self):
+        if not hasattr(self, 'confluent_custom_ctx'):
+            return None
+        if self.confluent_custom_ctx.get('nologon'):
+            raise _CancelSsh('nologon')
+        if 'initialpassword' in self.confluent_custom_ctx:
+            initpassword = self.confluent_custom_ctx.get('initialpassword')
+            del self.confluent_custom_ctx['initialpassword']
+            return initpassword
+        elif 'password' in self.confluent_custom_ctx:
+            password = self.confluent_custom_ctx.get('password')
+            del self.confluent_custom_ctx['password']
+            return password
+        else:
+            return None
+
+    def password_change_requested(self, prompt, lang):
+        print(repr(prompt))
+        print(repr(lang))
+
+    def password_change_failed(self):
+        print("pcf")
+
+    def password_changed(self):
+        print("pc")
+
+    def confluent_set_context(self, ctx):
+        self.confluent_custom_ctx = ctx
+
+
+def connect(target, context=None, disable_hostkey_validation=False, known_hosts='/etc/ssh/ssh_known_hosts', nodename=None, configmanager=None, keyattrib='pubkeys.ssh', **kwargs):
+    if context is None:
+        context = {}
+    def make_client():
+        client = _MyClient()
+        if disable_hostkey_validation:
+            client.disable_host_key_validation()
+        client.set_configmanager(configmanager)
+        client.set_nodename(nodename)
+        client.set_keyattrib(keyattrib) 
+
+        client.confluent_set_context(context)
+        return client
+    sco = asyncssh.SSHClientConnectionOptions(
+        client_factory=make_client,
+        x509_trusted_cert_paths=None,
+        known_hosts=known_hosts)
+    return asyncssh.connect(target, options=sco, **kwargs)
+
+
+async def get_ssh_banner(target):
+    mycontext = {'nologon': True}
+    try:
+        async with connect(target, disable_hostkey_validation=True, known_hosts=(), context=mycontext):
+            pass
+    except _CancelSsh:
+        pass
+    return mycontext.get('banner')
+
+
+if __name__ == '__main__':
+    import asyncio
+    import sys
+    asyncio.run(get_ssh_banner(sys.argv[1]))