featherbit/sessions/
redis.rs

1//! Redis/Valkey-backed [`SessionStore`] (`redis-store` feature).
2//!
3//! Keys are Cluster hash-tagged on the session id (spec "Cluster
4//! readiness"): a session's payload, meta, and lock always share a slot.
5//! The subject index is a SET per sha256(subject), lazily pruned. Payloads
6//! arrive already sealed (AES-256-GCM via the plugin's `CookieSealer`) —
7//! nothing readable sits in Redis. Errors bump
8//! `gateway_session_store_errors_total{store}` and surface as
9//! [`StoreError`] (503 at the plugin).
10
11use std::sync::Arc;
12use std::time::Duration;
13
14use async_trait::async_trait;
15use redis::AsyncCommands;
16use ring::digest::{digest, SHA256};
17
18use crate::metrics::GatewayMetrics;
19use crate::stores::redis_store::RedisStoreClient;
20
21use super::{SessionFilter, SessionId, SessionMeta, SessionPage, SessionStore, StoreError};
22
23pub struct RedisSessionStore {
24    client: Arc<RedisStoreClient>,
25    metrics: Option<Arc<GatewayMetrics>>,
26}
27
28fn sha256_hex(input: &str) -> String {
29    digest(&SHA256, input.as_bytes())
30        .as_ref()
31        .iter()
32        .map(|b| format!("{b:02x}"))
33        .collect()
34}
35
36/// Key builders — pure, unit-tested. Braces are literal Redis Cluster hash
37/// tags, so `sess`/`meta`/`lock` for one id always share a slot.
38pub(crate) fn sess_key(prefix: &str, id: &str) -> String {
39    format!("{prefix}:{}:{{{id}}}", crate::stores::namespaces::SESSIONS)
40}
41fn meta_key(prefix: &str, id: &str) -> String {
42    format!(
43        "{prefix}:{}:{{{id}}}:meta",
44        crate::stores::namespaces::SESSIONS
45    )
46}
47pub(crate) fn lock_key(prefix: &str, id: &str) -> String {
48    format!(
49        "{prefix}:{}:{{{id}}}",
50        crate::stores::namespaces::SESSION_LOCKS
51    )
52}
53pub(crate) fn subj_key(prefix: &str, subject: &str) -> String {
54    format!(
55        "{prefix}:{}:{{{}}}",
56        crate::stores::namespaces::SESSION_SUBJECTS,
57        sha256_hex(subject)
58    )
59}
60/// Extracts the id from a meta key produced by [`meta_key`].
61fn id_of_meta_key(prefix: &str, key: &str) -> Option<String> {
62    key.strip_prefix(&format!(
63        "{prefix}:{}:{{",
64        crate::stores::namespaces::SESSIONS
65    ))?
66    .strip_suffix("}:meta")
67    .map(str::to_string)
68}
69
70impl RedisSessionStore {
71    pub fn new(client: Arc<RedisStoreClient>, metrics: Option<Arc<GatewayMetrics>>) -> Self {
72        Self { client, metrics }
73    }
74
75    fn err(&self, msg: String) -> StoreError {
76        if let Some(ref m) = self.metrics {
77            m.session_store_errors
78                .with_label_values(&[self.client.name()])
79                .inc();
80        }
81        tracing::warn!(store = %self.client.name(), "session store error: {}", msg);
82        StoreError(msg)
83    }
84
85    async fn conn(&self) -> Result<crate::stores::redis_store::StoreConn, StoreError> {
86        self.client.conn().await.map_err(|e| self.err(e))
87    }
88}
89
90#[async_trait]
91impl SessionStore for RedisSessionStore {
92    async fn put(
93        &self,
94        id: &SessionId,
95        sealed: &[u8],
96        ttl: Duration,
97        meta: &SessionMeta,
98    ) -> Result<(), StoreError> {
99        let p = self.client.key_prefix();
100        let ttl_secs = ttl.as_secs().max(1);
101        let meta_json =
102            serde_json::to_string(meta).map_err(|e| self.err(format!("meta serialize: {e}")))?;
103        let mut conn = self.conn().await?;
104        let mut pipe = redis::pipe();
105        pipe.set_ex(sess_key(p, id.as_str()), sealed, ttl_secs)
106            .set_ex(meta_key(p, id.as_str()), meta_json, ttl_secs);
107        if !meta.subject.is_empty() {
108            let sk = subj_key(p, &meta.subject);
109            // NX then GT: a fresh index gets the TTL; an existing one only
110            // ever grows — it must outlive its longest-lived member, and a
111            // shorter new session must not shrink it.
112            pipe.sadd(&sk, id.as_str());
113            pipe.cmd("EXPIRE").arg(&sk).arg(ttl_secs).arg("NX");
114            pipe.cmd("EXPIRE").arg(&sk).arg(ttl_secs).arg("GT");
115        }
116        pipe.query_async::<()>(&mut conn)
117            .await
118            .map_err(|e| self.err(format!("put: {e}")))
119    }
120
121    async fn get(&self, id: &SessionId) -> Result<Option<Vec<u8>>, StoreError> {
122        let p = self.client.key_prefix();
123        let mut conn = self.conn().await?;
124        conn.get::<_, Option<Vec<u8>>>(sess_key(p, id.as_str()))
125            .await
126            .map_err(|e| self.err(format!("get: {e}")))
127    }
128
129    async fn delete(&self, id: &SessionId) -> Result<(), StoreError> {
130        let p = self.client.key_prefix();
131        let mut conn = self.conn().await?;
132        // Read the meta first so the subject index entry can be pruned.
133        let meta: Option<String> = conn
134            .get(meta_key(p, id.as_str()))
135            .await
136            .map_err(|e| self.err(format!("delete meta read: {e}")))?;
137        let _: () = conn
138            .del(&[sess_key(p, id.as_str()), meta_key(p, id.as_str())])
139            .await
140            .map_err(|e| self.err(format!("delete: {e}")))?;
141        if let Some(m) = meta.and_then(|s| serde_json::from_str::<SessionMeta>(&s).ok()) {
142            if !m.subject.is_empty() {
143                let _: () = conn
144                    .srem(subj_key(p, &m.subject), id.as_str())
145                    .await
146                    .map_err(|e| self.err(format!("delete srem: {e}")))?;
147            }
148        }
149        Ok(())
150    }
151
152    async fn delete_subject(&self, subject: &str) -> Result<u64, StoreError> {
153        let p = self.client.key_prefix();
154        let mut conn = self.conn().await?;
155        let sk = subj_key(p, subject);
156        let ids: Vec<String> = conn
157            .smembers(&sk)
158            .await
159            .map_err(|e| self.err(format!("delete_subject smembers: {e}")))?;
160        let mut removed = 0u64;
161        for id in &ids {
162            // Individual DELs: cross-slot under Cluster, so no multi-key op.
163            let n: u64 = conn
164                .del(&[sess_key(p, id), meta_key(p, id)])
165                .await
166                .map_err(|e| self.err(format!("delete_subject del: {e}")))?;
167            if n > 0 {
168                removed += 1;
169            }
170        }
171        let _: () = conn
172            .del(&sk)
173            .await
174            .map_err(|e| self.err(format!("delete_subject index del: {e}")))?;
175        Ok(removed)
176    }
177
178    async fn list(&self, filter: &SessionFilter) -> Result<SessionPage, StoreError> {
179        let p = self.client.key_prefix();
180        let mut conn = self.conn().await?;
181        let cursor: u64 = filter
182            .cursor
183            .as_deref()
184            .unwrap_or("0")
185            .parse()
186            .map_err(|_| StoreError("invalid cursor".to_string()))?;
187        let count = if filter.limit > 0 { filter.limit } else { 50 };
188        let (next, keys): (u64, Vec<String>) = redis::cmd("SCAN")
189            .arg(cursor)
190            .arg("MATCH")
191            .arg(format!("{p}:sess:*:meta"))
192            .arg("COUNT")
193            .arg(count)
194            .query_async(&mut conn)
195            .await
196            .map_err(|e| self.err(format!("list scan: {e}")))?;
197        let mut sessions = Vec::new();
198        for key in keys {
199            let Some(id) = id_of_meta_key(p, &key) else {
200                continue;
201            };
202            let raw: Option<String> = conn
203                .get(&key)
204                .await
205                .map_err(|e| self.err(format!("list get: {e}")))?;
206            let Some(mut meta) = raw.and_then(|s| serde_json::from_str::<SessionMeta>(&s).ok())
207            else {
208                continue;
209            };
210            meta.id = id;
211            if filter.subject.as_deref().is_some_and(|s| meta.subject != s) {
212                continue;
213            }
214            if filter.plugin.as_deref().is_some_and(|pl| meta.plugin != pl) {
215                continue;
216            }
217            sessions.push(meta);
218        }
219        Ok(SessionPage {
220            sessions,
221            next_cursor: if next == 0 {
222                None
223            } else {
224                Some(next.to_string())
225            },
226        })
227    }
228
229    async fn try_lock(&self, id: &SessionId, ttl: Duration) -> Result<bool, StoreError> {
230        let p = self.client.key_prefix();
231        let mut conn = self.conn().await?;
232        let acquired: Option<String> = redis::cmd("SET")
233            .arg(lock_key(p, id.as_str()))
234            .arg("1")
235            .arg("NX")
236            .arg("PX")
237            .arg(ttl.as_millis().max(1) as u64)
238            .query_async(&mut conn)
239            .await
240            .map_err(|e| self.err(format!("try_lock: {e}")))?;
241        Ok(acquired.is_some())
242    }
243
244    async fn unlock(&self, id: &SessionId) -> Result<(), StoreError> {
245        let p = self.client.key_prefix();
246        let mut conn = self.conn().await?;
247        conn.del::<_, ()>(lock_key(p, id.as_str()))
248            .await
249            .map_err(|e| self.err(format!("unlock: {e}")))
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256
257    /// The hash-tag layout is the cluster-readiness contract: one id's keys
258    /// share a slot, and the meta-key parser inverts the builder.
259    #[test]
260    fn test_key_layout_and_meta_parse() {
261        assert_eq!(sess_key("fb", "ab12"), "fb:sess:{ab12}");
262        assert_eq!(meta_key("fb", "ab12"), "fb:sess:{ab12}:meta");
263        assert_eq!(lock_key("fb", "ab12"), "fb:lock:{ab12}");
264        assert!(subj_key("fb", "alice").starts_with("fb:subj:{"));
265        assert!(subj_key("fb", "alice").ends_with('}'));
266        assert_ne!(subj_key("fb", "alice"), subj_key("fb", "bob"));
267        assert_eq!(
268            id_of_meta_key("fb", "fb:sess:{ab12}:meta").as_deref(),
269            Some("ab12")
270        );
271        assert_eq!(id_of_meta_key("fb", "fb:sess:{ab12}"), None);
272    }
273
274    /// Live round-trip; skipped unless FEATHERBIT_TEST_REDIS_URL is set.
275    #[tokio::test]
276    async fn test_redis_session_store_live() {
277        let Ok(url) = std::env::var("FEATHERBIT_TEST_REDIS_URL") else {
278            eprintln!("skipping test_redis_session_store_live: FEATHERBIT_TEST_REDIS_URL not set");
279            return;
280        };
281        let cfg: crate::config::StoreConfig = serde_yaml::from_str(&format!(
282            "name: live\ntype: redis\nurl: {url}\nkey_prefix: fbsess{}\n",
283            std::process::id()
284        ))
285        .unwrap();
286        let client = Arc::new(RedisStoreClient::build(&cfg).unwrap());
287        let store = RedisSessionStore::new(client.clone(), None);
288
289        let id = SessionId::random();
290        let meta = SessionMeta {
291            id: String::new(),
292            subject: "alice".to_string(),
293            plugin: "openid-connect".to_string(),
294            policy: "p".to_string(),
295            route: "r".to_string(),
296            created_at: 1,
297            expires_at: 9999999999,
298        };
299        store
300            .put(&id, b"sealed-bytes", Duration::from_secs(60), &meta)
301            .await
302            .unwrap();
303        assert_eq!(
304            store.get(&id).await.unwrap().as_deref(),
305            Some(&b"sealed-bytes"[..])
306        );
307
308        // NX must have set a TTL on the fresh subject-index key (this is
309        // the FIRST put for "alice" in this test run, so the key was just
310        // created by SADD with no TTL of its own) — without it, EXPIRE ...
311        // GT alone would never apply (a key with no TTL is "infinite" for
312        // GT's comparison) and the index would persist forever.
313        let prefix = format!("fbsess{}", std::process::id());
314        let mut raw_conn = client.conn().await.unwrap();
315        let fresh_ttl: i64 = redis::cmd("TTL")
316            .arg(subj_key(&prefix, "alice"))
317            .query_async(&mut raw_conn)
318            .await
319            .unwrap();
320        assert!(
321            fresh_ttl > 0,
322            "fresh subject-index key should have a TTL set by NX, got {fresh_ttl}"
323        );
324
325        // Lock: winner/loser then release.
326        assert!(store.try_lock(&id, Duration::from_secs(5)).await.unwrap());
327        assert!(!store.try_lock(&id, Duration::from_secs(5)).await.unwrap());
328        store.unlock(&id).await.unwrap();
329
330        // List finds it (drain SCAN cursors until exhausted).
331        let mut cursor: Option<String> = None;
332        let mut found = false;
333        loop {
334            let page = store
335                .list(&SessionFilter {
336                    subject: Some("alice".to_string()),
337                    plugin: None,
338                    limit: 10,
339                    cursor: cursor.clone(),
340                })
341                .await
342                .unwrap();
343            if page.sessions.iter().any(|m| m.id == id.as_str()) {
344                found = true;
345            }
346            match page.next_cursor {
347                Some(c) => cursor = Some(c),
348                None => break,
349            }
350        }
351        assert!(found);
352
353        // Subject-index TTL must only grow (EXPIRE ... GT), never shrink: a
354        // short-lived session put after a long-lived one must not truncate
355        // the index below the longer member's lifetime. Put a second,
356        // short-lived session for the same subject and confirm
357        // `delete_subject` still revokes BOTH — without GT the second put's
358        // unconditional `expire()` would shrink the index to 1s, it would
359        // expire out from under the first (60s) member, and this would
360        // silently drop to 0/1 instead of 2.
361        let id2 = SessionId::random();
362        let meta2 = SessionMeta {
363            id: String::new(),
364            subject: "alice".to_string(),
365            plugin: "openid-connect".to_string(),
366            policy: "p".to_string(),
367            route: "r".to_string(),
368            created_at: 1,
369            expires_at: 9999999999,
370        };
371        store
372            .put(&id2, b"sealed-bytes-2", Duration::from_secs(1), &meta2)
373            .await
374            .unwrap();
375
376        // Revoke by subject.
377        assert_eq!(store.delete_subject("alice").await.unwrap(), 2);
378        assert_eq!(store.get(&id).await.unwrap(), None);
379        assert_eq!(store.get(&id2).await.unwrap(), None);
380    }
381}
featherbit/sessions/redis.rs

featherbit/sessions/
redis.rs
