From 0e9d55271299a283ae7fe0fdc9f59cb6703c24c1 Mon Sep 17 00:00:00 2001 From: weisd Date: Tue, 10 Dec 2024 16:59:27 +0800 Subject: [PATCH] AssumeRoleHandle done --- Cargo.lock | 3 ++ ecstore/src/disk/error.rs | 2 + ecstore/src/error.rs | 1 + ecstore/src/set_disk.rs | 2 + ecstore/src/store.rs | 2 +- ecstore/src/store_api.rs | 1 - ecstore/src/utils/path.rs | 4 +- iam/Cargo.toml | 2 + iam/src/auth/credentials.rs | 54 ++++++++++++++++++++-- iam/src/cache.rs | 2 +- iam/src/error.rs | 9 ++++ iam/src/lib.rs | 36 ++++++++++++++- iam/src/manager.rs | 58 ++++++++++++++++++++---- iam/src/policy.rs | 10 ++++- iam/src/store.rs | 3 +- iam/src/store/object.rs | 62 +++++++++++++++---------- iam/src/utils.rs | 10 ++++- rustfs/Cargo.toml | 1 + rustfs/src/admin/handlers.rs | 87 ++++++++++++++++++++++++++++++++---- rustfs/src/auth.rs | 39 ++++++++++++++++ rustfs/src/main.rs | 10 +++-- scripts/run.sh | 2 +- scripts/test.sh | 0 23 files changed, 341 insertions(+), 59 deletions(-) create mode 100644 rustfs/src/auth.rs mode change 100644 => 100755 scripts/test.sh diff --git a/Cargo.lock b/Cargo.lock index 755756bfd..9d8a9643e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1287,6 +1287,7 @@ dependencies = [ "futures", "ipnetwork", "itertools", + "jsonwebtoken", "log", "rand", "serde", @@ -1296,6 +1297,7 @@ dependencies = [ "thiserror 2.0.3", "time", "tokio", + "tracing", ] [[package]] @@ -2521,6 +2523,7 @@ dependencies = [ "hyper", "hyper-util", "iam", + "jsonwebtoken", "lazy_static", "lock", "log", diff --git a/ecstore/src/disk/error.rs b/ecstore/src/disk/error.rs index 36e49f46c..f3e0ee43b 100644 --- a/ecstore/src/disk/error.rs +++ b/ecstore/src/disk/error.rs @@ -1,5 +1,7 @@ use std::io::{self, ErrorKind}; +use tracing::error; + use crate::{ error::{Error, Result}, quorum::CheckErrorFn, diff --git a/ecstore/src/error.rs b/ecstore/src/error.rs index 786594330..e7912cd8e 100644 --- a/ecstore/src/error.rs +++ b/ecstore/src/error.rs @@ -1,5 +1,6 @@ use std::io; +use tracing::warn; use tracing_error::{SpanTrace, SpanTraceStatus}; use crate::disk::error::{clone_disk_err, DiskError}; diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 85d5790ca..76da71f7b 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -178,6 +178,7 @@ impl SetDisks { } #[tracing::instrument(level = "debug", skip(disks, file_infos))] + #[allow(clippy::type_complexity)] async fn rename_data( disks: &[Option], src_bucket: &str, @@ -1630,6 +1631,7 @@ impl SetDisks { Ok((fi, parts_metadata, online_disks)) } + #[allow(clippy::too_many_arguments)] async fn get_object_with_fileinfo( // &self, bucket: &str, diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 81463aae7..1ad26ca32 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -58,10 +58,10 @@ use std::{ time::Duration, }; use time::OffsetDateTime; +use tokio::select; use tokio::sync::mpsc::Sender; use tokio::sync::{broadcast, mpsc, RwLock}; use tokio::time::{interval, sleep}; -use tokio::{fs, select}; use tracing::{debug, info}; use uuid::Uuid; diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 48a860fee..fc37c3ccc 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -9,7 +9,6 @@ use crate::{ use futures::StreamExt; use http::HeaderMap; use madmin::heal_commands::HealResultItem; -use madmin::info_commands::DiskMetrics; use rmp_serde::Serializer; use s3s::{dto::StreamingBlob, Body}; use serde::{Deserialize, Serialize}; diff --git a/ecstore/src/utils/path.rs b/ecstore/src/utils/path.rs index 654333ec2..cd538bc07 100644 --- a/ecstore/src/utils/path.rs +++ b/ecstore/src/utils/path.rs @@ -222,9 +222,9 @@ pub fn split(path: &str) -> (&str, &str) { (path, "") } -pub fn dir(path: &str) -> &str { +pub fn dir(path: &str) -> String { let (a, _) = split(path); - a + clean(a) } #[cfg(test)] mod tests { diff --git a/iam/Cargo.toml b/iam/Cargo.toml index 8b67c3536..8387981d3 100644 --- a/iam/Cargo.toml +++ b/iam/Cargo.toml @@ -26,6 +26,8 @@ itertools = "0.13.0" futures.workspace = true rand.workspace = true base64-simd = "0.8.0" +jsonwebtoken = "9.3.0" +tracing.workspace = true [dev-dependencies] test-case.workspace = true diff --git a/iam/src/auth/credentials.rs b/iam/src/auth/credentials.rs index ac4590c2a..822f237b1 100644 --- a/iam/src/auth/credentials.rs +++ b/iam/src/auth/credentials.rs @@ -1,13 +1,21 @@ +use crate::policy::{Policy, Validator}; +use crate::service_type::ServiceType; +use crate::{utils, Error}; +use jsonwebtoken::{encode, Algorithm, Header}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; use std::cell::LazyCell; use std::collections::HashMap; use time::format_description::BorrowedFormatItem; -use time::{Date, OffsetDateTime}; +use time::{Date, Duration, OffsetDateTime}; -use crate::policy::{Policy, Validator}; -use crate::service_type::ServiceType; -use crate::{utils, Error}; +const ACCESS_KEY_MIN_LEN: usize = 3; +const ACCESS_KEY_MAX_LEN: usize = 20; +const SECRET_KEY_MIN_LEN: usize = 8; +const SECRET_KEY_MAX_LEN: usize = 40; + +const ACCOUNT_ON: &str = "on"; +const ACCOUNT_OFF: &str = "off"; #[cfg_attr(test, derive(PartialEq, Eq, Debug))] struct CredentialHeader { @@ -102,6 +110,44 @@ impl Credentials { Self::check_key_value(header) } + pub fn get_new_credentials_with_metadata( + claims: &T, + token_secret: &str, + exp: Option, + ) -> crate::Result { + let ak = utils::gen_access_key(20).unwrap_or_default(); + let sk = utils::gen_secret_key(32).unwrap_or_default(); + + Self::create_new_credentials_with_metadata(&ak, &sk, claims, token_secret, exp) + } + + pub fn create_new_credentials_with_metadata( + ak: &str, + sk: &str, + claims: &T, + token_secret: &str, + exp: Option, + ) -> crate::Result { + if ak.len() < ACCESS_KEY_MIN_LEN || ak.len() > ACCESS_KEY_MAX_LEN { + return Err(Error::InvalidAccessKeyLength); + } + + if sk.len() < SECRET_KEY_MIN_LEN || sk.len() > SECRET_KEY_MAX_LEN { + return Err(Error::InvalidAccessKeyLength); + } + + let token = utils::generate_jwt(claims, token_secret).map_err(Error::JWTError)?; + + Ok(Self { + access_key: ak.to_owned(), + secret_key: sk.to_owned(), + session_token: token, + status: ACCOUNT_ON.to_owned(), + expiration: exp.map(|v| OffsetDateTime::now_utc().saturating_add(Duration::seconds(v as i64))), + ..Default::default() + }) + } + pub fn check_key_value(_header: CredentialHeader) -> crate::Result { todo!() } diff --git a/iam/src/cache.rs b/iam/src/cache.rs index c3dbef97c..41a3c711f 100644 --- a/iam/src/cache.rs +++ b/iam/src/cache.rs @@ -89,7 +89,7 @@ impl Cache { impl CacheInner { #[inline] - fn get_user<'a>(&self, user_name: &'a str) -> Option<&UserIdentity> { + pub fn get_user<'a>(&self, user_name: &'a str) -> Option<&UserIdentity> { self.users.get(user_name).or_else(|| self.sts_accounts.get(user_name)) } diff --git a/iam/src/error.rs b/iam/src/error.rs index cc92131d1..70117183e 100644 --- a/iam/src/error.rs +++ b/iam/src/error.rs @@ -28,6 +28,15 @@ pub enum Error { #[error("malformed credential")] ErrCredMalformed, + + #[error("CredNotInitialized")] + CredNotInitialized, + + #[error("invalid key length")] + InvalidAccessKeyLength, + + #[error("jwt err {0}")] + JWTError(jsonwebtoken::errors::Error), } pub type Result = std::result::Result; diff --git a/iam/src/lib.rs b/iam/src/lib.rs index 5d41c7900..419e45105 100644 --- a/iam/src/lib.rs +++ b/iam/src/lib.rs @@ -7,7 +7,7 @@ use std::sync::{Arc, OnceLock}; use store::object::ObjectStore; use time::OffsetDateTime; -mod cache; +pub mod cache; mod format; mod handler; @@ -24,6 +24,38 @@ pub use error::{Error, Result}; static IAM_SYS: OnceLock>> = OnceLock::new(); +static GLOBAL_ACTIVE_CRED: OnceLock = OnceLock::new(); + +pub fn init_global_action_cred(ak: Option, sk: Option) -> Result<()> { + let ak = { + if let Some(k) = ak { + k + } else { + utils::gen_access_key(20).unwrap_or_default() + } + }; + + let sk = { + if let Some(k) = sk { + k + } else { + utils::gen_secret_key(32).unwrap_or_default() + } + }; + + GLOBAL_ACTIVE_CRED + .set(Credentials { + access_key: ak, + secret_key: sk, + ..Default::default() + }) + .map_err(|_e| Error::CredNotInitialized) +} + +pub fn get_global_action_cred() -> Option { + GLOBAL_ACTIVE_CRED.get().cloned() +} + pub async fn init_iam_sys(ecstore: Arc) -> crate::Result<()> { debug!("init iam system"); let s = IamCache::new(ObjectStore::new(ecstore)).await; @@ -33,7 +65,7 @@ pub async fn init_iam_sys(ecstore: Arc) -> crate::Result<()> { #[inline] pub fn get() -> crate::Result>> { - IAM_SYS.get().map(|x| Arc::clone(x)).ok_or(Error::IamSysNotInitialized) + IAM_SYS.get().map(Arc::clone).ok_or(Error::IamSysNotInitialized) } pub async fn is_allowed<'a>(args: Args<'a>) -> crate::Result { diff --git a/iam/src/manager.rs b/iam/src/manager.rs index 84daf8757..a58aa8b15 100644 --- a/iam/src/manager.rs +++ b/iam/src/manager.rs @@ -109,11 +109,12 @@ where } // todo, 判断是否存在，是否可以重试 + #[tracing::instrument(level = "debug", skip(self))] async fn save_iam_formatter(self: Arc) -> crate::Result<()> { match self.api.load_iam_config::(Format::PATH).await { Ok((format, _)) if format.version >= 1 => return Ok(()), Err(Error::EcstoreError(e)) if !ecstore::disk::error::is_err_file_not_found(&e) => { - return Err(Error::EcstoreError(e)) + return Err(Error::EcstoreError(e)); } _ => {} } @@ -127,13 +128,14 @@ where Ok(users .values() .filter_map(|x| { - if !access_key.is_empty() && x.credentials.parent_user.as_str() == access_key { - if x.credentials.is_service_account() { - let mut c = x.credentials.clone(); - c.secret_key = String::new(); - c.session_token = String::new(); - return Some(c); - } + if !access_key.is_empty() + && x.credentials.parent_user.as_str() == access_key + && x.credentials.is_service_account() + { + let mut c = x.credentials.clone(); + c.secret_key = String::new(); + c.session_token = String::new(); + return Some(c); } None @@ -220,4 +222,44 @@ where _ => Ok(None), } } + pub async fn policy_db_get(&self, name: &str, _groups: Option>) -> crate::Result> { + // let user = self.cache.users.load(); + // let Some(u) = user.get(name) else { + // return Err(Error::StringError("no service account".into())); + // }; + + let policies = self.cache.user_policies.load(); + + let user_policies = { + if let Some(p) = policies.get(name) { + p.to_slice() + } else { + Vec::new() + } + }; + + // TODO: groups + + Ok(user_policies) + } + + pub async fn set_temp_user(&self, _access_key: &str, cred: &Credentials, _policy_name: &str) -> crate::Result<()> { + let user_entiry = UserIdentity::from(cred.clone()); + let path = format!( + "config/iam/{}{}/identity.json", + UserType::Sts.prefix(), + user_entiry.credentials.access_key + ); + debug!("save object: {path:?}"); + self.api.save_iam_config(&user_entiry, path).await?; + + Cache::add_or_update( + &self.cache.users, + &user_entiry.credentials.access_key, + &user_entiry, + OffsetDateTime::now_utc(), + ); + + Ok(()) + } } diff --git a/iam/src/policy.rs b/iam/src/policy.rs index 114c56621..b60b017bf 100644 --- a/iam/src/policy.rs +++ b/iam/src/policy.rs @@ -17,7 +17,7 @@ pub use id::ID; pub use policy::{default::DEFAULT_POLICIES, Policy}; pub use resource::ResourceSet; use serde::{Deserialize, Serialize}; -use serde_json::Value; +use serde_json::{to_string, Value}; pub use statement::Statement; use std::collections::HashMap; use time::OffsetDateTime; @@ -37,6 +37,14 @@ impl MappedPolicy { update_at: OffsetDateTime::now_utc(), } } + + pub fn to_slice(&self) -> Vec { + self.policies + .split(",") + .filter(|v| !v.trim().is_empty()) + .map(|v| v.to_string()) + .collect() + } } pub struct GroupInfo { diff --git a/iam/src/store.rs b/iam/src/store.rs index 7944eeaf8..73053bf2d 100644 --- a/iam/src/store.rs +++ b/iam/src/store.rs @@ -8,7 +8,7 @@ use serde::{de::DeserializeOwned, Serialize}; use crate::{ auth::UserIdentity, cache::Cache, - policy::{PolicyDoc, UserType, DEFAULT_POLICIES}, + policy::{MappedPolicy, PolicyDoc, UserType, DEFAULT_POLICIES}, }; #[async_trait::async_trait] @@ -40,4 +40,5 @@ pub trait Store: Clone + Send + Sync + 'static { async fn load_users(&self, user_type: UserType) -> crate::Result>; async fn load_policy_docs(&self) -> crate::Result>; + async fn load_mapped_policy(&self, user_type: UserType, name: &str, is_group: bool) -> crate::Result; } diff --git a/iam/src/store/object.rs b/iam/src/store/object.rs index f884e2472..57c00ec26 100644 --- a/iam/src/store/object.rs +++ b/iam/src/store/object.rs @@ -7,8 +7,9 @@ use ecstore::{ utils::path::dir, }; use futures::future::try_join_all; -use log::debug; +use log::{debug, warn}; use serde::{de::DeserializeOwned, Serialize}; +use tracing::error; use super::Store; use crate::{ @@ -101,14 +102,6 @@ impl ObjectStore { Ok(Some(user)) } - - async fn load_mapped_policy(&self, user_type: UserType, name: &str, _is_group: bool) -> crate::Result { - let (p, _) = self - .load_iam_config::(&format!("{base}{name}.json", base = user_type.prefix(), name = name)) - .await?; - - Ok(p) - } } #[async_trait::async_trait] @@ -139,6 +132,7 @@ impl Store for ObjectStore { )) } + #[tracing::instrument(level = "debug", skip(self, item, path))] async fn save_iam_config(&self, item: Item, path: impl AsRef + Send) -> crate::Result<()> { let data = serde_json::to_vec(&item).map_err(|e| crate::Error::StringError(e.to_string()))?; // let data = crypto::encrypt_data(&[], &data)?; @@ -171,8 +165,8 @@ impl Store for ObjectStore { .await?; if policy_doc.version == 0 { - policy_doc.create_date = object_info.mod_time.clone(); - policy_doc.update_date = object_info.mod_time.clone(); + policy_doc.create_date = object_info.mod_time; + policy_doc.update_date = object_info.mod_time; } result.insert(name.to_str().unwrap().to_owned(), policy_doc); @@ -221,7 +215,7 @@ impl Store for ObjectStore { "policydb/groups/", "service-accounts/", "policydb/sts-users/", - "sts", + "sts/", ], ) .await?; @@ -236,12 +230,13 @@ impl Store for ObjectStore { ); // 一次读取32个元素 - let mut iter = items + let iter = items .iter() .map(|item| item.trim_start_matches("config/iam/")) .map(|item| split_path(item, item.starts_with("policydb/"))) .filter_map(|(list_key, trimmed_item)| { debug!("list_key: {list_key}, trimmed_item: {trimmed_item}"); + if list_key == "format.json" { return None; } @@ -257,13 +252,14 @@ impl Store for ObjectStore { Some(async move { match list_key { "policies/" => { - let name = dir(trimmed_item).trim_end_matches('/'); + let trimmed_item = dir(trimmed_item); + let name = trimmed_item.trim_end_matches('/'); let policy_doc = self.load_policy(name).await?; policy_docs.lock().await.insert(name.to_owned(), policy_doc); } "users/" => { let name = dir(trimmed_item); - if let Some(user) = self.load_user_identity(UserType::Reg, name).await? { + if let Some(user) = self.load_user_identity(UserType::Reg, &name).await? { users.lock().await.insert(name.to_owned(), user); }; } @@ -278,7 +274,8 @@ impl Store for ObjectStore { } } "service-accounts/" => { - let name = dir(trimmed_item).trim_end_matches('/'); + let trimmed_item = dir(trimmed_item); + let name = trimmed_item.trim_end_matches('/'); let Some(user) = self.load_user_identity(UserType::Svc, name).await? else { return Ok(()); }; @@ -301,7 +298,8 @@ impl Store for ObjectStore { } "sts/" => { let name = dir(trimmed_item); - if let Some(user) = self.load_user_identity(UserType::Sts, name).await? { + if let Some(user) = self.load_user_identity(UserType::Sts, &name).await? { + warn!("sts_accounts insert {}, user {:?}", name, &user.credentials.access_key); sts_accounts.lock().await.insert(name.to_owned(), user); }; } @@ -321,7 +319,7 @@ impl Store for ObjectStore { let mut all_futures = Vec::with_capacity(32); - while let Some(f) = iter.next() { + for f in iter { all_futures.push(f); if all_futures.len() == 32 { @@ -334,12 +332,30 @@ impl Store for ObjectStore { try_join_all(all_futures).await?; } - Arc::into_inner(users).map(|x| cache.users.store(Arc::new(x.into_inner().update_load_time()))); - Arc::into_inner(policy_docs).map(|x| cache.policy_docs.store(Arc::new(x.into_inner().update_load_time()))); - Arc::into_inner(user_policies).map(|x| cache.user_policies.store(Arc::new(x.into_inner().update_load_time()))); - Arc::into_inner(sts_policies).map(|x| cache.sts_policies.store(Arc::new(x.into_inner().update_load_time()))); - Arc::into_inner(sts_accounts).map(|x| cache.sts_accounts.store(Arc::new(x.into_inner().update_load_time()))); + if let Some(x) = Arc::into_inner(users) { + cache.users.store(Arc::new(x.into_inner().update_load_time())) + } + + if let Some(x) = Arc::into_inner(policy_docs) { + cache.policy_docs.store(Arc::new(x.into_inner().update_load_time())) + } + if let Some(x) = Arc::into_inner(user_policies) { + cache.user_policies.store(Arc::new(x.into_inner().update_load_time())) + } + if let Some(x) = Arc::into_inner(sts_policies) { + cache.sts_policies.store(Arc::new(x.into_inner().update_load_time())) + } + if let Some(x) = Arc::into_inner(sts_accounts) { + cache.sts_accounts.store(Arc::new(x.into_inner().update_load_time())) + } Ok(()) } + async fn load_mapped_policy(&self, user_type: UserType, name: &str, _is_group: bool) -> crate::Result { + let (p, _) = self + .load_iam_config::(&format!("{base}{name}.json", base = user_type.prefix(), name = name)) + .await?; + + Ok(p) + } } diff --git a/iam/src/utils.rs b/iam/src/utils.rs index 30285d52d..f3cd2260b 100644 --- a/iam/src/utils.rs +++ b/iam/src/utils.rs @@ -1,6 +1,7 @@ -use rand::{Rng, RngCore}; - use crate::Error; +use jsonwebtoken::{encode, Algorithm, EncodingKey, Header}; +use rand::{Rng, RngCore}; +use serde::Serialize; pub fn gen_access_key(length: usize) -> crate::Result { const ALPHA_NUMERIC_TABLE: [char; 36] = [ @@ -39,6 +40,11 @@ pub fn gen_secret_key(length: usize) -> crate::Result { Ok(key_str) } +pub fn generate_jwt(claims: &T, secret: &str) -> Result { + let header = Header::new(Algorithm::HS512); + encode(&header, &claims, &EncodingKey::from_secret(secret.as_bytes())) +} + #[cfg(test)] mod tests { use super::{gen_access_key, gen_secret_key}; diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 82b13aaca..8b2431193 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -67,6 +67,7 @@ atoi = "2.0.0" serde_urlencoded = "0.7.1" crypto = { path = "../crypto" } iam = { path = "../iam" } +jsonwebtoken = "9.3.0" [build-dependencies] prost-build.workspace = true diff --git a/rustfs/src/admin/handlers.rs b/rustfs/src/admin/handlers.rs index 297fff8f7..48039e064 100644 --- a/rustfs/src/admin/handlers.rs +++ b/rustfs/src/admin/handlers.rs @@ -20,6 +20,7 @@ use ecstore::GLOBAL_Endpoints; use futures::{Stream, StreamExt}; use http::Uri; use hyper::StatusCode; +use iam::get_global_action_cred; use madmin::metrics::RealtimeMetrics; use madmin::utils::parse_duration; use matchit::Params; @@ -47,6 +48,9 @@ use tracing::{error, info, warn}; pub mod service_account; pub mod trace; +const ASSUME_ROLE_ACTION: &str = "AssumeRole"; +const ASSUME_ROLE_VERSION: &str = "2011-06-15"; + #[derive(Deserialize, Debug, Default)] #[serde(rename_all = "PascalCase", default)] pub struct AssumeRoleRequest { @@ -85,13 +89,30 @@ pub struct AssumeRoleRequest { // pub parent_user: String, // } +#[derive(Debug, Serialize, Default)] +pub struct STSClaims { + parent: String, + exp: usize, + access_key: String, +} + +fn get_token_signing_key() -> Option { + if let Some(s) = get_global_action_cred() { + Some(s.secret_key.clone()) + } else { + None + } +} + pub struct AssumeRoleHandle {} #[async_trait::async_trait] impl Operation for AssumeRoleHandle { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { warn!("handle AssumeRoleHandle"); - let Some(cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) }; + let Some(user) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) }; + + // TODO: 判断权限, 不允许sts访问 let mut input = req.input; @@ -100,17 +121,67 @@ impl Operation for AssumeRoleHandle { }; let body: AssumeRoleRequest = from_bytes(&bytes).map_err(|_e| s3_error!(InvalidRequest, "get body failed"))?; - warn!("AssumeRole get body {:?}", body); + if body.action.as_str() != ASSUME_ROLE_ACTION { + return Err(s3_error!(InvalidArgument, "not suport action")); + } - let exp = OffsetDateTime::now_utc().saturating_add(Duration::days(1)); + if body.version.as_str() != ASSUME_ROLE_VERSION { + return Err(s3_error!(InvalidArgument, "not suport version")); + } + + warn!("AssumeRole get cred {:?}", &user); + warn!("AssumeRole get body {:?}", &body); + + let Ok(iam_store) = iam::get() else { return Err(s3_error!(InvalidRequest, "iam not init")) }; + + if let Err(_err) = iam_store.policy_db_get(&user.access_key, None).await { + return Err(s3_error!(InvalidArgument, "invalid policy arg")); + } + + let Some(secret) = get_token_signing_key() else { + return Err(s3_error!(InvalidArgument, "sk not init")); + }; + + let exp = { + if body.duration_seconds > 0 { + body.duration_seconds + } else { + 3600 + } + }; + + let mut claims = STSClaims { + parent: user.access_key.clone(), + exp, + ..Default::default() + }; + + let ak = iam::utils::gen_access_key(20).unwrap_or_default(); + let sk = iam::utils::gen_secret_key(32).unwrap_or_default(); + + claims.access_key = ak.clone(); + + let mut cred = match iam::auth::Credentials::create_new_credentials_with_metadata(&ak, &sk, &claims, &secret, Some(exp)) { + Ok(res) => res, + Err(_er) => return Err(s3_error!(InvalidRequest, "")), + }; + + cred.parent_user = user.access_key.clone(); + + if let Err(err) = iam_store.set_temp_user(&cred.access_key, &cred, "").await { + error!("set_temp_user err {:?}", err); + return Err(s3_error!(InternalError, "set_temp_user failed")); + } - // TODO: create tmp access_key let resp = AssumeRoleOutput { credentials: Some(Credentials { access_key_id: cred.access_key, - expiration: Timestamp::from(exp), - secret_access_key: cred.secret_key.expose().to_string(), - session_token: "sdfsdf".to_owned(), + expiration: Timestamp::from( + cred.expiration + .unwrap_or(OffsetDateTime::now_utc().saturating_add(Duration::seconds(3600))), + ), + secret_access_key: cred.secret_key, + session_token: cred.session_token, }), ..Default::default() }; @@ -119,8 +190,6 @@ impl Operation for AssumeRoleHandle { let output = xml::serialize::(&resp).unwrap(); Ok(S3Response::new((StatusCode::OK, Body::from(output)))) - - // return Err(s3_error!(NotImplemented)); } } diff --git a/rustfs/src/auth.rs b/rustfs/src/auth.rs new file mode 100644 index 000000000..f76fdfcbb --- /dev/null +++ b/rustfs/src/auth.rs @@ -0,0 +1,39 @@ +use iam::cache::CacheInner; +use s3s::auth::S3Auth; +use s3s::auth::SecretKey; +use s3s::auth::SimpleAuth; +use s3s::s3_error; +use s3s::S3Result; + +pub struct IAMAuth { + simple_auth: SimpleAuth, +} + +impl IAMAuth { + pub fn new(ak: impl Into, sk: impl Into) -> Self { + let simple_auth = SimpleAuth::from_single(ak, sk); + Self { simple_auth } + } +} + +#[async_trait::async_trait] +impl S3Auth for IAMAuth { + async fn get_secret_key(&self, access_key: &str) -> S3Result { + if access_key.is_empty() { + return Err(s3_error!(NotSignedUp, "Your account is not signed up")); + } + + if let Ok(key) = self.simple_auth.get_secret_key(access_key).await { + return Ok(key); + } + + if let Ok(iam_store) = iam::get() { + let c = CacheInner::from(&iam_store.cache); + if let Some(id) = c.get_user(access_key) { + return Ok(SecretKey::from(id.credentials.secret_key.clone())); + } + } + + Err(s3_error!(NotSignedUp, "Your account is not signed up2")) + } +} diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index e02254249..35c0dfc53 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -1,8 +1,10 @@ mod admin; +mod auth; mod config; mod grpc; mod service; mod storage; +use crate::auth::IAMAuth; use clap::Parser; use common::{ error::{Error, Result}, @@ -26,7 +28,7 @@ use hyper_util::{ }; use iam::init_iam_sys; use protos::proto_gen::node_service::node_service_server::NodeServiceServer; -use s3s::{auth::SimpleAuth, service::S3ServiceBuilder}; +use s3s::service::S3ServiceBuilder; use service::hybrid; use std::{io::IsTerminal, net::SocketAddr, str::FromStr}; use tokio::net::TcpListener; @@ -87,6 +89,7 @@ async fn run(opt: config::Opt) -> Result<()> { debug!("server_address {}", &server_address); + iam::init_global_action_cred(None, None).unwrap(); set_global_rustfs_port(server_port); //监听地址,端口从参数中获取 @@ -133,7 +136,7 @@ async fn run(opt: config::Opt) -> Result<()> { } //显示info信息 info!("authentication is enabled {}, {}", &access_key, &secret_key); - b.set_auth(SimpleAuth::from_single(access_key, secret_key)); + b.set_auth(IAMAuth::new(access_key, secret_key)); b.set_access(store.clone()); @@ -212,6 +215,8 @@ async fn run(opt: config::Opt) -> Result<()> { })?; warn!(" init store success!"); + init_iam_sys(store.clone()).await.unwrap(); + new_global_notification_sys(endpoint_pools.clone()).await.map_err(|err| { error!("new_global_notification_sys faild {:?}", &err); Error::from_string(err.to_string()) @@ -221,7 +226,6 @@ async fn run(opt: config::Opt) -> Result<()> { init_data_scanner().await; // init auto heal init_auto_heal().await; - init_iam_sys(store.clone()).await.unwrap(); info!("server was started"); diff --git a/scripts/run.sh b/scripts/run.sh index dbc40a437..4f1942d12 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -11,7 +11,7 @@ mkdir -p ./target/volume/test{0..4} if [ -z "$RUST_LOG" ]; then - export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug,reader=debug,router=debug" + export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug,iam=debug" fi # export RUSTFS_ERASURE_SET_DRIVE_COUNT=5 diff --git a/scripts/test.sh b/scripts/test.sh old mode 100644 new mode 100755