From d6bff637db60dc381816fdd9f29bac7a89a82698 Mon Sep 17 00:00:00 2001 From: Jarrod Johnson Date: Fri, 23 Feb 2024 11:56:07 -0500 Subject: [PATCH] Commence work on async --- confluent_server/bin/confluent | 3 + confluent_server/confluent/core.py | 37 +++++---- confluent_server/confluent/messages.py | 5 ++ .../plugins/hardwaremanagement/enclosure.py | 27 +++---- .../plugins/hardwaremanagement/ipmi.py | 34 +++++---- .../confluent/plugins/shell/ssh.py | 75 ++++++------------- 6 files changed, 83 insertions(+), 98 deletions(-) diff --git a/confluent_server/bin/confluent b/confluent_server/bin/confluent index db66e587..708f5d73 100755 --- a/confluent_server/bin/confluent +++ b/confluent_server/bin/confluent @@ -17,6 +17,9 @@ import sys import os +import eventlet.hubs +eventlet.hubs.use_hub("eventlet.hubs.asyncio") + path = os.path.dirname(os.path.realpath(__file__)) path = os.path.realpath(os.path.join(path, '..', 'lib', 'python')) if path.startswith('/opt'): diff --git a/confluent_server/confluent/core.py b/confluent_server/confluent/core.py index a8a4412b..09a298e2 100644 --- a/confluent_server/confluent/core.py +++ b/confluent_server/confluent/core.py @@ -946,7 +946,7 @@ def _forward_rsp(connection, res): connection.sendall(r) -def handle_node_request(configmanager, inputdata, operation, +async def handle_node_request(configmanager, inputdata, operation, pathcomponents, autostrip=True): if log.logfull: raise exc.TargetResourceUnavailable('Filesystem full, free up space and restart confluent service') @@ -1027,7 +1027,7 @@ def handle_node_request(configmanager, inputdata, operation, else: raise Exception("TODO here") del pathcomponents[0:2] - passvalues = queue.Queue() + passvalues = asyncio.Queue() plugroute = routespec.routeinfo _plugin = None @@ -1095,23 +1095,23 @@ def handle_node_request(configmanager, inputdata, operation, numworkers = 0 for hfunc in nodesbyhandler: numworkers += 1 - workers.spawn(addtoqueue, passvalues, hfunc, {'nodes': nodesbyhandler[hfunc], + asyncio.create_task(addtoqueue(passvalues, hfunc, {'nodes': nodesbyhandler[hfunc], 'element': pathcomponents, 'configmanager': configmanager, 'inputdata': _get_input_data(_plugin, pathcomponents, operation, inputdata,nodes, - isnoderange, configmanager)}) + isnoderange, configmanager)})) for manager in nodesbymanager: numworkers += 1 - workers.spawn(addtoqueue, passvalues, dispatch_request, { + asyncio.create_task(addtoqueue(passvalues, dispatch_request, { 'nodes': nodesbymanager[manager], 'manager': manager, 'element': pathcomponents, 'configmanager': configmanager, - 'inputdata': inputdata, 'operation': operation, 'isnoderange': isnoderange}) + 'inputdata': inputdata, 'operation': operation, 'isnoderange': isnoderange})) if isnoderange or not autostrip: - return iterate_queue(numworkers, passvalues) + return await iterate_queue(numworkers, passvalues) else: if numworkers > 0: - return iterate_queue(numworkers, passvalues, nodes[0]) + return await iterate_queue(numworkers, passvalues, nodes[0]) else: raise exc.NotImplementedException() @@ -1132,10 +1132,10 @@ def _get_input_data(plugin_ext, pathcomponents, operation, inputdata, nodes, isnoderange,configmanager) -def iterate_queue(numworkers, passvalues, strip=False): +async def iterate_queue(numworkers, passvalues, strip=False): completions = 0 while completions < numworkers: - nv = passvalues.get() + nv = await passvalues.get() if nv == 'theend': completions += 1 else: @@ -1146,18 +1146,23 @@ def iterate_queue(numworkers, passvalues, strip=False): yield nv -def addtoqueue(theq, fun, kwargs): +async def addtoqueue(theq, fun, kwargs): try: result = fun(**kwargs) if isinstance(result, console.Console): - theq.put(result) + await theq.put(result) else: - for pv in result: - theq.put(pv) + + if isinstance(result, types.AsyncGeneratorType): + async for pv in result: + await theq.put(pv) + else: + for pv in result: + await theq.put(pv) except Exception as e: - theq.put(e) + await theq.put(e) finally: - theq.put('theend') + await theq.put('theend') def dispatch_request(nodes, manager, element, configmanager, inputdata, diff --git a/confluent_server/confluent/messages.py b/confluent_server/confluent/messages.py index ce36344d..7d8d3946 100644 --- a/confluent_server/confluent/messages.py +++ b/confluent_server/confluent/messages.py @@ -590,6 +590,11 @@ class InputFirmwareUpdate(ConfluentMessage): @property def filename(self): + # TODO: get the currennt_user and cross reference if that user is allowed to + # read... however, not sure wwhat to do if user is pure confluent user + # though the staging may get an explicit pass, which should cover the web case... + # media and firmware are ways to currently push things out, but if we allow profile export + # what then? if self._complexname: raise Exception('User requested substitutions, but code is ' 'written against old api, code must be fixed or ' diff --git a/confluent_server/confluent/plugins/hardwaremanagement/enclosure.py b/confluent_server/confluent/plugins/hardwaremanagement/enclosure.py index a59422c0..d910400b 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/enclosure.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/enclosure.py @@ -15,11 +15,9 @@ import confluent.core as core import confluent.messages as msg import pyghmi.exceptions as pygexc import confluent.exceptions as exc -import eventlet.queue as queue -import eventlet.greenpool as greenpool -def reseat_bays(encmgr, bays, configmanager, rspq): +async def reseat_bays(encmgr, bays, configmanager, rspq): try: for encbay in bays: node = bays[encbay] @@ -28,13 +26,13 @@ def reseat_bays(encmgr, bays, configmanager, rspq): '/nodes/{0}/_enclosure/reseat_bay'.format(encmgr), 'update', configmanager, inputdata={'reseat': int(encbay)}): - rspq.put(rsp) + await rspq.put(rsp) except pygexc.UnsupportedFunctionality as uf: - rspq.put(msg.ConfluentNodeError(node, str(uf))) + await rspq.put(msg.ConfluentNodeError(node, str(uf))) except exc.TargetEndpointUnreachable as uf: - rspq.put(msg.ConfluentNodeError(node, str(uf))) + await rspq.put(msg.ConfluentNodeError(node, str(uf))) finally: - rspq.put(None) + await rspq.put(None) def update(nodes, element, configmanager, inputdata): emebs = configmanager.get_node_attributes( @@ -54,17 +52,16 @@ def update(nodes, element, configmanager, inputdata): if em not in baysbyencmgr: baysbyencmgr[em] = {} baysbyencmgr[em][eb] = node - rspq = queue.Queue() - gp = greenpool.GreenPool(64) + reseattasks = [] + rspq = asyncio.Queue() for encmgr in baysbyencmgr: - gp.spawn_n(reseat_bays, encmgr, baysbyencmgr[encmgr], configmanager, rspq) - while gp.running(): - nrsp = rspq.get() + currtask = asyncio.create_task(reseat_bays(encmgr, baysbyencmgr[encmgr], configmanager, rspq)) + reseattasks.append(currtask) + while not all([task.done() for task in reseattasks]):; + nrsp = await rspq.get() if nrsp is not None: yield nrsp while not rspq.empty(): - nrsp = rspq.get() + nrsp = await rspq.get() if nrsp is not None: yield nrsp - - diff --git a/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py b/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py index 06a8c444..103df7c2 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/ipmi.py @@ -30,11 +30,11 @@ import eventlet.support.greendns from fnmatch import fnmatch import os import pwd -import pyghmi.constants as pygconstants -import pyghmi.exceptions as pygexc -import pyghmi.storage as storage +import aiohmi.constants as pygconstants +import aiohmi.exceptions as pygexc +import aiohmi.storage as storage console = eventlet.import_patched('pyghmi.ipmi.console') -ipmicommand = eventlet.import_patched('pyghmi.ipmi.command') +import aiohmi.ipmi.command as ipmicommand import socket import ssl import traceback @@ -174,7 +174,9 @@ def sanitize_invdata(indata): class IpmiCommandWrapper(ipmicommand.Command): - def __init__(self, node, cfm, **kwargs): + @classmethod + async def create(cls, node, cfm, **kwargs): + self = cls() self.cfm = cfm self.node = node self.sensormap = {} @@ -185,7 +187,7 @@ class IpmiCommandWrapper(ipmicommand.Command): (node,), ('secret.hardwaremanagementuser', 'collective.manager', 'secret.hardwaremanagementpassword', 'secret.ipmikg', 'hardwaremanagement.manager'), self._attribschanged) - super(self.__class__, self).__init__(**kwargs) + amait super().create(**kwargs) self.setup_confluent_keyhandler() try: os.makedirs('/var/cache/confluent/ipmi/') @@ -380,7 +382,7 @@ class IpmiConsole(conapi.Console): self.solconnection.send_break() -def perform_requests(operator, nodes, element, cfg, inputdata, realop): +async def perform_requests(operator, nodes, element, cfg, inputdata, realop): cryptit = cfg.decrypt cfg.decrypt = True configdata = cfg.get_node_attributes(nodes, _configattributes) @@ -430,7 +432,7 @@ def perform_requests(operator, nodes, element, cfg, inputdata, realop): def perform_request(operator, node, element, configdata, inputdata, cfg, results, realop): try: - return IpmiHandler(operator, node, element, configdata, inputdata, + return IpmiHandler.create(operator, node, element, configdata, inputdata, cfg, results, realop).handle_request() except pygexc.IpmiException as ipmiexc: excmsg = str(ipmiexc) @@ -460,9 +462,11 @@ def perform_request(operator, node, element, persistent_ipmicmds = {} -class IpmiHandler(object): - def __init__(self, operation, node, element, cfd, inputdata, cfg, output, +class IpmiHandler: + @classmethod + async def create(cls, operation, node, element, cfd, inputdata, cfg, output, realop): + self = cls() self.cfm = cfg self.invmap = {} self.output = output @@ -492,7 +496,7 @@ class IpmiHandler(object): except KeyError: # was no previous session pass try: - persistent_ipmicmds[(node, tenant)] = IpmiCommandWrapper( + persistent_ipmicmds[(node, tenant)] = await IpmiCommandWrapper.create( node, cfg, bmc=connparams['bmc'], userid=connparams['username'], password=connparams['passphrase'], kg=connparams['kg'], @@ -1636,14 +1640,14 @@ def initthread(): _ipmithread = eventlet.spawn(_ipmi_evtloop) -def create(nodes, element, configmanager, inputdata, realop='create'): +async def create(nodes, element, configmanager, inputdata, realop='create'): initthread() if element == ['_console', 'session']: if len(nodes) > 1: raise Exception("_console/session does not support multiple nodes") return IpmiConsole(nodes[0], configmanager) else: - return perform_requests( + return await perform_requests( 'update', nodes, element, configmanager, inputdata, realop) @@ -1652,7 +1656,7 @@ def update(nodes, element, configmanager, inputdata): return create(nodes, element, configmanager, inputdata, 'update') -def retrieve(nodes, element, configmanager, inputdata): +async def retrieve(nodes, element, configmanager, inputdata): initthread() if '/'.join(element).startswith('inventory/firmware/updates/active'): return firmwaremanager.list_updates(nodes, configmanager.tenant, @@ -1664,7 +1668,7 @@ def retrieve(nodes, element, configmanager, inputdata): return firmwaremanager.list_updates(nodes, configmanager.tenant, element, 'ffdc') else: - return perform_requests('read', nodes, element, configmanager, + return await perform_requests('read', nodes, element, configmanager, inputdata, 'read') def delete(nodes, element, configmanager, inputdata): diff --git a/confluent_server/confluent/plugins/shell/ssh.py b/confluent_server/confluent/plugins/shell/ssh.py index 2a6b65ec..98f98605 100644 --- a/confluent_server/confluent/plugins/shell/ssh.py +++ b/confluent_server/confluent/plugins/shell/ssh.py @@ -22,35 +22,29 @@ import confluent.exceptions as cexc import confluent.interface.console as conapi import confluent.log as log -try: - import cryptography -except ImportError: - # Using older, non-crypography based paramiko - cryptography = None -import eventlet import hashlib import sys sys.modules['gssapi'] = None -paramiko = eventlet.import_patched('paramiko') -warnhostkey = False -if cryptography and cryptography.__version__.split('.') < ['1', '5']: - # older cryptography with paramiko breaks most key support except - # ed25519 - warnhostkey = True - paramiko.transport.Transport._preferred_keys = filter( - lambda x: 'ed25519' in x, - paramiko.transport.Transport._preferred_keys) +#paramiko = eventlet.import_patched('paramiko') +import asyncio +import asyncssh -class HostKeyHandler(paramiko.client.MissingHostKeyPolicy): + +class HostKeyHandler: def __init__(self, configmanager, node): self.cfm = configmanager self.node = node def missing_host_key(self, client, hostname, key): + # 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 fingerprint = 'sha512$' + hashlib.sha512(key.asbytes()).hexdigest() cfg = self.cfm.get_node_attributes( self.node, ('pubkeys.ssh', 'pubkeys.addpolicy')) @@ -94,11 +88,12 @@ class SshShell(conapi.Console): self.height = height if not self.connected: return + # asyncssh channel has change_terminal_size, hurray self.shell.resize_pty(width=width, height=height) - def recvdata(self): + async def recvdata(self): while self.connected: - pendingdata = self.shell.recv(8192) + pendingdata = await self.shell.stdout.read(8192) if not pendingdata: self.ssh.close() if self.datacallback: @@ -123,31 +118,18 @@ class SshShell(conapi.Console): self.inputmode = -3 eventlet.spawn_n(self.do_logon) - def do_logon(self): - self.ssh = paramiko.SSHClient() - self.ssh.set_missing_host_key_policy( - HostKeyHandler(self.nodeconfig, self.node)) + 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 + #sco.client_fatory = SSHKnownHostsLookup + try: self.datacallback('\r\nConnecting to {}...'.format(self.node)) - self.ssh.connect(self.node, username=self.username, - password=self.password, allow_agent=False, - look_for_keys=False) - except paramiko.AuthenticationException as e: - self.ssh.close() - self.inputmode = 0 - self.username = b'' - self.password = b'' - self.datacallback('\r\nError connecting to {0}:\r\n {1}\r\n'.format(self.node, str(e))) - self.datacallback('\r\nlogin as: ') - return - except paramiko.ssh_exception.NoValidConnectionsError as e: - self.ssh.close() - self.datacallback('\r\nError connecting to {0}:\r\n {1}\r\n'.format(self.node, str(e))) - self.inputmode = 0 - self.username = b'' - self.password = b'' - self.datacallback('\r\nlogin as: ') - return + try: + self.ssh = await asyncssh.connect(self,node, username=self.username, password=self.password, known_hosts='/etc/ssh/ssh_known_hosts') + except ValueError: + #TODO: non-cert ssh targets + raise except cexc.PubkeyInvalid as pi: self.ssh.close() self.keyaction = b'' @@ -158,17 +140,6 @@ class SshShell(conapi.Console): self.inputmode = -1 self.datacallback('\r\nEnter "disconnect" or "accept": ') return - except paramiko.SSHException as pi: - self.ssh.close() - self.inputmode = -2 - warn = str(pi) - if warnhostkey: - warn += ' (Older cryptography package on this host only ' \ - 'works with ed25519, check ssh startup on target ' \ - 'and permissions on /etc/ssh/*key)\r\n' \ - 'Press Enter to close...' - self.datacallback('\r\n' + warn) - return except Exception as e: self.ssh.close() self.ssh.close()