From 42549de7a888385d3d586ea863ec248122fe2564 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Mon, 12 Oct 2026 11:27:28 -0700 Subject: [PATCH] fix(live): harden section admission and relay connections - Reject invalid sections before retaining a writer. - Try resolved relay addresses within one TCP deadline. - Bracket IPv6 proxy CONNECT authorities. Fixes #99 Fixes #139 Fixes #140 Assisted-by: gpt-6.1-sol --- crates/notebook/src/live/share/batch.rs | 43 ++++-- crates/notebook/src/live/transport.rs | 60 ++++++-- crates/notebook/src/live/transport_tests.rs | 152 +++++++++++++++++++- 3 files changed, 226 insertions(+), 29 deletions(-) diff --git a/crates/notebook/src/live/share/batch.rs b/crates/notebook/src/live/share/batch.rs index e2c77ee6a98f2766522112ac3916fd47a120cec3..9830ffaa2c6d86e2145bcfe8bee15222ecb4e370 100644 --- a/crates/notebook/src/live/share/batch.rs +++ b/crates/notebook/src/live/share/batch.rs @@ -58,7 +58,9 @@ impl Served { pub(super) fn batch(self: &Arc, peer: &[u8; 16], request: Request) -> Result { let path = request.path.clone(); if !self.writers.lock().unwrap().contains_key(&path) { - self.image(&path)?; + let image = self.image(&path)?; + let arena = Arena::default(); + Section::open(&arena, (*image).clone())?; } let mut writers = self.writers.lock().unwrap(); let writer = writers.entry(path.clone()).or_insert_with(|| { @@ -318,7 +320,7 @@ mod tests { #[cfg(not(target_arch = "wasm32"))] #[test] - fn missing_sections_do_not_retain_writers_and_can_be_created_later() { + fn invalid_sections_do_not_retain_writers_and_can_be_repaired() { let directory = tempfile::tempdir().unwrap(); let folder = directory.path().join("notebook"); std::fs::create_dir(&folder).unwrap(); @@ -347,19 +349,32 @@ mod tests { held: Vec::new(), }, ); - for at in 0..128 { - let reply = served.handle( - &peer, - kind::EDITS, - &minicbor::to_vec(Request { - path: format!("Missing{at}.one"), - ..Request::default() - }) - .unwrap(), - ); - assert!(reply.failure.is_some()); + let toc = onestore::create_table_of_contents("Invalid.onetoc2", &[]).unwrap(); + { + let store = Store::parse(&toc).unwrap(); + RevisionIndex::parse(&store).unwrap(); + } + let arena = Arena::default(); + assert!(Section::open(&arena, toc.clone()).is_err()); + for image in [None, Some(vec![0; 1024]), Some(toc)] { + for at in 0..128 { + let path = format!("Missing{at}.one"); + if let Some(image) = &image { + std::fs::write(folder.join(&path), image).unwrap(); + } + let reply = served.handle( + &peer, + kind::EDITS, + &minicbor::to_vec(Request { + path, + ..Request::default() + }) + .unwrap(), + ); + assert!(reply.failure.is_some()); + } + assert!(served.writers.lock().unwrap().is_empty()); } - assert!(served.writers.lock().unwrap().is_empty()); let path = "Missing0.one"; let image = onestore::create_section(path, "Original text", "Fixture").unwrap(); diff --git a/crates/notebook/src/live/transport.rs b/crates/notebook/src/live/transport.rs index 33fc6e21debadd3c452d70c2c54226da91e97e44..32c3ff50c87eeca02f15207a5a6e71ee2021e1ac 100644 --- a/crates/notebook/src/live/transport.rs +++ b/crates/notebook/src/live/transport.rs @@ -11,13 +11,13 @@ use base64::Engine; use rustls::{ClientConfig, ClientConnection, RootCertStore, pki_types::ServerName}; use std::{ io::{self, BufReader, Read, Write}, - net::{Shutdown, TcpStream, ToSocketAddrs}, + net::{Shutdown, SocketAddr, TcpStream, ToSocketAddrs}, sync::{ Arc, Condvar, Mutex, OnceLock, atomic::{AtomicBool, Ordering}, }, thread, - time::Duration, + time::{Duration, Instant}, }; const CONNECT: Duration = Duration::from_secs(10); @@ -128,6 +128,38 @@ struct Stream { tcp: TcpStream, } +fn connect_resolved( + targets: impl Iterator, + deadline: Instant, + mut connect: impl FnMut(&SocketAddr, Duration) -> io::Result, +) -> Result { + let targets: Vec<_> = targets.collect(); + let mut failure = (Trouble::Dns, io::ErrorKind::NotFound.into()); + for (at, target) in targets.iter().enumerate() { + let remaining = deadline.saturating_duration_since(Instant::now()); + let attempts = u32::try_from(targets.len() - at).unwrap_or(u32::MAX); + let timeout = remaining / attempts; + if timeout.is_zero() { + return Err((Trouble::TimedOut, io::ErrorKind::TimedOut.into())); + } + let connected = connect(target, timeout); + if Instant::now() >= deadline { + return Err((Trouble::TimedOut, io::ErrorKind::TimedOut.into())); + } + match connected { + Ok(tcp) => return Ok(tcp), + Err(error) => { + let trouble = match error.kind() { + io::ErrorKind::TimedOut => Trouble::TimedOut, + _ => Trouble::Unreachable, + }; + failure = (trouble, error); + } + } + } + Err(failure) +} + fn open(address: &Address) -> Result { let proxy = proxy::for_host(&address.host, address.tls); let (host, port) = match &proxy { @@ -138,18 +170,15 @@ fn open(address: &Address) -> Result { Some(_) => Trouble::ProxyUnreachable, None => trouble, }; - let target = (host, port) + let targets = (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) - })?; + .map_err(|error| Failure::Trouble(reached(Trouble::Dns), error))?; + let tcp = connect_resolved( + targets, + Instant::now() + CONNECT, + TcpStream::connect_timeout, + ) + .map_err(|(trouble, error)| Failure::Trouble(reached(trouble), error))?; tcp.set_nodelay(true)?; tcp.set_read_timeout(Some(QUIET))?; tcp.set_write_timeout(Some(QUIET))?; @@ -196,7 +225,10 @@ fn open(address: &Address) -> Result { /// 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 target = match address.host.contains(':') { + true => format!("[{}]:{}", address.host, address.port), + false => 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"); diff --git a/crates/notebook/src/live/transport_tests.rs b/crates/notebook/src/live/transport_tests.rs index c005cb9f6f4f3ed588d6ff67b5c47e1dd5abdd7b..4d57b13b235d09ae58cad75fb36ba8191850999e 100644 --- a/crates/notebook/src/live/transport_tests.rs +++ b/crates/notebook/src/live/transport_tests.rs @@ -1,5 +1,5 @@ use super::*; -use std::{net::TcpListener, thread::JoinHandle, time::Instant}; +use std::{net::TcpListener, thread::JoinHandle}; fn request(stream: &mut TcpStream) -> io::Result<(String, Vec)> { let head = ws::head(stream)?; @@ -203,3 +203,153 @@ fn repeated_get_failures_return_without_a_reconnect_loop() { assert_eq!(attempts, 1); assert!(elapsed < Duration::from_secs(1)); } + +#[test] +fn later_resolved_addresses_can_connect_after_the_first_refuses() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let at = listener.local_addr().unwrap(); + let refused = SocketAddr::from(([127, 0, 0, 1], 0)); + let connected = connect_resolved( + [refused, at].into_iter(), + Instant::now() + Duration::from_secs(2), + TcpStream::connect_timeout, + ); + let mut client = match connected { + Ok(client) => client, + Err((trouble, error)) => panic!("later address failed: {trouble:?}: {error}"), + }; + assert_eq!(client.peer_addr().unwrap(), at); + client + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + client + .set_write_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut server = accept(&listener).expect("the second address never connected"); + server + .set_write_timeout(Some(Duration::from_secs(2))) + .unwrap(); + client.write_all(b"request").unwrap(); + let mut request = [0; 7]; + server.read_exact(&mut request).unwrap(); + assert_eq!(&request, b"request"); + server.write_all(b"answer").unwrap(); + let mut answer = [0; 6]; + client.read_exact(&mut answer).unwrap(); + assert_eq!(&answer, b"answer"); +} + +#[test] +fn resolved_connection_failures_keep_the_last_error_and_share_the_budget() { + let targets = [ + SocketAddr::from(([127, 0, 0, 1], 1)), + SocketAddr::from(([127, 0, 0, 1], 2)), + SocketAddr::from(([127, 0, 0, 1], 3)), + ]; + let budget = Duration::from_secs(10); + for (last, expected) in [ + (io::ErrorKind::AddrNotAvailable, Trouble::Unreachable), + (io::ErrorKind::TimedOut, Trouble::TimedOut), + ] { + let mut attempts = 0; + let failed = connect_resolved( + targets.into_iter(), + Instant::now() + budget, + |target, timeout| { + assert_eq!(*target, targets[attempts]); + assert!(!timeout.is_zero()); + assert!(timeout <= budget / (3 - attempts as u32)); + attempts += 1; + Err(io::Error::new( + if attempts == 3 { + last + } else { + io::ErrorKind::ConnectionRefused + }, + format!("attempt {attempts}"), + )) + }, + ); + let (trouble, error) = failed.unwrap_err(); + assert_eq!(attempts, 3); + assert_eq!(trouble, expected); + assert_eq!(error.kind(), last); + assert_eq!(error.to_string(), "attempt 3"); + } +} + +#[test] +fn empty_resolution_and_expired_deadlines_do_not_attempt_a_connection() { + let empty = connect_resolved(std::iter::empty(), Instant::now() + CONNECT, |_, _| { + panic!("empty resolution tried to connect") + }); + let (trouble, error) = empty.unwrap_err(); + assert_eq!(trouble, Trouble::Dns); + assert_eq!(error.kind(), io::ErrorKind::NotFound); + let expired = connect_resolved( + [SocketAddr::from(([127, 0, 0, 1], 0))].into_iter(), + Instant::now(), + |_, _| panic!("an expired deadline tried to connect"), + ); + let (trouble, error) = expired.unwrap_err(); + assert_eq!(trouble, Trouble::TimedOut); + assert_eq!(error.kind(), io::ErrorKind::TimedOut); +} + +#[test] +fn proxy_tunnels_bracket_ipv6_authorities_and_remain_usable() { + for (url, target) in [ + ("ws://[::1]:123", "[::1]:123"), + ("wss://[2001:db8::1]:8443", "[2001:db8::1]:8443"), + ("ws://[::1]", "[::1]:80"), + ("wss://[::1]", "[::1]:443"), + ("ws://127.0.0.1:123", "127.0.0.1:123"), + ("ws://127.0.0.1", "127.0.0.1:80"), + ("wss://relay.example", "relay.example:443"), + ] { + let address = parse(url).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let at = listener.local_addr().unwrap(); + let proxy = Proxy { + host: "127.0.0.1".into(), + port: at.port(), + credentials: None, + }; + let mut client = TcpStream::connect_timeout(&at, Duration::from_secs(2)).unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + client + .set_write_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let serving = thread::spawn(move || { + let mut stream = accept(&listener).expect("the proxy connection never arrived"); + stream + .set_write_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let (head, body) = request(&mut stream).unwrap(); + assert_eq!( + head, + format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n") + ); + assert!(body.is_empty()); + answer(&mut stream, 200, b"").unwrap(); + let (head, body) = request(&mut stream).unwrap(); + assert_eq!(head, "GET /v1/room/test?poll=1 HTTP/1.1\r\n\r\n"); + assert!(body.is_empty()); + answer(&mut stream, 200, b"session test").unwrap(); + }); + let exchange = if tunnel(&client, &proxy, &address).is_ok() { + (|| -> io::Result<_> { + client.write_all(b"GET /v1/room/test?poll=1 HTTP/1.1\r\n\r\n")?; + request(&mut client) + })() + } else { + Err(io::Error::other("the proxy refused the tunnel")) + }; + serving.join().unwrap(); + let (head, body) = exchange.unwrap(); + assert!(head.starts_with("HTTP/1.1 200 ")); + assert_eq!(body, b"session test"); + } +} -- 2.54.0