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

rns_core/transport/
engine_state.rs
