diff --git a/s7commplus/server.py b/s7commplus/server.py index 08a060f8..3fe01718 100644 --- a/s7commplus/server.py +++ b/s7commplus/server.py @@ -668,25 +668,38 @@ def _process_request( request_data = payload[14:] if function_code == FunctionCode.INIT_SSL: - return self._handle_init_ssl(seq_num), False + response = self._handle_init_ssl(seq_num) + rst = False elif function_code == FunctionCode.CREATE_OBJECT: - return self._handle_create_object(seq_num, request_data), False + response = self._handle_create_object(seq_num, request_data) + rst = False elif function_code == FunctionCode.DELETE_OBJECT: - return self._handle_delete_object(seq_num, req_session_id), False + response = self._handle_delete_object(seq_num, req_session_id) + rst = False elif function_code == FunctionCode.EXPLORE: - return self._handle_explore(seq_num, req_session_id, request_data), False + response = self._handle_explore(seq_num, req_session_id, request_data) + rst = False elif function_code == FunctionCode.GET_MULTI_VARIABLES: - resp = self._handle_get_multi_variables(seq_num, req_session_id, request_data) + response = self._handle_get_multi_variables(seq_num, req_session_id, request_data) rst = self._rst_after_symbolic_read and session_id != 0 - return resp, rst elif function_code == FunctionCode.SET_MULTI_VARIABLES: - return self._handle_set_multi_variables(seq_num, req_session_id, request_data), False + response = self._handle_set_multi_variables(seq_num, req_session_id, request_data) + rst = False elif function_code == FunctionCode.GET_VAR_SUBSTREAMED: - return self._handle_get_var_substreamed(seq_num, req_session_id, request_data), False + response = self._handle_get_var_substreamed(seq_num, req_session_id, request_data) + rst = False elif function_code == FunctionCode.SET_VAR_SUBSTREAMED: - return self._handle_set_var_substreamed(seq_num, req_session_id, request_data), False + response = self._handle_set_var_substreamed(seq_num, req_session_id, request_data) + rst = False else: - return self._build_error_response(seq_num, req_session_id, function_code), False + response = self._build_error_response(seq_num, req_session_id, function_code) + rst = False + + if self._protocol_version >= ProtocolVersion.V2 and req_session_id != 0: + integrity_id = integrity_id_read if function_code in READ_FUNCTION_CODES else integrity_id_write + response += encode_uint32_vlq(integrity_id) + + return response, rst def _build_response_header(self, function_code: int, seq_num: int) -> bytes: """Build a 10-byte S7CommPlus data-response header. @@ -694,8 +707,8 @@ def _build_response_header(self, function_code: int, seq_num: int) -> bytes: Unlike requests (which carry a 4-byte SessionId, giving a 14-byte header), real S7-1500 *responses* omit the SessionId field, so the data header is 10 bytes: opcode + reserved + function + reserved + - seqnr + transport. For V2+, the IntegrityId travels at the *end* of - the payload (appended by the individual handlers), not in the header. + seqnr + transport. For V2+, _process_request appends the IntegrityId + at the *end* of the payload, not in the header. Args: function_code: Response function code @@ -951,8 +964,10 @@ def _handle_get_multi_variables(self, seq_num: int, session_id: int, request_dat # Terminate error list response += encode_uint32_vlq(0) - # IntegrityId - response += encode_uint32_vlq(0) + # V1 responses retain the legacy zero IntegrityId field. V2+ responses + # receive the current per-client counter in _process_request(). + if self._protocol_version < ProtocolVersion.V2: + response += encode_uint32_vlq(0) return bytes(response) @@ -974,7 +989,8 @@ def _handle_set_multi_variables(self, seq_num: int, session_id: int, request_dat logger.debug("SetMultiVariables: accepting session setup write") response += encode_uint64_vlq(0) # ReturnValue: success response += encode_uint32_vlq(0) # Empty error list - response += encode_uint32_vlq(0) # IntegrityId + if self._protocol_version < ProtocolVersion.V2: + response += encode_uint32_vlq(0) # Legacy V1 IntegrityId return bytes(response) # Parse request payload for DB writes @@ -1000,8 +1016,8 @@ def _handle_set_multi_variables(self, seq_num: int, session_id: int, request_dat # Terminate error list response += encode_uint32_vlq(0) - # IntegrityId - response += encode_uint32_vlq(0) + if self._protocol_version < ProtocolVersion.V2: + response += encode_uint32_vlq(0) # Legacy V1 IntegrityId return bytes(response) @@ -1050,7 +1066,10 @@ def _handle_get_var_substreamed(self, seq_num: int, session_id: int, request_dat response += bytes([0x10, DataType.USINT]) response += encode_uint32_vlq(0) - response += encode_uint32_vlq(0) # IntegrityId + # V1 responses retain the legacy zero IntegrityId field. V2+ responses + # receive the current per-client counter in _process_request(). + if self._protocol_version < ProtocolVersion.V2: + response += encode_uint32_vlq(0) return bytes(response) def _handle_set_var_substreamed(self, seq_num: int, session_id: int, request_data: bytes) -> bytes: diff --git a/tests/test_s7_tls.py b/tests/test_s7_tls.py index 39828fc7..58d3bf7c 100644 --- a/tests/test_s7_tls.py +++ b/tests/test_s7_tls.py @@ -217,6 +217,29 @@ async def test_integrity_id_tracking_enabled(self, tls_server: tuple[S7CommPlusS finally: await client.disconnect() + @pytest.mark.asyncio + async def test_sequential_reads_and_writes_keep_integrity_counters_in_sync( + self, tls_server: tuple[S7CommPlusServer, str, str] + ) -> None: + """Multiple V2 operations should succeed as both counters advance.""" + _, cert_path, _ = tls_server + + client = S7CommPlusAsyncClient() + await client.connect("127.0.0.1", port=TEST_PORT_V2_TLS, use_tls=True, tls_ca=cert_path) + + try: + initial_read_id = client._integrity_id_read + initial_write_id = client._integrity_id_write + + for value in (b"first", b"second", b"third"): + await client.db_write(1, 0, value) + assert await client.db_read(1, 0, len(value)) == value + + assert client._integrity_id_read == initial_read_id + 3 + assert client._integrity_id_write == initial_write_id + 3 + finally: + await client.disconnect() + @pytest.mark.asyncio async def test_protocol_version_is_v2(self, tls_server: tuple[S7CommPlusServer, str, str]) -> None: """V2 server should report protocol version 2.""" diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index 2db8a36c..6a0d8ad4 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -41,6 +41,7 @@ Opcode, ProtocolVersion, ) +from s7commplus.server import S7CommPlusServer from s7commplus.vlq import decode_uint32_vlq, encode_uint32_vlq from snap7.error import S7ConnectionError @@ -315,6 +316,90 @@ def test_tls_v2_response_application_payload_is_not_stripped(self) -> None: assert conn._send_s7_data.call_args[0][0][17] == 0x34 +class TestServerResponseIntegrityId: + """Test V2 response IntegrityId selection and encoding.""" + + @staticmethod + def _request(function_code: int) -> bytes: + request = struct.pack( + ">BHHHHIB", + Opcode.REQUEST, + 0, + function_code, + 0, + 1, + 0x12345678, + 0x34, + ) + return encode_header(ProtocolVersion.V2, len(request)) + request + + @pytest.mark.parametrize( + ("function_code", "expected_integrity_id"), + [ + (FunctionCode.GET_MULTI_VARIABLES, 128), + (FunctionCode.EXPLORE, 128), + (FunctionCode.GET_VAR_SUBSTREAMED, 128), + (FunctionCode.SET_MULTI_VARIABLES, 16384), + (FunctionCode.SET_VAR_SUBSTREAMED, 16384), + (FunctionCode.DELETE_OBJECT, 16384), + ], + ) + def test_v2_response_appends_function_counter(self, function_code: int, expected_integrity_id: int) -> None: + server = S7CommPlusServer(protocol_version=ProtocolVersion.V2) + request = self._request(function_code) + + initial_response, initial_rst = server._process_request(request, 0x12345678) + advanced_response, advanced_rst = server._process_request( + request, 0x12345678, integrity_id_read=128, integrity_id_write=16384 + ) + + assert initial_response is not None + assert advanced_response is not None + assert advanced_response == initial_response[:-1] + encode_uint32_vlq(expected_integrity_id) + assert not initial_rst + assert not advanced_rst + + def test_v1_response_keeps_legacy_integrity_field(self) -> None: + server = S7CommPlusServer(protocol_version=ProtocolVersion.V1) + request = self._request(FunctionCode.GET_MULTI_VARIABLES) + + initial_response, initial_rst = server._process_request(request, 0x12345678) + advanced_response, advanced_rst = server._process_request(request, 0x12345678, integrity_id_read=128) + + assert advanced_response == initial_response + assert not initial_rst + assert not advanced_rst + + def test_v2_substreamed_response_has_one_integrity_id(self) -> None: + server = S7CommPlusServer(protocol_version=ProtocolVersion.V2) + request = self._request(FunctionCode.GET_VAR_SUBSTREAMED) + + response, rst = server._process_request(request, 0x12345678, integrity_id_read=128) + + assert response == server._handle_get_var_substreamed(1, 0x12345678, b"") + encode_uint32_vlq(128) + assert not rst + + def test_init_ssl_response_has_no_integrity_id(self) -> None: + server = S7CommPlusServer(protocol_version=ProtocolVersion.V2) + request = bytearray(self._request(FunctionCode.INIT_SSL)) + request[13:17] = bytes(4) # InitSSL runs before a session id exists. + + response, rst = server._process_request(bytes(request), 0, integrity_id_write=128) + + assert response == server._handle_init_ssl(1) + assert not rst + + def test_in_session_error_response_uses_write_integrity_id(self) -> None: + server = S7CommPlusServer(protocol_version=ProtocolVersion.V2) + unsupported_function = 0xFFFF + request = self._request(unsupported_function) + + response, rst = server._process_request(request, 0x12345678, integrity_id_write=128) + + assert response == server._build_error_response(1, 0x12345678, unsupported_function) + encode_uint32_vlq(128) + assert not rst + + class TestIntegrityIdVlqEncoding: """Test VLQ encoding used for IntegrityId values."""