diff --git a/CHANGES.md b/CHANGES.md index 5c2b789f..62ddf448 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -9,6 +9,9 @@ Major release: new `s7commplus` package with S7CommPlus protocol support. * Correlate S7CommPlus responses by opcode, function, and sequence; discard bounded stale replies from earlier requests, preserve interleaved notifications, and serialize synchronous wire requests. +* Verify authenticated V3 responses and cumulative fragment digests before + parsing, invalidating the connection with a dedicated integrity error on any + mismatch or truncated envelope. * Decode corroborating CPU execution attributes so S7CommPlus `get_cpu_state()` distinguishes RUN from STOP on S7-1500 and returns UNKNOWN for absent or inconsistent state attributes, including S7-1200 responses that omit them. diff --git a/s7commplus/connection.py b/s7commplus/connection.py index bb80879d..49ef081a 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -284,20 +284,37 @@ def _set_s7_groups(ctx: ssl.SSLContext) -> None: ) -def _verify_v3_hmac(protected: bytes, session_key: bytes) -> bytes: - """Verify and remove the V3 HMAC prefix from application data.""" - from snap7.error import S7ConnectionError +def _verify_v3_hmac(protected: bytes, session_key: bytes, digest_state: hmac.HMAC | None = None) -> bytes: + """Verify and remove a V3 HMAC prefix. + + ``digest_state`` implements the legacy fragmented-response behavior: the + first digest covers the first fragment and each later digest covers all + application bytes accumulated so far. The state advances only after a + successful constant-time comparison. + """ + from snap7.error import S7IntegrityError if not protected: - raise S7ConnectionError("Empty V3 frame") + raise S7IntegrityError("Empty authenticated S7CommPlus V3 frame; reconnect before retrying") digest_length = protected[0] - if digest_length != hashlib.sha256().digest_size or len(protected) < 1 + digest_length: - raise S7ConnectionError(f"Invalid V3 HMAC length: {digest_length}") + expected_length = hashlib.sha256().digest_size + if digest_length != expected_length: + raise S7IntegrityError( + f"Invalid S7CommPlus V3 digest length {digest_length}, expected {expected_length}; reconnect before retrying" + ) + if len(protected) < 1 + digest_length: + raise S7IntegrityError( + f"Truncated S7CommPlus V3 digest: received {len(protected) - 1} of {digest_length} bytes; reconnect before retrying" + ) received_digest = protected[1 : 1 + digest_length] application_data = protected[1 + digest_length :] - expected_digest = hmac.new(session_key[:24], application_data, hashlib.sha256).digest() + verifier = digest_state.copy() if digest_state is not None else hmac.new(session_key[:24], digestmod=hashlib.sha256) + verifier.update(application_data) + expected_digest = verifier.digest() if not hmac.compare_digest(received_digest, expected_digest): - raise S7ConnectionError("Invalid V3 HMAC") + raise S7IntegrityError("S7CommPlus V3 response integrity check failed; reconnect before retrying") + if digest_state is not None: + digest_state.update(application_data) return bytes(application_data) @@ -845,19 +862,18 @@ def _send_legitimation_legacy(self, response: bytes) -> None: _check_set_variable_response(resp_payload) def collect_explore_frames(self, first_payload: bytes) -> bytes: - """Collect multi-fragment EXPLORE continuation frames for V3 PLCs. + """Collect unauthenticated multi-fragment EXPLORE continuation frames. On V3 PLCs (FW >= V4.5) a large EXPLORE response (e.g. RID 0x8A11FFFF) spans multiple TPKT frames. The first frame is the normal response (already stripped of its 10-byte header by send_request). Continuation - frames carry **no** response header — they are raw BLOB data protected - only by a V3 HMAC prefix. The caller must concatenate them before - parsing. + frames carry no response header. Authenticated callers must use + ``send_request(..., reassemble=True)`` because this legacy helper no + longer has the first frame bytes needed to verify cumulative digests. Termination: a ``frag_len == 0`` frame is the standard S7CommPlus - end-of-stream trailer. As a fallback, a frame whose body (after HMAC - strip) is measurably shorter than the first frame body is treated as the - last fragment (5-byte tolerance). + end-of-stream trailer. As a fallback, a measurably shorter frame body is + treated as the last fragment (5-byte tolerance). Collection is capped by ``_MAX_REASSEMBLED_FRAGMENTS`` and ``_MAX_REASSEMBLED_BYTES`` to prevent unbounded allocation on malformed @@ -870,6 +886,14 @@ def collect_explore_frames(self, first_payload: bytes) -> bytes: Returns: All fragment payloads concatenated (first_payload + continuations). """ + if self._session_key is not None: + from snap7.error import S7ProtocolError + + raise S7ProtocolError( + "Authenticated Explore continuations require send_request(..., reassemble=True) so cumulative digests " + "can be verified" + ) + # The first frame body (already header-stripped) was originally # len(first_payload) + 10 bytes on the wire (10-byte response header). # Continuation frames of the same "full" size will be that long after @@ -895,10 +919,6 @@ def collect_explore_frames(self, first_payload: bytes) -> bytes: if frag_len == 0: break # standard S7CommPlus end-of-stream trailer body = raw[4 : 4 + frag_len] - # V3 non-TLS: strip the HMAC prefix ([hash_len][hash_bytes]) - if self._protocol_version >= ProtocolVersion.V3 and len(body) > 33: - hash_len = body[0] - body = body[1 + hash_len :] if not body: break all_data += body @@ -943,6 +963,46 @@ def disconnect(self) -> None: self._notification_frames.clear() self._iso_conn.disconnect() + def _invalidate_integrity_failure(self) -> None: + """Close an untrusted stream without sending protocol data on it.""" + self._session_ready = False + self._session_id = 0 + self.disconnect() + + def _verify_v3_hmac(self, protected: bytes, digest_state: hmac.HMAC | None = None) -> bytes: + """Verify authenticated data and make any failure terminal for this connection.""" + from snap7.error import S7IntegrityError + + if self._session_key is None: + self._invalidate_integrity_failure() + raise S7IntegrityError("Authenticated S7CommPlus V3 response arrived without a session key; reconnect") + try: + return _verify_v3_hmac(protected, self._session_key, digest_state) + except S7IntegrityError: + self._invalidate_integrity_failure() + raise + + def _verify_v3_frame(self, frame: bytes) -> None: + """Verify a complete V3 frame before its opcode is inspected or queued.""" + from snap7.error import S7IntegrityError + + version, data_length, consumed = decode_header(frame) + if version != ProtocolVersion.V3: + if self._session_key is not None: + self._invalidate_integrity_failure() + raise S7IntegrityError( + f"Authenticated S7CommPlus response used unauthenticated frame version {version}; reconnect" + ) + return + frame_end = consumed + data_length + if len(frame) < frame_end: + self._invalidate_integrity_failure() + raise S7IntegrityError( + f"Truncated authenticated S7CommPlus V3 frame: declared {data_length} data bytes, " + f"received {max(0, len(frame) - consumed)}; reconnect before retrying" + ) + self._verify_v3_hmac(bytes(frame[consumed:frame_end])) + def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: int = 4, reassemble: bool = False) -> bytes: """Serialize one request/response exchange on the connection.""" with self._request_lock: @@ -1067,11 +1127,7 @@ def _send_request(self, function_code: int, payload: bytes, integrity_tail: int, # V3 responses have a hash-length byte + HMAC prefix before the payload. if version == ProtocolVersion.V3: - if self._session_key is None: - from snap7.error import S7ConnectionError - - raise S7ConnectionError("V3 response received without a session key") - response = _verify_v3_hmac(response, self._session_key) + response = self._verify_v3_hmac(response) logger.debug(" V3 HMAC verified") logger.debug(f" Response data ({len(response)} bytes): {response.hex(' ')}") @@ -1126,6 +1182,7 @@ def _recv_response_frame(self, expected_sequence: Optional[int] = None) -> bytes if system_events > _MAX_SYSTEM_EVENTS_PER_RESPONSE: raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response") continue + self._verify_v3_frame(response_frame) if data_length < 10: return response_frame opcode = _incoming_frame_opcode(response_frame) @@ -1173,6 +1230,7 @@ def receive_notification(self) -> bytes: raise S7ConnectionError("Not connected") frame = self._notification_frames.popleft() if self._notification_frames else self._recv_s7_data() + self._verify_v3_frame(frame) if not self._is_notification_frame(frame): from snap7.error import S7ConnectionError @@ -1206,11 +1264,27 @@ def ensure(n: int) -> None: data = bytearray() fragments = 0 + expected_version: int | None = None + digest_state = hmac.new(self._session_key[:24], digestmod=hashlib.sha256) if self._session_key is not None else None while True: ensure(4) if buf[0] != 0x72: raise S7ConnectionError("Expected S7CommPlus fragment header (0x72)") fragment_version = buf[1] + if expected_version is None: + expected_version = fragment_version + elif fragment_version != expected_version: + if self._session_key is not None: + from snap7.error import S7IntegrityError + + self._invalidate_integrity_failure() + raise S7IntegrityError( + f"Authenticated S7CommPlus response changed fragment version from {expected_version} " + f"to {fragment_version}; reconnect" + ) + raise S7ConnectionError( + f"S7CommPlus response changed fragment version from {expected_version} to {fragment_version}" + ) frag_len = (buf[2] << 8) | buf[3] del buf[:4] if frag_len == 0: @@ -1219,9 +1293,7 @@ def ensure(n: int) -> None: fragment_data = bytes(buf[:frag_len]) del buf[:frag_len] if fragment_version == ProtocolVersion.V3: - if self._session_key is None: - raise S7ConnectionError("V3 response received without a session key") - fragment_data = _verify_v3_hmac(fragment_data, self._session_key) + fragment_data = self._verify_v3_hmac(fragment_data, digest_state) data.extend(fragment_data) fragments += 1 if fragments > self._MAX_REASSEMBLED_FRAGMENTS or len(data) > self._MAX_REASSEMBLED_BYTES: @@ -1229,7 +1301,7 @@ def ensure(n: int) -> None: # The next 4 bytes are either the trailer (0x72 ver 0x0000) or the next # fragment's header (0x72 ver len>0). ensure(4) - if buf[0] == 0x72 and buf[2] == 0 and buf[3] == 0: + if buf[0] == 0x72 and buf[1] == expected_version and buf[2] == 0 and buf[3] == 0: del buf[:4] # consume trailer — last fragment break return bytes(data) diff --git a/snap7/error.py b/snap7/error.py index 16108475..aca1bd1d 100644 --- a/snap7/error.py +++ b/snap7/error.py @@ -28,6 +28,12 @@ class S7ProtocolError(S7Error): pass +class S7IntegrityError(S7ProtocolError): + """Raised when authenticated S7CommPlus traffic fails integrity checks.""" + + pass + + class S7TimeoutError(S7Error): """Raised when S7 operation times out.""" diff --git a/tests/test_error.py b/tests/test_error.py index 7e32f9e4..8792a149 100644 --- a/tests/test_error.py +++ b/tests/test_error.py @@ -6,6 +6,7 @@ S7Error, S7ConnectionError, S7ProtocolError, + S7IntegrityError, S7TimeoutError, S7AuthenticationError, S7StalePacketError, @@ -33,6 +34,7 @@ def test_s7error_without_code(self) -> None: def test_subclass_hierarchy(self) -> None: assert issubclass(S7ConnectionError, S7Error) assert issubclass(S7ProtocolError, S7Error) + assert issubclass(S7IntegrityError, S7ProtocolError) assert issubclass(S7TimeoutError, S7Error) assert issubclass(S7AuthenticationError, S7Error) assert issubclass(S7StalePacketError, S7ProtocolError) @@ -42,6 +44,7 @@ def test_all_subclasses_instantiate(self) -> None: for cls in ( S7ConnectionError, S7ProtocolError, + S7IntegrityError, S7TimeoutError, S7AuthenticationError, S7StalePacketError, diff --git a/tests/test_s7_legacy_request_layout.py b/tests/test_s7_legacy_request_layout.py index 2769d61e..a2367da0 100644 --- a/tests/test_s7_legacy_request_layout.py +++ b/tests/test_s7_legacy_request_layout.py @@ -33,7 +33,9 @@ def test_v1_substreamed_request_matches_accepted_tia_packet() -> None: conn._integrity_id_read = 1 conn._send_s7_data = MagicMock() body = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.GET_VAR_SUBSTREAMED, 0, 3, 0x34) + bytes(4) - conn._recv_s7_data = MagicMock(return_value=encode_header(ProtocolVersion.V2, len(body)) + body) + response_digest = hmac.new(conn._session_key, body, hashlib.sha256).digest() + protected_body = bytes([len(response_digest)]) + response_digest + body + conn._recv_s7_data = MagicMock(return_value=encode_header(ProtocolVersion.V3, len(protected_body)) + protected_body) conn.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload) frame = conn._send_s7_data.call_args.args[0] assert frame[37:-4] == expected diff --git a/tests/test_s7_server.py b/tests/test_s7_server.py index e6bfd701..89e2ad39 100644 --- a/tests/test_s7_server.py +++ b/tests/test_s7_server.py @@ -9,7 +9,7 @@ import pytest -from snap7.error import S7ConnectionError +from snap7.error import S7ConnectionError, S7IntegrityError from s7commplus.async_client import S7CommPlusAsyncClient from s7commplus.client import S7CommPlusClient from s7commplus.connection import _parse_get_var_substreamed_response, _verify_v3_hmac @@ -489,7 +489,7 @@ def test_v3_hmac_verification(self) -> None: tampered = protected[:-1] + bytes([protected[-1] ^ 0x01]) with pytest.raises(ConnectionError, match="Invalid V3 HMAC"): S7CommPlusServer._verify_v3_data(tampered, TEST_SESSION_KEY) - with pytest.raises(S7ConnectionError, match="Invalid V3 HMAC"): + with pytest.raises(S7IntegrityError, match="integrity check failed"): _verify_v3_hmac(tampered, TEST_SESSION_KEY) def test_create_object_response_contains_fingerprint_and_challenge(self) -> None: diff --git a/tests/test_s7_subscription.py b/tests/test_s7_subscription.py index 5ff62f16..d180c740 100644 --- a/tests/test_s7_subscription.py +++ b/tests/test_s7_subscription.py @@ -1,12 +1,14 @@ """Tests for S7CommPlus symbolic data subscriptions.""" +import hashlib +import hmac import struct from unittest.mock import MagicMock import pytest from s7commplus.client import S7CommPlusClient -from s7commplus.codec import encode_header, encode_pvalue_blob +from s7commplus.codec import decode_header, encode_header, encode_pvalue_blob from s7commplus.connection import S7CommPlusConnection from s7commplus.protocol import DataType, FunctionCode, Ids, Opcode, ProtocolVersion from s7commplus.subscription import ( @@ -16,6 +18,7 @@ parse_subscription_notification, ) from s7commplus.vlq import encode_uint32_vlq, encode_uint64_vlq +from snap7.error import S7IntegrityError def _response_frame(function_code: int, sequence: int, payload: bytes) -> bytes: @@ -192,3 +195,26 @@ def test_send_request_queues_interleaved_notification(self) -> None: assert connection.send_request(FunctionCode.GET_VARIABLE, b"\x00\x00\x00\x00") == b"\x00" assert connection.receive_notification() == notification assert connection._recv_s7_data.call_count == 2 + + def test_authenticated_notification_is_verified_before_queueing(self) -> None: + connection = S7CommPlusConnection("127.0.0.1") + connection._connected = True + connection._protocol_version = ProtocolVersion.V3 + connection._session_id = 1 + connection._session_key = bytes(range(24)) + connection._iso_conn.disconnect = MagicMock() + + notification = _notification_frame(version=ProtocolVersion.V3) + _, data_length, consumed = decode_header(notification) + data = notification[consumed : consumed + data_length] + digest = hmac.new(connection._session_key, data, hashlib.sha256).digest() + protected = bytearray(bytes([len(digest)]) + digest + data) + protected[1] ^= 1 + notification = encode_header(ProtocolVersion.V3, len(protected)) + protected + connection._send_s7_data = MagicMock() + connection._recv_s7_data = MagicMock(return_value=bytes(notification)) + + with pytest.raises(S7IntegrityError, match="integrity check failed"): + connection.send_request(FunctionCode.GET_VARIABLE, bytes(4)) + assert not connection.connected + assert not connection._notification_frames diff --git a/tests/test_s7_unit.py b/tests/test_s7_unit.py index e4673075..83a3ff71 100644 --- a/tests/test_s7_unit.py +++ b/tests/test_s7_unit.py @@ -3,7 +3,7 @@ import hashlib import hmac import struct -from unittest.mock import MagicMock, call +from unittest.mock import MagicMock, call, patch import pytest @@ -22,7 +22,7 @@ _build_symbolic_write_payload, _build_substreamed_write_payload, ) -from s7commplus.connection import S7CommPlusConnection, _strip_paom_string_in_session_version +from s7commplus.connection import S7CommPlusConnection, _strip_paom_string_in_session_version, _verify_v3_hmac from s7commplus.codec import encode_header, encode_object_qualifier, encode_pvalue_blob from s7commplus.codec import _pvalue_element_size as _element_size from s7commplus.codec import skip_typed_value, parse_server_session_version @@ -745,18 +745,53 @@ def test_multiple_fragments_split_across_reads(self) -> None: conn = self._conn_yielding([self._frag(b"abc"), self._frag(b"de"), self._TRAILER]) assert conn._recv_reassembled_payload() == b"abcde" - def test_v3_session_key_hmac_is_stripped_from_each_fragment(self) -> None: + def test_v3_session_key_hmac_uses_cumulative_fragment_digest(self) -> None: conn = self._conn_yielding([]) conn._session_key = bytes(24) + digest_state = hmac.new(conn._session_key, digestmod=hashlib.sha256) def v3_frag(data: bytes) -> bytes: - digest = hmac.new(conn._session_key, data, hashlib.sha256).digest() + digest_state.update(data) + digest = digest_state.digest() protected = bytes([len(digest)]) + digest + data return bytes([0x72, ProtocolVersion.V3, 0, len(protected)]) + protected initial = v3_frag(b"abc") + v3_frag(b"de") + bytes([0x72, ProtocolVersion.V3, 0, 0]) assert conn._recv_reassembled_payload(initial) == b"abcde" + def test_v3_cumulative_fragment_rejects_independent_second_digest(self) -> None: + from snap7.error import S7IntegrityError + + conn = self._conn_yielding([]) + conn._connected = True + conn._session_key = bytes(24) + + def independent_fragment(data: bytes) -> bytes: + digest = hmac.new(conn._session_key, data, hashlib.sha256).digest() + protected = bytes([len(digest)]) + digest + data + return bytes([0x72, ProtocolVersion.V3, 0, len(protected)]) + protected + + initial = independent_fragment(b"abc") + independent_fragment(b"de") + with pytest.raises(S7IntegrityError, match="integrity check failed"): + conn._recv_reassembled_payload(initial) + assert not conn.connected + + def test_authenticated_reassembly_rejects_fragment_version_downgrade(self) -> None: + from snap7.error import S7IntegrityError + + conn = self._conn_yielding([]) + conn._connected = True + conn._session_key = bytes(24) + first_data = b"abc" + digest = hmac.new(conn._session_key, first_data, hashlib.sha256).digest() + first_protected = b"\x20" + digest + first_data + first = encode_header(ProtocolVersion.V3, len(first_protected)) + first_protected + downgraded = bytes([0x72, ProtocolVersion.V2, 0, 2]) + b"de" + + with pytest.raises(S7IntegrityError, match="changed fragment version"): + conn._recv_reassembled_payload(first + downgraded) + assert not conn.connected + def test_bad_fragment_header_raises(self) -> None: from snap7.error import S7ConnectionError @@ -779,3 +814,81 @@ def test_fragment_count_cap(self) -> None: conn._MAX_REASSEMBLED_FRAGMENTS = 2 with pytest.raises(S7ConnectionError, match="exceeds limits"): conn._recv_reassembled_payload() + + +class TestV3ResponseIntegrity: + KEY = bytes(range(24)) + + @classmethod + def _protected(cls, data: bytes, key: bytes | None = None) -> bytes: + digest = hmac.new(key or cls.KEY, data, hashlib.sha256).digest() + return bytes([len(digest)]) + digest + data + + def test_valid_digest_uses_constant_time_comparison(self) -> None: + protected = self._protected(b"authenticated response") + with patch("s7commplus.connection.hmac.compare_digest", wraps=hmac.compare_digest) as compare: + assert _verify_v3_hmac(protected, self.KEY) == b"authenticated response" + compare.assert_called_once() + + @pytest.mark.parametrize("mutation", ["wrong-key", "payload", "digest"]) + def test_changed_digest_covered_data_is_rejected(self, mutation: str) -> None: + from snap7.error import S7IntegrityError + + signing_key = bytes(reversed(self.KEY)) if mutation == "wrong-key" else None + protected = self._protected(b"authenticated response", signing_key) + if mutation == "payload": + protected = protected[:-1] + bytes([protected[-1] ^ 1]) + elif mutation == "digest": + protected = protected[:1] + bytes([protected[1] ^ 1]) + protected[2:] + with pytest.raises(S7IntegrityError, match="integrity check failed"): + _verify_v3_hmac(protected, self.KEY) + + @pytest.mark.parametrize( + "protected, message", + [(b"", "Empty authenticated"), (b"\x1f" + bytes(31), "digest length"), (b"\x20" + bytes(12), "Truncated")], + ) + def test_invalid_or_truncated_digest_is_rejected(self, protected: bytes, message: str) -> None: + from snap7.error import S7IntegrityError + + with pytest.raises(S7IntegrityError, match=message): + _verify_v3_hmac(protected, self.KEY) + + def test_failure_invalidates_connection(self) -> None: + from snap7.error import S7IntegrityError + + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._session_ready = True + conn._session_id = 123 + conn._session_key = self.KEY + conn._iso_conn.disconnect = MagicMock() + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + protected = bytearray(self._protected(response)) + protected[1] ^= 1 + frame = encode_header(ProtocolVersion.V3, len(protected)) + protected + conn._recv_s7_data = MagicMock(return_value=bytes(frame)) + conn._send_s7_data = MagicMock() + + with pytest.raises(S7IntegrityError, match="integrity check failed"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + assert not conn.connected + assert conn._session_key is None + conn._iso_conn.disconnect.assert_called_once_with() + + def test_authenticated_response_rejects_frame_version_downgrade(self) -> None: + from snap7.error import S7IntegrityError + + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._session_ready = True + conn._session_id = 123 + conn._session_key = self.KEY + conn._iso_conn.disconnect = MagicMock() + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + frame = encode_header(ProtocolVersion.V2, len(response)) + response + conn._recv_s7_data = MagicMock(return_value=frame) + conn._send_s7_data = MagicMock() + + with pytest.raises(S7IntegrityError, match="unauthenticated frame version"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + assert not conn.connected