fix(s7commplus): encode generated SessionKey descriptor

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

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

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."""