2
0
mirror of https://opendev.org/x/pyghmi synced 2026-09-04 12:07:59 +00:00

refactor: support IPMI cipher 3, 8, and 17, and add corresponding tests

Change-Id: Ie643d1168ff6fc2f087ed1543e034e1d336df8b2
This commit is contained in:
Maik Rebaum
2026-08-28 10:16:19 -04:00
parent fb75c224f2
commit 7f826446c1
2 changed files with 349 additions and 23 deletions
+104 -23
View File
@@ -29,6 +29,25 @@ import pyghmi.ipmi.private.constants as constants
import pyghmi.ipmi.private.session as ipmisession
# Supported cipher suite configurations
# Maps (auth_algo, integrity_algo) -> (hashlib_func, hash_truncate_len)
# Per IPMI spec Tables 13-17, 13-18, 13-19
#
# Cipher suite 3: HMAC-SHA1 / HMAC-SHA1-96 / AES-CBC-128
# Cipher suite 8: HMAC-MD5 / HMAC-MD5-128 / AES-CBC-128
# Cipher suite 17: HMAC-SHA256 / HMAC-SHA256-128 / AES-CBC-128
CIPHER_SUITE_MAP = {
(1, 1): (hashlib.sha1, 12), # Cipher suite 3
(2, 2): (hashlib.md5, 16), # Cipher suite 8
(3, 4): (hashlib.sha256, 16), # Cipher suite 17
}
# Default cipher suite 3 for backward compatibility
DEFAULT_AUTH_ALGO = 1 # HMAC-SHA1
DEFAULT_INTEGRITY_ALGO = 1 # HMAC-SHA1-96
DEFAULT_CONF_ALGO = 1 # AES-CBC-128
class ServerSession(ipmisession.Session):
def __new__(cls, authdata, kg, clientaddr, netsocket, request, uuid,
bmc):
@@ -42,19 +61,46 @@ class ServerSession(ipmisession.Session):
clienttag = request[0]
# role = request[1]
self.clientsessionid = request[4:8]
# TODO(jbjohnso): intelligently handle integrity/auth/conf
# for now, forcibly do cipher suite 3
self.managedsessionid = os.urandom(4)
# table 13-17, 1 for now (hmac-sha1), 3 should also be supported
# table 13-18, integrity, 1 for now is hmac-sha1-96, 4 is sha256
# confidentiality: 1 is aes-cbc-128, the only one
# Parse client's requested algorithms from Open Session Request
# Per IPMI spec, the request payload after the header contains:
# Bytes 8-15: Auth payload [type=0, 0, 0, len=8, algo, 0, 0, 0]
# Bytes 16-23: Integrity payload [type=1, 0, 0, len=8, algo, 0, 0, 0]
# Bytes 24-31: Conf payload [type=2, 0, 0, len=8, algo, 0, 0, 0]
# The algorithm byte is at offset 4 within each 8-byte payload
client_auth = request[12] if len(request) > 12 else DEFAULT_AUTH_ALGO
client_integrity = request[20] if len(request) > 20 else \
DEFAULT_INTEGRITY_ALGO
client_conf = request[28] if len(request) > 28 else DEFAULT_CONF_ALGO
# Check if we support this cipher suite combination
cipher_key = (client_auth, client_integrity)
if cipher_key in CIPHER_SUITE_MAP and client_conf == 1:
# Accepted - configure session for this cipher suite
self.auth_algo = client_auth
self.integrity_algo = client_integrity
self.conf_algo = client_conf
self.currhashlib, self.currhashlen = CIPHER_SUITE_MAP[cipher_key]
status_code = 0 # Success
else:
# Unsupported cipher suite combination
# Return RMCP+ status code 0x11: No Cipher suite match
status_code = 0x11
# Set defaults for response structure (session will fail)
self.auth_algo = DEFAULT_AUTH_ALGO
self.integrity_algo = DEFAULT_INTEGRITY_ALGO
self.conf_algo = DEFAULT_CONF_ALGO
self.currhashlib = hashlib.sha1
self.currhashlen = 12
self.privlevel = 4
response = (bytearray([clienttag, 0, self.privlevel, 0])
response = (bytearray([clienttag, status_code, self.privlevel, 0])
+ self.clientsessionid + self.managedsessionid
+ bytearray([
0, 0, 0, 8, 1, 0, 0, 0, # auth
1, 0, 0, 8, 1, 0, 0, 0, # integrity
2, 0, 0, 8, 1, 0, 0, 0, # privacy
0, 0, 0, 8, self.auth_algo, 0, 0, 0, # auth
1, 0, 0, 8, self.integrity_algo, 0, 0, 0, # integrity
2, 0, 0, 8, self.conf_algo, 0, 0, 0, # privacy
]))
return response
@@ -62,8 +108,13 @@ class ServerSession(ipmisession.Session):
bmc):
# begin conversation per RMCP+ open session request
self.uuid = uuid
# Default to cipher suite 3 (SHA1) - will be overridden by
# create_open_session_response() based on client's request
self.currhashlib = hashlib.sha1
self.currhashlen = 12
self.auth_algo = DEFAULT_AUTH_ALGO
self.integrity_algo = DEFAULT_INTEGRITY_ALGO
self.conf_algo = DEFAULT_CONF_ALGO
self.rqaddr = constants.IPMI_BMC_ADDRESS
self.authdata = authdata
self.servermode = True
@@ -123,7 +174,7 @@ class ServerSession(ipmisession.Session):
if self.kg is None:
self.kg = self.kuid
authcode = hmac.new(
self.kuid, bytes(hmacdata), hashlib.sha1).digest()
self.kuid, bytes(hmacdata), self.currhashlib).digest()
# regretably, ipmi mandates the server send out an hmac first
# akin to a leak of /etc/shadow, not too worrisome if the secret
# is complex, but terrible for most likely passwords selected by
@@ -146,15 +197,15 @@ class ServerSession(ipmisession.Session):
self.sik = hmac.new(self.kg,
bytes(RmRc)
+ struct.pack("2B", self.rolem, len(self.username))
+ self.username, hashlib.sha1).digest()
self.k1 = hmac.new(self.sik, b'\x01' * 20, hashlib.sha1).digest()
self.k2 = hmac.new(self.sik, b'\x02' * 20, hashlib.sha1).digest()
+ self.username, self.currhashlib).digest()
self.k1 = hmac.new(self.sik, b'\x01' * 20, self.currhashlib).digest()
self.k2 = hmac.new(self.sik, b'\x02' * 20, self.currhashlib).digest()
self.aeskey = self.k2[0:16]
hmacdata = (self.Rc + self.clientsessionid
+ struct.pack("2B", self.rolem, len(self.username))
+ self.username)
expectedauthcode = hmac.new(self.kuid, bytes(hmacdata), hashlib.sha1
).digest()
expectedauthcode = hmac.new(
self.kuid, bytes(hmacdata), self.currhashlib).digest()
authcode = struct.pack("%dB" % len(data[8:]), *data[8:])
if expectedauthcode != authcode:
# TODO(jjohnson2): RMCP error back at invalid rakp3
@@ -192,12 +243,19 @@ class ServerSession(ipmisession.Session):
[tagvalue, statuscode, 0, 0]) + self.clientsessionid
hmacdata = self.Rm + self.managedsessionid + self.uuiddata
hmacdata = struct.pack('%dB' % len(hmacdata), *hmacdata)
authdata = hmac.new(self.sik, hmacdata, hashlib.sha1).digest()[:12]
authdata = hmac.new(self.sik, hmacdata,
self.currhashlib).digest()[:self.currhashlen]
payload += authdata
self.send_payload(payload, constants.payload_types['rakp4'],
retry=False)
self.confalgo = 'aes'
self.integrityalgo = 'sha1'
# Set integrity algorithm name based on negotiated cipher suite
if self.integrity_algo == 1:
self.integrityalgo = 'sha1'
elif self.integrity_algo == 2:
self.integrityalgo = 'md5'
else:
self.integrityalgo = 'sha256'
self.sequencenumber = 1
self.sessionid = struct.unpack(
'<I', struct.pack('4B', *self.clientsessionid))[0]
@@ -297,20 +355,43 @@ class IpmiServer(object):
def send_cipher_suites(self, myaddr, mylun, clientaddr, clientlun,
clientseq, data, sockaddr):
# the last two bytes is length of message, fixed at 14 for now
# the rest is boilerplate ipmi, follow along in ipmi spec
# 'example ipmi over lan' if desired
# Advertise supported cipher suites per IPMI Table 22-17
# Format: 0xC0 (standard suite), suite_id, auth|0x40, integ|0x40,
# conf|0x80. We support cipher suites 3, 8, and 17.
#
# Cipher suite record format (per IPMI spec):
# - Start of record: 0xC0 for standard cipher suite
# - Cipher Suite ID
# - Authentication algorithm (0x40 bit for standard algorithms)
# - Integrity algorithm (0x40 bit for standard algorithms)
# - Confidentiality algorithm (0x80 bit for standard algorithms)
cipher_data = bytearray([
0x00, # List index (channel number, 0 for this channel)
0x01, # Reserved
# Cipher suite 3: HMAC-SHA1 / HMAC-SHA1-96 / AES-CBC-128
0xc0, 0x03, 0x01, 0x41, 0x81,
# Cipher suite 8: HMAC-MD5 / HMAC-MD5-128 / AES-CBC-128
0xc0, 0x08, 0x02, 0x42, 0x81,
# Cipher suite 17: HMAC-SHA256 / HMAC-SHA256-128 / AES-CBC-128
0xc0, 0x11, 0x03, 0x44, 0x81,
])
# Calculate total message length
# rq header (3) + cipher_data + checksum (1)
msg_len = 3 + len(cipher_data) + 1
header = bytearray(
b'\x06\x00\xff\x07\x06\x00\x00\x00\x00'
b'\x00\x00\x00\x00\x00\x0e\x00')
b'\x00\x00\x00\x00\x00')
header.append(msg_len & 0xff)
header.append((msg_len >> 8) & 0xff)
# now the generic inner ipmi packet, per figure-13-4,
# ipmi lan message formats
ipmihdr = bytearray([clientaddr, clientlun | (7 << 2)])
hdrsum = ipmisession._checksum(*ipmihdr)
ipmihdr.append(hdrsum)
rq = bytearray([myaddr, mylun | clientseq << 2, 0x54])
# for now, hard code a cipher suite 3 only response
rq.extend(bytearray(b'\x00\x01\xc0\x03\x01\x41\x81'))
rq.extend(cipher_data)
hdrsum = ipmisession._checksum(*rq)
rq.append(hdrsum)
pkt = header + ipmihdr + rq
@@ -0,0 +1,245 @@
# Copyright 2026 Canonical Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Unit tests for ServerSession cipher suite negotiation."""
import hashlib
from unittest import mock
from pyghmi.ipmi.private import serversession
from pyghmi.tests.unit import base
class TestCipherSuiteMap(base.TestCase):
"""Test cipher suite configuration constants."""
def test_cipher_suite_3_exists(self):
"""Cipher suite 3 should use SHA1 with 12-byte truncation."""
self.assertIn((1, 1), serversession.CIPHER_SUITE_MAP)
hashfunc, hashlen = serversession.CIPHER_SUITE_MAP[(1, 1)]
self.assertEqual(hashfunc, hashlib.sha1)
self.assertEqual(hashlen, 12)
def test_cipher_suite_8_exists(self):
"""Cipher suite 8 should use MD5 with 16-byte truncation."""
self.assertIn((2, 2), serversession.CIPHER_SUITE_MAP)
hashfunc, hashlen = serversession.CIPHER_SUITE_MAP[(2, 2)]
self.assertEqual(hashfunc, hashlib.md5)
self.assertEqual(hashlen, 16)
def test_cipher_suite_17_exists(self):
"""Cipher suite 17 should use SHA256 with 16-byte truncation."""
self.assertIn((3, 4), serversession.CIPHER_SUITE_MAP)
hashfunc, hashlen = serversession.CIPHER_SUITE_MAP[(3, 4)]
self.assertEqual(hashfunc, hashlib.sha256)
self.assertEqual(hashlen, 16)
def test_default_cipher_suite_is_3(self):
"""Default cipher suite should be 3 (SHA1)."""
self.assertEqual(serversession.DEFAULT_AUTH_ALGO, 1)
self.assertEqual(serversession.DEFAULT_INTEGRITY_ALGO, 1)
self.assertEqual(serversession.DEFAULT_CONF_ALGO, 1)
class TestOpenSessionResponse(base.TestCase):
"""Test create_open_session_response cipher negotiation."""
def _make_open_session_request(self, tag, auth, integrity, conf):
"""Create an Open Session Request payload.
Per IPMI spec, the Open Session Request format is:
- Byte 0: Message tag
- Byte 1: Requested max privilege level
- Bytes 2-3: Reserved
- Bytes 4-7: Remote console session ID (little-endian)
- Bytes 8-15: Auth algorithm payload [type, 0, 0, len, algo, 0, 0, 0]
- Bytes 16-23: Integrity algorithm payload
- Bytes 24-31: Confidentiality algorithm payload
"""
return bytearray([
tag, # Message tag
0x00, # Max privilege
0x00, 0x00, # Reserved
0x01, 0x02, 0x03, 0x04, # Remote console session ID
0, 0, 0, 8, auth, 0, 0, 0, # Auth algorithm payload
1, 0, 0, 8, integrity, 0, 0, 0, # Integrity algorithm payload
2, 0, 0, 8, conf, 0, 0, 0, # Conf algorithm payload
])
def _create_mock_session(self):
"""Create a mock ServerSession for testing."""
session = mock.Mock(spec=serversession.ServerSession)
# Set up required attributes
session.clientsessionid = None
session.managedsessionid = None
session.auth_algo = None
session.integrity_algo = None
session.conf_algo = None
session.currhashlib = None
session.currhashlen = None
session.privlevel = None
return session
def test_accepts_cipher_suite_3(self):
"""Server should accept cipher suite 3 (SHA1/SHA1-96/AES)."""
session = self._create_mock_session()
request = self._make_open_session_request(
tag=0x01, auth=1, integrity=1, conf=1)
# Call the actual method with mock session
with mock.patch('os.urandom', return_value=b'\x00' * 4):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Check response status code (byte 1) is 0 (success)
self.assertEqual(response[1], 0)
# Check negotiated algorithms
self.assertEqual(session.auth_algo, 1)
self.assertEqual(session.integrity_algo, 1)
self.assertEqual(session.conf_algo, 1)
self.assertEqual(session.currhashlib, hashlib.sha1)
self.assertEqual(session.currhashlen, 12)
def test_accepts_cipher_suite_8(self):
"""Server should accept cipher suite 8 (MD5/MD5-128/AES)."""
session = self._create_mock_session()
request = self._make_open_session_request(
tag=0x02, auth=2, integrity=2, conf=1)
with mock.patch('os.urandom', return_value=b'\x00' * 4):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Check response status code is 0 (success)
self.assertEqual(response[1], 0)
self.assertEqual(session.auth_algo, 2)
self.assertEqual(session.integrity_algo, 2)
self.assertEqual(session.currhashlib, hashlib.md5)
self.assertEqual(session.currhashlen, 16)
def test_accepts_cipher_suite_17(self):
"""Server should accept cipher suite 17 (SHA256/SHA256-128/AES)."""
session = self._create_mock_session()
request = self._make_open_session_request(
tag=0x03, auth=3, integrity=4, conf=1)
with mock.patch('os.urandom', return_value=b'\x00' * 4):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Check response status code is 0 (success)
self.assertEqual(response[1], 0)
self.assertEqual(session.auth_algo, 3)
self.assertEqual(session.integrity_algo, 4)
self.assertEqual(session.currhashlib, hashlib.sha256)
self.assertEqual(session.currhashlen, 16)
def test_rejects_unsupported_auth_algo(self):
"""Unsupported auth algorithm should return error 0x11."""
session = self._create_mock_session()
# Auth algo 99 is not supported
request = self._make_open_session_request(
tag=0x04, auth=99, integrity=1, conf=1)
with mock.patch('os.urandom', return_value=b'\x00' * 4):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Check response status code is 0x11 (No Cipher suite match)
self.assertEqual(response[1], 0x11)
def test_rejects_unsupported_integrity_algo(self):
"""Unsupported integrity algorithm should return error 0x11."""
session = self._create_mock_session()
# Auth=1 with integrity=4 is not a valid combination
request = self._make_open_session_request(
tag=0x05, auth=1, integrity=4, conf=1)
with mock.patch('os.urandom', return_value=b'\x00' * 4):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Check response status code is 0x11
self.assertEqual(response[1], 0x11)
def test_rejects_unsupported_conf_algo(self):
"""Non-AES confidentiality should return error 0x11."""
session = self._create_mock_session()
# Conf algo 2 (XRC4) is not supported
request = self._make_open_session_request(
tag=0x06, auth=1, integrity=1, conf=2)
with mock.patch('os.urandom', return_value=b'\x00' * 4):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Check response status code is 0x11
self.assertEqual(response[1], 0x11)
def test_rejects_cipher_suite_12(self):
"""Cipher suite 12 (MD5/MD5-128 raw) is not supported."""
session = self._create_mock_session()
# Cipher suite 12: auth=2 (HMAC-MD5), integrity=3 (MD5-128 raw)
request = self._make_open_session_request(
tag=0x07, auth=2, integrity=3, conf=1)
with mock.patch('os.urandom', return_value=b'\x00' * 4):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Should be rejected since we don't support MD5-128 (raw hash)
self.assertEqual(response[1], 0x11)
def test_response_contains_correct_algorithms(self):
"""Response should echo back the negotiated algorithms."""
session = self._create_mock_session()
request = self._make_open_session_request(
tag=0x08, auth=3, integrity=4, conf=1)
with mock.patch('os.urandom', return_value=b'\xaa\xbb\xcc\xdd'):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Response format:
# Bytes 0: tag, 1: status, 2: priv, 3: reserved
# Bytes 4-7: client session ID
# Bytes 8-11: managed session ID
# Bytes 12-19: auth payload
# Bytes 20-27: integrity payload
# Bytes 28-35: conf payload
self.assertEqual(response[0], 0x08) # tag echoed
self.assertEqual(response[1], 0) # success
self.assertEqual(response[16], 3) # auth algo at offset 12+4
self.assertEqual(response[24], 4) # integrity algo at offset 20+4
self.assertEqual(response[32], 1) # conf algo at offset 28+4
def test_short_request_uses_defaults(self):
"""Short request should use default algorithms."""
session = self._create_mock_session()
# Minimal request - just tag, privilege, reserved, session ID
request = bytearray([
0x09, # Message tag
0x00, # Max privilege
0x00, 0x00, # Reserved
0x01, 0x02, 0x03, 0x04, # Remote console session ID
])
with mock.patch('os.urandom', return_value=b'\x00' * 4):
response = serversession.ServerSession.\
create_open_session_response(session, request)
# Should succeed with defaults (cipher suite 3)
self.assertEqual(response[1], 0)
self.assertEqual(session.auth_algo, 1)
self.assertEqual(session.integrity_algo, 1)