alpine_protocol_sdk/discovery/
client.rs

1use std::{
2    collections::HashMap,
3    fmt, io,
4    net::{IpAddr, SocketAddr, UdpSocket},
5    sync::{Mutex, OnceLock},
6    time::Duration,
7};
8
9use alpine::attestation::{verify_device_identity_attestation, AttesterRegistry};
10use alpine::messages::{DiscoveryReply, DiscoveryRequest};
11use rand::{rngs::OsRng, RngCore};
12use serde_cbor;
13use tracing::{info, warn};
14use uuid::Uuid;
15
16use socket2::{Domain, Protocol, Socket, Type};
17
18use crate::error::AlpineSdkError;
19use crate::phase::{current_phase, Phase};
20use crate::quarantine::mark_quarantine;
21
22const DEFAULT_MULTICAST_IPV4: &str = "239.255.255.250:19455";
23const DEFAULT_MULTICAST_IPV6: &str = "[ff12::1]:19455";
24const DEFAULT_BROADCAST_IPV4: &str = "255.255.255.255:19455";
25const DISCOVERY_SAFE_MTU: usize = 1200;
26
27static IDENTITY_SEEN: OnceLock<Mutex<HashMap<IpAddr, Vec<u8>>>> = OnceLock::new();
28
29fn identity_map() -> &'static Mutex<HashMap<IpAddr, Vec<u8>>> {
30    IDENTITY_SEEN.get_or_init(|| Mutex::new(HashMap::new()))
31}
32
33/// Options used to configure the blocking discovery helper.
34pub struct DiscoveryClientOptions {
35    pub remote_addr: SocketAddr,
36    pub local_addr: SocketAddr,
37    pub timeout: Duration,
38    pub prefer_multicast: bool,
39    pub allow_broadcast: bool,
40    pub interface: Option<String>,
41    pub attester_registry: Option<AttesterRegistry>,
42}
43
44impl DiscoveryClientOptions {
45    /// Creates options with the provided remote socket and a default timeout.
46    pub fn new(remote_addr: SocketAddr, local_addr: SocketAddr, timeout: Duration) -> Self {
47        Self {
48            remote_addr,
49            local_addr,
50            timeout,
51            prefer_multicast: false,
52            allow_broadcast: true,
53            interface: None,
54            attester_registry: None,
55        }
56    }
57
58    pub fn disable_multicast(mut self) -> Self {
59        self.prefer_multicast = false;
60        self
61    }
62
63    pub fn disable_broadcast(mut self) -> Self {
64        self.allow_broadcast = false;
65        self
66    }
67
68    pub fn with_attester_registry(mut self, registry: AttesterRegistry) -> Self {
69        self.attester_registry = Some(registry);
70        self
71    }
72}
73
74/// Errors that can happen while sending or receiving discovery payloads.
75#[derive(Debug)]
76pub enum DiscoveryError {
77    Io(io::Error),
78    Decode(serde_cbor::Error),
79    Timeout,
80    PermissionDenied,
81    MulticastUnavailable,
82    BroadcastBlocked,
83}
84
85impl DiscoveryError {
86    pub fn label(&self) -> &'static str {
87        match self {
88            DiscoveryError::Io(_) => "io error",
89            DiscoveryError::Decode(_) => "cbor decode error",
90            DiscoveryError::Timeout => "discovery timed out",
91            DiscoveryError::PermissionDenied => "discovery channel permission denied",
92            DiscoveryError::MulticastUnavailable => "multicast discovery unavailable",
93            DiscoveryError::BroadcastBlocked => "broadcast discovery blocked",
94        }
95    }
96
97    pub fn hint(&self) -> Option<&'static str> {
98        match self {
99            DiscoveryError::PermissionDenied => Some("udp send/recv denied by OS policy"),
100            DiscoveryError::BroadcastBlocked => Some("broadcast disabled on this network"),
101            DiscoveryError::MulticastUnavailable => Some("multicast not permitted on this network"),
102            DiscoveryError::Timeout => Some("no discovery replies before timeout"),
103            _ => None,
104        }
105    }
106}
107
108impl fmt::Display for DiscoveryError {
109    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
110        match self {
111            DiscoveryError::Io(err) => write!(f, "io error: {}", err),
112            DiscoveryError::Decode(err) => write!(f, "cbors serialization error: {}", err),
113            DiscoveryError::Timeout => write!(f, "discovery timed out"),
114            DiscoveryError::PermissionDenied => {
115                write!(f, "discovery channel permission denied")
116            }
117            DiscoveryError::MulticastUnavailable => {
118                write!(f, "multicast discovery unavailable")
119            }
120            DiscoveryError::BroadcastBlocked => write!(f, "broadcast discovery blocked"),
121        }
122    }
123}
124
125impl std::error::Error for DiscoveryError {}
126
127impl From<io::Error> for DiscoveryError {
128    fn from(err: io::Error) -> Self {
129        match err.kind() {
130            io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => DiscoveryError::Timeout,
131            _ => DiscoveryError::Io(err),
132        }
133    }
134}
135
136impl From<serde_cbor::Error> for DiscoveryError {
137    fn from(err: serde_cbor::Error) -> Self {
138        DiscoveryError::Decode(err)
139    }
140}
141
142#[derive(Debug, Clone, Copy, PartialEq, Eq)]
143pub enum DiscoveryTargetKind {
144    Multicast,
145    Broadcast,
146    UnicastConfigured,
147    UnicastFallback,
148}
149
150impl DiscoveryTargetKind {
151    pub fn as_str(&self) -> &'static str {
152        match self {
153            DiscoveryTargetKind::Multicast => "multicast",
154            DiscoveryTargetKind::Broadcast => "broadcast",
155            DiscoveryTargetKind::UnicastConfigured => "unicast-configured",
156            DiscoveryTargetKind::UnicastFallback => "unicast",
157        }
158    }
159}
160
161#[derive(Debug, Clone)]
162pub struct DiscoveryErrorDetails {
163    pub label: &'static str,
164    pub hint: Option<&'static str>,
165    pub kind: Option<io::ErrorKind>,
166    pub message: Option<String>,
167}
168
169#[derive(Debug, Clone)]
170pub struct DiscoveryAttempt {
171    pub target: SocketAddr,
172    pub mode: DiscoveryTargetKind,
173    pub local_bind: SocketAddr,
174    pub payload_len: usize,
175    pub bytes_sent: Option<usize>,
176    pub error: Option<DiscoveryErrorDetails>,
177}
178
179impl DiscoveryAttempt {
180    pub fn debug_string(&self) -> String {
181        let error = self.error.as_ref().map(|err| err.label).unwrap_or("ok");
182        format!(
183            "target={} mode={} local_bind={} payload_len={} bytes_sent={} error={}",
184            self.target,
185            self.mode.as_str(),
186            self.local_bind,
187            self.payload_len,
188            self.bytes_sent
189                .map(|val| val.to_string())
190                .unwrap_or_else(|| "-".to_string()),
191            error
192        )
193    }
194}
195
196#[derive(Debug, Clone)]
197pub struct DiscoveryDiagnostics {
198    pub attempts: Vec<DiscoveryAttempt>,
199    pub recv_error: Option<DiscoveryErrorDetails>,
200}
201
202impl DiscoveryDiagnostics {
203    fn new() -> Self {
204        Self {
205            attempts: Vec::new(),
206            recv_error: None,
207        }
208    }
209}
210
211#[derive(Debug)]
212pub struct DiscoveryResult {
213    pub outcome: Option<DiscoveryOutcome>,
214    pub diagnostics: DiscoveryDiagnostics,
215    pub error: Option<DiscoveryError>,
216}
217
218impl DiscoveryResult {
219    pub fn into_result(self) -> Result<DiscoveryOutcome, DiscoveryError> {
220        match (self.outcome, self.error) {
221            (Some(outcome), _) => Ok(outcome),
222            (None, Some(err)) => Err(err),
223            (None, None) => Err(DiscoveryError::Timeout),
224        }
225    }
226}
227
228/// The outcome of a discovery request.
229#[must_use]
230#[derive(Debug, Clone)]
231pub struct DiscoveryOutcome {
232    pub reply: DiscoveryReply,
233    pub peer: SocketAddr,
234    pub client_nonce: Vec<u8>,
235    pub local_addr: SocketAddr,
236    pub device_identity_pubkey: Option<Vec<u8>>,
237    pub device_identity_trusted: bool,
238    pub device_identity_attestation_error: Option<String>,
239    pub interface: Option<String>,
240    pub run_id: String,
241}
242
243#[derive(Debug, Clone)]
244pub struct TrustDecision {
245    pub state: DeviceTrustState,
246    pub reason: String,
247}
248
249impl TrustDecision {
250    pub fn explain(&self) -> &str {
251        &self.reason
252    }
253}
254
255#[derive(Debug, Clone)]
256pub struct ProtocolNegotiationReport {
257    pub client_version: String,
258    pub device_version: String,
259    pub compatible: bool,
260    pub note: String,
261}
262
263#[must_use]
264#[derive(Debug, Clone)]
265pub struct TrustedDiscoveryOutcome {
266    pub outcome: DiscoveryOutcome,
267}
268
269#[derive(Debug, Clone, PartialEq, Eq)]
270pub enum DeviceTrustState {
271    Trusted,
272    UntrustedNoAttestation,
273    UntrustedNoRegistry,
274    UntrustedInvalid(String),
275}
276
277impl DeviceTrustState {
278    pub fn as_str(&self) -> &'static str {
279        match self {
280            DeviceTrustState::Trusted => "trusted",
281            DeviceTrustState::UntrustedNoAttestation => "untrusted-no-attestation",
282            DeviceTrustState::UntrustedNoRegistry => "untrusted-no-registry",
283            DeviceTrustState::UntrustedInvalid(_) => "untrusted-invalid",
284        }
285    }
286}
287
288#[derive(Debug, Clone, Default)]
289pub struct DeviceSelectionPolicy {
290    pub allow_device_ids: Vec<String>,
291    pub deny_device_ids: Vec<String>,
292    pub allow_models: Vec<String>,
293    pub deny_models: Vec<String>,
294    pub allow_manufacturers: Vec<String>,
295    pub deny_manufacturers: Vec<String>,
296}
297
298#[derive(Debug, Clone)]
299pub struct DeviceSelectionDecision {
300    pub allowed: bool,
301    pub reason: String,
302}
303
304impl DeviceSelectionPolicy {
305    pub fn evaluate(&self, outcome: &DiscoveryOutcome) -> DeviceSelectionDecision {
306        let reply = &outcome.reply;
307        if self.deny_device_ids.iter().any(|id| id == &reply.device_id) {
308            return DeviceSelectionDecision {
309                allowed: false,
310                reason: format!("device_id {} denied", reply.device_id),
311            };
312        }
313        if self
314            .deny_models
315            .iter()
316            .any(|model| model == &reply.model_id)
317        {
318            return DeviceSelectionDecision {
319                allowed: false,
320                reason: format!("model_id {} denied", reply.model_id),
321            };
322        }
323        if self
324            .deny_manufacturers
325            .iter()
326            .any(|mfg| mfg == &reply.manufacturer_id)
327        {
328            return DeviceSelectionDecision {
329                allowed: false,
330                reason: format!("manufacturer_id {} denied", reply.manufacturer_id),
331            };
332        }
333        if !self.allow_device_ids.is_empty()
334            && !self
335                .allow_device_ids
336                .iter()
337                .any(|id| id == &reply.device_id)
338        {
339            return DeviceSelectionDecision {
340                allowed: false,
341                reason: format!("device_id {} not in allowlist", reply.device_id),
342            };
343        }
344        if !self.allow_models.is_empty()
345            && !self
346                .allow_models
347                .iter()
348                .any(|model| model == &reply.model_id)
349        {
350            return DeviceSelectionDecision {
351                allowed: false,
352                reason: format!("model_id {} not in allowlist", reply.model_id),
353            };
354        }
355        if !self.allow_manufacturers.is_empty()
356            && !self
357                .allow_manufacturers
358                .iter()
359                .any(|mfg| mfg == &reply.manufacturer_id)
360        {
361            return DeviceSelectionDecision {
362                allowed: false,
363                reason: format!("manufacturer_id {} not in allowlist", reply.manufacturer_id),
364            };
365        }
366        DeviceSelectionDecision {
367            allowed: true,
368            reason: "policy allowed".to_string(),
369        }
370    }
371
372    pub fn enforce(&self, outcome: &DiscoveryOutcome) -> Result<(), AlpineSdkError> {
373        let decision = self.evaluate(outcome);
374        if decision.allowed {
375            Ok(())
376        } else {
377            Err(AlpineSdkError::SelectionDenied(decision.reason))
378        }
379    }
380}
381
382impl DiscoveryOutcome {
383    pub fn trust_state(&self) -> DeviceTrustState {
384        if self.device_identity_trusted {
385            return DeviceTrustState::Trusted;
386        }
387        match self.device_identity_attestation_error.as_deref() {
388            Some("device identity attestation missing") => DeviceTrustState::UntrustedNoAttestation,
389            Some("attester registry not configured") => DeviceTrustState::UntrustedNoRegistry,
390            Some(other) => DeviceTrustState::UntrustedInvalid(other.to_string()),
391            None => DeviceTrustState::UntrustedInvalid("attestation failed".to_string()),
392        }
393    }
394
395    pub fn require_trusted(self) -> Result<TrustedDiscoveryOutcome, AlpineSdkError> {
396        match self.trust_state() {
397            DeviceTrustState::Trusted => Ok(TrustedDiscoveryOutcome { outcome: self }),
398            DeviceTrustState::UntrustedNoAttestation => Err(AlpineSdkError::UntrustedDevice(
399                "device identity attestation missing".into(),
400            )),
401            DeviceTrustState::UntrustedNoRegistry => Err(AlpineSdkError::UntrustedDevice(
402                "attester registry not configured".into(),
403            )),
404            DeviceTrustState::UntrustedInvalid(reason) => {
405                Err(AlpineSdkError::UntrustedDevice(reason))
406            }
407        }
408    }
409
410    pub fn trust_decision(&self) -> TrustDecision {
411        let state = self.trust_state();
412        let reason = match &state {
413            DeviceTrustState::Trusted => "attestation verified".to_string(),
414            DeviceTrustState::UntrustedNoAttestation => {
415                "device identity attestation missing".to_string()
416            }
417            DeviceTrustState::UntrustedNoRegistry => "attester registry not configured".to_string(),
418            DeviceTrustState::UntrustedInvalid(reason) => reason.clone(),
419        };
420        TrustDecision { state, reason }
421    }
422
423    pub fn protocol_report(&self) -> ProtocolNegotiationReport {
424        let client_version = alpine::messages::ALPINE_VERSION.to_string();
425        let device_version = self.reply.alpine_version.clone();
426        let compatible = client_version == device_version;
427        let note = if compatible {
428            "protocol versions match".to_string()
429        } else {
430            format!(
431                "client {} vs device {}; upgrade/downgrade may be required",
432                client_version, device_version
433            )
434        };
435        ProtocolNegotiationReport {
436            client_version,
437            device_version,
438            compatible,
439            note,
440        }
441    }
442
443    pub fn require_capabilities(
444        &self,
445        required: &alpine::messages::CapabilitySet,
446    ) -> Result<(), AlpineSdkError> {
447        validate_capability_subset(required, &self.reply.capabilities)
448    }
449}
450/// Stateless discovery helper that wraps the protocol request/response models.
451pub struct DiscoveryClient {
452    socket: UdpSocket,
453    remote_addr: SocketAddr,
454    prefer_multicast: bool,
455    allow_broadcast: bool,
456    ipv6: bool,
457    interface: Option<String>,
458    attester_registry: Option<AttesterRegistry>,
459}
460
461impl DiscoveryClient {
462    /// Creates a client that will send discovery packets to `remote_addr`.
463    pub fn new(options: DiscoveryClientOptions) -> Result<Self, DiscoveryError> {
464        let domain = if options.remote_addr.is_ipv4() {
465            Domain::IPV4
466        } else {
467            Domain::IPV6
468        };
469        let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
470        if options.allow_broadcast && options.remote_addr.is_ipv4() {
471            socket.set_broadcast(true)?;
472        }
473        socket.bind(&options.local_addr.into())?;
474
475        let socket: UdpSocket = socket.into();
476        socket.set_read_timeout(Some(options.timeout))?;
477        // TTL of 4 keeps traffic on the local segment but makes sure it leaves the host.
478        if options.prefer_multicast || options.remote_addr.ip().is_multicast() {
479            let _ = socket.set_multicast_ttl_v4(4);
480        }
481        let local_addr = socket.local_addr().unwrap_or(options.local_addr);
482        info!(
483            "[ALPINE][DISCOVERY][SOCKET] discovery socket created local_addr={} remote_addr={} prefer_multicast={}",
484            local_addr,
485            options.remote_addr,
486            options.prefer_multicast
487        );
488        Ok(Self {
489            socket,
490            remote_addr: options.remote_addr,
491            prefer_multicast: options.prefer_multicast,
492            allow_broadcast: options.allow_broadcast,
493            ipv6: options.remote_addr.is_ipv6() || options.local_addr.is_ipv6(),
494            interface: options.interface.clone(),
495            attester_registry: options.attester_registry.clone(),
496        })
497    }
498
499    /// Sends a discovery payload with the requested capability names and waits for a reply.
500    pub fn discover(&self, requested: &[String]) -> Result<DiscoveryOutcome, DiscoveryError> {
501        self.discover_with_report(requested).into_result()
502    }
503
504    /// Sends discovery and returns diagnostics for each attempted target.
505    pub fn discover_with_report(&self, requested: &[String]) -> DiscoveryResult {
506        let phase = current_phase();
507        if phase == Phase::Handshake {
508            warn!(
509                "[ALPINE][BUG] discovery attempted during handshake phase (no sends expected); local_addr={}",
510                self.socket
511                    .local_addr()
512                    .unwrap_or_else(|_| SocketAddr::from(([0, 0, 0, 0], 0)))
513            );
514        }
515
516        let mut diagnostics = DiscoveryDiagnostics::new();
517
518        let mut nonce = vec![0u8; 32];
519        OsRng.fill_bytes(&mut nonce);
520        let request = DiscoveryRequest::new(requested.to_vec(), nonce.clone());
521        let payload = match serde_cbor::to_vec(&request) {
522            Ok(payload) => payload,
523            Err(err) => {
524                return DiscoveryResult {
525                    outcome: None,
526                    diagnostics,
527                    error: Some(DiscoveryError::from(err)),
528                };
529            }
530        };
531        let payload_len = payload.len();
532        debug_assert!(
533            payload_len > 8,
534            "discovery payload unexpectedly small; framing may have drifted"
535        );
536        if payload_len > DISCOVERY_SAFE_MTU {
537            warn!(
538                "[ALPINE][DISCOVERY][WARN] payload_len={} exceeds safe MTU {}; fragmentation likely",
539                payload_len, DISCOVERY_SAFE_MTU
540            );
541        }
542
543        let mut send_error: Option<DiscoveryError> = None;
544        let mut sent = false;
545        for target in self.discovery_targets() {
546            let mode = classify_target(target, self.remote_addr);
547            let local_bind_addr = self
548                .socket
549                .local_addr()
550                .unwrap_or_else(|_| SocketAddr::from(([0, 0, 0, 0], 0)));
551            info!(
552                "[ALPINE][DISCOVERY][TX][attempt] target={} mode={} local_bind={} payload_len={}",
553                target,
554                mode.as_str(),
555                local_bind_addr,
556                payload.len()
557            );
558            match self.socket.send_to(&payload, target) {
559                Ok(bytes) => {
560                    info!(
561                        "[ALPINE][DISCOVERY][TX][result] target={} mode={} local_bind={} bytes_sent={}",
562                        target,
563                        mode.as_str(),
564                        local_bind_addr,
565                        bytes
566                    );
567                    diagnostics.attempts.push(DiscoveryAttempt {
568                        target,
569                        mode,
570                        local_bind: local_bind_addr,
571                        payload_len,
572                        bytes_sent: Some(bytes),
573                        error: None,
574                    });
575                    sent = true;
576                }
577                Err(err) => {
578                    let kind = err.kind();
579                    let message = err.to_string();
580                    let mapped = self.map_send_error(err, target);
581                    warn!(
582                        "[ALPINE][DISCOVERY][TX][result] target={} mode={} local_bind={} phase={} error={}",
583                        target,
584                        mode.as_str(),
585                        local_bind_addr,
586                        phase.label(),
587                        mapped
588                    );
589                    diagnostics.attempts.push(DiscoveryAttempt {
590                        target,
591                        mode,
592                        local_bind: local_bind_addr,
593                        payload_len,
594                        bytes_sent: None,
595                        error: Some(DiscoveryErrorDetails {
596                            label: mapped.label(),
597                            hint: mapped.hint(),
598                            kind: Some(kind),
599                            message: Some(message),
600                        }),
601                    });
602                    send_error = Some(mapped);
603                    if !self.should_continue_after_error(&kind) {
604                        return DiscoveryResult {
605                            outcome: None,
606                            diagnostics,
607                            error: send_error,
608                        };
609                    }
610                }
611            }
612        }
613
614        if !sent {
615            let error = send_error.unwrap_or(DiscoveryError::PermissionDenied);
616            warn!(
617                "[ALPINE][DISCOVERY][WARN] no discovery payloads sent; possible UDP egress block"
618            );
619            return DiscoveryResult {
620                outcome: None,
621                diagnostics,
622                error: Some(error),
623            };
624        }
625
626        let timeout_ms = self
627            .socket
628            .read_timeout()
629            .ok()
630            .flatten()
631            .map(|d| d.as_millis())
632            .unwrap_or_default();
633        let local_port = self
634            .socket
635            .local_addr()
636            .map(|addr| addr.port())
637            .unwrap_or_default();
638        info!(
639            "[ALPINE][DISCOVERY] awaiting reply timeout_ms={} local_port={} remote_hint={}",
640            timeout_ms, local_port, self.remote_addr
641        );
642
643        let mut buf = vec![0u8; 2048];
644        let (len, peer) = match self.socket.recv_from(&mut buf) {
645            Ok(res) => res,
646            Err(err) => {
647                let kind = err.kind();
648                let message = err.to_string();
649                let mapped = self.map_recv_error(err);
650                if matches!(mapped, DiscoveryError::Timeout) && sent {
651                    warn!(
652                        "[ALPINE][DISCOVERY][WARN] discovery sends ok but no replies; inbound UDP may be blocked"
653                    );
654                }
655                diagnostics.recv_error = Some(DiscoveryErrorDetails {
656                    label: mapped.label(),
657                    hint: mapped.hint(),
658                    kind: Some(kind),
659                    message: Some(message),
660                });
661                return DiscoveryResult {
662                    outcome: None,
663                    diagnostics,
664                    error: Some(mapped),
665                };
666            }
667        };
668        let reply: DiscoveryReply = match serde_cbor::from_slice(&buf[..len]) {
669            Ok(reply) => reply,
670            Err(err) => {
671                return DiscoveryResult {
672                    outcome: None,
673                    diagnostics,
674                    error: Some(DiscoveryError::from(err)),
675                };
676            }
677        };
678        let local_addr = match self.socket.local_addr() {
679            Ok(local_addr) => local_addr,
680            Err(err) => {
681                return DiscoveryResult {
682                    outcome: None,
683                    diagnostics,
684                    error: Some(DiscoveryError::from(err)),
685                };
686            }
687        };
688        info!(
689            "[ALPINE][DISCOVERY][RX] reply received peer={} local_addr={} iface={:?} bytes={}",
690            peer, local_addr, self.interface, len
691        );
692        let (device_identity_trusted, device_identity_attestation_error) =
693            self.verify_attestation(&reply);
694        if !reply.device_identity_pubkey.is_empty() {
695            let peer_ip = peer.ip();
696            if let Some(reason) = record_identity_change(peer_ip, &reply.device_identity_pubkey) {
697                warn!(
698                    "[ALPINE][DISCOVERY][WARN] identity change detected peer={} reason={}",
699                    peer, reason
700                );
701                let _ = mark_quarantine(peer_ip, reason);
702            }
703        }
704        let outcome = DiscoveryOutcome {
705            reply: reply.clone(),
706            peer,
707            client_nonce: nonce,
708            local_addr,
709            device_identity_pubkey: if reply.device_identity_pubkey.is_empty() {
710                None
711            } else {
712                Some(reply.device_identity_pubkey.clone())
713            },
714            device_identity_trusted,
715            device_identity_attestation_error,
716            interface: self.interface.clone(),
717            run_id: Uuid::new_v4().to_string(),
718        };
719        DiscoveryResult {
720            outcome: Some(outcome),
721            diagnostics,
722            error: None,
723        }
724    }
725
726    fn discovery_targets(&self) -> Vec<SocketAddr> {
727        let mut targets = Vec::new();
728
729        if self.prefer_multicast {
730            if !self.ipv6 {
731                if let Ok(addr) = DEFAULT_MULTICAST_IPV4.parse() {
732                    push_if_unique(&mut targets, addr);
733                }
734            }
735            if self.ipv6 {
736                if let Ok(addr) = DEFAULT_MULTICAST_IPV6.parse() {
737                    push_if_unique(&mut targets, addr);
738                }
739            }
740        }
741
742        push_if_unique(&mut targets, self.remote_addr);
743
744        if self.allow_broadcast && self.remote_addr.ip().is_ipv4() && !self.ipv6 {
745            if let Ok(addr) = DEFAULT_BROADCAST_IPV4.parse() {
746                push_if_unique(&mut targets, addr);
747            }
748        }
749
750        targets
751    }
752
753    fn map_send_error(&self, err: io::Error, target: SocketAddr) -> DiscoveryError {
754        match err.kind() {
755            io::ErrorKind::PermissionDenied => {
756                if target.ip().is_multicast() {
757                    DiscoveryError::MulticastUnavailable
758                } else if is_broadcast_addr(target.ip()) {
759                    DiscoveryError::BroadcastBlocked
760                } else {
761                    DiscoveryError::PermissionDenied
762                }
763            }
764            io::ErrorKind::ConnectionReset | io::ErrorKind::WouldBlock => {
765                DiscoveryError::PermissionDenied
766            }
767            _ => DiscoveryError::Io(err),
768        }
769    }
770
771    fn map_recv_error(&self, err: io::Error) -> DiscoveryError {
772        match err.kind() {
773            io::ErrorKind::TimedOut => DiscoveryError::Timeout,
774            io::ErrorKind::PermissionDenied | io::ErrorKind::ConnectionReset => {
775                DiscoveryError::PermissionDenied
776            }
777            io::ErrorKind::WouldBlock => DiscoveryError::Timeout,
778            _ => DiscoveryError::Io(err),
779        }
780    }
781
782    fn should_continue_after_error(&self, kind: &io::ErrorKind) -> bool {
783        matches!(
784            kind,
785            io::ErrorKind::PermissionDenied
786                | io::ErrorKind::WouldBlock
787                | io::ErrorKind::ConnectionReset
788        )
789    }
790
791    fn verify_attestation(&self, reply: &DiscoveryReply) -> (bool, Option<String>) {
792        if reply.device_identity_attestation.is_empty() {
793            return (false, Some("device identity attestation missing".into()));
794        }
795        let Some(registry) = &self.attester_registry else {
796            return (false, Some("attester registry not configured".into()));
797        };
798        match verify_device_identity_attestation(reply, registry, std::time::SystemTime::now()) {
799            Ok(_) => (true, None),
800            Err(err) => {
801                warn!(
802                    "[ALPINE][DISCOVERY][TRUST] attestation verification failed device_id={} err={}",
803                    reply.device_id,
804                    err
805                );
806                (false, Some(err.to_string()))
807            }
808        }
809    }
810}
811
812fn record_identity_change(peer: IpAddr, pubkey: &[u8]) -> Option<String> {
813    let mut guard = identity_map()
814        .lock()
815        .unwrap_or_else(|poisoned| poisoned.into_inner());
816    if let Some(existing) = guard.get(&peer) {
817        if existing != pubkey {
818            guard.insert(peer, pubkey.to_vec());
819            return Some("device identity pubkey changed for peer".to_string());
820        }
821        return None;
822    }
823    guard.insert(peer, pubkey.to_vec());
824    None
825}
826
827impl Drop for DiscoveryClient {
828    fn drop(&mut self) {
829        if let Ok(local) = self.socket.local_addr() {
830            info!(
831                "[ALPINE][DISCOVERY][SOCKET] discovery socket dropped local_addr={} remote_addr={}",
832                local, self.remote_addr
833            );
834        } else {
835            info!(
836                "[ALPINE][DISCOVERY][SOCKET] discovery socket dropped remote_addr={}",
837                self.remote_addr
838            );
839        }
840    }
841}
842
843fn is_broadcast_addr(ip: IpAddr) -> bool {
844    matches!(ip, IpAddr::V4(addr) if addr.is_broadcast())
845}
846
847fn push_if_unique(targets: &mut Vec<SocketAddr>, candidate: SocketAddr) {
848    if !targets.contains(&candidate) {
849        targets.push(candidate);
850    }
851}
852
853fn classify_target(target: SocketAddr, configured: SocketAddr) -> DiscoveryTargetKind {
854    if target.ip().is_multicast() {
855        DiscoveryTargetKind::Multicast
856    } else if is_broadcast_addr(target.ip()) {
857        DiscoveryTargetKind::Broadcast
858    } else if target == configured {
859        DiscoveryTargetKind::UnicastConfigured
860    } else {
861        DiscoveryTargetKind::UnicastFallback
862    }
863}
864
865fn validate_capability_subset(
866    required: &alpine::messages::CapabilitySet,
867    device: &alpine::messages::CapabilitySet,
868) -> Result<(), AlpineSdkError> {
869    for format in required.channel_formats.iter() {
870        if !device.channel_formats.contains(format) {
871            return Err(AlpineSdkError::InvalidCapabilities(format!(
872                "channel format {:?} not supported by device",
873                format
874            )));
875        }
876    }
877    if required.max_channels > device.max_channels {
878        return Err(AlpineSdkError::InvalidCapabilities(format!(
879            "max channels {} exceeds device max {}",
880            required.max_channels, device.max_channels
881        )));
882    }
883    if required.grouping_supported && !device.grouping_supported {
884        return Err(AlpineSdkError::InvalidCapabilities(
885            "grouping not supported by device".into(),
886        ));
887    }
888    if required.streaming_supported && !device.streaming_supported {
889        return Err(AlpineSdkError::InvalidCapabilities(
890            "streaming not supported by device".into(),
891        ));
892    }
893    if required.encryption_supported && !device.encryption_supported {
894        return Err(AlpineSdkError::InvalidCapabilities(
895            "encryption not supported by device".into(),
896        ));
897    }
898    if let Some(extensions) = required.vendor_extensions.as_ref() {
899        if let Some(device_extensions) = device.vendor_extensions.as_ref() {
900            for key in extensions.keys() {
901                if !device_extensions.contains_key(key) {
902                    return Err(AlpineSdkError::InvalidCapabilities(format!(
903                        "vendor extension {} not supported by device",
904                        key
905                    )));
906                }
907            }
908        } else {
909            return Err(AlpineSdkError::InvalidCapabilities(
910                "vendor extensions not supported by device".into(),
911            ));
912        }
913    }
914    Ok(())
915}
alpine_protocol_sdk/discovery/client.rs

alpine_protocol_sdk/discovery/
client.rs
