diff --git a/crates/notebook/Cargo.toml b/crates/notebook/Cargo.toml index 63ef60a46c7b52d4346aa3058b0923db2c9cb6c7..b28ab9b2c05445d9a2d0109fe2e36663a6b1efe8 100644 --- a/crates/notebook/Cargo.toml +++ b/crates/notebook/Cargo.toml @@ -52,7 +52,7 @@ rsqlite-vfs = "0.1.1" nix = { version = "0.31", default-features = false, features = ["fs"] } [target.'cfg(windows)'.dependencies] -windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } +windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem", "Win32_System_Registry"] } [dev-dependencies] libc = "0.2" diff --git a/crates/notebook/src/live.rs b/crates/notebook/src/live.rs index 36b03a9bd8a82874e764bcb6e6bee771cf2591f4..b04de9bc9999cdd860bc0edbcdd8b29ba5583db2 100644 --- a/crates/notebook/src/live.rs +++ b/crates/notebook/src/live.rs @@ -8,8 +8,11 @@ //! from scratch. pub use ::relay::code; +pub mod proxy; mod relay; pub mod share; +mod transport; +pub use transport::Trouble; pub mod wire; pub use wire::{Caret, Guid, Hello, Presence, Spot}; @@ -156,8 +159,8 @@ pub enum Relayed { Joined, /// Answered with this HTTP status, and how long it asked to wait. Refused(u16, Option), - /// Not reached. - Unreachable, + /// Not reached, and why. + Unreachable(Trouble), } /// Presence on the network while it lives; dropping it leaves. @@ -755,6 +758,30 @@ fn discover( Ok(daemon) } +/// `text` with `%XX` escapes decoded, as a URL's name and password. +fn decode(text: &str) -> String { + let bytes = text.as_bytes(); + let mut decoded = Vec::with_capacity(bytes.len()); + let mut at = 0; + while at < bytes.len() { + match text + .get(at + 1..at + 3) + .filter(|_| bytes[at] == b'%') + .and_then(|hex| u8::from_str_radix(hex, 16).ok()) + { + Some(byte) => { + decoded.push(byte); + at += 3; + } + None => { + decoded.push(bytes[at]); + at += 1; + } + } + } + String::from_utf8_lossy(&decoded).into_owned() +} + fn hex(bytes: &[u8]) -> String { bytes.iter().map(|byte| format!("{byte:02x}")).collect() } diff --git a/crates/notebook/src/live/proxy.rs b/crates/notebook/src/live/proxy.rs new file mode 100644 index 0000000000000000000000000000000000000000..6fcb5911e051e27dc62bcd5305885584f132a2e5 --- /dev/null +++ b/crates/notebook/src/live/proxy.rs @@ -0,0 +1,290 @@ +//! The proxy a relay connection, and only a relay connection, goes through: `HTTPS_PROXY` +//! (`HTTP_PROXY` for `ws://`), then `ALL_PROXY`, each in lower case too, short of `NO_PROXY`; +//! else the system's: macOS's network settings, Windows's Internet settings, GNOME's. Only +//! HTTP proxies, reached by `CONNECT`, with a name and password where the URL has them; a +//! proxy auto-configuration script is not read. + +use base64::Engine; +use std::sync::Mutex; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Proxy { + pub host: String, + pub port: u16, + /// The name and password `Proxy-Authorization` sends. + pub credentials: Option<(String, String)>, +} + +impl Proxy { + /// `http://[name:password@]host[:port][/]`, or `host:port`; none for another scheme. + pub fn parse(url: &str) -> Option { + let url = url.trim(); + let rest = match url.split_once("://") { + Some(("http" | "https", rest)) => rest, + Some(_) => return None, + None => url, + }; + let rest = rest.split('/').next()?; + let (credentials, authority) = match rest.rsplit_once('@') { + Some((credentials, authority)) => { + let (name, password) = credentials.split_once(':').unwrap_or((credentials, "")); + ( + Some((crate::live::decode(name), crate::live::decode(password))), + authority, + ) + } + None => (None, rest), + }; + let (host, port) = match authority.rsplit_once(':') { + Some((host, port)) if !port.contains(']') => (host, port.parse().ok()?), + _ => (authority, 80), + }; + let host = host.trim_start_matches('[').trim_end_matches(']'); + (!host.is_empty()).then(|| Self { + host: host.to_owned(), + port, + credentials, + }) + } + + /// The `Proxy-Authorization` header's value, where it has credentials. + pub fn authorization(&self) -> Option { + let (name, password) = self.credentials.as_ref()?; + let token = base64::engine::general_purpose::STANDARD.encode(format!("{name}:{password}")); + Some(format!("Basic {token}")) + } +} + +impl std::fmt::Display for Proxy { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + write!(f, "{}:{}", self.host, self.port) + } +} + +/// A proxy `use_proxy` set in place of what the environment and the system say. +static CHOSEN: Mutex>> = Mutex::new(None); + +/// Uses `proxy`, or no proxy with `Some(None)`, for every relay connection from now on; `None` +/// goes back to the environment's and the system's. +pub fn use_proxy(proxy: Option>) { + *CHOSEN + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = proxy; +} + +/// The proxy a connection to `host` goes through, over TLS where `tls`. +pub fn for_host(host: &str, tls: bool) -> Option { + if let Some(chosen) = CHOSEN.lock().unwrap_or_else(|p| p.into_inner()).clone() { + return chosen; + } + let variable = |name: &str| { + [name.to_owned(), name.to_lowercase()] + .into_iter() + .find_map(|name| std::env::var(name).ok().filter(|value| !value.is_empty())) + }; + let named = [if tls { "HTTPS_PROXY" } else { "HTTP_PROXY" }, "ALL_PROXY"] + .into_iter() + .find_map(variable); + match named { + Some(url) => { + let bypass = variable("NO_PROXY").unwrap_or_default(); + (!bypassed(host, bypass.split(','))).then(|| Proxy::parse(&url))? + } + None => system(host, tls), + } +} + +/// Whether `host` is one of `list`'s: a name, a suffix after a dot, or `*` for every host. +fn bypassed<'a>(host: &str, list: impl Iterator) -> bool { + let host = host.to_ascii_lowercase(); + list.map(|entry| { + entry + .trim() + .trim_start_matches("*.") + .trim_start_matches('.') + }) + .filter(|entry| !entry.is_empty()) + .any(|entry| { + let entry = entry.to_ascii_lowercase(); + entry == "*" || host == entry || host.ends_with(&format!(".{entry}")) + }) +} + +/// What `scutil --proxy` says of the network settings in use. +#[cfg(target_os = "macos")] +fn system(host: &str, tls: bool) -> Option { + let output = std::process::Command::new("/usr/sbin/scutil") + .arg("--proxy") + .output() + .ok()?; + let text = String::from_utf8(output.stdout).ok()?; + let value = |key: &str| { + text.lines().find_map(|line| { + let (name, value) = line.split_once(" : ")?; + (name.trim() == key).then(|| value.trim().to_owned()) + }) + }; + let kind = if tls { "HTTPS" } else { "HTTP" }; + if value(&format!("{kind}Enable")).as_deref() != Some("1") { + return None; + } + let exceptions: Vec = text + .lines() + .skip_while(|line| !line.contains("ExceptionsList")) + .skip(1) + .take_while(|line| !line.contains('}')) + .filter_map(|line| Some(line.split_once(" : ")?.1.trim().to_owned())) + .collect(); + if bypassed(host, exceptions.iter().map(String::as_str)) { + return None; + } + Some(Proxy { + host: value(&format!("{kind}Proxy"))?, + port: value(&format!("{kind}Port"))?.parse().ok()?, + credentials: None, + }) +} + +/// What Windows's Internet settings say, as WinHTTP and Internet Explorer read them for this +/// user: `ProxyServer`, as `host:port` or `https=host:port;http=...`, where `ProxyEnable`. +#[cfg(windows)] +fn system(host: &str, tls: bool) -> Option { + let key = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"; + if registry::dword(key, "ProxyEnable")? == 0 { + return None; + } + let server = registry::string(key, "ProxyServer")?; + let overrides = registry::string(key, "ProxyOverride").unwrap_or_default(); + if bypassed( + host, + overrides.split(';').filter(|entry| *entry != ""), + ) { + return None; + } + let scheme = if tls { "https=" } else { "http=" }; + let server = match server.contains('=') { + true => server + .split(';') + .find_map(|entry| entry.trim().strip_prefix(scheme))?, + false => server.as_str(), + }; + Proxy::parse(server) +} + +#[cfg(windows)] +#[allow(unsafe_code)] +mod registry { + use windows_sys::Win32::System::Registry::{ + HKEY_CURRENT_USER, RRF_RT_REG_DWORD, RRF_RT_REG_SZ, RegGetValueW, + }; + + fn wide(text: &str) -> Vec { + text.encode_utf16().chain([0]).collect() + } + + pub fn dword(key: &str, name: &str) -> Option { + let (key, name) = (wide(key), wide(name)); + let mut value = 0u32; + let mut size = 4u32; + // SAFETY: the key and name are NUL-terminated and outlive the call; the value is four + // bytes, as `size` says. + let result = unsafe { + RegGetValueW( + HKEY_CURRENT_USER, + key.as_ptr(), + name.as_ptr(), + RRF_RT_REG_DWORD, + std::ptr::null_mut(), + (&mut value as *mut u32).cast(), + &mut size, + ) + }; + (result == 0).then_some(value) + } + + pub fn string(key: &str, name: &str) -> Option { + let (key, name) = (wide(key), wide(name)); + let mut buffer = vec![0u16; 2048]; + let mut size = (buffer.len() * 2) as u32; + // SAFETY: as in `dword`, with `size` the buffer's length in bytes. + let result = unsafe { + RegGetValueW( + HKEY_CURRENT_USER, + key.as_ptr(), + name.as_ptr(), + RRF_RT_REG_SZ, + std::ptr::null_mut(), + buffer.as_mut_ptr().cast(), + &mut size, + ) + }; + if result != 0 { + return None; + } + let length = buffer + .iter() + .position(|unit| *unit == 0) + .unwrap_or(buffer.len()); + Some(String::from_utf16_lossy(&buffer[..length])) + } +} + +/// What GNOME's proxy settings say, where they are manual. +#[cfg(not(any(target_os = "macos", windows)))] +fn system(host: &str, tls: bool) -> Option { + let get = |schema: &str, key: &str| { + let output = std::process::Command::new("gsettings") + .args(["get", schema, key]) + .output() + .ok()?; + let text = String::from_utf8(output.stdout).ok()?; + Some(text.trim().trim_matches('\'').to_owned()) + }; + if get("org.gnome.system.proxy", "mode")? != "manual" { + return None; + } + let ignored = get("org.gnome.system.proxy", "ignore-hosts").unwrap_or_default(); + let ignored = ignored.trim_matches(['[', ']']).replace('\'', ""); + if bypassed(host, ignored.split(',')) { + return None; + } + let schema = if tls { + "org.gnome.system.proxy.https" + } else { + "org.gnome.system.proxy.http" + }; + Some(Proxy { + host: get(schema, "host").filter(|host| !host.is_empty())?, + port: get(schema, "port")? + .parse() + .ok() + .filter(|port| *port != 0)?, + credentials: None, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn proxies_read_as_written() { + assert_eq!( + Proxy::parse("http://ada:p%40ss@proxy.example:3128/"), + Some(Proxy { + host: "proxy.example".into(), + port: 3128, + credentials: Some(("ada".into(), "p@ss".into())), + }) + ); + assert_eq!(Proxy::parse("proxy:8080").unwrap().port, 8080); + assert_eq!(Proxy::parse("http://[::1]:8080").unwrap().host, "::1"); + assert!(Proxy::parse("socks5://proxy:1080").is_none()); + let proxy = Proxy::parse("http://ada:secret@proxy:1").unwrap(); + assert_eq!(proxy.authorization().unwrap(), "Basic YWRhOnNlY3JldA=="); + let list = || ["localhost", ".internal", "*.corp.example"].into_iter(); + assert!(bypassed("localhost", list()) && bypassed("files.internal", list())); + assert!(bypassed("a.corp.example", list()) && !bypassed("relay.example", list())); + assert!(bypassed("anything", ["*"].into_iter())); + } +} diff --git a/crates/notebook/src/live/relay.rs b/crates/notebook/src/live/relay.rs index bd6c7c80d1e5a6f3a0f02e7673f0898aa1ade6db..553b41ae388f9ad0c51a8129cf68bf54b37575f4 100644 --- a/crates/notebook/src/live/relay.rs +++ b/crates/notebook/src/live/relay.rs @@ -3,23 +3,21 @@ //! the relay sees only who talks to whom, when, and how much. A stream that breaks makes this //! end join the room again, which ends every stream it had there; peers then meet afresh. -use super::{Event, OPENING, PATIENCE, Pipe, Relayed as Answer, Shared, Side, code_parts}; +use super::{ + Event, OPENING, PATIENCE, Pipe, Relayed as Answer, Shared, Side, code_parts, + transport::{self, Address, Failure, parse}, +}; use ::relay::{Notice, SLOT, Verdict, ws}; -use base64::Engine; -use rustls::{ClientConfig, ClientConnection, RootCertStore, pki_types::ServerName}; use std::{ collections::{HashMap, HashSet}, io::{self, BufReader, Read, Write}, - net::{Shutdown, TcpStream, ToSocketAddrs}, - sync::{Arc, Mutex, OnceLock, atomic::Ordering, mpsc}, + sync::{Arc, Mutex, atomic::Ordering, mpsc}, thread, time::{Duration, Instant}, }; -const CONNECT: Duration = Duration::from_secs(10); -/// How often a quiet connection pings the relay, and how long it waits to hear anything. +/// How often a quiet connection pings the relay. const KEEPALIVE: Duration = Duration::from_secs(30); -const QUIET: Duration = Duration::from_secs(75); /// A connection that lasted this long was no failure, so the next waits only a second. const STEADY: Duration = Duration::from_secs(60); /// The most sent in one message, well under any relay's cap. @@ -40,54 +38,6 @@ pub(super) fn join(shared: &Arc, url: &str, port: u16) -> io::Result<()> Ok(()) } -/// A relay's address: `ws://` or `wss://`, a host, and the path the relay's `/v1/` follows. -#[derive(Debug, PartialEq)] -struct Address { - tls: bool, - /// The host and port as the URL gave them, for the `Host` header. - authority: String, - host: String, - port: u16, - path: String, -} - -fn parse(url: &str) -> io::Result
{ - let bad = || io::Error::new(io::ErrorKind::InvalidInput, format!("Not a relay: {url}")); - let (tls, rest) = match url.split_once("://") { - Some(("wss", rest)) => (true, rest), - Some(("ws", rest)) => (false, rest), - _ => return Err(bad()), - }; - let (authority, path) = rest.split_at(rest.find('/').unwrap_or(rest.len())); - let (host, port) = match authority.rsplit_once(':') { - Some((host, port)) if !port.contains(']') => (host, port.parse().map_err(|_| bad())?), - _ => (authority, if tls { 443 } else { 80 }), - }; - let host = host.trim_start_matches('[').trim_end_matches(']'); - if host.is_empty() { - return Err(bad()); - } - Ok(Address { - tls, - authority: authority.into(), - host: host.into(), - port, - path: path.trim_end_matches('/').into(), - }) -} - -enum Failure { - Network(io::Error), - /// The relay answered, but with this HTTP status and how long to wait. - Refused(u16, Option), -} - -impl From for Failure { - fn from(error: io::Error) -> Self { - Failure::Network(error) - } -} - /// Records how the relay answered, telling `shared`'s events where it changed. fn answered(shared: &Shared, answer: Answer) { let mut state = shared.state.lock().unwrap(); @@ -135,9 +85,9 @@ fn keep(shared: &Arc, address: &Address, port: u16) { } wait = wait.max(retry.unwrap_or_default()); } - Err(Failure::Network(error)) => { + Err(Failure::Trouble(trouble, error)) => { eprintln!("Live: no relay at {}: {error}", address.authority); - answered(shared, Answer::Unreachable); + answered(shared, Answer::Unreachable(trouble)); } } if began.elapsed() >= STEADY { @@ -148,170 +98,23 @@ fn keep(shared: &Arc, address: &Address, port: u16) { } } -/// Opens a WebSocket to `path` at `address`: the connection, and its reading half. +/// Opens the relay's `path` at `address`: the connection, and what reads its messages. fn connect(address: &Address, path: &str, owner: bool) -> Result<(Arc, Reader), Failure> { - let target = (address.host.as_str(), address.port) - .to_socket_addrs()? - .next() - .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "No address for the relay"))?; - let tcp = TcpStream::connect_timeout(&target, CONNECT)?; - tcp.set_nodelay(true)?; - tcp.set_read_timeout(Some(QUIET))?; - tcp.set_write_timeout(Some(QUIET))?; - let (reading, mut writing): (Box, Box) = if address.tls { - let name = ServerName::try_from(address.host.clone()) - .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?; - let mut connection = ClientConnection::new(tls(), name).map_err(io::Error::other)?; - while connection.is_handshaking() { - connection.complete_io(&mut &tcp)?; - } - let connection = Arc::new(Mutex::new(connection)); - ( - Box::new(TlsReader { - tcp: tcp.try_clone()?, - tls: Arc::clone(&connection), - plain: Vec::new(), - at: 0, - }), - Box::new(TlsWriter { - tcp: tcp.try_clone()?, - tls: connection, - }), - ) - } else { - (Box::new(tcp.try_clone()?), Box::new(tcp.try_clone()?)) - }; - let mut key = [0; 16]; - getrandom::fill(&mut key).map_err(|_| io::Error::other("System random source failed"))?; - let key = base64::engine::general_purpose::STANDARD.encode(key); - write!( - writing, - "GET {path} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\ - Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\nUser-Agent: Snowbound/{}\r\n\r\n", - address.authority, - env!("CARGO_PKG_VERSION"), - )?; - let mut reader = BufReader::new(reading); - let head = ws::head(&mut reader)?; - let status = head - .split(' ') - .nth(1) - .and_then(|status| status.parse().ok()) - .unwrap_or(0); - if status != 101 { - let retry = ws::header(&head, "Retry-After") - .and_then(|seconds| seconds.parse().ok()) - .map(Duration::from_secs); - return Err(Failure::Refused(status, retry)); - } - if ws::header(&head, "Sec-WebSocket-Accept") != Some(ws::accept(&key).as_str()) { - return Err(io::Error::new(io::ErrorKind::InvalidData, "Not a relay").into()); - } + let connection = transport::connect(address, path)?; + let reader = ws::Reader::new(BufReader::new(connection.reader), MOST, false); let socket = Arc::new(Socket { - send: Mutex::new(writing), - tcp, + send: Mutex::new(connection.writer), + close: connection.close, owner, links: Mutex::default(), }); - Ok((socket, ws::Reader::new(reader, MOST, false))) -} - -/// The certificate authorities the system trusts, then Mozilla's for a system whose store is -/// missing or stale, as updates trust them. -fn tls() -> Arc { - static CONFIG: OnceLock> = OnceLock::new(); - Arc::clone(CONFIG.get_or_init(|| { - let mut roots = RootCertStore::empty(); - roots.add_parsable_certificates(rustls_native_certs::load_native_certs().certs); - roots.add_parsable_certificates(webpki_root_certs::TLS_SERVER_ROOT_CERTS.iter().cloned()); - let provider = Arc::new(rustls::crypto::ring::default_provider()); - Arc::new( - ClientConfig::builder_with_provider(provider) - .with_safe_default_protocol_versions() - .expect("ring speaks TLS 1.2 and 1.3") - .with_root_certificates(roots) - .with_no_client_auth(), - ) - })) -} - -/// TLS read on one thread while another writes: the socket is read without the lock, and -/// what arrives is decrypted under it. -struct TlsReader { - tcp: TcpStream, - tls: Arc>, - plain: Vec, - at: usize, -} - -impl Read for TlsReader { - fn read(&mut self, buffer: &mut [u8]) -> io::Result { - while self.at == self.plain.len() { - self.plain.clear(); - self.at = 0; - let mut raw = vec![0; 16 << 10]; - let length = self.tcp.read(&mut raw)?; - if length == 0 { - return Ok(0); - } - let mut tls = self.tls.lock().unwrap(); - let mut arrived = &raw[..length]; - let mut closed = false; - while !arrived.is_empty() { - tls.read_tls(&mut arrived)?; - tls.process_new_packets() - .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; - let mut chunk = [0; 4096]; - loop { - match tls.reader().read(&mut chunk) { - Ok(0) => { - closed = true; - break; - } - Ok(length) => self.plain.extend_from_slice(&chunk[..length]), - Err(error) if error.kind() == io::ErrorKind::WouldBlock => break, - Err(error) => return Err(error), - } - } - } - while tls.wants_write() { - tls.write_tls(&mut &self.tcp)?; - } - if closed && self.plain.is_empty() { - return Ok(0); - } - } - let length = buffer.len().min(self.plain.len() - self.at); - buffer[..length].copy_from_slice(&self.plain[self.at..self.at + length]); - self.at += length; - Ok(length) - } -} - -struct TlsWriter { - tcp: TcpStream, - tls: Arc>, -} - -impl Write for TlsWriter { - fn write(&mut self, bytes: &[u8]) -> io::Result { - let mut tls = self.tls.lock().unwrap(); - let length = tls.writer().write(bytes)?; - while tls.wants_write() { - tls.write_tls(&mut &self.tcp)?; - } - Ok(length) - } - - fn flush(&mut self) -> io::Result<()> { - Ok(()) - } + Ok((socket, reader)) } /// One connection to a relay's room. pub(super) struct Socket { send: Mutex>, - tcp: TcpStream, + close: Box, /// Whether this end claimed the room for its code, and so tells the relay who knew it. owner: bool, links: Mutex, @@ -336,7 +139,7 @@ impl Socket { } pub(super) fn hang_up(&self) { - let _ = self.tcp.shutdown(Shutdown::Both); + (self.close)(); } fn forget(&self, slot: u32) { diff --git a/crates/notebook/src/live/share.rs b/crates/notebook/src/live/share.rs index 84d0fe1a5f612efe3a8d23383988ab25e9857bc4..ff7f9a8a8581860d9491332cf6c74c3fc42dcfcf 100644 --- a/crates/notebook/src/live/share.rs +++ b/crates/notebook/src/live/share.rs @@ -114,8 +114,8 @@ pub enum Refusal { TooMany(Option), /// The relay is full. Busy, - /// The relay couldn't be reached, and no one answered on this network. - Unreachable, + /// The relay couldn't be reached, and why, and no one answered on this network. + Unreachable(super::Trouble), /// The relay let this end in, but no one answered. TimedOut, } @@ -152,7 +152,7 @@ pub fn join( } }, ) - .map_err(|_| Refusal::Unreachable)?; + .map_err(|_| Refusal::Unreachable(super::Trouble::Other))?; let start = Instant::now(); loop { if let Ok(welcome) = welcome.try_recv() { @@ -174,8 +174,8 @@ pub fn join( Relayed::Refused(410, _) => return Err(Refusal::Expired), Relayed::Refused(429, wait) => return Err(Refusal::TooMany(wait)), Relayed::Refused(503, _) if settled => return Err(Refusal::Busy), - Relayed::Unreachable if waited > Duration::from_secs(10) => { - return Err(Refusal::Unreachable); + Relayed::Unreachable(trouble) if waited > Duration::from_secs(10) => { + return Err(Refusal::Unreachable(trouble)); } Relayed::Unknown if relay.is_none() && waited > Duration::from_secs(10) => { return Err(Refusal::NoOne); diff --git a/crates/notebook/src/live/transport.rs b/crates/notebook/src/live/transport.rs new file mode 100644 index 0000000000000000000000000000000000000000..5435b9d6f9fb1f5bc86b49e6f72cad8e86ed1505 --- /dev/null +++ b/crates/notebook/src/live/transport.rs @@ -0,0 +1,628 @@ +//! The connection to a relay, wherever the network lets one through: straight to it, or +//! through the proxy `proxy` names by `CONNECT`, over TLS that trusts what the system trusts +//! (so a proxy that inspects HTTPS with its own authority works once the system trusts it); +//! as a WebSocket, or where something on the way refuses WebSockets, as HTTPS requests the +//! relay answers the same messages over (`GET` waits for what is to come, `POST` sends). What +//! fails is told apart, so a person can be told what to try. + +use super::proxy::{self, Proxy}; +use ::relay::ws; +use base64::Engine; +use rustls::{ClientConfig, ClientConnection, RootCertStore, pki_types::ServerName}; +use std::{ + io::{self, BufReader, Read, Write}, + net::{Shutdown, TcpStream, ToSocketAddrs}, + sync::{ + Arc, Condvar, Mutex, OnceLock, + atomic::{AtomicBool, Ordering}, + }, + thread, + time::Duration, +}; + +const CONNECT: Duration = Duration::from_secs(10); +/// How long a connection may hear nothing: the relay hears a ping every 30 s and answers. +pub(super) const QUIET: Duration = Duration::from_secs(75); +/// The most a poll's answer holds. +const MOST: usize = 4 << 20; +/// Set once WebSockets were refused, so later connections go straight to polling. +static POLLING: AtomicBool = AtomicBool::new(false); + +/// Why a relay could not be reached, as a person can act on it. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Trouble { + /// The relay's name, or the proxy's, didn't resolve. + Dns, + /// Nothing answered at the relay's address: a firewall, or the relay is down. + Unreachable, + TimedOut, + /// The proxy itself couldn't be reached. + ProxyUnreachable, + /// The proxy asks for a name and password (407). + ProxyAuthentication, + /// The proxy refused to connect to the relay, with this status. + ProxyRefused(u16), + /// The relay's certificate wasn't one the system trusts: something on the way presents its + /// own. + Certificate, + /// Something on the way refused both WebSockets and plain requests to the relay. + Blocked, + Other, +} + +pub(super) enum Failure { + Trouble(Trouble, io::Error), + /// The relay answered, with this HTTP status and how long to wait. + Refused(u16, Option), +} + +impl Failure { + fn other(error: io::Error) -> Self { + Failure::Trouble(Trouble::Other, error) + } +} + +impl From for Failure { + fn from(error: io::Error) -> Self { + Failure::other(error) + } +} + +/// A relay's address: `ws://` or `wss://`, a host, and the path the relay's `/v1/` follows. +#[derive(Debug, PartialEq)] +pub(super) struct Address { + pub tls: bool, + /// The host and port as the URL gave them, for the `Host` header. + pub authority: String, + pub host: String, + pub port: u16, + pub path: String, +} + +pub(super) fn parse(url: &str) -> io::Result
{ + let bad = || io::Error::new(io::ErrorKind::InvalidInput, format!("Not a relay: {url}")); + let (tls, rest) = match url.split_once("://") { + Some(("wss", rest)) => (true, rest), + Some(("ws", rest)) => (false, rest), + _ => return Err(bad()), + }; + let (authority, path) = rest.split_at(rest.find('/').unwrap_or(rest.len())); + let (host, port) = match authority.rsplit_once(':') { + Some((host, port)) if !port.contains(']') => (host, port.parse().map_err(|_| bad())?), + _ => (authority, if tls { 443 } else { 80 }), + }; + let host = host.trim_start_matches('[').trim_end_matches(']'); + if host.is_empty() { + return Err(bad()); + } + Ok(Address { + tls, + authority: authority.into(), + host: host.into(), + port, + path: path.trim_end_matches('/').into(), + }) +} + +/// A connection to the relay: what it writes the relay's messages into, what it reads them +/// from, and how to hang up. +pub(super) struct Connection { + pub writer: Box, + pub reader: Box, + pub close: Box, +} + +/// Opens the relay's `path` at `address`, as a WebSocket, else as HTTPS requests where +/// something on the way refused the WebSocket. +pub(super) fn connect(address: &Address, path: &str) -> Result { + if !POLLING.load(Ordering::Acquire) { + match websocket(address, path) { + Err(Failure::Refused(status, _)) if !relay_status(status) => { + POLLING.store(true, Ordering::Release); + } + Err(Failure::Trouble(Trouble::Other, _)) => { + POLLING.store(true, Ordering::Release); + } + connected => return connected, + } + } + poll(address, path).map_err(|failure| match failure { + Failure::Refused(status, _) if !relay_status(status) => Failure::Trouble( + Trouble::Blocked, + io::Error::other(format!("Refused with {status}")), + ), + failure => failure, + }) +} + +/// Whether the relay itself answers with `status`, rather than something on the way. +fn relay_status(status: u16) -> bool { + matches!(status, 404 | 410 | 429 | 503) +} + +/// A stream to the relay, through the proxy where there is one, over TLS where its address +/// asks. +struct Stream { + reader: Box, + writer: Box, + tcp: TcpStream, +} + +fn open(address: &Address) -> Result { + let proxy = proxy::for_host(&address.host, address.tls); + let (host, port) = match &proxy { + Some(proxy) => (proxy.host.as_str(), proxy.port), + None => (address.host.as_str(), address.port), + }; + let reached = |trouble| match proxy { + Some(_) => Trouble::ProxyUnreachable, + None => trouble, + }; + let target = (host, port) + .to_socket_addrs() + .map_err(|error| Failure::Trouble(reached(Trouble::Dns), error))? + .next() + .ok_or_else(|| Failure::Trouble(reached(Trouble::Dns), io::ErrorKind::NotFound.into()))?; + let tcp = TcpStream::connect_timeout(&target, CONNECT).map_err(|error| { + let trouble = match error.kind() { + io::ErrorKind::TimedOut => Trouble::TimedOut, + _ => Trouble::Unreachable, + }; + Failure::Trouble(reached(trouble), error) + })?; + tcp.set_nodelay(true)?; + tcp.set_read_timeout(Some(QUIET))?; + tcp.set_write_timeout(Some(QUIET))?; + if let Some(proxy) = &proxy { + tunnel(&tcp, proxy, address)?; + } + if !address.tls { + return Ok(Stream { + reader: Box::new(tcp.try_clone()?), + writer: Box::new(tcp.try_clone()?), + tcp, + }); + } + let name = ServerName::try_from(address.host.clone()) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?; + let mut connection = ClientConnection::new(tls(), name).map_err(io::Error::other)?; + while connection.is_handshaking() { + connection.complete_io(&mut &tcp).map_err(|error| { + let rejected = error + .get_ref() + .and_then(|inner| inner.downcast_ref::()) + .is_some_and(|tls| matches!(tls, rustls::Error::InvalidCertificate(_))); + match rejected { + true => Failure::Trouble(Trouble::Certificate, error), + false => Failure::other(error), + } + })?; + } + let connection = Arc::new(Mutex::new(connection)); + Ok(Stream { + reader: Box::new(TlsReader { + tcp: tcp.try_clone()?, + tls: Arc::clone(&connection), + plain: Vec::new(), + at: 0, + }), + writer: Box::new(TlsWriter { + tcp: tcp.try_clone()?, + tls: connection, + }), + tcp, + }) +} + +/// Asks `proxy` on `tcp` to connect through to the relay. +fn tunnel(tcp: &TcpStream, proxy: &Proxy, address: &Address) -> Result<(), Failure> { + let target = format!("{}:{}", address.host, address.port); + let mut request = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n"); + if let Some(authorization) = proxy.authorization() { + request += &format!("Proxy-Authorization: {authorization}\r\n"); + } + request += "\r\n"; + (&mut &*tcp) + .write_all(request.as_bytes()) + .map_err(|error| Failure::Trouble(Trouble::ProxyUnreachable, error))?; + let head = + ws::head(&mut &*tcp).map_err(|error| Failure::Trouble(Trouble::ProxyUnreachable, error))?; + match status(&head) { + 200 => Ok(()), + 407 => Err(Failure::Trouble( + Trouble::ProxyAuthentication, + io::Error::new(io::ErrorKind::PermissionDenied, "The proxy asks for a name"), + )), + status => Err(Failure::Trouble( + Trouble::ProxyRefused(status), + io::Error::other(format!("The proxy answered {status}")), + )), + } +} + +fn status(head: &str) -> u16 { + head.split(' ') + .nth(1) + .and_then(|status| status.parse().ok()) + .unwrap_or(0) +} + +fn retry(head: &str) -> Option { + ws::header(head, "Retry-After") + .and_then(|seconds| seconds.parse().ok()) + .map(Duration::from_secs) +} + +/// Opens a WebSocket to `path` at `address`. +fn websocket(address: &Address, path: &str) -> Result { + let Stream { + reader, + mut writer, + tcp, + } = open(address)?; + let mut key = [0; 16]; + getrandom::fill(&mut key).map_err(|_| io::Error::other("System random source failed"))?; + let key = base64::engine::general_purpose::STANDARD.encode(key); + write!( + writer, + "GET {path} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\ + Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\nUser-Agent: Snowbound/{}\r\n\r\n", + address.authority, + env!("CARGO_PKG_VERSION"), + )?; + let mut reader = BufReader::new(reader); + let head = ws::head(&mut reader)?; + if status(&head) != 101 { + return Err(Failure::Refused(status(&head), retry(&head))); + } + if ws::header(&head, "Sec-WebSocket-Accept") != Some(ws::accept(&key).as_str()) { + return Err(io::Error::new(io::ErrorKind::InvalidData, "Not a relay").into()); + } + Ok(Connection { + writer, + reader: Box::new(reader), + close: Box::new(move || { + let _ = tcp.shutdown(Shutdown::Both); + }), + }) +} + +/// An HTTP/1.1 connection that answers one request after another. +struct Http { + reader: BufReader>, + writer: Box, + tcp: TcpStream, +} + +impl Http { + fn open(address: &Address) -> Result { + let Stream { + reader, + writer, + tcp, + } = open(address)?; + Ok(Self { + reader: BufReader::new(reader), + writer, + tcp, + }) + } + + /// Sends `method` on `path` with `body`: the status, the head and the body answered. + fn ask( + &mut self, + address: &Address, + method: &str, + path: &str, + body: &[u8], + ) -> io::Result<(u16, String, Vec)> { + write!( + self.writer, + "{method} {path} HTTP/1.1\r\nHost: {}\r\nUser-Agent: Snowbound/{}\r\n\ + Content-Length: {}\r\nContent-Type: application/octet-stream\r\n\r\n", + address.authority, + env!("CARGO_PKG_VERSION"), + body.len() + )?; + self.writer.write_all(body)?; + let head = ws::head(&mut self.reader)?; + let length: usize = ws::header(&head, "Content-Length") + .and_then(|length| length.parse().ok()) + .unwrap_or(0); + if length > MOST { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "An answer too large", + )); + } + let mut answer = vec![0; length]; + self.reader.read_exact(&mut answer)?; + Ok((status(&head), head, answer)) + } +} + +/// What a polled session shares between its reading, its sending and hanging up. +struct Session { + /// Messages written and not yet sent. + pending: Mutex>, + ready: Condvar, + closed: AtomicBool, + /// The connections open now, to shut on hanging up. + open: Mutex>, +} + +impl Session { + fn close(&self) { + self.closed.store(true, Ordering::Release); + self.ready.notify_all(); + for tcp in self.open.lock().unwrap().drain(..) { + let _ = tcp.shutdown(Shutdown::Both); + } + } +} + +/// Joins `path` as HTTPS requests: the relay's answer names a session, whose messages a +/// `GET` waits for and a `POST` sends, each a run of WebSocket frames as on a WebSocket. +fn poll(address: &Address, path: &str) -> Result { + let mut http = Http::open(address)?; + let joined = match path.contains('?') { + true => format!("{path}&poll=1"), + false => format!("{path}?poll=1"), + }; + let (status, head, body) = http.ask(address, "GET", &joined, &[])?; + if status != 200 { + return Err(Failure::Refused(status, retry(&head))); + } + let session_id = String::from_utf8_lossy(&body) + .strip_prefix("session ") + .map(|id| id.trim().to_owned()) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Not a relay"))?; + let at = format!("{}/v1/poll/{session_id}", address.path); + let session = Arc::new(Session { + pending: Mutex::default(), + ready: Condvar::new(), + closed: AtomicBool::new(false), + open: Mutex::new(vec![http.tcp.try_clone()?]), + }); + let address = Arc::new(Address { + tls: address.tls, + authority: address.authority.clone(), + host: address.host.clone(), + port: address.port, + path: address.path.clone(), + }); + // Sends what is written, a batch a request, on a connection of its own. + let (sending, to, posting) = (Arc::clone(&session), at.clone(), Arc::clone(&address)); + thread::Builder::new() + .name("live relay post".into()) + .spawn(move || post(&sending, &posting, &to))?; + let closing = Arc::clone(&session); + Ok(Connection { + writer: Box::new(PollWriter(Arc::clone(&session))), + reader: Box::new(PollReader { + session, + address, + at, + http: Some(http), + arrived: Vec::new(), + read: 0, + }), + close: Box::new(move || closing.close()), + }) +} + +fn post(session: &Session, address: &Address, at: &str) { + let mut http: Option = None; + loop { + let batch = { + let mut pending = session.pending.lock().unwrap(); + while pending.is_empty() && !session.closed.load(Ordering::Acquire) { + pending = session.ready.wait(pending).unwrap(); + } + if session.closed.load(Ordering::Acquire) { + return; + } + std::mem::take(&mut *pending) + }; + // A connection the relay or the proxy closed meanwhile is opened again, once. + let sent = (0..2).any(|_| { + if http.is_none() { + http = Http::open(address).ok(); + if let (Some(opened), Ok(mut open)) = (&http, session.open.lock()) { + open.extend(opened.tcp.try_clone()); + } + } + let answered = http + .as_mut() + .map(|connection| connection.ask(address, "POST", at, &batch)); + match answered { + Some(Ok((200 | 204, ..))) => true, + Some(Ok((410, ..))) => { + session.close(); + true + } + _ => { + http = None; + false + } + } + }); + if !sent { + session.close(); + return; + } + } +} + +struct PollWriter(Arc); + +impl Write for PollWriter { + fn write(&mut self, bytes: &[u8]) -> io::Result { + if self.0.closed.load(Ordering::Acquire) { + return Err(io::ErrorKind::BrokenPipe.into()); + } + self.0.pending.lock().unwrap().extend_from_slice(bytes); + self.0.ready.notify_all(); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +/// Reads what each `GET` brings, waiting for the next when it is all read. +struct PollReader { + session: Arc, + address: Arc
, + at: String, + http: Option, + arrived: Vec, + read: usize, +} + +impl Read for PollReader { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + while self.read == self.arrived.len() { + if self.session.closed.load(Ordering::Acquire) { + return Ok(0); + } + if self.http.is_none() { + let opened = Http::open(&self.address).map_err(|failure| match failure { + Failure::Trouble(_, error) => error, + Failure::Refused(status, _) => io::Error::other(format!("{status}")), + })?; + self.session + .open + .lock() + .unwrap() + .extend(opened.tcp.try_clone()); + self.http = Some(opened); + } + let answered = + self.http + .as_mut() + .expect("opened above") + .ask(&self.address, "GET", &self.at, &[]); + match answered { + Ok((200, _, body)) => { + self.arrived = body; + self.read = 0; + } + Ok((410, ..)) => { + self.session.close(); + return Ok(0); + } + Ok((status, ..)) => { + return Err(io::Error::other(format!("The relay answered {status}"))); + } + Err(error) => { + // Opened again once; a second failure ends the session. + if self.http.take().is_none() { + return Err(error); + } + self.http = Http::open(&self.address).ok(); + if self.http.is_none() { + return Err(error); + } + } + } + } + let length = buffer.len().min(self.arrived.len() - self.read); + buffer[..length].copy_from_slice(&self.arrived[self.read..self.read + length]); + self.read += length; + Ok(length) + } +} + +/// The certificate authorities the system trusts, then Mozilla's for a system whose store is +/// missing or stale, as updates trust them. +fn tls() -> Arc { + static CONFIG: OnceLock> = OnceLock::new(); + Arc::clone(CONFIG.get_or_init(|| { + let mut roots = RootCertStore::empty(); + roots.add_parsable_certificates(rustls_native_certs::load_native_certs().certs); + roots.add_parsable_certificates(webpki_root_certs::TLS_SERVER_ROOT_CERTS.iter().cloned()); + let provider = Arc::new(rustls::crypto::ring::default_provider()); + Arc::new( + ClientConfig::builder_with_provider(provider) + .with_safe_default_protocol_versions() + .expect("ring speaks TLS 1.2 and 1.3") + .with_root_certificates(roots) + .with_no_client_auth(), + ) + })) +} + +/// TLS read on one thread while another writes: the socket is read without the lock, and +/// what arrives is decrypted under it. +struct TlsReader { + tcp: TcpStream, + tls: Arc>, + plain: Vec, + at: usize, +} + +impl Read for TlsReader { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + while self.at == self.plain.len() { + self.plain.clear(); + self.at = 0; + let mut raw = vec![0; 16 << 10]; + let length = self.tcp.read(&mut raw)?; + if length == 0 { + return Ok(0); + } + let mut tls = self.tls.lock().unwrap(); + let mut arrived = &raw[..length]; + let mut closed = false; + while !arrived.is_empty() { + tls.read_tls(&mut arrived)?; + tls.process_new_packets() + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + let mut chunk = [0; 4096]; + loop { + match tls.reader().read(&mut chunk) { + Ok(0) => { + closed = true; + break; + } + Ok(length) => self.plain.extend_from_slice(&chunk[..length]), + Err(error) if error.kind() == io::ErrorKind::WouldBlock => break, + Err(error) => return Err(error), + } + } + } + while tls.wants_write() { + tls.write_tls(&mut &self.tcp)?; + } + if closed && self.plain.is_empty() { + return Ok(0); + } + } + let length = buffer.len().min(self.plain.len() - self.at); + buffer[..length].copy_from_slice(&self.plain[self.at..self.at + length]); + self.at += length; + Ok(length) + } +} + +struct TlsWriter { + tcp: TcpStream, + tls: Arc>, +} + +impl Write for TlsWriter { + fn write(&mut self, bytes: &[u8]) -> io::Result { + let mut tls = self.tls.lock().unwrap(); + let length = tls.writer().write(bytes)?; + while tls.wants_write() { + tls.write_tls(&mut &self.tcp)?; + } + Ok(length) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} diff --git a/crates/notebook/tests/live_networks.rs b/crates/notebook/tests/live_networks.rs new file mode 100644 index 0000000000000000000000000000000000000000..727a163ef2bb85aa7882f4ab76001cb05f3054ed --- /dev/null +++ b/crates/notebook/tests/live_networks.rs @@ -0,0 +1,168 @@ +//! Live Share on hostile networks: through a proxy that asks for a password, through one that +//! refuses WebSockets (the relay is then reached by plain requests), and the troubles named +//! when the relay can't be reached. One test, as the proxy chosen holds for the process. +#![cfg(feature = "live")] + +use notebook::live::{ + Trouble, + proxy::{self, Proxy}, + share::{self, Refusal, Sharing}, +}; +use std::{ + io::{self, Read, Write}, + net::{SocketAddr, TcpListener, TcpStream}, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + thread, +}; + +#[path = "support/live.rs"] +mod live; +use live::*; + +/// What a proxy on this computer did. +#[derive(Default)] +struct Seen { + tunnels: AtomicUsize, + refused: AtomicUsize, +} + +/// An HTTP proxy that tunnels `CONNECT`s, asking for `credentials` where given and, with +/// `block_websockets`, refusing a WebSocket's upgrade inside the tunnel, as a proxy that +/// inspects HTTPS does. +fn proxy(credentials: Option<&'static str>, block_websockets: bool) -> (SocketAddr, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let seen = Arc::new(Seen::default()); + let counting = Arc::clone(&seen); + thread::spawn(move || { + for client in listener.incoming().flatten() { + let seen = Arc::clone(&counting); + thread::spawn(move || { + let _ = tunnel(client, credentials, block_websockets, &seen); + }); + } + }); + (address, seen) +} + +fn tunnel( + mut client: TcpStream, + credentials: Option<&str>, + block_websockets: bool, + seen: &Seen, +) -> io::Result<()> { + let head = relay::ws::head(&mut client)?; + let target = head.split(' ').nth(1).unwrap_or_default().to_owned(); + let authorized = credentials + .is_none_or(|expected| relay::ws::header(&head, "Proxy-Authorization") == Some(expected)); + if !head.starts_with("CONNECT ") || !authorized { + client.write_all( + b"HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic\r\n\ + Content-Length: 0\r\n\r\n", + )?; + return Ok(()); + } + let mut upstream = TcpStream::connect(&target)?; + client.write_all(b"HTTP/1.1 200 Connection established\r\n\r\n")?; + seen.tunnels.fetch_add(1, Ordering::Relaxed); + if block_websockets { + let request = relay::ws::head(&mut client)?; + if relay::ws::header(&request, "Upgrade").is_some() { + seen.refused.fetch_add(1, Ordering::Relaxed); + client.write_all(b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\n\r\n")?; + return Ok(()); + } + upstream.write_all(request.as_bytes())?; + } + let (mut from, mut to) = (client.try_clone()?, upstream.try_clone()?); + thread::spawn(move || { + let _ = io::copy(&mut from, &mut to); + let _ = to.shutdown(std::net::Shutdown::Both); + }); + let mut buffer = [0; 16 << 10]; + loop { + let length = upstream.read(&mut buffer)?; + if length == 0 { + return Ok(()); + } + client.write_all(&buffer[..length])?; + } +} + +fn through(address: SocketAddr, credentials: Option<(&str, &str)>) { + proxy::use_proxy(Some(Some(Proxy { + host: address.ip().to_string(), + port: address.port(), + credentials: credentials.map(|(name, password)| (name.into(), password.into())), + }))); +} + +/// Shares a notebook and has a guest join it and publish an edit, all through `url`. +fn share_and_edit(directory: &std::path::Path, url: &str) { + let folder = notebook(directory); + let host = host( + &folder, + &directory.join("host"), + &Sharing::new("").unwrap(), + url, + ); + let (guest, notebook) = guest("Grace", &code(&host), url, &directory.join("grace")); + let section = open(¬ebook, &guest, "Garden.one", None); + let file = folder.join("Garden.one"); + let id = replace(§ion, &std::fs::read(&file).unwrap(), 0..8, "Through"); + published(§ion, id); + assert_eq!( + server::text(&std::fs::read(&file).unwrap()).2, + "Through text" + ); +} + +fn refusal(code: &str, url: &str) -> Refusal { + share::join(hello("Mallory"), code, "", None, Some(url)).unwrap_err() +} + +#[test] +fn hostile_networks() { + let directory = tempfile::tempdir().unwrap(); + let url = relay(Default::default()); + let code = notebook::live::code::format(412, "4MZ9XR").unwrap(); + + // A proxy that asks for a password: refused without, through with it. + let (address, seen) = proxy(Some("Basic YWRhOnNlY3JldA=="), false); + through(address, None); + assert_eq!( + refusal(&code, &url), + Refusal::Unreachable(Trouble::ProxyAuthentication) + ); + through(address, Some(("ada", "secret"))); + share_and_edit(&directory.path().join("connect"), &url); + assert!( + seen.tunnels.load(Ordering::Relaxed) >= 3, + "host, guest and joiner tunnel" + ); + + // A proxy that refuses WebSockets: the relay is reached by requests instead. + let (address, seen) = proxy(None, true); + through(address, None); + share_and_edit(&directory.path().join("blocked"), &url); + assert!(seen.refused.load(Ordering::Relaxed) >= 1); + + // A proxy that isn't there, and a relay whose name doesn't resolve. + let closed = TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap(); + through(closed, None); + assert_eq!( + refusal(&code, &url), + Refusal::Unreachable(Trouble::ProxyUnreachable) + ); + proxy::use_proxy(Some(None)); + assert_eq!( + refusal(&code, "wss://relay.invalid"), + Refusal::Unreachable(Trouble::Dns) + ); +} diff --git a/crates/notebook/tests/live_share.rs b/crates/notebook/tests/live_share.rs index 7903784fbed1f62a5c482292e1bf8c1c57aa261b..929ad88fd02d4d1f72118fb54635356d4d9f8ab4 100644 --- a/crates/notebook/tests/live_share.rs +++ b/crates/notebook/tests/live_share.rs @@ -4,159 +4,15 @@ #![cfg(feature = "live")] use notebook::{ - EditStatus, Replica, - live::{ - Hello, - share::{self, Guest, Host, Refusal, Sharing}, - }, - session::{Notebook, Section, SyncState}, + live::share::{self, Refusal, Sharing}, + session::{Notebook, SyncState}, }; -use onestore::{ - Arena, ExGuid, - op::{Edit, Op, PageOp}, - protected::{Key, rekey}, -}; -use std::{ - net::TcpListener, - path::Path, - sync::Arc, - thread, - time::{Duration, Instant}, -}; - -#[path = "support/server.rs"] -mod server; - -const PASSWORD: &str = "fixture password"; - -/// A relay on this computer with `config`'s limits: its URL. -fn relay(config: relay::server::Config) -> String { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); - let url = format!("ws://{}", listener.local_addr().unwrap()); - thread::spawn(move || relay::server::serve(listener, config)); - url -} - -/// Another secret than `secret`, as a guess makes one. -fn mistaken(secret: &str) -> String { - let first = if secret.starts_with('A') { 'B' } else { 'A' }; - format!("{first}{}", &secret[1..]) -} - -fn hello(name: &str) -> Hello { - Hello::new(name.into(), None).unwrap() -} - -fn until(what: &str, done: impl Fn() -> bool) { - let deadline = Instant::now() + Duration::from_secs(30); - while !done() { - assert!(Instant::now() < deadline, "{what}"); - thread::sleep(Duration::from_millis(20)); - } -} - -/// A notebook folder holding `Garden.one`, a page reading "Original text", and a protected -/// `Sealed.one` reading "Sealed text". -fn notebook(root: &Path) -> std::path::PathBuf { - let folder = root.join("Garden"); - std::fs::create_dir_all(&folder).unwrap(); - std::fs::write( - folder.join("Garden.one"), - onestore::create_section("Garden.one", "Original text", "Fixture").unwrap(), - ) - .unwrap(); - let plain = onestore::create_section("Sealed.one", "Sealed text", "Fixture").unwrap(); - let key = Key::new(PASSWORD).unwrap(); - std::fs::write( - folder.join("Sealed.one"), - rekey(&plain, None, Some(&key)).unwrap(), - ) - .unwrap(); - folder -} - -fn host(folder: &Path, cache: &Path, sharing: &Sharing, url: &str) -> Host { - let storage = Notebook::open(folder, cache).unwrap().into_storage(); - Host::start( - storage, - hello("Ada"), - sharing.clone(), - "Garden", - None, - Some(url), - || {}, - ) - .unwrap() -} - -/// The host's code once the relay has numbered it. -fn code(host: &Host) -> String { - until("the code was never numbered", || { - host.code().is_some_and(|code| share::code(&code).is_some()) - }); - host.code().unwrap() -} - -/// `name` joins with `code` and opens the notebook in `cache`. -fn guest(name: &str, code: &str, url: &str, cache: &Path) -> (Arc, Notebook) { - let welcome = share::join(hello(name), code, "", None, Some(url)).unwrap(); - assert_eq!( - (welcome.notebook.as_str(), welcome.host.as_str()), - ("Garden", "Ada") - ); - let guest = Guest::start( - hello(name), - welcome.share, - welcome.secret, - None, - Some(url), - || {}, - ) - .unwrap(); - until("the host was never met", || guest.host().is_some()); - let notebook = Notebook::open_hosted(Arc::clone(&guest), cache).unwrap(); - (guest, notebook) -} - -fn open(notebook: &Notebook, guest: &Arc, path: &str, key: Option<&Key>) -> Section { - let replica = notebook.replica_path(path).unwrap(); - std::fs::create_dir_all(replica.parent().unwrap()).unwrap(); - let replica = Replica::open_or_create(&replica, key, || notebook.read_section(path)).unwrap(); - Section::resume_hosted(path.into(), replica, Arc::clone(guest), || {}).unwrap() -} - -fn replace(section: &Section, image: &[u8], range: std::ops::Range, with: &str) -> u64 { - let (space, text, _) = server::text(image); - replaced(section, space, text, range, with) -} - -fn replaced( - section: &Section, - space: ExGuid, - text: ExGuid, - range: std::ops::Range, - with: &str, -) -> u64 { - let op = PageOp::Text { - text, - range, - with: with.into(), - }; - let edit = Edit { - at: 134_000_000_000_000_000, - ops: vec![Op::Page { space, op }], - }; - section.replica().apply("Guest", edit).unwrap() -} +use onestore::Arena; +use std::sync::Arc; -fn published(section: &Section, id: u64) { - until("the edit was never published", || { - matches!( - section.status(id).unwrap(), - Some(EditStatus::Published { .. }) - ) - }); -} +#[path = "support/live.rs"] +mod live; +use live::*; /// A guest opens a section through the host, its edit lands in the host's file, and the /// host's own edit reaches the guest. diff --git a/crates/notebook/tests/support/live.rs b/crates/notebook/tests/support/live.rs new file mode 100644 index 0000000000000000000000000000000000000000..911b42f72d5abeb99cb1b7f9ed7c75b40c093645 --- /dev/null +++ b/crates/notebook/tests/support/live.rs @@ -0,0 +1,157 @@ +//! What the Live Share tests share: a relay, a host and guests on a notebook of two sections. +#![allow(dead_code)] + +use notebook::{ + EditStatus, Replica, + live::{ + Hello, + share::{self, Guest, Host, Sharing}, + }, + session::{Notebook, Section}, +}; +use onestore::{ + ExGuid, + op::{Edit, Op, PageOp}, + protected::{Key, rekey}, +}; +use std::{ + net::TcpListener, + path::Path, + sync::Arc, + thread, + time::{Duration, Instant}, +}; + +#[path = "server.rs"] +pub mod server; + +pub const PASSWORD: &str = "fixture password"; + +/// A relay on this computer with `config`'s limits: its URL. +pub fn relay(config: relay::server::Config) -> String { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + thread::spawn(move || relay::server::serve(listener, config)); + url +} + +/// Another secret than `secret`, as a guess makes one. +pub fn mistaken(secret: &str) -> String { + let first = if secret.starts_with('A') { 'B' } else { 'A' }; + format!("{first}{}", &secret[1..]) +} + +pub fn hello(name: &str) -> Hello { + Hello::new(name.into(), None).unwrap() +} + +pub fn until(what: &str, done: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(30); + while !done() { + assert!(Instant::now() < deadline, "{what}"); + thread::sleep(Duration::from_millis(20)); + } +} + +/// A notebook folder holding `Garden.one`, a page reading "Original text", and a protected +/// `Sealed.one` reading "Sealed text". +pub fn notebook(root: &Path) -> std::path::PathBuf { + let folder = root.join("Garden"); + std::fs::create_dir_all(&folder).unwrap(); + std::fs::write( + folder.join("Garden.one"), + onestore::create_section("Garden.one", "Original text", "Fixture").unwrap(), + ) + .unwrap(); + let plain = onestore::create_section("Sealed.one", "Sealed text", "Fixture").unwrap(); + let key = Key::new(PASSWORD).unwrap(); + std::fs::write( + folder.join("Sealed.one"), + rekey(&plain, None, Some(&key)).unwrap(), + ) + .unwrap(); + folder +} + +pub fn host(folder: &Path, cache: &Path, sharing: &Sharing, url: &str) -> Host { + let storage = Notebook::open(folder, cache).unwrap().into_storage(); + Host::start( + storage, + hello("Ada"), + sharing.clone(), + "Garden", + None, + Some(url), + || {}, + ) + .unwrap() +} + +/// The host's code once the relay has numbered it. +pub fn code(host: &Host) -> String { + until("the code was never numbered", || { + host.code().is_some_and(|code| share::code(&code).is_some()) + }); + host.code().unwrap() +} + +/// `name` joins with `code` and opens the notebook in `cache`. +pub fn guest(name: &str, code: &str, url: &str, cache: &Path) -> (Arc, Notebook) { + let welcome = share::join(hello(name), code, "", None, Some(url)).unwrap(); + assert_eq!( + (welcome.notebook.as_str(), welcome.host.as_str()), + ("Garden", "Ada") + ); + let guest = Guest::start( + hello(name), + welcome.share, + welcome.secret, + None, + Some(url), + || {}, + ) + .unwrap(); + until("the host was never met", || guest.host().is_some()); + let notebook = Notebook::open_hosted(Arc::clone(&guest), cache).unwrap(); + (guest, notebook) +} + +pub fn open(notebook: &Notebook, guest: &Arc, path: &str, key: Option<&Key>) -> Section { + let replica = notebook.replica_path(path).unwrap(); + std::fs::create_dir_all(replica.parent().unwrap()).unwrap(); + let replica = Replica::open_or_create(&replica, key, || notebook.read_section(path)).unwrap(); + Section::resume_hosted(path.into(), replica, Arc::clone(guest), || {}).unwrap() +} + +pub fn replace(section: &Section, image: &[u8], range: std::ops::Range, with: &str) -> u64 { + let (space, text, _) = server::text(image); + replaced(section, space, text, range, with) +} + +pub fn replaced( + section: &Section, + space: ExGuid, + text: ExGuid, + range: std::ops::Range, + with: &str, +) -> u64 { + let op = PageOp::Text { + text, + range, + with: with.into(), + }; + let edit = Edit { + at: 134_000_000_000_000_000, + ops: vec![Op::Page { space, op }], + }; + section.replica().apply("Guest", edit).unwrap() +} + +pub fn published(section: &Section, id: u64) { + until("the edit was never published", || { + matches!( + section.status(id).unwrap(), + Some(EditStatus::Published { .. }) + ) + }); +} diff --git a/crates/relay/README.md b/crates/relay/README.md index e708f15f835a21d3cad1d4eca0f4606d5fc08721..5237d02a6ade9aa33c0e8a58629b0b71dd9ac37e 100644 --- a/crates/relay/README.md +++ b/crates/relay/README.md @@ -68,6 +68,15 @@ Share names another relay. could otherwise claim any address. Listening on a public address without TLS works but lets anyone on the path see room tags and nameplates. +## Where WebSockets don't get through + +Some networks' proxies refuse WebSockets. A peer then joins with `?poll=1` on the same path, +and the relay answers `session `; `GET /v1/poll/` then waits up to 25 seconds for what +is to go to it, and `POST /v1/poll/` brings what it sends, each body a run of WebSocket +frames as the socket would carry them, so the room sees no difference. A session that asks +nothing for a minute leaves. Through Caddy or nginx this needs nothing more than the +WebSocket's own proxying; keep nginx's `proxy_read_timeout` above 25 seconds. + ## The site: snowbound.paperclover.net `snowbound-site` serves the hosted web build's folder, and for a path that is a Live Share diff --git a/crates/relay/src/server.rs b/crates/relay/src/server.rs index c52054b3cb7119039cf69fe89e0dbb0ec05ce5ba..afb95e7527796d75a7caab439a0a08a2c714050e 100644 --- a/crates/relay/src/server.rs +++ b/crates/relay/src/server.rs @@ -65,6 +65,11 @@ impl Default for Config { } const HANDSHAKE: Duration = Duration::from_secs(10); +/// How long a polled session's `GET` waits for something to bring, how long one of its +/// connections may idle between requests, and how long a session may ask nothing. +const WAIT: Duration = Duration::from_secs(25); +const KEPT: Duration = Duration::from_secs(60); +const IDLE_POLL: Duration = Duration::from_secs(60); const WRITE: Duration = Duration::from_secs(30); const MINUTE: Duration = Duration::from_secs(60); const HOUR: Duration = Duration::from_secs(3600); @@ -122,6 +127,17 @@ struct Relay { struct State { rooms: HashMap, addresses: HashMap, + /// Peers that reach the relay by requests rather than a WebSocket, by session. + polls: HashMap, +} + +/// A peer in a room by requests: where its messages wait for its next `GET`. +struct Poll { + tag: String, + slot: u32, + outbox: Arc, + /// When it last asked anything; one quiet past `IDLE_POLL` has gone. + last: Instant, } struct Room { @@ -169,6 +185,8 @@ enum Refusal { } impl Relay { + /// Answers a connection's requests one after another, as a polled session sends them, + /// until one takes the connection over as a WebSocket or it ends. fn connection(&self, stream: TcpStream) { let _ = stream.set_nodelay(true); let _ = stream.set_read_timeout(Some(HANDSHAKE)); @@ -177,16 +195,27 @@ impl Relay { return; }; let mut reader = BufReader::new(reading); - let Ok(head) = ws::head(&mut reader) else { - return; - }; - let address = self.address(&stream, &head); + while let Ok(head) = ws::head(&mut reader) { + if !self.request(&stream, &mut reader, &head) { + return; + } + // Between a session's requests, a connection may idle as long as one waits. + let _ = stream.set_read_timeout(Some(KEPT)); + } + } + + /// Answers the request `head`: whether the connection serves another. + fn request(&self, stream: &TcpStream, reader: &mut BufReader, head: &str) -> bool { + let address = self.address(stream, head); let target = head.split(' ').nth(1).unwrap_or_default(); let (path, query) = target.split_once('?').unwrap_or((target, "")); + if let Some(id) = path.strip_prefix("/v1/poll/") { + return self.polled(stream, reader, head, id); + } let ask = match path { "/health" => { - respond(&stream, "200 OK", "application/json", "", &self.health()); - return; + respond(stream, "200 OK", "application/json", "", &self.health()); + return false; } "/v1/claim" => Ask::Claim( query @@ -197,53 +226,59 @@ impl Relay { _ => match path.strip_prefix("/v1/room/").filter(|tag| valid(tag)) { Some(tag) => Ask::Room(tag.into()), None => { - respond(&stream, "404 Not Found", "text/plain", "", "No such page\n"); - return; + respond(stream, "404 Not Found", "text/plain", "", "No such page\n"); + return false; } }, }; - let upgrade = ws::header(&head, "Upgrade") + if query.split('&').any(|pair| pair == "poll=1") { + let outbox = Arc::new(Outbox::mailbox(self.config.queue)); + return match self.join(address, ask, &outbox, Instant::now()) { + Ok((tag, slot)) => { + let mut id = [0; 16]; + if getrandom::fill(&mut id).is_err() { + return false; + } + let id: String = id.iter().map(|byte| format!("{byte:02x}")).collect(); + let poll = Poll { + tag, + slot, + outbox, + last: Instant::now(), + }; + self.state.lock().unwrap().polls.insert(id.clone(), poll); + reply(stream, "200 OK", format!("session {id}").as_bytes()) + } + Err(refusal) => { + refuse(stream, refusal); + false + } + }; + } + let upgrade = ws::header(head, "Upgrade") .is_some_and(|value| value.eq_ignore_ascii_case("websocket")); let (true, Some(key)) = ( upgrade && head.starts_with("GET "), - ws::header(&head, "Sec-WebSocket-Key"), + ws::header(head, "Sec-WebSocket-Key"), ) else { respond( - &stream, + stream, "400 Bad Request", "text/plain", "", - "A WebSocket only\n", + "A WebSocket, or ?poll=1\n", ); - return; + return false; }; let Ok(writing) = stream.try_clone() else { - return; + return false; }; let outbox = Arc::new(Outbox::new(writing, self.config.queue)); let (tag, slot) = match self.join(address, ask, &outbox, Instant::now()) { Ok(joined) => joined, Err(refusal) => { - let (status, headers, body) = match refusal { - Refusal::NotFound => ("404 Not Found", String::new(), "No such code\n"), - Refusal::Gone => ("410 Gone", String::new(), "The code has expired\n"), - Refusal::Wait(wait) => ( - "429 Too Many Requests", - // Rounded up, so that a client waiting so long finds it over. - format!( - "Retry-After: {}\r\n", - wait.as_secs() + u64::from(wait.subsec_nanos() > 0) - ), - "Too many tries\n", - ), - Refusal::Full => ( - "503 Service Unavailable", - "Retry-After: 30\r\n".into(), - "The relay is full\n", - ), - }; - respond(&stream, status, "text/plain", &headers, body); - return; + refuse(stream, refusal); + return false; } }; let switching = format!( @@ -252,7 +287,7 @@ impl Relay { ws::accept(key) ); let writer = Arc::clone(&outbox); - if (&stream).write_all(switching.as_bytes()).is_ok() + if (&*stream).write_all(switching.as_bytes()).is_ok() && thread::Builder::new() .stack_size(STACK) .spawn(move || writer.drain()) @@ -268,6 +303,75 @@ impl Relay { } let mut state = self.state.lock().unwrap(); depart(&mut state, &self.config, &tag, slot, Instant::now()); + false + } + + /// A polled session's request: `GET` waits for what is to go to it, `POST` brings what + /// it sends, each a run of WebSocket frames. Whether the connection serves another. + fn polled( + &self, + stream: &TcpStream, + reader: &mut BufReader, + head: &str, + id: &str, + ) -> bool { + let length: usize = ws::header(head, "Content-Length") + .and_then(|length| length.parse().ok()) + .unwrap_or(0); + if length > self.config.max_message * 4 { + return false; + } + let mut body = vec![0; length]; + if std::io::Read::read_exact(reader, &mut body).is_err() { + return false; + } + let session = { + let mut state = self.state.lock().unwrap(); + state.polls.get_mut(id).map(|poll| { + poll.last = Instant::now(); + (poll.tag.clone(), poll.slot, Arc::clone(&poll.outbox)) + }) + }; + let Some((tag, slot, outbox)) = session else { + return reply(stream, "410 Gone", b"No such session\n"); + }; + if head.starts_with("POST ") { + let mut frames = ws::Reader::new(&body[..], self.config.max_message, true); + while let Ok(message) = frames.read() { + if !self.heard(&tag, slot, &outbox, message) { + self.end_poll(id); + return reply(stream, "410 Gone", b"Closed\n"); + } + } + return reply(stream, "200 OK", b""); + } + let _ = stream.set_write_timeout(Some(WRITE)); + match outbox.take(WAIT) { + Some(bytes) => { + if let Some(poll) = self.state.lock().unwrap().polls.get_mut(id) { + poll.last = Instant::now(); + } + reply(stream, "200 OK", &bytes) + } + None => { + self.end_poll(id); + reply(stream, "410 Gone", b"Closed\n") + } + } + } + + /// Ends the polled session `id`: it leaves its room. + fn end_poll(&self, id: &str) { + let mut state = self.state.lock().unwrap(); + if let Some(poll) = state.polls.remove(id) { + depart( + &mut state, + &self.config, + &poll.tag, + poll.slot, + Instant::now(), + ); + } } /// Where a peer counts: the address it connected from, or its proxy says it did. @@ -313,7 +417,9 @@ impl Relay { ) -> Result<(String, u32), Refusal> { let config = &self.config; let mut state = self.state.lock().unwrap(); - let State { rooms, addresses } = &mut *state; + let State { + rooms, addresses, .. + } = &mut *state; if !addresses.contains_key(&address) && addresses.len() >= ADDRESSES { return Err(Refusal::Full); } @@ -484,7 +590,9 @@ impl Relay { /// Takes a code's owner's word on the peer in a slot waiting to meet it. fn judge(&self, tag: &str, from: u32, verdict: Verdict) { let mut state = self.state.lock().unwrap(); - let State { rooms, addresses } = &mut *state; + let State { + rooms, addresses, .. + } = &mut *state; let Some(room) = rooms.get_mut(tag).filter(|room| room.owner == Some(from)) else { return; }; @@ -537,6 +645,15 @@ impl Relay { for (tag, slot) in late { depart(&mut state, config, &tag, slot, now); } + let quiet: Vec = (state.polls.iter()) + .filter(|(_, poll)| now.saturating_duration_since(poll.last) >= IDLE_POLL) + .map(|(id, _)| id.clone()) + .collect(); + for id in quiet { + if let Some(poll) = state.polls.remove(&id) { + depart(&mut state, config, &poll.tag, poll.slot, now); + } + } state.addresses.retain(|_, address| { address.refresh(now); !address.quiet(config, now) @@ -547,7 +664,9 @@ impl Relay { /// Takes `slot` out of room `tag` and hangs up on it. One still waiting to meet a code's /// owner tried a wrong code; the owner leaving excuses those waiting for it. fn depart(state: &mut State, config: &Config, tag: &str, slot: u32, now: Instant) { - let State { rooms, addresses } = state; + let State { + rooms, addresses, .. + } = state; let Some(room) = rooms.get_mut(tag) else { return; }; @@ -613,6 +732,38 @@ fn valid(tag: &str) -> bool { .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-') } +/// Answers a polled session's request, keeping the connection: whether that worked. +fn reply(mut stream: &TcpStream, status: &str, body: &[u8]) -> bool { + let head = format!( + "HTTP/1.1 {status}\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\n\ + Cache-Control: no-store\r\n\r\n", + body.len() + ); + stream.write_all(head.as_bytes()).is_ok() && stream.write_all(body).is_ok() +} + +fn refuse(stream: &TcpStream, refusal: Refusal) { + let (status, headers, body) = match refusal { + Refusal::NotFound => ("404 Not Found", String::new(), "No such code\n"), + Refusal::Gone => ("410 Gone", String::new(), "The code has expired\n"), + Refusal::Wait(wait) => ( + "429 Too Many Requests", + // Rounded up, so that a client waiting so long finds it over. + format!( + "Retry-After: {}\r\n", + wait.as_secs() + u64::from(wait.subsec_nanos() > 0) + ), + "Too many tries\n", + ), + Refusal::Full => ( + "503 Service Unavailable", + "Retry-After: 30\r\n".into(), + "The relay is full\n", + ), + }; + respond(stream, status, "text/plain", &headers, body); +} + fn respond(mut stream: &TcpStream, status: &str, kind: &str, headers: &str, body: &str) { let _ = write!( stream, @@ -729,7 +880,8 @@ impl Bucket { /// What waits to be written to one peer, capped in bytes. struct Outbox { - stream: TcpStream, + /// Where its frames go; none for a polled session's, which its `GET`s take. + stream: Option, queue: Mutex, ready: Condvar, most: usize, @@ -745,13 +897,35 @@ struct Queue { impl Outbox { fn new(stream: TcpStream, most: usize) -> Self { Self { - stream, + stream: Some(stream), queue: Mutex::default(), ready: Condvar::new(), most, } } + fn mailbox(most: usize) -> Self { + Self { + stream: None, + queue: Mutex::default(), + ready: Condvar::new(), + most, + } + } + + /// Everything queued, waiting up to `wait` for something; none once closed. + fn take(&self, wait: Duration) -> Option> { + let mut queue = self.queue.lock().unwrap(); + if queue.frames.is_empty() && !queue.closed { + queue = self.ready.wait_timeout(queue, wait).unwrap().0; + } + if queue.closed { + return None; + } + queue.bytes = 0; + Some(queue.frames.drain(..).flatten().collect()) + } + /// Queues `frame`, or hangs up on a peer that reads too slowly to take it. fn push(&self, frame: Vec) { let mut queue = self.queue.lock().unwrap(); @@ -776,7 +950,9 @@ impl Outbox { }; self.ready.notify_one(); drop(queue); - let _ = self.stream.shutdown(Shutdown::Both); + if let Some(stream) = &self.stream { + let _ = stream.shutdown(Shutdown::Both); + } } /// Writes what is queued until closed. @@ -795,7 +971,10 @@ impl Outbox { queue = self.ready.wait(queue).unwrap(); } }; - if (&self.stream).write_all(&frame).is_err() { + let Some(mut stream) = self.stream.as_ref() else { + return; + }; + if stream.write_all(&frame).is_err() { self.close(); return; } diff --git a/crates/snowbound/src/live.rs b/crates/snowbound/src/live.rs index ced80f73d5afe913e4ce96e33e402608054260c2..a63ece404594daf420b52a6b86947c3d7bf90542 100644 --- a/crates/snowbound/src/live.rs +++ b/crates/snowbound/src/live.rs @@ -236,7 +236,7 @@ pub(crate) fn join( code: &str, password: &str, ) -> Result { - let me = hello().map_err(|_| live::share::Refusal::Unreachable)?; + let me = hello().map_err(|_| live::share::Refusal::Unreachable(live::Trouble::Other))?; live::share::join(me, code, password, reach(), relay().as_deref()) } diff --git a/crates/snowbound/src/share.rs b/crates/snowbound/src/share.rs index 8844f2b117edc56be535f0eec3a3410a891c4b28..2dd2910aa0ec334d4313ae91ddf8707fbb515211 100644 --- a/crates/snowbound/src/share.rs +++ b/crates/snowbound/src/share.rs @@ -4,7 +4,7 @@ use crate::{Library, State}; use accesskit::Role; use notebook::live::{ - Relayed, + Relayed, Trouble, share::{self, Refusal, Sharing}, wire::Welcome, }; @@ -90,13 +90,43 @@ fn refusal(refusal: &Refusal) -> String { None => "Too many wrong codes from this network. Try again later.".into(), }, Refusal::Busy => "The Live Share relay is busy. Try again in a minute.".into(), - Refusal::Unreachable => { - "Can’t reach the Live Share relay. Check your internet connection.".into() - } + Refusal::Unreachable(trouble) => unreachable(*trouble), Refusal::TimedOut => "The computer sharing didn’t answer. Try again.".into(), } } +/// What to tell someone whose computer can't reach the relay, and what to try. +fn unreachable(trouble: Trouble) -> String { + match trouble { + Trouble::Dns => "Can’t find the Live Share relay. Check your internet connection, or \ + try another network." + .into(), + Trouble::Unreachable => "Can’t connect to the Live Share relay. A firewall may block \ + it. Try another network." + .into(), + Trouble::TimedOut => "The Live Share relay didn’t answer. A firewall may block it. Try \ + another network." + .into(), + Trouble::ProxyUnreachable => "Can’t reach your proxy server. Check its address in your \ + network settings." + .into(), + Trouble::ProxyAuthentication => "Your proxy server asks for a name and password. Set \ + HTTPS_PROXY to http://name:password@proxy:port, then \ + try again." + .into(), + Trouble::ProxyRefused(status) => format!( + "Your proxy server won’t connect to the Live Share relay ({status}). Ask whoever \ + runs your network to allow relay.snowbound.paperclover.net." + ), + Trouble::Certificate => "Something on your network replaced the relay’s security \ + certificate. If your network inspects secure connections, add \ + its certificate to this computer’s trusted certificates." + .into(), + Trouble::Blocked => "Your network blocks the Live Share relay. Try another network.".into(), + Trouble::Other => "Can’t reach the Live Share relay. Try again in a moment.".into(), + } +} + /// A dialog's frame, titled `title` with the BETA badge. fn frame(ui: &mut Ui, id: Id, title: &str) { let theme = ui.theme.clone(); @@ -346,9 +376,17 @@ impl State { true, ); match host.relayed() { - Relayed::Unreachable | Relayed::Refused(..) => status( + Relayed::Unreachable(trouble) => status( ui, - "Can’t reach the Live Share relay. Only people on this network can join.", + &format!( + "{} Until then, only people on this network can join.", + unreachable(trouble) + ), + ), + Relayed::Refused(..) => status( + ui, + "The Live Share relay turned this computer away. Only people on this \ + network can join.", ), _ => {} } diff --git a/crates/snowbound/src/update.rs b/crates/snowbound/src/update.rs index fc19bf4d14ba5bea4b82e4b1d6eebd5187b85000..568629556a4a8d82e7ec5b63a5079c85c222d003 100644 --- a/crates/snowbound/src/update.rs +++ b/crates/snowbound/src/update.rs @@ -563,7 +563,23 @@ fn download(path: &str, limit: u64) -> Result, String> { let roots = (system.iter().chain(bundled)) .map(|der| Certificate::from_der(der).to_owned()) .collect(); + // The proxy Live Share's relay goes through, the system's included; ureq reads only the + // environment's. + #[cfg(feature = "live")] + let proxy = { + let host = BASE.trim_start_matches("https://").split('/').next(); + notebook::live::proxy::for_host(host.unwrap_or_default(), true).and_then(|proxy| { + let credentials = (proxy.credentials.as_ref()) + .map_or_else(String::new, |(name, password)| { + format!("{name}:{password}@") + }); + ureq::Proxy::new(&format!("http://{credentials}{proxy}")).ok() + }) + }; + #[cfg(not(feature = "live"))] + let proxy = ureq::Proxy::try_from_env(); let agent: ureq::Agent = ureq::Agent::config_builder() + .proxy(proxy) .tls_config( TlsConfig::builder() .root_certs(RootCerts::Specific(Arc::new(roots)))