rns_core/transport/
inbound_engine.rs

1use super::*;
2
3pub(super) fn extra_link_proof_timeout(interface: Option<&InterfaceInfo>) -> f64 {
4    interface
5        .and_then(|interface| interface.bitrate)
6        .filter(|bitrate| *bitrate > 0)
7        .map_or(0.0, |bitrate| {
8            (constants::MTU as f64 * 8.0) / bitrate as f64
9        })
10}
11
12impl TransportEngine {
13    fn record_invalid_announce(
14        &self,
15        announce: &AnnounceData,
16        interface: InterfaceId,
17        now: f64,
18        actions: &mut Vec<TransportAction>,
19    ) {
20        let identity_hash = crate::hash::truncated_hash(&announce.public_key);
21        if !self.is_blackholed(&identity_hash, now) {
22            actions.push(TransportAction::ProtocolViolation { interface });
23        }
24    }
25
26    /// Return whether an inbound frame passes parsing, hop, and packet filtering.
27    /// This does not mutate deduplication or routing state.
28    pub fn accepts_inbound_frame(&self, frame: InboundFrame<'_>) -> bool {
29        self.prepare_inbound_packet(frame).is_some()
30    }
31
32    /// Process an inbound raw packet from a network interface.
33    ///
34    /// Returns a list of actions for the caller to execute.
35    pub fn handle_inbound(
36        &mut self,
37        frame: InboundFrame<'_>,
38        rng: &mut dyn Rng,
39    ) -> Vec<TransportAction> {
40        self.handle_inbound_with_announce_queue(frame, rng, None)
41    }
42
43    pub fn handle_inbound_with_announce_queue(
44        &mut self,
45        frame: InboundFrame<'_>,
46        rng: &mut dyn Rng,
47        announce_queue: Option<&mut AnnounceVerifyQueue>,
48    ) -> Vec<TransportAction> {
49        let Some(ctx) = self.prepare_inbound_packet(frame) else {
50            return Vec::new();
51        };
52        let mut actions = Vec::new();
53
54        self.remember_inbound_packet_hash(&ctx.packet);
55        self.bridge_plain_broadcast(&ctx, &mut actions);
56        self.handle_transport_forwarding(&ctx, &mut actions);
57        self.handle_link_table_routing(&ctx, &mut actions);
58        self.handle_inbound_announce(&ctx, rng, announce_queue, &mut actions);
59
60        if ctx.packet.flags.packet_type == constants::PACKET_TYPE_PROOF {
61            self.process_inbound_proof(&ctx, &mut actions);
62        }
63
64        self.handle_inbound_local_delivery(&ctx, &mut actions);
65        actions
66    }
67
68    pub(super) fn prepare_inbound_packet(
69        &self,
70        frame: InboundFrame<'_>,
71    ) -> Option<InboundPacketCtx> {
72        let mut packet = RawPacket::unpack(frame.raw).ok()?;
73        let from_local_client = self
74            .interfaces
75            .get(&frame.iface)
76            .map(|i| i.is_local_client)
77            .unwrap_or(false);
78        packet.hops = packet.hops.checked_add(1)?;
79        packet.rssi = frame.rx.rssi;
80        packet.snr = frame.rx.snr;
81        if from_local_client {
82            packet.hops = packet.hops.saturating_sub(1);
83        }
84        if !self.packet_filter(&packet) {
85            return None;
86        }
87        let retain_original_raw = packet.flags.packet_type == constants::PACKET_TYPE_ANNOUNCE;
88        Some(InboundPacketCtx {
89            packet,
90            original_raw: if retain_original_raw {
91                Some(frame.raw.to_vec())
92            } else {
93                None
94            },
95            iface: frame.iface,
96            now: frame.now,
97            from_local_client,
98        })
99    }
100
101    fn remember_inbound_packet_hash(&mut self, packet: &RawPacket) {
102        let remember_hash = !(self.link_table.contains_key(&packet.destination_hash)
103            || (packet.flags.packet_type == constants::PACKET_TYPE_PROOF
104                && packet.context == constants::CONTEXT_LRPROOF));
105        if remember_hash {
106            self.packet_hashlist.add(packet.packet_hash);
107        }
108    }
109
110    fn bridge_plain_broadcast(&self, ctx: &InboundPacketCtx, actions: &mut Vec<TransportAction>) {
111        if ctx.packet.flags.destination_type != constants::DESTINATION_PLAIN
112            || ctx.packet.flags.transport_type != constants::TRANSPORT_BROADCAST
113            || !self.has_local_clients()
114        {
115            return;
116        }
117
118        if ctx.from_local_client {
119            actions.push(TransportAction::ForwardPlainBroadcast {
120                raw: PacketBytes::from(ctx.packet.raw.clone()),
121                to_local: false,
122                exclude: Some(ctx.iface),
123            });
124        } else {
125            actions.push(TransportAction::ForwardPlainBroadcast {
126                raw: PacketBytes::from(ctx.packet.raw.clone()),
127                to_local: true,
128                exclude: None,
129            });
130        }
131    }
132
133    fn handle_transport_forwarding(
134        &mut self,
135        ctx: &InboundPacketCtx,
136        actions: &mut Vec<TransportAction>,
137    ) {
138        if !(self.config.transport_enabled || self.config.identity_hash.is_some()) {
139            return;
140        }
141        if ctx.packet.transport_id.is_none()
142            || ctx.packet.flags.packet_type == constants::PACKET_TYPE_ANNOUNCE
143        {
144            if ctx.packet.flags.packet_type == constants::PACKET_TYPE_DATA {
145                log::debug!(
146                    "TransportForward: DATA dest={:02x}{:02x}{:02x}{:02x}.. not transport-addressed header={} iface={}",
147                    ctx.packet.destination_hash[0],
148                    ctx.packet.destination_hash[1],
149                    ctx.packet.destination_hash[2],
150                    ctx.packet.destination_hash[3],
151                    ctx.packet.flags.header_type,
152                    ctx.iface.0
153                );
154            }
155            return;
156        }
157
158        let Some(identity_hash) = self.config.identity_hash else {
159            return;
160        };
161        if ctx.packet.transport_id != Some(identity_hash) {
162            if ctx.packet.flags.packet_type == constants::PACKET_TYPE_DATA {
163                log::debug!(
164                    "TransportForward: DATA dest={:02x}{:02x}{:02x}{:02x}.. transport mismatch got={:02x?} own={:02x?} iface={}",
165                    ctx.packet.destination_hash[0],
166                    ctx.packet.destination_hash[1],
167                    ctx.packet.destination_hash[2],
168                    ctx.packet.destination_hash[3],
169                    ctx.packet.transport_id.as_ref().map(|id| &id[..4]),
170                    &identity_hash[..4],
171                    ctx.iface.0
172                );
173            }
174            return;
175        }
176
177        let Some(path_entry) = self
178            .path_table
179            .get(&ctx.packet.destination_hash)
180            .and_then(|ps| ps.primary())
181        else {
182            if ctx.packet.flags.packet_type == constants::PACKET_TYPE_DATA {
183                log::debug!(
184                    "TransportForward: DATA dest={:02x}{:02x}{:02x}{:02x}.. addressed to us but no path iface={}",
185                    ctx.packet.destination_hash[0],
186                    ctx.packet.destination_hash[1],
187                    ctx.packet.destination_hash[2],
188                    ctx.packet.destination_hash[3],
189                    ctx.iface.0
190                );
191            }
192            return;
193        };
194
195        let next_hop = path_entry.next_hop;
196        let remaining_hops = path_entry.hops;
197        let outbound_interface = path_entry.receiving_interface;
198        let outbound_is_local_client = self
199            .interfaces
200            .get(&outbound_interface)
201            .map(|info| info.is_local_client)
202            .unwrap_or(false);
203        let forwarded_remaining_hops = if outbound_is_local_client {
204            0
205        } else {
206            remaining_hops
207        };
208        if ctx.packet.flags.packet_type == constants::PACKET_TYPE_DATA {
209            log::debug!(
210                "TransportForward: DATA dest={:02x}{:02x}{:02x}{:02x}.. remaining_hops={} out_iface={} next_hop={:02x?}",
211                ctx.packet.destination_hash[0],
212                ctx.packet.destination_hash[1],
213                ctx.packet.destination_hash[2],
214                ctx.packet.destination_hash[3],
215                remaining_hops,
216                outbound_interface.0,
217                &next_hop[..4]
218            );
219        }
220        let mut new_raw = forward_transport_packet(
221            &ctx.packet,
222            next_hop,
223            forwarded_remaining_hops,
224            outbound_interface,
225        );
226        if self.config.local_hops_delta != 0
227            && ctx.from_local_client
228            && !outbound_is_local_client
229            && ctx.packet.hops == 0
230            && ctx.packet.flags.destination_type != constants::DESTINATION_PLAIN
231            && ctx.packet.flags.destination_type != constants::DESTINATION_GROUP
232            && new_raw.len() > 1
233        {
234            new_raw[1] = self.config.local_hops_delta;
235        }
236
237        if ctx.packet.flags.packet_type == constants::PACKET_TYPE_LINKREQUEST {
238            let extra_proof_timeout =
239                extra_link_proof_timeout(self.interfaces.get(&outbound_interface));
240            let proof_timeout = ctx.now
241                + constants::LINK_ESTABLISHMENT_TIMEOUT_PER_HOP * (remaining_hops.max(1) as f64)
242                + extra_proof_timeout;
243            let (link_id, link_entry) = create_link_entry(
244                &ctx.packet,
245                next_hop,
246                outbound_interface,
247                remaining_hops,
248                ctx.iface,
249                ctx.now,
250                proof_timeout,
251            );
252            self.link_table.insert(link_id, link_entry);
253            actions.push(TransportAction::LinkRequestReceived {
254                link_id,
255                destination_hash: ctx.packet.destination_hash,
256                receiving_interface: ctx.iface,
257            });
258        } else {
259            let (trunc_hash, reverse_entry) =
260                create_reverse_entry(&ctx.packet, outbound_interface, ctx.iface, ctx.now);
261            self.reverse_table.insert(trunc_hash, reverse_entry);
262        }
263
264        actions.push(TransportAction::SendOnInterface {
265            interface: outbound_interface,
266            raw: new_raw.into(),
267        });
268
269        if let Some(entry) = self
270            .path_table
271            .get_mut(&ctx.packet.destination_hash)
272            .and_then(|ps| ps.primary_mut())
273        {
274            entry.timestamp = ctx.now;
275        }
276    }
277
278    fn handle_link_table_routing(
279        &mut self,
280        ctx: &InboundPacketCtx,
281        actions: &mut Vec<TransportAction>,
282    ) {
283        if !self.config.transport_enabled && self.config.identity_hash.is_none() {
284            return;
285        }
286        if ctx.packet.flags.packet_type == constants::PACKET_TYPE_ANNOUNCE
287            || ctx.packet.flags.packet_type == constants::PACKET_TYPE_LINKREQUEST
288            || ctx.packet.context == constants::CONTEXT_LRPROOF
289        {
290            return;
291        }
292
293        // This single keyed lookup is the native link fast path. The engine is
294        // already exclusively borrowed, and the typed entry contains every
295        // routing and hop-rewrite field, so a second denormalized cache would
296        // add invalidation risk without avoiding locks or linear searches.
297        // Keep this authoritative-table path even when profiling suggests a
298        // cache: upstream's corresponding denormalized experiment was removed.
299        let Some(link_entry) = self.link_table.get(&ctx.packet.destination_hash).cloned() else {
300            return;
301        };
302        if !link_entry.validated {
303            return;
304        }
305        let instance_local_link = self.interface_is_local_client(link_entry.next_hop_interface)
306            && self.interface_is_local_client(link_entry.received_interface);
307        let Some((outbound_iface, new_raw)) = route_via_link_table(
308            &ctx.packet,
309            &link_entry,
310            ctx.iface,
311            LocalHopRewrite {
312                local_hops_delta: self.config.local_hops_delta,
313                from_local_client: ctx.from_local_client,
314                skip_local_hops_delta: instance_local_link,
315            },
316        ) else {
317            // A link entry with no route from this receiving interface is an
318            // expected fast-path miss, not an operator-facing warning.
319            return;
320        };
321
322        self.packet_hashlist.add(ctx.packet.packet_hash);
323        actions.push(TransportAction::SendOnInterface {
324            interface: outbound_iface,
325            raw: new_raw.into(),
326        });
327
328        if let Some(entry) = self.link_table.get_mut(&ctx.packet.destination_hash) {
329            entry.timestamp = ctx.now;
330        }
331    }
332
333    fn handle_inbound_announce(
334        &mut self,
335        ctx: &InboundPacketCtx,
336        rng: &mut dyn Rng,
337        announce_queue: Option<&mut AnnounceVerifyQueue>,
338        actions: &mut Vec<TransportAction>,
339    ) {
340        if ctx.packet.flags.packet_type != constants::PACKET_TYPE_ANNOUNCE {
341            return;
342        }
343
344        if let Some(queue) = announce_queue {
345            self.try_enqueue_announce(ctx, rng, queue, actions);
346        } else {
347            let original_raw = ctx
348                .original_raw
349                .as_deref()
350                .expect("announce packets retain original raw bytes");
351            self.process_inbound_announce(
352                &ctx.packet,
353                original_raw,
354                ctx.iface,
355                ctx.now,
356                rng,
357                actions,
358            );
359        }
360    }
361
362    fn handle_inbound_local_delivery(
363        &self,
364        ctx: &InboundPacketCtx,
365        actions: &mut Vec<TransportAction>,
366    ) {
367        if (ctx.packet.flags.packet_type == constants::PACKET_TYPE_LINKREQUEST
368            || ctx.packet.flags.packet_type == constants::PACKET_TYPE_DATA)
369            // The engine's exclusive borrow makes this a direct, lock-free
370            // destination lookup for data and link requests alike; delivery
371            // reuses the same keyed registry.
372            && self
373                .local_destinations
374                .contains_key(&ctx.packet.destination_hash)
375        {
376            let mut delivery_raw = ctx.packet.raw.clone();
377            // Link responders learn the post-ingress hop metric from the
378            // authenticated LRRTT packet. Preserve that field when crossing
379            // the action boundary without changing ordinary callback bytes.
380            if ctx.packet.context == constants::CONTEXT_LRRTT && delivery_raw.len() >= 2 {
381                delivery_raw[1] = ctx.packet.hops;
382            }
383            actions.push(TransportAction::DeliverLocal {
384                destination_hash: ctx.packet.destination_hash,
385                raw: PacketBytes::from(delivery_raw),
386                packet_hash: ctx.packet.packet_hash,
387                receiving_interface: ctx.iface,
388            });
389        }
390    }
391
392    // =========================================================================
393    // Inbound announce processing
394    // =========================================================================
395
396    fn process_inbound_announce(
397        &mut self,
398        packet: &RawPacket,
399        original_raw: &[u8],
400        iface: InterfaceId,
401        now: f64,
402        rng: &mut dyn Rng,
403        actions: &mut Vec<TransportAction>,
404    ) {
405        if packet.flags.destination_type != constants::DESTINATION_SINGLE {
406            return;
407        }
408
409        let has_ratchet = packet.flags.context_flag == constants::FLAG_SET;
410
411        // Unpack and validate announce
412        let announce = match AnnounceData::unpack(&packet.data, has_ratchet) {
413            Ok(a) => a,
414            Err(_) => {
415                actions.push(TransportAction::ProtocolViolation { interface: iface });
416                return;
417            }
418        };
419
420        if self.should_hold_announce(packet, original_raw, iface, now) {
421            return;
422        }
423
424        let sig_cache_key =
425            Self::announce_sig_cache_key(packet.destination_hash, &announce.signature);
426
427        let validated = if self.announce_sig_cache.contains(&sig_cache_key) {
428            announce.to_validated_unchecked()
429        } else {
430            match announce.validate(&packet.destination_hash) {
431                Ok(v) => {
432                    self.announce_sig_cache.insert(sig_cache_key, now);
433                    v
434                }
435                Err(_) => {
436                    self.record_invalid_announce(&announce, iface, now, actions);
437                    return;
438                }
439            }
440        };
441
442        let received_from = self.announce_received_from(packet, now);
443        let random_blob = match extract_random_blob(&packet.data) {
444            Some(b) => b,
445            None => {
446                actions.push(TransportAction::ProtocolViolation { interface: iface });
447                return;
448            }
449        };
450        let announce_emitted = timebase_from_random_blob(&random_blob);
451
452        self.process_verified_announce(
453            VerifiedAnnounceCtx {
454                packet,
455                original_raw,
456                iface,
457                now,
458                validated,
459                received_from,
460                random_blob,
461                announce_emitted,
462            },
463            rng,
464            actions,
465        );
466    }
467
468    fn announce_raw_for_local_clients(&self, packet: &RawPacket) -> PacketBytes {
469        let Some(identity_hash) = self.config.identity_hash else {
470            return PacketBytes::from(packet.raw.clone());
471        };
472
473        if packet.raw.len() < 2 {
474            return PacketBytes::from(packet.raw.clone());
475        }
476
477        let payload_start = if packet.flags.header_type == constants::HEADER_2 {
478            18usize
479        } else {
480            2usize
481        };
482        if packet.raw.len() < payload_start {
483            return PacketBytes::from(packet.raw.clone());
484        }
485
486        let flags = (constants::HEADER_2 << 6)
487            | (constants::TRANSPORT_TRANSPORT << 4)
488            | (packet.raw[0] & 0x0F);
489        let mut raw = Vec::with_capacity(18 + packet.raw.len() - payload_start);
490        raw.push(flags);
491        raw.push(packet.hops);
492        raw.extend_from_slice(&identity_hash);
493        raw.extend_from_slice(&packet.raw[payload_start..]);
494        PacketBytes::from(raw)
495    }
496
497    pub(super) fn announce_sig_cache_key(
498        destination_hash: [u8; 16],
499        signature: &[u8; 64],
500    ) -> [u8; 32] {
501        let mut material = [0u8; 80];
502        material[..16].copy_from_slice(&destination_hash);
503        material[16..].copy_from_slice(signature);
504        hash::full_hash(&material)
505    }
506
507    fn announce_received_from(&mut self, packet: &RawPacket, now: f64) -> [u8; 16] {
508        if let Some(transport_id) = packet.transport_id {
509            if self.config.transport_enabled {
510                if let Some(announce_entry) = self.announce_table.get_mut(&packet.destination_hash)
511                {
512                    if packet.hops.checked_sub(1) == Some(announce_entry.hops) {
513                        announce_entry.local_rebroadcasts += 1;
514                        if announce_entry.retries > 0
515                            && announce_entry.local_rebroadcasts
516                                >= constants::LOCAL_REBROADCASTS_MAX
517                        {
518                            self.announce_table.remove(&packet.destination_hash);
519                        }
520                    }
521                    if let Some(announce_entry) = self.announce_table.get(&packet.destination_hash)
522                    {
523                        if packet.hops.checked_sub(1) == Some(announce_entry.hops + 1)
524                            && announce_entry.retries > 0
525                            && now < announce_entry.retransmit_timeout
526                        {
527                            self.announce_table.remove(&packet.destination_hash);
528                        }
529                    }
530                }
531            }
532            transport_id
533        } else {
534            packet.destination_hash
535        }
536    }
537
538    fn should_hold_announce(
539        &mut self,
540        packet: &RawPacket,
541        original_raw: &[u8],
542        iface: InterfaceId,
543        now: f64,
544    ) -> bool {
545        if self.has_path(&packet.destination_hash) {
546            return false;
547        }
548        if self
549            .discovery_path_requests
550            .contains_key(&packet.destination_hash)
551        {
552            return false;
553        }
554        let Some(info) = self.interfaces.get(&iface) else {
555            return false;
556        };
557        if packet.context == constants::CONTEXT_PATH_RESPONSE
558            || !self.ingress_control.should_ingress_limit(
559                iface,
560                &info.ingress_control,
561                info.ia_freq,
562                info.started,
563                now,
564            )
565        {
566            return false;
567        }
568        self.ingress_control.hold_announce(
569            iface,
570            &info.ingress_control,
571            packet.destination_hash,
572            ingress_control::HeldAnnounce {
573                raw: original_raw.to_vec(),
574                hops: packet.hops,
575                receiving_interface: iface,
576                rx: RxMetadata {
577                    rssi: packet.rssi,
578                    snr: packet.snr,
579                },
580                timestamp: now,
581            },
582        );
583        true
584    }
585
586    fn try_enqueue_announce(
587        &mut self,
588        ctx: &InboundPacketCtx,
589        rng: &mut dyn Rng,
590        announce_queue: &mut AnnounceVerifyQueue,
591        actions: &mut Vec<TransportAction>,
592    ) {
593        if ctx.packet.flags.destination_type != constants::DESTINATION_SINGLE {
594            return;
595        }
596
597        let has_ratchet = ctx.packet.flags.context_flag == constants::FLAG_SET;
598        let announce = match AnnounceData::unpack(&ctx.packet.data, has_ratchet) {
599            Ok(a) => a,
600            Err(_) => {
601                actions.push(TransportAction::ProtocolViolation {
602                    interface: ctx.iface,
603                });
604                return;
605            }
606        };
607
608        let received_from = self.announce_received_from(&ctx.packet, ctx.now);
609
610        // Announce verification and this registry read share the engine's
611        // exclusive processing turn, so no destinations-map lock is needed.
612        if self
613            .local_destinations
614            .contains_key(&ctx.packet.destination_hash)
615        {
616            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
617                "Announce:skipping local destination {:02x}{:02x}{:02x}{:02x}..",
618                ctx.packet.destination_hash[0],
619                ctx.packet.destination_hash[1],
620                ctx.packet.destination_hash[2],
621                ctx.packet.destination_hash[3],
622            );
623            return;
624        }
625
626        let original_raw = ctx
627            .original_raw
628            .as_deref()
629            .expect("announce packets retain original raw bytes");
630        if self.should_hold_announce(&ctx.packet, original_raw, ctx.iface, ctx.now) {
631            return;
632        }
633
634        let sig_cache_key =
635            Self::announce_sig_cache_key(ctx.packet.destination_hash, &announce.signature);
636        if self.announce_sig_cache.contains(&sig_cache_key) {
637            let validated = announce.to_validated_unchecked();
638            let random_blob = match extract_random_blob(&ctx.packet.data) {
639                Some(b) => b,
640                None => return,
641            };
642            let announce_emitted = timebase_from_random_blob(&random_blob);
643            self.process_verified_announce(
644                VerifiedAnnounceCtx {
645                    packet: &ctx.packet,
646                    original_raw,
647                    iface: ctx.iface,
648                    now: ctx.now,
649                    validated,
650                    received_from,
651                    random_blob,
652                    announce_emitted,
653                },
654                rng,
655                actions,
656            );
657            return;
658        }
659
660        if ctx.packet.context == constants::CONTEXT_PATH_RESPONSE {
661            let Ok(validated) = announce.validate(&ctx.packet.destination_hash) else {
662                self.record_invalid_announce(&announce, ctx.iface, ctx.now, actions);
663                return;
664            };
665            self.announce_sig_cache.insert(sig_cache_key, ctx.now);
666            let random_blob = match extract_random_blob(&ctx.packet.data) {
667                Some(b) => b,
668                None => return,
669            };
670            let announce_emitted = timebase_from_random_blob(&random_blob);
671            self.process_verified_announce(
672                VerifiedAnnounceCtx {
673                    packet: &ctx.packet,
674                    original_raw,
675                    iface: ctx.iface,
676                    now: ctx.now,
677                    validated,
678                    received_from,
679                    random_blob,
680                    announce_emitted,
681                },
682                rng,
683                actions,
684            );
685            return;
686        }
687
688        let random_blob = match extract_random_blob(&ctx.packet.data) {
689            Some(b) => b,
690            None => {
691                actions.push(TransportAction::ProtocolViolation {
692                    interface: ctx.iface,
693                });
694                return;
695            }
696        };
697        let announce_emitted = timebase_from_random_blob(&random_blob);
698        let key = AnnounceVerifyKey {
699            destination_hash: ctx.packet.destination_hash,
700            random_blob,
701            received_from,
702        };
703        let pending = PendingAnnounce {
704            original_raw: original_raw.to_vec(),
705            packet: ctx.packet.clone(),
706            interface: ctx.iface,
707            received_from,
708            queued_at: ctx.now,
709            best_hops: ctx.packet.hops,
710            emission_ts: announce_emitted,
711            random_blob,
712        };
713        let _ = announce_queue.enqueue(key, pending);
714    }
715
716    pub fn complete_verified_announce(
717        &mut self,
718        pending: PendingAnnounce,
719        validated: crate::announce::ValidatedAnnounce,
720        sig_cache_key: [u8; 32],
721        now: f64,
722        rng: &mut dyn Rng,
723    ) -> Vec<TransportAction> {
724        self.announce_sig_cache.insert(sig_cache_key, now);
725        let mut actions = Vec::new();
726        self.process_verified_announce(
727            VerifiedAnnounceCtx {
728                packet: &pending.packet,
729                original_raw: &pending.original_raw,
730                iface: pending.interface,
731                now,
732                validated,
733                received_from: pending.received_from,
734                random_blob: pending.random_blob,
735                announce_emitted: pending.emission_ts,
736            },
737            rng,
738            &mut actions,
739        );
740        actions
741    }
742
743    pub fn clear_failed_verified_announce(&mut self, _sig_cache_key: [u8; 32], _now: f64) {}
744
745    fn process_verified_announce(
746        &mut self,
747        ctx: VerifiedAnnounceCtx<'_>,
748        rng: &mut dyn Rng,
749        actions: &mut Vec<TransportAction>,
750    ) {
751        if self.is_blackholed(&ctx.validated.identity_hash, ctx.now) {
752            return;
753        }
754        if ctx.packet.hops > constants::PATHFINDER_M {
755            return;
756        }
757
758        let existing_set = self.path_table.get(&ctx.packet.destination_hash);
759        let was_unknown_destination = existing_set.is_none_or(|ps| ps.is_empty());
760
761        // Reset stale path state before first-path installation so path-state handling
762        // cannot race ahead of the path table for previously unknown destinations.
763        if was_unknown_destination {
764            self.path_states.remove(&ctx.packet.destination_hash);
765        }
766
767        // Multi-path aware decision
768        let is_unresponsive = self.path_is_unresponsive(&ctx.packet.destination_hash);
769
770        let current_gravity = existing_set
771            .and_then(|path_set| path_set.primary())
772            .and_then(|path| self.interfaces.get(&path.receiving_interface))
773            .map(|interface| interface.gravity);
774        let announce_gravity = self
775            .interfaces
776            .get(&ctx.iface)
777            .map(|interface| interface.gravity);
778        let higher_gravity_replacement = existing_set.is_some_and(|path_set| {
779            pathfinder::is_higher_gravity_replacement(
780                path_set,
781                ctx.packet.hops,
782                ctx.announce_emitted,
783                current_gravity,
784                announce_gravity,
785            )
786        });
787        let mp_decision = pathfinder::decide_announce_multipath_with_gravity(
788            existing_set,
789            ctx.packet.hops,
790            ctx.announce_emitted,
791            &ctx.random_blob,
792            &ctx.received_from,
793            is_unresponsive,
794            ctx.now,
795            self.config.prefer_shorter_path,
796            current_gravity,
797            announce_gravity,
798        );
799
800        if mp_decision == MultiPathDecision::Reject {
801            log::trace!(target: crate::logging::PATHING_LOG_TARGET,
802                "Announce:path decision REJECT for dest={:02x}{:02x}{:02x}{:02x}..",
803                ctx.packet.destination_hash[0],
804                ctx.packet.destination_hash[1],
805                ctx.packet.destination_hash[2],
806                ctx.packet.destination_hash[3],
807            );
808            return;
809        }
810        if higher_gravity_replacement {
811            log::log!(
812                target: crate::logging::PATHING_LOG_TARGET,
813                crate::logging::GRAVITY_UPDATE_LOG_LEVEL,
814                "Replacing path table entry for {:02x}{:02x}{:02x}{:02x}.. due to higher gravity ({:?}->{:?})",
815                ctx.packet.destination_hash[0],
816                ctx.packet.destination_hash[1],
817                ctx.packet.destination_hash[2],
818                ctx.packet.destination_hash[3],
819                current_gravity,
820                announce_gravity,
821            );
822        }
823
824        // Rate limiting
825        let rate_blocked = if ctx.packet.context != constants::CONTEXT_PATH_RESPONSE {
826            if let Some(iface_info) = self.interfaces.get(&ctx.iface) {
827                self.rate_limiter.check_and_update(
828                    &ctx.packet.destination_hash,
829                    ctx.now,
830                    iface_info.announce_rate_target,
831                    iface_info.announce_rate_grace,
832                    iface_info.announce_rate_penalty,
833                )
834            } else {
835                false
836            }
837        } else {
838            false
839        };
840
841        // Get interface mode for expiry calculation
842        let interface_mode = self
843            .interfaces
844            .get(&ctx.iface)
845            .map(|i| i.mode)
846            .unwrap_or(constants::MODE_FULL);
847
848        let expires = compute_path_expires(ctx.now, interface_mode);
849
850        // Get existing random blobs from the matching path (same next_hop) or empty
851        let existing_blobs = self
852            .path_table
853            .get(&ctx.packet.destination_hash)
854            .and_then(|ps| ps.find_by_next_hop(&ctx.received_from))
855            .map(|e| e.random_blobs.clone())
856            .unwrap_or_default();
857
858        // Generate RNG value for retransmit timeout
859        let mut rng_bytes = [0u8; 8];
860        rng.fill_bytes(&mut rng_bytes);
861        let rng_value = (u64::from_le_bytes(rng_bytes) as f64) / (u64::MAX as f64);
862
863        let is_path_response = ctx.packet.context == constants::CONTEXT_PATH_RESPONSE;
864
865        let (path_entry, announce_entry) = announce_proc::process_validated_announce(
866            ctx.packet.destination_hash,
867            ctx.packet.hops,
868            &ctx.packet.data,
869            &ctx.packet.raw,
870            ctx.packet.packet_hash,
871            ctx.packet.flags.context_flag,
872            ctx.received_from,
873            ctx.iface,
874            ctx.now,
875            existing_blobs,
876            ctx.random_blob,
877            expires,
878            rng_value,
879            self.config.transport_enabled,
880            is_path_response,
881            rate_blocked,
882            Some(ctx.original_raw.to_vec()),
883        );
884
885        // Emit CacheAnnounce for disk caching (pre-hop-increment raw)
886        actions.push(TransportAction::CacheAnnounce {
887            packet_hash: ctx.packet.packet_hash,
888            raw: ctx.original_raw.to_vec().into(),
889        });
890
891        // Store path via upsert into PathSet
892        match mp_decision {
893            MultiPathDecision::ReplacePrimary => self.upsert_primary_path_destination(
894                ctx.packet.destination_hash,
895                path_entry,
896                ctx.now,
897            ),
898            MultiPathDecision::AddAlternative => {
899                self.upsert_path_destination(ctx.packet.destination_hash, path_entry, ctx.now)
900            }
901            MultiPathDecision::Reject => unreachable!("rejected decisions returned above"),
902        }
903
904        // If receiving interface has a tunnel_id, store path in tunnel table too
905        if let Some(tunnel_id) = self.interfaces.get(&ctx.iface).and_then(|i| i.tunnel_id) {
906            let blobs = self
907                .path_table
908                .get(&ctx.packet.destination_hash)
909                .and_then(|ps| ps.find_by_next_hop(&ctx.received_from))
910                .map(|e| e.random_blobs.clone())
911                .unwrap_or_default();
912            self.tunnel_table.store_tunnel_path(
913                &tunnel_id,
914                ctx.packet.destination_hash,
915                tunnel::TunnelPath {
916                    timestamp: ctx.now,
917                    received_from: ctx.received_from,
918                    hops: ctx.packet.hops,
919                    expires,
920                    random_blobs: blobs,
921                    packet_hash: ctx.packet.packet_hash,
922                },
923                ctx.now,
924                self.config.destination_timeout_secs,
925                self.config.max_tunnel_destinations_total,
926            );
927        }
928
929        // Re-apply the path-state reset after storing the path entry so any transient
930        // stale state is also cleared once the destination exists in the path table.
931        self.path_states.remove(&ctx.packet.destination_hash);
932
933        // Store announce for retransmission
934        if let Some(ann) = announce_entry {
935            self.insert_announce_entry(ctx.packet.destination_hash, ann, ctx.now);
936        }
937
938        // Emit actions
939        actions.push(TransportAction::AnnounceReceived {
940            destination_hash: ctx.packet.destination_hash,
941            identity_hash: ctx.validated.identity_hash,
942            public_key: ctx.validated.public_key,
943            name_hash: ctx.validated.name_hash,
944            random_hash: ctx.validated.random_hash,
945            ratchet: ctx.validated.ratchet,
946            app_data: ctx.validated.app_data,
947            hops: ctx.packet.hops,
948            receiving_interface: ctx.iface,
949            rx: RxMetadata {
950                rssi: ctx.packet.rssi,
951                snr: ctx.packet.snr,
952            },
953        });
954
955        actions.push(TransportAction::PathUpdated {
956            destination_hash: ctx.packet.destination_hash,
957            hops: ctx.packet.hops,
958            next_hop: ctx.received_from,
959            interface: ctx.iface,
960        });
961
962        // Forward announce to local clients if any are connected
963        if self.has_local_clients() {
964            actions.push(TransportAction::ForwardToLocalClients {
965                raw: self.announce_raw_for_local_clients(ctx.packet),
966                exclude: Some(ctx.iface),
967            });
968        }
969
970        // Check for discovery path requests waiting for this announce
971        if let Some(requesting_interfaces) =
972            self.discovery_path_requests_waiting(&ctx.packet.destination_hash)
973        {
974            // Build a path response announce and queue it
975            let entry = AnnounceEntry {
976                timestamp: ctx.now,
977                retransmit_timeout: ctx.now,
978                retries: constants::PATHFINDER_R,
979                received_from: ctx.received_from,
980                hops: ctx.packet.hops,
981                packet_raw: ctx.packet.raw.clone(),
982                packet_data: ctx.packet.data.clone(),
983                destination_hash: ctx.packet.destination_hash,
984                context_flag: ctx.packet.flags.context_flag,
985                local_rebroadcasts: 0,
986                block_rebroadcasts: true,
987                attached_interface: requesting_interfaces.first().copied(),
988            };
989            if let Some(identity_hash) = self.config.identity_hash {
990                let raw = announce_proc::build_retransmit_announce(&entry, &identity_hash);
991                for interface in requesting_interfaces.iter().skip(1) {
992                    actions.push(TransportAction::SendOnInterface {
993                        interface: *interface,
994                        raw: raw.clone().into(),
995                    });
996                }
997            }
998            self.insert_announce_entry(ctx.packet.destination_hash, entry, ctx.now);
999        }
1000    }
1001
1002    pub fn announce_sig_cache_contains(&self, sig_cache_key: &[u8; 32]) -> bool {
1003        self.announce_sig_cache.contains(sig_cache_key)
1004    }
1005
1006    /// Check if there's a waiting discovery path request for a destination.
1007    /// Consumes the request if found (one-shot: the caller queues the announce response).
1008    pub(super) fn discovery_path_requests_waiting(
1009        &mut self,
1010        dest_hash: &[u8; 16],
1011    ) -> Option<Vec<InterfaceId>> {
1012        let request = self
1013            .discovery_path_requests
1014            .remove(dest_hash)
1015            .map(|req| req.requesting_interfaces);
1016        self.discovery_path_request_deadlines.remove(dest_hash);
1017        request
1018    }
1019
1020    // =========================================================================
1021    // Inbound proof processing
1022    // =========================================================================
1023
1024    fn process_inbound_proof(
1025        &mut self,
1026        ctx: &InboundPacketCtx,
1027        actions: &mut Vec<TransportAction>,
1028    ) {
1029        let packet = &ctx.packet;
1030        if packet.context == constants::CONTEXT_LRPROOF {
1031            // Link request proof routing
1032            if (self.config.transport_enabled)
1033                && self.link_table.contains_key(&packet.destination_hash)
1034            {
1035                let link_entry = self.link_table.get(&packet.destination_hash).cloned();
1036                if let Some(entry) = link_entry {
1037                    let instance_local_link = self
1038                        .interface_is_local_client(entry.next_hop_interface)
1039                        && self.interface_is_local_client(entry.received_interface);
1040                    if let Some((outbound_interface, new_raw)) = route_via_link_table(
1041                        packet,
1042                        &entry,
1043                        ctx.iface,
1044                        LocalHopRewrite {
1045                            local_hops_delta: self.config.local_hops_delta,
1046                            from_local_client: ctx.from_local_client,
1047                            skip_local_hops_delta: instance_local_link,
1048                        },
1049                    ) {
1050                        // Forward the proof (simplified: skip signature validation
1051                        // which requires Identity recall)
1052
1053                        // Mark link as validated
1054                        if let Some(le) = self.link_table.get_mut(&packet.destination_hash) {
1055                            le.validated = true;
1056                        }
1057
1058                        actions.push(TransportAction::LinkEstablished {
1059                            link_id: packet.destination_hash,
1060                            interface: outbound_interface,
1061                        });
1062
1063                        actions.push(TransportAction::SendOnInterface {
1064                            interface: outbound_interface,
1065                            raw: new_raw.into(),
1066                        });
1067                    } else if link_route_hops_match(packet.hops, &entry, ctx.iface) {
1068                        log::debug!(
1069                            "Link request proof received on wrong interface {}, not transporting it (expected {} or {})",
1070                            ctx.iface.0,
1071                            entry.next_hop_interface.0,
1072                            entry.received_interface.0,
1073                        );
1074                    } else {
1075                        log::debug!("{}", lrproof_hop_mismatch_diagnostic(packet.hops, &entry));
1076                    }
1077                }
1078            } else {
1079                // Could be for a local pending link - deliver locally
1080                let mut delivery_raw = packet.raw.clone();
1081                // LinkManager must see the same post-ingress hop metric that
1082                // was authenticated and considered by transport.
1083                if delivery_raw.len() >= 2 {
1084                    delivery_raw[1] = packet.hops;
1085                }
1086                actions.push(TransportAction::DeliverLocal {
1087                    destination_hash: packet.destination_hash,
1088                    raw: PacketBytes::from(delivery_raw),
1089                    packet_hash: packet.packet_hash,
1090                    receiving_interface: ctx.iface,
1091                });
1092            }
1093        } else {
1094            // Regular proof: check reverse table
1095            if self.config.transport_enabled {
1096                if let Some(reverse_entry) = self.reverse_table.remove(&packet.destination_hash) {
1097                    let proof_for_local_client =
1098                        self.interface_is_local_client(reverse_entry.receiving_interface);
1099                    if let Some(action) = route_proof_via_reverse(
1100                        packet,
1101                        &reverse_entry,
1102                        ctx.iface,
1103                        LocalHopRewrite {
1104                            local_hops_delta: self.config.local_hops_delta,
1105                            from_local_client: ctx.from_local_client,
1106                            skip_local_hops_delta: proof_for_local_client,
1107                        },
1108                    ) {
1109                        actions.push(action);
1110                    }
1111                }
1112            }
1113
1114            // Deliver to local receipts
1115            actions.push(TransportAction::DeliverLocal {
1116                destination_hash: packet.destination_hash,
1117                raw: PacketBytes::from(packet.raw.clone()),
1118                packet_hash: packet.packet_hash,
1119                receiving_interface: ctx.iface,
1120            });
1121        }
1122    }
1123}
rns_core/transport/inbound_engine.rs

rns_core/transport/
inbound_engine.rs
