diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 9ca7600c..93f4b4aa 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,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: 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..756fd5e3 100644 --- a/s7commplus/connection.py +++ b/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: 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..a93efa6f 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,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")