delegated-redis 0.2.2

#![doc = include_str!("../README.md")]
#![deny(missing_docs)]

//! Redis-backed trust state for [`delegated`] evaluators.
//!
//! This crate provides [`RedisTrustState`], a production-oriented implementation of
//! [`TrustStateStore`] and [`TrustStateAdmin`] using Redis atomic `SET NX` for nonce
//! consumption and persistent flags for revocation and emergency deny.
//!
//! See the [operations guide](https://github.com/abendrothj/delegated/blob/main/crates/delegated-redis/OPERATIONS.md)
//! for key layout, Sentinel/cluster notes, and lifecycle procedures.
//!
//! [`delegated`]: https://docs.rs/delegated
//! [`TrustStateStore`]: delegated::TrustStateStore
//! [`TrustStateAdmin`]: delegated::TrustStateAdmin

use chrono::{DateTime, Utc};
use delegated::{TrustStateAdmin, TrustStateError, TrustStateStore};
use redis::{Client, Commands, RedisError};

/// Shared trust state backed by Redis `SET NX EX` nonce consumption.
///
/// Connect one instance per process and reuse it across requests. Constructing a
/// fresh store per request adds connection overhead without security benefit.
///
/// # Examples
///
/// ```
/// use delegated::TrustStateStore;
/// use delegated_redis::RedisTrustState;
///
/// # fn example() -> Result<(), delegated::TrustStateError> {
/// let state = RedisTrustState::new("redis://127.0.0.1/")?;
/// assert!(!state.is_token_revoked("https://issuer.example", "token-1")?);
/// # Ok(())
/// # }
/// ```
pub struct RedisTrustState {
    client: Client,
    prefix: String,
}

impl RedisTrustState {
    /// Connects to Redis with the default `delegated` key prefix.
    ///
    /// `redis_url` must be a URL understood by the `redis` crate, for example
    /// `redis://127.0.0.1:6379` or `rediss://user:pass@host:6380/0`.
    pub fn new(redis_url: &str) -> Result<Self, TrustStateError> {
        Self::with_prefix(redis_url, "delegated")
    }

    /// Connects to Redis with a custom key prefix for multi-tenant clusters.
    ///
    /// Use a distinct prefix per environment or authorization surface that shares
    /// the same Redis deployment.
    pub fn with_prefix(
        redis_url: &str,
        prefix: impl Into<String>,
    ) -> Result<Self, TrustStateError> {
        let client = Client::open(redis_url).map_err(map_error)?;
        Ok(Self {
            client,
            prefix: prefix.into(),
        })
    }

    fn key(&self, category: &str, issuer: &str, identifier: &str) -> String {
        format!("{}:{}:{}:{}", self.prefix, category, issuer, identifier)
    }

    fn connection(&self) -> Result<redis::Connection, TrustStateError> {
        self.client.get_connection().map_err(map_error)
    }

    fn exists(&self, key: &str) -> Result<bool, TrustStateError> {
        let mut connection = self.connection()?;
        connection.exists(key).map_err(map_error)
    }

    fn set_flag(&self, key: &str) -> Result<(), TrustStateError> {
        let mut connection = self.connection()?;
        connection.set::<_, _, ()>(key, "1").map_err(map_error)
    }
}

impl TrustStateStore for RedisTrustState {
    fn is_token_revoked(&self, issuer: &str, token_id: &str) -> Result<bool, TrustStateError> {
        self.exists(&self.key("revoked", issuer, token_id))
    }

    fn is_agent_denied(&self, issuer: &str, agent_id: &str) -> Result<bool, TrustStateError> {
        self.exists(&self.key("denied", issuer, agent_id))
    }

    fn consume_nonce(
        &self,
        issuer: &str,
        nonce: &str,
        observed_at: DateTime<Utc>,
        valid_until: DateTime<Utc>,
    ) -> Result<bool, TrustStateError> {
        let ttl = (valid_until - observed_at)
            .num_seconds()
            .max(1)
            .min(i64::from(u32::MAX)) as u64;
        let key = self.key("nonce", issuer, nonce);
        let mut connection = self.connection()?;
        let set: Option<String> = redis::cmd("SET")
            .arg(&key)
            .arg("1")
            .arg("NX")
            .arg("EX")
            .arg(ttl)
            .query(&mut connection)
            .map_err(map_error)?;
        Ok(set.is_some())
    }
}

impl TrustStateAdmin for RedisTrustState {
    fn revoke_token(&self, issuer: &str, token_id: &str) -> Result<(), TrustStateError> {
        self.set_flag(&self.key("revoked", issuer, token_id))
    }

