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::{OsRng, 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 pending_requests: HashMap<[u8; 16], PendingRequest>,
64 incoming_splits: HashMap<[u8; 32], IncomingSplitTransfer>,
66 outgoing_splits: HashMap<[u8; 32], OutgoingSplitTransfer>,
68 resource_strategy: ResourceStrategy,
70 max_request_size: Option<usize>,
72 route_interface: Option<rns_core::transport::types::InterfaceId>,
74 route_transport_id: Option<[u8; 16]>,
79}
80
81struct IncomingSplitTransfer {
82 total_segments: u64,
83 completed_segments: u64,
84 current_segment_index: u64,
85 current_received_parts: usize,
86 current_total_parts: usize,
87 data: Vec<u8>,
88 metadata: Option<Vec<u8>>,
89 is_request: bool,
90 is_response: bool,
91}
92
93struct OutgoingSplitTransfer {
94 total_segments: u64,
95 completed_segments: u64,
96 current_segment_index: u64,
97 current_sent_parts: usize,
98 current_total_parts: usize,
99}
100
101struct PendingRequest {
102 deadline: Option<f64>,
103 max_response_size: Option<usize>,
104}
105
106struct LinkDestination {
108 sig_prv: Ed25519PrivateKey,
109 sig_pub_bytes: [u8; 32],
110 resource_strategy: ResourceStrategy,
111 max_request_size: Option<usize>,
112}
113
114#[derive(Debug, Clone, PartialEq, Eq)]
116pub enum RequestResponse {
117 Bytes(Vec<u8>),
119 Resource {
121 data: Vec<u8>,
122 metadata: Option<Vec<u8>>,
123 auto_compress: bool,
124 },
125}
126
127#[derive(Debug, Clone, Copy, PartialEq, Eq)]
129pub enum RequestFailure {
130 ResponseTooLarge { size: u64, maximum: usize },
132}
133
134impl From<Vec<u8>> for RequestResponse {
135 fn from(data: Vec<u8>) -> Self {
136 RequestResponse::Bytes(data)
137 }
138}
139
140struct RequestHandlerEntry {
142 path: String,
144 path_hash: [u8; 16],
146 allowed_list: Option<Vec<[u8; 16]>>,
148 handler: Box<
150 dyn Fn(LinkId, &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<RequestResponse>
151 + Send,
152 >,
153}
154
155#[derive(Debug)]
157pub enum LinkManagerAction {
158 SendPacket {
160 raw: Vec<u8>,
161 dest_type: u8,
162 attached_interface: Option<rns_core::transport::types::InterfaceId>,
163 },
164 LinkEstablished {
166 link_id: LinkId,
167 dest_hash: [u8; 16],
168 hops: u8,
169 rebalanced_at: Option<f64>,
170 rtt: f64,
171 is_initiator: bool,
172 },
173 LinkClosed {
175 link_id: LinkId,
176 reason: Option<TeardownReason>,
177 },
178 RemoteIdentified {
180 link_id: LinkId,
181 identity_hash: [u8; 16],
182 public_key: [u8; 64],
183 },
184 RegisterLinkDest { link_id: LinkId },
186 DeregisterLinkDest { link_id: LinkId },
188 ManagementRequest {
191 link_id: LinkId,
192 path_hash: [u8; 16],
193 data: Vec<u8>,
195 request_id: [u8; 16],
197 remote_identity: Option<([u8; 16], [u8; 64])>,
198 },
199 ResourceReceived {
201 link_id: LinkId,
202 data: Vec<u8>,
203 metadata: Option<Vec<u8>>,
204 },
205 ResourceCompleted { link_id: LinkId },
207 ResourceFailed { link_id: LinkId, error: String },
209 ResourceProgress {
211 link_id: LinkId,
212 received: usize,
213 total: usize,
214 },
215 ResourceAcceptQuery {
217 link_id: LinkId,
218 resource_hash: Vec<u8>,
219 transfer_size: u64,
220 has_metadata: bool,
221 },
222 ChannelMessageReceived {
224 link_id: LinkId,
225 msgtype: u16,
226 payload: Vec<u8>,
227 },
228 LinkDataReceived {
230 link_id: LinkId,
231 context: u8,
232 data: Vec<u8>,
233 },
234 ResponseReceived {
236 link_id: LinkId,
237 request_id: [u8; 16],
238 data: Vec<u8>,
239 metadata: Option<Vec<u8>>,
240 },
241 RequestFailed {
243 link_id: LinkId,
244 request_id: [u8; 16],
245 reason: RequestFailure,
246 },
247 LinkRequestReceived {
249 link_id: LinkId,
250 receiving_interface: rns_core::transport::types::InterfaceId,
251 },
252}
253
254pub struct LinkManager {
256 links: HashMap<LinkId, ManagedLink>,
257 link_destinations: HashMap<[u8; 16], LinkDestination>,
258 request_handlers: Vec<RequestHandlerEntry>,
259 management_paths: Vec<[u8; 16]>,
262}
263
264#[derive(Debug, Clone, Copy, PartialEq, Eq)]
265pub struct LinkRouteHint {
266 pub interface: rns_core::transport::types::InterfaceId,
267 pub transport_id: Option<[u8; 16]>,
268}
269
270impl LinkManager {
271 fn resource_sdu_for_link(link: &ManagedLink) -> usize {
272 let mtu = link.engine.mtu() as usize;
275 let derived = mtu.saturating_sub(constants::HEADER_MAXSIZE + constants::IFAC_MIN_SIZE);
276 if derived > 0 {
277 derived
278 } else {
279 constants::RESOURCE_SDU
280 }
281 }
282
283 fn split_progress_parts(
284 segment_index: u64,
285 total_segments: u64,
286 current_done: usize,
287 current_total: usize,
288 sdu: usize,
289 ) -> (usize, usize) {
290 let max_parts_per_segment = constants::RESOURCE_MAX_EFFICIENT_SIZE.div_ceil(sdu.max(1));
291 let total = (total_segments as usize).saturating_mul(max_parts_per_segment);
292 let completed_segments = segment_index.saturating_sub(1) as usize;
293 let completed = completed_segments.saturating_mul(max_parts_per_segment);
294 let current = if current_total == 0 {
295 0
296 } else if current_total < max_parts_per_segment {
297 let scaled =
298 (current_done as f64) * (max_parts_per_segment as f64 / current_total as f64);
299 scaled.floor() as usize
300 } else {
301 current_done
302 };
303 (completed.saturating_add(current).min(total), total)
304 }
305
306 fn resource_hash_key(hash: &[u8]) -> Option<[u8; 32]> {
307 let mut key = [0u8; 32];
308 if hash.len() != key.len() {
309 return None;
310 }
311 key.copy_from_slice(hash);
312 Some(key)
313 }
314
315 fn incoming_split_progress(split: &IncomingSplitTransfer, sdu: usize) -> (usize, usize) {
316 Self::split_progress_parts(
317 split.current_segment_index,
318 split.total_segments,
319 split.current_received_parts,
320 split.current_total_parts,
321 sdu,
322 )
323 }
324
325 fn outgoing_split_progress(split: &OutgoingSplitTransfer, sdu: usize) -> (usize, usize) {
326 Self::split_progress_parts(
327 split.current_segment_index,
328 split.total_segments,
329 split.current_sent_parts,
330 split.current_total_parts,
331 sdu,
332 )
333 }
334
335 pub fn new() -> Self {
337 LinkManager {
338 links: HashMap::new(),
339 link_destinations: HashMap::new(),
340 request_handlers: Vec::new(),
341 management_paths: Vec::new(),
342 }
343 }
344
345 pub fn register_management_path(&mut self, path_hash: [u8; 16]) {
349 if !self.management_paths.contains(&path_hash) {
350 self.management_paths.push(path_hash);
351 }
352 }
353
354 pub fn get_derived_key(&self, link_id: &LinkId) -> Option<Vec<u8>> {
356 self.links
357 .get(link_id)
358 .and_then(|link| link.engine.derived_key().map(|dk| dk.to_vec()))
359 }
360
361 pub fn get_link_route_hint(&self, link_id: &LinkId) -> Option<LinkRouteHint> {
363 self.links.get(link_id).and_then(|link| {
364 link.route_interface.map(|interface| LinkRouteHint {
365 interface,
366 transport_id: link.route_transport_id,
367 })
368 })
369 }
370
371 pub fn set_link_route_hint(
373 &mut self,
374 link_id: &LinkId,
375 interface: rns_core::transport::types::InterfaceId,
376 transport_id: Option<[u8; 16]>,
377 ) -> bool {
378 let Some(link) = self.links.get_mut(link_id) else {
379 return false;
380 };
381 link.route_interface = Some(interface);
382 link.route_transport_id = transport_id;
383 true
384 }
385
386 pub fn register_link_destination(
388 &mut self,
389 dest_hash: [u8; 16],
390 sig_prv: Ed25519PrivateKey,
391 sig_pub_bytes: [u8; 32],
392 resource_strategy: ResourceStrategy,
393 ) {
394 self.link_destinations.insert(
395 dest_hash,
396 LinkDestination {
397 sig_prv,
398 sig_pub_bytes,
399 resource_strategy,
400 max_request_size: None,
401 },
402 );
403 }
404
405 pub fn set_link_destination_max_request_size(
410 &mut self,
411 dest_hash: &[u8; 16],
412 max_request_size: Option<usize>,
413 ) -> bool {
414 let Some(destination) = self.link_destinations.get_mut(dest_hash) else {
415 return false;
416 };
417 destination.max_request_size = max_request_size;
418 for link in self.links.values_mut() {
419 if link.dest_hash == *dest_hash && link.dest_sig_pub_bytes.is_none() {
420 link.max_request_size = max_request_size;
421 }
422 }
423 true
424 }
425
426 pub fn deregister_link_destination(&mut self, dest_hash: &[u8; 16]) {
428 self.link_destinations.remove(dest_hash);
429 }
430
431 pub fn register_request_handler<F>(
437 &mut self,
438 path: &str,
439 allowed_list: Option<Vec<[u8; 16]>>,
440 handler: F,
441 ) where
442 F: Fn(LinkId, &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<Vec<u8>>
443 + Send
444 + 'static,
445 {
446 let path_hash = compute_path_hash(path);
447 self.request_handlers.push(RequestHandlerEntry {
448 path: path.to_string(),
449 path_hash,
450 allowed_list,
451 handler: Box::new(move |link_id, p, data, remote| {
452 handler(link_id, p, data, remote).map(RequestResponse::Bytes)
453 }),
454 });
455 }
456
457 pub fn register_request_handler_response<F>(
459 &mut self,
460 path: &str,
461 allowed_list: Option<Vec<[u8; 16]>>,
462 handler: F,
463 ) where
464 F: Fn(LinkId, &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<RequestResponse>
465 + Send
466 + 'static,
467 {
468 let path_hash = compute_path_hash(path);
469 self.request_handlers.push(RequestHandlerEntry {
470 path: path.to_string(),
471 path_hash,
472 allowed_list,
473 handler: Box::new(handler),
474 });
475 }
476
477 pub fn create_link(
485 &mut self,
486 dest_hash: &[u8; 16],
487 dest_sig_pub_bytes: &[u8; 32],
488 hops: u8,
489 mtu: u32,
490 rng: &mut dyn Rng,
491 ) -> (LinkId, Vec<LinkManagerAction>) {
492 let mode = LinkMode::Aes256Cbc;
493 let (mut engine, request_data) =
494 LinkEngine::new_initiator(dest_hash, hops, mode, Some(mtu), time::now(), rng);
495
496 let flags = PacketFlags {
498 header_type: constants::HEADER_1,
499 context_flag: constants::FLAG_UNSET,
500 transport_type: constants::TRANSPORT_BROADCAST,
501 destination_type: constants::DESTINATION_SINGLE,
502 packet_type: constants::PACKET_TYPE_LINKREQUEST,
503 };
504
505 let packet = match RawPacket::pack(
506 flags,
507 0,
508 dest_hash,
509 None,
510 constants::CONTEXT_NONE,
511 &request_data,
512 ) {
513 Ok(p) => p,
514 Err(_) => {
515 return ([0u8; 16], Vec::new());
517 }
518 };
519
520 engine.set_link_id_from_hashable(&packet.get_hashable_part(), request_data.len());
521 let link_id = *engine.link_id();
522
523 engine.record_outbound_traffic(packet.data.len());
524 let managed = ManagedLink {
525 engine,
526 channel: None,
527 pending_channel_packets: HashMap::new(),
528 channel_send_ok: 0,
529 channel_send_not_ready: 0,
530 channel_send_too_big: 0,
531 channel_send_other_error: 0,
532 channel_messages_received: 0,
533 channel_proofs_sent: 0,
534 channel_proofs_received: 0,
535 dest_hash: *dest_hash,
536 remote_identity: None,
537 dest_sig_pub_bytes: Some(*dest_sig_pub_bytes),
538 incoming_resources: Vec::new(),
539 outgoing_resources: Vec::new(),
540 pending_requests: HashMap::new(),
541 incoming_splits: HashMap::new(),
542 outgoing_splits: HashMap::new(),
543 resource_strategy: ResourceStrategy::default(),
544 max_request_size: None,
545 route_interface: None,
546 route_transport_id: None,
547 };
548 self.links.insert(link_id, managed);
549
550 let mut actions = Vec::new();
551 actions.push(LinkManagerAction::RegisterLinkDest { link_id });
553 actions.push(LinkManagerAction::SendPacket {
555 raw: packet.raw,
556 dest_type: constants::DESTINATION_LINK,
557 attached_interface: None,
558 });
559
560 (link_id, actions)
561 }
562
563 pub fn handle_local_delivery(
569 &mut self,
570 dest_hash: [u8; 16],
571 raw: &[u8],
572 packet_hash: [u8; 32],
573 receiving_interface: rns_core::transport::types::InterfaceId,
574 rng: &mut dyn Rng,
575 ) -> Vec<LinkManagerAction> {
576 let packet = match RawPacket::unpack(raw) {
577 Ok(p) => p,
578 Err(_) => return Vec::new(),
579 };
580
581 if packet.flags.packet_type != constants::PACKET_TYPE_LINKREQUEST {
584 if let Some(link) = self.links.get_mut(&dest_hash) {
585 link.engine.record_inbound_traffic(packet.data.len());
586 }
587 }
588
589 match packet.flags.packet_type {
590 constants::PACKET_TYPE_LINKREQUEST => {
591 self.handle_linkrequest(&dest_hash, &packet, receiving_interface, rng)
592 }
593 constants::PACKET_TYPE_PROOF if packet.context == constants::CONTEXT_LRPROOF => {
594 self.handle_lrproof(&dest_hash, &packet, receiving_interface, rng)
596 }
597 constants::PACKET_TYPE_PROOF if packet.context == constants::CONTEXT_RESOURCE_PRF => {
598 self.handle_resource_prf(&dest_hash, &packet.data, rng)
601 }
602 constants::PACKET_TYPE_PROOF => self.handle_link_proof(&dest_hash, &packet, rng),
603 constants::PACKET_TYPE_DATA => {
604 self.handle_link_data(&dest_hash, &packet, packet_hash, receiving_interface, rng)
605 }
606 _ => Vec::new(),
607 }
608 }
609
610 fn handle_linkrequest(
612 &mut self,
613 dest_hash: &[u8; 16],
614 packet: &RawPacket,
615 receiving_interface: rns_core::transport::types::InterfaceId,
616 rng: &mut dyn Rng,
617 ) -> Vec<LinkManagerAction> {
618 let ld = match self.link_destinations.get(dest_hash) {
620 Some(ld) => ld,
621 None => return Vec::new(),
622 };
623
624 let hashable = packet.get_hashable_part();
625 let now = time::now();
626
627 let (engine, lrproof_data) = match LinkEngine::new_responder(
629 &ld.sig_prv,
630 &ld.sig_pub_bytes,
631 &packet.data,
632 &hashable,
633 dest_hash,
634 packet.hops,
635 now,
636 rng,
637 ) {
638 Ok(r) => r,
639 Err(e) => {
640 log::debug!("LINKREQUEST rejected: {}", e);
641 return Vec::new();
642 }
643 };
644
645 let link_id = *engine.link_id();
646 log::debug!(
647 "LINKREQUEST accepted: link={:02x?} iface={} header_type={} transport_id_present={} hops={}",
648 &link_id[..4],
649 receiving_interface.0,
650 packet.flags.header_type,
651 packet.transport_id.is_some(),
652 packet.hops
653 );
654
655 let managed = ManagedLink {
656 engine,
657 channel: None,
658 pending_channel_packets: HashMap::new(),
659 channel_send_ok: 0,
660 channel_send_not_ready: 0,
661 channel_send_too_big: 0,
662 channel_send_other_error: 0,
663 channel_messages_received: 0,
664 channel_proofs_sent: 0,
665 channel_proofs_received: 0,
666 dest_hash: *dest_hash,
667 remote_identity: None,
668 dest_sig_pub_bytes: None,
669 incoming_resources: Vec::new(),
670 outgoing_resources: Vec::new(),
671 pending_requests: HashMap::new(),
672 incoming_splits: HashMap::new(),
673 outgoing_splits: HashMap::new(),
674 resource_strategy: ld.resource_strategy,
675 max_request_size: ld.max_request_size,
676 route_interface: Some(receiving_interface),
677 route_transport_id: if packet.flags.header_type == constants::HEADER_2 {
678 packet.transport_id
679 } else {
680 None
681 },
682 };
683 self.links.insert(link_id, managed);
684
685 let flags = PacketFlags {
687 header_type: constants::HEADER_1,
688 context_flag: constants::FLAG_UNSET,
689 transport_type: constants::TRANSPORT_BROADCAST,
690 destination_type: constants::DESTINATION_LINK,
691 packet_type: constants::PACKET_TYPE_PROOF,
692 };
693
694 let mut actions = Vec::new();
695
696 actions.push(LinkManagerAction::RegisterLinkDest { link_id });
698
699 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
700 flags,
701 0,
702 &link_id,
703 None,
704 constants::CONTEXT_LRPROOF,
705 &lrproof_data,
706 ) {
707 log::debug!(
708 "LRPROOF queued: link={:02x?} route_iface={} route_tid_present={} hops=0",
709 &link_id[..4],
710 receiving_interface.0,
711 packet.transport_id.is_some()
712 );
713 actions.push(LinkManagerAction::SendPacket {
714 raw,
715 dest_type: constants::DESTINATION_LINK,
716 attached_interface: None,
717 });
718 }
719
720 actions.push(LinkManagerAction::LinkRequestReceived {
722 link_id,
723 receiving_interface,
724 });
725
726 actions
727 }
728
729 fn handle_link_proof(
730 &mut self,
731 link_id: &LinkId,
732 packet: &RawPacket,
733 rng: &mut dyn Rng,
734 ) -> Vec<LinkManagerAction> {
735 if packet.data.len() < 32 {
736 return Vec::new();
737 }
738
739 let mut tracked_hash = [0u8; 32];
740 tracked_hash.copy_from_slice(&packet.data[..32]);
741
742 let Some(link) = self.links.get_mut(link_id) else {
743 return Vec::new();
744 };
745 let Some(sequence) = link.pending_channel_packets.remove(&tracked_hash) else {
746 return Vec::new();
747 };
748 link.channel_proofs_received += 1;
749 let Some(channel) = link.channel.as_mut() else {
750 return Vec::new();
751 };
752
753 let chan_actions = channel.packet_delivered(sequence);
754 let _ = channel;
755 let _ = link;
756 self.process_channel_actions(link_id, chan_actions, rng)
757 }
758
759 fn build_link_packet_proof(
760 &mut self,
761 link_id: &LinkId,
762 packet_hash: &[u8; 32],
763 ) -> Vec<LinkManagerAction> {
764 let dest_hash = match self.links.get(link_id) {
765 Some(link) => link.dest_hash,
766 None => return Vec::new(),
767 };
768 let Some(ld) = self.link_destinations.get(&dest_hash) else {
769 return Vec::new();
770 };
771 if let Some(link) = self.links.get_mut(link_id) {
772 link.channel_proofs_sent += 1;
773 }
774
775 let signature = ld.sig_prv.sign(packet_hash);
776 let mut proof_data = Vec::with_capacity(96);
777 proof_data.extend_from_slice(packet_hash);
778 proof_data.extend_from_slice(&signature);
779
780 let flags = PacketFlags {
781 header_type: constants::HEADER_1,
782 context_flag: constants::FLAG_UNSET,
783 transport_type: constants::TRANSPORT_BROADCAST,
784 destination_type: constants::DESTINATION_LINK,
785 packet_type: constants::PACKET_TYPE_PROOF,
786 };
787 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
788 flags,
789 0,
790 link_id,
791 None,
792 constants::CONTEXT_NONE,
793 &proof_data,
794 ) {
795 vec![LinkManagerAction::SendPacket {
796 raw,
797 dest_type: constants::DESTINATION_LINK,
798 attached_interface: None,
799 }]
800 } else {
801 Vec::new()
802 }
803 }
804
805 fn handle_lrproof(
807 &mut self,
808 link_id_bytes: &[u8; 16],
809 packet: &RawPacket,
810 receiving_interface: rns_core::transport::types::InterfaceId,
811 rng: &mut dyn Rng,
812 ) -> Vec<LinkManagerAction> {
813 let link = match self.links.get_mut(link_id_bytes) {
814 Some(l) => l,
815 None => return Vec::new(),
816 };
817
818 link.route_interface = Some(receiving_interface);
819 if packet.flags.header_type == constants::HEADER_2 {
820 if let Some(transport_id) = packet.transport_id {
821 link.route_transport_id = Some(transport_id);
822 }
823 }
824 log::debug!(
825 "LRPROOF received: link={:02x?} iface={} header_type={} transport_id_present={}",
826 &link_id_bytes[..4],
827 receiving_interface.0,
828 packet.flags.header_type,
829 packet.transport_id.is_some()
830 );
831
832 if link.engine.state() != LinkState::Pending || !link.engine.is_initiator() {
833 return Vec::new();
834 }
835
836 let dest_sig_pub_bytes = match link.dest_sig_pub_bytes {
838 Some(b) => b,
839 None => {
840 log::debug!("LRPROOF: no destination signing key available");
841 return Vec::new();
842 }
843 };
844
845 let now = time::now();
846 let (lrrtt_encrypted, link_actions) = match link.engine.handle_lrproof_with_hops(
847 &packet.data,
848 &dest_sig_pub_bytes,
849 Some(packet.hops),
850 now,
851 rng,
852 ) {
853 Ok(r) => r,
854 Err(e) => {
855 log::debug!("LRPROOF validation failed: {}", e);
856 return Vec::new();
857 }
858 };
859
860 let link_id = *link.engine.link_id();
861 let mut actions = Vec::new();
862
863 actions.extend(self.process_link_actions(&link_id, &link_actions));
865
866 let flags = PacketFlags {
868 header_type: constants::HEADER_1,
869 context_flag: constants::FLAG_UNSET,
870 transport_type: constants::TRANSPORT_BROADCAST,
871 destination_type: constants::DESTINATION_LINK,
872 packet_type: constants::PACKET_TYPE_DATA,
873 };
874
875 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
876 flags,
877 0,
878 &link_id,
879 None,
880 constants::CONTEXT_LRRTT,
881 &lrrtt_encrypted,
882 ) {
883 actions.push(LinkManagerAction::SendPacket {
884 raw,
885 dest_type: constants::DESTINATION_LINK,
886 attached_interface: None,
887 });
888 }
889
890 if let Some(link) = self.links.get_mut(&link_id) {
892 if link.engine.state() == LinkState::Active {
893 let rtt = link.engine.rtt().unwrap_or(1.0);
894 link.channel = Some(Channel::new(rtt));
895 }
896 }
897
898 actions
899 }
900
901 fn handle_link_data(
907 &mut self,
908 link_id_bytes: &[u8; 16],
909 packet: &RawPacket,
910 packet_hash: [u8; 32],
911 receiving_interface: rns_core::transport::types::InterfaceId,
912 rng: &mut dyn Rng,
913 ) -> Vec<LinkManagerAction> {
914 enum LinkDataResult<'a> {
916 Lrrtt {
917 link_id: LinkId,
918 link_actions: Vec<LinkAction>,
919 },
920 Identify {
921 link_id: LinkId,
922 link_actions: Vec<LinkAction>,
923 },
924 Keepalive {
925 link_id: LinkId,
926 inbound_actions: Vec<LinkAction>,
927 reply: bool,
928 received_at: f64,
929 },
930 LinkClose {
931 link_id: LinkId,
932 teardown_actions: Vec<LinkAction>,
933 },
934 Channel {
935 link_id: LinkId,
936 inbound_actions: Vec<LinkAction>,
937 plaintext: Vec<u8>,
938 packet_hash: [u8; 32],
939 },
940 Request {
941 link_id: LinkId,
942 inbound_actions: Vec<LinkAction>,
943 plaintext: Vec<u8>,
944 request_id: [u8; 16],
945 },
946 Response {
947 link_id: LinkId,
948 inbound_actions: Vec<LinkAction>,
949 plaintext: Vec<u8>,
950 },
951 Generic {
952 link_id: LinkId,
953 inbound_actions: Vec<LinkAction>,
954 plaintext: Vec<u8>,
955 context: u8,
956 packet_hash: [u8; 32],
957 },
958 ResourceAdv {
960 link_id: LinkId,
961 inbound_actions: Vec<LinkAction>,
962 plaintext: Vec<u8>,
963 },
964 ResourceReq {
966 link_id: LinkId,
967 inbound_actions: Vec<LinkAction>,
968 plaintext: Vec<u8>,
969 },
970 ResourceHmu {
972 link_id: LinkId,
973 inbound_actions: Vec<LinkAction>,
974 plaintext: Vec<u8>,
975 },
976 ResourcePart {
978 link_id: LinkId,
979 inbound_actions: Vec<LinkAction>,
980 raw_data: &'a [u8],
981 },
982 ResourcePrf {
984 link_id: LinkId,
985 inbound_actions: Vec<LinkAction>,
986 plaintext: Vec<u8>,
987 },
988 ResourceIcl {
990 link_id: LinkId,
991 inbound_actions: Vec<LinkAction>,
992 },
993 ResourceRcl {
995 link_id: LinkId,
996 inbound_actions: Vec<LinkAction>,
997 },
998 Error,
999 }
1000
1001 let result = {
1002 let link = match self.links.get_mut(link_id_bytes) {
1003 Some(l) => l,
1004 None => return Vec::new(),
1005 };
1006
1007 link.route_interface = Some(receiving_interface);
1008 if packet.flags.header_type == constants::HEADER_2 {
1009 if let Some(transport_id) = packet.transport_id {
1010 link.route_transport_id = Some(transport_id);
1011 }
1012 } else {
1013 link.route_transport_id = None;
1014 }
1015
1016 match packet.context {
1017 constants::CONTEXT_LRRTT => {
1018 match link.engine.handle_lrrtt_with_hops(
1019 &packet.data,
1020 Some(packet.hops),
1021 time::now(),
1022 ) {
1023 Ok(link_actions) => {
1024 let link_id = *link.engine.link_id();
1025 LinkDataResult::Lrrtt {
1026 link_id,
1027 link_actions,
1028 }
1029 }
1030 Err(e) => {
1031 log::debug!("LRRTT handling failed: {}", e);
1032 LinkDataResult::Error
1033 }
1034 }
1035 }
1036 constants::CONTEXT_LINKIDENTIFY => {
1037 match link.engine.handle_identify(&packet.data) {
1038 Ok(link_actions) => {
1039 let link_id = *link.engine.link_id();
1040 link.remote_identity = link.engine.remote_identity().cloned();
1041 LinkDataResult::Identify {
1042 link_id,
1043 link_actions,
1044 }
1045 }
1046 Err(e) => {
1047 log::debug!("LINKIDENTIFY failed: {}", e);
1048 LinkDataResult::Error
1049 }
1050 }
1051 }
1052 constants::CONTEXT_KEEPALIVE => {
1053 let received_at = time::now();
1054 let reply = link
1055 .engine
1056 .should_reply_keepalive(&packet.data, received_at);
1057 let inbound_actions = link.engine.record_inbound(received_at);
1058 let link_id = *link.engine.link_id();
1059 LinkDataResult::Keepalive {
1060 link_id,
1061 inbound_actions,
1062 reply,
1063 received_at,
1064 }
1065 }
1066 constants::CONTEXT_LINKCLOSE => match link.engine.decrypt(&packet.data) {
1067 Ok(plaintext) if plaintext.as_slice() == link_id_bytes => {
1068 let teardown_actions = link.engine.handle_teardown();
1069 let link_id = *link.engine.link_id();
1070 LinkDataResult::LinkClose {
1071 link_id,
1072 teardown_actions,
1073 }
1074 }
1075 _ => LinkDataResult::Error,
1076 },
1077 constants::CONTEXT_CHANNEL => match link.engine.decrypt(&packet.data) {
1078 Ok(plaintext) => {
1079 let inbound_actions = link.engine.record_inbound(time::now());
1080 let link_id = *link.engine.link_id();
1081 LinkDataResult::Channel {
1082 link_id,
1083 inbound_actions,
1084 plaintext,
1085 packet_hash,
1086 }
1087 }
1088 Err(_) => LinkDataResult::Error,
1089 },
1090 constants::CONTEXT_REQUEST => match link.engine.decrypt(&packet.data) {
1091 Ok(plaintext) => {
1092 let inbound_actions = link.engine.record_inbound(time::now());
1093 let link_id = *link.engine.link_id();
1094 let request_id = packet.get_truncated_hash();
1095 LinkDataResult::Request {
1096 link_id,
1097 inbound_actions,
1098 plaintext,
1099 request_id,
1100 }
1101 }
1102 Err(_) => LinkDataResult::Error,
1103 },
1104 constants::CONTEXT_RESPONSE => match link.engine.decrypt(&packet.data) {
1105 Ok(plaintext) => {
1106 let inbound_actions = link.engine.record_inbound(time::now());
1107 let link_id = *link.engine.link_id();
1108 LinkDataResult::Response {
1109 link_id,
1110 inbound_actions,
1111 plaintext,
1112 }
1113 }
1114 Err(_) => LinkDataResult::Error,
1115 },
1116 constants::CONTEXT_RESOURCE_ADV => match link.engine.decrypt(&packet.data) {
1118 Ok(plaintext) => {
1119 let inbound_actions = link.engine.record_inbound(time::now());
1120 let link_id = *link.engine.link_id();
1121 LinkDataResult::ResourceAdv {
1122 link_id,
1123 inbound_actions,
1124 plaintext,
1125 }
1126 }
1127 Err(_) => LinkDataResult::Error,
1128 },
1129 constants::CONTEXT_RESOURCE_REQ => match link.engine.decrypt(&packet.data) {
1130 Ok(plaintext) => {
1131 let inbound_actions = link.engine.record_inbound(time::now());
1132 let link_id = *link.engine.link_id();
1133 LinkDataResult::ResourceReq {
1134 link_id,
1135 inbound_actions,
1136 plaintext,
1137 }
1138 }
1139 Err(_) => LinkDataResult::Error,
1140 },
1141 constants::CONTEXT_RESOURCE_HMU => match link.engine.decrypt(&packet.data) {
1142 Ok(plaintext) => {
1143 let inbound_actions = link.engine.record_inbound(time::now());
1144 let link_id = *link.engine.link_id();
1145 LinkDataResult::ResourceHmu {
1146 link_id,
1147 inbound_actions,
1148 plaintext,
1149 }
1150 }
1151 Err(_) => LinkDataResult::Error,
1152 },
1153 constants::CONTEXT_RESOURCE => {
1154 let inbound_actions = link.engine.record_inbound(time::now());
1156 let link_id = *link.engine.link_id();
1157 LinkDataResult::ResourcePart {
1158 link_id,
1159 inbound_actions,
1160 raw_data: &packet.data,
1161 }
1162 }
1163 constants::CONTEXT_RESOURCE_PRF => match link.engine.decrypt(&packet.data) {
1164 Ok(plaintext) => {
1165 let inbound_actions = link.engine.record_inbound(time::now());
1166 let link_id = *link.engine.link_id();
1167 LinkDataResult::ResourcePrf {
1168 link_id,
1169 inbound_actions,
1170 plaintext,
1171 }
1172 }
1173 Err(_) => LinkDataResult::Error,
1174 },
1175 constants::CONTEXT_RESOURCE_ICL => {
1176 let _ = link.engine.decrypt(&packet.data); let inbound_actions = link.engine.record_inbound(time::now());
1178 let link_id = *link.engine.link_id();
1179 LinkDataResult::ResourceIcl {
1180 link_id,
1181 inbound_actions,
1182 }
1183 }
1184 constants::CONTEXT_RESOURCE_RCL => {
1185 let _ = link.engine.decrypt(&packet.data); let inbound_actions = link.engine.record_inbound(time::now());
1187 let link_id = *link.engine.link_id();
1188 LinkDataResult::ResourceRcl {
1189 link_id,
1190 inbound_actions,
1191 }
1192 }
1193 _ => match link.engine.decrypt(&packet.data) {
1194 Ok(plaintext) => {
1195 let inbound_actions = link.engine.record_inbound(time::now());
1196 let link_id = *link.engine.link_id();
1197 LinkDataResult::Generic {
1198 link_id,
1199 inbound_actions,
1200 plaintext,
1201 context: packet.context,
1202 packet_hash,
1203 }
1204 }
1205 Err(_) => LinkDataResult::Error,
1206 },
1207 }
1208 }; let mut actions = Vec::new();
1212 match result {
1213 LinkDataResult::Lrrtt {
1214 link_id,
1215 link_actions,
1216 } => {
1217 actions.extend(self.process_link_actions(&link_id, &link_actions));
1218 if let Some(link) = self.links.get_mut(&link_id) {
1220 if link.engine.state() == LinkState::Active {
1221 let rtt = link.engine.rtt().unwrap_or(1.0);
1222 link.channel = Some(Channel::new(rtt));
1223 }
1224 }
1225 }
1226 LinkDataResult::Identify {
1227 link_id,
1228 link_actions,
1229 } => {
1230 actions.extend(self.process_link_actions(&link_id, &link_actions));
1231 }
1232 LinkDataResult::Keepalive {
1233 link_id,
1234 inbound_actions,
1235 reply,
1236 received_at,
1237 } => {
1238 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1239 if reply {
1240 let flags = PacketFlags {
1241 header_type: constants::HEADER_1,
1242 context_flag: constants::FLAG_UNSET,
1243 transport_type: constants::TRANSPORT_BROADCAST,
1244 destination_type: constants::DESTINATION_LINK,
1245 packet_type: constants::PACKET_TYPE_DATA,
1246 };
1247 if let Ok((raw, _)) = RawPacket::pack_raw_with_hash(
1248 flags,
1249 0,
1250 &link_id,
1251 None,
1252 constants::CONTEXT_KEEPALIVE,
1253 &[0xfe],
1254 ) {
1255 actions.push(LinkManagerAction::SendPacket {
1256 raw,
1257 dest_type: constants::DESTINATION_LINK,
1258 attached_interface: None,
1259 });
1260 if let Some(link) = self.links.get_mut(&link_id) {
1261 link.engine.record_outbound(received_at, true);
1262 }
1263 }
1264 }
1265 }
1266 LinkDataResult::LinkClose {
1267 link_id,
1268 teardown_actions,
1269 } => {
1270 actions.extend(self.process_link_actions(&link_id, &teardown_actions));
1271 }
1272 LinkDataResult::Channel {
1273 link_id,
1274 inbound_actions,
1275 plaintext,
1276 packet_hash,
1277 } => {
1278 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1279 if let Some(link) = self.links.get_mut(&link_id) {
1281 if let Some(ref mut channel) = link.channel {
1282 let chan_actions = channel.receive(&plaintext, time::now());
1283 link.channel_messages_received += chan_actions
1284 .iter()
1285 .filter(|action| {
1286 matches!(
1287 action,
1288 rns_core::channel::ChannelAction::MessageReceived { .. }
1289 )
1290 })
1291 .count()
1292 as u64;
1293 let _ = link;
1295 actions.extend(self.process_channel_actions(&link_id, chan_actions, rng));
1296 }
1297 }
1298 actions.extend(self.build_link_packet_proof(&link_id, &packet_hash));
1299 }
1300 LinkDataResult::Request {
1301 link_id,
1302 inbound_actions,
1303 plaintext,
1304 request_id,
1305 } => {
1306 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1307 let size_ok = self
1308 .links
1309 .get(&link_id)
1310 .and_then(|link| link.max_request_size)
1311 .is_none_or(|limit| plaintext.len() <= limit);
1312 if size_ok {
1313 actions.extend(self.handle_request(&link_id, &plaintext, request_id, rng));
1314 } else {
1315 log::debug!(
1316 "ignored request with excessive size {} bytes on link {:02x?}",
1317 plaintext.len(),
1318 &link_id[..4]
1319 );
1320 }
1321 }
1322 LinkDataResult::Response {
1323 link_id,
1324 inbound_actions,
1325 plaintext,
1326 } => {
1327 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1328 actions.extend(self.handle_response(&link_id, &plaintext, None, None));
1330 }
1331 LinkDataResult::Generic {
1332 link_id,
1333 inbound_actions,
1334 plaintext,
1335 context,
1336 packet_hash,
1337 } => {
1338 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1339 actions.push(LinkManagerAction::LinkDataReceived {
1340 link_id,
1341 context,
1342 data: plaintext,
1343 });
1344
1345 actions.extend(self.build_link_packet_proof(&link_id, &packet_hash));
1346 }
1347 LinkDataResult::ResourceAdv {
1348 link_id,
1349 inbound_actions,
1350 plaintext,
1351 } => {
1352 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1353 actions.extend(self.handle_resource_adv(&link_id, &plaintext, rng));
1354 }
1355 LinkDataResult::ResourceReq {
1356 link_id,
1357 inbound_actions,
1358 plaintext,
1359 } => {
1360 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1361 actions.extend(self.handle_resource_req(&link_id, &plaintext, rng));
1362 }
1363 LinkDataResult::ResourceHmu {
1364 link_id,
1365 inbound_actions,
1366 plaintext,
1367 } => {
1368 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1369 actions.extend(self.handle_resource_hmu(&link_id, &plaintext, rng));
1370 }
1371 LinkDataResult::ResourcePart {
1372 link_id,
1373 inbound_actions,
1374 raw_data,
1375 } => {
1376 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1377 actions.extend(self.handle_resource_part(&link_id, &raw_data, rng));
1378 }
1379 LinkDataResult::ResourcePrf {
1380 link_id,
1381 inbound_actions,
1382 plaintext,
1383 } => {
1384 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1385 actions.extend(self.handle_resource_prf(&link_id, &plaintext, rng));
1386 }
1387 LinkDataResult::ResourceIcl {
1388 link_id,
1389 inbound_actions,
1390 } => {
1391 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1392 actions.extend(self.handle_resource_icl(&link_id));
1393 }
1394 LinkDataResult::ResourceRcl {
1395 link_id,
1396 inbound_actions,
1397 } => {
1398 actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1399 actions.extend(self.handle_resource_rcl(&link_id));
1400 }
1401 LinkDataResult::Error => {}
1402 }
1403
1404 actions
1405 }
1406
1407 fn handle_request(
1409 &mut self,
1410 link_id: &LinkId,
1411 plaintext: &[u8],
1412 request_id: [u8; 16],
1413 rng: &mut dyn Rng,
1414 ) -> Vec<LinkManagerAction> {
1415 use rns_core::msgpack::{self, Value};
1416
1417 let arr = match msgpack::unpack_exact(plaintext) {
1419 Ok(Value::Array(arr)) if arr.len() >= 3 => arr,
1420 _ => return Vec::new(),
1421 };
1422
1423 let path_hash_bytes = match &arr[1] {
1424 Value::Bin(b) if b.len() == 16 => b,
1425 _ => return Vec::new(),
1426 };
1427 let mut path_hash = [0u8; 16];
1428 path_hash.copy_from_slice(path_hash_bytes);
1429
1430 let request_data = msgpack::pack(&arr[2]);
1432
1433 if self.management_paths.contains(&path_hash) {
1435 let remote_identity = self
1436 .links
1437 .get(link_id)
1438 .and_then(|l| l.remote_identity)
1439 .map(|(h, k)| (h, k));
1440 return vec![LinkManagerAction::ManagementRequest {
1441 link_id: *link_id,
1442 path_hash,
1443 data: request_data,
1444 request_id,
1445 remote_identity,
1446 }];
1447 }
1448
1449 let handler_idx = self
1451 .request_handlers
1452 .iter()
1453 .position(|h| h.path_hash == path_hash);
1454 let handler_idx = match handler_idx {
1455 Some(i) => i,
1456 None => return Vec::new(),
1457 };
1458
1459 let remote_identity = self
1461 .links
1462 .get(link_id)
1463 .and_then(|l| l.remote_identity.as_ref());
1464 let handler = &self.request_handlers[handler_idx];
1465 if let Some(ref allowed) = handler.allowed_list {
1466 match remote_identity {
1467 Some((identity_hash, _)) => {
1468 if !allowed.contains(identity_hash) {
1469 log::debug!("Request denied: identity not in allowed list");
1470 return Vec::new();
1471 }
1472 }
1473 None => {
1474 log::debug!("Request denied: peer not identified");
1475 return Vec::new();
1476 }
1477 }
1478 }
1479
1480 let path = handler.path.clone();
1482 let response = (handler.handler)(*link_id, &path, &request_data, remote_identity);
1483
1484 let mut actions = Vec::new();
1485 if let Some(response) = response {
1486 match response {
1487 RequestResponse::Bytes(response_data) => {
1488 let mut response_actions =
1489 self.build_response_packet(link_id, &request_id, &response_data, rng);
1490 if response_actions.is_empty() {
1491 response_actions.extend(self.send_response_resource(
1492 link_id,
1493 &request_id,
1494 &response_data,
1495 None,
1496 true,
1497 rng,
1498 ));
1499 }
1500 actions.extend(response_actions);
1501 }
1502 RequestResponse::Resource {
1503 data,
1504 metadata,
1505 auto_compress,
1506 } => {
1507 actions.extend(self.send_response_resource(
1508 link_id,
1509 &request_id,
1510 &data,
1511 metadata.as_deref(),
1512 auto_compress,
1513 rng,
1514 ));
1515 }
1516 }
1517 }
1518
1519 actions
1520 }
1521
1522 fn build_response_packet(
1525 &self,
1526 link_id: &LinkId,
1527 request_id: &[u8; 16],
1528 response_data: &[u8],
1529 rng: &mut dyn Rng,
1530 ) -> Vec<LinkManagerAction> {
1531 use rns_core::msgpack::{self, Value};
1532
1533 let response_value = msgpack::unpack_exact(response_data)
1534 .unwrap_or_else(|_| Value::Bin(response_data.to_vec()));
1535
1536 let response_array = Value::Array(vec![Value::Bin(request_id.to_vec()), response_value]);
1537 let response_plaintext = msgpack::pack(&response_array);
1538
1539 let mut actions = Vec::new();
1540 if let Some(link) = self.links.get(link_id) {
1541 if let Ok(encrypted) = link.engine.encrypt(&response_plaintext, rng) {
1542 let flags = PacketFlags {
1543 header_type: constants::HEADER_1,
1544 context_flag: constants::FLAG_UNSET,
1545 transport_type: constants::TRANSPORT_BROADCAST,
1546 destination_type: constants::DESTINATION_LINK,
1547 packet_type: constants::PACKET_TYPE_DATA,
1548 };
1549 let max_mtu = link.engine.mtu() as usize;
1550 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
1551 flags,
1552 0,
1553 link_id,
1554 None,
1555 constants::CONTEXT_RESPONSE,
1556 &encrypted,
1557 max_mtu,
1558 ) {
1559 actions.push(LinkManagerAction::SendPacket {
1560 raw,
1561 dest_type: constants::DESTINATION_LINK,
1562 attached_interface: None,
1563 });
1564 }
1565 }
1566 }
1567 actions
1568 }
1569
1570 fn send_response_resource(
1571 &mut self,
1572 link_id: &LinkId,
1573 request_id: &[u8; 16],
1574 response_data: &[u8],
1575 metadata: Option<&[u8]>,
1576 auto_compress: bool,
1577 rng: &mut dyn Rng,
1578 ) -> Vec<LinkManagerAction> {
1579 use rns_core::msgpack::{self, Value};
1580
1581 let link = match self.links.get_mut(link_id) {
1582 Some(l) => l,
1583 None => return Vec::new(),
1584 };
1585
1586 if link.engine.state() != LinkState::Active {
1587 return Vec::new();
1588 }
1589
1590 let now = time::now();
1591
1592 let response_value = msgpack::unpack_exact(response_data)
1596 .unwrap_or_else(|_| Value::Bin(response_data.to_vec()));
1597 let response_array = Value::Array(vec![Value::Bin(request_id.to_vec()), response_value]);
1598 let resource_payload = msgpack::pack(&response_array);
1599
1600 let senders = match Self::build_resource_senders(
1601 link,
1602 &resource_payload,
1603 metadata,
1604 auto_compress,
1605 true, Some(request_id.to_vec()),
1607 rng,
1608 now,
1609 ) {
1610 Ok(s) => s,
1611 Err(e) => {
1612 log::debug!("Failed to create response ResourceSender: {}", e);
1613 return Vec::new();
1614 }
1615 };
1616
1617 let adv_actions = Self::start_resource_senders(link, senders, now);
1618
1619 let _ = link;
1620 self.process_resource_actions(link_id, adv_actions, rng)
1621 }
1622
1623 pub fn send_management_response(
1626 &mut self,
1627 link_id: &LinkId,
1628 request_id: &[u8; 16],
1629 response_data: &[u8],
1630 rng: &mut dyn Rng,
1631 ) -> Vec<LinkManagerAction> {
1632 let mut actions = self.build_response_packet(link_id, request_id, response_data, rng);
1633 if actions.is_empty() {
1634 actions.extend(self.send_response_resource(
1635 link_id,
1636 request_id,
1637 response_data,
1638 None,
1639 true,
1640 rng,
1641 ));
1642 }
1643 actions
1644 }
1645
1646 pub fn send_request(
1654 &mut self,
1655 link_id: &LinkId,
1656 path: &str,
1657 data: &[u8],
1658 rng: &mut dyn Rng,
1659 ) -> Vec<LinkManagerAction> {
1660 self.send_request_with_max_response_size(link_id, path, data, None, rng)
1661 }
1662
1663 pub fn send_request_with_max_response_size(
1665 &mut self,
1666 link_id: &LinkId,
1667 path: &str,
1668 data: &[u8],
1669 max_response_size: Option<usize>,
1670 rng: &mut dyn Rng,
1671 ) -> Vec<LinkManagerAction> {
1672 use rns_core::msgpack::{self, Value};
1673
1674 let link = match self.links.get_mut(link_id) {
1675 Some(l) => l,
1676 None => return Vec::new(),
1677 };
1678
1679 if link.engine.state() != LinkState::Active {
1680 return Vec::new();
1681 }
1682
1683 let path_hash = compute_path_hash(path);
1684
1685 let data_value = msgpack::unpack_exact(data).unwrap_or_else(|_| Value::Bin(data.to_vec()));
1687
1688 let request_array = Value::Array(vec![
1690 Value::Float(time::now()),
1691 Value::Bin(path_hash.to_vec()),
1692 data_value,
1693 ]);
1694 let plaintext = msgpack::pack(&request_array);
1695
1696 if plaintext.len() > link.engine.mdu() {
1697 let request_id = rns_core::hash::truncated_hash(&plaintext);
1698 let now = time::now();
1699 let senders = match Self::build_resource_senders(
1700 link,
1701 &plaintext,
1702 None,
1703 true,
1704 false,
1705 Some(request_id.to_vec()),
1706 rng,
1707 now,
1708 ) {
1709 Ok(senders) => senders,
1710 Err(e) => {
1711 log::debug!("Failed to create request ResourceSender: {}", e);
1712 return Vec::new();
1713 }
1714 };
1715 link.pending_requests.insert(
1718 request_id,
1719 PendingRequest {
1720 deadline: None,
1721 max_response_size,
1722 },
1723 );
1724 let adv_actions = Self::start_resource_senders(link, senders, now);
1725 let _ = link;
1726 return self.process_resource_actions(link_id, adv_actions, rng);
1727 }
1728
1729 let encrypted = match link.engine.encrypt(&plaintext, rng) {
1730 Ok(e) => e,
1731 Err(_) => return Vec::new(),
1732 };
1733
1734 let flags = PacketFlags {
1735 header_type: constants::HEADER_1,
1736 context_flag: constants::FLAG_UNSET,
1737 transport_type: constants::TRANSPORT_BROADCAST,
1738 destination_type: constants::DESTINATION_LINK,
1739 packet_type: constants::PACKET_TYPE_DATA,
1740 };
1741
1742 let mut actions = Vec::new();
1743 let max_mtu = link.engine.mtu() as usize;
1744 if let Ok((raw, packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
1745 flags,
1746 0,
1747 link_id,
1748 None,
1749 constants::CONTEXT_REQUEST,
1750 &encrypted,
1751 max_mtu,
1752 ) {
1753 let mut request_id = [0u8; 16];
1754 request_id.copy_from_slice(&packet_hash[..16]);
1755 let deadline = Self::request_response_deadline(link, time::now());
1756 link.pending_requests.insert(
1757 request_id,
1758 PendingRequest {
1759 deadline: Some(deadline),
1760 max_response_size,
1761 },
1762 );
1763 actions.push(LinkManagerAction::SendPacket {
1764 raw,
1765 dest_type: constants::DESTINATION_LINK,
1766 attached_interface: None,
1767 });
1768 }
1769 actions
1770 }
1771
1772 pub fn send_on_link(
1774 &self,
1775 link_id: &LinkId,
1776 plaintext: &[u8],
1777 context: u8,
1778 rng: &mut dyn Rng,
1779 ) -> Vec<LinkManagerAction> {
1780 let link = match self.links.get(link_id) {
1781 Some(l) => l,
1782 None => return Vec::new(),
1783 };
1784
1785 if link.engine.state() != LinkState::Active {
1786 return Vec::new();
1787 }
1788
1789 let encrypted = match link.engine.encrypt(plaintext, rng) {
1790 Ok(e) => e,
1791 Err(_) => return Vec::new(),
1792 };
1793
1794 let flags = PacketFlags {
1795 header_type: constants::HEADER_1,
1796 context_flag: constants::FLAG_UNSET,
1797 transport_type: constants::TRANSPORT_BROADCAST,
1798 destination_type: constants::DESTINATION_LINK,
1799 packet_type: constants::PACKET_TYPE_DATA,
1800 };
1801
1802 let mut actions = Vec::new();
1803 if let Ok((raw, _packet_hash)) =
1804 RawPacket::pack_raw_with_hash(flags, 0, link_id, None, context, &encrypted)
1805 {
1806 actions.push(LinkManagerAction::SendPacket {
1807 raw,
1808 dest_type: constants::DESTINATION_LINK,
1809 attached_interface: None,
1810 });
1811 }
1812 actions
1813 }
1814
1815 pub fn identify(
1817 &self,
1818 link_id: &LinkId,
1819 identity: &rns_crypto::identity::Identity,
1820 rng: &mut dyn Rng,
1821 ) -> Vec<LinkManagerAction> {
1822 let link = match self.links.get(link_id) {
1823 Some(l) => l,
1824 None => return Vec::new(),
1825 };
1826
1827 let encrypted = match link.engine.build_identify(identity, rng) {
1828 Ok(e) => e,
1829 Err(_) => return Vec::new(),
1830 };
1831
1832 let flags = PacketFlags {
1833 header_type: constants::HEADER_1,
1834 context_flag: constants::FLAG_UNSET,
1835 transport_type: constants::TRANSPORT_BROADCAST,
1836 destination_type: constants::DESTINATION_LINK,
1837 packet_type: constants::PACKET_TYPE_DATA,
1838 };
1839
1840 let mut actions = Vec::new();
1841 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
1842 flags,
1843 0,
1844 link_id,
1845 None,
1846 constants::CONTEXT_LINKIDENTIFY,
1847 &encrypted,
1848 ) {
1849 actions.push(LinkManagerAction::SendPacket {
1850 raw,
1851 dest_type: constants::DESTINATION_LINK,
1852 attached_interface: None,
1853 });
1854 }
1855 actions
1856 }
1857
1858 pub fn teardown_link(&mut self, link_id: &LinkId) -> Vec<LinkManagerAction> {
1860 let link = match self.links.get_mut(link_id) {
1861 Some(l) => l,
1862 None => return Vec::new(),
1863 };
1864
1865 let previous_state = link.engine.state();
1866 if previous_state == LinkState::Closed {
1867 return Vec::new();
1868 }
1869
1870 let encrypted_close = if previous_state == LinkState::Pending {
1871 None
1872 } else {
1873 let mut rng = OsRng;
1874 link.engine.encrypt(link_id, &mut rng).ok()
1875 };
1876 let teardown_actions = link.engine.teardown();
1877 if let Some(ref mut channel) = link.channel {
1878 channel.shutdown();
1879 }
1880
1881 let mut actions = self.process_link_actions(link_id, &teardown_actions);
1882
1883 if previous_state == LinkState::Pending {
1886 return actions;
1887 }
1888
1889 let flags = PacketFlags {
1891 header_type: constants::HEADER_1,
1892 context_flag: constants::FLAG_UNSET,
1893 transport_type: constants::TRANSPORT_BROADCAST,
1894 destination_type: constants::DESTINATION_LINK,
1895 packet_type: constants::PACKET_TYPE_DATA,
1896 };
1897 if let Some(encrypted_close) = encrypted_close {
1898 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
1899 flags,
1900 0,
1901 link_id,
1902 None,
1903 constants::CONTEXT_LINKCLOSE,
1904 &encrypted_close,
1905 ) {
1906 actions.push(LinkManagerAction::SendPacket {
1907 raw,
1908 dest_type: constants::DESTINATION_LINK,
1909 attached_interface: None,
1910 });
1911 }
1912 }
1913
1914 actions
1915 }
1916
1917 pub fn teardown_all_links(&mut self) -> Vec<LinkManagerAction> {
1919 let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
1920 let mut actions = Vec::new();
1921 for link_id in link_ids {
1922 actions.extend(self.teardown_link(&link_id));
1923 }
1924 actions
1925 }
1926
1927 fn handle_response(
1929 &mut self,
1930 link_id: &LinkId,
1931 plaintext: &[u8],
1932 metadata: Option<Vec<u8>>,
1933 resource_request_id: Option<[u8; 16]>,
1934 ) -> Vec<LinkManagerAction> {
1935 use rns_core::msgpack;
1936
1937 let packet_response = msgpack::unpack_exact(plaintext).ok().and_then(|value| {
1939 let msgpack::Value::Array(arr) = value else {
1940 return None;
1941 };
1942 if arr.len() < 2 {
1943 return None;
1944 }
1945 let msgpack::Value::Bin(request_id_bytes) = &arr[0] else {
1946 return None;
1947 };
1948 if request_id_bytes.len() != 16 {
1949 return None;
1950 }
1951 let mut request_id = [0u8; 16];
1952 request_id.copy_from_slice(request_id_bytes);
1953 Some((request_id, msgpack::pack(&arr[1])))
1954 });
1955
1956 let (request_id, response_data, check_size) = match packet_response {
1957 Some((request_id, response_data)) => (request_id, response_data, true),
1958 None => {
1959 let Some(request_id) = resource_request_id else {
1960 return Vec::new();
1961 };
1962 (
1963 request_id,
1964 msgpack::pack(&msgpack::Value::Bin(plaintext.to_vec())),
1965 false,
1966 )
1967 }
1968 };
1969
1970 let Some(link) = self.links.get_mut(link_id) else {
1971 return Vec::new();
1972 };
1973 let Some(pending) = link.pending_requests.get(&request_id) else {
1974 return Vec::new();
1975 };
1976 if check_size {
1977 let response_size = response_data.len().saturating_sub(2) as u64;
1981 if let Some(maximum) = pending.max_response_size {
1982 if response_size > maximum as u64 {
1983 link.pending_requests.remove(&request_id);
1984 log::debug!(
1985 "rejected response with excessive size {} bytes on link {:02x?}",
1986 response_size,
1987 &link_id[..4]
1988 );
1989 return vec![LinkManagerAction::RequestFailed {
1990 link_id: *link_id,
1991 request_id,
1992 reason: RequestFailure::ResponseTooLarge {
1993 size: response_size,
1994 maximum,
1995 },
1996 }];
1997 }
1998 }
1999 }
2000 link.pending_requests.remove(&request_id);
2001
2002 vec![LinkManagerAction::ResponseReceived {
2003 link_id: *link_id,
2004 request_id,
2005 data: response_data,
2006 metadata,
2007 }]
2008 }
2009
2010 fn response_request_id(request_id: &Option<Vec<u8>>) -> Option<[u8; 16]> {
2011 let bytes = request_id.as_deref()?;
2012 if bytes.len() != 16 {
2013 return None;
2014 }
2015 let mut out = [0u8; 16];
2016 out.copy_from_slice(bytes);
2017 Some(out)
2018 }
2019
2020 fn request_response_deadline(link: &ManagedLink, now: f64) -> f64 {
2021 now + link.engine.rtt().unwrap_or(1.0) * constants::LINK_TRAFFIC_TIMEOUT_FACTOR
2022 + constants::RESOURCE_RESPONSE_MAX_GRACE_TIME * 1.125
2023 }
2024
2025 fn build_resource_senders(
2026 link: &ManagedLink,
2027 data: &[u8],
2028 metadata: Option<&[u8]>,
2029 auto_compress: bool,
2030 is_response: bool,
2031 request_id: Option<Vec<u8>>,
2032 rng: &mut dyn Rng,
2033 now: f64,
2034 ) -> Result<Vec<ResourceSender>, rns_core::resource::ResourceError> {
2035 let link_rtt = link.engine.rtt().unwrap_or(1.0);
2036 let resource_sdu = Self::resource_sdu_for_link(link);
2037 let metadata_overhead = metadata.map(|m| 3 + m.len()).unwrap_or(0);
2038 let logical_size = metadata_overhead + data.len();
2039
2040 if logical_size <= constants::RESOURCE_MAX_EFFICIENT_SIZE {
2041 let enc_rng = std::cell::RefCell::new(rns_crypto::OsRng);
2042 let encrypt_fn = |plaintext: &[u8]| -> Vec<u8> {
2043 link.engine
2044 .encrypt(plaintext, &mut *enc_rng.borrow_mut())
2045 .unwrap_or_else(|_| plaintext.to_vec())
2046 };
2047 return ResourceSender::new(
2048 data,
2049 metadata,
2050 resource_sdu,
2051 &encrypt_fn,
2052 &Bzip2Compressor,
2053 rng,
2054 now,
2055 auto_compress,
2056 is_response,
2057 request_id,
2058 1,
2059 1,
2060 None,
2061 link_rtt,
2062 6.0,
2063 )
2064 .map(|sender| vec![sender]);
2065 }
2066
2067 if metadata_overhead > constants::RESOURCE_MAX_EFFICIENT_SIZE {
2068 return Err(rns_core::resource::ResourceError::TooLarge);
2069 }
2070
2071 let first_payload_len = core::cmp::min(
2072 data.len(),
2073 constants::RESOURCE_MAX_EFFICIENT_SIZE - metadata_overhead,
2074 );
2075 let remaining = data.len().saturating_sub(first_payload_len);
2076 let total_segments = 1 + remaining.div_ceil(constants::RESOURCE_MAX_EFFICIENT_SIZE) as u64;
2077
2078 let enc_rng = std::cell::RefCell::new(rns_crypto::OsRng);
2079 let encrypt_fn = |plaintext: &[u8]| -> Vec<u8> {
2080 link.engine
2081 .encrypt(plaintext, &mut *enc_rng.borrow_mut())
2082 .unwrap_or_else(|_| plaintext.to_vec())
2083 };
2084
2085 let mut senders = Vec::new();
2086 let mut first = ResourceSender::new(
2087 &data[..first_payload_len],
2088 metadata,
2089 resource_sdu,
2090 &encrypt_fn,
2091 &Bzip2Compressor,
2092 rng,
2093 now,
2094 auto_compress,
2095 is_response,
2096 request_id.clone(),
2097 1,
2098 total_segments,
2099 None,
2100 link_rtt,
2101 6.0,
2102 )?;
2103 first.data_size = logical_size;
2104 let original_hash = first.original_hash;
2105 let has_metadata = metadata.is_some();
2106 senders.push(first);
2107
2108 let mut offset = first_payload_len;
2109 let mut segment_index = 2;
2110 while offset < data.len() {
2111 let end = core::cmp::min(offset + constants::RESOURCE_MAX_EFFICIENT_SIZE, data.len());
2112 let mut sender = ResourceSender::new(
2113 &data[offset..end],
2114 None,
2115 resource_sdu,
2116 &encrypt_fn,
2117 &Bzip2Compressor,
2118 rng,
2119 now,
2120 auto_compress,
2121 is_response,
2122 request_id.clone(),
2123 segment_index,
2124 total_segments,
2125 Some(original_hash),
2126 link_rtt,
2127 6.0,
2128 )?;
2129 sender.data_size = logical_size;
2130 sender.flags.has_metadata = has_metadata;
2131 senders.push(sender);
2132 segment_index += 1;
2133 offset = end;
2134 }
2135
2136 Ok(senders)
2137 }
2138
2139 fn start_resource_senders(
2140 link: &mut ManagedLink,
2141 mut senders: Vec<ResourceSender>,
2142 now: f64,
2143 ) -> Vec<ResourceAction> {
2144 if senders.is_empty() {
2145 return Vec::new();
2146 }
2147
2148 let mut first = senders.remove(0);
2149 let adv_actions = first.advertise(now);
2150
2151 if first.total_segments > 1 {
2152 let original_hash = first.original_hash;
2153 let split = OutgoingSplitTransfer {
2154 total_segments: first.total_segments,
2155 completed_segments: 0,
2156 current_segment_index: first.segment_index,
2157 current_sent_parts: 0,
2158 current_total_parts: first.total_parts(),
2159 };
2160 link.outgoing_splits.insert(original_hash, split);
2161 }
2162
2163 link.outgoing_resources.push(first);
2164 link.outgoing_resources.extend(senders);
2165 adv_actions
2166 }
2167
2168 fn handle_resource_adv(
2170 &mut self,
2171 link_id: &LinkId,
2172 adv_plaintext: &[u8],
2173 rng: &mut dyn Rng,
2174 ) -> Vec<LinkManagerAction> {
2175 let has_request_handlers =
2176 !self.request_handlers.is_empty() || !self.management_paths.is_empty();
2177 let link = match self.links.get_mut(link_id) {
2178 Some(l) => l,
2179 None => return Vec::new(),
2180 };
2181
2182 let link_rtt = link.engine.rtt().unwrap_or(1.0);
2183 let resource_sdu = Self::resource_sdu_for_link(link);
2184 let now = time::now();
2185
2186 let receiver = match ResourceReceiver::from_advertisement(
2187 adv_plaintext,
2188 resource_sdu,
2189 link_rtt,
2190 now,
2191 None,
2192 None,
2193 ) {
2194 Ok(r) => r,
2195 Err(e) => {
2196 log::debug!("Resource ADV rejected: {}", e);
2197 let _ = link;
2198 return self.teardown_link(link_id);
2199 }
2200 };
2201
2202 let strategy = link.resource_strategy;
2203 let resource_hash = receiver.resource_hash.clone();
2204 let transfer_size = receiver.transfer_size;
2205 let has_metadata = receiver.has_metadata;
2206 let is_request = receiver.flags.is_request;
2207 let is_response = receiver.flags.is_response;
2208 let is_split = receiver.flags.split;
2209 let segment_index = receiver.segment_index;
2210 let total_segments = receiver.total_segments;
2211 let original_hash = match Self::resource_hash_key(&receiver.original_hash) {
2212 Some(key) => key,
2213 None => return Vec::new(),
2214 };
2215
2216 if is_split && segment_index > 1 {
2217 let should_accept = link
2218 .incoming_splits
2219 .get(&original_hash)
2220 .is_some_and(|split| {
2221 split.completed_segments + 1 == segment_index
2222 && split.total_segments == total_segments
2223 });
2224
2225 if !should_accept {
2226 let reject_actions = {
2227 let mut r = receiver;
2228 r.reject()
2229 };
2230 let _ = link;
2231 return self.process_resource_actions(link_id, reject_actions, rng);
2232 }
2233
2234 let current_total_parts = receiver.total_parts;
2235 link.incoming_resources.push(receiver);
2236 let idx = link.incoming_resources.len() - 1;
2237 if let Some(split) = link.incoming_splits.get_mut(&original_hash) {
2238 split.current_segment_index = segment_index;
2239 split.current_received_parts = 0;
2240 split.current_total_parts = current_total_parts;
2241 }
2242 let resource_actions = link.incoming_resources[idx].accept(now);
2243 let _ = link;
2244 return self.process_resource_actions(link_id, resource_actions, rng);
2245 }
2246
2247 if is_request {
2248 if !has_request_handlers || Self::response_request_id(&receiver.request_id).is_none() {
2251 return Vec::new();
2252 }
2253 if link
2254 .max_request_size
2255 .is_some_and(|limit| receiver.data_size > limit as u64)
2256 {
2257 log::debug!(
2258 "ignored request with excessive size {} bytes on link {:02x?}",
2259 receiver.data_size,
2260 &link_id[..4]
2261 );
2262 let reject_actions = {
2263 let mut receiver = receiver;
2264 receiver.reject()
2265 };
2266 let _ = link;
2267 return self.process_resource_actions(link_id, reject_actions, rng);
2268 }
2269 if is_split {
2270 link.incoming_splits.insert(
2271 original_hash,
2272 IncomingSplitTransfer {
2273 total_segments,
2274 completed_segments: 0,
2275 current_segment_index: segment_index,
2276 current_received_parts: 0,
2277 current_total_parts: receiver.total_parts,
2278 data: Vec::new(),
2279 metadata: None,
2280 is_request: true,
2281 is_response: false,
2282 },
2283 );
2284 }
2285 link.incoming_resources.push(receiver);
2286 let idx = link.incoming_resources.len() - 1;
2287 let resource_actions = link.incoming_resources[idx].accept(now);
2288 let _ = link;
2289 return self.process_resource_actions(link_id, resource_actions, rng);
2290 }
2291
2292 if is_response {
2293 let Some(request_id) = Self::response_request_id(&receiver.request_id) else {
2296 return Vec::new();
2297 };
2298 let Some(pending) = link.pending_requests.get(&request_id) else {
2299 return Vec::new();
2300 };
2301 if let Some(maximum) = pending.max_response_size {
2302 if receiver.data_size > maximum as u64 {
2303 let response_size = receiver.data_size;
2304 link.pending_requests.remove(&request_id);
2305 log::debug!(
2306 "rejected response with excessive size {} bytes on link {:02x?}",
2307 response_size,
2308 &link_id[..4]
2309 );
2310 let reject_actions = {
2311 let mut receiver = receiver;
2312 receiver.reject()
2313 };
2314 let _ = link;
2315 let mut actions = self.process_resource_actions(link_id, reject_actions, rng);
2316 actions.push(LinkManagerAction::RequestFailed {
2317 link_id: *link_id,
2318 request_id,
2319 reason: RequestFailure::ResponseTooLarge {
2320 size: response_size,
2321 maximum,
2322 },
2323 });
2324 return actions;
2325 }
2326 }
2327 if is_split {
2328 link.incoming_splits.insert(
2329 original_hash,
2330 IncomingSplitTransfer {
2331 total_segments,
2332 completed_segments: 0,
2333 current_segment_index: segment_index,
2334 current_received_parts: 0,
2335 current_total_parts: receiver.total_parts,
2336 data: Vec::new(),
2337 metadata: None,
2338 is_request: false,
2339 is_response,
2340 },
2341 );
2342 }
2343 link.incoming_resources.push(receiver);
2344 let idx = link.incoming_resources.len() - 1;
2345 let resource_actions = link.incoming_resources[idx].accept(now);
2346 let _ = link;
2347 return self.process_resource_actions(link_id, resource_actions, rng);
2348 }
2349
2350 match strategy {
2351 ResourceStrategy::AcceptNone => {
2352 let reject_actions = {
2354 let mut r = receiver;
2355 r.reject()
2356 };
2357 self.process_resource_actions(link_id, reject_actions, rng)
2358 }
2359 ResourceStrategy::AcceptAll => {
2360 if is_split {
2361 link.incoming_splits.insert(
2362 original_hash,
2363 IncomingSplitTransfer {
2364 total_segments,
2365 completed_segments: 0,
2366 current_segment_index: segment_index,
2367 current_received_parts: 0,
2368 current_total_parts: receiver.total_parts,
2369 data: Vec::new(),
2370 metadata: None,
2371 is_request: false,
2372 is_response,
2373 },
2374 );
2375 }
2376 link.incoming_resources.push(receiver);
2377 let idx = link.incoming_resources.len() - 1;
2378 let resource_actions = link.incoming_resources[idx].accept(now);
2379 let _ = link;
2380 self.process_resource_actions(link_id, resource_actions, rng)
2381 }
2382 ResourceStrategy::AcceptApp => {
2383 link.incoming_resources.push(receiver);
2384 vec![LinkManagerAction::ResourceAcceptQuery {
2386 link_id: *link_id,
2387 resource_hash,
2388 transfer_size,
2389 has_metadata,
2390 }]
2391 }
2392 }
2393 }
2394
2395 pub fn accept_resource(
2397 &mut self,
2398 link_id: &LinkId,
2399 resource_hash: &[u8],
2400 accept: bool,
2401 rng: &mut dyn Rng,
2402 ) -> Vec<LinkManagerAction> {
2403 let link = match self.links.get_mut(link_id) {
2404 Some(l) => l,
2405 None => return Vec::new(),
2406 };
2407
2408 let now = time::now();
2409 let idx = link
2410 .incoming_resources
2411 .iter()
2412 .position(|r| r.resource_hash == resource_hash);
2413 let idx = match idx {
2414 Some(i) => i,
2415 None => return Vec::new(),
2416 };
2417
2418 if accept && link.incoming_resources[idx].flags.split {
2419 if let Some(original_hash) =
2420 Self::resource_hash_key(&link.incoming_resources[idx].original_hash)
2421 {
2422 link.incoming_splits
2423 .entry(original_hash)
2424 .or_insert_with(|| IncomingSplitTransfer {
2425 total_segments: link.incoming_resources[idx].total_segments,
2426 completed_segments: 0,
2427 current_segment_index: link.incoming_resources[idx].segment_index,
2428 current_received_parts: 0,
2429 current_total_parts: link.incoming_resources[idx].total_parts,
2430 data: Vec::new(),
2431 metadata: None,
2432 is_request: link.incoming_resources[idx].flags.is_request,
2433 is_response: link.incoming_resources[idx].flags.is_response,
2434 });
2435 }
2436 }
2437
2438 let resource_actions = if accept {
2439 link.incoming_resources[idx].accept(now)
2440 } else {
2441 link.incoming_resources[idx].reject()
2442 };
2443
2444 let _ = link;
2445 self.process_resource_actions(link_id, resource_actions, rng)
2446 }
2447
2448 fn handle_resource_req(
2450 &mut self,
2451 link_id: &LinkId,
2452 plaintext: &[u8],
2453 rng: &mut dyn Rng,
2454 ) -> Vec<LinkManagerAction> {
2455 let link = match self.links.get_mut(link_id) {
2456 Some(l) => l,
2457 None => return Vec::new(),
2458 };
2459
2460 let now = time::now();
2461 let mut all_actions = Vec::new();
2462 let mut progress_update = None;
2463 for sender in &mut link.outgoing_resources {
2464 if sender.flags.split && sender.status == rns_core::resource::ResourceStatus::Queued {
2465 continue;
2466 }
2467 let before_sent = sender.sent_parts;
2468 let resource_actions = sender.handle_request(plaintext, now);
2469 if !resource_actions.is_empty() {
2470 if sender.sent_parts != before_sent {
2471 if sender.flags.split {
2472 if let Some(split) = link.outgoing_splits.get_mut(&sender.original_hash) {
2473 split.current_segment_index = sender.segment_index;
2474 split.current_sent_parts = sender.sent_parts;
2475 split.current_total_parts = sender.total_parts();
2476 progress_update =
2477 Some(Self::outgoing_split_progress(split, sender.sdu));
2478 }
2479 } else {
2480 progress_update = Some((sender.sent_parts, sender.total_parts()));
2481 }
2482 }
2483 all_actions.extend(resource_actions);
2484 break;
2485 }
2486 }
2487
2488 let _ = link;
2489 let mut out = self.process_resource_actions(link_id, all_actions, rng);
2490 if let Some((received, total)) = progress_update {
2491 out.push(LinkManagerAction::ResourceProgress {
2492 link_id: *link_id,
2493 received,
2494 total,
2495 });
2496 }
2497 out
2498 }
2499
2500 fn handle_resource_hmu(
2502 &mut self,
2503 link_id: &LinkId,
2504 plaintext: &[u8],
2505 rng: &mut dyn Rng,
2506 ) -> Vec<LinkManagerAction> {
2507 let link = match self.links.get_mut(link_id) {
2508 Some(l) => l,
2509 None => return Vec::new(),
2510 };
2511
2512 let now = time::now();
2513 let mut all_actions = Vec::new();
2514 for receiver in &mut link.incoming_resources {
2515 let resource_actions = receiver.handle_hashmap_update(plaintext, now);
2516 if !resource_actions.is_empty() {
2517 all_actions.extend(resource_actions);
2518 break;
2519 }
2520 }
2521
2522 let _ = link;
2523 self.process_resource_actions(link_id, all_actions, rng)
2524 }
2525
2526 fn handle_resource_part(
2528 &mut self,
2529 link_id: &LinkId,
2530 raw_data: &[u8],
2531 rng: &mut dyn Rng,
2532 ) -> Vec<LinkManagerAction> {
2533 let link = match self.links.get_mut(link_id) {
2534 Some(l) => l,
2535 None => return Vec::new(),
2536 };
2537
2538 let now = time::now();
2539 let resource_sdu = Self::resource_sdu_for_link(link);
2540 let mut all_actions = Vec::new();
2541 let mut assemble_idx = None;
2542 let mut assembled_is_request = false;
2543 let mut assembled_is_response = false;
2544 let mut request_request_id = None;
2545 let mut response_request_id = None;
2546
2547 for (idx, receiver) in link.incoming_resources.iter_mut().enumerate() {
2548 if receiver.status >= rns_core::resource::ResourceStatus::Complete {
2549 continue;
2550 }
2551 let resource_actions = receiver.receive_part(raw_data, now);
2552 if !resource_actions.is_empty() {
2553 if receiver.received_count == receiver.total_parts {
2554 assemble_idx = Some(idx);
2555 }
2556 if receiver.flags.split {
2557 if let Some(key) = Self::resource_hash_key(&receiver.original_hash) {
2558 if let Some(split) = link.incoming_splits.get_mut(&key) {
2559 split.current_segment_index = receiver.segment_index;
2560 split.current_received_parts = receiver.received_count;
2561 split.current_total_parts = receiver.total_parts;
2562 let (received, total) =
2563 Self::incoming_split_progress(split, resource_sdu);
2564 for action in resource_actions {
2565 match action {
2566 ResourceAction::ProgressUpdate { .. } => {
2567 all_actions.push(ResourceAction::ProgressUpdate {
2568 received,
2569 total,
2570 });
2571 }
2572 other => all_actions.push(other),
2573 }
2574 }
2575 } else {
2576 all_actions.extend(resource_actions);
2577 }
2578 } else {
2579 all_actions.extend(resource_actions);
2580 }
2581 } else {
2582 all_actions.extend(resource_actions);
2583 }
2584 break;
2585 }
2586 }
2587
2588 if let Some(idx) = assemble_idx {
2589 let split_key = if link.incoming_resources[idx].flags.split {
2590 Self::resource_hash_key(&link.incoming_resources[idx].original_hash)
2591 } else {
2592 None
2593 };
2594 let split_segment_index = link.incoming_resources[idx].segment_index;
2595 let split_segment_total = link.incoming_resources[idx].total_segments;
2596 let split_segment_parts = link.incoming_resources[idx].total_parts;
2597 let split_is_request = link.incoming_resources[idx].flags.is_request;
2598 let split_is_response = link.incoming_resources[idx].flags.is_response;
2599 let resource_request_id =
2600 Self::response_request_id(&link.incoming_resources[idx].request_id);
2601 if split_is_request {
2602 request_request_id = resource_request_id;
2603 } else if split_is_response {
2604 response_request_id = resource_request_id;
2605 }
2606 let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
2607 link.engine.decrypt(ciphertext).map_err(|_| ())
2608 };
2609 let mut assemble_actions =
2610 link.incoming_resources[idx].assemble(&decrypt_fn, &Bzip2Compressor);
2611 assembled_is_request = split_is_request;
2612 assembled_is_response = split_is_response;
2613
2614 if let Some(key) = split_key {
2615 let mut converted_actions = Vec::new();
2616 let mut segment_data = None;
2617 let mut segment_metadata = None;
2618 for action in assemble_actions {
2619 match action {
2620 ResourceAction::DataReceived { data, metadata } => {
2621 segment_data = Some(data);
2622 segment_metadata = metadata;
2623 }
2624 ResourceAction::Completed => {}
2625 other => converted_actions.push(other),
2626 }
2627 }
2628
2629 if let Some(data) = segment_data {
2630 if let Some(split) = link.incoming_splits.get_mut(&key) {
2631 split.data.extend_from_slice(&data);
2632 if segment_metadata.is_some() {
2633 split.metadata = segment_metadata;
2634 }
2635 split.completed_segments = split_segment_index;
2636 split.current_segment_index = split_segment_index;
2637 split.current_received_parts = split_segment_parts;
2638 split.current_total_parts = split_segment_parts;
2639 }
2640
2641 if split_segment_index == split_segment_total {
2642 if let Some(split) = link.incoming_splits.remove(&key) {
2643 assembled_is_request = split.is_request;
2644 assembled_is_response = split.is_response;
2645 converted_actions.push(ResourceAction::DataReceived {
2646 data: split.data,
2647 metadata: split.metadata,
2648 });
2649 converted_actions.push(ResourceAction::Completed);
2650 }
2651 }
2652 }
2653
2654 assemble_actions = converted_actions;
2655 }
2656 all_actions.extend(assemble_actions);
2657 }
2658
2659 let _ = link;
2660 let mut out = self.process_resource_actions(link_id, all_actions, rng);
2661
2662 if assembled_is_request {
2663 let mut converted = Vec::new();
2664 for action in out {
2665 match action {
2666 LinkManagerAction::ResourceReceived { data, .. } => {
2667 if let Some(request_id) = request_request_id {
2668 converted.extend(self.handle_request(link_id, &data, request_id, rng));
2669 }
2670 }
2671 LinkManagerAction::ResourceAcceptQuery { .. } => {
2672 }
2674 other => converted.push(other),
2675 }
2676 }
2677 out = converted;
2678 } else if assembled_is_response {
2679 let mut converted = Vec::new();
2680 for action in out {
2681 match action {
2682 LinkManagerAction::ResourceReceived { data, metadata, .. } => {
2683 converted.extend(self.handle_response(
2684 link_id,
2685 &data,
2686 metadata,
2687 response_request_id,
2688 ));
2689 }
2690 LinkManagerAction::ResourceAcceptQuery { .. } => {
2691 }
2693 other => converted.push(other),
2694 }
2695 }
2696 out = converted;
2697 }
2698
2699 out
2700 }
2701
2702 fn handle_resource_prf(
2704 &mut self,
2705 link_id: &LinkId,
2706 plaintext: &[u8],
2707 rng: &mut dyn Rng,
2708 ) -> Vec<LinkManagerAction> {
2709 let link = match self.links.get_mut(link_id) {
2710 Some(l) => l,
2711 None => return Vec::new(),
2712 };
2713
2714 let now = time::now();
2715 let mut result_actions = Vec::new();
2716 let mut completed_sender = None;
2717 let mut completed_request_id = None;
2718 let mut failed_split = None;
2719 let mut failed_request_id = None;
2720 let proof_hash = plaintext.get(..32);
2721 for sender in &mut link.outgoing_resources {
2722 if proof_hash.is_some_and(|hash| hash != sender.resource_hash.as_slice()) {
2723 continue;
2724 }
2725 let resource_actions = sender.handle_proof(plaintext, now);
2726 if !resource_actions.is_empty() {
2727 if resource_actions
2728 .iter()
2729 .any(|action| matches!(action, ResourceAction::Completed))
2730 {
2731 completed_sender = Some((
2732 sender.original_hash,
2733 sender.segment_index,
2734 sender.total_segments,
2735 sender.total_parts(),
2736 ));
2737 if sender.flags.is_request && sender.segment_index == sender.total_segments {
2738 completed_request_id = Self::response_request_id(&sender.request_id);
2739 }
2740 }
2741 if sender.flags.split
2742 && resource_actions
2743 .iter()
2744 .any(|action| matches!(action, ResourceAction::Failed(_)))
2745 {
2746 failed_split = Some(sender.original_hash);
2747 }
2748 if resource_actions
2749 .iter()
2750 .any(|action| matches!(action, ResourceAction::Failed(_)))
2751 && sender.flags.is_request
2752 {
2753 failed_request_id = Self::response_request_id(&sender.request_id);
2754 }
2755 result_actions.extend(resource_actions);
2756 break;
2757 }
2758 }
2759
2760 let mut actions = Vec::new();
2762 let mut advertise_next = None;
2763 for ra in result_actions {
2764 match ra {
2765 ResourceAction::Completed => {
2766 if let Some((original_hash, segment_index, total_segments, total_parts)) =
2767 completed_sender
2768 {
2769 if total_segments > 1 && segment_index < total_segments {
2770 if let Some(split) = link.outgoing_splits.get_mut(&original_hash) {
2771 split.completed_segments = segment_index;
2772 split.current_segment_index = segment_index;
2773 split.current_sent_parts = total_parts;
2774 split.current_total_parts = total_parts;
2775 if let Some(next) = link.outgoing_resources.iter_mut().find(|s| {
2776 s.flags.split
2777 && s.original_hash == original_hash
2778 && s.segment_index == segment_index + 1
2779 }) {
2780 split.current_segment_index = next.segment_index;
2781 split.current_sent_parts = 0;
2782 split.current_total_parts = next.total_parts();
2783 advertise_next = Some(next.advertise(now));
2784 }
2785 }
2786 } else {
2787 link.outgoing_splits.remove(&original_hash);
2788 actions
2789 .push(LinkManagerAction::ResourceCompleted { link_id: *link_id });
2790 }
2791 } else {
2792 actions.push(LinkManagerAction::ResourceCompleted { link_id: *link_id });
2793 }
2794 }
2795 ResourceAction::Failed(e) => {
2796 if let Some(original_hash) = failed_split {
2797 link.outgoing_splits.remove(&original_hash);
2798 }
2799 actions.push(LinkManagerAction::ResourceFailed {
2800 link_id: *link_id,
2801 error: format!("{}", e),
2802 });
2803 }
2804 _ => {}
2805 }
2806 }
2807
2808 if let Some(request_id) = completed_request_id {
2809 let deadline = Self::request_response_deadline(link, now);
2810 if let Some(entry) = link.pending_requests.get_mut(&request_id) {
2811 entry.deadline = Some(deadline);
2812 }
2813 }
2814 if let Some(request_id) = failed_request_id {
2815 link.pending_requests.remove(&request_id);
2816 }
2817
2818 link.outgoing_resources
2820 .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
2821
2822 let _ = link;
2823 if let Some(next_actions) = advertise_next {
2824 actions.extend(self.process_resource_actions(link_id, next_actions, rng));
2825 }
2826
2827 actions
2828 }
2829
2830 fn handle_resource_icl(&mut self, link_id: &LinkId) -> Vec<LinkManagerAction> {
2832 let link = match self.links.get_mut(link_id) {
2833 Some(l) => l,
2834 None => return Vec::new(),
2835 };
2836
2837 let mut actions = Vec::new();
2838 for receiver in &mut link.incoming_resources {
2839 let ra = receiver.handle_cancel();
2840 for a in ra {
2841 if let ResourceAction::Failed(ref e) = a {
2842 actions.push(LinkManagerAction::ResourceFailed {
2843 link_id: *link_id,
2844 error: format!("{}", e),
2845 });
2846 }
2847 }
2848 }
2849 link.incoming_resources
2850 .retain(|r| r.status < rns_core::resource::ResourceStatus::Complete);
2851 link.incoming_splits.clear();
2852 actions
2853 }
2854
2855 fn handle_resource_rcl(&mut self, link_id: &LinkId) -> Vec<LinkManagerAction> {
2857 let link = match self.links.get_mut(link_id) {
2858 Some(l) => l,
2859 None => return Vec::new(),
2860 };
2861
2862 let request_ids: Vec<[u8; 16]> = link
2863 .outgoing_resources
2864 .iter()
2865 .filter(|sender| sender.flags.is_request)
2866 .filter_map(|sender| Self::response_request_id(&sender.request_id))
2867 .collect();
2868 let mut actions = Vec::new();
2869 for sender in &mut link.outgoing_resources {
2870 let ra = sender.handle_reject();
2871 for a in ra {
2872 if let ResourceAction::Failed(ref e) = a {
2873 actions.push(LinkManagerAction::ResourceFailed {
2874 link_id: *link_id,
2875 error: format!("{}", e),
2876 });
2877 }
2878 }
2879 }
2880 link.outgoing_resources
2881 .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
2882 link.outgoing_splits.clear();
2883 for request_id in request_ids {
2884 link.pending_requests.remove(&request_id);
2885 }
2886 actions
2887 }
2888
2889 fn resource_link_is_active(&self, link_id: &LinkId) -> bool {
2890 self.links
2891 .get(link_id)
2892 .is_some_and(|link| link.engine.state() == LinkState::Active)
2893 }
2894
2895 fn abort_resources_on_inactive_link(&mut self, link_id: &LinkId) {
2896 let Some(link) = self.links.get_mut(link_id) else {
2897 return;
2898 };
2899 for sender in &mut link.outgoing_resources {
2900 let _ = sender.cancel();
2901 }
2902 for receiver in &mut link.incoming_resources {
2903 let _ = receiver.cancel();
2904 }
2905 link.outgoing_resources.clear();
2906 link.incoming_resources.clear();
2907 link.outgoing_splits.clear();
2908 link.incoming_splits.clear();
2909 link.pending_requests.clear();
2910 }
2911
2912 fn process_resource_actions(
2914 &mut self,
2915 link_id: &LinkId,
2916 actions: Vec<ResourceAction>,
2917 rng: &mut dyn Rng,
2918 ) -> Vec<LinkManagerAction> {
2919 let mut result = Vec::new();
2920 for action in actions {
2921 let requires_active_link = matches!(
2922 &action,
2923 ResourceAction::SendAdvertisement(_)
2924 | ResourceAction::SendPart(_)
2925 | ResourceAction::SendRequest(_)
2926 | ResourceAction::SendHmu(_)
2927 | ResourceAction::SendProof(_)
2928 );
2929 if requires_active_link && !self.resource_link_is_active(link_id) {
2930 self.abort_resources_on_inactive_link(link_id);
2931 continue;
2932 }
2933
2934 match action {
2935 ResourceAction::SendAdvertisement(data) => {
2936 let encrypted = self
2938 .links
2939 .get(link_id)
2940 .and_then(|link| link.engine.encrypt(&data, rng).ok());
2941 if let Some(encrypted) = encrypted {
2942 result.extend(self.build_link_packet(
2943 link_id,
2944 constants::CONTEXT_RESOURCE_ADV,
2945 &encrypted,
2946 ));
2947 }
2948 }
2949 ResourceAction::SendPart(data) => {
2950 result.extend(self.build_link_packet(
2952 link_id,
2953 constants::CONTEXT_RESOURCE,
2954 &data,
2955 ));
2956 }
2957 ResourceAction::SendRequest(data) => {
2958 let encrypted = self
2959 .links
2960 .get(link_id)
2961 .and_then(|link| link.engine.encrypt(&data, rng).ok());
2962 if let Some(encrypted) = encrypted {
2963 result.extend(self.build_link_packet(
2964 link_id,
2965 constants::CONTEXT_RESOURCE_REQ,
2966 &encrypted,
2967 ));
2968 }
2969 }
2970 ResourceAction::SendHmu(data) => {
2971 let encrypted = self
2972 .links
2973 .get(link_id)
2974 .and_then(|link| link.engine.encrypt(&data, rng).ok());
2975 if let Some(encrypted) = encrypted {
2976 result.extend(self.build_link_packet(
2977 link_id,
2978 constants::CONTEXT_RESOURCE_HMU,
2979 &encrypted,
2980 ));
2981 }
2982 }
2983 ResourceAction::SendProof(data) => {
2984 let flags = PacketFlags {
2985 header_type: constants::HEADER_1,
2986 context_flag: constants::FLAG_UNSET,
2987 transport_type: constants::TRANSPORT_BROADCAST,
2988 destination_type: constants::DESTINATION_LINK,
2989 packet_type: constants::PACKET_TYPE_PROOF,
2990 };
2991 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
2992 flags,
2993 0,
2994 link_id,
2995 None,
2996 constants::CONTEXT_RESOURCE_PRF,
2997 &data,
2998 ) {
2999 result.push(LinkManagerAction::SendPacket {
3000 raw,
3001 dest_type: constants::DESTINATION_LINK,
3002 attached_interface: None,
3003 });
3004 }
3005 }
3006 ResourceAction::SendCancelInitiator(data) => {
3007 let encrypted = self
3008 .links
3009 .get(link_id)
3010 .filter(|link| link.engine.state() == LinkState::Active)
3011 .and_then(|link| link.engine.encrypt(&data, rng).ok());
3012 if let Some(encrypted) = encrypted {
3013 result.extend(self.build_link_packet(
3014 link_id,
3015 constants::CONTEXT_RESOURCE_ICL,
3016 &encrypted,
3017 ));
3018 }
3019 }
3020 ResourceAction::SendCancelReceiver(data) => {
3021 let encrypted = self
3022 .links
3023 .get(link_id)
3024 .filter(|link| link.engine.state() == LinkState::Active)
3025 .and_then(|link| link.engine.encrypt(&data, rng).ok());
3026 if let Some(encrypted) = encrypted {
3027 result.extend(self.build_link_packet(
3028 link_id,
3029 constants::CONTEXT_RESOURCE_RCL,
3030 &encrypted,
3031 ));
3032 }
3033 }
3034 ResourceAction::DataReceived { data, metadata } => {
3035 result.push(LinkManagerAction::ResourceReceived {
3036 link_id: *link_id,
3037 data,
3038 metadata,
3039 });
3040 }
3041 ResourceAction::Completed => {
3042 result.push(LinkManagerAction::ResourceCompleted { link_id: *link_id });
3043 }
3044 ResourceAction::Failed(e) => {
3045 result.push(LinkManagerAction::ResourceFailed {
3046 link_id: *link_id,
3047 error: format!("{}", e),
3048 });
3049 }
3050 ResourceAction::TeardownLink => {
3051 let teardown_actions = match self.links.get_mut(link_id) {
3052 Some(link) => link.engine.handle_teardown(),
3053 None => Vec::new(),
3054 };
3055 result.extend(self.process_link_actions(link_id, &teardown_actions));
3056 }
3057 ResourceAction::ProgressUpdate { received, total } => {
3058 result.push(LinkManagerAction::ResourceProgress {
3059 link_id: *link_id,
3060 received,
3061 total,
3062 });
3063 }
3064 }
3065 }
3066 result
3067 }
3068
3069 fn build_link_packet(
3071 &self,
3072 link_id: &LinkId,
3073 context: u8,
3074 data: &[u8],
3075 ) -> Vec<LinkManagerAction> {
3076 let flags = PacketFlags {
3077 header_type: constants::HEADER_1,
3078 context_flag: constants::FLAG_UNSET,
3079 transport_type: constants::TRANSPORT_BROADCAST,
3080 destination_type: constants::DESTINATION_LINK,
3081 packet_type: constants::PACKET_TYPE_DATA,
3082 };
3083 let mut actions = Vec::new();
3084 let max_mtu = self
3085 .links
3086 .get(link_id)
3087 .map(|l| l.engine.mtu() as usize)
3088 .unwrap_or(constants::MTU);
3089 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
3090 flags, 0, link_id, None, context, data, max_mtu,
3091 ) {
3092 actions.push(LinkManagerAction::SendPacket {
3093 raw,
3094 dest_type: constants::DESTINATION_LINK,
3095 attached_interface: None,
3096 });
3097 }
3098 actions
3099 }
3100
3101 pub fn send_resource(
3103 &mut self,
3104 link_id: &LinkId,
3105 data: &[u8],
3106 metadata: Option<&[u8]>,
3107 rng: &mut dyn Rng,
3108 ) -> Vec<LinkManagerAction> {
3109 self.send_resource_with_auto_compress(link_id, data, metadata, true, rng)
3110 }
3111
3112 pub fn send_resource_with_auto_compress(
3114 &mut self,
3115 link_id: &LinkId,
3116 data: &[u8],
3117 metadata: Option<&[u8]>,
3118 auto_compress: bool,
3119 rng: &mut dyn Rng,
3120 ) -> Vec<LinkManagerAction> {
3121 let link = match self.links.get_mut(link_id) {
3122 Some(l) => l,
3123 None => return Vec::new(),
3124 };
3125
3126 if link.engine.state() != LinkState::Active {
3127 return Vec::new();
3128 }
3129
3130 let now = time::now();
3131
3132 let senders = match Self::build_resource_senders(
3133 link,
3134 data,
3135 metadata,
3136 auto_compress,
3137 false, None, rng,
3140 now,
3141 ) {
3142 Ok(s) => s,
3143 Err(e) => {
3144 log::debug!("Failed to create ResourceSender: {}", e);
3145 return Vec::new();
3146 }
3147 };
3148
3149 let adv_actions = Self::start_resource_senders(link, senders, now);
3150
3151 let _ = link;
3152 self.process_resource_actions(link_id, adv_actions, rng)
3153 }
3154
3155 pub fn set_resource_strategy(&mut self, link_id: &LinkId, strategy: ResourceStrategy) {
3157 if let Some(link) = self.links.get_mut(link_id) {
3158 link.resource_strategy = strategy;
3159 }
3160 }
3161
3162 pub fn flush_channel_tx(&mut self, link_id: &LinkId) {
3165 if let Some(link) = self.links.get_mut(link_id) {
3166 if let Some(ref mut channel) = link.channel {
3167 channel.flush_tx();
3168 }
3169 }
3170 }
3171
3172 pub fn send_channel_message(
3174 &mut self,
3175 link_id: &LinkId,
3176 msgtype: u16,
3177 payload: &[u8],
3178 rng: &mut dyn Rng,
3179 ) -> Result<Vec<LinkManagerAction>, String> {
3180 let link = match self.links.get_mut(link_id) {
3181 Some(l) => l,
3182 None => return Err("unknown link".to_string()),
3183 };
3184
3185 let channel = match link.channel {
3186 Some(ref mut ch) => ch,
3187 None => return Err("link has no active channel".to_string()),
3188 };
3189
3190 let link_mdu = link.engine.mdu();
3191 let now = time::now();
3192 let chan_actions = match channel.send(msgtype, payload, now, link_mdu) {
3193 Ok(a) => {
3194 link.channel_send_ok += 1;
3195 a
3196 }
3197 Err(e) => {
3198 log::debug!("Channel send failed: {:?}", e);
3199 match e {
3200 rns_core::channel::ChannelError::NotReady => link.channel_send_not_ready += 1,
3201 rns_core::channel::ChannelError::MessageTooBig => {
3202 link.channel_send_too_big += 1;
3203 }
3204 rns_core::channel::ChannelError::InvalidEnvelope => {
3205 link.channel_send_other_error += 1;
3206 }
3207 }
3208 return Err(e.to_string());
3209 }
3210 };
3211
3212 let _ = link;
3213 Ok(self.process_channel_actions(link_id, chan_actions, rng))
3214 }
3215
3216 pub fn tick(&mut self, rng: &mut dyn Rng) -> Vec<LinkManagerAction> {
3218 let now = time::now();
3219 let mut all_actions = Vec::new();
3220
3221 let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
3223
3224 for link_id in &link_ids {
3225 let link = match self.links.get_mut(link_id) {
3226 Some(l) => l,
3227 None => continue,
3228 };
3229
3230 let tick_actions = link.engine.tick(now);
3232 all_actions.extend(self.process_link_actions(link_id, &tick_actions));
3233
3234 let link = match self.links.get_mut(link_id) {
3236 Some(l) => l,
3237 None => continue,
3238 };
3239 if link.engine.needs_keepalive(now) {
3240 let flags = PacketFlags {
3243 header_type: constants::HEADER_1,
3244 context_flag: constants::FLAG_UNSET,
3245 transport_type: constants::TRANSPORT_BROADCAST,
3246 destination_type: constants::DESTINATION_LINK,
3247 packet_type: constants::PACKET_TYPE_DATA,
3248 };
3249 if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
3250 flags,
3251 0,
3252 link_id,
3253 None,
3254 constants::CONTEXT_KEEPALIVE,
3255 &[0xff],
3256 ) {
3257 all_actions.push(LinkManagerAction::SendPacket {
3258 raw,
3259 dest_type: constants::DESTINATION_LINK,
3260 attached_interface: None,
3261 });
3262 link.engine.record_outbound(now, true);
3263 }
3264 }
3265
3266 if let Some(channel) = link.channel.as_mut() {
3267 let chan_actions = channel.tick(now);
3268 let _ = channel;
3269 let _ = link;
3270 all_actions.extend(self.process_channel_actions(link_id, chan_actions, rng));
3271 }
3272 }
3273
3274 for link_id in &link_ids {
3276 let link = match self.links.get_mut(link_id) {
3277 Some(l) => l,
3278 None => continue,
3279 };
3280
3281 let mut sender_actions = Vec::new();
3283 for sender in &mut link.outgoing_resources {
3284 sender_actions.extend(sender.tick(now));
3285 }
3286
3287 let mut receiver_actions = Vec::new();
3289 for receiver in &mut link.incoming_resources {
3290 let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
3291 link.engine.decrypt(ciphertext).map_err(|_| ())
3292 };
3293 receiver_actions.extend(receiver.tick(now, &decrypt_fn, &Bzip2Compressor));
3294 }
3295
3296 let failed_request_ids: Vec<[u8; 16]> = link
3297 .outgoing_resources
3298 .iter()
3299 .filter(|sender| {
3300 sender.flags.is_request
3301 && sender.status >= rns_core::resource::ResourceStatus::Failed
3302 })
3303 .filter_map(|sender| Self::response_request_id(&sender.request_id))
3304 .collect();
3305 for request_id in failed_request_ids {
3306 link.pending_requests.remove(&request_id);
3307 }
3308 link.pending_requests
3309 .retain(|_, request| request.deadline.is_none_or(|expires_at| now <= expires_at));
3310
3311 link.outgoing_resources
3313 .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
3314 link.incoming_resources
3315 .retain(|r| r.status < rns_core::resource::ResourceStatus::Assembling);
3316 let active_split_hashes: Vec<[u8; 32]> = link
3317 .outgoing_resources
3318 .iter()
3319 .filter(|s| s.flags.split)
3320 .map(|s| s.original_hash)
3321 .collect();
3322 link.outgoing_splits.retain(|original_hash, split| {
3323 split.completed_segments < split.total_segments
3324 && active_split_hashes.contains(original_hash)
3325 });
3326
3327 let _ = link;
3328 all_actions.extend(self.process_resource_actions(link_id, sender_actions, rng));
3329 all_actions.extend(self.process_resource_actions(link_id, receiver_actions, rng));
3330 }
3331
3332 let closed: Vec<LinkId> = self
3334 .links
3335 .iter()
3336 .filter(|(_, l)| l.engine.state() == LinkState::Closed)
3337 .map(|(id, _)| *id)
3338 .collect();
3339 for id in closed {
3340 self.links.remove(&id);
3341 all_actions.push(LinkManagerAction::DeregisterLinkDest { link_id: id });
3342 }
3343
3344 all_actions
3345 }
3346
3347 pub fn is_link_destination(&self, dest_hash: &[u8; 16]) -> bool {
3349 self.links.contains_key(dest_hash) || self.link_destinations.contains_key(dest_hash)
3350 }
3351
3352 pub fn link_state(&self, link_id: &LinkId) -> Option<LinkState> {
3354 self.links.get(link_id).map(|l| l.engine.state())
3355 }
3356
3357 pub fn link_rtt(&self, link_id: &LinkId) -> Option<f64> {
3359 self.links.get(link_id).and_then(|l| l.engine.rtt())
3360 }
3361
3362 pub fn set_link_rtt(&mut self, link_id: &LinkId, rtt: f64) {
3364 if let Some(link) = self.links.get_mut(link_id) {
3365 link.engine.set_rtt(rtt);
3366 }
3367 }
3368
3369 pub fn record_link_inbound(&mut self, link_id: &LinkId) {
3371 if let Some(link) = self.links.get_mut(link_id) {
3372 link.engine.record_inbound(time::now());
3373 }
3374 }
3375
3376 pub fn set_link_mtu(&mut self, link_id: &LinkId, mtu: u32) {
3378 if let Some(link) = self.links.get_mut(link_id) {
3379 link.engine.set_mtu(mtu);
3380 }
3381 }
3382
3383 pub fn link_count(&self) -> usize {
3385 self.links.len()
3386 }
3387
3388 pub fn resource_transfer_count(&self) -> usize {
3390 self.links
3391 .values()
3392 .map(|managed| {
3393 managed
3394 .incoming_resources
3395 .iter()
3396 .filter(|resource| !resource.flags.split)
3397 .count()
3398 + managed.incoming_splits.len()
3399 + managed
3400 .outgoing_resources
3401 .iter()
3402 .filter(|resource| !resource.flags.split)
3403 .count()
3404 + managed.outgoing_splits.len()
3405 })
3406 .sum()
3407 }
3408
3409 pub fn cancel_all_resources(&mut self, rng: &mut dyn Rng) -> Vec<LinkManagerAction> {
3411 let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
3412 let mut all_actions = Vec::new();
3413
3414 for link_id in &link_ids {
3415 let link = match self.links.get_mut(link_id) {
3416 Some(l) => l,
3417 None => continue,
3418 };
3419
3420 let mut sender_actions = Vec::new();
3421 for sender in &mut link.outgoing_resources {
3422 sender_actions.extend(sender.cancel());
3423 }
3424
3425 let mut receiver_actions = Vec::new();
3426 for receiver in &mut link.incoming_resources {
3427 receiver_actions.extend(receiver.cancel());
3428 }
3429
3430 link.outgoing_resources
3431 .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
3432 link.incoming_resources
3433 .retain(|r| r.status < rns_core::resource::ResourceStatus::Assembling);
3434 link.outgoing_splits.clear();
3435 link.incoming_splits.clear();
3436
3437 let _ = link;
3438 all_actions.extend(self.process_resource_actions(link_id, sender_actions, rng));
3439 all_actions.extend(self.process_resource_actions(link_id, receiver_actions, rng));
3440 }
3441
3442 all_actions
3443 }
3444
3445 pub fn link_entries(&self) -> Vec<crate::event::LinkInfoEntry> {
3447 self.links
3448 .iter()
3449 .map(|(link_id, managed)| {
3450 let state = match managed.engine.state() {
3451 LinkState::Pending => "pending",
3452 LinkState::Handshake => "handshake",
3453 LinkState::Active => "active",
3454 LinkState::Stale => "stale",
3455 LinkState::Closed => "closed",
3456 };
3457 crate::event::LinkInfoEntry {
3458 link_id: *link_id,
3459 state: state.to_string(),
3460 is_initiator: managed.engine.is_initiator(),
3461 dest_hash: managed.dest_hash,
3462 remote_identity: managed.remote_identity.as_ref().map(|(h, _)| *h),
3463 rtt: managed.engine.rtt(),
3464 expected_hops: managed.engine.expected_hops(),
3465 tx_packets: managed.engine.tx_packets(),
3466 rx_packets: managed.engine.rx_packets(),
3467 tx_bytes: managed.engine.tx_bytes(),
3468 rx_bytes: managed.engine.rx_bytes(),
3469 channel_window: managed.channel.as_ref().map(|c| c.window()),
3470 channel_outstanding: managed.channel.as_ref().map(|c| c.outstanding()),
3471 pending_channel_packets: managed.pending_channel_packets.len(),
3472 channel_send_ok: managed.channel_send_ok,
3473 channel_send_not_ready: managed.channel_send_not_ready,
3474 channel_send_too_big: managed.channel_send_too_big,
3475 channel_send_other_error: managed.channel_send_other_error,
3476 channel_messages_received: managed.channel_messages_received,
3477 channel_proofs_sent: managed.channel_proofs_sent,
3478 channel_proofs_received: managed.channel_proofs_received,
3479 }
3480 })
3481 .collect()
3482 }
3483
3484 pub fn record_outbound_packet(&mut self, packet: &RawPacket) {
3488 if packet.flags.destination_type == constants::DESTINATION_LINK {
3489 if let Some(link) = self.links.get_mut(&packet.destination_hash) {
3490 link.engine.record_outbound_traffic(packet.data.len());
3491 }
3492 }
3493 }
3494
3495 pub fn resource_entries(&self) -> Vec<crate::event::ResourceInfoEntry> {
3497 let mut entries = Vec::new();
3498 for (link_id, managed) in &self.links {
3499 let resource_sdu = Self::resource_sdu_for_link(managed);
3500 for split in managed.incoming_splits.values() {
3501 let (received, total) = Self::incoming_split_progress(split, resource_sdu);
3502 entries.push(crate::event::ResourceInfoEntry {
3503 link_id: *link_id,
3504 direction: "incoming".to_string(),
3505 total_parts: total,
3506 transferred_parts: received,
3507 complete: received >= total && total > 0,
3508 });
3509 }
3510 for recv in &managed.incoming_resources {
3511 if recv.flags.split {
3512 continue;
3513 }
3514 let (received, total) = recv.progress();
3515 entries.push(crate::event::ResourceInfoEntry {
3516 link_id: *link_id,
3517 direction: "incoming".to_string(),
3518 total_parts: total,
3519 transferred_parts: received,
3520 complete: received >= total && total > 0,
3521 });
3522 }
3523 for split in managed.outgoing_splits.values() {
3524 let (sent, total) = Self::outgoing_split_progress(split, resource_sdu);
3525 entries.push(crate::event::ResourceInfoEntry {
3526 link_id: *link_id,
3527 direction: "outgoing".to_string(),
3528 total_parts: total,
3529 transferred_parts: sent,
3530 complete: sent >= total && total > 0,
3531 });
3532 }
3533 for send in &managed.outgoing_resources {
3534 if send.flags.split {
3535 continue;
3536 }
3537 let total = send.total_parts();
3538 let sent = send.sent_parts;
3539 entries.push(crate::event::ResourceInfoEntry {
3540 link_id: *link_id,
3541 direction: "outgoing".to_string(),
3542 total_parts: total,
3543 transferred_parts: sent,
3544 complete: sent >= total && total > 0,
3545 });
3546 }
3547 }
3548 entries
3549 }
3550
3551 fn process_link_actions(
3553 &self,
3554 link_id: &LinkId,
3555 actions: &[LinkAction],
3556 ) -> Vec<LinkManagerAction> {
3557 let mut result = Vec::new();
3558 for action in actions {
3559 match action {
3560 LinkAction::StateChanged {
3561 new_state, reason, ..
3562 } => match new_state {
3563 LinkState::Closed => {
3564 result.push(LinkManagerAction::LinkClosed {
3565 link_id: *link_id,
3566 reason: *reason,
3567 });
3568 }
3569 _ => {}
3570 },
3571 LinkAction::LinkEstablished {
3572 rtt, is_initiator, ..
3573 } => {
3574 let dest_hash = self
3575 .links
3576 .get(link_id)
3577 .map(|l| l.dest_hash)
3578 .unwrap_or([0u8; 16]);
3579 let hops = self
3580 .links
3581 .get(link_id)
3582 .map(|link| link.engine.expected_hops())
3583 .unwrap_or(rns_core::constants::PATHFINDER_M);
3584 let rebalanced_at = self
3585 .links
3586 .get(link_id)
3587 .and_then(|link| link.engine.rebalanced_at());
3588 result.push(LinkManagerAction::LinkEstablished {
3589 link_id: *link_id,
3590 dest_hash,
3591 hops,
3592 rebalanced_at,
3593 rtt: *rtt,
3594 is_initiator: *is_initiator,
3595 });
3596 }
3597 LinkAction::RemoteIdentified {
3598 identity_hash,
3599 public_key,
3600 ..
3601 } => {
3602 result.push(LinkManagerAction::RemoteIdentified {
3603 link_id: *link_id,
3604 identity_hash: *identity_hash,
3605 public_key: *public_key,
3606 });
3607 }
3608 LinkAction::DataReceived { .. } => {
3609 }
3611 }
3612 }
3613 result
3614 }
3615
3616 fn process_channel_actions(
3618 &mut self,
3619 link_id: &LinkId,
3620 actions: Vec<rns_core::channel::ChannelAction>,
3621 rng: &mut dyn Rng,
3622 ) -> Vec<LinkManagerAction> {
3623 let mut result = Vec::new();
3624 for action in actions {
3625 match action {
3626 rns_core::channel::ChannelAction::SendOnLink { raw, sequence } => {
3627 let encrypted = match self.links.get(link_id) {
3631 Some(link) => match link.engine.encrypt(&raw, rng) {
3632 Ok(encrypted) => encrypted,
3633 Err(_) => {
3634 if let Some(link_mut) = self.links.get_mut(link_id) {
3635 if let Some(channel) = link_mut.channel.as_mut() {
3636 channel.cancel_send(sequence);
3637 }
3638 }
3639 continue;
3640 }
3641 },
3642 None => continue,
3643 };
3644 let flags = PacketFlags {
3645 header_type: constants::HEADER_1,
3646 context_flag: constants::FLAG_UNSET,
3647 transport_type: constants::TRANSPORT_BROADCAST,
3648 destination_type: constants::DESTINATION_LINK,
3649 packet_type: constants::PACKET_TYPE_DATA,
3650 };
3651 match RawPacket::pack_raw_with_hash(
3652 flags,
3653 0,
3654 link_id,
3655 None,
3656 constants::CONTEXT_CHANNEL,
3657 &encrypted,
3658 ) {
3659 Ok((raw_bytes, packet_hash)) => {
3660 if let Some(link_mut) = self.links.get_mut(link_id) {
3661 link_mut
3662 .pending_channel_packets
3663 .insert(packet_hash, sequence);
3664 }
3665 result.push(LinkManagerAction::SendPacket {
3666 raw: raw_bytes,
3667 dest_type: constants::DESTINATION_LINK,
3668 attached_interface: None,
3669 });
3670 }
3671 Err(_) => {
3672 if let Some(link_mut) = self.links.get_mut(link_id) {
3673 if let Some(channel) = link_mut.channel.as_mut() {
3674 channel.cancel_send(sequence);
3675 }
3676 }
3677 }
3678 }
3679 }
3680 rns_core::channel::ChannelAction::MessageReceived {
3681 msgtype, payload, ..
3682 } => {
3683 result.push(LinkManagerAction::ChannelMessageReceived {
3684 link_id: *link_id,
3685 msgtype,
3686 payload,
3687 });
3688 }
3689 rns_core::channel::ChannelAction::TeardownLink => {
3690 result.push(LinkManagerAction::LinkClosed {
3691 link_id: *link_id,
3692 reason: Some(TeardownReason::Timeout),
3693 });
3694 }
3695 }
3696 }
3697 result
3698 }
3699}
3700
3701impl Default for LinkManager {
3702 fn default() -> Self {
3703 Self::new()
3704 }
3705}
3706
3707fn compute_path_hash(path: &str) -> [u8; 16] {
3710 let full = rns_core::hash::full_hash(path.as_bytes());
3711 let mut result = [0u8; 16];
3712 result.copy_from_slice(&full[..16]);
3713 result
3714}
3715
3716#[cfg(test)]
3717mod tests {
3718 use super::*;
3719 use rns_crypto::identity::Identity;
3720 use rns_crypto::{FixedRng, OsRng};
3721
3722 fn make_rng(seed: u8) -> FixedRng {
3723 FixedRng::new(&[seed; 128])
3724 }
3725
3726 fn make_dest_keys(rng: &mut dyn Rng) -> (Ed25519PrivateKey, [u8; 32]) {
3727 let sig_prv = Ed25519PrivateKey::generate(rng);
3728 let sig_pub_bytes = sig_prv.public_key().public_bytes();
3729 (sig_prv, sig_pub_bytes)
3730 }
3731
3732 #[test]
3733 fn test_register_link_destination() {
3734 let mut mgr = LinkManager::new();
3735 let mut rng = make_rng(0x01);
3736 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3737 let dest_hash = [0xDD; 16];
3738
3739 mgr.register_link_destination(
3740 dest_hash,
3741 sig_prv,
3742 sig_pub_bytes,
3743 ResourceStrategy::AcceptNone,
3744 );
3745 assert!(mgr.is_link_destination(&dest_hash));
3746
3747 mgr.deregister_link_destination(&dest_hash);
3748 assert!(!mgr.is_link_destination(&dest_hash));
3749 }
3750
3751 #[test]
3752 fn test_create_link() {
3753 let mut mgr = LinkManager::new();
3754 let mut rng = OsRng;
3755 let dest_hash = [0xDD; 16];
3756
3757 let sig_pub_bytes = [0xAA; 32]; let (link_id, actions) = mgr.create_link(
3759 &dest_hash,
3760 &sig_pub_bytes,
3761 1,
3762 constants::MTU as u32,
3763 &mut rng,
3764 );
3765 assert_ne!(link_id, [0u8; 16]);
3766 assert_eq!(actions.len(), 2);
3768 assert!(matches!(
3769 actions[0],
3770 LinkManagerAction::RegisterLinkDest { .. }
3771 ));
3772 assert!(matches!(actions[1], LinkManagerAction::SendPacket { .. }));
3773
3774 assert_eq!(mgr.link_state(&link_id), Some(LinkState::Pending));
3776 }
3777
3778 #[test]
3779 fn test_full_handshake_via_manager() {
3780 let mut rng = OsRng;
3781 let dest_hash = [0xDD; 16];
3782
3783 let mut responder_mgr = LinkManager::new();
3785 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3786 responder_mgr.register_link_destination(
3787 dest_hash,
3788 sig_prv,
3789 sig_pub_bytes,
3790 ResourceStrategy::AcceptNone,
3791 );
3792
3793 let mut initiator_mgr = LinkManager::new();
3795
3796 let (link_id, init_actions) = initiator_mgr.create_link(
3798 &dest_hash,
3799 &sig_pub_bytes,
3800 1,
3801 constants::MTU as u32,
3802 &mut rng,
3803 );
3804 assert_eq!(init_actions.len(), 2);
3805
3806 let linkrequest_raw = match &init_actions[1] {
3808 LinkManagerAction::SendPacket { raw, .. } => raw.clone(),
3809 _ => panic!("Expected SendPacket"),
3810 };
3811
3812 let lr_packet = RawPacket::unpack(&linkrequest_raw).unwrap();
3814
3815 let resp_actions = responder_mgr.handle_local_delivery(
3817 lr_packet.destination_hash,
3818 &linkrequest_raw,
3819 lr_packet.packet_hash,
3820 rns_core::transport::types::InterfaceId(0),
3821 &mut rng,
3822 );
3823 assert!(resp_actions.len() >= 2);
3825 assert!(matches!(
3826 resp_actions[0],
3827 LinkManagerAction::RegisterLinkDest { .. }
3828 ));
3829
3830 let lrproof_raw = match &resp_actions[1] {
3832 LinkManagerAction::SendPacket { raw, .. } => raw.clone(),
3833 _ => panic!("Expected SendPacket for LRPROOF"),
3834 };
3835
3836 let lrproof_packet = RawPacket::unpack(&lrproof_raw).unwrap();
3838 let init_actions2 = initiator_mgr.handle_local_delivery(
3839 lrproof_packet.destination_hash,
3840 &lrproof_raw,
3841 lrproof_packet.packet_hash,
3842 rns_core::transport::types::InterfaceId(0),
3843 &mut rng,
3844 );
3845
3846 let has_established = init_actions2
3848 .iter()
3849 .any(|a| matches!(a, LinkManagerAction::LinkEstablished { .. }));
3850 assert!(has_established, "Initiator should emit LinkEstablished");
3851
3852 let lrrtt_raw = init_actions2
3854 .iter()
3855 .find_map(|a| match a {
3856 LinkManagerAction::SendPacket { raw, .. } => Some(raw.clone()),
3857 _ => None,
3858 })
3859 .expect("Should have LRRTT SendPacket");
3860
3861 let lrrtt_packet = RawPacket::unpack(&lrrtt_raw).unwrap();
3863 let resp_link_id = lrrtt_packet.destination_hash;
3864 let resp_actions2 = responder_mgr.handle_local_delivery(
3865 resp_link_id,
3866 &lrrtt_raw,
3867 lrrtt_packet.packet_hash,
3868 rns_core::transport::types::InterfaceId(0),
3869 &mut rng,
3870 );
3871
3872 let has_established = resp_actions2
3873 .iter()
3874 .any(|a| matches!(a, LinkManagerAction::LinkEstablished { .. }));
3875 assert!(has_established, "Responder should emit LinkEstablished");
3876
3877 assert_eq!(initiator_mgr.link_state(&link_id), Some(LinkState::Active));
3879 assert_eq!(responder_mgr.link_state(&link_id), Some(LinkState::Active));
3880
3881 assert!(initiator_mgr.link_rtt(&link_id).is_some());
3883 assert!(responder_mgr.link_rtt(&link_id).is_some());
3884 }
3885
3886 #[test]
3887 fn test_encrypted_data_exchange() {
3888 let mut rng = OsRng;
3889 let dest_hash = [0xDD; 16];
3890 let mut resp_mgr = LinkManager::new();
3891 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3892 resp_mgr.register_link_destination(
3893 dest_hash,
3894 sig_prv,
3895 sig_pub_bytes,
3896 ResourceStrategy::AcceptNone,
3897 );
3898 let mut init_mgr = LinkManager::new();
3899
3900 let (link_id, init_actions) = init_mgr.create_link(
3902 &dest_hash,
3903 &sig_pub_bytes,
3904 1,
3905 constants::MTU as u32,
3906 &mut rng,
3907 );
3908 let lr_raw = extract_send_packet(&init_actions);
3909 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
3910 let resp_actions = resp_mgr.handle_local_delivery(
3911 lr_pkt.destination_hash,
3912 &lr_raw,
3913 lr_pkt.packet_hash,
3914 rns_core::transport::types::InterfaceId(0),
3915 &mut rng,
3916 );
3917 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
3918 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
3919 let init_actions2 = init_mgr.handle_local_delivery(
3920 lrproof_pkt.destination_hash,
3921 &lrproof_raw,
3922 lrproof_pkt.packet_hash,
3923 rns_core::transport::types::InterfaceId(0),
3924 &mut rng,
3925 );
3926 let lrrtt_raw = extract_any_send_packet(&init_actions2);
3927 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
3928 resp_mgr.handle_local_delivery(
3929 lrrtt_pkt.destination_hash,
3930 &lrrtt_raw,
3931 lrrtt_pkt.packet_hash,
3932 rns_core::transport::types::InterfaceId(0),
3933 &mut rng,
3934 );
3935
3936 let actions =
3938 init_mgr.send_on_link(&link_id, b"hello link!", constants::CONTEXT_NONE, &mut rng);
3939 assert_eq!(actions.len(), 1);
3940 assert!(matches!(actions[0], LinkManagerAction::SendPacket { .. }));
3941 }
3942
3943 #[test]
3944 fn test_request_response() {
3945 let mut rng = OsRng;
3946 let dest_hash = [0xDD; 16];
3947 let mut resp_mgr = LinkManager::new();
3948 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
3949 resp_mgr.register_link_destination(
3950 dest_hash,
3951 sig_prv,
3952 sig_pub_bytes,
3953 ResourceStrategy::AcceptNone,
3954 );
3955
3956 resp_mgr.register_request_handler("/status", None, |_link_id, _path, _data, _remote| {
3958 Some(b"OK".to_vec())
3959 });
3960
3961 let mut init_mgr = LinkManager::new();
3962
3963 let (link_id, init_actions) = init_mgr.create_link(
3965 &dest_hash,
3966 &sig_pub_bytes,
3967 1,
3968 constants::MTU as u32,
3969 &mut rng,
3970 );
3971 let lr_raw = extract_send_packet(&init_actions);
3972 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
3973 let resp_actions = resp_mgr.handle_local_delivery(
3974 lr_pkt.destination_hash,
3975 &lr_raw,
3976 lr_pkt.packet_hash,
3977 rns_core::transport::types::InterfaceId(0),
3978 &mut rng,
3979 );
3980 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
3981 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
3982 let init_actions2 = init_mgr.handle_local_delivery(
3983 lrproof_pkt.destination_hash,
3984 &lrproof_raw,
3985 lrproof_pkt.packet_hash,
3986 rns_core::transport::types::InterfaceId(0),
3987 &mut rng,
3988 );
3989 let lrrtt_raw = extract_any_send_packet(&init_actions2);
3990 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
3991 resp_mgr.handle_local_delivery(
3992 lrrtt_pkt.destination_hash,
3993 &lrrtt_raw,
3994 lrrtt_pkt.packet_hash,
3995 rns_core::transport::types::InterfaceId(0),
3996 &mut rng,
3997 );
3998
3999 let req_actions = init_mgr.send_request(&link_id, "/status", b"query", &mut rng);
4001 assert_eq!(req_actions.len(), 1);
4002
4003 let req_raw = extract_send_packet_from(&req_actions);
4005 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
4006 let resp_actions = resp_mgr.handle_local_delivery(
4007 req_pkt.destination_hash,
4008 &req_raw,
4009 req_pkt.packet_hash,
4010 rns_core::transport::types::InterfaceId(0),
4011 &mut rng,
4012 );
4013
4014 let has_response = resp_actions
4016 .iter()
4017 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
4018 assert!(has_response, "Handler should produce a response packet");
4019 }
4020
4021 #[test]
4022 fn packet_request_at_exact_size_limit_is_dispatched() {
4023 use std::sync::{Arc, Mutex};
4024
4025 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4026 let mut rng = OsRng;
4027 let observed = Arc::new(Mutex::new(false));
4028 resp_mgr.register_request_handler("/bounded", None, {
4029 let observed = Arc::clone(&observed);
4030 move |_link_id, _path, _data, _remote| {
4031 *observed.lock().unwrap() = true;
4032 None
4033 }
4034 });
4035
4036 let request_actions = init_mgr.send_request(&link_id, "/bounded", b"\xc0", &mut rng);
4037 let raw = extract_any_send_packet(&request_actions);
4038 let packet = RawPacket::unpack(&raw).unwrap();
4039 let packed_size = resp_mgr.links[&link_id]
4040 .engine
4041 .decrypt(&packet.data)
4042 .unwrap()
4043 .len();
4044 resp_mgr.links.get_mut(&link_id).unwrap().max_request_size = Some(packed_size);
4045
4046 resp_mgr.handle_local_delivery(
4047 packet.destination_hash,
4048 &raw,
4049 packet.packet_hash,
4050 rns_core::transport::types::InterfaceId(0),
4051 &mut rng,
4052 );
4053
4054 assert!(*observed.lock().unwrap());
4055 }
4056
4057 #[test]
4058 fn packet_request_one_byte_over_size_limit_is_ignored_before_dispatch() {
4059 use std::sync::{Arc, Mutex};
4060
4061 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4062 let mut rng = OsRng;
4063 let observed = Arc::new(Mutex::new(false));
4064 resp_mgr.register_request_handler("/bounded", None, {
4065 let observed = Arc::clone(&observed);
4066 move |_link_id, _path, _data, _remote| {
4067 *observed.lock().unwrap() = true;
4068 None
4069 }
4070 });
4071
4072 let request_actions = init_mgr.send_request(&link_id, "/bounded", b"\xc0", &mut rng);
4073 let raw = extract_any_send_packet(&request_actions);
4074 let packet = RawPacket::unpack(&raw).unwrap();
4075 let packed_size = resp_mgr.links[&link_id]
4076 .engine
4077 .decrypt(&packet.data)
4078 .unwrap()
4079 .len();
4080 resp_mgr.links.get_mut(&link_id).unwrap().max_request_size = Some(packed_size - 1);
4081
4082 let actions = resp_mgr.handle_local_delivery(
4083 packet.destination_hash,
4084 &raw,
4085 packet.packet_hash,
4086 rns_core::transport::types::InterfaceId(0),
4087 &mut rng,
4088 );
4089
4090 assert!(actions.is_empty());
4091 assert!(!*observed.lock().unwrap());
4092 }
4093
4094 #[test]
4095 fn test_send_request_wraps_invalid_msgpack_data_as_bin() {
4096 use std::sync::{Arc, Mutex};
4097
4098 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4099 let mut rng = OsRng;
4100
4101 let invalid = vec![0xC1];
4102 let expected = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(invalid.clone()));
4103 let captured = Arc::new(Mutex::new(None::<Vec<u8>>));
4104 let captured_for_handler = Arc::clone(&captured);
4105
4106 resp_mgr.register_request_handler("/bin", None, move |_link_id, _path, data, _remote| {
4107 *captured_for_handler.lock().unwrap() = Some(data.to_vec());
4108 Some(rns_core::msgpack::pack(&rns_core::msgpack::Value::Bool(
4109 true,
4110 )))
4111 });
4112
4113 let req_actions = init_mgr.send_request(&link_id, "/bin", &invalid, &mut rng);
4114 let req_raw = extract_send_packet_from(&req_actions);
4115 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
4116 let resp_actions = resp_mgr.handle_local_delivery(
4117 req_pkt.destination_hash,
4118 &req_raw,
4119 req_pkt.packet_hash,
4120 rns_core::transport::types::InterfaceId(0),
4121 &mut rng,
4122 );
4123
4124 assert!(
4125 resp_actions
4126 .iter()
4127 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. })),
4128 "handler should still produce a response"
4129 );
4130 assert_eq!(*captured.lock().unwrap(), Some(expected));
4131 }
4132
4133 #[test]
4134 fn test_invalid_response_bytes_are_returned_as_msgpack_bin() {
4135 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4136 let mut rng = OsRng;
4137 let invalid_response = vec![0xC1];
4138 let expected =
4139 rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(invalid_response.clone()));
4140
4141 resp_mgr.register_request_handler("/invalid-response", None, {
4142 let invalid_response = invalid_response.clone();
4143 move |_link_id, _path, _data, _remote| Some(invalid_response.clone())
4144 });
4145
4146 let req_actions = init_mgr.send_request(&link_id, "/invalid-response", b"\xc0", &mut rng);
4147 let req_raw = extract_send_packet_from(&req_actions);
4148 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
4149 let resp_actions = resp_mgr.handle_local_delivery(
4150 req_pkt.destination_hash,
4151 &req_raw,
4152 req_pkt.packet_hash,
4153 rns_core::transport::types::InterfaceId(0),
4154 &mut rng,
4155 );
4156
4157 let resp_raw = extract_any_send_packet(&resp_actions);
4158 let resp_pkt = RawPacket::unpack(&resp_raw).unwrap();
4159 let init_actions = init_mgr.handle_local_delivery(
4160 resp_pkt.destination_hash,
4161 &resp_raw,
4162 resp_pkt.packet_hash,
4163 rns_core::transport::types::InterfaceId(0),
4164 &mut rng,
4165 );
4166
4167 let response_data = init_actions
4168 .iter()
4169 .find_map(|action| match action {
4170 LinkManagerAction::ResponseReceived { data, .. } => Some(data.clone()),
4171 _ => None,
4172 })
4173 .expect("initiator should receive a response");
4174 assert_eq!(response_data, expected);
4175 }
4176
4177 #[test]
4178 fn packet_response_at_exact_size_limit_is_accepted() {
4179 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4180 let mut rng = OsRng;
4181 let payload = vec![0xA5; 64];
4182 let response_value =
4183 rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload.clone()));
4184 resp_mgr.register_request_handler("/bounded-response", None, {
4185 let response_value = response_value.clone();
4186 move |_, _, _, _| Some(response_value.clone())
4187 });
4188
4189 let request_actions = init_mgr.send_request_with_max_response_size(
4190 &link_id,
4191 "/bounded-response",
4192 b"\xc0",
4193 Some(payload.len()),
4194 &mut rng,
4195 );
4196 let request_raw = extract_any_send_packet(&request_actions);
4197 let request = RawPacket::unpack(&request_raw).unwrap();
4198 let response_actions = resp_mgr.handle_local_delivery(
4199 request.destination_hash,
4200 &request_raw,
4201 request.packet_hash,
4202 rns_core::transport::types::InterfaceId(0),
4203 &mut rng,
4204 );
4205 let response_raw = extract_any_send_packet(&response_actions);
4206 let response = RawPacket::unpack(&response_raw).unwrap();
4207 let actions = init_mgr.handle_local_delivery(
4208 response.destination_hash,
4209 &response_raw,
4210 response.packet_hash,
4211 rns_core::transport::types::InterfaceId(0),
4212 &mut rng,
4213 );
4214
4215 assert!(actions.iter().any(|action| matches!(
4216 action,
4217 LinkManagerAction::ResponseReceived { data, .. } if data == &response_value
4218 )));
4219 assert!(!actions
4220 .iter()
4221 .any(|action| matches!(action, LinkManagerAction::RequestFailed { .. })));
4222 }
4223
4224 #[test]
4225 fn packet_response_one_byte_over_size_limit_fails_request() {
4226 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4227 let mut rng = OsRng;
4228 let payload = vec![0xA5; 64];
4229 let response_value =
4230 rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload.clone()));
4231 resp_mgr.register_request_handler("/bounded-response", None, {
4232 let response_value = response_value.clone();
4233 move |_, _, _, _| Some(response_value.clone())
4234 });
4235
4236 let request_actions = init_mgr.send_request_with_max_response_size(
4237 &link_id,
4238 "/bounded-response",
4239 b"\xc0",
4240 Some(payload.len() - 1),
4241 &mut rng,
4242 );
4243 let request_raw = extract_any_send_packet(&request_actions);
4244 let request = RawPacket::unpack(&request_raw).unwrap();
4245 let request_id = request.get_truncated_hash();
4246 let response_actions = resp_mgr.handle_local_delivery(
4247 request.destination_hash,
4248 &request_raw,
4249 request.packet_hash,
4250 rns_core::transport::types::InterfaceId(0),
4251 &mut rng,
4252 );
4253 let response_raw = extract_any_send_packet(&response_actions);
4254 let response = RawPacket::unpack(&response_raw).unwrap();
4255 let actions = init_mgr.handle_local_delivery(
4256 response.destination_hash,
4257 &response_raw,
4258 response.packet_hash,
4259 rns_core::transport::types::InterfaceId(0),
4260 &mut rng,
4261 );
4262
4263 assert!(actions.iter().any(|action| matches!(
4264 action,
4265 LinkManagerAction::RequestFailed {
4266 request_id: failed_id,
4267 reason: RequestFailure::ResponseTooLarge { size: 64, maximum: 63 },
4268 ..
4269 } if failed_id == &request_id
4270 )));
4271 assert!(!actions
4272 .iter()
4273 .any(|action| matches!(action, LinkManagerAction::ResponseReceived { .. })));
4274 assert!(!init_mgr.links[&link_id]
4275 .pending_requests
4276 .contains_key(&request_id));
4277 assert!(init_mgr
4278 .handle_local_delivery(
4279 response.destination_hash,
4280 &response_raw,
4281 response.packet_hash,
4282 rns_core::transport::types::InterfaceId(0),
4283 &mut rng,
4284 )
4285 .is_empty());
4286 }
4287
4288 #[test]
4289 fn test_request_acl_deny_unidentified() {
4290 let mut rng = OsRng;
4291 let dest_hash = [0xDD; 16];
4292 let mut resp_mgr = LinkManager::new();
4293 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
4294 resp_mgr.register_link_destination(
4295 dest_hash,
4296 sig_prv,
4297 sig_pub_bytes,
4298 ResourceStrategy::AcceptNone,
4299 );
4300
4301 resp_mgr.register_request_handler(
4303 "/restricted",
4304 Some(vec![[0xAA; 16]]),
4305 |_link_id, _path, _data, _remote| Some(b"secret".to_vec()),
4306 );
4307
4308 let mut init_mgr = LinkManager::new();
4309
4310 let (link_id, init_actions) = init_mgr.create_link(
4312 &dest_hash,
4313 &sig_pub_bytes,
4314 1,
4315 constants::MTU as u32,
4316 &mut rng,
4317 );
4318 let lr_raw = extract_send_packet(&init_actions);
4319 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
4320 let resp_actions = resp_mgr.handle_local_delivery(
4321 lr_pkt.destination_hash,
4322 &lr_raw,
4323 lr_pkt.packet_hash,
4324 rns_core::transport::types::InterfaceId(0),
4325 &mut rng,
4326 );
4327 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
4328 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
4329 let init_actions2 = init_mgr.handle_local_delivery(
4330 lrproof_pkt.destination_hash,
4331 &lrproof_raw,
4332 lrproof_pkt.packet_hash,
4333 rns_core::transport::types::InterfaceId(0),
4334 &mut rng,
4335 );
4336 let lrrtt_raw = extract_any_send_packet(&init_actions2);
4337 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
4338 resp_mgr.handle_local_delivery(
4339 lrrtt_pkt.destination_hash,
4340 &lrrtt_raw,
4341 lrrtt_pkt.packet_hash,
4342 rns_core::transport::types::InterfaceId(0),
4343 &mut rng,
4344 );
4345
4346 let req_actions = init_mgr.send_request(&link_id, "/restricted", b"query", &mut rng);
4348 let req_raw = extract_send_packet_from(&req_actions);
4349 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
4350 let resp_actions = resp_mgr.handle_local_delivery(
4351 req_pkt.destination_hash,
4352 &req_raw,
4353 req_pkt.packet_hash,
4354 rns_core::transport::types::InterfaceId(0),
4355 &mut rng,
4356 );
4357
4358 let has_response = resp_actions
4360 .iter()
4361 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
4362 assert!(!has_response, "Unidentified peer should be denied");
4363 }
4364
4365 #[test]
4366 fn test_teardown_link() {
4367 let mut rng = OsRng;
4368 let dest_hash = [0xDD; 16];
4369 let mut mgr = LinkManager::new();
4370
4371 let dummy_sig = [0xAA; 32];
4372 let (link_id, _) =
4373 mgr.create_link(&dest_hash, &dummy_sig, 1, constants::MTU as u32, &mut rng);
4374 assert_eq!(mgr.link_count(), 1);
4375
4376 let actions = mgr.teardown_link(&link_id);
4377 let has_close = actions
4378 .iter()
4379 .any(|a| matches!(a, LinkManagerAction::LinkClosed { .. }));
4380 assert!(has_close);
4381 assert!(!actions
4382 .iter()
4383 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. })));
4384 assert!(mgr.teardown_link(&link_id).is_empty());
4385
4386 let tick_actions = mgr.tick(&mut rng);
4388 let has_deregister = tick_actions
4389 .iter()
4390 .any(|a| matches!(a, LinkManagerAction::DeregisterLinkDest { .. }));
4391 assert!(has_deregister);
4392 assert_eq!(mgr.link_count(), 0);
4393 }
4394
4395 #[test]
4396 fn active_teardown_sends_linkclose_and_remote_closed_teardown_is_noop() {
4397 let (mut initiator, mut responder, link_id) = setup_active_link();
4398 let actions = initiator.teardown_link(&link_id);
4399 let close_packet = actions
4400 .iter()
4401 .find_map(|action| match action {
4402 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
4403 .ok()
4404 .filter(|packet| packet.context == constants::CONTEXT_LINKCLOSE),
4405 _ => None,
4406 })
4407 .expect("active teardown must emit LINKCLOSE");
4408 assert!(close_packet.raw.len() > constants::HEADER_MINSIZE);
4409 let plaintext = responder
4410 .links
4411 .get(&link_id)
4412 .unwrap()
4413 .engine
4414 .decrypt(&close_packet.data)
4415 .expect("LINKCLOSE payload must authenticate");
4416 assert_eq!(plaintext, link_id);
4417 assert!(initiator.teardown_link(&link_id).is_empty());
4418
4419 responder
4420 .links
4421 .get_mut(&link_id)
4422 .unwrap()
4423 .engine
4424 .handle_teardown();
4425 assert!(responder.teardown_link(&link_id).is_empty());
4426 }
4427
4428 #[test]
4429 fn forged_or_empty_linkclose_does_not_close_an_active_link() {
4430 let (_initiator, mut responder, link_id) = setup_active_link();
4431 let flags = PacketFlags {
4432 header_type: constants::HEADER_1,
4433 context_flag: constants::FLAG_UNSET,
4434 transport_type: constants::TRANSPORT_BROADCAST,
4435 destination_type: constants::DESTINATION_LINK,
4436 packet_type: constants::PACKET_TYPE_DATA,
4437 };
4438 let mut rng = OsRng;
4439
4440 for forged_payload in [Vec::new(), vec![0x55; 48]] {
4441 let packet = RawPacket::pack(
4442 flags,
4443 0,
4444 &link_id,
4445 None,
4446 constants::CONTEXT_LINKCLOSE,
4447 &forged_payload,
4448 )
4449 .unwrap();
4450 let actions = responder.handle_local_delivery(
4451 link_id,
4452 &packet.raw,
4453 [0u8; 32],
4454 rns_core::transport::types::InterfaceId(1),
4455 &mut rng,
4456 );
4457
4458 assert_eq!(responder.link_state(&link_id), Some(LinkState::Active));
4459 assert!(!actions
4460 .iter()
4461 .any(|action| matches!(action, LinkManagerAction::LinkClosed { .. })));
4462 }
4463 }
4464
4465 #[test]
4466 fn test_identify_on_link() {
4467 let mut rng = OsRng;
4468 let dest_hash = [0xDD; 16];
4469 let mut resp_mgr = LinkManager::new();
4470 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
4471 resp_mgr.register_link_destination(
4472 dest_hash,
4473 sig_prv,
4474 sig_pub_bytes,
4475 ResourceStrategy::AcceptNone,
4476 );
4477 let mut init_mgr = LinkManager::new();
4478
4479 let (link_id, init_actions) = init_mgr.create_link(
4481 &dest_hash,
4482 &sig_pub_bytes,
4483 1,
4484 constants::MTU as u32,
4485 &mut rng,
4486 );
4487 let lr_raw = extract_send_packet(&init_actions);
4488 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
4489 let resp_actions = resp_mgr.handle_local_delivery(
4490 lr_pkt.destination_hash,
4491 &lr_raw,
4492 lr_pkt.packet_hash,
4493 rns_core::transport::types::InterfaceId(0),
4494 &mut rng,
4495 );
4496 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
4497 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
4498 let init_actions2 = init_mgr.handle_local_delivery(
4499 lrproof_pkt.destination_hash,
4500 &lrproof_raw,
4501 lrproof_pkt.packet_hash,
4502 rns_core::transport::types::InterfaceId(0),
4503 &mut rng,
4504 );
4505 let lrrtt_raw = extract_any_send_packet(&init_actions2);
4506 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
4507 resp_mgr.handle_local_delivery(
4508 lrrtt_pkt.destination_hash,
4509 &lrrtt_raw,
4510 lrrtt_pkt.packet_hash,
4511 rns_core::transport::types::InterfaceId(0),
4512 &mut rng,
4513 );
4514
4515 let identity = Identity::new(&mut rng);
4517 let id_actions = init_mgr.identify(&link_id, &identity, &mut rng);
4518 assert_eq!(id_actions.len(), 1);
4519
4520 let id_raw = extract_send_packet_from(&id_actions);
4522 let id_pkt = RawPacket::unpack(&id_raw).unwrap();
4523 let resp_actions = resp_mgr.handle_local_delivery(
4524 id_pkt.destination_hash,
4525 &id_raw,
4526 id_pkt.packet_hash,
4527 rns_core::transport::types::InterfaceId(0),
4528 &mut rng,
4529 );
4530
4531 let has_identified = resp_actions
4532 .iter()
4533 .any(|a| matches!(a, LinkManagerAction::RemoteIdentified { .. }));
4534 assert!(has_identified, "Responder should emit RemoteIdentified");
4535 }
4536
4537 #[test]
4538 fn test_path_hash_computation() {
4539 let h1 = compute_path_hash("/status");
4540 let h2 = compute_path_hash("/path");
4541 assert_ne!(h1, h2);
4542
4543 assert_eq!(h1, compute_path_hash("/status"));
4545 }
4546
4547 #[test]
4548 fn test_link_count() {
4549 let mut mgr = LinkManager::new();
4550 let mut rng = OsRng;
4551
4552 assert_eq!(mgr.link_count(), 0);
4553
4554 let dummy_sig = [0xAA; 32];
4555 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
4556 assert_eq!(mgr.link_count(), 1);
4557
4558 mgr.create_link(&[0x22; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
4559 assert_eq!(mgr.link_count(), 2);
4560 }
4561
4562 fn extract_send_packet(actions: &[LinkManagerAction]) -> Vec<u8> {
4565 extract_send_packet_at(actions, actions.len() - 1)
4566 }
4567
4568 fn extract_send_packet_at(actions: &[LinkManagerAction], idx: usize) -> Vec<u8> {
4569 match &actions[idx] {
4570 LinkManagerAction::SendPacket { raw, .. } => raw.clone(),
4571 other => panic!("Expected SendPacket at index {}, got {:?}", idx, other),
4572 }
4573 }
4574
4575 fn extract_any_send_packet(actions: &[LinkManagerAction]) -> Vec<u8> {
4576 actions
4577 .iter()
4578 .find_map(|a| match a {
4579 LinkManagerAction::SendPacket { raw, .. } => Some(raw.clone()),
4580 _ => None,
4581 })
4582 .expect("Expected at least one SendPacket action")
4583 }
4584
4585 fn extract_send_packet_from(actions: &[LinkManagerAction]) -> Vec<u8> {
4586 extract_any_send_packet(actions)
4587 }
4588
4589 fn setup_active_link() -> (LinkManager, LinkManager, LinkId) {
4592 setup_active_link_with_max_request_size(None)
4593 }
4594
4595 fn setup_active_link_with_max_request_size(
4596 max_request_size: Option<usize>,
4597 ) -> (LinkManager, LinkManager, LinkId) {
4598 let mut rng = OsRng;
4599 let dest_hash = [0xDD; 16];
4600 let mut resp_mgr = LinkManager::new();
4601 let (sig_prv, sig_pub_bytes) = make_dest_keys(&mut rng);
4602 resp_mgr.register_link_destination(
4603 dest_hash,
4604 sig_prv,
4605 sig_pub_bytes,
4606 ResourceStrategy::AcceptNone,
4607 );
4608 assert!(resp_mgr.set_link_destination_max_request_size(&dest_hash, max_request_size));
4609 let mut init_mgr = LinkManager::new();
4610
4611 let (link_id, init_actions) = init_mgr.create_link(
4612 &dest_hash,
4613 &sig_pub_bytes,
4614 1,
4615 constants::MTU as u32,
4616 &mut rng,
4617 );
4618 let lr_raw = extract_send_packet(&init_actions);
4619 let lr_pkt = RawPacket::unpack(&lr_raw).unwrap();
4620 let resp_actions = resp_mgr.handle_local_delivery(
4621 lr_pkt.destination_hash,
4622 &lr_raw,
4623 lr_pkt.packet_hash,
4624 rns_core::transport::types::InterfaceId(0),
4625 &mut rng,
4626 );
4627 let lrproof_raw = extract_send_packet_at(&resp_actions, 1);
4628 let lrproof_pkt = RawPacket::unpack(&lrproof_raw).unwrap();
4629 let init_actions2 = init_mgr.handle_local_delivery(
4630 lrproof_pkt.destination_hash,
4631 &lrproof_raw,
4632 lrproof_pkt.packet_hash,
4633 rns_core::transport::types::InterfaceId(0),
4634 &mut rng,
4635 );
4636 let lrrtt_raw = extract_any_send_packet(&init_actions2);
4637 let lrrtt_pkt = RawPacket::unpack(&lrrtt_raw).unwrap();
4638 resp_mgr.handle_local_delivery(
4639 lrrtt_pkt.destination_hash,
4640 &lrrtt_raw,
4641 lrrtt_pkt.packet_hash,
4642 rns_core::transport::types::InterfaceId(0),
4643 &mut rng,
4644 );
4645
4646 assert_eq!(init_mgr.link_state(&link_id), Some(LinkState::Active));
4647 assert_eq!(resp_mgr.link_state(&link_id), Some(LinkState::Active));
4648
4649 (init_mgr, resp_mgr, link_id)
4650 }
4651
4652 #[test]
4653 fn request_size_limit_is_inherited_when_link_is_accepted() {
4654 let (_init_mgr, resp_mgr, link_id) = setup_active_link_with_max_request_size(Some(321));
4655 assert_eq!(resp_mgr.links[&link_id].max_request_size, Some(321));
4656 }
4657
4658 #[test]
4659 fn request_size_limit_updates_existing_responder_links() {
4660 let (_init_mgr, mut resp_mgr, link_id) = setup_active_link();
4661 let dest_hash = resp_mgr.links[&link_id].dest_hash;
4662 assert!(resp_mgr.set_link_destination_max_request_size(&dest_hash, Some(654)));
4663 assert_eq!(resp_mgr.links[&link_id].max_request_size, Some(654));
4664 }
4665
4666 #[test]
4667 fn setting_request_size_limit_for_unknown_destination_fails() {
4668 let mut mgr = LinkManager::new();
4669 assert!(!mgr.set_link_destination_max_request_size(&[0xEE; 16], Some(1)));
4670 }
4671
4672 fn packed_keepalive(link_id: &LinkId, payload: &[u8]) -> (Vec<u8>, RawPacket) {
4673 let flags = PacketFlags {
4674 header_type: constants::HEADER_1,
4675 context_flag: constants::FLAG_UNSET,
4676 transport_type: constants::TRANSPORT_BROADCAST,
4677 destination_type: constants::DESTINATION_LINK,
4678 packet_type: constants::PACKET_TYPE_DATA,
4679 };
4680 let (raw, _) = RawPacket::pack_raw_with_hash(
4681 flags,
4682 0,
4683 link_id,
4684 None,
4685 constants::CONTEXT_KEEPALIVE,
4686 payload,
4687 )
4688 .unwrap();
4689 let packet = RawPacket::unpack(&raw).unwrap();
4690 (raw, packet)
4691 }
4692
4693 fn keepalive_packets(actions: &[LinkManagerAction]) -> Vec<RawPacket> {
4694 actions
4695 .iter()
4696 .filter_map(|action| match action {
4697 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw).ok(),
4698 _ => None,
4699 })
4700 .filter(|packet| packet.context == constants::CONTEXT_KEEPALIVE)
4701 .collect()
4702 }
4703
4704 #[test]
4705 fn initiator_tick_emits_ff_keepalive_probe_when_outbound_is_quiet() {
4706 let (mut initiator, _, link_id) = setup_active_link();
4707 let now = time::now();
4708 let keepalive = initiator.links[&link_id].engine.keepalive_interval();
4709 let engine = &mut initiator.links.get_mut(&link_id).unwrap().engine;
4710 engine.record_inbound(now);
4711 engine.record_outbound(now - keepalive - 1.0, true);
4712
4713 let packets = keepalive_packets(&initiator.tick(&mut OsRng));
4714 assert_eq!(packets.len(), 1);
4715 assert_eq!(packets[0].data, [0xff]);
4716 }
4717
4718 #[test]
4719 fn responder_tick_never_originates_keepalive_probe() {
4720 let (_, mut responder, link_id) = setup_active_link();
4721 let now = time::now();
4722 let keepalive = responder.links[&link_id].engine.keepalive_interval();
4723 let engine = &mut responder.links.get_mut(&link_id).unwrap().engine;
4724 engine.record_inbound(now - keepalive - 1.0);
4725 engine.record_outbound(now - keepalive - 1.0, false);
4726
4727 assert!(keepalive_packets(&responder.tick(&mut OsRng)).is_empty());
4728 }
4729
4730 #[test]
4731 fn responder_acknowledges_ff_probe_after_outbound_silence() {
4732 let (_, mut responder, link_id) = setup_active_link();
4733 let now = time::now();
4734 let keepalive = responder.links[&link_id].engine.keepalive_interval();
4735 responder
4736 .links
4737 .get_mut(&link_id)
4738 .unwrap()
4739 .engine
4740 .record_outbound(now - keepalive - 1.0, false);
4741 let (raw, packet) = packed_keepalive(&link_id, &[0xff]);
4742
4743 let actions = responder.handle_local_delivery(
4744 packet.destination_hash,
4745 &raw,
4746 packet.packet_hash,
4747 rns_core::transport::types::InterfaceId(0),
4748 &mut OsRng,
4749 );
4750 let packets = keepalive_packets(&actions);
4751 assert_eq!(packets.len(), 1);
4752 assert_eq!(packets[0].data, [0xfe]);
4753 }
4754
4755 #[test]
4756 fn recent_responder_traffic_suppresses_keepalive_ack() {
4757 let (_, mut responder, link_id) = setup_active_link();
4758 responder
4759 .links
4760 .get_mut(&link_id)
4761 .unwrap()
4762 .engine
4763 .record_outbound(time::now(), false);
4764 let (raw, packet) = packed_keepalive(&link_id, &[0xff]);
4765
4766 let actions = responder.handle_local_delivery(
4767 packet.destination_hash,
4768 &raw,
4769 packet.packet_hash,
4770 rns_core::transport::types::InterfaceId(0),
4771 &mut OsRng,
4772 );
4773 assert!(keepalive_packets(&actions).is_empty());
4774 }
4775
4776 #[test]
4777 fn keepalive_ack_does_not_create_ping_pong_reply() {
4778 let (mut initiator, _, link_id) = setup_active_link();
4779 let (raw, packet) = packed_keepalive(&link_id, &[0xfe]);
4780
4781 let actions = initiator.handle_local_delivery(
4782 packet.destination_hash,
4783 &raw,
4784 packet.packet_hash,
4785 rns_core::transport::types::InterfaceId(0),
4786 &mut OsRng,
4787 );
4788 assert!(keepalive_packets(&actions).is_empty());
4789 }
4790
4791 #[test]
4796 fn test_resource_strategy_default() {
4797 let mut mgr = LinkManager::new();
4798 let mut rng = OsRng;
4799 let dummy_sig = [0xAA; 32];
4800 let (link_id, _) =
4801 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
4802
4803 let link = mgr.links.get(&link_id).unwrap();
4805 assert_eq!(link.resource_strategy, ResourceStrategy::AcceptNone);
4806 }
4807
4808 #[test]
4809 fn test_set_resource_strategy() {
4810 let mut mgr = LinkManager::new();
4811 let mut rng = OsRng;
4812 let dummy_sig = [0xAA; 32];
4813 let (link_id, _) =
4814 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
4815
4816 mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
4817 assert_eq!(
4818 mgr.links.get(&link_id).unwrap().resource_strategy,
4819 ResourceStrategy::AcceptAll
4820 );
4821
4822 mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
4823 assert_eq!(
4824 mgr.links.get(&link_id).unwrap().resource_strategy,
4825 ResourceStrategy::AcceptApp
4826 );
4827 }
4828
4829 #[test]
4830 fn test_send_resource_on_active_link() {
4831 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4832 let mut rng = OsRng;
4833
4834 let data = vec![0xAB; 100]; let actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
4837
4838 let has_send = actions
4840 .iter()
4841 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
4842 assert!(
4843 has_send,
4844 "send_resource should emit advertisement SendPacket"
4845 );
4846 }
4847
4848 #[test]
4849 fn test_large_send_request_uses_request_resource() {
4850 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4851 let mut rng = OsRng;
4852
4853 let data = vec![0xAB; 2048];
4854 let actions = init_mgr.send_request(&link_id, "/large", &data, &mut rng);
4855
4856 let adv = first_resource_advertisement(&init_mgr, &link_id, &actions);
4857 assert!(adv.is_request());
4858 assert!(!adv.is_response());
4859 assert_eq!(adv.request_id.as_ref().map(Vec::len), Some(16));
4860 let request_id = LinkManager::response_request_id(&adv.request_id).unwrap();
4861 assert_eq!(
4862 init_mgr.links[&link_id]
4863 .pending_requests
4864 .get(&request_id)
4865 .and_then(|request| request.deadline),
4866 None,
4867 "resource request timeout must wait for delivery proof"
4868 );
4869 assert!(init_mgr.links[&link_id]
4870 .pending_requests
4871 .contains_key(&request_id));
4872
4873 let has_request_packet = actions.iter().any(|action| match action {
4874 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
4875 .map(|pkt| pkt.context == constants::CONTEXT_REQUEST)
4876 .unwrap_or(false),
4877 _ => false,
4878 });
4879 assert!(!has_request_packet);
4880 }
4881
4882 #[test]
4883 fn unanswered_packet_request_expires_from_pending_set() {
4884 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
4885 let mut rng = OsRng;
4886 let actions = init_mgr.send_request(&link_id, "/no-response", b"\xc0", &mut rng);
4887 let raw = extract_any_send_packet(&actions);
4888 let request_id = RawPacket::unpack(&raw).unwrap().get_truncated_hash();
4889 assert!(init_mgr.links[&link_id]
4890 .pending_requests
4891 .get(&request_id)
4892 .is_some_and(|request| request.deadline.is_some()));
4893
4894 init_mgr
4895 .links
4896 .get_mut(&link_id)
4897 .unwrap()
4898 .pending_requests
4899 .get_mut(&request_id)
4900 .unwrap()
4901 .deadline = Some(time::now() - 1.0);
4902 init_mgr.tick(&mut rng);
4903
4904 assert!(!init_mgr.links[&link_id]
4905 .pending_requests
4906 .contains_key(&request_id));
4907 }
4908
4909 #[test]
4910 fn invalid_resource_advertisement_tears_down_link_once() {
4911 let (_init_mgr, mut resp_mgr, link_id) = setup_active_link();
4912 let mut rng = OsRng;
4913
4914 let actions = resp_mgr.handle_resource_adv(&link_id, &[0xc1], &mut rng);
4915
4916 assert_eq!(resp_mgr.link_state(&link_id), Some(LinkState::Closed));
4917 assert_eq!(
4918 actions
4919 .iter()
4920 .filter(|action| matches!(action, LinkManagerAction::LinkClosed { .. }))
4921 .count(),
4922 1
4923 );
4924 assert!(resp_mgr
4925 .handle_resource_adv(&link_id, &[0xc1], &mut rng)
4926 .is_empty());
4927 }
4928
4929 #[test]
4930 fn oversized_resource_advertisement_tears_down_link() {
4931 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4932 let mut rng = OsRng;
4933 let adv_actions = init_mgr.send_resource_with_auto_compress(
4934 &link_id,
4935 &deterministic_bytes(1024),
4936 None,
4937 false,
4938 &mut rng,
4939 );
4940 let mut adv = first_resource_advertisement(&init_mgr, &link_id, &adv_actions);
4941 adv.transfer_size = (constants::RESOURCE_MAX_EFFICIENT_SIZE * 3 + 1) as u64;
4942
4943 let actions = resp_mgr.handle_resource_adv(&link_id, &adv.pack(0), &mut rng);
4944
4945 assert_eq!(resp_mgr.link_state(&link_id), Some(LinkState::Closed));
4946 assert!(actions
4947 .iter()
4948 .any(|action| matches!(action, LinkManagerAction::LinkClosed { .. })));
4949 }
4950
4951 #[test]
4952 fn request_resource_without_any_handler_is_ignored() {
4953 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4954 let mut rng = OsRng;
4955 let data = deterministic_bytes(4096);
4956
4957 let adv_actions = init_mgr.send_request(&link_id, "/missing", &data, &mut rng);
4958 let adv_raw = extract_any_send_packet(&adv_actions);
4959 let adv_pkt = RawPacket::unpack(&adv_raw).unwrap();
4960 assert_eq!(adv_pkt.context, constants::CONTEXT_RESOURCE_ADV);
4961
4962 let actions = resp_mgr.handle_local_delivery(
4963 adv_pkt.destination_hash,
4964 &adv_raw,
4965 adv_pkt.packet_hash,
4966 rns_core::transport::types::InterfaceId(0),
4967 &mut rng,
4968 );
4969
4970 assert!(
4971 actions.is_empty(),
4972 "unhandled request resource must be ignored"
4973 );
4974 assert!(resp_mgr.links[&link_id].incoming_resources.is_empty());
4975 }
4976
4977 #[test]
4978 fn request_resource_at_exact_declared_size_limit_is_accepted() {
4979 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
4980 let mut rng = OsRng;
4981 resp_mgr.register_request_handler("/bounded", None, |_, _, _, _| None);
4982 let request = deterministic_bytes(4096);
4983 let adv_actions = init_mgr.send_request(&link_id, "/bounded", &request, &mut rng);
4984 let adv = first_resource_advertisement(&init_mgr, &link_id, &adv_actions);
4985 resp_mgr.links.get_mut(&link_id).unwrap().max_request_size = Some(adv.data_size as usize);
4986 let raw = extract_any_send_packet(&adv_actions);
4987 let packet = RawPacket::unpack(&raw).unwrap();
4988
4989 let actions = resp_mgr.handle_local_delivery(
4990 packet.destination_hash,
4991 &raw,
4992 packet.packet_hash,
4993 rns_core::transport::types::InterfaceId(0),
4994 &mut rng,
4995 );
4996
4997 assert!(
4998 !actions.is_empty(),
4999 "accepted resource should request parts"
5000 );
5001 assert_eq!(resp_mgr.links[&link_id].incoming_resources.len(), 1);
5002 }
5003
5004 #[test]
5005 fn request_resource_one_byte_over_declared_size_limit_is_rejected() {
5006 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5007 let mut rng = OsRng;
5008 resp_mgr.register_request_handler("/bounded", None, |_, _, _, _| None);
5009 let adv_actions =
5010 init_mgr.send_request(&link_id, "/bounded", &deterministic_bytes(4096), &mut rng);
5011 let adv = first_resource_advertisement(&init_mgr, &link_id, &adv_actions);
5012 resp_mgr.links.get_mut(&link_id).unwrap().max_request_size =
5013 Some(adv.data_size as usize - 1);
5014 let raw = extract_any_send_packet(&adv_actions);
5015 let packet = RawPacket::unpack(&raw).unwrap();
5016
5017 let actions = resp_mgr.handle_local_delivery(
5018 packet.destination_hash,
5019 &raw,
5020 packet.packet_hash,
5021 rns_core::transport::types::InterfaceId(0),
5022 &mut rng,
5023 );
5024
5025 assert!(!actions.is_empty(), "rejected resource should send an RCL");
5026 assert!(resp_mgr.links[&link_id].incoming_resources.is_empty());
5027 }
5028
5029 #[test]
5030 fn request_size_limit_does_not_reject_ordinary_resources() {
5031 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5032 let mut rng = OsRng;
5033 let link = resp_mgr.links.get_mut(&link_id).unwrap();
5034 link.max_request_size = Some(0);
5035 link.resource_strategy = ResourceStrategy::AcceptAll;
5036 let adv_actions = init_mgr.send_resource_with_auto_compress(
5037 &link_id,
5038 &deterministic_bytes(4096),
5039 None,
5040 false,
5041 &mut rng,
5042 );
5043 let raw = extract_any_send_packet(&adv_actions);
5044 let packet = RawPacket::unpack(&raw).unwrap();
5045
5046 let actions = resp_mgr.handle_local_delivery(
5047 packet.destination_hash,
5048 &raw,
5049 packet.packet_hash,
5050 rns_core::transport::types::InterfaceId(0),
5051 &mut rng,
5052 );
5053
5054 assert!(!actions.is_empty(), "ordinary resource should be accepted");
5055 assert_eq!(resp_mgr.links[&link_id].incoming_resources.len(), 1);
5056 }
5057
5058 #[test]
5059 fn unsolicited_response_resource_is_ignored() {
5060 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5061 let mut rng = OsRng;
5062 let unsolicited_id = [0xEF; 16];
5063
5064 let adv_actions = resp_mgr.send_response_resource(
5065 &link_id,
5066 &unsolicited_id,
5067 &deterministic_bytes(4096),
5068 None,
5069 false,
5070 &mut rng,
5071 );
5072 let adv_raw = extract_any_send_packet(&adv_actions);
5073 let adv_pkt = RawPacket::unpack(&adv_raw).unwrap();
5074 let actions = init_mgr.handle_local_delivery(
5075 adv_pkt.destination_hash,
5076 &adv_raw,
5077 adv_pkt.packet_hash,
5078 rns_core::transport::types::InterfaceId(0),
5079 &mut rng,
5080 );
5081
5082 assert!(
5083 actions.is_empty(),
5084 "unsolicited response resource must be ignored"
5085 );
5086 assert!(init_mgr.links[&link_id].incoming_resources.is_empty());
5087 }
5088
5089 #[test]
5090 fn resource_packets_are_suppressed_after_link_closes() {
5091 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5092 let mut rng = OsRng;
5093 init_mgr.send_resource_with_auto_compress(
5094 &link_id,
5095 &deterministic_bytes(4096),
5096 None,
5097 false,
5098 &mut rng,
5099 );
5100 init_mgr
5101 .links
5102 .get_mut(&link_id)
5103 .unwrap()
5104 .engine
5105 .handle_teardown();
5106
5107 for action in [
5108 ResourceAction::SendAdvertisement(vec![1]),
5109 ResourceAction::SendPart(vec![2]),
5110 ResourceAction::SendRequest(vec![3]),
5111 ResourceAction::SendHmu(vec![4]),
5112 ResourceAction::SendProof(vec![5; 64]),
5113 ] {
5114 let emitted = init_mgr.process_resource_actions(&link_id, vec![action], &mut rng);
5115 assert!(
5116 !emitted
5117 .iter()
5118 .any(|item| matches!(item, LinkManagerAction::SendPacket { .. })),
5119 "closed link emitted a resource packet"
5120 );
5121 }
5122 assert!(init_mgr.links[&link_id]
5123 .outgoing_resources
5124 .iter()
5125 .all(|sender| sender.status == rns_core::resource::ResourceStatus::Failed));
5126 }
5127
5128 #[test]
5129 fn large_request_resource_reaches_registered_handler_and_returns_response() {
5130 use std::sync::{Arc, Mutex};
5131
5132 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5133 let mut rng = OsRng;
5134 let request_value =
5135 rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(deterministic_bytes(4096)));
5136 let observed = Arc::new(Mutex::new(None));
5137 resp_mgr.register_request_handler("/large-request", None, {
5138 let observed = Arc::clone(&observed);
5139 move |_link_id, _path, data, _remote| {
5140 *observed.lock().unwrap() = Some(data.to_vec());
5141 Some(rns_core::msgpack::pack(&rns_core::msgpack::Value::Bool(
5142 true,
5143 )))
5144 }
5145 });
5146
5147 let initial = init_mgr.send_request(&link_id, "/large-request", &request_value, &mut rng);
5148 let mut pending: Vec<(char, LinkManagerAction)> =
5149 initial.into_iter().map(|action| ('i', action)).collect();
5150 let mut response = None;
5151
5152 for _ in 0..300 {
5153 if pending.is_empty() || response.is_some() {
5154 break;
5155 }
5156 let mut next = Vec::new();
5157 for (source, action) in pending.drain(..) {
5158 let LinkManagerAction::SendPacket { raw, .. } = action else {
5159 continue;
5160 };
5161 let packet = RawPacket::unpack(&raw).unwrap();
5162 let actions = if source == 'i' {
5163 resp_mgr.handle_local_delivery(
5164 packet.destination_hash,
5165 &raw,
5166 packet.packet_hash,
5167 rns_core::transport::types::InterfaceId(0),
5168 &mut rng,
5169 )
5170 } else {
5171 init_mgr.handle_local_delivery(
5172 packet.destination_hash,
5173 &raw,
5174 packet.packet_hash,
5175 rns_core::transport::types::InterfaceId(0),
5176 &mut rng,
5177 )
5178 };
5179 for action in &actions {
5180 if let LinkManagerAction::ResponseReceived { data, .. } = action {
5181 response = Some(data.clone());
5182 }
5183 assert!(!matches!(
5184 action,
5185 LinkManagerAction::ResourceAcceptQuery { .. }
5186 ));
5187 }
5188 let next_source = if source == 'i' { 'r' } else { 'i' };
5189 next.extend(actions.into_iter().map(|action| (next_source, action)));
5190 }
5191 pending = next;
5192 }
5193
5194 assert_eq!(*observed.lock().unwrap(), Some(request_value));
5195 assert_eq!(
5196 response,
5197 Some(rns_core::msgpack::pack(&rns_core::msgpack::Value::Bool(
5198 true
5199 )))
5200 );
5201 assert!(init_mgr.links[&link_id].pending_requests.is_empty());
5202 }
5203
5204 fn first_resource_advertisement(
5205 mgr: &LinkManager,
5206 link_id: &LinkId,
5207 actions: &[LinkManagerAction],
5208 ) -> rns_core::resource::ResourceAdvertisement {
5209 let adv_raw = actions
5210 .iter()
5211 .find_map(|action| match action {
5212 LinkManagerAction::SendPacket { raw, .. } => {
5213 let pkt = RawPacket::unpack(raw).ok()?;
5214 (pkt.context == constants::CONTEXT_RESOURCE_ADV).then_some(raw)
5215 }
5216 _ => None,
5217 })
5218 .expect("sender should emit a resource advertisement");
5219 let adv_pkt = RawPacket::unpack(adv_raw).unwrap();
5220 let plaintext = mgr
5221 .links
5222 .get(link_id)
5223 .unwrap()
5224 .engine
5225 .decrypt(&adv_pkt.data)
5226 .unwrap();
5227 rns_core::resource::ResourceAdvertisement::unpack(&plaintext).unwrap()
5228 }
5229
5230 fn deterministic_bytes(len: usize) -> Vec<u8> {
5231 let mut state = 0x1234_5678u32;
5232 (0..len)
5233 .map(|_| {
5234 state = state.wrapping_mul(1664525).wrapping_add(1013904223);
5235 (state >> 16) as u8
5236 })
5237 .collect()
5238 }
5239
5240 fn drive_link_manager_packets(
5241 init_mgr: &mut LinkManager,
5242 resp_mgr: &mut LinkManager,
5243 initial_actions: Vec<LinkManagerAction>,
5244 initial_source: char,
5245 rng: &mut dyn Rng,
5246 max_rounds: usize,
5247 ) -> (
5248 Option<Vec<u8>>,
5249 bool,
5250 Vec<(char, usize, usize)>,
5251 Vec<(char, String)>,
5252 usize,
5253 ) {
5254 let mut pending: Vec<(char, LinkManagerAction)> = initial_actions
5255 .into_iter()
5256 .map(|a| (initial_source, a))
5257 .collect();
5258 let mut rounds = 0;
5259 let mut received_data = None;
5260 let mut sender_completed = false;
5261 let mut progress = Vec::new();
5262 let mut failures = Vec::new();
5263
5264 while !pending.is_empty() && rounds < max_rounds {
5265 rounds += 1;
5266 let mut next = Vec::new();
5267 for (source, action) in pending.drain(..) {
5268 let LinkManagerAction::SendPacket { raw, .. } = action else {
5269 continue;
5270 };
5271 let pkt = RawPacket::unpack(&raw).unwrap();
5272 let target_actions = if source == 'i' {
5273 resp_mgr.handle_local_delivery(
5274 pkt.destination_hash,
5275 &raw,
5276 pkt.packet_hash,
5277 rns_core::transport::types::InterfaceId(0),
5278 rng,
5279 )
5280 } else {
5281 init_mgr.handle_local_delivery(
5282 pkt.destination_hash,
5283 &raw,
5284 pkt.packet_hash,
5285 rns_core::transport::types::InterfaceId(0),
5286 rng,
5287 )
5288 };
5289 let target_source = if source == 'i' { 'r' } else { 'i' };
5290 for action in &target_actions {
5291 match action {
5292 LinkManagerAction::ResourceReceived { data, .. } => {
5293 received_data = Some(data.clone());
5294 }
5295 LinkManagerAction::ResourceCompleted { .. } => {
5296 sender_completed = true;
5297 }
5298 LinkManagerAction::ResourceProgress {
5299 received, total, ..
5300 } => {
5301 progress.push((target_source, *received, *total));
5302 }
5303 LinkManagerAction::ResourceFailed { error, .. } => {
5304 failures.push((target_source, error.clone()));
5305 }
5306 _ => {}
5307 }
5308 }
5309 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
5310 }
5311 pending = next;
5312 }
5313
5314 (received_data, sender_completed, progress, failures, rounds)
5315 }
5316
5317 #[test]
5318 fn test_send_resource_auto_compress_option_controls_adv_flag() {
5319 let data = vec![0x41; 2048];
5320
5321 let (mut compressed_mgr, _resp_mgr, link_id) = setup_active_link();
5322 let mut rng = OsRng;
5323 let actions =
5324 compressed_mgr.send_resource_with_auto_compress(&link_id, &data, None, true, &mut rng);
5325 let adv = first_resource_advertisement(&compressed_mgr, &link_id, &actions);
5326 assert!(
5327 adv.flags.compressed,
5328 "compressible resource should compress"
5329 );
5330
5331 let (mut plain_mgr, _resp_mgr, link_id) = setup_active_link();
5332 let mut rng = OsRng;
5333 let actions =
5334 plain_mgr.send_resource_with_auto_compress(&link_id, &data, None, false, &mut rng);
5335 let adv = first_resource_advertisement(&plain_mgr, &link_id, &actions);
5336 assert!(
5337 !adv.flags.compressed,
5338 "auto_compress=false should keep resource uncompressed"
5339 );
5340 }
5341
5342 #[test]
5343 fn test_send_resource_on_inactive_link() {
5344 let mut mgr = LinkManager::new();
5345 let mut rng = OsRng;
5346 let dummy_sig = [0xAA; 32];
5347 let (link_id, _) =
5348 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
5349
5350 let actions = mgr.send_resource(&link_id, b"data", None, &mut rng);
5352 assert!(actions.is_empty(), "Cannot send resource on inactive link");
5353 }
5354
5355 #[test]
5356 fn test_send_resource_without_session_key_uses_encrypt_fallback_path() {
5357 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5358 let mut rng = OsRng;
5359 init_mgr
5360 .links
5361 .get_mut(&link_id)
5362 .unwrap()
5363 .engine
5364 .clear_session_for_testing();
5365
5366 let actions = init_mgr.send_resource(&link_id, b"data", None, &mut rng);
5367
5368 assert!(
5369 actions.is_empty(),
5370 "without a session key, no advertisement should be emitted"
5371 );
5372 assert_eq!(
5373 init_mgr
5374 .links
5375 .get(&link_id)
5376 .map(|managed| managed.outgoing_resources.len()),
5377 Some(1)
5378 );
5379 }
5380
5381 #[test]
5382 fn test_resource_adv_rejected_by_accept_none() {
5383 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5384 let mut rng = OsRng;
5385
5386 let data = vec![0xCD; 100];
5389 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5390
5391 for action in &adv_actions {
5393 if let LinkManagerAction::SendPacket { raw, .. } = action {
5394 let pkt = RawPacket::unpack(raw).unwrap();
5395 let resp_actions = resp_mgr.handle_local_delivery(
5396 pkt.destination_hash,
5397 raw,
5398 pkt.packet_hash,
5399 rns_core::transport::types::InterfaceId(0),
5400 &mut rng,
5401 );
5402 let has_resource_received = resp_actions
5404 .iter()
5405 .any(|a| matches!(a, LinkManagerAction::ResourceReceived { .. }));
5406 assert!(
5407 !has_resource_received,
5408 "AcceptNone should not accept resource"
5409 );
5410 }
5411 }
5412 }
5413
5414 #[test]
5415 fn test_resource_adv_accepted_by_accept_all() {
5416 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5417 let mut rng = OsRng;
5418
5419 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
5421
5422 let data = vec![0xCD; 100];
5424 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5425
5426 for action in &adv_actions {
5428 if let LinkManagerAction::SendPacket { raw, .. } = action {
5429 let pkt = RawPacket::unpack(raw).unwrap();
5430 let resp_actions = resp_mgr.handle_local_delivery(
5431 pkt.destination_hash,
5432 raw,
5433 pkt.packet_hash,
5434 rns_core::transport::types::InterfaceId(0),
5435 &mut rng,
5436 );
5437 let has_send = resp_actions
5439 .iter()
5440 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
5441 assert!(has_send, "AcceptAll should accept and request parts");
5442 }
5443 }
5444 }
5445
5446 #[test]
5447 fn test_resource_accept_app_query() {
5448 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5449 let mut rng = OsRng;
5450
5451 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
5453
5454 let data = vec![0xCD; 100];
5456 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5457
5458 let mut got_query = false;
5460 for action in &adv_actions {
5461 if let LinkManagerAction::SendPacket { raw, .. } = action {
5462 let pkt = RawPacket::unpack(raw).unwrap();
5463 let resp_actions = resp_mgr.handle_local_delivery(
5464 pkt.destination_hash,
5465 raw,
5466 pkt.packet_hash,
5467 rns_core::transport::types::InterfaceId(0),
5468 &mut rng,
5469 );
5470 for a in &resp_actions {
5471 if matches!(a, LinkManagerAction::ResourceAcceptQuery { .. }) {
5472 got_query = true;
5473 }
5474 }
5475 }
5476 }
5477 assert!(got_query, "AcceptApp should emit ResourceAcceptQuery");
5478 }
5479
5480 #[test]
5481 fn test_resource_accept_app_accept() {
5482 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5483 let mut rng = OsRng;
5484
5485 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
5486
5487 let data = vec![0xCD; 100];
5488 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5489
5490 for action in &adv_actions {
5491 if let LinkManagerAction::SendPacket { raw, .. } = action {
5492 let pkt = RawPacket::unpack(raw).unwrap();
5493 let resp_actions = resp_mgr.handle_local_delivery(
5494 pkt.destination_hash,
5495 raw,
5496 pkt.packet_hash,
5497 rns_core::transport::types::InterfaceId(0),
5498 &mut rng,
5499 );
5500 for a in &resp_actions {
5501 if let LinkManagerAction::ResourceAcceptQuery {
5502 link_id: lid,
5503 resource_hash,
5504 ..
5505 } = a
5506 {
5507 let accept_actions =
5509 resp_mgr.accept_resource(lid, resource_hash, true, &mut rng);
5510 let has_send = accept_actions
5512 .iter()
5513 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
5514 assert!(
5515 has_send,
5516 "Accepting resource should produce request for parts"
5517 );
5518 }
5519 }
5520 }
5521 }
5522 }
5523
5524 #[test]
5525 fn test_resource_accept_app_reject() {
5526 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5527 let mut rng = OsRng;
5528
5529 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
5530
5531 let data = vec![0xCD; 100];
5532 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5533
5534 for action in &adv_actions {
5535 if let LinkManagerAction::SendPacket { raw, .. } = action {
5536 let pkt = RawPacket::unpack(raw).unwrap();
5537 let resp_actions = resp_mgr.handle_local_delivery(
5538 pkt.destination_hash,
5539 raw,
5540 pkt.packet_hash,
5541 rns_core::transport::types::InterfaceId(0),
5542 &mut rng,
5543 );
5544 for a in &resp_actions {
5545 if let LinkManagerAction::ResourceAcceptQuery {
5546 link_id: lid,
5547 resource_hash,
5548 ..
5549 } = a
5550 {
5551 let reject_actions =
5553 resp_mgr.accept_resource(lid, resource_hash, false, &mut rng);
5554 let has_resource_received = reject_actions
5557 .iter()
5558 .any(|a| matches!(a, LinkManagerAction::ResourceReceived { .. }));
5559 assert!(!has_resource_received);
5560 }
5561 }
5562 }
5563 }
5564 }
5565
5566 #[test]
5567 fn test_resource_full_transfer() {
5568 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5569 let mut rng = OsRng;
5570
5571 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
5573
5574 let original_data = b"Hello, Resource Transfer!".to_vec();
5576 let adv_actions = init_mgr.send_resource(&link_id, &original_data, None, &mut rng);
5577
5578 let mut pending: Vec<(char, LinkManagerAction)> =
5581 adv_actions.into_iter().map(|a| ('i', a)).collect();
5582 let mut rounds = 0;
5583 let max_rounds = 50;
5584 let mut resource_received = false;
5585 let mut sender_completed = false;
5586
5587 while !pending.is_empty() && rounds < max_rounds {
5588 rounds += 1;
5589 let mut next: Vec<(char, LinkManagerAction)> = Vec::new();
5590
5591 for (source, action) in pending.drain(..) {
5592 if let LinkManagerAction::SendPacket { raw, .. } = action {
5593 let pkt = RawPacket::unpack(&raw).unwrap();
5594
5595 let target_actions = if source == 'i' {
5597 resp_mgr.handle_local_delivery(
5598 pkt.destination_hash,
5599 &raw,
5600 pkt.packet_hash,
5601 rns_core::transport::types::InterfaceId(0),
5602 &mut rng,
5603 )
5604 } else {
5605 init_mgr.handle_local_delivery(
5606 pkt.destination_hash,
5607 &raw,
5608 pkt.packet_hash,
5609 rns_core::transport::types::InterfaceId(0),
5610 &mut rng,
5611 )
5612 };
5613
5614 let target_source = if source == 'i' { 'r' } else { 'i' };
5615 for a in &target_actions {
5616 match a {
5617 LinkManagerAction::ResourceReceived { data, .. } => {
5618 assert_eq!(*data, original_data);
5619 resource_received = true;
5620 }
5621 LinkManagerAction::ResourceCompleted { .. } => {
5622 sender_completed = true;
5623 }
5624 _ => {}
5625 }
5626 }
5627 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
5628 }
5629 }
5630 pending = next;
5631 }
5632
5633 assert!(
5634 resource_received,
5635 "Responder should receive resource data (rounds={})",
5636 rounds
5637 );
5638 assert!(
5639 sender_completed,
5640 "Sender should get completion proof (rounds={})",
5641 rounds
5642 );
5643 }
5644
5645 #[test]
5646 fn test_split_resource_advertisement_and_progress_entries() {
5647 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5648 let mut rng = OsRng;
5649 let data = deterministic_bytes(constants::RESOURCE_MAX_EFFICIENT_SIZE + 1024);
5650
5651 let actions =
5652 init_mgr.send_resource_with_auto_compress(&link_id, &data, None, false, &mut rng);
5653 let adv = first_resource_advertisement(&init_mgr, &link_id, &actions);
5654
5655 assert!(adv.flags.split);
5656 assert_eq!(adv.segment_index, 1);
5657 assert_eq!(adv.total_segments, 2);
5658 assert_eq!(adv.data_size, data.len() as u64);
5659
5660 let managed = init_mgr.links.get(&link_id).unwrap();
5661 assert_eq!(managed.outgoing_splits.len(), 1);
5662 assert_eq!(
5663 managed
5664 .outgoing_resources
5665 .iter()
5666 .filter(|sender| sender.flags.split)
5667 .count(),
5668 2
5669 );
5670
5671 let entries = init_mgr.resource_entries();
5672 assert_eq!(entries.len(), 1);
5673 assert_eq!(entries[0].direction, "outgoing");
5674 assert!(entries[0].total_parts > managed.outgoing_resources[0].total_parts());
5675 assert_eq!(entries[0].transferred_parts, 0);
5676 }
5677
5678 #[test]
5679 fn test_split_resource_full_transfer_and_monotonic_progress() {
5680 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5681 let mut rng = OsRng;
5682 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
5683
5684 let original_data = deterministic_bytes(constants::RESOURCE_MAX_EFFICIENT_SIZE + 2048);
5685 let initial_actions = init_mgr.send_resource_with_auto_compress(
5686 &link_id,
5687 &original_data,
5688 None,
5689 false,
5690 &mut rng,
5691 );
5692
5693 let (received_data, sender_completed, progress, failures, rounds) =
5694 drive_link_manager_packets(
5695 &mut init_mgr,
5696 &mut resp_mgr,
5697 initial_actions,
5698 'i',
5699 &mut rng,
5700 10_000,
5701 );
5702
5703 assert!(
5704 received_data.as_ref().is_some_and(|data| data == &original_data),
5705 "split transfer did not deliver payload in {rounds} rounds; sender_completed={sender_completed}; failures={failures:?}; last_progress={:?}; init_entries={:?}; resp_entries={:?}",
5706 progress.last(),
5707 init_mgr.resource_entries(),
5708 resp_mgr.resource_entries()
5709 );
5710 assert!(
5711 sender_completed,
5712 "sender did not complete in {rounds} rounds"
5713 );
5714 assert!(
5715 progress
5716 .iter()
5717 .any(|(_, received, total)| received == total),
5718 "expected final progress update"
5719 );
5720
5721 let mut init_last = 0;
5722 let mut resp_last = 0;
5723 for (side, received, total) in progress {
5724 assert!(received <= total);
5725 match side {
5726 'i' => {
5727 assert!(
5728 received >= init_last,
5729 "initiator progress regressed from {init_last} to {received}"
5730 );
5731 init_last = received;
5732 }
5733 'r' => {
5734 assert!(
5735 received >= resp_last,
5736 "responder progress regressed from {resp_last} to {received}"
5737 );
5738 resp_last = received;
5739 }
5740 _ => unreachable!(),
5741 }
5742 }
5743 }
5744
5745 #[test]
5746 fn test_split_resource_accept_app_queries_only_first_segment() {
5747 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5748 let mut rng = OsRng;
5749 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
5750
5751 let original_data = deterministic_bytes(constants::RESOURCE_MAX_EFFICIENT_SIZE + 1024);
5752 let adv_actions = init_mgr.send_resource_with_auto_compress(
5753 &link_id,
5754 &original_data,
5755 None,
5756 false,
5757 &mut rng,
5758 );
5759 let adv_raw = extract_any_send_packet(&adv_actions);
5760 let adv_pkt = RawPacket::unpack(&adv_raw).unwrap();
5761 let query_actions = resp_mgr.handle_local_delivery(
5762 adv_pkt.destination_hash,
5763 &adv_raw,
5764 adv_pkt.packet_hash,
5765 rns_core::transport::types::InterfaceId(0),
5766 &mut rng,
5767 );
5768
5769 let queries: Vec<_> = query_actions
5770 .iter()
5771 .filter_map(|action| match action {
5772 LinkManagerAction::ResourceAcceptQuery { resource_hash, .. } => {
5773 Some(resource_hash.clone())
5774 }
5775 _ => None,
5776 })
5777 .collect();
5778 assert_eq!(queries.len(), 1);
5779
5780 let accept_actions = resp_mgr.accept_resource(&link_id, &queries[0], true, &mut rng);
5781 let (received_data, sender_completed, _progress, failures, rounds) =
5782 drive_link_manager_packets(
5783 &mut init_mgr,
5784 &mut resp_mgr,
5785 accept_actions,
5786 'r',
5787 &mut rng,
5788 10_000,
5789 );
5790
5791 assert!(
5792 failures.is_empty(),
5793 "split AcceptApp transfer failed: {failures:?}"
5794 );
5795 assert!(
5796 received_data
5797 .as_ref()
5798 .is_some_and(|data| data == &original_data),
5799 "split AcceptApp transfer did not deliver in {rounds} rounds"
5800 );
5801 assert!(sender_completed);
5802 }
5803
5804 #[test]
5805 fn test_resource_cancel_icl() {
5806 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5807 let mut rng = OsRng;
5808
5809 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
5810
5811 let data = vec![0xAB; 2000];
5813 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
5814
5815 for action in &adv_actions {
5817 if let LinkManagerAction::SendPacket { raw, .. } = action {
5818 let pkt = RawPacket::unpack(raw).unwrap();
5819 resp_mgr.handle_local_delivery(
5820 pkt.destination_hash,
5821 raw,
5822 pkt.packet_hash,
5823 rns_core::transport::types::InterfaceId(0),
5824 &mut rng,
5825 );
5826 }
5827 }
5828
5829 assert!(!resp_mgr
5831 .links
5832 .get(&link_id)
5833 .unwrap()
5834 .incoming_resources
5835 .is_empty());
5836
5837 let icl_actions = resp_mgr.handle_resource_icl(&link_id);
5839
5840 let has_failed = icl_actions
5842 .iter()
5843 .any(|a| matches!(a, LinkManagerAction::ResourceFailed { .. }));
5844 assert!(has_failed, "ICL should produce ResourceFailed");
5845 }
5846
5847 #[test]
5848 fn test_resource_cancel_rcl() {
5849 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5850 let mut rng = OsRng;
5851
5852 let data = vec![0xAB; 2000];
5854 init_mgr.send_resource(&link_id, &data, None, &mut rng);
5855
5856 assert!(!init_mgr
5858 .links
5859 .get(&link_id)
5860 .unwrap()
5861 .outgoing_resources
5862 .is_empty());
5863
5864 let rcl_actions = init_mgr.handle_resource_rcl(&link_id);
5866
5867 let has_failed = rcl_actions
5868 .iter()
5869 .any(|a| matches!(a, LinkManagerAction::ResourceFailed { .. }));
5870 assert!(has_failed, "RCL should produce ResourceFailed");
5871 }
5872
5873 #[test]
5874 fn test_cancel_all_resources_clears_active_transfers() {
5875 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5876 let mut rng = OsRng;
5877
5878 let actions = init_mgr.send_resource(&link_id, b"resource body", None, &mut rng);
5879 assert!(!actions.is_empty());
5880 assert_eq!(init_mgr.resource_transfer_count(), 1);
5881
5882 let cancel_actions = init_mgr.cancel_all_resources(&mut rng);
5883
5884 assert_eq!(init_mgr.resource_transfer_count(), 0);
5885 assert!(cancel_actions.iter().any(|action| match action {
5886 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
5887 .map(|packet| packet.context == constants::CONTEXT_RESOURCE_ICL)
5888 .unwrap_or(false),
5889 _ => false,
5890 }));
5891 }
5892
5893 #[test]
5894 fn receiver_cancel_sends_rcl_while_active_and_closed_cleanup_is_local_only() {
5895 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5896 let mut rng = OsRng;
5897 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
5898
5899 let advertisement = init_mgr.send_resource(&link_id, b"incoming body", None, &mut rng);
5900 let raw = extract_any_send_packet(&advertisement);
5901 let packet = RawPacket::unpack(&raw).unwrap();
5902 resp_mgr.handle_local_delivery(
5903 packet.destination_hash,
5904 &raw,
5905 packet.packet_hash,
5906 rns_core::transport::types::InterfaceId(0),
5907 &mut rng,
5908 );
5909 assert_eq!(resp_mgr.resource_transfer_count(), 1);
5910 let active_cancel = resp_mgr.cancel_all_resources(&mut rng);
5911 assert_eq!(resp_mgr.resource_transfer_count(), 0);
5912 assert!(active_cancel.iter().any(|action| match action {
5913 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
5914 .map(|packet| packet.context == constants::CONTEXT_RESOURCE_RCL)
5915 .unwrap_or(false),
5916 _ => false,
5917 }));
5918
5919 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
5920 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptApp);
5921 let advertisement = init_mgr.send_resource(&link_id, b"second body", None, &mut rng);
5922 let raw = extract_any_send_packet(&advertisement);
5923 let packet = RawPacket::unpack(&raw).unwrap();
5924 resp_mgr.handle_local_delivery(
5925 packet.destination_hash,
5926 &raw,
5927 packet.packet_hash,
5928 rns_core::transport::types::InterfaceId(0),
5929 &mut rng,
5930 );
5931 resp_mgr
5932 .links
5933 .get_mut(&link_id)
5934 .unwrap()
5935 .engine
5936 .handle_teardown();
5937 let closed_cancel = resp_mgr.cancel_all_resources(&mut rng);
5938 assert_eq!(resp_mgr.resource_transfer_count(), 0);
5939 assert!(!closed_cancel
5940 .iter()
5941 .any(|action| matches!(action, LinkManagerAction::SendPacket { .. })));
5942 }
5943
5944 #[test]
5945 fn stale_sender_cancel_cleans_up_without_icl() {
5946 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5947 let mut rng = OsRng;
5948 init_mgr.send_resource(&link_id, b"stale transfer", None, &mut rng);
5949 init_mgr
5950 .links
5951 .get_mut(&link_id)
5952 .unwrap()
5953 .engine
5954 .tick(time::now() + 100_000.0);
5955 assert_eq!(init_mgr.link_state(&link_id), Some(LinkState::Stale));
5956
5957 let actions = init_mgr.cancel_all_resources(&mut rng);
5958 assert_eq!(init_mgr.resource_transfer_count(), 0);
5959 assert!(!actions
5960 .iter()
5961 .any(|action| matches!(action, LinkManagerAction::SendPacket { .. })));
5962 }
5963
5964 #[test]
5965 fn test_resource_tick_cleans_up() {
5966 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
5967 let mut rng = OsRng;
5968
5969 let data = vec![0xAB; 100];
5970 init_mgr.send_resource(&link_id, &data, None, &mut rng);
5971
5972 assert!(!init_mgr
5973 .links
5974 .get(&link_id)
5975 .unwrap()
5976 .outgoing_resources
5977 .is_empty());
5978
5979 init_mgr.handle_resource_rcl(&link_id);
5981
5982 init_mgr.tick(&mut rng);
5984
5985 assert!(
5986 init_mgr
5987 .links
5988 .get(&link_id)
5989 .unwrap()
5990 .outgoing_resources
5991 .is_empty(),
5992 "Tick should clean up completed/failed outgoing resources"
5993 );
5994 }
5995
5996 #[test]
5997 fn test_build_link_packet() {
5998 let (init_mgr, _resp_mgr, link_id) = setup_active_link();
5999
6000 let actions =
6001 init_mgr.build_link_packet(&link_id, constants::CONTEXT_RESOURCE, b"test data");
6002 assert_eq!(actions.len(), 1);
6003 if let LinkManagerAction::SendPacket { raw, dest_type, .. } = &actions[0] {
6004 let pkt = RawPacket::unpack(raw).unwrap();
6005 assert_eq!(pkt.context, constants::CONTEXT_RESOURCE);
6006 assert_eq!(*dest_type, constants::DESTINATION_LINK);
6007 } else {
6008 panic!("Expected SendPacket");
6009 }
6010 }
6011
6012 #[test]
6013 fn link_traffic_accounting_uses_ciphertext_data_once() {
6014 let (mut initiator, mut responder, link_id) = setup_active_link();
6015 let before_tx = initiator.links[&link_id].engine.tx_packets();
6016 let before_tx_bytes = initiator.links[&link_id].engine.tx_bytes();
6017 let before_rx = responder.links[&link_id].engine.rx_packets();
6018 let before_rx_bytes = responder.links[&link_id].engine.rx_bytes();
6019 let actions = initiator.send_on_link(
6020 &link_id,
6021 b"short plaintext",
6022 constants::CONTEXT_NONE,
6023 &mut OsRng,
6024 );
6025 let raw = actions
6026 .iter()
6027 .find_map(|action| match action {
6028 LinkManagerAction::SendPacket { raw, .. } => Some(raw.clone()),
6029 _ => None,
6030 })
6031 .unwrap();
6032 let packet = RawPacket::unpack(&raw).unwrap();
6033 assert!(packet.data.len() > b"short plaintext".len());
6034
6035 initiator.record_outbound_packet(&packet);
6036 assert_eq!(initiator.links[&link_id].engine.tx_packets(), before_tx + 1);
6037 assert_eq!(
6038 initiator.links[&link_id].engine.tx_bytes(),
6039 before_tx_bytes + packet.data.len() as u64
6040 );
6041
6042 responder.handle_local_delivery(
6043 link_id,
6044 &raw,
6045 packet.packet_hash,
6046 rns_core::transport::types::InterfaceId(1),
6047 &mut OsRng,
6048 );
6049 assert_eq!(responder.links[&link_id].engine.rx_packets(), before_rx + 1);
6050 assert_eq!(
6051 responder.links[&link_id].engine.rx_bytes(),
6052 before_rx_bytes + packet.data.len() as u64
6053 );
6054 }
6055
6056 #[test]
6057 fn test_build_link_packet_returns_empty_when_mtu_too_small() {
6058 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
6059 init_mgr.set_link_mtu(&link_id, 84);
6060
6061 let actions =
6062 init_mgr.build_link_packet(&link_id, constants::CONTEXT_RESOURCE, &[0xAA; 200]);
6063 assert!(actions.is_empty(), "oversized packet should not be built");
6064 }
6065
6066 #[test]
6067 fn test_process_resource_actions_encrypted_variants_drop_on_encrypt_failure() {
6068 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
6069 let mut rng = OsRng;
6070 init_mgr
6071 .links
6072 .get_mut(&link_id)
6073 .unwrap()
6074 .engine
6075 .clear_session_for_testing();
6076
6077 let cases = vec![
6078 ResourceAction::SendAdvertisement(vec![1, 2, 3]),
6079 ResourceAction::SendRequest(vec![4, 5, 6]),
6080 ResourceAction::SendHmu(vec![7, 8, 9]),
6081 ResourceAction::SendCancelInitiator(vec![13, 14, 15]),
6082 ResourceAction::SendCancelReceiver(vec![16, 17, 18]),
6083 ];
6084
6085 for action in cases {
6086 let out = init_mgr.process_resource_actions(&link_id, vec![action], &mut rng);
6087 assert!(
6088 out.is_empty(),
6089 "encrypt failure should suppress packet emission"
6090 );
6091 }
6092 }
6093
6094 #[test]
6095 fn test_resource_proof_is_raw_proof_packet() {
6096 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
6097 let mut rng = OsRng;
6098 let proof = vec![0x22; 64];
6099
6100 let out = init_mgr.process_resource_actions(
6101 &link_id,
6102 vec![ResourceAction::SendProof(proof.clone())],
6103 &mut rng,
6104 );
6105 let raw = extract_any_send_packet(&out);
6106 let pkt = RawPacket::unpack(&raw).unwrap();
6107
6108 assert_eq!(pkt.flags.packet_type, constants::PACKET_TYPE_PROOF);
6109 assert_eq!(pkt.context, constants::CONTEXT_RESOURCE_PRF);
6110 assert_eq!(pkt.data, proof);
6111 }
6112
6113 #[test]
6118 fn test_channel_message_delivery() {
6119 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
6120 let mut rng = OsRng;
6121
6122 let chan_actions = init_mgr
6124 .send_channel_message(&link_id, 42, b"channel data", &mut rng)
6125 .expect("active link channel send should succeed");
6126 assert!(!chan_actions.is_empty());
6127
6128 let mut got_channel_msg = false;
6130 for action in &chan_actions {
6131 if let LinkManagerAction::SendPacket { raw, .. } = action {
6132 let pkt = RawPacket::unpack(raw).unwrap();
6133 let resp_actions = resp_mgr.handle_local_delivery(
6134 pkt.destination_hash,
6135 raw,
6136 pkt.packet_hash,
6137 rns_core::transport::types::InterfaceId(0),
6138 &mut rng,
6139 );
6140 for a in &resp_actions {
6141 if let LinkManagerAction::ChannelMessageReceived {
6142 msgtype, payload, ..
6143 } = a
6144 {
6145 assert_eq!(*msgtype, 42);
6146 assert_eq!(*payload, b"channel data");
6147 got_channel_msg = true;
6148 }
6149 }
6150 }
6151 }
6152 assert!(got_channel_msg, "Responder should receive channel message");
6153 }
6154
6155 #[test]
6156 fn test_channel_send_drops_packet_when_encrypt_fails() {
6157 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
6158 let mut rng = OsRng;
6159 init_mgr
6160 .links
6161 .get_mut(&link_id)
6162 .unwrap()
6163 .engine
6164 .clear_session_for_testing();
6165
6166 let actions = init_mgr
6167 .send_channel_message(&link_id, 42, b"channel data", &mut rng)
6168 .expect("channel should still accept the message locally");
6169
6170 assert!(
6171 actions.is_empty(),
6172 "encrypt failure should suppress channel packet"
6173 );
6174 assert!(
6175 init_mgr
6176 .links
6177 .get(&link_id)
6178 .unwrap()
6179 .pending_channel_packets
6180 .is_empty(),
6181 "failed packet encryption must not track a pending channel proof"
6182 );
6183 }
6184
6185 #[test]
6186 fn test_channel_proof_reopens_send_window() {
6187 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
6188 let mut rng = OsRng;
6189
6190 init_mgr
6191 .send_channel_message(&link_id, 42, b"first", &mut rng)
6192 .expect("first send should succeed");
6193 init_mgr
6194 .send_channel_message(&link_id, 42, b"second", &mut rng)
6195 .expect("second send should succeed");
6196
6197 let err = init_mgr
6198 .send_channel_message(&link_id, 42, b"third", &mut rng)
6199 .expect_err("third send should hit the initial channel window");
6200 assert_eq!(err, "Channel is not ready to send");
6201
6202 let queued_packets = init_mgr
6203 .links
6204 .get(&link_id)
6205 .unwrap()
6206 .pending_channel_packets
6207 .clone();
6208 assert_eq!(queued_packets.len(), 2);
6209 for tracked_hash in queued_packets.keys().take(1) {
6210 let mut proof_data = Vec::with_capacity(96);
6211 proof_data.extend_from_slice(tracked_hash);
6212 proof_data.extend_from_slice(&[0x11; 64]);
6213 let flags = PacketFlags {
6214 header_type: constants::HEADER_1,
6215 context_flag: constants::FLAG_UNSET,
6216 transport_type: constants::TRANSPORT_BROADCAST,
6217 destination_type: constants::DESTINATION_LINK,
6218 packet_type: constants::PACKET_TYPE_PROOF,
6219 };
6220 let proof = RawPacket::pack(
6221 flags,
6222 0,
6223 &link_id,
6224 None,
6225 constants::CONTEXT_NONE,
6226 &proof_data,
6227 )
6228 .expect("proof packet should pack");
6229 let ack_actions = init_mgr.handle_local_delivery(
6230 link_id,
6231 &proof.raw,
6232 proof.packet_hash,
6233 rns_core::transport::types::InterfaceId(0),
6234 &mut rng,
6235 );
6236 assert!(
6237 ack_actions.is_empty(),
6238 "proof delivery should only update channel state"
6239 );
6240 }
6241
6242 init_mgr
6243 .send_channel_message(&link_id, 42, b"third", &mut rng)
6244 .expect("proof should free one channel slot");
6245 }
6246
6247 #[test]
6248 fn test_generic_link_data_delivery() {
6249 let (init_mgr, mut resp_mgr, link_id) = setup_active_link();
6250 let mut rng = OsRng;
6251
6252 let actions = init_mgr.send_on_link(&link_id, b"raw stuff", 0x42, &mut rng);
6254 assert_eq!(actions.len(), 1);
6255
6256 let raw = extract_any_send_packet(&actions);
6258 let pkt = RawPacket::unpack(&raw).unwrap();
6259 let resp_actions = resp_mgr.handle_local_delivery(
6260 pkt.destination_hash,
6261 &raw,
6262 pkt.packet_hash,
6263 rns_core::transport::types::InterfaceId(0),
6264 &mut rng,
6265 );
6266
6267 let has_data = resp_actions
6268 .iter()
6269 .any(|a| matches!(a, LinkManagerAction::LinkDataReceived { context: 0x42, .. }));
6270 assert!(
6271 has_data,
6272 "Responder should receive LinkDataReceived for unknown context"
6273 );
6274 }
6275
6276 #[test]
6277 fn test_invalid_encrypted_contexts_are_ignored() {
6278 let (_init_mgr, mut resp_mgr, link_id) = setup_active_link();
6279 let mut rng = OsRng;
6280 let contexts = [
6281 constants::CONTEXT_CHANNEL,
6282 constants::CONTEXT_REQUEST,
6283 constants::CONTEXT_RESPONSE,
6284 constants::CONTEXT_RESOURCE_ADV,
6285 constants::CONTEXT_RESOURCE_REQ,
6286 constants::CONTEXT_RESOURCE_HMU,
6287 constants::CONTEXT_RESOURCE_PRF,
6288 0x42,
6289 ];
6290
6291 for context in contexts {
6292 let flags = PacketFlags {
6293 header_type: constants::HEADER_1,
6294 context_flag: constants::FLAG_UNSET,
6295 transport_type: constants::TRANSPORT_BROADCAST,
6296 destination_type: constants::DESTINATION_LINK,
6297 packet_type: constants::PACKET_TYPE_DATA,
6298 };
6299 let pkt = RawPacket::pack(flags, 0, &link_id, None, context, b"invalid-ciphertext")
6300 .expect("test packet should pack");
6301 let actions = resp_mgr.handle_local_delivery(
6302 pkt.destination_hash,
6303 &pkt.raw,
6304 pkt.packet_hash,
6305 rns_core::transport::types::InterfaceId(0),
6306 &mut rng,
6307 );
6308 assert!(
6309 actions.is_empty(),
6310 "invalid ciphertext for context {context:#x} should be ignored"
6311 );
6312 }
6313 }
6314
6315 #[test]
6316 fn test_resource_part_without_matching_receiver_is_ignored() {
6317 let (_init_mgr, mut resp_mgr, link_id) = setup_active_link();
6318 let mut rng = OsRng;
6319 let flags = PacketFlags {
6320 header_type: constants::HEADER_1,
6321 context_flag: constants::FLAG_UNSET,
6322 transport_type: constants::TRANSPORT_BROADCAST,
6323 destination_type: constants::DESTINATION_LINK,
6324 packet_type: constants::PACKET_TYPE_DATA,
6325 };
6326 let pkt = RawPacket::pack(
6327 flags,
6328 0,
6329 &link_id,
6330 None,
6331 constants::CONTEXT_RESOURCE,
6332 b"orphan-part",
6333 )
6334 .expect("test packet should pack");
6335
6336 let actions = resp_mgr.handle_local_delivery(
6337 pkt.destination_hash,
6338 &pkt.raw,
6339 pkt.packet_hash,
6340 rns_core::transport::types::InterfaceId(0),
6341 &mut rng,
6342 );
6343
6344 assert!(actions.is_empty(), "orphan resource part should be ignored");
6345 }
6346
6347 #[test]
6348 fn test_response_delivery() {
6349 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
6350 let mut rng = OsRng;
6351
6352 resp_mgr.register_request_handler("/echo", None, |_link_id, _path, data, _remote| {
6354 Some(data.to_vec())
6355 });
6356
6357 let req_actions = init_mgr.send_request(&link_id, "/echo", b"\xc0", &mut rng); assert!(!req_actions.is_empty());
6360
6361 let req_raw = extract_any_send_packet(&req_actions);
6363 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
6364 let resp_actions = resp_mgr.handle_local_delivery(
6365 req_pkt.destination_hash,
6366 &req_raw,
6367 req_pkt.packet_hash,
6368 rns_core::transport::types::InterfaceId(0),
6369 &mut rng,
6370 );
6371 let has_resp_send = resp_actions
6372 .iter()
6373 .any(|a| matches!(a, LinkManagerAction::SendPacket { .. }));
6374 assert!(has_resp_send, "Handler should produce response");
6375
6376 let resp_raw = extract_any_send_packet(&resp_actions);
6378 let resp_pkt = RawPacket::unpack(&resp_raw).unwrap();
6379 let init_actions = init_mgr.handle_local_delivery(
6380 resp_pkt.destination_hash,
6381 &resp_raw,
6382 resp_pkt.packet_hash,
6383 rns_core::transport::types::InterfaceId(0),
6384 &mut rng,
6385 );
6386
6387 let has_response_received = init_actions
6388 .iter()
6389 .any(|a| matches!(a, LinkManagerAction::ResponseReceived { .. }));
6390 assert!(
6391 has_response_received,
6392 "Initiator should receive ResponseReceived"
6393 );
6394 }
6395
6396 #[test]
6397 fn test_large_response_uses_resource_fallback() {
6398 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
6399 let mut rng = OsRng;
6400
6401 let large_payload: Vec<u8> = (0..5000u32).map(|i| (i & 0xFF) as u8).collect();
6404 resp_mgr.register_request_handler("/large", None, {
6405 let large_payload = large_payload.clone();
6406 move |_link_id, _path, _data, _remote| Some(large_payload.clone())
6407 });
6408
6409 let req_actions = init_mgr.send_request(&link_id, "/large", b"\xc0", &mut rng);
6411 assert!(!req_actions.is_empty());
6412
6413 let req_raw = extract_any_send_packet(&req_actions);
6415 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
6416 let resp_actions = resp_mgr.handle_local_delivery(
6417 req_pkt.destination_hash,
6418 &req_raw,
6419 req_pkt.packet_hash,
6420 rns_core::transport::types::InterfaceId(0),
6421 &mut rng,
6422 );
6423
6424 let mut has_resource_adv = false;
6425 let mut has_direct_response = false;
6426 for action in &resp_actions {
6427 if let LinkManagerAction::SendPacket { raw, .. } = action {
6428 let pkt = RawPacket::unpack(raw).unwrap();
6429 if pkt.context == constants::CONTEXT_RESOURCE_ADV {
6430 has_resource_adv = true;
6431 }
6432 if pkt.context == constants::CONTEXT_RESPONSE {
6433 has_direct_response = true;
6434 }
6435 }
6436 }
6437
6438 assert!(
6439 has_resource_adv,
6440 "Large response should advertise a response resource"
6441 );
6442 assert!(
6443 !has_direct_response,
6444 "Large response should not use direct CONTEXT_RESPONSE packet"
6445 );
6446 }
6447
6448 #[test]
6449 fn test_send_management_response_without_session_key_uses_resource_fallback_path() {
6450 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
6451 let mut rng = OsRng;
6452 init_mgr
6453 .links
6454 .get_mut(&link_id)
6455 .unwrap()
6456 .engine
6457 .clear_session_for_testing();
6458
6459 let large_response: Vec<u8> = (0..5000u32).map(|i| (i & 0xFF) as u8).collect();
6460 let actions =
6461 init_mgr.send_management_response(&link_id, &[0x11; 16], &large_response, &mut rng);
6462
6463 assert!(
6464 actions.is_empty(),
6465 "without a session key, no response packets should be emitted"
6466 );
6467 assert_eq!(
6468 init_mgr
6469 .links
6470 .get(&link_id)
6471 .map(|managed| managed.outgoing_resources.len()),
6472 Some(1)
6473 );
6474 }
6475
6476 #[test]
6477 fn test_send_channel_message_on_no_channel() {
6478 let mut mgr = LinkManager::new();
6479 let mut rng = OsRng;
6480 let dummy_sig = [0xAA; 32];
6481 let (link_id, _) =
6482 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
6483
6484 let err = mgr
6486 .send_channel_message(&link_id, 1, b"test", &mut rng)
6487 .expect_err("pending link should reject channel send");
6488 assert_eq!(err, "link has no active channel");
6489 }
6490
6491 #[test]
6492 fn test_send_on_link_requires_active() {
6493 let mut mgr = LinkManager::new();
6494 let mut rng = OsRng;
6495 let dummy_sig = [0xAA; 32];
6496 let (link_id, _) =
6497 mgr.create_link(&[0x11; 16], &dummy_sig, 1, constants::MTU as u32, &mut rng);
6498
6499 let actions = mgr.send_on_link(&link_id, b"test", constants::CONTEXT_NONE, &mut rng);
6500 assert!(actions.is_empty(), "Cannot send on pending link");
6501 }
6502
6503 #[test]
6504 fn test_send_on_link_unknown_link() {
6505 let mgr = LinkManager::new();
6506 let mut rng = OsRng;
6507
6508 let actions = mgr.send_on_link(&[0xFF; 16], b"test", constants::CONTEXT_NONE, &mut rng);
6509 assert!(actions.is_empty());
6510 }
6511
6512 #[test]
6513 fn test_resource_full_transfer_large() {
6514 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
6515 let mut rng = OsRng;
6516
6517 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
6518
6519 let original_data: Vec<u8> = (0..2000u32)
6521 .map(|i| {
6522 let pos = i as usize;
6523 (pos ^ (pos >> 8) ^ (pos >> 16)) as u8
6524 })
6525 .collect();
6526
6527 let adv_actions = init_mgr.send_resource(&link_id, &original_data, None, &mut rng);
6528
6529 let mut pending: Vec<(char, LinkManagerAction)> =
6530 adv_actions.into_iter().map(|a| ('i', a)).collect();
6531 let mut rounds = 0;
6532 let max_rounds = 200;
6533 let mut resource_received = false;
6534 let mut sender_completed = false;
6535
6536 while !pending.is_empty() && rounds < max_rounds {
6537 rounds += 1;
6538 let mut next: Vec<(char, LinkManagerAction)> = Vec::new();
6539
6540 for (source, action) in pending.drain(..) {
6541 if let LinkManagerAction::SendPacket { raw, .. } = action {
6542 let pkt = match RawPacket::unpack(&raw) {
6543 Ok(p) => p,
6544 Err(_) => continue,
6545 };
6546
6547 let target_actions = if source == 'i' {
6548 resp_mgr.handle_local_delivery(
6549 pkt.destination_hash,
6550 &raw,
6551 pkt.packet_hash,
6552 rns_core::transport::types::InterfaceId(0),
6553 &mut rng,
6554 )
6555 } else {
6556 init_mgr.handle_local_delivery(
6557 pkt.destination_hash,
6558 &raw,
6559 pkt.packet_hash,
6560 rns_core::transport::types::InterfaceId(0),
6561 &mut rng,
6562 )
6563 };
6564
6565 let target_source = if source == 'i' { 'r' } else { 'i' };
6566 for a in &target_actions {
6567 match a {
6568 LinkManagerAction::ResourceReceived { data, .. } => {
6569 assert_eq!(*data, original_data);
6570 resource_received = true;
6571 }
6572 LinkManagerAction::ResourceCompleted { .. } => {
6573 sender_completed = true;
6574 }
6575 _ => {}
6576 }
6577 }
6578 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
6579 }
6580 }
6581 pending = next;
6582 }
6583
6584 assert!(
6585 resource_received,
6586 "Should receive large resource (rounds={})",
6587 rounds
6588 );
6589 assert!(
6590 sender_completed,
6591 "Sender should complete (rounds={})",
6592 rounds
6593 );
6594 }
6595
6596 #[test]
6597 fn test_resource_receiver_stores_original_advertisement_plaintext() {
6598 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
6599 let mut rng = OsRng;
6600
6601 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
6602
6603 let data = vec![0x41; 256];
6604 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
6605
6606 let adv_raw = adv_actions
6607 .iter()
6608 .find_map(|action| match action {
6609 LinkManagerAction::SendPacket { raw, .. } => {
6610 let pkt = RawPacket::unpack(raw).ok()?;
6611 (pkt.context == constants::CONTEXT_RESOURCE_ADV).then_some(raw.clone())
6612 }
6613 _ => None,
6614 })
6615 .expect("sender should emit a resource advertisement");
6616
6617 let adv_pkt = RawPacket::unpack(&adv_raw).unwrap();
6618 let adv_plaintext = resp_mgr
6619 .links
6620 .get(&link_id)
6621 .unwrap()
6622 .engine
6623 .decrypt(&adv_pkt.data)
6624 .unwrap();
6625
6626 let _resp_actions = resp_mgr.handle_local_delivery(
6627 adv_pkt.destination_hash,
6628 &adv_raw,
6629 adv_pkt.packet_hash,
6630 rns_core::transport::types::InterfaceId(0),
6631 &mut rng,
6632 );
6633
6634 let receiver = resp_mgr
6635 .links
6636 .get(&link_id)
6637 .and_then(|managed| managed.incoming_resources.first())
6638 .expect("advertisement should create an incoming receiver");
6639 assert_eq!(receiver.advertisement_packet, adv_plaintext);
6640 assert_eq!(
6641 receiver.max_decompressed_size,
6642 constants::RESOURCE_AUTO_COMPRESS_MAX_SIZE
6643 );
6644 }
6645
6646 #[test]
6647 fn test_corrupt_compressed_resource_rejects_and_tears_down_link() {
6648 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
6649 let mut rng = OsRng;
6650
6651 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
6652
6653 let data = vec![b'A'; 4096];
6654 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
6655
6656 let mut request_actions = Vec::new();
6657 for action in &adv_actions {
6658 let LinkManagerAction::SendPacket { raw, .. } = action else {
6659 continue;
6660 };
6661 let pkt = RawPacket::unpack(raw).unwrap();
6662 let actions = resp_mgr.handle_local_delivery(
6663 pkt.destination_hash,
6664 raw,
6665 pkt.packet_hash,
6666 rns_core::transport::types::InterfaceId(0),
6667 &mut rng,
6668 );
6669 request_actions.extend(actions);
6670 }
6671
6672 {
6673 let receiver = resp_mgr
6674 .links
6675 .get_mut(&link_id)
6676 .and_then(|managed| managed.incoming_resources.first_mut())
6677 .expect("receiver should exist after advertisement");
6678 assert!(receiver.flags.compressed, "test data should be compressed");
6679 receiver.max_decompressed_size = 64;
6680 }
6681
6682 let mut responder_actions = Vec::new();
6683 for action in request_actions {
6684 let LinkManagerAction::SendPacket { raw, .. } = action else {
6685 continue;
6686 };
6687 let pkt = RawPacket::unpack(&raw).unwrap();
6688 let actions = init_mgr.handle_local_delivery(
6689 pkt.destination_hash,
6690 &raw,
6691 pkt.packet_hash,
6692 rns_core::transport::types::InterfaceId(0),
6693 &mut rng,
6694 );
6695
6696 for action in actions {
6697 let LinkManagerAction::SendPacket { raw, .. } = &action else {
6698 continue;
6699 };
6700 let pkt = RawPacket::unpack(raw).unwrap();
6701 let delivered = resp_mgr.handle_local_delivery(
6702 pkt.destination_hash,
6703 raw,
6704 pkt.packet_hash,
6705 rns_core::transport::types::InterfaceId(0),
6706 &mut rng,
6707 );
6708 responder_actions.extend(delivered);
6709 }
6710 }
6711
6712 assert!(
6713 responder_actions.iter().any(|action| matches!(
6714 action,
6715 LinkManagerAction::ResourceFailed { error, .. }
6716 if error == "Resource too large"
6717 )),
6718 "corrupt oversized resource should fail with TooLarge"
6719 );
6720 assert!(
6721 responder_actions.iter().any(|action| matches!(
6722 action,
6723 LinkManagerAction::LinkClosed { link_id: closed_id, .. } if *closed_id == link_id
6724 )),
6725 "corrupt oversized resource should tear down the link"
6726 );
6727 assert!(
6728 responder_actions.iter().any(|action| match action {
6729 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
6730 .map(|pkt| pkt.context == constants::CONTEXT_RESOURCE_RCL)
6731 .unwrap_or(false),
6732 _ => false,
6733 }),
6734 "corrupt oversized resource should send a receiver cancel/reject packet"
6735 );
6736 assert_eq!(
6737 resp_mgr
6738 .links
6739 .get(&link_id)
6740 .map(|managed| managed.engine.state()),
6741 Some(LinkState::Closed)
6742 );
6743 }
6744
6745 #[test]
6746 fn test_resource_hmu_timeout_extension_in_link_manager_flow() {
6747 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
6748 let mut rng = OsRng;
6749
6750 resp_mgr.set_resource_strategy(&link_id, ResourceStrategy::AcceptAll);
6751
6752 let mut state = 0x1234_5678u32;
6755 let data: Vec<u8> = (0..50000)
6756 .map(|_| {
6757 state = state.wrapping_mul(1664525).wrapping_add(1013904223);
6758 (state >> 16) as u8
6759 })
6760 .collect();
6761 let adv_actions = init_mgr.send_resource(&link_id, &data, None, &mut rng);
6762 let mut pending: Vec<(char, LinkManagerAction)> =
6763 adv_actions.into_iter().map(|a| ('i', a)).collect();
6764
6765 let mut rounds = 0;
6766
6767 while rounds < 300 {
6770 rounds += 1;
6771 let mut next: Vec<(char, LinkManagerAction)> = Vec::new();
6772
6773 for (source, action) in pending.drain(..) {
6774 let LinkManagerAction::SendPacket { raw, .. } = action else {
6775 continue;
6776 };
6777
6778 let pkt = match RawPacket::unpack(&raw) {
6779 Ok(p) => p,
6780 Err(_) => continue,
6781 };
6782
6783 let target_actions = if source == 'i' {
6784 resp_mgr.handle_local_delivery(
6785 pkt.destination_hash,
6786 &raw,
6787 pkt.packet_hash,
6788 rns_core::transport::types::InterfaceId(0),
6789 &mut rng,
6790 )
6791 } else {
6792 init_mgr.handle_local_delivery(
6793 pkt.destination_hash,
6794 &raw,
6795 pkt.packet_hash,
6796 rns_core::transport::types::InterfaceId(0),
6797 &mut rng,
6798 )
6799 };
6800
6801 let target_source = if source == 'i' { 'r' } else { 'i' };
6802 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
6803 }
6804
6805 if resp_mgr
6806 .links
6807 .get(&link_id)
6808 .and_then(|managed| managed.incoming_resources.first())
6809 .is_some_and(|receiver| receiver.waiting_for_hmu)
6810 {
6811 break;
6812 }
6813
6814 pending = next;
6815 }
6816
6817 assert!(
6818 resp_mgr
6819 .links
6820 .get(&link_id)
6821 .and_then(|managed| managed.incoming_resources.first())
6822 .is_some_and(|receiver| receiver.waiting_for_hmu),
6823 "expected receiver to reach a live HMU wait state"
6824 );
6825
6826 let prime_actions = {
6829 let managed = resp_mgr.links.get_mut(&link_id).unwrap();
6830 let receiver = managed.incoming_resources.first_mut().unwrap();
6831 let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
6832 managed.engine.decrypt(ciphertext).map_err(|_| ())
6833 };
6834 receiver.tick(
6835 receiver.last_activity + 0.0001,
6836 &decrypt_fn,
6837 &Bzip2Compressor,
6838 )
6839 };
6840 assert!(
6841 !prime_actions
6842 .iter()
6843 .any(|a| matches!(a, ResourceAction::SendRequest(_))),
6844 "fresh HMU wait state should not immediately emit a retry request"
6845 );
6846
6847 let (late_delta, retries_before) = {
6848 let managed = resp_mgr
6849 .links
6850 .get_mut(&link_id)
6851 .expect("receiver link should still exist");
6852 let receiver = managed
6853 .incoming_resources
6854 .first_mut()
6855 .expect("receiver should have an active incoming resource");
6856
6857 assert!(
6858 receiver.waiting_for_hmu,
6859 "receiver should be waiting for HMU"
6860 );
6861
6862 let eifr = receiver.eifr.unwrap_or_else(|| {
6863 (constants::RESOURCE_SDU as f64 * 8.0) / receiver.rtt.unwrap_or(0.5)
6864 });
6865 let expected_tof = if receiver.outstanding_parts > 0 {
6866 (receiver.outstanding_parts as f64 * constants::RESOURCE_SDU as f64 * 8.0) / eifr
6867 } else {
6868 (3.0 * constants::RESOURCE_SDU as f64) / eifr
6869 };
6870 let expected_hmu_wait =
6871 (constants::RESOURCE_SDU as f64 * 8.0 * constants::RESOURCE_HMU_WAIT_FACTOR) / eifr;
6872 let old_delta = constants::RESOURCE_PART_TIMEOUT_FACTOR_AFTER_RTT * expected_tof
6873 + constants::RESOURCE_RETRY_GRACE_TIME;
6874 (
6875 old_delta + expected_hmu_wait + expected_hmu_wait.max(1.0),
6876 receiver.retries_left,
6877 )
6878 };
6879 {
6880 let managed = resp_mgr.links.get(&link_id).unwrap();
6881 let receiver = managed.incoming_resources.first().unwrap();
6882 assert_eq!(receiver.retries_left, retries_before);
6883 assert!(
6884 receiver.eifr.is_some(),
6885 "receiver tick should have populated EIFR"
6886 );
6887 }
6888
6889 let late_resource_actions = {
6890 let managed = resp_mgr.links.get_mut(&link_id).unwrap();
6891 let receiver = managed.incoming_resources.first_mut().unwrap();
6892 let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
6893 managed.engine.decrypt(ciphertext).map_err(|_| ())
6894 };
6895 receiver.tick(
6896 receiver.last_activity + late_delta,
6897 &decrypt_fn,
6898 &Bzip2Compressor,
6899 )
6900 };
6901 let late_actions =
6902 resp_mgr.process_resource_actions(&link_id, late_resource_actions, &mut rng);
6903 let retry_raw = late_actions
6904 .iter()
6905 .find_map(|a| match a {
6906 LinkManagerAction::SendPacket { raw, .. } => {
6907 let pkt = RawPacket::unpack(raw).ok()?;
6908 (pkt.context == constants::CONTEXT_RESOURCE_REQ).then_some(raw.clone())
6909 }
6910 _ => None,
6911 })
6912 .expect("receiver should emit a resource retry request after extended timeout");
6913
6914 {
6915 let managed = resp_mgr.links.get(&link_id).unwrap();
6916 let receiver = managed.incoming_resources.first().unwrap();
6917 assert_eq!(receiver.retries_left, retries_before - 1);
6918 }
6919
6920 let retry_pkt = RawPacket::unpack(&retry_raw).unwrap();
6921 let retry_plaintext = resp_mgr
6922 .links
6923 .get(&link_id)
6924 .unwrap()
6925 .engine
6926 .decrypt(&retry_pkt.data)
6927 .expect("retry request should decrypt");
6928 assert_eq!(retry_plaintext[0], constants::RESOURCE_HASHMAP_IS_EXHAUSTED);
6929
6930 let retry_to_sender = init_mgr.handle_local_delivery(
6933 retry_pkt.destination_hash,
6934 &retry_raw,
6935 retry_pkt.packet_hash,
6936 rns_core::transport::types::InterfaceId(0),
6937 &mut rng,
6938 );
6939 assert!(
6940 retry_to_sender.iter().any(|a| match a {
6941 LinkManagerAction::SendPacket { raw, .. } => RawPacket::unpack(raw)
6942 .map(|pkt| pkt.context == constants::CONTEXT_RESOURCE_HMU)
6943 .unwrap_or(false),
6944 _ => false,
6945 }),
6946 "sender should answer the exhausted retry request with a live HMU packet"
6947 );
6948 }
6949
6950 #[test]
6951 fn test_process_resource_actions_mapping() {
6952 let (mut init_mgr, _resp_mgr, link_id) = setup_active_link();
6953 let mut rng = OsRng;
6954
6955 let actions = vec![
6957 ResourceAction::DataReceived {
6958 data: vec![1, 2, 3],
6959 metadata: Some(vec![4, 5]),
6960 },
6961 ResourceAction::Completed,
6962 ResourceAction::Failed(rns_core::resource::ResourceError::Timeout),
6963 ResourceAction::ProgressUpdate {
6964 received: 10,
6965 total: 20,
6966 },
6967 ResourceAction::TeardownLink,
6968 ];
6969
6970 let result = init_mgr.process_resource_actions(&link_id, actions, &mut rng);
6971
6972 assert!(matches!(
6973 result[0],
6974 LinkManagerAction::ResourceReceived { .. }
6975 ));
6976 assert!(matches!(
6977 result[1],
6978 LinkManagerAction::ResourceCompleted { .. }
6979 ));
6980 assert!(matches!(
6981 result[2],
6982 LinkManagerAction::ResourceFailed { .. }
6983 ));
6984 assert!(matches!(
6985 result[3],
6986 LinkManagerAction::ResourceProgress {
6987 received: 10,
6988 total: 20,
6989 ..
6990 }
6991 ));
6992 assert!(result
6993 .iter()
6994 .any(|action| matches!(action, LinkManagerAction::LinkClosed { .. })));
6995 }
6996
6997 #[test]
6998 fn test_link_state_empty() {
6999 let mgr = LinkManager::new();
7000 let fake_id = [0xAA; 16];
7001 assert!(mgr.link_state(&fake_id).is_none());
7002 }
7003
7004 #[test]
7005 fn test_large_response_resource_completes_as_response() {
7006 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
7007 let mut rng = OsRng;
7008
7009 let large_payload: Vec<u8> = (0..5000u32).map(|i| (i & 0xFF) as u8).collect();
7010 let response_value = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(large_payload));
7011 resp_mgr.register_request_handler("/large", None, {
7012 let response_value = response_value.clone();
7013 move |_link_id, _path, _data, _remote| Some(response_value.clone())
7014 });
7015
7016 let req_actions = init_mgr.send_request(&link_id, "/large", b"\xc0", &mut rng);
7017 let req_raw = extract_any_send_packet(&req_actions);
7018 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
7019 let request_id = req_pkt.get_truncated_hash();
7020 let resp_actions = resp_mgr.handle_local_delivery(
7021 req_pkt.destination_hash,
7022 &req_raw,
7023 req_pkt.packet_hash,
7024 rns_core::transport::types::InterfaceId(0),
7025 &mut rng,
7026 );
7027
7028 let mut pending: Vec<(char, LinkManagerAction)> =
7029 resp_actions.into_iter().map(|a| ('r', a)).collect();
7030 let mut rounds = 0;
7031 let mut received_response = None;
7032
7033 while !pending.is_empty() && rounds < 200 {
7034 rounds += 1;
7035 let mut next = Vec::new();
7036
7037 for (source, action) in pending.drain(..) {
7038 let LinkManagerAction::SendPacket { raw, .. } = action else {
7039 continue;
7040 };
7041 let pkt = RawPacket::unpack(&raw).unwrap();
7042 let target_actions = if source == 'r' {
7043 init_mgr.handle_local_delivery(
7044 pkt.destination_hash,
7045 &raw,
7046 pkt.packet_hash,
7047 rns_core::transport::types::InterfaceId(0),
7048 &mut rng,
7049 )
7050 } else {
7051 resp_mgr.handle_local_delivery(
7052 pkt.destination_hash,
7053 &raw,
7054 pkt.packet_hash,
7055 rns_core::transport::types::InterfaceId(0),
7056 &mut rng,
7057 )
7058 };
7059
7060 let target_source = if source == 'r' { 'i' } else { 'r' };
7061 for target_action in &target_actions {
7062 match target_action {
7063 LinkManagerAction::ResponseReceived {
7064 request_id: rid,
7065 data,
7066 ..
7067 } => {
7068 received_response = Some((*rid, data.clone()));
7069 }
7070 LinkManagerAction::ResourceReceived { .. } => {
7071 panic!("response resources must complete as ResponseReceived")
7072 }
7073 LinkManagerAction::ResourceAcceptQuery { .. } => {
7074 panic!("response resources must bypass application acceptance")
7075 }
7076 _ => {}
7077 }
7078 }
7079 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
7080 }
7081
7082 pending = next;
7083 }
7084
7085 let (received_request_id, received_data) = received_response.unwrap_or_else(|| {
7086 panic!(
7087 "large response resource did not complete as ResponseReceived after {} rounds",
7088 rounds
7089 )
7090 });
7091 assert_eq!(received_request_id, request_id);
7092 assert_eq!(received_data, response_value);
7093 }
7094
7095 #[test]
7096 fn response_resource_at_exact_declared_size_limit_is_accepted() {
7097 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
7098 let mut rng = OsRng;
7099 let payload = deterministic_bytes(4096);
7100 let response_value =
7101 rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload.clone()));
7102 let expected_size = rns_core::msgpack::pack(&rns_core::msgpack::Value::Array(vec![
7103 rns_core::msgpack::Value::Bin(vec![0; 16]),
7104 rns_core::msgpack::Value::Bin(payload),
7105 ]))
7106 .len();
7107 resp_mgr.register_request_handler_response("/bounded-resource", None, {
7108 let response_value = response_value.clone();
7109 move |_, _, _, _| {
7110 Some(RequestResponse::Resource {
7111 data: response_value.clone(),
7112 metadata: None,
7113 auto_compress: false,
7114 })
7115 }
7116 });
7117
7118 let request_actions = init_mgr.send_request_with_max_response_size(
7119 &link_id,
7120 "/bounded-resource",
7121 b"\xc0",
7122 Some(expected_size),
7123 &mut rng,
7124 );
7125 let request_raw = extract_any_send_packet(&request_actions);
7126 let request = RawPacket::unpack(&request_raw).unwrap();
7127 let response_actions = resp_mgr.handle_local_delivery(
7128 request.destination_hash,
7129 &request_raw,
7130 request.packet_hash,
7131 rns_core::transport::types::InterfaceId(0),
7132 &mut rng,
7133 );
7134 let adv = first_resource_advertisement(&resp_mgr, &link_id, &response_actions);
7135 assert_eq!(adv.data_size as usize, expected_size);
7136 let adv_raw = extract_any_send_packet(&response_actions);
7137 let adv_packet = RawPacket::unpack(&adv_raw).unwrap();
7138 let actions = init_mgr.handle_local_delivery(
7139 adv_packet.destination_hash,
7140 &adv_raw,
7141 adv_packet.packet_hash,
7142 rns_core::transport::types::InterfaceId(0),
7143 &mut rng,
7144 );
7145
7146 assert!(!actions
7147 .iter()
7148 .any(|action| matches!(action, LinkManagerAction::RequestFailed { .. })));
7149 assert_eq!(init_mgr.links[&link_id].incoming_resources.len(), 1);
7150 }
7151
7152 #[test]
7153 fn response_resource_one_byte_over_declared_size_limit_is_rejected() {
7154 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
7155 let mut rng = OsRng;
7156 let payload = deterministic_bytes(4096);
7157 let response_value =
7158 rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload.clone()));
7159 let expected_size = rns_core::msgpack::pack(&rns_core::msgpack::Value::Array(vec![
7160 rns_core::msgpack::Value::Bin(vec![0; 16]),
7161 rns_core::msgpack::Value::Bin(payload),
7162 ]))
7163 .len();
7164 resp_mgr.register_request_handler_response("/bounded-resource", None, {
7165 let response_value = response_value.clone();
7166 move |_, _, _, _| {
7167 Some(RequestResponse::Resource {
7168 data: response_value.clone(),
7169 metadata: None,
7170 auto_compress: false,
7171 })
7172 }
7173 });
7174
7175 let request_actions = init_mgr.send_request_with_max_response_size(
7176 &link_id,
7177 "/bounded-resource",
7178 b"\xc0",
7179 Some(expected_size - 1),
7180 &mut rng,
7181 );
7182 let request_raw = extract_any_send_packet(&request_actions);
7183 let request = RawPacket::unpack(&request_raw).unwrap();
7184 let request_id = request.get_truncated_hash();
7185 let response_actions = resp_mgr.handle_local_delivery(
7186 request.destination_hash,
7187 &request_raw,
7188 request.packet_hash,
7189 rns_core::transport::types::InterfaceId(0),
7190 &mut rng,
7191 );
7192 let adv_raw = extract_any_send_packet(&response_actions);
7193 let adv_packet = RawPacket::unpack(&adv_raw).unwrap();
7194 let actions = init_mgr.handle_local_delivery(
7195 adv_packet.destination_hash,
7196 &adv_raw,
7197 adv_packet.packet_hash,
7198 rns_core::transport::types::InterfaceId(0),
7199 &mut rng,
7200 );
7201
7202 assert!(actions.iter().any(|action| matches!(
7203 action,
7204 LinkManagerAction::RequestFailed {
7205 request_id: failed_id,
7206 reason: RequestFailure::ResponseTooLarge { size, maximum },
7207 ..
7208 } if failed_id == &request_id
7209 && *size == expected_size as u64
7210 && *maximum == expected_size - 1
7211 )));
7212 assert!(init_mgr.links[&link_id].incoming_resources.is_empty());
7213 assert!(!init_mgr.links[&link_id]
7214 .pending_requests
7215 .contains_key(&request_id));
7216 }
7217
7218 #[test]
7219 fn test_response_resource_preserves_metadata() {
7220 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
7221 let mut rng = OsRng;
7222
7223 let payload = b"bundle-data".to_vec();
7224 let metadata = b"git-status-ok".to_vec();
7225 let response_value = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload));
7226 resp_mgr.register_request_handler_response("/fetch", None, {
7227 let response_value = response_value.clone();
7228 let metadata = metadata.clone();
7229 move |_link_id, _path, _data, _remote| {
7230 Some(RequestResponse::Resource {
7231 data: response_value.clone(),
7232 metadata: Some(metadata.clone()),
7233 auto_compress: false,
7234 })
7235 }
7236 });
7237
7238 let req_actions = init_mgr.send_request(&link_id, "/fetch", b"\xc0", &mut rng);
7239 let req_raw = extract_any_send_packet(&req_actions);
7240 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
7241 let request_id = req_pkt.get_truncated_hash();
7242 let resp_actions = resp_mgr.handle_local_delivery(
7243 req_pkt.destination_hash,
7244 &req_raw,
7245 req_pkt.packet_hash,
7246 rns_core::transport::types::InterfaceId(0),
7247 &mut rng,
7248 );
7249
7250 let mut pending: Vec<(char, LinkManagerAction)> =
7251 resp_actions.into_iter().map(|a| ('r', a)).collect();
7252 let mut received_response = None;
7253
7254 for _ in 0..200 {
7255 if pending.is_empty() || received_response.is_some() {
7256 break;
7257 }
7258
7259 let mut next = Vec::new();
7260 for (source, action) in pending.drain(..) {
7261 let LinkManagerAction::SendPacket { raw, .. } = action else {
7262 continue;
7263 };
7264 let pkt = RawPacket::unpack(&raw).unwrap();
7265 let target_actions = if source == 'r' {
7266 init_mgr.handle_local_delivery(
7267 pkt.destination_hash,
7268 &raw,
7269 pkt.packet_hash,
7270 rns_core::transport::types::InterfaceId(0),
7271 &mut rng,
7272 )
7273 } else {
7274 resp_mgr.handle_local_delivery(
7275 pkt.destination_hash,
7276 &raw,
7277 pkt.packet_hash,
7278 rns_core::transport::types::InterfaceId(0),
7279 &mut rng,
7280 )
7281 };
7282
7283 let target_source = if source == 'r' { 'i' } else { 'r' };
7284 for target_action in &target_actions {
7285 match target_action {
7286 LinkManagerAction::ResponseReceived {
7287 request_id: rid,
7288 data,
7289 metadata: response_metadata,
7290 ..
7291 } => {
7292 received_response =
7293 Some((*rid, data.clone(), response_metadata.clone()));
7294 }
7295 LinkManagerAction::ResourceReceived { .. } => {
7296 panic!("response resources must complete as ResponseReceived")
7297 }
7298 _ => {}
7299 }
7300 }
7301 next.extend(target_actions.into_iter().map(|a| (target_source, a)));
7302 }
7303 pending = next;
7304 }
7305
7306 let (received_request_id, received_data, received_metadata) = received_response
7307 .expect("resource response with metadata should complete as ResponseReceived");
7308 assert_eq!(received_request_id, request_id);
7309 assert_eq!(received_data, response_value);
7310 assert_eq!(received_metadata, Some(metadata));
7311 }
7312
7313 #[test]
7314 fn test_negotiated_mtu_response_uses_resource_before_global_mtu() {
7315 let (mut init_mgr, mut resp_mgr, link_id) = setup_active_link();
7316 let mut rng = OsRng;
7317
7318 init_mgr.set_link_mtu(&link_id, 300);
7319 resp_mgr.set_link_mtu(&link_id, 300);
7320
7321 let payload = vec![0xAB; 350];
7322 let response_value = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload));
7323 resp_mgr.register_request_handler("/mtu", None, {
7324 let response_value = response_value.clone();
7325 move |_link_id, _path, _data, _remote| Some(response_value.clone())
7326 });
7327
7328 let req_actions = init_mgr.send_request(&link_id, "/mtu", b"\xc0", &mut rng);
7329 let req_raw = extract_any_send_packet(&req_actions);
7330 let req_pkt = RawPacket::unpack(&req_raw).unwrap();
7331 let resp_actions = resp_mgr.handle_local_delivery(
7332 req_pkt.destination_hash,
7333 &req_raw,
7334 req_pkt.packet_hash,
7335 rns_core::transport::types::InterfaceId(0),
7336 &mut rng,
7337 );
7338
7339 let mut has_resource_adv = false;
7340 let mut direct_response_len = None;
7341 for action in &resp_actions {
7342 if let LinkManagerAction::SendPacket { raw, .. } = action {
7343 let pkt = RawPacket::unpack(raw).unwrap();
7344 has_resource_adv |= pkt.context == constants::CONTEXT_RESOURCE_ADV;
7345 if pkt.context == constants::CONTEXT_RESPONSE {
7346 direct_response_len = Some(raw.len());
7347 }
7348 }
7349 }
7350
7351 assert!(
7352 has_resource_adv,
7353 "responses larger than the negotiated link MTU should use resource fallback"
7354 );
7355 assert!(
7356 direct_response_len.is_none(),
7357 "sent direct response of {} bytes on a 300 byte negotiated MTU",
7358 direct_response_len.unwrap_or_default()
7359 );
7360 }
7361
7362 #[test]
7363 fn test_large_management_response_uses_resource_fallback() {
7364 let (_init_mgr, mut resp_mgr, link_id) = setup_active_link();
7365 let mut rng = OsRng;
7366
7367 let payload = vec![0xBC; 5000];
7368 let response_value = rns_core::msgpack::pack(&rns_core::msgpack::Value::Bin(payload));
7369 let actions =
7370 resp_mgr.send_management_response(&link_id, &[0x55; 16], &response_value, &mut rng);
7371
7372 let mut has_resource_adv = false;
7373 let mut has_direct_response = false;
7374 for action in &actions {
7375 if let LinkManagerAction::SendPacket { raw, .. } = action {
7376 let pkt = RawPacket::unpack(raw).unwrap();
7377 has_resource_adv |= pkt.context == constants::CONTEXT_RESOURCE_ADV;
7378 has_direct_response |= pkt.context == constants::CONTEXT_RESPONSE;
7379 }
7380 }
7381
7382 assert!(
7383 has_resource_adv,
7384 "large management responses should advertise a response resource"
7385 );
7386 assert!(
7387 !has_direct_response,
7388 "large management responses should not use a direct CONTEXT_RESPONSE packet"
7389 );
7390 }
7391}