| # 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 |
| |
| from .colors import color |
| from .att import ATT_CID, ATT_PDU |
| from .smp import SMP_CID, SMP_Command |
| from .core import name_or_number |
| from .l2cap import ( |
| L2CAP_PDU, |
| L2CAP_CONNECTION_REQUEST, |
| L2CAP_CONNECTION_RESPONSE, |
| L2CAP_SIGNALING_CID, |
| L2CAP_LE_SIGNALING_CID, |
| L2CAP_Control_Frame, |
| L2CAP_Connection_Response, |
| ) |
| from .hci import ( |
| HCI_EVENT_PACKET, |
| HCI_ACL_DATA_PACKET, |
| HCI_DISCONNECTION_COMPLETE_EVENT, |
| HCI_AclDataPacketAssembler, |
| ) |
| from .rfcomm import RFCOMM_Frame, RFCOMM_PSM |
| from .sdp import SDP_PDU, SDP_PSM |
| from .avdtp import MessageAssembler as AVDTP_MessageAssembler, AVDTP_PSM |
| |
| # ----------------------------------------------------------------------------- |
| # Logging |
| # ----------------------------------------------------------------------------- |
| logger = logging.getLogger(__name__) |
| |
| |
| # ----------------------------------------------------------------------------- |
| PSM_NAMES = { |
| RFCOMM_PSM: 'RFCOMM', |
| SDP_PSM: 'SDP', |
| AVDTP_PSM: 'AVDTP' |
| # TODO: add more PSM values |
| } |
| |
| |
| # ----------------------------------------------------------------------------- |
| class PacketTracer: |
| class AclStream: |
| def __init__(self, analyzer): |
| self.analyzer = analyzer |
| self.packet_assembler = HCI_AclDataPacketAssembler(self.on_acl_pdu) |
| self.avdtp_assemblers = {} # AVDTP assemblers, by source_cid |
| self.psms = {} # PSM, by source_cid |
| self.peer = None # ACL stream in the other direction |
| |
| # pylint: disable=too-many-nested-blocks |
| def on_acl_pdu(self, pdu): |
| l2cap_pdu = L2CAP_PDU.from_bytes(pdu) |
| |
| if l2cap_pdu.cid == ATT_CID: |
| att_pdu = ATT_PDU.from_bytes(l2cap_pdu.payload) |
| self.analyzer.emit(att_pdu) |
| elif l2cap_pdu.cid == SMP_CID: |
| smp_command = SMP_Command.from_bytes(l2cap_pdu.payload) |
| self.analyzer.emit(smp_command) |
| elif l2cap_pdu.cid in (L2CAP_SIGNALING_CID, L2CAP_LE_SIGNALING_CID): |
| control_frame = L2CAP_Control_Frame.from_bytes(l2cap_pdu.payload) |
| self.analyzer.emit(control_frame) |
| |
| # Check if this signals a new channel |
| if control_frame.code == L2CAP_CONNECTION_REQUEST: |
| self.psms[control_frame.source_cid] = control_frame.psm |
| elif control_frame.code == L2CAP_CONNECTION_RESPONSE: |
| if ( |
| control_frame.result |
| == L2CAP_Connection_Response.CONNECTION_SUCCESSFUL |
| ): |
| if self.peer: |
| if psm := self.peer.psms.get(control_frame.source_cid): |
| # Found a pending connection |
| self.psms[control_frame.destination_cid] = psm |
| |
| # For AVDTP connections, create a packet assembler for |
| # each direction |
| if psm == AVDTP_PSM: |
| self.avdtp_assemblers[ |
| control_frame.source_cid |
| ] = AVDTP_MessageAssembler(self.on_avdtp_message) |
| self.peer.avdtp_assemblers[ |
| control_frame.destination_cid |
| ] = AVDTP_MessageAssembler( |
| self.peer.on_avdtp_message |
| ) |
| |
| else: |
| # Try to find the PSM associated with this PDU |
| if self.peer and (psm := self.peer.psms.get(l2cap_pdu.cid)): |
| if psm == SDP_PSM: |
| sdp_pdu = SDP_PDU.from_bytes(l2cap_pdu.payload) |
| self.analyzer.emit(sdp_pdu) |
| elif psm == RFCOMM_PSM: |
| rfcomm_frame = RFCOMM_Frame.from_bytes(l2cap_pdu.payload) |
| self.analyzer.emit(rfcomm_frame) |
| elif psm == AVDTP_PSM: |
| self.analyzer.emit( |
| f'{color("L2CAP", "green")} [CID={l2cap_pdu.cid}, ' |
| f'PSM=AVDTP]: {l2cap_pdu.payload.hex()}' |
| ) |
| assembler = self.avdtp_assemblers.get(l2cap_pdu.cid) |
| if assembler: |
| assembler.on_pdu(l2cap_pdu.payload) |
| else: |
| psm_string = name_or_number(PSM_NAMES, psm) |
| self.analyzer.emit( |
| f'{color("L2CAP", "green")} [CID={l2cap_pdu.cid}, ' |
| f'PSM={psm_string}]: {l2cap_pdu.payload.hex()}' |
| ) |
| else: |
| self.analyzer.emit(l2cap_pdu) |
| |
| def on_avdtp_message(self, transaction_label, message): |
| self.analyzer.emit( |
| f'{color("AVDTP", "green")} [{transaction_label}] {message}' |
| ) |
| |
| def feed_packet(self, packet): |
| self.packet_assembler.feed_packet(packet) |
| |
| class Analyzer: |
| def __init__(self, label, emit_message): |
| self.label = label |
| self.emit_message = emit_message |
| self.acl_streams = {} # ACL streams, by connection handle |
| self.peer = None # Analyzer in the other direction |
| |
| def start_acl_stream(self, connection_handle): |
| logger.info( |
| f'[{self.label}] +++ Creating ACL stream for connection ' |
| f'0x{connection_handle:04X}' |
| ) |
| stream = PacketTracer.AclStream(self) |
| self.acl_streams[connection_handle] = stream |
| |
| # Associate with a peer stream if we can |
| if peer_stream := self.peer.acl_streams.get(connection_handle): |
| stream.peer = peer_stream |
| peer_stream.peer = stream |
| |
| return stream |
| |
| def end_acl_stream(self, connection_handle): |
| if connection_handle in self.acl_streams: |
| logger.info( |
| f'[{self.label}] --- Removing ACL stream for connection ' |
| f'0x{connection_handle:04X}' |
| ) |
| del self.acl_streams[connection_handle] |
| |
| # Let the other forwarder know so it can cleanup its stream as well |
| self.peer.end_acl_stream(connection_handle) |
| |
| def on_packet(self, packet): |
| self.emit(packet) |
| |
| if packet.hci_packet_type == HCI_ACL_DATA_PACKET: |
| # Look for an existing stream for this handle, create one if it is the |
| # first ACL packet for that connection handle |
| if (stream := self.acl_streams.get(packet.connection_handle)) is None: |
| stream = self.start_acl_stream(packet.connection_handle) |
| stream.feed_packet(packet) |
| elif packet.hci_packet_type == HCI_EVENT_PACKET: |
| if packet.event_code == HCI_DISCONNECTION_COMPLETE_EVENT: |
| self.end_acl_stream(packet.connection_handle) |
| |
| def emit(self, message): |
| self.emit_message(f'[{self.label}] {message}') |
| |
| def trace(self, packet, direction=0): |
| if direction == 0: |
| self.host_to_controller_analyzer.on_packet(packet) |
| else: |
| self.controller_to_host_analyzer.on_packet(packet) |
| |
| def __init__( |
| self, |
| host_to_controller_label=color('HOST->CONTROLLER', 'blue'), |
| controller_to_host_label=color('CONTROLLER->HOST', 'cyan'), |
| emit_message=logger.info, |
| ): |
| self.host_to_controller_analyzer = PacketTracer.Analyzer( |
| host_to_controller_label, emit_message |
| ) |
| self.controller_to_host_analyzer = PacketTracer.Analyzer( |
| controller_to_host_label, emit_message |
| ) |
| self.host_to_controller_analyzer.peer = self.controller_to_host_analyzer |
| self.controller_to_host_analyzer.peer = self.host_to_controller_analyzer |
