feat: expose the PLC protection level on the S7CommPlus clients

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,13 @@ async def connect(
             else:
                 logger.warning("PLC did not provide ServerSessionVersion - session setup incomplete")
                 self._session_setup_ok = False
+
+            # 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}, "
                 f"version=V{self._protocol_version}, session={self._session_id}, "
@@ -244,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,
@@ -326,6 +346,19 @@ 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) -> 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)
+        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."""
         from .protocol import LegitimationId
@@ -378,6 +411,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:

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,

s7commplus/connection.py

@@ -151,6 +151,36 @@ 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
+
+    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:
     """Build a SetVariable request set around an already encoded PValue."""
     payload = struct.pack(">I", in_object_id)
@@ -299,6 +329,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 +382,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,
@@ -431,6 +469,13 @@ def connect(
             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}, "
@@ -490,6 +535,25 @@ def authenticate(self, password: str, username: str = "") -> None:
 
         logger.info("PLC legitimation completed successfully")
 
+        # 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)
+        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.
 
@@ -613,6 +677,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:

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+).
 

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,68 @@ 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_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
+        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")