matc/
discover.rs

1//! Module with very simple mdns based discovery of matter devices.
2//! Usually application shall discover devices using these methods and filter according discriminator.
3//! This module tries to send mdns using ipv4 and ipv6 multicast at same time.
4//! If more control over discovery mechanism is required, it may be better to use some external mdns library.
5
6use crate::{mdns::{self, DnsMessage}, mdns2};
7use anyhow::{Context, Result};
8use byteorder::ReadBytesExt;
9use std::{
10    collections::{BTreeMap, HashMap},
11    io::{Cursor, Read},
12    net::{IpAddr, Ipv4Addr, Ipv6Addr},
13    time::Duration,
14};
15use tokio_util::bytes::Buf;
16
17#[derive(Debug, Clone)]
18pub enum CommissioningMode {
19    No,
20    Yes,
21    WithPasscode,
22}
23
24#[derive(Debug, Clone)]
25pub struct MatterDeviceInfo {
26    pub instance: String,
27    pub device: String,
28    pub ips: Vec<IpAddr>,
29    pub name: Option<String>,
30    pub vendor_id: Option<String>,
31    pub product_id: Option<String>,
32    pub discriminator: Option<String>,
33    pub commissioning_mode: Option<CommissioningMode>,
34    pub pairing_hint: Option<String>,
35    pub source_ip: String,
36    pub port: Option<u16>,
37    /// MRP idle interval (SII TXT key, milliseconds)
38    pub session_idle_interval_ms: Option<u32>,
39    /// MRP active interval (SAI TXT key, milliseconds)
40    pub session_active_interval_ms: Option<u32>,
41    /// MRP active threshold (SAT TXT key, milliseconds)
42    pub session_active_threshold_ms: Option<u32>,
43    /// Device type (DT TXT key) from the commissionable advertisement, decimal string.
44    pub device_type: Option<String>,
45    /// scope_id (interface index) for the device's link-local IPv6 addresses: the
46    /// interface on which its mDNS reply arrived. Needed to send to `fe80::...`.
47    pub scope_id: Option<u32>,
48}
49
50impl MatterDeviceInfo {
51    /// MRP timing parameters from the advertised SII/SAI/SAT values,
52    /// with spec defaults for missing keys.
53    pub fn mrp_params(&self) -> crate::mrp::MrpParameters {
54        crate::mrp::MrpParameters::from_txt_ms(
55            self.session_idle_interval_ms,
56            self.session_active_interval_ms,
57            self.session_active_threshold_ms,
58        )
59    }
60
61    pub fn print_compact(&self) {
62        let mut info = format!("{} ({})", self.instance, self.device);
63        if let Some(name) = &self.name {
64            info += &format!(", name: {}", name);
65        }
66        if let Some(vendor_id) = &self.vendor_id {
67            info += &format!(", vendor_id: {}", vendor_id);
68        }
69        if let Some(product_id) = &self.product_id {
70            info += &format!(", product_id: {}", product_id);
71        }
72        if let Some(discriminator) = &self.discriminator {
73            info += &format!(", discriminator: {}", discriminator);
74        }
75        if let Some(cm) = &self.commissioning_mode {
76            info += &format!(", commissioning_mode: {:?}", cm);
77        }
78        if let Some(pairing_hint) = &self.pairing_hint {
79            info += &format!(", pairing_hint: {}", pairing_hint);
80        }
81        if let Some(port) = &self.port {
82            info += &format!(", port: {}", port);
83        }
84        if let Some(sii) = &self.session_idle_interval_ms {
85            info += &format!(", sii_ms: {}", sii);
86        }
87        if let Some(sai) = &self.session_active_interval_ms {
88            info += &format!(", sai_ms: {}", sai);
89        }
90        println!("{}", info);
91        if !self.ips.is_empty() {
92            println!("  ips:");
93            for ip in &self.ips {
94                println!("      {}", ip);
95            }
96        }
97
98    }
99}
100
101
102pub fn parse_txt_records(data: &[u8]) -> Result<HashMap<String, String>> {
103    let mut cursor = Cursor::new(data);
104    let mut out = HashMap::new();
105    while cursor.remaining() > 0 {
106        let len = cursor.read_u8()?;
107        let mut buf = vec![0; len as usize];
108        cursor.read_exact(buf.as_mut_slice())?;
109        let splitstr = std::str::from_utf8(&buf)?.splitn(2, "=");
110        let x: Vec<&str> = splitstr.collect();
111        if x.len() == 2 {
112            out.insert(x[0].to_owned(), x[1].to_owned());
113        }
114    }
115    Ok(out)
116}
117
118/// Extract (SII, SAI, SAT) millisecond values from parsed TXT records.
119/// Unparseable values are ignored.
120fn parse_mrp_txt(rec: &HashMap<String, String>) -> (Option<u32>, Option<u32>, Option<u32>) {
121    let get = |key: &str| rec.get(key).and_then(|v| v.parse::<u32>().ok());
122    (get("SII"), get("SAI"), get("SAT"))
123}
124
125fn remove_string_suffix(string: &str, suffix: &str) -> String {
126    if let Some(s) = string.strip_suffix(suffix) {
127        s.to_owned()
128    } else {
129        string.to_owned()
130    }
131}
132
133pub fn to_matter_info2(msg: &DnsMessage, svc: &str) -> Result<Vec<MatterDeviceInfo>> {
134    let mut out = Vec::new();
135    let mut matter_service = false;
136    let svcname = ".".to_owned() + svc + ".";
137    for answer in &msg.answers {
138        if answer.name == svcname[1..] {
139            matter_service = true
140        }
141    }
142    if !matter_service {
143        return Err(anyhow::anyhow!("not matter service"));
144    }
145    let mut services = HashMap::new();
146    let mut targets = HashMap::new();
147    for additional in &msg.additional {
148        if additional.typ == mdns::TYPE_A {
149            let arr: [u8; 4] = match additional.rdata.clone().try_into() {
150                Ok(v) => v,
151                Err(_e) => return Err(anyhow::anyhow!("A record is not correct")),
152            };
153            let val = IpAddr::V4(Ipv4Addr::from_bits(u32::from_be_bytes(arr)));
154            if !targets.contains_key(&additional.name) {
155                targets.insert(additional.name.clone(), Vec::new());
156            }
157            targets.get_mut(&additional.name).unwrap().push(val);
158        }
159        if additional.typ == mdns::TYPE_AAAA {
160            let arr: [u8; 16] = match additional.rdata.clone().try_into() {
161                Ok(v) => v,
162                Err(_e) => return Err(anyhow::anyhow!("AAAA record is not correct")),
163            };
164            let val = IpAddr::V6(Ipv6Addr::from_bits(u128::from_be_bytes(arr)));
165            if !targets.contains_key(&additional.name) {
166                targets.insert(additional.name.clone(), Vec::new());
167            }
168            targets.get_mut(&additional.name).unwrap().push(val);
169        }
170    }
171    let mut all = msg.additional.to_vec();
172    all.append(&mut msg.answers.to_vec());
173    for additional in &all {
174        if additional.typ == mdns::TYPE_SRV {
175            let service_name = remove_string_suffix(&additional.name, &svcname);
176            if additional.rdata.len() < 6 {
177                continue;
178            }
179            let port = ((additional.rdata[4] as u16) << 8) | (additional.rdata[5] as u16);
180            let target_name = {
181                if let Some(at) = additional.target.as_ref() {
182                    at
183                } else {
184                    continue;
185                }
186            };
187            let target_ip = targets.get(target_name).cloned().unwrap_or_default();
188            let mi = MatterDeviceInfo {
189                instance: service_name.clone(),
190                device: remove_string_suffix(target_name, ".local.").to_owned(),
191                ips: target_ip,
192                name: None,
193                discriminator: None,
194                commissioning_mode: None,
195                pairing_hint: None,
196                source_ip: msg.source.to_string(),
197                vendor_id: None,
198                product_id: None,
199                port: Some(port),
200                session_idle_interval_ms: None,
201                session_active_interval_ms: None,
202                session_active_threshold_ms: None,
203                device_type: None,
204                scope_id: None,
205            };
206            services.insert(service_name, mi);
207        }
208    }
209    for s in services.values() {
210        out.push(s.clone());
211    }
212
213    Ok(out)
214}
215
216pub fn to_matter_info(msg: &DnsMessage, svc: &str) -> Result<MatterDeviceInfo> {
217    let mut device = None;
218    let mut service = None;
219    let mut ips = BTreeMap::new();
220    let mut name = None;
221    let mut discriminator = None;
222    let mut cm = None;
223    let mut pairing_hint = None;
224    let mut vendor_id = None;
225    let mut product_id = None;
226    let mut port: Option<u16> = None;
227    let mut mrp = (None, None, None);
228    let mut device_type = None;
229
230    let mut matter_service = false;
231    let svcname = ".".to_owned() + svc + ".";
232    for answer in &msg.answers {
233        if answer.name == svcname[1..] {
234            matter_service = true
235        }
236    }
237    for additional in &msg.additional {
238        if additional.typ == mdns::TYPE_A {
239            let arr: [u8; 4] = match additional.rdata.clone().try_into() {
240                Ok(v) => v,
241                Err(_e) => return Err(anyhow::anyhow!("A record is not correct")),
242            };
243            let val = IpAddr::V4(Ipv4Addr::from_bits(u32::from_be_bytes(arr)));
244            ips.insert(val, true);
245            device = Some(remove_string_suffix(&additional.name, ".local."));
246        }
247        if additional.typ == mdns::TYPE_AAAA {
248            let arr: [u8; 16] = match additional.rdata.clone().try_into() {
249                Ok(v) => v,
250                Err(_e) => return Err(anyhow::anyhow!("AAAA record is not correct")),
251            };
252            let val = IpAddr::V6(Ipv6Addr::from_bits(u128::from_be_bytes(arr)));
253            ips.insert(val, true);
254            device = Some(remove_string_suffix(&additional.name, ".local."));
255        }
256        if additional.typ == mdns::TYPE_SRV {
257            service = Some(remove_string_suffix(&additional.name, &svcname));
258            if additional.rdata.len() >= 6 {
259                port = Some(((additional.rdata[4] as u16) << 8) | (additional.rdata[5] as u16))
260            }
261        }
262        if additional.typ == mdns::TYPE_TXT {
263            let rec = parse_txt_records(&additional.rdata)?;
264            name = rec.get("DN").cloned();
265            discriminator = rec.get("D").cloned();
266            pairing_hint = rec.get("PH").cloned();
267            device_type = rec.get("DT").cloned();
268            mrp = parse_mrp_txt(&rec);
269            if let Some(vp) = rec.get("VP") {
270                let mut split = vp.split("+");
271                vendor_id = split.next().map(str::to_owned);
272                product_id = split.next().map(str::to_owned);
273            }
274            cm = match rec.get("CM") {
275                Some(v) => match v.as_str() {
276                    "0" => Some(CommissioningMode::No),
277                    "1" => Some(CommissioningMode::Yes),
278                    "2" => Some(CommissioningMode::WithPasscode),
279                    _ => None,
280                },
281                None => None,
282            };
283        }
284    }
285
286    if !matter_service {
287        return Err(anyhow::anyhow!("not matter service"));
288    }
289
290    Ok(MatterDeviceInfo {
291        instance: service.context("service name not detected")?,
292        device: device.context("device name not detected")?,
293        ips: ips.into_keys().collect(),
294        name,
295        discriminator,
296        commissioning_mode: cm,
297        pairing_hint,
298        source_ip: msg.source.to_string(),
299        vendor_id,
300        product_id,
301        port,
302        session_idle_interval_ms: mrp.0,
303        session_active_interval_ms: mrp.1,
304        session_active_threshold_ms: mrp.2,
305        device_type,
306        scope_id: None,
307    })
308}
309
310async fn discover_common(timeout: Duration, svc_type: &str) -> Result<Vec<MatterDeviceInfo>> {
311    let stop = tokio_util::sync::CancellationToken::new();
312    let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel::<DnsMessage>();
313
314    mdns::discover(svc_type, mdns::QTYPE_ANY, sender, stop.child_token()).await?;
315
316    tokio::spawn(async move {
317        tokio::time::sleep(timeout).await;
318        stop.cancel();
319    });
320    let mut cache = HashMap::new();
321    let mut out = Vec::new();
322    while let Some(dns) = receiver.recv().await {
323        if cache.contains_key(&dns) {
324            continue;
325        }
326        let info = match to_matter_info(&dns, svc_type) {
327            Ok(info) => info,
328            Err(_) => continue,
329        };
330        out.push(info);
331        cache.insert(dns, true);
332    }
333    Ok(out)
334}
335
336/// Discover commissionable devices using mdns
337pub async fn discover_commissionable(timeout: Duration) -> Result<Vec<MatterDeviceInfo>> {
338    discover_common(timeout, "_matterc._udp.local").await
339}
340
341/// Discover commissioned devices using mdns
342pub async fn discover_commissioned(timeout: Duration) -> Result<Vec<MatterDeviceInfo>> {
343    discover_common(timeout, "_matter._tcp.local").await
344}
345
346
347async fn discover_common2(timeout: Duration, svc_type: &str) -> Result<Vec<MatterDeviceInfo>> {
348    let stop = tokio_util::sync::CancellationToken::new();
349    let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel::<DnsMessage>();
350
351    mdns::discover(svc_type, mdns::QTYPE_ANY, sender, stop.child_token()).await?;
352
353    tokio::spawn(async move {
354        tokio::time::sleep(timeout).await;
355        stop.cancel();
356    });
357    let mut cache = HashMap::new();
358    let mut out: Vec<MatterDeviceInfo> = Vec::new();
359    while let Some(dns) = receiver.recv().await {
360        if cache.contains_key(&dns) {
361            continue;
362        }
363        let info = match to_matter_info2(&dns, svc_type) {
364            Ok(info) => info,
365            Err(e) => {
366                log::trace!("failed to parse mdns message from {}: {:?}", dns.source, e);
367                continue;
368            },
369        };
370        for i in &info {
371            out.push(i.clone());
372        }
373        cache.insert(dns, true);
374    }
375    Ok(out)
376}
377
378/// Discover commissionable devices using mdns
379pub async fn discover_commissionable2(timeout: Duration) -> Result<Vec<MatterDeviceInfo>> {
380    discover_common2(timeout, "_matterc._udp.local").await
381}
382
383/// Discover commissioned devices using mdns
384pub async fn discover_commissioned2(timeout: Duration, device: &Option<String>) -> Result<Vec<MatterDeviceInfo>> {
385    let query = {
386        match device {
387            None => "_matter._tcp.local".to_owned(),
388            Some(d) => format!("{}._matter._tcp.local", d),
389        }
390    };
391    discover_common2(timeout, &query).await
392}
393
394
395
396/// Discover the first device matching a predicate.
397///
398/// Subscribes to the broadcast channel, sends `query` as an active mDNS lookup, then
399/// drains events until one matching `service_name` passes `predicate`. Lag events (dropped
400/// due to buffer overflow) are logged and skipped; discovery continues normally.
401///
402/// `predicate` receives the full instance target string and the parsed `MatterDeviceInfo`.
403pub async fn discover_one<F>(
404    mdns: &mdns2::MdnsService,
405    query: &str,
406    service_name: &str,
407    timeout: Duration,
408    predicate: F,
409) -> Result<(String, MatterDeviceInfo)>
410where
411    F: Fn(&str, &MatterDeviceInfo) -> bool,
412{
413    let mut rx = mdns.subscribe();
414    mdns.active_lookup(query, mdns::QTYPE_ANY).await;
415    let deadline = std::time::Instant::now() + timeout;
416    loop {
417        let remaining = deadline.saturating_duration_since(std::time::Instant::now());
418        if remaining.is_zero() {
419            anyhow::bail!("mDNS discovery timeout for {}", query);
420        }
421        match tokio::time::timeout(remaining, rx.recv()).await {
422            Err(_) => anyhow::bail!("mDNS discovery timeout for {}", query),
423            Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(n))) => {
424                log::warn!("mDNS discovery: dropped {} events due to lag, continuing", n);
425            }
426            Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => {
427                anyhow::bail!("mDNS service closed");
428            }
429            Ok(Ok(mdns2::MdnsEvent::ServiceExpired { .. })) => {}
430            Ok(Ok(mdns2::MdnsEvent::ServiceDiscovered { name, target, .. })) => {
431                if name != service_name {
432                    continue;
433                }
434                let info = match extract_matter_info(&target, mdns).await {
435                    Ok(i) => i,
436                    Err(e) => {
437                        log::debug!("failed to extract Matter info from {}: {}", target, e);
438                        continue;
439                    }
440                };
441                if predicate(&target, &info) {
442                    return Ok((target, info));
443                }
444            }
445        }
446    }
447}
448
449/// Discover all matching devices until the timeout expires.
450///
451/// Like [`discover_one`] but collects every device whose `ServiceDiscovered` event
452/// matches `service_name` and for which `extract_matter_info` succeeds, until `timeout`
453/// elapses. Returns an empty `Vec` if no devices are found (not an error).
454pub async fn discover_all(
455    mdns: &mdns2::MdnsService,
456    query: &str,
457    service_name: &str,
458    timeout: Duration,
459) -> Result<Vec<(String, MatterDeviceInfo)>> {
460    let mut rx = mdns.subscribe();
461    mdns.active_lookup(query, mdns::QTYPE_ANY).await;
462    let deadline = std::time::Instant::now() + timeout;
463    let mut out = Vec::new();
464    loop {
465        let remaining = deadline.saturating_duration_since(std::time::Instant::now());
466        if remaining.is_zero() {
467            break;
468        }
469        match tokio::time::timeout(remaining, rx.recv()).await {
470            Err(_) => break,
471            Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(n))) => {
472                log::warn!("mDNS discover_all: dropped {} events due to lag, continuing", n);
473            }
474            Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => break,
475            Ok(Ok(mdns2::MdnsEvent::ServiceExpired { .. })) => {}
476            Ok(Ok(mdns2::MdnsEvent::ServiceDiscovered { name, target, .. })) => {
477                if name != service_name {
478                    continue;
479                }
480                match extract_matter_info(&target, mdns).await {
481                    Ok(info) => out.push((target, info)),
482                    Err(e) => {
483                        log::debug!("failed to extract Matter info from {}: {}", target, e);
484                    }
485                }
486            }
487        }
488    }
489    Ok(out)
490}
491
492pub async fn extract_matter_info(target: &str, mdns: &mdns2::MdnsService) -> Result<MatterDeviceInfo> {
493    let txt_records = mdns.lookup(target, mdns::TYPE_TXT).await;
494    let mut txt_info = HashMap::new();
495    for txt_rr in txt_records {
496        txt_info.extend(parse_txt_records(&txt_rr.rdata)?);
497    }
498    let srv_records = mdns.lookup(target, mdns::TYPE_SRV).await;
499    let srv_rr = srv_records.first().ok_or_else(|| anyhow::anyhow!("No SRV record found for {}", target))?;
500    let (srv_target, port) = match srv_rr.data {
501        mdns::RRData::SRV { ref target, port, .. } => (target.clone(), port),
502        _ => return Err(anyhow::anyhow!("Invalid SRV record for {}", target)),
503    };
504    let mut ips = Vec::new();
505    let a_records = mdns.lookup(&srv_target, mdns::TYPE_A).await;
506    for a_rr in a_records {
507        if let mdns::RRData::A(ip) = a_rr.data {
508            ips.push(ip.into());
509        }
510    }
511    let aaaa_records = mdns.lookup(&srv_target, mdns::TYPE_AAAA).await;
512    for aaaa_rr in aaaa_records {
513        if let mdns::RRData::AAAA(ip) = aaaa_rr.data {
514            ips.push(ip.into());
515        }
516    }
517    let (vendor_id, product_id) = {
518        let vp = txt_info.get("VP");
519        if let Some(vp) = vp {
520            let mut parts = vp.split('+');
521            let vendor_id = parts.next();
522            let product_id = parts.next();
523            (vendor_id.map(|v| v.to_owned()), product_id.map(|p| p.to_owned()))
524        } else {
525            (None, None)
526        }
527    };
528    let discriminator = txt_info.get("D").cloned();
529    let name = txt_info.get("DN").cloned();
530    let commissioning_mode = match txt_info.get("CM") {
531                Some(v) => match v.as_str() {
532                    "0" => Some(CommissioningMode::No),
533                    "1" => Some(CommissioningMode::Yes),
534                    "2" => Some(CommissioningMode::WithPasscode),
535                    _ => None,
536                },
537                None => None,
538            };
539    let pairing_hint = txt_info.get("PH").cloned();
540    let device_type = txt_info.get("DT").cloned();
541    let (sii, sai, sat) = parse_mrp_txt(&txt_info);
542    // Correct scope_id for link-local addresses: the interface on which mDNS
543    // received the device's reply (see MdnsService::scope_for).
544    let mut scope_id = None;
545    for ip in &ips {
546        if let IpAddr::V6(v6) = ip {
547            if (v6.segments()[0] & 0xffc0) == 0xfe80 {
548                scope_id = mdns.scope_for(v6).await;
549                if scope_id.is_some() {
550                    break;
551                }
552            }
553        }
554    }
555    Ok(MatterDeviceInfo {
556        name,
557        instance: target.trim_end_matches('.').to_owned(),
558        device: srv_target.trim_end_matches('.').to_owned(),
559        ips,
560        vendor_id,
561        product_id,
562        discriminator,
563        commissioning_mode,
564        pairing_hint,
565        source_ip: "".to_owned(),
566        port: Some(port),
567        session_idle_interval_ms: sii,
568        session_active_interval_ms: sai,
569        session_active_threshold_ms: sat,
570        device_type,
571        scope_id,
572    })
573}
574
575/// Build the address string for a UDP connection. For link-local IPv6 it appends
576/// the zone `%<scope_id>` (interface index); without it the OS cannot send to
577/// `fe80::...`.
578pub fn addr_string(ip: &IpAddr, port: u16, scope_id: Option<u32>) -> String {
579    match ip {
580        IpAddr::V6(v6) if (v6.segments()[0] & 0xffc0) == 0xfe80 => match scope_id {
581            Some(idx) => format!("[{}%{}]:{}", v6, idx, port),
582            None => format!("[{}]:{}", v6, port),
583        },
584        IpAddr::V6(v6) => format!("[{}]:{}", v6, port),
585        IpAddr::V4(v4) => format!("{}:{}", v4, port),
586    }
587}
588#[cfg(test)]
589mod tests {
590    use super::*;
591
592    fn txt_rdata(entries: &[&str]) -> Vec<u8> {
593        let mut out = Vec::new();
594        for e in entries {
595            out.push(e.len() as u8);
596            out.extend_from_slice(e.as_bytes());
597        }
598        out
599    }
600
601    #[test]
602    fn test_parse_mrp_txt() {
603        let rec = parse_txt_records(&txt_rdata(&["SII=5000", "SAI=300", "SAT=4000", "D=840"]))
604            .unwrap();
605        assert_eq!(parse_mrp_txt(&rec), (Some(5000), Some(300), Some(4000)));
606
607        let rec = parse_txt_records(&txt_rdata(&["SII=abc", "D=840"])).unwrap();
608        assert_eq!(parse_mrp_txt(&rec), (None, None, None));
609
610        let rec = parse_txt_records(&txt_rdata(&["D=840"])).unwrap();
611        assert_eq!(parse_mrp_txt(&rec), (None, None, None));
612    }
613}
matc/discover.rs

matc/
discover.rs
