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        gravity: 0,
83        recursive_prs: false,
84        announces_from_internal: true,
85        announces_to_internal: None,
86        out_capable: true,
87        in_capable: true,
88        bitrate: None,
89        airtime_profile: None,
90        announce_rate_target: None,
91        announce_rate_grace: 0,
92        announce_rate_penalty: 0.0,
93        announce_cap: 1.0,
94        is_local_client: false,
95        wants_tunnel: false,
96        tunnel_id: None,
97        mtu: 1400,
98        ia_freq: 0.0,
99        ip_freq: 0.0,
100        op_freq: 0.0,
101        op_samples: 0,
102        started: 0.0,
103        ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
104    };
105
106    let running_clone = running.clone();
107
108    // Spawn reader + keepalive thread
109    thread::Builder::new()
110        .name(format!("direct-udp-{}", &name))
111        .spawn(move || {
112            run_reader(
113                socket,
114                peer_addr,
115                interface_id,
116                session_id,
117                punch_token,
118                tx,
119                running_clone,
120            );
121        })?;
122
123    Ok((Box::new(writer), info))
124}
125
126fn run_reader(
127    socket: Arc<UdpSocket>,
128    peer_addr: SocketAddr,
129    interface_id: InterfaceId,
130    session_id: [u8; 16],
131    punch_token: [u8; 32],
132    tx: EventSender,
133    running: Arc<AtomicBool>,
134) {
135    let mut buf = [0u8; 2048];
136    let mut last_inbound = std::time::Instant::now();
137    let mut last_keepalive = std::time::Instant::now();
138    let keepalive_pkt = puncher::build_keepalive_packet(&session_id, &punch_token);
139
140    while running.load(Ordering::Relaxed) {
141        // Send keepalive if due
142        if last_keepalive.elapsed() >= KEEPALIVE_INTERVAL {
143            let _ = socket.send_to(&keepalive_pkt, peer_addr);
144            last_keepalive = std::time::Instant::now();
145        }
146
147        // Check inactivity timeout
148        if last_inbound.elapsed() >= INACTIVITY_TIMEOUT {
149            log::info!("[{}] Direct UDP interface timed out", interface_id.0);
150            let _ = tx.send(Event::InterfaceDown(interface_id));
151            break;
152        }
153
154        match socket.recv_from(&mut buf) {
155            Ok((len, src)) => {
156                // Only accept packets from our verified peer
157                if src != peer_addr {
158                    continue;
159                }
160                last_inbound = std::time::Instant::now();
161
162                // Skip keepalive/punch packets (they have RNSH/RNSA magic)
163                if len >= 4 && (buf[..4] == *b"RNSH" || buf[..4] == *b"RNSA") {
164                    continue;
165                }
166
167                // Deliver as Reticulum frame
168                if len > 0 {
169                    let _ = tx.send(Event::Frame {
170                        interface_id,
171                        data: buf[..len].to_vec(),
172                        rssi: None,
173                        snr: None,
174                    });
175                }
176            }
177            Err(ref e)
178                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut =>
179            {
180                continue;
181            }
182            Err(e) => {
183                log::warn!("[{}] Direct UDP recv error: {}", interface_id.0, e);
184                let _ = tx.send(Event::InterfaceDown(interface_id));
185                break;
186            }
187        }
188    }
189}
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194
195    #[test]
196    fn test_udp_direct_writer() {
197        // Create a pair of sockets
198        let sock_a = UdpSocket::bind("127.0.0.1:0").unwrap();
199        let sock_b = UdpSocket::bind("127.0.0.1:0").unwrap();
200        let addr_b = sock_b.local_addr().unwrap();
201        sock_b
202            .set_read_timeout(Some(Duration::from_secs(1)))
203            .unwrap();
204
205        let mut writer = UdpDirectWriter {
206            socket: Arc::new(sock_a),
207            peer_addr: addr_b,
208            running: Arc::new(AtomicBool::new(true)),
209        };
210
211        // Send a frame
212        let data = b"hello direct peer";
213        writer.send_frame(data).unwrap();
214
215        // Receive on other end
216        let mut buf = [0u8; 64];
217        let (len, _src) = sock_b.recv_from(&mut buf).unwrap();
218        assert_eq!(&buf[..len], data);
219    }
220}
rns_net/holepunch/udp_direct.rs

rns_net/holepunch/
udp_direct.rs
