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

rns_core/transport/
engine_state.rs
