rocketmq_auth/authentication/provider/
local_authentication_metadata_provider.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//! Local Authentication Metadata Provider Implementation (Rust 2021 Standard)
16
17use std::any::Any;
18use std::collections::HashMap;
19use std::future::Future;
20use std::path::Path;
21use std::path::PathBuf;
22use std::pin::Pin;
23use std::sync::Arc;
24
25use rocketmq_runtime::MetadataDeadline;
26use rocketmq_runtime::MetadataGeneration;
27use rocketmq_runtime::MetadataIoActor;
28use tokio::fs;
29
30use crate::authentication::model::user::User;
31use crate::config::AuthConfig;
32use crate::metadata_snapshot::classify_snapshot_persistence;
33use crate::metadata_snapshot::ReconciliationGeneration;
34use crate::metadata_snapshot::SnapshotPersistence;
35use crate::AuthFailureKind;
36use crate::AuthOperation;
37use crate::AuthServiceError;
38use crate::AuthServiceResult;
39
40use super::authentication_metadata_provider::AuthenticationMetadataProvider;
41
42/// Local authentication metadata provider backed by an in-memory snapshot and an
43/// optional JSON snapshot file.
44pub struct LocalAuthenticationMetadataProvider {
45    storage: Arc<tokio::sync::RwLock<HashMap<String, User>>>,
46    storage_path: Option<PathBuf>,
47    write_lock: Arc<tokio::sync::Mutex<()>>,
48    metadata_io: Option<MetadataIoActor>,
49    /// The one user snapshot generation whose durability was not confirmed.
50    reconciliation: ReconciliationGeneration,
51}
52
53impl LocalAuthenticationMetadataProvider {
54    pub(crate) async fn flush_shared(&self) -> AuthServiceResult<()> {
55        let _writer = self.write_lock.lock().await;
56        let users = self.storage.read().await.clone();
57        // A flush re-encodes memory that is already published, so only the
58        // reported error matters here.
59        match self.persist_users(&users).await.into_error() {
60            Some(error) => Err(error),
61            None => Ok(()),
62        }
63    }
64
65    pub(crate) async fn close_shared(&self) -> AuthServiceResult<()> {
66        let _writer = self.write_lock.lock().await;
67        self.storage.write().await.clear();
68        Ok(())
69    }
70
71    /// Create a new local authentication metadata provider.
72    pub fn new() -> Self {
73        Self {
74            storage: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
75            storage_path: None,
76            write_lock: Arc::new(tokio::sync::Mutex::new(())),
77            reconciliation: ReconciliationGeneration::default(),
78            metadata_io: None,
79        }
80    }
81
82    pub fn with_config(config: &AuthConfig) -> AuthServiceResult<Self> {
83        Self::with_config_and_metadata_io(config, None)
84    }
85
86    pub fn with_config_and_metadata_io(
87        config: &AuthConfig,
88        metadata_io: Option<MetadataIoActor>,
89    ) -> AuthServiceResult<Self> {
90        let storage_path = auth_metadata_snapshot_path(config, "users.json");
91        let users = match &storage_path {
92            Some(path) => read_users_snapshot_blocking(path)?,
93            None => HashMap::new(),
94        };
95        Ok(Self {
96            storage: Arc::new(tokio::sync::RwLock::new(users)),
97            storage_path,
98            write_lock: Arc::new(tokio::sync::Mutex::new(())),
99            reconciliation: ReconciliationGeneration::default(),
100            metadata_io,
101        })
102    }
103
104    /// Writes the user snapshot and reports what the provider confirmed.
105    ///
106    /// The result is not an error type on purpose: a caller publishes the
107    /// accepted change even when the replacement is unconfirmed, and only a
108    /// definite pre-commit failure may hold it back.
109    async fn persist_users(&self, users: &HashMap<String, User>) -> SnapshotPersistence {
110        let Some(path) = &self.storage_path else {
111            return SnapshotPersistence::Durable {
112                generation: MetadataGeneration::new(0),
113            };
114        };
115        let content = match serde_json::to_vec_pretty(users) {
116            Ok(content) => content,
117            Err(source) => {
118                return SnapshotPersistence::NotWritten(AuthServiceError::with_source(
119                    AuthOperation::EncodeMetadata,
120                    AuthFailureKind::InvalidData,
121                    source,
122                ));
123            }
124        };
125        if let Some(metadata_io) = &self.metadata_io {
126            let persistence = match metadata_io
127                .submit_next_observed(
128                    "auth.authentication-users",
129                    path,
130                    content,
131                    MetadataDeadline::after(std::time::Duration::from_secs(5)),
132                )
133                .await
134            {
135                Ok(observation) => classify_snapshot_persistence(observation),
136                Err(error) => SnapshotPersistence::NotWritten(AuthServiceError::metadata_io(error)),
137            };
138            match &persistence {
139                SnapshotPersistence::Durable { generation } => self.reconciliation.release(*generation),
140                SnapshotPersistence::Unconfirmed { generation, .. } => self.reconciliation.record(*generation),
141                SnapshotPersistence::NotWritten(_) => {}
142            }
143            return persistence;
144        }
145        match write_users_snapshot(path, content).await {
146            Ok(()) => SnapshotPersistence::Durable {
147                generation: MetadataGeneration::new(0),
148            },
149            Err(error) => SnapshotPersistence::NotWritten(error),
150        }
151    }
152
153    /// Returns whether a metadata write is still unconfirmed.
154    pub fn reconciliation_required(&self) -> bool {
155        self.reconciliation.required()
156    }
157
158    /// Returns the generation whose durability was not confirmed.
159    pub fn reconciliation_generation(&self) -> Option<MetadataGeneration> {
160        self.reconciliation.generation()
161    }
162
163    async fn mutate_users<F>(&self, mutation: F) -> AuthServiceResult<()>
164    where
165        F: FnOnce(&mut HashMap<String, User>) -> AuthServiceResult<()>,
166    {
167        let _write_guard = self.write_lock.lock().await;
168        let mut snapshot = {
169            let storage = self.storage.read().await;
170            storage.clone()
171        };
172        mutation(&mut snapshot)?;
173        let persistence = self.persist_users(&snapshot).await;
174        if !persistence.may_have_written() {
175            return Err(persistence
176                .into_error()
177                .expect("not-written persistence carries its error"));
178        }
179        // Published even when unconfirmed: memory must not be older than a file
180        // that may already hold the change.
181        let mut storage = self.storage.write().await;
182        *storage = snapshot;
183        match persistence.into_error() {
184            Some(error) => Err(error),
185            None => Ok(()),
186        }
187    }
188}
189
190impl Default for LocalAuthenticationMetadataProvider {
191    fn default() -> Self {
192        Self::new()
193    }
194}
195
196impl AuthenticationMetadataProvider for LocalAuthenticationMetadataProvider {
197    /// Initialize the provider with storage.
198    fn initialize<'a>(
199        &'a mut self,
200        config: AuthConfig,
201        _metadata_service: Option<Arc<dyn Any + Send + Sync>>,
202    ) -> Pin<Box<dyn Future<Output = AuthServiceResult<()>> + Send + 'a>> {
203        Box::pin(async move {
204            let storage_path = auth_metadata_snapshot_path(&config, "users.json");
205            let users = if let Some(path) = storage_path.clone() {
206                let users = read_users_snapshot(&path).await?;
207                tracing::info!("LocalAuthenticationMetadataProvider initialized at {:?}", path);
208                users
209            } else {
210                tracing::info!("LocalAuthenticationMetadataProvider initialized in memory-only mode");
211                HashMap::new()
212            };
213            let mut storage = self.storage.write().await;
214            *storage = users;
215            self.storage_path = storage_path;
216            Ok(())
217        })
218    }
219
220    /// Shutdown the provider and release resources.
221    fn shutdown(&mut self) -> Pin<Box<dyn Future<Output = AuthServiceResult<()>> + Send + '_>> {
222        Box::pin(async move {
223            let mut storage = self.storage.write().await;
224            storage.clear();
225            tracing::info!("LocalAuthenticationMetadataProvider shutdown successfully");
226            Ok(())
227        })
228    }
229
230    /// Create a new user.
231    fn create_user<'a>(&'a self, user: User) -> Pin<Box<dyn Future<Output = AuthServiceResult<()>> + Send + 'a>> {
232        Box::pin(async move {
233            let username = user.username().to_string();
234            self.mutate_users(|storage| {
235                if storage.contains_key(&username) {
236                    return Err(AuthServiceError::new(
237                        AuthOperation::ManageMetadata,
238                        AuthFailureKind::Conflict,
239                    ));
240                }
241                storage.insert(username.clone(), user);
242                Ok(())
243            })
244            .await?;
245
246            tracing::info!("User '{}' created successfully", username);
247            Ok(())
248        })
249    }
250
251    /// Delete a user.
252    fn delete_user<'a>(
253        &'a self,
254        username: &'a str,
255    ) -> Pin<Box<dyn Future<Output = AuthServiceResult<()>> + Send + 'a>> {
256        Box::pin(async move {
257            self.mutate_users(|storage| {
258                storage.remove(username);
259                Ok(())
260            })
261            .await?;
262
263            tracing::info!("User '{}' deleted successfully", username);
264            Ok(())
265        })
266    }
267
268    /// Update a user.
269    fn update_user<'a>(&'a self, user: User) -> Pin<Box<dyn Future<Output = AuthServiceResult<()>> + Send + 'a>> {
270        Box::pin(async move {
271            let username = user.username().to_string();
272            self.mutate_users(|storage| {
273                storage.insert(username.clone(), user);
274                Ok(())
275            })
276            .await?;
277
278            tracing::info!("User '{}' updated successfully", username);
279            Ok(())
280        })
281    }
282
283    /// Get a user by username.
284    fn get_user<'a>(&'a self, username: &'a str) -> Pin<Box<dyn Future<Output = AuthServiceResult<User>> + Send + 'a>> {
285        Box::pin(async move {
286            let storage = self.storage.read().await;
287
288            storage
289                .get(username)
290                .cloned()
291                .ok_or_else(|| AuthServiceError::new(AuthOperation::ReadMetadata, AuthFailureKind::NotFound))
292        })
293    }
294
295    /// List all users, optionally filtered by username pattern.
296    fn list_user<'a>(
297        &'a self,
298        filter: Option<&'a str>,
299    ) -> Pin<Box<dyn Future<Output = AuthServiceResult<Vec<User>>> + Send + 'a>> {
300        Box::pin(async move {
301            let storage = self.storage.read().await;
302            let mut result = Vec::new();
303
304            for (username, user) in storage.iter() {
305                // Apply filter if provided
306                if let Some(filter_str) = filter {
307                    if !filter_str.is_empty() && !username.contains(filter_str) {
308                        continue;
309                    }
310                }
311
312                result.push(user.clone());
313            }
314
315            tracing::debug!("Listed {} users (filter: {:?})", result.len(), filter);
316            Ok(result)
317        })
318    }
319}
320
321fn auth_metadata_snapshot_path(config: &AuthConfig, file_name: &str) -> Option<PathBuf> {
322    let raw_path = config.auth_config_path.as_str().trim();
323    if raw_path.is_empty() {
324        return None;
325    }
326    let path = PathBuf::from(raw_path);
327    let root = if path.extension().is_some() {
328        path.with_extension("")
329    } else {
330        path
331    };
332    Some(root.join(file_name))
333}
334
335async fn read_users_snapshot(path: &Path) -> AuthServiceResult<HashMap<String, User>> {
336    let bytes = match fs::read(path).await {
337        Ok(bytes) => bytes,
338        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(HashMap::new()),
339        Err(error) => return Err(AuthServiceError::storage_read_failed(error)),
340    };
341    decode_users_snapshot(path, &bytes)
342}
343
344fn read_users_snapshot_blocking(path: &Path) -> AuthServiceResult<HashMap<String, User>> {
345    let bytes = match std::fs::read(path) {
346        Ok(bytes) => bytes,
347        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(HashMap::new()),
348        Err(error) => return Err(AuthServiceError::storage_read_failed(error)),
349    };
350    decode_users_snapshot(path, &bytes)
351}
352
353fn decode_users_snapshot(_path: &Path, bytes: &[u8]) -> AuthServiceResult<HashMap<String, User>> {
354    if bytes.iter().all(u8::is_ascii_whitespace) {
355        return Ok(HashMap::new());
356    }
357    serde_json::from_slice(bytes).map_err(|source| {
358        AuthServiceError::with_source(AuthOperation::DecodeMetadata, AuthFailureKind::InvalidData, source)
359    })
360}
361
362async fn write_users_snapshot(path: &Path, content: Vec<u8>) -> AuthServiceResult<()> {
363    if let Some(parent) = path.parent() {
364        fs::create_dir_all(parent)
365            .await
366            .map_err(AuthServiceError::storage_write_failed)?;
367    }
368
369    let temp_file = temp_snapshot_path(path);
370    fs::write(&temp_file, content)
371        .await
372        .map_err(AuthServiceError::storage_write_failed)?;
373
374    match fs::rename(&temp_file, path).await {
375        Ok(()) => Ok(()),
376        Err(_rename_error) => {
377            fs::copy(&temp_file, path)
378                .await
379                .map_err(AuthServiceError::storage_write_failed)?;
380            let _ = fs::remove_file(&temp_file).await;
381            Ok(())
382        }
383    }
384}
385
386fn temp_snapshot_path(path: &Path) -> PathBuf {
387    let file_name = path
388        .file_name()
389        .and_then(|value| value.to_str())
390        .unwrap_or("users.json");
391    let nanos = std::time::SystemTime::now()
392        .duration_since(std::time::UNIX_EPOCH)
393        .map_or(0, |duration| duration.as_nanos());
394    path.with_file_name(format!(".{file_name}.{nanos}.tmp"))
395}
396
397#[cfg(test)]
398mod tests {
399    use std::fs;
400
401    use cheetah_string::CheetahString;
402    use tempfile::TempDir;
403
404    use super::*;
405    use crate::authentication::enums::user_status::UserStatus;
406
407    #[tokio::test]
408    async fn test_local_provider_initialization() {
409        let config = AuthConfig::default();
410        let mut provider = LocalAuthenticationMetadataProvider::new();
411
412        assert!(provider.initialize(config, None).await.is_ok());
413    }
414
415    #[tokio::test]
416    async fn test_create_and_get_user() {
417        let mut provider = LocalAuthenticationMetadataProvider::new();
418        provider.initialize(AuthConfig::default(), None).await.unwrap();
419
420        let user = User::of("testuser");
421        provider.create_user(user).await.unwrap();
422
423        let retrieved = provider.get_user("testuser").await.unwrap();
424        assert_eq!(retrieved.username().to_string(), "testuser");
425    }
426
427    #[tokio::test]
428    async fn create_user_rejects_duplicate_without_overwriting_existing_secret() {
429        let mut provider = LocalAuthenticationMetadataProvider::new();
430        provider.initialize(AuthConfig::default(), None).await.unwrap();
431
432        provider
433            .create_user(User::of_with_password("duplicate", "first"))
434            .await
435            .unwrap();
436
437        let error = provider
438            .create_user(User::of_with_password("duplicate", "second"))
439            .await
440            .unwrap_err();
441        assert_eq!(error.operation(), AuthOperation::ManageMetadata);
442        assert_eq!(error.kind(), AuthFailureKind::Conflict);
443
444        let user = provider.get_user("duplicate").await.unwrap();
445        assert_eq!(user.password().map(|value| value.as_str()), Some("first"));
446    }
447
448    #[tokio::test]
449    async fn test_update_user() {
450        let mut provider = LocalAuthenticationMetadataProvider::new();
451        provider.initialize(AuthConfig::default(), None).await.unwrap();
452
453        let user = User::of_with_password("testuser", "password");
454        provider.create_user(user).await.unwrap();
455
456        let updated_user = User::of_with_password("testuser", "newpassword");
457        provider.update_user(updated_user).await.unwrap();
458
459        let retrieved = provider.get_user("testuser").await.unwrap();
460        assert_eq!(retrieved.username().to_string(), "testuser");
461    }
462
463    #[tokio::test]
464    async fn test_delete_user() {
465        let mut provider = LocalAuthenticationMetadataProvider::new();
466        provider.initialize(AuthConfig::default(), None).await.unwrap();
467
468        let user = User::of("testuser");
469        provider.create_user(user).await.unwrap();
470
471        provider.delete_user("testuser").await.unwrap();
472
473        assert!(provider.get_user("testuser").await.is_err());
474    }
475
476    #[tokio::test]
477    async fn test_list_users() {
478        let mut provider = LocalAuthenticationMetadataProvider::new();
479        provider.initialize(AuthConfig::default(), None).await.unwrap();
480
481        provider.create_user(User::of("user1")).await.unwrap();
482        provider.create_user(User::of("user2")).await.unwrap();
483        provider.create_user(User::of("admin1")).await.unwrap();
484
485        let all_users = provider.list_user(None).await.unwrap();
486        assert_eq!(all_users.len(), 3);
487
488        let filtered_users = provider.list_user(Some("user")).await.unwrap();
489        assert_eq!(filtered_users.len(), 2);
490    }
491
492    #[tokio::test]
493    async fn local_provider_persists_users_across_reinitialization() {
494        let temp = TempDir::new().unwrap();
495        let config = AuthConfig {
496            auth_config_path: CheetahString::from_string(temp.path().join("auth.json").to_string_lossy().into_owned()),
497            ..AuthConfig::default()
498        };
499
500        let mut provider = LocalAuthenticationMetadataProvider::new();
501        provider.initialize(config.clone(), None).await.unwrap();
502        let mut user = User::of_with_password("persisted", "secret");
503        user.set_user_status(UserStatus::Enable);
504        provider.create_user(user).await.unwrap();
505
506        let mut restarted = LocalAuthenticationMetadataProvider::new();
507        restarted.initialize(config, None).await.unwrap();
508        let restored = restarted.get_user("persisted").await.unwrap();
509
510        assert_eq!(restored.username().as_str(), "persisted");
511        assert_eq!(restored.password().map(|value| value.as_str()), Some("secret"));
512        assert_eq!(restored.user_status(), Some(UserStatus::Enable));
513    }
514
515    #[tokio::test]
516    async fn metadata_io_actor_durably_persists_authentication_snapshot() {
517        let temp = TempDir::new().unwrap();
518        let config = AuthConfig {
519            auth_config_path: CheetahString::from_string(temp.path().join("auth.json").to_string_lossy().into_owned()),
520            ..AuthConfig::default()
521        };
522        let context = rocketmq_runtime::RuntimeContext::try_from_current("auth-metadata-io-test").unwrap();
523        let actor = rocketmq_runtime::MetadataIoConfig::default()
524            .into_plan()
525            .expect("default metadata I/O config is valid")
526            .start(&context.service_context("auth"))
527            .unwrap();
528        let provider =
529            LocalAuthenticationMetadataProvider::with_config_and_metadata_io(&config, Some(actor.clone())).unwrap();
530
531        provider
532            .create_user(User::of_with_password("durable-user", "secret"))
533            .await
534            .unwrap();
535
536        let snapshot = actor.snapshot();
537        let users = snapshot
538            .resources
539            .iter()
540            .find(|resource| resource.resource.as_ref() == "auth.authentication-users")
541            .expect("authentication metadata resource should be tracked");
542        assert!(users.durable_generation.is_some());
543        assert_eq!(snapshot.pending_operations, 0);
544        assert!(temp.path().join("auth").join("users.json").is_file());
545
546        let report = actor
547            .shutdown_until(rocketmq_runtime::MetadataDeadline::after(
548                std::time::Duration::from_secs(5),
549            ))
550            .await;
551        assert!(!report.timed_out);
552    }
553
554    #[tokio::test]
555    async fn local_provider_persists_user_update_and_delete_across_reinitialization() {
556        let temp = TempDir::new().unwrap();
557        let config = AuthConfig {
558            auth_config_path: CheetahString::from_string(temp.path().join("auth.json").to_string_lossy().into_owned()),
559            ..AuthConfig::default()
560        };
561
562        let mut provider = LocalAuthenticationMetadataProvider::new();
563        provider.initialize(config.clone(), None).await.unwrap();
564        provider
565            .create_user(User::of_with_password("persisted", "first"))
566            .await
567            .unwrap();
568        provider
569            .update_user(User::of_with_password("persisted", "second"))
570            .await
571            .unwrap();
572
573        let mut restarted = LocalAuthenticationMetadataProvider::new();
574        restarted.initialize(config.clone(), None).await.unwrap();
575        let restored = restarted.get_user("persisted").await.unwrap();
576        assert_eq!(restored.password().map(|value| value.as_str()), Some("second"));
577
578        restarted.delete_user("persisted").await.unwrap();
579
580        let mut deleted_restart = LocalAuthenticationMetadataProvider::new();
581        deleted_restart.initialize(config, None).await.unwrap();
582        assert!(deleted_restart.get_user("persisted").await.is_err());
583    }
584
585    #[tokio::test]
586    async fn local_provider_rejects_corrupted_user_snapshot() {
587        let temp = TempDir::new().unwrap();
588        let config = AuthConfig {
589            auth_config_path: CheetahString::from_string(temp.path().join("auth.json").to_string_lossy().into_owned()),
590            ..AuthConfig::default()
591        };
592        let snapshot = temp.path().join("auth").join("users.json");
593        fs::create_dir_all(snapshot.parent().unwrap()).unwrap();
594        fs::write(&snapshot, b"{not valid json").unwrap();
595
596        let mut provider = LocalAuthenticationMetadataProvider::new();
597        let error = provider.initialize(config, None).await.unwrap_err();
598
599        assert_eq!(error.operation(), AuthOperation::DecodeMetadata);
600        assert_eq!(error.kind(), AuthFailureKind::InvalidData);
601        assert!(!error.to_string().contains("users.json"));
602    }
603
604    #[tokio::test]
605    async fn initialize_without_storage_path_clears_previous_snapshot() {
606        let temp = TempDir::new().unwrap();
607        let config = AuthConfig {
608            auth_config_path: CheetahString::from_string(temp.path().join("auth.json").to_string_lossy().into_owned()),
609            ..AuthConfig::default()
610        };
611
612        let mut provider = LocalAuthenticationMetadataProvider::new();
613        provider.initialize(config, None).await.unwrap();
614        provider
615            .create_user(User::of_with_password("stale", "secret"))
616            .await
617            .unwrap();
618
619        provider.initialize(AuthConfig::default(), None).await.unwrap();
620
621        assert!(provider.get_user("stale").await.is_err());
622    }
623}
rocketmq_auth/authentication/provider/local_authentication_metadata_provider.rs

rocketmq_auth/authentication/provider/
local_authentication_metadata_provider.rs
