1use super::*;
2use std::time::Instant;
3
4/// The code for room `number` and `secret`.
5fn code(number: u32, secret: &str) -> String {
6 super::code::format(number, secret).unwrap()
7}
8
9fn hello(name: &str) -> Hello {
10 Hello::new(name.into(), Some(vec![1, 2, 3])).unwrap()
11}
12
13/// Waits until `done` holds of `live`'s peers.
14fn until(live: &Live, done: impl Fn(&[Peer]) -> bool) -> Vec<Peer> {
15 let deadline = Instant::now() + Duration::from_secs(10);
16 loop {
17 let peers = live.peers();
18 if done(&peers) {
19 return peers;
20 }
21 assert!(Instant::now() < deadline, "peers stayed {peers:?}");
22 thread::sleep(Duration::from_millis(20));
23 }
24}
25
26fn caret(offset: u32) -> Presence {
27 let spot = Spot {
28 text: Guid {
29 guid: [7; 16],
30 n: 3,
31 },
32 offset,
33 };
34 Presence {
35 section: Some([5; 16]),
36 page: Some(Guid {
37 guid: [6; 16],
38 n: 1,
39 }),
40 caret: Some(Caret {
41 anchor: spot,
42 focus: spot,
43 }),
44 }
45}
46
47/// Two ends of one code meet, greet each other by name and picture, hear each other's newest
48/// caret, and see the other leave.
49#[test]
50fn peers_meet_and_follow_presence() {
51 let room = Room::join(&code(7, "ABCDEF"), "");
52 let ada = Live::start(hello("Ada"), &room, None, None, |_| {}).unwrap();
53 let grace = Live::start(hello("Grace"), &room, None, None, |_| {}).unwrap();
54 ada.set_presence(caret(1));
55 ada.connect(grace.address());
56 let seen = until(&grace, |peers| {
57 peers.iter().any(|peer| peer.presence.is_some())
58 });
59 assert_eq!(seen[0].hello.name, "Ada");
60 assert_eq!(seen[0].hello.picture.as_deref(), Some(&[1, 2, 3][..]));
61 assert_eq!(seen[0].presence, Some(caret(1)));
62 until(&ada, |peers| {
63 peers.len() == 1 && peers[0].hello.name == "Grace"
64 });
65 for offset in 2..20 {
66 ada.set_presence(caret(offset));
67 }
68 until(&grace, |peers| peers[0].presence == Some(caret(19)));
69 drop(ada);
70 until(&grace, <[Peer]>::is_empty);
71}
72
73fn lan_accept(listener: &TcpListener) -> TcpStream {
74 let deadline = Instant::now() + Duration::from_secs(10);
75 loop {
76 match listener.accept() {
77 Ok((stream, _)) => {
78 stream.set_nonblocking(false).unwrap();
79 stream.set_read_timeout(Some(OPENING)).unwrap();
80 stream.set_write_timeout(Some(OPENING)).unwrap();
81 return stream;
82 }
83 Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
84 assert!(Instant::now() < deadline, "LAN did not reconnect");
85 thread::sleep(Duration::from_millis(10));
86 }
87 Err(error) => panic!("{error}"),
88 }
89 }
90}
91
92fn lan_opening(stream: &mut TcpStream) -> Vec<u8> {
93 let mut length = [0; 4];
94 stream.read_exact(&mut length).unwrap();
95 let length = u32::from_be_bytes(length) as usize;
96 assert!(length < 1 << 20);
97 let mut opening = vec![0; length];
98 stream.read_exact(&mut opening).unwrap();
99 opening
100}
101
102#[test]
103fn lan_retries_opening_io_and_reconnect_refusal_with_fresh_keys() {
104 let room = Room::Notebook([71; 16]);
105 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
106 let address = listener.local_addr().unwrap();
107 listener.set_nonblocking(true).unwrap();
108 let (fault, faults) = mpsc::channel();
109 let (resume, resumed) = mpsc::channel();
110 let server_room = room.clone();
111 let server = thread::spawn(move || {
112 let mut abandoned = lan_accept(&listener);
113 let first = lan_opening(&mut abandoned);
114 drop(abandoned);
115 let host = hello("Host");
116 let mut openings = vec![first];
117 let mut delivered = Vec::new();
118 let mut listener = Some(listener);
119 for edit in [41_u64, 42] {
120 let mut stream = lan_accept(listener.as_ref().unwrap());
121 let theirs = lan_opening(&mut stream);
122 assert!(openings.iter().all(|old| *old != theirs), "LAN reused keys");
123 openings.push(theirs.clone());
124 let (opening, ours) = wire::Opening::new(
125 Side::Responder,
126 &server_room.tag().unwrap(),
127 &server_room.secret(),
128 )
129 .unwrap();
130 stream
131 .write_all(&(ours.len() as u32).to_be_bytes())
132 .unwrap();
133 stream.write_all(&ours).unwrap();
134 let (mut send, mut receive) = opening.finish(&theirs).unwrap();
135 send.send(&mut stream, kind::HELLO, &host).unwrap();
136 assert_eq!(receive.receive(&mut stream).unwrap().0, kind::HELLO);
137 send.send(&mut stream, kind::EDITS, &edit).unwrap();
138 loop {
139 let (kind, body) = receive.receive(&mut stream).unwrap();
140 if kind == kind::EDITS {
141 delivered.push(minicbor::decode::<u64>(&body).unwrap());
142 break;
143 }
144 }
145 assert_eq!(
146 listener.as_ref().unwrap().accept().unwrap_err().kind(),
147 io::ErrorKind::WouldBlock,
148 "duplicate LAN dial workers connected",
149 );
150 if edit == 41 {
151 drop(listener.take());
152 stream.shutdown(Shutdown::Both).unwrap();
153 fault.send(()).unwrap();
154 resumed.recv_timeout(Duration::from_secs(10)).unwrap();
155 let reopened = TcpListener::bind(address).unwrap();
156 reopened.set_nonblocking(true).unwrap();
157 listener = Some(reopened);
158 } else {
159 loop {
160 let (kind, _) = receive.receive(&mut stream).unwrap();
161 assert_ne!(kind, kind::EDITS, "an edit was delivered twice");
162 if kind == kind::BYE {
163 break;
164 }
165 }
166 }
167 }
168 delivered
169 });
170 let (frame, frames) = mpsc::channel();
171 let guest = Live::start(hello("Guest"), &room, None, None, move |event| {
172 if let Event::Frame { kind, body, .. } = event {
173 frame.send((kind, body.to_vec())).unwrap();
174 }
175 })
176 .unwrap();
177 for _ in 0..20 {
178 guest.connect(address);
179 }
180 for edit in [41_u64, 42] {
181 let (kind, body) = frames.recv_timeout(Duration::from_secs(10)).unwrap();
182 assert_eq!(kind, kind::EDITS);
183 assert_eq!(minicbor::decode::<u64>(&body).unwrap(), edit);
184 let peer = until(&guest, |peers| peers.len() == 1).remove(0);
185 let line = guest.line(&peer.hello.peer).unwrap();
186 line.send(kind::EDITS, &edit).unwrap();
187 if edit == 41 {
188 faults.recv_timeout(Duration::from_secs(10)).unwrap();
189 until(&guest, <[Peer]>::is_empty);
190 assert_eq!(guest.failed(), 0);
191 assert_eq!(
192 TcpStream::connect(address).unwrap_err().kind(),
193 io::ErrorKind::ConnectionRefused
194 );
195 thread::sleep(Duration::from_millis(2300));
196 resume.send(()).unwrap();
197 } else {
198 line.hang_up("left");
199 }
200 }
201 assert_eq!(server.join().unwrap(), [41, 42]);
202 assert!(matches!(frames.try_recv(), Err(mpsc::TryRecvError::Empty)));
203 assert_eq!(guest.failed(), 0);
204 until(&guest, <[Peer]>::is_empty);
205 let stopped = Arc::downgrade(&guest.shared);
206 drop(guest);
207 let deadline = Instant::now() + Duration::from_secs(1);
208 while stopped.upgrade().is_some() {
209 assert!(
210 Instant::now() < deadline,
211 "LAN helpers retained stopped Live"
212 );
213 thread::sleep(Duration::from_millis(10));
214 }
215}
216
217#[test]
218fn lan_retries_initial_refusal_and_stops_its_backoff() {
219 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
220 let address = listener.local_addr().unwrap();
221 drop(listener);
222 let room = Room::Notebook([72; 16]);
223 let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap();
224 guest.connect(address);
225 thread::sleep(Duration::from_millis(1200));
226 let listener = TcpListener::bind(address).unwrap();
227 listener.set_nonblocking(true).unwrap();
228 let mut stream = lan_accept(&listener);
229 let first = lan_opening(&mut stream);
230 assert!(!first.is_empty());
231 drop(listener);
232 drop(stream);
233 thread::sleep(Duration::from_millis(100));
234 let stopped = Arc::downgrade(&guest.shared);
235 drop(guest);
236 let deadline = Instant::now() + Duration::from_secs(1);
237 while stopped.upgrade().is_some() {
238 assert!(
239 Instant::now() < deadline,
240 "LAN backoff retained stopped Live"
241 );
242 thread::sleep(Duration::from_millis(10));
243 }
244 let listener = TcpListener::bind(address).unwrap();
245 listener.set_nonblocking(true).unwrap();
246 thread::sleep(Duration::from_millis(2200));
247 assert_eq!(
248 listener.accept().unwrap_err().kind(),
249 io::ErrorKind::WouldBlock
250 );
251}
252
253#[test]
254fn lan_authentication_and_version_failures_stop_retrying() {
255 for version in [false, true] {
256 let room = Room::Notebook([73; 16]);
257 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
258 listener.set_nonblocking(true).unwrap();
259 let address = listener.local_addr().unwrap();
260 let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap();
261 guest.connect(address);
262 let mut stream = lan_accept(&listener);
263 if version {
264 lan_opening(&mut stream);
265 let mut opening = minicbor::Encoder::new(Vec::new());
266 opening
267 .map(3)
268 .unwrap()
269 .u8(0)
270 .unwrap()
271 .u16(wire::VERSION + 1)
272 .unwrap()
273 .u8(1)
274 .unwrap()
275 .str(&room.tag().unwrap())
276 .unwrap()
277 .u8(2)
278 .unwrap()
279 .bytes(&[0; 33])
280 .unwrap();
281 let opening = opening.into_writer();
282 stream
283 .write_all(&(opening.len() as u32).to_be_bytes())
284 .unwrap();
285 stream.write_all(&opening).unwrap();
286 until(&guest, |_| guest.other_version() == Some(wire::VERSION + 1));
287 assert_eq!(guest.failed(), 0);
288 } else {
289 let (mut send, _) = wire::open(
290 &mut stream,
291 Side::Responder,
292 &room.tag().unwrap(),
293 b"wrong secret",
294 )
295 .unwrap();
296 send.send(&mut stream, kind::HELLO, &hello("Host")).unwrap();
297 until(&guest, |_| guest.failed() == 1);
298 }
299 drop(stream);
300 thread::sleep(Duration::from_millis(1200));
301 assert_eq!(
302 listener.accept().unwrap_err().kind(),
303 io::ErrorKind::WouldBlock
304 );
305 assert_eq!(guest.failed(), u32::from(!version));
306 guest.connect(address);
307 let mut stream = lan_accept(&listener);
308 assert!(
309 !lan_opening(&mut stream).is_empty(),
310 "finished dial was never released"
311 );
312 drop(guest);
313 }
314}
315
316#[test]
317fn lan_duplicate_peer_stops_retrying_without_replacing_its_line() {
318 let room = Room::Notebook([74; 16]);
319 let host_hello = hello("Host");
320 let (frame, frames) = mpsc::channel();
321 let host = Live::start(host_hello.clone(), &room, None, None, move |event| {
322 if let Event::Frame { kind, body, .. } = event {
323 frame.send((kind, body.to_vec())).unwrap();
324 }
325 })
326 .unwrap();
327 let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap();
328 guest.connect(host.address());
329 until(&guest, |peers| peers.len() == 1);
330 let line = guest.line(&host_hello.peer).unwrap();
331 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
332 listener.set_nonblocking(true).unwrap();
333 guest.connect(listener.local_addr().unwrap());
334 let mut stream = lan_accept(&listener);
335 let (mut send, mut receive) = wire::open(
336 &mut stream,
337 Side::Responder,
338 &room.tag().unwrap(),
339 &room.secret(),
340 )
341 .unwrap();
342 send.send(&mut stream, kind::HELLO, &host_hello).unwrap();
343 assert_eq!(receive.receive(&mut stream).unwrap().0, kind::HELLO);
344 assert_eq!(stream.read(&mut [0; 1]).unwrap(), 0);
345 thread::sleep(Duration::from_millis(1200));
346 assert_eq!(
347 listener.accept().unwrap_err().kind(),
348 io::ErrorKind::WouldBlock
349 );
350 assert_eq!(guest.peers().len(), 1);
351 assert_eq!(guest.failed(), 0);
352 line.send(kind::EDITS, &43_u64).unwrap();
353 let (kind, body) = frames.recv_timeout(Duration::from_secs(10)).unwrap();
354 assert_eq!(kind, kind::EDITS);
355 assert_eq!(minicbor::decode::<u64>(&body).unwrap(), 43);
356 assert!(matches!(frames.try_recv(), Err(mpsc::TryRecvError::Empty)));
357}
358
359/// A peer holding another code never meets: its first frame does not open.
360#[test]
361fn another_code_never_meets() {
362 let ada = Live::start(
363 hello("Ada"),
364 &Room::join(&code(7, "ABCDEF"), ""),
365 None,
366 None,
367 |_| {},
368 )
369 .unwrap();
370 let mallory = Live::start(
371 hello("Mallory"),
372 &Room::join(&code(7, "ABCDEG"), ""),
373 None,
374 None,
375 |_| {},
376 )
377 .unwrap();
378 mallory.connect(ada.address());
379 thread::sleep(Duration::from_millis(500));
380 assert!(ada.peers().is_empty() && mallory.peers().is_empty());
381}
382
383/// A field or a kind a later version adds is skipped by this one.
384#[test]
385fn later_fields_and_kinds_are_skipped() {
386 #[derive(Encode)]
387 #[cbor(map)]
388 struct Later {
389 #[cbor(n(0), with = "minicbor::bytes")]
390 section: Option<[u8; 16]>,
391 #[n(9)]
392 mood: String,
393 }
394 use minicbor::Encode;
395 let body = minicbor::to_vec(Later {
396 section: Some([5; 16]),
397 mood: "curious".into(),
398 })
399 .unwrap();
400 let presence: Presence = minicbor::decode(&body).unwrap();
401 assert_eq!(
402 presence,
403 Presence {
404 section: Some([5; 16]),
405 ..Presence::default()
406 }
407 );
408
409 let room = Room::join(&code(4, "QJETHR"), "");
410 let grace = Live::start(hello("Grace"), &room, None, None, |_| {}).unwrap();
411 // A later version: it greets, says something new, then where it is.
412 let later = thread::spawn(move || {
413 let mut stream = TcpStream::connect(grace.address()).unwrap();
414 let (mut send, mut receive) = wire::open(
415 &mut stream,
416 Side::Initiator,
417 &room.tag().unwrap(),
418 &room.secret(),
419 )
420 .unwrap();
421 send.send(&mut stream, kind::HELLO, &hello("Later"))
422 .unwrap();
423 receive.receive(&mut stream).unwrap();
424 send.send(&mut stream, 999, &"a chat message").unwrap();
425 send.send(&mut stream, kind::PRESENCE, &caret(4)).unwrap();
426 until(&grace, |peers| {
427 peers
428 .first()
429 .is_some_and(|peer| peer.presence == Some(caret(4)))
430 });
431 });
432 later.join().unwrap();
433}
434
435/// Two ends of one notebook find each other by mDNS on this computer's loopback.
436#[test]
437#[ignore = "multicasts mDNS on the loopback interface"]
438fn peers_find_each_other_on_loopback() {
439 let room = Room::Notebook([9; 16]);
440 let ada = Live::start(hello("Ada"), &room, Some(Reach::Loopback), None, |_| {}).unwrap();
441 let grace = Live::start(hello("Grace"), &room, Some(Reach::Loopback), None, |_| {}).unwrap();
442 until(&ada, |peers| peers.len() == 1);
443 until(&grace, |peers| peers.len() == 1);
444}
445
446/// Two sealers that met over loopback: Ada's to send with and Grace's to receive with.
447fn sealers() -> (Sealer, Sealer) {
448 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
449 let address = listener.local_addr().unwrap();
450 let ada = thread::spawn(move || {
451 let mut stream = TcpStream::connect(address).unwrap();
452 wire::open(&mut stream, Side::Initiator, "room", b"secret")
453 .unwrap()
454 .0
455 });
456 let (mut stream, _) = listener.accept().unwrap();
457 let grace = wire::open(&mut stream, Side::Responder, "room", b"secret")
458 .unwrap()
459 .1;
460 (ada.join().unwrap(), grace)
461}
462
463/// Ada's carets at offsets `0..count`, each frame as its own block of bytes.
464fn frames(ada: &mut Sealer, count: u32) -> Vec<Vec<u8>> {
465 (0..count)
466 .map(|offset| {
467 let mut block = Vec::new();
468 ada.send(&mut block, kind::PRESENCE, &caret(offset))
469 .unwrap();
470 block
471 })
472 .collect()
473}
474
475/// A frame lost, repeated, reordered, altered, forged or sealed for another meeting is caught
476/// where it lands, before anything in it or after it is read.
477#[test]
478fn frames_out_of_place_are_caught() {
479 let (mut other, _) = sealers();
480 let elsewhere = frames(&mut other, 3).remove(2);
481 type Edit = dyn Fn(&mut Vec<Vec<u8>>);
482 let cases: [(&str, Box<Edit>, &str); 6] = [
483 (
484 "lost",
485 Box::new(|sent| drop(sent.remove(2))),
486 "Frame 3 came where frame 2 was due",
487 ),
488 (
489 "repeated",
490 Box::new(|sent| {
491 let again = sent[1].clone();
492 sent.insert(2, again);
493 }),
494 "Frame 1 came where frame 2 was due",
495 ),
496 (
497 "reordered",
498 Box::new(|sent| sent.swap(2, 3)),
499 "Frame 3 came where frame 2 was due",
500 ),
501 (
502 "altered",
503 Box::new(|sent| *sent[2].last_mut().unwrap() ^= 1),
504 "Frame 2 does not open",
505 ),
506 (
507 "forged",
508 Box::new(|sent| {
509 // Its length and number kept, the rest made up.
510 let mut forged = sent[2].clone();
511 forged[12..].fill(7);
512 sent.insert(2, forged);
513 }),
514 "Frame 2 does not open",
515 ),
516 (
517 "from another meeting",
518 Box::new(move |sent| sent.insert(2, elsewhere.clone())),
519 "Frame 2 does not open",
520 ),
521 ];
522 for (name, tamper, expected) in cases {
523 let (mut ada, mut grace) = sealers();
524 let mut sent = frames(&mut ada, 5);
525 tamper(&mut sent);
526 let stream = sent.concat();
527 let mut arriving = &stream[..];
528 for offset in 0..2 {
529 let (message, body) = grace.receive(&mut arriving).unwrap();
530 assert_eq!(message, kind::PRESENCE);
531 assert_eq!(minicbor::decode::<Presence>(&body).unwrap(), caret(offset));
532 }
533 let error = grace.receive(&mut arriving).unwrap_err();
534 assert_eq!(error.kind(), io::ErrorKind::InvalidData, "{name}");
535 assert!(error.to_string().starts_with(expected), "{name}: {error}");
536 }
537}
538
539/// A relay on this computer, with `config`'s limits: its URL.
540fn relay(config: ::relay::server::Config) -> (String, SocketAddr) {
541 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
542 let address = listener.local_addr().unwrap();
543 thread::spawn(move || ::relay::server::serve(listener, config));
544 (format!("ws://{address}"), address)
545}
546
547/// The status a relay at `address` answers a WebSocket to `path` with.
548fn status(address: SocketAddr, path: &str) -> String {
549 let mut stream = TcpStream::connect(address).unwrap();
550 write!(
551 stream,
552 "GET {path} HTTP/1.1\r\nHost: relay\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
553 Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n"
554 )
555 .unwrap();
556 let head = ::relay::ws::head(&mut stream).unwrap();
557 head.lines().next().unwrap().to_owned()
558}
559
560/// Two ends of one notebook that share no network meet in the relay's room, and see each
561/// other leave.
562#[test]
563fn peers_meet_through_a_relay() {
564 let (url, _) = relay(Default::default());
565 let room = Room::Notebook([3; 16]);
566 let ada = Live::start(hello("Ada"), &room, None, Some(&url), |_| {}).unwrap();
567 ada.set_presence(caret(1));
568 let grace = Live::start(hello("Grace"), &room, None, Some(&url), |_| {}).unwrap();
569 until(&grace, |peers| {
570 peers.len() == 1 && peers[0].presence == Some(caret(1))
571 });
572 until(&ada, |peers| {
573 peers.len() == 1 && peers[0].hello.name == "Grace"
574 });
575 ada.set_presence(caret(2));
576 until(&grace, |peers| peers[0].presence == Some(caret(2)));
577 drop(ada);
578 until(&grace, <[Peer]>::is_empty);
579}
580
581#[test]
582fn relay_timeouts_reopen_with_fresh_keys() {
583 use ::relay::ws::{self, Message};
584
585 let (url, upstream) = relay(Default::default());
586 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
587 let delayed = format!("ws://{}", listener.local_addr().unwrap());
588 listener.set_nonblocking(true).unwrap();
589 let (opening, openings) = mpsc::channel();
590 let proxy = thread::spawn(move || {
591 for attempt in 0..3 {
592 let deadline = Instant::now() + Duration::from_secs(15);
593 let mut client = loop {
594 match listener.accept() {
595 Ok((client, _)) => break client,
596 Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
597 assert!(Instant::now() < deadline, "the relay room never reopened");
598 thread::sleep(Duration::from_millis(20));
599 }
600 Err(error) => panic!("{error}"),
601 }
602 };
603 client.set_nonblocking(false).unwrap();
604 client
605 .set_read_timeout(Some(Duration::from_secs(15)))
606 .unwrap();
607 let mut server = TcpStream::connect(upstream).unwrap();
608 let head = ws::head(&mut client).unwrap();
609 server.write_all(head.as_bytes()).unwrap();
610 let (mut down, mut to_client) =
611 (server.try_clone().unwrap(), client.try_clone().unwrap());
612 let replies = thread::spawn(move || {
613 let _ = io::copy(&mut down, &mut to_client);
614 let _ = to_client.shutdown(Shutdown::Both);
615 });
616 let mut messages = ws::Reader::new(client, 1 << 20, true);
617 let mut first = true;
618 while let Ok(message) = messages.read() {
619 let (opcode, data) = match message {
620 Message::Binary(data) => {
621 let slot = u32::from_be_bytes(data[..4].try_into().unwrap());
622 if first && slot > 0 && slot < ::relay::GROUP {
623 first = false;
624 opening.send(data[4..].to_vec()).unwrap();
625 // Keep the first opening delayed until its stream is abandoned.
626 if attempt == 0 {
627 continue;
628 }
629 }
630 (ws::BINARY, data)
631 }
632 Message::Text(text) => (ws::TEXT, text.into_bytes()),
633 Message::Ping(data) => (ws::PING, data),
634 Message::Pong => (ws::PONG, vec![]),
635 Message::Close => break,
636 };
637 if server
638 .write_all(&ws::frame(opcode, &data, Some([1, 2, 3, 4])))
639 .is_err()
640 {
641 break;
642 }
643 }
644 let _ = server.shutdown(Shutdown::Both);
645 replies.join().unwrap();
646 }
647 });
648
649 let room = Room::Notebook([7; 16]);
650 let mut host = hello("Host");
651 host.serves = Some([8; 16]);
652 let (met, host_met) = mpsc::channel();
653 let (frame, frames) = mpsc::channel();
654 let host = Live::start(host, &room, None, Some(&url), move |event| match event {
655 Event::Met(_, line) => {
656 let _ = met.send(line.clone());
657 }
658 Event::Frame { kind, body, .. } => {
659 let _ = frame.send((kind, body.to_vec()));
660 }
661 _ => {}
662 })
663 .unwrap();
664 let (met, guest_met) = mpsc::channel();
665 let (left, guest_left) = mpsc::channel();
666 let (frame, guest_frames) = mpsc::channel();
667 let guest = Live::start(
668 hello("Guest"),
669 &room,
670 None,
671 Some(&delayed),
672 move |event| match event {
673 Event::Met(_, line) => {
674 let _ = met.send(line.clone());
675 }
676 Event::Left(_) => {
677 let _ = left.send(());
678 }
679 Event::Frame { kind, body, .. } => {
680 let _ = frame.send((kind, body.to_vec()));
681 }
682 _ => {}
683 },
684 )
685 .unwrap();
686 let first = openings.recv_timeout(Duration::from_secs(10)).unwrap();
687 assert!(matches!(
688 host_met.try_recv(),
689 Err(mpsc::TryRecvError::Empty)
690 ));
691 assert!(matches!(
692 guest_met.try_recv(),
693 Err(mpsc::TryRecvError::Empty)
694 ));
695 let fresh = openings.recv_timeout(Duration::from_secs(10)).unwrap();
696 assert_ne!(first, fresh, "the retry reused the abandoned key exchange");
697 let host_line = host_met.recv_timeout(Duration::from_secs(10)).unwrap();
698 guest_met.recv_timeout(Duration::from_secs(10)).unwrap();
699 let before_timeout = guest.shared.state.lock().unwrap().relay.clone().unwrap();
700 guest
701 .shared
702 .state
703 .lock()
704 .unwrap()
705 .peers
706 .values()
707 .next()
708 .unwrap()
709 .pipe
710 .set_read_timeout(Duration::from_secs(1))
711 .unwrap();
712 host_line.send(kind::EDITS, &40_u64).unwrap();
713 let (kind, body) = guest_frames.recv_timeout(Duration::from_secs(10)).unwrap();
714 assert_eq!(kind, kind::EDITS);
715 assert_eq!(minicbor::decode::<u64>(&body).unwrap(), 40);
716 guest_left.recv_timeout(Duration::from_secs(10)).unwrap();
717 let reopened = openings.recv_timeout(Duration::from_secs(10)).unwrap();
718 assert_ne!(
719 fresh, reopened,
720 "the retry reused the timed-out stream's keys"
721 );
722 host_met.recv_timeout(Duration::from_secs(10)).unwrap();
723 let line = guest_met.recv_timeout(Duration::from_secs(10)).unwrap();
724 assert_eq!((host.failed(), guest.failed()), (0, 0));
725 let before = guest.shared.state.lock().unwrap().relay.clone().unwrap();
726 assert!(!Arc::ptr_eq(&before_timeout, &before));
727 for edit in [41_u64, 42] {
728 line.send(kind::EDITS, &edit).unwrap();
729 }
730 line.hang_up("left");
731 let mut delivered = Vec::new();
732 loop {
733 let (kind, body) = frames.recv_timeout(Duration::from_secs(10)).unwrap();
734 match kind {
735 kind::EDITS => delivered.push(minicbor::decode::<u64>(&body).unwrap()),
736 kind::BYE => break,
737 _ => {}
738 }
739 }
740 assert_eq!(delivered, [41, 42]);
741 guest_left.recv_timeout(Duration::from_secs(10)).unwrap();
742 host.set_presence(caret(43));
743 until(&guest, |peers| {
744 peers
745 .first()
746 .is_some_and(|peer| peer.presence == Some(caret(43)))
747 });
748 let state = guest.shared.state.lock().unwrap();
749 assert!(Arc::ptr_eq(&before, state.relay.as_ref().unwrap()));
750 drop(state);
751 assert!(matches!(
752 host_met.try_recv(),
753 Err(mpsc::TryRecvError::Empty)
754 ));
755 assert!(matches!(
756 guest_met.try_recv(),
757 Err(mpsc::TryRecvError::Empty)
758 ));
759 assert!(matches!(
760 guest_frames.try_recv(),
761 Err(mpsc::TryRecvError::Empty)
762 ));
763 drop(guest);
764 drop(host);
765 proxy.join().unwrap();
766}
767
768/// The end sharing a code asks the relay to number it; the other types the whole code. One
769/// with the wrong words never meets, and the relay, told so by the end sharing, burns the
770/// code once too many have tried.
771#[test]
772fn a_relay_numbers_a_code_and_burns_it_after_wrong_tries() {
773 let (url, address) = relay(::relay::server::Config {
774 burn_after: 2,
775 ..Default::default()
776 });
777 let host = Live::start(
778 hello("Ada"),
779 &Room::share("ABCDEF", ""),
780 None,
781 Some(&url),
782 |_| {},
783 )
784 .unwrap();
785 let deadline = Instant::now() + Duration::from_secs(10);
786 let code = loop {
787 if let Some(code) = host.code() {
788 break code;
789 }
790 assert!(Instant::now() < deadline, "no code");
791 thread::sleep(Duration::from_millis(20));
792 };
793 let (number, secret) = super::code::parse(&code).unwrap();
794 assert_eq!(secret, "ABCDEF");
795 let guest = Live::start(
796 hello("Grace"),
797 &Room::join(&code, ""),
798 None,
799 Some(&url),
800 |_| {},
801 )
802 .unwrap();
803 until(&host, |peers| peers.len() == 1);
804 until(&guest, |peers| peers.len() == 1);
805
806 let path = format!("/v1/room/code-{number}");
807 let wrong = Room::join(&self::code(number, "ABCDEG"), "");
808 // An end that typed a wrong code gives up at once; Mallory tries twice, and the second
809 // wrong try burns the code.
810 for _ in 0..2 {
811 let mallory = Live::start(hello("Mallory"), &wrong, None, Some(&url), |_| {}).unwrap();
812 let deadline = Instant::now() + Duration::from_secs(10);
813 while mallory.failed() == 0 {
814 assert!(Instant::now() < deadline, "Mallory never tried");
815 thread::sleep(Duration::from_millis(20));
816 }
817 assert!(mallory.peers().is_empty());
818 }
819 until(&host, |_| host.burned());
820 assert_eq!(status(address, &path), "HTTP/1.1 410 Gone");
821 assert_eq!(host.peers().len(), 1, "Grace stays");
822}
823
824/// What a malicious relay does to one message on its way.
825#[derive(Clone, Copy, Debug)]
826enum Tamper {
827 Drop,
828 Repeat,
829 Reorder,
830 Alter,
831 Inject,
832}
833
834/// A relay in the middle of Grace's connection that passes on what the real one at
835/// `upstream` says, except the first message to her from a peer once `armed`, which it
836/// tampers with. Her next connection waits for `release`.
837struct Malicious {
838 url: String,
839 armed: Arc<std::sync::atomic::AtomicBool>,
840 tampered: mpsc::Receiver<()>,
841 rejoined: mpsc::Receiver<()>,
842 release: mpsc::Sender<()>,
843}
844
845fn malicious(upstream: SocketAddr, tamper: Tamper) -> Malicious {
846 use ::relay::ws::{self, Message};
847 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
848 let url = format!("ws://{}", listener.local_addr().unwrap());
849 let (tampered, told) = mpsc::channel();
850 let (rejoined, heard) = mpsc::channel();
851 let (release, released) = mpsc::channel();
852 let armed = Arc::new(std::sync::atomic::AtomicBool::new(false));
853 let arming = Arc::clone(&armed);
854 thread::spawn(move || {
855 for (index, client) in listener.incoming().enumerate() {
856 let mut client = client.unwrap();
857 if index > 0 {
858 let _ = rejoined.send(());
859 let _ = released.recv();
860 }
861 let server = TcpStream::connect(upstream).unwrap();
862 let (mut up, mut to_server) =
863 (client.try_clone().unwrap(), server.try_clone().unwrap());
864 thread::spawn(move || {
865 let _ = io::copy(&mut up, &mut to_server);
866 let _ = to_server.shutdown(Shutdown::Both);
867 });
868 let (tampered, armed) = (tampered.clone(), Arc::clone(&arming));
869 thread::spawn(move || {
870 let mut reading = io::BufReader::new(server);
871 let head = ws::head(&mut reading).unwrap();
872 client.write_all(head.as_bytes()).unwrap();
873 let mut messages = ws::Reader::new(reading, 1 << 20, false);
874 let mut held = None;
875 while let Ok(message) = messages.read() {
876 let frames: Vec<Vec<u8>> = match message {
877 Message::Binary(mut data) => {
878 let mut out = vec![];
879 if index == 0 && armed.swap(false, std::sync::atomic::Ordering::AcqRel)
880 {
881 match tamper {
882 Tamper::Drop => {}
883 Tamper::Repeat => out = vec![data.clone(), data],
884 Tamper::Reorder => held = Some(data),
885 Tamper::Alter => {
886 *data.last_mut().unwrap() ^= 1;
887 out = vec![data];
888 }
889 Tamper::Inject => {
890 // A slot and part of the sender's id, then
891 // made-up bytes.
892 let mut forged = data.clone();
893 forged[16..].fill(7);
894 out = vec![forged, data];
895 }
896 }
897 let _ = tampered.send(());
898 } else {
899 out.push(data);
900 out.extend(held.take());
901 }
902 out.into_iter()
903 .map(|data| ws::frame(ws::BINARY, &data, None))
904 .collect()
905 }
906 Message::Text(text) => vec![ws::frame(ws::TEXT, text.as_bytes(), None)],
907 Message::Ping(payload) => vec![ws::frame(ws::PING, &payload, None)],
908 Message::Pong => vec![ws::frame(ws::PONG, &[], None)],
909 Message::Close => break,
910 };
911 if frames.iter().any(|frame| client.write_all(frame).is_err()) {
912 break;
913 }
914 }
915 let _ = client.shutdown(Shutdown::Both);
916 });
917 }
918 });
919 Malicious {
920 url,
921 armed,
922 tampered: told,
923 rejoined: heard,
924 release,
925 }
926}
927
928/// Starts `name`, recording each presence of its peer as applied, and `None` when the peer
929/// goes.
930fn recording(name: &str, room: &Room, relay: &str) -> (Live, Arc<Mutex<Vec<Option<Presence>>>>) {
931 let heard = Arc::new(Mutex::new(Vec::new()));
932 let shared: Arc<std::sync::OnceLock<std::sync::Weak<Shared>>> = Arc::default();
933 let (recorded, watched) = (Arc::clone(&heard), Arc::clone(&shared));
934 let live = Live::start(hello(name), room, None, Some(relay), move |event| {
935 if !matches!(event, Event::Changed) {
936 return;
937 }
938 let Some(shared) = watched.get().and_then(std::sync::Weak::upgrade) else {
939 return;
940 };
941 let entry = match shared.peers().first() {
942 None => Some(None),
943 Some(peer) => peer.presence.clone().map(Some),
944 };
945 recorded.lock().unwrap().extend(entry);
946 })
947 .unwrap();
948 shared.set(Arc::downgrade(&live.shared)).ok().unwrap();
949 (live, heard)
950}
951
952/// Whatever a malicious relay does to a frame, the end it was for never applies it or
953/// anything after it, drops the connection as broken, joins again and hears the newest
954/// presence from scratch.
955#[test]
956fn a_malicious_relay_is_caught() {
957 let room = Room::Notebook([4; 16]);
958 for tamper in [
959 Tamper::Drop,
960 Tamper::Repeat,
961 Tamper::Reorder,
962 Tamper::Alter,
963 Tamper::Inject,
964 ] {
965 let (url, address) = relay(Default::default());
966 let ada = Live::start(hello("Ada"), &room, None, Some(&url), |_| {}).unwrap();
967 ada.set_presence(caret(1));
968 let relay = malicious(address, tamper);
969 let (grace, heard) = recording("Grace", &room, &relay.url);
970 until(&grace, |peers| {
971 peers
972 .first()
973 .is_some_and(|peer| peer.presence == Some(caret(1)))
974 });
975 // Ada's next presence goes to everyone, so its loss is caught too.
976 relay
977 .armed
978 .store(true, std::sync::atomic::Ordering::Release);
979 ada.set_presence(caret(2));
980 relay
981 .tampered
982 .recv_timeout(Duration::from_secs(10))
983 .unwrap();
984 ada.set_presence(caret(3));
985 relay
986 .rejoined
987 .recv_timeout(Duration::from_secs(10))
988 .unwrap();
989 let before: Vec<_> = heard.lock().unwrap().clone();
990 let gone = before.iter().position(Option::is_none).unwrap();
991 assert!(
992 !before[..gone].contains(&Some(caret(3))),
993 "{tamper:?}: applied after the tampering: {before:?}"
994 );
995 assert!(grace.peers().is_empty(), "{tamper:?}");
996 relay.release.send(()).unwrap();
997 until(&grace, |peers| {
998 peers
999 .first()
1000 .is_some_and(|peer| peer.presence == Some(caret(3)))
1001 });
1002 }
1003}
1004
1005/// Ends of two versions each learn the other's version from the opening, before any key is
1006/// agreed, so each can say which should update.
1007#[test]
1008fn another_version_is_named_before_any_key() {
1009 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
1010 let address = listener.local_addr().unwrap();
1011 let older = thread::spawn(move || {
1012 let mut stream = TcpStream::connect(address).unwrap();
1013 #[derive(minicbor::Encode)]
1014 #[cbor(map)]
1015 struct Open {
1016 #[n(0)]
1017 version: u16,
1018 #[n(1)]
1019 room: String,
1020 #[cbor(n(2), with = "minicbor::bytes")]
1021 pake: Vec<u8>,
1022 }
1023 let open = minicbor::to_vec(Open {
1024 version: 1,
1025 room: "room".into(),
1026 pake: vec![0; 33],
1027 })
1028 .unwrap();
1029 stream
1030 .write_all(&[&(open.len() as u32).to_be_bytes()[..], &open].concat())
1031 .unwrap();
1032 let mut length = [0; 4];
1033 stream.read_exact(&mut length).unwrap();
1034 let mut answer = vec![0; u32::from_be_bytes(length) as usize];
1035 stream.read_exact(&mut answer).unwrap();
1036 answer
1037 });
1038 let (mut stream, _) = listener.accept().unwrap();
1039 let Err(error) = wire::open(&mut stream, Side::Responder, "room", b"secret") else {
1040 panic!("met a peer of another version");
1041 };
1042 assert_eq!(error.kind(), io::ErrorKind::Unsupported);
1043 let version = error.get_ref().unwrap().downcast_ref::<wire::Version>();
1044 assert_eq!(version, Some(&wire::Version(1)));
1045 // The older end heard this one's opening, and with it its version.
1046 assert!(!older.join().unwrap().is_empty());
1047}