use std::sync::Arc;
use tracing::debug;
use tracing::warn;
use crate::authentication::enums::subject_type::SubjectType;
use crate::authentication::model::subject::Subject;
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;
pub type ManagerResult<T> = Result<T, AuthorizationError>;
pub struct AuthorizationMetadataManagerImpl {
authorization_provider: Arc<LocalAuthorizationMetadataProvider>,
authentication_provider: Option<Arc<dyn std::any::Any + Send + Sync>>,
}
impl AuthorizationMetadataManagerImpl {
pub fn new(
authorization_provider: Arc<LocalAuthorizationMetadataProvider>,
authentication_provider: Option<Arc<dyn std::any::Any + Send + Sync>>,
) -> Self {
Self {
authorization_provider,
authentication_provider,
}
}
pub fn from_config(_config: &AuthConfig) -> ManagerResult<Self> {
Err(AuthorizationError::ConfigurationError(
"AuthorizationMetadataManagerImpl::from_config requires factory implementation. Use new() constructor \
with explicit providers instead."
.to_string(),
))
}
pub async fn shutdown(&mut self) {
debug!("Shutting down AuthorizationMetadataManagerImpl");
if self.authentication_provider.is_some() {
debug!("Shutting down authentication provider");
}
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) {
let mut policies = acl.policies().clone();
for policy in policies.iter_mut() {
if policy.policy_type() == PolicyType::Custom {
}
}
acl.set_policies(policies);
}
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 {
warn!(
"Subject validation for USER type is not yet implemented (subject: {})",
acl.subject_key()
);
}
Ok(())
}
async fn verify_subject_by_ref<S: Subject + Send + Sync>(&self, subject: &S) -> ManagerResult<()> {
if subject.subject_type() == SubjectType::User {
warn!(
"Subject validation for USER type is not yet implemented (subject: {})",
subject.subject_key()
);
}
Ok(())
}
#[allow(dead_code)]
fn get_authorization_provider(&self) -> &Arc<LocalAuthorizationMetadataProvider> {
&self.authorization_provider
}
#[allow(dead_code)]
fn get_authentication_provider(&self) -> ManagerResult<&Arc<dyn std::any::Any + Send + Sync>> {
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 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")); }
}