camel_auth/
introspection.rs

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
68/// ADR-0051 credential boundary: manual-redaction
69pub struct CachingTokenIntrospector {
70    endpoint: String,
71    client_id: String,
72    client_secret: Zeroizing<String>,
73    http: reqwest::Client,
74    pub(crate) cache: Arc<RwLock<HashMap<String, CachedEntry>>>,
75    in_flight: Mutex<HashMap<String, Arc<Mutex<()>>>>,
76    max_cache_size: usize,
77    default_ttl: Duration,
78    negative_ttl: Duration,
79}
80
81impl CachingTokenIntrospector {
82    pub async fn new(
83        endpoint: String,
84        client_id: String,
85        client_secret: String,
86        options: IntrospectionCacheOptions,
87        policy: SsrfPolicy,
88    ) -> Result<Self, AuthError> {
89        let http = build_ssrf_pinned_client(
90            &endpoint,
91            "introspection endpoint",
92            &SsrfClientOptions::new(policy)
93                .with_connect_timeout(Duration::from_secs(5))
94                .with_request_timeout(Duration::from_secs(10)),
95        )
96        .await?;
97        Ok(Self::with_client(
98            endpoint,
99            client_id,
100            Zeroizing::new(client_secret),
101            options,
102            http,
103        ))
104    }
105
106    #[doc(hidden)]
107    pub fn new_unchecked_for_test(
108        endpoint: String,
109        client_id: String,
110        client_secret: String,
111        options: IntrospectionCacheOptions,
112    ) -> Result<Self, AuthError> {
113        // Typed error instead of a panic (rc-3j4mq sibling sweep): on a
114        // CA-less platform (e.g. Android/Termux) the platform verifier
115        // fails this build — callers get a config-class error, matching
116        // the AuthError::ConfigError precedent in http_client.rs.
117        let http = reqwest::Client::builder()
118            .connect_timeout(Duration::from_secs(5))
119            .timeout(Duration::from_secs(10))
120            .build()
121            .map_err(|e| {
122                AuthError::ConfigError(format!("introspection HTTP client build failed: {e}"))
123            })?;
124        Ok(Self::with_client(
125            endpoint,
126            client_id,
127            Zeroizing::new(client_secret),
128            options,
129            http,
130        ))
131    }
132
133    fn with_client(
134        endpoint: String,
135        client_id: String,
136        client_secret: Zeroizing<String>,
137        options: IntrospectionCacheOptions,
138        http: reqwest::Client,
139    ) -> Self {
140        Self {
141            endpoint,
142            client_id,
143            client_secret,
144            http,
145            cache: Arc::new(RwLock::new(HashMap::new())),
146            in_flight: Mutex::new(HashMap::new()),
147            max_cache_size: options.max_entries,
148            default_ttl: options.default_ttl,
149            negative_ttl: options.negative_ttl,
150        }
151    }
152
153    fn token_hash(token: &str) -> String {
154        let mut hasher = Sha256::new();
155        hasher.update(token.as_bytes());
156        hex::encode(hasher.finalize())
157    }
158
159    fn compute_ttl(&self, result: &IntrospectionResult) -> Duration {
160        if !result.active {
161            return self.negative_ttl;
162        }
163        if let Some(exp) = result.exp {
164            let now = std::time::SystemTime::now()
165                .duration_since(std::time::UNIX_EPOCH)
166                .unwrap_or_default()
167                .as_secs();
168            if exp > now {
169                let remaining = Duration::from_secs(exp - now);
170                return remaining.min(self.default_ttl);
171            }
172        }
173        self.default_ttl
174    }
175
176    async fn evict_if_needed(&self) {
177        let mut cache = self.cache.write().await;
178        if cache.len() < self.max_cache_size {
179            return;
180        }
181        let now = Instant::now();
182        cache.retain(|_, entry| entry.expires_at > now);
183        if cache.len() >= self.max_cache_size {
184            let oldest_key = cache
185                .iter()
186                .min_by_key(|(_, e)| e.expires_at)
187                .map(|(k, _)| k.clone());
188            if let Some(key) = oldest_key {
189                cache.remove(&key);
190            }
191        }
192    }
193}
194
195impl fmt::Debug for CachingTokenIntrospector {
196    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
197        f.debug_struct("CachingTokenIntrospector")
198            .field("endpoint", &self.endpoint)
199            .field("client_id", &self.client_id)
200            .field("client_secret", &"[REDACTED]")
201            .field("max_cache_size", &self.max_cache_size)
202            .field("default_ttl", &self.default_ttl)
203            .field("negative_ttl", &self.negative_ttl)
204            .finish_non_exhaustive()
205    }
206}
207
208#[async_trait]
209impl TokenIntrospector for CachingTokenIntrospector {
210    async fn introspect(&self, token: &str) -> Result<IntrospectionResult, AuthError> {
211        let key = Self::token_hash(token);
212
213        {
214            let cache = self.cache.read().await;
215            if let Some(entry) = cache.get(&key)
216                && entry.expires_at > Instant::now()
217            {
218                tracing::debug!(target: "camel_auth::introspection", cache_outcome = "hit");
219                return Ok(entry.result.clone());
220            }
221        }
222
223        // Get-or-insert a per-key mutex so introspection of different tokens runs
224        // in parallel while introspection of the same token is still single-flighted.
225        //
226        // Duplicated from CachingPermissionEvaluator::evaluate (permission_cache.rs).
227        // Two identical call sites — extraction would add indirection without
228        // reducing total LoC. Keep both in sync when modifying.
229        let key_mutex = {
230            let mut in_flight_map = self.in_flight.lock().await;
231            in_flight_map
232                .entry(key.clone())
233                .or_insert_with(|| Arc::new(Mutex::new(())))
234                .clone()
235        };
236
237        // Do the work inside a block that catches all `?` early returns. The
238        // per-key guard drops at the end of the block on both success and error
239        // paths, releasing the single-flight slot before cleanup runs.
240        let result: Result<IntrospectionResult, AuthError> = async {
241            let _guard = key_mutex.lock().await;
242            // double-check cache (hit-after-wait)
243            {
244                let cache = self.cache.read().await;
245                if let Some(entry) = cache.get(&key)
246                    && entry.expires_at > Instant::now()
247                {
248                    tracing::debug!(target: "camel_auth::introspection", cache_outcome = "hit_after_wait");
249                    return Ok(entry.result.clone());
250                }
251            }
252
253            tracing::debug!(
254                target: "camel_auth::introspection",
255                cache_outcome = "miss"
256            );
257
258            let response = self
259                .http
260                .post(&self.endpoint)
261                .form(&[
262                    ("token", token),
263                    ("client_id", &self.client_id),
264                    ("client_secret", self.client_secret.as_str()),
265                ])
266                .send()
267                .await
268                .map_err(|e| {
269                    AuthError::ProviderUnavailable(format!("introspection request failed: {e}"))
270                })?;
271
272            let status = response.status();
273            if status.as_u16() == 401 || status.as_u16() == 403 {
274                return Err(AuthError::ProviderUnavailable(
275                    "introspection client unauthorized".into(),
276                ));
277            }
278            if status.is_server_error() {
279                return Err(AuthError::ProviderUnavailable(format!(
280                    "introspection endpoint returned {}",
281                    status
282                )));
283            }
284            if status.is_client_error() {
285                return Err(AuthError::TokenInvalid(format!(
286                    "introspection endpoint returned client error {}",
287                    status
288                )));
289            }
290
291            let result: IntrospectionResult = response.json().await.map_err(|e| {
292                AuthError::ProviderUnavailable(format!("invalid introspection response: {e}"))
293            })?;
294
295            let ttl = self.compute_ttl(&result);
296            let entry = CachedEntry {
297                result: result.clone(),
298                expires_at: Instant::now() + ttl,
299            };
300
301            self.evict_if_needed().await;
302            {
303                let mut cache = self.cache.write().await;
304                cache.insert(key.clone(), entry);
305            }
306
307            Ok(result)
308        }
309        .await;
310
311        // Release our clone of the Arc before cleanup; then the map entry is the
312        // sole owner (strong_count == 1) iff no other caller is in flight for the
313        // same key. The outer mutex serializes this test-and-remove against the
314        // get-or-insert above, so concurrent callers cannot race the removal.
315        drop(key_mutex);
316        {
317            let mut in_flight_map = self.in_flight.lock().await;
318            if let Some(arc) = in_flight_map.get(&key)
319                && Arc::strong_count(arc) == 1
320            {
321                in_flight_map.remove(&key);
322            }
323        }
324        result
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use super::*;
331    use wiremock::matchers::{body_string_contains, method};
332    use wiremock::{Mock, MockServer, ResponseTemplate};
333
334    #[test]
335    fn deserialize_minimal_active() {
336        let json = r#"{"active": true}"#;
337        let result: IntrospectionResult = serde_json::from_str(json).unwrap();
338        assert!(result.active);
339        assert!(result.sub.is_none());
340        assert!(result.extra.is_empty());
341    }
342
343    #[test]
344    fn deserialize_full_rfc7662() {
345        let json = r#"{
346            "active": true,
347            "sub": "user-1",
348            "exp": 1700000000,
349            "iat": 1699999999,
350            "nbf": 1699999900,
351            "scope": "read write",
352            "client_id": "my-client",
353            "token_type": "Bearer",
354            "iss": "https://kc.example.com/realms/test",
355            "aud": ["my-api"],
356            "realm_access": {"roles": ["admin", "user"]},
357            "resource_access": {"my-client": {"roles": ["client-role"]}}
358        }"#;
359        let result: IntrospectionResult = serde_json::from_str(json).unwrap();
360        assert!(result.active);
361        assert_eq!(result.sub.as_deref(), Some("user-1"));
362        assert_eq!(result.exp, Some(1700000000));
363        assert_eq!(result.scope.as_deref(), Some("read write"));
364        assert_eq!(result.client_id.as_deref(), Some("my-client"));
365        assert_eq!(result.token_type.as_deref(), Some("Bearer"));
366        assert_eq!(
367            result.iss.as_deref(),
368            Some("https://kc.example.com/realms/test")
369        );
370        assert!(result.extra.contains_key("realm_access"));
371        assert!(result.extra.contains_key("resource_access"));
372    }
373
374    #[test]
375    fn deserialize_inactive() {
376        let json = r#"{"active": false}"#;
377        let result: IntrospectionResult = serde_json::from_str(json).unwrap();
378        assert!(!result.active);
379    }
380
381    #[test]
382    fn deserialize_unknown_fields_go_to_extra() {
383        let json = r#"{"active": true, "custom_field": "hello"}"#;
384        let result: IntrospectionResult = serde_json::from_str(json).unwrap();
385        assert_eq!(result.extra["custom_field"], "hello");
386    }
387
388    #[test]
389    fn cache_options_defaults() {
390        let opts = IntrospectionCacheOptions::default();
391        assert_eq!(opts.max_entries, 10_000);
392        assert_eq!(opts.default_ttl, Duration::from_secs(60));
393        assert_eq!(opts.negative_ttl, Duration::from_secs(5));
394    }
395
396    fn test_cache_opts() -> IntrospectionCacheOptions {
397        IntrospectionCacheOptions {
398            max_entries: 100,
399            default_ttl: Duration::from_secs(60),
400            negative_ttl: Duration::from_secs(2),
401        }
402    }
403
404    #[tokio::test]
405    async fn cache_hit_returns_cached_result_without_http_call() {
406        let server = MockServer::start().await;
407        Mock::given(method("POST"))
408            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
409                "active": true,
410                "sub": "cached-user"
411            })))
412            .expect(1)
413            .mount(&server)
414            .await;
415
416        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
417            server.uri(),
418            "client-id".into(),
419            "client-secret".into(),
420            test_cache_opts(),
421        )
422        .expect("introspector"); // allow-unwrap(test)
423
424        let r1 = introspector.introspect("token-a").await.unwrap();
425        let r2 = introspector.introspect("token-a").await.unwrap();
426        assert_eq!(r1.sub, r2.sub);
427        assert!(r1.active);
428    }
429
430    #[tokio::test]
431    async fn expired_entry_re_introspects() {
432        let server = MockServer::start().await;
433        Mock::given(method("POST"))
434            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
435                "active": true, "sub": "user"
436            })))
437            .expect(2)
438            .mount(&server)
439            .await;
440
441        let opts = IntrospectionCacheOptions {
442            max_entries: 100,
443            default_ttl: Duration::from_millis(50),
444            negative_ttl: Duration::from_secs(2),
445        };
446        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
447            server.uri(),
448            "cid".into(),
449            "cs".into(),
450            opts,
451        )
452        .expect("introspector"); // allow-unwrap(test)
453
454        introspector.introspect("tok").await.unwrap();
455        tokio::time::sleep(Duration::from_millis(80)).await;
456        introspector.introspect("tok").await.unwrap();
457    }
458
459    #[tokio::test]
460    async fn inactive_token_cached_with_negative_ttl() {
461        let server = MockServer::start().await;
462        Mock::given(method("POST"))
463            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
464                "active": false
465            })))
466            .expect(1)
467            .mount(&server)
468            .await;
469
470        let opts = IntrospectionCacheOptions {
471            max_entries: 100,
472            default_ttl: Duration::from_secs(60),
473            negative_ttl: Duration::from_secs(10),
474        };
475        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
476            server.uri(),
477            "cid".into(),
478            "cs".into(),
479            opts,
480        )
481        .expect("introspector"); // allow-unwrap(test)
482
483        let r = introspector.introspect("dead-token").await.unwrap();
484        assert!(!r.active);
485        let r2 = introspector.introspect("dead-token").await.unwrap();
486        assert!(!r2.active);
487    }
488
489    #[tokio::test]
490    async fn cache_key_does_not_contain_raw_token() {
491        let server = MockServer::start().await;
492        Mock::given(method("POST"))
493            .respond_with(
494                ResponseTemplate::new(200).set_body_json(serde_json::json!({"active": true})),
495            )
496            .mount(&server)
497            .await;
498
499        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
500            server.uri(),
501            "cid".into(),
502            "cs".into(),
503            test_cache_opts(),
504        )
505        .expect("introspector"); // allow-unwrap(test)
506
507        introspector.introspect("secret-token-value").await.unwrap();
508        let cache = introspector.cache.read().await;
509        for key in cache.keys() {
510            assert!(
511                !key.contains("secret-token-value"),
512                "cache key must not contain raw token"
513            );
514        }
515    }
516
517    #[tokio::test]
518    async fn eviction_removes_oldest_when_over_capacity() {
519        let server = MockServer::start().await;
520        for i in 0..5 {
521            Mock::given(method("POST"))
522                .and(body_string_contains(format!("token-{i}"))) // allow-secret
523                .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
524                    "active": true, "sub": format!("user-{i}")
525                })))
526                .mount(&server)
527                .await;
528        }
529
530        let opts = IntrospectionCacheOptions {
531            max_entries: 2,
532            default_ttl: Duration::from_secs(600),
533            negative_ttl: Duration::from_secs(5),
534        };
535        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
536            server.uri(),
537            "cid".into(),
538            "cs".into(),
539            opts,
540        )
541        .expect("introspector"); // allow-unwrap(test)
542
543        introspector.introspect("token-0").await.unwrap();
544        introspector.introspect("token-1").await.unwrap();
545        introspector.introspect("token-2").await.unwrap();
546        introspector.introspect("token-3").await.unwrap();
547        introspector.introspect("token-4").await.unwrap();
548
549        let cache = introspector.cache.read().await;
550        assert!(cache.len() <= 2, "cache must respect max_entries");
551    }
552
553    #[test]
554    fn debug_redacts_client_secret() {
555        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
556            "https://example.com".into(),
557            "cid".into(),
558            "super-secret-value".into(),
559            test_cache_opts(),
560        )
561        .expect("introspector"); // allow-unwrap(test)
562        let debug = format!("{introspector:?}");
563        assert!(
564            !debug.contains("super-secret-value"),
565            "Debug must not leak client_secret"
566        );
567        assert!(debug.contains("REDACTED"));
568    }
569
570    #[tokio::test]
571    async fn production_constructor_rejects_http_endpoint() {
572        let result = CachingTokenIntrospector::new(
573            "http://insecure.example.com/introspect".into(),
574            "cid".into(),
575            "cs".into(),
576            test_cache_opts(),
577            SsrfPolicy::PublicHttpsOnly,
578        )
579        .await;
580        assert!(result.is_err());
581        let err = result.unwrap_err();
582        assert!(matches!(err, AuthError::ConfigError(ref s) if s.contains("HTTPS")));
583    }
584
585    #[tokio::test]
586    async fn production_constructor_rejects_localhost() {
587        let result = CachingTokenIntrospector::new(
588            "https://localhost:8080/introspect".into(),
589            "cid".into(),
590            "cs".into(),
591            test_cache_opts(),
592            SsrfPolicy::PublicHttpsOnly,
593        )
594        .await;
595        assert!(result.is_err());
596        let err = result.unwrap_err();
597        assert!(
598            matches!(err, AuthError::ConfigError(ref s) if s.contains("private") || s.contains("loopback"))
599        );
600    }
601
602    #[tokio::test]
603    async fn http_error_500_returns_provider_unavailable() {
604        let server = MockServer::start().await;
605        Mock::given(method("POST"))
606            .respond_with(ResponseTemplate::new(500))
607            .mount(&server)
608            .await;
609
610        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
611            server.uri(),
612            "cid".into(),
613            "cs".into(),
614            test_cache_opts(),
615        )
616        .expect("introspector"); // allow-unwrap(test)
617        let result = introspector.introspect("tok").await;
618        assert!(result.is_err());
619        let err = result.unwrap_err();
620        assert!(matches!(err, AuthError::ProviderUnavailable(_)));
621    }
622
623    #[tokio::test]
624    async fn http_401_returns_provider_unavailable() {
625        let server = MockServer::start().await;
626        Mock::given(method("POST"))
627            .respond_with(ResponseTemplate::new(401))
628            .mount(&server)
629            .await;
630
631        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
632            server.uri(),
633            "cid".into(),
634            "cs".into(),
635            test_cache_opts(),
636        )
637        .expect("introspector"); // allow-unwrap(test)
638        let result = introspector.introspect("tok").await;
639        assert!(result.is_err());
640        let err = result.unwrap_err();
641        assert!(matches!(err, AuthError::ProviderUnavailable(ref s) if s.contains("unauthorized")));
642    }
643
644    #[tokio::test]
645    async fn concurrent_different_tokens_no_head_of_line_blocking() {
646        let server = MockServer::start().await;
647        Mock::given(method("POST"))
648            .respond_with(
649                ResponseTemplate::new(200)
650                    .set_delay(Duration::from_millis(500))
651                    .set_body_json(serde_json::json!({"active": true})),
652            )
653            .expect(2)
654            .mount(&server)
655            .await;
656
657        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
658            server.uri(),
659            "cid".into(),
660            "cs".into(),
661            test_cache_opts(),
662        )
663        .expect("introspector"); // allow-unwrap(test)
664
665        let start = Instant::now();
666        let (r1, r2) = tokio::join!(
667            introspector.introspect("token-parallel-a"),
668            introspector.introspect("token-parallel-b"),
669        );
670        let elapsed = start.elapsed();
671
672        assert!(r1.is_ok(), "first introspect failed: {:?}", r1.err());
673        assert!(r2.is_ok(), "second introspect failed: {:?}", r2.err());
674        assert!(r1.unwrap().active);
675        assert!(r2.unwrap().active);
676        assert!(
677            elapsed < Duration::from_millis(800),
678            "expected parallel introspection (<800ms), got {elapsed:?} (serial would be ~1000ms)"
679        );
680    }
681
682    #[tokio::test]
683    async fn concurrent_same_token_dedup_preserved() {
684        let server = MockServer::start().await;
685        Mock::given(method("POST"))
686            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
687                "active": true
688            })))
689            .expect(1)
690            .mount(&server)
691            .await;
692
693        let introspector = CachingTokenIntrospector::new_unchecked_for_test(
694            server.uri(),
695            "cid".into(),
696            "cs".into(),
697            test_cache_opts(),
698        )
699        .expect("introspector"); // allow-unwrap(test)
700
701        let (r1, r2) = tokio::join!(
702            introspector.introspect("same-dedup-token"),
703            introspector.introspect("same-dedup-token"),
704        );
705
706        assert!(r1.is_ok(), "first caller failed: {:?}", r1.err());
707        assert!(r2.is_ok(), "second caller failed: {:?}", r2.err());
708    }
709}
camel_auth/introspection.rs

camel_auth/
introspection.rs
