diff --git a/AGENTS.md b/AGENTS.md index 8264ca6d5bed92cc8bda7c9673153951a533a3d3..1736a094db53fa216108f3dc5a72253f9661b228 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -13,7 +13,8 @@ novel UI kit, each new platform port is extremely lightweight. | Path | Owns | Depends on | | --- | --- | --- | | `crates/onestore` | The file format: revision stores (`.one`, `.onetoc2`), the page model, ops, the commit protocol. No network, no SQLite, no `unsafe`. | none | -| `crates/notebook` | From editor to disk or share: discovery, notebook structure, sessions, the SQLite replica, sync and merging, conflict pages, the embedded SMB client (feature `smb`), live presence between peers (feature `live`). | onestore | +| `crates/notebook` | From editor to disk or share: discovery, notebook structure, sessions, the SQLite replica, sync and merging, conflict pages, the embedded SMB client (feature `smb`), live presence between peers (feature `live`). | onestore, relay | +| `crates/relay` | `snowbound-relay`, the WebSocket relay Live Share meets through off the LAN, and the framing and notices its clients share with it. Sees only sealed frames. | none | | `crates/draw` | The wgpu renderer that page and chrome both paint through, and the text-editing core (keys, chords, carets) they share. | none | | `crates/canvas` | The page: editor, OneNote-faithful layout, page scene, interaction, the page's accessibility tree. | onestore, draw | | `crates/ui` | The immediate-mode interface kit and OneNote's chrome controls. Knows nothing of notebooks. | draw | diff --git a/Cargo.lock b/Cargo.lock index a45e40ee46a458129d14eaa17fe7a28f94f27e50..eb72c37fc618d978e8ea54d1b1887c8087f5d4c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2189,8 +2189,11 @@ dependencies = [ "minicbor", "nix", "onestore", + "relay", "rsqlite-vfs", "rusqlite", + "rustls", + "rustls-native-certs", "serde", "serde_json", "sha2", @@ -2202,6 +2205,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-time", + "webpki-root-certs", "windows-sys 0.61.2", "zeroize", ] @@ -3132,6 +3136,15 @@ dependencies = [ "bitflags 2.13.1", ] +[[package]] +name = "relay" +version = "0.1.0" +dependencies = [ + "base64", + "getrandom 0.4.3", + "sha1", +] + [[package]] name = "renderdoc-sys" version = "1.1.0" diff --git a/crates/notebook/Cargo.toml b/crates/notebook/Cargo.toml index 11fb79992c50189bb798ca5191c0f8a901f09c10..1c12191aad2634504ded7d482ed21a8358cf3fe8 100644 --- a/crates/notebook/Cargo.toml +++ b/crates/notebook/Cargo.toml @@ -6,9 +6,9 @@ publish = false [features] smb = ["dep:smb2", "dep:tokio"] -# Live presence: peers found on the network, met through a shared secret, over an encrypted -# stream (resources/live-share.md). -live = ["dep:mdns-sd", "dep:minicbor", "dep:spake2"] +# Live presence: peers found on the network or met through a relay, through a shared secret, +# over an encrypted stream (resources/live-share.md). +live = ["dep:mdns-sd", "dep:minicbor", "dep:spake2", "dep:relay", "dep:rustls", "dep:rustls-native-certs", "dep:webpki-root-certs"] [dependencies] onestore = { path = "../onestore" } @@ -29,6 +29,11 @@ hmac = "0.13.0" mdns-sd = { version = "0.21.4", default-features = false, optional = true } minicbor = { version = "2.3.0", features = ["derive", "alloc"], optional = true } spake2 = { version = "=0.5.0-pre.0", features = ["getrandom"], optional = true } +relay = { path = "../relay", optional = true } +# A relay's TLS, trusting what the app's updates trust: the system's authorities, then Mozilla's. +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"], optional = true } +rustls-native-certs = { version = "0.8", optional = true } +webpki-root-certs = { version = "1.0", optional = true } # OneNote packages (.onepkg) are cabinet files. cab = "0.6" # std's clock where there is one; the browser's on wasm32-unknown-unknown, whose std has none. diff --git a/crates/notebook/README.md b/crates/notebook/README.md index db6126f899eddf38c68d54fb0c934ed25e609243..23699b632c0432db56f7a074f2748385b85c2f53 100644 --- a/crates/notebook/README.md +++ b/crates/notebook/README.md @@ -531,16 +531,22 @@ physical power-loss durability. ## Live presence (feature `live`) -`live::Live::start(hello, room, reach, notify)` listens on a TCP port and, with a `Reach`, -advertises `_snowbound._tcp` by mDNS on every network or on loopback alone, connecting to the -peers in the same `Room` that it finds: a notebook's identity, or a code typed on both -(`Room::Code("4-violet-otter")`). `connect(address)` meets a peer discovery did not find. -Peers meet through SPAKE2 on the room's secret, then every frame is AES-256-GCM under the keys -it agreed: a message kind and a CBOR map (`live::wire`). A reader skips kinds and map keys it -doesn't know, so later versions add both freely. `set_presence` says which section, page and -caret this end has (text object and UTF-16 offset, as ops address text); a connection sends -only the newest. `peers()` lists each connected peer's `Hello` (name, picture) and presence, -and `notify` runs whenever that changes. Dropping the `Live` leaves. +`live::Live::start(hello, room, reach, relay, notify)` listens on a TCP port and, with a +`Reach`, advertises `_snowbound._tcp` by mDNS on every network or on loopback alone, +connecting to the peers in the same `Room` that it finds: a notebook's identity, or a code +typed on both (`Room::Code("4-violet-otter")`). With a `relay` (`wss://live.example.net`, +`crates/relay`) it also joins the room there and meets its peers through it; a code's words +alone (`Room::Code("violet-otter")`) ask the relay for a number, and `code()` then has the +whole code. `connect(address)` meets a peer discovery did not find. Peers meet through +SPAKE2 on the room's secret, then every frame is AES-256-GCM under the keys it agreed: its +number, which is also its nonce, then a message kind and a CBOR map (`live::wire`). A frame +lost, repeated, reordered or forged on the way fails where it lands; the connection is +dropped as broken, nothing from it after the fault is applied, and the ends meet again from +scratch. A reader skips kinds and map keys it doesn't know, so later versions add both +freely. `set_presence` says which section, page and caret this end has (text object and +UTF-16 offset, as ops address text); a connection sends only the newest. `peers()` lists +each connected peer's `Hello` (name, picture) and presence, and `notify` runs whenever that +changes. Dropping the `Live` leaves. ## Queue measurement diff --git a/crates/notebook/src/live.rs b/crates/notebook/src/live.rs index a6bfeb8275cfbd90d6db066c47026104e2725a0a..17347158e5fb8b25185b5eb2350b47cdbdb9fcb1 100644 --- a/crates/notebook/src/live.rs +++ b/crates/notebook/src/live.rs @@ -1,9 +1,12 @@ //! Live presence: who else has the notebook open, the page they are on and their caret, -//! straight from one Snowbound to another. Peers find each other with mDNS -//! (`_snowbound._tcp`) and meet through a secret both hold, a notebook's identity or a code -//! typed on both, which SPAKE2 turns into the keys every frame after the opening is sealed -//! with. The lower peer id connects; each connection has a thread reading and one writing. +//! straight from one Snowbound to another, or through a relay (`crates/relay`) where they +//! aren't on one network. Peers find each other with mDNS (`_snowbound._tcp`) or in the +//! relay's room, and meet through a secret both hold, a notebook's identity or a code typed +//! on both, which SPAKE2 turns into the keys every frame after the opening is sealed with. +//! Each connection has a thread reading and one writing. A connection whose frames arrive +//! out of order is dropped and met again from scratch. +mod relay; pub mod wire; pub use wire::{Caret, Guid, Hello, Presence, Spot}; @@ -11,7 +14,7 @@ use mdns_sd::{IfKind, ServiceDaemon, ServiceEvent, ServiceInfo}; use sha2::{Digest, Sha256}; use std::{ collections::BTreeMap, - io, + io::{self, Read, Write}, net::{IpAddr, Ipv4Addr, Shutdown, SocketAddr, TcpListener, TcpStream}, sync::{ Arc, Mutex, @@ -28,6 +31,8 @@ const SERVICE: &str = "_snowbound._tcp.local."; const PING: Duration = Duration::from_secs(15); const GONE: Duration = Duration::from_secs(45); const OPENING: Duration = Duration::from_secs(5); +/// The longest wait before meeting again. +const PATIENCE: Duration = Duration::from_secs(30); /// The secret peers meet through. #[derive(Clone, Debug, PartialEq, Eq)] @@ -36,27 +41,40 @@ pub enum Room { /// hold. Notebook([u8; 16]), /// A code typed on both: `7-violet-otter`, whose number names it on the network and whose - /// words only the two people know. + /// words only the two people know. Its words alone (`violet-otter`) ask the relay for a + /// free number, and `Live::code` then has the whole code. Code(String), } impl Room { - /// What names the room in the clear: a hash of a notebook's identity, a code's number. - fn tag(&self) -> String { + /// What names the room in the clear: a hash of a notebook's identity, a code's number; + /// none for a code the relay hasn't numbered. + fn tag(&self) -> Option { match self { - Room::Notebook(id) => hex(&Sha256::digest([&b"Snowbound room "[..], id].concat())[..8]), - Room::Code(code) => format!("code-{}", code.split('-').next().unwrap_or_default()), + Room::Notebook(id) => Some(hex(&Sha256::digest( + [&b"Snowbound room "[..], id].concat(), + )[..8])), + Room::Code(code) => code_parts(code).0.map(|number| format!("code-{number}")), } } fn secret(&self) -> Vec { match self { Room::Notebook(id) => id.to_vec(), - Room::Code(code) => code.trim().to_lowercase().into_bytes(), + Room::Code(code) => code_parts(code).1.to_lowercase().into_bytes(), } } } +/// A code's number, if it has one, and its words. +fn code_parts(code: &str) -> (Option, &str) { + let code = code.trim(); + match code.split_once('-') { + Some((number, words)) if number.parse::().is_ok() => (number.parse().ok(), words), + _ => (None, code), + } +} + /// Where peers are looked for. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Reach { @@ -82,7 +100,7 @@ pub struct Live { struct Shared { me: Hello, - tag: String, + room: Room, secret: Vec, state: Mutex, notify: Box, @@ -96,25 +114,93 @@ struct State { /// Counts changes to `presence`, so a writer sends only the newest. generation: u64, peers: BTreeMap<[u8; 16], Link>, + /// The code others type, once known. + code: Option, + /// The relay connection open now, to hang up on leaving. + relay: Option>, } struct Link { connection: u64, peer: Peer, wake: mpsc::Sender<()>, - stream: TcpStream, + pipe: Arc, +} + +/// A stream to one peer, read by one thread and written by another: a TCP connection, or +/// one carried through a relay. +trait Pipe: Send + Sync { + fn read(&self, buffer: &mut [u8]) -> io::Result; + fn write(&self, bytes: &[u8]) -> io::Result; + fn set_read_timeout(&self, timeout: Duration) -> io::Result<()>; + /// Hangs up, ending the thread reading. + fn shutdown(&self); + /// Whether it goes straight to the peer, which is better than through a relay. + fn direct(&self) -> bool; + /// Hears whether the peer at the other end knew the secret. + fn met(&self, _met: bool) {} + /// Hears that frames arrived lost, repeated, reordered or forged, after which this end + /// meets its peers again. + fn broken(&self) {} +} + +impl Pipe for TcpStream { + fn read(&self, buffer: &mut [u8]) -> io::Result { + Read::read(&mut &*self, buffer) + } + + fn write(&self, bytes: &[u8]) -> io::Result { + Write::write(&mut &*self, bytes) + } + + fn set_read_timeout(&self, timeout: Duration) -> io::Result<()> { + TcpStream::set_read_timeout(self, Some(timeout)) + } + + fn shutdown(&self) { + let _ = TcpStream::shutdown(self, Shutdown::Both); + } + + fn direct(&self) -> bool { + true + } +} + +impl Read for &dyn Pipe { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + Pipe::read(*self, buffer) + } +} + +impl Write for &dyn Pipe { + fn write(&mut self, bytes: &[u8]) -> io::Result { + Pipe::write(*self, bytes) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } } impl Live { /// Starts listening as `me` in `room`, advertised and looked for where `reach` says, or - /// not at all with `None`, leaving peers to `connect`. `notify` runs on a network thread - /// whenever `peers` changes. + /// not at all with `None`, leaving peers to `connect`, and in the room at `relay` + /// (`wss://live.example.net`) where given. `notify` runs on a network thread whenever + /// `peers` or `code` changes. pub fn start( me: Hello, room: &Room, reach: Option, + relay: Option<&str>, notify: impl Fn() + Send + Sync + 'static, ) -> io::Result { + let tag = room.tag(); + if tag.is_none() && relay.is_none() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "Only a relay can number a code", + )); + } let host = match reach { Some(Reach::Network) => IpAddr::V4(Ipv4Addr::UNSPECIFIED), Some(Reach::Loopback) | None => IpAddr::V4(Ipv4Addr::LOCALHOST), @@ -124,16 +210,27 @@ impl Live { if address.ip().is_unspecified() { address.set_ip(IpAddr::V4(Ipv4Addr::LOCALHOST)); } + let code = match room { + Room::Code(code) if tag.is_some() => Some(code.trim().to_owned()), + _ => None, + }; let shared = Arc::new(Shared { me, - tag: room.tag(), + room: room.clone(), secret: room.secret(), - state: Mutex::default(), + state: Mutex::new(State { + code, + ..State::default() + }), notify: Box::new(notify), stopped: AtomicBool::new(false), connections: AtomicU64::new(0), }); + if let Some(relay) = relay { + relay::join(&shared, relay)?; + } let accepting = Arc::clone(&shared); + let accepted = tag.clone(); thread::Builder::new() .name("live accept".into()) .spawn(move || { @@ -141,17 +238,17 @@ impl Live { if accepting.stopped.load(Ordering::Acquire) { return; } - if let Ok(stream) = stream { + if let (Ok(stream), Some(tag)) = (stream, accepted.clone()) { let shared = Arc::clone(&accepting); - thread::spawn(move || shared.run(stream, Side::Responder)); + thread::spawn(move || shared.run(Arc::new(stream), Side::Responder, &tag)); } } })?; - let daemon = match reach { - Some(reach) => { - Some(advertise(&shared, reach, address.port()).map_err(io::Error::other)?) + let daemon = match (reach, tag) { + (Some(reach), Some(tag)) => { + Some(advertise(&shared, reach, tag, address.port()).map_err(io::Error::other)?) } - None => None, + _ => None, }; Ok(Live { shared, @@ -165,6 +262,12 @@ impl Live { self.address } + /// The code others type to meet this end: the one it was given, or the one the relay + /// numbered. + pub fn code(&self) -> Option { + self.shared.state.lock().unwrap().code.clone() + } + /// Connects to a peer at `address` that discovery did not find. pub fn connect(&self, address: SocketAddr) { let shared = Arc::clone(&self.shared); @@ -199,22 +302,27 @@ impl Drop for Live { } // Wakes the accepting thread to see it has stopped. let _ = TcpStream::connect_timeout(&self.address, OPENING); - for link in self.shared.state.lock().unwrap().peers.values() { - let _ = link.stream.shutdown(Shutdown::Both); + let state = self.shared.state.lock().unwrap(); + for link in state.peers.values() { + link.pipe.shutdown(); + } + if let Some(relay) = &state.relay { + relay.hang_up(); } } } -/// Advertises `shared` on `port` and connects to the peers in its room that discovery finds -/// with a higher id than its own, which leave the connecting to it. -fn advertise(shared: &Arc, reach: Reach, port: u16) -> mdns_sd::Result { +/// Advertises `shared` as in room `tag` on `port` and connects to the peers in its room that +/// discovery finds with a higher id than its own, which leave the connecting to it. +fn advertise( + shared: &Arc, + reach: Reach, + tag: String, + port: u16, +) -> mdns_sd::Result { let daemon = ServiceDaemon::new()?; let id = hex(&shared.me.peer); - let properties = [ - ("v", "1"), - ("room", shared.tag.as_str()), - ("peer", id.as_str()), - ]; + let properties = [("v", "1"), ("room", tag.as_str()), ("peer", id.as_str())]; let host = format!("snowbound-{id}.local."); let info = match reach { Reach::Network => { @@ -249,7 +357,7 @@ fn advertise(shared: &Arc, reach: Reach, port: u16) -> mdns_sd::Result = @@ -272,21 +380,35 @@ impl Shared { state.peers.keys().any(|id| hex(id) == peer) } + /// Connects to `address`, and again while the peer is there and the connection was the + /// one this end kept. fn dial(self: Arc, address: SocketAddr) { - if let Ok(stream) = TcpStream::connect_timeout(&address, OPENING) { - self.run(stream, Side::Initiator); + let Some(tag) = self.room.tag() else { + return; + }; + let mut wait = Duration::from_secs(1); + while let Ok(stream) = TcpStream::connect_timeout(&address, OPENING) { + if !Arc::clone(&self).run(Arc::new(stream), Side::Initiator, &tag) + || self.stopped.load(Ordering::Acquire) + { + return; + } + thread::sleep(wait); + wait = (wait * 2).min(PATIENCE); } } - /// Meets the peer at the other end of `stream`, then reads from it until it goes. - fn run(self: Arc, mut stream: TcpStream, side: Side) { + /// Meets the peer at the other end of `pipe` in room `tag`, then reads from it until it + /// goes: whether it was the connection kept to that peer. + fn run(self: Arc, pipe: Arc, side: Side, tag: &str) -> bool { if self.stopped.load(Ordering::Acquire) { - return; + pipe.shutdown(); + return false; } + let mut stream: &dyn Pipe = &*pipe; let met = (|| { - stream.set_read_timeout(Some(OPENING))?; - stream.set_nodelay(true)?; - let (mut send, mut receive) = wire::open(&mut stream, side, &self.tag, &self.secret)?; + stream.set_read_timeout(OPENING)?; + let (mut send, mut receive) = wire::open(&mut stream, side, tag, &self.secret)?; send.send(&mut stream, kind::HELLO, &self.me)?; let (first, body) = receive.receive(&mut stream)?; if first != kind::HELLO { @@ -294,27 +416,40 @@ impl Shared { } let hello: Hello = minicbor::decode(&body) .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "A malformed hello"))?; - stream.set_read_timeout(Some(GONE))?; + stream.set_read_timeout(GONE)?; Ok((send, receive, hello)) })(); + pipe.met(met.is_ok()); let (send, mut receive, hello) = match met { Ok(met) => met, Err(error) => { - eprintln!("Live: no meeting with {:?}: {error}", stream.peer_addr()); - return; + eprintln!("Live: no meeting in {tag}: {error}"); + pipe.shutdown(); + return false; } }; let peer = hello.peer; + let name = hello.name.clone(); let connection = self.connections.fetch_add(1, Ordering::Relaxed); let (wake, woken) = mpsc::channel(); { let mut state = self.state.lock().unwrap(); - if peer == self.me.peer || state.peers.contains_key(&peer) { - return; - } - let Ok(writing) = stream.try_clone() else { - return; + // A peer met both directly and through a relay keeps the direct connection, as + // both ends then agree. + let kept = match state.peers.get(&peer) { + _ if peer == self.me.peer => false, + Some(link) if link.pipe.direct() || !pipe.direct() => false, + Some(link) => { + link.pipe.shutdown(); + true + } + None => true, }; + if !kept { + drop(state); + pipe.shutdown(); + return false; + } let _ = wake.send(()); state.peers.insert( peer, @@ -325,28 +460,35 @@ impl Shared { presence: None, }, wake, - stream: writing, + pipe: Arc::clone(&pipe), }, ); } (self.notify)(); - if let Ok(writing) = stream.try_clone() { - let shared = Arc::clone(&self); - thread::spawn(move || shared.write(writing, send, woken)); - } - while let Ok((message, body)) = receive.receive(&mut stream) { + let writing = Arc::clone(&pipe); + let shared = Arc::clone(&self); + thread::spawn(move || shared.write(&*writing, send, woken)); + let ended = loop { + let (message, body) = match receive.receive(&mut stream) { + Ok(frame) => frame, + Err(error) => break Some(error), + }; if message != kind::PRESENCE { continue; } let Ok(presence) = minicbor::decode::(&body) else { - break; + break None; }; if let Some(link) = self.state.lock().unwrap().peers.get_mut(&peer) { link.peer.presence = Some(presence); } (self.notify)(); + }; + if let Some(error) = ended.filter(|error| error.kind() == io::ErrorKind::InvalidData) { + eprintln!("Live: the connection to {name} broke ({error}); meeting again"); + pipe.broken(); } - let _ = stream.shutdown(Shutdown::Both); + pipe.shutdown(); let mut state = self.state.lock().unwrap(); if state .peers @@ -357,10 +499,12 @@ impl Shared { drop(state); (self.notify)(); } + true } /// Sends the newest presence whenever woken, and a ping when quiet. - fn write(&self, mut stream: TcpStream, mut send: Sealer, woken: mpsc::Receiver<()>) { + fn write(&self, pipe: &dyn Pipe, mut send: Sealer, woken: mpsc::Receiver<()>) { + let mut stream = pipe; let mut sent = None; loop { let result = match woken.recv_timeout(PING) { @@ -379,7 +523,7 @@ impl Shared { Err(mpsc::RecvTimeoutError::Disconnected) => return, }; if result.is_err() { - let _ = stream.shutdown(Shutdown::Both); + pipe.shutdown(); return; } } diff --git a/crates/notebook/src/live/relay.rs b/crates/notebook/src/live/relay.rs new file mode 100644 index 0000000000000000000000000000000000000000..cfa5256e878dc2fb997bc58179249c78016fe9b6 --- /dev/null +++ b/crates/notebook/src/live/relay.rs @@ -0,0 +1,537 @@ +//! Meeting peers through a relay (`crates/relay`): one WebSocket to the room, carrying a +//! stream to each peer in it, each opened with SPAKE2 and sealed end to end as on a LAN, so +//! 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::{OPENING, PATIENCE, Pipe, Shared, Side, code_parts}; +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}, + 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. +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. +const CHUNK: usize = 64 << 10; +/// The largest message read from the relay. +const MOST: usize = 1 << 20; + +type Reader = ws::Reader>>; + +/// Joins `shared`'s room at the relay at `url`, and keeps joining whenever it falls out. +pub(super) fn join(shared: &Arc, url: &str) -> io::Result<()> { + let address = parse(url)?; + let shared = Arc::clone(shared); + thread::Builder::new() + .name("live relay".into()) + .spawn(move || keep(&shared, &address))?; + 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) + } +} + +/// Stays in the room at `address`, joining again whenever the connection ends, until +/// `shared` stops. +fn keep(shared: &Arc, address: &Address) { + let mut wait = Duration::from_secs(1); + // The number the relay gave this end's code, asked for again on joining again. + let mut nameplate = None; + while !shared.stopped.load(Ordering::Acquire) { + let began = Instant::now(); + let tag = shared.room.tag(); + let path = match &tag { + Some(tag) => format!("{}/v1/room/{tag}", address.path), + None => match nameplate { + Some(number) => format!("{}/v1/claim?nameplate={number}", address.path), + None => format!("{}/v1/claim", address.path), + }, + }; + match connect(address, &path, tag.is_none()) { + Ok((socket, reader)) => { + shared.state.lock().unwrap().relay = Some(Arc::clone(&socket)); + if !shared.stopped.load(Ordering::Acquire) { + session(shared, &socket, reader, &mut nameplate); + } + shared.state.lock().unwrap().relay = None; + socket.hang_up(); + } + Err(Failure::Refused(410, _)) => { + eprintln!("Live: the code has expired; ask for a new one"); + return; + } + Err(Failure::Refused(status, retry)) => { + eprintln!("Live: the relay refused to let this end in ({status})"); + wait = wait.max(retry.unwrap_or_default()); + } + Err(Failure::Network(error)) => { + eprintln!("Live: no relay at {}: {error}", address.authority); + } + } + if began.elapsed() >= STEADY { + wait = Duration::from_secs(1); + } + thread::sleep(wait); + wait = (wait * 2).min(PATIENCE); + } +} + +/// Opens a WebSocket to `path` at `address`: the connection, and its reading half. +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 socket = Arc::new(Socket { + send: Mutex::new(writing), + tcp, + 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(()) + } +} + +/// One connection to a relay's room. +pub(super) struct Socket { + send: Mutex>, + tcp: TcpStream, + /// Whether this end claimed the room for its code, and so tells the relay who knew it. + owner: bool, + links: Mutex, +} + +/// The streams a socket carries, by the slot of the peer at the other end. +#[derive(Default)] +struct Links { + /// This end's slot. + me: u32, + inboxes: HashMap>>, + /// Slots whose stream ended; a peer that comes back has a new slot. + ended: HashSet, +} + +impl Socket { + fn send(&self, opcode: u8, payload: &[u8]) -> io::Result<()> { + let mut mask = [0; 4]; + getrandom::fill(&mut mask).map_err(|_| io::Error::other("System random source failed"))?; + let frame = ws::frame(opcode, payload, Some(mask)); + self.send.lock().unwrap().write_all(&frame) + } + + pub(super) fn hang_up(&self) { + let _ = self.tcp.shutdown(Shutdown::Both); + } + + fn forget(&self, slot: u32) { + let mut links = self.links.lock().unwrap(); + links.inboxes.remove(&slot); + links.ended.insert(slot); + } +} + +/// Reads notices and peers' bytes from the relay until the connection ends, opening a stream +/// to each peer in the room. +fn session( + shared: &Arc, + socket: &Arc, + mut reader: Reader, + nameplate: &mut Option, +) { + let mut tag = shared.room.tag(); + // Pings while the session lasts: dropping `_beat` at its end stops them. + let (_beat, beats) = mpsc::channel::<()>(); + let pinging = Arc::clone(socket); + thread::spawn(move || { + while let Err(mpsc::RecvTimeoutError::Timeout) = beats.recv_timeout(KEEPALIVE) { + if pinging.send(ws::PING, &[]).is_err() { + return; + } + } + }); + while let Ok(message) = reader.read() { + match message { + ws::Message::Text(text) => match text.parse() { + Ok(Notice::Nameplate(number)) => { + *nameplate = Some(number); + tag = Some(format!("code-{number}")); + let super::Room::Code(words) = &shared.room else { + continue; + }; + let code = format!("{number}-{}", code_parts(words).1); + let mut state = shared.state.lock().unwrap(); + if state.code.as_ref() != Some(&code) { + eprintln!("Live: the code is {code}"); + state.code = Some(code); + drop(state); + (shared.notify)(); + } + } + Ok(Notice::Welcome { you, members }) => { + socket.links.lock().unwrap().me = you; + for slot in members { + meet(shared, socket, tag.as_deref(), slot, None); + } + } + Ok(Notice::Joined(slot)) => meet(shared, socket, tag.as_deref(), slot, None), + Ok(Notice::Left(slot)) => socket.forget(slot), + Ok(Notice::Burned) => { + // Coming back, ask for another number. + *nameplate = None; + eprintln!("Live: too many wrong tries; the code admits no one new"); + } + Err(()) => {} + }, + ws::Message::Binary(data) => { + if let Some((slot, bytes)) = data.split_first_chunk::() { + let slot = u32::from_be_bytes(*slot); + meet(shared, socket, tag.as_deref(), slot, Some(bytes.to_vec())); + } + } + ws::Message::Ping(payload) => { + let _ = socket.send(ws::PONG, &payload); + } + ws::Message::Pong => {} + ws::Message::Close => break, + } + } + socket.links.lock().unwrap().inboxes.clear(); +} + +/// Hands `bytes` from the peer in `slot` to its stream, or opens one: this end opens a stream +/// to a peer with a lower slot when told of it, and answers one with a higher slot when its +/// first bytes come. +fn meet( + shared: &Arc, + socket: &Arc, + tag: Option<&str>, + slot: u32, + bytes: Option>, +) { + let mut links = socket.links.lock().unwrap(); + if let Some(inbox) = links.inboxes.get(&slot) { + if let Some(bytes) = bytes { + let _ = inbox.send(bytes); + } + return; + } + let side = match bytes { + None if slot < links.me => Side::Initiator, + Some(_) if slot > links.me => Side::Responder, + _ => return, + }; + let Some(tag) = tag.filter(|_| !links.ended.contains(&slot)) else { + return; + }; + let (inbox, arriving) = mpsc::channel(); + if let Some(bytes) = bytes { + let _ = inbox.send(bytes); + } + links.inboxes.insert(slot, inbox); + let pipe = Arc::new(Relayed { + socket: Arc::clone(socket), + slot, + inbox: Mutex::new(Inbox { + arriving, + chunk: Vec::new(), + at: 0, + }), + timeout: Mutex::new(OPENING), + }); + let shared = Arc::clone(shared); + let tag = tag.to_owned(); + thread::spawn(move || shared.run(pipe, side, &tag)); +} + +/// The stream to one peer through the relay. +struct Relayed { + socket: Arc, + slot: u32, + inbox: Mutex, + timeout: Mutex, +} + +struct Inbox { + arriving: mpsc::Receiver>, + chunk: Vec, + at: usize, +} + +impl Pipe for Relayed { + fn read(&self, buffer: &mut [u8]) -> io::Result { + let mut inbox = self.inbox.lock().unwrap(); + while inbox.at == inbox.chunk.len() { + let timeout = *self.timeout.lock().unwrap(); + inbox.chunk = match inbox.arriving.recv_timeout(timeout) { + Ok(chunk) => chunk, + Err(mpsc::RecvTimeoutError::Timeout) => return Err(io::ErrorKind::TimedOut.into()), + Err(mpsc::RecvTimeoutError::Disconnected) => return Ok(0), + }; + inbox.at = 0; + } + let at = inbox.at; + let length = buffer.len().min(inbox.chunk.len() - at); + buffer[..length].copy_from_slice(&inbox.chunk[at..at + length]); + inbox.at += length; + Ok(length) + } + + fn write(&self, bytes: &[u8]) -> io::Result { + if self.socket.links.lock().unwrap().ended.contains(&self.slot) { + return Err(io::ErrorKind::BrokenPipe.into()); + } + let length = bytes.len().min(CHUNK); + let message = [&self.slot.to_be_bytes()[..], &bytes[..length]].concat(); + self.socket.send(ws::BINARY, &message)?; + Ok(length) + } + + fn set_read_timeout(&self, timeout: Duration) -> io::Result<()> { + *self.timeout.lock().unwrap() = timeout; + Ok(()) + } + + fn shutdown(&self) { + self.socket.forget(self.slot); + } + + fn direct(&self) -> bool { + false + } + + fn met(&self, met: bool) { + if self.socket.owner { + let verdict = if met { Verdict::Met } else { Verdict::Failed }(self.slot); + let _ = self.socket.send(ws::TEXT, verdict.to_string().as_bytes()); + } + } + + fn broken(&self) { + self.socket.hang_up(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn relay_addresses_parse() { + let address = parse("wss://live.example.net").unwrap(); + assert_eq!( + address, + Address { + tls: true, + authority: "live.example.net".into(), + host: "live.example.net".into(), + port: 443, + path: String::new(), + } + ); + let address = parse("ws://[::1]:7650/snowbound/").unwrap(); + assert_eq!((address.host.as_str(), address.port), ("::1", 7650)); + assert_eq!(address.path, "/snowbound"); + assert!(parse("https://live.example.net").is_err()); + assert!(parse("ws://:80").is_err()); + } +} diff --git a/crates/notebook/src/live/tests.rs b/crates/notebook/src/live/tests.rs index 2d5efa137c81df365a89a7adab3aed54cefd9d3d..d25172625af521d330ee38f7e0f38c3af1be4c36 100644 --- a/crates/notebook/src/live/tests.rs +++ b/crates/notebook/src/live/tests.rs @@ -44,8 +44,8 @@ fn caret(offset: u32) -> Presence { #[test] fn peers_meet_and_follow_presence() { let room = Room::Code("7-violet-otter".into()); - let ada = Live::start(hello("Ada"), &room, None, || {}).unwrap(); - let grace = Live::start(hello("Grace"), &room, None, || {}).unwrap(); + let ada = Live::start(hello("Ada"), &room, None, None, || {}).unwrap(); + let grace = Live::start(hello("Grace"), &room, None, None, || {}).unwrap(); ada.set_presence(caret(1)); ada.connect(grace.address()); let seen = until(&grace, |peers| { @@ -72,6 +72,7 @@ fn another_code_never_meets() { hello("Ada"), &Room::Code("7-violet-otter".into()), None, + None, || {}, ) .unwrap(); @@ -79,6 +80,7 @@ fn another_code_never_meets() { hello("Mallory"), &Room::Code("7-violet-ocelot".into()), None, + None, || {}, ) .unwrap(); @@ -114,12 +116,17 @@ fn later_fields_and_kinds_are_skipped() { ); let room = Room::Code("4-quiet-heron".into()); - let grace = Live::start(hello("Grace"), &room, None, || {}).unwrap(); + let grace = Live::start(hello("Grace"), &room, None, None, || {}).unwrap(); // A later version: it greets, says something new, then where it is. let later = thread::spawn(move || { let mut stream = TcpStream::connect(grace.address()).unwrap(); - let (mut send, mut receive) = - wire::open(&mut stream, Side::Initiator, &room.tag(), &room.secret()).unwrap(); + let (mut send, mut receive) = wire::open( + &mut stream, + Side::Initiator, + &room.tag().unwrap(), + &room.secret(), + ) + .unwrap(); send.send(&mut stream, kind::HELLO, &hello("Later")) .unwrap(); receive.receive(&mut stream).unwrap(); @@ -139,8 +146,362 @@ fn later_fields_and_kinds_are_skipped() { #[ignore = "multicasts mDNS on the loopback interface"] fn peers_find_each_other_on_loopback() { let room = Room::Notebook([9; 16]); - let ada = Live::start(hello("Ada"), &room, Some(Reach::Loopback), || {}).unwrap(); - let grace = Live::start(hello("Grace"), &room, Some(Reach::Loopback), || {}).unwrap(); + let ada = Live::start(hello("Ada"), &room, Some(Reach::Loopback), None, || {}).unwrap(); + let grace = Live::start(hello("Grace"), &room, Some(Reach::Loopback), None, || {}).unwrap(); until(&ada, |peers| peers.len() == 1); until(&grace, |peers| peers.len() == 1); } + +/// Two sealers that met over loopback: Ada's to send with and Grace's to receive with. +fn sealers() -> (Sealer, Sealer) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let ada = thread::spawn(move || { + let mut stream = TcpStream::connect(address).unwrap(); + wire::open(&mut stream, Side::Initiator, "room", b"secret") + .unwrap() + .0 + }); + let (mut stream, _) = listener.accept().unwrap(); + let grace = wire::open(&mut stream, Side::Responder, "room", b"secret") + .unwrap() + .1; + (ada.join().unwrap(), grace) +} + +/// Ada's carets at offsets `0..count`, each frame as its own block of bytes. +fn frames(ada: &mut Sealer, count: u32) -> Vec> { + (0..count) + .map(|offset| { + let mut block = Vec::new(); + ada.send(&mut block, kind::PRESENCE, &caret(offset)) + .unwrap(); + block + }) + .collect() +} + +/// A frame lost, repeated, reordered, altered, forged or sealed for another meeting is caught +/// where it lands, before anything in it or after it is read. +#[test] +fn frames_out_of_place_are_caught() { + let (mut other, _) = sealers(); + let elsewhere = frames(&mut other, 3).remove(2); + type Edit = dyn Fn(&mut Vec>); + let cases: [(&str, Box, &str); 6] = [ + ( + "lost", + Box::new(|sent| drop(sent.remove(2))), + "Frame 3 came where frame 2 was due", + ), + ( + "repeated", + Box::new(|sent| { + let again = sent[1].clone(); + sent.insert(2, again); + }), + "Frame 1 came where frame 2 was due", + ), + ( + "reordered", + Box::new(|sent| sent.swap(2, 3)), + "Frame 3 came where frame 2 was due", + ), + ( + "altered", + Box::new(|sent| *sent[2].last_mut().unwrap() ^= 1), + "Frame 2 does not open", + ), + ( + "forged", + Box::new(|sent| { + // Its length and number kept, the rest made up. + let mut forged = sent[2].clone(); + forged[12..].fill(7); + sent.insert(2, forged); + }), + "Frame 2 does not open", + ), + ( + "from another meeting", + Box::new(move |sent| sent.insert(2, elsewhere.clone())), + "Frame 2 does not open", + ), + ]; + for (name, tamper, expected) in cases { + let (mut ada, mut grace) = sealers(); + let mut sent = frames(&mut ada, 5); + tamper(&mut sent); + let stream = sent.concat(); + let mut arriving = &stream[..]; + for offset in 0..2 { + let (message, body) = grace.receive(&mut arriving).unwrap(); + assert_eq!(message, kind::PRESENCE); + assert_eq!(minicbor::decode::(&body).unwrap(), caret(offset)); + } + let error = grace.receive(&mut arriving).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::InvalidData, "{name}"); + assert!(error.to_string().starts_with(expected), "{name}: {error}"); + } +} + +/// A relay on this computer, with `config`'s limits: its URL. +fn relay(config: ::relay::server::Config) -> (String, SocketAddr) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + thread::spawn(move || ::relay::server::serve(listener, config)); + (format!("ws://{address}"), address) +} + +/// The status a relay at `address` answers a WebSocket to `path` with. +fn status(address: SocketAddr, path: &str) -> String { + let mut stream = TcpStream::connect(address).unwrap(); + write!( + stream, + "GET {path} HTTP/1.1\r\nHost: relay\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\ + Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n" + ) + .unwrap(); + let head = ::relay::ws::head(&mut stream).unwrap(); + head.lines().next().unwrap().to_owned() +} + +/// Two ends of one notebook that share no network meet in the relay's room, and see each +/// other leave. +#[test] +fn peers_meet_through_a_relay() { + let (url, _) = relay(Default::default()); + let room = Room::Notebook([3; 16]); + let ada = Live::start(hello("Ada"), &room, None, Some(&url), || {}).unwrap(); + ada.set_presence(caret(1)); + let grace = Live::start(hello("Grace"), &room, None, Some(&url), || {}).unwrap(); + until(&grace, |peers| { + peers.len() == 1 && peers[0].presence == Some(caret(1)) + }); + until(&ada, |peers| { + peers.len() == 1 && peers[0].hello.name == "Grace" + }); + ada.set_presence(caret(2)); + until(&grace, |peers| peers[0].presence == Some(caret(2))); + drop(ada); + until(&grace, <[Peer]>::is_empty); +} + +/// The end sharing a code asks the relay to number it; the other types the whole code. One +/// with the wrong words never meets, and the relay, told so by the end sharing, burns the +/// code once too many have tried. +#[test] +fn a_relay_numbers_a_code_and_burns_it_after_wrong_tries() { + let (url, address) = relay(::relay::server::Config { + burn_after: 2, + ..Default::default() + }); + let host = Live::start( + hello("Ada"), + &Room::Code("violet-otter".into()), + None, + Some(&url), + || {}, + ) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(10); + let code = loop { + if let Some(code) = host.code() { + break code; + } + assert!(Instant::now() < deadline, "no code"); + thread::sleep(Duration::from_millis(20)); + }; + let (number, words) = code.split_once('-').unwrap(); + assert_eq!(words, "violet-otter"); + let guest = Live::start( + hello("Grace"), + &Room::Code(code.clone()), + None, + Some(&url), + || {}, + ) + .unwrap(); + until(&host, |peers| peers.len() == 1); + until(&guest, |peers| peers.len() == 1); + + let path = format!("/v1/room/code-{number}"); + let wrong = Room::Code(format!("{number}-violet-ocelot")); + let mallory = Live::start(hello("Mallory"), &wrong, None, Some(&url), || {}).unwrap(); + // Mallory tries again a second later, and the second wrong try burns the code. Asking + // sooner would count as a try itself. + thread::sleep(Duration::from_secs(5)); + assert_eq!(status(address, &path), "HTTP/1.1 410 Gone"); + assert!(mallory.peers().is_empty()); + assert_eq!(host.peers().len(), 1, "Grace stays"); +} + +/// What a malicious relay does to one message on its way. +#[derive(Clone, Copy, Debug)] +enum Tamper { + Drop, + Repeat, + Reorder, + Alter, + Inject, +} + +/// A relay in the middle of Grace's connection that passes on what the real one at +/// `upstream` says, except the message to her numbered `at` among those from peers, which it +/// tampers with. Her next connection waits for `release`. +struct Malicious { + url: String, + tampered: mpsc::Receiver<()>, + rejoined: mpsc::Receiver<()>, + release: mpsc::Sender<()>, +} + +fn malicious(upstream: SocketAddr, tamper: Tamper, at: usize) -> Malicious { + use ::relay::ws::{self, Message}; + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let (tampered, told) = mpsc::channel(); + let (rejoined, heard) = mpsc::channel(); + let (release, released) = mpsc::channel(); + thread::spawn(move || { + for (index, client) in listener.incoming().enumerate() { + let mut client = client.unwrap(); + if index > 0 { + let _ = rejoined.send(()); + let _ = released.recv(); + } + let server = TcpStream::connect(upstream).unwrap(); + let (mut up, mut to_server) = + (client.try_clone().unwrap(), server.try_clone().unwrap()); + thread::spawn(move || { + let _ = io::copy(&mut up, &mut to_server); + let _ = to_server.shutdown(Shutdown::Both); + }); + let tampered = tampered.clone(); + thread::spawn(move || { + let mut reading = io::BufReader::new(server); + let head = ws::head(&mut reading).unwrap(); + client.write_all(head.as_bytes()).unwrap(); + let mut messages = ws::Reader::new(reading, 1 << 20, false); + let (mut count, mut held) = (0, None); + while let Ok(message) = messages.read() { + let frames: Vec> = match message { + Message::Binary(mut data) => { + count += 1; + let mut out = vec![]; + if index == 0 && count - 1 == at { + match tamper { + Tamper::Drop => {} + Tamper::Repeat => out = vec![data.clone(), data], + Tamper::Reorder => held = Some(data), + Tamper::Alter => { + *data.last_mut().unwrap() ^= 1; + out = vec![data]; + } + Tamper::Inject => { + // A slot, length and number, then made-up bytes. + let mut forged = data.clone(); + forged[16..].fill(7); + out = vec![forged, data]; + } + } + let _ = tampered.send(()); + } else { + out.push(data); + out.extend(held.take()); + } + out.into_iter() + .map(|data| ws::frame(ws::BINARY, &data, None)) + .collect() + } + Message::Text(text) => vec![ws::frame(ws::TEXT, text.as_bytes(), None)], + Message::Ping(payload) => vec![ws::frame(ws::PING, &payload, None)], + Message::Pong => vec![ws::frame(ws::PONG, &[], None)], + Message::Close => break, + }; + if frames.iter().any(|frame| client.write_all(frame).is_err()) { + break; + } + } + let _ = client.shutdown(Shutdown::Both); + }); + } + }); + Malicious { + url, + tampered: told, + rejoined: heard, + release, + } +} + +/// Starts `name`, recording each presence of its peer as applied, and `None` when the peer +/// goes. +fn recording(name: &str, room: &Room, relay: &str) -> (Live, Arc>>>) { + let heard = Arc::new(Mutex::new(Vec::new())); + let shared: Arc>> = Arc::default(); + let (recorded, watched) = (Arc::clone(&heard), Arc::clone(&shared)); + let live = Live::start(hello(name), room, None, Some(relay), move || { + let Some(shared) = watched.get().and_then(std::sync::Weak::upgrade) else { + return; + }; + let entry = match shared.state.lock().unwrap().peers.values().next() { + None => Some(None), + Some(link) => link.peer.presence.clone().map(Some), + }; + recorded.lock().unwrap().extend(entry); + }) + .unwrap(); + shared.set(Arc::downgrade(&live.shared)).ok().unwrap(); + (live, heard) +} + +/// Whatever a malicious relay does to a frame, the end it was for never applies it or +/// anything after it, drops the connection as broken, joins again and hears the newest +/// presence from scratch. +#[test] +fn a_malicious_relay_is_caught() { + let room = Room::Notebook([4; 16]); + for tamper in [ + Tamper::Drop, + Tamper::Repeat, + Tamper::Reorder, + Tamper::Alter, + Tamper::Inject, + ] { + let (url, address) = relay(Default::default()); + let ada = Live::start(hello("Ada"), &room, None, Some(&url), || {}).unwrap(); + ada.set_presence(caret(1)); + // Ada's opening, hello and first presence reach Grace; the next is tampered with. + let relay = malicious(address, tamper, 3); + let (grace, heard) = recording("Grace", &room, &relay.url); + until(&grace, |peers| { + peers + .first() + .is_some_and(|peer| peer.presence == Some(caret(1))) + }); + ada.set_presence(caret(2)); + relay + .tampered + .recv_timeout(Duration::from_secs(10)) + .unwrap(); + ada.set_presence(caret(3)); + relay + .rejoined + .recv_timeout(Duration::from_secs(10)) + .unwrap(); + let before: Vec<_> = heard.lock().unwrap().clone(); + let gone = before.iter().position(Option::is_none).unwrap(); + assert!( + !before[..gone].contains(&Some(caret(3))), + "{tamper:?}: applied after the tampering: {before:?}" + ); + assert!(grace.peers().is_empty(), "{tamper:?}"); + relay.release.send(()).unwrap(); + until(&grace, |peers| { + peers + .first() + .is_some_and(|peer| peer.presence == Some(caret(3))) + }); + } +} diff --git a/crates/notebook/src/live/wire.rs b/crates/notebook/src/live/wire.rs index ecc5af93e9ded61624ef4b94eb1d73bd6c5e6d5a..bfdb8d4737da092c29c6ad106d77e9dff7c8e09b 100644 --- a/crates/notebook/src/live/wire.rs +++ b/crates/notebook/src/live/wire.rs @@ -136,7 +136,9 @@ struct Open { pake: Vec, } -/// One direction's AEAD key and the count of frames sealed under it, which is each frame's nonce. +/// One direction's AEAD key and the count of frames sealed under it. Each frame carries its +/// number, which is also its nonce, so a frame lost, repeated, reordered or forged on the way +/// is caught before anything in it or after it is read. pub struct Sealer { cipher: Aes256Gcm, count: u64, @@ -152,13 +154,6 @@ impl Sealer { } } - fn nonce(&mut self) -> [u8; 12] { - let mut nonce = [0; 12]; - nonce[4..].copy_from_slice(&self.count.to_be_bytes()); - self.count += 1; - nonce - } - /// Writes message `kind` holding `body`. pub fn send( &mut self, @@ -168,28 +163,54 @@ impl Sealer { ) -> io::Result<()> { let mut clear = kind.to_be_bytes().to_vec(); minicbor::encode(body, &mut clear).map_err(io::Error::other)?; - let nonce = self.nonce(); + let number = self.count.to_be_bytes(); let sealed = self .cipher - .encrypt(&nonce.into(), clear.as_slice()) + .encrypt(&nonce(number).into(), clear.as_slice()) .map_err(|_| io::Error::other("A frame could not be sealed"))?; - write_block(to, &sealed) + self.count += 1; + write_block(to, &[&number[..], &sealed].concat()) } - /// Reads the next message: its kind and body. + /// Reads the next message: its kind and body. An error of kind `InvalidData` means the + /// stream broke, and nothing more on it can be trusted. pub fn receive(&mut self, from: &mut impl Read) -> io::Result<(u16, Vec)> { - let sealed = read_block(from)?; - let nonce = self.nonce(); + let block = read_block(from)?; + let (number, sealed) = block + .split_first_chunk::<8>() + .ok_or_else(|| invalid("A frame without its number"))?; + let due = self.count; + if u64::from_be_bytes(*number) != due { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "Frame {} came where frame {due} was due", + u64::from_be_bytes(*number) + ), + )); + } let mut clear = self .cipher - .decrypt(&nonce.into(), sealed.as_slice()) - .map_err(|_| invalid("A frame does not open under the agreed key"))?; + .decrypt(&nonce(*number).into(), sealed) + .map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("Frame {due} does not open under the agreed key"), + ) + })?; + self.count += 1; let body = clear.split_off(2.min(clear.len())); let kind = u16::from_be_bytes(clear.try_into().map_err(|_| invalid("An empty frame"))?); Ok((kind, body)) } } +fn nonce(number: [u8; 8]) -> [u8; 12] { + let mut nonce = [0; 12]; + nonce[4..].copy_from_slice(&number); + nonce +} + /// Which end of the connection this is. #[derive(Clone, Copy, PartialEq, Eq)] pub enum Side { diff --git a/crates/relay/Cargo.toml b/crates/relay/Cargo.toml new file mode 100644 index 0000000000000000000000000000000000000000..c9aefd33faa8225c66dfb90faa0755627628f257 --- /dev/null +++ b/crates/relay/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "relay" +version = "0.1.0" +edition = "2024" +publish = false + +# The relay Live Share meets through off the LAN (resources/live-share.md), and the WebSocket +# framing and notices its clients share with it. +[[bin]] +name = "snowbound-relay" +path = "src/main.rs" + +[dependencies] +base64 = { version = "0.23.1", default-features = false, features = ["std"] } +getrandom = "0.4.3" +sha1 = "0.11.0" diff --git a/crates/relay/README.md b/crates/relay/README.md new file mode 100644 index 0000000000000000000000000000000000000000..b751f0843f2b55f89a9644c172a2707973edb63b --- /dev/null +++ b/crates/relay/README.md @@ -0,0 +1,95 @@ +# snowbound-relay + +The relay Live Share meets through when two Snowbounds aren't on one network. Peers join a +room named by a tag (a hash of the notebook's secret, or a code's number), and the relay +passes their messages between them. Every message after the opening is sealed end to end +and numbered inside the seal, so the relay can't read, alter, drop or reorder one unseen; +it learns who talks to whom, when, and how much. `src/lib.rs` describes the protocol. + +One static Linux executable with no configuration file and nothing on disk. It speaks plain +HTTP and WebSocket; a proxy in front of it terminates TLS. + +## Build + +```sh +python3 tools/release_relay.py # target/relay/snowbound-relay-linux-{x86_64,aarch64} +``` + +It links with Rust's own lld against Rust's own musl, so it needs only `rustup`. The folder +it makes holds both executables, `SHA256SUMS`, this README and `snowbound-relay.service`. + +## Deploy + +```sh +scp target/relay/snowbound-relay-linux-x86_64 vps:/tmp/snowbound-relay +scp target/relay/snowbound-relay.service vps:/tmp/ +ssh vps +sudo install -m 755 /tmp/snowbound-relay /usr/local/bin/snowbound-relay +sudo install -m 644 /tmp/snowbound-relay.service /etc/systemd/system/ +sudo systemctl daemon-reload +sudo systemctl enable --now snowbound-relay +curl -s http://127.0.0.1:7650/health # {"rooms":0,"peers":0,"connections":1,"seconds":3} +``` + +Then point a name at the server and put a TLS proxy in front. Caddy fetches its own +certificate and passes WebSocket upgrades through as they are: + +```text +live.example.net { + reverse_proxy 127.0.0.1:7650 +} +``` + +nginx, with a certificate from certbot: + +```nginx +server { + listen 443 ssl; + server_name live.example.net; + ssl_certificate /etc/letsencrypt/live/live.example.net/fullchain.pem; + ssl_certificate_key /etc/letsencrypt/live/live.example.net/privkey.pem; + location / { + proxy_pass http://127.0.0.1:7650; + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "upgrade"; + # Replaced, not appended to, so a client can't name its own address. + proxy_set_header X-Forwarded-For $remote_addr; + proxy_read_timeout 1h; + } +} +``` + +The app then uses `wss://live.example.net` (for now, `SNOWBOUND_LIVE_RELAY=wss://live.example.net` +with a build that has the `live` feature). + +`--trust-forwarded true`, as the unit sets it, counts each peer by the last +`X-Forwarded-For` entry, the one the proxy added. Without a proxy, leave it off: a client +could otherwise claim any address. Listening on a public address without TLS works but lets +anyone on the path see room tags and nameplates. + +## Options + +`snowbound-relay --help` lists every option and its default. Each can also be set in the +unit's environment as `SNOWBOUND_RELAY_