avdtp: bound message assembler to drop truncated PDUs (DoS prevention)

bumble/avdtp.py

@@ -311,6 +311,13 @@ def reset(self) -> None:
     def on_pdu(self, pdu: bytes) -> None:
         self.packet_count += 1
 
+        # Drop empty PDUs sent by remote — accessing pdu[0] below would
+        # raise IndexError, propagating up to the L2CAP read loop and
+        # tearing down the channel. Same class as #912 (ATT empty PDU).
+        if not pdu:
+            logger.warning('AVDTP message assembler: empty PDU dropped')
+            return
+
         transaction_label = pdu[0] >> 4
         packet_type = Protocol.PacketType((pdu[0] >> 2) & 3)
         message_type = Message.MessageType(pdu[0] & 3)
@@ -324,6 +331,23 @@ def on_pdu(self, pdu: bytes) -> None:
             Protocol.PacketType.SINGLE_PACKET,
             Protocol.PacketType.START_PACKET,
         ):
+            # Both single and start packets carry the signal identifier in
+            # pdu[1]; start packets additionally carry the packet count in
+            # pdu[2]. Guard each access so a malformed remote frame can't
+            # crash the message assembler.
+            if len(pdu) < 2:
+                logger.warning(
+                    'AVDTP %s packet too short (%d bytes); dropped',
+                    packet_type.name,
+                    len(pdu),
+                )
+                return
+            if packet_type == Protocol.PacketType.START_PACKET and len(pdu) < 3:
+                logger.warning(
+                    'AVDTP START packet missing signal-packet count; dropped'
+                )
+                return
+
             if self.message is not None:
                 # The previous message has not been terminated
                 logger.warning(

tests/avdtp_test.py

@@ -120,6 +120,31 @@ def test_messages(message: avdtp.Message):
     assert message.payload == parsed.payload
 
 
+# -----------------------------------------------------------------------------
+@pytest.mark.parametrize(
+    'pdu',
+    (
+        b'',  # empty PDU — would IndexError on pdu[0]
+        b'\x00',  # 1-byte SINGLE_PACKET — would IndexError on pdu[1]
+        b'\x04',  # 1-byte START_PACKET — would IndexError on pdu[1]
+        b'\x44\x10',  # 2-byte START_PACKET — would IndexError on pdu[2]
+    ),
+)
+def test_message_assembler_truncated_pdu(pdu: bytes):
+    """Truncated AVDTP PDUs from a remote peer must NOT raise IndexError —
+    same DoS class as #912 (ATT empty PDU). The assembler is required to
+    log + drop and stay alive so the L2CAP channel survives."""
+    completed = []
+
+    def callback(transaction_label, message):
+        completed.append((transaction_label, message))
+
+    assembler = avdtp.MessageAssembler(callback)
+    # Must not raise; nothing should be delivered to callback either.
+    assembler.on_pdu(pdu)
+    assert not completed
+
+
 # -----------------------------------------------------------------------------
 def test_rtp():
     packet = bytes.fromhex(