Revert speculative SessionKey send; keep public-key checksum capture

Wireshark's s7comm-plus dissector reveals that the value at address
1830 is a Struct(1800) named ``StructSecurityKey`` carrying a
``SessionKey`` — specifically a 180-byte ``SecurityKeyEncryptedKey``
blob with the layout:

  0-3   uint32 LE  magic = 0xFEE1DEAD
  4-7   uint32 LE  blobsize (0xB4)
  16-23 uint64 LE  symmetric_key_checksum
  32-39 uint64 LE  public_key_checksum    (what attribute 233 carries)
  48-N            encrypted_random_seed   (RSA-encrypted)
  N..   16 bytes   AES-CBC IV
  N+16..56 bytes   encrypted_challenge

So generating this requires Siemens' RSA public key (identified by the
8-byte fingerprint in attribute 233's "01:HEX" string) plus the KDF
that turns the seed into the AES-CBC key. We don't have either, so the
TIA-replay we shipped in a2111e7 was always going to fail against any
new session — the PLC cannot decrypt a static blob.

Roll back the address-1830 send. Keep the public-key checksum
extraction as a diagnostic capture (renamed from `_oms_session_uuid`
to `_public_key_checksum` to match the dissector's terminology); a
future commit can use it to dispatch to a key-aware handshake once we
have the keys, or to clearly log "this firmware needs a SessionKey we
can't generate" when we don't.

Net effect on V1-initial S7-1200 (FW < 4.5): same as before this PR —
SetupSession still fails, the unified client falls back to legacy
PUT/GET, db_read works, browse() requires CommPlus and so still
raises. PR #713 stays a draft pending the key situation.

Refs #710 #712 #713

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: gijzelaerr

s7/connection.py

@@ -47,7 +47,6 @@
 from snap7.connection import ISOTCPConnection
 from .protocol import (
     FunctionCode,
-    LegitimationId,
     Opcode,
     ProtocolVersion,
     ElementID,
@@ -63,111 +62,6 @@
 logger = logging.getLogger(__name__)
 
 
-# V2 session-setup legitimation BLOB observed in TIA Portal V19 captures
-# against an S7-1200 FW 4.2.2 (V1-initial firmware). 180 bytes; written to
-# address 1830 alongside ServerSessionVersion (306) in a single
-# SetMultiVariables. Without it the PLC drops the TCP connection silently
-# after the setup write.
-#
-# Most of the bytes are opaque — likely identity/integrity material that
-# TIA either pre-computes or signs with key material we don't have. The
-# only field we know we have to make per-session is the embedded PLC OMS
-# UUID at offset 32 (little-endian). Everything else is replayed verbatim
-# from one TIA capture as a first-pass experiment; if a later test shows
-# the PLC validates other byte ranges, we'll have to reverse those too.
-#
-# Reference capture: ``TIAPortalV19AccessibleDevices.pcapng`` frame 31
-# (attached to PR #713 / issue #710).
-_V2_SETUP_LEGITIMATION_BLOB_TEMPLATE = bytes.fromhex(
-    "ad de e1 fe b4 00 00 00 01 00 00 00 01 00 00 00"
-    "ce 9b 9f 3a 94 03 98 6b 01 01 00 00 00 00 00 00"
-    "00 00 00 00 00 00 00 00 10 01 00 00 00 00 00 00"  # OMS UUID slot at offset 32
-    "3f c8 df c2 7c 03 7f d9 99 4c ea c7 e2 b9 bb 1a"
-    "53 be 2a cd 00 49 53 3a fc 0e 43 64 49 5a c2 8e"
-    "20 ce ef 14 09 fe b4 aa e8 1c 54 08 e4 53 4c 08"
-    "2d ed 4b e6 31 ff b4 c5 1e 3b 56 ee b3 d4 1a d9"
-    "4a 83 b7 bf ac 3c ec 28 9d 5c dd f9 2e 12 59 87"
-    "f1 03 c7 08 f1 0b dc e1 e9 40 63 b5 2b 84 dc 58"
-    "9b d1 4c b6 26 a1 16 36 12 22 8e 5b 3d da c0 a0"
-    "b8 37 97 76 2f b1 f1 bb b2 b0 fb 44 f4 6a 50 9e"
-    "03 42 d5 6f"
-)
-_V2_SETUP_LEGITIMATION_OMS_UUID_OFFSET = 32
-
-
-def _build_v2_session_setup_legitimation_value(oms_session_uuid: bytes) -> bytes:
-    """Build the typed value written to address 1830 during session setup.
-
-    The PObject tree shape is:
-
-    Struct(1800)
-        1801: UDInt(0)
-        1802: USInt(0)
-        1803: Struct(1825)
-            1826: ULInt (opaque, replayed)
-            1827: UDInt(272)
-            1828: UDInt(0)
-        1804: Struct(1825)
-            1826: ULInt (opaque, replayed)
-            1827: UDInt(65793)
-            1828: UDInt(0)
-        1805: Blob(180 bytes — see template above; OMS UUID patched in LE)
-
-    Args:
-        oms_session_uuid: 8-byte OMS session UUID parsed from the
-            CreateObject response (attribute 233 ``"01:HEX"`` string).
-
-    Returns:
-        Encoded typed value (flags + datatype + struct body), ready to
-        append to a SetMultiVariables item.
-    """
-    if len(oms_session_uuid) != 8:
-        raise ValueError(f"OMS session UUID must be 8 bytes, got {len(oms_session_uuid)}")
-
-    blob = bytearray(_V2_SETUP_LEGITIMATION_BLOB_TEMPLATE)
-    blob[_V2_SETUP_LEGITIMATION_OMS_UUID_OFFSET : _V2_SETUP_LEGITIMATION_OMS_UUID_OFFSET + 8] = oms_session_uuid[::-1]
-
-    # PValue helpers. Each returns a complete typed value: flags + datatype
-    # + value bytes. _elem prepends the element-key VLQ for placement
-    # inside a struct.
-    def _elem(elem_id: int, typed_value: bytes) -> bytes:
-        return encode_uint32_vlq(elem_id) + typed_value
-
-    def _udint(value: int) -> bytes:
-        return bytes([0x00, DataType.UDINT]) + encode_uint32_vlq(value)
-
-    def _usint(value: int) -> bytes:
-        return bytes([0x00, DataType.USINT, value & 0xFF])
-
-    def _ulint_raw(raw_vlq: bytes) -> bytes:
-        return bytes([0x00, DataType.ULINT]) + raw_vlq
-
-    def _struct(struct_id: int, body: bytes) -> bytes:
-        # Struct PValue: flags(0) + STRUCT + 4-byte ID + body + terminator(0)
-        return bytes([0x00, DataType.STRUCT]) + struct.pack(">I", struct_id) + body + bytes([0x00])
-
-    def _blob(data: bytes) -> bytes:
-        return bytes([0x00, DataType.BLOB]) + encode_uint32_vlq(0) + encode_uint32_vlq(len(data)) + data
-
-    # Opaque ULInt VLQs from the TIA reference capture. Each is a 9-byte
-    # VLQ. Treated as opaque payload rather than decoded numeric values
-    # since we don't know what they encode.
-    opaque_ulint_1 = bytes.fromhex("de d0 cd b0 c8 fc 90 f3 1a")
-    opaque_ulint_2 = bytes.fromhex("b5 e6 80 b9 a1 ea bf 9b ce")
-
-    inner_1803 = _elem(1826, _ulint_raw(opaque_ulint_1)) + _elem(1827, _udint(272)) + _elem(1828, _udint(0))
-    inner_1804 = _elem(1826, _ulint_raw(opaque_ulint_2)) + _elem(1827, _udint(65793)) + _elem(1828, _udint(0))
-
-    outer_body = b""
-    outer_body += _elem(1801, _udint(0))
-    outer_body += _elem(1802, _usint(0))
-    outer_body += _elem(1803, _struct(1825, inner_1803))
-    outer_body += _elem(1804, _struct(1825, inner_1804))
-    outer_body += _elem(1805, _blob(bytes(blob)))
-
-    return _struct(1800, outer_body)
-
-
 def _strip_paom_string_in_session_version(struct_bytes: bytes) -> bytes:
     """Replace element 319 in a captured ServerSessionVersion struct with an empty WString.
 
@@ -242,11 +136,13 @@ def __init__(
         self._server_session_version: Optional[int] = None
         self._server_session_version_raw: Optional[bytes] = None
         self._session_setup_ok: bool = False
-        # PLC-provided 8-byte OMS session UUID (parsed from the
+        # PLC-provided 8-byte public-key checksum (parsed from the
         # ObjectVariableTypeName "01:HEX" attribute in the CreateObject
-        # response). Required to build the V2 session-setup legitimation
-        # value on V1-initial S7-1200 firmware.
-        self._oms_session_uuid: Optional[bytes] = None
+        # response). It identifies which Siemens RSA key the PLC expects
+        # for the V2 SessionKey handshake. Captured for diagnostics and
+        # future use; we don't generate the SessionKey blob ourselves
+        # because we don't have the matching public key.
+        self._public_key_checksum: Optional[bytes] = None
 
         # V2+ IntegrityId tracking
         self._integrity_id_read: int = 0
@@ -934,9 +830,12 @@ def _parse_create_object_response(self, payload: bytes) -> None:
 
                 elif attr_id == 233:
                     # ObjectVariableTypeName: a WString shaped "01:HEX",
-                    # where HEX is the PLC's OMS session UUID encoded as
-                    # 16 hex characters. Captured for the V2 session-setup
-                    # legitimation value (address 1830).
+                    # where HEX is the 8-byte fingerprint of the Siemens
+                    # RSA public key the PLC expects for the V2 SessionKey
+                    # handshake (Wireshark s7comm-plus dissector calls
+                    # this "publickeychecksum"). Captured for diagnostics
+                    # and so future code can decide whether the matching
+                    # public key is available.
                     if offset + 2 > len(payload):
                         break
                     _flags = payload[offset]
@@ -950,8 +849,8 @@ def _parse_create_object_response(self, payload: bytes) -> None:
                         try:
                             text = raw.decode("utf-8")
                             if len(text) == 19 and text[2] == ":":
-                                self._oms_session_uuid = bytes.fromhex(text[3:])
-                                logger.info(f"OMS session UUID captured: {text}")
+                                self._public_key_checksum = bytes.fromhex(text[3:])
+                                logger.info(f"Public key checksum captured: {text}")
                         except (UnicodeDecodeError, ValueError):
                             logger.debug(f"Unparseable ObjectVariableTypeName: {raw!r}")
                     else:
@@ -1070,7 +969,7 @@ def _skip_typed_value(self, data: bytes, offset: int, datatype: int, flags: int)
             return offset
 
     def _setup_session(self) -> bool:
-        """Send V2 SetMultiVariables to complete the session handshake.
+        """Send V2 SetMultiVariables to echo ServerSessionVersion back to the PLC.
 
         Always uses V2 framing (`72 02 ...`), transport flags 0x34, and no
         IntegrityId — these are the established session-setup conventions
@@ -1080,16 +979,16 @@ def _setup_session(self) -> bool:
         Without this step, the PLC rejects all subsequent data operations
         with ERROR2 (0x05A9).
 
-        Two-item form (V1-initial S7-1200, FW < 4.07): writes both
-        ``SESSION_SETUP_LEGITIMATION`` (1830) and ``SERVER_SESSION_VERSION``
-        (306) in one SetMultiVariables. TIA Portal V19 does this on FW 4.2.2
-        captures; without the legitimation item the PLC closes the TCP
-        connection silently right after the setup write. We use the two-item
-        form whenever the PLC sent us its OMS session UUID in the
-        CreateObject response (a V1-initial-firmware behaviour).
-
-        One-item form (everything else): just the ``SERVER_SESSION_VERSION``
-        echo, matching the C# driver's ``SetSessionSetupData``.
+        Note on V1-initial S7-1200 (FW < 4.5): TIA Portal V19 also writes a
+        ``SessionKey`` value at address 1830 in the same frame, carrying an
+        RSA-encrypted random seed and an AES-CBC-encrypted challenge.
+        Without that key, the PLC drops the connection right after this
+        write — so CommPlus data ops (incl. ``browse()``) never come up on
+        V1-initial firmware. The unified client transparently falls back to
+        legacy PUT/GET in that case. See PR #713 / issue #710 / issue #712
+        for the diagnosis (Wireshark s7comm-plus dissector confirmed
+        Struct(1800) is ``StructSecurityKey``; we don't have Siemens'
+        public keys to generate the encrypted seed).
 
         Returns:
             True if session setup succeeded (return_value == 0).
@@ -1112,36 +1011,23 @@ def _setup_session(self) -> bool:
             0x34,
         )
 
+        payload = bytearray()
+        payload += struct.pack(">I", self._session_id)  # InObjectId
+        payload += encode_uint32_vlq(1)  # ItemCount
+        payload += encode_uint32_vlq(1)  # AddressCount
+        payload += encode_uint32_vlq(ObjectId.SERVER_SESSION_VERSION)  # Address: 306
+        payload += encode_uint32_vlq(1)  # ItemNumber
+
         if self._server_session_version_raw is not None:
             # Echo the Struct(314) value, but strip element 319 (the device
             # PAOM string) to empty. TIA Portal does this; writing the PLC's
             # own identity back appears to be rejected — the V1-initial
             # S7-1200 silently drops the connection if we don't strip it.
-            session_version_value = _strip_paom_string_in_session_version(self._server_session_version_raw)
+            payload += _strip_paom_string_in_session_version(self._server_session_version_raw)
         else:
             # Test/emulator path: scalar UDInt fallback.
-            session_version_value = bytes([0x00, DataType.UDINT]) + encode_uint32_vlq(self._server_session_version or 0)
-
-        oms_uuid = self._oms_session_uuid
-
-        payload = bytearray()
-        payload += struct.pack(">I", self._session_id)  # InObjectId
-
-        if oms_uuid is not None:
-            payload += encode_uint32_vlq(2)  # ItemCount
-            payload += encode_uint32_vlq(2)  # AddressCount
-            payload += encode_uint32_vlq(LegitimationId.SESSION_SETUP_LEGITIMATION)  # 1830
-            payload += encode_uint32_vlq(ObjectId.SERVER_SESSION_VERSION)  # 306
-            payload += encode_uint32_vlq(1)  # ItemNumber 1
-            payload += _build_v2_session_setup_legitimation_value(oms_uuid)
-            payload += encode_uint32_vlq(2)  # ItemNumber 2
-            payload += session_version_value
-        else:
-            payload += encode_uint32_vlq(1)  # ItemCount
-            payload += encode_uint32_vlq(1)  # AddressCount
-            payload += encode_uint32_vlq(ObjectId.SERVER_SESSION_VERSION)  # 306
-            payload += encode_uint32_vlq(1)  # ItemNumber
-            payload += session_version_value
+            payload += bytes([0x00, DataType.UDINT])
+            payload += encode_uint32_vlq(self._server_session_version or 0)
 
         payload += bytes([0x00])  # Fill byte
         payload += encode_object_qualifier()

tests/test_s7_unit.py

@@ -13,7 +13,6 @@
 from s7.codec import encode_pvalue_blob
 from s7.connection import (
     S7CommPlusConnection,
-    _build_v2_session_setup_legitimation_value,
     _element_size,
     _strip_paom_string_in_session_version,
 )
@@ -451,52 +450,11 @@ def test_strip_paom_string_no_match_returns_unchanged(self) -> None:
         nopaom = bytes.fromhex("00170000013a823b0004840000")
         assert _strip_paom_string_in_session_version(nopaom) == nopaom
 
-    def test_build_v2_session_setup_legitimation_value(self) -> None:
-        # The PObject tree we send at address 1830 must be byte-identical
-        # to the value TIA Portal V19 sent on a real V1-initial S7-1200,
-        # with the OMS UUID patched in. Reference: frame 31 of
-        # ``TIAPortalV19AccessibleDevices.pcapng`` (PR #713 / issue #710).
-        oms_uuid = bytes.fromhex("BD426B091F08731A")
-        got = _build_v2_session_setup_legitimation_value(oms_uuid)
-        expected_hex = (
-            "00 17 00 00 07 08"
-            "8e 09 00 04 00"
-            "8e 0a 00 02 00"
-            "8e 0b 00 17 00 00 07 21 8e 22 00 05 de d0 cd b0 c8 fc 90 f3 1a 8e 23 00 04 82 10 8e 24 00 04 00 00"
-            "8e 0c 00 17 00 00 07 21 8e 22 00 05 b5 e6 80 b9 a1 ea bf 9b ce 8e 23 00 04 84 82 01 8e 24 00 04 00 00"
-            "8e 0d 00 14 00 81 34"
-            "ad de e1 fe b4 00 00 00 01 00 00 00 01 00 00 00"
-            "ce 9b 9f 3a 94 03 98 6b 01 01 00 00 00 00 00 00"
-            "1a 73 08 1f 09 6b 42 bd 10 01 00 00 00 00 00 00"
-            "3f c8 df c2 7c 03 7f d9 99 4c ea c7 e2 b9 bb 1a"
-            "53 be 2a cd 00 49 53 3a fc 0e 43 64 49 5a c2 8e"
-            "20 ce ef 14 09 fe b4 aa e8 1c 54 08 e4 53 4c 08"
-            "2d ed 4b e6 31 ff b4 c5 1e 3b 56 ee b3 d4 1a d9"
-            "4a 83 b7 bf ac 3c ec 28 9d 5c dd f9 2e 12 59 87"
-            "f1 03 c7 08 f1 0b dc e1 e9 40 63 b5 2b 84 dc 58"
-            "9b d1 4c b6 26 a1 16 36 12 22 8e 5b 3d da c0 a0"
-            "b8 37 97 76 2f b1 f1 bb b2 b0 fb 44 f4 6a 50 9e"
-            "03 42 d5 6f"
-            "00"
-        )
-        assert got == bytes.fromhex(expected_hex)
-
-    def test_build_v2_session_setup_legitimation_value_patches_oms_uuid(self) -> None:
-        # A different OMS UUID should appear in the blob in little-endian
-        # byte order at the documented offset, leaving the rest of the
-        # template untouched.
-        oms_uuid = bytes.fromhex("0102030405060708")
-        got = _build_v2_session_setup_legitimation_value(oms_uuid)
-        # Locate the BLOB length marker (`81 34`) and check the next 32 bytes
-        # land us at the OMS UUID slot.
-        idx = got.index(bytes.fromhex("81 34"))
-        oms_le = got[idx + 2 + 32 : idx + 2 + 40]
-        assert oms_le == bytes.fromhex("0807060504030201")
-
-    def test_parse_oms_session_uuid(self) -> None:
-        # CreateObject response carries the PLC's OMS session UUID as a
-        # WString shaped "01:HEX". The parser captures it as raw bytes for
-        # the V2 session-setup legitimation value.
+    def test_parse_public_key_checksum(self) -> None:
+        # CreateObject response carries the Siemens RSA public-key
+        # checksum as a WString shaped "01:HEX" in attribute 233
+        # (ObjectVariableTypeName). The parser captures it as raw bytes
+        # for diagnostics.
         conn = S7CommPlusConnection("127.0.0.1")
         payload = bytearray()
         payload += bytes([ElementID.ATTRIBUTE])
@@ -506,7 +464,7 @@ def test_parse_oms_session_uuid(self) -> None:
         payload += encode_uint32_vlq(len(text))
         payload += text.encode("utf-8")
         conn._parse_create_object_response(bytes(payload))
-        assert conn._oms_session_uuid == bytes.fromhex("BD426B091F08731A")
+        assert conn._public_key_checksum == bytes.fromhex("BD426B091F08731A")
 
     def test_parse_struct_version(self) -> None:
         # Real S7-1200/1500 PLCs send ServerSessionVersion as Struct(314)