fix(j1939-21): respect sequence number for fragmented frames on reass…

j1939/j1939_21.py

@@ -321,6 +321,7 @@ def _process_tp_cm(self, mid, dest_address, data, timestamp):
                         'pgn': pgn,
                         'message_size': message_size,
                         'num_packages': num_packages,
+                        'next_expected_packet': 1,
                         'next_packet': min(self._max_cmdt_packets, max_num_packages),
                         'max_cmdt_packages': self._max_cmdt_packets,
                         'num_packages_max_rec': min(self._max_cmdt_packets, max_num_packages),
@@ -386,6 +387,7 @@ def _process_tp_cm(self, mid, dest_address, data, timestamp):
                         "pgn": pgn,
                         "message_size": message_size,
                         "num_packages": num_packages,
+                        "next_expected_packet": 1,
                         "next_packet": 1,
                         "max_cmdt_packages": self._max_cmdt_packets,
                         "data": [],
@@ -417,8 +419,33 @@ def _process_tp_dt(self, mid, dest_address, data, timestamp):
                 # TODO: LOG/TRACE/EXCEPTION?
                 return
 
+            expected_sequence_number = self._rcv_buffer[buffer_hash]['next_expected_packet']
+            if sequence_number != expected_sequence_number:
+                sequence_error = (
+                    "duplicate"
+                    if 0 < sequence_number < expected_sequence_number
+                    else "invalid"
+                )
+                if dest_address != ParameterGroupNumber.Address.GLOBAL:
+                    # Keep the wire value in the standardized reason range. The
+                    # specific sequence error remains available in the exception.
+                    self.__send_tp_abort(
+                        dest_address,
+                        src_address,
+                        self.ConnectionAbortReason.RESOURCES,
+                        self._rcv_buffer[buffer_hash]['pgn'],
+                    )
+                del self._rcv_buffer[buffer_hash]
+                self.__job_thread_wakeup()
+                raise ValueError(
+                    "J1939-21 TP.DT packet out of sequence "
+                    f"({sequence_error}): "
+                    f"expected {expected_sequence_number}, received {sequence_number}"
+                )
+
             # get data
             self._rcv_buffer[buffer_hash]['data'].extend(data[1:])
+            self._rcv_buffer[buffer_hash]['next_expected_packet'] += 1
 
             # message is complete with sending an acknowledge
             if len(self._rcv_buffer[buffer_hash]['data']) >= self._rcv_buffer[buffer_hash]['message_size']:

test/test_ecu.py

@@ -4,6 +4,8 @@
 import pytest
 
 import j1939
+from j1939.j1939_21 import J1939_21
+from j1939.message_id import MessageId
 from test.helpers.feeder import Feeder
 
 # def test_connect(self):
@@ -59,6 +61,98 @@ def test_broadcast_receive_long(feeder):
     feeder.receive()
 
 
+def test_broadcast_receive_out_of_sequence_packet_raises():
+    """Reject and terminate a BAM session with an invalid sequence number."""
+    sent = []
+    notified = []
+    dll = J1939_21(
+        send_message=lambda *args: sent.append(args),
+        job_thread_wakeup=lambda: None,
+        notify_subscribers=lambda *args: notified.append(args),
+        max_cmdt_packets=1,
+        minimum_tp_rts_cts_dt_interval=None,
+        minimum_tp_bam_dt_interval=None,
+        ecu_is_message_acceptable=lambda dest: True,
+    )
+    bam_mid = MessageId(can_id=0x00ECFF01)
+    dll._process_tp_cm(bam_mid, 0xFF, [32, 20, 0, 3, 255, 0xB0, 0xFE, 0], 0.0)
+    buffer_hash = dll._buffer_hash(0x01, 0xFF)
+    mid = MessageId(can_id=0x00EBFF01)
+
+    with pytest.raises(ValueError, match='out of sequence'):
+        dll._process_tp_dt(mid, 0xFF, [2, 8, 9, 10, 11, 12, 13, 14], 0.0)
+
+    assert sent == []
+    assert notified == []
+    assert buffer_hash not in dll._rcv_buffer
+
+
+def test_peer_to_peer_receive_out_of_sequence_packet_aborts():
+    """Reject an out-of-sequence CMDT packet and abort the session."""
+    sent = []
+    dll = J1939_21(
+        send_message=lambda *args: sent.append(args),
+        job_thread_wakeup=lambda: None,
+        notify_subscribers=lambda *args: None,
+        max_cmdt_packets=1,
+        minimum_tp_rts_cts_dt_interval=None,
+        minimum_tp_bam_dt_interval=None,
+        ecu_is_message_acceptable=lambda dest: True,
+    )
+    rts_mid = MessageId(can_id=0x00EC0201)
+    dll._process_tp_cm(rts_mid, 0x02, [16, 20, 0, 3, 1, 0, 223, 0], 0.0)
+    sent.clear()
+
+    dt_mid = MessageId(can_id=0x00EB0201)
+    with pytest.raises(ValueError, match='out of sequence'):
+        dll._process_tp_dt(dt_mid, 0x02, [2, 1, 2, 3, 4, 5, 6, 7], 0.0)
+
+    assert sent == [
+        (
+            0x1CEC0102,
+            True,
+            [255, 2, 255, 255, 255, 0, 223, 0],
+        )
+    ]
+    assert not dll._rcv_buffer
+
+    dll._process_tp_cm(rts_mid, 0x02, [16, 20, 0, 3, 1, 0, 223, 0], 0.0)
+    sent.clear()
+
+    with pytest.raises(ValueError, match='out of sequence'):
+        dll._process_tp_dt(dt_mid, 0x02, [0, 1, 2, 3, 4, 5, 6, 7], 0.0)
+
+    assert sent[0][2][1] == 2
+
+
+def test_peer_to_peer_sequence_gap_after_valid_packet_aborts():
+    """Reject a sequence gap without delivering a partial RTS/CTS payload."""
+    sent = []
+    notified = []
+    dll = J1939_21(
+        send_message=lambda *args: sent.append(args),
+        job_thread_wakeup=lambda: None,
+        notify_subscribers=lambda *args: notified.append(args),
+        max_cmdt_packets=2,
+        minimum_tp_rts_cts_dt_interval=None,
+        minimum_tp_bam_dt_interval=None,
+        ecu_is_message_acceptable=lambda dest: True,
+    )
+    rts_mid = MessageId(can_id=0x00EC0201)
+    dll._process_tp_cm(rts_mid, 0x02, [16, 20, 0, 3, 2, 0, 223, 0], 0.0)
+    sent.clear()
+
+    dt_mid = MessageId(can_id=0x00EB0201)
+    dll._process_tp_dt(dt_mid, 0x02, [1, 1, 2, 3, 4, 5, 6, 7], 0.0)
+
+    with pytest.raises(ValueError, match='out of sequence'):
+        dll._process_tp_dt(dt_mid, 0x02, [3, 8, 9, 10, 11, 12, 13, 14], 0.0)
+
+    assert sent[0][2][1] == 2
+    assert notified == []
+    assert not dll._rcv_buffer
+
+
 def test_peer_to_peer_receive_short(feeder):
     """Test the receivement of a normal peer-to-peer message