SessionKey handshake for V1-initial S7-1200 PLCs

.pre-commit-config.yaml

@@ -3,7 +3,9 @@ repos:
     rev: v6.0.0
     hooks:
       - id: trailing-whitespace
+        exclude: '\.bin$'
       - id: end-of-file-fixer
+        exclude: '\.bin$'
       - id: check-ast
       - id: check-shebang-scripts-are-executable
       - id: check-json

CHANGES.md

@@ -44,6 +44,7 @@ Major release: new `s7commplus` package with S7CommPlus protocol support.
 
 ### Thanks
 
+* [@bonk-dev](https://github.com/bonk-dev) — [HarpoS7](https://github.com/bonk-dev/HarpoS7): the session authentication implementation in `s7commplus/session_auth/` is a Python port of HarpoS7 (MIT license, see `s7commplus/session_auth/LICENSE-HarpoS7`)
 * [@hs2bws-hash](https://github.com/hs2bws-hash) — extensive real PLC testing of Partner BSend/BRecv (#668)
 * [@QuakeString](https://github.com/QuakeString) — read optimizer inspiration via python-snap7-optimized fork
 

doc/development.rst

@@ -81,6 +81,8 @@ Special thanks to:
 * Davide Nardella for creating snap7
 * Thomas Hergenhahn for his libnodave
 * Thomas W for his S7comm wireshark plugin
+* `bonk-dev <https://github.com/bonk-dev>`_ for `HarpoS7 <https://github.com/bonk-dev/HarpoS7>`_ — the S7CommPlus session authentication (``s7commplus/session_auth/``) is a Python port of HarpoS7 (MIT)
+* Thomas W (thomas-v2) for `S7CommPlusDriver <https://github.com/thomas-v2/S7CommPlusDriver>`_ — the zlib preset dictionaries and S7CommPlus protocol reference (LGPL-3.0)
 * `Fabian Beitler <https://github.com/swamper123>`_
 * `Nikteliy <https://github.com/nikteliy>`_
 * `Lautaro Nahuel Dapino <https://github.com/lautarodapin>`_

pyproject.toml

@@ -41,7 +41,7 @@ discovery = ["pnio-dcp"]
 
 [tool.setuptools.package-data]
 snap7 = ["py.typed"]
-s7commplus = ["py.typed"]
+s7commplus = ["py.typed", "session_auth/**/*.bin"]
 
 [tool.setuptools.packages.find]
 include = ["snap7*", "s7*", "s7commplus*"]

s7commplus/client.py

@@ -93,9 +93,10 @@ def connect(
             tls_cert=tls_cert,
             tls_key=tls_key,
             tls_ca=tls_ca,
+            password=password or "",
         )
 
-        if password is not None and self._connection.tls_active:
+        if password is not None and self._connection.tls_active and not self._connection.requires_substreamed:
             logger.info("Performing PLC legitimation (password authentication)")
             self._connection.authenticate(password)
 
@@ -119,6 +120,9 @@ def db_read(self, db_number: int, start: int, size: int) -> bytes:
         if self._connection is None:
             raise RuntimeError("Not connected")
 
+        if self._connection.requires_substreamed:
+            return self._db_read_substreamed(db_number, start, size)
+
         payload = _build_read_payload([(db_number, start, size)])
         response = self._connection.send_request(FunctionCode.GET_MULTI_VARIABLES, payload)
         results = _parse_read_response(response)
@@ -128,6 +132,15 @@ def db_read(self, db_number: int, start: int, size: int) -> bytes:
             raise RuntimeError("Read failed: PLC returned error for item")
         return results[0]
 
+    def _db_read_substreamed(self, db_number: int, start: int, size: int) -> bytes:
+        assert self._connection is not None
+        access_area = Ids.DB_ACCESS_AREA_BASE + (db_number & 0xFFFF)
+        payload = _build_substreamed_read_payload(
+            self._connection.session_id, access_area, Ids.DB_VALUE_ACTUAL, [start + 1, size]
+        )
+        response = self._connection.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload)
+        return _parse_substreamed_read_response(response)
+
     def db_write(self, db_number: int, start: int, data: bytes) -> None:
         """Write raw bytes to a data block.
 
@@ -139,10 +152,22 @@ def db_write(self, db_number: int, start: int, data: bytes) -> None:
         if self._connection is None:
             raise RuntimeError("Not connected")
 
+        if self._connection.requires_substreamed:
+            self._db_write_substreamed(db_number, start, data)
+            return
+
         payload = _build_write_payload([(db_number, start, data)])
         response = self._connection.send_request(FunctionCode.SET_MULTI_VARIABLES, payload)
         _parse_write_response(response)
 
+    def _db_write_substreamed(self, db_number: int, start: int, data: bytes) -> None:
+        assert self._connection is not None
+        access_area = Ids.DB_ACCESS_AREA_BASE + (db_number & 0xFFFF)
+        payload = _build_substreamed_write_payload(
+            self._connection.session_id, access_area, Ids.DB_VALUE_ACTUAL, [start + 1, len(data)], data
+        )
+        self._connection.send_request(FunctionCode.SET_VAR_SUBSTREAMED, payload)
+
     def db_read_multi(self, items: list[tuple[int, int, int]]) -> list[bytes]:
         """Read multiple data block regions in a single request.
 
@@ -155,6 +180,9 @@ def db_read_multi(self, items: list[tuple[int, int, int]]) -> list[bytes]:
         if self._connection is None:
             raise RuntimeError("Not connected")
 
+        if self._connection.requires_substreamed:
+            return [self._db_read_substreamed(db, start, size) for db, start, size in items]
+
         payload = _build_read_payload(items)
         response = self._connection.send_request(FunctionCode.GET_MULTI_VARIABLES, payload)
         parsed = _parse_read_response(response)
@@ -175,6 +203,13 @@ def read_area(self, area_rid: int, start: int, size: int) -> bytes:
         if self._connection is None:
             raise RuntimeError("Not connected")
 
+        if self._connection.requires_substreamed:
+            payload = _build_substreamed_read_payload(
+                self._connection.session_id, area_rid, Ids.CONTROLLER_AREA_VALUE_ACTUAL, [start + 1, size]
+            )
+            response = self._connection.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload)
+            return _parse_substreamed_read_response(response)
+
         payload = _build_area_read_payload(area_rid, start, size)
         response = self._connection.send_request(FunctionCode.GET_MULTI_VARIABLES, payload)
         results = _parse_read_response(response)
@@ -193,6 +228,13 @@ def write_area(self, area_rid: int, start: int, data: bytes) -> None:
         if self._connection is None:
             raise RuntimeError("Not connected")
 
+        if self._connection.requires_substreamed:
+            payload = _build_substreamed_write_payload(
+                self._connection.session_id, area_rid, Ids.CONTROLLER_AREA_VALUE_ACTUAL, [start + 1, len(data)], data
+            )
+            self._connection.send_request(FunctionCode.SET_VAR_SUBSTREAMED, payload)
+            return
+
         payload = _build_area_write_payload(area_rid, start, data)
         response = self._connection.send_request(FunctionCode.SET_MULTI_VARIABLES, payload)
         _parse_write_response(response)
@@ -280,7 +322,10 @@ def explore(self, explore_id: int = 0) -> bytes:
         if self._connection is None:
             raise RuntimeError("Not connected")
 
-        payload = _build_explore_payload(explore_id)
+        if self._connection._session_key is not None:
+            payload = _build_explore_payload_v3(explore_id if explore_id else 0x38)
+        else:
+            payload = _build_explore_payload(explore_id)
         response = self._connection.send_request(FunctionCode.EXPLORE, payload, integrity_tail=5, reassemble=True)
         return response
 
@@ -384,7 +429,14 @@ def list_datablocks(self) -> list[dict[str, Any]]:
         if self._connection is None:
             raise RuntimeError("Not connected")
 
-        payload = _build_explore_request(Ids.NATIVE_THE_PLC_PROGRAM_RID, [Ids.OBJECT_VARIABLE_TYPE_NAME, Ids.BLOCK_BLOCK_NUMBER])
+        if self._connection._session_key is not None:
+            # V1-initial PLCs: explore the DB wildcard address (0x8A11FFFF)
+            # matching TIA Portal's browse pattern
+            payload = _build_explore_payload_v3(0x8A11FFFF)
+        else:
+            payload = _build_explore_request(
+                Ids.NATIVE_THE_PLC_PROGRAM_RID, [Ids.OBJECT_VARIABLE_TYPE_NAME, Ids.BLOCK_BLOCK_NUMBER]
+            )
         response = self._connection.send_request(FunctionCode.EXPLORE, payload, integrity_tail=5, reassemble=True)
         return _parse_explore_datablocks(response)
 
@@ -467,6 +519,8 @@ def _read_typeinfo_rid(self, db_rid: int) -> int:
 
     def _explore_type_info_container(self) -> list["typeinfo.PObject"]:
         """EXPLORE the OMS type-info container and return its per-type objects."""
+        if self._connection is None:
+            raise RuntimeError("Not connected")
         payload = _build_explore_request(Ids.OBJECT_OMS_TYPE_INFO_CONTAINER, [])
         response = self._connection.send_request(FunctionCode.EXPLORE, payload, integrity_tail=5, reassemble=True)
         return typeinfo.extract_type_info_objects(response)
@@ -682,6 +736,66 @@ def _parse_write_response(response: bytes) -> None:
         raise RuntimeError(f"Write failed: {err_str}")
 
 
+def _build_substreamed_read_payload(session_id: int, access_area: int, access_sub_area: int, lids: list[int]) -> bytes:
+    """Build a GET_VAR_SUBSTREAMED payload for data access on V1-initial PLCs."""
+    oq = encode_object_qualifier()
+    payload = bytearray()
+    payload += struct.pack(">I", session_id)
+    payload += bytes([0x20, 0x04])
+    num_fields = 4 + len(lids)
+    payload += encode_uint32_vlq(num_fields)
+    payload += encode_uint32_vlq(0)  # SymbolCRC
+    payload += encode_uint32_vlq(access_area)
+    payload += encode_uint32_vlq(len(lids) + 1)
+    payload += encode_uint32_vlq(access_sub_area)
+    for lid in lids:
+        payload += encode_uint32_vlq(lid)
+    payload += oq
+    payload += bytes([0x00])
+    payload += encode_uint32_vlq(1)
+    payload += encode_uint32_vlq(1)
+    payload += struct.pack(">I", 0)
+    return bytes(payload)
+
+
+def _build_substreamed_write_payload(
+    session_id: int, access_area: int, access_sub_area: int, lids: list[int], data: bytes
+) -> bytes:
+    """Build a SET_VAR_SUBSTREAMED payload for data access on V1-initial PLCs."""
+    oq = encode_object_qualifier()
+    payload = bytearray()
+    payload += struct.pack(">I", session_id)
+    payload += bytes([0x20, 0x04])
+    num_fields = 4 + len(lids)
+    payload += encode_uint32_vlq(num_fields)
+    payload += encode_uint32_vlq(0)  # SymbolCRC
+    payload += encode_uint32_vlq(access_area)
+    payload += encode_uint32_vlq(len(lids) + 1)
+    payload += encode_uint32_vlq(access_sub_area)
+    for lid in lids:
+        payload += encode_uint32_vlq(lid)
+    payload += oq
+    payload += bytes([0x00])
+    payload += encode_uint32_vlq(1)
+    payload += encode_pvalue_blob(data)
+    payload += encode_uint32_vlq(1)
+    payload += struct.pack(">I", 0)
+    return bytes(payload)
+
+
+def _parse_substreamed_read_response(response: bytes) -> bytes:
+    """Parse a GET_VAR_SUBSTREAMED response and extract the data bytes."""
+    offset = 0
+    return_value, consumed = decode_uint64_vlq(response, offset)
+    offset += consumed
+    if return_value != 0:
+        raise RuntimeError(f"Substreamed read failed with return value 0x{return_value:X}")
+    if offset >= len(response):
+        raise RuntimeError("Substreamed read response empty")
+    raw_bytes, consumed = decode_pvalue_to_bytes(response, offset)
+    return raw_bytes
+
+
 def _build_area_read_payload(area_rid: int, start: int, size: int) -> bytes:
     """Build a GetMultiVariables payload for controller memory area access.
 
@@ -809,6 +923,19 @@ def _build_explore_payload(explore_id: int = 0) -> bytes:
     return bytes(payload)
 
 
+def _build_explore_payload_v3(explore_id: int, sequence: int = 10) -> bytes:
+    """Build a V3-style EXPLORE request payload matching TIA Portal format.
+
+    V1-initial PLCs use a 4-byte big-endian InObjectId followed by
+    fixed parameters, rather than the VLQ-based format.
+    """
+    payload = struct.pack(">I", explore_id)
+    payload += bytes([0x00, 0x01, 0x00, 0x01, 0x00, 0x00])
+    payload += bytes([sequence & 0xFF])
+    payload += bytes([0x00, 0x00, 0x00, 0x00, 0x00])
+    return payload
+
+
 def _build_invoke_payload(state: int) -> bytes:
     """Build an INVOKE request payload for SetPlcOperatingState.
 

s7commplus/codec.py

@@ -632,13 +632,33 @@ def parse_create_object_session_id(body: bytes) -> tuple[list[int], int, int]:
     return object_ids, boff, return_value
 
 
-def parse_server_session_version(payload: bytes) -> Optional[bytes]:
-    """Scan a CreateObject response payload for the ServerSessionVersion attribute (306).
+class CreateObjectAttributes:
+    """Attributes parsed from a CreateObject response PObject tree."""
 
-    Returns the raw typed value (flags + datatype + data) to echo back during session
-    setup, or ``None`` if the attribute is not present. Real S7-1500 PLCs send it as a
-    Struct (0x17).
+    __slots__ = ("server_session_version", "public_key_fingerprint", "session_challenge")
+
+    def __init__(self) -> None:
+        self.server_session_version: Optional[bytes] = None
+        self.public_key_fingerprint: Optional[str] = None
+        self.session_challenge: Optional[bytes] = None
+
+
+def parse_create_object_attributes(payload: bytes) -> CreateObjectAttributes:
+    """Scan a CreateObject response payload for session-setup attributes.
+
+    Extracts up to three attributes from the PObject tree:
+
+    - ``306`` (ServerSessionVersion) — raw typed value echoed back in session setup.
+    - ``233`` (ObjectVariableTypeName) — ``"FF:HHHHHHHHHHHHHHHH"`` public key
+      fingerprint on V1-initial PLCs with SessionKey auth.
+    - ``303`` (ServerSessionRequest) — 20-byte session challenge (USINT array)
+      on V1-initial PLCs.
+
+    Returns a :class:`CreateObjectAttributes` with whichever attributes were found.
     """
+    from .protocol import DataType
+
+    result = CreateObjectAttributes()
     offset = 0
     while offset < len(payload):
         tag = payload[offset]
@@ -658,24 +678,61 @@ def parse_server_session_version(payload: bytes) -> Optional[bytes]:
             if attr_id == ObjectId.SERVER_SESSION_VERSION:
                 value_start = offset
                 end = skip_typed_value(payload, offset + 2, datatype, flags)
-                return bytes(payload[value_start:end])
+                result.server_session_version = bytes(payload[value_start:end])
+                offset = end
+
+            elif attr_id == Ids.OBJECT_VARIABLE_TYPE_NAME and datatype == DataType.WSTRING:
+                offset += 2
+                length, consumed = decode_uint32_vlq(payload, offset)
+                offset += consumed
+                raw = bytes(payload[offset : offset + length])
+                offset += length
+                try:
+                    text = raw.decode("utf-8")
+                    if len(text) == 19 and text[2] == ":":
+                        result.public_key_fingerprint = text
+                except (UnicodeDecodeError, ValueError):
+                    pass
+
+            elif attr_id == 303 and result.session_challenge is None:
+                is_array = bool(flags & 0x10)
+                offset += 2
+                if is_array and datatype == DataType.USINT:
+                    count, consumed = decode_uint32_vlq(payload, offset)
+                    offset += consumed
+                    result.session_challenge = bytes(payload[offset : offset + count])
+                    offset += count
+                else:
+                    offset = skip_typed_value(payload, offset, datatype, flags)
+
             else:
                 offset = skip_typed_value(payload, offset + 2, datatype, flags)
 
         elif tag == ElementID.START_OF_OBJECT:
             offset += 1
             if offset + 4 > len(payload):
                 break
-            offset += 4  # RelationId (fixed)
-            for _ in range(3):  # ClassId, ClassFlags, AttributeId (each VLQ)
+            offset += 4
+            for _ in range(3):
                 _, consumed = decode_uint32_vlq(payload, offset)
                 offset += consumed
 
         elif tag == ElementID.TERMINATING_OBJECT:
             offset += 1
         elif tag == 0x00:
-            offset += 1  # null terminator / padding
+            offset += 1
         else:
-            offset += 1  # unknown tag — skip
+            offset += 1
 
-    return None
+    return result
+
+
+def parse_server_session_version(payload: bytes) -> Optional[bytes]:
+    """Scan a CreateObject response payload for the ServerSessionVersion attribute (306).
+
+    Returns the raw typed value (flags + datatype + data) to echo back during session
+    setup, or ``None`` if the attribute is not present.
+
+    .. deprecated:: Use :func:`parse_create_object_attributes` for new code.
+    """
+    return parse_create_object_attributes(payload).server_session_version