Merge pull request #209 from rustfs/feat/admin-api

feat: add admin user api

crypto/src/encdec/encrypt.rs

@@ -44,6 +44,9 @@ fn encrypt<T: aes_gcm::aead::Aead>(
     use crate::error::Error;
 
     let nonce = T::generate_nonce(rand::thread_rng());
+
+    println!("encrypt nonce len {}, {:?}", nonce.len(), &id);
+
     let encryptor = stream.encrypt(&nonce, data).map_err(Error::ErrEncryptFailed)?;
 
     let mut ciphertext = Vec::with_capacity(salt.len() + 1 + nonce.len() + encryptor.len());

crypto/src/encdec/id.rs

@@ -3,6 +3,7 @@ use pbkdf2::pbkdf2_hmac;
 use sha2::Sha256;
 
 #[repr(u8)]
+#[derive(Debug, Clone, Copy)]
 pub(crate) enum ID {
     Argon2idAESGCM = 0x00,
     Argon2idChaCHa20Poly1305 = 0x01,

ecstore/src/admin_server_info.rs

@@ -340,7 +340,7 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
             )
             .await
             {
-                let data_usage_info = cache.dui(DATA_USAGE_ROOT, &vec![]);
+                let data_usage_info = cache.dui(DATA_USAGE_ROOT, &[]);
                 erasure_set.objects_count = data_usage_info.objects_total_count;
                 erasure_set.versions_count = data_usage_info.versions_total_count;
                 erasure_set.delete_markers_count = data_usage_info.delete_markers_total_count;

ecstore/src/set_disk.rs

@@ -3526,6 +3526,42 @@ impl SetDisks {
         defer.await;
         Ok(())
     }
+
+    async fn delete_prefix(&self, bucket: &str, prefix: &str) -> Result<()> {
+        let disks = self.get_disks_internal().await;
+        let write_quorum = disks.len() / 2 + 1;
+
+        let mut futures = Vec::with_capacity(disks.len());
+
+        for disk_op in disks.iter() {
+            let bucket = bucket.to_string();
+            let prefix = prefix.to_string();
+            futures.push(async move {
+                if let Some(disk) = disk_op {
+                    disk.delete(
+                        &bucket,
+                        &prefix,
+                        DeleteOptions {
+                            recursive: true,
+                            immediate: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                } else {
+                    Ok(())
+                }
+            });
+        }
+
+        let errs = join_all(futures).await.into_iter().map(|v| v.err()).collect::<Vec<_>>();
+
+        if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) {
+            return Err(err);
+        }
+
+        Ok(())
+    }
 }
 
 #[async_trait::async_trait]
@@ -4048,7 +4084,14 @@ impl StorageAPI for SetDisks {
 
         Ok((del_objects, del_errs))
     }
-    async fn delete_object(&self, _bucket: &str, _object: &str, _opts: ObjectOptions) -> Result<ObjectInfo> {
+    async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
+        if opts.delete_prefix {
+            self.delete_prefix(bucket, object)
+                .await
+                .map_err(|e| to_object_err(e, vec![bucket, object]))?;
+
+            return Ok(ObjectInfo::default());
+        }
         unimplemented!()
     }
 

ecstore/src/sets.rs

@@ -500,7 +500,7 @@ impl StorageAPI for Sets {
         Ok((del_objects, del_errs))
     }
     async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
-        if opts.delete_prefix {
+        if opts.delete_prefix && !opts.delete_prefix_object {
             self.delete_prefix(bucket, object).await?;
             return Ok(ObjectInfo::default());
         }

ecstore/src/store.rs

@@ -398,8 +398,20 @@ impl ECStore {
 
         Ok(())
     }
-    async fn delete_prefix(&self, _bucket: &str, _object: &str) -> Result<()> {
-        unimplemented!()
+    async fn delete_prefix(&self, bucket: &str, object: &str) -> Result<()> {
+        for pool in self.pools.iter() {
+            pool.delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    delete_prefix: true,
+                    ..Default::default()
+                },
+            )
+            .await?;
+        }
+
+        Ok(())
     }
 
     async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option<usize> {

ecstore/src/store_api.rs

@@ -579,6 +579,7 @@ pub struct ObjectOptions {
     pub part_number: Option<usize>,
 
     pub delete_prefix: bool,
+    pub delete_prefix_object: bool,
     pub version_id: Option<String>,
     pub no_lock: bool,
 

iam/src/auth/credentials.rs

@@ -18,6 +18,23 @@ const SECRET_KEY_MAX_LEN: usize = 40;
 pub const ACCOUNT_ON: &str = "on";
 pub const ACCOUNT_OFF: &str = "off";
 
+const RESERVED_CHARS: &str = "=,";
+
+// ContainsReservedChars - returns whether the input string contains reserved characters.
+pub fn contains_reserved_chars(s: &str) -> bool {
+    s.contains(RESERVED_CHARS)
+}
+
+// IsAccessKeyValid - validate access key for right length.
+pub fn is_access_key_valid(access_key: &str) -> bool {
+    access_key.len() >= ACCESS_KEY_MIN_LEN
+}
+
+// IsSecretKeyValid - validate secret key for right length.
+pub fn is_secret_key_valid(secret_key: &str) -> bool {
+    secret_key.len() >= SECRET_KEY_MIN_LEN
+}
+
 #[cfg_attr(test, derive(PartialEq, Eq, Debug))]
 struct CredentialHeader {
     access_key: String,
@@ -116,6 +133,10 @@ impl Credentials {
     }
 
     pub fn is_expired(&self) -> bool {
+        if self.expiration.is_none() {
+            return false;
+        }
+
         self.expiration
             .as_ref()
             .map(|e| time::OffsetDateTime::now_utc() > *e)

iam/src/error.rs

@@ -32,9 +32,15 @@ pub enum Error {
     #[error("CredNotInitialized")]
     CredNotInitialized,
 
-    #[error("invalid key length")]
+    #[error("invalid access key length")]
     InvalidAccessKeyLength,
 
+    #[error("invalid secret key length")]
+    InvalidSecretKeyLength,
+
+    #[error("access key contains reserved characters =,")]
+    ContainsReservedChars,
+
     #[error("jwt err {0}")]
     JWTError(jsonwebtoken::errors::Error),
 

iam/src/lib.rs

@@ -1,8 +1,9 @@
-use auth::{Credentials, UserIdentity};
+use auth::{contains_reserved_chars, is_access_key_valid, is_secret_key_valid, Credentials, UserIdentity};
 use ecstore::store::ECStore;
 use log::debug;
+use madmin::AccountStatus;
 use manager::IamCache;
-use policy::{Args, Policy};
+use policy::{Args, Policy, UserType};
 use std::{
     collections::HashMap,
     sync::{Arc, OnceLock},
@@ -71,7 +72,7 @@ pub fn get() -> crate::Result<Arc<IamCache<ObjectStore>>> {
     IAM_SYS.get().map(Arc::clone).ok_or(Error::IamSysNotInitialized)
 }
 
-pub async fn is_allowed<'a>(args: Args<'a>) -> crate::Result<bool> {
+pub async fn is_allowed(args: Args<'_>) -> crate::Result<bool> {
     Ok(get()?.is_allowed(args).await)
 }
 
@@ -106,6 +107,42 @@ pub async fn get_user(ak: &str) -> crate::Result<(Option<UserIdentity>, bool)> {
 }
 
 pub async fn create_user(ak: &str, sk: &str, status: &str) -> crate::Result<OffsetDateTime> {
+    if !is_access_key_valid(ak) {
+        return Err(Error::InvalidAccessKeyLength);
+    }
+
+    if contains_reserved_chars(ak) {
+        return Err(Error::ContainsReservedChars);
+    }
+
+    if !is_secret_key_valid(sk) {
+        return Err(Error::InvalidSecretKeyLength);
+    }
     get()?.add_user(ak, sk, status).await
     // notify
 }
+
+pub async fn delete_user(ak: &str, _notify: bool) -> crate::Result<()> {
+    get()?.delete_user(ak, UserType::Reg).await
+    // TODO NOTIFY
+}
+
+pub async fn is_temp_user(ak: &str) -> crate::Result<(bool, String)> {
+    if let Some(user) = get()?.get_user(ak).await? {
+        Ok((user.credentials.is_temp(), user.credentials.parent_user))
+    } else {
+        Err(Error::NoSuchUser(ak.to_string()))
+    }
+}
+
+pub async fn get_user_info(ak: &str) -> crate::Result<madmin::UserInfo> {
+    get()?.get_user_info(ak).await
+}
+
+pub async fn set_user_status(ak: &str, status: AccountStatus) -> crate::Result<OffsetDateTime> {
+    get()?.set_user_status(ak, status).await
+}
+
+pub async fn list_service_accounts(ak: &str) -> crate::Result<Vec<Credentials>> {
+    get()?.list_service_accounts(ak).await
+}

iam/src/manager.rs

@@ -9,6 +9,7 @@ use std::{
 
 use ecstore::store_err::is_err_object_not_found;
 use log::{debug, warn};
+use madmin::AccountStatus;
 use time::OffsetDateTime;
 use tokio::{
     select,
@@ -295,6 +296,8 @@ where
                 ..Default::default()
             };
 
+            warn!("uinfo {:?}", u);
+
             if let Some(p) = policies.get(k) {
                 u.policy_name = Some(p.policies.clone());
                 u.updated_at = Some(p.update_at);
@@ -313,7 +316,7 @@ where
     pub async fn add_user(&self, access_key: &str, secret_key: &str, status: &str) -> crate::Result<OffsetDateTime> {
         let status = {
             match status {
-                "disabled" => auth::ACCOUNT_ON,
+                val if val == AccountStatus::Enabled.as_ref() => auth::ACCOUNT_ON,
                 auth::ACCOUNT_ON => auth::ACCOUNT_ON,
                 _ => auth::ACCOUNT_OFF,
             }
@@ -321,6 +324,7 @@ where
 
         let users = self.cache.users.load();
         if let Some(x) = users.get(access_key) {
+            warn!("user already exists: {:?}", x);
             if x.credentials.is_temp() {
                 return Err(crate::Error::IAMActionNotAllowed);
             }
@@ -329,7 +333,121 @@ where
         let user_entiry = UserIdentity::from(Credentials {
             access_key: access_key.to_string(),
             secret_key: secret_key.to_string(),
-            status: status.to_string(),
+            status: status.to_owned(),
+            ..Default::default()
+        });
+        let path = format!(
+            "config/iam/{}{}/identity.json",
+            UserType::Reg.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(user_entiry.update_at.unwrap_or(OffsetDateTime::now_utc()))
+    }
+
+    pub async fn delete_user(&self, access_key: &str, utype: UserType) -> crate::Result<()> {
+        let users = self.cache.users.load();
+        if let Some(x) = users.get(access_key) {
+            if x.credentials.is_temp() {
+                return Err(crate::Error::IAMActionNotAllowed);
+            }
+        }
+
+        // if utype == UserType::Reg {}
+
+        let path = format!("config/iam/{}{}/identity.json", UserType::Reg.prefix(), access_key);
+        debug!("delete object: {path:?}");
+        self.api.delete_iam_config(path).await?;
+
+        // delete cache
+        Cache::delete(&self.cache.users, access_key, OffsetDateTime::now_utc());
+
+        Ok(())
+    }
+
+    pub async fn get_user(&self, access_key: &str) -> crate::Result<Option<UserIdentity>> {
+        let u = self
+            .cache
+            .users
+            .load()
+            .get(access_key)
+            .cloned()
+            .or_else(|| self.cache.sts_accounts.load().get(access_key).cloned());
+
+        Ok(u)
+    }
+
+    pub async fn get_user_info(&self, access_key: &str) -> crate::Result<madmin::UserInfo> {
+        let users = self.cache.users.load();
+        let policies = self.cache.user_policies.load();
+        let group_members = self.cache.user_group_memeberships.load();
+
+        let u = match users.get(access_key) {
+            Some(u) => u,
+            None => return Err(Error::NoSuchUser(access_key.to_string())),
+        };
+
+        if u.credentials.is_temp() || u.credentials.is_service_account() {
+            return Err(Error::IAMActionNotAllowed);
+        }
+
+        let mut uinfo = madmin::UserInfo {
+            status: if u.credentials.is_valid() {
+                madmin::AccountStatus::Enabled
+            } else {
+                madmin::AccountStatus::Disabled
+            },
+            updated_at: u.update_at,
+            ..Default::default()
+        };
+
+        if let Some(p) = policies.get(access_key) {
+            uinfo.policy_name = Some(p.policies.clone());
+            uinfo.updated_at = Some(p.update_at);
+        }
+
+        if let Some(members) = group_members.get(access_key) {
+            uinfo.member_of = Some(members.iter().cloned().collect());
+        }
+
+        Ok(uinfo)
+    }
+
+    pub async fn set_user_status(&self, access_key: &str, status: AccountStatus) -> crate::Result<OffsetDateTime> {
+        if access_key.is_empty() {
+            return Err(Error::InvalidArgument);
+        }
+
+        let users = self.cache.users.load();
+        let u = match users.get(access_key) {
+            Some(u) => u,
+            None => return Err(Error::NoSuchUser(access_key.to_string())),
+        };
+
+        if u.credentials.is_temp() || u.credentials.is_service_account() {
+            return Err(Error::IAMActionNotAllowed);
+        }
+
+        let status = {
+            match status {
+                AccountStatus::Enabled => auth::ACCOUNT_ON,
+                _ => auth::ACCOUNT_OFF,
+            }
+        };
+
+        let user_entiry = UserIdentity::from(Credentials {
+            access_key: access_key.to_string(),
+            secret_key: u.credentials.secret_key.clone(),
+            status: status.to_owned(),
             ..Default::default()
         });
         let path = format!(

iam/src/policy.rs

@@ -16,7 +16,7 @@ pub use function::Functions;
 pub use id::ID;
 pub use policy::{default::DEFAULT_POLICIES, Policy};
 pub use resource::ResourceSet;
-use serde::{Deserialize, Serialize};
+use serde::{de, Deserialize, Serialize};
 use serde_json::Value;
 pub use statement::Statement;
 use std::collections::HashMap;
@@ -93,6 +93,7 @@ pub trait Validator {
     }
 }
 
+#[derive(Debug, PartialEq, Eq)]
 pub enum UserType {
     Svc,
     Sts,

iam/src/store.rs

@@ -18,6 +18,7 @@ pub trait Store: Clone + Send + Sync + 'static {
         Item: DeserializeOwned;
 
     async fn save_iam_config<Item: Serialize + Send>(&self, item: Item, path: impl AsRef<str> + Send) -> crate::Result<()>;
+    async fn delete_iam_config(&self, path: impl AsRef<str> + Send) -> crate::Result<()>;
 
     async fn load_all(&self, cache: &Cache) -> crate::Result<()>;
 

iam/src/store/object.rs

@@ -6,6 +6,7 @@ use ecstore::{
     store_api::{ObjectIO, ObjectInfo, ObjectOptions, PutObjReader},
     store_list_objects::{ObjectInfoOrErr, WalkOptions},
     utils::path::{dir, SLASH_SEPARATOR},
+    StorageAPI,
 };
 use futures::future::try_join_all;
 use log::{debug, warn};
@@ -121,6 +122,22 @@ impl ObjectStore {
 
 #[async_trait::async_trait]
 impl Store for ObjectStore {
+    async fn delete_iam_config(&self, path: impl AsRef<str> + Send) -> crate::Result<()> {
+        self.object_api
+            .delete_object(
+                Self::BUCKET_NAME,
+                path.as_ref(),
+                ObjectOptions {
+                    delete_prefix: true,
+                    delete_prefix_object: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .map_err(crate::Error::EcstoreError)?;
+
+        Ok(())
+    }
     async fn load_iam_config<Item>(&self, path: impl AsRef<str> + Send) -> crate::Result<(Item, ObjectInfo)>
     where
         Item: DeserializeOwned,

madmin/src/user.rs

@@ -10,6 +10,27 @@ pub enum AccountStatus {
     Disabled,
 }
 
+impl AsRef<str> for AccountStatus {
+    fn as_ref(&self) -> &str {
+        match self {
+            AccountStatus::Enabled => "enabled",
+            AccountStatus::Disabled => "disabled",
+        }
+    }
+}
+
+impl TryFrom<&str> for AccountStatus {
+    type Error = String;
+
+    fn try_from(s: &str) -> Result<Self, Self::Error> {
+        match s {
+            "enabled" => Ok(AccountStatus::Enabled),
+            "disabled" => Ok(AccountStatus::Disabled),
+            _ => Err(format!("invalid account status: {}", s)),
+        }
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub enum UserAuthType {
     #[serde(rename = "builtin")]
@@ -50,3 +71,69 @@ pub struct UserInfo {
     #[serde(rename = "updatedAt")]
     pub updated_at: Option<OffsetDateTime>,
 }
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AddOrUpdateUserReq {
+    #[serde(rename = "secretKey")]
+    pub secret_key: String,
+
+    #[serde(rename = "policy", skip_serializing_if = "Option::is_none")]
+    pub policy: Option<String>,
+
+    #[serde(rename = "status")]
+    pub status: AccountStatus,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ServiceAccountInfo {
+    #[serde(rename = "parentUser")]
+    pub parent_user: String,
+
+    #[serde(rename = "accountStatus")]
+    pub account_status: String,
+
+    #[serde(rename = "impliedPolicy")]
+    pub implied_policy: bool,
+
+    #[serde(rename = "accessKey")]
+    pub access_key: String,
+
+    #[serde(rename = "name", skip_serializing_if = "Option::is_none")]
+    pub name: Option<String>,
+
+    #[serde(rename = "description", skip_serializing_if = "Option::is_none")]
+    pub description: Option<String>,
+
+    #[serde(rename = "expiration", skip_serializing_if = "Option::is_none")]
+    pub expiration: Option<OffsetDateTime>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ListServiceAccountsResp {
+    #[serde(rename = "accounts")]
+    pub accounts: Vec<ServiceAccountInfo>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AddServiceAccountReq {
+    #[serde(rename = "policy", skip_serializing_if = "Option::is_none")]
+    pub policy: Option<String>,
+
+    #[serde(rename = "targetUser", skip_serializing_if = "Option::is_none")]
+    pub target_user: Option<String>,
+
+    #[serde(rename = "accessKey")]
+    pub access_key: String,
+
+    #[serde(rename = "secretKey")]
+    pub secret_key: String,
+
+    #[serde(rename = "name")]
+    pub name: String,
+
+    #[serde(rename = "description", skip_serializing_if = "Option::is_none")]
+    pub description: Option<String>,
+
+    #[serde(rename = "expiration", skip_serializing_if = "Option::is_none")]
+    pub expiration: Option<OffsetDateTime>,
+}

rustfs/build.rs

@@ -1,3 +1,4 @@
 fn main() -> shadow_rs::SdResult<()> {
-    shadow_rs::new()
+    shadow_rs::ShadowBuilder::builder().build()?;
+    Ok(())
 }

rustfs/src/admin/handlers/service_account.rs

@@ -8,26 +8,32 @@ use iam::{
         Args,
     },
 };
+use madmin::{AddServiceAccountReq, ListServiceAccountsResp, ServiceAccountInfo};
 use matchit::Params;
-use s3s::{s3_error, Body, S3Request, S3Response, S3Result};
+use s3s::{s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
+use serde_urlencoded::from_bytes;
 use tracing::{debug, warn};
 
-use crate::admin::models::service_account::{AddServiceAccountReq, AddServiceAccountResp, Credentials, InfoServiceAccountResp};
 use crate::admin::router::Operation;
+use crate::admin::{
+    handlers::check_key_valid,
+    models::service_account::{AddServiceAccountResp, Credentials, InfoServiceAccountResp},
+};
 
 pub struct AddServiceAccount {}
 #[async_trait::async_trait]
 impl Operation for AddServiceAccount {
-    async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        warn!("handle AddServiceAccount, req: {req:?}");
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        warn!("handle AddServiceAccount ");
 
-        let Some(cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) };
+        let Some(input_cred) = req.credentials else {
+            return Err(s3_error!(InvalidRequest, "get cred failed"));
+        };
         let _is_owner = true; // 先按true处理，后期根据请求决定。
-        let body = req.input.store_all_unlimited().await.unwrap();
-        let body = crypto::decrypt_data(cred.secret_key.expose().as_bytes(), &body[..])
-            .map_err(|_| s3_error!(InternalError, "encrypt data failed"))?;
 
-        debug!("body: {:?}", String::from_utf8_lossy(&body));
+        let Some(body) = req.input.bytes() else {
+            return Err(s3_error!(InvalidRequest, "get body failed"));
+        };
 
         let mut create_req: AddServiceAccountReq =
             serde_json::from_slice(&body[..]).map_err(|e| s3_error!(InvalidRequest, "unmarshal body failed, e: {:?}", e))?;
@@ -38,8 +44,19 @@ impl Operation for AddServiceAccount {
             return Err(s3_error!(InvalidRequest, "access key has spaces"));
         }
 
+        let (cred, _) = check_key_valid(None, &input_cred.access_key).await.map_err(|e| {
+            debug!("check key failed: {e:?}");
+            s3_error!(InternalError, "check key failed")
+        })?;
+
+        // TODO check create_req validity
+
         // 校验合法性, Name, Expiration, Description
-        let _target_user = create_req.target_user.as_ref().unwrap_or(&cred.access_key);
+        let target_user = if let Some(u) = create_req.target_user {
+            u
+        } else {
+            cred.access_key
+        };
         let _deny_only = true;
 
         // todo 校验权限
@@ -63,18 +80,15 @@ impl Operation for AddServiceAccount {
         //
 
         let cred = CredentialsBuilder::new()
-            .parent_user(match create_req.target_user {
-                Some(target_user) => target_user,
-                _ => cred.access_key,
-            })
+            .parent_user(target_user)
             .access_key(create_req.access_key)
             .secret_key(create_req.secret_key)
-            .description(create_req.description)
+            .description(create_req.description.unwrap_or_default())
             .expiration(create_req.expiration)
             .session_policy({
                 match create_req.policy {
                     Some(p) if !p.is_empty() => {
-                        Some(serde_json::from_slice(&p).map_err(|_| s3_error!(InvalidRequest, "invalid policy"))?)
+                        Some(serde_json::from_slice(p.as_bytes()).map_err(|_| s3_error!(InvalidRequest, "invalid policy"))?)
                     }
                     _ => None,
                 }
@@ -203,12 +217,67 @@ impl Operation for InfoServiceAccount {
     }
 }
 
+#[derive(Debug, Default, serde::Deserialize)]
+pub struct ListServiceAccountQuery {
+    pub user: Option<String>,
+}
+
 pub struct ListServiceAccount {}
 #[async_trait::async_trait]
 impl Operation for ListServiceAccount {
-    async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle ListServiceAccount");
-        todo!()
+        let query = {
+            if let Some(query) = req.uri.query() {
+                let input: ListServiceAccountQuery =
+                    from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?;
+                input
+            } else {
+                ListServiceAccountQuery::default()
+            }
+        };
+
+        let target_account = if let Some(user) = query.user {
+            user
+        } else {
+            let Some(input_cred) = req.credentials else {
+                return Err(s3_error!(InvalidRequest, "get cred failed"));
+            };
+
+            let (cred, _owner) = check_key_valid(None, &input_cred.access_key).await.map_err(|e| {
+                debug!("check key failed: {e:?}");
+                s3_error!(InternalError, "check key failed")
+            })?;
+
+            if cred.parent_user.is_empty() {
+                input_cred.access_key
+            } else {
+                cred.parent_user
+            }
+        };
+
+        let service_accounts = iam::list_service_accounts(&target_account).await.map_err(|e| {
+            debug!("list service account failed: {e:?}");
+            s3_error!(InternalError, "list service account failed")
+        })?;
+
+        let accounts: Vec<ServiceAccountInfo> = service_accounts
+            .into_iter()
+            .map(|sa| ServiceAccountInfo {
+                parent_user: sa.parent_user,
+                account_status: sa.status,
+                implied_policy: true, // or set according to your logic
+                access_key: sa.access_key,
+                name: sa.name,
+                description: sa.description,
+                expiration: sa.expiration,
+            })
+            .collect();
+
+        let data = serde_json::to_vec(&ListServiceAccountsResp { accounts })
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal users err {}", e)))?;
+
+        Ok(S3Response::new((StatusCode::OK, Body::from(data))))
     }
 }
 

rustfs/src/admin/handlers/user.rs

@@ -1,6 +1,6 @@
-use futures::TryFutureExt;
 use http::StatusCode;
 use iam::get_global_action_cred;
+use madmin::{AccountStatus, AddOrUpdateUserReq};
 use matchit::Params;
 use s3s::{s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
 use serde::Deserialize;
@@ -16,6 +16,7 @@ use crate::admin::{
 pub struct AddUserQuery {
     #[serde(rename = "accessKey")]
     pub access_key: Option<String>,
+    pub status: Option<String>,
 }
 
 pub struct AddUser {}
@@ -39,6 +40,26 @@ impl Operation for AddUser {
 
         let ak = query.access_key.as_deref().unwrap_or_default();
 
+        if ak.is_empty() {
+            return Err(s3_error!(InvalidArgument, "access key is empty"));
+        }
+
+        let Some(body) = req.input.bytes() else {
+            return Err(s3_error!(InvalidRequest, "get body failed"));
+        };
+
+        // let body_bytes = decrypt_data(input_cred.secret_key.expose().as_bytes(), &body)
+        //     .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidArgument, format!("decrypt_data err {}", e)))?;
+
+        let args: AddOrUpdateUserReq = serde_json::from_slice(&body)
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("unmarshal body err {}", e)))?;
+
+        warn!("add user args {:?}", args);
+
+        if args.secret_key.is_empty() {
+            return Err(s3_error!(InvalidArgument, "access key is empty"));
+        }
+
         if let Some(sys_cred) = get_global_action_cred() {
             if sys_cred.access_key == ak {
                 return Err(s3_error!(InvalidArgument, "can't create user with system access key"));
@@ -62,15 +83,52 @@ impl Operation for AddUser {
             return Err(s3_error!(InvalidArgument, "can't create user with service account access key"));
         }
 
-        return Err(s3_error!(NotImplemented));
+        iam::create_user(ak, &args.secret_key, "enabled")
+            .await
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("create_user err {}", e)))?;
+
+        Ok(S3Response::new((StatusCode::OK, Body::empty())))
     }
 }
 
 pub struct SetUserStatus {}
 #[async_trait::async_trait]
 impl Operation for SetUserStatus {
-    async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        return Err(s3_error!(NotImplemented));
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        warn!("handle SetUserStatus");
+
+        let query = {
+            if let Some(query) = req.uri.query() {
+                let input: AddUserQuery =
+                    from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?;
+                input
+            } else {
+                AddUserQuery::default()
+            }
+        };
+
+        let ak = query.access_key.as_deref().unwrap_or_default();
+
+        if ak.is_empty() {
+            return Err(s3_error!(InvalidArgument, "access key is empty"));
+        }
+
+        let Some(input_cred) = req.credentials else {
+            return Err(s3_error!(InvalidRequest, "get cred failed"));
+        };
+
+        if input_cred.access_key == ak {
+            return Err(s3_error!(InvalidArgument, "can't change status of self"));
+        }
+
+        let status = AccountStatus::try_from(query.status.as_deref().unwrap_or_default())
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidArgument, e))?;
+
+        iam::set_user_status(ak, status)
+            .await
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("set_user_status err {}", e)))?;
+
+        Ok(S3Response::new((StatusCode::OK, Body::empty())))
     }
 }
 
@@ -83,25 +141,85 @@ impl Operation for ListUsers {
             .await
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
 
-        let body = serde_json::to_string(&users)
+        let data = serde_json::to_vec(&users)
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal users err {}", e)))?;
-        Ok(S3Response::new((StatusCode::OK, Body::from(body))))
+
+        // let Some(input_cred) = req.credentials else {
+        //     return Err(s3_error!(InvalidRequest, "get cred failed"));
+        // };
+
+        // let body = encrypt_data(input_cred.secret_key.expose().as_bytes(), &data)
+        //     .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidArgument, format!("encrypt_data err {}", e)))?;
+
+        Ok(S3Response::new((StatusCode::OK, Body::from(data))))
     }
 }
 
 pub struct RemoveUser {}
 #[async_trait::async_trait]
 impl Operation for RemoveUser {
-    async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         warn!("handle RemoveUser");
-        return Err(s3_error!(NotImplemented));
+        let query = {
+            if let Some(query) = req.uri.query() {
+                let input: AddUserQuery =
+                    from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?;
+                input
+            } else {
+                AddUserQuery::default()
+            }
+        };
+
+        let ak = query.access_key.as_deref().unwrap_or_default();
+
+        if ak.is_empty() {
+            return Err(s3_error!(InvalidArgument, "access key is empty"));
+        }
+
+        let (is_temp, _) = iam::is_temp_user(ak)
+            .await
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("is_temp_user err {}", e)))?;
+
+        if is_temp {
+            return Err(s3_error!(InvalidArgument, "can't remove temp user"));
+        }
+
+        iam::delete_user(ak, true)
+            .await
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("delete_user err {}", e)))?;
+
+        Ok(S3Response::new((StatusCode::OK, Body::empty())))
     }
 }
 
 pub struct GetUserInfo {}
 #[async_trait::async_trait]
 impl Operation for GetUserInfo {
-    async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        return Err(s3_error!(NotImplemented));
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        warn!("handle GetUserInfo");
+        let query = {
+            if let Some(query) = req.uri.query() {
+                let input: AddUserQuery =
+                    from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?;
+                input
+            } else {
+                AddUserQuery::default()
+            }
+        };
+
+        let ak = query.access_key.as_deref().unwrap_or_default();
+
+        if ak.is_empty() {
+            return Err(s3_error!(InvalidArgument, "access key is empty"));
+        }
+
+        let info = iam::get_user_info(ak)
+            .await
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
+
+        let data = serde_json::to_vec(&info)
+            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal user err {}", e)))?;
+
+        Ok(S3Response::new((StatusCode::OK, Body::from(data))))
     }
 }

rustfs/src/admin/mod.rs

@@ -127,29 +127,34 @@ fn regist_user_route(r: &mut S3Router<AdminOperation>) -> Result<()> {
         AdminOperation(&handlers::AccountInfoHandler {}),
     )?;
 
+    // 1
     r.insert(
         Method::GET,
         format!("{}{}", ADMIN_PREFIX, "/v3/list-users").as_str(),
         AdminOperation(&user::ListUsers {}),
     )?;
 
+    // 1
     r.insert(
         Method::GET,
         format!("{}{}", ADMIN_PREFIX, "/v3/user-info").as_str(),
         AdminOperation(&user::GetUserInfo {}),
     )?;
 
+    // 1
     r.insert(
         Method::DELETE,
         format!("{}{}", ADMIN_PREFIX, "/v3/remove-user").as_str(),
         AdminOperation(&user::RemoveUser {}),
     )?;
 
+    // 1
     r.insert(
         Method::PUT,
         format!("{}{}", ADMIN_PREFIX, "/v3/add-user").as_str(),
         AdminOperation(&user::AddUser {}),
     )?;
+    // 1
     r.insert(
         Method::PUT,
         format!("{}{}", ADMIN_PREFIX, "/v3/set-user-status").as_str(),
@@ -169,6 +174,7 @@ fn regist_user_route(r: &mut S3Router<AdminOperation>) -> Result<()> {
         AdminOperation(&InfoServiceAccount {}),
     )?;
 
+    // 1
     r.insert(
         Method::GET,
         format!("{}{}", ADMIN_PREFIX, "/v3/list-service-accounts").as_str(),

rustfs/src/admin/models/service_account.rs

@@ -1,20 +1,20 @@
-use serde::{Deserialize, Serialize};
+use serde::Serialize;
 use time::OffsetDateTime;
 
-#[derive(Deserialize, Default)]
-#[serde(rename_all = "camelCase", default)]
-pub struct AddServiceAccountReq {
-    pub access_key: String,
-    pub secret_key: String,
+// #[derive(Deserialize, Default)]
+// #[serde(rename_all = "camelCase", default)]
+// pub struct AddServiceAccountReq {
+//     pub access_key: String,
+//     pub secret_key: String,
 
-    pub policy: Option<Vec<u8>>,
-    pub target_user: Option<String>,
-    pub name: String,
-    pub description: String,
-    #[serde(with = "time::serde::rfc3339::option")]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub expiration: Option<OffsetDateTime>,
-}
+//     pub policy: Option<Vec<u8>>,
+//     pub target_user: Option<String>,
+//     pub name: String,
+//     pub description: String,
+//     #[serde(with = "time::serde::rfc3339::option")]
+//     #[serde(skip_serializing_if = "Option::is_none")]
+//     pub expiration: Option<OffsetDateTime>,
+// }
 
 #[derive(Serialize)]
 #[serde(rename_all = "camelCase")]