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                    // Implausibly-future emission times are ignored (deliberate
397                    // deviation from Python, see MAX_EMISSION_TIME_SKEW_SECS).
398                    if tunnel_path.hops <= existing.hops || existing.expires < now {
399                        let existing_timebase =
400                            trusted_timebase_from_random_blobs(&existing.random_blobs, now);
401                        let tunnel_timebase =
402                            trusted_timebase_from_random_blobs(&tunnel_path.random_blobs, now);
403                        tunnel_timebase >= existing_timebase
404                    } else {
405                        false
406                    }
407                }
408                None => now < tunnel_path.expires,
409            };
410
411            if should_restore {
412                let entry = PathEntry {
413                    timestamp: tunnel_path.timestamp,
414                    next_hop: tunnel_path.received_from,
415                    hops: tunnel_path.hops,
416                    expires: tunnel_path.expires,
417                    random_blobs: tunnel_path.random_blobs.clone(),
418                    receiving_interface: interface,
419                    packet_hash: tunnel_path.packet_hash,
420                    announce_raw: None,
421                };
422                self.upsert_path_destination(*dest_hash, entry, now);
423                log::trace!(target: crate::logging::PATHING_LOG_TARGET,
424                    "Restored tunnel path to {:02x?}: hops={} via={:02x?} interface={}",
425                    &dest_hash[..4],
426                    tunnel_path.hops,
427                    &tunnel_path.received_from[..4],
428                    interface.0,
429                );
430            } else {
431                log::trace!(target: crate::logging::PATHING_LOG_TARGET,
432                    "Did not restore tunnel path to {:02x?}: existing path is preferred or tunnel path expired",
433                    &dest_hash[..4],
434                );
435            }
436        }
437
438        actions.push(TransportAction::TunnelEstablished {
439            tunnel_id,
440            interface,
441        });
442
443        actions
444    }
445
446    /// Synthesize a tunnel on an interface.
447    ///
448    /// `identity`: the transport identity (must have private key for signing)
449    /// `interface_id`: which interface to send the synthesis on
450    /// `rng`: random number generator
451    ///
452    /// Returns TunnelSynthesize action to send the synthesis packet.
453    pub fn synthesize_tunnel(
454        &self,
455        identity: &rns_crypto::identity::Identity,
456        interface_id: InterfaceId,
457        rng: &mut dyn Rng,
458    ) -> Vec<TransportAction> {
459        let mut actions = Vec::new();
460
461        let interface_hash = if let Some(interface_hash) = self.interface_hashes.get(&interface_id)
462        {
463            *interface_hash
464        } else {
465            log::warn!(
466                "Cannot synthesize tunnel on {:?}: unknown interface or missing cached hash",
467                interface_id
468            );
469            return actions;
470        };
471
472        match tunnel::build_tunnel_synthesize_data(identity, &interface_hash, rng) {
473            Ok((data, _tunnel_id)) => {
474                let dest_hash = crate::destination::destination_hash(
475                    "rnstransport",
476                    &["tunnel", "synthesize"],
477                    None,
478                );
479                actions.push(TransportAction::TunnelSynthesize {
480                    interface: interface_id,
481                    data,
482                    dest_hash,
483                });
484            }
485            Err(e) => {
486                log::warn!("Cannot synthesize tunnel on {:?}: {}", interface_id, e);
487            }
488        }
489
490        actions
491    }
492
493    /// Void a tunnel's interface connection (tunnel disconnected).
494    pub fn void_tunnel_interface(&mut self, tunnel_id: &[u8; 32]) {
495        self.tunnel_table.void_tunnel_interface(tunnel_id);
496    }
497
498    /// Access the tunnel table for queries.
499    pub fn tunnel_table(&self) -> &TunnelTable {
500        &self.tunnel_table
501    }
502
503    // =========================================================================
504    // Packet filter
505    // =========================================================================
506
507    /// Check if any local client interfaces are registered.
508    pub(super) fn has_local_clients(&self) -> bool {
509        self.interfaces.values().any(|i| i.is_local_client)
510    }
511
512    pub(super) fn interface_is_local_client(&self, iface: InterfaceId) -> bool {
513        self.interfaces
514            .get(&iface)
515            .map(|i| i.is_local_client)
516            .unwrap_or(false)
517    }
518
519    /// Whether `iface` is a local client this engine serves as a shared
520    /// instance (Transport.py `from_local_client`). A shared client also
521    /// flags its interface to the daemon as a local client, but it has no
522    /// transport identity.
523    pub(super) fn serves_local_client(&self, iface: InterfaceId) -> bool {
524        self.config.identity_hash.is_some() && self.interface_is_local_client(iface)
525    }
526
527    pub(super) fn serves_local_clients(&self) -> bool {
528        self.config.identity_hash.is_some() && self.has_local_clients()
529    }
530
531    /// Packet filter: dedup + basic validity.
532    ///
533    /// Transport.py:1187-1238
534    #[cfg(test)]
535    pub(super) fn packet_filter(&self, packet: &RawPacket) -> bool {
536        self.packet_filter_at_hops(packet, packet.hops)
537    }
538
539    pub(super) fn packet_filter_at_hops(&self, packet: &RawPacket, hops: u8) -> bool {
540        // Filter packets for other transport instances
541        if packet.transport_id.is_some()
542            && packet.flags.packet_type != constants::PACKET_TYPE_ANNOUNCE
543        {
544            if let Some(ref identity_hash) = self.config.identity_hash {
545                if packet.transport_id.as_ref() != Some(identity_hash) {
546                    return false;
547                }
548            }
549        }
550
551        // Allow certain contexts unconditionally
552        match packet.context {
553            constants::CONTEXT_KEEPALIVE
554            | constants::CONTEXT_RESOURCE_REQ
555            | constants::CONTEXT_RESOURCE_PRF
556            | constants::CONTEXT_RESOURCE
557            | constants::CONTEXT_CACHE_REQUEST
558            | constants::CONTEXT_CHANNEL => return true,
559            _ => {}
560        }
561
562        // PLAIN/GROUP checks
563        if packet.flags.destination_type == constants::DESTINATION_PLAIN
564            || packet.flags.destination_type == constants::DESTINATION_GROUP
565        {
566            if packet.flags.packet_type != constants::PACKET_TYPE_ANNOUNCE {
567                return hops <= 1;
568            } else {
569                // PLAIN/GROUP ANNOUNCE is invalid
570                return false;
571            }
572        }
573
574        // Deduplication
575        if !self.packet_hashlist.is_duplicate(&packet.packet_hash) {
576            return true;
577        }
578
579        // Duplicate announce for SINGLE dest is allowed (path update)
580        if packet.flags.packet_type == constants::PACKET_TYPE_ANNOUNCE
581            && packet.flags.destination_type == constants::DESTINATION_SINGLE
582        {
583            return true;
584        }
585
586        false
587    }
588}
rns_core/transport/engine_state.rs

rns_core/transport/
engine_state.rs
