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