rocketmq_client_rust/common/
acl.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::collections::BTreeMap;
16
17use rocketmq_protocol::protocol::remoting_command::RemotingCommand;
18
19/// Java-compatible ACL exception type.
20pub type AclException = rocketmq_auth::AuthServiceError;
21
22/// Java-compatible ACL signing algorithm type.
23pub type SigningAlgorithm = rocketmq_auth::SignatureAlgorithm;
24
25/// Java-compatible ACL constants.
26pub struct AclConstants;
27
28impl AclConstants {
29    pub const CONFIG_ACCESS_KEY: &'static str = "accessKey";
30    pub const CONFIG_SECRET_KEY: &'static str = "secretKey";
31    pub const PUB: &'static str = "PUB";
32    pub const SUB: &'static str = "SUB";
33    pub const DENY: &'static str = "DENY";
34    pub const PUB_SUB: &'static str = "PUB|SUB";
35    pub const SUB_PUB: &'static str = "SUB|PUB";
36}
37
38/// Java-compatible ACL permission bits.
39pub struct Permission;
40
41impl Permission {
42    pub const DENY: u8 = 1;
43    pub const ANY: u8 = 1 << 1;
44    pub const PUB: u8 = 1 << 2;
45    pub const SUB: u8 = 1 << 3;
46
47    pub fn parse_perm_from_string(permission: Option<&str>) -> u8 {
48        rocketmq_auth::Permission::parse(permission).bits()
49    }
50}
51
52/// Java-compatible ACL signer facade.
53pub struct AclSigner;
54
55impl AclSigner {
56    pub const DEFAULT_CHARSET: &'static str = rocketmq_auth::DEFAULT_CHARSET;
57    pub const DEFAULT_ALGORITHM: SigningAlgorithm = SigningAlgorithm::HmacSha1;
58
59    pub fn cal_signature(data: &[u8], secret_key: &str) -> Result<String, AclException> {
60        rocketmq_auth::cal_signature(data, secret_key)
61    }
62
63    pub fn cal_signature_str(data: &str, secret_key: &str) -> Result<String, AclException> {
64        Self::cal_signature(data.as_bytes(), secret_key)
65    }
66
67    pub fn cal_signature_with_algorithm(
68        data: &[u8],
69        secret_key: &str,
70        algorithm: SigningAlgorithm,
71    ) -> Result<String, AclException> {
72        rocketmq_auth::cal_signature_with_algorithm(data, secret_key, algorithm)
73    }
74}
75
76/// Java-compatible ACL utility facade.
77pub struct AclUtils;
78
79impl AclUtils {
80    fn net_address_scope_error(net_address: &str) -> AclException {
81        let _ = net_address;
82        AclException::new(
83            rocketmq_auth::AuthOperation::BuildContext,
84            rocketmq_auth::AuthFailureKind::InvalidInput,
85        )
86    }
87
88    pub fn combine_request_content(request: &RemotingCommand, fields: &BTreeMap<String, String>) -> Vec<u8> {
89        let value_len = fields
90            .iter()
91            .filter(|(key, _)| key.as_str() != "Signature")
92            .map(|(_, value)| value.len())
93            .sum::<usize>();
94        let body_len = request.body().map_or(0, |body| body.len());
95        let mut content = Vec::with_capacity(value_len + body_len);
96
97        for (key, value) in fields {
98            if key.as_str() != "Signature" {
99                content.extend_from_slice(value.as_bytes());
100            }
101        }
102        if let Some(body) = request.body() {
103            content.extend_from_slice(body);
104        }
105
106        content
107    }
108
109    pub fn combine_bytes(first: Option<&[u8]>, second: Option<&[u8]>) -> Vec<u8> {
110        match (
111            first.filter(|bytes| !bytes.is_empty()),
112            second.filter(|bytes| !bytes.is_empty()),
113        ) {
114            (None, None) => Vec::new(),
115            (Some(first), None) => first.to_vec(),
116            (None, Some(second)) => second.to_vec(),
117            (Some(first), Some(second)) => {
118                let mut combined = Vec::with_capacity(first.len() + second.len());
119                combined.extend_from_slice(first);
120                combined.extend_from_slice(second);
121                combined
122            }
123        }
124    }
125
126    pub fn cal_signature(data: &[u8], secret_key: &str) -> Result<String, AclException> {
127        AclSigner::cal_signature(data, secret_key)
128    }
129
130    pub fn ipv6_address_check(net_address: &str) -> Result<(), AclException> {
131        if !Self::is_asterisk(net_address) && !Self::is_minus(net_address) {
132            return Ok(());
133        }
134
135        let asterisk = net_address.find('*');
136        let minus = net_address.find('-');
137        if let Some(asterisk) = asterisk {
138            if asterisk != net_address.len() - 1 {
139                return Err(Self::net_address_scope_error(net_address));
140            }
141        }
142
143        let Some(minus) = minus else {
144            return Ok(());
145        };
146        match asterisk {
147            None => {
148                if let Some(last_colon) = net_address.rfind(':') {
149                    if minus <= last_colon {
150                        return Err(Self::net_address_scope_error(net_address));
151                    }
152                }
153            }
154            Some(_) => {
155                let Some(last_colon) = net_address.rfind(':') else {
156                    return Ok(());
157                };
158                if let Some(second_last_colon) = net_address[..last_colon].rfind(':') {
159                    if minus <= second_last_colon {
160                        return Err(Self::net_address_scope_error(net_address));
161                    }
162                }
163            }
164        }
165
166        Ok(())
167    }
168
169    pub fn v6ip_process(net_address: &str) -> String {
170        let is_asterisk = Self::is_asterisk(net_address);
171        let is_minus = Self::is_minus(net_address);
172        let (sub_address, part) = if is_asterisk && is_minus {
173            let last_colon = net_address.rfind(':').unwrap_or(net_address.len());
174            let second_last_colon = net_address[..last_colon].rfind(':').unwrap_or(0);
175            (&net_address[..second_last_colon], 6)
176        } else if !is_asterisk && !is_minus {
177            (net_address, 8)
178        } else {
179            let last_colon = net_address.rfind(':').unwrap_or(net_address.len());
180            (&net_address[..last_colon], 7)
181        };
182        Self::expand_ip(sub_address, part)
183    }
184
185    pub fn verify(net_address: &str, index: usize) -> Result<(), AclException> {
186        if Self::is_scope_address(net_address, index) {
187            Ok(())
188        } else {
189            Err(Self::net_address_scope_error(net_address))
190        }
191    }
192
193    pub fn is_scope_address(net_address: &str, index: usize) -> bool {
194        if Self::is_colon(net_address) {
195            let net_address = Self::expand_ip(net_address, 8);
196            let parts = net_address
197                .split(':')
198                .filter(|part| !part.is_empty())
199                .collect::<Vec<_>>();
200            return Self::is_ipv6_scope_parts(&parts, index);
201        }
202
203        let parts = net_address
204            .split('.')
205            .filter(|part| !part.is_empty())
206            .collect::<Vec<_>>();
207        parts.len() == 4 && Self::is_scope_parts(&parts, index)
208    }
209
210    pub fn is_scope_parts(parts: &[&str], index: usize) -> bool {
211        parts
212            .iter()
213            .take(index)
214            .all(|part| part.trim().parse::<i32>().is_ok_and(Self::is_scope_value))
215    }
216
217    pub fn is_colon(net_address: &str) -> bool {
218        net_address.contains(':')
219    }
220
221    pub fn is_scope_value(num: i32) -> bool {
222        (0..=255).contains(&num)
223    }
224
225    pub fn is_asterisk(asterisk: &str) -> bool {
226        asterisk.contains('*')
227    }
228
229    pub fn is_comma(comma: &str) -> bool {
230        comma.contains(',')
231    }
232
233    pub fn is_minus(minus: &str) -> bool {
234        minus.contains('-')
235    }
236
237    pub fn is_ipv6_scope_parts(parts: &[&str], index: usize) -> bool {
238        parts.iter().take(index).all(|part| {
239            i32::from_str_radix(part, 16)
240                .map(Self::is_ipv6_scope_value)
241                .unwrap_or(false)
242        })
243    }
244
245    pub fn is_ipv6_scope_value(num: i32) -> bool {
246        (0..=0xffff).contains(&num)
247    }
248
249    pub fn expand_ip(net_address: &str, part: usize) -> String {
250        let mut net_address = net_address.to_uppercase();
251        let separator_count = net_address.matches(':').count();
252        if part > separator_count {
253            let pad_str = ":".repeat(part - separator_count + 1);
254            net_address = net_address.replace("::", &pad_str);
255        }
256
257        net_address
258            .split(':')
259            .map(|part| {
260                if part.len() < 4 {
261                    format!("{part:0>4}")
262                } else {
263                    part.to_string()
264                }
265            })
266            .collect::<Vec<_>>()
267            .join(":")
268    }
269
270    #[allow(non_snake_case)]
271    pub fn IPv6AddressCheck(net_address: &str) -> Result<(), AclException> {
272        Self::ipv6_address_check(net_address)
273    }
274
275    #[allow(non_snake_case)]
276    pub fn v6ipProcess(net_address: &str) -> String {
277        Self::v6ip_process(net_address)
278    }
279
280    #[allow(non_snake_case)]
281    pub fn expandIP(net_address: &str, part: usize) -> String {
282        Self::expand_ip(net_address, part)
283    }
284}
285
286#[cfg(test)]
287mod tests {
288    use bytes::Bytes;
289
290    use super::*;
291
292    #[test]
293    fn acl_constants_match_java_values() {
294        assert_eq!(AclConstants::CONFIG_ACCESS_KEY, "accessKey");
295        assert_eq!(AclConstants::CONFIG_SECRET_KEY, "secretKey");
296        assert_eq!(AclConstants::PUB, "PUB");
297        assert_eq!(AclConstants::SUB, "SUB");
298        assert_eq!(AclConstants::DENY, "DENY");
299        assert_eq!(AclConstants::PUB_SUB, "PUB|SUB");
300        assert_eq!(AclConstants::SUB_PUB, "SUB|PUB");
301    }
302
303    #[test]
304    fn permission_parse_perm_from_string_matches_java_bits() {
305        assert_eq!(Permission::parse_perm_from_string(None), Permission::DENY);
306        assert_eq!(Permission::parse_perm_from_string(Some("PUB")), Permission::PUB);
307        assert_eq!(Permission::parse_perm_from_string(Some("SUB")), Permission::SUB);
308        assert_eq!(
309            Permission::parse_perm_from_string(Some("PUB|SUB")),
310            Permission::PUB | Permission::SUB
311        );
312        assert_eq!(
313            Permission::parse_perm_from_string(Some("SUB|PUB")),
314            Permission::PUB | Permission::SUB
315        );
316        assert_eq!(Permission::parse_perm_from_string(Some("UNKNOWN")), Permission::DENY);
317    }
318
319    #[test]
320    fn acl_signer_matches_java_hmac_sha1_base64() {
321        assert_eq!(
322            AclSigner::cal_signature_str("alicetopic-a", "secret").unwrap(),
323            "3yomro7y0WqWcbV+9tQa4av5w3Q="
324        );
325        assert_eq!(
326            AclSigner::cal_signature_with_algorithm(b"alicetopic-a", "secret", SigningAlgorithm::HmacSha256).unwrap(),
327            "+VoSl/q1CHGyZYyeI+H1C8SK6N+inDouZi6grscuA/Q="
328        );
329        assert_eq!(AclSigner::DEFAULT_ALGORITHM, SigningAlgorithm::HmacSha1);
330        assert_eq!(AclSigner::DEFAULT_CHARSET, "UTF-8");
331    }
332
333    #[test]
334    fn acl_utils_combine_bytes_matches_java_null_and_empty_rules() {
335        assert_eq!(AclUtils::combine_bytes(None, None), b"");
336        assert_eq!(AclUtils::combine_bytes(Some(b""), Some(b"body")), b"body");
337        assert_eq!(AclUtils::combine_bytes(Some(b"head"), Some(b"")), b"head");
338        assert_eq!(AclUtils::combine_bytes(Some(b"head"), Some(b"body")), b"headbody");
339    }
340
341    #[test]
342    fn acl_utils_combine_request_content_uses_sorted_values_and_skips_signature_like_java() {
343        let request = RemotingCommand::create_remoting_command(10).set_body(Bytes::from_static(b"body"));
344        let fields = BTreeMap::from([
345            ("Signature".to_string(), "stale".to_string()),
346            ("a".to_string(), "first".to_string()),
347            ("z".to_string(), "last".to_string()),
348        ]);
349
350        assert_eq!(AclUtils::combine_request_content(&request, &fields), b"firstlastbody");
351    }
352
353    #[test]
354    fn acl_utils_expand_ip_matches_java_padding_rules() {
355        assert_eq!(
356            AclUtils::expand_ip("2::ac5:78", 8),
357            "0002:0000:0000:0000:0000:0000:0AC5:0078"
358        );
359        assert_eq!(AclUtils::expandIP("2::ac5:78", 6), "0002:0000:0000:0000:0AC5:0078");
360        assert_eq!(
361            AclUtils::expand_ip("2001:db8::1", 8),
362            "2001:0DB8:0000:0000:0000:0000:0000:0001"
363        );
364    }
365
366    #[test]
367    fn acl_utils_v6ip_process_matches_java_wildcard_and_range_rules() {
368        assert_eq!(
369            AclUtils::v6ip_process("2::ac5:78:1-200:*"),
370            "0002:0000:0000:0000:0AC5:0078"
371        );
372        assert_eq!(
373            AclUtils::v6ipProcess("2::ac5:78:1-200"),
374            "0002:0000:0000:0000:0000:0AC5:0078"
375        );
376        assert_eq!(
377            AclUtils::v6ip_process("2::ac5:78:*"),
378            "0002:0000:0000:0000:0000:0AC5:0078"
379        );
380    }
381
382    #[test]
383    fn acl_utils_ipv6_address_check_matches_java_scope_rules() {
384        assert!(AclUtils::ipv6_address_check("2::ac5:78:1-200:*").is_ok());
385        assert!(AclUtils::ipv6_address_check("2::ac5:78:1-200").is_ok());
386        assert!(AclUtils::IPv6AddressCheck("2::ac5:78:*").is_ok());
387        assert!(AclUtils::ipv6_address_check("2::ac5:78:*:1-200").is_err());
388        assert!(AclUtils::ipv6_address_check("2::ac5:1-200:78:*").is_err());
389        assert!(AclUtils::ipv6_address_check("2::ac5:1-200:78").is_err());
390    }
391
392    #[test]
393    fn acl_utils_verify_returns_typed_error_for_out_of_scope_addresses() {
394        assert!(AclUtils::verify("192.168.0.1", 4).is_ok());
395        assert!(AclUtils::verify("192.168.256.1", 4).is_err());
396        assert!(AclUtils::verify("2::ac5:78", 8).is_ok());
397        assert!(AclUtils::verify("GG::1", 8).is_err());
398    }
399}
rocketmq_client_rust/common/acl.rs

rocketmq_client_rust/common/
acl.rs
