use std::sync::Arc;
use tracing::debug;
use crate::authentication::enums::subject_type::SubjectType;
use crate::authentication::model::subject::Subject;
use crate::authentication::provider::AuthenticationMetadataProvider;
use crate::authorization::enums::policy_type::PolicyType;
use crate::authorization::metadata_provider::local::LocalAuthorizationMetadataProvider;
use crate::authorization::metadata_provider::AuthorizationMetadataProvider;
use crate::authorization::model::acl::Acl;
use crate::authorization::model::environment::Environment;
use crate::authorization::model::policy::Policy;
use crate::authorization::model::policy_entry::PolicyEntry;
use crate::authorization::model::resource::Resource;
use crate::authorization::provider::AuthorizationError;
use crate::config::AuthConfig;
use crate::ProviderRegistry;
use rocketmq_error::AuthError;
use rocketmq_error::RocketMQError;
pub type ManagerResult<T> = Result<T, AuthorizationError>;
pub struct AuthorizationMetadataManagerImpl {
authorization_provider: Arc<LocalAuthorizationMetadataProvider>,
authentication_provider: Option<Arc<dyn AuthenticationMetadataProvider>>,
}
impl AuthorizationMetadataManagerImpl {
pub fn new(
authorization_provider: Arc<LocalAuthorizationMetadataProvider>,
authentication_provider: Option<Arc<dyn AuthenticationMetadataProvider>>,
) -> Self {
Self {
authorization_provider,
authentication_provider,
}
}
pub fn from_config(config: &AuthConfig) -> ManagerResult<Self> {
let registry = ProviderRegistry::local(config)
.map_err(|error| AuthorizationError::ConfigurationError(error.to_string()))?;
let authentication_provider: Arc<dyn AuthenticationMetadataProvider> =
registry.authentication_metadata_provider();
Ok(Self::new(
registry.authorization_metadata_provider(),
Some(authentication_provider),
))
}
pub async fn shutdown(&mut self) {
debug!("Shutting down AuthorizationMetadataManagerImpl");
if self.authentication_provider.is_some() {
debug!("Authentication provider is shared; owner is responsible for provider shutdown");
}
debug!("Authorization provider shutdown requested (requires mutable reference)");
debug!("AuthorizationMetadataManagerImpl shutdown complete");
}
pub async fn create_acl(&self, mut acl: Acl) -> ManagerResult<()> {
self.validate_acl(&acl)?;
Self::init_acl(&mut acl);
self.verify_subject_exists(&acl).await?;
let subject_key = acl.subject_key();
let subject_type = acl.subject_type();
let existing_acl = self
.authorization_provider
.get_acl(&SubjectWrapper {
key: subject_key.to_string(),
subject_type,
})
.await?;
match existing_acl {
None => {
debug!("Creating new ACL for subject: {}", subject_key);
self.authorization_provider.create_acl(acl).await
}
Some(mut old_acl) => {
debug!("ACL already exists for subject {}, merging policies", subject_key);
old_acl.update_policies(acl.policies().clone());
self.authorization_provider.update_acl(old_acl).await
}
}
}
pub async fn update_acl(&self, mut acl: Acl) -> ManagerResult<()> {
self.validate_acl(&acl)?;
Self::init_acl(&mut acl);
self.verify_subject_exists(&acl).await?;
let subject_key = acl.subject_key();
let subject_type = acl.subject_type();
let existing_acl = self
.authorization_provider
.get_acl(&SubjectWrapper {
key: subject_key.to_string(),
subject_type,
})
.await?;
match existing_acl {
None => {
debug!("No existing ACL for subject {}, creating new", subject_key);
self.authorization_provider.create_acl(acl).await
}
Some(mut old_acl) => {
debug!("Updating existing ACL for subject: {}", subject_key);
old_acl.update_policies(acl.policies().clone());
self.authorization_provider.update_acl(old_acl).await
}
}
}
pub async fn delete_acl<S: Subject + Send + Sync>(&self, subject: &S) -> ManagerResult<()> {
self.delete_acl_with_filter(subject, None, None).await
}
pub async fn delete_acl_with_filter<S: Subject + Send + Sync>(
&self,
subject: &S,
policy_type: Option<PolicyType>,
resource: Option<&Resource>,
) -> ManagerResult<()> {
self.verify_subject_by_ref(subject).await?;
let policy_type = policy_type.unwrap_or(PolicyType::Custom);
let mut acl = self.authorization_provider.get_acl(subject).await?.ok_or_else(|| {
AuthorizationError::InternalError(format!(
"The ACL for subject '{}' does not exist",
subject.subject_key()
))
})?;
if let Some(resource) = resource {
debug!(
"Deleting resource {:?} from {:?} policy for subject {}",
resource,
policy_type,
subject.subject_key()
);
acl.delete_policy(policy_type, resource);
if acl.policies().is_empty() {
self.authorization_provider.delete_acl(subject).await
} else {
self.authorization_provider.update_acl(acl).await
}
} else {
debug!("Deleting entire ACL for subject: {}", subject.subject_key());
self.authorization_provider.delete_acl(subject).await
}
}
pub async fn get_acl<S: Subject + Send + Sync>(&self, subject: &S) -> ManagerResult<Option<Acl>> {
self.verify_subject_by_ref(subject).await?;
self.authorization_provider.get_acl(subject).await
}
pub async fn list_acl(
&self,
subject_filter: Option<&str>,
resource_filter: Option<&str>,
) -> ManagerResult<Vec<Acl>> {
self.authorization_provider
.list_acl(subject_filter, resource_filter)
.await
}
fn init_acl(_acl: &mut Acl) {}
fn validate_acl(&self, acl: &Acl) -> ManagerResult<()> {
if acl.policies().is_empty() {
return Err(AuthorizationError::InvalidContext("The policies is empty.".to_string()));
}
for policy in acl.policies() {
self.validate_policy(policy)?;
}
Ok(())
}
fn validate_policy(&self, policy: &Policy) -> ManagerResult<()> {
if policy.entries().is_empty() {
return Err(AuthorizationError::InvalidContext(
"The policy entries is empty.".to_string(),
));
}
for entry in policy.entries() {
self.validate_policy_entry(entry)?;
}
Ok(())
}
fn validate_policy_entry(&self, entry: &PolicyEntry) -> ManagerResult<()> {
let resource = entry.resource();
if let Some(name) = resource.resource_name() {
if name.is_empty() {
return Err(AuthorizationError::InvalidContext(
"The resource pattern is empty.".to_string(),
));
}
} else {
return Err(AuthorizationError::InvalidContext(
"The resource pattern is null.".to_string(),
));
}
if entry.actions().is_empty() {
return Err(AuthorizationError::InvalidContext("The actions is empty.".to_string()));
}
use rocketmq_common::common::action::Action;
if entry.actions().contains(&Action::Any) {
return Err(AuthorizationError::InvalidContext(
"The actions can not be Any.".to_string(),
));
}
if let Some(environment) = entry.environment() {
self.validate_environment(environment)?;
}
Ok(())
}
fn validate_environment(&self, environment: &Environment) -> ManagerResult<()> {
let source_ips = environment.source_ips();
if !source_ips.is_empty() {
for source_ip in source_ips {
if source_ip.trim().is_empty() {
return Err(AuthorizationError::InvalidContext(
"The source ip is empty.".to_string(),
));
}
if !Self::is_valid_ip_or_cidr(source_ip) {
return Err(AuthorizationError::InvalidContext(format!(
"The source ip '{}' is invalid.",
source_ip
)));
}
}
}
Ok(())
}
fn is_valid_ip_or_cidr(ip: &str) -> bool {
use std::net::IpAddr;
if ip.parse::<IpAddr>().is_ok() {
return true;
}
if let Some((addr, prefix)) = ip.split_once('/') {
if addr.parse::<IpAddr>().is_ok() {
if let Ok(prefix_len) = prefix.parse::<u8>() {
if addr.contains(':') {
return prefix_len <= 128;
} else {
return prefix_len <= 32;
}
}
}
}
false
}
async fn verify_subject_exists(&self, acl: &Acl) -> ManagerResult<()> {
if acl.subject_type() == SubjectType::User {
self.verify_user_subject_key(acl.subject_key()).await?;
}
Ok(())
}
async fn verify_subject_by_ref<S: Subject + Send + Sync>(&self, subject: &S) -> ManagerResult<()> {
if subject.subject_type() == SubjectType::User {
self.verify_user_subject_key(subject.subject_key()).await?;
}
Ok(())
}
async fn verify_user_subject_key(&self, subject_key: &str) -> ManagerResult<()> {
let provider = self.authentication_provider.as_ref().ok_or_else(|| {
AuthorizationError::NotInitialized("The authenticationMetadataProvider is not configured.".to_string())
})?;
let username = subject_key
.split_once(':')
.map(|(_, username)| username)
.unwrap_or(subject_key)
.trim();
if username.is_empty() {
return Err(AuthorizationError::SubjectNotFound(subject_key.to_string()));
}
match provider.get_user(username).await {
Ok(_) => Ok(()),
Err(RocketMQError::Authentication(AuthError::UserNotFound(_))) => Err(AuthorizationError::SubjectNotFound(
format!("The subject of {subject_key} is not exist."),
)),
Err(error) => Err(AuthorizationError::MetadataServiceError(error.to_string())),
}
}
#[allow(dead_code)]
fn get_authorization_provider(&self) -> &Arc<LocalAuthorizationMetadataProvider> {
&self.authorization_provider
}
#[allow(dead_code)]
fn get_authentication_provider(&self) -> ManagerResult<&Arc<dyn AuthenticationMetadataProvider>> {
self.authentication_provider.as_ref().ok_or_else(|| {
AuthorizationError::NotInitialized("The authenticationMetadataProvider is not configured.".to_string())
})
}
}
struct SubjectWrapper {
key: String,
subject_type: SubjectType,
}
impl Subject for SubjectWrapper {
fn subject_key(&self) -> &str {
&self.key
}
fn subject_type(&self) -> SubjectType {
self.subject_type
}
}
#[cfg(test)]
mod tests {
use rocketmq_common::common::action::Action;
use crate::authentication::enums::user_status::UserStatus;
use crate::authentication::enums::user_type::UserType;
use crate::authentication::model::user::User;
use crate::authentication::provider::LocalAuthenticationMetadataProvider;
use crate::authorization::enums::decision::Decision;
use super::*;
#[test]
fn test_is_valid_ip_or_cidr() {
assert!(AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("192.168.0.1"));
assert!(AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("10.0.0.0"));
assert!(AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("192.168.0.0/24"));
assert!(AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("10.0.0.0/8"));
assert!(AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("2001:db8::1"));
assert!(AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("::1"));
assert!(AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("2001:db8::/32"));
assert!(!AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("invalid"));
assert!(!AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr(
"256.256.256.256"
));
assert!(!AuthorizationMetadataManagerImpl::is_valid_ip_or_cidr("192.168.0.0/33")); }
#[tokio::test]
async fn create_acl_validates_user_subject_with_authentication_provider() {
let mut authorization_provider = LocalAuthorizationMetadataProvider::new();
authorization_provider
.initialize(AuthConfig::default(), None)
.expect("authorization provider should initialize");
let authentication_provider = Arc::new(LocalAuthenticationMetadataProvider::new());
let manager = AuthorizationMetadataManagerImpl::new(
Arc::new(authorization_provider),
Some(authentication_provider.clone() as Arc<dyn AuthenticationMetadataProvider>),
);
let acl = Acl::of(
"alice",
SubjectType::User,
Policy::of(
vec![Resource::of_topic("topic-a")],
vec![Action::Pub],
None,
Decision::Allow,
),
);
let missing_error = manager
.create_acl(acl.clone())
.await
.expect_err("missing user subject should be rejected");
assert!(matches!(missing_error, AuthorizationError::SubjectNotFound(_)));
let mut user = User::of_with_type("alice", "secret", UserType::Normal);
user.set_user_status(UserStatus::Enable);
authentication_provider
.create_user(user)
.await
.expect("user should be created");
manager
.create_acl(acl)
.await
.expect("existing user subject should be accepted");
}
}