1use super::*;
2use rns_core::transport::{InboundFrame, RxMetadata};
3
4impl Driver {
5 #[cfg(test)]
6 pub(crate) fn handle_frame_event(
7 &mut self,
8 interface_id: InterfaceId,
9 data: Vec<u8>,
10 rssi: Option<i16>,
11 snr: Option<f32>,
12 ) {
13 self.handle_classified_frame_event(interface_id, data, rssi, snr, false);
14 }
15
16 fn handle_classified_frame_event(
17 &mut self,
18 interface_id: InterfaceId,
19 data: Vec<u8>,
20 rssi: Option<i16>,
21 snr: Option<f32>,
22 ingress_limited: bool,
23 ) {
24 let received_size = data.len();
25 if data.len() > 2 && (data[0] & 0x03) == 0x01 {
26 log::debug!(
27 "Announce:frame from iface {} (len={}, flags=0x{:02x})",
28 interface_id.0,
29 data.len(),
30 data[0]
31 );
32 }
33 let Some(entry) = self.interfaces.get(&interface_id) else {
34 log::warn!("[{}] dropping frame from unknown interface", interface_id.0);
35 return;
36 };
37 if !entry.enabled || !entry.online {
38 return;
39 }
40 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
41 entry.stats.rxb += data.len() as u64;
42 entry.stats.rx_packets += 1;
43 }
44
45 let packet = if let Some(entry) = self.interfaces.get(&interface_id) {
46 if let Some(ref ifac_state) = entry.ifac {
47 match ifac::unmask_inbound(&data, ifac_state) {
48 Some(unmasked) => unmasked,
49 None => {
50 log::debug!("[{}] IFAC rejected packet", interface_id.0);
51 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
52 entry.stats.ifac_violations += 1;
53 }
54 return;
55 }
56 }
57 } else {
58 if data.len() > 2 && data[0] & 0x80 == 0x80 {
59 log::debug!(
60 "[{}] dropping packet with IFAC flag on non-IFAC interface",
61 interface_id.0
62 );
63 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
64 entry.stats.ifac_violations += 1;
65 }
66 return;
67 }
68 data
69 }
70 } else {
71 data
72 };
73
74 let parsed_packet = match RawPacket::unpack(&packet) {
75 Ok(packet) => packet,
76 Err(_) => {
77 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
78 entry.stats.protocol_violations += 1;
79 }
80 return;
81 }
82 };
83 if parsed_packet.flags.packet_type == rns_core::constants::PACKET_TYPE_ANNOUNCE
84 && packet.len() > rns_core::constants::MTU
85 {
86 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
87 entry.stats.protocol_violations += 1;
88 }
89 return;
90 }
91 let is_path_request = parsed_packet.destination_hash == self.path_request_dest;
92 let tagless_path_request = is_path_request && parsed_packet.data.len() <= 16;
93 if tagless_path_request || self.engine.is_unvalidated_link_packet(&parsed_packet) {
94 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
95 entry.stats.protocol_violations += 1;
96 }
97 return;
98 }
99 if is_path_request && parsed_packet.data.len() > 48 {
100 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
101 entry.stats.protocol_violations += 1;
102 }
103 }
104 let filter_frame = InboundFrame {
105 raw: &packet,
106 iface: interface_id,
107 now: time::now(),
108 rx: RxMetadata { rssi, snr },
109 };
110 if !self.engine.accepts_inbound_frame(filter_frame) {
111 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
112 if self
113 .engine
114 .is_packet_filter_protocol_violation(&parsed_packet, interface_id)
115 {
116 entry.stats.protocol_violations += 1;
117 } else {
118 entry.stats.packet_filter_hits += 1;
119 }
120 }
121 return;
122 }
123
124 #[cfg(feature = "hooks")]
125 {
126 let pkt_ctx = rns_hooks::PacketContext {
127 flags: if packet.is_empty() { 0 } else { packet[0] },
128 hops: if packet.len() > 1 { packet[1] } else { 0 },
129 destination_hash: extract_dest_hash(&packet),
130 context: 0,
131 packet_hash: [0; 32],
132 interface_id: interface_id.0,
133 data_offset: 0,
134 data_len: packet.len() as u32,
135 };
136 let ctx = HookContext::Packet {
137 ctx: &pkt_ctx,
138 raw: &packet,
139 };
140 let now = time::now();
141 let engine_ref = EngineRef {
142 engine: &self.engine,
143 interfaces: &self.interfaces,
144 link_manager: &self.link_manager,
145 now,
146 };
147 let provider_events_enabled = self.provider_events_enabled();
148 if let Some(ref e) = run_hook_inner(
149 &mut self.hook_slots[HookPoint::PreIngress as usize].programs,
150 &self.hook_manager,
151 &engine_ref,
152 &ctx,
153 now,
154 provider_events_enabled,
155 ) {
156 self.forward_hook_side_effects("PreIngress", e);
157 if e.hook_result.as_ref().is_some_and(|r| r.is_drop()) {
158 return;
159 }
160 }
161 }
162
163 if packet.len() > 2 && (packet[0] & 0x03) == 0x01 {
164 let now = time::now();
165 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
166 entry.stats.record_incoming_announce(now);
167 }
168 }
169
170 if let Some(entry) = self.interfaces.get(&interface_id) {
171 self.engine.update_interface_freqs(
172 interface_id,
173 entry.stats.incoming_announce_freq(),
174 entry.stats.incoming_path_request_freq(),
175 entry.stats.outgoing_path_request_freq(),
176 entry.stats.outgoing_path_request_samples(),
177 );
178 }
179
180 if let Some((link_id, destination_hash, packet_hops, proof_data)) = self
185 .engine
186 .inbound_lrproof_rebalance_candidate(&packet, interface_id)
187 {
188 if let Some(announced) = self.known_destination_announced(&destination_hash) {
189 let mut sig_pub = [0u8; 32];
190 sig_pub.copy_from_slice(&announced.public_key[32..64]);
191 if self.engine.rebalance_link_path_from_lrproof(
192 &link_id,
193 packet_hops,
194 interface_id,
195 &proof_data,
196 &sig_pub,
197 ) {
198 log::log!(
199 target: crate::logging::PATHING_LOG_TARGET,
200 crate::logging::REBALANCE_LOG_LEVEL,
201 "Re-balanced path to {:02x?} from link-request proof",
202 &destination_hash[..4],
203 );
204 }
205 }
206 }
207
208 let inbound_frame = InboundFrame {
209 raw: &packet,
210 iface: interface_id,
211 now: time::now(),
212 rx: RxMetadata { rssi, snr },
213 };
214
215 let actions = if self.async_announce_verification {
216 let mut announce_queue = self
217 .announce_verify_queue
218 .lock()
219 .unwrap_or_else(|poisoned| poisoned.into_inner());
220 self.engine.handle_inbound_with_announce_queue(
221 inbound_frame,
222 &mut self.rng,
223 Some(&mut announce_queue),
224 )
225 } else {
226 self.engine.handle_inbound(inbound_frame, &mut self.rng)
227 };
228 if actions.iter().any(|action| {
229 matches!(
230 action,
231 TransportAction::AnnounceReceived {
232 receiving_interface,
233 ..
234 } if *receiving_interface == interface_id
235 )
236 }) {
237 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
238 entry.stats.arxc = entry.stats.arxc.saturating_add(1);
239 entry.stats.arxb = entry.stats.arxb.saturating_add(received_size as u64);
240 }
241 }
242
243 #[cfg(feature = "hooks")]
244 {
245 let pkt_ctx = rns_hooks::PacketContext {
246 flags: if packet.is_empty() { 0 } else { packet[0] },
247 hops: if packet.len() > 1 { packet[1] } else { 0 },
248 destination_hash: extract_dest_hash(&packet),
249 context: 0,
250 packet_hash: [0; 32],
251 interface_id: interface_id.0,
252 data_offset: 0,
253 data_len: packet.len() as u32,
254 };
255 let ctx = HookContext::Packet {
256 ctx: &pkt_ctx,
257 raw: &packet,
258 };
259 let now = time::now();
260 let engine_ref = EngineRef {
261 engine: &self.engine,
262 interfaces: &self.interfaces,
263 link_manager: &self.link_manager,
264 now,
265 };
266 let provider_events_enabled = self.provider_events_enabled();
267 if let Some(ref e) = run_hook_inner(
268 &mut self.hook_slots[HookPoint::PreDispatch as usize].programs,
269 &self.hook_manager,
270 &engine_ref,
271 &ctx,
272 now,
273 provider_events_enabled,
274 ) {
275 self.forward_hook_side_effects("PreDispatch", e);
276 }
277 }
278
279 self.dispatch_all_with_ingress_class(actions, ingress_limited);
280 self.event_tx.set_ingress_bursts(
281 interface_id,
282 self.engine
283 .burst_active(&interface_id)
284 .then(|| self.engine.burst_activated(&interface_id)),
285 self.engine
286 .pr_burst_active(&interface_id)
287 .then(|| self.engine.pr_burst_activated(&interface_id)),
288 );
289 }
290
291 pub(crate) fn handle_announce_verified_event(
292 &mut self,
293 key: rns_core::transport::announce_verify_queue::AnnounceVerifyKey,
294 validated: rns_core::announce::ValidatedAnnounce,
295 sig_cache_key: [u8; 32],
296 ) {
297 let pending = {
298 let mut announce_queue = self
299 .announce_verify_queue
300 .lock()
301 .unwrap_or_else(|poisoned| poisoned.into_inner());
302 announce_queue.complete_success(&key)
303 };
304 if let Some(pending) = pending {
305 let interface = pending.interface;
306 let received_size = pending.original_raw.len();
307 let actions = self.engine.complete_verified_announce(
308 pending,
309 validated,
310 sig_cache_key,
311 time::now(),
312 &mut self.rng,
313 );
314 if actions
315 .iter()
316 .any(|action| matches!(action, TransportAction::AnnounceReceived { .. }))
317 {
318 if let Some(entry) = self.interfaces.get_mut(&interface) {
319 entry.stats.arxc = entry.stats.arxc.saturating_add(1);
320 entry.stats.arxb = entry.stats.arxb.saturating_add(received_size as u64);
321 }
322 }
323 self.dispatch_all(actions);
324 }
325 }
326
327 pub(crate) fn handle_tick_event(&mut self) {
328 #[cfg(feature = "hooks")]
329 {
330 let ctx = HookContext::Tick;
331 let now = time::now();
332 let engine_ref = EngineRef {
333 engine: &self.engine,
334 interfaces: &self.interfaces,
335 link_manager: &self.link_manager,
336 now,
337 };
338 let provider_events_enabled = self.provider_events_enabled();
339 if let Some(ref e) = run_hook_inner(
340 &mut self.hook_slots[HookPoint::Tick as usize].programs,
341 &self.hook_manager,
342 &engine_ref,
343 &ctx,
344 now,
345 provider_events_enabled,
346 ) {
347 self.forward_hook_side_effects("Tick", e);
348 }
349 }
350
351 let now = time::now();
352 self.sample_interface_traffic(now);
353 for (id, entry) in &self.interfaces {
354 self.engine.update_interface_freqs(
355 *id,
356 entry.stats.incoming_announce_freq(),
357 entry.stats.incoming_path_request_freq(),
358 entry.stats.outgoing_path_request_freq(),
359 entry.stats.outgoing_path_request_samples(),
360 );
361 }
362 let actions = self.engine.tick(now, &mut self.rng);
363 self.dispatch_all(actions);
364 let link_actions = self.link_manager.tick(&mut self.rng);
365 self.dispatch_link_actions(link_actions);
366 self.enforce_drain_deadline();
367 {
368 let tx = self.get_event_sender();
369 let hp_actions = self.holepunch_manager.tick(&tx);
370 self.dispatch_holepunch_actions(hp_actions);
371 }
372 self.tick_management_announces(now);
373 self.sent_packets
374 .retain(|_, (_, sent_time)| now - *sent_time < 60.0);
375 self.completed_proofs
376 .retain(|_, (_, received)| now - *received < 120.0);
377
378 self.tick_discovery_announcer(now);
379 #[cfg(feature = "iface-backbone")]
380 self.maintain_backbone_peer_pool();
381
382 self.memory_stats_counter += 1;
383 if self.memory_stats_counter >= 300 {
384 self.memory_stats_counter = 0;
385 self.log_memory_stats();
386 }
387
388 if self.discover_interfaces {
389 self.discovery_cleanup_counter += 1;
390 if self.discovery_cleanup_counter >= self.discovery_cleanup_interval_ticks {
391 self.discovery_cleanup_counter = 0;
392 if let Ok(removed) = self
393 .discovered_interfaces
394 .cleanup_with_blackholes(|identity| self.engine.is_blackholed(identity, now))
395 {
396 if removed > 0 {
397 log::info!("Discovery cleanup: removed {} stale entries", removed);
398 }
399 #[cfg(feature = "iface-backbone")]
400 self.cull_stale_discovered_backbone_peer_pool_candidates();
401 }
402 }
403 }
404
405 self.finish_ratchet_cleanup_if_ready();
406 if self.known_destination_cleanup.is_none() {
407 self.cache_cleanup_counter += 1;
408 if self.cache_cleanup_counter >= self.known_destinations_cleanup_interval_ticks {
409 self.cache_cleanup_counter = 0;
410 self.known_destination_cleanup = Some(KnownDestinationCleanup {
411 candidates: self.known_destinations.keys().copied().collect(),
412 cursor: 0,
413 started_at: Instant::now(),
414 removed: 0,
415 });
416 log::debug!("Cleaning known destinations at background priority...");
417 }
418 }
419 self.continue_known_destination_cleanup(now);
420
421 self.announce_cache_cleanup_counter += 1;
422 if self.announce_cache_cleanup_counter >= self.announce_cache_cleanup_interval_ticks {
423 self.announce_cache_cleanup_counter = 0;
424 if self.announce_cache.is_some() && self.cache_cleanup_active_hashes.is_none() {
425 self.cache_cleanup_active_hashes = Some(self.engine.active_packet_hashes());
426 self.cache_cleanup_entries = None;
427 self.cache_cleanup_removed = 0;
428 }
429 }
430
431 if self.cache_cleanup_active_hashes.is_some() {
432 if let Some(ref cache) = self.announce_cache {
433 if self.cache_cleanup_entries.is_none() {
434 match cache.entries() {
435 Ok(entries) => self.cache_cleanup_entries = Some(entries),
436 Err(e) => {
437 log::warn!("Announce cache cleanup failed to open directory: {}", e);
438 self.cache_cleanup_active_hashes = None;
439 self.cache_cleanup_entries = None;
440 }
441 }
442 }
443 }
444
445 if let Some(ref cache) = self.announce_cache {
446 let Some(active_hashes) = self.cache_cleanup_active_hashes.as_ref() else {
447 self.cache_cleanup_entries = None;
448 return;
449 };
450 let entries = match self.cache_cleanup_entries.as_mut() {
451 Some(entries) => entries,
452 None => return,
453 };
454 match cache.clean_batch(
455 active_hashes,
456 entries,
457 self.announce_cache_cleanup_batch_size,
458 ) {
459 Ok((removed, finished)) => {
460 self.cache_cleanup_removed += removed;
461 if finished {
462 if self.cache_cleanup_removed > 0 {
463 log::info!(
464 "Announce cache cleanup complete: removed {} stale files",
465 self.cache_cleanup_removed
466 );
467 }
468 self.cache_cleanup_active_hashes = None;
469 self.cache_cleanup_entries = None;
470 }
471 }
472 Err(e) => {
473 log::warn!("Announce cache cleanup failed: {}", e);
474 self.cache_cleanup_active_hashes = None;
475 self.cache_cleanup_entries = None;
476 }
477 }
478 } else {
479 self.cache_cleanup_active_hashes = None;
480 self.cache_cleanup_entries = None;
481 }
482 }
483 }
484
485 pub(super) fn sample_interface_traffic(&mut self, now: f64) {
486 self.traffic_samples
487 .retain(|id, _| self.interfaces.contains_key(id));
488 for (id, entry) in &mut self.interfaces {
489 let Some(previous) = self.traffic_samples.get(id).copied() else {
490 self.traffic_samples
491 .insert(*id, TrafficSample::from_stats(now, &entry.stats));
492 entry.stats.traffic_rates = Default::default();
493 continue;
494 };
495 let elapsed = now - previous.sampled_at;
496 if elapsed < 1.0 {
497 continue;
498 }
499 let bps =
500 |current: u64, prior: u64| current.saturating_sub(prior) as f64 * 8.0 / elapsed;
501 let pps = |current: u64, prior: u64| current.saturating_sub(prior) as f64 / elapsed;
502 entry.stats.traffic_rates = crate::interface::TrafficRates {
503 rxs: bps(entry.stats.rxb, previous.rxb),
504 txs: bps(entry.stats.txb, previous.txb),
505 rxpps: pps(entry.stats.rx_packets, previous.rx_packets),
506 txpps: pps(entry.stats.tx_packets, previous.tx_packets),
507 arxs: bps(entry.stats.arxb, previous.arxb),
508 atxs: bps(entry.stats.atxb, previous.atxb),
509 prxs: bps(entry.stats.prxb, previous.prxb),
510 ptxs: bps(entry.stats.ptxb, previous.ptxb),
511 };
512 self.traffic_samples
513 .insert(*id, TrafficSample::from_stats(now, &entry.stats));
514 }
515 }
516
517 fn continue_known_destination_cleanup(&mut self, now: f64) {
518 let Some(mut cleanup) = self.known_destination_cleanup.take() else {
519 return;
520 };
521 let end =
522 (cleanup.cursor + KNOWN_DESTINATION_CLEANUP_BATCH_SIZE).min(cleanup.candidates.len());
523 for destination_hash in &cleanup.candidates[cleanup.cursor..end] {
524 let remove = self
525 .known_destinations
526 .get(destination_hash)
527 .is_some_and(|state| {
528 !self.engine.has_path(destination_hash)
529 && !self.local_destinations.contains_key(destination_hash)
530 && !state.retained
531 && now - Self::known_destination_relevance_time(state)
532 >= self.known_destinations_ttl
533 });
534 if remove && self.known_destinations.remove(destination_hash).is_some() {
535 cleanup.removed += 1;
536 }
537 }
538 cleanup.cursor = end;
539 if cleanup.cursor < cleanup.candidates.len() {
540 self.known_destination_cleanup = Some(cleanup);
541 return;
542 }
543
544 let evicted = self.enforce_known_destination_cap(false);
545 let active_destinations = self.engine.active_destination_hashes();
546 let rate_limiters =
547 self.engine
548 .cull_rate_limiter(&active_destinations, now, self.rate_limiter_ttl_secs);
549 log::debug!(
550 "Cleaned known destinations in {:?}",
551 cleanup.started_at.elapsed()
552 );
553 if cleanup.removed > 0 || evicted > 0 || rate_limiters > 0 {
554 log::info!(
555 "Memory cleanup: removed {} known_destinations, evicted {} known_destinations, {} rate_limiter entries",
556 cleanup.removed,
557 evicted,
558 rate_limiters,
559 );
560 }
561 self.start_ratchet_cleanup(now);
562 }
563
564 fn finish_ratchet_cleanup_if_ready(&mut self) {
565 let finished = self
566 .ratchet_cleanup_handle
567 .as_ref()
568 .is_some_and(std::thread::JoinHandle::is_finished);
569 if finished {
570 if let Some(handle) = self.ratchet_cleanup_handle.take() {
571 let _ = handle.join();
572 }
573 }
574 }
575
576 fn start_ratchet_cleanup(&mut self, now: f64) {
577 if self.ratchet_cleanup_handle.is_some() {
578 return;
579 }
580 let Some(store) = self.ratchet_store.clone() else {
581 return;
582 };
583 let known_destinations = self.known_destinations.keys().copied().collect();
584 let expiry = self.ratchet_expiry_secs;
585 match std::thread::Builder::new()
586 .name("rns-destination-cleanup".into())
587 .spawn(move || {
588 #[cfg(target_family = "unix")]
589 unsafe {
590 libc::nice(5);
591 }
592 match store.cleanup(&known_destinations, now, expiry) {
593 Ok(stats) if stats.removed > 0 => log::info!(
594 "Known-destination cleanup removed {} stale ratchets",
595 stats.removed
596 ),
597 Ok(_) => {}
598 Err(err) => log::warn!("failed to clean ratchets: {}", err),
599 }
600 }) {
601 Ok(handle) => self.ratchet_cleanup_handle = Some(handle),
602 Err(err) => log::warn!("failed to start ratchet cleanup worker: {}", err),
603 }
604 }
605
606 pub(crate) fn handle_interface_up_event(
607 &mut self,
608 id: InterfaceId,
609 new_writer: Option<Box<dyn crate::interface::Writer>>,
610 info: Option<rns_core::transport::types::InterfaceInfo>,
611 ) {
612 let wants_tunnel;
613 let mut replay_shared_announces = false;
614 if let Some(mut info) = info {
615 log::info!("[{}] dynamic interface registered", id.0);
616 self.apply_announce_rate_defaults(&mut info);
617 self.apply_ingress_control_defaults(&mut info);
618 wants_tunnel = info.wants_tunnel;
619 let iface_type = infer_interface_type(&info.name);
620 info.started = time::now();
621 self.register_interface_runtime_defaults(&info);
622 self.engine.register_interface(info.clone());
623 if let Some(writer) = new_writer {
624 let (writer, async_writer_metrics) =
625 self.wrap_interface_writer(id, &info.name, writer);
626 self.interfaces.insert(
627 id,
628 InterfaceEntry {
629 id,
630 info,
631 writer,
632 async_writer_metrics: Some(async_writer_metrics),
633 enabled: true,
634 online: true,
635 dynamic: true,
636 ifac: None,
637 stats: InterfaceStats {
638 started: time::now(),
639 ..Default::default()
640 },
641 interface_type: iface_type,
642 send_retry_at: None,
643 send_retry_backoff: Duration::ZERO,
644 },
645 );
646 }
647 self.callbacks.on_interface_up(id);
648 #[cfg(feature = "hooks")]
649 {
650 let ctx = HookContext::Interface { interface_id: id.0 };
651 let now = time::now();
652 let engine_ref = EngineRef {
653 engine: &self.engine,
654 interfaces: &self.interfaces,
655 link_manager: &self.link_manager,
656 now,
657 };
658 let provider_events_enabled = self.provider_events_enabled();
659 if let Some(ref e) = run_hook_inner(
660 &mut self.hook_slots[HookPoint::InterfaceUp as usize].programs,
661 &self.hook_manager,
662 &engine_ref,
663 &ctx,
664 now,
665 provider_events_enabled,
666 ) {
667 self.forward_hook_side_effects("InterfaceUp", e);
668 }
669 }
670 } else {
671 let is_local_client = self
672 .interfaces
673 .get(&id)
674 .map(|entry| entry.info.is_local_client)
675 .unwrap_or(false);
676 replay_shared_announces =
677 is_local_client && self.shared_reconnect_pending.remove(&id).unwrap_or(false);
678 let interface_name = self
679 .interfaces
680 .get(&id)
681 .map(|entry| entry.info.name.clone())
682 .unwrap_or_else(|| format!("iface-{}", id.0));
683 let wrapped_writer =
684 new_writer.map(|writer| self.wrap_interface_writer(id, &interface_name, writer));
685 if let Some(entry) = self.interfaces.get_mut(&id) {
686 log::info!("[{}] interface online", id.0);
687 wants_tunnel = entry.info.wants_tunnel;
688 entry.online = true;
689 if let Some((writer, async_writer_metrics)) = wrapped_writer {
690 log::info!("[{}] writer refreshed after reconnect", id.0);
691 entry.writer = writer;
692 entry.async_writer_metrics = Some(async_writer_metrics);
693 }
694 self.callbacks.on_interface_up(id);
695 #[cfg(feature = "hooks")]
696 {
697 let ctx = HookContext::Interface { interface_id: id.0 };
698 let now = time::now();
699 let engine_ref = EngineRef {
700 engine: &self.engine,
701 interfaces: &self.interfaces,
702 link_manager: &self.link_manager,
703 now,
704 };
705 let provider_events_enabled = self.provider_events_enabled();
706 if let Some(ref e) = run_hook_inner(
707 &mut self.hook_slots[HookPoint::InterfaceUp as usize].programs,
708 &self.hook_manager,
709 &engine_ref,
710 &ctx,
711 now,
712 provider_events_enabled,
713 ) {
714 self.forward_hook_side_effects("InterfaceUp", e);
715 }
716 }
717 } else {
718 wants_tunnel = false;
719 }
720 }
721
722 if wants_tunnel {
723 self.synthesize_tunnel_for_interface(id);
724 }
725 if replay_shared_announces {
726 self.replay_shared_announces();
727 }
728 }
729
730 pub(crate) fn handle_dynamic_interface_up_event(
731 &mut self,
732 id: InterfaceId,
733 writer: Box<dyn crate::interface::Writer>,
734 mut registration: crate::event::DynamicInterfaceRegistration,
735 ) {
736 let parent_id = registration.parent_id;
737 let parent_ifac = self.interfaces.get(&parent_id).and_then(|parent| {
738 registration.info.mode = parent.info.mode;
739 registration.info.gravity = parent.info.gravity;
740 registration.info.recursive_prs = parent.info.recursive_prs;
741 registration.info.announces_from_internal = parent.info.announces_from_internal;
742 registration.info.announces_to_internal = parent.info.announces_to_internal;
743 registration.info.announce_rate_target = parent.info.announce_rate_target;
744 registration.info.announce_rate_grace = parent.info.announce_rate_grace;
745 registration.info.announce_rate_penalty = parent.info.announce_rate_penalty;
746 registration.info.ingress_control = parent.info.ingress_control;
747 parent.ifac.clone()
748 });
749 let inherited_ifac = registration.ifac.take().or(parent_ifac);
750
751 let interface_type = registration.interface_type;
752 let telemetry = registration.telemetry;
753 self.handle_interface_up_event(id, Some(writer), Some(registration.info));
754 if let Some(entry) = self.interfaces.get_mut(&id) {
755 self.dynamic_interface_parents.insert(id, parent_id);
756 self.event_tx.register_dynamic_parent(id, parent_id);
757 entry.interface_type = interface_type;
758 entry.ifac = inherited_ifac;
759 Self::apply_interface_telemetry(entry, telemetry);
760 }
761 }
762
763 fn apply_interface_telemetry(
764 entry: &mut crate::interface::InterfaceEntry,
765 telemetry: crate::event::InterfaceTelemetry,
766 ) {
767 entry.stats.cpu_load = telemetry.cpu_load;
768 entry.stats.mem_load = telemetry.mem_load;
769 entry.stats.switch_id = telemetry.switch_id;
770 entry.stats.endpoint_id = telemetry.endpoint_id;
771 entry.stats.via_switch_id = telemetry.via_switch_id;
772 entry.stats.peers = telemetry.peers;
773 }
774
775 pub(crate) fn handle_interface_telemetry_event(
776 &mut self,
777 interface_id: InterfaceId,
778 telemetry: crate::event::InterfaceTelemetry,
779 ) {
780 if let Some(entry) = self.interfaces.get_mut(&interface_id) {
781 Self::apply_interface_telemetry(entry, telemetry);
782 }
783 }
784
785 pub(crate) fn handle_interface_down_event(&mut self, id: InterfaceId) {
786 self.event_tx.remove_interface(id);
787 if let Some(entry) = self.interfaces.get(&id) {
788 if let Some(tunnel_id) = entry.info.tunnel_id {
789 self.engine.void_tunnel_interface(&tunnel_id);
790 }
791 }
792
793 if let Some(entry) = self.interfaces.get(&id) {
794 let is_dynamic = entry.dynamic;
795 let is_local_client = entry.info.is_local_client;
796 let interface_name = entry.info.name.clone();
797 if is_dynamic {
798 log::info!("[{}] dynamic interface removed", id.0);
799 self.interface_runtime_defaults.remove(&interface_name);
800 self.engine.deregister_interface(id);
801 self.interfaces.remove(&id);
802 self.dynamic_interface_parents.remove(&id);
803 } else {
804 log::info!("[{}] interface offline", id.0);
805 if let Some(entry) = self.interfaces.get_mut(&id) {
806 entry.online = false;
807 } else {
808 log::warn!(
809 "interface {} disappeared while handling interface-down",
810 id.0
811 );
812 return;
813 }
814 if is_local_client {
815 self.handle_shared_interface_down(id);
816 }
817 }
818 self.callbacks.on_interface_down(id);
819 #[cfg(feature = "hooks")]
820 {
821 let ctx = HookContext::Interface { interface_id: id.0 };
822 let now = time::now();
823 let engine_ref = EngineRef {
824 engine: &self.engine,
825 interfaces: &self.interfaces,
826 link_manager: &self.link_manager,
827 now,
828 };
829 let provider_events_enabled = self.provider_events_enabled();
830 if let Some(ref e) = run_hook_inner(
831 &mut self.hook_slots[HookPoint::InterfaceDown as usize].programs,
832 &self.hook_manager,
833 &engine_ref,
834 &ctx,
835 now,
836 provider_events_enabled,
837 ) {
838 self.forward_hook_side_effects("InterfaceDown", e);
839 }
840 }
841 }
842 #[cfg(feature = "iface-backbone")]
843 self.handle_backbone_peer_pool_down(id);
844 }
845
846 pub(crate) fn known_destination_route_hint(
847 &self,
848 dest_hash: &[u8; 16],
849 ) -> Option<(InterfaceId, u8)> {
850 let announced = &self.known_destinations.get(dest_hash)?.announced;
851 let iface = announced.receiving_interface;
852 if iface.0 == 0 {
853 return None;
854 }
855
856 self.interfaces
857 .get(&iface)
858 .filter(|entry| entry.online)
859 .map(|_| (iface, announced.hops))
860 }
861
862 pub(crate) fn handle_send_outbound_event(
863 &mut self,
864 raw: Vec<u8>,
865 dest_type: u8,
866 attached_interface: Option<InterfaceId>,
867 ) {
868 if self.is_draining() {
869 self.reject_new_work("send outbound packet");
870 return;
871 }
872 match RawPacket::unpack(&raw) {
873 Ok(packet) => {
874 let is_announce =
875 packet.flags.packet_type == rns_core::constants::PACKET_TYPE_ANNOUNCE;
876 if is_announce {
877 log::debug!(
878 "SendOutbound: ANNOUNCE for {:02x?} (len={}, dest_type={}, attached={:?})",
879 &packet.destination_hash[..4],
880 raw.len(),
881 dest_type,
882 attached_interface
883 );
884 }
885 if packet.flags.packet_type == rns_core::constants::PACKET_TYPE_DATA {
886 log::debug!(
887 "SendOutbound: DATA dest={:02x}{:02x}{:02x}{:02x}.. header={} transport={:02x?} attached={:?}",
888 packet.destination_hash[0],
889 packet.destination_hash[1],
890 packet.destination_hash[2],
891 packet.destination_hash[3],
892 packet.flags.header_type,
893 packet.transport_id.as_ref().map(|id| &id[..4]),
894 attached_interface
895 );
896 self.sent_packets
897 .insert(packet.packet_hash, (packet.destination_hash, time::now()));
898 }
899 let actions = self.engine.handle_outbound(
900 &packet,
901 dest_type,
902 attached_interface,
903 time::now(),
904 );
905 if packet.flags.packet_type == rns_core::constants::PACKET_TYPE_DATA {
906 log::debug!(
907 "SendOutbound: DATA routed to {} actions: {:?}",
908 actions.len(),
909 actions
910 .iter()
911 .map(|a| match a {
912 TransportAction::SendOnInterface { interface, raw } =>
913 format!("SendOn({};{}b)", interface.0, raw.len()),
914 TransportAction::BroadcastOnAllInterfaces { raw, .. } =>
915 format!("BroadcastAll({}b)", raw.len()),
916 _ => "other".to_string(),
917 })
918 .collect::<Vec<_>>()
919 );
920 }
921 if is_announce {
922 log::debug!(
923 "SendOutbound: announce routed to {} actions: {:?}",
924 actions.len(),
925 actions
926 .iter()
927 .map(|a| match a {
928 TransportAction::SendOnInterface { interface, .. } =>
929 format!("SendOn({})", interface.0),
930 TransportAction::BroadcastOnAllInterfaces { .. } =>
931 "BroadcastAll".to_string(),
932 _ => "other".to_string(),
933 })
934 .collect::<Vec<_>>()
935 );
936 }
937 self.dispatch_all(actions);
938 }
939 Err(e) => {
940 log::warn!("SendOutbound: failed to unpack packet: {:?}", e);
941 }
942 }
943 }
944
945 pub fn run(&mut self) {
947 loop {
948 let received = match self.rx.recv_classified() {
949 Ok(e) => e,
950 Err(_) => break, };
952
953 match received.event {
954 Event::Frame {
955 interface_id,
956 data,
957 rssi,
958 snr,
959 } => {
960 self.handle_classified_frame_event(
961 interface_id,
962 data,
963 rssi,
964 snr,
965 received.ingress_limited,
966 );
967 }
968 Event::AnnounceVerified {
969 key,
970 validated,
971 sig_cache_key,
972 } => {
973 self.handle_announce_verified_event(key, validated, sig_cache_key);
974 }
975 Event::AnnounceVerifyFailed { key, .. } => {
976 let mut announce_queue = self
977 .announce_verify_queue
978 .lock()
979 .unwrap_or_else(|poisoned| poisoned.into_inner());
980 let interface = announce_queue.pending_interface(&key);
981 let blackholed = announce_queue
982 .pending_identity_hash(&key)
983 .is_some_and(|hash| self.engine.is_blackholed(&hash, time::now()));
984 if announce_queue.complete_failure(&key) {
985 if let Some(entry) = interface.and_then(|id| self.interfaces.get_mut(&id)) {
986 if !blackholed {
987 entry.stats.protocol_violations += 1;
988 }
989 }
990 }
991 }
992 Event::Tick => self.handle_tick_event(),
993 Event::BeginDrain { timeout } => {
994 self.begin_drain(timeout);
995 }
996 Event::InterfaceUp(id, new_writer, info) => {
997 self.handle_interface_up_event(id, new_writer, info);
998 }
999 Event::DynamicInterfaceUp {
1000 id,
1001 writer,
1002 registration,
1003 } => self.handle_dynamic_interface_up_event(id, writer, registration),
1004 Event::InterfaceTelemetry {
1005 interface_id,
1006 telemetry,
1007 } => self.handle_interface_telemetry_event(interface_id, telemetry),
1008 Event::InterfaceDown(id) => self.handle_interface_down_event(id),
1009 Event::SendOutbound {
1010 raw,
1011 dest_type,
1012 attached_interface,
1013 } => self.handle_send_outbound_event(raw, dest_type, attached_interface),
1014 Event::RegisterDestination {
1015 dest_hash,
1016 dest_type,
1017 } => {
1018 self.engine.register_destination(dest_hash, dest_type);
1019 self.local_destinations.insert(dest_hash, dest_type);
1020 }
1021 Event::StoreSharedAnnounce {
1022 dest_hash,
1023 name_hash,
1024 identity_prv_key,
1025 app_data,
1026 } => {
1027 self.shared_announces.insert(
1028 dest_hash,
1029 SharedAnnounceRecord {
1030 name_hash,
1031 identity_prv_key,
1032 app_data,
1033 },
1034 );
1035 }
1036 Event::DeregisterDestination { dest_hash } => {
1037 self.engine.deregister_destination(&dest_hash);
1038 self.local_destinations.remove(&dest_hash);
1039 self.shared_announces.remove(&dest_hash);
1040 }
1041 Event::Query(request, response_tx) => {
1042 let response = self.handle_query_mut(request);
1043 let _ = response_tx.send(response);
1044 }
1045 Event::DeregisterLinkDestination { dest_hash } => {
1046 self.link_manager.deregister_link_destination(&dest_hash);
1047 }
1048 Event::RegisterLinkDestination {
1049 dest_hash,
1050 sig_prv_bytes,
1051 sig_pub_bytes,
1052 resource_strategy,
1053 max_request_size,
1054 } => {
1055 let sig_prv =
1056 rns_crypto::ed25519::Ed25519PrivateKey::from_bytes(&sig_prv_bytes);
1057 let strat = match resource_strategy {
1058 1 => crate::link_manager::ResourceStrategy::AcceptAll,
1059 2 => crate::link_manager::ResourceStrategy::AcceptApp,
1060 _ => crate::link_manager::ResourceStrategy::AcceptNone,
1061 };
1062 self.link_manager.register_link_destination(
1063 dest_hash,
1064 sig_prv,
1065 sig_pub_bytes,
1066 strat,
1067 );
1068 self.link_manager
1069 .set_link_destination_max_request_size(&dest_hash, max_request_size);
1070 self.engine
1072 .register_destination(dest_hash, rns_core::constants::DESTINATION_SINGLE);
1073 self.local_destinations
1074 .insert(dest_hash, rns_core::constants::DESTINATION_SINGLE);
1075 }
1076 Event::RegisterRequestHandler {
1077 path,
1078 allowed_list,
1079 handler,
1080 } => {
1081 self.link_manager.register_request_handler(
1082 &path,
1083 allowed_list,
1084 move |link_id, p, data, remote| handler(link_id, p, data, remote),
1085 );
1086 }
1087 Event::RegisterRequestHandlerResponse {
1088 path,
1089 allowed_list,
1090 handler,
1091 } => {
1092 self.link_manager.register_request_handler_response(
1093 &path,
1094 allowed_list,
1095 move |link_id, p, data, remote| handler(link_id, p, data, remote),
1096 );
1097 }
1098 Event::RegisterDeferredRequestHandler {
1099 path,
1100 allowed_list,
1101 handler,
1102 } => {
1103 self.link_manager.register_deferred_request_handler(
1104 &path,
1105 allowed_list,
1106 move |link_id, path, request_id, data, remote| {
1107 handler(link_id, path, request_id, data, remote)
1108 },
1109 );
1110 }
1111 Event::CreateLink {
1112 dest_hash,
1113 dest_sig_pub_bytes,
1114 response_tx,
1115 } => {
1116 if self.is_draining() {
1117 self.reject_new_work("create link");
1118 let _ = (dest_hash, dest_sig_pub_bytes);
1119 let _ = response_tx.send([0u8; 16]);
1120 continue;
1121 }
1122 let next_hop_interface = self.engine.next_hop_interface(&dest_hash);
1123 let recalled_route_hint = if next_hop_interface.is_none() {
1124 self.known_destination_route_hint(&dest_hash)
1125 } else {
1126 None
1127 };
1128 if recalled_route_hint.is_some() {
1129 let _ = self.mark_known_destination_used(&dest_hash);
1130 }
1131 let attached_interface =
1132 next_hop_interface.or(recalled_route_hint.map(|(iface, _)| iface));
1133 let hops = self
1134 .engine
1135 .hops_to(&dest_hash)
1136 .or_else(|| recalled_route_hint.map(|(_, hops)| hops))
1137 .unwrap_or(0);
1138 let mtu = attached_interface
1139 .and_then(|iface_id| self.interfaces.get(&iface_id))
1140 .map(|entry| entry.info.mtu)
1141 .unwrap_or(rns_core::constants::MTU as u32);
1142 let (link_id, mut link_actions) = self.link_manager.create_link(
1143 &dest_hash,
1144 &dest_sig_pub_bytes,
1145 hops,
1146 mtu,
1147 &mut self.rng,
1148 );
1149 if let Some(iface) = attached_interface {
1150 self.link_manager.set_link_route_hint(&link_id, iface, None);
1151 }
1152 if next_hop_interface.is_none() {
1153 if let Some(iface) = attached_interface {
1154 for action in &mut link_actions {
1155 if let LinkManagerAction::SendPacket {
1156 dest_type,
1157 attached_interface,
1158 ..
1159 } = action
1160 {
1161 if *dest_type == rns_core::constants::DESTINATION_LINK
1162 && attached_interface.is_none()
1163 {
1164 *attached_interface = Some(iface);
1165 }
1166 }
1167 }
1168 }
1169 }
1170 let _ = response_tx.send(link_id);
1171 self.dispatch_link_actions(link_actions);
1172 }
1173 Event::SendRequest {
1174 link_id,
1175 path,
1176 data,
1177 max_response_size,
1178 } => {
1179 if self.is_draining() {
1180 self.reject_new_work("send link request");
1181 let _ = (link_id, path, data, max_response_size);
1182 continue;
1183 }
1184 let link_actions = self.link_manager.send_request_with_max_response_size(
1185 &link_id,
1186 &path,
1187 &data,
1188 max_response_size,
1189 &mut self.rng,
1190 );
1191 self.dispatch_link_actions(link_actions);
1192 }
1193 Event::SendDeferredResponse {
1194 link_id,
1195 request_id,
1196 data,
1197 } => {
1198 let link_actions = self.link_manager.send_deferred_response(
1199 &link_id,
1200 &request_id,
1201 &data,
1202 &mut self.rng,
1203 );
1204 self.dispatch_link_actions(link_actions);
1205 }
1206 Event::IdentifyOnLink {
1207 link_id,
1208 identity_prv_key,
1209 } => {
1210 if self.is_draining() {
1211 self.reject_new_work("identify on link");
1212 let _ = (link_id, identity_prv_key);
1213 continue;
1214 }
1215 let identity =
1216 rns_crypto::identity::Identity::from_private_key(&identity_prv_key);
1217 let link_actions =
1218 self.link_manager
1219 .identify(&link_id, &identity, &mut self.rng);
1220 self.dispatch_link_actions(link_actions);
1221 }
1222 Event::TeardownLink { link_id } => {
1223 let link_actions = self.link_manager.teardown_link(&link_id);
1224 self.dispatch_link_actions(link_actions);
1225 }
1226 Event::SendResource {
1227 link_id,
1228 data,
1229 metadata,
1230 auto_compress,
1231 } => {
1232 if self.is_draining() {
1233 self.reject_new_work("send resource");
1234 let _ = (link_id, data, metadata, auto_compress);
1235 continue;
1236 }
1237 let link_actions = self.link_manager.send_resource_with_auto_compress(
1238 &link_id,
1239 &data,
1240 metadata.as_deref(),
1241 auto_compress,
1242 &mut self.rng,
1243 );
1244 self.dispatch_link_actions(link_actions);
1245 }
1246 Event::SendResourceStream {
1247 link_id,
1248 transfer_id,
1249 reader,
1250 declared_length,
1251 metadata,
1252 auto_compress,
1253 } => {
1254 if self.is_draining() {
1255 self.callbacks.on_resource_stream_failed(
1256 rns_core::types::LinkId(link_id),
1257 transfer_id,
1258 crate::resource::ResourceTransferError::Cancelled,
1259 );
1260 continue;
1261 }
1262 let link_actions = self.link_manager.send_resource_stream(
1263 &link_id,
1264 transfer_id,
1265 reader,
1266 declared_length,
1267 metadata,
1268 auto_compress,
1269 &mut self.rng,
1270 );
1271 self.dispatch_link_actions(link_actions);
1272 }
1273 Event::SetResourceStrategy { link_id, strategy } => {
1274 use crate::link_manager::ResourceStrategy;
1275 let strat = match strategy {
1276 0 => ResourceStrategy::AcceptNone,
1277 1 => ResourceStrategy::AcceptAll,
1278 2 => ResourceStrategy::AcceptApp,
1279 _ => ResourceStrategy::AcceptNone,
1280 };
1281 self.link_manager.set_resource_strategy(&link_id, strat);
1282 }
1283 Event::SetResourceReceiveMode { link_id, mode } => {
1284 self.link_manager.set_resource_receive_mode(&link_id, mode);
1285 }
1286 Event::AcceptResource {
1287 link_id,
1288 resource_hash,
1289 accept,
1290 } => {
1291 if self.is_draining() && accept {
1292 self.reject_new_work("accept resource");
1293 let _ = (link_id, resource_hash, accept);
1294 continue;
1295 }
1296 let link_actions = self.link_manager.accept_resource(
1297 &link_id,
1298 &resource_hash,
1299 accept,
1300 &mut self.rng,
1301 );
1302 self.dispatch_link_actions(link_actions);
1303 }
1304 Event::SendChannelMessage {
1305 link_id,
1306 msgtype,
1307 payload,
1308 response_tx,
1309 } => {
1310 if self.is_draining() {
1311 self.reject_new_work("send channel message");
1312 let _ = response_tx.send(Err(self.drain_error("send channel message")));
1313 continue;
1314 }
1315 match self.link_manager.send_channel_message(
1316 &link_id,
1317 msgtype,
1318 &payload,
1319 &mut self.rng,
1320 ) {
1321 Ok(link_actions) => {
1322 self.dispatch_link_actions(link_actions);
1323 let _ = response_tx.send(Ok(()));
1324 }
1325 Err(err) => {
1326 let _ = response_tx.send(Err(err));
1327 }
1328 }
1329 }
1330 Event::SendOnLink {
1331 link_id,
1332 data,
1333 context,
1334 response_tx,
1335 } => {
1336 if self.is_draining() {
1337 self.reject_new_work("send link payload");
1338 if let Some(tx) = response_tx {
1339 let _ = tx.send(Err(crate::event::LinkDatagramError::DriverStopped));
1340 }
1341 continue;
1342 }
1343 let result =
1344 self.link_manager
1345 .try_send_on_link(&link_id, &data, context, &mut self.rng);
1346 match result {
1347 Ok(link_actions) => {
1348 self.dispatch_link_actions(link_actions);
1349 if let Some(tx) = response_tx {
1350 let _ = tx.send(Ok(()));
1351 }
1352 }
1353 Err(error) => {
1354 if let Some(tx) = response_tx {
1355 let _ = tx.send(Err(error));
1356 }
1357 }
1358 }
1359 }
1360 Event::RequestPath { dest_hash } => {
1361 if self.is_draining() {
1362 self.reject_new_work("request path");
1363 let _ = dest_hash;
1364 continue;
1365 }
1366 self.handle_request_path(dest_hash);
1367 }
1368 Event::RegisterProofStrategy {
1369 dest_hash,
1370 strategy,
1371 signing_key,
1372 } => {
1373 let identity = signing_key
1374 .map(|key| rns_crypto::identity::Identity::from_private_key(&key));
1375 self.proof_strategies
1376 .insert(dest_hash, (strategy, identity));
1377 }
1378 Event::ProposeDirectConnect { link_id } => {
1379 if self.is_draining() {
1380 self.reject_new_work("propose direct connect");
1381 let _ = link_id;
1382 continue;
1383 }
1384 let derived_key = self.link_manager.get_derived_key(&link_id);
1385 if let Some(dk) = derived_key {
1386 let tx = self.get_event_sender();
1387 let hp_actions =
1388 self.holepunch_manager
1389 .propose(link_id, &dk, &mut self.rng, &tx);
1390 self.dispatch_holepunch_actions(hp_actions);
1391 } else {
1392 log::warn!(
1393 "Cannot propose direct connect: no derived key for link {:02x?}",
1394 &link_id[..4]
1395 );
1396 }
1397 }
1398 Event::SetDirectConnectPolicy { policy } => {
1399 self.holepunch_manager.set_policy(policy);
1400 }
1401 Event::HolePunchProbeResult {
1402 link_id,
1403 session_id,
1404 observed_addr,
1405 socket,
1406 probe_server,
1407 } => {
1408 let hp_actions = self.holepunch_manager.handle_probe_result(
1409 link_id,
1410 session_id,
1411 observed_addr,
1412 socket,
1413 probe_server,
1414 );
1415 self.dispatch_holepunch_actions(hp_actions);
1416 }
1417 Event::HolePunchProbeFailed {
1418 link_id,
1419 session_id,
1420 } => {
1421 let hp_actions = self
1422 .holepunch_manager
1423 .handle_probe_failed(link_id, session_id);
1424 self.dispatch_holepunch_actions(hp_actions);
1425 }
1426 Event::LoadHook {
1427 name,
1428 wasm_bytes,
1429 attach_point,
1430 priority,
1431 response_tx,
1432 } => {
1433 #[cfg(feature = "hooks")]
1434 {
1435 let result = (|| -> Result<(), String> {
1436 let point_idx = crate::config::parse_hook_point(&attach_point)
1437 .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1438 let mgr = self
1439 .hook_manager
1440 .as_ref()
1441 .ok_or_else(|| "hook manager not available".to_string())?;
1442 let program = mgr
1443 .compile(name.clone(), &wasm_bytes, priority)
1444 .map_err(|e| format!("compile error: {}", e))?;
1445 self.hook_slots[point_idx].attach(program);
1446 log::info!(
1447 "Loaded hook '{}' at point {} (priority {})",
1448 name,
1449 attach_point,
1450 priority
1451 );
1452 Ok(())
1453 })();
1454 let _ = response_tx.send(result);
1455 }
1456 #[cfg(not(feature = "hooks"))]
1457 {
1458 let _ = (name, wasm_bytes, attach_point, priority);
1459 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1460 }
1461 }
1462 Event::LoadHookFile {
1463 name,
1464 path,
1465 hook_type,
1466 attach_point,
1467 priority,
1468 response_tx,
1469 } => {
1470 #[cfg(feature = "hooks")]
1471 {
1472 let result = (|| -> Result<(), String> {
1473 let point_idx = crate::config::parse_hook_point(&attach_point)
1474 .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1475 let backend = crate::config::parse_hook_backend(&hook_type)?;
1476 let mgr = self
1477 .hook_manager
1478 .as_ref()
1479 .ok_or_else(|| "hook manager not available".to_string())?;
1480 let program = mgr
1481 .load_file_backend(
1482 name.clone(),
1483 std::path::Path::new(&path),
1484 priority,
1485 backend,
1486 )
1487 .map_err(|e| format!("load error: {}", e))?;
1488 self.hook_slots[point_idx].attach(program);
1489 log::info!(
1490 "Loaded {} hook '{}' at point {} (priority {})",
1491 backend.as_str(),
1492 name,
1493 attach_point,
1494 priority
1495 );
1496 Ok(())
1497 })();
1498 let _ = response_tx.send(result);
1499 }
1500 #[cfg(not(feature = "hooks"))]
1501 {
1502 let _ = (name, path, hook_type, attach_point, priority);
1503 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1504 }
1505 }
1506 Event::LoadBuiltinHook {
1507 name,
1508 builtin_id,
1509 attach_point,
1510 priority,
1511 response_tx,
1512 } => {
1513 #[cfg(feature = "hooks")]
1514 {
1515 let result = (|| -> Result<(), String> {
1516 let point_idx = crate::config::parse_hook_point(&attach_point)
1517 .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1518 let mgr = self
1519 .hook_manager
1520 .as_ref()
1521 .ok_or_else(|| "hook manager not available".to_string())?;
1522 let program = mgr
1523 .load_builtin(name.clone(), builtin_id.as_str(), priority)
1524 .map_err(|e| format!("load error: {}", e))?;
1525 self.hook_slots[point_idx].attach(program);
1526 log::info!(
1527 "Loaded built-in hook '{}' ({}) at point {} (priority {})",
1528 name,
1529 builtin_id,
1530 attach_point,
1531 priority
1532 );
1533 Ok(())
1534 })();
1535 let _ = response_tx.send(result);
1536 }
1537 #[cfg(not(feature = "hooks"))]
1538 {
1539 let _ = (name, builtin_id, attach_point, priority);
1540 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1541 }
1542 }
1543 Event::UnloadHook {
1544 name,
1545 attach_point,
1546 response_tx,
1547 } => {
1548 #[cfg(feature = "hooks")]
1549 {
1550 let result = (|| -> Result<(), String> {
1551 let point_idx = crate::config::parse_hook_point(&attach_point)
1552 .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1553 match self.hook_slots[point_idx].detach(&name) {
1554 Some(_) => {
1555 log::info!(
1556 "Unloaded hook '{}' from point {}",
1557 name,
1558 attach_point
1559 );
1560 Ok(())
1561 }
1562 None => Err(format!(
1563 "hook '{}' not found at point '{}'",
1564 name, attach_point
1565 )),
1566 }
1567 })();
1568 let _ = response_tx.send(result);
1569 }
1570 #[cfg(not(feature = "hooks"))]
1571 {
1572 let _ = (name, attach_point);
1573 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1574 }
1575 }
1576 Event::ReloadHook {
1577 name,
1578 attach_point,
1579 wasm_bytes,
1580 response_tx,
1581 } => {
1582 #[cfg(feature = "hooks")]
1583 {
1584 let result = (|| -> Result<(), String> {
1585 let point_idx = crate::config::parse_hook_point(&attach_point)
1586 .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1587 let old =
1588 self.hook_slots[point_idx].detach(&name).ok_or_else(|| {
1589 format!("hook '{}' not found at point '{}'", name, attach_point)
1590 })?;
1591 let priority = old.priority;
1592 let mgr = match self.hook_manager.as_ref() {
1593 Some(m) => m,
1594 None => {
1595 self.hook_slots[point_idx].attach(old);
1596 return Err("hook manager not available".to_string());
1597 }
1598 };
1599 match mgr.compile(name.clone(), &wasm_bytes, priority) {
1600 Ok(program) => {
1601 self.hook_slots[point_idx].attach(program);
1602 log::info!(
1603 "Reloaded hook '{}' at point {} (priority {})",
1604 name,
1605 attach_point,
1606 priority
1607 );
1608 Ok(())
1609 }
1610 Err(e) => {
1611 self.hook_slots[point_idx].attach(old);
1612 Err(format!("compile error: {}", e))
1613 }
1614 }
1615 })();
1616 let _ = response_tx.send(result);
1617 }
1618 #[cfg(not(feature = "hooks"))]
1619 {
1620 let _ = (name, attach_point, wasm_bytes);
1621 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1622 }
1623 }
1624 Event::ReloadHookFile {
1625 name,
1626 attach_point,
1627 path,
1628 hook_type,
1629 response_tx,
1630 } => {
1631 #[cfg(feature = "hooks")]
1632 {
1633 let result = (|| -> Result<(), String> {
1634 let point_idx = crate::config::parse_hook_point(&attach_point)
1635 .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1636 let old =
1637 self.hook_slots[point_idx].detach(&name).ok_or_else(|| {
1638 format!("hook '{}' not found at point '{}'", name, attach_point)
1639 })?;
1640 let priority = old.priority;
1641 let backend = match crate::config::parse_hook_backend(&hook_type) {
1642 Ok(backend) => backend,
1643 Err(e) => {
1644 self.hook_slots[point_idx].attach(old);
1645 return Err(e);
1646 }
1647 };
1648 let mgr = match self.hook_manager.as_ref() {
1649 Some(m) => m,
1650 None => {
1651 self.hook_slots[point_idx].attach(old);
1652 return Err("hook manager not available".to_string());
1653 }
1654 };
1655 match mgr.load_file_backend(
1656 name.clone(),
1657 std::path::Path::new(&path),
1658 priority,
1659 backend,
1660 ) {
1661 Ok(program) => {
1662 self.hook_slots[point_idx].attach(program);
1663 log::info!(
1664 "Reloaded {} hook '{}' at point {} (priority {})",
1665 backend.as_str(),
1666 name,
1667 attach_point,
1668 priority
1669 );
1670 Ok(())
1671 }
1672 Err(e) => {
1673 self.hook_slots[point_idx].attach(old);
1674 Err(format!("load error: {}", e))
1675 }
1676 }
1677 })();
1678 let _ = response_tx.send(result);
1679 }
1680 #[cfg(not(feature = "hooks"))]
1681 {
1682 let _ = (name, attach_point, path, hook_type);
1683 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1684 }
1685 }
1686 Event::ReloadBuiltinHook {
1687 name,
1688 attach_point,
1689 builtin_id,
1690 response_tx,
1691 } => {
1692 #[cfg(feature = "hooks")]
1693 {
1694 let result = (|| -> Result<(), String> {
1695 let point_idx = crate::config::parse_hook_point(&attach_point)
1696 .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1697 let old =
1698 self.hook_slots[point_idx].detach(&name).ok_or_else(|| {
1699 format!("hook '{}' not found at point '{}'", name, attach_point)
1700 })?;
1701 let priority = old.priority;
1702 let mgr = match self.hook_manager.as_ref() {
1703 Some(m) => m,
1704 None => {
1705 self.hook_slots[point_idx].attach(old);
1706 return Err("hook manager not available".to_string());
1707 }
1708 };
1709 match mgr.load_builtin(name.clone(), builtin_id.as_str(), priority) {
1710 Ok(program) => {
1711 self.hook_slots[point_idx].attach(program);
1712 log::info!(
1713 "Reloaded built-in hook '{}' ({}) at point {} (priority {})",
1714 name,
1715 builtin_id,
1716 attach_point,
1717 priority
1718 );
1719 Ok(())
1720 }
1721 Err(e) => {
1722 self.hook_slots[point_idx].attach(old);
1723 Err(format!("load error: {}", e))
1724 }
1725 }
1726 })();
1727 let _ = response_tx.send(result);
1728 }
1729 #[cfg(not(feature = "hooks"))]
1730 {
1731 let _ = (name, attach_point, builtin_id);
1732 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1733 }
1734 }
1735 Event::SetHookEnabled {
1736 name,
1737 attach_point,
1738 enabled,
1739 response_tx,
1740 } => {
1741 #[cfg(feature = "hooks")]
1742 {
1743 let result = self.update_hook_program(&name, &attach_point, |program| {
1744 program.enabled = enabled;
1745 });
1746 if result.is_ok() {
1747 log::info!(
1748 "{} hook '{}' at point {}",
1749 if enabled { "Enabled" } else { "Disabled" },
1750 name,
1751 attach_point,
1752 );
1753 }
1754 let _ = response_tx.send(result);
1755 }
1756 #[cfg(not(feature = "hooks"))]
1757 {
1758 let _ = (name, attach_point, enabled);
1759 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1760 }
1761 }
1762 Event::SetHookPriority {
1763 name,
1764 attach_point,
1765 priority,
1766 response_tx,
1767 } => {
1768 #[cfg(feature = "hooks")]
1769 {
1770 let result = self.update_hook_program(&name, &attach_point, |program| {
1771 program.priority = priority;
1772 });
1773 if result.is_ok() {
1774 if let Some(point_idx) = crate::config::parse_hook_point(&attach_point)
1775 {
1776 self.hook_slots[point_idx]
1777 .programs
1778 .sort_by_key(|program| std::cmp::Reverse(program.priority));
1779 log::info!(
1780 "Updated hook '{}' at point {} to priority {}",
1781 name,
1782 attach_point,
1783 priority,
1784 );
1785 } else {
1786 log::error!(
1787 "hook point '{}' became invalid during priority update",
1788 attach_point
1789 );
1790 }
1791 }
1792 let _ = response_tx.send(result);
1793 }
1794 #[cfg(not(feature = "hooks"))]
1795 {
1796 let _ = (name, attach_point, priority);
1797 let _ = response_tx.send(Err("hooks not enabled".to_string()));
1798 }
1799 }
1800 Event::ListHooks { response_tx } => {
1801 #[cfg(feature = "hooks")]
1802 {
1803 let hook_point_names = [
1804 "PreIngress",
1805 "PreDispatch",
1806 "AnnounceReceived",
1807 "PathUpdated",
1808 "AnnounceRetransmit",
1809 "LinkRequestReceived",
1810 "LinkEstablished",
1811 "LinkClosed",
1812 "InterfaceUp",
1813 "InterfaceDown",
1814 "InterfaceConfigChanged",
1815 "BackbonePeerConnected",
1816 "BackbonePeerDisconnected",
1817 "BackbonePeerIdleTimeout",
1818 "BackbonePeerWriteStall",
1819 "BackbonePeerPenalty",
1820 "SendOnInterface",
1821 "BroadcastOnAllInterfaces",
1822 "DeliverLocal",
1823 "TunnelSynthesize",
1824 "Tick",
1825 ];
1826 let mut infos = Vec::new();
1827 for (idx, slot) in self.hook_slots.iter().enumerate() {
1828 let point_name = hook_point_names.get(idx).unwrap_or(&"Unknown");
1829 for prog in &slot.programs {
1830 infos.push(crate::event::HookInfo {
1831 name: prog.name.clone(),
1832 hook_type: prog.backend_name().to_string(),
1833 attach_point: point_name.to_string(),
1834 priority: prog.priority,
1835 enabled: prog.enabled,
1836 consecutive_traps: prog.consecutive_traps,
1837 });
1838 }
1839 }
1840 let _ = response_tx.send(infos);
1841 }
1842 #[cfg(not(feature = "hooks"))]
1843 {
1844 let _ = response_tx.send(Vec::new());
1845 }
1846 }
1847 Event::InterfaceConfigChanged(id) => {
1848 #[cfg(feature = "hooks")]
1849 {
1850 let ctx = HookContext::Interface { interface_id: id.0 };
1851 let now = time::now();
1852 let engine_ref = EngineRef {
1853 engine: &self.engine,
1854 interfaces: &self.interfaces,
1855 link_manager: &self.link_manager,
1856 now,
1857 };
1858 let provider_events_enabled = self.provider_events_enabled();
1859 if let Some(ref e) = run_hook_inner(
1860 &mut self.hook_slots[HookPoint::InterfaceConfigChanged as usize]
1861 .programs,
1862 &self.hook_manager,
1863 &engine_ref,
1864 &ctx,
1865 now,
1866 provider_events_enabled,
1867 ) {
1868 self.forward_hook_side_effects("InterfaceConfigChanged", e);
1869 }
1870 }
1871 #[cfg(not(feature = "hooks"))]
1872 let _ = id;
1873 }
1874 Event::BackbonePeerConnected {
1875 server_interface_id,
1876 peer_interface_id,
1877 peer_ip,
1878 peer_port,
1879 } => {
1880 #[cfg(feature = "hooks")]
1881 {
1882 self.run_backbone_peer_hook(
1883 "BackbonePeerConnected",
1884 HookPoint::BackbonePeerConnected,
1885 &BackbonePeerHookEvent {
1886 server_interface_id,
1887 peer_interface_id: Some(peer_interface_id),
1888 peer_ip,
1889 peer_port,
1890 connected_for: Duration::ZERO,
1891 had_received_data: false,
1892 penalty_level: 0,
1893 blacklist_for: Duration::ZERO,
1894 },
1895 );
1896 }
1897 #[cfg(not(feature = "hooks"))]
1898 let _ = (server_interface_id, peer_interface_id, peer_ip, peer_port);
1899 }
1900 Event::BackbonePeerDisconnected {
1901 server_interface_id,
1902 peer_interface_id,
1903 peer_ip,
1904 peer_port,
1905 connected_for,
1906 had_received_data,
1907 } => {
1908 #[cfg(feature = "hooks")]
1909 {
1910 self.run_backbone_peer_hook(
1911 "BackbonePeerDisconnected",
1912 HookPoint::BackbonePeerDisconnected,
1913 &BackbonePeerHookEvent {
1914 server_interface_id,
1915 peer_interface_id: Some(peer_interface_id),
1916 peer_ip,
1917 peer_port,
1918 connected_for,
1919 had_received_data,
1920 penalty_level: 0,
1921 blacklist_for: Duration::ZERO,
1922 },
1923 );
1924 }
1925 #[cfg(not(feature = "hooks"))]
1926 let _ = (
1927 server_interface_id,
1928 peer_interface_id,
1929 peer_ip,
1930 peer_port,
1931 connected_for,
1932 had_received_data,
1933 );
1934 }
1935 Event::BackbonePeerIdleTimeout {
1936 server_interface_id,
1937 peer_interface_id,
1938 peer_ip,
1939 peer_port,
1940 connected_for,
1941 } => {
1942 #[cfg(feature = "hooks")]
1943 {
1944 self.run_backbone_peer_hook(
1945 "BackbonePeerIdleTimeout",
1946 HookPoint::BackbonePeerIdleTimeout,
1947 &BackbonePeerHookEvent {
1948 server_interface_id,
1949 peer_interface_id: Some(peer_interface_id),
1950 peer_ip,
1951 peer_port,
1952 connected_for,
1953 had_received_data: false,
1954 penalty_level: 0,
1955 blacklist_for: Duration::ZERO,
1956 },
1957 );
1958 }
1959 #[cfg(not(feature = "hooks"))]
1960 let _ = (
1961 server_interface_id,
1962 peer_interface_id,
1963 peer_ip,
1964 peer_port,
1965 connected_for,
1966 );
1967 }
1968 Event::BackbonePeerWriteStall {
1969 server_interface_id,
1970 peer_interface_id,
1971 peer_ip,
1972 peer_port,
1973 connected_for,
1974 } => {
1975 #[cfg(feature = "hooks")]
1976 {
1977 self.run_backbone_peer_hook(
1978 "BackbonePeerWriteStall",
1979 HookPoint::BackbonePeerWriteStall,
1980 &BackbonePeerHookEvent {
1981 server_interface_id,
1982 peer_interface_id: Some(peer_interface_id),
1983 peer_ip,
1984 peer_port,
1985 connected_for,
1986 had_received_data: false,
1987 penalty_level: 0,
1988 blacklist_for: Duration::ZERO,
1989 },
1990 );
1991 }
1992 #[cfg(not(feature = "hooks"))]
1993 let _ = (
1994 server_interface_id,
1995 peer_interface_id,
1996 peer_ip,
1997 peer_port,
1998 connected_for,
1999 );
2000 }
2001 Event::BackbonePeerPenalty {
2002 server_interface_id,
2003 peer_ip,
2004 penalty_level,
2005 blacklist_for,
2006 } => {
2007 #[cfg(feature = "hooks")]
2008 {
2009 self.run_backbone_peer_hook(
2010 "BackbonePeerPenalty",
2011 HookPoint::BackbonePeerPenalty,
2012 &BackbonePeerHookEvent {
2013 server_interface_id,
2014 peer_interface_id: None,
2015 peer_ip,
2016 peer_port: 0,
2017 connected_for: Duration::ZERO,
2018 had_received_data: false,
2019 penalty_level,
2020 blacklist_for,
2021 },
2022 );
2023 }
2024 #[cfg(not(feature = "hooks"))]
2025 let _ = (server_interface_id, peer_ip, penalty_level, blacklist_for);
2026 }
2027 Event::Shutdown => {
2028 self.graceful_shutdown();
2029 break;
2030 }
2031 }
2032 }
2033 }
2034 pub(crate) fn handle_tunnel_synth_delivery(
2035 &mut self,
2036 raw: &[u8],
2037 receiving_interface: InterfaceId,
2038 ) {
2039 let packet = match RawPacket::unpack(raw) {
2041 Ok(p) => p,
2042 Err(_) => return,
2043 };
2044
2045 match rns_core::transport::tunnel::validate_tunnel_synthesize_data(&packet.data) {
2046 Ok(validated) => {
2047 let iface_id = self
2050 .interfaces
2051 .iter()
2052 .find(|(_, entry)| entry.info.wants_tunnel && entry.online && entry.enabled)
2053 .map(|(id, _)| *id);
2054
2055 if let Some(iface) = iface_id {
2056 let now = time::now();
2057 let tunnel_actions = self.engine.handle_tunnel(validated.tunnel_id, iface, now);
2058 self.dispatch_all(tunnel_actions);
2059 }
2060 }
2061 Err(e) => {
2062 log::debug!("Tunnel synthesis validation failed: {}", e);
2063 if let Some(entry) = self.interfaces.get_mut(&receiving_interface) {
2064 entry.stats.protocol_violations += 1;
2065 }
2066 }
2067 }
2068 }
2069
2070 pub(crate) fn synthesize_tunnel_for_interface(&mut self, interface: InterfaceId) {
2074 if let Some(ref identity) = self.transport_identity {
2075 let actions = self
2076 .engine
2077 .synthesize_tunnel(identity, interface, &mut self.rng);
2078 self.dispatch_all(actions);
2079 }
2080 }
2081
2082 pub(crate) fn handle_request_path(&mut self, dest_hash: [u8; 16]) {
2084 let mut data = Vec::with_capacity(48);
2086 data.extend_from_slice(&dest_hash);
2087
2088 if self.engine.transport_enabled() {
2089 if let Some(id_hash) = self.engine.identity_hash() {
2090 data.extend_from_slice(id_hash);
2091 }
2092 }
2093
2094 let mut tag = [0u8; 16];
2096 self.rng.fill_bytes(&mut tag);
2097 data.extend_from_slice(&tag);
2098
2099 let flags = rns_core::packet::PacketFlags {
2101 header_type: rns_core::constants::HEADER_1,
2102 context_flag: rns_core::constants::FLAG_UNSET,
2103 transport_type: rns_core::constants::TRANSPORT_BROADCAST,
2104 destination_type: rns_core::constants::DESTINATION_PLAIN,
2105 packet_type: rns_core::constants::PACKET_TYPE_DATA,
2106 };
2107
2108 if let Ok(packet) = RawPacket::pack(
2109 flags,
2110 0,
2111 &self.path_request_dest,
2112 None,
2113 rns_core::constants::CONTEXT_NONE,
2114 &data,
2115 ) {
2116 let now = time::now();
2117 self.engine.record_outbound_path_request(dest_hash, now);
2118 let actions = self.engine.handle_outbound(
2119 &packet,
2120 rns_core::constants::DESTINATION_PLAIN,
2121 None,
2122 now,
2123 );
2124 self.dispatch_all(actions);
2125 }
2126 }
2127
2128 pub(crate) fn get_event_sender(&self) -> crate::event::EventSender {
2131 self.event_tx.clone()
2135 }
2136
2137 const MANAGEMENT_ANNOUNCE_DELAY: f64 = 5.0;
2139
2140 pub(crate) fn tick_discovery_announcer(&mut self, now: f64) {
2142 let announcer = match self.interface_announcer.as_mut() {
2143 Some(a) => a,
2144 None => return,
2145 };
2146
2147 announcer.maybe_start(now);
2148
2149 let stamp_result = match announcer.poll_ready() {
2150 Some(r) => r,
2151 None => return,
2152 };
2153
2154 if !announcer.contains_interface(&stamp_result.interface_name) {
2155 log::debug!(
2156 "Discovery: dropping completed stamp for removed interface '{}'",
2157 stamp_result.interface_name
2158 );
2159 return;
2160 }
2161
2162 let app_data = match stamp_result.app_data {
2163 Ok(app_data) => app_data,
2164 Err(error) => {
2165 log::error!(
2166 "Discovery: suppressing announce for '{}': {}",
2167 stamp_result.interface_name,
2168 error
2169 );
2170 return;
2171 }
2172 };
2173
2174 let identity = match self.transport_identity.as_ref() {
2175 Some(id) => id,
2176 None => {
2177 log::warn!("Discovery: stamp ready but no transport identity");
2178 return;
2179 }
2180 };
2181
2182 let identity_hash = identity.hash();
2184 let disc_dest = rns_core::destination::destination_hash(
2185 crate::discovery::APP_NAME,
2186 &["discovery", "interface"],
2187 Some(identity_hash),
2188 );
2189 let name_hash = self.discovery_name_hash;
2190 let mut random_hash = [0u8; 10];
2191 self.rng.fill_bytes(&mut random_hash);
2192
2193 let (announce_data, _) = match rns_core::announce::AnnounceData::pack(
2194 identity,
2195 &disc_dest,
2196 &name_hash,
2197 &random_hash,
2198 None,
2199 Some(&app_data),
2200 ) {
2201 Ok(v) => v,
2202 Err(e) => {
2203 log::warn!("Discovery: failed to pack announce: {}", e);
2204 return;
2205 }
2206 };
2207
2208 let flags = rns_core::packet::PacketFlags {
2209 header_type: rns_core::constants::HEADER_1,
2210 context_flag: rns_core::constants::FLAG_UNSET,
2211 transport_type: rns_core::constants::TRANSPORT_BROADCAST,
2212 destination_type: rns_core::constants::DESTINATION_SINGLE,
2213 packet_type: rns_core::constants::PACKET_TYPE_ANNOUNCE,
2214 };
2215
2216 let packet = match RawPacket::pack(
2217 flags,
2218 0,
2219 &disc_dest,
2220 None,
2221 rns_core::constants::CONTEXT_NONE,
2222 &announce_data,
2223 ) {
2224 Ok(p) => p,
2225 Err(e) => {
2226 log::warn!("Discovery: failed to pack packet: {}", e);
2227 return;
2228 }
2229 };
2230
2231 self.engine.register_destination(
2232 packet.destination_hash,
2233 rns_core::constants::DESTINATION_SINGLE,
2234 );
2235 let outbound_actions = self.engine.handle_outbound(
2236 &packet,
2237 rns_core::constants::DESTINATION_SINGLE,
2238 None,
2239 now,
2240 );
2241 log::debug!(
2242 "Discovery announce sent for interface '{}' ({} actions, dest={:02x?})",
2243 stamp_result.interface_name,
2244 outbound_actions.len(),
2245 &disc_dest[..4],
2246 );
2247 self.dispatch_all(outbound_actions);
2248 }
2249
2250 pub(crate) fn rss_mb() -> Option<f64> {
2252 let statm = std::fs::read_to_string("/proc/self/statm").ok()?;
2253 let rss_pages: u64 = statm.split_whitespace().nth(1)?.parse().ok()?;
2254 Some(rss_pages as f64 * 4096.0 / (1024.0 * 1024.0))
2255 }
2256
2257 pub(crate) fn parse_proc_kib(contents: &str, key: &str) -> Option<u64> {
2258 contents.lines().find_map(|line| {
2259 let value = line.strip_prefix(key)?;
2260 value.split_whitespace().next()?.parse().ok()
2261 })
2262 }
2263
2264 pub(crate) fn proc_status_mb() -> Option<(f64, f64, f64, f64)> {
2265 let status = std::fs::read_to_string("/proc/self/status").ok()?;
2266 let vm_rss = Self::parse_proc_kib(&status, "VmRSS:")? as f64 / 1024.0;
2267 let vm_hwm = Self::parse_proc_kib(&status, "VmHWM:")? as f64 / 1024.0;
2268 let vm_data = Self::parse_proc_kib(&status, "VmData:")? as f64 / 1024.0;
2269 let vm_swap = Self::parse_proc_kib(&status, "VmSwap:").unwrap_or(0) as f64 / 1024.0;
2270 Some((vm_rss, vm_hwm, vm_data, vm_swap))
2271 }
2272
2273 pub(crate) fn smaps_rollup_mb() -> Option<MemoryRollup> {
2274 let smaps = std::fs::read_to_string("/proc/self/smaps_rollup").ok()?;
2275 let rss_kib = Self::parse_proc_kib(&smaps, "Rss:")?;
2276 let anon_kib = Self::parse_proc_kib(&smaps, "Anonymous:")?;
2277 let shared_clean_kib = Self::parse_proc_kib(&smaps, "Shared_Clean:").unwrap_or(0);
2278 let shared_dirty_kib = Self::parse_proc_kib(&smaps, "Shared_Dirty:").unwrap_or(0);
2279 let private_clean_kib = Self::parse_proc_kib(&smaps, "Private_Clean:").unwrap_or(0);
2280 let private_dirty_kib = Self::parse_proc_kib(&smaps, "Private_Dirty:").unwrap_or(0);
2281 let swap_kib = Self::parse_proc_kib(&smaps, "Swap:").unwrap_or(0);
2282 let file_est_kib = rss_kib.saturating_sub(anon_kib);
2283 Some((
2284 rss_kib as f64 / 1024.0,
2285 anon_kib as f64 / 1024.0,
2286 file_est_kib as f64 / 1024.0,
2287 shared_clean_kib as f64 / 1024.0,
2288 shared_dirty_kib as f64 / 1024.0,
2289 private_clean_kib as f64 / 1024.0,
2290 private_dirty_kib as f64 / 1024.0,
2291 swap_kib as f64 / 1024.0,
2292 ))
2293 }
2294
2295 pub(crate) fn log_memory_stats(&self) {
2297 let rss = Self::rss_mb()
2298 .map(|v| format!("{:.1}", v))
2299 .unwrap_or_else(|| "N/A".into());
2300 let (vm_rss, vm_hwm, vm_data, vm_swap) = Self::proc_status_mb()
2301 .map(|(rss, hwm, data, swap)| {
2302 (
2303 format!("{rss:.1}"),
2304 format!("{hwm:.1}"),
2305 format!("{data:.1}"),
2306 format!("{swap:.1}"),
2307 )
2308 })
2309 .unwrap_or_else(|| ("N/A".into(), "N/A".into(), "N/A".into(), "N/A".into()));
2310 let (
2311 smaps_rss,
2312 smaps_anon,
2313 smaps_file_est,
2314 smaps_shared_clean,
2315 smaps_shared_dirty,
2316 smaps_private_clean,
2317 smaps_private_dirty,
2318 smaps_swap,
2319 ) = Self::smaps_rollup_mb()
2320 .map(
2321 |(
2322 rss,
2323 anon,
2324 file_est,
2325 shared_clean,
2326 shared_dirty,
2327 private_clean,
2328 private_dirty,
2329 swap,
2330 )| {
2331 (
2332 format!("{rss:.1}"),
2333 format!("{anon:.1}"),
2334 format!("{file_est:.1}"),
2335 format!("{shared_clean:.1}"),
2336 format!("{shared_dirty:.1}"),
2337 format!("{private_clean:.1}"),
2338 format!("{private_dirty:.1}"),
2339 format!("{swap:.1}"),
2340 )
2341 },
2342 )
2343 .unwrap_or_else(|| {
2344 (
2345 "N/A".into(),
2346 "N/A".into(),
2347 "N/A".into(),
2348 "N/A".into(),
2349 "N/A".into(),
2350 "N/A".into(),
2351 "N/A".into(),
2352 "N/A".into(),
2353 )
2354 });
2355 log::info!(
2356 "MEMSTATS rss_mb={} vmrss_mb={} vmhwm_mb={} vmdata_mb={} vmswap_mb={} smaps_rss_mb={} smaps_anon_mb={} smaps_file_est_mb={} smaps_shared_clean_mb={} smaps_shared_dirty_mb={} smaps_private_clean_mb={} smaps_private_dirty_mb={} smaps_swap_mb={} known_dest={} known_dest_cap_evict={} path={} path_cap_evict={} announce={} reverse={} link={} held_ann={} hashlist={} sig_cache={} ann_verify_q={} rate_lim={} blackhole={} tunnel={} ann_q_ifaces={} ann_q_nonempty={} ann_q_entries={} ann_q_bytes={} ann_q_iface_drop={} pr_tags={} disc_pr={} sent_pkt={} completed={} local_dest={} shared_ann={} lm_links={} hp_sessions={} proof_strat={}",
2357 rss,
2358 vm_rss,
2359 vm_hwm,
2360 vm_data,
2361 vm_swap,
2362 smaps_rss,
2363 smaps_anon,
2364 smaps_file_est,
2365 smaps_shared_clean,
2366 smaps_shared_dirty,
2367 smaps_private_clean,
2368 smaps_private_dirty,
2369 smaps_swap,
2370 self.known_destinations.len(),
2371 self.known_destinations_cap_evict_count,
2372 self.engine.path_table_count(),
2373 self.engine.path_destination_cap_evict_count(),
2374 self.engine.announce_table_count(),
2375 self.engine.reverse_table_count(),
2376 self.engine.link_table_count(),
2377 self.engine.held_announces_count(),
2378 self.engine.packet_hashlist_len(),
2379 self.engine.announce_sig_cache_len(),
2380 self.announce_verify_queue
2381 .lock()
2382 .map(|queue| queue.len())
2383 .unwrap_or(0),
2384 self.engine.rate_limiter_count(),
2385 self.engine.blackholed_count(),
2386 self.engine.tunnel_count(),
2387 self.engine.announce_queue_count(),
2388 self.engine.nonempty_announce_queue_count(),
2389 self.engine.queued_announce_count(),
2390 self.engine.queued_announce_bytes(),
2391 self.engine.announce_queue_interface_cap_drop_count(),
2392 self.engine.discovery_pr_tags_count(),
2393 self.engine.discovery_path_requests_count(),
2394 self.sent_packets.len(),
2395 self.completed_proofs.len(),
2396 self.local_destinations.len(),
2397 self.shared_announces.len(),
2398 self.link_manager.link_count(),
2399 self.holepunch_manager.session_count(),
2400 self.proof_strategies.len(),
2401 );
2402 }
2403
2404 pub(crate) fn tick_management_announces(&mut self, now: f64) {
2406 if self.transport_identity.is_none() {
2407 return;
2408 }
2409
2410 let uptime = now - self.started;
2411
2412 if !self.initial_announce_sent {
2414 if uptime < Self::MANAGEMENT_ANNOUNCE_DELAY {
2415 return;
2416 }
2417 self.initial_announce_sent = true;
2418 self.emit_management_announces(now);
2419 return;
2420 }
2421
2422 if now - self.last_management_announce >= self.management_announce_interval_secs {
2424 self.emit_management_announces(now);
2425 }
2426 }
2427
2428 pub(crate) fn emit_management_announces(&mut self, now: f64) {
2430 use crate::management;
2431
2432 self.last_management_announce = now;
2433
2434 let identity = match self.transport_identity {
2435 Some(ref id) => id,
2436 None => return,
2437 };
2438
2439 let mgmt_raw = if self.management_config.enable_remote_management {
2441 management::build_management_announce(identity, &mut self.rng)
2442 } else {
2443 None
2444 };
2445
2446 let bh_raw = if self.management_config.publish_blackhole {
2447 management::build_blackhole_announce(identity, &mut self.rng)
2448 } else {
2449 None
2450 };
2451
2452 let probe_raw = if self.probe_responder_hash.is_some() {
2453 management::build_probe_announce(identity, &mut self.rng)
2454 } else {
2455 None
2456 };
2457
2458 self.emit_local_management_announce(mgmt_raw, now, "management destination");
2459 self.emit_local_management_announce(bh_raw, now, "blackhole info");
2460 self.emit_local_management_announce(probe_raw, now, "probe responder");
2461 }
2462
2463 fn emit_local_management_announce(&mut self, raw: Option<Vec<u8>>, now: f64, label: &str) {
2464 if let Some(raw) = raw {
2465 if let Ok(packet) = RawPacket::unpack(&raw) {
2466 self.engine.register_destination(
2467 packet.destination_hash,
2468 rns_core::constants::DESTINATION_SINGLE,
2469 );
2470 let actions = self.engine.handle_outbound(
2471 &packet,
2472 rns_core::constants::DESTINATION_SINGLE,
2473 None,
2474 now,
2475 );
2476 self.dispatch_all(actions);
2477 log::debug!("Emitted {} announce", label);
2478 }
2479 }
2480 }
2481
2482 pub(crate) fn handle_management_request(
2484 &mut self,
2485 link_id: [u8; 16],
2486 path_hash: [u8; 16],
2487 data: Vec<u8>,
2488 request_id: [u8; 16],
2489 remote_identity: Option<([u8; 16], [u8; 64])>,
2490 ) {
2491 use crate::management;
2492
2493 let is_restricted = path_hash == management::status_path_hash()
2495 || path_hash == management::path_path_hash();
2496
2497 if is_restricted && !self.management_config.remote_management_allowed.is_empty() {
2498 match remote_identity {
2499 Some((identity_hash, _)) => {
2500 if !self
2501 .management_config
2502 .remote_management_allowed
2503 .contains(&identity_hash)
2504 {
2505 log::debug!("Management request denied: identity not in allowed list");
2506 return;
2507 }
2508 }
2509 None => {
2510 log::debug!("Management request denied: peer not identified");
2511 return;
2512 }
2513 }
2514 }
2515
2516 let response_data = if path_hash == management::status_path_hash() {
2517 {
2518 let views: Vec<&dyn management::InterfaceStatusView> = self
2519 .interfaces
2520 .values()
2521 .map(|e| e as &dyn management::InterfaceStatusView)
2522 .collect();
2523 management::handle_status_request(
2524 &data,
2525 &self.engine,
2526 &views,
2527 self.started,
2528 self.probe_responder_hash,
2529 )
2530 }
2531 } else if path_hash == management::path_path_hash() {
2532 management::handle_path_request(&data, &self.engine)
2533 } else if path_hash == management::list_path_hash() {
2534 management::handle_blackhole_list_request(&self.engine)
2535 } else {
2536 log::warn!("Unknown management path_hash: {:02x?}", &path_hash[..4]);
2537 None
2538 };
2539
2540 if let Some(response) = response_data {
2541 let actions = self.link_manager.send_management_response(
2542 &link_id,
2543 &request_id,
2544 &response,
2545 &mut self.rng,
2546 );
2547 self.dispatch_link_actions(actions);
2548 }
2549 }
2550}