xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-08-28 01:26:44 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #267 from Obihoernchen/pyrefly

Browse Source 
Add pyrefly for async correctness
This commit is contained in:
Jarrod Johnson
2026-08-10 09:10:34 -04:00
committed by GitHub
parent 4fe1532da9 a860211817
commit cfcb406dba
32 changed files with 645 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+7
View File
@@ -52,6 +52,13 @@ jobs:
# workspace can be handed over as-is.
args: check --output-format=github
pyrefly:
name: Pyrefly (async correctness)
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- uses: facebook/pyrefly@main
python-compileall:
name: Python compileall
runs-on: ubuntu-latest
confluent_osdeploy/common/profile/scripts/confignet
+2 -2
View File
@@ -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 = ''
confluent_osdeploy/debian/profiles/default/scripts/confignet
+2 -2
View File
@@ -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 = ''
confluent_server/aiohmi/cmd/fakebmc.py
+4 -3
View File
@@ -27,6 +27,7 @@ Sent cold reset command to MC
"""
import argparse
import asyncio
import sys
import aiohmi.ipmi.bmc as bmc
@@ -72,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():
@@ -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__':
confluent_server/aiohmi/cmd/pyghmicons.py
+36 -19
View File
@@ -14,32 +14,43 @@
""" 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):
async def _feed_input(sol, inqueue):
"""Sends what the reader below collected, one chunk at a time"""
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)
await sol.send_data(await inqueue.get())
def _print(data):
def _got_input(inqueue):
"""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:
# 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
inqueue.put_nowait(data)
async def _print(data):
bailout = False
if not isinstance(data, str):
bailout = True
@@ -50,7 +61,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
@@ -72,10 +83,16 @@ 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()
sol.main_loop()
await sol.connect()
inqueue = asyncio.Queue()
loop = asyncio.get_running_loop()
loop.add_reader(sys.stdin, _got_input, inqueue)
try:
# 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:
loop.remove_reader(sys.stdin)
except Exception:
currfl = fcntl.fcntl(sys.stdin.fileno(), fcntl.F_GETFL)
@@ -85,4 +102,4 @@ def main():
if __name__ == '__main__':
sys.exit(main())
sys.exit(asyncio.run(main()))
confluent_server/aiohmi/cmd/pyghmiutil.py
+22 -31
View File
@@ -17,54 +17,51 @@ it isn't conceived as a general utility to actually use, just help developers
understand how the ipmi_command class workes.
"""
import functools
import asyncio
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:]]))
def main():
async def main():
if (len(sys.argv) < 3) or 'IPMIPASSWORD' not in os.environ:
print("Usage:")
print(" IPMIPASSWORD=password %s bmc username <cmd> <optarg>" %
@@ -73,20 +70,14 @@ 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:
ipmicmd.eventloop()
ipmicmd = await command.Command.create(
bmc=bmc, userid=userid, password=password)
await docommand(sys.argv[3:], ipmicmd)
if __name__ == '__main__':
sys.exit(main())
sys.exit(asyncio.run(main()))
confluent_server/aiohmi/cmd/virshbmc.py
+56 -12
View File
@@ -15,6 +15,7 @@ control a VM
"""
import argparse
import asyncio
import sys
import threading
@@ -38,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):
@@ -53,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
@@ -106,24 +111,63 @@ 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):
# 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:
# 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()
def deactivate_payload(self, request, session):
self.run_console = False
self.sol_thread.join()
super(LibvirtBmc, self).deactivate_payload(request, session)
async def deactivate_payload(self, request, session):
if self.activated and self.sol_thread:
self.run_console = False
# 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, 1)
if not self.sol_thread.is_alive():
self.sol_thread = None
await super(LibvirtBmc, self).deactivate_payload(request, session)
def iohandler(self, data):
async def iohandler(self, data):
if self.stream:
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():
@@ -154,7 +198,7 @@ def main():
hypervisor=args.hypervisor,
domain=args.domain,
port=args.port)
mybmc.listen()
asyncio.run(mybmc.listen())
if __name__ == '__main__':
confluent_server/aiohmi/ipmi/bmc.py
+43 -41
View File
@@ -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,81 @@ 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])
# 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]
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)
confluent_server/aiohmi/ipmi/command.py
+17 -12
View File
@@ -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):
@@ -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):
confluent_server/aiohmi/ipmi/console.py
+21 -17
View File
@@ -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
@@ -194,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"""
@@ -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')
@@ -415,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
@@ -428,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):
@@ -460,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"""
@@ -506,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()
@@ -524,7 +525,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.
@@ -534,16 +535,19 @@ 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)
# 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=False,
needskeepalive=needskeepalive)
def close(self):
"""Shut down an SOL session"""
confluent_server/aiohmi/ipmi/private/serversession.py
+62 -41
View File
@@ -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('<I', self.managedsessionid)[0]
self.ipmicallback = self.handle_client_request
self._send_rakp4(clienttag, 0)
await self._send_rakp4(clienttag, 0)
def handle_client_request(self, request):
async def handle_client_request(self, request):
if request['netfn'] == 6 and request['command'] == 0x3b:
pendingpriv = request['data'][0]
returncode = 0
@@ -176,37 +185,37 @@ class ServerSession(ipmisession.Session):
returncode = 0x81
else:
self.clientpriv = request['data'][0]
self._send_ipmi_net_payload(code=returncode,
data=[self.clientpriv])
await self._send_ipmi_net_payload(code=returncode,
data=[self.clientpriv])
elif request['netfn'] == 6 and request['command'] == 0x3c:
self.send_ipmi_response()
await self.send_ipmi_response()
self.close_server_session()
else:
self.bmc.handle_raw_request(request, self)
await self.bmc.handle_raw_request(request, self)
def close_server_session(self):
pass
def _send_rakp4(self, tagvalue, statuscode):
async def _send_rakp4(self, tagvalue, statuscode):
payload = bytearray(
[tagvalue, statuscode, 0, 0]) + self.clientsessionid
hmacdata = self.Rm + self.managedsessionid + self.uuiddata
hmacdata = struct.pack('%dB' % len(hmacdata), *hmacdata)
authdata = hmac.new(self.sik, hmacdata, hashlib.sha1).digest()[:12]
payload += authdata
self.send_payload(payload, constants.payload_types['rakp4'],
retry=False)
await self.send_payload(payload, constants.payload_types['rakp4'],
retry=False)
self.confalgo = 'aes'
self.integrityalgo = 'sha1'
self.sequencenumber = 1
self.sessionid = struct.unpack(
'<I', struct.pack('4B', *self.clientsessionid))[0]
def _got_rakp4(self, data):
async def _got_rakp4(self, data):
# stub, server should not think about rakp4
pass
def _timedout(self):
async def _timedout(self):
"""Expire a client session after a period of inactivity
After the session inactivity timeout, this invalidate the client
@@ -223,10 +232,10 @@ class ServerSession(ipmisession.Session):
"""
pass
def send_ipmi_response(self, data=[], code=0):
self._send_ipmi_net_payload(data=data, code=code)
async def send_ipmi_response(self, data=[], code=0):
await self._send_ipmi_net_payload(data=data, code=code)
def logout(self):
async def logout(self, sessionok=True):
pass
@@ -272,9 +281,20 @@ class IpmiServer(object):
self.kg = None
self.timeout = 60
self.port = port
addrinfo = socket.getaddrinfo(address, port, 0,
socket.SOCK_DGRAM)[0]
self.serversocket = ipmisession.Session._assignsocket(addrinfo)
self.addrinfo = socket.getaddrinfo(address, port, 0,
socket.SOCK_DGRAM)[0]
self.serversocket = None
async def bind(self):
"""Open the socket this server listens on
Deferred from __init__ because assigning a socket is a coroutine.
listen() does it, so a caller that uses listen() need not care.
"""
if self.serversocket is not None:
return
self.serversocket = await ipmisession.Session._assignsocket(
self.addrinfo)
ipmisession.Session.bmc_handlers[self.serversocket] = {0: self}
def send_auth_cap(self, myaddr, mylun, clientaddr, clientlun, clientseq,
@@ -290,10 +310,10 @@ class IpmiServer(object):
header.append(ipmisession._checksum(*bodydata))
ipmisession._io_sendto(self.serversocket, header, sockaddr)
def process_pktqueue(self):
async def process_pktqueue(self):
while self.pktqueue:
pkt = self.pktqueue.popleft()
self.sessionless_data(pkt[0], pkt[1])
await self.sessionless_data(pkt[0], pkt[1])
def send_cipher_suites(self, myaddr, mylun, clientaddr, clientlun,
clientseq, data, sockaddr):
@@ -316,7 +336,7 @@ class IpmiServer(object):
pkt = header + ipmihdr + rq
ipmisession._io_sendto(self.serversocket, pkt, sockaddr)
def sessionless_data(self, data, sockaddr):
async def sessionless_data(self, data, sockaddr):
"""Examines unsolocited packet and decides appropriate action.
For a listening IpmiServer, a packet without an active session
@@ -335,9 +355,10 @@ class IpmiServer(object):
if payloadtype not in (0, 16):
return
if payloadtype == 16: # new session to handle conversation
ServerSession(self.authdata, self.kg, sockaddr,
self.serversocket, data[16:], self.uuid,
bmc=self)
newsession = ServerSession(self.authdata, self.kg, sockaddr,
self.serversocket, data[16:],
self.uuid, bmc=self)
await newsession.send_open_session_response()
return
# ditch two byte, because ipmi2 header is two
# bytes longer than ipmi1 (payload type added, payload length 2).
@@ -379,17 +400,17 @@ class IpmiServer(object):
except AttributeError:
self.kg = kg
def send_device_id(self, session):
async def send_device_id(self, session):
response = [self.deviceid, self.revision, self.firmwaremajor,
self.firmwareminor, self.ipmiversion,
self.additionaldevices]
response += struct.unpack('4B', struct.pack('<I', self.mfgid))
response += struct.unpack('4B', struct.pack('<I', self.prodid))
session.send_ipmi_response(data=response)
await session.send_ipmi_response(data=response)
def handle_raw_request(self, request, session):
async def handle_raw_request(self, request, session):
# per table 5-2, completion code 0xc1 is 'unrecognized'
session.send_ipmi_response(code=0xc1)
await session.send_ipmi_response(code=0xc1)
def logout(self):
async def logout(self, sessionok=True):
pass
confluent_server/aiohmi/ipmi/private/session.py
+3
View File
@@ -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
confluent_server/aiohmi/redfish/oem/lenovo/xcc.py
+5 -5
View File
@@ -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
confluent_server/aiohmi/util/webclient.py
+2 -1
View File
@@ -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':
confluent_server/confluent/discovery/handlers/megarac.py
+1 -1
View File
@@ -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]))
confluent_server/confluent/discovery/handlers/redfishbmc.py
+1 -1
View File
@@ -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]))
confluent_server/confluent/discovery/handlers/smm.py
+58 -54
View File
@@ -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):
@@ -49,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'
@@ -62,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:
@@ -80,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)
@@ -103,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:
@@ -116,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 '<statusCode>0' not in rspdata:
raise Exception("Error configuring SMM Network")
return
@@ -128,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)
@@ -181,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)
@@ -245,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)
confluent_server/confluent/discovery/handlers/tsm.py
+1 -1
View File
@@ -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]))
confluent_server/confluent/discovery/handlers/xcc3.py
+4 -3
View File
@@ -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]))
confluent_server/confluent/discovery/protocols/mdns.py
+1 -1
View File
@@ -480,4 +480,4 @@ from pprint import pprint
if __name__ == '__main__':
def printit(rsp):
print(repr(rsp))
snoop(pprint)
asyncio.run(snoop(pprint))
confluent_server/confluent/discovery/protocols/pxe.py
+1 -1
View File
@@ -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))
confluent_server/confluent/discovery/protocols/slp.py
+1 -1
View File
@@ -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))
confluent_server/confluent/networking/lldp.py
-1
View File
@@ -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))
confluent_server/confluent/plugins/console/tsmsol.py
+19 -8
View File
@@ -83,6 +83,7 @@ class TsmConsole(conapi.Console):
self.nodeconfig = config
self.connected = False
self.recvr = None
self.clisess = None
async def recvdata(self):
@@ -108,17 +109,23 @@ 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.ws = await self.clisess.ws_connect(
'wss://{0}/sol?CSRFTOKEN={1}'.format(self.bmc, rc.oem.csrftok),
ssl=self.ssl)
self.clisess = aiohttp.ClientSession(cookie_jar=wc.cookies)
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
@@ -136,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
confluent_server/confluent/plugins/hardwaremanagement/deltapdu.py
+1 -1
View File
@@ -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):
confluent_server/confluent/plugins/hardwaremanagement/eatonpdu.py
+97 -81
View File
@@ -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 = "<SET_OBJECT><OBJECT name='PDU.OutletSystem.Outlet[{}].DelayBefore{}'>0</OBJECT>".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:
confluent_server/confluent/plugins/hardwaremanagement/proxmox.py
+11 -7
View File
@@ -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())
confluent_server/confluent/plugins/hardwaremanagement/redfish.py
+6 -6
View File
@@ -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':
confluent_server/confluent/plugins/hardwaremanagement/vcenter.py
+12 -8
View File
@@ -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())
confluent_server/confluent/sockapi.py
+2 -2
View File
@@ -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 '
@@ -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),
confluent_server/confluent/sshutil.py
+9 -4
View File
@@ -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())
pyrefly.toml
+138
View File
@@ -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"