# 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 # ----------------------------------------------------------------------------- from __future__ import annotations import asyncio import dataclasses import enum import itertools import logging import struct from collections import deque from collections.abc import Callable, Iterable, Sequence from typing import TYPE_CHECKING, Any, ClassVar, SupportsBytes, TypeVar from typing_extensions import override from bumble import hci, utils from bumble.colors import color from bumble.core import ( InvalidArgumentError, InvalidPacketError, InvalidStateError, OutOfResourcesError, ProtocolError, ) if TYPE_CHECKING: from bumble.device import Connection from bumble.host import Host # ----------------------------------------------------------------------------- # Logging # ----------------------------------------------------------------------------- logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- # Constants # ----------------------------------------------------------------------------- # fmt: off # pylint: disable=line-too-long L2CAP_SIGNALING_CID = 0x01 L2CAP_LE_SIGNALING_CID = 0x05 L2CAP_MIN_LE_MTU = 23 L2CAP_MIN_BR_EDR_MTU = 48 L2CAP_MAX_BR_EDR_MTU = 65535 L2CAP_DEFAULT_MTU = 2048 # Default value for the MTU we are willing to accept L2CAP_DEFAULT_MPS = 1010 # Default value for the MPS we are willing to accept DEFAULT_TX_WINDOW_SIZE = 63 DEFAULT_MAX_RETRANSMISSION = 1 DEFAULT_RETRANSMISSION_TIMEOUT = 2.0 DEFAULT_MONITOR_TIMEOUT = 12.0 L2CAP_DEFAULT_CONNECTIONLESS_MTU = 1024 # See Bluetooth spec @ Vol 3, Part A - Table 2.1: CID name space on ACL-U, ASB-U, and AMP-U logical links L2CAP_ACL_U_DYNAMIC_CID_RANGE_START = 0x0040 L2CAP_ACL_U_DYNAMIC_CID_RANGE_END = 0xFFFF # See Bluetooth spec @ Vol 3, Part A - Table 2.2: CID name space on LE-U logical link L2CAP_LE_U_DYNAMIC_CID_RANGE_START = 0x0040 L2CAP_LE_U_DYNAMIC_CID_RANGE_END = 0x007F # PSM Range - See Bluetooth spec @ Vol 3, Part A / Table 4.5: PSM ranges and usage L2CAP_PSM_DYNAMIC_RANGE_START = 0x1001 L2CAP_PSM_DYNAMIC_RANGE_END = 0xFFFF # LE PSM Ranges - See Bluetooth spec @ Vol 3, Part A / Table 4.19: LE Credit Based Connection Request LE_PSM ranges L2CAP_LE_PSM_DYNAMIC_RANGE_START = 0x0080 L2CAP_LE_PSM_DYNAMIC_RANGE_END = 0x00FF class CommandCode(hci.SpecableEnum): L2CAP_COMMAND_REJECT = 0x01 L2CAP_CONNECTION_REQUEST = 0x02 L2CAP_CONNECTION_RESPONSE = 0x03 L2CAP_CONFIGURE_REQUEST = 0x04 L2CAP_CONFIGURE_RESPONSE = 0x05 L2CAP_DISCONNECTION_REQUEST = 0x06 L2CAP_DISCONNECTION_RESPONSE = 0x07 L2CAP_ECHO_REQUEST = 0x08 L2CAP_ECHO_RESPONSE = 0x09 L2CAP_INFORMATION_REQUEST = 0x0A L2CAP_INFORMATION_RESPONSE = 0x0B L2CAP_CREATE_CHANNEL_REQUEST = 0x0C L2CAP_CREATE_CHANNEL_RESPONSE = 0x0D L2CAP_MOVE_CHANNEL_REQUEST = 0x0E L2CAP_MOVE_CHANNEL_RESPONSE = 0x0F L2CAP_MOVE_CHANNEL_CONFIRMATION = 0x10 L2CAP_MOVE_CHANNEL_CONFIRMATION_RESPONSE = 0x11 L2CAP_CONNECTION_PARAMETER_UPDATE_REQUEST = 0x12 L2CAP_CONNECTION_PARAMETER_UPDATE_RESPONSE = 0x13 L2CAP_LE_CREDIT_BASED_CONNECTION_REQUEST = 0x14 L2CAP_LE_CREDIT_BASED_CONNECTION_RESPONSE = 0x15 L2CAP_LE_FLOW_CONTROL_CREDIT = 0x16 L2CAP_CREDIT_BASED_CONNECTION_REQUEST = 0x17 L2CAP_CREDIT_BASED_CONNECTION_RESPONSE = 0x18 L2CAP_CREDIT_BASED_RECONFIGURE_REQUEST = 0x19 L2CAP_CREDIT_BASED_RECONFIGURE_RESPONSE = 0x1A L2CAP_CONNECTION_PARAMETERS_ACCEPTED_RESULT = 0x0000 L2CAP_CONNECTION_PARAMETERS_REJECTED_RESULT = 0x0001 L2CAP_COMMAND_NOT_UNDERSTOOD_REASON = 0x0000 L2CAP_SIGNALING_MTU_EXCEEDED_REASON = 0x0001 L2CAP_INVALID_CID_IN_REQUEST_REASON = 0x0002 L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_CREDITS = 65535 L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MTU = 23 L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MTU = 65535 L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MPS = 23 L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MPS = 65533 L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MTU = 2048 L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MPS = 2048 L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_INITIAL_CREDITS = 256 # fmt: on # pylint: enable=line-too-long class TransmissionMode(utils.OpenIntEnum): '''See Bluetooth spec @ Vol 3, Part A - 5.4. Retransmission and Flow Control option''' BASIC = 0x00 RETRANSMISSION = 0x01 FLOW_CONTROL = 0x02 ENHANCED_RETRANSMISSION = 0x03 STREAMING = 0x04 # ----------------------------------------------------------------------------- # Classes # ----------------------------------------------------------------------------- # pylint: disable=invalid-name class L2capError(ProtocolError): def __init__(self, error_code, error_name='', details=''): super().__init__(error_code, 'L2CAP', error_name, details) @dataclasses.dataclass class ClassicChannelSpec: '''Spec of L2CAP Channel over Classic Transport. Attributes: psm: PSM of channel. This is optional for server, and when it is None, a PSM will be allocated. mtu: Maximum Transmission Unit. mps: Maximum PDU payload Size. tx_window_size: The size of the transmission window for Flow Control mode, Retransmission mode, and Enhanced Retransmission mode. max_retransmission: The number of transmissions of a single I-frame that L2CAP is allowed to try in Retransmission mode and Enhanced Retransmission mode. retransmission_timeout: The timeout of retransmission in seconds. monitor_timeout: The interval at which S-frames should be transmitted on the return channel when no frames are received on the forward channel. mode: The transmission mode to use. fcs_enabled: Whether to enable FCS (Frame Check Sequence). ''' psm: int | None = None mtu: int = L2CAP_DEFAULT_MTU mps: int = L2CAP_DEFAULT_MPS tx_window_size: int = DEFAULT_TX_WINDOW_SIZE max_retransmission: int = DEFAULT_MAX_RETRANSMISSION retransmission_timeout: float = DEFAULT_RETRANSMISSION_TIMEOUT monitor_timeout: float = DEFAULT_MONITOR_TIMEOUT mode: TransmissionMode = TransmissionMode.BASIC fcs_enabled: bool = False @dataclasses.dataclass class LeCreditBasedChannelSpec: psm: int | None = None mtu: int = L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MTU mps: int = L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MPS max_credits: int = L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_INITIAL_CREDITS def __post_init__(self): if ( self.max_credits < 1 or self.max_credits > L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_CREDITS ): raise InvalidArgumentError('max credits out of range') if ( self.mtu < L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MTU or self.mtu > L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MTU ): raise InvalidArgumentError('MTU out of range') if ( self.mps < L2CAP_LE_CREDIT_BASED_CONNECTION_MIN_MPS or self.mps > L2CAP_LE_CREDIT_BASED_CONNECTION_MAX_MPS ): raise InvalidArgumentError('MPS out of range') class L2CAP_PDU: ''' See Bluetooth spec @ Vol 3, Part A - 3 DATA PACKET FORMAT ''' @classmethod def from_bytes(cls, data: bytes) -> L2CAP_PDU: # Check parameters if len(data) < 4: raise InvalidPacketError('not enough data for L2CAP header') length, l2cap_pdu_cid = struct.unpack_from(' bytes: return self.to_bytes(with_fcs=False) def to_bytes(self, with_fcs: bool = False) -> bytes: length = len(self.payload) if with_fcs: length += 2 header = struct.pack(' None: self.cid = cid self.payload = payload def __str__(self) -> str: return f'{color("L2CAP", "green")} [CID={self.cid}]: {self.payload.hex()}' class ControlField: ''' See Bluetooth spec @ Vol 3, Part A - 3.3.2 Control field. ''' class FieldType(utils.OpenIntEnum): I_FRAME = 0x00 S_FRAME = 0x01 class SegmentationAndReassembly(utils.OpenIntEnum): UNSEGMENTED = 0x00 START = 0x01 END = 0x02 CONTINUATION = 0x03 class SupervisoryFunction(utils.OpenIntEnum): # Receiver Ready RR = 0 # Reject REJ = 1 # Receiver Not Ready RNR = 2 # Select Reject SREJ = 3 class RetransmissionBit(utils.OpenIntEnum): NORMAL = 0x00 RETRANSMISSION = 0x01 req_seq: int frame_type: ClassVar[FieldType] def __bytes__(self) -> bytes: raise NotImplementedError() class EnhancedControlField(ControlField): """Base control field used in Enhanced Retransmission and Streaming Mode.""" final: int @classmethod def from_bytes(cls, data: bytes) -> EnhancedControlField: frame_type = data[0] & 0x01 if frame_type == cls.FieldType.I_FRAME: return InformationEnhancedControlField.from_bytes(data) elif frame_type == cls.FieldType.S_FRAME: return SupervisoryEnhancedControlField.from_bytes(data) else: raise InvalidArgumentError(f'Invalid frame type: {frame_type}') @dataclasses.dataclass class InformationEnhancedControlField(EnhancedControlField): tx_seq: int sar: int req_seq: int = 0 final: int = 1 frame_type = EnhancedControlField.FieldType.I_FRAME @classmethod def from_bytes(cls, data: bytes) -> EnhancedControlField: return cls( tx_seq=(data[0] >> 1) & 0b0111111, final=(data[0] >> 7) & 0b1, req_seq=(data[1] & 0b00111111), sar=(data[1] >> 6) & 0b11, ) def __bytes__(self) -> bytes: return bytes( [ self.frame_type | (self.tx_seq << 1) | (self.final << 7), self.req_seq | (self.sar << 6), ] ) @dataclasses.dataclass class SupervisoryEnhancedControlField(EnhancedControlField): supervision_function: int = ControlField.SupervisoryFunction.RR poll: int = 0 req_seq: int = 0 final: int = 0 frame_type = EnhancedControlField.FieldType.S_FRAME @classmethod def from_bytes(cls, data: bytes) -> EnhancedControlField: return cls( supervision_function=(data[0] >> 2) & 0b11, poll=(data[0] >> 4) & 0b1, final=(data[0] >> 7) & 0b1, req_seq=(data[1] & 0b1111111), ) def __bytes__(self) -> bytes: return bytes( [ ( self.frame_type | (self.supervision_function << 2) | self.poll << 7 | (self.final << 7) ), self.req_seq, ] ) # ----------------------------------------------------------------------------- @dataclasses.dataclass class L2CAP_Control_Frame: ''' See Bluetooth spec @ Vol 3, Part A - 4 SIGNALING PACKET FORMATS ''' classes: ClassVar[dict[int, type[L2CAP_Control_Frame]]] = {} fields: ClassVar[hci.Fields] = () code: int = dataclasses.field(default=0, init=False) name: str = dataclasses.field(default='', init=False) _payload: bytes | None = dataclasses.field(default=None, init=False) identifier: int @classmethod def from_bytes(cls, pdu: bytes) -> L2CAP_Control_Frame: code, identifier, length = struct.unpack_from(" list[tuple[L2CAP_Configure_Request.ParameterType, bytes]]: options = [] while len(data) >= 2: value_type = data[0] length = data[1] value = data[2 : 2 + length] data = data[2 + length :] options.append((L2CAP_Configure_Request.ParameterType(value_type), value)) return options @staticmethod def encode_configuration_options(options: list[tuple[int, bytes]]) -> bytes: return b''.join( [bytes([option[0], len(option[1])]) + option[1] for option in options] ) _ControlFrame = TypeVar('_ControlFrame', bound='L2CAP_Control_Frame') @classmethod def subclass(cls, subclass: type[_ControlFrame]) -> type[_ControlFrame]: subclass.name = subclass.__name__.upper() subclass.code = CommandCode[subclass.name] subclass.fields = hci.HCI_Object.fields_from_dataclass(subclass) # Register a factory for this class L2CAP_Control_Frame.classes[subclass.code] = subclass return subclass @property def payload(self) -> bytes: if self._payload is None: self._payload = hci.HCI_Object.dict_to_bytes(self.__dict__, self.fields) return self._payload @payload.setter def payload(self, payload: bytes) -> None: self._payload = payload def __bytes__(self) -> bytes: return ( struct.pack(' str: result = f'{color(self.name, "yellow")} [ID={self.identifier}]' if fields := getattr(self, 'fields', None): result += ':\n' + hci.HCI_Object.format_fields(self.__dict__, fields, ' ') else: if len(self.payload) > 1: result += f': {self.payload.hex()}' return result # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Command_Reject(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.1 COMMAND REJECT ''' class Reason(hci.SpecableEnum): COMMAND_NOT_UNDERSTOOD = 0x0000 SIGNALING_MTU_EXCEEDED = 0x0001 INVALID_CID_IN_REQUEST = 0x0002 reason: int = dataclasses.field(metadata=Reason.type_metadata(2)) data: bytes = dataclasses.field(metadata=hci.metadata('*')) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Connection_Request(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.2 CONNECTION REQUEST ''' @staticmethod def parse_psm(data: bytes, offset: int = 0) -> tuple[int, int]: psm_length = 2 psm = data[offset] | data[offset + 1] << 8 # The PSM field extends until the first even octet (inclusive) while data[offset + psm_length - 1] % 2 == 1: psm |= data[offset + psm_length] << (8 * psm_length) psm_length += 1 return offset + psm_length, psm @staticmethod def serialize_psm(psm: int) -> bytes: serialized = struct.pack('>= 16 while psm: serialized += bytes([psm & 0xFF]) psm >>= 8 return serialized psm: int = dataclasses.field( metadata=hci.metadata( { 'parser': lambda data, offset: L2CAP_Connection_Request.parse_psm( data, offset ), 'serializer': lambda value: L2CAP_Connection_Request.serialize_psm( value ), } ) ) source_cid: int = dataclasses.field(metadata=hci.metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Connection_Response(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.3 CONNECTION RESPONSE ''' class Result(hci.SpecableEnum): CONNECTION_SUCCESSFUL = 0x0000 CONNECTION_PENDING = 0x0001 CONNECTION_REFUSED_PSM_NOT_SUPPORTED = 0x0002 CONNECTION_REFUSED_SECURITY_BLOCK = 0x0003 CONNECTION_REFUSED_NO_RESOURCES_AVAILABLE = 0x0004 CONNECTION_REFUSED_INVALID_SOURCE_CID = 0x0006 CONNECTION_REFUSED_SOURCE_CID_ALREADY_ALLOCATED = 0x0007 CONNECTION_REFUSED_UNACCEPTABLE_PARAMETERS = 0x000B destination_cid: int = dataclasses.field(metadata=hci.metadata(2)) source_cid: int = dataclasses.field(metadata=hci.metadata(2)) result: int = dataclasses.field(metadata=Result.type_metadata(2)) status: int = dataclasses.field(metadata=hci.metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Configure_Request(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.4 CONFIGURATION REQUEST ''' class ParameterType(utils.OpenIntEnum): MTU = 0x01 FLUSH_TIMEOUT = 0x02 QOS = 0x03 RETRANSMISSION_AND_FLOW_CONTROL = 0x04 FCS = 0x05 EXTENDED_FLOW_SPEC = 0x06 EXTENDED_WINDOW_SIZE = 0x07 destination_cid: int = dataclasses.field(metadata=hci.metadata(2)) flags: int = dataclasses.field(metadata=hci.metadata(2)) options: bytes = dataclasses.field(metadata=hci.metadata('*')) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Configure_Response(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.5 CONFIGURATION RESPONSE ''' class Result(hci.SpecableEnum): SUCCESS = 0x0000 FAILURE_UNACCEPTABLE_PARAMETERS = 0x0001 FAILURE_REJECTED = 0x0002 FAILURE_UNKNOWN_OPTIONS = 0x0003 PENDING = 0x0004 FAILURE_FLOW_SPEC_REJECTED = 0x0005 source_cid: int = dataclasses.field(metadata=hci.metadata(2)) flags: int = dataclasses.field(metadata=hci.metadata(2)) result: int = dataclasses.field(metadata=Result.type_metadata(2)) options: bytes = dataclasses.field(metadata=hci.metadata('*')) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Disconnection_Request(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.6 DISCONNECTION REQUEST ''' destination_cid: int = dataclasses.field(metadata=hci.metadata(2)) source_cid: int = dataclasses.field(metadata=hci.metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Disconnection_Response(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.7 DISCONNECTION RESPONSE ''' destination_cid: int = dataclasses.field(metadata=hci.metadata(2)) source_cid: int = dataclasses.field(metadata=hci.metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Echo_Request(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.8 ECHO REQUEST ''' data: bytes = dataclasses.field(metadata=hci.metadata('*')) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Echo_Response(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.9 ECHO RESPONSE ''' data: bytes = dataclasses.field(metadata=hci.metadata('*')) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Information_Request(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.10 INFORMATION REQUEST ''' class InfoType(hci.SpecableEnum): CONNECTIONLESS_MTU = 0x0001 EXTENDED_FEATURES_SUPPORTED = 0x0002 FIXED_CHANNELS_SUPPORTED = 0x0003 class ExtendedFeatures(hci.SpecableFlag): FLOW_MODE_CONTROL = 0x0001 RETRANSMISSION_MODE = 0x0002 BIDIRECTIONAL_QOS = 0x0004 ENHANCED_RETRANSMISSION_MODE = 0x0008 STREAMING_MODE = 0x0010 FCS_OPTION = 0x0020 EXTENDED_FLOW_SPEC = 0x0040 FIXED_CHANNELS = 0x0080 EXTENDED_WINDOW_SIZE = 0x0100 UNICAST_CONNECTIONLESS_DATA = 0x0200 ENHANCED_CREDIT_BASE_FLOW_CONTROL = 0x0400 info_type: int = dataclasses.field(metadata=InfoType.type_metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Information_Response(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.11 INFORMATION RESPONSE ''' class Result(hci.SpecableEnum): SUCCESS = 0x00 NOT_SUPPORTED = 0x01 info_type: int = dataclasses.field( metadata=L2CAP_Information_Request.InfoType.type_metadata(2) ) result: int = dataclasses.field(metadata=Result.type_metadata(2)) data: bytes = dataclasses.field(metadata=hci.metadata('*')) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Connection_Parameter_Update_Request(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.20 CONNECTION PARAMETER UPDATE REQUEST ''' interval_min: int = dataclasses.field(metadata=hci.metadata(2)) interval_max: int = dataclasses.field(metadata=hci.metadata(2)) latency: int = dataclasses.field(metadata=hci.metadata(2)) timeout: int = dataclasses.field(metadata=hci.metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Connection_Parameter_Update_Response(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.21 CONNECTION PARAMETER UPDATE RESPONSE ''' result: int = dataclasses.field(metadata=hci.metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_LE_Credit_Based_Connection_Request(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.22 LE CREDIT BASED CONNECTION REQUEST (CODE 0x14) ''' le_psm: int = dataclasses.field(metadata=hci.metadata(2)) source_cid: int = dataclasses.field(metadata=hci.metadata(2)) mtu: int = dataclasses.field(metadata=hci.metadata(2)) mps: int = dataclasses.field(metadata=hci.metadata(2)) initial_credits: int = dataclasses.field(metadata=hci.metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_LE_Credit_Based_Connection_Response(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.23 LE CREDIT BASED CONNECTION RESPONSE (CODE 0x15) ''' class Result(hci.SpecableEnum): CONNECTION_SUCCESSFUL = 0x0000 CONNECTION_REFUSED_LE_PSM_NOT_SUPPORTED = 0x0002 CONNECTION_REFUSED_NO_RESOURCES_AVAILABLE = 0x0004 CONNECTION_REFUSED_INSUFFICIENT_AUTHENTICATION = 0x0005 CONNECTION_REFUSED_INSUFFICIENT_AUTHORIZATION = 0x0006 CONNECTION_REFUSED_INSUFFICIENT_ENCRYPTION_KEY_SIZE = 0x0007 CONNECTION_REFUSED_INSUFFICIENT_ENCRYPTION = 0x0008 CONNECTION_REFUSED_INVALID_SOURCE_CID = 0x0009 CONNECTION_REFUSED_SOURCE_CID_ALREADY_ALLOCATED = 0x000A CONNECTION_REFUSED_UNACCEPTABLE_PARAMETERS = 0x000B destination_cid: int = dataclasses.field(metadata=hci.metadata(2)) mtu: int = dataclasses.field(metadata=hci.metadata(2)) mps: int = dataclasses.field(metadata=hci.metadata(2)) initial_credits: int = dataclasses.field(metadata=hci.metadata(2)) result: int = dataclasses.field(metadata=Result.type_metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_LE_Flow_Control_Credit(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.24 LE FLOW CONTROL CREDIT (CODE 0x16) ''' cid: int = dataclasses.field(metadata=hci.metadata(2)) credits: int = dataclasses.field(metadata=hci.metadata(2)) # ----------------------------------------------------------------------------- @L2CAP_Control_Frame.subclass @dataclasses.dataclass class L2CAP_Credit_Based_Connection_Request(L2CAP_Control_Frame): ''' See Bluetooth spec @ Vol 3, Part A - 4.25 L2CAP_CREDIT_BASED_CONNECTION_REQ (0x17). ''' @classmethod def parse_cid_list(cls, data: bytes, offset: int) -> tuple[int, list[int]]: count = (len(data) - offset) // 2 return len(data), list(struct.unpack_from("<" + ("H" * count), data, offset)) @classmethod def serialize_cid_list(cls, cids: Sequence[int]) -> bytes: return b"".join([struct.pack(" None: self.channel = channel def send_sdu(self, sdu: bytes) -> None: self.channel.send_pdu(sdu) def on_pdu(self, pdu: bytes) -> None: self.channel.on_sdu(pdu) # TODO: Handle retransmission class EnhancedRetransmissionProcessor(Processor): MAX_SEQ_NUM = 64 @dataclasses.dataclass class _PendingPdu: payload: bytes tx_seq: int sar: InformationEnhancedControlField.SegmentationAndReassembly sdu_length: int = 0 req_seq: int = 0 def __bytes__(self) -> bytes: return ( bytes( InformationEnhancedControlField( tx_seq=self.tx_seq, req_seq=self.req_seq, sar=self.sar, ) ) + ( struct.pack(' None: if ( self.peer_max_retransmission <= 0 or self._num_receiver_ready_polls_sent < self.peer_max_retransmission ): self._send_receiver_ready_poll() self._start_monitor() else: logger.error("Max retransmission exceeded") def _receiver_ready_poll(self) -> None: self._send_receiver_ready_poll() self._start_monitor() def _start_monitor(self) -> None: if self._monitor_handle: self._monitor_handle.cancel() self._monitor_handle = asyncio.get_running_loop().call_later( self.monitor_timeout, self._monitor ) def _start_receiver_ready_poll(self) -> None: if self._receiver_ready_poll_handle: self._receiver_ready_poll_handle.cancel() self._num_receiver_ready_polls_sent = 0 self._receiver_ready_poll_handle = asyncio.get_running_loop().call_later( self.retransmission_timeout, self._receiver_ready_poll ) def _send_receiver_ready_poll(self) -> None: self._num_receiver_ready_polls_sent += 1 self._send_s_frame( supervision_function=SupervisoryEnhancedControlField.SupervisoryFunction.RR, final=1, ) def _get_next_tx_seq(self) -> int: seq_num = self._next_tx_seq self._next_tx_seq = (self._next_tx_seq + 1) % self.MAX_SEQ_NUM return seq_num @override def send_sdu(self, sdu: bytes) -> None: if len(sdu) <= self.peer_mps: pdu = self._PendingPdu( payload=sdu, tx_seq=self._get_next_tx_seq(), req_seq=self._req_seq_num, sar=InformationEnhancedControlField.SegmentationAndReassembly.UNSEGMENTED, ) self._pending_pdus.append(pdu) else: for offset in range(0, len(sdu), self.peer_mps): payload = sdu[offset : offset + self.peer_mps] if offset == 0: sar = ( InformationEnhancedControlField.SegmentationAndReassembly.START ) elif offset + len(payload) >= len(sdu): sar = InformationEnhancedControlField.SegmentationAndReassembly.END else: sar = ( InformationEnhancedControlField.SegmentationAndReassembly.CONTINUATION ) pdu = self._PendingPdu( payload=payload, tx_seq=self._get_next_tx_seq(), req_seq=self._req_seq_num, sar=sar, sdu_length=len(sdu), ) self._pending_pdus.append(pdu) self._process_output() @override def on_pdu(self, pdu: bytes) -> None: control_field = EnhancedControlField.from_bytes(pdu) self._update_ack_seq(control_field.req_seq, control_field.final != 0) if isinstance(control_field, InformationEnhancedControlField): if control_field.tx_seq != self._req_seq_num: logger.error( "tx_seq != self._req_seq_num, tx_seq: %d, self._req_seq_num: %d", control_field.tx_seq, self._req_seq_num, ) return self._req_seq_num = (control_field.tx_seq + 1) % self.MAX_SEQ_NUM if ( control_field.sar == InformationEnhancedControlField.SegmentationAndReassembly.START ): # Drop Control Field(2) + SDU Length(2) self._in_sdu += pdu[4:] else: # Drop Control Field(2) self._in_sdu += pdu[2:] if control_field.sar in ( InformationEnhancedControlField.SegmentationAndReassembly.END, InformationEnhancedControlField.SegmentationAndReassembly.UNSEGMENTED, ): self.channel.on_sdu(self._in_sdu) self._in_sdu = b'' # If sink doesn't trigger any I-frame, ack this frame. if self._req_seq_num != self._last_acked_rx_seq: self._send_s_frame( supervision_function=SupervisoryEnhancedControlField.SupervisoryFunction.RR, final=0, ) elif isinstance(control_field, SupervisoryEnhancedControlField): self._remote_is_busy = ( control_field.supervision_function == SupervisoryEnhancedControlField.SupervisoryFunction.RNR ) if control_field.supervision_function in ( SupervisoryEnhancedControlField.SupervisoryFunction.RR, SupervisoryEnhancedControlField.SupervisoryFunction.RNR, ): if control_field.poll: self._send_s_frame( supervision_function=SupervisoryEnhancedControlField.SupervisoryFunction.RR, final=1, ) else: # TODO: Handle Retransmission. pass def _process_output(self) -> None: if self._remote_is_busy: logger.debug("Remote is busy") return if self._monitor_handle: logger.debug("Monitor handle is not None") return pdu_to_send = self.peer_tx_window_size - len(self._tx_window) for pdu in itertools.islice(self._pending_pdus, pdu_to_send): self._send_i_frame(pdu) self._pending_pdus = self._pending_pdus[pdu_to_send:] def _send_i_frame(self, pdu: _PendingPdu) -> None: pdu.req_seq = self._req_seq_num self._start_receiver_ready_poll() self._tx_window.append(pdu) self.channel.send_pdu(bytes(pdu)) self._last_acked_rx_seq = self._req_seq_num def _send_s_frame( self, supervision_function: SupervisoryEnhancedControlField.SupervisoryFunction, final: int, ) -> None: self.channel.send_pdu( SupervisoryEnhancedControlField( supervision_function=supervision_function, final=final, req_seq=self._req_seq_num, ) ) self._last_acked_rx_seq = self._req_seq_num def _update_ack_seq(self, new_seq: int, is_poll_response: bool) -> None: num_frames_acked = (new_seq - self._last_acked_tx_seq) % self.MAX_SEQ_NUM if num_frames_acked > len(self._tx_window): logger.error( "Received acknowledgment for %d frames but only %d frames are pending", num_frames_acked, len(self._tx_window), ) return if is_poll_response and self._monitor_handle: self._monitor_handle.cancel() self._monitor_handle = None del self._tx_window[:num_frames_acked] self._last_acked_tx_seq = new_seq if ( self._last_acked_tx_seq == self._next_tx_seq and self._receiver_ready_poll_handle ): self._receiver_ready_poll_handle.cancel() self._receiver_ready_poll_handle = None self._process_output() # ----------------------------------------------------------------------------- class ClassicChannel(utils.EventEmitter): class State(enum.IntEnum): # States CLOSED = 0x00 WAIT_CONNECT = 0x01 WAIT_CONNECT_RSP = 0x02 OPEN = 0x03 WAIT_DISCONNECT = 0x04 WAIT_CREATE = 0x05 WAIT_CREATE_RSP = 0x06 WAIT_MOVE = 0x07 WAIT_MOVE_RSP = 0x08 WAIT_MOVE_CONFIRM = 0x09 WAIT_CONFIRM_RSP = 0x0A # CONFIG substates WAIT_CONFIG = 0x10 WAIT_SEND_CONFIG = 0x11 WAIT_CONFIG_REQ_RSP = 0x12 WAIT_CONFIG_RSP = 0x13 WAIT_CONFIG_REQ = 0x14 WAIT_IND_FINAL_RSP = 0x15 WAIT_FINAL_RSP = 0x16 WAIT_CONTROL_IND = 0x17 EVENT_OPEN = "open" EVENT_CLOSE = "close" connection_result: asyncio.Future[None] | None disconnection_result: asyncio.Future[None] | None response: asyncio.Future[bytes] | None sink: Callable[[bytes], Any] | None state: State connection: Connection mtu: int peer_mtu: int processor: Processor def __init__( self, manager: ChannelManager, connection: Connection, signaling_cid: int, psm: int, source_cid: int, spec: ClassicChannelSpec, ) -> None: super().__init__() self.manager = manager self.connection = connection self.signaling_cid = signaling_cid self.state = self.State.CLOSED self.mtu = spec.mtu self.peer_mtu = L2CAP_MIN_BR_EDR_MTU self.psm = psm self.source_cid = source_cid self.destination_cid = 0 self.connection_result = None self.disconnection_result = None self.sink = None self.fcs_enabled = spec.fcs_enabled self.spec = spec self.mode = spec.mode # Configure mode-specific processor later on configure request. self.processor = Processor(self) if self.mode not in ( TransmissionMode.BASIC, TransmissionMode.ENHANCED_RETRANSMISSION, ): raise InvalidArgumentError(f"Mode {spec.mode} is not supported") def _change_state(self, new_state: State) -> None: logger.debug(f'{self} state change -> {color(new_state.name, "cyan")}') self.state = new_state def write(self, sdu: bytes) -> None: self.processor.send_sdu(sdu) def send_pdu(self, pdu: SupportsBytes | bytes) -> None: if self.state != self.State.OPEN: raise InvalidStateError('channel not open') self.manager.send_pdu( self.connection, self.destination_cid, pdu, self.fcs_enabled ) def send_control_frame(self, frame: L2CAP_Control_Frame) -> None: self.manager.send_control_frame(self.connection, self.signaling_cid, frame) def on_pdu(self, pdu: bytes) -> None: if self.fcs_enabled: # Drop FCS. pdu = pdu[:-2] self.processor.on_pdu(pdu) def on_sdu(self, sdu: bytes) -> None: if self.sink: # pylint: disable=not-callable self.sink(sdu) else: logger.warning( color('received sdu without a pending request or sink', 'red') ) async def connect(self) -> None: if self.state != self.State.CLOSED: raise InvalidStateError('invalid state') # Check that we can start a new connection if self.connection_result: raise InvalidStateError('connection already pending') self._change_state(self.State.WAIT_CONNECT_RSP) self.send_control_frame( L2CAP_Connection_Request( identifier=self.manager.next_identifier(self.connection), psm=self.psm, source_cid=self.source_cid, ) ) # Create a future to wait for the state machine to get to a success or error # state self.connection_result = asyncio.get_running_loop().create_future() # Wait for the connection to succeed or fail try: return await self.connection.cancel_on_disconnection(self.connection_result) finally: self.connection_result = None def _disconnect_sync(self) -> None: """For internal sync disconnection.""" self._change_state(self.State.WAIT_DISCONNECT) self.send_control_frame( L2CAP_Disconnection_Request( identifier=self.manager.next_identifier(self.connection), destination_cid=self.destination_cid, source_cid=self.source_cid, ) ) # Create a future to wait for the state machine to get to a success or error # state self.disconnection_result = asyncio.get_running_loop().create_future() def _abort_connection_result(self, message: str = 'Connection failure') -> None: # Cancel pending connection result. if self.connection_result and not self.connection_result.done(): self.connection_result.set_exception( L2capError(error_code=0, error_name=message) ) async def disconnect(self) -> None: if self.state != self.State.OPEN: raise InvalidStateError('invalid state') self._disconnect_sync() if self.disconnection_result: return await self.disconnection_result def abort(self) -> None: if self.state == self.State.OPEN: self._change_state(self.State.CLOSED) self.emit(self.EVENT_CLOSE) def send_configure_request(self) -> None: options: list[tuple[int, bytes]] = [ ( L2CAP_Configure_Request.ParameterType.MTU, struct.pack(' None: self.destination_cid = request.source_cid self._change_state(self.State.WAIT_CONNECT) self.send_control_frame( L2CAP_Connection_Response( identifier=request.identifier, destination_cid=self.source_cid, source_cid=self.destination_cid, result=L2CAP_Connection_Response.Result.CONNECTION_SUCCESSFUL, status=0x0000, ) ) self._change_state(self.State.WAIT_CONFIG) self.send_configure_request() self._change_state(self.State.WAIT_CONFIG_REQ_RSP) def on_connection_response(self, response: L2CAP_Connection_Response): if self.state != self.State.WAIT_CONNECT_RSP: logger.warning(color('invalid state', 'red')) return if response.result == L2CAP_Connection_Response.Result.CONNECTION_SUCCESSFUL: self.destination_cid = response.destination_cid self._change_state(self.State.WAIT_CONFIG) self.send_configure_request() self._change_state(self.State.WAIT_CONFIG_REQ_RSP) elif response.result == L2CAP_Connection_Response.Result.CONNECTION_PENDING: pass else: self._change_state(self.State.CLOSED) if self.connection_result: self.connection_result.set_exception( L2capError( response.result, L2CAP_Connection_Response.Result(response.result).name, ) ) self.connection_result = None def on_configure_request(self, request: L2CAP_Configure_Request) -> None: if self.state not in ( self.State.WAIT_CONFIG, self.State.WAIT_CONFIG_REQ, self.State.WAIT_CONFIG_REQ_RSP, ): logger.warning(color('invalid state', 'red')) return # Decode the options options = L2CAP_Control_Frame.decode_configuration_options(request.options) # Result to options replied_options = list[tuple[int, bytes]]() result = L2CAP_Configure_Response.Result.SUCCESS new_mode = TransmissionMode.BASIC for option in options: match option[0]: case L2CAP_Configure_Request.ParameterType.MTU: self.peer_mtu = struct.unpack(' None: if response.result == L2CAP_Configure_Response.Result.SUCCESS: if self.state == self.State.WAIT_CONFIG_REQ_RSP: self._change_state(self.State.WAIT_CONFIG_REQ) elif self.state in ( self.State.WAIT_CONFIG_RSP, self.State.WAIT_CONTROL_IND, ): self._change_state(self.State.OPEN) if self.connection_result: self.connection_result.set_result(None) self.connection_result = None self.emit(self.EVENT_OPEN) else: logger.warning(color('invalid state', 'red')) elif ( response.result == L2CAP_Configure_Response.Result.FAILURE_UNACCEPTABLE_PARAMETERS ): # Re-configure with what's suggested in the response self.send_control_frame( L2CAP_Configure_Request( identifier=self.manager.next_identifier(self.connection), destination_cid=self.destination_cid, flags=0x0000, options=response.options, ) ) else: logger.warning( color( '!!! configuration rejected: ' f'{L2CAP_Configure_Response.Result(response.result).name}', 'red', ) ) # TODO: decide how to fail gracefully def on_disconnection_request(self, request: L2CAP_Disconnection_Request) -> None: self.send_control_frame( L2CAP_Disconnection_Response( identifier=request.identifier, destination_cid=request.destination_cid, source_cid=request.source_cid, ) ) self._abort_connection_result() self._change_state(self.State.CLOSED) self.emit(self.EVENT_CLOSE) self.manager.on_channel_closed(self) def on_disconnection_response(self, response: L2CAP_Disconnection_Response) -> None: if ( response.destination_cid != self.destination_cid or response.source_cid != self.source_cid ): logger.warning('unexpected source or destination CID') return self._change_state(self.State.CLOSED) if self.disconnection_result: self.disconnection_result.set_result(None) self.disconnection_result = None self.emit(self.EVENT_CLOSE) self.manager.on_channel_closed(self) def __str__(self) -> str: return ( f'Channel({self.source_cid}->{self.destination_cid}, ' f'PSM={self.psm}, ' f'MTU={self.mtu}/{self.peer_mtu}, ' f'state={self.state.name})' ) # ----------------------------------------------------------------------------- class LeCreditBasedChannel(utils.EventEmitter): """ LE Credit-based Connection Oriented Channel """ class State(enum.IntEnum): INIT = 0 CONNECTED = 1 CONNECTING = 2 DISCONNECTING = 3 DISCONNECTED = 4 CONNECTION_ERROR = 5 out_queue: deque[bytes] connection_result: asyncio.Future[LeCreditBasedChannel] | None disconnection_result: asyncio.Future[None] | None in_sdu: bytes | None out_sdu: bytes | None state: State connection: Connection sink: Callable[[bytes], Any] | None EVENT_OPEN = "open" EVENT_CLOSE = "close" EVENT_ATT_MTU_UPDATE = "att_mtu_update" def __init__( self, manager: ChannelManager, connection: Connection, psm: int, source_cid: int, destination_cid: int, mtu: int, mps: int, credits: int, # pylint: disable=redefined-builtin peer_mtu: int, peer_mps: int, peer_credits: int, connected: bool, ) -> None: super().__init__() self.manager = manager self.connection = connection self.psm = psm self.source_cid = source_cid self.destination_cid = destination_cid self.mtu = mtu self.mps = mps self.credits = credits self.peer_mtu = peer_mtu self.peer_mps = peer_mps self.peer_credits = peer_credits self.peer_max_credits = self.peer_credits self.peer_credits_threshold = self.peer_max_credits // 2 self.in_sdu = None self.in_sdu_length = 0 self.out_queue = deque() self.out_sdu = None self.sink = None self.connected = False self.connection_result = None self.disconnection_result = None self.drained = asyncio.Event() # Core Specification Vol 3, Part G, 5.3.1 ATT_MTU # ATT_MTU shall be set to the minimum of the MTU field values of the two devices. self.att_mtu = min(mtu, peer_mtu) self.drained.set() if connected: self.state = self.State.CONNECTED else: self.state = self.State.INIT def _change_state(self, new_state: State) -> None: logger.debug(f'{self} state change -> {color(new_state.name, "cyan")}') self.state = new_state if new_state == self.State.CONNECTED: self.emit(self.EVENT_OPEN) elif new_state == self.State.DISCONNECTED: self.emit(self.EVENT_CLOSE) def send_pdu(self, pdu: SupportsBytes | bytes) -> None: self.manager.send_pdu(self.connection, self.destination_cid, pdu) def send_control_frame(self, frame: L2CAP_Control_Frame) -> None: self.manager.send_control_frame(self.connection, L2CAP_LE_SIGNALING_CID, frame) async def connect(self) -> LeCreditBasedChannel: # Check that we're in the right state if self.state != self.State.INIT: raise InvalidStateError('not in a connectable state') # Check that we can start a new connection identifier = self.manager.next_identifier(self.connection) if identifier in self.manager.le_coc_requests: raise InvalidStateError('too many concurrent connection requests') self._change_state(self.State.CONNECTING) request = L2CAP_LE_Credit_Based_Connection_Request( identifier=identifier, le_psm=self.psm, source_cid=self.source_cid, mtu=self.mtu, mps=self.mps, initial_credits=self.peer_credits, ) self.manager.le_coc_requests[identifier] = request self.send_control_frame(request) # Create a future to wait for the response self.connection_result = asyncio.get_running_loop().create_future() # Wait for the connection to succeed or fail return await self.connection_result async def disconnect(self) -> None: # Check that we're connected if self.state != self.State.CONNECTED: raise InvalidStateError('not connected') self._change_state(self.State.DISCONNECTING) self.flush_output() self.send_control_frame( L2CAP_Disconnection_Request( identifier=self.manager.next_identifier(self.connection), destination_cid=self.destination_cid, source_cid=self.source_cid, ) ) # Create a future to wait for the state machine to get to a success or error # state self.disconnection_result = asyncio.get_running_loop().create_future() return await self.disconnection_result def abort(self) -> None: if self.state == self.State.CONNECTED: self._change_state(self.State.DISCONNECTED) if self.state == self.State.CONNECTING: if self.connection_result is not None: self.connection_result.cancel() def on_pdu(self, pdu: bytes) -> None: if self.sink is None: logger.warning('received pdu without a sink') return if self.state != self.State.CONNECTED: logger.warning('received PDU while not connected, dropping') # Manage the peer credits if self.peer_credits == 0: logger.warning('received LE frame when peer out of credits') else: self.peer_credits -= 1 if self.peer_credits <= self.peer_credits_threshold: # The credits fell below the threshold, replenish them to the max self.send_control_frame( L2CAP_LE_Flow_Control_Credit( identifier=self.manager.next_identifier(self.connection), cid=self.source_cid, credits=self.peer_max_credits - self.peer_credits, ) ) self.peer_credits = self.peer_max_credits # Check if this starts a new SDU if self.in_sdu is None: # Start a new SDU self.in_sdu = pdu else: # Continue an SDU self.in_sdu += pdu # Check if the SDU is complete if self.in_sdu_length == 0: # We don't know the size yet, check if we have received the header to # compute it if len(self.in_sdu) >= 2: self.in_sdu_length = struct.unpack_from(' None: # Look for a matching pending response result if self.connection_result is None: logger.warning( f'received unexpected connection response (id={response.identifier})' ) return if ( response.result == L2CAP_LE_Credit_Based_Connection_Response.Result.CONNECTION_SUCCESSFUL ): self.destination_cid = response.destination_cid self.peer_mtu = response.mtu self.peer_mps = response.mps self.credits = response.initial_credits self.connected = True self.connection_result.set_result(self) self._change_state(self.State.CONNECTED) else: self.connection_result.set_exception( L2capError( response.result, L2CAP_LE_Credit_Based_Connection_Response.Result( response.result ).name, ) ) self._change_state(self.State.CONNECTION_ERROR) # Cleanup self.connection_result = None def on_enhanced_connection_response( self, destination_cid: int, response: L2CAP_Credit_Based_Connection_Response ) -> None: if ( response.result == L2CAP_Credit_Based_Connection_Response.Result.ALL_CONNECTIONS_SUCCESSFUL ): self.destination_cid = destination_cid self.peer_mtu = response.mtu self.peer_mps = response.mps self.credits = response.initial_credits self.connected = True self._change_state(self.State.CONNECTED) else: self._change_state(self.State.CONNECTION_ERROR) def on_credits(self, credits: int) -> None: # pylint: disable=redefined-builtin self.credits += credits logger.debug(f'received {credits} credits, total = {self.credits}') # Try to send more data if we have any queued up self.process_output() def on_disconnection_request(self, request: L2CAP_Disconnection_Request) -> None: self.send_control_frame( L2CAP_Disconnection_Response( identifier=request.identifier, destination_cid=request.destination_cid, source_cid=request.source_cid, ) ) self._change_state(self.State.DISCONNECTED) self.flush_output() def on_disconnection_response(self, response: L2CAP_Disconnection_Response) -> None: if self.state != self.State.DISCONNECTING: logger.warning(color('invalid state', 'red')) return if ( response.destination_cid != self.destination_cid or response.source_cid != self.source_cid ): logger.warning('unexpected source or destination CID') return self._change_state(self.State.DISCONNECTED) if self.disconnection_result: self.disconnection_result.set_result(None) self.disconnection_result = None def on_att_mtu_update(self, mtu: int) -> None: self.att_mtu = mtu self.emit(self.EVENT_ATT_MTU_UPDATE, mtu) def flush_output(self) -> None: self.out_queue.clear() self.out_sdu = None def process_output(self) -> None: while self.credits > 0: if self.out_sdu is not None: # Finish the current SDU packet = self.out_sdu[: self.peer_mps] self.send_pdu(packet) self.credits -= 1 logger.debug(f'sent {len(packet)} bytes, {self.credits} credits left') if len(packet) == len(self.out_sdu): # We sent everything self.out_sdu = None else: # Keep what's still left to send self.out_sdu = self.out_sdu[len(packet) :] continue if self.out_queue: # Create the next SDU (2 bytes header plus up to MTU bytes payload) logger.debug( f'assembling SDU from {len(self.out_queue)} packets in output queue' ) payload = b'' while self.out_queue and len(payload) < self.peer_mtu: # We can add more data to the payload chunk = self.out_queue[0][: self.peer_mtu - len(payload)] payload += chunk self.out_queue[0] = self.out_queue[0][len(chunk) :] if len(self.out_queue[0]) == 0: # We consumed the entire buffer, remove it self.out_queue.popleft() logger.debug( f'packet completed, {len(self.out_queue)} left in queue' ) # Construct the SDU with its header assert len(payload) != 0 logger.debug(f'SDU complete: {len(payload)} payload bytes') self.out_sdu = struct.pack(' None: if self.state != self.State.CONNECTED: logger.warning('not connected, dropping data') return # Queue the data self.out_queue.append(data) self.drained.clear() logger.debug( f'{len(data)} bytes packet queued, {len(self.out_queue)} packets in queue' ) # Send what we can self.process_output() async def drain(self) -> None: await self.drained.wait() def pause_reading(self) -> None: # TODO: not implemented yet pass def resume_reading(self) -> None: # TODO: not implemented yet pass def __str__(self) -> str: return ( f'CoC({self.source_cid}->{self.destination_cid}, ' f'State={self.state.name}, ' f'PSM={self.psm}, ' f'MTU={self.mtu}/{self.peer_mtu}, ' f'MPS={self.mps}/{self.peer_mps}, ' f'credits={self.credits}/{self.peer_credits})' ) # ----------------------------------------------------------------------------- class ClassicChannelServer(utils.EventEmitter): EVENT_CONNECTION = "connection" def __init__( self, manager: ChannelManager, psm: int, handler: Callable[[ClassicChannel], Any] | None, spec: ClassicChannelSpec, ) -> None: super().__init__() self.manager = manager self.handler = handler self.psm = psm self.spec = spec def on_connection(self, channel: ClassicChannel) -> None: self.emit(self.EVENT_CONNECTION, channel) if self.handler: self.handler(channel) def close(self) -> None: if self.psm in self.manager.servers: del self.manager.servers[self.psm] # ----------------------------------------------------------------------------- class LeCreditBasedChannelServer(utils.EventEmitter): EVENT_CONNECTION = "connection" def __init__( self, manager: ChannelManager, psm: int, handler: Callable[[LeCreditBasedChannel], Any] | None, max_credits: int, mtu: int, mps: int, ) -> None: super().__init__() self.manager = manager self.handler = handler self.psm = psm self.max_credits = max_credits self.mtu = mtu self.mps = mps def on_connection(self, channel: LeCreditBasedChannel) -> None: self.emit(self.EVENT_CONNECTION, channel) if self.handler: self.handler(channel) def close(self) -> None: if self.psm in self.manager.le_coc_servers: del self.manager.le_coc_servers[self.psm] # ----------------------------------------------------------------------------- class ChannelManager: identifiers: dict[int, int] channels: dict[int, dict[int, ClassicChannel | LeCreditBasedChannel]] servers: dict[int, ClassicChannelServer] le_coc_channels: dict[int, dict[int, LeCreditBasedChannel]] le_coc_servers: dict[int, LeCreditBasedChannelServer] le_coc_requests: dict[int, L2CAP_LE_Credit_Based_Connection_Request] fixed_channels: dict[int, Callable[[int, bytes], Any] | None] pending_credit_based_connections: dict[ int, dict[ int, tuple[ asyncio.Future[None], list[LeCreditBasedChannel], ], ], ] _host: Host | None connection_parameters_update_response: asyncio.Future[int] | None def __init__( self, extended_features: Iterable[int] = (), connectionless_mtu: int = L2CAP_DEFAULT_CONNECTIONLESS_MTU, ) -> None: self._host = None self.identifiers = {} # Incrementing identifier values by connection self.channels = {} # All channels, mapped by connection and source cid self.fixed_channels = { # Fixed channel handlers, mapped by cid L2CAP_SIGNALING_CID: None, L2CAP_LE_SIGNALING_CID: None, } self.servers = {} # Servers accepting connections, by PSM self.le_coc_channels = ( {} ) # LE CoC channels, mapped by connection and destination cid self.le_coc_servers = {} # LE CoC - Servers accepting connections, by PSM self.le_coc_requests = {} # LE CoC connection requests, by identifier self.pending_credit_based_connections = ( {} ) # Credit-based connection request contexts, by connection handle and identifier self.extended_features = set(extended_features) self.connectionless_mtu = connectionless_mtu self.connection_parameters_update_response = None @property def host(self) -> Host: assert self._host return self._host @host.setter def host(self, host: Host) -> None: if self._host is not None: self._host.remove_listener('disconnection', self.on_disconnection) self._host = host if host is not None: host.on('disconnection', self.on_disconnection) def find_channel(self, connection_handle: int, cid: int): if connection_channels := self.channels.get(connection_handle): return connection_channels.get(cid) return None def find_le_coc_channel(self, connection_handle: int, cid: int): if connection_channels := self.le_coc_channels.get(connection_handle): return connection_channels.get(cid) return None @staticmethod def find_free_br_edr_cid(channels: Iterable[int]) -> int: # Pick the smallest valid CID that's not already in the list # (not necessarily the most efficient algorithm, but the list of CID is # very small in practice) for cid in range( L2CAP_ACL_U_DYNAMIC_CID_RANGE_START, L2CAP_ACL_U_DYNAMIC_CID_RANGE_END + 1 ): if cid not in channels: return cid raise OutOfResourcesError('no free CID available') @classmethod def find_free_le_cid(cls, channels: Iterable[int]) -> int | None: cids = cls.find_free_le_cids(channels, 1) return cids[0] if cids else None @classmethod def find_free_le_cids(cls, channels: Iterable[int], count: int) -> list[int]: # Pick the smallest valid CID that's not already in the list # (not necessarily the most efficient algorithm, but the list of CID is # very small in practice) cids: list[int] = [] for cid in range( L2CAP_LE_U_DYNAMIC_CID_RANGE_START, L2CAP_LE_U_DYNAMIC_CID_RANGE_END + 1 ): if cid not in channels: cids.append(cid) if len(cids) == count: return cids return [] def next_identifier(self, connection: Connection) -> int: identifier = (self.identifiers.setdefault(connection.handle, 0) + 1) % 256 # 0x00 is an invalid ID (BT Core Spec, Vol 3, Part A, Sect 4 if identifier == 0: identifier = 1 self.identifiers[connection.handle] = identifier return identifier def register_fixed_channel( self, cid: int, handler: Callable[[int, bytes], Any] ) -> None: self.fixed_channels[cid] = handler def deregister_fixed_channel(self, cid: int) -> None: if cid in self.fixed_channels: del self.fixed_channels[cid] def create_classic_server( self, spec: ClassicChannelSpec, handler: Callable[[ClassicChannel], Any] | None = None, ) -> ClassicChannelServer: if not spec.psm: # Find a free PSM for candidate in range( L2CAP_PSM_DYNAMIC_RANGE_START, L2CAP_PSM_DYNAMIC_RANGE_END + 1, 2 ): if (candidate >> 8) % 2 == 1: continue if candidate in self.servers: continue spec.psm = candidate break else: raise InvalidStateError('no free PSM') else: # Check that the PSM isn't already in use if spec.psm in self.servers: raise InvalidArgumentError('PSM already in use') # Check that the PSM is valid if spec.psm % 2 == 0: raise InvalidArgumentError('invalid PSM (not odd)') check = spec.psm >> 8 while check: if check % 2 != 0: raise InvalidArgumentError('invalid PSM') check >>= 8 self.servers[spec.psm] = ClassicChannelServer(self, spec.psm, handler, spec) return self.servers[spec.psm] def create_le_credit_based_server( self, spec: LeCreditBasedChannelSpec, handler: Callable[[LeCreditBasedChannel], Any] | None = None, ) -> LeCreditBasedChannelServer: if not spec.psm: # Find a free PSM for candidate in range( L2CAP_LE_PSM_DYNAMIC_RANGE_START, L2CAP_LE_PSM_DYNAMIC_RANGE_END + 1 ): if candidate in self.le_coc_servers: continue spec.psm = candidate break else: raise InvalidStateError('no free PSM') else: # Check that the PSM isn't already in use if spec.psm in self.le_coc_servers: raise InvalidArgumentError('PSM already in use') self.le_coc_servers[spec.psm] = LeCreditBasedChannelServer( self, spec.psm, handler, max_credits=spec.max_credits, mtu=spec.mtu, mps=spec.mps, ) return self.le_coc_servers[spec.psm] def on_disconnection(self, connection_handle: int, reason: int) -> None: del reason # unused. logger.debug('disconnection from %d, cleaning up channels', connection_handle) if channels := self.channels.pop(connection_handle, None): for channel in channels.values(): channel.abort() if le_coc_channels := self.le_coc_channels.pop(connection_handle, None): for le_coc_channel in le_coc_channels.values(): le_coc_channel.abort() if pending_credit_based_connections := self.pending_credit_based_connections.pop( connection_handle, None ): for future, _ in pending_credit_based_connections.values(): if not future.done(): future.cancel("ACL disconnected") self.identifiers.pop(connection_handle, None) def send_pdu( self, connection: Connection, cid: int, pdu: SupportsBytes | bytes, with_fcs: bool = False, ) -> None: pdu_str = pdu.hex() if isinstance(pdu, bytes) else str(pdu) pdu_bytes = bytes(pdu) logger.debug( f'{color(">>> Sending L2CAP PDU", "blue")} ' f'on connection [0x{connection.handle:04X}] (CID={cid}) ' f'{connection.peer_address}: {len(pdu_bytes)} bytes, {pdu_str}' ) self.host.send_acl_sdu( connection.handle, L2CAP_PDU(cid, bytes(pdu)).to_bytes(with_fcs=with_fcs) ) def on_pdu(self, connection: Connection, cid: int, pdu: bytes) -> None: if cid in (L2CAP_SIGNALING_CID, L2CAP_LE_SIGNALING_CID): # Parse the L2CAP payload into a Control Frame object control_frame = L2CAP_Control_Frame.from_bytes(pdu) self.on_control_frame(connection, cid, control_frame) elif cid in self.fixed_channels: handler = self.fixed_channels[cid] assert handler is not None handler(connection.handle, pdu) else: if (channel := self.find_channel(connection.handle, cid)) is None: logger.warning( color( f'channel not found for 0x{connection.handle:04X}:{cid}', 'red' ) ) return channel.on_pdu(pdu) def send_control_frame( self, connection: Connection, cid: int, control_frame: L2CAP_Control_Frame ) -> None: logger.debug( f'{color(">>> Sending L2CAP Signaling Control Frame", "blue")} ' f'on connection [0x{connection.handle:04X}] (CID={cid}) ' f'{connection.peer_address}:\n{control_frame}' ) self.host.send_l2cap_pdu(connection.handle, cid, bytes(control_frame)) def on_control_frame( self, connection: Connection, cid: int, control_frame: L2CAP_Control_Frame ) -> None: logger.debug( f'{color("<<< Received L2CAP Signaling Control Frame", "green")} ' f'on connection [0x{connection.handle:04X}] (CID={cid}) ' f'{connection.peer_address}:\n{control_frame}' ) # Find the handler method handler_name = f'on_{control_frame.name.lower()}' handler = getattr(self, handler_name, None) if handler: try: handler(connection, cid, control_frame) except Exception: logger.exception(color("!!! Exception in handler:", "red")) self.send_control_frame( connection, cid, L2CAP_Command_Reject( identifier=control_frame.identifier, reason=L2CAP_COMMAND_NOT_UNDERSTOOD_REASON, data=b'', ), ) raise else: logger.error(color('Channel Manager command not handled???', 'red')) self.send_control_frame( connection, cid, L2CAP_Command_Reject( identifier=control_frame.identifier, reason=L2CAP_COMMAND_NOT_UNDERSTOOD_REASON, data=b'', ), ) def on_l2cap_command_reject( self, _connection: Connection, _cid: int, packet: L2CAP_Command_Reject ) -> None: logger.warning(f'{color("!!! Command rejected:", "red")} {packet.reason}') def on_l2cap_connection_request( self, connection: Connection, cid: int, request: L2CAP_Connection_Request ) -> None: # Check if there's a server for this PSM server = self.servers.get(request.psm) if server: # Find a free CID for this new channel connection_channels = self.channels.setdefault(connection.handle, {}) source_cid = self.find_free_br_edr_cid(connection_channels) if source_cid is None: # Should never happen! self.send_control_frame( connection, cid, L2CAP_Connection_Response( identifier=request.identifier, destination_cid=0, source_cid=request.source_cid, result=L2CAP_Connection_Response.Result.CONNECTION_REFUSED_NO_RESOURCES_AVAILABLE, status=0x0000, ), ) return # Create a new channel logger.debug( f'creating server channel with cid={source_cid} for psm {request.psm}' ) channel = ClassicChannel( manager=self, connection=connection, signaling_cid=cid, psm=request.psm, source_cid=source_cid, spec=server.spec, ) connection_channels[source_cid] = channel # Notify server.on_connection(channel) channel.on_connection_request(request) else: logger.warning( f'No server for connection 0x{connection.handle:04X} ' f'on PSM {request.psm}' ) self.send_control_frame( connection, cid, L2CAP_Connection_Response( identifier=request.identifier, destination_cid=0, source_cid=request.source_cid, result=L2CAP_Connection_Response.Result.CONNECTION_REFUSED_PSM_NOT_SUPPORTED, status=0x0000, ), ) def on_l2cap_connection_response( self, connection: Connection, cid: int, response: L2CAP_Connection_Response, ) -> None: if ( channel := self.find_channel(connection.handle, response.source_cid) ) is None: logger.warning( color( f'channel {response.source_cid} not found for ' f'0x{connection.handle:04X}:{cid}', 'red', ) ) return channel.on_connection_response(response) def on_l2cap_configure_request( self, connection: Connection, cid: int, request: L2CAP_Configure_Request ) -> None: if ( channel := self.find_channel(connection.handle, request.destination_cid) ) is None: logger.warning( color( f'channel {request.destination_cid} not found for ' f'0x{connection.handle:04X}:{cid}', 'red', ) ) return channel.on_configure_request(request) def on_l2cap_configure_response( self, connection: Connection, cid: int, response: L2CAP_Configure_Response ) -> None: if ( channel := self.find_channel(connection.handle, response.source_cid) ) is None: logger.warning( color( f'channel {response.source_cid} not found for ' f'0x{connection.handle:04X}:{cid}', 'red', ) ) return channel.on_configure_response(response) def on_l2cap_disconnection_request( self, connection: Connection, cid: int, request: L2CAP_Disconnection_Request ) -> None: if ( channel := self.find_channel(connection.handle, request.destination_cid) ) is None: logger.warning( color( f'channel {request.destination_cid} not found for ' f'0x{connection.handle:04X}:{cid}', 'red', ) ) return channel.on_disconnection_request(request) def on_l2cap_disconnection_response( self, connection: Connection, cid: int, response: L2CAP_Disconnection_Response ) -> None: if ( channel := self.find_channel(connection.handle, response.source_cid) ) is None: logger.warning( color( f'channel {response.source_cid} not found for ' f'0x{connection.handle:04X}:{cid}', 'red', ) ) return channel.on_disconnection_response(response) def on_l2cap_echo_request( self, connection: Connection, cid: int, request: L2CAP_Echo_Request ) -> None: logger.debug(f'<<< Echo request: data={request.data.hex()}') self.send_control_frame( connection, cid, L2CAP_Echo_Response(identifier=request.identifier, data=request.data), ) def on_l2cap_echo_response( self, _connection: Connection, _cid: int, response: L2CAP_Echo_Response ) -> None: logger.debug(f'<<< Echo response: data={response.data.hex()}') # TODO notify listeners def on_l2cap_information_request( self, connection: Connection, cid: int, request: L2CAP_Information_Request ) -> None: if request.info_type == L2CAP_Information_Request.InfoType.CONNECTIONLESS_MTU: result = L2CAP_Information_Response.Result.SUCCESS data = self.connectionless_mtu.to_bytes(2, 'little') elif ( request.info_type == L2CAP_Information_Request.InfoType.EXTENDED_FEATURES_SUPPORTED ): result = L2CAP_Information_Response.Result.SUCCESS data = sum(self.extended_features).to_bytes(4, 'little') elif ( request.info_type == L2CAP_Information_Request.InfoType.FIXED_CHANNELS_SUPPORTED ): result = L2CAP_Information_Response.Result.SUCCESS data = sum(1 << cid for cid in self.fixed_channels).to_bytes(8, 'little') else: result = L2CAP_Information_Response.Result.NOT_SUPPORTED data = b'' self.send_control_frame( connection, cid, L2CAP_Information_Response( identifier=request.identifier, info_type=request.info_type, result=result, data=data, ), ) def on_l2cap_connection_parameter_update_request( self, connection: Connection, cid: int, request: L2CAP_Connection_Parameter_Update_Request, ): if connection.role == hci.Role.CENTRAL: self.send_control_frame( connection, cid, L2CAP_Connection_Parameter_Update_Response( identifier=request.identifier, result=L2CAP_CONNECTION_PARAMETERS_ACCEPTED_RESULT, ), ) self.host.send_command_sync( hci.HCI_LE_Connection_Update_Command( connection_handle=connection.handle, connection_interval_min=request.interval_min, connection_interval_max=request.interval_max, max_latency=request.latency, supervision_timeout=request.timeout, min_ce_length=0, max_ce_length=0, ) ) else: self.send_control_frame( connection, cid, L2CAP_Connection_Parameter_Update_Response( identifier=request.identifier, result=L2CAP_CONNECTION_PARAMETERS_REJECTED_RESULT, ), ) async def update_connection_parameters( self, connection: Connection, interval_min: int, interval_max: int, latency: int, timeout: int, ) -> int: # Check that there isn't already a request pending if self.connection_parameters_update_response: raise InvalidStateError('request already pending') self.connection_parameters_update_response = ( asyncio.get_running_loop().create_future() ) self.send_control_frame( connection, L2CAP_LE_SIGNALING_CID, L2CAP_Connection_Parameter_Update_Request( identifier=self.next_identifier(connection), interval_min=interval_min, interval_max=interval_max, latency=latency, timeout=timeout, ), ) return await self.connection_parameters_update_response def on_l2cap_connection_parameter_update_response( self, connection: Connection, cid: int, response: L2CAP_Connection_Parameter_Update_Response, ) -> None: if self.connection_parameters_update_response: self.connection_parameters_update_response.set_result(response.result) self.connection_parameters_update_response = None else: logger.warning( color( 'received l2cap_connection_parameter_update_response without a pending request', 'red', ) ) def on_l2cap_le_credit_based_connection_request( self, connection: Connection, cid: int, request: L2CAP_LE_Credit_Based_Connection_Request, ) -> None: if not (server := self.le_coc_servers.get(request.le_psm)): logger.info( f'No LE server for connection 0x{connection.handle:04X} ' f'on PSM {request.le_psm}' ) self.send_control_frame( connection, cid, L2CAP_LE_Credit_Based_Connection_Response( identifier=request.identifier, destination_cid=0, mtu=L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MTU, mps=L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MPS, initial_credits=0, result=L2CAP_LE_Credit_Based_Connection_Response.Result.CONNECTION_REFUSED_LE_PSM_NOT_SUPPORTED, ), ) return # Check that the CID isn't already used le_connection_channels = self.le_coc_channels.setdefault(connection.handle, {}) if request.source_cid in le_connection_channels: logger.warning(f'source CID {request.source_cid} already in use') self.send_control_frame( connection, cid, L2CAP_LE_Credit_Based_Connection_Response( identifier=request.identifier, destination_cid=0, mtu=server.mtu, mps=server.mps, initial_credits=0, result=L2CAP_LE_Credit_Based_Connection_Response.Result.CONNECTION_REFUSED_SOURCE_CID_ALREADY_ALLOCATED, ), ) return # Find a free CID for this new channel connection_channels = self.channels.setdefault(connection.handle, {}) source_cid = self.find_free_le_cid(connection_channels) if source_cid is None: # Should never happen! self.send_control_frame( connection, cid, L2CAP_LE_Credit_Based_Connection_Response( identifier=request.identifier, destination_cid=0, mtu=server.mtu, mps=server.mps, initial_credits=0, result=L2CAP_LE_Credit_Based_Connection_Response.Result.CONNECTION_REFUSED_NO_RESOURCES_AVAILABLE, ), ) return # Create a new channel logger.debug( f'creating LE CoC server channel with cid={source_cid} for psm ' f'{request.le_psm}' ) channel = LeCreditBasedChannel( self, connection, request.le_psm, source_cid, request.source_cid, server.mtu, server.mps, request.initial_credits, request.mtu, request.mps, server.max_credits, True, ) connection_channels[source_cid] = channel le_connection_channels[request.source_cid] = channel # Respond self.send_control_frame( connection, cid, L2CAP_LE_Credit_Based_Connection_Response( identifier=request.identifier, destination_cid=source_cid, mtu=server.mtu, mps=server.mps, initial_credits=server.max_credits, # pylint: disable=line-too-long result=L2CAP_LE_Credit_Based_Connection_Response.Result.CONNECTION_SUCCESSFUL, ), ) # Notify server.on_connection(channel) def on_l2cap_le_credit_based_connection_response( self, connection: Connection, _cid: int, response: L2CAP_LE_Credit_Based_Connection_Response, ) -> None: # Find the pending request by identifier if not (request := self.le_coc_requests.pop(response.identifier, None)): logger.warning(color('!!! received response for unknown request', 'red')) return # Find the channel for this request channel = self.find_channel(connection.handle, request.source_cid) if channel is None: logger.warning( color( 'received connection response for an unknown channel ' f'(cid={request.source_cid})', 'red', ) ) return # Process the response channel.on_connection_response(response) def on_l2cap_credit_based_connection_request( self, connection: Connection, cid: int, request: L2CAP_Credit_Based_Connection_Request, ) -> None: if not (server := self.le_coc_servers.get(request.spsm)): logger.info( 'No LE server for connection 0x%04X ' 'on PSM %d', connection.handle, request.spsm, ) self.send_control_frame( connection, cid, L2CAP_Credit_Based_Connection_Response( identifier=request.identifier, destination_cid=[], mtu=L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MTU, mps=L2CAP_LE_CREDIT_BASED_CONNECTION_DEFAULT_MPS, initial_credits=0, result=L2CAP_Credit_Based_Connection_Response.Result.ALL_CONNECTIONS_REFUSED_SPSM_NOT_SUPPORTED, ), ) return # Check that the CID isn't already used le_connection_channels = self.le_coc_channels.setdefault(connection.handle, {}) if cid_in_use := set(request.source_cid).intersection( set(le_connection_channels) ): logger.warning('source CID already in use: %s', cid_in_use) self.send_control_frame( connection, cid, L2CAP_Credit_Based_Connection_Response( identifier=request.identifier, mtu=server.mtu, mps=server.mps, initial_credits=0, result=L2CAP_Credit_Based_Connection_Response.Result.SOME_CONNECTIONS_REFUSED_SOURCE_CID_ALREADY_ALLOCATED, destination_cid=[], ), ) return # Find free CIDs for new channels connection_channels = self.channels.setdefault(connection.handle, {}) source_cids = self.find_free_le_cids( connection_channels, len(request.source_cid) ) if not source_cids: self.send_control_frame( connection, cid, L2CAP_Credit_Based_Connection_Response( identifier=request.identifier, destination_cid=[], mtu=server.mtu, mps=server.mps, initial_credits=server.max_credits, result=L2CAP_Credit_Based_Connection_Response.Result.SOME_CONNECTIONS_REFUSED_INSUFFICIENT_RESOURCES_AVAILABLE, ), ) return for destination_cid in request.source_cid: # TODO: Handle Classic channels. if not (source_cid := self.find_free_le_cid(connection_channels)): logger.warning("No free CIDs available") break # Create a new channel logger.debug( 'creating LE CoC server channel with cid=%s for psm %s', source_cid, request.spsm, ) channel = LeCreditBasedChannel( self, connection, request.spsm, source_cid, destination_cid, server.mtu, server.mps, request.initial_credits, request.mtu, request.mps, server.max_credits, True, ) connection_channels[source_cid] = channel le_connection_channels[source_cid] = channel server.on_connection(channel) # Respond self.send_control_frame( connection, cid, L2CAP_Credit_Based_Connection_Response( identifier=request.identifier, destination_cid=source_cids, mtu=server.mtu, mps=server.mps, initial_credits=server.max_credits, result=L2CAP_Credit_Based_Connection_Response.Result.ALL_CONNECTIONS_SUCCESSFUL, ), ) def on_l2cap_credit_based_connection_response( self, connection: Connection, _cid: int, response: L2CAP_Credit_Based_Connection_Response, ) -> None: # Find the pending request by identifier pending_connections = self.pending_credit_based_connections.setdefault( connection.handle, {} ) if not ( pending_connection := pending_connections.pop(response.identifier, None) ): logger.warning(color('!!! received response for unknown request', 'red')) return connection_result, channels = pending_connection # Process the response for channel, destination_cid in zip(channels, response.destination_cid): channel.on_enhanced_connection_response(destination_cid, response) if ( response.result == L2CAP_Credit_Based_Connection_Response.Result.ALL_CONNECTIONS_SUCCESSFUL ): connection_result.set_result(None) else: connection_result.set_exception( L2capError(response.result, response.result.name) ) def on_l2cap_le_flow_control_credit( self, connection: Connection, _cid: int, credit: L2CAP_LE_Flow_Control_Credit ) -> None: channel = self.find_le_coc_channel(connection.handle, credit.cid) if channel is None: logger.warning(f'received credits for an unknown channel (cid={credit.cid}') return channel.on_credits(credit.credits) def on_channel_closed(self, channel: ClassicChannel) -> None: connection_channels = self.channels.get(channel.connection.handle) if connection_channels: if channel.source_cid in connection_channels: del connection_channels[channel.source_cid] async def create_le_credit_based_channel( self, connection: Connection, spec: LeCreditBasedChannelSpec, ) -> LeCreditBasedChannel: # Find a free CID for the new channel connection_channels = self.channels.setdefault(connection.handle, {}) source_cid = self.find_free_le_cid(connection_channels) if source_cid is None: # Should never happen! raise OutOfResourcesError('all CIDs already in use') if spec.psm is None: raise InvalidArgumentError('PSM cannot be None') # Create the channel logger.debug(f'creating coc channel with cid={source_cid} for psm {spec.psm}') channel = LeCreditBasedChannel( manager=self, connection=connection, psm=spec.psm, source_cid=source_cid, destination_cid=0, mtu=spec.mtu, mps=spec.mps, credits=0, peer_mtu=0, peer_mps=0, peer_credits=spec.max_credits, connected=False, ) connection_channels[source_cid] = channel # Connect try: await channel.connect() except Exception: logger.exception('connection failed') del connection_channels[source_cid] raise # Remember the channel by source CID and destination CID le_connection_channels = self.le_coc_channels.setdefault(connection.handle, {}) le_connection_channels[channel.destination_cid] = channel return channel async def create_classic_channel( self, connection: Connection, spec: ClassicChannelSpec ) -> ClassicChannel: # NOTE: this implementation hard-codes BR/EDR # Find a free CID for a new channel connection_channels = self.channels.setdefault(connection.handle, {}) source_cid = self.find_free_br_edr_cid(connection_channels) if source_cid is None: # Should never happen! raise OutOfResourcesError('all CIDs already in use') if spec.psm is None: raise InvalidArgumentError('PSM cannot be None') # Create the channel logger.debug( f'creating client channel with cid={source_cid} for psm {spec.psm}' ) channel = ClassicChannel( manager=self, connection=connection, signaling_cid=L2CAP_SIGNALING_CID, psm=spec.psm, source_cid=source_cid, spec=spec, ) connection_channels[source_cid] = channel # Connect try: await channel.connect() except BaseException as e: connection_channels.pop(source_cid, None) raise e return channel async def create_enhanced_credit_based_channels( self, connection: Connection, spec: LeCreditBasedChannelSpec, count: int, ) -> list[LeCreditBasedChannel]: # Find a free CID for the new channel connection_channels = self.channels.setdefault(connection.handle, {}) source_cids = self.find_free_le_cids(connection_channels, count) if not source_cids: # Should never happen! raise OutOfResourcesError('all CIDs already in use') if spec.psm is None: raise InvalidArgumentError('PSM cannot be None') # Create the channel logger.debug( 'creating coc channel with cid=%s for psm %s', source_cids, spec.psm ) channels: list[LeCreditBasedChannel] = [] for source_cid in source_cids: channel = LeCreditBasedChannel( manager=self, connection=connection, psm=spec.psm, source_cid=source_cid, destination_cid=0, mtu=spec.mtu, mps=spec.mps, credits=0, peer_mtu=0, peer_mps=0, peer_credits=spec.max_credits, connected=False, ) connection_channels[source_cid] = channel channels.append(channel) identifier = self.next_identifier(connection) request = L2CAP_Credit_Based_Connection_Request( identifier=identifier, spsm=spec.psm, mtu=spec.mtu, mps=spec.mps, initial_credits=spec.max_credits, source_cid=source_cids, ) connection_result = asyncio.get_running_loop().create_future() pending_connections = self.pending_credit_based_connections.setdefault( connection.handle, {} ) pending_connections[identifier] = (connection_result, channels) self.send_control_frame( connection, L2CAP_LE_SIGNALING_CID, request, ) # Connect try: await connection_result except Exception: logger.exception('connection failed') for cid in source_cids: del connection_channels[cid] raise # Remember the channel by source CID and destination CID le_connection_channels = self.le_coc_channels.setdefault(connection.handle, {}) for channel in channels: le_connection_channels[channel.destination_cid] = channel return channels @classmethod def make_mode_processor( self, channel: ClassicChannel, mode: TransmissionMode, peer_tx_window_size: int, peer_max_retransmission: int, peer_retransmission_timeout: int, peer_monitor_timeout: int, peer_mps: int, ) -> Processor: del peer_retransmission_timeout, peer_monitor_timeout # Unused. if mode == TransmissionMode.BASIC: return Processor(channel) elif mode == TransmissionMode.ENHANCED_RETRANSMISSION: return EnhancedRetransmissionProcessor( channel, peer_tx_window_size, peer_max_retransmission, peer_mps ) raise InvalidArgumentError("Mode %s is not implemented", mode.name)