Merge pull request #235 from Obihoernchen/aiohmi

Various asyncio fixes

confluent_server/aiohmi/ipmi/command.py

@@ -329,8 +329,8 @@ class Command(object):
         await self.oem_init()
 
         if hasattr(self._oem, 'set_power'):
-            return self._oem.set_power(powerstate,
-                                       bridge_request=bridge_request)
+            return await self._oem.set_power(powerstate,
+                                             bridge_request=bridge_request)
 
         if hasattr(self._oem, 'process_power_state'):
             powerstate = self._oem.process_power_state(
@@ -675,7 +675,7 @@ class Command(object):
         await self.init_sdr()
         for fruid in self._sdr.fru:
             if self._sdr.fru[fruid].fru_name == component:
-                return self._oem.process_fru(fru.FRU(
+                return await self._oem.process_fru(fru.FRU(
                     ipmicmd=self, fruid=fruid,
                     sdr=self._sdr.fru[fruid]).info, component)
         return await self._oem.get_inventory_of_component(component)
@@ -738,7 +738,8 @@ class Command(object):
         This provides a detailed view of the LEDs of the managed system.
         """
         await self.oem_init()
-        return await self._oem.get_leds()
+        async for led in self._oem.get_leds():
+            yield led
 
     async def get_ntp_enabled(self):
         await self.oem_init()
@@ -806,7 +807,7 @@ class Command(object):
                 rsp = await self.raw_command(command=0x2d, netfn=4,
                                        rslun=currsensor.sensor_lun,
                                        data=(currsensor.sensor_number,))
-                return self._sdr.sensors[sensor].decode_sensor_reading(
+                return await self._sdr.sensors[sensor].decode_sensor_reading(
                     self, rsp['data'])
         await self.oem_init()
         return await self._oem.get_sensor_reading(sensorname)
@@ -853,9 +854,9 @@ class Command(object):
         else:
             raise Exception("Unrecognized data format " + repr(fetchdata))
 
-    def get_extended_bmc_configuration(self):
-        self.oem_init()
-        return self._oem.get_extended_bmc_configuration()
+    async def get_extended_bmc_configuration(self):
+        await self.oem_init()
+        return await self._oem.get_extended_bmc_configuration()
 
     async def get_bmc_configuration(self):
         await self.oem_init()
@@ -1091,7 +1092,7 @@ class Command(object):
                 if rsp['code'] == 203:  # Sensor does not exist, optional dev
                     continue
                 raise exc.IpmiException(rsp['error'], code=rsp['code'])
-            yield self._sdr.sensors[sensor].\
+            yield await self._sdr.sensors[sensor].\
                 decode_sensor_reading(self, rsp['data'])
         await self.oem_init()
         async for reading in self._oem.get_sensor_data():
@@ -1581,7 +1582,7 @@ class Command(object):
         }
         b |= privilege_levels[privilege_level] & 0b00000111
         data.append(b)
-        response = self.raw_command(netfn=0x06, command=0x40, data=data)
+        response = await self.raw_command(netfn=0x06, command=0x40, data=data)
         if 'error' in response:
             raise Exception(response['error'])
         return True
@@ -2311,4 +2312,4 @@ class Command(object):
 
         """
         await self.oem_init()
-        return await self._oem.set_oem_extended_privilleges(uid)
+        return await self._oem.set_oem_extended_privilleges(uid)

confluent_server/aiohmi/ipmi/console.py

@@ -133,7 +133,7 @@ class Console(object):
                     response['code'])
                 return
         if 'error' in response:
-            self._print_error(response['error'])
+            await self._print_error(response['error'])
             return
         self.activated = True
         # data[0:3] is reserved except for the test mode, which we don't use
@@ -381,7 +381,7 @@ class Console(object):
                 await self.send_payload(ackpayload, retry=False)
             except exc.IpmiException:
                 # if the session is broken, then close the SOL session
-                self.close()
+                await self.close()
         if self.myseq != 0 and ackseq == self.myseq:  # the bmc has something
             # to say about last xmit
             self.awaitingack = False

confluent_server/aiohmi/ipmi/oem/lenovo/config.py

@@ -74,7 +74,7 @@ async def run_command_with_retry(connection, data):
         except pygexc.IpmiException as e:
             if e.ipmicode != 0xa or not tries:
                 raise
-            connection.ipmi_session.pause(1)
+            await connection.ipmi_session.pause(1)
 
 
 def _convert_syntax(raw):
@@ -198,7 +198,7 @@ class LenovoFirmwareConfig(object):
                     break
             except KeyError:
                 pass
-            self.connection.ipmi_session.pause(5)
+            await self.connection.ipmi_session.pause(5)
         filehandle = response['data'][3:7]
         filehandle = struct.unpack("<I", filehandle)[0]
         return filehandle

confluent_server/aiohmi/ipmi/oem/lenovo/energy.py

@@ -38,7 +38,7 @@ class EnergyManager(object):
                                         'Node Power', 'Total Power')
                 self._mypowermeters = ('node power', 'total power', 'gpu power', 'riser 1 power', 'riser 2 power')
                 self._usefapm = True
-                return
+                return self
         except pygexc.IpmiException:
             pass
 
@@ -80,7 +80,7 @@ class EnergyManager(object):
                 break
             except pygexc.IpmiException as ie:
                 if tries and ie.ipmicode == 0xc3:
-                    ipmicmd.ipmi_session.pause(0.1)
+                    await ipmicmd.ipmi_session.pause(0.1)
                     continue
                 raise
         if rsp is None:
@@ -159,8 +159,14 @@ class Energy(object):
 
 
 if __name__ == '__main__':
+    import asyncio
     import os
     import aiohmi.ipmi.command as cmd
     import sys
-    c = cmd.Command(sys.argv[1], os.environ['BMCUSER'], os.environ['BMCPASS'])
-    EnergyManager(c).get_dc_energy(c)
+
+    async def main():
+        c = await cmd.Command.create(sys.argv[1], os.environ['BMCUSER'], os.environ['BMCPASS'])
+        manager = await EnergyManager.create(c)
+        print(await manager.get_dc_energy(c))
+
+    asyncio.run(main())

confluent_server/aiohmi/ipmi/oem/lenovo/handler.py

@@ -336,9 +336,9 @@ class OEMHandler(generic.OEMHandler):
             event['component'] += ' {0}'.format(evdata[1] & 0b11111)
 
     async def reseat_bay(self, bay):
-        if self.is_fpc:
+        if await self.is_fpc():
             return await self.smmhandler.reseat_bay(bay)
-        elif self.has_xcc and bay == -1:
+        elif await self.has_xcc() and bay == -1:
             return await self.immhandler.reseat()
         return await super(OEMHandler, self).reseat_bay(bay)
 
@@ -349,7 +349,7 @@ class OEMHandler(generic.OEMHandler):
         elif await self.is_fpc():
             return await self.smmhandler.get_ntp_enabled(self._fpc_variant)
         elif self.has_tsma:
-            return self.tsmahandler.get_ntp_enabled()
+            return await self.tsmahandler.get_ntp_enabled()
         return None
 
     async def get_ntp_servers(self):
@@ -360,9 +360,9 @@ class OEMHandler(generic.OEMHandler):
             srvs.append(ntpres['data'][129:257].rstrip('\x00'))
             return srvs
         if await self.is_fpc():
-            return await self.smmhandler.get_ntp_servers()
+            return self.smmhandler.get_ntp_servers()
         if self.has_tsma:
-            return self.tsmahandler.get_ntp_servers()
+            return await self.tsmahandler.get_ntp_servers()
         return ()
 
     async def set_ntp_enabled(self, enabled):
@@ -375,7 +375,7 @@ class OEMHandler(generic.OEMHandler):
                     netfn=0x32, command=0xa8, data=(3, 0), timeout=15)
             return True
         if await self.is_fpc():
-            await self.smmhandler.set_ntp_enabled(enabled)
+            self.smmhandler.set_ntp_enabled(enabled)
             return True
         if self.has_tsma:
             await self.tsmahandler.set_ntp_enabled(enabled)
@@ -393,7 +393,7 @@ class OEMHandler(generic.OEMHandler):
             if not 0 <= index <= 2:
                 raise pygexc.InvalidParameterValue(
                     'SMM supports indexes 0 through 2')
-            await self.smmhandler.set_ntp_server(server, index)
+            self.smmhandler.set_ntp_server(server, index)
             return True
         elif self.has_tsma:
             if not (0 <= index <= 1):
@@ -493,8 +493,7 @@ class OEMHandler(generic.OEMHandler):
         if await self.has_tsm() or await self.has_ami() or await self.has_asrock():
             # Thinkserver with TSM
             if not self.oem_inventory_info:
-                async for desc in self._collect_tsm_inventory():
-                    yield desc
+                await self._collect_tsm_inventory()
             for compname in self.oem_inventory_info:
                 yield compname
         elif await self.has_imm():
@@ -516,21 +515,22 @@ class OEMHandler(generic.OEMHandler):
         elif await self.is_fpc():
             async for compname in self.smmhandler.get_inventory_descriptions(
                     self.ipmicmd, await self.is_fpc()):
