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
41/// The hosts guests may reach (gasm:net and gasm:fetch): none, all (`--allow-net`)
42/// or a list (`--allow-net=api.met.no,*.example.org`; `*.` matches subdomains). With
43/// `consent` (the window runner) any other host is the player's choice.
44#[derive(Clone, Debug, Default)]
45pub struct NetPolicy {
46    pub allowed: bool,
47    /// empty: every host (when allowed)
48    pub hosts: Vec<String>,
49    pub consent: Option<crate::consent::Consent>,
50}
51
52impl NetPolicy {
53    pub fn new(allowed: bool, hosts: Vec<String>) -> NetPolicy {
54        NetPolicy { allowed, hosts: hosts.into_iter().map(|h| h.to_ascii_lowercase()).collect(), consent: None }
55    }
56
57    pub fn with_consent(mut self, consent: Option<crate::consent::Consent>) -> NetPolicy {
58        self.consent = consent;
59        self
60    }
61
62    /// Why a URL is refused, None if it may be reached now, or Err(()) while the
63    /// player hasn't answered (the question is queued).
64    pub fn verdict(&self, url: &str) -> Result<Option<String>, ()> {
65        let Some(why) = self.refusal(url) else { return Ok(None) };
66        let Some(c) = &self.consent else { return Ok(Some(why)) };
67        let host = url_host(url).unwrap_or_default();
68        match c.lock().unwrap().check(&crate::consent::net_subject(&host)) {
69            Some(true) => Ok(None),
70            Some(false) => Ok(Some(format!("the player said no to {host}"))),
71            None => Err(()),
72        }
73    }
74
75    /// Block (on a connection's own thread) until the URL may be reached: Err(why) if
76    /// refused, or once `cancelled` says the request is gone.
77    pub fn wait(&self, url: &str, cancelled: impl Fn() -> bool) -> Result<(), String> {
78        loop {
79            match self.verdict(url) {
80                Ok(None) => return Ok(()),
81                Ok(Some(why)) => return Err(why),
82                Err(()) if cancelled() => return Err("cancelled".into()),
83                Err(()) => std::thread::sleep(Duration::from_millis(50)),
84            }
85        }
86    }
87
88    pub fn permits(&self, host: &str) -> bool {
89        if !self.allowed {
90            return false;
91        }
92        let host = host.trim_end_matches('.').to_ascii_lowercase();
93        self.hosts.is_empty()
94            || self.hosts.iter().any(|p| match p.strip_prefix("*.") {
95                Some(domain) => host.len() > domain.len() + 1 && host.ends_with(domain) && host.as_bytes()[host.len() - domain.len() - 1] == b'.',
96                None => host == *p,
97            })
98    }
99
100    /// Why a URL is refused (for the log), or None if it may be reached.
101    pub fn refusal(&self, url: &str) -> Option<String> {
102        if !self.allowed {
103            return Some("run with --allow-net".into());
104        }
105        let host = url_host(url).unwrap_or_default();
106        (!self.permits(&host)).then(|| format!("{host} is not in --allow-net={}", self.hosts.join(",")))
107    }
108}
109
110/// The host of an absolute URL (no userinfo, no port, brackets kept for IPv6).
111pub fn url_host(url: &str) -> Option<String> {
112    let rest = url.split_once("://")?.1;
113    let authority = rest.split(['/', '?', '#']).next()?;
114    let authority = authority.rsplit_once('@').map_or(authority, |(_, a)| a);
115    let host = if authority.starts_with('[') {
116        &authority[..=authority.find(']')?]
117    } else {
118        authority.split(':').next()?
119    };
120    (!host.is_empty()).then(|| host.to_ascii_lowercase())
121}
122
123pub struct Net {
124    policy: NetPolicy,
125    conns: HashMap<i32, Conn>,
126    next: i32,
127}
128
129impl Net {
130    pub fn new(allowed: bool) -> Net {
131        Net::with_policy(NetPolicy::new(allowed, Vec::new()))
132    }
133
134    pub fn with_policy(policy: NetPolicy) -> Net {
135        Net { policy, conns: HashMap::new(), next: 1 }
136    }
137
138    pub fn open(&mut self, url: &str) -> i32 {
139        // refused now, or (Err) the player is asked and the thread waits for the answer
140        if let Ok(Some(why)) = self.policy.verdict(url) {
141            eprintln!("[gasm] net: denied connection to {url} ({why})");
142            return -1;
143        }
144        if !url.starts_with("ws://") && !url.starts_with("wss://") {
145            eprintln!("[gasm] net: only ws:// and wss:// URLs are supported: {url}");
146            return -1;
147        }
148        if self.conns.len() >= MAX_CONNECTIONS {
149            eprintln!("[gasm] net: too many connections (max {MAX_CONNECTIONS}): {url}");
150            return -1;
151        }
152        install_crypto_provider();
153        let state = Arc::new(AtomicU32::new(CONNECTING));
154        let (out_tx, out_rx) = mpsc::sync_channel::<Vec<u8>>(QUEUE);
155        let (in_tx, in_rx) = mpsc::sync_channel::<Vec<u8>>(QUEUE);
156        let (st, url_owned, policy) = (state.clone(), url.to_owned(), self.policy.clone());
157        let thread = std::thread::spawn(move || {
158            // the player's answer first (closing the connection meanwhile drops out_tx)
159            let gone = || matches!(out_rx.try_recv(), Err(TryRecvError::Disconnected));
160            if let Err(why) = policy.wait(&url_owned, gone) {
161                eprintln!("[gasm] net: denied connection to {url_owned} ({why})");
162                st.store(ERROR, Ordering::Release);
163                return;
164            }
165            run(&url_owned, &st, out_rx, in_tx)
166        });
167        let h = self.next;
168        self.next += 1;
169        self.conns.insert(h, Conn { state, outgoing: out_tx, incoming: in_rx, queue: VecDeque::new(), thread: Some(thread) });
170        h
171    }
172
173    /// A handle `open` never returned is an error (the guest traps); a closed
174    /// one is `Ok(false)`.
175    fn check(&self, h: i32) -> Result<bool, String> {
176        if h <= 0 || h >= self.next {
177            return Err(format!("gasm:net: invalid connection handle {h}"));
178        }
179        Ok(self.conns.contains_key(&h))
180    }
181
182    pub fn state(&self, h: i32) -> Result<u32, String> {
183        Ok(match self.check(h)? {
184            true => self.conns[&h].state.load(Ordering::Acquire),
185            false => CLOSED,
186        })
187    }
188
189    /// 0, or -1 if not open or the send queue is full.
190    pub fn send(&mut self, h: i32, msg: Vec<u8>) -> Result<i32, String> {
191        if !self.check(h)? {
192            return Ok(-1);
193        }
194        let c = &self.conns[&h];
195        if c.state.load(Ordering::Acquire) != OPEN {
196            return Ok(-1);
197        }
198        Ok(match c.outgoing.try_send(msg) {
199            Ok(()) => 0,
200            Err(TrySendError::Full(_)) => {
201                eprintln!("[gasm] net: send queue full ({QUEUE} messages)");
202                -1
203            }
204            Err(TrySendError::Disconnected(_)) => -1,
205        })
206    }
207
208    /// Next message without removing it; Ok(Err(0)) = none yet, Ok(Err(-1)) = closed and drained.
209    pub fn peek(&mut self, h: i32) -> Result<Result<&[u8], i32>, String> {
210        if !self.check(h)? {
211            return Ok(Err(-1));
212        }
213        Ok(self.peek_open(h))
214    }
215
216    fn peek_open(&mut self, h: i32) -> Result<&[u8], i32> {
217        let Some(c) = self.conns.get_mut(&h) else { return Err(-1) };
218        while let Ok(m) = c.incoming.try_recv() {
219            c.queue.push_back(m);
220        }
221        if c.queue.is_empty() {
222            return Err(if c.state.load(Ordering::Acquire) >= CLOSED { -1 } else { 0 });
223        }
224        Ok(c.queue.front().unwrap())
225    }
226
227    pub fn pop(&mut self, h: i32) {
228        if let Some(c) = self.conns.get_mut(&h) {
229            c.queue.pop_front();
230        }
231    }
232
233    /// Closing a closed connection does nothing.
234    pub fn close(&mut self, h: i32) -> Result<(), String> {
235        self.check(h)?;
236        self.conns.remove(&h); // dropping the sender ends the thread (after flushing)
237        Ok(())
238    }
239}
240
241impl Drop for Net {
242    /// Let connection threads flush what the guest already sent (e.g. the last
243    /// lockstep inputs before a game exits) instead of dropping it on exit.
244    fn drop(&mut self) {
245        let threads: Vec<_> = self.conns.drain().filter_map(|(_, mut c)| c.thread.take()).collect();
246        let deadline = std::time::Instant::now() + Duration::from_secs(2);
247        for t in threads {
248            while !t.is_finished() && std::time::Instant::now() < deadline {
249                std::thread::sleep(Duration::from_millis(5));
250            }
251        }
252    }
253}
254
255/// Connection thread: connect, then alternate between flushing outgoing
256/// messages and reading with a short timeout.
257fn run(url: &str, state: &AtomicU32, outgoing: Receiver<Vec<u8>>, incoming: SyncSender<Vec<u8>>) {
258    let mut ws = match connect(url) {
259        Ok(ws) => ws,
260        Err(e) => {
261            eprintln!("[gasm] net: {url}: {e}");
262            state.store(ERROR, Ordering::Release);
263            return;
264        }
265    };
266    // Non-blocking after the handshake: one thread alternates between sending and
267    // receiving. (Not read timeouts: on Windows a timed-out receive leaves the
268    // socket in an indeterminate state.)
269    let tcp = match ws.get_mut() {
270        MaybeTlsStream::Plain(s) => Some(&*s),
271        MaybeTlsStream::Rustls(s) => Some(&s.sock),
272        _ => None,
273    };
274    if let Some(tcp) = tcp {
275        let _ = tcp.set_nodelay(true);
276        if let Err(e) = tcp.set_nonblocking(true) {
277            eprintln!("[gasm] net: {url}: {e}");
278            state.store(ERROR, Ordering::Release);
279            return;
280        }
281    }
282    state.store(OPEN, Ordering::Release);
283    let end = pump(&mut ws, &outgoing, &incoming);
284    if let Err(e) = &end {
285        eprintln!("[gasm] net: {url}: {e}");
286    }
287    state.store(if end.is_ok() { CLOSED } else { ERROR }, Ordering::Release);
288    close_gracefully(&mut ws);
289}
290
291/// TCP connect and TLS/WebSocket handshakes, each bounded by [`CONNECT_TIMEOUT`].
292fn connect(url: &str) -> Result<WebSocket<MaybeTlsStream<std::net::TcpStream>>, String> {
293    use std::net::ToSocketAddrs;
294    let uri: tungstenite::http::Uri = url.parse().map_err(|e| format!("bad URL: {e}"))?;
295    let host = uri.host().ok_or("URL has no host")?.trim_start_matches('[').trim_end_matches(']');
296    let port = uri.port_u16().unwrap_or(if uri.scheme_str() == Some("wss") { 443 } else { 80 });
297    let mut last = String::from("no address");
298    for addr in (host, port).to_socket_addrs().map_err(|e| e.to_string())? {
299        match std::net::TcpStream::connect_timeout(&addr, CONNECT_TIMEOUT) {
300            Ok(tcp) => {
301                tcp.set_read_timeout(Some(CONNECT_TIMEOUT)).map_err(|e| e.to_string())?;
302                tcp.set_write_timeout(Some(CONNECT_TIMEOUT)).map_err(|e| e.to_string())?;
303                let (ws, _) = tungstenite::client_tls(url, tcp).map_err(|e| e.to_string())?;
304                let tcp = match ws.get_ref() {
305                    MaybeTlsStream::Plain(s) => s,
306                    MaybeTlsStream::Rustls(s) => &s.sock,
307                    _ => return Ok(ws),
308                };
309                tcp.set_read_timeout(None).map_err(|e| e.to_string())?;
310                tcp.set_write_timeout(None).map_err(|e| e.to_string())?;
311                return Ok(ws);
312            }
313            Err(e) => last = format!("{addr}: {e}"),
314        }
315    }
316    Err(last)
317}
318
319/// WebSocket close handshake, draining incoming data until the peer confirms.
320/// Exiting with unread data makes the OS reset the connection, and a reset can
321/// make the other side discard messages it hasn't read yet (e.g. our last inputs).
322fn close_gracefully(ws: &mut WebSocket<MaybeTlsStream<std::net::TcpStream>>) {
323    if ws.close(None).is_err() {
324        return;
325    }
326    let deadline = std::time::Instant::now() + Duration::from_secs(1);
327    while std::time::Instant::now() < deadline {
328        match ws.read() {
329            Ok(_) => {}
330            Err(e) if would_block(&e) => std::thread::sleep(Duration::from_millis(1)),
331            Err(_) => break, // ConnectionClosed: handshake complete
332        }
333    }
334}
335
336/// rustls needs a process-wide crypto backend; we use ring (see Cargo.toml).
337pub fn install_crypto_provider() {
338    static ONCE: std::sync::Once = std::sync::Once::new();
339    ONCE.call_once(|| {
340        let _ = rustls::crypto::ring::default_provider().install_default();
341    });
342}
343
344fn would_block(e: &tungstenite::Error) -> bool {
345    matches!(e, tungstenite::Error::Io(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut | ErrorKind::Interrupted))
346}
347
348/// Non-blocking pump: flush outgoing messages, read what's there, and sleep
349/// briefly only when idle. A send that can't complete now stays buffered in
350/// tungstenite and goes out with the next flush.
351pub(crate) fn pump(
352    ws: &mut WebSocket<MaybeTlsStream<std::net::TcpStream>>,
353    outgoing: &Receiver<Vec<u8>>,
354    incoming: &SyncSender<Vec<u8>>,
355) -> Result<(), String> {
356    // a message the guest's full receive queue couldn't take yet: read nothing more until it fits
357    let mut pending: Option<Vec<u8>> = None;
358    loop {
359        let mut busy = false;
360        loop {
361            match outgoing.try_recv() {
362                Ok(m) => {
363                    busy = true;
364                    match ws.send(Message::Binary(m.into())) {
365                        Ok(()) => {}
366                        Err(e) if would_block(&e) => {}
367                        Err(e) => return Err(e.to_string()),
368                    }
369                }
370                Err(TryRecvError::Empty) => break,
371                Err(TryRecvError::Disconnected) => {
372                    flush_all(ws);
373                    return Ok(()); // guest closed it
374                }
375            }
376        }
377        match ws.flush() {
378            Ok(()) => {}
379            Err(e) if would_block(&e) => {}
380            Err(e) => return Err(e.to_string()),
381        }
382        if let Some(m) = pending.take() {
383            match incoming.try_send(m) {
384                Ok(()) => busy = true,
385                Err(TrySendError::Full(m)) => pending = Some(m),
386                Err(TrySendError::Disconnected(_)) => return Ok(()),
387            }
388        }
389        let read = if pending.is_some() { Err(tungstenite::Error::Io(ErrorKind::WouldBlock.into())) } else { ws.read() };
390        match read {
391            Ok(Message::Binary(b)) => {
392                busy = true;
393                pending = Some(b.to_vec());
394            }
395            Ok(Message::Text(t)) => {
396                busy = true;
397                pending = Some(t.as_bytes().to_vec());
398            }
399            Ok(Message::Close(_)) => return Ok(()),
400            Ok(_) => busy = true,
401            Err(e) if would_block(&e) => {}
402            Err(tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed) => return Ok(()),
403            Err(e) => return Err(e.to_string()),
404        }
405        if !busy {
406            std::thread::sleep(Duration::from_millis(1));
407        }
408    }
409}
410
411/// Push out anything still buffered (bounded), e.g. before closing.
412fn flush_all(ws: &mut WebSocket<MaybeTlsStream<std::net::TcpStream>>) {
413    let deadline = std::time::Instant::now() + Duration::from_secs(1);
414    while std::time::Instant::now() < deadline {
415        match ws.flush() {
416            Err(e) if would_block(&e) => std::thread::sleep(Duration::from_millis(1)),
417            _ => return,
418        }
419    }
420}
421
422#[cfg(test)]
423mod policy_tests {
424    use super::*;
425
426    #[test]
427    fn hosts() {
428        assert_eq!(url_host("https://User@API.met.no:443/x?y").as_deref(), Some("api.met.no"));
429        assert_eq!(url_host("http://[::1]:8080/").as_deref(), Some("[::1]"));
430        assert_eq!(url_host("ws://localhost").as_deref(), Some("localhost"));
431        let p = NetPolicy::new(true, vec!["api.met.no".into(), "*.example.org".into()]);
432        assert!(p.permits("api.met.no") && p.permits("API.MET.NO.") && p.permits("a.b.example.org"));
433        assert!(!p.permits("met.no") && !p.permits("example.org") && !p.permits("badexample.org"));
434        assert!(NetPolicy::new(true, vec![]).permits("anything"));
435        assert!(!NetPolicy::new(false, vec![]).permits("anything"));
436    }
437}
438
439#[cfg(test)]
440mod tests {
441    use super::*;
442
443    /// Wait (up to 10 s) for a connection to leave CONNECTING; return its state.
444    fn settle(net: &Net, h: i32) -> u32 {
445        let t0 = std::time::Instant::now();
446        while net.state(h).unwrap() == CONNECTING && t0.elapsed() < Duration::from_secs(10) {
447            std::thread::sleep(Duration::from_millis(20));
448        }
449        net.state(h).unwrap()
450    }
451
452    /// Public TLS endpoint, OS trust store. Run with: cargo test --release -- --ignored
453    #[test]
454    #[ignore = "needs internet"]
455    fn wss_public_echo() {
456        let mut net = Net::new(true);
457        let h = net.open("wss://echo.websocket.org");
458        assert!(h > 0);
459        assert_eq!(settle(&net, h), OPEN, "TLS connection did not open");
460        assert_eq!(net.send(h, b"gasm-tls-check".to_vec()), Ok(0));
461        let t0 = std::time::Instant::now();
462        let mut echoed = false;
463        while t0.elapsed() < Duration::from_secs(10) && !echoed {
464            match net.peek(h).unwrap() {
465                Ok(m) => {
466                    echoed = m == b"gasm-tls-check";
467                    net.pop(h);
468                }
469                Err(0) => std::thread::sleep(Duration::from_millis(20)),
470                Err(_) => break,
471            }
472        }
473        assert!(echoed, "no echo over wss://");
474    }
475
476    #[test]
477    fn denied_without_permission() {
478        assert_eq!(Net::new(false).open("wss://example.com"), -1);
479        assert_eq!(Net::new(true).open("http://example.com"), -1);
480    }
481
482    #[test]
483    fn unknown_handles_are_errors_closed_ones_are_not() {
484        let mut net = Net::new(true);
485        assert!(net.state(1).is_err() && net.state(0).is_err() && net.close(7).is_err());
486        let h = net.open("ws://127.0.0.1:1"); // nothing listens on port 1
487        assert!(h > 0);
488        net.close(h).unwrap();
489        assert_eq!(net.state(h), Ok(CLOSED));
490        assert_eq!(net.send(h, vec![1]), Ok(-1));
491        assert_eq!(net.peek(h).unwrap().err(), Some(-1));
492        net.close(h).unwrap(); // again: nothing happens
493    }
494}
gasm_host/net.rs

gasm_host/
net.rs
