authorclover caruso <git@paperclover.net> 2026-10-12 11:27:28-07:00
committerclover caruso <git@paperclover.net> 2026-10-12 11:29:25-07:00
log42549de7a888385d3d586ea863ec248122fe2564
tree5f3cb563bc5e4ba460ee943fd1d04f535d9ee00d
parent464e096c972a1146718f05b4ede4dc49588eeae1
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

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

3 files changed, 226 insertions(+), 29 deletions(-)

crates/notebook/src/live/share/batch.rs+29-14
...@@ -58,7 +58,9 @@ impl Served {...@@ -58,7 +58,9 @@ impl Served {
58 pub(super) fn batch(self: &Arc<Self>, peer: &[u8; 16], request: Request) -> Result<Reply> {58 pub(super) fn batch(self: &Arc<Self>, peer: &[u8; 16], request: Request) -> Result<Reply> {
59 let path = request.path.clone();59 let path = request.path.clone();
60 if !self.writers.lock().unwrap().contains_key(&path) {60 if !self.writers.lock().unwrap().contains_key(&path) {
61 self.image(&path)?;61 let image = self.image(&path)?;
62 let arena = Arena::default();
63 Section::open(&arena, (*image).clone())?;
62 }64 }
63 let mut writers = self.writers.lock().unwrap();65 let mut writers = self.writers.lock().unwrap();
64 let writer = writers.entry(path.clone()).or_insert_with(|| {66 let writer = writers.entry(path.clone()).or_insert_with(|| {
...@@ -318,7 +320,7 @@ mod tests {...@@ -318,7 +320,7 @@ mod tests {
318320
319 #[cfg(not(target_arch = "wasm32"))]321 #[cfg(not(target_arch = "wasm32"))]
320 #[test]322 #[test]
321 fn missing_sections_do_not_retain_writers_and_can_be_created_later() {323 fn invalid_sections_do_not_retain_writers_and_can_be_repaired() {
322 let directory = tempfile::tempdir().unwrap();324 let directory = tempfile::tempdir().unwrap();
323 let folder = directory.path().join("notebook");325 let folder = directory.path().join("notebook");
324 std::fs::create_dir(&folder).unwrap();326 std::fs::create_dir(&folder).unwrap();
...@@ -347,19 +349,32 @@ mod tests {...@@ -347,19 +349,32 @@ mod tests {
347 held: Vec::new(),349 held: Vec::new(),
348 },350 },
349 );351 );
350 for at in 0..128 {352 let toc = onestore::create_table_of_contents("Invalid.onetoc2", &[]).unwrap();
351 let reply = served.handle(353 {
352 &peer,354 let store = Store::parse(&toc).unwrap();
353 kind::EDITS,355 RevisionIndex::parse(&store).unwrap();
354 &minicbor::to_vec(Request {356 }
355 path: format!("Missing{at}.one"),357 let arena = Arena::default();
356 ..Request::default()358 assert!(Section::open(&arena, toc.clone()).is_err());
357 })359 for image in [None, Some(vec![0; 1024]), Some(toc)] {
358 .unwrap(),360 for at in 0..128 {
359 );361 let path = format!("Missing{at}.one");
360 assert!(reply.failure.is_some());362 if let Some(image) = &image {
363 std::fs::write(folder.join(&path), image).unwrap();
364 }
365 let reply = served.handle(
366 &peer,
367 kind::EDITS,
368 &minicbor::to_vec(Request {
369 path,
370 ..Request::default()
371 })
372 .unwrap(),
373 );
374 assert!(reply.failure.is_some());
375 }
376 assert!(served.writers.lock().unwrap().is_empty());
361 }377 }
362 assert!(served.writers.lock().unwrap().is_empty());
363378
364 let path = "Missing0.one";379 let path = "Missing0.one";
365 let image = onestore::create_section(path, "Original text", "Fixture").unwrap();380 let image = onestore::create_section(path, "Original text", "Fixture").unwrap();
crates/notebook/src/live/transport.rs+46-14
...@@ -11,13 +11,13 @@ use base64::Engine;...@@ -11,13 +11,13 @@ use base64::Engine;
11use rustls::{ClientConfig, ClientConnection, RootCertStore, pki_types::ServerName};11use rustls::{ClientConfig, ClientConnection, RootCertStore, pki_types::ServerName};
12use std::{12use std::{
13 io::{self, BufReader, Read, Write},13 io::{self, BufReader, Read, Write},
14 net::{Shutdown, TcpStream, ToSocketAddrs},14 net::{Shutdown, SocketAddr, TcpStream, ToSocketAddrs},
15 sync::{15 sync::{
16 Arc, Condvar, Mutex, OnceLock,16 Arc, Condvar, Mutex, OnceLock,
17 atomic::{AtomicBool, Ordering},17 atomic::{AtomicBool, Ordering},
18 },18 },
19 thread,19 thread,
20 time::Duration,20 time::{Duration, Instant},
21};21};
2222
23const CONNECT: Duration = Duration::from_secs(10);23const CONNECT: Duration = Duration::from_secs(10);
...@@ -128,6 +128,38 @@ struct Stream {...@@ -128,6 +128,38 @@ struct Stream {
128 tcp: TcpStream,128 tcp: TcpStream,
129}129}
130130
131fn connect_resolved(
132 targets: impl Iterator<Item = SocketAddr>,
133 deadline: Instant,
134 mut connect: impl FnMut(&SocketAddr, Duration) -> io::Result<TcpStream>,
135) -> Result<TcpStream, (Trouble, io::Error)> {
136 let targets: Vec<_> = targets.collect();
137 let mut failure = (Trouble::Dns, io::ErrorKind::NotFound.into());
138 for (at, target) in targets.iter().enumerate() {
139 let remaining = deadline.saturating_duration_since(Instant::now());
140 let attempts = u32::try_from(targets.len() - at).unwrap_or(u32::MAX);
141 let timeout = remaining / attempts;
142 if timeout.is_zero() {
143 return Err((Trouble::TimedOut, io::ErrorKind::TimedOut.into()));
144 }
145 let connected = connect(target, timeout);
146 if Instant::now() >= deadline {
147 return Err((Trouble::TimedOut, io::ErrorKind::TimedOut.into()));
148 }
149 match connected {
150 Ok(tcp) => return Ok(tcp),
151 Err(error) => {
152 let trouble = match error.kind() {
153 io::ErrorKind::TimedOut => Trouble::TimedOut,
154 _ => Trouble::Unreachable,
155 };
156 failure = (trouble, error);
157 }
158 }
159 }
160 Err(failure)
161}
162
131fn open(address: &Address) -> Result<Stream, Failure> {163fn open(address: &Address) -> Result<Stream, Failure> {
132 let proxy = proxy::for_host(&address.host, address.tls);164 let proxy = proxy::for_host(&address.host, address.tls);
133 let (host, port) = match &proxy {165 let (host, port) = match &proxy {
...@@ -138,18 +170,15 @@ fn open(address: &Address) -> Result<Stream, Failure> {...@@ -138,18 +170,15 @@ fn open(address: &Address) -> Result<Stream, Failure> {
138 Some(_) => Trouble::ProxyUnreachable,170 Some(_) => Trouble::ProxyUnreachable,
139 None => trouble,171 None => trouble,
140 };172 };
141 let target = (host, port)173 let targets = (host, port)
142 .to_socket_addrs()174 .to_socket_addrs()
143 .map_err(|error| Failure::Trouble(reached(Trouble::Dns), error))?175 .map_err(|error| Failure::Trouble(reached(Trouble::Dns), error))?;
144 .next()176 let tcp = connect_resolved(
145 .ok_or_else(|| Failure::Trouble(reached(Trouble::Dns), io::ErrorKind::NotFound.into()))?;177 targets,
146 let tcp = TcpStream::connect_timeout(&target, CONNECT).map_err(|error| {178 Instant::now() + CONNECT,
147 let trouble = match error.kind() {179 TcpStream::connect_timeout,
148 io::ErrorKind::TimedOut => Trouble::TimedOut,180 )
149 _ => Trouble::Unreachable,181 .map_err(|(trouble, error)| Failure::Trouble(reached(trouble), error))?;
150 };
151 Failure::Trouble(reached(trouble), error)
152 })?;
153 tcp.set_nodelay(true)?;182 tcp.set_nodelay(true)?;
154 tcp.set_read_timeout(Some(QUIET))?;183 tcp.set_read_timeout(Some(QUIET))?;
155 tcp.set_write_timeout(Some(QUIET))?;184 tcp.set_write_timeout(Some(QUIET))?;
...@@ -196,7 +225,10 @@ fn open(address: &Address) -> Result<Stream, Failure> {...@@ -196,7 +225,10 @@ fn open(address: &Address) -> Result<Stream, Failure> {
196225
197/// Asks `proxy` on `tcp` to connect through to the relay.226/// Asks `proxy` on `tcp` to connect through to the relay.
198fn tunnel(tcp: &TcpStream, proxy: &Proxy, address: &Address) -> Result<(), Failure> {227fn tunnel(tcp: &TcpStream, proxy: &Proxy, address: &Address) -> Result<(), Failure> {
199 let target = format!("{}:{}", address.host, address.port);228 let target = match address.host.contains(':') {
229 true => format!("[{}]:{}", address.host, address.port),
230 false => format!("{}:{}", address.host, address.port),
231 };
200 let mut request = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n");232 let mut request = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n");
201 if let Some(authorization) = proxy.authorization() {233 if let Some(authorization) = proxy.authorization() {
202 request += &format!("Proxy-Authorization: {authorization}\r\n");234 request += &format!("Proxy-Authorization: {authorization}\r\n");
crates/notebook/src/live/transport_tests.rs+151-1
...@@ -1,5 +1,5 @@...@@ -1,5 +1,5 @@
1use super::*;1use super::*;
2use std::{net::TcpListener, thread::JoinHandle, time::Instant};2use std::{net::TcpListener, thread::JoinHandle};
33
4fn request(stream: &mut TcpStream) -> io::Result<(String, Vec<u8>)> {4fn request(stream: &mut TcpStream) -> io::Result<(String, Vec<u8>)> {
5 let head = ws::head(stream)?;5 let head = ws::head(stream)?;
...@@ -203,3 +203,153 @@ fn repeated_get_failures_return_without_a_reconnect_loop() {...@@ -203,3 +203,153 @@ fn repeated_get_failures_return_without_a_reconnect_loop() {
203 assert_eq!(attempts, 1);203 assert_eq!(attempts, 1);
204 assert!(elapsed < Duration::from_secs(1));204 assert!(elapsed < Duration::from_secs(1));
205}205}
206
207#[test]
208fn later_resolved_addresses_can_connect_after_the_first_refuses() {
209 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
210 let at = listener.local_addr().unwrap();
211 let refused = SocketAddr::from(([127, 0, 0, 1], 0));
212 let connected = connect_resolved(
213 [refused, at].into_iter(),
214 Instant::now() + Duration::from_secs(2),
215 TcpStream::connect_timeout,
216 );
217 let mut client = match connected {
218 Ok(client) => client,
219 Err((trouble, error)) => panic!("later address failed: {trouble:?}: {error}"),
220 };
221 assert_eq!(client.peer_addr().unwrap(), at);
222 client
223 .set_read_timeout(Some(Duration::from_secs(2)))
224 .unwrap();
225 client
226 .set_write_timeout(Some(Duration::from_secs(2)))
227 .unwrap();
228 let mut server = accept(&listener).expect("the second address never connected");
229 server
230 .set_write_timeout(Some(Duration::from_secs(2)))
231 .unwrap();
232 client.write_all(b"request").unwrap();
233 let mut request = [0; 7];
234 server.read_exact(&mut request).unwrap();
235 assert_eq!(&request, b"request");
236 server.write_all(b"answer").unwrap();
237 let mut answer = [0; 6];
238 client.read_exact(&mut answer).unwrap();
239 assert_eq!(&answer, b"answer");
240}
241
242#[test]
243fn resolved_connection_failures_keep_the_last_error_and_share_the_budget() {
244 let targets = [
245 SocketAddr::from(([127, 0, 0, 1], 1)),
246 SocketAddr::from(([127, 0, 0, 1], 2)),
247 SocketAddr::from(([127, 0, 0, 1], 3)),
248 ];
249 let budget = Duration::from_secs(10);
250 for (last, expected) in [
251 (io::ErrorKind::AddrNotAvailable, Trouble::Unreachable),
252 (io::ErrorKind::TimedOut, Trouble::TimedOut),
253 ] {
254 let mut attempts = 0;
255 let failed = connect_resolved(
256 targets.into_iter(),
257 Instant::now() + budget,
258 |target, timeout| {
259 assert_eq!(*target, targets[attempts]);
260 assert!(!timeout.is_zero());
261 assert!(timeout <= budget / (3 - attempts as u32));
262 attempts += 1;
263 Err(io::Error::new(
264 if attempts == 3 {
265 last
266 } else {
267 io::ErrorKind::ConnectionRefused
268 },
269 format!("attempt {attempts}"),
270 ))
271 },
272 );
273 let (trouble, error) = failed.unwrap_err();
274 assert_eq!(attempts, 3);
275 assert_eq!(trouble, expected);
276 assert_eq!(error.kind(), last);
277 assert_eq!(error.to_string(), "attempt 3");
278 }
279}
280
281#[test]
282fn empty_resolution_and_expired_deadlines_do_not_attempt_a_connection() {
283 let empty = connect_resolved(std::iter::empty(), Instant::now() + CONNECT, |_, _| {
284 panic!("empty resolution tried to connect")
285 });
286 let (trouble, error) = empty.unwrap_err();
287 assert_eq!(trouble, Trouble::Dns);
288 assert_eq!(error.kind(), io::ErrorKind::NotFound);
289 let expired = connect_resolved(
290 [SocketAddr::from(([127, 0, 0, 1], 0))].into_iter(),
291 Instant::now(),
292 |_, _| panic!("an expired deadline tried to connect"),
293 );
294 let (trouble, error) = expired.unwrap_err();
295 assert_eq!(trouble, Trouble::TimedOut);
296 assert_eq!(error.kind(), io::ErrorKind::TimedOut);
297}
298
299#[test]
300fn proxy_tunnels_bracket_ipv6_authorities_and_remain_usable() {
301 for (url, target) in [
302 ("ws://[::1]:123", "[::1]:123"),
303 ("wss://[2001:db8::1]:8443", "[2001:db8::1]:8443"),
304 ("ws://[::1]", "[::1]:80"),
305 ("wss://[::1]", "[::1]:443"),
306 ("ws://127.0.0.1:123", "127.0.0.1:123"),
307 ("ws://127.0.0.1", "127.0.0.1:80"),
308 ("wss://relay.example", "relay.example:443"),
309 ] {
310 let address = parse(url).unwrap();
311 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
312 let at = listener.local_addr().unwrap();
313 let proxy = Proxy {
314 host: "127.0.0.1".into(),
315 port: at.port(),
316 credentials: None,
317 };
318 let mut client = TcpStream::connect_timeout(&at, Duration::from_secs(2)).unwrap();
319 client
320 .set_read_timeout(Some(Duration::from_secs(2)))
321 .unwrap();
322 client
323 .set_write_timeout(Some(Duration::from_secs(2)))
324 .unwrap();
325 let serving = thread::spawn(move || {
326 let mut stream = accept(&listener).expect("the proxy connection never arrived");
327 stream
328 .set_write_timeout(Some(Duration::from_secs(2)))
329 .unwrap();
330 let (head, body) = request(&mut stream).unwrap();
331 assert_eq!(
332 head,
333 format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n\r\n")
334 );
335 assert!(body.is_empty());
336 answer(&mut stream, 200, b"").unwrap();
337 let (head, body) = request(&mut stream).unwrap();
338 assert_eq!(head, "GET /v1/room/test?poll=1 HTTP/1.1\r\n\r\n");
339 assert!(body.is_empty());
340 answer(&mut stream, 200, b"session test").unwrap();
341 });
342 let exchange = if tunnel(&client, &proxy, &address).is_ok() {
343 (|| -> io::Result<_> {
344 client.write_all(b"GET /v1/room/test?poll=1 HTTP/1.1\r\n\r\n")?;
345 request(&mut client)
346 })()
347 } else {
348 Err(io::Error::other("the proxy refused the tunnel"))
349 };
350 serving.join().unwrap();
351 let (head, body) = exchange.unwrap();
352 assert!(head.starts_with("HTTP/1.1 200 "));
353 assert_eq!(body, b"session test");
354 }
355}