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                data: resource_payload,
255                metadata,
256                auto_compress,
257                is_response: true,
258                request_id: Some(request_id.to_vec()),
259                rng,
260                now,
261            },
262        ) {
263            Ok(s) => s,
264            Err(e) => {
265                log::debug!("Failed to create response ResourceSender: {}", e);
266                return Vec::new();
267            }
268        };
269
270        let adv_actions = Self::start_resource_senders(link, senders, now);
271
272        let _ = link;
273        self.process_resource_actions(link_id, adv_actions, rng)
274    }
275
276    /// Send the value for a previously accepted deferred request.
277    pub fn send_deferred_response(
278        &mut self,
279        link_id: &LinkId,
280        request_id: &[u8; 16],
281        response_data: &[u8],
282        rng: &mut dyn Rng,
283    ) -> Vec<LinkManagerAction> {
284        let packet_actions = self.build_response_packet(link_id, request_id, response_data, rng);
285        if !packet_actions.is_empty() {
286            packet_actions
287        } else {
288            self.send_response_resource(link_id, request_id, response_data, None, true, rng)
289        }
290    }
291
292    /// Send a management response on a link.
293    /// Called by the driver after building the response for a ManagementRequest.
294    pub fn send_management_response(
295        &mut self,
296        link_id: &LinkId,
297        request_id: &[u8; 16],
298        response_data: &[u8],
299        rng: &mut dyn Rng,
300    ) -> Vec<LinkManagerAction> {
301        let mut actions = self.build_response_packet(link_id, request_id, response_data, rng);
302        if actions.is_empty() {
303            actions.extend(self.send_response_resource(
304                link_id,
305                request_id,
306                response_data,
307                None,
308                true,
309                rng,
310            ));
311        }
312        actions
313    }
314
315    /// Send a request on a link.
316    ///
317    /// `data` is the msgpack-encoded request data value (e.g. msgpack([True]) for /status).
318    ///
319    /// Uses Python-compatible format: plaintext = msgpack([timestamp, path_hash_bytes, data_value]).
320    /// Returns actions (the encrypted request packet). The response will arrive
321    /// later via handle_local_delivery with CONTEXT_RESPONSE.
322    pub fn send_request(
323        &mut self,
324        link_id: &LinkId,
325        path: &str,
326        data: &[u8],
327        rng: &mut dyn Rng,
328    ) -> Vec<LinkManagerAction> {
329        self.send_request_with_max_response_size(link_id, path, data, None, rng)
330    }
331
332    /// Send a request with an optional maximum accepted response size.
333    pub fn send_request_with_max_response_size(
334        &mut self,
335        link_id: &LinkId,
336        path: &str,
337        data: &[u8],
338        max_response_size: Option<usize>,
339        rng: &mut dyn Rng,
340    ) -> Vec<LinkManagerAction> {
341        use rns_core::msgpack::{self, Value};
342
343        let link = match self.links.get_mut(link_id) {
344            Some(l) => l,
345            None => return Vec::new(),
346        };
347
348        if link.engine.state() != LinkState::Active {
349            return Vec::new();
350        }
351
352        let path_hash = compute_path_hash(path);
353
354        // Decode data bytes to msgpack Value (or use Bin if can't decode)
355        let data_value = msgpack::unpack_exact(data).unwrap_or_else(|_| Value::Bin(data.to_vec()));
356
357        // Python-compatible format: msgpack([timestamp, Bin(path_hash), data_value])
358        let request_array = Value::Array(vec![
359            Value::Float(time::now()),
360            Value::Bin(path_hash.to_vec()),
361            data_value,
362        ]);
363        let plaintext = msgpack::pack(&request_array);
364
365        if plaintext.len() > link.engine.mdu() {
366            let request_id = rns_core::hash::truncated_hash(&plaintext);
367            let now = time::now();
368            let senders = match Self::build_resource_senders(
369                link,
370                ResourceSendParams {
371                    data: &plaintext,
372                    metadata: None,
373                    auto_compress: true,
374                    is_response: false,
375                    request_id: Some(request_id.to_vec()),
376                    rng,
377                    now,
378                },
379            ) {
380                Ok(senders) => senders,
381                Err(e) => {
382                    log::debug!("Failed to create request ResourceSender: {}", e);
383                    return Vec::new();
384                }
385            };
386            // Resource request response timing starts once the remote proves
387            // receipt of the final request-resource segment.
388            link.pending_requests.insert(
389                request_id,
390                PendingRequest {
391                    deadline: None,
392                    max_response_size,
393                },
394            );
395            let adv_actions = Self::start_resource_senders(link, senders, now);
396            let _ = link;
397            return self.process_resource_actions(link_id, adv_actions, rng);
398        }
399
400        let encrypted = match link.engine.encrypt(&plaintext, rng) {
401            Ok(e) => e,
402            Err(_) => return Vec::new(),
403        };
404
405        let flags = PacketFlags {
406            header_type: constants::HEADER_1,
407            context_flag: constants::FLAG_UNSET,
408            transport_type: constants::TRANSPORT_BROADCAST,
409            destination_type: constants::DESTINATION_LINK,
410            packet_type: constants::PACKET_TYPE_DATA,
411        };
412
413        let mut actions = Vec::new();
414        let max_mtu = link.engine.mtu() as usize;
415        if let Ok((raw, packet_hash)) = RawPacket::pack_raw_with_hash_with_max_mtu(
416            flags,
417            0,
418            link_id,
419            None,
420            constants::CONTEXT_REQUEST,
421            &encrypted,
422            max_mtu,
423        ) {
424            let mut request_id = [0u8; 16];
425            request_id.copy_from_slice(&packet_hash[..16]);
426            let deadline = Self::request_response_deadline(link, time::now());
427            link.pending_requests.insert(
428                request_id,
429                PendingRequest {
430                    deadline: Some(deadline),
431                    max_response_size,
432                },
433            );
434            actions.push(LinkManagerAction::SendPacket {
435                raw,
436                dest_type: constants::DESTINATION_LINK,
437                attached_interface: None,
438            });
439        }
440        actions
441    }
442
443    /// Send encrypted data on a link with a given context.
444    /// Handle a response on a link.
445    pub(super) fn handle_response(
446        &mut self,
447        link_id: &LinkId,
448        plaintext: &[u8],
449        metadata: Option<Vec<u8>>,
450        resource_request_id: Option<[u8; 16]>,
451    ) -> Vec<LinkManagerAction> {
452        use rns_core::msgpack;
453
454        // Python-compatible packet response: msgpack([Bin(request_id), response_value]).
455        let packet_response = msgpack::unpack_exact(plaintext).ok().and_then(|value| {
456            let msgpack::Value::Array(arr) = value else {
457                return None;
458            };
459            if arr.len() < 2 {
460                return None;
461            }
462            let msgpack::Value::Bin(request_id_bytes) = &arr[0] else {
463                return None;
464            };
465            if request_id_bytes.len() != 16 {
466                return None;
467            }
468            let mut request_id = [0u8; 16];
469            request_id.copy_from_slice(request_id_bytes);
470            Some((request_id, msgpack::pack(&arr[1])))
471        });
472
473        let (request_id, response_data, check_size) = match packet_response {
474            Some((request_id, response_data)) => (request_id, response_data, true),
475            None => {
476                let Some(request_id) = resource_request_id else {
477                    return Vec::new();
478                };
479                (
480                    request_id,
481                    msgpack::pack(&msgpack::Value::Bin(plaintext.to_vec())),
482                    false,
483                )
484            }
485        };
486
487        let Some(link) = self.links.get_mut(link_id) else {
488            return Vec::new();
489        };
490        let Some(pending) = link.pending_requests.get(&request_id) else {
491            return Vec::new();
492        };
493        if check_size {
494            // Upstream excludes the two-byte MessagePack binary wrapper from
495            // packet response sizes. Resource responses are checked from the
496            // advertisement before their transfer is accepted.
497            let response_size = response_data.len().saturating_sub(2) as u64;
498            if let Some(maximum) = pending.max_response_size {
499                if response_size > maximum as u64 {
500                    link.pending_requests.remove(&request_id);
501                    log::debug!(
502                        "rejected response with excessive size {} bytes on link {:02x?}",
503                        response_size,
504                        &link_id[..4]
505                    );
506                    return vec![LinkManagerAction::RequestFailed {
507                        link_id: *link_id,
508                        request_id,
509                        reason: RequestFailure::ResponseTooLarge {
510                            size: response_size,
511                            maximum,
512                        },
513                    }];
514                }
515            }
516        }
517        link.pending_requests.remove(&request_id);
518
519        vec![LinkManagerAction::ResponseReceived {
520            link_id: *link_id,
521            request_id,
522            data: response_data,
523            metadata,
524        }]
525    }
526
527    pub(super) fn response_request_id(request_id: &Option<Vec<u8>>) -> Option<[u8; 16]> {
528        let bytes = request_id.as_deref()?;
529        if bytes.len() != 16 {
530            return None;
531        }
532        let mut out = [0u8; 16];
533        out.copy_from_slice(bytes);
534        Some(out)
535    }
536}
rns_net/common/link_manager/request_handling.rs

rns_net/common/link_manager/
request_handling.rs
