1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
|
# 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 asyncio
import sys
import bumble.core
import bumble.logging
from bumble.colors import color
from bumble.core import BT_L2CAP_PROTOCOL_ID, BT_RFCOMM_PROTOCOL_ID, PhysicalTransport
from bumble.device import Device
from bumble.rfcomm import Client
from bumble.sdp import (
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
)
from bumble.sdp import Client as SDP_Client
from bumble.sdp import DataElement, ServiceAttribute
from bumble.transport import open_transport
# -----------------------------------------------------------------------------
async def list_rfcomm_channels(connection):
# Connect to the SDP Server
sdp_client = SDP_Client(connection)
await sdp_client.connect()
# Search for services with an L2CAP service attribute
search_result = await sdp_client.search_attributes(
[BT_L2CAP_PROTOCOL_ID],
[
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
],
)
print(color('==================================', 'blue'))
print(color('RFCOMM Services:', 'yellow'))
# pylint: disable-next=too-many-nested-blocks
for attribute_list in search_result:
# Look for the RFCOMM Channel number
protocol_descriptor_list = ServiceAttribute.find_attribute_in_list(
attribute_list, SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID
)
if protocol_descriptor_list:
for protocol_descriptor in protocol_descriptor_list.value:
if len(protocol_descriptor.value) >= 2:
if protocol_descriptor.value[0].value == BT_RFCOMM_PROTOCOL_ID:
print(color('SERVICE:', 'green'))
print(
color(' RFCOMM Channel:', 'cyan'),
protocol_descriptor.value[1].value,
)
# List profiles
bluetooth_profile_descriptor_list = (
ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
)
)
if bluetooth_profile_descriptor_list:
if bluetooth_profile_descriptor_list.value:
if (
bluetooth_profile_descriptor_list.value[0].type
== DataElement.SEQUENCE
):
bluetooth_profile_descriptors = (
bluetooth_profile_descriptor_list.value
)
else:
# Sometimes, instead of a list of lists, we just
# find a list. Fix that
bluetooth_profile_descriptors = [
bluetooth_profile_descriptor_list
]
print(color(' Profiles:', 'green'))
for (
bluetooth_profile_descriptor
) in bluetooth_profile_descriptors:
version_major = (
bluetooth_profile_descriptor.value[1].value >> 8
)
version_minor = (
bluetooth_profile_descriptor.value[1].value
& 0xFF
)
print(
' '
f'{bluetooth_profile_descriptor.value[0].value}'
f' - version {version_major}.{version_minor}'
)
# List service classes
service_class_id_list = ServiceAttribute.find_attribute_in_list(
attribute_list, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
)
if service_class_id_list:
if service_class_id_list.value:
print(color(' Service Classes:', 'green'))
for service_class_id in service_class_id_list.value:
print(' ', service_class_id.value)
await sdp_client.disconnect()
# -----------------------------------------------------------------------------
class TcpServerProtocol(asyncio.Protocol):
def __init__(self, rfcomm_session):
self.rfcomm_session = rfcomm_session
self.transport = None
def connection_made(self, transport):
peer_name = transport.get_extra_info('peer_name')
print(f'<<< TCP Server: connection from {peer_name}')
self.transport = transport
self.rfcomm_session.sink = self.rfcomm_data_received
def rfcomm_data_received(self, data):
print(f'<<< RFCOMM Data: {data.hex()}')
if self.transport:
self.transport.write(data)
else:
print('!!! no TCP connection, dropping data')
def data_received(self, data):
print(f'<<< TCP Server: data received: {len(data)} bytes - {data.hex()}')
self.rfcomm_session.write(data)
# -----------------------------------------------------------------------------
async def tcp_server(tcp_port, rfcomm_session):
print(f'$$$ Starting TCP server on port {tcp_port}')
server = await asyncio.get_running_loop().create_server(
lambda: TcpServerProtocol(rfcomm_session), '127.0.0.1', tcp_port
)
await asyncio.get_running_loop().create_future()
async with server:
await server.serve_forever()
# -----------------------------------------------------------------------------
async def main() -> None:
if len(sys.argv) < 5:
print(
'Usage: run_rfcomm_client.py <device-config> <transport-spec> '
'<bluetooth-address> <channel>|discover [tcp-port]'
)
print(
' specifying a channel number, or "discover" to list all RFCOMM channels'
)
print('example: run_rfcomm_client.py classic1.json usb:0 E1:CA:72:48:C4:E8 8')
return
print('<<< connecting to HCI...')
async with await open_transport(sys.argv[2]) as hci_transport:
print('<<< connected')
# Create a device
device = Device.from_config_file_with_hci(
sys.argv[1], hci_transport.source, hci_transport.sink
)
device.classic_enabled = True
await device.power_on()
# Connect to a peer
target_address = sys.argv[3]
print(f'=== Connecting to {target_address}...')
connection = await device.connect(
target_address, transport=PhysicalTransport.BR_EDR
)
print(f'=== Connected to {connection.peer_address}!')
channel_str = sys.argv[4]
if channel_str == 'discover':
await list_rfcomm_channels(connection)
return
# Request authentication
print('*** Authenticating...')
await connection.authenticate()
print('*** Authenticated')
# Enable encryption
print('*** Enabling encryption...')
await connection.encrypt()
print('*** Encryption on')
# Create a client and start it
print('@@@ Starting RFCOMM client...')
rfcomm_client = Client(connection)
rfcomm_mux = await rfcomm_client.start()
print('@@@ Started')
channel = int(channel_str)
print(f'### Opening session for channel {channel}...')
try:
session = await rfcomm_mux.open_dlc(channel)
print('### Session open', session)
except bumble.core.ConnectionError as error:
print(f'### Session open failed: {error}')
await rfcomm_mux.disconnect()
print('@@@ Disconnected from RFCOMM server')
return
if len(sys.argv) == 6:
# A TCP port was specified, start listening
tcp_port = int(sys.argv[5])
asyncio.create_task(tcp_server(tcp_port, session))
await hci_transport.source.wait_for_termination()
# -----------------------------------------------------------------------------
bumble.logging.setup_basic_logging('DEBUG')
asyncio.run(main())
|
