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

@@ -135,6 +135,7 @@ Feature and bug fix release for the pure Python S7 communication library.
 
 ### 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", "zlib_dicts/*.xml"]
+s7commplus = ["py.typed", "session_auth/**/*.bin", "zlib_dicts/*.xml"]
 
 [tool.setuptools.packages.find]
 include = ["snap7*", "s7*", "s7commplus*"]

s7commplus/async_client.py

@@ -9,46 +9,53 @@
 import struct
 from typing import Any, Optional
 
-from .connection import _S7_CIPHERS, _set_s7_groups
-from .protocol import (
-    DataType,
-    ElementID,
-    FunctionCode,
-    ObjectId,
-    Opcode,
-    ProtocolVersion,
-    READ_FUNCTION_CODES,
-    S7COMMPLUS_LOCAL_TSAP,
-    S7COMMPLUS_REMOTE_TSAP,
+from . import typeinfo
+from .blob_decompressor import find_and_decompress
+from .client import (
+    _build_area_read_payload,
+    _build_area_write_payload,
+    _build_explore_payload,
+    _build_explore_request,
+    _build_invoke_payload,
+    _build_read_payload,
+    _build_subscription_request,
+    _build_symbolic_read_payload,
+    _build_symbolic_write_payload,
+    _build_write_payload,
+    _parse_explore_datablocks,
+    _parse_read_response,
+    _parse_write_response,
 )
 from .codec import (
-    encode_header,
     decode_header,
-    encode_typed_value,
+    encode_header,
     encode_object_qualifier,
     encode_pvalue_blob,
+    encode_typed_value,
     parse_create_object_session_id,
     parse_server_session_version,
 )
-from .vlq import encode_uint32_vlq, decode_uint32_vlq, decode_uint64_vlq
-from .blob_decompressor import find_and_decompress
-from .client import (
-    _build_read_payload,
-    _parse_read_response,
-    _build_write_payload,
-    _parse_write_response,
-    _build_area_read_payload,
-    _build_area_write_payload,
-    _build_symbolic_read_payload,
-    _build_symbolic_write_payload,
-    _build_explore_payload,
-    _build_invoke_payload,
-    _build_explore_request,
-    _parse_explore_datablocks,
-    _build_subscription_request,
+from .connection import (
+    _S7_CIPHERS,
+    _build_get_var_substreamed_payload,
+    _build_set_variable_payload,
+    _check_set_variable_response,
+    _parse_get_var_substreamed_response,
+    _set_s7_groups,
 )
-from . import typeinfo
-from .protocol import Ids
+from .protocol import (
+    READ_FUNCTION_CODES,
+    S7COMMPLUS_LOCAL_TSAP,
+    S7COMMPLUS_REMOTE_TSAP,
+    DataType,
+    ElementID,
+    FunctionCode,
+    Ids,
+    ObjectId,
+    Opcode,
+    ProtocolVersion,
+)
+from .vlq import decode_uint32_vlq, decode_uint64_vlq, encode_uint32_vlq
 
 logger = logging.getLogger(__name__)
 
@@ -321,89 +328,33 @@ async def _tls_read_incoming(self) -> None:
 
     async def _get_legitimation_challenge(self) -> bytes:
         """Request legitimation challenge from PLC."""
-        from .protocol import LegitimationId, DataType as DT
+        from .protocol import LegitimationId
 
-        payload = bytearray()
-        payload += struct.pack(">I", self._session_id)
-        payload += encode_uint32_vlq(1)
-        payload += encode_uint32_vlq(1)
-        payload += encode_uint32_vlq(LegitimationId.SERVER_SESSION_REQUEST)
-        payload += struct.pack(">I", 0)
-
-        resp_payload = await self._send_request(FunctionCode.GET_VAR_SUBSTREAMED, bytes(payload))
-
-        offset = 0
-        return_value, consumed = decode_uint64_vlq(resp_payload, offset)
-        offset += consumed
-
-        if return_value != 0:
-            from snap7.error import S7ConnectionError
-
-            raise S7ConnectionError(f"GetVarSubStreamed for challenge failed: return_value={return_value}")
-
-        if offset + 2 > len(resp_payload):
-            from snap7.error import S7ConnectionError
-
-            raise S7ConnectionError("Challenge response too short")
-
-        _flags = resp_payload[offset]
-        datatype = resp_payload[offset + 1]
-        offset += 2
-
-        if datatype == DT.BLOB:
-            length, consumed = decode_uint32_vlq(resp_payload, offset)
-            offset += consumed
-            return bytes(resp_payload[offset : offset + length])
-        else:
-            count, consumed = decode_uint32_vlq(resp_payload, offset)
-            offset += consumed
-            return bytes(resp_payload[offset : offset + count])
+        payload = _build_get_var_substreamed_payload(self._session_id, LegitimationId.SERVER_SESSION_REQUEST)
+        resp_payload = await self._send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload, integrity_tail=4)
+        return _parse_get_var_substreamed_response(resp_payload)
 
     async def _send_legitimation_new(self, encrypted_response: bytes) -> None:
         """Send new-style legitimation response (AES-256-CBC encrypted)."""
-        from .protocol import LegitimationId, DataType as DT
-
-        payload = bytearray()
-        payload += struct.pack(">I", self._session_id)
-        payload += encode_uint32_vlq(1)
-        payload += encode_uint32_vlq(LegitimationId.LEGITIMATE)
-        payload += bytes([0x00, DT.BLOB])
-        payload += encode_uint32_vlq(len(encrypted_response))
-        payload += encrypted_response
-        payload += struct.pack(">I", 0)
-
-        resp_payload = await self._send_request(FunctionCode.SET_VARIABLE, bytes(payload))
-
-        if len(resp_payload) >= 1:
-            return_value, _ = decode_uint64_vlq(resp_payload, 0)
-            if return_value < 0:
-                from snap7.error import S7ConnectionError
+        from .protocol import LegitimationId
 
-                raise S7ConnectionError(f"Legitimation rejected by PLC: return_value={return_value}")
-            logger.debug(f"New legitimation return_value={return_value}")
+        value = bytes([0x00, DataType.BLOB, 0x00])
+        value += encode_uint32_vlq(len(encrypted_response))
+        value += encrypted_response
+        payload = _build_set_variable_payload(self._session_id, LegitimationId.LEGITIMATE, value)
+        resp_payload = await self._send_request(FunctionCode.SET_VARIABLE, payload, integrity_tail=4)
+        _check_set_variable_response(resp_payload)
 
     async def _send_legitimation_legacy(self, response: bytes) -> None:
         """Send legacy legitimation response (SHA-1 XOR)."""
-        from .protocol import LegitimationId, DataType as DT
-
-        payload = bytearray()
-        payload += struct.pack(">I", self._session_id)
-        payload += encode_uint32_vlq(1)
-        payload += encode_uint32_vlq(LegitimationId.SERVER_SESSION_RESPONSE)
-        payload += bytes([0x10, DT.USINT])  # flags=0x10 (array)
-        payload += encode_uint32_vlq(len(response))
-        payload += response
-        payload += struct.pack(">I", 0)
-
-        resp_payload = await self._send_request(FunctionCode.SET_VARIABLE, bytes(payload))
-
-        if len(resp_payload) >= 1:
-            return_value, _ = decode_uint64_vlq(resp_payload, 0)
-            if return_value < 0:
-                from snap7.error import S7ConnectionError
-
-                raise S7ConnectionError(f"Legacy legitimation rejected by PLC: return_value={return_value}")
-            logger.debug(f"Legacy legitimation return_value={return_value}")
+        from .protocol import LegitimationId
+
+        value = bytes([0x10, DataType.USINT])
+        value += encode_uint32_vlq(len(response))
+        value += response
+        payload = _build_set_variable_payload(self._session_id, LegitimationId.SERVER_SESSION_RESPONSE, value)
+        resp_payload = await self._send_request(FunctionCode.SET_VARIABLE, payload, integrity_tail=4)
+        _check_set_variable_response(resp_payload)
 
     async def disconnect(self) -> None:
         """Disconnect from PLC."""