diff --git a/apps/usb_probe.py b/apps/usb_probe.py index 81a266f2..922446f5 100644 --- a/apps/usb_probe.py +++ b/apps/usb_probe.py @@ -39,6 +39,7 @@ # Constants # ----------------------------------------------------------------------------- USB_DEVICE_CLASS_DEVICE = 0x00 +USB_DEVICE_CLASS_MISCELLANEOUS = 0xEF USB_DEVICE_CLASS_WIRELESS_CONTROLLER = 0xE0 USB_DEVICE_SUBCLASS_RF_CONTROLLER = 0x01 USB_DEVICE_PROTOCOL_BLUETOOTH_PRIMARY_CONTROLLER = 0x01 @@ -155,8 +156,12 @@ def is_bluetooth_hci(device): ) == USB_BT_HCI_CLASS_TUPLE: return True - # If the device class is 'Device', look for a matching interface - if device.getDeviceClass() == USB_DEVICE_CLASS_DEVICE: + # If the device class is 'Device' or 'Miscellaneous' (composite devices), look + # for a matching interface + if device.getDeviceClass() in ( + USB_DEVICE_CLASS_DEVICE, + USB_DEVICE_CLASS_MISCELLANEOUS, + ): for configuration in device: for interface in configuration: for setting in interface: diff --git a/bumble/transport/usb.py b/bumble/transport/usb.py index 2dcb48aa..2185b9b5 100644 --- a/bumble/transport/usb.py +++ b/bumble/transport/usb.py @@ -56,8 +56,10 @@ def _safe_call_soon(loop: asyncio.AbstractEventLoop, callback, *args) -> None: # ----------------------------------------------------------------------------- # pylint: disable=invalid-name USB_RECIPIENT_DEVICE = 0x00 +USB_RECIPIENT_INTERFACE = 0x01 USB_REQUEST_TYPE_CLASS = 0x01 << 5 USB_DEVICE_CLASS_DEVICE = 0x00 +USB_DEVICE_CLASS_MISCELLANEOUS = 0xEF USB_DEVICE_CLASS_WIRELESS_CONTROLLER = 0xE0 USB_DEVICE_SUBCLASS_RF_CONTROLLER = 0x01 USB_DEVICE_PROTOCOL_BLUETOOTH_PRIMARY_CONTROLLER = 0x01 @@ -240,10 +242,11 @@ def find_endpoints(device, forced_mode, sco_alternate=None): class UsbPacketSink: - def __init__(self, device, bulk_out, isochronous_out) -> None: + def __init__(self, device, bulk_out, isochronous_out, interface_number=0) -> None: self.device = device self.bulk_out = bulk_out self.isochronous_out = isochronous_out + self.interface_number = interface_number self.bulk_or_control_out_transfer = device.getTransfer() self.isochronous_out_transfer = ( device.getTransfer( @@ -315,11 +318,20 @@ async def process_queue(self): self.bulk_or_control_out_transfer.submit() submitted = True elif packet_type == hci.HCI_COMMAND_PACKET: + # A controller that isn't the first interface of a composite + # device must be addressed by its interface number, or the + # command goes to whichever function owns interface 0 + # (Core Spec Vol 4, Part B, 2.2.2) self.bulk_or_control_out_transfer.setControl( - USB_RECIPIENT_DEVICE | USB_REQUEST_TYPE_CLASS, - 0, + ( + USB_RECIPIENT_INTERFACE + if self.interface_number + else USB_RECIPIENT_DEVICE + ) + | USB_REQUEST_TYPE_CLASS, 0, 0, + self.interface_number, packet_payload, callback=self.transfer_callback, ) @@ -857,8 +869,12 @@ def device_is_bluetooth_hci(device): ) == USB_BT_HCI_CLASS_TUPLE: return True - # If the device class is 'Device', look for a matching interface - if device.getDeviceClass() == USB_DEVICE_CLASS_DEVICE: + # If the device class is 'Device' or 'Miscellaneous' (composite + # devices), look for a matching interface + if device.getDeviceClass() in ( + USB_DEVICE_CLASS_DEVICE, + USB_DEVICE_CLASS_MISCELLANEOUS, + ): for configuration in device: for interface in configuration: for setting in interface: @@ -955,7 +971,9 @@ def device_is_bluetooth_hci(device): source = UsbPacketSource( device, device_metadata, interrupt_in, bulk_in, isochronous_in ) - sink = UsbPacketSink(device, bulk_out, isochronous_out) + sink = UsbPacketSink( + device, bulk_out, isochronous_out, acl_interface.getNumber() + ) return UsbTransport(context, device, acl_interface, sco_interface, source, sink) except usb1.USBError as error: logger.warning(color(f'!!! failed to open USB device: {error}', 'red')) diff --git a/tests/transport/usb_test.py b/tests/transport/usb_test.py index 2235c172..6295282e 100644 --- a/tests/transport/usb_test.py +++ b/tests/transport/usb_test.py @@ -118,3 +118,90 @@ async def test_usb_packet_sink_iso_routing_with_iso_endpoint(): await sink.queue_task except asyncio.CancelledError: pass + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ('interface_number', 'request_type'), + [ + # Single function controller: addressed to the device + (0, 0x20), + # Controller behind other functions in a composite device: addressed to + # its interface + (2, 0x21), + ], +) +async def test_usb_packet_sink_command_addressing(interface_number, request_type): + mock_device = mock.Mock() + mock_bulk_out = mock.Mock() + mock_transfer = mock.Mock() + mock_device.getTransfer.return_value = mock_transfer + + sink = usb.UsbPacketSink( + mock_device, + mock_bulk_out, + isochronous_out=None, + interface_number=interface_number, + ) + sink.start() + + # Send HCI_Reset + sink.on_packet(bytes([hci.HCI_COMMAND_PACKET, 0x03, 0x0C, 0x00])) + + # Yield control to let the queue processor run + await asyncio.sleep(0.01) + + mock_transfer.setControl.assert_called_once_with( + request_type, + 0, + 0, + interface_number, + bytes([0x03, 0x0C, 0x00]), + callback=sink.transfer_callback, + ) + mock_transfer.submit.assert_called_once() + + if sink.queue_task: + sink.queue_task.cancel() + try: + await sink.queue_task + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'device_class', + [ + # Class defined per interface + 0x00, + # Miscellaneous: composite device with interface associations + 0xEF, + ], +) +async def test_open_usb_transport_by_index_checks_interfaces(device_class): + # A Bluetooth interface without endpoints: enough to be selected by index, + # but not to open a transport + setting = mock.MagicMock() + setting.getClass.return_value = 0xE0 + setting.getSubClass.return_value = 0x01 + setting.getProtocol.return_value = 0x01 + setting.__iter__.side_effect = lambda: iter([]) + interface = mock.MagicMock() + interface.__iter__.side_effect = lambda: iter([setting]) + configuration = mock.MagicMock() + configuration.__iter__.side_effect = lambda: iter([interface]) + device = mock.MagicMock() + device.getDeviceClass.return_value = device_class + device.getDeviceSubClass.return_value = 0x02 + device.getDeviceProtocol.return_value = 0x01 + device.__iter__.side_effect = lambda: iter([configuration]) + context = mock.Mock() + context.getDeviceIterator.return_value = [device] + + with ( + mock.patch.object(usb, 'load_libusb'), + mock.patch.object(usb.usb1, 'USBContext', return_value=context), + pytest.raises(usb.TransportInitError, match='no compatible interface'), + ): + await usb.open_usb_transport('0')