authkestra_devsig/
replay.rs

1//! Replay protection.
2//!
3//! Deliberately not authkestra's `SessionStore`: this holds short-TTL, single-use `jti` markers,
4//! not sessions, and has entirely different lifecycle semantics (insert-once, never updated,
5//! evicted on TTL rather than on logout).
6
7use std::collections::HashMap;
8use std::sync::Mutex;
9use std::time::{Duration, Instant};
10
11use async_trait::async_trait;
12use thiserror::Error;
13
14/// Failure modes for a replay store. A request must fail closed on *any* `Err` here — not just
15/// [`ReplayError::Unavailable`] — meaning `verify()` treats a store outage identically to a
16/// genuine replay: reject, never silently allow.
17#[derive(Debug, Error, Clone, PartialEq, Eq)]
18pub enum ReplayError {
19    /// The store could not be reached (network partition, connection pool exhausted, etc.).
20    #[error("replay store unavailable: {0}")]
21    Unavailable(String),
22}
23
24/// Atomic single-use marker store, keyed by `jti`.
25///
26/// A real deployment should back this with something shared across every verifier instance in
27/// the trust domain (e.g. Redis with `SET key val NX PX ttl`) — if several verifiers each keep an
28/// independent local cache, a request replayed to a *different* verifier instance succeeds. See
29/// the crate-level docs for the "replay scope" requirement this trait exists to satisfy.
30#[async_trait]
31pub trait ReplayStore: Send + Sync + 'static {
32    /// Atomically records `jti` if and only if it is not already present.
33    ///
34    /// Returns `Ok(true)` if `jti` was newly inserted (request may proceed), `Ok(false)` if it
35    /// was already present (replay — reject). `Err` means the store itself failed; the caller
36    /// must treat this identically to `Ok(false)` (fail closed, never fail open).
37    async fn put_if_absent(&self, jti: &str, ttl: Duration) -> Result<bool, ReplayError>;
38}
39
40struct Entry {
41    inserted_at: Instant,
42    ttl: Duration,
43}
44
45impl Entry {
46    fn is_expired(&self, now: Instant) -> bool {
47        now.duration_since(self.inserted_at) >= self.ttl
48    }
49}
50
51/// A `HashMap`-backed [`ReplayStore`] for tests and single-process deployments.
52///
53/// Expired entries are lazily swept on each call rather than on a background timer — fine for a
54/// single process, but this does not share state across processes. Use a distributed store (e.g.
55/// Redis) for anything with more than one verifier instance.
56#[derive(Default)]
57pub struct InMemoryReplayStore {
58    entries: Mutex<HashMap<String, Entry>>,
59}
60
61impl InMemoryReplayStore {
62    /// Creates an empty store.
63    pub fn new() -> Self {
64        Self {
65            entries: Mutex::new(HashMap::new()),
66        }
67    }
68}
69
70#[async_trait]
71impl ReplayStore for InMemoryReplayStore {
72    async fn put_if_absent(&self, jti: &str, ttl: Duration) -> Result<bool, ReplayError> {
73        let now = Instant::now();
74        let mut entries = self.entries.lock().expect("replay store mutex poisoned");
75
76        entries.retain(|_, entry| !entry.is_expired(now));
77
78        if entries.contains_key(jti) {
79            return Ok(false);
80        }
81
82        entries.insert(
83            jti.to_string(),
84            Entry {
85                inserted_at: now,
86                ttl,
87            },
88        );
89        Ok(true)
90    }
91}
92
93/// A [`ReplayStore`] that always fails — for exercising the fail-closed path in tests (a replay
94/// store outage must still reject).
95pub struct UnavailableReplayStore;
96
97#[async_trait]
98impl ReplayStore for UnavailableReplayStore {
99    async fn put_if_absent(&self, _jti: &str, _ttl: Duration) -> Result<bool, ReplayError> {
100        Err(ReplayError::Unavailable("simulated outage".to_string()))
101    }
102}
authkestra_devsig/replay.rs

authkestra_devsig/
replay.rs
