diff --git a/pyghmi/ipmi/private/serversession.py b/pyghmi/ipmi/private/serversession.py index ee23d317..16480471 100644 --- a/pyghmi/ipmi/private/serversession.py +++ b/pyghmi/ipmi/private/serversession.py @@ -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( '> 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 diff --git a/pyghmi/tests/unit/ipmi/test_serversession.py b/pyghmi/tests/unit/ipmi/test_serversession.py new file mode 100644 index 00000000..6b6f3d7d --- /dev/null +++ b/pyghmi/tests/unit/ipmi/test_serversession.py @@ -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)