rust_raknet/
nethernet.rs

1//! TCP signaling forwarding for Bedrock servers using NetherNet.
2//!
3//! NetherNet uses HTTP signaling to exchange WebRTC session descriptions, then
4//! sends gameplay data directly between the client and the Bedrock host. This
5//! proxy forwards only the TCP signaling connection to a NetherNet-capable
6//! backend; it does not implement WebRTC or relay gameplay traffic.
7
8use std::{io, net::SocketAddr};
9use tokio::{
10    io::copy_bidirectional,
11    net::{TcpListener, TcpStream},
12};
13
14/// Forwards NetherNet's TCP signaling connections to a NetherNet-capable server.
15///
16/// The client must still be able to reach the backend's advertised WebRTC ICE
17/// candidate directly for gameplay. This is useful when the signaling endpoint
18/// needs a different local address or port from the backend.
19///
20/// # Example
21/// ```no_run
22/// # async fn example() -> std::io::Result<()> {
23/// let proxy = rust_raknet::NetherNetProxy::bind(
24///     "127.0.0.1:19144".parse().unwrap(),
25///     "127.0.0.1:19142".parse().unwrap(),
26/// ).await?;
27/// proxy.run().await
28/// # }
29/// ```
30pub struct NetherNetProxy {
31    listener: TcpListener,
32    upstream: SocketAddr,
33    connection_limit: usize,
34}
35
36impl NetherNetProxy {
37    /// Binds a TCP signaling endpoint and forwards each connection to `upstream`.
38    pub async fn bind(listen: SocketAddr, upstream: SocketAddr) -> io::Result<Self> {
39        Ok(Self {
40            listener: TcpListener::bind(listen).await?,
41            upstream,
42            connection_limit: 1024,
43        })
44    }
45
46    /// Returns the local TCP address used for NetherNet signaling.
47    pub fn local_addr(&self) -> io::Result<SocketAddr> {
48        self.listener.local_addr()
49    }
50
51    /// Sets the maximum number of simultaneous signaling connections (default: 1024).
52    pub fn with_connection_limit(mut self, limit: std::num::NonZeroUsize) -> Self {
53        self.connection_limit = limit.get();
54        self
55    }
56
57    /// Accepts and forwards signaling connections until an I/O error occurs.
58    ///
59    /// Call this in a Tokio task if the application needs to stop it by
60    /// cancelling that task. Cancellation also aborts its active forwarding tasks.
61    pub async fn run(&self) -> io::Result<()> {
62        let mut connections = tokio::task::JoinSet::new();
63        loop {
64            tokio::select! {
65                _ = connections.join_next(), if !connections.is_empty() => {}
66                accepted = self.listener.accept(), if connections.len() < self.connection_limit => {
67                    let (client, peer) = accepted?;
68                    let upstream = self.upstream;
69                    connections.spawn(async move {
70                        if let Err(error) = forward(client, upstream).await {
71                            crate::raknet_log_error!("NetherNet signaling relay for {peer} failed: {error}");
72                        }
73                    });
74                }
75            }
76        }
77    }
78}
79
80async fn forward(mut client: TcpStream, upstream: SocketAddr) -> io::Result<()> {
81    let mut upstream_stream = tokio::time::timeout(
82        std::time::Duration::from_secs(10),
83        TcpStream::connect(upstream),
84    )
85    .await
86    .map_err(|_| {
87        io::Error::new(
88            io::ErrorKind::TimedOut,
89            "NetherNet upstream connection timed out",
90        )
91    })??;
92    copy_bidirectional(&mut client, &mut upstream_stream).await?;
93    Ok(())
94}
95
96#[cfg(test)]
97mod tests {
98    use super::NetherNetProxy;
99    use std::time::Duration;
100    use tokio::{
101        io::{AsyncReadExt, AsyncWriteExt},
102        net::{TcpListener, TcpStream},
103        time::timeout,
104    };
105
106    #[tokio::test]
107    async fn closes_client_when_upstream_is_unavailable() {
108        let unavailable = TcpListener::bind("127.0.0.1:0").await.unwrap();
109        let upstream_addr = unavailable.local_addr().unwrap();
110        drop(unavailable);
111
112        let proxy = NetherNetProxy::bind("127.0.0.1:0".parse().unwrap(), upstream_addr)
113            .await
114            .unwrap();
115        let proxy_addr = proxy.local_addr().unwrap();
116        let proxy_task = tokio::spawn(async move { proxy.run().await });
117
118        let result = timeout(Duration::from_secs(3), async {
119            let mut client = TcpStream::connect(proxy_addr).await.unwrap();
120            client
121                .write_all(b"GET /v1/join HTTP/1.1\r\n\r\n")
122                .await
123                .unwrap();
124            let mut response = Vec::new();
125            client.read_to_end(&mut response).await.map(|_| response)
126        })
127        .await
128        .expect("client was not closed after the upstream connection failed");
129
130        assert!(result.is_err() || result.as_ref().is_ok_and(Vec::is_empty));
131        proxy_task.abort();
132        let _ = proxy_task.await;
133    }
134
135    #[tokio::test]
136    async fn forwards_signaling_bytes_in_both_directions() {
137        let upstream = TcpListener::bind("127.0.0.1:0").await.unwrap();
138        let upstream_addr = upstream.local_addr().unwrap();
139        let upstream_task = tokio::spawn(async move {
140            let (mut stream, _) = upstream.accept().await.unwrap();
141            let mut request = Vec::new();
142            let mut chunk = [0; 512];
143            loop {
144                let count = stream.read(&mut chunk).await.unwrap();
145                if count == 0 {
146                    break;
147                }
148                request.extend_from_slice(&chunk[..count]);
149                if request.windows(4).any(|window| window == b"\r\n\r\n") {
150                    break;
151                }
152            }
153            assert!(request.starts_with(b"GET /v1/join HTTP/1.1\r\n"));
154            stream
155                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
156                .await
157                .unwrap();
158        });
159
160        let proxy = NetherNetProxy::bind("127.0.0.1:0".parse().unwrap(), upstream_addr)
161            .await
162            .unwrap();
163        let proxy_addr = proxy.local_addr().unwrap();
164        let proxy_task = tokio::spawn(async move { proxy.run().await });
165
166        let result = timeout(Duration::from_secs(3), async {
167            let mut client = TcpStream::connect(proxy_addr).await.unwrap();
168            client
169                .write_all(b"GET /v1/join HTTP/1.1\r\nHost: localhost\r\n\r\n")
170                .await
171                .unwrap();
172            client.shutdown().await.unwrap();
173            let mut response = Vec::new();
174            client.read_to_end(&mut response).await.unwrap();
175            response
176        })
177        .await
178        .expect("signaling forward timed out");
179
180        assert!(result.starts_with(b"HTTP/1.1 200 OK\r\n"));
181        assert!(result.ends_with(b"\r\nok"));
182        upstream_task.await.unwrap();
183        proxy_task.abort();
184        let _ = proxy_task.await;
185    }
186}
rust_raknet/nethernet.rs

rust_raknet/
nethernet.rs
