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