rocketmq_auth/acl/
loader.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
15use std::collections::hash_map::DefaultHasher;
16use std::collections::HashSet;
17use std::hash::Hash;
18use std::hash::Hasher;
19use std::path::Path;
20use std::path::PathBuf;
21
22use cheetah_string::CheetahString;
23use rocketmq_runtime::BlockingExecutor;
24use tokio::fs;
25
26use crate::acl::validator::validate_acl_config;
27use crate::migration::alc::acl_config::AclConfig;
28use crate::migration::alc::plain_access_config::PlainAccessConfig;
29use crate::migration::alc::plain_access_data::DataVersion;
30use crate::migration::alc::plain_access_data::PlainAccessData;
31use crate::runtime_bridge::AuthBlockingExecutor;
32use crate::AuthFailureKind;
33use crate::AuthOperation;
34use crate::AuthServiceError;
35use crate::AuthServiceResult;
36
37#[derive(Clone, Debug)]
38pub struct FileAclConfigLoader {
39    roots: Vec<PathBuf>,
40    blocking: AuthBlockingExecutor,
41}
42
43#[derive(Clone, Debug)]
44pub struct FileAclConfigStore {
45    file: PathBuf,
46}
47
48#[derive(Clone, Copy, Debug, PartialEq, Eq)]
49pub struct AclConfigFingerprint(u64);
50
51impl FileAclConfigLoader {
52    pub fn new(root: impl Into<PathBuf>, blocking: BlockingExecutor) -> Self {
53        Self {
54            roots: vec![root.into()],
55            blocking: AuthBlockingExecutor::new(blocking),
56        }
57    }
58
59    pub async fn load_with_fingerprint(&self) -> AuthServiceResult<(AclConfig, AclConfigFingerprint)> {
60        let files = self.discover_files().await?;
61        let (config, fingerprint) = load_files(files).await?;
62        validate_acl_config(&config)?;
63        Ok((config, fingerprint))
64    }
65
66    pub async fn discover_files(&self) -> AuthServiceResult<Vec<PathBuf>> {
67        let roots = self.roots.clone();
68        self.blocking
69            .spawn_io("auth.acl.discover_files", move || discover_acl_files(&roots))
70            .await
71            .map_err(|source| {
72                AuthServiceError::with_source(AuthOperation::ReadMetadata, AuthFailureKind::Unavailable, source)
73            })?
74    }
75}
76
77impl FileAclConfigStore {
78    pub fn new(file: impl Into<PathBuf>) -> Self {
79        Self { file: file.into() }
80    }
81
82    pub async fn update_global_white_remote_addresses<I, S>(&self, addresses: I) -> AuthServiceResult<()>
83    where
84        I: IntoIterator<Item = S>,
85        S: AsRef<str>,
86    {
87        let addresses: Vec<_> = addresses
88            .into_iter()
89            .map(|address| address.as_ref().trim().to_owned())
90            .filter(|address| !address.is_empty())
91            .map(CheetahString::from)
92            .collect();
93        if addresses.is_empty() {
94            return Err(AuthServiceError::new(
95                AuthOperation::ManageMetadata,
96                AuthFailureKind::InvalidInput,
97            ));
98        }
99
100        let mut data = self.load_plain_access_data().await?;
101        data.set_global_white_remote_addresses(addresses);
102        bump_data_version(&mut data);
103        self.write_plain_access_data(&data).await
104    }
105
106    async fn load_plain_access_data(&self) -> AuthServiceResult<PlainAccessData> {
107        let bytes = match fs::read(&self.file).await {
108            Ok(bytes) => bytes,
109            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(PlainAccessData::default()),
110            Err(error) => return Err(AuthServiceError::storage_read_failed(error)),
111        };
112
113        if bytes.iter().all(u8::is_ascii_whitespace) {
114            return Ok(PlainAccessData::default());
115        }
116
117        serde_yaml::from_slice(&bytes).map_err(|source| {
118            AuthServiceError::with_source(AuthOperation::DecodeMetadata, AuthFailureKind::InvalidData, source)
119        })
120    }
121
122    async fn write_plain_access_data(&self, data: &PlainAccessData) -> AuthServiceResult<()> {
123        let content = serde_yaml::to_string(data).map_err(|source| {
124            AuthServiceError::with_source(AuthOperation::EncodeMetadata, AuthFailureKind::InvalidData, source)
125        })?;
126
127        if let Some(parent) = self.file.parent() {
128            fs::create_dir_all(parent)
129                .await
130                .map_err(AuthServiceError::storage_write_failed)?;
131        }
132
133        if fs::metadata(&self.file).await.is_ok() {
134            let backup = backup_file_path(&self.file);
135            fs::copy(&self.file, &backup)
136                .await
137                .map_err(AuthServiceError::storage_write_failed)?;
138        }
139
140        let temp_file = temp_file_path(&self.file);
141        fs::write(&temp_file, content)
142            .await
143            .map_err(AuthServiceError::storage_write_failed)?;
144
145        match fs::rename(&temp_file, &self.file).await {
146            Ok(()) => Ok(()),
147            Err(_rename_error) => {
148                fs::copy(&temp_file, &self.file)
149                    .await
150                    .map_err(AuthServiceError::storage_write_failed)?;
151                let _ = fs::remove_file(&temp_file).await;
152                Ok(())
153            }
154        }
155    }
156}
157
158async fn load_files(files: Vec<PathBuf>) -> AuthServiceResult<(AclConfig, AclConfigFingerprint)> {
159    let mut accounts = Vec::new();
160    let mut global_white_addrs = Vec::new();
161    let mut seen_access_keys = HashSet::new();
162    let mut hasher = DefaultHasher::new();
163
164    for file in files {
165        file.hash(&mut hasher);
166        let bytes = fs::read(&file).await.map_err(AuthServiceError::storage_read_failed)?;
167        bytes.hash(&mut hasher);
168        if bytes.iter().all(u8::is_ascii_whitespace) {
169            continue;
170        }
171
172        let data: PlainAccessData = serde_yaml::from_slice(&bytes).map_err(|source| {
173            AuthServiceError::with_source(AuthOperation::DecodeMetadata, AuthFailureKind::InvalidData, source)
174        })?;
175
176        global_white_addrs.extend(data.global_white_remote_addresses().iter().cloned());
177        extend_accounts(&mut accounts, &mut seen_access_keys, data.accounts());
178    }
179
180    let mut config = AclConfig::new();
181    if !global_white_addrs.is_empty() {
182        config.set_global_white_addrs(global_white_addrs);
183    }
184    if !accounts.is_empty() {
185        config.set_plain_access_configs(accounts);
186    }
187    Ok((config, AclConfigFingerprint(hasher.finish())))
188}
189
190fn extend_accounts(
191    accounts: &mut Vec<PlainAccessConfig>,
192    seen_access_keys: &mut HashSet<String>,
193    incoming: &[PlainAccessConfig],
194) {
195    for account in incoming {
196        let Some(access_key) = account.access_key() else {
197            continue;
198        };
199        if seen_access_keys.insert(access_key.to_string()) {
200            accounts.push(account.clone());
201        }
202    }
203}
204
205fn discover_acl_files(roots: &[PathBuf]) -> AuthServiceResult<Vec<PathBuf>> {
206    let mut files = Vec::new();
207    for root in roots {
208        collect_acl_files(root, &mut files)?;
209    }
210    Ok(files)
211}
212
213fn collect_acl_files(path: &Path, files: &mut Vec<PathBuf>) -> AuthServiceResult<()> {
214    let metadata = std::fs::metadata(path).map_err(AuthServiceError::storage_read_failed)?;
215
216    if metadata.is_file() {
217        if is_acl_yaml(path) {
218            files.push(path.to_path_buf());
219        }
220        return Ok(());
221    }
222
223    if metadata.is_dir() {
224        let mut entries = Vec::new();
225        for entry in std::fs::read_dir(path).map_err(AuthServiceError::storage_read_failed)? {
226            let entry = entry.map_err(AuthServiceError::storage_read_failed)?;
227            entries.push(entry.path());
228        }
229        entries.sort();
230        for entry in entries {
231            collect_acl_files(&entry, files)?;
232        }
233    }
234
235    Ok(())
236}
237
238fn is_acl_yaml(path: &Path) -> bool {
239    let Some(extension) = path.extension().and_then(|value| value.to_str()) else {
240        return false;
241    };
242    if !extension.eq_ignore_ascii_case("yml") && !extension.eq_ignore_ascii_case("yaml") {
243        return false;
244    }
245    path.file_name()
246        .and_then(|value| value.to_str())
247        .is_none_or(|file_name| !file_name.eq_ignore_ascii_case("tools.yml"))
248}
249
250fn backup_file_path(path: &Path) -> PathBuf {
251    let mut backup = path.as_os_str().to_os_string();
252    backup.push(".bak");
253    PathBuf::from(backup)
254}
255
256fn temp_file_path(path: &Path) -> PathBuf {
257    let file_name = path
258        .file_name()
259        .and_then(|value| value.to_str())
260        .unwrap_or("plain_acl.yml");
261    let nanos = std::time::SystemTime::now()
262        .duration_since(std::time::UNIX_EPOCH)
263        .map(|duration| duration.as_nanos())
264        .unwrap_or_default();
265    path.with_file_name(format!(".{file_name}.{}.tmp", nanos))
266}
267
268fn bump_data_version(data: &mut PlainAccessData) {
269    let counter = data
270        .latest_version()
271        .map_or(1, |version| version.counter.saturating_add(1));
272    data.set_data_version(vec![DataVersion {
273        timestamp: current_time_millis(),
274        counter,
275    }]);
276}
277
278fn current_time_millis() -> u64 {
279    std::time::SystemTime::now()
280        .duration_since(std::time::UNIX_EPOCH)
281        .map_or(0, |duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
282}
283
284#[cfg(test)]
285mod tests {
286    use std::fs;
287
288    use cheetah_string::CheetahString;
289    use rocketmq_runtime::RuntimeContext;
290    use tempfile::TempDir;
291
292    use super::*;
293
294    fn metadata_blocking() -> BlockingExecutor {
295        RuntimeContext::from_current("auth-acl-loader-test")
296            .service_context("auth-acl-loader")
297            .metadata_io()
298            .clone()
299    }
300
301    #[tokio::test]
302    async fn discovers_yaml_files_recursively_and_skips_tools_file() {
303        let temp = TempDir::new().unwrap();
304        let root = temp.path();
305        let nested = root.join("nested");
306        fs::create_dir_all(&nested).unwrap();
307        fs::write(root.join("a.yml"), "").unwrap();
308        fs::write(root.join("tools.yml"), "").unwrap();
309        fs::write(nested.join("b.yaml"), "").unwrap();
310        fs::write(nested.join("ignored.txt"), "").unwrap();
311
312        let files = FileAclConfigLoader::new(root, metadata_blocking())
313            .discover_files()
314            .await
315            .unwrap();
316
317        assert_eq!(files.len(), 2);
318        assert!(files.iter().any(|path| path.ends_with("a.yml")));
319        assert!(files.iter().any(|path| path.ends_with("b.yaml")));
320        assert!(!files.iter().any(|path| path.ends_with("tools.yml")));
321    }
322
323    #[tokio::test]
324    async fn loads_java_acl_files_and_keeps_first_duplicate_access_key() {
325        let temp = TempDir::new().unwrap();
326        let first = temp.path().join("first.yml");
327        let second = temp.path().join("second.yml");
328        fs::write(
329            &first,
330            r#"
331globalWhiteRemoteAddresses:
332  - 10.0.0.1
333accounts:
334  - accessKey: ak
335    secretKey: first-secret
336    admin: true
337"#,
338        )
339        .unwrap();
340        fs::write(
341            &second,
342            r#"
343globalWhiteRemoteAddresses:
344  - 10.0.0.2
345accounts:
346  - accessKey: ak
347    secretKey: second-secret
348  - accessKey: bk
349    secretKey: bk-secret
350"#,
351        )
352        .unwrap();
353
354        let (config, _) = FileAclConfigLoader::new(temp.path(), metadata_blocking())
355            .load_with_fingerprint()
356            .await
357            .unwrap();
358
359        let white_addrs = config.global_white_addrs().unwrap();
360        assert_eq!(
361            white_addrs,
362            &[CheetahString::from("10.0.0.1"), CheetahString::from("10.0.0.2")]
363        );
364
365        let accounts = config.plain_access_configs().unwrap();
366        assert_eq!(accounts.len(), 2);
367        assert_eq!(accounts[0].access_key().unwrap().as_str(), "ak");
368        assert_eq!(accounts[0].secret_key().unwrap().as_str(), "first-secret");
369        assert_eq!(accounts[1].access_key().unwrap().as_str(), "bk");
370    }
371
372    #[tokio::test]
373    async fn empty_files_load_as_empty_config() {
374        let temp = TempDir::new().unwrap();
375        let file = temp.path().join("plain_acl.yml");
376        fs::write(&file, " \n\t").unwrap();
377
378        let (config, _) = FileAclConfigLoader::new(file, metadata_blocking())
379            .load_with_fingerprint()
380            .await
381            .unwrap();
382
383        assert!(config.global_white_addrs().is_none());
384        assert!(config.plain_access_configs().is_none());
385    }
386
387    #[tokio::test]
388    async fn fingerprint_changes_when_acl_file_content_changes() {
389        let temp = TempDir::new().unwrap();
390        let file = temp.path().join("plain_acl.yml");
391        fs::write(
392            &file,
393            r#"
394accounts:
395  - accessKey: ak
396    secretKey: first
397"#,
398        )
399        .unwrap();
400
401        let loader = FileAclConfigLoader::new(&file, metadata_blocking());
402        let (_, first_fingerprint) = loader.load_with_fingerprint().await.unwrap();
403        let (_, same_fingerprint) = loader.load_with_fingerprint().await.unwrap();
404        assert_eq!(first_fingerprint, same_fingerprint);
405
406        fs::write(
407            &file,
408            r#"
409accounts:
410  - accessKey: ak
411    secretKey: second
412"#,
413        )
414        .unwrap();
415
416        let (_, updated_fingerprint) = loader.load_with_fingerprint().await.unwrap();
417        assert_ne!(first_fingerprint, updated_fingerprint);
418    }
419
420    #[tokio::test]
421    async fn rejects_acl_file_with_missing_secret_key() {
422        let temp = TempDir::new().unwrap();
423        let file = temp.path().join("plain_acl.yml");
424        fs::write(
425            &file,
426            r#"
427accounts:
428  - accessKey: ak
429"#,
430        )
431        .unwrap();
432
433        let error = FileAclConfigLoader::new(file, metadata_blocking())
434            .load_with_fingerprint()
435            .await
436            .unwrap_err();
437
438        assert_eq!(error.kind(), AuthFailureKind::InvalidConfiguration);
439        assert!(!error.to_string().contains("secret="));
440    }
441
442    #[tokio::test]
443    async fn store_updates_global_white_remote_addresses_and_preserves_accounts() {
444        let temp = TempDir::new().unwrap();
445        let file = temp.path().join("plain_acl.yml");
446        fs::write(
447            &file,
448            r#"
449globalWhiteRemoteAddresses:
450  - 10.10.*.*
451accounts:
452  - accessKey: ak
453    secretKey: sk
454    defaultTopicPerm: PUB
455"#,
456        )
457        .unwrap();
458
459        FileAclConfigStore::new(&file)
460            .update_global_white_remote_addresses(["172.16.*.*", "192.168.0.*"])
461            .await
462            .unwrap();
463
464        let content = fs::read_to_string(&file).unwrap();
465        let data: PlainAccessData = serde_yaml::from_str(&content).unwrap();
466        assert_eq!(
467            data.global_white_remote_addresses(),
468            &[
469                CheetahString::from_static_str("172.16.*.*"),
470                CheetahString::from_static_str("192.168.0.*")
471            ]
472        );
473        assert_eq!(data.accounts().len(), 1);
474        assert_eq!(data.accounts()[0].access_key().unwrap().as_str(), "ak");
475        assert_eq!(data.accounts()[0].secret_key().unwrap().as_str(), "sk");
476        assert!(content.contains("globalWhiteRemoteAddresses"));
477        assert!(content.contains("accessKey"));
478    }
479
480    #[tokio::test]
481    async fn store_creates_missing_acl_file_for_global_white_remote_addresses() {
482        let temp = TempDir::new().unwrap();
483        let file = temp.path().join("conf").join("plain_acl.yml");
484
485        FileAclConfigStore::new(&file)
486            .update_global_white_remote_addresses(["10.10.*.*"])
487            .await
488            .unwrap();
489
490        let data: PlainAccessData = serde_yaml::from_str(&fs::read_to_string(&file).unwrap()).unwrap();
491        assert_eq!(
492            data.global_white_remote_addresses(),
493            &[CheetahString::from_static_str("10.10.*.*")]
494        );
495        assert!(data.accounts().is_empty());
496    }
497
498    #[tokio::test]
499    async fn store_bumps_data_version_when_global_white_remote_addresses_change() {
500        let temp = TempDir::new().unwrap();
501        let file = temp.path().join("plain_acl.yml");
502        fs::write(
503            &file,
504            r#"
505globalWhiteRemoteAddresses:
506  - 10.10.*.*
507dataVersion:
508  - timestamp: 100
509    counter: 7
510accounts:
511  - accessKey: ak
512    secretKey: sk
513"#,
514        )
515        .unwrap();
516
517        FileAclConfigStore::new(&file)
518            .update_global_white_remote_addresses(["172.16.*.*"])
519            .await
520            .unwrap();
521
522        let data: PlainAccessData = serde_yaml::from_str(&fs::read_to_string(&file).unwrap()).unwrap();
523        let version = data.latest_version().unwrap();
524        assert_eq!(version.counter, 8);
525        assert!(version.timestamp > 100);
526    }
527
528    #[tokio::test]
529    async fn store_creates_data_version_when_missing() {
530        let temp = TempDir::new().unwrap();
531        let file = temp.path().join("plain_acl.yml");
532
533        FileAclConfigStore::new(&file)
534            .update_global_white_remote_addresses(["10.10.*.*"])
535            .await
536            .unwrap();
537
538        let data: PlainAccessData = serde_yaml::from_str(&fs::read_to_string(&file).unwrap()).unwrap();
539        let version = data.latest_version().unwrap();
540        assert_eq!(version.counter, 1);
541        assert!(version.timestamp > 0);
542    }
543}
rocketmq_auth/acl/loader.rs

rocketmq_auth/acl/
loader.rs
