use std::any::Any;
use std::collections::HashMap;
use cheetah_string::CheetahString;
use rocketmq_common::common::action::Action;
use rocketmq_common::common::resource::resource_pattern::ResourcePattern;
use rocketmq_common::common::resource::resource_type::ResourceType;
use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
use rocketmq_remoting::code::request_code::RequestCode;
use rocketmq_remoting::net::channel::Channel;
use rocketmq_remoting::protocol::body::request::lock_batch_request_body::LockBatchRequestBody;
use rocketmq_remoting::protocol::body::unlock_batch_request_body::UnlockBatchRequestBody;
use rocketmq_remoting::protocol::header::get_consumer_listby_group_request_header::GetConsumerListByGroupRequestHeader;
use rocketmq_remoting::protocol::header::query_consumer_offset_request_header::QueryConsumerOffsetRequestHeader;
use rocketmq_remoting::protocol::header::unregister_client_request_header::UnregisterClientRequestHeader;
use rocketmq_remoting::protocol::header::update_consumer_offset_header::UpdateConsumerOffsetRequestHeader;
use rocketmq_remoting::protocol::heartbeat::heartbeat_data::HeartbeatData;
use rocketmq_remoting::protocol::namespace_util::NamespaceUtil;
use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
use rocketmq_remoting::runtime::connection_handler_context::ConnectionHandlerContext;
use rocketmq_remoting::runtime::connection_handler_context::ConnectionHandlerContextWrapper;
use crate::authentication::enums::subject_type::SubjectType;
use crate::authorization::builder::AuthorizationContextBuilder;
use crate::authorization::context::default_authorization_context::DefaultAuthorizationContext;
use crate::authorization::model::resource::Resource;
use crate::authorization::provider::AuthorizationError;
use crate::authorization::provider::AuthorizationResult;
use crate::config::AuthConfig;
const ACCESS_KEY: &str = "AccessKey";
const TOPIC: &str = "topic";
const GROUP: &str = "group";
const CONSUMER_GROUP: &str = "consumerGroup";
const B: &str = "b";
#[derive(Clone, Debug)]
pub struct DefaultAuthorizationContextBuilder {
#[allow(dead_code)]
auth_config: AuthConfig,
}
impl DefaultAuthorizationContextBuilder {
pub fn new(auth_config: AuthConfig) -> Self {
Self { auth_config }
}
fn build_context(
&self,
subject_key: Option<&str>,
resource: Resource,
actions: Vec<Action>,
source_ip: &str,
rpc_code: &str,
) -> DefaultAuthorizationContext {
let mut context = DefaultAuthorizationContext::default();
if let Some(subject_key) = subject_key {
context.set_subject(subject_key, SubjectType::User);
}
context.set_resource(resource);
context.set_actions(actions);
context.set_source_ip(source_ip);
context.set_rpc_code(rpc_code);
context
}
fn field_value<'a>(&self, fields: &'a HashMap<CheetahString, CheetahString>, key: &str) -> Option<&'a str> {
fields
.get(&CheetahString::from(key))
.map(|value| value.as_str())
.filter(|value| !value.is_empty())
}
fn raw_field_value<'a>(&self, fields: &'a HashMap<CheetahString, CheetahString>, key: &str) -> Option<&'a str> {
fields.get(&CheetahString::from(key)).map(|value| value.as_str())
}
fn subject_key(&self, command: &RemotingCommand) -> Option<String> {
command
.ext_fields()
.and_then(|fields| self.field_value(fields, ACCESS_KEY))
.map(|username| format!("{}:{}", SubjectType::User.name(), username))
}
fn source_ip(&self, channel_context: &dyn Any) -> String {
if let Some(ctx) = channel_context.downcast_ref::<ConnectionHandlerContext>() {
return ctx.remote_address().ip().to_string();
}
if let Some(ctx) = channel_context.downcast_ref::<ConnectionHandlerContextWrapper>() {
return ctx.remote_address().ip().to_string();
}
if let Some(channel) = channel_context.downcast_ref::<Channel>() {
return channel.remote_address().ip().to_string();
}
String::from("unknown")
}
fn push_topic_sub_if_not_retry(
&self,
contexts: &mut Vec<DefaultAuthorizationContext>,
subject_key: Option<&str>,
topic: &str,
actions: Vec<Action>,
source_ip: &str,
rpc_code: &str,
) {
if NamespaceUtil::is_retry_topic(topic) {
return;
}
contexts.push(self.build_context(subject_key, Resource::of_topic(topic), actions, source_ip, rpc_code));
}
fn push_java_annotation_resource(
&self,
contexts: &mut Vec<DefaultAuthorizationContext>,
subject_key: Option<&str>,
resource_type: ResourceType,
resource_value: &str,
actions: Vec<Action>,
source_ip: &str,
rpc_code: &str,
) {
let resource = match resource_type {
ResourceType::Cluster => Resource::of_cluster(resource_value),
ResourceType::Topic if NamespaceUtil::is_retry_topic(resource_value) => {
Resource::of_group(resource_value.to_string())
}
ResourceType::Topic => Resource::of(
ResourceType::Topic,
Some(resource_value.to_string()),
ResourcePattern::Literal,
),
ResourceType::Group => Resource::of_group(resource_value.to_string()),
other => Resource::of(other, Some(resource_value.to_string()), ResourcePattern::Literal),
};
contexts.push(self.build_context(subject_key, resource, actions, source_ip, rpc_code));
}
fn push_java_annotation_field(
&self,
contexts: &mut Vec<DefaultAuthorizationContext>,
subject_key: Option<&str>,
fields: &HashMap<CheetahString, CheetahString>,
field_name: &str,
splitter: Option<&str>,
resource_type: ResourceType,
actions: Vec<Action>,
source_ip: &str,
rpc_code: &str,
) {
let Some(value) = self.raw_field_value(fields, field_name) else {
return;
};
let values: Vec<&str> = match splitter {
Some(splitter) => value.split(splitter).filter(|value| !value.is_empty()).collect(),
None => vec![value],
};
for value in values {
self.push_java_annotation_resource(
contexts,
subject_key,
resource_type,
value,
actions.clone(),
source_ip,
rpc_code,
);
}
}
fn push_java_annotation_cluster_default(
&self,
contexts: &mut Vec<DefaultAuthorizationContext>,
subject_key: Option<&str>,
actions: Vec<Action>,
source_ip: &str,
rpc_code: &str,
) {
self.push_java_annotation_resource(
contexts,
subject_key,
ResourceType::Cluster,
self.auth_config.cluster_name.as_str(),
actions,
source_ip,
rpc_code,
);
}
fn build_context_by_java_annotation_mapping(
&self,
contexts: &mut Vec<DefaultAuthorizationContext>,
subject_key: Option<&str>,
fields: &HashMap<CheetahString, CheetahString>,
source_ip: &str,
rpc_code: &str,
request_code: RequestCode,
) {
match request_code {
RequestCode::UpdateAndCreateTopic => self.push_java_annotation_field(
contexts,
subject_key,
fields,
TOPIC,
None,
ResourceType::Topic,
vec![Action::Create],
source_ip,
rpc_code,
),
RequestCode::DeleteTopicInBroker => self.push_java_annotation_field(
contexts,
subject_key,
fields,
TOPIC,
None,
ResourceType::Topic,
vec![Action::Delete],
source_ip,
rpc_code,
),
RequestCode::DeleteSubscriptionGroup => self.push_java_annotation_field(
contexts,
subject_key,
fields,
"groupName",
None,
ResourceType::Group,
vec![Action::Delete],
source_ip,
rpc_code,
),
RequestCode::GetTopicConfig
| RequestCode::GetTopicStatsInfo
| RequestCode::GetMaxOffset
| RequestCode::GetMinOffset
| RequestCode::GetEarliestMsgStoreTime
| RequestCode::ViewMessageById
| RequestCode::SearchOffsetByTimestamp
| RequestCode::QueryTopicConsumeByWho
| RequestCode::CheckRocksdbCqWriteProgress => self.push_java_annotation_field(
contexts,
subject_key,
fields,
TOPIC,
None,
ResourceType::Topic,
vec![Action::Get],
source_ip,
rpc_code,
),
RequestCode::CheckTransactionState => self.push_java_annotation_field(
contexts,
subject_key,
fields,
TOPIC,
None,
ResourceType::Topic,
vec![Action::Pub],
source_ip,
rpc_code,
),
RequestCode::ResumeCheckHalfMessage => self.push_java_annotation_field(
contexts,
subject_key,
fields,
TOPIC,
None,
ResourceType::Topic,
vec![Action::Update],
source_ip,
rpc_code,
),
RequestCode::GetSubscriptionGroupConfig
| RequestCode::GetConsumerConnectionList
| RequestCode::GetConsumerRunningInfo
| RequestCode::QueryTopicsByConsumer => {
let field = match request_code {
RequestCode::GetConsumerConnectionList | RequestCode::GetConsumerRunningInfo => CONSUMER_GROUP,
_ => GROUP,
};
self.push_java_annotation_field(
contexts,
subject_key,
fields,
field,
None,
ResourceType::Group,
vec![Action::Get],
source_ip,
rpc_code,
);
}
RequestCode::UpdateAndGetGroupForbidden => {
self.push_java_annotation_field(
contexts,
subject_key,
fields,
GROUP,
None,
ResourceType::Group,
vec![Action::Update],
source_ip,
rpc_code,
);
self.push_java_annotation_field(
contexts,
subject_key,
fields,
TOPIC,
None,
ResourceType::Topic,
vec![Action::Update],
source_ip,
rpc_code,
);
}
RequestCode::GetConsumeStats => {
self.push_java_annotation_field(
contexts,
subject_key,
fields,
CONSUMER_GROUP,
None,
ResourceType::Group,
vec![Action::Get],
source_ip,
rpc_code,
);
self.push_java_annotation_field(
contexts,
subject_key,
fields,
TOPIC,
None,
ResourceType::Topic,
vec![Action::Get],
source_ip,
rpc_code,
);
self.push_java_annotation_field(
contexts,
subject_key,
fields,
"topicList",
Some(";"),
ResourceType::Topic,
vec![Action::Get],
source_ip,
rpc_code,
);
}
RequestCode::QueryCorrectionOffset => {
self.push_java_annotation_field(
contexts,
subject_key,
fields,
"filterGroups",
Some(","),
ResourceType::Group,
vec![Action::Get],
source_ip,
rpc_code,
);
self.push_java_annotation_field(
contexts,
subject_key,
fields,
"compareGroup",
None,
ResourceType::Group,
vec![Action::Get],
source_ip,
rpc_code,
);
self.push_java_annotation_field(
contexts,
subject_key,
fields,
TOPIC,
None,
ResourceType::Topic,
vec![Action::Get],
source_ip,
rpc_code,
);
}
RequestCode::AuthCreateUser
| RequestCode::AuthUpdateUser
| RequestCode::AuthDeleteUser
| RequestCode::AuthCreateAcl
| RequestCode::AuthUpdateAcl
| RequestCode::AuthDeleteAcl
| RequestCode::AddBroker
| RequestCode::RemoveBroker
| RequestCode::RegisterBroker
| RequestCode::UnregisterBroker
| RequestCode::RegisterTopicInNamesrv
| RequestCode::DeleteTopicInNamesrv
| RequestCode::UpdateBrokerConfig
| RequestCode::PutKvConfig
| RequestCode::DeleteKvConfig
| RequestCode::UpdateNamesrvConfig
| RequestCode::AddWritePermOfBroker
| RequestCode::WipeWritePermOfBroker
| RequestCode::ControllerAlterSyncStateSet
| RequestCode::ControllerElectMaster
| RequestCode::ControllerRegisterBroker
| RequestCode::ControllerApplyBrokerId
| RequestCode::BrokerHeartbeat
| RequestCode::NotifyBrokerRoleChanged
| RequestCode::NotifyMinBrokerIdChange
| RequestCode::ExchangeBrokerHaInfo
| RequestCode::ResetMasterFlushOffset
| RequestCode::CleanBrokerData => self.push_java_annotation_cluster_default(
contexts,
subject_key,
vec![Action::Update],
source_ip,
rpc_code,
),
RequestCode::AuthGetUser
| RequestCode::AuthListUsers
| RequestCode::AuthGetAcl
| RequestCode::AuthListAcl
| RequestCode::GetAllProducerInfo
| RequestCode::GetProducerConnectionList
| RequestCode::GetBrokerConsumeStats
| RequestCode::GetKvConfig
| RequestCode::GetKvlistByNamespace
| RequestCode::QueryDataVersion
| RequestCode::GetBrokerRuntimeInfo
| RequestCode::ViewBrokerStatsData
| RequestCode::ExportRocksdbConfigToJson
| RequestCode::ControllerGetReplicaInfo
| RequestCode::ControllerGetNextBrokerId
| RequestCode::GetBrokerHaStatus => {
self.push_java_annotation_cluster_default(contexts, subject_key, vec![Action::Get], source_ip, rpc_code)
}
_ => {}
}
}
}
impl AuthorizationContextBuilder for DefaultAuthorizationContextBuilder {
fn build_from_remoting(
&self,
channel_context: &(dyn Any + Send + Sync),
command: &RemotingCommand,
) -> AuthorizationResult<Vec<DefaultAuthorizationContext>> {
let mut contexts = Vec::new();
let fields = match command.ext_fields() {
Some(fields) => fields,
None => return Ok(contexts),
};
let subject_key = self.subject_key(command);
let subject_key = subject_key.as_deref();
let source_ip = self.source_ip(channel_context);
let rpc_code = command.code().to_string();
match RequestCode::from(command.code()) {
RequestCode::GetRouteinfoByTopic => {
if let Some(topic) = self.field_value(fields, TOPIC) {
if NamespaceUtil::is_retry_topic(topic) {
contexts.push(self.build_context(
subject_key,
Resource::of_group(topic.to_string()),
vec![Action::Sub, Action::Get],
&source_ip,
&rpc_code,
));
} else {
contexts.push(self.build_context(
subject_key,
Resource::of_topic(topic),
vec![Action::Pub, Action::Sub, Action::Get],
&source_ip,
&rpc_code,
));
}
}
}
RequestCode::SendMessage => {
if let Some(topic) = self.field_value(fields, TOPIC) {
if NamespaceUtil::is_retry_topic(topic) {
contexts.push(self.build_context(
subject_key,
Resource::of_group(topic.to_string()),
vec![Action::Sub],
&source_ip,
&rpc_code,
));
} else {
contexts.push(self.build_context(
subject_key,
Resource::of_topic(topic),
vec![Action::Pub],
&source_ip,
&rpc_code,
));
}
}
}
RequestCode::SendMessageV2 | RequestCode::SendBatchMessage => {
if let Some(topic) = self.field_value(fields, B) {
if NamespaceUtil::is_retry_topic(topic) {
contexts.push(self.build_context(
subject_key,
Resource::of_group(topic.to_string()),
vec![Action::Sub],
&source_ip,
&rpc_code,
));
} else {
contexts.push(self.build_context(
subject_key,
Resource::of_topic(topic),
vec![Action::Pub],
&source_ip,
&rpc_code,
));
}
}
}
RequestCode::RecallMessage => {
if let Some(topic) = self.field_value(fields, TOPIC) {
contexts.push(self.build_context(
subject_key,
Resource::of_topic(topic),
vec![Action::Pub],
&source_ip,
&rpc_code,
));
}
}
RequestCode::EndTransaction => {
if let Some(topic) = self.field_value(fields, TOPIC) {
contexts.push(self.build_context(
subject_key,
Resource::of_topic(topic),
vec![Action::Pub],
&source_ip,
&rpc_code,
));
}
}
RequestCode::ConsumerSendMsgBack => {
if let Some(group) = self.field_value(fields, GROUP) {
contexts.push(self.build_context(
subject_key,
Resource::of_group(group.to_string()),
vec![Action::Sub],
&source_ip,
&rpc_code,
));
}
}
RequestCode::PullMessage => {
if let Some(topic) = self.field_value(fields, TOPIC) {
self.push_topic_sub_if_not_retry(
&mut contexts,
subject_key,
topic,
vec![Action::Sub],
&source_ip,
&rpc_code,
);
}
if let Some(group) = self.field_value(fields, CONSUMER_GROUP) {
contexts.push(self.build_context(
subject_key,
Resource::of_group(group.to_string()),
vec![Action::Sub],
&source_ip,
&rpc_code,
));
}
}
RequestCode::QueryMessage => {
if let Some(topic) = self.field_value(fields, TOPIC) {
contexts.push(self.build_context(
subject_key,
Resource::of_topic(topic),
vec![Action::Sub, Action::Get],
&source_ip,
&rpc_code,
));
}
}
RequestCode::HeartBeat => {
if let Some(body) = command.body() {
let heartbeat = SerdeJsonUtils::from_json_bytes::<HeartbeatData>(body)
.map_err(|error| AuthorizationError::InvalidContext(error.to_string()))?;
for consumer in heartbeat.consumer_data_set {
contexts.push(self.build_context(
subject_key,
Resource::of_group(consumer.group_name.to_string()),
vec![Action::Sub],
&source_ip,
&rpc_code,
));
for subscription in consumer.subscription_data_set {
self.push_topic_sub_if_not_retry(
&mut contexts,
subject_key,
subscription.topic.as_str(),
vec![Action::Sub],
&source_ip,
&rpc_code,
);
}
}
}
}
RequestCode::UnregisterClient => {
let header = command
.decode_command_custom_header::<UnregisterClientRequestHeader>()
.map_err(|error| AuthorizationError::InvalidContext(error.to_string()))?;
if let Some(group) = header.consumer_group.as_deref() {
contexts.push(self.build_context(
subject_key,
Resource::of_group(group.to_string()),
vec![Action::Sub],
&source_ip,
&rpc_code,
));
}
}
RequestCode::GetConsumerListByGroup => {
let header = command
.decode_command_custom_header::<GetConsumerListByGroupRequestHeader>()
.map_err(|error| AuthorizationError::InvalidContext(error.to_string()))?;
contexts.push(self.build_context(
subject_key,
Resource::of_group(header.consumer_group.to_string()),
vec![Action::Sub, Action::Get],
&source_ip,
&rpc_code,
));
}
RequestCode::QueryConsumerOffset => {
let header = command
.decode_command_custom_header::<QueryConsumerOffsetRequestHeader>()
.map_err(|error| AuthorizationError::InvalidContext(error.to_string()))?;
self.push_topic_sub_if_not_retry(
&mut contexts,
subject_key,
header.topic.as_str(),
vec![Action::Sub, Action::Get],
&source_ip,
&rpc_code,
);
contexts.push(self.build_context(
subject_key,
Resource::of_group(header.consumer_group.to_string()),
vec![Action::Sub, Action::Get],
&source_ip,
&rpc_code,
));
}
RequestCode::UpdateConsumerOffset => {
let header = command
.decode_command_custom_header::<UpdateConsumerOffsetRequestHeader>()
.map_err(|error| AuthorizationError::InvalidContext(error.to_string()))?;
self.push_topic_sub_if_not_retry(
&mut contexts,
subject_key,
header.topic.as_str(),
vec![Action::Sub, Action::Update],
&source_ip,
&rpc_code,
);
contexts.push(self.build_context(
subject_key,
Resource::of_group(header.consumer_group.to_string()),
vec![Action::Sub, Action::Update],
&source_ip,
&rpc_code,
));
}
RequestCode::LockBatchMq => {
if let Some(body) = command.body() {
let body = SerdeJsonUtils::from_json_bytes::<LockBatchRequestBody>(body)
.map_err(|error| AuthorizationError::InvalidContext(error.to_string()))?;
if let Some(group) = body.consumer_group.as_deref() {
contexts.push(self.build_context(
subject_key,
Resource::of_group(group.to_string()),
vec![Action::Sub],
&source_ip,
&rpc_code,
));
}
for mq in body.mq_set {
self.push_topic_sub_if_not_retry(
&mut contexts,
subject_key,
mq.topic().as_str(),
vec![Action::Sub],
&source_ip,
&rpc_code,
);
}
}
}
RequestCode::UnlockBatchMq => {
if let Some(body) = command.body() {
let body = serde_json::from_slice::<UnlockBatchRequestBody>(body)
.map_err(|error| AuthorizationError::InvalidContext(error.to_string()))?;
if let Some(group) = body.consumer_group.as_deref() {
contexts.push(self.build_context(
subject_key,
Resource::of_group(group.to_string()),
vec![Action::Sub],
&source_ip,
&rpc_code,
));
}
for mq in body.mq_set {
self.push_topic_sub_if_not_retry(
&mut contexts,
subject_key,
mq.topic().as_str(),
vec![Action::Sub],
&source_ip,
&rpc_code,
);
}
}
}
request_code => self.build_context_by_java_annotation_mapping(
&mut contexts,
subject_key,
fields,
&source_ip,
&rpc_code,
request_code,
),
}
Ok(contexts)
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::collections::HashSet;
use rocketmq_common::common::message::message_queue::MessageQueue;
use rocketmq_remoting::code::request_code::RequestCode;
use super::*;
use crate::config::AuthConfig;
fn command_with_fields(code: RequestCode, fields: &[(&str, &str)]) -> RemotingCommand {
let ext_fields = fields
.iter()
.map(|(key, value)| (CheetahString::from(*key), CheetahString::from(*value)))
.collect::<HashMap<_, _>>();
RemotingCommand::create_remoting_command(code.to_i32()).set_ext_fields(ext_fields)
}
#[test]
fn test_build_send_message_context() {
let builder = DefaultAuthorizationContextBuilder::new(AuthConfig::default());
let command = command_with_fields(
RequestCode::SendMessage,
&[("AccessKey", "alice"), ("topic", "test-topic")],
);
let contexts = builder.build_from_remoting(&(), &command).unwrap();
assert_eq!(contexts.len(), 1);
assert_eq!(contexts[0].subject_key(), Some("User:alice"));
assert_eq!(contexts[0].resource_key(), Some("Topic:test-topic".to_string()));
assert_eq!(contexts[0].actions(), &[Action::Pub]);
}
#[test]
fn test_build_pull_message_contexts() {
let builder = DefaultAuthorizationContextBuilder::new(AuthConfig::default());
let command = command_with_fields(
RequestCode::PullMessage,
&[
("AccessKey", "alice"),
("topic", "test-topic"),
("consumerGroup", "group-a"),
],
);
let contexts = builder.build_from_remoting(&(), &command).unwrap();
assert_eq!(contexts.len(), 2);
assert_eq!(contexts[0].subject_key(), Some("User:alice"));
assert_eq!(contexts[1].resource_key(), Some("Group:group-a".to_string()));
}
#[test]
fn test_build_heartbeat_contexts() {
let builder = DefaultAuthorizationContextBuilder::new(AuthConfig::default());
let mut ext_fields = HashMap::new();
ext_fields.insert(CheetahString::from("AccessKey"), CheetahString::from("alice"));
let mut heartbeat = HeartbeatData::default();
let mut consumer = rocketmq_remoting::protocol::heartbeat::consumer_data::ConsumerData {
group_name: CheetahString::from("group-a"),
..Default::default()
};
let subscription = rocketmq_remoting::protocol::heartbeat::subscription_data::SubscriptionData {
topic: CheetahString::from("topic-a"),
..Default::default()
};
consumer.subscription_data_set.insert(subscription);
heartbeat.consumer_data_set.insert(consumer);
let command = RemotingCommand::create_remoting_command(RequestCode::HeartBeat.to_i32())
.set_ext_fields(ext_fields)
.set_body(serde_json::to_vec(&heartbeat).unwrap());
let contexts = builder.build_from_remoting(&(), &command).unwrap();
assert_eq!(contexts.len(), 2);
assert_eq!(contexts[0].subject_key(), Some("User:alice"));
assert_eq!(contexts[0].resource_key(), Some("Group:group-a".to_string()));
assert_eq!(contexts[1].resource_key(), Some("Topic:topic-a".to_string()));
}
#[test]
fn test_build_lock_batch_contexts() {
let builder = DefaultAuthorizationContextBuilder::new(AuthConfig::default());
let mut ext_fields = HashMap::new();
ext_fields.insert(CheetahString::from("AccessKey"), CheetahString::from("alice"));
let mut mq_set = HashSet::new();
mq_set.insert(MessageQueue::from_parts("topic-a", "broker-a", 0));
let body = LockBatchRequestBody {
consumer_group: Some(CheetahString::from("group-a")),
mq_set,
..Default::default()
};
let command = RemotingCommand::create_remoting_command(RequestCode::LockBatchMq.to_i32())
.set_ext_fields(ext_fields)
.set_body(serde_json::to_vec(&body).unwrap());
let contexts = builder.build_from_remoting(&(), &command).unwrap();
assert_eq!(contexts.len(), 2);
assert_eq!(contexts[0].resource_key(), Some("Group:group-a".to_string()));
assert_eq!(contexts[1].resource_key(), Some("Topic:topic-a".to_string()));
}
#[test]
fn test_build_annotation_fallback_topic_and_cluster_contexts() {
let builder = DefaultAuthorizationContextBuilder::new(AuthConfig {
cluster_name: CheetahString::from_static_str("DefaultCluster"),
..AuthConfig::default()
});
let topic_command = command_with_fields(
RequestCode::UpdateAndCreateTopic,
&[("AccessKey", "rocketmq"), ("topic", "topic")],
);
let topic_contexts = builder.build_from_remoting(&(), &topic_command).unwrap();
assert_eq!(topic_contexts.len(), 1);
assert_eq!(topic_contexts[0].subject_key(), Some("User:rocketmq"));
assert_eq!(topic_contexts[0].resource_key(), Some("Topic:topic".to_string()));
assert_eq!(topic_contexts[0].actions(), &[Action::Create]);
assert_eq!(
topic_contexts[0].rpc_code(),
Some(RequestCode::UpdateAndCreateTopic.to_i32().to_string().as_str())
);
let cluster_command = command_with_fields(
RequestCode::AuthCreateUser,
&[("AccessKey", "rocketmq"), ("username", "alice")],
);
let cluster_contexts = builder.build_from_remoting(&(), &cluster_command).unwrap();
assert_eq!(cluster_contexts.len(), 1);
assert_eq!(cluster_contexts[0].subject_key(), Some("User:rocketmq"));
assert_eq!(
cluster_contexts[0].resource_key(),
Some("Cluster:DefaultCluster".to_string())
);
assert_eq!(cluster_contexts[0].actions(), &[Action::Update]);
assert_eq!(
cluster_contexts[0].rpc_code(),
Some(RequestCode::AuthCreateUser.to_i32().to_string().as_str())
);
}
}