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

rocketmq_client_rust/common/
acl.rs
