Skip to content

Implement extended advertising emulation #827

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zxzxwu merged 2 commits into from
Dec 1, 2025
Merged

Implement extended advertising emulation #827

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d2a4c2a
Implement extended advertising emulation
zxzxwu Nov 27, 2025
a84f027
Refactor LE emulation with LL and Air Interface
zxzxwu Nov 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
852 changes: 566 additions & 286 deletions bumble/controller.py
View file
Open in desktop

Large diffs are not rendered by default.

30 changes: 24 additions & 6 deletions bumble/device.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,12 +265,22 @@ def from_advertising_report(

# -----------------------------------------------------------------------------
class AdvertisementDataAccumulator:
last_advertisement: Advertisement | None
last_data: bytes
passive: bool

def __init__(self, passive: bool = False):
self.passive = passive
self.last_advertisement = None
self.last_data = b''

def update(self, report):
def update(
self,
report: (
hci.HCI_LE_Advertising_Report_Event.Report
| hci.HCI_LE_Extended_Advertising_Report_Event.Report
),
) -> Advertisement | None:
advertisement = Advertisement.from_advertising_report(report)
if advertisement is None:
return None
Expand All @@ -283,10 +293,12 @@ def update(self, report):
and not self.last_advertisement.is_scan_response
):
# This is the response to a scannable advertisement
result = Advertisement.from_advertising_report(report)
result.is_connectable = self.last_advertisement.is_connectable
result.is_scannable = True
result.data = AdvertisingData.from_bytes(self.last_data + report.data)
if result := Advertisement.from_advertising_report(report):
result.is_connectable = self.last_advertisement.is_connectable
result.is_scannable = True
result.data = AdvertisingData.from_bytes(
self.last_data + report.data
)
self.last_data = b''
else:
if (
Expand Down Expand Up @@ -3333,7 +3345,13 @@ def is_scanning(self):
return self.scanning

@host_event_handler
def on_advertising_report(self, report):
def on_advertising_report(
self,
report: (
hci.HCI_LE_Advertising_Report_Event.Report
| hci.HCI_LE_Extended_Advertising_Report_Event.Report
),
) -> None:
if not (accumulator := self.advertisement_accumulators.get(report.address)):
accumulator = AdvertisementDataAccumulator(passive=self.scanning_is_passive)
self.advertisement_accumulators[report.address] = accumulator
Expand Down
183 changes: 25 additions & 158 deletions bumble/link.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,21 +19,19 @@
# Imports
# -----------------------------------------------------------------------------
import logging
from typing import Optional
from typing import TYPE_CHECKING, Optional

from bumble import controller, core, hci, lmp
from bumble import core, hci, ll, lmp

if TYPE_CHECKING:
from bumble import controller

# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logger = logging.getLogger(__name__)


# -----------------------------------------------------------------------------
# Utils
# -----------------------------------------------------------------------------


# -----------------------------------------------------------------------------
# TODO: add more support for various LL exchanges
# (see Vol 6, Part B - 2.4 DATA CHANNEL PDU)
Expand All @@ -47,7 +45,6 @@ class LocalLink:

def __init__(self):
self.controllers = set()
self.pending_connection = None
self.pending_classic_connection = None

############################################################
Expand All @@ -61,10 +58,11 @@ def add_controller(self, controller: controller.Controller):
def remove_controller(self, controller: controller.Controller):
self.controllers.remove(controller)

def find_controller(self, address: hci.Address) -> controller.Controller | None:
def find_le_controller(self, address: hci.Address) -> controller.Controller | None:
for controller in self.controllers:
if controller.random_address == address:
return controller
for connection in controller.le_connections.values():
if connection.self_address == address:
return controller
return None

def find_classic_controller(
Expand All @@ -75,22 +73,13 @@ def find_classic_controller(
return controller
return None

def get_pending_connection(self):
return self.pending_connection

############################################################
# LE handlers
############################################################

def on_address_changed(self, controller):
pass

def send_advertising_data(self, sender_address: hci.Address, data: bytes):
# Send the advertising data to all controllers, except the sender
for controller in self.controllers:
if controller.random_address != sender_address:
controller.on_link_advertising_data(sender_address, data)

def send_acl_data(
self,
sender_controller: controller.Controller,
Expand All @@ -100,7 +89,7 @@ def send_acl_data(
):
# Send the data to the first controller with a matching address
if transport == core.PhysicalTransport.LE:
destination_controller = self.find_controller(destination_address)
destination_controller = self.find_le_controller(destination_address)
source_address = sender_controller.random_address
elif transport == core.PhysicalTransport.BR_EDR:
destination_controller = self.find_classic_controller(destination_address)
Expand All @@ -115,152 +104,30 @@ def send_acl_data(
)
)

def on_connection_complete(self) -> None:
# Check that we expect this call
if not self.pending_connection:
logger.warning('on_connection_complete with no pending connection')
return

central_address, le_create_connection_command = self.pending_connection
self.pending_connection = None

# Find the controller that initiated the connection
if not (central_controller := self.find_controller(central_address)):
logger.warning('!!! Initiating controller not found')
return

# Connect to the first controller with a matching address
if peripheral_controller := self.find_controller(
le_create_connection_command.peer_address
):
central_controller.on_link_peripheral_connection_complete(
le_create_connection_command, hci.HCI_SUCCESS
)
peripheral_controller.on_link_central_connected(central_address)
return

# No peripheral found
central_controller.on_link_peripheral_connection_complete(
le_create_connection_command, hci.HCI_CONNECTION_ACCEPT_TIMEOUT_ERROR
)

def connect(
def send_advertising_pdu(
self,
central_address: hci.Address,
le_create_connection_command: hci.HCI_LE_Create_Connection_Command,
sender_controller: controller.Controller,
packet: ll.AdvertisingPdu,
):
logger.debug(
f'$$$ CONNECTION {central_address} -> '
f'{le_create_connection_command.peer_address}'
)
self.pending_connection = (central_address, le_create_connection_command)
asyncio.get_running_loop().call_soon(self.on_connection_complete)
loop = asyncio.get_running_loop()
for c in self.controllers:
if c != sender_controller:
loop.call_soon(c.on_ll_advertising_pdu, packet)

def on_disconnection_complete(
def send_ll_control_pdu(
self,
initiating_address: hci.Address,
target_address: hci.Address,
disconnect_command: hci.HCI_Disconnect_Command,
sender_address: hci.Address,
receiver_address: hci.Address,
packet: ll.ControlPdu,
):
# Find the controller that initiated the disconnection
if not (initiating_controller := self.find_controller(initiating_address)):
logger.warning('!!! Initiating controller not found')
return

# Disconnect from the first controller with a matching address
if target_controller := self.find_controller(target_address):
target_controller.on_link_disconnected(
initiating_address, disconnect_command.reason
if not (receiver_controller := self.find_le_controller(receiver_address)):
raise core.InvalidArgumentError(
f"Unable to find controller for address {receiver_address}"
)

initiating_controller.on_link_disconnection_complete(
disconnect_command, hci.HCI_SUCCESS
)

def disconnect(
self,
initiating_address: hci.Address,
target_address: hci.Address,
disconnect_command: hci.HCI_Disconnect_Command,
):
logger.debug(
f'$$$ DISCONNECTION {initiating_address} -> '
f'{target_address}: reason = {disconnect_command.reason}'
)
asyncio.get_running_loop().call_soon(
lambda: self.on_disconnection_complete(
initiating_address, target_address, disconnect_command
)
lambda: receiver_controller.on_ll_control_pdu(sender_address, packet)
)

def on_connection_encrypted(
self,
central_address: hci.Address,
peripheral_address: hci.Address,
rand: bytes,
ediv: int,
ltk: bytes,
):
logger.debug(f'*** ENCRYPTION {central_address} -> {peripheral_address}')

if central_controller := self.find_controller(central_address):
central_controller.on_link_encrypted(peripheral_address, rand, ediv, ltk)

if peripheral_controller := self.find_controller(peripheral_address):
peripheral_controller.on_link_encrypted(central_address, rand, ediv, ltk)

def create_cis(
self,
central_controller: controller.Controller,
peripheral_address: hci.Address,
cig_id: int,
cis_id: int,
) -> None:
logger.debug(
f'$$$ CIS Request {central_controller.random_address} -> {peripheral_address}'
)
if peripheral_controller := self.find_controller(peripheral_address):
asyncio.get_running_loop().call_soon(
peripheral_controller.on_link_cis_request,
central_controller.random_address,
cig_id,
cis_id,
)

def accept_cis(
self,
peripheral_controller: controller.Controller,
central_address: hci.Address,
cig_id: int,
cis_id: int,
) -> None:
logger.debug(
f'$$$ CIS Accept {peripheral_controller.random_address} -> {central_address}'
)
if central_controller := self.find_controller(central_address):
loop = asyncio.get_running_loop()
loop.call_soon(central_controller.on_link_cis_established, cig_id, cis_id)
loop.call_soon(
peripheral_controller.on_link_cis_established, cig_id, cis_id
)

def disconnect_cis(
self,
initiator_controller: controller.Controller,
peer_address: hci.Address,
cig_id: int,
cis_id: int,
) -> None:
logger.debug(
f'$$$ CIS Disconnect {initiator_controller.random_address} -> {peer_address}'
)
if peer_controller := self.find_controller(peer_address):
loop = asyncio.get_running_loop()
loop.call_soon(
initiator_controller.on_link_cis_disconnected, cig_id, cis_id
)
loop.call_soon(peer_controller.on_link_cis_disconnected, cig_id, cis_id)

############################################################
# Classic handlers
############################################################
Expand Down
Loading
Loading