rns_net/holepunch/
udp_direct.rs

1//! Point-to-point UDP direct interface.
2//!
3//! After hole punching succeeds, this module wraps the punched UDP socket
4//! as a Reticulum interface (implementing Writer + reader thread).
5//! No HDLC framing needed — one datagram = one packet (same as UdpInterface).
6
7use std::io;
8use std::net::{SocketAddr, UdpSocket};
9use std::sync::atomic::{AtomicBool, Ordering};
10use std::sync::Arc;
11use std::thread;
12use std::time::Duration;
13
14use rns_core::transport::types::{InterfaceId, InterfaceInfo};
15
16use crate::event::{Event, EventSender};
17use crate::interface::Writer;
18
19use super::puncher;
20
21/// Keepalive interval for maintaining NAT mappings.
22const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
23
24/// Timeout for considering the direct connection dead.
25const INACTIVITY_TIMEOUT: Duration = Duration::from_secs(120);
26
27/// Writer for a direct UDP peer-to-peer connection.
28///
29/// When dropped, signals the reader thread to stop.
30pub struct UdpDirectWriter {
31    socket: Arc<UdpSocket>,
32    peer_addr: SocketAddr,
33    running: Arc<AtomicBool>,
34}
35
36impl Writer for UdpDirectWriter {
37    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
38        self.socket.send_to(data, self.peer_addr)?;
39        Ok(())
40    }
41}
42
43impl Drop for UdpDirectWriter {
44    fn drop(&mut self) {
45        self.running.store(false, Ordering::Relaxed);
46    }
47}
48
49/// Start a direct UDP interface from a punched socket.
50///
51/// Returns `(interface_id, writer, InterfaceInfo)` and spawns a reader thread.
52/// The reader thread sends `Event::Frame` for incoming packets and
53/// `Event::InterfaceDown` if the connection times out.
54pub fn start_direct_interface(
55    socket: UdpSocket,
56    peer_addr: SocketAddr,
57    interface_id: InterfaceId,
58    session_id: [u8; 16],
59    punch_token: [u8; 32],
60    tx: EventSender,
61) -> io::Result<(Box<dyn Writer>, InterfaceInfo)> {
62    let socket = Arc::new(socket);
63    socket.set_read_timeout(Some(Duration::from_secs(1)))?;
64
65    let running = Arc::new(AtomicBool::new(true));
66
67    let writer = UdpDirectWriter {
68        socket: socket.clone(),
69        peer_addr,
70        running: running.clone(),
71    };
72
73    let name = format!(
74        "DirectPeer/{:02x}{:02x}{:02x}{:02x}",
75        session_id[0], session_id[1], session_id[2], session_id[3]
76    );
77
78    let info = InterfaceInfo {
79        id: interface_id,
80        name: name.clone(),
81        mode: rns_core::constants::MODE_FULL,
82        recursive_prs: false,
83        announces_from_internal: true,
84        out_capable: true,
85        in_capable: true,
86        bitrate: None,
87        airtime_profile: None,
88        announce_rate_target: None,
89        announce_rate_grace: 0,
90        announce_rate_penalty: 0.0,
91        announce_cap: 1.0,
92        is_local_client: false,
93        wants_tunnel: false,
94        tunnel_id: None,
95        mtu: 1400,
96        ia_freq: 0.0,
97        ip_freq: 0.0,
98        op_freq: 0.0,
99        op_samples: 0,
100        started: 0.0,
101        ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
102    };
103
104    let running_clone = running.clone();
105
106    // Spawn reader + keepalive thread
107    thread::Builder::new()
108        .name(format!("direct-udp-{}", &name))
109        .spawn(move || {
110            run_reader(
111                socket,
112                peer_addr,
113                interface_id,
114                session_id,
115                punch_token,
116                tx,
117                running_clone,
118            );
119        })?;
120
121    Ok((Box::new(writer), info))
122}
123
124fn run_reader(
125    socket: Arc<UdpSocket>,
126    peer_addr: SocketAddr,
127    interface_id: InterfaceId,
128    session_id: [u8; 16],
129    punch_token: [u8; 32],
130    tx: EventSender,
131    running: Arc<AtomicBool>,
132) {
133    let mut buf = [0u8; 2048];
134    let mut last_inbound = std::time::Instant::now();
135    let mut last_keepalive = std::time::Instant::now();
136    let keepalive_pkt = puncher::build_keepalive_packet(&session_id, &punch_token);
137
138    while running.load(Ordering::Relaxed) {
139        // Send keepalive if due
140        if last_keepalive.elapsed() >= KEEPALIVE_INTERVAL {
141            let _ = socket.send_to(&keepalive_pkt, peer_addr);
142            last_keepalive = std::time::Instant::now();
143        }
144
145        // Check inactivity timeout
146        if last_inbound.elapsed() >= INACTIVITY_TIMEOUT {
147            log::info!("[{}] Direct UDP interface timed out", interface_id.0);
148            let _ = tx.send(Event::InterfaceDown(interface_id));
149            break;
150        }
151
152        match socket.recv_from(&mut buf) {
153            Ok((len, src)) => {
154                // Only accept packets from our verified peer
155                if src != peer_addr {
156                    continue;
157                }
158                last_inbound = std::time::Instant::now();
159
160                // Skip keepalive/punch packets (they have RNSH/RNSA magic)
161                if len >= 4 && (buf[..4] == *b"RNSH" || buf[..4] == *b"RNSA") {
162                    continue;
163                }
164
165                // Deliver as Reticulum frame
166                if len > 0 {
167                    let _ = tx.send(Event::Frame {
168                        interface_id,
169                        data: buf[..len].to_vec(),
170                        rssi: None,
171                        snr: None,
172                    });
173                }
174            }
175            Err(ref e)
176                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut =>
177            {
178                continue;
179            }
180            Err(e) => {
181                log::warn!("[{}] Direct UDP recv error: {}", interface_id.0, e);
182                let _ = tx.send(Event::InterfaceDown(interface_id));
183                break;
184            }
185        }
186    }
187}
188
189#[cfg(test)]
190mod tests {
191    use super::*;
192
193    #[test]
194    fn test_udp_direct_writer() {
195        // Create a pair of sockets
196        let sock_a = UdpSocket::bind("127.0.0.1:0").unwrap();
197        let sock_b = UdpSocket::bind("127.0.0.1:0").unwrap();
198        let addr_b = sock_b.local_addr().unwrap();
199        sock_b
200            .set_read_timeout(Some(Duration::from_secs(1)))
201            .unwrap();
202
203        let mut writer = UdpDirectWriter {
204            socket: Arc::new(sock_a),
205            peer_addr: addr_b,
206            running: Arc::new(AtomicBool::new(true)),
207        };
208
209        // Send a frame
210        let data = b"hello direct peer";
211        writer.send_frame(data).unwrap();
212
213        // Receive on other end
214        let mut buf = [0u8; 64];
215        let (len, _src) = sock_b.recv_from(&mut buf).unwrap();
216        assert_eq!(&buf[..len], data);
217    }
218}
rns_net/holepunch/udp_direct.rs

rns_net/holepunch/
udp_direct.rs
