fips_core/node/
endpoint_traffic.rs

1use super::*;
2
3#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
4pub(in crate::node) struct FmpPlaintextTrafficClass {
5    pub(in crate::node) bulk_endpoint_data: bool,
6    pub(in crate::node) drop_on_backpressure: bool,
7}
8
9/// Priority/bulk lane selected for an app-owned endpoint payload.
10#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
11pub enum EndpointPayloadLane {
12    #[default]
13    Priority,
14    Bulk,
15}
16
17impl EndpointPayloadLane {
18    fn command_lane(self) -> EndpointCommandLane {
19        match self {
20            Self::Priority => EndpointCommandLane::Priority,
21            Self::Bulk => EndpointCommandLane::Bulk,
22        }
23    }
24}
25
26/// Traffic policy selected for an app-owned endpoint payload.
27#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
28pub struct EndpointPayloadClass {
29    lane: EndpointPayloadLane,
30    drop_on_backpressure: bool,
31}
32
33impl EndpointPayloadClass {
34    pub fn lane(self) -> EndpointPayloadLane {
35        self.lane
36    }
37
38    pub fn is_latency_sensitive(self) -> bool {
39        self.lane == EndpointPayloadLane::Priority
40    }
41
42    pub fn drop_on_backpressure(self) -> bool {
43        self.drop_on_backpressure
44    }
45}
46
47#[cfg(unix)]
48pub(in crate::node) struct FmpWorkerSendReservation {
49    pub(in crate::node) counter: u64,
50    pub(in crate::node) header: [u8; ESTABLISHED_HEADER_SIZE],
51    pub(in crate::node) cipher: ring::aead::LessSafeKey,
52}
53
54#[cfg(unix)]
55pub(in crate::node) fn reserve_fmp_worker_send(
56    session: &mut crate::noise::NoiseSession,
57    their_index: crate::utils::index::SessionIndex,
58    flags: u8,
59    payload_len: u16,
60) -> Result<Option<FmpWorkerSendReservation>, crate::noise::NoiseError> {
61    let Some(cipher) = session.send_cipher_clone() else {
62        return Ok(None);
63    };
64    let counter = session.take_send_counter()?;
65    let header = build_established_header(their_index, counter, flags, payload_len);
66    Ok(Some(FmpWorkerSendReservation {
67        counter,
68        header,
69        cipher,
70    }))
71}
72
73#[derive(Clone, Copy, Debug, PartialEq, Eq)]
74pub(crate) enum EndpointCommandLane {
75    Priority,
76    Bulk,
77}
78
79pub(in crate::node) fn classify_fmp_plaintext_traffic(
80    plaintext: &[u8],
81) -> FmpPlaintextTrafficClass {
82    let bulk_endpoint_data = fmp_plaintext_is_bulk_session_datagram(plaintext);
83    // At this layer established FSP payloads are already end-to-end encrypted,
84    // so a bulk SessionDatagram may still be TCP endpoint traffic. Keep it out
85    // of the control lane, but only the pre-FSP endpoint path may mark known
86    // non-TCP packets as discardable under sender backpressure.
87    FmpPlaintextTrafficClass {
88        bulk_endpoint_data,
89        drop_on_backpressure: false,
90    }
91}
92
93pub(in crate::node) fn fmp_plaintext_is_bulk_session_datagram(plaintext: &[u8]) -> bool {
94    if plaintext
95        .first()
96        .is_none_or(|ty| *ty != LinkMessageType::SessionDatagram.to_byte())
97    {
98        return false;
99    }
100    let Some(fsp_payload) = plaintext.get(crate::protocol::SESSION_DATAGRAM_HEADER_SIZE..) else {
101        return false;
102    };
103    FspCommonPrefix::parse(fsp_payload).is_some_and(|prefix| {
104        prefix.phase == FSP_PHASE_ESTABLISHED && !prefix.is_unencrypted() && !prefix.has_coords()
105    })
106}
107
108/// Classify an app-owned endpoint payload for queue admission and pressure policy.
109pub fn classify_endpoint_payload(payload: &[u8]) -> EndpointPayloadClass {
110    const IPPROTO_ICMP: u8 = 1;
111    const IPPROTO_TCP: u8 = 6;
112    const IPPROTO_ICMPV6: u8 = 58;
113
114    match parse_endpoint_payload_ip_proto(payload) {
115        Some((IPPROTO_ICMP, _)) => EndpointPayloadClass::default(),
116        Some((IPPROTO_ICMPV6, _)) => EndpointPayloadClass::default(),
117        Some((IPPROTO_TCP, offset)) => {
118            let latency_sensitive = endpoint_tcp_payload_is_latency_sensitive(payload, offset);
119            EndpointPayloadClass {
120                lane: if latency_sensitive {
121                    EndpointPayloadLane::Priority
122                } else {
123                    EndpointPayloadLane::Bulk
124                },
125                drop_on_backpressure: false,
126            }
127        }
128        _ => EndpointPayloadClass {
129            lane: EndpointPayloadLane::Bulk,
130            drop_on_backpressure: true,
131        },
132    }
133}
134
135/// Return true when an app-owned endpoint payload should retain priority-lane progress.
136///
137/// Embedders that stage packets before calling `FipsEndpoint::send*_to_peer`
138/// can use this to apply the same priority/bulk policy as the FIPS endpoint
139/// command queue without duplicating IP/TCP parsing.
140pub fn endpoint_payload_is_latency_sensitive(payload: &[u8]) -> bool {
141    classify_endpoint_payload(payload).is_latency_sensitive()
142}
143
144#[cfg(test)]
145pub(crate) fn endpoint_command_lane_for_payload(payload: &[u8]) -> EndpointCommandLane {
146    if endpoint_payload_is_latency_sensitive(payload) {
147        EndpointCommandLane::Priority
148    } else {
149        EndpointCommandLane::Bulk
150    }
151}
152
153/// Endpoint payload bytes plus the traffic policy selected at app ingress.
154#[derive(Clone, Debug, PartialEq, Eq)]
155pub(crate) struct EndpointDataPayload {
156    bytes: Vec<u8>,
157    traffic_class: EndpointPayloadClass,
158}
159
160impl EndpointDataPayload {
161    pub(crate) fn new(bytes: Vec<u8>) -> Self {
162        let traffic_class = classify_endpoint_payload(&bytes);
163        Self {
164            bytes,
165            traffic_class,
166        }
167    }
168
169    pub(crate) fn from_classified(bytes: Vec<u8>, traffic_class: EndpointPayloadClass) -> Self {
170        Self {
171            bytes,
172            traffic_class,
173        }
174    }
175
176    pub(crate) fn lane(&self) -> EndpointCommandLane {
177        self.traffic_class.lane().command_lane()
178    }
179
180    pub(crate) fn bulk_endpoint_data(&self) -> bool {
181        self.traffic_class.lane() == EndpointPayloadLane::Bulk
182    }
183
184    pub(crate) fn drop_on_backpressure(&self) -> bool {
185        self.traffic_class.drop_on_backpressure()
186    }
187
188    pub(crate) fn as_slice(&self) -> &[u8] {
189        &self.bytes
190    }
191
192    pub(crate) fn len(&self) -> usize {
193        self.bytes.len()
194    }
195}
196
197impl From<Vec<u8>> for EndpointDataPayload {
198    fn from(bytes: Vec<u8>) -> Self {
199        Self::new(bytes)
200    }
201}
202
203/// Outbound endpoint data plus the peer identity it is bound to.
204#[derive(Debug)]
205pub(crate) struct EndpointDataSend {
206    dest_addr: NodeAddr,
207    dest_pubkey: secp256k1::PublicKey,
208    payload: EndpointDataPayload,
209}
210
211impl EndpointDataSend {
212    pub(crate) fn new(remote: PeerIdentity, payload: EndpointDataPayload) -> Self {
213        Self {
214            dest_addr: *remote.node_addr(),
215            dest_pubkey: remote.pubkey_full(),
216            payload,
217        }
218    }
219
220    pub(crate) fn dest_addr(&self) -> NodeAddr {
221        self.dest_addr
222    }
223
224    pub(crate) fn dest_pubkey(&self) -> secp256k1::PublicKey {
225        self.dest_pubkey
226    }
227
228    pub(crate) fn payload(&self) -> &EndpointDataPayload {
229        &self.payload
230    }
231
232    pub(crate) fn into_payload(self) -> EndpointDataPayload {
233        self.payload
234    }
235}
236
237/// Admission result for a bounded pending endpoint-data queue.
238#[derive(Clone, Copy, Debug, PartialEq, Eq)]
239pub(crate) struct PendingEndpointDataQueueAdmission {
240    dropped_oldest: bool,
241}
242
243impl PendingEndpointDataQueueAdmission {
244    pub(crate) fn dropped_oldest(&self) -> bool {
245        self.dropped_oldest
246    }
247}
248
249/// Per-destination endpoint payloads waiting for session establishment.
250#[derive(Debug, Default)]
251pub(crate) struct PendingEndpointDataQueue {
252    payloads: VecDeque<EndpointDataPayload>,
253}
254
255impl PendingEndpointDataQueue {
256    pub(crate) fn push_bounded(
257        &mut self,
258        payload: EndpointDataPayload,
259        capacity: usize,
260    ) -> PendingEndpointDataQueueAdmission {
261        let dropped_oldest = self.payloads.len() >= capacity;
262        if dropped_oldest {
263            self.payloads.pop_front();
264        }
265        self.payloads.push_back(payload);
266        PendingEndpointDataQueueAdmission { dropped_oldest }
267    }
268
269    pub(crate) fn len(&self) -> usize {
270        self.payloads.len()
271    }
272
273    pub(crate) fn into_payloads(self) -> VecDeque<EndpointDataPayload> {
274        self.payloads
275    }
276
277    #[cfg(test)]
278    pub(crate) fn iter(&self) -> impl Iterator<Item = &EndpointDataPayload> {
279        self.payloads.iter()
280    }
281}
282
283/// Admission result for a bounded pending TUN packet queue.
284#[derive(Clone, Copy, Debug, PartialEq, Eq)]
285pub(crate) struct PendingTunPacketQueueAdmission {
286    dropped_oldest: bool,
287}
288
289impl PendingTunPacketQueueAdmission {
290    pub(crate) fn dropped_oldest(&self) -> bool {
291        self.dropped_oldest
292    }
293}
294
295/// Per-destination TUN packets waiting for session establishment.
296#[derive(Debug, Default)]
297pub(crate) struct PendingTunPacketQueue {
298    packets: VecDeque<Vec<u8>>,
299}
300
301impl PendingTunPacketQueue {
302    pub(crate) fn push_bounded(
303        &mut self,
304        packet: Vec<u8>,
305        capacity: usize,
306    ) -> PendingTunPacketQueueAdmission {
307        let dropped_oldest = self.packets.len() >= capacity;
308        if dropped_oldest {
309            self.packets.pop_front();
310        }
311        self.packets.push_back(packet);
312        PendingTunPacketQueueAdmission { dropped_oldest }
313    }
314
315    pub(crate) fn len(&self) -> usize {
316        self.packets.len()
317    }
318
319    pub(crate) fn into_packets(self) -> VecDeque<Vec<u8>> {
320        self.packets
321    }
322
323    #[cfg(test)]
324    pub(crate) fn iter(&self) -> impl Iterator<Item = &Vec<u8>> {
325        self.packets.iter()
326    }
327}
328
329/// Admission result for pending session-establishment traffic.
330#[derive(Clone, Copy, Debug, PartialEq, Eq)]
331pub(crate) struct PendingSessionTrafficAdmission {
332    destination_dropped: bool,
333    dropped_oldest: bool,
334}
335
336impl PendingSessionTrafficAdmission {
337    pub(crate) fn destination_dropped(&self) -> bool {
338        self.destination_dropped
339    }
340
341    pub(crate) fn dropped_oldest(&self) -> bool {
342        self.dropped_oldest
343    }
344}
345
346/// Queued TUN and endpoint traffic removed for one destination.
347#[derive(Debug, Default)]
348pub(crate) struct PendingDestinationTraffic {
349    tun_packets: Option<PendingTunPacketQueue>,
350    endpoint_data: Option<PendingEndpointDataQueue>,
351}
352
353impl PendingDestinationTraffic {
354    pub(crate) fn tun_packets(&self) -> Option<&PendingTunPacketQueue> {
355        self.tun_packets.as_ref()
356    }
357
358    pub(crate) fn into_tun_packets(self) -> Option<PendingTunPacketQueue> {
359        self.tun_packets
360    }
361
362    pub(crate) fn endpoint_data(&self) -> Option<&PendingEndpointDataQueue> {
363        self.endpoint_data.as_ref()
364    }
365}
366
367/// Pending traffic waiting for session establishment.
368#[derive(Debug, Default)]
369pub(crate) struct PendingSessionTrafficQueues {
370    pending_destinations: HashSet<NodeAddr>,
371    tun_packets: HashMap<NodeAddr, PendingTunPacketQueue>,
372    endpoint_data: HashMap<NodeAddr, PendingEndpointDataQueue>,
373}
374
375impl PendingSessionTrafficQueues {
376    pub(crate) fn push_tun_packet(
377        &mut self,
378        dest_addr: NodeAddr,
379        packet: Vec<u8>,
380        max_destinations: usize,
381        packets_per_dest: usize,
382    ) -> PendingSessionTrafficAdmission {
383        if !self.tun_packets.contains_key(&dest_addr) && self.tun_packets.len() >= max_destinations
384        {
385            return PendingSessionTrafficAdmission {
386                destination_dropped: true,
387                dropped_oldest: false,
388            };
389        }
390
391        let admission = self
392            .tun_packets
393            .entry(dest_addr)
394            .or_default()
395            .push_bounded(packet, packets_per_dest);
396        self.pending_destinations.insert(dest_addr);
397        PendingSessionTrafficAdmission {
398            destination_dropped: false,
399            dropped_oldest: admission.dropped_oldest(),
400        }
401    }
402
403    pub(crate) fn push_endpoint_data(
404        &mut self,
405        dest_addr: NodeAddr,
406        payload: impl Into<EndpointDataPayload>,
407        max_destinations: usize,
408        packets_per_dest: usize,
409    ) -> PendingSessionTrafficAdmission {
410        if !self.endpoint_data.contains_key(&dest_addr)
411            && self.endpoint_data.len() >= max_destinations
412        {
413            return PendingSessionTrafficAdmission {
414                destination_dropped: true,
415                dropped_oldest: false,
416            };
417        }
418
419        let admission = self
420            .endpoint_data
421            .entry(dest_addr)
422            .or_default()
423            .push_bounded(payload.into(), packets_per_dest);
424        self.pending_destinations.insert(dest_addr);
425        PendingSessionTrafficAdmission {
426            destination_dropped: false,
427            dropped_oldest: admission.dropped_oldest(),
428        }
429    }
430
431    pub(crate) fn remove_destination(&mut self, dest_addr: &NodeAddr) -> PendingDestinationTraffic {
432        self.pending_destinations.remove(dest_addr);
433        PendingDestinationTraffic {
434            tun_packets: self.tun_packets.remove(dest_addr),
435            endpoint_data: self.endpoint_data.remove(dest_addr),
436        }
437    }
438
439    pub(crate) fn take_tun_packets(
440        &mut self,
441        dest_addr: &NodeAddr,
442    ) -> Option<PendingTunPacketQueue> {
443        let packets = self.tun_packets.remove(dest_addr);
444        if packets.is_some() && !self.endpoint_data.contains_key(dest_addr) {
445            self.pending_destinations.remove(dest_addr);
446        }
447        packets
448    }
449
450    pub(crate) fn take_endpoint_data(
451        &mut self,
452        dest_addr: &NodeAddr,
453    ) -> Option<PendingEndpointDataQueue> {
454        let payloads = self.endpoint_data.remove(dest_addr);
455        if payloads.is_some() && !self.tun_packets.contains_key(dest_addr) {
456            self.pending_destinations.remove(dest_addr);
457        }
458        payloads
459    }
460
461    pub(crate) fn has_traffic_for(&self, dest_addr: &NodeAddr) -> bool {
462        self.pending_destinations.contains(dest_addr)
463    }
464
465    pub(crate) fn tun_packets_for(&self, dest_addr: &NodeAddr) -> Option<&PendingTunPacketQueue> {
466        self.tun_packets.get(dest_addr)
467    }
468
469    pub(crate) fn endpoint_data_for(
470        &self,
471        dest_addr: &NodeAddr,
472    ) -> Option<&PendingEndpointDataQueue> {
473        self.endpoint_data.get(dest_addr)
474    }
475
476    pub(crate) fn tun_destination_count(&self) -> usize {
477        self.tun_packets.len()
478    }
479
480    pub(crate) fn tun_packet_count(&self) -> usize {
481        self.tun_packets.values().map(|q| q.len()).sum()
482    }
483}
484
485fn endpoint_tcp_payload_is_latency_sensitive(payload: &[u8], tcp_offset: usize) -> bool {
486    const TCP_MIN_HEADER_LEN: usize = 20;
487    const TCP_FLAG_FIN: u8 = 0x01;
488    const TCP_FLAG_SYN: u8 = 0x02;
489    const TCP_FLAG_RST: u8 = 0x04;
490    const INTERACTIVE_TCP_PAYLOAD_MAX: usize = 256;
491
492    if payload.len() < tcp_offset + TCP_MIN_HEADER_LEN {
493        return true;
494    }
495
496    let tcp_header_len = usize::from(payload[tcp_offset + 12] >> 4) * 4;
497    if tcp_header_len < TCP_MIN_HEADER_LEN || payload.len() < tcp_offset + tcp_header_len {
498        return true;
499    }
500
501    let flags = payload[tcp_offset + 13];
502    if flags & (TCP_FLAG_FIN | TCP_FLAG_SYN | TCP_FLAG_RST) != 0 {
503        return true;
504    }
505
506    let payload_len = endpoint_ip_payload_len(payload)
507        .and_then(|ip_payload_len| ip_payload_len.checked_sub(tcp_header_len))
508        .unwrap_or_else(|| payload.len().saturating_sub(tcp_offset + tcp_header_len));
509    payload_len <= INTERACTIVE_TCP_PAYLOAD_MAX
510}
511
512fn endpoint_ip_payload_len(payload: &[u8]) -> Option<usize> {
513    const IPV4_MIN_HEADER_LEN: usize = 20;
514    const IPV6_HEADER_LEN: usize = 40;
515
516    let version_ihl = payload.first().copied()?;
517    match version_ihl >> 4 {
518        4 => {
519            if payload.len() < IPV4_MIN_HEADER_LEN {
520                return None;
521            }
522            let header_len = usize::from(version_ihl & 0x0f) * 4;
523            if header_len < IPV4_MIN_HEADER_LEN || payload.len() < header_len {
524                return None;
525            }
526            let total_len = usize::from(u16::from_be_bytes([payload[2], payload[3]]));
527            total_len.checked_sub(header_len)
528        }
529        6 => {
530            if payload.len() < IPV6_HEADER_LEN {
531                return None;
532            }
533            Some(usize::from(u16::from_be_bytes([payload[4], payload[5]])))
534        }
535        _ => None,
536    }
537}
538
539fn parse_endpoint_payload_ip_proto(payload: &[u8]) -> Option<(u8, usize)> {
540    const IPV4_MIN_HEADER_LEN: usize = 20;
541
542    let version_ihl = payload.first().copied()?;
543
544    match version_ihl >> 4 {
545        4 => {
546            if payload.len() < IPV4_MIN_HEADER_LEN {
547                return None;
548            }
549            let header_len = usize::from(version_ihl & 0x0f) * 4;
550            if header_len >= IPV4_MIN_HEADER_LEN && payload.len() >= header_len {
551                Some((payload[9], header_len))
552            } else {
553                None
554            }
555        }
556        6 => ipv6_payload_next_header(payload),
557        _ => None,
558    }
559}
560
561#[cfg(test)]
562pub(in crate::node) fn endpoint_payload_is_tcp(payload: &[u8]) -> bool {
563    const IPPROTO_TCP: u8 = 6;
564    parse_endpoint_payload_ip_proto(payload).is_some_and(|(proto, _)| proto == IPPROTO_TCP)
565}
566
567fn ipv6_payload_next_header(payload: &[u8]) -> Option<(u8, usize)> {
568    const IPV6_HEADER_LEN: usize = 40;
569    const IPV6_FRAGMENT_HEADER_LEN: usize = 8;
570
571    if payload.len() < IPV6_HEADER_LEN || payload[0] >> 4 != 6 {
572        return None;
573    }
574
575    let mut next_header = payload[6];
576    let mut offset = IPV6_HEADER_LEN;
577    let mut extension_count = 0usize;
578    while ipv6_extension_header_is_skippable(next_header) {
579        if next_header == 44 {
580            if payload.len() < offset + IPV6_FRAGMENT_HEADER_LEN {
581                return None;
582            }
583            next_header = payload[offset];
584            offset += IPV6_FRAGMENT_HEADER_LEN;
585        } else if next_header == 51 {
586            if payload.len() < offset + 2 {
587                return None;
588            }
589            let header_len = (usize::from(payload[offset + 1]) + 2) * 4;
590            if payload.len() < offset + header_len {
591                return None;
592            }
593            next_header = payload[offset];
594            offset += header_len;
595        } else {
596            if payload.len() < offset + 2 {
597                return None;
598            }
599            let header_len = (usize::from(payload[offset + 1]) + 1) * 8;
600            if payload.len() < offset + header_len {
601                return None;
602            }
603            next_header = payload[offset];
604            offset += header_len;
605        }
606        extension_count += 1;
607        if extension_count > 8 {
608            return None;
609        }
610    }
611
612    Some((next_header, offset))
613}
614
615fn ipv6_extension_header_is_skippable(next_header: u8) -> bool {
616    matches!(next_header, 0 | 43 | 44 | 51 | 60 | 135)
617}
fips_core/node/endpoint_traffic.rs

fips_core/node/
endpoint_traffic.rs
