day_part_zeroconf/
lib.rs

1// Copyright © The Daybrite Project
2// SPDX-License-Identifier: MPL-2.0
3//! Native DNS-SD discovery and advertisement. No async runtime is required.
4//!
5//! Operations own their native registrations. Dropping an operation cancels it;
6//! `stop` waits for the coordinator to request native cleanup. Mobile OS teardown
7//! may complete asynchronously. Call `next_event().await` from any executor, or use
8//! `recv_timeout` on a worker thread. Events contain protocol data, not UI copy.
9use std::{
10    collections::{BTreeMap, BTreeSet, VecDeque},
11    fmt,
12    future::Future,
13    net::SocketAddr,
14    pin::Pin,
15    sync::{
16        Arc, Condvar, Mutex,
17        atomic::{AtomicBool, Ordering},
18        mpsc,
19    },
20    task::{Context, Poll, Waker},
21    thread,
22    time::Duration,
23};
24
25pub const ENUMERATION_TYPE: &str = "_services._dns-sd._udp";
26const MAX_SERVICES: usize = 1024;
27const MAX_TYPES: usize = 128;
28
29#[derive(Clone, Debug, PartialEq, Eq)]
30#[non_exhaustive]
31pub enum Error {
32    Unsupported,
33    InvalidInput(String),
34    PermissionDenied,
35    Unavailable(String),
36    Native { code: i64, message: String },
37    Overflow,
38    Cancelled,
39}
40impl fmt::Display for Error {
41    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
42        write!(f, "{self:?}")
43    }
44}
45impl std::error::Error for Error {}
46pub type Result<T, E = Error> = std::result::Result<T, E>;
47
48pub(crate) fn trace_enabled() -> bool {
49    static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
50    *ENABLED.get_or_init(|| std::env::var("DAY_ZEROCONF_TRACE").is_ok_and(|v| v == "1"))
51}
52// Called by workers, never from a native callback. Ignore stderr write failures.
53pub(crate) fn trace(message: impl fmt::Display) {
54    if trace_enabled() {
55        use std::io::Write;
56        let _ = writeln!(std::io::stderr().lock(), "zeroconf: {message}");
57    }
58}
59
60/// A service observation belongs to an interface. Names are never persistent device IDs.
61#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
62pub struct ServiceId {
63    pub name: String,
64    pub service_type: String,
65    pub domain: String,
66    pub interface: u32,
67}
68#[derive(Clone, Debug, PartialEq, Eq)]
69pub struct Service {
70    pub id: ServiceId,
71    pub hostname: String,
72    pub port: u16,
73    pub addresses: Vec<SocketAddr>,
74    pub txt: TxtRecord,
75}
76/// Values are opaque bytes. None (key only) differs from Some(empty) (key=).
77#[derive(Clone, Debug, Default, PartialEq, Eq)]
78pub struct TxtRecord(pub Vec<(String, Option<Vec<u8>>)>);
79impl TxtRecord {
80    pub fn get(&self, key: &str) -> Option<&[u8]> {
81        self.0
82            .iter()
83            .find(|(k, _)| k.eq_ignore_ascii_case(key))
84            .and_then(|(_, v)| v.as_deref())
85    }
86    pub fn encode(&self) -> Result<Vec<u8>> {
87        let mut out = Vec::new();
88        let mut seen = BTreeSet::new();
89        for (key, value) in &self.0 {
90            if key.is_empty()
91                || !key.bytes().all(|b| (0x20..=0x7e).contains(&b) && b != b'=')
92                || !seen.insert(key.to_ascii_lowercase())
93            {
94                return Err(Error::InvalidInput("TXT key".into()));
95            }
96            let n = key.len() + value.as_ref().map_or(0, |v| v.len() + 1);
97            if n > 255 {
98                return Err(Error::InvalidInput("TXT item exceeds 255 bytes".into()));
99            }
100            out.push(n as u8);
101            out.extend_from_slice(key.as_bytes());
102            if let Some(v) = value {
103                out.push(b'=');
104                out.extend_from_slice(v)
105            }
106        }
107        if out.len() > 1300 {
108            return Err(Error::InvalidInput(
109                "TXT record exceeds portable 1300 byte limit".into(),
110            ));
111        }
112        Ok(out)
113    }
114    pub fn decode(bytes: &[u8]) -> Result<Self> {
115        let mut rest = bytes;
116        let mut result = Self::default();
117        let mut seen = BTreeSet::new();
118        while let Some((&n, tail)) = rest.split_first() {
119            let n = n as usize;
120            if tail.len() < n {
121                return Err(Error::InvalidInput("truncated TXT record".into()));
122            }
123            let item = &tail[..n];
124            rest = &tail[n..];
125            if item.is_empty() {
126                continue;
127            }
128            let split = item.iter().position(|b| *b == b'=');
129            let key = &item[..split.unwrap_or(n)];
130            let Ok(key) = std::str::from_utf8(key) else {
131                continue;
132            };
133            if key.is_empty()
134                || !key.bytes().all(|b| (0x20..=0x7e).contains(&b))
135                || !seen.insert(key.to_ascii_lowercase())
136            {
137                continue;
138            }
139            result
140                .0
141                .push((key.into(), split.map(|i| item[i + 1..].to_vec())));
142        }
143        Ok(result)
144    }
145}
146
147#[derive(Clone, Debug)]
148pub struct BrowseOptions {
149    /// None enumerates advertised service types, then browses each.
150    pub service_type: Option<String>,
151    pub interface: u32,
152}
153impl BrowseOptions {
154    pub fn all() -> Self {
155        Self {
156            service_type: None,
157            interface: 0,
158        }
159    }
160    pub fn service(service_type: impl Into<String>) -> Self {
161        Self {
162            service_type: Some(service_type.into()),
163            interface: 0,
164        }
165    }
166}
167#[derive(Clone, Debug)]
168pub struct Advertisement {
169    pub name: String,
170    pub service_type: String,
171    pub port: u16,
172    pub txt: TxtRecord,
173    pub interface: u32,
174}
175#[derive(Clone, Debug, PartialEq, Eq)]
176#[non_exhaustive]
177pub enum Event {
178    Started,
179    ServiceType(String),
180    Found(ServiceId),
181    Resolved(Service),
182    Removed(ServiceId),
183    Registered(ServiceId),
184    Error(Error),
185    Stopped,
186}
187#[derive(Clone, Copy, Debug)]
188pub struct Capabilities {
189    /// A backend exists; availability and permissions are checked when it starts.
190    pub native: bool,
191    /// Enumerates service types. Mobile fallback uses the default IPv4 multicast route.
192    pub browse_all: bool,
193    /// Arbitrary byte values can be advertised without conversion.
194    pub binary_txt: bool,
195    /// Key-only TXT attributes can be advertised without becoming empty values.
196    pub key_only_txt: bool,
197    /// A caller may choose a native interface index instead of zero (automatic).
198    pub interface_selection: bool,
199}
200pub fn capabilities() -> Capabilities {
201    backend::capabilities()
202}
203
204pub fn validate_type(value: &str) -> Result<String> {
205    let value = value.trim_end_matches('.');
206    let value = value.strip_suffix(".local").unwrap_or(value);
207    let Some((name, transport)) = value.split_once('.') else {
208        return Err(Error::InvalidInput("service type".into()));
209    };
210    if !name.starts_with('_')
211        || name.len() < 2
212        || name.len() > 63
213        || !name[1..]
214            .bytes()
215            .all(|b| b.is_ascii_alphanumeric() || b == b'-')
216        || !matches!(transport, "_tcp" | "_udp")
217    {
218        return Err(Error::InvalidInput("service type".into()));
219    }
220    Ok(value.to_owned())
221}
222
223struct Queue {
224    events: VecDeque<Event>,
225    closed: bool,
226    waker: Option<Waker>,
227}
228struct Inbox {
229    state: Mutex<Queue>,
230    ready: Condvar,
231}
232impl Inbox {
233    fn push(&self, event: Event) {
234        let mut s = self.state.lock().unwrap();
235        if s.closed {
236            return;
237        }
238        if s.events.len() >= MAX_SERVICES {
239            s.events.clear();
240            s.events.push_back(Event::Error(Error::Overflow));
241            s.closed = true
242        } else {
243            s.events.push_back(event)
244        }
245        let waker = s.waker.take();
246        drop(s);
247        if let Some(w) = waker {
248            w.wake();
249        }
250        self.ready.notify_all();
251    }
252    fn close(&self) {
253        let mut s = self.state.lock().unwrap();
254        s.closed = true;
255        let waker = s.waker.take();
256        drop(s);
257        if let Some(w) = waker {
258            w.wake();
259        }
260        self.ready.notify_all()
261    }
262}
263/// A single-consumer event subscription. Native callbacks never run application code.
264pub struct Operation {
265    inbox: Arc<Inbox>,
266    cancelled: Arc<AtomicBool>,
267    worker: Option<thread::JoinHandle<()>>,
268}
269/// Cancels an operation without blocking the UI thread. Clones share one cancellation flag.
270#[derive(Clone)]
271pub struct Cancellation(Arc<AtomicBool>);
272impl Cancellation {
273    pub fn cancel(&self) {
274        self.0.store(true, Ordering::Release);
275    }
276}
277impl Operation {
278    pub fn cancellation(&self) -> Cancellation {
279        Cancellation(self.cancelled.clone())
280    }
281    pub fn next_event(&mut self) -> Next<'_> {
282        Next(self)
283    }
284    pub fn recv_timeout(&mut self, timeout: Duration) -> Option<Event> {
285        let s = self.inbox.state.lock().unwrap();
286        let (mut s, _) = self
287            .inbox
288            .ready
289            .wait_timeout_while(s, timeout, |s| s.events.is_empty() && !s.closed)
290            .unwrap();
291        s.events.pop_front()
292    }
293    pub fn stop(&mut self) {
294        self.cancelled.store(true, Ordering::Release);
295        if let Some(t) = self.worker.take() {
296            let _ = t.join();
297        }
298        self.inbox.close()
299    }
300}
301impl Drop for Operation {
302    fn drop(&mut self) {
303        self.cancelled.store(true, Ordering::Release);
304        self.inbox.close();
305    }
306}
307pub struct Next<'a>(&'a mut Operation);
308impl Future for Next<'_> {
309    type Output = Option<Event>;
310    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
311        let mut s = self.0.inbox.state.lock().unwrap();
312        if let Some(e) = s.events.pop_front() {
313            Poll::Ready(Some(e))
314        } else if s.closed {
315            Poll::Ready(None)
316        } else {
317            s.waker = Some(cx.waker().clone());
318            Poll::Pending
319        }
320    }
321}
322
323#[derive(Clone, Debug)]
324pub(crate) enum Request {
325    Browse(String, u32),
326    Resolve(ServiceId),
327    Advertise(Advertisement),
328}
329#[derive(Clone)]
330pub(crate) struct Sink {
331    sender: mpsc::SyncSender<(u64, Event)>,
332    generation: u64,
333    overflow: Arc<AtomicBool>,
334}
335pub(crate) trait Native: Send {}
336impl<T: Send> Native for T {}
337pub(crate) fn emit(sink: &Sink, event: Event) {
338    if let Err(mpsc::TrySendError::Full(_)) = sink.sender.try_send((sink.generation, event)) {
339        sink.overflow.store(true, Ordering::Release);
340    }
341}
342
343/// Browse local services, automatically resolving each observation. Keep the operation alive.
344pub fn browse(mut options: BrowseOptions) -> Result<Operation> {
345    if let Some(t) = &options.service_type {
346        options.service_type = Some(validate_type(t)?)
347    }
348    start(Request::Browse(
349        options
350            .service_type
351            .unwrap_or_else(|| ENUMERATION_TYPE.into()),
352        options.interface,
353    ))
354}
355/// Resolve a known local identity. Some backends emit updates; refresh to guarantee freshness.
356pub fn resolve(mut id: ServiceId) -> Result<Operation> {
357    id.service_type = validate_type(&id.service_type)?;
358    if id.name.is_empty()
359        || id.name.len() > 63
360        || id.name.contains('\0')
361        || !matches!(id.domain.as_str(), "local" | "local.")
362    {
363        return Err(Error::InvalidInput("local service identity".into()));
364    }
365    id.domain = "local.".into();
366    start(Request::Resolve(id))
367}
368/// Advertise a listener owned by the caller. Registered reports the actual instance name.
369/// Dropping the returned operation withdraws the advertisement; it does not close the listener.
370pub fn advertise(mut service: Advertisement) -> Result<Operation> {
371    service.service_type = validate_type(&service.service_type)?;
372    if service.name.is_empty()
373        || service.name.len() > 63
374        || service.name.contains('\0')
375        || service.port == 0
376    {
377        return Err(Error::InvalidInput("service name or port".into()));
378    }
379    service.txt.encode()?;
380    start(Request::Advertise(service))
381}
382fn start(request: Request) -> Result<Operation> {
383    if !capabilities().native {
384        return Err(Error::Unsupported);
385    }
386    let inbox = Arc::new(Inbox {
387        state: Mutex::new(Queue {
388            events: VecDeque::new(),
389            closed: false,
390            waker: None,
391        }),
392        ready: Condvar::new(),
393    });
394    let cancelled = Arc::new(AtomicBool::new(false));
395    let output = inbox.clone();
396    let stop = cancelled.clone();
397    let worker = thread::Builder::new()
398        .name("zeroconf".into())
399        .spawn(move || {
400            match &request {
401                Request::Advertise(a) => trace(format_args!(
402                    "advertise {:?} {} port={} interface={}",
403                    a.name, a.service_type, a.port, a.interface
404                )),
405                _ => trace(format_args!("starting {request:?}")),
406            }
407            let (sender, rx) = mpsc::sync_channel(MAX_SERVICES);
408            let overflow = Arc::new(AtomicBool::new(false));
409            let tx = Sink {
410                sender,
411                generation: 0,
412                overflow: overflow.clone(),
413            };
414            let browsing = matches!(&request, Request::Browse(..));
415            let all = matches!(&request,Request::Browse(t,_) if t==ENUMERATION_TYPE);
416            let interface = match &request {
417                Request::Browse(_, i) => *i,
418                Request::Resolve(id) => id.interface,
419                Request::Advertise(a) => a.interface,
420            };
421            let mut handles = Vec::new();
422            let mut types = BTreeSet::new();
423            let mut resolves: BTreeMap<ServiceId, (Box<dyn Native>, u64)> = BTreeMap::new();
424            let mut next_generation = 1;
425            emit(&tx, Event::Started);
426            match backend::start(request, tx.clone()) {
427                Ok(h) => handles.push(h),
428                Err(e) => {
429                    output.push(Event::Error(e));
430                    output.close();
431                    return;
432                }
433            }
434            while !stop.load(Ordering::Acquire) {
435                if overflow.load(Ordering::Acquire) {
436                    output.push(Event::Error(Error::Overflow));
437                    break;
438                }
439                if output.state.lock().unwrap().closed {
440                    break;
441                }
442                let (generation, event) = match rx.recv_timeout(Duration::from_millis(50)) {
443                    Ok(e) => e,
444                    Err(mpsc::RecvTimeoutError::Timeout) => continue,
445                    Err(_) => break,
446                };
447                match &event {
448                    Event::Resolved(s) => trace(format_args!(
449                        "generation {generation}: resolved {:?}, port={}, address_count={}, txt_count={}",
450                        s.id, s.port, s.addresses.len(), s.txt.0.len()
451                    )),
452                    _ => trace(format_args!("generation {generation}: {event:?}")),
453                }
454                // Ignore callbacks queued by a resolver that has already been removed.
455                if browsing && generation != 0 && !resolves.values().any(|(_, g)| *g == generation)
456                {
457                    continue;
458                }
459                match &event {
460                    Event::Resolved(service) if browsing && !resolves.contains_key(&service.id) => {
461                        continue;
462                    }
463                    Event::ServiceType(t) if all => {
464                        if validate_type(t).is_err() {
465                            continue;
466                        }
467                        if !types.insert(t.clone()) {
468                            continue;
469                        }
470                        if types.len() > MAX_TYPES {
471                            output.push(Event::Error(Error::Overflow));
472                            break;
473                        }
474                        match backend::start(Request::Browse(t.clone(), interface), tx.clone()) {
475                            Ok(h) => handles.push(h),
476                            Err(e) => output.push(Event::Error(e)),
477                        }
478                    }
479                    Event::Found(id) if !resolves.contains_key(id) => {
480                        if resolves.len() >= MAX_SERVICES {
481                            output.push(Event::Error(Error::Overflow));
482                            break;
483                        }
484                        let mut resolution_sink = tx.clone();
485                        resolution_sink.generation = next_generation;
486                        next_generation += 1;
487                        let generation = resolution_sink.generation;
488                        match backend::start(Request::Resolve(id.clone()), resolution_sink) {
489                            Ok(h) => {
490                                resolves.insert(id.clone(), (h, generation));
491                            }
492                            Err(e) => output.push(Event::Error(e)),
493                        }
494                    }
495                    Event::Removed(id) => {
496                        resolves.remove(id);
497                    }
498                    _ => {}
499                }
500                output.push(event);
501            }
502            drop(resolves);
503            drop(handles);
504            output.push(Event::Stopped);
505            output.close();
506        })
507        .map_err(|e| Error::Unavailable(e.to_string()))?;
508    Ok(Operation {
509        inbox,
510        cancelled,
511        worker: Some(worker),
512    })
513}
514
515#[cfg(any(target_os = "macos", target_os = "ios"))]
516#[path = "apple.rs"]
517mod backend;
518#[cfg(all(target_os = "linux", not(target_env = "ohos")))]
519#[path = "linux.rs"]
520mod backend;
521#[cfg(target_os = "windows")]
522#[path = "windows.rs"]
523mod backend;
524#[cfg(any(target_os = "android", target_env = "ohos"))]
525#[path = "mobile.rs"]
526mod backend;
527#[cfg(any(test, target_os = "windows"))]
528mod windows_names;
529#[cfg(not(any(
530    target_os = "macos",
531    target_os = "ios",
532    target_os = "android",
533    target_os = "linux",
534    target_os = "windows"
535)))]
536mod backend {
537    use super::*;
538    pub fn capabilities() -> Capabilities {
539        Capabilities {
540            native: false,
541            browse_all: false,
542            binary_txt: false,
543            key_only_txt: false,
544            interface_selection: false,
545        }
546    }
547    pub fn start(_: Request, _: Sink) -> Result<Box<dyn Native>> {
548        Err(Error::Unsupported)
549    }
550}
551
552#[cfg(test)]
553mod tests {
554    use super::*;
555    #[test]
556    fn txt_binary_and_empty_roundtrip() {
557        let t = TxtRecord(vec![
558            ("flag".into(), None),
559            ("empty".into(), Some(vec![])),
560            ("binary".into(), Some(vec![0, 255, b'=', 1])),
561        ]);
562        assert_eq!(TxtRecord::decode(&t.encode().unwrap()).unwrap(), t);
563    }
564    #[test]
565    fn txt_invalid_and_duplicate() {
566        assert!(TxtRecord::decode(&[4, 1]).is_err());
567        assert!(
568            TxtRecord(vec![("A".into(), None), ("a".into(), None)])
569                .encode()
570                .is_err()
571        );
572        assert_eq!(
573            TxtRecord::decode(&[3, b'a', b'=', b'1', 3, b'A', b'=', b'2'])
574                .unwrap()
575                .get("a"),
576            Some(&b"1"[..])
577        );
578    }
579    #[test]
580    fn types_are_validated() {
581        assert_eq!(
582            validate_type("_calibre._tcp.local.").unwrap(),
583            "_calibre._tcp"
584        );
585        for s in ["http", "_http._tcp.evil", "_http._sctp", "_x\0._tcp"] {
586            assert!(validate_type(s).is_err());
587        }
588        assert!(validate_type(&format!("_{}._tcp", "a".repeat(62))).is_ok());
589        assert!(validate_type(&format!("_{}._tcp", "a".repeat(63))).is_err());
590    }
591    #[test]
592    fn oversized_txt_rejected() {
593        assert!(
594            TxtRecord(vec![("x".into(), Some(vec![0; 255]))])
595                .encode()
596                .is_err()
597        );
598    }
599}
600
601#[cfg(any(test, target_os = "android", target_env = "ohos"))]
602mod wire;
603#[cfg(test)]
604mod lifecycle_tests {
605    use super::*;
606    fn inbox() -> Arc<Inbox> {
607        Arc::new(Inbox {
608            state: Mutex::new(Queue {
609                events: VecDeque::new(),
610                closed: false,
611                waker: None,
612            }),
613            ready: Condvar::new(),
614        })
615    }
616    #[test]
617    fn slow_consumer_is_told_to_resync() {
618        let q = inbox();
619        for _ in 0..=MAX_SERVICES {
620            q.push(Event::Started)
621        }
622        let s = q.state.lock().unwrap();
623        assert!(s.closed);
624        assert_eq!(s.events.len(), 1);
625        assert_eq!(s.events.front(), Some(&Event::Error(Error::Overflow)));
626    }
627    #[test]
628    fn full_native_channel_signals_overflow_without_blocking() {
629        let (sender, _) = mpsc::sync_channel(1);
630        let overflow = Arc::new(AtomicBool::new(false));
631        let sink = Sink {
632            sender,
633            generation: 0,
634            overflow: overflow.clone(),
635        };
636        emit(&sink, Event::Started);
637        emit(&sink, Event::Stopped); /* Disconnected is harmless after cancellation. */
638        assert!(!overflow.load(Ordering::Acquire));
639        let (sender, _rx) = mpsc::sync_channel(1);
640        let sink = Sink {
641            sender,
642            generation: 0,
643            overflow: overflow.clone(),
644        };
645        emit(&sink, Event::Started);
646        emit(&sink, Event::Stopped);
647        assert!(overflow.load(Ordering::Acquire));
648    }
649    #[test]
650    fn dropping_operation_closes_reader_and_cancels() {
651        let q = inbox();
652        let c = Arc::new(AtomicBool::new(false));
653        let op = Operation {
654            inbox: q.clone(),
655            cancelled: c.clone(),
656            worker: None,
657        };
658        drop(op);
659        assert!(c.load(Ordering::Acquire));
660        assert!(q.state.lock().unwrap().closed);
661    }
662    #[test]
663    fn invalid_identity_rejected_before_native_call() {
664        let id = ServiceId {
665            name: "bad\0name".into(),
666            service_type: "_http._tcp".into(),
667            domain: "local.".into(),
668            interface: 0,
669        };
670        assert!(resolve(id).is_err());
671    }
672}
673
674/// Whether this host requires an explicit runtime permission request before discovery.
675/// Currently Android 17+ apps targeting API 37+. Request ACCESS_LOCAL_NETWORK through
676/// day-part-permissions; Apple presents its own prompt on first native access.
677pub fn requires_explicit_permission() -> bool {
678    #[cfg(any(target_os = "android", target_env = "ohos"))]
679    {
680        backend::requires_explicit_permission()
681    }
682    #[cfg(not(any(target_os = "android", target_env = "ohos")))]
683    {
684        false
685    }
686}
687
688#[cfg(any(test, target_os = "android", target_env = "ohos"))]
689mod enumerate;
day_part_zeroconf/lib.rs

day_part_zeroconf/
lib.rs
