pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

add broadcast code encoding

Browse Source
This commit is contained in:
Gilles Boccon-Gibod
2025-03-17 19:56:02 -04:00
parent b2d9541f8f
commit 5e55c0e358
3 changed files with 39 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
apps/auracast.py
View File

@@ -99,6 +99,25 @@ def codec_config_string(
return '\n'.join(indent + line for line in lines) return '\n'.join(indent + line for line in lines)
def broadcast_code_bytes(broadcast_code: str) -> bytes:
"""
Convert a broadcast code string to a 16-byte value.
If `broadcast_code` is `0x` followed by 32 hex characters, it is interpreted as a
raw 16-byte raw broadcast code in big-endian byte order.
Otherwise, `broadcast_code` is converted to a 16-byte value as specified in
BLUETOOTH CORE SPECIFICATION Version 6.0 | Vol 3, Part C , section 3.2.6.3
"""
if broadcast_code.startswith("0x") and len(broadcast_code) == 34:
return bytes.fromhex(broadcast_code[2:])[::-1]
broadcast_code_utf8 = broadcast_code.encode("utf-8")
if len(broadcast_code_utf8) > 16:
raise ValueError("broadcast code must be <= 16 bytes in utf-8 encoding")
padding = bytes(16 - len(broadcast_code_utf8))
return broadcast_code_utf8 + padding
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
# Scan For Broadcasts # Scan For Broadcasts
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
@@ -234,22 +253,14 @@ class BroadcastScanner(pyee.EventEmitter):
if self.biginfo: if self.biginfo:
print(color(' BIG:', 'cyan')) print(color(' BIG:', 'cyan'))
print( print(color(' Number of BIS:', 'magenta'), self.biginfo.num_bis)
color(' Number of BIS:', 'magenta'), print(color(' ISO Interval: ', 'magenta'), self.biginfo.iso_interval)
self.biginfo.num_bis, print(color(' Max PDU: ', 'magenta'), self.biginfo.max_pdu)
) print(color(' SDU Interval: ', 'magenta'), self.biginfo.sdu_interval)
print( print(color(' Max SDU: ', 'magenta'), self.biginfo.max_sdu)
color(' PHY: ', 'magenta'), print(color(' PHY: ', 'magenta'), self.biginfo.phy.name)
self.biginfo.phy.name, print(color(' Framed: ', 'magenta'), self.biginfo.framed)
) print(color(' Encrypted: ', 'magenta'), self.biginfo.encrypted)
print(
color(' Framed: ', 'magenta'),
self.biginfo.framed,
)
print(
color(' Encrypted: ', 'magenta'),
self.biginfo.encrypted,
)
def on_sync_establishment(self) -> None: def on_sync_establishment(self) -> None:
self.emit('sync_establishment') self.emit('sync_establishment')
@@ -702,14 +713,13 @@ async def run_receive(
def on_change() -> None: def on_change() -> None:
if ( if (
broadcast.basic_audio_announcement broadcast.basic_audio_announcement and broadcast.biginfo
and not basic_audio_announcement_scanned.is_set() ) and not basic_audio_announcement_scanned.is_set():
):
basic_audio_announcement_scanned.set() basic_audio_announcement_scanned.set()
broadcast.on('change', on_change) broadcast.on('change', on_change)
if not broadcast.basic_audio_announcement: if not broadcast.basic_audio_announcement or not broadcast.biginfo:
print('Wait for Basic Audio Announcement...') print('Wait for Basic Audio Announcement and BIG Info...')
await basic_audio_announcement_scanned.wait() await basic_audio_announcement_scanned.wait()
print('Basic Audio Announcement found') print('Basic Audio Announcement found')
broadcast.print() broadcast.print()
@@ -730,7 +740,7 @@ async def run_receive(
big_sync_timeout=0x4000, big_sync_timeout=0x4000,
bis=[bis.index for bis in subgroup.bis], bis=[bis.index for bis in subgroup.bis],
broadcast_code=( broadcast_code=(
bytes.fromhex(broadcast_code) if broadcast_code else None broadcast_code_bytes(broadcast_code) if broadcast_code else None
), ),
), ),
) )
@@ -944,7 +954,7 @@ async def run_transmit(
max_transport_latency=65, max_transport_latency=65,
rtn=4, rtn=4,
broadcast_code=( broadcast_code=(
bytes.fromhex(broadcast_code) if broadcast_code else None broadcast_code_bytes(broadcast_code) if broadcast_code else None
), ),
), ),
) )
@@ -1088,7 +1098,7 @@ def pair(ctx, transport, address):
'--broadcast-code', '--broadcast-code',
metavar='BROADCAST_CODE', metavar='BROADCAST_CODE',
type=str, type=str,
help='Broadcast encryption code in hex format', help='Broadcast encryption code (string or raw hex format prefixed with 0x)',
) )
@click.option( @click.option(
'--sync-timeout', '--sync-timeout',

2
bumble/profiles/pbp.py
View File

@@ -40,7 +40,7 @@ class PublicBroadcastAnnouncement:
def from_bytes(cls, data: bytes) -> Self: def from_bytes(cls, data: bytes) -> Self:
features = cls.Features(data[0]) features = cls.Features(data[0])
metadata_length = data[1] metadata_length = data[1]
metadata_ltv = data[1 : 1 + metadata_length] metadata_ltv = data[2 : 2 + metadata_length]
return cls( return cls(
features=features, metadata=le_audio.Metadata.from_bytes(metadata_ltv) features=features, metadata=le_audio.Metadata.from_bytes(metadata_ltv)
) )

3
bumble/transport/common.py
View File

@@ -302,7 +302,10 @@ class ParserSource(BaseSource):
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
class StreamPacketSource(asyncio.Protocol, ParserSource): class StreamPacketSource(asyncio.Protocol, ParserSource):
def data_received(self, data: bytes) -> None: def data_received(self, data: bytes) -> None:
try:
self.parser.feed_data(data) self.parser.feed_data(data)
except core.InvalidPacketError:
logger.warning("invalid packet, ignoring data")
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------