rocketmq_auth/authorization/
factory.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::any::Any;
16use std::collections::HashMap;
17use std::sync::Arc;
18use std::sync::Mutex;
19use std::sync::OnceLock;
20
21use rocketmq_protocol::protocol::remoting_command::RemotingCommand;
22
23use crate::authorization::builder::default_authorization_context_builder::DefaultAuthorizationContextBuilder;
24use crate::authorization::builder::AuthorizationContextBuilder;
25use crate::authorization::context::default_authorization_context::DefaultAuthorizationContext;
26use crate::authorization::evaluator::AuthorizationEvaluator;
27use crate::authorization::manager::metadata_manager_impl::AuthorizationMetadataManagerImpl;
28use crate::authorization::metadata_provider::AuthorizationMetadataProvider;
29use crate::authorization::metadata_provider::LocalAuthorizationMetadataProvider;
30use crate::authorization::provider::AuthorizationProvider;
31use crate::authorization::provider::DefaultAuthorizationProvider;
32use crate::authorization::strategy::AuthorizationStrategy;
33use crate::authorization::strategy::StatefulAuthorizationStrategy;
34use crate::authorization::strategy::StatelessAuthorizationStrategy;
35use crate::config::AuthConfig;
36use crate::AuthFailureKind;
37use crate::AuthOperation;
38use crate::AuthServiceError;
39use crate::AuthServiceResult;
40
41static INSTANCE_CACHE: OnceLock<Mutex<HashMap<String, Arc<dyn Any + Send + Sync>>>> = OnceLock::new();
42
43const PROVIDER_PREFIX: &str = "PROVIDER_";
44const METADATA_PROVIDER_PREFIX: &str = "METADATA_PROVIDER_";
45const EVALUATOR_PREFIX: &str = "EVALUATOR_";
46
47/// Compatibility factory for standalone callers. Runtime composition uses an
48/// injected [`crate::ProviderRegistry`]; only legacy calls without one consult
49/// the process cache by config name.
50pub struct AuthorizationFactory;
51
52impl AuthorizationFactory {
53    pub fn get_provider(config: &AuthConfig) -> AuthServiceResult<Arc<DefaultAuthorizationProvider>> {
54        if !is_blank_or_supported(
55            config.authorization_provider.as_str(),
56            &["DefaultAuthorizationProvider", "default"],
57        ) {
58            return Err(unsupported(
59                "authorizationProvider",
60                config.authorization_provider.as_str(),
61            ));
62        }
63
64        let key = format!("{}{}", PROVIDER_PREFIX, config.config_name);
65        Self::compute_if_absent(&key, || {
66            let provider = new_initialized_default_authorization_provider(config.clone(), None)?;
67            Ok(Arc::new(provider) as Arc<dyn Any + Send + Sync>)
68        })
69        .and_then(|value| {
70            value
71                .downcast::<DefaultAuthorizationProvider>()
72                .map_err(|_| AuthServiceError::new(AuthOperation::Initialize, AuthFailureKind::Internal))
73        })
74    }
75
76    pub fn get_metadata_provider(
77        config: &AuthConfig,
78    ) -> AuthServiceResult<Option<Arc<LocalAuthorizationMetadataProvider>>> {
79        if config.authorization_metadata_provider.is_empty() {
80            return Ok(None);
81        }
82        if !is_supported(
83            config.authorization_metadata_provider.as_str(),
84            &["LocalAuthorizationMetadataProvider", "local"],
85        ) {
86            return Err(unsupported(
87                "authorizationMetadataProvider",
88                config.authorization_metadata_provider.as_str(),
89            ));
90        }
91
92        let key = format!("{}{}", METADATA_PROVIDER_PREFIX, config.config_name);
93        Self::compute_if_absent(&key, || {
94            let mut provider = LocalAuthorizationMetadataProvider::new();
95            provider.initialize(config.clone(), None)?;
96            Ok(Arc::new(provider) as Arc<dyn Any + Send + Sync>)
97        })
98        .and_then(|value| {
99            value
100                .downcast::<LocalAuthorizationMetadataProvider>()
101                .map_err(|_| AuthServiceError::new(AuthOperation::Initialize, AuthFailureKind::Internal))
102        })
103        .map(Some)
104    }
105
106    pub fn get_metadata_manager(config: &AuthConfig) -> AuthServiceResult<AuthorizationMetadataManagerImpl> {
107        AuthorizationMetadataManagerImpl::from_config(config)
108    }
109
110    pub fn get_strategy(
111        config: &AuthConfig,
112        metadata_service: Option<Box<dyn Any + Send + Sync>>,
113    ) -> AuthServiceResult<Box<dyn AuthorizationStrategy>> {
114        let strategy = config.authorization_strategy.as_str();
115        if strategy.trim().is_empty()
116            || strategy.ends_with("StatelessAuthorizationStrategy")
117            || strategy.eq_ignore_ascii_case("stateless")
118        {
119            return StatelessAuthorizationStrategy::new(config.clone(), metadata_service)
120                .map(|strategy| Box::new(strategy) as Box<dyn AuthorizationStrategy>);
121        }
122        if strategy.ends_with("StatefulAuthorizationStrategy") || strategy.eq_ignore_ascii_case("stateful") {
123            return StatefulAuthorizationStrategy::new(config.clone(), metadata_service)
124                .map(|strategy| Box::new(strategy) as Box<dyn AuthorizationStrategy>);
125        }
126        Err(AuthServiceError::configuration_error(format!(
127            "Unsupported authorization strategy: {strategy}"
128        )))
129    }
130
131    pub fn get_evaluator(
132        config: &AuthConfig,
133    ) -> AuthServiceResult<Arc<AuthorizationEvaluator<Box<dyn AuthorizationStrategy>>>> {
134        let key = format!("{}{}", EVALUATOR_PREFIX, config.config_name);
135        Self::compute_if_absent(&key, || {
136            let strategy = Self::get_strategy(config, None)?;
137            Ok(Arc::new(AuthorizationEvaluator::new(strategy)) as Arc<dyn Any + Send + Sync>)
138        })
139        .and_then(|value| {
140            value
141                .downcast::<AuthorizationEvaluator<Box<dyn AuthorizationStrategy>>>()
142                .map_err(|_| AuthServiceError::configuration_error("Failed to downcast authorization evaluator"))
143        })
144    }
145
146    pub fn new_contexts_from_command(
147        config: &AuthConfig,
148        auth_context: &crate::RemotingAuthContext,
149        command: &RemotingCommand,
150    ) -> AuthServiceResult<Option<Vec<DefaultAuthorizationContext>>> {
151        let _provider = Self::get_provider(config)?;
152        let builder = DefaultAuthorizationContextBuilder::new(config.clone());
153        builder.build_from_remoting(auth_context, command).map(Some)
154    }
155
156    fn compute_if_absent<F>(key: &str, factory: F) -> AuthServiceResult<Arc<dyn Any + Send + Sync>>
157    where
158        F: FnOnce() -> AuthServiceResult<Arc<dyn Any + Send + Sync>>,
159    {
160        let cache = INSTANCE_CACHE.get_or_init(|| Mutex::new(HashMap::new()));
161        {
162            let guard = cache
163                .lock()
164                .map_err(|_| AuthServiceError::new(AuthOperation::Initialize, AuthFailureKind::InvalidConfiguration))?;
165            if let Some(value) = guard.get(key) {
166                return Ok(Arc::clone(value));
167            }
168        }
169
170        let value = factory()?;
171
172        let mut guard = cache
173            .lock()
174            .map_err(|_| AuthServiceError::new(AuthOperation::Initialize, AuthFailureKind::InvalidConfiguration))?;
175        if let Some(value) = guard.get(key) {
176            return Ok(Arc::clone(value));
177        }
178        guard.insert(key.to_string(), Arc::clone(&value));
179        Ok(value)
180    }
181}
182
183impl AuthorizationStrategy for Box<dyn AuthorizationStrategy> {
184    fn evaluate<'a>(
185        &'a self,
186        context: &'a crate::authorization::context::default_authorization_context::DefaultAuthorizationContext,
187    ) -> crate::authorization::strategy::AuthorizationFuture<'a> {
188        self.as_ref().evaluate(context)
189    }
190}
191
192fn is_blank_or_supported(configured: &str, supported: &[&str]) -> bool {
193    configured.trim().is_empty() || is_supported(configured, supported)
194}
195
196fn is_supported(configured: &str, supported: &[&str]) -> bool {
197    supported
198        .iter()
199        .any(|value| configured.eq_ignore_ascii_case(value) || configured.ends_with(value))
200}
201
202fn unsupported(key: &'static str, configured: &str) -> AuthServiceError {
203    let _ = (key, configured);
204    AuthServiceError::new(AuthOperation::Initialize, AuthFailureKind::Unsupported)
205}
206
207fn new_initialized_default_authorization_provider(
208    config: AuthConfig,
209    metadata_service: Option<Box<dyn Any + Send + Sync>>,
210) -> AuthServiceResult<DefaultAuthorizationProvider> {
211    let mut provider = DefaultAuthorizationProvider::new();
212    provider.initialize_with_metadata(config, metadata_service)?;
213    Ok(provider)
214}
215
216#[cfg(test)]
217mod tests {
218    use cheetah_string::CheetahString;
219    use rocketmq_protocol::code::request_code::RequestCode;
220    use rocketmq_protocol::protocol::remoting_command::RemotingCommand;
221    use rocketmq_security_api::Action;
222
223    use super::*;
224
225    fn config(name: &str) -> AuthConfig {
226        AuthConfig {
227            config_name: CheetahString::from(name),
228            cluster_name: CheetahString::from_static_str("DefaultCluster"),
229            ..AuthConfig::default()
230        }
231    }
232
233    #[test]
234    fn get_provider_caches_by_config_name() {
235        let config = config("authorization-factory-cache");
236
237        let first = AuthorizationFactory::get_provider(&config).unwrap();
238        let second = AuthorizationFactory::get_provider(&config).unwrap();
239
240        assert!(Arc::ptr_eq(&first, &second));
241    }
242
243    #[test]
244    fn new_contexts_from_command_uses_default_builder() {
245        let config = config("authorization-factory-contexts");
246        let mut fields = HashMap::new();
247        fields.insert(
248            CheetahString::from_static_str("AccessKey"),
249            CheetahString::from_static_str("alice"),
250        );
251        fields.insert(
252            CheetahString::from_static_str("topic"),
253            CheetahString::from_static_str("TopicA"),
254        );
255        let command =
256            RemotingCommand::create_remoting_command(RequestCode::SendMessage.to_i32()).set_ext_fields(fields);
257
258        let contexts = AuthorizationFactory::new_contexts_from_command(
259            &config,
260            &crate::RemotingAuthContext::embedded("test-session"),
261            &command,
262        )
263        .unwrap()
264        .unwrap();
265
266        assert_eq!(contexts.len(), 1);
267        assert_eq!(contexts[0].subject_key(), Some("User:alice"));
268        assert_eq!(contexts[0].resource_key(), Some("Topic:TopicA".to_string()));
269        assert_eq!(contexts[0].actions(), &[Action::Pub]);
270    }
271}
rocketmq_auth/authorization/factory.rs

rocketmq_auth/authorization/
factory.rs
