rocketmq_auth/authentication/provider/
default_authentication_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//! Default Authentication Provider Implementation (Rust 2021 Standard)
16
17use std::any::Any;
18use std::collections::HashMap;
19use std::sync::Arc;
20
21use rocketmq_protocol::protocol::remoting_command::RemotingCommand;
22use tracing::debug;
23use tracing::info;
24
25use crate::authentication::builder::default_authentication_context_builder::DefaultAuthenticationContextBuilder;
26use crate::authentication::builder::AuthenticationContextBuilder;
27use crate::authentication::chain::default_authentication_handler::DefaultAuthenticationHandler;
28use crate::authentication::chain::handler::AuthenticationHandler;
29use crate::authentication::context::default_authentication_context::DefaultAuthenticationContext;
30use crate::authentication::provider::authentication_metadata_provider::AuthenticationMetadataProvider;
31use crate::authentication::provider::local_authentication_metadata_provider::LocalAuthenticationMetadataProvider;
32use crate::config::AuthConfig;
33use crate::runtime::ProviderRegistry;
34use crate::AuthFailureKind;
35use crate::AuthMetrics;
36use crate::AuthOperation;
37use crate::AuthServiceError;
38use crate::AuthServiceResult;
39
40use super::authentication_provider::AuthenticationProvider;
41
42/// Default authentication provider.
43pub struct DefaultAuthenticationProvider {
44    /// Authentication configuration.
45    auth_config: Option<AuthConfig>,
46
47    /// Metadata service supplier.
48    metadata_service: Option<Arc<dyn Any + Send + Sync>>,
49
50    /// Local metadata provider used by the default handler chain.
51    metadata_provider: Option<Arc<crate::UserMetadataHandle>>,
52
53    /// Authentication context builder.
54    authentication_context_builder: DefaultAuthenticationContextBuilder,
55
56    /// Authentication counters shared with the owning runtime when available.
57    metrics: AuthMetrics,
58}
59
60impl DefaultAuthenticationProvider {
61    /// Create a new default authentication provider.
62    pub fn new() -> Self {
63        Self {
64            auth_config: None,
65            metadata_service: None,
66            metadata_provider: None,
67            authentication_context_builder: DefaultAuthenticationContextBuilder::new(),
68            metrics: AuthMetrics::default(),
69        }
70    }
71
72    pub fn metadata_provider(&self) -> Option<Arc<crate::UserMetadataHandle>> {
73        self.metadata_provider.clone()
74    }
75
76    pub fn initialize_with_registry(
77        &mut self,
78        config: AuthConfig,
79        provider_registry: ProviderRegistry,
80    ) -> AuthServiceResult<()> {
81        self.auth_config = Some(config);
82        self.metadata_service = None;
83        self.authentication_context_builder = DefaultAuthenticationContextBuilder::new();
84        self.metadata_provider = Some(provider_registry.authentication_metadata_provider());
85        self.metrics = provider_registry.metrics();
86        Ok(())
87    }
88
89    /// Perform audit logging.
90    fn do_audit_log(&self, context: &DefaultAuthenticationContext, error: Option<&str>) {
91        if let Some(username) = context.username() {
92            if username.is_empty() {
93                return;
94            }
95
96            if let Some(err) = error {
97                info!(
98                    "[AUTHENTICATION] User:{} is authenticated failed. Error: {}",
99                    username, err
100                );
101            } else {
102                debug!("[AUTHENTICATION] User:{} is authenticated success.", username);
103            }
104        }
105    }
106
107    /// Internal authentication logic.
108    async fn authenticate_internal(&self, context: &DefaultAuthenticationContext) -> AuthServiceResult<()> {
109        let metadata_provider = self
110            .metadata_provider
111            .as_ref()
112            .ok_or_else(|| AuthServiceError::new(AuthOperation::Authenticate, AuthFailureKind::Unavailable))?;
113        let signature_algorithm = self
114            .auth_config
115            .as_ref()
116            .map(|config| config.signature_algorithm)
117            .unwrap_or_default();
118        let request_timestamp_expired_millis = self
119            .auth_config
120            .as_ref()
121            .map(|config| config.request_timestamp_expired_millis)
122            .unwrap_or_default();
123        let handler = DefaultAuthenticationHandler::with_options(
124            metadata_provider.clone(),
125            signature_algorithm,
126            request_timestamp_expired_millis,
127            self.metrics.clone(),
128        );
129        handler.handle(context).await
130    }
131}
132
133impl Default for DefaultAuthenticationProvider {
134    fn default() -> Self {
135        Self::new()
136    }
137}
138
139impl AuthenticationProvider for DefaultAuthenticationProvider {
140    type Context = DefaultAuthenticationContext;
141
142    /// Initialize the provider.
143    async fn initialize(
144        &mut self,
145        config: AuthConfig,
146        metadata_service: Option<Arc<dyn Any + Send + Sync>>,
147    ) -> AuthServiceResult<()> {
148        if let Some(service) = metadata_service {
149            let registry = service.downcast_ref::<ProviderRegistry>().ok_or_else(|| {
150                AuthServiceError::new(AuthOperation::InitializeProvider, AuthFailureKind::Unsupported)
151            })?;
152            return self.initialize_with_registry(config, registry.clone());
153        }
154        self.auth_config = Some(config.clone());
155        self.metadata_service = None;
156        self.authentication_context_builder = DefaultAuthenticationContextBuilder::new();
157        let mut provider = LocalAuthenticationMetadataProvider::new();
158        provider.initialize(config, None).await?;
159        self.metadata_provider = Some(crate::UserMetadataHandle::legacy(Arc::new(provider)));
160        self.metrics = AuthMetrics::default();
161        Ok(())
162    }
163
164    /// Authenticate a request.
165    async fn authenticate(&self, context: &Self::Context) -> AuthServiceResult<()> {
166        // Note: This uses handler chain pattern. For full chain support,
167        // you'd need to implement a handler chain.
168
169        let result = self.authenticate_internal(context).await;
170
171        // Audit log
172        match &result {
173            Ok(_) => self.do_audit_log(context, None),
174            Err(e) => self.do_audit_log(context, Some(&e.to_string())),
175        }
176
177        self.metrics.record_authentication_result(result.is_ok());
178        result
179    }
180
181    /// Create context from gRPC metadata.
182    fn new_context_from_metadata(
183        &self,
184        metadata: &HashMap<String, String>,
185        request: Box<dyn Any + Send>,
186    ) -> Self::Context {
187        self.authentication_context_builder
188            .build_from_grpc_metadata_map(metadata, request.as_ref())
189            .unwrap_or_else(|_| DefaultAuthenticationContext::new())
190    }
191
192    /// Create context from remoting command.
193    fn new_context_from_command(&self, command: &RemotingCommand) -> Self::Context {
194        self.authentication_context_builder
195            .build_from_remoting(command, None)
196            .unwrap_or_else(|_| DefaultAuthenticationContext::new())
197    }
198}
199
200#[cfg(test)]
201mod tests {
202    use super::*;
203    use cheetah_string::CheetahString;
204
205    #[tokio::test]
206    async fn test_default_provider_initialization() {
207        let mut provider = DefaultAuthenticationProvider::new();
208        let config = AuthConfig::default();
209
210        assert!(provider.initialize(config, None).await.is_ok());
211    }
212
213    #[test]
214    fn test_default_provider_creation() {
215        let provider = DefaultAuthenticationProvider::new();
216        assert!(provider.auth_config.is_none());
217    }
218
219    #[test]
220    fn test_new_context_from_metadata_uses_grpc_header_parser() {
221        let provider = DefaultAuthenticationProvider::new();
222        let mut metadata = HashMap::new();
223        metadata.insert(
224            "authorization".to_owned(),
225            "MQv2-HMAC-SHA1 Credential=alice, SignedHeaders=x-mq-date-time, Signature=48656C6C6F".to_owned(),
226        );
227        metadata.insert("x-mq-date-time".to_owned(), "20231227T194619Z".to_owned());
228        metadata.insert("channel-id".to_owned(), "channel-a".to_owned());
229
230        let context = provider.new_context_from_metadata(&metadata, Box::new(()));
231
232        assert_eq!(context.username(), Some(&CheetahString::from("alice")));
233        assert_eq!(context.signature(), Some(&CheetahString::from("SGVsbG8=")));
234        assert_eq!(context.content(), Some(b"20231227T194619Z".as_slice()));
235        assert_eq!(context.base.channel_id(), Some(&CheetahString::from("channel-a")));
236    }
237
238    #[tokio::test]
239    async fn test_authenticate() {
240        let mut provider = DefaultAuthenticationProvider::new();
241        provider.initialize(AuthConfig::default(), None).await.unwrap();
242
243        let context = DefaultAuthenticationContext::new();
244
245        let result = provider.authenticate(&context).await;
246        assert!(result.is_err());
247    }
248
249    #[tokio::test]
250    async fn test_authenticate_success() {
251        use crate::authentication::chain::acl_signer;
252        use crate::authentication::enums::user_status::UserStatus;
253        use crate::authentication::enums::user_type::UserType;
254
255        let mut provider = DefaultAuthenticationProvider::new();
256        provider.initialize(AuthConfig::default(), None).await.unwrap();
257
258        let metadata_provider = provider.metadata_provider().unwrap();
259        let mut user = crate::authentication::model::user::User::of_with_type("alice", "secret", UserType::Normal);
260        user.set_user_status(UserStatus::Enable);
261        metadata_provider.create_user(user).await.unwrap();
262
263        let content = b"test-content".to_vec();
264        let signature = acl_signer::cal_signature(&content, "secret").unwrap();
265
266        let mut context = DefaultAuthenticationContext::new();
267        context.set_username("alice".into());
268        context.set_content(content);
269        context.set_signature(signature.into());
270
271        assert!(provider.authenticate(&context).await.is_ok());
272    }
273
274    #[tokio::test]
275    async fn test_authenticate_uses_configured_signature_algorithm() {
276        use crate::authentication::chain::acl_signer;
277        use crate::authentication::chain::acl_signer::SignatureAlgorithm;
278        use crate::authentication::enums::user_status::UserStatus;
279        use crate::authentication::enums::user_type::UserType;
280
281        let mut provider = DefaultAuthenticationProvider::new();
282        provider
283            .initialize(
284                AuthConfig {
285                    signature_algorithm: SignatureAlgorithm::HmacSha256,
286                    ..AuthConfig::default()
287                },
288                None,
289            )
290            .await
291            .unwrap();
292
293        let metadata_provider = provider.metadata_provider().unwrap();
294        let mut user = crate::authentication::model::user::User::of_with_type("alice", "secret", UserType::Normal);
295        user.set_user_status(UserStatus::Enable);
296        metadata_provider.create_user(user).await.unwrap();
297
298        let content = b"test-content".to_vec();
299        let signature =
300            acl_signer::cal_signature_with_algorithm(&content, "secret", SignatureAlgorithm::HmacSha256).unwrap();
301
302        let mut context = DefaultAuthenticationContext::new();
303        context.set_username("alice".into());
304        context.set_content(content);
305        context.set_signature(signature.into());
306
307        assert!(provider.authenticate(&context).await.is_ok());
308    }
309
310    #[tokio::test]
311    async fn test_authenticate_rejects_expired_request_timestamp() {
312        use crate::authentication::chain::acl_signer;
313        use crate::authentication::enums::user_status::UserStatus;
314        use crate::authentication::enums::user_type::UserType;
315
316        let mut provider = DefaultAuthenticationProvider::new();
317        provider
318            .initialize(
319                AuthConfig {
320                    request_timestamp_expired_millis: 1,
321                    ..AuthConfig::default()
322                },
323                None,
324            )
325            .await
326            .unwrap();
327
328        let metadata_provider = provider.metadata_provider().unwrap();
329        let mut user = crate::authentication::model::user::User::of_with_type("alice", "secret", UserType::Normal);
330        user.set_user_status(UserStatus::Enable);
331        metadata_provider.create_user(user).await.unwrap();
332
333        let content = b"test-content".to_vec();
334        let signature = acl_signer::cal_signature(&content, "secret").unwrap();
335
336        let mut context = DefaultAuthenticationContext::new();
337        context.set_username("alice".into());
338        context.set_content(content);
339        context.set_signature(signature.into());
340        context.set_request_timestamp("1".into());
341        context.set_request_timestamp_millis(1);
342
343        let result = provider.authenticate(&context).await;
344
345        let error = result.unwrap_err();
346        assert_eq!(error.operation(), AuthOperation::Authenticate);
347        assert_eq!(error.kind(), AuthFailureKind::Expired);
348    }
349}
rocketmq_auth/authentication/provider/default_authentication_provider.rs

rocketmq_auth/authentication/provider/
default_authentication_provider.rs
