diff --git a/snap7/server/__init__.py b/snap7/server/__init__.py index f9b43686..1a652cc7 100644 --- a/snap7/server/__init__.py +++ b/snap7/server/__init__.py @@ -2705,10 +2705,9 @@ def _build_cotp_cc(self) -> bytes: pdu_size_param = struct.pack(">BBB", self.COTP_PARAM_PDU_SIZE, 1, self.tpdu_size) pdu_length = 6 + len(pdu_size_param) base_pdu = struct.pack( - ">BBBHHB", + ">BBHHB", pdu_length, # PDU length self.COTP_CC, # PDU type - 0x00, # Reserved / CDT self.dst_ref, # Destination reference (client's source ref) self.src_ref, # Source reference (our ref) 0x00, # Class/option diff --git a/tests/test_server.py b/tests/test_server.py index b7af39a3..6928bfc8 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -316,6 +316,17 @@ def test_server_area_management(self) -> None: @pytest.mark.server class TestServerISOConnectionLimits: + def test_connection_confirm_has_valid_length_and_tpdu_size(self) -> None: + client_socket = MagicMock() + connection = ServerISOConnection(client_socket) + connection.dst_ref = 0x000F + connection.tpdu_size = 0x09 + + connection_confirm = connection._build_cotp_cc() + + assert connection_confirm == bytes.fromhex("09d0000f000100c00109") + assert connection_confirm[0] == len(connection_confirm) - 1 + def test_partial_frame_timeout_closes_connection(self) -> None: client_socket = MagicMock() client_socket.recv.side_effect = [b"\x03", TimeoutError()]