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: BTreeMap::new(),
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 ctx = self.prepare_inbound_packet(InboundFrame {
203            raw,
204            iface: receiving_interface,
205            now: 0.0,
206            rx: RxMetadata::default(),
207        })?;
208        if ctx.packet.flags.packet_type != constants::PACKET_TYPE_PROOF
209            || ctx.packet.context != constants::CONTEXT_LRPROOF
210        {
211            return None;
212        }
213        let link_id = ctx.packet.destination_hash;
214        let destination_hash =
215            self.link_rebalance_destination(&link_id, ctx.packet.hops, receiving_interface)?;
216        Some((link_id, destination_hash, ctx.packet.hops, ctx.packet.data))
217    }
218
219    /// Validate a mismatched-hop LRPROOF and update the relay link route and
220    /// destination path atomically enough for normal proof routing to resume.
221    ///
222    /// The engine is exclusively borrowed for the whole operation, so the
223    /// link and path mutations need neither separate path-table locks nor
224    /// repeated map lookups. A path can legitimately be absent; that does not
225    /// invalidate the authenticated link-route update.
226    pub fn rebalance_link_path_from_lrproof(
227        &mut self,
228        link_id: &[u8; 16],
229        packet_hops: u8,
230        receiving_interface: InterfaceId,
231        proof_data: &[u8],
232        destination_sig_pub_bytes: &[u8; 32],
233    ) -> bool {
234        if self
235            .link_rebalance_destination(link_id, packet_hops, receiving_interface)
236            .is_none()
237        {
238            return false;
239        }
240
241        let destination_sig_pub =
242            rns_crypto::ed25519::Ed25519PublicKey::from_bytes(destination_sig_pub_bytes);
243        if crate::link::handshake::validate_lrproof(
244            proof_data,
245            link_id,
246            &destination_sig_pub,
247            destination_sig_pub_bytes,
248        )
249        .is_err()
250        {
251            return false;
252        }
253
254        let destination_hash = match self.link_table.get_mut(link_id) {
255            Some(entry) if !entry.validated => {
256                entry.remaining_hops = packet_hops;
257                entry.destination_hash
258            }
259            _ => return false,
260        };
261        if let Some(paths) = self.path_table.get_mut(&destination_hash) {
262            paths.update_primary_hops(packet_hops);
263        }
264        true
265    }
266
267    /// Update the current path metric after a terminus validates an LRPROOF.
268    pub fn rebalance_destination_path_hops(
269        &mut self,
270        destination_hash: &[u8; 16],
271        packet_hops: u8,
272    ) -> bool {
273        self.path_table
274            .get_mut(destination_hash)
275            .is_some_and(|paths| paths.update_primary_hops(packet_hops))
276    }
277
278    // =========================================================================
279    // Blackhole management
280    // =========================================================================
281
282    /// Add an identity hash to the blackhole list.
283    ///
284    /// `identity_hash` is the 16-byte identity hash to blackhole. `now` is the
285    /// current Unix timestamp. If `duration_hours` is `Some` and greater than
286    /// zero, the entry expires after that many hours; otherwise it does not
287    /// expire. `reason` is optional descriptive text retained with the entry.
288    pub fn blackhole_identity(
289        &mut self,
290        identity_hash: [u8; 16],
291        now: f64,
292        duration_hours: Option<f64>,
293        reason: Option<String>,
294    ) {
295        let expires = match duration_hours {
296            Some(h) if h > 0.0 => now + h * 3600.0,
297            _ => 0.0, // never expires
298        };
299        self.blackholed_identities.insert(
300            identity_hash,
301            BlackholeEntry {
302                created: now,
303                expires,
304                reason,
305            },
306        );
307    }
308
309    /// Remove an identity hash from the blackhole list.
310    ///
311    /// Returns `true` if an entry was removed, or `false` if the identity was
312    /// not blackholed.
313    pub fn unblackhole_identity(&mut self, identity_hash: &[u8; 16]) -> bool {
314        self.blackholed_identities.remove(identity_hash).is_some()
315    }
316
317    /// Check if an identity hash is blackholed (and not expired).
318    pub fn is_blackholed(&self, identity_hash: &[u8; 16], now: f64) -> bool {
319        if let Some(entry) = self.blackholed_identities.get(identity_hash) {
320            if entry.expires == 0.0 || entry.expires > now {
321                return true;
322            }
323        }
324        false
325    }
326
327    /// Get all blackhole entries (for queries).
328    pub fn blackholed_entries(&self) -> impl Iterator<Item = (&[u8; 16], &BlackholeEntry)> {
329        self.blackholed_identities.iter()
330    }
331
332    /// Cull expired blackhole entries.
333    pub(super) fn cull_blackholed(&mut self, now: f64) {
334        self.blackholed_identities
335            .retain(|_, entry| entry.expires == 0.0 || entry.expires > now);
336    }
337
338    // =========================================================================
339    // Tunnel management
340    // =========================================================================
341
342    /// Handle a validated tunnel synthesis — create new or reattach.
343    ///
344    /// Returns actions for any restored paths.
345    pub fn handle_tunnel(
346        &mut self,
347        tunnel_id: [u8; 32],
348        interface: InterfaceId,
349        now: f64,
350    ) -> Vec<TransportAction> {
351        let mut actions = Vec::new();
352        let reattaching = self.tunnel_table.get(&tunnel_id).is_some();
353        if reattaching {
354            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
355                "Tunnel endpoint {:02x?} reappeared on interface {}; restoring paths",
356                &tunnel_id[..4],
357                interface.0,
358            );
359        } else {
360            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
361                "Tunnel endpoint {:02x?} established on interface {}",
362                &tunnel_id[..4],
363                interface.0,
364            );
365        }
366
367        // Set tunnel_id on the interface
368        if let Some(info) = self.interfaces.get_mut(&interface) {
369            info.tunnel_id = Some(tunnel_id);
370        }
371
372        let restored_paths = self.tunnel_table.handle_tunnel(
373            tunnel_id,
374            interface,
375            now,
376            self.config.destination_timeout_secs,
377        );
378
379        // Restore paths to path table if they're better than existing
380        for (dest_hash, tunnel_path) in &restored_paths {
381            let should_restore = match self.path_table.get(dest_hash).and_then(|ps| ps.primary()) {
382                Some(existing) => {
383                    // Restore if fewer/equal hops or existing expired, but never
384                    // overwrite a path learned from a more recent announce.
385                    if tunnel_path.hops <= existing.hops || existing.expires < now {
386                        let existing_timebase = timebase_from_random_blobs(&existing.random_blobs);
387                        let tunnel_timebase = timebase_from_random_blobs(&tunnel_path.random_blobs);
388                        tunnel_timebase >= existing_timebase
389                    } else {
390                        false
391                    }
392                }
393                None => now < tunnel_path.expires,
394            };
395
396            if should_restore {
397                let entry = PathEntry {
398                    timestamp: tunnel_path.timestamp,
399                    next_hop: tunnel_path.received_from,
400                    hops: tunnel_path.hops,
401                    expires: tunnel_path.expires,
402                    random_blobs: tunnel_path.random_blobs.clone(),
403                    receiving_interface: interface,
404                    packet_hash: tunnel_path.packet_hash,
405                    announce_raw: None,
406                };
407                self.upsert_path_destination(*dest_hash, entry, now);
408                log::trace!(target: crate::logging::PATHING_LOG_TARGET,
409                    "Restored tunnel path to {:02x?}: hops={} via={:02x?} interface={}",
410                    &dest_hash[..4],
411                    tunnel_path.hops,
412                    &tunnel_path.received_from[..4],
413                    interface.0,
414                );
415            } else {
416                log::trace!(target: crate::logging::PATHING_LOG_TARGET,
417                    "Did not restore tunnel path to {:02x?}: existing path is preferred or tunnel path expired",
418                    &dest_hash[..4],
419                );
420            }
421        }
422
423        actions.push(TransportAction::TunnelEstablished {
424            tunnel_id,
425            interface,
426        });
427
428        actions
429    }
430
431    /// Synthesize a tunnel on an interface.
432    ///
433    /// `identity`: the transport identity (must have private key for signing)
434    /// `interface_id`: which interface to send the synthesis on
435    /// `rng`: random number generator
436    ///
437    /// Returns TunnelSynthesize action to send the synthesis packet.
438    pub fn synthesize_tunnel(
439        &self,
440        identity: &rns_crypto::identity::Identity,
441        interface_id: InterfaceId,
442        rng: &mut dyn Rng,
443    ) -> Vec<TransportAction> {
444        let mut actions = Vec::new();
445
446        let interface_hash = if let Some(interface_hash) = self.interface_hashes.get(&interface_id)
447        {
448            *interface_hash
449        } else {
450            log::warn!(
451                "Cannot synthesize tunnel on {:?}: unknown interface or missing cached hash",
452                interface_id
453            );
454            return actions;
455        };
456
457        match tunnel::build_tunnel_synthesize_data(identity, &interface_hash, rng) {
458            Ok((data, _tunnel_id)) => {
459                let dest_hash = crate::destination::destination_hash(
460                    "rnstransport",
461                    &["tunnel", "synthesize"],
462                    None,
463                );
464                actions.push(TransportAction::TunnelSynthesize {
465                    interface: interface_id,
466                    data,
467                    dest_hash,
468                });
469            }
470            Err(e) => {
471                log::warn!("Cannot synthesize tunnel on {:?}: {}", interface_id, e);
472            }
473        }
474
475        actions
476    }
477
478    /// Void a tunnel's interface connection (tunnel disconnected).
479    pub fn void_tunnel_interface(&mut self, tunnel_id: &[u8; 32]) {
480        self.tunnel_table.void_tunnel_interface(tunnel_id);
481    }
482
483    /// Access the tunnel table for queries.
484    pub fn tunnel_table(&self) -> &TunnelTable {
485        &self.tunnel_table
486    }
487
488    // =========================================================================
489    // Packet filter
490    // =========================================================================
491
492    /// Check if any local client interfaces are registered.
493    pub(super) fn has_local_clients(&self) -> bool {
494        self.interfaces.values().any(|i| i.is_local_client)
495    }
496
497    pub(super) fn interface_is_local_client(&self, iface: InterfaceId) -> bool {
498        self.interfaces
499            .get(&iface)
500            .map(|i| i.is_local_client)
501            .unwrap_or(false)
502    }
503
504    /// Packet filter: dedup + basic validity.
505    ///
506    /// Transport.py:1187-1238
507    pub(super) fn packet_filter(&self, packet: &RawPacket) -> bool {
508        // Filter packets for other transport instances
509        if packet.transport_id.is_some()
510            && packet.flags.packet_type != constants::PACKET_TYPE_ANNOUNCE
511        {
512            if let Some(ref identity_hash) = self.config.identity_hash {
513                if packet.transport_id.as_ref() != Some(identity_hash) {
514                    return false;
515                }
516            }
517        }
518
519        // Allow certain contexts unconditionally
520        match packet.context {
521            constants::CONTEXT_KEEPALIVE
522            | constants::CONTEXT_RESOURCE_REQ
523            | constants::CONTEXT_RESOURCE_PRF
524            | constants::CONTEXT_RESOURCE
525            | constants::CONTEXT_CACHE_REQUEST
526            | constants::CONTEXT_CHANNEL => return true,
527            _ => {}
528        }
529
530        // PLAIN/GROUP checks
531        if packet.flags.destination_type == constants::DESTINATION_PLAIN
532            || packet.flags.destination_type == constants::DESTINATION_GROUP
533        {
534            if packet.flags.packet_type != constants::PACKET_TYPE_ANNOUNCE {
535                return packet.hops <= 1;
536            } else {
537                // PLAIN/GROUP ANNOUNCE is invalid
538                return false;
539            }
540        }
541
542        // Deduplication
543        if !self.packet_hashlist.is_duplicate(&packet.packet_hash) {
544            return true;
545        }
546
547        // Duplicate announce for SINGLE dest is allowed (path update)
548        if packet.flags.packet_type == constants::PACKET_TYPE_ANNOUNCE
549            && packet.flags.destination_type == constants::DESTINATION_SINGLE
550        {
551            return true;
552        }
553
554        false
555    }
556}
rns_core/transport/engine_state.rs

rns_core/transport/
engine_state.rs
