use std::collections::{HashMap, VecDeque};
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::time::Instant;
use matter_commissioning::driver::AsyncDatagram;
use matter_commissioning::NocRng;
use matter_transport::{
DecodeInboundOutput, Discovery, MrpEvent, MrpFlags, ProtocolHeader, ProtocolId, QueryHandle,
ServiceKind, SessionId, SessionManager, SessionRole,
};
use tokio::sync::{mpsc, oneshot};
use crate::error::Error;
use crate::fabric::FabricConfig;
use crate::snapshot;
use crate::state::ControllerState;
use crate::store::ControllerStore;
use crate::subscription::{AttributeReport, SubscriptionEvent, SUBSCRIPTION_CHANNEL_CAP};
const OP_SUBSCRIBE_REQUEST: u8 = 0x03;
const OP_SUBSCRIBE_RESPONSE: u8 = 0x04;
const OP_REPORT_DATA: u8 = 0x05;
const OP_STATUS_RESPONSE: u8 = 0x01;
const OP_TIMED_REQUEST: u8 = 0x0a;
const OP_WRITE_REQUEST: u8 = 0x06;
const OP_WRITE_RESPONSE: u8 = 0x07;
const NEEDS_TIMED_INTERACTION: u8 = 0xc6;
fn response_needs_timed(opcode: u8, payload: &[u8]) -> bool {
if matches!(
matter_interaction::parse_status_response(payload),
Ok(Some(NEEDS_TIMED_INTERACTION))
) {
return true;
}
match opcode {
crate::node::OP_INVOKE_REQUEST => matches!(
matter_interaction::parse_invoke_response_batch(payload),
Ok(entries) if entries.iter().any(|e| matches!(
e.response,
matter_interaction::InvokeResponse::Status(s)
if s.to_u8() == NEEDS_TIMED_INTERACTION
))
),
OP_WRITE_REQUEST => matches!(
matter_interaction::parse_write_response(payload),
Ok(statuses)
if statuses.iter().any(|(_, s)| s.to_u8() == NEEDS_TIMED_INTERACTION)
),
_ => false,
}
}
const LIVENESS_TICK: std::time::Duration = std::time::Duration::from_millis(250);
const IDLE_PARK_MAX: std::time::Duration = std::time::Duration::from_secs(3600);
#[cfg(not(test))]
const RESOLVE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30);
#[cfg(test)]
const RESOLVE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(2);
const SEEN_RECORD_TTL: std::time::Duration = std::time::Duration::from_secs(60);
const SEEN_RECORD_CAP: usize = 256;
const MAX_READ_CHUNKS: usize = 64;
const MAX_READ_BYTES: usize = 256 * 1024;
const MATTER_GROUP_PORT: u16 = 5540;
const GROUP_COUNTER_BLOCK: u32 = 64;
const RESUB_MAX_FIBONACCI_STEP_INDEX: u32 = 14;
const RESUB_WAIT_TIME_MULTIPLIER_MS: u64 = 10_000;
const RESUB_MAX_RETRY_WAIT_INTERVAL_MS: u64 = 5_538_000;
const RESUB_MIN_WAIT_PERCENT: u64 = 30;
const LIVENESS_GRACE: std::time::Duration = std::time::Duration::from_secs(5);
fn fibonacci(n: u32) -> u64 {
let (mut a, mut b) = (0u64, 1u64);
for _ in 0..n {
let next = a + b;
a = b;
b = next;
}
a
}
fn resubscribe_backoff(rng: &dyn NocRng, retry_count: u32) -> std::time::Duration {
let max_wait_ms = if retry_count <= RESUB_MAX_FIBONACCI_STEP_INDEX {
fibonacci(retry_count).saturating_mul(RESUB_WAIT_TIME_MULTIPLIER_MS)
} else {
RESUB_MAX_RETRY_WAIT_INTERVAL_MS
};
let min_wait_ms = (RESUB_MIN_WAIT_PERCENT * max_wait_ms) / 100;
let span = max_wait_ms - min_wait_ms;
let jitter = if span == 0 {
0
} else {
let mut buf = [0u8; 8];
let _ = rng.fill(&mut buf);
u64::from_le_bytes(buf) % span
};
std::time::Duration::from_millis(min_wait_ms + jitter)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub(crate) struct SubId(pub(crate) u64);
struct ReportSink {
report_tx: mpsc::Sender<SubscriptionEvent>,
ctrl_tx: mpsc::UnboundedSender<SubscriptionEvent>,
dropped: usize,
}
impl ReportSink {
fn try_send_report(&mut self, report: AttributeReport) -> bool {
if self.dropped > 0 {
match self.report_tx.try_send(SubscriptionEvent::Lagged {
dropped: self.dropped,
}) {
Ok(()) => self.dropped = 0,
Err(mpsc::error::TrySendError::Closed(_)) => return false,
Err(mpsc::error::TrySendError::Full(_)) => {}
}
}
match self.report_tx.try_send(SubscriptionEvent::Report(report)) {
Ok(()) => true,
Err(mpsc::error::TrySendError::Full(_)) => {
self.dropped += 1;
true
}
Err(mpsc::error::TrySendError::Closed(_)) => false,
}
}
fn try_send_event(&mut self, event: matter_interaction::EventReport) -> bool {
if self.dropped > 0 {
match self.report_tx.try_send(SubscriptionEvent::Lagged {
dropped: self.dropped,
}) {
Ok(()) => self.dropped = 0,
Err(mpsc::error::TrySendError::Closed(_)) => return false,
Err(mpsc::error::TrySendError::Full(_)) => {}
}
}
match self.report_tx.try_send(SubscriptionEvent::Event(event)) {
Ok(()) => true,
Err(mpsc::error::TrySendError::Full(_)) => {
self.dropped += 1;
true
}
Err(mpsc::error::TrySendError::Closed(_)) => false,
}
}
fn send_control(&self, event: SubscriptionEvent) -> bool {
self.ctrl_tx.send(event).is_ok()
}
}
struct SubEntry {
tx: ReportSink,
peer: SocketAddr,
reassembler: ReportReassembler,
session_id: SessionId,
wire_sub_id: u32,
node_id: u64,
paths: Vec<matter_interaction::ReadPath>,
event_paths: Vec<matter_interaction::EventPath>,
event_filters: Vec<matter_interaction::EventFilter>,
min_interval: u16,
max_interval: u16,
liveness_deadline: Instant,
}
struct PendingResubscribe {
sub_id: SubId,
attempt_at: Instant,
node_id: u64,
paths: Vec<matter_interaction::ReadPath>,
event_paths: Vec<matter_interaction::EventPath>,
event_filters: Vec<matter_interaction::EventFilter>,
min_interval: u16,
max_interval: u16,
retry_count: u32,
tx: ReportSink,
}
struct Pending {
node_id: u64,
peer: SocketAddr,
request: PendingRequest,
retried: bool,
reply: PendingReply,
}
struct PendingRequest {
opcode: u8,
protocol_id: ProtocolId,
payload: Vec<u8>,
}
enum PendingReply {
RoundTrip(oneshot::Sender<Result<Vec<u8>, Error>>),
TimedAction {
action_opcode: u8,
action_payload: Vec<u8>,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
},
Action {
opcode: u8,
timed_payload: TimedPayload,
keys: Vec<(u32, u32)>,
timeout_ms: u16,
node_id: u64,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
},
ChunkedWrite {
reply: oneshot::Sender<
Result<
Vec<(
matter_interaction::AttributePath,
matter_interaction::ImStatus,
)>,
Error,
>,
>,
remaining: VecDeque<Vec<u8>>,
statuses: Vec<(
matter_interaction::AttributePath,
matter_interaction::ImStatus,
)>,
},
Read {
reply: oneshot::Sender<Result<Vec<matter_interaction::ReportData>, Error>>,
chunks: Vec<matter_interaction::ReportData>,
total_bytes: usize,
},
Subscribe {
sub_id: SubId,
reply: Option<oneshot::Sender<Result<SubEstablished, Error>>>,
report_tx: ReportSink,
report_rx: Option<SubReceivers>,
priming: Box<ReportReassembler>,
node_id: u64,
paths: Vec<matter_interaction::ReadPath>,
event_paths: Vec<matter_interaction::EventPath>,
event_filters: Vec<matter_interaction::EventFilter>,
min_interval: u16,
max_interval: u16,
retry_count: u32,
},
}
pub(crate) struct SubReceivers {
pub(crate) report_rx: mpsc::Receiver<SubscriptionEvent>,
pub(crate) ctrl_rx: mpsc::UnboundedReceiver<SubscriptionEvent>,
}
pub(crate) type SubEstablished = (SubReceivers, SubId);
const MAX_SUB_CHUNKS: usize = 64;
#[derive(Default)]
struct ReportReassembler {
acc: matter_interaction::ReportAccumulator,
pending_chunks: usize,
}
impl ReportReassembler {
fn push_parsed(
&mut self,
rd: matter_interaction::ReportData,
) -> Option<Vec<(matter_interaction::AttributePath, matter_codec::Value)>> {
let more = rd.more_chunked_messages;
if self.acc.push(rd).is_err() {
self.acc = matter_interaction::ReportAccumulator::default();
self.pending_chunks = 0;
return None;
}
if !more {
self.pending_chunks = 0;
return Some(std::mem::take(&mut self.acc).finish());
}
self.pending_chunks += 1;
if self.pending_chunks > MAX_SUB_CHUNKS {
self.acc = matter_interaction::ReportAccumulator::default();
self.pending_chunks = 0;
}
None
}
#[cfg(test)]
fn push(
&mut self,
payload: &[u8],
) -> Option<Vec<(matter_interaction::AttributePath, matter_codec::Value)>> {
let rd = matter_interaction::parse_report_data(payload).ok()?;
self.push_parsed(rd)
}
}
pub(crate) type TimedPayload = Box<dyn FnOnce() -> Vec<u8> + Send>;
pub(crate) enum Command {
CreateFabric {
cfg: FabricConfig,
reply: oneshot::Sender<Result<u64, Error>>,
},
#[cfg(test)]
RoundTrip {
node_id: u64,
opcode: u8,
protocol_id: matter_transport::ProtocolId,
payload: Vec<u8>,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
},
Read {
node_id: u64,
payload: Vec<u8>,
reply: oneshot::Sender<Result<Vec<matter_interaction::ReportData>, Error>>,
},
Commission {
setup_payload: matter_commissioning::SetupPayload,
label: Option<String>,
reply: oneshot::Sender<Result<crate::NodeInfo, Error>>,
},
#[cfg(feature = "ble")]
CommissionBle {
setup_payload: matter_commissioning::SetupPayload,
network: matter_commissioning::NetworkCredentials,
label: Option<String>,
reply: oneshot::Sender<Result<crate::NodeInfo, Error>>,
},
SetNodeVidPid {
node_id: u64,
vendor_id: Option<u16>,
product_id: Option<u16>,
reply: oneshot::Sender<Result<(), Error>>,
},
Subscribe {
node_id: u64,
paths: Vec<matter_interaction::ReadPath>,
event_paths: Vec<matter_interaction::EventPath>,
event_filters: Vec<matter_interaction::EventFilter>,
min_interval: u16,
max_interval: u16,
reply: oneshot::Sender<Result<SubEstablished, Error>>,
},
Action {
node_id: u64,
opcode: u8, plain_payload: Vec<u8>,
timed_payload: TimedPayload,
keys: Vec<(u32, u32)>,
timeout_ms: u16,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
},
TimedRoundTrip {
node_id: u64,
timeout_ms: u16,
action_opcode: u8,
action_payload: Vec<u8>,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
},
ChunkedWrite {
node_id: u64,
chunks: Vec<Vec<u8>>,
reply: oneshot::Sender<
Result<
Vec<(
matter_interaction::AttributePath,
matter_interaction::ImStatus,
)>,
Error,
>,
>,
},
CommissionerNodeId {
reply: oneshot::Sender<Result<u64, Error>>,
},
ListNodes {
reply: oneshot::Sender<Vec<crate::NodeInfo>>,
},
ForgetNode {
node_id: u64,
reply: oneshot::Sender<Result<bool, Error>>,
},
ResumptionRecordFor {
node_id: u64,
reply: oneshot::Sender<Result<Option<Vec<u8>>, Error>>,
},
StoreResumptionRecord {
node_id: u64,
record_bytes: Vec<u8>,
reply: oneshot::Sender<Result<(), Error>>,
},
PersistIcdRegistration {
registration: crate::icd::IcdRegistration,
reply: oneshot::Sender<Result<(), Error>>,
},
CreateGroup {
key_set_id: u16,
epoch_start_time: u64,
reply: oneshot::Sender<Result<crate::GroupKeySet, Error>>,
},
InvokeGroup {
group_id: u16,
key_set_id: u16,
path: matter_interaction::CommandPath,
fields_tlv: Vec<u8>,
reply: oneshot::Sender<Result<(), Error>>,
},
CancelSubscription { key: SubId },
#[cfg(test)]
SessionCount { reply: oneshot::Sender<usize> },
}
struct CachedSession {
session_id: SessionId,
peer: std::net::SocketAddr,
}
struct SaveJob {
store: Arc<dyn ControllerStore>,
bytes: Vec<u8>,
seq: u64,
gate: Arc<std::sync::Mutex<u64>>,
}
impl SaveJob {
fn run(self) -> Result<(), crate::store::StoreError> {
let mut last = match self.gate.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if self.seq < *last {
return Ok(());
}
self.store.save(&self.bytes)?;
*last = self.seq;
Ok(())
}
}
async fn save_offloaded(job: SaveJob) -> Result<(), Error> {
match tokio::task::spawn_blocking(move || job.run()).await {
Ok(saved) => Ok(saved?),
Err(join_err) => Err(Error::Operational(format!(
"persistence task failed: {join_err}"
))),
}
}
pub(crate) struct Actor<T: AsyncDatagram, D: Discovery> {
transport: T,
discovery: D,
sessions: SessionManager,
store: Arc<dyn ControllerStore>,
rng: Arc<dyn NocRng>,
state: ControllerState,
cache: HashMap<(u64, u64), CachedSession>, trust: Option<Arc<crate::trust::AttestationTrust>>,
admin_vendor_id: u16,
subscriptions: HashMap<SubId, SubEntry>,
sub_index: HashMap<(SessionId, u32), SubId>,
pending: HashMap<(SessionId, u16), Pending>,
next_sub_id: u64,
snapshot_seq: u64,
save_gate: Arc<std::sync::Mutex<u64>>,
resubscribes: Vec<PendingResubscribe>,
timed_paths: std::collections::HashSet<(u32, u32)>,
commission_tx: mpsc::Sender<CommissionCompletion>,
commission_rx: mpsc::Receiver<CommissionCompletion>,
pending_connects: HashMap<u64, Vec<ConnectWaiter>>,
connect_mrp: HashMap<u64, matter_transport::MrpConfig>,
pending_resolves: Vec<PendingResolve>,
next_resolve_poll: Instant,
resolve_query: Option<QueryHandle>,
seen_records: HashMap<String, SeenRecord>,
multicast_if: Option<u32>,
group_counters: HashMap<u64, u32>,
group_key_cache: HashMap<u64, GroupKeyCacheEntry>,
connect_inbound: HashMap<u64, mpsc::Sender<(Vec<u8>, SocketAddr)>>,
connect_routes: HashMap<SocketAddr, u64>,
connect_outbound_tx: mpsc::Sender<crate::handshake_socket::HandshakeOutbound>,
connect_outbound_rx: mpsc::Receiver<crate::handshake_socket::HandshakeOutbound>,
connect_done_tx: mpsc::Sender<ConnectCompletion>,
connect_done_rx: mpsc::Receiver<ConnectCompletion>,
}
struct GroupKeyCacheEntry {
epoch_key: [u8; 16],
root_public_key: [u8; 65],
op_group_key: [u8; 16],
group_session_id: u16,
privacy_key: [u8; 16],
}
struct ConnectCompletion {
node_id: u64,
result: Result<(matter_crypto::CaseSessionOutput, SocketAddr), Error>,
}
struct PendingResolve {
fabric_id: u64,
node_id: u64,
target: String,
deadline: Instant,
}
struct SeenRecord {
peer: SocketAddr,
peer_mrp: matter_transport::MrpConfig,
seen: Instant,
}
enum ConnectWaiter {
Command(Command),
ResendPending(Pending),
Resubscribe(PendingResubscribe),
}
fn route_key(addr: SocketAddr) -> SocketAddr {
let ip = match addr.ip() {
IpAddr::V6(v6) => v6.to_ipv4_mapped().map_or(IpAddr::V6(v6), IpAddr::V4),
IpAddr::V4(v4) => IpAddr::V4(v4),
};
SocketAddr::new(ip, addr.port())
}
fn command_target_node(cmd: &Command) -> Option<u64> {
match cmd {
Command::Read { node_id, .. }
| Command::Action { node_id, .. }
| Command::TimedRoundTrip { node_id, .. }
| Command::ChunkedWrite { node_id, .. }
| Command::Subscribe { node_id, .. } => Some(*node_id),
#[cfg(test)]
Command::RoundTrip { node_id, .. } => Some(*node_id),
_ => None,
}
}
fn fail_command(cmd: Command, err: Error) {
match cmd {
Command::Read { reply, .. } => {
let _ = reply.send(Err(err));
}
Command::Action { reply, .. } | Command::TimedRoundTrip { reply, .. } => {
let _ = reply.send(Err(err));
}
Command::ChunkedWrite { reply, .. } => {
let _ = reply.send(Err(err));
}
Command::Subscribe { reply, .. } => {
let _ = reply.send(Err(err));
}
#[cfg(test)]
Command::RoundTrip { reply, .. } => {
let _ = reply.send(Err(err));
}
_ => {}
}
}
#[allow(clippy::too_many_arguments)]
async fn run_connect_task(
node_id: u64,
fabric_id: u64,
local_session_id: u16,
credentials: matter_crypto::CaseCredentials,
roots: matter_cert::TrustedRoots,
now: matter_cert::MatterTime,
peer: SocketAddr,
inbound_rx: mpsc::Receiver<(Vec<u8>, SocketAddr)>,
outbound_tx: mpsc::Sender<crate::handshake_socket::HandshakeOutbound>,
done_tx: mpsc::Sender<ConnectCompletion>,
) {
let socket = crate::handshake_socket::HandshakeSocket::new(node_id, outbound_tx, inbound_rx);
let result = matter_commissioning::driver::run_case_establish(
&socket,
peer,
local_session_id,
credentials,
roots,
node_id,
fabric_id,
now,
)
.await
.map(|output| (output, peer))
.map_err(Error::from);
let _ = done_tx.send(ConnectCompletion { node_id, result }).await;
}
struct CommissionCompletion {
fabric_id: u64,
result: Result<matter_commissioning::CommissionedFabric, Error>,
label: Option<String>,
reply: oneshot::Sender<Result<crate::NodeInfo, Error>>,
}
#[allow(clippy::too_many_arguments)]
async fn run_commission_task(
setup_payload: matter_commissioning::SetupPayload,
trust: Arc<crate::trust::AttestationTrust>,
fabric_record: matter_commissioning::FabricRecord,
commissioner_node_id: u64,
ipk_epoch_key: [u8; 16],
commissioner_noc: matter_cert::MatterCertificate,
commissioner_pkcs8: Vec<u8>,
assigned_node_id: u64,
admin_vendor_id: u16,
now: matter_cert::MatterTime,
rng: Arc<dyn matter_commissioning::NocRng>,
multicast_if: Option<u32>,
) -> Result<matter_commissioning::CommissionedFabric, Error> {
use matter_commissioning::driver::{commission, DriverConfig};
use matter_commissioning::CommissionerConfig;
let transport = matter_transport::TokioUdpTransport::bind_with_multicast_if(0, multicast_if)
.await
.map_err(|e| Error::Operational(format!("commission bind: {e}")))?;
let mut discovery = matter_transport::MdnsSdDiscovery::new()
.map_err(|e| Error::Operational(format!("commission mdns: {e}")))?;
let commissioner = CommissionerConfig {
pase_attestation_challenge: [0u8; 16], fabric: &fabric_record,
setup_payload: &setup_payload,
paa_trust_store: &trust.paa,
cd_signing_roots: &trust.cd,
commissioner_node_id,
assigned_node_id,
ipk_epoch_key,
case_admin_subject: commissioner_node_id,
admin_vendor_id,
now,
rng,
network: matter_commissioning::NetworkCredentials::AlreadyOnNetwork,
};
let config = DriverConfig {
commissioner,
commissionable_addr: None, passcode: setup_payload.passcode.as_u32(),
commissioner_noc: &commissioner_noc,
commissioner_signer_pkcs8: &commissioner_pkcs8,
};
commission(&transport, &mut discovery, config)
.await
.map_err(Error::from)
}
impl<T: AsyncDatagram, D: Discovery> Actor<T, D> {
pub(crate) fn new(
transport: T,
discovery: D,
store: Arc<dyn ControllerStore>,
rng: Arc<dyn NocRng>,
state: ControllerState,
trust: Option<crate::trust::AttestationTrust>,
admin_vendor_id: u16,
) -> Self {
Self::new_inner(
transport,
discovery,
store,
rng,
state,
trust,
admin_vendor_id,
)
}
pub(crate) fn with_multicast_if(mut self, multicast_if: Option<u32>) -> Self {
self.multicast_if = multicast_if;
self
}
#[allow(clippy::too_many_arguments)] fn new_inner(
transport: T,
discovery: D,
store: Arc<dyn ControllerStore>,
rng: Arc<dyn NocRng>,
state: ControllerState,
trust: Option<crate::trust::AttestationTrust>,
admin_vendor_id: u16,
) -> Self {
let (commission_tx, commission_rx) = mpsc::channel(8);
let (connect_outbound_tx, connect_outbound_rx) = mpsc::channel(64);
let (connect_done_tx, connect_done_rx) = mpsc::channel(8);
Self {
transport,
discovery,
sessions: SessionManager::new(),
store,
rng,
state,
cache: HashMap::new(),
trust: trust.map(Arc::new),
admin_vendor_id,
subscriptions: HashMap::new(),
sub_index: HashMap::new(),
pending: HashMap::new(),
next_sub_id: 1,
snapshot_seq: 0,
save_gate: Arc::new(std::sync::Mutex::new(0)),
resubscribes: Vec::new(),
timed_paths: std::collections::HashSet::new(),
commission_tx,
commission_rx,
pending_connects: HashMap::new(),
connect_mrp: HashMap::new(),
pending_resolves: Vec::new(),
next_resolve_poll: Instant::now(),
resolve_query: None,
seen_records: HashMap::new(),
multicast_if: None,
group_counters: HashMap::new(),
group_key_cache: HashMap::new(),
connect_inbound: HashMap::new(),
connect_routes: HashMap::new(),
connect_outbound_tx,
connect_outbound_rx,
connect_done_tx,
connect_done_rx,
}
}
pub(crate) async fn run(mut self, mut rx: mpsc::Receiver<Command>) {
loop {
let now = Instant::now();
let next_deadline = self.next_timer_deadline();
if next_deadline.is_some_and(|d| d <= now) {
self.drive_mrp().await;
self.check_liveness();
self.drive_resubscribes().await;
self.drive_pending_resolves();
continue;
}
let sleep_for =
next_deadline.map_or(IDLE_PARK_MAX, |d| d.saturating_duration_since(now));
tokio::select! {
biased;
maybe = rx.recv() => match maybe {
Some(cmd) => self.dispatch(cmd).await,
None => return self.shutdown_discovery(),
},
Some(completion) = self.commission_rx.recv() => {
self.handle_commission_completion(completion).await;
}
Some(out) = self.connect_outbound_rx.recv() => {
self.handle_connect_outbound(out).await;
}
Some(done) = self.connect_done_rx.recv() => {
self.handle_connect_done(done).await;
}
recv = self.transport.recv_from() => {
if let Ok((packet, from)) = recv {
self.handle_inbound(&packet, from).await;
}
}
() = tokio::time::sleep(sleep_for) => {
tracing::trace!(target: "matter_controller::actor", "timer wake");
self.drive_mrp().await;
self.check_liveness();
self.drive_resubscribes().await;
self.drive_pending_resolves();
}
}
}
}
fn next_timer_deadline(&self) -> Option<Instant> {
let mrp = self.sessions.poll_timeout();
let liveness = self
.subscriptions
.values()
.map(|e| e.liveness_deadline)
.min();
let resub = self.resubscribes.iter().map(|pr| pr.attempt_at).min();
let resolve = if self.pending_resolves.is_empty() {
None
} else {
Some(self.next_resolve_poll)
};
[mrp, liveness, resub, resolve].into_iter().flatten().min()
}
async fn dispatch(&mut self, cmd: Command) {
if let Some(node_id) = command_target_node(&cmd) {
if let Ok(fabric_id) = self.sole_fabric().map(|f| f.fabric_id) {
if !self.cache.contains_key(&(fabric_id, node_id)) {
self.enqueue_connect_waiter(fabric_id, node_id, ConnectWaiter::Command(cmd));
return;
}
}
}
self.dispatch_ready(cmd).await;
}
#[allow(clippy::too_many_lines)]
async fn dispatch_ready(&mut self, cmd: Command) {
match cmd {
Command::CreateFabric { cfg, reply } => {
let _ = reply.send(self.handle_create_fabric(&cfg).await);
}
#[cfg(test)]
Command::RoundTrip {
node_id,
opcode,
protocol_id,
payload,
reply,
} => {
self.start_round_trip(node_id, opcode, protocol_id, payload, reply)
.await;
}
Command::Read {
node_id,
payload,
reply,
} => {
self.start_read(node_id, payload, reply).await;
}
Command::Commission {
setup_payload,
label,
reply,
} => {
self.spawn_commission(setup_payload, label, reply);
}
#[cfg(feature = "ble")]
Command::CommissionBle {
setup_payload,
network,
label,
reply,
} => {
self.spawn_commission_ble(setup_payload, network, label, reply);
}
Command::Subscribe {
node_id,
paths,
event_paths,
event_filters,
min_interval,
max_interval,
reply,
} => {
self.start_subscribe(
node_id,
paths,
event_paths,
event_filters,
min_interval,
max_interval,
reply,
)
.await;
}
Command::Action {
node_id,
opcode,
plain_payload,
timed_payload,
keys,
timeout_ms,
reply,
} => {
self.handle_action(
node_id,
opcode,
plain_payload,
timed_payload,
keys,
timeout_ms,
reply,
)
.await;
}
Command::TimedRoundTrip {
node_id,
timeout_ms,
action_opcode,
action_payload,
reply,
} => {
self.start_timed_round_trip(
node_id,
timeout_ms,
action_opcode,
action_payload,
reply,
)
.await;
}
Command::ChunkedWrite {
node_id,
chunks,
reply,
} => {
self.handle_chunked_write(node_id, chunks, reply).await;
}
Command::CommissionerNodeId { reply } => {
let _ = reply.send(self.sole_fabric().map(|f| f.commissioner.node_id));
}
Command::ListNodes { reply } => {
let nodes = self
.state
.fabrics
.iter()
.flat_map(|f| {
f.devices.iter().map(move |d| crate::NodeInfo {
node_id: d.node_id,
fabric_id: f.fabric_id,
vendor_id: d.vendor_id,
product_id: d.product_id,
label: d.label.clone(),
})
})
.collect();
let _ = reply.send(nodes);
}
Command::ForgetNode { node_id, reply } => {
let mut removed = false;
for fabric in &mut self.state.fabrics {
let before = fabric.devices.len();
let fabric_id = fabric.fabric_id;
fabric.devices.retain(|d| d.node_id != node_id);
if fabric.devices.len() != before {
removed = true;
if let Some(c) = self.cache.remove(&(fabric_id, node_id)) {
self.sessions.remove(c.session_id);
}
}
}
self.fail_connect_waiters(node_id, &Error::Operational("node forgotten".into()));
self.remove_subscriptions_for_node(node_id);
self.resubscribes.retain(|pr| pr.node_id != node_id);
self.pending.retain(|_, p| p.node_id != node_id);
let outcome = if removed {
match self.durable_save_inputs() {
Ok(job) => save_offloaded(job).await.map(|()| true),
Err(e) => Err(e),
}
} else {
Ok(false)
};
let _ = reply.send(outcome);
}
Command::SetNodeVidPid {
node_id,
vendor_id,
product_id,
reply,
} => {
let _ = reply.send(
self.handle_set_node_vid_pid(node_id, vendor_id, product_id)
.await,
);
}
Command::ResumptionRecordFor { node_id, reply } => {
let result = self.sole_fabric().map(|f| {
f.devices
.iter()
.find(|d| d.node_id == node_id)
.and_then(|d| d.resumption_record.clone())
});
let _ = reply.send(result);
}
Command::StoreResumptionRecord {
node_id,
record_bytes,
reply,
} => {
let _ = reply.send(self.handle_store_resumption_record(node_id, record_bytes));
}
Command::PersistIcdRegistration {
registration,
reply,
} => {
let _ = reply.send(self.handle_persist_icd_registration(registration).await);
}
Command::CreateGroup {
key_set_id,
epoch_start_time,
reply,
} => {
let _ = reply.send(self.handle_create_group(key_set_id, epoch_start_time).await);
}
Command::InvokeGroup {
group_id,
key_set_id,
path,
fields_tlv,
reply,
} => {
let _ = reply.send(
self.handle_invoke_group(group_id, key_set_id, path, &fields_tlv)
.await,
);
}
Command::CancelSubscription { key } => {
self.remove_subscription(key);
self.resubscribes.retain(|pr| pr.sub_id != key);
}
#[cfg(test)]
Command::SessionCount { reply } => {
let _ = reply.send(self.cache.len());
}
}
}
async fn handle_create_fabric(&mut self, cfg: &FabricConfig) -> Result<u64, Error> {
let entry = crate::fabric::create_fabric(cfg, self.rng.as_ref())?;
let fabric_id = entry.fabric_id;
self.state.fabrics.push(entry);
let job = self.durable_save_inputs()?;
save_offloaded(job).await?;
Ok(fabric_id)
}
async fn handle_persist_icd_registration(
&mut self,
registration: crate::icd::IcdRegistration,
) -> Result<(), Error> {
let fabric = self.sole_fabric_mut()?;
fabric
.icd_clients
.retain(|r| r.node_id != registration.node_id);
fabric.icd_clients.push(registration);
let job = self.durable_save_inputs()?;
save_offloaded(job).await?;
Ok(())
}
fn spawn_commission(
&mut self,
setup_payload: matter_commissioning::SetupPayload,
label: Option<String>,
reply: oneshot::Sender<Result<crate::NodeInfo, Error>>,
) {
let Some(trust) = self.trust.clone() else {
let _ = reply.send(Err(Error::NoTrust));
return;
};
let admin_vendor_id = self.admin_vendor_id;
let snapshot = match self.sole_fabric() {
Ok(fabric) => match fabric.to_fabric_record() {
Ok(fabric_record) => Ok((
fabric_record,
fabric.fabric_id,
fabric.commissioner.node_id,
fabric.ipk,
fabric.commissioner.noc.clone(),
fabric.commissioner.operational_pkcs8.clone(),
crate::commission::next_device_node_id(fabric),
)),
Err(e) => Err(e),
},
Err(e) => Err(e),
};
let (
fabric_record,
fabric_id,
commissioner_node_id,
ipk_epoch_key,
commissioner_noc,
commissioner_pkcs8,
assigned_node_id,
) = match snapshot {
Ok(s) => s,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
let now = match current_matter_time() {
Ok(n) => n,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
let rng = self.rng.clone();
let tx = self.commission_tx.clone();
let multicast_if = self.multicast_if;
tokio::spawn(async move {
let result = run_commission_task(
setup_payload,
trust,
fabric_record,
commissioner_node_id,
ipk_epoch_key,
commissioner_noc,
commissioner_pkcs8,
assigned_node_id,
admin_vendor_id,
now,
rng,
multicast_if,
)
.await;
let _ = tx
.send(CommissionCompletion {
fabric_id,
result,
label,
reply,
})
.await;
});
}
#[cfg(feature = "ble")]
fn spawn_commission_ble(
&mut self,
setup_payload: matter_commissioning::SetupPayload,
network: matter_commissioning::NetworkCredentials,
label: Option<String>,
reply: oneshot::Sender<Result<crate::NodeInfo, Error>>,
) {
let Some(trust) = self.trust.clone() else {
let _ = reply.send(Err(Error::NoTrust));
return;
};
let admin_vendor_id = self.admin_vendor_id;
let snapshot = match self.sole_fabric() {
Ok(fabric) => match fabric.to_fabric_record() {
Ok(fabric_record) => Ok((
fabric_record,
fabric.fabric_id,
fabric.commissioner.node_id,
fabric.ipk,
fabric.commissioner.noc.clone(),
fabric.commissioner.operational_pkcs8.clone(),
crate::commission::next_device_node_id(fabric),
)),
Err(e) => Err(e),
},
Err(e) => Err(e),
};
let (
fabric_record,
fabric_id,
commissioner_node_id,
ipk_epoch_key,
commissioner_noc,
commissioner_pkcs8,
assigned_node_id,
) = match snapshot {
Ok(s) => s,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
let now = match current_matter_time() {
Ok(n) => n,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
let rng = self.rng.clone();
let tx = self.commission_tx.clone();
tokio::spawn(async move {
let result = crate::ble_commission::run_commission_ble_task(
setup_payload,
trust,
fabric_record,
commissioner_node_id,
ipk_epoch_key,
commissioner_noc,
commissioner_pkcs8,
assigned_node_id,
admin_vendor_id,
now,
rng,
network,
)
.await;
let _ = tx
.send(CommissionCompletion {
fabric_id,
result,
label,
reply,
})
.await;
});
}
async fn handle_commission_completion(&mut self, completion: CommissionCompletion) {
let CommissionCompletion {
fabric_id,
result,
label,
reply,
} = completion;
let outcome = match result {
Ok(commissioned) => {
let mut device = crate::commission::device_entry_from_commissioned(&commissioned);
device.label = label;
let info = crate::NodeInfo {
node_id: device.node_id,
fabric_id,
vendor_id: None,
product_id: None,
label: device.label.clone(),
};
if let Some(fabric) = self
.state
.fabrics
.iter_mut()
.find(|f| f.fabric_id == fabric_id)
{
fabric.devices.push(device);
}
match self.durable_save_inputs() {
Ok(job) => save_offloaded(job).await.map(|()| info),
Err(e) => Err(e),
}
}
Err(e) => Err(e),
};
let _ = reply.send(outcome);
}
async fn handle_set_node_vid_pid(
&mut self,
node_id: u64,
vendor_id: Option<u16>,
product_id: Option<u16>,
) -> Result<(), Error> {
let mut changed = false;
for fabric in &mut self.state.fabrics {
if let Some(device) = fabric.devices.iter_mut().find(|d| d.node_id == node_id) {
if vendor_id.is_some() && device.vendor_id != vendor_id {
device.vendor_id = vendor_id;
changed = true;
}
if product_id.is_some() && device.product_id != product_id {
device.product_id = product_id;
changed = true;
}
break;
}
}
if !changed {
return Ok(());
}
let job = self.durable_save_inputs()?;
save_offloaded(job).await
}
fn durable_save_inputs(&mut self) -> Result<SaveJob, Error> {
let bytes = snapshot::serialize(&self.state)?;
Ok(self.save_job(bytes))
}
fn save_job(&mut self, bytes: Vec<u8>) -> SaveJob {
self.snapshot_seq += 1;
SaveJob {
store: self.store.clone(),
bytes,
seq: self.snapshot_seq,
gate: self.save_gate.clone(),
}
}
fn persist_best_effort(&mut self) {
let Ok(bytes) = snapshot::serialize(&self.state) else {
return;
};
let job = self.save_job(bytes);
drop(tokio::task::spawn_blocking(move || {
let _ = job.run();
}));
}
fn sole_fabric(&self) -> Result<&crate::state::FabricEntry, Error> {
match self.state.fabrics.as_slice() {
[one] => Ok(one),
[] => Err(Error::NotCommissioned("no fabric created yet".into())),
_ => Err(Error::NotCommissioned(
"multiple fabrics; fabric(id).node(id) addressing is not in M8.2".into(),
)),
}
}
fn sole_fabric_mut(&mut self) -> Result<&mut crate::state::FabricEntry, Error> {
match self.state.fabrics.as_mut_slice() {
[one] => Ok(one),
[] => Err(Error::NotCommissioned("no fabric created yet".into())),
_ => Err(Error::NotCommissioned(
"multiple fabrics; fabric(id).node(id) addressing is not in M8.2".into(),
)),
}
}
async fn handle_create_group(
&mut self,
key_set_id: u16,
epoch_start_time: u64,
) -> Result<crate::GroupKeySet, Error> {
let mut epoch_key = [0u8; 16];
matter_crypto::random_bytes(&mut epoch_key)
.map_err(|e| Error::Operational(format!("group epoch-key generation failed: {e}")))?;
let fabric = self.sole_fabric_mut()?;
fabric.group_keys.retain(|k| k.key_set_id != key_set_id);
fabric.group_keys.push(crate::state::GroupKeySetConfig::new(
key_set_id,
epoch_key,
epoch_start_time,
));
let job = self.durable_save_inputs()?;
save_offloaded(job).await?;
Ok(crate::GroupKeySet::new(
key_set_id,
epoch_key.to_vec(),
epoch_start_time,
))
}
fn group_keys_for(
&mut self,
fabric_id: u64,
root_public_key: &[u8; 65],
epoch_key: [u8; 16],
) -> Result<&GroupKeyCacheEntry, Error> {
use std::collections::hash_map::Entry;
match self.group_key_cache.entry(fabric_id) {
Entry::Occupied(e)
if e.get().epoch_key == epoch_key
&& &e.get().root_public_key == root_public_key =>
{
Ok(e.into_mut())
}
slot => {
let compressed_fabric_id =
matter_crypto::derive_compressed_fabric_id(root_public_key, fabric_id)
.map_err(|e| Error::Operational(format!("compressed fabric id: {e}")))?;
let op_group_key =
matter_crypto::derive_operational_ipk(&epoch_key, &compressed_fabric_id)
.map_err(|e| Error::Operational(format!("operational group key: {e}")))?;
let group_session_id = matter_crypto::derive_group_session_id(&op_group_key)
.map_err(|e| Error::Operational(format!("group session id: {e}")))?;
let privacy_key = matter_crypto::derive_group_privacy_key(&op_group_key)
.map_err(|e| Error::Operational(format!("group privacy key: {e}")))?;
let fresh = GroupKeyCacheEntry {
epoch_key,
root_public_key: *root_public_key,
op_group_key,
group_session_id,
privacy_key,
};
Ok(match slot {
Entry::Occupied(mut e) => {
e.insert(fresh);
e.into_mut()
}
Entry::Vacant(e) => e.insert(fresh),
})
}
}
}
async fn handle_invoke_group(
&mut self,
group_id: u16,
key_set_id: u16,
path: matter_interaction::CommandPath,
fields_tlv: &[u8],
) -> Result<(), Error> {
let (fabric_id, source_node_id, root_public_key, epoch_key, ceiling) = {
let fabric = self.sole_fabric()?;
let epoch_key = fabric
.group_keys
.iter()
.rfind(|k| k.key_set_id == key_set_id)
.map(|k| k.epoch_key)
.ok_or(Error::GroupNotProvisioned(key_set_id))?;
let root_public_key = *fabric.rcac_cert.public_key().as_bytes();
(
fabric.fabric_id,
fabric.commissioner.node_id,
root_public_key,
epoch_key,
fabric.outbound_group_counter,
)
};
let entry = self.group_keys_for(fabric_id, &root_public_key, epoch_key)?;
let op_group_key = entry.op_group_key;
let group_session_id = entry.group_session_id;
let privacy_key = entry.privacy_key;
let mcast = matter_crypto::group_multicast_ipv6(fabric_id, group_id);
let next = *self.group_counters.entry(fabric_id).or_insert(ceiling);
if next >= ceiling {
let new_ceiling = next
.checked_add(GROUP_COUNTER_BLOCK)
.or_else(|| (next < u32::MAX).then_some(u32::MAX))
.ok_or_else(|| {
Error::Operational("group counter exhausted — re-key the group".into())
})?;
self.sole_fabric_mut()?.outbound_group_counter = new_ceiling;
let saved = match self.durable_save_inputs() {
Ok(job) => save_offloaded(job).await,
Err(e) => Err(e),
};
if let Err(e) = saved {
debug_assert!(
self.sole_fabric().is_ok(),
"the fabric that was just reserved against must still exist"
);
if let Ok(fabric) = self.sole_fabric_mut() {
fabric.outbound_group_counter = ceiling;
}
return Err(e);
}
}
self.group_counters
.insert(fabric_id, next.saturating_add(1));
let counter = next;
let payload = matter_interaction::build_invoke_request_group(path, fields_tlv);
let mut eid = [0u8; 2];
matter_crypto::random_bytes(&mut eid)
.map_err(|e| Error::Operational(format!("exchange-id generation failed: {e}")))?;
let protocol_header = ProtocolHeader {
exchange_flags: matter_transport::ExchangeFlags::INITIATOR,
opcode: crate::node::OP_INVOKE_REQUEST,
exchange_id: u16::from_le_bytes(eid),
protocol_id: ProtocolId::INTERACTION_MODEL,
ack_counter: None,
};
let wire = matter_transport::encode_group_secured_with_privacy_key(
&op_group_key,
&privacy_key,
group_session_id,
source_node_id,
group_id,
counter,
&protocol_header,
&payload,
)?;
let scope_id = self.multicast_if.filter(|&i| i != 0).unwrap_or_else(|| {
std::env::var("MATTER_MULTICAST_IF")
.ok()
.and_then(|s| s.parse::<u32>().ok())
.unwrap_or(0)
});
let dest = SocketAddr::V6(std::net::SocketAddrV6::new(
mcast,
MATTER_GROUP_PORT,
0,
scope_id,
));
self.transport
.send_to(&wire, dest)
.await
.map_err(|e| Error::Operational(format!("group send: {e}")))?;
Ok(())
}
fn enqueue_connect_waiter(&mut self, fabric_id: u64, node_id: u64, waiter: ConnectWaiter) {
let already_connecting = self.pending_connects.contains_key(&node_id);
self.pending_connects
.entry(node_id)
.or_default()
.push(waiter);
if !already_connecting {
self.spawn_connect(fabric_id, node_id);
}
}
fn spawn_connect(&mut self, fabric_id: u64, node_id: u64) {
let compressed = self.sole_fabric().and_then(|fabric| {
matter_crypto::derive_compressed_fabric_id(
fabric.rcac_cert.public_key().as_bytes(),
fabric.fabric_id,
)
.map_err(|e| Error::Operational(e.to_string()))
});
let compressed = match compressed {
Ok(c) => c,
Err(e) => {
self.fail_connect_waiters(node_id, &e);
return;
}
};
let target = matter_commissioning::driver::operational_instance_name(compressed, node_id);
if self.resolve_query.is_none() {
match self.discovery.query(ServiceKind::Operational) {
Ok(h) => self.resolve_query = Some(h),
Err(e) => {
let err = Error::from(matter_commissioning::driver::DriverError::Transport(e));
self.fail_connect_waiters(node_id, &err);
return;
}
}
}
self.park_resolve(fabric_id, node_id, target);
self.drive_pending_resolves();
}
fn park_resolve(&mut self, fabric_id: u64, node_id: u64, target: String) {
self.pending_resolves.push(PendingResolve {
fabric_id,
node_id,
target,
deadline: Instant::now() + RESOLVE_DEADLINE,
});
}
fn record_key(instance_name: &str) -> String {
instance_name.to_ascii_lowercase()
}
fn record_seen(&mut self, services: &[matter_transport::MatterService], now: Instant) {
self.seen_records
.retain(|_, r| now.saturating_duration_since(r.seen) < SEEN_RECORD_TTL);
for svc in services {
let Some(addr) = matter_commissioning::driver::preferred_address(&svc.addresses) else {
continue;
};
let key = Self::record_key(&svc.instance_name);
if self.seen_records.len() >= SEEN_RECORD_CAP && !self.seen_records.contains_key(&key) {
let oldest = self
.seen_records
.iter()
.min_by_key(|(_, r)| r.seen)
.map(|(k, _)| k.clone());
if let Some(k) = oldest {
self.seen_records.remove(&k);
}
}
self.seen_records.insert(
key,
SeenRecord {
peer: SocketAddr::new(addr, svc.port),
peer_mrp: svc.peer_mrp_config(),
seen: now,
},
);
}
}
fn shutdown_discovery(&mut self) {
self.pending_resolves.clear();
self.release_resolve_query_if_idle();
}
fn release_resolve_query_if_idle(&mut self) {
if self.pending_resolves.is_empty() {
if let Some(handle) = self.resolve_query.take() {
self.discovery.stop_query(handle);
}
}
}
fn cancel_pending_resolve(&mut self, node_id: u64) {
self.pending_resolves.retain(|pr| pr.node_id != node_id);
self.release_resolve_query_if_idle();
}
fn drive_pending_resolves(&mut self) {
self.next_resolve_poll = Instant::now() + LIVENESS_TICK;
let Some(handle) = self.resolve_query else {
return;
};
if self.pending_resolves.is_empty() {
return;
}
let services = self.discovery.poll_results(handle);
let now = Instant::now();
self.record_seen(&services, now);
let mut resolved: Vec<(u64, u64, SocketAddr, matter_transport::MrpConfig)> = Vec::new();
let mut expired: Vec<(u64, String)> = Vec::new();
let mut still_parked = Vec::with_capacity(self.pending_resolves.len());
for pr in std::mem::take(&mut self.pending_resolves) {
if let Some(rec) = self.seen_records.get(&Self::record_key(&pr.target)) {
resolved.push((pr.fabric_id, pr.node_id, rec.peer, rec.peer_mrp));
} else if now >= pr.deadline {
expired.push((pr.node_id, pr.target));
} else {
still_parked.push(pr);
}
}
self.pending_resolves = still_parked;
self.release_resolve_query_if_idle();
for (node_id, target) in expired {
let err = Error::from(matter_commissioning::driver::DriverError::Discovery(
format!("operational node {target} not found via mDNS"),
));
self.fail_connect_waiters(node_id, &err);
}
for (fabric_id, node_id, peer, peer_mrp) in resolved {
self.finish_spawn_connect(fabric_id, node_id, peer, peer_mrp);
}
}
fn finish_spawn_connect(
&mut self,
fabric_id: u64,
node_id: u64,
peer: SocketAddr,
peer_mrp: matter_transport::MrpConfig,
) {
let creds = match self.sole_fabric() {
Ok(fabric) => crate::credentials::operational_credentials(fabric),
Err(e) => Err(e),
};
let (credentials, roots, _compressed) = match creds {
Ok(c) => c,
Err(e) => {
self.fail_connect_waiters(node_id, &e);
return;
}
};
let now = match current_matter_time() {
Ok(n) => n,
Err(e) => {
self.fail_connect_waiters(node_id, &e);
return;
}
};
self.connect_mrp.insert(node_id, peer_mrp);
let local_session_id = self.sessions.allocate_session_id().0;
let (inbound_tx, inbound_rx) = mpsc::channel(16);
self.connect_inbound.insert(node_id, inbound_tx);
let outbound_tx = self.connect_outbound_tx.clone();
let done_tx = self.connect_done_tx.clone();
tokio::spawn(run_connect_task(
node_id,
fabric_id,
local_session_id,
credentials,
roots,
now,
peer,
inbound_rx,
outbound_tx,
done_tx,
));
}
async fn handle_connect_outbound(&mut self, out: crate::handshake_socket::HandshakeOutbound) {
let crate::handshake_socket::HandshakeOutbound {
node_id,
bytes,
peer,
} = out;
if self.connect_inbound.contains_key(&node_id) {
self.connect_routes.insert(route_key(peer), node_id);
}
let _ = self.transport.send_to(&bytes, peer).await;
}
async fn handle_connect_done(&mut self, done: ConnectCompletion) {
let ConnectCompletion { node_id, result } = done;
self.connect_inbound.remove(&node_id);
self.connect_routes.retain(|_, n| *n != node_id);
let (output, peer) = match result {
Ok(ok) => ok,
Err(e) => {
self.fail_connect_waiters(node_id, &e);
return;
}
};
let fabric_id = match self.sole_fabric() {
Ok(fabric) => fabric.fabric_id,
Err(e) => {
self.fail_connect_waiters(node_id, &e);
return;
}
};
let old_session = self.cache.get(&(fabric_id, node_id)).map(|c| c.session_id);
if let Some(old) = old_session {
self.sessions.remove(old);
}
let record_bytes = output
.resumption_record
.as_ref()
.and_then(|r| crate::resumption::serialize_record(r).ok());
let peer_mrp = self.connect_mrp.remove(&node_id).unwrap_or_default();
let sid = self
.sessions
.register_case_with_mrp(&output, SessionRole::Initiator, peer_mrp);
if let Some(s) = self.sessions.get_mut(sid) {
s.peer_addr = Some(peer);
}
self.upsert_device(fabric_id, node_id, peer, record_bytes);
self.cache.insert(
(fabric_id, node_id),
CachedSession {
session_id: sid,
peer,
},
);
if let Some(old) = old_session {
self.resubscribe_stranded(old);
}
if let Some(waiters) = self.pending_connects.remove(&node_id) {
for waiter in waiters {
match waiter {
ConnectWaiter::Command(cmd) => self.dispatch_ready(cmd).await,
ConnectWaiter::ResendPending(p) => {
self.resume_resend_pending(p, sid, peer).await;
}
ConnectWaiter::Resubscribe(pr) => {
self.resume_resubscribe(pr, sid, peer).await;
}
}
}
}
}
fn fail_connect_waiters(&mut self, node_id: u64, err: &Error) {
self.connect_inbound.remove(&node_id);
self.connect_routes.retain(|_, n| *n != node_id);
self.connect_mrp.remove(&node_id);
self.cancel_pending_resolve(node_id);
if let Some(waiters) = self.pending_connects.remove(&node_id) {
let msg = err.to_string();
let fail_err =
|| Error::Operational(format!("connect to node {node_id} failed: {msg}"));
for waiter in waiters {
match waiter {
ConnectWaiter::Command(cmd) => fail_command(cmd, fail_err()),
ConnectWaiter::ResendPending(p) => Self::fail_pending(p, fail_err()),
ConnectWaiter::Resubscribe(pr) => self.reschedule_resubscribe(pr),
}
}
}
}
async fn resume_resend_pending(&mut self, p: Pending, sid: SessionId, peer: SocketAddr) {
let sent = self
.send_request(
sid,
peer,
p.request.opcode,
p.request.protocol_id,
&p.request.payload,
)
.await;
match sent {
Ok(exchange) => {
let mut np = p;
np.peer = peer;
np.retried = true;
match &mut np.reply {
PendingReply::Read {
chunks,
total_bytes,
..
} => {
chunks.clear();
*total_bytes = 0;
}
PendingReply::Subscribe { priming, .. } => {
**priming = ReportReassembler::default();
}
PendingReply::RoundTrip(_)
| PendingReply::TimedAction { .. }
| PendingReply::Action { .. }
| PendingReply::ChunkedWrite { .. } => {}
}
self.pending.insert((sid, exchange), np);
}
Err(e) => Self::fail_pending(p, e),
}
}
async fn resume_resubscribe(
&mut self,
pr: PendingResubscribe,
sid: SessionId,
peer: SocketAddr,
) {
let req =
matter_interaction::build_subscribe_request(&matter_interaction::SubscribeRequest {
keep_subscriptions: false,
min_interval_floor: pr.min_interval,
max_interval_ceiling: pr.max_interval,
paths: pr.paths.clone(),
event_paths: pr.event_paths.clone(),
event_filters: pr.event_filters.clone(),
});
match self
.send_request(
sid,
peer,
OP_SUBSCRIBE_REQUEST,
ProtocolId::INTERACTION_MODEL,
&req,
)
.await
{
Ok(exchange) => {
self.pending.insert(
(sid, exchange),
Pending {
node_id: pr.node_id,
peer,
request: PendingRequest {
opcode: OP_SUBSCRIBE_REQUEST,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload: req,
},
retried: true,
reply: PendingReply::Subscribe {
sub_id: pr.sub_id,
reply: None,
report_tx: pr.tx,
report_rx: None,
priming: Box::new(ReportReassembler::default()),
node_id: pr.node_id,
paths: pr.paths,
event_paths: pr.event_paths,
event_filters: pr.event_filters,
min_interval: pr.min_interval,
max_interval: pr.max_interval,
retry_count: pr.retry_count,
},
},
);
}
Err(_) => self.reschedule_resubscribe(pr),
}
}
async fn connect(&mut self, node_id: u64) -> Result<(SessionId, std::net::SocketAddr), Error> {
let fabric_id = self.sole_fabric()?.fabric_id;
let (credentials, roots, compressed) =
crate::credentials::operational_credentials(self.sole_fabric()?)?;
let peer = matter_commissioning::driver::resolve_operational(
&mut self.discovery,
compressed,
node_id,
)
.await?;
let now = current_matter_time()?;
let sid = matter_commissioning::driver::run_case(
&self.transport,
&mut self.sessions,
peer,
credentials,
roots,
node_id,
fabric_id,
now,
)
.await?;
if let Some(s) = self.sessions.get_mut(sid) {
s.peer_addr = Some(peer);
}
let old_session = self.cache.get(&(fabric_id, node_id)).map(|c| c.session_id);
if let Some(old) = old_session {
self.sessions.remove(old);
}
self.upsert_device(fabric_id, node_id, peer, None);
self.cache.insert(
(fabric_id, node_id),
CachedSession {
session_id: sid,
peer,
},
);
if let Some(old) = old_session {
self.resubscribe_stranded(old);
}
Ok((sid, peer))
}
fn resubscribe_stranded(&mut self, old_session: SessionId) {
let stranded: Vec<SubId> = self
.subscriptions
.iter()
.filter(|(_, e)| e.session_id == old_session)
.map(|(id, _)| *id)
.collect();
for id in stranded {
self.begin_resubscribe(id, Error::Operational("session replaced".into()));
}
}
fn upsert_device(
&mut self,
fabric_id: u64,
node_id: u64,
peer: std::net::SocketAddr,
resumption_record: Option<Vec<u8>>,
) {
let addr = peer.to_string();
let mut changed = false;
if let Some(fabric) = self
.state
.fabrics
.iter_mut()
.find(|f| f.fabric_id == fabric_id)
{
if let Some(dev) = fabric.devices.iter_mut().find(|d| d.node_id == node_id) {
if dev.last_known_addr.as_deref() != Some(addr.as_str()) {
dev.last_known_addr = Some(addr);
changed = true;
}
if let Some(rr) = resumption_record {
dev.resumption_record = Some(rr);
changed = true;
}
}
}
if changed {
self.persist_best_effort();
}
}
fn handle_store_resumption_record(
&mut self,
node_id: u64,
record_bytes: Vec<u8>,
) -> Result<(), Error> {
let fabric = self.sole_fabric()?;
let fabric_id = fabric.fabric_id;
let Some(dev) = self
.state
.fabrics
.iter_mut()
.find(|f| f.fabric_id == fabric_id)
.and_then(|f| f.devices.iter_mut().find(|d| d.node_id == node_id))
else {
return Err(Error::Operational(format!(
"no device entry for node {node_id:#x} to store a resumption record on"
)));
};
dev.resumption_record = Some(record_bytes);
self.persist_best_effort();
Ok(())
}
async fn session_for(&mut self, node_id: u64) -> Result<(SessionId, SocketAddr), Error> {
let fabric_id = self.sole_fabric()?.fabric_id;
if let Some((sid, peer)) = self
.cache
.get(&(fabric_id, node_id))
.map(|c| (c.session_id, c.peer))
{
return Ok((sid, peer));
}
self.connect(node_id).await
}
async fn send_request(
&mut self,
sid: SessionId,
peer: SocketAddr,
opcode: u8,
protocol_id: ProtocolId,
payload: &[u8],
) -> Result<u16, Error> {
let out = self.sessions.encode_outbound(
sid,
None,
opcode,
protocol_id,
payload,
MrpFlags { reliable: true },
Instant::now(),
)?;
let exchange = out.exchange_id;
self.transport
.send_to(&out.wire_bytes, peer)
.await
.map_err(|e| Error::Operational(format!("request send: {e}")))?;
Ok(exchange)
}
#[cfg(test)]
async fn start_round_trip(
&mut self,
node_id: u64,
opcode: u8,
protocol_id: ProtocolId,
payload: Vec<u8>,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
) {
let (sid, peer) = match self.session_for(node_id).await {
Ok(v) => v,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
match self
.send_request(sid, peer, opcode, protocol_id, &payload)
.await
{
Ok(exchange) => {
self.pending.insert(
(sid, exchange),
Pending {
node_id,
peer,
request: PendingRequest {
opcode,
protocol_id,
payload,
},
retried: false,
reply: PendingReply::RoundTrip(reply),
},
);
}
Err(e) => {
let _ = reply.send(Err(e));
}
}
}
#[allow(clippy::too_many_arguments)] async fn begin_timed(
&mut self,
sid: SessionId,
peer: SocketAddr,
node_id: u64,
timeout_ms: u16,
action_opcode: u8,
action_payload: Vec<u8>,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
) {
let req = matter_interaction::build_timed_request(timeout_ms);
match self
.send_request(
sid,
peer,
OP_TIMED_REQUEST,
ProtocolId::INTERACTION_MODEL,
&req,
)
.await
{
Ok(exchange) => {
self.pending.insert(
(sid, exchange),
Pending {
node_id,
peer,
request: PendingRequest {
opcode: OP_TIMED_REQUEST,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload: req,
},
retried: false,
reply: PendingReply::TimedAction {
action_opcode,
action_payload,
reply,
},
},
);
}
Err(e) => {
let _ = reply.send(Err(e));
}
}
}
async fn start_timed_round_trip(
&mut self,
node_id: u64,
timeout_ms: u16,
action_opcode: u8,
action_payload: Vec<u8>,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
) {
let (sid, peer) = match self.session_for(node_id).await {
Ok(v) => v,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
self.begin_timed(
sid,
peer,
node_id,
timeout_ms,
action_opcode,
action_payload,
reply,
)
.await;
}
#[allow(clippy::too_many_arguments)] async fn handle_action(
&mut self,
node_id: u64,
opcode: u8,
plain_payload: Vec<u8>,
timed_payload: TimedPayload,
keys: Vec<(u32, u32)>,
timeout_ms: u16,
reply: oneshot::Sender<Result<Vec<u8>, Error>>,
) {
let (sid, peer) = match self.session_for(node_id).await {
Ok(v) => v,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
if keys.iter().any(|k| self.timed_paths.contains(k)) {
self.begin_timed(
sid,
peer,
node_id,
timeout_ms,
opcode,
timed_payload(),
reply,
)
.await;
return;
}
match self
.send_request(
sid,
peer,
opcode,
ProtocolId::INTERACTION_MODEL,
&plain_payload,
)
.await
{
Ok(exchange) => {
self.pending.insert(
(sid, exchange),
Pending {
node_id,
peer,
request: PendingRequest {
opcode,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload: plain_payload,
},
retried: false,
reply: PendingReply::Action {
opcode,
timed_payload,
keys,
timeout_ms,
node_id,
reply,
},
},
);
}
Err(e) => {
let _ = reply.send(Err(e));
}
}
}
async fn resolve_action(&mut self, sid: SessionId, exchange: u16, payload: Vec<u8>) {
let Some(p) = self.pending.remove(&(sid, exchange)) else {
return;
};
let PendingReply::Action {
opcode,
timed_payload,
keys,
timeout_ms,
node_id,
reply,
} = p.reply
else {
return;
};
if !response_needs_timed(opcode, &payload) {
let _ = reply.send(Ok(payload));
return;
}
for k in keys {
self.timed_paths.insert(k);
}
self.begin_timed(
sid,
p.peer,
node_id,
timeout_ms,
opcode,
timed_payload(),
reply,
)
.await;
}
async fn start_read(
&mut self,
node_id: u64,
payload: Vec<u8>,
reply: oneshot::Sender<Result<Vec<matter_interaction::ReportData>, Error>>,
) {
let (sid, peer) = match self.session_for(node_id).await {
Ok(v) => v,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
let opcode = crate::node::OP_READ_REQUEST;
match self
.send_request(sid, peer, opcode, ProtocolId::INTERACTION_MODEL, &payload)
.await
{
Ok(exchange) => {
self.pending.insert(
(sid, exchange),
Pending {
node_id,
peer,
request: PendingRequest {
opcode,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload,
},
retried: false,
reply: PendingReply::Read {
reply,
chunks: Vec::new(),
total_bytes: 0,
},
},
);
}
Err(e) => {
let _ = reply.send(Err(e));
}
}
}
async fn handle_chunked_write(
&mut self,
node_id: u64,
chunks: Vec<Vec<u8>>,
reply: oneshot::Sender<
Result<
Vec<(
matter_interaction::AttributePath,
matter_interaction::ImStatus,
)>,
Error,
>,
>,
) {
if chunks.is_empty() {
let _ = reply.send(Err(Error::Operational(
"chunked_write requires at least one chunk".into(),
)));
return;
}
let (sid, peer) = match self.session_for(node_id).await {
Ok(v) => v,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
let exchange = match self
.send_request(
sid,
peer,
OP_WRITE_REQUEST,
ProtocolId::INTERACTION_MODEL,
&chunks[0],
)
.await
{
Ok(ex) => ex,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
let mut remaining: VecDeque<Vec<u8>> = chunks.into();
let first = remaining.pop_front().unwrap_or_default();
self.pending.insert(
(sid, exchange),
Pending {
node_id,
peer,
request: PendingRequest {
opcode: OP_WRITE_REQUEST,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload: first,
},
retried: true,
reply: PendingReply::ChunkedWrite {
reply,
remaining,
statuses: Vec::new(),
},
},
);
}
async fn handle_inbound(&mut self, packet: &[u8], from: SocketAddr) {
if packet.len() >= 3 && packet[1] == 0 && packet[2] == 0 {
if let Some(&node_id) = self.connect_routes.get(&route_key(from)) {
if let Some(tx) = self.connect_inbound.get(&node_id) {
let _ = tx.send((packet.to_vec(), from)).await;
}
}
return;
}
let Ok(decoded) = self.sessions.decode_inbound(packet, Instant::now()) else {
return;
};
match decoded {
DecodeInboundOutput::AppMessage {
session_id,
exchange_id,
opcode,
payload,
..
} => {
if self.pending.contains_key(&(session_id, exchange_id)) {
self.resolve_pending(session_id, exchange_id, opcode, payload)
.await;
} else if opcode == OP_REPORT_DATA {
self.deliver_report(session_id, exchange_id, &payload).await;
}
}
DecodeInboundOutput::DuplicateReliableAckResent { ack_packet, .. } => {
let _ = self.transport.send_to(&ack_packet, from).await;
}
_ => {}
}
}
async fn resolve_pending(
&mut self,
session_id: SessionId,
exchange_id: u16,
opcode: u8,
payload: Vec<u8>,
) {
enum Kind {
RoundTrip,
ChunkedWrite,
Read,
Subscribe,
Timed,
Action,
}
let key = (session_id, exchange_id);
let kind = match self.pending.get(&key) {
Some(p) => match &p.reply {
PendingReply::RoundTrip(_) => Kind::RoundTrip,
PendingReply::ChunkedWrite { .. } => Kind::ChunkedWrite,
PendingReply::Read { .. } => Kind::Read,
PendingReply::Subscribe { .. } => Kind::Subscribe,
PendingReply::TimedAction { .. } => Kind::Timed,
PendingReply::Action { .. } => Kind::Action,
},
None => return,
};
match kind {
Kind::RoundTrip => {
if let Some(PendingReply::RoundTrip(reply)) =
self.pending.remove(&key).map(|p| p.reply)
{
let _ = reply.send(Ok(payload));
}
}
Kind::ChunkedWrite => {
self.resolve_chunked_write(session_id, exchange_id, opcode, payload)
.await;
}
Kind::Read => {
let peer = self.pending.get(&key).map(|p| p.peer);
let chunk_len = payload.len();
let rd = match matter_interaction::parse_report_data(&payload) {
Ok(rd) => rd,
Err(e) => {
if let Some(PendingReply::Read { reply, .. }) =
self.pending.remove(&key).map(|p| p.reply)
{
let _ = reply.send(Err(Error::InteractionModel(e)));
}
return;
}
};
let more = rd.more_chunked_messages;
let over = match self.pending.get_mut(&key).map(|p| &mut p.reply) {
Some(PendingReply::Read {
chunks,
total_bytes,
..
}) => {
*total_bytes = total_bytes.saturating_add(chunk_len);
chunks.push(rd);
chunks.len() > MAX_READ_CHUNKS || *total_bytes > MAX_READ_BYTES
}
_ => return,
};
if over {
if let Some(PendingReply::Read { reply, .. }) =
self.pending.remove(&key).map(|p| p.reply)
{
let _ = reply.send(Err(Error::Operational("read too large".into())));
}
} else if more {
if let Some(peer) = peer {
let _ = self.send_chunk_ack(session_id, exchange_id, peer).await;
}
} else if let Some(PendingReply::Read { reply, chunks, .. }) =
self.pending.remove(&key).map(|p| p.reply)
{
let _ = reply.send(Ok(chunks));
}
}
Kind::Subscribe => {
self.resolve_subscribe(session_id, exchange_id, opcode, payload)
.await;
}
Kind::Timed => {
self.resolve_timed(session_id, exchange_id, payload).await;
}
Kind::Action => {
self.resolve_action(session_id, exchange_id, payload).await;
}
}
}
async fn resolve_timed(&mut self, sid: SessionId, exchange: u16, payload: Vec<u8>) {
let success = matches!(
matter_interaction::parse_status_response(&payload),
Ok(Some(0))
);
let Some(p) = self.pending.remove(&(sid, exchange)) else {
return;
};
let PendingReply::TimedAction {
action_opcode,
action_payload,
reply,
} = p.reply
else {
return;
};
if !success {
let _ = reply.send(Ok(payload));
return;
}
if let Err(e) = self
.send_on_exchange(sid, exchange, p.peer, action_opcode, &action_payload)
.await
{
let _ = reply.send(Err(e));
return;
}
self.pending.insert(
(sid, exchange),
Pending {
node_id: p.node_id,
peer: p.peer,
request: PendingRequest {
opcode: action_opcode,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload: action_payload,
},
retried: true, reply: PendingReply::RoundTrip(reply),
},
);
}
async fn resolve_chunked_write(
&mut self,
sid: SessionId,
exchange: u16,
opcode: u8,
payload: Vec<u8>,
) {
let Some(p) = self.pending.remove(&(sid, exchange)) else {
return;
};
let PendingReply::ChunkedWrite {
reply,
mut remaining,
mut statuses,
} = p.reply
else {
return;
};
if opcode != OP_WRITE_RESPONSE {
let err = if opcode == OP_STATUS_RESPONSE {
match matter_interaction::parse_status_response(&payload) {
Ok(Some(status)) => Error::Operational(format!(
"chunked write rejected by device: IM status 0x{status:02x}"
)),
_ => Error::Operational(
"chunked write rejected by device: malformed StatusResponse".into(),
),
}
} else {
Error::Operational(format!(
"chunked write: unexpected response opcode 0x{opcode:02x} (expected WriteResponse)"
))
};
let _ = reply.send(Err(err));
return;
}
let chunk_statuses = match matter_interaction::parse_write_response(&payload) {
Ok(s) => s,
Err(e) => {
let _ = reply.send(Err(Error::InteractionModel(e)));
return;
}
};
statuses.extend(chunk_statuses);
let Some(next) = remaining.pop_front() else {
let _ = reply.send(Ok(statuses));
return;
};
if let Err(e) = self
.send_on_exchange(sid, exchange, p.peer, OP_WRITE_REQUEST, &next)
.await
{
let _ = reply.send(Err(e));
return;
}
self.pending.insert(
(sid, exchange),
Pending {
node_id: p.node_id,
peer: p.peer,
request: PendingRequest {
opcode: OP_WRITE_REQUEST,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload: next,
},
retried: true, reply: PendingReply::ChunkedWrite {
reply,
remaining,
statuses,
},
},
);
}
async fn send_on_exchange(
&mut self,
sid: SessionId,
exchange: u16,
peer: SocketAddr,
opcode: u8,
payload: &[u8],
) -> Result<(), Error> {
let out = self.sessions.encode_outbound(
sid,
Some(exchange),
opcode,
ProtocolId::INTERACTION_MODEL,
payload,
MrpFlags { reliable: true },
Instant::now(),
)?;
self.transport
.send_to(&out.wire_bytes, peer)
.await
.map_err(|e| Error::Operational(format!("timed action send: {e}")))?;
Ok(())
}
async fn send_chunk_ack(
&mut self,
sid: SessionId,
exchange: u16,
peer: SocketAddr,
) -> Result<(), Error> {
let status = matter_interaction::build_status_response(0);
let out = self.sessions.encode_outbound(
sid,
Some(exchange),
OP_STATUS_RESPONSE,
ProtocolId::INTERACTION_MODEL,
&status,
MrpFlags { reliable: true },
Instant::now(),
)?;
self.transport
.send_to(&out.wire_bytes, peer)
.await
.map_err(|e| Error::Operational(format!("chunk ack send: {e}")))?;
Ok(())
}
async fn deliver_report(&mut self, session_id: SessionId, exchange_id: u16, payload: &[u8]) {
let Ok(mut rd) = matter_interaction::parse_report_data(payload) else {
return;
};
let Some(wire_sub_id) = rd.subscription_id else {
return; };
let now = Instant::now();
let Some(&sub_id) = self.sub_index.get(&(session_id, wire_sub_id)) else {
return;
};
let Some(entry) = self.subscriptions.get_mut(&sub_id) else {
debug_assert!(false, "sub_index points at a missing subscription");
return;
};
entry.liveness_deadline =
now + std::time::Duration::from_secs(u64::from(entry.max_interval)) + LIVENESS_GRACE;
let peer = entry.peer;
let mut consumer_gone = false;
for ev in std::mem::take(&mut rd.events) {
if !entry.tx.try_send_event(ev) {
consumer_gone = true;
break;
}
}
if !consumer_gone {
if let Some(attrs) = entry.reassembler.push_parsed(rd) {
for (path, value) in attrs {
if !entry.tx.try_send_report(AttributeReport { path, value }) {
consumer_gone = true;
break;
}
}
}
}
if consumer_gone {
self.remove_subscription(sub_id);
return;
}
let _ = self.send_status_ack(session_id, exchange_id, peer).await;
}
#[allow(clippy::too_many_arguments)]
async fn start_subscribe(
&mut self,
node_id: u64,
paths: Vec<matter_interaction::ReadPath>,
event_paths: Vec<matter_interaction::EventPath>,
event_filters: Vec<matter_interaction::EventFilter>,
min_interval: u16,
max_interval: u16,
reply: oneshot::Sender<Result<SubEstablished, Error>>,
) {
let (sid, peer) = match self.session_for(node_id).await {
Ok(v) => v,
Err(e) => {
let _ = reply.send(Err(e));
return;
}
};
let req =
matter_interaction::build_subscribe_request(&matter_interaction::SubscribeRequest {
keep_subscriptions: false,
min_interval_floor: min_interval,
max_interval_ceiling: max_interval,
paths: paths.clone(),
event_paths: event_paths.clone(),
event_filters: event_filters.clone(),
});
match self
.send_request(
sid,
peer,
OP_SUBSCRIBE_REQUEST,
ProtocolId::INTERACTION_MODEL,
&req,
)
.await
{
Ok(exchange) => {
let sub_id = SubId(self.next_sub_id);
self.next_sub_id += 1;
let (report_tx, report_rx) =
mpsc::channel::<SubscriptionEvent>(SUBSCRIPTION_CHANNEL_CAP);
let (ctrl_tx, ctrl_rx) = mpsc::unbounded_channel::<SubscriptionEvent>();
let report_tx = ReportSink {
report_tx,
ctrl_tx,
dropped: 0,
};
let report_rx = SubReceivers { report_rx, ctrl_rx };
self.pending.insert(
(sid, exchange),
Pending {
node_id,
peer,
request: PendingRequest {
opcode: OP_SUBSCRIBE_REQUEST,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload: req,
},
retried: false,
reply: PendingReply::Subscribe {
sub_id,
reply: Some(reply),
report_tx,
report_rx: Some(report_rx),
priming: Box::new(ReportReassembler::default()),
node_id,
paths,
event_paths,
event_filters,
min_interval,
max_interval,
retry_count: 0,
},
},
);
}
Err(e) => {
let _ = reply.send(Err(e));
}
}
}
async fn resolve_subscribe(
&mut self,
session_id: SessionId,
exchange_id: u16,
opcode: u8,
payload: Vec<u8>,
) {
let key = (session_id, exchange_id);
if opcode == OP_REPORT_DATA {
if let Some(peer) = self.pending.get(&key).map(|p| p.peer) {
let _ = self.send_status_ack(session_id, exchange_id, peer).await;
}
if let Some(Pending {
reply:
PendingReply::Subscribe {
report_tx, priming, ..
},
..
}) = self.pending.get_mut(&key)
{
let Ok(mut rd) = matter_interaction::parse_report_data(&payload) else {
return;
};
for ev in std::mem::take(&mut rd.events) {
if !report_tx.try_send_event(ev) {
break;
}
}
if let Some(attrs) = priming.push_parsed(rd) {
for (path, value) in attrs {
if !report_tx.try_send_report(AttributeReport { path, value }) {
break;
}
}
}
}
} else if opcode == OP_SUBSCRIBE_RESPONSE {
let Some(p) = self.pending.remove(&key) else {
return;
};
let PendingReply::Subscribe {
sub_id,
reply,
report_tx,
report_rx,
node_id,
paths,
event_paths,
event_filters,
min_interval,
..
} = p.reply
else {
return;
};
match matter_interaction::parse_subscribe_response(&payload) {
Ok(resp) => {
let deadline = Instant::now()
+ std::time::Duration::from_secs(u64::from(resp.max_interval))
+ LIVENESS_GRACE;
if !report_tx.send_control(SubscriptionEvent::Established {
subscription_id: resp.subscription_id,
}) {
return;
}
self.insert_subscription(
sub_id,
SubEntry {
tx: report_tx,
peer: p.peer,
reassembler: ReportReassembler::default(),
session_id,
wire_sub_id: resp.subscription_id,
node_id,
paths,
event_paths,
event_filters,
min_interval,
max_interval: resp.max_interval,
liveness_deadline: deadline,
},
);
if let (Some(reply), Some(rx)) = (reply, report_rx) {
let _ = reply.send(Ok((rx, sub_id)));
}
}
Err(e) => {
if let Some(reply) = reply {
let _ = reply.send(Err(Error::InteractionModel(e)));
}
}
}
}
}
async fn send_status_ack(
&mut self,
sid: SessionId,
exchange: u16,
peer: SocketAddr,
) -> Result<(), Error> {
let status = matter_interaction::build_status_response(0);
let out = self.sessions.encode_outbound(
sid,
Some(exchange),
OP_STATUS_RESPONSE,
ProtocolId::INTERACTION_MODEL,
&status,
MrpFlags { reliable: false },
Instant::now(),
)?;
self.transport
.send_to(&out.wire_bytes, peer)
.await
.map_err(|e| Error::Operational(format!("status ack send: {e}")))?;
Ok(())
}
async fn drive_mrp(&mut self) {
for event in self.sessions.handle_timeout(Instant::now()) {
match event {
MrpEvent::Retransmit {
session_id, packet, ..
}
| MrpEvent::SendStandaloneAck {
session_id, packet, ..
} => {
if let Some(peer) = self.peer_for_session(session_id) {
let _ = self.transport.send_to(&packet, peer).await;
}
}
MrpEvent::Expired {
session_id,
exchange_id,
..
} => {
self.on_pending_timeout(session_id, exchange_id).await;
}
_ => {}
}
}
}
async fn on_pending_timeout(&mut self, session_id: SessionId, exchange_id: u16) {
let Some(p) = self.pending.remove(&(session_id, exchange_id)) else {
return;
};
if matches!(&p.reply, PendingReply::Subscribe { reply: None, .. }) {
if let PendingReply::Subscribe {
sub_id,
report_tx,
node_id,
paths,
event_paths,
event_filters,
min_interval,
max_interval,
retry_count,
..
} = p.reply
{
if let Ok(fabric_id) = self.sole_fabric().map(|f| f.fabric_id) {
if self
.cache
.get(&(fabric_id, node_id))
.is_some_and(|c| c.session_id == session_id)
{
if let Some(old) = self.cache.remove(&(fabric_id, node_id)) {
self.sessions.remove(old.session_id);
}
}
}
self.reschedule_resubscribe(PendingResubscribe {
sub_id,
attempt_at: Instant::now(),
node_id,
paths,
event_paths,
event_filters,
min_interval,
max_interval,
retry_count,
tx: report_tx,
});
}
return;
}
if !p.retried {
let Ok(fabric_id) = self.sole_fabric().map(|f| f.fabric_id) else {
Self::fail_pending(p, Error::Operational("round-trip timed out".into()));
return;
};
if self
.cache
.get(&(fabric_id, p.node_id))
.is_some_and(|c| c.session_id == session_id)
{
self.cache.remove(&(fabric_id, p.node_id));
}
if let Some((sid, peer)) = self
.cache
.get(&(fabric_id, p.node_id))
.map(|c| (c.session_id, c.peer))
{
self.resume_resend_pending(p, sid, peer).await;
} else {
self.enqueue_connect_waiter(fabric_id, p.node_id, ConnectWaiter::ResendPending(p));
}
return;
}
Self::fail_pending(p, Error::Operational("round-trip timed out".into()));
}
fn fail_pending(p: Pending, err: Error) {
match p.reply {
PendingReply::RoundTrip(reply)
| PendingReply::TimedAction { reply, .. }
| PendingReply::Action { reply, .. } => {
let _ = reply.send(Err(err));
}
PendingReply::ChunkedWrite { reply, .. } => {
let _ = reply.send(Err(err));
}
PendingReply::Read { reply, .. } => {
let _ = reply.send(Err(err));
}
PendingReply::Subscribe { reply, .. } => {
if let Some(reply) = reply {
let _ = reply.send(Err(err));
}
}
}
}
fn insert_subscription(&mut self, sub_id: SubId, entry: SubEntry) {
let new_key = (entry.session_id, entry.wire_sub_id);
if let Some(old) = self.subscriptions.insert(sub_id, entry) {
let old_key = (old.session_id, old.wire_sub_id);
if self.sub_index.get(&old_key) == Some(&sub_id) {
self.sub_index.remove(&old_key);
}
}
match self.sub_index.get(&new_key) {
None => {
self.sub_index.insert(new_key, sub_id);
}
Some(&existing) if existing == sub_id => {
}
Some(_) => {
}
}
}
fn remove_subscription(&mut self, sub_id: SubId) -> Option<SubEntry> {
let entry = self.subscriptions.remove(&sub_id)?;
let key = (entry.session_id, entry.wire_sub_id);
if self.sub_index.get(&key) == Some(&sub_id) {
self.sub_index.remove(&key);
}
Some(entry)
}
fn remove_subscriptions_for_node(&mut self, node_id: u64) {
let ids: Vec<SubId> = self
.subscriptions
.iter()
.filter(|(_, s)| s.node_id == node_id)
.map(|(id, _)| *id)
.collect();
for id in ids {
self.remove_subscription(id);
}
}
fn check_liveness(&mut self) {
let now = Instant::now();
let stale: Vec<SubId> = self
.subscriptions
.iter()
.filter(|(_, e)| e.liveness_deadline <= now)
.map(|(id, _)| *id)
.collect();
for id in stale {
self.begin_resubscribe(
id,
Error::Operational("subscription liveness timeout".into()),
);
}
}
fn begin_resubscribe(&mut self, sub_id: SubId, cause: Error) {
let Some(entry) = self.remove_subscription(sub_id) else {
return;
};
if !entry
.tx
.send_control(SubscriptionEvent::Resubscribing { cause })
{
return;
}
let wait = resubscribe_backoff(self.rng.as_ref(), 0);
self.resubscribes.push(PendingResubscribe {
sub_id,
attempt_at: Instant::now() + wait,
node_id: entry.node_id,
paths: entry.paths,
event_paths: entry.event_paths,
event_filters: entry.event_filters,
min_interval: entry.min_interval,
max_interval: entry.max_interval,
retry_count: 0,
tx: entry.tx,
});
}
async fn drive_resubscribes(&mut self) {
let now = Instant::now();
let mut due = Vec::new();
let mut i = 0;
while i < self.resubscribes.len() {
if self.resubscribes[i].attempt_at <= now {
due.push(self.resubscribes.swap_remove(i));
} else {
i += 1;
}
}
for pr in due {
self.attempt_resubscribe(pr).await;
}
}
async fn attempt_resubscribe(&mut self, pr: PendingResubscribe) {
if pr.tx.report_tx.is_closed() && pr.tx.ctrl_tx.is_closed() {
return;
}
let Ok(fabric_id) = self.sole_fabric().map(|f| f.fabric_id) else {
self.reschedule_resubscribe(pr);
return;
};
if let Some((sid, peer)) = self
.cache
.get(&(fabric_id, pr.node_id))
.map(|c| (c.session_id, c.peer))
{
self.resume_resubscribe(pr, sid, peer).await;
} else {
self.enqueue_connect_waiter(fabric_id, pr.node_id, ConnectWaiter::Resubscribe(pr));
}
}
fn reschedule_resubscribe(&mut self, mut pr: PendingResubscribe) {
if pr.tx.report_tx.is_closed() && pr.tx.ctrl_tx.is_closed() {
return;
}
pr.retry_count = pr.retry_count.saturating_add(1);
let wait = resubscribe_backoff(self.rng.as_ref(), pr.retry_count);
pr.attempt_at = Instant::now() + wait;
self.resubscribes.push(pr);
}
fn peer_for_session(&self, sid: SessionId) -> Option<SocketAddr> {
if let Some(addr) = self.sessions.get(sid).and_then(|s| s.peer_addr) {
return Some(addr);
}
self.subscriptions
.values()
.find(|e| e.session_id == sid)
.map(|e| e.peer)
.or_else(|| {
self.pending
.iter()
.find(|((s, _), _)| *s == sid)
.map(|(_, p)| p.peer)
})
.or_else(|| {
self.cache
.values()
.find(|c| c.session_id == sid)
.map(|c| c.peer)
})
}
}
pub(crate) fn current_matter_time() -> Result<matter_cert::MatterTime, Error> {
let secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_err(|e| Error::Operational(format!("clock: {e}")))?
.as_secs();
Ok(matter_cert::MatterTime::from_unix_secs(secs))
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
use super::*;
use crate::fabric::FabricConfig;
use crate::store::ControllerStore;
use matter_cert::{MatterTime, TrustAnchor, TrustedRoots};
use matter_commissioning::driver::{
decode_unsecured, encode_unsecured, operational_instance_name, InMemoryDatagram,
};
use matter_commissioning::{issue_noc, SystemNocRng, VerifiedCsr};
use matter_crypto::{
derive_compressed_fabric_id, derive_operational_ipk, CaseCredentials, CaseResponder,
RingSigner, Sigma1Outcome, Signer,
};
use matter_transport::{
DecodeInboundOutput, Discovery, MatterService, MrpFlags, ProtocolId, QueryHandle,
ServiceKind, SessionManager, SessionRole,
};
use std::time::Instant;
fn build_write_response_status(
endpoint: u16,
cluster: u32,
attribute: u32,
status: u8,
) -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.start_array(Tag::Context(0)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(2), u64::from(endpoint)).unwrap();
w.put_uint(Tag::Context(3), u64::from(cluster)).unwrap();
w.put_uint(Tag::Context(4), u64::from(attribute)).unwrap();
w.end_container().unwrap();
w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), u64::from(status)).unwrap();
w.end_container().unwrap();
w.end_container().unwrap();
w.end_container().unwrap();
w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
}
#[test]
fn needs_timed_detects_message_level_status_response() {
let payload = matter_interaction::build_status_response(NEEDS_TIMED_INTERACTION);
assert!(response_needs_timed(
crate::node::OP_INVOKE_REQUEST,
&payload
));
assert!(response_needs_timed(OP_WRITE_REQUEST, &payload));
}
#[test]
fn needs_timed_detects_per_command_invoke_status() {
let path = matter_interaction::CommandPath {
endpoint: 1,
cluster: 0x0101, command: 0x00, };
let payload = matter_interaction::build_invoke_response_status(
path,
matter_interaction::ImStatus::Failure(NEEDS_TIMED_INTERACTION),
);
assert!(
response_needs_timed(crate::node::OP_INVOKE_REQUEST, &payload),
"0xc6 carried in an InvokeResponse must trigger the timed retry"
);
}
#[test]
fn needs_timed_detects_per_attribute_write_status() {
let payload = build_write_response_status(0, 0x0028, 0x05, NEEDS_TIMED_INTERACTION);
assert!(
response_needs_timed(OP_WRITE_REQUEST, &payload),
"0xc6 carried in a WriteResponse must trigger the timed retry"
);
}
#[test]
fn needs_timed_false_for_success_and_other_failures() {
let path = matter_interaction::CommandPath {
endpoint: 1,
cluster: 0x0101,
command: 0x00,
};
let ok = matter_interaction::build_invoke_response_status(
path,
matter_interaction::ImStatus::Success,
);
assert!(!response_needs_timed(crate::node::OP_INVOKE_REQUEST, &ok));
let other = matter_interaction::build_invoke_response_status(
path,
matter_interaction::ImStatus::Failure(0x01),
);
assert!(!response_needs_timed(
crate::node::OP_INVOKE_REQUEST,
&other
));
let wok = build_write_response_status(0, 0x0028, 0x05, 0x00);
assert!(!response_needs_timed(OP_WRITE_REQUEST, &wok));
}
struct NullDiscovery;
impl Discovery for NullDiscovery {
fn publish(&mut self, _s: &MatterService) -> matter_transport::Result<()> {
Ok(())
}
fn unpublish(&mut self, _n: &str, _k: ServiceKind) -> matter_transport::Result<()> {
Ok(())
}
fn query(&mut self, _k: ServiceKind) -> matter_transport::Result<QueryHandle> {
Ok(QueryHandle(0))
}
fn stop_query(&mut self, _h: QueryHandle) {}
fn poll_results(&mut self, _h: QueryHandle) -> Vec<MatterService> {
Vec::new()
}
}
#[derive(Default)]
struct MemStore(std::sync::Mutex<Option<Vec<u8>>>);
impl ControllerStore for MemStore {
fn load(&self) -> Result<Option<Vec<u8>>, crate::store::StoreError> {
Ok(self.0.lock().unwrap().clone())
}
fn save(&self, snapshot: &[u8]) -> Result<(), crate::store::StoreError> {
*self.0.lock().unwrap() = Some(snapshot.to_vec());
Ok(())
}
}
fn cfg() -> FabricConfig {
FabricConfig {
fabric_id: 0xAABB_CCDD_0000_0001,
rcac_id: 1,
commissioner_node_id: 1,
validity: (
MatterTime::from_unix_secs(1_700_000_000),
MatterTime::NO_EXPIRY,
),
issue_icac: false,
}
}
#[tokio::test]
async fn create_fabric_persists_and_reopens() {
let store = Arc::new(MemStore::default());
let (io, _peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store.clone(),
io,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let fid = controller
.create_fabric(cfg())
.await
.expect("create_fabric");
assert_eq!(fid, 0xAABB_CCDD_0000_0001);
let bytes = store.load().expect("load").expect("snapshot present");
let restored = crate::snapshot::deserialize(&bytes).expect("deserialize");
assert_eq!(restored.fabrics.len(), 1);
assert_eq!(restored.fabrics[0].commissioner.node_id, 1);
}
#[tokio::test]
async fn nodes_lists_commissioned_devices_with_metadata() {
let mut fabric =
crate::fabric::create_fabric(&cfg(), &SystemNocRng).expect("create_fabric");
let fabric_id = fabric.fabric_id;
let node_id: u64 = 0x0000_0000_0000_0042;
fabric.devices.push(crate::state::DeviceEntry {
node_id,
peer_noc_public_key: [0u8; 65],
resumption_record: None,
last_known_addr: None,
vendor_id: Some(0xFFF1),
product_id: Some(0x8000),
label: Some("plug".to_string()),
});
let store = Arc::new(MemStore::default());
store
.save(
&crate::snapshot::serialize(&ControllerState {
fabrics: vec![fabric],
})
.unwrap(),
)
.unwrap();
let (io, _peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store,
io,
NullDiscovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let nodes = controller.nodes().await.expect("nodes");
assert_eq!(
nodes,
vec![crate::NodeInfo {
node_id,
fabric_id,
vendor_id: Some(0xFFF1),
product_id: Some(0x8000),
label: Some("plug".to_string()),
}]
);
}
#[tokio::test]
#[allow(clippy::too_many_lines)] async fn forget_node_drops_all_local_state_without_device_contact() {
let mut fabric =
crate::fabric::create_fabric(&cfg(), &SystemNocRng).expect("create_fabric");
let fabric_id = fabric.fabric_id;
let node_id: u64 = 0x0000_0000_0000_0042;
fabric.devices.push(crate::state::DeviceEntry {
node_id,
peer_noc_public_key: [0u8; 65],
resumption_record: None,
last_known_addr: None,
vendor_id: Some(0xFFF1),
product_id: Some(0x8000),
label: Some("plug".to_string()),
});
let store = Arc::new(MemStore::default());
let (io, _peer) = InMemoryDatagram::pair();
let mut actor = Actor::new(
io,
NullDiscovery,
store.clone(),
Arc::new(SystemNocRng),
ControllerState {
fabrics: vec![fabric],
},
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
);
actor.cache.insert(
(fabric_id, node_id),
CachedSession {
session_id: SessionId(7),
peer: "127.0.0.1:5540".parse().unwrap(),
},
);
let (sink, _report_rx, _ctrl_rx) = test_report_sink();
actor.insert_subscription(
SubId(1),
SubEntry {
tx: sink,
peer: "127.0.0.1:5540".parse().unwrap(),
reassembler: ReportReassembler::default(),
session_id: SessionId(7),
wire_sub_id: 0x1234,
node_id,
paths: vec![matter_interaction::ReadPath::all()],
event_paths: vec![],
event_filters: vec![],
min_interval: 1,
max_interval: 30,
liveness_deadline: Instant::now(),
},
);
let (nodes_tx, nodes_rx) = oneshot::channel();
actor
.dispatch_ready(Command::ListNodes { reply: nodes_tx })
.await;
assert_eq!(
nodes_rx.await.unwrap().len(),
1,
"fixture device must be listed before forget"
);
let (reply, rx) = oneshot::channel();
actor
.dispatch_ready(Command::ForgetNode { node_id, reply })
.await;
assert!(
rx.await.unwrap().expect("forget_node"),
"a device was found and removed"
);
assert!(
!actor.cache.contains_key(&(fabric_id, node_id)),
"the cached session must be evicted"
);
assert!(
actor.subscriptions.values().all(|s| s.node_id != node_id),
"the live subscription to the node must be dropped (else the \
resubscribe engine reconnects to the forgotten node)"
);
let (after_tx, after_rx) = oneshot::channel();
actor
.dispatch_ready(Command::ListNodes { reply: after_tx })
.await;
assert!(
after_rx.await.unwrap().is_empty(),
"nodes() must no longer list the forgotten node"
);
let bytes = store.load().unwrap().expect("snapshot saved");
let restored = crate::snapshot::deserialize(&bytes).unwrap();
assert!(
restored
.fabrics
.iter()
.all(|f| f.devices.iter().all(|d| d.node_id != node_id)),
"the persisted snapshot must no longer contain the forgotten device"
);
let (repeat_reply, repeat_rx) = oneshot::channel();
actor
.dispatch_ready(Command::ForgetNode {
node_id,
reply: repeat_reply,
})
.await;
assert!(
!repeat_rx.await.unwrap().expect("forget_node second call"),
"a second forget of the same node finds nothing"
);
}
#[test]
fn sub_index_tracks_subscribe_resubscribe_cancel_and_forget() {
let mut actor = actor_with_one_fabric();
let sid_a = SessionId(7);
let sid_b = SessionId(9);
let peer: std::net::SocketAddr = "127.0.0.1:5540".parse().unwrap();
let entry_for = |session_id, wire_sub_id, node_id| {
let (sink, _report_rx, _ctrl_rx) = test_report_sink();
SubEntry {
tx: sink,
peer,
reassembler: ReportReassembler::default(),
session_id,
wire_sub_id,
node_id,
paths: vec![matter_interaction::ReadPath::all()],
event_paths: vec![],
event_filters: vec![],
min_interval: 1,
max_interval: 30,
liveness_deadline: Instant::now(),
}
};
let id = SubId(1);
actor.insert_subscription(id, entry_for(sid_a, 0x1111, 7));
assert_eq!(actor.sub_index.get(&(sid_a, 0x1111)), Some(&id));
actor.insert_subscription(id, entry_for(sid_b, 0x2222, 7));
assert!(!actor.sub_index.contains_key(&(sid_a, 0x1111)));
assert_eq!(actor.sub_index.get(&(sid_b, 0x2222)), Some(&id));
assert_eq!(actor.sub_index.len(), 1);
actor.remove_subscription(id);
assert!(actor.sub_index.is_empty() && actor.subscriptions.is_empty());
actor.insert_subscription(SubId(2), entry_for(sid_a, 0x3333, 9));
actor.insert_subscription(SubId(3), entry_for(sid_b, 0x4444, 9));
actor.remove_subscriptions_for_node(9);
assert!(actor.sub_index.is_empty() && actor.subscriptions.is_empty());
let key = (sid_a, 0x5555);
let sub_a = SubId(10);
let sub_b = SubId(11);
actor.insert_subscription(sub_a, entry_for(sid_a, 0x5555, 20));
actor.insert_subscription(sub_b, entry_for(sid_a, 0x5555, 21));
assert_eq!(
actor.sub_index.get(&key),
Some(&sub_a),
"the index keeps the first owner of a colliding key"
);
assert!(
actor.subscriptions.contains_key(&sub_a) && actor.subscriptions.contains_key(&sub_b),
"both colliding subscriptions stay live in the primary map"
);
actor.remove_subscription(sub_b);
assert_eq!(
actor.sub_index.get(&key),
Some(&sub_a),
"removing the shadowed loser leaves the winner's index entry intact"
);
assert!(!actor.subscriptions.contains_key(&sub_b));
actor.remove_subscription(sub_a);
assert!(!actor.sub_index.contains_key(&key));
assert!(actor.subscriptions.is_empty());
}
#[tokio::test]
async fn upsert_device_is_update_only_and_never_resurrects_a_forgotten_node() {
let mut actor = actor_with_one_fabric();
let fabric_id = actor.sole_fabric().unwrap().fabric_id;
let peer: std::net::SocketAddr = "127.0.0.1:5540".parse().unwrap();
actor.upsert_device(fabric_id, 0x0000_0000_0000_0042, peer, Some(vec![1, 2, 3]));
assert!(
actor.sole_fabric().unwrap().devices.is_empty(),
"upsert_device must not fabricate a DeviceEntry for an unknown/forgotten node"
);
let node_id = 0x0000_0000_0000_0007;
actor
.state
.fabrics
.iter_mut()
.find(|f| f.fabric_id == fabric_id)
.unwrap()
.devices
.push(crate::state::DeviceEntry {
node_id,
peer_noc_public_key: [0x04; 65],
resumption_record: None,
last_known_addr: None,
vendor_id: None,
product_id: None,
label: None,
});
actor.upsert_device(fabric_id, node_id, peer, Some(vec![9, 9]));
let dev = actor.sole_fabric().unwrap().devices[0].clone();
assert_eq!(dev.last_known_addr.as_deref(), Some("127.0.0.1:5540"));
assert_eq!(dev.resumption_record, Some(vec![9, 9]));
assert_eq!(
dev.peer_noc_public_key, [0x04; 65],
"the existing NOC key must be preserved, never zeroed"
);
assert_eq!(
actor.sole_fabric().unwrap().devices.len(),
1,
"no duplicate entry was created for the known node"
);
}
#[tokio::test]
async fn invoke_group_emits_decodable_multicast_frame() {
use matter_codec::{Tag, TlvReader, Value};
use matter_transport::{decode_group_secured, DestNodeId, NodeId, SecuredMessageFlags};
let store = Arc::new(MemStore::default());
let (io, peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store.clone(),
io,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let fabric_id = controller
.create_fabric(cfg())
.await
.expect("create_fabric");
let key_set_id = 0x0042u16;
let group_key = controller
.create_group(key_set_id, 0)
.await
.expect("create_group");
assert_eq!(group_key.key_set_id, key_set_id);
assert_eq!(group_key.epoch_key.len(), 16, "epoch key must be 16 bytes");
let snap = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
assert_eq!(snap.fabrics[0].group_keys.len(), 1);
assert_eq!(snap.fabrics[0].outbound_group_counter, 0);
let commissioner_node_id = snap.fabrics[0].commissioner.node_id;
let root_public_key = *snap.fabrics[0].rcac_cert.public_key().as_bytes();
let epoch_key: [u8; 16] = group_key.epoch_key.clone().try_into().unwrap();
let group_id = 0xBEEFu16;
let path = crate::CommandPath {
endpoint: 0,
cluster: 0x0006,
command: 0x01,
};
let fields = Value::Structure(vec![]); controller
.invoke_group(group_id, key_set_id, path, fields.clone())
.await
.expect("invoke_group");
let snap2 = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
assert_eq!(
snap2.fabrics[0].outbound_group_counter, GROUP_COUNTER_BLOCK,
"the reservation ceiling must be persisted before the send"
);
let (wire, _from) = peer.recv_from().await.expect("frame emitted");
let compressed = derive_compressed_fabric_id(&root_public_key, fabric_id).unwrap();
let op_group_key = derive_operational_ipk(&epoch_key, &compressed).unwrap();
let (header, plaintext) = decode_group_secured(&wire, &op_group_key).expect("decode group");
assert!(header.flags.contains(SecuredMessageFlags::SOURCE_PRESENT));
assert!(header.flags.contains(SecuredMessageFlags::DEST_GROUP));
assert_eq!(header.source_node_id, Some(NodeId(commissioner_node_id)));
assert_eq!(
header.destination_node_id,
Some(DestNodeId::Group(group_id))
);
assert_eq!(header.message_counter.0, 0);
let (ph, app) = matter_transport::decode_protocol_header(&plaintext).unwrap();
assert_eq!(ph.opcode, crate::node::OP_INVOKE_REQUEST);
assert_eq!(ph.protocol_id, ProtocolId::INTERACTION_MODEL);
let fields_tlv = crate::node::value_to_tlv(&fields).unwrap();
let expected = matter_interaction::build_invoke_request_group(path, &fields_tlv);
assert_eq!(app, &expected[..], "IM payload must be the InvokeRequest");
let (_t, msg) = TlvReader::new(app).read_value().unwrap();
let Value::Structure(members) = msg else {
panic!("InvokeRequest is a structure")
};
assert_eq!(
members
.iter()
.find(|(t, _)| *t == Tag::Context(0))
.map(|(_, v)| v),
Some(&Value::Bool(true))
);
let invoke_requests = members
.iter()
.find(|(t, _)| *t == Tag::Context(2))
.map(|(_, v)| v)
.unwrap();
let Value::Array(command_list) = invoke_requests else {
panic!("InvokeRequests is an array")
};
let Value::Structure(first_command) = &command_list[0] else {
panic!("CommandDataIB is a structure")
};
let cmd_path = first_command
.iter()
.find(|(t, _)| *t == Tag::Context(0))
.map(|(_, v)| v)
.unwrap();
let Value::List(path_members) = cmd_path else {
panic!("CommandPath is a list")
};
assert_eq!(path_members[0], (Tag::Context(0), Value::Uint(0)));
assert_eq!(path_members[1], (Tag::Context(1), Value::Uint(0x0006)));
assert_eq!(path_members[2], (Tag::Context(2), Value::Uint(0x01)));
}
#[tokio::test]
async fn create_group_upserts_and_invoke_uses_newest_key() {
use matter_transport::decode_group_secured;
let store = Arc::new(MemStore::default());
let (io, _peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store.clone(),
io,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let fabric_id = controller
.create_fabric(cfg())
.await
.expect("create_fabric");
let key_set_id = 0x0042u16;
let first = controller
.create_group(key_set_id, 0)
.await
.expect("create_group #1");
let second = controller
.create_group(key_set_id, 0)
.await
.expect("create_group #2");
assert_ne!(first.epoch_key, second.epoch_key, "fresh key each create");
let snap = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
assert_eq!(
snap.fabrics[0].group_keys.len(),
1,
"create_group must upsert, not append a duplicate"
);
assert_eq!(
snap.fabrics[0].group_keys[0].epoch_key[..],
second.epoch_key[..],
"the stored key must be the newest one"
);
let mut poisoned = snap;
poisoned.fabrics[0].group_keys.insert(
0,
crate::state::GroupKeySetConfig::new(key_set_id, [0xAA; 16], 0),
);
store
.save(&crate::snapshot::serialize(&poisoned).unwrap())
.unwrap();
let (io2, peer2) = InMemoryDatagram::pair();
let controller2 = crate::controller::MatterController::with_components(
store.clone(),
io2,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("reopen");
let path = crate::CommandPath {
endpoint: 0,
cluster: 0x0006,
command: 0x01,
};
controller2
.invoke_group(
0x0008,
key_set_id,
path,
matter_codec::Value::Structure(vec![]),
)
.await
.expect("invoke_group");
let (wire, _from) = peer2.recv_from().await.expect("frame emitted");
let root_public_key = *poisoned.fabrics[0].rcac_cert.public_key().as_bytes();
let compressed = derive_compressed_fabric_id(&root_public_key, fabric_id).unwrap();
let newest_epoch: [u8; 16] = second.epoch_key.clone().try_into().unwrap();
let op_newest = derive_operational_ipk(&newest_epoch, &compressed).unwrap();
decode_group_secured(&wire, &op_newest)
.expect("frame must decrypt under the NEWEST key for this key set id");
let op_stale = derive_operational_ipk(&[0xAA; 16], &compressed).unwrap();
assert!(
decode_group_secured(&wire, &op_stale).is_err(),
"frame must NOT be encrypted under the stale first-match key"
);
}
#[tokio::test]
async fn invoke_group_unprovisioned_key_set_errors() {
let store = Arc::new(MemStore::default());
let (io, _peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store.clone(),
io,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.create_fabric(cfg())
.await
.expect("create_fabric");
let path = crate::CommandPath {
endpoint: 0,
cluster: 0x0006,
command: 0x01,
};
let err = controller
.invoke_group(0xBEEF, 0x0099, path, matter_codec::Value::Structure(vec![]))
.await
.expect_err("must reject unprovisioned key set");
assert!(matches!(err, Error::GroupNotProvisioned(0x0099)));
}
fn group_test_actor(
key_set_id: u16,
epoch_key: [u8; 16],
) -> (Actor<InMemoryDatagram, NullDiscovery>, InMemoryDatagram) {
let (io, peer) = InMemoryDatagram::pair();
let mut fabric =
crate::fabric::create_fabric(&cfg(), &SystemNocRng).expect("create_fabric");
fabric.group_keys.push(crate::state::GroupKeySetConfig::new(
key_set_id, epoch_key, 0,
));
let actor = Actor::new(
io,
NullDiscovery,
Arc::new(MemStore::default()),
Arc::new(SystemNocRng),
ControllerState {
fabrics: vec![fabric],
},
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
);
(actor, peer)
}
fn empty_fields_tlv() -> Vec<u8> {
crate::node::value_to_tlv(&matter_codec::Value::Structure(vec![])).unwrap()
}
#[tokio::test]
async fn group_key_cache_reused_across_sends() {
let key_set_id = 0x0042u16;
let epoch_key = [0x11u8; 16];
let (mut actor, peer) = group_test_actor(key_set_id, epoch_key);
let fabric_id = actor.sole_fabric().unwrap().fabric_id;
let op_group_key = op_group_key_of(&actor.state, &epoch_key);
let fields_tlv = empty_fields_tlv();
assert!(
actor.group_key_cache.is_empty(),
"cache starts cold — nothing is derived before the first send"
);
actor
.handle_invoke_group(0xBEEF, key_set_id, on_command_path(), &fields_tlv)
.await
.expect("first group send");
assert_eq!(
actor.group_key_cache.len(),
1,
"the first send must populate exactly one cache entry"
);
actor
.handle_invoke_group(0xBEEF, key_set_id, on_command_path(), &fields_tlv)
.await
.expect("second group send");
assert_eq!(
actor.group_key_cache.len(),
1,
"the second send must reuse the entry, not add another"
);
let entry = actor
.group_key_cache
.get(&fabric_id)
.expect("entry keyed by fabric id");
assert_eq!(
entry.epoch_key, epoch_key,
"the cached entry must record the epoch key it was derived from"
);
assert_eq!(
entry.op_group_key, op_group_key,
"the cached operational group key must equal the from-scratch derivation"
);
assert_eq!(
entry.privacy_key,
matter_crypto::derive_group_privacy_key(&op_group_key).unwrap(),
"the cached privacy key must equal the from-scratch derivation"
);
let (wire1, _from) = peer.recv_from().await.expect("frame 1 emitted");
let (wire2, _from) = peer.recv_from().await.expect("frame 2 emitted");
assert_eq!(wire_group_counter(&wire1, &op_group_key), 0);
assert_eq!(wire_group_counter(&wire2, &op_group_key), 1);
let (_h, plaintext) =
matter_transport::decode_group_secured(&wire2, &op_group_key).expect("decode");
let (_ph, app) = matter_transport::decode_protocol_header(&plaintext).unwrap();
assert_eq!(
app,
&matter_interaction::build_invoke_request_group(on_command_path(), &fields_tlv)[..],
"the cached path must still carry the same IM payload"
);
}
#[tokio::test]
async fn group_key_cache_invalidated_on_epoch_rotation() {
let key_set_id = 0x0042u16;
let old_epoch = [0x11u8; 16];
let new_epoch = [0x22u8; 16];
let (mut actor, peer) = group_test_actor(key_set_id, old_epoch);
let fabric_id = actor.sole_fabric().unwrap().fabric_id;
let op_old = op_group_key_of(&actor.state, &old_epoch);
let op_new = op_group_key_of(&actor.state, &new_epoch);
let fields_tlv = empty_fields_tlv();
actor
.handle_invoke_group(0xBEEF, key_set_id, on_command_path(), &fields_tlv)
.await
.expect("send under the old epoch key");
let (wire_old, _from) = peer.recv_from().await.expect("frame 1 emitted");
matter_transport::decode_group_secured(&wire_old, &op_old)
.expect("first frame must decrypt under the old key");
{
let fabric = actor.sole_fabric_mut().unwrap();
let slot = fabric
.group_keys
.iter_mut()
.rfind(|k| k.key_set_id == key_set_id)
.expect("provisioned key set");
*slot = crate::state::GroupKeySetConfig::new(key_set_id, new_epoch, 0);
}
actor
.handle_invoke_group(0xBEEF, key_set_id, on_command_path(), &fields_tlv)
.await
.expect("send after rotation");
let (wire_new, _from) = peer.recv_from().await.expect("frame 2 emitted");
matter_transport::decode_group_secured(&wire_new, &op_new)
.expect("frame must decrypt under the NEW epoch key's derivation");
assert!(
matter_transport::decode_group_secured(&wire_new, &op_old).is_err(),
"frame must NOT still be encrypted under the cached stale key"
);
let entry = actor
.group_key_cache
.get(&fabric_id)
.expect("entry keyed by fabric id");
assert_eq!(
entry.epoch_key, new_epoch,
"the refreshed entry must record the new epoch key"
);
assert_eq!(entry.op_group_key, op_new);
assert_eq!(
actor.group_key_cache.len(),
1,
"invalidation replaces the entry rather than accumulating"
);
}
#[tokio::test]
async fn group_key_cache_invalidated_on_root_key_change() {
let key_set_id = 0x0042u16;
let epoch_key = [0x11u8; 16];
let (mut actor, peer) = group_test_actor(key_set_id, epoch_key);
let fabric_id = actor.sole_fabric().unwrap().fabric_id;
let op_real = op_group_key_of(&actor.state, &epoch_key);
let real_root = *actor
.sole_fabric()
.unwrap()
.rcac_cert
.public_key()
.as_bytes();
let fields_tlv = empty_fields_tlv();
actor
.handle_invoke_group(0xBEEF, key_set_id, on_command_path(), &fields_tlv)
.await
.expect("first send");
let (_wire, _from) = peer.recv_from().await.expect("frame 1 emitted");
{
let entry = actor
.group_key_cache
.get_mut(&fabric_id)
.expect("warmed entry");
entry.root_public_key = [0x04; 65];
entry.op_group_key = [0xEE; 16];
entry.privacy_key = [0xEE; 16];
}
actor
.handle_invoke_group(0xBEEF, key_set_id, on_command_path(), &fields_tlv)
.await
.expect("second send");
let (wire, _from) = peer.recv_from().await.expect("frame 2 emitted");
matter_transport::decode_group_secured(&wire, &op_real)
.expect("frame must be re-derived under the fabric's REAL root key");
assert!(
matter_transport::decode_group_secured(&wire, &[0xEE; 16]).is_err(),
"frame must not be encrypted under the poisoned cache entry"
);
let entry = actor
.group_key_cache
.get(&fabric_id)
.expect("entry keyed by fabric id");
assert_eq!(
entry.root_public_key, real_root,
"the refreshed entry must record the root key it was derived under"
);
assert_eq!(entry.op_group_key, op_real);
}
#[derive(Default)]
struct CountingStore {
inner: std::sync::Mutex<Option<Vec<u8>>>,
saves: std::sync::atomic::AtomicUsize,
fail: std::sync::atomic::AtomicBool,
}
impl CountingStore {
fn saves(&self) -> usize {
self.saves.load(std::sync::atomic::Ordering::SeqCst)
}
fn set_failing(&self, fail: bool) {
self.fail.store(fail, std::sync::atomic::Ordering::SeqCst);
}
}
impl ControllerStore for CountingStore {
fn load(&self) -> Result<Option<Vec<u8>>, crate::store::StoreError> {
Ok(self.inner.lock().unwrap().clone())
}
fn save(&self, snapshot: &[u8]) -> Result<(), crate::store::StoreError> {
if self.fail.load(std::sync::atomic::Ordering::SeqCst) {
return Err(crate::store::StoreError::Io(std::io::Error::other(
"disk full",
)));
}
*self.inner.lock().unwrap() = Some(snapshot.to_vec());
self.saves.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
}
}
fn op_group_key_of(snap: &ControllerState, epoch_key: &[u8; 16]) -> [u8; 16] {
let root_public_key = *snap.fabrics[0].rcac_cert.public_key().as_bytes();
let compressed =
derive_compressed_fabric_id(&root_public_key, snap.fabrics[0].fabric_id).unwrap();
derive_operational_ipk(epoch_key, &compressed).unwrap()
}
fn wire_group_counter(wire: &[u8], op_group_key: &[u8; 16]) -> u32 {
let (header, _plaintext) =
matter_transport::decode_group_secured(wire, op_group_key).expect("decode group");
header.message_counter.0
}
fn on_command_path() -> crate::CommandPath {
crate::CommandPath {
endpoint: 0,
cluster: 0x0006,
command: 0x01,
}
}
#[tokio::test]
async fn group_sends_share_one_reservation() {
let store = Arc::new(CountingStore::default());
let (io, peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store.clone(),
io,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.create_fabric(cfg())
.await
.expect("create_fabric");
let key_set_id = 0x0042u16;
let group_key = controller
.create_group(key_set_id, 0)
.await
.expect("create_group");
let epoch_key: [u8; 16] = group_key.epoch_key.clone().try_into().unwrap();
let baseline = store.saves();
let mut counters = Vec::new();
for _ in 0..3 {
controller
.invoke_group(
0xBEEF,
key_set_id,
on_command_path(),
matter_codec::Value::Structure(vec![]),
)
.await
.expect("invoke_group");
let (wire, _from) = peer.recv_from().await.expect("frame emitted");
let snap = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let op = op_group_key_of(&snap, &epoch_key);
counters.push(wire_group_counter(&wire, &op));
}
assert_eq!(
store.saves() - baseline,
1,
"3 group sends inside one reserved block must cost exactly 1 store write"
);
assert_eq!(
counters,
vec![0, 1, 2],
"counters must still be strictly increasing across the block"
);
let snap = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
assert_eq!(
snap.fabrics[0].outbound_group_counter, GROUP_COUNTER_BLOCK,
"the serialized counter must be the reservation ceiling"
);
}
#[tokio::test]
async fn group_counter_survives_restart_without_reuse() {
let store: Arc<CountingStore> = Arc::new(CountingStore::default());
let (io, peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store.clone(),
io,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.create_fabric(cfg())
.await
.expect("create_fabric");
let key_set_id = 0x0042u16;
let group_key = controller
.create_group(key_set_id, 0)
.await
.expect("create_group");
let epoch_key: [u8; 16] = group_key.epoch_key.clone().try_into().unwrap();
let mut sent = Vec::new();
for _ in 0..2 {
controller
.invoke_group(
0xBEEF,
key_set_id,
on_command_path(),
matter_codec::Value::Structure(vec![]),
)
.await
.expect("invoke_group");
let (wire, _from) = peer.recv_from().await.expect("frame emitted");
let snap = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let op = op_group_key_of(&snap, &epoch_key);
sent.push(wire_group_counter(&wire, &op));
}
drop(controller);
let persisted_ceiling = crate::snapshot::deserialize(&store.load().unwrap().unwrap())
.unwrap()
.fabrics[0]
.outbound_group_counter;
let (io2, peer2) = InMemoryDatagram::pair();
let controller2 = crate::controller::MatterController::with_components(
store.clone(),
io2,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("reopen");
controller2
.invoke_group(
0xBEEF,
key_set_id,
on_command_path(),
matter_codec::Value::Structure(vec![]),
)
.await
.expect("invoke_group after restart");
let (wire, _from) = peer2.recv_from().await.expect("frame emitted");
let snap = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let op = op_group_key_of(&snap, &epoch_key);
let after_restart = wire_group_counter(&wire, &op);
assert!(
after_restart >= persisted_ceiling,
"post-restart counter {after_restart} must resume at or above the persisted ceiling {persisted_ceiling}"
);
assert!(
sent.iter().all(|&c| after_restart > c),
"post-restart counter {after_restart} must exceed every counter already sent ({sent:?})"
);
}
#[tokio::test]
async fn failed_reservation_rolls_back_and_burns_no_counter() {
let store = Arc::new(CountingStore::default());
let (io, peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store.clone(),
io,
NullDiscovery,
Arc::new(matter_commissioning::SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.create_fabric(cfg())
.await
.expect("create_fabric");
let key_set_id = 0x0042u16;
let group_key = controller
.create_group(key_set_id, 0)
.await
.expect("create_group");
let epoch_key: [u8; 16] = group_key.epoch_key.clone().try_into().unwrap();
let snap_before = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
assert_eq!(
snap_before.fabrics[0].outbound_group_counter, 0,
"no reservation yet"
);
store.set_failing(true);
let err = controller
.invoke_group(
0xBEEF,
key_set_id,
on_command_path(),
matter_codec::Value::Structure(vec![]),
)
.await
.expect_err("a failed reservation save must fail the send");
assert!(
format!("{err}").to_lowercase().contains("disk full"),
"expected the store error to propagate, got: {err}"
);
let emitted = tokio::time::timeout(
std::time::Duration::from_millis(100),
Box::pin(peer.recv_from()),
)
.await;
assert!(
emitted.is_err(),
"a send whose reservation never reached the store must emit no datagram"
);
let snap_failed = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
assert_eq!(
snap_failed.fabrics[0].outbound_group_counter, 0,
"a failed save must not have persisted the raised ceiling"
);
store.set_failing(false);
controller
.invoke_group(
0xBEEF,
key_set_id,
on_command_path(),
matter_codec::Value::Structure(vec![]),
)
.await
.expect("invoke_group after the store recovers");
let (wire, _from) = peer.recv_from().await.expect("frame emitted");
let snap_after = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let op = op_group_key_of(&snap_after, &epoch_key);
assert_eq!(
wire_group_counter(&wire, &op),
0,
"the failed send burned no counter: the retry reuses it"
);
assert_eq!(
snap_after.fabrics[0].outbound_group_counter, GROUP_COUNTER_BLOCK,
"the retry must have re-run the reservation — proving the ceiling was rolled back"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn best_effort_snapshot_mid_block_is_safe() {
let store = Arc::new(CountingStore::default());
let key_set_id = 0x0042u16;
let fields_tlv =
crate::node::value_to_tlv(&matter_codec::Value::Structure(vec![])).unwrap();
let (io, peer) = InMemoryDatagram::pair();
let fabric = crate::fabric::create_fabric(&cfg(), &SystemNocRng).unwrap();
let mut actor = Actor::new(
io,
NullDiscovery,
store.clone(),
Arc::new(SystemNocRng),
ControllerState {
fabrics: vec![fabric],
},
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
);
let (reply, rx) = oneshot::channel();
actor
.dispatch_ready(Command::CreateGroup {
key_set_id,
epoch_start_time: 0,
reply,
})
.await;
let group_key = rx.await.unwrap().expect("create_group");
let epoch_key: [u8; 16] = group_key.epoch_key.clone().try_into().unwrap();
let (reply, rx) = oneshot::channel();
actor
.dispatch_ready(Command::InvokeGroup {
group_id: 0xBEEF,
key_set_id,
path: on_command_path(),
fields_tlv: fields_tlv.clone(),
reply,
})
.await;
rx.await.unwrap().expect("invoke_group");
let (wire, _from) = peer.recv_from().await.expect("frame emitted");
let snap = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let op = op_group_key_of(&snap, &epoch_key);
let first = wire_group_counter(&wire, &op);
let before = store.saves();
actor.persist_best_effort();
let mut landed = false;
for _ in 0..200 {
if store.saves() > before {
landed = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(landed, "the detached best-effort save must have run");
let mid = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
assert_eq!(
mid.fabrics[0].outbound_group_counter, GROUP_COUNTER_BLOCK,
"a mid-block best-effort snapshot must serialize the ceiling, not the live counter"
);
let (io2, peer2) = InMemoryDatagram::pair();
let mut actor2 = Actor::new(
io2,
NullDiscovery,
store.clone(),
Arc::new(SystemNocRng),
mid,
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
);
let (reply, rx) = oneshot::channel();
actor2
.dispatch_ready(Command::InvokeGroup {
group_id: 0xBEEF,
key_set_id,
path: on_command_path(),
fields_tlv,
reply,
})
.await;
rx.await.unwrap().expect("invoke_group after crash");
let (wire2, _from) = peer2.recv_from().await.expect("frame emitted");
let after = wire_group_counter(&wire2, &op);
assert!(
after >= GROUP_COUNTER_BLOCK,
"post-crash counter {after} must resume at the ceiling {GROUP_COUNTER_BLOCK}"
);
assert!(
after > first,
"post-crash counter {after} must not reuse the already-sent {first}"
);
}
struct FixedDiscovery {
addr: std::net::SocketAddr,
instance_name: String,
}
impl Discovery for FixedDiscovery {
fn publish(&mut self, _s: &MatterService) -> matter_transport::Result<()> {
Ok(())
}
fn unpublish(&mut self, _n: &str, _k: ServiceKind) -> matter_transport::Result<()> {
Ok(())
}
fn query(&mut self, _k: ServiceKind) -> matter_transport::Result<QueryHandle> {
Ok(QueryHandle(1))
}
fn stop_query(&mut self, _h: QueryHandle) {}
fn poll_results(&mut self, _h: QueryHandle) -> Vec<MatterService> {
vec![MatterService::new(
self.instance_name.clone(),
ServiceKind::Operational,
vec![self.addr.ip()],
self.addr.port(),
std::collections::HashMap::new(),
)]
}
}
struct DrainingDiscovery {
addr: std::net::SocketAddr,
instance_name: String,
drained: bool,
}
impl Discovery for DrainingDiscovery {
fn publish(&mut self, _s: &MatterService) -> matter_transport::Result<()> {
Ok(())
}
fn unpublish(&mut self, _n: &str, _k: ServiceKind) -> matter_transport::Result<()> {
Ok(())
}
fn query(&mut self, _k: ServiceKind) -> matter_transport::Result<QueryHandle> {
self.drained = false; Ok(QueryHandle(1))
}
fn stop_query(&mut self, _h: QueryHandle) {}
fn poll_results(&mut self, _h: QueryHandle) -> Vec<MatterService> {
if self.drained {
return Vec::new();
}
self.drained = true;
vec![MatterService::new(
self.instance_name.clone(),
ServiceKind::Operational,
vec![self.addr.ip()],
self.addr.port(),
std::collections::HashMap::new(),
)]
}
}
fn test_report_sink() -> (
ReportSink,
mpsc::Receiver<SubscriptionEvent>,
mpsc::UnboundedReceiver<SubscriptionEvent>,
) {
let (report_tx, report_rx) = mpsc::channel::<SubscriptionEvent>(SUBSCRIPTION_CHANNEL_CAP);
let (ctrl_tx, ctrl_rx) = mpsc::unbounded_channel::<SubscriptionEvent>();
(
ReportSink {
report_tx,
ctrl_tx,
dropped: 0,
},
report_rx,
ctrl_rx,
)
}
fn build_report_data(ep: u16, cl: u32, at: u32, value: &matter_codec::Value) -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap(); w.put_uint(Tag::Context(0), 0x1234_5678).unwrap(); w.start_array(Tag::Context(1)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_structure(Tag::Context(1)).unwrap(); w.start_list(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(2), u64::from(ep)).unwrap();
w.put_uint(Tag::Context(3), u64::from(cl)).unwrap();
w.put_uint(Tag::Context(4), u64::from(at)).unwrap();
w.end_container().unwrap(); w.write_value(Tag::Context(2), value).unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap(); w.end_container().unwrap(); buf
}
fn build_report_data_event(
ep: u16,
cl: u32,
ev: u32,
event_number: u64,
value: &matter_codec::Value,
) -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap(); w.put_uint(Tag::Context(0), 0x1234_5678).unwrap(); w.start_array(Tag::Context(2)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_structure(Tag::Context(1)).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(1), u64::from(ep)).unwrap();
w.put_uint(Tag::Context(2), u64::from(cl)).unwrap();
w.put_uint(Tag::Context(3), u64::from(ev)).unwrap();
w.end_container().unwrap(); w.put_uint(Tag::Context(1), event_number).unwrap(); w.put_uint(Tag::Context(2), 2).unwrap(); w.put_uint(Tag::Context(3), 0).unwrap(); w.write_value(Tag::Context(7), value).unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap(); buf
}
fn build_report_data_chunk(
ep: u16,
cl: u32,
at: u32,
value: &matter_codec::Value,
more: bool,
) -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.put_uint(Tag::Context(0), 0x1234_5678).unwrap(); w.start_array(Tag::Context(1)).unwrap();
w.start_structure(Tag::Anonymous).unwrap();
w.start_structure(Tag::Context(1)).unwrap();
w.start_list(Tag::Context(1)).unwrap();
w.put_uint(Tag::Context(2), u64::from(ep)).unwrap();
w.put_uint(Tag::Context(3), u64::from(cl)).unwrap();
w.put_uint(Tag::Context(4), u64::from(at)).unwrap();
w.end_container().unwrap();
w.write_value(Tag::Context(2), value).unwrap();
w.end_container().unwrap();
w.end_container().unwrap();
w.end_container().unwrap(); if more {
w.put_bool(Tag::Context(3), true).unwrap(); }
w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
}
fn build_report_data_append(
ep: u16,
cl: u32,
at: u32,
value: &matter_codec::Value,
more: bool,
) -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.put_uint(Tag::Context(0), 0x1234_5678).unwrap(); w.start_array(Tag::Context(1)).unwrap();
w.start_structure(Tag::Anonymous).unwrap();
w.start_structure(Tag::Context(1)).unwrap();
w.start_list(Tag::Context(1)).unwrap();
w.put_uint(Tag::Context(2), u64::from(ep)).unwrap();
w.put_uint(Tag::Context(3), u64::from(cl)).unwrap();
w.put_uint(Tag::Context(4), u64::from(at)).unwrap();
w.put_null(Tag::Context(5)).unwrap(); w.end_container().unwrap();
w.write_value(Tag::Context(2), value).unwrap();
w.end_container().unwrap();
w.end_container().unwrap();
w.end_container().unwrap();
if more {
w.put_bool(Tag::Context(3), true).unwrap();
}
w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
}
async fn run_chunked_read_device(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
chunk0: Vec<u8>,
chunk1: Vec<u8>,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
let (wire, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage { exchange_id, .. } =
sessions.decode_inbound(&wire, Instant::now()).unwrap()
else {
panic!("expected ReadRequest");
};
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x05,
ProtocolId::INTERACTION_MODEL,
&chunk0,
MrpFlags { reliable: true },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
let (ack, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage {
opcode,
exchange_id: ack_exchange,
..
} = sessions.decode_inbound(&ack, Instant::now()).unwrap()
else {
panic!("expected StatusResponse ack");
};
assert_eq!(opcode, 0x01, "controller must ack the chunk");
assert_eq!(
ack_exchange, exchange_id,
"StatusResponse must ride the read exchange (enables the chunk-ack piggyback)"
);
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x05,
ProtocolId::INTERACTION_MODEL,
&chunk1,
MrpFlags { reliable: true },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
}
#[allow(clippy::too_many_arguments)] async fn run_loopback_device(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
echoes: usize,
reply_payload: Vec<u8>,
expect_timed: bool,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31, ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes()); body.extend_from_slice(&0u32.to_le_bytes()); body.extend_from_slice(&0u16.to_le_bytes()); let report = encode_unsecured(
201,
m.exchange_id,
0x40, ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
if expect_timed {
let (wire, _) = io.recv_from().await.unwrap();
let decoded = sessions.decode_inbound(&wire, Instant::now()).unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id,
opcode,
..
} = decoded
else {
panic!("expected a TimedRequest app message");
};
assert_eq!(opcode, 0x0a, "expected TimedRequest opcode 0x0a");
let status = matter_interaction::build_status_response(0);
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x01, ProtocolId::INTERACTION_MODEL,
&status,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
}
for _ in 0..echoes {
let (wire, _) = io.recv_from().await.unwrap();
let decoded = sessions.decode_inbound(&wire, Instant::now()).unwrap();
let DecodeInboundOutput::AppMessage { exchange_id, .. } = decoded else {
panic!("expected an IM request app message");
};
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x05, ProtocolId::INTERACTION_MODEL,
&reply_payload,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
}
}
async fn ack_timed_then_reply(
io: &InMemoryDatagram,
sessions: &mut SessionManager,
sid: SessionId,
ctrl_addr: std::net::SocketAddr,
write_response: &[u8],
) {
let (w, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id,
opcode,
..
} = sessions.decode_inbound(&w, Instant::now()).unwrap()
else {
panic!("expected a TimedRequest app message");
};
assert_eq!(opcode, 0x0a, "expected TimedRequest opcode 0x0a");
let status = matter_interaction::build_status_response(0);
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x01,
ProtocolId::INTERACTION_MODEL,
&status,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
let (w2, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id: e2,
opcode: op2,
..
} = sessions.decode_inbound(&w2, Instant::now()).unwrap()
else {
panic!("expected a timed WriteRequest app message");
};
assert_eq!(op2, 0x06, "expected timed WriteRequest opcode 0x06");
let out2 = sessions
.encode_outbound(
sid,
Some(e2),
0x07,
ProtocolId::INTERACTION_MODEL,
write_response,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out2.wire_bytes, ctrl_addr).await.unwrap();
}
async fn run_timed_retry_device(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
write_response: Vec<u8>,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
let (w, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id,
opcode,
..
} = sessions.decode_inbound(&w, Instant::now()).unwrap()
else {
panic!("expected a plain WriteRequest app message");
};
assert_eq!(opcode, 0x06, "cycle 1 must start with a plain WriteRequest");
let reject = matter_interaction::build_status_response(0xc6);
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x01,
ProtocolId::INTERACTION_MODEL,
&reject,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
ack_timed_then_reply(&io, &mut sessions, sid, ctrl_addr, &write_response).await;
ack_timed_then_reply(&io, &mut sessions, sid, ctrl_addr, &write_response).await;
}
fn build_subscribe_response(subscription_id: u32, max_interval: u16) -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.put_uint(Tag::Context(0), u64::from(subscription_id))
.unwrap();
w.put_uint(Tag::Context(2), u64::from(max_interval))
.unwrap();
w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
}
async fn run_subscription_device(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
reports: Vec<Vec<u8>>,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
let (wire, _) = io.recv_from().await.unwrap();
let decoded = sessions.decode_inbound(&wire, Instant::now()).unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id,
opcode,
..
} = decoded
else {
panic!("expected SubscribeRequest");
};
assert_eq!(opcode, 0x03, "expected SubscribeRequest opcode");
let sub_resp = build_subscribe_response(0x1234_5678, 30);
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x04,
ProtocolId::INTERACTION_MODEL,
&sub_resp,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
for report in &reports {
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x05,
ProtocolId::INTERACTION_MODEL,
report,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
let _ =
tokio::time::timeout(std::time::Duration::from_millis(100), io.recv_from()).await;
}
}
#[allow(clippy::too_many_lines)] async fn run_concurrent_sub_roundtrip_device(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
let (wire, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id: sub_exchange,
opcode,
..
} = sessions.decode_inbound(&wire, Instant::now()).unwrap()
else {
panic!("expected SubscribeRequest");
};
assert_eq!(opcode, 0x03, "expected SubscribeRequest opcode");
let sub_resp = build_subscribe_response(0x1234_5678, 30);
let out = sessions
.encode_outbound(
sid,
Some(sub_exchange),
0x04,
ProtocolId::INTERACTION_MODEL,
&sub_resp,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
let (wire, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id: rt_exchange,
opcode: rt_opcode,
..
} = sessions.decode_inbound(&wire, Instant::now()).unwrap()
else {
panic!("expected round-trip request");
};
assert_eq!(rt_opcode, 0x02, "expected the round-trip request opcode");
let steady = build_report_data(1, 0x06, 0x0000, &matter_codec::Value::Bool(true));
let out = sessions
.encode_outbound(
sid,
Some(sub_exchange),
0x05,
ProtocolId::INTERACTION_MODEL,
&steady,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
let out = sessions
.encode_outbound(
sid,
Some(rt_exchange),
0x05,
ProtocolId::INTERACTION_MODEL,
b"pong",
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
let _ = tokio::time::timeout(std::time::Duration::from_millis(100), io.recv_from()).await;
}
#[allow(clippy::too_many_lines)] async fn run_resubscribe_device(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
let wire_ids = [0x1111_1111_u32, 0x2222_2222_u32];
let mut cycle = 0usize;
loop {
let Ok(Ok((wire, _))) =
tokio::time::timeout(std::time::Duration::from_secs(30), io.recv_from()).await
else {
return; };
if wire.len() >= 3 && wire[1] == 0 && wire[2] == 0 {
continue; }
let Ok(decoded) = sessions.decode_inbound(&wire, Instant::now()) else {
continue;
};
let DecodeInboundOutput::AppMessage {
exchange_id,
opcode,
..
} = decoded
else {
continue; };
if opcode != 0x03 {
continue; }
let prime = build_report_data(1, 0x06, 0x0000, &matter_codec::Value::Bool(true));
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x05,
ProtocolId::INTERACTION_MODEL,
&prime,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
let sub_resp = build_subscribe_response(wire_ids[cycle.min(1)], 0);
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x04,
ProtocolId::INTERACTION_MODEL,
&sub_resp,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
cycle += 1;
if cycle >= 2 {
let _ = tokio::time::timeout(std::time::Duration::from_millis(200), io.recv_from())
.await;
return;
}
}
}
struct Harness {
store: Arc<MemStore>,
ctrl_io: InMemoryDatagram,
dev_io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
discovery: FixedDiscovery,
device_creds: CaseCredentials,
device_roots: TrustedRoots,
device_node_id: u64,
}
fn loopback_harness() -> Harness {
let mut fabric = {
let cfg = FabricConfig {
fabric_id: 0x0102_0304_0506_0708,
rcac_id: 1,
commissioner_node_id: 1,
validity: (
MatterTime::from_unix_secs(1_700_000_000),
MatterTime::NO_EXPIRY,
),
issue_icac: false,
};
crate::fabric::create_fabric(&cfg, &SystemNocRng).unwrap()
};
let device_node_id: u64 = 0x0000_0000_0000_0042;
let device_record = fabric.to_fabric_record().unwrap();
let (device_signer, _pkcs8) = RingSigner::generate().unwrap();
let device_pubkey = *device_signer.public_key().as_bytes();
fabric.devices.push(crate::state::DeviceEntry {
node_id: device_node_id,
peer_noc_public_key: device_pubkey,
resumption_record: None,
last_known_addr: None,
vendor_id: None,
product_id: None,
label: None,
});
let device_noc = issue_noc(
&device_record,
&VerifiedCsr {
public_key: device_signer.public_key().clone(),
},
device_node_id,
&[],
(
MatterTime::from_unix_secs(1_700_000_000),
MatterTime::NO_EXPIRY,
),
&SystemNocRng,
)
.unwrap();
let compressed =
derive_compressed_fabric_id(fabric.rcac_cert.public_key().as_bytes(), fabric.fabric_id)
.unwrap();
let device_ipk = derive_operational_ipk(&fabric.ipk, &compressed).unwrap();
let mut device_roots = TrustedRoots::new();
device_roots.add(TrustAnchor::from_root_cert(&fabric.rcac_cert));
let device_creds = CaseCredentials {
noc: device_noc,
icac: None,
signer: Box::new(device_signer),
fabric_id: fabric.fabric_id,
node_id: device_node_id,
ipk: device_ipk,
rcac_public_key: *fabric.rcac_cert.public_key().as_bytes(),
};
let store = Arc::new(MemStore::default());
store
.save(
&crate::snapshot::serialize(&ControllerState {
fabrics: vec![fabric],
})
.unwrap(),
)
.unwrap();
let (ctrl_io, dev_io) = InMemoryDatagram::pair();
let ctrl_addr = ctrl_io.local_addr();
let dev_addr = dev_io.local_addr();
let discovery = FixedDiscovery {
addr: dev_addr,
instance_name: operational_instance_name(compressed, device_node_id),
};
Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
}
}
#[tokio::test]
async fn connects_caches_and_round_trips_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
2,
b"pong".to_vec(),
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let resp1 = node
.round_trip(0x02, ProtocolId::INTERACTION_MODEL, b"ping".to_vec())
.await
.expect("first round-trip");
assert_eq!(resp1, b"pong");
assert_eq!(controller.session_count().await, 1, "session cached");
let resp2 = node
.round_trip(0x02, ProtocolId::INTERACTION_MODEL, b"ping".to_vec())
.await
.expect("second round-trip");
assert_eq!(resp2, b"pong");
assert_eq!(
controller.session_count().await,
1,
"still one session — reused, not re-established"
);
let record = controller
.resumption_record_for(device_node_id)
.await
.expect("fetch resumption record")
.expect("connect must persist a resumption record");
assert_eq!(record.peer.node_id, device_node_id);
device.await.unwrap();
}
#[tokio::test]
async fn actor_stays_live_while_resolve_pends() {
const UNRESOLVABLE_NODE: u64 = 0x0000_0000_0000_0099;
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
1,
b"pong".to_vec(),
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let unresolvable = controller.node(UNRESOLVABLE_NODE);
let mut pending_verb = Box::pin(unresolvable.round_trip(
0x02,
ProtocolId::INTERACTION_MODEL,
b"ping".to_vec(),
));
assert!(
tokio::time::timeout(std::time::Duration::from_millis(150), &mut pending_verb)
.await
.is_err(),
"the parked resolve must not have failed yet"
);
let node = controller.node(device_node_id);
let resp = tokio::time::timeout(
std::time::Duration::from_secs(2),
node.round_trip(0x02, ProtocolId::INTERACTION_MODEL, b"ping".to_vec()),
)
.await
.expect("actor stayed live while the other node's resolve was parked")
.expect("loopback round-trip");
assert_eq!(resp, b"pong");
assert!(
tokio::time::timeout(std::time::Duration::from_millis(50), &mut pending_verb)
.await
.is_err(),
"the parked resolve must outlive the other session's traffic"
);
let err = tokio::time::timeout(RESOLVE_DEADLINE * 3, pending_verb)
.await
.expect("the parked resolve must expire at its deadline")
.expect_err("an unresolvable node must fail its waiters");
let msg = err.to_string();
assert!(
msg.contains("not found via mDNS"),
"expiry must report the mDNS-not-found error, got: {msg}"
);
device.await.unwrap();
}
#[tokio::test]
async fn record_drained_before_its_resolve_parks_is_not_lost() {
const OFFLINE_NODE: u64 = 0x0000_0000_0000_0099;
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let discovery = DrainingDiscovery {
addr: discovery.addr,
instance_name: discovery.instance_name,
drained: false,
};
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
1,
b"pong".to_vec(),
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let offline = controller.node(OFFLINE_NODE);
let mut parked_verb =
Box::pin(offline.round_trip(0x02, ProtocolId::INTERACTION_MODEL, b"ping".to_vec()));
assert!(
tokio::time::timeout(std::time::Duration::from_millis(150), &mut parked_verb)
.await
.is_err(),
"the offline node's resolve must still be parked, holding the browse open"
);
let node = controller.node(device_node_id);
let resp = tokio::time::timeout(
std::time::Duration::from_secs(2),
node.round_trip(0x02, ProtocolId::INTERACTION_MODEL, b"ping".to_vec()),
)
.await
.expect("an online device must not be starved by a record drained before it parked")
.expect("loopback round-trip");
assert_eq!(resp, b"pong");
device.await.unwrap();
}
#[test]
fn route_key_unifies_mapped_ipv4_and_strips_scope() {
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV6};
let resolved: SocketAddr = (Ipv4Addr::new(192, 0, 2, 7), 5540).into();
let arrived_mapped = SocketAddr::new(
IpAddr::V6(Ipv4Addr::new(192, 0, 2, 7).to_ipv6_mapped()),
5540,
);
assert_eq!(route_key(resolved), route_key(arrived_mapped));
assert_eq!(
route_key(resolved),
resolved,
"canonical form is the IPv4 one"
);
let ll = Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 0x1d42);
let resolved_v6 = SocketAddr::V6(SocketAddrV6::new(ll, 5540, 0, 0));
let arrived_scoped = SocketAddr::V6(SocketAddrV6::new(ll, 5540, 0, 7));
assert_eq!(route_key(resolved_v6), route_key(arrived_scoped));
let a: SocketAddr = (Ipv4Addr::LOCALHOST, 5540).into();
let b: SocketAddr = (Ipv4Addr::LOCALHOST, 5541).into();
assert_ne!(route_key(a), route_key(b));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn connect_handshake_runs_off_loop_which_stays_responsive() {
let Harness {
store,
ctrl_io,
dev_io,
discovery,
device_node_id,
..
} = loopback_harness();
let device = tokio::spawn(async move {
let _ = dev_io.recv_from().await; tokio::time::sleep(std::time::Duration::from_secs(3)).await; });
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let parked = tokio::spawn(async move {
let _ = node
.round_trip(0x02, ProtocolId::INTERACTION_MODEL, b"ping".to_vec())
.await;
});
let count = tokio::time::timeout(
std::time::Duration::from_secs(1),
controller.session_count(),
)
.await
.expect("session_count must return while a connect handshake is stalled");
assert_eq!(count, 0, "the stalled connect established no session");
parked.abort();
device.abort();
}
#[tokio::test]
async fn read_verb_returns_report_data_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let report_blob = build_report_data(1, 0x06, 0x0000, &matter_codec::Value::Bool(true));
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
1,
report_blob,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let report = node
.read(&[matter_interaction::ReadPath::concrete(1, 0x06, 0x0000)])
.await
.expect("read");
assert_eq!(report.len(), 1);
let (path, value) = &report[0];
assert_eq!(path.endpoint, 1);
assert_eq!(path.cluster, 0x06);
assert_eq!(path.attribute, 0x0000);
assert_eq!(*value, matter_codec::Value::Bool(true));
device.await.unwrap();
}
#[tokio::test]
async fn read_events_returns_event_report_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let report_blob = build_report_data_event(0, 0x28, 0x00, 1, &matter_codec::Value::Uint(7));
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
1,
report_blob,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let events = node
.read_events(
&[matter_interaction::EventPath::concrete(0, 0x28, 0x00)],
&[],
)
.await
.expect("read_events");
assert_eq!(events.len(), 1);
match &events[0] {
matter_interaction::EventReport::Data(it) => {
assert_eq!(it.path.endpoint, Some(0));
assert_eq!(it.path.cluster, Some(0x28));
assert_eq!(it.path.event, Some(0x00));
assert_eq!(it.event_number, 1);
assert_eq!(it.value, matter_codec::Value::Uint(7));
}
other => panic!("expected EventReport::Data, got {other:?}"),
}
device.await.unwrap();
}
#[tokio::test]
async fn write_timed_does_handshake_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let resp = {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.start_array(Tag::Context(0)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(2), 0).unwrap();
w.put_uint(Tag::Context(3), 0x28).unwrap();
w.put_uint(Tag::Context(4), 0x05).unwrap();
w.end_container().unwrap();
w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), 0).unwrap(); w.end_container().unwrap();
w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
};
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D8,
1,
resp,
true, ));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let statuses = node
.write_timed(
&[(
matter_interaction::AttributePath {
endpoint: 0,
cluster: 0x28,
attribute: 0x05,
},
matter_codec::Value::Utf8("x".to_string()),
)],
None,
)
.await
.expect("write_timed");
assert_eq!(statuses.len(), 1);
assert_eq!(statuses[0].1, matter_interaction::ImStatus::Success);
device.await.unwrap();
}
#[tokio::test]
async fn write_auto_upgrades_and_caches_timed() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let resp = {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.start_array(Tag::Context(0)).unwrap();
w.start_structure(Tag::Anonymous).unwrap();
w.start_list(Tag::Context(0)).unwrap();
w.put_uint(Tag::Context(2), 0).unwrap();
w.put_uint(Tag::Context(3), 0x28).unwrap();
w.put_uint(Tag::Context(4), 0x05).unwrap();
w.end_container().unwrap();
w.start_structure(Tag::Context(1)).unwrap();
w.put_uint(Tag::Context(0), 0).unwrap();
w.end_container().unwrap();
w.end_container().unwrap();
w.end_container().unwrap();
w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
};
let device = tokio::spawn(run_timed_retry_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D9,
resp,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let path = matter_interaction::AttributePath {
endpoint: 0,
cluster: 0x28,
attribute: 0x05,
};
let s1 = node
.write(&[(path, matter_codec::Value::Utf8("a".to_string()))])
.await
.expect("write 1 (auto-upgrade)");
assert_eq!(s1[0].1, matter_interaction::ImStatus::Success);
let s2 = node
.write(&[(path, matter_codec::Value::Utf8("b".to_string()))])
.await
.expect("write 2 (cached timed)");
assert_eq!(s2[0].1, matter_interaction::ImStatus::Success);
device.await.unwrap();
}
#[tokio::test]
async fn read_reassembles_chunked_report_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let chunk0 =
build_report_data_chunk(0, 0x28, 0x0002, &matter_codec::Value::Uint(5010), true);
let chunk1 =
build_report_data_chunk(1, 0x06, 0x0000, &matter_codec::Value::Bool(true), false);
let device = tokio::spawn(run_chunked_read_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
chunk0,
chunk1,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let report = node
.read(&[matter_interaction::ReadPath::all()])
.await
.expect("chunked read");
assert_eq!(report.len(), 2, "both chunks reassembled");
assert_eq!(report[0].0.endpoint, 0);
assert_eq!(report[0].1, matter_codec::Value::Uint(5010));
assert_eq!(report[1].0.endpoint, 1);
assert_eq!(report[1].0.cluster, 0x06);
assert_eq!(report[1].1, matter_codec::Value::Bool(true));
device.await.unwrap();
}
#[test]
fn push_parsed_matches_byte_path() {
let c0 = build_report_data_chunk(0, 0x28, 0x0002, &matter_codec::Value::Uint(5010), true);
let c1 = build_report_data_chunk(1, 0x06, 0x0000, &matter_codec::Value::Bool(true), false);
let mut bytes_path = ReportReassembler::default();
assert!(bytes_path.push(&c0).is_none());
let via_bytes = bytes_path.push(&c1).expect("final chunk flushes");
let mut parsed_path = ReportReassembler::default();
let rd0 = matter_interaction::parse_report_data(&c0).expect("parse chunk 0");
let rd1 = matter_interaction::parse_report_data(&c1).expect("parse chunk 1");
assert!(parsed_path.push_parsed(rd0).is_none());
let via_parsed = parsed_path.push_parsed(rd1).expect("final chunk flushes");
assert_eq!(
via_parsed, via_bytes,
"single-parse path is content-identical"
);
assert_eq!(via_parsed.len(), 2);
assert_eq!(via_parsed[0].0.endpoint, 0);
assert_eq!(via_parsed[0].1, matter_codec::Value::Uint(5010));
assert_eq!(via_parsed[1].0.endpoint, 1);
assert_eq!(via_parsed[1].1, matter_codec::Value::Bool(true));
}
#[test]
fn reassembler_flushes_only_on_final_chunk() {
let mut r = ReportReassembler::default();
let c0 = build_report_data_chunk(0, 0x28, 0x0002, &matter_codec::Value::Uint(5010), true);
assert!(r.push(&c0).is_none(), "non-final chunk must not flush");
let c1 = build_report_data_chunk(1, 0x06, 0x0000, &matter_codec::Value::Bool(true), false);
let merged = r.push(&c1).expect("final chunk flushes");
assert_eq!(merged.len(), 2);
assert_eq!(merged[0].0.endpoint, 0);
assert_eq!(merged[1].0.endpoint, 1);
}
#[test]
fn reassembler_single_message_flushes_immediately() {
let mut r = ReportReassembler::default();
let only = build_report_data(1, 0x06, 0x0000, &matter_codec::Value::Bool(true));
let merged = r
.push(&only)
.expect("single-message report flushes at once");
assert_eq!(merged.len(), 1);
assert_eq!(merged[0].0.cluster, 0x06);
}
#[test]
fn reassembler_drops_runaway_sequence() {
let mut r = ReportReassembler::default();
let runaway = build_report_data_chunk(0, 0x28, 0x0002, &matter_codec::Value::Uint(1), true);
for _ in 0..=MAX_SUB_CHUNKS {
assert!(r.push(&runaway).is_none(), "non-final chunk never flushes");
}
let last =
build_report_data_chunk(1, 0x06, 0x0000, &matter_codec::Value::Bool(true), false);
let merged = r.push(&last).expect("final chunk flushes");
assert_eq!(merged.len(), 1, "runaway partial was dropped, not merged");
assert_eq!(merged[0].0.cluster, 0x06);
}
#[test]
fn fibonacci_matches_chip_sequence() {
assert_eq!(fibonacci(0), 0);
assert_eq!(fibonacci(1), 1);
assert_eq!(fibonacci(2), 1);
assert_eq!(fibonacci(3), 2);
assert_eq!(fibonacci(5), 5);
assert_eq!(fibonacci(14), 377);
}
#[test]
fn resubscribe_backoff_respects_chip_bounds() {
let rng = SystemNocRng;
assert_eq!(resubscribe_backoff(&rng, 0), std::time::Duration::ZERO);
for _ in 0..32 {
let d = u64::try_from(resubscribe_backoff(&rng, 3).as_millis()).unwrap();
assert!(
(6_000..=20_000).contains(&d),
"n=3 wait {d} out of [6000,20000]"
);
}
for _ in 0..32 {
let d = u64::try_from(resubscribe_backoff(&rng, 99).as_millis()).unwrap();
assert!(
(1_661_400..=5_538_000).contains(&d),
"n=99 wait {d} out of cap band"
);
}
}
#[tokio::test]
async fn subscribe_streams_reports_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_subscription_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
vec![build_report_data(1, 0x06, 0x0000, &matter_codec::Value::Bool(true)); 3],
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let mut sub = node
.subscribe(
&[matter_interaction::ReadPath::concrete(1, 0x06, 0x0000)],
&[],
1,
30,
)
.await
.expect("subscribe");
match sub.next().await {
Some(SubscriptionEvent::Established { subscription_id }) => {
assert_eq!(subscription_id, 0x1234_5678);
}
other => panic!("expected Established, got {other:?}"),
}
for _ in 0..3 {
let Some(SubscriptionEvent::Report(report)) = sub.next().await else {
panic!("expected a Report event");
};
assert_eq!(report.path.endpoint, 1);
assert_eq!(report.path.cluster, 0x06);
assert_eq!(report.value, matter_codec::Value::Bool(true));
}
device.await.unwrap();
sub.cancel().await.expect("cancel");
}
#[tokio::test]
async fn subscribe_reassembles_list_append_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let chunk0 = build_report_data_chunk(
1,
0x1d,
0x0003,
&matter_codec::Value::Array(Vec::new()),
true,
);
let chunk1 =
build_report_data_append(1, 0x1d, 0x0003, &matter_codec::Value::Uint(7), false);
let device = tokio::spawn(run_subscription_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D6,
vec![chunk0, chunk1],
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let mut sub = node
.subscribe(
&[matter_interaction::ReadPath::cluster(1, 0x1d)],
&[],
1,
30,
)
.await
.expect("subscribe");
match sub.next().await {
Some(SubscriptionEvent::Established { .. }) => {}
other => panic!("expected Established, got {other:?}"),
}
let Some(SubscriptionEvent::Report(report)) = sub.next().await else {
panic!("expected the merged list Report");
};
assert_eq!(report.path.endpoint, 1);
assert_eq!(report.path.cluster, 0x1d);
assert_eq!(report.path.attribute, 0x0003);
assert_eq!(
report.value,
matter_codec::Value::Array(vec![matter_codec::Value::Uint(7)]),
"list-append must reassemble into the full list"
);
device.await.unwrap();
sub.cancel().await.expect("cancel");
}
#[tokio::test]
async fn subscribe_streams_event_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let event_blob = build_report_data_event(0, 0x28, 0x00, 1, &matter_codec::Value::Uint(7));
let device = tokio::spawn(run_subscription_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D7,
vec![event_blob],
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let mut sub = node
.subscribe(
&[matter_interaction::ReadPath::cluster(1, 0x06)],
&[matter_interaction::EventPath::cluster(0, 0x28)],
1,
30,
)
.await
.expect("subscribe");
match sub.next().await {
Some(SubscriptionEvent::Established { .. }) => {}
other => panic!("expected Established, got {other:?}"),
}
loop {
match sub.next().await {
Some(SubscriptionEvent::Event(matter_interaction::EventReport::Data(it))) => {
assert_eq!(it.path.endpoint, Some(0));
assert_eq!(it.path.cluster, Some(0x28));
assert_eq!(it.path.event, Some(0x00));
assert_eq!(it.event_number, 1);
assert_eq!(it.value, matter_codec::Value::Uint(7));
break;
}
Some(_) => {}
None => panic!("subscription ended before an event arrived"),
}
}
device.await.unwrap();
sub.cancel().await.expect("cancel");
}
#[tokio::test]
async fn concurrent_round_trip_does_not_drop_subscription_report() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_concurrent_sub_roundtrip_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let mut sub = node
.subscribe(
&[matter_interaction::ReadPath::concrete(1, 0x06, 0x0000)],
&[],
1,
30,
)
.await
.expect("subscribe");
match sub.next().await {
Some(SubscriptionEvent::Established { .. }) => {}
other => panic!("expected Established, got {other:?}"),
}
let resp = node
.round_trip(0x02, ProtocolId::INTERACTION_MODEL, b"ping".to_vec())
.await
.expect("round-trip completes");
assert_eq!(resp, b"pong");
let event = tokio::time::timeout(std::time::Duration::from_secs(5), sub.next())
.await
.expect("steady report must arrive (not dropped by the concurrent round-trip)")
.expect("subscription still live");
let SubscriptionEvent::Report(report) = event else {
panic!("expected a Report event, got {event:?}");
};
assert_eq!(report.path.endpoint, 1);
assert_eq!(report.path.cluster, 0x06);
assert_eq!(report.value, matter_codec::Value::Bool(true));
device.await.unwrap();
sub.cancel().await.expect("cancel");
}
#[tokio::test]
async fn liveness_timeout_triggers_resubscribe() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_resubscribe_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00D2,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let mut sub = node
.subscribe(
&[matter_interaction::ReadPath::concrete(1, 0x06, 0x0000)],
&[],
1,
0,
)
.await
.expect("subscribe");
let mut establishes = 0u32;
let mut saw_resubscribing = false;
let mut reprimed_after_resub = false;
let overall = tokio::time::Instant::now() + std::time::Duration::from_secs(25);
while tokio::time::Instant::now() < overall
&& !(saw_resubscribing && establishes >= 2 && reprimed_after_resub)
{
match tokio::time::timeout(std::time::Duration::from_secs(15), sub.next()).await {
Ok(Some(SubscriptionEvent::Established { .. })) => establishes += 1,
Ok(Some(SubscriptionEvent::Resubscribing { .. })) => saw_resubscribing = true,
Ok(Some(SubscriptionEvent::Report(_))) => {
if saw_resubscribing {
reprimed_after_resub = true;
}
}
Ok(Some(_)) => {}
Ok(None) | Err(_) => break,
}
}
assert!(saw_resubscribing, "expected a Resubscribing event");
assert!(
establishes >= 2,
"expected a second Established after resubscribe, saw {establishes}"
);
assert!(
reprimed_after_resub,
"expected a re-primed Report after the resubscribe"
);
let _ = device.await;
sub.cancel().await.ok();
}
#[test]
fn resubscribe_stranded_moves_only_subs_on_the_replaced_session() {
let (io, _peer) = InMemoryDatagram::pair();
let mut actor = Actor::new(
io,
NullDiscovery,
Arc::new(MemStore::default()),
Arc::new(matter_commissioning::SystemNocRng),
ControllerState { fabrics: vec![] },
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
);
let peer: SocketAddr = "127.0.0.1:5540".parse().unwrap();
let mk = |tx, session_id| SubEntry {
tx,
peer,
reassembler: ReportReassembler::default(),
session_id,
wire_sub_id: 0x1234,
node_id: 2,
paths: vec![matter_interaction::ReadPath::all()],
event_paths: vec![],
event_filters: vec![],
min_interval: 1,
max_interval: 30,
liveness_deadline: Instant::now() + std::time::Duration::from_secs(60),
};
let (sink_a, _report_rx_a, mut rx_a) = test_report_sink();
let (sink_b, _report_rx_b, mut rx_b) = test_report_sink();
actor.insert_subscription(SubId(1), mk(sink_a, SessionId(7)));
actor.insert_subscription(SubId(2), mk(sink_b, SessionId(9)));
actor.resubscribe_stranded(SessionId(7));
assert!(
!actor.subscriptions.contains_key(&SubId(1)),
"stranded sub removed from the active map"
);
assert!(
actor.resubscribes.iter().any(|pr| pr.sub_id == SubId(1)),
"stranded sub scheduled for resubscribe"
);
assert!(
matches!(rx_a.try_recv(), Ok(SubscriptionEvent::Resubscribing { .. })),
"consumer notified with Resubscribing"
);
assert!(actor.subscriptions.contains_key(&SubId(2)));
assert!(!actor.resubscribes.iter().any(|pr| pr.sub_id == SubId(2)));
assert!(rx_b.try_recv().is_err(), "unaffected sub gets no event");
}
#[tokio::test]
async fn dropped_subscription_reaps_pending_resubscribe() {
let (io, _peer) = InMemoryDatagram::pair();
let mut actor = Actor::new(
io,
NullDiscovery,
Arc::new(MemStore::default()),
Arc::new(matter_commissioning::SystemNocRng),
ControllerState { fabrics: vec![] },
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
);
let (sink, report_rx, ctrl_rx) = test_report_sink();
drop(report_rx);
drop(ctrl_rx);
actor.resubscribes.push(PendingResubscribe {
sub_id: SubId(1),
attempt_at: Instant::now()
.checked_sub(std::time::Duration::from_secs(1))
.expect("instant minus 1s is representable"),
node_id: 2,
paths: vec![matter_interaction::ReadPath::all()],
event_paths: vec![],
event_filters: vec![],
min_interval: 1,
max_interval: 30,
retry_count: 0,
tx: sink,
});
actor.drive_resubscribes().await;
assert!(
actor.resubscribes.is_empty(),
"zombie resubscribe entry must be reaped, not rescheduled"
);
assert!(
actor.pending_connects.is_empty(),
"no connect should be enqueued for a consumer that is gone"
);
}
fn actor_with_one_fabric() -> Actor<InMemoryDatagram, NullDiscovery> {
let (io, _peer) = InMemoryDatagram::pair();
let fabric = {
let cfg = FabricConfig {
fabric_id: 0x0A0B_0C0D_0E0F_1011,
rcac_id: 1,
commissioner_node_id: 1,
validity: (
MatterTime::from_unix_secs(1_700_000_000),
MatterTime::NO_EXPIRY,
),
issue_icac: false,
};
crate::fabric::create_fabric(&cfg, &SystemNocRng).unwrap()
};
Actor::new(
io,
NullDiscovery,
Arc::new(MemStore::default()),
Arc::new(SystemNocRng),
ControllerState {
fabrics: vec![fabric],
},
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
}
#[test]
fn peer_for_session_uses_stamped_addr_and_dies_with_the_session() {
use matter_crypto::pase::PaseSessionKeys;
let mut actor = actor_with_one_fabric();
let keys = PaseSessionKeys {
ke: [0u8; 16],
i2r_key: [1u8; 16],
r2i_key: [2u8; 16],
attestation_key: [3u8; 16],
};
let sid = actor.sessions.register_pase(
keys,
SessionRole::Initiator,
1,
matter_transport::PeerHint::default(),
);
let peer: SocketAddr = "[::1]:5540".parse().unwrap();
if let Some(s) = actor.sessions.get_mut(sid) {
s.peer_addr = Some(peer);
}
assert_eq!(actor.peer_for_session(sid), Some(peer));
actor.sessions.remove(sid);
assert_eq!(actor.peer_for_session(sid), None);
}
fn seed_pending_round_trip(
actor: &mut Actor<InMemoryDatagram, NullDiscovery>,
session: SessionId,
exchange: u16,
node_id: u64,
) {
let (reply_tx, _reply_rx) = oneshot::channel();
actor.pending.insert(
(session, exchange),
Pending {
node_id,
peer: "127.0.0.1:5540".parse().unwrap(),
request: PendingRequest {
opcode: 0x02,
protocol_id: ProtocolId::INTERACTION_MODEL,
payload: vec![],
},
retried: false,
reply: PendingReply::RoundTrip(reply_tx),
},
);
}
#[tokio::test]
async fn late_timeout_on_superseded_session_does_not_evict_current_session() {
let mut actor = actor_with_one_fabric();
let fabric_id = actor.sole_fabric().unwrap().fabric_id;
let node_id = 0x42u64;
let old_session = SessionId(7);
let new_session = SessionId(9);
actor.cache.insert(
(fabric_id, node_id),
CachedSession {
session_id: new_session,
peer: "127.0.0.1:5540".parse().unwrap(),
},
);
seed_pending_round_trip(&mut actor, old_session, 0xABCD, node_id);
actor.on_pending_timeout(old_session, 0xABCD).await;
let cached = actor
.cache
.get(&(fabric_id, node_id))
.expect("current healthy session must remain cached");
assert_eq!(
cached.session_id, new_session,
"late timeout on a superseded session must not evict the current session"
);
assert!(
actor.resubscribes.is_empty(),
"no resubscribe churn should be scheduled by a superseded-session timeout"
);
}
#[tokio::test]
async fn timeout_on_current_session_evicts_it() {
let mut actor = actor_with_one_fabric();
let fabric_id = actor.sole_fabric().unwrap().fabric_id;
let node_id = 0x42u64;
let session = SessionId(7);
actor.cache.insert(
(fabric_id, node_id),
CachedSession {
session_id: session,
peer: "127.0.0.1:5540".parse().unwrap(),
},
);
seed_pending_round_trip(&mut actor, session, 0xABCD, node_id);
actor.on_pending_timeout(session, 0xABCD).await;
assert!(
!actor.cache.contains_key(&(fabric_id, node_id)),
"timeout on the current session must evict it so genuine reconnect happens"
);
}
fn mk_report(seq: usize) -> AttributeReport {
AttributeReport {
path: matter_interaction::AttributePath {
endpoint: 1,
cluster: 0x06,
attribute: u32::try_from(seq).unwrap_or(u32::MAX),
},
value: matter_codec::Value::Bool(true),
}
}
#[tokio::test]
async fn report_overflow_drops_and_surfaces_lagged_without_blocking() {
let (mut sink, mut report_rx, _ctrl_rx) = test_report_sink();
let overflow = 100usize;
let total = SUBSCRIPTION_CHANNEL_CAP + overflow;
for i in 0..total {
assert!(
sink.try_send_report(mk_report(i)),
"actor must never block or fail on a full buffer (live consumer)"
);
}
assert_eq!(
sink.dropped, overflow,
"exactly the over-capacity reports were dropped + counted"
);
assert_eq!(
report_rx.len(),
SUBSCRIPTION_CHANNEL_CAP,
"buffered reports are bounded by the channel capacity"
);
let first = report_rx.try_recv().expect("a buffered report");
assert!(matches!(first, SubscriptionEvent::Report(_)));
let _ = report_rx.try_recv().expect("a buffered report");
assert!(
sink.try_send_report(mk_report(9999)),
"post-drain send still succeeds"
);
assert_eq!(
sink.dropped, 0,
"Lagged flush cleared the dropped counter and the new report fit"
);
let mut saw_lagged = None;
let mut reports = 1usize; while let Ok(ev) = report_rx.try_recv() {
match ev {
SubscriptionEvent::Lagged { dropped } => {
assert!(saw_lagged.is_none(), "drops are coalesced into one Lagged");
saw_lagged = Some(dropped);
}
SubscriptionEvent::Report(_) => reports += 1,
other => panic!("unexpected event on report channel: {other:?}"),
}
}
assert_eq!(
saw_lagged,
Some(overflow),
"a single Lagged surfaced the exact dropped count"
);
assert!(
reports < total,
"the flood was bounded: delivered fewer reports than were sent"
);
}
#[tokio::test]
async fn report_send_reports_consumer_gone_when_receiver_dropped() {
let (mut sink, report_rx, _ctrl_rx) = test_report_sink();
drop(report_rx);
assert!(
!sink.try_send_report(mk_report(0)),
"a closed report receiver signals the consumer is gone"
);
}
#[tokio::test]
async fn control_events_delivered_even_when_report_channel_saturated() {
let (mut sink, report_rx, ctrl_rx) = test_report_sink();
for i in 0..(SUBSCRIPTION_CHANNEL_CAP + 50) {
assert!(sink.try_send_report(mk_report(i)));
}
assert!(
sink.send_control(SubscriptionEvent::Established {
subscription_id: 0xABCD,
}),
"Established must be delivered under report backpressure"
);
assert!(
sink.send_control(SubscriptionEvent::Resubscribing {
cause: Error::ControllerStopped,
}),
"Resubscribing must be delivered under report backpressure"
);
let (cmd_tx, _cmd_rx) = mpsc::channel::<Command>(8);
let mut sub = crate::subscription::Subscription {
rx: report_rx,
ctrl_rx,
tx: cmd_tx,
key: SubId(1),
cancelled: true, };
match sub.next().await {
Some(SubscriptionEvent::Established { subscription_id }) => {
assert_eq!(subscription_id, 0xABCD);
}
other => panic!("expected Established first, got {other:?}"),
}
match sub.next().await {
Some(SubscriptionEvent::Resubscribing { .. }) => {}
other => panic!("expected Resubscribing second, got {other:?}"),
}
match sub.next().await {
Some(SubscriptionEvent::Report(_)) => {}
other => panic!("expected a buffered Report next, got {other:?}"),
}
}
#[derive(Default)]
struct FailingStore;
impl ControllerStore for FailingStore {
fn load(&self) -> Result<Option<Vec<u8>>, crate::store::StoreError> {
Ok(None)
}
fn save(&self, _snapshot: &[u8]) -> Result<(), crate::store::StoreError> {
Err(crate::store::StoreError::Io(std::io::Error::other(
"disk full",
)))
}
}
#[derive(Default)]
struct BlockingStore {
inner: std::sync::Mutex<Option<Vec<u8>>>,
saves: std::sync::atomic::AtomicUsize,
gate: std::sync::Mutex<()>,
}
impl ControllerStore for BlockingStore {
fn load(&self) -> Result<Option<Vec<u8>>, crate::store::StoreError> {
Ok(self.inner.lock().unwrap().clone())
}
fn save(&self, snapshot: &[u8]) -> Result<(), crate::store::StoreError> {
let _held = self.gate.lock().unwrap();
self.saves.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
*self.inner.lock().unwrap() = Some(snapshot.to_vec());
Ok(())
}
}
#[tokio::test]
async fn durable_persist_surfaces_store_error() {
let store: Arc<dyn ControllerStore> = Arc::new(FailingStore);
let (io, _peer) = InMemoryDatagram::pair();
let controller = crate::controller::MatterController::with_components(
store,
io,
NullDiscovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let err = controller
.create_fabric(cfg())
.await
.expect_err("a failing store must fail create_fabric");
let msg = format!("{err}");
assert!(
msg.contains("disk full") || msg.to_lowercase().contains("i/o"),
"expected the store I/O error to propagate, got: {msg}"
);
}
fn test_actor(store: Arc<dyn ControllerStore>) -> Actor<InMemoryDatagram, NullDiscovery> {
let (io, _peer) = InMemoryDatagram::pair();
Actor::new(
io,
NullDiscovery,
store,
Arc::new(SystemNocRng),
ControllerState { fabrics: vec![] },
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn best_effort_persist_does_not_block_on_fsync() {
let store = Arc::new(BlockingStore::default());
let mut actor = test_actor(store.clone());
let held = store.gate.lock().unwrap();
let seq_before = actor.snapshot_seq;
let start = std::time::Instant::now();
actor.persist_best_effort();
assert!(
start.elapsed() < std::time::Duration::from_millis(500),
"best-effort persist must not block on the fsync"
);
assert_eq!(
actor.snapshot_seq,
seq_before + 1,
"persist_best_effort must advance the actor's snapshot sequence"
);
assert_eq!(store.saves.load(std::sync::atomic::Ordering::SeqCst), 0);
drop(held);
let mut ran = false;
for _ in 0..200 {
if store.saves.load(std::sync::atomic::Ordering::SeqCst) >= 1 {
ran = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(ran, "the offloaded best-effort save must eventually run");
let mut gated = false;
for _ in 0..200 {
if *actor.save_gate.lock().unwrap() == actor.snapshot_seq {
gated = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(
gated,
"the detached best-effort job must share the actor's save gate \
(gate {}, expected {})",
*actor.save_gate.lock().unwrap(),
actor.snapshot_seq
);
}
#[tokio::test]
async fn durable_persist_inputs_offload_round_trip() {
let store = Arc::new(MemStore::default());
let mut actor = test_actor(store.clone());
let job = actor.durable_save_inputs().expect("serialize");
save_offloaded(job).await.expect("durable save ok");
assert!(
store.load().expect("load").is_some(),
"durable save must have written the snapshot"
);
let mut actor = test_actor(Arc::new(FailingStore));
let job = actor.durable_save_inputs().expect("serialize");
let err = save_offloaded(job)
.await
.expect_err("a failing store must surface its error");
assert!(
format!("{err}").to_lowercase().contains("disk full")
|| format!("{err}").to_lowercase().contains("i/o"),
"expected the store error to propagate, got: {err}"
);
}
#[test]
fn stale_snapshot_does_not_clobber_newer() {
let store = Arc::new(MemStore::default());
let gate = Arc::new(std::sync::Mutex::new(0u64));
let newer = SaveJob {
store: store.clone(),
bytes: b"B".to_vec(),
seq: 2,
gate: gate.clone(),
};
let stale = SaveJob {
store: store.clone(),
bytes: b"A".to_vec(),
seq: 1,
gate: gate.clone(),
};
newer.run().expect("newer save ok");
stale.run().expect("stale save is a no-op, not an error");
assert_eq!(
store.load().expect("load"),
Some(b"B".to_vec()),
"the older snapshot must not have clobbered the newer one"
);
assert_eq!(*gate.lock().unwrap(), 2, "the gate tracks the newest write");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn best_effort_after_durable_is_noop() {
let store = Arc::new(MemStore::default());
let mut actor = test_actor(store.clone());
let job = actor.durable_save_inputs().expect("serialize");
save_offloaded(job).await.expect("durable save ok");
let durable = store.load().expect("load");
assert!(durable.is_some(), "durable save must have written");
let stale = SaveJob {
store: store.clone(),
bytes: b"stale-snapshot".to_vec(),
seq: 0,
gate: actor.save_gate.clone(),
};
let ran = Arc::new(std::sync::atomic::AtomicBool::new(false));
let flag = ran.clone();
drop(tokio::task::spawn_blocking(move || {
let _ = stale.run();
flag.store(true, std::sync::atomic::Ordering::SeqCst);
}));
let mut done = false;
for _ in 0..200 {
if ran.load(std::sync::atomic::Ordering::SeqCst) {
done = true;
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
assert!(done, "the detached best-effort job must have run");
assert_eq!(
store.load().expect("load"),
durable,
"a stale best-effort save must not roll the store back"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn liveness_timer_fires_under_inbound_flood() {
let (io, peer) = InMemoryDatagram::pair();
let mut actor = Actor::new(
io,
NullDiscovery,
Arc::new(MemStore::default()),
Arc::new(SystemNocRng),
ControllerState { fabrics: vec![] },
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
);
let (sink, _report_rx, mut ctrl_rx) = test_report_sink();
actor.insert_subscription(
SubId(1),
SubEntry {
tx: sink,
peer: "127.0.0.1:5540".parse().unwrap(),
reassembler: ReportReassembler::default(),
session_id: SessionId(7),
wire_sub_id: 0x1234,
node_id: 2,
paths: vec![matter_interaction::ReadPath::all()],
event_paths: vec![],
event_filters: vec![],
min_interval: 1,
max_interval: 30,
liveness_deadline: Instant::now()
.checked_sub(std::time::Duration::from_secs(1))
.expect("instant minus 1s is representable"),
},
);
let (cmd_tx, cmd_rx) = mpsc::channel::<Command>(8);
let loop_handle = tokio::spawn(actor.run(cmd_rx));
let flood = tokio::spawn(async move {
loop {
if peer
.send_to(b"junk-datagram-pressure", peer.local_addr())
.await
.is_err()
{
break;
}
tokio::task::yield_now().await;
}
});
let got = tokio::time::timeout(std::time::Duration::from_secs(2), ctrl_rx.recv()).await;
flood.abort();
drop(cmd_tx); let _ = loop_handle.await;
assert!(
matches!(got, Ok(Some(SubscriptionEvent::Resubscribing { .. }))),
"liveness timer must fire under inbound flood (got {got:?})"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn commission_completion_drains_while_loop_stays_responsive() {
let actor = actor_with_one_fabric();
let fabric_id = actor.sole_fabric().unwrap().fabric_id;
let completion_tx = actor.commission_tx.clone();
let (cmd_tx, cmd_rx) = mpsc::channel::<Command>(8);
let loop_handle = tokio::spawn(actor.run(cmd_rx));
let (count_tx, count_rx) = oneshot::channel();
cmd_tx
.send(Command::SessionCount { reply: count_tx })
.await
.unwrap();
let count = tokio::time::timeout(std::time::Duration::from_secs(1), count_rx)
.await
.expect("the loop must service SessionCount while a commission is outstanding")
.expect("SessionCount reply");
assert_eq!(count, 0, "no sessions cached yet");
let (reply_tx, reply_rx) = oneshot::channel();
completion_tx
.send(CommissionCompletion {
fabric_id,
result: Err(Error::Operational("simulated commission failure".into())),
label: None,
reply: reply_tx,
})
.await
.unwrap();
let outcome = tokio::time::timeout(std::time::Duration::from_secs(1), reply_rx)
.await
.expect("the completions arm must resolve the commission reply")
.expect("reply channel");
assert!(
matches!(outcome, Err(Error::Operational(_))),
"the commission error must propagate to the caller (got {outcome:?})"
);
drop(cmd_tx);
let _ = loop_handle.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[allow(clippy::too_many_lines)] async fn commission_completion_persists_label_on_device_entry() {
let (io, _peer) = InMemoryDatagram::pair();
let fabric = {
let cfg = FabricConfig {
fabric_id: 0x0A0B_0C0D_0E0F_1011,
rcac_id: 1,
commissioner_node_id: 1,
validity: (
MatterTime::from_unix_secs(1_700_000_000),
MatterTime::NO_EXPIRY,
),
issue_icac: false,
};
crate::fabric::create_fabric(&cfg, &SystemNocRng).unwrap()
};
let fabric_id = fabric.fabric_id;
let fabric_record = fabric.to_fabric_record().expect("fabric record");
let actor = Actor::new(
io,
NullDiscovery,
Arc::new(MemStore::default()),
Arc::new(SystemNocRng),
ControllerState {
fabrics: vec![fabric],
},
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
);
let completion_tx = actor.commission_tx.clone();
let (cmd_tx, cmd_rx) = mpsc::channel::<Command>(8);
let loop_handle = tokio::spawn(actor.run(cmd_rx));
let commissioned = matter_commissioning::test_support::commissioned_fabric_for_test(
fabric_record,
2,
[0x04; 65],
);
let (reply_tx, reply_rx) = oneshot::channel();
completion_tx
.send(CommissionCompletion {
fabric_id,
result: Ok(commissioned),
label: Some("plug".to_string()),
reply: reply_tx,
})
.await
.unwrap();
let outcome = tokio::time::timeout(std::time::Duration::from_secs(1), reply_rx)
.await
.expect("the completions arm must resolve the commission reply")
.expect("reply channel");
let info = outcome.expect("commission reply");
assert_eq!(info.node_id, 2, "node id must be the assigned peer id");
assert_eq!(info.fabric_id, fabric_id, "NodeInfo carries the fabric id");
assert_eq!(
info.label,
Some("plug".to_string()),
"NodeInfo carries the label"
);
assert_eq!(
(info.vendor_id, info.product_id),
(None, None),
"vid/pid are filled by the controller's post-commission read, not the actor"
);
let (nodes_tx, nodes_rx) = oneshot::channel();
cmd_tx
.send(Command::ListNodes { reply: nodes_tx })
.await
.unwrap();
let nodes = nodes_rx.await.unwrap();
assert_eq!(nodes.len(), 1, "the device entry must have been pushed");
assert_eq!(
nodes[0].label,
Some("plug".to_string()),
"the caller-supplied label must be persisted on the device entry"
);
let (set_tx, set_rx) = oneshot::channel();
cmd_tx
.send(Command::SetNodeVidPid {
node_id: 2,
vendor_id: Some(0xFFF1),
product_id: Some(0x8000),
reply: set_tx,
})
.await
.unwrap();
set_rx.await.unwrap().expect("SetNodeVidPid persist");
let (nodes2_tx, nodes2_rx) = oneshot::channel();
cmd_tx
.send(Command::ListNodes { reply: nodes2_tx })
.await
.unwrap();
let nodes2 = nodes2_rx.await.unwrap();
assert_eq!(
(nodes2[0].vendor_id, nodes2[0].product_id),
(Some(0xFFF1), Some(0x8000)),
"SetNodeVidPid must persist vid/pid onto the device entry"
);
assert_eq!(
nodes2[0].label,
Some("plug".to_string()),
"SetNodeVidPid must not disturb the existing label"
);
let (miss_tx, miss_rx) = oneshot::channel();
cmd_tx
.send(Command::SetNodeVidPid {
node_id: 0xDEAD,
vendor_id: Some(1),
product_id: Some(2),
reply: miss_tx,
})
.await
.unwrap();
miss_rx
.await
.unwrap()
.expect("SetNodeVidPid for an unknown node is a no-op success");
drop(cmd_tx);
let _ = loop_handle.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn commission_dispatch_hands_off_without_blocking() {
let actor = actor_with_one_fabric();
let (cmd_tx, cmd_rx) = mpsc::channel::<Command>(8);
let loop_handle = tokio::spawn(actor.run(cmd_rx));
let (c_tx, c_rx) = oneshot::channel();
cmd_tx
.send(Command::Commission {
setup_payload: matter_commissioning::parse_manual_code("11693312331")
.expect("valid sample manual pairing code"),
label: None,
reply: c_tx,
})
.await
.unwrap();
let commission = tokio::time::timeout(std::time::Duration::from_secs(1), c_rx)
.await
.expect("Commission must be dispatched without blocking the loop")
.expect("commission reply");
assert!(
matches!(commission, Err(Error::NoTrust)),
"no trust configured → NoTrust (got {commission:?})"
);
let (count_tx, count_rx) = oneshot::channel();
cmd_tx
.send(Command::SessionCount { reply: count_tx })
.await
.unwrap();
let count = tokio::time::timeout(std::time::Duration::from_secs(1), count_rx)
.await
.expect("the loop must remain responsive after a Commission dispatch")
.expect("SessionCount reply");
assert_eq!(count, 0);
drop(cmd_tx);
let _ = loop_handle.await;
}
fn build_invoke_status_success() -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.put_bool(Tag::Context(0), false).unwrap();
w.start_array(Tag::Context(1)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_structure(Tag::Context(1)).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(0), 0).unwrap(); w.put_uint(
Tag::Context(1),
u64::from(crate::admin::ADMIN_COMMISSIONING_CLUSTER),
)
.unwrap(); w.put_uint(
Tag::Context(2),
u64::from(crate::admin::CMD_OPEN_COMMISSIONING_WINDOW),
)
.unwrap(); w.end_container().unwrap(); w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), 0x00).unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
}
#[tokio::test]
async fn open_commissioning_window_with_does_timed_invoke_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = build_invoke_status_success();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
true,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let node = controller.node(device_node_id);
let win = node
.open_commissioning_window_with(180, 20_202_021, &[0x01; 32], 3840, 1000, None, None)
.await
.expect("open window");
assert_eq!(win.passcode, 20_202_021);
assert_eq!(win.discriminator, 3840);
assert_eq!(win.manual_code.len(), 11);
assert!(win.qr_code.is_none());
device.await.unwrap();
}
#[tokio::test]
async fn open_basic_commissioning_window_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
build_invoke_status_success(),
true,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.open_basic_commissioning_window(180)
.await
.expect("open basic");
device.await.unwrap();
}
#[cfg(feature = "ota")]
#[tokio::test]
async fn announce_ota_provider_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = matter_interaction::build_invoke_response_status(
matter_interaction::CommandPath {
endpoint: 0,
cluster: 0x002A,
command: 0x00,
},
matter_interaction::ImStatus::Success,
);
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.announce_ota_provider(
0x1122_3344_5566_7788,
0xFFF1,
0,
)
.await
.expect("announce ota provider");
device.await.unwrap();
}
#[tokio::test]
async fn register_icd_client_over_loopback_persists_registration() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let resp_fields = {
use matter_codec::{Tag, TlvWriter};
let mut b = Vec::new();
let mut w = TlvWriter::new(&mut b);
w.start_structure(Tag::Anonymous).unwrap();
w.put_uint(Tag::Context(0), 42).unwrap();
w.end_container().unwrap();
b
};
let reply = matter_interaction::build_invoke_response_command(
matter_interaction::CommandPath {
endpoint: 0,
cluster: 0x0046,
command: 0x01,
},
&resp_fields,
);
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let reg = controller
.node(device_node_id)
.register_icd_client(1, crate::IcdClientType::Permanent)
.await
.expect("register_icd_client");
assert_eq!(reg.node_id, device_node_id);
assert_eq!(reg.start_counter, 42);
assert_eq!(reg.check_in_node_id, 1); assert_eq!(reg.monitored_subject, 1);
device.await.unwrap();
}
#[tokio::test]
async fn stay_active_request_over_loopback_returns_promised() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let resp_fields = {
use matter_codec::{Tag, TlvWriter};
let mut b = Vec::new();
let mut w = TlvWriter::new(&mut b);
w.start_structure(Tag::Anonymous).unwrap();
w.put_uint(Tag::Context(0), 5000).unwrap();
w.end_container().unwrap();
b
};
let reply = matter_interaction::build_invoke_response_command(
matter_interaction::CommandPath {
endpoint: 0,
cluster: 0x0046,
command: 0x04,
},
&resp_fields,
);
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let promised = controller
.node(device_node_id)
.stay_active_request(3000)
.await
.expect("stay_active_request");
assert_eq!(promised, 5000);
device.await.unwrap();
}
#[tokio::test]
async fn provider_server_accepts_case_and_dispatches_invoke_over_loopback() {
use crate::provider_server::ProviderServer;
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr: _,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let server = tokio::spawn(async move {
ProviderServer::new(
dev_io,
vec![device_creds],
device_roots,
0x55,
MatterTime::from_unix_secs(2_000_000_000),
)
.accept_and_dispatch_once(
|req: &matter_interaction::ParsedInvokeRequest| {
let path = req.commands[0].path;
matter_interaction::build_invoke_response_status(
path,
matter_interaction::ImStatus::Success,
)
},
1,
)
.await
});
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let path = matter_interaction::CommandPath {
endpoint: 0,
cluster: 0x0029,
command: 0x00,
};
let result = controller
.node(device_node_id)
.invoke(path, matter_codec::Value::Structure(vec![]))
.await
.expect("invoke");
assert!(matches!(
result,
crate::InvokeResult::Status(matter_interaction::ImStatus::Success)
));
let dispatched = server.await.unwrap().expect("server ok");
assert_eq!(dispatched, 1);
}
#[cfg(feature = "ota")]
#[tokio::test]
async fn serve_ota_once_full_flow_over_loopback() {
use crate::provider_server::ProviderServer;
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr: _,
discovery: _,
device_creds,
device_roots,
device_node_id: _,
} = loopback_harness();
let state = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let fabric = &state.fabrics[0];
let (provider_creds, provider_roots, _compressed) =
crate::credentials::operational_credentials(fabric).unwrap();
let provider_node_id = fabric.commissioner.node_id;
let fabric_id = fabric.fabric_id;
let image: Vec<u8> = (0..2500u32)
.map(|i| u8::try_from(i % 251).unwrap())
.collect();
let offer = matter_ota::ImageOffer {
software_version: 2,
software_version_string: "2.0".into(),
image_uri: format!("bdx://{provider_node_id:016X}/fw.ota"),
update_token: vec![0xAB; 16],
};
let provider_addr = dev_io.local_addr();
let image_for_assert = image.clone();
let sunk: std::sync::Arc<std::sync::Mutex<Vec<matter_crypto::ResumptionRecord>>> =
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sunk_in = sunk.clone();
let server = tokio::spawn(async move {
ProviderServer::new(
dev_io,
vec![provider_creds],
provider_roots,
0x55,
MatterTime::from_unix_secs(2_000_000_000),
)
.with_record_sink(Box::new(move |r| sunk_in.lock().unwrap().push(r)))
.serve_ota_once(offer, image, 256)
.await
});
let reassembled = ota_test_requestor(
ctrl_io,
provider_addr,
device_creds,
device_roots,
provider_node_id,
fabric_id,
1,
)
.await;
assert_eq!(
reassembled, image_for_assert,
"requestor reassembled the served image"
);
server.await.unwrap().expect("provider served OTA");
let records = sunk.lock().unwrap();
assert_eq!(
records.len(),
1,
"full-path accept must yield a resumption record to persist"
);
}
#[cfg(feature = "ota")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_ota_once_resumed_session_over_loopback() {
use crate::provider_server::ProviderServer;
use matter_crypto::{PeerInfo, ResumptionId, ResumptionRecord};
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr: _,
discovery: _,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let state = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let fabric = &state.fabrics[0];
let (provider_creds, provider_roots, _compressed) =
crate::credentials::operational_credentials(fabric).unwrap();
let provider_node_id = fabric.commissioner.node_id;
let fabric_id = fabric.fabric_id;
let prior_id = ResumptionId([0x42; 16]);
let prior_secret = [0x24u8; 32];
let provider_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: device_node_id,
fabric_id,
noc: device_creds.noc.clone(),
session_id: 0,
},
expires_at: None,
};
let requestor_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: provider_node_id,
fabric_id,
noc: provider_creds.noc.clone(),
session_id: 0,
},
expires_at: None,
};
let image: Vec<u8> = (0..2500u32)
.map(|i| u8::try_from(i % 251).unwrap())
.collect();
let offer = matter_ota::ImageOffer {
software_version: 2,
software_version_string: "2.0".into(),
image_uri: format!("bdx://{provider_node_id:016X}/fw.ota"),
update_token: vec![0xAB; 16],
};
let provider_addr = dev_io.local_addr();
let image_for_assert = image.clone();
let sunk: std::sync::Arc<std::sync::Mutex<Vec<matter_crypto::ResumptionRecord>>> =
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sunk_in = sunk.clone();
let server = tokio::spawn(async move {
ProviderServer::new(
dev_io,
vec![provider_creds],
provider_roots,
0x55,
MatterTime::from_unix_secs(2_000_000_000),
)
.with_resumption_records(vec![provider_record])
.with_record_sink(Box::new(move |r| sunk_in.lock().unwrap().push(r)))
.serve_ota_once(offer, image, 256)
.await
});
let reassembled = ota_test_requestor_resumed(
ctrl_io,
provider_addr,
device_creds,
device_roots,
provider_node_id,
fabric_id,
1,
requestor_record,
)
.await;
assert_eq!(
reassembled, image_for_assert,
"requestor reassembled the served image over the RESUMED session"
);
server
.await
.unwrap()
.expect("provider served OTA on resumed session");
let records = sunk.lock().unwrap();
assert_eq!(records.len(), 1);
assert_ne!(records[0].id, prior_id, "Sigma2_Resume rotates the id");
assert_eq!(
records[0].shared_secret, prior_secret,
"the shared secret carries over unchanged"
);
}
#[cfg(feature = "ota")]
#[allow(clippy::too_many_lines)] async fn ota_test_requestor(
io: matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
creds: matter_crypto::CaseCredentials,
roots: matter_cert::TrustedRoots,
provider_node_id: u64,
fabric_id: u64,
current_version: u32,
) -> Vec<u8> {
use matter_commissioning::driver::run_case;
use matter_transport::SessionManager;
let now = MatterTime::from_unix_secs(2_000_000_000);
let mut sessions = SessionManager::new();
let sid = run_case(
&io,
&mut sessions,
provider_addr,
creds,
roots,
provider_node_id,
fabric_id,
now,
)
.await
.unwrap();
drive_ota_flow(&io, provider_addr, &mut sessions, sid, current_version).await
}
#[cfg(feature = "ota")]
#[allow(clippy::too_many_arguments)] async fn ota_test_requestor_resumed(
io: matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
creds: matter_crypto::CaseCredentials,
roots: matter_cert::TrustedRoots,
provider_node_id: u64,
fabric_id: u64,
current_version: u32,
record: matter_crypto::ResumptionRecord,
) -> Vec<u8> {
let (mut sessions, sid) = resume_case_handshake(
&io,
provider_addr,
creds,
roots,
provider_node_id,
fabric_id,
record,
0x0021,
)
.await;
drive_ota_flow(&io, provider_addr, &mut sessions, sid, current_version).await
}
#[cfg(feature = "ota")]
async fn drive_ota_flow(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
sessions: &mut matter_transport::SessionManager,
sid: matter_transport::SessionId,
current_version: u32,
) -> Vec<u8> {
let (reassembled, token) =
drive_ota_download_and_apply(io, provider_addr, sessions, sid, current_version).await;
send_notify_update_applied(
io,
provider_addr,
sessions,
sid,
&token,
current_version + 1,
)
.await;
reassembled
}
#[cfg(feature = "ota")]
async fn drive_ota_download_and_apply(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
sessions: &mut matter_transport::SessionManager,
sid: matter_transport::SessionId,
current_version: u32,
) -> (Vec<u8>, Vec<u8>) {
let update_token = ota_query_image(io, provider_addr, sessions, sid, current_version).await;
let length = bdx_receive_init(io, provider_addr, sessions, sid).await;
let reassembled =
bdx_pull_blocks_and_ack_eof(io, provider_addr, sessions, sid, length).await;
ota_apply_update(
io,
provider_addr,
sessions,
sid,
&update_token,
current_version + 1,
)
.await;
(reassembled, update_token)
}
#[cfg(feature = "ota")]
async fn ota_query_image(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
sessions: &mut matter_transport::SessionManager,
sid: matter_transport::SessionId,
current_version: u32,
) -> Vec<u8> {
use matter_clusters::gen::ota_software_update_provider as prov;
use matter_commissioning::driver::secured_round_trip;
use matter_interaction::{
build_invoke_request, parse_invoke_response, CommandPath, InvokeResponse,
};
use matter_transport::ProtocolId;
const IM: u8 = 0x08;
let qi = prov::encode_query_image(
0xFFF1,
0x8000,
current_version,
&vec![prov::DownloadProtocolEnum::BdxSynchronous],
None,
None,
None,
None,
);
let qi_req = build_invoke_request(
CommandPath {
endpoint: 0,
cluster: prov::CLUSTER_ID,
command: prov::command_id::QUERY_IMAGE,
},
&qi,
);
let resp = secured_round_trip(
io,
sessions,
sid,
provider_addr,
IM,
ProtocolId::INTERACTION_MODEL,
&qi_req,
)
.await
.unwrap();
match parse_invoke_response(&resp.payload).unwrap() {
InvokeResponse::Command { fields_tlv, .. } => {
let qir = prov::QueryImageResponse::decode(&fields_tlv).unwrap();
assert_eq!(qir.status, prov::StatusEnum::UpdateAvailable);
qir.update_token.unwrap()
}
other @ InvokeResponse::Status(_) => {
panic!("expected QueryImageResponse command, got {other:?}")
}
}
}
#[cfg(feature = "ota")]
async fn bdx_receive_init(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
sessions: &mut matter_transport::SessionManager,
sid: matter_transport::SessionId,
) -> usize {
use matter_bdx::{ReceiveAccept, TransferControl, TransferInit};
use matter_commissioning::driver::secured_round_trip;
use matter_transport::ProtocolId;
let init = TransferInit {
control: TransferControl::RECEIVER_DRIVE,
version: 0,
max_block_size: 256,
start_offset: 0,
max_length: 0,
file_designator: b"fw.ota".to_vec(),
metadata: Vec::new(),
};
let acc = secured_round_trip(
io,
sessions,
sid,
provider_addr,
matter_bdx::MessageType::ReceiveInit.to_u8(),
ProtocolId::BDX,
&init.encode(),
)
.await
.unwrap();
let accept = ReceiveAccept::decode(&acc.payload).unwrap();
usize::try_from(accept.length).unwrap()
}
#[cfg(feature = "ota")]
async fn bdx_query_one_block(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
sessions: &mut matter_transport::SessionManager,
sid: matter_transport::SessionId,
block_counter: u32,
) -> Vec<u8> {
use matter_bdx::{CounterMessage, DataBlock};
use matter_commissioning::driver::secured_round_trip;
use matter_transport::ProtocolId;
let q = CounterMessage { block_counter }.encode();
let blk = secured_round_trip(
io,
sessions,
sid,
provider_addr,
matter_bdx::MessageType::BlockQuery.to_u8(),
ProtocolId::BDX,
&q,
)
.await
.unwrap();
DataBlock::decode(&blk.payload).unwrap().data
}
#[cfg(feature = "ota")]
async fn bdx_pull_blocks_and_ack_eof(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
sessions: &mut matter_transport::SessionManager,
sid: matter_transport::SessionId,
length: usize,
) -> Vec<u8> {
use matter_bdx::CounterMessage;
use matter_transport::{MrpFlags, ProtocolId};
use std::time::Instant;
let mut reassembled = Vec::new();
let mut counter = 0u32;
while reassembled.len() < length {
let data = bdx_query_one_block(io, provider_addr, sessions, sid, counter).await;
reassembled.extend_from_slice(&data);
counter += 1;
}
let ack = CounterMessage {
block_counter: counter - 1,
}
.encode();
let out = sessions
.encode_outbound(
sid,
None,
matter_bdx::MessageType::BlockAckEof.to_u8(),
ProtocolId::BDX,
&ack,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, provider_addr).await.unwrap();
reassembled
}
#[cfg(feature = "ota")]
async fn ota_apply_update(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
sessions: &mut matter_transport::SessionManager,
sid: matter_transport::SessionId,
update_token: &[u8],
target_version: u32,
) {
use matter_clusters::gen::ota_software_update_provider as prov;
use matter_commissioning::driver::secured_round_trip;
use matter_interaction::{
build_invoke_request, parse_invoke_response, CommandPath, InvokeResponse,
};
use matter_transport::ProtocolId;
const IM: u8 = 0x08;
let aur = prov::encode_apply_update_request(&update_token.to_vec(), target_version);
let aur_req = build_invoke_request(
CommandPath {
endpoint: 0,
cluster: prov::CLUSTER_ID,
command: prov::command_id::APPLY_UPDATE_REQUEST,
},
&aur,
);
let ar = secured_round_trip(
io,
sessions,
sid,
provider_addr,
IM,
ProtocolId::INTERACTION_MODEL,
&aur_req,
)
.await
.unwrap();
match parse_invoke_response(&ar.payload).unwrap() {
InvokeResponse::Command { fields_tlv, .. } => {
let r = prov::ApplyUpdateResponse::decode(&fields_tlv).unwrap();
assert_eq!(r.action, prov::ApplyUpdateActionEnum::Proceed);
}
other @ InvokeResponse::Status(_) => {
panic!("expected ApplyUpdateResponse command, got {other:?}")
}
}
}
async fn send_notify_update_applied(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
sessions: &mut matter_transport::SessionManager,
sid: matter_transport::SessionId,
token: &[u8],
software_version: u32,
) {
use matter_clusters::gen::ota_software_update_provider as prov;
use matter_commissioning::driver::secured_round_trip;
use matter_interaction::{
build_invoke_request, parse_invoke_response, CommandPath, InvokeResponse,
};
use matter_transport::ProtocolId;
const IM: u8 = 0x08;
let token_vec = token.to_vec();
let nua = prov::encode_notify_update_applied(&token_vec, software_version);
let nua_req = build_invoke_request(
CommandPath {
endpoint: 0,
cluster: prov::CLUSTER_ID,
command: prov::command_id::NOTIFY_UPDATE_APPLIED,
},
&nua,
);
let nr = secured_round_trip(
io,
sessions,
sid,
provider_addr,
IM,
ProtocolId::INTERACTION_MODEL,
&nua_req,
)
.await
.unwrap();
assert!(matches!(
parse_invoke_response(&nr.payload).unwrap(),
InvokeResponse::Status(matter_interaction::ImStatus::Success)
));
}
#[allow(clippy::too_many_arguments)] async fn resume_case_handshake(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
creds: matter_crypto::CaseCredentials,
roots: matter_cert::TrustedRoots,
provider_node_id: u64,
fabric_id: u64,
record: matter_crypto::ResumptionRecord,
session_id: u16,
) -> (
matter_transport::SessionManager,
matter_transport::SessionId,
) {
use matter_commissioning::driver::{decode_unsecured, encode_unsecured};
use matter_transport::{ProtocolId, SessionManager};
const OP_SIGMA1: u8 = 0x30;
const OP_SIGMA2_RESUME: u8 = 0x33;
const OP_STATUS_REPORT: u8 = 0x40;
let exchange: u16 = 0x7000 | session_id;
let now = MatterTime::from_unix_secs(2_000_000_000);
let mut initiator = matter_crypto::CaseInitiator::new_with_resumption(
creds,
roots,
provider_node_id,
fabric_id,
record,
session_id,
now,
)
.unwrap();
let sigma1 = initiator.start().unwrap();
let wire = encode_unsecured(
1,
exchange,
OP_SIGMA1,
ProtocolId::SECURE_CHANNEL,
true,
true,
None,
None,
&sigma1,
);
io.send_to(&wire, provider_addr).await.unwrap();
let (bytes, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&bytes).unwrap();
assert_eq!(m.opcode, OP_SIGMA2_RESUME, "expected Sigma2_Resume");
initiator.handle_sigma2_resume(&m.payload).unwrap();
let mut body = Vec::with_capacity(8);
body.extend_from_slice(&0u16.to_le_bytes()); body.extend_from_slice(&0u32.to_le_bytes()); body.extend_from_slice(&0u16.to_le_bytes()); let report = encode_unsecured(
2,
exchange,
OP_STATUS_REPORT,
ProtocolId::SECURE_CHANNEL,
true,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, provider_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = initiator.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, matter_transport::SessionRole::Initiator);
(sessions, sid)
}
fn make_device_creds_for_fabric(
fabric: &crate::state::FabricEntry,
) -> (matter_crypto::CaseCredentials, matter_cert::TrustedRoots) {
let device_node_id: u64 = 0x0000_0000_0000_0042;
let device_record = fabric.to_fabric_record().unwrap();
let (device_signer, _pkcs8) = RingSigner::generate().unwrap();
let device_noc = issue_noc(
&device_record,
&VerifiedCsr {
public_key: device_signer.public_key().clone(),
},
device_node_id,
&[],
(
MatterTime::from_unix_secs(1_700_000_000),
MatterTime::NO_EXPIRY,
),
&SystemNocRng,
)
.unwrap();
let compressed =
derive_compressed_fabric_id(fabric.rcac_cert.public_key().as_bytes(), fabric.fabric_id)
.unwrap();
let device_ipk = derive_operational_ipk(&fabric.ipk, &compressed).unwrap();
let mut device_roots = TrustedRoots::new();
device_roots.add(TrustAnchor::from_root_cert(&fabric.rcac_cert));
let device_creds = CaseCredentials {
noc: device_noc,
icac: None,
signer: Box::new(device_signer),
fabric_id: fabric.fabric_id,
node_id: device_node_id,
ipk: device_ipk,
rcac_public_key: *fabric.rcac_cert.public_key().as_bytes(),
};
(device_creds, device_roots)
}
#[cfg(feature = "ota")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_ota_survives_stray_frames_before_sigma1() {
use crate::provider_server::ProviderServer;
use matter_crypto::{PeerInfo, ResumptionId, ResumptionRecord};
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr: _,
discovery: _,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let state = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let fabric = &state.fabrics[0];
let (provider_creds, provider_roots, _compressed) =
crate::credentials::operational_credentials(fabric).unwrap();
let provider_node_id = fabric.commissioner.node_id;
let fabric_id = fabric.fabric_id;
let prior_id = ResumptionId([0x43; 16]);
let prior_secret = [0x25u8; 32];
let provider_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: device_node_id,
fabric_id,
noc: device_creds.noc.clone(),
session_id: 0,
},
expires_at: None,
};
let requestor_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: provider_node_id,
fabric_id,
noc: provider_creds.noc.clone(),
session_id: 0,
},
expires_at: None,
};
let image: Vec<u8> = (0..500u32)
.map(|i| u8::try_from(i % 251).unwrap())
.collect();
let offer = matter_ota::ImageOffer {
software_version: 2,
software_version_string: "2.0".into(),
image_uri: format!("bdx://{provider_node_id:016X}/fw.ota"),
update_token: vec![0xAB; 16],
};
let provider_addr = dev_io.local_addr();
let image_for_assert = image.clone();
let server = tokio::spawn(async move {
ProviderServer::new(
dev_io,
vec![provider_creds], provider_roots,
0x56,
MatterTime::from_unix_secs(2_000_000_000),
)
.with_resumption_records(vec![provider_record])
.serve_ota_once(offer, image, 256)
.await
});
ctrl_io
.send_to(&[0xDE, 0xAD, 0xBE, 0xEF], provider_addr)
.await
.unwrap();
let stray_ack = matter_commissioning::driver::encode_unsecured(
1,
0x7777,
0x10, matter_transport::ProtocolId::SECURE_CHANNEL,
true,
false,
Some(1),
None,
&[],
);
ctrl_io.send_to(&stray_ack, provider_addr).await.unwrap();
let reassembled = ota_test_requestor_resumed(
ctrl_io,
provider_addr,
device_creds,
device_roots,
provider_node_id,
fabric_id,
1,
requestor_record,
)
.await;
assert_eq!(
reassembled, image_for_assert,
"the single-credential serve must survive pre-Sigma1 strays"
);
server
.await
.unwrap()
.expect("serve completed despite stray frames");
}
#[cfg(feature = "ota")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_ota_rejects_unpinned_peer() {
use crate::provider_server::ProviderServer;
use matter_crypto::{PeerInfo, ResumptionId, ResumptionRecord};
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr: _,
discovery: _,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let state = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let fabric = &state.fabrics[0];
let (provider_creds, provider_roots, _compressed) =
crate::credentials::operational_credentials(fabric).unwrap();
let provider_node_id = fabric.commissioner.node_id;
let fabric_id = fabric.fabric_id;
let prior_id = ResumptionId([0x44; 16]);
let prior_secret = [0x26u8; 32];
let provider_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: device_node_id,
fabric_id,
noc: device_creds.noc.clone(),
session_id: 0,
},
expires_at: None,
};
let requestor_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: provider_node_id,
fabric_id,
noc: provider_creds.noc.clone(),
session_id: 0,
},
expires_at: None,
};
let offer = matter_ota::ImageOffer {
software_version: 2,
software_version_string: "2.0".into(),
image_uri: format!("bdx://{provider_node_id:016X}/fw.ota"),
update_token: vec![0xAB; 16],
};
let provider_addr = dev_io.local_addr();
let server = tokio::spawn(async move {
ProviderServer::new(
dev_io,
vec![provider_creds], provider_roots,
0x57,
MatterTime::from_unix_secs(2_000_000_000),
)
.with_resumption_records(vec![provider_record])
.with_expected_peer(device_node_id + 1)
.serve_ota_once(offer, vec![0u8; 64], 256)
.await
});
let requestor = tokio::spawn(async move {
let _ = resume_case_handshake(
&ctrl_io,
provider_addr,
device_creds,
device_roots,
provider_node_id,
fabric_id,
requestor_record,
0x0031,
)
.await;
});
let err = server
.await
.unwrap()
.expect_err("pinned serve must reject the wrong peer");
assert!(
err.to_string().contains("not the expected"),
"unexpected error: {err}"
);
requestor.abort();
}
#[cfg(feature = "ota")]
#[allow(clippy::too_many_lines)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_ota_spans_sessions_for_post_reboot_notify() {
use crate::provider_server::ProviderServer;
use matter_crypto::{PeerInfo, ResumptionId, ResumptionRecord};
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr: _,
discovery: _,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let state = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let fabric = &state.fabrics[0];
let (provider_creds1, provider_roots, _compressed) =
crate::credentials::operational_credentials(fabric).unwrap();
let (provider_creds2, _, _) = crate::credentials::operational_credentials(fabric).unwrap();
let provider_node_id = fabric.commissioner.node_id;
let fabric_id = fabric.fabric_id;
let provider_noc = provider_creds1.noc.clone();
let (device_creds2, device_roots2) = make_device_creds_for_fabric(fabric);
let prior_id = ResumptionId([0x42; 16]);
let prior_secret = [0x24u8; 32];
let provider_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: device_node_id,
fabric_id,
noc: device_creds.noc.clone(),
session_id: 0,
},
expires_at: None,
};
let requestor_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: provider_node_id,
fabric_id,
noc: provider_noc.clone(),
session_id: 0,
},
expires_at: None,
};
let image: Vec<u8> = (0..2500u32)
.map(|i| u8::try_from(i % 251).unwrap())
.collect();
let offer = matter_ota::ImageOffer {
software_version: 2,
software_version_string: "2.0".into(),
image_uri: format!("bdx://{provider_node_id:016X}/fw.ota"),
update_token: vec![0xAB; 16],
};
let provider_addr = dev_io.local_addr();
let image_for_assert = image.clone();
let sunk: std::sync::Arc<std::sync::Mutex<Vec<matter_crypto::ResumptionRecord>>> =
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sunk_in = sunk.clone();
let server = tokio::spawn(async move {
ProviderServer::new(
dev_io,
vec![provider_creds1, provider_creds2],
provider_roots,
0x55,
MatterTime::from_unix_secs(2_000_000_000),
)
.with_resumption_records(vec![provider_record])
.with_record_sink(Box::new(move |r| sunk_in.lock().unwrap().push(r)))
.serve_ota_once(offer, image, 256)
.await
});
let (mut s1, sid1) = resume_case_handshake(
&ctrl_io,
provider_addr,
device_creds,
device_roots,
provider_node_id,
fabric_id,
requestor_record,
0x0021,
)
.await;
let (reassembled, token) =
drive_ota_download_and_apply(&ctrl_io, provider_addr, &mut s1, sid1, 1).await;
assert_eq!(
reassembled, image_for_assert,
"session-1 download must reassemble the full image"
);
let rotated = sunk.lock().unwrap()[0].clone();
let requestor_record2 = ResumptionRecord {
id: rotated.id,
shared_secret: rotated.shared_secret,
peer: PeerInfo {
node_id: provider_node_id,
fabric_id,
noc: provider_noc, session_id: 0,
},
expires_at: None,
};
let (mut s2, sid2) = resume_case_handshake(
&ctrl_io,
provider_addr,
device_creds2,
device_roots2,
provider_node_id,
fabric_id,
requestor_record2,
0x0022,
)
.await;
send_notify_update_applied(&ctrl_io, provider_addr, &mut s2, sid2, &token, 2).await;
server
.await
.unwrap()
.expect("serve_ota_once must complete on cross-session Notify");
let records = sunk.lock().unwrap();
assert_eq!(
records.len(),
2,
"one record per accept (one per CASE session)"
);
assert_ne!(
records[0].id, records[1].id,
"each accept rotates the resumption id"
);
}
#[allow(clippy::too_many_arguments)] async fn full_case_handshake(
io: &matter_commissioning::driver::InMemoryDatagram,
provider_addr: std::net::SocketAddr,
creds: matter_crypto::CaseCredentials,
roots: matter_cert::TrustedRoots,
provider_node_id: u64,
fabric_id: u64,
session_id: u16,
send_final_ack: bool,
) -> (
matter_transport::SessionManager,
matter_transport::SessionId,
) {
use matter_commissioning::driver::{decode_unsecured, encode_unsecured};
use matter_transport::{ProtocolId, SessionManager};
const OP_SIGMA1: u8 = 0x30;
const OP_SIGMA2: u8 = 0x31;
const OP_SIGMA3: u8 = 0x32;
const OP_STATUS_REPORT: u8 = 0x40;
const OP_MRP_STANDALONE_ACK: u8 = 0x10;
let exchange: u16 = 0x6000 | session_id;
let now = MatterTime::from_unix_secs(2_000_000_000);
let mut initiator = matter_crypto::CaseInitiator::new(
creds,
roots,
provider_node_id,
fabric_id,
session_id,
now,
)
.unwrap();
let sigma1 = initiator.start().unwrap();
let wire = encode_unsecured(
1,
exchange,
OP_SIGMA1,
ProtocolId::SECURE_CHANNEL,
true,
true,
None,
None,
&sigma1,
);
io.send_to(&wire, provider_addr).await.unwrap();
let (bytes, _) = io.recv_from().await.unwrap();
let m2 = decode_unsecured(&bytes).unwrap();
assert_eq!(m2.opcode, OP_SIGMA2, "expected Sigma2");
initiator.handle_sigma2(&m2.payload).unwrap();
let sigma3 = initiator.next_message().unwrap();
let wire = encode_unsecured(
2,
exchange,
OP_SIGMA3,
ProtocolId::SECURE_CHANNEL,
true,
true,
Some(m2.message_counter),
None,
&sigma3,
);
io.send_to(&wire, provider_addr).await.unwrap();
let (bytes, _) = io.recv_from().await.unwrap();
let report = decode_unsecured(&bytes).unwrap();
assert_eq!(report.opcode, OP_STATUS_REPORT, "expected StatusReport");
assert_eq!(
report.payload.get(0..2),
Some(&[0u8, 0u8][..]),
"handshake must close with a success StatusReport"
);
if send_final_ack {
let ack = encode_unsecured(
3,
exchange,
OP_MRP_STANDALONE_ACK,
ProtocolId::SECURE_CHANNEL,
true,
false,
Some(report.message_counter),
None,
&[],
);
io.send_to(&ack, provider_addr).await.unwrap();
}
let output = initiator.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, matter_transport::SessionRole::Initiator);
(sessions, sid)
}
#[cfg(feature = "ota")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_ota_carries_sigma1_arriving_in_place_of_close_ack() {
use crate::provider_server::ProviderServer;
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr: _,
discovery: _,
device_creds,
device_roots,
device_node_id: _,
} = loopback_harness();
let state = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let fabric = &state.fabrics[0];
let (provider_creds1, provider_roots, _compressed) =
crate::credentials::operational_credentials(fabric).unwrap();
let (provider_creds2, _, _) = crate::credentials::operational_credentials(fabric).unwrap();
let provider_node_id = fabric.commissioner.node_id;
let fabric_id = fabric.fabric_id;
let (device_creds2, device_roots2) = make_device_creds_for_fabric(fabric);
let image: Vec<u8> = (0..2500u32)
.map(|i| u8::try_from(i % 251).unwrap())
.collect();
let offer = matter_ota::ImageOffer {
software_version: 2,
software_version_string: "2.0".into(),
image_uri: format!("bdx://{provider_node_id:016X}/fw.ota"),
update_token: vec![0xAB; 16],
};
let provider_addr = dev_io.local_addr();
let image_for_assert = image.clone();
let sunk: std::sync::Arc<std::sync::Mutex<Vec<matter_crypto::ResumptionRecord>>> =
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sunk_in = sunk.clone();
let server = tokio::spawn(async move {
ProviderServer::new(
dev_io,
vec![provider_creds1, provider_creds2],
provider_roots,
0x55,
MatterTime::from_unix_secs(2_000_000_000),
)
.with_record_sink(Box::new(move |r| sunk_in.lock().unwrap().push(r)))
.serve_ota_once(offer, image, 256)
.await
});
let (_s1, _sid1) = full_case_handshake(
&ctrl_io,
provider_addr,
device_creds,
device_roots,
provider_node_id,
fabric_id,
0x0021,
false,
)
.await;
let (mut s2, sid2) = tokio::time::timeout(
std::time::Duration::from_secs(10),
full_case_handshake(
&ctrl_io,
provider_addr,
device_creds2,
device_roots2,
provider_node_id,
fabric_id,
0x0022,
true,
),
)
.await
.expect("provider must answer the fast Sigma1 (pre-fix it was absorbed as the close ack)");
let reassembled = drive_ota_flow(&ctrl_io, provider_addr, &mut s2, sid2, 1).await;
assert_eq!(reassembled, image_for_assert);
server
.await
.unwrap()
.expect("serve must complete on session 2's Notify");
assert_eq!(
sunk.lock().unwrap().len(),
2,
"one resumption record per accept"
);
}
#[cfg(feature = "ota")]
#[allow(clippy::too_many_lines)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_ota_rearms_bdx_for_cross_session_receive_init() {
use crate::provider_server::ProviderServer;
use matter_crypto::{PeerInfo, ResumptionId, ResumptionRecord};
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr: _,
discovery: _,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let state = crate::snapshot::deserialize(&store.load().unwrap().unwrap()).unwrap();
let fabric = &state.fabrics[0];
let (provider_creds1, provider_roots, _compressed) =
crate::credentials::operational_credentials(fabric).unwrap();
let (provider_creds2, _, _) = crate::credentials::operational_credentials(fabric).unwrap();
let provider_node_id = fabric.commissioner.node_id;
let fabric_id = fabric.fabric_id;
let provider_noc = provider_creds1.noc.clone();
let (device_creds2, device_roots2) = make_device_creds_for_fabric(fabric);
let prior_id = ResumptionId([0x45; 16]);
let prior_secret = [0x27u8; 32];
let provider_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: device_node_id,
fabric_id,
noc: device_creds.noc.clone(),
session_id: 0,
},
expires_at: None,
};
let requestor_record = ResumptionRecord {
id: prior_id,
shared_secret: prior_secret,
peer: PeerInfo {
node_id: provider_node_id,
fabric_id,
noc: provider_noc.clone(),
session_id: 0,
},
expires_at: None,
};
let image: Vec<u8> = (0..2500u32)
.map(|i| u8::try_from(i % 251).unwrap())
.collect();
let offer = matter_ota::ImageOffer {
software_version: 2,
software_version_string: "2.0".into(),
image_uri: format!("bdx://{provider_node_id:016X}/fw.ota"),
update_token: vec![0xAB; 16],
};
let provider_addr = dev_io.local_addr();
let image_for_assert = image.clone();
let sunk: std::sync::Arc<std::sync::Mutex<Vec<matter_crypto::ResumptionRecord>>> =
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sunk_in = sunk.clone();
let server = tokio::spawn(async move {
ProviderServer::new(
dev_io,
vec![provider_creds1, provider_creds2],
provider_roots,
0x58,
MatterTime::from_unix_secs(2_000_000_000),
)
.with_resumption_records(vec![provider_record])
.with_record_sink(Box::new(move |r| sunk_in.lock().unwrap().push(r)))
.serve_ota_once(offer, image, 256)
.await
});
let (mut s1, sid1) = resume_case_handshake(
&ctrl_io,
provider_addr,
device_creds,
device_roots,
provider_node_id,
fabric_id,
requestor_record,
0x0023,
)
.await;
let token = ota_query_image(&ctrl_io, provider_addr, &mut s1, sid1, 1).await;
let length = bdx_receive_init(&ctrl_io, provider_addr, &mut s1, sid1).await;
let first_block = bdx_query_one_block(&ctrl_io, provider_addr, &mut s1, sid1, 0).await;
assert!(
!first_block.is_empty() && first_block.len() < length,
"session 1 must stop mid-transfer"
);
let rotated = sunk.lock().unwrap()[0].clone();
let requestor_record2 = ResumptionRecord {
id: rotated.id,
shared_secret: rotated.shared_secret,
peer: PeerInfo {
node_id: provider_node_id,
fabric_id,
noc: provider_noc,
session_id: 0,
},
expires_at: None,
};
let (mut s2, sid2) = resume_case_handshake(
&ctrl_io,
provider_addr,
device_creds2,
device_roots2,
provider_node_id,
fabric_id,
requestor_record2,
0x0024,
)
.await;
let length2 = bdx_receive_init(&ctrl_io, provider_addr, &mut s2, sid2).await;
assert_eq!(
length2,
image_for_assert.len(),
"re-armed transfer serves the full image"
);
let reassembled =
bdx_pull_blocks_and_ack_eof(&ctrl_io, provider_addr, &mut s2, sid2, length2).await;
assert_eq!(
reassembled, image_for_assert,
"the re-initiated transfer must serve the image from the start"
);
ota_apply_update(&ctrl_io, provider_addr, &mut s2, sid2, &token, 2).await;
send_notify_update_applied(&ctrl_io, provider_addr, &mut s2, sid2, &token, 2).await;
server
.await
.unwrap()
.expect("serve must survive the cross-session ReceiveInit");
}
#[tokio::test]
async fn set_utc_time_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = matter_interaction::build_invoke_response_status(
matter_interaction::CommandPath {
endpoint: 0,
cluster: 0x0038,
command: 0x00,
},
matter_interaction::ImStatus::Success,
);
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.set_utc_time(1_000_000, crate::TimeGranularity::Seconds)
.await
.expect("set utc time");
device.await.unwrap();
}
#[tokio::test]
async fn set_time_zone_over_loopback_returns_dst_required() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let resp_fields = {
use matter_codec::{Tag, TlvWriter};
let mut b = Vec::new();
let mut w = TlvWriter::new(&mut b);
w.start_structure(Tag::Anonymous).unwrap();
w.put_bool(Tag::Context(0), true).unwrap();
w.end_container().unwrap();
b
};
let reply = matter_interaction::build_invoke_response_command(
matter_interaction::CommandPath {
endpoint: 0,
cluster: 0x0038,
command: 0x03,
},
&resp_fields,
);
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let dst_required = controller
.node(device_node_id)
.set_time_zone(&[crate::TimeZoneEntry::new(3600, 0, Some("CET".into()))])
.await
.expect("set time zone");
assert!(dst_required);
device.await.unwrap();
}
#[tokio::test]
async fn set_dst_offset_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = matter_interaction::build_invoke_response_status(
matter_interaction::CommandPath {
endpoint: 0,
cluster: 0x0038,
command: 0x04,
},
matter_interaction::ImStatus::Success,
);
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.set_dst_offset(&[crate::DstOffsetEntry::new(3600, 0, None)])
.await
.expect("set dst offset");
device.await.unwrap();
}
#[tokio::test]
async fn revoke_commissioning_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
build_invoke_status_success(),
true,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.revoke_commissioning()
.await
.expect("revoke");
device.await.unwrap();
}
#[tokio::test]
async fn commissioning_window_status_reads_window_status_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = build_report_data(0, 0x003C, 0x0000, &matter_codec::Value::Uint(1));
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let ws = controller
.node(device_node_id)
.commissioning_window_status()
.await
.expect("status");
assert_eq!(
ws.status,
crate::admin::CommissioningWindowStatus::EnhancedWindowOpen
);
device.await.unwrap();
}
#[tokio::test]
async fn list_fabrics_reads_fabrics_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let fabric = matter_codec::Value::Structure(vec![
(
matter_codec::Tag::Context(1),
matter_codec::Value::Bytes(vec![4u8; 65]),
),
(
matter_codec::Tag::Context(2),
matter_codec::Value::Uint(0xFFF1),
),
(
matter_codec::Tag::Context(3),
matter_codec::Value::Uint(0xAABB),
),
(
matter_codec::Tag::Context(4),
matter_codec::Value::Uint(0x1234),
),
(
matter_codec::Tag::Context(5),
matter_codec::Value::Utf8("home".into()),
),
(
matter_codec::Tag::Context(254),
matter_codec::Value::Uint(1),
),
]);
let reply = build_report_data(0, 0x003E, 0x0001, &matter_codec::Value::Array(vec![fabric]));
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let fabrics = controller
.node(device_node_id)
.list_fabrics()
.await
.expect("list");
assert_eq!(fabrics.len(), 1);
assert_eq!(fabrics[0].fabric_index, 1);
assert_eq!(fabrics[0].fabric_id, 0xAABB);
device.await.unwrap();
}
#[tokio::test]
async fn read_acl_reads_acl_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let entry = matter_codec::Value::Structure(vec![
(
matter_codec::Tag::Context(1),
matter_codec::Value::Uint(5), ),
(
matter_codec::Tag::Context(2),
matter_codec::Value::Uint(2), ),
(
matter_codec::Tag::Context(3),
matter_codec::Value::Array(vec![matter_codec::Value::Uint(0x1234)]),
),
(matter_codec::Tag::Context(4), matter_codec::Value::Null),
(
matter_codec::Tag::Context(254),
matter_codec::Value::Uint(1), ),
]);
let reply = build_report_data(0, 0x001F, 0x0000, &matter_codec::Value::Array(vec![entry]));
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let acl = controller
.node(device_node_id)
.read_acl()
.await
.expect("read_acl");
assert_eq!(acl.len(), 1);
assert_eq!(acl[0].privilege, crate::acl::AclPrivilege::Administer);
device.await.unwrap();
}
fn build_invoke_response_noc(status: u8, fabric_index: Option<u8>) -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap(); w.put_bool(Tag::Context(0), false).unwrap(); w.start_array(Tag::Context(1)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_structure(Tag::Context(0)).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(0), 0).unwrap(); w.put_uint(
Tag::Context(1),
u64::from(crate::opcreds::OPERATIONAL_CREDENTIALS_CLUSTER),
)
.unwrap(); w.put_uint(Tag::Context(2), 0x08).unwrap(); w.end_container().unwrap(); w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), u64::from(status)).unwrap(); if let Some(fi) = fabric_index {
w.put_uint(Tag::Context(1), u64::from(fi)).unwrap(); }
w.end_container().unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap(); w.end_container().unwrap(); buf
}
async fn run_loopback_device_seq(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
replies: Vec<Vec<u8>>,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
for reply_payload in &replies {
let (wire, _) = io.recv_from().await.unwrap();
let decoded = sessions.decode_inbound(&wire, Instant::now()).unwrap();
let DecodeInboundOutput::AppMessage { exchange_id, .. } = decoded else {
panic!("expected an IM request app message");
};
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x05,
ProtocolId::INTERACTION_MODEL,
reply_payload,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
}
}
fn write_request_has_more_chunked(msg: &[u8]) -> bool {
use matter_codec::{Element, Tag, TlvReader};
let mut r = TlvReader::new(msg);
let _ = r.next(); loop {
match r.next() {
Ok(Some(Element::Scalar {
tag: Tag::Context(3),
value: matter_codec::Value::Bool(b),
})) => return b,
Ok(Some(Element::ContainerStart { .. })) => {
let mut depth = 1usize;
while depth > 0 {
match r.next() {
Ok(Some(Element::ContainerStart { .. })) => depth += 1,
Ok(Some(Element::ContainerEnd)) => depth -= 1,
Ok(Some(_)) => {}
Ok(None) | Err(_) => return false,
}
}
}
Ok(Some(Element::ContainerEnd) | None) | Err(_) => return false,
Ok(Some(_)) => {}
}
}
}
#[allow(clippy::too_many_lines)] async fn run_chunked_write_device(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
expected_chunks: usize,
write_response: Vec<u8>,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
let mut exchange_seen: Option<u16> = None;
for i in 0..expected_chunks {
let (w, _) = io.recv_from().await.unwrap();
let recv_at = Instant::now();
let DecodeInboundOutput::AppMessage {
exchange_id,
opcode,
payload,
..
} = sessions.decode_inbound(&w, recv_at).unwrap()
else {
panic!("expected a WriteRequest app message for chunk {i}");
};
assert_eq!(opcode, 0x06, "chunk {i} must be a WriteRequest (0x06)");
match exchange_seen {
None => exchange_seen = Some(exchange_id),
Some(ex) => assert_eq!(
ex, exchange_id,
"every chunk must reuse the same exchange (one-exchange invariant)"
),
}
let more = write_request_has_more_chunked(&payload);
assert_eq!(
more,
i + 1 != expected_chunks,
"chunk {i} MoreChunkedMessages flag disagrees with expected_chunks"
);
tokio::select! {
biased;
extra = io.recv_from() => {
let len = extra.map_or(0, |(b, _)| b.len());
panic!(
"chunked-write pipelining detected: a WriteRequest for chunk {} \
({len} bytes) arrived before the device replied to chunk {i} — \
the client must gate each chunk on its WriteResponse",
i + 1
);
}
() = tokio::task::yield_now() => {}
}
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x07, ProtocolId::INTERACTION_MODEL,
&write_response,
MrpFlags { reliable: true },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
}
}
#[tokio::test]
async fn chunked_write_sends_all_chunks_one_exchange() {
use matter_codec::{Tag, TlvWriter};
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let path = matter_interaction::AttributePath {
endpoint: 1,
cluster: 0x001F, attribute: 0x0000,
};
let elems: Vec<Vec<u8>> = (0u64..4)
.map(|n| {
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.put_uint(Tag::Anonymous, n).unwrap();
buf
})
.collect();
let chunks = matter_interaction::build_list_write_chunks(path, &elems, 40, false);
assert!(
chunks.len() >= 2,
"test needs a multi-chunk write; got {} chunk(s)",
chunks.len()
);
for (i, c) in chunks.iter().enumerate() {
assert_eq!(
write_request_has_more_chunked(c),
i + 1 != chunks.len(),
"chunk {i} MoreChunkedMessages flag"
);
}
let n_chunks = chunks.len();
let write_response = {
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.start_array(Tag::Context(0)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(2), u64::from(path.endpoint))
.unwrap();
w.put_uint(Tag::Context(3), u64::from(path.cluster))
.unwrap();
w.put_uint(Tag::Context(4), u64::from(path.attribute))
.unwrap();
w.end_container().unwrap();
w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), 0).unwrap(); w.end_container().unwrap();
w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
};
let device = tokio::spawn(run_chunked_write_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00DC,
n_chunks,
write_response,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let statuses = controller
.node(device_node_id)
.chunked_write(chunks)
.await
.expect("chunked_write");
assert_eq!(statuses.len(), n_chunks);
for (path, status) in &statuses {
assert_eq!(*status, matter_interaction::ImStatus::Success);
assert_eq!(path.cluster, 0x001F);
}
device.await.unwrap();
}
#[allow(clippy::too_many_lines)] #[allow(clippy::too_many_arguments)] async fn run_chunked_write_device_pumps_all_despite_failure(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
expected_chunks: usize,
first_response: Vec<u8>,
rest_response: Vec<u8>,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
let mut exchange_seen: Option<u16> = None;
for i in 0..expected_chunks {
let (w, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id,
opcode,
payload,
..
} = sessions.decode_inbound(&w, Instant::now()).unwrap()
else {
panic!("expected a WriteRequest app message for chunk {i}");
};
assert_eq!(opcode, 0x06, "chunk {i} must be a WriteRequest (0x06)");
match exchange_seen {
None => exchange_seen = Some(exchange_id),
Some(ex) => assert_eq!(
ex, exchange_id,
"every chunk must reuse the same exchange (one-exchange invariant)"
),
}
let more = write_request_has_more_chunked(&payload);
assert_eq!(
more,
i + 1 != expected_chunks,
"chunk {i} MoreChunkedMessages flag disagrees with expected_chunks"
);
tokio::select! {
biased;
extra = io.recv_from() => {
let len = extra.map_or(0, |(b, _)| b.len());
panic!(
"chunked-write pipelining detected: a WriteRequest for chunk {} \
({len} bytes) arrived before the device replied to chunk {i}",
i + 1
);
}
() = tokio::task::yield_now() => {}
}
let response = if i == 0 {
&first_response
} else {
&rest_response
};
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x07, ProtocolId::INTERACTION_MODEL,
response,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
}
}
#[tokio::test]
async fn chunked_write_pumps_all_chunks_despite_non_success_element_status() {
use matter_codec::{Tag, TlvWriter};
fn write_response_with_status(
path: matter_interaction::AttributePath,
status: u8,
) -> Vec<u8> {
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.start_array(Tag::Context(0)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(2), u64::from(path.endpoint))
.unwrap();
w.put_uint(Tag::Context(3), u64::from(path.cluster))
.unwrap();
w.put_uint(Tag::Context(4), u64::from(path.attribute))
.unwrap();
w.end_container().unwrap();
w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), u64::from(status)).unwrap();
w.end_container().unwrap();
w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
}
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let path = matter_interaction::AttributePath {
endpoint: 1,
cluster: 0x001F,
attribute: 0x0000,
};
let elems: Vec<Vec<u8>> = (0u64..4)
.map(|n| {
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.put_uint(Tag::Anonymous, n).unwrap();
buf
})
.collect();
let chunks = matter_interaction::build_list_write_chunks(path, &elems, 40, false);
assert!(
chunks.len() >= 2,
"test needs a multi-chunk write; got {} chunk(s)",
chunks.len()
);
let n_chunks = chunks.len();
let first_response = write_response_with_status(path, 0x01); let rest_response = write_response_with_status(path, 0x00);
let device = tokio::spawn(run_chunked_write_device_pumps_all_despite_failure(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00DD,
n_chunks,
first_response,
rest_response,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let statuses = controller
.node(device_node_id)
.chunked_write(chunks)
.await
.expect("chunked_write must still resolve Ok — a bad element status is not terminal");
assert_eq!(statuses.len(), n_chunks);
assert_eq!(statuses[0].1, matter_interaction::ImStatus::Failure(0x01));
for status in &statuses[1..] {
assert_eq!(status.1, matter_interaction::ImStatus::Success);
}
device.await.unwrap();
}
async fn run_chunked_write_device_rejects_with_status(
io: InMemoryDatagram,
ctrl_addr: std::net::SocketAddr,
creds: CaseCredentials,
roots: TrustedRoots,
responder_session_id: u16,
status: u8,
) {
let mut responder = CaseResponder::new(
creds,
roots,
responder_session_id,
MatterTime::from_unix_secs(2_000_000_000),
)
.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
assert!(matches!(
responder.handle_sigma1(&m.payload).unwrap(),
Sigma1Outcome::NewSession
));
let sigma2 = responder.next_message().unwrap();
let wire = encode_unsecured(
200,
m.exchange_id,
0x31,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&sigma2,
);
io.send_to(&wire, ctrl_addr).await.unwrap();
let (p, _) = io.recv_from().await.unwrap();
let m = decode_unsecured(&p).unwrap();
responder.handle_sigma3(&m.payload).unwrap();
let mut body = Vec::new();
body.extend_from_slice(&0u16.to_le_bytes());
body.extend_from_slice(&0u32.to_le_bytes());
body.extend_from_slice(&0u16.to_le_bytes());
let report = encode_unsecured(
201,
m.exchange_id,
0x40,
ProtocolId::SECURE_CHANNEL,
false,
true,
Some(m.message_counter),
None,
&body,
);
io.send_to(&report, ctrl_addr).await.unwrap();
let _ack = io.recv_from().await.unwrap();
let output = responder.finish().unwrap();
let mut sessions = SessionManager::new();
let sid = sessions.register_case(&output, SessionRole::Responder);
let (w, _) = io.recv_from().await.unwrap();
let DecodeInboundOutput::AppMessage {
exchange_id,
opcode,
payload,
..
} = sessions.decode_inbound(&w, Instant::now()).unwrap()
else {
panic!("expected a WriteRequest app message for chunk 0");
};
assert_eq!(opcode, 0x06, "chunk 0 must be a WriteRequest (0x06)");
assert!(
write_request_has_more_chunked(&payload),
"chunk 0 must carry MoreChunkedMessages (test needs a multi-chunk write)"
);
let status_bytes = matter_interaction::build_status_response(status);
let out = sessions
.encode_outbound(
sid,
Some(exchange_id),
0x01, ProtocolId::INTERACTION_MODEL,
&status_bytes,
MrpFlags { reliable: false },
Instant::now(),
)
.unwrap();
io.send_to(&out.wire_bytes, ctrl_addr).await.unwrap();
let grace =
tokio::time::timeout(std::time::Duration::from_millis(200), io.recv_from()).await;
assert!(
grace.is_err(),
"client sent a further chunk after the device rejected the chunked write"
);
}
#[tokio::test]
async fn chunked_write_terminal_on_status_response_rejection() {
use matter_codec::{Tag, TlvWriter};
const BUSY: u8 = 0x9C;
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let path = matter_interaction::AttributePath {
endpoint: 1,
cluster: 0x001F,
attribute: 0x0000,
};
let elems: Vec<Vec<u8>> = (0u64..4)
.map(|n| {
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.put_uint(Tag::Anonymous, n).unwrap();
buf
})
.collect();
let chunks = matter_interaction::build_list_write_chunks(path, &elems, 40, false);
assert!(
chunks.len() >= 2,
"test needs a multi-chunk write; got {} chunk(s)",
chunks.len()
);
let device = tokio::spawn(run_chunked_write_device_rejects_with_status(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x00DE,
BUSY,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let err = controller
.node(device_node_id)
.chunked_write(chunks)
.await
.expect_err("a StatusResponse rejection must be an Err, not Ok(vec![])");
let msg = err.to_string();
assert!(
msg.to_lowercase().contains("0x9c"),
"error must name the IM status (0x9c); got: {msg}"
);
device.await.unwrap();
}
#[tokio::test]
async fn commissioner_node_id_returns_stored_id() {
let Harness {
store,
ctrl_io,
discovery,
device_node_id,
..
} = loopback_harness();
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let id = controller
.node(device_node_id)
.commissioner_node_id()
.await
.expect("commissioner_node_id");
assert_eq!(id, 1);
}
#[tokio::test]
async fn remove_fabric_refuses_self_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = build_report_data(
0,
crate::opcreds::OPERATIONAL_CREDENTIALS_CLUSTER,
crate::opcreds::ATTR_CURRENT_FABRIC_INDEX,
&matter_codec::Value::Uint(1),
);
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let err = controller
.node(device_node_id)
.remove_fabric(1)
.await
.unwrap_err();
assert!(
matches!(err, crate::error::Error::WouldRemoveSelf),
"expected WouldRemoveSelf, got {err:?}"
);
device.await.unwrap();
}
#[tokio::test]
async fn remove_fabric_removes_other_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let replies = vec![
build_report_data(
0,
crate::opcreds::OPERATIONAL_CREDENTIALS_CLUSTER,
crate::opcreds::ATTR_CURRENT_FABRIC_INDEX,
&matter_codec::Value::Uint(1),
),
build_invoke_response_noc(0, Some(2)),
];
let device = tokio::spawn(run_loopback_device_seq(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
replies,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.remove_fabric(2)
.await
.expect("remove fabric 2 must succeed");
device.await.unwrap();
}
#[tokio::test]
async fn remove_fabric_fails_closed_when_fabric_index_unreadable() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = build_report_data(
0,
crate::opcreds::OPERATIONAL_CREDENTIALS_CLUSTER,
0x0001, &matter_codec::Value::Array(vec![]),
);
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
reply,
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let err = controller
.node(device_node_id)
.remove_fabric(2)
.await
.unwrap_err();
assert!(
matches!(err, crate::error::Error::Operational(_)),
"expected Operational error when CurrentFabricIndex unreadable, got {err:?}"
);
device.await.unwrap();
}
fn build_write_response_acl_success() -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.start_array(Tag::Context(0)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(2), 0).unwrap(); w.put_uint(Tag::Context(3), 0x001F).unwrap(); w.put_uint(Tag::Context(4), 0x0000).unwrap(); w.end_container().unwrap();
w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), 0).unwrap(); w.end_container().unwrap();
w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap(); w.end_container().unwrap();
buf
}
#[tokio::test]
async fn write_acl_single_chunk_round_trip() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = build_write_response_acl_success();
let device = tokio::spawn(run_chunked_write_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x60,
1,
reply,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let our_node_id: u64 = 1;
let entries = vec![crate::acl::AclEntry {
privilege: crate::acl::AclPrivilege::Administer,
auth_mode: crate::acl::AclAuthMode::Case,
subjects: Some(vec![our_node_id]),
targets: None,
fabric_index: None,
}];
let statuses = controller
.node(device_node_id)
.write_acl(&entries)
.await
.expect("write_acl must succeed");
assert_eq!(statuses.len(), 1);
assert_eq!(
statuses[0].1,
matter_interaction::ImStatus::Success,
"device must reply Success"
);
assert_eq!(statuses[0].0.cluster, crate::acl::ACCESS_CONTROL_CLUSTER);
assert_eq!(statuses[0].0.attribute, crate::acl::ATTR_ACL);
device.await.unwrap();
}
#[tokio::test]
async fn write_acl_refuses_lockout() {
let Harness {
store,
ctrl_io,
discovery,
..
} = loopback_harness();
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let entries = vec![crate::acl::AclEntry {
privilege: crate::acl::AclPrivilege::Operate,
auth_mode: crate::acl::AclAuthMode::Case,
subjects: Some(vec![1]),
targets: None,
fabric_index: None,
}];
let err = controller
.node(42) .write_acl(&entries)
.await
.unwrap_err();
assert!(
matches!(err, crate::error::Error::AclWouldLockOut),
"expected AclWouldLockOut, got {err:?}"
);
}
#[tokio::test]
async fn write_acl_multi_chunk_reassembles() {
use matter_codec::{Tag, TlvWriter};
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let path = matter_interaction::AttributePath {
endpoint: 0,
cluster: crate::acl::ACCESS_CONTROL_CLUSTER,
attribute: crate::acl::ATTR_ACL,
};
let elems: Vec<Vec<u8>> = (0u64..4)
.map(|n| {
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.put_uint(Tag::Anonymous, n).unwrap();
buf
})
.collect();
let chunks = matter_interaction::build_list_write_chunks(path, &elems, 40, false);
assert!(
chunks.len() >= 2,
"test requires multi-chunk write; got {} chunk(s)",
chunks.len()
);
let n_chunks = chunks.len();
let write_response = build_write_response_acl_success();
let device = tokio::spawn(run_chunked_write_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x61,
n_chunks,
write_response,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let statuses = controller
.node(device_node_id)
.chunked_write(chunks)
.await
.expect("chunked_write must succeed");
assert_eq!(statuses.len(), n_chunks);
for (_, status) in &statuses {
assert_eq!(*status, matter_interaction::ImStatus::Success);
}
device.await.unwrap();
}
fn build_write_response_binding_success() -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
w.start_structure(Tag::Anonymous).unwrap();
w.start_array(Tag::Context(0)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(2), 1).unwrap(); w.put_uint(Tag::Context(3), 0x001E).unwrap(); w.put_uint(Tag::Context(4), 0x0000).unwrap(); w.end_container().unwrap();
w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), 0).unwrap(); w.end_container().unwrap();
w.end_container().unwrap();
w.end_container().unwrap();
w.put_uint(Tag::Context(0xFF), 11).unwrap();
w.end_container().unwrap();
buf
}
#[tokio::test]
async fn write_binding_single_chunk_round_trip() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_chunked_write_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x61,
1,
build_write_response_binding_success(),
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let statuses = controller
.node(device_node_id)
.write_binding(
1,
&[crate::BindingTarget::new(
Some(0x1122),
None,
Some(1),
Some(0x0006),
)],
)
.await
.expect("write_binding");
assert_eq!(statuses.len(), 1);
assert_eq!(statuses[0].1, matter_interaction::ImStatus::Success);
assert_eq!(statuses[0].0.cluster, 0x001E);
assert_eq!(statuses[0].0.attribute, 0x0000);
device.await.unwrap();
}
#[allow(clippy::too_many_lines)]
#[tokio::test]
async fn write_acl_multi_chunk_via_dispatch() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let entries = vec![
crate::acl::AclEntry {
privilege: crate::acl::AclPrivilege::Administer,
auth_mode: crate::acl::AclAuthMode::Case,
subjects: Some(vec![1u64]),
targets: None,
fabric_index: None,
},
crate::acl::AclEntry {
privilege: crate::acl::AclPrivilege::Operate,
auth_mode: crate::acl::AclAuthMode::Case,
subjects: Some(vec![2u64]),
targets: None,
fabric_index: None,
},
crate::acl::AclEntry {
privilege: crate::acl::AclPrivilege::View,
auth_mode: crate::acl::AclAuthMode::Case,
subjects: Some(vec![3u64]),
targets: None,
fabric_index: None,
},
];
let acl_path = matter_interaction::AttributePath {
endpoint: 0,
cluster: crate::acl::ACCESS_CONTROL_CLUSTER,
attribute: crate::acl::ATTR_ACL,
};
let element_tlvs: Vec<Vec<u8>> = entries
.iter()
.map(|e| crate::node::value_to_tlv(&crate::acl::acl_entry_value(e)).expect("encode"))
.collect();
let expected_chunks =
matter_interaction::build_list_write_chunks(acl_path, &element_tlvs, 40, false).len();
assert!(
expected_chunks >= 2,
"test requires multi-chunk write; got {expected_chunks} chunk(s)"
);
let device = tokio::spawn(run_chunked_write_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x62,
expected_chunks,
build_write_response_acl_success(),
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let statuses = controller
.node(device_node_id)
.write_acl_with_budget(&entries, 40)
.await
.expect("write_acl_with_budget must succeed");
assert_eq!(statuses.len(), expected_chunks);
for (path, status) in &statuses {
assert_eq!(*status, matter_interaction::ImStatus::Success);
assert_eq!(path.cluster, crate::acl::ACCESS_CONTROL_CLUSTER);
assert_eq!(path.attribute, crate::acl::ATTR_ACL);
}
device.await.unwrap();
}
#[test]
fn write_acl_single_chunk_byte_parity() {
use crate::acl::{acl_entry_value, AclAuthMode, AclEntry, AclPrivilege};
use matter_codec::{Tag, TlvWriter};
use matter_interaction::AttributePath;
use std::{fs, path::PathBuf};
#[derive(serde::Deserialize)]
struct Fixture {
entry_tlv_hex: String,
expected_message_hex: String,
}
fn hex_decode(s: &str) -> Vec<u8> {
(0..s.len())
.step_by(2)
.map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap())
.collect()
}
let fixture_path: PathBuf = {
let mut p: PathBuf = env!("CARGO_MANIFEST_DIR").into();
p.push("..");
p.push("..");
p.push("test-vectors");
p.push("acl");
p.push("write_acl_single_chunk.json");
p
};
let Ok(raw) = fs::read_to_string(&fixture_path) else {
eprintln!("skipping write_acl_single_chunk_byte_parity: fixture not found");
return;
};
let f: Fixture = serde_json::from_str(&raw).unwrap();
let expected_entry_tlv = hex_decode(&f.entry_tlv_hex);
let expected_message = hex_decode(&f.expected_message_hex);
let entry = AclEntry {
privilege: AclPrivilege::Administer,
auth_mode: AclAuthMode::Case,
subjects: Some(vec![1u64]),
targets: None,
fabric_index: None,
};
let mut entry_tlv: Vec<u8> = Vec::new();
TlvWriter::new(&mut entry_tlv)
.write_value(Tag::Anonymous, &acl_entry_value(&entry))
.unwrap();
assert_eq!(
entry_tlv, expected_entry_tlv,
"acl_entry_value TLV does not match fixture entry_tlv_hex"
);
let path = AttributePath {
endpoint: 0,
cluster: crate::acl::ACCESS_CONTROL_CLUSTER,
attribute: crate::acl::ATTR_ACL,
};
let chunks = matter_interaction::build_list_write_chunks(path, &[entry_tlv], 4096, false);
assert_eq!(chunks.len(), 1, "must be single chunk with big budget");
assert_eq!(
chunks[0], expected_message,
"build_list_write_chunks single-chunk does not match fixture expected_message_hex"
);
}
#[tokio::test]
async fn update_fabric_label_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
build_invoke_response_noc(0, Some(1)),
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.update_fabric_label("living-room")
.await
.expect("relabel");
device.await.unwrap();
}
#[tokio::test]
async fn write_group_key_set_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
build_invoke_status_success(),
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let set = crate::group::GroupKeySet::new(42, vec![0xABu8; 16], 0);
controller
.node(device_node_id)
.write_group_key_set(&set)
.await
.expect("write_group_key_set");
device.await.unwrap();
}
fn build_write_response_group_key_map_success() -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
#[allow(clippy::unwrap_used)] {
w.start_structure(Tag::Anonymous).unwrap();
w.start_array(Tag::Context(0)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(2), 0).unwrap(); w.put_uint(Tag::Context(3), 0x003F).unwrap(); w.put_uint(Tag::Context(4), 0x0000).unwrap(); w.end_container().unwrap();
w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), 0).unwrap(); w.end_container().unwrap();
w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap(); w.end_container().unwrap();
}
buf
}
#[tokio::test]
async fn write_group_key_map_single_chunk_round_trip() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let reply = build_write_response_group_key_map_success();
let device = tokio::spawn(run_chunked_write_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x61,
1,
reply,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
let entries = vec![crate::group::GroupKeyMapEntry::new(7, 42)];
let statuses = controller
.node(device_node_id)
.write_group_key_map(&entries)
.await
.expect("write_group_key_map must succeed");
assert_eq!(statuses.len(), 1);
assert_eq!(
statuses[0].1,
matter_interaction::ImStatus::Success,
"device must reply Success"
);
assert_eq!(
statuses[0].0.cluster,
crate::group::GROUP_KEY_MANAGEMENT_CLUSTER
);
assert_eq!(statuses[0].0.attribute, crate::group::ATTR_GROUP_KEY_MAP);
device.await.unwrap();
}
fn build_add_group_response(status: u8, group_id: u16) -> Vec<u8> {
use matter_codec::{Tag, TlvWriter};
let mut buf = Vec::new();
let mut w = TlvWriter::new(&mut buf);
#[allow(clippy::unwrap_used)] {
w.start_structure(Tag::Anonymous).unwrap(); w.put_bool(Tag::Context(0), false).unwrap(); w.start_array(Tag::Context(1)).unwrap(); w.start_structure(Tag::Anonymous).unwrap(); w.start_structure(Tag::Context(0)).unwrap(); w.start_list(Tag::Context(0)).unwrap(); w.put_uint(Tag::Context(0), 1).unwrap(); w.put_uint(Tag::Context(1), u64::from(crate::group::GROUPS_CLUSTER))
.unwrap(); w.put_uint(Tag::Context(2), 0x00).unwrap(); w.end_container().unwrap(); w.start_structure(Tag::Context(1)).unwrap(); w.put_uint(Tag::Context(0), u64::from(status)).unwrap(); w.put_uint(Tag::Context(1), u64::from(group_id)).unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.end_container().unwrap(); w.put_uint(Tag::Context(0xFF), 11).unwrap(); w.end_container().unwrap(); }
buf
}
#[tokio::test]
async fn add_group_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
build_add_group_response(0, 7),
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.add_group(1, 7, "test")
.await
.expect("add_group must succeed");
device.await.unwrap();
}
#[tokio::test]
async fn remove_group_over_loopback() {
let Harness {
store,
ctrl_io,
dev_io,
ctrl_addr,
discovery,
device_creds,
device_roots,
device_node_id,
} = loopback_harness();
let device = tokio::spawn(run_loopback_device(
dev_io,
ctrl_addr,
device_creds,
device_roots,
0x55,
1,
build_add_group_response(0, 7),
false,
));
let controller = crate::controller::MatterController::with_components(
store,
ctrl_io,
discovery,
Arc::new(SystemNocRng),
None,
crate::builder::DEFAULT_ADMIN_VENDOR_ID,
)
.expect("open");
controller
.node(device_node_id)
.remove_group(1, 7)
.await
.expect("remove_group must succeed");
device.await.unwrap();
}
}