gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-18 10:43:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of github.com:rustfs/s3-rustfs into feature/Systemd.service

Browse Source
This commit is contained in:
houseme
2025-04-11 16:48:46 +08:00
parent ab8b19eb5d 1b24fbdb00
commit 42c8890d4a
24 changed files with 802 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/Cargo.toml
+12 -3
View File
@@ -44,7 +44,7 @@ http.workspace = true
http-body.workspace = true
iam = { path = "../iam" }
jsonwebtoken = "9.3.0"
libsystemd = { workspace = true, optional = true }
lock.workspace = true
local-ip-address = { workspace = true }
matchit = { workspace = true }
@@ -72,7 +72,12 @@ shadow-rs.workspace = true
tracing.workspace = true
time = { workspace = true, features = ["parsing", "formatting", "serde"] }
tokio-util.workspace = true
tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "signal"] }
tokio = { workspace = true, features = [
"rt-multi-thread",
"macros",
"net",
"signal",
] }
tokio-rustls.workspace = true
lazy_static.workspace = true
tokio-stream.workspace = true
@@ -102,7 +107,11 @@ ecstore = { path = "../ecstore" }
s3s.workspace = true
clap = { workspace = true }
tracing-subscriber = { workspace = true, features = ["env-filter", "time"] }
hyper-util = { workspace = true, features = ["tokio", "server-auto", "server-graceful"] }
hyper-util = { workspace = true, features = [
"tokio",
"server-auto",
"server-graceful",
] }
transform-stream = { workspace = true }
netif = "0.1.6"
shadow-rs.workspace = true
rustfs/src/admin/handlers.rs
+184 -40
View File
@@ -1,12 +1,12 @@
use super::router::Operation;
use crate::auth::check_key_valid;
use crate::auth::get_condition_values;
use crate::auth::get_session_token;
use crate::storage::error::to_s3_error;
use ::policy::policy::action::{Action, S3Action};
use ::policy::policy::resource::Resource;
use ::policy::policy::statement::BPStatement;
use ::policy::policy::{ActionSet, BucketPolicy, Effect, ResourceSet};
use bytes::Bytes;
use common::error::Error as ec_Error;
use ecstore::admin_server_info::get_server_info;
use ecstore::bucket::versioning_sys::BucketVersioningSys;
use ecstore::global::GLOBAL_ALlHealState;
use ecstore::heal::data_usage::load_data_usage_from_backend;
use ecstore::heal::heal_commands::HealOpts;
@@ -15,15 +15,23 @@ use ecstore::metrics_realtime::{collect_local_metrics, CollectMetricsOpts, Metri
use ecstore::new_object_layer_fn;
use ecstore::peer::is_reserved_or_invalid_bucket;
use ecstore::store::is_valid_object_prefix;
use ecstore::store_api::BucketOptions;
use ecstore::store_api::StorageAPI;
use ecstore::utils::path::path_join;
use ecstore::GLOBAL_Endpoints;
use futures::{Stream, StreamExt};
use http::{HeaderMap, Uri};
use hyper::StatusCode;
use iam::get_global_action_cred;
use iam::store::MappedPolicy;
use madmin::metrics::RealtimeMetrics;
use madmin::utils::parse_duration;
use matchit::Params;
use policy::policy::action::Action;
use policy::policy::action::S3Action;
use policy::policy::default::DEFAULT_POLICIES;
use policy::policy::Args;
use policy::policy::BucketPolicy;
use s3s::header::CONTENT_TYPE;
use s3s::stream::{ByteStream, DynByteStream};
use s3s::{s3_error, Body, S3Error, S3Request, S3Response, S3Result};
@@ -43,7 +51,7 @@ use tokio_stream::wrappers::ReceiverStream;
use tracing::{error, info, warn};
pub mod group;
pub mod policy;
pub mod policys;
pub mod service_account;
pub mod sts;
pub mod trace;
@@ -61,49 +69,187 @@ pub struct AccountInfoHandler {}
#[async_trait::async_trait]
impl Operation for AccountInfoHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle AccountInfoHandler");
let Some(cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) };
warn!("AccountInfoHandler cread {:?}", &cred);
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
// test policy
let Some(input_cred) = req.credentials else {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let mut s3_all_act = HashSet::with_capacity(1);
s3_all_act.insert(Action::S3Action(S3Action::AllActions));
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let mut all_res = HashSet::with_capacity(1);
all_res.insert(Resource::S3("*".to_string()));
let Ok(iam_store) = iam::get() else { return Err(s3_error!(InvalidRequest, "iam not init")) };
let bucket_policy = BucketPolicy {
id: "".into(),
version: "2012-10-17".to_owned(),
statements: vec![BPStatement {
sid: "".into(),
effect: Effect::Allow,
actions: ActionSet(s3_all_act.clone()),
resources: ResourceSet(all_res),
let default_claims = HashMap::new();
let claims = cred.claims.as_ref().unwrap_or(&default_claims);
let cred_clone = cred.clone();
let conditions = get_condition_values(&req.headers, &cred_clone);
let cred_clone = Arc::new(cred_clone);
let conditions = Arc::new(conditions);
let is_allow = Box::new({
let iam_clone = Arc::clone(&iam_store);
let cred_clone = Arc::clone(&cred_clone);
let conditions = Arc::clone(&conditions);
move |name: String| {
let iam_clone = Arc::clone(&iam_clone);
let cred_clone = Arc::clone(&cred_clone);
let conditions = Arc::clone(&conditions);
async move {
let (mut rd, mut wr) = (false, false);
if !iam_clone
.is_allowed(&Args {
account: &cred_clone.access_key,
groups: &cred_clone.groups,
action: Action::S3Action(S3Action::ListBucketAction),
bucket: &name,
conditions: &conditions,
is_owner: owner,
object: "",
claims,
deny_only: false,
})
.await
{
rd = true
}
if !iam_clone
.is_allowed(&Args {
account: &cred_clone.access_key,
groups: &cred_clone.groups,
action: Action::S3Action(S3Action::GetBucketLocationAction),
bucket: &name,
conditions: &conditions,
is_owner: owner,
object: "",
claims,
deny_only: false,
})
.await
{
rd = true
}
if !iam_clone
.is_allowed(&Args {
account: &cred_clone.access_key,
groups: &cred_clone.groups,
action: Action::S3Action(S3Action::PutObjectAction),
bucket: &name,
conditions: &conditions,
is_owner: owner,
object: "",
claims,
deny_only: false,
})
.await
{
wr = true
}
(rd, wr)
}
}
});
let account_name = if cred.is_temp() || cred.is_service_account() {
cred.parent_user.clone()
} else {
cred.access_key.clone()
};
let claims_args = Args {
account: "",
groups: &None,
action: Action::None,
bucket: "",
conditions: &HashMap::new(),
is_owner: false,
object: "",
claims,
deny_only: false,
};
let role_arn = claims_args.get_role_arn();
// TODO: get_policies_from_claims(claims);
let Some(admin_cred) = get_global_action_cred() else {
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
"get_global_action_cred failed".to_string(),
));
};
let mut effective_policy: policy::policy::Policy = Default::default();
if account_name == admin_cred.access_key {
for (name, p) in DEFAULT_POLICIES.iter() {
if *name == "consoleAdmin" {
effective_policy = p.clone();
break;
}
}
} else if let Some(arn) = role_arn {
let (_, policy_name) = iam_store
.get_role_policy(arn)
.await
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
let policies = MappedPolicy::new(&policy_name).to_slice();
effective_policy = iam_store.get_combined_policy(&policies).await;
} else {
let policies = iam_store
.policy_db_get(&account_name, &cred.groups)
.await
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("get policy failed: {}", e)))?;
effective_policy = iam_store.get_combined_policy(&policies).await;
};
let policy_str = serde_json::to_string(&effective_policy)
.map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse policy failed"))?;
let mut account_info = madmin::AccountInfo {
account_name,
server: store.backend_info().await,
policy: serde_json::Value::String(policy_str),
..Default::default()
};
// TODO: bucket policy
let buckets = store
.list_bucket(&BucketOptions {
cached: true,
..Default::default()
}],
};
})
.await
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
// let policy = bucket_policy
// .marshal_msg()
// .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse policy failed"))?;
for bucket in buckets.iter() {
let (rd, wr) = is_allow(bucket.name.clone()).await;
if rd || wr {
// TODO: BucketQuotaSys
// TODO: other attributes
account_info.buckets.push(madmin::BucketAccessInfo {
name: bucket.name.clone(),
details: Some(madmin::BucketDetails {
versioning: BucketVersioningSys::enabled(bucket.name.as_str()).await,
versioning_suspended: BucketVersioningSys::suspended(bucket.name.as_str()).await,
..Default::default()
}),
created: bucket.created,
access: madmin::AccountAccess { read: rd, write: wr },
..Default::default()
});
}
}
let backend_info = store.backend_info().await;
let info = AccountInfo {
account_name: cred.access_key,
server: backend_info,
policy: bucket_policy,
};
let data = serde_json::to_vec(&info)
let data = serde_json::to_vec(&account_info)
.map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse accountInfo failed"))?;
let mut header = HeaderMap::new();
@@ -128,8 +274,6 @@ pub struct ServerInfoHandler {}
#[async_trait::async_trait]
impl Operation for ServerInfoHandler {
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle ServerInfoHandler");
let info = get_server_info(true).await;
let data = serde_json::to_vec(&info)
rustfs/src/admin/handlers/policy.rs → rustfs/src/admin/handlers/policys.rs
+1 -2
View File
@@ -1,5 +1,3 @@
use std::collections::HashMap;
use crate::admin::{router::Operation, utils::has_space_be};
use http::{HeaderMap, StatusCode};
use iam::{error::is_err_no_such_user, get_global_action_cred, store::MappedPolicy};
@@ -8,6 +6,7 @@ use policy::policy::Policy;
use s3s::{header::CONTENT_TYPE, s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
use serde::Deserialize;
use serde_urlencoded::from_bytes;
use std::collections::HashMap;
use tracing::warn;
#[derive(Debug, Deserialize, Default)]
rustfs/src/admin/handlers/service_account.rs
+140 -19
View File
@@ -1,5 +1,5 @@
use crate::admin::utils::has_space_be;
use crate::auth::get_session_token;
use crate::auth::{get_condition_values, get_session_token};
use crate::{admin::router::Operation, auth::check_key_valid};
use http::HeaderMap;
use hyper::StatusCode;
@@ -13,7 +13,8 @@ use madmin::{
ServiceAccountInfo, UpdateServiceAccountReq,
};
use matchit::Params;
use policy::policy::Policy;
use policy::policy::action::{Action, AdminAction};
use policy::policy::{Args, Policy};
use s3s::S3ErrorCode::InvalidRequest;
use s3s::{header::CONTENT_TYPE, s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
use serde::Deserialize;
@@ -30,7 +31,7 @@ impl Operation for AddServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) =
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &req_cred.access_key).await?;
let mut input = req.input;
@@ -91,6 +92,25 @@ impl Operation for AddServiceAccount {
let Ok(iam_store) = iam::get() else { return Err(s3_error!(InvalidRequest, "iam not init")) };
let deny_only = cred.access_key == target_user || cred.parent_user == target_user;
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
groups: &cred.groups,
action: Action::AdminAction(AdminAction::CreateServiceAccountAdminAction),
bucket: "",
conditions: &get_condition_values(&req.headers, &cred),
is_owner: owner,
object: "",
claims: cred.claims.as_ref().unwrap_or(&HashMap::new()),
deny_only,
})
.await
{
return Err(s3_error!(AccessDenied, "access denied"));
}
if target_user != cred.access_key {
let has_user = iam_store.get_user(&target_user).await;
if has_user.is_none() && target_user != sys_cred.access_key {
@@ -212,7 +232,31 @@ impl Operation for UpdateServiceAccount {
update_req
.validate()
.map_err(|e| S3Error::with_message(InvalidRequest, e.to_string()))?;
// TODO: is_allowed
let Some(input_cred) = req.credentials else {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
groups: &cred.groups,
action: Action::AdminAction(AdminAction::UpdateServiceAccountAdminAction),
bucket: "",
conditions: &get_condition_values(&req.headers, &cred),
is_owner: owner,
object: "",
claims: cred.claims.as_ref().unwrap_or(&HashMap::new()),
deny_only: false,
})
.await
{
return Err(s3_error!(AccessDenied, "access denied"));
}
let sp = {
if let Some(policy) = update_req.new_policy {
let sp = Policy::parse_config(policy.as_bytes()).map_err(|e| {
@@ -280,7 +324,36 @@ impl Operation for InfoServiceAccount {
s3_error!(InternalError, "get service account failed")
})?;
// TODO: is_allowed
let Some(input_cred) = req.credentials else {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
groups: &cred.groups,
action: Action::AdminAction(AdminAction::ListServiceAccountsAdminAction),
bucket: "",
conditions: &get_condition_values(&req.headers, &cred),
is_owner: owner,
object: "",
claims: cred.claims.as_ref().unwrap_or(&HashMap::new()),
deny_only: false,
})
.await
{
let user = if cred.parent_user.is_empty() {
&cred.access_key
} else {
&cred.parent_user
};
if user != &svc_account.parent_user {
return Err(s3_error!(AccessDenied, "access denied"));
}
}
let implied_policy = if let Some(policy) = session_policy.as_ref() {
policy.version.is_empty() && policy.statements.is_empty()
@@ -359,7 +432,7 @@ impl Operation for ListServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) =
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key)
.await
.map_err(|e| {
@@ -367,22 +440,47 @@ impl Operation for ListServiceAccount {
s3_error!(InternalError, "check key failed")
})?;
let target_account = if let Some(user) = query.user {
if user != input_cred.access_key {
user
} else if cred.parent_user.is_empty() {
input_cred.access_key
} else {
cred.parent_user
// let target_account = if let Some(user) = query.user {
// if user != input_cred.access_key {
// user
// } else if cred.parent_user.is_empty() {
// input_cred.access_key
// } else {
// cred.parent_user
// }
// } else if cred.parent_user.is_empty() {
// input_cred.access_key
// } else {
// cred.parent_user
// };
let Ok(iam_store) = iam::get() else { return Err(s3_error!(InvalidRequest, "iam not init")) };
let target_account = if query.user.as_ref().is_some_and(|v| v != &cred.access_key) {
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
groups: &cred.groups,
action: Action::AdminAction(AdminAction::UpdateServiceAccountAdminAction),
bucket: "",
conditions: &get_condition_values(&req.headers, &cred),
is_owner: owner,
object: "",
claims: cred.claims.as_ref().unwrap_or(&HashMap::new()),
deny_only: false,
})
.await
{
return Err(s3_error!(AccessDenied, "access denied"));
}
query.user.unwrap_or_default()
} else if cred.parent_user.is_empty() {
input_cred.access_key
cred.access_key
} else {
cred.parent_user
};
let Ok(iam_store) = iam::get() else { return Err(s3_error!(InvalidRequest, "iam not init")) };
let service_accounts = iam_store.list_service_accounts(&target_account).await.map_err(|e| {
debug!("list service account failed: {e:?}");
s3_error!(InternalError, "list service account failed")
@@ -420,7 +518,7 @@ impl Operation for DeleteServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (_cred, _owner) =
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key)
.await
.map_err(|e| {
@@ -444,7 +542,7 @@ impl Operation for DeleteServiceAccount {
let Ok(iam_store) = iam::get() else { return Err(s3_error!(InvalidRequest, "iam not init")) };
let _svc_account = match iam_store.get_service_account(&query.access_key).await {
let svc_account = match iam_store.get_service_account(&query.access_key).await {
Ok((res, _)) => Some(res),
Err(err) => {
if is_err_no_such_service_account(&err) {
@@ -455,7 +553,30 @@ impl Operation for DeleteServiceAccount {
}
};
// TODO: is_allowed
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
groups: &cred.groups,
action: Action::AdminAction(AdminAction::RemoveServiceAccountAdminAction),
bucket: "",
conditions: &get_condition_values(&req.headers, &cred),
is_owner: owner,
object: "",
claims: cred.claims.as_ref().unwrap_or(&HashMap::new()),
deny_only: false,
})
.await
{
let user = if cred.parent_user.is_empty() {
&cred.access_key
} else {
&cred.parent_user
};
if svc_account.is_some_and(|v| &v.parent_user != user) {
return Err(s3_error!(InvalidRequest, "service account not exist"));
}
}
iam_store.delete_service_account(&query.access_key).await.map_err(|e| {
debug!("delete service account failed, e: {:?}", e);
rustfs/src/admin/handlers/user.rs
+51 -20
View File
@@ -1,9 +1,13 @@
use std::str::from_utf8;
use std::{collections::HashMap, str::from_utf8};
use http::{HeaderMap, StatusCode};
use iam::get_global_action_cred;
use madmin::{AccountStatus, AddOrUpdateUserReq};
use matchit::Params;
use policy::policy::{
action::{Action, AdminAction},
Args,
};
use s3s::{header::CONTENT_TYPE, s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
use serde::Deserialize;
use serde_urlencoded::from_bytes;
@@ -11,7 +15,7 @@ use tracing::warn;
use crate::{
admin::{router::Operation, utils::has_space_be},
auth::{check_key_valid, get_session_token},
auth::{check_key_valid, get_condition_values, get_session_token},
};
#[derive(Debug, Deserialize, Default)]
@@ -39,7 +43,7 @@ impl Operation for AddUser {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) =
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let ak = query.access_key.as_deref().unwrap_or_default();
@@ -63,8 +67,6 @@ impl Operation for AddUser {
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"));
}
@@ -89,13 +91,24 @@ impl Operation for AddUser {
return Err(s3_error!(InvalidArgument, "access key is not utf8"));
}
// let check_deny_only = if ak == cred.access_key {
// true
// } else {
// false
// };
// TODO: is_allowed
let deny_only = ak == cred.access_key;
let conditions = get_condition_values(&req.headers, &cred);
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
groups: &cred.groups,
action: Action::AdminAction(AdminAction::CreateUserAdminAction),
bucket: "",
conditions: &conditions,
is_owner: owner,
object: "",
claims: cred.claims.as_ref().unwrap_or(&HashMap::new()),
deny_only,
})
.await
{
return Err(s3_error!(AccessDenied, "access denied"));
}
iam_store
.create_user(ak, &args)
@@ -113,8 +126,6 @@ pub struct SetUserStatus {}
#[async_trait::async_trait]
impl Operation for SetUserStatus {
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 =
@@ -165,8 +176,6 @@ pub struct ListUsers {}
#[async_trait::async_trait]
impl Operation for ListUsers {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle ListUsers");
let query = {
if let Some(query) = req.uri.query() {
let input: BucketQuery =
@@ -214,8 +223,6 @@ pub struct RemoveUser {}
#[async_trait::async_trait]
impl Operation for RemoveUser {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle RemoveUser");
let query = {
if let Some(query) = req.uri.query() {
let input: AddUserQuery =
@@ -277,8 +284,6 @@ pub struct GetUserInfo {}
#[async_trait::async_trait]
impl Operation for GetUserInfo {
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 =
@@ -297,6 +302,32 @@ impl Operation for GetUserInfo {
let Ok(iam_store) = iam::get() else { return Err(s3_error!(InvalidRequest, "iam not init")) };
let Some(input_cred) = req.credentials else {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
let deny_only = ak == cred.access_key;
let conditions = get_condition_values(&req.headers, &cred);
if !iam_store
.is_allowed(&Args {
account: &cred.access_key,
groups: &cred.groups,
action: Action::AdminAction(AdminAction::GetUserAdminAction),
bucket: "",
conditions: &conditions,
is_owner: owner,
object: "",
claims: cred.claims.as_ref().unwrap_or(&HashMap::new()),
deny_only,
})
.await
{
return Err(s3_error!(AccessDenied, "access denied"));
}
let info = iam_store
.get_user_info(ak)
.await
rustfs/src/admin/mod.rs
+7 -6
View File
@@ -6,10 +6,11 @@ pub mod utils;
use common::error::Result;
// use ecstore::global::{is_dist_erasure, is_erasure};
use handlers::{
group, policy,
group, policys,
service_account::{AddServiceAccount, DeleteServiceAccount, InfoServiceAccount, ListServiceAccount, UpdateServiceAccount},
sts, user,
};
use hyper::Method;
use router::{AdminOperation, S3Router};
use rpc::regist_rpc_route;
@@ -231,35 +232,35 @@ fn regist_user_route(r: &mut S3Router<AdminOperation>) -> Result<()> {
r.insert(
Method::GET,
format!("{}{}", ADMIN_PREFIX, "/v3/list-canned-policies").as_str(),
AdminOperation(&policy::ListCannedPolicies {}),
AdminOperation(&policys::ListCannedPolicies {}),
)?;
// info-canned-policy?name=xxx
r.insert(
Method::GET,
format!("{}{}", ADMIN_PREFIX, "/v3/info-canned-policy").as_str(),
AdminOperation(&policy::InfoCannedPolicy {}),
AdminOperation(&policys::InfoCannedPolicy {}),
)?;
// add-canned-policy?name=xxx
r.insert(
Method::PUT,
format!("{}{}", ADMIN_PREFIX, "/v3/add-canned-policy").as_str(),
AdminOperation(&policy::AddCannedPolicy {}),
AdminOperation(&policys::AddCannedPolicy {}),
)?;
// remove-canned-policy?name=xxx
r.insert(
Method::DELETE,
format!("{}{}", ADMIN_PREFIX, "/v3/remove-canned-policy").as_str(),
AdminOperation(&policy::RemoveCannedPolicy {}),
AdminOperation(&policys::RemoveCannedPolicy {}),
)?;
// set-user-or-group-policy?policyName=xxx&userOrGroup=xxx&isGroup=xxx
r.insert(
Method::PUT,
format!("{}{}", ADMIN_PREFIX, "/v3/set-user-or-group-policy").as_str(),
AdminOperation(&policy::SetPolicyForUserOrGroup {}),
AdminOperation(&policys::SetPolicyForUserOrGroup {}),
)?;
Ok(())
rustfs/src/admin/router.rs
+4 -2
View File
@@ -67,14 +67,16 @@ where
uri.path().starts_with(ADMIN_PREFIX) || uri.path().starts_with(RPC_PREFIX)
}
async fn call(&self, req: S3Request<Body>) -> S3Result<S3Response<(StatusCode, Body)>> {
async fn call(&self, req: S3Request<Body>) -> S3Result<S3Response<Body>> {
let uri = format!("{}|{}", &req.method, req.uri.path());
// warn!("get uri {}", &uri);
if let Ok(mat) = self.router.at(&uri) {
let op: &T = mat.value;
return op.call(req, mat.params).await;
let mut resp = op.call(req, mat.params).await?;
resp.status = Some(resp.output.0);
return Ok(resp.map_output(|x| x.1));
}
return Err(s3_error!(NotImplemented));
rustfs/src/main.rs
+2 -3
View File
@@ -205,7 +205,7 @@ async fn run(opt: config::Opt) -> Result<()> {
// Setup S3 service
// 本项目使用 s3s 库来实现 s3 服务
let service = {
let s3_service = {
let store = storage::ecfs::FS::new();
// let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new(server_address.clone(), endpoint_pools).await?);
let mut b = S3ServiceBuilder::new(store.clone());
@@ -293,11 +293,10 @@ async fn run(opt: config::Opt) -> Result<()> {
let mut sigterm_inner = sigterm_inner;
let mut sigint_inner = sigint_inner;
let hyper_service = service.into_shared();
let hybrid_service = TowerToHyperService::new(
tower::ServiceBuilder::new()
.layer(CorsLayer::permissive())
.service(hybrid(hyper_service, rpc_service)),
.service(hybrid(s3_service, rpc_service)),
);
let http_server = ConnBuilder::new(TokioExecutor::new());
rustfs/src/storage/ecfs.rs
+43 -28
View File
@@ -261,14 +261,7 @@ impl S3 for FS {
.await
.map_err(to_s3_error)?;
let version_id = opts
.version_id
.as_ref()
.map(|v| match Uuid::parse_str(v) {
Ok(id) => Some(id),
Err(_) => None,
})
.unwrap_or_default();
let version_id = opts.version_id.as_ref().map(|v| Uuid::parse_str(v).ok()).unwrap_or_default();
let dobj = ObjectToDelete {
object_name: key,
version_id,
@@ -325,14 +318,7 @@ impl S3 for FS {
.objects
.iter()
.map(|v| {
let version_id = v
.version_id
.as_ref()
.map(|v| match Uuid::parse_str(v) {
Ok(id) => Some(id),
Err(_) => None,
})
.unwrap_or_default();
let version_id = v.version_id.as_ref().map(|v| Uuid::parse_str(v).ok()).unwrap_or_default();
ObjectToDelete {
object_name: v.key.clone(),
version_id,
@@ -407,8 +393,6 @@ impl S3 for FS {
async fn get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
// mc get 3
// warn!("get_object input {:?}, vid {:?}", &req.input, req.input.version_id);
let GetObjectInput {
bucket,
key,
@@ -447,8 +431,6 @@ impl S3 for FS {
return Err(s3_error!(InvalidArgument, "range and part_number invalid"));
}
// let metadata = extract_metadata(&req.headers);
let opts: ObjectOptions = get_opts(&bucket, &key, version_id, part_number, &req.headers)
.await
.map_err(to_s3_error)?;
@@ -516,16 +498,49 @@ impl S3 for FS {
#[tracing::instrument(level = "debug", skip(self, req))]
async fn head_object(&self, req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
// mc get 2
let HeadObjectInput { bucket, key, .. } = req.input;
let HeadObjectInput {
bucket,
key,
version_id,
part_number,
range,
..
} = req.input;
let part_number = part_number.map(|v| v as usize);
if let Some(part_num) = part_number {
if part_num == 0 {
return Err(s3_error!(InvalidArgument, "part_numer invalid"));
}
}
let rs = range.map(|v| match v {
Range::Int { first, last } => HTTPRangeSpec {
is_suffix_length: false,
start: first as usize,
end: last.map(|v| v as usize),
},
Range::Suffix { length } => HTTPRangeSpec {
is_suffix_length: true,
start: length as usize,
end: None,
},
});
if rs.is_some() && part_number.is_some() {
return Err(s3_error!(InvalidArgument, "range and part_number invalid"));
}
let opts: ObjectOptions = get_opts(&bucket, &key, version_id, part_number, &req.headers)
.await
.map_err(to_s3_error)?;
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let info = store
.get_object_info(&bucket, &key, &ObjectOptions::default())
.await
.map_err(to_s3_error)?;
let info = store.get_object_info(&bucket, &key, &opts).await.map_err(to_s3_error)?;
// warn!("head_object info {:?}", &info);
@@ -1054,11 +1069,11 @@ impl S3 for FS {
#[tracing::instrument(level = "debug", skip(self))]
async fn get_bucket_tagging(&self, req: S3Request<GetBucketTaggingInput>) -> S3Result<S3Response<GetBucketTaggingOutput>> {
let GetBucketTaggingInput { bucket, .. } = req.input;
let bucket = req.input.bucket.clone();
// check bucket exists.
let _bucket = self
.head_bucket(S3Request::new(HeadBucketInput {
bucket: bucket.clone(),
.head_bucket(req.map_input(|input| HeadBucketInput {
bucket: input.bucket,
expected_bucket_owner: None,
}))
.await?;