Merge pull request #381 from zxzxwu/offload

Support non-directed address generation offload
This commit is contained in:
zxzxwu
2024-01-28 22:08:32 +08:00
committed by GitHub
3 changed files with 20 additions and 8 deletions

View File

@@ -21,6 +21,7 @@ import functools
import json import json
import asyncio import asyncio
import logging import logging
import secrets
from contextlib import asynccontextmanager, AsyncExitStack, closing from contextlib import asynccontextmanager, AsyncExitStack, closing
from dataclasses import dataclass from dataclasses import dataclass
from collections.abc import Iterable from collections.abc import Iterable
@@ -994,12 +995,15 @@ class DeviceConfiguration:
irk = config.get('irk') irk = config.get('irk')
if irk: if irk:
self.irk = bytes.fromhex(irk) self.irk = bytes.fromhex(irk)
else: elif self.address != Address(DEVICE_DEFAULT_ADDRESS):
# Construct an IRK from the address bytes # Construct an IRK from the address bytes
# NOTE: this is not secure, but will always give the same IRK for the same # NOTE: this is not secure, but will always give the same IRK for the same
# address # address
address_bytes = bytes(self.address) address_bytes = bytes(self.address)
self.irk = (address_bytes * 3)[:16] self.irk = (address_bytes * 3)[:16]
else:
# Fallback - when both IRK and address are not set, randomly generate an IRK.
self.irk = secrets.token_bytes(16)
# Load advertising data # Load advertising data
advertising_data = config.get('advertising_data') advertising_data = config.get('advertising_data')
@@ -1581,6 +1585,16 @@ class Device(CompositeEventEmitter):
if self.address_resolution_offload: if self.address_resolution_offload:
await self.send_command(HCI_LE_Clear_Resolving_List_Command()) await self.send_command(HCI_LE_Clear_Resolving_List_Command())
# Add an empty entry for non-directed address generation.
await self.send_command(
HCI_LE_Add_Device_To_Resolving_List_Command(
peer_identity_address_type=Address.ANY.address_type,
peer_identity_address=Address.ANY,
peer_irk=bytes(16),
local_irk=self.irk,
)
)
for irk, address in resolving_keys: for irk, address in resolving_keys:
await self.send_command( await self.send_command(
HCI_LE_Add_Device_To_Resolving_List_Command( HCI_LE_Add_Device_To_Resolving_List_Command(
@@ -1681,8 +1695,8 @@ class Device(CompositeEventEmitter):
peer_address = target peer_address = target
peer_address_type = target.address_type peer_address_type = target.address_type
else: else:
peer_address = Address('00:00:00:00:00:00') peer_address = Address.ANY
peer_address_type = Address.PUBLIC_DEVICE_ADDRESS peer_address_type = Address.ANY.address_type
# Set the advertising parameters # Set the advertising parameters
await self.send_command( await self.send_command(

View File

@@ -1993,10 +1993,8 @@ class Manager(EventEmitter):
) -> None: ) -> None:
# Store the keys in the key store # Store the keys in the key store
if self.device.keystore and identity_address is not None: if self.device.keystore and identity_address is not None:
self.device.abort_on( # Make sure on_pairing emits after key update.
'flush', self.device.update_keys(str(identity_address), keys) await self.device.update_keys(str(identity_address), keys)
)
# Notify the device # Notify the device
self.device.on_pairing(session.connection, identity_address, keys, session.sc) self.device.on_pairing(session.connection, identity_address, keys, session.sc)

View File

@@ -99,7 +99,7 @@ development =
types-protobuf >= 4.21.0 types-protobuf >= 4.21.0
avatar = avatar =
pandora-avatar == 0.0.5 pandora-avatar == 0.0.5
rootcanal == 1.4.0 ; python_version>='3.10' rootcanal == 1.6.0 ; python_version>='3.10'
documentation = documentation =
mkdocs >= 1.4.0 mkdocs >= 1.4.0
mkdocs-material >= 8.5.6 mkdocs-material >= 8.5.6