From cc2d2f27002565d4bab61c76c8dac5ccbe45aabd Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:19:49 +0200 Subject: [PATCH 01/16] msgs: add the missing commas to the SEL message fields The __fields__ of SetSelTimeReq, SetSelTimeRsp and AddSelEntryReq were a single field instead of a tuple, because the trailing comma was missing. Creating one of these messages failed with a TypeError, as the field is not iterable. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/msgs/sel.py | 8 ++++---- tests/msgs/test_sel.py | 18 ++++++++++++++++++ 2 files changed, 22 insertions(+), 4 deletions(-) diff --git a/pyipmi/msgs/sel.py b/pyipmi/msgs/sel.py index 54668c8c..1d536580 100644 --- a/pyipmi/msgs/sel.py +++ b/pyipmi/msgs/sel.py @@ -116,7 +116,7 @@ class AddSelEntryReq(Message): __cmdid__ = constants.CMDID_ADD_SEL_ENTRY __netfn__ = constants.NETFN_STORAGE __fields__ = ( - ByteArray('record_data', 16) + ByteArray('record_data', 16), ) @@ -185,7 +185,7 @@ class GetSelTimeRsp(Message): __netfn__ = constants.NETFN_STORAGE | 1 __fields__ = ( CompletionCode(), - Timestamp('timestamp') + Timestamp('timestamp'), ) @@ -194,7 +194,7 @@ class SetSelTimeReq(Message): __cmdid__ = constants.CMDID_SET_SEL_TIME __netfn__ = constants.NETFN_STORAGE __fields__ = ( - Timestamp('timestamp') + Timestamp('timestamp'), ) @@ -203,5 +203,5 @@ class SetSelTimeRsp(Message): __cmdid__ = constants.CMDID_SET_SEL_TIME __netfn__ = constants.NETFN_STORAGE | 1 __fields__ = ( - CompletionCode() + CompletionCode(), ) diff --git a/tests/msgs/test_sel.py b/tests/msgs/test_sel.py index 115971c7..d98347fd 100644 --- a/tests/msgs/test_sel.py +++ b/tests/msgs/test_sel.py @@ -38,3 +38,21 @@ def test_getselentry_encode_valid_rsp(): m.record_data = array('B', b'\x01\x02\x03\x04') data = encode_message(m) assert data == b'\x00\x02\x01\x01\x02\x03\x04' + + +def test_setseltime_encode_req(): + m = pyipmi.msgs.sel.SetSelTimeReq() + m.timestamp = 0x01020304 + assert encode_message(m) == b'\x04\x03\x02\x01' + + +def test_setseltime_decode_rsp(): + m = pyipmi.msgs.sel.SetSelTimeRsp() + decode_message(m, b'\x00') + assert m.completion_code == 0x00 + + +def test_addselentry_encode_req(): + m = pyipmi.msgs.sel.AddSelEntryReq() + m.record_data = array('B', range(16)) + assert encode_message(m) == bytes(range(16)) From 3c55ac65a062a0eb4c768c5bc5ccf80625d2d127 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:19:58 +0200 Subject: [PATCH 02/16] fields: decode BCD+ strings given as array The SDR passes the device ID string as array to SdrTypeLengthString. An array has no decode() method, so a BCD+ encoded device ID string failed with an AttributeError. Convert the data to bytes before decoding it. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/fields.py | 2 +- tests/test_fields.py | 11 ++++++++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/pyipmi/fields.py b/pyipmi/fields.py index bbcd88fd..6e26aaad 100644 --- a/pyipmi/fields.py +++ b/pyipmi/fields.py @@ -98,7 +98,7 @@ def _from_data(self, data: bytes, offset: int = 0, self.raw = data[offset+1:offset+1+self.length] if self.field_type == self.TYPE_BCD_PLUS: - self.string = self.raw.decode('bcd+') + self.string = bytes(self.raw).decode('bcd+') elif self.field_type == self.TYPE_6BIT_ASCII: self.string = _unpack6bitascii(self.raw) else: diff --git a/tests/test_fields.py b/tests/test_fields.py index 99291001..2640f7aa 100644 --- a/tests/test_fields.py +++ b/tests/test_fields.py @@ -2,7 +2,10 @@ import pytest -from pyipmi.fields import (VersionField, FruTypeLengthString) +from array import array + +from pyipmi.fields import (VersionField, FruTypeLengthString, + SdrTypeLengthString) from pyipmi.errors import DecodingError @@ -30,3 +33,9 @@ def test_versionfield_decoding_error(): def test_FruTypeLengthString_6bitascii(): f = FruTypeLengthString(b'\x83d\xc9\xb2\xde', 0) assert f.string == 'DELL' + + +def test_SdrTypeLengthString_bcd_plus_from_array(): + # the SDR passes the device ID string as array + f = SdrTypeLengthString(data=array('B', [0x42, 0x12, 0x34])) + assert f.string == '1234' From 868dc9bb381ea55de3af7c3dd4125ebb85957978 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:20:20 +0200 Subject: [PATCH 03/16] fru: read to the end of the area if only the offset is given read_fru_data() with an offset but without a count failed with a TypeError, because the end of the read was calculated from the count. Read from the offset to the end of the FRU inventory area instead. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/fru.py | 4 ++-- tests/test_fru.py | 5 +++++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/pyipmi/fru.py b/pyipmi/fru.py index 3d10d705..5056b3d8 100644 --- a/pyipmi/fru.py +++ b/pyipmi/fru.py @@ -66,9 +66,9 @@ def read_fru_data(self, offset: int | None = None, data = array.array('B') # first check for maximum area size - if offset is None: + if offset is None or count is None: area_size = self.get_fru_inventory_area_info(fru_id) - off = 0 + off = offset or 0 else: area_size = offset + count off = offset diff --git a/tests/test_fru.py b/tests/test_fru.py index ef35dec0..6a88e553 100644 --- a/tests/test_fru.py +++ b/tests/test_fru.py @@ -218,6 +218,11 @@ def test_read_fru_data(): assert fru.requests == [(0, 32), (32, 32), (64, 32), (96, 4)] +def test_read_fru_data_offset_without_count(): + fru = FakeFruDevice(FRU_DATA) + assert fru.read_fru_data(offset=90) == FRU_DATA[90:] + + @pytest.mark.parametrize('cc', [constants.CC_CANT_RET_NUM_REQ_BYTES, constants.CC_REQ_DATA_FIELD_EXCEED, constants.CC_PARAM_OUT_OF_RANGE]) From 8e1d396bfc79a779b996da9373013b0f5749fcd7 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:20:30 +0200 Subject: [PATCH 04/16] fru: raise DataNotFound if the requested area is not present The area offsets of the common header are None for areas that are not present. Requesting such an area read the FRU data with the offset None, which is the whole FRU inventory, and decoded it as the area. Raise DataNotFound instead. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/fru.py | 8 ++++++-- tests/test_fru.py | 14 +++++++++++++- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/pyipmi/fru.py b/pyipmi/fru.py index 5056b3d8..d53abeda 100644 --- a/pyipmi/fru.py +++ b/pyipmi/fru.py @@ -21,7 +21,7 @@ import datetime import os -from .errors import DecodingError, CompletionCodeError, RetryError +from .errors import DecodingError, CompletionCodeError, RetryError, DataNotFound from .helper import ReadLength from .msgs import constants from .utils import bcd_search, chunks, py3_array_tobytes @@ -124,12 +124,14 @@ def _check_fru_size(header: InventoryCommonHeader | None, name: str, f'{length} bytes exceeds the FRU size of ' f'{fru_size} bytes') - def _read_fru_area(self, offset: int, fru_id: int = 0, + def _read_fru_area(self, offset: int | None, fru_id: int = 0, header: InventoryCommonHeader | None = None, name: str = 'area') -> bytes: # The first read returns the area length and, if possible, all of the # area data: read up to the start of the next area. For the last area # only its minimal size of 8 bytes is known to be within the FRU data. + if offset is None: + raise DataNotFound(f'FRU has no {name}') count = FRU_AREA_MIN_LENGTH if header is not None: next_offsets = [o for o in header.area_offsets() if o > offset] @@ -194,6 +196,8 @@ def get_fru_multirecord_area(self, fru_id: int = 0, header = self._get_header(fru_id, ignore_checksum, header) # we have to determine the length of the area first + if header.multirecord_area_offset is None: + raise DataNotFound('FRU has no multirecord area') offset = header.multirecord_area_offset count = 0 diff --git a/tests/test_fru.py b/tests/test_fru.py index 6a88e553..c41d673f 100644 --- a/tests/test_fru.py +++ b/tests/test_fru.py @@ -5,7 +5,7 @@ import pytest -from pyipmi.errors import CompletionCodeError, DecodingError +from pyipmi.errors import CompletionCodeError, DataNotFound, DecodingError from pyipmi.msgs import constants from pyipmi.fru import (Fru, FruData, FruInventory, @@ -357,3 +357,15 @@ def test_get_fru_inventory_without_area_info(): header = fru.get_fru_inventory_header() assert header.fru_size is None fru.get_fru_inventory() + + +@pytest.mark.parametrize('method', [ + 'get_fru_chassis_area', 'get_fru_board_area', 'get_fru_product_area', + 'get_fru_multirecord_area', +]) +def test_get_fru_area_not_present(method): + # common header without any area + fru = FakeFruDevice(b'\x01\x00\x00\x00\x00\x00\x00\xff') + with pytest.raises(DataNotFound): + getattr(fru, method)() + assert fru.requests == [(0, 8)] From 6d3f995705f0115bf1f566ed22c92939dc3cab9b Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:20:53 +0200 Subject: [PATCH 05/16] ipmitool: add the missing arguments of 'picmg channel power' The command called send_channel_power() with the channel only, the required enable and current limit arguments were missing, so the command always failed with a TypeError. Add the arguments 'on'/'off' and the current limit in amperes. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- man/ipmitool.py.1 | 10 +++++++++- pyipmi/ipmitool.py | 7 ++++++- tests/test_ipmitool.py | 2 +- 3 files changed, 16 insertions(+), 3 deletions(-) diff --git a/man/ipmitool.py.1 b/man/ipmitool.py.1 index 5838c0b4..809dbaa7 100644 --- a/man/ipmitool.py.1 +++ b/man/ipmitool.py.1 @@ -401,13 +401,21 @@ Request the power channel status starting power channel .SH COMMAND \fI\,'ipmitool.py picmg channel power'\/\fR -usage: ipmitool.py picmg channel power [\-h] channel +usage: ipmitool.py picmg channel power [\-h] channel {on,off} current_limit Send channel power .TP \fBchannel\fR +.TP +\fB{on,off}\fR +enable or disable the payload power + +.TP +\fBcurrent_limit\fR +current limit in amperes + .SH COMMAND \fI\,'ipmitool.py picmg send'\/\fR usage: ipmitool.py picmg send [\-h] ... diff --git a/pyipmi/ipmitool.py b/pyipmi/ipmitool.py index 631dc0f9..2934f189 100755 --- a/pyipmi/ipmitool.py +++ b/pyipmi/ipmitool.py @@ -451,7 +451,8 @@ def cmd_picmg_send_pm_heartbeat(ipmi: pyipmi.Ipmi, def cmd_picmg_send_channel_power(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: - ipmi.send_channel_power(args.channel) + ipmi.send_channel_power(args.channel, args.state == 'on', + args.current_limit) VITA_LED_COLORS = ('reserved', 'BLUE', 'RED', 'GREEN', 'AMBER', 'ORANGE', @@ -1098,6 +1099,10 @@ def build_parser() -> argparse.ArgumentParser: p = sub.command('power', cmd_picmg_send_channel_power, 'Send channel power') p.add_argument('channel', type=auto_int) + p.add_argument('state', choices=('on', 'off'), + help='enable or disable the payload power') + p.add_argument('current_limit', type=float, + help='current limit in amperes') sub = group.group('send', 'Send') sub.command('heartbeat', cmd_picmg_send_pm_heartbeat, 'Send PM heartbeat') diff --git a/tests/test_ipmitool.py b/tests/test_ipmitool.py index 9581ae0b..48fdcebf 100644 --- a/tests/test_ipmitool.py +++ b/tests/test_ipmitool.py @@ -167,7 +167,7 @@ def test_defaults(self): ('picmg portstate get 1 0', 'cmd_picmg_get_portstate'), ('picmg portstate getall', 'cmd_picmg_get_portstate_all'), ('picmg channel status 1', 'cmd_picmg_getpower_channel_status'), - ('picmg channel power 1', 'cmd_picmg_send_channel_power'), + ('picmg channel power 1 on 2.5', 'cmd_picmg_send_channel_power'), ('picmg send heartbeat', 'cmd_picmg_send_pm_heartbeat'), ('vita properties', 'cmd_vita_properties'), ('vita led set 0 1 255 0 3', 'cmd_vita_led_set'), From e9b31f175761a2eb239823745c9f7054acc0666e Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:20:53 +0200 Subject: [PATCH 06/16] ipmitool: fix sensors without states and unsupported ports 'sdr show' failed with a TypeError for a sensor reading without assertion states, print 'na' instead. Select the output by the class of the SDR instead of comparing the record type by identity. get_port_state() returns None for a port that is not supported, which let 'picmg portstate get' and 'getall' fail with an AttributeError. Print 'Port not supported' for a single port and skip the port when all ports are listed. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/ipmitool.py | 27 +++++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/pyipmi/ipmitool.py b/pyipmi/ipmitool.py index 2934f189..622957ec 100755 --- a/pyipmi/ipmitool.py +++ b/pyipmi/ipmitool.py @@ -120,6 +120,12 @@ def format_analog_value(value: float | None) -> str: return '%.3f' % value +def format_states(states: int | None) -> str: + if states is None: + return 'na' + return '0x%x' % states + + def sdr_show(ipmi: pyipmi.Ipmi, s: pyipmi.sdr.SdrCommon) -> None: print("SDR record ID: 0x%04x" % s.id) @@ -129,7 +135,7 @@ def sdr_show(ipmi: pyipmi.Ipmi, s: pyipmi.sdr.SdrCommon) -> None: print("Device Id string: %s" % s.device_id_string) if hasattr(s, 'entity_id'): print("Entity: %s.%s" % (s.entity_id, s.entity_instance)) - if s.type is pyipmi.sdr.SDR_TYPE_FULL_SENSOR_RECORD: + if isinstance(s, pyipmi.sdr.SdrFullSensorRecord): (raw, states) = ipmi.get_sensor_reading(s.number, s.owner_lun) value = format_analog_value(s.convert_sensor_raw_to_value(raw)) t_unr = format_analog_value( @@ -145,19 +151,19 @@ def sdr_show(ipmi: pyipmi.Ipmi, s: pyipmi.sdr.SdrCommon) -> None: t_lnr = format_analog_value( s.convert_sensor_raw_to_value(s.threshold['lnr'])) print("Reading value: %s" % value) - print("Reading state: 0x%x" % states) + print("Reading state: %s" % format_states(states)) print("UNR: %s" % t_unr) print("UCR: %s" % t_ucr) print("UNC: %s" % t_unc) print("LNC: %s" % t_lnc) print("LCR: %s" % t_lcr) print("LNR: %s" % t_lnr) - elif s.type is pyipmi.sdr.SDR_TYPE_COMPACT_SENSOR_RECORD: + elif isinstance(s, pyipmi.sdr.SdrCompactSensorRecord): (raw, states) = ipmi.get_sensor_reading(s.number) print("Reading: %s" % raw) - print("Reading state: 0x%x" % states) - elif s.type is \ - pyipmi.sdr.SDR_TYPE_MANAGEMENT_CONTROLLER_CONFIRMATION_RECORD: + print("Reading state: %s" % format_states(states)) + elif isinstance(s, + pyipmi.sdr.SdrManagementControllerConfirmationRecord): print("Slave address: 0x%02x" % (s.device_slave_address << 1)) print("Device ID: 0x%02x" % s.device_id) print("Device revision: %d" % s.device_revision) @@ -406,7 +412,11 @@ def cmd_picmg_get_power(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: print(pwr) -def print_link_state(p: pyipmi.picmg.LinkDescriptor, s: int) -> None: +def print_link_state(p: pyipmi.picmg.LinkDescriptor | None, + s: int | None) -> None: + if p is None or s is None: + print('Port not supported') + return intf_str = pyipmi.picmg.LinkDescriptor().get_interface_string(p.interface) link_str = pyipmi.picmg.LinkDescriptor().get_link_type_string( p.type, p.extension, p.sig_class) @@ -421,7 +431,8 @@ def cmd_picmg_get_portstate_all(ipmi: pyipmi.Ipmi, for channel in range(16): try: (p, s) = ipmi.get_port_state(channel, interface) - print_link_state(p, s) + if p is not None: + print_link_state(p, s) except pyipmi.errors.CompletionCodeError as e: if e.cc == 0xcc: continue From 7fc14b9a0f3bcefaadfd6714f301c68e5008392a Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:21:09 +0200 Subject: [PATCH 07/16] interfaces/ipmitool: fix the check for a missing session The check hasattr(self, '_session') was always true, because _session is set to None by the constructor. Without a session the ipmitool command failed with an AttributeError on None instead of the intended RuntimeError. Check for None in a helper that returns the session. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/interfaces/ipmitool.py | 36 +++++++++++++++++++---------------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/pyipmi/interfaces/ipmitool.py b/pyipmi/interfaces/ipmitool.py index bb130f2d..00683b96 100644 --- a/pyipmi/interfaces/ipmitool.py +++ b/pyipmi/interfaces/ipmitool.py @@ -100,11 +100,16 @@ def __init__(self, interface_type: str = 'lan', cipher: int | None = None, r".*RAKP [0-9]+ HMAC.*") self.re_raw_request = re.compile( r".*RAW REQUEST\s*\((\d+)\s*bytes?\)") - self._session = None + self._session: Session | None = None def establish_session(self, session: Session) -> None: self._session = session + def _get_session(self) -> Session: + if self._session is None: + raise RuntimeError('Session needs to be set') + return self._session + def rmcp_ping(self) -> None: if self._interface_type == 'serial-terminal': @@ -112,15 +117,16 @@ def rmcp_ping(self) -> None: 'rcmp_ping not supported on "serial-terminal" interface') # for now this uses impitool.. + session = self._get_session() cmd = self.IPMITOOL_PATH cmd += (' -I %s' % self._interface_type) - cmd += (' -H %s' % self._session.rmcp_host) - cmd += (' -p %s' % self._session.rmcp_port) + cmd += (' -H %s' % session.rmcp_host) + cmd += (' -p %s' % session.rmcp_port) cmd += (' -v') cmd += self._build_ipmitool_retries() - if self._session.auth_type == Session.AUTH_TYPE_NONE: + if session.auth_type == Session.AUTH_TYPE_NONE: cmd += (' -A NONE') - elif self._session.auth_type == Session.AUTH_TYPE_PASSWORD: + elif session.auth_type == Session.AUTH_TYPE_PASSWORD: cmd += self._build_ipmitool_credentials() cmd += (' session info all') @@ -319,27 +325,26 @@ def _build_ipmitool_priv_level(self, level: int) -> str: def _build_ipmitool_cmd(self, target: Target, lun: int, netfn: int, raw_bytes: bytes) -> str: - if not hasattr(self, '_session'): - raise RuntimeError('Session needs to be set') + session = self._get_session() cmd = self.IPMITOOL_PATH cmd += (' -I %s' % self._interface_type) - cmd += (' -H %s' % self._session.rmcp_host) - cmd += (' -p %s' % self._session.rmcp_port) + cmd += (' -H %s' % session.rmcp_host) + cmd += (' -p %s' % session.rmcp_port) cmd += (' -v') - cmd += self._build_ipmitool_priv_level(self._session.priv_level) + cmd += self._build_ipmitool_priv_level(session.priv_level) if self._cipher: cmd += (' -C %s' % self._cipher) cmd += self._build_ipmitool_retries() - if self._session.auth_type == Session.AUTH_TYPE_NONE: + if session.auth_type == Session.AUTH_TYPE_NONE: cmd += ' -P ""' - elif self._session.auth_type == Session.AUTH_TYPE_PASSWORD: + elif session.auth_type == Session.AUTH_TYPE_PASSWORD: cmd += self._build_ipmitool_credentials() else: raise RuntimeError('Session type %d not supported' % - self._session.auth_type) + session.auth_type) cmd += self._build_ipmitool_target(target) cmd += self._build_ipmitool_raw_data(lun, netfn, raw_bytes) @@ -349,11 +354,10 @@ def _build_ipmitool_cmd(self, target: Target, lun: int, netfn: int, def _build_serial_ipmitool_cmd(self, target: Target, lun: int, netfn: int, raw_bytes: bytes) -> str: - if not hasattr(self, '_session'): - raise RuntimeError('Session needs to be set') + session = self._get_session() cmd = (f'{self.IPMITOOL_PATH} -I {self._interface_type} ' - f'-D {self._session.serial_port}:{self._session.serial_baudrate}') + f'-D {session.serial_port}:{session.serial_baudrate}') cmd += self._build_ipmitool_target(target) cmd += self._build_ipmitool_raw_data(lun, netfn, raw_bytes) From 10eb75dd37a83209532956ec736acb11926244a1 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:21:19 +0200 Subject: [PATCH 08/16] session: support user name and password as bytes everywhere The credentials can be set as str or bytes. Only the password of the RMCP interface handled bytes, a bytes user name failed in the RMCP interface, and bytes credentials failed in the RMCP+ and ipmitool interfaces. The RMCP interface padded a str user name to 16 characters instead of 16 bytes. Add auth_username_bytes and auth_password_bytes to the session, they return the UTF-8 encoded credentials, and use them in the interfaces. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/interfaces/ipmitool.py | 7 +++++-- pyipmi/interfaces/rmcp.py | 10 +++++----- pyipmi/interfaces/rmcpplus.py | 4 ++-- pyipmi/session.py | 25 ++++++++++++++++++++++--- 4 files changed, 34 insertions(+), 12 deletions(-) diff --git a/pyipmi/interfaces/ipmitool.py b/pyipmi/interfaces/ipmitool.py index 00683b96..c9a398c3 100644 --- a/pyipmi/interfaces/ipmitool.py +++ b/pyipmi/interfaces/ipmitool.py @@ -311,8 +311,11 @@ def _build_ipmitool_credentials(self) -> str: # The command is executed by a shell, so the credentials have to be # quoted to prevent the shell from interpreting characters like # '$', '`', '"' or '\'. - return (f' -U {shlex.quote(self._session.auth_username)}' - f' -P {shlex.quote(self._session.auth_password)}') + session = self._get_session() + username = session.auth_username_bytes.decode() + password = session.auth_password_bytes.decode() + return (f' -U {shlex.quote(username)}' + f' -P {shlex.quote(password)}') def _build_ipmitool_priv_level(self, level: int) -> str: LEVELS = { diff --git a/pyipmi/interfaces/rmcp.py b/pyipmi/interfaces/rmcp.py index 616eae87..a1d5df91 100644 --- a/pyipmi/interfaces/rmcp.py +++ b/pyipmi/interfaces/rmcp.py @@ -263,9 +263,9 @@ def _padd_password(self) -> bytes: The password/key is 0 padded to 16-bytes for all specified authentication types. """ - password = self.session._auth_password - if isinstance(password, str): - password = str.encode(password) + password = b'' + if self.session is not None: + password = self.session.auth_password_bytes return password.ljust(16, b'\x00') def _pack_auth_code_straight(self) -> bytes: @@ -499,8 +499,8 @@ def _get_session_challenge(self, session: Session) -> Message: req = create_request_by_name('GetSessionChallenge') req.target = self.host_target req.authentication.type = session.auth_type - if session._auth_username: - req.user_name = session._auth_username.ljust(16, '\x00') + if session.auth_username: + req.user_name = session.auth_username_bytes.ljust(16, b'\x00') rsp = self.send_and_receive(req) check_rsp_completion_code(rsp) return rsp diff --git a/pyipmi/interfaces/rmcpplus.py b/pyipmi/interfaces/rmcpplus.py index 536f4f62..1bdc79a5 100644 --- a/pyipmi/interfaces/rmcpplus.py +++ b/pyipmi/interfaces/rmcpplus.py @@ -444,8 +444,8 @@ def _open_session(self, session: Session, suite: CipherSuite) -> int: def _rakp(self, session: Session, suite: CipherSuite, managed_session_id: int) -> bytes: """Do the RAKP handshake, returns the session integrity key.""" - username = (session.auth_username or '').encode() - password = (session.auth_password or '').encode() + username = session.auth_username_bytes + password = session.auth_password_bytes if len(username) > MAX_USER_NAME_LENGTH: raise AuthenticationError('user name longer than %d bytes' % MAX_USER_NAME_LENGTH) diff --git a/pyipmi/session.py b/pyipmi/session.py index fa9ff3bf..e7c82f4e 100644 --- a/pyipmi/session.py +++ b/pyipmi/session.py @@ -20,6 +20,14 @@ from typing import Any +def _to_bytes(value: str | bytes | None) -> bytes: + if value is None: + return b'' + if isinstance(value, str): + return value.encode() + return value + + class Session: AUTH_TYPE_NONE = 0x00 AUTH_TYPE_MD2 = 0x01 @@ -105,19 +113,30 @@ def _set_auth_type(self, auth_type: int) -> None: def _get_auth_type(self) -> int: return self._auth_type - def set_auth_type_user(self, username: str, password: str) -> None: + def set_auth_type_user(self, username: str | bytes, + password: str | bytes) -> None: self._auth_type = self.AUTH_TYPE_PASSWORD self._auth_username = username self._auth_password = password @property - def auth_username(self) -> str | None: + def auth_username(self) -> str | bytes | None: return self._auth_username @property - def auth_password(self) -> str | None: + def auth_password(self) -> str | bytes | None: return self._auth_password + @property + def auth_username_bytes(self) -> bytes: + """The user name as UTF-8 encoded bytes, empty if not set.""" + return _to_bytes(self._auth_username) + + @property + def auth_password_bytes(self) -> bytes: + """The password as UTF-8 encoded bytes, empty if not set.""" + return _to_bytes(self._auth_password) + def establish(self) -> None: if hasattr(self.interface, 'establish_session'): self.interface.establish_session(self) From e479e18e2e26d8620cd4f834d7f211e62ae359bb Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:21:32 +0200 Subject: [PATCH 09/16] msgs: fix the error message of a duplicate message registration The group extension is None for most messages, so formatting it with '%d' failed with a TypeError instead of raising the DescriptionError. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/msgs/registry.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyipmi/msgs/registry.py b/pyipmi/msgs/registry.py index aaf04f7f..3dba12f6 100644 --- a/pyipmi/msgs/registry.py +++ b/pyipmi/msgs/registry.py @@ -51,7 +51,7 @@ def register_class(self, cls: type[Message]) -> type[Message]: cls.__name__) msg_id = (cls.__netfn__, cls.__cmdid__, cls.__group_extension__) if msg_id in self.registry: - raise DescriptionError('Message (%d,%d,%d) already registered (%s)' + raise DescriptionError('Message (%d,%d,%s) already registered (%s)' % (msg_id[0], msg_id[1], msg_id[2], self.registry[msg_id])) From 19681a58826a442248cd2b04e26052e2f7e7ba93 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:21:39 +0200 Subject: [PATCH 10/16] msgs: annotate the message framework for type checking The fields of a message are created as attributes at runtime, so a type checker reports every access to a field as an error. Declare __getattr__ and __setattr__ of Message and Bitfield.BitWrapper for type checking only, the runtime behaviour does not change. Declare the class attributes set by the message definitions (__netfn__, __cmdid__, __fields__) and the type of the message registry. Annotate the helpers in utils: ByteBuffer is iterable and can be sliced, ByteSequence is the raw data in any of the forms used by the library, is_string() is a type guard and the BCD codec functions match the codec protocol. ResponseDecoder declares _from_response(). Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/msgs/message.py | 45 ++++++++++++++++++++++++++++++++--------- pyipmi/msgs/registry.py | 3 ++- pyipmi/state.py | 3 +++ pyipmi/utils.py | 25 +++++++++++++++++------ 4 files changed, 60 insertions(+), 16 deletions(-) diff --git a/pyipmi/msgs/message.py b/pyipmi/msgs/message.py index d7135f47..eb0005e1 100644 --- a/pyipmi/msgs/message.py +++ b/pyipmi/msgs/message.py @@ -17,7 +17,7 @@ from __future__ import annotations from array import array -from typing import Any +from typing import Any, TYPE_CHECKING from collections.abc import Callable from . import constants @@ -46,7 +46,7 @@ def create(self) -> Any: class ByteArray(BaseField): - def __init__(self, name: str, length: int, + def __init__(self, name: str, length: int | None, default: bytes | None = None) -> None: BaseField.__init__(self, name, length) if default is not None: @@ -55,6 +55,7 @@ def __init__(self, name: str, length: int, self.default = None def _length(self, obj: Message) -> int: + assert self.length is not None return self.length def encode(self, obj: Message, data: ByteBuffer) -> None: @@ -72,7 +73,8 @@ def decode(self, obj: Message, data: ByteBuffer) -> None: bytes.append(data.pop_unsigned_int(1)) setattr(obj, self.name, array('B', bytes)) - def create(self) -> array: + def create(self) -> array | None: + assert self.length is not None if self.default is not None: return array('B', self.default) else: @@ -97,6 +99,8 @@ def create(self) -> None: class UnsignedInt(BaseField): + length: int + def encode(self, obj: Message, data: ByteBuffer) -> None: value = getattr(obj, self.name) data.push_unsigned_int(value, self.length) @@ -113,6 +117,8 @@ def create(self) -> int: class String(BaseField): + length: int + def encode(self, obj: Message, data: ByteBuffer) -> None: value = getattr(obj, self.name) data.push_string(value) @@ -201,7 +207,7 @@ def encode(self, obj: Message, data: ByteBuffer) -> None: data.extend(a) def decode(self, obj: Message, data: ByteBuffer) -> None: - setattr(obj, self.name, array('B', data[:])) + setattr(obj, self.name, array('B', data.array)) del data.array[:] def create(self) -> array: @@ -215,6 +221,8 @@ def __init__(self, name: str, width: int = 1, self.name = name self._width = width self.default = default + # set by the Bitfield the bit is added to + self.offset = 0 class ReservedBit(Bit): counter = 0 @@ -270,7 +278,13 @@ def _set_value(self, value: int) -> None: _value = property(_get_value, _set_value) + if TYPE_CHECKING: + # the bits are created as attributes at runtime + def __getattr__(self, name: str) -> Any: ... + def __setattr__(self, name: str, value: Any) -> None: ... + reserved_bit_counter = 0 + length: int def __init__(self, name: str, length: int, *bits: Bit) -> None: BaseField.__init__(self, name, length) @@ -317,13 +331,26 @@ def __init__(self, value: int | None = None) -> None: UnsignedInt.__init__(self, 'event_message_rev', 1, value) +Field = BaseField | Conditional | Optional + + class Message: RESERVED_FIELD_NAMES = ['cmdid', 'netfn', 'lun', 'group_extension'] + # set by the message definitions, intentionally without a default value + __netfn__: int + __cmdid__: int + __fields__: tuple[Field, ...] + __default_lun__ = 0 - __group_extension__ = None + __group_extension__: int | None = None __not_implemented__ = False + if TYPE_CHECKING: + # the message fields are created as attributes at runtime + def __getattr__(self, name: str) -> Any: ... + def __setattr__(self, name: str, value: Any) -> None: ... + def __init__(self, *args: Any, **kwargs: Any) -> None: """Message constructor with ([buf], [field=val,...]) prototype. @@ -341,7 +368,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: # set default lun self.lun = self.__default_lun__ - self.data = '' + self.data: Any = '' if args: self._decode(args[0]) else: @@ -390,17 +417,17 @@ def _decode(self, data: bytes) -> None: if not hasattr(self, '__fields__'): return - data = ByteBuffer(data) + buf = ByteBuffer(data) cc = None for field in self.__fields__: try: - field.decode(self, data) + field.decode(self, buf) except CompletionCodeError as e: # stop decoding on completion code != 0 cc = e.cc break - if (cc is None or cc == 0) and len(data) > 0: + if (cc is None or cc == 0) and len(buf) > 0: raise DecodingError('Data has extra bytes') def _is_request(self) -> bool: diff --git a/pyipmi/msgs/registry.py b/pyipmi/msgs/registry.py index 3dba12f6..bedd5b23 100644 --- a/pyipmi/msgs/registry.py +++ b/pyipmi/msgs/registry.py @@ -25,7 +25,8 @@ class MessageRegistry: def __init__(self) -> None: - self.registry = dict() + self.registry: dict[str | tuple[int, int, int | None], + type[Message]] = dict() def register_class(self, cls: type[Message]) -> type[Message]: # some sanity checks diff --git a/pyipmi/state.py b/pyipmi/state.py index 6d9e5164..b055bfb7 100644 --- a/pyipmi/state.py +++ b/pyipmi/state.py @@ -15,6 +15,9 @@ def __init__(self, rsp: Message | None = None) -> None: if rsp: self._from_response(rsp) + def _from_response(self, rsp: Message) -> None: + raise NotImplementedError() + class State(DefaultProperties, ResponseDecoder): """This is a container that represents a state. diff --git a/pyipmi/utils.py b/pyipmi/utils.py index eb5219d0..4a380fb4 100644 --- a/pyipmi/utils.py +++ b/pyipmi/utils.py @@ -18,8 +18,8 @@ import codecs from array import array -from typing import Any, TYPE_CHECKING -from collections.abc import Generator +from typing import Any, TYPE_CHECKING, TypeGuard, overload +from collections.abc import Generator, Iterator, Sequence from .msgs import constants from .errors import DecodingError, CompletionCodeError @@ -138,18 +138,31 @@ def __delslice__(self, a: int, b: int) -> None: def __len__(self) -> int: return len(self.array) - def __getitem__(self, idx: int) -> int: + def __iter__(self) -> Iterator[int]: + return iter(self.array) + + @overload + def __getitem__(self, idx: int) -> int: ... + + @overload + def __getitem__(self, idx: slice) -> array: ... + + def __getitem__(self, idx: int | slice) -> int | array: return self.array[idx] +# raw data given as bytes, list of ints, array or ByteBuffer +ByteSequence = Sequence[int] | ByteBuffer + + BCD_MAP = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', ' ', '-', '.'] -def bcd_encode(input: str, errors: str = 'strict') -> None: +def bcd_encode(input: str, errors: str = 'strict') -> tuple[bytes, int]: raise NotImplementedError() -def bcd_decode(encoded_input: Any) -> tuple[str, int]: +def bcd_decode(encoded_input: Any, errors: str = 'strict') -> tuple[str, int]: chars = list() try: for data in encoded_input: @@ -166,5 +179,5 @@ def bcd_search(name: str) -> codecs.CodecInfo | None: return codecs.CodecInfo(name='bcd+', encode=bcd_encode, decode=bcd_decode) -def is_string(string: Any) -> bool: +def is_string(string: Any) -> TypeGuard[str]: return isinstance(string, str) From c0fedfb0537540af7b32353497380ef34bb01e8a Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:21:50 +0200 Subject: [PATCH 11/16] pyipmi: declare the Ipmi API used by the command groups The Ipmi class is composed of the command groups (Bmc, Chassis, Fru, ...). They call send_message() and other methods provided by Ipmi or by another command group, which a type checker cannot know. Add the base class IpmiMixin that declares these methods for type checking only, the command groups inherit from it. Use the property decorator for interface, session and target of Ipmi, so they override the declarations of IpmiMixin, and annotate the attributes of Target. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/__init__.py | 40 +++++++++++++++++---------------- pyipmi/bmc.py | 3 ++- pyipmi/chassis.py | 3 ++- pyipmi/dcmi.py | 3 ++- pyipmi/event.py | 3 ++- pyipmi/fru.py | 3 ++- pyipmi/hpm.py | 3 ++- pyipmi/lan.py | 3 ++- pyipmi/messaging.py | 3 ++- pyipmi/mixin.py | 55 +++++++++++++++++++++++++++++++++++++++++++++ pyipmi/picmg.py | 3 ++- pyipmi/sdr.py | 3 ++- pyipmi/sel.py | 3 ++- pyipmi/sensor.py | 3 ++- pyipmi/vita.py | 3 ++- 15 files changed, 102 insertions(+), 32 deletions(-) create mode 100644 pyipmi/mixin.py diff --git a/pyipmi/__init__.py b/pyipmi/__init__.py index 99794be9..89a791bd 100644 --- a/pyipmi/__init__.py +++ b/pyipmi/__init__.py @@ -18,7 +18,7 @@ import time import ast -from typing import Any +from typing import Any, Literal from . import bmc from . import chassis @@ -94,8 +94,8 @@ def __str__(self) -> str: class Target: """The Target class represents an IPMI target.""" - routing = None - ipmb_address = None + routing: list[Routing] | None = None + ipmb_address: int | None = None def __init__(self, ipmb_address: int | None = None, routing: str | list[tuple] | None = None) -> None: @@ -183,14 +183,14 @@ def __init__(self, interface: Any = None, target: Target | None = None, self.requester = requester if requester is not None else NullRequester() for base in Ipmi.__bases__: - base.__init__(self) + base.__init__(self) # type: ignore[misc] def __enter__(self) -> Ipmi: self.open() return self def __exit__(self, exception_type: Any, exception_value: Any, - traceback: Any) -> bool: + traceback: Any) -> Literal[False]: self.close() return False @@ -259,33 +259,35 @@ def raw_command(self, lun: int, netfn: int, raw_bytes: bytes) -> bytes: return self.interface.send_and_receive_raw(self.target, lun, netfn, raw_bytes) - def _get_interface(self) -> Any: + @property + def interface(self) -> Any: try: return self._interface except AttributeError: raise RuntimeError('No interface has been set') from None - def _get_session(self) -> Session: + @interface.setter + def interface(self, interface: Any) -> None: + self._interface = interface + + @property + def session(self) -> Session: try: return self._session except AttributeError: raise RuntimeError('No IPMI session has been set') from None - def _get_target(self) -> Target: + @session.setter + def session(self, session: Session) -> None: + self._session = session + + @property + def target(self) -> Target | None: try: return self._target except AttributeError: raise RuntimeError('No IPMI target has been set') from None - def _set_interface(self, interface: Any) -> None: - self._interface = interface - - def _set_session(self, session: Session) -> None: - self._session = session - - def _set_target(self, target: Target) -> None: + @target.setter + def target(self, target: Target | None) -> None: self._target = target - - target = property(_get_target, _set_target) - interface = property(_get_interface, _set_interface) - session = property(_get_session, _set_session) diff --git a/pyipmi/bmc.py b/pyipmi/bmc.py index f0cd2f1d..674dbc40 100644 --- a/pyipmi/bmc.py +++ b/pyipmi/bmc.py @@ -22,9 +22,10 @@ from .utils import check_completion_code from .state import State from .fields import VersionField +from .mixin import IpmiMixin -class Bmc: +class Bmc(IpmiMixin): def get_device_id(self) -> DeviceId: return DeviceId(self.send_message_with_name('GetDeviceId')) diff --git a/pyipmi/chassis.py b/pyipmi/chassis.py index 08e25fb4..432cfdb2 100644 --- a/pyipmi/chassis.py +++ b/pyipmi/chassis.py @@ -23,6 +23,7 @@ from .msgs import create_request_by_name, Message from .utils import check_completion_code, check_rsp_completion_code, ByteBuffer from .state import State +from .mixin import IpmiMixin from .msgs.chassis import \ CONTROL_POWER_DOWN, CONTROL_POWER_UP, CONTROL_POWER_CYCLE, \ @@ -146,7 +147,7 @@ def boot_options_to_data(boot_device: BootDevice, boot_mode: str, return data -class Chassis: +class Chassis(IpmiMixin): def get_chassis_status(self) -> ChassisStatus: return ChassisStatus(self.send_message_with_name('GetChassisStatus')) diff --git a/pyipmi/dcmi.py b/pyipmi/dcmi.py index 2047abce..1adf9af4 100644 --- a/pyipmi/dcmi.py +++ b/pyipmi/dcmi.py @@ -23,6 +23,7 @@ ENTITY_ID_DCMI_BASEBOARD) from .msgs import Message, create_request_by_name from .utils import check_rsp_completion_code +from .mixin import IpmiMixin PARAM_SUPPORTED_DCMI_CAPABILITIES = 1 @@ -50,7 +51,7 @@ ENTITY_ID_DCMI_BASEBOARD) -class Dcmi: +class Dcmi(IpmiMixin): def get_dcmi_capabilities(self, selector: int) -> Message: rsp = self.send_message_with_name('GetDcmiCapabilities', parameter_selector=selector) diff --git a/pyipmi/event.py b/pyipmi/event.py index ec06eb36..0c4ace3c 100644 --- a/pyipmi/event.py +++ b/pyipmi/event.py @@ -18,12 +18,13 @@ from .utils import check_completion_code from .msgs import create_request_by_name +from .mixin import IpmiMixin EVENT_ASSERTION = 0 EVENT_DEASSERTION = 1 -class Event: +class Event(IpmiMixin): def set_event_receiver(self, ipmb_address: int, lun: int) -> None: req = create_request_by_name('SetEventReceiver') req.event_receiver.ipmb_i2c_slave_address = ipmb_address diff --git a/pyipmi/fru.py b/pyipmi/fru.py index d53abeda..ab2fb17c 100644 --- a/pyipmi/fru.py +++ b/pyipmi/fru.py @@ -26,6 +26,7 @@ from .msgs import constants from .utils import bcd_search, chunks, py3_array_tobytes from .fields import FruTypeLengthString +from .mixin import IpmiMixin codecs.register(bcd_search) @@ -33,7 +34,7 @@ FRU_AREA_MIN_LENGTH = 8 -class Fru: +class Fru(IpmiMixin): def __init__(self) -> None: self.write_length = 16 # read length per FRU device, a reduced length is kept for the diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index 8cd69485..6e76be56 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -31,6 +31,7 @@ from .utils import py3dec_unic_bytes_fix, py3_array_tobytes from .state import State from .fields import VersionField +from .mixin import IpmiMixin PROPERTY_GENERAL_PROPERTIES = 0 @@ -76,7 +77,7 @@ CC_ABORT_UPGRADE_CANNOT_RESUME_OPERATION = 0x81 -class Hpm: +class Hpm(IpmiMixin): @staticmethod def _get_component_count(components: int) -> int: diff --git a/pyipmi/lan.py b/pyipmi/lan.py index 7764940d..3943a3d4 100644 --- a/pyipmi/lan.py +++ b/pyipmi/lan.py @@ -21,6 +21,7 @@ from .errors import CompletionCodeError, DataNotFound from .msgs import constants, create_request_by_name from .utils import check_rsp_completion_code, ByteBuffer +from .mixin import IpmiMixin LAN_PARAMETER_SET_IN_PROGRESS = 0 LAN_PARAMETER_AUTHENTICATION_TYPE_SUPPORT = 1 @@ -173,7 +174,7 @@ def vlan_to_data(vlan: int) -> ByteBuffer: LAN_CHANNEL_SEARCH_RANGE = range(1, 0x0c) -class Lan: +class Lan(IpmiMixin): def __init__(self) -> None: self._lan_channel: int | None = None diff --git a/pyipmi/messaging.py b/pyipmi/messaging.py index 018ea199..9f3db974 100644 --- a/pyipmi/messaging.py +++ b/pyipmi/messaging.py @@ -22,6 +22,7 @@ from .msgs import create_request_by_name, Message from .utils import check_completion_code, check_rsp_completion_code from .state import State +from .mixin import IpmiMixin class PasswordOperation(int, Enum): @@ -62,7 +63,7 @@ class UserPrivilegeLevel(str, Enum): } -class Messaging: +class Messaging(IpmiMixin): def get_channel_authentication_capabilities( self, channel: int, priv_lvl: int) -> ChannelAuthenticationCapabilities: diff --git a/pyipmi/mixin.py b/pyipmi/mixin.py new file mode 100644 index 00000000..6eaa9a3e --- /dev/null +++ b/pyipmi/mixin.py @@ -0,0 +1,55 @@ +# Copyright (c) 2026 Kontron Europe GmbH +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +from __future__ import annotations + +from typing import Any, TYPE_CHECKING + +if TYPE_CHECKING: + from . import Target + from .bmc import DeviceId + from .messaging import ChannelInfo + from .msgs import Message + + +class IpmiMixin: + """Base class of the command groups the `Ipmi` class is composed of. + + The command groups use the messaging API of the `Ipmi` class and the + commands of other command groups. These are declared here for static + type checking only. + """ + + if TYPE_CHECKING: + @property + def interface(self) -> Any: ... + + @property + def target(self) -> Target | None: ... + + def send_message(self, req: Message, retry: int = 3) -> Message: ... + + def send_message_with_name(self, name: str, *args: Any, + **kwargs: Any) -> Message: ... + + # bmc.Bmc + def get_device_id(self) -> DeviceId: ... + + # messaging.Messaging + def get_channel_info(self, channel: int) -> ChannelInfo: ... + + # sensor.Sensor + def reserve_device_sdr_repository(self) -> int: ... diff --git a/pyipmi/picmg.py b/pyipmi/picmg.py index 18c0a351..607610c2 100644 --- a/pyipmi/picmg.py +++ b/pyipmi/picmg.py @@ -21,6 +21,7 @@ from .msgs import picmg from .utils import check_completion_code from .state import State +from .mixin import IpmiMixin from .msgs.picmg import \ FRU_CONTROL_COLD_RESET, FRU_CONTROL_WARM_RESET, \ @@ -28,7 +29,7 @@ FRU_ACTIVATION_FRU_ACTIVATE, FRU_ACTIVATION_FRU_DEACTIVATE -class Picmg: +class Picmg(IpmiMixin): def get_picmg_properties(self) -> Message: return self.send_message_with_name('GetPicmgProperties') diff --git a/pyipmi/sdr.py b/pyipmi/sdr.py index 04fe5cf1..0780df32 100644 --- a/pyipmi/sdr.py +++ b/pyipmi/sdr.py @@ -30,6 +30,7 @@ from .helper import get_sdr_data_helper, clear_repository_helper from .helper import get_sdr_chunk_helper, ReadLength from .state import State +from .mixin import IpmiMixin SDR_TYPE_FULL_SENSOR_RECORD = 0x01 SDR_TYPE_COMPACT_SENSOR_RECORD = 0x02 @@ -58,7 +59,7 @@ L_CUBERT = 11 -class Sdr: +class Sdr(IpmiMixin): def __init__(self) -> None: # read length of the SDR repository, a reduced length is kept for the # following records diff --git a/pyipmi/sel.py b/pyipmi/sel.py index 6e3897da..9e8eabe8 100644 --- a/pyipmi/sel.py +++ b/pyipmi/sel.py @@ -28,9 +28,10 @@ from .helper import clear_repository_helper from .state import State +from .mixin import IpmiMixin -class Sel: +class Sel(IpmiMixin): def get_sel_entries_count(self) -> int: info = SelInfo(self.send_message_with_name('GetSelInfo')) return info.entries diff --git a/pyipmi/sensor.py b/pyipmi/sensor.py index 30ee7039..d92543a2 100644 --- a/pyipmi/sensor.py +++ b/pyipmi/sensor.py @@ -26,6 +26,7 @@ ReadLength) from . import sdr +from .mixin import IpmiMixin # THRESHOLD BASED STATES @@ -109,7 +110,7 @@ SENSOR_TYPE_VITA_IPMC_RESET_TYPE = 0xf8 -class Sensor: +class Sensor(IpmiMixin): def __init__(self) -> None: # read length of the device SDRs, a reduced length is kept for the # following records diff --git a/pyipmi/vita.py b/pyipmi/vita.py index 0babf158..522ea97a 100644 --- a/pyipmi/vita.py +++ b/pyipmi/vita.py @@ -20,6 +20,7 @@ from .msgs import create_request_by_name, Message from .utils import check_completion_code +from .mixin import IpmiMixin from .msgs.vita import (VITA_FRU_CONTROL_COLD_RESET, # noqa: F401 VITA_FRU_CONTROL_WARM_RESET, VITA_FRU_CONTROL_GRACEFUL_REBOOT, @@ -70,7 +71,7 @@ } -class Vita: +class Vita(IpmiMixin): """VITA 46.11 commands of a VSO (VITA Standards Organization) IPMC.""" def get_vita_vso_capabilities(self) -> Message: From aae769b4ac068f8b70b325a7070c299d0d88b1ff Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:22:01 +0200 Subject: [PATCH 12/16] hpm: fix the type errors Annotate the attributes of the upgrade image records and narrow the optional values where they are used: - the FORMAT fields of UpgradeImageHeaderRecord are declared - UpgradeImage.actions is an empty list instead of None without a file - ComponentProperty declares _from_rsp_data() and from_data() returns None explicitly for an unknown property - find_component_id_by_descriptor() checks for the description string property by its class - upload_binary() uses a flag to resend a block with a reduced size instead of setting the chunk to None - the image header prints the signature as text instead of b'...' - the lists in the __str__() methods do not shadow the str builtin Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 116 ++++++++++++++++++++++++++++++-------------------- 1 file changed, 69 insertions(+), 47 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index 6e76be56..a30229f6 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -23,6 +23,7 @@ import time from array import array +from collections.abc import Sequence from .errors import CompletionCodeError, HpmError, IpmiTimeoutError from .msgs import create_request_by_name, Message @@ -115,7 +116,7 @@ def find_component_id_by_descriptor(self, descriptor: str) -> int | None: for component_id in caps.components: prop = self.get_component_property(component_id, PROPERTY_DESCRIPTION_STRING) - if prop is not None: + if isinstance(prop, ComponentPropertyDescriptionString): if prop.description == descriptor: return component_id return None @@ -175,8 +176,10 @@ def _determine_max_block_size(self) -> int: size = IPMB_MAX_REQUEST_DATA_SIZE \ - (bridges - 1) * SEND_MESSAGE_OVERHEAD else: - size = getattr(self.interface, 'MAX_REQUEST_DATA_SIZE', None) - if not isinstance(size, int): + max_size = getattr(self.interface, 'MAX_REQUEST_DATA_SIZE', None) + if isinstance(max_size, int): + size = max_size + else: size = IPMB_MAX_REQUEST_DATA_SIZE return size - UPLOAD_FIRMWARE_BLOCK_HEADER_SIZE @@ -194,6 +197,7 @@ def upload_binary(self, binary: bytes, timeout: float = 2, while offset < len(binary): chunk = binary[offset:offset + block_size] + block_size_reduced = False # a timed out block is sent again, up to `retry` times for attempt in range(retry): try: @@ -207,7 +211,7 @@ def upload_binary(self, binary: bytes, timeout: float = 2, constants.CC_REQ_DATA_FIELD_EXCEED) and not block_size_accepted and block_size > 1): block_size -= 1 - chunk = None + block_size_reduced = True else: raise HpmError('upload_firmware_block CC=0x%02x' % e.cc) from e @@ -217,7 +221,7 @@ def upload_binary(self, binary: bytes, timeout: float = 2, continue break - if chunk is None: + if block_size_reduced: # send the block again with the reduced size continue @@ -394,7 +398,8 @@ def upgrade_stage(self, image: UpgradeImage, component: int, are skipped. """ for action in image.actions: - if action.components & (1 << component) == 0: + if action.components is None \ + or action.components & (1 << component) == 0: continue if isinstance(action, UpgradeActionRecordUploadForUpgrade): if compare: @@ -410,6 +415,8 @@ def upgrade_stage(self, image: UpgradeImage, component: int, self.initiate_upgrade_action_and_wait(1 << component, upgrade_action) if isinstance(action, UpgradeActionRecordUploadForUpgrade): + assert action.firmware_image_data is not None + assert action.firmware_length is not None self.upload_binary(action.firmware_image_data) self.finish_upload_and_wait(component, action.firmware_length) @@ -516,12 +523,16 @@ def __str__(self) -> str: class ComponentProperty: - def __init__(self, data: bytes | None = None) -> None: + def __init__(self, data: Sequence[int] | None = None) -> None: if (data): self._from_rsp_data(data) + def _from_rsp_data(self, data: Sequence[int]) -> None: + raise NotImplementedError() + @staticmethod - def from_data(component_id: int, data: bytes) -> ComponentProperty | None: + def from_data(component_id: int, + data: str | Sequence[int]) -> ComponentProperty | None: if isinstance(data, str): data = [ord(c) for c in data] @@ -537,6 +548,7 @@ def from_data(component_id: int, data: bytes) -> ComponentProperty | None: return ComponentPropertyDeferredVersion(data) elif component_id in PROPERTY_OEM: raise NotImplementedError + return None class ComponentPropertyGeneral(ComponentProperty): @@ -547,7 +559,7 @@ class ComponentPropertyGeneral(ComponentProperty): DEFERRED_ACTIVATION_SUPPORT_MASK = 0x10 PAYLOAD_COLD_RESET_REQ_SUPPORT_MASK = 0x20 - def _from_rsp_data(self, data: bytes) -> None: + def _from_rsp_data(self, data: Sequence[int]) -> None: support = [] cap = data[0] @@ -577,7 +589,7 @@ def __str__(self) -> str: class ComponentPropertyCurrentVersion(ComponentProperty): - def _from_rsp_data(self, data: bytes) -> None: + def _from_rsp_data(self, data: Sequence[int]) -> None: self.version = VersionField(data) def __str__(self) -> str: @@ -586,9 +598,8 @@ def __str__(self) -> str: class ComponentPropertyDescriptionString(ComponentProperty): - def _from_rsp_data(self, data: bytes) -> None: - descr = py3_array_tobytes(array('B', data)) - descr = py3dec_unic_bytes_fix(descr) + def _from_rsp_data(self, data: Sequence[int]) -> None: + descr = py3dec_unic_bytes_fix(py3_array_tobytes(array('B', data))) # strip '\x00' descr = descr.replace('\0', '') self.description = descr @@ -599,7 +610,7 @@ def __str__(self) -> str: class ComponentPropertyRollbackVersion(ComponentProperty): - def _from_rsp_data(self, data: bytes) -> None: + def _from_rsp_data(self, data: Sequence[int]) -> None: self.version = VersionField(data) def __str__(self) -> str: @@ -608,7 +619,7 @@ def __str__(self) -> str: class ComponentPropertyDeferredVersion(ComponentProperty): - def _from_rsp_data(self, data: bytes) -> None: + def _from_rsp_data(self, data: Sequence[int]) -> None: self.version = VersionField(data) def __str__(self) -> str: @@ -617,7 +628,7 @@ def __str__(self) -> str: class ComponentPropertyOem(ComponentProperty): - def _from_rsp_data(self, data: bytes) -> None: + def _from_rsp_data(self, data: Sequence[int]) -> None: self.oem_data = data def __str__(self) -> str: @@ -673,6 +684,17 @@ class UpgradeImageHeaderRecord: image_header('oem_data_length', ' None: for a in self.FORMAT: setattr(self, a.field_name, None) @@ -695,7 +717,7 @@ def _from_data(self, data: bytes) -> None: data = [ord(c) for c in data] self.manufacturer_id = data[10] | data[11] << 8 | data[12] << 16 - self.components = [] + self.components: list[int] = [] for i in range(8): if data[20] & (1 << i): self.components.append(i) @@ -711,23 +733,23 @@ def _from_data(self, data: bytes) -> None: self.length = 34 + self.oem_data_length+1 def __str__(self) -> str: - str = [] - str.append("HPM Upgrade Image header") - str.append(" Signature: %s" % self.signature) - str.append(" Format Version: %s" % self.format_version) - str.append(" Device ID: %s" % self.device_id) - str.append(" Manufacturer: %s" % self.manufacturer_id) - str.append(" Product ID: %s" % self.product_id) - str.append(" Time: %s" % self.time) - str.append(" Image Cap: 0x%02x" % self.capabilities) - str.append(" Components: %s" % self.components) - str.append(" Selftest Timeout: %s" % self.selftest_timeout) - str.append(" Rollback Timeout: %s" % self.rollback_timeout) - str.append(" Inacc. Timeout: %s" % self.inaccessibility_timeout) - str.append(" Earliest comp.: %s" % self.earliest_compatible_revision) - str.append(" firmware Revision:%s" % self.firmware_revision) - str.append(" OEM data len: %s" % self.oem_data_length) - return "\n".join(str) + string = [] + string.append("HPM Upgrade Image header") + string.append(" Signature: %s" % self.signature.decode()) + string.append(" Format Version: %s" % self.format_version) + string.append(" Device ID: %s" % self.device_id) + string.append(" Manufacturer: %s" % self.manufacturer_id) + string.append(" Product ID: %s" % self.product_id) + string.append(" Time: %s" % self.time) + string.append(" Image Cap: 0x%02x" % self.capabilities) + string.append(" Components: %s" % self.components) + string.append(" Selftest Timeout: %s" % self.selftest_timeout) + string.append(" Rollback Timeout: %s" % self.rollback_timeout) + string.append(" Inacc. Timeout: %s" % self.inaccessibility_timeout) + string.append(" Earliest comp.: %s" % self.earliest_compatible_revision) + string.append(" firmware Revision:%s" % self.firmware_revision) + string.append(" OEM data len: %s" % self.oem_data_length) + return "\n".join(string) class UpgradeActionRecord: @@ -765,11 +787,12 @@ def create_from_data(data: bytes) -> UpgradeActionRecord: % action_type) def __str__(self) -> str: - str = [] - str.append("Action Record Type: 0x%x (%s) " % + assert self.action is not None and self.components is not None + string = [] + string.append("Action Record Type: 0x%x (%s) " % (self.action, self.ACTIONS[self.action])) - str.append(" Components: 0x%02x" % self.components) - return "\n".join(str) + string.append(" Components: 0x%02x" % self.components) + return "\n".join(string) class UpgradeActionRecordBackup(UpgradeActionRecord): @@ -812,12 +835,12 @@ def __init__(self, data: bytes | None = None) -> None: self.length += 31 + self.firmware_length def __str__(self) -> str: - str = [UpgradeActionRecord.__str__(self)] - str.append(" Firmware Version: %s" % self.firmware_version) - str.append(" Description: %s" + string = [UpgradeActionRecord.__str__(self)] + string.append(" Firmware Version: %s" % self.firmware_version) + string.append(" Description: %s" % self.firmware_description_string) - str.append(" Firmware Length: %s" % self.firmware_length) - return "\n".join(str) + string.append(" Firmware Length: %s" % self.firmware_length) + return "\n".join(string) class ImageChecksumRecord: @@ -834,14 +857,14 @@ def _from_data(self, data: bytes) -> None: class UpgradeImage: def __init__(self, filename: str | None = None) -> None: - self.actions = None + self.actions: list[UpgradeActionRecord] = [] if filename: self._from_file(filename) def __str__(self) -> str: - str = [] - return "\n".join(str) + string: list[str] = [] + return "\n".join(string) def _check_md5_sum(self, filedata: bytes) -> None: self.checksum_actual = hashlib.md5( @@ -867,7 +890,6 @@ def _from_file(self, filename: str) -> None: ################################ # Upgrade Actions - self.actions = [] while (off + HPM_IMAGE_CHECKSUM_SIZE) < len(file_data): action = UpgradeActionRecord.create_from_data(file_data[off:]) self.actions.append(action) From 0d823a101ec9cc2db1807828bf21bb22ebdfdbbc Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:22:01 +0200 Subject: [PATCH 13/16] pyipmi: fix the type errors of the command groups Annotate the attributes that are initialized with None or an empty list (ChassisStatus, Session, LinkDescriptor, FruInventory) and accept the raw data as any sequence of ints where bytes, lists and arrays are passed (FRU, SDR and type/length fields). Other changes: - get_port_state() is annotated to return None for an unsupported port - SelEntry is decoded from the record data by its own constructor instead of the response decoder of State - VersionField loses the 'version' attribute, it was always None - the command specific completion code is only looked up with a NetFn Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/chassis.py | 22 +++++++++++----------- pyipmi/errors.py | 2 +- pyipmi/fields.py | 27 ++++++++++++++------------- pyipmi/fru.py | 43 ++++++++++++++++++++++--------------------- pyipmi/picmg.py | 12 ++++++++++-- pyipmi/sdr.py | 42 +++++++++++++++++++++--------------------- pyipmi/sel.py | 10 ++++++++-- pyipmi/session.py | 20 ++++++++++---------- 8 files changed, 97 insertions(+), 81 deletions(-) diff --git a/pyipmi/chassis.py b/pyipmi/chassis.py index 432cfdb2..4d87ea18 100644 --- a/pyipmi/chassis.py +++ b/pyipmi/chassis.py @@ -228,17 +228,17 @@ def set_boot_options(self, boot_device: BootDevice, boot_mode: str, class ChassisStatus(State): - power_on = None - overload = None - interlock = None - fault = None - control_fault = None - restore_policy = None - id_cmd_state_info_support = None - chassis_id_state = None - front_panel_button_capabilities = None - last_event = [] - chassis_state = [] + power_on: bool | None = None + overload: bool | None = None + interlock: bool | None = None + fault: bool | None = None + control_fault: bool | None = None + restore_policy: int | None = None + id_cmd_state_info_support: bool | None = None + chassis_id_state: int | None = None + front_panel_button_capabilities: int | None = None + last_event: list[str] = [] + chassis_state: list[str] = [] def _from_response(self, rsp: Message) -> None: # don't append to the lists shared by all instances diff --git a/pyipmi/errors.py b/pyipmi/errors.py index 9af70659..bbda5e99 100644 --- a/pyipmi/errors.py +++ b/pyipmi/errors.py @@ -54,7 +54,7 @@ def find_cc_desc(error_cc: int, cmdid: int | None = None, if error_cc == cc[0]: return cc[1] # Then search in command specific completion codes - if cmdid is not None: + if cmdid is not None and netfn is not None: command_cc = CC_ERR_CMD_SPECIFIC_DESC.get((netfn, cmdid, group_extension), {}) descr = command_cc.get(error_cc, "Unknown error description") return descr diff --git a/pyipmi/fields.py b/pyipmi/fields.py index 6e26aaad..fd6efef6 100644 --- a/pyipmi/fields.py +++ b/pyipmi/fields.py @@ -4,6 +4,7 @@ from __future__ import annotations import array +from collections.abc import Sequence from .errors import DecodingError from .utils import py3_array_tobytes @@ -18,20 +19,20 @@ class VersionField: VERSION_FIELD_LEN = 2 VERSION_WITH_AUX_FIELD_LEN = 6 - def __init__(self, data: bytes | str | None = None) -> None: - self.major = None - self.minor = None + def __init__(self, data: str | Sequence[int] | None = None) -> None: + self.major: int | None = None + self.minor: int | None = None if data: self._from_data(data) - def _from_data(self, data: bytes | str) -> None: + def _from_data(self, data: str | Sequence[int]) -> None: if isinstance(data, str): data = [ord(c) for c in data] - data = array.array('B', data) - self.version = self._decode_data(data[0:2]) - if len(data) == self.VERSION_WITH_AUX_FIELD_LEN: - self.auxiliary = data[2:6] + buf = array.array('B', data) + self._decode_data(buf[0:2]) + if len(buf) == self.VERSION_WITH_AUX_FIELD_LEN: + self.auxiliary = buf[2:6] def __str__(self) -> str: return self.version_to_string() @@ -51,7 +52,7 @@ def version_to_string(self) -> str: return ''.join("%s.%s" % (self.major, self.minor)) -def _unpack6bitascii(data: bytes) -> str: +def _unpack6bitascii(data: Sequence[int]) -> str: """Unpack the 6bit ascii encoded string.""" string = '' for i in range(0, len(data), 3): @@ -78,7 +79,7 @@ class TypeLengthString: TYPE_6BIT_ASCII = 2 TYPE_ASCII_OR_UTF16 = 3 - def __init__(self, data: bytes | None = None, offset: int = 0, + def __init__(self, data: Sequence[int] | None = None, offset: int = 0, force_lang_eng: bool = False, sdr: bool = False) -> None: if data: self._from_data(data, offset, force_lang_eng) @@ -89,7 +90,7 @@ def __str__(self) -> str: else: return self.string.replace('\x00', '') - def _from_data(self, data: bytes, offset: int = 0, + def _from_data(self, data: Sequence[int], offset: int = 0, force_lang_eng: bool = False) -> None: self.offset = offset self.field_type = data[offset] >> 6 & 0x3 @@ -108,13 +109,13 @@ def _from_data(self, data: bytes, offset: int = 0, class FruTypeLengthString(TypeLengthString): - def __init__(self, data: bytes | None = None, offset: int = 0, + def __init__(self, data: Sequence[int] | None = None, offset: int = 0, force_lang_eng: bool = False) -> None: super().__init__(data, offset, force_lang_eng, sdr=False) class SdrTypeLengthString(TypeLengthString): - def __init__(self, data: bytes | None = None, offset: int = 0, + def __init__(self, data: Sequence[int] | None = None, offset: int = 0, force_lang_eng: bool = False) -> None: super().__init__(data, sdr=True) diff --git a/pyipmi/fru.py b/pyipmi/fru.py index ab2fb17c..5379f044 100644 --- a/pyipmi/fru.py +++ b/pyipmi/fru.py @@ -20,6 +20,7 @@ import codecs import datetime import os +from collections.abc import Sequence from .errors import DecodingError, CompletionCodeError, RetryError, DataNotFound from .helper import ReadLength @@ -281,7 +282,7 @@ def get_fru_inventory_from_file(filename: str, CUSTOM_FIELD_END = 0xc1 -def _decode_custom_fields(data: bytes) -> list[FruTypeLengthString]: +def _decode_custom_fields(data: Sequence[int]) -> list[FruTypeLengthString]: offset = 0 fields = [] while data[offset] != CUSTOM_FIELD_END: @@ -292,7 +293,7 @@ def _decode_custom_fields(data: bytes) -> list[FruTypeLengthString]: class FruData: - def __init__(self, data: bytes | str | None = None, + def __init__(self, data: str | Sequence[int] | None = None, ignore_checksum: bool = False) -> None: if data: if isinstance(data, str): @@ -306,7 +307,7 @@ class InventoryCommonHeader(FruData): # size of the FRU data, if read from a device that reports it fru_size: int | None = None - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: if len(data) < 8: raise DecodingError('InventoryCommonHeader length != 8') self.format_version = data[0] & 0x0f @@ -329,7 +330,7 @@ def area_offsets(self) -> list[int]: class CommonInfoArea(FruData): - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: self.format_version = data[0] & 0x0f if self.format_version != 1: raise DecodingError('unsupported format version (%d)' % @@ -364,7 +365,7 @@ class InventoryChassisInfoArea(CommonInfoArea): TYPE_RAID_CHASSIS = 22 TYPE_RACK_MOUNT_CHASSIS = 23 - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: CommonInfoArea._from_data(self, data) self.type = data[2] offset = 3 @@ -376,7 +377,7 @@ def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: class InventoryBoardInfoArea(CommonInfoArea): - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: CommonInfoArea._from_data(self, data, ignore_checksum=ignore_checksum) self.language_code = data[2] minutes = data[5] << 16 | data[4] << 8 | data[3] @@ -397,7 +398,7 @@ def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: class InventoryProductInfoArea(CommonInfoArea): - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: CommonInfoArea._from_data(self, data) self.language_code = data[2] offset = 3 @@ -433,7 +434,7 @@ def __str__(self) -> str: return '%02x: %s' % (self.record_type_id, ' '.join('%02x' % b for b in self.raw)) - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: if len(data) < 5: raise DecodingError('data too short') self.record_type_id = data[0] @@ -447,7 +448,7 @@ def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: raise DecodingError('FruDataMultiRecord record checksum failed') @staticmethod - def create_from_record_id(data: bytes) -> FruDataMultiRecord: + def create_from_record_id(data: Sequence[int]) -> FruDataMultiRecord: if data[0] == FruDataMultiRecord.TYPE_OEM_PICMG: return FruPicmgRecord.create_from_record_id(data) else: @@ -487,11 +488,11 @@ class FruPicmgRecord(FruDataMultiRecord): PICMG_RECORD_ID_CARRIER_BUSED_CONNECTIVITY = 0x31 PICMG_RECORD_ID_ZONE_3_INTERFACE_DOCUMENTATION = 0x32 - def __init__(self, data: bytes) -> None: + def __init__(self, data: Sequence[int]) -> None: FruDataMultiRecord.__init__(self, data) @staticmethod - def create_from_record_id(data: bytes) -> FruPicmgRecord: + def create_from_record_id(data: Sequence[int]) -> FruPicmgRecord: picmg_record = FruPicmgRecord(data) if picmg_record.picmg_record_type_id ==\ FruPicmgRecord.PICMG_RECORD_ID_MTCA_POWER_MODULE_CAPABILITY: @@ -499,7 +500,7 @@ def create_from_record_id(data: bytes) -> FruPicmgRecord: return FruPicmgRecord(data) - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: if len(data) < 10: raise DecodingError('data too short') data = array.array('B', data) @@ -511,7 +512,7 @@ def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: class FruPicmgPowerModuleCapabilityRecord(FruPicmgRecord): - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: if len(data) < 12: raise DecodingError('data too short') FruPicmgRecord._from_data(self, data) @@ -520,11 +521,11 @@ def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: class InventoryMultiRecordArea: - def __init__(self, data: bytes, ignore_checksum: bool = False) -> None: + def __init__(self, data: Sequence[int], ignore_checksum: bool = False) -> None: if data: self._from_data(data) - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: self.records = list() offset = 0 while True: @@ -536,17 +537,17 @@ def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: class FruInventory: - def __init__(self, data: bytes | None = None, + def __init__(self, data: Sequence[int] | None = None, ignore_checksum: bool = False) -> None: - self.chassis_info_area = None - self.board_info_area = None - self.product_info_area = None - self.multirecord_area = None + self.chassis_info_area: InventoryChassisInfoArea | None = None + self.board_info_area: InventoryBoardInfoArea | None = None + self.product_info_area: InventoryProductInfoArea | None = None + self.multirecord_area: InventoryMultiRecordArea | None = None if data: self._from_data(data, ignore_checksum=ignore_checksum) - def _from_data(self, data: bytes, ignore_checksum: bool = False) -> None: + def _from_data(self, data: Sequence[int], ignore_checksum: bool = False) -> None: self.raw = data self.common_header = InventoryCommonHeader(data[:8]) diff --git a/pyipmi/picmg.py b/pyipmi/picmg.py index 607610c2..8d589b16 100644 --- a/pyipmi/picmg.py +++ b/pyipmi/picmg.py @@ -149,8 +149,8 @@ def set_port_state(self, link_descr: LinkDescriptor, state: int) -> None: rsp = self.send_message(req) check_completion_code(rsp.completion_code) - def get_port_state(self, channel_number: int, - channel_interface: int) -> tuple[LinkDescriptor, int]: + def get_port_state(self, channel_number: int, channel_interface: int, + ) -> tuple[LinkDescriptor | None, int | None]: req = create_request_by_name('GetPortState') req.channel.number = channel_number req.channel.interface = channel_interface @@ -258,6 +258,14 @@ class LinkDescriptor(State): STATE_DISABLE = picmg.LINK_STATE_DISABLE STATE_ENABLE = picmg.LINK_STATE_ENABLE + channel: int + interface: int + link_flags: int + type: int + sig_class: int + extension: int + grouping_id: int + __properties__ = [ # (property, description) ('channel', ''), diff --git a/pyipmi/sdr.py b/pyipmi/sdr.py index 0780df32..8f442af9 100644 --- a/pyipmi/sdr.py +++ b/pyipmi/sdr.py @@ -24,7 +24,7 @@ from .errors import DecodingError from .fields import SdrTypeLengthString -from .utils import check_completion_code, ByteBuffer +from .utils import check_completion_code, ByteBuffer, ByteSequence from .msgs import create_request_by_name, Message from .helper import get_sdr_data_helper, clear_repository_helper @@ -192,7 +192,7 @@ def _from_response(self, rsp: Message) -> None: class SdrCommon: - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: if data: self.data = data @@ -214,7 +214,7 @@ def __str__(self) -> str: (' '.join(['%02x' % b for b in self.data])) return s - def _common_header(self, data: bytes) -> None: + def _common_header(self, data: ByteSequence) -> None: buffer = ByteBuffer(data[:]) try: self.id = buffer.pop_unsigned_int(2) @@ -242,7 +242,7 @@ def _device_id_string(self, buffer: ByteBuffer) -> None: # buffer.pop_string(self.device_id_string_length & 0x3f) @staticmethod - def from_data(data: bytes, next_id: int | None = None) -> SdrCommon: + def from_data(data: ByteSequence, next_id: int | None = None) -> SdrCommon: sdr_type = data[3] cls = { @@ -274,7 +274,7 @@ class SdrFullSensorRecord(SdrCommon): DATA_FMT_2S_COMPLEMENT = 2 DATA_FMT_NONE = 3 - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: super().__init__(data, next_id) @@ -296,9 +296,9 @@ def convert_sensor_raw_to_value(self, raw: int | None) -> float | None: elif (fmt == self.DATA_FMT_2S_COMPLEMENT): if raw & 0x80: raw = -((raw & 0x7f) ^ 0x7f) - 1 - raw = float(raw) - return self.lin((self.m * raw + (self.b * 10**self.k1)) * 10**self.k2) + return self.lin((self.m * float(raw) + + (self.b * 10**self.k1)) * 10**self.k2) def convert_sensor_value_to_raw(self, value: float) -> int: linearization = self.linearization & 0x7f @@ -401,7 +401,7 @@ def _decode_capabilities(self, capabilities: int) -> None: if (capabilities & 0x03) == 3: pass - def _from_data(self, data: bytes) -> None: + def _from_data(self, data: ByteSequence) -> None: buffer = ByteBuffer(data[5:]) # record key bytes self._common_record_key(buffer.pop_slice(3)) @@ -505,7 +505,7 @@ def _from_data(self, data: bytes) -> None: # SDR type 0x02 ################################################## class SdrCompactSensorRecord(SdrCommon): - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: super().__init__(data, next_id) @@ -515,7 +515,7 @@ def __str__(self) -> str: ' '.join(['%02x' % b for b in self.data])) return s - def _from_data(self, data: bytes) -> None: + def _from_data(self, data: ByteSequence) -> None: buffer = ByteBuffer(data[5:]) # record key bytes @@ -546,14 +546,14 @@ def _from_data(self, data: bytes) -> None: # SDR type 0x03 ################################################## class SdrEventOnlySensorRecord(SdrCommon): - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: super().__init__(data, next_id) def __str__(self) -> str: return 'Not supported yet.' - def _from_data(self, data: bytes) -> None: + def _from_data(self, data: ByteSequence) -> None: buffer = ByteBuffer(data[5:]) # record key bytes @@ -574,7 +574,7 @@ def _from_data(self, data: bytes) -> None: # SDR type 0x11 ################################################## class SdrFruDeviceLocator(SdrCommon): - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: super().__init__(data, next_id) @@ -584,7 +584,7 @@ def __str__(self) -> str: ' '.join(['%02x' % b for b in self.data])) return s - def _from_data(self, data: bytes) -> None: + def _from_data(self, data: ByteSequence) -> None: buffer = ByteBuffer(data[5:]) self.device_access_address = buffer.pop_unsigned_int(1) >> 1 self.fru_device_id = buffer.pop_unsigned_int(1) @@ -602,7 +602,7 @@ def _from_data(self, data: bytes) -> None: # SDR type 0x12 ################################################## class SdrManagementControllerDeviceLocator(SdrCommon): - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: super().__init__( data, next_id) @@ -613,7 +613,7 @@ def __str__(self) -> str: ' '.join(['%02x' % b for b in self.data])) return s - def _from_data(self, data: bytes) -> None: + def _from_data(self, data: ByteSequence) -> None: buffer = ByteBuffer(data[5:]) self.device_slave_address = buffer.pop_unsigned_int(1) >> 1 self.channel_number = buffer.pop_unsigned_int(1) & 0xf @@ -630,12 +630,12 @@ def _from_data(self, data: bytes) -> None: # SDR type 0x13 ################################################## class SdrManagementControllerConfirmationRecord(SdrCommon): - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: super().__init__( data, next_id) - def _from_data(self, data: bytes) -> None: + def _from_data(self, data: ByteSequence) -> None: buffer = ByteBuffer(data[5:]) self.device_slave_address = buffer.pop_unsigned_int(1) >> 1 self.device_id = buffer.pop_unsigned_int(1) @@ -654,14 +654,14 @@ def _from_data(self, data: bytes) -> None: # SDR type 0xC0 ################################################## class SdrOEMSensorRecord(SdrCommon): - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: super().__init__(data, next_id) def __str__(self) -> str: return 'Not supported yet.' - def _from_data(self, data: bytes) -> None: + def _from_data(self, data: ByteSequence) -> None: buffer = ByteBuffer(data[5:]) # record key bytes @@ -670,7 +670,7 @@ def _from_data(self, data: bytes) -> None: # Any SDR type not known or not implemented class SdrUnknownSensorRecord(SdrCommon): - def __init__(self, data: bytes | None = None, + def __init__(self, data: ByteSequence | None = None, next_id: int | None = None) -> None: super().__init__(data, next_id) diff --git a/pyipmi/sel.py b/pyipmi/sel.py index 9e8eabe8..6ed85df7 100644 --- a/pyipmi/sel.py +++ b/pyipmi/sel.py @@ -21,7 +21,7 @@ from collections.abc import Generator from .errors import CompletionCodeError, DecodingError -from .utils import check_completion_code, ByteBuffer +from .utils import check_completion_code, ByteBuffer, ByteSequence from .msgs import create_request_by_name, Message from .msgs import constants from .event import EVENT_ASSERTION, EVENT_DEASSERTION @@ -158,6 +158,12 @@ class SelEntry(State): TYPE_OEM_TIMESTAMPED_RANGE = list(range(0xc0, 0xe0)) TYPE_OEM_NON_TIMESTAMPED_RANGE = list(range(0xe0, 0x100)) + def __init__(self, data: ByteSequence | None = None) -> None: + # a SEL entry is decoded from the record data, not from a response + super().__init__() + if data: + self._from_data(data) + def __str__(self) -> str: raw = '[%s]' % (' '.join(['0x%02x' % b for b in self.data])) string = [] @@ -185,7 +191,7 @@ def type_to_string(entry_type: int) -> str | None: string = 'OEM non-timestamped (0x%02x)' % entry_type return string - def _from_response(self, data: ByteBuffer) -> None: + def _from_data(self, data: ByteSequence) -> None: if len(data) != 16: raise DecodingError('Invalid SEL record length (%d)' % len(data)) diff --git a/pyipmi/session.py b/pyipmi/session.py index e7c82f4e..0dd793ce 100644 --- a/pyipmi/session.py +++ b/pyipmi/session.py @@ -40,16 +40,16 @@ class Session: PRIV_LEVEL_ADMINISTRATOR = 4 PRIV_LEVEL_OEM = 5 - session_id = None - _interface = None + session_id: int | None = None + _interface: Any = None _priv_level = PRIV_LEVEL_ADMINISTRATOR _auth_type = AUTH_TYPE_NONE - _auth_username = None - _auth_password = None - _rmcp_host = None - _rmcp_port = None - _serial_port = None - _serial_baudrate = None + _auth_username: str | bytes | None = None + _auth_password: str | bytes | None = None + _rmcp_host: str | None = None + _rmcp_port: int | None = None + _serial_port: str | None = None + _serial_baudrate: int | None = None def __init__(self) -> None: self.established = False @@ -155,8 +155,8 @@ def __str__(self) -> str: string += ' Seq: 0x%08x\n' % self.sequence_number string += ' Host: %s:%s\n' % (self._rmcp_host, self._rmcp_port) string += ' Auth.: %s\n' % self.auth_type - string += ' User: %s\n' % self._auth_username - string += ' Password: %s\n' % self._auth_password + string += ' User: %s\n' % str(self._auth_username) + string += ' Password: %s\n' % str(self._auth_password) string += '\n' return string From 3a6686f6fcdf85a3cbc0b19f07a3cea631657f51 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:22:06 +0200 Subject: [PATCH 14/16] interfaces: fix the type errors Annotate the attributes initialized with None and narrow the optional values where they are used: - the IPMB header fields are declared as int without a None default, they are always set before a header is encoded - the IPMB, RMCP and ipmidev interfaces raise a ValueError if the target has no IPMB address or a bridge has no channel, the shared helper target_ipmb_address() checks the address - ipmidev raises a RuntimeError if the device is not open - a received RMCP message without data is returned as b'' instead of failing in the debug output, an MD5 authenticated message without data could not be packed - the ipmitool interface prints the output of a failed command as text Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/interfaces/__init__.py | 6 +-- pyipmi/interfaces/ipmb.py | 61 ++++++++++++++++++------------- pyipmi/interfaces/ipmidev.py | 17 ++++++--- pyipmi/interfaces/ipmitool.py | 4 +- pyipmi/interfaces/openipmblink.py | 5 ++- pyipmi/interfaces/rmcp.py | 33 +++++++++-------- pyipmi/interfaces/rmcpplus.py | 3 ++ pyipmi/interfaces/router.py | 2 + 8 files changed, 80 insertions(+), 51 deletions(-) diff --git a/pyipmi/interfaces/__init__.py b/pyipmi/interfaces/__init__.py index 37e3055d..b837886f 100644 --- a/pyipmi/interfaces/__init__.py +++ b/pyipmi/interfaces/__init__.py @@ -20,6 +20,7 @@ from .ipmitool import Ipmitool from .aardvark import Aardvark +from .base import Interface from .ipmbdev import IpmbDev from .ipmidev import IpmiDev from .mock import Mock @@ -27,7 +28,7 @@ from .rmcp import Rmcp from .rmcpplus import RmcpPlus -INTERFACES = [ +INTERFACES: list[type[Interface]] = [ Ipmitool, Aardvark, IpmbDev, @@ -42,7 +43,6 @@ def create_interface(interface: str, *args: Any, **kwargs: Any) -> Any: for intf in INTERFACES: if intf.NAME == interface: - intf = intf(*args, **kwargs) - return intf + return intf(*args, **kwargs) raise RuntimeError('unknown interface with name %s' % interface) diff --git a/pyipmi/interfaces/ipmb.py b/pyipmi/interfaces/ipmb.py index 19847786..18859f6e 100644 --- a/pyipmi/interfaces/ipmb.py +++ b/pyipmi/interfaces/ipmb.py @@ -21,7 +21,7 @@ import time from array import array from typing import TYPE_CHECKING -from collections.abc import Iterable +from collections.abc import Iterable, Sequence from .. import Routing, Target from ..errors import IpmiTimeoutError @@ -62,19 +62,22 @@ class IpmbHeader: *-------*--------------*----------*-------*---------------*-------* """ - rs_sa = None - rs_lun = None - rq_sa = None - rq_lun = None - rq_seq = None - netfn = None - cmdid = None - checksum = None + rs_sa: int + rs_lun: int + rq_sa: int + rq_lun: int + rq_seq: int + netfn: int + cmdid: int + checksum: int - def __init__(self, data: bytes | None = None) -> None: + def __init__(self, data: Sequence[int] | None = None) -> None: if data: self.decode(data) + def decode(self, data: Sequence[int]) -> None: + raise NotImplementedError() + def __str__(self) -> str: return f'rs_sa=0x{self.rs_sa:02x}, rs_lun={self.rs_lun}, ' \ f'rq_sa=0x{self.rq_sa:02x}, rq_lun={self.rq_lun}, ' \ @@ -97,10 +100,9 @@ def encode(self) -> bytes: data.append(self.cmdid) return py3_array_tobytes(data) - def decode(self, data: bytes) -> None: + def decode(self, data: Sequence[int]) -> None: """Decode the header.""" - msg = array('B') - py3_array_frombytes(msg, data) + msg = array('B', data) self.rs_sa = msg[0] self.netfn = msg[1] >> 2 self.rs_lun = msg[1] & 3 @@ -125,17 +127,17 @@ def encode(self) -> bytes: data.append(self.cmdid) return py3_array_tobytes(data) - def decode(self, data: bytes) -> None: + def decode(self, data: Sequence[int]) -> None: """Decode the header.""" - data = array('B', data) - self.rq_sa = data[0] - self.netfn = data[1] >> 2 - self.rq_lun = data[1] & 3 - self.checksum = data[2] - self.rs_sa = data[3] - self.rq_seq = data[4] >> 2 - self.rs_lun = data[4] & 3 - self.cmdid = data[5] + msg = array('B', data) + self.rq_sa = msg[0] + self.netfn = msg[1] >> 2 + self.rq_lun = msg[1] & 3 + self.checksum = msg[2] + self.rs_sa = msg[3] + self.rq_seq = msg[4] >> 2 + self.rs_lun = msg[4] & 3 + self.cmdid = msg[5] def from_req_header(self, req_header: IpmbHeaderReq) -> None: """Set up the header of the response to the given request. @@ -218,6 +220,9 @@ def encode_bridged_message(routing: list[Routing], header: IpmbHeaderReq, tx_data = encode_ipmb_msg(header, payload) for bridge in reversed(routing[:-1]): + if bridge.channel is None: + raise ValueError('bridge channel of routing entry missing: %s' + % bridge) tx_data = encode_send_message(tx_data, rq_sa=bridge.rq_sa, rs_sa=bridge.rs_sa, @@ -246,6 +251,12 @@ def decode_bridged_message(rx_data: bytes) -> bytes: return rx_data +def target_ipmb_address(target: Target) -> int: + if target.ipmb_address is None: + raise ValueError('IPMB address of the target missing: %s' % target) + return target.ipmb_address + + def rx_filter(header: IpmbHeaderReq, data: bytes | array, rq_sa: bool = False, rs_sa: bool = False, rq_lun: bool = False, rs_lun: bool = True, rq_seq: bool = True) -> bool: @@ -404,7 +415,7 @@ def is_ipmc_accessible(self, target: Target) -> bool: header = IpmbHeaderReq() header.netfn = 6 header.rs_lun = 0 - header.rs_sa = target.ipmb_address + header.rs_sa = target_ipmb_address(target) header.rq_seq = self.next_sequence_number header.rq_lun = 0 header.rq_sa = self.slave_address @@ -434,7 +445,7 @@ def _send_and_receive(self, target: Target, lun: int, netfn: int, header = IpmbHeaderReq() header.netfn = netfn header.rs_lun = lun - header.rs_sa = target.ipmb_address + header.rs_sa = target_ipmb_address(target) header.rq_seq = self._inc_sequence_number() header.rq_lun = 0 header.rq_sa = self.slave_address diff --git a/pyipmi/interfaces/ipmidev.py b/pyipmi/interfaces/ipmidev.py index 603179a4..98125027 100644 --- a/pyipmi/interfaces/ipmidev.py +++ b/pyipmi/interfaces/ipmidev.py @@ -34,7 +34,7 @@ try: import fcntl except ImportError: - fcntl = None + fcntl = None # type: ignore[assignment] # definitions of include/uapi/linux/ipmi.h @@ -139,6 +139,11 @@ def close(self) -> None: os.close(self._dev) self._dev = None + def _get_dev(self) -> int: + if self._dev is None: + raise RuntimeError('Device %s is not open' % self.port) + return self._dev + def is_ipmc_accessible(self, target: Target) -> bool: try: self.send_and_receive_raw(target, 0, constants.NETFN_APP, @@ -153,6 +158,7 @@ def _encode_address(self, target: Target, lun: int) -> ctypes.Structure: raise RuntimeError('ipmidev supports only one bridge, routing: %s' % ', '.join(str(r) for r in routing)) + address: int | None if len(routing) == 2: channel = routing[0].channel address = routing[1].rs_sa @@ -180,16 +186,17 @@ def _send(self, addr: ctypes.Structure, netfn: int, cmdid: int, req.msg.cmd = cmdid req.msg.data_len = len(data) req.msg.data = ctypes.addressof(data_buf) - fcntl.ioctl(self._dev, IPMICTL_SEND_COMMAND, req) + fcntl.ioctl(self._get_dev(), IPMICTL_SEND_COMMAND, req) def _receive(self, msgid: int, netfn: int, cmdid: int) -> bytes: + dev = self._get_dev() deadline = time.monotonic() + self.timeout while True: timeout = deadline - time.monotonic() if timeout <= 0: raise IpmiTimeoutError() - r, _, _ = select.select([self._dev], [], [], timeout) - if self._dev not in r: + r, _, _ = select.select([dev], [], [], timeout) + if dev not in r: raise IpmiTimeoutError() addr_buf = ctypes.create_string_buffer( @@ -200,7 +207,7 @@ def _receive(self, msgid: int, netfn: int, cmdid: int) -> bytes: recv.addr_len = ctypes.sizeof(addr_buf) recv.msg.data = ctypes.addressof(data_buf) recv.msg.data_len = IPMI_MAX_MSG_LENGTH - fcntl.ioctl(self._dev, IPMICTL_RECEIVE_MSG_TRUNC, recv) + fcntl.ioctl(dev, IPMICTL_RECEIVE_MSG_TRUNC, recv) rx_data = data_buf.raw[:recv.msg.data_len] if (recv.recv_type == IPMI_RESPONSE_RECV_TYPE diff --git a/pyipmi/interfaces/ipmitool.py b/pyipmi/interfaces/ipmitool.py index c9a398c3..55e4bc80 100644 --- a/pyipmi/interfaces/ipmitool.py +++ b/pyipmi/interfaces/ipmitool.py @@ -172,7 +172,8 @@ def _parse_output(self, output: bytes) -> tuple[int | None, array | None]: # Check for error opening ipmi device if self.re_could_not_open.match(line): - raise RuntimeError(f'ipmitool failed: {output}') + raise RuntimeError('ipmitool failed: ' + f'{py3dec_unic_bytes_fix(output)}') if self.re_long_password.match(line): raise IpmiLongPasswordError(line) @@ -264,6 +265,7 @@ def _build_ipmitool_raw_data(lun: int, netfn: int, raw: bytes) -> str: @staticmethod def _routing_channel(target: Target, index: int) -> int: """Return the channel of a routing entry, ipmitool needs it.""" + assert target.routing is not None channel = target.routing[index].channel if channel is None: raise ValueError('the ipmitool interface needs the bridge channel ' diff --git a/pyipmi/interfaces/openipmblink.py b/pyipmi/interfaces/openipmblink.py index 6d65f9b1..a4b3e02d 100644 --- a/pyipmi/interfaces/openipmblink.py +++ b/pyipmi/interfaces/openipmblink.py @@ -61,7 +61,7 @@ def __init__(self, port: str, cmd_timeout: float = 1.0) -> None: self.port = port self.cmd_timeout = cmd_timeout self.info: dict | None = None - self._ser = None + self._ser: Any = None self._rx_buf = bytearray() self._replies: queue.Queue = queue.Queue() self._command_lock = threading.Lock() @@ -232,7 +232,8 @@ def _read_loop(self) -> None: self._dispatch(packet) def _dispatch(self, packet: dict) -> None: - listener = self._listeners.get(packet.get('bus')) + bus = packet.get('bus') + listener = self._listeners.get(bus) if isinstance(bus, int) else None if listener is None: return try: diff --git a/pyipmi/interfaces/rmcp.py b/pyipmi/interfaces/rmcp.py index a1d5df91..9a35caf5 100644 --- a/pyipmi/interfaces/rmcp.py +++ b/pyipmi/interfaces/rmcp.py @@ -35,7 +35,7 @@ from .base import Interface from ..interfaces.ipmb import (IpmbHeaderReq, encode_ipmb_msg, encode_bridged_message, decode_bridged_message, - rx_filter) + rx_filter, target_ipmb_address) from ..utils import (check_completion_code, check_rsp_completion_code, py3_array_tobytes) @@ -213,6 +213,8 @@ def pack(self) -> bytes: def unpack(self, sdu: bytes) -> None: AsfMsg.unpack(self, sdu) + # check_header() made sure that the data is present + assert self.data is not None # header_len = struct.calcsize(self.ASF_HEADER_FORMAT) (self.oem_iana_enterprise_number, self.oem_defined, self.supported_entities, self.supported_interactions) =\ @@ -229,7 +231,8 @@ def check_data(self) -> None: def check_header(self) -> None: if self.asf_type != self.ASF_TYPE_PRESENCE_PONG: raise DecodingError('type does not match') - if len(self.data) != struct.calcsize(self.DATA_FORMAT): + if self.data is None \ + or len(self.data) != struct.calcsize(self.DATA_FORMAT): raise DecodingError('Data length mismatch') @@ -304,7 +307,7 @@ def pack(self, sdu: bytes | None) -> bytes: elif auth_type == Session.AUTH_TYPE_PASSWORD: pdu += self._pack_auth_code_straight() elif auth_type == Session.AUTH_TYPE_MD5: - pdu += self._pack_auth_code_md5(sdu) + pdu += self._pack_auth_code_md5(sdu or b'') else: raise NotSupportedError('authentication type %s' % auth_type) @@ -404,15 +407,15 @@ def __init__(self, slave_address: int = 0x81, quirks_cfg={'rmcp_ignore_rq_seq': True} ) """ - self.host = None - self.port = None + self.host: str | None = None + self.port: int | None = None self.seq_number = 0xff self.slave_address = slave_address self.host_target = Target(host_target_address) self.max_retries = max_retries self.next_sequence_number = 0 self.keep_alive_interval = keep_alive_interval - self._stop_keep_alive = None + self._stop_keep_alive: Callable[[], None] | None = None self._last_request_time = 0.0 self._timeout: float | None = None self.transaction_lock = threading.Lock() @@ -453,12 +456,12 @@ def _send_ipmi_msg(self, data: bytes) -> None: tx_data = ipmi.pack(data) self._send_rmcp_msg(tx_data, RMCP_CLASS_IPMI) - def _receive_ipmi_msg(self, ignore_sdu_length: bool = False) -> bytes | None: + def _receive_ipmi_msg(self, ignore_sdu_length: bool = False) -> bytes: (_, class_of_msg, pdu) = self._receive_rmcp_msg() if class_of_msg != RMCP_CLASS_IPMI: raise DecodingError('invalid class field in ASF message') msg = IpmiMsg(ignore_sdu_length=ignore_sdu_length) - data = msg.unpack(pdu) + data = msg.unpack(pdu) or b'' logger.debug('IPMI RX: {:s}'.format( ' '.join('%02x' % b for b in array('B', data)))) return data @@ -512,7 +515,7 @@ def _activate_session(self, session: Session, challenge: bytes) -> Message: req.authentication.type = session.auth_type req.privilege_level.maximum_requested = session.priv_level req.challenge_string = challenge - req.session_id = self._session.sid + req.session_id = session.sid req.initial_outbound_sequence_number = random.randrange(1, 0xffffffff) rsp = self.send_and_receive(req) check_rsp_completion_code(rsp) @@ -541,8 +544,8 @@ def _keep_alive(self) -> None: def establish_session(self, session: Session) -> None: self._session = None - self.host = session._rmcp_host - self.port = session._rmcp_port + self.host = session.rmcp_host + self.port = session.rmcp_port self._sock.connect((self.host, self.port)) # 0 - Ping @@ -565,9 +568,9 @@ def establish_session(self, session: Session) -> None: # 3 - Activate Session logger.debug('Activate Session') rsp = self._activate_session(session, session_challenge) - self._session.sid = rsp.session_id - self._session.sequence_number = rsp.initial_inbound_sequence_number - self._session.activated = True + session.sid = rsp.session_id + session.sequence_number = rsp.initial_inbound_sequence_number + session.activated = True logger.debug('Set Session Privilege Level') # 4 - Set Session Privilege Level @@ -646,7 +649,7 @@ def _send_and_receive(self, target: Target, lun: int, netfn: int, header = IpmbHeaderReq() header.netfn = netfn header.rs_lun = lun - header.rs_sa = target.ipmb_address + header.rs_sa = target_ipmb_address(target) header.rq_seq = self.next_sequence_number header.rq_lun = 0 header.rq_sa = self.slave_address diff --git a/pyipmi/interfaces/rmcpplus.py b/pyipmi/interfaces/rmcpplus.py index 1bdc79a5..72851eee 100644 --- a/pyipmi/interfaces/rmcpplus.py +++ b/pyipmi/interfaces/rmcpplus.py @@ -315,6 +315,7 @@ def __init__(self, slave_address: int = 0x81, keep_alive_interval=keep_alive_interval, max_retries=max_retries, quirks_cfg=quirks_cfg) + self.cipher_suites: tuple[int, ...] if cipher_suite is None: self.cipher_suites = DEFAULT_CIPHER_SUITES elif cipher_suite in CIPHER_SUITES: @@ -334,6 +335,8 @@ def _next_message_tag(self) -> int: def _send_payload(self, payload_type: int, payload: bytes) -> None: if self._keys is not None: + # the keys are set when the session is established + assert self._session is not None self._session.increment_sequence_number() pdu = pack_rmcpplus(payload_type, payload, self._session.sid, self._session.sequence_number, self._keys) diff --git a/pyipmi/interfaces/router.py b/pyipmi/interfaces/router.py index d601faff..0d9824ff 100644 --- a/pyipmi/interfaces/router.py +++ b/pyipmi/interfaces/router.py @@ -154,6 +154,7 @@ def request(self, interface: IpmbInterface, header: IpmbHeaderReq, interface.send_frame(encode_ipmb_msg(header, payload)) if not pending.event.wait(timeout): raise IpmiTimeoutError() + assert pending.frame is not None return pending.frame finally: with self._lock: @@ -206,6 +207,7 @@ def _handle_request(self, interface: IpmbInterface, frame: bytes) -> None: logger.debug('IPMB RX request [%s]', header) handler = self._find_handler(header, data) + rsp_data: bytes | None if handler is None: if self.unhandled_cc is None: return From 1a01d749a447b446a7343a9b216fbb062735f79a Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:22:12 +0200 Subject: [PATCH 15/16] ipmitool: fix the type errors Use separate variables for the FRU areas and the chassis status JSON instead of reusing one variable for different types, parse the interface options from the option string only, and close the connection only if one was created. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/ipmitool.py | 90 +++++++++++++++++++++------------------------- 1 file changed, 40 insertions(+), 50 deletions(-) diff --git a/pyipmi/ipmitool.py b/pyipmi/ipmitool.py index 622957ec..244c5fb4 100755 --- a/pyipmi/ipmitool.py +++ b/pyipmi/ipmitool.py @@ -27,6 +27,7 @@ import traceback from array import array from collections.abc import Callable +from typing import Any import pyipmi import pyipmi.interfaces @@ -204,18 +205,11 @@ def cmd_sdr_show_all(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: def print_sdr_list_entry(record_id: int, number: int | str | None, id_string: str | None, value: object, states: int | None) -> None: - if number: - number = str(number) - else: - number = 'na' - - if states: - states = hex(states) - else: - states = 'na' + number_str = str(number) if number else 'na' + states_str = hex(states) if states else 'na' - print("0x%04x | %3s | %-18s | %9s | %s" % (record_id, number, - id_string, value, states)) + print("0x%04x | %3s | %-18s | %9s | %s" % (record_id, number_str, + id_string, value, states_str)) def cmd_sdr_list(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: @@ -237,17 +231,17 @@ def cmd_sdr_list(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: for s in iter_fct(): try: number = None - value = None + value: int | str | None = None states = None - if s.type is pyipmi.sdr.SDR_TYPE_FULL_SENSOR_RECORD: - (value, states) = ipmi.get_sensor_reading(s.number) + if isinstance(s, pyipmi.sdr.SdrFullSensorRecord): + (raw, states) = ipmi.get_sensor_reading(s.number) number = s.number - if value is not None: + if raw is not None: value = format_analog_value( - s.convert_sensor_raw_to_value(value)) + s.convert_sensor_raw_to_value(raw)) - elif s.type is pyipmi.sdr.SDR_TYPE_COMPACT_SENSOR_RECORD: + elif isinstance(s, pyipmi.sdr.SdrCompactSensorRecord): (value, states) = ipmi.get_sensor_reading(s.number) number = s.number @@ -273,23 +267,23 @@ def cmd_fru_print(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: inv = ipmi.get_fru_inventory(args.fru_id) # Chassis Info Area - area = inv.chassis_info_area - if area: + chassis_area = inv.chassis_info_area + if chassis_area: print(''' Chassis Info Area: Type: %(type)d Part Number: %(part_number)s Serial Number: %(serial_number)s -'''[1:-1] % area.__dict__) +'''[1:-1] % chassis_area.__dict__) - if len(area.custom_chassis_info) != 0: + if len(chassis_area.custom_chassis_info) != 0: print(' Custom Chassis Info Records:') - for field in area.custom_chassis_info: + for field in chassis_area.custom_chassis_info: print(' %s' % field) # Board Info Area - area = inv.board_info_area - if area: + board_area = inv.board_info_area + if board_area: print(''' Board Info Area: Mfg. Date / Time: %(mfg_date)s @@ -298,16 +292,16 @@ def cmd_fru_print(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: Serial Number: %(serial_number)s Part Number: %(part_number)s FRU File ID: %(fru_file_id)s -'''[1:-1] % area.__dict__) +'''[1:-1] % board_area.__dict__) - if len(area.custom_mfg_info) != 0: + if len(board_area.custom_mfg_info) != 0: print(' Custom Board Info Records:') - for field in area.custom_mfg_info: + for field in board_area.custom_mfg_info: print(' %s' % field) # Product Info Area - area = inv.product_info_area - if area: + product_area = inv.product_info_area + if product_area: print(''' Product Info Area: Manufacturer: %(manufacturer)s @@ -317,19 +311,19 @@ def cmd_fru_print(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: Serial Number: %(serial_number)s Asset: %(asset_tag)s FRU File ID: %(fru_file_id)s -'''[1:-1] % area.__dict__) +'''[1:-1] % product_area.__dict__) - if len(area.custom_mfg_info) != 0: + if len(product_area.custom_mfg_info) != 0: print(' Custom Board Info Records:') - for field in area.custom_mfg_info: + for field in product_area.custom_mfg_info: print(' %s' % field) # Multirecords - area = inv.multirecord_area - if area: + multirecord_area = inv.multirecord_area + if multirecord_area: print('Multirecord Area:') if args.all == 'all': - for record in area.records: + for record in multirecord_area.records: print(' %s' % record) else: print(' Skipped. Use "print all"') @@ -367,15 +361,14 @@ def cmd_chassis_status(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: status = ipmi.get_chassis_status() if args.json: - status = { + print(json.dumps({ 'power_on': status.power_on, 'overload': status.overload, 'interlock': status.interlock, 'fault': status.fault, 'ctrl_fault': status.control_fault, 'restore_policy': status.restore_policy - } - print(json.dumps(status)) + })) else: print(''' @@ -615,13 +608,11 @@ def cmd_vita_led_set(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: print('LED state has been updated') -def parse_interface_options(interface_name: str, options: str | list) -> dict: - if options: - options = options.split(',') - - interface_options = {} +def parse_interface_options(interface_name: str, + options: str | None) -> dict[str, Any]: + interface_options: dict[str, Any] = {} - for option in options: + for option in options.split(',') if options else []: (name, value) = option.split('=', 1) if interface_name == 'aardvark': if name == 'serial': @@ -681,14 +672,13 @@ def parse_interface_options(interface_name: str, options: str | list) -> dict: return interface_options -def create_ipmi_connection(interface_name: str, interface_options: str | list, +def create_ipmi_connection(interface_name: str, options: str | None, target_address: int, target_routing: str | list | None, rmcp_host: str | None, rmcp_port: int, rmcp_user: str, rmcp_password: str, - rmcp_priv_level: str | None) -> pyipmi.Ipmi: - interface_options = parse_interface_options(interface_name, - interface_options) + rmcp_priv_level: str | None) -> pyipmi.Ipmi | None: + interface_options = parse_interface_options(interface_name, options) try: interface = pyipmi.interfaces.create_interface(interface_name, @@ -1282,7 +1272,7 @@ def main(argv: list[str] | None = None) -> None: sys.exit(1) # interface could not be created, error is printed try: - if args.needs_connection: + if ipmi is not None: ipmi.open() # this will open interface and session args.func(ipmi, args) except pyipmi.errors.CompletionCodeError as e: @@ -1305,7 +1295,7 @@ def main(argv: list[str] | None = None) -> None: traceback.print_exc() sys.exit(1) finally: - if args.needs_connection: + if ipmi is not None: ipmi.close() # this will close interface and session From dbd179cd4fdd824e4cb2e482f8a28b4fb6627d90 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 19:22:12 +0200 Subject: [PATCH 16/16] ci: add a type check with mypy Run mypy on the pyipmi package in the lint workflow. The configuration is in setup.cfg, the optional dependencies pyserial, pyaardvark and cryptography are not installed for the check. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- .github/workflows/lint.yml | 18 ++++++++++++++++++ setup.cfg | 8 ++++++++ 2 files changed, 26 insertions(+) diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 685f60bd..468068af 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -20,6 +20,24 @@ jobs: - name: Lint with ruff run: ruff check --output-format=github . + mypy: + name: Type check with mypy + runs-on: ubuntu-24.04 + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Install dependencies + run: pip install mypy==2.4.0 + + # the files and options are configured in setup.cfg + - name: Type check with mypy + run: mypy + codespell: name: Check spelling with codespell runs-on: ubuntu-24.04 diff --git a/setup.cfg b/setup.cfg index c41db37c..ef26e4df 100644 --- a/setup.cfg +++ b/setup.cfg @@ -13,3 +13,11 @@ ignore = D100,D101,D102,D103,D104,D105,D203,D204 [codespell] skip = ./.git,*.bin,*.pyc,./tests/fru_bin,./tests/hpm_bin,./man/ipmitool.py.1 ignore-words-list = pres + +[mypy] +files = pyipmi + +# optional dependencies, they are not installed for the type check, and +# the version module, which is generated by setup.py +[mypy-serial.*,pyaardvark.*,cryptography.*,pyipmi.version] +ignore_missing_imports = True