diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 685f60b..468068a 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/man/ipmitool.py.1 b/man/ipmitool.py.1 index 5838c0b..809dbaa 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/__init__.py b/pyipmi/__init__.py index 99794be..89a791b 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 f0cd2f1..674dbc4 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 08e25fb..4d87ea1 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')) @@ -227,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/dcmi.py b/pyipmi/dcmi.py index 2047abc..1adf9af 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/errors.py b/pyipmi/errors.py index 9af7065..bbda5e9 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/event.py b/pyipmi/event.py index ec06eb3..0c4ace3 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/fields.py b/pyipmi/fields.py index bbcd88f..fd6efef 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 @@ -98,7 +99,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: @@ -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 3d10d70..5379f04 100644 --- a/pyipmi/fru.py +++ b/pyipmi/fru.py @@ -20,12 +20,14 @@ import codecs import datetime import os +from collections.abc import Sequence -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 from .fields import FruTypeLengthString +from .mixin import IpmiMixin codecs.register(bcd_search) @@ -33,7 +35,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 @@ -66,9 +68,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 @@ -124,12 +126,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 +198,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 @@ -276,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: @@ -287,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): @@ -301,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 @@ -324,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)' % @@ -359,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 @@ -371,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] @@ -392,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 @@ -428,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] @@ -442,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: @@ -482,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: @@ -494,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) @@ -506,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) @@ -515,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: @@ -531,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/hpm.py b/pyipmi/hpm.py index 8cd6948..a30229f 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 @@ -31,6 +32,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 +78,7 @@ CC_ABORT_UPGRADE_CANNOT_RESUME_OPERATION = 0x81 -class Hpm: +class Hpm(IpmiMixin): @staticmethod def _get_component_count(components: int) -> int: @@ -114,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 @@ -174,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 @@ -193,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: @@ -206,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 @@ -216,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 @@ -393,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: @@ -409,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) @@ -515,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] @@ -536,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): @@ -546,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] @@ -576,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: @@ -585,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 @@ -598,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: @@ -607,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: @@ -616,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: @@ -672,6 +684,17 @@ class UpgradeImageHeaderRecord: image_header('oem_data_length', ' None: for a in self.FORMAT: setattr(self, a.field_name, None) @@ -694,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) @@ -710,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: @@ -764,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): @@ -811,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: @@ -833,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( @@ -866,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) diff --git a/pyipmi/interfaces/__init__.py b/pyipmi/interfaces/__init__.py index 37e3055..b837886 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 1984778..18859f6 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 603179a..9812502 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 bb130f2..55e4bc8 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') @@ -166,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) @@ -258,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 ' @@ -305,8 +313,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 = { @@ -319,27 +330,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 +359,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) diff --git a/pyipmi/interfaces/openipmblink.py b/pyipmi/interfaces/openipmblink.py index 6d65f9b..a4b3e02 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 616eae8..9a35caf 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') @@ -263,9 +266,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: @@ -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 @@ -499,8 +502,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 @@ -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 536f4f6..72851ee 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) @@ -444,8 +447,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/interfaces/router.py b/pyipmi/interfaces/router.py index d601faf..0d9824f 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 diff --git a/pyipmi/ipmitool.py b/pyipmi/ipmitool.py index 631dc0f..244c5fb 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 @@ -120,6 +121,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 +136,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 +152,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) @@ -198,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' + number_str = str(number) if number else 'na' + states_str = hex(states) if states else 'na' - if states: - states = hex(states) - else: - states = '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: @@ -231,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 @@ -267,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 @@ -292,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 @@ -311,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"') @@ -361,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(''' @@ -406,7 +405,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 +424,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 @@ -451,7 +455,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', @@ -603,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': @@ -669,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, @@ -1098,6 +1100,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') @@ -1266,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: @@ -1289,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 diff --git a/pyipmi/lan.py b/pyipmi/lan.py index 7764940..3943a3d 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 018ea19..9f3db97 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 0000000..6eaa9a3 --- /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/msgs/message.py b/pyipmi/msgs/message.py index d7135f4..eb0005e 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 aaf04f7..bedd5b2 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 @@ -51,7 +52,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])) diff --git a/pyipmi/msgs/sel.py b/pyipmi/msgs/sel.py index 54668c8..1d53658 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/pyipmi/picmg.py b/pyipmi/picmg.py index 18c0a35..8d589b1 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') @@ -148,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 @@ -257,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 04fe5cf..8f442af 100644 --- a/pyipmi/sdr.py +++ b/pyipmi/sdr.py @@ -24,12 +24,13 @@ 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 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 @@ -191,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 @@ -213,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) @@ -241,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 = { @@ -273,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) @@ -295,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 @@ -400,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)) @@ -504,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) @@ -514,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 @@ -545,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 @@ -573,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) @@ -583,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) @@ -601,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) @@ -612,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 @@ -629,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) @@ -653,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 @@ -669,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 6e3897d..6ed85df 100644 --- a/pyipmi/sel.py +++ b/pyipmi/sel.py @@ -21,16 +21,17 @@ 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 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 @@ -157,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 = [] @@ -184,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/sensor.py b/pyipmi/sensor.py index 30ee703..d92543a 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/session.py b/pyipmi/session.py index fa9ff3b..0dd793c 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 @@ -32,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 @@ -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) @@ -136,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 diff --git a/pyipmi/state.py b/pyipmi/state.py index 6d9e516..b055bfb 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 eb5219d..4a380fb 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) diff --git a/pyipmi/vita.py b/pyipmi/vita.py index 0babf15..522ea97 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: diff --git a/setup.cfg b/setup.cfg index c41db37..ef26e4d 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 diff --git a/tests/msgs/test_sel.py b/tests/msgs/test_sel.py index 115971c..d98347f 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)) diff --git a/tests/test_fields.py b/tests/test_fields.py index 9929100..2640f7a 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' diff --git a/tests/test_fru.py b/tests/test_fru.py index ef35dec..c41d673 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, @@ -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]) @@ -352,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)] diff --git a/tests/test_ipmitool.py b/tests/test_ipmitool.py index 9581ae0..48fdceb 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'),