Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions bumble/hci.py
Original file line number Diff line number Diff line change
Expand Up @@ -8174,6 +8174,8 @@ def from_bytes(cls, packet: bytes) -> HCI_IsoDataPacket:
packet_status_flag: int | None = None

pos = 1
if len(packet) < pos + 4:
raise InvalidPacketError(f'ISO data packet too short: {len(packet)} bytes')
pdu_info, data_total_length = struct.unpack_from('<HH', packet, pos)
connection_handle = pdu_info & 0xFFF
pb_flag = (pdu_info >> 12) & 0b11
Expand All @@ -8186,10 +8188,14 @@ def from_bytes(cls, packet: bytes) -> HCI_IsoDataPacket:
if ts_flag:
if not should_include_sdu_info:
logger.warning(f'Timestamp included when pb_flag={bin(pb_flag)}')
if len(packet) < pos + 4:
raise InvalidPacketError('ISO data packet truncated (timestamp)')
time_stamp, *_ = struct.unpack_from('<I', packet, pos)
pos += 4

if should_include_sdu_info:
if len(packet) < pos + 4:
raise InvalidPacketError('ISO data packet truncated (SDU info)')
packet_sequence_number, sdu_info = struct.unpack_from('<HH', packet, pos)
iso_sdu_length = sdu_info & 0xFFF
packet_status_flag = (sdu_info >> 15) & 1
Expand Down
10 changes: 10 additions & 0 deletions tests/hci_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -701,6 +701,16 @@ def test_iso_data_packet():
assert bytes(packet) == data


# -----------------------------------------------------------------------------
def test_iso_data_packet_too_short():
# A truncated ISO data packet must raise InvalidPacketError, not the raw
# struct.error from unpack_from. Here pb_flag=0b00 asks for 4 bytes of SDU
# info at offset 5, but the packet ends at offset 5.
data = bytes.fromhex('0561000000') # type + pdu_info(0x0061) + len(0) = 5 bytes
with pytest.raises(hci.InvalidPacketError):
hci.HCI_IsoDataPacket.from_bytes(data)


# -----------------------------------------------------------------------------
def run_test_events():
test_HCI_Event()
Expand Down