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