Disconnect LE credit-based channels on invalid SDU lengths

bumble/l2cap.py

@@ -1742,6 +1742,7 @@ def on_pdu(self, pdu: bytes) -> None:
 
         if self.state != self.State.CONNECTED:
             logger.warning('received PDU while not connected, dropping')
+            return
 
         # Manage the peer credits
         if self.peer_credits == 0:
@@ -1768,29 +1769,33 @@ def on_pdu(self, pdu: bytes) -> None:
             self.in_sdu += pdu
 
         # Check if the SDU is complete
-        if self.in_sdu_length == 0:
-            # We don't know the size yet, check if we have received the header to
-            # compute it
-            if len(self.in_sdu) >= 2:
-                self.in_sdu_length = struct.unpack_from('<H', self.in_sdu, 0)[0]
-        if self.in_sdu_length == 0:
-            # We'll compute it later
+        if len(self.in_sdu) < 2:
+            # We don't know the size yet, we'll compute it later
             return
-        if len(self.in_sdu) < 2 + self.in_sdu_length:
-            # Not complete yet
-            logger.debug(
-                f'SDU: {len(self.in_sdu) - 2} of {self.in_sdu_length} bytes received'
-            )
-            return
-        if len(self.in_sdu) != 2 + self.in_sdu_length:
+        self.in_sdu_length = struct.unpack_from('<H', self.in_sdu, 0)[0]
+        if self.in_sdu_length > self.mtu or len(self.in_sdu) > 2 + self.in_sdu_length:
             # Overflow
             logger.warning(
-                f'SDU overflow: sdu_length={self.in_sdu_length}, '
+                f'SDU overflow: sdu_length={self.in_sdu_length}, mtu={self.mtu}, '
                 f'received {len(self.in_sdu) - 2}'
             )
-            # TODO: we should disconnect
             self.in_sdu = None
             self.in_sdu_length = 0
+            self._change_state(self.State.DISCONNECTING)
+            self.flush_output()
+            self.send_control_frame(
+                L2CAP_Disconnection_Request(
+                    identifier=self.manager.next_identifier(self.connection),
+                    destination_cid=self.destination_cid,
+                    source_cid=self.source_cid,
+                )
+            )
+            return
+        if len(self.in_sdu) < 2 + self.in_sdu_length:
+            # Not complete yet
+            logger.debug(
+                f'SDU: {len(self.in_sdu) - 2} of {self.in_sdu_length} bytes received'
+            )
             return
 
         # Send the SDU to the sink

tests/l2cap_test.py

@@ -531,6 +531,68 @@ async def test_disconnection_collision():
     assert server_channel.state == l2cap.LeCreditBasedChannel.State.DISCONNECTED
 
 
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_zero_length_sdu():
+    devices = await TwoDevices.create_with_connection()
+
+    server_channels = asyncio.Queue[l2cap.LeCreditBasedChannel]()
+    server = devices[1].create_l2cap_server(
+        spec=l2cap.LeCreditBasedChannelSpec(),
+        handler=server_channels.put_nowait,
+    )
+    assert (connection := devices.connections[0])
+    client = await connection.create_l2cap_channel(
+        spec=l2cap.LeCreditBasedChannelSpec(server.psm)
+    )
+    server_channel = await server_channels.get()
+    received = asyncio.Queue[bytes]()
+    server_channel.sink = received.put_nowait
+
+    # A zero-length SDU must not hold up the SDUs that follow it
+    client.send_pdu(bytes(2))
+    client.write(b'hello')
+    assert await asyncio.wait_for(received.get(), timeout=1.0) == b''
+    assert await asyncio.wait_for(received.get(), timeout=1.0) == b'hello'
+    await client.disconnect()
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "sdu_length, payload_length",
+    (
+        (101, 0),  # SDU length larger than the MTU
+        (0, 48),  # More payload than the SDU length
+        (10, 11),
+    ),
+)
+async def test_invalid_sdu_length(sdu_length: int, payload_length: int):
+    devices = await TwoDevices.create_with_connection()
+
+    server_channels = asyncio.Queue[l2cap.LeCreditBasedChannel]()
+    server = devices[1].create_l2cap_server(
+        spec=l2cap.LeCreditBasedChannelSpec(mtu=100),
+        handler=server_channels.put_nowait,
+    )
+    assert (connection := devices.connections[0])
+    client = await connection.create_l2cap_channel(
+        spec=l2cap.LeCreditBasedChannelSpec(server.psm)
+    )
+    server_channel = await server_channels.get()
+    server_channel.sink = mock.Mock()
+    closed = asyncio.Event()
+    server_channel.once(server_channel.EVENT_CLOSE, closed.set)
+
+    # The receiver must disconnect the channel rather than buffer or deliver
+    client.send_pdu(sdu_length.to_bytes(2, 'little') + bytes(payload_length))
+    await asyncio.wait_for(closed.wait(), timeout=1.0)
+
+    server_channel.sink.assert_not_called()
+    assert server_channel.in_sdu is None
+    assert client.state == l2cap.LeCreditBasedChannel.State.DISCONNECTED
+
+
 # -----------------------------------------------------------------------------
 async def run():
     test_helpers()