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 axum::Extension(stop): axum::Extension<tokio_util::sync::CancellationToken>,
452 ws: WebSocketUpgrade,
453 request: axum::extract::Request,
454) -> Response {
455 if request.headers().contains_key(axum::http::header::ORIGIN) {
460 return (StatusCode::FORBIDDEN, "pairing is not open to web pages").into_response();
461 }
462 let from = crate::auth::routes::client_ip(&request);
463 if !OPENS.allow(from) {
464 return (StatusCode::TOO_MANY_REQUESTS, "too many pairings").into_response();
465 }
466 let query: Vec<(String, String)> = form_urlencoded::parse(raw.unwrap_or_default().as_bytes())
467 .into_owned()
468 .collect();
469 let param = |key: &str| {
470 query
471 .iter()
472 .find(|(k, _)| k == key)
473 .map(|(_, v)| v.as_str())
474 };
475 match pairings().open(param("name").unwrap_or_default(), param("device"), from) {
476 Ok(opened) => ws.on_upgrade(move |socket| session(socket, opened, stop)),
477 Err(OpenError::Full) => {
478 (StatusCode::SERVICE_UNAVAILABLE, "too many pairings waiting").into_response()
479 }
480 Err(OpenError::Busy) => (
481 StatusCode::TOO_MANY_REQUESTS,
482 "too many pairings waiting from this address",
483 )
484 .into_response(),
485 Err(OpenError::Entropy) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
486 }
487}
488
489async fn session(
492 mut socket: WebSocket,
493 mut opened: Opened<'static>,
494 stop: tokio_util::sync::CancellationToken,
495) {
496 let waited = tokio::select! {
497 waited = wait(&mut socket, &mut opened) => waited,
498 () = stop.cancelled() => None,
500 };
501 opened.outcome.close();
505 let late = opened.outcome.try_recv().ok();
506 let Some(waited) = waited else { return };
507 let settled = late.unwrap_or(waited);
509 log::info!(
510 "pair: {} {}",
511 opened.code,
512 match settled.message {
513 PairMessage::Approved { .. } => "approved",
514 PairMessage::Declined => "declined",
515 _ => "expired",
516 }
517 );
518 let sent = send(&mut socket, &settled.message).await.is_ok();
519 if let Some(done) = settled.sent {
520 let _ = done.send(sent);
521 }
522 let _ = socket.send(Message::Close(None)).await;
523}
524
525async fn wait(socket: &mut WebSocket, opened: &mut Opened<'static>) -> Option<Settled> {
528 let pending = PairMessage::Pending {
529 id: opened.id.clone(),
530 code: opened.code.clone(),
531 expires_in: opened.pairings.ttl.as_secs(),
532 };
533 send(socket, &pending).await.ok()?;
534 log::info!("pair: {} waiting", opened.code);
535 let lapse = tokio::time::sleep(opened.pairings.ttl);
536 tokio::pin!(lapse);
537 let keepalive = opened.pairings.keepalive;
538 let mut ping = tokio::time::interval_at(tokio::time::Instant::now() + keepalive, keepalive);
539 let silence = opened.pairings.silence();
540 let quiet = tokio::time::sleep(silence);
541 tokio::pin!(quiet);
542 loop {
543 tokio::select! {
544 outcome = &mut opened.outcome => {
546 return Some(outcome.unwrap_or_else(|_| PairMessage::Expired.into()));
547 }
548 () = &mut lapse => return Some(PairMessage::Expired.into()),
549 () = &mut quiet => {
550 log::info!("pair: {} stopped answering", opened.code);
551 return None;
552 }
553 _ = ping.tick() => socket.send(Message::Ping(Default::default())).await.ok()?,
554 msg = socket.recv() => match msg {
555 Some(Ok(Message::Close(_)) | Err(_)) | None => return None,
556 Some(Ok(_)) => quiet.as_mut().reset(tokio::time::Instant::now() + silence),
557 },
558 }
559 }
560}
561
562async fn send(socket: &mut WebSocket, message: &PairMessage) -> Result<(), axum::Error> {
563 let text = serde_json::to_string(message).unwrap_or_default();
564 socket.send(Message::Text(text.into())).await
565}
566
567#[cfg(test)]
568mod tests {
569 use super::*;
570
571 const ME: i64 = 1;
573 const LAN: IpAddr = IpAddr::V4(std::net::Ipv4Addr::new(192, 168, 1, 20));
574
575 fn host(i: usize) -> IpAddr {
577 IpAddr::V4(std::net::Ipv4Addr::new(
578 10,
579 9,
580 (i / 256) as u8,
581 (i % 256) as u8,
582 ))
583 }
584
585 fn leaked(ttl: Duration) -> &'static Pairings {
586 Box::leak(Box::new(Pairings::new(ttl)))
587 }
588
589 #[test]
590 fn codes_are_crockford_and_dashed() {
591 let p = leaked(TTL);
592 for _ in 0..50 {
593 let o = p.open("tv", None, LAN).unwrap();
594 let (a, b) = o.code.split_once('-').unwrap();
595 assert_eq!((a.len(), b.len()), (4, 4));
596 assert!(
597 format!("{a}{b}")
598 .bytes()
599 .all(|c| CODE_ALPHABET.contains(&c) && !b"ILOU".contains(&c))
600 );
601 }
602 }
603
604 #[test]
605 fn ids_are_unguessable_and_distinct() {
606 let p = leaked(TTL);
607 let opened: Vec<_> = (0..100)
608 .map(|i| p.open("tv", None, host(i)).unwrap())
609 .collect();
610 let ids: std::collections::HashSet<_> = opened.iter().map(|o| o.id.clone()).collect();
611 assert_eq!(ids.len(), 100);
612 assert!(opened.iter().all(|o| o.id.len() == 43));
613 let codes: std::collections::HashSet<_> = opened.iter().map(|o| o.code.clone()).collect();
614 assert_eq!(codes.len(), 100);
615 }
616
617 #[test]
618 fn a_code_is_found_however_it_is_typed() {
619 let p = leaked(TTL);
620 let o = p.open("Living room\u{7} TV", None, LAN).unwrap();
621 assert_eq!(
622 p.info(&o.id, ME).ok().map(|i| i.device).as_deref(),
623 Some("Living room TV")
624 );
625 let bare = o.code.replace('-', "");
626 for typed in [
627 o.code.clone(),
628 o.code.to_lowercase(),
629 bare.clone(),
630 format!(" {} {} ", &bare[..4], &bare[4..]),
631 ] {
632 assert_eq!(
633 p.info(&typed, ME).ok().map(|i| i.device).as_deref(),
634 Some("Living room TV"),
635 "{typed}"
636 );
637 }
638 assert_eq!(normalise_code("o1l1-IOAB").as_deref(), Some("011110AB"));
639 assert_eq!(normalise_code("ABCD-EFGU"), None);
640 assert_eq!(normalise_code("ABC"), None);
641 }
642
643 #[tokio::test]
644 async fn approving_delivers_the_key_to_the_waiting_device() {
645 let p = leaked(TTL);
646 let mut o = p.open("tv", None, LAN).unwrap();
647 let taken = p.take(&o.code, ME).unwrap();
648 assert_eq!(taken.device(), "tv");
649 assert!(p.take(&o.id, ME).is_err());
651 taken
652 .settle(PairMessage::Approved {
653 username: "alice".into(),
654 api_key: "key".into(),
655 })
656 .unwrap();
657 assert_eq!(
658 (&mut o.outcome).await.unwrap().message,
659 PairMessage::Approved {
660 username: "alice".into(),
661 api_key: "key".into(),
662 }
663 );
664 }
665
666 fn database() -> (tempfile::TempDir, koan_core::db::connection::Database, i64) {
667 let dir = tempfile::tempdir().unwrap();
668 let db = koan_core::db::connection::Database::open(&dir.path().join("koan.db")).unwrap();
669 let id = koan_core::db::queries::auth::create_user(
670 &db.conn,
671 "alice",
672 "hunter2",
673 koan_core::auth::Role::Readonly,
674 )
675 .unwrap();
676 (dir, db, id)
677 }
678
679 #[test]
680 fn approving_mints_a_key_for_the_approver() {
681 let (_dir, db, alice) = database();
682 let p = leaked(TTL);
683 let o = p.open("Living room TV", Some("0192-tv"), LAN).unwrap();
684 let (id, code) = (o.id.clone(), o.code.clone());
685 let device = o.device(true);
686 let settled = p.settle(&db.conn, &code, alice, "alice", false).unwrap();
687 assert_eq!(
688 settled,
689 PairInfo {
690 device: "Living room TV".into(),
691 device_id: Some("0192-tv".into()),
692 from: LAN,
693 }
694 );
695 let PairMessage::Approved { username, api_key } = device.join().unwrap() else {
696 panic!("not approved");
697 };
698 assert_eq!(username, "alice");
699 let user = koan_core::db::queries::api_keys::authenticate_api_key(&db.conn, &api_key)
700 .unwrap()
701 .unwrap();
702 assert_eq!(user.id, alice);
703 let keys = koan_core::db::queries::api_keys::list_api_keys(&db.conn, Some(alice)).unwrap();
704 assert_eq!(keys[0].name, "Living room TV");
705 assert_eq!(
707 p.settle(&db.conn, &id, alice, "alice", false),
708 Err(PairError::NotFound)
709 );
710 }
711
712 #[test]
713 fn a_key_the_socket_would_not_take_is_revoked() {
714 let (_dir, db, alice) = database();
715 let p = leaked(TTL);
716 let o = p.open("tv", None, LAN).unwrap();
717 let id = o.id.clone();
718 let device = o.device(false);
719 assert_eq!(
721 p.settle(&db.conn, &id, alice, "alice", false),
722 Err(PairError::NotFound)
723 );
724 let PairMessage::Approved { api_key, .. } = device.join().unwrap() else {
725 panic!("not approved");
726 };
727 assert!(
728 koan_core::db::queries::api_keys::authenticate_api_key(&db.conn, &api_key)
729 .unwrap()
730 .is_none()
731 );
732 let keys = koan_core::db::queries::api_keys::list_api_keys(&db.conn, Some(alice)).unwrap();
733 assert!(keys.is_empty());
734 }
735
736 #[tokio::test]
737 async fn declining_mints_nothing() {
738 let (_dir, db, alice) = database();
739 let p = leaked(TTL);
740 let mut o = p.open("tv", None, LAN).unwrap();
741 p.settle(&db.conn, &o.id, alice, "alice", true).unwrap();
742 assert_eq!(
743 (&mut o.outcome).await.unwrap().message,
744 PairMessage::Declined
745 );
746 let keys = koan_core::db::queries::api_keys::list_api_keys(&db.conn, Some(alice)).unwrap();
747 assert!(keys.is_empty());
748 }
749
750 #[test]
751 fn a_key_for_a_gone_device_is_revoked() {
752 let (_dir, db, alice) = database();
753 let p = leaked(TTL);
754 let o = p.open("tv", None, LAN).unwrap();
755 let id = o.id.clone();
756 let taken = p.take(&id, ME).unwrap();
758 drop(o);
759 p.entries.lock().insert(taken.id, taken.entry);
760 assert_eq!(
761 p.settle(&db.conn, &id, alice, "alice", false),
762 Err(PairError::NotFound)
763 );
764 let keys = koan_core::db::queries::api_keys::list_api_keys(&db.conn, Some(alice)).unwrap();
765 assert!(keys.is_empty());
766 }
767
768 #[test]
769 fn approving_an_unknown_pairing_is_not_found() {
770 let (_dir, db, alice) = database();
771 let p = leaked(TTL);
772 assert_eq!(
773 p.settle(&db.conn, "ABCD-EFGH", alice, "alice", false),
774 Err(PairError::NotFound)
775 );
776 }
777
778 #[test]
779 fn a_gone_device_gives_the_outcome_back() {
780 let p = leaked(TTL);
781 let o = p.open("tv", None, LAN).unwrap();
782 let taken = p.take(&o.id, ME).unwrap();
783 drop(o);
784 assert!(taken.settle(PairMessage::Declined).is_err());
785 }
786
787 #[tokio::test]
788 async fn a_lapsed_pairing_is_expired_and_unknown() {
789 let p = leaked(Duration::ZERO);
790 let mut o = p.open("tv", None, LAN).unwrap();
791 assert!(p.info(&o.id, ME).is_err());
792 assert!(p.take(&o.code, ME).is_err());
793 assert_eq!(
794 (&mut o.outcome).await.unwrap().message,
795 PairMessage::Expired
796 );
797 }
798
799 #[test]
800 fn the_requesting_address_is_kept() {
801 let p = leaked(TTL);
802 let mapped: IpAddr = "::ffff:203.0.113.9".parse().unwrap();
803 let o = p.open("tv", None, mapped).unwrap();
804 let info = p.info(&o.code, ME).unwrap();
805 assert_eq!(info.from.to_string(), "203.0.113.9");
806 assert!(!info.local());
807 assert_eq!(p.take(&o.id, ME).unwrap().info(), info);
808 }
809
810 #[test]
811 fn private_addresses_are_local() {
812 for local in [
813 "10.1.2.3",
814 "172.16.0.1",
815 "192.168.1.20",
816 "169.254.10.1",
817 "127.0.0.1",
818 "100.64.0.1",
819 "100.101.102.103",
820 "100.127.255.254",
821 "::1",
822 "fd12:3456::1",
823 "fe80::1",
824 "::ffff:192.168.0.5",
825 ] {
826 assert!(is_local(local.parse().unwrap()), "{local}");
827 }
828 for public in [
829 "203.0.113.9",
830 "8.8.8.8",
831 "172.32.0.1",
832 "100.63.255.255",
833 "100.128.0.1",
834 "2001:db8::1",
835 "2a00:1450::1",
836 "::ffff:8.8.8.8",
837 ] {
838 assert!(!is_local(public.parse().unwrap()), "{public}");
839 }
840 }
841
842 #[test]
843 fn unknown_pairings_are_not_found() {
844 let p = leaked(TTL);
845 let o = p.open("tv", None, LAN).unwrap();
846 let other = if o.code == "0000-0000" {
847 "1111-1111"
848 } else {
849 "0000-0000"
850 };
851 assert!(p.info(other, ME).is_err());
852 assert!(p.take("not-a-pairing", ME).is_err());
853 assert!(p.info("", ME).is_err());
854 }
855
856 #[test]
857 fn misses_are_limited_per_account() {
858 let p = leaked(TTL);
859 let o = p.open("tv", None, LAN).unwrap();
860 let other = if o.code == "0000-0000" {
861 "1111-1111"
862 } else {
863 "0000-0000"
864 };
865 for _ in 0..MISSES_PER_MINUTE {
866 assert_eq!(p.info(other, ME).err(), Some(PairError::NotFound));
867 }
868 assert_eq!(p.info(&o.code, ME).err(), Some(PairError::Limited));
870 assert_eq!(p.info(other, ME).err(), Some(PairError::Limited));
871 assert!(p.info(&o.id, ME).is_ok());
873 for _ in 0..=MISSES_PER_MINUTE {
875 assert!(p.info(&o.code, ME + 1).is_ok());
876 }
877 }
878
879 async fn serve(p: &'static Pairings) -> std::net::SocketAddr {
881 let app = axum::Router::new().route(
882 "/",
883 axum::routing::get(move |ws: WebSocketUpgrade| async move {
884 let opened = p.open("tv", None, LAN).unwrap();
885 ws.on_upgrade(move |socket| session(socket, opened, Default::default()))
886 }),
887 );
888 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
889 let addr = listener.local_addr().unwrap();
890 tokio::spawn(async move { axum::serve(listener, app).await });
891 addr
892 }
893
894 fn quick(keepalive: Duration) -> &'static Pairings {
895 let mut p = Pairings::new(TTL);
896 p.keepalive = keepalive;
897 Box::leak(Box::new(p))
898 }
899
900 fn pending_id(
901 socket: &mut tungstenite::WebSocket<impl std::io::Read + std::io::Write>,
902 ) -> String {
903 let first = socket.read().unwrap();
904 match serde_json::from_str(first.to_text().unwrap()).unwrap() {
905 PairMessage::Pending { id, .. } => id,
906 other => panic!("{other:?}"),
907 }
908 }
909
910 #[tokio::test(flavor = "multi_thread")]
911 async fn a_device_that_stops_answering_is_dropped() {
912 let p = quick(Duration::from_secs(1));
915 let addr = serve(p).await;
916 let (socket, id) = tokio::task::spawn_blocking(move || {
918 let (mut socket, _) = tungstenite::connect(format!("ws://{addr}/")).unwrap();
919 let id = pending_id(&mut socket);
920 (socket, id)
921 })
922 .await
923 .unwrap();
924 assert!(p.info(&id, ME).is_ok());
925 tokio::time::sleep(p.silence() * 2).await;
926 assert_eq!(p.info(&id, ME).err(), Some(PairError::NotFound));
927 drop(socket);
928 }
929
930 #[tokio::test(flavor = "multi_thread")]
931 async fn a_device_that_answers_waits_and_gets_its_key() {
932 let (_dir, db, alice) = database();
933 let p = quick(Duration::from_secs(1));
934 let addr = serve(p).await;
935 let (id_tx, id_rx) = oneshot::channel();
936 let device = tokio::task::spawn_blocking(move || {
937 let (mut socket, _) = tungstenite::connect(format!("ws://{addr}/")).unwrap();
938 id_tx.send(pending_id(&mut socket)).unwrap();
939 loop {
941 if let tungstenite::Message::Text(text) = socket.read().unwrap() {
942 return serde_json::from_str::<PairMessage>(&text).unwrap();
943 }
944 }
945 });
946 let id = id_rx.await.unwrap();
947 tokio::time::sleep(p.silence() * 2).await;
948 let db = tokio::task::spawn_blocking(move || {
950 p.settle(&db.conn, &id, alice, "alice", false).unwrap();
951 db
952 })
953 .await
954 .unwrap();
955 let PairMessage::Approved { api_key, .. } = device.await.unwrap() else {
956 panic!("not approved");
957 };
958 assert!(
959 koan_core::db::queries::api_keys::authenticate_api_key(&db.conn, &api_key)
960 .unwrap()
961 .is_some()
962 );
963 }
964
965 #[test]
966 fn a_closed_socket_removes_its_pairing() {
967 let p = leaked(TTL);
968 let o = p.open("tv", None, LAN).unwrap();
969 let id = o.id.clone();
970 drop(o);
971 assert!(p.info(&id, ME).is_err());
972 }
973
974 #[test]
975 fn pairings_are_capped() {
976 let p = leaked(TTL);
977 let held: Vec<_> = (0..MAX_PENDING)
978 .map(|i| p.open("tv", None, host(i)).unwrap())
979 .collect();
980 assert_eq!(p.open("tv", None, LAN).err(), Some(OpenError::Full));
981 drop(held);
982 assert!(p.open("tv", None, LAN).is_ok());
983 }
984
985 #[test]
986 fn one_address_holds_three_at_most() {
987 let p = leaked(TTL);
988 let held: Vec<_> = (0..3).map(|_| p.open("tv", None, LAN).unwrap()).collect();
989 assert_eq!(p.open("tv", None, LAN).err(), Some(OpenError::Busy));
990 let mapped: IpAddr = "::ffff:192.168.1.20".parse().unwrap();
992 assert_eq!(p.open("tv", None, mapped).err(), Some(OpenError::Busy));
993 assert!(p.open("tv", None, host(1)).is_ok());
994
995 let v6: Vec<_> = ["2001:db8:1:2::1", "2001:db8:1:2::2", "2001:db8:1:2:ffff::9"]
997 .into_iter()
998 .map(|a| p.open("tv", None, a.parse().unwrap()).unwrap())
999 .collect();
1000 assert_eq!(
1001 p.open("tv", None, "2001:db8:1:2:abcd::1".parse().unwrap())
1002 .err(),
1003 Some(OpenError::Busy)
1004 );
1005 assert!(
1006 p.open("tv", None, "2001:db8:1:3::1".parse().unwrap())
1007 .is_ok()
1008 );
1009
1010 drop(held);
1011 assert!(p.open("tv", None, LAN).is_ok());
1012 drop(v6);
1013 }
1014}