From 10560b719d93e476bd7b76a54449940fdea01177 Mon Sep 17 00:00:00 2001 From: gba Date: Mon, 27 Jul 2026 06:38:10 -0700 Subject: [PATCH] Add BlueZ capture backend: Bluetooth Remote ID with no extra hardware ble_capture requires a Sniffle nRF52 dongle, so a box with no add-on hardware decodes no Bluetooth Remote ID at all. Every Raspberry Pi already has a radio that can hear ASTM F3411 advertisements -- verified by capturing real DroneBeacon DB120 frames with btmon on an AryaOS box. New bluez_capture module, selected by FEED_URL scheme ble+hci://hci0. It reuses ble_parse for ODID extraction and emits the same (pack, meta) contract BleWorker uses, so the whole downstream chain is unchanged. Two cooperating pieces: * Scan driver (always): BlueZ D-Bus SetDiscoveryFilter(Transport=le) + StartDiscovery. An ordinary D-Bus client, so it coexists with bluetoothd and an active Bluetooth PAN instead of seizing the adapter. * Reader, selectable. 'monitor' uses an HCI_CHANNEL_MONITOR socket -- the passive tap btmon uses -- for every advertising report with full AD bytes and per-frame RSSI, nothing coalesced; needs CAP_NET_RAW. 'dbus' reads ServiceData off org.bluez.Device1, needs no capability but BlueZ may coalesce repeats, and for Remote ID a coalesced repeat is a lost message TYPE rather than a duplicate. 'auto' tries monitor then falls back. CPython's socket.bind() for BTPROTO_HCI accepts only (device_id,) and raises "bind(): wrong format" for a channel, so it cannot select the monitor channel at all; bind_hci_monitor() falls back to calling bind(2) via ctypes with a real struct sockaddr_hci. Confirmed correct against a live adapter: the kernel returns EPERM (the CAP_NET_RAW check) rather than EINVAL, so the channel is parsed as MONITOR. Routing note: "ble+hci" contains "ble", so its branch must precede the substring test in create_tasks(). Covered by a test. Captures LEGACY advertising only -- the ASTM long-range profile uses BT5 extended advertising on the Coded PHY, which the Pi's CYW43455 is not known to receive. Documented as complementary to a Sniffle dongle or DroneScout receiver, not a replacement. 18 new tests, decoding real captured advertisement bytes through synthesized HCI monitor frames so they run in CI with no adapter. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_0197da7dhvcPoHxYKamrYqyM --- docs/configuration.md | 42 ++- src/dronecot/__init__.py | 1 + src/dronecot/bluez_capture.py | 521 ++++++++++++++++++++++++++++++++++ src/dronecot/classes.py | 61 ++++ src/dronecot/functions.py | 7 +- tests/test_bluez_capture.py | 264 +++++++++++++++++ 6 files changed, 894 insertions(+), 2 deletions(-) create mode 100644 src/dronecot/bluez_capture.py create mode 100644 tests/test_bluez_capture.py diff --git a/docs/configuration.md b/docs/configuration.md index 7d7761d..1a5f187 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -39,10 +39,11 @@ Selects the input worker. DroneCOT inspects the URL scheme: | `wifi://wlan0` | Wi-Fi | Linux monitor mode (Beacon + NAN) | | `wifi+pcap:///path/file.pcapng` | Wi-Fi | Offline pcap replay | | `ble:///dev/ttyUSB0` or `ble://auto` | BLE | Sniffle USB dongle | +| `ble+hci://hci0` | BLE | Onboard BlueZ adapter — no extra hardware | | `wireless://wlan0` | Wi-Fi + BLE | Both workers (set `BLE_SERIAL` if needed) | !!! note - Worker selection is based on substrings in `FEED_URL` (`wireless` is checked before `wifi`). See [Feeds](feeds.md). + Worker selection is based on substrings in `FEED_URL` (`wireless` is checked before `wifi`, and `ble+hci` before `ble`). See [Feeds](feeds.md). **Serial URL forms:** @@ -94,6 +95,45 @@ Requires [Sniffle](https://github.com/nccgroup/Sniffle) Python CLI on `PYTHONPAT --- +## BLE (onboard BlueZ adapter) + +Used when `FEED_URL` uses the `ble+hci://` scheme. Captures Remote ID on the +Bluetooth radio the board already has — no Sniffle dongle, no extra hardware. + +| Key | Default | Description | +|-----|---------|-------------| +| `BLE_ADAPTER` | from `FEED_URL` or `hci0` | BlueZ adapter to use | +| `BLE_READER` | `auto` | `monitor`, `dbus`, or `auto` (see below) | +| `BLE_RSSI_THRESHOLD` | — | Drop advertisements weaker than this (dBm) | + +DroneCOT always drives a continuous LE scan over BlueZ D-Bus +(`SetDiscoveryFilter` with `Transport: le`, then `StartDiscovery`). Because that +is an ordinary D-Bus client it **coexists with `bluetoothd` and with an active +Bluetooth PAN** rather than seizing the adapter. + +Two ways to read the advertisements: + +- **`monitor`** — an `HCI_CHANNEL_MONITOR` socket, the same passive tap `btmon` + uses. Delivers every advertising report with full advertisement bytes and + per-frame RSSI, with nothing coalesced. Needs `CAP_NET_RAW` + (systemd: `AmbientCapabilities=CAP_NET_RAW`). +- **`dbus`** — reads the `ServiceData` property off `org.bluez.Device1`. Needs no + elevated capability, but BlueZ may coalesce repeated advertisements. For + Remote ID a coalesced repeat is a lost message *type*, not a duplicate, so + prefer `monitor` where you can. +- **`auto`** (default) — try `monitor`, fall back to `dbus`. + +Query options may also be given in the URL: +`ble+hci://hci0?reader=monitor&rssi=-95&duplicates=1` + +!!! warning "Legacy advertising only" + This path captures **legacy** advertisements. The ASTM long-range profile + uses BT5 extended advertising on the Coded PHY, which the Raspberry Pi's + CYW43455 is not known to receive. Treat onboard capture as complementary to + a Sniffle dongle or a DroneScout receiver, not a replacement. + +--- + ## MQTT Used when `FEED_URL` contains `mqtt`. Broker host and port are taken from `FEED_URL` (not separate `MQTT_BROKER` / `MQTT_PORT` variables). diff --git a/src/dronecot/__init__.py b/src/dronecot/__init__.py index 5710e66..aec0c20 100644 --- a/src/dronecot/__init__.py +++ b/src/dronecot/__init__.py @@ -90,6 +90,7 @@ from .classes import ( # NOQA BleWorker, + BlueZWorker, MQTTWorker, RIDWorker, RXMockWorker, diff --git a/src/dronecot/bluez_capture.py b/src/dronecot/bluez_capture.py new file mode 100644 index 0000000..ea379ec --- /dev/null +++ b/src/dronecot/bluez_capture.py @@ -0,0 +1,521 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# +# Copyright Sensors & Signals LLC https://www.snstac.com +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Bluetooth Remote ID capture on a stock BlueZ adapter -- no Sniffle dongle. + +``ble_capture`` needs a Sniffle nRF52 dongle, which means a box with no extra +hardware decodes no Bluetooth Remote ID at all. Every Raspberry Pi already has a +Bluetooth radio that can hear ASTM F3411 advertisements, so this module reads +them through BlueZ instead. + +Two cooperating pieces +---------------------- +**Scan driver** (always): BlueZ D-Bus ``SetDiscoveryFilter`` + ``StartDiscovery`` +puts the controller into a continuous LE scan. This is a normal D-Bus client, so +it coexists with ``bluetoothd`` and with an active Bluetooth PAN rather than +seizing the adapter. + +**Reader** (selectable): + +``monitor`` + An ``HCI_CHANNEL_MONITOR`` socket -- the same passive tap ``btmon`` uses. + Delivers every LE Advertising Report the controller receives, with the full + advertisement bytes and per-frame RSSI, and nothing is coalesced. Needs + ``CAP_NET_RAW``. It is purely passive, so it only sees traffic while the + scan driver (or anything else) is scanning. + +``dbus`` + Reads the ``ServiceData`` property off ``org.bluez.Device1``. Requires no + elevated capability, but BlueZ may coalesce repeated advertisements from one + device, which for Remote ID means dropped messages -- a transmitter rotates + through message types, so a dropped repeat is a lost message *type*, not a + duplicate. + +``auto`` (default) + Try ``monitor``; fall back to ``dbus`` if the socket cannot be opened. + +Known limitation +---------------- +This captures **legacy** advertisements. The ASTM long-range profile uses BT5 +extended advertising on the Coded PHY, which the Pi's CYW43455 is not known to +receive. Treat onboard capture as complementary to a Sniffle dongle or a +DroneScout receiver, not a replacement. +""" + +import logging +import os +import socket +import struct +import threading + +from typing import Callable, Dict, List, Optional, Tuple +from urllib.parse import parse_qs, urlparse + +_logger = logging.getLogger(__name__) + +# --- BlueZ / HCI constants ------------------------------------------------ + +AF_BLUETOOTH = 31 +BTPROTO_HCI = 1 +HCI_CHANNEL_MONITOR = 2 +HCI_DEV_NONE = 0xFFFF + +# struct hci_mon_hdr { __le16 opcode; __le16 index; __le16 len; } +HCI_MON_HDR_LEN = 6 +HCI_MON_EVENT_PKT = 0x0003 + +HCI_EVENT_LE_META = 0x3E +LE_ADVERTISING_REPORT = 0x02 +LE_EXTENDED_ADVERTISING_REPORT = 0x0D + +BLUEZ_SERVICE = "org.bluez" +ADAPTER_IFACE = "org.bluez.Adapter1" +DEVICE_IFACE = "org.bluez.Device1" + +# 16-bit ASTM UUID 0xFFFA as BlueZ spells it on D-Bus. +ASTM_UUID_DBUS = "0000fffa-0000-1000-8000-00805f9b34fb" + +DEFAULT_ADAPTER = "hci0" + + +def parse_bluez_feed_url(feed_url: str) -> dict: + """Parse a ``ble+hci://`` FEED_URL. + + Examples + -------- + ``ble+hci://hci0`` + ``ble+hci://hci0?reader=monitor&rssi=-95`` + """ + parsed = urlparse(feed_url) + path = (parsed.path or "").strip() + if path.startswith("//"): + path = path[2:] + adapter = parsed.hostname or path.strip("/") or DEFAULT_ADAPTER + if adapter == "auto": + adapter = DEFAULT_ADAPTER + + qs = parse_qs(parsed.query or "") + reader = qs.get("reader", ["auto"])[0].lower() + if reader not in {"auto", "monitor", "dbus"}: + _logger.warning("Unknown BLE reader %r, using 'auto'", reader) + reader = "auto" + + rssi_raw = qs.get("rssi", [None])[0] + duplicates = qs.get("duplicates", ["1"])[0].lower() in {"1", "true", "yes"} + + return { + "adapter": adapter, + "reader": reader, + "rssi_threshold": int(rssi_raw) if rssi_raw is not None else None, + "duplicates": duplicates, + } + + +def adapter_index(adapter: str) -> int: + """``hci0`` -> ``0``. Used to filter monitor traffic to one controller.""" + digits = "".join(ch for ch in adapter if ch.isdigit()) + return int(digits) if digits else 0 + + +def format_mac(addr_le: bytes) -> str: + """HCI reports addresses little-endian; render them the human way.""" + return ":".join(f"{b:02X}" for b in reversed(addr_le[:6])) + + +def bind_hci_monitor(sock: socket.socket) -> None: + """Bind ``sock`` to the HCI monitor channel. + + CPython's ``socket.bind()`` for ``BTPROTO_HCI`` accepts only ``(device_id,)`` + on current builds -- passing a channel raises "bind(): wrong format" -- so it + cannot select ``HCI_CHANNEL_MONITOR`` at all. Try the two-tuple first in case + the interpreter does support it, then fall back to calling ``bind(2)`` + directly with a proper ``struct sockaddr_hci``. + + Raises ``OSError`` if the bind fails (typically missing ``CAP_NET_RAW``). + """ + try: + sock.bind((HCI_DEV_NONE, HCI_CHANNEL_MONITOR)) + return + except OSError: + pass # interpreter lacks channel support; fall through to ctypes + + import ctypes # pylint: disable=import-outside-toplevel + + libc = ctypes.CDLL("libc.so.6", use_errno=True) + # struct sockaddr_hci { sa_family_t family; unsigned short dev, channel; } + addr = struct.pack(" List[Tuple[str, bytes, Optional[int]]]: + """Decode an LE Advertising Report subevent body. + + Returns ``[(mac, adv_data, rssi), ...]``. Malformed or truncated reports + yield nothing rather than raising -- this runs on radio input. + """ + out: List[Tuple[str, bytes, Optional[int]]] = [] + if len(params) < 2: + return out + + num_reports = params[1] + pos = 2 + for _ in range(num_reports): + # event_type(1) addr_type(1) addr(6) data_len(1) + if pos + 9 > len(params): + break + addr = params[pos + 2 : pos + 8] + data_len = params[pos + 8] + pos += 9 + if pos + data_len > len(params): + break + adv_data = params[pos : pos + data_len] + pos += data_len + rssi = None + if pos < len(params): + rssi = struct.unpack("b", params[pos : pos + 1])[0] + pos += 1 + out.append((format_mac(addr), adv_data, rssi)) + return out + + +def parse_le_extended_advertising_report( + params: bytes, +) -> List[Tuple[str, bytes, Optional[int]]]: + """Decode an LE Extended Advertising Report subevent body (BT5).""" + out: List[Tuple[str, bytes, Optional[int]]] = [] + if len(params) < 2: + return out + + num_reports = params[1] + pos = 2 + for _ in range(num_reports): + # event_type(2) addr_type(1) addr(6) primary_phy(1) secondary_phy(1) + # adv_sid(1) tx_power(1) rssi(1) periodic_interval(2) + # direct_addr_type(1) direct_addr(6) data_len(1) + if pos + 24 > len(params): + break + addr = params[pos + 3 : pos + 9] + rssi = struct.unpack("b", params[pos + 13 : pos + 14])[0] + data_len = params[pos + 23] + pos += 24 + if pos + data_len > len(params): + break + adv_data = params[pos : pos + data_len] + pos += data_len + out.append((format_mac(addr), adv_data, rssi)) + return out + + +def parse_monitor_packet(packet: bytes, index: int) -> List[Tuple[str, bytes, Optional[int]]]: + """Extract advertising reports from one HCI monitor frame. + + ``index`` restricts decoding to a single controller, so a box with more than + one adapter does not attribute another radio's traffic to this sensor. + """ + if len(packet) < HCI_MON_HDR_LEN: + return [] + + opcode, pkt_index, length = struct.unpack_from(" Optional[socket.socket]: + """Open a passive HCI monitor socket, or return None if not permitted.""" + try: + sock = socket.socket( + AF_BLUETOOTH, socket.SOCK_RAW | socket.SOCK_CLOEXEC, BTPROTO_HCI + ) + except (OSError, AttributeError) as exc: + _logger.debug("HCI monitor socket unavailable: %s", exc) + return None + + try: + bind_hci_monitor(sock) + except OSError as exc: + sock.close() + _logger.debug( + "Cannot bind HCI monitor channel (%s); CAP_NET_RAW is required", exc + ) + return None + + return sock + + def _run_monitor(self) -> None: + from dronecot import ble_parse # pylint: disable=import-outside-toplevel + + index = adapter_index(self.adapter) + sock = self._sock + if sock is None: + return + sock.settimeout(1.0) + + while not self._stop.is_set(): + try: + packet = sock.recv(4096) + except socket.timeout: + continue + except OSError as exc: + if self._stop.is_set(): + break + _logger.debug("HCI monitor read error: %s", exc) + continue + + for mac, adv_data, rssi in parse_monitor_packet(packet, index): + self._emit(ble_parse, adv_data, mac, rssi) + + def _emit(self, ble_parse, adv_data: bytes, mac: str, rssi: Optional[int]) -> None: + """Extract ODID from an advertisement and hand it upstream.""" + if self.rssi_threshold is not None and rssi is not None: + if rssi < self.rssi_threshold: + return + + pack = ble_parse.extract_odid_from_adv_data(adv_data) + if not pack: + return + + meta: Dict = {"type": "BLE legacy (BlueZ)", "MAC address": mac} + if rssi is not None: + meta["RSSI"] = rssi + self.on_packet(pack, meta) + + # -- reader: BlueZ D-Bus ServiceData ------------------------------------ + + def _handle_dbus_properties(self, address: Optional[str], props: dict) -> None: + from dronecot import ble_parse # pylint: disable=import-outside-toplevel + + service_data = props.get("ServiceData") + if not service_data: + return + + raw = None + for uuid, value in service_data.items(): + if str(uuid).lower() == ASTM_UUID_DBUS: + raw = bytes(bytearray(value)) + break + if raw is None: + return + + rssi = props.get("RSSI") + rssi = int(rssi) if rssi is not None else None + if self.rssi_threshold is not None and rssi is not None: + if rssi < self.rssi_threshold: + return + + # BlueZ hands back the service-data payload with the 16-bit UUID already + # stripped; put it back so the shared AD parser sees its usual layout. + pack = ble_parse.parse_odid_service_data(ble_parse.ASTM_BLE_UUID + raw) + if not pack: + return + + meta: Dict = {"type": "BLE legacy (BlueZ)"} + if address: + meta["MAC address"] = str(address).upper() + if rssi is not None: + meta["RSSI"] = rssi + self.on_packet(pack, meta) + + def _run_dbus(self) -> None: + """Drive a continuous LE scan and, on the dbus reader, decode from it.""" + try: + import dbus # pylint: disable=import-outside-toplevel + import dbus.mainloop.glib # pylint: disable=import-outside-toplevel + from gi.repository import GLib # pylint: disable=import-outside-toplevel + except ImportError as exc: + raise ImportError( + "BlueZ capture requires python3-dbus and python3-gi " + "(Debian: apt install python3-dbus python3-gi)" + ) from exc + + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + bus = dbus.SystemBus() + adapter_path = f"/org/bluez/{self.adapter}" + + adapter = dbus.Interface( + bus.get_object(BLUEZ_SERVICE, adapter_path), ADAPTER_IFACE + ) + + # Transport 'le' keeps Classic (and therefore an active PAN) untouched. + # DuplicateData asks BlueZ to report repeated advertisement payloads + # rather than coalescing them -- essential when each repeat may carry a + # different ODID message type. + discovery_filter = { + "Transport": "le", + "DuplicateData": dbus.Boolean(self.duplicates), + } + if self.rssi_threshold is not None: + discovery_filter["RSSI"] = dbus.Int16(self.rssi_threshold) + + try: + adapter.SetDiscoveryFilter(discovery_filter) + except dbus.DBusException as exc: + _logger.warning("SetDiscoveryFilter failed (%s); scanning unfiltered", exc) + + try: + adapter.StartDiscovery() + except dbus.DBusException as exc: + # Another client may already be discovering, which is fine -- the + # radio is scanning either way and that is all the tap needs. + _logger.info("StartDiscovery: %s (continuing)", exc) + + if self._active_reader == "dbus": + self._connect_dbus_signals(bus) + + self._loop = GLib.MainLoop() + try: + self._loop.run() + finally: + try: + adapter.StopDiscovery() + except Exception: # pylint: disable=broad-except + pass + + def _connect_dbus_signals(self, bus) -> None: + """Subscribe to advertisement properties for the D-Bus reader.""" + + def on_interfaces_added(_path, interfaces): + props = interfaces.get(DEVICE_IFACE) + if props: + self._handle_dbus_properties(props.get("Address"), props) + + def on_properties_changed(interface, changed, _invalidated, path=None): + if interface != DEVICE_IFACE: + return + address = None + if path: + # /org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF + tail = str(path).rsplit("/", 1)[-1] + if tail.startswith("dev_"): + address = tail[4:].replace("_", ":") + self._handle_dbus_properties(address, changed) + + bus.add_signal_receiver( + on_interfaces_added, + dbus_interface="org.freedesktop.DBus.ObjectManager", + signal_name="InterfacesAdded", + ) + bus.add_signal_receiver( + on_properties_changed, + dbus_interface="org.freedesktop.DBus.Properties", + signal_name="PropertiesChanged", + arg0=DEVICE_IFACE, + path_keyword="path", + ) + + # -- lifecycle ---------------------------------------------------------- + + def start(self) -> None: + self._stop.clear() + + if self.reader in {"auto", "monitor"}: + self._sock = self._open_monitor_socket() + if self._sock is None and self.reader == "monitor": + raise IOError( + "HCI monitor socket unavailable: dronecot needs CAP_NET_RAW " + "for reader=monitor (systemd: AmbientCapabilities=CAP_NET_RAW)" + ) + + self._active_reader = "monitor" if self._sock is not None else "dbus" + if self.reader == "dbus": + self._active_reader = "dbus" + if self._sock is not None: + self._sock.close() + self._sock = None + + # The scan driver runs either way: the monitor tap is passive and sees + # nothing unless the controller is actually scanning. + self._spawn(self._run_dbus, "dronecot-ble-scan") + if self._active_reader == "monitor": + self._spawn(self._run_monitor, "dronecot-ble-monitor") + + _logger.info( + "BlueZ Remote ID capture started on %s (reader=%s)", + self.adapter, + self._active_reader, + ) + + def _spawn(self, target: Callable, name: str) -> None: + thread = threading.Thread(target=target, daemon=True, name=name) + thread.start() + self._threads.append(thread) + + def stop(self) -> None: + self._stop.set() + if self._loop is not None: + try: + self._loop.quit() + except Exception: # pylint: disable=broad-except + pass + for thread in self._threads: + thread.join(timeout=3) + self._threads = [] + if self._sock is not None: + try: + self._sock.close() + except OSError: + pass + self._sock = None diff --git a/src/dronecot/classes.py b/src/dronecot/classes.py index 5d0f0f8..57f6867 100644 --- a/src/dronecot/classes.py +++ b/src/dronecot/classes.py @@ -577,6 +577,67 @@ async def run(self, _=-1) -> None: self._sniffer.stop() +class BlueZWorker(pytak.QueueWorker): + """Queue Worker for BLE Open Drone ID capture on a stock BlueZ adapter. + + Unlike BleWorker this needs no external dongle -- it uses the Bluetooth + radio the board already has. Captures legacy advertisements only; see + bluez_capture for the BT5 Coded PHY caveat. + """ + + def __init__(self, queue, config): + super().__init__(queue, config) + self.config = config + self._loop = None + self._sniffer = None + + def _on_ble_packet(self, pack: bytes, meta: dict) -> None: + merged = {**dronecot.rid_normalize.uas_meta_defaults(self.config), **meta} + pl = dronecot.rid_normalize.bytes_to_rid_dict(pack, merged) + if pl and self._loop: + asyncio.run_coroutine_threadsafe(self.put_queue(pl), self._loop) + + async def run(self, _=-1) -> None: + self._logger.info("Running BlueZWorker") + self._loop = asyncio.get_running_loop() + + try: + from dronecot.bluez_capture import BlueZSniffer, parse_bluez_feed_url + except ImportError as exc: + self._logger.error("BlueZ capture unavailable: %s", exc) + return + + feed = parse_bluez_feed_url(str(self.config.get("FEED_URL", ""))) + + adapter = self.config.get("BLE_ADAPTER") or feed.get("adapter") + reader = self.config.get("BLE_READER") or feed.get("reader") + rssi_threshold = self.config.get("BLE_RSSI_THRESHOLD") + if rssi_threshold is None or str(rssi_threshold) == "": + rssi_threshold = feed.get("rssi_threshold") + else: + rssi_threshold = int(rssi_threshold) + + self._sniffer = BlueZSniffer( + on_packet=self._on_ble_packet, + adapter=adapter, + reader=reader, + rssi_threshold=rssi_threshold, + duplicates=feed.get("duplicates", True), + ) + + try: + self._sniffer.start() + while True: + await asyncio.sleep(3600) + except asyncio.CancelledError: + raise + except Exception as exc: # pylint: disable=broad-except + self._logger.error("BlueZ capture failed: %s", exc) + finally: + if self._sniffer: + self._sniffer.stop() + + class RIDWorker(pytak.QueueWorker): """Queue Worker for RID.""" diff --git a/src/dronecot/functions.py b/src/dronecot/functions.py index 70655ac..17cf85a 100644 --- a/src/dronecot/functions.py +++ b/src/dronecot/functions.py @@ -105,7 +105,8 @@ def create_tasks(config: SectionProxy, clitool: pytak.CLITool) -> Set[pytak.Work UDP_RID_PORT set -> UDPRIDWorker + RIDWorker (pre-decoded RID JSON) wireless:// -> WifiWorker + BleWorker + RIDWorker wifi:// -> WifiWorker + RIDWorker - ble:// -> BleWorker + RIDWorker + ble:// -> BleWorker + RIDWorker (Sniffle dongle) + ble+hci:// -> BlueZWorker + RIDWorker (onboard BlueZ adapter) udp:// -> UDPRIDWorker + RIDWorker mqtt:// -> MQTTWorker + RIDWorker serial:// -> SerialWorker + RIDWorker (default when no key/scheme set) @@ -131,6 +132,10 @@ def create_tasks(config: SectionProxy, clitool: pytak.CLITool) -> Set[pytak.Work elif "wifi" in feed_url: tasks.add(dronecot.WifiWorker(net_queue, config)) tasks.add(dronecot.RIDWorker(clitool.tx_queue, net_queue, config)) + # Must precede the generic "ble" test -- "ble+hci" contains "ble". + elif parsed.scheme == "ble+hci": + tasks.add(dronecot.BlueZWorker(net_queue, config)) + tasks.add(dronecot.RIDWorker(clitool.tx_queue, net_queue, config)) elif "ble" in feed_url: tasks.add(dronecot.BleWorker(net_queue, config)) tasks.add(dronecot.RIDWorker(clitool.tx_queue, net_queue, config)) diff --git a/tests/test_bluez_capture.py b/tests/test_bluez_capture.py new file mode 100644 index 0000000..0c086bd --- /dev/null +++ b/tests/test_bluez_capture.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""Tests for BlueZ-based Bluetooth Remote ID capture. + +The HCI decoding is exercised offline by synthesizing monitor frames around a +REAL ASTM F3411 advertisement captured with btmon on an AryaOS box, so these +run in CI with no Bluetooth adapter present. +""" + +import struct +import unittest + +from dronecot import ble_parse, bluez_capture + + +# Real BLE advertising payload from a BlueMark DroneBeacon DB120, captured off +# the air. AD length 0x1e, type 0x16 (Service Data), UUID 0xFFFA, app code 0x0D, +# message counter 0x22, then a 25-byte ODID System message (type 4, version 2). +CAPTURED_AD = bytes.fromhex( + "1e16faff0d2242008eb78116f15bfcb601000000000000004c08eec73b0e00" +) +CAPTURED_MAC_LE = bytes.fromhex("95 6b d2 11 72 df".replace(" ", "")) +CAPTURED_MAC = "DF:72:11:D2:6B:95" + + +def _monitor_frame(event_params: bytes, index: int = 0) -> bytes: + """Wrap HCI LE Meta event params in the monitor + HCI event headers.""" + hci_event = bytes([bluez_capture.HCI_EVENT_LE_META, len(event_params)]) + event_params + header = struct.pack( + " bytes: + """Build an LE Advertising Report subevent body with one report.""" + return ( + bytes([bluez_capture.LE_ADVERTISING_REPORT, 1]) + + bytes([0x00, 0x01]) # event_type, addr_type + + mac_le + + bytes([len(adv_data)]) + + adv_data + + struct.pack("b", rssi) + ) + + +def _extended_report(adv_data: bytes, mac_le: bytes, rssi: int) -> bytes: + """Build an LE Extended Advertising Report subevent body with one report.""" + return ( + bytes([bluez_capture.LE_EXTENDED_ADVERTISING_REPORT, 1]) + + struct.pack("> 4, 4) # System message + + def test_decodes_extended_advertisement(self): + frame = _monitor_frame(_extended_report(CAPTURED_AD, CAPTURED_MAC_LE, -80)) + reports = bluez_capture.parse_monitor_packet(frame, 0) + + self.assertEqual(len(reports), 1) + mac, adv_data, rssi = reports[0] + self.assertEqual(mac, CAPTURED_MAC) + self.assertEqual(rssi, -80) + self.assertEqual(adv_data, CAPTURED_AD) + + def test_ignores_other_adapters(self): + frame = _monitor_frame(_legacy_report(CAPTURED_AD, CAPTURED_MAC_LE, -67), index=1) + self.assertEqual(bluez_capture.parse_monitor_packet(frame, 0), []) + + def test_ignores_non_event_packets(self): + header = struct.pack("> 4, 4) + self.assertEqual(meta["MAC address"], CAPTURED_MAC) + self.assertEqual(meta["RSSI"], -67) + self.assertEqual(meta["type"], "BLE legacy (BlueZ)") + + def test_non_odid_advertisement_is_ignored(self): + sniffer = self._sniffer() + # A plain "Flags" AD structure, nothing to do with Remote ID. + sniffer._emit(ble_parse, bytes([0x02, 0x01, 0x06]), CAPTURED_MAC, -50) + self.assertEqual(self.packets, []) + + def test_rssi_threshold_filters_weak_frames(self): + sniffer = self._sniffer(rssi_threshold=-70) + sniffer._emit(ble_parse, CAPTURED_AD, CAPTURED_MAC, -90) + self.assertEqual(self.packets, []) + sniffer._emit(ble_parse, CAPTURED_AD, CAPTURED_MAC, -60) + self.assertEqual(len(self.packets), 1) + + +class TestDbusServiceData(unittest.TestCase): + def test_service_data_is_rewrapped_for_the_shared_parser(self): + """BlueZ strips the UUID; we must put it back before parsing.""" + packets = [] + sniffer = bluez_capture.BlueZSniffer( + on_packet=lambda pack, meta: packets.append((pack, meta)) + ) + + # ServiceData value is everything after the 16-bit UUID. + service_value = CAPTURED_AD[4:] + sniffer._handle_dbus_properties( + CAPTURED_MAC, + {"ServiceData": {bluez_capture.ASTM_UUID_DBUS: service_value}, "RSSI": -72}, + ) + + self.assertEqual(len(packets), 1) + pack, meta = packets[0] + self.assertEqual(pack[0] >> 4, 4) + self.assertEqual(meta["MAC address"], CAPTURED_MAC) + self.assertEqual(meta["RSSI"], -72) + + def test_other_service_uuids_ignored(self): + packets = [] + sniffer = bluez_capture.BlueZSniffer( + on_packet=lambda pack, meta: packets.append((pack, meta)) + ) + sniffer._handle_dbus_properties( + CAPTURED_MAC, + {"ServiceData": {"0000180f-0000-1000-8000-00805f9b34fb": b"\x64"}}, + ) + self.assertEqual(packets, []) + + def test_no_service_data_is_ignored(self): + packets = [] + sniffer = bluez_capture.BlueZSniffer( + on_packet=lambda pack, meta: packets.append((pack, meta)) + ) + sniffer._handle_dbus_properties(CAPTURED_MAC, {"RSSI": -60}) + self.assertEqual(packets, []) + + +class TestRouting(unittest.TestCase): + """`ble+hci` contains the substring `ble`, so branch order matters.""" + + def _workers(self, feed_url): + import asyncio + import dronecot + + class _CliTool: # minimal stand-in for pytak.CLITool + def __init__(self): + self.tx_queue = asyncio.Queue() + + async def build(): + return dronecot.create_tasks({"FEED_URL": feed_url}, _CliTool()) + + return {type(t).__name__ for t in asyncio.run(build())} + + def test_ble_hci_routes_to_bluez_worker(self): + names = self._workers("ble+hci://hci0") + self.assertIn("BlueZWorker", names) + self.assertNotIn("BleWorker", names) + self.assertIn("RIDWorker", names) + + def test_plain_ble_still_routes_to_sniffle_worker(self): + names = self._workers("ble://auto") + self.assertIn("BleWorker", names) + self.assertNotIn("BlueZWorker", names) + + +if __name__ == "__main__": + unittest.main()