1use std::collections::HashMap;
2use std::fmt;
3use std::sync::Arc;
4use std::time::{Duration, Instant};
5
6use async_trait::async_trait;
7use serde::{Deserialize, Serialize};
8use sha2::{Digest, Sha256};
9use tokio::sync::{Mutex, RwLock};
10
11use crate::http_client::build_ssrf_pinned_client;
12use crate::types::AuthError;
13use zeroize::Zeroizing;
14
15#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
16pub struct IntrospectionResult {
17 pub active: bool,
18 #[serde(default)]
19 pub sub: Option<String>,
20 #[serde(default)]
21 pub exp: Option<u64>,
22 #[serde(default)]
23 pub iat: Option<u64>,
24 #[serde(default)]
25 pub nbf: Option<u64>,
26 #[serde(default)]
27 pub scope: Option<String>,
28 #[serde(default)]
29 pub client_id: Option<String>,
30 #[serde(default)]
31 pub token_type: Option<String>,
32 #[serde(default)]
33 pub iss: Option<String>,
34 #[serde(default)]
35 pub aud: Option<serde_json::Value>,
36 #[serde(flatten)]
37 pub extra: serde_json::Map<String, serde_json::Value>,
38}
39
40#[derive(Debug, Clone)]
41pub struct IntrospectionCacheOptions {
42 pub max_entries: usize,
43 pub default_ttl: Duration,
44 pub negative_ttl: Duration,
45}
46
47impl Default for IntrospectionCacheOptions {
48 fn default() -> Self {
49 Self {
50 max_entries: 10_000,
51 default_ttl: Duration::from_secs(60),
52 negative_ttl: Duration::from_secs(5),
53 }
54 }
55}
56
57#[async_trait]
58pub trait TokenIntrospector: Send + Sync {
59 async fn introspect(&self, token: &str) -> Result<IntrospectionResult, AuthError>;
60}
61
62pub(crate) struct CachedEntry {
63 result: IntrospectionResult,
64 expires_at: Instant,
65}
66
67pub struct CachingTokenIntrospector {
68 endpoint: String,
69 client_id: String,
70 client_secret: Zeroizing<String>,
71 http: reqwest::Client,
72 pub(crate) cache: Arc<RwLock<HashMap<String, CachedEntry>>>,
73 in_flight: Mutex<()>,
74 max_cache_size: usize,
75 default_ttl: Duration,
76 negative_ttl: Duration,
77}
78
79impl CachingTokenIntrospector {
80 pub async fn new(
81 endpoint: String,
82 client_id: String,
83 client_secret: String,
84 options: IntrospectionCacheOptions,
85 ) -> Result<Self, AuthError> {
86 let http = build_ssrf_pinned_client(
87 &endpoint,
88 "introspection endpoint",
89 Duration::from_secs(5),
90 Duration::from_secs(10),
91 )
92 .await?;
93 Ok(Self::with_client(
94 endpoint,
95 client_id,
96 Zeroizing::new(client_secret),
97 options,
98 http,
99 ))
100 }
101
102 #[doc(hidden)]
103 pub fn new_unchecked_for_test(
104 endpoint: String,
105 client_id: String,
106 client_secret: String,
107 options: IntrospectionCacheOptions,
108 ) -> Self {
109 let http = reqwest::Client::builder()
110 .connect_timeout(Duration::from_secs(5))
111 .timeout(Duration::from_secs(10))
112 .build()
113 .expect("hardened HTTP client builder config is valid"); Self::with_client(
115 endpoint,
116 client_id,
117 Zeroizing::new(client_secret),
118 options,
119 http,
120 )
121 }
122
123 fn with_client(
124 endpoint: String,
125 client_id: String,
126 client_secret: Zeroizing<String>,
127 options: IntrospectionCacheOptions,
128 http: reqwest::Client,
129 ) -> Self {
130 Self {
131 endpoint,
132 client_id,
133 client_secret,
134 http,
135 cache: Arc::new(RwLock::new(HashMap::new())),
136 in_flight: Mutex::new(()),
137 max_cache_size: options.max_entries,
138 default_ttl: options.default_ttl,
139 negative_ttl: options.negative_ttl,
140 }
141 }
142
143 fn token_hash(token: &str) -> String {
144 let mut hasher = Sha256::new();
145 hasher.update(token.as_bytes());
146 hex::encode(hasher.finalize())
147 }
148
149 fn compute_ttl(&self, result: &IntrospectionResult) -> Duration {
150 if !result.active {
151 return self.negative_ttl;
152 }
153 if let Some(exp) = result.exp {
154 let now = std::time::SystemTime::now()
155 .duration_since(std::time::UNIX_EPOCH)
156 .unwrap_or_default()
157 .as_secs();
158 if exp > now {
159 let remaining = Duration::from_secs(exp - now);
160 return remaining.min(self.default_ttl);
161 }
162 }
163 self.default_ttl
164 }
165
166 async fn evict_if_needed(&self) {
167 let mut cache = self.cache.write().await;
168 if cache.len() < self.max_cache_size {
169 return;
170 }
171 let now = Instant::now();
172 cache.retain(|_, entry| entry.expires_at > now);
173 if cache.len() >= self.max_cache_size {
174 let oldest_key = cache
175 .iter()
176 .min_by_key(|(_, e)| e.expires_at)
177 .map(|(k, _)| k.clone());
178 if let Some(key) = oldest_key {
179 cache.remove(&key);
180 }
181 }
182 }
183}
184
185impl fmt::Debug for CachingTokenIntrospector {
186 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
187 f.debug_struct("CachingTokenIntrospector")
188 .field("endpoint", &self.endpoint)
189 .field("client_id", &self.client_id)
190 .field("client_secret", &"[REDACTED]")
191 .field("max_cache_size", &self.max_cache_size)
192 .field("default_ttl", &self.default_ttl)
193 .field("negative_ttl", &self.negative_ttl)
194 .finish_non_exhaustive()
195 }
196}
197
198#[async_trait]
199impl TokenIntrospector for CachingTokenIntrospector {
200 async fn introspect(&self, token: &str) -> Result<IntrospectionResult, AuthError> {
201 let key = Self::token_hash(token);
202
203 {
204 let cache = self.cache.read().await;
205 if let Some(entry) = cache.get(&key)
206 && entry.expires_at > Instant::now()
207 {
208 tracing::debug!(target: "camel_auth::introspection", cache_outcome = "hit");
209 return Ok(entry.result.clone());
210 }
211 }
212
213 let _guard = self.in_flight.lock().await;
214
215 {
216 let cache = self.cache.read().await;
217 if let Some(entry) = cache.get(&key)
218 && entry.expires_at > Instant::now()
219 {
220 tracing::debug!(target: "camel_auth::introspection", cache_outcome = "hit_after_wait");
221 return Ok(entry.result.clone());
222 }
223 }
224
225 tracing::debug!(
226 target: "camel_auth::introspection",
227 cache_outcome = "miss"
228 );
229
230 let response = self
231 .http
232 .post(&self.endpoint)
233 .form(&[
234 ("token", token),
235 ("client_id", &self.client_id),
236 ("client_secret", self.client_secret.as_str()),
237 ])
238 .send()
239 .await
240 .map_err(|e| {
241 AuthError::ProviderUnavailable(format!("introspection request failed: {e}"))
242 })?;
243
244 let status = response.status();
245 if status.as_u16() == 401 || status.as_u16() == 403 {
246 return Err(AuthError::ProviderUnavailable(
247 "introspection client unauthorized".into(),
248 ));
249 }
250 if status.is_server_error() {
251 return Err(AuthError::ProviderUnavailable(format!(
252 "introspection endpoint returned {}",
253 status
254 )));
255 }
256 if status.is_client_error() {
257 return Err(AuthError::TokenInvalid(format!(
258 "introspection endpoint returned client error {}",
259 status
260 )));
261 }
262
263 let result: IntrospectionResult = response.json().await.map_err(|e| {
264 AuthError::ProviderUnavailable(format!("invalid introspection response: {e}"))
265 })?;
266
267 let ttl = self.compute_ttl(&result);
268 let entry = CachedEntry {
269 result: result.clone(),
270 expires_at: Instant::now() + ttl,
271 };
272
273 self.evict_if_needed().await;
274 {
275 let mut cache = self.cache.write().await;
276 cache.insert(key, entry);
277 }
278
279 Ok(result)
280 }
281}
282
283#[cfg(test)]
284mod tests {
285 use super::*;
286 use wiremock::matchers::{body_string_contains, method};
287 use wiremock::{Mock, MockServer, ResponseTemplate};
288
289 #[test]
290 fn deserialize_minimal_active() {
291 let json = r#"{"active": true}"#;
292 let result: IntrospectionResult = serde_json::from_str(json).unwrap();
293 assert!(result.active);
294 assert!(result.sub.is_none());
295 assert!(result.extra.is_empty());
296 }
297
298 #[test]
299 fn deserialize_full_rfc7662() {
300 let json = r#"{
301 "active": true,
302 "sub": "user-1",
303 "exp": 1700000000,
304 "iat": 1699999999,
305 "nbf": 1699999900,
306 "scope": "read write",
307 "client_id": "my-client",
308 "token_type": "Bearer",
309 "iss": "https://kc.example.com/realms/test",
310 "aud": ["my-api"],
311 "realm_access": {"roles": ["admin", "user"]},
312 "resource_access": {"my-client": {"roles": ["client-role"]}}
313 }"#;
314 let result: IntrospectionResult = serde_json::from_str(json).unwrap();
315 assert!(result.active);
316 assert_eq!(result.sub.as_deref(), Some("user-1"));
317 assert_eq!(result.exp, Some(1700000000));
318 assert_eq!(result.scope.as_deref(), Some("read write"));
319 assert_eq!(result.client_id.as_deref(), Some("my-client"));
320 assert_eq!(result.token_type.as_deref(), Some("Bearer"));
321 assert_eq!(
322 result.iss.as_deref(),
323 Some("https://kc.example.com/realms/test")
324 );
325 assert!(result.extra.contains_key("realm_access"));
326 assert!(result.extra.contains_key("resource_access"));
327 }
328
329 #[test]
330 fn deserialize_inactive() {
331 let json = r#"{"active": false}"#;
332 let result: IntrospectionResult = serde_json::from_str(json).unwrap();
333 assert!(!result.active);
334 }
335
336 #[test]
337 fn deserialize_unknown_fields_go_to_extra() {
338 let json = r#"{"active": true, "custom_field": "hello"}"#;
339 let result: IntrospectionResult = serde_json::from_str(json).unwrap();
340 assert_eq!(result.extra["custom_field"], "hello");
341 }
342
343 #[test]
344 fn cache_options_defaults() {
345 let opts = IntrospectionCacheOptions::default();
346 assert_eq!(opts.max_entries, 10_000);
347 assert_eq!(opts.default_ttl, Duration::from_secs(60));
348 assert_eq!(opts.negative_ttl, Duration::from_secs(5));
349 }
350
351 fn test_cache_opts() -> IntrospectionCacheOptions {
352 IntrospectionCacheOptions {
353 max_entries: 100,
354 default_ttl: Duration::from_secs(60),
355 negative_ttl: Duration::from_secs(2),
356 }
357 }
358
359 #[tokio::test]
360 async fn cache_hit_returns_cached_result_without_http_call() {
361 let server = MockServer::start().await;
362 Mock::given(method("POST"))
363 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
364 "active": true,
365 "sub": "cached-user"
366 })))
367 .expect(1)
368 .mount(&server)
369 .await;
370
371 let introspector = CachingTokenIntrospector::new_unchecked_for_test(
372 server.uri(),
373 "client-id".into(),
374 "client-secret".into(),
375 test_cache_opts(),
376 );
377
378 let r1 = introspector.introspect("token-a").await.unwrap();
379 let r2 = introspector.introspect("token-a").await.unwrap();
380 assert_eq!(r1.sub, r2.sub);
381 assert!(r1.active);
382 }
383
384 #[tokio::test]
385 async fn expired_entry_re_introspects() {
386 let server = MockServer::start().await;
387 Mock::given(method("POST"))
388 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
389 "active": true, "sub": "user"
390 })))
391 .expect(2)
392 .mount(&server)
393 .await;
394
395 let opts = IntrospectionCacheOptions {
396 max_entries: 100,
397 default_ttl: Duration::from_millis(50),
398 negative_ttl: Duration::from_secs(2),
399 };
400 let introspector = CachingTokenIntrospector::new_unchecked_for_test(
401 server.uri(),
402 "cid".into(),
403 "cs".into(),
404 opts,
405 );
406
407 introspector.introspect("tok").await.unwrap();
408 tokio::time::sleep(Duration::from_millis(80)).await;
409 introspector.introspect("tok").await.unwrap();
410 }
411
412 #[tokio::test]
413 async fn inactive_token_cached_with_negative_ttl() {
414 let server = MockServer::start().await;
415 Mock::given(method("POST"))
416 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
417 "active": false
418 })))
419 .expect(1)
420 .mount(&server)
421 .await;
422
423 let opts = IntrospectionCacheOptions {
424 max_entries: 100,
425 default_ttl: Duration::from_secs(60),
426 negative_ttl: Duration::from_secs(10),
427 };
428 let introspector = CachingTokenIntrospector::new_unchecked_for_test(
429 server.uri(),
430 "cid".into(),
431 "cs".into(),
432 opts,
433 );
434
435 let r = introspector.introspect("dead-token").await.unwrap();
436 assert!(!r.active);
437 let r2 = introspector.introspect("dead-token").await.unwrap();
438 assert!(!r2.active);
439 }
440
441 #[tokio::test]
442 async fn cache_key_does_not_contain_raw_token() {
443 let server = MockServer::start().await;
444 Mock::given(method("POST"))
445 .respond_with(
446 ResponseTemplate::new(200).set_body_json(serde_json::json!({"active": true})),
447 )
448 .mount(&server)
449 .await;
450
451 let introspector = CachingTokenIntrospector::new_unchecked_for_test(
452 server.uri(),
453 "cid".into(),
454 "cs".into(),
455 test_cache_opts(),
456 );
457
458 introspector.introspect("secret-token-value").await.unwrap();
459 let cache = introspector.cache.read().await;
460 for key in cache.keys() {
461 assert!(
462 !key.contains("secret-token-value"),
463 "cache key must not contain raw token"
464 );
465 }
466 }
467
468 #[tokio::test]
469 async fn eviction_removes_oldest_when_over_capacity() {
470 let server = MockServer::start().await;
471 for i in 0..5 {
472 Mock::given(method("POST"))
473 .and(body_string_contains(format!("token-{i}"))) .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
475 "active": true, "sub": format!("user-{i}")
476 })))
477 .mount(&server)
478 .await;
479 }
480
481 let opts = IntrospectionCacheOptions {
482 max_entries: 2,
483 default_ttl: Duration::from_secs(600),
484 negative_ttl: Duration::from_secs(5),
485 };
486 let introspector = CachingTokenIntrospector::new_unchecked_for_test(
487 server.uri(),
488 "cid".into(),
489 "cs".into(),
490 opts,
491 );
492
493 introspector.introspect("token-0").await.unwrap();
494 introspector.introspect("token-1").await.unwrap();
495 introspector.introspect("token-2").await.unwrap();
496 introspector.introspect("token-3").await.unwrap();
497 introspector.introspect("token-4").await.unwrap();
498
499 let cache = introspector.cache.read().await;
500 assert!(cache.len() <= 2, "cache must respect max_entries");
501 }
502
503 #[test]
504 fn debug_redacts_client_secret() {
505 let introspector = CachingTokenIntrospector::new_unchecked_for_test(
506 "https://example.com".into(),
507 "cid".into(),
508 "super-secret-value".into(),
509 test_cache_opts(),
510 );
511 let debug = format!("{introspector:?}");
512 assert!(
513 !debug.contains("super-secret-value"),
514 "Debug must not leak client_secret"
515 );
516 assert!(debug.contains("REDACTED"));
517 }
518
519 #[tokio::test]
520 async fn production_constructor_rejects_http_endpoint() {
521 let result = CachingTokenIntrospector::new(
522 "http://insecure.example.com/introspect".into(),
523 "cid".into(),
524 "cs".into(),
525 test_cache_opts(),
526 )
527 .await;
528 assert!(result.is_err());
529 let err = result.unwrap_err();
530 assert!(matches!(err, AuthError::ConfigError(ref s) if s.contains("HTTPS")));
531 }
532
533 #[tokio::test]
534 async fn production_constructor_rejects_localhost() {
535 let result = CachingTokenIntrospector::new(
536 "https://localhost:8080/introspect".into(),
537 "cid".into(),
538 "cs".into(),
539 test_cache_opts(),
540 )
541 .await;
542 assert!(result.is_err());
543 let err = result.unwrap_err();
544 assert!(
545 matches!(err, AuthError::ConfigError(ref s) if s.contains("private") || s.contains("loopback"))
546 );
547 }
548
549 #[tokio::test]
550 async fn http_error_500_returns_provider_unavailable() {
551 let server = MockServer::start().await;
552 Mock::given(method("POST"))
553 .respond_with(ResponseTemplate::new(500))
554 .mount(&server)
555 .await;
556
557 let introspector = CachingTokenIntrospector::new_unchecked_for_test(
558 server.uri(),
559 "cid".into(),
560 "cs".into(),
561 test_cache_opts(),
562 );
563 let result = introspector.introspect("tok").await;
564 assert!(result.is_err());
565 let err = result.unwrap_err();
566 assert!(matches!(err, AuthError::ProviderUnavailable(_)));
567 }
568
569 #[tokio::test]
570 async fn http_401_returns_provider_unavailable() {
571 let server = MockServer::start().await;
572 Mock::given(method("POST"))
573 .respond_with(ResponseTemplate::new(401))
574 .mount(&server)
575 .await;
576
577 let introspector = CachingTokenIntrospector::new_unchecked_for_test(
578 server.uri(),
579 "cid".into(),
580 "cs".into(),
581 test_cache_opts(),
582 );
583 let result = introspector.introspect("tok").await;
584 assert!(result.is_err());
585 let err = result.unwrap_err();
586 assert!(matches!(err, AuthError::ProviderUnavailable(ref s) if s.contains("unauthorized")));
587 }
588}