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

camel_auth/
introspection.rs
