rns_core/transport/
engine_state.rs

1use super::*;
2
3impl TransportEngine {
4    pub fn new(config: TransportConfig) -> Self {
5        let packet_hashlist_max_entries = config.packet_hashlist_max_entries;
6        let packet_hashlist_allocation = config.packet_hashlist_allocation;
7        let sig_cache_max = if config.announce_sig_cache_enabled {
8            config.announce_sig_cache_max_entries
9        } else {
10            0
11        };
12        let sig_cache_ttl = config.announce_sig_cache_ttl_secs;
13        let announce_queue_max_interfaces = config.announce_queue_max_interfaces;
14        TransportEngine {
15            config,
16            path_table: BTreeMap::new(),
17            announce_table: BTreeMap::new(),
18            reverse_table: tables::ReverseTable::default(),
19            link_table: BTreeMap::new(),
20            held_announces: BTreeMap::new(),
21            packet_hashlist: PacketHashlist::with_allocation(
22                packet_hashlist_max_entries,
23                packet_hashlist_allocation,
24            ),
25            announce_sig_cache: AnnounceSignatureCache::new(sig_cache_max, sig_cache_ttl),
26            rate_limiter: AnnounceRateLimiter::new(),
27            path_states: BTreeMap::new(),
28            interfaces: BTreeMap::new(),
29            interface_hashes: BTreeMap::new(),
30            local_destinations: BTreeMap::new(),
31            blackholed_identities: BTreeMap::new(),
32            announce_queues: AnnounceQueues::new(announce_queue_max_interfaces),
33            ingress_control: IngressControl::new(),
34            tunnel_table: TunnelTable::new(),
35            discovery_pr_tags: VecDeque::new(),
36            discovery_pr_tag_set: BTreeSet::new(),
37            path_requests: BTreeMap::new(),
38            discovery_path_requests: BTreeMap::new(),
39            discovery_path_request_deadlines: BTreeMap::new(),
40            pending_local_path_requests: BTreeMap::new(),
41            path_destination_cap_evict_count: 0,
42            announces_last_checked: 0.0,
43            tables_last_culled: 0.0,
44        }
45    }
46
47    // =========================================================================
48    // Interface management
49    // =========================================================================
50
51    pub fn register_interface(&mut self, info: InterfaceInfo) {
52        self.interface_hashes
53            .insert(info.id, hash::full_hash(info.name.as_bytes()));
54        self.interfaces.insert(info.id, info);
55    }
56
57    pub fn deregister_interface(&mut self, id: InterfaceId) {
58        self.interfaces.remove(&id);
59        self.interface_hashes.remove(&id);
60        self.drop_paths_for_interface(id);
61        self.drop_reverse_for_interface(id);
62        self.drop_links_for_interface(id);
63        self.announce_queues.remove_interface(id);
64        self.ingress_control.remove_interface(&id);
65        self.pending_local_path_requests
66            .retain(|_, (interface, _)| *interface != id);
67    }
68
69    // =========================================================================
70    // Destination management
71    // =========================================================================
72
73    pub fn register_destination(&mut self, dest_hash: [u8; 16], dest_type: u8) {
74        self.local_destinations.insert(dest_hash, dest_type);
75    }
76
77    pub fn deregister_destination(&mut self, dest_hash: &[u8; 16]) {
78        self.local_destinations.remove(dest_hash);
79    }
80
81    // =========================================================================
82    // Path queries
83    // =========================================================================
84
85    pub fn has_path(&self, dest_hash: &[u8; 16]) -> bool {
86        self.path_table
87            .get(dest_hash)
88            .is_some_and(|ps| !ps.is_empty())
89    }
90
91    pub fn hops_to(&self, dest_hash: &[u8; 16]) -> Option<u8> {
92        self.path_table
93            .get(dest_hash)
94            .and_then(|ps| ps.primary())
95            .map(|e| e.hops)
96    }
97
98    pub fn next_hop(&self, dest_hash: &[u8; 16]) -> Option<[u8; 16]> {
99        self.path_table
100            .get(dest_hash)
101            .and_then(|ps| ps.primary())
102            .map(|e| e.next_hop)
103    }
104
105    pub fn next_hop_interface(&self, dest_hash: &[u8; 16]) -> Option<InterfaceId> {
106        self.path_table
107            .get(dest_hash)
108            .and_then(|ps| ps.primary())
109            .map(|e| e.receiving_interface)
110    }
111
112    // =========================================================================
113    // Path state
114    // =========================================================================
115
116    /// Mark a path as unresponsive.
117    ///
118    /// If `receiving_interface` is provided and points to a MODE_BOUNDARY interface,
119    /// the marking is skipped — boundary interfaces must not poison path tables.
120    /// (Python Transport.py: mark_path_unknown/unresponsive boundary exemption)
121    pub fn mark_path_unresponsive(
122        &mut self,
123        dest_hash: &[u8; 16],
124        receiving_interface: Option<InterfaceId>,
125    ) {
126        if let Some(iface_id) = receiving_interface {
127            if let Some(info) = self.interfaces.get(&iface_id) {
128                if info.mode == constants::MODE_BOUNDARY {
129                    return;
130                }
131            }
132        }
133
134        // Failover: if we have alternative paths, promote the next one
135        if let Some(ps) = self.path_table.get_mut(dest_hash) {
136            if ps.len() > 1 {
137                ps.failover(false); // demote old primary to back
138                                    // Clear unresponsive state since we promoted a fresh primary
139                self.path_states.remove(dest_hash);
140                return;
141            }
142        }
143
144        self.path_states
145            .insert(*dest_hash, constants::STATE_UNRESPONSIVE);
146    }
147
148    pub fn mark_path_responsive(&mut self, dest_hash: &[u8; 16]) {
149        self.path_states
150            .insert(*dest_hash, constants::STATE_RESPONSIVE);
151    }
152
153    pub fn path_is_unresponsive(&self, dest_hash: &[u8; 16]) -> bool {
154        self.path_states.get(dest_hash) == Some(&constants::STATE_UNRESPONSIVE)
155    }
156
157    pub fn expire_path(&mut self, dest_hash: &[u8; 16]) {
158        if let Some(ps) = self.path_table.get_mut(dest_hash) {
159            ps.expire_all();
160        }
161    }
162
163    // =========================================================================
164    // Link table
165    // =========================================================================
166
167    pub fn register_link(&mut self, link_id: [u8; 16], entry: LinkEntry) {
168        self.link_table.insert(link_id, entry);
169    }
170
171    pub fn validate_link(&mut self, link_id: &[u8; 16]) {
172        if let Some(entry) = self.link_table.get_mut(link_id) {
173            entry.validated = true;
174        }
175    }
176
177    pub fn remove_link(&mut self, link_id: &[u8; 16]) {
178        self.link_table.remove(link_id);
179    }
180
181    /// Return the destination whose unvalidated link route can be rebalanced
182    /// by a mismatched-hop LRPROOF received from its recorded next hop.
183    pub fn link_rebalance_destination(
184        &self,
185        link_id: &[u8; 16],
186        packet_hops: u8,
187        receiving_interface: InterfaceId,
188    ) -> Option<[u8; 16]> {
189        let entry = self.link_table.get(link_id)?;
190        (self.config.transport_enabled
191            && !entry.validated
192            && packet_hops != entry.remaining_hops
193            && receiving_interface == entry.next_hop_interface)
194            .then_some(entry.destination_hash)
195    }
196
197    /// Parse and filter an inbound frame before offering it for LRPROOF path
198    /// rebalancing. The returned hops are the post-ingress metric used by the
199    /// normal transport pipeline.
200    pub fn inbound_lrproof_rebalance_candidate(
201        &self,
202        raw: &[u8],
203        receiving_interface: InterfaceId,
204    ) -> Option<super::LrproofRebalanceCandidate> {
205        let decoded = DecodedPacket::unpack(raw).ok()?;
206        self.decoded_lrproof_rebalance_candidate(&decoded, receiving_interface)
207    }
208
209    pub fn decoded_lrproof_rebalance_candidate(
210        &self,
211        decoded: &DecodedPacket,
212        receiving_interface: InterfaceId,
213    ) -> Option<super::LrproofRebalanceCandidate> {
214        let packet = &decoded.packet;
215        let (hops, _) = self.filtered_inbound_hops(packet, receiving_interface)?;
216        if packet.flags.packet_type != constants::PACKET_TYPE_PROOF
217            || packet.context != constants::CONTEXT_LRPROOF
218        {
219            return None;
220        }
221        let link_id = packet.destination_hash;
222        let destination_hash =
223            self.link_rebalance_destination(&link_id, hops, receiving_interface)?;
224        Some((link_id, destination_hash, hops, packet.data().to_vec()))
225    }
226
227    /// Validate a mismatched-hop LRPROOF and update the relay link route and
228    /// destination path atomically enough for normal proof routing to resume.
229    ///
230    /// The engine is exclusively borrowed for the whole operation, so the
231    /// link and path mutations need neither separate path-table locks nor
232    /// repeated map lookups. A path can legitimately be absent; that does not
233    /// invalidate the authenticated link-route update.
234    pub fn rebalance_link_path_from_lrproof(
235        &mut self,
236        link_id: &[u8; 16],
237        packet_hops: u8,
238        receiving_interface: InterfaceId,
239        proof_data: &[u8],
240        destination_sig_pub_bytes: &[u8; 32],
241    ) -> bool {
242        if self
243            .link_rebalance_destination(link_id, packet_hops, receiving_interface)
244            .is_none()
245        {
246            return false;
247        }
248
249        let Ok(destination_sig_pub) =
250            rns_crypto::ed25519::Ed25519PublicKey::from_bytes(destination_sig_pub_bytes)
251        else {
252            return false;
253        };
254        if crate::link::handshake::validate_lrproof(
255            proof_data,
256            link_id,
257            &destination_sig_pub,
258            destination_sig_pub_bytes,
259        )
260        .is_err()
261        {
262            return false;
263        }
264
265        let destination_hash = match self.link_table.get_mut(link_id) {
266            Some(entry) if !entry.validated => {
267                entry.remaining_hops = packet_hops;
268                entry.destination_hash
269            }
270            _ => return false,
271        };
272        if let Some(paths) = self.path_table.get_mut(&destination_hash) {
273            paths.update_primary_hops(packet_hops);
274        }
275        true
276    }
277
278    /// Update the current path metric after a terminus validates an LRPROOF.
279    pub fn rebalance_destination_path_hops(
280        &mut self,
281        destination_hash: &[u8; 16],
282        packet_hops: u8,
283    ) -> bool {
284        self.path_table
285            .get_mut(destination_hash)
286            .is_some_and(|paths| paths.update_primary_hops(packet_hops))
287    }
288
289    // =========================================================================
290    // Blackhole management
291    // =========================================================================
292
293    /// Add an identity hash to the blackhole list.
294    ///
295    /// `identity_hash` is the 16-byte identity hash to blackhole. `now` is the
296    /// current Unix timestamp. If `duration_hours` is `Some` and greater than
297    /// zero, the entry expires after that many hours; otherwise it does not
298    /// expire. `reason` is optional descriptive text retained with the entry.
299    pub fn blackhole_identity(
300        &mut self,
301        identity_hash: [u8; 16],
302        now: f64,
303        duration_hours: Option<f64>,
304        reason: Option<String>,
305    ) {
306        let expires = match duration_hours {
307            Some(h) if h > 0.0 => now + h * 3600.0,
308            _ => 0.0, // never expires
309        };
310        self.blackholed_identities.insert(
311            identity_hash,
312            BlackholeEntry {
313                created: now,
314                expires,
315                reason,
316            },
317        );
318    }
319
320    /// Remove an identity hash from the blackhole list.
321    ///
322    /// Returns `true` if an entry was removed, or `false` if the identity was
323    /// not blackholed.
324    pub fn unblackhole_identity(&mut self, identity_hash: &[u8; 16]) -> bool {
325        self.blackholed_identities.remove(identity_hash).is_some()
326    }
327
328    /// Check if an identity hash is blackholed (and not expired).
329    pub fn is_blackholed(&self, identity_hash: &[u8; 16], now: f64) -> bool {
330        if let Some(entry) = self.blackholed_identities.get(identity_hash) {
331            if entry.expires == 0.0 || entry.expires > now {
332                return true;
333            }
334        }
335        false
336    }
337
338    /// Get all blackhole entries (for queries).
339    pub fn blackholed_entries(&self) -> impl Iterator<Item = (&[u8; 16], &BlackholeEntry)> {
340        self.blackholed_identities.iter()
341    }
342
343    /// Cull expired blackhole entries.
344    pub(super) fn cull_blackholed(&mut self, now: f64) {
345        self.blackholed_identities
346            .retain(|_, entry| entry.expires == 0.0 || entry.expires > now);
347    }
348
349    // =========================================================================
350    // Tunnel management
351    // =========================================================================
352
353    /// Handle a validated tunnel synthesis — create new or reattach.
354    ///
355    /// Returns actions for any restored paths.
356    pub fn handle_tunnel(
357        &mut self,
358        tunnel_id: [u8; 32],
359        interface: InterfaceId,
360        now: f64,
361    ) -> Vec<TransportAction> {
362        let mut actions = Vec::new();
363        let reattaching = self.tunnel_table.get(&tunnel_id).is_some();
364        if reattaching {
365            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
366                "Tunnel endpoint {:02x?} reappeared on interface {}; restoring paths",
367                &tunnel_id[..4],
368                interface.0,
369            );
370        } else {
371            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
372                "Tunnel endpoint {:02x?} established on interface {}",
373                &tunnel_id[..4],
374                interface.0,
375            );
376        }
377
378        // Set tunnel_id on the interface
379        if let Some(info) = self.interfaces.get_mut(&interface) {
380            info.tunnel_id = Some(tunnel_id);
381        }
382
383        let restored_paths = self.tunnel_table.handle_tunnel(
384            tunnel_id,
385            interface,
386            now,
387            self.config.destination_timeout_secs,
388        );
389
390        // Restore paths to path table if they're better than existing
391        for (dest_hash, tunnel_path) in &restored_paths {
392            let should_restore = match self.path_table.get(dest_hash).and_then(|ps| ps.primary()) {
393                Some(existing) => {
394                    // Restore if fewer/equal hops or existing expired, but never
395                    // overwrite a path learned from a more recent announce.
396                    if tunnel_path.hops <= existing.hops || existing.expires < now {
397                        let existing_timebase = timebase_from_random_blobs(&existing.random_blobs);
398                        let tunnel_timebase = timebase_from_random_blobs(&tunnel_path.random_blobs);
399                        tunnel_timebase >= existing_timebase
400                    } else {
401                        false
402                    }
403                }
404                None => now < tunnel_path.expires,
405            };
406
407            if should_restore {
408                let entry = PathEntry {
409                    timestamp: tunnel_path.timestamp,
410                    next_hop: tunnel_path.received_from,
411                    hops: tunnel_path.hops,
412                    expires: tunnel_path.expires,
413                    random_blobs: tunnel_path.random_blobs.clone(),
414                    receiving_interface: interface,
415                    packet_hash: tunnel_path.packet_hash,
416                    announce_raw: None,
417                };
418                self.upsert_path_destination(*dest_hash, entry, now);
419                log::trace!(target: crate::logging::PATHING_LOG_TARGET,
420                    "Restored tunnel path to {:02x?}: hops={} via={:02x?} interface={}",
421                    &dest_hash[..4],
422                    tunnel_path.hops,
423                    &tunnel_path.received_from[..4],
424                    interface.0,
425                );
426            } else {
427                log::trace!(target: crate::logging::PATHING_LOG_TARGET,
428                    "Did not restore tunnel path to {:02x?}: existing path is preferred or tunnel path expired",
429                    &dest_hash[..4],
430                );
431            }
432        }
433
434        actions.push(TransportAction::TunnelEstablished {
435            tunnel_id,
436            interface,
437        });
438
439        actions
440    }
441
442    /// Synthesize a tunnel on an interface.
443    ///
444    /// `identity`: the transport identity (must have private key for signing)
445    /// `interface_id`: which interface to send the synthesis on
446    /// `rng`: random number generator
447    ///
448    /// Returns TunnelSynthesize action to send the synthesis packet.
449    pub fn synthesize_tunnel(
450        &self,
451        identity: &rns_crypto::identity::Identity,
452        interface_id: InterfaceId,
453        rng: &mut dyn Rng,
454    ) -> Vec<TransportAction> {
455        let mut actions = Vec::new();
456
457        let interface_hash = if let Some(interface_hash) = self.interface_hashes.get(&interface_id)
458        {
459            *interface_hash
460        } else {
461            log::warn!(
462                "Cannot synthesize tunnel on {:?}: unknown interface or missing cached hash",
463                interface_id
464            );
465            return actions;
466        };
467
468        match tunnel::build_tunnel_synthesize_data(identity, &interface_hash, rng) {
469            Ok((data, _tunnel_id)) => {
470                let dest_hash = crate::destination::destination_hash(
471                    "rnstransport",
472                    &["tunnel", "synthesize"],
473                    None,
474                );
475                actions.push(TransportAction::TunnelSynthesize {
476                    interface: interface_id,
477                    data,
478                    dest_hash,
479                });
480            }
481            Err(e) => {
482                log::warn!("Cannot synthesize tunnel on {:?}: {}", interface_id, e);
483            }
484        }
485
486        actions
487    }
488
489    /// Void a tunnel's interface connection (tunnel disconnected).
490    pub fn void_tunnel_interface(&mut self, tunnel_id: &[u8; 32]) {
491        self.tunnel_table.void_tunnel_interface(tunnel_id);
492    }
493
494    /// Access the tunnel table for queries.
495    pub fn tunnel_table(&self) -> &TunnelTable {
496        &self.tunnel_table
497    }
498
499    // =========================================================================
500    // Packet filter
501    // =========================================================================
502
503    /// Check if any local client interfaces are registered.
504    pub(super) fn has_local_clients(&self) -> bool {
505        self.interfaces.values().any(|i| i.is_local_client)
506    }
507
508    pub(super) fn interface_is_local_client(&self, iface: InterfaceId) -> bool {
509        self.interfaces
510            .get(&iface)
511            .map(|i| i.is_local_client)
512            .unwrap_or(false)
513    }
514
515    /// Whether `iface` is a local client this engine serves as a shared
516    /// instance (Transport.py `from_local_client`). A shared client also
517    /// flags its interface to the daemon as a local client, but it has no
518    /// transport identity.
519    pub(super) fn serves_local_client(&self, iface: InterfaceId) -> bool {
520        self.config.identity_hash.is_some() && self.interface_is_local_client(iface)
521    }
522
523    pub(super) fn serves_local_clients(&self) -> bool {
524        self.config.identity_hash.is_some() && self.has_local_clients()
525    }
526
527    /// Packet filter: dedup + basic validity.
528    ///
529    /// Transport.py:1187-1238
530    #[cfg(test)]
531    pub(super) fn packet_filter(&self, packet: &RawPacket) -> bool {
532        self.packet_filter_at_hops(packet, packet.hops)
533    }
534
535    pub(super) fn packet_filter_at_hops(&self, packet: &RawPacket, hops: u8) -> bool {
536        // Filter packets for other transport instances
537        if packet.transport_id.is_some()
538            && packet.flags.packet_type != constants::PACKET_TYPE_ANNOUNCE
539        {
540            if let Some(ref identity_hash) = self.config.identity_hash {
541                if packet.transport_id.as_ref() != Some(identity_hash) {
542                    return false;
543                }
544            }
545        }
546
547        // Allow certain contexts unconditionally
548        match packet.context {
549            constants::CONTEXT_KEEPALIVE
550            | constants::CONTEXT_RESOURCE_REQ
551            | constants::CONTEXT_RESOURCE_PRF
552            | constants::CONTEXT_RESOURCE
553            | constants::CONTEXT_CACHE_REQUEST
554            | constants::CONTEXT_CHANNEL => return true,
555            _ => {}
556        }
557
558        // PLAIN/GROUP checks
559        if packet.flags.destination_type == constants::DESTINATION_PLAIN
560            || packet.flags.destination_type == constants::DESTINATION_GROUP
561        {
562            if packet.flags.packet_type != constants::PACKET_TYPE_ANNOUNCE {
563                return hops <= 1;
564            } else {
565                // PLAIN/GROUP ANNOUNCE is invalid
566                return false;
567            }
568        }
569
570        // Deduplication
571        if !self.packet_hashlist.is_duplicate(&packet.packet_hash) {
572            return true;
573        }
574
575        // Duplicate announce for SINGLE dest is allowed (path update)
576        if packet.flags.packet_type == constants::PACKET_TYPE_ANNOUNCE
577            && packet.flags.destination_type == constants::DESTINATION_SINGLE
578        {
579            return true;
580        }
581
582        false
583    }
584}
rns_core/transport/engine_state.rs

rns_core/transport/
engine_state.rs
