Feature/hpm improvements

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
 

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
 

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:
@@ -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()

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),

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

tests/test_hpm.py

@@ -1,9 +1,12 @@
 #!/usr/bin/env python
 
+import hashlib
 import os
+import struct
 
 import pytest
 
+import pyipmi
 from pyipmi.errors import HpmError, IpmiTimeoutError
 from pyipmi.hpm import (Hpm, ComponentProperty,
                         ComponentPropertyDescriptionString,
@@ -14,12 +17,13 @@
                         ComponentPropertyRollbackVersion,
                         UpgradeActionRecord, UpgradeActionRecordBackup,
                         UpgradeActionRecordPrepare,
-                        UpgradeActionRecordUploadForUpgrade,
-                        UpgradeActionRecordUploadForCompare, UpgradeImage,
+                        UpgradeActionRecordUploadForUpgrade, UpgradeImage,
+                        UpgradeImageHeaderRecord,
                         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
 
@@ -70,7 +74,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,
@@ -113,9 +117,126 @@ 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('<L', len(image)) + image)
+
+
+@pytest.mark.parametrize('cls', [
+    UpgradeActionRecordBackup, UpgradeActionRecordPrepare,
+    UpgradeActionRecordUploadForUpgrade,
+])
+def test_upgradeactionrecord_without_data(cls):
+    record = cls()
+    assert record.action is None
+    assert record.components is None
+
+
+def test_upgradeactionrecord_upload_truncated():
+    data = _upload_record(b'\x11\x22\x33')
+    with pytest.raises(HpmError, match='truncated'):
+        UpgradeActionRecord.create_from_data(data[:-1])
+
+
+def test_upgradeactionrecord_upload_str():
+    record = UpgradeActionRecord.create_from_data(
+        _upload_record(b'\x11\x22\x33'))
+    assert str(record) == '\n'.join([
+        'Action Record Type: 0x2 (Upload Firmware Image) ',
+        ' Components: 0x02',
+        ' Firmware Version: 1.2',
+        ' Description:      firmware',
+        ' 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)
+
+
+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',
+    '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():
@@ -191,7 +312,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'
 
@@ -324,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))),
     ]
 
 
@@ -350,6 +471,64 @@ 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(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\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():
@@ -422,7 +601,7 @@ def test_upgrade_image_header_str():
     assert 'Device ID:        4' in s
     assert 'Manufacturer:     15000' in s
     assert str(image.actions[1]).startswith(
-        'Action Record Type: 0x2 (Upload for Upgrade)')
+        'Action Record Type: 0x2 (Upload Firmware Image)')
 
 
 def test_upgrade_action_record_invalid():
@@ -468,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',
@@ -480,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):
@@ -500,5 +684,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))