1#![deny(missing_docs)]
48
49use async_trait::async_trait;
50use base64::engine::general_purpose::URL_SAFE_NO_PAD;
51use base64::Engine;
52use churust_core::{Error, SessionStore, SESSION_ID_KEY};
53use std::collections::BTreeMap;
54use std::sync::Arc;
55
56const ID_BYTES: usize = 32;
58const ID_CHARS: usize = 43;
60const DEFAULT_TTL: u64 = 24 * 60 * 60;
62const DEFAULT_PREFIX: &str = "churust:session:";
64
65#[async_trait]
71trait Backend: Send + Sync + 'static {
72 async fn get(&self, key: &str) -> Option<String>;
73 async fn set(&self, key: &str, value: &str, ttl: u64);
74 async fn touch(&self, key: &str, ttl: u64);
75 async fn del(&self, key: &str) -> bool;
78}
79
80#[derive(Clone)]
82pub struct RedisStore {
83 backend: Arc<dyn Backend>,
84 prefix: String,
85 ttl: u64,
86 sliding: bool,
87}
88
89impl std::fmt::Debug for RedisStore {
90 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91 f.debug_struct("RedisStore")
92 .field("prefix", &self.prefix)
93 .field("ttl", &self.ttl)
94 .field("sliding", &self.sliding)
95 .finish_non_exhaustive()
96 }
97}
98
99impl RedisStore {
100 pub async fn connect(url: &str) -> Result<Self, redis::RedisError> {
113 let client = redis::Client::open(url)?;
114 let connection = client.get_multiplexed_async_connection().await?;
117 Ok(Self::from_backend(RedisBackend {
118 client,
119 connection: tokio::sync::Mutex::new(Some(connection)),
120 }))
121 }
122
123 pub fn from_client(client: redis::Client) -> Self {
128 Self::from_backend(RedisBackend {
129 client,
130 connection: tokio::sync::Mutex::new(None),
131 })
132 }
133
134 fn from_backend(backend: impl Backend) -> Self {
135 Self {
136 backend: Arc::new(backend),
137 prefix: DEFAULT_PREFIX.to_string(),
138 ttl: DEFAULT_TTL,
139 sliding: true,
140 }
141 }
142
143 pub fn ttl(mut self, secs: u64) -> Self {
149 assert!(secs > 0, "session ttl must be at least one second");
150 self.ttl = secs;
151 self
152 }
153
154 pub fn prefix(mut self, prefix: impl Into<String>) -> Self {
156 self.prefix = prefix.into();
157 self
158 }
159
160 pub fn sliding(mut self, yes: bool) -> Self {
167 self.sliding = yes;
168 self
169 }
170
171 fn key(&self, id: &str) -> String {
172 format!("{}{}", self.prefix, id)
173 }
174}
175
176fn new_id() -> String {
184 let mut bytes = [0u8; ID_BYTES];
185 getrandom::fill(&mut bytes).expect("the OS must be able to supply randomness for a session id");
186 URL_SAFE_NO_PAD.encode(bytes)
187}
188
189fn is_well_formed(raw: &str) -> bool {
194 raw.len() == ID_CHARS
195 && raw
196 .bytes()
197 .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
198}
199
200#[async_trait]
201impl SessionStore for RedisStore {
202 async fn load(&self, raw: &str) -> Option<BTreeMap<String, String>> {
203 if !is_well_formed(raw) {
204 return None;
205 }
206 let key = self.key(raw);
207 let stored = self.backend.get(&key).await?;
208 let mut data: BTreeMap<String, String> = serde_json::from_str(&stored).ok()?;
209
210 if self.sliding {
211 self.backend.touch(&key, self.ttl).await;
212 }
213
214 data.insert(SESSION_ID_KEY.to_string(), raw.to_string());
217 Some(data)
218 }
219
220 async fn store(
221 &self,
222 data: &BTreeMap<String, String>,
223 previous: Option<&str>,
224 ) -> Result<Option<String>, Error> {
225 if data.is_empty() {
229 if let Some(old) = previous.filter(|raw| is_well_formed(raw)) {
230 if !self.backend.del(&self.key(old)).await {
231 return Err(revocation_failed());
232 }
233 }
234 return Ok(None);
235 }
236
237 let carried = data.get(SESSION_ID_KEY).filter(|id| is_well_formed(id));
238 let id = match carried {
239 Some(id) => id.clone(),
240 None => new_id(),
241 };
242
243 if let Some(old) = previous.filter(|raw| is_well_formed(raw) && *raw != id) {
248 if !self.backend.del(&self.key(old)).await {
254 return Err(revocation_failed());
255 }
256 }
257
258 let mut payload = data.clone();
261 payload.remove(SESSION_ID_KEY);
262 let Ok(encoded) = serde_json::to_string(&payload) else {
263 return Ok(None);
264 };
265
266 self.backend.set(&self.key(&id), &encoded, self.ttl).await;
271 Ok(Some(id))
272 }
273}
274
275fn revocation_failed() -> Error {
282 Error::internal("the session could not be ended; please try again")
283}
284
285struct RedisBackend {
287 client: redis::Client,
288 connection: tokio::sync::Mutex<Option<redis::aio::MultiplexedConnection>>,
291}
292
293impl RedisBackend {
294 async fn run<T: redis::FromRedisValue>(&self, cmd: &redis::Cmd) -> Option<T> {
307 for attempt in 0..2 {
308 let mut conn = self.connection().await?;
309
310 match cmd.query_async::<T>(&mut conn).await {
311 Ok(value) => return Some(value),
312 Err(_) if attempt == 0 => {
313 *self.connection.lock().await = None;
315 }
316 Err(_) => return None,
317 }
318 }
319 None
320 }
321
322 async fn connection(&self) -> Option<redis::aio::MultiplexedConnection> {
339 if let Some(conn) = self.connection.lock().await.as_ref().cloned() {
342 return Some(conn);
343 }
344
345 let fresh = self.client.get_multiplexed_async_connection().await.ok()?;
346
347 let mut guard = self.connection.lock().await;
348 match guard.as_ref() {
349 Some(existing) => Some(existing.clone()),
352 None => {
353 *guard = Some(fresh.clone());
354 Some(fresh)
355 }
356 }
357 }
358}
359
360#[async_trait]
361impl Backend for RedisBackend {
362 async fn get(&self, key: &str) -> Option<String> {
363 self.run::<Option<String>>(redis::cmd("GET").arg(key))
364 .await
365 .flatten()
366 }
367
368 async fn set(&self, key: &str, value: &str, ttl: u64) {
369 let _ = self
372 .run::<()>(redis::cmd("SET").arg(key).arg(value).arg("EX").arg(ttl))
373 .await;
374 }
375
376 async fn touch(&self, key: &str, ttl: u64) {
377 let _ = self.run::<()>(redis::cmd("EXPIRE").arg(key).arg(ttl)).await;
378 }
379
380 async fn del(&self, key: &str) -> bool {
381 self.run::<()>(redis::cmd("DEL").arg(key)).await.is_some()
387 }
388}
389
390#[cfg(test)]
391mod tests {
392 use super::*;
393 use std::sync::Mutex;
394 use std::time::{Duration, Instant};
395
396 #[derive(Default)]
399 struct MemoryBackend {
400 entries: Mutex<BTreeMap<String, (String, Instant)>>,
401 }
402
403 impl MemoryBackend {
404 fn live(&self, key: &str) -> Option<String> {
405 let entries = self.entries.lock().unwrap();
406 let (value, expires) = entries.get(key)?;
407 (*expires > Instant::now()).then(|| value.clone())
408 }
409
410 fn len(&self) -> usize {
411 let now = Instant::now();
412 self.entries
413 .lock()
414 .unwrap()
415 .values()
416 .filter(|(_, expires)| *expires > now)
417 .count()
418 }
419 }
420
421 #[async_trait]
422 impl Backend for Arc<MemoryBackend> {
423 async fn get(&self, key: &str) -> Option<String> {
424 self.live(key)
425 }
426
427 async fn set(&self, key: &str, value: &str, ttl: u64) {
428 self.entries.lock().unwrap().insert(
429 key.to_string(),
430 (value.to_string(), Instant::now() + Duration::from_secs(ttl)),
431 );
432 }
433
434 async fn touch(&self, key: &str, ttl: u64) {
435 if let Some(entry) = self.entries.lock().unwrap().get_mut(key) {
436 entry.1 = Instant::now() + Duration::from_secs(ttl);
437 }
438 }
439
440 async fn del(&self, key: &str) -> bool {
441 self.entries.lock().unwrap().remove(key);
442 true
443 }
444 }
445
446 fn store() -> (RedisStore, Arc<MemoryBackend>) {
447 let backend = Arc::new(MemoryBackend::default());
448 let store = RedisStore {
449 backend: Arc::new(backend.clone()),
450 prefix: DEFAULT_PREFIX.to_string(),
451 ttl: DEFAULT_TTL,
452 sliding: true,
453 };
454 (store, backend)
455 }
456
457 struct RefusesDeletes(Arc<MemoryBackend>);
464
465 #[async_trait]
466 impl Backend for RefusesDeletes {
467 async fn get(&self, key: &str) -> Option<String> {
468 self.0.get(key).await
469 }
470
471 async fn set(&self, key: &str, value: &str, ttl: u64) {
472 self.0.set(key, value, ttl).await
473 }
474
475 async fn touch(&self, key: &str, ttl: u64) {
476 self.0.touch(key, ttl).await
477 }
478
479 async fn del(&self, _key: &str) -> bool {
480 false
481 }
482 }
483
484 fn refusing_store() -> (RedisStore, Arc<MemoryBackend>) {
485 let backend = Arc::new(MemoryBackend::default());
486 let store = RedisStore {
487 backend: Arc::new(RefusesDeletes(backend.clone())),
488 prefix: DEFAULT_PREFIX.to_string(),
489 ttl: DEFAULT_TTL,
490 sliding: true,
491 };
492 (store, backend)
493 }
494
495 fn data(pairs: &[(&str, &str)]) -> BTreeMap<String, String> {
496 pairs
497 .iter()
498 .map(|(k, v)| (k.to_string(), v.to_string()))
499 .collect()
500 }
501
502 #[tokio::test]
503 async fn a_session_round_trips_through_the_backend() {
504 let (store, _) = store();
505 let id = store
506 .store(&data(&[("user", "ana")]), None)
507 .await
508 .expect("the write must succeed")
509 .expect("a new session gets an id");
510
511 let back = store.load(&id).await.expect("it must load again");
512 assert_eq!(back.get("user").map(String::as_str), Some("ana"));
513 }
514
515 #[tokio::test]
516 async fn the_cookie_carries_only_an_identifier() {
517 let (store, backend) = store();
518 let id = store
519 .store(&data(&[("user", "ana"), ("secret", "hunter2")]), None)
520 .await
521 .unwrap()
522 .unwrap();
523
524 assert!(is_well_formed(&id));
525 assert!(
526 !id.contains("ana") && !id.contains("hunter2"),
527 "session contents must not travel in the cookie: {id}"
528 );
529 let raw = backend.live(&format!("{DEFAULT_PREFIX}{id}")).unwrap();
530 assert!(raw.contains("hunter2"), "the value lives server side");
531 }
532
533 #[tokio::test]
534 async fn identifiers_are_unpredictable_and_distinct() {
535 let (store, _) = store();
536 let mut seen = std::collections::HashSet::new();
537 for _ in 0..256 {
538 let id = store
539 .store(&data(&[("k", "v")]), None)
540 .await
541 .unwrap()
542 .unwrap();
543 assert_eq!(id.len(), ID_CHARS);
544 assert!(seen.insert(id), "a session id was reused");
545 }
546 }
547
548 #[tokio::test]
549 async fn logging_out_deletes_the_record() {
550 let (store, backend) = store();
551 let id = store
552 .store(&data(&[("user", "ana")]), None)
553 .await
554 .unwrap()
555 .unwrap();
556 assert_eq!(backend.len(), 1);
557
558 let reissued = store.store(&BTreeMap::new(), Some(&id)).await;
560 assert!(
561 matches!(reissued, Ok(None)),
562 "a completed logout has no new cookie to set"
563 );
564 assert_eq!(
565 backend.len(),
566 0,
567 "the record must be gone, not merely stale"
568 );
569 assert!(
570 store.load(&id).await.is_none(),
571 "a cookie copied before logout must stop working"
572 );
573 }
574
575 #[tokio::test]
576 async fn a_logout_the_backend_refused_is_not_reported_as_a_logout() {
577 let (store, backend) = refusing_store();
578 let id = store
579 .store(&data(&[("user", "ana")]), None)
580 .await
581 .unwrap()
582 .unwrap();
583
584 let outcome = store.store(&BTreeMap::new(), Some(&id)).await;
585
586 assert!(
587 store.load(&id).await.is_some(),
588 "the stand-in must have kept the record, or this proves nothing"
589 );
590 assert_eq!(backend.len(), 1, "the record survived the failed delete");
591 assert!(
592 outcome.is_err(),
593 "a session that is still valid must not be reported as withdrawn"
594 );
595 }
596
597 #[tokio::test]
598 async fn a_rotation_whose_withdrawal_failed_is_refused_too() {
599 let (store, _) = refusing_store();
604 let first = store
605 .store(&data(&[("cart", "3")]), None)
606 .await
607 .unwrap()
608 .unwrap();
609
610 let mut rotated = store.load(&first).await.unwrap();
611 rotated.remove(SESSION_ID_KEY);
612 rotated.insert("user".into(), "ana".into());
613
614 assert!(
615 store.store(&rotated, Some(&first)).await.is_err(),
616 "the pre-login identifier still resolves, so the rotation failed"
617 );
618 assert!(
619 store.load(&first).await.is_some(),
620 "the stand-in must have kept the record, or this proves nothing"
621 );
622 }
623
624 #[tokio::test]
625 async fn rotating_mints_a_new_id_and_withdraws_the_old_one() {
626 let (store, backend) = store();
627 let first = store
628 .store(&data(&[("cart", "3")]), None)
629 .await
630 .unwrap()
631 .unwrap();
632
633 let mut rotated = store.load(&first).await.unwrap();
635 rotated.remove(SESSION_ID_KEY);
636 rotated.insert("user".into(), "ana".into());
637
638 let second = store.store(&rotated, Some(&first)).await.unwrap().unwrap();
639 assert_ne!(first, second, "a rotated session must change identifier");
640 assert!(
641 store.load(&first).await.is_none(),
642 "the pre-login identifier must not still resolve"
643 );
644 let carried = store.load(&second).await.unwrap();
645 assert_eq!(carried.get("cart").map(String::as_str), Some("3"));
646 assert_eq!(carried.get("user").map(String::as_str), Some("ana"));
647 assert_eq!(backend.len(), 1, "the old record was not left behind");
648 }
649
650 #[tokio::test]
651 async fn an_unchanged_session_keeps_its_identifier() {
652 let (store, backend) = store();
653 let id = store
654 .store(&data(&[("user", "ana")]), None)
655 .await
656 .unwrap()
657 .unwrap();
658
659 let mut loaded = store.load(&id).await.unwrap();
660 loaded.insert("theme".into(), "dark".into());
661 let again = store.store(&loaded, Some(&id)).await.unwrap().unwrap();
662
663 assert_eq!(id, again, "an ordinary write must not rotate the session");
664 assert_eq!(backend.len(), 1);
665 }
666
667 #[tokio::test]
668 async fn a_malformed_identifier_is_refused_without_a_lookup() {
669 let (store, _) = store();
670 for hostile in [
671 "",
672 "short",
673 "../../etc/passwd",
674 "churust:session:*",
675 "a b",
676 &"x".repeat(4096),
677 ] {
678 assert!(!is_well_formed(hostile), "{hostile:?} should not be valid");
679 assert!(store.load(hostile).await.is_none());
680 }
681 }
682
683 #[tokio::test]
684 async fn an_unknown_identifier_loads_nothing() {
685 let (store, _) = store();
686 assert!(store.load(&new_id()).await.is_none());
687 }
688
689 #[tokio::test]
690 async fn the_stored_value_does_not_repeat_the_identifier() {
691 let (store, backend) = store();
692 let id = store
693 .store(&data(&[("user", "ana")]), None)
694 .await
695 .unwrap()
696 .unwrap();
697 let raw = backend.live(&format!("{DEFAULT_PREFIX}{id}")).unwrap();
698 assert!(
699 !raw.contains(SESSION_ID_KEY),
700 "the key is the identifier; storing it twice invites disagreement: {raw}"
701 );
702 }
703
704 #[tokio::test]
705 async fn expiry_removes_a_session() {
706 let (mut store, _) = store();
707 store.ttl = 1;
708 let id = store
709 .store(&data(&[("user", "ana")]), None)
710 .await
711 .unwrap()
712 .unwrap();
713 assert!(store.load(&id).await.is_some());
714
715 tokio::time::sleep(Duration::from_millis(1100)).await;
716 assert!(
717 store.load(&id).await.is_none(),
718 "a session past its ttl must not load"
719 );
720 }
721
722 #[tokio::test]
723 async fn sliding_expiry_extends_on_read() {
724 let (mut store, backend) = store();
725 store.ttl = 2;
726 let id = store
727 .store(&data(&[("user", "ana")]), None)
728 .await
729 .unwrap()
730 .unwrap();
731
732 for _ in 0..3 {
734 tokio::time::sleep(Duration::from_millis(800)).await;
735 assert!(store.load(&id).await.is_some());
736 }
737 assert_eq!(backend.len(), 1);
738 }
739
740 #[tokio::test]
741 async fn absolute_expiry_does_not_extend_on_read() {
742 let (mut store, _) = store();
743 store.ttl = 1;
744 store.sliding = false;
745 let id = store
746 .store(&data(&[("user", "ana")]), None)
747 .await
748 .unwrap()
749 .unwrap();
750
751 tokio::time::sleep(Duration::from_millis(600)).await;
752 assert!(store.load(&id).await.is_some());
753 tokio::time::sleep(Duration::from_millis(600)).await;
754 assert!(
755 store.load(&id).await.is_none(),
756 "reading must not have extended the deadline"
757 );
758 }
759
760 #[tokio::test]
761 async fn a_custom_prefix_is_applied() {
762 let (mut store, backend) = store();
763 store.prefix = "app:sess:".into();
764 let id = store
765 .store(&data(&[("user", "ana")]), None)
766 .await
767 .unwrap()
768 .unwrap();
769 assert!(backend.live(&format!("app:sess:{id}")).is_some());
770 }
771
772 #[test]
773 #[should_panic(expected = "at least one second")]
774 fn a_zero_ttl_is_refused() {
775 let (store, _) = store();
776 let _ = store.ttl(0);
777 }
778
779 #[tokio::test]
795 async fn a_cold_dial_does_not_serialise_every_other_session_operation() {
796 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
797 let addr = listener.local_addr().unwrap();
798
799 let seen = Arc::new(std::sync::atomic::AtomicUsize::new(0));
800 let counter = seen.clone();
801 tokio::spawn(async move {
802 while let Ok((sock, _)) = listener.accept().await {
804 counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
805 std::mem::forget(sock);
806 }
807 });
808
809 let client = redis::Client::open(format!("redis://{addr}")).expect("a client");
810 let backend = Arc::new(RedisBackend {
811 client,
812 connection: tokio::sync::Mutex::new(None),
813 });
814
815 const CALLERS: usize = 5;
816 let mut tasks = Vec::new();
817 for _ in 0..CALLERS {
818 let backend = backend.clone();
819 tasks.push(tokio::spawn(async move { backend.get("k").await }));
820 }
821
822 tokio::time::sleep(Duration::from_millis(600)).await;
825 let dials = seen.load(std::sync::atomic::Ordering::Relaxed);
826 for t in tasks {
827 t.abort();
828 }
829
830 assert!(
831 dials > 1,
832 "only {dials} of {CALLERS} callers reached a dial: they are queued \
833 behind one another on the connection lock"
834 );
835 }
836}