This commit is contained in:
houseme
2025-05-19 16:32:56 +08:00
parent be8a615cd7
commit 791780dd68
5 changed files with 69 additions and 40 deletions
+9 -8
View File
@@ -55,7 +55,8 @@ impl Cache {
fn exec<T: Clone>(target: &ArcSwap<CacheEntity<T>>, t: OffsetDateTime, mut op: impl FnMut(&mut CacheEntity<T>)) { fn exec<T: Clone>(target: &ArcSwap<CacheEntity<T>>, t: OffsetDateTime, mut op: impl FnMut(&mut CacheEntity<T>)) {
let mut cur = target.load(); let mut cur = target.load();
loop { loop {
// 当前的更新时间晚于执行时间,说明后台任务加载完毕,不需要执行当前操作。 // If the current update time is later than the execution time,
// the background task is loaded and the current operation does not need to be performed.
if cur.load_time >= t { if cur.load_time >= t {
return; return;
} }
@@ -63,7 +64,7 @@ impl Cache {
let mut new = CacheEntity::clone(&cur); let mut new = CacheEntity::clone(&cur);
op(&mut new); op(&mut new);
// 使用 cas 原子替换内容 // Replace content with CAS atoms
let prev = target.compare_and_swap(&*cur, Arc::new(new)); let prev = target.compare_and_swap(&*cur, Arc::new(new));
let swapped = Self::ptr_eq(&*cur, &*prev); let swapped = Self::ptr_eq(&*cur, &*prev);
if swapped { if swapped {
@@ -88,17 +89,17 @@ impl Cache {
pub fn build_user_group_memberships(&self) { pub fn build_user_group_memberships(&self) {
let groups = self.groups.load(); let groups = self.groups.load();
let mut user_group_memeberships = HashMap::new(); let mut user_group_memberships = HashMap::new();
for (group_name, group) in groups.iter() { for (group_name, group) in groups.iter() {
for user_name in &group.members { for user_name in &group.members {
user_group_memeberships user_group_memberships
.entry(user_name.clone()) .entry(user_name.clone())
.or_insert_with(HashSet::new) .or_insert_with(HashSet::new)
.insert(group_name.clone()); .insert(group_name.clone());
} }
} }
self.user_group_memberships self.user_group_memberships
.store(Arc::new(CacheEntity::new(user_group_memeberships))); .store(Arc::new(CacheEntity::new(user_group_memberships)));
} }
} }
@@ -164,7 +165,7 @@ impl CacheInner {
#[derive(Clone)] #[derive(Clone)]
pub struct CacheEntity<T> { pub struct CacheEntity<T> {
map: HashMap<String, T>, map: HashMap<String, T>,
/// 重新加载的时间 /// The time of the reload
load_time: OffsetDateTime, load_time: OffsetDateTime,
} }
@@ -215,7 +216,7 @@ pub struct CacheInner {
pub sts_accounts: G<UserIdentity>, pub sts_accounts: G<UserIdentity>,
pub sts_policies: G<MappedPolicy>, pub sts_policies: G<MappedPolicy>,
pub groups: G<GroupInfo>, pub groups: G<GroupInfo>,
pub user_group_memeberships: G<HashSet<String>>, pub user_group_memberships: G<HashSet<String>>,
pub group_policies: G<MappedPolicy>, pub group_policies: G<MappedPolicy>,
} }
@@ -228,7 +229,7 @@ impl From<&Cache> for CacheInner {
sts_accounts: value.sts_accounts.load(), sts_accounts: value.sts_accounts.load(),
sts_policies: value.sts_policies.load(), sts_policies: value.sts_policies.load(),
groups: value.groups.load(), groups: value.groups.load(),
user_group_memeberships: value.user_group_memberships.load(), user_group_memberships: value.user_group_memberships.load(),
group_policies: value.group_policies.load(), group_policies: value.group_policies.load(),
} }
} }
+1 -1
View File
@@ -7,7 +7,7 @@ pub enum Error {
#[error(transparent)] #[error(transparent)]
PolicyError(#[from] PolicyError), PolicyError(#[from] PolicyError),
#[error("ecsotre error: {0}")] #[error("ecstore error: {0}")]
EcstoreError(common::error::Error), EcstoreError(common::error::Error),
#[error("{0}")] #[error("{0}")]
+22 -22
View File
@@ -75,7 +75,7 @@ where
T: Store, T: Store,
{ {
pub(crate) async fn new(api: T) -> Arc<Self> { pub(crate) async fn new(api: T) -> Arc<Self> {
let (sender, reciver) = mpsc::channel::<i64>(100); let (sender, receiver) = mpsc::channel::<i64>(100);
let sys = Arc::new(Self { let sys = Arc::new(Self {
api, api,
@@ -86,20 +86,20 @@ where
last_timestamp: AtomicI64::new(0), last_timestamp: AtomicI64::new(0),
}); });
sys.clone().init(reciver).await.unwrap(); sys.clone().init(receiver).await.unwrap();
sys sys
} }
async fn init(self: Arc<Self>, reciver: Receiver<i64>) -> Result<()> { async fn init(self: Arc<Self>, receiver: Receiver<i64>) -> Result<()> {
self.clone().save_iam_formatter().await?; self.clone().save_iam_formatter().await?;
self.clone().load().await?; self.clone().load().await?;
// 后台线程开启定时更新或者接收到信号更新 // The background thread enables scheduled updates or receives signal updates
tokio::spawn({ tokio::spawn({
let s = Arc::clone(&self); let s = Arc::clone(&self);
async move { async move {
let ticker = tokio::time::interval(Duration::from_secs(120)); let ticker = tokio::time::interval(Duration::from_secs(120));
tokio::pin!(ticker, reciver); tokio::pin!(ticker, receiver);
loop { loop {
select! { select! {
_ = ticker.tick() => { _ = ticker.tick() => {
@@ -107,7 +107,7 @@ where
error!("iam load err {:?}", err); error!("iam load err {:?}", err);
} }
}, },
i = reciver.recv() => { i = receiver.recv() => {
match i { match i {
Some(t) => { Some(t) => {
let last = s.last_timestamp.load(Ordering::Relaxed); let last = s.last_timestamp.load(Ordering::Relaxed);
@@ -142,7 +142,7 @@ where
Ok(()) Ok(())
} }
// todo, 判断是否存在,是否可以重试 // todo,Check whether it exists and whether it can be retried
#[tracing::instrument(level = "debug", skip(self))] #[tracing::instrument(level = "debug", skip(self))]
async fn save_iam_formatter(self: Arc<Self>) -> Result<()> { async fn save_iam_formatter(self: Arc<Self>) -> Result<()> {
let path = get_iam_format_file_path(); let path = get_iam_format_file_path();
@@ -798,7 +798,7 @@ where
let mp = MappedPolicy::new(policy); let mp = MappedPolicy::new(policy);
let (_, combined_policy_stmt) = filter_policies(&self.cache, &mp.policies, "temp"); let (_, combined_policy_stmt) = filter_policies(&self.cache, &mp.policies, "temp");
if combined_policy_stmt.is_empty() { if combined_policy_stmt.is_empty() {
return Err(Error::msg(format!("need poliy not found {}", IamError::NoSuchPolicy))); return Err(Error::msg(format!("need policy not found {}", IamError::NoSuchPolicy)));
} }
self.api self.api
@@ -971,7 +971,7 @@ where
_ => auth::ACCOUNT_OFF, _ => auth::ACCOUNT_OFF,
} }
}; };
let user_entiry = UserIdentity::from(Credentials { let user_entity = UserIdentity::from(Credentials {
access_key: access_key.to_string(), access_key: access_key.to_string(),
secret_key: args.secret_key.to_string(), secret_key: args.secret_key.to_string(),
status: status.to_owned(), status: status.to_owned(),
@@ -979,10 +979,10 @@ where
}); });
self.api self.api
.save_user_identity(access_key, UserType::Reg, user_entiry.clone(), None) .save_user_identity(access_key, UserType::Reg, user_entity.clone(), None)
.await?; .await?;
self.update_user_with_claims(access_key, user_entiry)?; self.update_user_with_claims(access_key, user_entity)?;
Ok(OffsetDateTime::now_utc()) Ok(OffsetDateTime::now_utc())
} }
@@ -1088,7 +1088,7 @@ where
} }
}; };
let user_entiry = UserIdentity::from(Credentials { let user_entity = UserIdentity::from(Credentials {
access_key: access_key.to_string(), access_key: access_key.to_string(),
secret_key: u.credentials.secret_key.clone(), secret_key: u.credentials.secret_key.clone(),
status: status.to_owned(), status: status.to_owned(),
@@ -1096,10 +1096,10 @@ where
}); });
self.api self.api
.save_user_identity(access_key, UserType::Reg, user_entiry.clone(), None) .save_user_identity(access_key, UserType::Reg, user_entity.clone(), None)
.await?; .await?;
self.update_user_with_claims(access_key, user_entiry)?; self.update_user_with_claims(access_key, user_entity)?;
Ok(OffsetDateTime::now_utc()) Ok(OffsetDateTime::now_utc())
} }
@@ -1167,9 +1167,9 @@ where
Cache::add_or_update(&self.cache.groups, group, &gi, OffsetDateTime::now_utc()); Cache::add_or_update(&self.cache.groups, group, &gi, OffsetDateTime::now_utc());
let user_group_memeberships = self.cache.user_group_memberships.load(); let user_group_memberships = self.cache.user_group_memberships.load();
members.iter().for_each(|member| { members.iter().for_each(|member| {
if let Some(m) = user_group_memeberships.get(member) { if let Some(m) = user_group_memberships.get(member) {
let mut m = m.clone(); let mut m = m.clone();
m.insert(group.to_string()); m.insert(group.to_string());
Cache::add_or_update(&self.cache.user_group_memberships, member, &m, OffsetDateTime::now_utc()); Cache::add_or_update(&self.cache.user_group_memberships, member, &m, OffsetDateTime::now_utc());
@@ -1252,9 +1252,9 @@ where
Cache::add_or_update(&self.cache.groups, name, &gi, OffsetDateTime::now_utc()); Cache::add_or_update(&self.cache.groups, name, &gi, OffsetDateTime::now_utc());
let user_group_memeberships = self.cache.user_group_memberships.load(); let user_group_memberships = self.cache.user_group_memberships.load();
members.iter().for_each(|member| { members.iter().for_each(|member| {
if let Some(m) = user_group_memeberships.get(member) { if let Some(m) = user_group_memberships.get(member) {
let mut m = m.clone(); let mut m = m.clone();
m.remove(name); m.remove(name);
Cache::add_or_update(&self.cache.user_group_memberships, member, &m, OffsetDateTime::now_utc()); Cache::add_or_update(&self.cache.user_group_memberships, member, &m, OffsetDateTime::now_utc());
@@ -1308,8 +1308,8 @@ where
} }
fn remove_group_from_memberships_map(&self, group: &str) { fn remove_group_from_memberships_map(&self, group: &str) {
let user_group_memeberships = self.cache.user_group_memberships.load(); let user_group_memberships = self.cache.user_group_memberships.load();
for (k, v) in user_group_memeberships.iter() { for (k, v) in user_group_memberships.iter() {
if v.contains(group) { if v.contains(group) {
let mut m = v.clone(); let mut m = v.clone();
m.remove(group); m.remove(group);
@@ -1319,9 +1319,9 @@ where
} }
fn update_group_memberships_map(&self, group: &str, gi: &GroupInfo) { fn update_group_memberships_map(&self, group: &str, gi: &GroupInfo) {
let user_group_memeberships = self.cache.user_group_memberships.load(); let user_group_memberships = self.cache.user_group_memberships.load();
for member in gi.members.iter() { for member in gi.members.iter() {
if let Some(m) = user_group_memeberships.get(member) { if let Some(m) = user_group_memberships.get(member) {
let mut m = m.clone(); let mut m = m.clone();
m.insert(group.to_string()); m.insert(group.to_string());
Cache::add_or_update(&self.cache.user_group_memberships, member, &m, OffsetDateTime::now_utc()); Cache::add_or_update(&self.cache.user_group_memberships, member, &m, OffsetDateTime::now_utc());
+8 -8
View File
@@ -246,7 +246,7 @@ impl<T: Store> IamSys<T> {
// set expiration time default to 1 hour // set expiration time default to 1 hour
m.insert( m.insert(
"exp".to_string(), "exp".to_string(),
serde_json::Value::Number(serde_json::Number::from( Value::Number(serde_json::Number::from(
opts.expiration opts.expiration
.map_or(OffsetDateTime::now_utc().unix_timestamp() + 3600, |t| t.unix_timestamp()), .map_or(OffsetDateTime::now_utc().unix_timestamp() + 3600, |t| t.unix_timestamp()),
)), )),
@@ -282,7 +282,7 @@ impl<T: Store> IamSys<T> {
self.store.list_service_accounts(access_key).await self.store.list_service_accounts(access_key).await
} }
pub async fn list_tmep_accounts(&self, access_key: &str) -> Result<Vec<UserIdentity>> { pub async fn list_temp_accounts(&self, access_key: &str) -> Result<Vec<UserIdentity>> {
self.store.list_temp_accounts(access_key).await self.store.list_temp_accounts(access_key).await
} }
@@ -637,17 +637,17 @@ impl<T: Store> IamSys<T> {
} }
fn is_allowed_by_session_policy(args: &Args<'_>) -> (bool, bool) { fn is_allowed_by_session_policy(args: &Args<'_>) -> (bool, bool) {
let Some(spolicy) = args.claims.get(SESSION_POLICY_NAME_EXTRACTED) else { let Some(policy) = args.claims.get(SESSION_POLICY_NAME_EXTRACTED) else {
return (false, false); return (false, false);
}; };
let has_session_policy = true; let has_session_policy = true;
let Some(spolicy_str) = spolicy.as_str() else { let Some(policy_str) = policy.as_str() else {
return (has_session_policy, false); return (has_session_policy, false);
}; };
let Ok(sub_policy) = Policy::parse_config(spolicy_str.as_bytes()) else { let Ok(sub_policy) = Policy::parse_config(policy_str.as_bytes()) else {
return (has_session_policy, false); return (has_session_policy, false);
}; };
@@ -662,17 +662,17 @@ fn is_allowed_by_session_policy(args: &Args<'_>) -> (bool, bool) {
} }
fn is_allowed_by_session_policy_for_service_account(args: &Args<'_>) -> (bool, bool) { fn is_allowed_by_session_policy_for_service_account(args: &Args<'_>) -> (bool, bool) {
let Some(spolicy) = args.claims.get(SESSION_POLICY_NAME_EXTRACTED) else { let Some(policy) = args.claims.get(SESSION_POLICY_NAME_EXTRACTED) else {
return (false, false); return (false, false);
}; };
let mut has_session_policy = true; let mut has_session_policy = true;
let Some(spolicy_str) = spolicy.as_str() else { let Some(policy_str) = policy.as_str() else {
return (has_session_policy, false); return (has_session_policy, false);
}; };
let Ok(sub_policy) = Policy::parse_config(spolicy_str.as_bytes()) else { let Ok(sub_policy) = Policy::parse_config(policy_str.as_bytes()) else {
return (has_session_policy, false); return (has_session_policy, false);
}; };
+29 -1
View File
@@ -3,6 +3,20 @@ use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header};
use rand::{Rng, RngCore}; use rand::{Rng, RngCore};
use serde::{de::DeserializeOwned, Serialize}; use serde::{de::DeserializeOwned, Serialize};
/// Generates a random access key of the specified length.
///
/// # Arguments
///
/// * `length` - The length of the access key to be generated.
///
/// # Returns
///
/// * `Result<String>` - A result containing the generated access key or an error if the length is invalid.
///
/// # Errors
///
/// * Returns an error if the length is less than 3.
///
pub fn gen_access_key(length: usize) -> Result<String> { pub fn gen_access_key(length: usize) -> Result<String> {
const ALPHA_NUMERIC_TABLE: [char; 36] = [ const ALPHA_NUMERIC_TABLE: [char; 36] = [
'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N',
@@ -23,7 +37,21 @@ pub fn gen_access_key(length: usize) -> Result<String> {
Ok(result) Ok(result)
} }
pub fn gen_secret_key(length: usize) -> crate::Result<String> { /// Generates a random secret key of the specified length.
///
/// # Arguments
///
/// * `length` - The length of the secret key to be generated.
///
/// # Returns
///
/// * `Result<String>` - A result containing the generated secret key or an error if the length is invalid.
///
/// # Errors
///
/// * Returns an error if the length is less than 8.
///
pub fn gen_secret_key(length: usize) -> Result<String> {
use base64_simd::URL_SAFE_NO_PAD; use base64_simd::URL_SAFE_NO_PAD;
if length < 8 { if length < 8 {