diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/config.rs | 264 | ||||
| -rw-r--r-- | src/controller.rs | 1332 | ||||
| -rw-r--r-- | src/host.rs | 88 | ||||
| -rw-r--r-- | src/main.rs | 374 | ||||
| -rw-r--r-- | src/output.rs | 444 | ||||
| -rw-r--r-- | src/pairing.rs | 129 | ||||
| -rw-r--r-- | src/state.rs | 194 | ||||
| -rw-r--r-- | src/time.rs | 367 | ||||
| -rw-r--r-- | src/tz.rs | 268 |
9 files changed, 3460 insertions, 0 deletions
diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..3e53975 --- /dev/null +++ b/src/config.rs @@ -0,0 +1,264 @@ +//! Validated JSON configuration, file-compatible with the TypeScript +//! implementation (`/etc/mattertimesync/config.json`, camelCase keys). +//! +//! Unknown fields are rejected loudly via serde's `deny_unknown_fields`; +//! device membership deliberately lives in controller storage, not here. + +use std::fmt; +use std::path::PathBuf; + +use jiff::tz::TimeZone; +use serde::Deserialize; + +pub const DEFAULT_CONFIG_PATH: &str = "/etc/mattertimesync/config.json"; + +/// Matter FabricDescriptorStruct label limit. +const FABRIC_LABEL_MAX_LENGTH: usize = 32; + +#[derive(Debug, thiserror::Error)] +pub enum ConfigError { + #[error("cannot read configuration file {path}: {source}")] + Unreadable { + path: PathBuf, + source: std::io::Error, + }, + #[error("configuration is not valid: {0}")] + Invalid(#[from] serde_json::Error), + #[error("\"timezone\" must be a valid IANA time-zone name (got {0:?})")] + BadTimezone(String), + #[error( + "\"fabricLabel\" must be a non-empty string of at most {FABRIC_LABEL_MAX_LENGTH} characters" + )] + BadFabricLabel, + #[error( + "\"storagePath\" must be an absolute path (got {0:?}); JSON configs get no shell expansion" + )] + RelativeStoragePath(PathBuf), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum LogLevel { + Debug, + #[default] + Info, + Warn, + Error, +} + +impl From<LogLevel> for log::LevelFilter { + fn from(level: LogLevel) -> Self { + match level { + LogLevel::Debug => log::LevelFilter::Debug, + LogLevel::Info => log::LevelFilter::Info, + LogLevel::Warn => log::LevelFilter::Warn, + LogLevel::Error => log::LevelFilter::Error, + } + } +} + +impl fmt::Display for LogLevel { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let name = match self { + LogLevel::Debug => "debug", + LogLevel::Info => "info", + LogLevel::Warn => "warn", + LogLevel::Error => "error", + }; + f.write_str(name) + } +} + +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct Config { + /// Where this configuration was loaded from; not a config field. + #[serde(skip)] + pub source: PathBuf, + /// Directory holding persistent Matter fabric state and service state. + /// Contains the controller's private keys; mode 0700. + pub storage_path: PathBuf, + /// IANA time-zone name, e.g. "America/Chicago". Never a fixed UTC offset. + #[serde(default = "default_timezone")] + pub timezone: String, + #[serde(default)] + pub log_level: LogLevel, + /// Fabric label other ecosystems display for this controller (e.g. the + /// Apple Home Connected Services subtitle). Must be unique per device. + #[serde(default = "default_fabric_label")] + pub fabric_label: String, +} + +fn default_timezone() -> String { + "America/Chicago".into() +} + +fn default_fabric_label() -> String { + "Matter Time Sync".into() +} + +impl Config { + pub fn load(path: &std::path::Path) -> Result<Self, ConfigError> { + let raw = std::fs::read_to_string(path).map_err(|source| ConfigError::Unreadable { + path: path.to_owned(), + source, + })?; + let mut config = Self::parse(&raw)?; + config.source = path.to_owned(); + Ok(config) + } + + pub fn parse(raw: &str) -> Result<Self, ConfigError> { + let config: Config = serde_json::from_str(raw)?; + config.validate()?; + Ok(config) + } + + fn validate(&self) -> Result<(), ConfigError> { + // No shell expansion happens on a JSON file, so a "~/..." or relative + // path would silently land wherever the process happens to run. + if !self.storage_path.is_absolute() { + return Err(ConfigError::RelativeStoragePath(self.storage_path.clone())); + } + // Resolving through jiff's tzdb is the validation; a bare offset or + // invented name fails here rather than at 3am on a DST transition. + TimeZone::get(&self.timezone) + .map_err(|_| ConfigError::BadTimezone(self.timezone.clone()))?; + if self.fabric_label.trim().is_empty() || self.fabric_label.len() > FABRIC_LABEL_MAX_LENGTH + { + return Err(ConfigError::BadFabricLabel); + } + Ok(()) + } + + /// Warnings for path components that look like they expected shell + /// expansion: a component starting with `~` or containing `$`. Such + /// directories can legitimately exist, so these cannot be errors; but + /// far more often they mean the config was written expecting a shell to + /// expand it, and the data would land in a literal `~foo` directory. + pub fn path_warnings(&self) -> Vec<String> { + self.storage_path + .components() + .filter_map(|component| { + let text = component.as_os_str().to_string_lossy(); + let looks_like = if text.starts_with('~') { + "a shell tilde" + } else if text.contains('$') { + "an unexpanded shell variable" + } else { + return None; + }; + Some(format!( + "storagePath component {text:?} looks like {looks_like}; JSON configs get no \ + shell expansion, so it will be used as a literal directory name" + )) + }) + .collect() + } + + /// The configured zone, resolved against the system tzdb. + pub fn time_zone(&self) -> TimeZone { + // Validated at load time; a tzdb that shrinks between then and now is + // not a scenario worth threading a Result through every caller for. + TimeZone::get(&self.timezone).expect("timezone was validated at config load") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const MINIMAL: &str = r#"{ "storagePath": "/var/lib/mattertimesync" }"#; + + #[test] + fn minimal_config_applies_defaults() { + let config = Config::parse(MINIMAL).unwrap(); + assert_eq!( + config.storage_path, + PathBuf::from("/var/lib/mattertimesync") + ); + assert_eq!(config.timezone, "America/Chicago"); + assert_eq!(config.log_level, LogLevel::Info); + assert_eq!(config.fabric_label, "Matter Time Sync"); + } + + #[test] + fn full_config_round_trips() { + let config = Config::parse( + r#"{ + "storagePath": "/tmp/x", + "timezone": "Europe/Berlin", + "logLevel": "debug", + "fabricLabel": "Lakeside Time Sync" + }"#, + ) + .unwrap(); + assert_eq!(config.timezone, "Europe/Berlin"); + assert_eq!(config.log_level, LogLevel::Debug); + assert_eq!(config.fabric_label, "Lakeside Time Sync"); + } + + #[test] + fn unknown_fields_are_rejected_loudly() { + // Fields from abandoned designs are rejected, not ignored. + for raw in [ + r#"{ "storagePath": "/x", "nodeId": "1" }"#, + r#"{ "storagePath": "/x", "syncIntervalHours": 24 }"#, + r#"{ "storagePath": "/x", "unexpected": 1 }"#, + ] { + let error = Config::parse(raw).unwrap_err(); + assert!(error.to_string().contains("unknown field"), "{error}"); + } + } + + #[test] + fn suspicious_path_components_warn_but_load() { + let config = Config::parse(r#"{ "storagePath": "/data/~backup" }"#).unwrap(); + assert_eq!(config.path_warnings().len(), 1); + assert!(config.path_warnings()[0].contains("shell tilde")); + + let config = Config::parse(r#"{ "storagePath": "/var/lib/$USER/mts" }"#).unwrap(); + assert!(config.path_warnings()[0].contains("unexpanded shell variable")); + + let config = Config::parse(r#"{ "storagePath": "/var/lib/mattertimesync" }"#).unwrap(); + assert!(config.path_warnings().is_empty()); + } + + #[test] + fn non_absolute_storage_paths_are_rejected() { + for path in ["~/mts-storage", "data", "./data"] { + let raw = format!(r#"{{ "storagePath": {path:?} }}"#); + let error = Config::parse(&raw).unwrap_err(); + assert!( + error.to_string().contains("absolute"), + "accepted {path:?}: {error}" + ); + } + } + + #[test] + fn storage_path_is_required() { + assert!(Config::parse(r#"{}"#).is_err()); + } + + #[test] + fn invalid_timezones_are_rejected() { + for tz in ["Central Time", "", "America/Springfield", "UTC-6"] { + let raw = format!(r#"{{ "storagePath": "/x", "timezone": {tz:?} }}"#); + assert!(Config::parse(&raw).is_err(), "accepted {tz:?}"); + } + } + + #[test] + fn invalid_log_levels_are_rejected() { + assert!(Config::parse(r#"{ "storagePath": "/x", "logLevel": "verbose" }"#).is_err()); + } + + #[test] + fn invalid_fabric_labels_are_rejected() { + for label in ["", " ", &"x".repeat(33)] { + let raw = format!(r#"{{ "storagePath": "/x", "fabricLabel": {label:?} }}"#); + assert!(Config::parse(&raw).is_err(), "accepted {label:?}"); + } + } +} diff --git a/src/controller.rs b/src/controller.rs new file mode 100644 index 0000000..675f58d --- /dev/null +++ b/src/controller.rs @@ -0,0 +1,1332 @@ +//! The rs-matter controller: persistent fabric, one-shot stack harness, and +//! the on-wire operations (commission, sync, decommission). +//! +//! Being a secondary Matter controller is fabric membership, not a running +//! process: every command builds the stack, races the transport and mDNS +//! pumps against the actual operation, and exits. The fabric identity +//! persists in `<storagePath>/matter/` (rs-matter's KV blobs) plus +//! `<storagePath>/identity.json` (the RCAC private key that signs device +//! NOCs, which rs-matter deliberately leaves to the caller). + +use std::net::UdpSocket; +use std::num::NonZeroU8; +use std::path::Path; +use std::pin::pin; + +use anyhow::{Context as _, anyhow, bail}; +use embassy_futures::select::{Either3, select3}; +use rs_matter::Matter; +use rs_matter::cert::MAX_CERT_TLV_AND_ASN1_LEN; +use rs_matter::cert::r#gen::VALID_FOREVER; +use rs_matter::crypto::{ + CanonAeadKey, CanonPkcSecretKey, Crypto, RngCore as _, SecretKey, SigningSecretKey, + default_crypto, +}; +use rs_matter::dm::devices::test::{DAC_PRIVKEY, TEST_DEV_ATT, TEST_DEV_COMM, TEST_DEV_DET}; +use rs_matter::error::Error as MatterError; +use rs_matter::fabric::FabricPersist; +use rs_matter::onboard::cac::RcacGenerator; +use rs_matter::onboard::noc::NocGenerator; +use rs_matter::transport::network::mdns::CommissionableFilter; +use rs_matter::transport::network::mdns::builtin::{BuiltinMdns, Host}; +use rs_matter::transport::network::mdns::{ + MDNS_IPV6_BROADCAST_ADDR, MDNS_SOCKET_DEFAULT_BIND_ADDR, +}; +use rs_matter::utils::init::InitMaybeUninit; +use serde::{Deserialize, Serialize}; +use static_cell::StaticCell; + +use crate::config::Config; +use crate::state; + +/// The identity material rs-matter leaves to the caller, persisted as JSON +/// with the RCAC private key hex-encoded. Mode 0600, next to the KV blobs. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Identity { + pub fabric_id: u64, + pub controller_node_id: u64, + pub next_device_node_id: u64, + pub rcac_privkey_hex: String, +} + +/// What `status` reports about the controller identity, derived purely from +/// the on-disk artifacts (no Matter stack startup). +#[derive(Debug)] +pub enum IdentityStatus { + /// Neither identity.json nor a fabric blob: first commission creates it. + NotCreated, + Created { + fabric_id: u64, + controller_node_id: u64, + }, + /// identity.json exists but cannot be read or parsed (typically + /// permissions: it is mode 0600, owned by the service user). + Unreadable, + /// One half of the identity pair is missing; the fabric-writing commands + /// will refuse or auto-recover per their guardrails. + Inconsistent(&'static str), +} + +/// Inspects the identity pair on disk. A fabric blob is any k_* file in the +/// matter/ KV directory. +pub fn identity_status(storage: &Path) -> IdentityStatus { + let identity_exists = identity_path(storage).exists(); + let fabric_exists = std::fs::read_dir(matter_kv_path(storage)) + .map(|entries| { + entries + .flatten() + .any(|e| e.file_name().to_string_lossy().starts_with("k_")) + }) + .unwrap_or(false); + match (identity_exists, fabric_exists) { + (false, false) => IdentityStatus::NotCreated, + (true, false) => { + IdentityStatus::Inconsistent("identity.json exists but the fabric storage is missing") + } + (false, true) => { + IdentityStatus::Inconsistent("fabric storage exists but identity.json is missing") + } + (true, true) => match std::fs::read_to_string(identity_path(storage)) + .ok() + .and_then(|raw| serde_json::from_str::<Identity>(&raw).ok()) + { + Some(identity) => IdentityStatus::Created { + fabric_id: identity.fabric_id, + controller_node_id: identity.controller_node_id, + }, + None => IdentityStatus::Unreadable, + }, + } +} + +fn identity_path(storage: &Path) -> std::path::PathBuf { + storage.join("identity.json") +} + +fn matter_kv_path(storage: &Path) -> std::path::PathBuf { + storage.join("matter") +} + +impl Identity { + fn generate(crypto: &impl Crypto) -> Result<Self, MatterError> { + let mut rng = crypto.rand()?; + let mut bytes = [0u8; 8]; + rng.fill_bytes(&mut bytes); + // Non-zero 64-bit fabric id; operational node ids for devices are + // small and sequential like matter.js's. + let fabric_id = u64::from_be_bytes(bytes) | 1; + let mut node_bytes = [0u8; 8]; + rng.fill_bytes(&mut node_bytes); + Ok(Self { + fabric_id, + controller_node_id: u64::from_be_bytes(node_bytes) | 1, + next_device_node_id: 1, + rcac_privkey_hex: String::new(), + }) + } + + fn rcac_privkey(&self) -> anyhow::Result<CanonPkcSecretKey> { + let bytes = hex_decode(&self.rcac_privkey_hex).context("identity.json rcac key")?; + let mut key = CanonPkcSecretKey::new(); + if bytes.len() != key.access_mut().len() { + bail!( + "identity.json rcac key has unexpected length {}", + bytes.len() + ); + } + key.access_mut().copy_from_slice(&bytes); + Ok(key) + } +} + +fn hex_encode(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +fn hex_decode(hex: &str) -> anyhow::Result<Vec<u8>> { + if !hex.len().is_multiple_of(2) { + bail!("odd-length hex"); + } + (0..hex.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).context("bad hex digit")) + .collect() +} + +static MATTER: StaticCell<Matter> = StaticCell::new(); + +/// Flat-file KV store with atomic writes, stand-in for rs-matter's +/// `DirKvBlobStore` (whose `store()` is a plain overwrite). The fabric +/// record lives here and must never be half-written; a temp file plus +/// rename makes every update all-or-nothing. Same `k_XXXX` file naming. +struct AtomicKvBlobStore(std::path::PathBuf); + +impl AtomicKvBlobStore { + fn key_path(&self, key: u16) -> std::path::PathBuf { + self.0.join(format!("k_{key:04x}")) + } +} + +impl rs_matter::persist::KvBlobStore for AtomicKvBlobStore { + fn load<'a>(&mut self, key: u16, buf: &'a mut [u8]) -> Result<Option<&'a [u8]>, MatterError> { + match std::fs::read(self.key_path(key)) { + Ok(data) => { + let slot = buf + .get_mut(..data.len()) + .ok_or(rs_matter::error::ErrorCode::NoSpace)?; + slot.copy_from_slice(&data); + Ok(Some(slot)) + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(_) => Err(rs_matter::error::ErrorCode::StdIoError.into()), + } + } + + fn store(&mut self, key: u16, data: &[u8], _buf: &mut [u8]) -> Result<(), MatterError> { + let path = self.key_path(key); + let write = || -> std::io::Result<()> { + std::fs::create_dir_all(self.0.as_path())?; + let tmp = path.with_extension("tmp"); + state::write_private(&tmp, data)?; + std::fs::rename(&tmp, &path) + }; + write().map_err(|_| rs_matter::error::ErrorCode::StdIoError.into()) + } + + fn remove(&mut self, key: u16, _buf: &mut [u8]) -> Result<(), MatterError> { + match std::fs::remove_file(self.key_path(key)) { + Ok(()) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(_) => Err(rs_matter::error::ErrorCode::StdIoError.into()), + } + } +} + +/// Everything an operation needs, borrowed from the harness scope. +pub struct Ctx<'a, C: Crypto> { + pub matter: &'a Matter<'a>, + pub crypto: &'a C, + pub fab_idx: NonZeroU8, + pub identity: Identity, + pub config: &'a Config, +} + +/// A controller operation, generic over the crypto backend the harness +/// picks. A trait (rather than an async closure) because Rust forbids a +/// nested `impl Trait` in closure parameters. +pub trait Op { + type Out; + async fn run<C: Crypto>(self, ctx: &Ctx<'_, C>) -> anyhow::Result<Self::Out>; +} + +/// Builds the Matter stack, ensures the persistent controller fabric exists, +/// then races the transport and mDNS pumps against `op`. The pumps never +/// finish on their own; `op` completing ends the run. +pub fn run<O: Op>(config: &Config, op: O) -> anyhow::Result<O::Out> { + std::fs::create_dir_all(&config.storage_path).context("create storage directory")?; + restrict_dir_mode(&config.storage_path)?; + // One controller process per storage directory. Held for the process + // lifetime (flock releases on exit, so no stale-lock handling); this + // makes the first-run identity bootstrap race-free when a manual run + // overlaps the timer. + let _lock = lock_storage(&config.storage_path)?; + + futures_lite::future::block_on(async { + let crypto = default_crypto(rand::thread_rng(), DAC_PRIVKEY); + let matter: &'static Matter = + MATTER + .uninit() + .init_with(Matter::init(&TEST_DEV_DET, TEST_DEV_COMM, &TEST_DEV_ATT, 0)); + + let kv_store = AtomicKvBlobStore(matter_kv_path(&config.storage_path)); + let kv = matter.kv(kv_store); + matter + .load_persist(&kv) + .await + .map_err(|e| anyhow!("load persisted Matter state: {e:?}"))?; + + let (fab_idx, identity) = ensure_fabric(config, matter, &crypto, &kv)?; + let ctx = Ctx { + matter, + crypto: &crypto, + fab_idx, + identity, + config, + }; + + // Matter transport on an ephemeral port. + let socket = + async_io::Async::<UdpSocket>::bind(([0u16; 8], 0)).context("bind Matter UDP socket")?; + let transport = matter.run(&crypto, &socket, &socket, &socket); + + // mDNS: the builtin responder everywhere. On the deployment + // platform (Linux) it owns port 5353 outright and behaves + // deterministically. On a macOS dev machine it shares 5353 with + // mDNSResponder and resolve answers occasionally race to the wrong + // socket; the connect retries absorb most of that. (The system + // dnssd backend was tried and is worse: its 10-item browse channel + // deterministically drops records on Matter-dense networks.) + let mdns_socket = bind_mdns_socket().context("bind mDNS socket")?; + let (mdns_host_ipv4, mdns_ipv6, mdns_interface) = pick_interface()?; + let hostname = format!( + "{:012X}", + ctx.identity.controller_node_id & 0xFFFF_FFFF_FFFF + ); + let host = Host { + hostname: &hostname, + ip: mdns_host_ipv4, + ipv6: mdns_ipv6, + }; + let mut mdns_runner = BuiltinMdns::new(); + // ipv4_interface: None disables the runner's IPv4 send path + // entirely; see bind_mdns_socket for why mDNS is IPv6-only here. + let mdns = mdns_runner.run( + &mdns_socket, + &mdns_socket, + &host, + None, + Some(mdns_interface), + matter, + &crypto, + ); + + let mut transport = pin!(transport); + let mut mdns = pin!(mdns); + let mut op_fut = pin!(op.run(&ctx)); + + match select3(&mut transport, &mut mdns, &mut op_fut).await { + Either3::First(r) => bail!("Matter transport exited prematurely: {r:?}"), + Either3::Second(r) => bail!("mDNS runner exited prematurely: {r:?}"), + Either3::Third(result) => result, + } + }) +} + +/// Loads or bootstraps the controller fabric. On first run this mints the +/// RCAC (whose private key we keep, RCAC-direct mode), our own operational +/// NOC, and the fabric IPK, then installs and persists the fabric. On later +/// runs the fabric comes back via rs-matter's KV persistence and only the +/// label is reconciled with the configuration. +fn ensure_fabric<C: Crypto>( + config: &Config, + matter: &Matter<'static>, + crypto: &C, + kv: &impl rs_matter::persist::KvBlobStoreAccess, +) -> anyhow::Result<(NonZeroU8, Identity)> { + let identity_file = identity_path(&config.storage_path); + let existing = matter.with_state(|state| state.fabrics.iter().map(|f| f.fab_idx()).next()); + + let registry: crate::state::NodeRegistry = + state::load_or_default(&state::registry_path(&config.storage_path)); + + if let Some(fab_idx) = existing { + match std::fs::read_to_string(&identity_file) { + Ok(raw) => { + let identity: Identity = + serde_json::from_str(&raw).context("parse identity.json")?; + reconcile_label(matter, crypto, kv, fab_idx, &config.fabric_label)?; + return Ok((fab_idx, identity)); + } + // A fabric blob without identity.json is an interrupted first + // run: identity.json is written last, and commissioning requires + // a completed bootstrap. With an empty registry nothing can + // reference this fabric, so discard it and bootstrap cleanly. + // With a non-empty registry something is deeply wrong; never + // guess. + Err(_) if registry.nodes.is_empty() => { + log::warn!( + "Discarding fabric left by an interrupted first run (no identity.json, \ + no commissioned devices); creating a fresh controller identity" + ); + matter + .with_state(|state| state.fabrics.remove(fab_idx)) + .map_err(|e| anyhow!("remove interrupted fabric: {e:?}"))?; + remove_persisted_fabric(kv, fab_idx)?; + } + Err(e) => { + bail!( + "cannot read {} but the node registry lists {} commissioned device(s); \ + refusing to touch the fabric. Restore identity.json from backup. ({e})", + identity_file.display(), + registry.nodes.len() + ); + } + } + } + + // Refuse to mint a fresh identity over the remains of an old one. An + // identity.json without a loadable fabric means the KV storage was lost + // or damaged AFTER commissioning began; overwriting the RCAC key would + // permanently orphan every device commissioned to it. Deleting the whole + // storage directory (after decommissioning, or accepting the orphans) is + // a deliberate human act, never something a run does implicitly. + if identity_file.exists() { + bail!( + "{} exists but no fabric was loaded from {}; refusing to create a new controller \ + identity over an old one. Restore the matter/ KV files from backup, or delete the \ + whole storage directory to deliberately start over.", + identity_file.display(), + matter_kv_path(&config.storage_path).display() + ); + } + if !registry.nodes.is_empty() { + bail!( + "the node registry lists {} commissioned device(s) but no fabric was loaded; \ + refusing to create a new controller identity. Restore the storage directory from \ + backup.", + registry.nodes.len() + ); + } + + log::info!("First run: creating the controller fabric (persistent identity)"); + let mut identity = Identity::generate(crypto).map_err(|e| anyhow!("{e:?}"))?; + + let mut rcac_buf = [0u8; MAX_CERT_TLV_AND_ASN1_LEN]; + let mut rcac_gen = RcacGenerator::new(&mut rcac_buf); + let (rcac_priv, rcac) = rcac_gen + .generate(crypto, identity.fabric_id, VALID_FOREVER) + .map_err(|e| anyhow!("generate RCAC: {e:?}"))?; + identity.rcac_privkey_hex = hex_encode(rcac_priv.reference().access()); + + // Our own operational identity: keypair, CSR, NOC signed by the RCAC. + let controller_key = crypto + .generate_secret_key() + .map_err(|e| anyhow!("generate controller key: {e:?}"))?; + let mut csr_buf = [0u8; 256]; + let csr = controller_key + .csr(&mut csr_buf) + .map_err(|e| anyhow!("controller CSR: {e:?}"))?; + let mut controller_key_canon = CanonPkcSecretKey::new(); + controller_key + .write_canon(&mut controller_key_canon) + .map_err(|e| anyhow!("{e:?}"))?; + + let mut noc_buf = [0u8; MAX_CERT_TLV_AND_ASN1_LEN]; + let mut noc_generator = NocGenerator::create(rcac_priv.reference(), rcac, &[], &mut noc_buf) + .map_err(|e| anyhow!("NOC generator: {e:?}"))?; + let controller_noc = noc_generator + .generate(crypto, csr, identity.controller_node_id, &[], VALID_FOREVER) + .map_err(|e| anyhow!("controller NOC: {e:?}"))?; + + let mut ipk = CanonAeadKey::new(); + crypto + .rand() + .map_err(|e| anyhow!("{e:?}"))? + .fill_bytes(ipk.access_mut()); + + let fab_idx = matter + .with_state(|state| { + let fab_idx = state + .fabrics + .add( + crypto, + controller_key_canon.reference(), + rcac, + controller_noc, + &[], + Some(ipk.reference()), + TEST_VENDOR_ID, + identity.controller_node_id, + )? + .fab_idx(); + let _ = state.fabrics.update_label(fab_idx, &config.fabric_label); + Ok::<_, MatterError>(fab_idx) + }) + .map_err(|e| anyhow!("install controller fabric: {e:?}"))?; + + persist_fabric(matter, kv, fab_idx)?; + + // Identity file after the fabric: a crash in between leaves a fabric + // without identity.json, which the next run reports loudly rather than + // silently minting a second identity. + state::store(&identity_file, &identity).context("write identity.json")?; + log::info!( + "Controller fabric created and persisted (fabric id {}, controller node id {})", + identity.fabric_id, + identity.controller_node_id + ); + Ok((fab_idx, identity)) +} + +/// CSA test vendor id: the correct value for an uncertified controller. +const TEST_VENDOR_ID: u16 = 0xFFF1; + +fn reconcile_label<C: Crypto>( + matter: &Matter<'static>, + _crypto: &C, + kv: &impl rs_matter::persist::KvBlobStoreAccess, + fab_idx: NonZeroU8, + label: &str, +) -> anyhow::Result<()> { + let changed = matter.with_state(|state| { + let current = state + .fabrics + .get(fab_idx) + .map(|f| f.label().to_string()) + .unwrap_or_default(); + if current == label { + return Ok::<_, MatterError>(false); + } + state.fabrics.update_label(fab_idx, label)?; + Ok(true) + }); + match changed { + Ok(true) => persist_fabric(matter, kv, fab_idx), + Ok(false) => Ok(()), + Err(e) => Err(anyhow!("update local fabric label: {e:?}")), + } +} + +fn remove_persisted_fabric( + kv: &impl rs_matter::persist::KvBlobStoreAccess, + fab_idx: NonZeroU8, +) -> anyhow::Result<()> { + let mut persist = FabricPersist::new(kv); + persist + .remove(fab_idx) + .and_then(|()| persist.run()) + .map_err(|e: MatterError| anyhow!("remove persisted fabric blob: {e:?}")) +} + +fn persist_fabric( + matter: &Matter<'static>, + kv: &impl rs_matter::persist::KvBlobStoreAccess, + fab_idx: NonZeroU8, +) -> anyhow::Result<()> { + matter + .with_state(|state| { + let fabric = state + .fabrics + .get(fab_idx) + .ok_or(rs_matter::error::ErrorCode::NotFound)?; + let mut persist = FabricPersist::new(kv); + persist.store(fabric)?; + persist.run() + }) + .map_err(|e: MatterError| anyhow!("persist fabric: {e:?}")) +} + +#[cfg(unix)] +fn lock_storage(storage: &Path) -> anyhow::Result<std::fs::File> { + use std::os::fd::AsRawFd; + let path = storage.join(".lock"); + let file = std::fs::File::create(&path).context("create storage lock file")?; + let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) }; + if rc != 0 { + bail!( + "another mattertimesync instance is already running against {}", + storage.display() + ); + } + Ok(file) +} + +#[cfg(not(unix))] +fn lock_storage(_storage: &Path) -> anyhow::Result<std::fs::File> { + anyhow::bail!("only unix hosts are supported") +} + +#[cfg(unix)] +fn restrict_dir_mode(path: &Path) -> anyhow::Result<()> { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)) + .context("chmod storage directory") +} + +#[cfg(not(unix))] +fn restrict_dir_mode(_path: &Path) -> anyhow::Result<()> { + Ok(()) +} + +fn bind_mdns_socket() -> anyhow::Result<async_io::Async<UdpSocket>> { + use socket2::{Domain, Protocol, Socket, Type}; + let socket = Socket::new(Domain::IPV6, Type::DGRAM, Some(Protocol::UDP))?; + socket.set_reuse_address(true)?; + // macOS requires SO_REUSEPORT (not just SO_REUSEADDR) to share 5353 + // with the system's own mDNSResponder; Linux is lenient either way. + #[cfg(unix)] + socket.set_reuse_port(true)?; + socket.set_only_v6(false)?; + socket.bind(&MDNS_SOCKET_DEFAULT_BIND_ADDR.into())?; + let socket = async_io::Async::<UdpSocket>::new_nonblocking(socket.into())?; + + // IPv6 only, everywhere. Matter mandates IPv6 (Thread addresses are + // IPv6-only and every Matter device advertises over IPv6 mDNS), so the + // IPv4 side would add nothing but platform variance: macOS refuses IPv4 + // operations on an IPv6 socket that Linux tolerates. Without the group + // join there is no IPv4 mDNS at all, quietly and identically on every + // host. A failed IPv6 join, by contrast, means discovery cannot work. + let (_ipv4, _ipv6, interface) = pick_interface()?; + socket + .get_ref() + .join_multicast_v6(&MDNS_IPV6_BROADCAST_ADDR, interface) + .context("join mDNS IPv6 multicast group")?; + Ok(socket) +} + +/// Picks the LAN interface for mDNS: the first non-loopback interface with +/// both an IPv6 address (preferring link-local) and an IPv4 address. +fn pick_interface() -> anyhow::Result<(std::net::Ipv4Addr, std::net::Ipv6Addr, u32)> { + let all = if_addrs::get_if_addrs().context("enumerate network interfaces")?; + let candidate = |v6_filter: fn(std::net::Ipv6Addr) -> bool| { + all.iter() + .filter(|ia| !ia.is_loopback()) + .filter_map(|ia| match ia.addr { + if_addrs::IfAddr::V6(ref v6) if v6_filter(v6.ip) => { + Some((ia.name.clone(), v6.ip, ia.index.unwrap_or(0))) + } + _ => None, + }) + .find_map(|(name, ipv6, index)| { + all.iter() + .filter(|ia| ia.name == name) + .find_map(|ia| match ia.addr { + if_addrs::IfAddr::V4(ref v4) => Some((v4.ip, ipv6, index)), + _ => None, + }) + }) + }; + candidate(|ip| ip.is_unicast_link_local()) + .or_else(|| candidate(|_| true)) + .ok_or_else(|| anyhow!("no network interface with IPv4 + IPv6 found for mDNS")) +} + +/// Commissionable-browse filter for a manual pairing code. +pub fn commissionable_filter(short_discriminator: u8) -> CommissionableFilter { + CommissionableFilter { + short_discriminator: Some(short_discriminator), + ..Default::default() + } +} + +// --------------------------------------------------------------------------- +// Operations +// --------------------------------------------------------------------------- + +use rs_matter::dm::clusters::decl::basic_information::BasicInformationClient; +use rs_matter::dm::clusters::decl::operational_credentials::OperationalCredentialsClient; +use rs_matter::dm::clusters::decl::time_synchronization::{ + GranularityEnum, TimeSourceEnum, TimeSynchronizationClient, +}; +use rs_matter::dm::endpoints::ROOT_ENDPOINT_ID; +use rs_matter::onboard::{CommissionOptions, Commissioner}; +use rs_matter::transport::exchange::Exchange; + +use jiff::Timestamp; + +use crate::pairing::Onboarding; +use crate::state::{NodeInfo, NodeRegistry}; + +const BROWSE_TIMEOUT_MS: u32 = 30_000; +/// Per-phase bound for commissioning (PASE handshake + invokes, CASE + complete). +const COMMISSION_TIMEOUT_SECS: u64 = 60; +/// Bound for opening a CASE exchange to a commissioned node (mDNS resolve + handshake). +const CONNECT_TIMEOUT_SECS: u64 = 30; + +/// Bounds a Matter operation that could otherwise hang (unreachable peer, +/// swallowed packets). The timeout is part of the caller's log line so an +/// operator watching a quiet log knows how long "waiting" can last. +async fn with_timeout<T>( + what: &str, + secs: u64, + fut: impl Future<Output = Result<T, MatterError>>, +) -> anyhow::Result<T> { + let mut fut = core::pin::pin!(fut); + let mut timer = core::pin::pin!(embassy_time::Timer::after( + embassy_time::Duration::from_secs(secs) + )); + match embassy_futures::select::select(&mut fut, &mut timer).await { + embassy_futures::select::Either::First(result) => { + result.map_err(|e| anyhow!("{what}: {e:?}")) + } + embassy_futures::select::Either::Second(()) => bail!("{what} timed out after {secs}s"), + } +} + +pub struct CommissionOp { + pub onboarding: Onboarding, +} + +#[derive(Debug, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CommissionOutcome { + pub node_id: crate::output::Id64, + pub fabric_id: crate::output::Id64, + pub vendor_name: Option<String>, + pub product_name: Option<String>, +} + +impl Op for CommissionOp { + type Out = CommissionOutcome; + + async fn run<C: Crypto>(self, ctx: &Ctx<'_, C>) -> anyhow::Result<CommissionOutcome> { + let matter = ctx.matter; + log::info!( + "Discovering commissionable device (_matterc._udp, short discriminator {}, timeout {}s)", + self.onboarding.short_discriminator, + BROWSE_TIMEOUT_MS / 1000 + ); + let (peer_addr, _instance) = matter + .transport() + .browse_commissionable( + &commissionable_filter(self.onboarding.short_discriminator), + &[], + BROWSE_TIMEOUT_MS, + ) + .await + .map_err(|e| { + anyhow!("no commissionable device found (is the pairing window open?): {e:?}") + })?; + log::info!("Found commissionable device at {:?}", peer_addr); + + let device_node_id = ctx.identity.next_device_node_id; + let rcac_privkey = ctx.identity.rcac_privkey()?; + let mut noc_buf = [0u8; MAX_CERT_TLV_AND_ASN1_LEN]; + let mut noc_generator = + NocGenerator::new(matter, rcac_privkey.reference(), ctx.fab_idx, &mut noc_buf) + .map_err(|e| anyhow!("NOC generator from persisted identity: {e:?}"))?; + + let mut commissioner_buf = [0u8; rs_matter::cert::MAX_CERT_TLV_LEN]; + let mut commissioner = Commissioner::new( + matter, + ctx.crypto, + ctx.fab_idx, + &mut noc_generator, + &mut commissioner_buf, + ); + let opts = CommissionOptions { + // Consumer devices carry vendor DACs we cannot verify without the + // DCL; matter.js accepted these the same way. + allow_test_attestation: true, + ..CommissionOptions::new() + }; + + log::info!( + "Commissioning as node {device_node_id} (PASE phase, timeout {COMMISSION_TIMEOUT_SECS}s)" + ); + let phase1 = with_timeout( + "commissioning over PASE", + COMMISSION_TIMEOUT_SECS, + commissioner.commission( + peer_addr, + self.onboarding.passcode, + &opts, + device_node_id, + VALID_FOREVER, + ), + ) + .await?; + log::info!("CASE phase: completing commissioning (timeout {COMMISSION_TIMEOUT_SECS}s)"); + with_timeout( + "CommissioningComplete over CASE", + COMMISSION_TIMEOUT_SECS, + commissioner.complete_via_case(peer_addr, &phase1), + ) + .await?; + + // Cache the device identity for the local registry; best-effort. + let vendor_name = read_vendor_name(ctx, device_node_id).await.ok(); + let product_name = read_product_name(ctx, device_node_id).await.ok(); + + let registry_file = state::registry_path(&ctx.config.storage_path); + let mut registry: NodeRegistry = state::load_or_default(®istry_file); + registry.nodes.insert( + device_node_id.to_string(), + NodeInfo { + vendor_name: vendor_name.clone(), + product_name: product_name.clone(), + }, + ); + state::store(®istry_file, ®istry)?; + + let mut identity = ctx.identity.clone(); + identity.next_device_node_id += 1; + state::store(&identity_path(&ctx.config.storage_path), &identity)?; + + state::update_node_state( + &state::service_state_path(&ctx.config.storage_path), + device_node_id, + |node| node.last_successful_connection = Some(Timestamp::now()), + )?; + + Ok(CommissionOutcome { + node_id: device_node_id.into(), + fabric_id: ctx.identity.fabric_id.into(), + vendor_name, + product_name, + }) + } +} + +async fn read_vendor_name<C: Crypto>(ctx: &Ctx<'_, C>, node_id: u64) -> anyhow::Result<String> { + let exchange = connect(ctx, node_id).await?; + let mut out = String::new(); + exchange + .basic_information() + .vendor_name_read_with(ROOT_ENDPOINT_ID, |value| { + out = value?.to_string(); + Ok::<_, MatterError>(()) + }) + .await + .map_err(|e| anyhow!("read vendorName: {e:?}"))? + .map_err(|e| anyhow!("parse vendorName: {e:?}"))?; + Ok(out) +} + +async fn read_product_name<C: Crypto>(ctx: &Ctx<'_, C>, node_id: u64) -> anyhow::Result<String> { + let exchange = connect(ctx, node_id).await?; + let mut out = String::new(); + exchange + .basic_information() + .product_name_read_with(ROOT_ENDPOINT_ID, |value| { + out = value?.to_string(); + Ok::<_, MatterError>(()) + }) + .await + .map_err(|e| anyhow!("read productName: {e:?}"))? + .map_err(|e| anyhow!("parse productName: {e:?}"))?; + Ok(out) +} + +/// How many times to attempt an mDNS resolve + CASE connect. Resolve +/// answers can be lost per-attempt (multicast loss, macOS socket sharing), +/// so single-shot failure is not conclusive. +const CONNECT_ATTEMPTS: u32 = 3; +/// Pause between connect attempts. A failed CASE handshake can linger +/// half-open on the device, which then answers an immediate re-knock with +/// Busy; a couple of seconds lets it reap the stale session. +const CONNECT_RETRY_DELAY_SECS: u64 = 2; + +/// Opens an exchange over CASE to a commissioned node (cached session when +/// available, mDNS operational resolve otherwise). One exchange = one IM +/// transaction, so every read/invoke starts here. +async fn connect<'a, C: Crypto>(ctx: &Ctx<'a, C>, node_id: u64) -> anyhow::Result<Exchange<'a>> { + let mut last_error = None; + for attempt in 1..=CONNECT_ATTEMPTS { + match with_timeout( + &format!("reaching node {node_id} (offline or unresolvable?)"), + CONNECT_TIMEOUT_SECS, + Exchange::initiate(ctx.matter, ctx.crypto, ctx.fab_idx, node_id), + ) + .await + { + Ok(exchange) => return Ok(exchange), + Err(error) => { + if attempt < CONNECT_ATTEMPTS { + log::warn!( + "Node {node_id}: connect attempt {attempt}/{CONNECT_ATTEMPTS} failed \ + ({error:#}); retrying in {CONNECT_RETRY_DELAY_SECS}s" + ); + embassy_time::Timer::after(embassy_time::Duration::from_secs( + CONNECT_RETRY_DELAY_SECS, + )) + .await; + } + last_error = Some(error); + } + } + } + Err(last_error.expect("at least one attempt ran")) +} + +// --- sync ----------------------------------------------------------------- + +use crate::time::{ClockAssessment, MatterMicros}; +use crate::tz::{build_dst_offset_list, build_time_zone_list}; + +const VERIFY_TOLERANCE_MICROS: i64 = 5_000_000; +const FEATURE_TIME_ZONE: u32 = 1 << 0; + +#[derive(Debug, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct SyncOutcome { + pub node_id: crate::output::Id64, + pub success: bool, + /// True when the device has no Time Synchronization cluster: reported + /// with a warning, not counted as a failure, so a permanently + /// incompatible device cannot fail every timer run. + pub skipped: bool, + pub error: Option<String>, + /// Human assessment of the device clock before the write. + pub clock_before: Option<String>, + pub epoch_shifted: bool, + pub utc_time_written: Option<String>, + pub time_zone_written: Option<String>, + pub dst_entries_written: usize, + pub delta_after_micros: Option<i64>, + pub verified: bool, +} + +pub struct SyncOp { + pub targets: Vec<u64>, + /// When set, write this instant instead of the current time: the + /// operator is deliberately setting an arbitrary wall clock. + pub manual_time: Option<MatterMicros>, +} + +impl Op for SyncOp { + type Out = Vec<SyncOutcome>; + + async fn run<C: Crypto>(self, ctx: &Ctx<'_, C>) -> anyhow::Result<Vec<SyncOutcome>> { + let state_path = state::service_state_path(&ctx.config.storage_path); + let mut outcomes = Vec::with_capacity(self.targets.len()); + for node_id in self.targets { + state::update_node_state(&state_path, node_id, |n| { + n.last_attempted_sync = Some(Timestamp::now()); + })?; + let outcome = match sync_one(ctx, node_id, self.manual_time).await { + Ok(outcome) => outcome, + Err(error) => SyncOutcome { + node_id: node_id.into(), + success: false, + skipped: false, + error: Some(format!("{error:#}")), + clock_before: None, + epoch_shifted: false, + utc_time_written: None, + time_zone_written: None, + dst_entries_written: 0, + delta_after_micros: None, + verified: false, + }, + }; + state::update_node_state(&state_path, node_id, |n| { + if outcome.success { + n.last_successful_sync = Some(Timestamp::now()); + n.last_successful_connection = Some(Timestamp::now()); + n.last_error = None; + } else { + n.last_error = outcome.error.clone(); + } + })?; + if let Some(error) = &outcome.error { + log::error!("Node {node_id}: sync failed: {error}"); + } + outcomes.push(outcome); + } + Ok(outcomes) + } +} + +async fn sync_one<C: Crypto>( + ctx: &Ctx<'_, C>, + node_id: u64, + manual_time: Option<MatterMicros>, +) -> anyhow::Result<SyncOutcome> { + let tz = ctx.config.time_zone(); + let now_ts = Timestamp::now(); + + // Capability discovery first: a device without the cluster is skipped + // with a warning, and the TimeZone feature gates SetTimeZone/SetDSTOffset. + let feature_map = match connect(ctx, node_id) + .await? + .time_synchronization() + .feature_map_read(ROOT_ENDPOINT_ID) + .await + { + Ok(map) => map, + Err(e) if e.code() == rs_matter::error::ErrorCode::ClusterNotFound => { + log::warn!("Node {node_id}: no Time Synchronization cluster; skipping"); + return Ok(SyncOutcome { + node_id: node_id.into(), + success: false, + skipped: true, + error: Some("no Time Synchronization cluster".into()), + clock_before: None, + epoch_shifted: false, + utc_time_written: None, + time_zone_written: None, + dst_entries_written: 0, + delta_after_micros: None, + verified: false, + }); + } + Err(e) => bail!("read featureMap: {e:?}"), + }; + let has_time_zone = feature_map & FEATURE_TIME_ZONE != 0; + + // Before: read the device clock live for the correction report. + let before = connect(ctx, node_id) + .await? + .time_synchronization() + .utc_time_read(ROOT_ENDPOINT_ID) + .await + .map_err(|e| anyhow!("read utcTime: {e:?}"))?; + let device_before = before.into_option().map(MatterMicros); + let assessment = ClockAssessment::compare(device_before, MatterMicros::now()); + log::info!("Node {node_id}: {assessment}"); + + // SetUTCTime, last-moment fresh. The host clock is NTP-disciplined and + // the timestamp microsecond-precise at send time; a device that already + // holds good time may reject a weaker claim (TimeNotAccepted). + let utc_write = manual_time.unwrap_or_else(MatterMicros::now); + connect(ctx, node_id) + .await? + .time_synchronization() + .set_utc_time(ROOT_ENDPOINT_ID, |b| { + b.utc_time(utc_write.0)? + .granularity(GranularityEnum::MicrosecondsGranularity)? + .time_source(Some(TimeSourceEnum::NonMatterSNTP))? + .end() + }) + .await + .map_err(|e| anyhow!("SetUTCTime rejected: {e:?}"))?; + log::info!("Node {node_id}: SetUTCTime {utc_write}"); + + let mut time_zone_written = None; + let mut dst_entries_written = 0; + if has_time_zone { + let zone_entries = build_time_zone_list(&tz, &ctx.config.timezone, now_ts); + let response = connect(ctx, node_id) + .await? + .time_synchronization() + .set_time_zone(ROOT_ENDPOINT_ID, |b| { + let mut list = b.time_zone()?; + for entry in &zone_entries { + list = list + .push()? + .offset(entry.offset_seconds)? + .valid_at(entry.valid_at.0)? + .name(Some(&entry.name))? + .end()?; + } + list.end()?.end() + }) + .await + .map_err(|e| anyhow!("SetTimeZone rejected: {e:?}"))?; + let dst_required = response + .response() + .map(|r| r.dst_offset_required().unwrap_or(true)) + .unwrap_or(true); + response + .complete() + .await + .map_err(|e| anyhow!("SetTimeZone completion: {e:?}"))?; + time_zone_written = Some(ctx.config.timezone.clone()); + log::info!("Node {node_id}: SetTimeZone {}", ctx.config.timezone); + + if dst_required { + let max_entries = connect(ctx, node_id) + .await? + .time_synchronization() + .dst_offset_list_max_size_read(ROOT_ENDPOINT_ID) + .await + .map_err(|e| anyhow!("read dstOffsetListMaxSize: {e:?}"))?; + let dst_entries = build_dst_offset_list(&tz, usize::from(max_entries), now_ts); + connect(ctx, node_id) + .await? + .time_synchronization() + .set_dst_offset(ROOT_ENDPOINT_ID, |b| { + let mut list = b.dst_offset()?; + for entry in &dst_entries { + list = list + .push()? + .offset(entry.offset_seconds)? + .valid_starting(entry.valid_starting.0)? + .valid_until(match entry.valid_until { + Some(until) => rs_matter::tlv::Nullable::some(until.0), + None => rs_matter::tlv::Nullable::none(), + })? + .end()?; + } + list.end()?.end() + }) + .await + .map_err(|e| anyhow!("SetDSTOffset rejected: {e:?}"))?; + dst_entries_written = dst_entries.len(); + log::info!("Node {node_id}: SetDSTOffset with {dst_entries_written} entries"); + } + } + + // Verify by reading the clock back. + let after = connect(ctx, node_id) + .await? + .time_synchronization() + .utc_time_read(ROOT_ENDPOINT_ID) + .await + .map_err(|e| anyhow!("read-back utcTime: {e:?}"))?; + // Verify against what was WRITTEN (not against "now"): the question is + // whether the device accepted our value, which also makes verification + // correct when a manual time was set deliberately far from now. + let after_assessment = + ClockAssessment::compare(after.into_option().map(MatterMicros), utc_write); + let delta_after = after_assessment.effective_delta_micros(); + let verified = delta_after.is_some_and(|d| d.abs() <= VERIFY_TOLERANCE_MICROS); + if !verified { + bail!("read-back verification failed: {after_assessment}"); + } + + ensure_fabric_label(ctx, node_id).await; + + Ok(SyncOutcome { + node_id: node_id.into(), + success: true, + skipped: false, + error: None, + clock_before: Some(assessment.to_string()), + epoch_shifted: assessment.is_epoch_shifted(), + utc_time_written: Some(utc_write.to_string()), + time_zone_written, + dst_entries_written, + delta_after_micros: delta_after, + verified, + }) +} + +/// Pushes the configured fabric label to the device when it differs from the +/// stored one. Best-effort: a cosmetic label must not fail a clock sync. +/// Labels are unique per device, so a conflict with another admin's label is +/// logged with guidance rather than retried. +async fn ensure_fabric_label<C: Crypto>(ctx: &Ctx<'_, C>, node_id: u64) { + if let Err(error) = try_ensure_fabric_label(ctx, node_id).await { + log::warn!("Node {node_id}: could not update fabric label: {error:#}"); + } +} + +async fn try_ensure_fabric_label<C: Crypto>(ctx: &Ctx<'_, C>, node_id: u64) -> anyhow::Result<()> { + let wanted = ctx.config.fabric_label.clone(); + let mut current: Option<String> = None; + connect(ctx, node_id) + .await? + .operational_credentials() + .fabrics_read_with(ROOT_ENDPOINT_ID, |reader| { + // Fabric-filtered read: the device returns only our own entry. + for item in reader? { + current = Some(item?.label()?.to_string()); + } + Ok::<_, MatterError>(()) + }) + .await + .map_err(|e| anyhow!("read fabrics: {e:?}"))? + .map_err(|e| anyhow!("parse fabrics: {e:?}"))?; + + if current.as_deref() == Some(wanted.as_str()) { + return Ok(()); + } + let response = connect(ctx, node_id) + .await? + .operational_credentials() + .update_fabric_label(ROOT_ENDPOINT_ID, |b| b.label(&wanted)?.end()) + .await + .map_err(|e| anyhow!("UpdateFabricLabel: {e:?}"))?; + let status = response.response().map(|r| r.status_code()); + response + .complete() + .await + .map_err(|e| anyhow!("UpdateFabricLabel completion: {e:?}"))?; + log::info!("Node {node_id}: fabric label updated to {wanted:?} (status {status:?})"); + Ok(()) +} + +// --- decommission ---------------------------------------------------------- + +pub struct DecommissionOp { + pub targets: Vec<u64>, +} + +#[derive(Debug, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct DecommissionOutcome { + pub node_id: crate::output::Id64, + pub success: bool, + pub error: Option<String>, +} + +impl Op for DecommissionOp { + type Out = Vec<DecommissionOutcome>; + + async fn run<C: Crypto>(self, ctx: &Ctx<'_, C>) -> anyhow::Result<Vec<DecommissionOutcome>> { + let mut outcomes = Vec::with_capacity(self.targets.len()); + for node_id in self.targets { + let result = decommission_one(ctx, node_id).await; + match result { + Ok(()) => { + let registry_file = state::registry_path(&ctx.config.storage_path); + let mut registry: NodeRegistry = state::load_or_default(®istry_file); + registry.nodes.remove(&node_id.to_string()); + state::store(®istry_file, ®istry)?; + state::remove_node_state( + &state::service_state_path(&ctx.config.storage_path), + node_id, + )?; + outcomes.push(DecommissionOutcome { + node_id: node_id.into(), + success: true, + error: None, + }); + } + Err(error) => { + log::error!("Node {node_id}: decommission failed: {error:#}"); + outcomes.push(DecommissionOutcome { + node_id: node_id.into(), + success: false, + error: Some(format!("{error:#}")), + }); + } + } + } + Ok(outcomes) + } +} + +/// The device drops this controller's fabric via RemoveFabric on our own +/// entry (found through a fabric-filtered read, so no other admin's entry +/// can even be addressed), while staying paired to its primary ecosystem. +async fn decommission_one<C: Crypto>(ctx: &Ctx<'_, C>, node_id: u64) -> anyhow::Result<()> { + let mut our_index: Option<u8> = None; + connect(ctx, node_id) + .await? + .operational_credentials() + .fabrics_read_with(ROOT_ENDPOINT_ID, |reader| { + for item in reader? { + our_index = item?.fabric_index()?; + } + Ok::<_, MatterError>(()) + }) + .await + .map_err(|e| anyhow!("read fabrics: {e:?}"))? + .map_err(|e| anyhow!("parse fabrics: {e:?}"))?; + let our_index = our_index.ok_or_else(|| anyhow!("device has no entry for our fabric"))?; + + log::info!( + "Decommissioning: removing our fabric (device index {our_index}) from node {node_id}" + ); + let response = connect(ctx, node_id) + .await? + .operational_credentials() + .remove_fabric(ROOT_ENDPOINT_ID, |b| b.fabric_index(our_index)?.end()) + .await + .map_err(|e| anyhow!("RemoveFabric: {e:?}"))?; + let status = response.response().map(|r| r.status_code()); + response + .complete() + .await + .map_err(|e| anyhow!("RemoveFabric completion: {e:?}"))?; + log::info!("Node {node_id}: fabric removed (status {status:?})"); + Ok(()) +} + +// --- inspect --------------------------------------------------------------- + +pub struct InspectOp { + pub targets: Vec<u64>, +} + +#[derive(Debug, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct InspectOutcome { + pub node_id: crate::output::Id64, + pub error: Option<String>, + pub vendor_name: Option<String>, + pub product_name: Option<String>, + pub time_sync: Option<TimeSyncCaps>, + pub our_fabric_label: Option<String>, + pub our_fabric_index: Option<u8>, +} + +#[derive(Debug, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TimeSyncCaps { + pub feature_map: u32, + pub time_zone_feature: bool, + pub utc_time: Option<String>, + pub granularity: u8, + pub dst_offset_list_max_size: u8, +} + +impl Op for InspectOp { + type Out = Vec<InspectOutcome>; + + async fn run<C: Crypto>(self, ctx: &Ctx<'_, C>) -> anyhow::Result<Vec<InspectOutcome>> { + let mut outcomes = Vec::with_capacity(self.targets.len()); + for node_id in self.targets { + match inspect_one(ctx, node_id).await { + Ok(outcome) => outcomes.push(outcome), + Err(error) => { + log::error!("Node {node_id}: inspect failed: {error:#}"); + outcomes.push(InspectOutcome { + node_id: node_id.into(), + error: Some(format!("{error:#}")), + vendor_name: None, + product_name: None, + time_sync: None, + our_fabric_label: None, + our_fabric_index: None, + }); + } + } + } + Ok(outcomes) + } +} + +async fn inspect_one<C: Crypto>(ctx: &Ctx<'_, C>, node_id: u64) -> anyhow::Result<InspectOutcome> { + let vendor_name = read_vendor_name(ctx, node_id).await.ok(); + let product_name = read_product_name(ctx, node_id).await.ok(); + + let feature_map = connect(ctx, node_id) + .await? + .time_synchronization() + .feature_map_read(ROOT_ENDPOINT_ID) + .await + .map_err(|e| anyhow!("read featureMap (device may lack Time Synchronization): {e:?}"))?; + let utc_time = connect(ctx, node_id) + .await? + .time_synchronization() + .utc_time_read(ROOT_ENDPOINT_ID) + .await + .map_err(|e| anyhow!("read utcTime: {e:?}"))? + .into_option() + .map(|v| MatterMicros(v).to_string()); + let granularity = connect(ctx, node_id) + .await? + .time_synchronization() + .granularity_read(ROOT_ENDPOINT_ID) + .await + .map_err(|e| anyhow!("read granularity: {e:?}"))? as u8; + let has_tz = feature_map & FEATURE_TIME_ZONE != 0; + let dst_max = if has_tz { + connect(ctx, node_id) + .await? + .time_synchronization() + .dst_offset_list_max_size_read(ROOT_ENDPOINT_ID) + .await + .unwrap_or(1) + } else { + 0 + }; + + let mut label = None; + let mut index = None; + connect(ctx, node_id) + .await? + .operational_credentials() + .fabrics_read_with(ROOT_ENDPOINT_ID, |reader| { + for item in reader? { + let item = item?; + label = Some(item.label()?.to_string()); + index = item.fabric_index()?; + } + Ok::<_, MatterError>(()) + }) + .await + .map_err(|e| anyhow!("read fabrics: {e:?}"))? + .map_err(|e| anyhow!("parse fabrics: {e:?}"))?; + + Ok(InspectOutcome { + node_id: node_id.into(), + error: None, + vendor_name, + product_name, + time_sync: Some(TimeSyncCaps { + feature_map, + time_zone_feature: has_tz, + utc_time, + granularity, + dst_offset_list_max_size: dst_max, + }), + our_fabric_label: label, + our_fabric_index: index, + }) +} diff --git a/src/host.rs b/src/host.rs new file mode 100644 index 0000000..feb5a34 --- /dev/null +++ b/src/host.rs @@ -0,0 +1,88 @@ +//! Host clock trust: the device must never be set from a clock we do not +//! trust. +//! +//! The primary check measures the actual clock error with a single SNTP +//! query (RFC 4330) and accepts the host when the offset is under a second. +//! That is a direct measurement, platform-independent, and stronger than +//! asking the OS whether it believes it is synchronized. When no NTP server +//! is reachable, systemd-timesyncd's verdict (`timedatectl`, Linux) is the +//! fallback; hosts with neither fail safe. + +use std::net::UdpSocket; +use std::process::Command; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +const NTP_SERVERS: [&str; 2] = ["time.apple.com:123", "pool.ntp.org:123"]; +/// Accept the host clock when it is within this many seconds of NTP time. +/// Well inside the 5s sync read-back tolerance, far above network jitter. +const MAX_OFFSET_SECONDS: f64 = 1.0; +/// Seconds between the NTP era (1900) and the Unix epoch (1970). +const NTP_UNIX_OFFSET: f64 = 2_208_988_800.0; +/// The NTP fraction field is 32-bit fixed-point in units of 1/2^32 seconds; +/// dividing by 2^32 converts it to seconds. +const NTP_FRACTION_SCALE: f64 = (1u64 << 32) as f64; + +pub fn clock_is_ntp_synchronized() -> bool { + for server in NTP_SERVERS { + if let Some(offset) = sntp_offset(server) { + let ok = offset.abs() <= MAX_OFFSET_SECONDS; + if ok { + log::debug!("Host clock is {offset:+.3}s from {server}; trusted"); + } else { + log::warn!("Host clock is {offset:+.3}s from {server}; not trusted"); + } + return ok; + } + } + log::debug!("No NTP server reachable; falling back to timedatectl"); + timedatectl_says_synchronized() +} + +/// One SNTP client exchange: returns the approximate offset of the local +/// clock relative to the server (positive = local clock ahead). Uses the +/// request/response midpoint, so the error is bounded by half the round +/// trip, which is milliseconds against a threshold of a second. +fn sntp_offset(server: &str) -> Option<f64> { + let socket = UdpSocket::bind(("0.0.0.0", 0)).ok()?; + socket.set_read_timeout(Some(Duration::from_secs(2))).ok()?; + socket.connect(server).ok()?; + + let mut request = [0u8; 48]; + request[0] = 0b00_100_011; // LI 0, version 4, mode 3 (client) + let sent_at = unix_now(); + socket.send(&request).ok()?; + + let mut response = [0u8; 48]; + let len = socket.recv(&mut response).ok()?; + let received_at = unix_now(); + if len < 48 || response[0] & 0x07 != 4 { + // Not a server-mode reply. + return None; + } + + // Transmit timestamp: seconds since 1900 plus a 32-bit binary fraction. + let seconds = u32::from_be_bytes(response[40..44].try_into().ok()?) as f64; + let fraction = + u32::from_be_bytes(response[44..48].try_into().ok()?) as f64 / NTP_FRACTION_SCALE; + let server_time = seconds + fraction - NTP_UNIX_OFFSET; + if server_time <= 0.0 { + return None; + } + Some((sent_at + received_at) / 2.0 - server_time) +} + +fn unix_now() -> f64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs_f64()) + .unwrap_or(0.0) +} + +/// systemd-timesyncd's opinion; false on hosts without timedatectl. +fn timedatectl_says_synchronized() -> bool { + Command::new("timedatectl") + .args(["show", "-p", "NTPSynchronized", "--value"]) + .output() + .map(|output| output.status.success() && output.stdout.trim_ascii() == b"yes") + .unwrap_or(false) +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..e36ad21 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,374 @@ +//! mattertimesync: a one-shot CLI Matter controller that sets the clocks of +//! Matter devices via the standard Time Synchronization cluster. +//! +//! # CLI invariants +//! +//! Every command upholds these; new commands and options must too. Numbers +//! 1, 2 and 5 are enforced structurally by the [`output::Output`] type: +//! handlers return a value of a closed enum instead of printing, the exit +//! code is derived from that value, and both renderings consume it. +//! +//! 1. **Channels.** stdout carries command output only; all logs go to +//! stderr. With `--json`, stdout is exactly one pretty-printed JSON +//! object with a descriptive top-level key ({"nodes": ...}, +//! {"commissioned": ...}), on failure included ({"error": ...}). +//! 2. **Exit codes.** 0 = success, including empty-but-valid results; +//! 1 = runtime failure, including any per-node failure; 2 = usage or +//! configuration error. Devices skipped as incompatible do not fail a +//! run. +//! 3. **--node semantics.** The local read-only command (status) takes a +//! [`Filter`]: an absent node yields an empty report, never an error. +//! Device-connecting commands (inspect, sync, decommission) take a +//! [`Target`]: the node must be commissioned, otherwise error. +//! commission takes neither because the node ID does not exist until it +//! assigns one. +//! 4. **Default targeting.** Device-connecting commands operate on every +//! commissioned device when --node is omitted. On an empty registry, +//! read-only inspect reports emptiness successfully; fabric-writing +//! sync and decommission error, because having nothing to write to is +//! an operator problem worth surfacing. +//! 5. **Per-node isolation.** Multi-node runs never abort the batch on one +//! node's failure; every node gets its own outcome entry. +//! 6. **Universal options.** -c/--config and -j/--json exist everywhere; +//! -n/--node everywhere except commission. + +mod config; +mod controller; +mod host; +mod output; +mod pairing; +mod state; +mod time; +mod tz; + +use std::path::PathBuf; +use std::process::ExitCode; + +use clap::{Parser, Subcommand}; +use jiff::Timestamp; + +use crate::config::{Config, DEFAULT_CONFIG_PATH}; +use crate::output::{ + ConfigReport, DstTransition, IdentityReport, NodeListing, Output, StatusReport, SyncReport, +}; +use crate::state::{NodeRegistry, ServiceState}; +use crate::time::MatterMicros; + +#[derive(Parser)] +#[command( + name = "mattertimesync", + version, + about, + disable_help_subcommand = true +)] +struct Cli { + /// Configuration file + #[arg(short = 'c', long, global = true, default_value = DEFAULT_CONFIG_PATH)] + config: PathBuf, + #[command(subcommand)] + command: Option<Command>, +} + +#[derive(Subcommand)] +enum Command { + /// Show controller and per-device state [local, read-only] + Status { + /// Machine-readable output (64-bit values as decimal strings) + #[arg(short = 'j', long)] + json: bool, + /// Show only this device's state + #[arg(short = 'n', long)] + node: Option<u64>, + }, + /// Connect to each commissioned device (or one with --node) and dump + /// identity, time-sync capabilities, and our fabric entry + /// [fabric interrogation, read-only] + Inspect { + /// Inspect only this device + #[arg(short = 'n', long)] + node: Option<u64>, + /// Machine-readable output (64-bit values as decimal strings) + #[arg(short = 'j', long)] + json: bool, + }, + /// Join a device as an additional Matter admin; the pairing code is + /// used once, never logged or stored [fabric-writing] + Commission { + /// Pairing code from the primary ecosystem's pairing mode + pairing_code: String, + /// Machine-readable output (64-bit values as decimal strings) + #[arg(short = 'j', long)] + json: bool, + }, + /// Set each device's clock: UTC time, time zone, DST offsets + /// [fabric-writing] + Sync { + /// Target one device instead of all commissioned devices + #[arg(short = 'n', long)] + node: Option<u64>, + /// Set this wall-clock time (e.g. 16:35 or 4:35pm, today in the + /// configured time zone) instead of the current time + #[arg(short = 't', long, value_name = "TIME")] + time: Option<String>, + /// Machine-readable output (64-bit values as decimal strings) + #[arg(short = 'j', long)] + json: bool, + }, + /// Drop this controller's fabric from each device, or one with --node; + /// primary ecosystems are untouched [fabric-writing] + Decommission { + /// Target one device instead of all commissioned devices + #[arg(short = 'n', long)] + node: Option<u64>, + /// Machine-readable output (64-bit values as decimal strings) + #[arg(short = 'j', long)] + json: bool, + }, +} + +fn main() -> ExitCode { + let cli = Cli::parse(); + + // No command: show the full help, not a terse error. + let Some(command) = cli.command else { + use clap::CommandFactory; + let _ = Cli::command().print_help(); + return ExitCode::from(2); + }; + + let config = match Config::load(&cli.config) { + Ok(config) => config, + Err(error) => { + eprintln!("Configuration error: {error}"); + return ExitCode::from(2); + } + }; + + env_logger::Builder::new() + .filter_level(config.log_level.into()) + .format_timestamp_millis() + .init(); + + for warning in config.path_warnings() { + log::warn!("{warning}"); + } + + let (json, result) = match command { + Command::Status { json, node } => (json, run_status(&config, Filter(node))), + Command::Inspect { json, node } => (json, run_inspect(&config, Target(node))), + Command::Sync { json, node, time } => { + (json, run_sync(&config, Target(node), time.as_deref())) + } + Command::Commission { json, pairing_code } => { + (json, run_commission(&config, &pairing_code)) + } + Command::Decommission { json, node } => (json, run_decommission(&config, Target(node))), + }; + + let output = result.unwrap_or_else(|error| { + log::error!("{error:#}"); + Output::Error { + error: format!("{error:#}"), + } + }); + if json { + output.print_json(); + } else { + output.render_human(); + } + output.exit_code() +} + +/// `--node` on a local, read-only command: filters a report. An absent node +/// yields an empty report, never an error. +struct Filter(Option<u64>); + +impl Filter { + fn retain<V>(&self, map: &mut std::collections::BTreeMap<String, V>) { + if let Some(node) = self.0 { + let key = node.to_string(); + map.retain(|k, _| *k == key); + } + } +} + +/// `--node` on a device-connecting command: selects targets. A requested +/// node must be commissioned; without one, every commissioned device. +struct Target(Option<u64>); + +impl Target { + /// Targets for fabric-writing commands: an empty registry is an error + /// (having nothing to write to is an operator problem worth surfacing). + fn resolve(&self, config: &Config) -> anyhow::Result<Vec<u64>> { + let targets = self.resolve_or_empty(config)?; + if targets.is_empty() { + anyhow::bail!("no devices are commissioned yet; run \"commission\" to add one"); + } + Ok(targets) + } + + /// Targets for read-only interrogation: an empty registry is a valid, + /// empty answer. An explicitly requested unknown node is still an error. + fn resolve_or_empty(&self, config: &Config) -> anyhow::Result<Vec<u64>> { + let registry: NodeRegistry = + state::load_or_default(&state::registry_path(&config.storage_path)); + let known: Vec<u64> = registry.node_ids().collect(); + match self.0 { + None => Ok(known), + Some(node) if known.contains(&node) => Ok(vec![node]), + Some(node) => anyhow::bail!( + "node {node} is not commissioned on this controller (known nodes: {})", + if known.is_empty() { + "none".into() + } else { + known + .iter() + .map(u64::to_string) + .collect::<Vec<_>>() + .join(", ") + } + ), + } + } +} + +fn run_status(config: &Config, filter: Filter) -> anyhow::Result<Output> { + let tz = config.time_zone(); + let now = Timestamp::now(); + let state: ServiceState = + state::load_or_default(&state::service_state_path(&config.storage_path)); + let mut registry: NodeRegistry = + state::load_or_default(&state::registry_path(&config.storage_path)); + filter.retain(&mut registry.nodes); + let nodes: Vec<NodeListing> = registry + .nodes + .into_iter() + .filter_map(|(key, info)| { + // Registry keys are decimal node IDs written by us; anything else + // is file corruption and is skipped, matching the lenient loader. + let node_id: u64 = key.parse().ok()?; + let node = state.nodes.get(&key).cloned().unwrap_or_default(); + Some(NodeListing { + node_id: node_id.into(), + vendor_name: info.vendor_name, + product_name: info.product_name, + last_successful_connection: node.last_successful_connection, + last_successful_sync: node.last_successful_sync, + last_attempted_sync: node.last_attempted_sync, + last_error: node.last_error, + }) + }) + .collect(); + + let identity = match controller::identity_status(&config.storage_path) { + controller::IdentityStatus::NotCreated => IdentityReport::NotCreated, + controller::IdentityStatus::Created { + fabric_id, + controller_node_id, + } => IdentityReport::Created { + fabric_id: fabric_id.to_string(), + controller_node_id: controller_node_id.to_string(), + }, + controller::IdentityStatus::Unreadable => IdentityReport::Unreadable, + controller::IdentityStatus::Inconsistent(reason) => IdentityReport::Inconsistent { + reason: reason.to_string(), + }, + }; + let offset_seconds = tz.to_offset(now).seconds(); + Ok(Output::Status(Box::new(StatusReport { + config: ConfigReport { + source: config.source.clone(), + storage_path: config.storage_path.clone(), + timezone: config.timezone.clone(), + log_level: config.log_level.to_string(), + fabric_label: config.fabric_label.clone(), + }, + storage_initialized: config.storage_path.is_dir(), + controller_identity: identity, + host_ntp_synchronized: host::clock_is_ntp_synchronized(), + current_utc_offset_seconds: offset_seconds, + current_utc_offset: tz::format_utc_offset(offset_seconds), + next_dst_transition: tz::next_offset_transition(&tz, now).map(|(at, before, after)| { + DstTransition { + at: at.to_string(), + offset_before_seconds: before, + offset_after_seconds: after, + } + }), + matter_time_now_microseconds: MatterMicros::now().0.to_string(), + nodes, + }))) +} + +fn run_inspect(config: &Config, target: Target) -> anyhow::Result<Output> { + let targets = target.resolve_or_empty(config)?; + let nodes = if targets.is_empty() { + Vec::new() + } else { + controller::run(config, controller::InspectOp { targets })? + }; + Ok(Output::Inspection { nodes }) +} + +fn run_sync(config: &Config, target: Target, time: Option<&str>) -> anyhow::Result<Output> { + // --time makes the operator the time source: compute today's date in + // the configured zone at the requested wall-clock time. + let manual_time = match time { + None => None, + Some(raw) => { + let wall = crate::time::parse_wall_clock(raw).map_err(|e| anyhow::anyhow!(e))?; + let tz = config.time_zone(); + let today = Timestamp::now().to_zoned(tz.clone()).date(); + let instant = today + .at(wall.hour(), wall.minute(), wall.second(), 0) + .to_zoned(tz) + .map_err(|e| anyhow::anyhow!("cannot place {raw:?} in {}: {e}", config.timezone))? + .timestamp(); + Some(MatterMicros::from_timestamp(instant)) + } + }; + + // Fail safe: never push time from a clock that is not NTP-disciplined. + // Not enforced with --time: the operator deliberately chose the value. + let ntp = host::clock_is_ntp_synchronized(); + if !ntp && manual_time.is_none() { + log::warn!("Host clock is not NTP-synchronized; refusing to set device time."); + return Ok(Output::Sync(SyncReport { + host_ntp_synchronized: false, + manual_time: None, + nodes: Vec::new(), + })); + } + let targets = target.resolve(config)?; + let nodes = controller::run( + config, + controller::SyncOp { + targets, + manual_time, + }, + )?; + Ok(Output::Sync(SyncReport { + host_ntp_synchronized: ntp, + manual_time: manual_time.map(|m| m.to_string()), + nodes, + })) +} + +fn run_commission(config: &Config, code: &str) -> anyhow::Result<Output> { + let onboarding = pairing::parse_manual_code(code)?; + let commissioned = controller::run(config, controller::CommissionOp { onboarding })?; + Ok(Output::Commissioned { + nodes: vec![commissioned], + }) +} + +fn run_decommission(config: &Config, target: Target) -> anyhow::Result<Output> { + let targets = target.resolve(config)?; + let nodes = controller::run(config, controller::DecommissionOp { targets })?; + let registry: NodeRegistry = + state::load_or_default(&state::registry_path(&config.storage_path)); + Ok(Output::Decommission { + nodes, + remaining_nodes: registry.nodes.into_keys().collect(), + }) +} diff --git a/src/output.rs b/src/output.rs new file mode 100644 index 0000000..4de5745 --- /dev/null +++ b/src/output.rs @@ -0,0 +1,444 @@ +//! The typed output layer. +//! +//! Every command handler returns an [`Output`]; printing happens exactly +//! once, in `main`, from that value. The CLI invariants are enforced here by +//! construction rather than by convention: +//! +//! - the set of top-level JSON shapes is closed (this enum), so a command +//! cannot invent an envelope or emit a naked array; +//! - JSON formatting is uniform because [`print_json`] is the only printer; +//! - the exit code is derived from the output value by [`Output::exit_code`], +//! so it cannot disagree with what was reported; +//! - human and JSON renderings are fed by the same data, so they cannot +//! drift apart in content. + +use std::path::PathBuf; +use std::process::ExitCode; + +use jiff::Timestamp; +use serde::Serialize; + +use crate::controller::{CommissionOutcome, DecommissionOutcome, InspectOutcome, SyncOutcome}; +use crate::time::CompactDuration; +use crate::tz::format_utc_offset; + +/// A 64-bit Matter identifier (node ID, fabric ID). Serializes as a decimal +/// string, never as a JSON number, which silently loses precision beyond +/// 2^53 in many consumers. The rule lives in the type, so an ID field +/// cannot be emitted wrongly. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct Id64(pub u64); + +impl Serialize for Id64 { + fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> { + serializer.serialize_str(&self.0.to_string()) + } +} + +impl std::fmt::Display for Id64 { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} + +impl From<u64> for Id64 { + fn from(value: u64) -> Self { + Self(value) + } +} + +/// Everything a command can say. Untagged: each variant serializes as its +/// own object with descriptive top-level keys. +#[derive(Debug, Serialize)] +#[serde(untagged)] +pub enum Output { + Status(Box<StatusReport>), + Inspection { + nodes: Vec<InspectOutcome>, + }, + Sync(SyncReport), + Commissioned { + nodes: Vec<CommissionOutcome>, + }, + #[serde(rename_all = "camelCase")] + Decommission { + nodes: Vec<DecommissionOutcome>, + remaining_nodes: Vec<String>, + }, + Error { + error: String, + }, +} + +/// Everything that was parsed from the configuration file, echoed back so +/// an operator can see exactly what the tool is running with. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ConfigReport { + pub source: PathBuf, + pub storage_path: PathBuf, + pub timezone: String, + pub log_level: String, + pub fabric_label: String, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StatusReport { + pub config: ConfigReport, + pub storage_initialized: bool, + pub controller_identity: IdentityReport, + pub host_ntp_synchronized: bool, + pub current_utc_offset_seconds: i32, + pub current_utc_offset: String, + pub next_dst_transition: Option<DstTransition>, + pub matter_time_now_microseconds: String, + /// One entry per commissioned device: registry identity merged with + /// the latest sync state. + pub nodes: Vec<NodeListing>, +} + +#[derive(Debug, Serialize)] +#[serde( + tag = "state", + rename_all = "camelCase", + rename_all_fields = "camelCase" +)] +pub enum IdentityReport { + NotCreated, + Created { + fabric_id: String, + controller_node_id: String, + }, + Unreadable, + Inconsistent { + reason: String, + }, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct DstTransition { + pub at: String, + pub offset_before_seconds: i32, + pub offset_after_seconds: i32, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct NodeListing { + pub node_id: Id64, + pub vendor_name: Option<String>, + pub product_name: Option<String>, + pub last_successful_connection: Option<Timestamp>, + pub last_successful_sync: Option<Timestamp>, + pub last_attempted_sync: Option<Timestamp>, + pub last_error: Option<String>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct SyncReport { + pub host_ntp_synchronized: bool, + /// Set when the operator supplied --time: the written instant. + pub manual_time: Option<String>, + pub nodes: Vec<SyncOutcome>, +} + +impl Output { + /// The exit code is a function of the reported outcome; a handler cannot + /// return one that disagrees with its output. 0 = success (including + /// empty-but-valid results), 1 = any failure. Skipped devices are not + /// failures. + pub fn exit_code(&self) -> ExitCode { + let ok = match self { + Output::Status(_) | Output::Commissioned { .. } => true, + Output::Inspection { nodes } => nodes.iter().all(|n| n.error.is_none()), + Output::Sync(report) => { + // With --time the operator is the time source; the NTP + // verdict is informational only. + (report.host_ntp_synchronized || report.manual_time.is_some()) + && report.nodes.iter().all(|n| n.success || n.skipped) + } + Output::Decommission { nodes, .. } => nodes.iter().all(|n| n.success), + Output::Error { .. } => false, + }; + if ok { + ExitCode::SUCCESS + } else { + ExitCode::FAILURE + } + } + + /// The one JSON printer: pretty, one object, always parseable. + pub fn print_json(&self) { + match serde_json::to_string_pretty(self) { + Ok(body) => println!("{body}"), + Err(error) => println!("{{\"error\": \"serialize output: {error}\"}}"), + } + } + + pub fn render_human(&self) { + match self { + Output::Status(report) => render_status(report), + Output::Inspection { nodes } => render_inspection(nodes), + Output::Sync(report) => render_sync(report), + Output::Commissioned { nodes } => nodes.iter().for_each(render_commissioned), + Output::Decommission { + nodes, + remaining_nodes, + } => render_decommission(nodes, remaining_nodes), + // Failures are already on stderr via the log; stdout stays quiet. + Output::Error { .. } => {} + } + } +} + +fn display_instant(at: Option<Timestamp>) -> String { + at.map_or_else(|| "never".into(), |at| at.to_string()) +} + +fn render_status(report: &StatusReport) { + println!( + "Configuration: loaded from {}", + report.config.source.display() + ); + println!( + " storagePath: {}", + report.config.storage_path.display() + ); + println!(" timezone: {}", report.config.timezone); + println!(" logLevel: {}", report.config.log_level); + println!(" fabricLabel: {}", report.config.fabric_label); + println!( + "Controller storage: {}", + if report.storage_initialized { + format!("present at {}", report.config.storage_path.display()) + } else { + "NOT INITIALIZED".into() + } + ); + match &report.controller_identity { + IdentityReport::NotCreated => { + println!("Controller identity: not created yet (\"commission\" will create it)") + } + IdentityReport::Created { + fabric_id, + controller_node_id, + } => println!( + "Controller identity: created (fabric id {fabric_id}, controller node id {controller_node_id})" + ), + IdentityReport::Unreadable => { + println!("Controller identity: present but unreadable (run as the service user?)") + } + IdentityReport::Inconsistent { reason } => { + println!("Controller identity: INCONSISTENT: {reason}") + } + } + println!( + "Commissioned nodes: {}", + if report.nodes.is_empty() { + "none recorded".into() + } else { + report + .nodes + .iter() + .map(|n| n.node_id.to_string()) + .collect::<Vec<_>>() + .join(", ") + } + ); + println!( + "Host NTP synced: {}", + if report.host_ntp_synchronized { + "yes" + } else { + "no (or not determinable on this host)" + } + ); + println!("Current UTC offset: {}", report.current_utc_offset); + match &report.next_dst_transition { + Some(transition) => println!( + "Next DST transition: {} ({} -> {})", + transition.at, + format_utc_offset(transition.offset_before_seconds), + format_utc_offset(transition.offset_after_seconds) + ), + None => println!("Next DST transition: none (fixed-offset zone)"), + } + println!( + "Matter time now: {} us since 2000-01-01T00:00:00Z", + report.matter_time_now_microseconds + ); + for node in &report.nodes { + let name = [node.vendor_name.as_deref(), node.product_name.as_deref()] + .into_iter() + .flatten() + .collect::<Vec<_>>() + .join(" "); + println!( + "Node {}: {}", + node.node_id, + if name.is_empty() { + "(no cached device info)" + } else { + &name + } + ); + println!( + " Last connection: {}", + display_instant(node.last_successful_connection) + ); + println!( + " Last successful sync: {}", + display_instant(node.last_successful_sync) + ); + println!( + " Last attempted sync: {}", + display_instant(node.last_attempted_sync) + ); + println!( + " Most recent error: {}", + node.last_error.as_deref().unwrap_or("none") + ); + } +} + +fn render_inspection(nodes: &[InspectOutcome]) { + if nodes.is_empty() { + println!("No devices are commissioned yet; run \"commission\" to add one."); + return; + } + for outcome in nodes { + println!("Node {}", outcome.node_id); + if let Some(error) = &outcome.error { + println!(" Error: {error}"); + println!(); + continue; + } + println!( + " Vendor: {}", + outcome.vendor_name.as_deref().unwrap_or("(unknown)") + ); + println!( + " Product: {}", + outcome.product_name.as_deref().unwrap_or("(unknown)") + ); + if let Some(time_sync) = &outcome.time_sync { + println!(" Time Synchronization cluster (endpoint 0):"); + println!( + " features: {:#x}{}", + time_sync.feature_map, + if time_sync.time_zone_feature { + " (timeZone)" + } else { + "" + } + ); + println!( + " utcTime: {}", + time_sync.utc_time.as_deref().unwrap_or("unset") + ); + println!(" granularity: {}", time_sync.granularity); + println!( + " dstOffsetListMaxSize: {}", + time_sync.dst_offset_list_max_size + ); + } + println!( + " Our fabric entry: label {:?}, device fabric index {}", + outcome.our_fabric_label.as_deref().unwrap_or("(none)"), + outcome + .our_fabric_index + .map(|i| i.to_string()) + .unwrap_or_else(|| "?".into()) + ); + println!(); + } +} + +fn render_sync(report: &SyncReport) { + if !report.host_ntp_synchronized && report.manual_time.is_none() { + // The refusal is already on stderr as a warning. + return; + } + if let Some(manual) = &report.manual_time { + println!("Manual time set: {manual} (host NTP state not enforced)"); + } + for outcome in &report.nodes { + println!("Node {}:", outcome.node_id); + if let Some(before) = &outcome.clock_before { + println!(" Assessment: {before}"); + } + if let Some(written) = &outcome.utc_time_written { + println!(" Time written: {written}"); + } + if let Some(zone) = &outcome.time_zone_written { + println!(" Time zone: {zone}"); + } + if outcome.dst_entries_written > 0 { + println!(" DST offsets: {} entries", outcome.dst_entries_written); + } + if let Some(delta) = outcome.delta_after_micros { + println!( + " Verification: device clock within {} of host after sync", + CompactDuration(delta.unsigned_abs()) + ); + } + match (&outcome.error, outcome.skipped) { + (None, _) => println!(" Result: OK"), + (Some(reason), true) => println!(" Result: SKIPPED ({reason})"), + (Some(error), false) => println!(" Result: FAILED ({error})"), + } + } +} + +fn render_commissioned(outcome: &CommissionOutcome) { + println!("Commissioning summary"); + println!(" Node ID: {}", outcome.node_id); + println!(" Fabric ID: {}", outcome.fabric_id); + println!( + " Device: {}", + [ + outcome.vendor_name.as_deref(), + outcome.product_name.as_deref() + ] + .into_iter() + .flatten() + .collect::<Vec<_>>() + .join(" ") + ); + println!(); + println!("The device remains paired with its primary ecosystem; this controller"); + println!("was added as an additional Matter administrator. Run \"sync\" to"); + println!("synchronize its clock now."); +} + +fn render_decommission(nodes: &[DecommissionOutcome], remaining: &[String]) { + for outcome in nodes { + if outcome.success { + println!( + "Device {} removed this controller's fabric; its primary ecosystem is untouched.", + outcome.node_id + ); + } else { + println!( + "Device {} could NOT be decommissioned: {}", + outcome.node_id, + outcome.error.as_deref().unwrap_or("unknown error") + ); + } + } + if remaining.is_empty() { + println!( + "No devices remain commissioned. The local controller identity remains in the \ + storage directory; deleting it is now safe." + ); + } else { + println!( + "{} device(s) remain commissioned: {}", + remaining.len(), + remaining.join(", ") + ); + } +} diff --git a/src/pairing.rs b/src/pairing.rs new file mode 100644 index 0000000..715d7a8 --- /dev/null +++ b/src/pairing.rs @@ -0,0 +1,129 @@ +//! Manual pairing code decoding (Matter Core spec 5.1.4.1). +//! +//! An 11-digit (no VID/PID) or 21-digit (with VID/PID) decimal code packs a +//! 4-bit short discriminator and the 27-bit setup passcode, guarded by a +//! trailing Verhoeff check digit. QR payloads ("MT:...") are not accepted; +//! the primary ecosystem always shows the numeric code. + +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub enum PairingCodeError { + #[error( + "QR payloads are not supported; enter the numeric pairing code shown by the primary ecosystem" + )] + QrPayload, + #[error("pairing code must be 11 or 21 digits (got {0})")] + BadLength(usize), + #[error("pairing code check digit is wrong; re-check the digits")] + BadCheckDigit, + #[error("pairing code decodes to an invalid setup passcode")] + BadPasscode, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Onboarding { + pub passcode: u32, + /// Upper 4 bits of the 12-bit discriminator, as carried in manual codes. + pub short_discriminator: u8, +} + +/// Setup passcodes the spec forbids (trivial/sequential values). +const INVALID_PASSCODES: [u32; 12] = [ + 0, 11111111, 22222222, 33333333, 44444444, 55555555, 66666666, 77777777, 88888888, 99999999, + 12345678, 87654321, +]; + +pub fn parse_manual_code(raw: &str) -> Result<Onboarding, PairingCodeError> { + let trimmed = raw.trim(); + if trimmed.starts_with("MT:") { + return Err(PairingCodeError::QrPayload); + } + let digits: Vec<u8> = trimmed + .chars() + .filter(|c| !c.is_whitespace() && *c != '-') + .map(|c| { + c.to_digit(10) + .map(|d| d as u8) + .ok_or(PairingCodeError::BadLength(0)) + }) + .collect::<Result<_, _>>() + .map_err(|_| PairingCodeError::BadLength(trimmed.len()))?; + + if digits.len() != 11 && digits.len() != 21 { + return Err(PairingCodeError::BadLength(digits.len())); + } + + let payload: String = digits[..digits.len() - 1] + .iter() + .map(|d| (b'0' + d) as char) + .collect(); + if verhoeff::calculate(payload.as_str()) != digits[digits.len() - 1] { + return Err(PairingCodeError::BadCheckDigit); + } + + let d1 = u32::from(digits[0]); + let chunk2 = digits[1..6] + .iter() + .fold(0u32, |acc, &d| acc * 10 + u32::from(d)); + let chunk3 = digits[6..10] + .iter() + .fold(0u32, |acc, &d| acc * 10 + u32::from(d)); + + // digit 1 = (VID_PID_PRESENT << 2) | (discriminator >> 10) + // digits 2-6 = ((discriminator & 0x300) << 6) | (passcode & 0x3FFF) + // digits 7-10 = passcode >> 14 + let short_discriminator = (((d1 & 0x3) << 2) | ((chunk2 >> 14) & 0x3)) as u8; + let passcode = (chunk3 << 14) | (chunk2 & 0x3FFF); + + if passcode >= 1 << 27 || INVALID_PASSCODES.contains(&passcode) { + return Err(PairingCodeError::BadPasscode); + } + Ok(Onboarding { + passcode, + short_discriminator, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn decodes_the_canonical_test_code() { + // The Matter SDK test onboarding payload: passcode 20202021, + // discriminator 3840 (0xF00) => short discriminator 15, manual code + // 34970112332. + let onboarding = parse_manual_code("34970112332").unwrap(); + assert_eq!( + onboarding, + Onboarding { + passcode: 20202021, + short_discriminator: 15, + } + ); + } + + #[test] + fn accepts_hyphenated_input() { + assert!(parse_manual_code("3497-011-2332").is_ok()); + } + + #[test] + fn rejects_qr_payloads_with_guidance() { + assert_eq!( + parse_manual_code("MT:Y.K90SO527JA0648G00"), + Err(PairingCodeError::QrPayload) + ); + } + + #[test] + fn rejects_wrong_lengths_and_check_digits() { + assert!(matches!( + parse_manual_code("1234"), + Err(PairingCodeError::BadLength(_)) + )); + assert_eq!( + parse_manual_code("34970112331"), + Err(PairingCodeError::BadCheckDigit) + ); + } +} diff --git a/src/state.rs b/src/state.rs new file mode 100644 index 0000000..887ac21 --- /dev/null +++ b/src/state.rs @@ -0,0 +1,194 @@ +//! Local service state kept next to the Matter fabric storage. +//! +//! Two small JSON files, both written atomically (temp file + rename) so a +//! power loss mid-write never corrupts them: +//! +//! - `service-state.json`: per-node sync results, file-compatible with the +//! TypeScript implementation. +//! - `nodes.json`: the node registry (id plus cached vendor/product names), +//! written at commissioning time so `nodes` and `status` never need to +//! start the Matter stack. + +use std::collections::BTreeMap; +use std::io; +use std::path::{Path, PathBuf}; + +use jiff::Timestamp; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", default)] +pub struct NodeState { + pub last_successful_connection: Option<Timestamp>, + pub last_successful_sync: Option<Timestamp>, + pub last_attempted_sync: Option<Timestamp>, + pub last_error: Option<String>, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +pub struct ServiceState { + /// Keyed by node ID as a decimal string, like the TypeScript state file. + pub nodes: BTreeMap<String, NodeState>, +} + +/// Cached identity of a commissioned node, captured at commissioning. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", default)] +pub struct NodeInfo { + pub vendor_name: Option<String>, + pub product_name: Option<String>, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +pub struct NodeRegistry { + pub nodes: BTreeMap<String, NodeInfo>, +} + +impl NodeRegistry { + pub fn node_ids(&self) -> impl Iterator<Item = u64> + '_ { + // Registry keys are written by us; a non-decimal key would be file + // corruption, which the lenient loader already reduces to "empty". + self.nodes.keys().filter_map(|key| key.parse().ok()) + } +} + +pub fn service_state_path(storage: &Path) -> PathBuf { + storage.join("service-state.json") +} + +pub fn registry_path(storage: &Path) -> PathBuf { + storage.join("nodes.json") +} + +/// Reads a state file leniently: missing or corrupt files yield the default. +/// These files hold status metadata only; refusing to run over them would +/// turn a scratched cache into an outage. +pub fn load_or_default<T: DeserializeOwned + Default>(path: &Path) -> T { + match std::fs::read_to_string(path) { + Ok(raw) => serde_json::from_str(&raw).unwrap_or_default(), + Err(_) => T::default(), + } +} + +/// Atomic JSON write: temp file in the same directory, then rename. +pub fn store<T: Serialize>(path: &Path, value: &T) -> io::Result<()> { + let parent = path.parent().unwrap_or_else(|| Path::new(".")); + std::fs::create_dir_all(parent)?; + let tmp = path.with_extension("json.tmp"); + let body = serde_json::to_string_pretty(value).expect("state serialization cannot fail"); + write_private(&tmp, body.as_bytes())?; + std::fs::rename(&tmp, path) +} + +#[cfg(unix)] +pub(crate) fn write_private(path: &Path, bytes: &[u8]) -> io::Result<()> { + use std::io::Write; + use std::os::unix::fs::OpenOptionsExt; + let mut file = std::fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .mode(0o600) + .open(path)?; + file.write_all(bytes)?; + file.write_all(b"\n") +} + +#[cfg(not(unix))] +pub(crate) fn write_private(path: &Path, bytes: &[u8]) -> io::Result<()> { + std::fs::write(path, [bytes, b"\n"].concat()) +} + +/// Merges a patch into one node's entry and persists the result. +pub fn update_node_state( + path: &Path, + node_id: u64, + patch: impl FnOnce(&mut NodeState), +) -> io::Result<()> { + let mut state: ServiceState = load_or_default(path); + patch(state.nodes.entry(node_id.to_string()).or_default()); + store(path, &state) +} + +/// Drops one node's entry (after decommissioning) and persists the result. +pub fn remove_node_state(path: &Path, node_id: u64) -> io::Result<()> { + let mut state: ServiceState = load_or_default(path); + state.nodes.remove(&node_id.to_string()); + store(path, &state) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tempdir() -> PathBuf { + let dir = std::env::temp_dir().join(format!("mts-state-test-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + #[test] + fn state_round_trips_and_merges() { + let path = tempdir().join("service-state.json"); + update_node_state(&path, 1, |node| { + node.last_error = Some("boom".into()); + }) + .unwrap(); + update_node_state(&path, 1, |node| { + node.last_error = None; + node.last_successful_sync = Some("2026-07-27T02:20:24.240Z".parse().unwrap()); + }) + .unwrap(); + + let state: ServiceState = load_or_default(&path); + let node = &state.nodes["1"]; + assert_eq!(node.last_error, None); + assert!(node.last_successful_sync.is_some()); + + remove_node_state(&path, 1).unwrap(); + let state: ServiceState = load_or_default(&path); + assert!(state.nodes.is_empty()); + } + + #[test] + fn reads_typescript_state_files() { + // Exact shape written by the TypeScript implementation. + let raw = r#"{ + "nodes": { + "1": { + "lastSuccessfulConnection": "2026-07-27T02:20:23.357Z", + "lastSuccessfulSync": "2026-07-27T02:20:24.240Z", + "lastAttemptedSync": "2026-07-27T02:20:22.437Z", + "lastError": null + } + } + }"#; + let state: ServiceState = serde_json::from_str(raw).unwrap(); + assert_eq!(state.nodes["1"].last_error, None); + assert!(state.nodes["1"].last_successful_sync.is_some()); + } + + #[test] + fn corrupt_files_reduce_to_defaults() { + let path = tempdir().join("corrupt.json"); + std::fs::write(&path, "{ not json").unwrap(); + let state: ServiceState = load_or_default(&path); + assert!(state.nodes.is_empty()); + } + + #[test] + fn registry_lists_node_ids() { + let mut registry = NodeRegistry::default(); + registry.nodes.insert( + "1".into(), + NodeInfo { + vendor_name: Some("IKEA of Sweden".into()), + product_name: Some("ALPSTUGA air quality monitor".into()), + }, + ); + assert_eq!(registry.node_ids().collect::<Vec<_>>(), vec![1]); + } +} diff --git a/src/time.rs b/src/time.rs new file mode 100644 index 0000000..bedc1fa --- /dev/null +++ b/src/time.rs @@ -0,0 +1,367 @@ +//! Matter epoch conversion and device clock assessment. +//! +//! Matter UTC time is microseconds since 2000-01-01T00:00:00Z (the "Matter +//! epoch"), not the Unix epoch. `jiff::Timestamp` is the lingua franca +//! everywhere else in this program; [`MatterMicros`] exists only at the wire +//! boundary. Unlike matter.js, rs-matter passes spec values through +//! unconverted, so no Unix-epoch API shift exists here. + +use std::fmt; + +use jiff::Timestamp; + +pub const MATTER_EPOCH_UNIX_SECONDS: i64 = 946_684_800; +const MATTER_EPOCH_UNIX_MICROS: i64 = MATTER_EPOCH_UNIX_SECONDS * 1_000_000; + +/// Microseconds since the Matter epoch, as carried in Time Synchronization +/// cluster attributes and commands. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct MatterMicros(pub u64); + +impl MatterMicros { + pub fn now() -> Self { + Self::from_timestamp(Timestamp::now()) + } + + /// Panics on pre-2000 timestamps, which cannot be represented on the + /// wire and cannot arise from a running host clock we already required + /// to be NTP-synchronized. + pub fn from_timestamp(at: Timestamp) -> Self { + let micros = at.as_microsecond() - MATTER_EPOCH_UNIX_MICROS; + Self(u64::try_from(micros).expect("timestamp precedes the Matter epoch")) + } + + pub fn to_timestamp(self) -> Timestamp { + Timestamp::from_microsecond(self.0 as i64 + MATTER_EPOCH_UNIX_MICROS) + .expect("Matter timestamp out of jiff range") + } + + /// Signed delta `self - other` in microseconds. + pub fn delta_micros(self, other: Self) -> i64 { + self.0 as i64 - other.0 as i64 + } +} + +impl fmt::Display for MatterMicros { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.to_timestamp().fmt(f) + } +} + +/// A device is treated as epoch-confused when its clock delta lands within +/// this window of the exact Unix/Matter epoch distance: firmware that encodes +/// Unix-epoch values on the wire reads as ~30 years ahead after decoding. A +/// week comfortably covers any real drift while remaining astronomically far +/// from every honest delta. +const EPOCH_SHIFT_DETECTION_WINDOW_MICROS: i64 = 7 * 86_400 * 1_000_000; + +/// How the device's reported clock relates to the host clock. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClockAssessment { + /// The device lost its clock (null utcTime, e.g. after a power outage). + Unset, + /// The device reported a time `delta_micros` away from the host's. + Offset { + /// Raw reported delta (device minus host), exact. + delta_micros: i64, + /// True when the delta is the Unix/Matter epoch distance: the device + /// firmware encodes the wrong epoch on the wire (off-spec). + epoch_shifted: bool, + }, +} + +impl ClockAssessment { + pub fn compare(device: Option<MatterMicros>, host: MatterMicros) -> Self { + let Some(device) = device else { + return Self::Unset; + }; + let delta_micros = device.delta_micros(host); + let shift_error = delta_micros - MATTER_EPOCH_UNIX_MICROS; + Self::Offset { + delta_micros, + epoch_shifted: shift_error.abs() <= EPOCH_SHIFT_DETECTION_WINDOW_MICROS, + } + } + + /// The device's real clock error: the raw delta with any detected epoch + /// shift folded out. `None` when the clock was unset. + pub fn effective_delta_micros(&self) -> Option<i64> { + match *self { + Self::Unset => None, + Self::Offset { + delta_micros, + epoch_shifted, + } => Some(if epoch_shifted { + delta_micros - MATTER_EPOCH_UNIX_MICROS + } else { + delta_micros + }), + } + } + + pub fn is_epoch_shifted(&self) -> bool { + matches!( + self, + Self::Offset { + epoch_shifted: true, + .. + } + ) + } +} + +impl fmt::Display for ClockAssessment { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Unset => f.write_str("device clock was unset"), + Self::Offset { epoch_shifted, .. } => { + let effective = self + .effective_delta_micros() + .expect("Offset always has a delta"); + if *epoch_shifted { + write!( + f, + "device encodes Unix-epoch time on the wire (off-spec); corrected, its clock was {}", + DescribeDelta(effective) + ) + } else { + write!(f, "device clock was {}", DescribeDelta(effective)) + } + } + } + } +} + +/// "within 1s of host time (412ms behind)" / "1m 23s ahead" for a signed delta. +struct DescribeDelta(i64); + +impl fmt::Display for DescribeDelta { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let magnitude = self.0.unsigned_abs(); + let direction = if self.0 < 0 { "behind" } else { "ahead" }; + if magnitude < 1_000_000 { + write!( + f, + "within 1s of host time ({} {direction})", + CompactDuration(magnitude) + ) + } else { + write!(f, "{} {direction}", CompactDuration(magnitude)) + } + } +} + +/// Compact human duration: 412ms, 3.2s, 1m 23s, 1d 1h 1m. +pub struct CompactDuration(pub u64); + +impl fmt::Display for CompactDuration { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let micros = self.0; + if micros < 1_000 { + return write!(f, "{micros}us"); + } + if micros < 1_000_000 { + return write!(f, "{}ms", micros / 1_000); + } + let total_seconds = micros / 1_000_000; + if total_seconds < 60 { + let tenths = (micros % 1_000_000) / 100_000; + return if tenths == 0 { + write!(f, "{total_seconds}s") + } else { + write!(f, "{total_seconds}.{tenths}s") + }; + } + let days = total_seconds / 86_400; + let hours = (total_seconds % 86_400) / 3_600; + let minutes = (total_seconds % 3_600) / 60; + let seconds = total_seconds % 60; + let mut parts: Vec<String> = Vec::new(); + if days > 0 { + parts.push(format!("{days}d")); + } + if hours > 0 { + parts.push(format!("{hours}h")); + } + if minutes > 0 { + parts.push(format!("{minutes}m")); + } + if seconds > 0 && days == 0 { + parts.push(format!("{seconds}s")); + } + f.write_str(&parts.join(" ")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ts(s: &str) -> Timestamp { + s.parse().unwrap() + } + + #[test] + fn matter_epoch_zero_is_year_2000() { + assert_eq!( + MatterMicros::from_timestamp(ts("2000-01-01T00:00:00Z")).0, + 0 + ); + assert_eq!(MatterMicros(0).to_timestamp(), ts("2000-01-01T00:00:00Z")); + assert_eq!( + MatterMicros(1_000_000).to_timestamp(), + ts("2000-01-01T00:00:01Z") + ); + } + + #[test] + fn conversion_round_trips_current_dates() { + let now = ts("2026-07-26T20:00:00.123456Z"); + assert_eq!(MatterMicros::from_timestamp(now).to_timestamp(), now); + } + + #[test] + fn compact_duration_scales_units() { + let cases = [ + (0, "0us"), + (999, "999us"), + (412_000, "412ms"), + (3_200_000, "3.2s"), + (59_000_000, "59s"), + (125_000_000, "2m 5s"), + (3_840_000_000, "1h 4m"), + (90_061_000_000, "1d 1h 1m"), + ]; + for (micros, expected) in cases { + assert_eq!(CompactDuration(micros).to_string(), expected); + } + } + + #[test] + fn assessment_reports_unset_clock() { + let host = MatterMicros::from_timestamp(ts("2026-07-26T20:00:00Z")); + let assessment = ClockAssessment::compare(None, host); + assert_eq!(assessment, ClockAssessment::Unset); + assert_eq!(assessment.to_string(), "device clock was unset"); + } + + #[test] + fn assessment_reports_direction_and_magnitude() { + let host = MatterMicros::from_timestamp(ts("2026-07-26T20:00:00Z")); + let behind = ClockAssessment::compare(Some(MatterMicros(host.0 - 83_000_000)), host); + assert_eq!(behind.to_string(), "device clock was 1m 23s behind"); + assert_eq!(behind.effective_delta_micros(), Some(-83_000_000)); + + let ahead = ClockAssessment::compare(Some(MatterMicros(host.0 + 5_500_000)), host); + assert_eq!(ahead.to_string(), "device clock was 5.5s ahead"); + + let close = ClockAssessment::compare(Some(MatterMicros(host.0 - 412_000)), host); + assert_eq!( + close.to_string(), + "device clock was within 1s of host time (412ms behind)" + ); + } + + #[test] + fn assessment_detects_wrong_epoch_encoding() { + let host = MatterMicros::from_timestamp(ts("2026-07-26T20:00:00Z")); + let shift = (MATTER_EPOCH_UNIX_SECONDS * 1_000_000) as u64; + let wrong = ClockAssessment::compare(Some(MatterMicros(host.0 + shift + 1_400_000)), host); + assert!(wrong.is_epoch_shifted()); + assert_eq!(wrong.effective_delta_micros(), Some(1_400_000)); + assert_eq!( + wrong.to_string(), + "device encodes Unix-epoch time on the wire (off-spec); corrected, its clock was 1.4s ahead" + ); + } + + #[test] + fn assessment_does_not_fold_out_ordinary_large_errors() { + let host = MatterMicros::from_timestamp(ts("2026-07-26T20:00:00Z")); + let shift = (MATTER_EPOCH_UNIX_SECONDS * 1_000_000) as u64; + let month = 30 * 86_400 * 1_000_000; + let broken = ClockAssessment::compare(Some(MatterMicros(host.0 + shift + month)), host); + assert!(!broken.is_epoch_shifted()); + } +} + +/// Parses an operator-supplied wall-clock time: "16:35", "4:35p", "4:35pm", +/// each with an optional ":ss". Meridiem suffixes imply 12-hour form. +pub fn parse_wall_clock(input: &str) -> Result<jiff::civil::Time, String> { + let lowered = input.trim().to_ascii_lowercase(); + let (digits, meridiem) = if let Some(rest) = lowered + .strip_suffix("am") + .or_else(|| lowered.strip_suffix('a')) + { + (rest.trim_end(), Some(false)) + } else if let Some(rest) = lowered + .strip_suffix("pm") + .or_else(|| lowered.strip_suffix('p')) + { + (rest.trim_end(), Some(true)) + } else { + (lowered.as_str(), None) + }; + + let parts: Vec<&str> = digits.split(':').collect(); + if !(2..=3).contains(&parts.len()) { + return Err(format!( + "cannot parse {input:?} as a time (expected HH:MM or HH:MM:SS, optionally with am/pm)" + )); + } + let numbers: Vec<u8> = parts + .iter() + .map(|p| { + p.parse() + .map_err(|_| format!("cannot parse {p:?} in {input:?} as a number")) + }) + .collect::<Result<_, _>>()?; + let (mut hour, minute, second) = (numbers[0], numbers[1], *numbers.get(2).unwrap_or(&0)); + + match meridiem { + Some(pm) => { + if !(1..=12).contains(&hour) { + return Err(format!("hour in {input:?} must be 1-12 with am/pm")); + } + hour = if pm { hour % 12 + 12 } else { hour % 12 }; + } + None if hour > 23 => return Err(format!("hour in {input:?} must be 0-23")), + None => {} + } + jiff::civil::Time::new(hour as i8, minute as i8, second as i8, 0) + .map_err(|e| format!("invalid time {input:?}: {e}")) +} + +#[cfg(test)] +mod wall_clock_tests { + use super::parse_wall_clock; + + #[test] + fn parses_12_and_24_hour_forms() { + let cases = [ + ("16:35", (16, 35, 0)), + ("4:35p", (16, 35, 0)), + ("4:35pm", (16, 35, 0)), + ("4:35a", (4, 35, 0)), + ("12:00am", (0, 0, 0)), + ("12:15PM", (12, 15, 0)), + ("16:35:20", (16, 35, 20)), + ("07:05", (7, 5, 0)), + ]; + for (input, (h, m, s)) in cases { + let time = parse_wall_clock(input).unwrap(); + assert_eq!( + (time.hour(), time.minute(), time.second()), + (h, m, s), + "{input}" + ); + } + } + + #[test] + fn rejects_nonsense() { + for input in ["25:00", "13:00pm", "4", "4:60", "0:00am", "banana"] { + assert!(parse_wall_clock(input).is_err(), "accepted {input:?}"); + } + } +} diff --git a/src/tz.rs b/src/tz.rs new file mode 100644 index 0000000..2820c59 --- /dev/null +++ b/src/tz.rs @@ -0,0 +1,268 @@ +//! IANA time-zone helpers: Matter TimeZone/DSTOffset structure generation +//! and offset introspection for status output. +//! +//! jiff exposes the real tzdb transition table, so upcoming DST changes are +//! read directly instead of probed for (the TypeScript implementation had to +//! binary-search ICU offsets; here `TimeZone::following` is exact by +//! construction). + +use jiff::Timestamp; +use jiff::tz::TimeZone; + +use crate::time::MatterMicros; + +/// Matter TimeZoneStruct: the zone's standard offset, DST carried separately. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MatterTimeZoneEntry { + /// Standard (non-DST) UTC offset in seconds. + pub offset_seconds: i32, + /// Matter-epoch microseconds at which the entry takes effect; 0 = always. + pub valid_at: MatterMicros, + pub name: String, +} + +/// Matter DSTOffsetStruct: one DST period, added on top of the TimeZone offset. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct MatterDstOffsetEntry { + pub offset_seconds: i32, + pub valid_starting: MatterMicros, + /// None = valid until further notice (last entry only). + pub valid_until: Option<MatterMicros>, +} + +/// The next instant the zone's UTC offset changes, with the offsets on both +/// sides. None for fixed-offset zones. +pub fn next_offset_transition(tz: &TimeZone, from: Timestamp) -> Option<(Timestamp, i32, i32)> { + let before = tz.to_offset(from).seconds(); + tz.following(from) + .map(|transition| (transition.timestamp(), transition.offset().seconds())) + .find(|&(_, offset)| offset != before) + .map(|(at, after)| (at, before, after)) +} + +/// The zone's standard (non-DST) UTC offset in seconds: the smaller of the +/// mid-January and mid-July offsets of the year. DST increases the offset in +/// every zone as consumed here (including Europe/Dublin, which the tzdb +/// models with a negative SAVE but presents as +00:00 winter / +01:00 summer). +pub fn standard_offset_seconds(tz: &TimeZone, at: Timestamp) -> i32 { + let year = at.to_zoned(TimeZone::UTC).year(); + let probe = |month: i8| -> i32 { + let date = jiff::civil::date(year, month, 15).at(0, 0, 0, 0); + let ts = date + .to_zoned(TimeZone::UTC) + .expect("UTC has no gaps") + .timestamp(); + tz.to_offset(ts).seconds() + }; + probe(1).min(probe(7)) +} + +/// The TimeZone list for SetTimeZone: a single entry carrying the zone's +/// standard offset and IANA name, valid from the beginning of time. +pub fn build_time_zone_list(tz: &TimeZone, name: &str, at: Timestamp) -> Vec<MatterTimeZoneEntry> { + vec![MatterTimeZoneEntry { + offset_seconds: standard_offset_seconds(tz, at), + valid_at: MatterMicros(0), + name: name.chars().take(64).collect(), + }] +} + +/// The DSTOffset list for SetDstOffset: the DST state in effect at `from` +/// followed by upcoming transitions, at most `max_entries` entries (the +/// device's DSTOffsetListMaxSize; spec minimum 1). Entries carry concrete +/// valid-until bounds where a next transition is known, so an unrefreshed +/// device falls back to standard time rather than trusting stale DST; the +/// periodic sync refreshes the list long before it expires. Zones without +/// transitions yield a single open-ended entry. +pub fn build_dst_offset_list( + tz: &TimeZone, + max_entries: usize, + from: Timestamp, +) -> Vec<MatterDstOffsetEntry> { + let limit = max_entries.max(1); + let standard = standard_offset_seconds(tz, from); + let mut entries = Vec::with_capacity(limit); + + let mut current_offset = tz.to_offset(from).seconds(); + let mut valid_starting = MatterMicros(0); + let mut transitions = tz + .following(from) + .map(|transition| (transition.timestamp(), transition.offset().seconds())); + + while entries.len() < limit { + // Skip tzdb transitions that don't change the offset (abbreviation or + // rule bookkeeping only); they are not DST boundaries. + let next = transitions.find(|&(_, offset)| offset != current_offset); + match next { + Some((at, offset_after)) => { + let until = MatterMicros::from_timestamp(at); + entries.push(MatterDstOffsetEntry { + offset_seconds: current_offset - standard, + valid_starting, + valid_until: Some(until), + }); + valid_starting = until; + current_offset = offset_after; + } + None => { + entries.push(MatterDstOffsetEntry { + offset_seconds: current_offset - standard, + valid_starting, + valid_until: None, + }); + break; + } + } + } + entries +} + +/// "UTC-05:00" style rendering for logs and status output. +pub fn format_utc_offset(offset_seconds: i32) -> String { + let sign = if offset_seconds < 0 { '-' } else { '+' }; + let magnitude = offset_seconds.unsigned_abs(); + format!( + "UTC{sign}{:02}:{:02}", + magnitude / 3600, + (magnitude % 3600) / 60 + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tz(name: &str) -> TimeZone { + TimeZone::get(name).unwrap() + } + + fn ts(s: &str) -> Timestamp { + s.parse().unwrap() + } + + fn matter(s: &str) -> MatterMicros { + MatterMicros::from_timestamp(ts(s)) + } + + // The two Chicago transitions after 2026-07-26. + fn fall_2026() -> MatterMicros { + matter("2026-11-01T07:00:00Z") + } + fn spring_2027() -> MatterMicros { + matter("2027-03-14T08:00:00Z") + } + + #[test] + fn standard_offset_ignores_season_and_hemisphere() { + let chicago = tz("America/Chicago"); + assert_eq!( + standard_offset_seconds(&chicago, ts("2026-07-15T00:00:00Z")), + -6 * 3600 + ); + assert_eq!( + standard_offset_seconds(&chicago, ts("2026-01-15T00:00:00Z")), + -6 * 3600 + ); + // Sydney: AEST UTC+10 standard, AEDT UTC+11 in southern summer. + assert_eq!( + standard_offset_seconds(&tz("Australia/Sydney"), ts("2026-01-15T00:00:00Z")), + 10 * 3600 + ); + assert_eq!( + standard_offset_seconds(&tz("UTC"), ts("2026-07-15T00:00:00Z")), + 0 + ); + assert_eq!( + standard_offset_seconds(&tz("Asia/Kolkata"), ts("2026-07-15T00:00:00Z")), + (5.5 * 3600.0) as i32 + ); + } + + #[test] + fn finds_exact_chicago_transitions() { + let chicago = tz("America/Chicago"); + let (at, before, after) = + next_offset_transition(&chicago, ts("2026-07-15T00:00:00Z")).unwrap(); + // 2026-11-01 02:00 CDT (UTC-5) -> 01:00 CST (UTC-6): 07:00:00 UTC. + assert_eq!(at, ts("2026-11-01T07:00:00Z")); + assert_eq!((before, after), (-5 * 3600, -6 * 3600)); + assert!(next_offset_transition(&tz("UTC"), ts("2026-01-15T00:00:00Z")).is_none()); + } + + #[test] + fn time_zone_list_is_single_standard_entry() { + let list = build_time_zone_list( + &tz("America/Chicago"), + "America/Chicago", + ts("2026-07-26T00:00:00Z"), + ); + assert_eq!( + list, + vec![MatterTimeZoneEntry { + offset_seconds: -6 * 3600, + valid_at: MatterMicros(0), + name: "America/Chicago".into(), + }] + ); + } + + #[test] + fn dst_list_covers_active_period_plus_next() { + let list = build_dst_offset_list(&tz("America/Chicago"), 2, ts("2026-07-26T00:00:00Z")); + assert_eq!( + list, + vec![ + MatterDstOffsetEntry { + offset_seconds: 3600, + valid_starting: MatterMicros(0), + valid_until: Some(fall_2026()), + }, + MatterDstOffsetEntry { + offset_seconds: 0, + valid_starting: fall_2026(), + valid_until: Some(spring_2027()), + }, + ] + ); + } + + #[test] + fn dst_list_respects_device_capacity() { + let list = build_dst_offset_list(&tz("America/Chicago"), 1, ts("2026-07-26T00:00:00Z")); + assert_eq!(list.len(), 1); + assert_eq!(list[0].valid_until, Some(fall_2026())); + } + + #[test] + fn dst_list_is_single_open_ended_zero_for_fixed_zones() { + for name in ["UTC", "Asia/Kolkata"] { + let list = build_dst_offset_list(&tz(name), 2, ts("2026-07-26T00:00:00Z")); + assert_eq!( + list, + vec![MatterDstOffsetEntry { + offset_seconds: 0, + valid_starting: MatterMicros(0), + valid_until: None, + }], + "zone {name}" + ); + } + } + + #[test] + fn dst_list_starts_from_standard_time_in_winter() { + let list = build_dst_offset_list(&tz("America/Chicago"), 2, ts("2026-01-15T00:00:00Z")); + let spring_2026 = matter("2026-03-08T08:00:00Z"); + assert_eq!(list[0].offset_seconds, 0); + assert_eq!(list[0].valid_until, Some(spring_2026)); + assert_eq!(list[1].offset_seconds, 3600); + assert_eq!(list[1].valid_starting, spring_2026); + } + + #[test] + fn formats_utc_offsets() { + assert_eq!(format_utc_offset(-6 * 3600), "UTC-06:00"); + assert_eq!(format_utc_offset(19_800), "UTC+05:30"); + assert_eq!(format_utc_offset(0), "UTC+00:00"); + } +} |