From dd0a3d6d76fffc3f26227a68a0b10c8831a1b323 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 14:25:23 +0200 Subject: [PATCH 01/12] hpm: fix spell check warning Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 2 +- tests/test_hpm.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index 136e598..43606b3 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -462,7 +462,7 @@ def _from_rsp_data(self, data: bytes) -> None: support.append('reserved') if cap & self.PREPARATION_SUPPORT_MASK: - support.append('prepartion') + support.append('preparation') if cap & self.COMPARISON_SUPPORT_MASK: support.append('comparison') if cap & self.DEFERRED_ACTIVATION_SUPPORT_MASK: diff --git a/tests/test_hpm.py b/tests/test_hpm.py index b085f14..bb83a12 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -70,7 +70,7 @@ def test_deferredversion(self): def test_str(self): prop = ComponentProperty().from_data(PROPERTY_GENERAL_PROPERTIES, b'\x15') - assert str(prop) == ('General: rollback_is_supported, prepartion, ' + assert str(prop) == ('General: rollback_is_supported, preparation, ' 'deferred_activation') prop = ComponentProperty().from_data(PROPERTY_CURRENT_VERSION, From 7f59a926d60ba4f40d6708bb5135e100176daf82 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:08 +0200 Subject: [PATCH 02/12] tests: fix spelling in hpm component properties test The spelling of 'preparation' was fixed in the general component properties, update the test accordingly. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- tests/test_hpm.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_hpm.py b/tests/test_hpm.py index bb83a12..641c874 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -191,7 +191,7 @@ def test_get_component_properties(): ComponentPropertyGeneral, ComponentPropertyCurrentVersion, ComponentPropertyDescriptionString, ComponentPropertyDeferredVersion] assert props[0].general == ['rollback_backup_not_supported', - 'prepartion', 'comparison', + 'preparation', 'comparison', 'payload_cold_reset_required'] assert props[2].description == 'fw' From a8b93920a898da2bbfc142967b6a4778710155a7 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:13 +0200 Subject: [PATCH 03/12] ipmitool: fix hpm check without connection The 'hpm check' command does not need a connection, so no Ipmi object is passed to the command function. Calling open_upgrade_image() on it failed with an AttributeError. Open the upgrade image directly. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/ipmitool.py | 2 +- tests/test_ipmitool.py | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/pyipmi/ipmitool.py b/pyipmi/ipmitool.py index 3ff0dae..b34ae1e 100755 --- a/pyipmi/ipmitool.py +++ b/pyipmi/ipmitool.py @@ -346,7 +346,7 @@ def cmd_hpm_capabilities(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: def cmd_hpm_check_file(ipmi: pyipmi.Ipmi, args: argparse.Namespace) -> None: - cap = ipmi.open_upgrade_image(args.file) + cap = pyipmi.hpm.UpgradeImage(args.file) print(cap.header) for action in cap.actions: diff --git a/tests/test_ipmitool.py b/tests/test_ipmitool.py index 9d96721..e61df5f 100644 --- a/tests/test_ipmitool.py +++ b/tests/test_ipmitool.py @@ -2,6 +2,7 @@ import argparse import logging +import os import pytest @@ -243,6 +244,14 @@ def test_dcmi_set_conf_param_invalid_selector(self, capsys): def test_hpm_check_needs_no_connection(self): assert not self.parse('hpm check file.img').needs_connection + def test_hpm_check_runs_without_connection(self, capsys): + path = os.path.join(os.path.dirname(__file__), 'hpm_bin', + 'firmware.hpm') + ipmitool.main(['hpm', 'check', path]) + out = capsys.readouterr().out + assert 'HPM Upgrade Image header' in out + assert 'Upload for Upgrade' in out + def test_invalid_choice(self, capsys): with pytest.raises(SystemExit): self.parse('chassis power sideways') From e57985857dd419c0c758a4d101efb22ab63b597a Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:17 +0200 Subject: [PATCH 04/12] hpm: check the signature of the upgrade image A file that is not an HPM.1 upgrade image (e.g. a gzip compressed package) was parsed anyway and failed later with a misleading BCD decoding error of the version fields. Check the 'PICMGFWU' signature of the image header and raise an HpmError. ipmitool.py prints an HpmError as error message instead of a traceback. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 5 +++++ pyipmi/ipmitool.py | 5 +++++ tests/test_hpm.py | 6 ++++++ 3 files changed, 16 insertions(+) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index 43606b3..abe02f7 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -580,8 +580,13 @@ def __init__(self, data: bytes | None = None) -> None: if data: self._from_data(data) + SIGNATURE = b'PICMGFWU' + def _from_data(self, data: bytes) -> None: self.signature = data[0:8] + if self.signature != self.SIGNATURE: + raise HpmError('no HPM.1 upgrade image (invalid signature %s)' + % bytes(self.signature).hex(' ')) for a in self.FORMAT: setattr(self, a.field_name, struct.unpack( diff --git a/pyipmi/ipmitool.py b/pyipmi/ipmitool.py index b34ae1e..631dc0f 100755 --- a/pyipmi/ipmitool.py +++ b/pyipmi/ipmitool.py @@ -1279,6 +1279,11 @@ def main(argv: list[str] | None = None) -> None: if args.verbose: traceback.print_exc() sys.exit(1) + except pyipmi.errors.HpmError as e: + print('HPM error: %s' % e) + if args.verbose: + traceback.print_exc() + sys.exit(1) except KeyboardInterrupt: if args.verbose: traceback.print_exc() diff --git a/tests/test_hpm.py b/tests/test_hpm.py index 641c874..9ff3b70 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -16,6 +16,7 @@ UpgradeActionRecordPrepare, UpgradeActionRecordUploadForUpgrade, UpgradeActionRecordUploadForCompare, UpgradeImage, + UpgradeImageHeaderRecord, PROPERTY_GENERAL_PROPERTIES, PROPERTY_CURRENT_VERSION, PROPERTY_DESCRIPTION_STRING, PROPERTY_ROLLBACK_VERSION, PROPERTY_DEFERRED_VERSION, PROPERTY_OEM, @@ -502,3 +503,8 @@ def send_and_receive(req): ipmi.interface.send_and_receive.side_effect = send_and_receive ipmi.wait_until_new_firmware_comes_up(timeout=5, interval=1) assert fake_time.sleeps == [1, 1, 5] + + +def test_upgradeimageheaderrecord_invalid_signature(): + with pytest.raises(HpmError): + UpgradeImageHeaderRecord(b'\x1f\x8b\x08\x00' + bytes(31)) From b14913b6450fe24b691762f7d0082b4fafee2c79 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:21 +0200 Subject: [PATCH 05/12] msgs/hpm: fix bit order of the target upgrade capabilities The bits of the global capabilities of the Get Target Upgrade Capabilities response were decoded in reverse order. Bit 0 is the IPMC self-test support and bit 7 'firmware upgrade undesirable', as defined by HPM.1 and also used by ipmitool. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/msgs/hpm.py | 14 +++++++------- tests/msgs/test_hpm.py | 19 +++++++++++++++++++ 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/pyipmi/msgs/hpm.py b/pyipmi/msgs/hpm.py index 5f81de5..7b4d93a 100644 --- a/pyipmi/msgs/hpm.py +++ b/pyipmi/msgs/hpm.py @@ -43,14 +43,14 @@ class GetTargetUpgradeCapabilitiesRsp(PicmgMessage): GroupExtensionIdentifier('picmg_identifier', PICMG_IDENTIFIER), UnsignedInt('hpm_1_version', 1), Bitfield('capabilities', 1, - Bitfield.Bit('firmware_upgrade_undesirable', 1), - Bitfield.Bit('automatic_rollback_overriden', 1), - Bitfield.Bit('ipmc_degraded_during_upgrade', 1), - Bitfield.Bit('deferred_activation', 1), - Bitfield.Bit('services_affected_by_upgrade', 1), - Bitfield.Bit('manual_rollback', 1), + Bitfield.Bit('selftest', 1), Bitfield.Bit('automatic_rollback', 1), - Bitfield.Bit('selftest', 1),), + Bitfield.Bit('manual_rollback', 1), + Bitfield.Bit('services_affected_by_upgrade', 1), + Bitfield.Bit('deferred_activation', 1), + Bitfield.Bit('ipmc_degraded_during_upgrade', 1), + Bitfield.Bit('automatic_rollback_overriden', 1), + Bitfield.Bit('firmware_upgrade_undesirable', 1),), Bitfield('timeout', 4, Bitfield.Bit('upgrade', 8), Bitfield.Bit('selftest', 8), diff --git a/tests/msgs/test_hpm.py b/tests/msgs/test_hpm.py index d34f06e..56092b2 100644 --- a/tests/msgs/test_hpm.py +++ b/tests/msgs/test_hpm.py @@ -42,3 +42,22 @@ def test_activatefirmwarereq_encode_valid_req_wo_optional(): m.rollback_override_policy = None data = encode_message(m) assert data == b'\x00' + + +def test_gettargetupgradecapabilitiesrsp_decode_capabilities(): + m = pyipmi.msgs.hpm.GetTargetUpgradeCapabilitiesRsp() + # self-test, automatic/manual rollback and deferred activation + decode_message(m, b'\x00\x00\x00\x17\x0c\x0c\x0c\x0c\x03') + caps = m.capabilities + assert caps.selftest == 1 + assert caps.automatic_rollback == 1 + assert caps.manual_rollback == 1 + assert caps.services_affected_by_upgrade == 0 + assert caps.deferred_activation == 1 + assert caps.ipmc_degraded_during_upgrade == 0 + assert caps.automatic_rollback_overriden == 0 + assert caps.firmware_upgrade_undesirable == 0 + + decode_message(m, b'\x00\x00\x00\x80\x0c\x0c\x0c\x0c\x03') + assert m.capabilities.firmware_upgrade_undesirable == 1 + assert m.capabilities.selftest == 0 From 5ef320a02379c114d31b484318de19ee165aaeed Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:26 +0200 Subject: [PATCH 06/12] hpm: fix upgrade action records HPM.1 defines only three action record types in an upgrade image: backup components, prepare components and upload firmware image. The 'upload for compare' is no record type, it is an action of the Initiate Upgrade Action command. Remove the UpgradeActionRecordUploadForCompare class, a record with a reserved type raises an HpmError. Further fixes of the action records: - a record can be created without data - a truncated firmware image raises an HpmError - strip the '\x00' padding of the firmware description string - print the firmware version, description and length of the upload record - use the HPM.1 names of the action types Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 64 +++++++++++++++++++++++++++++++----------- tests/test_hpm.py | 57 +++++++++++++++++++++++++++++++++---- tests/test_ipmitool.py | 2 +- 3 files changed, 100 insertions(+), 23 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index abe02f7..8f99d12 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -41,11 +41,17 @@ PROPERTY_DEFERRED_VERSION = 4 PROPERTY_OEM = list(range(192, 255)) +# actions of the Initiate Upgrade Action command ACTION_BACKUP_COMPONENT = 0x00 ACTION_PREPARE_COMPONENT = 0x01 ACTION_UPLOAD_FOR_UPGRADE = 0x02 ACTION_UPLOAD_FOR_COMPARE = 0x03 +# action types of the upgrade image action records +IMAGE_ACTION_BACKUP_COMPONENTS = 0x00 +IMAGE_ACTION_PREPARE_COMPONENTS = 0x01 +IMAGE_ACTION_UPLOAD_FIRMWARE_IMAGE = 0x02 + CC_LONG_DURATION_CMD_IN_PROGRESS = 0x80 CC_GET_COMP_PROP_UPGRADE_NOT_SUPPORTED_OVER_INTF = 0x81 @@ -634,32 +640,36 @@ def __str__(self) -> str: class UpgradeActionRecord: ACTIONS = ( - "Backup", - "Prepare", - "Upload for Upgrade", - "Upload for Compare" + "Backup Components", + "Prepare Components", + "Upload Firmware Image", ) + HEADER_LENGTH = 3 + def __init__(self, data: bytes | None = None) -> None: - self.action_type = array('B', data)[0] + self.action_type = None + self.action = None + self.components = None + self.checksum = None + self.length = self.HEADER_LENGTH if data: (self.action, self.components, self.checksum) \ = struct.unpack('BBB', data[0:3]) - self.length = 3 + self.action_type = self.action @staticmethod def create_from_data(data: bytes) -> UpgradeActionRecord: action_type = array('B', data)[0] - if action_type == ACTION_BACKUP_COMPONENT: + if action_type == IMAGE_ACTION_BACKUP_COMPONENTS: return UpgradeActionRecordBackup(data) - elif action_type == ACTION_PREPARE_COMPONENT: + elif action_type == IMAGE_ACTION_PREPARE_COMPONENTS: return UpgradeActionRecordPrepare(data) - elif action_type == ACTION_UPLOAD_FOR_UPGRADE: + elif action_type == IMAGE_ACTION_UPLOAD_FIRMWARE_IMAGE: return UpgradeActionRecordUploadForUpgrade(data) - elif action_type == ACTION_UPLOAD_FOR_COMPARE: - return UpgradeActionRecordUploadForCompare(data) else: - raise HpmError('unsupported ActionRecord') + raise HpmError('unsupported ActionRecord type 0x%02x' + % action_type) def __str__(self) -> str: str = [] @@ -678,21 +688,43 @@ class UpgradeActionRecordPrepare(UpgradeActionRecord): class UpgradeActionRecordUploadForUpgrade(UpgradeActionRecord): + """Upload Firmware Image action record. + + The action header is followed by the firmware version (6 bytes), the + description string (21 bytes), the firmware length (4 bytes) and the + firmware image. The image is uploaded for upgrade or for compare, + this is selected by the Initiate Upgrade Action command. + """ + def __init__(self, data: bytes | None = None) -> None: UpgradeActionRecord.__init__(self, data) + self.firmware_version = None + self.firmware_description_string = None + self.firmware_length = None + self.firmware_image_data = None if data: self.firmware_version = \ VersionField( data[3:3 + VersionField.VERSION_WITH_AUX_FIELD_LEN]) + # strip the '\x00' padding self.firmware_description_string \ - = py3dec_unic_bytes_fix(data[9:30]) + = py3dec_unic_bytes_fix(data[9:30]).rstrip('\0') self.firmware_length = struct.unpack(' str: + str = [UpgradeActionRecord.__str__(self)] + str.append(" Firmware Version: %s" % self.firmware_version) + str.append(" Description: %s" + % self.firmware_description_string) + str.append(" Firmware Length: %s" % self.firmware_length) + return "\n".join(str) class ImageChecksumRecord: diff --git a/tests/test_hpm.py b/tests/test_hpm.py index 9ff3b70..b97973b 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -1,6 +1,7 @@ #!/usr/bin/env python import os +import struct import pytest @@ -14,8 +15,7 @@ ComponentPropertyRollbackVersion, UpgradeActionRecord, UpgradeActionRecordBackup, UpgradeActionRecordPrepare, - UpgradeActionRecordUploadForUpgrade, - UpgradeActionRecordUploadForCompare, UpgradeImage, + UpgradeActionRecordUploadForUpgrade, UpgradeImage, UpgradeImageHeaderRecord, PROPERTY_GENERAL_PROPERTIES, PROPERTY_CURRENT_VERSION, PROPERTY_DESCRIPTION_STRING, PROPERTY_ROLLBACK_VERSION, @@ -114,9 +114,54 @@ def test_upgradeactionrecord_create_from_data(): assert record.firmware_description_string == '012345678901234567890' assert record.firmware_length == 4 - record = UpgradeActionRecord.create_from_data(b'\x03\x08\x02') - assert record.action == 3 - assert type(record) is UpgradeActionRecordUploadForCompare + record = UpgradeActionRecord.create_from_data( + _upload_record(b'\x11\x22\x33')) + assert type(record) is UpgradeActionRecordUploadForUpgrade + assert record.firmware_version.version_to_string() == '1.2' + assert record.firmware_description_string == 'firmware' + assert record.firmware_length == 3 + assert record.firmware_image_data == b'\x11\x22\x33' + assert record.length == 3 + 31 + 3 + + +def test_upgradeactionrecord_reserved_type(): + # compare is no action record type, only an Initiate Upgrade Action + with pytest.raises(HpmError, match='unsupported ActionRecord type 0x03'): + UpgradeActionRecord.create_from_data(b'\x03\x02\xfb') + + +def _upload_record(image, components=0x02): + return (bytes((0x02, components, 0)) + + b'\x01\x02\x00\x00\x00\x00' + + b'firmware'.ljust(21, b'\x00') + + struct.pack(' Date: Tue, 6 Oct 2026 15:16:30 +0200 Subject: [PATCH 07/12] hpm: add the compare of a component Add compare_component_from_image() and compare_component_from_file() to compare the firmware of an image with the active copy of a component. In compare mode the upgrade stage skips the backup and prepare actions and uploads the firmware image with the 'upload for compare' action. The upgrade stage maps the action record types to the actions of the Initiate Upgrade Action command instead of using the record type directly. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 39 +++++++++++++++++++++++++-- tests/test_hpm.py | 67 ++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 103 insertions(+), 3 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index 8f99d12..58038ae 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -342,12 +342,30 @@ def preparation_stage(self, image: UpgradeImage) -> None: if support is not True: raise HpmError('no supported component in image') - def upgrade_stage(self, image: UpgradeImage, component: int) -> None: + def upgrade_stage(self, image: UpgradeImage, component: int, + compare: bool = False) -> None: + """Perform the action records of the image for the component. + + With `compare` the firmware image is uploaded for comparison with + the active copy of the component, the backup and prepare actions + are skipped. + """ for action in image.actions: if action.components & (1 << component) == 0: continue + if isinstance(action, UpgradeActionRecordUploadForUpgrade): + if compare: + upgrade_action = ACTION_UPLOAD_FOR_COMPARE + else: + upgrade_action = ACTION_UPLOAD_FOR_UPGRADE + elif compare: + continue + elif isinstance(action, UpgradeActionRecordBackup): + upgrade_action = ACTION_BACKUP_COMPONENT + else: + upgrade_action = ACTION_PREPARE_COMPONENT self.initiate_upgrade_action_and_wait(1 << component, - action.action_type) + upgrade_action) if isinstance(action, UpgradeActionRecordUploadForUpgrade): self.upload_binary(action.firmware_image_data) self.finish_upload_and_wait(component, action.firmware_length) @@ -388,6 +406,23 @@ def install_component_from_file(self, filename: str, component: int) -> None: image = UpgradeImage(filename) self.install_component_from_image(image, component) + def compare_component_from_image(self, image: UpgradeImage, + component: int) -> None: + """Compare the firmware of the image with the active copy. + + A mismatch is reported by the Finish Firmware Upload command. + """ + self.abort_firmware_upgrade() + if component not in image.header.components: + raise HpmError('component=%d not in image' % component) + self.preparation_stage(image) + self.upgrade_stage(image, component, compare=True) + + def compare_component_from_file(self, filename: str, + component: int) -> None: + image = UpgradeImage(filename) + self.compare_component_from_image(image, component) + class UpgradeStatus(State): diff --git a/tests/test_hpm.py b/tests/test_hpm.py index b97973b..86ee978 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -1,5 +1,6 @@ #!/usr/bin/env python +import hashlib import os import struct @@ -20,7 +21,8 @@ PROPERTY_GENERAL_PROPERTIES, PROPERTY_CURRENT_VERSION, PROPERTY_DESCRIPTION_STRING, PROPERTY_ROLLBACK_VERSION, PROPERTY_DEFERRED_VERSION, PROPERTY_OEM, - ACTION_PREPARE_COMPONENT, ACTION_UPLOAD_FOR_UPGRADE) + ACTION_BACKUP_COMPONENT, ACTION_PREPARE_COMPONENT, + ACTION_UPLOAD_FOR_UPGRADE, ACTION_UPLOAD_FOR_COMPARE) from .ipmi_helper import create_ipmi @@ -164,6 +166,69 @@ def test_upgradeactionrecord_upload_str(): ' Firmware Length: 3']) +def _image_with_actions(tmp_path, actions): + """Write an image for device ID 4, manufacturer 15000, product 1701 + with component 1 and return the filename.""" + header = (b'PICMGFWU\x00\x04\x98\x3a\x00\xa5\x06' + bytes(5) + + b'\x02' + bytes(13)) + header += bytes(((-sum(header)) & 0xff,)) + data = header + b''.join(actions) + path = tmp_path / 'image.hpm' + path.write_bytes(data + hashlib.md5(data).digest()) + return str(path) + + +UPGRADE_STAGE_RSP = { + 'InitiateUpgradeAction': b'\x00\x00', + 'UploadFirmwareBlock': b'\x00\x00', + 'FinishFirmwareUpload': b'\x00\x00', +} + + +def _initiated_actions(ipmi): + return [data[-1] for (name, data) in ipmi.requests + if name == 'InitiateUpgradeActionReq'] + + +def test_upgrade_stage_actions(tmp_path): + image = UpgradeImage(_image_with_actions(tmp_path, [ + b'\x00\x02\xfe', b'\x01\x02\xfd', _upload_record(bytes(30))])) + ipmi = create_ipmi(UPGRADE_STAGE_RSP) + ipmi.upgrade_stage(image, 1) + assert _initiated_actions(ipmi) == [ + ACTION_BACKUP_COMPONENT, ACTION_PREPARE_COMPONENT, + ACTION_UPLOAD_FOR_UPGRADE] + assert ipmi.requests[-1][0] == 'FinishFirmwareUploadReq' + + +def test_upgrade_stage_compare(tmp_path): + image = UpgradeImage(_image_with_actions(tmp_path, [ + b'\x00\x02\xfe', b'\x01\x02\xfd', _upload_record(bytes(30))])) + ipmi = create_ipmi(UPGRADE_STAGE_RSP) + ipmi.upgrade_stage(image, 1, compare=True) + # backup and prepare are skipped + assert _initiated_actions(ipmi) == [ACTION_UPLOAD_FOR_COMPARE] + names = [name for (name, _) in ipmi.requests] + assert names.count('UploadFirmwareBlockReq') == 2 + assert names[-1] == 'FinishFirmwareUploadReq' + + +def test_compare_component_from_file(tmp_path): + filename = _image_with_actions(tmp_path, [ + b'\x01\x02\xfd', _upload_record(bytes(30))]) + ipmi = create_ipmi({ + **UPGRADE_STAGE_RSP, + 'AbortFirmwareUpgrade': b'\x00\x00', + 'GetDeviceId': DEVICE_ID_RSP, + 'GetTargetUpgradeCapabilities': TARGET_CAPS_RSP, + }) + ipmi.compare_component_from_file(filename, 1) + names = [name for (name, _) in ipmi.requests] + assert _initiated_actions(ipmi) == [ACTION_UPLOAD_FOR_COMPARE] + assert 'ActivateFirmwareReq' not in names + assert names[-1] == 'FinishFirmwareUploadReq' + + def test_upgrade_image(): path = os.path.dirname(os.path.abspath(__file__)) hpm_file = os.path.join(path, 'hpm_bin/firmware.hpm') From 773dcf5b52eeaf1ee44d163d4dc11404b7fb3e58 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:35 +0200 Subject: [PATCH 08/12] hpm: verify the MD5 checksum of the upgrade image The MD5 checksum was never verified and the calculated checksum was always None, because the return value of hashlib's update() was used instead of the digest. Calculate the digest and raise an HpmError on a mismatch. The checksum is verified after the image header, so a file that is no upgrade image is reported by the signature check. Read the file with a context manager, a missing file raised a NameError after printing an error message. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 31 +++++++++++-------------------- tests/test_hpm.py | 9 +++++++++ 2 files changed, 20 insertions(+), 20 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index 58038ae..41f431e 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -16,7 +16,6 @@ from __future__ import annotations -import os import codecs import struct import collections @@ -786,33 +785,27 @@ def __str__(self) -> str: return "\n".join(str) def _check_md5_sum(self, filedata: bytes) -> None: - summer = hashlib.md5() - self.checksum_actual \ - = summer.update(filedata[:-HPM_IMAGE_CHECKSUM_SIZE]) + self.checksum_actual = hashlib.md5( + filedata[:-HPM_IMAGE_CHECKSUM_SIZE]).digest() self.checksum_expected = filedata[-HPM_IMAGE_CHECKSUM_SIZE:] + if self.checksum_actual != self.checksum_expected: + raise HpmError('image MD5 checksum mismatch') def _from_file(self, filename: str) -> None: - try: - file = open(filename, "rb") - except OSError: - print('Error open file "%s"' % filename) - - ################################ - # get file size - file_size = os.stat(filename).st_size - file_data = file.read(file_size) - - ################################ - # get image checksum - self._check_md5_sum(file_data) - # XXX verify checksum + with open(filename, "rb") as file: + file_data = file.read() + file_size = len(file_data) ################################ # Upgrade Image Header self.header = UpgradeImageHeaderRecord(file_data) off = self.header.length + ################################ + # verify image checksum + self._check_md5_sum(file_data) + ################################ # Upgrade Actions self.actions = [] @@ -824,5 +817,3 @@ def _from_file(self, filename: str) -> None: ################################ # Image checksum self.checksum = ImageChecksumRecord(file_data[off:file_size]) - - file.close() diff --git a/tests/test_hpm.py b/tests/test_hpm.py index 86ee978..e7eb170 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -178,6 +178,15 @@ def _image_with_actions(tmp_path, actions): return str(path) +def test_upgrade_image_md5_mismatch(tmp_path): + filename = _image_with_actions(tmp_path, [b'\x01\x02\xfd']) + with open(filename, 'r+b') as f: + f.seek(-1, os.SEEK_END) + f.write(b'\x00') + with pytest.raises(HpmError, match='MD5'): + UpgradeImage(filename) + + UPGRADE_STAGE_RSP = { 'InitiateUpgradeAction': b'\x00\x00', 'UploadFirmwareBlock': b'\x00\x00', From 1dfee8ebe747dd6a505ee5ac69bd9cc255babbe0 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:39 +0200 Subject: [PATCH 09/12] hpm: resend an upload block after a timeout On a timeout of the Upload Firmware Block command the retry counter was decremented, but the block was not sent again. The next block was uploaded instead and the firmware data of the timed out block was missing. Send the same block again, up to 'retry' times. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 31 +++++++++++++++++-------------- tests/test_hpm.py | 22 ++++++++++++++++++++++ 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index 41f431e..4937523 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -164,20 +164,23 @@ def upload_binary(self, binary: bytes, timeout: float = 2, block_size = self._determine_max_block_size() for chunk in chunks(binary, block_size): - try: - self.upload_firmware_block(block_number, chunk) - except CompletionCodeError as e: - if e.cc == CC_LONG_DURATION_CMD_IN_PROGRESS: - - self.wait_for_long_duration_command( - constants.CMDID_HPM_UPLOAD_FIRMWARE_BLOCK, - timeout, interval) - else: - raise HpmError('upload_firmware_block CC=0x%02x' % e.cc) from e - except IpmiTimeoutError: - retry -= 1 - if retry == 0: - raise IpmiTimeoutError() from None + # a timed out block is sent again, up to `retry` times + for attempt in range(retry): + try: + self.upload_firmware_block(block_number, chunk) + except CompletionCodeError as e: + if e.cc == CC_LONG_DURATION_CMD_IN_PROGRESS: + self.wait_for_long_duration_command( + constants.CMDID_HPM_UPLOAD_FIRMWARE_BLOCK, + timeout, interval) + else: + raise HpmError('upload_firmware_block CC=0x%02x' + % e.cc) from e + except IpmiTimeoutError: + if attempt == retry - 1: + raise IpmiTimeoutError() from None + continue + break block_number += 1 block_number &= 0xff diff --git a/tests/test_hpm.py b/tests/test_hpm.py index e7eb170..d5dc3c2 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -470,6 +470,28 @@ def test_upload_binary_timeout(): ipmi.interface.send_and_receive.side_effect = IpmiTimeoutError() with pytest.raises(IpmiTimeoutError): ipmi.upload_binary(bytes(100), retry=2) + # the first block is tried `retry` times + assert ipmi.interface.send_and_receive.call_count == 2 + + +def test_upload_binary_timeout_resends_block(): + ipmi = create_ipmi(b'\x00\x00') + respond = ipmi.interface.send_and_receive.side_effect + errors = [IpmiTimeoutError()] + + def send_and_receive(req): + # the second block times out once + if req.number == 1 and errors: + raise errors.pop() + return respond(req) + + ipmi.interface.send_and_receive.side_effect = send_and_receive + ipmi.upload_binary(bytes(range(50))) + assert ipmi.requests == [ + ('UploadFirmwareBlockReq', b'\x00\x00' + bytes(range(22))), + ('UploadFirmwareBlockReq', b'\x00\x01' + bytes(range(22, 44))), + ('UploadFirmwareBlockReq', b'\x00\x02' + bytes(range(44, 50))), + ] def test_finish_firmware_upload(): From b42525aad7010226bc8e5d3746b8260116f23638 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:43 +0200 Subject: [PATCH 10/12] hpm: stop waiting when the new firmware comes up wait_until_new_firmware_comes_up() polled the controller until the timeout, also when the controller answered again. Return as soon as the controller answers after it was not accessible. Before it is not accessible, an answer may still be from the old firmware, so wait until the timeout in this case. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 16 +++++++++++++--- tests/test_hpm.py | 11 ++++++++++- 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index 4937523..e09ad76 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -377,15 +377,25 @@ def _activation_state_do_self_testing(self) -> None: def wait_until_new_firmware_comes_up(self, timeout: float, interval: float) -> None: + """Wait until the controller answers again after the activation. + + Only an answer after the controller was not accessible is from the + new firmware, before it may still be the old one. If the controller + does not become inaccessible, wait until the timeout. + """ + was_inaccessible = False start_time = time.time() while time.time() < start_time + timeout: try: self.get_upgrade_status() self.get_device_id() - except IpmiTimeoutError: - time.sleep(interval) - except OSError: + except (IpmiTimeoutError, OSError): + was_inaccessible = True time.sleep(interval) + continue + if was_inaccessible: + break + time.sleep(interval) time.sleep(5) def activation_stage(self, image: UpgradeImage, component: int) -> None: diff --git a/tests/test_hpm.py b/tests/test_hpm.py index d5dc3c2..6becfab 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -642,10 +642,19 @@ def send_and_receive(req): return respond(req) ipmi.interface.send_and_receive.side_effect = send_and_receive - ipmi.wait_until_new_firmware_comes_up(timeout=5, interval=1) + ipmi.wait_until_new_firmware_comes_up(timeout=50, interval=1) + # returns when the controller answers again assert fake_time.sleeps == [1, 1, 5] +def test_wait_until_new_firmware_comes_up_without_reset(fake_time): + ipmi = create_ipmi({'GetUpgradeStatus': b'\x00\x00\x00\x00', + 'GetDeviceId': DEVICE_ID_RSP}) + ipmi.wait_until_new_firmware_comes_up(timeout=5, interval=1) + # the answers may be from the old firmware, wait until the timeout + assert fake_time.sleeps == [1] * 5 + [5] + + def test_upgradeimageheaderrecord_invalid_signature(): with pytest.raises(HpmError): UpgradeImageHeaderRecord(b'\x1f\x8b\x08\x00' + bytes(31)) From e87eb85ff3b801d27a17cbbf308112b7e443e5ca Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:48 +0200 Subject: [PATCH 11/12] hpm: determine the upload block size from interface and routing The firmware block size was fixed at 22 bytes. An IPMB message has 25 bytes request data, 2 bytes are used by the PICMG identifier and the block number, so a block can hold 23 bytes. A message sent through two bridges is embedded in a Send Message request on the IPMB and the fixed size was too large. Determine the block size like ipmitool: - a message sent directly by the interface is limited by the interface, the new MAX_REQUEST_DATA_SIZE attribute (38 bytes for RMCP) - a bridged message is limited by the IPMB message length, 8 bytes less for each additional bridge If the target rejects the length of the first block (CC 0xc7 or 0xc8), the block size is reduced until a block is accepted. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 52 ++++++++++++++++++++++++++++++++++----- pyipmi/interfaces/base.py | 4 +++ pyipmi/interfaces/rmcp.py | 2 ++ tests/test_hpm.py | 49 +++++++++++++++++++++++++++++++----- 4 files changed, 95 insertions(+), 12 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index e09ad76..ab7bb07 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -27,7 +27,7 @@ from .errors import CompletionCodeError, HpmError, IpmiTimeoutError from .msgs import create_request_by_name, Message from .msgs import constants -from .utils import check_completion_code, bcd_search, chunks +from .utils import check_completion_code, bcd_search from .utils import py3dec_unic_bytes_fix, py3_array_tobytes from .state import State from .fields import VersionField @@ -53,6 +53,13 @@ CC_LONG_DURATION_CMD_IN_PROGRESS = 0x80 +# an IPMB message is limited to 32 bytes, this leaves 25 bytes request data +IPMB_MAX_REQUEST_DATA_SIZE = 25 +# a bridged message is embedded in a Send Message request on the IPMB +SEND_MESSAGE_OVERHEAD = 8 +# PICMG identifier and block number of the Upload Firmware Block request +UPLOAD_FIRMWARE_BLOCK_HEADER_SIZE = 2 + CC_GET_COMP_PROP_UPGRADE_NOT_SUPPORTED_OVER_INTF = 0x81 CC_GET_COMP_PROP_INVALID_COMPONENT = 0x82 CC_GET_COMP_PROP_INVALID_PROPERTIES_SELECTOR = 0x83 @@ -153,17 +160,39 @@ def upload_firmware_block(self, block_number: int, data: bytes) -> None: self.send_message_with_name('UploadFirmwareBlock', number=block_number, data=data) - @staticmethod - def _determine_max_block_size() -> int: - return 22 + def _determine_max_block_size(self) -> int: + """Return the maximum firmware data length of an upload block. + + A message sent directly by the interface is limited by the + interface, a bridged message by the IPMB message length. Each + additional bridge embeds the message in another Send Message + request on the IPMB. + """ + routing = getattr(self.target, 'routing', None) + bridges = len(routing) - 1 if routing else 0 + if bridges > 0: + 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): + size = IPMB_MAX_REQUEST_DATA_SIZE + return size - UPLOAD_FIRMWARE_BLOCK_HEADER_SIZE def upload_binary(self, binary: bytes, timeout: float = 2, interval: float = 0.1, retry: int = 3) -> None: - """Upload all firmware blocks from a binary.""" + """Upload all firmware blocks from a binary. + + If the target rejects the length of the first block, the block size + is reduced until a block is accepted. + """ block_number = 0 block_size = self._determine_max_block_size() + block_size_accepted = False + offset = 0 - for chunk in chunks(binary, block_size): + while offset < len(binary): + chunk = binary[offset:offset + block_size] # a timed out block is sent again, up to `retry` times for attempt in range(retry): try: @@ -173,6 +202,11 @@ def upload_binary(self, binary: bytes, timeout: float = 2, self.wait_for_long_duration_command( constants.CMDID_HPM_UPLOAD_FIRMWARE_BLOCK, timeout, interval) + elif (e.cc in (constants.CC_REQ_DATA_INV_LENGTH, + constants.CC_REQ_DATA_FIELD_EXCEED) + and not block_size_accepted and block_size > 1): + block_size -= 1 + chunk = None else: raise HpmError('upload_firmware_block CC=0x%02x' % e.cc) from e @@ -182,6 +216,12 @@ def upload_binary(self, binary: bytes, timeout: float = 2, continue break + if chunk is None: + # send the block again with the reduced size + continue + + block_size_accepted = True + offset += len(chunk) block_number += 1 block_number &= 0xff diff --git a/pyipmi/interfaces/base.py b/pyipmi/interfaces/base.py index 8b6a6f3..daecebb 100644 --- a/pyipmi/interfaces/base.py +++ b/pyipmi/interfaces/base.py @@ -34,6 +34,10 @@ class Interface: NAME: str | None = None + # maximum request data length of a message sent directly (not bridged) + # by the interface, None for the IPMB default + MAX_REQUEST_DATA_SIZE: int | None = None + def open(self) -> None: pass diff --git a/pyipmi/interfaces/rmcp.py b/pyipmi/interfaces/rmcp.py index 18c0c94..616eae8 100644 --- a/pyipmi/interfaces/rmcp.py +++ b/pyipmi/interfaces/rmcp.py @@ -366,6 +366,8 @@ def check_header(self) -> None: class Rmcp(Interface): NAME = 'rmcp' + # 45 bytes LAN message length minus 7 bytes message header and checksums + MAX_REQUEST_DATA_SIZE = 38 _session: Session | None = None diff --git a/tests/test_hpm.py b/tests/test_hpm.py index 6becfab..f9b60d5 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -6,6 +6,7 @@ import pytest +import pyipmi from pyipmi.errors import HpmError, IpmiTimeoutError from pyipmi.hpm import (Hpm, ComponentProperty, ComponentPropertyDescriptionString, @@ -444,9 +445,9 @@ def test_upload_binary(): ipmi = create_ipmi(b'\x00\x00') ipmi.upload_binary(bytes(range(50))) assert ipmi.requests == [ - ('UploadFirmwareBlockReq', b'\x00\x00' + bytes(range(22))), - ('UploadFirmwareBlockReq', b'\x00\x01' + bytes(range(22, 44))), - ('UploadFirmwareBlockReq', b'\x00\x02' + bytes(range(44, 50))), + ('UploadFirmwareBlockReq', b'\x00\x00' + bytes(range(23))), + ('UploadFirmwareBlockReq', b'\x00\x01' + bytes(range(23, 46))), + ('UploadFirmwareBlockReq', b'\x00\x02' + bytes(range(46, 50))), ] @@ -488,12 +489,48 @@ def send_and_receive(req): ipmi.interface.send_and_receive.side_effect = send_and_receive ipmi.upload_binary(bytes(range(50))) assert ipmi.requests == [ + ('UploadFirmwareBlockReq', b'\x00\x00' + bytes(range(23))), + ('UploadFirmwareBlockReq', b'\x00\x01' + bytes(range(23, 46))), + ('UploadFirmwareBlockReq', b'\x00\x02' + bytes(range(46, 50))), + ] + + +@pytest.mark.parametrize('max_request_data_size, routing, block_size', [ + # directly on the IPMB + (None, None, 23), + # LAN interface directly to the BMC + (38, None, 36), + # bridged once, the message is on the IPMB + (38, [(0x81, 0x20, 0), (0x20, 0x88, None)], 23), + # bridged twice, the message is in a Send Message on the IPMB + (38, [(0x81, 0x20, 0), (0x20, 0x82, 7), (0x20, 0x72, None)], 15), +]) +def test_determine_max_block_size(max_request_data_size, routing, block_size): + ipmi = create_ipmi(b'\x00\x00') + ipmi.interface.MAX_REQUEST_DATA_SIZE = max_request_data_size + ipmi.target = pyipmi.Target(0x88, routing=routing) + assert ipmi._determine_max_block_size() == block_size + + +def test_upload_binary_reduces_block_size(): + ipmi = create_ipmi({'UploadFirmwareBlock': [ + b'\xc7', b'\xc8', b'\x00\x00', b'\x00\x00', b'\x00\x00']}) + ipmi.upload_binary(bytes(range(50))) + assert ipmi.requests == [ + ('UploadFirmwareBlockReq', b'\x00\x00' + bytes(range(23))), ('UploadFirmwareBlockReq', b'\x00\x00' + bytes(range(22))), - ('UploadFirmwareBlockReq', b'\x00\x01' + bytes(range(22, 44))), - ('UploadFirmwareBlockReq', b'\x00\x02' + bytes(range(44, 50))), + ('UploadFirmwareBlockReq', b'\x00\x00' + bytes(range(21))), + ('UploadFirmwareBlockReq', b'\x00\x01' + bytes(range(21, 42))), + ('UploadFirmwareBlockReq', b'\x00\x02' + bytes(range(42, 50))), ] +def test_upload_binary_length_error_after_accepted_block(): + ipmi = create_ipmi({'UploadFirmwareBlock': [b'\x00\x00', b'\xc7']}) + with pytest.raises(HpmError, match='CC=0xc7'): + ipmi.upload_binary(bytes(50)) + + def test_finish_firmware_upload(): ipmi = create_ipmi(b'\x00\x00') ipmi.finish_upload_and_wait(1, 0x12345) @@ -610,7 +647,7 @@ def test_install_component_from_file(fake_time): names = [name for (name, _) in ipmi.requests] image = UpgradeImage(HPM_FILE) - blocks = -(-image.actions[1].firmware_length // 22) + blocks = -(-image.actions[1].firmware_length // 23) assert names[:6] == [ 'AbortFirmwareUpgradeReq', 'GetDeviceIdReq', 'GetTargetUpgradeCapabilitiesReq', From 4909944b8c57d91a374094cca19656ce7130dd87 Mon Sep 17 00:00:00 2001 From: Heiko Thiery Date: Tue, 6 Oct 2026 15:16:52 +0200 Subject: [PATCH 12/12] hpm: check the component before contacting the controller Check if the component is in the image before the firmware upgrade is aborted, so nothing is sent to the controller on a wrong component. The error message lists the components of the image. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Heiko Thiery --- pyipmi/hpm.py | 13 +++++++++---- tests/test_hpm.py | 11 ++++++++--- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/pyipmi/hpm.py b/pyipmi/hpm.py index ab7bb07..8cd6948 100644 --- a/pyipmi/hpm.py +++ b/pyipmi/hpm.py @@ -445,11 +445,17 @@ def activation_stage(self, image: UpgradeImage, component: int) -> None: image.header.inaccessibility_timeout, 1) self._activation_state_do_self_testing() + @staticmethod + def _check_component_in_image(image: UpgradeImage, + component: int) -> None: + if component not in image.header.components: + raise HpmError('component=%d not in image (image components: %s)' + % (component, image.header.components)) + def install_component_from_image(self, image: UpgradeImage, component: int) -> None: + self._check_component_in_image(image, component) self.abort_firmware_upgrade() - if component not in image.header.components: - raise HpmError('component=%d not in image' % component) self.preparation_stage(image) self.upgrade_stage(image, component) self.activation_stage(image, component) @@ -464,9 +470,8 @@ def compare_component_from_image(self, image: UpgradeImage, A mismatch is reported by the Finish Firmware Upload command. """ + self._check_component_in_image(image, component) self.abort_firmware_upgrade() - if component not in image.header.components: - raise HpmError('component=%d not in image' % component) self.preparation_stage(image) self.upgrade_stage(image, component, compare=True) diff --git a/tests/test_hpm.py b/tests/test_hpm.py index f9b60d5..dcaa9ec 100644 --- a/tests/test_hpm.py +++ b/tests/test_hpm.py @@ -659,10 +659,15 @@ def test_install_component_from_file(fake_time): assert 'ActivateFirmwareReq' in names -def test_install_component_not_in_image(): +@pytest.mark.parametrize('method', ['install_component_from_image', + 'compare_component_from_image']) +def test_component_not_in_image(method): ipmi = create_ipmi(b'\x00\x00') - with pytest.raises(HpmError, match='component=0 not in image'): - ipmi.install_component_from_image(UpgradeImage(HPM_FILE), 0) + with pytest.raises(HpmError, match=r'component=0 not in image ' + r'\(image components: \[1\]\)'): + getattr(ipmi, method)(UpgradeImage(HPM_FILE), 0) + # nothing is sent to the controller + assert ipmi.requests == [] def test_wait_until_new_firmware_comes_up(fake_time):