1use std::collections::HashMap;
12use std::net::IpAddr;
13use std::sync::LazyLock;
14use std::time::{Duration, Instant};
15
16use axum::extract::RawQuery;
17use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
18use axum::http::StatusCode;
19use axum::response::{IntoResponse, Response};
20use koan_core::remote::pair::PairMessage;
21use parking_lot::Mutex;
22use tokio::sync::oneshot;
23
24use crate::auth::password::FailureLimiter;
25use crate::auth::routes::{RateLimiter, network};
26
27pub const TTL: Duration = Duration::from_secs(600);
29const MAX_PENDING: usize = 256;
31const MAX_PENDING_PER_ADDRESS: usize = 3;
35const CODE_ALPHABET: &[u8; 32] = b"0123456789ABCDEFGHJKMNPQRSTVWXYZ";
38const CODE_LEN: usize = 8;
39const KEEPALIVE: Duration = Duration::from_secs(30);
42const MISSES_PER_MINUTE: u32 = 10;
46const DELIVERY: Duration = Duration::from_secs(10);
49
50struct Entry {
51 code: String,
53 device: String,
54 from: IpAddr,
56 created: Instant,
57 outcome: oneshot::Sender<Settled>,
58}
59
60pub struct Settled {
62 pub message: PairMessage,
63 sent: Option<std::sync::mpsc::SyncSender<bool>>,
65}
66
67impl From<PairMessage> for Settled {
68 fn from(message: PairMessage) -> Self {
69 Self {
70 message,
71 sent: None,
72 }
73 }
74}
75
76pub struct Pairings {
77 ttl: Duration,
78 keepalive: Duration,
79 entries: Mutex<HashMap<String, Entry>>,
80 misses: FailureLimiter<i64>,
82}
83
84pub fn pairings() -> &'static Pairings {
87 static PAIRINGS: LazyLock<Pairings> = LazyLock::new(|| Pairings::new(ttl_from_env()));
88 &PAIRINGS
89}
90
91fn ttl_from_env() -> Duration {
94 std::env::var("KOAN_PAIR_TTL_SECS")
95 .ok()
96 .and_then(|s| s.trim().parse().ok())
97 .filter(|&secs| secs > 0)
98 .map_or(TTL, Duration::from_secs)
99}
100
101static OPENS: LazyLock<RateLimiter> = LazyLock::new(|| RateLimiter::new(60, 10));
104
105pub struct Opened<'a> {
108 pub id: String,
109 pub code: String,
111 outcome: oneshot::Receiver<Settled>,
112 pairings: &'a Pairings,
113}
114
115#[cfg(test)]
116impl Opened<'static> {
117 pub(crate) fn device(mut self, sent: bool) -> std::thread::JoinHandle<PairMessage> {
120 std::thread::spawn(move || {
121 let outcome = std::mem::replace(&mut self.outcome, oneshot::channel().1);
123 let settled = outcome.blocking_recv().unwrap();
124 if let Some(done) = settled.sent {
125 done.send(sent).unwrap();
126 }
127 settled.message
128 })
129 }
130}
131
132impl Drop for Opened<'_> {
133 fn drop(&mut self) {
134 self.pairings.entries.lock().remove(&self.id);
135 }
136}
137
138#[derive(Debug, Clone, PartialEq, Eq)]
140pub struct PairInfo {
141 pub device: String,
142 pub from: IpAddr,
143}
144
145impl PairInfo {
146 pub fn local(&self) -> bool {
149 is_local(self.from)
150 }
151}
152
153fn is_local(ip: IpAddr) -> bool {
156 match ip.to_canonical() {
157 IpAddr::V4(v4) => {
158 let [a, b, ..] = v4.octets();
159 v4.is_private()
160 || (a == 100 && (b & 0xc0) == 64)
161 || v4.is_link_local()
162 || v4.is_loopback()
163 }
164 IpAddr::V6(v6) => {
165 v6.is_loopback()
166 || (v6.segments()[0] & 0xfe00) == 0xfc00
167 || (v6.segments()[0] & 0xffc0) == 0xfe80
168 }
169 }
170}
171
172pub struct Taken {
174 id: String,
175 entry: Entry,
176}
177
178impl Taken {
179 pub fn device(&self) -> &str {
180 &self.entry.device
181 }
182
183 pub fn info(&self) -> PairInfo {
184 PairInfo {
185 device: self.entry.device.clone(),
186 from: self.entry.from,
187 }
188 }
189
190 pub fn settle(self, outcome: PairMessage) -> Result<(), PairMessage> {
192 self.entry
193 .outcome
194 .send(outcome.into())
195 .map_err(|settled| settled.message)
196 }
197}
198
199#[derive(Debug, PartialEq, Eq)]
200pub enum OpenError {
201 Full,
202 Busy,
204 Entropy,
205}
206
207impl Pairings {
208 fn new(ttl: Duration) -> Self {
209 Self {
210 ttl,
211 keepalive: KEEPALIVE,
212 entries: Mutex::default(),
213 misses: FailureLimiter::new(MISSES_PER_MINUTE),
214 }
215 }
216
217 fn silence(&self) -> Duration {
220 self.keepalive * 5 / 2
221 }
222
223 pub fn open(&self, device: &str, from: IpAddr) -> Result<Opened<'_>, OpenError> {
225 let id = koan_core::auth::random_api_key().map_err(|_| OpenError::Entropy)?;
226 let (tx, rx) = oneshot::channel();
227 let mut entries = self.entries.lock();
228 self.sweep(&mut entries);
229 if entries.len() >= MAX_PENDING {
230 return Err(OpenError::Full);
231 }
232 let from = from.to_canonical();
233 let waiting = entries
234 .values()
235 .filter(|e| network(e.from) == network(from))
236 .count();
237 if waiting >= MAX_PENDING_PER_ADDRESS {
238 return Err(OpenError::Busy);
239 }
240 let code = loop {
241 let code = new_code()?;
242 if !entries.values().any(|e| e.code == code) {
243 break code;
244 }
245 };
246 entries.insert(
247 id.clone(),
248 Entry {
249 code: code.clone(),
250 device: koan_core::invite::device_name(device),
251 from,
252 created: Instant::now(),
253 outcome: tx,
254 },
255 );
256 Ok(Opened {
257 id,
258 code: format!("{}-{}", &code[..4], &code[4..]),
259 outcome: rx,
260 pairings: self,
261 })
262 }
263
264 pub fn info(&self, pair: &str, caller: i64) -> Result<PairInfo, PairError> {
267 let mut entries = self.entries.lock();
268 self.sweep(&mut entries);
269 let id = self.lookup(&entries, pair, caller)?;
270 let e = &entries[&id];
271 Ok(PairInfo {
272 device: e.device.clone(),
273 from: e.from,
274 })
275 }
276
277 pub fn take(&self, pair: &str, caller: i64) -> Result<Taken, PairError> {
280 let mut entries = self.entries.lock();
281 self.sweep(&mut entries);
282 let id = self.lookup(&entries, pair, caller)?;
283 let entry = entries.remove(&id).ok_or(PairError::NotFound)?;
284 Ok(Taken { id, entry })
285 }
286
287 fn lookup(
292 &self,
293 entries: &HashMap<String, Entry>,
294 pair: &str,
295 caller: i64,
296 ) -> Result<String, PairError> {
297 let pair = pair.trim();
298 if entries.contains_key(pair) {
299 return Ok(pair.to_owned());
300 }
301 if self.misses.exhausted(&caller) {
302 return Err(PairError::Limited);
303 }
304 find(entries, pair).ok_or_else(|| {
305 self.misses.record(caller);
306 PairError::NotFound
307 })
308 }
309
310 pub fn put_back(&self, taken: Taken) {
312 if !taken.entry.outcome.is_closed() {
313 self.entries.lock().insert(taken.id, taken.entry);
314 }
315 }
316
317 pub fn settle(
321 &self,
322 conn: &rusqlite::Connection,
323 pair: &str,
324 user_id: i64,
325 username: &str,
326 decline: bool,
327 ) -> Result<PairInfo, PairError> {
328 use koan_core::db::queries::api_keys;
329 let taken = self.take(pair, user_id)?;
330 let info = taken.info();
331 let device = info.device.clone();
332 if decline {
333 let _ = taken.settle(PairMessage::Declined);
334 log::info!("pair: {device} declined by {username}");
335 return Ok(info);
336 }
337 let api_key = match api_keys::create_api_key(conn, user_id, &device) {
338 Ok((_, key)) => key,
339 Err(e) => {
340 self.put_back(taken);
341 return Err(PairError::Internal(e.to_string()));
342 }
343 };
344 let approved = PairMessage::Approved {
345 username: username.to_owned(),
346 api_key: api_key.clone(),
347 };
348 let (sent_tx, sent_rx) = std::sync::mpsc::sync_channel(1);
351 let settled = Settled {
352 message: approved,
353 sent: Some(sent_tx),
354 };
355 let sent = taken.entry.outcome.send(settled).is_ok()
356 && sent_rx.recv_timeout(DELIVERY).unwrap_or(false);
357 if !sent {
358 let _ = api_keys::revoke_api_key_value(conn, &api_key);
361 log::warn!("pair: {device} went before its key reached it");
362 return Err(PairError::NotFound);
363 }
364 log::info!("pair: {device} signed in as {username}");
365 Ok(info)
366 }
367
368 fn sweep(&self, entries: &mut HashMap<String, Entry>) {
370 let lapsed: Vec<String> = entries
371 .iter()
372 .filter(|(_, e)| e.created.elapsed() >= self.ttl)
373 .map(|(id, _)| id.clone())
374 .collect();
375 for id in lapsed {
376 if let Some(e) = entries.remove(&id) {
377 let _ = e.outcome.send(PairMessage::Expired.into());
378 }
379 }
380 }
381}
382
383#[derive(Debug, PartialEq, Eq)]
384pub enum PairError {
385 NotFound,
387 Limited,
389 Internal(String),
390}
391
392fn new_code() -> Result<String, OpenError> {
393 let mut bytes = [0u8; CODE_LEN];
394 getrandom::fill(&mut bytes).map_err(|_| OpenError::Entropy)?;
395 Ok(bytes
397 .iter()
398 .map(|b| CODE_ALPHABET[(b % 32) as usize] as char)
399 .collect())
400}
401
402fn normalise_code(typed: &str) -> Option<String> {
405 let code: String = typed
406 .chars()
407 .filter(|c| !matches!(c, '-' | ' '))
408 .map(|c| match c.to_ascii_uppercase() {
409 'I' | 'L' => '1',
410 'O' => '0',
411 c => c,
412 })
413 .collect();
414 (code.len() == CODE_LEN && code.bytes().all(|b| CODE_ALPHABET.contains(&b))).then_some(code)
415}
416
417fn find(entries: &HashMap<String, Entry>, pair: &str) -> Option<String> {
418 let pair = pair.trim();
419 if entries.contains_key(pair) {
420 return Some(pair.to_owned());
421 }
422 let code = normalise_code(pair)?;
423 entries
424 .iter()
425 .find(|(_, e)| e.code == code)
426 .map(|(id, _)| id.clone())
427}
428
429pub(crate) async fn route(
432 RawQuery(raw): RawQuery,
433 ws: WebSocketUpgrade,
434 request: axum::extract::Request,
435) -> Response {
436 if request.headers().contains_key(axum::http::header::ORIGIN) {
441 return (StatusCode::FORBIDDEN, "pairing is not open to web pages").into_response();
442 }
443 let from = crate::auth::routes::client_ip(&request);
444 if !OPENS.allow(from) {
445 return (StatusCode::TOO_MANY_REQUESTS, "too many pairings").into_response();
446 }
447 let name = form_urlencoded::parse(raw.unwrap_or_default().as_bytes())
448 .find(|(k, _)| k == "name")
449 .map(|(_, v)| v.into_owned())
450 .unwrap_or_default();
451 match pairings().open(&name, from) {
452 Ok(opened) => ws.on_upgrade(move |socket| session(socket, opened)),
453 Err(OpenError::Full) => {
454 (StatusCode::SERVICE_UNAVAILABLE, "too many pairings waiting").into_response()
455 }
456 Err(OpenError::Busy) => (
457 StatusCode::TOO_MANY_REQUESTS,
458 "too many pairings waiting from this address",
459 )
460 .into_response(),
461 Err(OpenError::Entropy) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
462 }
463}
464
465async fn session(mut socket: WebSocket, mut opened: Opened<'static>) {
468 let waited = wait(&mut socket, &mut opened).await;
469 opened.outcome.close();
473 let late = opened.outcome.try_recv().ok();
474 let Some(waited) = waited else { return };
475 let settled = late.unwrap_or(waited);
477 log::info!(
478 "pair: {} {}",
479 opened.code,
480 match settled.message {
481 PairMessage::Approved { .. } => "approved",
482 PairMessage::Declined => "declined",
483 _ => "expired",
484 }
485 );
486 let sent = send(&mut socket, &settled.message).await.is_ok();
487 if let Some(done) = settled.sent {
488 let _ = done.send(sent);
489 }
490 let _ = socket.send(Message::Close(None)).await;
491}
492
493async fn wait(socket: &mut WebSocket, opened: &mut Opened<'static>) -> Option<Settled> {
496 let pending = PairMessage::Pending {
497 id: opened.id.clone(),
498 code: opened.code.clone(),
499 expires_in: opened.pairings.ttl.as_secs(),
500 };
501 send(socket, &pending).await.ok()?;
502 log::info!("pair: {} waiting", opened.code);
503 let lapse = tokio::time::sleep(opened.pairings.ttl);
504 tokio::pin!(lapse);
505 let keepalive = opened.pairings.keepalive;
506 let mut ping = tokio::time::interval_at(tokio::time::Instant::now() + keepalive, keepalive);
507 let silence = opened.pairings.silence();
508 let quiet = tokio::time::sleep(silence);
509 tokio::pin!(quiet);
510 loop {
511 tokio::select! {
512 outcome = &mut opened.outcome => {
514 return Some(outcome.unwrap_or_else(|_| PairMessage::Expired.into()));
515 }
516 () = &mut lapse => return Some(PairMessage::Expired.into()),
517 () = &mut quiet => {
518 log::info!("pair: {} stopped answering", opened.code);
519 return None;
520 }
521 _ = ping.tick() => socket.send(Message::Ping(Default::default())).await.ok()?,
522 msg = socket.recv() => match msg {
523 Some(Ok(Message::Close(_)) | Err(_)) | None => return None,
524 Some(Ok(_)) => quiet.as_mut().reset(tokio::time::Instant::now() + silence),
525 },
526 }
527 }
528}
529
530async fn send(socket: &mut WebSocket, message: &PairMessage) -> Result<(), axum::Error> {
531 let text = serde_json::to_string(message).unwrap_or_default();
532 socket.send(Message::Text(text.into())).await
533}
534
535#[cfg(test)]
536mod tests {
537 use super::*;
538
539 const ME: i64 = 1;
541 const LAN: IpAddr = IpAddr::V4(std::net::Ipv4Addr::new(192, 168, 1, 20));
542
543 fn host(i: usize) -> IpAddr {
545 IpAddr::V4(std::net::Ipv4Addr::new(
546 10,
547 9,
548 (i / 256) as u8,
549 (i % 256) as u8,
550 ))
551 }
552
553 fn leaked(ttl: Duration) -> &'static Pairings {
554 Box::leak(Box::new(Pairings::new(ttl)))
555 }
556
557 #[test]
558 fn codes_are_crockford_and_dashed() {
559 let p = leaked(TTL);
560 for _ in 0..50 {
561 let o = p.open("tv", LAN).unwrap();
562 let (a, b) = o.code.split_once('-').unwrap();
563 assert_eq!((a.len(), b.len()), (4, 4));
564 assert!(
565 format!("{a}{b}")
566 .bytes()
567 .all(|c| CODE_ALPHABET.contains(&c) && !b"ILOU".contains(&c))
568 );
569 }
570 }
571
572 #[test]
573 fn ids_are_unguessable_and_distinct() {
574 let p = leaked(TTL);
575 let opened: Vec<_> = (0..100).map(|i| p.open("tv", host(i)).unwrap()).collect();
576 let ids: std::collections::HashSet<_> = opened.iter().map(|o| o.id.clone()).collect();
577 assert_eq!(ids.len(), 100);
578 assert!(opened.iter().all(|o| o.id.len() == 43));
579 let codes: std::collections::HashSet<_> = opened.iter().map(|o| o.code.clone()).collect();
580 assert_eq!(codes.len(), 100);
581 }
582
583 #[test]
584 fn a_code_is_found_however_it_is_typed() {
585 let p = leaked(TTL);
586 let o = p.open("Living room\u{7} TV", LAN).unwrap();
587 assert_eq!(
588 p.info(&o.id, ME).ok().map(|i| i.device).as_deref(),
589 Some("Living room TV")
590 );
591 let bare = o.code.replace('-', "");
592 for typed in [
593 o.code.clone(),
594 o.code.to_lowercase(),
595 bare.clone(),
596 format!(" {} {} ", &bare[..4], &bare[4..]),
597 ] {
598 assert_eq!(
599 p.info(&typed, ME).ok().map(|i| i.device).as_deref(),
600 Some("Living room TV"),
601 "{typed}"
602 );
603 }
604 assert_eq!(normalise_code("o1l1-IOAB").as_deref(), Some("011110AB"));
605 assert_eq!(normalise_code("ABCD-EFGU"), None);
606 assert_eq!(normalise_code("ABC"), None);
607 }
608
609 #[tokio::test]
610 async fn approving_delivers_the_key_to_the_waiting_device() {
611 let p = leaked(TTL);
612 let mut o = p.open("tv", LAN).unwrap();
613 let taken = p.take(&o.code, ME).unwrap();
614 assert_eq!(taken.device(), "tv");
615 assert!(p.take(&o.id, ME).is_err());
617 taken
618 .settle(PairMessage::Approved {
619 username: "alice".into(),
620 api_key: "key".into(),
621 })
622 .unwrap();
623 assert_eq!(
624 (&mut o.outcome).await.unwrap().message,
625 PairMessage::Approved {
626 username: "alice".into(),
627 api_key: "key".into(),
628 }
629 );
630 }
631
632 fn database() -> (tempfile::TempDir, koan_core::db::connection::Database, i64) {
633 let dir = tempfile::tempdir().unwrap();
634 let db = koan_core::db::connection::Database::open(&dir.path().join("koan.db")).unwrap();
635 let id = koan_core::db::queries::auth::create_user(
636 &db.conn,
637 "alice",
638 "hunter2",
639 koan_core::auth::Role::Readonly,
640 )
641 .unwrap();
642 (dir, db, id)
643 }
644
645 #[test]
646 fn approving_mints_a_key_for_the_approver() {
647 let (_dir, db, alice) = database();
648 let p = leaked(TTL);
649 let o = p.open("Living room TV", LAN).unwrap();
650 let (id, code) = (o.id.clone(), o.code.clone());
651 let device = o.device(true);
652 let settled = p.settle(&db.conn, &code, alice, "alice", false).unwrap();
653 assert_eq!(
654 settled,
655 PairInfo {
656 device: "Living room TV".into(),
657 from: LAN,
658 }
659 );
660 let PairMessage::Approved { username, api_key } = device.join().unwrap() else {
661 panic!("not approved");
662 };
663 assert_eq!(username, "alice");
664 let user = koan_core::db::queries::api_keys::authenticate_api_key(&db.conn, &api_key)
665 .unwrap()
666 .unwrap();
667 assert_eq!(user.id, alice);
668 let keys = koan_core::db::queries::api_keys::list_api_keys(&db.conn, Some(alice)).unwrap();
669 assert_eq!(keys[0].name, "Living room TV");
670 assert_eq!(
672 p.settle(&db.conn, &id, alice, "alice", false),
673 Err(PairError::NotFound)
674 );
675 }
676
677 #[test]
678 fn a_key_the_socket_would_not_take_is_revoked() {
679 let (_dir, db, alice) = database();
680 let p = leaked(TTL);
681 let o = p.open("tv", LAN).unwrap();
682 let id = o.id.clone();
683 let device = o.device(false);
684 assert_eq!(
686 p.settle(&db.conn, &id, alice, "alice", false),
687 Err(PairError::NotFound)
688 );
689 let PairMessage::Approved { api_key, .. } = device.join().unwrap() else {
690 panic!("not approved");
691 };
692 assert!(
693 koan_core::db::queries::api_keys::authenticate_api_key(&db.conn, &api_key)
694 .unwrap()
695 .is_none()
696 );
697 let keys = koan_core::db::queries::api_keys::list_api_keys(&db.conn, Some(alice)).unwrap();
698 assert!(keys.is_empty());
699 }
700
701 #[tokio::test]
702 async fn declining_mints_nothing() {
703 let (_dir, db, alice) = database();
704 let p = leaked(TTL);
705 let mut o = p.open("tv", LAN).unwrap();
706 p.settle(&db.conn, &o.id, alice, "alice", true).unwrap();
707 assert_eq!(
708 (&mut o.outcome).await.unwrap().message,
709 PairMessage::Declined
710 );
711 let keys = koan_core::db::queries::api_keys::list_api_keys(&db.conn, Some(alice)).unwrap();
712 assert!(keys.is_empty());
713 }
714
715 #[test]
716 fn a_key_for_a_gone_device_is_revoked() {
717 let (_dir, db, alice) = database();
718 let p = leaked(TTL);
719 let o = p.open("tv", LAN).unwrap();
720 let id = o.id.clone();
721 let taken = p.take(&id, ME).unwrap();
723 drop(o);
724 p.entries.lock().insert(taken.id, taken.entry);
725 assert_eq!(
726 p.settle(&db.conn, &id, alice, "alice", false),
727 Err(PairError::NotFound)
728 );
729 let keys = koan_core::db::queries::api_keys::list_api_keys(&db.conn, Some(alice)).unwrap();
730 assert!(keys.is_empty());
731 }
732
733 #[test]
734 fn approving_an_unknown_pairing_is_not_found() {
735 let (_dir, db, alice) = database();
736 let p = leaked(TTL);
737 assert_eq!(
738 p.settle(&db.conn, "ABCD-EFGH", alice, "alice", false),
739 Err(PairError::NotFound)
740 );
741 }
742
743 #[test]
744 fn a_gone_device_gives_the_outcome_back() {
745 let p = leaked(TTL);
746 let o = p.open("tv", LAN).unwrap();
747 let taken = p.take(&o.id, ME).unwrap();
748 drop(o);
749 assert!(taken.settle(PairMessage::Declined).is_err());
750 }
751
752 #[tokio::test]
753 async fn a_lapsed_pairing_is_expired_and_unknown() {
754 let p = leaked(Duration::ZERO);
755 let mut o = p.open("tv", LAN).unwrap();
756 assert!(p.info(&o.id, ME).is_err());
757 assert!(p.take(&o.code, ME).is_err());
758 assert_eq!(
759 (&mut o.outcome).await.unwrap().message,
760 PairMessage::Expired
761 );
762 }
763
764 #[test]
765 fn the_requesting_address_is_kept() {
766 let p = leaked(TTL);
767 let mapped: IpAddr = "::ffff:203.0.113.9".parse().unwrap();
768 let o = p.open("tv", mapped).unwrap();
769 let info = p.info(&o.code, ME).unwrap();
770 assert_eq!(info.from.to_string(), "203.0.113.9");
771 assert!(!info.local());
772 assert_eq!(p.take(&o.id, ME).unwrap().info(), info);
773 }
774
775 #[test]
776 fn private_addresses_are_local() {
777 for local in [
778 "10.1.2.3",
779 "172.16.0.1",
780 "192.168.1.20",
781 "169.254.10.1",
782 "127.0.0.1",
783 "100.64.0.1",
784 "100.101.102.103",
785 "100.127.255.254",
786 "::1",
787 "fd12:3456::1",
788 "fe80::1",
789 "::ffff:192.168.0.5",
790 ] {
791 assert!(is_local(local.parse().unwrap()), "{local}");
792 }
793 for public in [
794 "203.0.113.9",
795 "8.8.8.8",
796 "172.32.0.1",
797 "100.63.255.255",
798 "100.128.0.1",
799 "2001:db8::1",
800 "2a00:1450::1",
801 "::ffff:8.8.8.8",
802 ] {
803 assert!(!is_local(public.parse().unwrap()), "{public}");
804 }
805 }
806
807 #[test]
808 fn unknown_pairings_are_not_found() {
809 let p = leaked(TTL);
810 let o = p.open("tv", LAN).unwrap();
811 let other = if o.code == "0000-0000" {
812 "1111-1111"
813 } else {
814 "0000-0000"
815 };
816 assert!(p.info(other, ME).is_err());
817 assert!(p.take("not-a-pairing", ME).is_err());
818 assert!(p.info("", ME).is_err());
819 }
820
821 #[test]
822 fn misses_are_limited_per_account() {
823 let p = leaked(TTL);
824 let o = p.open("tv", LAN).unwrap();
825 let other = if o.code == "0000-0000" {
826 "1111-1111"
827 } else {
828 "0000-0000"
829 };
830 for _ in 0..MISSES_PER_MINUTE {
831 assert_eq!(p.info(other, ME).err(), Some(PairError::NotFound));
832 }
833 assert_eq!(p.info(&o.code, ME).err(), Some(PairError::Limited));
835 assert_eq!(p.info(other, ME).err(), Some(PairError::Limited));
836 assert!(p.info(&o.id, ME).is_ok());
838 for _ in 0..=MISSES_PER_MINUTE {
840 assert!(p.info(&o.code, ME + 1).is_ok());
841 }
842 }
843
844 async fn serve(p: &'static Pairings) -> std::net::SocketAddr {
846 let app = axum::Router::new().route(
847 "/",
848 axum::routing::get(move |ws: WebSocketUpgrade| async move {
849 let opened = p.open("tv", LAN).unwrap();
850 ws.on_upgrade(move |socket| session(socket, opened))
851 }),
852 );
853 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
854 let addr = listener.local_addr().unwrap();
855 tokio::spawn(async move { axum::serve(listener, app).await });
856 addr
857 }
858
859 fn quick(keepalive: Duration) -> &'static Pairings {
860 let mut p = Pairings::new(TTL);
861 p.keepalive = keepalive;
862 Box::leak(Box::new(p))
863 }
864
865 fn pending_id(
866 socket: &mut tungstenite::WebSocket<impl std::io::Read + std::io::Write>,
867 ) -> String {
868 let first = socket.read().unwrap();
869 match serde_json::from_str(first.to_text().unwrap()).unwrap() {
870 PairMessage::Pending { id, .. } => id,
871 other => panic!("{other:?}"),
872 }
873 }
874
875 #[tokio::test(flavor = "multi_thread")]
876 async fn a_device_that_stops_answering_is_dropped() {
877 let p = quick(Duration::from_secs(1));
880 let addr = serve(p).await;
881 let (socket, id) = tokio::task::spawn_blocking(move || {
883 let (mut socket, _) = tungstenite::connect(format!("ws://{addr}/")).unwrap();
884 let id = pending_id(&mut socket);
885 (socket, id)
886 })
887 .await
888 .unwrap();
889 assert!(p.info(&id, ME).is_ok());
890 tokio::time::sleep(p.silence() * 2).await;
891 assert_eq!(p.info(&id, ME).err(), Some(PairError::NotFound));
892 drop(socket);
893 }
894
895 #[tokio::test(flavor = "multi_thread")]
896 async fn a_device_that_answers_waits_and_gets_its_key() {
897 let (_dir, db, alice) = database();
898 let p = quick(Duration::from_secs(1));
899 let addr = serve(p).await;
900 let (id_tx, id_rx) = oneshot::channel();
901 let device = tokio::task::spawn_blocking(move || {
902 let (mut socket, _) = tungstenite::connect(format!("ws://{addr}/")).unwrap();
903 id_tx.send(pending_id(&mut socket)).unwrap();
904 loop {
906 if let tungstenite::Message::Text(text) = socket.read().unwrap() {
907 return serde_json::from_str::<PairMessage>(&text).unwrap();
908 }
909 }
910 });
911 let id = id_rx.await.unwrap();
912 tokio::time::sleep(p.silence() * 2).await;
913 let db = tokio::task::spawn_blocking(move || {
915 p.settle(&db.conn, &id, alice, "alice", false).unwrap();
916 db
917 })
918 .await
919 .unwrap();
920 let PairMessage::Approved { api_key, .. } = device.await.unwrap() else {
921 panic!("not approved");
922 };
923 assert!(
924 koan_core::db::queries::api_keys::authenticate_api_key(&db.conn, &api_key)
925 .unwrap()
926 .is_some()
927 );
928 }
929
930 #[test]
931 fn a_closed_socket_removes_its_pairing() {
932 let p = leaked(TTL);
933 let o = p.open("tv", LAN).unwrap();
934 let id = o.id.clone();
935 drop(o);
936 assert!(p.info(&id, ME).is_err());
937 }
938
939 #[test]
940 fn pairings_are_capped() {
941 let p = leaked(TTL);
942 let held: Vec<_> = (0..MAX_PENDING)
943 .map(|i| p.open("tv", host(i)).unwrap())
944 .collect();
945 assert_eq!(p.open("tv", LAN).err(), Some(OpenError::Full));
946 drop(held);
947 assert!(p.open("tv", LAN).is_ok());
948 }
949
950 #[test]
951 fn one_address_holds_three_at_most() {
952 let p = leaked(TTL);
953 let held: Vec<_> = (0..3).map(|_| p.open("tv", LAN).unwrap()).collect();
954 assert_eq!(p.open("tv", LAN).err(), Some(OpenError::Busy));
955 let mapped: IpAddr = "::ffff:192.168.1.20".parse().unwrap();
957 assert_eq!(p.open("tv", mapped).err(), Some(OpenError::Busy));
958 assert!(p.open("tv", host(1)).is_ok());
959
960 let v6: Vec<_> = ["2001:db8:1:2::1", "2001:db8:1:2::2", "2001:db8:1:2:ffff::9"]
962 .into_iter()
963 .map(|a| p.open("tv", a.parse().unwrap()).unwrap())
964 .collect();
965 assert_eq!(
966 p.open("tv", "2001:db8:1:2:abcd::1".parse().unwrap()).err(),
967 Some(OpenError::Busy)
968 );
969 assert!(p.open("tv", "2001:db8:1:3::1".parse().unwrap()).is_ok());
970
971 drop(held);
972 assert!(p.open("tv", LAN).is_ok());
973 drop(v6);
974 }
975}