-                yield (compname, self.smmhandler.get_inventory_of_component(
+                yield (compname, await self.smmhandler.get_inventory_of_component(
                     self.ipmicmd, compname))
 
     async def get_sensor_data(self):
         if await self.has_imm():
             async for name in self.immhandler.get_oem_sensor_names(self.ipmicmd):
-                yield self.immhandler.get_oem_sensor_reading(name,
-                                                             self.ipmicmd)
+                yield await self.immhandler.get_oem_sensor_reading(name,
+                                                                   self.ipmicmd)
         elif await self.is_fpc():
             for name in nextscale.get_sensor_names(self.ipmicmd,
                                                    self._fpc_variant):
-                yield nextscale.get_sensor_reading(name, self.ipmicmd,
-                                                   self._fpc_variant)
+                yield await nextscale.get_sensor_reading(name, self.ipmicmd,
+                                                         self._fpc_variant)
         elif await self.has_ami():
-            self.get_ami_sensor_data()
+            async for reading in self.get_ami_sensor_data():
+                yield reading
 
     async def get_sensor_descriptions(self):
         if await self.has_imm():
@@ -548,26 +548,26 @@ class OEMHandler(generic.OEMHandler):
 
     async def get_sensor_reading(self, sensorname):
         if await self.has_imm():
-            return self.immhandler.get_oem_sensor_reading(sensorname,
-                                                          self.ipmicmd)
+            return await self.immhandler.get_oem_sensor_reading(sensorname,
+                                                                self.ipmicmd)
         elif await self.is_fpc():
-            return nextscale.get_sensor_reading(sensorname, self.ipmicmd,
-                                                self._fpc_variant)
+            return await nextscale.get_sensor_reading(sensorname, self.ipmicmd,
+                                                      self._fpc_variant)
         elif await self.has_ami():
-            self.get_ami_sensor_reading(sensorname)
+            return await self.get_ami_sensor_reading(sensorname)
         return ()
 
     async def get_inventory_of_component(self, component):
         if await self.has_tsm() or await self.has_ami() or await self.has_asrock():
             await self._collect_tsm_inventory()
-            return await self.oem_inventory_info.get(component, None)
+            return self.oem_inventory_info.get(component, None)
         if await self.has_imm():
             return await self.immhandler.get_component_inventory(component)
         if await self.is_fpc():
-            return await self.smmhandler.get_inventory_of_component(component)
+            return await self.smmhandler.get_inventory_of_component(self.ipmicmd, component)
 
-    def get_cmd_type(self, categorie_item, catspec):
-        if self.has_asrock:
+    async def get_cmd_type(self, categorie_item, catspec):
+        if await self.has_asrock():
             cmd_type = catspec["command"]["asrock"]
         elif categorie_item in categorie_items:
             cmd_type = catspec["command"]["lenovo"]
@@ -578,9 +578,7 @@ class OEMHandler(generic.OEMHandler):
 
     async def _collect_tsm_inventory(self):
         self.oem_inventory_info = {}
-        asrock = False
-        if self.has_asrock:
-            asrock = True
+        asrock = await self.has_asrock()
         for catid, catspec in inventory.categories.items():
             # skip the inventory fields if the system is RS160
             if asrock and catid not in categorie_items:
@@ -589,7 +587,7 @@ class OEMHandler(generic.OEMHandler):
                     and catspec["workaround_bmc_bug"](
                         "ami" if await self.has_ami() else "lenovo")):
                 rsp = None
-                cmd = self.get_cmd_type(catid, catspec)
+                cmd = await self.get_cmd_type(catid, catspec)
                 tmp_command = dict(cmd)
                 tmp_command["data"] = list(tmp_command["data"])
                 count = 0
@@ -616,7 +614,7 @@ class OEMHandler(generic.OEMHandler):
                 rsp["data"] = buffer(bytearray(rsp["data"]))
             else:
                 try:
-                    cmd = self.get_cmd_type(catid, catspec)
+                    cmd = await self.get_cmd_type(catid, catspec)
                     rsp = await self.ipmicmd.raw_command(**cmd)
                 except pygexc.IpmiException:
                     continue
@@ -768,7 +766,7 @@ class OEMHandler(generic.OEMHandler):
             except (AttributeError, KeyError, IndexError):
                 pass
             if await self.has_xcc() and name and name.startswith('PSU '):
-                self.immhandler.augment_psu_info(fru, name)
+                await self.immhandler.augment_psu_info(fru, name)
             if (await self.has_xcc() and 'memory_type' in fru
                     and fru['memory_type'] == 'Unknown'):
                 await self.immhandler.fetch_dimm(name, fru)
@@ -851,7 +849,7 @@ class OEMHandler(generic.OEMHandler):
     async def get_oem_firmware(self, bmcver, components, category):
         if await self.has_tsm() or await self.has_ami() or await self.has_asrock():
             command = firmware.get_categories()["firmware"]
-            fw_cmd = self.get_cmd_type("firmware", command)
+            fw_cmd = await self.get_cmd_type("firmware", command)
             rsp = await self.ipmicmd.raw_command(**fw_cmd)
 
             # the newest Lenovo ThinkServer versions are returning Bios version
@@ -859,7 +857,7 @@ class OEMHandler(generic.OEMHandler):
             bios_versions = None
             if await self.has_tsm() or await self.has_asrock():
                 bios_command = firmware.get_categories()["bios_version"]
-                bios_cmd = self.get_cmd_type("bios_version", bios_command)
+                bios_cmd = await self.get_cmd_type("bios_version", bios_command)
                 bios_rsp = await self.ipmicmd.raw_command(**bios_cmd)
                 if await self.has_asrock():
                     bios_versions = bios_rsp['data']
@@ -942,7 +940,7 @@ class OEMHandler(generic.OEMHandler):
                 name += rsp['data'][:]
             return name.rstrip('\x00')
         elif await self.is_fpc():
-            return await self.smmhandler.get_domain()
+            return self.smmhandler.get_domain()
 
     async def set_oem_domain_name(self, name):
         if await self.has_tsm():
@@ -961,20 +959,20 @@ class OEMHandler(generic.OEMHandler):
             await self._restart_dns()
             return
         elif await self.is_fpc():
-            await self.smmhandler.set_domain(name)
+            self.smmhandler.set_domain(name)
 
     async def set_hostname(self, hostname):
         if await self.has_xcc():
             return await self.immhandler.set_hostname(hostname)
         elif await self.is_fpc():
-            return await self.smmhandler.set_hostname(hostname)
+            return self.smmhandler.set_hostname(hostname)
         return await super(OEMHandler, self).set_hostname(hostname)
 
     async def get_hostname(self):
         if await self.has_xcc():
             return await self.immhandler.get_hostname()
         elif await self.is_fpc():
-            return await self.smmhandler.get_hostname()
+            return self.smmhandler.get_hostname()
         return await super(OEMHandler, self).get_hostname()
 
     """ Gets a remote console launcher for a Lenovo ThinkServer.
@@ -1029,14 +1027,14 @@ class OEMHandler(generic.OEMHandler):
 
     async def add_extra_net_configuration(self, netdata, channel=None):
         if await self.has_tsm():
-            ipv6_addr = await self.ipmicmd.raw_command(
+            ipv6_addr = (await self.ipmicmd.raw_command(
                 netfn=0x0c, command=0x02,
-                data=(0x01, 0xc5, 0x00, 0x00))["data"][1:]
+                data=(0x01, 0xc5, 0x00, 0x00)))['data'][1:]
             if not ipv6_addr:
                 return
-            rspdata = await self.ipmicmd.raw_command(
+            rspdata = (await self.ipmicmd.raw_command(
                 netfn=0xc, command=0x02,
-                data=(0x1, 0xc6, 0, 0))['data']
+                data=(0x1, 0xc6, 0, 0)))['data']
             ipv6_prefix_ba = bytearray(rspdata)
             ipv6_prefix = ipv6_prefix_ba[1]
 
@@ -1135,9 +1133,9 @@ class OEMHandler(generic.OEMHandler):
         targshortname = _megarac_abbrev_image(imagename)
         shortnames = await self._megarac_fetch_image_shortnames()
         while targshortname not in shortnames:
-            self.ipmicmd.wait_for_rsp(1)
+            await self.ipmicmd.wait_for_rsp(1)
             shortnames = await self._megarac_fetch_image_shortnames()
-        self.ipmicmd.ipmi_session.pause(10)
+        await self.ipmicmd.ipmi_session.pause(10)
         try:
             await self.ipmicmd.raw_command(netfn=0x32, command=0xa0, data=(1, 0))
             await self.ipmicmd.ipmi_session.pause(5)

confluent_server/aiohmi/ipmi/oem/lenovo/imm.py

@@ -839,7 +839,7 @@ class IMMClient(object):
                     hwmap[aname] = bdata
             self.datacache['lenovo_cached_hwmap'] = (hwmap,
                                                      util._monotonic_time())
-        self.weblogout()
+        await self.weblogout()
         return hwmap
 
     async def get_firmware_inventory(self, bmcver, components, category):
@@ -1434,7 +1434,7 @@ class XCCClient(IMMClient):
                 if controller != disk.id[0]:
                     raise pygexc.UnsupportedFunctionality(
                         'Cannot span arrays across controllers')
-            raidmap = self._raid_number_map(controller)
+            raidmap = await self._raid_number_map(controller)
             if not raidmap:
                 raise pygexc.InvalidParameterValue(
                     'There are no available drives for a new array')
@@ -1487,21 +1487,21 @@ class XCCClient(IMMClient):
             # TODO(): adding new volume to existing array would be here
             pass
 
-    def _make_jbod(self, disk, realcfg):
+    async def _make_jbod(self, disk, realcfg):
         currstatus = self._get_status(disk, realcfg)
         if currstatus.lower() == 'jbod':
             return
-        self._make_available(disk, realcfg)
-        self._set_drive_state(disk, 16)
+        await self._make_available(disk, realcfg)
+        await self._set_drive_state(disk, 16)
 
-    def _make_global_hotspare(self, disk, realcfg):
+    async def _make_global_hotspare(self, disk, realcfg):
         currstatus = self._get_status(disk, realcfg)
         if currstatus.lower() == 'global hot spare':
             return
-        self._make_available(disk, realcfg)
-        self._set_drive_state(disk, 1)
+        await self._make_available(disk, realcfg)
+        await self._set_drive_state(disk, 1)
 
-    def _make_available(self, disk, realcfg):
+    async def _make_available(self, disk, realcfg):
         # 8 if jbod, 4 if hotspare.., leave alone if already...
         currstatus = self._get_status(disk, realcfg)
         newstate = None
@@ -1511,7 +1511,7 @@ class XCCClient(IMMClient):
             newstate = 4
         elif currstatus.lower() == 'jbod':
             newstate = 8
-        self._set_drive_state(disk, newstate)
+        await self._set_drive_state(disk, newstate)
 
     def _get_status(self, disk, realcfg):
         for cfgdisk in realcfg.disks:
@@ -1536,7 +1536,7 @@ class XCCClient(IMMClient):
         wc = await self.wc()
         rsp = await wc.grab_json_response(
             '/api/function', {'raidlink_ClearRaidConf': '1'})
-        self.weblogout()
+        await self.weblogout()
         if rsp['return'] != 0:
             raise Exception('Unexpected return to clear config: ' + repr(rsp))
 
@@ -1560,10 +1560,10 @@ class XCCClient(IMMClient):
                 if rsp.get('return', -1) != 0:
                     raise Exception(
                         'Unexpected return to volume deletion: ' + repr(rsp))
-                self._wait_storage_async()
+                await self._wait_storage_async()
         for disk in cfgspec.disks:
             await self._make_available(disk, realcfg)
-        self.weblogout()
+        await self.weblogout()
 
     async def apply_storage_configuration(self, cfgspec):
         realcfg = await self.get_storage_configuration(False)
@@ -2168,7 +2168,7 @@ class XCCClient(IMMClient):
             retry = 3
             while not complete and retry > 0:
                 try:
-                    pgress, status = self.grab_redfish_response_with_status(
+                    pgress, status = await self.grab_redfish_response_with_status(
                         monitorurl)
                 except socket.timeout:
                     pgress = None                
@@ -2205,10 +2205,10 @@ class XCCClient(IMMClient):
                 return 'complete'
             return 'pending'
         finally:
-            self.grab_redfish_response_with_status(
+            await self.grab_redfish_response_with_status(
                 '/redfish/v1/UpdateService',
                 {'HttpPushUriTargetsBusy': False}, method='PATCH')
-            self.grab_redfish_response_with_status(
+            await self.grab_redfish_response_with_status(
                 '/redfish/v1/UpdateService',
                 {'HttpPushUriTargets': []}, method='PATCH')
 
@@ -2365,7 +2365,7 @@ class XCCClient(IMMClient):
             if verifystatus == 2:
                 raise Exception('Failed to verify firmware image')
             if verifystatus != 1:
-                ipmisession.Session.pause(1)
+                await ipmisession.Session.pause(1)
             if verifystatus not in (0, 1, 255):
                 errmsg = repr(rsp) if rsp else wc.lastjsonerror
                 raise Exception(
@@ -2410,7 +2410,7 @@ class XCCClient(IMMClient):
                 errmsg = repr(rsp) if rsp else wc.lastjsonerror
                 raise Exception(
                     'Unexpected result from PCI select: ' + errmsg)
-            self.set_property('/v2/ibmc/uefi/force-inventory', 1)
+            await self.set_property('/v2/ibmc/uefi/force-inventory', 1)
         else:
             await self._refresh_token()
             rsp = await wc.grab_json_response(
@@ -2435,7 +2435,7 @@ class XCCClient(IMMClient):
         complete = False
         while not complete:
             await self._refresh_token()
-            ipmisession.Session.pause(3)
+            await ipmisession.Session.pause(3)
             rsp = await wc.grab_json_response(
                 '/api/dataset/imm_firmware_progress')
             progress({'phase': 'apply',

confluent_server/aiohmi/ipmi/oem/lenovo/nextscale.py

@@ -671,14 +671,14 @@ class SMMClient(object):
             for bayn in range(1, numbays + 1):
                 if fnmatch.fnmatch('bay{0}_user_cap'.format(bayn),
                                    key.lower()):
-                    self.set_bay_cap(bayn, changeset[key]['value'])
+                    await self.set_bay_cap(bayn, changeset[key]['value'])
                 if fnmatch.fnmatch(
                         'bay{0}_user_cap_active'.format(bayn), key.lower()):
-                    self.set_bay_cap_active(bayn, changeset[key]['value'])
+                    await self.set_bay_cap_active(bayn, changeset[key]['value'])
             if fnmatch.fnmatch('chassis_user_cap', key.lower()):
-                self.set_bay_cap(numbays + 1, changeset[key]['value'])
+                await self.set_bay_cap(numbays + 1, changeset[key]['value'])
             if fnmatch.fnmatch('chassis_user_cap_active', key.lower()):
-                self.set_bay_cap_active(numbays + 1, changeset[key]['value'])
+                await self.set_bay_cap_active(numbays + 1, changeset[key]['value'])
             if fnmatch.fnmatch('fanspeed', key.lower()):
                 for mode in self.fanmodes:
                     byteval = mode
@@ -784,7 +784,7 @@ class SMMClient(object):
         if progress:
             progress({'phase': 'initializing', 'progress': initpct})
         while bytearray(rsp['data'])[0] != 0:
-            ipmisession.Session.pause(3)
+            await ipmisession.Session.pause(3)
             initpct += 3.0
             if initpct > 99.0:
                 initpct = 99.0
@@ -816,10 +816,10 @@ class SMMClient(object):
     async def process_fru(self, fru):
         smmv1 = self.smm_variant & 0xf0 == 0
         # TODO(jjohnson2): can also get EIOM, SMM, and riser data if warranted
-        snum = bytes(await self.ipmicmd.raw_command(
-            netfn=0x32, command=0xb0, data=(5, 1))['data'][:])
-        mnum = bytes(await self.ipmicmd.raw_command(
-            netfn=0x32, command=0xb0, data=(5, 0))['data'][:])
+        snum = bytes((await self.ipmicmd.raw_command(
+            netfn=0x32, command=0xb0, data=(5, 1)))['data'][:])
+        mnum = bytes((await self.ipmicmd.raw_command(
+            netfn=0x32, command=0xb0, data=(5, 0)))['data'][:])
         if not smmv1:
             snum = snum[2:]
             mnum = mnum[2:]
@@ -1039,7 +1039,7 @@ class SMMClient(object):
         complete = False
         tries = 0
         while not complete:
-            ipmisession.Session.pause(3)
+            await ipmisession.Session.pause(3)
             wc.request('POST', '/data', 'get=fwProgress,fwUpdate')
             try:
                 rsp = wc.getresponse()
@@ -1061,16 +1061,16 @@ class SMMClient(object):
             complete = percent >= 100.0
         return 'complete'
 
-    def get_inventory_descriptions(self, ipmicmd, variant):
+    async def get_inventory_descriptions(self, ipmicmd, variant):
         if variant >> 5 == 0:
             return
-        psucount = get_psu_count(ipmicmd, variant)
+        psucount = await get_psu_count(ipmicmd, variant)
         for idx in range(psucount):
             yield 'PSU {}'.format(idx + 1)
 
-    def get_inventory_of_component(self, ipmicmd, component):
+    async def get_inventory_of_component(self, ipmicmd, component):
         psuidx = int(component.replace('PSU ', ''))
-        return self.get_psu_info(ipmicmd, psuidx)
+        return await self.get_psu_info(ipmicmd, psuidx)
 
     async def get_psu_info(self, ipmicmd, psunum):
         psuinfo = await ipmicmd.raw_command(0x34, 0x6, data=(psunum,))

confluent_server/aiohmi/ipmi/private/localsession.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import ctypes
 import fcntl
 from select import select

confluent_server/aiohmi/ipmi/private/session.py

@@ -1023,11 +1023,11 @@ class Session(object):
             # tried ipmi 2.0 against a 1.5 which should work, but some bmcs
             # thought 'reserved' meant 'must be zero'
             self.ipmi15only = 1
-            return self._get_channel_auth_cap()
+            return await self._get_channel_auth_cap()
         mysuffix = " while trying to get channel authentication capabalities"
         errstr = get_ipmi_error(response, suffix=mysuffix)
         if errstr:
-            self.onlogon({'error': errstr})
+            await self.onlogon({'error': errstr})
             return
         data = response['data']
         self.currentchannel = data[0]
@@ -1035,7 +1035,7 @@ class Session(object):
             self.ipmiversion = 2.0
         if self.ipmiversion == 1.5:
             if not (data[1] & 0b100):
-                self.onlogon(
+                await self.onlogon(
                     {'error':
                      "MD5 required but not enabled/available on target BMC"})
                 return
@@ -1043,16 +1043,16 @@ class Session(object):
         elif self.ipmiversion == 2.0:
             await self._open_rmcpplus_request()
 
-    def _got_session_challenge(self, response):
+    async def _got_session_challenge(self, response):
         errstr = get_ipmi_error(response,
                                 suffix=" while getting session challenge")
         if errstr:
-            self.onlogon({'error': errstr})
+            await self.onlogon({'error': errstr})
             return
         data = response['data']
         self.sessionid = struct.unpack("<I", bytes(data[0:4]))[0]
         self.authtype = 2
-        self._activate_session(data[4:])
+        await self._activate_session(data[4:])
 
     # NOTE(jbjohnso):
     # This sends the activate session payload.  We pick '1' as the requested
@@ -1067,7 +1067,7 @@ class Session(object):
     async def _activated_session(self, response):
         errstr = get_ipmi_error(response)
         if errstr:
-            self.onlogon({'error': errstr})
+            await self.onlogon({'error': errstr})
             return
         data = response['data']
         self.sessionid = struct.unpack("<I", bytes(data[1:5]))[0]
@@ -1085,8 +1085,8 @@ class Session(object):
                 # some implementations will let us get this far,
                 # but suddenly get skiddish.  Try again in such a case
                 self.privlevel = 3
-                response = self.raw_command(netfn=0x6, command=0x3b,
-                                            data=[self.privlevel])
+                response = await self.raw_command(netfn=0x6, command=0x3b,
+                                                  data=[self.privlevel])
             if response['code']:
                 self.logged = 0
                 self.onlogpayload = None
@@ -1097,7 +1097,7 @@ class Session(object):
                     self.privlevel, self.userid)
                 errstr = get_ipmi_error(response, suffix=mysuffix)
                 if errstr:
-                    self.onlogon({'error': errstr})
+                    await self.onlogon({'error': errstr})
                     return
         self.logging = False
         self.logoutexpiry = None
@@ -1548,7 +1548,7 @@ class Session(object):
                 errstr = constants.rmcp_codes[data[1]]
             else:
                 errstr = "Unrecognized RMCP code %d" % data[1]
-            self.onlogon({'error': errstr})
+            await self.onlogon({'error': errstr})
             return -9
         self.allowedpriv = data[2]
         # NOTE(jbjohnso): At this point, the BMC has no idea about what user
@@ -1692,7 +1692,7 @@ class Session(object):
         aclen = len(expectedauthcode)
         authcode = struct.pack("%dB" % aclen, *data[8:aclen + 8])
         if authcode != expectedauthcode:
-            self.onlogon({'error': "Invalid RAKP4 integrity code (wrong Kg?)"})
+            await self.onlogon({'error': "Invalid RAKP4 integrity code (wrong Kg?)"})
             return
         self.sessionid = self.pendingsessionid
         self.integrityalgo = self.attemptedhash

confluent_server/aiohmi/ipmi/sdr.py

@@ -475,7 +475,7 @@ class SDREntry(object):
             health = const.Health.Ok
         return desc, health
 
-    def decode_sensor_reading(self, ipmicmd, reading):
+    async def decode_sensor_reading(self, ipmicmd, reading):
         numeric = None
         output = {
             'name': self.sensor_name,
@@ -495,8 +495,8 @@ class SDREntry(object):
         if numeric is not None:
             lowerbound = numeric - (0.5 + (self.tolerance / 2.0))
             upperbound = numeric + (0.5 + (self.tolerance / 2.0))
-            lowerbound = self.decode_value(ipmicmd, lowerbound)
-            upperbound = self.decode_value(ipmicmd, upperbound)
+            lowerbound = await self.decode_value(ipmicmd, lowerbound)
+            upperbound = await self.decode_value(ipmicmd, upperbound)
             output['value'] = (lowerbound + upperbound) / 2.0
             output['imprecision'] = output['value'] - lowerbound
             discrete = False
@@ -552,13 +552,13 @@ class SDREntry(object):
                 output['state_ids'].append(self.assert_trap_value(6))
         return SensorReading(output, self.unit_suffix)
 
-    def _set_tmp_formula(self, ipmicmd, value):
-        rsp = ipmicmd.raw_command(netfn=4, command=0x23,
-                                  data=(self.sensor_number, value))
+    async def _set_tmp_formula(self, ipmicmd, value):
+        rsp = await ipmicmd.raw_command(netfn=4, command=0x23,
+                                        data=(self.sensor_number, value))
         # skip next reading field, not used in on-demand situation
         self.decode_formula(rsp['data'][1:])
 
-    def decode_value(self, ipmicmd, value):
+    async def decode_value(self, ipmicmd, value):
         # Take the input value and return meaningful value
         linearization = self.linearization
         if linearization > 11:  # direct calling code to get factors
@@ -568,7 +568,7 @@ class SDREntry(object):
             # fashion.  However for now opt for retrieving rows as needed
             # rather than tracking all that information for a relatively
             # rare behavior
-            self._set_tmp_formula(ipmicmd, value)
+            await self._set_tmp_formula(ipmicmd, value)
             linearization = 0
         # time to compute the pre-linearization value.
         decoded = float((value * self.m + self.b)

confluent_server/aiohmi/redfish/command.py

@@ -1035,7 +1035,7 @@ class Command(object):
                 'Address': static_gateway,
             }]
         if patch:
-            nicurl = self._get_bmc_nic_url(name)
+            nicurl = await self._get_bmc_nic_url(name)
             await self._do_web_request(nicurl, patch, 'PATCH')
 
     async def set_net_configuration(self, ipv4_address=None, ipv4_configuration=None,
@@ -1458,22 +1458,22 @@ class Command(object):
                     del self.wc.stdheaders['Authorization']            
             return
         for vmurl in vmurls:
-            vminfo = self._do_web_request(vmurl, cache=False)
+            vminfo = await self._do_web_request(vmurl, cache=False)
             if vminfo.get('ConnectedVia', None) != 'NotConnected':
                 continue
             inserturl = vminfo.get(
                 'Actions', {}).get(
                     '#VirtualMedia.InsertMedia', {}).get('target', None)
             if inserturl:
-                self._do_web_request(inserturl, {'Image': url})
+                await self._do_web_request(inserturl, {'Image': url})
             else:
                 try:
-                    self._do_web_request(vmurl,
-                                         {'Image': url, 'Inserted': True},
-                                         'PATCH')
+                    await self._do_web_request(vmurl,
+                                               {'Image': url, 'Inserted': True},
+                                               'PATCH')
                 except exc.RedfishError as re:
                     if re.msgid.endswith(u'PropertyUnknown'):
-                        self._do_web_request(vmurl, {'Image': url}, 'PATCH')
+                        await self._do_web_request(vmurl, {'Image': url}, 'PATCH')
                     else:
                         raise
             break
@@ -1586,6 +1586,10 @@ class Command(object):
         return await oem.apply_license(filename, self, progress, data)
 
 if __name__ == '__main__':
-    print(repr(
-        Command(sys.argv[1], os.environ['BMCUSER'], os.environ['BMCPASS'],
-                verifycallback=lambda x: True).get_power()))
+    async def main():
+        cmd = await Command.create(
+            sys.argv[1], os.environ['BMCUSER'], os.environ['BMCPASS'],
+            verifycallback=lambda x: True)
+        print(repr(await cmd.get_power()))
+
+    asyncio.run(main())

confluent_server/aiohmi/redfish/oem/dell/idrac.py

@@ -17,8 +17,8 @@ import aiohmi.redfish.oem.generic as generic
 
 class OEMHandler(generic.OEMHandler):
 
-    def set_bootdev(self, bootdev, persist=False, uefiboot=None,
-                    fishclient=None):
+    async def set_bootdev(self, bootdev, persist=False, uefiboot=None,
+                          fishclient=None):
         # gleaned from web console, under configuration, system settings,
         # hardware, first boot device. iDrac presumes that the standard
         # explicitly refers only to physical devices. I think the intent
@@ -26,8 +26,8 @@ class OEMHandler(generic.OEMHandler):
         # the 'physical' standard to the vFDD/VCD-DVD seen in the idrac
         # web gui
         if bootdev not in ('floppy', 'cd'):
-            return super(OEMHandler, self).set_bootdev(bootdev, persist,
-                                                       uefiboot, fishclient)
+            return await super(OEMHandler, self).set_bootdev(bootdev, persist,
+                                                             uefiboot, fishclient)
         payload = {'Attributes': {}}
         if persist:
             payload['Attributes']['ServerBoot.1.BootOnce'] = 'Disabled'
@@ -37,7 +37,7 @@ class OEMHandler(generic.OEMHandler):
             payload['Attributes']['ServerBoot.1.FirstBootDevice'] = 'vFDD'
         elif bootdev == 'cd':
             payload['Attributes']['ServerBoot.1.FirstBootDevice'] = 'VCD-DVD'
-        fishclient._do_web_request(
+        await fishclient._do_web_request(
             '/redfish/v1/Managers/iDRAC.Embedded.1/Attributes',
             payload, method='PATCH')
         return {'bootdev': bootdev}

confluent_server/aiohmi/redfish/oem/generic.py

@@ -660,8 +660,8 @@ class OEMHandler(object):
                 procurl = sysinfo.get('Processors', {}).get('@odata.id',
                                                           None)
                 if procurl:
-                    for cpu in await fishclient._do_web_request(procurl).get(
-                            'Members', []):
+                    procinfo = await fishclient._do_web_request(procurl)
+                    for cpu in procinfo.get('Members', []):
                         cinfo = await fishclient._do_web_request(cpu['@odata.id'])
                         if cinfo.get('Status', {}).get(
                                 'State', None) == 'Absent':
@@ -1004,8 +1004,9 @@ class OEMHandler(object):
     def _extract_fwinfo(self, inf):
         return {}
 
-    def get_firmware_inventory(self, components, fishclient, category=None):
-        return []
+    async def get_firmware_inventory(self, components, fishclient, category=None):
+        return
+        yield
 
     def set_credentials(self, username, password):
         try:
@@ -1240,7 +1241,7 @@ class OEMHandler(object):
         if not foundmacs:
             # No PCIe device inventory, but *maybe* ethernet inventory...
             idxsbyname = {}
-            for nicinfo in self._get_eth_urls():
+            for nicinfo in await self._get_eth_urls():
                 nicinfo = await self._do_web_request(nicinfo)
                 nicname = nicinfo.get('Name', None)
                 nicinfo = nicinfo.get('MACAddress', nicinfo.get('PermanentAddress', None))
@@ -1675,7 +1676,7 @@ class OEMHandler(object):
         await fishclient._do_web_request(licenses, licinfo)
 
 
-    def get_user_expiration(self, uid):
+    async def get_user_expiration(self, uid):
         return None
 
     async def reseat_bay(self, bay):

confluent_server/aiohmi/redfish/oem/lenovo/main.py

@@ -39,7 +39,7 @@ async def get_handler(sysinfo, sysurl, webclient, cache, cmd, rootinfo={}):
     if not leninf:
         bmcinfo = await cmd.bmcinfo()
         if 'Ami' in bmcinfo.get('Oem', {}):
-            return tsma.TsmHandler(sysinfo, sysurl, webclient, cache)
+            return await tsma.TsmHandler.create(sysinfo, sysurl, webclient, cache, gpool=cmd._gpool)
     elif 'xclarity controller' in mgrinfo.get('Model', '').lower():
         if mgrinfo['Model'].endswith('3'):
             return await xcc3.OEMHandler.create(sysinfo, sysurl, webclient, cache,
@@ -60,9 +60,9 @@ async def get_handler(sysinfo, sysurl, webclient, cache, cmd, rootinfo={}):
         devdesc = await webclient.grab_json_response_with_status('/DeviceDescription.json')
         if devdesc[1] == 200:
             if devdesc[0]['type'].lower() in ('lenovo-smm3', 'smm3'):
-                return smm3.OEMHandler(sysinfo, sysurl, webclient, cache,
-                                      gpool=cmd._gpool)
+                return await smm3.OEMHandler.create(sysinfo, sysurl, webclient, cache,
+                                                    gpool=cmd._gpool)
     except Exception:
         pass
-    return generic.OEMHandler(sysinfo, sysurl, webclient, cache,
-                            gpool=cmd._gpool)
+    return await generic.OEMHandler.create(sysinfo, sysurl, webclient, cache,
+                                           gpool=cmd._gpool)

confluent_server/aiohmi/redfish/oem/lenovo/smm3.py

@@ -286,7 +286,8 @@ class OEMHandler(generic.OEMHandler):
         rsp = await self._do_web_request(url, method='POST')
 
     async def get_event_log(self, clear=False, fishclient=None):
-        return await super().get_event_log(clear, fishclient, extraurls=[{'@odata.id':'/redfish/v1/Chassis/chassis1/LogServices/EventLog'}])
+        async for event in super().get_event_log(clear, fishclient, extraurls=[{'@odata.id':'/redfish/v1/Chassis/chassis1/LogServices/EventLog'}]):
+            yield event
 
     async def get_description(self, fishclient):
         return {'height': 13, 'slot': 0, 'slots': [8, 2]}

confluent_server/aiohmi/redfish/oem/lenovo/tsma.py

@@ -101,8 +101,8 @@ class TsmHandler(generic.OEMHandler):
     @classmethod
     async def create(cls, sysinfo, sysurl, webclient, cache=None, fish=None,
                  gpool=None):
-        self = await super(TsmHandler, cls).create(sysinfo, sysurl, webclient, cache, fish,
-                                         gpool)
+        self = await super(TsmHandler, cls).create(sysinfo, sysurl, webclient, cache,
+                                                  gpool=gpool)
         if cache is None:
             cache = {}
         self._wc = None
@@ -186,7 +186,7 @@ class TsmHandler(generic.OEMHandler):
             if 'dns_domain'.startswith(key.lower()):
                 dnschgs['domain_name'] = currval
             if 'password_complexity'.startswith(key.lower()):
-                self._set_pass_complexity(currval, wc)
+                await self._set_pass_complexity(currval, wc)
             if 'password_login_failures'.startswith(key.lower()):
                 await self._set_pass_lockout(currval, wc)
         if dnschgs:
@@ -731,7 +731,7 @@ class TsmHandler(generic.OEMHandler):
             hddslots += 1
         else:
             raise exc.UnsupportedFunctionality('Unknown slot type requested')
-        gensettings = wc.grab_json_response('/api/settings/media/general')
+        gensettings = await wc.grab_json_response('/api/settings/media/general')
         samesettings = gensettings['same_settings'] == 1
         if samesettings:
             hds = gensettings['cd_remote_server_address']
@@ -772,10 +772,10 @@ class TsmHandler(generic.OEMHandler):
         gensettings['remote_media_support'] = 1
         gensettings['cd_remote_password'] = ''
         gensettings['hd_remote_password'] = ''
-        wc.grab_json_response_with_status('/api/settings/media/general',
+        await wc.grab_json_response_with_status('/api/settings/media/general',
                                           gensettings, method='PUT')
         # need to calibrate instances correctly
-        currinfo, status = wc.grab_json_response_with_status(
+        currinfo, status = await wc.grab_json_response_with_status(
             '/api/settings/media/instance')
         currinfo['num_cd'] = cdslots
         currinfo['num_hd'] = hddslots
@@ -783,9 +783,9 @@ class TsmHandler(generic.OEMHandler):
             currinfo['kvm_num_cd'] = cdslots
         if currinfo['kvm_num_hd'] > hddslots:
             currinfo['kvm_num_hd'] = hddslots
-        wc.grab_json_response_with_status(
+        await wc.grab_json_response_with_status(
             '/api/settings/media/instance', currinfo, method='PUT')
-        images = wc.grab_json_response('/api/settings/media/remote/images')
+        images = await wc.grab_json_response('/api/settings/media/remote/images')
         tries = 20
         while tries and not images:
             tries -= 1

confluent_server/aiohmi/redfish/oem/lenovo/xcc.py

@@ -184,7 +184,7 @@ class OEMHandler(generic.OEMHandler):
         try:
             self.fwo = await self.fwc.get_fw_options(fetchimm=fetchimm)
         except config.Unsupported:
-            return super(OEMHandler, self).get_system_configuration(
+            return await super(OEMHandler, self).get_system_configuration(
                 hideadvanced, fishclient)
         except Exception:
             raise Exception('%s failed to retrieve UEFI configuration'
@@ -831,7 +831,7 @@ class OEMHandler(generic.OEMHandler):
                 wc.set_header('X-XSRF-TOKEN', cookie.value)
             wc.vintage = util._monotonic_time()
 
-    def _make_available(self, disk, realcfg):
+    async def _make_available(self, disk, realcfg):
         # 8 if jbod, 4 if hotspare.., leave alone if already...
         currstatus = self._get_status(disk, realcfg)
         newstate = None
@@ -841,21 +841,21 @@ class OEMHandler(generic.OEMHandler):
             newstate = 4
         elif currstatus.lower() == 'jbod':
             newstate = 8
-        self._set_drive_state(disk, newstate)
+        await self._set_drive_state(disk, newstate)
 
-    def _make_jbod(self, disk, realcfg):
+    async def _make_jbod(self, disk, realcfg):
         currstatus = self._get_status(disk, realcfg)
         if currstatus.lower() == 'jbod':
             return
-        self._make_available(disk, realcfg)
-        self._set_drive_state(disk, 16)
+        await self._make_available(disk, realcfg)
+        await self._set_drive_state(disk, 16)
 
-    def _make_global_hotspare(self, disk, realcfg):
+    async def _make_global_hotspare(self, disk, realcfg):
         currstatus = self._get_status(disk, realcfg)
         if currstatus.lower() == 'global hot spare':
             return
-        self._make_available(disk, realcfg)
-        self._set_drive_state(disk, 1)
+        await self._make_available(disk, realcfg)
+        await self._set_drive_state(disk, 1)
 
     def _get_status(self, disk, realcfg):
         for cfgdisk in realcfg.disks:
@@ -978,7 +978,7 @@ class OEMHandler(generic.OEMHandler):
             pass
 
     async def _create_array(self, pool):
-        params = self._parse_array_spec(pool)
+        params = await self._parse_array_spec(pool)
         cid = params['controller'].split(',')[0]
         cslotno = params['controller'].split(',')[1]
         url = '/api/function/raid_conf?params=raidlink_GetDefaultVolProp'
@@ -1443,7 +1443,7 @@ class OEMHandler(generic.OEMHandler):
 
     async def update_firmware_backend(self, filename, data=None, progress=None,
                                 bank=None):
-        self._refresh_token()
+        await self._refresh_token()
         wc = await self.wc()
         rsv = await wc.grab_json_response('/api/providers/fwupdate', json.dumps(
             {'UPD_WebReserve': 1}))
@@ -1470,7 +1470,7 @@ class OEMHandler(generic.OEMHandler):
                     raise Exception('File is larger than supported')
                 raise Exception('Unexpected result:' + repr(rsp))
             uploadstate = rsp['state']
-            self._refresh_token()
+            await self._refresh_token()
         while uploadstate != 'done':
             rsp = await wc.grab_json_response(
                 '/upload/progress?X-Progress-ID={0}'.format(xid))
@@ -1757,18 +1757,18 @@ class OEMHandler(generic.OEMHandler):
                 else:
                     return days
 
-    def get_inventory_descriptions(self, withids=False):
-        hwmap = self.hardware_inventory_map()
+    async def get_inventory_descriptions(self, withids=False):
+        hwmap = await self.hardware_inventory_map()
         yield "System"
         for key in natural_sort(hwmap):
             yield key
-        for cpuinv in self._get_cpu_inventory():
+        async for cpuinv in self._get_cpu_inventory():
             yield cpuinv[0]
-        for meminv in self._get_mem_inventory():
+        async for meminv in self._get_mem_inventory():
             yield meminv[0]
 
 
-    def get_inventory_of_component(self, compname):
+    async def get_inventory_of_component(self, compname):
         if compname.lower() == 'system':
             sysinfo = {
                 'UUID': self._varsysinfo.get('UUID', ''),
@@ -1779,14 +1779,14 @@ class OEMHandler(generic.OEMHandler):
                     'SKU', self._varsysinfo.get('PartNumber', '')),
             }
             return sysinfo
-        hwmap = self.hardware_inventory_map()
+        hwmap = await self.hardware_inventory_map()
         try:
             return hwmap[compname]
         except KeyError:
-            for cpuinv in self._get_cpu_inventory():
+            async for cpuinv in self._get_cpu_inventory():
                 if cpuinv[0] == compname:
                     return cpuinv[1]
-            for meminv in self._get_mem_inventory():
+            async for meminv in self._get_mem_inventory():
                 if meminv[0] == compname:
                     return meminv[1]
 

confluent_server/aiohmi/redfish/oem/lenovo/xcc3.py

@@ -432,7 +432,7 @@ class OEMHandler(generic.OEMHandler):
             await self._make_available(disk, realcfg)
         self._urlcache.clear()
 
-    def _parse_array_spec(self, arrayspec):
+    async def _parse_array_spec(self, arrayspec):
         controller = None
         if arrayspec.disks:
             for disk in list(arrayspec.disks) + list(arrayspec.hotspares):
@@ -441,7 +441,7 @@ class OEMHandler(generic.OEMHandler):
                 if controller != disk.id[0]:
                     raise pygexc.UnsupportedFunctionality(
                         'Cannot span arrays across controllers')
-            raidmap = self._raid_number_map(controller)
+            raidmap = await self._raid_number_map(controller)
             if not raidmap:
                 raise pygexc.InvalidParameterValue(
                     'No RAID Type supported on this controller')
@@ -511,7 +511,7 @@ class OEMHandler(generic.OEMHandler):
         return themap
 
     async def _create_array(self, pool):
-        params = self._parse_array_spec(pool)
+        params = await self._parse_array_spec(pool)
         cid = params['controller'].split(',')[0]
         c_capabilities, code = await self.webclient.grab_json_response_with_status(
             f'/redfish/v1/Systems/1/Storage/{cid}/Volumes/Capabilities')
@@ -696,14 +696,14 @@ class OEMHandler(generic.OEMHandler):
             if write_policy:
                 request_data["WriteCachePolicy"] = write_policy
             
-            msg, code=self.webclient.grab_json_response_with_status(
+            msg, code = await self.webclient.grab_json_response_with_status(
                 f'/redfish/v1/Systems/1/Storage/{cid}/Volumes',
                 method='POST',
                 data=request_data)
             if code == 500 and not stripsize:
                     # Mystery error can be a mandatory strip size, default to 64k to match WebUI behavior
                     request_data["StripSizeBytes"] = 65536
-                    msg, code=self.webclient.grab_json_response_with_status(
+                    msg, code = await self.webclient.grab_json_response_with_status(
                         f'/redfish/v1/Systems/1/Storage/{cid}/Volumes',
                         method='POST',
                         data=request_data)
@@ -906,7 +906,7 @@ class OEMHandler(generic.OEMHandler):
                                                  cache=False)
         rawsettings = rawsettings.get('Attributes', {})
         pendingsettings = {}
-        ret = self._set_redfish_settings(
+        ret = await self._set_redfish_settings(
             changeset, fishclient, currsettings, rawsettings,
             pendingsettings, self.lenovobmcattrdeps, reginfo,
             '/redfish/v1/Managers/1/Oem/Lenovo/BMCSettings')
@@ -1039,7 +1039,7 @@ class OEMHandler(generic.OEMHandler):
         if usbsettings:
             await self.apply_usb_configuration(usbsettings)
         if bmchangeset:
-            self._set_xcc3_settings(bmchangeset, self)        
+            await self._set_xcc3_settings(bmchangeset, self)
 
     async def apply_usb_configuration(self, usbsettings):
         bmcattribs = {}
@@ -1220,5 +1220,3 @@ class OEMHandler(generic.OEMHandler):
                     'Name': 'HPM-FPGA Pending',
                     'build': pendinghpm}
         raise pygexc.BypassGenericBehavior()
-
-

confluent_server/aiohmi/redfish/oem/lookup.py

@@ -27,23 +27,23 @@ OEMMAP = {
 }
 
 
-def get_oem_handler(sysinfo, sysurl, webclient, cache, cmd, rootinfo={}):
+async def get_oem_handler(sysinfo, sysurl, webclient, cache, cmd, rootinfo={}):
     if rootinfo.get('Vendor', None) in OEMMAP:
-        return OEMMAP[rootinfo['Vendor']].get_handler(sysinfo, sysurl,
-                                                     webclient, cache, cmd, rootinfo)
+        return await OEMMAP[rootinfo['Vendor']].get_handler(sysinfo, sysurl,
+                                                            webclient, cache, cmd, rootinfo)
     for oem in sysinfo.get('Oem', {}):
         if oem in OEMMAP:
-            return OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache,
-                                           cmd, rootinfo)
+            return await OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache,
+                                                 cmd, rootinfo)
     for oem in sysinfo.get('Links', {}).get('OEM', []):
         if oem in OEMMAP:
-            return OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache,
-                                           cmd, rootinfo)
+            return await OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache,
+                                                 cmd, rootinfo)
     if rootinfo:  # rootinfo indicates early invocation, bmcinfo not ready yet
-        return generic.OEMHandler(sysinfo, sysurl, webclient, cache, cmd._gpool, rootinfo)    
-    bmcinfo = cmd.bmcinfo
+        return await generic.OEMHandler.create(sysinfo, sysurl, webclient, cache, cmd._gpool, rootinfo)
+    bmcinfo = await cmd.bmcinfo()
     for oem in bmcinfo.get('Oem', {}):
         if oem in OEMMAP:
-            return OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache,
-                                           cmd, rootinfo)
-    return generic.OEMHandler(sysinfo, sysurl, webclient, cache, cmd._gpool, rootinfo)
+            return await OEMMAP[oem].get_handler(sysinfo, sysurl, webclient, cache,
+                                                 cmd, rootinfo)
+    return await generic.OEMHandler.create(sysinfo, sysurl, webclient, cache, cmd._gpool, rootinfo)

confluent_server/confluent/asynchttp.py

@@ -57,8 +57,8 @@ class AsyncTermRelation(object):
         self.asynchdl = asynchdl
         self.termid = termid
 
-    def got_data(self, data):
-        self.asynchdl.add(self.termid, data)
+    async def got_data(self, data):
+        await self.asynchdl.add(self.termid, data)
 
 
 class AsyncSession(object):

confluent_server/confluent/collective/manager.py

@@ -187,7 +187,7 @@ async def connect_to_leader(cert=None, name=None, leader=None, remote=None, isre
                 await cfm.ConfigManager(tenant=None)._load_from_json(dbjson,
                                                                sync=False)
                 cfm.commit_clear()
-            except Exception:
+            except Exception as e:
                 print(repr(e))
                 await cfm.stop_following()
                 cfm.rollback_clear()
@@ -656,8 +656,8 @@ async def handle_connection(connection, cert, request, local=False):
             retrythread = None
         async with leader_init:
             cnn = tlvdata.get_socket(connection)
-            cfm.update_collective_address(request['name'],
-                                          cnn.getpeername()[0])
+            await cfm.update_collective_address(request['name'],
+                                                cnn.getpeername()[0])
             await tlvdata.send(connection, cfm._dump_keys(None, False))
             await tlvdata.send(connection, cfm._cfgstore['collective'])
             await tlvdata.send(connection, {'confluent_uuid': cfm.get_global('confluent_uuid')}) # cfm.get_globals())
@@ -895,7 +895,7 @@ async def check_managers():
                 targets = sorted(expandednoderanges[targets], key=availmanagers.get)
                 if not targets:
                     continue
-                c.set_node_attributes({node: {'collective.manager': {'value': targets[0]}}})
+                await c.set_node_attributes({node: {'collective.manager': {'value': targets[0]}}})
                 availmanagers[targets[0]] += 1
         await _assimilate_missing()
     failovercheck = None

confluent_server/confluent/config/configmanager.py

@@ -743,7 +743,7 @@ async def relay_slaved_requests(name, listener):
                 _newquorum = has_quorum()
             _hasquorum = _newquorum
             if _hasquorum and _pending_collective_updates:
-                apply_pending_collective_updates()
+                await apply_pending_collective_updates()
             msg = await lh.get_next_msg()
             while msg:
                 if name not in cfgstreams:
@@ -1050,23 +1050,23 @@ def _true_del_collective_member(name, sync=True):
 _pending_collective_updates = {}
 
 
-def update_collective_address(name ,address):
+async def update_collective_address(name, address):
     fprint = _cfgstore['collective'][name]['fingerprint']
     oldaddress = _cfgstore['collective'][name]['address']
     if oldaddress == address:
         return
     try:
         check_quorum()
-        add_collective_member(name, address, fprint)
+        await add_collective_member(name, address, fprint)
     except exc.DegradedCollective:
         _pending_collective_updates[name] = address
 
 
-def apply_pending_collective_updates():
+async def apply_pending_collective_updates():
     for name in list(_pending_collective_updates):
         fprint = _cfgstore['collective'][name]['fingerprint']
         address = _pending_collective_updates[name]
-        add_collective_member(name, address, fprint)
+        await add_collective_member(name, address, fprint)
         del _pending_collective_updates[name]
 
 
@@ -3232,7 +3232,7 @@ async def dump_db_to_directory(location, password, redact=None, skipkeys=False,
     try:
         for tenant in os.listdir(
                 os.path.join(ConfigManager._cfgdir, '/tenants/')):
-            tenant_data = ConfigManager(tenant=tenant)._dump_to_json(redact=redact)
+            tenant_data = await ConfigManager(tenant=tenant)._dump_to_json(redact=redact)
             with open(os.path.join(location, 'tenants', tenant, f'main.{format}'), 'wb' if format == 'json' else 'w') as cfgfile:
                 if format == 'json':
                     cfgfile.write(tenant_data)

confluent_server/confluent/consoleserver.py

@@ -225,13 +225,13 @@ class ConsoleHandler(object):
             'value', None)
         if list(configmodule.list_collective()) and not myc:
             self._is_local = False
-            self._detach()
+            await self._detach()
             await self._disconnect()
         if myc and myc != collective.get_myname():
             # Do not do console connect for nodes managed by another
             # confluent collective member
             self._is_local = False
-            self._detach()
+            await self._detach()
             await self._disconnect()
         else:
             self._is_local = True
@@ -301,9 +301,9 @@ class ConsoleHandler(object):
             'or the console simply not having any output since last connection]')
         self.clearpending = True
 
-    def _detach(self):
+    async def _detach(self):
         for ses in list(self.livesessions):
-            ses.detach()
+            await ses.detach()
 
     async def _disconnect(self):
         if self.connectionthread:
@@ -451,8 +451,7 @@ class ConsoleHandler(object):
         await self._send_rcpts({'deleting': True})
         await self._disconnect()
         if self._console:
-
-            self._console.close()
+            await self._console.close()
             self._console = None
         if self.connectionthread:
             self.connectionthread.cancel()
@@ -670,7 +669,7 @@ class ProxyConsole(object):
 
     def _attribschanged(self, nodeattribs, configmanager, **kwargs):
         if self.clisession:
-            self.clisession.detach()
+            tasks.spawn(self.clisession.detach())
             self.clisession = None
 
     async def relay_data(self):
@@ -743,15 +742,15 @@ class ProxyConsole(object):
                 pass
         self.clisession = None
 
-    def send_break(self):
-        tlvdata.send(self.remote, {'operation': 'break'})
+    async def send_break(self):
+        await tlvdata.send(self.remote, {'operation': 'break'})
 
-    def reopen(self):
-        tlvdata.send(self.remote, {'operation': 'reopen'})
+    async def reopen(self):
+        await tlvdata.send(self.remote, {'operation': 'reopen'})
 
     def resize(self, width, height):
-        tlvdata.send(self.remote, {'operation': 'resize', 'width': width,
-                                   'height': height})
+        tasks.spawn(tlvdata.send(self.remote, {'operation': 'resize', 'width': width,
+                                              'height': height}))
 
 
 # this represents some api view of a console handler.  This handles things like
@@ -825,10 +824,10 @@ class ConsoleSession(object):
         self.conshdl = await connect_node(self.node, self.configmanager,
                                     self.username, self.direct, self.width,
                                     self.height)
-    def send_break(self):
+    async def send_break(self):
         """Send break to remote system
         """
-        self.conshdl.send_break()
+        await self.conshdl.send_break()
 
     def resize(self, width, height):
         self.conshdl.resize(width, height)
@@ -871,7 +870,7 @@ class ConsoleSession(object):
             await self.conshdl.attachsession(self)
             self.write = self.conshdl.write
 
-    def got_data(self, data):
+    async def got_data(self, data):
         """Receive data from console and buffer
 
         If the caller does not provide a callback and instead will be polling

confluent_server/confluent/credserver.py

@@ -103,7 +103,7 @@ class CredServer(object):
                     now = datetime.datetime.utcnow()
                     expiry = datetime.datetime.strptime(apiarmed, "%Y-%m-%dT%H:%M:%SZ")
                     if now > expiry:
-                        self.cfm.set_node_attributes({nodename: {'deployment.apiarmed': ''}})
+                        await self.cfm.set_node_attributes({nodename: {'deployment.apiarmed': ''}})
                         client.close()
                         return
             await cloop.sock_sendall(client, b'\x02\x20')
@@ -133,7 +133,7 @@ class CredServer(object):
             await cloop.sock_recv(client, 2)  # drain end of message
             await cloop.sock_sendall(client, b'\x05\x00') # report success
             if hmackey and apiarmed != 'continuous':
-                self.cfm.clear_node_attributes([nodename], ['secret.selfapiarmtoken'])
+                await self.cfm.clear_node_attributes([nodename], ['secret.selfapiarmtoken'])
             if apiarmed != 'continuous':
                 disarm = {nodename: {'deployment.sealedapikey': '', 'deployment.apiarmed': ''}}
         finally:

confluent_server/confluent/discovery/core.py

@@ -580,14 +580,14 @@ async def handle_api_request(configmanager, inputdata, operation, pathcomponents
         return (msg.KeyValueData({'rescan': 'started'}),)
     elif operation in ('update', 'create') and pathcomponents[:2] == ['discovery', 'subscriptions']:
         target = pathcomponents[2]
-        affluent.subscribe_discovery(target, configmanager, collective.get_myname())
+        await affluent.subscribe_discovery(target, configmanager, collective.get_myname())
         currsubs = get_subscriptions()
         currsubs[target] = {}
         save_subscriptions(currsubs)
         return (msg.KeyValueData({'status': 'subscribed'}),)
     elif operation == 'delete' and pathcomponents[:2] == ['discovery', 'subscriptions']:
         target = pathcomponents[2]
-        affluent.unsubscribe_discovery(target, configmanager, collective.get_myname())
+        await affluent.unsubscribe_discovery(target, configmanager, collective.get_myname())
         currsubs = get_subscriptions()
         if target in currsubs:
             del currsubs[target]
@@ -597,7 +597,7 @@ async def handle_api_request(configmanager, inputdata, operation, pathcomponents
         if pathcomponents == ['discovery', 'register']:
             if 'addresses' not in inputdata:
                 raise exc.InvalidArgumentException('Missing address in input')
-            return await register_remote_addrs(inputdata['addresses'], configmanager)
+            return register_remote_addrs(inputdata['addresses'], configmanager)
         if 'node' not in inputdata:
             raise exc.InvalidArgumentException('Missing node name in input')
         mac = _get_mac_from_query(pathcomponents)
@@ -920,7 +920,7 @@ async def detected(info):
                     else:
                         policies = set([])
                     if policies & {'open', 'permissive'}:
-                        cfg.set_node_attributes({nodename: {'id.uuid': info['uuid']}})
+                        await cfg.set_node_attributes({nodename: {'id.uuid': info['uuid']}})
             return  # already known, no need for more
     #TODO(jjohnson2): We might have to get UUID for certain searches...
     #for now defer probe until inside eval_node.  We might not have
@@ -975,7 +975,7 @@ async def get_chained_smm_name(nodename, cfg, handler, nl=None, checkswitch=True
     mycert = await handler.get_https_cert()
     if checkswitch:
         fprints = macmap.get_node_fingerprints(nodename, cfg)
-        for fprint in fprints:
+        async for fprint in fprints:
             if util.cert_matches(fprint[0], mycert):
                 # ok we have a direct match, it is this node
                 return nodename, fprint[1]
@@ -1380,7 +1380,7 @@ async def eval_node(cfg, handler, info, nodename, manual=False):
                     # The candidate nodename is the head of a chain, we must
                     # validate the smm certificate by the switch
                     fprints = macmap.get_node_fingerprints(nodename, cfg)
-                    for fprint in fprints:
+                    async for fprint in fprints:
                         if util.cert_matches(fprint[0], await handler.get_https_cert()):
                             if not await discover_node(cfg, handler, info,
                                                  nodename, manual):
@@ -1434,7 +1434,7 @@ async def discover_node(cfg, handler, info, nodename, manual):
     # in some product or another.
     curruuid = info.get('uuid', False)
     if 'pxe' in policies and info['handler'] == pxeh:
-        return do_pxe_discovery(cfg, handler, info, manual, nodename, policies)
+        return await do_pxe_discovery(cfg, handler, info, manual, nodename, policies)
     elif ('permissive' in policies and handler.https_supported and lastfp and
             not util.cert_matches(lastfp, await handler.get_https_cert()) and not manual):
         info['discofailure'] = 'fingerprint'
@@ -1455,7 +1455,7 @@ async def discover_node(cfg, handler, info, nodename, manual):
                 return False
         info['nodename'] = nodename
         if info['handler'] == pxeh:
-            return do_pxe_discovery(cfg, handler, info, manual, nodename, policies)
+            return await do_pxe_discovery(cfg, handler, info, manual, nodename, policies)
         elif manual or not util.cert_matches(lastfp, await handler.get_https_cert()):
             # only 'discover' if it is not the same as last time
             try:
@@ -1554,7 +1554,7 @@ async def wait_for_connection(bmcaddr):
                 continue
         await asyncio.sleep(1)
 
-def do_pxe_discovery(cfg, handler, info, manual, nodename, policies):
+async def do_pxe_discovery(cfg, handler, info, manual, nodename, policies):
     # use uuid based scheme in lieu of tls cert, ideally only
     # for stateless 'discovery' targets like pxe, where data does not
     # change
@@ -1585,7 +1585,7 @@ def do_pxe_discovery(cfg, handler, info, manual, nodename, policies):
             for checkattr in attribs:
                 checkval = currattrs.get(nodename, {}).get(checkattr, {}).get('value', None)
                 if checkval != attribs[checkattr]:
-                    cfg.set_node_attributes({nodename: attribs})
+                    await cfg.set_node_attributes({nodename: attribs})
                     break
     if info['uuid'] in known_pxe_uuids:
         return True

confluent_server/confluent/discovery/handlers/bmc.py

@@ -48,8 +48,8 @@ class NodeHandler(generic.NodeHandler):
         return
         # TODO(jjohnson2): probe serial number and uuid
 
-    def config(self, nodename, reset=False):
-        self._bmcconfig(nodename, reset)
+    async def config(self, nodename, reset=False):
+        await self._bmcconfig(nodename, reset)
 
     async def _bmcconfig(self, nodename, reset=False, customconfig=None, vc=None):
         # TODO(jjohnson2): set ip parameters, user/pass, alert cfg maybe
@@ -108,7 +108,7 @@ class NodeHandler(generic.NodeHandler):
                 # Use existing account that has been created
                 newuserslot = uid
                 if newpass != passwd:  # don't mess with existing if no change
-                    ic.set_user_password(newuserslot, password=newpass)
+                    await ic.set_user_password(newuserslot, password=newpass)
                     ic = await self._get_ipmicmd(user, passwd)
                     if vc:
                         ic.register_key_handler(vc)
@@ -118,8 +118,8 @@ class NodeHandler(generic.NodeHandler):
             if newuserslot < 2:
                 newuserslot = 2
             if newpass != passwd:  # don't mess with existing if no change
-                ic.set_user_password(newuserslot, password=newpass)
-            ic.set_user_name(newuserslot, newuser)
+                await ic.set_user_password(newuserslot, password=newpass)
+            await ic.set_user_name(newuserslot, newuser)
             if havecustomcreds:
                 ic = await self._get_ipmicmd(user, passwd)
                 if vc:
@@ -132,17 +132,17 @@ class NodeHandler(generic.NodeHandler):
         for uid in currusers:
             if uid != newuserslot:
                 if uid <= lockedusers:  # we cannot delete, settle for disable
-                    ic.disable_user(uid, 'disable')
+                    await ic.disable_user(uid, 'disable')
                 else:
                     # lead with the most critical thing, removing user access
-                    ic.set_user_access(uid, channel=None, callback=False,
-                                       link_auth=False, ipmi_msg=False,
-                                       privilege_level='no_access')
+                    await ic.set_user_access(uid, channel=None, callback=False,
+                                             link_auth=False, ipmi_msg=False,
+                                             privilege_level='no_access')
                     # next, try to disable the password
-                    ic.set_user_password(uid, mode='disable', password=None)
+                    await ic.set_user_password(uid, mode='disable', password=None)
                     # ok, now we can be less paranoid
                     try:
-                        ic.user_delete(uid)
+                        await ic.user_delete(uid)
                     except pygexc.IpmiException as ie:
                         if ie.ipmicode != 0xd5:  # some response to the 0xff
                             # name...
@@ -165,20 +165,20 @@ class NodeHandler(generic.NodeHandler):
             netconfig = await netutil.get_nic_config(cfg, nodename, ip=newip)
             plen = netconfig['prefix']
             newip = '{0}/{1}'.format(newip, plen)
-            currcfg = ic.get_net_configuration()
+            currcfg = await ic.get_net_configuration()
             if currcfg['ipv4_address'] != newip:
                 # do not change the ipv4_config if the current config looks
                 # like it is already accurate
-                ic.set_net_configuration(ipv4_address=newip,
-                                         ipv4_configuration='static',
-                                         ipv4_gateway=netconfig[
-                                             'ipv4_gateway'])
+                await ic.set_net_configuration(ipv4_address=newip,
+                                                ipv4_configuration='static',
+                                                ipv4_gateway=netconfig[
+                                                    'ipv4_gateway'])
         elif self.ipaddr.startswith('fe80::'):
-            cfg.set_node_attributes(
+            await cfg.set_node_attributes(
                 {nodename: {'hardwaremanagement.manager': self.ipaddr}})
         else:
             raise exc.TargetEndpointUnreachable(
                 'hardwaremanagement.manager must be set to desired address')
         if reset:
-            ic.reset_bmc()
+            await ic.reset_bmc()
         return ic

confluent_server/confluent/discovery/handlers/imm.py

@@ -89,13 +89,13 @@ class NodeHandler(bmchandler.NodeHandler):
             guiddata = await ipmicmd.xraw_command(netfn=6, command=8)
             self.info['uuid'] = pygutil.decode_wireformat_uuid(
                 guiddata['data']).lower()
-            ipmicmd.oem_init()
-            bayid = ipmicmd._oem.immhandler.get_property(
+            await ipmicmd.oem_init()
+            bayid = await ipmicmd._oem.immhandler.get_property(
                 '/v2/cmm/sp/7')
             if not bayid:
                 return
             self.info['enclosure.bay'] = int(bayid)
-            smmid = ipmicmd._oem.immhandler.get_property(
+            smmid = await ipmicmd._oem.immhandler.get_property(
                 '/v2/ibmc/smm/chassis/uuid')
             if not smmid:
                 return
@@ -108,4 +108,3 @@ class NodeHandler(bmchandler.NodeHandler):
         except pygexc.IpmiException as ie:
             print(repr(ie))
             raise
-

confluent_server/confluent/discovery/handlers/redfishbmc.py

@@ -266,8 +266,9 @@ class NodeHandler(generic.NodeHandler):
             newip = newipinfo[-1][0]
             if ':' in newip:
                 raise exc.NotImplementedException('IPv6 remote config TODO')
-            hifurls = await get_host_interface_urls(wc, self.mgrinfo(wc))
-            mgtnicinfo = self.mgrinfo(wc)['EthernetInterfaces']['@odata.id']
+            mgrinfo = await self.mgrinfo(wc)
+            hifurls = await get_host_interface_urls(wc, mgrinfo)
+            mgtnicinfo = mgrinfo['EthernetInterfaces']['@odata.id']
             mgtnicinfo = await wc.grab_json_response(mgtnicinfo)
             mgtnics = [x['@odata.id'] for x in mgtnicinfo.get('Members', [])]
             actualnics = []

confluent_server/confluent/discovery/handlers/smm.py

@@ -99,7 +99,7 @@ class NodeHandler(bmchandler.NodeHandler):
         if smmip:
             smmip = smmip.split('/', 1)[0]
         if smmip and ':' not in smmip:
-            smmip = await asyncio.get_running_loop().getaddrinfo(smmip, 0)[0]
+            smmip = (await asyncio.get_running_loop().getaddrinfo(smmip, 0))[0]
             smmip = smmip[-1][0]
             if smmip and ':' in smmip:
                 raise exc.NotImplementedException('IPv6 not supported')
@@ -125,7 +125,7 @@ class NodeHandler(bmchandler.NodeHandler):
         if smmip and ':' in smmip and not smmip.startswith('fe80::'):
             raise exc.NotImplementedException('IPv6 configuration TODO')
         if self.ipaddr.startswith('fe80::'):
-            cfg.set_node_attributes(
+            await cfg.set_node_attributes(
                     {nodename: {'hardwaremanagement.manager': self.ipaddr}})
 
     def _webconfigcreds(self, username, password):

confluent_server/confluent/discovery/handlers/smm3.py

@@ -46,7 +46,7 @@ async def remote_nodecfg(nodename, cfm):
     ipaddr = cfg.get(nodename, {}).get('hardwaremanagement.manager', {}).get(
         'value', None)
     ipaddr = ipaddr.split('/', 1)[0]
-    ipaddr = await asyncio.get_running_loop().getaddrinfo(ipaddr, 0)[0][-1]
+    ipaddr = (await asyncio.get_running_loop().getaddrinfo(ipaddr, 0))[0][-1]
     if not ipaddr:
         raise Exception('Cannot remote configure a system without known '
                         'address')
@@ -63,4 +63,3 @@ if __name__ == '__main__':
     print(repr(info))
     testr = NodeHandler(info, c)
     asyncio.run(testr.config(sys.argv[2]))
-

confluent_server/confluent/discovery/handlers/tsm.py

@@ -225,7 +225,7 @@ class NodeHandler(generic.NodeHandler):
         rsp, status = await wc.grab_json_response_with_status('/api/session', method='DELETE')
 
 
-def remote_nodecfg(nodename, cfm):
+async def remote_nodecfg(nodename, cfm):
     cfg = cfm.get_node_attributes(
             nodename, 'hardwaremanagement.manager')
     ipaddr = cfg.get(nodename, {}).get('hardwaremanagement.manager', {}).get(
@@ -237,7 +237,7 @@ def remote_nodecfg(nodename, cfm):
                          'address')
     info = {'addresses': [ipaddr]}
     nh = NodeHandler(info, cfm)
-    nh.config(nodename)
+    await nh.config(nodename)
 
 if __name__ == '__main__':
     import confluent.config.configmanager as cfm

confluent_server/confluent/discovery/handlers/xcc.py

@@ -171,8 +171,8 @@ class NodeHandler(immhandler.NodeHandler):
                     {'IPMI': {'ProtocolEnabled': True}}, method='PATCH')
         ipmicmd = None
         try:
-            ipmicmd = self._get_ipmicmd(self._currcreds[0], self._currcreds[1])
-            ipmicmd.xraw_command(netfn=0x3a, command=0xf1, data=(1,))
+            ipmicmd = await self._get_ipmicmd(self._currcreds[0], self._currcreds[1])
+            await ipmicmd.xraw_command(netfn=0x3a, command=0xf1, data=(1,))
         except pygexc.IpmiException as e:
             if (e.ipmicode != 193 and 'Unauthorized name' not in str(e) and
                     'Incorrect password' not in str(e) and
@@ -711,4 +711,3 @@ async def remote_nodecfg(nodename, cfm):
     else:
         nh = xcc3handler.NodeHandler(info, cfm)
     await nh.config(nodename)
-

confluent_server/confluent/discovery/protocols/mdns.py

@@ -336,7 +336,7 @@ async def check_fish(urldata, port=443, verifycallback=None):
         except (IndexError, KeyError):
             return None
             url = '/redfish/v1/'
-            peerinfo = wc.grab_json_response('/redfish/v1/')
+            peerinfo = await wc.grab_json_response('/redfish/v1/')
     if url == '/redfish/v1/':
         if 'UUID' in peerinfo:
             if 'services' not in data:

confluent_server/confluent/discovery/protocols/ssdp.py

@@ -298,7 +298,7 @@ async def snoop(handler, byehandler=None, protocol=None, uuidlookup=None):
                                     if await netutil.ip_on_same_subnet(theip, 'fe80::', 64):
                                         if '%' in peer[0]:
                                             ifidx = peer[0].split('%', 1)[1]
-                                            iface = await cloop.getaddrinfo(peer[0], 0, socket.AF_INET6, socket.SOCK_DGRAM)[0][-1][-1]
+                                            iface = (await cloop.getaddrinfo(peer[0], 0, socket.AF_INET6, socket.SOCK_DGRAM))[0][-1][-1]
                                         else:
                                             ifidx = '{}'.format(peer[-1])
                                             iface = peer[-1]
@@ -502,7 +502,7 @@ async def check_fish(urldata, port=443, verifycallback=None):
     if url == '/DeviceDescription.json':
         if not peerinfo:
             if data.get('services', None) == ['urn::dmtf-org:service:redfish-rest:']:
-                peerinfo = wc.grab_json_response('/redfish/v1/')
+                peerinfo = await wc.grab_json_response('/redfish/v1/')
                 if peerinfo:
                     data['services'] = ['lenovo-smm3']
                     data['uuid'] = peerinfo['UUID'].lower()
@@ -609,5 +609,3 @@ if __name__ == '__main__':
     def printit(rsp):
         pass # print(repr(rsp))
     asyncio.run(active_scan(printit))
-
-

confluent_server/confluent/httpapi.py

@@ -571,7 +571,7 @@ async def wsock_handler(req):
                                     width=msg['width'], height=msg['height'])
                             if action == 'break':
                                 clientsessid = '{0}'.format(msg['sessid'])
-                                myconsoles[clientsessid].send_break()
+                                await myconsoles[clientsessid].send_break()
                             elif action == 'stop':
                                 sessid = '{0}'.format(msg.get('sessid', None))
                                 if sessid in myconsoles:
@@ -585,21 +585,24 @@ async def wsock_handler(req):
             if asess:
                 await asess.destroy()
         return rsp
-    if '/console/session' in ws.path or '/shell/sessions/' in ws.path:
-        def datacallback(data):
-            ws.send(websockify_data(data))
-        geom = ws.wait()
-        geom = geom[1:]
+    path = req.rel_url.path
+    if '/console/session' in path or '/shell/sessions/' in path:
+        async def datacallback(data):
+            await rsp.send_str(websockify_data(data))
+        geom = await rsp.receive()
+        if geom.type != WSMsgType.TEXT:
+            return rsp
+        geom = geom.data[1:]
         geom = json.loads(geom)
         width = geom['width']
         height = geom['height']
         skipreplay = geom.get('skipreplay', False)
         #hard bake JSON into this path, do not support other incarnations
-        if '/console/session' in ws.path:
-            prefix, _, _ = ws.path.partition('/console/session')
+        if '/console/session' in path:
+            prefix, _, _ = path.partition('/console/session')
             shellsession = False
-        elif '/shell/sessions/' in ws.path:
-            prefix, _, _ = ws.path.partition('/shell/sessions')
+        elif '/shell/sessions/' in path:
+            prefix, _, _ = path.partition('/shell/sessions')
             shellsession = True
         _, _, nodename = prefix.rpartition('/')
 
@@ -618,24 +621,26 @@ async def wsock_handler(req):
                 )
         except exc.NotFoundException:
             return
-        clientmsg = ws.wait()
+        clientmsg = await rsp.receive()
         try:
-            while clientmsg is not None:
+            while clientmsg.type == WSMsgType.TEXT:
+                clientmsg = clientmsg.data
                 if clientmsg[0] == ' ':
-                    consession.write(clientmsg[1:])
+                    await consession.write(clientmsg[1:])
                 elif clientmsg[0] == '!':
                     cmd = json.loads(clientmsg[1:])
                     action = cmd.get('action', None)
                     if action == 'break':
-                        consession.send_break()
+                        await consession.send_break()
                     elif action == 'resize':
                         consession.resize(
                             width=cmd['width'], height=cmd['height'])
                 elif clientmsg[0] == '?':
-                    ws.send(u'?')
-                clientmsg = ws.wait()
+                    await rsp.send_str(u'?')
+                clientmsg = await rsp.receive()
         finally:
-            consession.destroy()
+            await consession.destroy()
+        return rsp
 
 
 async def resourcehandler(request):
@@ -927,14 +932,14 @@ async def resourcehandler_backend(req, make_response):
             await rsp.write(json.dumps({'session': querydict['session']}))
             return rsp # client has requests to send or receive, not both...
         elif 'closesession' in querydict:
-            consolesessions[querydict['session']]['session'].destroy()
+            await consolesessions[querydict['session']]['session'].destroy()
             del consolesessions[querydict['session']]
             rsp = await make_response('application/json', 200)
             await rsp.write(b'{"sessionclosed": true}')
             return rsp
         elif 'action' in querydict:
             if querydict['action'] == 'break':
-                consolesessions[querydict['session']]['session'].send_break()
+                await consolesessions[querydict['session']]['session'].send_break()
             elif querydict['action'] == 'resize':
                 consolesessions[querydict['session']]['session'].resize(
                     width=querydict['width'], height=querydict['height'])

confluent_server/confluent/networking/lldp.py

@@ -132,8 +132,8 @@ def b64tohex(b64str):
     bd = bytearray(bd)
     return ''.join(['{0:02x}'.format(x) for x in bd])
 
-def get_fingerprint(switch, port, configmanager, portmatch):
-    update_switch_data(switch, configmanager)
+async def get_fingerprint(switch, port, configmanager, portmatch):
+    await update_switch_data(switch, configmanager)
     for neigh in _neighbypeerid:
         info = _neighbypeerid[neigh]
         if neigh == '!!vintage' or info.get('switch', None) != switch:
@@ -492,4 +492,3 @@ async def _handle_neighbor_query(pathcomponents, configmanager):
             if isinstance(x, Exception):
                 raise x
     return list_info(parms, listrequested)
-

confluent_server/confluent/networking/macmap.py

@@ -255,6 +255,7 @@ async def _start_offloader():
 
 
 async def _recv_offload():
+    global _offloader
     try:
         upacker = msgpack.Unpacker(encoding='utf8')
     except TypeError:
@@ -262,6 +263,15 @@ async def _recv_offload():
     #instream = _offloader.stdout.fileno()
     while True:
         datum = await _offloader.stdout.read(512)
+        if not datum:
+            _offloader = None
+            pending = list(_offloadevts.values())
+            _offloadevts.clear()
+            for future in pending:
+                if not future.done():
+                    future.set_exception(
+                        RuntimeError('MAC map offload process exited'))
+            return
         upacker.feed(datum)
         for result in upacker:
             if result[0] not in _offloadevts:
@@ -628,15 +638,15 @@ async def handle_api_request(configmanager, inputdata, operation, pathcomponents
             operation, '/'.join(pathcomponents)))
 
 
-def get_node_fingerprints(nodename, configmanager):
+async def get_node_fingerprints(nodename, configmanager):
     cfg = configmanager.get_node_attributes(nodename, ['net*.switch',
                                                        'net*.switchport'])
     for attrkey in cfg[nodename]:
         if attrkey.endswith('switch'):
             switch = cfg[nodename][attrkey]['value']
             port = cfg[nodename][attrkey + 'port']['value']
-            yield get_fingerprint(switch, port, configmanager,
-                                       _namesmatch)
+            yield await get_fingerprint(switch, port, configmanager,
+                                        _namesmatch)
 
 
 async def handle_read_api_request(pathcomponents, configmanager):

confluent_server/confluent/plugins/configuration/attributes.py

@@ -254,10 +254,10 @@ def yield_rename_resources(namemap, isnode):
         else:
             yield msg.RenamedResource(node, namemap[node])
 
-def update_locks(nodes, configmanager, inputdata):
+async def update_locks(nodes, configmanager, inputdata):
     for node in nodes:
         updatestate = inputdata.inputbynode[node]
-        configmanager.set_node_attributes({node: {'deployment.lock': updatestate}})
+        await configmanager.set_node_attributes({node: {'deployment.lock': updatestate}})
         yield msg.DeploymentLock(node, updatestate)
 
 async def update_nodes(nodes, element, configmanager, inputdata):

confluent_server/confluent/plugins/deployment/identimage.py

@@ -46,7 +46,7 @@ async def create_ident_image(node, configmanager):
     tmpd = tempfile.mkdtemp()
     ident = { 'nodename': node }
     apikey = create_apikey()
-    configmanager.set_node_attributes({node: {'secret.selfapiarmtoken': apikey}})
+    await configmanager.set_node_attributes({node: {'secret.selfapiarmtoken': apikey}})
     ident['apitoken'] = apikey
     # This particular mechanism does not (yet) do anything smart with collective
     # It would be a reasonable enhancement to list all collective server addresses
@@ -79,4 +79,3 @@ async def update(nodes, element, configmanager, inputdata):
         yield msg.CreatedResource(
                 'nodes/{0}/deployment/ident_image'.format(node))
 
-

confluent_server/confluent/plugins/hardwaremanagement/affluent.py

@@ -72,10 +72,10 @@ async def renotify_me(node, configmanager, myname):
     creds = configmanager.get_node_attributes(
         node, ['secret.hardwaremanagementuser', 'secret.hardwaremanagementpassword'], decrypt=True)
     wc = WebClient(node, configmanager, creds)
-    res, status = wc.wc.grab_json_response_with_status('/affluent/systems/renotify', {'subscriber': myname})
+    await wc.wc.grab_json_response_with_status('/affluent/systems/renotify', {'subscriber': myname})
 
 
-def subscribe_discovery(node, configmanager, myname):
+async def subscribe_discovery(node, configmanager, myname):
     creds = configmanager.get_node_attributes(
         node, ['secret.hardwaremanagementuser', 'secret.hardwaremanagementpassword'], decrypt=True)
     tsock = socket.create_connection((node, 443))
@@ -87,19 +87,19 @@ def subscribe_discovery(node, configmanager, myname):
     wc = WebClient(node, configmanager, creds)
     with open('/etc/confluent/tls/cacert.pem') as cain:
         cacert = cain.read()
-    wc.wc.grab_json_response('/affluent/cert_authorities/{0}'.format(myname), cacert)
-    res, status = wc.wc.grab_json_response_with_status('/affluent/discovery_subscribers/{0}'.format(myname), {'url': myurl, 'authname': node})
+    await wc.wc.grab_json_response('/affluent/cert_authorities/{0}'.format(myname), cacert)
+    res, status = await wc.wc.grab_json_response_with_status('/affluent/discovery_subscribers/{0}'.format(myname), {'url': myurl, 'authname': node})
     if status == 200:
         agentkey = res['cryptkey']
-        configmanager.set_node_attributes({node: {'crypted.selfapikey': {'hashvalue': agentkey}}})
-    res, status = wc.wc.grab_json_response_with_status('/affluent/systems/renotify', {'subscriber': myname})
+        await configmanager.set_node_attributes({node: {'crypted.selfapikey': {'hashvalue': agentkey}}})
+    await wc.wc.grab_json_response_with_status('/affluent/systems/renotify', {'subscriber': myname})
 
-def unsubscribe_discovery(node, configmanager, myname):
+async def unsubscribe_discovery(node, configmanager, myname):
     creds = configmanager.get_node_attributes(
         node, ['secret.hardwaremanagementuser', 'secret.hardwaremanagementpassword'], decrypt=True)
     wc = WebClient(node, configmanager, creds)
-    res, status = wc.wc.grab_json_response_with_status('/affluent/cert_authorities/{0}'.format(myname), method='DELETE')
-    res, status = wc.wc.grab_json_response_with_status('/affluent/discovery_subscribers/{0}'.format(myname), method='DELETE')
+    await wc.wc.grab_json_response_with_status('/affluent/cert_authorities/{0}'.format(myname), method='DELETE')
+    await wc.wc.grab_json_response_with_status('/affluent/discovery_subscribers/{0}'.format(myname), method='DELETE')
 
 
 def update(nodes, element, configmanager, inputdata):

confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py

@@ -163,7 +163,7 @@ class PDUClient(object):
         self.wc.grab_response('/logout_wait.htm')
 
     async def get_outlet(self, outlet):
-        rsp = await self.wc.grab_response('/setting_admin4.xml')
+        rsp = self.wc.grab_response('/setting_admin4.xml')
         xd = fromstring(rsp[0])
         for ch in xd:
             if 'relay' not in ch.tag:
@@ -177,7 +177,7 @@ class PDUClient(object):
         outlet = int(outlet)
         ident = self.map_outlets[outlet]
         sitem = '/SetParm?item={}?content={}'.format(ident, state)
-        await self.wc.grab_response(sitem)
+        self.wc.grab_response(sitem)
 
 async def retrieve(nodes, element, configmanager, inputdata):
     if 'outlets' not in element:

confluent_server/confluent/plugins/hardwaremanagement/ipmi.py

@@ -554,16 +554,16 @@ class IpmiHandler:
             if (self.error == 'timeout' or
                     'Insufficient resources' in self.error):
                 self.error = self.error.replace(' reported in RAKP4', '')
-                self.output.put(msg.ConfluentTargetTimeout(
+                await self.output.put(msg.ConfluentTargetTimeout(
                     self.node, self.error))
                 return
             elif 'Invalid Session ID' in self.error:
-                self.output.put(msg.ConfluentTargetTimeout(
+                await self.output.put(msg.ConfluentTargetTimeout(
                     self.node, 'Temporary Login Error'))
                 return
             elif ('Unauthorized' in self.error or
                     'Incorrect password' in self.error):
-                self.output.put(
+                await self.output.put(
                     msg.ConfluentTargetInvalidCredentials(self.node))
                 return
             else:
@@ -708,7 +708,7 @@ class IpmiHandler:
             if tmpvarbind.endswith('3183.1.1'):
                 varbinddata = inputdata[tmpvarbind]
         varbinddata = hex2bin(varbinddata)
-        event = self.ipmicmd.decode_pet(specifictrap, varbinddata)
+        event = await self.ipmicmd.decode_pet(specifictrap, varbinddata)
         self.pyghmi_event_to_confluent(event)
         await self.output.put(msg.EventCollection((event,), name=self.node))
 
@@ -1007,7 +1007,7 @@ class IpmiHandler:
         if activeupdates:
             await self.output.put(msg.KeyValueData({'status': 'active'}, self.node))
         else:
-            status = self.ipmicmd.get_update_status()
+            status = await self.ipmicmd.get_update_status()
             await self.output.put(msg.KeyValueData({'status': status}, self.node))
 
     async def handle_inventory(self):
@@ -1025,12 +1025,12 @@ class IpmiHandler:
 
     async def list_leds(self):
         await self.output.put(msg.ChildCollection('all'))
-        for category, info in self.ipmicmd.get_leds():
+        async for category, info in self.ipmicmd.get_leds():
             await self.output.put(msg.ChildCollection(simplify_name(category)))
 
     async def read_leds(self, component):
         led_categories = []
-        for category, info in self.ipmicmd.get_leds():
+        async for category, info in self.ipmicmd.get_leds():
             if component == 'all' or component == simplify_name(category):
                 led_categories.append({category: info})
         await self.output.put(msg.LEDStatus(led_categories, self.node))
@@ -1054,7 +1054,7 @@ class IpmiHandler:
                 await self.make_inventory_map()
                 compname = self.invmap.get(component, None)
                 if compname is None:
-                    self.output.put(msg.ConfluentTargetNotFound())
+                    await self.output.put(msg.ConfluentTargetNotFound())
                     return
                 invdata = await self.ipmicmd.get_inventory_of_component(compname)
                 if invdata is None:
@@ -1118,7 +1118,7 @@ class IpmiHandler:
         if len(storelem) < 2 or storelem[0] != 'volumes':
             raise exc.InvalidArgumentException('Must target a specific volume')
         volname = storelem[-1]
-        curr = self.ipmicmd.get_storage_configuration()
+        curr = await self.ipmicmd.get_storage_configuration()
         volumes = []
         volsfound = False
         toremove = storage.ConfigSpec(arrays=[storage.Array(volumes=volumes)])
@@ -1131,7 +1131,7 @@ class IpmiHandler:
             await self.output.put(msg.ConfluentTargetNotFound(
                 self.node, "No volume named '{0}' found".format(volname)))
             return
-        self.ipmicmd.remove_storage_configuration(toremove)
+        await self.ipmicmd.remove_storage_configuration(toremove)
         await self.output.put(msg.DeletedResource(volname))
 
     async def _create_storage(self, storelem):
@@ -1139,7 +1139,7 @@ class IpmiHandler:
             raise exc.InvalidArgumentException('Can only create volumes')
         vols = []
         thedisks = None
-        currcfg = self.ipmicmd.get_storage_configuration()
+        currcfg = await self.ipmicmd.get_storage_configuration()
         currnames = []
         for arr in currcfg.arrays:
             arrname = '{0}-{1}'.format(*arr.id)
@@ -1345,7 +1345,7 @@ class IpmiHandler:
 
     async def list_sensors(self):
         try:
-            sensors = await self.ipmicmd.get_sensor_descriptions()
+            sensors = [sensor async for sensor in self.ipmicmd.get_sensor_descriptions()]
         except pygexc.IpmiException:
             await self.output.put(msg.ConfluentTargetTimeout(self.node))
             return
@@ -1668,7 +1668,7 @@ class IpmiHandler:
     async def handle_ikvm(self):
         methods = await self.ipmicmd.get_ikvm_methods()
         if 'openbmc' in methods:
-            url = vinzmanager.get_url(self.node, self.inputdata)
+            url = await vinzmanager.get_url(self.node, self.inputdata)
             await self.output.put(msg.ChildCollection(url))
             return
         launchdata = await self.ipmicmd.get_ikvm_launchdata()
@@ -1760,4 +1760,3 @@ def delete(nodes, element, configmanager, inputdata):
                                               element, type='ffdc')
     return perform_requests(
         'delete', nodes, element, configmanager, inputdata, 'delete')
-

confluent_server/confluent/plugins/hardwaremanagement/pdu.py

@@ -71,9 +71,10 @@ async def readpdu(pdu, outletmap, configmanager, rspq):
         for outlet in outletmap:
             node, pgroup = outletmap[outlet]
             try:
-                for rsp in core.handle_path(
+                responses = await core.handle_path(
                         '/nodes/{0}/power/outlets/{1}'.format(pdu, outlet),
-                        'retrieve', configmanager):
+                        'retrieve', configmanager)
+                async for rsp in core.iterate_responses(responses):
                     await rspq.put(msg.KeyValueData({pgroup: rsp.kvpairs['state']['value']}, node))
             except exc.TargetEndpointBadCredentials:
                     await rspq.put(msg.ConfluentTargetInvalidCredentials(pdu))
@@ -137,8 +138,9 @@ async def updatepdu(pdu, outletmap, configmanager, inputdata, rspq):
     try:
         for outlet in outletmap:
             node, pgroup = outletmap[outlet]
-            for rsp in core.handle_path('/nodes/{0}/power/outlets/{1}'.format(pdu, outlet),
-                                        'update', configmanager, inputdata={'state': inputdata.powerstate(node)}):
+            responses = await core.handle_path('/nodes/{0}/power/outlets/{1}'.format(pdu, outlet),
+                                               'update', configmanager, inputdata={'state': inputdata.powerstate(node)})
+            async for rsp in core.iterate_responses(responses):
                 await rspq.put(msg.KeyValueData({pgroup: rsp.kvpairs['state']['value']}, node))
     finally:
         await rspq.put(TaskDone())

confluent_server/confluent/plugins/hardwaremanagement/redfish.py

@@ -399,7 +399,7 @@ class IpmiHandler:
         tenant = cfg.tenant
         if (node, tenant) not in persistent_ipmicmds:
             try:
-                await persistent_ipmicmds[(node, tenant)].close_confluent()
+                persistent_ipmicmds[(node, tenant)].close_confluent()
             except KeyError:  # was no previous session
                 pass
             try:
@@ -830,7 +830,7 @@ class IpmiHandler:
 
     async def make_sensor_map(self, sensors=None):
         if sensors is None:
-            sensors = await self.ipmicmd.get_sensor_descriptions()
+            sensors = self.ipmicmd.get_sensor_descriptions()
         async for sensor in sensors:
             resourcename = sensor['name']
             self.sensormap[simplify_name(resourcename)] = resourcename
@@ -1075,7 +1075,7 @@ class IpmiHandler:
                     volsfound = True
                     volumes.append(vol)
         if not volsfound:
-            self.output.put(msg.ConfluentTargetNotFound(
+            await self.output.put(msg.ConfluentTargetNotFound(
                 self.node, "No volume named '{0}' found".format(volname)))
             return
         await self.ipmicmd.remove_storage_configuration(toremove)
@@ -1128,14 +1128,14 @@ class IpmiHandler:
                                            vol.status, arrname))
                 return
             else:
-                self._show_storage(storelem[:1] + [vol['name']])
+                await self._show_storage(storelem[:1] + [vol['name']])
 
     async def _update_storage(self, storelem):
         if storelem[0] == 'disks':
             if len(storelem) == 1:
                 raise exc.InvalidArgumentException('Must target a disk')
-            self.set_disk(storelem[-1],
-                          self.inputdata.inputbynode[self.node])
+            await self.set_disk(storelem[-1],
+                                self.inputdata.inputbynode[self.node])
         await self._show_storage(storelem)
 
     async def _show_storage(self, storelem):
@@ -1292,7 +1292,7 @@ class IpmiHandler:
 
     async def list_sensors(self):
         try:
-            sensors = await self.ipmicmd.get_sensor_descriptions()
+            sensors = [sensor async for sensor in self.ipmicmd.get_sensor_descriptions()]
         except pygexc.IpmiException:
             await self.output.put(msg.ConfluentTargetTimeout(self.node))
             return

confluent_server/confluent/plugins/hardwaremanagement/srlinux.py

@@ -55,7 +55,7 @@ async def retrieve_node_backend(node, element, user, pwd, configmanager, inputda
             if element[-1] == 'all' or simplify_name(sensor['name']) == element[-1]:
                 await results.put(msg.SensorReadings([sensor], node))
     else:
-        results.put(msg.ConfluentNodeError(node, 'Not supported'))
+        await results.put(msg.ConfluentNodeError(node, 'Not supported'))
 
 
 async def retrieve(nodes, element, configmanager, inputdata):

confluent_server/confluent/plugins/shell/ssh.py

@@ -146,7 +146,7 @@ class SshShell(conapi.Console):
             if b'\r' in self.keyaction:
                 action = self.keyaction.split(b'\r')[0]
                 if action.lower() == b'accept':
-                    self.nodeconfig.set_node_attributes(
+                    await self.nodeconfig.set_node_attributes(
                         {self.node:
                              {self.keyattrname: self.candidatefprint}})
                     await self.datacallback('\r\n')
@@ -157,10 +157,10 @@ class SshShell(conapi.Console):
                     self.keyaction = b''
                     await self.datacallback('\r\nEnter "disconnect" or "accept": ')
             elif len(data) > 0:
-                self.datacallback(data)
+                await self.datacallback(data)
         elif self.inputmode == 0:
             while len(data) and data[0:1] == b'\x7f' and len(self.username):
-                self.datacallback('\b \b')  # erase previously echoed value
+                await self.datacallback('\b \b')  # erase previously echoed value
                 self.username = self.username[:-1]
                 data = data[1:]
             while len(data) and data[0:1] == b'\x7f':

confluent_server/confluent/selfservice.py

@@ -252,7 +252,7 @@ async def handle_request(req, make_response, mimetype):
         if not bmcaddr:
             return await make_response(mimetype, 500, 'Internal Server Error', body='Missing value in hardwaremanagement.manager')
         bmcaddr = bmcaddr.split('/', 1)[0]
-        bmcaddr = await asyncio.get_running_loop().getaddrinfo(bmcaddr, 0)[0]
+        bmcaddr = (await asyncio.get_running_loop().getaddrinfo(bmcaddr, 0))[0]
         bmcaddr = bmcaddr[-1][0]
         if '.' in bmcaddr:  # ipv4 is allowed
             netconfig = await netutil.get_nic_config(cfg, nodename, ip=bmcaddr)
@@ -456,9 +456,9 @@ async def handle_request(req, make_response, mimetype):
             reqbody = None
         cfgmod = reqbody.get('configmod', 'unspecified')
         if cfgmod == 'xcc':
-            xcc.remote_nodecfg(nodename, cfg)
+            await xcc.remote_nodecfg(nodename, cfg)
         elif cfgmod == 'tsm':
-            tsm.remote_nodecfg(nodename, cfg)
+            await tsm.remote_nodecfg(nodename, cfg)
         else:
             return await make_response(mimetype, 500, 'unsupported configmod', body='Unsupported configmod "{}"'.format(cfgmod))
         return await make_response(mimetype, 200, 'Ok', body='complete')

confluent_server/confluent/shellmodule.py

@@ -30,6 +30,7 @@ import fcntl
 import os
 import pty
 import random
+import select
 import subprocess
 
 
@@ -54,12 +55,12 @@ class ExecConsole(conapi.Console):
                 try:
                     somedata = os.read(self._master, 128)
                     while somedata:
-                        self._datacallback(somedata)
+                        await self._datacallback(somedata)
                         await asyncio.sleep(0)
                         somedata = os.read(self._master, 128)
                 except OSError as e:
                     if e.errno == 5:
-                        self._datacallback(conapi.ConsoleEvent.Disconnect)
+                        await self._datacallback(conapi.ConsoleEvent.Disconnect)
                         self.subproc = None
                         return
                     if e.errno != 11:
@@ -68,7 +69,7 @@ class ExecConsole(conapi.Console):
                 try:
                     somedata = self.subproc.stderr.read()
                     while somedata:
-                        self._datacallback(somedata)
+                        await self._datacallback(somedata)
                         await asyncio.sleep(0)
                         somedata = self.subproc.stderr.read()
                 except IOError as e:
@@ -76,10 +77,10 @@ class ExecConsole(conapi.Console):
                         raise
             childstate = self.subproc.poll()
             if childstate is not None:
-                self._datacallback(conapi.ConsoleEvent.Disconnect)
+                await self._datacallback(conapi.ConsoleEvent.Disconnect)
                 self.subproc = None
 
-    def connect(self, callback):
+    async def connect(self, callback):
         self._datacallback = callback
         master, slave = pty.openpty()
         self._master = master
@@ -90,14 +91,16 @@ class ExecConsole(conapi.Console):
                 stderr=subprocess.PIPE, close_fds=True)
         except OSError:
             print("Unable to execute " + self.executable + " (permissions?)")
-            self.close()
+            os.close(master)
+            os.close(slave)
+            self._master = None
             return
         os.close(slave)
         fcntl.fcntl(master, fcntl.F_SETFL, os.O_NONBLOCK)
         fcntl.fcntl(self.subproc.stderr.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
         self.readerthread = tasks.spawn(self.relaydata())
 
-    def write(self, data):
+    async def write(self, data):
         os.write(self._master, data)
 
     async def close(self):

confluent_server/confluent/shellserver.py

@@ -37,7 +37,7 @@ async def reapsessions():
             for sesshdl in list(currcli):
                 currsess = currcli[sesshdl]
                 if currsess.numusers == 0 and currsess.expiry < time.time():
-                    currsess.close()
+                    await currsess.close()
                     del activesessions[clientid][sesshdl]
 
 class _ShellHandler(consoleserver.ConsoleHandler):
@@ -54,7 +54,7 @@ class _ShellHandler(consoleserver.ConsoleHandler):
             _reaper = tasks.spawn(reapsessions())
 
 
-    def check_collective(self, attrvalue):
+    async def check_collective(self, attrvalue):
         return
 
     def log(self, *args, **kwargs):
@@ -69,7 +69,7 @@ class _ShellHandler(consoleserver.ConsoleHandler):
     #    #retdata, connstate = await super(_ShellHandler, self).get_recent()
     #    return '', {} # connstate
 
-    def _got_disconnected(self):
+    async def _got_disconnected(self):
         self.connectstate = 'closed'
         tasks.spawn(self._bgdisconnect())
 
@@ -77,7 +77,7 @@ class _ShellHandler(consoleserver.ConsoleHandler):
         await self._send_rcpts({'connectstate': self.connectstate})
         for session in list(self.livesessions):
             await session.destroy()
-        self.feedbuffer('\x1bc')
+        await self.feedbuffer('\x1bc')
 
 
 

confluent_server/confluent/sockapi.py

@@ -319,7 +319,7 @@ async def term_interact(authdata, authname, ccons, cfm, connection, consession,
                 await consession.destroy()
                 break
             elif data['operation'] == 'break':
-                consession.send_break()
+                await consession.send_break()
                 continue
             elif data['operation'] == 'reopen':
                 await consession.reopen()

confluent_server/confluent/sshutil.py

@@ -7,6 +7,7 @@ import confluent.collective.manager as collective
 import confluent.util as util
 import glob
 import os
+import signal
 import shutil
 import subprocess
 import tempfile

confluent_server/confluent/webauthn.py

@@ -248,7 +248,7 @@ async def handle_api_request(url, req, username, cfm, reqbody, authorized):
     if url == '/registration_options':
         userinfo = cfm.get_user(username)
         if not userinfo:
-            cfm.create_user(username, role='Stub')
+            await cfm.create_user(username, role='Stub')
             userinfo = cfm.get_user(username)
         authid = userinfo.get('webauthid', None)
         if not authid:  # TODO: index users by authid as well as name
@@ -293,4 +293,3 @@ async def handle_api_request(url, req, username, cfm, reqbody, authorized):
         if rsp.get('verified', False):
             return json.dumps({'status': 'Success'})
 
-