    fn deny_agent(&self, issuer: &str, agent_id: &str) -> Result<(), TrustStateError> {
        self.set_flag(&self.key("denied", issuer, agent_id))
    }
}

fn map_error(error: RedisError) -> TrustStateError {
    TrustStateError::new(error.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::{Duration, TimeZone, Utc};
    use delegated::{DelegationTokenBuilder, Evaluator, OperationContext, PinnedIssuerKeys};
    use ed25519_dalek::SigningKey;
    use std::sync::{Arc, Barrier};
    use std::thread;

    fn redis_url() -> Option<String> {
        std::env::var("REDIS_URL")
            .ok()
            .filter(|value| !value.is_empty())
    }

    fn unique_prefix() -> String {
        format!(
            "delegated-test-{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_nanos()
        )
    }

    fn now() -> DateTime<Utc> {
        Utc.with_ymd_and_hms(2026, 7, 1, 12, 0, 0).single().unwrap()
    }

    #[test]
    fn redis_nonce_consumption_is_single_use() {
        let Some(url) = redis_url() else {
            eprintln!("skipping redis_nonce_consumption_is_single_use: REDIS_URL unset");
            return;
        };
        let state = RedisTrustState::with_prefix(&url, unique_prefix()).unwrap();
        assert!(
            state
                .consume_nonce(
                    "issuer",
                    "nonce-000000000001",
                    now(),
                    now() + Duration::hours(1)
                )
                .unwrap()
        );
        assert!(
            !state
                .consume_nonce(
                    "issuer",
                    "nonce-000000000001",
                    now(),
                    now() + Duration::hours(1)
                )
                .unwrap()
        );
    }

    #[test]
    fn redis_revocation_and_deny_are_visible_to_evaluator() {
        let Some(url) = redis_url() else {
            eprintln!(
                "skipping redis_revocation_and_deny_are_visible_to_evaluator: REDIS_URL unset"
            );
            return;
        };
        let state = RedisTrustState::with_prefix(&url, unique_prefix()).unwrap();
        state
            .revoke_token("https://issuer.example", "token-1")
            .unwrap();
        assert!(
            state
                .is_token_revoked("https://issuer.example", "token-1")
                .unwrap()
        );
        state
            .deny_agent("https://issuer.example", "agent:bad")
            .unwrap();
        assert!(
            state
                .is_agent_denied("https://issuer.example", "agent:bad")
                .unwrap()
        );
    }

    #[test]
    fn redis_nonce_is_atomic_under_concurrency() {
        let Some(url) = redis_url() else {
            eprintln!("skipping redis_nonce_is_atomic_under_concurrency: REDIS_URL unset");
            return;
        };
        let prefix = unique_prefix();
        let key = SigningKey::from_bytes(&[7; 32]);
        let keys = PinnedIssuerKeys::new();
        keys.insert("https://issuer.example", "issuer-1", key.verifying_key())
            .unwrap();
        let issued = now() - Duration::minutes(1);
        let expires = now() + Duration::minutes(10);
        let token = DelegationTokenBuilder::new()
            .token_id("token-1")
            .issuer("https://issuer.example")
            .agent_id("agent:scheduler")
            .delegator_id("user:alice")
            .audience("calendar-api")
            .allowed_action("calendar.create")
            .allowed_resource("calendar:alice")
            .max_delegation_depth(0)
            .issued_at(issued)
            .expires_at(expires)
            .nonce("nonce-concurrent-001")
            .key_id("issuer-1")
            .build_and_sign(&key)
            .unwrap();
        let request =
            serde_json::to_vec(&delegated::envelope(token, Some("request-1".into()))).unwrap();
        let operation = OperationContext::new("calendar-api", "calendar.create")
            .with_resource("calendar:alice")
            .with_delegation_depth(0);

        let state = Arc::new(RedisTrustState::with_prefix(&url, prefix).unwrap());
        let request = Arc::new(request);
        let keys = Arc::new(keys);
        let barrier = Arc::new(Barrier::new(8));
        let mut workers = Vec::new();
        for _ in 0..8 {
            let state = Arc::clone(&state);
            let request = Arc::clone(&request);
            let keys = Arc::clone(&keys);
            let barrier = Arc::clone(&barrier);
            let operation = operation.clone();
            workers.push(thread::spawn(move || {
                barrier.wait();
                Evaluator::new(keys.as_ref(), state.as_ref())
                    .evaluate(request.as_ref(), &operation, now())
                    .0
                    .allowed
            }));
        }
        let allowed = workers
            .into_iter()
            .map(|worker| worker.join().unwrap())
            .filter(|allowed| *allowed)
            .count();
        assert_eq!(allowed, 1);
    }
}