Issue when using bleak on linux when everything worked well on windows

aefrtyi commented 1 year ago

bleak version: 0.15.1
Python version: 3.9
Operating System: Debian GNU/Linux 11 (bullseye)
BlueZ version (bluetoothctl -v) in case of Linux: 5.55

Description

I have a ble peripheral device that has a custom scan response data in EIR 0x42. I previously used bluepy on linux to scan and communicate with it, but I wanted to be able to use my code on both windows and linux, so I switch to bleak.

I managed to scan my device with the expected data on windows and communicate with it successfully, however, when I try to do so on linux, BleakClient.get_services() returns an empty dict, and I am unable to find the adv data I'm looking for.

What I Did

Scanner:

import asyncio
import json
import platform
import struct
from typing import Dict, List, Tuple

from bleak import BleakScanner
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData

CURRENT_SYSTEM: str = platform.system()

def __format_adv_data(device: BLEDevice) -> Dict[bytes, bytes]:
        adv_data: dict = dict()
        if CURRENT_SYSTEM == "Windows":
            for detail in device.details:
                # device possess advertising details that are enumerable. In some cases, some of those details may be empty
                if detail is not None:
                    # if detail is not empty, then get the adv data from the details obtained
                    for data in detail.advertisement.data_sections:
                        unpacked_adv_data = [dt[0] for dt in struct.iter_unpack('<B', data.data)]
                        # Convert adv data to match correct format to use later on
                        bytes_adv_data = b"".join([(dt.to_bytes(1, "big")) for dt in unpacked_adv_data])
                        data_type = int(data.data_type).to_bytes(1, 'big')
                        adv_data[data_type] = bytes_adv_data
        elif CURRENT_SYSTEM == "Linux":
            detail = device.details.get('props')
            print(f"details {device.details} | type {type(device.details)}")
                # print(json.loads(detail).get('props'))
        return adv_data

def detection_callback(device: BLEDevice, adv_data: AdvertisementData):
    if device.address.lower() == '53:4e:48:00:00:07':
        if CURRENT_SYSTEM == "Windows":
            _, raw_adv_data = adv_data.platform_data
            new_adv_data, scan_data = raw_adv_data
            if new_adv_data is not None and scan_data is not None:
                formated_adv_data = dict()
                # if detail is not empty, then get the adv data from the details obtained
                for data in new_adv_data.advertisement.data_sections:
                    unpacked_adv_data = [dt[0] for dt in struct.iter_unpack('<B', data.data)]
                    # Convert adv data to match correct format to use later on
                    bytes_adv_data = b"".join([(dt.to_bytes(1, "big")) for dt in unpacked_adv_data])
                    data_type = int(data.data_type).to_bytes(1, 'big')
                    formated_adv_data[data_type] = bytes_adv_data
                for data in scan_data.advertisement.data_sections:
                    unpacked_adv_data = [dt[0] for dt in struct.iter_unpack('<B', data.data)]
                    # Convert adv data to match correct format to use later on
                    bytes_adv_data = b"".join([(dt.to_bytes(1, "big")) for dt in unpacked_adv_data])
                    data_type = int(data.data_type).to_bytes(1, 'big')
                    formated_adv_data[data_type] = bytes_adv_data
                print(formated_adv_data)
        elif CURRENT_SYSTEM == "Linux":
            print(f"local_name: {adv_data.local_name}")
            print(f"manufacturer_data: {adv_data.manufacturer_data}")
            print(f"service_data: {adv_data.service_data}")
            print(f"platform_data: {adv_data.platform_data}")
            for key, value in adv_data.platform_data.items():
                print(f"{key}: {value} | {type(value)}")
            print('-----------------------------------')

