1use super::super::super::{GONE, OPENING, Pipe, Shared, State};
2use super::*;
3use std::{
4 io::{Read, Write},
5 net::{Shutdown, TcpListener, TcpStream},
6 sync::atomic::AtomicUsize,
7};
8
9struct Barrier {
10 reached: mpsc::Sender<()>,
11 resume: mpsc::Receiver<()>,
12}
13
14impl Barrier {
15 fn wait(self) -> io::Result<()> {
16 self.reached.send(()).map_err(io::Error::other)?;
17 self.resume.recv_timeout(GONE).map_err(io::Error::other)
18 }
19}
20
21fn barrier() -> (Barrier, mpsc::Receiver<()>, mpsc::Sender<()>) {
22 let (reached, waiting) = mpsc::channel();
23 let (resume, continued) = mpsc::channel();
24 (
25 Barrier {
26 reached,
27 resume: continued,
28 },
29 waiting,
30 resume,
31 )
32}
33
34struct HandoverPipe {
35 stream: TcpStream,
36 direct: bool,
37 body: Mutex<Option<Barrier>>,
38 retired: mpsc::Sender<()>,
39}
40
41impl Pipe for HandoverPipe {
42 fn read(&self, buffer: &mut [u8]) -> io::Result<usize> {
43 let gate = if buffer.len() > 4 {
44 self.body.lock().unwrap().take()
45 } else {
46 None
47 };
48 if let Some(gate) = gate {
49 // Preserve a complete copied ciphertext block across the old pipe's shutdown.
50 Read::read_exact(&mut &self.stream, buffer)?;
51 gate.wait()?;
52 Ok(buffer.len())
53 } else {
54 Read::read(&mut &self.stream, buffer)
55 }
56 }
57
58 fn write(&self, bytes: &[u8]) -> io::Result<usize> {
59 Write::write(&mut &self.stream, bytes)
60 }
61
62 fn set_read_timeout(&self, timeout: Duration) -> io::Result<()> {
63 self.stream.set_read_timeout(Some(timeout))
64 }
65
66 fn shutdown(&self) {
67 let _ = self.stream.shutdown(Shutdown::Both);
68 }
69
70 fn direct(&self) -> bool {
71 self.direct
72 }
73}
74
75impl Drop for HandoverPipe {
76 fn drop(&mut self) {
77 let _ = self.retired.send(());
78 }
79}
80
81struct Handover {
82 shared: Arc<Shared>,
83 guest: Arc<Inner>,
84 pipes: Vec<Arc<HandoverPipe>>,
85 readers: Vec<(mpsc::Receiver<()>, thread::JoinHandle<bool>)>,
86 retired: (mpsc::Sender<()>, mpsc::Receiver<()>),
87 observed: mpsc::Receiver<()>,
88 releases: Vec<mpsc::Sender<()>>,
89}
90
91impl Handover {
92 fn new(met: Option<Barrier>) -> Self {
93 let me = Hello::new("Guest".into(), None).unwrap();
94 let guest = Arc::new(Inner {
95 share: [41; 16],
96 me: me.clone(),
97 reach: None,
98 relay: None,
99 events: Arc::new(|| {}),
100 presence: Mutex::default(),
101 here: Mutex::default(),
102 host: Mutex::default(),
103 pending: Mutex::default(),
104 next: AtomicU64::new(1),
105 watch: Mutex::default(),
106 ended: Mutex::default(),
107 asked: Default::default(),
108 images: Mutex::default(),
109 current: Mutex::default(),
110 });
111 let heard = Arc::clone(&guest);
112 let met = Mutex::new(met);
113 let (events, observed) = mpsc::channel();
114 let room = Room::Notebook([42; 16]);
115 let shared = Arc::new(Shared {
116 me,
117 secret: room.secret(),
118 room,
119 reach: None,
120 state: Mutex::new(State::default()),
121 events: Box::new(move |source, event| {
122 let meeting = matches!(&event, Event::Met(..));
123 if meeting {
124 let gate = met.lock().unwrap().take();
125 if let Some(gate) = gate {
126 gate.wait().unwrap();
127 }
128 }
129 heard.heard(source, event);
130 if meeting {
131 let _ = events.send(());
132 }
133 }),
134 stopped: AtomicBool::new(false),
135 connections: AtomicU64::new(0),
136 });
137 Self {
138 shared,
139 guest,
140 pipes: Vec::new(),
141 readers: Vec::new(),
142 retired: mpsc::channel(),
143 observed,
144 releases: Vec::new(),
145 }
146 }
147
148 fn open(&mut self, host: &Hello, direct: bool) -> (TcpStream, wire::Sealer, wire::Sealer) {
149 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
150 let mut remote =
151 TcpStream::connect_timeout(&listener.local_addr().unwrap(), OPENING).unwrap();
152 remote.set_read_timeout(Some(OPENING)).unwrap();
153 remote.set_write_timeout(Some(OPENING)).unwrap();
154 let (local, _) = listener.accept().unwrap();
155 local.set_write_timeout(Some(OPENING)).unwrap();
156 let pipe = Arc::new(HandoverPipe {
157 stream: local,
158 direct,
159 body: Mutex::default(),
160 retired: self.retired.0.clone(),
161 });
162 self.pipes.push(Arc::clone(&pipe));
163 let shared = Arc::clone(&self.shared);
164 let tag = shared.room.tag().unwrap();
165 let reader_tag = tag.clone();
166 let (finished, done) = mpsc::channel();
167 let reader = thread::spawn(move || {
168 let retry = shared.run(pipe, wire::Side::Responder, &reader_tag);
169 let _ = finished.send(());
170 retry
171 });
172 self.readers.push((done, reader));
173 let (mut send, mut receive) = wire::open(
174 &mut remote,
175 wire::Side::Initiator,
176 &tag,
177 &self.shared.secret,
178 )
179 .unwrap();
180 send.send(&mut remote, kind::HELLO, host).unwrap();
181 let (message, body) = receive.receive(&mut remote).unwrap();
182 assert_eq!(message, kind::HELLO);
183 assert_eq!(minicbor::decode::<Hello>(&body).unwrap(), self.shared.me);
184 (remote, send, receive)
185 }
186
187 fn heard(&self) {
188 self.observed.recv_timeout(OPENING).unwrap();
189 }
190}
191
192impl Drop for Handover {
193 fn drop(&mut self) {
194 self.shared.stopped.store(true, Ordering::Release);
195 for release in self.releases.drain(..) {
196 let _ = release.send(());
197 }
198 for pipe in &self.pipes {
199 pipe.shutdown();
200 }
201 self.shared.state.lock().unwrap().peers.clear();
202 let readers_done = self
203 .readers
204 .iter()
205 .all(|(done, _)| done.recv_timeout(OPENING) != Err(mpsc::RecvTimeoutError::Timeout));
206 self.guest.host.lock().unwrap().take();
207 let count = self.pipes.len();
208 self.pipes.clear();
209 let joined = readers_done
210 && (0..count).all(|_| self.retired.1.recv_timeout(OPENING).is_ok())
211 && (self.readers.drain(..)).all(|(_, reader)| reader.join().is_ok());
212 if !thread::panicking() {
213 assert!(
214 joined,
215 "handover reader/writer owners did not retire and join"
216 );
217 }
218 }
219}
220
221#[test]
222fn handover_stale_frame_cannot_end_promoted_guest() {
223 let mut fixture = Handover::new(None);
224 let mut host = Hello::new("Old host route".into(), None).unwrap();
225 host.serves = Some(fixture.guest.share);
226 let (mut old, mut send, _) = fixture.open(&host, false);
227 fixture.heard();
228 let (gate, copied, resume) = barrier();
229 fixture.releases.push(resume.clone());
230 *fixture.pipes[0].body.lock().unwrap() = Some(gate);
231 send.send(
232 &mut old,
233 kind::BYE,
234 &wire::Bye {
235 reason: STOPPED.into(),
236 },
237 )
238 .unwrap();
239 copied.recv_timeout(OPENING).unwrap();
240 host.name = "Promoted host route".into();
241 let (_new, _, _) = fixture.open(&host, true);
242 fixture.heard();
243 assert!(
244 fixture.shared.state.lock().unwrap().peers[&host.peer]
245 .pipe
246 .direct()
247 );
248 assert!(fixture.guest.ended.lock().unwrap().is_none());
249 resume.send(()).unwrap();
250 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
251 let ended = *fixture.guest.ended.lock().unwrap();
252 drop(fixture);
253 assert_eq!(ended, None, "stale frame ended the promoted guest");
254}
255
256#[test]
257fn handover_reply_from_retired_route_reaches_its_waiter() {
258 let mut fixture = Handover::new(None);
259 let mut host = Hello::new("Old host route".into(), None).unwrap();
260 host.serves = Some(fixture.guest.share);
261 let (mut old, mut send, _) = fixture.open(&host, false);
262 fixture.heard();
263 let (answer, waiting) = crate::task::response::<Reply>();
264 fixture.guest.pending.lock().unwrap().insert(75, answer);
265 let (gate, copied, resume) = barrier();
266 fixture.releases.push(resume.clone());
267 *fixture.pipes[0].body.lock().unwrap() = Some(gate);
268 let reply = Reply {
269 id: 75,
270 length: Some(12),
271 ..Reply::default()
272 };
273 send.send(&mut old, kind::REPLY, &reply).unwrap();
274 copied.recv_timeout(OPENING).unwrap();
275 host.name = "Promoted host route".into();
276 let (_new, _, _) = fixture.open(&host, true);
277 fixture.heard();
278 resume.send(()).unwrap();
279 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
280 let answered = waiting.try_recv().ok();
281 drop(fixture);
282 assert_eq!(
283 answered,
284 Some(reply),
285 "a reply read off a retired route missed its waiter"
286 );
287}
288
289#[test]
290fn handover_stale_met_cannot_replace_promoted_guest_line() {
291 let (gate, delayed, resume) = barrier();
292 let mut fixture = Handover::new(Some(gate));
293 fixture.releases.push(resume.clone());
294 let mut host = Hello::new("Old host route".into(), None).unwrap();
295 host.serves = Some(fixture.guest.share);
296 let (_old, _, _) = fixture.open(&host, false);
297 delayed.recv_timeout(OPENING).unwrap();
298 host.name = "Promoted host route".into();
299 let (_new, _, _) = fixture.open(&host, true);
300 fixture.heard();
301 assert_eq!(
302 fixture.guest.host.lock().unwrap().as_ref().unwrap().0.name,
303 host.name
304 );
305 resume.send(()).unwrap();
306 fixture.heard();
307 assert_eq!(
308 fixture.shared.state.lock().unwrap().peers[&host.peer]
309 .peer
310 .hello
311 .name,
312 host.name
313 );
314 let selected = fixture
315 .guest
316 .host
317 .lock()
318 .unwrap()
319 .as_ref()
320 .unwrap()
321 .0
322 .name
323 .clone();
324 drop(fixture);
325 assert_eq!(
326 selected, host.name,
327 "stale Met replaced the promoted guest Line"
328 );
329}
330
331impl Handover {
332 fn intercept(
333 &mut self,
334 before: impl Fn(&Arc<Shared>, &Event<'_>) + Send + Sync + 'static,
335 ) -> mpsc::Receiver<u16> {
336 let guest = Arc::clone(&self.guest);
337 let (observed, waiting) = mpsc::channel();
338 Arc::get_mut(&mut self.shared).unwrap().events = Box::new(move |source, event| {
339 let kind = match &event {
340 Event::Met(..) => kind::HELLO,
341 Event::Left(..) => kind::BYE,
342 Event::Frame { kind, .. } => *kind,
343 Event::Changed => 0,
344 };
345 before(source, &event);
346 guest.heard(source, event);
347 let _ = observed.send(kind);
348 });
349 waiting
350 }
351}
352
353fn observed(events: &mpsc::Receiver<u16>, wanted: u16) {
354 let end = std::time::Instant::now() + OPENING;
355 loop {
356 let kind = events
357 .recv_timeout(end.saturating_duration_since(std::time::Instant::now()))
358 .unwrap();
359 if kind == wanted {
360 return;
361 }
362 }
363}
364
365#[test]
366fn handover_old_left_preserves_promoted_host_and_pending_reply() {
367 let mut fixture = Handover::new(None);
368 let (gate, removed, resume) = barrier();
369 fixture.releases.push(resume.clone());
370 let gate = Mutex::new(Some(gate));
371 let events = fixture.intercept(move |_, event| {
372 if matches!(event, Event::Left(_))
373 && let Some(gate) = gate.lock().unwrap().take()
374 {
375 gate.wait().unwrap();
376 }
377 });
378 let mut host = Hello::new("Old host route".into(), None).unwrap();
379 host.serves = Some(fixture.guest.share);
380 let (old, _, _) = fixture.open(&host, false);
381 observed(&events, kind::HELLO);
382 old.shutdown(Shutdown::Both).unwrap();
383 removed.recv_timeout(OPENING).unwrap();
384 host.name = "Promoted host route".into();
385 let (_new, _, _) = fixture.open(&host, true);
386 observed(&events, kind::HELLO);
387 let (answer, waiting) = crate::task::response::<Reply>();
388 fixture.guest.pending.lock().unwrap().insert(73, answer);
389 resume.send(()).unwrap();
390 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
391 let host_kept = fixture
392 .guest
393 .host
394 .lock()
395 .unwrap()
396 .as_ref()
397 .is_some_and(|(hello, _)| hello.name == host.name);
398 let pending_kept = fixture.guest.pending.lock().unwrap().contains_key(&73);
399 let answer_kept = matches!(waiting.try_recv(), Err(mpsc::TryRecvError::Empty));
400 drop(fixture);
401 assert!(
402 host_kept && pending_kept && answer_kept,
403 "retired Left cleared the promoted host or pending reply"
404 );
405}
406
407#[test]
408fn handover_selected_loss_clears_host_and_disconnects_pending_reply() {
409 let mut fixture = Handover::new(None);
410 let mut host = Hello::new("Selected host".into(), None).unwrap();
411 host.serves = Some(fixture.guest.share);
412 let (remote, _, _) = fixture.open(&host, true);
413 fixture.heard();
414 let (answer, waiting) = crate::task::response::<Reply>();
415 fixture.guest.pending.lock().unwrap().insert(74, answer);
416 remote.shutdown(Shutdown::Both).unwrap();
417 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
418 let host_cleared = fixture.guest.host.lock().unwrap().is_none();
419 let pending_cleared = fixture.guest.pending.lock().unwrap().is_empty();
420 let answer_lost = matches!(waiting.try_recv(), Err(mpsc::TryRecvError::Disconnected));
421 drop(fixture);
422 assert!(
423 host_cleared && pending_cleared && answer_lost,
424 "selected route loss retained the host or pending reply"
425 );
426}
427
428fn served() -> (tempfile::TempDir, Arc<Served>) {
429 let directory = tempfile::tempdir().unwrap();
430 let folder = directory.path().join("notebook");
431 std::fs::create_dir(&folder).unwrap();
432 let served = Arc::new(Served {
433 storage: crate::session::Notebook::open(&folder, directory.path().join("cache"))
434 .unwrap()
435 .into_storage(),
436 images: Mutex::default(),
437 snapshots: Mutex::default(),
438 puts: Mutex::default(),
439 guests: Mutex::default(),
440 writers: Mutex::default(),
441 host: Mutex::default(),
442 room: Mutex::default(),
443 });
444 (directory, served)
445}
446
447#[test]
448fn handover_copied_host_request_cannot_enter_promoted_queue() {
449 let (_directory, served) = served();
450 let mut fixture = Handover::new(None);
451 let serving = Arc::clone(&served);
452 let events = fixture.intercept(move |source, event| match event {
453 Event::Met(hello, line) => assert!(serving.admit(source, hello, line)),
454 Event::Frame { from, kind, body } => serving.queue(source, from, *kind, body),
455 Event::Left(hello) => serving.forget(&hello.peer, Some((source, hello))),
456 _ => {}
457 });
458 let mut host = Hello::new("Old route".into(), None).unwrap();
459 host.serves = Some(fixture.guest.share);
460 let (mut old, mut send, _) = fixture.open(&host, false);
461 observed(&events, kind::HELLO);
462 let (gate, copied, resume) = barrier();
463 fixture.releases.push(resume.clone());
464 *fixture.pipes[0].body.lock().unwrap() = Some(gate);
465 send.send(
466 &mut old,
467 kind::PUT,
468 &Request {
469 id: 80,
470 handle: Some(1),
471 offset: Some(0),
472 bytes: Some(vec![1, 2, 3]),
473 ..Request::default()
474 },
475 )
476 .unwrap();
477 copied.recv_timeout(OPENING).unwrap();
478 host.name = "Promoted route".into();
479 let (mut new, mut send, mut receive) = fixture.open(&host, true);
480 observed(&events, kind::HELLO);
481 resume.send(()).unwrap();
482 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
483 send.send(
484 &mut new,
485 kind::PUT,
486 &Request {
487 id: 81,
488 handle: Some(2),
489 offset: Some(0),
490 bytes: Some(vec![4, 5, 6]),
491 ..Request::default()
492 },
493 )
494 .unwrap();
495 let reply = loop {
496 let (kind, body) = receive.receive(&mut new).unwrap();
497 if kind == kind::REPLY {
498 break minicbor::decode::<Reply>(&body).unwrap();
499 }
500 };
501 let puts = served
502 .puts
503 .lock()
504 .unwrap()
505 .keys()
506 .copied()
507 .collect::<Vec<_>>();
508 served.guests.lock().unwrap().clear();
509 drop(fixture);
510 assert_eq!(
511 (reply.id, reply.failure, puts),
512 (81, None, vec![(host.peer, 2)]),
513 "retired request reached the promoted guest or current request was lost"
514 );
515}
516
517#[test]
518fn handover_old_left_preserves_promoted_host_uploads_and_snapshots() {
519 let (_directory, served) = served();
520 let mut fixture = Handover::new(None);
521 let (gate, removed, resume) = barrier();
522 fixture.releases.push(resume.clone());
523 let gate = Mutex::new(Some(gate));
524 let serving = Arc::clone(&served);
525 let events = fixture.intercept(move |source, event| match event {
526 Event::Met(hello, line) => assert!(serving.admit(source, hello, line)),
527 Event::Left(hello) => {
528 if let Some(gate) = gate.lock().unwrap().take() {
529 gate.wait().unwrap();
530 }
531 serving.forget(&hello.peer, Some((source, hello)));
532 }
533 _ => {}
534 });
535 let mut host = Hello::new("Old route".into(), None).unwrap();
536 host.serves = Some(fixture.guest.share);
537 let (old, _, _) = fixture.open(&host, false);
538 observed(&events, kind::HELLO);
539 old.shutdown(Shutdown::Both).unwrap();
540 removed.recv_timeout(OPENING).unwrap();
541 host.name = "Promoted route".into();
542 let (_new, _, _) = fixture.open(&host, true);
543 observed(&events, kind::HELLO);
544 served
545 .puts
546 .lock()
547 .unwrap()
548 .insert((host.peer, 90), vec![7, 8, 9]);
549 served
550 .snapshots
551 .lock()
552 .unwrap()
553 .insert((host.peer, 91), (Arc::new(vec![10, 11]), Instant::now()));
554 resume.send(()).unwrap();
555 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
556 let guest_kept = served
557 .guests
558 .lock()
559 .unwrap()
560 .get(&host.peer)
561 .is_some_and(|guest| guest.hello.name == host.name);
562 let upload_kept = served.puts.lock().unwrap().get(&(host.peer, 90)) == Some(&vec![7, 8, 9]);
563 let snapshot_kept = served
564 .snapshots
565 .lock()
566 .unwrap()
567 .get(&(host.peer, 91))
568 .is_some_and(|(bytes, _)| **bytes == vec![10, 11]);
569 served.guests.lock().unwrap().clear();
570 drop(fixture);
571 assert!(
572 guest_kept && upload_kept && snapshot_kept,
573 "retired Left removed promoted host resources"
574 );
575}
576
577#[test]
578fn handover_deferred_presence_cannot_replace_promoted_host_room() {
579 let mut fixture = Handover::new(None);
580 let mut host = Hello::new("Old host route".into(), None).unwrap();
581 host.serves = Some(fixture.guest.share);
582 let (_old, _, _) = fixture.open(&host, false);
583 fixture.heard();
584 let from = Arc::clone(&fixture.guest.host.lock().unwrap().as_ref().unwrap().0);
585 let secret = [61; 16];
586 let guest = Arc::clone(&fixture.guest);
587 let source = Arc::clone(&fixture.shared);
588 let here_owner = Arc::clone(&fixture.guest);
589 let here = here_owner.here.lock().unwrap();
590 let (entered, started) = mpsc::channel();
591 let (finished, done) = mpsc::channel();
592 let installer = thread::spawn(move || {
593 entered.send(()).unwrap();
594 guest.join_presence(&source, &from, secret);
595 let _ = finished.send(());
596 true
597 });
598 fixture.readers.push((done, installer));
599 started.recv_timeout(OPENING).unwrap();
600 host.name = "Promoted host route".into();
601 let (_new, _, _) = fixture.open(&host, true);
602 fixture.heard();
603 drop(here);
604 fixture.readers[1].0.recv_timeout(OPENING).unwrap();
605 let refused = fixture.guest.presence.lock().unwrap().is_none();
606 let promoted = fixture
607 .guest
608 .host
609 .lock()
610 .unwrap()
611 .as_ref()
612 .is_some_and(|(hello, _)| hello.name == host.name);
613 drop(fixture);
614 assert!(
615 refused && promoted,
616 "deferred old WELCOME installed its presence room"
617 );
618}
619
620#[test]
621fn handover_presence_install_seeds_latest_here() {
622 let mut fixture = Handover::new(None);
623 let mut host = Hello::new("Current host".into(), None).unwrap();
624 host.serves = Some(fixture.guest.share);
625 let (_remote, _, _) = fixture.open(&host, true);
626 fixture.heard();
627 let from = Arc::clone(&fixture.guest.host.lock().unwrap().as_ref().unwrap().0);
628 let secret = [62; 16];
629 let guest = Arc::clone(&fixture.guest);
630 let source = Arc::clone(&fixture.shared);
631 let mut here = fixture.guest.here.lock().unwrap();
632 let latest = Presence {
633 section: Some([63; 16]),
634 ..Presence::default()
635 };
636 let (entered, started) = mpsc::channel();
637 let (finished, done) = mpsc::channel();
638 let installer = thread::spawn(move || {
639 entered.send(()).unwrap();
640 guest.join_presence(&source, &from, secret);
641 let _ = finished.send(());
642 true
643 });
644 fixture.readers.push((done, installer));
645 started.recv_timeout(OPENING).unwrap();
646 *here = latest.clone();
647 drop(here);
648 fixture.readers[1].0.recv_timeout(OPENING).unwrap();
649 let seeded = fixture
650 .guest
651 .presence
652 .lock()
653 .unwrap()
654 .as_ref()
655 .is_some_and(|(held, live)| {
656 *held == secret && live.shared.state.lock().unwrap().presence == latest
657 });
658 let installed = fixture.guest.presence.lock().unwrap().take();
659 drop(installed);
660 drop(fixture);
661 assert!(
662 seeded,
663 "presence installation overwrote current here with a stale seed"
664 );
665}
666
667pub(super) fn promoted_routes<T>(
668 room: Room,
669 before: impl FnOnce(&Arc<Shared>, &Arc<Hello>, &Line),
670 check: impl FnOnce(&Arc<Shared>, &Arc<Hello>, &Line, &Arc<Hello>, &Line) -> T,
671) -> T {
672 let mut fixture = Handover::new(None);
673 let shared = Arc::get_mut(&mut fixture.shared).unwrap();
674 shared.secret = room.secret();
675 shared.room = room;
676 let host = Hello::new("Pairing device".into(), None).unwrap();
677 let (_old, _, _) = fixture.open(&host, false);
678 fixture.heard();
679 let (old, old_line) = {
680 let state = fixture.shared.state.lock().unwrap();
681 let link = &state.peers[&host.peer];
682 (Arc::clone(&link.peer.hello), link.line.clone())
683 };
684 before(&fixture.shared, &old, &old_line);
685 let (_new, _, _) = fixture.open(&host, true);
686 fixture.heard();
687 let (new, new_line) = {
688 let state = fixture.shared.state.lock().unwrap();
689 let link = &state.peers[&host.peer];
690 (Arc::clone(&link.peer.hello), link.line.clone())
691 };
692 let result = check(&fixture.shared, &old, &old_line, &new, &new_line);
693 drop((old_line, new_line));
694 drop(fixture);
695 result
696}
697
698#[test]
699fn handover_pair_pending_keeps_promoted_identity_and_rejects_retired_allow() {
700 let directory = tempfile::tempdir().unwrap();
701 let folder = directory.path().join("notebook");
702 std::fs::create_dir(&folder).unwrap();
703 let sharing = Sharing {
704 approve: true,
705 code: super::super::super::code::format(701, "ABCDEF").unwrap(),
706 ..Sharing::new("").unwrap()
707 };
708 let room = Room::share(&sharing.code, &sharing.password);
709 let host = Host::start(
710 crate::session::Notebook::open(&folder, directory.path().join("cache"))
711 .unwrap()
712 .into_storage(),
713 Hello::new("Notebook owner".into(), None).unwrap(),
714 sharing,
715 "Fixture",
716 None,
717 None,
718 || {},
719 |_| Ok(()),
720 )
721 .unwrap();
722 let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared);
723 let valid = promoted_routes(
724 room,
725 |source, old, old_line| {
726 (callback.events)(source, Event::Met(old, old_line));
727 },
728 |source, old, old_line, new, new_line| {
729 host.allow(&old.peer).unwrap();
730 let no_grant = host.sharing().members.is_empty();
731 (callback.events)(source, Event::Met(new, new_line));
732 (callback.events)(source, Event::Met(old, old_line));
733 (callback.events)(source, Event::Left(old));
734 let kept = host
735 .requests()
736 .first()
737 .is_some_and(|request| Arc::ptr_eq(request, new));
738 host.stop();
739 kept && no_grant
740 },
741 );
742 assert!(
743 valid,
744 "pairing admitted a retired route or retired Left removed the promoted request"
745 );
746}
747
748#[test]
749fn handover_pair_retired_met_cannot_persist_another_credential() {
750 let directory = tempfile::tempdir().unwrap();
751 let folder = directory.path().join("notebook");
752 std::fs::create_dir(&folder).unwrap();
753 let sharing = Sharing {
754 approve: false,
755 code: super::super::super::code::format(702, "ABCDEF").unwrap(),
756 ..Sharing::new("").unwrap()
757 };
758 let room = Room::share(&sharing.code, &sharing.password);
759 let kept = Arc::new(AtomicUsize::new(0));
760 let count = Arc::clone(&kept);
761 let host = Host::start(
762 crate::session::Notebook::open(&folder, directory.path().join("cache"))
763 .unwrap()
764 .into_storage(),
765 Hello::new("Notebook owner".into(), None).unwrap(),
766 sharing,
767 "Fixture",
768 None,
769 None,
770 || {},
771 move |_| {
772 count.fetch_add(1, Ordering::SeqCst);
773 Ok(())
774 },
775 )
776 .unwrap();
777 let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared);
778 let valid = promoted_routes(
779 room,
780 |_, _, _| {},
781 |source, old, old_line, new, new_line| {
782 (callback.events)(source, Event::Met(new, new_line));
783 let after_current = kept.load(Ordering::SeqCst);
784 (callback.events)(source, Event::Met(old, old_line));
785 let refused = kept.load(Ordering::SeqCst) == after_current
786 && after_current == 2
787 && host.sharing().members.len() == 1;
788 host.stop();
789 refused
790 },
791 );
792 assert!(valid, "retired pairing Met persisted another credential");
793}
794
795#[test]
796fn handover_retired_membership_rejects_deferred_access_met() {
797 let directory = tempfile::tempdir().unwrap();
798 let folder = directory.path().join("notebook");
799 std::fs::create_dir(&folder).unwrap();
800 let secret = [71; 16];
801 let sharing = Sharing {
802 code: super::super::super::code::format(703, "ABCDEF").unwrap(),
803 members: vec![Device {
804 secret,
805 name: "Retired device".into(),
806 device: None,
807 }],
808 ..Sharing::new("").unwrap()
809 };
810 let host = Host::start(
811 crate::session::Notebook::open(&folder, directory.path().join("cache"))
812 .unwrap()
813 .into_storage(),
814 Hello::new("Notebook owner".into(), None).unwrap(),
815 sharing,
816 "Fixture",
817 None,
818 None,
819 || {},
820 |_| Ok(()),
821 )
822 .unwrap();
823 let callback = Arc::clone(&host.members.access.lock().unwrap()[&secret].shared);
824 let refused = promoted_routes(
825 Room::Notebook(secret),
826 |_, _, _| {},
827 |source, _, _, current, line| {
828 host.remove(&secret).unwrap();
829 (callback.events)(source, Event::Met(current, line));
830 let refused = host.members.served.guests.lock().unwrap().is_empty()
831 && host.sharing().members.is_empty();
832 host.stop();
833 refused
834 },
835 );
836 assert!(
837 refused,
838 "deferred access callback readmitted a removed device"
839 );
840}