1use std::collections::HashMap;
10
11use super::compressor::Bzip2Compressor;
12use rns_core::channel::{Channel, Sequence};
13use rns_core::constants;
14use rns_core::link::types::{LinkId, LinkState, TeardownReason};
15use rns_core::link::{LinkAction, LinkEngine, LinkMode};
16use rns_core::packet::{PacketFlags, RawPacket};
17use rns_core::resource::{ResourceAction, ResourceReceiver, ResourceSender};
18use rns_crypto::ed25519::Ed25519PrivateKey;
19use rns_crypto::Rng;
20
21use super::time;
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum ResourceStrategy {
26 AcceptNone,
28 AcceptAll,
30 AcceptApp,
32}
33
34impl Default for ResourceStrategy {
35 fn default() -> Self {
36 ResourceStrategy::AcceptNone
37 }
38}
39
40struct ManagedLink {
42 engine: LinkEngine,
43 channel: Option<Channel>,
44 pending_channel_packets: HashMap<[u8; 32], Sequence>,
45 channel_send_ok: u64,
46 channel_send_not_ready: u64,
47 channel_send_too_big: u64,
48 channel_send_other_error: u64,
49 channel_messages_received: u64,
50 channel_proofs_sent: u64,
51 channel_proofs_received: u64,
52 dest_hash: [u8; 16],
54 remote_identity: Option<([u8; 16], [u8; 64])>,
56 dest_sig_pub_bytes: Option<[u8; 32]>,
58 incoming_resources: Vec<ResourceReceiver>,
60 outgoing_resources: Vec<ResourceSender>,
62 incoming_splits: HashMap<[u8; 32], IncomingSplitTransfer>,
64 outgoing_splits: HashMap<[u8; 32], OutgoingSplitTransfer>,
66 resource_strategy: ResourceStrategy,
68 route_interface: Option<rns_core::transport::types::InterfaceId>,
70 route_transport_id: Option<[u8; 16]>,
75}
76
77struct IncomingSplitTransfer {
78 total_segments: u64,
79 completed_segments: u64,
80 current_segment_index: u64,
81 current_received_parts: usize,
82 current_total_parts: usize,
83 data: Vec<u8>,
84 metadata: Option<Vec<u8>>,
85 is_response: bool,
86}
87
88struct OutgoingSplitTransfer {
89 total_segments: u64,
90 completed_segments: u64,
91 current_segment_index: u64,
92 current_sent_parts: usize,
93 current_total_parts: usize,
94}
95
96struct LinkDestination {
98 sig_prv: Ed25519PrivateKey,
99 sig_pub_bytes: [u8; 32],
100 resource_strategy: ResourceStrategy,
101}
102
103#[derive(Debug, Clone, PartialEq, Eq)]
105pub enum RequestResponse {
106 Bytes(Vec<u8>),
108 Resource {
110 data: Vec<u8>,
111 metadata: Option<Vec<u8>>,
112 auto_compress: bool,
113 },
114}
115
116impl From<Vec<u8>> for RequestResponse {
117 fn from(data: Vec<u8>) -> Self {
118 RequestResponse::Bytes(data)
119 }
120}
121
122struct RequestHandlerEntry {
124 path: String,
126 path_hash: [u8; 16],
128 allowed_list: Option<Vec<[u8; 16]>>,
130 handler: Box<
132 dyn Fn(LinkId, &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<RequestResponse>
133 + Send,
134 >,
135}
136
137#[derive(Debug)]
139pub enum LinkManagerAction {
140 SendPacket {
142 raw: Vec<u8>,
143 dest_type: u8,
144 attached_interface: Option<rns_core::transport::types::InterfaceId>,
145 },
146 LinkEstablished {
148 link_id: LinkId,
149 dest_hash: [u8; 16],
150 rtt: f64,
151 is_initiator: bool,
152 },
153 LinkClosed {
155 link_id: LinkId,
156 reason: Option<TeardownReason>,
157 },
158 RemoteIdentified {
160 link_id: LinkId,
161 identity_hash: [u8; 16],
162 public_key: [u8; 64],
163 },
164 RegisterLinkDest { link_id: LinkId },
166 DeregisterLinkDest { link_id: LinkId },
168 ManagementRequest {
171 link_id: LinkId,
172 path_hash: [u8; 16],
173 data: Vec<u8>,
175 request_id: [u8; 16],
177 remote_identity: Option<([u8; 16], [u8; 64])>,
178 },
179 ResourceReceived {
181 link_id: LinkId,
182 data: Vec<u8>,
183 metadata: Option<Vec<u8>>,
184 },
185 ResourceCompleted { link_id: LinkId },
187 ResourceFailed { link_id: LinkId, error: String },
189 ResourceProgress {
191 link_id: LinkId,
192 received: usize,
193 total: usize,
194 },
195 ResourceAcceptQuery {
197 link_id: LinkId,
198 resource_hash: Vec<u8>,
199 transfer_size: u64,
200 has_metadata: bool,
201 },
202 ChannelMessageReceived {
204 link_id: LinkId,
205 msgtype: u16,
206 payload: Vec<u8>,
207 },
208 LinkDataReceived {
210 link_id: LinkId,
211 context: u8,
212 data: Vec<u8>,
213 },
214 ResponseReceived {
216 link_id: LinkId,
217 request_id: [u8; 16],
218 data: Vec<u8>,
219 metadata: Option<Vec<u8>>,
220 },
221 LinkRequestReceived {
223 link_id: LinkId,
224 receiving_interface: rns_core::transport::types::InterfaceId,
225 },
226}
227
228pub struct LinkManager {
230 links: HashMap<LinkId, ManagedLink>,
231 link_destinations: HashMap<[u8; 16], LinkDestination>,
232 request_handlers: Vec<RequestHandlerEntry>,
233 management_paths: Vec<[u8; 16]>,
236}
237
238#[derive(Debug, Clone, Copy, PartialEq, Eq)]
239pub struct LinkRouteHint {
240 pub interface: rns_core::transport::types::InterfaceId,
241 pub transport_id: Option<[u8; 16]>,
242}
243
244impl LinkManager {
245 fn resource_sdu_for_link(link: &ManagedLink) -> usize {
246 let mtu = link.engine.mtu() as usize;
249 let derived = mtu.saturating_sub(constants::HEADER_MAXSIZE + constants::IFAC_MIN_SIZE);
250 if derived > 0 {
251 derived
252 } else {
253 constants::RESOURCE_SDU
254 }
255 }
256
257 fn split_progress_parts(
258 segment_index: u64,
259 total_segments: u64,
260 current_done: usize,
261 current_total: usize,
262 sdu: usize,
263 ) -> (usize, usize) {
264 let max_parts_per_segment = constants::RESOURCE_MAX_EFFICIENT_SIZE.div_ceil(sdu.max(1));
265 let total = (total_segments as usize).saturating_mul(max_parts_per_segment);
266 let completed_segments = segment_index.saturating_sub(1) as usize;
267 let completed = completed_segments.saturating_mul(max_parts_per_segment);
268 let current = if current_total == 0 {
269 0
270 } else if current_total < max_parts_per_segment {
271 let scaled =
272 (current_done as f64) * (max_parts_per_segment as f64 / current_total as f64);
273 scaled.floor() as usize
274 } else {
275 current_done
276 };
277 (completed.saturating_add(current).min(total), total)
278 }
279
280 fn resource_hash_key(hash: &[u8]) -> Option<[u8; 32]> {
281 let mut key = [0u8; 32];
282 if hash.len() != key.len() {
283 return None;
284 }
285 key.copy_from_slice(hash);
286 Some(key)
287 }
288
289 fn incoming_split_progress(split: &IncomingSplitTransfer, sdu: usize) -> (usize, usize) {
290 Self::split_progress_parts(
291 split.current_segment_index,
292 split.total_segments,
293 split.current_received_parts,
294 split.current_total_parts,
295 sdu,
296 )
297 }
298
299 fn outgoing_split_progress(split: &OutgoingSplitTransfer, sdu: usize) -> (usize, usize) {
300 Self::split_progress_parts(
301 split.current_segment_index,
302 split.total_segments,
303 split.current_sent_parts,
304 split.current_total_parts,
305 sdu,
306 )
307 }
308
309 pub fn new() -> Self {
311 LinkManager {
312 links: HashMap::new(),
313 link_destinations: HashMap::new(),
314 request_handlers: Vec::new(),
315 management_paths: Vec::new(),
316 }
317 }
318
319 pub fn register_management_path(&mut self, path_hash: [u8; 16]) {
323 if !self.management_paths.contains(&path_hash) {
324 self.management_paths.push(path_hash);
325 }
326 }
327
328 pub fn get_derived_key(&self, link_id: &LinkId) -> Option<Vec<u8>> {
330 self.links
331 .get(link_id)
332 .and_then(|link| link.engine.derived_key().map(|dk| dk.to_vec()))
333 }
334
335 pub fn get_link_route_hint(&self, link_id: &LinkId) -> Option<LinkRouteHint> {
337 self.links.get(link_id).and_then(|link| {
338 link.route_interface.map(|interface| LinkRouteHint {
339 interface,
340 transport_id: link.route_transport_id,
341 })
342 })
343 }
344
345 pub fn set_link_route_hint(
347 &mut self,
348 link_id: &LinkId,
349 interface: rns_core::transport::types::InterfaceId,
350 transport_id: Option<[u8; 16]>,
351 ) -> bool {
352 let Some(link) = self.links.get_mut(link_id) else {
353 return false;
354 };
355 link.route_interface = Some(interface);
356 link.route_transport_id = transport_id;
357 true
358 }
359
360 pub fn register_link_destination(
362 &mut self,
363 dest_hash: [u8; 16],
364 sig_prv: Ed25519PrivateKey,
365 sig_pub_bytes: [u8; 32],
366 resource_strategy: ResourceStrategy,
367 ) {
368 self.link_destinations.insert(
369 dest_hash,
370 LinkDestination {
371 sig_prv,
372 sig_pub_bytes,
373 resource_strategy,
374 },
375 );
376 }
377
378 pub fn deregister_link_destination(&mut self, dest_hash: &[u8; 16]) {
380 self.link_destinations.remove(dest_hash);
381 }
382
383 pub fn register_request_handler<F>(
389 &mut self,
390 path: &str,
391 allowed_list: Option<Vec<[u8; 16]>>,
392 handler: F,
393 ) where
394 F: Fn(LinkId, &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<Vec<u8>>
395 + Send
396 + 'static,
397 {
398 let path_hash = compute_path_hash(path);
399 self.request_handlers.push(RequestHandlerEntry {
400 path: path.to_string(),
401 path_hash,
402 allowed_list,
403 handler: Box::new(move |link_id, p, data, remote| {
404 handler(link_id, p, data, remote).map(RequestResponse::Bytes)
405 }),
406 });
407 }
408
409 pub fn register_request_handler_response<F>(
411 &mut self,
412 path: &str,
413 allowed_list: Option<Vec<[u8; 16]>>,
414 handler: F,
415 ) where
416 F: Fn(LinkId, &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<RequestResponse>
417 + Send
418 + 'static,
419 {
420 let path_hash = compute_path_hash(path);
421 self.request_handlers.push(RequestHandlerEntry {
422 path: path.to_string(),
423 path_hash,
424 allowed_list,
425 handler: Box::new(handler),
426 });
427 }
428
429 pub fn create_link(
437 &mut self,
438 dest_hash: &[u8; 16],
439 dest_sig_pub_bytes: &[u8; 32],
440 hops: u8,
441 mtu: u32,
442 rng: &mut dyn Rng,
443 ) -> (LinkId, Vec<LinkManagerAction>) {
444 let mode = LinkMode::Aes256Cbc;
445 let (mut engine, request_data) =
446 LinkEngine::new_initiator(dest_hash, hops, mode, Some(mtu), time::now(), rng);
447
448 let flags = PacketFlags {
450 header_type: constants::HEADER_1,
451 context_flag: constants::FLAG_UNSET,
452 transport_type: constants::TRANSPORT_BROADCAST,
453 destination_type: constants::DESTINATION_SINGLE,
454 packet_type: constants::PACKET_TYPE_LINKREQUEST,
455 };
456
457 let packet = match RawPacket::pack(
458 flags,
459 0,
460 dest_hash,
461 None,
462 constants::CONTEXT_NONE,
463 &request_data,
464 ) {
465 Ok(p) => p,
466 Err(_) => {
467 return ([0u8; 16], Vec::new());
469 }
470 };
471
472 engine.set_link_id_from_hashable(&packet.get_hashable_part(), request_data.len());
473 let link_id = *engine.link_id();
474
475 let managed = ManagedLink {
476 engine,
477 channel: None,
478 pending_channel_packets: HashMap::new(),
479 channel_send_ok: 0,
480 channel_send_not_ready: 0,
481 channel_send_too_big: 0,
482 channel_send_other_error: 0,
483 channel_messages_received: 0,
484 channel_proofs_sent: 0,
485 channel_proofs_received: 0,
486 dest_hash: *dest_hash,
487 remote_identity: None,
488 dest_sig_pub_bytes: Some(*dest_sig_pub_bytes),
489 incoming_resources: Vec::new(),
490 outgoing_resources: Vec::new(),
491 incoming_splits: HashMap::new(),
492 outgoing_splits: HashMap::new(),
493 resource_strategy: ResourceStrategy::default(),
494 route_interface: None,
495 route_transport_id: None,
496 };
497 self.links.insert(link_id, managed);
498
499 let mut actions = Vec::new();
500 actions.push(LinkManagerAction::RegisterLinkDest { link_id });
502 actions.push(LinkManagerAction::SendPacket {
504 raw: packet.raw,
505 dest_type: constants::DESTINATION_LINK,
506 attached_interface: None,
507 });
508
509 (link_id, actions)
510 }
511
512 pub fn handle_local_delivery(
518 &mut self,
519 dest_hash: [u8; 16],
520 raw: &[u8],
521 packet_hash: [u8; 32],
522 receiving_interface: rns_core::transport::types::InterfaceId,
523 rng: &mut dyn Rng,
524 ) -> Vec<LinkManagerAction> {
525 let packet = match RawPacket::unpack(raw) {
526 Ok(p) => p,
527 Err(_) => return Vec::new(),
528 };
529
530 match packet.flags.packet_type {
531 constants::PACKET_TYPE_LINKREQUEST => {
532 self.handle_linkrequest(&dest_hash, &packet, receiving_interface, rng)
533 }
534 constants::PACKET_TYPE_PROOF if packet.context == constants::CONTEXT_LRPROOF => {
535 self.handle_lrproof(&dest_hash, &packet, receiving_interface, rng)
537 }
538 constants::PACKET_TYPE_PROOF if packet.context == constants::CONTEXT_RESOURCE_PRF => {
539 self.handle_resource_prf(&dest_hash, &packet.data, rng)
542 }
543 constants::PACKET_TYPE_PROOF => self.handle_link_proof(&dest_hash, &packet, rng),
544 constants::PACKET_TYPE_DATA => {
545 self.handle_link_data(&dest_hash, &packet, packet_hash, receiving_interface, rng)
546 }
547 _ => Vec::new(),
548 }
549 }
550
551 fn handle_linkrequest(
553 &mut self,
554 dest_hash: &[u8; 16],
555 packet: &RawPacket,
556 receiving_interface: rns_core::transport::types::InterfaceId,
557 rng: &mut dyn Rng,
558 ) -> Vec<LinkManagerAction> {
559 let ld = match self.link_destinations.get(dest_hash) {
561 Some(ld) => ld,
562 None => return Vec::new(),
563 };
564
565 let hashable = packet.get_hashable_part();
566 let now = time::now();
567
568 let (engine, lrproof_data) = match LinkEngine::new_responder(
570 &ld.sig_prv,
571 &ld.sig_pub_bytes,
572 &packet.data,
573 &hashable,
574 dest_hash,
575 packet.hops,
576 now,
577 rng,
578 ) {
579 Ok(r) => r,
580 Err(e) => {
581 log::debug!("LINKREQUEST rejected: {}", e);
582 return Vec::new();
583 }
584 };
585
586 let link_id = *engine.link_id();
587 log::debug!(
588 "LINKREQUEST accepted: link={:02x?} iface={} header_type={} transport_id_present={} hops={}",
589 &link_id[..4],
590 receiving_interface.0,
591 packet.flags.header_type,
592 packet.transport_id.is_some(),
593 packet.hops
594 );
595
596 let managed = ManagedLink {
597 engine,
598 channel: None,
599 pending_channel_packets: HashMap::new(),
600 channel_send_ok: 0,
601 channel_send_not_ready: 0,
602 channel_send_too_big: 0,
603 channel_send_other_error: 0,
604 channel_messages_received: 0,
605 channel_proofs_sent: 0,
606 channel_proofs_received: 0,
607 dest_hash: *dest_hash,
608 remote_identity: None,
609 dest_sig_pub_bytes: None,
610 incoming_resources: Vec::new(),
611 outgoing_resources: Vec::new(),
612 incoming_splits: HashMap::new(),
613 outgoing_splits: HashMap::new(),
614 resource_strategy: ld.resource_strategy,
615 route_interface: Some(receiving_interface),
616 route_transport_id: if packet.flags.header_type == constants::HEADER_2 {
617 packet.transport_id
618 } else {
619 None
620 },
621 };
622 self.links.insert(link_id, managed);
623
624 let flags = PacketFlags {
626 header_type: constants::HEADER_1,
627 context_flag: constants::FLAG_UNSET,
628 transport_type: constants::TRANSPORT_BROADCAST,
629 destination_type: constants::DESTINATION_LINK,
630 packet_type: constants::PACKET_TYPE_PROOF,
631 };
632
633 let mut actions = Vec::new();
634
635 actions.push(LinkManagerAction::RegisterLinkDest { link_id });
637
638 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
639 flags,
640 packet.hops,
641 &link_id,
642 None,
643 constants::CONTEXT_LRPROOF,
644 &lrproof_data,
645 ) {
646 log::debug!(
647 "LRPROOF queued: link={:02x?} route_iface={} route_tid_present={} hops={}",
648 &link_id[..4],
649 receiving_interface.0,
650 packet.transport_id.is_some(),
651 packet.hops
652 );
653 actions.push(LinkManagerAction::SendPacket {
654 raw,
655 dest_type: constants::DESTINATION_LINK,
656 attached_interface: None,
657 });
658 }
659
660 if packet.hops != 0 {
663 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
664 flags,
665 0,
666 &link_id,
667 None,
668 constants::CONTEXT_LRPROOF,
669 &lrproof_data,
670 ) {
671 log::debug!(
672 "LRPROOF fallback queued: link={:02x?} route_iface={} hops=0",
673 &link_id[..4],
674 receiving_interface.0
675 );
676 actions.push(LinkManagerAction::SendPacket {
677 raw,
678 dest_type: constants::DESTINATION_LINK,
679 attached_interface: None,
680 });
681 }
682 }
683
684 if packet.hops < u8::MAX {
688 let hops_plus_one = packet.hops + 1;
689 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
690 flags,
691 hops_plus_one,
692 &link_id,
693 None,
694 constants::CONTEXT_LRPROOF,
695 &lrproof_data,
696 ) {
697 log::debug!(
698 "LRPROOF +1 queued: link={:02x?} route_iface={} hops={}",
699 &link_id[..4],
700 receiving_interface.0,
701 hops_plus_one
702 );
703 actions.push(LinkManagerAction::SendPacket {
704 raw,
705 dest_type: constants::DESTINATION_LINK,
706 attached_interface: None,
707 });
708 }
709 }
710
711 actions.push(LinkManagerAction::LinkRequestReceived {
713 link_id,
714 receiving_interface,
715 });
716
717 actions
718 }
719
720 fn handle_link_proof(
721 &mut self,
722 link_id: &LinkId,
723 packet: &RawPacket,
724 rng: &mut dyn Rng,
725 ) -> Vec<LinkManagerAction> {
726 if packet.data.len() < 32 {
727 return Vec::new();
728 }
729
730 let mut tracked_hash = [0u8; 32];
731 tracked_hash.copy_from_slice(&packet.data[..32]);
732
733 let Some(link) = self.links.get_mut(link_id) else {
734 return Vec::new();
735 };
736 let Some(sequence) = link.pending_channel_packets.remove(&tracked_hash) else {
737 return Vec::new();
738 };
739 link.channel_proofs_received += 1;
740 let Some(channel) = link.channel.as_mut() else {
741 return Vec::new();
742 };
743
744 let chan_actions = channel.packet_delivered(sequence);
745 let _ = channel;
746 let _ = link;
747 self.process_channel_actions(link_id, chan_actions, rng)
748 }
749
750 fn build_link_packet_proof(
751 &mut self,
752 link_id: &LinkId,
753 packet_hash: &[u8; 32],
754 ) -> Vec<LinkManagerAction> {
755 let dest_hash = match self.links.get(link_id) {
756 Some(link) => link.dest_hash,
757 None => return Vec::new(),
758 };
759 let Some(ld) = self.link_destinations.get(&dest_hash) else {
760 return Vec::new();
761 };
762 if let Some(link) = self.links.get_mut(link_id) {
763 link.channel_proofs_sent += 1;
764 }
765
766 let signature = ld.sig_prv.sign(packet_hash);
767 let mut proof_data = Vec::with_capacity(96);
768 proof_data.extend_from_slice(packet_hash);
769 proof_data.extend_from_slice(&signature);
770
771 let flags = PacketFlags {
772 header_type: constants::HEADER_1,
773 context_flag: constants::FLAG_UNSET,
774 transport_type: constants::TRANSPORT_BROADCAST,
775 destination_type: constants::DESTINATION_LINK,
776 packet_type: constants::PACKET_TYPE_PROOF,
777 };
778 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
779 flags,
780 0,
781 link_id,
782 None,
783 constants::CONTEXT_NONE,
784 &proof_data,
785 ) {
786 vec![LinkManagerAction::SendPacket {
787 raw,
788 dest_type: constants::DESTINATION_LINK,
789 attached_interface: None,
790 }]
791 } else {
792 Vec::new()
793 }
794 }
795
796 fn handle_lrproof(
798 &mut self,
799 link_id_bytes: &[u8; 16],
800 packet: &RawPacket,
801 receiving_interface: rns_core::transport::types::InterfaceId,
802 rng: &mut dyn Rng,
803 ) -> Vec<LinkManagerAction> {
804 let link = match self.links.get_mut(link_id_bytes) {
805 Some(l) => l,
806 None => return Vec::new(),
807 };
808
809 link.route_interface = Some(receiving_interface);
810 if packet.flags.header_type == constants::HEADER_2 {
811 if let Some(transport_id) = packet.transport_id {
812 link.route_transport_id = Some(transport_id);
813 }
814 }
815 log::debug!(
816 "LRPROOF received: link={:02x?} iface={} header_type={} transport_id_present={}",
817 &link_id_bytes[..4],
818 receiving_interface.0,
819 packet.flags.header_type,
820 packet.transport_id.is_some()
821 );
822
823 if link.engine.state() != LinkState::Pending || !link.engine.is_initiator() {
824 return Vec::new();
825 }
826
827 let dest_sig_pub_bytes = match link.dest_sig_pub_bytes {
829 Some(b) => b,
830 None => {
831 log::debug!("LRPROOF: no destination signing key available");
832 return Vec::new();
833 }
834 };
835
836 let now = time::now();
837 let (lrrtt_encrypted, link_actions) =
838 match link
839 .engine
840 .handle_lrproof(&packet.data, &dest_sig_pub_bytes, now, rng)
841 {
842 Ok(r) => r,
843 Err(e) => {
844 log::debug!("LRPROOF validation failed: {}", e);
845 return Vec::new();
846 }
847 };
848
849 let link_id = *link.engine.link_id();
850 let mut actions = Vec::new();
851
852 actions.extend(self.process_link_actions(&link_id, &link_actions));
854
855 let flags = PacketFlags {
857 header_type: constants::HEADER_1,
858 context_flag: constants::FLAG_UNSET,
859 transport_type: constants::TRANSPORT_BROADCAST,
860 destination_type: constants::DESTINATION_LINK,
861 packet_type: constants::PACKET_TYPE_DATA,
862 };
863
864 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
865 flags,
866 0,
867 &link_id,
868 None,
869 constants::CONTEXT_LRRTT,
870 &lrrtt_encrypted,
871 ) {
872 actions.push(LinkManagerAction::SendPacket {
873 raw,
874 dest_type: constants::DESTINATION_LINK,
875 attached_interface: None,
876 });
877 }
878
879 if let Some(link) = self.links.get_mut(&link_id) {
881 if link.engine.state() == LinkState::Active {
882 let rtt = link.engine.rtt().unwrap_or(1.0);
883 link.channel = Some(Channel::new(rtt));
884 }
885 }
886
887 actions
888 }
889
890 fn handle_link_data(
896 &mut self,
897 link_id_bytes: &[u8; 16],
898 packet: &RawPacket,
899 packet_hash: [u8; 32],
900 receiving_interface: rns_core::transport::types::InterfaceId,
901 rng: &mut dyn Rng,
902 ) -> Vec<LinkManagerAction> {
903 enum LinkDataResult<'a> {
905 Lrrtt {
906 link_id: LinkId,
907 link_actions: Vec<LinkAction>,
908 },
909 Identify {
910 link_id: LinkId,
911 link_actions: Vec<LinkAction>,
912 },
913 Keepalive {
914 link_id: LinkId,
915 inbound_actions: Vec<LinkAction>,
916 },
917 LinkClose {
918 link_id: LinkId,
919 teardown_actions: Vec<LinkAction>,
920 },
921 Channel {
922 link_id: LinkId,
923 inbound_actions: Vec<LinkAction>,
924 plaintext: Vec<u8>,
925 packet_hash: [u8; 32],
926 },
927 Request {
928 link_id: LinkId,
929 inbound_actions: Vec<LinkAction>,
930 plaintext: Vec<u8>,
931 request_id: [u8; 16],
932 },
933 Response {
934 link_id: LinkId,
935 inbound_actions: Vec<LinkAction>,
936 plaintext: Vec<u8>,
937 },
938 Generic {
939 link_id: LinkId,
940 inbound_actions: Vec<LinkAction>,
941 plaintext: Vec<u8>,
942 context: u8,
943 packet_hash: [u8; 32],
944 },
945 ResourceAdv {
947 link_id: LinkId,
948 inbound_actions: Vec<LinkAction>,
949 plaintext: Vec<u8>,
950 },
951 ResourceReq {
953 link_id: LinkId,
954 inbound_actions: Vec<LinkAction>,
955 plaintext: Vec<u8>,
956 },
957 ResourceHmu {
959 link_id: LinkId,
960 inbound_actions: Vec<LinkAction>,
961 plaintext: Vec<u8>,
962 },
963 ResourcePart {
965 link_id: LinkId,
966 inbound_actions: Vec<LinkAction>,
967 raw_data: &'a [u8],
968 },
969 ResourcePrf {
971 link_id: LinkId,
972 inbound_actions: Vec<LinkAction>,
973 plaintext: Vec<u8>,
974 },
975 ResourceIcl {
977 link_id: LinkId,
978 inbound_actions: Vec<LinkAction>,
979 },
980 ResourceRcl {
982 link_id: LinkId,
983 inbound_actions: Vec<LinkAction>,
984 },
985 Error,
986 }
987
988 let result = {
989 let link = match self.links.get_mut(link_id_bytes) {
990 Some(l) => l,
991 None => return Vec::new(),
992 };
993
994 link.route_interface = Some(receiving_interface);
995 if packet.flags.header_type == constants::HEADER_2 {
996 if let Some(transport_id) = packet.transport_id {
997 link.route_transport_id = Some(transport_id);
998 }
999 } else {
1000 link.route_transport_id = None;
1001 }
1002
1003 match packet.context {
1004 constants::CONTEXT_LRRTT => {
1005 match link.engine.handle_lrrtt(&packet.data, time::now()) {
1006 Ok(link_actions) => {
1007 let link_id = *link.engine.link_id();
1008 LinkDataResult::Lrrtt {
1009 link_id,
1010 link_actions,
1011 }
1012 }
1013 Err(e) => {
1014 log::debug!("LRRTT handling failed: {}", e);
1015 LinkDataResult::Error
1016 }
1017 }
1018 }
1019 constants::CONTEXT_LINKIDENTIFY => {
1020 match link.engine.handle_identify(&packet.data) {
1021 Ok(link_actions) => {
1022 let link_id = *link.engine.link_id();
1023 link.remote_identity = link.engine.remote_identity().cloned();
1024 LinkDataResult::Identify {
1025 link_id,
1026 link_actions,
1027 }
1028 }
1029 Err(e) => {
1030 log::debug!("LINKIDENTIFY failed: {}", e);
1031 LinkDataResult::Error
1032 }
1033 }
1034 }
1035 constants::CONTEXT_KEEPALIVE => {
1036 let inbound_actions = link.engine.record_inbound(time::now());
1037 let link_id = *link.engine.link_id();
1038 LinkDataResult::Keepalive {
1039 link_id,
1040 inbound_actions,
1041 }
1042 }
1043 constants::CONTEXT_LINKCLOSE => {
1044 let teardown_actions = link.engine.handle_teardown();
1045 let link_id = *link.engine.link_id();
1046 LinkDataResult::LinkClose {
1047 link_id,
1048 teardown_actions,
1049 }
1050 }
1051 constants::CONTEXT_CHANNEL => match link.engine.decrypt(&packet.data) {
1052 Ok(plaintext) => {
1053 let inbound_actions = link.engine.record_inbound(time::now());
1054 let link_id = *link.engine.link_id();
1055 LinkDataResult::Channel {
1056 link_id,
1057 inbound_actions,
1058 plaintext,
1059 packet_hash,
1060 }
1061 }
1062 Err(_) => LinkDataResult::Error,
1063 },
1064 constants::CONTEXT_REQUEST => match link.engine.decrypt(&packet.data) {
1065 Ok(plaintext) => {
1066 let inbound_actions = link.engine.record_inbound(time::now());
1067 let link_id = *link.engine.link_id();
1068 let request_id = packet.get_truncated_hash();
1069 LinkDataResult::Request {
1070 link_id,
1071 inbound_actions,
1072 plaintext,
1073 request_id,
1074 }
1075 }
1076 Err(_) => LinkDataResult::Error,
1077 },
1078 constants::CONTEXT_RESPONSE => match link.engine.decrypt(&packet.data) {
1079 Ok(plaintext) => {
1080 let inbound_actions = link.engine.record_inbound(time::now());
1081 let link_id = *link.engine.link_id();
1082 LinkDataResult::Response {
1083 link_id,
1084 inbound_actions,
1085 plaintext,
1086 }
1087 }
1088 Err(_) => LinkDataResult::Error,
1089 },
1090 constants::CONTEXT_RESOURCE_ADV => match link.engine.decrypt(&packet.data) {
1092 Ok(plaintext) => {
1093 let inbound_actions = link.engine.record_inbound(time::now());
1094 let link_id = *link.engine.link_id();
1095 LinkDataResult::ResourceAdv {
1096 link_id,
1097 inbound_actions,
1098 plaintext,
1099 }
1100 }
1101 Err(_) => LinkDataResult::Error,
1102 },
1103 constants::CONTEXT_RESOURCE_REQ => match link.engine.decrypt(&packet.data) {
1104 Ok(plaintext) => {
1105 let inbound_actions = link.engine.record_inbound(time::now());
1106 let link_id = *link.engine.link_id();
1107 LinkDataResult::ResourceReq {
1108 link_id,
1109 inbound_actions,
1110 plaintext,
1111 }
1112 }
1113 Err(_) => LinkDataResult::Error,
1114 },
1115 constants::CONTEXT_RESOURCE_HMU => match link.engine.decrypt(&packet.data) {
1116 Ok(plaintext) => {
1117 let inbound_actions = link.engine.record_inbound(time::now());
1118 let link_id = *link.engine.link_id();
1119 LinkDataResult::ResourceHmu {
1120 link_id,
1121 inbound_actions,
1122 plaintext,
1123 }
1124 }
1125 Err(_) => LinkDataResult::Error,
1126 },
1127 constants::CONTEXT_RESOURCE => {
1128 let inbound_actions = link.engine.record_inbound(time::now());
1130 let link_id = *link.engine.link_id();
1131 LinkDataResult::ResourcePart {
1132 link_id,
1133 inbound_actions,
1134 raw_data: &packet.data,
1135 }
1136 }
1137 constants::CONTEXT_RESOURCE_PRF => match link.engine.decrypt(&packet.data) {
1138 Ok(plaintext) => {
1139 let inbound_actions = link.engine.record_inbound(time::now());
1140 let link_id = *link.engine.link_id();
1141 LinkDataResult::ResourcePrf {
1142 link_id,
1143 inbound_actions,
1144 plaintext,
1145 }
1146 }
1147 Err(_) => LinkDataResult::Error,
1148 },
1149 constants::CONTEXT_RESOURCE_ICL => {
1150 let _ = link.engine.decrypt(&packet.data); let inbound_actions = link.engine.record_inbound(time::now());
1152 let link_id = *link.engine.link_id();
1153 LinkDataResult::ResourceIcl {
1154 link_id,
1155 inbound_actions,
1156 }
1157 }
1158 constants::CONTEXT_RESOURCE_RCL => {
1159 let _ = link.engine.decrypt(&packet.data); let inbound_actions = link.engine.record_inbound(time::now());
1161 let link_id = *link.engine.link_id();
1162 LinkDataResult::ResourceRcl {
1163 link_id,
1164 inbound_actions,
1165 }
1166 }
1167 _ => match link.engine.decrypt(&packet.data) {
1168 Ok(plaintext) => {
1169 let inbound_actions = link.engine.record_inbound(time::now());
1170 let link_id = *link.engine.link_id();
1171 LinkDataResult::Generic {
1172 link_id,
1173 inbound_actions,
1174 plaintext,
1175 context: packet.context,
1176 packet_hash,
1177 }
1178 }
1179 Err(_) => LinkDataResult::Error,
1180 },
1181 }
1182 }; let mut actions = Vec::new();
1186 match result {
1187 LinkDataResult::Lrrtt {
1188 link_id,
1189 link_actions,
1190 } => {
1191 actions.extend(self.process_link_actions(&link_id, &link_actions));
1192 if let Some(link) = self.links.get_mut(&link_id) {
1194 if link.engine.state() == LinkState::Active {
1195 let rtt = link.engine.rtt().unwrap_or(1.0);
1196 link.channel = Some(Channel::new(rtt));
1197 }
1198 }
1199 }
1200 LinkDataResult::Identify {
1201 link_id,
1202 link_actions,
1203 } => {
1204 actions.extend(self.process_link_actions(&link_id, &link_actions));
1205 }
1206 LinkDataResult::Keepalive {
1207 link_id,
1208 inbound_actions,
1209 } => {
1210 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1211 }
1217 LinkDataResult::LinkClose {
1218 link_id,
1219 teardown_actions,
1220 } => {
1221 actions.extend(self.process_link_actions(&link_id, &teardown_actions));
1222 }
1223 LinkDataResult::Channel {
1224 link_id,
1225 inbound_actions,
1226 plaintext,
1227 packet_hash,
1228 } => {
1229 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1230 if let Some(link) = self.links.get_mut(&link_id) {
1232 if let Some(ref mut channel) = link.channel {
1233 let chan_actions = channel.receive(&plaintext, time::now());
1234 link.channel_messages_received += chan_actions
1235 .iter()
1236 .filter(|action| {
1237 matches!(
1238 action,
1239 rns_core::channel::ChannelAction::MessageReceived { .. }
1240 )
1241 })
1242 .count()
1243 as u64;
1244 let _ = link;
1246 actions.extend(self.process_channel_actions(&link_id, chan_actions, rng));
1247 }
1248 }
1249 actions.extend(self.build_link_packet_proof(&link_id, &packet_hash));
1250 }
1251 LinkDataResult::Request {
1252 link_id,
1253 inbound_actions,
1254 plaintext,
1255 request_id,
1256 } => {
1257 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1258 actions.extend(self.handle_request(&link_id, &plaintext, request_id, rng));
1259 }
1260 LinkDataResult::Response {
1261 link_id,
1262 inbound_actions,
1263 plaintext,
1264 } => {
1265 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1266 actions.extend(self.handle_response(&link_id, &plaintext, None, None));
1268 }
1269 LinkDataResult::Generic {
1270 link_id,
1271 inbound_actions,
1272 plaintext,
1273 context,
1274 packet_hash,
1275 } => {
1276 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1277 actions.push(LinkManagerAction::LinkDataReceived {
1278 link_id,
1279 context,
1280 data: plaintext,
1281 });
1282
1283 actions.extend(self.build_link_packet_proof(&link_id, &packet_hash));
1284 }
1285 LinkDataResult::ResourceAdv {
1286 link_id,
1287 inbound_actions,
1288 plaintext,
1289 } => {
1290 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1291 actions.extend(self.handle_resource_adv(&link_id, &plaintext, rng));
1292 }
1293 LinkDataResult::ResourceReq {
1294 link_id,
1295 inbound_actions,
1296 plaintext,
1297 } => {
1298 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1299 actions.extend(self.handle_resource_req(&link_id, &plaintext, rng));
1300 }
1301 LinkDataResult::ResourceHmu {
1302 link_id,
1303 inbound_actions,
1304 plaintext,
1305 } => {
1306 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1307 actions.extend(self.handle_resource_hmu(&link_id, &plaintext, rng));
1308 }
1309 LinkDataResult::ResourcePart {
1310 link_id,
1311 inbound_actions,
1312 raw_data,
1313 } => {
1314 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1315 actions.extend(self.handle_resource_part(&link_id, &raw_data, rng));
1316 }
1317 LinkDataResult::ResourcePrf {
1318 link_id,
1319 inbound_actions,
1320 plaintext,
1321 } => {
1322 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1323 actions.extend(self.handle_resource_prf(&link_id, &plaintext, rng));
1324 }
1325 LinkDataResult::ResourceIcl {
1326 link_id,
1327 inbound_actions,
1328 } => {
1329 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1330 actions.extend(self.handle_resource_icl(&link_id));
1331 }
1332 LinkDataResult::ResourceRcl {
1333 link_id,
1334 inbound_actions,
1335 } => {
1336 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1337 actions.extend(self.handle_resource_rcl(&link_id));
1338 }
1339 LinkDataResult::Error => {}
1340 }
1341
1342 actions
1343 }
1344
1345 fn handle_request(
1347 &mut self,
1348 link_id: &LinkId,
1349 plaintext: &[u8],
1350 request_id: [u8; 16],
1351 rng: &mut dyn Rng,
1352 ) -> Vec<LinkManagerAction> {
1353 use rns_core::msgpack::{self, Value};
1354
1355 let arr = match msgpack::unpack_exact(plaintext) {
1357 Ok(Value::Array(arr)) if arr.len() >= 3 => arr,
1358 _ => return Vec::new(),
1359 };
1360
1361 let path_hash_bytes = match &arr[1] {
1362 Value::Bin(b) if b.len() == 16 => b,
1363 _ => return Vec::new(),
1364 };
1365 let mut path_hash = [0u8; 16];
1366 path_hash.copy_from_slice(path_hash_bytes);
1367
1368 let request_data = msgpack::pack(&arr[2]);
1370
1371 if self.management_paths.contains(&path_hash) {
1373 let remote_identity = self
1374 .links
1375 .get(link_id)
1376 .and_then(|l| l.remote_identity)
1377 .map(|(h, k)| (h, k));
1378 return vec![LinkManagerAction::ManagementRequest {
1379 link_id: *link_id,
1380 path_hash,
1381 data: request_data,
1382 request_id,
1383 remote_identity,
1384 }];
1385 }
1386
1387 let handler_idx = self
1389 .request_handlers
1390 .iter()
1391 .position(|h| h.path_hash == path_hash);
1392 let handler_idx = match handler_idx {
1393 Some(i) => i,
1394 None => return Vec::new(),
1395 };
1396
1397 let remote_identity = self
1399 .links
1400 .get(link_id)
1401 .and_then(|l| l.remote_identity.as_ref());
1402 let handler = &self.request_handlers[handler_idx];
1403 if let Some(ref allowed) = handler.allowed_list {
1404 match remote_identity {
1405 Some((identity_hash, _)) => {
1406 if !allowed.contains(identity_hash) {
1407 log::debug!("Request denied: identity not in allowed list");
1408 return Vec::new();
1409 }
1410 }
1411 None => {
1412 log::debug!("Request denied: peer not identified");
1413 return Vec::new();
1414 }
1415 }
1416 }
1417
1418 let path = handler.path.clone();
1420 let response = (handler.handler)(*link_id, &path, &request_data, remote_identity);
1421
1422 let mut actions = Vec::new();
1423 if let Some(response) = response {
1424 match response {
1425 RequestResponse::Bytes(response_data) => {
1426 let mut response_actions =
1427 self.build_response_packet(link_id, &request_id, &response_data, rng);
1428 if response_actions.is_empty() {
1429 response_actions.extend(self.send_response_resource(
1430 link_id,
1431 &request_id,
1432 &response_data,
1433 None,
1434 true,
1435 rng,
1436 ));
1437 }
1438 actions.extend(response_actions);
1439 }
1440 RequestResponse::Resource {
1441 data,
1442 metadata,
1443 auto_compress,
1444 } => {
1445 actions.extend(self.send_response_resource(
1446 link_id,
1447 &request_id,
1448 &data,
1449 metadata.as_deref(),
1450 auto_compress,
1451 rng,
1452 ));
1453 }
1454 }
1455 }
1456
1457 actions
1458 }
1459
1460 fn build_response_packet(
1463 &self,
1464 link_id: &LinkId,
1465 request_id: &[u8; 16],
1466 response_data: &[u8],
1467 rng: &mut dyn Rng,
1468 ) -> Vec<LinkManagerAction> {
1469 use rns_core::msgpack::{self, Value};
1470
1471 let response_value = msgpack::unpack_exact(response_data)
1472 .unwrap_or_else(|_| Value::Bin(response_data.to_vec()));
1473
1474 let response_array = Value::Array(vec![Value::Bin(request_id.to_vec()), response_value]);
1475 let response_plaintext = msgpack::pack(&response_array);
1476
1477 let mut actions = Vec::new();
1478 if let Some(link) = self.links.get(link_id) {
1479 if let Ok(encrypted) = link.engine.encrypt(&response_plaintext, rng) {
1480 let flags = PacketFlags {
1481 header_type: constants::HEADER_1,
1482 context_flag: constants::FLAG_UNSET,
1483 transport_type: constants::TRANSPORT_BROADCAST,
1484 destination_type: constants::DESTINATION_LINK,
1485 packet_type: constants::PACKET_TYPE_DATA,
1486 };
1487 let max_mtu = link.engine.mtu() as usize;
1488 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
1489 flags,
1490 0,
1491 link_id,
1492 None,
1493 constants::CONTEXT_RESPONSE,
1494 &encrypted,
1495 max_mtu,
1496 ) {
1497 actions.push(LinkManagerAction::SendPacket {
1498 raw,
1499 dest_type: constants::DESTINATION_LINK,
1500 attached_interface: None,
1501 });
1502 }
1503 }
1504 }
1505 actions
1506 }
1507
1508 fn send_response_resource(
1509 &mut self,
1510 link_id: &LinkId,
1511 request_id: &[u8; 16],
1512 response_data: &[u8],
1513 metadata: Option<&[u8]>,
1514 auto_compress: bool,
1515 rng: &mut dyn Rng,
1516 ) -> Vec<LinkManagerAction> {
1517 use rns_core::msgpack::{self, Value};
1518
1519 let link = match self.links.get_mut(link_id) {
1520 Some(l) => l,
1521 None => return Vec::new(),
1522 };
1523
1524 if link.engine.state() != LinkState::Active {
1525 return Vec::new();
1526 }
1527
1528 let now = time::now();
1529
1530 let response_value = msgpack::unpack_exact(response_data)
1534 .unwrap_or_else(|_| Value::Bin(response_data.to_vec()));
1535 let response_array = Value::Array(vec![Value::Bin(request_id.to_vec()), response_value]);
1536 let resource_payload = msgpack::pack(&response_array);
1537
1538 let senders = match Self::build_resource_senders(
1539 link,
1540 &resource_payload,
1541 metadata,
1542 auto_compress,
1543 true, Some(request_id.to_vec()),
1545 rng,
1546 now,
1547 ) {
1548 Ok(s) => s,
1549 Err(e) => {
1550 log::debug!("Failed to create response ResourceSender: {}", e);
1551 return Vec::new();
1552 }
1553 };
1554
1555 let adv_actions = Self::start_resource_senders(link, senders, now);
1556
1557 let _ = link;
1558 self.process_resource_actions(link_id, adv_actions, rng)
1559 }
1560
1561 pub fn send_management_response(
1564 &mut self,
1565 link_id: &LinkId,
1566 request_id: &[u8; 16],
1567 response_data: &[u8],
1568 rng: &mut dyn Rng,
1569 ) -> Vec<LinkManagerAction> {
1570 let mut actions = self.build_response_packet(link_id, request_id, response_data, rng);
1571 if actions.is_empty() {
1572 actions.extend(self.send_response_resource(
1573 link_id,
1574 request_id,
1575 response_data,
1576 None,
1577 true,
1578 rng,
1579 ));
1580 }
1581 actions
1582 }
1583
1584 pub fn send_request(
1592 &mut self,
1593 link_id: &LinkId,
1594 path: &str,
1595 data: &[u8],
1596 rng: &mut dyn Rng,
1597 ) -> Vec<LinkManagerAction> {
1598 use rns_core::msgpack::{self, Value};
1599
1600 let link = match self.links.get_mut(link_id) {
1601 Some(l) => l,
1602 None => return Vec::new(),
1603 };
1604
1605 if link.engine.state() != LinkState::Active {
1606 return Vec::new();
1607 }
1608
1609 let path_hash = compute_path_hash(path);
1610
1611 let data_value = msgpack::unpack_exact(data).unwrap_or_else(|_| Value::Bin(data.to_vec()));
1613
1614 let request_array = Value::Array(vec![
1616 Value::Float(time::now()),
1617 Value::Bin(path_hash.to_vec()),
1618 data_value,
1619 ]);
1620 let plaintext = msgpack::pack(&request_array);
1621
1622 if plaintext.len() > link.engine.mdu() {
1623 let request_id = rns_core::hash::truncated_hash(&plaintext);
1624 let now = time::now();
1625 let senders = match Self::build_resource_senders(
1626 link,
1627 &plaintext,
1628 None,
1629 true,
1630 false,
1631 Some(request_id.to_vec()),
1632 rng,
1633 now,
1634 ) {
1635 Ok(senders) => senders,
1636 Err(e) => {
1637 log::debug!("Failed to create request ResourceSender: {}", e);
1638 return Vec::new();
1639 }
1640 };
1641 let adv_actions = Self::start_resource_senders(link, senders, now);
1642 let _ = link;
1643 return self.process_resource_actions(link_id, adv_actions, rng);
1644 }
1645
1646 let encrypted = match link.engine.encrypt(&plaintext, rng) {
1647 Ok(e) => e,
1648 Err(_) => return Vec::new(),
1649 };
1650
1651 let flags = PacketFlags {
1652 header_type: constants::HEADER_1,
1653 context_flag: constants::FLAG_UNSET,
1654 transport_type: constants::TRANSPORT_BROADCAST,
1655 destination_type: constants::DESTINATION_LINK,
1656 packet_type: constants::PACKET_TYPE_DATA,
1657 };
1658
1659 let mut actions = Vec::new();
1660 let max_mtu = link.engine.mtu() as usize;
1661 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
1662 flags,
1663 0,
1664 link_id,
1665 None,
1666 constants::CONTEXT_REQUEST,
1667 &encrypted,
1668 max_mtu,
1669 ) {
1670 actions.push(LinkManagerAction::SendPacket {
1671 raw,
1672 dest_type: constants::DESTINATION_LINK,
1673 attached_interface: None,
1674 });
1675 }
1676 actions
1677 }
1678
1679 pub fn send_on_link(
1681 &self,
1682 link_id: &LinkId,
1683 plaintext: &[u8],
1684 context: u8,
1685 rng: &mut dyn Rng,
1686 ) -> Vec<LinkManagerAction> {
1687 let link = match self.links.get(link_id) {
1688 Some(l) => l,
1689 None => return Vec::new(),
1690 };
1691
1692 if link.engine.state() != LinkState::Active {
1693 return Vec::new();
1694 }
1695
1696 let encrypted = match link.engine.encrypt(plaintext, rng) {
1697 Ok(e) => e,
1698 Err(_) => return Vec::new(),
1699 };
1700
1701 let flags = PacketFlags {
1702 header_type: constants::HEADER_1,
1703 context_flag: constants::FLAG_UNSET,
1704 transport_type: constants::TRANSPORT_BROADCAST,
1705 destination_type: constants::DESTINATION_LINK,
1706 packet_type: constants::PACKET_TYPE_DATA,
1707 };
1708
1709 let mut actions = Vec::new();
1710 if let Ok((raw, _packet_hash)) =
1711 RawPacket::pack_raw_with_hash(flags, 0, link_id, None, context, &encrypted)
1712 {
1713 actions.push(LinkManagerAction::SendPacket {
1714 raw,
1715 dest_type: constants::DESTINATION_LINK,
1716 attached_interface: None,
1717 });
1718 }
1719 actions
1720 }
1721
1722 pub fn identify(
1724 &self,
1725 link_id: &LinkId,
1726 identity: &rns_crypto::identity::Identity,
1727 rng: &mut dyn Rng,
1728 ) -> Vec<LinkManagerAction> {
1729 let link = match self.links.get(link_id) {
1730 Some(l) => l,
1731 None => return Vec::new(),
1732 };
1733
1734 let encrypted = match link.engine.build_identify(identity, rng) {
1735 Ok(e) => e,
1736 Err(_) => return Vec::new(),
1737 };
1738
1739 let flags = PacketFlags {
1740 header_type: constants::HEADER_1,
1741 context_flag: constants::FLAG_UNSET,
1742 transport_type: constants::TRANSPORT_BROADCAST,
1743 destination_type: constants::DESTINATION_LINK,
1744 packet_type: constants::PACKET_TYPE_DATA,
1745 };
1746
1747 let mut actions = Vec::new();
1748 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
1749 flags,
1750 0,
1751 link_id,
1752 None,
1753 constants::CONTEXT_LINKIDENTIFY,
1754 &encrypted,
1755 ) {
1756 actions.push(LinkManagerAction::SendPacket {
1757 raw,
1758 dest_type: constants::DESTINATION_LINK,
1759 attached_interface: None,
1760 });
1761 }
1762 actions
1763 }
1764
1765 pub fn teardown_link(&mut self, link_id: &LinkId) -> Vec<LinkManagerAction> {
1767 let link = match self.links.get_mut(link_id) {
1768 Some(l) => l,
1769 None => return Vec::new(),
1770 };
1771
1772 let teardown_actions = link.engine.teardown();
1773 if let Some(ref mut channel) = link.channel {
1774 channel.shutdown();
1775 }
1776
1777 let mut actions = self.process_link_actions(link_id, &teardown_actions);
1778
1779 let flags = PacketFlags {
1781 header_type: constants::HEADER_1,
1782 context_flag: constants::FLAG_UNSET,
1783 transport_type: constants::TRANSPORT_BROADCAST,
1784 destination_type: constants::DESTINATION_LINK,
1785 packet_type: constants::PACKET_TYPE_DATA,
1786 };
1787 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
1788 flags,
1789 0,
1790 link_id,
1791 None,
1792 constants::CONTEXT_LINKCLOSE,
1793 &[],
1794 ) {
1795 actions.push(LinkManagerAction::SendPacket {
1796 raw,
1797 dest_type: constants::DESTINATION_LINK,
1798 attached_interface: None,
1799 });
1800 }
1801
1802 actions
1803 }
1804
1805 pub fn teardown_all_links(&mut self) -> Vec<LinkManagerAction> {
1807 let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
1808 let mut actions = Vec::new();
1809 for link_id in link_ids {
1810 actions.extend(self.teardown_link(&link_id));
1811 }
1812 actions
1813 }
1814
1815 fn handle_response(
1817 &self,
1818 link_id: &LinkId,
1819 plaintext: &[u8],
1820 metadata: Option<Vec<u8>>,
1821 resource_request_id: Option<[u8; 16]>,
1822 ) -> Vec<LinkManagerAction> {
1823 use rns_core::msgpack;
1824
1825 let packet_response = msgpack::unpack_exact(plaintext).ok().and_then(|value| {
1827 let msgpack::Value::Array(arr) = value else {
1828 return None;
1829 };
1830 if arr.len() < 2 {
1831 return None;
1832 }
1833 let msgpack::Value::Bin(request_id_bytes) = &arr[0] else {
1834 return None;
1835 };
1836 if request_id_bytes.len() != 16 {
1837 return None;
1838 }
1839 let mut request_id = [0u8; 16];
1840 request_id.copy_from_slice(request_id_bytes);
1841 Some((request_id, msgpack::pack(&arr[1])))
1842 });
1843
1844 let (request_id, response_data) = match packet_response {
1845 Some(response) => response,
1846 None => {
1847 let Some(request_id) = resource_request_id else {
1848 return Vec::new();
1849 };
1850 (
1851 request_id,
1852 msgpack::pack(&msgpack::Value::Bin(plaintext.to_vec())),
1853 )
1854 }
1855 };
1856
1857 vec![LinkManagerAction::ResponseReceived {
1858 link_id: *link_id,
1859 request_id,
1860 data: response_data,
1861 metadata,
1862 }]
1863 }
1864
1865 fn response_request_id(request_id: &Option<Vec<u8>>) -> Option<[u8; 16]> {
1866 let bytes = request_id.as_deref()?;
1867 if bytes.len() != 16 {
1868 return None;
1869 }
1870 let mut out = [0u8; 16];
1871 out.copy_from_slice(bytes);
1872 Some(out)
1873 }
1874
1875 fn build_resource_senders(
1876 link: &ManagedLink,
1877 data: &[u8],
1878 metadata: Option<&[u8]>,
1879 auto_compress: bool,
1880 is_response: bool,
1881 request_id: Option<Vec<u8>>,
1882 rng: &mut dyn Rng,
1883 now: f64,
1884 ) -> Result<Vec<ResourceSender>, rns_core::resource::ResourceError> {
1885 let link_rtt = link.engine.rtt().unwrap_or(1.0);
1886 let resource_sdu = Self::resource_sdu_for_link(link);
1887 let metadata_overhead = metadata.map(|m| 3 + m.len()).unwrap_or(0);
1888 let logical_size = metadata_overhead + data.len();
1889
1890 if logical_size <= constants::RESOURCE_MAX_EFFICIENT_SIZE {
1891 let enc_rng = std::cell::RefCell::new(rns_crypto::OsRng);
1892 let encrypt_fn = |plaintext: &[u8]| -> Vec<u8> {
1893 link.engine
1894 .encrypt(plaintext, &mut *enc_rng.borrow_mut())
1895 .unwrap_or_else(|_| plaintext.to_vec())
1896 };
1897 return ResourceSender::new(
1898 data,
1899 metadata,
1900 resource_sdu,
1901 &encrypt_fn,
1902 &Bzip2Compressor,
1903 rng,
1904 now,
1905 auto_compress,
1906 is_response,
1907 request_id,
1908 1,
1909 1,
1910 None,
1911 link_rtt,
1912 6.0,
1913 )
1914 .map(|sender| vec![sender]);
1915 }
1916
1917 if metadata_overhead > constants::RESOURCE_MAX_EFFICIENT_SIZE {
1918 return Err(rns_core::resource::ResourceError::TooLarge);
1919 }
1920
1921 let first_payload_len = core::cmp::min(
1922 data.len(),
1923 constants::RESOURCE_MAX_EFFICIENT_SIZE - metadata_overhead,
1924 );
1925 let remaining = data.len().saturating_sub(first_payload_len);
1926 let total_segments = 1 + remaining.div_ceil(constants::RESOURCE_MAX_EFFICIENT_SIZE) as u64;
1927
1928 let enc_rng = std::cell::RefCell::new(rns_crypto::OsRng);
1929 let encrypt_fn = |plaintext: &[u8]| -> Vec<u8> {
1930 link.engine
1931 .encrypt(plaintext, &mut *enc_rng.borrow_mut())
1932 .unwrap_or_else(|_| plaintext.to_vec())
1933 };
1934
1935 let mut senders = Vec::new();
1936 let mut first = ResourceSender::new(
1937 &data[..first_payload_len],
1938 metadata,
1939 resource_sdu,
1940 &encrypt_fn,
1941 &Bzip2Compressor,
1942 rng,
1943 now,
1944 auto_compress,
1945 is_response,
1946 request_id.clone(),
1947 1,
1948 total_segments,
1949 None,
1950 link_rtt,
1951 6.0,
1952 )?;
1953 first.data_size = logical_size;
1954 let original_hash = first.original_hash;
1955 let has_metadata = metadata.is_some();
1956 senders.push(first);
1957
1958 let mut offset = first_payload_len;
1959 let mut segment_index = 2;
1960 while offset < data.len() {
1961 let end = core::cmp::min(offset + constants::RESOURCE_MAX_EFFICIENT_SIZE, data.len());
1962 let mut sender = ResourceSender::new(
1963 &data[offset..end],
1964 None,
1965 resource_sdu,
1966 &encrypt_fn,
1967 &Bzip2Compressor,
1968 rng,
1969 now,
1970 auto_compress,
1971 is_response,
1972 request_id.clone(),
1973 segment_index,
1974 total_segments,
1975 Some(original_hash),
1976 link_rtt,
1977 6.0,
1978 )?;
1979 sender.data_size = logical_size;
1980 sender.flags.has_metadata = has_metadata;
1981 senders.push(sender);
1982 segment_index += 1;
1983 offset = end;
1984 }
1985
1986 Ok(senders)
1987 }
1988
1989 fn start_resource_senders(
1990 link: &mut ManagedLink,
1991 mut senders: Vec<ResourceSender>,
1992 now: f64,
1993 ) -> Vec<ResourceAction> {
1994 if senders.is_empty() {
1995 return Vec::new();
1996 }
1997
1998 let mut first = senders.remove(0);
1999 let adv_actions = first.advertise(now);
2000
2001 if first.total_segments > 1 {
2002 let original_hash = first.original_hash;
2003 let split = OutgoingSplitTransfer {
2004 total_segments: first.total_segments,
2005 completed_segments: 0,
2006 current_segment_index: first.segment_index,
2007 current_sent_parts: 0,
2008 current_total_parts: first.total_parts(),
2009 };
2010 link.outgoing_splits.insert(original_hash, split);
2011 }
2012
2013 link.outgoing_resources.push(first);
2014 link.outgoing_resources.extend(senders);
2015 adv_actions
2016 }
2017
2018 fn handle_resource_adv(
2020 &mut self,
2021 link_id: &LinkId,
2022 adv_plaintext: &[u8],
2023 rng: &mut dyn Rng,
2024 ) -> Vec<LinkManagerAction> {
2025 let link = match self.links.get_mut(link_id) {
2026 Some(l) => l,
2027 None => return Vec::new(),
2028 };
2029
2030 let link_rtt = link.engine.rtt().unwrap_or(1.0);
2031 let resource_sdu = Self::resource_sdu_for_link(link);
2032 let now = time::now();
2033
2034 let receiver = match ResourceReceiver::from_advertisement(
2035 adv_plaintext,
2036 resource_sdu,
2037 link_rtt,
2038 now,
2039 None,
2040 None,
2041 ) {
2042 Ok(r) => r,
2043 Err(e) => {
2044 log::debug!("Resource ADV rejected: {}", e);
2045 return Vec::new();
2046 }
2047 };
2048
2049 let strategy = link.resource_strategy;
2050 let resource_hash = receiver.resource_hash.clone();
2051 let transfer_size = receiver.transfer_size;
2052 let has_metadata = receiver.has_metadata;
2053 let is_response = receiver.flags.is_response;
2054 let is_split = receiver.flags.split;
2055 let segment_index = receiver.segment_index;
2056 let total_segments = receiver.total_segments;
2057 let original_hash = match Self::resource_hash_key(&receiver.original_hash) {
2058 Some(key) => key,
2059 None => return Vec::new(),
2060 };
2061
2062 if is_split && segment_index > 1 {
2063 let should_accept = link
2064 .incoming_splits
2065 .get(&original_hash)
2066 .is_some_and(|split| {
2067 split.completed_segments + 1 == segment_index
2068 && split.total_segments == total_segments
2069 });
2070
2071 if !should_accept {
2072 let reject_actions = {
2073 let mut r = receiver;
2074 r.reject()
2075 };
2076 let _ = link;
2077 return self.process_resource_actions(link_id, reject_actions, rng);
2078 }
2079
2080 let current_total_parts = receiver.total_parts;
2081 link.incoming_resources.push(receiver);
2082 let idx = link.incoming_resources.len() - 1;
2083 if let Some(split) = link.incoming_splits.get_mut(&original_hash) {
2084 split.current_segment_index = segment_index;
2085 split.current_received_parts = 0;
2086 split.current_total_parts = current_total_parts;
2087 }
2088 let resource_actions = link.incoming_resources[idx].accept(now);
2089 let _ = link;
2090 return self.process_resource_actions(link_id, resource_actions, rng);
2091 }
2092
2093 if is_response {
2094 if is_split {
2097 link.incoming_splits.insert(
2098 original_hash,
2099 IncomingSplitTransfer {
2100 total_segments,
2101 completed_segments: 0,
2102 current_segment_index: segment_index,
2103 current_received_parts: 0,
2104 current_total_parts: receiver.total_parts,
2105 data: Vec::new(),
2106 metadata: None,
2107 is_response,
2108 },
2109 );
2110 }
2111 link.incoming_resources.push(receiver);
2112 let idx = link.incoming_resources.len() - 1;
2113 let resource_actions = link.incoming_resources[idx].accept(now);
2114 let _ = link;
2115 return self.process_resource_actions(link_id, resource_actions, rng);
2116 }
2117
2118 match strategy {
2119 ResourceStrategy::AcceptNone => {
2120 let reject_actions = {
2122 let mut r = receiver;
2123 r.reject()
2124 };
2125 self.process_resource_actions(link_id, reject_actions, rng)
2126 }
2127 ResourceStrategy::AcceptAll => {
2128 if is_split {
2129 link.incoming_splits.insert(
2130 original_hash,
2131 IncomingSplitTransfer {
2132 total_segments,
2133 completed_segments: 0,
2134 current_segment_index: segment_index,
2135 current_received_parts: 0,
2136 current_total_parts: receiver.total_parts,
2137 data: Vec::new(),
2138 metadata: None,
2139 is_response,
2140 },
2141 );
2142 }
2143 link.incoming_resources.push(receiver);
2144 let idx = link.incoming_resources.len() - 1;
2145 let resource_actions = link.incoming_resources[idx].accept(now);
2146 let _ = link;
2147 self.process_resource_actions(link_id, resource_actions, rng)
2148 }
2149 ResourceStrategy::AcceptApp => {
2150 link.incoming_resources.push(receiver);
2151 vec![LinkManagerAction::ResourceAcceptQuery {
2153 link_id: *link_id,
2154 resource_hash,
2155 transfer_size,
2156 has_metadata,
2157 }]
2158 }
2159 }
2160 }
2161
2162 pub fn accept_resource(
2164 &mut self,
2165 link_id: &LinkId,
2166 resource_hash: &[u8],
2167 accept: bool,
2168 rng: &mut dyn Rng,
2169 ) -> Vec<LinkManagerAction> {
2170 let link = match self.links.get_mut(link_id) {
2171 Some(l) => l,
2172 None => return Vec::new(),
2173 };
2174
2175 let now = time::now();
2176 let idx = link
2177 .incoming_resources
2178 .iter()
2179 .position(|r| r.resource_hash == resource_hash);
2180 let idx = match idx {
2181 Some(i) => i,
2182 None => return Vec::new(),
2183 };
2184
2185 if accept && link.incoming_resources[idx].flags.split {
2186 if let Some(original_hash) =
2187 Self::resource_hash_key(&link.incoming_resources[idx].original_hash)
2188 {
2189 link.incoming_splits
2190 .entry(original_hash)
2191 .or_insert_with(|| IncomingSplitTransfer {
2192 total_segments: link.incoming_resources[idx].total_segments,
2193 completed_segments: 0,
2194 current_segment_index: link.incoming_resources[idx].segment_index,
2195 current_received_parts: 0,
2196 current_total_parts: link.incoming_resources[idx].total_parts,
2197 data: Vec::new(),
2198 metadata: None,
2199 is_response: link.incoming_resources[idx].flags.is_response,
2200 });
2201 }
2202 }
2203
2204 let resource_actions = if accept {
2205 link.incoming_resources[idx].accept(now)
2206 } else {
2207 link.incoming_resources[idx].reject()
2208 };
2209
2210 let _ = link;
2211 self.process_resource_actions(link_id, resource_actions, rng)
2212 }
2213
2214 fn handle_resource_req(
2216 &mut self,
2217 link_id: &LinkId,
2218 plaintext: &[u8],
2219 rng: &mut dyn Rng,
2220 ) -> Vec<LinkManagerAction> {
2221 let link = match self.links.get_mut(link_id) {
2222 Some(l) => l,
2223 None => return Vec::new(),
2224 };
2225
2226 let now = time::now();
2227 let mut all_actions = Vec::new();
2228 let mut progress_update = None;
2229 for sender in &mut link.outgoing_resources {
2230 if sender.flags.split && sender.status == rns_core::resource::ResourceStatus::Queued {
2231 continue;
2232 }
2233 let before_sent = sender.sent_parts;
2234 let resource_actions = sender.handle_request(plaintext, now);
2235 if !resource_actions.is_empty() {
2236 if sender.sent_parts != before_sent {
2237 if sender.flags.split {
2238 if let Some(split) = link.outgoing_splits.get_mut(&sender.original_hash) {
2239 split.current_segment_index = sender.segment_index;
2240 split.current_sent_parts = sender.sent_parts;
2241 split.current_total_parts = sender.total_parts();
2242 progress_update =
2243 Some(Self::outgoing_split_progress(split, sender.sdu));
2244 }
2245 } else {
2246 progress_update = Some((sender.sent_parts, sender.total_parts()));
2247 }
2248 }
2249 all_actions.extend(resource_actions);
2250 break;
2251 }
2252 }
2253
2254 let _ = link;
2255 let mut out = self.process_resource_actions(link_id, all_actions, rng);
2256 if let Some((received, total)) = progress_update {
2257 out.push(LinkManagerAction::ResourceProgress {
2258 link_id: *link_id,
2259 received,
2260 total,
2261 });
2262 }
2263 out
2264 }
2265
2266 fn handle_resource_hmu(
2268 &mut self,
2269 link_id: &LinkId,
2270 plaintext: &[u8],
2271 rng: &mut dyn Rng,
2272 ) -> Vec<LinkManagerAction> {
2273 let link = match self.links.get_mut(link_id) {
2274 Some(l) => l,
2275 None => return Vec::new(),
2276 };
2277
2278 let now = time::now();
2279 let mut all_actions = Vec::new();
2280 for receiver in &mut link.incoming_resources {
2281 let resource_actions = receiver.handle_hashmap_update(plaintext, now);
2282 if !resource_actions.is_empty() {
2283 all_actions.extend(resource_actions);
2284 break;
2285 }
2286 }
2287
2288 let _ = link;
2289 self.process_resource_actions(link_id, all_actions, rng)
2290 }
2291
2292 fn handle_resource_part(
2294 &mut self,
2295 link_id: &LinkId,
2296 raw_data: &[u8],
2297 rng: &mut dyn Rng,
2298 ) -> Vec<LinkManagerAction> {
2299 let link = match self.links.get_mut(link_id) {
2300 Some(l) => l,
2301 None => return Vec::new(),
2302 };
2303
2304 let now = time::now();
2305 let resource_sdu = Self::resource_sdu_for_link(link);
2306 let mut all_actions = Vec::new();
2307 let mut assemble_idx = None;
2308 let mut assembled_is_response = false;
2309 let mut response_request_id = None;
2310
2311 for (idx, receiver) in link.incoming_resources.iter_mut().enumerate() {
2312 if receiver.status >= rns_core::resource::ResourceStatus::Complete {
2313 continue;
2314 }
2315 let resource_actions = receiver.receive_part(raw_data, now);
2316 if !resource_actions.is_empty() {
2317 if receiver.received_count == receiver.total_parts {
2318 assemble_idx = Some(idx);
2319 }
2320 if receiver.flags.split {
2321 if let Some(key) = Self::resource_hash_key(&receiver.original_hash) {
2322 if let Some(split) = link.incoming_splits.get_mut(&key) {
2323 split.current_segment_index = receiver.segment_index;
2324 split.current_received_parts = receiver.received_count;
2325 split.current_total_parts = receiver.total_parts;
2326 let (received, total) =
2327 Self::incoming_split_progress(split, resource_sdu);
2328 for action in resource_actions {
2329 match action {
2330 ResourceAction::ProgressUpdate { .. } => {
2331 all_actions.push(ResourceAction::ProgressUpdate {
2332 received,
2333 total,
2334 });
2335 }
2336 other => all_actions.push(other),
2337 }
2338 }
2339 } else {
2340 all_actions.extend(resource_actions);
2341 }
2342 } else {
2343 all_actions.extend(resource_actions);
2344 }
2345 } else {
2346 all_actions.extend(resource_actions);
2347 }
2348 break;
2349 }
2350 }
2351
2352 if let Some(idx) = assemble_idx {
2353 let split_key = if link.incoming_resources[idx].flags.split {
2354 Self::resource_hash_key(&link.incoming_resources[idx].original_hash)
2355 } else {
2356 None
2357 };
2358 let split_segment_index = link.incoming_resources[idx].segment_index;
2359 let split_segment_total = link.incoming_resources[idx].total_segments;
2360 let split_segment_parts = link.incoming_resources[idx].total_parts;
2361 let split_is_response = link.incoming_resources[idx].flags.is_response;
2362 response_request_id =
2363 Self::response_request_id(&link.incoming_resources[idx].request_id);
2364 let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
2365 link.engine.decrypt(ciphertext).map_err(|_| ())
2366 };
2367 let mut assemble_actions =
2368 link.incoming_resources[idx].assemble(&decrypt_fn, &Bzip2Compressor);
2369 assembled_is_response = split_is_response;
2370
2371 if let Some(key) = split_key {
2372 let mut converted_actions = Vec::new();
2373 let mut segment_data = None;
2374 let mut segment_metadata = None;
2375 for action in assemble_actions {
2376 match action {
2377 ResourceAction::DataReceived { data, metadata } => {
2378 segment_data = Some(data);
2379 segment_metadata = metadata;
2380 }
2381 ResourceAction::Completed => {}
2382 other => converted_actions.push(other),
2383 }
2384 }
2385
2386 if let Some(data) = segment_data {
2387 if let Some(split) = link.incoming_splits.get_mut(&key) {
2388 split.data.extend_from_slice(&data);
2389 if segment_metadata.is_some() {
2390 split.metadata = segment_metadata;
2391 }
2392 split.completed_segments = split_segment_index;
2393 split.current_segment_index = split_segment_index;
2394 split.current_received_parts = split_segment_parts;
2395 split.current_total_parts = split_segment_parts;
2396 }
2397
2398 if split_segment_index == split_segment_total {
2399 if let Some(split) = link.incoming_splits.remove(&key) {
2400 assembled_is_response = split.is_response;
2401 converted_actions.push(ResourceAction::DataReceived {
2402 data: split.data,
2403 metadata: split.metadata,
2404 });
2405 converted_actions.push(ResourceAction::Completed);
2406 }
2407 }
2408 }
2409
2410 assemble_actions = converted_actions;
2411 }
2412 all_actions.extend(assemble_actions);
2413 }
2414
2415 let _ = link;
2416 let mut out = self.process_resource_actions(link_id, all_actions, rng);
2417
2418 if assembled_is_response || response_request_id.is_some() {
2419 let mut converted = Vec::new();
2420 for action in out {
2421 match action {
2422 LinkManagerAction::ResourceReceived { data, metadata, .. } => {
2423 converted.extend(self.handle_response(
2424 link_id,
2425 &data,
2426 metadata,
2427 response_request_id,
2428 ));
2429 }
2430 LinkManagerAction::ResourceAcceptQuery { .. } => {
2431 }
2433 other => converted.push(other),
2434 }
2435 }
2436 out = converted;
2437 }
2438
2439 out
2440 }
2441
2442 fn handle_resource_prf(
2444 &mut self,
2445 link_id: &LinkId,
2446 plaintext: &[u8],
2447 rng: &mut dyn Rng,
2448 ) -> Vec<LinkManagerAction> {
2449 let link = match self.links.get_mut(link_id) {
2450 Some(l) => l,
2451 None => return Vec::new(),
2452 };
2453
2454 let now = time::now();
2455 let mut result_actions = Vec::new();
2456 let mut completed_sender = None;
2457 let mut failed_split = None;
2458 let proof_hash = plaintext.get(..32);
2459 for sender in &mut link.outgoing_resources {
2460 if proof_hash.is_some_and(|hash| hash != sender.resource_hash.as_slice()) {
2461 continue;
2462 }
2463 let resource_actions = sender.handle_proof(plaintext, now);
2464 if !resource_actions.is_empty() {
2465 if resource_actions
2466 .iter()
2467 .any(|action| matches!(action, ResourceAction::Completed))
2468 {
2469 completed_sender = Some((
2470 sender.original_hash,
2471 sender.segment_index,
2472 sender.total_segments,
2473 sender.total_parts(),
2474 ));
2475 }
2476 if sender.flags.split
2477 && resource_actions
2478 .iter()
2479 .any(|action| matches!(action, ResourceAction::Failed(_)))
2480 {
2481 failed_split = Some(sender.original_hash);
2482 }
2483 result_actions.extend(resource_actions);
2484 break;
2485 }
2486 }
2487
2488 let mut actions = Vec::new();
2490 let mut advertise_next = None;
2491 for ra in result_actions {
2492 match ra {
2493 ResourceAction::Completed => {
2494 if let Some((original_hash, segment_index, total_segments, total_parts)) =
2495 completed_sender
2496 {
2497 if total_segments > 1 && segment_index < total_segments {
2498 if let Some(split) = link.outgoing_splits.get_mut(&original_hash) {
2499 split.completed_segments = segment_index;
2500 split.current_segment_index = segment_index;
2501 split.current_sent_parts = total_parts;
2502 split.current_total_parts = total_parts;
2503 if let Some(next) = link.outgoing_resources.iter_mut().find(|s| {
2504 s.flags.split
2505 && s.original_hash == original_hash
2506 && s.segment_index == segment_index + 1
2507 }) {
2508 split.current_segment_index = next.segment_index;
2509 split.current_sent_parts = 0;
2510 split.current_total_parts = next.total_parts();
2511 advertise_next = Some(next.advertise(now));
2512 }
2513 }
2514 } else {
2515 link.outgoing_splits.remove(&original_hash);
2516 actions
2517 .push(LinkManagerAction::ResourceCompleted { link_id: *link_id });
2518 }
2519 } else {
2520 actions.push(LinkManagerAction::ResourceCompleted { link_id: *link_id });
2521 }
2522 }
2523 ResourceAction::Failed(e) => {
2524 if let Some(original_hash) = failed_split {
2525 link.outgoing_splits.remove(&original_hash);
2526 }
2527 actions.push(LinkManagerAction::ResourceFailed {
2528 link_id: *link_id,
2529 error: format!("{}", e),
2530 });
2531 }
2532 _ => {}
2533 }
2534 }
2535
2536 link.outgoing_resources
2538 .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
2539
2540 let _ = link;
2541 if let Some(next_actions) = advertise_next {
2542 actions.extend(self.process_resource_actions(link_id, next_actions, rng));
2543 }
2544
2545 actions
2546 }
2547
2548 fn handle_resource_icl(&mut self, link_id: &LinkId) -> Vec<LinkManagerAction> {
2550 let link = match self.links.get_mut(link_id) {
2551 Some(l) => l,
2552 None => return Vec::new(),
2553 };
2554
2555 let mut actions = Vec::new();
2556 for receiver in &mut link.incoming_resources {
2557 let ra = receiver.handle_cancel();
2558 for a in ra {
2559 if let ResourceAction::Failed(ref e) = a {
2560 actions.push(LinkManagerAction::ResourceFailed {
2561 link_id: *link_id,
2562 error: format!("{}", e),
2563 });
2564 }
2565 }
2566 }
2567 link.incoming_resources
2568 .retain(|r| r.status < rns_core::resource::ResourceStatus::Complete);
2569 link.incoming_splits.clear();
2570 actions
2571 }
2572
2573 fn handle_resource_rcl(&mut self, link_id: &LinkId) -> Vec<LinkManagerAction> {
2575 let link = match self.links.get_mut(link_id) {
2576 Some(l) => l,
2577 None => return Vec::new(),
2578 };
2579
2580 let mut actions = Vec::new();
2581 for sender in &mut link.outgoing_resources {
2582 let ra = sender.handle_reject();
2583 for a in ra {
2584 if let ResourceAction::Failed(ref e) = a {
2585 actions.push(LinkManagerAction::ResourceFailed {
2586 link_id: *link_id,
2587 error: format!("{}", e),
2588 });
2589 }
2590 }
2591 }
2592 link.outgoing_resources
2593 .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
2594 link.outgoing_splits.clear();
2595 actions
2596 }
2597
2598 fn process_resource_actions(
2600 &mut self,
2601 link_id: &LinkId,
2602 actions: Vec<ResourceAction>,
2603 rng: &mut dyn Rng,
2604 ) -> Vec<LinkManagerAction> {
2605 let mut result = Vec::new();
2606 for action in actions {
2607 match action {
2608 ResourceAction::SendAdvertisement(data) => {
2609 let encrypted = self
2611 .links
2612 .get(link_id)
2613 .and_then(|link| link.engine.encrypt(&data, rng).ok());
2614 if let Some(encrypted) = encrypted {
2615 result.extend(self.build_link_packet(
2616 link_id,
2617 constants::CONTEXT_RESOURCE_ADV,
2618 &encrypted,
2619 ));
2620 }
2621 }
2622 ResourceAction::SendPart(data) => {
2623 result.extend(self.build_link_packet(
2625 link_id,
2626 constants::CONTEXT_RESOURCE,
2627 &data,
2628 ));
2629 }
2630 ResourceAction::SendRequest(data) => {
2631 let encrypted = self
2632 .links
2633 .get(link_id)
2634 .and_then(|link| link.engine.encrypt(&data, rng).ok());
2635 if let Some(encrypted) = encrypted {
2636 result.extend(self.build_link_packet(
2637 link_id,
2638 constants::CONTEXT_RESOURCE_REQ,
2639 &encrypted,
2640 ));
2641 }
2642 }
2643 ResourceAction::SendHmu(data) => {
2644 let encrypted = self
2645 .links
2646 .get(link_id)
2647 .and_then(|link| link.engine.encrypt(&data, rng).ok());
2648 if let Some(encrypted) = encrypted {
2649 result.extend(self.build_link_packet(
2650 link_id,
2651 constants::CONTEXT_RESOURCE_HMU,
2652 &encrypted,
2653 ));
2654 }
2655 }
2656 ResourceAction::SendProof(data) => {
2657 let flags = PacketFlags {
2658 header_type: constants::HEADER_1,
2659 context_flag: constants::FLAG_UNSET,
2660 transport_type: constants::TRANSPORT_BROADCAST,
2661 destination_type: constants::DESTINATION_LINK,
2662 packet_type: constants::PACKET_TYPE_PROOF,
2663 };
2664 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
2665 flags,
2666 0,
2667 link_id,
2668 None,
2669 constants::CONTEXT_RESOURCE_PRF,
2670 &data,
2671 ) {
2672 result.push(LinkManagerAction::SendPacket {
2673 raw,
2674 dest_type: constants::DESTINATION_LINK,
2675 attached_interface: None,
2676 });
2677 }
2678 }
2679 ResourceAction::SendCancelInitiator(data) => {
2680 let encrypted = self
2681 .links
2682 .get(link_id)
2683 .and_then(|link| link.engine.encrypt(&data, rng).ok());
2684 if let Some(encrypted) = encrypted {
2685 result.extend(self.build_link_packet(
2686 link_id,
2687 constants::CONTEXT_RESOURCE_ICL,
2688 &encrypted,
2689 ));
2690 }
2691 }
2692 ResourceAction::SendCancelReceiver(data) => {
2693 let encrypted = self
2694 .links
2695 .get(link_id)
2696 .and_then(|link| link.engine.encrypt(&data, rng).ok());
2697 if let Some(encrypted) = encrypted {
2698 result.extend(self.build_link_packet(
2699 link_id,
2700 constants::CONTEXT_RESOURCE_RCL,
2701 &encrypted,
2702 ));
2703 }
2704 }
2705 ResourceAction::DataReceived { data, metadata } => {
2706 result.push(LinkManagerAction::ResourceReceived {
2707 link_id: *link_id,
2708 data,
2709 metadata,
2710 });
2711 }
2712 ResourceAction::Completed => {
2713 result.push(LinkManagerAction::ResourceCompleted { link_id: *link_id });
2714 }
2715 ResourceAction::Failed(e) => {
2716 result.push(LinkManagerAction::ResourceFailed {
2717 link_id: *link_id,
2718 error: format!("{}", e),
2719 });
2720 }
2721 ResourceAction::TeardownLink => {
2722 let teardown_actions = match self.links.get_mut(link_id) {
2723 Some(link) => link.engine.handle_teardown(),
2724 None => Vec::new(),
2725 };
2726 result.extend(self.process_link_actions(link_id, &teardown_actions));
2727 }
2728 ResourceAction::ProgressUpdate { received, total } => {
2729 result.push(LinkManagerAction::ResourceProgress {
2730 link_id: *link_id,
2731 received,
2732 total,
2733 });
2734 }
2735 }
2736 }
2737 result
2738 }
2739
2740 fn build_link_packet(
2742 &self,
2743 link_id: &LinkId,
2744 context: u8,
2745 data: &[u8],
2746 ) -> Vec<LinkManagerAction> {
2747 let flags = PacketFlags {
2748 header_type: constants::HEADER_1,
2749 context_flag: constants::FLAG_UNSET,
2750 transport_type: constants::TRANSPORT_BROADCAST,
2751 destination_type: constants::DESTINATION_LINK,
2752 packet_type: constants::PACKET_TYPE_DATA,
2753 };
2754 let mut actions = Vec::new();
2755 let max_mtu = self
2756 .links
2757 .get(link_id)
2758 .map(|l| l.engine.mtu() as usize)
2759 .unwrap_or(constants::MTU);
2760 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
2761 flags, 0, link_id, None, context, data, max_mtu,
2762 ) {
2763 actions.push(LinkManagerAction::SendPacket {
2764 raw,
2765 dest_type: constants::DESTINATION_LINK,
2766 attached_interface: None,
2767 });
2768 }
2769 actions
2770 }
2771
2772 pub fn send_resource(
2774 &mut self,
2775 link_id: &LinkId,
2776 data: &[u8],
2777 metadata: Option<&[u8]>,
2778 rng: &mut dyn Rng,
2779 ) -> Vec<LinkManagerAction> {
2780 self.send_resource_with_auto_compress(link_id, data, metadata, true, rng)
2781 }
2782
2783 pub fn send_resource_with_auto_compress(
2785 &mut self,
2786 link_id: &LinkId,
2787 data: &[u8],
2788 metadata: Option<&[u8]>,
2789 auto_compress: bool,
2790 rng: &mut dyn Rng,
2791 ) -> Vec<LinkManagerAction> {
2792 let link = match self.links.get_mut(link_id) {
2793 Some(l) => l,
2794 None => return Vec::new(),
2795 };
2796
2797 if link.engine.state() != LinkState::Active {
2798 return Vec::new();
2799 }
2800
2801 let now = time::now();
2802
2803 let senders = match Self::build_resource_senders(
2804 link,
2805 data,
2806 metadata,
2807 auto_compress,
2808 false, None, rng,
2811 now,
2812 ) {
2813 Ok(s) => s,
2814 Err(e) => {
2815 log::debug!("Failed to create ResourceSender: {}", e);
2816 return Vec::new();
2817 }
2818 };
2819
2820 let adv_actions = Self::start_resource_senders(link, senders, now);
2821
2822 let _ = link;
2823 self.process_resource_actions(link_id, adv_actions, rng)
2824 }
2825
2826 pub fn set_resource_strategy(&mut self, link_id: &LinkId, strategy: ResourceStrategy) {
2828 if let Some(link) = self.links.get_mut(link_id) {
2829 link.resource_strategy = strategy;
2830 }
2831 }
2832
2833 pub fn flush_channel_tx(&mut self, link_id: &LinkId) {
2836 if let Some(link) = self.links.get_mut(link_id) {
2837 if let Some(ref mut channel) = link.channel {
2838 channel.flush_tx();
2839 }
2840 }
2841 }
2842
2843 pub fn send_channel_message(
2845 &mut self,
2846 link_id: &LinkId,
2847 msgtype: u16,
2848 payload: &[u8],
2849 rng: &mut dyn Rng,
2850 ) -> Result<Vec<LinkManagerAction>, String> {
2851 let link = match self.links.get_mut(link_id) {
2852 Some(l) => l,
2853 None => return Err("unknown link".to_string()),
2854 };
2855
2856 let channel = match link.channel {
2857 Some(ref mut ch) => ch,
2858 None => return Err("link has no active channel".to_string()),
2859 };
2860
2861 let link_mdu = link.engine.mdu();
2862 let now = time::now();
2863 let chan_actions = match channel.send(msgtype, payload, now, link_mdu) {
2864 Ok(a) => {
2865 link.channel_send_ok += 1;
2866 a
2867 }
2868 Err(e) => {
2869 log::debug!("Channel send failed: {:?}", e);
2870 match e {
2871 rns_core::channel::ChannelError::NotReady => link.channel_send_not_ready += 1,
2872 rns_core::channel::ChannelError::MessageTooBig => {
2873 link.channel_send_too_big += 1;
2874 }
2875 rns_core::channel::ChannelError::InvalidEnvelope => {
2876 link.channel_send_other_error += 1;
2877 }
2878 }
2879 return Err(e.to_string());
2880 }
2881 };
2882
2883 let _ = link;
2884 Ok(self.process_channel_actions(link_id, chan_actions, rng))
2885 }
2886
2887 pub fn tick(&mut self, rng: &mut dyn Rng) -> Vec<LinkManagerAction> {
2889 let now = time::now();
2890 let mut all_actions = Vec::new();
2891
2892 let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
2894
2895 for link_id in &link_ids {
2896 let link = match self.links.get_mut(link_id) {
2897 Some(l) => l,
2898 None => continue,
2899 };
2900
2901 let tick_actions = link.engine.tick(now);
2903 all_actions.extend(self.process_link_actions(link_id, &tick_actions));
2904
2905 let link = match self.links.get_mut(link_id) {
2907 Some(l) => l,
2908 None => continue,
2909 };
2910 if link.engine.needs_keepalive(now) {
2911 let flags = PacketFlags {
2913 header_type: constants::HEADER_1,
2914 context_flag: constants::FLAG_UNSET,
2915 transport_type: constants::TRANSPORT_BROADCAST,
2916 destination_type: constants::DESTINATION_LINK,
2917 packet_type: constants::PACKET_TYPE_DATA,
2918 };
2919 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
2920 flags,
2921 0,
2922 link_id,
2923 None,
2924 constants::CONTEXT_KEEPALIVE,
2925 &[],
2926 ) {
2927 all_actions.push(LinkManagerAction::SendPacket {
2928 raw,
2929 dest_type: constants::DESTINATION_LINK,
2930 attached_interface: None,
2931 });
2932 link.engine.record_outbound(now, true);
2933 }
2934 }
2935
2936 if let Some(channel) = link.channel.as_mut() {
2937 let chan_actions = channel.tick(now);
2938 let _ = channel;
2939 let _ = link;
2940 all_actions.extend(self.process_channel_actions(link_id, chan_actions, rng));
2941 }
2942 }
2943
2944 for link_id in &link_ids {
2946 let link = match self.links.get_mut(link_id) {
2947 Some(l) => l,
2948 None => continue,
2949 };
2950
2951 let mut sender_actions = Vec::new();
2953 for sender in &mut link.outgoing_resources {
2954 sender_actions.extend(sender.tick(now));
2955 }
2956
2957 let mut receiver_actions = Vec::new();
2959 for receiver in &mut link.incoming_resources {
2960 let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
2961 link.engine.decrypt(ciphertext).map_err(|_| ())
2962 };
2963 receiver_actions.extend(receiver.tick(now, &decrypt_fn, &Bzip2Compressor));
2964 }
2965
2966 link.outgoing_resources
2968 .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
2969 link.incoming_resources
2970 .retain(|r| r.status < rns_core::resource::ResourceStatus::Assembling);
2971 let active_split_hashes: Vec<[u8; 32]> = link
2972 .outgoing_resources
2973 .iter()
2974 .filter(|s| s.flags.split)
2975 .map(|s| s.original_hash)
2976 .collect();
2977 link.outgoing_splits.retain(|original_hash, split| {
2978 split.completed_segments < split.total_segments
2979 && active_split_hashes.contains(original_hash)
2980 });
2981
2982 let _ = link;
2983 all_actions.extend(self.process_resource_actions(link_id, sender_actions, rng));
2984 all_actions.extend(self.process_resource_actions(link_id, receiver_actions, rng));
2985 }
2986
2987 let closed: Vec<LinkId> = self
2989 .links
2990 .iter()
2991 .filter(|(_, l)| l.engine.state() == LinkState::Closed)
2992 .map(|(id, _)| *id)
2993 .collect();
2994 for id in closed {
2995 self.links.remove(&id);
2996 all_actions.push(LinkManagerAction::DeregisterLinkDest { link_id: id });
2997 }
2998
2999 all_actions
3000 }
3001
3002 pub fn is_link_destination(&self, dest_hash: &[u8; 16]) -> bool {
3004 self.links.contains_key(dest_hash) || self.link_destinations.contains_key(dest_hash)
3005 }
3006
3007 pub fn link_state(&self, link_id: &LinkId) -> Option<LinkState> {
3009 self.links.get(link_id).map(|l| l.engine.state())
3010 }
3011
3012 pub fn link_rtt(&self, link_id: &LinkId) -> Option<f64> {
3014 self.links.get(link_id).and_then(|l| l.engine.rtt())
3015 }
3016
3017 pub fn set_link_rtt(&mut self, link_id: &LinkId, rtt: f64) {
3019 if let Some(link) = self.links.get_mut(link_id) {
3020 link.engine.set_rtt(rtt);
3021 }
3022 }
3023
3024 pub fn record_link_inbound(&mut self, link_id: &LinkId) {
3026 if let Some(link) = self.links.get_mut(link_id) {
3027 link.engine.record_inbound(time::now());
3028 }
3029 }
3030
3031 pub fn set_link_mtu(&mut self, link_id: &LinkId, mtu: u32) {
3033 if let Some(link) = self.links.get_mut(link_id) {
3034 link.engine.set_mtu(mtu);
3035 }
3036 }
3037
3038 pub fn link_count(&self) -> usize {
3040 self.links.len()
3041 }
3042
3043 pub fn resource_transfer_count(&self) -> usize {
3045 self.links
3046 .values()
3047 .map(|managed| {
3048 managed
3049 .incoming_resources
3050 .iter()
3051 .filter(|resource| !resource.flags.split)
3052 .count()
3053 + managed.incoming_splits.len()
3054 + managed
3055 .outgoing_resources
3056 .iter()
3057 .filter(|resource| !resource.flags.split)
3058 .count()
3059 + managed.outgoing_splits.len()
3060 })
3061 .sum()
3062 }
3063
3064 pub fn cancel_all_resources(&mut self, rng: &mut dyn Rng) -> Vec<LinkManagerAction> {
3066 let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
3067 let mut all_actions = Vec::new();
3068
3069 for link_id in &link_ids {
3070 let link = match self.links.get_mut(link_id) {
3071 Some(l) => l,
3072 None => continue,
3073 };
3074
3075 let mut sender_actions = Vec::new();
3076 for sender in &mut link.outgoing_resources {
3077 sender_actions.extend(sender.cancel());
3078 }
3079
3080 let mut receiver_actions = Vec::new();
3081 for receiver in &mut link.incoming_resources {
3082 receiver_actions.extend(receiver.reject());
3083 }
3084
3085 link.outgoing_resources
3086 .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
3087 link.incoming_resources
3088 .retain(|r| r.status < rns_core::resource::ResourceStatus::Assembling);
3089 link.outgoing_splits.clear();
3090 link.incoming_splits.clear();
3091
3092 let _ = link;
3093 all_actions.extend(self.process_resource_actions(link_id, sender_actions, rng));
3094 all_actions.extend(self.process_resource_actions(link_id, receiver_actions, rng));
3095 }
3096
3097 all_actions
3098 }
3099
3100 pub fn link_entries(&self) -> Vec<crate::event::LinkInfoEntry> {
3102 self.links
3103 .iter()
3104 .map(|(link_id, managed)| {
3105 let state = match managed.engine.state() {
3106 LinkState::Pending => "pending",
3107 LinkState::Handshake => "handshake",
3108 LinkState::Active => "active",
3109 LinkState::Stale => "stale",
3110 LinkState::Closed => "closed",
3111 };
3112 crate::event::LinkInfoEntry {
3113 link_id: *link_id,
3114 state: state.to_string(),
3115 is_initiator: managed.engine.is_initiator(),
3116 dest_hash: managed.dest_hash,
3117 remote_identity: managed.remote_identity.as_ref().map(|(h, _)| *h),
3118 rtt: managed.engine.rtt(),
3119 channel_window: managed.channel.as_ref().map(|c| c.window()),
3120 channel_outstanding: managed.channel.as_ref().map(|c| c.outstanding()),
3121 pending_channel_packets: managed.pending_channel_packets.len(),
3122 channel_send_ok: managed.channel_send_ok,
3123 channel_send_not_ready: managed.channel_send_not_ready,
3124 channel_send_too_big: managed.channel_send_too_big,
3125 channel_send_other_error: managed.channel_send_other_error,
3126 channel_messages_received: managed.channel_messages_received,
3127 channel_proofs_sent: managed.channel_proofs_sent,
3128 channel_proofs_received: managed.channel_proofs_received,
3129 }
3130 })
3131 .collect()
3132 }
3133
3134 pub fn resource_entries(&self) -> Vec<crate::event::ResourceInfoEntry> {
3136 let mut entries = Vec::new();
3137 for (link_id, managed) in &self.links {
3138 let resource_sdu = Self::resource_sdu_for_link(managed);
3139 for split in managed.incoming_splits.values() {
3140 let (received, total) = Self::incoming_split_progress(split, resource_sdu);
3141 entries.push(crate::event::ResourceInfoEntry {
3142 link_id: *link_id,
3143 direction: "incoming".to_string(),
3144 total_parts: total,
3145 transferred_parts: received,
3146 complete: received >= total && total > 0,
3147 });
3148 }
3149 for recv in &managed.incoming_resources {
3150 if recv.flags.split {
3151 continue;
3152 }
3153 let (received, total) = recv.progress();
3154 entries.push(crate::event::ResourceInfoEntry {
3155 link_id: *link_id,
3156 direction: "incoming".to_string(),
3157 total_parts: total,
3158 transferred_parts: received,
3159 complete: received >= total && total > 0,
3160 });
3161 }
3162 for split in managed.outgoing_splits.values() {
3163 let (sent, total) = Self::outgoing_split_progress(split, resource_sdu);
3164 entries.push(crate::event::ResourceInfoEntry {
3165 link_id: *link_id,
3166 direction: "outgoing".to_string(),
3167 total_parts: total,
3168 transferred_parts: sent,
3169 complete: sent >= total && total > 0,
3170 });
3171 }
3172 for send in &managed.outgoing_resources {
3173 if send.flags.split {
3174 continue;
3175 }
3176 let total = send.total_parts();
3177 let sent = send.sent_parts;
3178 entries.push(crate::event::ResourceInfoEntry {
3179 link_id: *link_id,
3180 direction: "outgoing".to_string(),
3181 total_parts: total,
3182 transferred_parts: sent,
3183 complete: sent >= total && total > 0,
3184 });
3185 }
3186 }
3187 entries
3188 }
3189
3190 fn process_link_actions(
3192 &self,
3193 link_id: &LinkId,
3194 actions: &[LinkAction],
3195 ) -> Vec<LinkManagerAction> {
3196 let mut result = Vec::new();
3197 for action in actions {
3198 match action {
3199 LinkAction::StateChanged {
3200 new_state, reason, ..
3201 } => match new_state {
3202 LinkState::Closed => {
3203 result.push(LinkManagerAction::LinkClosed {
3204 link_id: *link_id,
3205 reason: *reason,
3206 });
3207 }
3208 _ => {}
3209 },
3210 LinkAction::LinkEstablished {
3211 rtt, is_initiator, ..
3212 } => {
3213 let dest_hash = self
3214 .links
3215 .get(link_id)
3216 .map(|l| l.dest_hash)
3217 .unwrap_or([0u8; 16]);
3218 result.push(LinkManagerAction::LinkEstablished {
3219 link_id: *link_id,
3220 dest_hash,
3221 rtt: *rtt,
3222 is_initiator: *is_initiator,
3223 });
3224 }
3225 LinkAction::RemoteIdentified {
3226 identity_hash,
3227 public_key,
3228 ..
3229 } => {
3230 result.push(LinkManagerAction::RemoteIdentified {
3231 link_id: *link_id,
3232 identity_hash: *identity_hash,
3233 public_key: *public_key,
3234 });
3235 }
3236 LinkAction::DataReceived { .. } => {
3237 }
3239 }
3240 }
3241 result
3242 }
3243
3244 fn process_channel_actions(
3246 &mut self,
3247 link_id: &LinkId,
3248 actions: Vec<rns_core::channel::ChannelAction>,
3249 rng: &mut dyn Rng,
3250 ) -> Vec<LinkManagerAction> {
3251 let mut result = Vec::new();
3252 for action in actions {
3253 match action {
3254 rns_core::channel::ChannelAction::SendOnLink { raw, sequence } => {
3255 let encrypted = match self.links.get(link_id) {
3259 Some(link) => match link.engine.encrypt(&raw, rng) {
3260 Ok(encrypted) => encrypted,
3261 Err(_) => {
3262 if let Some(link_mut) = self.links.get_mut(link_id) {
3263 if let Some(channel) = link_mut.channel.as_mut() {
3264 channel.cancel_send(sequence);
3265 }
3266 }
3267 continue;
3268 }
3269 },
3270 None => continue,
3271 };
3272 let flags = PacketFlags {
3273 header_type: constants::HEADER_1,
3274 context_flag: constants::FLAG_UNSET,
3275 transport_type: constants::TRANSPORT_BROADCAST,
3276 destination_type: constants::DESTINATION_LINK,
3277 packet_type: constants::PACKET_TYPE_DATA,
3278 };
3279 match RawPacket::pack_raw_with_hash(
3280 flags,
3281 0,
3282 link_id,
3283 None,
3284 constants::CONTEXT_CHANNEL,
3285 &encrypted,
3286 ) {
3287 Ok((raw_bytes, packet_hash)) => {
3288 if let Some(link_mut) = self.links.get_mut(link_id) {
3289 link_mut
3290 .pending_channel_packets
3291 .insert(packet_hash, sequence);
3292 }
3293 result.push(LinkManagerAction::SendPacket {
3294 raw: raw_bytes,
3295 dest_type: constants::DESTINATION_LINK,
3296 attached_interface: None,
3297 });
3298 }
3299 Err(_) => {
3300 if let Some(link_mut) = self.links.get_mut(link_id) {
3301 if let Some(channel) = link_mut.channel.as_mut() {
3302 channel.cancel_send(sequence);
3303 }
3304 }
3305 }
3306 }
3307 }
3308 rns_core::channel::ChannelAction::MessageReceived {
3309 msgtype, payload, ..
3310 } => {
3311 result.push(LinkManagerAction::ChannelMessageReceived {
3312 link_id: *link_id,
3313 msgtype,
3314 payload,
3315 });
3316 }
3317 rns_core::channel::ChannelAction::TeardownLink => {
3318 result.push(LinkManagerAction::LinkClosed {
3319 link_id: *link_id,
3320 reason: Some(TeardownReason::Timeout),
3321 });
3322 }
3323 }
3324 }
3325 result
3326 }
3327}
3328
3329impl Default for LinkManager {
3330 fn default() -> Self {
3331 Self::new()
3332 }
3333}
3334
3335fn compute_path_hash(path: &str) -> [u8; 16] {
3338 let full = rns_core::hash::full_hash(path.as_bytes());
3339 let mut result = [0u8; 16];
3340 result.copy_from_slice(&full[..16]);
3341 result
3342}
3343
3344#[cfg(test)]
3345mod tests {
3346 use super::*;
3347 use rns_crypto::identity::Identity;
3348 use rns_crypto::{FixedRng, OsRng};
3349
3350 fn make_rng(seed: u8) -> FixedRng {
3351 FixedRng::new(&[seed; 128])
3352 }
3353
3354 fn make_dest_keys(rng: &mut dyn Rng) -> (Ed25519PrivateKey, [u8; 32]) {
3355 let sig_prv = Ed25519PrivateKey::generate(rng);
3356 let sig_pub_bytes = sig_prv.public_key().public_bytes();
3357 (sig_prv, sig_pub_bytes)
3358 }
3359
3360 #[test]
3361 fn test_register_link_destination() {
3362 let mut mgr = LinkManager::new();
3363 let mut rng = make_rng(0x01);
3364 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3365 let dest_hash = [0xDD; 16];
3366
3367 mgr.register_link_destination(
3368 dest_hash,
3369 sig_prv,
3370 sig_pub_bytes,
3371 ResourceStrategy::AcceptNone,
3372 );
3373 assert!(mgr.is_link_destination(&dest_hash));
3374
3375 mgr.deregister_link_destination(&dest_hash);
3376 assert!(!mgr.is_link_destination(&dest_hash));
3377 }
3378
3379 #[test]
3380 fn test_create_link() {
3381 let mut mgr = LinkManager::new();
3382 let mut rng = OsRng;
3383 let dest_hash = [0xDD; 16];
3384
3385 let sig_pub_bytes = [0xAA; 32]; let (link_id, actions) = mgr.create_link(
3387 &dest_hash,
3388 &sig_pub_bytes,
3389 1,
3390 constants::MTU as u32,
3391 &mut rng,
3392 );
3393 assert_ne!(link_id, [0u8; 16]);
3394 assert_eq!(actions.len(), 2);
3396 assert!(matches!(
3397 actions[0],
3398 LinkManagerAction::RegisterLinkDest { .. }
3399 ));
3400 assert!(matches!(actions[1], LinkManagerAction::SendPacket { .. }));
3401
3402 assert_eq!(mgr.link_state(&link_id), Some(LinkState::Pending));
3404 }
3405
3406 #[test]
3407 fn test_full_handshake_via_manager() {
3408 let mut rng = OsRng;
3409 let dest_hash = [0xDD; 16];
3410
3411 let mut responder_mgr = LinkManager::new();
3413 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3414 responder_mgr.register_link_destination(
3415 dest_hash,
3416 sig_prv,
3417 sig_pub_bytes,
3418 ResourceStrategy::AcceptNone,
3419 );
3420
3421 let mut initiator_mgr = LinkManager::new();
3423
3424 let (link_id, init_actions) = initiator_mgr.create_link(
3426 &dest_hash,
3427 &sig_pub_bytes,
3428 1,
3429 constants::MTU as u32,
3430 &mut rng,
3431 );
3432 assert_eq!(init_actions.len(), 2);
3433
3434 let linkrequest_raw = match &init_actions[1] {
3436 LinkManagerAction::SendPacket { raw, .. } => raw.clone(),
3437 _ => panic!("Expected SendPacket"),
3438 };
3439
3440 let lr_packet = RawPacket::unpack(&linkrequest_raw).unwrap();
3442
3443 let resp_actions = responder_mgr.handle_local_delivery(
3445 lr_packet.destination_hash,
3446 &linkrequest_raw,
3447 lr_packet.packet_hash,
3448 rns_core::transport::types::InterfaceId(0),
3449 &mut rng,
3450 );
3451 assert!(resp_actions.len() >= 2);
3453 assert!(matches!(
3454 resp_actions[0],
3455 LinkManagerAction::RegisterLinkDest { .. }
3456 ));
3457
3458 let lrproof_raw = match &resp_actions[1] {
3460 LinkManagerAction::SendPacket { raw, .. } => raw.clone(),
3461 _ => panic!("Expected SendPacket for LRPROOF"),
3462 };
3463
3464 let lrproof_packet = RawPacket::unpack(&lrproof_raw).unwrap();
3466 let init_actions2 = initiator_mgr.handle_local_delivery(
3467 lrproof_packet.destination_hash,
3468 &lrproof_raw,
3469 lrproof_packet.packet_hash,
3470 rns_core::transport::types::InterfaceId(0),
3471 &mut rng,
3472 );
3473
3474 let has_established = init_actions2
3476 .iter()
3477 .any(|a| matches!(a, LinkManagerAction::LinkEstablished { .. }));
3478 assert!(has_established, "Initiator should emit LinkEstablished");
3479
3480 let lrrtt_raw = init_actions2
3482 .iter()
3483 .find_map(|a| match a {
3484 LinkManagerAction::SendPacket { raw, .. } => Some(raw.clone()),
3485 _ => None,
3486 })
3487 .expect("Should have LRRTT SendPacket");
3488
3489 let lrrtt_packet = RawPacket::unpack(&lrrtt_raw).unwrap();
3491 let resp_link_id = lrrtt_packet.destination_hash;
3492 let resp_actions2 = responder_mgr.handle_local_delivery(
3493 resp_link_id,
3494 &lrrtt_raw,
3495 lrrtt_packet.packet_hash,
3496 rns_core::transport::types::InterfaceId(0),
3497 &mut rng,
3498 );
3499
3500 let has_established = resp_actions2
3501 .iter()
3502 .any(|a| matches!(a, LinkManagerAction::LinkEstablished { .. }));
3503 assert!(has_established, "Responder should emit LinkEstablished");
3504
3505 assert_eq!(initiator_mgr.link_state(&link_id), Some(LinkState::Active));
3507 assert_eq!(responder_mgr.link_state(&link_id), Some(LinkState::Active));
3508
3509 assert!(initiator_mgr.link_rtt(&link_id).is_some());
3511 assert!(responder_mgr.link_rtt(&link_id).is_some());
3512 }
3513
3514 #[test]
3515 fn test_encrypted_data_exchange() {
3516 let mut rng = OsRng;
3517 let dest_hash = [0xDD; 16];
3518 let mut resp_mgr = LinkManager::new();
3519 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3520 resp_mgr.register_link_destination(
3521 dest_hash,
3522 sig_prv,
3523 sig_pub_bytes,
3524 ResourceStrategy::AcceptNone,
3525 );
3526 let mut init_mgr = LinkManager::new();
3527
3528 let (link_id, init_actions) = init_mgr.create_link(
3530 &dest_hash,
3531 &sig_pub_bytes,
3532 1,
3533 constants::MTU as u32,
3534 &mut rng,
3535 );
3536 let lr_raw = extract_send_packet(&init_actions);
3537 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
3538 let resp_actions = resp_mgr.handle_local_delivery(
3539 lr_pkt.destination_hash,
3540 &lr_raw,
3541 lr_pkt.packet_hash,
3542 rns_core::transport::types::InterfaceId(0),
3543 &mut rng,
3544 );
3545 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
3546 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
3547 let init_actions2 = init_mgr.handle_local_delivery(
3548 lrproof_pkt.destination_hash,
3549 &lrproof_raw,
3550 lrproof_pkt.packet_hash,
3551 rns_core::transport::types::InterfaceId(0),
3552 &mut rng,
3553 );
3554 let lrrtt_raw = extract_any_send_packet(&init_actions2);
3555 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
3556 resp_mgr.handle_local_delivery(
3557 lrrtt_pkt.destination_hash,
3558 &lrrtt_raw,
3559 lrrtt_pkt.packet_hash,
3560 rns_core::transport::types::InterfaceId(0),
3561 &mut rng,
3562 );
3563
3564 let actions =
3566 init_mgr.send_on_link(&link_id, b"hello link!", constants::CONTEXT_NONE, &mut rng);
3567 assert_eq!(actions.len(), 1);
3568 assert!(matches!(actions[0], LinkManagerAction::SendPacket { .. }));
3569 }
3570
3571 #[test]
3572 fn test_request_response() {
3573 let mut rng = OsRng;
3574 let dest_hash = [0xDD; 16];
3575 let mut resp_mgr = LinkManager::new();
3576 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3577 resp_mgr.register_link_destination(
3578 dest_hash,
3579 sig_prv,
3580 sig_pub_bytes,
3581 ResourceStrategy::AcceptNone,
3582 );
3583
3584 resp_mgr.register_request_handler("/status", None, |_link_id, _path, _data, _remote| {
3586 Some(b"OK".to_vec())
3587 });
3588
3589 let mut init_mgr = LinkManager::new();
3590
3591 let (link_id, init_actions) = init_mgr.create_link(
3593 &dest_hash,
3594 &sig_pub_bytes,
3595 1,
3596 constants::MTU as u32,
3597 &mut rng,
3598 );
3599 let lr_raw = extract_send_packet(&init_actions);
3600 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
3601 let resp_actions = resp_mgr.handle_local_delivery(
3602 lr_pkt.destination_hash,
3603 &lr_raw,
3604 lr_pkt.packet_hash,
3605 rns_core::transport::types::InterfaceId(0),
3606 &mut rng,
3607 );
3608 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
3609 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
3610 let init_actions2 = init_mgr.handle_local_delivery(
3611 lrproof_pkt.destination_hash,
3612 &lrproof_raw,
3613 lrproof_pkt.packet_hash,
3614 rns_core::transport::types::InterfaceId(0),
3615 &mut rng,
3616 );
3617 let lrrtt_raw = extract_any_send_packet(&init_actions2);
3618 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
3619 resp_mgr.handle_local_delivery(
3620 lrrtt_pkt.destination_hash,
3621 &lrrtt_raw,
3622 lrrtt_pkt.packet_hash,
3623 rns_core::transport::types::InterfaceId(0),
3624 &mut rng,
3625 );
3626
3627 let req_actions = init_mgr.send_request(&link_id, "/status", b"query", &mut rng);
3629 assert_eq!(req_actions.len(), 1);
3630
3631 let req_raw = extract_send_packet_from(&req_actions);
3633 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
3634 let resp_actions = resp_mgr.handle_local_delivery(
3635 req_pkt.destination_hash,
3636 &req_raw,
3637 req_pkt.packet_hash,
3638 rns_core::transport::types::InterfaceId(0),
3639 &mut rng,
3640 );
3641
3642 let has_response = resp_actions
3644 .iter()
3645 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
3646 assert!(has_response, "Handler should produce a response packet");
3647 }
3648
3649 #[test]
3650 fn test_send_request_wraps_invalid_msgpack_data_as_bin() {
3651 use std::sync::{Arc, Mutex};
3652
3653 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
3654 let mut rng = OsRng;
3655
3656 let invalid = vec![0xC1];
3657 let expected = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(invalid.clone()));
3658 let captured = Arc::new(Mutex::new(None::<Vec<u8>>));
3659 let captured_for_handler = Arc::clone(&captured);
3660
3661 resp_mgr.register_request_handler("/bin", None, move |_link_id, _path, data, _remote| {
3662 *captured_for_handler.lock().unwrap() = Some(data.to_vec());
3663 Some(rns_core::msgpack::pack(&rns_core::msgpack::Value::Bool(
3664 true,
3665 )))
3666 });
3667
3668 let req_actions = init_mgr.send_request(&link_id, "/bin", &invalid, &mut rng);
3669 let req_raw = extract_send_packet_from(&req_actions);
3670 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
3671 let resp_actions = resp_mgr.handle_local_delivery(
3672 req_pkt.destination_hash,
3673 &req_raw,
3674 req_pkt.packet_hash,
3675 rns_core::transport::types::InterfaceId(0),
3676 &mut rng,
3677 );
3678
3679 assert!(
3680 resp_actions
3681 .iter()
3682 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. })),
3683 "handler should still produce a response"
3684 );
3685 assert_eq!(*captured.lock().unwrap(), Some(expected));
3686 }
3687
3688 #[test]
3689 fn test_invalid_response_bytes_are_returned_as_msgpack_bin() {
3690 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
3691 let mut rng = OsRng;
3692 let invalid_response = vec![0xC1];
3693 let expected =
3694 rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(invalid_response.clone()));
3695
3696 resp_mgr.register_request_handler("/invalid-response", None, {
3697 let invalid_response = invalid_response.clone();
3698 move |_link_id, _path, _data, _remote| Some(invalid_response.clone())
3699 });
3700
3701 let req_actions = init_mgr.send_request(&link_id, "/invalid-response", b"\xc0", &mut rng);
3702 let req_raw = extract_send_packet_from(&req_actions);
3703 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
3704 let resp_actions = resp_mgr.handle_local_delivery(
3705 req_pkt.destination_hash,
3706 &req_raw,
3707 req_pkt.packet_hash,
3708 rns_core::transport::types::InterfaceId(0),
3709 &mut rng,
3710 );
3711
3712 let resp_raw = extract_any_send_packet(&resp_actions);
3713 let resp_pkt = RawPacket::unpack(&resp_raw).unwrap();
3714 let init_actions = init_mgr.handle_local_delivery(
3715 resp_pkt.destination_hash,
3716 &resp_raw,
3717 resp_pkt.packet_hash,
3718 rns_core::transport::types::InterfaceId(0),
3719 &mut rng,
3720 );
3721
3722 let response_data = init_actions
3723 .iter()
3724 .find_map(|action| match action {
3725 LinkManagerAction::ResponseReceived { data, .. } => Some(data.clone()),
3726 _ => None,
3727 })
3728 .expect("initiator should receive a response");
3729 assert_eq!(response_data, expected);
3730 }
3731
3732 #[test]
3733 fn test_request_acl_deny_unidentified() {
3734 let mut rng = OsRng;
3735 let dest_hash = [0xDD; 16];
3736 let mut resp_mgr = LinkManager::new();
3737 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3738 resp_mgr.register_link_destination(
3739 dest_hash,
3740 sig_prv,
3741 sig_pub_bytes,
3742 ResourceStrategy::AcceptNone,
3743 );
3744
3745 resp_mgr.register_request_handler(
3747 "/restricted",
3748 Some(vec![[0xAA; 16]]),
3749 |_link_id, _path, _data, _remote| Some(b"secret".to_vec()),
3750 );
3751
3752 let mut init_mgr = LinkManager::new();
3753
3754 let (link_id, init_actions) = init_mgr.create_link(
3756 &dest_hash,
3757 &sig_pub_bytes,
3758 1,
3759 constants::MTU as u32,
3760 &mut rng,
3761 );
3762 let lr_raw = extract_send_packet(&init_actions);
3763 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
3764 let resp_actions = resp_mgr.handle_local_delivery(
3765 lr_pkt.destination_hash,
3766 &lr_raw,
3767 lr_pkt.packet_hash,
3768 rns_core::transport::types::InterfaceId(0),
3769 &mut rng,
3770 );
3771 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
3772 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
3773 let init_actions2 = init_mgr.handle_local_delivery(
3774 lrproof_pkt.destination_hash,
3775 &lrproof_raw,
3776 lrproof_pkt.packet_hash,
3777 rns_core::transport::types::InterfaceId(0),
3778 &mut rng,
3779 );
3780 let lrrtt_raw = extract_any_send_packet(&init_actions2);
3781 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
3782 resp_mgr.handle_local_delivery(
3783 lrrtt_pkt.destination_hash,
3784 &lrrtt_raw,
3785 lrrtt_pkt.packet_hash,
3786 rns_core::transport::types::InterfaceId(0),
3787 &mut rng,
3788 );
3789
3790 let req_actions = init_mgr.send_request(&link_id, "/restricted", b"query", &mut rng);
3792 let req_raw = extract_send_packet_from(&req_actions);
3793 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
3794 let resp_actions = resp_mgr.handle_local_delivery(
3795 req_pkt.destination_hash,
3796 &req_raw,
3797 req_pkt.packet_hash,
3798 rns_core::transport::types::InterfaceId(0),
3799 &mut rng,
3800 );
3801
3802 let has_response = resp_actions
3804 .iter()
3805 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
3806 assert!(!has_response, "Unidentified peer should be denied");
3807 }
3808
3809 #[test]
3810 fn test_teardown_link() {
3811 let mut rng = OsRng;
3812 let dest_hash = [0xDD; 16];
3813 let mut mgr = LinkManager::new();
3814
3815 let dummy_sig = [0xAA; 32];
3816 let (link_id, _) =
3817 mgr.create_link(&dest_hash, &dummy_sig, 1, constants::MTU as u32, &mut rng);
3818 assert_eq!(mgr.link_count(), 1);
3819
3820 let actions = mgr.teardown_link(&link_id);
3821 let has_close = actions
3822 .iter()
3823 .any(|a| matches!(a, LinkManagerAction::LinkClosed { .. }));
3824 assert!(has_close);
3825
3826 let tick_actions = mgr.tick(&mut rng);
3828 let has_deregister = tick_actions
3829 .iter()
3830 .any(|a| matches!(a, LinkManagerAction::DeregisterLinkDest { .. }));
3831 assert!(has_deregister);
3832 assert_eq!(mgr.link_count(), 0);
3833 }
3834
3835 #[test]
3836 fn test_identify_on_link() {
3837 let mut rng = OsRng;
3838 let dest_hash = [0xDD; 16];
3839 let mut resp_mgr = LinkManager::new();
3840 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3841 resp_mgr.register_link_destination(
3842 dest_hash,
3843 sig_prv,
3844 sig_pub_bytes,
3845 ResourceStrategy::AcceptNone,
3846 );
3847 let mut init_mgr = LinkManager::new();
3848
3849 let (link_id, init_actions) = init_mgr.create_link(
3851 &dest_hash,
3852 &sig_pub_bytes,
3853 1,
3854 constants::MTU as u32,
3855 &mut rng,
3856 );
3857 let lr_raw = extract_send_packet(&init_actions);
3858 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
3859 let resp_actions = resp_mgr.handle_local_delivery(
3860 lr_pkt.destination_hash,
3861 &lr_raw,
3862 lr_pkt.packet_hash,
3863 rns_core::transport::types::InterfaceId(0),
3864 &mut rng,
3865 );
3866 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
3867 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
3868 let init_actions2 = init_mgr.handle_local_delivery(
3869 lrproof_pkt.destination_hash,
3870 &lrproof_raw,
3871 lrproof_pkt.packet_hash,
3872 rns_core::transport::types::InterfaceId(0),
3873 &mut rng,
3874 );
3875 let lrrtt_raw = extract_any_send_packet(&init_actions2);
3876 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
3877 resp_mgr.handle_local_delivery(
3878 lrrtt_pkt.destination_hash,
3879 &lrrtt_raw,
3880 lrrtt_pkt.packet_hash,
3881 rns_core::transport::types::InterfaceId(0),
3882 &mut rng,
3883 );
3884
3885 let identity = Identity::new(&mut rng);
3887 let id_actions = init_mgr.identify(&link_id, &identity, &mut rng);
3888 assert_eq!(id_actions.len(), 1);
3889
3890 let id_raw = extract_send_packet_from(&id_actions);
3892 let id_pkt = RawPacket::unpack(&id_raw).unwrap();
3893 let resp_actions = resp_mgr.handle_local_delivery(
3894 id_pkt.destination_hash,
3895 &id_raw,
3896 id_pkt.packet_hash,
3897 rns_core::transport::types::InterfaceId(0),
3898 &mut rng,
3899 );
3900
3901 let has_identified = resp_actions
3902 .iter()
3903 .any(|a| matches!(a, LinkManagerAction::RemoteIdentified { .. }));
3904 assert!(has_identified, "Responder should emit RemoteIdentified");
3905 }
3906
3907 #[test]
3908 fn test_path_hash_computation() {
3909 let h1 = compute_path_hash("/status");
3910 let h2 = compute_path_hash("/path");
3911 assert_ne!(h1, h2);
3912
3913 assert_eq!(h1, compute_path_hash("/status"));
3915 }
3916
3917 #[test]
3918 fn test_link_count() {
3919 let mut mgr = LinkManager::new();
3920 let mut rng = OsRng;
3921
3922 assert_eq!(mgr.link_count(), 0);
3923
3924 let dummy_sig = [0xAA; 32];
3925 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
3926 assert_eq!(mgr.link_count(), 1);
3927
3928 mgr.create_link(&[0x22; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
3929 assert_eq!(mgr.link_count(), 2);
3930 }
3931
3932 fn extract_send_packet(actions: &[LinkManagerAction]) -> Vec<u8> {
3935 extract_send_packet_at(actions, actions.len() - 1)
3936 }
3937
3938 fn extract_send_packet_at(actions: &[LinkManagerAction], idx: usize) -> Vec<u8> {
3939 match &actions[idx] {
3940 LinkManagerAction::SendPacket { raw, .. } => raw.clone(),
3941 other => panic!("Expected SendPacket at index {}, got {:?}", idx, other),
3942 }
3943 }
3944
3945 fn extract_any_send_packet(actions: &[LinkManagerAction]) -> Vec<u8> {
3946 actions
3947 .iter()
3948 .find_map(|a| match a {
3949 LinkManagerAction::SendPacket { raw, .. } => Some(raw.clone()),
3950 _ => None,
3951 })
3952 .expect("Expected at least one SendPacket action")
3953 }
3954
3955 fn extract_send_packet_from(actions: &[LinkManagerAction]) -> Vec<u8> {
3956 extract_any_send_packet(actions)
3957 }
3958
3959 fn setup_active_link() -> (LinkManager, LinkManager, LinkId) {
3962 let mut rng = OsRng;
3963 let dest_hash = [0xDD; 16];
3964 let mut resp_mgr = LinkManager::new();
3965 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3966 resp_mgr.register_link_destination(
3967 dest_hash,
3968 sig_prv,
3969 sig_pub_bytes,
3970 ResourceStrategy::AcceptNone,
3971 );
3972 let mut init_mgr = LinkManager::new();
3973
3974 let (link_id, init_actions) = init_mgr.create_link(
3975 &dest_hash,
3976 &sig_pub_bytes,
3977 1,
3978 constants::MTU as u32,
3979 &mut rng,
3980 );
3981 let lr_raw = extract_send_packet(&init_actions);
3982 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
3983 let resp_actions = resp_mgr.handle_local_delivery(
3984 lr_pkt.destination_hash,
3985 &lr_raw,
3986 lr_pkt.packet_hash,
3987 rns_core::transport::types::InterfaceId(0),
3988 &mut rng,
3989 );
3990 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
3991 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
3992 let init_actions2 = init_mgr.handle_local_delivery(
3993 lrproof_pkt.destination_hash,
3994 &lrproof_raw,
3995 lrproof_pkt.packet_hash,
3996 rns_core::transport::types::InterfaceId(0),
3997 &mut rng,
3998 );
3999 let lrrtt_raw = extract_any_send_packet(&init_actions2);
4000 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
4001 resp_mgr.handle_local_delivery(
4002 lrrtt_pkt.destination_hash,
4003 &lrrtt_raw,
4004 lrrtt_pkt.packet_hash,
4005 rns_core::transport::types::InterfaceId(0),
4006 &mut rng,
4007 );
4008
4009 assert_eq!(init_mgr.link_state(&link_id), Some(LinkState::Active));
4010 assert_eq!(resp_mgr.link_state(&link_id), Some(LinkState::Active));
4011
4012 (init_mgr, resp_mgr, link_id)
4013 }
4014
4015 #[test]
4020 fn test_resource_strategy_default() {
4021 let mut mgr = LinkManager::new();
4022 let mut rng = OsRng;
4023 let dummy_sig = [0xAA; 32];
4024 let (link_id, _) =
4025 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
4026
4027 let link = mgr.links.get(&link_id).unwrap();
4029 assert_eq!(link.resource_strategy, ResourceStrategy::AcceptNone);
4030 }
4031
4032 #[test]
4033 fn test_set_resource_strategy() {
4034 let mut mgr = LinkManager::new();
4035 let mut rng = OsRng;
4036 let dummy_sig = [0xAA; 32];
4037 let (link_id, _) =
4038 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
4039
4040 mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
4041 assert_eq!(
4042 mgr.links.get(&link_id).unwrap().resource_strategy,
4043 ResourceStrategy::AcceptAll
4044 );
4045
4046 mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
4047 assert_eq!(
4048 mgr.links.get(&link_id).unwrap().resource_strategy,
4049 ResourceStrategy::AcceptApp
4050 );
4051 }
4052
4053 #[test]
4054 fn test_send_resource_on_active_link() {
4055 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4056 let mut rng = OsRng;
4057
4058 let data = vec![0xAB; 100]; let actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
4061
4062 let has_send = actions
4064 .iter()
4065 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
4066 assert!(
4067 has_send,
4068 "send_resource should emit advertisement SendPacket"
4069 );
4070 }
4071
4072 #[test]
4073 fn test_large_send_request_uses_request_resource() {
4074 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4075 let mut rng = OsRng;
4076
4077 let data = vec![0xAB; 2048];
4078 let actions = init_mgr.send_request(&link_id, "/large", &data, &mut rng);
4079
4080 let adv = first_resource_advertisement(&init_mgr, &link_id, &actions);
4081 assert!(adv.is_request());
4082 assert!(!adv.is_response());
4083 assert_eq!(adv.request_id.as_ref().map(Vec::len), Some(16));
4084
4085 let has_request_packet = actions.iter().any(|action| match action {
4086 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
4087 .map(|pkt| pkt.context == constants::CONTEXT_REQUEST)
4088 .unwrap_or(false),
4089 _ => false,
4090 });
4091 assert!(!has_request_packet);
4092 }
4093
4094 fn first_resource_advertisement(
4095 mgr: &LinkManager,
4096 link_id: &LinkId,
4097 actions: &[LinkManagerAction],
4098 ) -> rns_core::resource::ResourceAdvertisement {
4099 let adv_raw = actions
4100 .iter()
4101 .find_map(|action| match action {
4102 LinkManagerAction::SendPacket { raw, .. } => {
4103 let pkt = RawPacket::unpack(raw).ok()?;
4104 (pkt.context == constants::CONTEXT_RESOURCE_ADV).then_some(raw)
4105 }
4106 _ => None,
4107 })
4108 .expect("sender should emit a resource advertisement");
4109 let adv_pkt = RawPacket::unpack(adv_raw).unwrap();
4110 let plaintext = mgr
4111 .links
4112 .get(link_id)
4113 .unwrap()
4114 .engine
4115 .decrypt(&adv_pkt.data)
4116 .unwrap();
4117 rns_core::resource::ResourceAdvertisement::unpack(&plaintext).unwrap()
4118 }
4119
4120 fn deterministic_bytes(len: usize) -> Vec<u8> {
4121 let mut state = 0x1234_5678u32;
4122 (0..len)
4123 .map(|_| {
4124 state = state.wrapping_mul(1664525).wrapping_add(1013904223);
4125 (state >> 16) as u8
4126 })
4127 .collect()
4128 }
4129
4130 fn drive_link_manager_packets(
4131 init_mgr: &mut LinkManager,
4132 resp_mgr: &mut LinkManager,
4133 initial_actions: Vec<LinkManagerAction>,
4134 initial_source: char,
4135 rng: &mut dyn Rng,
4136 max_rounds: usize,
4137 ) -> (
4138 Option<Vec<u8>>,
4139 bool,
4140 Vec<(char, usize, usize)>,
4141 Vec<(char, String)>,
4142 usize,
4143 ) {
4144 let mut pending: Vec<(char, LinkManagerAction)> = initial_actions
4145 .into_iter()
4146 .map(|a| (initial_source, a))
4147 .collect();
4148 let mut rounds = 0;
4149 let mut received_data = None;
4150 let mut sender_completed = false;
4151 let mut progress = Vec::new();
4152 let mut failures = Vec::new();
4153
4154 while !pending.is_empty() && rounds < max_rounds {
4155 rounds += 1;
4156 let mut next = Vec::new();
4157 for (source, action) in pending.drain(..) {
4158 let LinkManagerAction::SendPacket { raw, .. } = action else {
4159 continue;
4160 };
4161 let pkt = RawPacket::unpack(&raw).unwrap();
4162 let target_actions = if source == 'i' {
4163 resp_mgr.handle_local_delivery(
4164 pkt.destination_hash,
4165 &raw,
4166 pkt.packet_hash,
4167 rns_core::transport::types::InterfaceId(0),
4168 rng,
4169 )
4170 } else {
4171 init_mgr.handle_local_delivery(
4172 pkt.destination_hash,
4173 &raw,
4174 pkt.packet_hash,
4175 rns_core::transport::types::InterfaceId(0),
4176 rng,
4177 )
4178 };
4179 let target_source = if source == 'i' { 'r' } else { 'i' };
4180 for action in &target_actions {
4181 match action {
4182 LinkManagerAction::ResourceReceived { data, .. } => {
4183 received_data = Some(data.clone());
4184 }
4185 LinkManagerAction::ResourceCompleted { .. } => {
4186 sender_completed = true;
4187 }
4188 LinkManagerAction::ResourceProgress {
4189 received, total, ..
4190 } => {
4191 progress.push((target_source, *received, *total));
4192 }
4193 LinkManagerAction::ResourceFailed { error, .. } => {
4194 failures.push((target_source, error.clone()));
4195 }
4196 _ => {}
4197 }
4198 }
4199 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
4200 }
4201 pending = next;
4202 }
4203
4204 (received_data, sender_completed, progress, failures, rounds)
4205 }
4206
4207 #[test]
4208 fn test_send_resource_auto_compress_option_controls_adv_flag() {
4209 let data = vec![0x41; 2048];
4210
4211 let (mut compressed_mgr, _resp_mgr, link_id) = setup_active_link();
4212 let mut rng = OsRng;
4213 let actions =
4214 compressed_mgr.send_resource_with_auto_compress(&link_id, &data, None, true, &mut rng);
4215 let adv = first_resource_advertisement(&compressed_mgr, &link_id, &actions);
4216 assert!(
4217 adv.flags.compressed,
4218 "compressible resource should compress"
4219 );
4220
4221 let (mut plain_mgr, _resp_mgr, link_id) = setup_active_link();
4222 let mut rng = OsRng;
4223 let actions =
4224 plain_mgr.send_resource_with_auto_compress(&link_id, &data, None, false, &mut rng);
4225 let adv = first_resource_advertisement(&plain_mgr, &link_id, &actions);
4226 assert!(
4227 !adv.flags.compressed,
4228 "auto_compress=false should keep resource uncompressed"
4229 );
4230 }
4231
4232 #[test]
4233 fn test_send_resource_on_inactive_link() {
4234 let mut mgr = LinkManager::new();
4235 let mut rng = OsRng;
4236 let dummy_sig = [0xAA; 32];
4237 let (link_id, _) =
4238 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
4239
4240 let actions = mgr.send_resource(&link_id, b"data", None, &mut rng);
4242 assert!(actions.is_empty(), "Cannot send resource on inactive link");
4243 }
4244
4245 #[test]
4246 fn test_send_resource_without_session_key_uses_encrypt_fallback_path() {
4247 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4248 let mut rng = OsRng;
4249 init_mgr
4250 .links
4251 .get_mut(&link_id)
4252 .unwrap()
4253 .engine
4254 .clear_session_for_testing();
4255
4256 let actions = init_mgr.send_resource(&link_id, b"data", None, &mut rng);
4257
4258 assert!(
4259 actions.is_empty(),
4260 "without a session key, no advertisement should be emitted"
4261 );
4262 assert_eq!(
4263 init_mgr
4264 .links
4265 .get(&link_id)
4266 .map(|managed| managed.outgoing_resources.len()),
4267 Some(1)
4268 );
4269 }
4270
4271 #[test]
4272 fn test_resource_adv_rejected_by_accept_none() {
4273 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4274 let mut rng = OsRng;
4275
4276 let data = vec![0xCD; 100];
4279 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
4280
4281 for action in &adv_actions {
4283 if let LinkManagerAction::SendPacket { raw, .. } = action {
4284 let pkt = RawPacket::unpack(raw).unwrap();
4285 let resp_actions = resp_mgr.handle_local_delivery(
4286 pkt.destination_hash,
4287 raw,
4288 pkt.packet_hash,
4289 rns_core::transport::types::InterfaceId(0),
4290 &mut rng,
4291 );
4292 let has_resource_received = resp_actions
4294 .iter()
4295 .any(|a| matches!(a, LinkManagerAction::ResourceReceived { .. }));
4296 assert!(
4297 !has_resource_received,
4298 "AcceptNone should not accept resource"
4299 );
4300 }
4301 }
4302 }
4303
4304 #[test]
4305 fn test_resource_adv_accepted_by_accept_all() {
4306 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4307 let mut rng = OsRng;
4308
4309 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
4311
4312 let data = vec![0xCD; 100];
4314 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
4315
4316 for action in &adv_actions {
4318 if let LinkManagerAction::SendPacket { raw, .. } = action {
4319 let pkt = RawPacket::unpack(raw).unwrap();
4320 let resp_actions = resp_mgr.handle_local_delivery(
4321 pkt.destination_hash,
4322 raw,
4323 pkt.packet_hash,
4324 rns_core::transport::types::InterfaceId(0),
4325 &mut rng,
4326 );
4327 let has_send = resp_actions
4329 .iter()
4330 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
4331 assert!(has_send, "AcceptAll should accept and request parts");
4332 }
4333 }
4334 }
4335
4336 #[test]
4337 fn test_resource_accept_app_query() {
4338 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4339 let mut rng = OsRng;
4340
4341 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
4343
4344 let data = vec![0xCD; 100];
4346 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
4347
4348 let mut got_query = false;
4350 for action in &adv_actions {
4351 if let LinkManagerAction::SendPacket { raw, .. } = action {
4352 let pkt = RawPacket::unpack(raw).unwrap();
4353 let resp_actions = resp_mgr.handle_local_delivery(
4354 pkt.destination_hash,
4355 raw,
4356 pkt.packet_hash,
4357 rns_core::transport::types::InterfaceId(0),
4358 &mut rng,
4359 );
4360 for a in &resp_actions {
4361 if matches!(a, LinkManagerAction::ResourceAcceptQuery { .. }) {
4362 got_query = true;
4363 }
4364 }
4365 }
4366 }
4367 assert!(got_query, "AcceptApp should emit ResourceAcceptQuery");
4368 }
4369
4370 #[test]
4371 fn test_resource_accept_app_accept() {
4372 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4373 let mut rng = OsRng;
4374
4375 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
4376
4377 let data = vec![0xCD; 100];
4378 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
4379
4380 for action in &adv_actions {
4381 if let LinkManagerAction::SendPacket { raw, .. } = action {
4382 let pkt = RawPacket::unpack(raw).unwrap();
4383 let resp_actions = resp_mgr.handle_local_delivery(
4384 pkt.destination_hash,
4385 raw,
4386 pkt.packet_hash,
4387 rns_core::transport::types::InterfaceId(0),
4388 &mut rng,
4389 );
4390 for a in &resp_actions {
4391 if let LinkManagerAction::ResourceAcceptQuery {
4392 link_id: lid,
4393 resource_hash,
4394 ..
4395 } = a
4396 {
4397 let accept_actions =
4399 resp_mgr.accept_resource(lid, resource_hash, true, &mut rng);
4400 let has_send = accept_actions
4402 .iter()
4403 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
4404 assert!(
4405 has_send,
4406 "Accepting resource should produce request for parts"
4407 );
4408 }
4409 }
4410 }
4411 }
4412 }
4413
4414 #[test]
4415 fn test_resource_accept_app_reject() {
4416 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4417 let mut rng = OsRng;
4418
4419 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
4420
4421 let data = vec![0xCD; 100];
4422 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
4423
4424 for action in &adv_actions {
4425 if let LinkManagerAction::SendPacket { raw, .. } = action {
4426 let pkt = RawPacket::unpack(raw).unwrap();
4427 let resp_actions = resp_mgr.handle_local_delivery(
4428 pkt.destination_hash,
4429 raw,
4430 pkt.packet_hash,
4431 rns_core::transport::types::InterfaceId(0),
4432 &mut rng,
4433 );
4434 for a in &resp_actions {
4435 if let LinkManagerAction::ResourceAcceptQuery {
4436 link_id: lid,
4437 resource_hash,
4438 ..
4439 } = a
4440 {
4441 let reject_actions =
4443 resp_mgr.accept_resource(lid, resource_hash, false, &mut rng);
4444 let has_resource_received = reject_actions
4447 .iter()
4448 .any(|a| matches!(a, LinkManagerAction::ResourceReceived { .. }));
4449 assert!(!has_resource_received);
4450 }
4451 }
4452 }
4453 }
4454 }
4455
4456 #[test]
4457 fn test_resource_full_transfer() {
4458 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4459 let mut rng = OsRng;
4460
4461 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
4463
4464 let original_data = b"Hello, Resource Transfer!".to_vec();
4466 let adv_actions = init_mgr.send_resource(&link_id, &original_data, None, &mut rng);
4467
4468 let mut pending: Vec<(char, LinkManagerAction)> =
4471 adv_actions.into_iter().map(|a| ('i', a)).collect();
4472 let mut rounds = 0;
4473 let max_rounds = 50;
4474 let mut resource_received = false;
4475 let mut sender_completed = false;
4476
4477 while !pending.is_empty() && rounds < max_rounds {
4478 rounds += 1;
4479 let mut next: Vec<(char, LinkManagerAction)> = Vec::new();
4480
4481 for (source, action) in pending.drain(..) {
4482 if let LinkManagerAction::SendPacket { raw, .. } = action {
4483 let pkt = RawPacket::unpack(&raw).unwrap();
4484
4485 let target_actions = if source == 'i' {
4487 resp_mgr.handle_local_delivery(
4488 pkt.destination_hash,
4489 &raw,
4490 pkt.packet_hash,
4491 rns_core::transport::types::InterfaceId(0),
4492 &mut rng,
4493 )
4494 } else {
4495 init_mgr.handle_local_delivery(
4496 pkt.destination_hash,
4497 &raw,
4498 pkt.packet_hash,
4499 rns_core::transport::types::InterfaceId(0),
4500 &mut rng,
4501 )
4502 };
4503
4504 let target_source = if source == 'i' { 'r' } else { 'i' };
4505 for a in &target_actions {
4506 match a {
4507 LinkManagerAction::ResourceReceived { data, .. } => {
4508 assert_eq!(*data, original_data);
4509 resource_received = true;
4510 }
4511 LinkManagerAction::ResourceCompleted { .. } => {
4512 sender_completed = true;
4513 }
4514 _ => {}
4515 }
4516 }
4517 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
4518 }
4519 }
4520 pending = next;
4521 }
4522
4523 assert!(
4524 resource_received,
4525 "Responder should receive resource data (rounds={})",
4526 rounds
4527 );
4528 assert!(
4529 sender_completed,
4530 "Sender should get completion proof (rounds={})",
4531 rounds
4532 );
4533 }
4534
4535 #[test]
4536 fn test_split_resource_advertisement_and_progress_entries() {
4537 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4538 let mut rng = OsRng;
4539 let data = deterministic_bytes(constants::RESOURCE_MAX_EFFICIENT_SIZE + 1024);
4540
4541 let actions =
4542 init_mgr.send_resource_with_auto_compress(&link_id, &data, None, false, &mut rng);
4543 let adv = first_resource_advertisement(&init_mgr, &link_id, &actions);
4544
4545 assert!(adv.flags.split);
4546 assert_eq!(adv.segment_index, 1);
4547 assert_eq!(adv.total_segments, 2);
4548 assert_eq!(adv.data_size, data.len() as u64);
4549
4550 let managed = init_mgr.links.get(&link_id).unwrap();
4551 assert_eq!(managed.outgoing_splits.len(), 1);
4552 assert_eq!(
4553 managed
4554 .outgoing_resources
4555 .iter()
4556 .filter(|sender| sender.flags.split)
4557 .count(),
4558 2
4559 );
4560
4561 let entries = init_mgr.resource_entries();
4562 assert_eq!(entries.len(), 1);
4563 assert_eq!(entries[0].direction, "outgoing");
4564 assert!(entries[0].total_parts > managed.outgoing_resources[0].total_parts());
4565 assert_eq!(entries[0].transferred_parts, 0);
4566 }
4567
4568 #[test]
4569 fn test_split_resource_full_transfer_and_monotonic_progress() {
4570 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4571 let mut rng = OsRng;
4572 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
4573
4574 let original_data = deterministic_bytes(constants::RESOURCE_MAX_EFFICIENT_SIZE + 2048);
4575 let initial_actions = init_mgr.send_resource_with_auto_compress(
4576 &link_id,
4577 &original_data,
4578 None,
4579 false,
4580 &mut rng,
4581 );
4582
4583 let (received_data, sender_completed, progress, failures, rounds) =
4584 drive_link_manager_packets(
4585 &mut init_mgr,
4586 &mut resp_mgr,
4587 initial_actions,
4588 'i',
4589 &mut rng,
4590 10_000,
4591 );
4592
4593 assert!(
4594 received_data.as_ref().is_some_and(|data| data == &original_data),
4595 "split transfer did not deliver payload in {rounds} rounds; sender_completed={sender_completed}; failures={failures:?}; last_progress={:?}; init_entries={:?}; resp_entries={:?}",
4596 progress.last(),
4597 init_mgr.resource_entries(),
4598 resp_mgr.resource_entries()
4599 );
4600 assert!(
4601 sender_completed,
4602 "sender did not complete in {rounds} rounds"
4603 );
4604 assert!(
4605 progress
4606 .iter()
4607 .any(|(_, received, total)| received == total),
4608 "expected final progress update"
4609 );
4610
4611 let mut init_last = 0;
4612 let mut resp_last = 0;
4613 for (side, received, total) in progress {
4614 assert!(received <= total);
4615 match side {
4616 'i' => {
4617 assert!(
4618 received >= init_last,
4619 "initiator progress regressed from {init_last} to {received}"
4620 );
4621 init_last = received;
4622 }
4623 'r' => {
4624 assert!(
4625 received >= resp_last,
4626 "responder progress regressed from {resp_last} to {received}"
4627 );
4628 resp_last = received;
4629 }
4630 _ => unreachable!(),
4631 }
4632 }
4633 }
4634
4635 #[test]
4636 fn test_split_resource_accept_app_queries_only_first_segment() {
4637 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4638 let mut rng = OsRng;
4639 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
4640
4641 let original_data = deterministic_bytes(constants::RESOURCE_MAX_EFFICIENT_SIZE + 1024);
4642 let adv_actions = init_mgr.send_resource_with_auto_compress(
4643 &link_id,
4644 &original_data,
4645 None,
4646 false,
4647 &mut rng,
4648 );
4649 let adv_raw = extract_any_send_packet(&adv_actions);
4650 let adv_pkt = RawPacket::unpack(&adv_raw).unwrap();
4651 let query_actions = resp_mgr.handle_local_delivery(
4652 adv_pkt.destination_hash,
4653 &adv_raw,
4654 adv_pkt.packet_hash,
4655 rns_core::transport::types::InterfaceId(0),
4656 &mut rng,
4657 );
4658
4659 let queries: Vec<_> = query_actions
4660 .iter()
4661 .filter_map(|action| match action {
4662 LinkManagerAction::ResourceAcceptQuery { resource_hash, .. } => {
4663 Some(resource_hash.clone())
4664 }
4665 _ => None,
4666 })
4667 .collect();
4668 assert_eq!(queries.len(), 1);
4669
4670 let accept_actions = resp_mgr.accept_resource(&link_id, &queries[0], true, &mut rng);
4671 let (received_data, sender_completed, _progress, failures, rounds) =
4672 drive_link_manager_packets(
4673 &mut init_mgr,
4674 &mut resp_mgr,
4675 accept_actions,
4676 'r',
4677 &mut rng,
4678 10_000,
4679 );
4680
4681 assert!(
4682 failures.is_empty(),
4683 "split AcceptApp transfer failed: {failures:?}"
4684 );
4685 assert!(
4686 received_data
4687 .as_ref()
4688 .is_some_and(|data| data == &original_data),
4689 "split AcceptApp transfer did not deliver in {rounds} rounds"
4690 );
4691 assert!(sender_completed);
4692 }
4693
4694 #[test]
4695 fn test_resource_cancel_icl() {
4696 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4697 let mut rng = OsRng;
4698
4699 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
4700
4701 let data = vec![0xAB; 2000];
4703 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
4704
4705 for action in &adv_actions {
4707 if let LinkManagerAction::SendPacket { raw, .. } = action {
4708 let pkt = RawPacket::unpack(raw).unwrap();
4709 resp_mgr.handle_local_delivery(
4710 pkt.destination_hash,
4711 raw,
4712 pkt.packet_hash,
4713 rns_core::transport::types::InterfaceId(0),
4714 &mut rng,
4715 );
4716 }
4717 }
4718
4719 assert!(!resp_mgr
4721 .links
4722 .get(&link_id)
4723 .unwrap()
4724 .incoming_resources
4725 .is_empty());
4726
4727 let icl_actions = resp_mgr.handle_resource_icl(&link_id);
4729
4730 let has_failed = icl_actions
4732 .iter()
4733 .any(|a| matches!(a, LinkManagerAction::ResourceFailed { .. }));
4734 assert!(has_failed, "ICL should produce ResourceFailed");
4735 }
4736
4737 #[test]
4738 fn test_resource_cancel_rcl() {
4739 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4740 let mut rng = OsRng;
4741
4742 let data = vec![0xAB; 2000];
4744 init_mgr.send_resource(&link_id, &data, None, &mut rng);
4745
4746 assert!(!init_mgr
4748 .links
4749 .get(&link_id)
4750 .unwrap()
4751 .outgoing_resources
4752 .is_empty());
4753
4754 let rcl_actions = init_mgr.handle_resource_rcl(&link_id);
4756
4757 let has_failed = rcl_actions
4758 .iter()
4759 .any(|a| matches!(a, LinkManagerAction::ResourceFailed { .. }));
4760 assert!(has_failed, "RCL should produce ResourceFailed");
4761 }
4762
4763 #[test]
4764 fn test_cancel_all_resources_clears_active_transfers() {
4765 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4766 let mut rng = OsRng;
4767
4768 let actions = init_mgr.send_resource(&link_id, b"resource body", None, &mut rng);
4769 assert!(!actions.is_empty());
4770 assert_eq!(init_mgr.resource_transfer_count(), 1);
4771
4772 let cancel_actions = init_mgr.cancel_all_resources(&mut rng);
4773
4774 assert_eq!(init_mgr.resource_transfer_count(), 0);
4775 assert!(cancel_actions
4776 .iter()
4777 .any(|action| matches!(action, LinkManagerAction::SendPacket { .. })));
4778 }
4779
4780 #[test]
4781 fn test_resource_tick_cleans_up() {
4782 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4783 let mut rng = OsRng;
4784
4785 let data = vec![0xAB; 100];
4786 init_mgr.send_resource(&link_id, &data, None, &mut rng);
4787
4788 assert!(!init_mgr
4789 .links
4790 .get(&link_id)
4791 .unwrap()
4792 .outgoing_resources
4793 .is_empty());
4794
4795 init_mgr.handle_resource_rcl(&link_id);
4797
4798 init_mgr.tick(&mut rng);
4800
4801 assert!(
4802 init_mgr
4803 .links
4804 .get(&link_id)
4805 .unwrap()
4806 .outgoing_resources
4807 .is_empty(),
4808 "Tick should clean up completed/failed outgoing resources"
4809 );
4810 }
4811
4812 #[test]
4813 fn test_build_link_packet() {
4814 let (init_mgr, _resp_mgr, link_id) = setup_active_link();
4815
4816 let actions =
4817 init_mgr.build_link_packet(&link_id, constants::CONTEXT_RESOURCE, b"test data");
4818 assert_eq!(actions.len(), 1);
4819 if let LinkManagerAction::SendPacket { raw, dest_type, .. } = &actions[0] {
4820 let pkt = RawPacket::unpack(raw).unwrap();
4821 assert_eq!(pkt.context, constants::CONTEXT_RESOURCE);
4822 assert_eq!(*dest_type, constants::DESTINATION_LINK);
4823 } else {
4824 panic!("Expected SendPacket");
4825 }
4826 }
4827
4828 #[test]
4829 fn test_build_link_packet_returns_empty_when_mtu_too_small() {
4830 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4831 init_mgr.set_link_mtu(&link_id, 84);
4832
4833 let actions =
4834 init_mgr.build_link_packet(&link_id, constants::CONTEXT_RESOURCE, &[0xAA; 200]);
4835 assert!(actions.is_empty(), "oversized packet should not be built");
4836 }
4837
4838 #[test]
4839 fn test_process_resource_actions_encrypted_variants_drop_on_encrypt_failure() {
4840 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4841 let mut rng = OsRng;
4842 init_mgr
4843 .links
4844 .get_mut(&link_id)
4845 .unwrap()
4846 .engine
4847 .clear_session_for_testing();
4848
4849 let cases = vec![
4850 ResourceAction::SendAdvertisement(vec![1, 2, 3]),
4851 ResourceAction::SendRequest(vec![4, 5, 6]),
4852 ResourceAction::SendHmu(vec![7, 8, 9]),
4853 ResourceAction::SendCancelInitiator(vec![13, 14, 15]),
4854 ResourceAction::SendCancelReceiver(vec![16, 17, 18]),
4855 ];
4856
4857 for action in cases {
4858 let out = init_mgr.process_resource_actions(&link_id, vec![action], &mut rng);
4859 assert!(
4860 out.is_empty(),
4861 "encrypt failure should suppress packet emission"
4862 );
4863 }
4864 }
4865
4866 #[test]
4867 fn test_resource_proof_is_raw_proof_packet() {
4868 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4869 let mut rng = OsRng;
4870 let proof = vec![0x22; 64];
4871
4872 let out = init_mgr.process_resource_actions(
4873 &link_id,
4874 vec![ResourceAction::SendProof(proof.clone())],
4875 &mut rng,
4876 );
4877 let raw = extract_any_send_packet(&out);
4878 let pkt = RawPacket::unpack(&raw).unwrap();
4879
4880 assert_eq!(pkt.flags.packet_type, constants::PACKET_TYPE_PROOF);
4881 assert_eq!(pkt.context, constants::CONTEXT_RESOURCE_PRF);
4882 assert_eq!(pkt.data, proof);
4883 }
4884
4885 #[test]
4890 fn test_channel_message_delivery() {
4891 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4892 let mut rng = OsRng;
4893
4894 let chan_actions = init_mgr
4896 .send_channel_message(&link_id, 42, b"channel data", &mut rng)
4897 .expect("active link channel send should succeed");
4898 assert!(!chan_actions.is_empty());
4899
4900 let mut got_channel_msg = false;
4902 for action in &chan_actions {
4903 if let LinkManagerAction::SendPacket { raw, .. } = action {
4904 let pkt = RawPacket::unpack(raw).unwrap();
4905 let resp_actions = resp_mgr.handle_local_delivery(
4906 pkt.destination_hash,
4907 raw,
4908 pkt.packet_hash,
4909 rns_core::transport::types::InterfaceId(0),
4910 &mut rng,
4911 );
4912 for a in &resp_actions {
4913 if let LinkManagerAction::ChannelMessageReceived {
4914 msgtype, payload, ..
4915 } = a
4916 {
4917 assert_eq!(*msgtype, 42);
4918 assert_eq!(*payload, b"channel data");
4919 got_channel_msg = true;
4920 }
4921 }
4922 }
4923 }
4924 assert!(got_channel_msg, "Responder should receive channel message");
4925 }
4926
4927 #[test]
4928 fn test_channel_send_drops_packet_when_encrypt_fails() {
4929 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4930 let mut rng = OsRng;
4931 init_mgr
4932 .links
4933 .get_mut(&link_id)
4934 .unwrap()
4935 .engine
4936 .clear_session_for_testing();
4937
4938 let actions = init_mgr
4939 .send_channel_message(&link_id, 42, b"channel data", &mut rng)
4940 .expect("channel should still accept the message locally");
4941
4942 assert!(
4943 actions.is_empty(),
4944 "encrypt failure should suppress channel packet"
4945 );
4946 assert!(
4947 init_mgr
4948 .links
4949 .get(&link_id)
4950 .unwrap()
4951 .pending_channel_packets
4952 .is_empty(),
4953 "failed packet encryption must not track a pending channel proof"
4954 );
4955 }
4956
4957 #[test]
4958 fn test_channel_proof_reopens_send_window() {
4959 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4960 let mut rng = OsRng;
4961
4962 init_mgr
4963 .send_channel_message(&link_id, 42, b"first", &mut rng)
4964 .expect("first send should succeed");
4965 init_mgr
4966 .send_channel_message(&link_id, 42, b"second", &mut rng)
4967 .expect("second send should succeed");
4968
4969 let err = init_mgr
4970 .send_channel_message(&link_id, 42, b"third", &mut rng)
4971 .expect_err("third send should hit the initial channel window");
4972 assert_eq!(err, "Channel is not ready to send");
4973
4974 let queued_packets = init_mgr
4975 .links
4976 .get(&link_id)
4977 .unwrap()
4978 .pending_channel_packets
4979 .clone();
4980 assert_eq!(queued_packets.len(), 2);
4981 for tracked_hash in queued_packets.keys().take(1) {
4982 let mut proof_data = Vec::with_capacity(96);
4983 proof_data.extend_from_slice(tracked_hash);
4984 proof_data.extend_from_slice(&[0x11; 64]);
4985 let flags = PacketFlags {
4986 header_type: constants::HEADER_1,
4987 context_flag: constants::FLAG_UNSET,
4988 transport_type: constants::TRANSPORT_BROADCAST,
4989 destination_type: constants::DESTINATION_LINK,
4990 packet_type: constants::PACKET_TYPE_PROOF,
4991 };
4992 let proof = RawPacket::pack(
4993 flags,
4994 0,
4995 &link_id,
4996 None,
4997 constants::CONTEXT_NONE,
4998 &proof_data,
4999 )
5000 .expect("proof packet should pack");
5001 let ack_actions = init_mgr.handle_local_delivery(
5002 link_id,
5003 &proof.raw,
5004 proof.packet_hash,
5005 rns_core::transport::types::InterfaceId(0),
5006 &mut rng,
5007 );
5008 assert!(
5009 ack_actions.is_empty(),
5010 "proof delivery should only update channel state"
5011 );
5012 }
5013
5014 init_mgr
5015 .send_channel_message(&link_id, 42, b"third", &mut rng)
5016 .expect("proof should free one channel slot");
5017 }
5018
5019 #[test]
5020 fn test_generic_link_data_delivery() {
5021 let (init_mgr, mut resp_mgr, link_id) = setup_active_link();
5022 let mut rng = OsRng;
5023
5024 let actions = init_mgr.send_on_link(&link_id, b"raw stuff", 0x42, &mut rng);
5026 assert_eq!(actions.len(), 1);
5027
5028 let raw = extract_any_send_packet(&actions);
5030 let pkt = RawPacket::unpack(&raw).unwrap();
5031 let resp_actions = resp_mgr.handle_local_delivery(
5032 pkt.destination_hash,
5033 &raw,
5034 pkt.packet_hash,
5035 rns_core::transport::types::InterfaceId(0),
5036 &mut rng,
5037 );
5038
5039 let has_data = resp_actions
5040 .iter()
5041 .any(|a| matches!(a, LinkManagerAction::LinkDataReceived { context: 0x42, .. }));
5042 assert!(
5043 has_data,
5044 "Responder should receive LinkDataReceived for unknown context"
5045 );
5046 }
5047
5048 #[test]
5049 fn test_invalid_encrypted_contexts_are_ignored() {
5050 let (_init_mgr, mut resp_mgr, link_id) = setup_active_link();
5051 let mut rng = OsRng;
5052 let contexts = [
5053 constants::CONTEXT_CHANNEL,
5054 constants::CONTEXT_REQUEST,
5055 constants::CONTEXT_RESPONSE,
5056 constants::CONTEXT_RESOURCE_ADV,
5057 constants::CONTEXT_RESOURCE_REQ,
5058 constants::CONTEXT_RESOURCE_HMU,
5059 constants::CONTEXT_RESOURCE_PRF,
5060 0x42,
5061 ];
5062
5063 for context in contexts {
5064 let flags = PacketFlags {
5065 header_type: constants::HEADER_1,
5066 context_flag: constants::FLAG_UNSET,
5067 transport_type: constants::TRANSPORT_BROADCAST,
5068 destination_type: constants::DESTINATION_LINK,
5069 packet_type: constants::PACKET_TYPE_DATA,
5070 };
5071 let pkt = RawPacket::pack(flags, 0, &link_id, None, context, b"invalid-ciphertext")
5072 .expect("test packet should pack");
5073 let actions = resp_mgr.handle_local_delivery(
5074 pkt.destination_hash,
5075 &pkt.raw,
5076 pkt.packet_hash,
5077 rns_core::transport::types::InterfaceId(0),
5078 &mut rng,
5079 );
5080 assert!(
5081 actions.is_empty(),
5082 "invalid ciphertext for context {context:#x} should be ignored"
5083 );
5084 }
5085 }
5086
5087 #[test]
5088 fn test_resource_part_without_matching_receiver_is_ignored() {
5089 let (_init_mgr, mut resp_mgr, link_id) = setup_active_link();
5090 let mut rng = OsRng;
5091 let flags = PacketFlags {
5092 header_type: constants::HEADER_1,
5093 context_flag: constants::FLAG_UNSET,
5094 transport_type: constants::TRANSPORT_BROADCAST,
5095 destination_type: constants::DESTINATION_LINK,
5096 packet_type: constants::PACKET_TYPE_DATA,
5097 };
5098 let pkt = RawPacket::pack(
5099 flags,
5100 0,
5101 &link_id,
5102 None,
5103 constants::CONTEXT_RESOURCE,
5104 b"orphan-part",
5105 )
5106 .expect("test packet should pack");
5107
5108 let actions = resp_mgr.handle_local_delivery(
5109 pkt.destination_hash,
5110 &pkt.raw,
5111 pkt.packet_hash,
5112 rns_core::transport::types::InterfaceId(0),
5113 &mut rng,
5114 );
5115
5116 assert!(actions.is_empty(), "orphan resource part should be ignored");
5117 }
5118
5119 #[test]
5120 fn test_response_delivery() {
5121 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5122 let mut rng = OsRng;
5123
5124 resp_mgr.register_request_handler("/echo", None, |_link_id, _path, data, _remote| {
5126 Some(data.to_vec())
5127 });
5128
5129 let req_actions = init_mgr.send_request(&link_id, "/echo", b"\xc0", &mut rng); assert!(!req_actions.is_empty());
5132
5133 let req_raw = extract_any_send_packet(&req_actions);
5135 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
5136 let resp_actions = resp_mgr.handle_local_delivery(
5137 req_pkt.destination_hash,
5138 &req_raw,
5139 req_pkt.packet_hash,
5140 rns_core::transport::types::InterfaceId(0),
5141 &mut rng,
5142 );
5143 let has_resp_send = resp_actions
5144 .iter()
5145 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
5146 assert!(has_resp_send, "Handler should produce response");
5147
5148 let resp_raw = extract_any_send_packet(&resp_actions);
5150 let resp_pkt = RawPacket::unpack(&resp_raw).unwrap();
5151 let init_actions = init_mgr.handle_local_delivery(
5152 resp_pkt.destination_hash,
5153 &resp_raw,
5154 resp_pkt.packet_hash,
5155 rns_core::transport::types::InterfaceId(0),
5156 &mut rng,
5157 );
5158
5159 let has_response_received = init_actions
5160 .iter()
5161 .any(|a| matches!(a, LinkManagerAction::ResponseReceived { .. }));
5162 assert!(
5163 has_response_received,
5164 "Initiator should receive ResponseReceived"
5165 );
5166 }
5167
5168 #[test]
5169 fn test_large_response_uses_resource_fallback() {
5170 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5171 let mut rng = OsRng;
5172
5173 let large_payload: Vec<u8> = (0..5000u32).map(|i| (i & 0xFF) as u8).collect();
5176 resp_mgr.register_request_handler("/large", None, {
5177 let large_payload = large_payload.clone();
5178 move |_link_id, _path, _data, _remote| Some(large_payload.clone())
5179 });
5180
5181 let req_actions = init_mgr.send_request(&link_id, "/large", b"\xc0", &mut rng);
5183 assert!(!req_actions.is_empty());
5184
5185 let req_raw = extract_any_send_packet(&req_actions);
5187 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
5188 let resp_actions = resp_mgr.handle_local_delivery(
5189 req_pkt.destination_hash,
5190 &req_raw,
5191 req_pkt.packet_hash,
5192 rns_core::transport::types::InterfaceId(0),
5193 &mut rng,
5194 );
5195
5196 let mut has_resource_adv = false;
5197 let mut has_direct_response = false;
5198 for action in &resp_actions {
5199 if let LinkManagerAction::SendPacket { raw, .. } = action {
5200 let pkt = RawPacket::unpack(raw).unwrap();
5201 if pkt.context == constants::CONTEXT_RESOURCE_ADV {
5202 has_resource_adv = true;
5203 }
5204 if pkt.context == constants::CONTEXT_RESPONSE {
5205 has_direct_response = true;
5206 }
5207 }
5208 }
5209
5210 assert!(
5211 has_resource_adv,
5212 "Large response should advertise a response resource"
5213 );
5214 assert!(
5215 !has_direct_response,
5216 "Large response should not use direct CONTEXT_RESPONSE packet"
5217 );
5218 }
5219
5220 #[test]
5221 fn test_send_management_response_without_session_key_uses_resource_fallback_path() {
5222 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5223 let mut rng = OsRng;
5224 init_mgr
5225 .links
5226 .get_mut(&link_id)
5227 .unwrap()
5228 .engine
5229 .clear_session_for_testing();
5230
5231 let large_response: Vec<u8> = (0..5000u32).map(|i| (i & 0xFF) as u8).collect();
5232 let actions =
5233 init_mgr.send_management_response(&link_id, &[0x11; 16], &large_response, &mut rng);
5234
5235 assert!(
5236 actions.is_empty(),
5237 "without a session key, no response packets should be emitted"
5238 );
5239 assert_eq!(
5240 init_mgr
5241 .links
5242 .get(&link_id)
5243 .map(|managed| managed.outgoing_resources.len()),
5244 Some(1)
5245 );
5246 }
5247
5248 #[test]
5249 fn test_send_channel_message_on_no_channel() {
5250 let mut mgr = LinkManager::new();
5251 let mut rng = OsRng;
5252 let dummy_sig = [0xAA; 32];
5253 let (link_id, _) =
5254 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
5255
5256 let err = mgr
5258 .send_channel_message(&link_id, 1, b"test", &mut rng)
5259 .expect_err("pending link should reject channel send");
5260 assert_eq!(err, "link has no active channel");
5261 }
5262
5263 #[test]
5264 fn test_send_on_link_requires_active() {
5265 let mut mgr = LinkManager::new();
5266 let mut rng = OsRng;
5267 let dummy_sig = [0xAA; 32];
5268 let (link_id, _) =
5269 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
5270
5271 let actions = mgr.send_on_link(&link_id, b"test", constants::CONTEXT_NONE, &mut rng);
5272 assert!(actions.is_empty(), "Cannot send on pending link");
5273 }
5274
5275 #[test]
5276 fn test_send_on_link_unknown_link() {
5277 let mgr = LinkManager::new();
5278 let mut rng = OsRng;
5279
5280 let actions = mgr.send_on_link(&[0xFF; 16], b"test", constants::CONTEXT_NONE, &mut rng);
5281 assert!(actions.is_empty());
5282 }
5283
5284 #[test]
5285 fn test_resource_full_transfer_large() {
5286 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5287 let mut rng = OsRng;
5288
5289 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
5290
5291 let original_data: Vec<u8> = (0..2000u32)
5293 .map(|i| {
5294 let pos = i as usize;
5295 (pos ^ (pos >> 8) ^ (pos >> 16)) as u8
5296 })
5297 .collect();
5298
5299 let adv_actions = init_mgr.send_resource(&link_id, &original_data, None, &mut rng);
5300
5301 let mut pending: Vec<(char, LinkManagerAction)> =
5302 adv_actions.into_iter().map(|a| ('i', a)).collect();
5303 let mut rounds = 0;
5304 let max_rounds = 200;
5305 let mut resource_received = false;
5306 let mut sender_completed = false;
5307
5308 while !pending.is_empty() && rounds < max_rounds {
5309 rounds += 1;
5310 let mut next: Vec<(char, LinkManagerAction)> = Vec::new();
5311
5312 for (source, action) in pending.drain(..) {
5313 if let LinkManagerAction::SendPacket { raw, .. } = action {
5314 let pkt = match RawPacket::unpack(&raw) {
5315 Ok(p) => p,
5316 Err(_) => continue,
5317 };
5318
5319 let target_actions = if source == 'i' {
5320 resp_mgr.handle_local_delivery(
5321 pkt.destination_hash,
5322 &raw,
5323 pkt.packet_hash,
5324 rns_core::transport::types::InterfaceId(0),
5325 &mut rng,
5326 )
5327 } else {
5328 init_mgr.handle_local_delivery(
5329 pkt.destination_hash,
5330 &raw,
5331 pkt.packet_hash,
5332 rns_core::transport::types::InterfaceId(0),
5333 &mut rng,
5334 )
5335 };
5336
5337 let target_source = if source == 'i' { 'r' } else { 'i' };
5338 for a in &target_actions {
5339 match a {
5340 LinkManagerAction::ResourceReceived { data, .. } => {
5341 assert_eq!(*data, original_data);
5342 resource_received = true;
5343 }
5344 LinkManagerAction::ResourceCompleted { .. } => {
5345 sender_completed = true;
5346 }
5347 _ => {}
5348 }
5349 }
5350 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
5351 }
5352 }
5353 pending = next;
5354 }
5355
5356 assert!(
5357 resource_received,
5358 "Should receive large resource (rounds={})",
5359 rounds
5360 );
5361 assert!(
5362 sender_completed,
5363 "Sender should complete (rounds={})",
5364 rounds
5365 );
5366 }
5367
5368 #[test]
5369 fn test_resource_receiver_stores_original_advertisement_plaintext() {
5370 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5371 let mut rng = OsRng;
5372
5373 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
5374
5375 let data = vec![0x41; 256];
5376 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5377
5378 let adv_raw = adv_actions
5379 .iter()
5380 .find_map(|action| match action {
5381 LinkManagerAction::SendPacket { raw, .. } => {
5382 let pkt = RawPacket::unpack(raw).ok()?;
5383 (pkt.context == constants::CONTEXT_RESOURCE_ADV).then_some(raw.clone())
5384 }
5385 _ => None,
5386 })
5387 .expect("sender should emit a resource advertisement");
5388
5389 let adv_pkt = RawPacket::unpack(&adv_raw).unwrap();
5390 let adv_plaintext = resp_mgr
5391 .links
5392 .get(&link_id)
5393 .unwrap()
5394 .engine
5395 .decrypt(&adv_pkt.data)
5396 .unwrap();
5397
5398 let _resp_actions = resp_mgr.handle_local_delivery(
5399 adv_pkt.destination_hash,
5400 &adv_raw,
5401 adv_pkt.packet_hash,
5402 rns_core::transport::types::InterfaceId(0),
5403 &mut rng,
5404 );
5405
5406 let receiver = resp_mgr
5407 .links
5408 .get(&link_id)
5409 .and_then(|managed| managed.incoming_resources.first())
5410 .expect("advertisement should create an incoming receiver");
5411 assert_eq!(receiver.advertisement_packet, adv_plaintext);
5412 assert_eq!(
5413 receiver.max_decompressed_size,
5414 constants::RESOURCE_AUTO_COMPRESS_MAX_SIZE
5415 );
5416 }
5417
5418 #[test]
5419 fn test_corrupt_compressed_resource_rejects_and_tears_down_link() {
5420 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5421 let mut rng = OsRng;
5422
5423 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
5424
5425 let data = vec![b'A'; 4096];
5426 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5427
5428 let mut request_actions = Vec::new();
5429 for action in &adv_actions {
5430 let LinkManagerAction::SendPacket { raw, .. } = action else {
5431 continue;
5432 };
5433 let pkt = RawPacket::unpack(raw).unwrap();
5434 let actions = resp_mgr.handle_local_delivery(
5435 pkt.destination_hash,
5436 raw,
5437 pkt.packet_hash,
5438 rns_core::transport::types::InterfaceId(0),
5439 &mut rng,
5440 );
5441 request_actions.extend(actions);
5442 }
5443
5444 {
5445 let receiver = resp_mgr
5446 .links
5447 .get_mut(&link_id)
5448 .and_then(|managed| managed.incoming_resources.first_mut())
5449 .expect("receiver should exist after advertisement");
5450 assert!(receiver.flags.compressed, "test data should be compressed");
5451 receiver.max_decompressed_size = 64;
5452 }
5453
5454 let mut responder_actions = Vec::new();
5455 for action in request_actions {
5456 let LinkManagerAction::SendPacket { raw, .. } = action else {
5457 continue;
5458 };
5459 let pkt = RawPacket::unpack(&raw).unwrap();
5460 let actions = init_mgr.handle_local_delivery(
5461 pkt.destination_hash,
5462 &raw,
5463 pkt.packet_hash,
5464 rns_core::transport::types::InterfaceId(0),
5465 &mut rng,
5466 );
5467
5468 for action in actions {
5469 let LinkManagerAction::SendPacket { raw, .. } = &action else {
5470 continue;
5471 };
5472 let pkt = RawPacket::unpack(raw).unwrap();
5473 let delivered = resp_mgr.handle_local_delivery(
5474 pkt.destination_hash,
5475 raw,
5476 pkt.packet_hash,
5477 rns_core::transport::types::InterfaceId(0),
5478 &mut rng,
5479 );
5480 responder_actions.extend(delivered);
5481 }
5482 }
5483
5484 assert!(
5485 responder_actions.iter().any(|action| matches!(
5486 action,
5487 LinkManagerAction::ResourceFailed { error, .. }
5488 if error == "Resource too large"
5489 )),
5490 "corrupt oversized resource should fail with TooLarge"
5491 );
5492 assert!(
5493 responder_actions.iter().any(|action| matches!(
5494 action,
5495 LinkManagerAction::LinkClosed { link_id: closed_id, .. } if *closed_id == link_id
5496 )),
5497 "corrupt oversized resource should tear down the link"
5498 );
5499 assert!(
5500 responder_actions.iter().any(|action| match action {
5501 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
5502 .map(|pkt| pkt.context == constants::CONTEXT_RESOURCE_RCL)
5503 .unwrap_or(false),
5504 _ => false,
5505 }),
5506 "corrupt oversized resource should send a receiver cancel/reject packet"
5507 );
5508 assert_eq!(
5509 resp_mgr
5510 .links
5511 .get(&link_id)
5512 .map(|managed| managed.engine.state()),
5513 Some(LinkState::Closed)
5514 );
5515 }
5516
5517 #[test]
5518 fn test_resource_hmu_timeout_extension_in_link_manager_flow() {
5519 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5520 let mut rng = OsRng;
5521
5522 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
5523
5524 let mut state = 0x1234_5678u32;
5527 let data: Vec<u8> = (0..50000)
5528 .map(|_| {
5529 state = state.wrapping_mul(1664525).wrapping_add(1013904223);
5530 (state >> 16) as u8
5531 })
5532 .collect();
5533 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5534 let mut pending: Vec<(char, LinkManagerAction)> =
5535 adv_actions.into_iter().map(|a| ('i', a)).collect();
5536
5537 let mut rounds = 0;
5538
5539 while rounds < 300 {
5542 rounds += 1;
5543 let mut next: Vec<(char, LinkManagerAction)> = Vec::new();
5544
5545 for (source, action) in pending.drain(..) {
5546 let LinkManagerAction::SendPacket { raw, .. } = action else {
5547 continue;
5548 };
5549
5550 let pkt = match RawPacket::unpack(&raw) {
5551 Ok(p) => p,
5552 Err(_) => continue,
5553 };
5554
5555 let target_actions = if source == 'i' {
5556 resp_mgr.handle_local_delivery(
5557 pkt.destination_hash,
5558 &raw,
5559 pkt.packet_hash,
5560 rns_core::transport::types::InterfaceId(0),
5561 &mut rng,
5562 )
5563 } else {
5564 init_mgr.handle_local_delivery(
5565 pkt.destination_hash,
5566 &raw,
5567 pkt.packet_hash,
5568 rns_core::transport::types::InterfaceId(0),
5569 &mut rng,
5570 )
5571 };
5572
5573 let target_source = if source == 'i' { 'r' } else { 'i' };
5574 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
5575 }
5576
5577 if resp_mgr
5578 .links
5579 .get(&link_id)
5580 .and_then(|managed| managed.incoming_resources.first())
5581 .is_some_and(|receiver| receiver.waiting_for_hmu)
5582 {
5583 break;
5584 }
5585
5586 pending = next;
5587 }
5588
5589 assert!(
5590 resp_mgr
5591 .links
5592 .get(&link_id)
5593 .and_then(|managed| managed.incoming_resources.first())
5594 .is_some_and(|receiver| receiver.waiting_for_hmu),
5595 "expected receiver to reach a live HMU wait state"
5596 );
5597
5598 let prime_actions = {
5601 let managed = resp_mgr.links.get_mut(&link_id).unwrap();
5602 let receiver = managed.incoming_resources.first_mut().unwrap();
5603 let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
5604 managed.engine.decrypt(ciphertext).map_err(|_| ())
5605 };
5606 receiver.tick(
5607 receiver.last_activity + 0.0001,
5608 &decrypt_fn,
5609 &Bzip2Compressor,
5610 )
5611 };
5612 assert!(
5613 !prime_actions
5614 .iter()
5615 .any(|a| matches!(a, ResourceAction::SendRequest(_))),
5616 "fresh HMU wait state should not immediately emit a retry request"
5617 );
5618
5619 let (late_delta, retries_before) = {
5620 let managed = resp_mgr
5621 .links
5622 .get_mut(&link_id)
5623 .expect("receiver link should still exist");
5624 let receiver = managed
5625 .incoming_resources
5626 .first_mut()
5627 .expect("receiver should have an active incoming resource");
5628
5629 assert!(
5630 receiver.waiting_for_hmu,
5631 "receiver should be waiting for HMU"
5632 );
5633
5634 let eifr = receiver.eifr.unwrap_or_else(|| {
5635 (constants::RESOURCE_SDU as f64 * 8.0) / receiver.rtt.unwrap_or(0.5)
5636 });
5637 let expected_tof = if receiver.outstanding_parts > 0 {
5638 (receiver.outstanding_parts as f64 * constants::RESOURCE_SDU as f64 * 8.0) / eifr
5639 } else {
5640 (3.0 * constants::RESOURCE_SDU as f64) / eifr
5641 };
5642 let expected_hmu_wait =
5643 (constants::RESOURCE_SDU as f64 * 8.0 * constants::RESOURCE_HMU_WAIT_FACTOR) / eifr;
5644 let old_delta = constants::RESOURCE_PART_TIMEOUT_FACTOR_AFTER_RTT * expected_tof
5645 + constants::RESOURCE_RETRY_GRACE_TIME;
5646 (
5647 old_delta + expected_hmu_wait + expected_hmu_wait.max(1.0),
5648 receiver.retries_left,
5649 )
5650 };
5651 {
5652 let managed = resp_mgr.links.get(&link_id).unwrap();
5653 let receiver = managed.incoming_resources.first().unwrap();
5654 assert_eq!(receiver.retries_left, retries_before);
5655 assert!(
5656 receiver.eifr.is_some(),
5657 "receiver tick should have populated EIFR"
5658 );
5659 }
5660
5661 let late_resource_actions = {
5662 let managed = resp_mgr.links.get_mut(&link_id).unwrap();
5663 let receiver = managed.incoming_resources.first_mut().unwrap();
5664 let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
5665 managed.engine.decrypt(ciphertext).map_err(|_| ())
5666 };
5667 receiver.tick(
5668 receiver.last_activity + late_delta,
5669 &decrypt_fn,
5670 &Bzip2Compressor,
5671 )
5672 };
5673 let late_actions =
5674 resp_mgr.process_resource_actions(&link_id, late_resource_actions, &mut rng);
5675 let retry_raw = late_actions
5676 .iter()
5677 .find_map(|a| match a {
5678 LinkManagerAction::SendPacket { raw, .. } => {
5679 let pkt = RawPacket::unpack(raw).ok()?;
5680 (pkt.context == constants::CONTEXT_RESOURCE_REQ).then_some(raw.clone())
5681 }
5682 _ => None,
5683 })
5684 .expect("receiver should emit a resource retry request after extended timeout");
5685
5686 {
5687 let managed = resp_mgr.links.get(&link_id).unwrap();
5688 let receiver = managed.incoming_resources.first().unwrap();
5689 assert_eq!(receiver.retries_left, retries_before - 1);
5690 }
5691
5692 let retry_pkt = RawPacket::unpack(&retry_raw).unwrap();
5693 let retry_plaintext = resp_mgr
5694 .links
5695 .get(&link_id)
5696 .unwrap()
5697 .engine
5698 .decrypt(&retry_pkt.data)
5699 .expect("retry request should decrypt");
5700 assert_eq!(retry_plaintext[0], constants::RESOURCE_HASHMAP_IS_EXHAUSTED);
5701
5702 let retry_to_sender = init_mgr.handle_local_delivery(
5705 retry_pkt.destination_hash,
5706 &retry_raw,
5707 retry_pkt.packet_hash,
5708 rns_core::transport::types::InterfaceId(0),
5709 &mut rng,
5710 );
5711 assert!(
5712 retry_to_sender.iter().any(|a| match a {
5713 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
5714 .map(|pkt| pkt.context == constants::CONTEXT_RESOURCE_HMU)
5715 .unwrap_or(false),
5716 _ => false,
5717 }),
5718 "sender should answer the exhausted retry request with a live HMU packet"
5719 );
5720 }
5721
5722 #[test]
5723 fn test_process_resource_actions_mapping() {
5724 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5725 let mut rng = OsRng;
5726
5727 let actions = vec![
5729 ResourceAction::DataReceived {
5730 data: vec![1, 2, 3],
5731 metadata: Some(vec![4, 5]),
5732 },
5733 ResourceAction::Completed,
5734 ResourceAction::Failed(rns_core::resource::ResourceError::Timeout),
5735 ResourceAction::ProgressUpdate {
5736 received: 10,
5737 total: 20,
5738 },
5739 ResourceAction::TeardownLink,
5740 ];
5741
5742 let result = init_mgr.process_resource_actions(&link_id, actions, &mut rng);
5743
5744 assert!(matches!(
5745 result[0],
5746 LinkManagerAction::ResourceReceived { .. }
5747 ));
5748 assert!(matches!(
5749 result[1],
5750 LinkManagerAction::ResourceCompleted { .. }
5751 ));
5752 assert!(matches!(
5753 result[2],
5754 LinkManagerAction::ResourceFailed { .. }
5755 ));
5756 assert!(matches!(
5757 result[3],
5758 LinkManagerAction::ResourceProgress {
5759 received: 10,
5760 total: 20,
5761 ..
5762 }
5763 ));
5764 assert!(result
5765 .iter()
5766 .any(|action| matches!(action, LinkManagerAction::LinkClosed { .. })));
5767 }
5768
5769 #[test]
5770 fn test_link_state_empty() {
5771 let mgr = LinkManager::new();
5772 let fake_id = [0xAA; 16];
5773 assert!(mgr.link_state(&fake_id).is_none());
5774 }
5775
5776 #[test]
5777 fn test_large_response_resource_completes_as_response() {
5778 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5779 let mut rng = OsRng;
5780
5781 let large_payload: Vec<u8> = (0..5000u32).map(|i| (i & 0xFF) as u8).collect();
5782 let response_value = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(large_payload));
5783 resp_mgr.register_request_handler("/large", None, {
5784 let response_value = response_value.clone();
5785 move |_link_id, _path, _data, _remote| Some(response_value.clone())
5786 });
5787
5788 let req_actions = init_mgr.send_request(&link_id, "/large", b"\xc0", &mut rng);
5789 let req_raw = extract_any_send_packet(&req_actions);
5790 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
5791 let request_id = req_pkt.get_truncated_hash();
5792 let resp_actions = resp_mgr.handle_local_delivery(
5793 req_pkt.destination_hash,
5794 &req_raw,
5795 req_pkt.packet_hash,
5796 rns_core::transport::types::InterfaceId(0),
5797 &mut rng,
5798 );
5799
5800 let mut pending: Vec<(char, LinkManagerAction)> =
5801 resp_actions.into_iter().map(|a| ('r', a)).collect();
5802 let mut rounds = 0;
5803 let mut received_response = None;
5804
5805 while !pending.is_empty() && rounds < 200 {
5806 rounds += 1;
5807 let mut next = Vec::new();
5808
5809 for (source, action) in pending.drain(..) {
5810 let LinkManagerAction::SendPacket { raw, .. } = action else {
5811 continue;
5812 };
5813 let pkt = RawPacket::unpack(&raw).unwrap();
5814 let target_actions = if source == 'r' {
5815 init_mgr.handle_local_delivery(
5816 pkt.destination_hash,
5817 &raw,
5818 pkt.packet_hash,
5819 rns_core::transport::types::InterfaceId(0),
5820 &mut rng,
5821 )
5822 } else {
5823 resp_mgr.handle_local_delivery(
5824 pkt.destination_hash,
5825 &raw,
5826 pkt.packet_hash,
5827 rns_core::transport::types::InterfaceId(0),
5828 &mut rng,
5829 )
5830 };
5831
5832 let target_source = if source == 'r' { 'i' } else { 'r' };
5833 for target_action in &target_actions {
5834 match target_action {
5835 LinkManagerAction::ResponseReceived {
5836 request_id: rid,
5837 data,
5838 ..
5839 } => {
5840 received_response = Some((*rid, data.clone()));
5841 }
5842 LinkManagerAction::ResourceReceived { .. } => {
5843 panic!("response resources must complete as ResponseReceived")
5844 }
5845 LinkManagerAction::ResourceAcceptQuery { .. } => {
5846 panic!("response resources must bypass application acceptance")
5847 }
5848 _ => {}
5849 }
5850 }
5851 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
5852 }
5853
5854 pending = next;
5855 }
5856
5857 let (received_request_id, received_data) = received_response.unwrap_or_else(|| {
5858 panic!(
5859 "large response resource did not complete as ResponseReceived after {} rounds",
5860 rounds
5861 )
5862 });
5863 assert_eq!(received_request_id, request_id);
5864 assert_eq!(received_data, response_value);
5865 }
5866
5867 #[test]
5868 fn test_response_resource_preserves_metadata() {
5869 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5870 let mut rng = OsRng;
5871
5872 let payload = b"bundle-data".to_vec();
5873 let metadata = b"git-status-ok".to_vec();
5874 let response_value = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload));
5875 resp_mgr.register_request_handler_response("/fetch", None, {
5876 let response_value = response_value.clone();
5877 let metadata = metadata.clone();
5878 move |_link_id, _path, _data, _remote| {
5879 Some(RequestResponse::Resource {
5880 data: response_value.clone(),
5881 metadata: Some(metadata.clone()),
5882 auto_compress: false,
5883 })
5884 }
5885 });
5886
5887 let req_actions = init_mgr.send_request(&link_id, "/fetch", b"\xc0", &mut rng);
5888 let req_raw = extract_any_send_packet(&req_actions);
5889 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
5890 let request_id = req_pkt.get_truncated_hash();
5891 let resp_actions = resp_mgr.handle_local_delivery(
5892 req_pkt.destination_hash,
5893 &req_raw,
5894 req_pkt.packet_hash,
5895 rns_core::transport::types::InterfaceId(0),
5896 &mut rng,
5897 );
5898
5899 let mut pending: Vec<(char, LinkManagerAction)> =
5900 resp_actions.into_iter().map(|a| ('r', a)).collect();
5901 let mut received_response = None;
5902
5903 for _ in 0..200 {
5904 if pending.is_empty() || received_response.is_some() {
5905 break;
5906 }
5907
5908 let mut next = Vec::new();
5909 for (source, action) in pending.drain(..) {
5910 let LinkManagerAction::SendPacket { raw, .. } = action else {
5911 continue;
5912 };
5913 let pkt = RawPacket::unpack(&raw).unwrap();
5914 let target_actions = if source == 'r' {
5915 init_mgr.handle_local_delivery(
5916 pkt.destination_hash,
5917 &raw,
5918 pkt.packet_hash,
5919 rns_core::transport::types::InterfaceId(0),
5920 &mut rng,
5921 )
5922 } else {
5923 resp_mgr.handle_local_delivery(
5924 pkt.destination_hash,
5925 &raw,
5926 pkt.packet_hash,
5927 rns_core::transport::types::InterfaceId(0),
5928 &mut rng,
5929 )
5930 };
5931
5932 let target_source = if source == 'r' { 'i' } else { 'r' };
5933 for target_action in &target_actions {
5934 match target_action {
5935 LinkManagerAction::ResponseReceived {
5936 request_id: rid,
5937 data,
5938 metadata: response_metadata,
5939 ..
5940 } => {
5941 received_response =
5942 Some((*rid, data.clone(), response_metadata.clone()));
5943 }
5944 LinkManagerAction::ResourceReceived { .. } => {
5945 panic!("response resources must complete as ResponseReceived")
5946 }
5947 _ => {}
5948 }
5949 }
5950 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
5951 }
5952 pending = next;
5953 }
5954
5955 let (received_request_id, received_data, received_metadata) = received_response
5956 .expect("resource response with metadata should complete as ResponseReceived");
5957 assert_eq!(received_request_id, request_id);
5958 assert_eq!(received_data, response_value);
5959 assert_eq!(received_metadata, Some(metadata));
5960 }
5961
5962 #[test]
5963 fn test_negotiated_mtu_response_uses_resource_before_global_mtu() {
5964 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5965 let mut rng = OsRng;
5966
5967 init_mgr.set_link_mtu(&link_id, 300);
5968 resp_mgr.set_link_mtu(&link_id, 300);
5969
5970 let payload = vec![0xAB; 350];
5971 let response_value = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload));
5972 resp_mgr.register_request_handler("/mtu", None, {
5973 let response_value = response_value.clone();
5974 move |_link_id, _path, _data, _remote| Some(response_value.clone())
5975 });
5976
5977 let req_actions = init_mgr.send_request(&link_id, "/mtu", b"\xc0", &mut rng);
5978 let req_raw = extract_any_send_packet(&req_actions);
5979 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
5980 let resp_actions = resp_mgr.handle_local_delivery(
5981 req_pkt.destination_hash,
5982 &req_raw,
5983 req_pkt.packet_hash,
5984 rns_core::transport::types::InterfaceId(0),
5985 &mut rng,
5986 );
5987
5988 let mut has_resource_adv = false;
5989 let mut direct_response_len = None;
5990 for action in &resp_actions {
5991 if let LinkManagerAction::SendPacket { raw, .. } = action {
5992 let pkt = RawPacket::unpack(raw).unwrap();
5993 has_resource_adv |= pkt.context == constants::CONTEXT_RESOURCE_ADV;
5994 if pkt.context == constants::CONTEXT_RESPONSE {
5995 direct_response_len = Some(raw.len());
5996 }
5997 }
5998 }
5999
6000 assert!(
6001 has_resource_adv,
6002 "responses larger than the negotiated link MTU should use resource fallback"
6003 );
6004 assert!(
6005 direct_response_len.is_none(),
6006 "sent direct response of {} bytes on a 300 byte negotiated MTU",
6007 direct_response_len.unwrap_or_default()
6008 );
6009 }
6010
6011 #[test]
6012 fn test_large_management_response_uses_resource_fallback() {
6013 let (_init_mgr, mut resp_mgr, link_id) = setup_active_link();
6014 let mut rng = OsRng;
6015
6016 let payload = vec![0xBC; 5000];
6017 let response_value = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload));
6018 let actions =
6019 resp_mgr.send_management_response(&link_id, &[0x55; 16], &response_value, &mut rng);
6020
6021 let mut has_resource_adv = false;
6022 let mut has_direct_response = false;
6023 for action in &actions {
6024 if let LinkManagerAction::SendPacket { raw, .. } = action {
6025 let pkt = RawPacket::unpack(raw).unwrap();
6026 has_resource_adv |= pkt.context == constants::CONTEXT_RESOURCE_ADV;
6027 has_direct_response |= pkt.context == constants::CONTEXT_RESPONSE;
6028 }
6029 }
6030
6031 assert!(
6032 has_resource_adv,
6033 "large management responses should advertise a response resource"
6034 );
6035 assert!(
6036 !has_direct_response,
6037 "large management responses should not use a direct CONTEXT_RESPONSE packet"
6038 );
6039 }
6040}