From 331a213df30f05471d547e6c92767fb21ad2e3a2 Mon Sep 17 00:00:00 2001 From: Vladimir Bondarevskiy Date: Sun, 27 Sep 2026 03:32:36 +0300 Subject: [PATCH 1/2] Fix Matter BLE discovery from service data advertisements (#1) --- .github/workflows/rust.yml | 14 + src/ble.rs | 152 +++++++--- src/ble/tests.rs | 599 +++++++++++++++++++++++++++++++++++++ src/ble/tests/backend.rs | 249 +++++++++++++++ src/ble/tests/fixtures.rs | 97 ++++++ 5 files changed, 1062 insertions(+), 49 deletions(-) create mode 100644 src/ble/tests.rs create mode 100644 src/ble/tests/backend.rs create mode 100644 src/ble/tests/fixtures.rs diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 9fd45e0..23c7dd3 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -20,3 +20,17 @@ jobs: run: cargo build --verbose - name: Run tests run: cargo test --verbose + + ble: + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v4 + - name: Install BLE build dependencies + run: | + sudo apt-get update + sudo apt-get install --yes libdbus-1-dev pkg-config + - name: Build BLE library and all targets + run: cargo build --locked --all-targets --features ble --verbose + - name: Run BLE-enabled tests without hardware + run: cargo test --locked --features ble --verbose diff --git a/src/ble.rs b/src/ble.rs index aab98a1..25d06b2 100644 --- a/src/ble.rs +++ b/src/ble.rs @@ -8,12 +8,14 @@ use std::time::Duration; use anyhow::{bail, Context, Result}; +use btleplug::api::{ScanFilter, WriteType}; +#[cfg(not(test))] use btleplug::{ - api::{ - Central, CentralEvent, Manager as _, Peripheral as _, ScanFilter, WriteType, - }, + api::{Central, CentralEvent, Manager as _, Peripheral as _}, platform::{Manager, Peripheral}, }; +#[cfg(test)] +use self::tests::backend::{CentralEvent, Manager, Peripheral}; use futures::StreamExt; use tokio::sync::mpsc; use uuid::Uuid; @@ -51,56 +53,95 @@ pub async fn find_by_discriminator(discriminator: u16, short_match: bool, scan_t let adapters = manager.adapters().await.context("listing BLE adapters")?; let central = adapters.into_iter().next().context("no BLE adapter found")?; - let filter = ScanFilter { - services: vec![MATTER_SERVICE_UUID], - }; - central.start_scan(filter).await.context("start BLE scan")?; - let mut events = central.events().await.context("BLE event stream")?; + // Matter UUID can be present only in service data, which BlueZ's UUID filter can omit. + central.start_scan(ScanFilter::default()).await.context("start BLE scan")?; let deadline = tokio::time::Instant::now() + scan_timeout; - loop { - let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); - if remaining.is_zero() { - bail!("BLE scan timeout: no device with discriminator {} found", discriminator); - } - let event = tokio::time::timeout(remaining, events.next()) - .await - .context("BLE scan timeout")? - .context("BLE event stream ended")?; + let found: Result = async { + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + bail!("BLE scan timeout: no device with discriminator {} found", discriminator); + } + let event = tokio::time::timeout(remaining, events.next()) + .await + .context("BLE scan timeout")? + .context("BLE event stream ended")?; - match event { - CentralEvent::DeviceDiscovered(id) | CentralEvent::DeviceUpdated(id) => { - let peripheral = central.peripheral(&id).await?; - let props = match peripheral.properties().await? { - Some(p) => p, - None => continue, - }; - // Look for service data for our UUID - let svc_data = props - .service_data - .get(&MATTER_SERVICE_UUID) - .cloned() - .unwrap_or_default(); - if svc_data.len() < 8 { - continue; + let (id, peripheral, svc_data) = match event { + CentralEvent::DeviceDiscovered(id) | CentralEvent::DeviceUpdated(id) => { + let peripheral = match central.peripheral(&id).await { + Ok(peripheral) => peripheral, + Err(error) => { + log::debug!( + "Skipping BLE candidate {}: peripheral lookup failed: {:?}", + id, error, + ); + continue; + } + }; + let props = match peripheral.properties().await { + Ok(Some(props)) => props, + Ok(None) => continue, + Err(error) => { + log::debug!( + "Skipping BLE candidate {}: properties unavailable: {:?}", + id, error, + ); + continue; + } + }; + ( + id, + Some(peripheral), + props.service_data.get(&MATTER_SERVICE_UUID).cloned(), + ) } - let (disc, vid, pid, cm_flag) = parse_service_data(&svc_data); - log::debug!("BLE found device: disc={} vid={} pid={} cm={}", disc, vid, pid, cm_flag); - let matches = if short_match { - disc >> 8 == discriminator >> 8 - } else { - disc == discriminator - }; - if matches { - central.stop_scan().await.ok(); - log::debug!("BLE device with matching discriminator found, connecting..."); - return connect_peripheral(peripheral).await; + CentralEvent::ServiceDataAdvertisement { id, service_data } => { + (id, None, service_data.get(&MATTER_SERVICE_UUID).cloned()) + } + _ => continue, + }; + let Some(svc_data) = svc_data else { + continue; + }; + if matches_discriminator(&svc_data, discriminator, short_match) { + if let Some(peripheral) = peripheral { + return Ok(peripheral); + } + match central.peripheral(&id).await { + Ok(peripheral) => return Ok(peripheral), + Err(error) => { + log::debug!( + "Skipping BLE candidate {}: peripheral lookup failed: {:?}", + id, error, + ); + } } } - _ => {} } } + .await; + if let Err(error) = central.stop_scan().await { + log::warn!("BLE stop_scan failed: {:?}", error); + } + let peripheral = found?; + log::debug!("BLE device with matching discriminator found, connecting..."); + connect_peripheral(peripheral).await +} + +fn matches_discriminator(svc_data: &[u8], discriminator: u16, short_match: bool) -> bool { + if svc_data.len() < 8 { + return false; + } + let (disc, vid, pid, cm_flag) = parse_service_data(svc_data); + log::debug!("BLE found device: disc={} vid={} pid={} cm={}", disc, vid, pid, cm_flag); + if short_match { + disc >> 8 == discriminator >> 8 + } else { + disc == discriminator + } } pub async fn scan_commissionable(scan_timeout: Duration) -> Result> { @@ -109,18 +150,28 @@ pub async fn scan_commissionable(scan_timeout: Duration) -> Result p, - None => continue, + let props = match peripheral.properties().await { + Ok(Some(props)) => props, + Ok(None) => continue, + Err(error) => { + log::debug!( + "Skipping BLE candidate {}: properties unavailable: {:?}", + peripheral.id(), error, + ); + continue; + } }; if let Some(svc_data) = props.service_data.get(&MATTER_SERVICE_UUID) { if svc_data.len() >= 8 { @@ -257,3 +308,6 @@ fn hex_dump(data: &[u8]) -> String { } s.trim_end().to_string() } + +#[cfg(test)] +mod tests; diff --git a/src/ble/tests.rs b/src/ble/tests.rs new file mode 100644 index 0000000..638c8b6 --- /dev/null +++ b/src/ble/tests.rs @@ -0,0 +1,599 @@ +use std::{collections::HashMap, time::Duration}; + +use btleplug::api::{PeripheralProperties, ScanFilter}; + +use super::{find_by_discriminator, scan_commissionable, MATTER_SERVICE_UUID}; +use backend::{Call, CentralEvent, Device, State}; +use fixtures::{raw_advertisement, raw_device, TestDeviceDefinition}; + +pub(super) mod backend; +mod fixtures; + +// Independent vectors retained from the original PR, not verified device captures. +// OpCode, discriminator + advertisement version (LE), VID (LE), PID (LE), flags. +const MATTER_SERVICE_DATA_ABC: [u8; 8] = [0x00, 0xbc, 0x0a, 0x2f, 0x13, 0x0d, 0x02, 0x00]; +const MATTER_SERVICE_DATA_560: [u8; 8] = [0x00, 0x60, 0x05, 0x2f, 0x13, 0x0d, 0x02, 0x00]; +const TIMEOUT: Duration = Duration::from_secs(30); + +fn finish(result: anyhow::Result) { + // No BTP handshake is run; stop the notification task created by connect_peripheral. + let peripheral = result.unwrap(); + peripheral.c2_abort.abort(); + (peripheral.disconnect)(); +} + +#[tokio::test(start_paused = true)] +async fn find_skips_disappeared_peripheral_and_unreadable_properties() { + let unreadable = TestDeviceDefinition { + id: 2, + discriminator: 0x0123, + ..Default::default() + }; + let target = TestDeviceDefinition { + id: 3, + discriminator: 0x0abc, + ..Default::default() + }; + let mut state = State { + events: vec![ + CentralEvent::DeviceDiscovered(1), + CentralEvent::DeviceUpdated(unreadable.id), + CentralEvent::DeviceDiscovered(target.id), + ], + ..Default::default() + }; + state.devices.insert( + unreadable.id, + Device { + properties_failures: 1, + ..unreadable.device() + }, + ); + state.devices.insert(target.id, target.device()); + + let (result, state) = backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + finish(result); + assert_eq!( + state.lock().unwrap().calls, + vec![ + Call::Events, + Call::Start(ScanFilter::default()), + Call::Lookup(1), + Call::Lookup(2), + Call::Properties(2), + Call::Lookup(3), + Call::Properties(3), + Call::Stop, + Call::Connect(3), + ] + ); +} + +#[tokio::test(start_paused = true)] +async fn find_retries_later_advertisement_after_failed_lookup() { + let target = TestDeviceDefinition { + id: 1, + discriminator: 0x0abc, + ..Default::default() + }; + let mut state = State { + events: vec![target.advertisement(), target.advertisement()], + ..Default::default() + }; + state.devices.insert( + target.id, + Device { + lookup_failures: 1, + ..Default::default() + }, + ); + + let (result, state) = backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + finish(result); + assert_eq!( + state.lock().unwrap().calls, + vec![ + Call::Events, + Call::Start(ScanFilter::default()), + Call::Lookup(1), + Call::Lookup(1), + Call::Stop, + Call::Connect(1), + ] + ); +} + +#[tokio::test(start_paused = true)] +async fn find_filters_fixed_service_data_vectors_before_looking_up_peripheral() { + for (data, discriminator, short_match) in [ + (MATTER_SERVICE_DATA_ABC, 0x0abc, false), + (MATTER_SERVICE_DATA_ABC, 0x0a00, true), + (MATTER_SERVICE_DATA_560, 0x0500, true), + ] { + let mut wrong = data; + wrong[1] ^= 1; + if short_match { + wrong[2] = 0x0b; + } + let mut state = State { + events: vec![ + CentralEvent::Other, + CentralEvent::ServiceDataAdvertisement { + id: 1, + service_data: HashMap::new(), + }, + raw_advertisement(1, &data[..7]), + raw_advertisement(1, &wrong), + raw_advertisement(2, &data), + ], + ..Default::default() + }; + // Service-data-only device: there are no cached properties or discovery events. + state.devices.insert(2, Device::default()); + let (result, state) = backend::run( + state, + find_by_discriminator(discriminator, short_match, TIMEOUT), + ) + .await; + finish(result); + assert_eq!( + state.lock().unwrap().calls, + vec![ + Call::Events, + Call::Start(ScanFilter::default()), + Call::Lookup(2), + Call::Stop, + Call::Connect(2), + ] + ); + } +} + +#[tokio::test(start_paused = true)] +async fn find_stops_after_timeout_even_when_candidates_fail() { + let state = State { + events: vec![CentralEvent::DeviceDiscovered(1)], + ..Default::default() + }; + let before = tokio::time::Instant::now(); + let (result, state) = backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + assert!(result + .err() + .unwrap() + .to_string() + .contains("BLE scan timeout")); + assert_eq!(before.elapsed(), TIMEOUT); + assert_eq!(state.lock().unwrap().calls.last(), Some(&Call::Stop)); +} + +#[tokio::test(start_paused = true)] +async fn find_preserves_scan_and_stream_errors() { + for (state, message, stops) in [ + ( + State { + events_error: true, + ..Default::default() + }, + "BLE event stream", + false, + ), + ( + State { + start_error: true, + ..Default::default() + }, + "start BLE scan", + false, + ), + ( + State { + end_events: true, + ..Default::default() + }, + "BLE event stream ended", + true, + ), + ] { + let (result, state) = + backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + assert_eq!(result.err().unwrap().to_string(), message); + assert_eq!(state.lock().unwrap().calls.contains(&Call::Stop), stops); + } +} + +#[tokio::test(start_paused = true)] +async fn find_preserves_connection_errors() { + let target = TestDeviceDefinition { + id: 1, + discriminator: 0x0abc, + ..Default::default() + }; + let mut state = State { + events: vec![target.advertisement()], + ..Default::default() + }; + state.devices.insert( + target.id, + Device { + connect_error: true, + ..Default::default() + }, + ); + let (result, state) = backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + assert_eq!(result.err().unwrap().to_string(), "BLE connect"); + assert!(state + .lock() + .unwrap() + .calls + .ends_with(&[Call::Stop, Call::Connect(1)])); +} + +#[tokio::test(start_paused = true)] +async fn scan_includes_service_data_only_devices_and_skips_bad_candidates() { + let mut state = State::default(); + state.devices.insert( + 1, + Device { + properties_failures: 1, + ..Default::default() + }, + ); + let target = TestDeviceDefinition { + id: 2, + discriminator: 0x0abc, + vendor_id: 0x1234, + product_id: 0x0042, + name: Some("Matter test device".to_owned()), + rssi: Some(-60), + tx_power: Some(-4), + ..Default::default() + }; + assert!(target.advertised_services.is_empty()); + state.devices.insert(target.id, target.device()); + state.devices.insert( + 3, + Device { + properties: Some(PeripheralProperties::default()), + ..Default::default() + }, + ); + state + .devices + .insert(4, raw_device(&MATTER_SERVICE_DATA_ABC[..7])); + state.devices.insert(5, Device::default()); + + let before = tokio::time::Instant::now(); + let (result, state) = backend::run(state, scan_commissionable(TIMEOUT)).await; + let found = result.unwrap(); + assert_eq!(before.elapsed(), TIMEOUT); + assert_eq!(found.len(), 1); + let found = &found[0]; + assert_eq!(found.discriminator, 0x0abc); + assert_eq!(found.vendor_id, 0x1234); + assert_eq!(found.product_id, 0x0042); + assert_eq!(found.name.as_deref(), Some("Matter test device")); + assert_eq!(found.rssi, Some(-60)); + assert_eq!(found.tx_power, Some(-4)); + assert_eq!(found.address, "2"); + assert_eq!(found.peripheral.id(), 2); + assert_eq!( + state.lock().unwrap().calls, + vec![ + Call::Start(ScanFilter::default()), + Call::Stop, + Call::List, + Call::Properties(1), + Call::Properties(2), + Call::Properties(3), + Call::Properties(4), + Call::Properties(5), + ] + ); +} + +#[tokio::test(start_paused = true)] +async fn scan_with_no_devices_returns_an_empty_list() { + let (result, state) = backend::run(State::default(), scan_commissionable(TIMEOUT)).await; + assert!(result.unwrap().is_empty()); + assert_eq!( + state.lock().unwrap().calls, + vec![Call::Start(ScanFilter::default()), Call::Stop, Call::List,] + ); +} + +#[tokio::test(start_paused = true)] +async fn scan_preserves_start_and_enumeration_errors() { + let (result, state) = backend::run( + State { + start_error: true, + ..Default::default() + }, + scan_commissionable(TIMEOUT), + ) + .await; + assert_eq!(result.err().unwrap().to_string(), "start BLE scan"); + assert_eq!( + state.lock().unwrap().calls, + vec![Call::Start(ScanFilter::default())] + ); + + let (result, state) = backend::run( + State { + list_error: true, + ..Default::default() + }, + scan_commissionable(TIMEOUT), + ) + .await; + assert_eq!(result.err().unwrap().to_string(), "Permission denied"); + assert_eq!( + state.lock().unwrap().calls, + vec![Call::Start(ScanFilter::default()), Call::Stop, Call::List,] + ); +} + +#[tokio::test(start_paused = true)] +async fn find_warns_when_stop_fails_without_blocking_connection() { + let target = TestDeviceDefinition { + id: 1, + discriminator: 0x0abc, + ..Default::default() + }; + let mut state = State { + events: vec![target.advertisement()], + stop_error: true, + ..Default::default() + }; + state.devices.insert(target.id, Device::default()); + let (result, state) = backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + finish(result); + let state = state.lock().unwrap(); + assert!(state.calls.ends_with(&[Call::Stop, Call::Connect(1)])); + assert_eq!( + state.warnings, + vec!["BLE stop_scan failed: PermissionDenied"] + ); +} + +#[tokio::test(start_paused = true)] +async fn find_warns_when_stop_fails_without_masking_search_or_connection_errors() { + for end_events in [false, true] { + let state = State { + stop_error: true, + end_events, + ..Default::default() + }; + let (result, state) = + backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + let expected = if end_events { + "BLE event stream ended" + } else { + "BLE scan timeout" + }; + assert_eq!(result.err().unwrap().to_string(), expected); + assert_eq!( + state.lock().unwrap().warnings, + vec!["BLE stop_scan failed: PermissionDenied"] + ); + } + + let target = TestDeviceDefinition { + id: 1, + discriminator: 0x0abc, + ..Default::default() + }; + let mut state = State { + events: vec![target.advertisement()], + stop_error: true, + ..Default::default() + }; + state.devices.insert( + target.id, + Device { + connect_error: true, + ..Default::default() + }, + ); + let (result, state) = backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + assert_eq!(result.err().unwrap().to_string(), "BLE connect"); + assert_eq!( + state.lock().unwrap().warnings, + vec!["BLE stop_scan failed: PermissionDenied"] + ); +} + +#[tokio::test(start_paused = true)] +async fn scan_warns_when_stop_fails_without_masking_results_or_enumeration_errors() { + let target = TestDeviceDefinition { + id: 1, + discriminator: 0x0abc, + ..Default::default() + }; + for list_error in [false, true] { + let mut state = State { + stop_error: true, + list_error, + ..Default::default() + }; + state.devices.insert(target.id, target.device()); + let (result, state) = backend::run(state, scan_commissionable(TIMEOUT)).await; + if list_error { + assert_eq!(result.err().unwrap().to_string(), "Permission denied"); + } else { + let found = result.unwrap(); + assert_eq!(found.len(), 1); + assert_eq!(found[0].discriminator, 0x0abc); + } + assert_eq!( + state.lock().unwrap().warnings, + vec!["BLE stop_scan failed: PermissionDenied"] + ); + } +} + +#[tokio::test(start_paused = true)] +async fn find_selects_the_matching_definition_among_different_devices() { + let neighbor = TestDeviceDefinition { + id: 1, + discriminator: 0x0247, + vendor_id: 0x1234, + product_id: 0x0001, + advertised_services: vec![MATTER_SERVICE_UUID], + ..Default::default() + }; + let target = TestDeviceDefinition { + id: 2, + discriminator: 0x0246, + vendor_id: 0x4321, + product_id: 0x0002, + ..Default::default() + }; + let other = TestDeviceDefinition { + id: 3, + discriminator: 0x024a, + ..target.clone() + }; + let mut state = State { + events: vec![ + CentralEvent::DeviceDiscovered(neighbor.id), + other.advertisement(), + CentralEvent::DeviceUpdated(target.id), + ], + ..Default::default() + }; + for definition in [&neighbor, &target, &other] { + state.devices.insert(definition.id, definition.device()); + } + let (result, state) = backend::run(state, find_by_discriminator(0x0246, false, TIMEOUT)).await; + finish(result); + assert_eq!( + state.lock().unwrap().calls, + vec![ + Call::Events, + Call::Start(ScanFilter::default()), + Call::Lookup(1), + Call::Properties(1), + Call::Lookup(2), + Call::Properties(2), + Call::Stop, + Call::Connect(2), + ] + ); +} + +#[tokio::test(start_paused = true)] +async fn find_uses_advertisement_data_independently_of_cached_properties() { + let cached = TestDeviceDefinition { + id: 7, + discriminator: 0x0111, + ..Default::default() + }; + let advertised = TestDeviceDefinition { + discriminator: 0x0abc, + ..cached.clone() + }; + let mut state = State { + events: vec![advertised.advertisement()], + ..Default::default() + }; + state.devices.insert(cached.id, cached.device()); + + let (result, state) = backend::run(state, find_by_discriminator(0x0abc, false, TIMEOUT)).await; + finish(result); + assert_eq!( + state.lock().unwrap().calls, + vec![ + Call::Events, + Call::Start(ScanFilter::default()), + Call::Lookup(7), + Call::Stop, + Call::Connect(7), + ] + ); +} + +#[tokio::test(start_paused = true)] +async fn scan_returns_metadata_for_multiple_device_definitions() { + let definitions = [ + TestDeviceDefinition { + id: 9, + discriminator: 0x0abc, + advertisement_version: 2, + vendor_id: 0x1234, + product_id: 0x5678, + additional_flags: 0x03, + name: Some("Service-data-only fixture".to_owned()), + rssi: Some(-60), + tx_power: Some(-4), + ..Default::default() + }, + TestDeviceDefinition { + id: 4, + discriminator: 0x0560, + vendor_id: 0x4321, + product_id: 0x00ff, + advertised_services: vec![MATTER_SERVICE_UUID], + ..Default::default() + }, + ]; + // Anchor fixture encoding independently: bit 12..15 is the advertisement version, + // not a commissioning flag. Flags occupy the final byte. + assert_eq!( + definitions[0].service_data(), + [0x00, 0xbc, 0x2a, 0x34, 0x12, 0x78, 0x56, 0x03] + ); + assert_eq!( + definitions[1].service_data(), + [0x00, 0x60, 0x05, 0x21, 0x43, 0xff, 0x00, 0x00] + ); + let mut state = State::default(); + for definition in &definitions { + state.devices.insert(definition.id, definition.device()); + } + let (result, state) = backend::run(state, scan_commissionable(TIMEOUT)).await; + let found = result.unwrap(); + assert_eq!(found.len(), definitions.len()); + for definition in &definitions { + let actual = found + .iter() + .find(|device| device.peripheral.id() == definition.id) + .unwrap(); + assert_eq!(actual.discriminator, definition.discriminator); + assert_eq!(actual.vendor_id, definition.vendor_id); + assert_eq!(actual.product_id, definition.product_id); + assert_eq!(actual.name, definition.name); + assert_eq!(actual.rssi, definition.rssi); + assert_eq!(actual.tx_power, definition.tx_power); + assert_eq!(actual.address, definition.id.to_string()); + } + assert_eq!( + state.lock().unwrap().calls, + vec![ + Call::Start(ScanFilter::default()), + Call::Stop, + Call::List, + Call::Properties(4), + Call::Properties(9), + ] + ); +} + +#[tokio::test(start_paused = true)] +async fn scan_reads_metadata_from_fixed_wire_vectors() { + let cases = [ + (MATTER_SERVICE_DATA_ABC, 0x0abc), + (MATTER_SERVICE_DATA_560, 0x0560), + ]; + let mut state = State::default(); + for (id, (data, _)) in cases.iter().enumerate() { + state.devices.insert(id as u64, raw_device(data)); + } + let (result, _) = backend::run(state, scan_commissionable(TIMEOUT)).await; + let found = result.unwrap(); + assert_eq!(found.len(), cases.len()); + for (actual, (_, discriminator)) in found.iter().zip(cases) { + assert_eq!(actual.discriminator, discriminator); + assert_eq!(actual.vendor_id, 0x132f); + assert_eq!(actual.product_id, 0x020d); + } +} diff --git a/src/ble/tests/backend.rs b/src/ble/tests/backend.rs new file mode 100644 index 0000000..3161da7 --- /dev/null +++ b/src/ble/tests/backend.rs @@ -0,0 +1,249 @@ +//! Test-only platform facade for exercising the public BLE functions without hardware. +//! Advertising properties, scan filters and GATT descriptors use the real btleplug types. + +use std::{ + collections::{BTreeMap, BTreeSet, HashMap}, + future::Future, + pin::Pin, + sync::{Arc, Mutex, Once}, +}; + +use btleplug::{ + api::{ + CharPropFlags, Characteristic, PeripheralProperties, ScanFilter, ValueNotification, + WriteType, + }, + Error, Result, +}; +use futures::{stream, Stream, StreamExt}; +use uuid::Uuid; + +use super::super::{C1_UUID, C2_UUID, MATTER_SERVICE_UUID}; + +#[derive(Debug)] +pub enum CentralEvent { + DeviceDiscovered(u64), + DeviceUpdated(u64), + ServiceDataAdvertisement { + id: u64, + service_data: HashMap>, + }, + Other, +} + +#[derive(Debug, PartialEq)] +pub enum Call { + Events, + Start(ScanFilter), + Stop, + List, + Lookup(u64), + Properties(u64), + Connect(u64), +} + +#[derive(Debug, Default)] +pub struct Device { + pub properties: Option, + pub lookup_failures: usize, + pub properties_failures: usize, + pub connect_error: bool, +} + +#[derive(Debug, Default)] +pub struct State { + pub events: Vec, + pub devices: BTreeMap, + pub calls: Vec, + pub end_events: bool, + pub start_error: bool, + pub events_error: bool, + pub list_error: bool, + pub stop_error: bool, + pub warnings: Vec, +} + +tokio::task_local! { + static CURRENT: Arc>; +} + +struct TestLogger; + +impl log::Log for TestLogger { + fn enabled(&self, metadata: &log::Metadata<'_>) -> bool { + metadata.level() <= log::Level::Warn + } + + fn log(&self, record: &log::Record<'_>) { + if self.enabled(record.metadata()) { + // Each test captures only logs from its own task-local scenario. + let _ = CURRENT.try_with(|state| { + state + .lock() + .unwrap() + .warnings + .push(record.args().to_string()); + }); + } + } + + fn flush(&self) {} +} + +pub async fn run(state: State, future: F) -> (F::Output, Arc>) { + static INIT_LOGGER: Once = Once::new(); + INIT_LOGGER.call_once(|| { + log::set_logger(&TestLogger).unwrap(); + log::set_max_level(log::LevelFilter::Warn); + }); + let state = Arc::new(Mutex::new(state)); + let output = CURRENT.scope(state.clone(), future).await; + (output, state) +} + +pub struct Manager(Arc>); + +impl Manager { + pub async fn new() -> Result { + Ok(Self(CURRENT.with(Arc::clone))) + } + + pub async fn adapters(&self) -> Result> { + Ok(vec![Adapter(self.0.clone())]) + } +} + +pub struct Adapter(Arc>); + +impl Adapter { + pub async fn events(&self) -> Result + Send>>> { + let mut state = self.0.lock().unwrap(); + state.calls.push(Call::Events); + if state.events_error { + return Err(Error::PermissionDenied); + } + let events = stream::iter(std::mem::take(&mut state.events)); + if state.end_events { + Ok(Box::pin(events)) + } else { + Ok(Box::pin(events.chain(stream::pending()))) + } + } + + pub async fn start_scan(&self, filter: ScanFilter) -> Result<()> { + let mut state = self.0.lock().unwrap(); + state.calls.push(Call::Start(filter)); + if state.start_error { + return Err(Error::PermissionDenied); + } + Ok(()) + } + + pub async fn stop_scan(&self) -> Result<()> { + let mut state = self.0.lock().unwrap(); + state.calls.push(Call::Stop); + if state.stop_error { + return Err(Error::PermissionDenied); + } + Ok(()) + } + + pub async fn peripheral(&self, id: &u64) -> Result { + let mut state = self.0.lock().unwrap(); + state.calls.push(Call::Lookup(*id)); + let device = state.devices.get_mut(id).ok_or(Error::DeviceNotFound)?; + if device.lookup_failures > 0 { + device.lookup_failures -= 1; + return Err(Error::DeviceNotFound); + } + Ok(Peripheral { + id: *id, + state: self.0.clone(), + }) + } + + pub async fn peripherals(&self) -> Result> { + let mut state = self.0.lock().unwrap(); + state.calls.push(Call::List); + if state.list_error { + return Err(Error::PermissionDenied); + } + Ok(state + .devices + .keys() + .map(|id| Peripheral { + id: *id, + state: self.0.clone(), + }) + .collect()) + } +} + +#[derive(Clone, Debug)] +pub struct Peripheral { + id: u64, + state: Arc>, +} + +impl Peripheral { + pub fn id(&self) -> u64 { + self.id + } + + pub async fn properties(&self) -> Result> { + let mut state = self.state.lock().unwrap(); + state.calls.push(Call::Properties(self.id)); + let device = state + .devices + .get_mut(&self.id) + .ok_or(Error::DeviceNotFound)?; + if device.properties_failures > 0 { + device.properties_failures -= 1; + return Err(Error::DeviceNotFound); + } + Ok(device.properties.clone()) + } + + pub async fn connect(&self) -> Result<()> { + let mut state = self.state.lock().unwrap(); + state.calls.push(Call::Connect(self.id)); + if state.devices[&self.id].connect_error { + return Err(Error::NotConnected); + } + Ok(()) + } + + pub async fn discover_services(&self) -> Result<()> { + Ok(()) + } + + pub fn characteristics(&self) -> BTreeSet { + [C1_UUID, C2_UUID] + .into_iter() + .map(|uuid| Characteristic { + uuid, + service_uuid: MATTER_SERVICE_UUID, + properties: CharPropFlags::WRITE | CharPropFlags::INDICATE, + descriptors: BTreeSet::new(), + }) + .collect() + } + + pub async fn notifications( + &self, + ) -> Result + Send>>> { + Ok(Box::pin(stream::pending())) + } + + pub async fn write(&self, _: &Characteristic, _: &[u8], _: WriteType) -> Result<()> { + panic!("discovery must not send GATT data") + } + + pub async fn subscribe(&self, _: &Characteristic) -> Result<()> { + panic!("discovery must not subscribe before the BTP handshake") + } + + pub async fn disconnect(&self) -> Result<()> { + Ok(()) + } +} diff --git a/src/ble/tests/fixtures.rs b/src/ble/tests/fixtures.rs new file mode 100644 index 0000000..9c9ed26 --- /dev/null +++ b/src/ble/tests/fixtures.rs @@ -0,0 +1,97 @@ +//! Named test-device data, independent of backend failures and event ordering. + +use std::collections::HashMap; + +use btleplug::api::PeripheralProperties; +use uuid::Uuid; + +use super::{ + backend::{CentralEvent, Device}, + MATTER_SERVICE_UUID, +}; + +#[derive(Clone, Debug, Default)] +pub struct TestDeviceDefinition { + pub id: u64, + pub discriminator: u16, + pub advertisement_version: u8, + pub vendor_id: u16, + pub product_id: u16, + pub additional_flags: u8, + pub name: Option, + pub rssi: Option, + pub tx_power: Option, + pub advertised_services: Vec, +} + +impl TestDeviceDefinition { + /// Encode the Matter service-data payload, excluding the 0xFFF6 service UUID. + /// Layout: + pub fn service_data(&self) -> [u8; 8] { + assert!( + self.discriminator <= 0x0fff, + "discriminator must fit in 12 bits" + ); + assert!( + self.advertisement_version <= 0x0f, + "advertisement version must fit in 4 bits" + ); + let discriminator_and_version = + self.discriminator | (u16::from(self.advertisement_version) << 12); + let [disc_low, disc_high] = discriminator_and_version.to_le_bytes(); + let [vendor_low, vendor_high] = self.vendor_id.to_le_bytes(); + let [product_low, product_high] = self.product_id.to_le_bytes(); + [ + 0x00, + disc_low, + disc_high, + vendor_low, + vendor_high, + product_low, + product_high, + self.additional_flags, + ] + } + + pub fn properties(&self) -> PeripheralProperties { + PeripheralProperties { + local_name: self.name.clone(), + rssi: self.rssi, + tx_power_level: self.tx_power, + services: self.advertised_services.clone(), + service_data: HashMap::from([(MATTER_SERVICE_UUID, self.service_data().to_vec())]), + ..Default::default() + } + } + + /// Build cached properties explicitly; creating an event does not populate this cache. + pub fn device(&self) -> Device { + Device { + properties: Some(self.properties()), + ..Default::default() + } + } + + /// Build an event independently of the peripheral's cached properties. + pub fn advertisement(&self) -> CentralEvent { + raw_advertisement(self.id, &self.service_data()) + } +} + +/// Bypass fixture encoding for independent wire vectors and malformed payloads. +pub fn raw_advertisement(id: u64, data: &[u8]) -> CentralEvent { + CentralEvent::ServiceDataAdvertisement { + id, + service_data: HashMap::from([(MATTER_SERVICE_UUID, data.to_vec())]), + } +} + +pub fn raw_device(data: &[u8]) -> Device { + Device { + properties: Some(PeripheralProperties { + service_data: HashMap::from([(MATTER_SERVICE_UUID, data.to_vec())]), + ..Default::default() + }), + ..Default::default() + } +} From 9d26263de58e0aa85f56c0dcd48fcb07b9780701 Mon Sep 17 00:00:00 2001 From: Vladimir Bondarevskiy Date: Fri, 30 Oct 2026 18:40:09 +0300 Subject: [PATCH 2/2] Fix CASE resumption BUSY and caller-owned connection deadlines (#4) * Fix CASE resumption BUSY and caller-owned connect deadlines * Test legacy CASE API compatibility with resumption BUSY * Preserve typed error diagnostics at deadline boundary * Validate TLV value lengths before allocation (#5) --- src/connection_budget.rs | 122 ++++++++ src/controller.rs | 93 +++++- src/controller/connect_deadline_tests.rs | 348 +++++++++++++++++++++++ src/devman/mod.rs | 123 ++++++-- src/lib.rs | 1 + src/tlv.rs | 37 ++- src/tlv/length_tests.rs | 128 +++++++++ 7 files changed, 809 insertions(+), 43 deletions(-) create mode 100644 src/connection_budget.rs create mode 100644 src/controller/connect_deadline_tests.rs create mode 100644 src/tlv/length_tests.rs diff --git a/src/connection_budget.rs b/src/connection_budget.rs new file mode 100644 index 0000000..f48b849 --- /dev/null +++ b/src/connection_budget.rs @@ -0,0 +1,122 @@ +//! Deadline-aware CASE work. Diagnostics contain only node, closed stage/outcome and timing. + +use std::{fmt, future::Future, time::Duration}; + +use anyhow::Result; +use tokio::time::{timeout_at, Instant}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ConnectStage { + Transport, + Resume, + Case, + BusyWait, + Rediscovery, +} + +impl ConnectStage { + pub fn as_str(self) -> &'static str { + match self { + Self::Transport => "transport", + Self::Resume => "resume", + Self::Case => "case", + Self::BusyWait => "busy_wait", + Self::Rediscovery => "rediscovery", + } + } +} + +#[derive(Clone, Copy, Debug)] +pub struct ConnectDeadline { + pub stage: ConnectStage, +} + +impl fmt::Display for ConnectDeadline { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Matter connect deadline at {}", self.stage.as_str()) + } +} + +impl std::error::Error for ConnectDeadline {} + +/// The responder's minimum wait did not fit in this attempt's deadline. +/// A caller must not start another attempt before `retry_at`. +#[derive(Clone, Copy, Debug)] +pub struct ConnectBusy { + pub retry_at: Instant, +} + +impl ConnectBusy { + pub fn retry_after(&self) -> Duration { + self.retry_at.saturating_duration_since(Instant::now()) + } +} + +impl fmt::Display for ConnectBusy { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("Matter responder BUSY; attempt deferred") + } +} + +impl std::error::Error for ConnectBusy {} + +pub(crate) struct ConnectionBudget { + node: u64, + deadline: Option, +} + +impl ConnectionBudget { + pub(crate) fn new(node: u64, deadline: Option) -> Self { + Self { node, deadline } + } + + pub(crate) async fn run( + &self, + stage: ConnectStage, + work: impl Future>, + ) -> Result { + let started = Instant::now(); + let result = match self.deadline { + Some(deadline) if started >= deadline => Err(ConnectDeadline { stage }.into()), + Some(deadline) => match timeout_at(deadline, work).await { + // Tokio may poll a ready future before the expired timer. Do not + // return late connection authority even when that future is ready. + Ok(Ok(_)) if Instant::now() >= deadline => Err(ConnectDeadline { stage }.into()), + Ok(result) => result, + Err(_) => Err(ConnectDeadline { stage }.into()), + }, + None => work.await, + }; + let outcome = match &result { + Ok(_) => "ok", + Err(error) if error.is::() => "deadline", + Err(error) if error.is::() => "busy", + Err(_) => "failed", + }; + log::info!( + "Matter connect stage node={} stage={} outcome={} elapsed_ms={}", + self.node, + stage.as_str(), + outcome, + started.elapsed().as_millis() + ); + result + } + + pub(crate) async fn busy_wait(&self, wait: Duration) -> Result<()> { + let retry_at = Instant::now() + wait; + if self.deadline.is_some_and(|deadline| retry_at >= deadline) { + log::info!( + "Matter connect stage node={} stage=busy_wait outcome=deferred wait_ms={}", + self.node, + wait.as_millis() + ); + return Err(ConnectBusy { retry_at }.into()); + } + self.run(ConnectStage::BusyWait, async { + tokio::time::sleep_until(retry_at).await; + Ok(()) + }) + .await + } +} diff --git a/src/controller.rs b/src/controller.rs index b6906ad..ad015bf 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -124,9 +124,33 @@ impl Controller { node_id: u64, controller_id: u64, ) -> Result { - if let Some(ses) = self.try_auth_sigma_resume(connection, node_id, controller_id).await? { - return Ok(ses); - } + self.auth_sigma_with_budget(connection, node_id, controller_id, None) + .await + } + + /// CASE with one caller-owned deadline, including resumption and BUSY waits. + /// A deferred BUSY is returned as `connection_budget::ConnectBusy`; it is not + /// a stale-address failure and must not trigger immediate rediscovery/retry. + pub async fn auth_sigma_with_busy_retry_until( + &self, + connection: &Arc, + node_id: u64, + controller_id: u64, + deadline: tokio::time::Instant, + ) -> Result { + self.auth_sigma_with_budget(connection, node_id, controller_id, Some(deadline)) + .await + } + + async fn auth_sigma_with_budget( + &self, + connection: &Arc, + node_id: u64, + controller_id: u64, + deadline: Option, + ) -> Result { + use crate::connection_budget::{ConnectStage, ConnectionBudget}; + let budget = ConnectionBudget::new(node_id, deadline); const MAX_BUSY_RETRIES: u32 = 5; const DEFAULT_BUSY_WAIT: Duration = Duration::from_millis(3000); @@ -134,31 +158,63 @@ impl Controller { let mut busy_retries = 0u32; loop { - match auth_sigma(connection.as_ref(), &self.fabric, self.certmanager.as_ref(), node_id, controller_id).await { - Ok((ses, resumption)) => { + let result = match budget + .run( + ConnectStage::Resume, + self.try_auth_sigma_resume(connection, node_id, controller_id), + ) + .await + { + Ok(Some(session)) => return Ok(session), + Ok(None) => { + budget + .run( + ConnectStage::Case, + auth_sigma( + connection.as_ref(), + &self.fabric, + self.certmanager.as_ref(), + node_id, + controller_id, + ), + ) + .await + } + Err(error) => Err(error), + }; + match result { + Ok((session, resumption)) => { if let Some(record) = resumption { self.resumption.lock().await.insert(node_id, record); } - return Ok(ses); + return Ok(session); } Err(e) => { if let Some(busy) = e.downcast_ref::() { if busy_retries < MAX_BUSY_RETRIES { - let wait = busy.wait_ms + let requested = busy + .wait_ms .map(|ms| Duration::from_millis(ms.into())) - .unwrap_or(DEFAULT_BUSY_WAIT) - .min(MAX_BUSY_WAIT); + .unwrap_or(DEFAULT_BUSY_WAIT); + // Bounded callers defer instead of shortening a peer's + // minimum wait. Retain the legacy unbounded API cap. + let wait = if deadline.is_some() { + requested + } else { + requested.min(MAX_BUSY_WAIT) + }; log::info!( - "CASE responder BUSY, waiting {:?} before retry ({}/{})", - wait, busy_retries + 1, MAX_BUSY_RETRIES + "Matter CASE BUSY node={} wait_ms={} retry={}", + node_id, + wait.as_millis(), + busy_retries + 1 ); - tokio::time::sleep(wait).await; + budget.busy_wait(wait).await?; busy_retries += 1; continue; } - return Err(e).context(format!( - "still BUSY after {} retries", MAX_BUSY_RETRIES - )); + return Err(e) + .context(format!("still BUSY after {} retries", MAX_BUSY_RETRIES)); } return Err(e); } @@ -201,6 +257,9 @@ impl Controller { if sigma2.protocol_header.protocol_id == messages::ProtocolMessageHeader::PROTOCOL_ID_SECURE_CHANNEL && sigma2.protocol_header.opcode == messages::ProtocolMessageHeader::OPCODE_STATUS { + if let Some(status) = sigma2.status_report_info.as_ref().filter(|status| status.is_busy()) { + return Err(SigmaBusy { wait_ms: status.minimum_wait_time_ms() }.into()); + } log::debug!( "CASE resume: responder rejected with status report, falling back to full SIGMA (exchange:{} {:?})", exchange, @@ -1018,6 +1077,10 @@ pub(crate) async fn auth_sigma( Ok((ses, resumption)) } +#[cfg(test)] +#[path = "controller/connect_deadline_tests.rs"] +mod connect_deadline_tests; + #[cfg(test)] mod tests { use super::*; diff --git a/src/controller/connect_deadline_tests.rs b/src/controller/connect_deadline_tests.rs new file mode 100644 index 0000000..67a1fe3 --- /dev/null +++ b/src/controller/connect_deadline_tests.rs @@ -0,0 +1,348 @@ +//! Public CASE entrypoints against an in-memory responder; no device I/O. +use super::*; +use crate::connection_budget::{ConnectBusy, ConnectDeadline, ConnectStage}; +use crate::messages::{ProtocolMessageHeader, SecureChannelGeneralCode, SecureChannelProtocolCode}; +use crate::{im, tlv, util}; +use std::collections::VecDeque; +use tokio::time::Instant; + +struct Certificates(p256::SecretKey); +impl certmanager::CertManager for Certificates { + fn get_ca_cert(&self) -> Result> { + anyhow::bail!("unused fixture certificate") + } + fn get_ca_key(&self) -> Result { + Ok(self.0.clone()) + } + fn get_ca_public_key(&self) -> Result> { + Ok(self.0.public_key().to_sec1_bytes().to_vec()) + } + fn get_user_cert(&self, _: u64) -> Result> { + anyhow::bail!("unused fixture certificate") + } + fn get_user_key(&self, _: u64) -> Result { + Ok(self.0.clone()) + } + fn get_fabric_id(&self) -> u64 { + 1 + } + fn get_ipk_epoch_key(&self) -> Vec { + vec![0; 16] + } +} + +enum Reply { + Busy(u32), + Resume, + Hang, + Fail, +} +struct Responder { + replies: tokio::sync::Mutex>, + current: tokio::sync::Mutex)>>, + requests: std::sync::Mutex>, +} +impl Responder { + fn new(replies: Vec<(Duration, Reply)>) -> Arc { + Arc::new(Self { + replies: tokio::sync::Mutex::new(replies.into()), + current: tokio::sync::Mutex::new(None), + requests: Default::default(), + }) + } +} + +#[async_trait::async_trait] +impl ConnectionTrait for Responder { + async fn send(&self, data: &[u8]) -> Result<()> { + let message = Message::decode(data)?; + assert_eq!( + message.protocol_header.protocol_id, 0, + "no physical IM command" + ); + if message.protocol_header.opcode != ProtocolMessageHeader::OPCODE_CASE_SIGMA1 { + return Ok(()); + } + self.requests + .lock() + .unwrap() + .push((Instant::now(), message.tlv.get_octet_string(&[6]).is_some())); + let (delay, reply) = self + .replies + .lock() + .await + .pop_front() + .unwrap_or((Duration::ZERO, Reply::Hang)); + let header = &message.protocol_header; + let bytes = match reply { + Reply::Busy(ms) => { + let mut body = crate::device_messages::status_report( + header.exchange_id, + SecureChannelGeneralCode::Busy as u16, + 0, + SecureChannelProtocolCode::Busy as u16, + message.message_header.message_counter as i64, + )?; + body.extend_from_slice(&ms.to_le_bytes()); + session::Session::new().encode_message(&body)? + } + Reply::Resume => { + let nonce = message.tlv.get_octet_string(&[1]).unwrap(); + let new_id = [3; 16]; + let mut salt = nonce.to_vec(); + salt.extend_from_slice(&new_id); + let key = util::cryptoutil::hkdf_sha256(&salt, &[2; 32], b"Sigma2_Resume", 16)?; + let mic = util::cryptoutil::aes128_ccm_encrypt( + key.as_slice().into(), + b"NCASE_SigmaS2", + &[], + &[], + )?; + let mut tlv = tlv::TlvBuffer::new(); + tlv.write_anon_struct()?; + tlv.write_octetstring(1, &new_id)?; + tlv.write_octetstring(2, &mic)?; + tlv.write_uint16(3, 42)?; + tlv.write_struct_end()?; + session::Session::new().encode_message(&crate::device_messages::sigma2_msg( + header.exchange_id, + &tlv.data, + message.message_header.message_counter as i64, + )?)? + } + Reply::Hang => return Ok(()), + Reply::Fail => { + *self.current.lock().await = Some((delay, Vec::new())); + return Ok(()); + } + }; + *self.current.lock().await = Some((delay, bytes)); + Ok(()) + } + async fn receive(&self, _: Duration) -> Result> { + let next = self.current.lock().await.take(); + if let Some((delay, bytes)) = next { + tokio::time::sleep(delay).await; + if bytes.is_empty() { + anyhow::bail!("opaque-secret-fixture-must-not-be-logged") + } + Ok(bytes) + } else { + std::future::pending().await + } + } + fn is_reliable(&self) -> bool { + true + } +} + +async fn controller(resume: bool) -> Arc { + let cm: Arc = Arc::new(Certificates(p256::SecretKey::random( + &mut rand::thread_rng(), + ))); + let transport = transport::Transport::new("127.0.0.1:0").await.unwrap(); + let controller = Controller::new(&cm, &transport, 1).unwrap(); + if resume { + controller.resumption.lock().await.insert( + 7, + sigma::ResumptionRecord { + resumption_id: [1; 16], + shared_secret: [2; 32], + }, + ); + } + controller +} + +#[tokio::test(start_paused = true)] +async fn resume_busy_waits_before_retry_and_then_authenticates() { + let controller = controller(true).await; + let responder = Responder::new(vec![ + (Duration::ZERO, Reply::Busy(5000)), + (Duration::ZERO, Reply::Resume), + ]); + let conn: Arc = responder.clone(); + let started = Instant::now(); + let result = controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, started + Duration::from_secs(10)) + .await + .unwrap(); + assert_eq!(result.session_id, 42); + let sent = responder.requests.lock().unwrap(); + assert_eq!(sent.len(), 2); + assert!(sent.iter().all(|(_, resume)| *resume)); + assert!(sent[1].0.duration_since(sent[0].0) >= Duration::from_secs(5)); +} + +#[tokio::test(start_paused = true)] +async fn full_case_busy_uses_the_same_wait_and_deferral_policy() { + let controller = controller(false).await; + let responder = Responder::new(vec![ + (Duration::ZERO, Reply::Busy(5000)), + (Duration::ZERO, Reply::Busy(5000)), + ]); + let conn: Arc = responder.clone(); + let started = Instant::now(); + let error = controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, started + Duration::from_secs(8)) + .await + .err() + .unwrap(); + let busy = error.downcast_ref::().unwrap(); + assert!(busy.retry_at >= started + Duration::from_secs(10)); + let sent = responder.requests.lock().unwrap(); + assert_eq!(sent.len(), 2); + assert!(sent.iter().all(|(_, resume)| !*resume)); + assert!(sent[1].0.duration_since(sent[0].0) >= Duration::from_secs(5)); +} + +#[tokio::test(start_paused = true)] +async fn legacy_public_api_also_honors_resume_busy_without_new_deadline() { + let controller = controller(true).await; + let responder = Responder::new(vec![ + (Duration::ZERO, Reply::Busy(5000)), + (Duration::ZERO, Reply::Resume), + ]); + let conn: Arc = responder.clone(); + assert_eq!( + controller + .auth_sigma_with_busy_retry(&conn, 7, 1) + .await + .unwrap() + .session_id, + 42 + ); + let sent = responder.requests.lock().unwrap(); + assert_eq!(sent.len(), 2); + assert!(sent.iter().all(|(_, resume)| *resume)); + assert!(sent[1].0.duration_since(sent[0].0) >= Duration::from_secs(5)); +} + +#[tokio::test(start_paused = true)] +async fn busy_longer_than_remaining_deadline_does_not_start_full_case() { + let controller = controller(true).await; + let responder = Responder::new(vec![(Duration::ZERO, Reply::Busy(5000))]); + let conn: Arc = responder.clone(); + let started = Instant::now(); + let error = controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, started + Duration::from_secs(2)) + .await + .err() + .unwrap(); + assert!(error.is::()); + assert_eq!(responder.requests.lock().unwrap().len(), 1); + assert!(Instant::now().duration_since(started) < Duration::from_secs(2)); +} + +#[tokio::test(start_paused = true)] +async fn expired_deadline_never_sends_and_hung_resume_returns_typed_stage() { + let controller = controller(true).await; + let responder = Responder::new(vec![]); + let conn: Arc = responder.clone(); + let error = controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, Instant::now()) + .await + .err() + .unwrap(); + assert_eq!( + error.downcast_ref::().unwrap().stage, + ConnectStage::Resume + ); + assert!(responder.requests.lock().unwrap().is_empty()); + let end = Instant::now() + Duration::from_secs(2); + let error = controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, end) + .await + .err() + .unwrap(); + assert_eq!( + error.downcast_ref::().unwrap().stage, + ConnectStage::Resume + ); + assert_eq!(Instant::now(), end); +} + +#[tokio::test(start_paused = true)] +async fn bounded_api_never_shortens_an_advertised_minimum_wait() { + let controller = controller(true).await; + let responder = Responder::new(vec![(Duration::ZERO, Reply::Busy(120_000))]); + let conn: Arc = responder; + let started = Instant::now(); + let error = controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, started + Duration::from_secs(10)) + .await + .err() + .unwrap(); + assert_eq!( + error.downcast_ref::().unwrap().retry_at, + started + Duration::from_secs(120) + ); +} + +#[tokio::test(start_paused = true)] +async fn late_valid_resume_cannot_return_connection_authority() { + let controller = controller(true).await; + let responder = Responder::new(vec![(Duration::from_secs(2), Reply::Resume)]); + let conn: Arc = responder; + let error = controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, Instant::now() + Duration::from_secs(2)) + .await + .err() + .unwrap(); + assert!(error.is::()); +} + +#[tokio::test(start_paused = true)] +async fn cancelling_busy_wait_never_retries_after_caller_is_gone() { + let controller = controller(true).await; + let responder = Responder::new(vec![(Duration::ZERO, Reply::Busy(5000))]); + let conn: Arc = responder.clone(); + let task = tokio::spawn(async move { + controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, Instant::now() + Duration::from_secs(20)) + .await + }); + while responder.requests.lock().unwrap().is_empty() { + tokio::task::yield_now().await; + } + tokio::task::yield_now().await; + task.abort(); + let _ = task.await; + tokio::time::advance(Duration::from_secs(30)).await; + assert_eq!(responder.requests.lock().unwrap().len(), 1); +} + +#[tokio::test(start_paused = true)] +async fn failed_case_does_not_close_an_unrelated_subscription() { + let controller = controller(true).await; + let responder = Responder::new(vec![(Duration::ZERO, Reply::Fail)]); + let conn: Arc = responder; + let other: Arc = Responder::new(vec![]); + let live = Connection::from_parts(other, session::Session::new()); + let rx = live.active.register_subscription(99); + let mut subscription = Subscription { + subscription_id: 99, + max_interval: 30, + priming_attribute_reports: vec![], + priming_event_reports: vec![], + rx, + registry: live.active.subscriptions_handle(), + }; + assert!(controller + .auth_sigma_with_busy_retry_until(&conn, 7, 1, Instant::now() + Duration::from_secs(2)) + .await + .is_err()); + live.active + .subscriptions_handle() + .lock() + .unwrap() + .get(&99) + .unwrap() + .try_send(im::ReportUpdate { + subscription_id: 99, + attribute_reports: vec![], + event_reports: vec![], + }) + .unwrap(); + assert_eq!(subscription.next().await.unwrap().subscription_id, 99); +} diff --git a/src/devman/mod.rs b/src/devman/mod.rs index a6b2d54..ae69a37 100644 --- a/src/devman/mod.rs +++ b/src/devman/mod.rs @@ -188,6 +188,27 @@ impl DeviceManager { self.connect_with_rediscovery(node_id, &address).await } + /// Connect under one absolute deadline. BUSY wait and deadline errors retain + /// their typed, safe diagnostics and never trigger immediate rediscovery. + pub async fn connect_until( + &self, + node_id: u64, + deadline: tokio::time::Instant, + ) -> Result { + let address = { + let reg = self + .registry + .lock() + .map_err(|e| anyhow::anyhow!("registry lock: {}", e))?; + reg.get(node_id) + .context("Matter device not found in registry")? + .address + .clone() + }; + self.connect_with_budget(node_id, &address, Some(deadline)) + .await + } + /// Connect to a previously commissioned device by friendly name. /// If the stored address fails, automatically re-discovers the device via operational mDNS. pub async fn connect_by_name(&self, name: &str) -> Result { @@ -204,7 +225,24 @@ impl DeviceManager { /// Connect with BUSY retry and one round of mDNS rediscovery on non-BUSY failure. /// BUSY handling is delegated to Controller::auth_sigma_with_busy_retry so that /// in-place reauth uses identical retry semantics. - async fn connect_with_rediscovery(&self, node_id: u64, address: &str) -> Result { + async fn connect_with_rediscovery( + &self, + node_id: u64, + address: &str, + ) -> Result { + self.connect_with_budget(node_id, address, None).await + } + + async fn connect_with_budget( + &self, + node_id: u64, + address: &str, + deadline: Option, + ) -> Result { + use crate::connection_budget::{ + ConnectBusy, ConnectDeadline, ConnectStage, ConnectionBudget, + }; + let budget = ConnectionBudget::new(node_id, deadline); let stored_mrp = self .registry .lock() @@ -214,31 +252,80 @@ impl DeviceManager { .unwrap_or_default(); let mut current_address = address.to_string(); // Create connection once and reuse across retries; only replace if address changes. - let mut conn = self.transport.create_connection(¤t_address).await; + let mut conn = budget + .run(ConnectStage::Transport, async { + Ok(self.transport.create_connection(¤t_address).await) + }) + .await?; conn.set_mrp_params(stored_mrp); - match self.controller.auth_sigma_with_busy_retry(&conn, node_id, self.config.controller_id).await { + let result = match deadline { + Some(deadline) => { + self.controller + .auth_sigma_with_busy_retry_until( + &conn, + node_id, + self.config.controller_id, + deadline, + ) + .await + } + None => { + self.controller + .auth_sigma_with_busy_retry(&conn, node_id, self.config.controller_id) + .await + } + }; + match result { Ok(ses) => Ok(controller::Connection::from_parts(conn, ses)), + Err(error) if error.is::() || error.is::() => Err(error), Err(e) => { // Try operational mDNS rediscovery once, then one more attempt. - log::info!( - "Connection to {} failed ({}), attempting operational rediscovery...", - current_address, e - ); - let (new_address, matter_info) = self - .discover_device_info(node_id, Duration::from_secs(10)) - .await - .context(format!("rediscovery for node {} after connect failure", node_id))?; - current_address = new_address; - conn = self.transport.create_connection(¤t_address).await; - conn.set_mrp_params(matter_info.mrp_params()); - let ses = self - .controller - .auth_sigma_with_busy_retry(&conn, node_id, self.config.controller_id) + if deadline.is_none() { + log::info!( + "Connection to {} failed ({}), attempting operational rediscovery...", + current_address, + e + ); + } + let (new_address, matter_info) = budget + .run( + ConnectStage::Rediscovery, + self.discover_device_info(node_id, Duration::from_secs(10)), + ) .await .context(format!( - "connection still failed after rediscovery at {}", current_address + "rediscovery for node {} after connect failure", + node_id ))?; + current_address = new_address; + conn = budget + .run(ConnectStage::Transport, async { + Ok(self.transport.create_connection(¤t_address).await) + }) + .await?; + conn.set_mrp_params(matter_info.mrp_params()); + let ses = match deadline { + Some(deadline) => { + self.controller + .auth_sigma_with_busy_retry_until( + &conn, + node_id, + self.config.controller_id, + deadline, + ) + .await + } + None => { + self.controller + .auth_sigma_with_busy_retry(&conn, node_id, self.config.controller_id) + .await + } + } + .context(format!( + "connection still failed after rediscovery at {}", + current_address + ))?; Ok(controller::Connection::from_parts(conn, ses)) } } diff --git a/src/lib.rs b/src/lib.rs index 4a8389a..a334e5a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -283,6 +283,7 @@ pub mod clusters; mod commission; pub use commission::NetworkCreds; pub mod controller; +pub mod connection_budget; pub mod device; mod device_messages; pub mod devman; diff --git a/src/tlv.rs b/src/tlv.rs index f9f9854..46d117f 100644 --- a/src/tlv.rs +++ b/src/tlv.rs @@ -425,6 +425,25 @@ fn read_tag(tagctrl: u8, cursor: &mut Cursor<&[u8]>) -> Result { } } +fn read_value_bytes(cursor: &mut Cursor<&[u8]>, size: u64) -> Result> { + let remaining = (cursor.get_ref().len() as u64).saturating_sub(cursor.position()); + if size > remaining { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "TLV value length exceeds remaining input", + )); + } + let size = usize::try_from(size).map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "TLV value length does not fit usize", + ) + })?; + let mut value = vec![0; size]; + cursor.read_exact(&mut value)?; + Ok(value) +} + fn decode(cursor: &mut Cursor<&[u8]>, container: &mut Vec) -> Result<()> { while cursor.position() < cursor.get_ref().len() as u64 { let fb = cursor.read_u8()?; @@ -529,8 +548,7 @@ fn decode(cursor: &mut Cursor<&[u8]>, container: &mut Vec) -> Result<() TYPE_UTF8_L1 => { // utf8 string let size = cursor.read_u8()?; - let mut value = vec![0; size as usize]; - cursor.read_exact(&mut value)?; + let value = read_value_bytes(cursor, u64::from(size))?; let str = String::from_utf8(value); let typ = match str { Ok(s) => TlvItemValue::String(s), @@ -542,8 +560,7 @@ fn decode(cursor: &mut Cursor<&[u8]>, container: &mut Vec) -> Result<() TYPE_OCTET_STRING_L1 => { // octet string let size = cursor.read_u8()?; - let mut value = vec![0; size as usize]; - cursor.read_exact(&mut value)?; + let value = read_value_bytes(cursor, u64::from(size))?; let item = TlvItem { tag, value: TlvItemValue::OctetString(value), @@ -553,8 +570,7 @@ fn decode(cursor: &mut Cursor<&[u8]>, container: &mut Vec) -> Result<() TYPE_OCTET_STRING_L2 => { // octet string large let size = cursor.read_u16::()?; - let mut value = vec![0; size as usize]; - cursor.read_exact(&mut value)?; + let value = read_value_bytes(cursor, u64::from(size))?; let item = TlvItem { tag, value: TlvItemValue::OctetString(value), @@ -564,8 +580,7 @@ fn decode(cursor: &mut Cursor<&[u8]>, container: &mut Vec) -> Result<() TYPE_OCTET_STRING_L4 => { // octet string very large let size = cursor.read_u32::()?; - let mut value = vec![0; size as usize]; - cursor.read_exact(&mut value)?; + let value = read_value_bytes(cursor, u64::from(size))?; let item = TlvItem { tag, value: TlvItemValue::OctetString(value), @@ -575,8 +590,7 @@ fn decode(cursor: &mut Cursor<&[u8]>, container: &mut Vec) -> Result<() TYPE_OCTET_STRING_L8 => { // octet string very very xtra large let size = cursor.read_u64::()?; - let mut value = vec![0; size as usize]; - cursor.read_exact(&mut value)?; + let value = read_value_bytes(cursor, size)?; let item = TlvItem { tag, value: TlvItemValue::OctetString(value), @@ -788,6 +802,9 @@ impl TlvItemEnc { } } +#[cfg(test)] +mod length_tests; + #[cfg(test)] mod tests { use super::{decode_tlv, TlvBuffer, TlvItemEnc, TlvItemValue, TlvItemValueEnc}; diff --git a/src/tlv/length_tests.rs b/src/tlv/length_tests.rs new file mode 100644 index 0000000..aa03c45 --- /dev/null +++ b/src/tlv/length_tests.rs @@ -0,0 +1,128 @@ +use super::{decode_tlv, TlvItemValue}; +use std::io::ErrorKind; + +const VALUE_TYPES: [(u8, usize); 5] = [(0x0c, 1), (0x10, 1), (0x11, 2), (0x12, 4), (0x13, 8)]; + +fn encoded_value(kind: u8, width: usize, declared: u64, payload: &[u8]) -> Vec { + let mut input = vec![kind]; + input.extend_from_slice(&declared.to_le_bytes()[..width]); + input.extend_from_slice(payload); + input +} + +fn expected_value(kind: u8, payload: &[u8]) -> TlvItemValue { + if kind == 0x0c { + TlvItemValue::String(String::from_utf8(payload.to_vec()).unwrap()) + } else { + TlvItemValue::OctetString(payload.to_vec()) + } +} + +#[test] +fn test_decode_empty_length_values() { + for (kind, width) in VALUE_TYPES { + let input = encoded_value(kind, width, 0, &[]); + assert_eq!(decode_tlv(&input).unwrap().value, expected_value(kind, &[])); + } +} + +#[test] +fn test_decode_exact_remaining_length_values() { + let payload = b"value"; + for (kind, width) in VALUE_TYPES { + let input = encoded_value(kind, width, payload.len() as u64, payload); + assert_eq!( + decode_tlv(&input).unwrap().value, + expected_value(kind, payload) + ); + } +} + +#[test] +fn test_decode_truncated_length_fields() { + for (kind, width) in VALUE_TYPES { + let input = encoded_value(kind, width, 1, &[]); + for length_bytes in 0..width { + assert_eq!( + decode_tlv(&input[..1 + length_bytes]).unwrap_err().kind(), + ErrorKind::UnexpectedEof, + "kind={kind:#x}, length_bytes={length_bytes}", + ); + } + } +} + +#[test] +fn test_decode_length_exceeding_remaining_input() { + for (kind, width) in VALUE_TYPES { + for payload in [&b""[..], &b"a"[..], &b"abc"[..]] { + let input = encoded_value(kind, width, 4, payload); + assert_eq!( + decode_tlv(&input).unwrap_err().kind(), + ErrorKind::UnexpectedEof, + "kind={kind:#x}, remaining={}", + payload.len(), + ); + } + } +} + +#[test] +fn test_decode_maximum_declared_lengths_without_payload() { + for (kind, width) in VALUE_TYPES { + let declared = match width { + 1 => u64::from(u8::MAX), + 2 => u64::from(u16::MAX), + 4 => u64::from(u32::MAX), + 8 => u64::MAX, + _ => unreachable!(), + }; + let input = encoded_value(kind, width, declared, &[]); + assert_eq!( + decode_tlv(&input).unwrap_err().kind(), + ErrorKind::UnexpectedEof, + "kind={kind:#x}", + ); + } +} + +#[test] +fn test_decode_large_declared_lengths_without_payload() { + for (kind, width) in [(0x12, 4), (0x13, 8)] { + let input = encoded_value(kind, width, 2_080_522_737, &[]); + assert_eq!( + decode_tlv(&input).unwrap_err().kind(), + ErrorKind::UnexpectedEof, + ); + } +} + +#[test] +fn test_decode_length_value_preserves_following_element() { + for (kind, width) in VALUE_TYPES { + let mut input = vec![0x15]; + input.push(kind | 0x20); + input.push(1); + input.extend_from_slice(&encoded_value(kind, width, 2, b"ok")[1..]); + input.extend_from_slice(&[0x24, 2, 7, 0x18]); + let decoded = decode_tlv(&input).unwrap(); + assert_eq!(decoded.get(&[1]), Some(&expected_value(kind, b"ok"))); + assert_eq!(decoded.get_int(&[2]), Some(7)); + } +} + +#[test] +fn test_decode_valid_large_values_without_transport_cap() { + let payload = vec![0x5a; 65_536]; + for (kind, width) in [(0x12, 4), (0x13, 8)] { + let input = encoded_value(kind, width, payload.len() as u64, &payload); + let decoded = decode_tlv(&input).unwrap(); + assert_eq!(decoded.get_octet_string(&[]), Some(payload.as_slice())); + } +} + +#[test] +fn test_decode_invalid_utf8_value_preserves_invalid_result() { + let input = encoded_value(0x0c, 1, 1, &[0xff]); + assert_eq!(decode_tlv(&input).unwrap().value, TlvItemValue::Invalid()); +}