rocketmq_client_rust/admin/default_mq_admin_ext_impl/
security.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 super::*;
16pub(super) fn validate_acl_file_path_for_global_white_addr_config(
17    acl_file_full_path: Option<&CheetahString>,
18) -> crate::ClientResult<()> {
19    if acl_file_full_path.is_some_and(|acl_file_full_path| !acl_file_full_path.is_empty()) {
20        return Err(ClientError::illegal_argument(
21            "acl_file_full_path is not supported by RocketMQ ACL 2.0 global white address updates",
22        ));
23    }
24    Ok(())
25}
26
27impl DefaultMQAdminExtImpl {
28    pub async fn create_acl_with_acl_info(
29        &self,
30        broker_addr: CheetahString,
31        acl_info: AclInfo,
32    ) -> crate::ClientResult<()> {
33        if acl_info.subject.as_ref().is_none_or(|subject| subject.is_empty()) {
34            return Err(crate::ClientError::illegal_argument("ACL subject is required"));
35        }
36
37        if let Some(ref client_instance) = self.client_instance {
38            client_instance
39                .get_mq_client_api_impl()?
40                .create_acl(broker_addr, &acl_info, self.remoting_timeout_millis()?)
41                .await
42        } else {
43            Err(crate::ClientError::not_started())
44        }
45    }
46
47    pub async fn update_acl_with_acl_info(
48        &self,
49        broker_addr: CheetahString,
50        acl_info: AclInfo,
51    ) -> crate::ClientResult<()> {
52        if acl_info.subject.as_ref().is_none_or(|subject| subject.is_empty()) {
53            return Err(crate::ClientError::illegal_argument("ACL subject is required"));
54        }
55
56        if let Some(ref client_instance) = self.client_instance {
57            client_instance
58                .get_mq_client_api_impl()?
59                .update_acl(broker_addr, &acl_info, self.remoting_timeout_millis()?)
60                .await
61        } else {
62            Err(crate::ClientError::not_started())
63        }
64    }
65
66    pub async fn create_user_with_user_info(
67        &self,
68        broker_addr: CheetahString,
69        user_info: UserInfo,
70    ) -> crate::ClientResult<()> {
71        let username = user_info
72            .username
73            .clone()
74            .ok_or_else(|| crate::ClientError::illegal_argument("User username is required"))?;
75
76        let password = user_info.password.clone().unwrap_or_default();
77        let user_type = user_info.user_type.clone().unwrap_or_default();
78
79        self.create_user(broker_addr, username, password, user_type).await
80    }
81
82    pub async fn update_user_with_user_info(
83        &self,
84        broker_addr: CheetahString,
85        user_info: UserInfo,
86    ) -> crate::ClientResult<()> {
87        let username = user_info
88            .username
89            .clone()
90            .ok_or_else(|| crate::ClientError::illegal_argument("User username is required"))?;
91
92        let password = user_info.password.clone().unwrap_or_default();
93        let user_type = user_info.user_type.clone().unwrap_or_default();
94        let user_status = user_info.user_status.clone().unwrap_or_default();
95
96        self.update_user(broker_addr, username, password, user_type, user_status)
97            .await
98    }
99}
100
101pub(super) fn build_acl_info(
102    subject: CheetahString,
103    resources: Vec<CheetahString>,
104    actions: Vec<CheetahString>,
105    source_ips: Vec<CheetahString>,
106    decision: CheetahString,
107) -> AclInfo {
108    let entries = if resources.is_empty() {
109        vec![PolicyEntryInfo {
110            resource: None,
111            actions: Some(actions),
112            source_ips: Some(source_ips),
113            decision: if decision.is_empty() { None } else { Some(decision) },
114        }]
115    } else {
116        resources
117            .into_iter()
118            .map(|resource| PolicyEntryInfo {
119                resource: Some(resource),
120                actions: Some(actions.clone()),
121                source_ips: Some(source_ips.clone()),
122                decision: if decision.is_empty() {
123                    None
124                } else {
125                    Some(decision.clone())
126                },
127            })
128            .collect()
129    };
130
131    AclInfo {
132        subject: Some(subject),
133        policies: Some(vec![PolicyInfo {
134            policy_type: Some(CheetahString::from_static_str("Custom")),
135            entries: Some(entries),
136        }]),
137    }
138}
rocketmq_client_rust/admin/default_mq_admin_ext_impl/security.rs

rocketmq_client_rust/admin/default_mq_admin_ext_impl/
security.rs
