mirror of
https://github.com/jmcorgan/fips.git
synced 2026-08-09 08:14:42 +00:00
Add BLE L2CAP transport with scan-based auto-connect
BLE transport implementation using L2CAP Connection-Oriented Channels (SeqPacket mode) via the bluer crate, behind cfg(feature = "ble"). Core transport: - BleTransport<I> generic over BleIo trait (BluerIo prod, MockBleIo test) - Connection pool with priority eviction (static > discovered, max 7) - Connect-on-send via connect_inline() matching TCP behavior - Per-connection receive loops with pool cleanup on disconnect Discovery and probing: - Combined scan_probe_loop using select! over scanner events and a BinaryHeap delay queue with per-entry random jitter (0-5s) to prevent herd effects when multiple nodes see the same beacon simultaneously - Pre-handshake pubkey exchange ([0x00][pubkey:32]) for IK identity - Cross-probe tie-breaker: smaller NodeAddr's outbound wins (same convention as FMP/FSP rekey dual-initiation) - Probed peers reported to DiscoveryBuffer; pool fills through normal node-layer auto-connect -> send_async -> connect_inline path Beacon management: - Periodic advertising: 1s burst every 30s (configurable via beacon_interval_secs / beacon_duration_secs) - FIPS service UUID for scan filtering Configuration (all fields optional with defaults): - adapter, psm, mtu, max_connections, connect_timeout_ms - advertise, scan, auto_connect, accept_connections - beacon_interval_secs (30), beacon_duration_secs (1) Hardware validated with two BLE nodes: - 2048-byte MTU, ~60-160ms RTT, zero-config auto-connect - BLE spike tool at testing/ble/ for standalone adapter validation 42 unit tests + 4 node-level integration tests, all CI-compatible via MockBleIo (no hardware required). tokio test-util added for time-dependent scan/probe tests.
This commit is contained in:
+25
-1
@@ -118,6 +118,30 @@ impl Node {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
} else if addr.transport == "ble" {
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
match self.resolve_ble_addr(&addr.addr) {
|
||||
Ok(result) => result,
|
||||
Err(e) => {
|
||||
debug!(
|
||||
transport = %addr.transport,
|
||||
addr = %addr.addr,
|
||||
error = %e,
|
||||
"Failed to resolve BLE address"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
{
|
||||
debug!(
|
||||
transport = %addr.transport,
|
||||
"BLE transport not available on this platform"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
// Find a transport matching this address type
|
||||
let tid = match self.find_transport_for_type(&addr.transport) {
|
||||
@@ -513,7 +537,7 @@ impl Node {
|
||||
self.packet_rx = Some(packet_rx);
|
||||
|
||||
// Initialize transports first (before TUN)
|
||||
let transport_handles = self.create_transports(&packet_tx);
|
||||
let transport_handles = self.create_transports(&packet_tx).await;
|
||||
|
||||
for mut handle in transport_handles {
|
||||
let transport_id = handle.transport_id();
|
||||
|
||||
+82
-1
@@ -675,7 +675,7 @@ impl Node {
|
||||
/// Create transport instances from configuration.
|
||||
///
|
||||
/// Returns a vector of TransportHandles for all configured transports.
|
||||
fn create_transports(&mut self, packet_tx: &PacketTx) -> Vec<TransportHandle> {
|
||||
async fn create_transports(&mut self, packet_tx: &PacketTx) -> Vec<TransportHandle> {
|
||||
let mut transports = Vec::new();
|
||||
|
||||
// Collect UDP configs with optional names to avoid borrow conflicts
|
||||
@@ -744,6 +744,47 @@ impl Node {
|
||||
transports.push(TransportHandle::Tor(tor));
|
||||
}
|
||||
|
||||
// Create BLE transport instances
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
let ble_instances: Vec<_> = self
|
||||
.config
|
||||
.transports
|
||||
.ble
|
||||
.iter()
|
||||
.map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
|
||||
.collect();
|
||||
|
||||
#[cfg(all(feature = "ble", not(test)))]
|
||||
for (name, ble_config) in ble_instances {
|
||||
let transport_id = self.allocate_transport_id();
|
||||
let adapter = ble_config.adapter().to_string();
|
||||
let mtu = ble_config.mtu();
|
||||
match crate::transport::ble::io::BluerIo::new(&adapter, mtu).await {
|
||||
Ok(io) => {
|
||||
let mut ble = crate::transport::ble::BleTransport::new(
|
||||
transport_id,
|
||||
name,
|
||||
ble_config,
|
||||
io,
|
||||
packet_tx.clone(),
|
||||
);
|
||||
ble.set_local_pubkey(self.identity.pubkey().serialize());
|
||||
transports.push(TransportHandle::Ble(ble));
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(adapter = %adapter, error = %e, "failed to initialize BLE adapter");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(any(not(feature = "ble"), test))]
|
||||
if !ble_instances.is_empty() {
|
||||
#[cfg(not(test))]
|
||||
tracing::warn!("BLE transport configured but 'ble' feature not enabled at compile time");
|
||||
}
|
||||
}
|
||||
|
||||
transports
|
||||
}
|
||||
|
||||
@@ -806,6 +847,46 @@ impl Node {
|
||||
Ok((transport_id, TransportAddr::from_bytes(&mac)))
|
||||
}
|
||||
|
||||
/// Resolve a BLE address string (`"adapter/AA:BB:CC:DD:EE:FF"`) to a
|
||||
/// (TransportId, TransportAddr) pair by finding the BLE transport
|
||||
/// instance matching the adapter name.
|
||||
#[cfg(target_os = "linux")]
|
||||
fn resolve_ble_addr(
|
||||
&self,
|
||||
addr_str: &str,
|
||||
) -> Result<(TransportId, TransportAddr), NodeError> {
|
||||
let ta = TransportAddr::from_string(addr_str);
|
||||
let adapter = crate::transport::ble::addr::adapter_from_addr(&ta)
|
||||
.ok_or_else(|| {
|
||||
NodeError::NoTransportForType(format!(
|
||||
"invalid BLE address format '{}': expected 'adapter/mac'",
|
||||
addr_str
|
||||
))
|
||||
})?;
|
||||
|
||||
// Find the BLE transport for this adapter
|
||||
let transport_id = self
|
||||
.transports
|
||||
.iter()
|
||||
.find(|(_, handle)| {
|
||||
handle.transport_type().name == "ble" && handle.is_operational()
|
||||
})
|
||||
.map(|(id, _)| *id)
|
||||
.ok_or_else(|| {
|
||||
NodeError::NoTransportForType(format!(
|
||||
"no operational BLE transport for adapter '{}'",
|
||||
adapter
|
||||
))
|
||||
})?;
|
||||
|
||||
// Validate the address format
|
||||
crate::transport::ble::addr::BleAddr::parse(addr_str).map_err(|e| {
|
||||
NodeError::NoTransportForType(format!("invalid BLE address '{}': {}", addr_str, e))
|
||||
})?;
|
||||
|
||||
Ok((transport_id, TransportAddr::from_string(addr_str)))
|
||||
}
|
||||
|
||||
// === Identity Accessors ===
|
||||
|
||||
/// Get this node's identity.
|
||||
|
||||
@@ -0,0 +1,277 @@
|
||||
//! BLE transport integration tests.
|
||||
//!
|
||||
//! Tests that the BLE transport works end-to-end at the node level:
|
||||
//! handshake, spanning tree convergence, mixed-transport routing.
|
||||
//! All tests use MockBleIo (in-memory channels, no hardware needed).
|
||||
|
||||
use super::*;
|
||||
use crate::config::BleConfig;
|
||||
use crate::transport::ble::addr::BleAddr;
|
||||
use crate::transport::ble::io::{MockBleIo, MockBleStream};
|
||||
use crate::transport::ble::BleTransport;
|
||||
use crate::transport::{packet_channel, Transport, TransportHandle, TransportId};
|
||||
use spanning_tree::{
|
||||
cleanup_nodes, drain_all_packets, initiate_handshake, verify_tree_convergence, TestNode,
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex as StdMutex};
|
||||
|
||||
/// Generate a deterministic BLE address for test node `n`.
|
||||
fn ble_addr(n: u8) -> BleAddr {
|
||||
BleAddr {
|
||||
adapter: "hci0".to_string(),
|
||||
device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n],
|
||||
}
|
||||
}
|
||||
|
||||
/// A pre-connected stream bank for MockBleIo connect handlers.
|
||||
///
|
||||
/// When a connect handler fires, it looks up the target address in this
|
||||
/// bank and returns the pre-created stream. The peer end should be
|
||||
/// injected into the target node's acceptor separately.
|
||||
type StreamBank = Arc<StdMutex<HashMap<String, MockBleStream>>>;
|
||||
|
||||
/// Create a test node with a BLE transport backed by MockBleIo.
|
||||
///
|
||||
/// Returns the TestNode and its MockBleIo (via Arc inside the transport)
|
||||
/// for test injection of connections and scan results.
|
||||
async fn make_test_node_ble(node_num: u8) -> TestNode {
|
||||
let mut node = make_node();
|
||||
let transport_id = TransportId::new(1);
|
||||
let addr = ble_addr(node_num);
|
||||
|
||||
let config = BleConfig {
|
||||
adapter: Some("hci0".to_string()),
|
||||
mtu: Some(2048),
|
||||
accept_connections: Some(true),
|
||||
scan: Some(false), // no auto-scan in tests
|
||||
advertise: Some(false), // no advertising in tests
|
||||
auto_connect: Some(false),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let io = MockBleIo::new("hci0", addr.clone());
|
||||
let (packet_tx, packet_rx) = packet_channel(256);
|
||||
let mut transport = BleTransport::new(transport_id, None, config, io, packet_tx);
|
||||
transport.start_async().await.unwrap();
|
||||
|
||||
let ta = addr.to_transport_addr();
|
||||
|
||||
node.transports
|
||||
.insert(transport_id, TransportHandle::Ble(transport));
|
||||
|
||||
TestNode {
|
||||
node,
|
||||
transport_id,
|
||||
packet_rx,
|
||||
addr: ta,
|
||||
}
|
||||
}
|
||||
|
||||
/// Extract the BleAddr from a TestNode's TransportAddr.
|
||||
fn node_ble_addr(node: &TestNode) -> BleAddr {
|
||||
BleAddr::parse(node.addr.as_str().unwrap()).unwrap()
|
||||
}
|
||||
|
||||
/// Wire a unidirectional BLE connection from node `i` to node `j`.
|
||||
///
|
||||
/// Creates a MockBleStream pair, deposits one end in a stream bank for
|
||||
/// node i's connect handler, and injects the other end into node j's
|
||||
/// accept loop. Must be called after `make_test_node_ble()` and before
|
||||
/// `initiate_handshake()`.
|
||||
async fn wire_ble_connection(nodes: &[TestNode], i: usize, j: usize, bank: &StreamBank) {
|
||||
let addr_i = node_ble_addr(&nodes[i]);
|
||||
let addr_j = node_ble_addr(&nodes[j]);
|
||||
|
||||
let (stream_i, stream_j) = MockBleStream::pair(addr_j.clone(), addr_i.clone(), 2048);
|
||||
|
||||
// Store stream_i in the bank keyed by node j's address string.
|
||||
// When node i connects to node j, the handler returns this stream.
|
||||
let key = nodes[j].addr.to_string();
|
||||
bank.lock().unwrap().insert(key, stream_i);
|
||||
|
||||
// Inject stream_j into node j's accept loop so it sees the inbound.
|
||||
let transport_j = nodes[j].node.transports.get(&nodes[j].transport_id).unwrap();
|
||||
match transport_j {
|
||||
TransportHandle::Ble(t) => {
|
||||
t.io().inject_inbound(stream_j).await;
|
||||
}
|
||||
_ => panic!("expected BLE transport"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Install a connect handler on node `i` that draws from the stream bank.
|
||||
fn install_connect_handler(nodes: &[TestNode], i: usize, bank: &StreamBank) {
|
||||
let bank = Arc::clone(bank);
|
||||
let transport_i = nodes[i].node.transports.get(&nodes[i].transport_id).unwrap();
|
||||
match transport_i {
|
||||
TransportHandle::Ble(t) => {
|
||||
t.io().set_connect_handler(move |addr, _psm| {
|
||||
let key = addr.to_transport_addr().to_string();
|
||||
let mut map = bank.lock().unwrap();
|
||||
match map.remove(&key) {
|
||||
Some(stream) => Ok(stream),
|
||||
None => Err(crate::transport::TransportError::ConnectionRefused),
|
||||
}
|
||||
});
|
||||
}
|
||||
_ => panic!("expected BLE transport"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Two BLE nodes complete a Noise handshake and establish bidirectional peering.
|
||||
#[tokio::test]
|
||||
async fn test_ble_two_node_handshake() {
|
||||
let mut nodes = vec![make_test_node_ble(1).await, make_test_node_ble(2).await];
|
||||
|
||||
// Wire connection: node 0 → node 1
|
||||
let bank: StreamBank = Arc::new(StdMutex::new(HashMap::new()));
|
||||
wire_ble_connection(&nodes, 0, 1, &bank).await;
|
||||
install_connect_handler(&nodes, 0, &bank);
|
||||
|
||||
// Initiate handshake (connect-on-send creates the BLE connection)
|
||||
initiate_handshake(&mut nodes, 0, 1).await;
|
||||
|
||||
// Drain all packets (handshake + TreeAnnounce exchange)
|
||||
let total = drain_all_packets(&mut nodes, false).await;
|
||||
assert!(total > 0, "should have processed packets");
|
||||
|
||||
// Verify bidirectional peering
|
||||
let addr_0 = *nodes[0].node.node_addr();
|
||||
let addr_1 = *nodes[1].node.node_addr();
|
||||
assert!(
|
||||
nodes[0].node.get_peer(&addr_1).is_some(),
|
||||
"node 0 should have node 1 as peer"
|
||||
);
|
||||
assert!(
|
||||
nodes[1].node.get_peer(&addr_0).is_some(),
|
||||
"node 1 should have node 0 as peer"
|
||||
);
|
||||
|
||||
cleanup_nodes(&mut nodes).await;
|
||||
}
|
||||
|
||||
/// Three BLE nodes in a chain converge to a consistent spanning tree.
|
||||
#[tokio::test]
|
||||
async fn test_ble_three_node_chain() {
|
||||
let mut nodes = vec![
|
||||
make_test_node_ble(1).await,
|
||||
make_test_node_ble(2).await,
|
||||
make_test_node_ble(3).await,
|
||||
];
|
||||
|
||||
let bank: StreamBank = Arc::new(StdMutex::new(HashMap::new()));
|
||||
|
||||
// Wire: 0 -- 1 -- 2
|
||||
wire_ble_connection(&nodes, 0, 1, &bank).await;
|
||||
wire_ble_connection(&nodes, 1, 2, &bank).await;
|
||||
install_connect_handler(&nodes, 0, &bank);
|
||||
install_connect_handler(&nodes, 1, &bank);
|
||||
|
||||
initiate_handshake(&mut nodes, 0, 1).await;
|
||||
initiate_handshake(&mut nodes, 1, 2).await;
|
||||
|
||||
let total = drain_all_packets(&mut nodes, false).await;
|
||||
assert!(total > 0, "should have processed packets");
|
||||
|
||||
// Verify spanning tree convergence
|
||||
verify_tree_convergence(&nodes);
|
||||
|
||||
// Verify correct root
|
||||
let expected_root = nodes.iter().map(|tn| *tn.node.node_addr()).min().unwrap();
|
||||
for tn in &nodes {
|
||||
assert_eq!(*tn.node.tree_state().root(), expected_root);
|
||||
}
|
||||
|
||||
// Verify peer counts
|
||||
assert_eq!(nodes[0].node.peer_count(), 1);
|
||||
assert_eq!(nodes[1].node.peer_count(), 2);
|
||||
assert_eq!(nodes[2].node.peer_count(), 1);
|
||||
|
||||
// Verify bloom filter reachability: node 0 → node 2
|
||||
let addr_2 = *nodes[2].node.node_addr();
|
||||
let reaches = nodes[0].node.peers().any(|p| p.may_reach(&addr_2));
|
||||
assert!(reaches, "node 0 should see node 2 as reachable");
|
||||
|
||||
cleanup_nodes(&mut nodes).await;
|
||||
}
|
||||
|
||||
/// Mixed transport: UDP and BLE nodes coexist in independent components.
|
||||
#[tokio::test]
|
||||
async fn test_ble_mixed_transport() {
|
||||
use spanning_tree::{make_test_node, verify_tree_convergence_components};
|
||||
|
||||
let udp_0 = make_test_node().await;
|
||||
let udp_1 = make_test_node().await;
|
||||
let ble_0 = make_test_node_ble(1).await;
|
||||
let ble_1 = make_test_node_ble(2).await;
|
||||
|
||||
let mut nodes = vec![udp_0, udp_1, ble_0, ble_1];
|
||||
|
||||
// Wire BLE pair
|
||||
let bank: StreamBank = Arc::new(StdMutex::new(HashMap::new()));
|
||||
wire_ble_connection(&nodes, 2, 3, &bank).await;
|
||||
install_connect_handler(&nodes, 2, &bank);
|
||||
|
||||
// Handshake within each component
|
||||
initiate_handshake(&mut nodes, 0, 1).await; // UDP pair
|
||||
initiate_handshake(&mut nodes, 2, 3).await; // BLE pair
|
||||
|
||||
let total = drain_all_packets(&mut nodes, false).await;
|
||||
assert!(total > 0);
|
||||
|
||||
// Verify each component converges independently
|
||||
verify_tree_convergence_components(&nodes, &[vec![0, 1], vec![2, 3]]);
|
||||
|
||||
// BLE component has its own root
|
||||
let ble_root = std::cmp::min(*nodes[2].node.node_addr(), *nodes[3].node.node_addr());
|
||||
assert_eq!(*nodes[2].node.tree_state().root(), ble_root);
|
||||
assert_eq!(*nodes[3].node.tree_state().root(), ble_root);
|
||||
|
||||
cleanup_nodes(&mut nodes).await;
|
||||
}
|
||||
|
||||
/// BLE scan+probe loop discovers peers via adapter scan events.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn test_ble_discovery() {
|
||||
let mut node = make_node();
|
||||
let transport_id = TransportId::new(1);
|
||||
let addr = ble_addr(1);
|
||||
|
||||
// Enable scanning so the scan+probe loop runs
|
||||
let config = BleConfig {
|
||||
adapter: Some("hci0".to_string()),
|
||||
mtu: Some(2048),
|
||||
accept_connections: Some(true),
|
||||
scan: Some(true),
|
||||
advertise: Some(false),
|
||||
auto_connect: Some(false),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let io = MockBleIo::new("hci0", addr.clone());
|
||||
let (packet_tx, packet_rx) = packet_channel(256);
|
||||
let mut transport = BleTransport::new(transport_id, None, config, io, packet_tx);
|
||||
transport.start_async().await.unwrap();
|
||||
|
||||
// Inject scan results via the I/O mock
|
||||
transport.io().inject_scan_result(ble_addr(2)).await;
|
||||
transport.io().inject_scan_result(ble_addr(3)).await;
|
||||
|
||||
// Let scan_probe_loop pick up results and schedule jitter
|
||||
tokio::task::yield_now().await;
|
||||
// Advance past max jitter so probes fire
|
||||
tokio::time::advance(std::time::Duration::from_secs(6)).await;
|
||||
tokio::task::yield_now().await;
|
||||
|
||||
// Without pubkey set, peers appear as bare MACs in discovery buffer
|
||||
let peers = transport.discover().unwrap();
|
||||
assert_eq!(peers.len(), 2);
|
||||
|
||||
let ta = addr.to_transport_addr();
|
||||
node.transports
|
||||
.insert(transport_id, TransportHandle::Ble(transport));
|
||||
|
||||
let mut nodes = vec![TestNode { node, transport_id, packet_rx, addr: ta }];
|
||||
cleanup_nodes(&mut nodes).await;
|
||||
}
|
||||
@@ -5,6 +5,8 @@ use crate::PeerIdentity;
|
||||
use std::time::Duration;
|
||||
|
||||
mod bloom;
|
||||
#[cfg(target_os = "linux")]
|
||||
mod ble;
|
||||
mod disconnect;
|
||||
mod discovery;
|
||||
#[cfg(target_os = "linux")]
|
||||
|
||||
Reference in New Issue
Block a user