rocketmq_auth/authentication/builder/
default_authentication_context_builder.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 implementation of authentication context builder.
16//!
17//! This module provides the standard implementation for building authentication contexts
18//! from both gRPC and Remoting protocol requests.
19
20use std::collections::BTreeMap;
21use std::collections::HashMap;
22
23use base64::Engine;
24use cheetah_string::CheetahString;
25use chrono::DateTime;
26use chrono::NaiveDateTime;
27use rocketmq_model::common::mix_all::UNIQUE_MSG_QUERY_FLAG;
28use rocketmq_model::common::mq_version::RocketMqVersion;
29use rocketmq_protocol::protocol::remoting_command::RemotingCommand;
30use tracing::warn;
31
32use crate::authentication::builder::AuthenticationContextBuilder;
33use crate::authentication::context::default_authentication_context::DefaultAuthenticationContext;
34use crate::AuthFailureKind;
35use crate::AuthOperation;
36use crate::AuthServiceError;
37use crate::AuthServiceResult;
38
39/// Constants for parsing authentication headers
40const CREDENTIAL: &str = "Credential";
41const SIGNATURE: &str = "Signature";
42const ACCESS_KEY: &str = "AccessKey";
43const SIGNATURE_FIELD: &str = "Signature";
44const X_MQ_DATE_TIME: &str = "x-mq-date-time";
45const DATE_TIME: &str = "DateTime";
46const TIMESTAMP: &str = "Timestamp";
47const TIMESTAMP_LOWERCASE: &str = "timestamp";
48const MALFORMED_GRPC_AUTHORIZATION_FRAGMENT_REASON: &str = "malformed_grpc_authorization_fragment";
49
50/// Separator constants
51const SPACE: char = ' ';
52const COMMA: char = ',';
53const EQUAL: char = '=';
54const SLASH: char = '/';
55
56/// Default authentication context builder.
57///
58/// This builder implements the standard RocketMQ authentication context construction
59/// logic, supporting both gRPC (with Authorization header parsing) and Remoting protocol
60/// (with extFields parsing) request types.
61///
62/// # Authentication Format
63///
64/// ## gRPC Authorization Header Format:
65/// ```text
66/// MQv2-HMAC-SHA1 Credential=<username>[/<additional>], SignedHeaders=x-mq-date-time, Signature=<hex-signature>
67/// ```
68///
69/// ## Remoting Protocol Format:
70/// ExtFields containing:
71/// - `AccessKey`: The username/access key
72/// - `Signature`: Base64-encoded signature
73/// - Other fields: Combined and sorted for signature verification
74///
75/// # Examples
76///
77/// ```rust,ignore
78/// use rocketmq_auth::DefaultAuthenticationContextBuilder;
79/// use rocketmq_auth::AuthenticationContextBuilder;
80///
81/// let builder = DefaultAuthenticationContextBuilder::new();
82///
83/// // From remoting command
84/// let context = builder.build_from_remoting(&command, Some("channel-123"))?;
85/// assert_eq!(context.username(), Some(&CheetahString::from("testuser")));
86/// ```
87#[derive(Debug, Default, Clone, Copy)]
88pub struct DefaultAuthenticationContextBuilder;
89
90impl DefaultAuthenticationContextBuilder {
91    /// Create a new default authentication context builder.
92    pub fn new() -> Self {
93        Self
94    }
95
96    /// Build a gRPC authentication context from plain metadata values.
97    ///
98    /// This keeps provider-level metadata parsing available without requiring callers
99    /// to construct a tonic `MetadataMap`.
100    pub fn build_from_grpc_metadata_map(
101        &self,
102        metadata: &HashMap<String, String>,
103        request: &dyn std::any::Any,
104    ) -> AuthServiceResult<DefaultAuthenticationContext> {
105        Self::build_from_grpc_values(
106            request,
107            Self::metadata_value(metadata, "authorization"),
108            Self::metadata_value(metadata, X_MQ_DATE_TIME),
109            Self::metadata_value(metadata, "channel-id"),
110        )
111    }
112
113    /// Convert hex-encoded string to base64-encoded string.
114    ///
115    /// This is used to convert the hex signature from gRPC Authorization header
116    /// to the base64 format expected by the authentication system.
117    ///
118    /// # Arguments
119    ///
120    /// * `hex_input` - Hex-encoded string
121    ///
122    /// # Returns
123    ///
124    /// * `Ok(String)` - Base64-encoded string
125    /// * `Err(AuthServiceError)` - Invalid hex input
126    fn hex_to_base64(hex_input: &str) -> AuthServiceResult<String> {
127        let bytes = hex::decode(hex_input).map_err(|source| {
128            AuthServiceError::with_source(AuthOperation::BuildContext, AuthFailureKind::InvalidInput, source)
129        })?;
130        Ok(base64::engine::general_purpose::STANDARD.encode(&bytes))
131    }
132
133    /// Combine request content for signature verification (Remoting protocol).
134    ///
135    /// This method combines the RemotingCommand fields into a single byte array
136    /// for signature verification.
137    ///
138    /// # Arguments
139    ///
140    /// * `request` - The remoting command
141    /// * `sorted_fields` - Sorted field map (excluding Signature)
142    ///
143    /// # Returns
144    ///
145    /// Byte vector containing the combined content
146    pub(crate) fn combine_request_content(
147        request: &RemotingCommand,
148        sorted_fields: &BTreeMap<CheetahString, CheetahString>,
149    ) -> Vec<u8> {
150        let mut content = Vec::new();
151
152        // Match Java AclUtils.combineRequestContent: concatenate sorted ext field
153        // values only, then append the request body. Field names are not signed.
154        for value in sorted_fields.values() {
155            content.extend_from_slice(value.as_bytes());
156        }
157
158        // Add request body if present
159        if let Some(body) = request.body() {
160            content.extend_from_slice(body);
161        }
162
163        content
164    }
165
166    fn parse_request_timestamp_millis(input: &str) -> Option<i64> {
167        let value = input.trim();
168        if value.is_empty() {
169            return None;
170        }
171
172        if value.bytes().all(|byte| byte.is_ascii_digit()) {
173            let timestamp = value.parse::<i64>().ok()?;
174            return if value.len() == 10 {
175                timestamp.checked_mul(1000)
176            } else {
177                Some(timestamp)
178            };
179        }
180
181        if let Ok(datetime) = NaiveDateTime::parse_from_str(value, "%Y%m%dT%H%M%SZ") {
182            return Some(datetime.and_utc().timestamp_millis());
183        }
184
185        DateTime::parse_from_rfc3339(value)
186            .ok()
187            .map(|datetime| datetime.timestamp_millis())
188    }
189
190    fn set_request_timestamp(context: &mut DefaultAuthenticationContext, value: &str) {
191        context.set_request_timestamp(CheetahString::from(value));
192        if let Some(timestamp_millis) = Self::parse_request_timestamp_millis(value) {
193            context.set_request_timestamp_millis(timestamp_millis);
194        }
195    }
196
197    fn metadata_value<'a>(metadata: &'a HashMap<String, String>, key: &str) -> Option<&'a str> {
198        metadata.get(key).map(String::as_str).or_else(|| {
199            metadata
200                .iter()
201                .find_map(|(candidate, value)| candidate.eq_ignore_ascii_case(key).then_some(value.as_str()))
202        })
203    }
204
205    fn build_from_grpc_values(
206        request: &dyn std::any::Any,
207        authorization: Option<&str>,
208        datetime: Option<&str>,
209        channel_id: Option<&str>,
210    ) -> AuthServiceResult<DefaultAuthenticationContext> {
211        let mut context = DefaultAuthenticationContext::new();
212
213        if let Some(channel_id) = channel_id {
214            context.base.set_channel_id(Some(CheetahString::from(channel_id)));
215        }
216
217        let rpc_code = std::any::type_name_of_val(request);
218        context.base.set_rpc_code(Some(CheetahString::from(rpc_code)));
219
220        let authorization = authorization.unwrap_or("");
221        if authorization.is_empty() {
222            return Ok(context);
223        }
224
225        let datetime = datetime
226            .ok_or_else(|| AuthServiceError::new(AuthOperation::BuildContext, AuthFailureKind::Unauthenticated))?;
227        Self::set_request_timestamp(&mut context, datetime);
228
229        let parts: Vec<&str> = authorization.splitn(2, SPACE).collect();
230        if parts.len() != 2 {
231            return Err(AuthServiceError::new(
232                AuthOperation::BuildContext,
233                AuthFailureKind::Unauthenticated,
234            ));
235        }
236
237        let key_values: Vec<&str> = parts[1].split(COMMA).collect();
238        for key_value in key_values {
239            let kv: Vec<&str> = key_value.trim().splitn(2, EQUAL).collect();
240            if kv.len() != 2 {
241                warn!(
242                    reason = MALFORMED_GRPC_AUTHORIZATION_FRAGMENT_REASON,
243                    "Ignoring malformed gRPC authorization fragment"
244                );
245                continue;
246            }
247
248            let auth_item = kv[0];
249            let value = kv[1];
250
251            match auth_item {
252                CREDENTIAL => {
253                    let credential_parts: Vec<&str> = value.split(SLASH).collect();
254                    if credential_parts.is_empty() {
255                        return Err(AuthServiceError::new(
256                            AuthOperation::BuildContext,
257                            AuthFailureKind::Unauthenticated,
258                        ));
259                    }
260                    context.set_username(CheetahString::from(credential_parts[0]));
261                }
262                SIGNATURE => {
263                    let base64_signature = Self::hex_to_base64(value)?;
264                    context.set_signature(CheetahString::from(base64_signature));
265                }
266                _ => {}
267            }
268        }
269
270        context.set_content(datetime.as_bytes().to_vec());
271
272        Ok(context)
273    }
274
275    fn remoting_request_timestamp(fields: &std::collections::HashMap<CheetahString, CheetahString>) -> Option<&str> {
276        fields
277            .get(&CheetahString::from(TIMESTAMP))
278            .or_else(|| fields.get(&CheetahString::from(TIMESTAMP_LOWERCASE)))
279            .or_else(|| fields.get(&CheetahString::from(DATE_TIME)))
280            .or_else(|| fields.get(&CheetahString::from(X_MQ_DATE_TIME)))
281            .map(CheetahString::as_str)
282    }
283}
284
285impl AuthenticationContextBuilder<DefaultAuthenticationContext> for DefaultAuthenticationContextBuilder {
286    #[cfg(feature = "grpc")]
287    fn build_from_grpc(
288        &self,
289        metadata: &tonic::metadata::MetadataMap,
290        request: &dyn std::any::Any,
291    ) -> AuthServiceResult<DefaultAuthenticationContext> {
292        Self::build_from_grpc_values(
293            request,
294            metadata.get("authorization").and_then(|v| v.to_str().ok()),
295            metadata.get(X_MQ_DATE_TIME).and_then(|v| v.to_str().ok()),
296            metadata.get("channel-id").and_then(|v| v.to_str().ok()),
297        )
298    }
299
300    fn build_from_remoting(
301        &self,
302        request: &RemotingCommand,
303        channel_id: Option<&str>,
304    ) -> AuthServiceResult<DefaultAuthenticationContext> {
305        let mut context = DefaultAuthenticationContext::new();
306
307        // Set channel ID
308        if let Some(id) = channel_id {
309            context.base.set_channel_id(Some(CheetahString::from(id)));
310        }
311
312        // Set RPC code from request code
313        context
314            .base
315            .set_rpc_code(Some(CheetahString::from(request.code().to_string())));
316
317        // Get extFields
318        let fields = match request.ext_fields() {
319            Some(fields) => fields,
320            None => return Ok(context), // No auth fields, return empty context
321        };
322
323        // Check for AccessKey
324        let access_key = match fields.get(&CheetahString::from(ACCESS_KEY)) {
325            Some(key) => key,
326            None => return Ok(context), // No AccessKey, return empty context
327        };
328
329        // Set username
330        context.set_username(access_key.clone());
331
332        // Set signature if present
333        if let Some(signature) = fields.get(&CheetahString::from(SIGNATURE_FIELD)) {
334            context.set_signature(signature.clone());
335        }
336
337        if let Some(request_timestamp) = Self::remoting_request_timestamp(fields) {
338            Self::set_request_timestamp(&mut context, request_timestamp);
339        }
340
341        // Build sorted content map for signature verification
342        let mut sorted_map = BTreeMap::new();
343
344        for (key, value) in fields.iter() {
345            // Filter out UNIQUE_MSG_QUERY_FLAG for old versions
346            if request.version() <= RocketMqVersion::V4_9_3.ordinal() as i32 && key.as_str() == UNIQUE_MSG_QUERY_FLAG {
347                continue;
348            }
349
350            // Exclude Signature field from content
351            if key.as_str() != SIGNATURE_FIELD {
352                sorted_map.insert(key.clone(), value.clone());
353            }
354        }
355
356        // Combine request content for signature verification
357        let content = Self::combine_request_content(request, &sorted_map);
358        context.set_content(content);
359
360        Ok(context)
361    }
362}
363
364#[cfg(test)]
365mod tests {
366    use super::*;
367    use std::collections::HashMap;
368    use std::sync::Arc;
369    use std::sync::Mutex;
370
371    use tracing_subscriber::fmt::MakeWriter;
372
373    #[derive(Clone)]
374    struct LogCapture {
375        output: Arc<Mutex<Vec<u8>>>,
376    }
377
378    impl LogCapture {
379        fn new() -> Self {
380            Self {
381                output: Arc::new(Mutex::new(Vec::new())),
382            }
383        }
384
385        fn contents(&self) -> String {
386            String::from_utf8(
387                self.output
388                    .lock()
389                    .expect("test log capture lock should not be poisoned")
390                    .clone(),
391            )
392            .expect("test log output should be valid UTF-8")
393        }
394    }
395
396    struct LogCaptureWriter {
397        output: Arc<Mutex<Vec<u8>>>,
398    }
399
400    impl std::io::Write for LogCaptureWriter {
401        fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
402            self.output
403                .lock()
404                .expect("test log capture lock should not be poisoned")
405                .extend_from_slice(buffer);
406            Ok(buffer.len())
407        }
408
409        fn flush(&mut self) -> std::io::Result<()> {
410            Ok(())
411        }
412    }
413
414    impl<'a> MakeWriter<'a> for LogCapture {
415        type Writer = LogCaptureWriter;
416
417        fn make_writer(&'a self) -> Self::Writer {
418            LogCaptureWriter {
419                output: Arc::clone(&self.output),
420            }
421        }
422    }
423
424    #[test]
425    fn test_hex_to_base64() {
426        let hex = "48656C6C6F";
427        let result = DefaultAuthenticationContextBuilder::hex_to_base64(hex).unwrap();
428        assert_eq!(result, "SGVsbG8=");
429    }
430
431    #[test]
432    fn test_hex_to_base64_invalid() {
433        let hex = "ZZZZ";
434        let result = DefaultAuthenticationContextBuilder::hex_to_base64(hex);
435        assert!(result.is_err());
436    }
437
438    #[test]
439    fn test_combine_request_content_matches_java_value_only_order() {
440        let request = RemotingCommand::create_remoting_command(10).set_body(Vec::from("body"));
441        let mut sorted_fields = BTreeMap::new();
442        sorted_fields.insert(CheetahString::from("AccessKey"), CheetahString::from("alice"));
443        sorted_fields.insert(CheetahString::from("topic"), CheetahString::from("topic-a"));
444
445        let content = DefaultAuthenticationContextBuilder::combine_request_content(&request, &sorted_fields);
446
447        assert_eq!(content, b"alicetopic-abody");
448    }
449
450    #[test]
451    fn parse_request_timestamp_millis_supports_java_grpc_format() {
452        let timestamp = DefaultAuthenticationContextBuilder::parse_request_timestamp_millis("20231227T194619Z");
453
454        assert_eq!(timestamp, Some(1_703_706_379_000));
455    }
456
457    #[test]
458    fn parse_request_timestamp_millis_supports_epoch_seconds_and_millis() {
459        assert_eq!(
460            DefaultAuthenticationContextBuilder::parse_request_timestamp_millis("1703706379"),
461            Some(1_703_706_379_000)
462        );
463        assert_eq!(
464            DefaultAuthenticationContextBuilder::parse_request_timestamp_millis("1703706379000"),
465            Some(1_703_706_379_000)
466        );
467    }
468
469    #[test]
470    fn parse_request_timestamp_millis_supports_rfc3339() {
471        let timestamp = DefaultAuthenticationContextBuilder::parse_request_timestamp_millis("2023-12-27T19:46:19Z");
472
473        assert_eq!(timestamp, Some(1_703_706_379_000));
474    }
475
476    #[test]
477    fn build_from_grpc_metadata_map_parses_authorization_header() {
478        let builder = DefaultAuthenticationContextBuilder::new();
479        let mut metadata = HashMap::new();
480        metadata.insert(
481            "Authorization".to_owned(),
482            "MQv2-HMAC-SHA1 Credential=alice/us-east/service, SignedHeaders=x-mq-date-time, Signature=48656C6C6F"
483                .to_owned(),
484        );
485        metadata.insert("X-Mq-Date-Time".to_owned(), "20231227T194619Z".to_owned());
486        metadata.insert("Channel-Id".to_owned(), "channel-a".to_owned());
487
488        let context = builder.build_from_grpc_metadata_map(&metadata, &()).unwrap();
489
490        assert_eq!(context.username(), Some(&CheetahString::from("alice")));
491        assert_eq!(context.signature(), Some(&CheetahString::from("SGVsbG8=")));
492        assert_eq!(context.content(), Some(b"20231227T194619Z".as_slice()));
493        assert_eq!(context.base.channel_id(), Some(&CheetahString::from("channel-a")));
494        assert_eq!(context.request_timestamp_millis(), Some(1_703_706_379_000));
495    }
496
497    #[test]
498    fn malformed_grpc_authorization_fragment_is_not_logged() {
499        const SYNTHETIC_CREDENTIAL: &str = "SYNTHETIC_CREDENTIAL_9720";
500        const SYNTHETIC_FRAGMENT: &str = "SYNTHETIC_FRAGMENT_9720";
501        const SYNTHETIC_SIGNATURE_LIKE: &str = "SYNTHETIC_SIGNATURE_LIKE_9720";
502        const SYNTHETIC_TOKEN: &str = "SYNTHETIC_TOKEN_9720";
503        const SYNTHETIC_BODY: &str = "SYNTHETIC_BODY_9720";
504
505        let authorization = format!(
506            "MQv2-HMAC-SHA1 Credential={SYNTHETIC_CREDENTIAL}/scope, SignedHeaders=x-mq-date-time, {SYNTHETIC_FRAGMENT}:{SYNTHETIC_SIGNATURE_LIKE}:{SYNTHETIC_TOKEN}:{SYNTHETIC_BODY}"
507        );
508        let capture = LogCapture::new();
509        let subscriber = tracing_subscriber::fmt()
510            .with_writer(capture.clone())
511            .with_ansi(false)
512            .without_time()
513            .with_max_level(tracing::Level::WARN)
514            .finish();
515        let guard = tracing::subscriber::set_default(subscriber);
516        let builder = DefaultAuthenticationContextBuilder::new();
517        let mut metadata = HashMap::new();
518        metadata.insert("authorization".to_owned(), authorization.clone());
519        metadata.insert("x-mq-date-time".to_owned(), "20231227T194619Z".to_owned());
520
521        let context = builder.build_from_grpc_metadata_map(&metadata, &()).unwrap();
522
523        drop(guard);
524        let logs = capture.contents();
525        assert_eq!(context.username(), Some(&CheetahString::from(SYNTHETIC_CREDENTIAL)));
526        assert!(logs.contains(MALFORMED_GRPC_AUTHORIZATION_FRAGMENT_REASON));
527        assert!(logs.contains("Ignoring malformed gRPC authorization fragment"));
528        assert!(!logs.contains(SYNTHETIC_FRAGMENT));
529        assert!(!logs.contains(SYNTHETIC_SIGNATURE_LIKE));
530        assert!(!logs.contains(SYNTHETIC_TOKEN));
531        assert!(!logs.contains(SYNTHETIC_BODY));
532        assert!(!logs.contains(SYNTHETIC_CREDENTIAL));
533        assert!(!logs.contains(&authorization));
534    }
535
536    #[test]
537    fn test_build_from_remoting_no_ext_fields() {
538        use rocketmq_protocol::protocol::command_custom_header::CommandCustomHeader;
539
540        #[derive(Debug, Clone)]
541        struct EmptyHeader;
542        impl CommandCustomHeader for EmptyHeader {
543            fn to_map(
544                &self,
545            ) -> Option<std::collections::HashMap<cheetah_string::CheetahString, cheetah_string::CheetahString>>
546            {
547                None
548            }
549        }
550
551        let builder = DefaultAuthenticationContextBuilder::new();
552        let request = RemotingCommand::create_request_command(1, EmptyHeader);
553        let result = builder.build_from_remoting(&request, Some("test-channel")).unwrap();
554
555        assert_eq!(result.base.channel_id(), Some(&CheetahString::from("test-channel")));
556        assert!(result.username().is_none());
557    }
558
559    #[test]
560    fn test_build_from_remoting_uses_java_signature_content() {
561        let mut ext_fields = HashMap::new();
562        ext_fields.insert(CheetahString::from("topic"), CheetahString::from("topic-a"));
563        ext_fields.insert(CheetahString::from("AccessKey"), CheetahString::from("alice"));
564        ext_fields.insert(CheetahString::from("Signature"), CheetahString::from("sig"));
565
566        let request = RemotingCommand::create_remoting_command(10)
567            .set_ext_fields(ext_fields)
568            .set_body(Vec::from("body"));
569        let builder = DefaultAuthenticationContextBuilder::new();
570
571        let context = builder.build_from_remoting(&request, Some("test-channel")).unwrap();
572
573        assert_eq!(context.username(), Some(&CheetahString::from("alice")));
574        assert_eq!(context.signature(), Some(&CheetahString::from("sig")));
575        assert_eq!(context.content(), Some(b"alicetopic-abody".as_slice()));
576    }
577
578    #[test]
579    fn test_build_from_remoting_extracts_request_timestamp() {
580        let mut ext_fields = HashMap::new();
581        ext_fields.insert(CheetahString::from("AccessKey"), CheetahString::from("alice"));
582        ext_fields.insert(CheetahString::from("Signature"), CheetahString::from("sig"));
583        ext_fields.insert(CheetahString::from("Timestamp"), CheetahString::from("1703706379000"));
584
585        let request = RemotingCommand::create_remoting_command(10).set_ext_fields(ext_fields);
586        let builder = DefaultAuthenticationContextBuilder::new();
587
588        let context = builder.build_from_remoting(&request, Some("test-channel")).unwrap();
589
590        assert_eq!(context.request_timestamp(), Some(&CheetahString::from("1703706379000")));
591        assert_eq!(context.request_timestamp_millis(), Some(1_703_706_379_000));
592    }
593}
rocketmq_auth/authentication/builder/default_authentication_context_builder.rs

rocketmq_auth/authentication/builder/
default_authentication_context_builder.rs
