gasm_host/
net.rs

1//! gasm:net for the native runner: WebSocket client connections.
2//!
3//! Each connection runs on a background thread; the guest-facing API is
4//! non-blocking (queues + an atomic state), matching the browser runner.
5//! `ws://` and `wss://` (rustls, trusting the OS certificate store; set
6//! `SSL_CERT_FILE` to use a specific CA bundle instead).
7//!
8//! Limits keep a misbehaving guest or peer from exhausting the host: at most
9//! [`MAX_CONNECTIONS`] open at once, bounded queues in both directions (a full
10//! send queue makes `send` fail; a full receive queue stops reading the socket,
11//! so TCP pushes back on the peer), and timeouts for connecting and the handshake.
12
13use std::collections::{HashMap, VecDeque};
14use std::io::ErrorKind;
15use std::sync::atomic::{AtomicU32, Ordering};
16use std::sync::mpsc::{self, Receiver, SyncSender, TryRecvError, TrySendError};
17use std::sync::Arc;
18use std::time::Duration;
19
20use tungstenite::stream::MaybeTlsStream;
21use tungstenite::{Message, WebSocket};
22
23pub const CONNECTING: u32 = 0;
24pub const OPEN: u32 = 1;
25pub const CLOSED: u32 = 2;
26pub const ERROR: u32 = 3;
27
28pub const MAX_CONNECTIONS: usize = 16;
29/// Messages queued per direction per connection.
30const QUEUE: usize = 4096;
31const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
32
33struct Conn {
34    state: Arc<AtomicU32>,
35    outgoing: SyncSender<Vec<u8>>,
36    thread: Option<std::thread::JoinHandle<()>>,
37    incoming: Receiver<Vec<u8>>,
38    queue: VecDeque<Vec<u8>>,
39}
40
41pub struct Net {
42    allowed: bool,
43    conns: HashMap<i32, Conn>,
44    next: i32,
45}
46
47impl Net {
48    pub fn new(allowed: bool) -> Net {
49        Net { allowed, conns: HashMap::new(), next: 1 }
50    }
51
52    pub fn open(&mut self, url: &str) -> i32 {
53        if !self.allowed {
54            eprintln!("[gasm] net: denied connection to {url} (run with --allow-net)");
55            return -1;
56        }
57        if !url.starts_with("ws://") && !url.starts_with("wss://") {
58            eprintln!("[gasm] net: only ws:// and wss:// URLs are supported: {url}");
59            return -1;
60        }
61        if self.conns.len() >= MAX_CONNECTIONS {
62            eprintln!("[gasm] net: too many connections (max {MAX_CONNECTIONS}): {url}");
63            return -1;
64        }
65        install_crypto_provider();
66        let state = Arc::new(AtomicU32::new(CONNECTING));
67        let (out_tx, out_rx) = mpsc::sync_channel::<Vec<u8>>(QUEUE);
68        let (in_tx, in_rx) = mpsc::sync_channel::<Vec<u8>>(QUEUE);
69        let (st, url_owned) = (state.clone(), url.to_owned());
70        let thread = std::thread::spawn(move || run(&url_owned, &st, out_rx, in_tx));
71        let h = self.next;
72        self.next += 1;
73        self.conns.insert(h, Conn { state, outgoing: out_tx, incoming: in_rx, queue: VecDeque::new(), thread: Some(thread) });
74        h
75    }
76
77    /// A handle `open` never returned is an error (the guest traps); a closed
78    /// one is `Ok(false)`.
79    fn check(&self, h: i32) -> Result<bool, String> {
80        if h <= 0 || h >= self.next {
81            return Err(format!("gasm:net: invalid connection handle {h}"));
82        }
83        Ok(self.conns.contains_key(&h))
84    }
85
86    pub fn state(&self, h: i32) -> Result<u32, String> {
87        Ok(match self.check(h)? {
88            true => self.conns[&h].state.load(Ordering::Acquire),
89            false => CLOSED,
90        })
91    }
92
93    /// 0, or -1 if not open or the send queue is full.
94    pub fn send(&mut self, h: i32, msg: Vec<u8>) -> Result<i32, String> {
95        if !self.check(h)? {
96            return Ok(-1);
97        }
98        let c = &self.conns[&h];
99        if c.state.load(Ordering::Acquire) != OPEN {
100            return Ok(-1);
101        }
102        Ok(match c.outgoing.try_send(msg) {
103            Ok(()) => 0,
104            Err(TrySendError::Full(_)) => {
105                eprintln!("[gasm] net: send queue full ({QUEUE} messages)");
106                -1
107            }
108            Err(TrySendError::Disconnected(_)) => -1,
109        })
110    }
111
112    /// Next message without removing it; Ok(Err(0)) = none yet, Ok(Err(-1)) = closed and drained.
113    pub fn peek(&mut self, h: i32) -> Result<Result<&[u8], i32>, String> {
114        if !self.check(h)? {
115            return Ok(Err(-1));
116        }
117        Ok(self.peek_open(h))
118    }
119
120    fn peek_open(&mut self, h: i32) -> Result<&[u8], i32> {
121        let Some(c) = self.conns.get_mut(&h) else { return Err(-1) };
122        while let Ok(m) = c.incoming.try_recv() {
123            c.queue.push_back(m);
124        }
125        if c.queue.is_empty() {
126            return Err(if c.state.load(Ordering::Acquire) >= CLOSED { -1 } else { 0 });
127        }
128        Ok(c.queue.front().unwrap())
129    }
130
131    pub fn pop(&mut self, h: i32) {
132        if let Some(c) = self.conns.get_mut(&h) {
133            c.queue.pop_front();
134        }
135    }
136
137    /// Closing a closed connection does nothing.
138    pub fn close(&mut self, h: i32) -> Result<(), String> {
139        self.check(h)?;
140        self.conns.remove(&h); // dropping the sender ends the thread (after flushing)
141        Ok(())
142    }
143}
144
145impl Drop for Net {
146    /// Let connection threads flush what the guest already sent (e.g. the last
147    /// lockstep inputs before a game exits) instead of dropping it on exit.
148    fn drop(&mut self) {
149        let threads: Vec<_> = self.conns.drain().filter_map(|(_, mut c)| c.thread.take()).collect();
150        let deadline = std::time::Instant::now() + Duration::from_secs(2);
151        for t in threads {
152            while !t.is_finished() && std::time::Instant::now() < deadline {
153                std::thread::sleep(Duration::from_millis(5));
154            }
155        }
156    }
157}
158
159/// Connection thread: connect, then alternate between flushing outgoing
160/// messages and reading with a short timeout.
161fn run(url: &str, state: &AtomicU32, outgoing: Receiver<Vec<u8>>, incoming: SyncSender<Vec<u8>>) {
162    let mut ws = match connect(url) {
163        Ok(ws) => ws,
164        Err(e) => {
165            eprintln!("[gasm] net: {url}: {e}");
166            state.store(ERROR, Ordering::Release);
167            return;
168        }
169    };
170    // Non-blocking after the handshake: one thread alternates between sending and
171    // receiving. (Not read timeouts: on Windows a timed-out receive leaves the
172    // socket in an indeterminate state.)
173    let tcp = match ws.get_mut() {
174        MaybeTlsStream::Plain(s) => Some(&*s),
175        MaybeTlsStream::Rustls(s) => Some(&s.sock),
176        _ => None,
177    };
178    if let Some(tcp) = tcp {
179        let _ = tcp.set_nodelay(true);
180        if let Err(e) = tcp.set_nonblocking(true) {
181            eprintln!("[gasm] net: {url}: {e}");
182            state.store(ERROR, Ordering::Release);
183            return;
184        }
185    }
186    state.store(OPEN, Ordering::Release);
187    let end = pump(&mut ws, &outgoing, &incoming);
188    if let Err(e) = &end {
189        eprintln!("[gasm] net: {url}: {e}");
190    }
191    state.store(if end.is_ok() { CLOSED } else { ERROR }, Ordering::Release);
192    close_gracefully(&mut ws);
193}
194
195/// TCP connect and TLS/WebSocket handshakes, each bounded by [`CONNECT_TIMEOUT`].
196fn connect(url: &str) -> Result<WebSocket<MaybeTlsStream<std::net::TcpStream>>, String> {
197    use std::net::ToSocketAddrs;
198    let uri: tungstenite::http::Uri = url.parse().map_err(|e| format!("bad URL: {e}"))?;
199    let host = uri.host().ok_or("URL has no host")?.trim_start_matches('[').trim_end_matches(']');
200    let port = uri.port_u16().unwrap_or(if uri.scheme_str() == Some("wss") { 443 } else { 80 });
201    let mut last = String::from("no address");
202    for addr in (host, port).to_socket_addrs().map_err(|e| e.to_string())? {
203        match std::net::TcpStream::connect_timeout(&addr, CONNECT_TIMEOUT) {
204            Ok(tcp) => {
205                tcp.set_read_timeout(Some(CONNECT_TIMEOUT)).map_err(|e| e.to_string())?;
206                tcp.set_write_timeout(Some(CONNECT_TIMEOUT)).map_err(|e| e.to_string())?;
207                let (ws, _) = tungstenite::client_tls(url, tcp).map_err(|e| e.to_string())?;
208                let tcp = match ws.get_ref() {
209                    MaybeTlsStream::Plain(s) => s,
210                    MaybeTlsStream::Rustls(s) => &s.sock,
211                    _ => return Ok(ws),
212                };
213                tcp.set_read_timeout(None).map_err(|e| e.to_string())?;
214                tcp.set_write_timeout(None).map_err(|e| e.to_string())?;
215                return Ok(ws);
216            }
217            Err(e) => last = format!("{addr}: {e}"),
218        }
219    }
220    Err(last)
221}
222
223/// WebSocket close handshake, draining incoming data until the peer confirms.
224/// Exiting with unread data makes the OS reset the connection, and a reset can
225/// make the other side discard messages it hasn't read yet (e.g. our last inputs).
226fn close_gracefully(ws: &mut WebSocket<MaybeTlsStream<std::net::TcpStream>>) {
227    if ws.close(None).is_err() {
228        return;
229    }
230    let deadline = std::time::Instant::now() + Duration::from_secs(1);
231    while std::time::Instant::now() < deadline {
232        match ws.read() {
233            Ok(_) => {}
234            Err(e) if would_block(&e) => std::thread::sleep(Duration::from_millis(1)),
235            Err(_) => break, // ConnectionClosed: handshake complete
236        }
237    }
238}
239
240/// rustls needs a process-wide crypto backend; we use ring (see Cargo.toml).
241pub fn install_crypto_provider() {
242    static ONCE: std::sync::Once = std::sync::Once::new();
243    ONCE.call_once(|| {
244        let _ = rustls::crypto::ring::default_provider().install_default();
245    });
246}
247
248fn would_block(e: &tungstenite::Error) -> bool {
249    matches!(e, tungstenite::Error::Io(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut | ErrorKind::Interrupted))
250}
251
252/// Non-blocking pump: flush outgoing messages, read what's there, and sleep
253/// briefly only when idle. A send that can't complete now stays buffered in
254/// tungstenite and goes out with the next flush.
255pub(crate) fn pump(
256    ws: &mut WebSocket<MaybeTlsStream<std::net::TcpStream>>,
257    outgoing: &Receiver<Vec<u8>>,
258    incoming: &SyncSender<Vec<u8>>,
259) -> Result<(), String> {
260    // a message the guest's full receive queue couldn't take yet: read nothing more until it fits
261    let mut pending: Option<Vec<u8>> = None;
262    loop {
263        let mut busy = false;
264        loop {
265            match outgoing.try_recv() {
266                Ok(m) => {
267                    busy = true;
268                    match ws.send(Message::Binary(m.into())) {
269                        Ok(()) => {}
270                        Err(e) if would_block(&e) => {}
271                        Err(e) => return Err(e.to_string()),
272                    }
273                }
274                Err(TryRecvError::Empty) => break,
275                Err(TryRecvError::Disconnected) => {
276                    flush_all(ws);
277                    return Ok(()); // guest closed it
278                }
279            }
280        }
281        match ws.flush() {
282            Ok(()) => {}
283            Err(e) if would_block(&e) => {}
284            Err(e) => return Err(e.to_string()),
285        }
286        if let Some(m) = pending.take() {
287            match incoming.try_send(m) {
288                Ok(()) => busy = true,
289                Err(TrySendError::Full(m)) => pending = Some(m),
290                Err(TrySendError::Disconnected(_)) => return Ok(()),
291            }
292        }
293        let read = if pending.is_some() { Err(tungstenite::Error::Io(ErrorKind::WouldBlock.into())) } else { ws.read() };
294        match read {
295            Ok(Message::Binary(b)) => {
296                busy = true;
297                pending = Some(b.to_vec());
298            }
299            Ok(Message::Text(t)) => {
300                busy = true;
301                pending = Some(t.as_bytes().to_vec());
302            }
303            Ok(Message::Close(_)) => return Ok(()),
304            Ok(_) => busy = true,
305            Err(e) if would_block(&e) => {}
306            Err(tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed) => return Ok(()),
307            Err(e) => return Err(e.to_string()),
308        }
309        if !busy {
310            std::thread::sleep(Duration::from_millis(1));
311        }
312    }
313}
314
315/// Push out anything still buffered (bounded), e.g. before closing.
316fn flush_all(ws: &mut WebSocket<MaybeTlsStream<std::net::TcpStream>>) {
317    let deadline = std::time::Instant::now() + Duration::from_secs(1);
318    while std::time::Instant::now() < deadline {
319        match ws.flush() {
320            Err(e) if would_block(&e) => std::thread::sleep(Duration::from_millis(1)),
321            _ => return,
322        }
323    }
324}
325
326#[cfg(test)]
327mod tests {
328    use super::*;
329
330    /// Wait (up to 10 s) for a connection to leave CONNECTING; return its state.
331    fn settle(net: &Net, h: i32) -> u32 {
332        let t0 = std::time::Instant::now();
333        while net.state(h).unwrap() == CONNECTING && t0.elapsed() < Duration::from_secs(10) {
334            std::thread::sleep(Duration::from_millis(20));
335        }
336        net.state(h).unwrap()
337    }
338
339    /// Public TLS endpoint, OS trust store. Run with: cargo test --release -- --ignored
340    #[test]
341    #[ignore = "needs internet"]
342    fn wss_public_echo() {
343        let mut net = Net::new(true);
344        let h = net.open("wss://echo.websocket.org");
345        assert!(h > 0);
346        assert_eq!(settle(&net, h), OPEN, "TLS connection did not open");
347        assert_eq!(net.send(h, b"gasm-tls-check".to_vec()), Ok(0));
348        let t0 = std::time::Instant::now();
349        let mut echoed = false;
350        while t0.elapsed() < Duration::from_secs(10) && !echoed {
351            match net.peek(h).unwrap() {
352                Ok(m) => {
353                    echoed = m == b"gasm-tls-check";
354                    net.pop(h);
355                }
356                Err(0) => std::thread::sleep(Duration::from_millis(20)),
357                Err(_) => break,
358            }
359        }
360        assert!(echoed, "no echo over wss://");
361    }
362
363    #[test]
364    fn denied_without_permission() {
365        assert_eq!(Net::new(false).open("wss://example.com"), -1);
366        assert_eq!(Net::new(true).open("http://example.com"), -1);
367    }
368
369    #[test]
370    fn unknown_handles_are_errors_closed_ones_are_not() {
371        let mut net = Net::new(true);
372        assert!(net.state(1).is_err() && net.state(0).is_err() && net.close(7).is_err());
373        let h = net.open("ws://127.0.0.1:1"); // nothing listens on port 1
374        assert!(h > 0);
375        net.close(h).unwrap();
376        assert_eq!(net.state(h), Ok(CLOSED));
377        assert_eq!(net.send(h, vec![1]), Ok(-1));
378        assert_eq!(net.peek(h).unwrap().err(), Some(-1));
379        net.close(h).unwrap(); // again: nothing happens
380    }
381}
gasm_host/net.rs

gasm_host/
net.rs
