From 37f32fe00d71a248ae58d3df34cbe607f42a4761 Mon Sep 17 00:00:00 2001 From: NOKIAO <675925864@qq.com> Date: Fri, 9 Oct 2026 15:24:36 +0800 Subject: [PATCH 1/3] Fix functional addressing with configurable ECU response addresses --- doipclient/client.py | 28 ++++++++++++++++++++----- tests/test_client.py | 50 +++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 70 insertions(+), 8 deletions(-) diff --git a/doipclient/client.py b/doipclient/client.py index 0744772..dfb7c5c 100644 --- a/doipclient/client.py +++ b/doipclient/client.py @@ -149,6 +149,11 @@ class DoIPClient: :type auto_reconnect_tcp: bool :param vm_specific: Optional 4 byte long int :type vm_specific: int, optional + :param allowed_response_addresses: Allowed source logical addresses for diagnostic responses. When omitted or empty, + only the current ecu_logical_address is accepted. Set this to ECU physical addresses when sending to a functional + address. ACK/NACK messages may originate from the request target or any address in this collection. + The response target must always match client_logical_address. + :type allowed_response_addresses: Collection[int], optional :raises ConnectionRefusedError: If the activation request fails :raises ValueError: If the IPAddress is neither an IPv4 nor an IPv6 address @@ -167,8 +172,10 @@ def __init__( use_secure=False, auto_reconnect_tcp=False, vm_specific=None, + allowed_response_addresses=None, ): self._ecu_logical_address = ecu_logical_address + self._allowed_response_addresses = allowed_response_addresses self._client_logical_address = client_logical_address self._client_ip_address = client_ip_address self._use_secure = use_secure @@ -728,6 +735,14 @@ def send_diagnostic(self, diagnostic_payload, timeout=A_PROCESSING_TIME): self._ecu_logical_address, diagnostic_payload, timeout ) + def _get_allowed_response_addresses(self, address=None): + response_addresses = self._allowed_response_addresses or { + self._ecu_logical_address + } + if address is not None: + return {address}.union(response_addresses) + return response_addresses + def send_diagnostic_to_address( self, address, diagnostic_payload, timeout=A_PROCESSING_TIME ): @@ -754,7 +769,7 @@ def send_diagnostic_to_address( result = self.read_doip() if type(result) == DiagnosticMessageNegativeAcknowledgement: if ( - result.source_address == address + result.source_address in self._get_allowed_response_addresses(address) and result.target_address == self._client_logical_address ): raise IOError( @@ -768,7 +783,7 @@ def send_diagnostic_to_address( ) elif type(result) == DiagnosticMessagePositiveAcknowledgement: if ( - result.source_address == address + result.source_address in self._get_allowed_response_addresses(address) and result.target_address == self._client_logical_address ): return @@ -786,6 +801,9 @@ def send_diagnostic_to_address( def receive_diagnostic(self, timeout=None): """Receive a raw diagnostic payload (ie: UDS) from the ECU. + Only source addresses in allowed_response_addresses are accepted, or ecu_logical_address + when that collection was not configured or is empty. The target must match client_logical_address. + :return: Raw UDS payload :rtype: bytearray :raises TimeoutError: No diagnostic response received in time @@ -801,16 +819,16 @@ def receive_diagnostic(self, timeout=None): result = self.read_doip() if type(result) == DiagnosticMessage: if ( - result.source_address == self._ecu_logical_address + result.source_address in self._get_allowed_response_addresses() and result.target_address == self._client_logical_address ): return result.user_data elif ( - result.source_address != self._ecu_logical_address + result.source_address not in self._get_allowed_response_addresses() and result.target_address == self._client_logical_address ): logger.warning( - "Received DiagnosticMessage with expected target address, but source address doesn't match expected ECU logical address. Ignoring." + "Received DiagnosticMessage with expected target address, but source address is not allowed. Ignoring." ) start_time = time.time() else: diff --git a/tests/test_client.py b/tests/test_client.py index 11c700d..8728c92 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -730,9 +730,16 @@ def test_send_diagnostic_to_address_times_out_when_only_invalid_target_ack_nack( assert mock_socket.tx_queue[-1] == diagnostic_request_to_address -def test_receive_diagnostic(mock_socket): - sut = DoIPClient(test_ip, test_logical_address) - mock_socket.rx_queue.append(diagnostic_result) +@pytest.mark.parametrize("allowed_addresses", [None, {}]) +@pytest.mark.parametrize("ecu_address", [1, 2]) +def test_receive_diagnostic(mock_socket, allowed_addresses, ecu_address): + sut = DoIPClient( + test_ip, test_logical_address, allowed_response_addresses=allowed_addresses + ) + sut._ecu_logical_address = ecu_address + response = bytearray(diagnostic_result) + response[8:10] = ecu_address.to_bytes(2, "big") + mock_socket.rx_queue.append(response) result = sut.receive_diagnostic() assert result == bytearray([0, 1, 2, 3]) @@ -743,6 +750,43 @@ def test_receive_diagnostic_timeout(mock_socket): sut.receive_diagnostic(timeout=0.01) +@pytest.mark.parametrize("source_address", [1, 2]) +def test_functional_diagnostic_accepts_allowed_responses(mock_socket, source_address): + sut = DoIPClient(test_ip, 0xE400, allowed_response_addresses={1, 2}) + ack, response = bytearray(diagnostic_ack), bytearray(diagnostic_result) + ack[8:10] = response[8:10] = source_address.to_bytes(2, "big") + mock_socket.rx_queue.extend([ack, response]) + sut.send_diagnostic(bytearray([0, 1, 2]), timeout=0.01) + assert mock_socket.tx_queue[-1][10:12] == b"\xe4\x00" + assert sut.receive_diagnostic(timeout=0.01) == bytearray([0, 1, 2, 3]) + + +def test_functional_diagnostic_accepts_allowed_nack(mock_socket): + sut = DoIPClient(test_ip, 0xE400, allowed_response_addresses={1}) + mock_socket.rx_queue.append(diagnostic_nack) + with pytest.raises(IOError, match="Diagnostic request rejected"): + sut.send_diagnostic(bytearray([0, 1, 2]), timeout=0.01) + + +def test_functional_diagnostic_accepts_default_response_address(mock_socket): + sut = DoIPClient(test_ip, test_logical_address) + sut._ecu_logical_address = 2 + mock_socket.rx_queue.append(diagnostic_ack_invalid_source) + assert sut.send_diagnostic_to_address(0xE400, bytearray([0, 1, 2]), timeout=0.01) is None + + +def test_receive_diagnostic_uses_allowed_addresses_and_checks_target(mock_socket): + sut = DoIPClient(test_ip, test_logical_address, allowed_response_addresses={2}) + sut._ecu_logical_address = 3 + valid_response = bytearray(diagnostic_result) + valid_response[8:10] = b"\x00\x02" + wrong_target = bytearray(valid_response) + wrong_target[10:12] = b"\x0e\x01" + mock_socket.rx_queue.extend([diagnostic_result, wrong_target, valid_response]) + assert sut.receive_diagnostic(timeout=0.01) == bytearray([0, 1, 2, 3]) + assert mock_socket.rx_queue == [] + + def test_request_vehicle_identification(mock_socket): sut = DoIPClient(test_ip, test_logical_address) mock_socket.rx_queue.append(vehicle_identification_response) From 11d8eb2aa18696c8d835fbafbd2966f1eb4c01b5 Mon Sep 17 00:00:00 2001 From: NOKIAO <675925864@qq.com> Date: Fri, 9 Oct 2026 15:28:05 +0800 Subject: [PATCH 2/3] update --- doipclient/client.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/doipclient/client.py b/doipclient/client.py index dfb7c5c..a04c038 100644 --- a/doipclient/client.py +++ b/doipclient/client.py @@ -801,9 +801,6 @@ def send_diagnostic_to_address( def receive_diagnostic(self, timeout=None): """Receive a raw diagnostic payload (ie: UDS) from the ECU. - Only source addresses in allowed_response_addresses are accepted, or ecu_logical_address - when that collection was not configured or is empty. The target must match client_logical_address. - :return: Raw UDS payload :rtype: bytearray :raises TimeoutError: No diagnostic response received in time From a0138e7eb3280ae65ceb9142f3c8af6ed5ff6bb8 Mon Sep 17 00:00:00 2001 From: NOKIAO <675925864@qq.com> Date: Fri, 9 Oct 2026 16:03:59 +0800 Subject: [PATCH 3/3] support callable for allowed_response_addresses --- doipclient/client.py | 46 ++++++++++++++++++++++++++------------------ tests/test_client.py | 36 +++++++++++++++++++++++++--------- 2 files changed, 54 insertions(+), 28 deletions(-) diff --git a/doipclient/client.py b/doipclient/client.py index a04c038..68efcf6 100644 --- a/doipclient/client.py +++ b/doipclient/client.py @@ -149,11 +149,13 @@ class DoIPClient: :type auto_reconnect_tcp: bool :param vm_specific: Optional 4 byte long int :type vm_specific: int, optional - :param allowed_response_addresses: Allowed source logical addresses for diagnostic responses. When omitted or empty, - only the current ecu_logical_address is accepted. Set this to ECU physical addresses when sending to a functional - address. ACK/NACK messages may originate from the request target or any address in this collection. + :param allowed_response_addresses: Additional allowed source logical addresses for diagnostic responses. + Set this to ECU physical addresses when sending to a functional address. Alternatively, pass a callable + that takes a source address and returns whether it is allowed. Diagnostic responses may originate from + the current ecu_logical_address or any source accepted by this configuration. ACK/NACK messages may + originate from the request target or any source accepted by this configuration. The response target must always match client_logical_address. - :type allowed_response_addresses: Collection[int], optional + :type allowed_response_addresses: Union[Collection[int], Callable[[int], bool]], optional :raises ConnectionRefusedError: If the activation request fails :raises ValueError: If the IPAddress is neither an IPv4 nor an IPv6 address @@ -735,13 +737,12 @@ def send_diagnostic(self, diagnostic_payload, timeout=A_PROCESSING_TIME): self._ecu_logical_address, diagnostic_payload, timeout ) - def _get_allowed_response_addresses(self, address=None): - response_addresses = self._allowed_response_addresses or { - self._ecu_logical_address - } - if address is not None: - return {address}.union(response_addresses) - return response_addresses + def _is_allowed_response_address(self, source_address, address): + if source_address == address: + return True + if callable(self._allowed_response_addresses): + return self._allowed_response_addresses(source_address) + return source_address in (self._allowed_response_addresses or ()) def send_diagnostic_to_address( self, address, diagnostic_payload, timeout=A_PROCESSING_TIME @@ -769,8 +770,10 @@ def send_diagnostic_to_address( result = self.read_doip() if type(result) == DiagnosticMessageNegativeAcknowledgement: if ( - result.source_address in self._get_allowed_response_addresses(address) - and result.target_address == self._client_logical_address + result.target_address == self._client_logical_address + and self._is_allowed_response_address( + result.source_address, address + ) ): raise IOError( "Diagnostic request rejected with negative acknowledge code: {}".format( @@ -783,8 +786,10 @@ def send_diagnostic_to_address( ) elif type(result) == DiagnosticMessagePositiveAcknowledgement: if ( - result.source_address in self._get_allowed_response_addresses(address) - and result.target_address == self._client_logical_address + result.target_address == self._client_logical_address + and self._is_allowed_response_address( + result.source_address, address + ) ): return else: @@ -815,14 +820,17 @@ def receive_diagnostic(self, timeout=None): else: result = self.read_doip() if type(result) == DiagnosticMessage: + source_address_allowed = self._is_allowed_response_address( + result.source_address, self._ecu_logical_address + ) if ( - result.source_address in self._get_allowed_response_addresses() - and result.target_address == self._client_logical_address + result.target_address == self._client_logical_address + and source_address_allowed ): return result.user_data elif ( - result.source_address not in self._get_allowed_response_addresses() - and result.target_address == self._client_logical_address + result.target_address == self._client_logical_address + and not source_address_allowed ): logger.warning( "Received DiagnosticMessage with expected target address, but source address is not allowed. Ignoring." diff --git a/tests/test_client.py b/tests/test_client.py index 8728c92..71a0123 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -730,7 +730,7 @@ def test_send_diagnostic_to_address_times_out_when_only_invalid_target_ack_nack( assert mock_socket.tx_queue[-1] == diagnostic_request_to_address -@pytest.mark.parametrize("allowed_addresses", [None, {}]) +@pytest.mark.parametrize("allowed_addresses", [None, {}, {0x1234}, lambda address: False]) @pytest.mark.parametrize("ecu_address", [1, 2]) def test_receive_diagnostic(mock_socket, allowed_addresses, ecu_address): sut = DoIPClient( @@ -751,8 +751,11 @@ def test_receive_diagnostic_timeout(mock_socket): @pytest.mark.parametrize("source_address", [1, 2]) -def test_functional_diagnostic_accepts_allowed_responses(mock_socket, source_address): - sut = DoIPClient(test_ip, 0xE400, allowed_response_addresses={1, 2}) +@pytest.mark.parametrize("allowed_addresses", [{1, 2}, lambda address: address in (1, 2)]) +def test_functional_diagnostic_accepts_allowed_responses( + mock_socket, source_address, allowed_addresses +): + sut = DoIPClient(test_ip, 0xE400, allowed_response_addresses=allowed_addresses) ack, response = bytearray(diagnostic_ack), bytearray(diagnostic_result) ack[8:10] = response[8:10] = source_address.to_bytes(2, "big") mock_socket.rx_queue.extend([ack, response]) @@ -761,22 +764,29 @@ def test_functional_diagnostic_accepts_allowed_responses(mock_socket, source_add assert sut.receive_diagnostic(timeout=0.01) == bytearray([0, 1, 2, 3]) -def test_functional_diagnostic_accepts_allowed_nack(mock_socket): - sut = DoIPClient(test_ip, 0xE400, allowed_response_addresses={1}) +@pytest.mark.parametrize("allowed_addresses", [{1}, lambda address: address == 1]) +def test_functional_diagnostic_accepts_allowed_nack(mock_socket, allowed_addresses): + sut = DoIPClient(test_ip, 0xE400, allowed_response_addresses=allowed_addresses) mock_socket.rx_queue.append(diagnostic_nack) with pytest.raises(IOError, match="Diagnostic request rejected"): sut.send_diagnostic(bytearray([0, 1, 2]), timeout=0.01) -def test_functional_diagnostic_accepts_default_response_address(mock_socket): +def test_functional_diagnostic_rejects_unlisted_ecu_ack(mock_socket): sut = DoIPClient(test_ip, test_logical_address) sut._ecu_logical_address = 2 mock_socket.rx_queue.append(diagnostic_ack_invalid_source) - assert sut.send_diagnostic_to_address(0xE400, bytearray([0, 1, 2]), timeout=0.01) is None + with pytest.raises(TimeoutError): + sut.send_diagnostic_to_address(0xE400, bytearray([0, 1, 2]), timeout=0.01) -def test_receive_diagnostic_uses_allowed_addresses_and_checks_target(mock_socket): - sut = DoIPClient(test_ip, test_logical_address, allowed_response_addresses={2}) +@pytest.mark.parametrize("allowed_addresses", [{2}, lambda address: address == 2]) +def test_receive_diagnostic_uses_allowed_addresses_and_checks_target( + mock_socket, allowed_addresses +): + sut = DoIPClient( + test_ip, test_logical_address, allowed_response_addresses=allowed_addresses + ) sut._ecu_logical_address = 3 valid_response = bytearray(diagnostic_result) valid_response[8:10] = b"\x00\x02" @@ -787,6 +797,14 @@ def test_receive_diagnostic_uses_allowed_addresses_and_checks_target(mock_socket assert mock_socket.rx_queue == [] +def test_diagnostic_ack_accepts_request_target_with_callable(mock_socket): + sut = DoIPClient( + test_ip, test_logical_address, allowed_response_addresses=lambda address: False + ) + mock_socket.rx_queue.append(diagnostic_ack_to_address) + assert sut.send_diagnostic_to_address(0x1234, bytearray([0, 1, 2]), timeout=0.01) is None + + def test_request_vehicle_identification(mock_socket): sut = DoIPClient(test_ip, test_logical_address) mock_socket.rx_queue.append(vehicle_identification_response)