diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2fd9b4f7..ec09f124 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -855,7 +855,7 @@ jobs: PeerStreamSession.swift ReticulumManager.swift mesh/Mesh.swift ble/MeshConnectionRegistry.swift \ ble/BleDiscoveryBootstrapPolicy.swift \ PeripheralRestorationAgeOutPolicy.swift \ - BleMessageNotificationPolicy.swift \ + BleHelloPolicy.swift BleMessageNotificationPolicy.swift \ BleAppTag.swift BleDensityPolicy.swift BleServiceInstanceSelection.swift \ Generated/offline_protocol.swift ForcedPresenceCheckQueue.swift \ AddressDeclarationPolicy.swift \ diff --git a/CHANGELOG.md b/CHANGELOG.md index bae5a975..89fb646e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -248,6 +248,56 @@ archived by series under [docs/changelog/](docs/changelog/); see the peripheral when a central is subscribed. The peripheral alone takes nothing while no central is subscribed, where a notification reached no one. No board run has been done yet. +- **An Android phone and a Mac running the Python service start a session + over Bluetooth LE.** They found and verified each other and never formed a + session: the Mac rotates its Bluetooth address, the phone kept a link to + each address, the Mac's peripheral then saw two centrals and attributed + every write to the placeholder `ble-peer`, and the core refused the phone's + key package as a transport identity mismatch (`TRANSPORT_IDENTITY_MISMATCH`). + The Python peripheral now names the central behind each write on macOS too, + serves the optional Hello characteristic, binds a central to the address its + hello proves and hands the core that central's fragments under that address. + The Python central writes its hello to a peer that serves one, bounded by + the characteristic's one-packet payload (bleak's BlueZ client reports a + 23-byte MTU, which skipped every hello from Linux), and announces the peer + only once the hello is answered, so the key package the announce pushes + never reaches the peer ahead of it. The iOS central now writes its hello + too, once per connection, after its Message subscription and before its + handshake reads (`BleHelloPolicy`). Without it an iPhone could + verify a Python box and never start a session through the link it opened. A + link bound by a hello is never rebound, and a message's `sender` field never + replaces the binding. The iOS hello is written only when it fits one ATT + packet, since iOS sends a longer with-response write as a prepared write, + which a peripheral refuses for a hello. A central that says hello and never + subscribes has its binding dropped after 5 seconds, and its peer reported + lost unless another link reaches it. Two Macs and a Galaxy M36 (Android 16) + were run; no iPhone was. +- **Android and the Python central keep one client link per peer.** A second + link that verifies as a peer already held (a peripheral advertising under an + old and a new address) is closed unannounced, and both leave that address + undialed for a minute, or until the kept link is gone. Kept, two such peers + filled Android's four connection slots, and the cap then refused every + inbound central. +- **On macOS a peripheral notification reaches its recipient, and survives a + full transmit queue.** A fragment for a peer whose central said hello goes to + that central alone rather than to every subscriber, and a notification + CoreBluetooth refuses because its queue is full is retried once the stack + says it is ready. bless ignored the refusal, so a burst (a message, its + receipt, a key package) lost fragments and waited for the core's next retry. +- **A dead Bluetooth LE link is dropped within about 20 seconds.** + CoreBluetooth kept a link to a Mac whose service had restarted "connected" + for over five minutes, and every frame written into it vanished. The Python + central now reads each link's Device id back every 15 seconds and drops a + link whose read fails, takes more than 5 seconds, or names another address, + reporting the peer lost unless its central is still bound on our own + peripheral, which still reaches it. +- **A Mac that may not use Bluetooth no longer wedges the service.** Started + where macOS cannot grant Bluetooth (over ssh), the Bluetooth LE peripheral + blocked the service inside bless's constructor for good: no API socket, and + the 30 second transport start deadline never fired, because the block was on + the event loop itself. The peripheral now refuses to start when macOS reports + Bluetooth denied or restricted, builds its server off the loop with a 20 + second deadline otherwise, and the service runs on its other transports. - **A resend is offered to the mesh when no carrier reaches the recipient.** Only a message's first send was handed to neighbours when no carrier took it. A message whose first attempt went into a direct link that had already diff --git a/bindings/python/offline_protocol_sdk/ble_manager.py b/bindings/python/offline_protocol_sdk/ble_manager.py index 727a4eee..ac3cd214 100644 --- a/bindings/python/offline_protocol_sdk/ble_manager.py +++ b/bindings/python/offline_protocol_sdk/ble_manager.py @@ -45,6 +45,12 @@ MESSAGE_CHAR_UUID = "6e400002-b5a3-f393-e0a9-e50e24dcca9e" DEVICE_ID_CHAR_UUID = "6e400003-b5a3-f393-e0a9-e50e24dcca9e" IDENTITY_CHAR_UUID = "6e400004-b5a3-f393-e0a9-e50e24dcca9e" +# Optional: a central writes its identity assertion here so the peripheral +# can bind the link to its address (docs/spec/ble-framing.md, Hello). +HELLO_CHAR_UUID = "6e400006-b5a3-f393-e0a9-e50e24dcca9e" +# The chapter's bounds on a hello value, checked by both roles. +HELLO_MIN_LEN = 96 +HELLO_MAX_LEN = 512 MAX_FRAGMENT_SIZE = 185 CONNECTION_TIMEOUT = 10.0 # seconds @@ -59,6 +65,21 @@ # the global per-minute connection budget that verifiable peers need. REFUSAL_BACKOFF_MAX = 600.0 # seconds PEER_LOST_TIMEOUT = 30.0 # seconds since last seen +# How often each connected link is read back, and how long the read may take. +# CoreBluetooth can keep a link "connected" for minutes after the peer's +# process (and its GATT service) is gone, and fragments are written without +# response, so nothing else ever fails on such a link. +LIVENESS_INTERVAL = 15.0 # seconds +# How long an address refused as a second link to a peer already held is left +# undialed, unless the kept link dies first. A rotating peripheral advertises +# under both addresses for a while, and without this the address was redialed, +# verified and refused again every cooldown. +DUPLICATE_LINK_SUPPRESS = 60.0 # seconds +LIVENESS_READ_TIMEOUT = 5.0 # seconds +# How long the hello's write response may take. The announce waits on it, so +# a peer that never answers must not hold the link unannounced; Android's +# hello watchdog has the same bound. +HELLO_WRITE_TIMEOUT = 3.0 # seconds SCAN_RESTART_INTERVAL = 30.0 @@ -116,6 +137,12 @@ def __init__(self, protocol: Any, device_id: str) -> None: self._connecting: set[str] = set() # address -> consecutive identity refusals; each doubles the cooldown self._refusals: dict[str, int] = {} + # address -> (peer it duplicated, monotonic time the refusal lapses) + self._duplicates: dict[str, tuple[str, float]] = {} + # Set by ProtocolManager: whether another link (a central bound on our + # own peripheral) still reaches a peer, so a client link going away + # does not report lost a peer the core can still route to. + self.peer_linked_elsewhere: Any = None # Metrics self._bytes_sent: int = 0 @@ -214,6 +241,7 @@ async def stop(self) -> None: self._device_id_to_addr.clear() self._last_seen.clear() self._connecting.clear() + self._duplicates.clear() try: self._protocol.ble_status_changed(is_available=False) @@ -288,6 +316,9 @@ def _on_advertisement( def _should_connect_locked(self, addr: str, now: float) -> bool: """Adaptive rate-limiting. Caller must hold ``_lock``.""" + if self._suppressed_duplicate_locked(addr, now): + return False + # Per-peripheral cooldown last_attempt = self._connection_attempts.get(addr, 0) if now - last_attempt < self._cooldown_locked(addr): @@ -301,6 +332,21 @@ def _should_connect_locked(self, addr: str, now: float) -> bool: return True + def _suppressed_duplicate_locked(self, addr: str, now: float) -> bool: + """Whether ``addr`` is inside a duplicate-link refusal. Caller must + hold ``_lock``. Lapses early once the kept link is gone, so a peer + whose kept link died is redialed at the address it still advertises.""" + entry = self._duplicates.get(addr) + if entry is None: + return False + peer, until = entry + held = self._device_id_to_addr.get(peer) + kept = held is not None and held != addr and held in self._clients + if now < until and kept: + return True + self._duplicates.pop(addr, None) + return False + def _cooldown_locked(self, addr: str) -> float: """Per-peripheral cooldown, doubled per consecutive identity refusal and capped at ``REFUSAL_BACKOFF_MAX``. Caller must hold ``_lock``.""" @@ -357,12 +403,49 @@ async def _connect_to_peer(self, device: BLEDevice) -> None: await client.disconnect() return + # A peripheral that rotates its address (macOS) advertises under + # the old and the new one at once, and both verify as one peer. A + # second client link to it buys nothing and makes the address map + # point at whichever verified last, so the first link is kept. + with self._lock: + held = self._device_id_to_addr.get(device_id) + duplicate = ( + held is not None + and held != addr + and held in self._clients + and self._clients[held].is_connected + ) + if duplicate: + with self._lock: + self._duplicates[addr] = ( + device_id, + time.monotonic() + DUPLICATE_LINK_SUPPRESS, + ) + self._emit_diagnostic("info", "Closed a second link to a peer already held", { + "address": addr, + "device_id": device_id, + "held_at": held, + }) + await client.disconnect() + return + with self._lock: self._refusals.pop(addr, None) self._clients[addr] = client self._peer_device_ids[addr] = device_id self._device_id_to_addr[device_id] = addr + # Subscribe to message notifications + await self._subscribe_to_messages(client, addr, device_id) + + # Say who we are on this link, after the subscription and before + # any Message write, as the chapter orders it. The announce below + # is what makes the core queue for this peer (it pushes a key + # package on it), so the hello goes first: a key package written + # ahead of it reaches a peripheral that cannot name this link and + # is refused as a transport identity mismatch. + await self._write_hello(client, addr) + # Announce the peer under the address it proved try: rssi = -70 # approximate; bleak doesn't expose RSSI post-connect @@ -370,9 +453,6 @@ async def _connect_to_peer(self, device: BLEDevice) -> None: except Exception: logger.debug("ble_peer_discovered failed for %s", device_id) - # Subscribe to message notifications - await self._subscribe_to_messages(client, addr, device_id) - self._emit_diagnostic("info", "Connected to peer", { "address": addr, "device_id": device_id, @@ -456,6 +536,94 @@ def on_notification(_sender: int, data: bytearray) -> None: "warning", f"Failed to subscribe to messages from {device_id}: {exc}" ) + async def _write_hello(self, client: BleakClient, addr: str) -> None: + """Write our identity assertion to the peer's Hello, when it has one. + + Without it a peripheral receiving our Message writes has no way to + tell which address they come from, and it refuses our key package, + whose sender must match the link's. Best effort, as the chapter + requires: no Hello, an assertion that does not fit one write, or a + refused write all leave the link as it was. + """ + try: + hello_char = client.services.get_characteristic(HELLO_CHAR_UUID) + except Exception: + return + if hello_char is None: + return + try: + assertion = bytes(self._protocol.identity_assertion([])) + except Exception as exc: + self._emit_diagnostic("debug", f"No identity assertion for a hello: {exc}") + return + if not HELLO_MIN_LEN <= len(assertion) <= HELLO_MAX_LEN: + return + # One ATT packet's payload, from the characteristic: bleak's BlueZ + # client reports a 23-byte MTU unless it was explicitly acquired, which + # bounded every Linux hello to 20 bytes so none was ever written. The + # characteristic reads BlueZ's MTU property (and on CoreBluetooth + # maximumWriteValueLength(.withoutResponse)), and a longer value would + # go out as a prepared write, which a peripheral refuses for a hello. + try: + payload_limit = int(hello_char.max_write_without_response_size) + except Exception: + payload_limit = 0 + if payload_limit and len(assertion) > payload_limit: + self._emit_diagnostic("debug", "Hello does not fit the MTU; skipped", { + "address": addr, + "length": len(assertion), + "mtu_payload": payload_limit, + }) + return + try: + await asyncio.wait_for( + client.write_gatt_char(HELLO_CHAR_UUID, assertion, response=True), + HELLO_WRITE_TIMEOUT, + ) + except Exception as exc: + self._emit_diagnostic("debug", f"Hello write to {addr} refused: {exc!r}") + + async def _probe_link(self, addr: str, client: BleakClient, device_id: str) -> bool: + """Read the peer's Device id back; False when the link is dead. + + A link whose peer process restarted can stay connected in + CoreBluetooth for minutes while every write into it vanishes. A read + of a characteristic that no longer exists, or that now names another + address, fails or times out, which this turns into a disconnect. + """ + try: + value = await asyncio.wait_for( + client.read_gatt_char(DEVICE_ID_CHAR_UUID), LIVENESS_READ_TIMEOUT + ) + return bytes(value).decode("utf-8").strip("\x00") == device_id + except Exception: + return False + + async def _check_liveness(self) -> None: + with self._lock: + links = [ + (addr, client, self._peer_device_ids[addr]) + for addr, client in self._clients.items() + if addr in self._peer_device_ids + ] + if not links: + return + results = await asyncio.gather( + *(self._probe_link(addr, client, device_id) for addr, client, device_id in links) + ) + for (addr, client, device_id), alive in zip(links, results): + if alive: + continue + self._emit_diagnostic("info", "Link failed its liveness read; dropping it", { + "address": addr, + "device_id": device_id, + }) + try: + await asyncio.wait_for(client.disconnect(), LIVENESS_READ_TIMEOUT) + except Exception: + logger.debug("disconnect after a failed liveness read failed", exc_info=True) + await self._on_peer_disconnected(addr) + def _on_fragment_received(self, sender_id: str, fragment: bytes) -> None: """Handle an incoming BLE fragment from a peer.""" with self._lock: @@ -476,18 +644,31 @@ async def _on_peer_disconnected(self, addr: str) -> None: if device_id is not None: self._device_id_to_addr.pop(device_id, None) - # A link that never proved a peer was never announced, so there is - # nothing to report lost. + # A link that never proved a peer was never announced, and one already + # torn down (a liveness drop runs this before the stack's own + # disconnect callback does) was reported then: nothing to report. if device_id is None: - self._emit_diagnostic("debug", f"Unverified link closed: {addr}") + self._emit_diagnostic("debug", f"Link closed with no proved peer on file: {addr}") return + self._emit_diagnostic("info", f"Peer disconnected: {device_id}") + if self._linked_elsewhere(device_id): + return try: self._protocol.ble_peer_lost(peer_id=device_id) except Exception: logger.debug("ble_peer_lost failed for %s", device_id) - self._emit_diagnostic("info", f"Peer disconnected: {device_id}") + def _linked_elsewhere(self, peer_id: str) -> bool: + """Whether another link of ours (the peripheral's) still reaches + ``peer_id``. Never raises: a probe that fails says no.""" + probe = self.peer_linked_elsewhere + if probe is None: + return False + try: + return bool(probe(peer_id)) + except Exception: + return False # -- outgoing fragment handling ------------------------------------------- @@ -562,6 +743,11 @@ async def write_fragment(self, recipient: str, data: bytes) -> bool: ) return True + def holds_peer(self, peer_id: str) -> bool: + """Whether a connected client link of ours has proved ``peer_id``.""" + client = self._find_client_for_peer(peer_id) + return client is not None and bool(client.is_connected) + def _find_client_for_peer(self, peer_id: str) -> BleakClient | None: """Find the BleakClient for a given device ID (O(1) via reverse map).""" with self._lock: @@ -576,7 +762,11 @@ async def _peer_cleanup_loop(self) -> None: """Periodically remove stale peers that haven't been seen recently.""" try: while True: - await asyncio.sleep(PEER_LOST_TIMEOUT / 2) + await asyncio.sleep(LIVENESS_INTERVAL) + try: + await self._check_liveness() + except Exception: + logger.debug("liveness check failed", exc_info=True) now = time.monotonic() stale_device_ids: list[str] = [] with self._lock: @@ -596,6 +786,8 @@ async def _peer_cleanup_loop(self) -> None: stale_device_ids.append(device_id) # Notify protocol outside the lock for device_id in stale_device_ids: + if self._linked_elsewhere(device_id): + continue try: self._protocol.ble_peer_lost(peer_id=device_id) except Exception: diff --git a/bindings/python/offline_protocol_sdk/ble_peripheral.py b/bindings/python/offline_protocol_sdk/ble_peripheral.py index 8fc7a270..0eb72e93 100644 --- a/bindings/python/offline_protocol_sdk/ble_peripheral.py +++ b/bindings/python/offline_protocol_sdk/ble_peripheral.py @@ -31,7 +31,7 @@ import sys import threading import time -from typing import Any +from typing import Any, Callable # bless is not installed on Windows (pyproject.toml leaves it out there: it # has no Windows backend), and this module is imported by the package's @@ -51,10 +51,14 @@ from .ble_manager import ( DEVICE_ID_CHAR_UUID, + HELLO_CHAR_UUID, + HELLO_MAX_LEN, + HELLO_MIN_LEN, IDENTITY_CHAR_UUID, MESSAGE_CHAR_UUID, SERVICE_UUID, ) +from .offline_protocol import verify_identity_assertion from .transport_manager import TransportError, TransportManager, TransportState logger = logging.getLogger(__name__) @@ -66,6 +70,192 @@ # How often to poll the bless delegate for subscription changes. _PEER_MONITOR_INTERVAL = 1.0 +# How long building the GATT server may take. bless's CoreBluetooth delegate +# blocks in its constructor, with no deadline, until the adapter reports +# powered on, which never happens when the process is not authorised to use +# Bluetooth. The local API server gives each transport 30 s to start; this is +# shorter so the refusal is this transport's own, with a reason. +_SERVER_INIT_TIMEOUT = 20.0 + +# ATT error codes a peripheral answers a refused write with (Core spec, Vol 3 +# Part F, 3.4.1.1). CoreBluetooth's CBATTError values are these numbers. +_ATT_SUCCESS = 0 +_ATT_INVALID_OFFSET = 0x07 +_ATT_INVALID_ATTRIBUTE_VALUE_LENGTH = 0x0D + +# A notification CoreBluetooth refused because its transmit queue was full is +# retried once the stack says it is ready again, this many times, waiting at +# most this long each. bless drops the refusal on the floor. +_NOTIFY_RETRIES = 4 +_NOTIFY_READY_TIMEOUT = 1.0 + +# A central that said hello but is not subscribed is given this long to +# subscribe before its binding is dropped and its peer reported lost. Without +# it a central that writes a hello and disconnects without ever subscribing +# (the monitor tracks subscribers only) left its binding, and a route to the +# address it proved, behind for the life of the process. +_UNSUBSCRIBED_HELLO_GRACE = 5.0 + +_ATTRIBUTING_DELEGATE: Any = None + + +def _prepared_hello_refusal(writes: list[tuple[str, int]]) -> int | None: + """The ATT code that refuses a batch carrying a hello, or None. + + ``writes`` is each request's ``(characteristic UUID, offset)``. + CoreBluetooth hands a prepared (long) write to the delegate as one batch + of requests at rising offsets, and the chapter requires a peripheral to + refuse a prepared or offset hello before verifying anything. Handled + request by request, the batch's first chunk (offset 0, a plausible + length) reached the verifier before the second chunk's offset refused it. + """ + hello = HELLO_CHAR_UUID.lower() + for char_uuid, offset in writes: + if char_uuid.lower() == hello and (len(writes) > 1 or offset != 0): + return _ATT_INVALID_OFFSET + return None +_DELEGATE_SWAP_LOCK = threading.Lock() +# How long a construction waits for another one's delegate swap. A swap is +# held for a whole construction, and one that wedged (see +# _SERVER_INIT_TIMEOUT) never releases it, so a later start says so instead +# of waiting out the init deadline and blaming the adapter. +_DELEGATE_SWAP_WAIT = 2.0 + + +def _attributing_delegate_class() -> Any: + """bless's CoreBluetooth delegate, made to say WHICH central did what. + + bless hands a write to its callback as ``(characteristic, value)``, + dropping the ``CBCentral`` the request came from, and responds success + to every write. With two centrals connected (a phone, and a Mac's own + central, or the two addresses of one rotating peer) a write could not be + attributed, and the core refused the writer's key package because its + sender did not match the link. This subclass passes each request's + central, offset and value to ``attributed_write_func``, answers with the + ATT code that returns, keeps the ``CBCentral`` objects of subscribers so + a notification can be addressed to one of them, and reports the stack's + "ready to update subscribers" to ``ready_func``. + + Built once per process: an Objective-C class name can be registered only + once. None where bless's CoreBluetooth backend is not importable. + """ + global _ATTRIBUTING_DELEGATE + if _ATTRIBUTING_DELEGATE is not None: + return _ATTRIBUTING_DELEGATE + try: + import objc # type: ignore[import-not-found] + from bless.backends.corebluetooth.peripheral_manager_delegate import ( # type: ignore[import-not-found] + PeripheralManagerDelegate, + ) + except Exception: # pragma: no cover - only reachable off macOS + return None + + class OfflineProtocolPeripheralDelegate(PeripheralManagerDelegate): # type: ignore[misc, valid-type] + def peripheralManager_didReceiveWriteRequests_( # noqa: N802 + self, peripheral_manager: Any, requests: Any + ) -> None: + handler = getattr(self, "attributed_write_func", None) + result = _ATT_SUCCESS + if handler is not None: + try: + refusal = _prepared_hello_refusal([ + (str(r.characteristic().UUID().UUIDString()), int(r.offset())) + for r in requests + ]) + except Exception: # never let a raise skip the response + refusal = None + if refusal is not None: + peripheral_manager.respondToRequest_withResult_(requests[0], refusal) + return + for request in requests: + char_uuid = str(request.characteristic().UUID().UUIDString()) + raw = request.value() + value = bytes(raw) if raw is not None else b"" + if handler is None: + self.write_request_func(char_uuid, value) + continue + central = str(request.central().identifier().UUIDString()) + try: + result = int(handler(central, char_uuid, value, int(request.offset()))) + except Exception: # never let a raise skip the response + result = _ATT_SUCCESS + if result != _ATT_SUCCESS: + break + peripheral_manager.respondToRequest_withResult_(requests[0], result) + + def peripheralManager_central_didSubscribeToCharacteristic_( # noqa: N802 + self, peripheral_manager: Any, central: Any, characteristic: Any + ) -> None: + objc.super( + OfflineProtocolPeripheralDelegate, self + ).peripheralManager_central_didSubscribeToCharacteristic_( + peripheral_manager, central, characteristic + ) + centrals = getattr(self, "subscribed_centrals", None) + if centrals is not None: + centrals[str(central.identifier().UUIDString())] = central + + def peripheralManager_central_didUnsubscribeFromCharacteristic_( # noqa: N802 + self, peripheral_manager: Any, central: Any, characteristic: Any + ) -> None: + central_id = str(central.identifier().UUIDString()) + try: + objc.super( + OfflineProtocolPeripheralDelegate, self + ).peripheralManager_central_didUnsubscribeFromCharacteristic_( + peripheral_manager, central, characteristic + ) + except (KeyError, ValueError): + pass + centrals = getattr(self, "subscribed_centrals", None) + if centrals is not None and central_id not in self._central_subscriptions: + centrals.pop(central_id, None) + + def peripheralManagerIsReadyToUpdateSubscribers_( # noqa: N802 + self, peripheral_manager: Any + ) -> None: + ready = getattr(self, "ready_func", None) + if ready is not None: + try: + ready() + except Exception: + pass + + _ATTRIBUTING_DELEGATE = OfflineProtocolPeripheralDelegate + return _ATTRIBUTING_DELEGATE + + +def _bluetooth_authorization_refusal() -> str | None: + """Why this process may not use Bluetooth on macOS, or None if it may. + + A process macOS has denied (or one that cannot ask, such as a service + started over ssh) would otherwise block in bless's constructor forever. + ``NotDetermined`` is allowed through: building the manager is what makes + macOS ask, and the construction deadline bounds the wait for an answer. + """ + if sys.platform != "darwin": + return None + try: + from CoreBluetooth import ( # type: ignore[import-not-found] + CBManager, + CBManagerAuthorizationDenied, + CBManagerAuthorizationRestricted, + ) + except Exception: + return None + try: + state = CBManager.authorization() + except Exception: + return None + if state == CBManagerAuthorizationDenied: + return ( + "Bluetooth is denied to this process: allow it in System Settings > " + "Privacy & Security > Bluetooth, or start the service from a GUI " + "session (macOS cannot ask a process started over ssh)" + ) + if state == CBManagerAuthorizationRestricted: + return "Bluetooth is restricted on this Mac by a profile or parental control" + return None _BLUEZ_SERVICE = "org.bluez" _BLUEZ_DEVICE_IFACE = "org.bluez.Device1" _BLUEZ_CHAR_IFACE = "org.bluez.GattCharacteristic1" @@ -296,7 +486,27 @@ def __init__( # _BlueZCentrals). None on CoreBluetooth, whose bless delegate # tracks them itself. self._bluez_centrals: _BlueZCentrals | None = None - + # Central id -> the address its hello proved. A link bound here feeds + # its fragments to the core under that address, which is what lets a + # key package from it through the core's transport-sender check. + # Never rebound: a hello on a bound link is ignored. + self._hello_bindings: dict[str, str] = {} + # Central id -> when its bound hello was first seen without a + # subscription (see _UNSUBSCRIBED_HELLO_GRACE). Monitor loop only. + self._unsubscribed_since: dict[str, float] = {} + # Set by ProtocolManager: whether another link (our own central's) + # already announced an address, so a hello does not announce it twice + # and a central leaving does not report a peer lost that is still up. + self.peer_linked_elsewhere: Any = None + # CoreBluetooth only: the attributing delegate's subscriber map and + # the stack's "ready to update subscribers" signal. + self._subscribed_centrals: dict[str, Any] = {} + self._notify_ready: asyncio.Event | None = None + # The monitor loop's clock. A test substitutes one it advances itself: + # on Windows time.monotonic() ticks every 15.6 ms, and asyncio runs a + # timer due within that at once, so a test sleeping in milliseconds + # never saw a grace period elapse. + self._clock: Callable[[], float] = time.monotonic # -- TransportManager interface ------------------------------------------- def is_available(self) -> bool: @@ -355,13 +565,16 @@ async def start(self) -> None: }) try: + refusal = _bluetooth_authorization_refusal() + if refusal is not None: + raise TransportError(refusal) # Truncate name — CoreBluetooth limits advertisement data to ~28 # bytes, and service UUIDs also consume space. name = self._device_id[:10] - self._server = BlessServer(name=name) + self._server = await self._build_server(name) + self._install_attribution() await self._setup_gatt_server() - # Register read/write callbacks self._server.read_request_func = self._on_read self._server.write_request_func = self._on_write @@ -437,8 +650,11 @@ async def stop(self) -> None: resolved_snapshot = dict(self._central_to_user_id) self._connected_centrals.clear() self._central_to_user_id.clear() + self._hello_bindings.clear() + self._unsubscribed_since.clear() self._last_known_central = None - + self._subscribed_centrals = {} + self._notify_ready = None for central_uuid in centrals_snapshot: try: self._protocol.ble_peer_lost(peer_id=central_uuid) @@ -459,6 +675,107 @@ async def stop(self) -> None: self._update_state(TransportState.STOPPED) self._emit_diagnostic("info", "BLE peripheral stopped") + async def _build_server(self, name: str) -> Any: + """Construct the bless server off the event loop, with a deadline. + + The CoreBluetooth constructor can block indefinitely (see + ``_SERVER_INIT_TIMEOUT``), and a block on the loop thread freezes the + whole service: no API socket, no other transport, and no deadline of + the caller's can fire. The thread is a daemon, so one that never + returns cannot hold the process open at exit. It is handed the running + loop, since bless otherwise looks one up in a thread that has none. + + BlueZ's constructor does not block, and it schedules its setup as a + task on the loop, which is safe only from the loop's own thread, so + there it runs here. + """ + loop = asyncio.get_running_loop() + if sys.platform != "darwin": + return self._construct_server(name, loop) + done: asyncio.Future[Any] = loop.create_future() + + def settle(result: Any, exc: BaseException | None) -> None: + if done.done(): + return + if exc is not None: + done.set_exception(exc) + else: + done.set_result(result) + + def build() -> None: + try: + server = self._construct_server(name, loop) + except BaseException as exc: # noqa: BLE001 - handed to the loop + result, error = None, exc + else: + result, error = server, None + try: + loop.call_soon_threadsafe(settle, result, error) + except RuntimeError: + pass # the loop closed while the stack was deciding + + threading.Thread(target=build, name="ble-peripheral-init", daemon=True).start() + try: + return await asyncio.wait_for(done, _SERVER_INIT_TIMEOUT) + except asyncio.TimeoutError as exc: + raise TransportError( + f"the Bluetooth stack did not come up within {_SERVER_INIT_TIMEOUT:.0f} s " + "(not authorised, or the adapter is off)" + ) from exc + + @staticmethod + def _construct_server(name: str, loop: Any = None) -> Any: + """``BlessServer(name)``, with the attributing delegate on macOS. + + bless's CoreBluetooth server builds its delegate from a name in its + own module; it is pointed at the subclass for this one construction, + under a lock, and put back. + """ + if sys.platform != "darwin": + return BlessServer(name=name, loop=loop) + # The class is built under the lock too: an Objective-C class name can + # be registered once per process, and two constructions racing to + # build it would both try. + if not _DELEGATE_SWAP_LOCK.acquire(timeout=_DELEGATE_SWAP_WAIT): + raise TransportError( + "an earlier Bluetooth stack construction in this process never " + "returned; restart the service once Bluetooth is allowed" + ) + try: + delegate_cls = _attributing_delegate_class() + if delegate_cls is None: + return BlessServer(name=name, loop=loop) + import bless.backends.corebluetooth.server as cb_server # type: ignore[import-not-found] + + original = cb_server.PeripheralManagerDelegate + cb_server.PeripheralManagerDelegate = delegate_cls + try: + return BlessServer(name=name, loop=loop) + finally: + cb_server.PeripheralManagerDelegate = original + finally: + _DELEGATE_SWAP_LOCK.release() + + def _install_attribution(self) -> None: + """Point the attributing delegate, when present, at this peripheral.""" + delegate = getattr(self._server, "peripheral_manager_delegate", None) + if delegate is None or not hasattr(delegate, "peripheralManagerIsReadyToUpdateSubscribers_"): + return + if _ATTRIBUTING_DELEGATE is None or not isinstance(delegate, _ATTRIBUTING_DELEGATE): + return + loop = self._loop + self._notify_ready = asyncio.Event() + ready = self._notify_ready + self._subscribed_centrals = {} + delegate.subscribed_centrals = self._subscribed_centrals + delegate.attributed_write_func = self._on_attributed_write + + def on_ready() -> None: + if loop is not None and not loop.is_closed(): + loop.call_soon_threadsafe(ready.set) + + delegate.ready_func = on_ready + def _attach_bluez_centrals(self) -> None: """Start observing who calls our GATT application on the bus. @@ -539,6 +856,15 @@ async def _setup_gatt_server(self) -> None: GATTAttributePermissions.readable, ) + # HELLO characteristic: a central writes its assertion here, once, + # so its Message writes can be attributed to the address it proved. + await self._server.add_new_characteristic( + SERVICE_UUID, + HELLO_CHAR_UUID, + GATTCharacteristicProperties.write, + None, + GATTAttributePermissions.writeable, + ) # -- GATT callbacks ------------------------------------------------------- @staticmethod @@ -580,33 +906,117 @@ def _on_write(self, char_uuid: Any, value: Any) -> None: connection**. Therefore the entire body is wrapped in try/except. """ try: - resolved = self._resolve_char_uuid(char_uuid) - logger.debug( - "_on_write called: uuid=%s, value_type=%s, value_len=%d", - resolved, type(value).__name__, len(value) if value else 0, + self._handle_write( + self._resolve_sender(), + self._resolve_char_uuid(char_uuid), + bytes(value) if value is not None else b"", + 0, ) - if resolved != MESSAGE_CHAR_UUID.lower(): - logger.debug("_on_write: ignoring write to %s", resolved) - return + except Exception as exc: + logger.exception("_on_write failed: %s", exc) - fragment = bytes(value) - logger.debug( - "_on_write: %d bytes, sender=%s", - len(fragment), self._resolve_sender(), - ) + def _on_attributed_write( + self, central: str, char_uuid: str, value: bytes, offset: int + ) -> int: + """A write the attributing delegate names the central of (macOS). + + Returns the ATT code to answer the request with. Runs on + CoreBluetooth's queue, so it never raises. + """ + try: with self._lock: - self._bytes_received += len(fragment) - self._fragments_received += 1 + self._last_known_central = central + return self._handle_write(central, char_uuid.lower(), value, offset) + except Exception as exc: + logger.exception("attributed write failed: %s", exc) + return _ATT_SUCCESS + + def _handle_write( + self, central: str, char_uuid: str, value: bytes, offset: int + ) -> int: + """One write from ``central``: a hello, or a Message fragment.""" + if char_uuid == HELLO_CHAR_UUID.lower(): + return self._on_hello(central, value, offset) + if char_uuid != MESSAGE_CHAR_UUID.lower(): + logger.debug("ignoring a write to %s", char_uuid) + return _ATT_SUCCESS + with self._lock: + sender_id = self._hello_bindings.get(central, central) + self._bytes_received += len(value) + self._fragments_received += 1 + logger.debug("fragment of %d bytes from %s as %s", len(value), central, sender_id) + self._protocol.ble_fragment_received(sender_id=sender_id, fragment=list(value)) + return _ATT_SUCCESS + + def _on_hello(self, central: str, value: bytes, offset: int) -> int: + """Bind ``central``'s link to the address its hello proves. + + The chapter's order: refuse a prepared or offset write and a value out + of bounds before verifying anything; verify with the one verifier; bind + to the derived address only; never rebind; announce the peer only if no + other link has. A value that fails is ignored, and the link stays as it + would have been without one. + """ + if offset != 0: + return _ATT_INVALID_OFFSET + if not HELLO_MIN_LEN <= len(value) <= HELLO_MAX_LEN: + return _ATT_INVALID_ATTRIBUTE_VALUE_LENGTH + if central == "ble-peer": + # The writer could not be named, so there is no link to bind. + return _ATT_SUCCESS + try: + address = verify_identity_assertion(list(value)) + except Exception as exc: + self._emit_diagnostic("warning", "A hello did not verify", { + "central": central, + "error": str(exc), + }) + return _ATT_SUCCESS + with self._lock: + bound = self._hello_bindings.get(central) + if bound is not None: + if bound != address: + logger.info("ignoring a hello naming %s on a link bound to %s", address, bound) + return _ATT_SUCCESS + self._hello_bindings[central] = address + self._central_to_user_id[central] = address + already = any( + other == address + for c, other in self._hello_bindings.items() + if c != central + ) + self._emit_diagnostic("info", "Central said hello", { + "central": central, + "address": address, + }) + if not already and not self._linked_elsewhere(address): + try: + self._protocol.ble_peer_discovered(peer_id=address, rssi=-50) + except Exception: + logger.debug("ble_peer_discovered failed for %s", address) + return _ATT_SUCCESS - sender_id = self._resolve_sender() + def holds_peer(self, address: str) -> bool: + """Whether a subscribed central of ours carries ``address``. - self._protocol.ble_fragment_received( - sender_id=sender_id, fragment=list(fragment) + ``BleManager`` asks this before reporting a peer lost when its client + link goes: a notification addressed to ``address`` still reaches that + central, so the core keeps a route it would otherwise drop. + """ + with self._lock: + return any( + self._central_to_user_id.get(central) == address + for central in self._connected_centrals ) - logger.debug("_on_write: fragment fed to protocol OK") - except Exception as exc: - logger.exception("_on_write failed: %s", exc) + def _linked_elsewhere(self, address: str) -> bool: + probe = self.peer_linked_elsewhere + if probe is None: + return False + try: + return bool(probe(address)) + except Exception: + return False def _resolve_sender(self) -> str: """Best-effort identification of the writing central. @@ -658,6 +1068,10 @@ def resolve_sender_identity(self, sender_user_id: str) -> None: ) return central_uuid = next(iter(self._connected_centrals)) + # A hello proved this link's address; a message's sender field + # proves nothing, so it never replaces one. + if central_uuid in self._hello_bindings: + return # Avoid redundant re-registration. if self._central_to_user_id.get(central_uuid) == sender_user_id: return @@ -728,12 +1142,9 @@ async def _drain_outgoing_fragments_inner(self) -> None: break data = bytes(frag.data) - logger.info( - "OUTGOING fragment: %d bytes, recipient=%s", - len(data), getattr(frag, "recipient_id", None), - ) - await self.notify_fragment(data) - + recipient = getattr(frag, "recipient_id", None) + logger.debug("outgoing fragment: %d bytes, recipient=%s", len(data), recipient) + await self.notify_fragment(data, recipient) # Return fragment to pool regardless of send outcome try: self._protocol.ble_return_fragment() @@ -769,13 +1180,13 @@ def has_subscriber(self) -> bool: with self._lock: return bool(self._connected_centrals) - async def notify_fragment(self, data: bytes) -> bool: + async def notify_fragment(self, data: bytes, recipient: str | None = None) -> bool: """Notify one fragment on the Message characteristic. - Every subscribed central receives it: neither backend of bless - addresses a notification to one central, so a peripheral with more - than one central fans every fragment out to all of them. - + On macOS, a fragment for a recipient whose central said hello goes to + that central alone; anything else goes to every subscribed central. + BlueZ offers no per-central notification, so there every subscriber + receives every fragment. A notify that raises is reported and still returns True, as ``BleManager.write_fragment`` does for a failed write: the fragment was this role's, the core's retry re-sends the message, and a drain @@ -789,7 +1200,10 @@ async def notify_fragment(self, data: bytes) -> bool: if char is None: return False char.value = data - server.update_value(SERVICE_UUID, MESSAGE_CHAR_UUID) + if self._notify_ready is not None: + await self._notify_corebluetooth(server, char, data, recipient) + else: + server.update_value(SERVICE_UUID, MESSAGE_CHAR_UUID) except Exception as exc: self._emit_diagnostic("warning", f"Failed to notify a fragment: {exc}") return True @@ -800,8 +1214,51 @@ async def notify_fragment(self, data: bytes) -> bool: await asyncio.sleep(_INTER_FRAME_DELAY) return True - # -- background tasks ----------------------------------------------------- + def _centrals_for(self, recipient: str | None) -> list[Any] | None: + """The subscribed ``CBCentral`` objects bound to ``recipient``, or + None to notify every subscriber.""" + if not recipient: + return None + with self._lock: + ids = [c for c, a in self._hello_bindings.items() if a == recipient] + # .get, not a membership test then an index: CoreBluetooth's queue + # removes an unsubscribing central between the two. + subscribed = self._subscribed_centrals + targets = [t for t in (subscribed.get(c) for c in ids) if t is not None] + return targets or None + + async def _notify_corebluetooth( + self, server: Any, char: Any, data: bytes, recipient: str | None + ) -> None: + """``updateValue`` with CoreBluetooth's back-pressure honoured. + ``updateValue`` returns NO when the transmit queue is full, and the + fragment is not sent. bless ignores that, so a burst (a message, its + receipt, a key package) lost fragments silently and the message + waited for the core's next retry. Here a refusal waits for the + stack's "ready to update subscribers" and tries again. + """ + manager = server.peripheral_manager_delegate.peripheral_manager + ready = self._notify_ready + targets = self._centrals_for(recipient) + for _ in range(_NOTIFY_RETRIES + 1): + if ready is not None: + ready.clear() + # PyObjC passes bytes where an NSData is expected. + if manager.updateValue_forCharacteristic_onSubscribedCentrals_(data, char.obj, targets): + return + if ready is None: + break + try: + await asyncio.wait_for(ready.wait(), _NOTIFY_READY_TIMEOUT) + except asyncio.TimeoutError: + pass + self._emit_diagnostic("warning", "CoreBluetooth kept refusing a notification", { + "recipient": recipient, + "length": len(data), + }) + + # -- background tasks ----------------------------------------------------- async def _current_centrals(self) -> set[str] | None: """The centrals connected now, or None when they cannot be read.""" server = self._server @@ -846,10 +1303,11 @@ async def _peer_monitor_loop(self) -> None: # Classify under the lock (the delegate thread races us on # _connected_centrals / _central_to_user_id). FFI + diagnostic # notifications happen afterward, outside the lock. - now = time.monotonic() + now = self._clock() added: list[str] = [] rejected: list[str] = [] removed: list[tuple[str, str | None]] = [] + dropped: list[tuple[str, str]] = [] with self._lock: capacity = self._max_connections @@ -865,8 +1323,28 @@ async def _peer_monitor_loop(self) -> None: resolved_uid = self._central_to_user_id.pop( central_uuid, None ) + self._hello_bindings.pop(central_uuid, None) + if resolved_uid is not None and resolved_uid in self._hello_bindings.values(): + resolved_uid = None # another central of ours still carries it removed.append((central_uuid, resolved_uid)) + # A central that said hello and never subscribed is not in + # `known`, so the sweep above never sees it leave. + orphans: list[str] = [] + for central_uuid in list(self._hello_bindings): + if central_uuid in current or central_uuid in self._connected_centrals: + self._unsubscribed_since.pop(central_uuid, None) + continue + since = self._unsubscribed_since.setdefault(central_uuid, now) + if now - since >= _UNSUBSCRIBED_HELLO_GRACE: + orphans.append(central_uuid) + for central_uuid in orphans: + self._unsubscribed_since.pop(central_uuid, None) + address = self._hello_bindings.pop(central_uuid) + self._central_to_user_id.pop(central_uuid, None) + if address in self._hello_bindings.values(): + continue # another central of ours still carries it + dropped.append((central_uuid, address)) for central_uuid in rejected: self._emit_diagnostic( "warning", @@ -890,7 +1368,7 @@ async def _peer_monitor_loop(self) -> None: self._protocol.ble_peer_lost(peer_id=central_uuid) except Exception: logger.debug("ble_peer_lost failed for %s", central_uuid) - if resolved_uid: + if resolved_uid and not self._linked_elsewhere(resolved_uid): try: self._protocol.ble_peer_lost(peer_id=resolved_uid) except Exception: @@ -901,6 +1379,18 @@ async def _peer_monitor_loop(self) -> None: "info", "Central disconnected", {"central": central_uuid} ) + for central_uuid, address in dropped: + self._emit_diagnostic("info", "Dropped a hello from a central that never subscribed", { + "central": central_uuid, + "address": address, + }) + if self._linked_elsewhere(address): + continue + try: + self._protocol.ble_peer_lost(peer_id=address) + except Exception: + logger.debug("ble_peer_lost failed for %s", address) + known = current.copy() # Update cached advertising state (CoreBluetooth only; on diff --git a/bindings/python/offline_protocol_sdk/protocol_manager.py b/bindings/python/offline_protocol_sdk/protocol_manager.py index 451e5c44..e1a2a12f 100644 --- a/bindings/python/offline_protocol_sdk/protocol_manager.py +++ b/bindings/python/offline_protocol_sdk/protocol_manager.py @@ -15,6 +15,7 @@ import os import platform import sys +import weakref from pathlib import Path from typing import Any, Callable @@ -135,7 +136,7 @@ async def _drain_shared_inner(self) -> None: if await ble.write_fragment(recipient, data): pass elif peripheral.has_subscriber(): - await peripheral.notify_fragment(data) + await peripheral.notify_fragment(data, recipient) else: # Rare: the core queues a fragment only for a peer it holds. # A drop means the message waits for the core's retry, which @@ -151,6 +152,23 @@ async def _drain_shared_inner(self) -> None: logger.debug("ble_return_fragment failed", exc_info=True) +def _weak_probe(method: Callable[[str], bool]) -> Callable[[str], bool]: + """``method`` without the strong reference a bound method holds. + + The two BLE roles each ask the other whether a peer is still reached, so + strong bound methods made them a cycle, and a dropped ProtocolManager + kept the core, and the stores it holds open, alive until a garbage + collection pass. A role that is gone reaches nothing. + """ + ref = weakref.WeakMethod(method) + + def probe(peer_id: str) -> bool: + target = ref() + return bool(target(peer_id)) if target is not None else False + + return probe + + class _WifiDirectTransportCallbackImpl(WifiDirectTransportCallback): """Forwards the peer-stream slot's wake to :class:`PeerStreamManager`. @@ -477,6 +495,14 @@ def __init__( self.ble_peripheral: BlePeripheral | None = None if getattr(config, "ble_enabled", False): self.ble_peripheral = BlePeripheral(self._protocol, device_id) + if self.ble is not None: + # A hello from a peer our own central already proved does not + # announce it twice, and the peer's central leaving does not + # report lost a peer our client link still reaches. + self.ble_peripheral.peer_linked_elsewhere = _weak_probe(self.ble.holds_peer) + # And the other way: our client link going away does not + # report lost a peer whose central is bound on our peripheral. + self.ble.peer_linked_elsewhere = _weak_probe(self.ble_peripheral.holds_peer) # The peer-stream transport behind the `wifi_direct` slot: TCP # streams to configured or discovered hosts, each proved by the diff --git a/bindings/python/tests/test_ble_hello.py b/bindings/python/tests/test_ble_hello.py new file mode 100644 index 00000000..a4f415a1 --- /dev/null +++ b/bindings/python/tests/test_ble_hello.py @@ -0,0 +1,754 @@ +"""The Bluetooth LE hello, and the link attribution it exists for. + +A peripheral attributes each Message write to a link, and hands the core the +link's id as the fragment's transport identity. The core refuses a hop-0 +control frame (a key package) whose sender is not that identity, so a link +known only by a central UUID, or by the ``ble-peer`` placeholder when several +centrals are connected, can never start a session. These tests pin the +pieces that fix it: the peripheral binds a central to the address its hello +proves and feeds its fragments under that address; the central writes its +hello; a second link to one rotating peer is refused; a link CoreBluetooth +keeps after its peer is gone is found and dropped; and a Bluetooth stack that +never comes up fails the start instead of wedging the service. + +All tests mock bless, bleak and the core, so they run without Bluetooth. +""" + +from __future__ import annotations + +import asyncio +import threading +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +import offline_protocol_sdk.ble_manager as ble_manager_module +import offline_protocol_sdk.ble_peripheral as peripheral_module +from offline_protocol_sdk.ble_manager import ( + DEVICE_ID_CHAR_UUID, + HELLO_CHAR_UUID, + HELLO_MAX_LEN, + HELLO_MIN_LEN, + IDENTITY_CHAR_UUID, + MESSAGE_CHAR_UUID, + BleManager, +) +from offline_protocol_sdk.ble_peripheral import BlePeripheral +from offline_protocol_sdk.transport_manager import TransportError + +LOCAL_ADDRESS = "off1qysluvwl5922yctzd0u9gpr06gn3k7ldfvgtwgvn" +PHONE = "off1q8g49e9g4s5umugk0jnwhhl6m6z8ghvp9gc2wu9p" +OTHER = "off1q83uk7kfcky8ve832a04zauh0q4lxzr8auylht52" +HELLO = bytes(range(96)) +PHONE_CENTRAL = "437D0AC9-6337-D5F4-D74A-9F5FF0F3717C" +MAC_CENTRAL = "BAD89907-37BE-47DC-D8CA-3F4F23E62D97" + + +@pytest.fixture +def protocol(): + p = MagicMock() + p.local_address = MagicMock(return_value=LOCAL_ADDRESS) + p.identity_assertion = MagicMock(return_value=list(HELLO)) + return p + + +@pytest.fixture +def peripheral(protocol): + return BlePeripheral(protocol, device_id="coordinator") + + +@pytest.fixture +def verifier(monkeypatch): + """The core's one verifier, answering PHONE for every value.""" + verify = MagicMock(return_value=PHONE) + monkeypatch.setattr(peripheral_module, "verify_identity_assertion", verify) + return verify + + +def advancing_clock(step): + """A monitor clock that moves ``step`` seconds per reading. Wall time + cannot be used: on Windows time.monotonic() ticks every 15.6 ms and + asyncio runs a millisecond sleep at once, so a grace never elapsed.""" + now = [0.0] + + def clock(): + now[0] += step + return now[0] + + return clock + + +def write(peripheral, central, uuid, value, offset=0): + return peripheral._on_attributed_write(central, uuid.upper(), value, offset) + + +# --------------------------------------------------------------------------- +# The peripheral binds a link to its hello +# --------------------------------------------------------------------------- + + +class TestPeripheralHello: + def test_a_bound_central_feeds_the_core_under_the_address_it_proved( + self, peripheral, protocol, verifier + ): + """The bug: with two centrals connected the phone's key package went + to the core as ``ble-peer`` and was refused as a transport identity + mismatch. Bound by its hello, it goes under the phone's address.""" + peripheral._connected_centrals.update({PHONE_CENTRAL: 1.0, MAC_CENTRAL: 2.0}) + assert write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) == 0 + write(peripheral, PHONE_CENTRAL, MESSAGE_CHAR_UUID, b"\x03\x00kp") + + protocol.ble_fragment_received.assert_called_once_with( + sender_id=PHONE, fragment=list(b"\x03\x00kp") + ) + verifier.assert_called_once_with(list(HELLO)) + + def test_an_unbound_central_keeps_its_own_id(self, peripheral, protocol, verifier): + write(peripheral, MAC_CENTRAL, MESSAGE_CHAR_UUID, b"\x03\x00x") + assert protocol.ble_fragment_received.call_args.kwargs["sender_id"] == MAC_CENTRAL + + def test_a_hello_announces_the_peer_once(self, peripheral, protocol, verifier): + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + write(peripheral, "SECOND-LINK", HELLO_CHAR_UUID, HELLO) + protocol.ble_peer_discovered.assert_called_once_with(peer_id=PHONE, rssi=-50) + + def test_a_peer_our_central_already_proved_is_not_announced_again( + self, peripheral, protocol, verifier + ): + peripheral.peer_linked_elsewhere = lambda address: address == PHONE + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + protocol.ble_peer_discovered.assert_not_called() + # Still bound: its writes must carry its address either way. + write(peripheral, PHONE_CENTRAL, MESSAGE_CHAR_UUID, b"\x03\x00x") + assert protocol.ble_fragment_received.call_args.kwargs["sender_id"] == PHONE + + @pytest.mark.parametrize("length", [HELLO_MIN_LEN - 1, HELLO_MAX_LEN + 1, 0]) + def test_a_value_out_of_bounds_is_refused_before_verifying( + self, peripheral, verifier, length + ): + assert write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, bytes(length)) == 0x0D + verifier.assert_not_called() + assert PHONE_CENTRAL not in peripheral._hello_bindings + + def test_a_write_at_an_offset_is_refused_before_verifying(self, peripheral, verifier): + assert write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO, offset=4) == 0x07 + verifier.assert_not_called() + + def test_a_hello_that_does_not_verify_is_ignored( + self, peripheral, protocol, monkeypatch + ): + monkeypatch.setattr( + peripheral_module, + "verify_identity_assertion", + MagicMock(side_effect=ValueError("bad signature")), + ) + assert write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) == 0 + assert PHONE_CENTRAL not in peripheral._hello_bindings + protocol.ble_peer_discovered.assert_not_called() + + def test_a_hello_never_rebinds_a_link(self, peripheral, protocol, monkeypatch): + verify = MagicMock(side_effect=[PHONE, OTHER]) + monkeypatch.setattr(peripheral_module, "verify_identity_assertion", verify) + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + assert peripheral._hello_bindings[PHONE_CENTRAL] == PHONE + + def test_an_unnamed_writer_binds_nothing(self, peripheral, verifier): + """Without the attributing delegate (or on BlueZ before the tracker + saw the writer) the writer is ``ble-peer``, and there is no link.""" + peripheral._connected_centrals.update({"A": 1.0, "B": 2.0}) + peripheral._on_write(HELLO_CHAR_UUID, bytearray(HELLO)) + assert peripheral._hello_bindings == {} + verifier.assert_not_called() + + def test_a_message_sender_never_replaces_a_hello_binding( + self, peripheral, verifier + ): + peripheral._connected_centrals[PHONE_CENTRAL] = 1.0 + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + peripheral.resolve_sender_identity(OTHER) + assert peripheral._central_to_user_id[PHONE_CENTRAL] == PHONE + + def test_writes_from_the_bless_callback_are_attributed_too( + self, peripheral, protocol, verifier + ): + """BlueZ: bless's own callback, with the tracker naming the writer.""" + peripheral._bluez_centrals = MagicMock(current="AA:BB:CC:DD:EE:FF") + peripheral._on_write(HELLO_CHAR_UUID, bytearray(HELLO)) + peripheral._on_write(MESSAGE_CHAR_UUID, bytearray(b"\x03\x00x")) + assert protocol.ble_fragment_received.call_args.kwargs["sender_id"] == PHONE + + +class TestPeripheralCentralLeaving: + @pytest.mark.asyncio + async def test_a_bound_central_leaving_reports_its_peer_lost( + self, peripheral, protocol, verifier + ): + await self._leave(peripheral, linked_elsewhere=False) + lost = [c.kwargs["peer_id"] for c in protocol.ble_peer_lost.call_args_list] + assert lost == [PHONE_CENTRAL, PHONE] + assert peripheral._hello_bindings == {} + + @pytest.mark.asyncio + async def test_a_peer_our_client_link_still_reaches_is_not_reported_lost( + self, peripheral, protocol, verifier + ): + await self._leave(peripheral, linked_elsewhere=True) + lost = [c.kwargs["peer_id"] for c in protocol.ble_peer_lost.call_args_list] + assert lost == [PHONE_CENTRAL] + + @pytest.mark.asyncio + async def test_a_hello_from_a_central_that_never_subscribes_is_dropped( + self, peripheral, protocol, verifier, monkeypatch + ): + """The monitor tracks subscribers only, so a central that said hello + and left without subscribing kept its binding, and the core a route + to its address, for the life of the process.""" + monkeypatch.setattr(peripheral_module, "_UNSUBSCRIBED_HELLO_GRACE", 0.01) + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + await self._poll(peripheral, lambda: PHONE_CENTRAL not in peripheral._hello_bindings) + assert PHONE_CENTRAL not in peripheral._hello_bindings + assert PHONE_CENTRAL not in peripheral._central_to_user_id + protocol.ble_peer_lost.assert_called_once_with(peer_id=PHONE) + + @pytest.mark.asyncio + async def test_a_hello_that_subscribes_within_the_grace_is_kept( + self, peripheral, protocol, verifier, monkeypatch + ): + monkeypatch.setattr(peripheral_module, "_UNSUBSCRIBED_HELLO_GRACE", 5.0) + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + # A hello lands before the subscription does: unsubscribed for the + # first polls, subscribed after. + polls = iter([set()] * 5) + await self._poll( + peripheral, + lambda: False, + current=lambda: next(polls, {PHONE_CENTRAL}), + ticks=60, + step=0.5, + ) + assert peripheral._hello_bindings == {PHONE_CENTRAL: PHONE} + protocol.ble_peer_lost.assert_not_called() + + @pytest.mark.asyncio + async def test_a_dropped_hello_another_link_still_reaches_is_not_lost( + self, peripheral, protocol, verifier, monkeypatch + ): + monkeypatch.setattr(peripheral_module, "_UNSUBSCRIBED_HELLO_GRACE", 0.01) + peripheral.peer_linked_elsewhere = lambda address: True + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + await self._poll(peripheral, lambda: PHONE_CENTRAL not in peripheral._hello_bindings) + assert PHONE_CENTRAL not in peripheral._hello_bindings + protocol.ble_peer_lost.assert_not_called() + + async def _poll(self, peripheral, done, current=lambda: set(), ticks=200, step=1.0): + peripheral._server = MagicMock() + peripheral._bluez_centrals = None + peripheral._clock = advancing_clock(step) + + async def centrals(): + return current() + + peripheral._current_centrals = centrals + with patch.object(peripheral_module, "_PEER_MONITOR_INTERVAL", 0.001): + task = asyncio.ensure_future(peripheral._peer_monitor_loop()) + for _ in range(ticks): + await asyncio.sleep(0.002) + if done(): + break + task.cancel() + await asyncio.wait({task}) + + async def _leave(self, peripheral, linked_elsewhere): + peripheral.peer_linked_elsewhere = lambda address: linked_elsewhere + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + peripheral._server = MagicMock() + peripheral._bluez_centrals = None + peripheral._clock = advancing_clock(1.0) + seen = [{PHONE_CENTRAL}, set()] + + async def current(): + return seen.pop(0) if seen else set() + + peripheral._current_centrals = current + with patch.object(peripheral_module, "_PEER_MONITOR_INTERVAL", 0.001): + task = asyncio.ensure_future(peripheral._peer_monitor_loop()) + for _ in range(200): + await asyncio.sleep(0.002) + if not seen and PHONE_CENTRAL not in peripheral._hello_bindings: + break + task.cancel() + await asyncio.wait({task}) + + +class TestPeripheralGatt: + @pytest.mark.asyncio + async def test_the_service_offers_a_writable_hello(self, peripheral, monkeypatch): + monkeypatch.setattr( + peripheral_module, + "GATTCharacteristicProperties", + MagicMock(read=1, write=2, write_without_response=4, notify=8), + ) + monkeypatch.setattr( + peripheral_module, + "GATTAttributePermissions", + MagicMock(readable=1, writeable=2), + ) + peripheral._server = MagicMock() + peripheral._server.add_new_service = AsyncMock() + peripheral._server.add_new_characteristic = AsyncMock() + peripheral._identity_bytes = HELLO + await peripheral._setup_gatt_server() + added = { + c.args[1]: (c.args[2], c.args[4]) + for c in peripheral._server.add_new_characteristic.call_args_list + } + assert added[HELLO_CHAR_UUID] == (2, 2) + + +# --------------------------------------------------------------------------- +# Notifications go to one central, and survive a full transmit queue +# --------------------------------------------------------------------------- + + +class TestPeripheralNotify: + def test_a_fragment_for_a_bound_peer_targets_its_central(self, peripheral, verifier): + central_obj = object() + peripheral._subscribed_centrals = {PHONE_CENTRAL: central_obj, MAC_CENTRAL: object()} + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + assert peripheral._centrals_for(PHONE) == [central_obj] + assert peripheral._centrals_for(OTHER) is None + assert peripheral._centrals_for(None) is None + + def test_a_central_that_unsubscribes_mid_lookup_falls_back_to_all( + self, peripheral, verifier + ): + """CoreBluetooth's queue removes an unsubscribing central between a + membership test and an index; the lookup must not raise.""" + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + + class Vanishing(dict): + def __contains__(self, key): + return True # present at the test, gone at the index + + peripheral._subscribed_centrals = Vanishing() + assert peripheral._centrals_for(PHONE) is None + + @pytest.mark.asyncio + async def test_a_refused_notification_is_sent_once_the_stack_is_ready(self, peripheral): + peripheral._notify_ready = asyncio.Event() + answers = [False, True] + sent = [] + + def update(value, char, centrals): + sent.append(value) + ok = answers.pop(0) + if not ok: + asyncio.get_running_loop().call_later(0.01, peripheral._notify_ready.set) + return ok + + server = MagicMock() + server.peripheral_manager_delegate.peripheral_manager.updateValue_forCharacteristic_onSubscribedCentrals_ = update + await peripheral._notify_corebluetooth(server, MagicMock(), b"frag", None) + assert sent == [b"frag", b"frag"] + + @pytest.mark.asyncio + async def test_a_notification_refused_every_time_gives_up(self, peripheral, monkeypatch): + monkeypatch.setattr(peripheral_module, "_NOTIFY_READY_TIMEOUT", 0.001) + peripheral._notify_ready = asyncio.Event() + server = MagicMock() + manager = server.peripheral_manager_delegate.peripheral_manager + manager.updateValue_forCharacteristic_onSubscribedCentrals_ = MagicMock(return_value=False) + await peripheral._notify_corebluetooth(server, MagicMock(), b"frag", None) + assert manager.updateValue_forCharacteristic_onSubscribedCentrals_.call_count == ( + peripheral_module._NOTIFY_RETRIES + 1 + ) + + +class TestPeripheralHoldsPeer: + def test_a_subscribed_central_bound_to_the_address_holds_it(self, peripheral, verifier): + """What BleManager asks before reporting a peer lost when its own + client link goes.""" + write(peripheral, PHONE_CENTRAL, HELLO_CHAR_UUID, HELLO) + assert not peripheral.holds_peer(PHONE), "said hello but not subscribed: no route" + peripheral._connected_centrals[PHONE_CENTRAL] = 1.0 + assert peripheral.holds_peer(PHONE) + assert not peripheral.holds_peer(OTHER) + + +class TestPreparedHello: + """CoreBluetooth hands a long write over as one batch at rising offsets. + The chapter refuses a prepared hello before verifying anything, so the + batch is judged whole, ahead of the per-request handler.""" + + def test_a_hello_in_a_batch_is_refused(self): + hello = HELLO_CHAR_UUID.upper() + assert peripheral_module._prepared_hello_refusal([(hello, 0), (hello, 96)]) == 0x07 + + def test_a_hello_batched_with_another_write_is_refused_at_offset_zero(self): + """A reliable write puts several requests at offset 0 in one batch; + a hello among them is still not one write.""" + assert peripheral_module._prepared_hello_refusal( + [(HELLO_CHAR_UUID, 0), (MESSAGE_CHAR_UUID, 0)] + ) == 0x07 + + def test_a_hello_at_an_offset_is_refused(self): + assert peripheral_module._prepared_hello_refusal([(HELLO_CHAR_UUID, 4)]) == 0x07 + + def test_a_single_hello_at_offset_zero_passes_to_the_handler(self): + assert peripheral_module._prepared_hello_refusal([(HELLO_CHAR_UUID, 0)]) is None + + def test_message_writes_are_not_judged(self): + assert peripheral_module._prepared_hello_refusal( + [(MESSAGE_CHAR_UUID, 0), (MESSAGE_CHAR_UUID, 0)] + ) is None + + +# --------------------------------------------------------------------------- +# A Bluetooth stack that never comes up fails the start, it does not wedge +# --------------------------------------------------------------------------- + + +class TestPeripheralStart: + @pytest.fixture(autouse=True) + def on_macos(self, monkeypatch): + """The off-loop construction is macOS's: BlueZ's never blocks.""" + monkeypatch.setattr(peripheral_module.sys, "platform", "darwin") + + @pytest.mark.asyncio + async def test_a_constructor_that_never_returns_times_out(self, peripheral, monkeypatch): + """bless's CoreBluetooth constructor waits, without a deadline, for a + power-on that never comes when Bluetooth is not authorised. On the + loop thread that froze the whole service; now it is a failed start.""" + monkeypatch.setattr(peripheral_module, "_SERVER_INIT_TIMEOUT", 0.05) + release = threading.Event() + monkeypatch.setattr( + BlePeripheral, "_construct_server", staticmethod(lambda name, loop: release.wait(5)) + ) + try: + with pytest.raises(TransportError, match="did not come up"): + await peripheral._build_server("coordinator") + finally: + release.set() + + @pytest.mark.asyncio + async def test_the_loop_keeps_running_while_the_stack_decides(self, peripheral, monkeypatch): + release = threading.Event() + monkeypatch.setattr( + BlePeripheral, "_construct_server", + staticmethod(lambda name, loop: (release.wait(5), ("server", loop))[1]), + ) + build = asyncio.ensure_future(peripheral._build_server("coordinator")) + await asyncio.sleep(0.01) + assert not build.done() + release.set() + # bless is handed the running loop: in the init thread it would look + # one up and find none. + assert await build == ("server", asyncio.get_running_loop()) + + @pytest.mark.asyncio + async def test_bluez_builds_on_the_loop_thread(self, peripheral, monkeypatch): + monkeypatch.setattr(peripheral_module.sys, "platform", "linux") + seen = {} + + def construct(name, loop): + seen["thread"] = threading.current_thread() + return "server" + + monkeypatch.setattr(BlePeripheral, "_construct_server", staticmethod(construct)) + assert await peripheral._build_server("coordinator") == "server" + assert seen["thread"] is threading.main_thread() + + @pytest.mark.asyncio + async def test_a_denied_process_is_refused_before_touching_the_stack( + self, peripheral, protocol, monkeypatch + ): + monkeypatch.setattr(peripheral, "is_available", lambda: True) + monkeypatch.setattr( + peripheral_module, + "_bluetooth_authorization_refusal", + lambda: "Bluetooth is denied to this process", + ) + construct = MagicMock() + monkeypatch.setattr(BlePeripheral, "_construct_server", staticmethod(construct)) + with pytest.raises(TransportError, match="denied"): + await peripheral.start() + construct.assert_not_called() + + +def test_the_attributing_delegate_is_swapped_in_for_one_construction_only(): + cb_server = pytest.importorskip("bless.backends.corebluetooth.server") + original = cb_server.PeripheralManagerDelegate + marker = MagicMock(name="delegate subclass") + seen = {} + + def fake_server(name, loop=None): + seen["delegate"] = cb_server.PeripheralManagerDelegate + return "server" + + with patch.object(peripheral_module, "_attributing_delegate_class", return_value=marker), \ + patch.object(peripheral_module, "BlessServer", fake_server), \ + patch.object(peripheral_module.sys, "platform", "darwin"): + assert BlePeripheral._construct_server("coordinator") == "server" + assert seen["delegate"] is marker + assert cb_server.PeripheralManagerDelegate is original + + +def test_a_construction_behind_a_wedged_one_says_so(monkeypatch): + """A constructor that never returned keeps the delegate swapped and the + lock held. A later start reports that, rather than waiting out the init + deadline and blaming the adapter.""" + pytest.importorskip("bless.backends.corebluetooth.server") + monkeypatch.setattr(peripheral_module, "_DELEGATE_SWAP_WAIT", 0.01) + monkeypatch.setattr(peripheral_module.sys, "platform", "darwin") + monkeypatch.setattr(peripheral_module, "_attributing_delegate_class", lambda: MagicMock()) + construct = MagicMock() + monkeypatch.setattr(peripheral_module, "BlessServer", construct) + assert peripheral_module._DELEGATE_SWAP_LOCK.acquire(timeout=1) + try: + with pytest.raises(TransportError, match="never returned"): + BlePeripheral._construct_server("coordinator") + finally: + peripheral_module._DELEGATE_SWAP_LOCK.release() + construct.assert_not_called() + # Released on the way out of a normal construction too. + BlePeripheral._construct_server("coordinator") + assert peripheral_module._DELEGATE_SWAP_LOCK.acquire(timeout=0) + peripheral_module._DELEGATE_SWAP_LOCK.release() + + +# --------------------------------------------------------------------------- +# The central says hello, keeps one link per peer, and drops dead links +# --------------------------------------------------------------------------- + + +def _client(reads, *, hello=True, mtu=247, room=244, connected=True): + client = MagicMock() + client.connect = AsyncMock() + client.disconnect = AsyncMock() + client.start_notify = AsyncMock() + client.write_gatt_char = AsyncMock() + client.is_connected = connected + client.mtu_size = mtu + hello_char = MagicMock(max_write_without_response_size=room) + client.services.get_characteristic = MagicMock( + side_effect=lambda uuid: hello_char if (uuid == HELLO_CHAR_UUID and hello) else None + ) + + async def read_gatt_char(uuid): + if uuid in reads: + return bytearray(reads[uuid]) + raise RuntimeError(f"no such characteristic {uuid}") + + client.read_gatt_char = AsyncMock(side_effect=read_gatt_char) + return client + + +@pytest.fixture +def manager(protocol, monkeypatch): + monkeypatch.setattr( + ble_manager_module, "verify_identity_assertion", MagicMock(return_value=PHONE) + ) + return BleManager(protocol, device_id="desktop-1") + + +PEER_READS = {DEVICE_ID_CHAR_UUID: PHONE.encode(), IDENTITY_CHAR_UUID: HELLO} + + +async def _connect(manager, client, addr): + device = MagicMock() + device.address = addr + manager._connecting.add(addr) + with patch("offline_protocol_sdk.ble_manager.BleakClient", return_value=client): + await manager._connect_to_peer(device) + + +class TestCentralHello: + @pytest.mark.asyncio + async def test_the_central_writes_its_assertion_with_response(self, manager): + client = _client(PEER_READS) + await _connect(manager, client, "AA:01") + client.write_gatt_char.assert_awaited_once_with(HELLO_CHAR_UUID, HELLO, response=True) + + @pytest.mark.asyncio + async def test_no_hello_characteristic_no_write(self, manager): + client = _client(PEER_READS, hello=False) + await _connect(manager, client, "AA:01") + client.write_gatt_char.assert_not_awaited() + client.disconnect.assert_not_awaited() + + @pytest.mark.asyncio + async def test_a_hello_that_does_not_fit_one_packet_is_skipped(self, manager): + client = _client(PEER_READS, room=20) + await _connect(manager, client, "AA:01") + client.write_gatt_char.assert_not_awaited() + client.disconnect.assert_not_awaited() + + @pytest.mark.asyncio + async def test_the_bound_is_the_characteristics_not_the_clients_mtu(self, manager): + """bleak's BlueZ client reports a 23-byte MTU until it is explicitly + acquired, which skipped every hello on Linux. The characteristic + reads BlueZ's negotiated MTU.""" + client = _client(PEER_READS, mtu=23, room=244) + await _connect(manager, client, "AA:01") + client.write_gatt_char.assert_awaited_once_with(HELLO_CHAR_UUID, HELLO, response=True) + + @pytest.mark.asyncio + async def test_the_hello_lands_before_the_peer_is_announced(self, manager, protocol): + """The announce makes the core push a key package. Written ahead of + the hello, it reached a peripheral that could not name the link, and + was refused as a transport identity mismatch.""" + order = [] + client = _client(PEER_READS) + client.start_notify = AsyncMock(side_effect=lambda *a: order.append("subscribe")) + client.write_gatt_char = AsyncMock(side_effect=lambda *a, **k: order.append("hello")) + protocol.ble_peer_discovered.side_effect = lambda **k: order.append("announce") + await _connect(manager, client, "AA:01") + assert order == ["subscribe", "hello", "announce"] + + @pytest.mark.asyncio + async def test_a_hello_that_is_never_answered_does_not_hold_the_announce( + self, manager, protocol, monkeypatch + ): + monkeypatch.setattr(ble_manager_module, "HELLO_WRITE_TIMEOUT", 0.01) + client = _client(PEER_READS) + + async def hang(*args, **kwargs): + await asyncio.sleep(10) + + client.write_gatt_char = AsyncMock(side_effect=hang) + # Bounded here too, so a central with no deadline fails rather than + # passing once the hang ends. + await asyncio.wait_for(_connect(manager, client, "AA:01"), 2.0) + protocol.ble_peer_discovered.assert_called_once_with(peer_id=PHONE, rssi=-70) + client.disconnect.assert_not_awaited() + + @pytest.mark.asyncio + async def test_a_refused_hello_keeps_the_link(self, manager, protocol): + client = _client(PEER_READS) + client.write_gatt_char = AsyncMock(side_effect=RuntimeError("refused")) + await _connect(manager, client, "AA:01") + client.disconnect.assert_not_awaited() + protocol.ble_peer_discovered.assert_called_once() + + +class TestCentralDuplicateLink: + @pytest.mark.asyncio + async def test_a_second_address_of_one_peer_is_closed_unannounced(self, manager, protocol): + first, second = _client(PEER_READS), _client(PEER_READS) + await _connect(manager, first, "AA:01") + await _connect(manager, second, "AA:02") + + second.disconnect.assert_awaited_once() + protocol.ble_peer_discovered.assert_called_once_with(peer_id=PHONE, rssi=-70) + assert manager._device_id_to_addr[PHONE] == "AA:01" + assert manager.holds_peer(PHONE) + + @pytest.mark.asyncio + async def test_a_refused_address_is_not_redialed_while_the_kept_link_lives( + self, manager, protocol + ): + """Without this the second address of a rotating peripheral was + redialed, verified and refused again every cooldown.""" + first, second = _client(PEER_READS), _client(PEER_READS) + await _connect(manager, first, "AA:01") + await _connect(manager, second, "AA:02") + # Past the per-address cooldown, inside the refusal: only the + # refusal can say no here. + inside = ( + ble_manager_module.time.monotonic() + + ble_manager_module.ADAPTIVE_COOLDOWN_PER_PERIPHERAL + + 1 + ) + assert inside < manager._duplicates["AA:02"][1] + with manager._lock: + manager._global_attempts.clear() + assert not manager._should_connect_locked("AA:02", inside) + # Lapses once the kept link is gone, inside the window. + manager._clients.pop("AA:01") + assert manager._should_connect_locked("AA:02", inside) + assert "AA:02" not in manager._duplicates + + @pytest.mark.asyncio + async def test_a_refusal_lapses_after_its_window(self, manager, protocol): + first, second = _client(PEER_READS), _client(PEER_READS) + await _connect(manager, first, "AA:01") + await _connect(manager, second, "AA:02") + later = ( + ble_manager_module.time.monotonic() + + ble_manager_module.DUPLICATE_LINK_SUPPRESS + + ble_manager_module.ADAPTIVE_COOLDOWN_PER_PERIPHERAL + + 1 + ) + with manager._lock: + manager._global_attempts.clear() + assert manager._should_connect_locked("AA:02", later) + + @pytest.mark.asyncio + async def test_a_peer_whose_kept_link_died_is_linked_again(self, manager, protocol): + first = _client(PEER_READS, connected=False) + await _connect(manager, first, "AA:01") + second = _client(PEER_READS) + await _connect(manager, second, "AA:02") + second.disconnect.assert_not_awaited() + assert manager._device_id_to_addr[PHONE] == "AA:02" + + +class TestCentralLiveness: + @pytest.mark.asyncio + async def test_a_link_whose_peer_is_gone_is_dropped_and_reported(self, manager, protocol): + client = _client(PEER_READS) + await _connect(manager, client, "AA:01") + client.read_gatt_char = AsyncMock(side_effect=RuntimeError("attribute not found")) + + await manager._check_liveness() + + client.disconnect.assert_awaited() + protocol.ble_peer_lost.assert_called_once_with(peer_id=PHONE) + assert not manager.holds_peer(PHONE) + + @pytest.mark.asyncio + async def test_a_peer_the_peripheral_still_reaches_is_not_reported_lost( + self, manager, protocol + ): + """The peer's central is bound on our peripheral, so a notification + still reaches it. Reported lost, the core dropped that route too.""" + manager.peer_linked_elsewhere = lambda peer: peer == PHONE + client = _client(PEER_READS) + await _connect(manager, client, "AA:01") + client.read_gatt_char = AsyncMock(side_effect=RuntimeError("attribute not found")) + + await manager._check_liveness() + + client.disconnect.assert_awaited() + protocol.ble_peer_lost.assert_not_called() + assert not manager.holds_peer(PHONE) + + @pytest.mark.asyncio + async def test_a_link_that_now_names_another_peer_is_dropped(self, manager, protocol): + client = _client(PEER_READS) + await _connect(manager, client, "AA:01") + client.read_gatt_char = AsyncMock(return_value=bytearray(OTHER.encode())) + await manager._check_liveness() + protocol.ble_peer_lost.assert_called_once_with(peer_id=PHONE) + + @pytest.mark.asyncio + async def test_a_live_link_is_kept(self, manager, protocol): + client = _client(PEER_READS) + await _connect(manager, client, "AA:01") + await manager._check_liveness() + client.disconnect.assert_not_awaited() + protocol.ble_peer_lost.assert_not_called() + + @pytest.mark.asyncio + async def test_a_read_that_hangs_counts_as_dead(self, manager, protocol, monkeypatch): + monkeypatch.setattr(ble_manager_module, "LIVENESS_READ_TIMEOUT", 0.01) + client = _client(PEER_READS) + await _connect(manager, client, "AA:01") + + async def hang(uuid): + await asyncio.sleep(10) + + client.read_gatt_char = AsyncMock(side_effect=hang) + await manager._check_liveness() + protocol.ble_peer_lost.assert_called_once_with(peer_id=PHONE) diff --git a/bindings/python/tests/test_ble_peripheral.py b/bindings/python/tests/test_ble_peripheral.py index 35763421..341b15b8 100644 --- a/bindings/python/tests/test_ble_peripheral.py +++ b/bindings/python/tests/test_ble_peripheral.py @@ -880,10 +880,13 @@ async def write_fragment(recipient, data): peripheral.sent = [] peripheral.has_subscriber = MagicMock(return_value=subscribed) - async def notify_fragment(data): + async def notify_fragment(data, recipient=None): peripheral.sent.append(data) + peripheral.recipients.append(recipient) return True + peripheral.recipients = [] + peripheral.notify_fragment = notify_fragment return _BleTransportCallbackImpl(central, peripheral), central, peripheral @@ -898,6 +901,9 @@ async def test_each_fragment_goes_to_the_role_that_holds_its_peer(self, mock_pro await router._drain_shared() assert central.sent == [("off1central", b"a1"), ("off1central", b"a2")] assert peripheral.sent == [b"b1"] + # The recipient travels with the fragment, so the peripheral can + # address it to that peer's central alone. + assert peripheral.recipients == ["AA:BB:CC:DD:EE:01"] assert mock_protocol.ble_get_next_fragment.call_count == 4 assert router.fragments_dropped == 0 diff --git a/bindings/python/tests/test_protocol_manager.py b/bindings/python/tests/test_protocol_manager.py index b5b7795e..9d8f6df4 100644 --- a/bindings/python/tests/test_protocol_manager.py +++ b/bindings/python/tests/test_protocol_manager.py @@ -172,6 +172,32 @@ def test_ble_created_when_enabled(self): assert pm.ble is not None assert pm.ble_peripheral is not None + def test_each_ble_role_asks_the_other_before_reporting_a_peer_lost(self): + """The central's client link going does not drop a peer whose central + is bound on our peripheral, and the other way round. The probes are + weak, so the two roles do not keep each other, and the core, alive.""" + import weakref + + config = _make_config(ble_enabled=True) + from offline_protocol_sdk.protocol_manager import ProtocolManager + + pm = ProtocolManager(config) + pm.ble._device_id_to_addr["off1peer"] = "AA:01" + pm.ble._clients["AA:01"] = MagicMock(is_connected=True) + assert pm.ble_peripheral.peer_linked_elsewhere("off1peer") + assert not pm.ble_peripheral.peer_linked_elsewhere("off1other") + + pm.ble_peripheral._connected_centrals["C1"] = 1.0 + pm.ble_peripheral._central_to_user_id["C1"] = "off1peer" + assert pm.ble.peer_linked_elsewhere("off1peer") + assert not pm.ble.peer_linked_elsewhere("off1other") + + peripheral = weakref.ref(pm.ble_peripheral) + probe = pm.ble.peer_linked_elsewhere + pm.ble_peripheral = None + assert peripheral() is None, "the central's probe must not keep the peripheral alive" + assert not probe("off1peer") + def test_ble_none_when_disabled(self): config = _make_config(ble_enabled=False) from offline_protocol_sdk.protocol_manager import ProtocolManager diff --git a/bindings/react-native/MeshSdk.podspec b/bindings/react-native/MeshSdk.podspec index 7394a121..075e44c5 100644 --- a/bindings/react-native/MeshSdk.podspec +++ b/bindings/react-native/MeshSdk.podspec @@ -35,6 +35,7 @@ Pod::Spec.new do |s| "ios/AddressDeclarationPolicy.swift", "ios/BleAppTag.swift", "ios/BleDensityPolicy.swift", + "ios/BleHelloPolicy.swift", "ios/BleMessageNotificationPolicy.swift", "ios/BleServiceInstanceSelection.swift", "ios/PeerIdentityBinding.swift", diff --git a/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/BleTransportFacade.kt b/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/BleTransportFacade.kt index a58cdc3a..fbe12a82 100644 --- a/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/BleTransportFacade.kt +++ b/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/BleTransportFacade.kt @@ -14,6 +14,7 @@ import android.os.Handler import android.os.HandlerThread import android.os.Looper import android.os.ParcelUuid +import android.os.SystemClock import android.util.Log import androidx.core.content.ContextCompat import com.offlineprotocol.BleAppTag @@ -2802,6 +2803,14 @@ class BleTransportFacade( return } + if (connections.isSuppressedDuplicate(device.address, SystemClock.elapsedRealtime())) { + if (logThrottler.shouldLog("duplicate_skip_${device.address}", intervalMs = 30000)) { + Log.d(TAG, "Not redialing ${device.address}: a second link to a peer already held") + } + connections.consumePendingRole(device.address) + return + } + if (currentConnectionCount() >= MAX_CONNECTIONS_PER_DEVICE) { if (logThrottler.shouldLog("mesh_conn_cap", intervalMs = 10000)) { Log.d(TAG, "Connection cap reached, not connecting to ${device.address}") diff --git a/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/CentralGattClient.kt b/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/CentralGattClient.kt index e4f68c30..53043006 100644 --- a/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/CentralGattClient.kt +++ b/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/CentralGattClient.kt @@ -12,6 +12,7 @@ import android.bluetooth.BluetoothStatusCodes import android.os.Build import android.os.Handler import android.os.Looper +import android.os.SystemClock import android.util.Log import com.offlineprotocol.BleServiceInstanceSelection import com.offlineprotocol.PeerIdentityBinding @@ -190,6 +191,11 @@ internal class CentralGattClient( * leave RSSI intact because they do not imply a quality problem. */ private const val GATT_GENERIC_ERROR = 133 + /** How long an address refused as a duplicate client link is left + * undialed. Long enough to outlast a rotating peripheral advertising + * under its old address. The refusal lapses early once the kept + * link is gone ([MeshConnectionRegistry.isSuppressedDuplicate]). */ + internal const val DUPLICATE_LINK_SUPPRESS_MS = 60_000L /** Hard ceiling on a single inbound BLE notification we accept from * a remote peripheral. Mirrors [PeripheralGattServer.MAX_INBOUND_WRITE_BYTES] * — well above mesh fragment size for MTU-negotiation headroom, but @@ -1243,6 +1249,10 @@ internal class CentralGattClient( return } is PeerIdentityBinding.Outcome.Verified -> { + if (host.connections.hasOtherClientLink(outcome.peerId, address)) { + refuseDuplicateLink(gatt, outcome.peerId) + return + } announceVerifiedPeer(gatt, outcome.peerId) // Identity proved → now enable notifications. enableNotificationsOnLink(gatt) @@ -1332,6 +1342,36 @@ internal class CentralGattClient( drainPendingInboundFor(address, peerId) } + /** + * Closes a client link that proved an identity we already hold a client + * link to at another address. + * + * A macOS peripheral rotates its address and advertises under the old and + * the new one at once, so the scanner dials both and both verify as the + * same peer. Kept, the pair spends two of [BleTransportFacade]'s four + * connection slots on one peer, and with two such peers the cap refuses + * every inbound central, which is how such a peripheral's own central + * would have learned who we are. The first link to verify stays; this one + * is closed unannounced, and its address is not redialed for + * [DUPLICATE_LINK_SUPPRESS_MS]. + */ + private fun refuseDuplicateLink(gatt: BluetoothGatt, peerId: String) { + val address = gatt.device.address + Log.i(TAG, "Closing a second client link to $peerId at $address") + diagnosticEmitter( + "info", + "BLE duplicate link closed", + mapOf("address" to address, "peer" to peerId), + ) + host.connections.suppressDuplicate( + address, + peerId, + SystemClock.elapsedRealtime() + DUPLICATE_LINK_SUPPRESS_MS, + ) + host.connections.consumePendingRole(address) + closeGattClient(gatt, "duplicate_identity") + } + /** * Closes a link whose peer could not prove the id it advertised. * diff --git a/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/MeshConnectionRegistry.kt b/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/MeshConnectionRegistry.kt index b89f9769..b91f9301 100644 --- a/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/MeshConnectionRegistry.kt +++ b/bindings/react-native/android/src/main/java/com/offlineprotocol/ble/MeshConnectionRegistry.kt @@ -30,6 +30,16 @@ class MeshConnectionRegistry { // still in flight, and a dial to a peer whose Bluetooth is off lasts as // long as the stack's connect timeout, so it cannot count as a link. private val establishedClients = Collections.newSetFromMap(ConcurrentHashMap()) + /** + * Addresses refused as a second client link to a peer we already hold a + * client link to, with that peer and the elapsed-realtime moment the + * refusal lapses. + * A peripheral that rotates its address (macOS does, on its own clock) + * advertises under both for a while, and without this the scanner would + * redial the second address on every advertisement, verify it, and be + * refused again, each round holding a connection slot. + */ + private val duplicateUntil = ConcurrentHashMap>() fun registerGatt(address: String, gatt: BluetoothGatt) { gattClients[address] = gatt @@ -108,6 +118,39 @@ class MeshConnectionRegistry { id == deviceId && (establishedClients.contains(address) || serverConnections.contains(address)) } + /** + * True when we hold a client link to [deviceId] at an address other than + * [excluding]. Server links do not count: a peer may hold one link per + * direction, and only a second link in the SAME direction is a duplicate. + */ + fun hasOtherClientLink(deviceId: String, excluding: String): Boolean = + addressToDevice.any { (address, id) -> + id == deviceId && address != excluding && gattClients.containsKey(address) + } + + /** + * Refuse client links to [address], a duplicate of [deviceId], until + * [untilMs] (elapsed realtime). + */ + fun suppressDuplicate(address: String, deviceId: String, untilMs: Long) { + duplicateUntil[address] = deviceId to untilMs + } + + /** + * Whether a dial to [address] is still inside a duplicate refusal. + * + * Lapses early once the client link it duplicated is gone: otherwise a + * peer whose kept link died just after the refusal stayed undialed at the + * address it still advertises for the rest of the window. + */ + fun isSuppressedDuplicate(address: String, nowMs: Long): Boolean { + val entry = duplicateUntil[address] ?: return false + val (deviceId, until) = entry + if (nowMs < until && hasOtherClientLink(deviceId, address)) return true + duplicateUntil.remove(address, entry) + return false + } + /** Every address currently mapped to [deviceId]. */ fun addressesForDevice(deviceId: String): List = addressToDevice.filterValues { it == deviceId }.keys.toList() @@ -161,5 +204,6 @@ class MeshConnectionRegistry { connectionRoles.clear() serverConnections.clear() establishedClients.clear() + duplicateUntil.clear() } } diff --git a/bindings/react-native/android/src/test/java/com/offlineprotocol/ble/DuplicateClientLinkTest.kt b/bindings/react-native/android/src/test/java/com/offlineprotocol/ble/DuplicateClientLinkTest.kt new file mode 100644 index 00000000..d93ea52a --- /dev/null +++ b/bindings/react-native/android/src/test/java/com/offlineprotocol/ble/DuplicateClientLinkTest.kt @@ -0,0 +1,107 @@ +package com.offlineprotocol.ble + +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config +import org.robolectric.shadows.ShadowBluetoothDevice +import org.robolectric.shadows.ShadowBluetoothGatt + +/** + * A macOS peripheral rotates its address and advertises under the old and the + * new one at once, so the scanner dials both and both verify as one peer. The + * second client link is the duplicate: it is refused, and its address is not + * redialed until the suppression lapses. A link in the other direction is not + * a duplicate. + */ +@RunWith(RobolectricTestRunner::class) +@Config(sdk = [34]) +class DuplicateClientLinkTest { + + private fun gatt(address: String) = + ShadowBluetoothGatt.newInstance(ShadowBluetoothDevice.newInstance(address)) + + @Test + fun `a second client link to a peer already held is a duplicate`() { + val registry = MeshConnectionRegistry() + registry.registerGatt("AA:AA:AA:AA:AA:01", gatt("AA:AA:AA:AA:AA:01")) + registry.setDeviceIdentifier("AA:AA:AA:AA:AA:01", "off1mac") + registry.registerGatt("AA:AA:AA:AA:AA:02", gatt("AA:AA:AA:AA:AA:02")) + + assertTrue(registry.hasOtherClientLink("off1mac", excluding = "AA:AA:AA:AA:AA:02")) + assertFalse( + "the kept link is never its own duplicate", + registry.hasOtherClientLink("off1mac", excluding = "AA:AA:AA:AA:AA:01"), + ) + } + + @Test + fun `a server link to the same peer is not a duplicate`() { + val registry = MeshConnectionRegistry() + registry.setDeviceIdentifier("BB:BB:BB:BB:BB:01", "off1mac") + registry.trackServerConnection("BB:BB:BB:BB:BB:01") + registry.registerGatt("BB:BB:BB:BB:BB:02", gatt("BB:BB:BB:BB:BB:02")) + + assertFalse(registry.hasOtherClientLink("off1mac", excluding = "BB:BB:BB:BB:BB:02")) + } + + @Test + fun `a closed client link no longer makes the next one a duplicate`() { + val registry = MeshConnectionRegistry() + registry.registerGatt("CC:CC:CC:CC:CC:01", gatt("CC:CC:CC:CC:CC:01")) + registry.setDeviceIdentifier("CC:CC:CC:CC:CC:01", "off1mac") + registry.removeGatt("CC:CC:CC:CC:CC:01") + + assertFalse(registry.hasOtherClientLink("off1mac", excluding = "CC:CC:CC:CC:CC:02")) + } + + /** A registry holding a client link to `off1mac` at the kept address. */ + private fun holding(kept: String): MeshConnectionRegistry { + val registry = MeshConnectionRegistry() + registry.registerGatt(kept, gatt(kept)) + registry.setDeviceIdentifier(kept, "off1mac") + return registry + } + + @Test + fun `a refused address stays undialed until the suppression lapses`() { + val registry = holding("DD:DD:DD:DD:DD:01") + registry.suppressDuplicate("DD:DD:DD:DD:DD:02", "off1mac", untilMs = 1_000L) + + assertTrue(registry.isSuppressedDuplicate("DD:DD:DD:DD:DD:02", nowMs = 999L)) + assertFalse(registry.isSuppressedDuplicate("DD:DD:DD:DD:DD:02", nowMs = 1_000L)) + assertFalse( + "a lapsed refusal is forgotten", + registry.isSuppressedDuplicate("DD:DD:DD:DD:DD:02", nowMs = 0L), + ) + assertFalse(registry.isSuppressedDuplicate("DD:DD:DD:DD:DD:03", nowMs = 0L)) + } + + @Test + fun `a refusal lapses early once the kept link is gone`() { + val registry = holding("FF:FF:FF:FF:FF:01") + registry.suppressDuplicate("FF:FF:FF:FF:FF:02", "off1mac", untilMs = Long.MAX_VALUE) + assertTrue(registry.isSuppressedDuplicate("FF:FF:FF:FF:FF:02", nowMs = 0L)) + + registry.removeGatt("FF:FF:FF:FF:FF:01") + + assertFalse( + "the peer must be redialable at the address it still advertises", + registry.isSuppressedDuplicate("FF:FF:FF:FF:FF:02", nowMs = 0L), + ) + } + + @Test + fun `clear forgets every refusal`() { + val registry = holding("EE:EE:EE:EE:EE:01") + registry.suppressDuplicate("EE:EE:EE:EE:EE:02", "off1mac", untilMs = Long.MAX_VALUE) + registry.clear() + // The kept link comes back, so only a forgotten refusal answers false. + registry.registerGatt("EE:EE:EE:EE:EE:01", gatt("EE:EE:EE:EE:EE:01")) + registry.setDeviceIdentifier("EE:EE:EE:EE:EE:01", "off1mac") + + assertFalse(registry.isSuppressedDuplicate("EE:EE:EE:EE:EE:02", nowMs = 0L)) + } +} diff --git a/bindings/react-native/ios/BleHelloPolicy.swift b/bindings/react-native/ios/BleHelloPolicy.swift new file mode 100644 index 00000000..13d8cd01 --- /dev/null +++ b/bindings/react-native/ios/BleHelloPolicy.swift @@ -0,0 +1,51 @@ +import Foundation + +/// Whether the central writes its hello on a link, and what it writes. +/// +/// The Bluetooth LE framing chapter's optional Hello characteristic +/// (`6E400006-...`) is how a central says who it is on the link it opened. A +/// peripheral that receives Message writes from a central it cannot name +/// otherwise has two choices, both bad: dial the central back to read its +/// identity (the iOS and Android peripherals do, at a cost of seconds to tens +/// of seconds per new link), or attribute the writes to the central's +/// connection id. The Python service's peripheral does the second, and the core +/// then refuses our key package, whose sender must be the link's identity, so +/// an iPhone could find and verify a Python box and never start a session +/// with it through that link. +/// +/// The value is the identity assertion this device serves on its own Identity +/// characteristic. It is written once per connection, with response, and only +/// when it fits one write. Anything else (a peer with no Hello, an assertion +/// not built yet, one out of the chapter's bounds) writes nothing, and the +/// link is resolved as it was before Hello existed: a hello is never a reason +/// to drop a link. +enum BleHelloPolicy { + /// The chapter's bounds on a hello value. A peripheral refuses anything + /// outside them before verifying, so writing one would only spend a write. + static let minLength = 96 + static let maxLength = 512 + + /// The bytes to write, or nil to write none. + /// + /// - Parameters: + /// - identity: the encoded assertion this device serves, if it has one. + /// - peerServesHello: whether the chosen service instance has Hello. + /// - alreadyWritten: whether this connection already wrote its hello. + /// Characteristic discovery replays on a live link, and the hello is + /// per connection. + /// - maximumWriteLength: `maximumWriteValueLength(for: .withoutResponse)`, + /// the payload of one ATT packet. Not the `.withResponse` figure: iOS + /// reports 512 there and sends anything longer than one packet as a + /// prepared write, which a peripheral must refuse for a hello. + static func helloToWrite( + identity: Data?, + peerServesHello: Bool, + alreadyWritten: Bool, + maximumWriteLength: Int + ) -> Data? { + guard peerServesHello, !alreadyWritten, let identity = identity else { return nil } + guard identity.count >= minLength, identity.count <= maxLength else { return nil } + guard identity.count <= maximumWriteLength else { return nil } + return identity + } +} diff --git a/bindings/react-native/ios/BleManager.swift b/bindings/react-native/ios/BleManager.swift index 38d9bb42..a5e66a9a 100644 --- a/bindings/react-native/ios/BleManager.swift +++ b/bindings/react-native/ios/BleManager.swift @@ -48,6 +48,8 @@ public class BleManager: NSObject, TransportManager { private let DEVICE_ID_CHAR_UUID = CBUUID(string: "6E400003-B5A3-F393-E0A9-E50E24DCCA9E") private let IDENTITY_CHAR_UUID = CBUUID(string: "6E400004-B5A3-F393-E0A9-E50E24DCCA9E") private let APP_TAG_CHAR_UUID = CBUUID(string: "6E400005-B5A3-F393-E0A9-E50E24DCCA9E") + /// Optional: written once by this central so the peer can name the link (see BleHelloPolicy). + private let HELLO_CHAR_UUID = CBUUID(string: "6E400006-B5A3-F393-E0A9-E50E24DCCA9E") // Fragment sizing is fully owned by the Rust transport now: it stores // a per-peer maximum usable payload seeded from @@ -159,6 +161,9 @@ public class BleManager: NSObject, TransportManager { /// Peripherals already announced via `blePeerDiscovered`, so a re-read of /// either characteristic on a live link cannot announce twice. private var announcedPeripherals: Set = [] + /// Links that already wrote their hello this connection. Cleared with + /// `announcedPeripherals`, so a reconnect says hello again. + private var helloWritten: Set = [] // MARK: - Service instance selection // @@ -789,6 +794,7 @@ public class BleManager: NSObject, TransportManager { advertisedDeviceIds.removeAll() verifiedPeerAddresses.removeAll() announcedPeripherals.removeAll() + helloWritten.removeAll() // The mesh counts the same links; without this it stays full. meshController.registerAllDisconnected() discoveredPeripherals.removeAll() @@ -1212,7 +1218,7 @@ public class BleManager: NSObject, TransportManager { clearServiceInstanceSelection(for: peripheral.identifier) let instances = peripheral.services?.filter { $0.uuid == SERVICE_UUID } ?? [] if instances.count == 1 { - peripheral.discoverCharacteristics([MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID], for: instances[0]) + peripheral.discoverCharacteristics([MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID, HELLO_CHAR_UUID], for: instances[0]) } else { peripheral.discoverServices([SERVICE_UUID]) } @@ -3253,6 +3259,7 @@ extension BleManager: CBCentralManagerDelegate { advertisedDeviceIds.removeValue(forKey: peripheral.identifier) verifiedPeerAddresses.removeValue(forKey: peripheral.identifier) announcedPeripherals.remove(peripheral.identifier) + helloWritten.remove(peripheral.identifier) clearServiceInstanceSelection(for: peripheral.identifier) if logThrottler.shouldLog(key: "disconnect_\(peripheral.identifier.uuidString)", interval: 10) { let errorDescription = (error as NSError?)?.localizedDescription ?? "none" @@ -3373,7 +3380,7 @@ extension BleManager: CBPeripheralDelegate { // instance is ever handshaken; if it is gone, so is the app // this link's identity came from. if instances.contains(where: { $0 === bound }) { - peripheral.discoverCharacteristics([MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID], for: bound) + peripheral.discoverCharacteristics([MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID, HELLO_CHAR_UUID], for: bound) } else { dropLinkForVanishedServiceInstance(peripheral) } @@ -3385,7 +3392,7 @@ extension BleManager: CBPeripheralDelegate { // make; for a link that never had several this clears nothing. clearServiceInstanceSelection(for: peripheral.identifier) for service in services where service.uuid == SERVICE_UUID { - peripheral.discoverCharacteristics([MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID], for: service) + peripheral.discoverCharacteristics([MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID, HELLO_CHAR_UUID], for: service) } } emitDiagnostic("info", "Discovered BLE services", context: ["peripheral": peripheral.identifier.uuidString]) @@ -3468,6 +3475,30 @@ extension BleManager: CBPeripheralDelegate { // replay that came back partial. guard !announcedPeripherals.contains(peripheral.identifier) else { return } + // Say who we are, once per connection, before the handshake reads and + // so before any Message write: CoreBluetooth runs a peripheral's GATT + // requests in the order they are issued, after the subscribe above. + // A peer that cannot otherwise name this link (the Python service's + // peripheral) refuses our key package without it. See BleHelloPolicy. + if let helloCharacteristic = characteristics.first(where: { $0.uuid == HELLO_CHAR_UUID }), + let hello = BleHelloPolicy.helloToWrite( + identity: currentSignedIdentity()?.encode(), + peerServesHello: true, + alreadyWritten: helloWritten.contains(peripheral.identifier), + // `.withoutResponse` is the single-packet payload (MTU - 3). + // The `.withResponse` figure is 512 on iOS because CoreBluetooth + // turns a longer write into a prepared write, which the chapter + // forbids for a hello and a peripheral refuses. + maximumWriteLength: peripheral.maximumWriteValueLength(for: .withoutResponse) + ) { + helloWritten.insert(peripheral.identifier) + peripheral.writeValue(hello, for: helloCharacteristic, type: .withResponse) + emitDiagnostic("info", "BLE hello written", context: [ + "peripheral": peripheral.identifier.uuidString, + "length": hello.count, + ]) + } + // Both reads are issued here and complete in either order; the peer is // announced by whichever one lands second, through // `completePeerHandshake`. Neither is sufficient alone — see that @@ -3802,7 +3833,7 @@ extension BleManager: CBPeripheralDelegate { ]) for instance in instances { peripheral.discoverCharacteristics( - [MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID, APP_TAG_CHAR_UUID], + [MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID, HELLO_CHAR_UUID, APP_TAG_CHAR_UUID], for: instance ) } @@ -3922,7 +3953,7 @@ extension BleManager: CBPeripheralDelegate { "tagged": probe.tags.count, "timedOut": timedOut, ]) - peripheral.discoverCharacteristics([MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID], for: chosen) + peripheral.discoverCharacteristics([MESSAGE_CHAR_UUID, DEVICE_ID_CHAR_UUID, IDENTITY_CHAR_UUID, HELLO_CHAR_UUID], for: chosen) } /// Drops everything selection holds for one link. A no-op for a link that @@ -4024,6 +4055,16 @@ extension BleManager: CBPeripheralDelegate { } public func peripheral(_ peripheral: CBPeripheral, didWriteValueFor characteristic: CBCharacteristic, error: Error?) { + if characteristic.uuid == HELLO_CHAR_UUID { + // Best effort: a refused hello leaves the link as it was. + if let error = error { + emitDiagnostic("info", "BLE hello refused", context: [ + "peripheral": peripheral.identifier.uuidString, + "error": error.localizedDescription, + ]) + } + return + } if let error = error { print("[BleManager] Error writing characteristic: \(error)") emitDiagnostic("error", "Error writing characteristic", context: ["error": error.localizedDescription]) diff --git a/bindings/react-native/ios/Package.swift b/bindings/react-native/ios/Package.swift index d5705334..9cb50ff6 100644 --- a/bindings/react-native/ios/Package.swift +++ b/bindings/react-native/ios/Package.swift @@ -61,6 +61,7 @@ let package = Package( "AddressDeclarationPolicy.swift", "BleAppTag.swift", "BleDensityPolicy.swift", + "BleHelloPolicy.swift", "BleMessageNotificationPolicy.swift", "BleServiceInstanceSelection.swift", "EncryptionConfigReader.swift", @@ -118,6 +119,7 @@ let package = Package( "AddressDeclarationPolicyTests.swift", "BleAppTagTests.swift", "BleDensityPolicyTests.swift", + "BleHelloPolicyTests.swift", "BleMessageNotificationPolicyTests.swift", "BleServiceInstanceSelectionTests.swift", "EncryptionConfigReaderTests.swift", diff --git a/bindings/react-native/ios/tests/BleHelloPolicyTests.swift b/bindings/react-native/ios/tests/BleHelloPolicyTests.swift new file mode 100644 index 00000000..63ee7ba9 --- /dev/null +++ b/bindings/react-native/ios/tests/BleHelloPolicyTests.swift @@ -0,0 +1,51 @@ +import XCTest +@testable import OfflineProtocol + +/// Covers `BleHelloPolicy` only. That `BleManager` writes what it returns, after +/// the subscription and before the handshake reads, is pinned in +/// `react_native_ios_central_writes_its_hello` in the uniffi crate. +final class BleHelloPolicyTests: XCTestCase { + + private let assertion = Data(repeating: 7, count: 115) + + func testWritesTheAssertionOncePerConnection() { + XCTAssertEqual( + BleHelloPolicy.helloToWrite(identity: assertion, peerServesHello: true, + alreadyWritten: false, maximumWriteLength: 512), + assertion + ) + XCTAssertNil( + BleHelloPolicy.helloToWrite(identity: assertion, peerServesHello: true, + alreadyWritten: true, maximumWriteLength: 512), + "a discovery replay on a live link must not write a second hello" + ) + } + + func testAPeerWithoutHelloGetsNone() { + XCTAssertNil(BleHelloPolicy.helloToWrite(identity: assertion, peerServesHello: false, + alreadyWritten: false, maximumWriteLength: 512)) + } + + func testNoAssertionYetWritesNone() { + XCTAssertNil(BleHelloPolicy.helloToWrite(identity: nil, peerServesHello: true, + alreadyWritten: false, maximumWriteLength: 512)) + } + + func testAnAssertionThatDoesNotFitOneWriteIsSkipped() { + XCTAssertNil(BleHelloPolicy.helloToWrite(identity: assertion, peerServesHello: true, + alreadyWritten: false, maximumWriteLength: 114)) + XCTAssertNotNil(BleHelloPolicy.helloToWrite(identity: assertion, peerServesHello: true, + alreadyWritten: false, maximumWriteLength: 115)) + } + + func testTheChaptersBoundsAreHeld() { + XCTAssertEqual(BleHelloPolicy.minLength, 96) + XCTAssertEqual(BleHelloPolicy.maxLength, 512) + for (count, expected) in [(95, false), (96, true), (512, true), (513, false)] { + let value = Data(repeating: 1, count: count) + let written = BleHelloPolicy.helloToWrite(identity: value, peerServesHello: true, + alreadyWritten: false, maximumWriteLength: 1024) + XCTAssertEqual(written != nil, expected, "length \(count)") + } + } +} diff --git a/crates/offline-protocol-uniffi/src/lib.rs b/crates/offline-protocol-uniffi/src/lib.rs index 0fb5e2e3..4038e6a7 100644 --- a/crates/offline-protocol-uniffi/src/lib.rs +++ b/crates/offline-protocol-uniffi/src/lib.rs @@ -13733,6 +13733,241 @@ mod tests { ); } + /// The iOS central writes its hello: once per connection, after the + /// Message subscription and before the handshake reads. + /// + /// Without it a peripheral that cannot otherwise name the link (the Python + /// service's) attributes our writes to a connection id, and the core + /// refuses our key package, whose sender must be the link's identity: an + /// iPhone could find and verify a Python box and never start a session + /// through that link. `BleHelloPolicy` is unit-tested, but a policy nobody + /// calls passes its own tests, and `BleManager` sits behind CoreBluetooth, + /// so the call site is pinned here: + /// + /// 1. The UUID is the Rust constant, and Hello is discovered on every path + /// that discovers the handshake characteristics. + /// 2. The write goes through the policy, with response, and sits below the + /// announce gate (one-shot per connection) and above the first handshake + /// read (so before any Message write, in CoreBluetooth's GATT order). + /// 3. The per-connection record is cleared wherever `announcedPeripherals` + /// is, or a reconnect would never say hello again. + #[test] + fn react_native_ios_central_writes_its_hello() { + use offline_protocol_transport::constants::BLE_HELLO_CHAR_UUID; + + let swift = rn_source_code_only("ios/BleManager.swift"); + assert!( + swift.contains(&format!("CBUUID(string: \"{BLE_HELLO_CHAR_UUID}\")")), + "BleManager.swift must declare the Hello characteristic as {BLE_HELLO_CHAR_UUID}" + ); + let discoveries: Vec<&str> = swift + .split("discoverCharacteristics(") + .skip(1) + .map(|rest| &rest[..rest.find(']').unwrap_or(rest.len())]) + .filter(|list| list.contains("IDENTITY_CHAR_UUID")) + .collect(); + assert!( + !discoveries.is_empty(), + "BleManager.swift must discover the handshake characteristics somewhere" + ); + for list in &discoveries { + assert!( + list.contains("HELLO_CHAR_UUID"), + "every discoverCharacteristics call that discovers the handshake characteristics \ + must also discover HELLO_CHAR_UUID: a path that omits it hands the discovery \ + delegate no Hello characteristic, and that link never says hello ({list}])" + ); + } + + let body_start = swift + .find( + "public func peripheral(_ peripheral: CBPeripheral, \ + didDiscoverCharacteristicsFor service: CBService, error: Error?) {", + ) + .expect("BleManager.swift must implement didDiscoverCharacteristicsFor"); + let body_end = swift + .find( + "public func peripheral(_ peripheral: CBPeripheral, \ + didUpdateValueFor characteristic: CBCharacteristic, error: Error?) {", + ) + .expect("BleManager.swift must implement didUpdateValueFor"); + let body = &swift[body_start..body_end]; + + let gate = body + .find("guard !announcedPeripherals.contains(peripheral.identifier) else { return }") + .expect("the discovery delegate must keep its announce gate"); + let policy = body.find("BleHelloPolicy.helloToWrite(").expect( + "the hello must be decided by BleHelloPolicy.helloToWrite in the discovery delegate", + ); + let write = body + .find("peripheral.writeValue(hello, for: helloCharacteristic, type: .withResponse)") + .expect("the hello must be written with response in the discovery delegate"); + let first_read = body + .find("peripheral.readValue(for: characteristic)") + .expect("the discovery delegate must issue the handshake reads"); + assert!( + gate < policy && policy < write && write < first_read, + "the hello must sit below the announce gate (one per connection) and above the first \ + handshake read (so it precedes every Message write in CoreBluetooth's GATT order)" + ); + assert!( + body.contains( + "maximumWriteLength: peripheral.maximumWriteValueLength(for: .withoutResponse)" + ), + "the hello must be bounded by one packet's payload: iOS reports 512 for \ + .withResponse and sends anything longer than one packet as a prepared write, which \ + a peripheral refuses for a hello" + ); + assert!( + body.contains("alreadyWritten: helloWritten.contains(peripheral.identifier)") + && body.contains("helloWritten.insert(peripheral.identifier)"), + "the hello must be recorded per connection, or a discovery replay writes it again" + ); + assert_eq!( + swift.matches("helloWritten.removeAll()").count(), + swift.matches("announcedPeripherals.removeAll()").count(), + "helloWritten must be cleared wherever announcedPeripherals is cleared" + ); + assert_eq!( + swift + .matches("helloWritten.remove(peripheral.identifier)") + .count(), + swift + .matches("announcedPeripherals.remove(peripheral.identifier)") + .count(), + "helloWritten must be cleared wherever announcedPeripherals is cleared for one link" + ); + } + + /// The Android central refuses a second client link to a peer it already + /// holds, before announcing it, and leaves the address undialed. + /// + /// A macOS peripheral rotates its address and advertises under both, so + /// the scanner dials both and both verify as one peer. Two such links per + /// peer filled the four connection slots, and the cap then refused every + /// inbound central. `MeshConnectionRegistry` is unit-tested; the order in + /// `handleIdentityRead` and the dial-time check are not reachable from a + /// JVM test, so they are pinned here: + /// + /// 1. The duplicate check runs on the verified arm, before + /// `announceVerifiedPeer`: an announce first would re-point the peer's + /// address at the duplicate and register a second mesh connection. + /// 2. The refusal records the peer with the deadline, so it lapses with + /// the kept link. + /// 3. `connectToDevice` consults the refusal before the connection cap. + #[test] + fn react_native_android_central_refuses_a_second_client_link_to_one_peer() { + let central = rn_source_code_only( + "android/src/main/java/com/offlineprotocol/ble/CentralGattClient.kt", + ); + let facade = rn_source_code_only( + "android/src/main/java/com/offlineprotocol/ble/BleTransportFacade.kt", + ); + let section = |source: &str, start: &str| -> String { + let from = source + .find(start) + .unwrap_or_else(|| panic!("missing {start}")); + let rest = &source[from + start.len()..]; + rest[..rn_kotlin_body_end(rest)].to_string() + }; + + let identity = section(¢ral, "private fun handleIdentityRead("); + let verified = identity + .find("is PeerIdentityBinding.Outcome.Verified ->") + .expect("handleIdentityRead must have a verified arm"); + let check = identity + .find("if (host.connections.hasOtherClientLink(outcome.peerId, address)) {") + .expect("the verified arm must check for another client link to the peer"); + let refuse = identity + .find("refuseDuplicateLink(gatt, outcome.peerId)") + .expect("a duplicate must be refused through refuseDuplicateLink"); + let announce = identity + .find("announceVerifiedPeer(gatt, outcome.peerId)") + .expect("the verified arm must announce the peer"); + assert!( + verified < check && check < refuse && refuse < announce, + "the duplicate check must run on the verified arm before announceVerifiedPeer" + ); + + let refusal = section(¢ral, "private fun refuseDuplicateLink("); + assert!( + refusal.contains("host.connections.suppressDuplicate( address, peerId,"), + "the refusal must record the peer it duplicated, so it lapses with the kept link" + ); + assert!( + refusal.contains("closeGattClient(gatt, \"duplicate_identity\")"), + "the duplicate link must be closed" + ); + assert!( + !refusal.contains("blePeerDiscovered"), + "a duplicate link is closed unannounced" + ); + + let dial = section(&facade, "private fun connectToDevice("); + let suppressed = dial + .find( + "connections.isSuppressedDuplicate(device.address, SystemClock.elapsedRealtime())", + ) + .expect("connectToDevice must consult the duplicate refusal"); + let cap = dial + .find("if (currentConnectionCount() >= MAX_CONNECTIONS_PER_DEVICE) {") + .expect("connectToDevice must keep its connection cap"); + let connect = dial + .find("device.connectGatt(") + .expect("connectToDevice must dial"); + assert!( + suppressed < cap && cap < connect, + "a refused duplicate address must be skipped before the cap and the dial" + ); + } + + /// The hello's length bounds are written down in three languages, and a + /// central and a peripheral that disagree fail silently: a peripheral + /// refuses a hello shorter than its floor before verifying, and a central + /// whose bound is wider spends a write that can only be refused. The + /// floor is the codec's ([`offline_protocol_sealed::IDENTITY_ASSERTION_MIN_LEN`]); + /// the ceiling is one attribute value, held by the Android server, which + /// was the first to serve Hello. + #[test] + fn ble_hello_length_bounds_agree_in_every_binding() { + let floor = offline_protocol_sealed::IDENTITY_ASSERTION_MIN_LEN; + let server = rn_source_code_only( + "android/src/main/java/com/offlineprotocol/ble/PeripheralGattServer.kt", + ); + let marker = "const val MAX_HELLO_WRITE_BYTES = "; + let at = server + .find(marker) + .expect("PeripheralGattServer.kt must declare MAX_HELLO_WRITE_BYTES") + + marker.len(); + let ceiling: usize = server[at..] + .split_whitespace() + .next() + .and_then(|v| v.parse().ok()) + .expect("MAX_HELLO_WRITE_BYTES must be an integer literal"); + + let python = python_module_constants("offline_protocol_sdk/ble_manager.py"); + assert_eq!( + python("HELLO_MIN_LEN"), + floor.to_string(), + "ble_manager.py HELLO_MIN_LEN must equal IDENTITY_ASSERTION_MIN_LEN" + ); + assert_eq!( + python("HELLO_MAX_LEN"), + ceiling.to_string(), + "ble_manager.py HELLO_MAX_LEN must equal Android's MAX_HELLO_WRITE_BYTES" + ); + + let swift = rn_source_code_only("ios/BleHelloPolicy.swift"); + assert!( + swift.contains(&format!("static let minLength = {floor}")), + "BleHelloPolicy.swift minLength must equal IDENTITY_ASSERTION_MIN_LEN ({floor})" + ); + assert!( + swift.contains(&format!("static let maxLength = {ceiling}")), + "BleHelloPolicy.swift maxLength must equal Android's MAX_HELLO_WRITE_BYTES ({ceiling})" + ); + } + /// A phone running several SDK apps presents one instance of the service /// per app behind one BLE link, and each central binds exactly one of them. /// diff --git a/docs/bridges/python.md b/docs/bridges/python.md index 8de5a180..1459ace1 100644 --- a/docs/bridges/python.md +++ b/docs/bridges/python.md @@ -406,7 +406,8 @@ is the caller of the call bless dispatches next only because bless's a coroutine method would be scheduled after the next call's hook, so a test reads the method dbus_next dispatches and refuses a coroutine. Each inbound write is attributed to its own caller, so two centrals no longer collapse into `ble-peer`. Notifications go to every -subscribed central: neither bless backend addresses one. +subscribed central on BlueZ, which offers no per-central notification; on +macOS see P15. The core's outbound fragment queue is shared by the two roles and offers no peek and no requeue (`ble_return_fragment` is a no-op), so a fragment one @@ -420,6 +421,88 @@ fragments behind it. The peripheral on its own takes nothing while nobody is subscribed. `test_ble_peripheral.py` pins both with a fake bus. Nothing here has run on a board. +## P15. A peripheral link carries the address its hello proved, or it cannot start a session + +**Invariant:** a fragment the peripheral hands to the core carries, as its +transport identity, the address the writing central proved with its hello, +and nothing else proves it. The core refuses a hop-0 control frame whose +sender is not the link's identity (`TRANSPORT_IDENTITY_MISMATCH`), and a key +package is such a frame. A link known by a central UUID, or by the +`ble-peer` placeholder, can carry ciphertext for a session that already +exists but can never start one. That is how an Android phone and a Mac found +each other, verified each other, and never formed a session: the Mac rotates +its address, the phone held a link to each address, so the Mac's peripheral +saw two centrals, attributed every write to `ble-peer`, and the core refused +the phone's key package. + +The pieces, each of which the failure needs only one of to come back: + +- **The writer of each write is named.** bless hands a CoreBluetooth write to + its callback without its `CBCentral`. The peripheral builds bless's server + with a subclass of its delegate (swapped into bless's module for that one + construction, under a lock) that passes each request's central, offset and + value through, and answers with the ATT code the peripheral returns. On + BlueZ the bus hook of P14 names the writer. +- **The peripheral serves Hello** ([the chapter](../spec/ble-framing.md), + writable with response) and runs the chapter's order: refuse an offset write + (`0x07`) or a value outside 96 to 512 bytes (`0x0D`) before verifying; + verify with the core's one verifier; bind the central to the derived address; + never rebind; announce the peer only when no other link has, which includes + the central role's client link (`ProtocolManager` hands the peripheral + `BleManager.holds_peer`). A message's `sender` field never replaces a hello + binding: it proves nothing. A central that says hello and never subscribes + is not a central the monitor tracks, so its binding is dropped, and its peer + reported lost unless another link reaches it, once it has gone 5 seconds + without a subscription. Kept, it left a route to the address it proved for + the life of the process. +- **The central writes its hello** after its Message subscription and before + its first write, when the peer serves Hello and the assertion fits one write. + So do the Android and iOS centrals, which is what lets a phone start a + session with this peripheral at all. "Fits one write" means one ATT packet: + iOS bounds it by `maximumWriteValueLength(for: .withoutResponse)`, because + the `.withResponse` figure is 512 there and a longer write goes out as a + prepared write, which this peripheral refuses. `BleManager` takes the bound + from the Hello characteristic's `max_write_without_response_size`, never from + the client's `mtu_size`: bleak's BlueZ client reports 23 until the MTU is + explicitly acquired, which skipped every hello from Linux. `BleManager` also + announces the peer only after the hello's response (or a 3 second deadline), + because the announce is what makes the core push a key package, and one + written ahead of the hello is refused by a peripheral that cannot yet name + the link. The peripheral refuses a batch carrying a hello (a prepared write) + as a whole, before any request in it reaches the verifier. +- **One client link per peer.** A second link that verifies as a peer already + held is closed unannounced, on Android and in `BleManager`, and its address + is left undialed for a minute or until the kept link is gone, whichever + comes first. Kept, two such peers fill Android's four connection slots, and + the cap then refuses every inbound central. Without the early lapse a peer + whose kept link died stayed undialed at the address it still advertises. +- **On macOS a notification goes to the recipient's central** when that + central said hello, through `updateValue:forCharacteristic:onSubscribedCentrals:`, + and a `NO` from it (a full transmit queue) waits for the stack's ready + signal and is retried. bless ignores that `NO`, which dropped fragments of + every burst. +- **A dead client link is found.** CoreBluetooth kept a link to a Mac whose + service process had restarted "connected" for over five minutes while every + write into it vanished. `BleManager` reads each link's Device id back every + 15 seconds and drops a link whose read fails, hangs past 5 seconds, or names + another address. A peer whose central is still bound on our peripheral is + not reported lost when our client link goes, in either direction + (`holds_peer` on each role, wired by `ProtocolManager`): a notification still + reaches it, and a lost report would drop the core's route. +- **A Bluetooth stack that never comes up fails the start.** bless's + CoreBluetooth delegate waits in its constructor, with no deadline, for a + power-on that never comes when the process is not authorised (a service + started over ssh). On the loop thread that froze the whole service past the + local API server's start deadline. The server is now built on a daemon + thread with a 20 second deadline, and a process macOS reports as denied or + restricted is refused before the stack is touched. A construction that + never returned keeps bless's delegate swapped and its lock held, so a later + start in the same process waits 2 seconds for it and then says so, rather + than waiting out the deadline and blaming the adapter. + +`test_ble_hello.py` pins each of these, and each has a mutation that turns it +red. The macOS pieces were run between two Macs and a Galaxy M36 (Android 16). + ## Testing ```bash diff --git a/examples/offline-first/README.md b/examples/offline-first/README.md index 2d9c1e8e..c7524ffe 100644 --- a/examples/offline-first/README.md +++ b/examples/offline-first/README.md @@ -118,10 +118,13 @@ each: it is retried down that stream until keepalive ends it (about 30 s), and only the next retry is handed to the neighbours. That is why scenario 4 stops C before taking it off the network A shares with it. -- **One phone per Linux box over Bluetooth LE.** The box's peripheral learns - which phone wrote to it, but it maps a phone to its user id only while that - phone is the one central connected, so with two a reply has no route back - through the peripheral. +- **Every phone receives every notification on a Linux box.** A phone that + writes the Bluetooth LE Hello (Android, and iOS from this release) is bound + to its address on the box's peripheral however many centrals are connected. + BlueZ offers no per-central notification, so on a Linux box every subscribed + phone receives every fragment, and the core drops what is not addressed to + it. An app built on an older SDK writes no hello and is mapped to its user id + only while it is the one central connected. - **An image built from 0.28.0 or earlier** never gets the receipt in scenario 4: the engine drops an acknowledgement for a message whose only route was the mesh (#537). Pass `--allow-missing-hop-receipt` to run the