async def main():
    timeout: int = 5
    scanner: BleakScanner = BleakScanner()
    scanner.register_detection_callback(detection_callback)
    await scanner.start()
    await asyncio.sleep(timeout)
    await scanner.stop()
    # print('-----------------------------------')
    # for device in scanner.discovered_devices:
    #     if device.address.lower().startswith('53:4e:48:'):
    #         for detail in device.details:
    #             print(type(detail))

    devices: List[BLEDevice] = await scanner.discover(timeout)
    formated_devices: List[Tuple[str, dict[bytes, bytes]]] = list()
    for device in devices:
        if device.address.lower() == "53:4e:48:00:00:07":
            adv_data: dict[bytes, bytes] = __format_adv_data(device)
            formated_devices.append((device.address.lower(), adv_data))
    # print(f"devices scanned: {devices}")
    print(f'res: {formated_devices}')

if __name__ == '__main__':
    asyncio.run(main())

which gives me the following result on linux:

root@iot-gate-imx8:/home/compulab/code/GatewaySmartphone# pipenv run python main.py
local_name: None
manufacturer_data: {}
service_data: {}
platform_data: {'Address': '53:4E:48:00:00:07', 'AddressType': 'random', 'Alias': '53-4E-48-00-00-07', 'Paired': False, 'Trusted': False, 'Blocked': False, 'LegacyPairing': False, 'RSSI': -34, 'Connected': False, 'UUIDs': [], 'Adapter': '/org/bluez/hci0', 'ServicesResolved': False}
Address: 53:4E:48:00:00:07 | <class 'str'>
AddressType: random | <class 'str'>
Alias: 53-4E-48-00-00-07 | <class 'str'>
Paired: False | <class 'bool'>
Trusted: False | <class 'bool'>
Blocked: False | <class 'bool'>
LegacyPairing: False | <class 'bool'>
RSSI: -34 | <class 'int'>
Connected: False | <class 'bool'>
UUIDs: [] | <class 'list'>
Adapter: /org/bluez/hci0 | <class 'str'>
ServicesResolved: False | <class 'bool'>
-----------------------------------
details {'path': '/org/bluez/hci0/dev_53_4E_48_00_00_07', 'props': {'Address': '53:4E:48:00:00:07', 'AddressType': 'random', 'Alias': '53-4E-48-00-00-07', 'Paired': False, 'Trusted': False, 'Blocked': False, 'LegacyPairing': False, 'Connected': False, 'UUIDs': [], 'Adapter': '/org/bluez/hci0', 'ServicesResolved': False, 'RSSI': -32}} | type <class 'dict'>
res: [('53:4e:48:00:00:07', {})]

Communication on windows:

import asyncio
import time
from enum import Enum
from typing import List
from uuid import UUID

from bleak import BleakClient, BleakScanner
from bleak.backends.characteristic import BleakGATTCharacteristic
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from bleak.backends.service import BleakGATTService, BleakGATTServiceCollection

def __extract_itp_header(msg):
    desc_id = msg[0]
    header = {"desc_id": desc_id}
    if desc_id == 0:
        instr = msg[1:]
    elif desc_id == 1:
        header["nb_frames"] = int.from_bytes(msg[1:3], "little")
        crc_tmp = msg[3:7]
        # print(f'crc obtained from implant: {crc_tmp.hex()}')
        header["crc"] = crc_tmp[::-1]
        header["offset"] = int.from_bytes(msg[7:11], "little")
        header["offset_start"] = int.from_bytes(msg[11:15], "little")
        instr = msg[15:]
    else:
        instr = msg
    return header, instr

##  Enumerate of service and characteristics uuids in the implant GATT server
class UUIDs(Enum):
    # App med
    service_uuid = "6d79686561727473656e74696e656c20"
    req_uuid = "6d79686561727473656e74696e656c21"
    res_uuid = "6d79686561727473656e74696e656c22"
    data_ctrl_uuid = "6d79686561727473656e74696e656c23"
    data_uuid = "6d79686561727473656e74696e656c24"

