xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-10-20 14:45:45 +00:00
Code Issues Projects Releases Wiki Activity

Port the IPMI server side to asyncio

Browse Source 
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.
This commit is contained in:
Markus Hilger
2026-08-10 04:39:28 +02:00
parent 7ecc2b2818
commit 8521c6ad4b
5 changed files with 119 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
confluent_server/aiohmi/cmd/fakebmc.py
+2 -1
View File
@@ -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__':
confluent_server/aiohmi/cmd/virshbmc.py
+6 -5
View File
@@ -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__':
confluent_server/aiohmi/ipmi/bmc.py
+41 -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,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)
confluent_server/aiohmi/ipmi/console.py
+9 -11
View File
@@ -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"""
confluent_server/aiohmi/ipmi/private/serversession.py
+61 -40
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):
pass