From e6f6b2330f375a260a531300f01e4fc6c76f79f2 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 14:04:31 +0200 Subject: [PATCH 01/23] Parse SMM answers as bytes, not as decoded text lxml refuses a str carrying an encoding declaration, and the SMM declares one when it answers /data/login, so _webconfigcreds has raised ValueError on the first thing it does after logging in ever since the switch to lxml. stdlib ElementTree took the same input, which is why it went unnoticed. fromstring already means to take either shape, so encode there. Confirmed against a DW612S. --- confluent_server/confluent/discovery/handlers/smm.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/confluent_server/confluent/discovery/handlers/smm.py b/confluent_server/confluent/discovery/handlers/smm.py index 86b8ab65..1ee2b3a1 100644 --- a/confluent_server/confluent/discovery/handlers/smm.py +++ b/confluent_server/confluent/discovery/handlers/smm.py @@ -38,6 +38,10 @@ def fromstring(inputdata): # The measures above should filter out the risky facets of xml # We don't need sophisticated feature support parser = etree.XMLParser(resolve_entities=False, no_network=True, huge_tree=False) + if not isinstance(inputdata, bytes): + # lxml refuses a str that declares an encoding, and the SMM declares + # one when answering a login + inputdata = inputdata.encode('utf8') return etree.fromstring(inputdata, parser=parser) def fixuuid(baduuid): From 0bd8a8504626c1ad782e48f822c19ff5de9858a7 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:27 +0200 Subject: [PATCH 02/23] Add a pyrefly job for async correctness Catches unawaited coroutines and awaits on non-awaitables, which ruff and compileall cannot see. Only those two kinds are enabled, and in pyrefly.toml rather than a --only flag so CI, local runs and editors agree; a full check reports thousands of errors on this largely unannotated tree. search-path is what lets imports resolve, and without it a large share of the findings, including everything in the SMM handler, goes unreported. Pyrefly walks *.py only and a glob does not lift that, so extensionless tools are named individually. No baseline: it matches by file, kind and column, so a new mistake at the same indentation as an old one would pass unnoticed. Advisory until the existing findings are dealt with. --- .github/workflows/ci.yml | 12 ++++ pyrefly.toml | 138 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 150 insertions(+) create mode 100644 pyrefly.toml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 62708ed7..641d3b18 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,6 +52,18 @@ jobs: # workspace can be handed over as-is. args: check --output-format=github + pyrefly: + name: Pyrefly (async correctness) + runs-on: ubuntu-latest + # Advisory for now: the tree still has findings, a good number of them + # real bugs, so this reports without failing the run. It also means an + # install failure or a checker crash passes silently. Remove the line once + # the findings are dealt with, which is the point of having it. + continue-on-error: true + steps: + - uses: actions/checkout@v7 + - uses: facebook/pyrefly@main + python-compileall: name: Python compileall runs-on: ubuntu-latest diff --git a/pyrefly.toml b/pyrefly.toml new file mode 100644 index 00000000..b25e736a --- /dev/null +++ b/pyrefly.toml @@ -0,0 +1,138 @@ +# Pyrefly configuration for confluent. +# +# The whole policy is the two error kinds enabled at the bottom of this file: +# a coroutine that is never awaited, where the call silently does nothing, and +# an await on something that is not awaitable, which raises TypeError as soon +# as the line runs. Those are the mistakes the asyncio port can make that no +# other check in CI can see. Everything else is off, because a full check +# reports several thousand errors on this largely unannotated tree, mostly +# attribute lookups it cannot verify. +# +# The policy lives here rather than in a --only flag on the command line, so +# that CI, a local run and the editor integration all behave the same way. +# +# A baseline file is deliberately not used. Pyrefly matches baselined findings +# by file, error kind and column, so a new mistake that happens to land at the +# same indentation as an existing one in the same file is silently accepted. + +project-includes = [ + # Every *.py in the tree, so this matches what ruff and compileall see. + ".", + # Pyrefly walks *.py only, and a glob in this list does not lift that + # restriction: it expands the glob and then drops everything without the + # extension. So every extensionless tool has to be named. This list is + # the one ruff.toml arrives at through extend-include, and is kept in + # step with it; anything missing here is skipped without a word. + "confluent_client/bin/collate", + "confluent_client/bin/confetty", + "confluent_client/bin/confluent2ansible", + "confluent_client/bin/confluent2dnsmasq", + "confluent_client/bin/confluent2hosts", + "confluent_client/bin/confluent2lxca", + "confluent_client/bin/confluent2xcat", + "confluent_client/bin/dir2img", + "confluent_client/bin/nodeapply", + "confluent_client/bin/nodeattrib", + "confluent_client/bin/nodebmcpassword", + "confluent_client/bin/nodebmcreset", + "confluent_client/bin/nodeboot", + "confluent_client/bin/nodecertutil", + "confluent_client/bin/nodeconfig", + "confluent_client/bin/nodeconsole", + "confluent_client/bin/nodedefine", + "confluent_client/bin/nodedeploy", + "confluent_client/bin/nodediscover", + "confluent_client/bin/nodeeventlog", + "confluent_client/bin/nodefirmware", + "confluent_client/bin/nodegroupattrib", + "confluent_client/bin/nodegroupdefine", + "confluent_client/bin/nodegrouplist", + "confluent_client/bin/nodegroupremove", + "confluent_client/bin/nodegrouprename", + "confluent_client/bin/nodehealth", + "confluent_client/bin/nodeidentify", + "confluent_client/bin/nodeinventory", + "confluent_client/bin/nodel2traceroute", + "confluent_client/bin/nodelicense", + "confluent_client/bin/nodelist", + "confluent_client/bin/nodemedia", + "confluent_client/bin/nodeping", + "confluent_client/bin/nodepower", + "confluent_client/bin/noderemove", + "confluent_client/bin/noderename", + "confluent_client/bin/nodereseat", + "confluent_client/bin/nodersync", + "confluent_client/bin/noderun", + "confluent_client/bin/nodesensors", + "confluent_client/bin/nodesetboot", + "confluent_client/bin/nodeshell", + "confluent_client/bin/nodestorage", + "confluent_client/bin/nodesupport", + "confluent_client/bin/stats", + "confluent_client/setup.py.tmpl", + "confluent_common/setup.py.tmpl", + "confluent_osdeploy/bluefield/bfb-autoinstall", + "confluent_osdeploy/bluefield/hostscripts/bfb-autoinstall", + "confluent_osdeploy/bluefield/profiles/default/nodedeploy-bfb", + "confluent_osdeploy/common/initramfs/opt/confluent/bin/apiclient", + "confluent_osdeploy/common/profile/scripts/autoconsole", + "confluent_osdeploy/common/profile/scripts/confignet", + "confluent_osdeploy/common/profile/scripts/getinstalldisk", + "confluent_osdeploy/common/profile/scripts/syncfileclient", + "confluent_osdeploy/debian/profiles/default/scripts/confignet", + "confluent_osdeploy/el10-diskless/profiles/default/scripts/add_local_repositories", + "confluent_osdeploy/el7-diskless/profiles/default/scripts/syncfileclient", + "confluent_osdeploy/el7/profiles/default/scripts/add_local_repositories", + "confluent_osdeploy/el7/profiles/default/scripts/configbmc", + "confluent_osdeploy/el7/profiles/default/scripts/syncfileclient", + "confluent_osdeploy/el8-diskless/profiles/default/scripts/add_local_repositories", + "confluent_osdeploy/el8/profiles/default/scripts/add_local_repositories", + "confluent_osdeploy/el8/profiles/default/scripts/configbmc", + "confluent_osdeploy/el9-diskless/profiles/default/scripts/add_local_repositories", + "confluent_osdeploy/esxi7/profiles/hypervisor/scripts/getinstalldisk", + "confluent_osdeploy/esxi7/profiles/hypervisor/scripts/makeksnet", + "confluent_osdeploy/genesis/profiles/default/scripts/configbmc", + "confluent_osdeploy/ubuntu22.04/profiles/default/scripts/mergetime", + "confluent_server/bin/collective", + "confluent_server/bin/confluent", + "confluent_server/bin/confluentdbutil", + "confluent_server/bin/confluent_selfcheck", + "confluent_server/bin/osdeploy", + "confluent_server/setup.py.tmpl", + "imgutil/imgutil", + "misc/filterpasswd", + "misc/getipsfromswitchport", + "misc/getusbnicaddr", +] + +project-excludes = [ + # A symlink to a unit file that only exists on a deployed system. + # Pyrefly refuses to walk the tree at all when it meets this. + "confluent_osdeploy/coreos/initramfs/etc/systemd/system/initrd-root-fs.target.requires/confluent-rootfs.service", +] + +# project-includes only picks the files to check. Imports are resolved against +# search-path, and without these roots pyrefly cannot tell that, for instance, +# aiohmi.util.webclient.request is async, and a large share of the findings, +# among them everything in the SMM discovery handler, goes unreported. The two +# trees both hold a package named confluent but share no module names, so one +# search path for each works; that would need revisiting if they ever overlap. +search-path = [ + "confluent_server", + "confluent_client", +] + +# The oldest interpreter parts of this tree still run on (el8, sles15). +python-version = "3.6" + +preset = "off" +# Both of these are load bearing rather than decoration. With unannotated +# bodies unchecked the check reports nothing at all, since almost nothing here +# is annotated, and without inferred return types it misses a good part of the +# rest. +check-unannotated-defs = true +infer-return-types = "checked" + +[errors] +unused-coroutine = "error" +not-async = "error" From 9148a9e1cf46e66a87fa3dc8c483e3dc5f63cc4d Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:27 +0200 Subject: [PATCH 03/23] Run the module self tests through asyncio These __main__ blocks called coroutines as if they were functions, so they did nothing at all. Single calls go through asyncio.run; sshutil, proxmox and vcenter needed an _selftest coroutine. Two were invisible to pyrefly because repr() and list() count as using the result: vcenter's get_vm_serial needs an await, and proxmox's get_vm_inventory is an async generator. lldp called _extract_neighbor_data twice, once correctly, so the bare call is dropped. xcc3.remote_nodecfg is not a self test: every other handler defines it as a coroutine and selfservice.py awaits it. --- .../confluent/discovery/handlers/megarac.py | 2 +- .../discovery/handlers/redfishbmc.py | 2 +- .../confluent/discovery/handlers/tsm.py | 2 +- .../confluent/discovery/handlers/xcc3.py | 7 ++++--- .../confluent/discovery/protocols/mdns.py | 2 +- .../confluent/discovery/protocols/pxe.py | 2 +- .../confluent/discovery/protocols/slp.py | 2 +- confluent_server/confluent/networking/lldp.py | 1 - .../plugins/hardwaremanagement/proxmox.py | 18 ++++++++++------- .../plugins/hardwaremanagement/vcenter.py | 20 +++++++++++-------- confluent_server/confluent/sshutil.py | 13 ++++++++---- 11 files changed, 42 insertions(+), 29 deletions(-) diff --git a/confluent_server/confluent/discovery/handlers/megarac.py b/confluent_server/confluent/discovery/handlers/megarac.py index 586604f0..2e767f9e 100644 --- a/confluent_server/confluent/discovery/handlers/megarac.py +++ b/confluent_server/confluent/discovery/handlers/megarac.py @@ -113,5 +113,5 @@ if __name__ == '__main__': info = {'addresses': [[sys.argv[1]]]} print(repr(info)) testr = NodeHandler(info, c) - testr.config(sys.argv[2]) + asyncio.run(testr.config(sys.argv[2])) diff --git a/confluent_server/confluent/discovery/handlers/redfishbmc.py b/confluent_server/confluent/discovery/handlers/redfishbmc.py index bf0fc087..5cdaeacc 100644 --- a/confluent_server/confluent/discovery/handlers/redfishbmc.py +++ b/confluent_server/confluent/discovery/handlers/redfishbmc.py @@ -357,4 +357,4 @@ if __name__ == '__main__': info = {'addresses': [[sys.argv[1]]] } print(repr(info)) testr = NodeHandler(info, c) - testr.config(sys.argv[2]) + asyncio.run(testr.config(sys.argv[2])) diff --git a/confluent_server/confluent/discovery/handlers/tsm.py b/confluent_server/confluent/discovery/handlers/tsm.py index 71ef812f..6b637e51 100644 --- a/confluent_server/confluent/discovery/handlers/tsm.py +++ b/confluent_server/confluent/discovery/handlers/tsm.py @@ -246,4 +246,4 @@ if __name__ == '__main__': info = {'addresses': [[sys.argv[1]]] } print(repr(info)) testr = NodeHandler(info, c) - testr.config(sys.argv[2]) + asyncio.run(testr.config(sys.argv[2])) diff --git a/confluent_server/confluent/discovery/handlers/xcc3.py b/confluent_server/confluent/discovery/handlers/xcc3.py index e2d57095..dbfa4e10 100644 --- a/confluent_server/confluent/discovery/handlers/xcc3.py +++ b/confluent_server/confluent/discovery/handlers/xcc3.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio import codecs import confluent.discovery.handlers.redfishbmc as redfishbmc import confluent.util as util @@ -90,7 +91,7 @@ class NodeHandler(redfishbmc.NodeHandler): -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( @@ -102,7 +103,7 @@ def remote_nodecfg(nodename, cfm): 'address') info = {'addresses': [ipaddr]} nh = NodeHandler(info, cfm) - nh.config(nodename) + await nh.config(nodename) if __name__ == '__main__': @@ -112,5 +113,5 @@ if __name__ == '__main__': info = {'addresses': [[sys.argv[1]]]} print(repr(info)) testr = NodeHandler(info, c) - testr.config(sys.argv[2]) + asyncio.run(testr.config(sys.argv[2])) diff --git a/confluent_server/confluent/discovery/protocols/mdns.py b/confluent_server/confluent/discovery/protocols/mdns.py index be889675..4435f218 100644 --- a/confluent_server/confluent/discovery/protocols/mdns.py +++ b/confluent_server/confluent/discovery/protocols/mdns.py @@ -480,4 +480,4 @@ from pprint import pprint if __name__ == '__main__': def printit(rsp): print(repr(rsp)) - snoop(pprint) + asyncio.run(snoop(pprint)) diff --git a/confluent_server/confluent/discovery/protocols/pxe.py b/confluent_server/confluent/discovery/protocols/pxe.py index db640bf3..743125a4 100644 --- a/confluent_server/confluent/discovery/protocols/pxe.py +++ b/confluent_server/confluent/discovery/protocols/pxe.py @@ -1004,4 +1004,4 @@ async def consider_discover(info, packet, sock, cfg, reqview, nodeguess, addr=No if __name__ == '__main__': def testsnoop(info): print(repr(info)) - snoop(testsnoop) + asyncio.run(snoop(testsnoop)) diff --git a/confluent_server/confluent/discovery/protocols/slp.py b/confluent_server/confluent/discovery/protocols/slp.py index 1e778829..54a59b55 100644 --- a/confluent_server/confluent/discovery/protocols/slp.py +++ b/confluent_server/confluent/discovery/protocols/slp.py @@ -698,4 +698,4 @@ async def scan(srvtypes=_slp_services, addresses=None, localonly=False): if __name__ == '__main__': def testsnoop(a): print(repr(a)) - snoop(testsnoop) + asyncio.run(snoop(testsnoop)) diff --git a/confluent_server/confluent/networking/lldp.py b/confluent_server/confluent/networking/lldp.py index 5ab5ccce..792b0a93 100644 --- a/confluent_server/confluent/networking/lldp.py +++ b/confluent_server/confluent/networking/lldp.py @@ -411,7 +411,6 @@ if __name__ == '__main__': # a quick one-shot test, args are switch and snmpv1 string for now # (should do three argument form for snmpv3 test import sys - _extract_neighbor_data((sys.argv[1], sys.argv[2], None, True)) asyncio.run(_extract_neighbor_data((sys.argv[1], sys.argv[2], None, True))) print(repr(_neighdata)) diff --git a/confluent_server/confluent/plugins/hardwaremanagement/proxmox.py b/confluent_server/confluent/plugins/hardwaremanagement/proxmox.py index abb32e28..614898bf 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/proxmox.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/proxmox.py @@ -481,7 +481,7 @@ async def create(nodes, element, configmanager, inputdata): return -if __name__ == '__main__': +async def _selftest(): import sys import os myuser = os.environ['PMXUSER'] @@ -489,12 +489,16 @@ if __name__ == '__main__': vc = PmxApiClient(sys.argv[1], myuser, mypass, None) vm = sys.argv[2] if sys.argv[3] == 'setboot': - vc.set_vm_bootdev(vm, sys.argv[4]) - vc.get_vm_bootdev(vm) + await vc.set_vm_bootdev(vm, sys.argv[4]) + await vc.get_vm_bootdev(vm) elif sys.argv[3] == 'power': - vc.set_vm_power(vm, sys.argv[4]) + await vc.set_vm_power(vm, sys.argv[4]) elif sys.argv[3] == 'getinfo': - print(repr(list(vc.get_vm_inventory(vm)))) - print("Bootdev: " + vc.get_vm_bootdev(vm)) - print("Power: " + vc.get_vm_power(vm)) + print(repr([datum async for datum in vc.get_vm_inventory(vm)])) + print("Bootdev: " + await vc.get_vm_bootdev(vm)) + print("Power: " + await vc.get_vm_power(vm)) #print("Serial: " + repr(vc.get_vm_serial(vm))) + + +if __name__ == '__main__': + asyncio.run(_selftest()) diff --git a/confluent_server/confluent/plugins/hardwaremanagement/vcenter.py b/confluent_server/confluent/plugins/hardwaremanagement/vcenter.py index 4772316e..4ed326cc 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/vcenter.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/vcenter.py @@ -411,7 +411,7 @@ async def create(nodes, element, configmanager, inputdata): -if __name__ == '__main__': +async def _selftest(): import sys import os myuser = os.environ['VMWUSER'] @@ -419,12 +419,16 @@ if __name__ == '__main__': vc = VmwApiClient(sys.argv[1], myuser, mypass, None) vm = sys.argv[2] if sys.argv[3] == 'setboot': - vc.set_vm_bootdev(vm, sys.argv[4]) - vc.get_vm_bootdev(vm) + await vc.set_vm_bootdev(vm, sys.argv[4]) + await vc.get_vm_bootdev(vm) elif sys.argv[3] == 'power': - vc.set_vm_power(vm, sys.argv[4]) + await vc.set_vm_power(vm, sys.argv[4]) elif sys.argv[3] == 'getinfo': - vc.get_vm(vm) - print("Bootdev: " + vc.get_vm_bootdev(vm)) - print("Power: " + vc.get_vm_power(vm)) - print("Serial: " + repr(vc.get_vm_serial(vm))) + await vc.get_vm(vm) + print("Bootdev: " + await vc.get_vm_bootdev(vm)) + print("Power: " + await vc.get_vm_power(vm)) + print("Serial: " + repr(await vc.get_vm_serial(vm))) + + +if __name__ == '__main__': + asyncio.run(_selftest()) diff --git a/confluent_server/confluent/sshutil.py b/confluent_server/confluent/sshutil.py index 7b2e71d3..cce58b39 100644 --- a/confluent_server/confluent/sshutil.py +++ b/confluent_server/confluent/sshutil.py @@ -271,8 +271,13 @@ def ca_exists(): return os.path.exists('/etc/confluent/ssh/ca') -if __name__ == '__main__': - initialize_root_key(True) +async def _selftest(): + await initialize_root_key(True) if not ca_exists(): - initialize_ca() - print(repr(sign_host_key(open('/etc/ssh/ssh_host_ed25519_key.pub').read(), collective.get_myname()))) + await initialize_ca() + print(repr(await sign_host_key( + open('/etc/ssh/ssh_host_ed25519_key.pub').read(), collective.get_myname()))) + + +if __name__ == '__main__': + asyncio.run(_selftest()) From 2ea7aed2cc10ce4146e81fac8d3c99836f719719 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:27 +0200 Subject: [PATCH 04/23] Port the SMM handler, Delta PDU logout and XCC config to async Taken from fix/asyncio-port-critical, limited to what pyrefly reports. The SMM handler still used the httplib style connect/request/getresponse that the async webclient does not have, so nothing was ever sent. It goes through grab_response_with_status now, with allow_redirects=False to preserve httplib behaviour, hence the new webclient parameter. That also fixes a login passing its headers as urlencode's second positional argument. Delta PDU's logout was a plain function both callers awaited, so the power paths raised TypeError. XCC's set_system_configuration was the last synchronous implementation of a method every caller awaits. --- .../aiohmi/redfish/oem/lenovo/xcc.py | 10 +- confluent_server/aiohmi/util/webclient.py | 3 +- .../confluent/discovery/handlers/smm.py | 108 +++++++++--------- .../plugins/hardwaremanagement/deltapdu.py | 2 +- 4 files changed, 62 insertions(+), 61 deletions(-) diff --git a/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py b/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py index a8343cd1..33eeaab3 100644 --- a/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py +++ b/confluent_server/aiohmi/redfish/oem/lenovo/xcc.py @@ -206,15 +206,15 @@ class OEMHandler(generic.OEMHandler): retcfg[opt]['sortid'] = self.fwo[opt]['sortid'] return retcfg - def set_system_configuration(self, changeset, fishclient): + async def set_system_configuration(self, changeset, fishclient): if not self.fwc: self.fwc = config.LenovoFirmwareConfig(self, useipmi=False) fetchimm = False if not self.fwo or util._monotonic_time() - self.fwovintage > 30: try: - self.fwo = self.fwc.get_fw_options(fetchimm=fetchimm) + self.fwo = await self.fwc.get_fw_options(fetchimm=fetchimm) except config.Unsupported: - return super(OEMHandler, self).set_system_configuration( + return await super(OEMHandler, self).set_system_configuration( changeset, fishclient) self.fwovintage = util._monotonic_time() for key in list(changeset): @@ -231,7 +231,7 @@ class OEMHandler(generic.OEMHandler): found = True if not found and not fetchimm: fetchimm = True - self.fwo = self.fwc.get_fw_options(fetchimm=fetchimm) + self.fwo = await self.fwc.get_fw_options(fetchimm=fetchimm) if key in self.fwo: continue else: @@ -254,7 +254,7 @@ class OEMHandler(generic.OEMHandler): self.merge_changeset(changeset) if changeset: try: - self.fwc.set_fw_options(self.fwo) + await self.fwc.set_fw_options(self.fwo) finally: self.fwo = None self.fwovintage = 0 diff --git a/confluent_server/aiohmi/util/webclient.py b/confluent_server/aiohmi/util/webclient.py index 9840dbeb..1fae311b 100644 --- a/confluent_server/aiohmi/util/webclient.py +++ b/confluent_server/aiohmi/util/webclient.py @@ -357,7 +357,7 @@ class WebConnection: return rsp, status async def grab_response_with_status(self, url, data=None, referer=None, - headers=None, method=None, expect_type=None): + headers=None, method=None, expect_type=None, allow_redirects=True): if not headers: headers = self.stdheaders.copy() else: @@ -378,6 +378,7 @@ class WebConnection: headers['Content-Type'] = 'application/json' elif data is not None: kwargs['data'] = data + kwargs['allow_redirects'] = allow_redirects async with thefunc(url, headers=headers, ssl=self.ssl, **kwargs) as rsp: if rsp.status >= 200 and rsp.status < 300: if expect_type == 'json': diff --git a/confluent_server/confluent/discovery/handlers/smm.py b/confluent_server/confluent/discovery/handlers/smm.py index 1ee2b3a1..8ea7a38e 100644 --- a/confluent_server/confluent/discovery/handlers/smm.py +++ b/confluent_server/confluent/discovery/handlers/smm.py @@ -53,6 +53,20 @@ def fixuuid(baduuid): uuid = (a[:8], a[8:12], a[12:16], baduuid[16:20], baduuid[20:]) return '-'.join(uuid).lower() + +async def _post(wc, url, data, headers=None): + if headers is None: + headers = wc.stdheaders.copy() + if data: + # What WebConnection.request used to fill in, and what the SMM + # expects of its /data endpoints. grab_response_with_status adds + # nothing, so aiohttp would label a str body text/plain instead. + headers['Content-Type'] = 'application/x-www-form-urlencoded' + body, _, _ = await wc.grab_response_with_status( + url, data, headers=headers, method='POST', allow_redirects=False) + return body + + class NodeHandler(bmchandler.NodeHandler): is_enclosure = True devname = 'SMM' @@ -66,7 +80,7 @@ class NodeHandler(bmchandler.NodeHandler): uuid = fixuuid(uuid[0]) self.info['uuid'] = uuid - def _webconfigrules(self, wc): + async def _webconfigrules(self, wc): rules = [] for rule in self.ruleset.split(','): if '=' not in rule: @@ -84,19 +98,16 @@ class NodeHandler(bmchandler.NodeHandler): rules.append('passwordReuseCheckNum:' + value) if rules: apirequest = 'set={0}'.format(','.join(rules)) - wc.request('POST', '/data', apirequest) - wc.getresponse().read() + await _post(wc, '/data', apirequest) async def _webconfignet(self, wc, nodename): cfg = self.configmanager if 'service:lenovo-smm2' in self.info.get('services', []): # need to enable ipmi for now.. - wc.request('POST', '/data', 'set=DoCmd(0x06,0x40,0x01,0x82,0x84)') - rsp = wc.getresponse() - rsp.read() - wc.request('POST', '/data', 'set=DoCmd(0x06,0x40,0x01,0x42,0x44)') - rsp = wc.getresponse() - rsp.read() + await _post(wc, '/data', + 'set=DoCmd(0x06,0x40,0x01,0x82,0x84)') + await _post(wc, '/data', + 'set=DoCmd(0x06,0x40,0x01,0x42,0x44)') cd = cfg.get_node_attributes( nodename, ['hardwaremanagement.manager']) smmip = cd.get(nodename, {}).get('hardwaremanagement.manager', {}).get('value', None) @@ -107,9 +118,8 @@ class NodeHandler(bmchandler.NodeHandler): smmip = smmip[-1][0] if smmip and ':' in smmip: raise exc.NotImplementedException('IPv6 not supported') - wc.request('POST', '/data', 'get=hostname') - rsp = wc.getresponse() - rspdata = fromstring(util.stringify(rsp.read())) + rspdata = fromstring(util.stringify( + await _post(wc, '/data', 'get=hostname'))) currip = rspdata.find('netConfig').find('ifConfigEntries').find( 'ifConfig').find('v4IPAddr').text if currip == smmip: @@ -120,9 +130,7 @@ class NodeHandler(bmchandler.NodeHandler): gateway = netconfig.get('ipv4_gateway', None) if gateway: setdata += ',v4Gateway:{0}'.format(gateway) - wc.request('POST', '/data', setdata) - rsp = wc.getresponse() - rspdata = util.stringify(rsp.read()) + rspdata = util.stringify(await _post(wc, '/data', setdata)) if '0' not in rspdata: raise Exception("Error configuring SMM Network") return @@ -132,48 +140,45 @@ class NodeHandler(bmchandler.NodeHandler): await cfg.set_node_attributes( {nodename: {'hardwaremanagement.manager': self.ipaddr}}) - def _webconfigcreds(self, username, password): - ip, port = self.get_web_port_and_ip() + async def _webconfigcreds(self, username, password): + ip, port = await self.get_web_port_and_ip() wc = webclient.WebConnection(ip, port, verifycallback=self.validate_cert) - wc.connect() authdata = { # start by trying factory defaults 'user': 'USERID', 'password': 'PASSW0RD', } headers = {'Connection': 'keep-alive', 'Content-Type': 'application/x-www-form-urlencoded'} - wc.request('POST', '/data/login', urlencode(authdata), headers) - rsp = wc.getresponse() - rspdata = util.stringify(rsp.read()) + rspdata = util.stringify( + await _post(wc, '/data/login', urlencode(authdata), headers)) if 'authResult>0' not in rspdata: # default credentials are refused, try with the actual authdata['user'] = username authdata['password'] = password - wc.request('POST', '/data/login', urlencode(authdata), headers) - rsp = wc.getresponse() - rspdata = util.stringify(rsp.read()) + rspdata = util.stringify( + await _post(wc, '/data/login', urlencode(authdata), headers)) if 'renew_account' in rspdata: tmppassword = 'Tmp42' + password[5:] tokens = fromstring(rspdata) st2 = tokens.findall('st2')[0].text wc.set_header('ST2', st2) - wc.request('POST', '/data/changepwd', 'oripwd={0}&newpwd={1}'.format(password, tmppassword)) - rsp = wc.getresponse() - rspdata = rsp.read().decode('utf8') + rspdata = await _post( + wc, '/data/changepwd', + 'oripwd={0}&newpwd={1}'.format(password, tmppassword)) + rspdata = rspdata.decode('utf8') bdata = 'user={0}&password={1}'.format(username, tmppassword) - wc.request('POST', '/data/login', bdata, headers) - rsp = wc.getresponse() - rspdata = rsp.read().decode('utf8') + rspdata = await _post(wc, '/data/login', bdata, headers) + rspdata = rspdata.decode('utf8') tokens = fromstring(rspdata) st2 = tokens.findall('st2')[0].text wc.set_header('ST2', st2) rules = 'set=passwordChangeInterval:0,passwordReuseCheckNum:0' - wc.request('POST', '/data', rules) - wc.getresponse().read() - wc.request('POST', '/data/changepwd', 'oripwd={0}&newpwd={1}'.format(tmppassword, password)) - wc.getresponse().read() - wc.request('POST', '/data/login', urlencode(authdata), headers) - rsp = wc.getresponse() - rspdata = util.stringify(rsp.read()) + await _post(wc, '/data', rules) + await _post( + wc, '/data/changepwd', + 'oripwd={0}&newpwd={1}'.format(tmppassword, password)) + rspdata = util.stringify( + await _post(wc, '/data/login', urlencode(authdata), + headers)) if 'authResult>0' not in rspdata: raise Exception('Unknown username/password on SMM') tokens = fromstring(rspdata) @@ -185,29 +190,24 @@ class NodeHandler(bmchandler.NodeHandler): tokens = fromstring(rspdata) st2 = tokens.findall('st2')[0].text wc.set_header('ST2', st2) - wc.request('POST', '/data/changepwd', urlencode(passwdchange)) - rsp = wc.getresponse() - rspdata = rsp.read() + rspdata = await _post( + wc, '/data/changepwd', urlencode(passwdchange)) authdata['password'] = password - wc.request('POST', '/data/login', urlencode(authdata), headers) - rsp = wc.getresponse() - rspdata = util.stringify(rsp.read()) + rspdata = util.stringify( + await _post(wc, '/data/login', urlencode(authdata), headers)) if 'authResult>0' in rspdata: tokens = fromstring(rspdata) st2 = tokens.findall('st2')[0].text wc.set_header('ST2', st2) if username == 'USERID': return wc - wc.request('POST', '/data', 'set=user(2,1,{0},511,,4,15,0)'.format(username)) - rsp = wc.getresponse() - rspdata = rsp.read() - wc.request('POST', '/data/logout') - rsp = wc.getresponse() - rspdata = rsp.read() + rspdata = await _post( + wc, '/data', + 'set=user(2,1,{0},511,,4,15,0)'.format(username)) + rspdata = await _post(wc, '/data/logout', None) authdata['user'] = username - wc.request('POST', '/data/login', urlencode(authdata, headers)) - rsp = wc.getresponse() - rspdata = rsp.read() + rspdata = await _post( + wc, '/data/login', urlencode(authdata), headers) tokens = fromstring(rspdata) st2 = tokens.findall('st2')[0].text wc.set_header('ST2', st2) @@ -249,8 +249,8 @@ class NodeHandler(bmchandler.NodeHandler): raise Exception('Using the default password is no longer supported') else: # Switch to full web based configuration, to mitigate risks with the SMM - wc = self._webconfigcreds(username, passwd) - self._webconfigrules(wc) + wc = await self._webconfigcreds(username, passwd) + await self._webconfigrules(wc) await self._webconfignet(wc, nodename) diff --git a/confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py b/confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py index 02df6406..3371b31c 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py @@ -158,7 +158,7 @@ class PDUClient(object): raise exc.TargetEndpointBadCredentials() - def logout(self): + async def logout(self): self.wc.grab_response('/logout_wait.htm') async def get_outlet(self, outlet): From 92c9abc74a3ce0c2d4c618d2e33dcdbc9ea33b8f Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 05/23] Await the remaining reachable coroutines sockapi sent its collective refusal without awaiting tlvdata.send. redfish handle_sensors returned a coroutine from most branches and None from the short ones, so the caller's await raised TypeError; it is a coroutine throughout now. console send_payload waited for a response without awaiting the wait. Two suppressions are pyrefly limitations rather than bugs: Session defines an async __new__, and the keepalive registry holds coroutine functions in an untyped dict. --- confluent_server/aiohmi/ipmi/console.py | 5 ++++- confluent_server/aiohmi/ipmi/private/session.py | 3 +++ .../confluent/plugins/hardwaremanagement/redfish.py | 12 ++++++------ confluent_server/confluent/sockapi.py | 2 +- 4 files changed, 14 insertions(+), 8 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/console.py b/confluent_server/aiohmi/ipmi/console.py index b7465ea4..81965907 100644 --- a/confluent_server/aiohmi/ipmi/console.py +++ b/confluent_server/aiohmi/ipmi/console.py @@ -73,6 +73,9 @@ class Console(object): password = self.password port = self.port kg = self.kg + # Session defines an async __new__, so the constructor call is a + # coroutine, which pyrefly does not model. + # pyrefly: ignore[not-async] self.ipmi_session = await session.Session( bmc=bmc, userid=userid, password=password, port=port, kg=kg) # induce one iteration of the loop, now that we would be @@ -301,7 +304,7 @@ class Console(object): async def send_payload(self, payload, payload_type=1, retry=True, needskeepalive=False): while not (self.connected or self.broken): - session.Session.wait_for_rsp(timeout=10) + await session.Session.wait_for_rsp(timeout=10) if self.ipmi_session is None or not self.ipmi_session.logged: await self._print_error('Session no longer connected') raise exc.IpmiException('Session no longer connected') diff --git a/confluent_server/aiohmi/ipmi/private/session.py b/confluent_server/aiohmi/ipmi/private/session.py index 1842d955..2bbb628e 100644 --- a/confluent_server/aiohmi/ipmi/private/session.py +++ b/confluent_server/aiohmi/ipmi/private/session.py @@ -1354,6 +1354,9 @@ class Session(object): # deregister continue if callable(cmd): + # registered callables are coroutine functions, + # but the registry itself is untyped + # pyrefly: ignore[not-async] await cmd() continue keptalive = True diff --git a/confluent_server/confluent/plugins/hardwaremanagement/redfish.py b/confluent_server/confluent/plugins/hardwaremanagement/redfish.py index ebc64d8f..86800d21 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/redfish.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/redfish.py @@ -1133,23 +1133,23 @@ class IpmiHandler: return await self._show_all_storage() - def handle_sensors(self): + async def handle_sensors(self): if self.element[-1] == '': self.element = self.element[:-1] if len(self.element) < 3: return self.sensorcategory = self.element[2] if self.sensorcategory == 'normalized': - return self.read_normalized(self.element[-1]) + return await self.read_normalized(self.element[-1]) # list sensors per category if len(self.element) == 3 and self.element[-2] == 'hardware': if self.sensorcategory == 'leds': - return self.list_leds() - return self.list_sensors() + return await self.list_leds() + return await self.list_sensors() elif len(self.element) == 4: # resource requested if self.sensorcategory == 'leds': - return self.read_leds(self.element[-1]) - return self.read_sensors(self.element[-1]) + return await self.read_leds(self.element[-1]) + return await self.read_sensors(self.element[-1]) def match_sensor(self, sensor): if self.sensorcategory == 'all': diff --git a/confluent_server/confluent/sockapi.py b/confluent_server/confluent/sockapi.py index 3bdc15f8..51b5dbdd 100644 --- a/confluent_server/confluent/sockapi.py +++ b/confluent_server/confluent/sockapi.py @@ -156,7 +156,7 @@ async def sessionhdl(connection, authname, skipauth=False, cert=None): return await collective.handle_connection( connection, None, request['collective'], local=True) else: - tlvdata.send( + await tlvdata.send( connection, {'collective': { 'error': 'collective management commands ' From b96d4b4103ee2ed7edb5035f8a1145147d35fa79 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 06/23] Give the aiohmi command line utilities an event loop Console.main_loop drives Session.wait_for_rsp, which is a coroutine, so it spun without ever waiting for a packet. It is a coroutine now, and pyghmicons runs its main under asyncio.run. pyghmiutil had the same shape around Command.eventloop. --- confluent_server/aiohmi/cmd/pyghmicons.py | 7 ++++--- confluent_server/aiohmi/cmd/pyghmiutil.py | 7 ++++--- confluent_server/aiohmi/ipmi/console.py | 4 ++-- 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/confluent_server/aiohmi/cmd/pyghmicons.py b/confluent_server/aiohmi/cmd/pyghmicons.py index 590ff3b5..0cb0d6f5 100755 --- a/confluent_server/aiohmi/cmd/pyghmicons.py +++ b/confluent_server/aiohmi/cmd/pyghmicons.py @@ -14,6 +14,7 @@ """ A simple little script to exemplify/test ipmi.console module """ +import asyncio import fcntl import os import select @@ -50,7 +51,7 @@ def _print(data): raise Exception(data) -def main(): +async def main(): tcattr = termios.tcgetattr(sys.stdin) newtcattr = tcattr # TODO(jbjohnso): add our exit handler @@ -75,7 +76,7 @@ def main(): inputthread = threading.Thread(target=_doinput, args=(sol,)) inputthread.daemon = True inputthread.start() - sol.main_loop() + await sol.main_loop() except Exception: currfl = fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL) @@ -85,4 +86,4 @@ def main(): if __name__ == '__main__': - sys.exit(main()) + sys.exit(asyncio.run(main())) diff --git a/confluent_server/aiohmi/cmd/pyghmiutil.py b/confluent_server/aiohmi/cmd/pyghmiutil.py index 2231d406..e16dbed9 100755 --- a/confluent_server/aiohmi/cmd/pyghmiutil.py +++ b/confluent_server/aiohmi/cmd/pyghmiutil.py @@ -17,6 +17,7 @@ it isn't conceived as a general utility to actually use, just help developers understand how the ipmi_command class workes. """ +import asyncio import functools import os import sys @@ -64,7 +65,7 @@ def docommand(args, result, ipmisession): data=map(lambda x: int(x, 16), args[2:]))) -def main(): +async def main(): if (len(sys.argv) < 3) or 'IPMIPASSWORD' not in os.environ: print("Usage:") print(" IPMIPASSWORD=password %s bmc username " % @@ -85,8 +86,8 @@ def main(): onlogon=functools.partial(docommand, sys.argv[3:])) if ipmicmd: - ipmicmd.eventloop() + await ipmicmd.eventloop() if __name__ == '__main__': - sys.exit(main()) + sys.exit(asyncio.run(main())) diff --git a/confluent_server/aiohmi/ipmi/console.py b/confluent_server/aiohmi/ipmi/console.py index 81965907..66312bdf 100644 --- a/confluent_server/aiohmi/ipmi/console.py +++ b/confluent_server/aiohmi/ipmi/console.py @@ -418,7 +418,7 @@ class Console(object): # sooner than timeout suggests is evidently a big deal await self.send_payload(payload=self.lastpayload, retry=False) - def main_loop(self): + async def main_loop(self): """Process all events until no more sessions exist. If a caller is a simple little utility, provide a function to @@ -431,7 +431,7 @@ class Console(object): # TODO(jbjohnso): wait_for_rsp is not returning a true value for our # own session while (1): - session.Session.wait_for_rsp(timeout=600) + await session.Session.wait_for_rsp(timeout=600) class ServerConsole(Console): From 8428a47d697e2b317974311d98bfaad0de5535d7 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 07/23] Await the console output flushes ServerConsole._got_sol_payload and Console._got_cons_input both flushed pending output without awaiting the flush, so nothing was written. --- confluent_server/aiohmi/ipmi/console.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/console.py b/confluent_server/aiohmi/ipmi/console.py index 66312bdf..cacf342e 100644 --- a/confluent_server/aiohmi/ipmi/console.py +++ b/confluent_server/aiohmi/ipmi/console.py @@ -197,12 +197,12 @@ class Console(object): else: self.pendingoutput[-1] += data - def _got_cons_input(self, handle): + async def _got_cons_input(self, handle): """Callback for handle events detected by ipmi session""" self._addpendingdata(handle.read()) if not self.awaitingack: - self._sendpendingoutput() + await self._sendpendingoutput() async def close(self): """Shut down an SOL session""" @@ -527,7 +527,7 @@ class ServerConsole(Console): else: self.pendingoutput = [newtext] + self.pendingoutput # self._sendpendingoutput() checks len(self._sendpendingoutput) - self._sendpendingoutput() + await self._sendpendingoutput() elif ackseq != 0 and self.awaitingack: # if an ack packet came in, but did not match what we # expected, retry our payload now. From ec56e2c67c84efaf6ce8fecf95c033e12f109a3f Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 08/23] Rebuild pyghmiutil on the async command API Command lost its constructor for an async create classmethod, so the utility raised TypeError before connecting. The onlogon callback it was built around is gone as well: create establishes the session itself, so both the callback and the eventloop that waited for it are unnecessary. docommand awaited nothing, so every operation produced a coroutine that was printed and dropped, and three of its calls are async generators. Each BMC is handled in turn now rather than only the last. --- confluent_server/aiohmi/cmd/pyghmiutil.py | 48 +++++++++-------------- 1 file changed, 19 insertions(+), 29 deletions(-) diff --git a/confluent_server/aiohmi/cmd/pyghmiutil.py b/confluent_server/aiohmi/cmd/pyghmiutil.py index e16dbed9..383f7f4b 100755 --- a/confluent_server/aiohmi/cmd/pyghmiutil.py +++ b/confluent_server/aiohmi/cmd/pyghmiutil.py @@ -18,51 +18,47 @@ understand how the ipmi_command class workes. """ import asyncio -import functools import os import sys from aiohmi.ipmi import command -def docommand(args, result, ipmisession): +async def docommand(args, ipmicmd): command = args[0] args = args[1:] - print("Logged into %s" % ipmisession.bmc) - if 'error' in result: - print(result['error']) - return + print("Logged into %s" % ipmicmd.bmc) if command == 'power': if args: - print(ipmisession.set_power(args[0], wait=True)) + print(await ipmicmd.set_power(args[0], wait=True)) else: - value = ipmisession.get_power() - print("%s: %s" % (ipmisession.bmc, value['powerstate'])) + value = await ipmicmd.get_power() + print("%s: %s" % (ipmicmd.bmc, value['powerstate'])) elif command == 'bootdev': if args: - print(ipmisession.set_bootdev(args[0])) + print(await ipmicmd.set_bootdev(args[0])) else: - print(ipmisession.get_bootdev()) + print(await ipmicmd.get_bootdev()) elif command == 'sensors': - for reading in ipmisession.get_sensor_data(): + async for reading in ipmicmd.get_sensor_data(): print(reading) elif command == 'health': - print(ipmisession.get_health()) + print(await ipmicmd.get_health()) elif command == 'inventory': - for item in ipmisession.get_inventory(): + async for item in ipmicmd.get_inventory(): print(item) elif command == 'leds': - for led in ipmisession.get_leds(): + async for led in ipmicmd.get_leds(): print(led) elif command == 'graphical': - print(ipmisession.get_graphical_console()) + print(await ipmicmd.get_graphical_console()) elif command == 'net': - print(ipmisession.get_net_configuration()) + print(await ipmicmd.get_net_configuration()) elif command == 'raw': - print(ipmisession.raw_command( + print(await ipmicmd.raw_command( netfn=int(args[0]), command=int(args[1]), - data=map(lambda x: int(x, 16), args[2:]))) + data=[int(x, 16) for x in args[2:]])) async def main(): @@ -74,19 +70,13 @@ async def main(): password = os.environ['IPMIPASSWORD'] os.environ['IPMIPASSWORD'] = "" - bmc = sys.argv[1] + bmcs = sys.argv[1].split(',') userid = sys.argv[2] - bmcs = bmc.split(',') - ipmicmd = None for bmc in bmcs: - # NOTE(etingof): is it right to have `ipmicmd` overridden? - ipmicmd = command.Command( - bmc=bmc, userid=userid, password=password, - onlogon=functools.partial(docommand, sys.argv[3:])) - - if ipmicmd: - await ipmicmd.eventloop() + ipmicmd = await command.Command.create( + bmc=bmc, userid=userid, password=password) + await docommand(sys.argv[3:], ipmicmd) if __name__ == '__main__': From b996a30a443c6291effe527050b099653ecb14bf Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 09/23] Finish porting pyghmicons to async The Console was never connected, so main_loop ran against a session that had never been established. Input arrived on a thread that called send_data and dropped the coroutine; the thread is gone and the loop watches stdin with add_reader instead. The output handler was a plain function that Console._print_data awaits. --- confluent_server/aiohmi/cmd/pyghmicons.py | 45 ++++++++++++++--------- 1 file changed, 27 insertions(+), 18 deletions(-) diff --git a/confluent_server/aiohmi/cmd/pyghmicons.py b/confluent_server/aiohmi/cmd/pyghmicons.py index 0cb0d6f5..2e9d642e 100755 --- a/confluent_server/aiohmi/cmd/pyghmicons.py +++ b/confluent_server/aiohmi/cmd/pyghmicons.py @@ -15,32 +15,39 @@ """ A simple little script to exemplify/test ipmi.console module """ import asyncio +import errno import fcntl import os -import select import sys import termios -import threading import tty from aiohmi.ipmi import console -def _doinput(sol): - while True: - select.select((sys.stdin,), (), (), 600) - try: - data = sys.stdin.read() - except (IOError, OSError) as e: - if e.errno == 11: - continue - raise - - sol.send_data(data) +# Tasks are kept alive here: the event loop only holds weak references to +# them, so a task that nothing else refers to can be collected mid flight. +_pending = set() -def _print(data): +def _got_input(sol): + """Called by the event loop whenever stdin has something to read""" + + try: + data = sys.stdin.read() + except (IOError, OSError) as e: + if e.errno == errno.EAGAIN: + return + raise + if not data: + return + task = asyncio.get_running_loop().create_task(sol.send_data(data)) + _pending.add(task) + task.add_done_callback(_pending.discard) + + +async def _print(data): bailout = False if not isinstance(data, str): bailout = True @@ -73,10 +80,12 @@ async def main(): sol = console.Console(bmc=sys.argv[1], userid=sys.argv[2], password=passwd, iohandler=_print, force=True) - inputthread = threading.Thread(target=_doinput, args=(sol,)) - inputthread.daemon = True - inputthread.start() - await sol.main_loop() + await sol.connect() + asyncio.get_running_loop().add_reader(sys.stdin, _got_input, sol) + try: + await sol.main_loop() + finally: + asyncio.get_running_loop().remove_reader(sys.stdin) except Exception: currfl = fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL) From dfde5736e982fd543f08a85eb6f91ad3b8526e8a Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 10/23] Stop the aiohmi event loop from spinning when it has nothing to do Command.eventloop called wait_for_rsp with no timeout. With nothing waiting or being kept alive there is no deadline to derive one from, so it returns without suspending and the loop runs flat out, measured at over 100000 iterations in two tenths of a second. MAX_IDLE gives it something to wait on, as a ceiling rather than a fixed delay: real deadlines still shorten it and an arriving packet still ends it early. --- confluent_server/aiohmi/ipmi/command.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/confluent_server/aiohmi/ipmi/command.py b/confluent_server/aiohmi/ipmi/command.py index 9d7f55b1..8ffb391d 100644 --- a/confluent_server/aiohmi/ipmi/command.py +++ b/confluent_server/aiohmi/ipmi/command.py @@ -209,7 +209,12 @@ class Command(object): @classmethod async def eventloop(cls): while True: - await session.Session.wait_for_rsp() + # A ceiling rather than no timeout at all: with nothing waiting or + # being kept alive, wait_for_rsp has nothing to wait on and returns + # at once, which would make this a busy loop. Sessions still shorten + # it to their own deadlines, and an arriving packet still ends the + # wait early. + await session.Session.wait_for_rsp(timeout=session.MAX_IDLE) @classmethod async def wait_for_rsp(cls, timeout): From e1071317ed262fed3d0d3e60d6fa63ab8580f599 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 11/23] Create shell sessions through the async factory ConsoleSession grew an async create and lost its constructor, and ShellSession inherits that. sockapi was updated for the console branch but not the shell branch immediately below it, so opening a shell session raised TypeError. It is the only place in the tree that builds one. --- confluent_server/confluent/sockapi.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/confluent_server/confluent/sockapi.py b/confluent_server/confluent/sockapi.py index 51b5dbdd..b3bcc289 100644 --- a/confluent_server/confluent/sockapi.py +++ b/confluent_server/confluent/sockapi.py @@ -288,7 +288,7 @@ async def start_term(authname, cfm, connection, params, path, authdata, skipauth sessionid = elems[5] else: sessionid = None - consession = shellserver.ShellSession( + consession = await shellserver.ShellSession.create( node=node, configmanager=cfm, username=authname, datacallback=ccons.sendall, skipreplay=skipreplay, sessionid=sessionid, width=params.get('width', 80), From 461385522c8df6d517bb0dd96ce56b710597b86c Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 12/23] Accept lastchance in every network manager The retry pass calls apply_configuration with lastchance=True, which only NetworkManager accepts. It is unreachable today, since only NetworkManager returns the 1 that fills the retry list, but it springs the moment either of the others grows a return, or the retry selection is brought in line with the first pass. Matching the signatures costs nothing. --- confluent_osdeploy/common/profile/scripts/confignet | 4 ++-- confluent_osdeploy/debian/profiles/default/scripts/confignet | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/confluent_osdeploy/common/profile/scripts/confignet b/confluent_osdeploy/common/profile/scripts/confignet index 9a972d55..0438f87f 100644 --- a/confluent_osdeploy/common/profile/scripts/confignet +++ b/confluent_osdeploy/common/profile/scripts/confignet @@ -159,7 +159,7 @@ class NetplanManager(object): currinfo[devname]['routes'] = routeinfo currcfg[devname] = currinfo[devname] - def apply_configuration(self, cfg): + def apply_configuration(self, cfg, lastchance=False): devnames = cfg['interfaces'] if len(devnames) > 1: teammode = cfg['settings'].get('team_mode', None) @@ -342,7 +342,7 @@ class WickedManager(object): v = v.strip() currcfg[k] = v - def apply_configuration(self, cfg): + def apply_configuration(self, cfg, lastchance=False): stgs = cfg['settings'] ipcfg = 'STARTMODE=auto\n' routecfg = '' diff --git a/confluent_osdeploy/debian/profiles/default/scripts/confignet b/confluent_osdeploy/debian/profiles/default/scripts/confignet index 61d8d4ff..25591b08 100644 --- a/confluent_osdeploy/debian/profiles/default/scripts/confignet +++ b/confluent_osdeploy/debian/profiles/default/scripts/confignet @@ -151,7 +151,7 @@ class NetplanManager(object): nicinfo[devname]['routes'] = routeinfo self.cfgbydev[devname] = nicinfo[devname] - def apply_configuration(self, cfg): + def apply_configuration(self, cfg, lastchance=False): devnames = cfg['interfaces'] if len(devnames) != 1: raise Exception('Multi-nic team/bonds not yet supported') @@ -266,7 +266,7 @@ class WickedManager(object): v = v.strip() currcfg[k] = v - def apply_configuration(self, cfg): + def apply_configuration(self, cfg, lastchance=False): stgs = cfg['settings'] ipcfg = 'STARTMODE=auto\n' routecfg = '' From 91960527aa9ec49374f1a6df708a5ab8742b41a2 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 13/23] Repair the TSM console connection Three faults in the same few lines. The redfish Command lost its constructor for an async create, so building one raised TypeError, which the except below reported as TargetEndpointUnreachable. await_redirect is defined nowhere in this repository's history, so that call raised too; create performs the session setup it was meant to trigger. And oem is a coroutine method rather than an attribute, with its web connection coming from get_wc, which is what performs the login that sets csrftok. --- .../confluent/plugins/console/tsmsol.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/confluent_server/confluent/plugins/console/tsmsol.py b/confluent_server/confluent/plugins/console/tsmsol.py index e022a4a9..62c1a948 100644 --- a/confluent_server/confluent/plugins/console/tsmsol.py +++ b/confluent_server/confluent/plugins/console/tsmsol.py @@ -108,16 +108,17 @@ class TsmConsole(conapi.Console): kv = util.TLSCertVerifier( self.nodeconfig, self.node, 'pubkeys.tls_hardwaremanager').verify_cert try: - rc = rcmd.Command(self.origbmc, self.username, - self.password, - verifycallback=kv) - await rc.await_redirect() + rc = await rcmd.Command.create(self.origbmc, self.username, + self.password, + verifycallback=kv) + rcoem = await rc.oem() + wc = await rcoem.get_wc() except Exception as e: raise cexc.TargetEndpointUnreachable(str(e)) self.ssl = CustomVerifier(kv) - self.clisess = aiohttp.ClientSession(cookie_jar=rc.oem.wc.cookies) + self.clisess = aiohttp.ClientSession(cookie_jar=wc.cookies) self.ws = await self.clisess.ws_connect( - 'wss://{0}/sol?CSRFTOKEN={1}'.format(self.bmc, rc.oem.csrftok), + 'wss://{0}/sol?CSRFTOKEN={1}'.format(self.bmc, rcoem.csrftok), ssl=self.ssl) self.connected = True self.recvr = tasks.spawn_task(self.recvdata()) From c9d7343e026e10e4572590da7b89c24019bc9d50 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 14/23] Feed console input through a queue A task per read swallowed failures, could deliver keystrokes out of order, and at end of input returned with the reader still registered, so a level triggered selector called it again for the same EOF. The reader queues now, one consumer sends in order, and it is gathered with the main loop so a failure reaches the caller. --- confluent_server/aiohmi/cmd/pyghmicons.py | 27 ++++++++++++++--------- 1 file changed, 17 insertions(+), 10 deletions(-) diff --git a/confluent_server/aiohmi/cmd/pyghmicons.py b/confluent_server/aiohmi/cmd/pyghmicons.py index 2e9d642e..4b2a23d5 100755 --- a/confluent_server/aiohmi/cmd/pyghmicons.py +++ b/confluent_server/aiohmi/cmd/pyghmicons.py @@ -26,12 +26,14 @@ import tty from aiohmi.ipmi import console -# Tasks are kept alive here: the event loop only holds weak references to -# them, so a task that nothing else refers to can be collected mid flight. -_pending = set() +async def _feed_input(sol, inqueue): + """Sends what the reader below collected, one chunk at a time""" + + while True: + await sol.send_data(await inqueue.get()) -def _got_input(sol): +def _got_input(inqueue): """Called by the event loop whenever stdin has something to read""" try: @@ -41,10 +43,11 @@ def _got_input(sol): return raise if not data: + # End of input. The reader is level triggered, so leaving it in place + # would have the loop call this again and again for the same EOF. + asyncio.get_running_loop().remove_reader(sys.stdin) return - task = asyncio.get_running_loop().create_task(sol.send_data(data)) - _pending.add(task) - task.add_done_callback(_pending.discard) + inqueue.put_nowait(data) async def _print(data): @@ -81,11 +84,15 @@ async def main(): sol = console.Console(bmc=sys.argv[1], userid=sys.argv[2], password=passwd, iohandler=_print, force=True) await sol.connect() - asyncio.get_running_loop().add_reader(sys.stdin, _got_input, sol) + inqueue = asyncio.Queue() + loop = asyncio.get_running_loop() + loop.add_reader(sys.stdin, _got_input, inqueue) try: - await sol.main_loop() + # gather rather than a detached task: a send that fails has to + # reach the caller instead of being collected in silence. + await asyncio.gather(sol.main_loop(), _feed_input(sol, inqueue)) finally: - asyncio.get_running_loop().remove_reader(sys.stdin) + loop.remove_reader(sys.stdin) except Exception: currfl = fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL) From e828ed4ff4cdc76dc8b52678e60109169c048d53 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 15/23] Close the TSM console web session TsmConsole created an aiohttp ClientSession and never closed it, and leaked it again when ws_connect failed. Neither was reachable before the connection path was repaired. It is closed on both paths, and starts as None so that closing before a connect does not trip over a missing attribute. --- .../confluent/plugins/console/tsmsol.py | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/confluent_server/confluent/plugins/console/tsmsol.py b/confluent_server/confluent/plugins/console/tsmsol.py index 62c1a948..8c5b00a4 100644 --- a/confluent_server/confluent/plugins/console/tsmsol.py +++ b/confluent_server/confluent/plugins/console/tsmsol.py @@ -83,6 +83,7 @@ class TsmConsole(conapi.Console): self.nodeconfig = config self.connected = False self.recvr = None + self.clisess = None async def recvdata(self): @@ -117,9 +118,14 @@ class TsmConsole(conapi.Console): raise cexc.TargetEndpointUnreachable(str(e)) self.ssl = CustomVerifier(kv) self.clisess = aiohttp.ClientSession(cookie_jar=wc.cookies) - self.ws = await self.clisess.ws_connect( - 'wss://{0}/sol?CSRFTOKEN={1}'.format(self.bmc, rcoem.csrftok), - ssl=self.ssl) + try: + self.ws = await self.clisess.ws_connect( + 'wss://{0}/sol?CSRFTOKEN={1}'.format(self.bmc, rcoem.csrftok), + ssl=self.ssl) + except Exception: + await self.clisess.close() + self.clisess = None + raise self.connected = True self.recvr = tasks.spawn_task(self.recvdata()) return @@ -137,6 +143,10 @@ class TsmConsole(conapi.Console): self.recvr = None if self.ws: await self.ws.close() + self.ws = None + if self.clisess: + await self.clisess.close() + self.clisess = None self.connected = False self.datacallback = None From 7ecc2b28180fa06953f0ed06316c573fe1773620 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 16/23] Replace the housekeeping thread with a loop owned task Housekeeper dates from when this work was a blocking select loop. Under asyncio the event loop is already that place, and a thread around it takes a captured loop reference, a scheduling step before the thread starts, and a daemon flag, only to sit waiting on a coroutine that never returns with no way to stop it. A task runs on the right loop by construction and cancels. The loop is looked up before the coroutine is built, so calling this without one raises rather than stranding it. Nothing in this tree used the class. Out of tree callers need start_housekeeping() instead, and gain the ability to stop it. --- confluent_server/aiohmi/ipmi/command.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/confluent_server/aiohmi/ipmi/command.py b/confluent_server/aiohmi/ipmi/command.py index 8ffb391d..8cdc0ca9 100644 --- a/confluent_server/aiohmi/ipmi/command.py +++ b/confluent_server/aiohmi/ipmi/command.py @@ -19,7 +19,6 @@ from itertools import chain import os import socket import struct -import threading import asyncio import aiohmi.constants as const @@ -107,19 +106,20 @@ def _cidr_to_mask(prefix): return struct.pack('>I', 2 ** prefix - 1 << (32 - prefix)) -class Housekeeper(threading.Thread): - """A Maintenance thread for housekeeping +def start_housekeeping(): + """Run aiohmi's recurring work on the current event loop - Long lived use of aiohmi may warrant some recurring asynchronous behavior. - This stock thread provides a simple minimal context for these housekeeping - tasks to run in. To use, do 'aiohmi.ipmi.command.Maintenance().start()' - and from that point forward, aiohmi should execute any needed ongoing - tasks automatically as needed. This is an alternative to calling - wait_for_rsp or eventloop in a thread of the callers design. + Long lived use of aiohmi may warrant some recurring asynchronous + behaviour. Call this once, from the loop the sessions belong to, and + aiohmi will service them as needed. Cancel the returned task to stop. + This is an alternative to calling wait_for_rsp or eventloop from the + caller's own code. """ - def run(self): - Command.eventloop() + # The loop is fetched first so that calling this without one raises + # before the coroutine exists, rather than leaving it unawaited. + loop = asyncio.get_running_loop() + return loop.create_task(Command.eventloop()) class Command(object): From 8521c6ad4be7e6695d473896bfb4a380f12542e5 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 17/23] Port the IPMI server side to asyncio bmc.py, serversession.py, fakebmc.py and virshbmc.py were byte identical to upstream pyghmi: the async port went through the session layer beneath them and left the server side alone. So every response created a coroutine and dropped it, and the overrides the now async parent awaits returned None. Running fakebmc bound no socket, spun a core, and answered nothing. Everything that sends a response is a coroutine now, and so is the dispatch that reaches it; the hooks a subclass implements stay ordinary functions, so an out of tree Bmc is unaffected unless it overrides the payload handlers. Two things had to leave their constructors, both being coroutines: assigning the server socket, into bind(), which is why listen() is no longer a classmethod, and answering the open session request, into send_open_session_response. Verified against upstream with ipmitool over nine commands, with identical output. SOL is not covered: fakebmc reports the payload disabled on both. --- confluent_server/aiohmi/cmd/fakebmc.py | 3 +- confluent_server/aiohmi/cmd/virshbmc.py | 11 +- confluent_server/aiohmi/ipmi/bmc.py | 82 +++++++------- confluent_server/aiohmi/ipmi/console.py | 20 ++-- .../aiohmi/ipmi/private/serversession.py | 101 +++++++++++------- 5 files changed, 119 insertions(+), 98 deletions(-) diff --git a/confluent_server/aiohmi/cmd/fakebmc.py b/confluent_server/aiohmi/cmd/fakebmc.py index ec24e3e8..5a222241 100755 --- a/confluent_server/aiohmi/cmd/fakebmc.py +++ b/confluent_server/aiohmi/cmd/fakebmc.py @@ -27,6 +27,7 @@ Sent cold reset command to MC """ import argparse +import asyncio import sys import aiohmi.ipmi.bmc as bmc @@ -90,7 +91,7 @@ def main(): help='Port to listen on; defaults to 623') args = parser.parse_args() mybmc = FakeBmc({'admin': 'password'}, port=args.port) - mybmc.listen() + asyncio.run(mybmc.listen()) if __name__ == '__main__': diff --git a/confluent_server/aiohmi/cmd/virshbmc.py b/confluent_server/aiohmi/cmd/virshbmc.py index 787db83c..d5e614e1 100755 --- a/confluent_server/aiohmi/cmd/virshbmc.py +++ b/confluent_server/aiohmi/cmd/virshbmc.py @@ -15,6 +15,7 @@ control a VM """ import argparse +import asyncio import sys import threading @@ -106,16 +107,16 @@ class LibvirtBmc(bmc.Bmc): return self.run_console - def activate_payload(self, request, session): - super(LibvirtBmc, self).activate_payload(request, session) + async def activate_payload(self, request, session): + await super(LibvirtBmc, self).activate_payload(request, session) self.run_console = True self.sol_thread = threading.Thread(target=self.loop) self.sol_thread.start() - def deactivate_payload(self, request, session): + async def deactivate_payload(self, request, session): self.run_console = False self.sol_thread.join() - super(LibvirtBmc, self).deactivate_payload(request, session) + await super(LibvirtBmc, self).deactivate_payload(request, session) def iohandler(self, data): if self.stream: @@ -154,7 +155,7 @@ def main(): hypervisor=args.hypervisor, domain=args.domain, port=args.port) - mybmc.listen() + asyncio.run(mybmc.listen()) if __name__ == '__main__': diff --git a/confluent_server/aiohmi/ipmi/bmc.py b/confluent_server/aiohmi/ipmi/bmc.py index 92e55968..1a35bfa3 100644 --- a/confluent_server/aiohmi/ipmi/bmc.py +++ b/confluent_server/aiohmi/ipmi/bmc.py @@ -60,48 +60,48 @@ class Bmc(serversession.IpmiServer): def is_active(self): raise NotImplementedError - def activate_payload(self, request, session): + async def activate_payload(self, request, session): if self.iohandler is None: - session.send_ipmi_response(code=0x81) + await session.send_ipmi_response(code=0x81) elif not self.is_active(): - session.send_ipmi_response(code=0x81) + await session.send_ipmi_response(code=0x81) elif self.activated: - session.send_ipmi_response(code=0x80) + await session.send_ipmi_response(code=0x80) else: self.activated = True solport = list(struct.unpack('BB', struct.pack('!H', self.port))) - session.send_ipmi_response( + await session.send_ipmi_response( data=[0, 0, 0, 0, 1, 0, 1, 0] + solport + [0xff, 0xff]) self.sol = console.ServerConsole(session, self.iohandler) - def deactivate_payload(self, request, session): + async def deactivate_payload(self, request, session): if self.iohandler is None: - session.send_ipmi_response(code=0x81) + await session.send_ipmi_response(code=0x81) elif not self.activated: - session.send_ipmi_response(code=0x80) + await session.send_ipmi_response(code=0x80) else: - session.send_ipmi_response() + await session.send_ipmi_response() self.sol.close() self.activated = False self.sol = None @staticmethod - def handle_missing_command(session): - session.send_ipmi_response(code=0xc1) + async def handle_missing_command(session): + await session.send_ipmi_response(code=0xc1) - def get_chassis_status(self, session): + async def get_chassis_status(self, session): try: powerstate = self.get_power_state() except NotImplementedError: - return session.send_ipmi_response(code=0xc1) + return await session.send_ipmi_response(code=0xc1) if powerstate in ipmicommand.power_states: powerstate = ipmicommand.power_states[powerstate] if powerstate not in (0, 1): raise Exception('BMC implementation mistake') statusdata = [powerstate, 0, 0] - session.send_ipmi_response(data=statusdata) + await session.send_ipmi_response(data=statusdata) - def control_chassis(self, request, session): + async def control_chassis(self, request, session): rc = 0 try: directive = request['data'][0] @@ -120,79 +120,79 @@ class Bmc(serversession.IpmiServer): rc = self.power_shutdown() if rc is None: rc = 0 - session.send_ipmi_response(code=rc) + await session.send_ipmi_response(code=rc) except NotImplementedError: - session.send_ipmi_response(code=0xcc) + await session.send_ipmi_response(code=0xcc) def get_boot_device(self): raise NotImplementedError - def get_system_boot_options(self, request, session): + async def get_system_boot_options(self, request, session): if request['data'][0] == 5: # boot flags try: bootdevice = self.get_boot_device() except NotImplementedError: - session.send_ipmi_response(data=[1, 5, 0, 0, 0, 0, 0]) + await session.send_ipmi_response(data=[1, 5, 0, 0, 0, 0, 0]) if (type(bootdevice) != int and bootdevice in ipmicommand.boot_devices): bootdevice = ipmicommand.boot_devices[bootdevice] paramdata = [1, 5, 0b10000000, bootdevice, 0, 0, 0] - return session.send_ipmi_response(data=paramdata) + return await session.send_ipmi_response(data=paramdata) else: - session.send_ipmi_response(code=0x80) + await session.send_ipmi_response(code=0x80) def set_boot_device(self, bootdevice): raise NotImplementedError - def set_system_boot_options(self, request, session): + async def set_system_boot_options(self, request, session): if request['data'][0] in (0, 3, 4): # for now, just smile and nod at boot flag bit clearing # implementing it is a burden and implementing it does more to # confuse users than serve a useful purpose - session.send_ipmi_response() + await session.send_ipmi_response() elif request['data'][0] == 5: bootdevice = (request['data'][2] >> 2) & 0b1111 try: bootdevice = ipmicommand.boot_devices[bootdevice] except KeyError: - session.send_ipmi_response(code=0xcc) + await session.send_ipmi_response(code=0xcc) return self.set_boot_device(bootdevice) - session.send_ipmi_response() + await session.send_ipmi_response() else: raise NotImplementedError - def handle_raw_request(self, request, session): + async def handle_raw_request(self, request, session): try: if request['netfn'] == 6: if request['command'] == 1: # get device id - return self.send_device_id(session) + return await self.send_device_id(session) elif request['command'] == 2: # cold reset - return session.send_ipmi_response(code=self.cold_reset()) + return await session.send_ipmi_response(code=self.cold_reset()) elif request['command'] == 0x37: # get system guid guid = self.get_system_guid() - return session.send_ipmi_response(code=0x00, data=guid.bytes_le) + return await session.send_ipmi_response(code=0x00, data=guid.bytes_le) elif request['command'] == 0x48: # activate payload - return self.activate_payload(request, session) + return await self.activate_payload(request, session) elif request['command'] == 0x49: # deactivate payload - return self.deactivate_payload(request, session) + return await self.deactivate_payload(request, session) elif request['netfn'] == 0: if request['command'] == 1: # get chassis status - return self.get_chassis_status(session) + return await self.get_chassis_status(session) elif request['command'] == 2: # chassis control - return self.control_chassis(request, session) + return await self.control_chassis(request, session) elif request['command'] == 8: # set boot options - return self.set_system_boot_options(request, session) + return await self.set_system_boot_options(request, session) elif request['command'] == 9: # get boot options - return self.get_system_boot_options(request, session) - session.send_ipmi_response(code=0xc1) + return await self.get_system_boot_options(request, session) + await session.send_ipmi_response(code=0xc1) except NotImplementedError: - session.send_ipmi_response(code=0xc1) + await session.send_ipmi_response(code=0xc1) except Exception: - session._send_ipmi_net_payload(code=0xff) + await session._send_ipmi_net_payload(code=0xff) traceback.print_exc() - @classmethod - def listen(cls, timeout=30): + async def listen(self, timeout=30): + await self.bind() while True: - ipmisession.Session.wait_for_rsp(timeout) + await ipmisession.Session.wait_for_rsp(timeout) diff --git a/confluent_server/aiohmi/ipmi/console.py b/confluent_server/aiohmi/ipmi/console.py index cacf342e..d204e126 100644 --- a/confluent_server/aiohmi/ipmi/console.py +++ b/confluent_server/aiohmi/ipmi/console.py @@ -463,8 +463,6 @@ class ServerConsole(Console): self.maxoutcount = 256 self.poweredon = True - session.Session.wait_for_rsp(0) - async def _got_sol_payload(self, payload): """SOL payload callback""" @@ -509,7 +507,7 @@ class ServerConsole(Console): # and might be hard to decide what to do in the context of # retry situation try: - self.send_payload(ackpayload, retry=False) + await self.send_payload(ackpayload, retry=False) except exc.IpmiException: # if the session is broken, then close the SOL session self.close() @@ -537,16 +535,16 @@ class ServerConsole(Console): # try to mitigate by avoiding overeager retries # occasional retry of a packet # sooner than timeout suggests is evidently a big deal - self.send_payload(payload=self.lastpayload) + await self.send_payload(payload=self.lastpayload) - def send_payload(self, payload, payload_type=1, retry=True, - needskeepalive=False): + async def send_payload(self, payload, payload_type=1, retry=True, + needskeepalive=False): while not (self.connected or self.broken): - session.Session.wait_for_rsp(timeout=10) - self.ipmi_session.send_payload(payload, - payload_type=payload_type, - retry=retry, - needskeepalive=needskeepalive) + await session.Session.wait_for_rsp(timeout=10) + await self.ipmi_session.send_payload(payload, + payload_type=payload_type, + retry=retry, + needskeepalive=needskeepalive) def close(self): """Shut down an SOL session""" diff --git a/confluent_server/aiohmi/ipmi/private/serversession.py b/confluent_server/aiohmi/ipmi/private/serversession.py index a96a2342..a3288ec2 100644 --- a/confluent_server/aiohmi/ipmi/private/serversession.py +++ b/confluent_server/aiohmi/ipmi/private/serversession.py @@ -86,19 +86,28 @@ class ServerSession(ipmisession.Session): ipmisession.Session.bmc_handlers[clientaddr] = {bmc.port: self} else: ipmisession.Session.bmc_handlers[clientaddr][bmc.port] = self - response = self.create_open_session_response(bytearray(request)) - self.send_payload(response, - constants.payload_types['rmcpplusopenresponse'], - retry=False) + self.initrequest = request - def _got_rmcp_openrequest(self, data): + async def send_open_session_response(self): + """Answer the open session request this was created for + + Sending is a coroutine now, so it cannot happen in __init__ where the + conversation used to begin. + """ + response = self.create_open_session_response( + bytearray(self.initrequest)) + await self.send_payload(response, + constants.payload_types['rmcpplusopenresponse'], + retry=False) + + async def _got_rmcp_openrequest(self, data): response = self.create_open_session_response( struct.pack('B' * len(data), *data)) - self.send_payload(response, - constants.payload_types['rmcpplusopenresponse'], - retry=False) + await self.send_payload(response, + constants.payload_types['rmcpplusopenresponse'], + retry=False) - def _got_rakp1(self, data): + async def _got_rakp1(self, data): clienttag = data[0] self.Rm = data[8:24] self.rolem = data[24] @@ -130,14 +139,14 @@ class ServerSession(ipmisession.Session): # a human newmessage = (bytearray([clienttag, 0, 0, 0]) + self.clientsessionid + self.Rc + uuidbytes + authcode) - self.send_payload(newmessage, constants.payload_types['rakp2'], - retry=False) + await self.send_payload(newmessage, constants.payload_types['rakp2'], + retry=False) - def _got_rakp2(self, data): + async def _got_rakp2(self, data): # stub, server should not think about rakp2 pass - def _got_rakp3(self, data): + async def _got_rakp3(self, data): # for now drop rakp3 with bad authcode # respond correctly a TODO(jjohnson2), since Kg being used # yet incorrect is a scenario why rakp3 could be bad @@ -165,9 +174,9 @@ class ServerSession(ipmisession.Session): return self.localsid = struct.unpack(' Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 18/23] Finish the server side SOL and cleanup paths The console awaits its output handler, but both sample BMCs supplied a plain function, and both dropped the send_data coroutine. virshbmc additionally receives its stream callback on a libvirt thread, so the send goes through run_coroutine_threadsafe against the loop captured at activation, called asyncloop because the class already has a loop method it uses as a thread target. ServerConsole asked the session layer to retry, which a ServerSession cannot do: it never runs Session.__init__, so it has no timeout, and its _timedout does nothing. IpmiServer.logout was synchronous and an argument short while _cleanup awaits logout(False). The boot options handler answered, then read an unbound name and answered again with 0xff. --- confluent_server/aiohmi/cmd/fakebmc.py | 4 ++-- confluent_server/aiohmi/cmd/virshbmc.py | 12 +++++++++--- confluent_server/aiohmi/ipmi/bmc.py | 4 +++- confluent_server/aiohmi/ipmi/console.py | 5 ++++- .../aiohmi/ipmi/private/serversession.py | 2 +- 5 files changed, 19 insertions(+), 8 deletions(-) diff --git a/confluent_server/aiohmi/cmd/fakebmc.py b/confluent_server/aiohmi/cmd/fakebmc.py index 5a222241..3ccbabf8 100755 --- a/confluent_server/aiohmi/cmd/fakebmc.py +++ b/confluent_server/aiohmi/cmd/fakebmc.py @@ -73,10 +73,10 @@ class FakeBmc(bmc.Bmc): def is_active(self): return self.powerstate == 'on' - def iohandler(self, data): + async def iohandler(self, data): print(data) if self.sol: - self.sol.send_data(data) + await self.sol.send_data(data) def main(): diff --git a/confluent_server/aiohmi/cmd/virshbmc.py b/confluent_server/aiohmi/cmd/virshbmc.py index d5e614e1..359ff97a 100755 --- a/confluent_server/aiohmi/cmd/virshbmc.py +++ b/confluent_server/aiohmi/cmd/virshbmc.py @@ -39,8 +39,11 @@ def stream_callback(stream, events, console): data = console.stream.recv(1024) except Exception: return - if console.sol: - console.sol.send_data(data) + if console.sol and console.asyncloop: + # libvirt calls this from its own event thread, and asyncio objects + # are not thread safe, so hand the send to the loop that owns them. + asyncio.run_coroutine_threadsafe(console.sol.send_data(data), + console.asyncloop) class LibvirtBmc(bmc.Bmc): @@ -54,6 +57,7 @@ class LibvirtBmc(bmc.Bmc): self.domain = self.conn.lookupByName(domain) self.state = self.domain.state(0) self.stream = None + self.asyncloop = None self.run_console = False self.conn.domainEventRegister(lifecycle_callback, self) self.sol_thread = None @@ -108,6 +112,8 @@ class LibvirtBmc(bmc.Bmc): return self.run_console async def activate_payload(self, request, session): + # captured for stream_callback, which runs on a libvirt thread + self.asyncloop = asyncio.get_running_loop() await super(LibvirtBmc, self).activate_payload(request, session) self.run_console = True self.sol_thread = threading.Thread(target=self.loop) @@ -118,7 +124,7 @@ class LibvirtBmc(bmc.Bmc): self.sol_thread.join() await super(LibvirtBmc, self).deactivate_payload(request, session) - def iohandler(self, data): + async def iohandler(self, data): if self.stream: self.stream.send(data) diff --git a/confluent_server/aiohmi/ipmi/bmc.py b/confluent_server/aiohmi/ipmi/bmc.py index 1a35bfa3..7ef57226 100644 --- a/confluent_server/aiohmi/ipmi/bmc.py +++ b/confluent_server/aiohmi/ipmi/bmc.py @@ -132,7 +132,9 @@ class Bmc(serversession.IpmiServer): try: bootdevice = self.get_boot_device() except NotImplementedError: - await session.send_ipmi_response(data=[1, 5, 0, 0, 0, 0, 0]) + # nothing more to say, and bootdevice below would be unbound + return await session.send_ipmi_response( + data=[1, 5, 0, 0, 0, 0, 0]) if (type(bootdevice) != int and bootdevice in ipmicommand.boot_devices): bootdevice = ipmicommand.boot_devices[bootdevice] diff --git a/confluent_server/aiohmi/ipmi/console.py b/confluent_server/aiohmi/ipmi/console.py index d204e126..c68ca774 100644 --- a/confluent_server/aiohmi/ipmi/console.py +++ b/confluent_server/aiohmi/ipmi/console.py @@ -541,9 +541,12 @@ class ServerConsole(Console): needskeepalive=False): while not (self.connected or self.broken): await session.Session.wait_for_rsp(timeout=10) + # retry is not passed on: a ServerSession has no retry timer, it + # never runs Session.__init__ and its _timedout does nothing, so + # asking for one only reaches for a timeout attribute it lacks. await self.ipmi_session.send_payload(payload, payload_type=payload_type, - retry=retry, + retry=False, needskeepalive=needskeepalive) def close(self): diff --git a/confluent_server/aiohmi/ipmi/private/serversession.py b/confluent_server/aiohmi/ipmi/private/serversession.py index a3288ec2..934ec0ca 100644 --- a/confluent_server/aiohmi/ipmi/private/serversession.py +++ b/confluent_server/aiohmi/ipmi/private/serversession.py @@ -412,5 +412,5 @@ class IpmiServer(object): # per table 5-2, completion code 0xc1 is 'unrecognized' await session.send_ipmi_response(code=0xc1) - def logout(self): + async def logout(self, sessionok=True): pass From bda403766e613b451268560945cdb47aaf372827 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 19/23] Start and stop the virsh console thread with the payload Activation started an event thread whichever way the base handler had just answered, so a refusal started one anyway and an already active console got a second. activated alone cannot tell the two refusals apart, being true already on the already active path, so the value from before the call decides. Deactivation joined that thread on the event loop, where it could stall every other session and its own response. The wait moves off the loop and is bounded, and the thread is a daemon. --- confluent_server/aiohmi/cmd/virshbmc.py | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/confluent_server/aiohmi/cmd/virshbmc.py b/confluent_server/aiohmi/cmd/virshbmc.py index 359ff97a..03456d6b 100755 --- a/confluent_server/aiohmi/cmd/virshbmc.py +++ b/confluent_server/aiohmi/cmd/virshbmc.py @@ -114,14 +114,32 @@ class LibvirtBmc(bmc.Bmc): async def activate_payload(self, request, session): # captured for stream_callback, which runs on a libvirt thread self.asyncloop = asyncio.get_running_loop() + wasactive = self.activated await super(LibvirtBmc, self).activate_payload(request, session) + if wasactive or not self.activated: + # the base handler refused: no io handler, or the domain is not + # running, so activated stayed false; or a console was already up, + # in which case activated was true before we asked and the thread + # for it is already running. Either way there is nothing to start. + return self.run_console = True self.sol_thread = threading.Thread(target=self.loop) + # virEventRunDefaultImpl can wait for an event that never comes, so + # this thread has no reliable end of its own + self.sol_thread.daemon = True self.sol_thread.start() async def deactivate_payload(self, request, session): - self.run_console = False - self.sol_thread.join() + if self.activated and self.sol_thread: + self.run_console = False + # Joining here would run on the event loop, and the thread sits in + # virEventRunDefaultImpl, which is documented as able to wait + # indefinitely: clearing run_console does not wake it. So wait off + # the loop, and not forever, rather than stall every other session + # and never send the response below. + await asyncio.get_running_loop().run_in_executor( + None, self.sol_thread.join, 5) + self.sol_thread = None await super(LibvirtBmc, self).deactivate_payload(request, session) async def iohandler(self, data): From 11dc8196b0461485eadbbf68db08169313c0cb70 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 20/23] Keep hold of a virsh console thread that will not stop Deactivation dropped its reference once the wait expired, whether or not the thread had stopped, so the next activation started a second one and revived the first by setting run_console again. The reference is cleared only when the thread is really gone, and activation refuses with 0x80 while one is alive. That makes the wait a courtesy rather than a correctness measure, so it drops to a second. --- confluent_server/aiohmi/cmd/virshbmc.py | 23 ++++++++++++++++------- 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/confluent_server/aiohmi/cmd/virshbmc.py b/confluent_server/aiohmi/cmd/virshbmc.py index 03456d6b..63bce006 100755 --- a/confluent_server/aiohmi/cmd/virshbmc.py +++ b/confluent_server/aiohmi/cmd/virshbmc.py @@ -114,6 +114,13 @@ class LibvirtBmc(bmc.Bmc): async def activate_payload(self, request, session): # captured for stream_callback, which runs on a libvirt thread self.asyncloop = asyncio.get_running_loop() + if self.sol_thread is not None and self.sol_thread.is_alive(): + # The thread from the previous console has not come back out of + # virEventRunDefaultImpl. A second one would run a second event + # loop against the same stream, and setting run_console below + # would revive the first one when it finally wakes. + return await session.send_ipmi_response(code=0x80) + self.sol_thread = None wasactive = self.activated await super(LibvirtBmc, self).activate_payload(request, session) if wasactive or not self.activated: @@ -132,14 +139,16 @@ class LibvirtBmc(bmc.Bmc): async def deactivate_payload(self, request, session): if self.activated and self.sol_thread: self.run_console = False - # Joining here would run on the event loop, and the thread sits in - # virEventRunDefaultImpl, which is documented as able to wait - # indefinitely: clearing run_console does not wake it. So wait off - # the loop, and not forever, rather than stall every other session - # and never send the response below. + # The thread only notices that after virEventRunDefaultImpl + # returns, which is documented as possibly never. Waiting on the + # event loop would stall every other session, so wait off it, and + # briefly: this is only to clear the state promptly in the normal + # case. A thread that outlives the wait stays owned here, and + # activate_payload refuses to start another until it is gone. await asyncio.get_running_loop().run_in_executor( - None, self.sol_thread.join, 5) - self.sol_thread = None + None, self.sol_thread.join, 1) + if not self.sol_thread.is_alive(): + self.sol_thread = None await super(LibvirtBmc, self).deactivate_payload(request, session) async def iohandler(self, data): From a53351a730861683586c17dfc6ae03b4028c5685 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:39:28 +0200 Subject: [PATCH 21/23] Give the virsh console loop a reason to wake virEventRunDefaultImpl waits for an event that an idle domain need not produce, so the thread could outlive a deactivation that reported success, and every later activation was refused while it did. Registering a timeout is what makes it return: measured, a thread with nothing registered was still running four seconds after being asked to stop, and with a half second timer it came out at once. --- confluent_server/aiohmi/cmd/virshbmc.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/confluent_server/aiohmi/cmd/virshbmc.py b/confluent_server/aiohmi/cmd/virshbmc.py index 63bce006..c9418dcd 100755 --- a/confluent_server/aiohmi/cmd/virshbmc.py +++ b/confluent_server/aiohmi/cmd/virshbmc.py @@ -156,8 +156,18 @@ class LibvirtBmc(bmc.Bmc): self.stream.send(data) def loop(self): - while self.check_console(): - libvirt.virEventRunDefaultImpl() + # virEventRunDefaultImpl waits for an event, and an idle domain can go + # a long time without producing one. Give it a reason to return, or + # the loop never reconsiders check_console and the thread cannot be + # stopped at all: measured as never waking without this, and returning + # at once with it. + timer = libvirt.virEventAddTimeout(500, lambda *args: None, None) + try: + while self.check_console(): + libvirt.virEventRunDefaultImpl() + finally: + if timer >= 0: + libvirt.virEventRemoveTimeout(timer) def main(): From 1c675c5f24eda70231a489d0b73e9b1b7f7e6f43 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 05:46:36 +0200 Subject: [PATCH 22/23] Port the Eaton PDU plugin to asyncio The plugin was written against the http.client based SecureHTTPConnection, and when that went away the reference was pointed at the aiohttp WebConnection, which shares the name and nothing else. Nothing in it could run: the transport called an async request() without awaiting it and then reached for a getresponse() the new class does not have, and three PDUClient methods that were never coroutines were awaited by the entry points. Two transports now, both local to this plugin. https is aiohttp and stays on the event loop, since the cert verifier records new fingerprints through tasks.spawn. http is http.client in a thread, with its own socket so it can still ask for a smaller segment size before connect: aiohttp only takes a socket factory from 3.12 on, newer than el9, el10, ubuntu 24.04 or Leap 16 ship. That side has no cert to verify and its credentials arrive already read, so the thread touches nothing. connect() establishes and authenticates, wc is just the accessor now, and logout() no longer sends a session id it never obtained. update() reports an unsupported element instead of raising NameError. On the https side cookies follow aiohttp's domain rules and the one POST with a body goes out as text/plain, where http.client replayed every cookie and sent no content type. The http side is as before, and neither can be settled without an Eaton PDU on the bench. Both transports were exercised against a stand-in: login, outlet read and set, sensors, logout, and a clamped segment size on the plaintext path. --- .../plugins/hardwaremanagement/eatonpdu.py | 178 ++++++++++-------- 1 file changed, 97 insertions(+), 81 deletions(-) diff --git a/confluent_server/confluent/plugins/hardwaremanagement/eatonpdu.py b/confluent_server/confluent/plugins/hardwaremanagement/eatonpdu.py index f6c0a3ac..fcc3d8dd 100644 --- a/confluent_server/confluent/plugins/hardwaremanagement/eatonpdu.py +++ b/confluent_server/confluent/plugins/hardwaremanagement/eatonpdu.py @@ -12,12 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. -#TODO: ASYNC asyncio conversion import asyncio import base64 import confluent.util as util import confluent.messages as msg import aiohmi.util.webclient as wc +import http.client as httplib +import http.cookies as Cookie import re import hashlib import json @@ -55,86 +56,93 @@ def answer_challenge(username, password, data): return {'sessionKey': skey.decode('utf8'), 'szResponse': rsp.decode('utf8'), 'szResponseValue': s2rsp.decode('utf8')} -import http.client as httplib -import http.cookies as Cookie +# carried over verbatim from the transport these replace +_fixedheaders = { + 'Host': 'pdu.cluster.net', + 'Accept': '*/*', + 'Accept-Language': 'en-US,en;q=0.9', + 'Connection': 'close', + 'Referer': 'http://pdu.cluster.net/setting_admin.htm', +} -# Delta PDU webserver always closes connection, -# replace conditionals with always close + +class SecureWebConnection(wc.WebConnection): + """The https transport, on the loop: the cert verifier needs it""" + + def __init__(self, host, verifycallback): + wc.WebConnection.__init__(self, host, 443, + verifycallback=verifycallback) + for header in _fixedheaders: + self.set_header(header, _fixedheaders[header]) + + async def grab_response(self, url, body=None, method=None): + # unfollowed, as http.client left them + rsp, status, _ = await self.grab_response_with_status( + url, body, method=method, allow_redirects=False) + return rsp, status + + +# the device closes the connection whatever it said it would do class WebResponse(httplib.HTTPResponse): def _check_close(self): return True -class WebConnection(wc.WebConnection): + +class PlainWebConnection(httplib.HTTPConnection): + """The http transport, for a PDU with TLS not turned on + + http.client, because the socket wants a smaller segment size set before + connect and aiohttp only takes a socket factory from 3.12 on, which is + newer than most distros ship. Requests run in a thread, which is safe + here: no cert to verify, and the credentials arrive already read. + """ + response_class = WebResponse - def __init__(self, host, secure, verifycallback): - if secure: - port = 443 - else: - port = 80 - wc.WebConnection.__init__(self, host, port, verifycallback=verifycallback) - self.secure = secure + + def __init__(self, host): + # http.client would otherwise wait forever + httplib.HTTPConnection.__init__(self, host, 80, timeout=60) + self.stdheaders = dict(_fixedheaders) self.cookies = {} - async def connect(self): - if self.secure: - return super(WebConnection, self).connect() - addrinfo = (await asyncio.get_running_loop().getaddrinfo(self.host, self.port))[0] - # workaround problems of too large mtu, moderately frequent occurance - # in this space + def connect(self): + addrinfo = socket.getaddrinfo(self.host, self.port, 0, + socket.SOCK_STREAM)[0] plainsock = socket.socket(addrinfo[0]) - plainsock.settimeout(self.mytimeout) + plainsock.settimeout(self.timeout) try: + # workaround problems of too large mtu, moderately frequent + # occurance in this space plainsock.setsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG, 1456) except socket.error: pass plainsock.connect(addrinfo[4]) self.sock = plainsock - - def getresponse(self): - try: - rsp = super(WebConnection, self).getresponse() - try: - hdrs = [x.split(':', 1) for x in rsp.msg.headers] - except AttributeError: - hdrs = rsp.msg.items() - for hdr in hdrs: - if hdr[0] == 'Set-Cookie': - c = Cookie.BaseCookie(hdr[1]) - for k in c: - self.cookies[k] = c[k].value - except httplib.BadStatusLine: - self.broken = True - raise - return rsp - def request(self, method, url, body=None): - headers = {} + def _blocking_grab(self, url, body, method): + headers = dict(self.stdheaders) if body: headers['Content-Length'] = len(body) - cookies = [] - for cookie in self.cookies: - cookies.append('{0}={1}'.format(cookie, self.cookies[cookie])) - headers['Cookie'] = ';'.join(cookies) - headers['Host'] = 'pdu.cluster.net' - headers['Accept'] = '*/*' - headers['Accept-Language'] = 'en-US,en;q=0.9' - headers['Connection'] = 'close' - headers['Referer'] = 'http://pdu.cluster.net/setting_admin.htm' - return super(WebConnection, self).request(method, url, body, headers) + if self.cookies: + headers['Cookie'] = ';'.join( + ['{0}={1}'.format(k, self.cookies[k]) for k in self.cookies]) + self.request(method, url, body, headers) + rsp = self.getresponse() + for hdr, value in rsp.getheaders(): + if hdr.lower() == 'set-cookie': + cookie = Cookie.BaseCookie(value) + for key in cookie: + self.cookies[key] = cookie[key].value + return rsp.read(), rsp.status - def grab_response(self, url, body=None, method=None): + async def grab_response(self, url, body=None, method=None): if method is None: method = 'GET' if body is None else 'POST' - if body: - self.request(method, url, body) - else: - self.request(method, url) - rsp = self.getresponse() - body = rsp.read() - return body, rsp.status + return await asyncio.to_thread(self._blocking_grab, url, body, method) + _sensors_by_node = {} -def get_sensor_data(element, node, configmanager): +async def get_sensor_data(element, node, configmanager): category, name = element[-2:] justnames = False readings = [] @@ -149,9 +157,9 @@ def get_sensor_data(element, node, configmanager): if not sn or sn[1] < time.time(): gc = PDUClient(node, configmanager) try: - sdata = gc.get_sensor_data() + sdata = await gc.get_sensor_data() finally: - gc.logout() + await gc.logout() _sensors_by_node[node] = [sdata, time.time() + 1] sn = _sensors_by_node.get(node, None) for outlet in sn[0]: @@ -183,6 +191,11 @@ class PDUClient(object): @property def wc(self): + # set by connect(); only login() reads it without connecting first + return self._wc + + async def connect(self): + # logging in is a coroutine, so this cannot be the wc property if self._wc: return self._wc targcfg = self.configmanager.get_node_attributes(self.node, @@ -197,18 +210,18 @@ class PDUClient(object): verifier = util.TLSCertVerifier( self.configmanager, self.node, 'pubkeys.tls_hardwaremanager') try: - self._wc = WebConnection(target, secure=True, verifycallback=verifier.verify_cert) - self.login(self.configmanager) + self._wc = SecureWebConnection(target, verifier.verify_cert) + await self.login(self.configmanager) except socket.error: pkey = self.configmanager.get_node_attributes(self.node, 'pubkeys.tls_hardwaremanager') pkey = pkey.get(self.node, {}).get('pubkeys.tls_hardwaremanager', {}).get('value', None) if pkey: raise - self._wc = WebConnection(target, secure=False, verifycallback=verifier.verify_cert) - self.login(self.configmanager) + self._wc = PlainWebConnection(target) + await self.login(self.configmanager) return self._wc - def login(self, configmanager): + async def login(self, configmanager): credcfg = configmanager.get_node_attributes(self.node, ['secret.hardwaremanagementuser', 'secret.hardwaremanagementpassword'], @@ -226,7 +239,7 @@ class PDUClient(object): raise Exception('Missing username or password') b64user = base64.b64encode(username.encode('utf8')).decode('utf8') b64pass = base64.b64encode(passwd.encode('utf8')).decode('utf8') - rsp = self.wc.grab_response('/config/gateway?page=cgi_authentication&login={}&_dc={}'.format(b64user, int(time.time()))) + rsp = await self.wc.grab_response('/config/gateway?page=cgi_authentication&login={}&_dc={}'.format(b64user, int(time.time()))) rsp = json.loads(sanitize_json(rsp[0])) self.sessid = rsp['data'][0] if rsp['data'][-1] == 'password': @@ -246,22 +259,24 @@ class PDUClient(object): parms['szResponseValue'], int(time.time()), ) - rsp = self.wc.grab_response(url) + rsp = await self.wc.grab_response(url) rsp = json.loads(sanitize_json(rsp[0])) if not rsp['success']: raise Exception('Failed to login to device') - rsp = self.wc.grab_response('/config/gateway?page=cgi_checkUserSession&sessionId={}&_dc={}'.format(self.sessid, int(time.time()))) + rsp = await self.wc.grab_response('/config/gateway?page=cgi_checkUserSession&sessionId={}&_dc={}'.format(self.sessid, int(time.time()))) - def do_request(self, suburl): - wc = self.wc + async def do_request(self, suburl): + wc = await self.connect() url = '/config/gateway?page={}&sessionId={}&_dc={}'.format(suburl, self.sessid, int(time.time())) - return wc.grab_response(url) + return await wc.grab_response(url) - def logout(self): - self.do_request('cgi_logout') + async def logout(self): + if self.sessid: + await self.do_request('cgi_logout') + self.sessid = None - def get_outlet(self, outlet): - rsp = self.do_request('cgi_pdu_outlets') + async def get_outlet(self, outlet): + rsp = await self.do_request('cgi_pdu_outlets') data = sanitize_json(rsp[0]) data = json.loads(data) data = data['data'][0] @@ -271,8 +286,8 @@ class PDUClient(object): return 'on' if outdata[3] else 'off' return - def get_sensor_data(self): - rsp = self.do_request('cgi_pdu_outlets') + async def get_sensor_data(self): + rsp = await self.do_request('cgi_pdu_outlets') data = sanitize_json(rsp[0]) data = json.loads(data) data = data['data'][0] @@ -293,8 +308,9 @@ class PDUClient(object): sdata[outletname] = outsense return sdata - def set_outlet(self, outlet, state): - rsp = self.do_request('cgi_pdu_outlets') + async def set_outlet(self, outlet, state): + wc = await self.connect() + rsp = await self.do_request('cgi_pdu_outlets') data = sanitize_json(rsp[0]) data = json.loads(data) data = data['data'][0] @@ -303,14 +319,14 @@ class PDUClient(object): outdata = outdata[0] if outdata[0] == outlet: payload = "0".format(idx, 'Startup' if state == 'on' else 'Shutdown') - rsp = self.wc.grab_response('/config/set_object_mass.xml?sessionId={}'.format(self.sessid), payload) + rsp = await wc.grab_response('/config/set_object_mass.xml?sessionId={}'.format(self.sessid), payload) return idx += 1 async def retrieve(nodes, element, configmanager, inputdata): if element[0] == 'sensors': for node in nodes: - for res in get_sensor_data(element, node, configmanager): + async for res in get_sensor_data(element, node, configmanager): yield res return elif 'outlets' not in element: From a8602118177e65dac4144a691ab9247bc8885215 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 10 Aug 2026 04:53:35 +0200 Subject: [PATCH 23/23] Make the pyrefly job blocking The tree is clean under it now, so the job can fail the run and catch the next async regression instead of only reporting one. Nothing is suppressed beyond the two ignores that state their reason at the line, for an async __new__ and an untyped callback registry, neither of which pyrefly can model. --- .github/workflows/ci.yml | 5 ----- 1 file changed, 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 641d3b18..7aa90e5e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,11 +55,6 @@ jobs: pyrefly: name: Pyrefly (async correctness) runs-on: ubuntu-latest - # Advisory for now: the tree still has findings, a good number of them - # real bugs, so this reports without failing the run. It also means an - # install failure or a checker crash passes silently. Remove the line once - # the findings are dealt with, which is the point of having it. - continue-on-error: true steps: - uses: actions/checkout@v7 - uses: facebook/pyrefly@main