diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 93f4b4aa..4d0f205a 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -44,14 +44,23 @@ _parse_protection_level_response, _set_s7_groups, ) +from .legitimation import ( + build_legacy_response, + build_new_response, + decide_legitimation_mode, + derive_legitimation_key, + extract_session_version_string, +) from .protocol import ( READ_FUNCTION_CODES, S7COMMPLUS_LOCAL_TSAP, S7COMMPLUS_REMOTE_TSAP, + AccessLevel, DataType, ElementID, FunctionCode, Ids, + LegitimationType, ObjectId, Opcode, ProtocolVersion, @@ -226,43 +235,76 @@ async def authenticate(self, password: str, username: str = "") -> None: Args: password: PLC password - username: Username for new-style auth (optional) + username: Username for the new-mode exchange (leave empty for legacy) Raises: - S7ConnectionError: If not connected, TLS not active, or auth fails + S7ConnectionError: If not connected, TLS is not active, the firmware + does not support legitimation, or the password was refused """ - if not self._connected: - from snap7.error import S7ConnectionError + from snap7.error import S7ConnectionError + if not self._connected: raise S7ConnectionError("Not connected") - if not self._tls_active or self._oms_secret is None: - from snap7.error import S7ConnectionError - + if not self._tls_active: raise S7ConnectionError("Legitimation requires TLS. Connect with use_tls=True.") + level_before = self._protection_level + if level_before is None: + raise S7ConnectionError("PLC does not report a protection level, so legitimation cannot be verified") + + if level_before <= AccessLevel.FULL_ACCESS: + logger.info("PLC already grants full access, legitimation is not required") + return + if not password: + logger.warning(f"PLC restricts access (level {level_before}) but no password was provided") + return + + # Step 1: Auto-detect legacy vs new from the firmware version + mode = self._decide_legitimation_mode() + if mode is None: + raise S7ConnectionError("PLC firmware version does not support legitimation") + logger.info(f"Using {mode.name.lower()} legitimation") + + # Step 2: Get challenge from PLC via GetVarSubStreamed challenge = await self._get_legitimation_challenge() logger.info(f"Received legitimation challenge ({len(challenge)} bytes)") - from .legitimation import build_legacy_response, build_new_response - - if username: - response_data = build_new_response(password, challenge, self._oms_secret, username) - await self._send_legitimation_new(response_data) + if mode is LegitimationType.LEGACY: + # A legacy challenge is XORed with a SHA-1 password hash, so it is that long. + if len(challenge) != 20: + raise S7ConnectionError(f"Unexpected legacy challenge length: {len(challenge)}") + await self._send_legitimation_legacy(build_legacy_response(password, challenge)) else: - try: - response_data = build_new_response(password, challenge, self._oms_secret, "") - await self._send_legitimation_new(response_data) - except NotImplementedError: - response_data = build_legacy_response(password, challenge) - await self._send_legitimation_legacy(response_data) - - logger.info("PLC legitimation completed successfully") + if self._oms_secret is None: + raise S7ConnectionError( + "New legitimation requires the TLS OMS exporter secret, which could not be derived from this TLS session." + ) + await self._send_legitimation_new(build_new_response(password, challenge, self._oms_secret, username)) + # The PLC rolls the key after every attempt; mirror it so a second + # legitimation on the same session encrypts with the same key. + self._oms_secret = derive_legitimation_key(self._oms_secret) - # Renew protection level + # Step 3: Renew protection level, which is what verifies the outcome 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}") + if self._protection_level is None: + raise S7ConnectionError("Legitimation outcome is unverifiable: the PLC stopped reporting its protection level") + if self._protection_level >= level_before: + raise S7ConnectionError( + f"Legitimation failed, protection level unchanged at {self._protection_level}: the password was refused" + ) + logger.info(f"PLC legitimation completed, protection level {level_before} -> {self._protection_level}") + + def _decide_legitimation_mode(self) -> Optional[LegitimationType]: + """Return the legitimation exchange the PLC firmware expects, None if unsupported.""" + if self._server_session_version is None: + return None + version_string = extract_session_version_string(self._server_session_version) + if version_string is None: + logger.warning("ServerSessionVersion carries no device string, cannot pick a legitimation mode") + return None + logger.debug(f"PLC device string: {version_string}") + return decide_legitimation_mode(version_string) async def _activate_tls( self, diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 756fd5e3..3c4aa837 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -51,15 +51,24 @@ from snap7.connection import ISOTCPConnection from .codec import decode_header, encode_header, encode_object_qualifier, parse_create_object_attributes +from .legitimation import ( + build_legacy_response, + build_new_response, + decide_legitimation_mode, + derive_legitimation_key, + extract_session_version_string, +) from .protocol import ( READ_FUNCTION_CODES, S7COMMPLUS_LOCAL_TSAP, S7COMMPLUS_REMOTE_TSAP, + AccessLevel, DataType, ElementID, FunctionCode, Ids, LegitimationId, + LegitimationType, ObjectId, Opcode, ProtocolVersion, @@ -194,7 +203,10 @@ def _build_set_variable_payload(in_object_id: int, address: int, value: bytes) - def _check_set_variable_response(payload: bytes) -> None: - """Raise when a SetVariable response reports a non-zero return value.""" + """Raise when a SetVariable response reports a refused legitimation. + + Reference: thomas-v2/S7CommPlusDriver/Legitimation/Legitimation.cs + """ from snap7.error import S7ConnectionError if not payload: @@ -203,7 +215,9 @@ def _check_set_variable_response(payload: bytes) -> None: return_value, _ = decode_uint64_vlq(payload, 0) except ValueError as exc: raise S7ConnectionError(f"Malformed SetVariable response: {exc}") from exc - if return_value != 0: + # The low 16 bits of the status word are a signed error code; the reference + # driver casts them with (Int16) and rejects negatives, i.e. the sign bit. + if return_value & 0x8000: raise S7ConnectionError(f"Legitimation rejected by PLC: return_value=0x{return_value:X}") @@ -470,7 +484,8 @@ def connect( self._session_activate() self._post_auth_legitimation(password=self._connect_password) - # Only a session that completed setup answers attribute reads + # Only a session that completed setup answers attribute reads; the + # V1-initial band falls back to legacy PUT/GET and never gets here. if self._session_setup_ok: self._protection_level = self._get_effective_protection_level() if self._protection_level is not None: @@ -497,48 +512,76 @@ def authenticate(self, password: str, username: str = "") -> None: Args: password: PLC password - username: Username for new-style auth (optional) + username: Username for the new-mode exchange (leave empty for legacy) Raises: - S7ConnectionError: If not connected, TLS not active, or auth fails + S7ConnectionError: If not connected, TLS is not active, the firmware + does not support legitimation, or the password was refused """ - if not self._connected: - from snap7.error import S7ConnectionError + from snap7.error import S7ConnectionError + if not self._connected: raise S7ConnectionError("Not connected") - if not self._tls_active or self._oms_secret is None: - from snap7.error import S7ConnectionError - + if not self._tls_active: raise S7ConnectionError("Legitimation requires TLS. Connect with use_tls=True.") - # Step 1: Get challenge from PLC via GetVarSubStreamed + level_before = self._protection_level + if level_before is None: + raise S7ConnectionError("PLC does not report a protection level, so legitimation cannot be verified") + + if level_before <= AccessLevel.FULL_ACCESS: + logger.info("PLC already grants full access, legitimation is not required") + return + if not password: + logger.warning(f"PLC restricts access (level {level_before}) but no password was provided") + return + + # Step 1: Auto-detect legacy vs new from the firmware version + mode = self._decide_legitimation_mode() + if mode is None: + raise S7ConnectionError("PLC firmware version does not support legitimation") + logger.info(f"Using {mode.name.lower()} legitimation") + + # Step 2: Get challenge from PLC via GetVarSubStreamed challenge = self._get_legitimation_challenge() logger.info(f"Received legitimation challenge ({len(challenge)} bytes)") - # Step 2: Build response (auto-detect legacy vs new) - from .legitimation import build_legacy_response, build_new_response - - if username: - # New-style auth with username always uses AES-256-CBC - response_data = build_new_response(password, challenge, self._oms_secret, username) - self._send_legitimation_new(response_data) + if mode is LegitimationType.LEGACY: + # A legacy challenge is XORed with a SHA-1 password hash, so it is that long. + if len(challenge) != 20: + raise S7ConnectionError(f"Unexpected legacy challenge length: {len(challenge)}") + self._send_legitimation_legacy(build_legacy_response(password, challenge)) else: - # Try new-style first, fall back to legacy SHA-1 XOR - try: - response_data = build_new_response(password, challenge, self._oms_secret, "") - self._send_legitimation_new(response_data) - except NotImplementedError: - # cryptography package not available, use legacy - response_data = build_legacy_response(password, challenge) - self._send_legitimation_legacy(response_data) - - logger.info("PLC legitimation completed successfully") + if self._oms_secret is None: + raise S7ConnectionError( + "New legitimation requires the TLS OMS exporter secret, which could not be derived from this TLS session." + ) + self._send_legitimation_new(build_new_response(password, challenge, self._oms_secret, username)) + # The PLC rolls the key after every attempt; mirror it so a second + # legitimation on the same session encrypts with the same key. + self._oms_secret = derive_legitimation_key(self._oms_secret) - # Renew protection level + # Step 3: Renew protection level, which is what verifies the outcome 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}") + if self._protection_level is None: + raise S7ConnectionError("Legitimation outcome is unverifiable: the PLC stopped reporting its protection level") + if self._protection_level >= level_before: + raise S7ConnectionError( + f"Legitimation failed, protection level unchanged at {self._protection_level}: the password was refused" + ) + logger.info(f"PLC legitimation completed, protection level {level_before} -> {self._protection_level}") + + def _decide_legitimation_mode(self) -> Optional[LegitimationType]: + """Return the legitimation exchange the PLC firmware expects, None if unsupported.""" + if self._server_session_version is None: + return None + version_string = extract_session_version_string(self._server_session_version) + if version_string is None: + logger.warning("ServerSessionVersion carries no device string, cannot pick a legitimation mode") + return None + logger.debug(f"PLC device string: {version_string}") + return decide_legitimation_mode(version_string) def _get_effective_protection_level(self) -> Optional[int]: """Read the session's effective protection level (see `AccessLevel`), None if request failed.""" diff --git a/s7commplus/legitimation.py b/s7commplus/legitimation.py index 2c8e197e..75b52d9c 100644 --- a/s7commplus/legitimation.py +++ b/s7commplus/legitimation.py @@ -14,11 +14,131 @@ import hashlib import logging +import struct from typing import Optional +from .protocol import DataType, Ids, LegitimationType +from .vlq import decode_uint32_vlq, encode_uint32_vlq + logger = logging.getLogger(__name__) +def _parse_paom_string(version_string: str) -> Optional[tuple[str, int]]: + """Read the device series and firmware number out of a ServerSessionVersion PAOM string. + + A PAOM string is `;;`. Only the leading digit + of the model number selects the series, and the firmware is compared as + `major * 100 + minor`, so those are what this returns. + + ```python + _parse_paom_string("1;6ES7 512-1CK01-0AB0;V2.9") # ("5", 209) + ``` + + The reference does this with one pattern over the whole string, + `^[^;]*;[^;]*[17]\\s?(\\d{3}).*;[VS](\\d{1,2}\\.\\d+)$`, which this matches except + that it also tolerates more than one space in front of the model number. + + :param version_string: PAOM string from `extract_session_version_string`. + :return: The series digit and firmware number, or None when either is unreadable. + + Reference: thomas-v2/S7CommPlusDriver/Legitimation/Legitimation.cs + """ + fields = version_string.split(";") + if len(fields) < 3: + return None + + # The model number ends the order number, behind a vendor prefix whose last + # digit is 1 or 7: "6ES7 512-1CK01-0AB0" -> prefix "6ES7", model "512". + order_number = fields[1].split("-")[0].rstrip() + model, prefix = order_number[-3:], order_number[:-3].rstrip() + if len(model) != 3 or not model.isdecimal() or prefix[-1:] not in ("1", "7"): + return None + + firmware = fields[-1] + if firmware[:1].upper() not in ("V", "S"): + return None + major, dot, minor = firmware[1:].partition(".") + if not dot or not (1 <= len(major) <= 2) or not major.isdecimal() or not minor.isdecimal(): + return None + + return model[0], int(major) * 100 + int(minor) + + +def extract_session_version_string(raw: bytes) -> Optional[str]: + """Extract the device PAOM string from a raw ServerSessionVersion value. + + The value is the typed ServerSessionVersion captured from the CreateObject + response. Element `Ids.SESSION_VERSION_SYSTEM_PAOM_STRING` holds the device + identity and firmware version as a WString. + + ```python + version = extract_session_version_string(connection.server_session_version) + # '1;6ES7 512-1CK01-0AB0;V2.9' + ``` + + :param raw: Raw typed ServerSessionVersion value (flags + datatype + struct data). + :return: The PAOM string, or None when the element is absent or undecodable. + """ + needle = encode_uint32_vlq(Ids.SESSION_VERSION_SYSTEM_PAOM_STRING) + search_from = 0 + while True: + index = raw.find(needle, search_from) + if index < 0: + return None + search_from = index + 1 + # [VLQ key][flags][datatype][VLQ length][utf-8 bytes] + length_at = index + len(needle) + 2 + if length_at > len(raw) or raw[length_at - 1] != DataType.WSTRING: + continue + try: + length, consumed = decode_uint32_vlq(raw, length_at) + except ValueError: + continue + start = length_at + consumed + if start + length > len(raw): + continue + try: + return raw[start : start + length].decode("utf-8") + except UnicodeDecodeError: + continue + + +def decide_legitimation_mode(version_string: str) -> Optional[LegitimationType]: + """Decide legacy (SHA-1 XOR) vs new (AES-256-CBC) legitimation from the firmware. + + ```python + decide_legitimation_mode("1;6ES7 512-1CK01-0AB0;V2.9") # LegitimationType.LEGACY + ``` + + :param version_string: PAOM string from `extract_session_version_string`. + :return: The mode to use, or None when the device or firmware does not + support legitimation at all. + + Reference: thomas-v2/S7CommPlusDriver/Legitimation/Legitimation.cs + """ + parsed = _parse_paom_string(version_string) + if parsed is None: + logger.warning(f"Could not extract the firmware version from {version_string!r}") + return None + series, firmware = parsed + + if series == "5": # S7-1500 + if firmware < 209: + return None + return LegitimationType.LEGACY if firmware < 301 else LegitimationType.NEW + if "50-0XB0" in version_string.upper() and series == "2": # S7-1200 G2 + return LegitimationType.NEW + if series == "2": # S7-1200 + if firmware < 403: + return None + return LegitimationType.LEGACY if firmware < 407 else LegitimationType.NEW + if series == "6": # S7-1507S software controller + if firmware < 2109: + return None + return LegitimationType.LEGACY + return None + + def derive_legitimation_key(oms_secret: bytes) -> bytes: """Derive AES-256 key from TLS OMS exporter secret. @@ -90,45 +210,42 @@ def build_new_response( def _build_legitimation_payload(password: str, username: str = "") -> bytes: - """Build the legitimation payload structure. + """Build the plaintext payload that new-mode legitimation encrypts. - The payload is a serialized ValueStruct with: - - 40401: LegitimationType (1=legacy, 2=new) - - 40402: Username (UTF-8 blob) - - 40403: Password or password hash (SHA-1) - """ - from .vlq import encode_uint32_vlq - from .protocol import DataType - - result = bytearray() + An empty username selects legacy-style credentials, where the password travels as its SHA-1 hash. + Reference: thomas-v2/S7CommPlusDriver/Legitimation/Legitimation.cs + """ if username: - legit_type = 2 + legitimation_type = LegitimationType.NEW password_data = password.encode("utf-8") else: - legit_type = 1 + legitimation_type = LegitimationType.LEGACY password_data = hashlib.sha1(password.encode("utf-8")).digest() # noqa: S324 - username_data = username.encode("utf-8") - # Struct with 3 elements + result = bytearray() result += bytes([0x00, DataType.STRUCT]) - result += encode_uint32_vlq(3) + result += struct.pack(">I", Ids.LEGITIMATION_PAYLOAD_STRUCT) # Element 1: LegitimationType - result += bytes([0x00, DataType.UDINT]) - result += encode_uint32_vlq(legit_type) + result += encode_uint32_vlq(Ids.LEGITIMATION_PAYLOAD_TYPE) + result += bytes([0x00, DataType.UDINT]) + encode_uint32_vlq(legitimation_type) # Element 2: Username blob + result += encode_uint32_vlq(Ids.LEGITIMATION_PAYLOAD_USERNAME) + username_data = username.encode("utf-8") result += bytes([0x00, DataType.BLOB]) result += encode_uint32_vlq(len(username_data)) result += username_data # Element 3: Password blob + result += encode_uint32_vlq(Ids.LEGITIMATION_PAYLOAD_PASSWORD) result += bytes([0x00, DataType.BLOB]) result += encode_uint32_vlq(len(password_data)) result += password_data + result += bytes([0x00]) # list terminator return bytes(result) diff --git a/s7commplus/protocol.py b/s7commplus/protocol.py index 4d869630..3db7d4ed 100644 --- a/s7commplus/protocol.py +++ b/s7commplus/protocol.py @@ -210,6 +210,16 @@ class Ids(IntEnum): EFFECTIVE_PROTECTION_LEVEL = 1842 ACTIVE_PROTECTION_LEVEL = 1843 + # ServerSessionVersion struct element carrying the device "PAOM string" + # which selects the legitimation mode. + SESSION_VERSION_SYSTEM_PAOM_STRING = 319 + + # Struct and element IDs of the encrypted new-mode legitimation payload + LEGITIMATION_PAYLOAD_STRUCT = 40400 + LEGITIMATION_PAYLOAD_TYPE = 40401 + LEGITIMATION_PAYLOAD_USERNAME = 40402 + LEGITIMATION_PAYLOAD_PASSWORD = 40403 + # DB AccessArea base (add DB number to get area ID) DB_ACCESS_AREA_BASE = 0x8A0E0000 @@ -247,6 +257,16 @@ class AccessLevel(IntEnum): NO_ACCESS = 4 +class LegitimationType(IntEnum): + """Legitimation mode. + + Reference: thomas-v2/S7CommPlusDriver/Legitimation/LegitimationType.cs + """ + + LEGACY = 1 + NEW = 2 + + class LegitimationId(IntEnum): """Legitimation IDs used in password authentication (V2+). diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py index 93f78cdd..1e5dcf20 100644 --- a/tests/test_coverage_gaps.py +++ b/tests/test_coverage_gaps.py @@ -24,6 +24,7 @@ LegitimationState, build_legacy_response, ) +from s7commplus.protocol import AccessLevel # ============================================================================ @@ -118,13 +119,14 @@ def test_authenticate_no_tls_raises(self) -> None: with pytest.raises(S7ConnectionError, match="requires TLS"): conn.authenticate("password") - def test_authenticate_tls_but_no_oms_secret_raises(self) -> None: + def test_authenticate_tls_without_oms_secret_is_allowed(self) -> None: + """The OMS secret is a new-mode requirement, not a precondition of authenticate().""" conn = S7CommPlusConnection("127.0.0.1") conn._connected = True conn._tls_active = True conn._oms_secret = None - with pytest.raises(S7ConnectionError, match="requires TLS"): - conn.authenticate("password") + conn._protection_level = AccessLevel.FULL_ACCESS + conn.authenticate("password") def test_legacy_response_empty_password(self) -> None: """Empty password should still produce a valid 20-byte response.""" diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index a93efa6f..1e4a6f3e 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -25,6 +25,7 @@ _build_legitimation_payload, build_legacy_response, derive_legitimation_key, + extract_session_version_string, ) from s7commplus.protocol import ( READ_FUNCTION_CODES, @@ -33,6 +34,7 @@ FunctionCode, Ids, LegitimationId, + ObjectId, ProtocolVersion, ) from s7commplus.vlq import decode_uint32_vlq, encode_uint32_vlq @@ -131,6 +133,52 @@ def test_legacy_response_zero_challenge(self) -> None: expected = hashlib.sha1(password.encode("utf-8")).digest() # noqa: S324 assert response == expected + def test_legacy_response_matches_reference_driver(self) -> None: + """ + SHA-1(password) XOR challenge, against a vector computed by the C# driver. + The challenge is a genuine 20-byte challenge from an S7-1512 (FW V2.9). + """ + challenge = bytes.fromhex("7d8f8470d20590efc1d740416b4a073296bf463b") + response = build_legacy_response("foobar", challenge) + assert response == bytes.fromhex("f5cc5389f613b1f2283cf9229406e5b3b32c6e43") + + +class TestExtractSessionVersionString: + """Test PAOM string extraction from a raw ServerSessionVersion value.""" + + @pytest.mark.parametrize( + "paom_string", + [ + "1;6ES7 214-1AG40-0XB0 ;V4.5", # S7-1214C, trailing space + "1;6ES7 510-1DJ01-0AB0;V2.9", # S7-1510SP + "1;6ES7 672-7FC01-0YA0;V21.9", # S7-1507SF + ], + ) + def test_extracts_paom_string(self, paom_string: str) -> None: + """Device strings from thomas-v2/S7CommPlusDriver, plain and behind a decoy key.""" + text = paom_string.encode("utf-8") + header = bytes([0x00, DataType.STRUCT]) + struct.pack(">I", ObjectId.SERVER_SESSION_VERSION) + # [VLQ key][flags][WString][VLQ length][utf-8 text] + element = encode_uint32_vlq(Ids.SESSION_VERSION_SYSTEM_PAOM_STRING) + element += bytes([0x00, DataType.WSTRING]) + encode_uint32_vlq(len(text)) + text + + assert extract_session_version_string(header + element) == paom_string + + # The needle also matches payload bytes, so the search must continue past them. + decoy = encode_uint32_vlq(Ids.EFFECTIVE_PROTECTION_LEVEL) + bytes([0x00, DataType.UDINT]) + decoy += encode_uint32_vlq(Ids.SESSION_VERSION_SYSTEM_PAOM_STRING) + assert extract_session_version_string(header + decoy + element) == paom_string + + def test_returns_none_when_unusable(self) -> None: + """Value truncated past the end of the buffer, or no element 319 at all.""" + text = b"1;6ES7 510-1DJ01-0AB0;V2.9" + value = bytes([0x00, DataType.STRUCT]) + struct.pack(">I", ObjectId.SERVER_SESSION_VERSION) + value += encode_uint32_vlq(Ids.SESSION_VERSION_SYSTEM_PAOM_STRING) + value += bytes([0x00, DataType.WSTRING]) + encode_uint32_vlq(len(text)) + text + + assert extract_session_version_string(value[:-1]) is None + assert extract_session_version_string(value[:2]) is None + class TestLegitimationPayload: """Test legitimation payload building.""" @@ -148,18 +196,16 @@ def test_payload_with_username(self) -> None: def test_payload_legit_type_1_without_username(self) -> None: """Without username, legitimation type should be 1 (legacy).""" payload = _build_legitimation_payload("password") - # After struct header (flags=0x00, type=0x17, count VLQ), the first - # element is flags=0x00, type=UDInt(0x04), then legit_type value - # The exact structure: [0x00, 0x17, count, 0x00, 0x04, legit_type, ...] - # legit_type=1 is at offset 5 (VLQ encoded) - assert payload[4] == 0x04 # UDInt type for legit_type - assert payload[5] == 0x01 # legit_type = 1 + # [flags=0x00, type=0x17, struct id (4 bytes), key VLQ (3 bytes), + # flags=0x00, type=UDInt(0x04), legit_type VLQ] + assert payload[10] == 0x04 # UDInt type for legit_type + assert payload[11] == 0x01 # legit_type = 1 def test_payload_legit_type_2_with_username(self) -> None: """With username, legitimation type should be 2 (new).""" payload = _build_legitimation_payload("password", "admin") - assert payload[4] == 0x04 # UDInt type for legit_type - assert payload[5] == 0x02 # legit_type = 2 + assert payload[10] == 0x04 # UDInt type for legit_type + assert payload[11] == 0x02 # legit_type = 2 class TestLegitimationState: @@ -330,6 +376,88 @@ async def test_async_challenge_uses_protocol_request_shape(self) -> None: integrity_tail=4, ) + @pytest.mark.conformance + def test_challenge_request_frame(self) -> None: + challenge = bytes.fromhex("7d8f8470d20590efc1d740416b4a073296bf463b") + payload = bytes([0x00, 0x00, 0x10, DataType.USINT]) + encode_uint32_vlq(len(challenge)) + challenge + bytes([0x00]) + body = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.GET_VAR_SUBSTREAMED, 0, 6, 0x34) + payload + + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + conn._session_id = 0x70000CB7 + conn._sequence_number = 6 + conn._with_integrity_id = True + conn._integrity_id_read = 3 + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock( + return_value=encode_header(ProtocolVersion.V2, len(body)) + body + struct.pack(">BBH", 0x72, 0x02, 0) + ) + + assert conn._get_legitimation_challenge() == challenge + conn._send_s7_data.assert_called_once_with( + bytes.fromhex( + "72020035" # header, data length 0x35 + "310000058600000006" # request, GetVarSubStreamed, seq 6 + "70000cb734" # session id, transport flags + "70000cb7" # InObjectId + "200401822f" # address array header + id 303 + "000004e88969001200000000896a001300896b00040000" # ObjectQualifier + "0001" # unknown + "03" # IntegrityId (read) + "00000000" # fill + "72020000" # trailer + ) + ) + + @pytest.mark.conformance + def test_legitimation_request_frame(self) -> None: + response = bytes.fromhex("f5cc5389f613b1f2283cf9229406e5b3b32c6e43") + body = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.SET_VARIABLE, 0, 7, 0x34) + encode_uint32_vlq(0) + + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + conn._session_id = 0x70000CB7 + conn._sequence_number = 7 + conn._with_integrity_id = True + conn._integrity_id_write = 1 + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock( + return_value=encode_header(ProtocolVersion.V2, len(body)) + body + struct.pack(">BBH", 0x72, 0x02, 0) + ) + + conn._send_legitimation_legacy(response) + + conn._send_s7_data.assert_called_once_with( + bytes.fromhex( + "72020049" # header, data length 0x49 + "31000004f200000007" # request, SetVariable, seq 7 + "70000cb734" # session id, transport flags + "70000cb7" # InObjectId + "018230" # always-1, address id 304 + "100214" # USInt array of 20 + ) + + response + + bytes.fromhex( + "000004e88969001200000000896a001300896b00040000" # ObjectQualifier + "00" # unknown + "01" # IntegrityId (write) + "00000000" # fill + "72020000" # trailer + ) + ) + + @pytest.mark.conformance + def test_non_zero_return_value_is_accepted(self) -> None: + """The PLC signals success with a non-zero status word. Response captured from an S7-1512.""" + _check_set_variable_response(bytes.fromhex("9381b0808099a68019")) + + @pytest.mark.conformance + def test_refusal_is_invisible_in_the_return_value(self) -> None: + """A refused password is indistinguishable from an accepted one here. Response captured from an S7-1512.""" + _check_set_variable_response(bytes.fromhex("9381b390809aca8016")) + class TestProtectionLevel: """The effective protection level read that precedes legitimation."""