rocketmq_client_rust/base/
validators.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::HashMap;
16
17use cheetah_string::CheetahString;
18use rocketmq_common::common::config::TopicConfig;
19use rocketmq_common::common::constant::PermName;
20use rocketmq_common::common::message::MessageConst;
21use rocketmq_common::common::message::MessageTrait;
22use rocketmq_common::common::topic::TopicValidator;
23use rocketmq_remoting::code::response_code::ResponseCode;
24
25use crate::producer::default_mq_producer::ProducerConfig;
26
27pub struct Validators;
28
29impl Validators {
30    pub const CHARACTER_MAX_LENGTH: usize = 255;
31    pub const TOPIC_MAX_LENGTH: usize = 127;
32    pub const GROUP_MAX_LENGTH: usize = 120;
33
34    pub fn check_group(group: &str) -> rocketmq_error::RocketMQResult<()> {
35        if group.trim().is_empty() {
36            return Err(mq_client_err!("the specified group is blank"));
37        }
38
39        if group.len() > Self::GROUP_MAX_LENGTH {
40            return Err(mq_client_err!(format!(
41                "the specified group[{}] is longer than group max length: {}.",
42                group,
43                Self::GROUP_MAX_LENGTH
44            )));
45        }
46
47        if TopicValidator::is_topic_or_group_illegal(group) {
48            return Err(mq_client_err!(format!(
49                "the specified group[{}] contains illegal characters, allowing only ^[%|a-zA-Z0-9_-]+$",
50                group
51            )));
52        }
53        Ok(())
54    }
55
56    pub fn check_message<M>(msg: Option<&M>, producer_config: &ProducerConfig) -> rocketmq_error::RocketMQResult<()>
57    where
58        M: MessageTrait,
59    {
60        let Some(msg) = msg else {
61            return Err(mq_client_err!(
62                ResponseCode::MessageIllegal as i32,
63                "the message is null".to_string()
64            ));
65        };
66        Self::check_topic(msg.topic())?;
67        Self::is_not_allowed_send_topic(msg.topic())?;
68
69        let Some(body) = msg.get_body() else {
70            return Err(mq_client_err!(
71                ResponseCode::MessageIllegal as i32,
72                "the message body is null".to_string()
73            ));
74        };
75
76        let length = body.len();
77        if length == 0 {
78            return Err(mq_client_err!(
79                ResponseCode::MessageIllegal as i32,
80                "the message body length is zero".to_string()
81            ));
82        }
83
84        if length > producer_config.max_message_size() as usize {
85            return Err(mq_client_err!(
86                ResponseCode::MessageIllegal as i32,
87                format!(
88                    "the message body size over max value, MAX: {}",
89                    producer_config.max_message_size()
90                )
91            ));
92        }
93
94        let lmq_path = msg.user_property(&CheetahString::from_static_str(
95            MessageConst::PROPERTY_INNER_MULTI_DISPATCH,
96        ));
97        if let Some(value) = lmq_path {
98            if value.contains(std::path::MAIN_SEPARATOR) {
99                return Err(mq_client_err!(
100                    ResponseCode::MessageIllegal as i32,
101                    format!(
102                        "INNER_MULTI_DISPATCH {} can not contains {} character",
103                        value,
104                        std::path::MAIN_SEPARATOR
105                    )
106                ));
107            }
108        }
109
110        Ok(())
111    }
112
113    pub fn check_topic(topic: &str) -> rocketmq_error::RocketMQResult<()> {
114        if topic.trim().is_empty() {
115            return Err(mq_client_err!("The specified topic is blank"));
116        }
117
118        if topic.len() > Self::TOPIC_MAX_LENGTH {
119            return Err(mq_client_err!(format!(
120                "The specified topic is longer than topic max length {}.",
121                Self::TOPIC_MAX_LENGTH
122            )));
123        }
124
125        if TopicValidator::is_topic_or_group_illegal(topic) {
126            return Err(mq_client_err!(format!(
127                "The specified topic[{}] contains illegal characters, allowing only ^[%|a-zA-Z0-9_-]+$",
128                topic
129            )));
130        }
131
132        Ok(())
133    }
134
135    pub fn is_system_topic(topic: &str) -> rocketmq_error::RocketMQResult<()> {
136        if TopicValidator::is_system_topic(topic) {
137            return Err(mq_client_err!(format!(
138                "The topic[{}] is conflict with system topic.",
139                topic
140            )));
141        }
142        Ok(())
143    }
144
145    pub fn is_not_allowed_send_topic(topic: &str) -> rocketmq_error::RocketMQResult<()> {
146        if TopicValidator::is_not_allowed_send_topic(topic) {
147            return Err(mq_client_err!(format!(
148                "Sending message to topic[{}] is forbidden.",
149                topic
150            )));
151        }
152
153        Ok(())
154    }
155
156    pub fn check_topic_config(topic_config: &TopicConfig) -> rocketmq_error::RocketMQResult<()> {
157        if !PermName::is_valid(topic_config.perm) {
158            return Err(mq_client_err!(
159                ResponseCode::NoPermission as i32,
160                format!("topicPermission value: {} is invalid.", topic_config.perm)
161            ));
162        }
163
164        Ok(())
165    }
166
167    pub fn check_broker_config(broker_config: &HashMap<String, String>) -> rocketmq_error::RocketMQResult<()> {
168        if let Some(broker_permission) = broker_config.get("brokerPermission") {
169            if !PermName::is_valid_str(broker_permission) {
170                return Err(mq_client_err!(format!(
171                    "brokerPermission value: {} is invalid.",
172                    broker_permission
173                )));
174            }
175        }
176
177        Ok(())
178    }
179}
180
181#[cfg(test)]
182mod tests {
183    use std::collections::HashMap;
184
185    use rocketmq_common::common::config::TopicConfig;
186
187    use super::*;
188
189    #[test]
190    fn check_group_blank_group() {
191        let result = Validators::check_group("");
192        assert!(result.is_err());
193    }
194
195    #[test]
196    fn check_group_long_group() {
197        let long_group = "a".repeat(256);
198        let result = Validators::check_group(&long_group);
199        assert!(result.is_err());
200    }
201
202    #[test]
203    fn check_group_matches_java_max_length() {
204        let max_group = "a".repeat(Validators::GROUP_MAX_LENGTH);
205        assert!(Validators::check_group(&max_group).is_ok());
206
207        let too_long_group = "a".repeat(Validators::GROUP_MAX_LENGTH + 1);
208        let result = Validators::check_group(&too_long_group);
209        assert!(result.is_err());
210        assert!(result
211            .unwrap_err()
212            .to_string()
213            .contains("longer than group max length: 120."));
214    }
215
216    #[test]
217    fn check_group_illegal_characters() {
218        let result = Validators::check_group("illegal@group");
219        assert!(result.is_err());
220    }
221
222    #[test]
223    fn check_group_valid_group() {
224        let result = Validators::check_group("valid_group");
225        assert!(result.is_ok());
226    }
227
228    #[test]
229    fn check_topic_blank_topic() {
230        let result = Validators::check_topic("");
231        assert!(result.is_err());
232    }
233
234    #[test]
235    fn check_topic_long_topic() {
236        let long_topic = "a".repeat(128);
237        let result = Validators::check_topic(&long_topic);
238        assert!(result.is_err());
239    }
240
241    #[test]
242    fn check_topic_illegal_characters() {
243        let result = Validators::check_topic("illegal@topic");
244        assert!(result.is_err());
245    }
246
247    #[test]
248    fn check_topic_valid_topic() {
249        let result = Validators::check_topic("valid_topic");
250        assert!(result.is_ok());
251    }
252
253    #[test]
254    fn check_topic_config_invalid_permission() {
255        let topic_config = TopicConfig {
256            perm: 999,
257            ..Default::default()
258        };
259        let result = Validators::check_topic_config(&topic_config);
260        assert!(result.is_err());
261    }
262
263    #[test]
264    fn check_topic_config_valid_permission() {
265        let topic_config = TopicConfig {
266            perm: 6,
267            ..Default::default()
268        };
269        let result = Validators::check_topic_config(&topic_config);
270        assert!(result.is_ok());
271    }
272
273    #[test]
274    fn check_broker_config_invalid_permission() {
275        let mut broker_config = HashMap::new();
276        broker_config.insert("brokerPermission".to_string(), "999".to_string());
277        let result = Validators::check_broker_config(&broker_config);
278        assert!(result.is_err());
279    }
280
281    #[test]
282    fn check_broker_config_rejects_non_numeric_permission_without_panic() {
283        let mut broker_config = HashMap::new();
284        broker_config.insert("brokerPermission".to_string(), "abc".to_string());
285        let result = Validators::check_broker_config(&broker_config);
286        assert!(result.is_err());
287    }
288
289    #[test]
290    fn check_broker_config_valid_permission() {
291        let mut broker_config = HashMap::new();
292        broker_config.insert("brokerPermission".to_string(), "6".to_string());
293        let result = Validators::check_broker_config(&broker_config);
294        assert!(result.is_ok());
295    }
296}
rocketmq_client_rust/base/validators.rs

rocketmq_client_rust/base/
validators.rs
