1use std::collections::HashSet;
2use std::sync::Arc;
3
4use rocketmq_error::RocketMQError;
5use rocketmq_error::RocketMQResult;
6use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
7
8use crate::authentication::builder::default_authentication_context_builder::DefaultAuthenticationContextBuilder;
9use crate::authentication::builder::AuthenticationContextBuilder;
10use crate::authentication::enums::user_status::UserStatus;
11use crate::authentication::enums::user_type::UserType;
12use crate::authentication::model::user::User;
13use crate::authentication::provider::authentication_metadata_provider::AuthenticationMetadataProvider;
14use crate::authentication::provider::AuthenticationProvider;
15use crate::authentication::provider::DefaultAuthenticationProvider;
16use crate::authentication::provider::LocalAuthenticationMetadataProvider;
17use crate::authorization::metadata_provider::AuthorizationMetadataProvider;
18use crate::authorization::metadata_provider::LocalAuthorizationMetadataProvider;
19use crate::authorization::provider::AuthorizationError;
20use crate::authorization::provider::AuthorizationProvider;
21use crate::authorization::provider::DefaultAuthorizationProvider;
22use crate::config::AuthConfig;
23
24#[derive(Clone)]
25pub struct ProviderRegistry {
26 authentication_metadata_provider: Arc<LocalAuthenticationMetadataProvider>,
27 authorization_metadata_provider: Arc<LocalAuthorizationMetadataProvider>,
28}
29
30impl ProviderRegistry {
31 pub fn local(config: &AuthConfig) -> RocketMQResult<Self> {
32 let authentication_metadata_provider = Arc::new(LocalAuthenticationMetadataProvider::new());
33 let mut authorization_metadata_provider = LocalAuthorizationMetadataProvider::new();
34 authorization_metadata_provider
35 .initialize(config.clone(), None)
36 .map_err(map_authorization_error)?;
37
38 Ok(Self {
39 authentication_metadata_provider,
40 authorization_metadata_provider: Arc::new(authorization_metadata_provider),
41 })
42 }
43
44 pub fn authentication_metadata_provider(&self) -> Arc<LocalAuthenticationMetadataProvider> {
45 self.authentication_metadata_provider.clone()
46 }
47
48 pub fn authorization_metadata_provider(&self) -> Arc<LocalAuthorizationMetadataProvider> {
49 self.authorization_metadata_provider.clone()
50 }
51}
52
53pub struct AuthRuntimeBuilder {
54 config: AuthConfig,
55 provider_registry: Option<ProviderRegistry>,
56}
57
58impl AuthRuntimeBuilder {
59 pub fn new(config: AuthConfig) -> Self {
60 Self {
61 config,
62 provider_registry: None,
63 }
64 }
65
66 pub fn with_provider_registry(mut self, provider_registry: ProviderRegistry) -> Self {
67 self.provider_registry = Some(provider_registry);
68 self
69 }
70
71 pub async fn build(self) -> RocketMQResult<AuthRuntime> {
72 let provider_registry = match self.provider_registry {
73 Some(provider_registry) => provider_registry,
74 None => ProviderRegistry::local(&self.config)?,
75 };
76
77 seed_initial_users(&provider_registry, &self.config).await?;
78
79 let mut authentication_provider = DefaultAuthenticationProvider::new();
80 authentication_provider.initialize_with_registry(self.config.clone(), provider_registry.clone())?;
81
82 let mut authorization_provider = DefaultAuthorizationProvider::new();
83 authorization_provider
84 .initialize_with_registry(self.config.clone(), provider_registry.clone())
85 .map_err(map_authorization_error)?;
86
87 let authentication_service = AuthenticationService::new(self.config.clone(), Arc::new(authentication_provider));
88 let authorization_service = AuthorizationService::new(self.config.clone(), Arc::new(authorization_provider));
89
90 Ok(AuthRuntime {
91 config: self.config,
92 provider_registry,
93 authentication_service,
94 authorization_service,
95 })
96 }
97}
98
99#[derive(Clone)]
100pub struct AuthRuntime {
101 config: AuthConfig,
102 provider_registry: ProviderRegistry,
103 authentication_service: AuthenticationService,
104 authorization_service: AuthorizationService,
105}
106
107impl AuthRuntime {
108 pub fn provider_registry(&self) -> &ProviderRegistry {
109 &self.provider_registry
110 }
111
112 pub fn authentication_service(&self) -> &AuthenticationService {
113 &self.authentication_service
114 }
115
116 pub fn authorization_service(&self) -> &AuthorizationService {
117 &self.authorization_service
118 }
119
120 pub fn enabled(&self) -> bool {
121 self.config.authentication_enabled || self.config.authorization_enabled
122 }
123
124 pub async fn check_remoting(
125 &self,
126 channel_context: &(dyn std::any::Any + Send + Sync),
127 command: &RemotingCommand,
128 ) -> RocketMQResult<()> {
129 self.authentication_service.authenticate_remoting(command, None).await?;
130 self.authorization_service
131 .authorize_remoting(channel_context, command)
132 .await
133 }
134}
135
136#[derive(Clone)]
137pub struct AuthenticationService {
138 config: AuthConfig,
139 whitelist: HashSet<String>,
140 provider: Arc<DefaultAuthenticationProvider>,
141 builder: DefaultAuthenticationContextBuilder,
142}
143
144impl AuthenticationService {
145 fn new(config: AuthConfig, provider: Arc<DefaultAuthenticationProvider>) -> Self {
146 Self {
147 whitelist: parse_whitelist(config.authentication_whitelist.as_str()),
148 config,
149 provider,
150 builder: DefaultAuthenticationContextBuilder::new(),
151 }
152 }
153
154 pub async fn authenticate_remoting(
155 &self,
156 command: &RemotingCommand,
157 channel_id: Option<&str>,
158 ) -> RocketMQResult<()> {
159 if !self.config.authentication_enabled || self.whitelist.contains(&command.code().to_string()) {
160 return Ok(());
161 }
162
163 let context = self
164 .builder
165 .build_from_remoting(command, channel_id)
166 .map_err(|error| RocketMQError::authentication_failed(error.to_string()))?;
167 self.provider.authenticate(&context).await
168 }
169}
170
171#[derive(Clone)]
172pub struct AuthorizationService {
173 config: AuthConfig,
174 whitelist: HashSet<String>,
175 provider: Arc<DefaultAuthorizationProvider>,
176}
177
178impl AuthorizationService {
179 fn new(config: AuthConfig, provider: Arc<DefaultAuthorizationProvider>) -> Self {
180 Self {
181 whitelist: parse_whitelist(config.authorization_whitelist.as_str()),
182 config,
183 provider,
184 }
185 }
186
187 pub async fn authorize_remoting(
188 &self,
189 channel_context: &(dyn std::any::Any + Send + Sync),
190 command: &RemotingCommand,
191 ) -> RocketMQResult<()> {
192 if !self.config.authorization_enabled || self.whitelist.contains(&command.code().to_string()) {
193 return Ok(());
194 }
195
196 let contexts = self
197 .provider
198 .new_contexts_from_remoting_command(channel_context, command)
199 .map_err(map_authorization_error)?;
200
201 for context in contexts {
202 self.provider
203 .authorize(&context)
204 .await
205 .map_err(map_authorization_error)?;
206 }
207
208 Ok(())
209 }
210}
211
212fn parse_whitelist(value: &str) -> HashSet<String> {
213 value
214 .split(',')
215 .map(str::trim)
216 .filter(|entry| !entry.is_empty())
217 .map(ToOwned::to_owned)
218 .collect()
219}
220
221async fn seed_initial_users(provider_registry: &ProviderRegistry, config: &AuthConfig) -> RocketMQResult<()> {
222 seed_init_authentication_user(provider_registry.authentication_metadata_provider(), config).await?;
223 seed_inner_client_user(provider_registry.authentication_metadata_provider(), config).await
224}
225
226async fn seed_init_authentication_user(
227 provider: Arc<LocalAuthenticationMetadataProvider>,
228 config: &AuthConfig,
229) -> RocketMQResult<()> {
230 let init_user = config.init_authentication_user.as_str().trim();
231 if init_user.is_empty() {
232 return Ok(());
233 }
234
235 if init_user.starts_with('{') {
236 if let Ok(mut user) = serde_json::from_str::<User>(init_user) {
237 user.set_user_type(UserType::Super);
238 if user.user_status().is_none() {
239 user.set_user_status(UserStatus::Enable);
240 }
241 return create_user_if_absent(provider, user).await;
242 }
243 }
244
245 let parts: Vec<&str> = init_user.splitn(2, ':').collect();
246 if parts.len() == 2 {
247 let username = parts[0].trim();
248 let password = parts[1].trim();
249 if !username.is_empty() && !password.is_empty() {
250 let mut user = User::of_with_type(username, password, UserType::Super);
251 user.set_user_status(UserStatus::Enable);
252 return create_user_if_absent(provider, user).await;
253 }
254 }
255
256 Ok(())
257}
258
259async fn seed_inner_client_user(
260 provider: Arc<LocalAuthenticationMetadataProvider>,
261 config: &AuthConfig,
262) -> RocketMQResult<()> {
263 #[derive(serde::Deserialize)]
264 struct SessionCredentials {
265 #[serde(rename = "accessKey")]
266 access_key: String,
267 #[serde(rename = "secretKey")]
268 secret_key: String,
269 }
270
271 let credentials = config.inner_client_authentication_credentials.as_str().trim();
272 if credentials.is_empty() {
273 return Ok(());
274 }
275
276 let Ok(credentials) = serde_json::from_str::<SessionCredentials>(credentials) else {
277 return Ok(());
278 };
279
280 let mut user = User::of_with_type(credentials.access_key, credentials.secret_key, UserType::Super);
281 user.set_user_status(UserStatus::Enable);
282 create_user_if_absent(provider, user).await
283}
284
285async fn create_user_if_absent(provider: Arc<LocalAuthenticationMetadataProvider>, user: User) -> RocketMQResult<()> {
286 if provider.get_user(user.username().as_str()).await.is_ok() {
287 return Ok(());
288 }
289 provider.create_user(user).await
290}
291
292fn map_authorization_error(error: AuthorizationError) -> RocketMQError {
293 RocketMQError::BrokerPermissionDenied {
294 operation: error.to_string(),
295 }
296}
297
298#[cfg(test)]
299mod tests {
300 use std::collections::HashMap;
301
302 use cheetah_string::CheetahString;
303 use rocketmq_common::common::action::Action;
304 use rocketmq_remoting::code::request_code::RequestCode;
305
306 use super::*;
307 use crate::authentication::acl_signer;
308 use crate::authentication::enums::subject_type::SubjectType;
309 use crate::authorization::enums::decision::Decision;
310 use crate::authorization::model::acl::Acl;
311 use crate::authorization::model::policy::Policy;
312 use crate::authorization::model::resource::Resource;
313
314 fn send_message_command(topic: &str, access_key: &str, signature: &str) -> RemotingCommand {
315 let mut ext_fields = HashMap::new();
316 ext_fields.insert(CheetahString::from_static_str("topic"), CheetahString::from(topic));
317 ext_fields.insert(
318 CheetahString::from_static_str("AccessKey"),
319 CheetahString::from(access_key),
320 );
321 ext_fields.insert(
322 CheetahString::from_static_str("Signature"),
323 CheetahString::from(signature),
324 );
325 RemotingCommand::create_remoting_command(RequestCode::SendMessage.to_i32()).set_ext_fields(ext_fields)
326 }
327
328 #[tokio::test]
329 async fn build_runtime_seeds_initial_super_user() {
330 let runtime = AuthRuntimeBuilder::new(AuthConfig {
331 init_authentication_user: CheetahString::from_static_str("admin:secret"),
332 ..AuthConfig::default()
333 })
334 .build()
335 .await
336 .unwrap();
337
338 let user = runtime
339 .provider_registry()
340 .authentication_metadata_provider()
341 .get_user("admin")
342 .await
343 .unwrap();
344 assert_eq!(user.user_type(), Some(UserType::Super));
345 assert_eq!(user.user_status(), Some(UserStatus::Enable));
346 }
347
348 #[tokio::test]
349 async fn runtime_authenticates_and_authorizes_remoting_command() {
350 let runtime = AuthRuntimeBuilder::new(AuthConfig {
351 authentication_enabled: true,
352 authorization_enabled: true,
353 ..AuthConfig::default()
354 })
355 .build()
356 .await
357 .unwrap();
358
359 let authn_provider = runtime.provider_registry().authentication_metadata_provider();
360 let mut user = User::of_with_type("alice", "secret", UserType::Normal);
361 user.set_user_status(UserStatus::Enable);
362 authn_provider.create_user(user).await.unwrap();
363
364 let authz_provider = runtime.provider_registry().authorization_metadata_provider();
365 let acl = Acl::of(
366 "alice",
367 SubjectType::User,
368 Policy::of(
369 vec![Resource::of_topic("topic-a")],
370 vec![Action::Pub],
371 None,
372 Decision::Allow,
373 ),
374 );
375 authz_provider.create_acl(acl).await.unwrap();
376
377 let signature = acl_signer::cal_signature("AccessKeyalicetopictopic-a".as_bytes(), "secret").unwrap();
378 let command = send_message_command("topic-a", "alice", &signature);
379
380 runtime.check_remoting(&(), &command).await.unwrap();
381 }
382}