gasm_host/
fetch.rs

1//! gasm:fetch for the native runner: HTTP(S) requests made for the guest
2//! (design/fetch.md).
3//!
4//! Each request runs on a background thread (ureq, rustls with ring, the OS
5//! certificate store or `SSL_CERT_FILE`); the guest polls a handle, so nothing
6//! blocks a frame. Redirects are followed here, not by ureq, so every hop is
7//! checked against the host list. The body arrives in chunks through a bounded
8//! channel: a guest that stops reading pauses the download.
9//!
10//! Headless runs can record responses (`--fetch-record DIR`) and replay them
11//! (`--fetch-replay DIR`): a replayed request completes at the start of the
12//! next frame, on every runner, so hashes compare. The record format and key
13//! are shared with runners/web/lib/fetch.js.
14
15use std::collections::{HashMap, VecDeque};
16use std::io::Read;
17use std::path::{Path, PathBuf};
18use std::sync::atomic::{AtomicBool, Ordering};
19use std::sync::mpsc::{self, Receiver, SyncSender, TryRecvError};
20use std::sync::{Arc, OnceLock};
21use std::time::Duration;
22
23use crate::net::{url_host, NetPolicy};
24
25pub const PENDING: u32 = 0;
26pub const HEADERS: u32 = 1;
27pub const DONE: u32 = 2;
28pub const FAILED: u32 = 3;
29
30pub const MAX_REQUESTS: usize = 16;
31/// Largest response body; a larger one fails the request.
32pub const MAX_BODY: u64 = 64 << 20;
33pub const MAX_REQUEST_BODY: usize = 16 << 20;
34const MAX_REDIRECTS: usize = 10;
35const CHUNK: usize = 64 << 10;
36/// Chunks in flight per request (4 MiB): then the thread waits for the guest.
37const QUEUE: usize = 64;
38
39pub const METHODS: [&str; 7] = ["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"];
40/// Request headers browsers don't let pages set (the Fetch standard's forbidden
41/// request headers, plus user-agent): a request with one fails on every runner,
42/// so a game can't come to depend on it. Same list in runners/web/lib/fetch.js.
43pub const FORBIDDEN_HEADERS: [&str; 22] = [
44    "accept-charset", "accept-encoding", "access-control-request-headers", "access-control-request-method",
45    "connection", "content-length", "cookie", "cookie2", "date", "dnt", "expect", "host", "keep-alive",
46    "origin", "referer", "set-cookie", "te", "trailer", "transfer-encoding", "upgrade", "user-agent", "via",
47];
48pub const FORBIDDEN_PREFIXES: [&str; 2] = ["proxy-", "sec-"];
49/// Response headers no runner reports: connection-level ones (browsers' fetch() hides
50/// them or they differ by HTTP version) and cookies (fetch() never exposes them).
51/// Same list in runners/web/lib/fetch.js.
52pub const HIDDEN_RESPONSE_HEADERS: [&str; 9] =
53    ["connection", "keep-alive", "proxy-connection", "set-cookie", "set-cookie2", "te", "trailer", "transfer-encoding", "upgrade"];
54
55/// Response headers as the guest sees them, the way fetch()'s Headers lists them:
56/// `name: value` lines sorted by (lowercase) name, repeated names joined with ", ".
57/// The body the guest reads is always decoded (gzip natively; browsers decode
58/// everything), so `content-encoding` is left out, and `content-length` too when it
59/// counted encoded bytes.
60pub fn header_lines<'a>(headers: impl IntoIterator<Item = (&'a str, String)>) -> String {
61    let mut by_name: std::collections::BTreeMap<String, Vec<String>> = Default::default();
62    for (k, v) in headers {
63        let k = k.to_ascii_lowercase();
64        if !HIDDEN_RESPONSE_HEADERS.contains(&k.as_str()) {
65            by_name.entry(k).or_default().push(v.trim().to_owned());
66        }
67    }
68    if let Some(enc) = by_name.remove("content-encoding") {
69        if enc.iter().any(|e| !e.eq_ignore_ascii_case("identity")) {
70            by_name.remove("content-length");
71        }
72    }
73    by_name.into_iter().map(|(k, v)| format!("{k}: {}\n", v.join(", "))).collect()
74}
75
76/// Where responses come from.
77#[derive(Clone, Debug, Default)]
78pub enum FetchMode {
79    #[default]
80    Live,
81    /// live, and every completed response is stored in the directory
82    Record(PathBuf),
83    /// never the network: responses from the directory, complete at the next frame
84    Replay(PathBuf),
85}
86
87/// A request the guest described (validated).
88#[derive(Debug, PartialEq)]
89pub struct Desc {
90    pub method: String,
91    pub url: String,
92    pub headers: Vec<(String, String)>,
93}
94
95/// Parse and check a request description; Err says why it's refused.
96pub fn parse_desc(json: &str, body_len: usize) -> Result<Desc, String> {
97    let v: serde_json::Value = serde_json::from_str(json).map_err(|e| format!("invalid JSON: {e}"))?;
98    let o = v.as_object().ok_or("the description must be a JSON object")?;
99    let method = match o.get("method") {
100        None => "GET".to_owned(),
101        Some(m) => m.as_str().ok_or("method must be a string")?.to_ascii_uppercase(),
102    };
103    if !METHODS.contains(&method.as_str()) {
104        return Err(format!("method {method} is not supported"));
105    }
106    let url = o.get("url").and_then(|u| u.as_str()).ok_or("url must be a string")?.to_owned();
107    let lower = url.to_ascii_lowercase();
108    if !(lower.starts_with("http://") || lower.starts_with("https://")) || url_host(&url).is_none() {
109        return Err(format!("not an absolute http(s) URL: {url}"));
110    }
111    if url.chars().any(|c| c.is_ascii_control() || c == ' ') {
112        return Err(format!("invalid URL: {url}"));
113    }
114    let mut headers = Vec::new();
115    if let Some(h) = o.get("headers") {
116        for (k, v) in h.as_object().ok_or("headers must be an object")? {
117            let v = v.as_str().ok_or_else(|| format!("header {k}: the value must be a string"))?;
118            check_header(k, v)?;
119            headers.push((k.to_ascii_lowercase(), v.to_owned()));
120        }
121    }
122    if body_len > 0 && (method == "GET" || method == "HEAD") {
123        return Err(format!("a {method} request can't have a body"));
124    }
125    if body_len > MAX_REQUEST_BODY {
126        return Err(format!("request body over {MAX_REQUEST_BODY} bytes"));
127    }
128    Ok(Desc { method, url, headers })
129}
130
131fn check_header(name: &str, value: &str) -> Result<(), String> {
132    let token = |c: u8| c.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|~".contains(&c);
133    if name.is_empty() || !name.bytes().all(token) {
134        return Err(format!("invalid header name {name:?}"));
135    }
136    if value.bytes().any(|c| c == b'\r' || c == b'\n' || c == 0) {
137        return Err(format!("invalid value for header {name}"));
138    }
139    let n = name.to_ascii_lowercase();
140    if FORBIDDEN_HEADERS.contains(&n.as_str()) || FORBIDDEN_PREFIXES.iter().any(|p| n.starts_with(p)) {
141        return Err(format!("header {name} can't be set (browsers refuse it)"));
142    }
143    Ok(())
144}
145
146/// FNV-1a 64 of method, URL and body: the name of a recorded response.
147pub fn record_key(method: &str, url: &str, body: &[u8]) -> String {
148    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
149    for b in method.bytes().chain([0]).chain(url.bytes()).chain([0]).chain(body.iter().copied()) {
150        h = (h ^ b as u64).wrapping_mul(0x0000_0100_0000_01b3);
151    }
152    format!("{h:016x}")
153}
154
155enum Event {
156    Head(i32, String),
157    Chunk(Vec<u8>),
158    Done,
159    Fail(String),
160}
161
162struct Req {
163    events: Option<Receiver<Event>>,
164    cancel: Arc<AtomicBool>,
165    state: u32,
166    status: i32,
167    headers: Option<String>,
168    body: VecDeque<u8>,
169    /// replayed: nothing is visible before this frame
170    ready_frame: u64,
171}
172
173pub struct Fetch {
174    policy: NetPolicy,
175    mode: FetchMode,
176    reqs: HashMap<i32, Req>,
177    next: i32,
178    /// hosts already logged as denied (once each)
179    denied: std::collections::HashSet<String>,
180}
181
182impl Default for Fetch {
183    fn default() -> Fetch {
184        Fetch::new(NetPolicy::default(), FetchMode::Live)
185    }
186}
187
188impl Fetch {
189    pub fn new(policy: NetPolicy, mode: FetchMode) -> Fetch {
190        Fetch { policy, mode, reqs: HashMap::new(), next: 1, denied: Default::default() }
191    }
192
193    /// Start a request; a handle > 0 or -1 (logged). `frame` is the current frame index.
194    pub fn request(&mut self, desc: &str, body: Vec<u8>, frame: u64) -> i32 {
195        let d = match parse_desc(desc, body.len()) {
196            Ok(d) => d,
197            Err(e) => {
198                eprintln!("[gasm] fetch: refused: {e}");
199                return -1;
200            }
201        };
202        if self.reqs.len() >= MAX_REQUESTS {
203            eprintln!("[gasm] fetch: too many requests (max {MAX_REQUESTS}): {}", d.url);
204            return -1;
205        }
206        let cancel = Arc::new(AtomicBool::new(false));
207        let mut req = Req { events: None, cancel: cancel.clone(), state: PENDING, status: 0, headers: None, body: VecDeque::new(), ready_frame: 0 };
208        match &self.mode {
209            FetchMode::Replay(dir) => {
210                req.ready_frame = frame + 1;
211                match load_record(dir, &record_key(&d.method, &d.url, &body)) {
212                    Some((status, headers, data)) => {
213                        (req.state, req.status, req.headers, req.body) = (DONE, status, Some(headers), data.into());
214                    }
215                    None => {
216                        eprintln!("[gasm] fetch: {} {} is not recorded in {}", d.method, d.url, dir.display());
217                        req.state = FAILED;
218                    }
219                }
220            }
221            mode => {
222                if let Some(why) = self.policy.refusal(&d.url) {
223                    let host = url_host(&d.url).unwrap_or_default();
224                    if self.denied.insert(host) {
225                        eprintln!("[gasm] fetch: denied {} ({why})", d.url);
226                    }
227                    return -1;
228                }
229                let record = match mode {
230                    FetchMode::Record(dir) => Some(dir.join(record_key(&d.method, &d.url, &body))),
231                    _ => None,
232                };
233                let (tx, rx) = mpsc::sync_channel(QUEUE);
234                let policy = self.policy.clone();
235                std::thread::spawn(move || run(d, body, policy, tx, cancel, record));
236                req.events = Some(rx);
237            }
238        }
239        let h = self.next;
240        self.next += 1;
241        self.reqs.insert(h, req);
242        h
243    }
244
245    /// A handle `request` never returned is an error (the guest traps); a closed one is None.
246    fn get(&mut self, h: i32) -> Result<Option<&mut Req>, String> {
247        if h <= 0 || h >= self.next {
248            return Err(format!("gasm:fetch: invalid handle {h}"));
249        }
250        let Some(r) = self.reqs.get_mut(&h) else { return Ok(None) };
251        if let Some(rx) = &r.events {
252            loop {
253                match rx.try_recv() {
254                    Ok(Event::Head(status, headers)) => (r.state, r.status, r.headers) = (HEADERS, status, Some(headers)),
255                    Ok(Event::Chunk(c)) => r.body.extend(c),
256                    Ok(Event::Done) => r.state = DONE,
257                    Ok(Event::Fail(e)) => {
258                        eprintln!("[gasm] fetch: {e}");
259                        (r.state, r.status) = (FAILED, 0);
260                        r.body.clear();
261                    }
262                    Err(TryRecvError::Empty) => break,
263                    Err(TryRecvError::Disconnected) => {
264                        if r.state < DONE {
265                            (r.state, r.status) = (FAILED, 0);
266                        }
267                        break;
268                    }
269                }
270            }
271            if r.state >= DONE {
272                r.events = None;
273            }
274        }
275        Ok(Some(r))
276    }
277
278    /// (state, status, headers) as the guest may see them now.
279    fn view(&mut self, h: i32, frame: u64) -> Result<(u32, i32, Option<String>), String> {
280        Ok(match self.get(h)? {
281            None => (FAILED, 0, None),
282            Some(r) if frame < r.ready_frame => (PENDING, 0, None),
283            Some(r) => (r.state, r.status, r.headers.clone()),
284        })
285    }
286
287    pub fn state(&mut self, h: i32, frame: u64) -> Result<u32, String> {
288        Ok(self.view(h, frame)?.0)
289    }
290
291    pub fn status(&mut self, h: i32, frame: u64) -> Result<i32, String> {
292        Ok(self.view(h, frame)?.1)
293    }
294
295    pub fn headers(&mut self, h: i32, frame: u64) -> Result<Option<String>, String> {
296        Ok(self.view(h, frame)?.2)
297    }
298
299    /// Up to `cap` body bytes: Ok(bytes) (empty: none yet), or Err(()) when done and drained or failed.
300    pub fn read(&mut self, h: i32, cap: usize, frame: u64) -> Result<Result<Vec<u8>, ()>, String> {
301        let Some(r) = self.get(h)? else { return Ok(Err(())) };
302        if frame < r.ready_frame || r.state == PENDING {
303            return Ok(Ok(Vec::new()));
304        }
305        if r.body.is_empty() {
306            return Ok(if r.state >= DONE { Err(()) } else { Ok(Vec::new()) });
307        }
308        let n = cap.min(r.body.len());
309        Ok(Ok(r.body.drain(..n).collect()))
310    }
311
312    pub fn close(&mut self, h: i32) -> Result<(), String> {
313        if h <= 0 || h >= self.next {
314            return Err(format!("gasm:fetch: invalid handle {h}"));
315        }
316        if let Some(r) = self.reqs.remove(&h) {
317            r.cancel.store(true, Ordering::Relaxed); // the thread also stops when its channel closes
318        }
319        Ok(())
320    }
321}
322
323fn load_record(dir: &Path, key: &str) -> Option<(i32, String, Vec<u8>)> {
324    let meta: serde_json::Value = serde_json::from_slice(&std::fs::read(dir.join(format!("{key}.json"))).ok()?).ok()?;
325    let body = std::fs::read(dir.join(format!("{key}.body"))).ok()?;
326    Some((meta.get("status")?.as_i64()? as i32, meta.get("headers")?.as_str()?.to_owned(), body))
327}
328
329fn save_record(path: &Path, method: &str, url: &str, status: i32, headers: &str, body: &[u8]) {
330    let meta = serde_json::json!({ "method": method, "url": url, "status": status, "headers": headers });
331    let write = || -> std::io::Result<()> {
332        if let Some(d) = path.parent() {
333            std::fs::create_dir_all(d)?;
334        }
335        std::fs::write(path.with_extension("body"), body)?;
336        std::fs::write(path.with_extension("json"), format!("{}\n", serde_json::to_string_pretty(&meta)?))
337    };
338    if let Err(e) = write() {
339        eprintln!("[gasm] fetch: --fetch-record {}: {e}", path.display());
340    }
341}
342
343fn agent() -> &'static ureq::Agent {
344    static AGENT: OnceLock<ureq::Agent> = OnceLock::new();
345    AGENT.get_or_init(|| {
346        // the OS's trusted roots (or SSL_CERT_FILE), as for wss:// in gasm:net
347        let roots = rustls_native_certs::load_native_certs();
348        for e in &roots.errors {
349            eprintln!("[gasm] fetch: certificates: {e}");
350        }
351        let certs = roots.certs.iter().map(|c| ureq::tls::Certificate::from_der(c.as_ref()).to_owned()).collect();
352        let tls = ureq::tls::TlsConfig::builder()
353            .provider(ureq::tls::TlsProvider::Rustls)
354            .root_certs(ureq::tls::RootCerts::Specific(Arc::new(certs)))
355            .unversioned_rustls_crypto_provider(Arc::new(rustls::crypto::ring::default_provider()))
356            .build();
357        ureq::Agent::config_builder()
358            .http_status_as_error(false)
359            .max_redirects(0)
360            .max_redirects_will_error(false)
361            .timeout_connect(Some(Duration::from_secs(30)))
362            .timeout_global(Some(Duration::from_secs(120)))
363            .tls_config(tls)
364            .build()
365            .into()
366    })
367}
368
369/// `location` relative to `base` (absolute, origin-relative, or path-relative).
370fn resolve(base: &str, location: &str) -> String {
371    if location.contains("://") {
372        return location.to_owned();
373    }
374    let (scheme, rest) = base.split_once("://").unwrap_or(("http", base));
375    let authority = rest.split(['/', '?', '#']).next().unwrap_or("");
376    if let Some(l) = location.strip_prefix("//") {
377        return format!("{scheme}://{l}");
378    }
379    if location.starts_with('/') {
380        return format!("{scheme}://{authority}{location}");
381    }
382    let path = rest[authority.len()..].split(['?', '#']).next().unwrap_or("");
383    let dir = &path[..path.rfind('/').map_or(0, |i| i + 1)];
384    format!("{scheme}://{authority}{}{location}", if dir.is_empty() { "/" } else { dir })
385}
386
387fn run(d: Desc, body: Vec<u8>, policy: NetPolicy, tx: SyncSender<Event>, cancel: Arc<AtomicBool>, record: Option<PathBuf>) {
388    let result = (|| -> Result<(), String> {
389        let (mut method, mut url, mut body) = (d.method.clone(), d.url.clone(), body);
390        let mut hops = 0;
391        let resp = loop {
392            let mut b = ureq::http::Request::builder().method(method.as_str()).uri(url.as_str());
393            for (k, v) in &d.headers {
394                b = b.header(k.as_str(), v.as_str());
395            }
396            let req = b.body(body.clone()).map_err(|e| format!("{url}: {e}"))?;
397            let resp = agent().run(req).map_err(|e| format!("{method} {url}: {e}"))?;
398            let status = resp.status().as_u16();
399            let location = resp.headers().get("location").and_then(|v| v.to_str().ok()).map(str::to_owned);
400            match (status, location) {
401                (301 | 302 | 303 | 307 | 308, Some(loc)) => {
402                    hops += 1;
403                    if hops > MAX_REDIRECTS {
404                        return Err(format!("{url}: too many redirects"));
405                    }
406                    let next = resolve(&url, &loc);
407                    if let Some(why) = policy.refusal(&next) {
408                        return Err(format!("{url}: redirect to {next} denied ({why})"));
409                    }
410                    // as browsers: 303, and 301/302 after a POST, continue as a GET without a body
411                    if status == 303 || (matches!(status, 301 | 302) && method == "POST") {
412                        if method != "HEAD" {
413                            method = "GET".into();
414                        }
415                        body.clear();
416                    }
417                    url = next;
418                }
419                _ => break resp,
420            }
421        };
422        let status = resp.status().as_u16() as i32;
423        let headers = header_lines(resp.headers().iter().map(|(k, v)| (k.as_str(), String::from_utf8_lossy(v.as_bytes()).into_owned())));
424        tx.send(Event::Head(status, headers.clone())).map_err(|_| String::new())?;
425        let mut reader = resp.into_body().into_with_config().limit(MAX_BODY).reader();
426        let mut buf = vec![0u8; CHUNK];
427        let mut all = Vec::new();
428        loop {
429            if cancel.load(Ordering::Relaxed) {
430                return Err(String::new());
431            }
432            let n = reader.read(&mut buf).map_err(|e| format!("{url}: {e}"))?;
433            if n == 0 {
434                break;
435            }
436            if record.is_some() {
437                all.extend_from_slice(&buf[..n]);
438            }
439            tx.send(Event::Chunk(buf[..n].to_vec())).map_err(|_| String::new())?;
440        }
441        if let Some(path) = &record {
442            save_record(path, &d.method, &d.url, status, &headers, &all);
443        }
444        tx.send(Event::Done).map_err(|_| String::new())
445    })();
446    if let Err(e) = result {
447        if !e.is_empty() {
448            let _ = tx.send(Event::Fail(e));
449        }
450    }
451}
452
453#[cfg(test)]
454mod tests {
455    use super::*;
456
457    #[test]
458    fn descriptions() {
459        let d = parse_desc(r#"{"url":"https://api.met.no/x?a=1","headers":{"Accept":"application/json"}}"#, 0).unwrap();
460        assert_eq!(d, Desc { method: "GET".into(), url: "https://api.met.no/x?a=1".into(), headers: vec![("accept".into(), "application/json".into())] });
461        assert_eq!(parse_desc(r#"{"method":"post","url":"http://h/"}"#, 3).unwrap().method, "POST");
462        for bad in [
463            r#"[]"#,
464            r#"{"url":"ftp://h/"}"#,
465            r#"{"url":"/relative"}"#,
466            r#"{"method":"TRACE","url":"http://h/"}"#,
467            r#"{"url":"http://h/","headers":{"Cookie":"a"}}"#,
468            r#"{"url":"http://h/","headers":{"sec-fetch-mode":"a"}}"#,
469            r#"{"url":"http://h/","headers":{"x":"a\r\nb"}}"#,
470            r#"{"url":"http://h/","headers":{"bad name":"a"}}"#,
471        ] {
472            assert!(parse_desc(bad, 0).is_err(), "{bad}");
473        }
474        assert!(parse_desc(r#"{"url":"http://h/"}"#, 1).is_err()); // GET with a body
475    }
476
477    #[test]
478    fn response_headers() {
479        let h = header_lines([("Vary", "a".into()), ("content-type", "text/plain".into()), ("Connection", "close".into()), ("vary", " b ".into()), ("set-cookie", "x=1".into())]);
480        assert_eq!(h, "content-type: text/plain\nvary: a, b\n");
481        let gz = header_lines([("content-encoding", "gzip".into()), ("content-length", "10".into()), ("etag", "x".into())]);
482        assert_eq!(gz, "etag: x\n");
483    }
484
485    #[test]
486    fn redirects_resolve() {
487        assert_eq!(resolve("https://a.b/x/y?q", "/z"), "https://a.b/z");
488        assert_eq!(resolve("https://a.b/x/y?q", "z"), "https://a.b/x/z");
489        assert_eq!(resolve("https://a.b", "z"), "https://a.b/z");
490        assert_eq!(resolve("https://a.b/x", "//c.d/e"), "https://c.d/e");
491        assert_eq!(resolve("https://a.b/x", "http://c.d/"), "http://c.d/");
492    }
493
494    #[test]
495    fn keys() {
496        // the same in runners/web/lib/fetch.js (recordKey)
497        assert_eq!(record_key("GET", "https://example.org/", b""), "a0eecdfd36f8bde7");
498    }
499
500    #[test]
501    fn replay_and_handles() {
502        let dir = std::env::temp_dir().join(format!("gasm-fetch-test-{}", std::process::id()));
503        let _ = std::fs::remove_dir_all(&dir);
504        let key = record_key("GET", "https://example.org/a", b"");
505        save_record(&dir.join(&key), "GET", "https://example.org/a", 200, "content-type: text/plain\n", b"hello");
506        let mut f = Fetch::new(NetPolicy::default(), FetchMode::Replay(dir.clone()));
507        let h = f.request(r#"{"url":"https://example.org/a"}"#, vec![], 5);
508        let miss = f.request(r#"{"url":"https://example.org/b"}"#, vec![], 5);
509        assert_eq!((f.state(h, 5).unwrap(), f.status(h, 5).unwrap(), f.read(h, 9, 5).unwrap()), (PENDING, 0, Ok(vec![])));
510        assert_eq!((f.state(h, 6).unwrap(), f.status(h, 6).unwrap()), (DONE, 200));
511        assert_eq!(f.headers(h, 6).unwrap().as_deref(), Some("content-type: text/plain\n"));
512        assert_eq!(f.read(h, 3, 6).unwrap(), Ok(b"hel".to_vec()));
513        assert_eq!(f.read(h, 9, 6).unwrap(), Ok(b"lo".to_vec()));
514        assert_eq!(f.read(h, 9, 6).unwrap(), Err(()));
515        assert_eq!(f.state(miss, 6).unwrap(), FAILED);
516        f.close(h).unwrap();
517        assert_eq!(f.state(h, 7).unwrap(), FAILED); // closed
518        assert!(f.state(99, 7).is_err()); // never returned: traps
519        let _ = std::fs::remove_dir_all(&dir);
520        // live requests are denied without --allow-net
521        assert_eq!(Fetch::default().request(r#"{"url":"https://example.org/"}"#, vec![], 0), -1);
522    }
523}
gasm_host/fetch.rs

gasm_host/
fetch.rs
