#![deny(missing_docs)]
use async_trait::async_trait;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine;
use churust_core::{Error, SessionStore, SESSION_ID_KEY};
use std::collections::BTreeMap;
use std::sync::Arc;
const ID_BYTES: usize = 32;
const ID_CHARS: usize = 43;
const DEFAULT_TTL: u64 = 24 * 60 * 60;
const DEFAULT_PREFIX: &str = "churust:session:";
#[async_trait]
trait Backend: Send + Sync + 'static {
async fn get(&self, key: &str) -> Option<String>;
async fn set(&self, key: &str, value: &str, ttl: u64);
async fn touch(&self, key: &str, ttl: u64);
async fn del(&self, key: &str) -> bool;
}
#[derive(Clone)]
pub struct RedisStore {
backend: Arc<dyn Backend>,
prefix: String,
ttl: u64,
sliding: bool,
}
impl std::fmt::Debug for RedisStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RedisStore")
.field("prefix", &self.prefix)
.field("ttl", &self.ttl)
.field("sliding", &self.sliding)
.finish_non_exhaustive()
}
}
impl RedisStore {
pub async fn connect(url: &str) -> Result<Self, redis::RedisError> {
let client = redis::Client::open(url)?;
let connection = client.get_multiplexed_async_connection().await?;
Ok(Self::from_backend(RedisBackend {
client,
connection: tokio::sync::Mutex::new(Some(connection)),
}))
}
pub fn from_client(client: redis::Client) -> Self {
Self::from_backend(RedisBackend {
client,
connection: tokio::sync::Mutex::new(None),
})
}
fn from_backend(backend: impl Backend) -> Self {
Self {
backend: Arc::new(backend),
prefix: DEFAULT_PREFIX.to_string(),
ttl: DEFAULT_TTL,
sliding: true,
}
}
pub fn ttl(mut self, secs: u64) -> Self {
assert!(secs > 0, "session ttl must be at least one second");
self.ttl = secs;
self
}
pub fn prefix(mut self, prefix: impl Into<String>) -> Self {
self.prefix = prefix.into();
self
}
pub fn sliding(mut self, yes: bool) -> Self {
self.sliding = yes;
self
}
fn key(&self, id: &str) -> String {
format!("{}{}", self.prefix, id)
}
}
fn new_id() -> String {
let mut bytes = [0u8; ID_BYTES];
getrandom::fill(&mut bytes).expect("the OS must be able to supply randomness for a session id");
URL_SAFE_NO_PAD.encode(bytes)
}
fn is_well_formed(raw: &str) -> bool {
raw.len() == ID_CHARS
&& raw
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
}
#[async_trait]
impl SessionStore for RedisStore {
async fn load(&self, raw: &str) -> Option<BTreeMap<String, String>> {
if !is_well_formed(raw) {
return None;
}
let key = self.key(raw);
let stored = self.backend.get(&key).await?;
let mut data: BTreeMap<String, String> = serde_json::from_str(&stored).ok()?;
if self.sliding {
self.backend.touch(&key, self.ttl).await;
}
data.insert(SESSION_ID_KEY.to_string(), raw.to_string());
Some(data)
}
async fn store(
&self,
data: &BTreeMap<String, String>,
previous: Option<&str>,
) -> Result<Option<String>, Error> {
if data.is_empty() {
if let Some(old) = previous.filter(|raw| is_well_formed(raw)) {
if !self.backend.del(&self.key(old)).await {
return Err(revocation_failed());
}
}
return Ok(None);
}
let carried = data.get(SESSION_ID_KEY).filter(|id| is_well_formed(id));
let id = match carried {
Some(id) => id.clone(),
None => new_id(),
};
if let Some(old) = previous.filter(|raw| is_well_formed(raw) && *raw != id) {
if !self.backend.del(&self.key(old)).await {
return Err(revocation_failed());
}
}
let mut payload = data.clone();
payload.remove(SESSION_ID_KEY);
let Ok(encoded) = serde_json::to_string(&payload) else {
return Ok(None);
};
self.backend.set(&self.key(&id), &encoded, self.ttl).await;
Ok(Some(id))
}
}
fn revocation_failed() -> Error {
Error::internal("the session could not be ended; please try again")
}
struct RedisBackend {
client: redis::Client,
connection: tokio::sync::Mutex<Option<redis::aio::MultiplexedConnection>>,
}
impl RedisBackend {
async fn run<T: redis::FromRedisValue>(&self, cmd: &redis::Cmd) -> Option<T> {
for attempt in 0..2 {
let mut guard = self.connection.lock().await;
if guard.is_none() {
match self.client.get_multiplexed_async_connection().await {
Ok(fresh) => *guard = Some(fresh),
Err(_) => return None,
}
}
let mut conn = guard.as_ref()?.clone();
drop(guard);
match cmd.query_async::<T>(&mut conn).await {
Ok(value) => return Some(value),
Err(_) if attempt == 0 => {
*self.connection.lock().await = None;
}
Err(_) => return None,
}
}
None
}
}
#[async_trait]
impl Backend for RedisBackend {
async fn get(&self, key: &str) -> Option<String> {
self.run::<Option<String>>(redis::cmd("GET").arg(key))
.await
.flatten()
}
async fn set(&self, key: &str, value: &str, ttl: u64) {
let _ = self
.run::<()>(redis::cmd("SET").arg(key).arg(value).arg("EX").arg(ttl))
.await;
}
async fn touch(&self, key: &str, ttl: u64) {
let _ = self.run::<()>(redis::cmd("EXPIRE").arg(key).arg(ttl)).await;
}
async fn del(&self, key: &str) -> bool {
self.run::<()>(redis::cmd("DEL").arg(key)).await.is_some()
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
use std::time::{Duration, Instant};
#[derive(Default)]
struct MemoryBackend {
entries: Mutex<BTreeMap<String, (String, Instant)>>,
}
impl MemoryBackend {
fn live(&self, key: &str) -> Option<String> {
let entries = self.entries.lock().unwrap();
let (value, expires) = entries.get(key)?;
(*expires > Instant::now()).then(|| value.clone())
}
fn len(&self) -> usize {
let now = Instant::now();
self.entries
.lock()
.unwrap()
.values()
.filter(|(_, expires)| *expires > now)
.count()
}
}
#[async_trait]
impl Backend for Arc<MemoryBackend> {
async fn get(&self, key: &str) -> Option<String> {
self.live(key)
}
async fn set(&self, key: &str, value: &str, ttl: u64) {
self.entries.lock().unwrap().insert(
key.to_string(),
(value.to_string(), Instant::now() + Duration::from_secs(ttl)),
);
}
async fn touch(&self, key: &str, ttl: u64) {
if let Some(entry) = self.entries.lock().unwrap().get_mut(key) {
entry.1 = Instant::now() + Duration::from_secs(ttl);
}
}
async fn del(&self, key: &str) -> bool {
self.entries.lock().unwrap().remove(key);
true
}
}
fn store() -> (RedisStore, Arc<MemoryBackend>) {
let backend = Arc::new(MemoryBackend::default());
let store = RedisStore {
backend: Arc::new(backend.clone()),
prefix: DEFAULT_PREFIX.to_string(),
ttl: DEFAULT_TTL,
sliding: true,
};
(store, backend)
}
struct RefusesDeletes(Arc<MemoryBackend>);
#[async_trait]
impl Backend for RefusesDeletes {
async fn get(&self, key: &str) -> Option<String> {
self.0.get(key).await
}
async fn set(&self, key: &str, value: &str, ttl: u64) {
self.0.set(key, value, ttl).await
}
async fn touch(&self, key: &str, ttl: u64) {
self.0.touch(key, ttl).await
}
async fn del(&self, _key: &str) -> bool {
false
}
}
fn refusing_store() -> (RedisStore, Arc<MemoryBackend>) {
let backend = Arc::new(MemoryBackend::default());
let store = RedisStore {
backend: Arc::new(RefusesDeletes(backend.clone())),
prefix: DEFAULT_PREFIX.to_string(),
ttl: DEFAULT_TTL,
sliding: true,
};
(store, backend)
}
fn data(pairs: &[(&str, &str)]) -> BTreeMap<String, String> {
pairs
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect()
}
#[tokio::test]
async fn a_session_round_trips_through_the_backend() {
let (store, _) = store();
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.expect("the write must succeed")
.expect("a new session gets an id");
let back = store.load(&id).await.expect("it must load again");
assert_eq!(back.get("user").map(String::as_str), Some("ana"));
}
#[tokio::test]
async fn the_cookie_carries_only_an_identifier() {
let (store, backend) = store();
let id = store
.store(&data(&[("user", "ana"), ("secret", "hunter2")]), None)
.await
.unwrap()
.unwrap();
assert!(is_well_formed(&id));
assert!(
!id.contains("ana") && !id.contains("hunter2"),
"session contents must not travel in the cookie: {id}"
);
let raw = backend.live(&format!("{DEFAULT_PREFIX}{id}")).unwrap();
assert!(raw.contains("hunter2"), "the value lives server side");
}
#[tokio::test]
async fn identifiers_are_unpredictable_and_distinct() {
let (store, _) = store();
let mut seen = std::collections::HashSet::new();
for _ in 0..256 {
let id = store
.store(&data(&[("k", "v")]), None)
.await
.unwrap()
.unwrap();
assert_eq!(id.len(), ID_CHARS);
assert!(seen.insert(id), "a session id was reused");
}
}
#[tokio::test]
async fn logging_out_deletes_the_record() {
let (store, backend) = store();
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.unwrap()
.unwrap();
assert_eq!(backend.len(), 1);
let reissued = store.store(&BTreeMap::new(), Some(&id)).await;
assert!(
matches!(reissued, Ok(None)),
"a completed logout has no new cookie to set"
);
assert_eq!(
backend.len(),
0,
"the record must be gone, not merely stale"
);
assert!(
store.load(&id).await.is_none(),
"a cookie copied before logout must stop working"
);
}
#[tokio::test]
async fn a_logout_the_backend_refused_is_not_reported_as_a_logout() {
let (store, backend) = refusing_store();
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.unwrap()
.unwrap();
let outcome = store.store(&BTreeMap::new(), Some(&id)).await;
assert!(
store.load(&id).await.is_some(),
"the stand-in must have kept the record, or this proves nothing"
);
assert_eq!(backend.len(), 1, "the record survived the failed delete");
assert!(
outcome.is_err(),
"a session that is still valid must not be reported as withdrawn"
);
}
#[tokio::test]
async fn a_rotation_whose_withdrawal_failed_is_refused_too() {
let (store, _) = refusing_store();
let first = store
.store(&data(&[("cart", "3")]), None)
.await
.unwrap()
.unwrap();
let mut rotated = store.load(&first).await.unwrap();
rotated.remove(SESSION_ID_KEY);
rotated.insert("user".into(), "ana".into());
assert!(
store.store(&rotated, Some(&first)).await.is_err(),
"the pre-login identifier still resolves, so the rotation failed"
);
assert!(
store.load(&first).await.is_some(),
"the stand-in must have kept the record, or this proves nothing"
);
}
#[tokio::test]
async fn rotating_mints_a_new_id_and_withdraws_the_old_one() {
let (store, backend) = store();
let first = store
.store(&data(&[("cart", "3")]), None)
.await
.unwrap()
.unwrap();
let mut rotated = store.load(&first).await.unwrap();
rotated.remove(SESSION_ID_KEY);
rotated.insert("user".into(), "ana".into());
let second = store.store(&rotated, Some(&first)).await.unwrap().unwrap();
assert_ne!(first, second, "a rotated session must change identifier");
assert!(
store.load(&first).await.is_none(),
"the pre-login identifier must not still resolve"
);
let carried = store.load(&second).await.unwrap();
assert_eq!(carried.get("cart").map(String::as_str), Some("3"));
assert_eq!(carried.get("user").map(String::as_str), Some("ana"));
assert_eq!(backend.len(), 1, "the old record was not left behind");
}
#[tokio::test]
async fn an_unchanged_session_keeps_its_identifier() {
let (store, backend) = store();
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.unwrap()
.unwrap();
let mut loaded = store.load(&id).await.unwrap();
loaded.insert("theme".into(), "dark".into());
let again = store.store(&loaded, Some(&id)).await.unwrap().unwrap();
assert_eq!(id, again, "an ordinary write must not rotate the session");
assert_eq!(backend.len(), 1);
}
#[tokio::test]
async fn a_malformed_identifier_is_refused_without_a_lookup() {
let (store, _) = store();
for hostile in [
"",
"short",
"../../etc/passwd",
"churust:session:*",
"a b",
&"x".repeat(4096),
] {
assert!(!is_well_formed(hostile), "{hostile:?} should not be valid");
assert!(store.load(hostile).await.is_none());
}
}
#[tokio::test]
async fn an_unknown_identifier_loads_nothing() {
let (store, _) = store();
assert!(store.load(&new_id()).await.is_none());
}
#[tokio::test]
async fn the_stored_value_does_not_repeat_the_identifier() {
let (store, backend) = store();
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.unwrap()
.unwrap();
let raw = backend.live(&format!("{DEFAULT_PREFIX}{id}")).unwrap();
assert!(
!raw.contains(SESSION_ID_KEY),
"the key is the identifier; storing it twice invites disagreement: {raw}"
);
}
#[tokio::test]
async fn expiry_removes_a_session() {
let (mut store, _) = store();
store.ttl = 1;
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.unwrap()
.unwrap();
assert!(store.load(&id).await.is_some());
tokio::time::sleep(Duration::from_millis(1100)).await;
assert!(
store.load(&id).await.is_none(),
"a session past its ttl must not load"
);
}
#[tokio::test]
async fn sliding_expiry_extends_on_read() {
let (mut store, backend) = store();
store.ttl = 2;
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.unwrap()
.unwrap();
for _ in 0..3 {
tokio::time::sleep(Duration::from_millis(800)).await;
assert!(store.load(&id).await.is_some());
}
assert_eq!(backend.len(), 1);
}
#[tokio::test]
async fn absolute_expiry_does_not_extend_on_read() {
let (mut store, _) = store();
store.ttl = 1;
store.sliding = false;
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.unwrap()
.unwrap();
tokio::time::sleep(Duration::from_millis(600)).await;
assert!(store.load(&id).await.is_some());
tokio::time::sleep(Duration::from_millis(600)).await;
assert!(
store.load(&id).await.is_none(),
"reading must not have extended the deadline"
);
}
#[tokio::test]
async fn a_custom_prefix_is_applied() {
let (mut store, backend) = store();
store.prefix = "app:sess:".into();
let id = store
.store(&data(&[("user", "ana")]), None)
.await
.unwrap()
.unwrap();
assert!(backend.live(&format!("app:sess:{id}")).is_some());
}
#[test]
#[should_panic(expected = "at least one second")]
fn a_zero_ttl_is_refused() {
let (store, _) = store();
let _ = store.ttl(0);
}
}