Snap for 15025362 from f623f430c4fc3661842af415167e18b5906837b5 to 26Q2-release Change-Id: I2f82ffb3d588b56f0123160f7fd7b20e9adafec6
diff --git a/METADATA b/METADATA index e9ea1f9..aea7fbd 100644 --- a/METADATA +++ b/METADATA
@@ -8,13 +8,13 @@ license_type: NOTICE last_upgrade_date { year: 2026 - month: 2 - day: 5 + month: 3 + day: 2 } homepage: "https://google.github.io/bumble/" identifier { type: "Git" value: "https://github.com/google/bumble" - version: "256a1a7405232aaa777b48a55a664490ba5c79aa" + version: "eb64debb622d5059dc5fba8d92d95b63947dcadb" } }
diff --git a/apps/auracast.py b/apps/auracast.py index 0a86b01..59cedbb 100644 --- a/apps/auracast.py +++ b/apps/auracast.py
@@ -24,13 +24,18 @@ import functools import logging import secrets +import sys from collections.abc import AsyncGenerator, Awaitable, Callable, Iterable, Sequence from typing import ( Any, ) import click -import tomli + +if sys.version_info >= (3, 11): + import tomllib +else: + import tomli as tomllib try: import lc3 # type: ignore # pylint: disable=E0401 @@ -114,7 +119,7 @@ broadcasts: list[Broadcast] = [] with open(filename, "rb") as config_file: - config = tomli.load(config_file) + config = tomllib.load(config_file) for broadcast in config.get("broadcasts", []): sources = [] for source in broadcast.get("sources", []):
diff --git a/apps/pair.py b/apps/pair.py index 135c9be..9bfa0a5 100644 --- a/apps/pair.py +++ b/apps/pair.py
@@ -20,11 +20,12 @@ import asyncio import logging import os +from typing import ClassVar import click from prompt_toolkit.shortcuts import PromptSession -from bumble import data_types +from bumble import data_types, smp from bumble.a2dp import make_audio_sink_service_sdp_records from bumble.att import ( ATT_INSUFFICIENT_AUTHENTICATION_ERROR, @@ -40,7 +41,7 @@ PhysicalTransport, ProtocolError, ) -from bumble.device import Device, Peer +from bumble.device import Connection, Device, Peer from bumble.gatt import ( GATT_DEVICE_NAME_CHARACTERISTIC, GATT_GENERIC_ACCESS_SERVICE, @@ -53,7 +54,6 @@ from bumble.keys import JsonKeyStore from bumble.pairing import OobData, PairingConfig, PairingDelegate from bumble.smp import OobContext, OobLegacyContext -from bumble.smp import error_name as smp_error_name from bumble.transport import open_transport from bumble.utils import AsyncRunner @@ -65,7 +65,7 @@ # ----------------------------------------------------------------------------- class Waiter: - instance: Waiter | None = None + instance: ClassVar[Waiter | None] = None def __init__(self, linger=False): self.done = asyncio.get_running_loop().create_future() @@ -319,12 +319,13 @@ # ----------------------------------------------------------------------------- @AsyncRunner.run_in_task() -async def on_pairing_failure(connection, reason): +async def on_pairing_failure(connection: Connection, reason: smp.ErrorCode): print(color('***-----------------------------------', 'red')) - print(color(f'*** Pairing failed: {smp_error_name(reason)}', 'red')) + print(color(f'*** Pairing failed: {reason.name}', 'red')) print(color('***-----------------------------------', 'red')) await connection.disconnect() - Waiter.instance.terminate() + if Waiter.instance: + Waiter.instance.terminate() # -----------------------------------------------------------------------------
diff --git a/bumble/a2dp.py b/bumble/a2dp.py index 4c9eb32..9bb2438 100644 --- a/bumble/a2dp.py +++ b/bumble/a2dp.py
@@ -88,13 +88,6 @@ SBC_STEREO_CHANNEL_MODE = 0x02 SBC_JOINT_STEREO_CHANNEL_MODE = 0x03 -SBC_CHANNEL_MODE_NAMES = { - SBC_MONO_CHANNEL_MODE: 'SBC_MONO_CHANNEL_MODE', - SBC_DUAL_CHANNEL_MODE: 'SBC_DUAL_CHANNEL_MODE', - SBC_STEREO_CHANNEL_MODE: 'SBC_STEREO_CHANNEL_MODE', - SBC_JOINT_STEREO_CHANNEL_MODE: 'SBC_JOINT_STEREO_CHANNEL_MODE' -} - SBC_BLOCK_LENGTHS = [4, 8, 12, 16] SBC_SUBBANDS = [4, 8] @@ -102,11 +95,6 @@ SBC_SNR_ALLOCATION_METHOD = 0x00 SBC_LOUDNESS_ALLOCATION_METHOD = 0x01 -SBC_ALLOCATION_METHOD_NAMES = { - SBC_SNR_ALLOCATION_METHOD: 'SBC_SNR_ALLOCATION_METHOD', - SBC_LOUDNESS_ALLOCATION_METHOD: 'SBC_LOUDNESS_ALLOCATION_METHOD' -} - SBC_MAX_FRAMES_IN_RTP_PAYLOAD = 15 MPEG_2_4_AAC_SAMPLING_FREQUENCIES = [ @@ -129,13 +117,6 @@ MPEG_4_AAC_LTP_OBJECT_TYPE = 0x02 MPEG_4_AAC_SCALABLE_OBJECT_TYPE = 0x03 -MPEG_2_4_OBJECT_TYPE_NAMES = { - MPEG_2_AAC_LC_OBJECT_TYPE: 'MPEG_2_AAC_LC_OBJECT_TYPE', - MPEG_4_AAC_LC_OBJECT_TYPE: 'MPEG_4_AAC_LC_OBJECT_TYPE', - MPEG_4_AAC_LTP_OBJECT_TYPE: 'MPEG_4_AAC_LTP_OBJECT_TYPE', - MPEG_4_AAC_SCALABLE_OBJECT_TYPE: 'MPEG_4_AAC_SCALABLE_OBJECT_TYPE' -} - OPUS_MAX_FRAMES_IN_RTP_PAYLOAD = 15 @@ -267,26 +248,27 @@ def create( cls, media_codec_type: int, data: bytes ) -> MediaCodecInformation | bytes: - if media_codec_type == CodecType.SBC: - return SbcMediaCodecInformation.from_bytes(data) - elif media_codec_type == CodecType.MPEG_2_4_AAC: - return AacMediaCodecInformation.from_bytes(data) - elif media_codec_type == CodecType.NON_A2DP: - vendor_media_codec_information = ( - VendorSpecificMediaCodecInformation.from_bytes(data) - ) - if ( - vendor_class_map := A2DP_VENDOR_MEDIA_CODEC_INFORMATION_CLASSES.get( - vendor_media_codec_information.vendor_id + match media_codec_type: + case CodecType.SBC: + return SbcMediaCodecInformation.from_bytes(data) + case CodecType.MPEG_2_4_AAC: + return AacMediaCodecInformation.from_bytes(data) + case CodecType.NON_A2DP: + vendor_media_codec_information = ( + VendorSpecificMediaCodecInformation.from_bytes(data) ) - ) and ( - media_codec_information_class := vendor_class_map.get( - vendor_media_codec_information.codec_id - ) - ): - return media_codec_information_class.from_bytes( - vendor_media_codec_information.value - ) + if ( + vendor_class_map := A2DP_VENDOR_MEDIA_CODEC_INFORMATION_CLASSES.get( + vendor_media_codec_information.vendor_id + ) + ) and ( + media_codec_information_class := vendor_class_map.get( + vendor_media_codec_information.codec_id + ) + ): + return media_codec_information_class.from_bytes( + vendor_media_codec_information.value + ) return vendor_media_codec_information @classmethod
diff --git a/bumble/at.py b/bumble/at.py index 9fe85ae..d49cf9c 100644 --- a/bumble/at.py +++ b/bumble/at.py
@@ -27,7 +27,7 @@ are ignored [..], unless they are embedded in numeric or string constants" Raises AtParsingError in case of invalid input string.""" - tokens = [] + tokens: list[bytearray] = [] in_quotes = False token = bytearray() for b in buffer: @@ -40,23 +40,24 @@ tokens.append(token[1:-1]) token = bytearray() else: - if char == b' ': - pass - elif char == b',' or char == b')': - tokens.append(token) - tokens.append(char) - token = bytearray() - elif char == b'(': - if len(token) > 0: - raise AtParsingError("open_paren following regular character") - tokens.append(char) - elif char == b'"': - if len(token) > 0: - raise AtParsingError("quote following regular character") - in_quotes = True - token.extend(char) - else: - token.extend(char) + match char: + case b' ': + pass + case b',' | b')': + tokens.append(token) + tokens.append(char) + token = bytearray() + case b'(': + if len(token) > 0: + raise AtParsingError("open_paren following regular character") + tokens.append(char) + case b'"': + if len(token) > 0: + raise AtParsingError("quote following regular character") + in_quotes = True + token.extend(char) + case _: + token.extend(char) tokens.append(token) return [bytes(token) for token in tokens if len(token) > 0] @@ -71,18 +72,19 @@ current: bytes | list = b'' for token in tokens: - if token == b',': - accumulator[-1].append(current) - current = b'' - elif token == b'(': - accumulator.append([]) - elif token == b')': - if len(accumulator) < 2: - raise AtParsingError("close_paren without matching open_paren") - accumulator[-1].append(current) - current = accumulator.pop() - else: - current = token + match token: + case b',': + accumulator[-1].append(current) + current = b'' + case b'(': + accumulator.append([]) + case b')': + if len(accumulator) < 2: + raise AtParsingError("close_paren without matching open_paren") + accumulator[-1].append(current) + current = accumulator.pop() + case _: + current = token accumulator[-1].append(current) if len(accumulator) > 1:
diff --git a/bumble/att.py b/bumble/att.py index 60e9b5c..07ebe86 100644 --- a/bumble/att.py +++ b/bumble/att.py
@@ -954,12 +954,13 @@ self.permissions = permissions # Convert the type to a UUID object if it isn't already - if isinstance(attribute_type, str): - self.type = UUID(attribute_type) - elif isinstance(attribute_type, bytes): - self.type = UUID.from_bytes(attribute_type) - else: - self.type = attribute_type + match attribute_type: + case str(): + self.type = UUID(attribute_type) + case bytes(): + self.type = UUID.from_bytes(attribute_type) + case _: + self.type = attribute_type self.value = value @@ -994,30 +995,31 @@ ) value: _T | None - if isinstance(self.value, AttributeValue): - try: - read_value = self.value.read(connection) - if inspect.isawaitable(read_value): - value = await read_value - else: - value = read_value - except ATT_Error as error: - raise ATT_Error( - error_code=error.error_code, att_handle=self.handle - ) from error - elif isinstance(self.value, AttributeValueV2): - try: - read_value = self.value.read(bearer) - if inspect.isawaitable(read_value): - value = await read_value - else: - value = read_value - except ATT_Error as error: - raise ATT_Error( - error_code=error.error_code, att_handle=self.handle - ) from error - else: - value = self.value + match self.value: + case AttributeValue(): + try: + read_value = self.value.read(connection) + if inspect.isawaitable(read_value): + value = await read_value + else: + value = read_value + except ATT_Error as error: + raise ATT_Error( + error_code=error.error_code, att_handle=self.handle + ) from error + case AttributeValueV2(): + try: + read_value = self.value.read(bearer) + if inspect.isawaitable(read_value): + value = await read_value + else: + value = read_value + except ATT_Error as error: + raise ATT_Error( + error_code=error.error_code, att_handle=self.handle + ) from error + case _: + value = self.value self.emit(self.EVENT_READ, connection, b'' if value is None else value) @@ -1049,26 +1051,27 @@ decoded_value = self.decode_value(value) - if isinstance(self.value, AttributeValue): - try: - result = self.value.write(connection, decoded_value) - if inspect.isawaitable(result): - await result - except ATT_Error as error: - raise ATT_Error( - error_code=error.error_code, att_handle=self.handle - ) from error - elif isinstance(self.value, AttributeValueV2): - try: - result = self.value.write(bearer, decoded_value) - if inspect.isawaitable(result): - await result - except ATT_Error as error: - raise ATT_Error( - error_code=error.error_code, att_handle=self.handle - ) from error - else: - self.value = decoded_value + match self.value: + case AttributeValue(): + try: + result = self.value.write(connection, decoded_value) + if inspect.isawaitable(result): + await result + except ATT_Error as error: + raise ATT_Error( + error_code=error.error_code, att_handle=self.handle + ) from error + case AttributeValueV2(): + try: + result = self.value.write(bearer, decoded_value) + if inspect.isawaitable(result): + await result + except ATT_Error as error: + raise ATT_Error( + error_code=error.error_code, att_handle=self.handle + ) from error + case _: + self.value = decoded_value self.emit(self.EVENT_WRITE, connection, decoded_value)
diff --git a/bumble/avrcp.py b/bumble/avrcp.py index 39754b4..a4bb669 100644 --- a/bumble/avrcp.py +++ b/bumble/avrcp.py
@@ -22,7 +22,14 @@ import functools import logging import struct -from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence +from collections.abc import ( + AsyncIterator, + Awaitable, + Callable, + Iterable, + Mapping, + Sequence, +) from dataclasses import dataclass, field from typing import ClassVar, SupportsBytes, TypeVar @@ -1049,11 +1056,9 @@ scope: Scope = field(metadata=Scope.type_metadata(1)) uid: int = field(metadata=_UINT64_BE_METADATA) uid_counter: int = field(metadata=hci.metadata('>2')) - start_item: int = field(metadata=hci.metadata('>4')) - end_item: int = field(metadata=hci.metadata('>4')) # When attributes is empty, all attributes will be requested. attributes: Sequence[MediaAttributeId] = field( - metadata=MediaAttributeId.type_metadata(1, list_begin=True, list_end=True) + metadata=MediaAttributeId.type_metadata(4, list_begin=True, list_end=True) ) @@ -1512,7 +1517,9 @@ @dataclass class TrackChangedEvent(Event): event_id = EventId.TRACK_CHANGED - identifier: bytes = field(metadata=hci.metadata('*')) + NO_TRACK = 0xFFFFFFFFFFFFFFFF + + uid: int = field(metadata=_UINT64_BE_METADATA) # ----------------------------------------------------------------------------- @@ -1536,16 +1543,19 @@ def __post_init__(self) -> None: super().__post_init__() - if self.attribute_id == ApplicationSetting.AttributeId.EQUALIZER_ON_OFF: - self.value_id = ApplicationSetting.EqualizerOnOffStatus(self.value_id) - elif self.attribute_id == ApplicationSetting.AttributeId.REPEAT_MODE: - self.value_id = ApplicationSetting.RepeatModeStatus(self.value_id) - elif self.attribute_id == ApplicationSetting.AttributeId.SHUFFLE_ON_OFF: - self.value_id = ApplicationSetting.ShuffleOnOffStatus(self.value_id) - elif self.attribute_id == ApplicationSetting.AttributeId.SCAN_ON_OFF: - self.value_id = ApplicationSetting.ScanOnOffStatus(self.value_id) - else: - self.value_id = ApplicationSetting.GenericValue(self.value_id) + match self.attribute_id: + case ApplicationSetting.AttributeId.EQUALIZER_ON_OFF: + self.value_id = ApplicationSetting.EqualizerOnOffStatus( + self.value_id + ) + case ApplicationSetting.AttributeId.REPEAT_MODE: + self.value_id = ApplicationSetting.RepeatModeStatus(self.value_id) + case ApplicationSetting.AttributeId.SHUFFLE_ON_OFF: + self.value_id = ApplicationSetting.ShuffleOnOffStatus(self.value_id) + case ApplicationSetting.AttributeId.SCAN_ON_OFF: + self.value_id = ApplicationSetting.ScanOnOffStatus(self.value_id) + case _: + self.value_id = ApplicationSetting.GenericValue(self.value_id) player_application_settings: Sequence[Setting] = field( metadata=hci.metadata(Setting.parse_from_bytes, list_begin=True, list_end=True) @@ -1619,6 +1629,8 @@ supported_events: list[EventId] supported_company_ids: list[int] + supported_player_app_settings: dict[ApplicationSetting.AttributeId, list[int]] + player_app_settings: dict[ApplicationSetting.AttributeId, int] volume: int playback_status: PlayStatus @@ -1626,11 +1638,23 @@ self, supported_events: Iterable[EventId] = (), supported_company_ids: Iterable[int] = (AVRCP_BLUETOOTH_SIG_COMPANY_ID,), + supported_player_app_settings: ( + Mapping[ApplicationSetting.AttributeId, Sequence[int]] | None + ) = None, ) -> None: self.supported_company_ids = list(supported_company_ids) self.supported_events = list(supported_events) self.volume = 0 self.playback_status = PlayStatus.STOPPED + self.supported_player_app_settings = ( + {key: list(value) for key, value in supported_player_app_settings.items()} + if supported_player_app_settings + else {} + ) + self.player_app_settings = {} + self.uid_counter = 0 + self.addressed_player_id = 0 + self.current_track_uid = TrackChangedEvent.NO_TRACK async def get_supported_events(self) -> list[EventId]: return self.supported_events @@ -1663,6 +1687,38 @@ async def get_playback_status(self) -> PlayStatus: return self.playback_status + async def get_supported_player_app_settings( + self, + ) -> dict[ApplicationSetting.AttributeId, list[int]]: + return self.supported_player_app_settings + + async def get_current_player_app_settings( + self, + ) -> dict[ApplicationSetting.AttributeId, int]: + return self.player_app_settings + + async def set_player_app_settings( + self, attribute: ApplicationSetting.AttributeId, value: int + ) -> None: + self.player_app_settings[attribute] = value + + async def play_item(self, scope: Scope, uid: int, uid_counter: int) -> None: + logger.debug( + "@@@ play_item: scope=%s, uid=%s, uid_counter=%s", + scope, + uid, + uid_counter, + ) + + async def get_uid_counter(self) -> int: + return self.uid_counter + + async def get_addressed_player_id(self) -> int: + return self.addressed_player_id + + async def get_current_track_uid(self) -> int: + return self.current_track_uid + # TODO add other delegate methods @@ -1910,6 +1966,51 @@ response = self._check_response(response_context, GetElementAttributesResponse) return list(response.attributes) + async def list_supported_player_app_settings( + self, attribute_ids: Sequence[ApplicationSetting.AttributeId] = () + ) -> dict[ApplicationSetting.AttributeId, list[int]]: + """Get element attributes from the connected peer.""" + response_context = await self.send_avrcp_command( + avc.CommandFrame.CommandType.STATUS, + ListPlayerApplicationSettingAttributesCommand(), + ) + if not attribute_ids: + list_attribute_response = self._check_response( + response_context, ListPlayerApplicationSettingAttributesResponse + ) + attribute_ids = list_attribute_response.attribute + + supported_settings: dict[ApplicationSetting.AttributeId, list[int]] = {} + for attribute_id in attribute_ids: + response_context = await self.send_avrcp_command( + avc.CommandFrame.CommandType.STATUS, + ListPlayerApplicationSettingValuesCommand(attribute_id), + ) + list_value_response = self._check_response( + response_context, ListPlayerApplicationSettingValuesResponse + ) + supported_settings[attribute_id] = list(list_value_response.value) + + return supported_settings + + async def get_player_app_settings( + self, attribute_ids: Sequence[ApplicationSetting.AttributeId] + ) -> dict[ApplicationSetting.AttributeId, int]: + """Get element attributes from the connected peer.""" + response_context = await self.send_avrcp_command( + avc.CommandFrame.CommandType.STATUS, + GetCurrentPlayerApplicationSettingValueCommand(attribute_ids), + ) + response: GetCurrentPlayerApplicationSettingValueResponse = ( + self._check_response( + response_context, GetCurrentPlayerApplicationSettingValueResponse + ) + ) + return { + attribute_id: value + for attribute_id, value in zip(response.attribute, response.value) + } + async def monitor_events( self, event_id: EventId, playback_interval: int = 0 ) -> AsyncIterator[Event]: @@ -1961,13 +2062,13 @@ async def monitor_track_changed( self, - ) -> AsyncIterator[bytes]: + ) -> AsyncIterator[int]: """Monitor Track changes from the connected peer.""" async for event in self.monitor_events(EventId.TRACK_CHANGED, 0): if not isinstance(event, TrackChangedEvent): logger.warning("unexpected event class") continue - yield event.identifier + yield event.uid async def monitor_playback_position( self, playback_interval: int @@ -2060,11 +2161,9 @@ """Notify the connected peer of a Playback Status change.""" self.notify_event(PlaybackStatusChangedEvent(status)) - def notify_track_changed(self, identifier: bytes) -> None: + def notify_track_changed(self, uid: int) -> None: """Notify the connected peer of a Track change.""" - if len(identifier) != 8: - raise core.InvalidArgumentError("identifier must be 8 bytes") - self.notify_event(TrackChangedEvent(identifier)) + self.notify_event(TrackChangedEvent(uid)) def notify_playback_position_changed(self, position: int) -> None: """Notify the connected peer of a Position change.""" @@ -2280,21 +2379,40 @@ ): # TODO: catch exceptions from delegates command = Command.from_bytes(pdu_id, pdu) - if isinstance(command, GetCapabilitiesCommand): - self._on_get_capabilities_command(transaction_label, command) - elif isinstance(command, SetAbsoluteVolumeCommand): - self._on_set_absolute_volume_command(transaction_label, command) - elif isinstance(command, RegisterNotificationCommand): - self._on_register_notification_command(transaction_label, command) - elif isinstance(command, GetPlayStatusCommand): - self._on_get_play_status_command(transaction_label, command) - else: - # Not supported. - # TODO: check that this is the right way to respond in this case. - logger.debug("unsupported PDU ID") - self.send_rejected_avrcp_response( - transaction_label, pdu_id, StatusCode.INVALID_PARAMETER - ) + match command: + case GetCapabilitiesCommand(): + self._on_get_capabilities_command(transaction_label, command) + case SetAbsoluteVolumeCommand(): + self._on_set_absolute_volume_command(transaction_label, command) + case RegisterNotificationCommand(): + self._on_register_notification_command(transaction_label, command) + case GetPlayStatusCommand(): + self._on_get_play_status_command(transaction_label, command) + case ListPlayerApplicationSettingAttributesCommand(): + self._on_list_player_application_setting_attributes_command( + transaction_label, command + ) + case ListPlayerApplicationSettingValuesCommand(): + self._on_list_player_application_setting_values_command( + transaction_label, command + ) + case SetPlayerApplicationSettingValueCommand(): + self._on_set_player_application_setting_value_command( + transaction_label, command + ) + case GetCurrentPlayerApplicationSettingValueCommand(): + self._on_get_current_player_application_setting_value_command( + transaction_label, command + ) + case PlayItemCommand(): + self._on_play_item_command(transaction_label, command) + case _: + # Not supported. + # TODO: check that this is the right way to respond in this case. + logger.debug("unsupported PDU ID") + self.send_rejected_avrcp_response( + transaction_label, pdu_id, StatusCode.INVALID_PARAMETER + ) else: logger.debug("unsupported command type") self.send_rejected_avrcp_response( @@ -2322,26 +2440,29 @@ # is Ok, but if/when more responses are supported, a lookup mechanism would be # more appropriate. response: Response | None = None - if response_code == avc.ResponseFrame.ResponseCode.REJECTED: - response = RejectedResponse(pdu_id=pdu_id, status_code=StatusCode(pdu[0])) - elif response_code == avc.ResponseFrame.ResponseCode.NOT_IMPLEMENTED: - response = NotImplementedResponse(pdu_id=pdu_id, parameters=pdu) - elif response_code in ( - avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE, - avc.ResponseFrame.ResponseCode.INTERIM, - avc.ResponseFrame.ResponseCode.CHANGED, - avc.ResponseFrame.ResponseCode.ACCEPTED, - ): - response = Response.from_bytes(pdu=pdu, pdu_id=PduId(pdu_id)) - else: - logger.debug("unexpected response code") - pending_command.response.set_exception( - core.ProtocolError( - error_code=None, - error_namespace="avrcp", - details="unexpected response code", + match response_code: + case avc.ResponseFrame.ResponseCode.REJECTED: + response = RejectedResponse( + pdu_id=pdu_id, status_code=StatusCode(pdu[0]) ) - ) + case avc.ResponseFrame.ResponseCode.NOT_IMPLEMENTED: + response = NotImplementedResponse(pdu_id=pdu_id, parameters=pdu) + case ( + avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE + | avc.ResponseFrame.ResponseCode.INTERIM + | avc.ResponseFrame.ResponseCode.CHANGED + | avc.ResponseFrame.ResponseCode.ACCEPTED + ): + response = Response.from_bytes(pdu=pdu, pdu_id=PduId(pdu_id)) + case _: + logger.debug("unexpected response code") + pending_command.response.set_exception( + core.ProtocolError( + error_code=None, + error_namespace="avrcp", + details="unexpected response code", + ) + ) if response is None: self.recycle_pending_command(pending_command) @@ -2512,22 +2633,18 @@ async def get_supported_events() -> None: capabilities: Sequence[bytes | SupportsBytes] - if ( - command.capability_id - == GetCapabilitiesCommand.CapabilityId.EVENTS_SUPPORTED - ): - capabilities = await self.delegate.get_supported_events() - elif ( - command.capability_id == GetCapabilitiesCommand.CapabilityId.COMPANY_ID - ): - company_ids = await self.delegate.get_supported_company_ids() - capabilities = [ - company_id.to_bytes(3, 'big') for company_id in company_ids - ] - else: - raise core.InvalidArgumentError( - f"Unsupported capability: {command.capability_id}" - ) + match command.capability_id: + case GetCapabilitiesCommand.CapabilityId.EVENTS_SUPPORTED: + capabilities = await self.delegate.get_supported_events() + case GetCapabilitiesCommand.CapabilityId.EVENTS_SUPPORTED.COMPANY_ID: + company_ids = await self.delegate.get_supported_company_ids() + capabilities = [ + company_id.to_bytes(3, 'big') for company_id in company_ids + ] + case _: + raise core.InvalidArgumentError( + f"Unsupported capability: {command.capability_id}" + ) self.send_avrcp_response( transaction_label, avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE, @@ -2572,6 +2689,121 @@ self._delegate_command(transaction_label, command, get_playback_status()) + def _on_list_player_application_setting_attributes_command( + self, + transaction_label: int, + command: ListPlayerApplicationSettingAttributesCommand, + ) -> None: + logger.debug("<<< AVRCP command PDU: %s", command) + + async def get_supported_player_app_settings() -> None: + supported_settings = await self.delegate.get_supported_player_app_settings() + self.send_avrcp_response( + transaction_label, + avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE, + ListPlayerApplicationSettingAttributesResponse( + list(supported_settings.keys()) + ), + ) + + self._delegate_command( + transaction_label, command, get_supported_player_app_settings() + ) + + def _on_list_player_application_setting_values_command( + self, + transaction_label: int, + command: ListPlayerApplicationSettingValuesCommand, + ) -> None: + logger.debug("<<< AVRCP command PDU: %s", command) + + async def get_supported_player_app_settings() -> None: + supported_settings = await self.delegate.get_supported_player_app_settings() + self.send_avrcp_response( + transaction_label, + avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE, + ListPlayerApplicationSettingValuesResponse( + supported_settings.get(command.attribute, []) + ), + ) + + self._delegate_command( + transaction_label, command, get_supported_player_app_settings() + ) + + def _on_get_current_player_application_setting_value_command( + self, + transaction_label: int, + command: GetCurrentPlayerApplicationSettingValueCommand, + ) -> None: + logger.debug("<<< AVRCP command PDU: %s", command) + + async def get_supported_player_app_settings() -> None: + current_settings = await self.delegate.get_current_player_app_settings() + + if not all( + attribute in current_settings for attribute in command.attribute + ): + self.send_not_implemented_avrcp_response( + transaction_label, + PduId.GET_CURRENT_PLAYER_APPLICATION_SETTING_VALUE, + ) + return + + self.send_avrcp_response( + transaction_label, + avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE, + GetCurrentPlayerApplicationSettingValueResponse( + attribute=command.attribute, + value=[ + current_settings[attribute] for attribute in command.attribute + ], + ), + ) + + self._delegate_command( + transaction_label, command, get_supported_player_app_settings() + ) + + def _on_set_player_application_setting_value_command( + self, + transaction_label: int, + command: SetPlayerApplicationSettingValueCommand, + ) -> None: + logger.debug("<<< AVRCP command PDU: %s", command) + + async def set_player_app_settings() -> None: + for attribute, value in zip(command.attribute, command.value): + await self.delegate.set_player_app_settings(attribute, value) + + self.send_avrcp_response( + transaction_label, + avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE, + SetPlayerApplicationSettingValueResponse(), + ) + + self._delegate_command(transaction_label, command, set_player_app_settings()) + + def _on_play_item_command( + self, + transaction_label: int, + command: PlayItemCommand, + ) -> None: + logger.debug("<<< AVRCP command PDU: %s", command) + + async def play_item() -> None: + await self.delegate.play_item( + scope=command.scope, uid=command.uid, uid_counter=command.uid_counter + ) + + self.send_avrcp_response( + transaction_label, + avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE, + PlayItemResponse(status=StatusCode.OPERATION_COMPLETED), + ) + + self._delegate_command(transaction_label, command, play_item()) + def _on_register_notification_command( self, transaction_label: int, command: RegisterNotificationCommand ) -> None: @@ -2587,26 +2819,51 @@ ) return - response: Response - if command.event_id == EventId.VOLUME_CHANGED: - volume = await self.delegate.get_absolute_volume() - response = RegisterNotificationResponse(VolumeChangedEvent(volume)) - elif command.event_id == EventId.PLAYBACK_STATUS_CHANGED: - playback_status = await self.delegate.get_playback_status() - response = RegisterNotificationResponse( - PlaybackStatusChangedEvent(play_status=playback_status) - ) - elif command.event_id == EventId.NOW_PLAYING_CONTENT_CHANGED: - playback_status = await self.delegate.get_playback_status() - response = RegisterNotificationResponse(NowPlayingContentChangedEvent()) - else: - logger.warning("Event supported but not handled %s", command.event_id) - return + event: Event + match command.event_id: + case EventId.VOLUME_CHANGED: + volume = await self.delegate.get_absolute_volume() + event = VolumeChangedEvent(volume) + case EventId.PLAYBACK_STATUS_CHANGED: + playback_status = await self.delegate.get_playback_status() + event = PlaybackStatusChangedEvent(play_status=playback_status) + case EventId.NOW_PLAYING_CONTENT_CHANGED: + event = NowPlayingContentChangedEvent() + case EventId.PLAYER_APPLICATION_SETTING_CHANGED: + settings = await self.delegate.get_current_player_app_settings() + event = PlayerApplicationSettingChangedEvent( + [ + PlayerApplicationSettingChangedEvent.Setting( + attribute, value # type: ignore + ) + for attribute, value in settings.items() + ] + ) + case EventId.AVAILABLE_PLAYERS_CHANGED: + event = AvailablePlayersChangedEvent() + case EventId.ADDRESSED_PLAYER_CHANGED: + event = AddressedPlayerChangedEvent( + AddressedPlayerChangedEvent.Player( + player_id=await self.delegate.get_addressed_player_id(), + uid_counter=await self.delegate.get_uid_counter(), + ) + ) + case EventId.UIDS_CHANGED: + event = UidsChangedEvent(await self.delegate.get_uid_counter()) + case EventId.TRACK_CHANGED: + event = TrackChangedEvent( + await self.delegate.get_current_track_uid() + ) + case _: + logger.warning( + "Event supported but not handled %s", command.event_id + ) + return self.send_avrcp_response( transaction_label, avc.ResponseFrame.ResponseCode.INTERIM, - response, + RegisterNotificationResponse(event), ) self._register_notification_listener(transaction_label, command)
diff --git a/bumble/controller.py b/bumble/controller.py index 12e997e..48329b1 100644 --- a/bumble/controller.py +++ b/bumble/controller.py
@@ -241,7 +241,7 @@ lmp_features: bytes = bytes.fromhex( '0000000060000000' ) # BR/EDR Not Supported, LE Supported (Controller) - manufacturer_name: int = 0xFFFF + manufacturer_company_identifier: int = 0xFFFF acl_data_packet_length: int = 27 total_num_acl_data_packets: int = 64 le_acl_data_packet_length: int = 27 @@ -403,27 +403,33 @@ ) # If the packet is a command, invoke the handler for this packet - if isinstance(packet, hci.HCI_Command): - self.on_hci_command_packet(packet) - elif isinstance(packet, hci.HCI_AclDataPacket): - self.on_hci_acl_data_packet(packet) - elif isinstance(packet, hci.HCI_Event): - self.on_hci_event_packet(packet) - else: - logger.warning(f'!!! unknown packet type {packet.hci_packet_type}') + match packet: + case hci.HCI_Command(): + self.on_hci_command_packet(packet) + case hci.HCI_AclDataPacket(): + self.on_hci_acl_data_packet(packet) + case hci.HCI_Event(): + self.on_hci_event_packet(packet) + case _: + logger.warning(f'!!! unknown packet type {packet.hci_packet_type}') def on_hci_command_packet(self, command: hci.HCI_Command) -> None: handler_name = f'on_{command.name.lower()}' handler = getattr(self, handler_name, self.on_hci_command) - result: bytes | None = handler(command) - if isinstance(result, bytes): + result: hci.HCI_ReturnParameters | None = handler(command) + if isinstance(command, hci.HCI_SyncCommand): + if result is None: + logger.error("Sync command handlers should return parameters, got None") + return self.send_hci_packet( hci.HCI_Command_Complete_Event( num_hci_command_packets=1, command_opcode=command.op_code, - return_parameters=hci.HCI_GenericReturnParameters(data=result), + return_parameters=result, ) ) + elif result is not None: + logger.error("Async command handlers should return None, got %s", result) def on_hci_event_packet(self, _event: hci.HCI_Packet) -> None: logger.warning('!!! unexpected event packet') @@ -448,6 +454,13 @@ if self.host: asyncio.get_running_loop().call_soon(self.host.on_packet, bytes(packet)) + def _send_hci_command_status(self, status: int, op_code: int) -> None: + self.send_hci_packet( + hci.HCI_Command_Status_Event( + status=status, num_hci_command_packets=1, command_opcode=op_code + ) + ) + # This method allows the controller to emulate the same API as a transport source async def wait_for_termination(self) -> None: await self.terminated @@ -481,6 +494,18 @@ return connection return None + def find_le_connection_by_handle(self, handle: int) -> Connection | None: + for connection in self.le_connections.values(): + if connection.handle == handle: + return connection + return None + + def find_classic_connection_by_handle(self, handle: int) -> Connection | None: + for connection in self.classic_connections.values(): + if connection.handle == handle: + return connection + return None + def find_classic_sco_link_by_handle(self, handle: int) -> ScoLink | None: for connection in self.sco_links.values(): if connection.handle == handle: @@ -505,26 +530,42 @@ logger.error("Cannot find a connection for %s", sender_address) return - if isinstance(packet, ll.TerminateInd): - self.on_le_disconnected(connection, packet.error_code) - elif isinstance(packet, ll.CisReq): - self.on_le_cis_request(connection, packet.cig_id, packet.cis_id) - elif isinstance(packet, ll.CisRsp): - self.on_le_cis_established(packet.cig_id, packet.cis_id) - connection.send_ll_control_pdu(ll.CisInd(packet.cig_id, packet.cis_id)) - elif isinstance(packet, ll.CisInd): - self.on_le_cis_established(packet.cig_id, packet.cis_id) - elif isinstance(packet, ll.CisTerminateInd): - self.on_le_cis_disconnected(packet.cig_id, packet.cis_id) - elif isinstance(packet, ll.EncReq): - self.on_le_encrypted(connection) + match packet: + case ll.TerminateInd(): + self.on_le_disconnected(connection, packet.error_code) + case ll.CisReq(): + self.on_le_cis_request(connection, packet.cig_id, packet.cis_id) + case ll.CisRsp(): + self.on_le_cis_established(packet.cig_id, packet.cis_id) + connection.send_ll_control_pdu(ll.CisInd(packet.cig_id, packet.cis_id)) + case ll.CisInd(): + self.on_le_cis_established(packet.cig_id, packet.cis_id) + case ll.CisTerminateInd(): + self.on_le_cis_disconnected(packet.cig_id, packet.cis_id) + case ll.EncReq(): + self.on_le_encrypted(connection) + case ll.FeatureReq() | ll.PeripheralFeatureReq(): + connection.send_ll_control_pdu( + ll.FeatureRsp( + feature_set=self.le_features.value.to_bytes(8, 'little') + ) + ) + case ll.FeatureRsp(feature_set): + self.send_hci_packet( + hci.HCI_LE_Read_Remote_Features_Complete_Event( + status=hci.HCI_ErrorCode.SUCCESS, + connection_handle=connection.handle, + le_features=feature_set, + ) + ) def on_ll_advertising_pdu(self, packet: ll.AdvertisingPdu) -> None: logger.debug("[%s] <<< Advertising PDU: %s", self.name, packet) - if isinstance(packet, ll.ConnectInd): - self.on_le_connect_ind(packet) - elif isinstance(packet, (ll.AdvInd, ll.AdvExtInd)): - self.on_advertising_pdu(packet) + match packet: + case ll.ConnectInd(): + self.on_le_connect_ind(packet) + case ll.AdvInd() | ll.AdvExtInd(): + self.on_advertising_pdu(packet) def on_le_connect_ind(self, packet: ll.ConnectInd) -> None: ''' @@ -571,7 +612,7 @@ # Then say that the connection has completed self.send_hci_packet( hci.HCI_LE_Connection_Complete_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=connection.handle, role=connection.role, peer_address_type=peer_address.address_type, @@ -586,7 +627,7 @@ if isinstance(advertiser, AdvertisingSet): self.send_hci_packet( hci.HCI_LE_Advertising_Set_Terminated_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, advertising_handle=advertiser.handle, connection_handle=connection.handle, num_completed_extended_advertising_events=0, @@ -598,12 +639,14 @@ # Send a disconnection complete event self.send_hci_packet( hci.HCI_Disconnection_Complete_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=connection.handle, reason=reason, ) ) + del self.le_connections[connection.peer_address] + def create_le_connection(self, peer_address: hci.Address) -> None: ''' Called when we receive advertisement matching connection filter. @@ -661,7 +704,7 @@ self.send_hci_packet( # pylint: disable=line-too-long hci.HCI_LE_Connection_Complete_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=connection.handle if connection else 0, role=hci.Role.CENTRAL, peer_address_type=peer_address.address_type, @@ -808,7 +851,7 @@ self.send_hci_packet( hci.HCI_LE_CIS_Established_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=cis_link.handle, # CIS parameters are ignored. cig_sync_delay=0, @@ -858,9 +901,9 @@ self.send_hci_packet( hci.HCI_Disconnection_Complete_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=cis_link.handle, - reason=hci.HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR, + reason=hci.HCI_ErrorCode.REMOTE_USER_TERMINATED_CONNECTION_ERROR, ) ) @@ -879,52 +922,54 @@ ] = loop.create_future() return future - def on_lmp_packet(self, sender_address: hci.Address, packet: lmp.Packet): - if isinstance(packet, (lmp.LmpAccepted, lmp.LmpAcceptedExt)): - if future := self.classic_pending_commands.setdefault( - sender_address, {} - ).get(packet.response_opcode): - future.set_result(hci.HCI_SUCCESS) - else: + def on_lmp_packet(self, sender_address: hci.Address, packet: lmp.Packet) -> None: + match packet: + case lmp.LmpAccepted() | lmp.LmpAcceptedExt(): + if future := self.classic_pending_commands.setdefault( + sender_address, {} + ).get(packet.response_opcode): + future.set_result(hci.HCI_ErrorCode.SUCCESS) + else: + logger.error("!!! Unhandled packet: %s", packet) + case lmp.LmpNotAccepted() | lmp.LmpNotAcceptedExt(): + if future := self.classic_pending_commands.setdefault( + sender_address, {} + ).get(packet.response_opcode): + future.set_result(packet.error_code) + else: + logger.error("!!! Unhandled packet: %s", packet) + case lmp.LmpHostConnectionReq(): + self.on_classic_connection_request( + sender_address, hci.HCI_Connection_Complete_Event.LinkType.ACL + ) + case lmp.LmpScoLinkReq(): + self.on_classic_connection_request( + sender_address, hci.HCI_Connection_Complete_Event.LinkType.SCO + ) + case lmp.LmpEscoLinkReq(): + self.on_classic_connection_request( + sender_address, hci.HCI_Connection_Complete_Event.LinkType.ESCO + ) + case lmp.LmpDetach(): + self.on_classic_disconnected( + sender_address, + hci.HCI_ErrorCode.REMOTE_USER_TERMINATED_CONNECTION_ERROR, + ) + case lmp.LmpSwitchReq(): + self.on_classic_role_change_request(sender_address) + case lmp.LmpRemoveScoLinkReq() | lmp.LmpRemoveEscoLinkReq(): + self.on_classic_sco_disconnected(sender_address, packet.error_code) + case lmp.LmpNameReq(): + self.on_classic_remote_name_request(sender_address, packet.name_offset) + case lmp.LmpNameRes(): + self.on_classic_remote_name_response( + sender_address, + packet.name_offset, + packet.name_length, + packet.name_fregment, + ) + case _: logger.error("!!! Unhandled packet: %s", packet) - elif isinstance(packet, (lmp.LmpNotAccepted, lmp.LmpNotAcceptedExt)): - if future := self.classic_pending_commands.setdefault( - sender_address, {} - ).get(packet.response_opcode): - future.set_result(packet.error_code) - else: - logger.error("!!! Unhandled packet: %s", packet) - elif isinstance(packet, (lmp.LmpHostConnectionReq)): - self.on_classic_connection_request( - sender_address, hci.HCI_Connection_Complete_Event.LinkType.ACL - ) - elif isinstance(packet, (lmp.LmpScoLinkReq)): - self.on_classic_connection_request( - sender_address, hci.HCI_Connection_Complete_Event.LinkType.SCO - ) - elif isinstance(packet, (lmp.LmpEscoLinkReq)): - self.on_classic_connection_request( - sender_address, hci.HCI_Connection_Complete_Event.LinkType.ESCO - ) - elif isinstance(packet, (lmp.LmpDetach)): - self.on_classic_disconnected( - sender_address, hci.HCI_REMOTE_USER_TERMINATED_CONNECTION_ERROR - ) - elif isinstance(packet, (lmp.LmpSwitchReq)): - self.on_classic_role_change_request(sender_address) - elif isinstance(packet, (lmp.LmpRemoveScoLinkReq, lmp.LmpRemoveEscoLinkReq)): - self.on_classic_sco_disconnected(sender_address, packet.error_code) - elif isinstance(packet, lmp.LmpNameReq): - self.on_classic_remote_name_request(sender_address, packet.name_offset) - elif isinstance(packet, lmp.LmpNameRes): - self.on_classic_remote_name_response( - sender_address, - packet.name_offset, - packet.name_length, - packet.name_fregment, - ) - else: - logger.error("!!! Unhandled packet: %s", packet) def on_classic_connection_request( self, peer_address: hci.Address, link_type: int @@ -958,7 +1003,7 @@ def on_classic_connection_complete( self, peer_address: hci.Address, status: int ) -> None: - if status == hci.HCI_SUCCESS: + if status == hci.HCI_ErrorCode.SUCCESS: # Allocate (or reuse) a connection handle peer_address = peer_address connection_handle = self.allocate_connection_handle() @@ -1005,7 +1050,7 @@ if connection := self.classic_connections.pop(peer_address, None): self.send_hci_packet( hci.HCI_Disconnection_Complete_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=connection.handle, reason=reason, ) @@ -1020,7 +1065,7 @@ if sco_link := self.sco_links.pop(peer_address, None): self.send_hci_packet( hci.HCI_Disconnection_Complete_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=sco_link.handle, reason=reason, ) @@ -1034,7 +1079,8 @@ self.send_lmp_packet( peer_address, lmp.LmpNotAccepted( - lmp.Opcode.LMP_SWITCH_REQ, hci.HCI_ROLE_CHANGE_NOT_ALLOWED_ERROR + lmp.Opcode.LMP_SWITCH_REQ, + hci.HCI_ErrorCode.ROLE_CHANGE_NOT_ALLOWED_ERROR, ), ) else: @@ -1053,7 +1099,7 @@ connection.role = new_role self.send_hci_packet( hci.HCI_Role_Change_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, bd_addr=connection.peer_address, new_role=new_role, ) @@ -1062,7 +1108,7 @@ def on_classic_sco_connection_complete( self, peer_address: hci.Address, status: int, link_type: int ) -> None: - if status == hci.HCI_SUCCESS: + if status == hci.HCI_ErrorCode.SUCCESS: # Allocate (or reuse) a connection handle connection_handle = self.allocate_connection_handle() sco_link = ScoLink( @@ -1092,7 +1138,7 @@ def on_classic_remote_name_request( self, peer_address: hci.Address, name_offset: int - ): + ) -> None: self.send_lmp_packet( peer_address, lmp.LmpNameRes( @@ -1108,10 +1154,10 @@ name_offset: int, name_length: int, name_fregment: bytes, - ): + ) -> None: self.send_hci_packet( hci.HCI_Remote_Name_Request_Complete_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, bd_addr=peer_address, remote_name=name_fregment, ) @@ -1130,13 +1176,17 @@ ############################################################ # HCI handlers ############################################################ - def on_hci_command(self, command: hci.HCI_Command) -> bytes | None: + def on_hci_command( + self, command: hci.HCI_Command + ) -> hci.HCI_StatusReturnParameters: logger.warning(color(f'--- Unsupported command {command}', 'red')) - return bytes([hci.HCI_UNKNOWN_HCI_COMMAND_ERROR]) + return hci.HCI_StatusReturnParameters( + hci.HCI_ErrorCode.UNKNOWN_HCI_COMMAND_ERROR + ) def on_hci_create_connection_command( self, command: hci.HCI_Create_Connection_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.1.5 Create Connection command ''' @@ -1147,12 +1197,8 @@ # Check that we don't already have a pending connection if self.pending_le_connection: - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_CONTROLLER_BUSY_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.CONTROLLER_BUSY_ERROR, command.op_code ) return None @@ -1169,54 +1215,41 @@ ) # Say that the connection is pending - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_STATUS_PENDING, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code) future = self.send_lmp_packet(command.bd_addr, lmp.LmpHostConnectionReq()) - def on_response(future: asyncio.Future[int]): + def on_response(future: asyncio.Future[int]) -> None: self.on_classic_connection_complete(command.bd_addr, future.result()) future.add_done_callback(on_response) return None - def on_hci_disconnect_command( - self, command: hci.HCI_Disconnect_Command - ) -> bytes | None: + def on_hci_disconnect_command(self, command: hci.HCI_Disconnect_Command) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.1.6 Disconnect Command ''' # First, say that the disconnection is pending - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_STATUS_PENDING, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code) # Notify the link of the disconnection handle = command.connection_handle - if connection := self.find_connection_by_handle(handle): + if connection := self.find_classic_connection_by_handle(handle): if self.link: - if connection.transport == PhysicalTransport.BR_EDR: - self.send_lmp_packet( - connection.peer_address, - lmp.LmpDetach(command.reason), - ) - self.on_classic_disconnected( - connection.peer_address, command.reason - ) - else: - connection.send_ll_control_pdu(ll.TerminateInd(command.reason)) - self.on_le_disconnected(connection, command.reason) + self.send_lmp_packet( + connection.peer_address, + lmp.LmpDetach(command.reason), + ) + self.on_classic_disconnected(connection.peer_address, command.reason) else: # Remove the connection del self.classic_connections[connection.peer_address] + elif connection := self.find_le_connection_by_handle(handle): + if self.link: + connection.send_ll_control_pdu(ll.TerminateInd(command.reason)) + self.on_le_disconnected(connection, command.reason) + else: + # Remove the connection + del self.le_connections[connection.peer_address] elif sco_link := self.find_classic_sco_link_by_handle(handle): if self.link: if ( @@ -1256,7 +1289,7 @@ def on_hci_accept_connection_request_command( self, command: hci.HCI_Accept_Connection_Request_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.1.8 Accept Connection Request command ''' @@ -1265,28 +1298,18 @@ return None if not (connection := self.classic_connections.get(command.bd_addr)): - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.UNKNOWN_CONNECTION_IDENTIFIER_ERROR, command.op_code ) return None - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_SUCCESS, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_ErrorCode.SUCCESS, command.op_code) if command.role == hci.Role.CENTRAL: # Perform role switching before accept. future = self.send_lmp_packet(command.bd_addr, lmp.LmpSwitchReq()) - def on_response(future: asyncio.Future[int]): - if (status := future.result()) == hci.HCI_SUCCESS: + def on_response(future: asyncio.Future[int]) -> None: + if (status := future.result()) == hci.HCI_ErrorCode.SUCCESS: self.classic_role_change(connection) # Continue connection setup. self.send_lmp_packet( @@ -1309,22 +1332,18 @@ command.bd_addr, lmp.LmpAccepted(lmp.Opcode.LMP_HOST_CONNECTION_REQ), ) - self.on_classic_connection_complete(command.bd_addr, hci.HCI_SUCCESS) + self.on_classic_connection_complete( + command.bd_addr, hci.HCI_ErrorCode.SUCCESS + ) return None def on_hci_remote_name_request_command( self, command: hci.HCI_Remote_Name_Request_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.1.19 Remote Name Request command ''' - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_SUCCESS, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_ErrorCode.SUCCESS, command.op_code) self.send_lmp_packet(command.bd_addr, lmp.LmpNameReq(0)) @@ -1332,7 +1351,7 @@ def on_hci_enhanced_setup_synchronous_connection_command( self, command: hci.HCI_Enhanced_Setup_Synchronous_Connection_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.1.45 Enhanced Setup Synchronous Connection command ''' @@ -1343,22 +1362,12 @@ if not ( connection := self.find_connection_by_handle(command.connection_handle) ): - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.UNKNOWN_CONNECTION_IDENTIFIER_ERROR, command.op_code ) return None - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_SUCCESS, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_ErrorCode.SUCCESS, command.op_code) future = self.send_lmp_packet( connection.peer_address, lmp.LmpEscoLinkReq( @@ -1377,7 +1386,7 @@ ), ) - def on_response(future: asyncio.Future[int]): + def on_response(future: asyncio.Future[int]) -> None: self.on_classic_sco_connection_complete( connection.peer_address, future.result(), @@ -1389,7 +1398,7 @@ def on_hci_enhanced_accept_synchronous_connection_request_command( self, command: hci.HCI_Enhanced_Accept_Synchronous_Connection_Request_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.1.46 Enhanced Accept Synchronous Connection Request command ''' @@ -1398,59 +1407,37 @@ return None if not (connection := self.classic_connections.get(command.bd_addr)): - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.UNKNOWN_CONNECTION_IDENTIFIER_ERROR, command.op_code ) return None - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_SUCCESS, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_ErrorCode.SUCCESS, command.op_code) self.send_lmp_packet( connection.peer_address, lmp.LmpAcceptedExt(lmp.Opcode.LMP_ESCO_LINK_REQ), ) self.on_classic_sco_connection_complete( connection.peer_address, - hci.HCI_SUCCESS, + hci.HCI_ErrorCode.SUCCESS, hci.HCI_Connection_Complete_Event.LinkType.ESCO, ) return None - def on_hci_sniff_mode_command( - self, command: hci.HCI_Sniff_Mode_Command - ) -> bytes | None: + def on_hci_sniff_mode_command(self, command: hci.HCI_Sniff_Mode_Command) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.2.2 Sniff Mode command ''' if self.link is None: - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.UNKNOWN_CONNECTION_IDENTIFIER_ERROR, command.op_code ) return None - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_SUCCESS, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_ErrorCode.SUCCESS, command.op_code) self.send_hci_packet( hci.HCI_Mode_Change_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=command.connection_handle, current_mode=hci.HCI_Mode_Change_Event.Mode.SNIFF, interval=2, @@ -1460,31 +1447,21 @@ def on_hci_exit_sniff_mode_command( self, command: hci.HCI_Exit_Sniff_Mode_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.2.3 Exit Sniff Mode command ''' if self.link is None: - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.UNKNOWN_CONNECTION_IDENTIFIER_ERROR, command.op_code ) return None - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_SUCCESS, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_ErrorCode.SUCCESS, command.op_code) self.send_hci_packet( hci.HCI_Mode_Change_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=command.connection_handle, current_mode=hci.HCI_Mode_Change_Event.Mode.ACTIVE, interval=2, @@ -1492,9 +1469,7 @@ ) return None - def on_hci_switch_role_command( - self, command: hci.HCI_Switch_Role_Command - ) -> bytes | None: + def on_hci_switch_role_command(self, command: hci.HCI_Switch_Role_Command) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.2.8 Switch hci.Role command ''' @@ -1504,21 +1479,11 @@ if connection := self.classic_connections.get(command.bd_addr): current_role = connection.role - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_SUCCESS, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_ErrorCode.SUCCESS, command.op_code) else: # Connection doesn't exist, reject. - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_DISALLOWED_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.COMMAND_DISALLOWED_ERROR, command.op_code ) return None @@ -1526,7 +1491,7 @@ if current_role == command.role: self.send_hci_packet( hci.HCI_Role_Change_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, bd_addr=command.bd_addr, new_role=current_role, ) @@ -1534,8 +1499,8 @@ else: future = self.send_lmp_packet(command.bd_addr, lmp.LmpSwitchReq()) - def on_response(future: asyncio.Future[int]): - if (status := future.result()) == hci.HCI_SUCCESS: + def on_response(future: asyncio.Future[int]) -> None: + if (status := future.result()) == hci.HCI_ErrorCode.SUCCESS: connection.role = hci.Role(command.role) self.send_hci_packet( hci.HCI_Role_Change_Event( @@ -1551,25 +1516,27 @@ def on_hci_set_event_mask_command( self, command: hci.HCI_Set_Event_Mask_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.1 Set Event Mask Command ''' self.event_mask = int.from_bytes( command.event_mask, byteorder='little', signed=False ) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) - def on_hci_reset_command(self, _command: hci.HCI_Reset_Command) -> bytes | None: + def on_hci_reset_command( + self, _command: hci.HCI_Reset_Command + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.2 Reset Command ''' # TODO: cleanup what needs to be reset - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_write_local_name_command( self, command: hci.HCI_Write_Local_Name_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.11 Write Local Name Command ''' @@ -1582,11 +1549,11 @@ self.local_name = str(local_name, 'utf-8') except UnicodeDecodeError: pass - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_read_local_name_command( self, _command: hci.HCI_Read_Local_Name_Command - ) -> bytes | None: + ) -> hci.HCI_Read_Local_Name_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.12 Read Local Name Command ''' @@ -1594,27 +1561,31 @@ if len(local_name) < 248: local_name = local_name + bytes(248 - len(local_name)) - return bytes([hci.HCI_SUCCESS]) + local_name + return hci.HCI_Read_Local_Name_ReturnParameters( + hci.HCI_ErrorCode.SUCCESS, local_name=local_name + ) def on_hci_read_class_of_device_command( self, _command: hci.HCI_Read_Class_Of_Device_Command - ) -> bytes | None: + ) -> hci.HCI_Read_Class_Of_Device_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.25 Read Class of Device Command ''' - return bytes([hci.HCI_SUCCESS, 0, 0, 0]) + return hci.HCI_Read_Class_Of_Device_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, class_of_device=0 + ) def on_hci_write_class_of_device_command( self, _command: hci.HCI_Write_Class_Of_Device_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.26 Write Class of Device Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_read_synchronous_flow_control_enable_command( self, _command: hci.HCI_Read_Synchronous_Flow_Control_Enable_Command - ) -> bytes | None: + ) -> hci.HCI_Read_Synchronous_Flow_Control_Enable_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.36 Read Synchronous Flow Control Enable Command @@ -1623,186 +1594,191 @@ ret = 1 else: ret = 0 - return bytes([hci.HCI_SUCCESS, ret]) + return hci.HCI_Read_Synchronous_Flow_Control_Enable_ReturnParameters( + hci.HCI_ErrorCode.SUCCESS, ret + ) def on_hci_write_synchronous_flow_control_enable_command( self, command: hci.HCI_Write_Synchronous_Flow_Control_Enable_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.37 Write Synchronous Flow Control Enable Command ''' - ret = hci.HCI_SUCCESS + ret = hci.HCI_ErrorCode.SUCCESS if command.synchronous_flow_control_enable == 1: self.sync_flow_control = True elif command.synchronous_flow_control_enable == 0: self.sync_flow_control = False else: - ret = hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR - return bytes([ret]) + ret = hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR + return hci.HCI_StatusReturnParameters(ret) def on_hci_set_controller_to_host_flow_control_command( self, _command: hci.HCI_Set_Controller_To_Host_Flow_Control_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.38 Set Controller To Host Flow Control Command ''' # For now we just accept the command but ignore the values. # TODO: respect the passed in values. - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_host_buffer_size_command( self, _command: hci.HCI_Host_Buffer_Size_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.39 Host Buffer Size Command ''' # For now we just accept the command but ignore the values. # TODO: respect the passed in values. - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_write_extended_inquiry_response_command( self, _command: hci.HCI_Write_Extended_Inquiry_Response_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.56 Write Extended Inquiry Response Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_write_simple_pairing_mode_command( self, _command: hci.HCI_Write_Simple_Pairing_Mode_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.59 Write Simple Pairing Mode Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_set_event_mask_page_2_command( self, command: hci.HCI_Set_Event_Mask_Page_2_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.69 Set Event Mask Page 2 Command ''' self.event_mask_page_2 = int.from_bytes( command.event_mask_page_2, byteorder='little', signed=False ) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_read_le_host_support_command( self, _command: hci.HCI_Read_LE_Host_Support_Command - ) -> bytes | None: + ) -> hci.HCI_Read_LE_Host_Support_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.78 Write LE Host Support Command ''' - return bytes([hci.HCI_SUCCESS, 1, 0]) + return hci.HCI_Read_LE_Host_Support_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, le_supported_host=1, unused=0 + ) def on_hci_write_le_host_support_command( self, _command: hci.HCI_Write_LE_Host_Support_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.79 Write LE Host Support Command ''' # TODO / Just ignore for now - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_write_authenticated_payload_timeout_command( self, command: hci.HCI_Write_Authenticated_Payload_Timeout_Command - ) -> bytes | None: + ) -> hci.HCI_StatusAndConnectionHandleReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.94 Write Authenticated Payload Timeout Command ''' # TODO - return struct.pack('<BH', hci.HCI_SUCCESS, command.connection_handle) + return hci.HCI_StatusAndConnectionHandleReturnParameters( + hci.HCI_ErrorCode.SUCCESS, command.connection_handle + ) def on_hci_read_local_version_information_command( self, _command: hci.HCI_Read_Local_Version_Information_Command - ) -> bytes | None: + ) -> hci.HCI_Read_Local_Version_Information_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.4.1 Read Local Version Information Command ''' - return struct.pack( - '<BBHBHH', - hci.HCI_SUCCESS, - self.hci_version, - self.hci_revision, - self.lmp_version, - self.manufacturer_name, - self.lmp_subversion, + return hci.HCI_Read_Local_Version_Information_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + hci_version=self.hci_version, + hci_subversion=self.hci_revision, + lmp_version=self.lmp_version, + company_identifier=self.manufacturer_company_identifier, + lmp_subversion=self.lmp_subversion, ) def on_hci_read_local_supported_commands_command( self, _command: hci.HCI_Read_Local_Supported_Commands_Command - ) -> bytes | None: + ) -> hci.HCI_Read_Local_Supported_Commands_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.4.2 Read Local Supported Commands Command ''' - return bytes([hci.HCI_SUCCESS]) + self.supported_commands + return hci.HCI_Read_Local_Supported_Commands_ReturnParameters( + hci.HCI_ErrorCode.SUCCESS, supported_commands=self.supported_commands + ) def on_hci_read_local_supported_features_command( self, _command: hci.HCI_Read_Local_Supported_Features_Command - ) -> bytes | None: + ) -> hci.HCI_Read_Local_Supported_Features_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.4.3 Read Local Supported Features Command ''' - return bytes([hci.HCI_SUCCESS]) + self.lmp_features[:8] + return hci.HCI_Read_Local_Supported_Features_ReturnParameters( + hci.HCI_ErrorCode.SUCCESS, lmp_features=self.lmp_features[:8] + ) def on_hci_read_local_extended_features_command( self, command: hci.HCI_Read_Local_Extended_Features_Command - ) -> bytes | None: + ) -> hci.HCI_Read_Local_Extended_Features_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.4.4 Read Local Extended Features Command ''' if command.page_number * 8 > len(self.lmp_features): - return bytes([hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR]) - return ( - bytes( - [ - # Status - hci.HCI_SUCCESS, - # Page number - command.page_number, - # Max page number - len(self.lmp_features) // 8 - 1, - ] + return hci.HCI_Read_Local_Extended_Features_ReturnParameters( + status=hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR, + page_number=command.page_number, + maximum_page_number=len(self.lmp_features) // 8 - 1, + extended_lmp_features=bytes(8), ) - # Features of the current page - + self.lmp_features[command.page_number * 8 : (command.page_number + 1) * 8] + return hci.HCI_Read_Local_Extended_Features_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + page_number=command.page_number, + maximum_page_number=len(self.lmp_features) // 8 - 1, + extended_lmp_features=self.lmp_features[ + command.page_number * 8 : (command.page_number + 1) * 8 + ], ) def on_hci_read_buffer_size_command( self, _command: hci.HCI_Read_Buffer_Size_Command - ) -> bytes | None: + ) -> hci.HCI_Read_Buffer_Size_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.4.5 Read Buffer Size Command ''' - return struct.pack( - '<BHBHH', - hci.HCI_SUCCESS, - self.acl_data_packet_length, - 0, - self.total_num_acl_data_packets, - 0, + return hci.HCI_Read_Buffer_Size_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + hc_acl_data_packet_length=self.acl_data_packet_length, + hc_synchronous_data_packet_length=0, + hc_total_num_acl_data_packets=self.total_num_acl_data_packets, + hc_total_num_synchronous_data_packets=0, ) def on_hci_read_bd_addr_command( self, _command: hci.HCI_Read_BD_ADDR_Command - ) -> bytes | None: + ) -> hci.HCI_Read_BD_ADDR_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.4.6 Read BD_ADDR Command ''' - bd_addr = ( - bytes(self._public_address) - if self._public_address is not None - else bytes(6) + return hci.HCI_Read_BD_ADDR_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + bd_addr=self._public_address or hci.Address.ANY, ) - return bytes([hci.HCI_SUCCESS]) + bd_addr def on_hci_le_set_default_subrate_command( self, command: hci.HCI_LE_Set_Default_Subrate_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 6, Part E - 7.8.123 LE Set Event Mask Command ''' @@ -1812,13 +1788,15 @@ or command.subrate_max < command.subrate_min or command.continuation_number >= command.subrate_max ): - return bytes([hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR]) + return hci.HCI_StatusReturnParameters( + hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR + ) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_subrate_request_command( self, command: hci.HCI_LE_Subrate_Request_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 6, Part E - 7.8.124 LE Subrate Request command ''' @@ -1828,19 +1806,15 @@ or command.subrate_max < command.subrate_min or command.continuation_number >= command.subrate_max ): - return bytes([hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR]) - - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_SUCCESS, - num_hci_command_packets=1, - command_opcode=command.op_code, + return self._send_hci_command_status( + hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR, command.op_code ) - ) + + self._send_hci_command_status(hci.HCI_ErrorCode.SUCCESS, command.op_code) self.send_hci_packet( hci.HCI_LE_Subrate_Change_Event( - status=hci.HCI_SUCCESS, + status=hci.HCI_ErrorCode.SUCCESS, connection_handle=command.connection_handle, subrate_factor=2, peripheral_latency=2, @@ -1852,77 +1826,78 @@ def on_hci_le_set_event_mask_command( self, command: hci.HCI_LE_Set_Event_Mask_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.1 LE Set Event Mask Command ''' self.le_event_mask = int.from_bytes( command.le_event_mask, byteorder='little', signed=False ) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_read_buffer_size_command( self, _command: hci.HCI_LE_Read_Buffer_Size_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Buffer_Size_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.2 LE Read Buffer Size Command ''' - return struct.pack( - '<BHB', - hci.HCI_SUCCESS, - self.le_acl_data_packet_length, - self.total_num_le_acl_data_packets, + return hci.HCI_LE_Read_Buffer_Size_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + le_acl_data_packet_length=self.le_acl_data_packet_length, + total_num_le_acl_data_packets=self.total_num_le_acl_data_packets, ) def on_hci_le_read_buffer_size_v2_command( self, _command: hci.HCI_LE_Read_Buffer_Size_V2_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Buffer_Size_V2_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.2 LE Read Buffer Size Command ''' - return struct.pack( - '<BHBHB', - hci.HCI_SUCCESS, - self.le_acl_data_packet_length, - self.total_num_le_acl_data_packets, - self.iso_data_packet_length, - self.total_num_iso_data_packets, + return hci.HCI_LE_Read_Buffer_Size_V2_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + le_acl_data_packet_length=self.le_acl_data_packet_length, + total_num_le_acl_data_packets=self.total_num_le_acl_data_packets, + iso_data_packet_length=self.iso_data_packet_length, + total_num_iso_data_packets=self.total_num_iso_data_packets, ) def on_hci_le_read_local_supported_features_command( self, _command: hci.HCI_LE_Read_Local_Supported_Features_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Local_Supported_Features_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.3 LE Read Local Supported Features Command ''' - return bytes([hci.HCI_SUCCESS]) + self.le_features.value.to_bytes(8, 'little') + return hci.HCI_LE_Read_Local_Supported_Features_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + le_features=self.le_features.value.to_bytes(8, 'little'), + ) def on_hci_le_read_all_local_supported_features_command( self, _command: hci.HCI_LE_Read_All_Local_Supported_Features_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_All_Local_Supported_Features_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.128 LE Read All Local Supported Features Command ''' - return ( - bytes([hci.HCI_SUCCESS]) - + bytes([0]) - + self.le_features.value.to_bytes(248, 'little') + return hci.HCI_LE_Read_All_Local_Supported_Features_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + max_page=0, + le_features=self.le_features.value.to_bytes(248, 'little'), ) def on_hci_le_set_random_address_command( self, command: hci.HCI_LE_Set_Random_Address_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.4 LE Set Random hci.Address Command ''' self.random_address = command.random_address - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_advertising_parameters_command( self, command: hci.HCI_LE_Set_Advertising_Parameters_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.5 LE Set Advertising Parameters Command ''' @@ -1942,39 +1917,41 @@ self.le_legacy_advertiser.advertising_filter_policy = ( command.advertising_filter_policy ) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_read_advertising_physical_channel_tx_power_command( self, _command: hci.HCI_LE_Read_Advertising_Physical_Channel_Tx_Power_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Advertising_Physical_Channel_Tx_Power_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.6 LE Read Advertising Physical Channel Tx Power Command ''' - return bytes([hci.HCI_SUCCESS, self.advertising_channel_tx_power]) + return hci.HCI_LE_Read_Advertising_Physical_Channel_Tx_Power_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, tx_power_level=0 + ) def on_hci_le_set_advertising_data_command( self, command: hci.HCI_LE_Set_Advertising_Data_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.7 LE Set Advertising Data Command ''' self.le_legacy_advertiser.advertising_data = command.advertising_data - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_scan_response_data_command( self, command: hci.HCI_LE_Set_Scan_Response_Data_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.8 LE Set Scan Response Data Command ''' self.le_legacy_advertiser.scan_response_data = command.scan_response_data - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_advertising_enable_command( self, command: hci.HCI_LE_Set_Advertising_Enable_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.9 LE Set Advertising Enable Command ''' @@ -1983,37 +1960,39 @@ else: self.le_legacy_advertiser.stop() - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_scan_parameters_command( self, command: hci.HCI_LE_Set_Scan_Parameters_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.10 LE Set Scan Parameters Command ''' if self.le_scan_enable: - return bytes([hci.HCI_COMMAND_DISALLOWED_ERROR]) + return hci.HCI_StatusReturnParameters( + hci.HCI_ErrorCode.COMMAND_DISALLOWED_ERROR + ) self.le_scan_type = command.le_scan_type self.le_scan_interval = command.le_scan_interval self.le_scan_window = command.le_scan_window self.le_scan_own_address_type = hci.AddressType(command.own_address_type) self.le_scanning_filter_policy = command.scanning_filter_policy - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_scan_enable_command( self, command: hci.HCI_LE_Set_Scan_Enable_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.11 LE Set Scan Enable Command ''' self.le_scan_enable = bool(command.le_scan_enable) self.filter_duplicates = bool(command.filter_duplicates) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_create_connection_command( self, command: hci.HCI_LE_Create_Connection_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.8.12 LE Create Connection Command ''' @@ -2025,162 +2004,140 @@ # Check that we don't already have a pending connection if self.pending_le_connection: - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_DISALLOWED_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.COMMAND_DISALLOWED_ERROR, command.op_code ) return None self.pending_le_connection = command # Say that the connection is pending - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_STATUS_PENDING, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code) return None def on_hci_le_create_connection_cancel_command( self, _command: hci.HCI_LE_Create_Connection_Cancel_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.13 LE Create Connection Cancel Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_extended_create_connection_command( self, command: hci.HCI_LE_Extended_Create_Connection_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.8.66 LE Extended Create Connection Command ''' if not self.link: - return None + return # Check pending if self.pending_le_connection: - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_DISALLOWED_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + self._send_hci_command_status( + hci.HCI_ErrorCode.COMMAND_DISALLOWED_ERROR, command.op_code ) - return None + return self.pending_le_connection = command - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_STATUS_PENDING, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) - return None + self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code) def on_hci_le_read_filter_accept_list_size_command( self, _command: hci.HCI_LE_Read_Filter_Accept_List_Size_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Filter_Accept_List_Size_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.14 LE Read Filter Accept List Size Command ''' - return bytes([hci.HCI_SUCCESS, self.filter_accept_list_size]) + return hci.HCI_LE_Read_Filter_Accept_List_Size_ReturnParameters( + hci.HCI_ErrorCode.SUCCESS, self.filter_accept_list_size + ) def on_hci_le_clear_filter_accept_list_command( self, _command: hci.HCI_LE_Clear_Filter_Accept_List_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.15 LE Clear Filter Accept List Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_add_device_to_filter_accept_list_command( self, _command: hci.HCI_LE_Add_Device_To_Filter_Accept_List_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.16 LE Add Device To Filter Accept List Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_remove_device_from_filter_accept_list_command( self, _command: hci.HCI_LE_Remove_Device_From_Filter_Accept_List_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.17 LE Remove Device From Filter Accept List Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_write_scan_enable_command( self, command: hci.HCI_Write_Scan_Enable_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.3.18 Write Scan Enable Command ''' self.classic_scan_enable = command.scan_enable - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_read_remote_features_command( self, command: hci.HCI_LE_Read_Remote_Features_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.8.21 LE Read Remote Features Command ''' handle = command.connection_handle - if not self.find_connection_by_handle(handle): - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) + if not (connection := self.find_le_connection_by_handle(handle)): + self._send_hci_command_status( + hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR, command.op_code ) return None # First, say that the command is pending - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_STATUS_PENDING, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code) - # Then send the remote features - self.send_hci_packet( - hci.HCI_LE_Read_Remote_Features_Complete_Event( - status=hci.HCI_SUCCESS, - connection_handle=handle, - le_features=bytes.fromhex('dd40000000000000'), + if connection.role == hci.Role.CENTRAL: + connection.send_ll_control_pdu( + ll.FeatureReq(feature_set=self.le_features.value.to_bytes(8, 'little')) ) - ) + else: + connection.send_ll_control_pdu( + ll.PeripheralFeatureReq( + feature_set=self.le_features.value.to_bytes(8, 'little') + ) + ) return None - def on_hci_le_rand_command(self, _command: hci.HCI_LE_Rand_Command) -> bytes | None: + def on_hci_le_rand_command( + self, _command: hci.HCI_LE_Rand_Command + ) -> hci.HCI_LE_Rand_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.23 LE Rand Command ''' - return bytes([hci.HCI_SUCCESS]) + struct.pack('Q', random.randint(0, 1 << 64)) + return hci.HCI_LE_Rand_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + random_number=struct.pack('Q', random.randint(0, 1 << 64)), + ) def on_hci_le_enable_encryption_command( self, command: hci.HCI_LE_Enable_Encryption_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.8.24 LE Enable Encryption Command ''' if not self.link: - return None + return # Check the parameters if ( @@ -2190,7 +2147,9 @@ or connection.transport != PhysicalTransport.LE ): logger.warning('connection not found') - return bytes([hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR]) + return self._send_hci_command_status( + hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR, command.op_code + ) connection.send_ll_control_pdu( ll.EncReq( @@ -2200,44 +2159,37 @@ ), ) - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_STATUS_PENDING, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code) # TODO: Handle authentication self.on_le_encrypted(connection) - return None - def on_hci_le_read_supported_states_command( self, _command: hci.HCI_LE_Read_Supported_States_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Supported_States_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.27 LE Read Supported States Command ''' - return bytes([hci.HCI_SUCCESS]) + self.le_states + return hci.HCI_LE_Read_Supported_States_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, le_states=self.le_states + ) def on_hci_le_read_suggested_default_data_length_command( self, _command: hci.HCI_LE_Read_Suggested_Default_Data_Length_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Suggested_Default_Data_Length_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.34 LE Read Suggested Default Data Length Command ''' - return struct.pack( - '<BHH', - hci.HCI_SUCCESS, - self.suggested_max_tx_octets, - self.suggested_max_tx_time, + return hci.HCI_LE_Read_Suggested_Default_Data_Length_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + suggested_max_tx_octets=self.suggested_max_tx_octets, + suggested_max_tx_time=self.suggested_max_tx_time, ) def on_hci_le_write_suggested_default_data_length_command( self, command: hci.HCI_LE_Write_Suggested_Default_Data_Length_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.35 LE Write Suggested Default Data Length Command @@ -2245,111 +2197,112 @@ self.suggested_max_tx_octets, self.suggested_max_tx_time = struct.unpack( '<HH', command.parameters[:4] ) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_read_local_p_256_public_key_command( self, _command: hci.HCI_LE_Read_Local_P_256_Public_Key_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.36 LE Read P-256 Public Key Command ''' # TODO create key and send hci.HCI_LE_Read_Local_P-256_Public_Key_Complete event - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_add_device_to_resolving_list_command( self, _command: hci.HCI_LE_Add_Device_To_Resolving_List_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.38 LE Add Device To Resolving List Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_clear_resolving_list_command( self, _command: hci.HCI_LE_Clear_Resolving_List_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.40 LE Clear Resolving List Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_read_resolving_list_size_command( self, _command: hci.HCI_LE_Read_Resolving_List_Size_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Resolving_List_Size_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.41 LE Read Resolving List Size Command ''' - return bytes([hci.HCI_SUCCESS, self.resolving_list_size]) + return hci.HCI_LE_Read_Resolving_List_Size_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + resolving_list_size=self.resolving_list_size, + ) def on_hci_le_set_address_resolution_enable_command( self, command: hci.HCI_LE_Set_Address_Resolution_Enable_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.44 LE Set hci.Address Resolution Enable Command ''' - ret = hci.HCI_SUCCESS + ret = hci.HCI_ErrorCode.SUCCESS if command.address_resolution_enable == 1: self.le_address_resolution = True elif command.address_resolution_enable == 0: self.le_address_resolution = False else: - ret = hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR - return bytes([ret]) + ret = hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR + return hci.HCI_StatusReturnParameters(ret) def on_hci_le_set_resolvable_private_address_timeout_command( self, command: hci.HCI_LE_Set_Resolvable_Private_Address_Timeout_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.45 LE Set Resolvable Private hci.Address Timeout Command ''' self.le_rpa_timeout = command.rpa_timeout - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_read_maximum_data_length_command( self, _command: hci.HCI_LE_Read_Maximum_Data_Length_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Maximum_Data_Length_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.46 LE Read Maximum Data Length Command ''' - return struct.pack( - '<BHHHH', - hci.HCI_SUCCESS, - self.supported_max_tx_octets, - self.supported_max_tx_time, - self.supported_max_rx_octets, - self.supported_max_rx_time, + return hci.HCI_LE_Read_Maximum_Data_Length_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + supported_max_tx_octets=self.supported_max_tx_octets, + supported_max_tx_time=self.supported_max_tx_time, + supported_max_rx_octets=self.supported_max_rx_octets, + supported_max_rx_time=self.supported_max_rx_time, ) def on_hci_le_read_phy_command( self, command: hci.HCI_LE_Read_PHY_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_PHY_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.47 LE Read PHY Command ''' - return struct.pack( - '<BHBB', - hci.HCI_SUCCESS, - command.connection_handle, - hci.HCI_LE_1M_PHY, - hci.HCI_LE_1M_PHY, + return hci.HCI_LE_Read_PHY_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + connection_handle=command.connection_handle, + tx_phy=hci.HCI_LE_1M_PHY, + rx_phy=hci.HCI_LE_1M_PHY, ) def on_hci_le_set_default_phy_command( self, command: hci.HCI_LE_Set_Default_PHY_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.48 LE Set Default PHY Command ''' self.default_phy['all_phys'] = command.all_phys self.default_phy['tx_phys'] = command.tx_phys self.default_phy['rx_phys'] = command.rx_phys - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_advertising_set_random_address_command( self, command: hci.HCI_LE_Set_Advertising_Set_Random_Address_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.52 LE Set Advertising Set Random hci.Address Command @@ -2360,11 +2313,11 @@ controller=self, handle=handle ) self.advertising_sets[handle].random_address = command.random_address - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_extended_advertising_parameters_command( self, command: hci.HCI_LE_Set_Extended_Advertising_Parameters_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Set_Extended_Advertising_Parameters_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.53 LE Set Extended Advertising Parameters Command @@ -2376,18 +2329,22 @@ ) self.advertising_sets[handle].parameters = command - return bytes([hci.HCI_SUCCESS, 0]) + return hci.HCI_LE_Set_Extended_Advertising_Parameters_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, selected_tx_power=0 + ) def on_hci_le_set_extended_advertising_data_command( self, command: hci.HCI_LE_Set_Extended_Advertising_Data_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.54 LE Set Extended Advertising Data Command ''' handle = command.advertising_handle if not (adv_set := self.advertising_sets.get(handle)): - return bytes([hci.HCI_UNKNOWN_ADVERTISING_IDENTIFIER_ERROR]) + return hci.HCI_StatusReturnParameters( + hci.HCI_ErrorCode.UNKNOWN_ADVERTISING_IDENTIFIER_ERROR + ) if command.operation in ( hci.HCI_LE_Set_Extended_Advertising_Data_Command.Operation.FIRST_FRAGMENT, @@ -2400,18 +2357,20 @@ ): adv_set.data.extend(command.advertising_data) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_extended_scan_response_data_command( self, command: hci.HCI_LE_Set_Extended_Scan_Response_Data_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.55 LE Set Extended Scan Response Data Command ''' handle = command.advertising_handle if not (adv_set := self.advertising_sets.get(handle)): - return bytes([hci.HCI_UNKNOWN_ADVERTISING_IDENTIFIER_ERROR]) + return hci.HCI_StatusReturnParameters( + hci.HCI_ErrorCode.UNKNOWN_ADVERTISING_IDENTIFIER_ERROR + ) if command.operation in ( hci.HCI_LE_Set_Extended_Advertising_Data_Command.Operation.FIRST_FRAGMENT, @@ -2424,11 +2383,11 @@ ): adv_set.scan_response_data.extend(command.scan_response_data) - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_extended_advertising_enable_command( self, command: hci.HCI_LE_Set_Extended_Advertising_Enable_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.56 LE Set Extended Advertising Enable Command @@ -2445,86 +2404,92 @@ for handle in command.advertising_handles: if advertising_set := self.advertising_sets.get(handle): advertising_set.stop() - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_remove_advertising_set_command( self, command: hci.HCI_LE_Remove_Advertising_Set_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.59 LE Remove Advertising Set Command ''' handle = command.advertising_handle if advertising_set := self.advertising_sets.pop(handle, None): advertising_set.stop() - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_clear_advertising_sets_command( self, _command: hci.HCI_LE_Clear_Advertising_Sets_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.60 LE Clear Advertising Sets Command ''' for advertising_set in self.advertising_sets.values(): advertising_set.stop() self.advertising_sets.clear() - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_read_maximum_advertising_data_length_command( self, _command: hci.HCI_LE_Read_Maximum_Advertising_Data_Length_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Maximum_Advertising_Data_Length_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.57 LE Read Maximum Advertising Data Length Command ''' - return struct.pack('<BH', hci.HCI_SUCCESS, 0x0672) + return hci.HCI_LE_Read_Maximum_Advertising_Data_Length_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, max_advertising_data_length=0x0672 + ) def on_hci_le_read_number_of_supported_advertising_sets_command( self, _command: hci.HCI_LE_Read_Number_Of_Supported_Advertising_Sets_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Number_Of_Supported_Advertising_Sets_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.58 LE Read Number of Supported Advertising Set Command ''' - return struct.pack('<BB', hci.HCI_SUCCESS, 0xF0) + return hci.HCI_LE_Read_Number_Of_Supported_Advertising_Sets_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, num_supported_advertising_sets=0xF0 + ) def on_hci_le_set_periodic_advertising_parameters_command( self, _command: hci.HCI_LE_Set_Periodic_Advertising_Parameters_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.61 LE Set Periodic Advertising Parameters Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_periodic_advertising_data_command( self, _command: hci.HCI_LE_Set_Periodic_Advertising_Data_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.62 LE Set Periodic Advertising Data Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_set_periodic_advertising_enable_command( self, _command: hci.HCI_LE_Set_Periodic_Advertising_Enable_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.63 LE Set Periodic Advertising Enable Command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS) def on_hci_le_read_transmit_power_command( self, _command: hci.HCI_LE_Read_Transmit_Power_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Read_Transmit_Power_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.74 LE Read Transmit Power Command ''' - return struct.pack('<BBB', hci.HCI_SUCCESS, 0, 0) + return hci.HCI_LE_Read_Transmit_Power_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, min_tx_power=0, max_tx_power=0 + ) def on_hci_le_set_cig_parameters_command( self, command: hci.HCI_LE_Set_CIG_Parameters_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Set_CIG_Parameters_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.97 LE Set CIG Parameter Command ''' @@ -2544,13 +2509,15 @@ cig_id=command.cig_id, handle=handle, ) - return struct.pack( - '<BBB', hci.HCI_SUCCESS, command.cig_id, len(handles) - ) + b''.join([struct.pack('<H', handle) for handle in handles]) + return hci.HCI_LE_Set_CIG_Parameters_ReturnParameters( + status=hci.HCI_ErrorCode.SUCCESS, + cig_id=command.cig_id, + connection_handle=handles, + ) def on_hci_le_create_cis_command( self, command: hci.HCI_LE_Create_CIS_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.8.99 LE Create CIS Command ''' @@ -2562,11 +2529,17 @@ ): if not (connection := self.find_connection_by_handle(acl_handle)): logger.error(f'Cannot find connection with handle={acl_handle}') - return bytes([hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR]) + self._send_hci_command_status( + hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR, command.op_code + ) + return if not (cis_link := self.central_cis_links.get(cis_handle)): logger.error(f'Cannot find CIS with handle={cis_handle}') - return bytes([hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR]) + self._send_hci_command_status( + hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR, command.op_code + ) + return cis_link.acl_connection = connection @@ -2574,35 +2547,28 @@ ll.CisReq(cig_id=cis_link.cig_id, cis_id=cis_link.cis_id) ) - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_STATUS_PENDING, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) - return None + self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code) def on_hci_le_remove_cig_command( self, command: hci.HCI_LE_Remove_CIG_Command - ) -> bytes | None: + ) -> hci.HCI_LE_Remove_CIG_ReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.100 LE Remove CIG Command ''' - status = hci.HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR + status = hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR cis_links = list(self.central_cis_links.items()) for cis_handle, cis_link in cis_links: if cis_link.cig_id == command.cig_id: self.central_cis_links.pop(cis_handle) - status = hci.HCI_SUCCESS + status = hci.HCI_ErrorCode.SUCCESS - return struct.pack('<BH', status, command.cig_id) + return hci.HCI_LE_Remove_CIG_ReturnParameters(status, command.cig_id) def on_hci_le_accept_cis_request_command( self, command: hci.HCI_LE_Accept_CIS_Request_Command - ) -> bytes | None: + ) -> None: ''' See Bluetooth spec Vol 4, Part E - 7.8.101 LE Accept CIS Request Command ''' @@ -2613,53 +2579,48 @@ pending_cis_link := self.peripheral_cis_links.get(command.connection_handle) ): logger.error(f'Cannot find CIS with handle={command.connection_handle}') - return bytes([hci.HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR]) + self._send_hci_command_status( + hci.HCI_ErrorCode.INVALID_COMMAND_PARAMETERS_ERROR, command.op_code + ) + return assert pending_cis_link.acl_connection pending_cis_link.acl_connection.send_ll_control_pdu( ll.CisRsp(cig_id=pending_cis_link.cig_id, cis_id=pending_cis_link.cis_id), ) - self.send_hci_packet( - hci.HCI_Command_Status_Event( - status=hci.HCI_COMMAND_STATUS_PENDING, - num_hci_command_packets=1, - command_opcode=command.op_code, - ) - ) + self._send_hci_command_status(hci.HCI_COMMAND_STATUS_PENDING, command.op_code) return None def on_hci_le_setup_iso_data_path_command( self, command: hci.HCI_LE_Setup_ISO_Data_Path_Command - ) -> bytes | None: + ) -> hci.HCI_StatusAndConnectionHandleReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.109 LE Setup ISO Data Path Command ''' if not (iso_link := self.find_iso_link_by_handle(command.connection_handle)): - return struct.pack( - '<BH', - hci.HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR, + return hci.HCI_StatusAndConnectionHandleReturnParameters( + hci.HCI_ErrorCode.UNKNOWN_CONNECTION_IDENTIFIER_ERROR, command.connection_handle, ) if command.data_path_direction in iso_link.data_paths: - return struct.pack( - '<BH', - hci.HCI_COMMAND_DISALLOWED_ERROR, - command.connection_handle, + return hci.HCI_StatusAndConnectionHandleReturnParameters( + hci.HCI_ErrorCode.COMMAND_DISALLOWED_ERROR, command.connection_handle ) iso_link.data_paths.add(command.data_path_direction) - return struct.pack('<BH', hci.HCI_SUCCESS, command.connection_handle) + return hci.HCI_StatusAndConnectionHandleReturnParameters( + hci.HCI_ErrorCode.SUCCESS, command.connection_handle + ) def on_hci_le_remove_iso_data_path_command( self, command: hci.HCI_LE_Remove_ISO_Data_Path_Command - ) -> bytes | None: + ) -> hci.HCI_StatusAndConnectionHandleReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.110 LE Remove ISO Data Path Command ''' if not (iso_link := self.find_iso_link_by_handle(command.connection_handle)): - return struct.pack( - '<BH', - hci.HCI_UNKNOWN_CONNECTION_IDENTIFIER_ERROR, + return hci.HCI_StatusAndConnectionHandleReturnParameters( + hci.HCI_ErrorCode.UNKNOWN_CONNECTION_IDENTIFIER_ERROR, command.connection_handle, ) data_paths: set[int] = set( @@ -2668,18 +2629,18 @@ if (1 << direction) & command.data_path_direction ) if not data_paths.issubset(iso_link.data_paths): - return struct.pack( - '<BH', - hci.HCI_COMMAND_DISALLOWED_ERROR, - command.connection_handle, + return hci.HCI_StatusAndConnectionHandleReturnParameters( + hci.HCI_ErrorCode.COMMAND_DISALLOWED_ERROR, command.connection_handle ) iso_link.data_paths.difference_update(data_paths) - return struct.pack('<BH', hci.HCI_SUCCESS, command.connection_handle) + return hci.HCI_StatusAndConnectionHandleReturnParameters( + hci.HCI_ErrorCode.SUCCESS, command.connection_handle + ) def on_hci_le_set_host_feature_command( self, _command: hci.HCI_LE_Set_Host_Feature_Command - ) -> bytes | None: + ) -> hci.HCI_StatusReturnParameters: ''' See Bluetooth spec Vol 4, Part E - 7.8.115 LE Set Host Feature command ''' - return bytes([hci.HCI_SUCCESS]) + return hci.HCI_StatusReturnParameters(hci.HCI_ErrorCode.SUCCESS)
diff --git a/bumble/core.py b/bumble/core.py index 8f68a2b..3be09de 100644 --- a/bumble/core.py +++ b/bumble/core.py
@@ -280,14 +280,15 @@ if not force_128: return self.uuid_bytes - if len(self.uuid_bytes) == 2: - return self.BASE_UUID + self.uuid_bytes + bytes([0, 0]) - elif len(self.uuid_bytes) == 4: - return self.BASE_UUID + self.uuid_bytes - elif len(self.uuid_bytes) == 16: - return self.uuid_bytes - else: - assert False, "unreachable" + match len(self.uuid_bytes): + case 2: + return self.BASE_UUID + self.uuid_bytes + bytes([0, 0]) + case 4: + return self.BASE_UUID + self.uuid_bytes + case 16: + return self.uuid_bytes + case _: + assert False, "unreachable" def to_pdu_bytes(self) -> bytes: ''' @@ -1769,66 +1770,71 @@ @classmethod def ad_data_to_string(cls, ad_type: int, ad_data: bytes) -> str: - if ad_type == AdvertisingData.FLAGS: - ad_type_str = 'Flags' - ad_data_str = AdvertisingData.flags_to_string(ad_data[0], short=True) - elif ad_type == AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS: - ad_type_str = 'Complete List of 16-bit Service Class UUIDs' - ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 2) - elif ad_type == AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS: - ad_type_str = 'Incomplete List of 16-bit Service Class UUIDs' - ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 2) - elif ad_type == AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS: - ad_type_str = 'Complete List of 32-bit Service Class UUIDs' - ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 4) - elif ad_type == AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS: - ad_type_str = 'Incomplete List of 32-bit Service Class UUIDs' - ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 4) - elif ad_type == AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS: - ad_type_str = 'Complete List of 128-bit Service Class UUIDs' - ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 16) - elif ad_type == AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS: - ad_type_str = 'Incomplete List of 128-bit Service Class UUIDs' - ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 16) - elif ad_type == AdvertisingData.SERVICE_DATA_16_BIT_UUID: - ad_type_str = 'Service Data' - uuid = UUID.from_bytes(ad_data[:2]) - ad_data_str = f'service={uuid}, data={ad_data[2:].hex()}' - elif ad_type == AdvertisingData.SERVICE_DATA_32_BIT_UUID: - ad_type_str = 'Service Data' - uuid = UUID.from_bytes(ad_data[:4]) - ad_data_str = f'service={uuid}, data={ad_data[4:].hex()}' - elif ad_type == AdvertisingData.SERVICE_DATA_128_BIT_UUID: - ad_type_str = 'Service Data' - uuid = UUID.from_bytes(ad_data[:16]) - ad_data_str = f'service={uuid}, data={ad_data[16:].hex()}' - elif ad_type == AdvertisingData.SHORTENED_LOCAL_NAME: - ad_type_str = 'Shortened Local Name' - ad_data_str = f'"{ad_data.decode("utf-8")}"' - elif ad_type == AdvertisingData.COMPLETE_LOCAL_NAME: - ad_type_str = 'Complete Local Name' - try: + match ad_type: + case AdvertisingData.FLAGS: + ad_type_str = 'Flags' + ad_data_str = AdvertisingData.flags_to_string(ad_data[0], short=True) + case AdvertisingData.COMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS: + ad_type_str = 'Complete List of 16-bit Service Class UUIDs' + ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 2) + case AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS: + ad_type_str = 'Incomplete List of 16-bit Service Class UUIDs' + ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 2) + case AdvertisingData.COMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS: + ad_type_str = 'Complete List of 32-bit Service Class UUIDs' + ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 4) + case AdvertisingData.INCOMPLETE_LIST_OF_32_BIT_SERVICE_CLASS_UUIDS: + ad_type_str = 'Incomplete List of 32-bit Service Class UUIDs' + ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 4) + case AdvertisingData.COMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS: + ad_type_str = 'Complete List of 128-bit Service Class UUIDs' + ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 16) + case AdvertisingData.INCOMPLETE_LIST_OF_128_BIT_SERVICE_CLASS_UUIDS: + ad_type_str = 'Incomplete List of 128-bit Service Class UUIDs' + ad_data_str = AdvertisingData.uuid_list_to_string(ad_data, 16) + case AdvertisingData.SERVICE_DATA_16_BIT_UUID: + ad_type_str = 'Service Data' + uuid = UUID.from_bytes(ad_data[:2]) + ad_data_str = f'service={uuid}, data={ad_data[2:].hex()}' + case AdvertisingData.SERVICE_DATA_32_BIT_UUID: + ad_type_str = 'Service Data' + uuid = UUID.from_bytes(ad_data[:4]) + ad_data_str = f'service={uuid}, data={ad_data[4:].hex()}' + case AdvertisingData.SERVICE_DATA_128_BIT_UUID: + ad_type_str = 'Service Data' + uuid = UUID.from_bytes(ad_data[:16]) + ad_data_str = f'service={uuid}, data={ad_data[16:].hex()}' + case AdvertisingData.SHORTENED_LOCAL_NAME: + ad_type_str = 'Shortened Local Name' ad_data_str = f'"{ad_data.decode("utf-8")}"' - except UnicodeDecodeError: + case AdvertisingData.COMPLETE_LOCAL_NAME: + ad_type_str = 'Complete Local Name' + try: + ad_data_str = f'"{ad_data.decode("utf-8")}"' + except UnicodeDecodeError: + ad_data_str = ad_data.hex() + case AdvertisingData.TX_POWER_LEVEL: + ad_type_str = 'TX Power Level' + ad_data_str = str(ad_data[0]) + case AdvertisingData.MANUFACTURER_SPECIFIC_DATA: + ad_type_str = 'Manufacturer Specific Data' + company_id = struct.unpack_from('<H', ad_data, 0)[0] + company_name = COMPANY_IDENTIFIERS.get( + company_id, f'0x{company_id:04X}' + ) + ad_data_str = f'company={company_name}, data={ad_data[2:].hex()}' + case AdvertisingData.APPEARANCE: + ad_type_str = 'Appearance' + appearance = Appearance.from_int( + struct.unpack_from('<H', ad_data, 0)[0] + ) + ad_data_str = str(appearance) + case AdvertisingData.BROADCAST_NAME: + ad_type_str = 'Broadcast Name' + ad_data_str = ad_data.decode('utf-8') + case _: + ad_type_str = AdvertisingData.Type(ad_type).name ad_data_str = ad_data.hex() - elif ad_type == AdvertisingData.TX_POWER_LEVEL: - ad_type_str = 'TX Power Level' - ad_data_str = str(ad_data[0]) - elif ad_type == AdvertisingData.MANUFACTURER_SPECIFIC_DATA: - ad_type_str = 'Manufacturer Specific Data' - company_id = struct.unpack_from('<H', ad_data, 0)[0] - company_name = COMPANY_IDENTIFIERS.get(company_id, f'0x{company_id:04X}') - ad_data_str = f'company={company_name}, data={ad_data[2:].hex()}' - elif ad_type == AdvertisingData.APPEARANCE: - ad_type_str = 'Appearance' - appearance = Appearance.from_int(struct.unpack_from('<H', ad_data, 0)[0]) - ad_data_str = str(appearance) - elif ad_type == AdvertisingData.BROADCAST_NAME: - ad_type_str = 'Broadcast Name' - ad_data_str = ad_data.decode('utf-8') - else: - ad_type_str = AdvertisingData.Type(ad_type).name - ad_data_str = ad_data.hex() return f'[{ad_type_str}]: {ad_data_str}'
diff --git a/bumble/drivers/intel.py b/bumble/drivers/intel.py index ccddd4f..5dbec38 100644 --- a/bumble/drivers/intel.py +++ b/bumble/drivers/intel.py
@@ -201,50 +201,51 @@ value = data[2 : 2 + value_length] typed_value: Any - if value_type == ValueType.END: - break + match value_type: + case ValueType.END: + break - if value_type in (ValueType.CNVI, ValueType.CNVR): - (v,) = struct.unpack("<I", value) - typed_value = ( - (((v >> 0) & 0xF) << 12) - | (((v >> 4) & 0xF) << 0) - | (((v >> 8) & 0xF) << 4) - | (((v >> 24) & 0xF) << 8) - ) - elif value_type == ValueType.HARDWARE_INFO: - (v,) = struct.unpack("<I", value) - typed_value = HardwareInfo( - HardwarePlatform((v >> 8) & 0xFF), HardwareVariant((v >> 16) & 0x3F) - ) - elif value_type in ( - ValueType.USB_VENDOR_ID, - ValueType.USB_PRODUCT_ID, - ValueType.DEVICE_REVISION, - ): - (typed_value,) = struct.unpack("<H", value) - elif value_type == ValueType.CURRENT_MODE_OF_OPERATION: - typed_value = ModeOfOperation(value[0]) - elif value_type in ( - ValueType.BUILD_TYPE, - ValueType.BUILD_NUMBER, - ValueType.SECURE_BOOT, - ValueType.OTP_LOCK, - ValueType.API_LOCK, - ValueType.DEBUG_LOCK, - ValueType.SECURE_BOOT_ENGINE_TYPE, - ): - typed_value = value[0] - elif value_type == ValueType.TIMESTAMP: - typed_value = Timestamp(value[0], value[1]) - elif value_type == ValueType.FIRMWARE_BUILD: - typed_value = FirmwareBuild(value[0], Timestamp(value[1], value[2])) - elif value_type == ValueType.BLUETOOTH_ADDRESS: - typed_value = hci.Address( - value, address_type=hci.Address.PUBLIC_DEVICE_ADDRESS - ) - else: - typed_value = value + case ValueType.CNVI | ValueType.CNVR: + (v,) = struct.unpack("<I", value) + typed_value = ( + (((v >> 0) & 0xF) << 12) + | (((v >> 4) & 0xF) << 0) + | (((v >> 8) & 0xF) << 4) + | (((v >> 24) & 0xF) << 8) + ) + case ValueType.HARDWARE_INFO: + (v,) = struct.unpack("<I", value) + typed_value = HardwareInfo( + HardwarePlatform((v >> 8) & 0xFF), HardwareVariant((v >> 16) & 0x3F) + ) + case ( + ValueType.USB_VENDOR_ID + | ValueType.USB_PRODUCT_ID + | ValueType.DEVICE_REVISION + ): + (typed_value,) = struct.unpack("<H", value) + case ValueType.CURRENT_MODE_OF_OPERATION: + typed_value = ModeOfOperation(value[0]) + case ( + ValueType.BUILD_TYPE + | ValueType.BUILD_NUMBER + | ValueType.SECURE_BOOT + | ValueType.OTP_LOCK + | ValueType.API_LOCK + | ValueType.DEBUG_LOCK + | ValueType.SECURE_BOOT_ENGINE_TYPE + ): + typed_value = value[0] + case ValueType.TIMESTAMP: + typed_value = Timestamp(value[0], value[1]) + case ValueType.FIRMWARE_BUILD: + typed_value = FirmwareBuild(value[0], Timestamp(value[1], value[2])) + case ValueType.BLUETOOTH_ADDRESS: + typed_value = hci.Address( + value, address_type=hci.Address.PUBLIC_DEVICE_ADDRESS + ) + case _: + typed_value = value result.append((value_type, typed_value)) data = data[2 + value_length :]
diff --git a/bumble/gap.py b/bumble/gap.py deleted file mode 100644 index 26a353c..0000000 --- a/bumble/gap.py +++ /dev/null
@@ -1,60 +0,0 @@ -# Copyright 2021-2022 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# https://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -# ----------------------------------------------------------------------------- -# Imports -# ----------------------------------------------------------------------------- -import logging -import struct - -from bumble.gatt import ( - GATT_APPEARANCE_CHARACTERISTIC, - GATT_DEVICE_NAME_CHARACTERISTIC, - GATT_GENERIC_ACCESS_SERVICE, - Characteristic, - Service, -) - -# ----------------------------------------------------------------------------- -# Logging -# ----------------------------------------------------------------------------- -logger = logging.getLogger(__name__) - - -# ----------------------------------------------------------------------------- -# Classes -# ----------------------------------------------------------------------------- - - -# ----------------------------------------------------------------------------- -class GenericAccessService(Service): - def __init__(self, device_name, appearance=(0, 0)): - device_name_characteristic = Characteristic( - GATT_DEVICE_NAME_CHARACTERISTIC, - Characteristic.Properties.READ, - Characteristic.READABLE, - device_name.encode('utf-8')[:248], - ) - - appearance_characteristic = Characteristic( - GATT_APPEARANCE_CHARACTERISTIC, - Characteristic.Properties.READ, - Characteristic.READABLE, - struct.pack('<H', (appearance[0] << 6) | appearance[1]), - ) - - super().__init__( - GATT_GENERIC_ACCESS_SERVICE, - [device_name_characteristic, appearance_characteristic], - )
diff --git a/bumble/hci.py b/bumble/hci.py index 46c41c7..0484cf2 100644 --- a/bumble/hci.py +++ b/bumble/hci.py
@@ -31,6 +31,7 @@ ClassVar, Generic, Literal, + SupportsBytes, TypeVar, cast, ) @@ -247,28 +248,6 @@ HCI_VERSION_BLUETOOTH_CORE_6_1 = SpecificationVersion.BLUETOOTH_CORE_6_1 HCI_VERSION_BLUETOOTH_CORE_6_2 = SpecificationVersion.BLUETOOTH_CORE_6_2 -HCI_VERSION_NAMES = { - HCI_VERSION_BLUETOOTH_CORE_1_0B: 'HCI_VERSION_BLUETOOTH_CORE_1_0B', - HCI_VERSION_BLUETOOTH_CORE_1_1: 'HCI_VERSION_BLUETOOTH_CORE_1_1', - HCI_VERSION_BLUETOOTH_CORE_1_2: 'HCI_VERSION_BLUETOOTH_CORE_1_2', - HCI_VERSION_BLUETOOTH_CORE_2_0_EDR: 'HCI_VERSION_BLUETOOTH_CORE_2_0_EDR', - HCI_VERSION_BLUETOOTH_CORE_2_1_EDR: 'HCI_VERSION_BLUETOOTH_CORE_2_1_EDR', - HCI_VERSION_BLUETOOTH_CORE_3_0_HS: 'HCI_VERSION_BLUETOOTH_CORE_3_0_HS', - HCI_VERSION_BLUETOOTH_CORE_4_0: 'HCI_VERSION_BLUETOOTH_CORE_4_0', - HCI_VERSION_BLUETOOTH_CORE_4_1: 'HCI_VERSION_BLUETOOTH_CORE_4_1', - HCI_VERSION_BLUETOOTH_CORE_4_2: 'HCI_VERSION_BLUETOOTH_CORE_4_2', - HCI_VERSION_BLUETOOTH_CORE_5_0: 'HCI_VERSION_BLUETOOTH_CORE_5_0', - HCI_VERSION_BLUETOOTH_CORE_5_1: 'HCI_VERSION_BLUETOOTH_CORE_5_1', - HCI_VERSION_BLUETOOTH_CORE_5_2: 'HCI_VERSION_BLUETOOTH_CORE_5_2', - HCI_VERSION_BLUETOOTH_CORE_5_3: 'HCI_VERSION_BLUETOOTH_CORE_5_3', - HCI_VERSION_BLUETOOTH_CORE_5_4: 'HCI_VERSION_BLUETOOTH_CORE_5_4', - HCI_VERSION_BLUETOOTH_CORE_6_0: 'HCI_VERSION_BLUETOOTH_CORE_6_0', - HCI_VERSION_BLUETOOTH_CORE_6_1: 'HCI_VERSION_BLUETOOTH_CORE_6_1', - HCI_VERSION_BLUETOOTH_CORE_6_2: 'HCI_VERSION_BLUETOOTH_CORE_6_2', -} - -LMP_VERSION_NAMES = HCI_VERSION_NAMES - # HCI Packet types HCI_COMMAND_PACKET = 0x01 HCI_ACL_DATA_PACKET = 0x02 @@ -1860,44 +1839,46 @@ field_type = field_type['parser'] # Parse the field - if field_type == '*': - # The rest of the bytes - field_value = data[offset:] - return (field_value, len(field_value)) - if field_type == 'v': - # Variable-length bytes field, with 1-byte length at the beginning - field_length = data[offset] - offset += 1 - field_value = data[offset : offset + field_length] - return (field_value, field_length + 1) - if field_type == 1: - # 8-bit unsigned - return (data[offset], 1) - if field_type == -1: - # 8-bit signed - return (struct.unpack_from('b', data, offset)[0], 1) - if field_type == 2: - # 16-bit unsigned - return (struct.unpack_from('<H', data, offset)[0], 2) - if field_type == '>2': - # 16-bit unsigned big-endian - return (struct.unpack_from('>H', data, offset)[0], 2) - if field_type == -2: - # 16-bit signed - return (struct.unpack_from('<h', data, offset)[0], 2) - if field_type == 3: - # 24-bit unsigned - padded = data[offset : offset + 3] + bytes([0]) - return (struct.unpack('<I', padded)[0], 3) - if field_type == 4: - # 32-bit unsigned - return (struct.unpack_from('<I', data, offset)[0], 4) - if field_type == '>4': - # 32-bit unsigned big-endian - return (struct.unpack_from('>I', data, offset)[0], 4) - if isinstance(field_type, int) and 4 < field_type <= 256: - # Byte array (from 5 up to 256 bytes) - return (data[offset : offset + field_type], field_type) + match field_type: + case '*': + # The rest of the bytes + field_value = data[offset:] + return (field_value, len(field_value)) + case 'v': + # Variable-length bytes field, with 1-byte length at the beginning + field_length = data[offset] + offset += 1 + field_value = data[offset : offset + field_length] + return (field_value, field_length + 1) + case 1: + # 8-bit unsigned + return (data[offset], 1) + case -1: + # 8-bit signed + return (struct.unpack_from('b', data, offset)[0], 1) + case 2: + # 16-bit unsigned + return (struct.unpack_from('<H', data, offset)[0], 2) + case '>2': + # 16-bit unsigned big-endian + return (struct.unpack_from('>H', data, offset)[0], 2) + case -2: + # 16-bit signed + return (struct.unpack_from('<h', data, offset)[0], 2) + case 3: + # 24-bit unsigned + padded = data[offset : offset + 3] + bytes([0]) + return (struct.unpack('<I', padded)[0], 3) + case 4: + # 32-bit unsigned + return (struct.unpack_from('<I', data, offset)[0], 4) + case '>4': + # 32-bit unsigned big-endian + return (struct.unpack_from('>I', data, offset)[0], 4) + case int() if 4 < field_type <= 256: + # Byte array (from 5 up to 256 bytes) + return (data[offset : offset + field_type], field_type) + if callable(field_type): new_offset, field_value = field_type(data, offset) return (field_value, new_offset - offset) @@ -1954,60 +1935,58 @@ # Serialize the field if serializer: - field_bytes = serializer(field_value) - elif field_type == 1: - # 8-bit unsigned - field_bytes = bytes([field_value]) - elif field_type == -1: - # 8-bit signed - field_bytes = struct.pack('b', field_value) - elif field_type == 2: - # 16-bit unsigned - field_bytes = struct.pack('<H', field_value) - elif field_type == '>2': - # 16-bit unsigned big-endian - field_bytes = struct.pack('>H', field_value) - elif field_type == -2: - # 16-bit signed - field_bytes = struct.pack('<h', field_value) - elif field_type == 3: - # 24-bit unsigned - field_bytes = struct.pack('<I', field_value)[0:3] - elif field_type == 4: - # 32-bit unsigned - field_bytes = struct.pack('<I', field_value) - elif field_type == '>4': - # 32-bit unsigned big-endian - field_bytes = struct.pack('>I', field_value) - elif field_type == '*': - if isinstance(field_value, int): - if 0 <= field_value <= 255: - field_bytes = bytes([field_value]) + return serializer(field_value) + match field_type: + case 1: + # 8-bit unsigned + return bytes([field_value]) + case -1: + # 8-bit signed + return struct.pack('b', field_value) + case 2: + # 16-bit unsigned + return struct.pack('<H', field_value) + case '>2': + # 16-bit unsigned big-endian + return struct.pack('>H', field_value) + case -2: + # 16-bit signed + return struct.pack('<h', field_value) + case 3: + # 24-bit unsigned + return struct.pack('<I', field_value)[0:3] + case 4: + # 32-bit unsigned + return struct.pack('<I', field_value) + case '>4': + # 32-bit unsigned big-endian + return struct.pack('>I', field_value) + case '*': + if isinstance(field_value, int): + if 0 <= field_value <= 255: + return bytes([field_value]) + else: + raise InvalidArgumentError('value too large for *-typed field') else: - raise InvalidArgumentError('value too large for *-typed field') - else: + return bytes(field_value) + case 'v': + # Variable-length bytes field, with 1-byte length at the beginning field_bytes = bytes(field_value) - elif field_type == 'v': - # Variable-length bytes field, with 1-byte length at the beginning - field_bytes = bytes(field_value) - field_length = len(field_bytes) - field_bytes = bytes([field_length]) + field_bytes - elif isinstance(field_value, (bytes, bytearray)) or hasattr( - field_value, '__bytes__' - ): + field_length = len(field_bytes) + return bytes([field_length]) + field_bytes + if isinstance(field_value, (bytes, bytearray, SupportsBytes)): field_bytes = bytes(field_value) if isinstance(field_type, int) and 4 < field_type <= 256: # Truncate or pad with zeros if the field is too long or too short if len(field_bytes) < field_type: - field_bytes += bytes(field_type - len(field_bytes)) + return field_bytes + bytes(field_type - len(field_bytes)) elif len(field_bytes) > field_type: - field_bytes = field_bytes[:field_type] - else: - raise InvalidArgumentError( - f"don't know how to serialize type {type(field_value)}" - ) + return field_bytes[:field_type] + return field_bytes - return field_bytes + raise InvalidArgumentError( + f"don't know how to serialize type {type(field_value)}" + ) @staticmethod def dict_to_bytes(hci_object, object_fields): @@ -4736,7 +4715,7 @@ # ----------------------------------------------------------------------------- @dataclasses.dataclass class HCI_LE_Read_Resolving_List_Size_ReturnParameters(HCI_StatusReturnParameters): - resolving_list_size: bytes = field(metadata=metadata(1)) + resolving_list_size: int = field(metadata=metadata(1)) @HCI_SyncCommand.sync_command(HCI_LE_Read_Resolving_List_Size_ReturnParameters)
diff --git a/bumble/hfp.py b/bumble/hfp.py index 3c623a9..7056517 100644 --- a/bumble/hfp.py +++ b/bumble/hfp.py
@@ -26,7 +26,7 @@ import re import traceback from collections.abc import Iterable -from typing import TYPE_CHECKING, Any, ClassVar +from typing import Any, ClassVar, Literal, overload from typing_extensions import Self @@ -420,61 +420,6 @@ # Hands-Free Control Interoperability Requirements # ----------------------------------------------------------------------------- -# Response codes. -RESPONSE_CODES = { - "+APLSIRI", - "+BAC", - "+BCC", - "+BCS", - "+BIA", - "+BIEV", - "+BIND", - "+BINP", - "+BLDN", - "+BRSF", - "+BTRH", - "+BVRA", - "+CCWA", - "+CHLD", - "+CHUP", - "+CIND", - "+CLCC", - "+CLIP", - "+CMEE", - "+CMER", - "+CNUM", - "+COPS", - "+IPHONEACCEV", - "+NREC", - "+VGM", - "+VGS", - "+VTS", - "+XAPL", - "A", - "D", -} - -# Unsolicited responses and statuses. -UNSOLICITED_CODES = { - "+APLSIRI", - "+BCS", - "+BIND", - "+BSIR", - "+BTRH", - "+BVRA", - "+CCWA", - "+CIEV", - "+CLIP", - "+VGM", - "+VGS", - "BLACKLISTED", - "BUSY", - "DELAYED", - "NO ANSWER", - "NO CARRIER", - "RING", -} - # Status codes STATUS_CODES = { "+CME ERROR", @@ -727,12 +672,9 @@ dlc: rfcomm.DLC command_lock: asyncio.Lock - if TYPE_CHECKING: - response_queue: asyncio.Queue[AtResponse] - unsolicited_queue: asyncio.Queue[AtResponse | None] - else: - response_queue: asyncio.Queue - unsolicited_queue: asyncio.Queue + pending_command: str | None = None + response_queue: asyncio.Queue[AtResponse] + unsolicited_queue: asyncio.Queue[AtResponse | None] read_buffer: bytearray active_codec: AudioCodec @@ -805,16 +747,39 @@ self.read_buffer = self.read_buffer[trailer + 2 :] # Forward the received code to the correct queue. - if self.command_lock.locked() and ( - response.code in STATUS_CODES or response.code in RESPONSE_CODES + if self.pending_command and ( + response.code in STATUS_CODES or response.code in self.pending_command ): self.response_queue.put_nowait(response) - elif response.code in UNSOLICITED_CODES: - self.unsolicited_queue.put_nowait(response) else: - logger.warning( - f"dropping unexpected response with code '{response.code}'" - ) + self.unsolicited_queue.put_nowait(response) + + @overload + async def execute_command( + self, + cmd: str, + timeout: float = 1.0, + *, + response_type: Literal[AtResponseType.NONE] = AtResponseType.NONE, + ) -> None: ... + + @overload + async def execute_command( + self, + cmd: str, + timeout: float = 1.0, + *, + response_type: Literal[AtResponseType.SINGLE], + ) -> AtResponse: ... + + @overload + async def execute_command( + self, + cmd: str, + timeout: float = 1.0, + *, + response_type: Literal[AtResponseType.MULTIPLE], + ) -> list[AtResponse]: ... async def execute_command( self, @@ -835,27 +800,34 @@ asyncio.TimeoutError: the status is not received after a timeout (default 1 second). ProtocolError: the status is not OK. """ - async with self.command_lock: - logger.debug(f">>> {cmd}") - self.dlc.write(cmd + '\r') - responses: list[AtResponse] = [] + try: + async with self.command_lock: + self.pending_command = cmd + logger.debug(f">>> {cmd}") + self.dlc.write(cmd + '\r') + responses: list[AtResponse] = [] - while True: - result = await asyncio.wait_for( - self.response_queue.get(), timeout=timeout - ) - if result.code == 'OK': - if response_type == AtResponseType.SINGLE and len(responses) != 1: - raise HfpProtocolError("NO ANSWER") + while True: + result = await asyncio.wait_for( + self.response_queue.get(), timeout=timeout + ) + if result.code == 'OK': + if ( + response_type == AtResponseType.SINGLE + and len(responses) != 1 + ): + raise HfpProtocolError("NO ANSWER") - if response_type == AtResponseType.MULTIPLE: - return responses - if response_type == AtResponseType.SINGLE: - return responses[0] - return None - if result.code in STATUS_CODES: - raise HfpProtocolError(result.code) - responses.append(result) + if response_type == AtResponseType.MULTIPLE: + return responses + if response_type == AtResponseType.SINGLE: + return responses[0] + return None + if result.code in STATUS_CODES: + raise HfpProtocolError(result.code) + responses.append(result) + finally: + self.pending_command = None async def initiate_slc(self): """4.2.1 Service Level Connection Initialization.""" @@ -1067,7 +1039,6 @@ responses = await self.execute_command( "AT+CLCC", response_type=AtResponseType.MULTIPLE ) - assert isinstance(responses, list) calls = [] for response in responses:
diff --git a/bumble/host.py b/bumble/host.py index 98fd131..f0d0f14 100644 --- a/bumble/host.py +++ b/bumble/host.py
@@ -22,7 +22,7 @@ import dataclasses import logging from collections.abc import Awaitable, Callable -from typing import TYPE_CHECKING, Any, TypeVar, cast, overload +from typing import TYPE_CHECKING, Any, TypeVar, overload from bumble import drivers, hci, utils from bumble.colors import color @@ -1002,18 +1002,19 @@ self.snooper.snoop(bytes(packet), Snooper.Direction.CONTROLLER_TO_HOST) # If the packet is a command, invoke the handler for this packet - if packet.hci_packet_type == hci.HCI_COMMAND_PACKET: - self.on_hci_command_packet(cast(hci.HCI_Command, packet)) - elif packet.hci_packet_type == hci.HCI_EVENT_PACKET: - self.on_hci_event_packet(cast(hci.HCI_Event, packet)) - elif packet.hci_packet_type == hci.HCI_ACL_DATA_PACKET: - self.on_hci_acl_data_packet(cast(hci.HCI_AclDataPacket, packet)) - elif packet.hci_packet_type == hci.HCI_SYNCHRONOUS_DATA_PACKET: - self.on_hci_sco_data_packet(cast(hci.HCI_SynchronousDataPacket, packet)) - elif packet.hci_packet_type == hci.HCI_ISO_DATA_PACKET: - self.on_hci_iso_data_packet(cast(hci.HCI_IsoDataPacket, packet)) - else: - logger.warning(f'!!! unknown packet type {packet.hci_packet_type}') + match packet: + case hci.HCI_Command(): + self.on_hci_command_packet(packet) + case hci.HCI_Event(): + self.on_hci_event_packet(packet) + case hci.HCI_AclDataPacket(): + self.on_hci_acl_data_packet(packet) + case hci.HCI_SynchronousDataPacket(): + self.on_hci_sco_data_packet(packet) + case hci.HCI_IsoDataPacket(): + self.on_hci_iso_data_packet(packet) + case _: + logger.warning(f'!!! unknown packet type {packet.hci_packet_type}') def on_hci_command_packet(self, command: hci.HCI_Command) -> None: logger.warning(f'!!! unexpected command packet: {command}')
diff --git a/bumble/ll.py b/bumble/ll.py index 08d40c7..0cbf3e9 100644 --- a/bumble/ll.py +++ b/bumble/ll.py
@@ -198,3 +198,24 @@ cig_id: int cis_id: int error_code: int + + +@dataclasses.dataclass +class FeatureReq(ControlPdu): + opcode = ControlPdu.Opcode.LL_FEATURE_REQ + + feature_set: bytes + + +@dataclasses.dataclass +class FeatureRsp(ControlPdu): + opcode = ControlPdu.Opcode.LL_FEATURE_RSP + + feature_set: bytes + + +@dataclasses.dataclass +class PeripheralFeatureReq(ControlPdu): + opcode = ControlPdu.Opcode.LL_PERIPHERAL_FEATURE_REQ + + feature_set: bytes
diff --git a/bumble/pairing.py b/bumble/pairing.py index b6509a6..d02946a 100644 --- a/bumble/pairing.py +++ b/bumble/pairing.py
@@ -21,18 +21,9 @@ import secrets from dataclasses import dataclass -from bumble import hci +from bumble import hci, smp from bumble.core import AdvertisingData, LeRole from bumble.smp import ( - SMP_DISPLAY_ONLY_IO_CAPABILITY, - SMP_DISPLAY_YES_NO_IO_CAPABILITY, - SMP_ENC_KEY_DISTRIBUTION_FLAG, - SMP_ID_KEY_DISTRIBUTION_FLAG, - SMP_KEYBOARD_DISPLAY_IO_CAPABILITY, - SMP_KEYBOARD_ONLY_IO_CAPABILITY, - SMP_LINK_KEY_DISTRIBUTION_FLAG, - SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY, - SMP_SIGN_KEY_DISTRIBUTION_FLAG, OobContext, OobLegacyContext, OobSharedData, @@ -96,11 +87,11 @@ # These are defined abstractly, and can be mapped to specific Classic pairing # and/or SMP constants. class IoCapability(enum.IntEnum): - NO_OUTPUT_NO_INPUT = SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY - KEYBOARD_INPUT_ONLY = SMP_KEYBOARD_ONLY_IO_CAPABILITY - DISPLAY_OUTPUT_ONLY = SMP_DISPLAY_ONLY_IO_CAPABILITY - DISPLAY_OUTPUT_AND_YES_NO_INPUT = SMP_DISPLAY_YES_NO_IO_CAPABILITY - DISPLAY_OUTPUT_AND_KEYBOARD_INPUT = SMP_KEYBOARD_DISPLAY_IO_CAPABILITY + NO_OUTPUT_NO_INPUT = smp.IoCapability.NO_INPUT_NO_OUTPUT + KEYBOARD_INPUT_ONLY = smp.IoCapability.KEYBOARD_ONLY + DISPLAY_OUTPUT_ONLY = smp.IoCapability.DISPLAY_ONLY + DISPLAY_OUTPUT_AND_YES_NO_INPUT = smp.IoCapability.DISPLAY_YES_NO + DISPLAY_OUTPUT_AND_KEYBOARD_INPUT = smp.IoCapability.KEYBOARD_DISPLAY # Direct names for backward compatibility. NO_OUTPUT_NO_INPUT = IoCapability.NO_OUTPUT_NO_INPUT @@ -111,10 +102,10 @@ # Key Distribution [LE only] class KeyDistribution(enum.IntFlag): - DISTRIBUTE_ENCRYPTION_KEY = SMP_ENC_KEY_DISTRIBUTION_FLAG - DISTRIBUTE_IDENTITY_KEY = SMP_ID_KEY_DISTRIBUTION_FLAG - DISTRIBUTE_SIGNING_KEY = SMP_SIGN_KEY_DISTRIBUTION_FLAG - DISTRIBUTE_LINK_KEY = SMP_LINK_KEY_DISTRIBUTION_FLAG + DISTRIBUTE_ENCRYPTION_KEY = smp.KeyDistribution.ENC_KEY + DISTRIBUTE_IDENTITY_KEY = smp.KeyDistribution.ID_KEY + DISTRIBUTE_SIGNING_KEY = smp.KeyDistribution.SIGN_KEY + DISTRIBUTE_LINK_KEY = smp.KeyDistribution.LINK_KEY DEFAULT_KEY_DISTRIBUTION: KeyDistribution = ( KeyDistribution.DISTRIBUTE_ENCRYPTION_KEY
diff --git a/bumble/profiles/ascs.py b/bumble/profiles/ascs.py index 1e9d591..50113c3 100644 --- a/bumble/profiles/ascs.py +++ b/bumble/profiles/ascs.py
@@ -664,46 +664,44 @@ responses = [] logger.debug(f'*** ASCS Write {operation} ***') - if isinstance(operation, ASE_Config_Codec): - for ase_id, *args in zip( - operation.ase_id, - operation.target_latency, - operation.target_phy, - operation.codec_id, - operation.codec_specific_configuration, + match operation: + case ASE_Config_Codec(): + for ase_id, *args in zip( + operation.ase_id, + operation.target_latency, + operation.target_phy, + operation.codec_id, + operation.codec_specific_configuration, + ): + responses.append(self.on_operation(operation.op_code, ase_id, args)) + case ASE_Config_QOS(): + for ase_id, *args in zip( + operation.ase_id, + operation.cig_id, + operation.cis_id, + operation.sdu_interval, + operation.framing, + operation.phy, + operation.max_sdu, + operation.retransmission_number, + operation.max_transport_latency, + operation.presentation_delay, + ): + responses.append(self.on_operation(operation.op_code, ase_id, args)) + case ASE_Enable() | ASE_Update_Metadata(): + for ase_id, *args in zip( + operation.ase_id, + operation.metadata, + ): + responses.append(self.on_operation(operation.op_code, ase_id, args)) + case ( + ASE_Receiver_Start_Ready() + | ASE_Disable() + | ASE_Receiver_Stop_Ready() + | ASE_Release() ): - responses.append(self.on_operation(operation.op_code, ase_id, args)) - elif isinstance(operation, ASE_Config_QOS): - for ase_id, *args in zip( - operation.ase_id, - operation.cig_id, - operation.cis_id, - operation.sdu_interval, - operation.framing, - operation.phy, - operation.max_sdu, - operation.retransmission_number, - operation.max_transport_latency, - operation.presentation_delay, - ): - responses.append(self.on_operation(operation.op_code, ase_id, args)) - elif isinstance(operation, (ASE_Enable, ASE_Update_Metadata)): - for ase_id, *args in zip( - operation.ase_id, - operation.metadata, - ): - responses.append(self.on_operation(operation.op_code, ase_id, args)) - elif isinstance( - operation, - ( - ASE_Receiver_Start_Ready, - ASE_Disable, - ASE_Receiver_Stop_Ready, - ASE_Release, - ), - ): - for ase_id in operation.ase_id: - responses.append(self.on_operation(operation.op_code, ase_id, [])) + for ase_id in operation.ase_id: + responses.append(self.on_operation(operation.op_code, ase_id, [])) control_point_notification = bytes( [operation.op_code, len(responses)]
diff --git a/bumble/profiles/bap.py b/bumble/profiles/bap.py index 49c2e3d..b569f72 100644 --- a/bumble/profiles/bap.py +++ b/bumble/profiles/bap.py
@@ -333,17 +333,18 @@ value = int.from_bytes(data[offset : offset + length - 1], 'little') offset += length - 1 - if type == CodecSpecificCapabilities.Type.SAMPLING_FREQUENCY: - supported_sampling_frequencies = SupportedSamplingFrequency(value) - elif type == CodecSpecificCapabilities.Type.FRAME_DURATION: - supported_frame_durations = SupportedFrameDuration(value) - elif type == CodecSpecificCapabilities.Type.AUDIO_CHANNEL_COUNT: - supported_audio_channel_count = bits_to_channel_counts(value) - elif type == CodecSpecificCapabilities.Type.OCTETS_PER_FRAME: - min_octets_per_sample = value & 0xFFFF - max_octets_per_sample = value >> 16 - elif type == CodecSpecificCapabilities.Type.CODEC_FRAMES_PER_SDU: - supported_max_codec_frames_per_sdu = value + match type: + case CodecSpecificCapabilities.Type.SAMPLING_FREQUENCY: + supported_sampling_frequencies = SupportedSamplingFrequency(value) + case CodecSpecificCapabilities.Type.FRAME_DURATION: + supported_frame_durations = SupportedFrameDuration(value) + case CodecSpecificCapabilities.Type.AUDIO_CHANNEL_COUNT: + supported_audio_channel_count = bits_to_channel_counts(value) + case CodecSpecificCapabilities.Type.OCTETS_PER_FRAME: + min_octets_per_sample = value & 0xFFFF + max_octets_per_sample = value >> 16 + case CodecSpecificCapabilities.Type.CODEC_FRAMES_PER_SDU: + supported_max_codec_frames_per_sdu = value # It is expected here that if some fields are missing, an error should be raised. # pylint: disable=possibly-used-before-assignment,used-before-assignment
diff --git a/bumble/profiles/gap.py b/bumble/profiles/gap.py index 3b818af..9ff374d 100644 --- a/bumble/profiles/gap.py +++ b/bumble/profiles/gap.py
@@ -55,14 +55,15 @@ def __init__( self, device_name: str, appearance: Appearance | tuple[int, int] | int = 0 ): - if isinstance(appearance, int): - appearance_int = appearance - elif isinstance(appearance, tuple): - appearance_int = (appearance[0] << 6) | appearance[1] - elif isinstance(appearance, Appearance): - appearance_int = int(appearance) - else: - raise TypeError() + match appearance: + case int(): + appearance_int = appearance + case tuple(): + appearance_int = (appearance[0] << 6) | appearance[1] + case Appearance(): + appearance_int = int(appearance) + case _: + raise TypeError() self.device_name_characteristic = Characteristic( GATT_DEVICE_NAME_CHARACTERISTIC,
diff --git a/bumble/smp.py b/bumble/smp.py index 76b4d00..9d0bb7c 100644 --- a/bumble/smp.py +++ b/bumble/smp.py
@@ -31,14 +31,13 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, ClassVar, TypeVar, cast -from bumble import crypto, utils +from bumble import crypto, hci, utils from bumble.colors import color from bumble.core import ( AdvertisingData, InvalidArgumentError, PhysicalTransport, ProtocolError, - name_or_number, ) from bumble.hci import ( Address, @@ -46,7 +45,6 @@ HCI_LE_Enable_Encryption_Command, HCI_Object, Role, - key_with_value, metadata, ) from bumble.keys import PairingKeys @@ -71,110 +69,110 @@ SMP_CID = 0x06 SMP_BR_CID = 0x07 -SMP_PAIRING_REQUEST_COMMAND = 0x01 -SMP_PAIRING_RESPONSE_COMMAND = 0x02 -SMP_PAIRING_CONFIRM_COMMAND = 0x03 -SMP_PAIRING_RANDOM_COMMAND = 0x04 -SMP_PAIRING_FAILED_COMMAND = 0x05 -SMP_ENCRYPTION_INFORMATION_COMMAND = 0x06 -SMP_MASTER_IDENTIFICATION_COMMAND = 0x07 -SMP_IDENTITY_INFORMATION_COMMAND = 0x08 -SMP_IDENTITY_ADDRESS_INFORMATION_COMMAND = 0x09 -SMP_SIGNING_INFORMATION_COMMAND = 0x0A -SMP_SECURITY_REQUEST_COMMAND = 0x0B -SMP_PAIRING_PUBLIC_KEY_COMMAND = 0x0C -SMP_PAIRING_DHKEY_CHECK_COMMAND = 0x0D -SMP_PAIRING_KEYPRESS_NOTIFICATION_COMMAND = 0x0E +class CommandCode(hci.SpecableEnum): + PAIRING_REQUEST = 0x01 + PAIRING_RESPONSE = 0x02 + PAIRING_CONFIRM = 0x03 + PAIRING_RANDOM = 0x04 + PAIRING_FAILED = 0x05 + ENCRYPTION_INFORMATION = 0x06 + MASTER_IDENTIFICATION = 0x07 + IDENTITY_INFORMATION = 0x08 + IDENTITY_ADDRESS_INFORMATION = 0x09 + SIGNING_INFORMATION = 0x0A + SECURITY_REQUEST = 0x0B + PAIRING_PUBLIC_KEY = 0x0C + PAIRING_DHKEY_CHECK = 0x0D + PAIRING_KEYPRESS_NOTIFICATION = 0x0E -SMP_COMMAND_NAMES = { - SMP_PAIRING_REQUEST_COMMAND: 'SMP_PAIRING_REQUEST_COMMAND', - SMP_PAIRING_RESPONSE_COMMAND: 'SMP_PAIRING_RESPONSE_COMMAND', - SMP_PAIRING_CONFIRM_COMMAND: 'SMP_PAIRING_CONFIRM_COMMAND', - SMP_PAIRING_RANDOM_COMMAND: 'SMP_PAIRING_RANDOM_COMMAND', - SMP_PAIRING_FAILED_COMMAND: 'SMP_PAIRING_FAILED_COMMAND', - SMP_ENCRYPTION_INFORMATION_COMMAND: 'SMP_ENCRYPTION_INFORMATION_COMMAND', - SMP_MASTER_IDENTIFICATION_COMMAND: 'SMP_MASTER_IDENTIFICATION_COMMAND', - SMP_IDENTITY_INFORMATION_COMMAND: 'SMP_IDENTITY_INFORMATION_COMMAND', - SMP_IDENTITY_ADDRESS_INFORMATION_COMMAND: 'SMP_IDENTITY_ADDRESS_INFORMATION_COMMAND', - SMP_SIGNING_INFORMATION_COMMAND: 'SMP_SIGNING_INFORMATION_COMMAND', - SMP_SECURITY_REQUEST_COMMAND: 'SMP_SECURITY_REQUEST_COMMAND', - SMP_PAIRING_PUBLIC_KEY_COMMAND: 'SMP_PAIRING_PUBLIC_KEY_COMMAND', - SMP_PAIRING_DHKEY_CHECK_COMMAND: 'SMP_PAIRING_DHKEY_CHECK_COMMAND', - SMP_PAIRING_KEYPRESS_NOTIFICATION_COMMAND: 'SMP_PAIRING_KEYPRESS_NOTIFICATION_COMMAND' -} -SMP_DISPLAY_ONLY_IO_CAPABILITY = 0x00 -SMP_DISPLAY_YES_NO_IO_CAPABILITY = 0x01 -SMP_KEYBOARD_ONLY_IO_CAPABILITY = 0x02 -SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY = 0x03 -SMP_KEYBOARD_DISPLAY_IO_CAPABILITY = 0x04 +class IoCapability(hci.SpecableEnum): + DISPLAY_ONLY = 0x00 + DISPLAY_YES_NO = 0x01 + KEYBOARD_ONLY = 0x02 + NO_INPUT_NO_OUTPUT = 0x03 + KEYBOARD_DISPLAY = 0x04 -SMP_IO_CAPABILITY_NAMES = { - SMP_DISPLAY_ONLY_IO_CAPABILITY: 'SMP_DISPLAY_ONLY_IO_CAPABILITY', - SMP_DISPLAY_YES_NO_IO_CAPABILITY: 'SMP_DISPLAY_YES_NO_IO_CAPABILITY', - SMP_KEYBOARD_ONLY_IO_CAPABILITY: 'SMP_KEYBOARD_ONLY_IO_CAPABILITY', - SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY: 'SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY', - SMP_KEYBOARD_DISPLAY_IO_CAPABILITY: 'SMP_KEYBOARD_DISPLAY_IO_CAPABILITY' -} +SMP_DISPLAY_ONLY_IO_CAPABILITY = IoCapability.DISPLAY_ONLY +SMP_DISPLAY_YES_NO_IO_CAPABILITY = IoCapability.DISPLAY_YES_NO +SMP_KEYBOARD_ONLY_IO_CAPABILITY = IoCapability.KEYBOARD_ONLY +SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY = IoCapability.NO_INPUT_NO_OUTPUT +SMP_KEYBOARD_DISPLAY_IO_CAPABILITY = IoCapability.KEYBOARD_DISPLAY -SMP_PASSKEY_ENTRY_FAILED_ERROR = 0x01 -SMP_OOB_NOT_AVAILABLE_ERROR = 0x02 -SMP_AUTHENTICATION_REQUIREMENTS_ERROR = 0x03 -SMP_CONFIRM_VALUE_FAILED_ERROR = 0x04 -SMP_PAIRING_NOT_SUPPORTED_ERROR = 0x05 -SMP_ENCRYPTION_KEY_SIZE_ERROR = 0x06 -SMP_COMMAND_NOT_SUPPORTED_ERROR = 0x07 -SMP_UNSPECIFIED_REASON_ERROR = 0x08 -SMP_REPEATED_ATTEMPTS_ERROR = 0x09 -SMP_INVALID_PARAMETERS_ERROR = 0x0A -SMP_DHKEY_CHECK_FAILED_ERROR = 0x0B -SMP_NUMERIC_COMPARISON_FAILED_ERROR = 0x0C -SMP_BD_EDR_PAIRING_IN_PROGRESS_ERROR = 0x0D -SMP_CROSS_TRANSPORT_KEY_DERIVATION_NOT_ALLOWED_ERROR = 0x0E +class ErrorCode(hci.SpecableEnum): + PASSKEY_ENTRY_FAILED = 0x01 + OOB_NOT_AVAILABLE = 0x02 + AUTHENTICATION_REQUIREMENTS = 0x03 + CONFIRM_VALUE_FAILED = 0x04 + PAIRING_NOT_SUPPORTED = 0x05 + ENCRYPTION_KEY_SIZE = 0x06 + COMMAND_NOT_SUPPORTED = 0x07 + UNSPECIFIED_REASON = 0x08 + REPEATED_ATTEMPTS = 0x09 + INVALID_PARAMETERS = 0x0A + DHKEY_CHECK_FAILED = 0x0B + NUMERIC_COMPARISON_FAILED = 0x0C + BD_EDR_PAIRING_IN_PROGRESS = 0x0D + CROSS_TRANSPORT_KEY_DERIVATION_NOT_ALLOWED = 0x0E -SMP_ERROR_NAMES = { - SMP_PASSKEY_ENTRY_FAILED_ERROR: 'SMP_PASSKEY_ENTRY_FAILED_ERROR', - SMP_OOB_NOT_AVAILABLE_ERROR: 'SMP_OOB_NOT_AVAILABLE_ERROR', - SMP_AUTHENTICATION_REQUIREMENTS_ERROR: 'SMP_AUTHENTICATION_REQUIREMENTS_ERROR', - SMP_CONFIRM_VALUE_FAILED_ERROR: 'SMP_CONFIRM_VALUE_FAILED_ERROR', - SMP_PAIRING_NOT_SUPPORTED_ERROR: 'SMP_PAIRING_NOT_SUPPORTED_ERROR', - SMP_ENCRYPTION_KEY_SIZE_ERROR: 'SMP_ENCRYPTION_KEY_SIZE_ERROR', - SMP_COMMAND_NOT_SUPPORTED_ERROR: 'SMP_COMMAND_NOT_SUPPORTED_ERROR', - SMP_UNSPECIFIED_REASON_ERROR: 'SMP_UNSPECIFIED_REASON_ERROR', - SMP_REPEATED_ATTEMPTS_ERROR: 'SMP_REPEATED_ATTEMPTS_ERROR', - SMP_INVALID_PARAMETERS_ERROR: 'SMP_INVALID_PARAMETERS_ERROR', - SMP_DHKEY_CHECK_FAILED_ERROR: 'SMP_DHKEY_CHECK_FAILED_ERROR', - SMP_NUMERIC_COMPARISON_FAILED_ERROR: 'SMP_NUMERIC_COMPARISON_FAILED_ERROR', - SMP_BD_EDR_PAIRING_IN_PROGRESS_ERROR: 'SMP_BD_EDR_PAIRING_IN_PROGRESS_ERROR', - SMP_CROSS_TRANSPORT_KEY_DERIVATION_NOT_ALLOWED_ERROR: 'SMP_CROSS_TRANSPORT_KEY_DERIVATION_NOT_ALLOWED_ERROR' -} +SMP_PASSKEY_ENTRY_FAILED_ERROR = ErrorCode.PASSKEY_ENTRY_FAILED +SMP_OOB_NOT_AVAILABLE_ERROR = ErrorCode.OOB_NOT_AVAILABLE +SMP_AUTHENTICATION_REQUIREMENTS_ERROR = ErrorCode.AUTHENTICATION_REQUIREMENTS +SMP_CONFIRM_VALUE_FAILED_ERROR = ErrorCode.CONFIRM_VALUE_FAILED +SMP_PAIRING_NOT_SUPPORTED_ERROR = ErrorCode.PAIRING_NOT_SUPPORTED +SMP_ENCRYPTION_KEY_SIZE_ERROR = ErrorCode.ENCRYPTION_KEY_SIZE +SMP_COMMAND_NOT_SUPPORTED_ERROR = ErrorCode.COMMAND_NOT_SUPPORTED +SMP_UNSPECIFIED_REASON_ERROR = ErrorCode.UNSPECIFIED_REASON +SMP_REPEATED_ATTEMPTS_ERROR = ErrorCode.REPEATED_ATTEMPTS +SMP_INVALID_PARAMETERS_ERROR = ErrorCode.INVALID_PARAMETERS +SMP_DHKEY_CHECK_FAILED_ERROR = ErrorCode.DHKEY_CHECK_FAILED +SMP_NUMERIC_COMPARISON_FAILED_ERROR = ErrorCode.NUMERIC_COMPARISON_FAILED +SMP_BD_EDR_PAIRING_IN_PROGRESS_ERROR = ErrorCode.BD_EDR_PAIRING_IN_PROGRESS +SMP_CROSS_TRANSPORT_KEY_DERIVATION_NOT_ALLOWED_ERROR = ErrorCode.CROSS_TRANSPORT_KEY_DERIVATION_NOT_ALLOWED -SMP_PASSKEY_ENTRY_STARTED_KEYPRESS_NOTIFICATION_TYPE = 0 -SMP_PASSKEY_DIGIT_ENTERED_KEYPRESS_NOTIFICATION_TYPE = 1 -SMP_PASSKEY_DIGIT_ERASED_KEYPRESS_NOTIFICATION_TYPE = 2 -SMP_PASSKEY_CLEARED_KEYPRESS_NOTIFICATION_TYPE = 3 -SMP_PASSKEY_ENTRY_COMPLETED_KEYPRESS_NOTIFICATION_TYPE = 4 - -SMP_KEYPRESS_NOTIFICATION_TYPE_NAMES = { - SMP_PASSKEY_ENTRY_STARTED_KEYPRESS_NOTIFICATION_TYPE: 'SMP_PASSKEY_ENTRY_STARTED_KEYPRESS_NOTIFICATION_TYPE', - SMP_PASSKEY_DIGIT_ENTERED_KEYPRESS_NOTIFICATION_TYPE: 'SMP_PASSKEY_DIGIT_ENTERED_KEYPRESS_NOTIFICATION_TYPE', - SMP_PASSKEY_DIGIT_ERASED_KEYPRESS_NOTIFICATION_TYPE: 'SMP_PASSKEY_DIGIT_ERASED_KEYPRESS_NOTIFICATION_TYPE', - SMP_PASSKEY_CLEARED_KEYPRESS_NOTIFICATION_TYPE: 'SMP_PASSKEY_CLEARED_KEYPRESS_NOTIFICATION_TYPE', - SMP_PASSKEY_ENTRY_COMPLETED_KEYPRESS_NOTIFICATION_TYPE: 'SMP_PASSKEY_ENTRY_COMPLETED_KEYPRESS_NOTIFICATION_TYPE' -} +class KeypressNotificationType(hci.SpecableEnum): + PASSKEY_ENTRY_STARTED = 0 + PASSKEY_DIGIT_ENTERED = 1 + PASSKEY_DIGIT_ERASED = 2 + PASSKEY_CLEARED = 3 + PASSKEY_ENTRY_COMPLETED = 4 # Bit flags for key distribution/generation -SMP_ENC_KEY_DISTRIBUTION_FLAG = 0b0001 -SMP_ID_KEY_DISTRIBUTION_FLAG = 0b0010 -SMP_SIGN_KEY_DISTRIBUTION_FLAG = 0b0100 -SMP_LINK_KEY_DISTRIBUTION_FLAG = 0b1000 +class KeyDistribution(hci.SpecableFlag): + ENC_KEY = 0b0001 + ID_KEY = 0b0010 + SIGN_KEY = 0b0100 + LINK_KEY = 0b1000 # AuthReq fields -SMP_BONDING_AUTHREQ = 0b00000001 -SMP_MITM_AUTHREQ = 0b00000100 -SMP_SC_AUTHREQ = 0b00001000 -SMP_KEYPRESS_AUTHREQ = 0b00010000 -SMP_CT2_AUTHREQ = 0b00100000 +class AuthReq(hci.SpecableFlag): + BONDING = 0b00000001 + MITM = 0b00000100 + SC = 0b00001000 + KEYPRESS = 0b00010000 + CT2 = 0b00100000 + + @classmethod + def from_booleans( + cls, + bonding: bool = False, + sc: bool = False, + mitm: bool = False, + keypress: bool = False, + ct2: bool = False, + ) -> AuthReq: + auth_req = AuthReq(0) + if bonding: + auth_req |= AuthReq.BONDING + if sc: + auth_req |= AuthReq.SC + if mitm: + auth_req |= AuthReq.MITM + if keypress: + auth_req |= AuthReq.KEYPRESS + if ct2: + auth_req |= AuthReq.CT2 + return auth_req # Crypto salt SMP_CTKD_H7_LEBR_SALT = bytes.fromhex('000000000000000000000000746D7031') @@ -188,8 +186,6 @@ # ----------------------------------------------------------------------------- # Utils # ----------------------------------------------------------------------------- -def error_name(error_code: int) -> str: - return name_or_number(SMP_ERROR_NAMES, error_code) # ----------------------------------------------------------------------------- @@ -201,20 +197,20 @@ See Bluetooth spec @ Vol 3, Part H - 3 SECURITY MANAGER PROTOCOL ''' - smp_classes: ClassVar[dict[int, type[SMP_Command]]] = {} + smp_classes: ClassVar[dict[CommandCode, type[SMP_Command]]] = {} fields: ClassVar[Fields] - code: int = field(default=0, init=False) + code: CommandCode = field(default=CommandCode(0), init=False) name: str = field(default='', init=False) _payload: bytes | None = field(default=None, init=False) @classmethod def from_bytes(cls, pdu: bytes) -> SMP_Command: - code = pdu[0] + code = CommandCode(pdu[0]) subclass = SMP_Command.smp_classes.get(code) if subclass is None: instance = SMP_Command() - instance.name = SMP_Command.command_name(code) + instance.name = code.name instance.code = code instance.payload = pdu return instance @@ -222,59 +218,14 @@ instance.payload = pdu[1:] return instance - @staticmethod - def command_name(code: int) -> str: - return name_or_number(SMP_COMMAND_NAMES, code) - - @staticmethod - def auth_req_str(value: int) -> str: - bonding_flags = value & 3 - mitm = (value >> 2) & 1 - sc = (value >> 3) & 1 - keypress = (value >> 4) & 1 - ct2 = (value >> 5) & 1 - - return ( - f'bonding_flags={bonding_flags}, ' - f'MITM={mitm}, sc={sc}, keypress={keypress}, ct2={ct2}' - ) - - @staticmethod - def io_capability_name(io_capability: int) -> str: - return name_or_number(SMP_IO_CAPABILITY_NAMES, io_capability) - - @staticmethod - def key_distribution_str(value: int) -> str: - key_types: list[str] = [] - if value & SMP_ENC_KEY_DISTRIBUTION_FLAG: - key_types.append('ENC') - if value & SMP_ID_KEY_DISTRIBUTION_FLAG: - key_types.append('ID') - if value & SMP_SIGN_KEY_DISTRIBUTION_FLAG: - key_types.append('SIGN') - if value & SMP_LINK_KEY_DISTRIBUTION_FLAG: - key_types.append('LINK') - return ','.join(key_types) - - @staticmethod - def keypress_notification_type_name(notification_type: int) -> str: - return name_or_number(SMP_KEYPRESS_NOTIFICATION_TYPE_NAMES, notification_type) - _Command = TypeVar("_Command", bound="SMP_Command") @classmethod def subclass(cls, subclass: type[_Command]) -> type[_Command]: - subclass.name = subclass.__name__.upper() - subclass.code = key_with_value(SMP_COMMAND_NAMES, subclass.name) - if subclass.code is None: - raise KeyError( - f'Command name {subclass.name} not found in SMP_COMMAND_NAMES' - ) subclass.fields = HCI_Object.fields_from_dataclass(subclass) - + subclass.name = subclass.__name__.upper() # Register a factory for this class SMP_Command.smp_classes[subclass.code] = subclass - return subclass @property @@ -308,19 +259,17 @@ See Bluetooth spec @ Vol 3, Part H - 3.5.1 Pairing Request ''' - io_capability: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.io_capability_name}) - ) + code = CommandCode.PAIRING_REQUEST + + io_capability: IoCapability = field(metadata=IoCapability.type_metadata(1)) oob_data_flag: int = field(metadata=metadata(1)) - auth_req: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.auth_req_str}) - ) + auth_req: AuthReq = field(metadata=AuthReq.type_metadata(1)) maximum_encryption_key_size: int = field(metadata=metadata(1)) - initiator_key_distribution: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.key_distribution_str}) + initiator_key_distribution: KeyDistribution = field( + metadata=KeyDistribution.type_metadata(1) ) - responder_key_distribution: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.key_distribution_str}) + responder_key_distribution: KeyDistribution = field( + metadata=KeyDistribution.type_metadata(1) ) @@ -332,19 +281,17 @@ See Bluetooth spec @ Vol 3, Part H - 3.5.2 Pairing Response ''' - io_capability: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.io_capability_name}) - ) + code = CommandCode.PAIRING_RESPONSE + + io_capability: IoCapability = field(metadata=IoCapability.type_metadata(1)) oob_data_flag: int = field(metadata=metadata(1)) - auth_req: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.auth_req_str}) - ) + auth_req: AuthReq = field(metadata=AuthReq.type_metadata(1)) maximum_encryption_key_size: int = field(metadata=metadata(1)) - initiator_key_distribution: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.key_distribution_str}) + initiator_key_distribution: KeyDistribution = field( + metadata=KeyDistribution.type_metadata(1) ) - responder_key_distribution: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.key_distribution_str}) + responder_key_distribution: KeyDistribution = field( + metadata=KeyDistribution.type_metadata(1) ) @@ -356,6 +303,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.5.3 Pairing Confirm ''' + code = CommandCode.PAIRING_CONFIRM + confirm_value: bytes = field(metadata=metadata(16)) @@ -367,6 +316,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.5.4 Pairing Random ''' + code = CommandCode.PAIRING_RANDOM + random_value: bytes = field(metadata=metadata(16)) @@ -378,7 +329,9 @@ See Bluetooth spec @ Vol 3, Part H - 3.5.5 Pairing Failed ''' - reason: int = field(metadata=metadata({'size': 1, 'mapper': error_name})) + code = CommandCode.PAIRING_FAILED + + reason: ErrorCode = field(metadata=ErrorCode.type_metadata(1)) # ----------------------------------------------------------------------------- @@ -389,6 +342,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.5.6 Pairing Public Key ''' + code = CommandCode.PAIRING_PUBLIC_KEY + public_key_x: bytes = field(metadata=metadata(32)) public_key_y: bytes = field(metadata=metadata(32)) @@ -401,6 +356,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.5.7 Pairing DHKey Check ''' + code = CommandCode.PAIRING_DHKEY_CHECK + dhkey_check: bytes = field(metadata=metadata(16)) @@ -412,10 +369,10 @@ See Bluetooth spec @ Vol 3, Part H - 3.5.8 Keypress Notification ''' - notification_type: int = field( - metadata=metadata( - {'size': 1, 'mapper': SMP_Command.keypress_notification_type_name} - ) + code = CommandCode.PAIRING_KEYPRESS_NOTIFICATION + + notification_type: KeypressNotificationType = field( + metadata=KeypressNotificationType.type_metadata(1) ) @@ -427,6 +384,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.6.2 Encryption Information ''' + code = CommandCode.ENCRYPTION_INFORMATION + long_term_key: bytes = field(metadata=metadata(16)) @@ -438,6 +397,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.6.3 Master Identification ''' + code = CommandCode.MASTER_IDENTIFICATION + ediv: int = field(metadata=metadata(2)) rand: bytes = field(metadata=metadata(8)) @@ -450,6 +411,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.6.4 Identity Information ''' + code = CommandCode.IDENTITY_INFORMATION + identity_resolving_key: bytes = field(metadata=metadata(16)) @@ -461,6 +424,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.6.5 Identity Address Information ''' + code = CommandCode.IDENTITY_ADDRESS_INFORMATION + addr_type: int = field(metadata=metadata(Address.ADDRESS_TYPE_SPEC)) bd_addr: Address = field(metadata=metadata(Address.parse_address_preceded_by_type)) @@ -473,6 +438,8 @@ See Bluetooth spec @ Vol 3, Part H - 3.6.6 Signing Information ''' + code = CommandCode.SIGNING_INFORMATION + signature_key: bytes = field(metadata=metadata(16)) @@ -484,25 +451,9 @@ See Bluetooth spec @ Vol 3, Part H - 3.6.7 Security Request ''' - auth_req: int = field( - metadata=metadata({'size': 1, 'mapper': SMP_Command.auth_req_str}) - ) + code = CommandCode.SECURITY_REQUEST - -# ----------------------------------------------------------------------------- -def smp_auth_req(bonding: bool, mitm: bool, sc: bool, keypress: bool, ct2: bool) -> int: - value = 0 - if bonding: - value |= SMP_BONDING_AUTHREQ - if mitm: - value |= SMP_MITM_AUTHREQ - if sc: - value |= SMP_SC_AUTHREQ - if keypress: - value |= SMP_KEYPRESS_AUTHREQ - if ct2: - value |= SMP_CT2_AUTHREQ - return value + auth_req: AuthReq = field(metadata=AuthReq.type_metadata(1)) # ----------------------------------------------------------------------------- @@ -676,8 +627,8 @@ self.ltk_rand = bytes(8) self.link_key: bytes | None = None self.maximum_encryption_key_size: int = 0 - self.initiator_key_distribution: int = 0 - self.responder_key_distribution: int = 0 + self.initiator_key_distribution: KeyDistribution = KeyDistribution(0) + self.responder_key_distribution: KeyDistribution = KeyDistribution(0) self.peer_random_value: bytes | None = None self.peer_public_key_x: bytes = bytes(32) self.peer_public_key_y = bytes(32) @@ -728,10 +679,10 @@ ) # Key Distribution (default values before negotiation) - self.initiator_key_distribution = ( + self.initiator_key_distribution = KeyDistribution( pairing_config.delegate.local_initiator_key_distribution ) - self.responder_key_distribution = ( + self.responder_key_distribution = KeyDistribution( pairing_config.delegate.local_responder_key_distribution ) @@ -743,7 +694,7 @@ self.ct2: bool = False # I/O Capabilities - self.io_capability = pairing_config.delegate.io_capability + self.io_capability = IoCapability(pairing_config.delegate.io_capability) self.peer_io_capability = SMP_NO_INPUT_NO_OUTPUT_IO_CAPABILITY # OOB @@ -822,8 +773,14 @@ return self.nx[0 if self.is_responder else 1] @property - def auth_req(self) -> int: - return smp_auth_req(self.bonding, self.mitm, self.sc, self.keypress, self.ct2) + def auth_req(self) -> AuthReq: + return AuthReq.from_booleans( + bonding=self.bonding, + sc=self.sc, + mitm=self.mitm, + keypress=self.keypress, + ct2=self.ct2, + ) def get_long_term_key(self, rand: bytes, ediv: int) -> bytes | None: if not self.sc and not self.completed: @@ -843,7 +800,7 @@ if self.connection.transport == PhysicalTransport.BR_EDR: self.pairing_method = PairingMethod.CTKD_OVER_CLASSIC return - if (not self.mitm) and (auth_req & SMP_MITM_AUTHREQ == 0): + if (not self.mitm) and (auth_req & AuthReq.MITM == 0): self.pairing_method = PairingMethod.JUST_WORKS return @@ -861,7 +818,7 @@ self.passkey_display = details[1 if self.is_initiator else 2] def check_expected_value( - self, expected: bytes, received: bytes, error: int + self, expected: bytes, received: bytes, error: ErrorCode ) -> bool: logger.debug(f'expected={expected.hex()} got={received.hex()}') if expected != received: @@ -881,7 +838,7 @@ except Exception: logger.exception('exception while confirm') - self.send_pairing_failed(SMP_CONFIRM_VALUE_FAILED_ERROR) + self.send_pairing_failed(ErrorCode.CONFIRM_VALUE_FAILED) self.connection.cancel_on_disconnection(prompt()) @@ -900,7 +857,7 @@ except Exception: logger.exception('exception while prompting') - self.send_pairing_failed(SMP_CONFIRM_VALUE_FAILED_ERROR) + self.send_pairing_failed(ErrorCode.CONFIRM_VALUE_FAILED) self.connection.cancel_on_disconnection(prompt()) @@ -911,13 +868,13 @@ passkey = await self.pairing_config.delegate.get_number() if passkey is None: logger.debug('Passkey request rejected') - self.send_pairing_failed(SMP_PASSKEY_ENTRY_FAILED_ERROR) + self.send_pairing_failed(ErrorCode.PASSKEY_ENTRY_FAILED) return logger.debug(f'user input: {passkey}') next_steps(passkey) except Exception: logger.exception('exception while prompting') - self.send_pairing_failed(SMP_PASSKEY_ENTRY_FAILED_ERROR) + self.send_pairing_failed(ErrorCode.PASSKEY_ENTRY_FAILED) self.connection.cancel_on_disconnection(prompt()) @@ -972,7 +929,7 @@ def send_command(self, command: SMP_Command) -> None: self.manager.send_command(self.connection, command) - def send_pairing_failed(self, error: int) -> None: + def send_pairing_failed(self, error: ErrorCode) -> None: self.send_command(SMP_Pairing_Failed_Command(reason=error)) self.on_pairing_failure(error) @@ -1144,7 +1101,7 @@ 'Try to derive LTK but host does not have the LK. Send a SMP_PAIRING_FAILED but the procedure will not be paused!' ) self.send_pairing_failed( - SMP_CROSS_TRANSPORT_KEY_DERIVATION_NOT_ALLOWED_ERROR + ErrorCode.CROSS_TRANSPORT_KEY_DERIVATION_NOT_ALLOWED ) else: self.ltk = self.derive_ltk(self.link_key, self.ct2) @@ -1155,14 +1112,14 @@ # CTKD: Derive LTK from LinkKey if ( self.connection.transport == PhysicalTransport.BR_EDR - and self.initiator_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG + and self.initiator_key_distribution & KeyDistribution.ENC_KEY ): self.ctkd_task = self.connection.cancel_on_disconnection( self.get_link_key_and_derive_ltk() ) elif not self.sc: # Distribute the LTK, EDIV and RAND - if self.initiator_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG: + if self.initiator_key_distribution & KeyDistribution.ENC_KEY: self.send_command( SMP_Encryption_Information_Command(long_term_key=self.ltk) ) @@ -1173,7 +1130,7 @@ ) # Distribute IRK & BD ADDR - if self.initiator_key_distribution & SMP_ID_KEY_DISTRIBUTION_FLAG: + if self.initiator_key_distribution & KeyDistribution.ID_KEY: self.send_command( SMP_Identity_Information_Command( identity_resolving_key=self.manager.device.irk @@ -1183,25 +1140,25 @@ # Distribute CSRK csrk = bytes(16) # FIXME: testing - if self.initiator_key_distribution & SMP_SIGN_KEY_DISTRIBUTION_FLAG: + if self.initiator_key_distribution & KeyDistribution.SIGN_KEY: self.send_command(SMP_Signing_Information_Command(signature_key=csrk)) # CTKD, calculate BR/EDR link key - if self.initiator_key_distribution & SMP_LINK_KEY_DISTRIBUTION_FLAG: + if self.initiator_key_distribution & KeyDistribution.LINK_KEY: self.link_key = self.derive_link_key(self.ltk, self.ct2) else: # CTKD: Derive LTK from LinkKey if ( self.connection.transport == PhysicalTransport.BR_EDR - and self.responder_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG + and self.responder_key_distribution & KeyDistribution.ENC_KEY ): self.ctkd_task = self.connection.cancel_on_disconnection( self.get_link_key_and_derive_ltk() ) # Distribute the LTK, EDIV and RAND elif not self.sc: - if self.responder_key_distribution & SMP_ENC_KEY_DISTRIBUTION_FLAG: + if self.responder_key_distribution & KeyDistribution.ENC_KEY: self.send_command( SMP_Encryption_Information_Command(long_term_key=self.ltk) ) @@ -1212,7 +1169,7 @@ ) # Distribute IRK & BD ADDR - if self.responder_key_distribution & SMP_ID_KEY_DISTRIBUTION_FLAG: + if self.responder_key_distribution & KeyDistribution.ID_KEY: self.send_command( SMP_Identity_Information_Command( identity_resolving_key=self.manager.device.irk @@ -1222,30 +1179,30 @@ # Distribute CSRK csrk = bytes(16) # FIXME: testing - if self.responder_key_distribution & SMP_SIGN_KEY_DISTRIBUTION_FLAG: + if self.responder_key_distribution & KeyDistribution.SIGN_KEY: self.send_command(SMP_Signing_Information_Command(signature_key=csrk)) # CTKD, calculate BR/EDR link key - if self.responder_key_distribution & SMP_LINK_KEY_DISTRIBUTION_FLAG: + if self.responder_key_distribution & KeyDistribution.LINK_KEY: self.link_key = self.derive_link_key(self.ltk, self.ct2) def compute_peer_expected_distributions(self, key_distribution_flags: int) -> None: # Set our expectations for what to wait for in the key distribution phase self.peer_expected_distributions = [] if not self.sc and self.connection.transport == PhysicalTransport.LE: - if key_distribution_flags & SMP_ENC_KEY_DISTRIBUTION_FLAG != 0: + if key_distribution_flags & KeyDistribution.ENC_KEY != 0: self.peer_expected_distributions.append( SMP_Encryption_Information_Command ) self.peer_expected_distributions.append( SMP_Master_Identification_Command ) - if key_distribution_flags & SMP_ID_KEY_DISTRIBUTION_FLAG != 0: + if key_distribution_flags & KeyDistribution.ID_KEY != 0: self.peer_expected_distributions.append(SMP_Identity_Information_Command) self.peer_expected_distributions.append( SMP_Identity_Address_Information_Command ) - if key_distribution_flags & SMP_SIGN_KEY_DISTRIBUTION_FLAG != 0: + if key_distribution_flags & KeyDistribution.SIGN_KEY != 0: self.peer_expected_distributions.append(SMP_Signing_Information_Command) logger.debug( 'expecting distributions: ' @@ -1258,7 +1215,7 @@ logger.warning( color('received key distribution on a non-encrypted connection', 'red') ) - self.send_pairing_failed(SMP_UNSPECIFIED_REASON_ERROR) + self.send_pairing_failed(ErrorCode.UNSPECIFIED_REASON) return # Check that this command class is expected @@ -1278,7 +1235,7 @@ 'red', ) ) - self.send_pairing_failed(SMP_UNSPECIFIED_REASON_ERROR) + self.send_pairing_failed(ErrorCode.UNSPECIFIED_REASON) async def pair(self) -> None: # Start pairing as an initiator @@ -1389,34 +1346,56 @@ ) await self.manager.on_pairing(self, peer_address, keys) - def on_pairing_failure(self, reason: int) -> None: - logger.warning(f'pairing failure ({error_name(reason)})') + def on_pairing_failure(self, reason: ErrorCode) -> None: + logger.warning('pairing failure (%s)', reason.name) if self.completed: return self.completed = True - error = ProtocolError(reason, 'smp', error_name(reason)) + error = ProtocolError(reason, 'smp', reason.name) if self.pairing_result is not None and not self.pairing_result.done(): self.pairing_result.set_exception(error) self.manager.on_pairing_failure(self, reason) def on_smp_command(self, command: SMP_Command) -> None: - # Find the handler method - handler_name = f'on_{command.name.lower()}' - handler = getattr(self, handler_name, None) - if handler is not None: - try: - handler(command) - except Exception: - logger.exception(color("!!! Exception in handler:", "red")) - response = SMP_Pairing_Failed_Command( - reason=SMP_UNSPECIFIED_REASON_ERROR - ) - self.send_command(response) - else: - logger.error(color('SMP command not handled???', 'red')) + try: + match command: + case SMP_Pairing_Request_Command(): + self.on_smp_pairing_request_command(command) + case SMP_Pairing_Response_Command(): + self.on_smp_pairing_response_command(command) + case SMP_Pairing_Confirm_Command(): + self.on_smp_pairing_confirm_command(command) + case SMP_Pairing_Random_Command(): + self.on_smp_pairing_random_command(command) + case SMP_Pairing_Failed_Command(): + self.on_smp_pairing_failed_command(command) + case SMP_Encryption_Information_Command(): + self.on_smp_encryption_information_command(command) + case SMP_Master_Identification_Command(): + self.on_smp_master_identification_command(command) + case SMP_Identity_Information_Command(): + self.on_smp_identity_information_command(command) + case SMP_Identity_Address_Information_Command(): + self.on_smp_identity_address_information_command(command) + case SMP_Signing_Information_Command(): + self.on_smp_signing_information_command(command) + case SMP_Pairing_Public_Key_Command(): + self.on_smp_pairing_public_key_command(command) + case SMP_Pairing_DHKey_Check_Command(): + self.on_smp_pairing_dhkey_check_command(command) + # case SMP_Security_Request_Command(): + # self.on_smp_security_request_command(command) + # case SMP_Pairing_Keypress_Notification_Command(): + # self.on_smp_pairing_keypress_notification_command(command) + case _: + logger.error(color('SMP command not handled', 'red')) + except Exception: + logger.exception(color("!!! Exception in handler:", "red")) + response = SMP_Pairing_Failed_Command(reason=ErrorCode.UNSPECIFIED_REASON) + self.send_command(response) def on_smp_pairing_request_command( self, command: SMP_Pairing_Request_Command @@ -1436,16 +1415,16 @@ accepted = False if not accepted: logger.debug('pairing rejected by delegate') - self.send_pairing_failed(SMP_PAIRING_NOT_SUPPORTED_ERROR) + self.send_pairing_failed(ErrorCode.PAIRING_NOT_SUPPORTED) return # Save the request self.preq = bytes(command) # Bonding and SC require both sides to request/support it - self.bonding = self.bonding and (command.auth_req & SMP_BONDING_AUTHREQ != 0) - self.sc = self.sc and (command.auth_req & SMP_SC_AUTHREQ != 0) - self.ct2 = self.ct2 and (command.auth_req & SMP_CT2_AUTHREQ != 0) + self.bonding = self.bonding and (command.auth_req & AuthReq.BONDING != 0) + self.sc = self.sc and (command.auth_req & AuthReq.SC != 0) + self.ct2 = self.ct2 and (command.auth_req & AuthReq.CT2 != 0) # Infer the pairing method if (self.sc and (self.oob_data_flag != 0 or command.oob_data_flag != 0)) or ( @@ -1456,7 +1435,7 @@ if not self.sc and self.tk is None: # For legacy OOB, TK is required. logger.warning("legacy OOB without TK") - self.send_pairing_failed(SMP_OOB_NOT_AVAILABLE_ERROR) + self.send_pairing_failed(ErrorCode.OOB_NOT_AVAILABLE) return if command.oob_data_flag == 0: # The peer doesn't have OOB data, use r=0 @@ -1475,8 +1454,11 @@ ( self.initiator_key_distribution, self.responder_key_distribution, - ) = await self.pairing_config.delegate.key_distribution_response( - command.initiator_key_distribution, command.responder_key_distribution + ) = map( + KeyDistribution, + await self.pairing_config.delegate.key_distribution_response( + command.initiator_key_distribution, command.responder_key_distribution + ), ) self.compute_peer_expected_distributions(self.initiator_key_distribution) @@ -1514,8 +1496,8 @@ self.peer_io_capability = command.io_capability # Bonding and SC require both sides to request/support it - self.bonding = self.bonding and (command.auth_req & SMP_BONDING_AUTHREQ != 0) - self.sc = self.sc and (command.auth_req & SMP_SC_AUTHREQ != 0) + self.bonding = self.bonding and (command.auth_req & AuthReq.BONDING != 0) + self.sc = self.sc and (command.auth_req & AuthReq.SC != 0) # Infer the pairing method if (self.sc and (self.oob_data_flag != 0 or command.oob_data_flag != 0)) or ( @@ -1526,7 +1508,7 @@ if not self.sc and self.tk is None: # For legacy OOB, TK is required. logger.warning("legacy OOB without TK") - self.send_pairing_failed(SMP_OOB_NOT_AVAILABLE_ERROR) + self.send_pairing_failed(ErrorCode.OOB_NOT_AVAILABLE) return if command.oob_data_flag == 0: # The peer doesn't have OOB data, use r=0 @@ -1546,7 +1528,7 @@ command.responder_key_distribution & ~self.responder_key_distribution != 0 ): # The response isn't a subset of the request - self.send_pairing_failed(SMP_INVALID_PARAMETERS_ERROR) + self.send_pairing_failed(ErrorCode.INVALID_PARAMETERS) return self.initiator_key_distribution = command.initiator_key_distribution self.responder_key_distribution = command.responder_key_distribution @@ -1624,7 +1606,7 @@ ) assert self.confirm_value if not self.check_expected_value( - self.confirm_value, confirm_verifier, SMP_CONFIRM_VALUE_FAILED_ERROR + self.confirm_value, confirm_verifier, ErrorCode.CONFIRM_VALUE_FAILED ): return @@ -1665,7 +1647,7 @@ self.pkb, self.pka, command.random_value, bytes([0]) ) if not self.check_expected_value( - self.confirm_value, confirm_verifier, SMP_CONFIRM_VALUE_FAILED_ERROR + self.confirm_value, confirm_verifier, ErrorCode.CONFIRM_VALUE_FAILED ): return elif self.pairing_method == PairingMethod.PASSKEY: @@ -1678,7 +1660,7 @@ bytes([0x80 + ((self.passkey >> self.passkey_step) & 1)]), ) if not self.check_expected_value( - self.confirm_value, confirm_verifier, SMP_CONFIRM_VALUE_FAILED_ERROR + self.confirm_value, confirm_verifier, ErrorCode.CONFIRM_VALUE_FAILED ): return @@ -1707,7 +1689,7 @@ bytes([0x80 + ((self.passkey >> self.passkey_step) & 1)]), ) if not self.check_expected_value( - self.confirm_value, confirm_verifier, SMP_CONFIRM_VALUE_FAILED_ERROR + self.confirm_value, confirm_verifier, ErrorCode.CONFIRM_VALUE_FAILED ): return @@ -1824,7 +1806,7 @@ if not self.check_expected_value( self.peer_oob_data.c, confirm_verifier, - SMP_CONFIRM_VALUE_FAILED_ERROR, + ErrorCode.CONFIRM_VALUE_FAILED, ): return @@ -1858,7 +1840,7 @@ expected = self.eb if self.is_initiator else self.ea assert expected if not self.check_expected_value( - expected, command.dhkey_check, SMP_DHKEY_CHECK_FAILED_ERROR + expected, command.dhkey_check, ErrorCode.DHKEY_CHECK_FAILED ): return @@ -1962,7 +1944,7 @@ ) # Security request is more than just pairing, so let applications handle them - if command.code == SMP_SECURITY_REQUEST_COMMAND: + if command.code == CommandCode.SECURITY_REQUEST: self.on_smp_security_request_command( connection, cast(SMP_Security_Request_Command, command) ) @@ -2002,15 +1984,13 @@ def request_pairing(self, connection: Connection) -> None: pairing_config = self.pairing_config_factory(connection) if pairing_config: - auth_req = smp_auth_req( - pairing_config.bonding, - pairing_config.mitm, - pairing_config.sc, - False, - False, + auth_req = AuthReq.from_booleans( + bonding=pairing_config.bonding, + sc=pairing_config.sc, + mitm=pairing_config.mitm, ) else: - auth_req = 0 + auth_req = AuthReq(0) self.send_command(connection, SMP_Security_Request_Command(auth_req=auth_req)) def on_session_start(self, session: Session) -> None: @@ -2026,7 +2006,7 @@ # Notify the device self.device.on_pairing(session.connection, identity_address, keys, session.sc) - def on_pairing_failure(self, session: Session, reason: int) -> None: + def on_pairing_failure(self, session: Session, reason: ErrorCode) -> None: self.device.on_pairing_failure(session.connection, reason) def on_session_end(self, session: Session) -> None:
diff --git a/examples/run_avrcp.py b/examples/run_avrcp.py index 13b34d5..a05361e 100644 --- a/examples/run_avrcp.py +++ b/examples/run_avrcp.py
@@ -133,10 +133,10 @@ utils.AsyncRunner.spawn(get_supported_events()) async def monitor_track_changed() -> None: - async for identifier in avrcp_protocol.monitor_track_changed(): - print("TRACK CHANGED:", identifier.hex()) + async for uid in avrcp_protocol.monitor_track_changed(): + print("TRACK CHANGED:", hex(uid)) websocket_server.send_message( - {"type": "track-changed", "params": {"identifier": identifier.hex()}} + {"type": "track-changed", "params": {"identifier": hex(uid)}} ) async def monitor_playback_status() -> None:
diff --git a/pyproject.toml b/pyproject.toml index 60c7f12..601c336 100644 --- a/pyproject.toml +++ b/pyproject.toml
@@ -37,7 +37,7 @@ "pyserial-asyncio >= 0.5; platform_system!='Emscripten'", "pyserial >= 3.5; platform_system!='Emscripten'", "pyusb >= 1.2; platform_system!='Emscripten'", - "tomli ~= 2.2.1; platform_system!='Emscripten'", + "tomli ~= 2.2.1; platform_system!='Emscripten' and python_version<'3.11'", "websockets >= 15.0.1; platform_system!='Emscripten'", ]
diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 01c859d..92b7e4b 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock
@@ -221,9 +221,9 @@ [[package]] name = "bytes" -version = "1.5.0" +version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2bd12c1caf447e69cd4528f47f94d203fd2582878ecb9e9465484c4148a8223" +checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" [[package]] name = "cc"
diff --git a/rust/Cargo.toml b/rust/Cargo.toml index f5ada30..f820a60 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml
@@ -30,7 +30,7 @@ itertools = "0.11.0" lazy_static = "1.4.0" thiserror = "1.0.41" -bytes = "1.5.0" +bytes = "1.11.1" pdl-derive = "0.2.0" pdl-runtime = "0.2.0" futures = "0.3.28"
diff --git a/tests/avrcp_test.py b/tests/avrcp_test.py index 755ff17..1c50f7f 100644 --- a/tests/avrcp_test.py +++ b/tests/avrcp_test.py
@@ -20,6 +20,7 @@ import asyncio import struct from collections.abc import Sequence +from unittest import mock import pytest @@ -118,8 +119,6 @@ scope=avrcp.Scope.NOW_PLAYING, uid=0, uid_counter=1, - start_item=0, - end_item=0, attributes=[avrcp.MediaAttributeId.DEFAULT_COVER_ART], ), avrcp.GetTotalNumberOfItemsCommand(scope=avrcp.Scope.NOW_PLAYING), @@ -136,7 +135,7 @@ "event,", [ avrcp.UidsChangedEvent(uid_counter=7), - avrcp.TrackChangedEvent(identifier=b'12356'), + avrcp.TrackChangedEvent(uid=12356), avrcp.VolumeChangedEvent(volume=9), avrcp.PlaybackStatusChangedEvent(play_status=avrcp.PlayStatus.PLAYING), avrcp.AddressedPlayerChangedEvent( @@ -583,6 +582,87 @@ # ----------------------------------------------------------------------------- @pytest.mark.asyncio +async def test_list_player_application_settings(): + two_devices: TwoDevices = await TwoDevices.create_with_avdtp() + + expected_settings = { + avrcp.ApplicationSetting.AttributeId.REPEAT_MODE: [ + avrcp.ApplicationSetting.RepeatModeStatus.ALL_TRACK_REPEAT, + avrcp.ApplicationSetting.RepeatModeStatus.GROUP_REPEAT, + avrcp.ApplicationSetting.RepeatModeStatus.SINGLE_TRACK_REPEAT, + avrcp.ApplicationSetting.RepeatModeStatus.OFF, + ], + avrcp.ApplicationSetting.AttributeId.SHUFFLE_ON_OFF: [ + avrcp.ApplicationSetting.ShuffleOnOffStatus.OFF, + avrcp.ApplicationSetting.ShuffleOnOffStatus.ALL_TRACKS_SHUFFLE, + avrcp.ApplicationSetting.ShuffleOnOffStatus.GROUP_SHUFFLE, + ], + } + two_devices.protocols[1].delegate = avrcp.Delegate( + supported_player_app_settings=expected_settings + ) + actual_settings = await two_devices.protocols[ + 0 + ].list_supported_player_app_settings() + assert actual_settings == expected_settings + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_get_set_player_app_settings(): + two_devices: TwoDevices = await TwoDevices.create_with_avdtp() + + delegate = two_devices.protocols[1].delegate + await two_devices.protocols[0].send_avrcp_command( + avc.CommandFrame.CommandType.CONTROL, + avrcp.SetPlayerApplicationSettingValueCommand( + attribute=[ + avrcp.ApplicationSetting.AttributeId.REPEAT_MODE, + avrcp.ApplicationSetting.AttributeId.SHUFFLE_ON_OFF, + ], + value=[ + avrcp.ApplicationSetting.RepeatModeStatus.ALL_TRACK_REPEAT, + avrcp.ApplicationSetting.ShuffleOnOffStatus.GROUP_SHUFFLE, + ], + ), + ) + expected_settings = { + avrcp.ApplicationSetting.AttributeId.REPEAT_MODE: avrcp.ApplicationSetting.RepeatModeStatus.ALL_TRACK_REPEAT, + avrcp.ApplicationSetting.AttributeId.SHUFFLE_ON_OFF: avrcp.ApplicationSetting.ShuffleOnOffStatus.GROUP_SHUFFLE, + } + assert delegate.player_app_settings == expected_settings + + actual_settings = await two_devices.protocols[0].get_player_app_settings( + [ + avrcp.ApplicationSetting.AttributeId.REPEAT_MODE, + avrcp.ApplicationSetting.AttributeId.SHUFFLE_ON_OFF, + ] + ) + assert actual_settings == expected_settings + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_play_item(): + two_devices: TwoDevices = await TwoDevices.create_with_avdtp() + + delegate = two_devices.protocols[1].delegate + + with mock.patch.object(delegate, delegate.play_item.__name__) as play_item_mock: + await two_devices.protocols[0].send_avrcp_command( + avc.CommandFrame.CommandType.CONTROL, + avrcp.PlayItemCommand( + scope=avrcp.Scope.MEDIA_PLAYER_LIST, uid=0, uid_counter=1 + ), + ) + + play_item_mock.assert_called_once_with( + scope=avrcp.Scope.MEDIA_PLAYER_LIST, uid=0, uid_counter=1 + ) + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio async def test_monitor_volume(): two_devices = await TwoDevices.create_with_avdtp() @@ -636,6 +716,102 @@ # ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_monitor_track_changed(): + two_devices = await TwoDevices.create_with_avdtp() + + delegate = two_devices.protocols[1].delegate = avrcp.Delegate( + [avrcp.EventId.TRACK_CHANGED] + ) + delegate.current_track_uid = avrcp.TrackChangedEvent.NO_TRACK + track_iter = two_devices.protocols[0].monitor_track_changed() + + # Interim + assert (await anext(track_iter)) == avrcp.TrackChangedEvent.NO_TRACK + # Changed + two_devices.protocols[1].notify_track_changed(1) + assert (await anext(track_iter)) == 1 + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_monitor_uid_changed(): + two_devices = await TwoDevices.create_with_avdtp() + + delegate = two_devices.protocols[1].delegate = avrcp.Delegate( + [avrcp.EventId.UIDS_CHANGED] + ) + delegate.uid_counter = 0 + uid_iter = two_devices.protocols[0].monitor_uids() + + # Interim + assert (await anext(uid_iter)) == 0 + # Changed + two_devices.protocols[1].notify_uids_changed(1) + assert (await anext(uid_iter)) == 1 + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_monitor_addressed_player(): + two_devices = await TwoDevices.create_with_avdtp() + + delegate = two_devices.protocols[1].delegate = avrcp.Delegate( + [avrcp.EventId.ADDRESSED_PLAYER_CHANGED] + ) + delegate.uid_counter = 0 + delegate.addressed_player_id = 0 + addressed_player_iter = two_devices.protocols[0].monitor_addressed_player() + + # Interim + assert ( + await anext(addressed_player_iter) + ) == avrcp.AddressedPlayerChangedEvent.Player(player_id=0, uid_counter=0) + # Changed + two_devices.protocols[1].notify_addressed_player_changed( + avrcp.AddressedPlayerChangedEvent.Player(player_id=1, uid_counter=1) + ) + assert ( + await anext(addressed_player_iter) + ) == avrcp.AddressedPlayerChangedEvent.Player(player_id=1, uid_counter=1) + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_monitor_player_app_settings(): + two_devices = await TwoDevices.create_with_avdtp() + + delegate = two_devices.protocols[1].delegate = avrcp.Delegate( + supported_events=[avrcp.EventId.PLAYER_APPLICATION_SETTING_CHANGED] + ) + delegate.player_app_settings = { + avrcp.ApplicationSetting.AttributeId.REPEAT_MODE: avrcp.ApplicationSetting.RepeatModeStatus.ALL_TRACK_REPEAT + } + settings_iter = two_devices.protocols[0].monitor_player_application_settings() + + # Interim + interim = await anext(settings_iter) + assert interim[0].attribute_id == avrcp.ApplicationSetting.AttributeId.REPEAT_MODE + assert ( + interim[0].value_id + == avrcp.ApplicationSetting.RepeatModeStatus.ALL_TRACK_REPEAT + ) + + # Changed + two_devices.protocols[1].notify_player_application_settings_changed( + [ + avrcp.PlayerApplicationSettingChangedEvent.Setting( + avrcp.ApplicationSetting.AttributeId.REPEAT_MODE, + avrcp.ApplicationSetting.RepeatModeStatus.GROUP_REPEAT, + ) + ] + ) + changed = await anext(settings_iter) + assert changed[0].attribute_id == avrcp.ApplicationSetting.AttributeId.REPEAT_MODE + assert changed[0].value_id == avrcp.ApplicationSetting.RepeatModeStatus.GROUP_REPEAT + + +# ----------------------------------------------------------------------------- if __name__ == '__main__': test_frame_parser() test_vendor_dependent_command()
diff --git a/tests/device_test.py b/tests/device_test.py index 74837d2..dcb8346 100644 --- a/tests/device_test.py +++ b/tests/device_test.py
@@ -311,6 +311,27 @@ # ----------------------------------------------------------------------------- @pytest.mark.asyncio +async def test_le_multiple_connects(): + devices = TwoDevices() + for controller in devices.controllers: + controller.le_features |= hci.LeFeatureMask.LE_EXTENDED_ADVERTISING + for dev in devices: + await dev.power_on() + await devices[0].start_advertising(auto_restart=True, advertising_interval_min=1.0) + + connection = await devices[1].connect(devices[0].random_address) + await connection.disconnect() + + await async_barrier() + await async_barrier() + + # a second connection attempt is working + connection = await devices[1].connect(devices[0].random_address) + await connection.disconnect() + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio async def test_advertising_and_scanning(): devices = TwoDevices() for dev in devices: @@ -445,7 +466,9 @@ devices = TwoDevices() await devices.setup_connection() - assert (await devices.connections[0].get_remote_le_features()) is not None + assert ( + await devices.connections[0].get_remote_le_features() + ) == devices.controllers[1].le_features # -----------------------------------------------------------------------------
diff --git a/tests/self_test.py b/tests/self_test.py index 17acfdb..069c9a6 100644 --- a/tests/self_test.py +++ b/tests/self_test.py
@@ -29,8 +29,7 @@ from bumble.hci import Role from bumble.pairing import PairingConfig, PairingDelegate from bumble.smp import ( - SMP_CONFIRM_VALUE_FAILED_ERROR, - SMP_PAIRING_NOT_SUPPORTED_ERROR, + ErrorCode, OobContext, OobLegacyContext, ) @@ -378,7 +377,7 @@ await _test_self_smp_with_configs(None, rejecting_pairing_config) paired = True except ProtocolError as error: - assert error.error_code == SMP_PAIRING_NOT_SUPPORTED_ERROR + assert error.error_code == ErrorCode.PAIRING_NOT_SUPPORTED assert not paired @@ -403,7 +402,7 @@ ) paired = True except ProtocolError as error: - assert error.error_code == SMP_CONFIRM_VALUE_FAILED_ERROR + assert error.error_code == ErrorCode.CONFIRM_VALUE_FAILED assert not paired @@ -534,11 +533,11 @@ with pytest.raises(ProtocolError) as error: await _test_self_smp_with_configs(pairing_config_1, pairing_config_4) - assert error.value.error_code == SMP_CONFIRM_VALUE_FAILED_ERROR + assert error.value.error_code == ErrorCode.CONFIRM_VALUE_FAILED with pytest.raises(ProtocolError): await _test_self_smp_with_configs(pairing_config_4, pairing_config_1) - assert error.value.error_code == SMP_CONFIRM_VALUE_FAILED_ERROR + assert error.value.error_code == ErrorCode.CONFIRM_VALUE_FAILED # -----------------------------------------------------------------------------