Skip to main content

rocketmq_auth/authentication/manager/
authentication_metadata_manager_impl.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Authentication metadata manager implementation.
16//!
17//! This module provides the default implementation of `AuthenticationMetadataManager`,
18//! coordinating authentication and authorization metadata providers.
19
20use 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
38/// Default implementation of `AuthenticationMetadataManager`.
39///
40/// This implementation coordinates between:
41/// - `AuthenticationMetadataProvider` - For user CRUD operations
42/// - `AuthorizationMetadataProvider` - For ACL operations (optional)
43///
44/// # Concurrency
45///
46/// Thread-safe through `Arc<RwLock<>>` for mutable provider references.
47/// Read operations acquire read locks, write operations acquire write locks.
48pub struct AuthenticationMetadataManagerImpl<A, B>
49where
50    A: AuthenticationMetadataProvider + 'static,
51    B: AuthorizationMetadataProvider + 'static,
52{
53    /// Authentication metadata provider for user operations.
54    authentication_metadata_provider: Arc<RwLock<Option<A>>>,
55
56    /// Authorization metadata provider for ACL operations (optional).
57    authorization_metadata_provider: Arc<RwLock<Option<B>>>,
58}
59
60impl
61    AuthenticationMetadataManagerImpl<
62        crate::LocalAuthenticationMetadataProvider,
63        crate::LocalAuthorizationMetadataProvider,
64    >
65{
66    /// Borrows the runtime's user and ACL ports for metadata management.
67    /// Shutting down the manager releases its handles; the runtime owns providers.
68    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    /// Create a new authentication metadata manager.
82    ///
83    /// # Arguments
84    ///
85    /// * `auth_provider` - Authentication metadata provider
86    /// * `authz_provider` - Authorization metadata provider (optional)
87    ///
88    /// # Example
89    ///
90    /// ```rust,ignore
91    /// let auth_provider = LocalAuthenticationMetadataProvider::new();
92    /// let authz_provider = Some(LocalAuthorizationMetadataProvider::new());
93    /// let manager = AuthenticationMetadataManagerImpl::new(auth_provider, authz_provider);
94    /// ```
95    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    /// Set the authentication metadata provider.
103    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    /// Set the authorization metadata provider.
109    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    /// Validate user for creation.
115    ///
116    /// # Validation Rules
117    ///
118    /// - Username must not be blank
119    /// - Password must not be blank (for create operation)
120    /// - User type must be set (for create operation)
121    fn validate_user(&self, user: &User, is_create: bool) -> ManagerResult<()> {
122        // Validate username
123        if user.username().is_empty() {
124            return Err(AuthServiceError::new(
125                AuthOperation::ManageMetadata,
126                AuthFailureKind::InvalidInput,
127            ));
128        }
129
130        // Validate password (required for creation)
131        if is_create && user.password().is_none() {
132            return Err(AuthServiceError::new(
133                AuthOperation::ManageMetadata,
134                AuthFailureKind::InvalidInput,
135            ));
136        }
137
138        // Validate user type (required for creation, optional for update)
139        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    /// Handle exceptions from provider operations.
150    ///
151    /// Preserves provider failures at the manager boundary.
152    fn handle_exception(&self, error: AuthServiceError) -> AuthServiceError {
153        // Log the error
154        warn!("Authentication metadata manager operation failed: {}", error);
155
156        // For now, just pass through the error
157        // In production, you might want to map specific error types
158        error
159    }
160
161    /// Initialize user from string (supports both JSON and "username:password" format)
162    async fn init_user_from_string(&self, init_user_str: &str, provider: &A) -> ManagerResult<()> {
163        // Try to parse as JSON first
164        if init_user_str.trim().starts_with('{') {
165            // JSON format - try to parse as User
166            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        // Fall back to "username:password" format
175        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(()); // Don't fail initialization
179        }
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    /// Initialize inner client user from SessionCredentials format
194    async fn init_inner_client_user(&self, credentials_str: &str, provider: &A) -> ManagerResult<()> {
195        // Try to parse as JSON with accessKey and secretKey fields
196        #[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    /// Create user if it doesn't exist
219    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        // Shutdown authentication provider
245        let mut auth_guard = self.authentication_metadata_provider.write().await;
246        if let Some(mut provider) = auth_guard.take() {
247            let _ = provider.shutdown().await; // Ignore shutdown errors in cleanup
248        }
249
250        // Shutdown authorization provider
251        let mut authz_guard = self.authorization_metadata_provider.write().await;
252        if let Some(mut provider) = authz_guard.take() {
253            provider.shutdown(); // AuthorizationMetadataProvider::shutdown is synchronous
254        }
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        // 1. Handle init_authentication_user (JSON format or simple "username:password")
274        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        // 2. Handle inner_client_authentication_credentials (SessionCredentials format)
280        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        // Validate user input
290        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        // Check if user already exists
303        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        // Set default values if not provided
315        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        // Validate user input (password not required for update)
334        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        // Get existing user first
347        let existing_user = provider.get_user(user.username().as_str()).await?;
348
349        // Merge fields: update only provided fields
350        let mut updated_user = existing_user.clone();
351
352        // Update password if provided
353        if let Some(new_password) = user.password() {
354            updated_user.set_password(new_password.clone());
355        }
356
357        // Update user type if provided
358        if let Some(new_type) = user.user_type() {
359            updated_user.set_user_type(new_type);
360        }
361
362        // Update user status if provided
363        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        // Delete user from authentication provider
395        let delete_user_result = auth_provider
396            .delete_user(username)
397            .await
398            .map_err(|e| self.handle_exception(e));
399
400        // Also delete related ACLs from authorization provider
401        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; // Ignore ACL deletion errors
405        }
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        // Manager should be created successfully
484        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        // Valid user
495        let user = User::of_with_type("test_user", "password", UserType::Normal);
496        assert!(manager.validate_user(&user, true).is_ok());
497
498        // Invalid: empty username
499        let user = User::of_with_type("", "password", UserType::Normal);
500        assert!(manager.validate_user(&user, true).is_err());
501
502        // Invalid: missing password for creation
503        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}