diff --git a/CHANGES.md b/CHANGES.md index fb4aaf69..9f31a057 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -16,6 +16,8 @@ Major release: new `s7commplus` package with S7CommPlus protocol support. * S7CommPlus symbolic data subscriptions and notification decoding (experimental) * TIA Portal XML import for SymbolTable (`SymbolTable.from_tia_xml()`) (experimental) * S7CommPlus CPU state reading and block transfer (upload/download) +* Fix the legacy SecurityKey descriptor to identify the newly generated + session key instead of an all-zero placeholder. * **Symbolic (LID-based) access for optimized DBs** (experimental): `Tag.from_access_string("8A0E0001.A", "REAL")` creates a symbolic Tag; `client.read_tag(tag)` routes to S7CommPlus LID-based access via the diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 5893ceef..3afe30c2 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -1326,7 +1326,7 @@ def _setup_session(self) -> bool: payload += encode_uint32_vlq(LegitimationId.SESSION_SETUP_LEGITIMATION) # 1830 payload += encode_uint32_vlq(ObjectId.SERVER_SESSION_VERSION) # 306 payload += encode_uint32_vlq(1) # ItemNumber for SecurityKey - payload += self._encode_security_key_struct(blob) + payload += self._encode_security_key_struct(blob, session_key) else: payload += encode_uint32_vlq(1) # ItemCount payload += encode_uint32_vlq(1) # AddressCount @@ -1524,7 +1524,7 @@ def _post_auth_legitimation(self, password: str = "") -> None: logger.info("Post-auth legitimation completed") - def _encode_security_key_struct(self, blob: bytes) -> bytes: + def _encode_security_key_struct(self, blob: bytes, session_key: bytes) -> bytes: """Encode the SecurityKey PObject struct (Struct 1800) wrapping the auth blob. Matches the wire format from TIA Portal / HarpoS7 PoC: @@ -1533,9 +1533,13 @@ def _encode_security_key_struct(self, blob: bytes) -> bytes: """ from .session_auth.utils import derive_key_id - public_key_id = derive_key_id(self._session_auth_public_key or b"\x00" * 24) - # The symmetric key ID is derived from the session key - symmetric_key_id = derive_key_id(self._session_key or b"\x00" * 24) + if not self._session_auth_public_key: + raise ValueError("SessionKey authentication requires public key material") + if not session_key: + raise ValueError("SessionKey authentication requires generated session key material") + + public_key_id = derive_key_id(self._session_auth_public_key) + symmetric_key_id = derive_key_id(session_key) # Determine key flags from family from .session_auth.blob_metadata import get_public_key_flags, get_symmetric_key_flags diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index f79bea2a..2db8a36c 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -719,6 +719,38 @@ def test_tls_v2_does_not_attempt_session_key_auth(self) -> None: assert conn._try_session_key_auth() is None assert conn._session_key is None + def test_security_key_descriptor_uses_pending_generated_key(self) -> None: + from s7commplus.session_auth.keys import KeyFamily, get_public_key + from s7commplus.session_auth.utils import derive_key_id + from s7commplus.vlq import encode_uint64_vlq + + conn = S7CommPlusConnection("127.0.0.1") + conn._session_auth_public_key = get_public_key("01:BD426B091F08731A") + conn._session_auth_family = KeyFamily.S7_1200 + generated_key = bytes(range(24)) + + assert conn._session_key is None + encoded = conn._encode_security_key_struct(bytes(180), generated_key) + symmetric_id = int.from_bytes(derive_key_id(generated_key), "little") + symmetric_descriptor = ( + encode_uint32_vlq(1804) + + bytes([0x00, DataType.STRUCT]) + + struct.pack(">I", Ids.SECURITY_KEY_ID) + + encode_uint32_vlq(1826) + + bytes([0x00, DataType.ULINT]) + + encode_uint64_vlq(symmetric_id) + ) + assert symmetric_descriptor in encoded + + def test_security_key_descriptor_rejects_missing_key_material(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + with pytest.raises(ValueError, match="public key material"): + conn._encode_security_key_struct(bytes(180), bytes(24)) + + conn._session_auth_public_key = bytes(40) + with pytest.raises(ValueError, match="generated session key material"): + conn._encode_security_key_struct(bytes(180), b"") + class TestProtocolVersionV2: """Test V2 protocol version constant."""