rns_net/common/link_manager/
request_handling.rs

1use super::*;
2
3impl LinkManager {
4    /// Handle a request on a link.
5    pub(super) fn handle_request(
6        &mut self,
7        link_id: &LinkId,
8        plaintext: &[u8],
9        request_id: [u8; 16],
10        rng: &mut dyn Rng,
11    ) -> Vec<LinkManagerAction> {
12        use rns_core::msgpack::{self, Value};
13
14        // Python-compatible format: msgpack([timestamp, Bin(path_hash), data_value])
15        let arr = match msgpack::unpack_exact(plaintext) {
16            Ok(Value::Array(arr)) if arr.len() >= 3 => arr,
17            _ => return Vec::new(),
18        };
19
20        let path_hash_bytes = match &arr[1] {
21            Value::Bin(b) if b.len() == 16 => b,
22            _ => return Vec::new(),
23        };
24        let mut path_hash = [0u8; 16];
25        path_hash.copy_from_slice(path_hash_bytes);
26
27        // Re-encode the data element for the handler
28        let request_data = msgpack::pack(&arr[2]);
29
30        // Check if this is a management path (handled by the driver)
31        if self.management_paths.contains(&path_hash) {
32            let remote_identity = self.links.get(link_id).and_then(|l| l.remote_identity);
33            return vec![LinkManagerAction::ManagementRequest {
34                link_id: *link_id,
35                path_hash,
36                data: request_data,
37                request_id,
38                remote_identity,
39            }];
40        }
41
42        if let Some(handler) = self
43            .deferred_request_handlers
44            .iter()
45            .find(|handler| handler.path_hash == path_hash)
46        {
47            let remote_identity = self
48                .links
49                .get(link_id)
50                .and_then(|link| link.remote_identity.as_ref());
51            if let Some(allowed) = &handler.allowed_list {
52                let Some((identity_hash, _)) = remote_identity else {
53                    log::debug!("Deferred request denied: peer not identified");
54                    return Vec::new();
55                };
56                if !allowed.contains(identity_hash) {
57                    log::debug!("Deferred request denied: identity not in allowed list");
58                    return Vec::new();
59                }
60            }
61            (handler.handler)(
62                *link_id,
63                &handler.path,
64                request_id,
65                &request_data,
66                remote_identity,
67            );
68            return Vec::new();
69        }
70
71        // Look up handler by path_hash
72        let handler_idx = self
73            .request_handlers
74            .iter()
75            .position(|h| h.path_hash == path_hash);
76        let handler_idx = match handler_idx {
77            Some(i) => i,
78            None => return Vec::new(),
79        };
80
81        // Check ACL
82        let remote_identity = self
83            .links
84            .get(link_id)
85            .and_then(|l| l.remote_identity.as_ref());
86        let handler = &self.request_handlers[handler_idx];
87        if let Some(ref allowed) = handler.allowed_list {
88            match remote_identity {
89                Some((identity_hash, _)) => {
90                    if !allowed.contains(identity_hash) {
91                        log::debug!("Request denied: identity not in allowed list");
92                        return Vec::new();
93                    }
94                }
95                None => {
96                    log::debug!("Request denied: peer not identified");
97                    return Vec::new();
98                }
99            }
100        }
101
102        // Call handler
103        let path = handler.path.clone();
104        let response = (handler.handler)(*link_id, &path, &request_data, remote_identity);
105
106        let mut actions = Vec::new();
107        if let Some(response) = response {
108            match response {
109                RequestResponse::Bytes(response_data) => {
110                    let mut response_actions =
111                        self.build_response_packet(link_id, &request_id, &response_data, rng);
112                    if response_actions.is_empty() {
113                        response_actions.extend(self.send_response_resource(
114                            link_id,
115                            &request_id,
116                            &response_data,
117                            None,
118                            true,
119                            rng,
120                        ));
121                    }
122                    actions.extend(response_actions);
123                }
124                RequestResponse::Resource {
125                    data,
126                    metadata,
127                    auto_compress,
128                } => {
129                    actions.extend(self.send_response_resource(
130                        link_id,
131                        &request_id,
132                        &data,
133                        metadata.as_deref(),
134                        auto_compress,
135                        rng,
136                    ));
137                }
138                RequestResponse::File {
139                    data,
140                    metadata,
141                    auto_compress,
142                } => {
143                    actions.extend(self.send_file_response_resource(
144                        link_id,
145                        &request_id,
146                        &data,
147                        Some(&metadata),
148                        auto_compress,
149                        rng,
150                    ));
151                }
152            }
153        }
154
155        actions
156    }
157
158    /// Build a response packet for a request.
159    /// `response_data` is the msgpack-encoded response value.
160    fn build_response_packet(
161        &self,
162        link_id: &LinkId,
163        request_id: &[u8; 16],
164        response_data: &[u8],
165        rng: &mut dyn Rng,
166    ) -> Vec<LinkManagerAction> {
167        use rns_core::msgpack::{self, Value};
168
169        let response_value = msgpack::unpack_exact(response_data)
170            .unwrap_or_else(|_| Value::Bin(response_data.to_vec()));
171
172        let response_array = Value::Array(vec![Value::Bin(request_id.to_vec()), response_value]);
173        let response_plaintext = msgpack::pack(&response_array);
174
175        let mut actions = Vec::new();
176        if let Some(link) = self.links.get(link_id) {
177            if let Ok(encrypted) = link.engine.encrypt(&response_plaintext, rng) {
178                let flags = PacketFlags {
179                    header_type: constants::HEADER_1,
180                    context_flag: constants::FLAG_UNSET,
181                    transport_type: constants::TRANSPORT_BROADCAST,
182                    destination_type: constants::DESTINATION_LINK,
183                    packet_type: constants::PACKET_TYPE_DATA,
184                };
185                let max_mtu = link.engine.mtu() as usize;
186                if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
187                    flags,
188                    0,
189                    link_id,
190                    None,
191                    constants::CONTEXT_RESPONSE,
192                    &encrypted,
193                    max_mtu,
194                ) {
195                    actions.push(LinkManagerAction::SendPacket {
196                        raw,
197                        dest_type: constants::DESTINATION_LINK,
198                        attached_interface: None,
199                    });
200                }
201            }
202        }
203        actions
204    }
205
206    pub(super) fn send_response_resource(
207        &mut self,
208        link_id: &LinkId,
209        request_id: &[u8; 16],
210        response_data: &[u8],
211        metadata: Option<&[u8]>,
212        auto_compress: bool,
213        rng: &mut dyn Rng,
214    ) -> Vec<LinkManagerAction> {
215        use rns_core::msgpack::{self, Value};
216
217        let response_value = msgpack::unpack_exact(response_data)
218            .unwrap_or_else(|_| Value::Bin(response_data.to_vec()));
219        let response_array = Value::Array(vec![Value::Bin(request_id.to_vec()), response_value]);
220        let resource_payload = msgpack::pack(&response_array);
221        self.send_file_response_resource(
222            link_id,
223            request_id,
224            &resource_payload,
225            metadata,
226            auto_compress,
227            rng,
228        )
229    }
230
231    fn send_file_response_resource(
232        &mut self,
233        link_id: &LinkId,
234        request_id: &[u8; 16],
235        resource_payload: &[u8],
236        metadata: Option<&[u8]>,
237        auto_compress: bool,
238        rng: &mut dyn Rng,
239    ) -> Vec<LinkManagerAction> {
240        let link = match self.links.get_mut(link_id) {
241            Some(l) => l,
242            None => return Vec::new(),
243        };
244
245        if link.engine.state() != LinkState::Active {
246            return Vec::new();
247        }
248
249        let now = time::now();
250
251        let senders = match Self::build_resource_senders(
252            link,
253            ResourceSendParams {
254                compressor: &Bzip2Compressor,
255                data: resource_payload,
256                metadata,
257                auto_compress,
258                is_response: true,
259                request_id: Some(request_id.to_vec()),
260                rng,
261                now,
262            },
263        ) {
264            Ok(s) => s,
265            Err(e) => {
266                log::debug!("Failed to create response ResourceSender: {}", e);
267                return Vec::new();
268            }
269        };
270
271        let adv_actions = Self::start_resource_senders(link, senders, now);
272
273        let _ = link;
274        self.process_resource_actions(link_id, adv_actions, rng)
275    }
276
277    /// Send the value for a previously accepted deferred request.
278    pub fn send_deferred_response(
279        &mut self,
280        link_id: &LinkId,
281        request_id: &[u8; 16],
282        response_data: &[u8],
283        rng: &mut dyn Rng,
284    ) -> Vec<LinkManagerAction> {
285        let packet_actions = self.build_response_packet(link_id, request_id, response_data, rng);
286        if !packet_actions.is_empty() {
287            packet_actions
288        } else {
289            self.send_response_resource(link_id, request_id, response_data, None, true, rng)
290        }
291    }
292
293    /// Send a management response on a link.
294    /// Called by the driver after building the response for a ManagementRequest.
295    pub fn send_management_response(
296        &mut self,
297        link_id: &LinkId,
298        request_id: &[u8; 16],
299        response_data: &[u8],
300        rng: &mut dyn Rng,
301    ) -> Vec<LinkManagerAction> {
302        let mut actions = self.build_response_packet(link_id, request_id, response_data, rng);
303        if actions.is_empty() {
304            actions.extend(self.send_response_resource(
305                link_id,
306                request_id,
307                response_data,
308                None,
309                true,
310                rng,
311            ));
312        }
313        actions
314    }
315
316    /// Send a request on a link.
317    ///
318    /// `data` is the msgpack-encoded request data value (e.g. msgpack([True]) for /status).
319    ///
320    /// Uses Python-compatible format: plaintext = msgpack([timestamp, path_hash_bytes, data_value]).
321    /// Returns actions (the encrypted request packet). The response will arrive
322    /// later via handle_local_delivery with CONTEXT_RESPONSE.
323    pub fn send_request(
324        &mut self,
325        link_id: &LinkId,
326        path: &str,
327        data: &[u8],
328        rng: &mut dyn Rng,
329    ) -> Vec<LinkManagerAction> {
330        self.send_request_with_max_response_size(link_id, path, data, None, rng)
331    }
332
333    /// Send a request with an optional maximum accepted response size.
334    pub fn send_request_with_max_response_size(
335        &mut self,
336        link_id: &LinkId,
337        path: &str,
338        data: &[u8],
339        max_response_size: Option<usize>,
340        rng: &mut dyn Rng,
341    ) -> Vec<LinkManagerAction> {
342        use rns_core::msgpack::{self, Value};
343
344        let link = match self.links.get_mut(link_id) {
345            Some(l) => l,
346            None => return Vec::new(),
347        };
348
349        if link.engine.state() != LinkState::Active {
350            return Vec::new();
351        }
352
353        let path_hash = compute_path_hash(path);
354
355        // Decode data bytes to msgpack Value (or use Bin if can't decode)
356        let data_value = msgpack::unpack_exact(data).unwrap_or_else(|_| Value::Bin(data.to_vec()));
357
358        // Python-compatible format: msgpack([timestamp, Bin(path_hash), data_value])
359        let request_array = Value::Array(vec![
360            Value::Float(time::now()),
361            Value::Bin(path_hash.to_vec()),
362            data_value,
363        ]);
364        let plaintext = msgpack::pack(&request_array);
365
366        if plaintext.len() > link.engine.mdu() {
367            let request_id = rns_core::hash::truncated_hash(&plaintext);
368            let now = time::now();
369            let senders = match Self::build_resource_senders(
370                link,
371                ResourceSendParams {
372                    compressor: &Bzip2Compressor,
373                    data: &plaintext,
374                    metadata: None,
375                    auto_compress: true,
376                    is_response: false,
377                    request_id: Some(request_id.to_vec()),
378                    rng,
379                    now,
380                },
381            ) {
382                Ok(senders) => senders,
383                Err(e) => {
384                    log::debug!("Failed to create request ResourceSender: {}", e);
385                    return Vec::new();
386                }
387            };
388            // Resource request response timing starts once the remote proves
389            // receipt of the final request-resource segment.
390            link.pending_requests.insert(
391                request_id,
392                PendingRequest {
393                    deadline: None,
394                    max_response_size,
395                },
396            );
397            let adv_actions = Self::start_resource_senders(link, senders, now);
398            let _ = link;
399            return self.process_resource_actions(link_id, adv_actions, rng);
400        }
401
402        let encrypted = match link.engine.encrypt(&plaintext, rng) {
403            Ok(e) => e,
404            Err(_) => return Vec::new(),
405        };
406
407        let flags = PacketFlags {
408            header_type: constants::HEADER_1,
409            context_flag: constants::FLAG_UNSET,
410            transport_type: constants::TRANSPORT_BROADCAST,
411            destination_type: constants::DESTINATION_LINK,
412            packet_type: constants::PACKET_TYPE_DATA,
413        };
414
415        let mut actions = Vec::new();
416        let max_mtu = link.engine.mtu() as usize;
417        if let Ok((raw, packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
418            flags,
419            0,
420            link_id,
421            None,
422            constants::CONTEXT_REQUEST,
423            &encrypted,
424            max_mtu,
425        ) {
426            let mut request_id = [0u8; 16];
427            request_id.copy_from_slice(&packet_hash[..16]);
428            let deadline = Self::request_response_deadline(link, time::now());
429            link.pending_requests.insert(
430                request_id,
431                PendingRequest {
432                    deadline: Some(deadline),
433                    max_response_size,
434                },
435            );
436            actions.push(LinkManagerAction::SendPacket {
437                raw,
438                dest_type: constants::DESTINATION_LINK,
439                attached_interface: None,
440            });
441        }
442        actions
443    }
444
445    /// Send encrypted data on a link with a given context.
446    /// Handle a response on a link.
447    pub(super) fn handle_response(
448        &mut self,
449        link_id: &LinkId,
450        plaintext: &[u8],
451        metadata: Option<Vec<u8>>,
452        resource_request_id: Option<[u8; 16]>,
453    ) -> Vec<LinkManagerAction> {
454        use rns_core::msgpack;
455
456        // Python-compatible packet response: msgpack([Bin(request_id), response_value]).
457        let packet_response = msgpack::unpack_exact(plaintext).ok().and_then(|value| {
458            let msgpack::Value::Array(arr) = value else {
459                return None;
460            };
461            if arr.len() < 2 {
462                return None;
463            }
464            let msgpack::Value::Bin(request_id_bytes) = &arr[0] else {
465                return None;
466            };
467            if request_id_bytes.len() != 16 {
468                return None;
469            }
470            let mut request_id = [0u8; 16];
471            request_id.copy_from_slice(request_id_bytes);
472            Some((request_id, msgpack::pack(&arr[1])))
473        });
474
475        let (request_id, response_data, check_size) = match packet_response {
476            Some((request_id, response_data)) => (request_id, response_data, true),
477            None => {
478                let Some(request_id) = resource_request_id else {
479                    return Vec::new();
480                };
481                (
482                    request_id,
483                    msgpack::pack(&msgpack::Value::Bin(plaintext.to_vec())),
484                    false,
485                )
486            }
487        };
488
489        let Some(link) = self.links.get_mut(link_id) else {
490            return Vec::new();
491        };
492        let Some(pending) = link.pending_requests.get(&request_id) else {
493            return Vec::new();
494        };
495        if check_size {
496            // Upstream excludes the two-byte MessagePack binary wrapper from
497            // packet response sizes. Resource responses are checked from the
498            // advertisement before their transfer is accepted.
499            let response_size = response_data.len().saturating_sub(2) as u64;
500            if let Some(maximum) = pending.max_response_size {
501                if response_size > maximum as u64 {
502                    link.pending_requests.remove(&request_id);
503                    log::debug!(
504                        "rejected response with excessive size {} bytes on link {:02x?}",
505                        response_size,
506                        &link_id[..4]
507                    );
508                    return vec![LinkManagerAction::RequestFailed {
509                        link_id: *link_id,
510                        request_id,
511                        reason: RequestFailure::ResponseTooLarge {
512                            size: response_size,
513                            maximum,
514                        },
515                    }];
516                }
517            }
518        }
519        link.pending_requests.remove(&request_id);
520
521        vec![LinkManagerAction::ResponseReceived {
522            link_id: *link_id,
523            request_id,
524            data: response_data,
525            metadata,
526        }]
527    }
528
529    pub(super) fn response_request_id(request_id: &Option<Vec<u8>>) -> Option<[u8; 16]> {
530        let bytes = request_id.as_deref()?;
531        if bytes.len() != 16 {
532            return None;
533        }
534        let mut out = [0u8; 16];
535        out.copy_from_slice(bytes);
536        Some(out)
537    }
538}
rns_net/common/link_manager/request_handling.rs

rns_net/common/link_manager/
request_handling.rs
