rustfs_audit/
registry.rs

1//  Copyright 2024 RustFS Team
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 crate::{AuditEntry, AuditError, AuditResult, factory::builtin_target_plugins};
16use rustfs_config::audit::AUDIT_ROUTE_PREFIX;
17use rustfs_config::server_config::{Config, KVS};
18use rustfs_targets::arn::TargetID;
19use rustfs_targets::{SharedTarget, Target, TargetError, TargetPluginRegistry, TargetRuntimeManager};
20use tracing::info;
21
22const LOG_COMPONENT_AUDIT: &str = "audit";
23const LOG_SUBSYSTEM_REGISTRY: &str = "registry";
24const EVENT_AUDIT_TARGET_REGISTRY_KEY_CREATED: &str = "audit_target_registry_key_created";
25const EVENT_AUDIT_TARGET_REGISTRY_STATE: &str = "audit_target_registry_state";
26
27/// Registry for managing audit targets
28pub struct AuditRegistry {
29    /// Storage for created targets
30    targets: TargetRuntimeManager<AuditEntry>,
31    /// Registered plugins for creating targets
32    plugins: TargetPluginRegistry<AuditEntry>,
33}
34
35impl Default for AuditRegistry {
36    fn default() -> Self {
37        Self::new()
38    }
39}
40
41impl AuditRegistry {
42    /// Creates a new AuditRegistry
43    pub fn new() -> Self {
44        let mut plugins = TargetPluginRegistry::new();
45        plugins.register_all(builtin_target_plugins());
46
47        AuditRegistry {
48            targets: TargetRuntimeManager::new(),
49            plugins,
50        }
51    }
52
53    pub fn supports_target_type(&self, target_type: &str) -> bool {
54        self.plugins.supports_target_type(target_type)
55    }
56
57    /// Creates a target of the specified type with the given ID and configuration
58    ///
59    /// # Arguments
60    /// * `target_type` - The type of the target (e.g., "webhook", "mqtt").
61    /// * `id` - The identifier for the target instance.
62    /// * `config` - The configuration key-value store for the target.
63    ///
64    /// # Returns
65    /// * `Result<Box<dyn Target<AuditEntry> + Send + Sync>, TargetError>` - The created target or an error.
66    pub async fn create_target(
67        &self,
68        target_type: &str,
69        id: String,
70        config: &KVS,
71    ) -> Result<Box<dyn Target<AuditEntry> + Send + Sync>, TargetError> {
72        self.plugins.create_target(target_type, id, config)
73    }
74
75    /// Creates all targets from a configuration
76    /// Create all notification targets from system configuration and environment variables.
77    /// This method processes the creation of each target concurrently as follows:
78    /// 1. Iterate through all registered target types (e.g. webhooks, mqtt).
79    /// 2. For each type, resolve its configuration in the configuration file and environment variables.
80    /// 3. Identify all target instance IDs that need to be created.
81    /// 4. Combine the default configuration, file configuration, and environment variable configuration for each instance.
82    /// 5. If the instance is enabled, create an asynchronous task for it to instantiate.
83    /// 6. Concurrency executes all creation tasks and collects results.
84    pub async fn create_audit_targets_from_config(
85        &self,
86        config: &Config,
87    ) -> AuditResult<Vec<Box<dyn Target<AuditEntry> + Send + Sync>>> {
88        self.plugins
89            .create_targets_from_config(config, AUDIT_ROUTE_PREFIX)
90            .await
91            .map_err(AuditError::from)
92    }
93
94    /// Adds a target to the registry
95    ///
96    /// # Arguments
97    /// * `id` - The identifier for the target.
98    /// * `target` - The target instance to be added.
99    pub fn add_target(&mut self, _id: String, target: Box<dyn Target<AuditEntry> + Send + Sync>) {
100        debug_assert_eq!(_id, target.id().to_string());
101        self.targets.add_boxed(target);
102    }
103
104    pub fn add_shared_target(&mut self, _id: String, target: SharedTarget<AuditEntry>) {
105        debug_assert_eq!(_id, target.id().to_string());
106        self.targets.add_arc(target);
107    }
108
109    /// Removes a target from the registry
110    ///
111    /// # Arguments
112    /// * `id` - The identifier for the target to be removed.
113    ///
114    /// # Returns
115    /// * `Option<SharedTarget<AuditEntry>>` - The removed target if it existed.
116    pub async fn remove_target(&mut self, id: &str) -> Option<SharedTarget<AuditEntry>> {
117        self.targets.remove_and_close(id).await
118    }
119
120    /// Gets a target from the registry
121    ///
122    /// # Arguments
123    /// * `id` - The identifier for the target to be retrieved.
124    ///
125    /// # Returns
126    /// * `Option<SharedTarget<AuditEntry>>` - The target if it exists.
127    pub fn get_target(&self, id: &str) -> Option<SharedTarget<AuditEntry>> {
128        self.targets.get(id)
129    }
130
131    /// Lists cloned target values for runtime inspection without exposing mutable registry access.
132    pub fn list_target_values(&self) -> Vec<SharedTarget<AuditEntry>> {
133        self.targets.values()
134    }
135
136    pub fn runtime_manager(&self) -> &TargetRuntimeManager<AuditEntry> {
137        &self.targets
138    }
139
140    pub fn runtime_manager_mut(&mut self) -> &mut TargetRuntimeManager<AuditEntry> {
141        &mut self.targets
142    }
143
144    /// Lists all target IDs
145    ///
146    /// # Returns
147    /// * `Vec<String>` - A vector of all target IDs in the registry.
148    pub fn list_targets(&self) -> Vec<String> {
149        self.targets.keys()
150    }
151
152    /// Closes all targets and clears the registry
153    ///
154    /// # Returns
155    /// * `AuditResult<()>` - Result indicating success or failure.
156    pub async fn close_all(&mut self) -> AuditResult<()> {
157        let mut first_error = None;
158
159        for target_id in self.targets.keys() {
160            if let Some(target) = self.targets.remove(&target_id)
161                && let Err(err) = target.close().await
162            {
163                tracing::error!(
164                    event = EVENT_AUDIT_TARGET_REGISTRY_STATE,
165                    component = LOG_COMPONENT_AUDIT,
166                    subsystem = LOG_SUBSYSTEM_REGISTRY,
167                    target_id = %target_id,
168                    state = "close_failed",
169                    error = %err,
170                    "Failed to close target during shutdown"
171                );
172                if first_error.is_none() {
173                    first_error = Some(err);
174                }
175            }
176        }
177
178        match first_error {
179            Some(err) => Err(AuditError::Target(err)),
180            None => Ok(()),
181        }
182    }
183
184    /// Creates a unique key for a target based on its type and ID
185    ///
186    /// # Arguments
187    /// * `target_type` - The type of the target (e.g., "webhook", "mqtt").
188    /// * `target_id` - The identifier for the target instance.
189    ///
190    /// # Returns
191    /// * `String` - The unique key for the target.
192    pub fn create_key(&self, target_type: &str, target_id: &str) -> String {
193        let key = TargetID::new(target_id.to_string(), target_type.to_string());
194        info!(
195            event = EVENT_AUDIT_TARGET_REGISTRY_KEY_CREATED,
196            component = LOG_COMPONENT_AUDIT,
197            subsystem = LOG_SUBSYSTEM_REGISTRY,
198            target_type = %target_type,
199            target_id = %target_id,
200            registry_key = %key,
201            "audit target registry state"
202        );
203        key.to_string()
204    }
205
206    /// Enables a target (placeholder, assumes target exists)
207    ///
208    /// # Arguments
209    /// * `target_type` - The type of the target (e.g., "webhook", "mqtt").
210    /// * `target_id` - The identifier for the target instance.
211    ///
212    /// # Returns
213    /// * `AuditResult<()>` - Result indicating success or failure.
214    pub fn enable_target(&self, target_type: &str, target_id: &str) -> AuditResult<()> {
215        let key = self.create_key(target_type, target_id);
216        if self.get_target(&key).is_some() {
217            info!(
218                event = EVENT_AUDIT_TARGET_REGISTRY_STATE,
219                component = LOG_COMPONENT_AUDIT,
220                subsystem = LOG_SUBSYSTEM_REGISTRY,
221                target_type = %target_type,
222                target_id = %target_id,
223                state = "enabled",
224                "audit target registry state"
225            );
226            Ok(())
227        } else {
228            Err(AuditError::Configuration(
229                format!("Target not found: {}-{}", target_type, target_id),
230                None,
231            ))
232        }
233    }
234
235    /// Disables a target (placeholder, assumes target exists)
236    ///
237    /// # Arguments
238    /// * `target_type` - The type of the target (e.g., "webhook", "mqtt").
239    /// * `target_id` - The identifier for the target instance.
240    ///
241    /// # Returns
242    /// * `AuditResult<()>` - Result indicating success or failure.
243    pub fn disable_target(&self, target_type: &str, target_id: &str) -> AuditResult<()> {
244        let key = self.create_key(target_type, target_id);
245        if self.get_target(&key).is_some() {
246            info!(
247                event = EVENT_AUDIT_TARGET_REGISTRY_STATE,
248                component = LOG_COMPONENT_AUDIT,
249                subsystem = LOG_SUBSYSTEM_REGISTRY,
250                target_type = %target_type,
251                target_id = %target_id,
252                state = "disabled",
253                "audit target registry state"
254            );
255            Ok(())
256        } else {
257            Err(AuditError::Configuration(
258                format!("Target not found: {}-{}", target_type, target_id),
259                None,
260            ))
261        }
262    }
263
264    /// Upserts a target into the registry
265    ///
266    /// # Arguments
267    /// * `target_type` - The type of the target (e.g., "webhook", "mqtt").
268    /// * `target_id` - The identifier for the target instance.
269    /// * `target` - The target instance to be upserted.
270    ///
271    /// # Returns
272    /// * `AuditResult<()>` - Result indicating success or failure.
273    pub fn upsert_target(
274        &mut self,
275        target_type: &str,
276        target_id: &str,
277        target: Box<dyn Target<AuditEntry> + Send + Sync>,
278    ) -> AuditResult<()> {
279        let key = self.create_key(target_type, target_id);
280        debug_assert_eq!(key, target.id().to_string());
281        self.targets.add_boxed(target);
282        Ok(())
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use super::AuditRegistry;
289    use crate::AuditError;
290    use rustfs_targets::TargetError;
291    use rustfs_targets::target::ChannelTargetType;
292    use rustfs_targets::testkit::MockTarget;
293
294    #[test]
295    fn registry_registers_amqp_factory() {
296        let registry = AuditRegistry::new();
297
298        assert!(registry.supports_target_type(ChannelTargetType::Amqp.as_str()));
299    }
300
301    #[tokio::test]
302    async fn close_all_returns_first_error_and_clears_targets() {
303        let mut registry = AuditRegistry::new();
304        let ok = MockTarget::new("ok", "webhook");
305        let ok_observer = ok.clone();
306        let fail = MockTarget::new("fail", "webhook")
307            .with_close_failures(usize::MAX)
308            .with_close_failure_error(|| TargetError::Unknown("close failed".to_string()));
309        let fail_observer = fail.clone();
310
311        registry.add_target(ok.target_id().to_string(), Box::new(ok));
312        registry.add_target(fail.target_id().to_string(), Box::new(fail));
313
314        let result = registry.close_all().await;
315
316        assert!(matches!(result, Err(AuditError::Target(TargetError::Unknown(_)))));
317        assert_eq!(ok_observer.close_call_count(), 1);
318        assert_eq!(fail_observer.close_call_count(), 1);
319        assert!(registry.list_targets().is_empty());
320    }
321}
rustfs_audit/registry.rs

rustfs_audit/
registry.rs
