rocketmq_auth/authentication/manager/
authentication_metadata_manager_impl.rs1use std::sync::Arc;
21
22use tokio::sync::RwLock;
23use tracing::info;
24use tracing::warn;
25
26use super::authentication_metadata_manager::AuthenticationMetadataManager;
27use super::authentication_metadata_manager::ManagerResult;
28use crate::authentication::enums::user_status::UserStatus;
29use crate::authentication::enums::user_type::UserType;
30use crate::authentication::model::user::User;
31use crate::authentication::provider::AuthenticationMetadataProvider;
32use crate::authorization::metadata_provider::AuthorizationMetadataProvider;
33use crate::config::AuthConfig;
34use crate::AuthFailureKind;
35use crate::AuthOperation;
36use crate::AuthServiceError;
37
38pub struct AuthenticationMetadataManagerImpl<A, B>
49where
50 A: AuthenticationMetadataProvider + 'static,
51 B: AuthorizationMetadataProvider + 'static,
52{
53 authentication_metadata_provider: Arc<RwLock<Option<A>>>,
55
56 authorization_metadata_provider: Arc<RwLock<Option<B>>>,
58}
59
60impl
61 AuthenticationMetadataManagerImpl<
62 crate::LocalAuthenticationMetadataProvider,
63 crate::LocalAuthorizationMetadataProvider,
64 >
65{
66 pub fn with_registry(registry: &crate::ProviderRegistry) -> impl AuthenticationMetadataManager {
69 AuthenticationMetadataManagerImpl::new(
70 Some(crate::provider_owner::legacy::RegistryUserMetadata::new(registry)),
71 Some(crate::provider_owner::legacy::RegistryAclMetadata::new(registry)),
72 )
73 }
74}
75
76impl<A, B> AuthenticationMetadataManagerImpl<A, B>
77where
78 A: AuthenticationMetadataProvider,
79 B: AuthorizationMetadataProvider,
80{
81 pub fn new(auth_provider: Option<A>, authz_provider: Option<B>) -> Self {
96 Self {
97 authentication_metadata_provider: Arc::new(RwLock::new(auth_provider)),
98 authorization_metadata_provider: Arc::new(RwLock::new(authz_provider)),
99 }
100 }
101
102 pub async fn set_authentication_provider(&self, provider: A) {
104 let mut guard = self.authentication_metadata_provider.write().await;
105 *guard = Some(provider);
106 }
107
108 pub async fn set_authorization_provider(&self, provider: B) {
110 let mut guard = self.authorization_metadata_provider.write().await;
111 *guard = Some(provider);
112 }
113
114 fn validate_user(&self, user: &User, is_create: bool) -> ManagerResult<()> {
122 if user.username().is_empty() {
124 return Err(AuthServiceError::new(
125 AuthOperation::ManageMetadata,
126 AuthFailureKind::InvalidInput,
127 ));
128 }
129
130 if is_create && user.password().is_none() {
132 return Err(AuthServiceError::new(
133 AuthOperation::ManageMetadata,
134 AuthFailureKind::InvalidInput,
135 ));
136 }
137
138 if is_create && user.user_type().is_none() {
140 return Err(AuthServiceError::new(
141 AuthOperation::ManageMetadata,
142 AuthFailureKind::InvalidInput,
143 ));
144 }
145
146 Ok(())
147 }
148
149 fn handle_exception(&self, error: AuthServiceError) -> AuthServiceError {
153 warn!("Authentication metadata manager operation failed: {}", error);
155
156 error
159 }
160
161 async fn init_user_from_string(&self, init_user_str: &str, provider: &A) -> ManagerResult<()> {
163 if init_user_str.trim().starts_with('{') {
165 if let Ok(mut user) = serde_json::from_str::<User>(init_user_str) {
167 user.set_user_type(UserType::Super);
168 self.create_user_if_not_exists(user, provider).await?;
169 return Ok(());
170 }
171 warn!("Failed to parse init_authentication_user as JSON, trying simple format");
172 }
173
174 let parts: Vec<&str> = init_user_str.splitn(2, ':').collect();
176 if parts.len() != 2 {
177 warn!("Invalid init_authentication_user format; expected username:password or JSON");
178 return Ok(()); }
180
181 let username = parts[0].trim();
182 let password = parts[1].trim();
183
184 if !username.is_empty() && !password.is_empty() {
185 let mut user = User::of_with_type(username, password, UserType::Super);
186 user.set_user_status(UserStatus::Enable);
187 self.create_user_if_not_exists(user, provider).await?;
188 }
189
190 Ok(())
191 }
192
193 async fn init_inner_client_user(&self, credentials_str: &str, provider: &A) -> ManagerResult<()> {
195 #[derive(serde::Deserialize)]
197 struct SessionCredentials {
198 #[serde(rename = "accessKey")]
199 access_key: String,
200 #[serde(rename = "secretKey")]
201 secret_key: String,
202 }
203
204 match serde_json::from_str::<SessionCredentials>(credentials_str) {
205 Ok(credentials) => {
206 let mut user = User::of_with_type(credentials.access_key, credentials.secret_key, UserType::Super);
207 user.set_user_status(UserStatus::Enable);
208 self.create_user_if_not_exists(user, provider).await?;
209 }
210 Err(_) => {
211 warn!("Failed to parse inner_client_authentication_credentials");
212 }
213 }
214
215 Ok(())
216 }
217
218 async fn create_user_if_not_exists(&self, user: User, provider: &A) -> ManagerResult<()> {
220 match provider.get_user(user.username().as_str()).await {
221 Ok(_existing) => {
222 info!("User '{}' already exists, skipping creation", user.username());
223 Ok(())
224 }
225 Err(error) if error.kind() == AuthFailureKind::NotFound => {
226 provider.create_user(user.clone()).await?;
227 info!("Created super user '{}'", user.username());
228 Ok(())
229 }
230 Err(error) => Err(error),
231 }
232 }
233}
234
235#[allow(async_fn_in_trait)]
236impl<A, B> AuthenticationMetadataManager for AuthenticationMetadataManagerImpl<A, B>
237where
238 A: AuthenticationMetadataProvider,
239 B: AuthorizationMetadataProvider,
240{
241 async fn shutdown(&mut self) -> ManagerResult<()> {
242 info!("Shutting down authentication metadata manager");
243
244 let mut auth_guard = self.authentication_metadata_provider.write().await;
246 if let Some(mut provider) = auth_guard.take() {
247 let _ = provider.shutdown().await; }
249
250 let mut authz_guard = self.authorization_metadata_provider.write().await;
252 if let Some(mut provider) = authz_guard.take() {
253 provider.shutdown(); }
255
256 info!("Authentication metadata manager shutdown complete");
257 Ok(())
258 }
259
260 async fn init_user(&self, auth_config: &AuthConfig) -> ManagerResult<()> {
261 info!("Initializing default super user from configuration");
262
263 let guard = self.authentication_metadata_provider.read().await;
264 let provider = match &*guard {
265 Some(p) => p,
266 None => {
267 return Err(AuthServiceError::not_initialized(
268 "Authentication metadata provider is not configured",
269 ));
270 }
271 };
272
273 let init_user_str = auth_config.init_authentication_user.as_str();
275 if !init_user_str.is_empty() {
276 self.init_user_from_string(init_user_str, provider).await?;
277 }
278
279 let inner_credentials = auth_config.inner_client_authentication_credentials.as_str();
281 if !inner_credentials.is_empty() {
282 self.init_inner_client_user(inner_credentials, provider).await?;
283 }
284
285 Ok(())
286 }
287
288 async fn create_user(&self, user: User) -> ManagerResult<()> {
289 self.validate_user(&user, true)?;
291
292 let guard = self.authentication_metadata_provider.read().await;
293 let provider = match &*guard {
294 Some(p) => p,
295 None => {
296 return Err(AuthServiceError::not_initialized(
297 "Authentication metadata provider is not configured",
298 ));
299 }
300 };
301
302 match provider.get_user(user.username().as_str()).await {
304 Ok(_) => {
305 return Err(AuthServiceError::new(
306 AuthOperation::ManageMetadata,
307 AuthFailureKind::Conflict,
308 ))
309 }
310 Err(error) if error.kind() == AuthFailureKind::NotFound => {}
311 Err(error) => return Err(error),
312 }
313
314 let mut user_to_create = user.clone();
316 if user_to_create.user_type().is_none() {
317 user_to_create.set_user_type(UserType::Normal);
318 }
319 if user_to_create.user_status().is_none() {
320 user_to_create.set_user_status(UserStatus::Enable);
321 }
322
323 provider
324 .create_user(user_to_create)
325 .await
326 .map_err(|e| self.handle_exception(e))?;
327
328 info!("User '{}' created successfully", user.username());
329 Ok(())
330 }
331
332 async fn update_user(&self, user: User) -> ManagerResult<()> {
333 self.validate_user(&user, false)?;
335
336 let guard = self.authentication_metadata_provider.read().await;
337 let provider = match &*guard {
338 Some(p) => p,
339 None => {
340 return Err(AuthServiceError::not_initialized(
341 "Authentication metadata provider is not configured",
342 ));
343 }
344 };
345
346 let existing_user = provider.get_user(user.username().as_str()).await?;
348
349 let mut updated_user = existing_user.clone();
351
352 if let Some(new_password) = user.password() {
354 updated_user.set_password(new_password.clone());
355 }
356
357 if let Some(new_type) = user.user_type() {
359 updated_user.set_user_type(new_type);
360 }
361
362 if let Some(new_status) = user.user_status() {
364 updated_user.set_user_status(new_status);
365 }
366
367 provider
368 .update_user(updated_user)
369 .await
370 .map_err(|e| self.handle_exception(e))?;
371
372 info!("User '{}' updated successfully", user.username());
373 Ok(())
374 }
375
376 async fn delete_user(&self, username: &str) -> ManagerResult<()> {
377 if username.is_empty() {
378 return Err(AuthServiceError::new(
379 AuthOperation::ManageMetadata,
380 AuthFailureKind::InvalidInput,
381 ));
382 }
383
384 let auth_guard = self.authentication_metadata_provider.read().await;
385 let auth_provider = match &*auth_guard {
386 Some(p) => p,
387 None => {
388 return Err(AuthServiceError::not_initialized(
389 "Authentication metadata provider is not configured",
390 ));
391 }
392 };
393
394 let delete_user_result = auth_provider
396 .delete_user(username)
397 .await
398 .map_err(|e| self.handle_exception(e));
399
400 let authz_guard = self.authorization_metadata_provider.read().await;
402 if let Some(authz_provider) = &*authz_guard {
403 let user = User::of(username);
404 let _ = authz_provider.delete_acl(&user).await; }
406
407 delete_user_result?;
408 info!("User '{}' and related ACLs deleted successfully", username);
409 Ok(())
410 }
411
412 async fn get_user(&self, username: &str) -> ManagerResult<User> {
413 if username.is_empty() {
414 return Err(AuthServiceError::new(
415 AuthOperation::ReadMetadata,
416 AuthFailureKind::InvalidInput,
417 ));
418 }
419
420 let guard = self.authentication_metadata_provider.read().await;
421 let provider = match &*guard {
422 Some(p) => p,
423 None => {
424 return Err(AuthServiceError::not_initialized(
425 "Authentication metadata provider is not configured",
426 ));
427 }
428 };
429
430 provider.get_user(username).await.map_err(|e| self.handle_exception(e))
431 }
432
433 async fn list_users(&self, filter: Option<&str>) -> ManagerResult<Vec<User>> {
434 let guard = self.authentication_metadata_provider.read().await;
435 let provider = match &*guard {
436 Some(p) => p,
437 None => {
438 return Err(AuthServiceError::not_initialized(
439 "Authentication metadata provider is not configured",
440 ));
441 }
442 };
443
444 provider.list_user(filter).await.map_err(|e| self.handle_exception(e))
445 }
446
447 async fn is_super_user(&self, username: &str) -> ManagerResult<bool> {
448 if username.is_empty() {
449 return Ok(false);
450 }
451
452 match self.get_user(username).await {
453 Ok(user) => Ok(matches!(user.user_type(), Some(UserType::Super))),
454 Err(error) if error.kind() == AuthFailureKind::NotFound => Ok(false),
455 Err(error) => Err(error),
456 }
457 }
458}
459
460impl<A, B> Default for AuthenticationMetadataManagerImpl<A, B>
461where
462 A: AuthenticationMetadataProvider,
463 B: AuthorizationMetadataProvider,
464{
465 fn default() -> Self {
466 Self::new(None, None)
467 }
468}
469
470#[cfg(test)]
471mod tests {
472 use super::*;
473 use crate::authentication::provider::local_authentication_metadata_provider::LocalAuthenticationMetadataProvider;
474 use crate::authorization::metadata_provider::local::LocalAuthorizationMetadataProvider;
475
476 #[tokio::test]
477 async fn test_manager_creation() {
478 let manager = AuthenticationMetadataManagerImpl::<
479 LocalAuthenticationMetadataProvider,
480 LocalAuthorizationMetadataProvider,
481 >::default();
482
483 assert!(manager.authentication_metadata_provider.read().await.is_none());
485 }
486
487 #[tokio::test]
488 async fn test_validate_user_creation() {
489 let manager = AuthenticationMetadataManagerImpl::<
490 LocalAuthenticationMetadataProvider,
491 LocalAuthorizationMetadataProvider,
492 >::default();
493
494 let user = User::of_with_type("test_user", "password", UserType::Normal);
496 assert!(manager.validate_user(&user, true).is_ok());
497
498 let user = User::of_with_type("", "password", UserType::Normal);
500 assert!(manager.validate_user(&user, true).is_err());
501
502 let mut user = User::of("test_user");
504 user.set_user_type(UserType::Normal);
505 assert!(manager.validate_user(&user, true).is_err());
506 }
507
508 #[tokio::test]
509 async fn test_is_super_user_with_empty_username() {
510 let manager = AuthenticationMetadataManagerImpl::<
511 LocalAuthenticationMetadataProvider,
512 LocalAuthorizationMetadataProvider,
513 >::default();
514
515 let result = manager.is_super_user("").await.unwrap();
516 assert!(!result);
517 }
518}