# Copyright (c) 2018 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 import socket import struct import hashlib import random import threading from array import array from queue import Queue from .. import Target from ..session import Session from ..msgs import (create_message, create_request_by_name, encode_message, decode_message, constants) from ..messaging import ChannelAuthenticationCapabilities from ..errors import DecodingError, NotSupportedError, RetryError from ..logger import log from ..interfaces.ipmb import (IpmbHeaderReq, encode_ipmb_msg, encode_bridged_message, decode_bridged_message, rx_filter) from ..utils import (check_completion_code, check_rsp_completion_code, py3_array_tobytes) CLASS_NORMAL_MSG = 0x00 CLASS_ACK_MSG = 0x80 RMCP_CLASS_ASF = 0x06 RMCP_CLASS_IPMI = 0x07 RMCP_CLASS_OEM = 0x08 def call_repeatedly(interval, func, *args): stopped = threading.Event() def loop(): # the first call is in `interval` secs while not stopped.wait(interval): try: func(*args) except socket.timeout: pass t = threading.Thread(target=loop) t.daemon = True t.start() return stopped.set class RmcpMsg(object): RMCP_HEADER_FORMAT = '!BxBB' ASF_RMCP_V_1_0 = 6 version = None seq_number = None class_of_msg = None def __init__(self, class_of_msg=None): if class_of_msg is not None: self.class_of_msg = class_of_msg def pack(self, sdu, seq_number): pdu = struct.pack(self.RMCP_HEADER_FORMAT, self.ASF_RMCP_V_1_0, seq_number, self.class_of_msg) if sdu is not None: pdu += sdu return pdu def unpack(self, pdu): header_len = struct.calcsize(self.RMCP_HEADER_FORMAT) header = pdu[:header_len] (self.version, self.seq_number, self.class_of_msg) = \ struct.unpack(self.RMCP_HEADER_FORMAT, header) sdu = pdu[header_len:] if self.version != self.ASF_RMCP_V_1_0: raise DecodingError('invalid RMCP version field') return sdu class AsfMsg(object): ASF_HEADER_FORMAT = '!IBBxB' ASF_TYPE_PRESENCE_PONG = 0x40 ASF_TYPE_PRESENCE_PING = 0x80 asf_type = 0 def __init__(self): self.iana_enterprise_number = 4542 self.tag = 0 self.data = None self.sdu = None def pack(self): if self.data: data_len = len(self.data) else: data_len = 0 pdu = struct.pack(self.ASF_HEADER_FORMAT, self.iana_enterprise_number, self.asf_type, self.tag, data_len) if self.data: pdu += self.data return pdu def unpack(self, sdu): self.sdu = sdu header_len = struct.calcsize(self.ASF_HEADER_FORMAT) header = sdu[:header_len] (self.iana_enterprise_number, self.asf_type, self.tag, data_len) = \ struct.unpack(self.ASF_HEADER_FORMAT, header) if len(sdu) < header_len + data_len: raise DecodingError('short SDU') elif len(sdu) > header_len + data_len: raise DecodingError('SDU has extra bytes') if data_len != 0: self.data = sdu[header_len:header_len + data_len] else: self.data = None if hasattr(self, 'check_header'): self.check_header() def __str__(self): if self.data: return ' '.join('%02x' % b for b in array('B', self.data)) if self.sdu: return ' '.join('%02x' % b for b in array('B', self.sdu)) return '' @staticmethod def from_data(sdu): asf = AsfMsg() asf.unpack(sdu) try: cls = { AsfMsg().ASF_TYPE_PRESENCE_PING: AsfPing, AsfMsg().ASF_TYPE_PRESENCE_PONG: AsfPong, }[asf.asf_type] except KeyError: raise DecodingError('Unsupported ASF type(0x%02x)' % asf.asf_type) instance = cls() instance.unpack(sdu) return instance class AsfPing(AsfMsg): def __init__(self): AsfMsg.__init__(self) self.asf_type = self.ASF_TYPE_PRESENCE_PING def check_header(self): if self.asf_type != self.ASF_TYPE_PRESENCE_PING: raise DecodingError('type does not match') if self.data: raise DecodingError('Data length is not zero') def __str__(self): return 'ping: ' + super(AsfMsg, self).__str__() class AsfPong(AsfMsg): DATA_FORMAT = '!IIBB6x' def __init__(self): self.asf_type = self.ASF_TYPE_PRESENCE_PONG self.oem_iana_enterprise_number = 4542 self.oem_defined = 0 self.supported_entities = 0 self.supported_interactions = 0 def pack(self): pdu = struct.pack(self.DATA_FORMAT, self.oem_iana_enterprise_number, self.oem_defined, self.supported_entities, self.supported_interactions) return pdu def unpack(self, sdu): AsfMsg.unpack(self, sdu) # header_len = struct.calcsize(self.ASF_HEADER_FORMAT) (self.oem_iana_enterprise_number, self.oem_defined, self.supported_entities, self.supported_interactions) =\ struct.unpack(self.DATA_FORMAT, self.data) self.check_data() def check_data(self): if self.oem_iana_enterprise_number == 4542 and self.oem_defined != 0: raise DecodingError('SDU malformed') if self.supported_interactions != 0: raise DecodingError('SDU malformed') def check_header(self): if self.asf_type != self.ASF_TYPE_PRESENCE_PONG: raise DecodingError('type does not match') if len(self.data) != struct.calcsize(self.DATA_FORMAT): raise DecodingError('Data length mismatch') class IpmiMsg(object): HEADER_FORMAT_NO_AUTH = '!BIIB' HEADER_FORMAT_AUTH = '!BII16BB' def __init__(self, session=None, ignore_sdu_length=False): self.session = session self.ignore_sdu_length = ignore_sdu_length def _pack_session_id(self): if self.session is not None: session_id = self.session.sid else: session_id = 0 return struct.unpack("I", session_id))[0] def _pack_sequence_number(self): if self.session is not None: seq = self.session.sequence_number else: seq = 0 return struct.unpack("I", seq))[0] def _padd_password(self): """Padd the password. 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) return password.ljust(16, b'\x00') def _pack_auth_code_straight(self): """Return the auth code as bytestring.""" return self._padd_password() def _pack_auth_code_md5(self, sdu): auth_code = struct.pack('>16s I %ds I 16s' % len(sdu), self._pack_auth_code_straight(), self._pack_session_id(), sdu, self._pack_sequence_number(), self._pack_auth_code_straight()) return hashlib.md5(auth_code).digest() def pack(self, sdu): if sdu is not None: data_len = len(sdu) else: data_len = 0 if self.session is not None: auth_type = self.session.auth_type if self.session.activated: self.session.increment_sequence_number() else: auth_type = Session.AUTH_TYPE_NONE pdu = struct.pack('!BII', auth_type, self._pack_sequence_number(), self._pack_session_id()) if auth_type == Session.AUTH_TYPE_NONE: pass 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) else: raise NotSupportedError('authentication type %s' % auth_type) pdu += py3_array_tobytes(array('B', [data_len])) if sdu is not None: pdu += sdu return pdu def unpack(self, pdu): auth_type = array('B', pdu)[0] if auth_type != 0: header_len = struct.calcsize(self.HEADER_FORMAT_AUTH) header = pdu[:header_len] # TBD .. find a way to do this better self.auth_type = array('B', pdu)[0] (self.sequence_number,) = struct.unpack('!I', pdu[1:5]) (self.session_id,) = struct.unpack('!I', pdu[5:9]) self.auth_code = [a for a in struct.unpack('!16B', pdu[9:25])] data_len = array('B', pdu)[25] else: header_len = struct.calcsize(self.HEADER_FORMAT_NO_AUTH) header = pdu[:header_len] (self.auth_type, self.sequence_number, self.session_id, data_len) = struct.unpack(self.HEADER_FORMAT_NO_AUTH, header) if not self.ignore_sdu_length: if len(pdu) < header_len + data_len: raise DecodingError('short SDU') elif len(pdu) > header_len + data_len: raise DecodingError( 'SDU has extra bytes ({:d},{:d},{:d} )'.format( len(pdu), header_len, data_len)) if hasattr(self, 'check_header'): self.check_header() if not self.ignore_sdu_length: if data_len != 0: sdu = pdu[header_len:header_len + data_len] else: sdu = None else: try: sdu = pdu[header_len:] sdu = None if sdu == b'' else sdu except IndexError: sdu = None return sdu def check_data(self): pass def check_header(self): pass class Rmcp(object): NAME = 'rmcp' _session = None def __init__(self, slave_address=0x81, host_target_address=0x20, keep_alive_interval=1, max_retries=0, quirks_cfg=dict()): """Native RMCP interface constructor Parameter `quirks_cfg`: a dict of additional configuration parameters for the RMCP object. Supported keys/values are : - `rmcp_ignore_sdu_length: bool` whether or not to verify the SDU length of a received IPMI message. If absent or `False` (default), an exception will be raised if there is a mismatch between the received SDU length and the value stored in `payload_length` field of the PDU header. If `True`, ignore the header field `payload_length` and unpack the PDU anyway. Example: interfaces.create_interface( interface="rmcp", quirks_cfg={'rmcp_ignore_sdu_length': True} ) - `rmcp_ignore_rq_seq: bool` wheater or not to verify the req_seq field for rx_filter. The default value is `False`. Example: interfaces.create_interface( interface="rmcp", quirks_cfg={'rmcp_ignore_rq_seq': True} ) """ self.host = None self.port = None self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.seq_number = 0xff self.slave_address = slave_address self.host_target = Target(host_target_address) self.max_retries = max_retries self.set_timeout(2.0) self.next_sequence_number = 0 self.keep_alive_interval = keep_alive_interval self._stop_keep_alive = None self._q = Queue() self.transaction_lock = threading.Lock() self.quirks_cfg = quirks_cfg self.ignore_sdu_length = quirks_cfg.get('rmcp_ignore_sdu_length', False) self.ignore_rq_seq = quirks_cfg.get('rmcp_ignore_rq_seq', False) def _send_rmcp_msg(self, sdu, class_of_msg): rmcp = RmcpMsg(class_of_msg) pdu = rmcp.pack(sdu, self.seq_number) self._sock.sendto(pdu, (self.host, self.port)) if self.seq_number != 255: self.seq_number = (self.seq_number + 1) % 254 def _receive_rmcp_msg(self): (pdu, _) = self._sock.recvfrom(4096) rmcp = RmcpMsg() sdu = rmcp.unpack(pdu) return (rmcp.seq_number, rmcp.class_of_msg, sdu) def set_timeout(self, timeout): self._sock.settimeout(timeout) def _send_ipmi_msg(self, data): log().debug('IPMI TX: {:s}'.format( ' '.join('%02x' % b for b in array('B', data)))) ipmi = IpmiMsg(self._session) tx_data = ipmi.pack(data) self._send_rmcp_msg(tx_data, RMCP_CLASS_IPMI) def _receive_ipmi_msg(self, ignore_sdu_length=False): (_, 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) log().debug('IPMI RX: {:s}'.format( ' '.join('%02x' % b for b in array('B', data)))) return data def _send_asf_msg(self, msg): log().debug('ASF TX: msg') self._send_rmcp_msg(msg.pack(), RMCP_CLASS_ASF) def _receive_asf_msg(self, cls): (_, class_of_msg, data) = self._receive_rmcp_msg() log().debug('ASF RX: msg') if class_of_msg != RMCP_CLASS_ASF: raise DecodingError('invalid class field in ASF message') msg = cls() msg.unpack(data) return msg def ping(self): ping = AsfPing() self._send_asf_msg(ping) self._receive_asf_msg(AsfPong) def _get_channel_auth_cap(self, session): CHANNEL_NUMBER_FOR_THIS = 0xe # get channel auth cap req = create_request_by_name('GetChannelAuthenticationCapabilities') req.target = self.host_target req.channel.number = CHANNEL_NUMBER_FOR_THIS req.privilege_level.requested = session.priv_level rsp = self.send_and_receive(req) check_completion_code(rsp.completion_code) caps = ChannelAuthenticationCapabilities(rsp) return caps def _get_session_challenge(self, session): # get session challenge 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') rsp = self.send_and_receive(req) check_rsp_completion_code(rsp) return rsp def _activate_session(self, session, challenge): # activate session req = create_request_by_name('ActivateSession') req.target = self.host_target 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.initial_outbound_sequence_number = random.randrange(1, 0xffffffff) rsp = self.send_and_receive(req) check_rsp_completion_code(rsp) return rsp def _set_session_privilege_level(self, level): req = create_request_by_name('SetSessionPrivilegeLevel') req.target = self.host_target req.privilege_level.requested = level rsp = self.send_and_receive(req) check_rsp_completion_code(rsp) return rsp def _get_device_id(self): req = create_request_by_name('GetDeviceId') req.target = self.host_target rsp = self.send_and_receive(req) check_completion_code(rsp.completion_code) def establish_session(self, session): self._session = None self.host = session._rmcp_host self.port = session._rmcp_port # 0 - Ping self.ping() # 1 - Get Channel Authentication Capabilities log().debug('Get Channel Authentication Capabilities') caps = self._get_channel_auth_cap(session) log().debug('%s' % caps) # 2 - Get Session Challenge log().debug('Get Session Challenge') session.auth_type = caps.get_max_auth_type() rsp = self._get_session_challenge(session) session_challenge = rsp.challenge_string session.sid = rsp.temporary_session_id self._session = session # 3 - Activate Session log().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 log().debug('Set Session Privilege Level') # 4 - Set Session Privilege Level self._set_session_privilege_level(session.priv_level) log().debug('Session opened') if self.keep_alive_interval: self._stop_keep_alive = call_repeatedly( self.keep_alive_interval, self._get_device_id) def close_session(self): if self._stop_keep_alive: self._stop_keep_alive() if self._session.activated is False: log().debug('Session already closed') return log().debug('Close Session %s' % self._session) req = create_request_by_name('CloseSession') req.target = self.host_target req.session_id = self._session.sid rsp = self.send_and_receive(req) check_completion_code(rsp.completion_code) self._session.activated = False # self._q.join() def _inc_sequence_number(self): self.next_sequence_number = (self.next_sequence_number + 1) % 64 def _send_and_receive(self, target, lun, netfn, cmdid, payload): """Send and receive data using RMCP interface. target: lun: netfn: cmdid: raw_bytes: IPMI message payload as bytestring Returns the received data as array. """ self._inc_sequence_number() header = IpmbHeaderReq() header.netfn = netfn header.rs_lun = lun header.rs_sa = target.ipmb_address header.rq_seq = self.next_sequence_number header.rq_lun = 0 header.rq_sa = self.slave_address header.cmdid = cmdid # Bridge message if target.routing: tx_data = encode_bridged_message(target.routing, header, payload, self.next_sequence_number) else: tx_data = encode_ipmb_msg(header, payload) with self.transaction_lock: retry = 0 while retry <= self.max_retries: try: self._send_ipmi_msg(tx_data) received = False received_retry = 0 while received is False and received_retry <= self.max_retries: if not self._q.empty(): rx_data = self._q.get() else: rx_data = self._receive_ipmi_msg(self.ignore_sdu_length) if array('B', rx_data)[5] == constants.CMDID_SEND_MESSAGE: rx_data = decode_bridged_message(rx_data) if not rx_data: # the forwarded reply is expected in the next packet # so we do not increment the retry counter as # it's not really a retry continue received = rx_filter(header, rx_data, rq_seq=not self.ignore_rq_seq) if not received: self._q.put(rx_data) received_retry += 1 if not received: raise RetryError("Max retry while checking received" "data against request header for rmcp" f"host {self.host}") break except socket.timeout: retry += 1 if retry > self.max_retries: raise RetryError("Max retry while sending and/or receiving ipmi" f"message for rmcp host {self.host}") return rx_data[6:-1] def send_and_receive_raw(self, target, lun, netfn, raw_bytes): """Interface function to send and receive raw message. target: IPMI target lun: logical unit number netfn: network function raw_bytes: RAW bytes as bytestring Returns the IPMI message response bytestring. """ return self._send_and_receive(target=target, lun=lun, netfn=netfn, cmdid=array('B', raw_bytes)[0], payload=raw_bytes[1:]) def send_and_receive(self, req): """Interface function to send and receive an IPMI message. target: IPMI target req: IPMI message request Returns the IPMI message response. """ rx_data = self._send_and_receive(target=req.target, lun=req.lun, netfn=req.netfn, cmdid=req.cmdid, payload=encode_message(req)) rsp = create_message(req.netfn + 1, req.cmdid, req.group_extension) decode_message(rsp, rx_data) return rsp