use chrono::{DateTime, Utc};
use delegated::{TrustStateAdmin, TrustStateError, TrustStateStore};
use redis::{Client, Commands, RedisError};
pub struct RedisTrustState {
client: Client,
prefix: String,
}
impl RedisTrustState {
pub fn new(redis_url: &str) -> Result<Self, TrustStateError> {
Self::with_prefix(redis_url, "delegated")
}
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);
}
}