class Chara():
    def __init__(self, bleak_chara: BleakGATTCharacteristic, bleak_client: BleakClient) -> None:
        self.chara: BleakGATTCharacteristic = bleak_chara
        self.client: BleakClient = bleak_client
        self.notifs: List[bytearray] = list()

    async def read(self) -> bytearray:
        value = await self.client.read_gatt_char(char_specifier=self.chara)
        return value

    async def write(self, value: bytes, with_response: bool = True) -> None:
        await self.client.write_gatt_char(char_specifier=self.chara, data=value, response=with_response)

    async def subscribe(self) -> None:
        await self.client.start_notify(char_specifier=self.chara, callback=self.__notification_handler)

    async def unsubscribe(self) -> None:
        await self.client.stop_notify(char_specifier=self.chara)

    async def wait_notification(self, timeout: int) -> bytes:
        response = None
        # Wait for a notification until there is a response in the list of notification of the characteristic
        start_time: float = time.time()
        while len(self.notifs) == 0:
            if (time.time() - start_time >= timeout):
                break
            await asyncio.sleep(0.5)
        # the response is the oldest notification for the current characteristic obtained
        if len(self.notifs) == 0:
            raise TimeoutError
        response = self.notifs.pop(0)
        return response

    def __notification_handler(self, sender: int, data: bytearray) -> None:
        print(f"Notification received: {data}")
        self.notifs.append(data)

async def main() -> None:
    instruction: bytes = bytes.fromhex('fefe797d8800e2d2a6a5f74c45df0523a23f51d38308a997df6a8a58646fd3b6be5a')
    mac_address: str = "53:4e:48:00:00:07"

    # Connect to device
    bleak_client: BleakClient = BleakClient(mac_address)
    await bleak_client.connect()
    print('Connected to bleak client')

    # Get services and characteristics
    services: BleakGATTServiceCollection = await bleak_client.get_services()
    service: BleakGATTService = services.get_service(UUID(hex=UUIDs.service_uuid.value))
    if service is None:
        print('Service not found')
        return
    req_chara: Chara = Chara(service.get_characteristic(UUID(hex=UUIDs.req_uuid.value)), bleak_client)
    res_chara: Chara = Chara(service.get_characteristic(UUID(hex=UUIDs.res_uuid.value)), bleak_client)
    data_ctrl_chara: Chara = Chara(service.get_characteristic(UUID(hex=UUIDs.data_ctrl_uuid.value)), bleak_client)
    data_chara: Chara = Chara(service.get_characteristic(UUID(hex=UUIDs.data_uuid.value)), bleak_client)
    print('Service and characteristics discovered')

    # Send instruction
    await res_chara.subscribe()
    print('Started notification for response characteristic')
    await data_chara.subscribe()
    print('Started notification for data characteristic')
    await req_chara.write(instruction)
    print(f'Wrote {instruction.hex()} to request characteristic')
    print('Wait for response characteristic to get implant response')
    await asyncio.sleep(1)
    res = await res_chara.wait_notification(20)
    print(f"Full res of instruction: {res.hex()}")
    header, instruction_res = __extract_itp_header(res)
    print(f"Header res of instruction: {header}")
    print(f"Res of instruction: {instruction_res.hex()}")
    print(f'response: {instruction_res.decode("utf-8")}')
    await res_chara.unsubscribe()
    print('Stopped notification for response characteristic')
    await data_chara.unsubscribe()
    print('Stopped notification for data characteristic')
    await bleak_client.disconnect()
    print('Disconnected from bleak client')

if __name__ == '__main__':
    asyncio.run(main())

As for the discovery of services, using the example get_services(), just changing the mac address, I obtain the following:

root@iot-gate-imx8:/home/compulab/code/GatewaySmartphone# pipenv run python get_services.py
Services:

I did find when reading the bluetooth service logs

Aug 09 07:53:36 iot-gate-imx8 bluetoothd[505]: src/device.c:load_gatt_db() No cache for 53:4E:48:00:00:07

but I'm unsure if it's the related to my issue.

I'm also attaching the bleak logs I got during scan and get_services() as well as the bluetooth trafic captured with tshark in case it could help. bleak_log_comm.log wireshark_scan.log wireshark_comm.log bleak_log_scan.log