From ce4c1c6bdf884deb0328f3619e17a3d32f576a65 Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Tue, 18 Aug 2026 21:43:09 +0200 Subject: [PATCH 1/2] feat: expose the PLC protection level on the S7CommPlus clients Read the session's effective protection level (id 1842) once the S7CommPlus session is established, and expose it as a protection_level property on the connection, the sync client and the async client, so callers can tell an unprotected PLC from one that needs legitimation. AccessLevel names the values the PLC reports. The request reuses the existing GetVarSubStreamed payload builder, whose framing matches the reference driver: the 0x0001 field following the ObjectQualifier is a protocol constant, and the IntegrityId is the separate VLQ that send_request splices in after it. --- s7commplus/async_client.py | 20 ++++++++++++++++++ s7commplus/client.py | 7 +++++++ s7commplus/connection.py | 32 +++++++++++++++++++++++++++++ s7commplus/protocol.py | 24 ++++++++++++++++++++++ tests/test_s7_v2.py | 42 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 125 insertions(+) diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 9ca7600c..246c5af0 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -41,6 +41,7 @@ _build_set_variable_payload, _check_set_variable_response, _parse_get_var_substreamed_response, + _parse_protection_level_response, _set_s7_groups, ) from .protocol import ( @@ -96,6 +97,8 @@ def __init__(self) -> None: # so it can be echoed back verbatim — real S7-1500 PLCs send it as a Struct. self._server_session_version: Optional[bytes] = None self._session_setup_ok: bool = False + # Effective protection level, read once the session is up + self._protection_level: Optional[int] = None @property def connected(self) -> bool: @@ -124,6 +127,11 @@ def oms_secret(self) -> Optional[bytes]: """OMS exporter secret from TLS session (None if TLS not active).""" return self._oms_secret + @property + def protection_level(self) -> Optional[int]: + """Effective protection level reported by the PLC (see `AccessLevel`).""" + return self._protection_level + async def connect( self, host: str, @@ -196,6 +204,9 @@ async def connect( else: logger.warning("PLC did not provide ServerSessionVersion - session setup incomplete") self._session_setup_ok = False + self._protection_level = await self._get_effective_protection_level() + logger.info(f"PLC reports protection level: {self._protection_level}") + logger.info( f"Async S7CommPlus connected to {host}:{port}, " f"version=V{self._protocol_version}, session={self._session_id}, " @@ -326,6 +337,14 @@ async def _tls_read_incoming(self) -> None: data = await self._recv_cotp_raw() self._incoming_bio.write(data) + async def _get_effective_protection_level(self) -> int: + """ + Read the session's effective protection level (see `AccessLevel`). + """ + payload = _build_get_var_substreamed_payload(self._session_id, Ids.EFFECTIVE_PROTECTION_LEVEL) + resp = await self._send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload, integrity_tail=4) + return _parse_protection_level_response(resp) + async def _get_legitimation_challenge(self) -> bytes: """Request legitimation challenge from PLC.""" from .protocol import LegitimationId @@ -378,6 +397,7 @@ async def disconnect(self) -> None: self._oms_secret = None self._server_session_version = None self._session_setup_ok = False + self._protection_level = None if self._writer: try: diff --git a/s7commplus/client.py b/s7commplus/client.py index 106d6f60..7a47c819 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -70,6 +70,13 @@ def tls_active(self) -> bool: return False return self._connection.tls_active + @property + def protection_level(self) -> Optional[int]: + """Effective protection level reported by the PLC (see `AccessLevel`).""" + if self._connection is None: + return None + return self._connection.protection_level + def connect( self, host: str, diff --git a/s7commplus/connection.py b/s7commplus/connection.py index e1f6bd83..0c119757 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -151,6 +151,18 @@ def _parse_get_var_substreamed_response(payload: bytes) -> bytes: raise S7ConnectionError(f"Malformed GetVarSubStreamed response: {exc}") from exc +def _parse_protection_level_response(payload: bytes) -> int: + """Extract the scalar UDInt value from a GetVarSubStreamed response payload.""" + from snap7.error import S7ConnectionError + + return_value, offset = decode_uint64_vlq(payload, 0) + if return_value != 0: + raise S7ConnectionError(f"GetVarSubStreamed for the protection level failed: return_value={return_value}") + offset += 1 # protocol-defined unknown byte + value, _ = decode_uint32_vlq(payload, offset + 2) + return value + + def _build_set_variable_payload(in_object_id: int, address: int, value: bytes) -> bytes: """Build a SetVariable request set around an already encoded PValue.""" payload = struct.pack(">I", in_object_id) @@ -299,6 +311,9 @@ def __init__( # Password for post-auth legitimation (V1-initial PLCs) self._connect_password: str = "" + # Effective protection level, read once the session is up + self._protection_level: Optional[int] = None + @property def connected(self) -> bool: return self._connected @@ -349,6 +364,11 @@ def oms_secret(self) -> Optional[bytes]: """OMS exporter secret from TLS session (for legitimation).""" return self._oms_secret + @property + def protection_level(self) -> Optional[int]: + """Effective protection level reported by the PLC (see `AccessLevel`).""" + return self._protection_level + def connect( self, timeout: float = 5.0, @@ -428,6 +448,9 @@ def connect( self._connected = True + self._protection_level = self._get_effective_protection_level() + logger.info(f"PLC reports protection level: {self._protection_level}") + if self._session_key is not None and self._session_setup_ok: self._session_activate() self._post_auth_legitimation(password=self._connect_password) @@ -490,6 +513,14 @@ def authenticate(self, password: str, username: str = "") -> None: logger.info("PLC legitimation completed successfully") + def _get_effective_protection_level(self) -> int: + """ + Read the session's effective protection level (see `AccessLevel`). + """ + payload = _build_get_var_substreamed_payload(self._session_id, Ids.EFFECTIVE_PROTECTION_LEVEL) + resp = self.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload, integrity_tail=4) + return _parse_protection_level_response(resp) + def _get_legitimation_challenge(self) -> bytes: """Request legitimation challenge from PLC. @@ -613,6 +644,7 @@ def disconnect(self) -> None: self._with_integrity_id = False self._integrity_id_read = 0 self._integrity_id_write = 0 + self._protection_level = None self._iso_conn.disconnect() def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: int = 4, reassemble: bool = False) -> bytes: diff --git a/s7commplus/protocol.py b/s7commplus/protocol.py index 3e13a627..4d869630 100644 --- a/s7commplus/protocol.py +++ b/s7commplus/protocol.py @@ -206,6 +206,10 @@ class Ids(IntEnum): ALARM_SUBSCRIPTION_REF_ALARM_DOMAIN = 2659 ALARM_SUBSCRIPTION_REF_ITS_ALARM_SUBSYSTEM = 2660 + # Session's effective protection level, readable via GetVarSubStreamed + EFFECTIVE_PROTECTION_LEVEL = 1842 + ACTIVE_PROTECTION_LEVEL = 1843 + # DB AccessArea base (add DB number to get area ID) DB_ACCESS_AREA_BASE = 0x8A0E0000 @@ -223,6 +227,26 @@ class Ids(IntEnum): ) +class AccessLevel(IntEnum): + """Protection levels reported by `Ids.EFFECTIVE_PROTECTION_LEVEL`. + + Lower is more privileged. A successful legitimation lowers the level; the + level reached depends on which password was configured for which role, so + it does not necessarily become `FULL_ACCESS`. + + `NONE` is not in the C# reference; PLCs with no protection configured + report it. + + Reference: thomas-v2/S7CommPlusDriver/Legitimation/AccessLevel.cs + """ + + NONE = 0 + FULL_ACCESS = 1 + READ_ACCESS = 2 + HMI_ACCESS = 3 + NO_ACCESS = 4 + + class LegitimationId(IntEnum): """Legitimation IDs used in password authentication (V2+). diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index 0d747c00..d9d3b85a 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -18,6 +18,7 @@ _build_set_variable_payload, _check_set_variable_response, _parse_get_var_substreamed_response, + _parse_protection_level_response, ) from s7commplus.legitimation import ( LegitimationState, @@ -27,8 +28,10 @@ ) from s7commplus.protocol import ( READ_FUNCTION_CODES, + AccessLevel, DataType, FunctionCode, + Ids, LegitimationId, ProtocolVersion, ) @@ -328,6 +331,45 @@ async def test_async_challenge_uses_protocol_request_shape(self) -> None: ) +class TestProtectionLevel: + """The effective protection level read that precedes legitimation.""" + + # Captured from a password-protected S7-1512: UDInt(4), trailing IntegrityId 7. + RESPONSE = bytes.fromhex("00000004040700000000") + + def test_parse_scalar_udint(self) -> None: + assert _parse_protection_level_response(self.RESPONSE) == AccessLevel.NO_ACCESS + + def test_parse_rejects_nonzero_return(self) -> None: + with pytest.raises(S7ConnectionError, match="return_value=4660"): + _parse_protection_level_response(encode_uint32_vlq(0x1234)) + + def test_sync_read_uses_protocol_request_shape(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._session_id = 0x01020304 + conn.send_request = MagicMock(return_value=self.RESPONSE) + + assert conn._get_effective_protection_level() == AccessLevel.NO_ACCESS + conn.send_request.assert_called_once_with( + FunctionCode.GET_VAR_SUBSTREAMED, + _build_get_var_substreamed_payload(0x01020304, Ids.EFFECTIVE_PROTECTION_LEVEL), + integrity_tail=4, + ) + + @pytest.mark.asyncio + async def test_async_read_uses_protocol_request_shape(self) -> None: + client = S7CommPlusAsyncClient() + client._session_id = 0x01020304 + client._send_request = AsyncMock(return_value=self.RESPONSE) + + assert await client._get_effective_protection_level() == AccessLevel.NO_ACCESS + client._send_request.assert_awaited_once_with( + FunctionCode.GET_VAR_SUBSTREAMED, + _build_get_var_substreamed_payload(0x01020304, Ids.EFFECTIVE_PROTECTION_LEVEL), + integrity_tail=4, + ) + + class TestSessionKeySelection: def test_tls_v2_does_not_attempt_session_key_auth(self) -> None: conn = S7CommPlusConnection("127.0.0.1") From a6bcfcedcaba257b15a118f1582300735a34c23d Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Wed, 19 Aug 2026 23:36:07 +0200 Subject: [PATCH 2/2] fix: Apply review suggestions. --- s7commplus/async_client.py | 30 +++++++++++++----- s7commplus/connection.py | 63 +++++++++++++++++++++++++++++--------- tests/test_s7_v2.py | 23 ++++++++++++++ 3 files changed, 93 insertions(+), 23 deletions(-) diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 246c5af0..93f4b4aa 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -204,8 +204,12 @@ async def connect( else: logger.warning("PLC did not provide ServerSessionVersion - session setup incomplete") self._session_setup_ok = False - self._protection_level = await self._get_effective_protection_level() - logger.info(f"PLC reports protection level: {self._protection_level}") + + # Only a session that completed setup answers attribute reads. + if self._session_setup_ok: + self._protection_level = await self._get_effective_protection_level() + if self._protection_level is not None: + logger.info(f"PLC reports protection level: {self._protection_level}") logger.info( f"Async S7CommPlus connected to {host}:{port}, " @@ -255,6 +259,11 @@ async def authenticate(self, password: str, username: str = "") -> None: logger.info("PLC legitimation completed successfully") + # Renew protection level + self._protection_level = await self._get_effective_protection_level() + if self._protection_level is not None: + logger.info(f"PLC reports protection level: {self._protection_level}") + async def _activate_tls( self, tls_cert: Optional[str] = None, @@ -337,13 +346,18 @@ async def _tls_read_incoming(self) -> None: data = await self._recv_cotp_raw() self._incoming_bio.write(data) - async def _get_effective_protection_level(self) -> int: - """ - Read the session's effective protection level (see `AccessLevel`). - """ + async def _get_effective_protection_level(self) -> Optional[int]: + """Read the session's effective protection level (see `AccessLevel`), None if request failed.""" + from snap7.error import S7ConnectionError + payload = _build_get_var_substreamed_payload(self._session_id, Ids.EFFECTIVE_PROTECTION_LEVEL) - resp = await self._send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload, integrity_tail=4) - return _parse_protection_level_response(resp) + try: + resp = await self._send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload, integrity_tail=4) + level = _parse_protection_level_response(resp) + except S7ConnectionError as exc: + logger.warning(f"PLC did not report a protection level: {exc}") + return None + return level async def _get_legitimation_challenge(self) -> bytes: """Request legitimation challenge from PLC.""" diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 0c119757..756fd5e3 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -155,12 +155,30 @@ def _parse_protection_level_response(payload: bytes) -> int: """Extract the scalar UDInt value from a GetVarSubStreamed response payload.""" from snap7.error import S7ConnectionError - return_value, offset = decode_uint64_vlq(payload, 0) - if return_value != 0: - raise S7ConnectionError(f"GetVarSubStreamed for the protection level failed: return_value={return_value}") - offset += 1 # protocol-defined unknown byte - value, _ = decode_uint32_vlq(payload, offset + 2) - return value + try: + return_value, offset = decode_uint64_vlq(payload, 0) + if return_value != 0: + raise S7ConnectionError(f"GetVarSubStreamed for the protection level failed: return_value={return_value}") + + if offset >= len(payload): + raise ValueError("missing response marker") + offset += 1 # protocol-defined unknown byte + + if offset + 2 > len(payload): + raise ValueError("missing PValue header") + flags = payload[offset] + datatype = payload[offset + 1] + offset += 2 + + if datatype != DataType.UDINT or flags & 0x10: + raise ValueError(f"expected a scalar UDInt, got flags=0x{flags:02X} datatype=0x{datatype:02X}") + + value, _ = decode_uint32_vlq(payload, offset) + return value + except S7ConnectionError: + raise + except (IndexError, ValueError) as exc: + raise S7ConnectionError(f"Malformed protection level response: {exc}") from exc def _build_set_variable_payload(in_object_id: int, address: int, value: bytes) -> bytes: @@ -448,12 +466,16 @@ def connect( self._connected = True - self._protection_level = self._get_effective_protection_level() - logger.info(f"PLC reports protection level: {self._protection_level}") - if self._session_key is not None and self._session_setup_ok: self._session_activate() self._post_auth_legitimation(password=self._connect_password) + + # Only a session that completed setup answers attribute reads + if self._session_setup_ok: + self._protection_level = self._get_effective_protection_level() + if self._protection_level is not None: + logger.info(f"PLC reports protection level: {self._protection_level}") + logger.info( f"S7CommPlus connected to {self.host}:{self.port}, " f"version=V{self._protocol_version}, session={self._session_id}, " @@ -513,13 +535,24 @@ def authenticate(self, password: str, username: str = "") -> None: logger.info("PLC legitimation completed successfully") - def _get_effective_protection_level(self) -> int: - """ - Read the session's effective protection level (see `AccessLevel`). - """ + # Renew protection level + self._protection_level = self._get_effective_protection_level() + if self._protection_level is not None: + logger.info(f"PLC reports protection level: {self._protection_level}") + + def _get_effective_protection_level(self) -> Optional[int]: + """Read the session's effective protection level (see `AccessLevel`), None if request failed.""" + from snap7.error import S7ConnectionError + payload = _build_get_var_substreamed_payload(self._session_id, Ids.EFFECTIVE_PROTECTION_LEVEL) - resp = self.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload, integrity_tail=4) - return _parse_protection_level_response(resp) + try: + level = _parse_protection_level_response( + self.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload, integrity_tail=4) + ) + except S7ConnectionError as exc: + logger.warning(f"PLC did not report a protection level: {exc}") + return None + return level def _get_legitimation_challenge(self) -> bytes: """Request legitimation challenge from PLC. diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index d9d3b85a..a93efa6f 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -344,6 +344,29 @@ def test_parse_rejects_nonzero_return(self) -> None: with pytest.raises(S7ConnectionError, match="return_value=4660"): _parse_protection_level_response(encode_uint32_vlq(0x1234)) + def test_parse_rejects_missing_response_marker(self) -> None: + with pytest.raises(S7ConnectionError, match="missing response marker"): + _parse_protection_level_response(bytes([0x00])) + + def test_parse_rejects_truncated_pvalue_header(self) -> None: + with pytest.raises(S7ConnectionError, match="missing PValue header"): + _parse_protection_level_response(bytes([0x00, 0x00, 0x00])) + + def test_parse_rejects_non_udint_datatype(self) -> None: + response = bytes([0x00, 0x00, 0x00, DataType.USINT, 0x04]) + with pytest.raises(S7ConnectionError, match="expected a scalar UDInt, got flags=0x00 datatype=0x02"): + _parse_protection_level_response(response) + + def test_parse_rejects_udint_array(self) -> None: + response = bytes([0x00, 0x00, 0x10, DataType.UDINT, 0x01, 0x04]) + with pytest.raises(S7ConnectionError, match="expected a scalar UDInt, got flags=0x10 datatype=0x04"): + _parse_protection_level_response(response) + + def test_parse_rejects_truncated_value(self) -> None: + response = bytes([0x00, 0x00, 0x00, DataType.UDINT, 0x84]) + with pytest.raises(S7ConnectionError, match="Malformed protection level response"): + _parse_protection_level_response(response) + def test_sync_read_uses_protocol_request_shape(self) -> None: conn = S7CommPlusConnection("127.0.0.1") conn._session_id = 0x01020304