rocketmq_auth/authentication/
acl_client_rpc_hook.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//! Java-compatible ACL client RPC hook.
16
17use std::collections::BTreeMap;
18use std::fmt;
19use std::fs;
20use std::net::SocketAddr;
21use std::path::Path;
22use std::sync::Arc;
23
24use cheetah_string::CheetahString;
25use rocketmq_error::Error as CanonicalError;
26use rocketmq_error::SharedError;
27use rocketmq_protocol::protocol::remoting_command::RemotingCommand;
28use rocketmq_security_api::OutboundSigner;
29use rocketmq_security_api::Secret;
30use rocketmq_security_api::SecurityOperation;
31use rocketmq_security_api::SecurityProviderError;
32use rocketmq_security_api::SecurityProviderFailure;
33use rocketmq_security_api::SecurityRequestView;
34use rocketmq_security_api::Signature;
35use rocketmq_transport::api::RPCHook;
36use serde::Deserialize;
37
38use crate::authentication::builder::DefaultAuthenticationContextBuilder;
39use crate::authentication::chain::acl_signer;
40use crate::authentication::chain::acl_signer::SignatureAlgorithm;
41use crate::config::AuthConfig;
42use crate::AuthFailureKind;
43use crate::AuthOperation;
44use crate::AuthServiceError;
45use crate::AuthServiceResult;
46
47const ACCESS_KEY: &str = "AccessKey";
48const SECURITY_TOKEN: &str = "SecurityToken";
49const SIGNATURE: &str = "Signature";
50
51/// Optional mounted JSON credentials used when inline inner-client credentials are absent.
52pub const INNER_CLIENT_CREDENTIALS_FILE_ENV: &str = "ROCKETMQ_INNER_CLIENT_CREDENTIALS_FILE";
53
54#[derive(Clone, PartialEq, Eq)]
55pub struct AclClientRpcHook {
56    access_key: CheetahString,
57    secret_key: CheetahString,
58    security_token: Option<CheetahString>,
59    signature_algorithm: SignatureAlgorithm,
60}
61
62impl fmt::Debug for AclClientRpcHook {
63    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
64        f.debug_struct("AclClientRpcHook")
65            .field("access_key", &self.access_key)
66            .field("secret_key", &"<redacted>")
67            .field("security_token", &self.security_token.as_ref().map(|_| "<redacted>"))
68            .field("signature_algorithm", &self.signature_algorithm)
69            .finish()
70    }
71}
72
73impl AclClientRpcHook {
74    pub fn new(
75        access_key: impl Into<CheetahString>,
76        secret_key: impl Into<CheetahString>,
77        security_token: Option<impl Into<CheetahString>>,
78        signature_algorithm: SignatureAlgorithm,
79    ) -> Option<Self> {
80        let access_key = access_key.into();
81        let secret_key = secret_key.into();
82        if access_key.trim().is_empty() || secret_key.trim().is_empty() {
83            return None;
84        }
85
86        Some(Self {
87            access_key: CheetahString::from(access_key.trim()),
88            secret_key: CheetahString::from(secret_key.trim()),
89            security_token: security_token.and_then(|token| non_blank(token.into())),
90            signature_algorithm,
91        })
92    }
93
94    pub fn from_auth_config(config: &AuthConfig) -> AuthServiceResult<Option<Self>> {
95        if config.inner_client_authentication_credentials.trim().is_empty() {
96            if let Some(path) = std::env::var_os(INNER_CLIENT_CREDENTIALS_FILE_ENV) {
97                return Self::from_required_credentials_file(Path::new(&path), config.signature_algorithm).map(Some);
98            }
99        }
100        let signer = Self::from_credentials_json(
101            config.inner_client_authentication_credentials.as_str(),
102            config.signature_algorithm,
103        )?;
104        if signer.is_none() && !config.inner_client_authentication_credentials.trim().is_empty() {
105            return Err(AuthServiceError::new(
106                AuthOperation::Initialize,
107                AuthFailureKind::InvalidConfiguration,
108            ));
109        }
110        Ok(signer)
111    }
112
113    /// Loads an explicitly configured Secret file; missing or empty credentials fail startup.
114    pub fn from_required_credentials_file(
115        path: impl AsRef<Path>,
116        signature_algorithm: SignatureAlgorithm,
117    ) -> AuthServiceResult<Self> {
118        let content = fs::read_to_string(path).map_err(|error| {
119            AuthServiceError::with_source(AuthOperation::LoadSecret, AuthFailureKind::Unavailable, error)
120        })?;
121        Self::from_credentials_json(&content, signature_algorithm)?
122            .ok_or_else(|| AuthServiceError::new(AuthOperation::Initialize, AuthFailureKind::InvalidConfiguration))
123    }
124
125    pub fn from_credentials_json(
126        credentials_json: &str,
127        signature_algorithm: SignatureAlgorithm,
128    ) -> AuthServiceResult<Option<Self>> {
129        if credentials_json.trim().is_empty() {
130            return Ok(None);
131        }
132
133        let credentials = serde_json::from_str::<ClientCredentials>(credentials_json).map_err(|error| {
134            AuthServiceError::with_source(AuthOperation::Initialize, AuthFailureKind::InvalidConfiguration, error)
135        })?;
136        Ok(Self::from_credentials(credentials, signature_algorithm))
137    }
138
139    pub fn from_tools_file(
140        path: impl AsRef<Path>,
141        signature_algorithm: SignatureAlgorithm,
142    ) -> AuthServiceResult<Option<Self>> {
143        let path = path.as_ref();
144        let content = match fs::read_to_string(path) {
145            Ok(content) => content,
146            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
147            Err(error) => {
148                return Err(AuthServiceError::with_source(
149                    AuthOperation::LoadSecret,
150                    AuthFailureKind::Unavailable,
151                    error,
152                ));
153            }
154        };
155        let credentials = serde_yaml::from_str::<ClientCredentials>(&content).map_err(|error| {
156            AuthServiceError::with_source(AuthOperation::Initialize, AuthFailureKind::InvalidConfiguration, error)
157        })?;
158        Ok(Self::from_credentials(credentials, signature_algorithm))
159    }
160
161    pub fn into_rpc_hook(self) -> Arc<dyn RPCHook> {
162        Arc::new(self)
163    }
164
165    pub fn into_outbound_signer(self) -> Arc<dyn OutboundSigner> {
166        Arc::new(self)
167    }
168
169    pub fn access_key(&self) -> &CheetahString {
170        &self.access_key
171    }
172
173    pub fn secret_key(&self) -> &CheetahString {
174        &self.secret_key
175    }
176
177    pub fn security_token(&self) -> Option<&CheetahString> {
178        self.security_token.as_ref()
179    }
180
181    fn from_credentials(credentials: ClientCredentials, signature_algorithm: SignatureAlgorithm) -> Option<Self> {
182        Self::new(
183            credentials.access_key?,
184            credentials.secret_key?,
185            credentials.security_token,
186            signature_algorithm,
187        )
188    }
189}
190
191impl RPCHook for AclClientRpcHook {
192    fn do_before_request(&self, _remote_addr: SocketAddr, request: &mut RemotingCommand) -> Result<(), SharedError> {
193        request.ensure_ext_fields_initialized();
194        request.add_ext_field(ACCESS_KEY, self.access_key.clone());
195        if let Some(security_token) = &self.security_token {
196            request.add_ext_field(SECURITY_TOKEN, security_token.clone());
197        }
198        request.make_custom_header_to_net();
199
200        let sorted_fields = request
201            .ext_fields()
202            .into_iter()
203            .flat_map(|fields| fields.iter())
204            .filter(|(key, _)| key.as_str() != SIGNATURE)
205            .map(|(key, value)| (key.clone(), value.clone()))
206            .collect::<BTreeMap<_, _>>();
207        let content = DefaultAuthenticationContextBuilder::combine_request_content(request, &sorted_fields);
208        let signature = acl_signer::cal_signature_with_algorithm(
209            content.as_slice(),
210            self.secret_key.as_str(),
211            self.signature_algorithm,
212        )
213        .map_err(|error| Arc::new(CanonicalError::from(error)))?;
214        request.add_ext_field(SIGNATURE, signature);
215        Ok(())
216    }
217
218    fn do_after_response(
219        &self,
220        _remote_addr: SocketAddr,
221        _request: &RemotingCommand,
222        _response: &mut RemotingCommand,
223    ) -> Result<(), SharedError> {
224        Ok(())
225    }
226}
227
228impl OutboundSigner for AclClientRpcHook {
229    fn sign(&self, request: SecurityRequestView<'_>) -> Result<Signature, SecurityProviderError> {
230        let mut sorted_fields = request
231            .fields()
232            .iter()
233            .filter(|(key, _)| key.as_str() != SIGNATURE)
234            .map(|(key, value)| (key.clone(), value.clone()))
235            .collect::<BTreeMap<_, _>>();
236        sorted_fields.insert(CheetahString::from(ACCESS_KEY), self.access_key.clone());
237        if let Some(security_token) = &self.security_token {
238            sorted_fields.insert(CheetahString::from(SECURITY_TOKEN), security_token.clone());
239        }
240
241        let mut content = Vec::new();
242        for value in sorted_fields.values() {
243            content.extend_from_slice(value.as_bytes());
244        }
245        if let Some(body) = request.body() {
246            content.extend_from_slice(body);
247        }
248        let signature = acl_signer::cal_signature_with_algorithm(
249            content.as_slice(),
250            self.secret_key.as_str(),
251            self.signature_algorithm,
252        )
253        .map_err(|source| {
254            SecurityProviderError::caused_by(
255                SecurityProviderFailure::OperationFailed,
256                SecurityOperation::SignRequest,
257                source,
258            )
259        })?;
260
261        let mut fields = vec![(CheetahString::from(ACCESS_KEY), Secret::new(self.access_key.clone()))];
262        if let Some(security_token) = &self.security_token {
263            fields.push((CheetahString::from(SECURITY_TOKEN), Secret::new(security_token.clone())));
264        }
265        fields.push((
266            CheetahString::from(SIGNATURE),
267            Secret::new(CheetahString::from(signature)),
268        ));
269        Ok(Signature::new(fields))
270    }
271}
272
273#[derive(Deserialize)]
274#[serde(rename_all = "camelCase")]
275struct ClientCredentials {
276    #[serde(alias = "AccessKey", alias = "access_key")]
277    access_key: Option<String>,
278    #[serde(alias = "SecretKey", alias = "secret_key")]
279    secret_key: Option<String>,
280    #[serde(alias = "SecurityToken", alias = "security_token")]
281    security_token: Option<String>,
282}
283
284fn non_blank(value: CheetahString) -> Option<CheetahString> {
285    let value = value.trim();
286    if value.is_empty() {
287        None
288    } else {
289        Some(CheetahString::from(value))
290    }
291}
292
293#[cfg(test)]
294mod tests {
295    use std::collections::HashMap;
296    use std::net::SocketAddr;
297
298    use rocketmq_protocol::protocol::header::empty_header::EmptyHeader;
299
300    use super::*;
301
302    fn remote_addr() -> SocketAddr {
303        "127.0.0.1:9876".parse().unwrap()
304    }
305
306    #[test]
307    fn required_secret_file_rejects_missing_malformed_and_incomplete_credentials() {
308        let directory = tempfile::tempdir().unwrap();
309        let path = directory.path().join("inner-client.json");
310        assert!(AclClientRpcHook::from_required_credentials_file(&path, SignatureAlgorithm::HmacSha1).is_err());
311        for invalid in [
312            "",
313            "not-json",
314            r#"{"accessKey":"inner"}"#,
315            r#"{"accessKey":"inner","secretKey":" "}"#,
316        ] {
317            std::fs::write(&path, invalid).unwrap();
318            let error =
319                AclClientRpcHook::from_required_credentials_file(&path, SignatureAlgorithm::HmacSha1).unwrap_err();
320            assert!(!error.to_string().contains("not-json"));
321        }
322        std::fs::write(&path, r#"{"accessKey":"inner","secretKey":"mounted-secret"}"#).unwrap();
323        let signer = AclClientRpcHook::from_required_credentials_file(&path, SignatureAlgorithm::HmacSha1).unwrap();
324        assert_eq!(signer.access_key().as_str(), "inner");
325        assert!(!format!("{signer:?}").contains("mounted-secret"));
326    }
327
328    #[test]
329    fn from_auth_config_skips_blank_or_incomplete_credentials() {
330        assert!(AclClientRpcHook::from_auth_config(&AuthConfig::default())
331            .unwrap()
332            .is_none());
333        assert!(
334            AclClientRpcHook::from_credentials_json(r#"{"accessKey":"ak"}"#, SignatureAlgorithm::HmacSha1)
335                .unwrap()
336                .is_none()
337        );
338        assert!(AclClientRpcHook::from_credentials_json(
339            r#"{"accessKey":" ","secretKey":"sk"}"#,
340            SignatureAlgorithm::HmacSha1
341        )
342        .unwrap()
343        .is_none());
344    }
345
346    #[test]
347    fn from_auth_config_parses_inner_client_credentials() {
348        let hook = AclClientRpcHook::from_credentials_json(
349            r#"{"accessKey":" inner ","secretKey":" secret ","securityToken":" token "}"#,
350            SignatureAlgorithm::HmacSha1,
351        )
352        .unwrap()
353        .unwrap();
354
355        assert_eq!(hook.access_key().as_str(), "inner");
356        assert_eq!(hook.secret_key().as_str(), "secret");
357        assert_eq!(hook.security_token().map(CheetahString::as_str), Some("token"));
358    }
359
360    #[test]
361    fn from_tools_file_reads_java_tools_credentials() {
362        let temp = tempfile::tempdir().unwrap();
363        let tools_file = temp.path().join("tools.yml");
364        fs::write(&tools_file, "accessKey: tools-ak\nsecretKey: tools-sk\n").unwrap();
365
366        let hook = AclClientRpcHook::from_tools_file(&tools_file, SignatureAlgorithm::HmacSha1)
367            .unwrap()
368            .unwrap();
369
370        assert_eq!(hook.access_key().as_str(), "tools-ak");
371        assert_eq!(hook.secret_key().as_str(), "tools-sk");
372    }
373
374    #[test]
375    fn do_before_request_adds_java_acl_fields_and_signature() {
376        let hook = AclClientRpcHook::new("alice", "secret", Some("token"), SignatureAlgorithm::HmacSha1).unwrap();
377        let mut request =
378            RemotingCommand::create_request_command(10, EmptyHeader {}).set_ext_fields(HashMap::from([(
379                CheetahString::from("topic"),
380                CheetahString::from("topic-a"),
381            )]));
382
383        hook.do_before_request(remote_addr(), &mut request).unwrap();
384
385        let fields = request.ext_fields().unwrap();
386        let expected_signature = acl_signer::cal_signature(b"alicetokentopic-a", "secret").unwrap();
387        assert_eq!(fields.get("AccessKey").map(CheetahString::as_str), Some("alice"));
388        assert_eq!(fields.get("SecurityToken").map(CheetahString::as_str), Some("token"));
389        assert_eq!(
390            fields.get("Signature").map(CheetahString::as_str),
391            Some(expected_signature.as_str())
392        );
393    }
394
395    #[test]
396    fn do_before_request_excludes_existing_signature_when_resigning_like_java() {
397        let hook = AclClientRpcHook::new("alice", "secret", Some("token"), SignatureAlgorithm::HmacSha1).unwrap();
398        let mut request = RemotingCommand::create_request_command(10, EmptyHeader {}).set_ext_fields(HashMap::from([
399            (CheetahString::from("topic"), CheetahString::from("topic-a")),
400            (CheetahString::from(SIGNATURE), CheetahString::from("stale-signature")),
401        ]));
402
403        hook.do_before_request(remote_addr(), &mut request).unwrap();
404
405        let fields = request.ext_fields().unwrap();
406        let expected_signature = acl_signer::cal_signature(b"alicetokentopic-a", "secret").unwrap();
407        assert_eq!(
408            fields.get(SIGNATURE).map(CheetahString::as_str),
409            Some(expected_signature.as_str())
410        );
411    }
412
413    #[test]
414    fn outbound_signer_matches_java_rpc_hook_signature_fields() {
415        let signer = AclClientRpcHook::new("alice", "secret", Some("token"), SignatureAlgorithm::HmacSha1).unwrap();
416        let fields = HashMap::from([(CheetahString::from("topic"), CheetahString::from("topic-a"))]);
417
418        let signature = signer
419            .sign(SecurityRequestView::new(10, 0, &fields, None, None))
420            .unwrap();
421        let signed_fields = signature
422            .fields()
423            .iter()
424            .map(|(key, value)| (key.as_str(), value.expose_secret().as_str()))
425            .collect::<HashMap<_, _>>();
426
427        assert_eq!(signed_fields.get(ACCESS_KEY), Some(&"alice"));
428        assert_eq!(signed_fields.get(SECURITY_TOKEN), Some(&"token"));
429        let expected = acl_signer::cal_signature(b"alicetokentopic-a", "secret").unwrap();
430        assert_eq!(signed_fields.get(SIGNATURE), Some(&expected.as_str()));
431    }
432}
rocketmq_auth/authentication/acl_client_rpc_hook.rs

rocketmq_auth/authentication/
acl_client_rpc_hook.rs
