fips_core/upper/
tun_outbound.rs

1use tokio::sync::mpsc;
2
3#[derive(Debug, Clone)]
4pub struct TunOutboundTx {
5    bulk: mpsc::Sender<Vec<u8>>,
6}
7
8#[derive(Debug)]
9pub struct TunOutboundRx {
10    bulk: mpsc::Receiver<Vec<u8>>,
11}
12
13pub(crate) fn tun_outbound_channel(capacity: usize) -> (TunOutboundTx, TunOutboundRx) {
14    let capacity = capacity.max(1);
15    let (bulk_tx, bulk_rx) = mpsc::channel(capacity);
16    (
17        TunOutboundTx { bulk: bulk_tx },
18        TunOutboundRx { bulk: bulk_rx },
19    )
20}
21
22impl TunOutboundTx {
23    pub async fn send(&self, packet: Vec<u8>) -> Result<(), mpsc::error::SendError<Vec<u8>>> {
24        self.bulk.send(packet).await
25    }
26
27    pub fn blocking_send(&self, packet: Vec<u8>) -> Result<(), mpsc::error::SendError<Vec<u8>>> {
28        self.bulk.blocking_send(packet)
29    }
30
31    pub fn try_send(&self, packet: Vec<u8>) -> Result<(), mpsc::error::TrySendError<Vec<u8>>> {
32        self.bulk.try_send(packet)
33    }
34
35    #[cfg(any(test, target_os = "linux", target_os = "macos", windows))]
36    pub(crate) fn admit_from_tun_reader(
37        &self,
38        packet: Vec<u8>,
39    ) -> Result<(), mpsc::error::SendError<Vec<u8>>> {
40        match self.bulk.try_send(packet) {
41            Ok(()) => Ok(()),
42            Err(mpsc::error::TrySendError::Full(packet)) => {
43                crate::perf_profile::record_tun_outbound_admission_drop();
44                tracing::debug!(
45                    len = packet.len(),
46                    "Dropping TUN outbound packet because admission queue is full"
47                );
48                Ok(())
49            }
50            Err(mpsc::error::TrySendError::Closed(packet)) => Err(mpsc::error::SendError(packet)),
51        }
52    }
53}
54
55impl TunOutboundRx {
56    pub(crate) async fn recv(&mut self) -> Option<Vec<u8>> {
57        self.bulk.recv().await
58    }
59
60    pub(crate) fn try_recv(&mut self) -> Result<Vec<u8>, mpsc::error::TryRecvError> {
61        self.bulk.try_recv()
62    }
63}
64
65#[cfg(test)]
66mod tests {
67    use super::*;
68
69    fn ipv4_packet(proto: u8, body_len: usize) -> Vec<u8> {
70        let total_len = 20 + body_len;
71        let mut packet = vec![0u8; total_len];
72        packet[0] = 0x45;
73        packet[2..4].copy_from_slice(&(total_len as u16).to_be_bytes());
74        packet[9] = proto;
75        packet
76    }
77
78    fn ipv4_tcp_bulk_packet() -> Vec<u8> {
79        let mut packet = ipv4_packet(6, 20 + 300);
80        let tcp_offset = 20;
81        packet[tcp_offset + 12] = 5 << 4;
82        packet[tcp_offset + 13] = 0x10;
83        packet
84    }
85
86    fn ipv4_icmp_packet() -> Vec<u8> {
87        ipv4_packet(1, 8)
88    }
89
90    #[tokio::test]
91    async fn tun_outbound_recv_preserves_app_packet_order() {
92        let (tx, mut rx) = tun_outbound_channel(4);
93        let bulk = ipv4_tcp_bulk_packet();
94        let icmp = ipv4_icmp_packet();
95
96        tx.try_send(bulk.clone())
97            .expect("first app packet should enqueue");
98        tx.try_send(icmp.clone())
99            .expect("second app packet should enqueue");
100
101        assert_eq!(rx.recv().await, Some(bulk));
102        assert_eq!(rx.recv().await, Some(icmp));
103    }
104
105    #[test]
106    fn tun_outbound_capacity_applies_to_icmp_app_payload() {
107        let (tx, mut rx) = tun_outbound_channel(1);
108        let first_bulk = ipv4_tcp_bulk_packet();
109        let icmp = ipv4_icmp_packet();
110
111        tx.try_send(first_bulk.clone())
112            .expect("first app packet should fit");
113        assert!(
114            tx.try_send(icmp).is_err(),
115            "ICMP app payload should share bounded app packet capacity"
116        );
117
118        assert_eq!(rx.try_recv(), Ok(first_bulk));
119    }
120
121    #[test]
122    fn tun_reader_admission_sheds_icmp_app_payload_when_full() {
123        let (tx, mut rx) = tun_outbound_channel(1);
124        let first_bulk = ipv4_tcp_bulk_packet();
125        let icmp = ipv4_icmp_packet();
126
127        assert!(tx.admit_from_tun_reader(first_bulk.clone()).is_ok());
128        assert!(tx.admit_from_tun_reader(icmp).is_ok());
129
130        assert_eq!(rx.try_recv(), Ok(first_bulk));
131        assert!(matches!(
132            rx.try_recv(),
133            Err(mpsc::error::TryRecvError::Empty)
134        ));
135    }
136}
fips_core/upper/tun_outbound.rs

fips_core/upper/
tun_outbound.rs
