rocketmq_auth/authentication/factory/
authentication_factory.rs1use std::any::Any;
36use std::collections::HashMap;
37use std::sync::Arc;
38use std::sync::Mutex;
39use std::sync::OnceLock;
40
41use rocketmq_protocol::protocol::remoting_command::RemotingCommand;
42
43use crate::authentication::context::default_authentication_context::DefaultAuthenticationContext;
44use crate::authentication::evaluator::AuthenticationEvaluator;
45use crate::authentication::manager::AuthenticationMetadataManagerImpl;
46use crate::authentication::provider::AuthenticationMetadataProvider;
47use crate::authentication::provider::AuthenticationProvider;
48use crate::authentication::provider::DefaultAuthenticationProvider;
49use crate::authentication::provider::LocalAuthenticationMetadataProvider;
50use crate::authentication::strategy::AuthenticationStrategy;
51use crate::authentication::strategy::StatefulAuthenticationStrategy;
52use crate::authentication::strategy::StatelessAuthenticationStrategy;
53use crate::authorization::metadata_provider::AuthorizationMetadataProvider;
54use crate::config::AuthConfig;
55use crate::AuthFailureKind;
56use crate::AuthOperation;
57use crate::AuthServiceError;
58use crate::AuthServiceResult;
59
60static INSTANCE_CACHE: OnceLock<Mutex<HashMap<String, Arc<dyn Any + Send + Sync>>>> = OnceLock::new();
65
66const PROVIDER_PREFIX: &str = "PROVIDER_";
68
69const METADATA_PROVIDER_PREFIX: &str = "METADATA_PROVIDER_";
71
72const EVALUATOR_PREFIX: &str = "EVALUATOR_";
74
75pub struct AuthenticationFactory;
86
87struct BoundAuthenticationStrategy {
88 strategy: Box<dyn AuthenticationStrategy>,
89 admission: Arc<crate::provider_owner::ProviderAdmission>,
90}
91
92impl AuthenticationStrategy for BoundAuthenticationStrategy {
93 fn authenticate<'a>(&'a self, context: &'a dyn crate::AuthenticationContext) -> crate::AuthenticationFuture<'a> {
94 Box::pin(async move {
95 let _operation = self.admission.enter()?;
96 self.strategy.authenticate(context).await
97 })
98 }
99}
100
101impl AuthenticationFactory {
102 pub async fn get_provider(config: &AuthConfig) -> AuthServiceResult<Arc<DefaultAuthenticationProvider>> {
123 if !is_blank_or_supported(
124 config.authentication_provider.as_str(),
125 &["DefaultAuthenticationProvider", "default"],
126 ) {
127 return Err(AuthServiceError::new(
128 AuthOperation::InitializeProvider,
129 AuthFailureKind::Unsupported,
130 ));
131 }
132
133 let key = format!("{}{}", PROVIDER_PREFIX, config.config_name);
134 if let Some(cached) = Self::cached(&key)? {
135 return cached
136 .downcast::<DefaultAuthenticationProvider>()
137 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal));
138 }
139 let provider = new_initialized_default_provider(config.clone(), None).await?;
140
141 Self::compute_if_absent(&key, || Ok(Arc::new(provider) as Arc<dyn Any + Send + Sync>)).and_then(|any_arc| {
142 any_arc
143 .downcast::<DefaultAuthenticationProvider>()
144 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal))
145 })
146 }
147
148 pub async fn get_metadata_provider(
160 config: &AuthConfig,
161 ) -> AuthServiceResult<Option<Arc<dyn AuthenticationMetadataProvider>>> {
162 Self::get_metadata_provider_with_service(config, None).await
163 }
164
165 pub async fn get_metadata_provider_with_service(
174 config: &AuthConfig,
175 metadata_service: Option<Arc<dyn Any + Send + Sync>>,
176 ) -> AuthServiceResult<Option<Arc<dyn AuthenticationMetadataProvider>>> {
177 if let Some(service) = metadata_service {
178 let registry = service.downcast_ref::<crate::ProviderRegistry>().ok_or_else(|| {
179 AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Unsupported)
180 })?;
181 return Ok(Some(Arc::new(
182 crate::provider_owner::legacy::RegistryUserMetadata::new(registry),
183 )));
184 }
185 let configured = config.authentication_metadata_provider.as_str();
186 if configured.trim().is_empty() {
187 return Ok(None);
188 }
189 if !is_supported(configured, &["LocalAuthenticationMetadataProvider", "local"]) {
190 return Err(AuthServiceError::new(
191 AuthOperation::InitializeProvider,
192 AuthFailureKind::Unsupported,
193 ));
194 }
195
196 let key = format!("{}{}", METADATA_PROVIDER_PREFIX, config.config_name);
197 if let Some(cached) = Self::cached(&key)? {
198 return cached
199 .downcast::<LocalAuthenticationMetadataProvider>()
200 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal))
201 .map(|provider| Some(provider as Arc<dyn AuthenticationMetadataProvider>));
202 }
203 let provider = new_initialized_local_metadata_provider(config.clone(), None).await?;
204 Self::compute_if_absent(&key, || Ok(Arc::new(provider) as Arc<dyn Any + Send + Sync>))
205 .and_then(|any_arc| {
206 any_arc
207 .downcast::<LocalAuthenticationMetadataProvider>()
208 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal))
209 })
210 .map(|provider| Some(provider as Arc<dyn AuthenticationMetadataProvider>))
211 }
212
213 pub async fn get_strategy(
219 config: &AuthConfig,
220 metadata_service: Option<Arc<dyn Any + Send + Sync>>,
221 ) -> AuthServiceResult<Box<dyn AuthenticationStrategy>> {
222 let registry = metadata_service
223 .as_ref()
224 .map(|service| {
225 service
226 .downcast_ref::<crate::ProviderRegistry>()
227 .cloned()
228 .ok_or_else(|| {
229 AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Unsupported)
230 })
231 })
232 .transpose()?;
233 let provider = if metadata_service.is_some() {
234 Arc::new(new_initialized_default_provider(config.clone(), metadata_service).await?)
235 } else {
236 Self::get_provider(config).await?
237 };
238 let strategy = config.authentication_strategy.as_str();
239 let strategy: Box<dyn AuthenticationStrategy> = if strategy.trim().is_empty()
240 || strategy.ends_with("StatelessAuthenticationStrategy")
241 || strategy.eq_ignore_ascii_case("stateless")
242 {
243 Box::new(StatelessAuthenticationStrategy::new(config.clone(), Some(provider)))
244 } else if strategy.ends_with("StatefulAuthenticationStrategy") || strategy.eq_ignore_ascii_case("stateful") {
245 Box::new(StatefulAuthenticationStrategy::new_with_acl_generation(
246 config.clone(),
247 Some(provider),
248 registry
249 .as_ref()
250 .map(crate::ProviderRegistry::acl_generation_counter)
251 .unwrap_or_default(),
252 ))
253 } else {
254 return Err(AuthServiceError::new(
255 AuthOperation::InitializeProvider,
256 AuthFailureKind::Unsupported,
257 ));
258 };
259 Ok(match registry {
260 Some(registry) => Box::new(BoundAuthenticationStrategy {
261 strategy,
262 admission: registry.admission(),
263 }),
264 None => strategy,
265 })
266 }
267
268 pub async fn get_evaluator(
270 config: &AuthConfig,
271 ) -> AuthServiceResult<Arc<AuthenticationEvaluator<Box<dyn AuthenticationStrategy>>>> {
272 Self::get_evaluator_with_service(config, None).await
273 }
274
275 pub async fn get_evaluator_with_service(
277 config: &AuthConfig,
278 metadata_service: Option<Arc<dyn Any + Send + Sync>>,
279 ) -> AuthServiceResult<Arc<AuthenticationEvaluator<Box<dyn AuthenticationStrategy>>>> {
280 if metadata_service.is_some() {
281 return Ok(Arc::new(AuthenticationEvaluator::new(
282 Self::get_strategy(config, metadata_service).await?,
283 )));
284 }
285 let key = format!("{}{}", EVALUATOR_PREFIX, config.config_name);
286 if let Some(cached) = Self::cached(&key)? {
287 return cached
288 .downcast::<AuthenticationEvaluator<Box<dyn AuthenticationStrategy>>>()
289 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal));
290 }
291 let strategy = Self::get_strategy(config, metadata_service).await?;
292 Self::compute_if_absent(&key, || {
293 Ok(Arc::new(AuthenticationEvaluator::new(strategy)) as Arc<dyn Any + Send + Sync>)
294 })
295 .and_then(|any_arc| {
296 any_arc
297 .downcast::<AuthenticationEvaluator<Box<dyn AuthenticationStrategy>>>()
298 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal))
299 })
300 }
301
302 pub fn get_metadata_manager<A, B>(
322 _config: &AuthConfig,
323 auth_provider: Option<A>,
324 authz_provider: Option<B>,
325 ) -> AuthenticationMetadataManagerImpl<A, B>
326 where
327 A: AuthenticationMetadataProvider + 'static,
328 B: AuthorizationMetadataProvider + 'static,
329 {
330 AuthenticationMetadataManagerImpl::new(auth_provider, authz_provider)
331 }
332
333 pub async fn new_context_from_metadata(
346 config: &AuthConfig,
347 metadata: &HashMap<String, String>,
348 request: Box<dyn Any + Send>,
349 ) -> AuthServiceResult<Option<DefaultAuthenticationContext>> {
350 let provider = Self::get_provider(config).await?;
351 Ok(Some(
352 <DefaultAuthenticationProvider as AuthenticationProvider>::new_context_from_metadata(
353 &provider, metadata, request,
354 ),
355 ))
356 }
357
358 pub async fn new_context_from_command(
370 config: &AuthConfig,
371 command: &RemotingCommand,
372 ) -> AuthServiceResult<Option<DefaultAuthenticationContext>> {
373 let provider = Self::get_provider(config).await?;
374 Ok(Some(
375 <DefaultAuthenticationProvider as AuthenticationProvider>::new_context_from_command(&provider, command),
376 ))
377 }
378
379 fn compute_if_absent<F>(key: &str, factory_fn: F) -> AuthServiceResult<Arc<dyn Any + Send + Sync>>
402 where
403 F: FnOnce() -> AuthServiceResult<Arc<dyn Any + Send + Sync>>,
404 {
405 let cache = INSTANCE_CACHE.get_or_init(|| Mutex::new(HashMap::new()));
406
407 {
408 let cache_guard = cache
409 .lock()
410 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal))?;
411 if let Some(cached) = cache_guard.get(key) {
412 return Ok(Arc::clone(cached));
413 }
414 }
415
416 let instance = factory_fn()?;
417
418 let mut cache_guard = cache
419 .lock()
420 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal))?;
421 if let Some(cached) = cache_guard.get(key) {
422 return Ok(Arc::clone(cached));
423 }
424
425 cache_guard.insert(key.to_string(), Arc::clone(&instance));
426 Ok(instance)
427 }
428
429 fn cached(key: &str) -> AuthServiceResult<Option<Arc<dyn Any + Send + Sync>>> {
430 let cache = INSTANCE_CACHE.get_or_init(|| Mutex::new(HashMap::new()));
431 let guard = cache
432 .lock()
433 .map_err(|_| AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Internal))?;
434 Ok(guard.get(key).cloned())
435 }
436}
437
438impl AuthenticationStrategy for Box<dyn AuthenticationStrategy> {
439 fn authenticate<'a>(
440 &'a self,
441 context: &'a dyn crate::authorization::context::authentication_context::AuthenticationContext,
442 ) -> crate::authentication::strategy::AuthenticationFuture<'a> {
443 self.as_ref().authenticate(context)
444 }
445}
446
447async fn new_initialized_default_provider(
448 config: AuthConfig,
449 metadata_service: Option<Arc<dyn Any + Send + Sync>>,
450) -> AuthServiceResult<DefaultAuthenticationProvider> {
451 let mut provider = DefaultAuthenticationProvider::new();
452 provider.initialize(config, metadata_service).await.map_err(|source| {
453 let kind = source.kind();
454 AuthServiceError::with_source(AuthOperation::InitializeProvider, kind, source)
455 })?;
456 Ok(provider)
457}
458
459async fn new_initialized_local_metadata_provider(
460 config: AuthConfig,
461 metadata_service: Option<Arc<dyn Any + Send + Sync>>,
462) -> AuthServiceResult<LocalAuthenticationMetadataProvider> {
463 let mut provider = LocalAuthenticationMetadataProvider::new();
464 provider.initialize(config, metadata_service).await.map_err(|source| {
465 let kind = source.kind();
466 AuthServiceError::with_source(AuthOperation::InitializeProvider, kind, source)
467 })?;
468 Ok(provider)
469}
470
471fn is_blank_or_supported(configured: &str, supported: &[&str]) -> bool {
472 configured.trim().is_empty() || is_supported(configured, supported)
473}
474
475fn is_supported(configured: &str, supported: &[&str]) -> bool {
476 supported
477 .iter()
478 .any(|value| configured.eq_ignore_ascii_case(value) || configured.ends_with(value))
479}
480
481#[cfg(test)]
482mod tests {
483 use cheetah_string::CheetahString;
484
485 use super::*;
486
487 fn create_test_config(config_name: &str) -> AuthConfig {
488 AuthConfig {
489 config_name: CheetahString::from_string(config_name.to_string()),
490 ..AuthConfig::default()
491 }
492 }
493
494 #[tokio::test]
495 async fn test_get_provider_default() {
496 let config = create_test_config("test_provider_default");
497 let provider = AuthenticationFactory::get_provider(&config).await;
498 assert!(provider.is_ok());
499 }
500
501 #[tokio::test]
502 async fn test_get_provider_unsupported() {
503 let mut config = create_test_config("test_provider_unsupported");
504 config.authentication_provider = CheetahString::from_static_str("UnsupportedAuthenticationProvider");
505
506 let error = match AuthenticationFactory::get_provider(&config).await {
507 Ok(_) => panic!("unsupported authentication provider should fail"),
508 Err(error) => error,
509 };
510
511 assert_eq!(error.operation(), AuthOperation::InitializeProvider);
512 assert_eq!(error.kind(), AuthFailureKind::Unsupported);
513 }
514
515 #[tokio::test]
516 async fn test_get_metadata_provider_not_configured() {
517 let config = create_test_config("test_metadata_provider_none");
518 let provider = AuthenticationFactory::get_metadata_provider(&config).await;
519 assert!(provider.is_ok());
520 assert!(provider.unwrap().is_none());
521 }
522
523 #[tokio::test]
524 async fn test_get_metadata_provider_local() {
525 let mut config = create_test_config("test_metadata_provider_local");
526 config.authentication_metadata_provider = CheetahString::from_static_str("LocalAuthenticationMetadataProvider");
527 let provider1 = AuthenticationFactory::get_metadata_provider(&config)
528 .await
529 .unwrap()
530 .unwrap();
531 let provider2 = AuthenticationFactory::get_metadata_provider(&config)
532 .await
533 .unwrap()
534 .unwrap();
535
536 assert!(Arc::ptr_eq(&provider1, &provider2));
537 }
538
539 #[tokio::test]
540 async fn test_get_metadata_provider_unsupported() {
541 let mut config = create_test_config("test_metadata_provider_unsupported");
542 config.authentication_metadata_provider = CheetahString::from_static_str("UnsupportedMetadataProvider");
543
544 let error = match AuthenticationFactory::get_metadata_provider(&config).await {
545 Ok(_) => panic!("unsupported metadata provider should fail"),
546 Err(error) => error,
547 };
548
549 assert_eq!(error.operation(), AuthOperation::InitializeProvider);
550 assert_eq!(error.kind(), AuthFailureKind::Unsupported);
551 }
552
553 #[tokio::test]
554 async fn test_provider_caching() {
555 let config = create_test_config("test_cache");
556 let provider1 = AuthenticationFactory::get_provider(&config).await.unwrap();
557 let provider2 = AuthenticationFactory::get_provider(&config).await.unwrap();
558 assert!(Arc::ptr_eq(&provider1, &provider2));
559 }
560
561 #[tokio::test]
562 async fn test_get_strategy_defaults_to_stateless() {
563 let config = create_test_config("test_strategy_default");
564 let strategy = AuthenticationFactory::get_strategy(&config, None).await.unwrap();
565 let context = DefaultAuthenticationContext::new();
566
567 assert!(strategy.authenticate(&context).await.is_ok());
568 }
569
570 #[tokio::test]
571 async fn test_get_strategy_supports_stateful_alias() {
572 let mut config = create_test_config("test_strategy_stateful");
573 config.authentication_strategy = CheetahString::from_static_str("stateful");
574 let strategy = AuthenticationFactory::get_strategy(&config, None).await.unwrap();
575 let context = DefaultAuthenticationContext::new();
576
577 assert!(strategy.authenticate(&context).await.is_ok());
578 }
579
580 #[tokio::test]
581 async fn test_get_strategy_unsupported() {
582 let mut config = create_test_config("test_strategy_unsupported");
583 config.authentication_strategy = CheetahString::from_static_str("UnsupportedAuthenticationStrategy");
584
585 let error = match AuthenticationFactory::get_strategy(&config, None).await {
586 Ok(_) => panic!("unsupported authentication strategy should fail"),
587 Err(error) => error,
588 };
589
590 assert_eq!(error.operation(), AuthOperation::InitializeProvider);
591 assert_eq!(error.kind(), AuthFailureKind::Unsupported);
592 }
593
594 #[tokio::test]
595 async fn test_get_evaluator_caching() {
596 let config = create_test_config("test_evaluator_cache");
597 let evaluator1 = AuthenticationFactory::get_evaluator(&config).await.unwrap();
598 let evaluator2 = AuthenticationFactory::get_evaluator(&config).await.unwrap();
599
600 assert!(Arc::ptr_eq(&evaluator1, &evaluator2));
601 }
602}