sdgoij/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-01-16 17:20:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

add s3 access check

Browse Source
This commit is contained in:
weisd
2025-03-06 15:17:17 +08:00
parent b821533b19
commit 937a0c7dee
11 changed files with 742 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
rustfs/src/admin/handlers.rs
View File

@@ -16,17 +16,11 @@ 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::StorageAPI;
use ecstore::utils::crypto::base64_encode;
use ecstore::utils::path::path_join;
use ecstore::GLOBAL_Endpoints;
use futures::{Stream, StreamExt};
use http::{HeaderMap, Uri};
use hyper::StatusCode;
use iam::auth::get_claims_from_token_with_secret;
use iam::error::Error as IamError;
use iam::policy::Policy;
use iam::sys::SESSION_POLICY_NAME;
use iam::{auth, get_global_action_cred};
use madmin::metrics::RealtimeMetrics;
use madmin::utils::parse_duration;
use matchit::Params;
@@ -35,7 +29,6 @@ use s3s::stream::{ByteStream, DynByteStream};
use s3s::{s3_error, Body, S3Error, S3Request, S3Response, S3Result};
use s3s::{S3ErrorCode, StdError};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use serde_urlencoded::from_bytes;
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
@@ -56,139 +49,6 @@ pub mod sts;
pub mod trace;
pub mod user;
// check_key_valid get auth.cred
pub async fn check_key_valid(security_token: Option<String>, ak: &str) -> S3Result<(auth::Credentials, bool)> {
let Some(mut cred) = get_global_action_cred() else {
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
format!("get_global_action_cred {:?}", IamError::IamSysNotInitialized),
));
};
let sys_cred = cred.clone();
// warn!("check_key_valid cred {:?}, as: {:?}", &cred, &ak);
if cred.access_key != ak {
let Ok(iam_store) = iam::get() else {
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
format!("check_key_valid {:?}", IamError::IamSysNotInitialized),
));
};
let (u, ok) = iam_store
.check_key(ak)
.await
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed1 {}", e)))?;
if !ok {
if let Some(u) = u {
if u.credentials.status == "off" {
return Err(s3_error!(InvalidRequest, "ErrAccessKeyDisabled"));
}
}
return Err(s3_error!(InvalidRequest, "check key failed"));
}
let Some(u) = u else {
return Err(s3_error!(InvalidRequest, "check key failed"));
};
cred = u.credentials;
}
// warn!("check_key_valid cred {:?}", &cred);
// warn!("check_key_valid security_token {:?}", &security_token);
let claims = check_claims_from_token(&security_token.unwrap_or_default(), &cred)
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed {}", e)))?;
cred.claims = if !claims.is_empty() { Some(claims) } else { None };
let mut owner = sys_cred.access_key == cred.access_key || cred.parent_user == sys_cred.access_key;
// permitRootAccess
if let Some(claims) = &cred.claims {
if claims.contains_key(SESSION_POLICY_NAME) {
owner = false
}
}
Ok((cred, owner))
}
pub fn check_claims_from_token(token: &str, cred: &auth::Credentials) -> S3Result<HashMap<String, Value>> {
if !token.is_empty() && cred.access_key.is_empty() {
return Err(s3_error!(InvalidRequest, "no access key"));
}
if token.is_empty() && cred.is_temp() && !cred.is_service_account() {
return Err(s3_error!(InvalidRequest, "invalid token"));
}
if !token.is_empty() && !cred.is_temp() {
return Err(s3_error!(InvalidRequest, "invalid token"));
}
if !cred.is_service_account() && cred.is_temp() && token != cred.session_token {
return Err(s3_error!(InvalidRequest, "invalid token"));
}
if cred.is_temp() && cred.is_expired() {
return Err(s3_error!(InvalidRequest, "invalid access key is temp and expired"));
}
let Some(sys_cred) = get_global_action_cred() else {
return Err(s3_error!(InternalError, "action cred not init"));
};
let mut secret = sys_cred.secret_key;
// TODO: REPLICATION
let mut token = token;
if cred.is_service_account() {
token = cred.session_token.as_str();
secret = cred.secret_key.clone();
}
if !token.is_empty() {
let claims: HashMap<String, Value> =
get_claims_from_token_with_secret(token, &secret).map_err(|_e| s3_error!(InvalidRequest, "invalid token"))?;
return Ok(claims);
}
Ok(HashMap::new())
}
pub fn get_session_token(hds: &HeaderMap) -> Option<String> {
hds.get("x-amz-security-token")
.map(|v| v.to_str().unwrap_or_default().to_string())
}
pub fn populate_session_policy(claims: &mut HashMap<String, Value>, policy: &str) -> S3Result<()> {
if !policy.is_empty() {
let session_policy = Policy::parse_config(policy.as_bytes())
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("parse policy err {}", e)))?;
if session_policy.version.is_empty() {
return Err(s3_error!(InvalidRequest, "invalid policy"));
}
let policy_buf = serde_json::to_vec(&session_policy)
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal policy err {}", e)))?;
if policy_buf.len() > 2048 {
return Err(s3_error!(InvalidRequest, "policy too large"));
}
claims.insert(SESSION_POLICY_NAME.to_string(), serde_json::Value::String(base64_encode(&policy_buf)));
}
Ok(())
}
#[derive(Debug, Serialize, Default)]
#[serde(rename_all = "PascalCase", default)]
pub struct AccountInfo {

32
rustfs/src/admin/handlers/service_account.rs
View File

@@ -1,5 +1,5 @@
use crate::admin::{handlers::check_key_valid, utils::has_space_be};
use crate::admin::{handlers::get_session_token, router::Operation};
use crate::admin::utils::has_space_be;
use crate::{admin::router::Operation, auth::check_key_valid};
use http::HeaderMap;
use hyper::StatusCode;
use iam::{
@@ -29,7 +29,7 @@ impl Operation for AddServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) = check_key_valid(get_session_token(&req.headers), &req_cred.access_key).await?;
let (cred, _owner) = check_key_valid(&req.headers, &req_cred.access_key).await?;
let mut input = req.input;
let body = match input.store_all_unlimited().await {
@@ -172,12 +172,6 @@ impl Operation for UpdateServiceAccount {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle UpdateServiceAccount");
// let Some(req_cred) = req.credentials else {
// return Err(s3_error!(InvalidRequest, "get cred failed"));
// };
// let (cred, _owner) = check_key_valid(get_session_token(&req.headers), &req_cred.access_key).await?;
let query = {
if let Some(query) = req.uri.query() {
let input: AccessKeyQuery =
@@ -363,12 +357,10 @@ impl Operation for ListServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) = check_key_valid(get_session_token(&req.headers), &input_cred.access_key)
.await
.map_err(|e| {
debug!("check key failed: {e:?}");
s3_error!(InternalError, "check key failed")
})?;
let (cred, _owner) = check_key_valid(&req.headers, &input_cred.access_key).await.map_err(|e| {
debug!("check key failed: {e:?}");
s3_error!(InternalError, "check key failed")
})?;
let target_account = if let Some(user) = query.user {
if user != input_cred.access_key {
@@ -423,12 +415,10 @@ impl Operation for DeleteServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (_cred, _owner) = check_key_valid(get_session_token(&req.headers), &input_cred.access_key)
.await
.map_err(|e| {
debug!("check key failed: {e:?}");
s3_error!(InternalError, "check key failed")
})?;
let (_cred, _owner) = check_key_valid(&req.headers, &input_cred.access_key).await.map_err(|e| {
debug!("check key failed: {e:?}");
s3_error!(InternalError, "check key failed")
})?;
let query = {
if let Some(query) = req.uri.query() {

36
rustfs/src/admin/handlers/sts.rs
View File

@@ -1,16 +1,19 @@
use crate::admin::{
handlers::{check_key_valid, get_session_token, populate_session_policy},
router::Operation,
use std::collections::HashMap;
use crate::{
admin::router::Operation,
auth::{check_key_valid, get_session_token},
};
use ecstore::utils::xml;
use ecstore::utils::{crypto::base64_encode, xml};
use http::StatusCode;
use iam::{auth::get_new_credentials_with_metadata, manager::get_token_signing_key};
use iam::{auth::get_new_credentials_with_metadata, manager::get_token_signing_key, policy::Policy, sys::SESSION_POLICY_NAME};
use matchit::Params;
use s3s::{
dto::{AssumeRoleOutput, Credentials, Timestamp},
s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result,
};
use serde::Deserialize;
use serde_json::Value;
use serde_urlencoded::from_bytes;
use time::{Duration, OffsetDateTime};
use tracing::{info, warn};
@@ -43,7 +46,7 @@ impl Operation for AssumeRoleHandle {
return Err(s3_error!(InvalidRequest, "AccessDenied1"));
}
let (cred, _owner) = check_key_valid(session_token, &user.access_key).await?;
let (cred, _owner) = check_key_valid(&req.headers, &user.access_key).await?;
// // TODO: 判断权限, 不允许sts访问
if cred.is_temp() || cred.is_service_account() {
@@ -137,3 +140,24 @@ impl Operation for AssumeRoleHandle {
Ok(S3Response::new((StatusCode::OK, Body::from(output))))
}
}
pub fn populate_session_policy(claims: &mut HashMap<String, Value>, policy: &str) -> S3Result<()> {
if !policy.is_empty() {
let session_policy = Policy::parse_config(policy.as_bytes())
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("parse policy err {}", e)))?;
if session_policy.version.is_empty() {
return Err(s3_error!(InvalidRequest, "invalid policy"));
}
let policy_buf = serde_json::to_vec(&session_policy)
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal policy err {}", e)))?;
if policy_buf.len() > 2048 {
return Err(s3_error!(InvalidRequest, "policy too large"));
}
claims.insert(SESSION_POLICY_NAME.to_string(), serde_json::Value::String(base64_encode(&policy_buf)));
}
Ok(())
}

11
rustfs/src/admin/handlers/user.rs
View File

@@ -9,10 +9,9 @@ use serde::Deserialize;
use serde_urlencoded::from_bytes;
use tracing::warn;
use crate::admin::{
handlers::{check_key_valid, get_session_token},
router::Operation,
utils::has_space_be,
use crate::{
admin::{router::Operation, utils::has_space_be},
auth::check_key_valid,
};
#[derive(Debug, Deserialize, Default)]
@@ -40,7 +39,7 @@ impl Operation for AddUser {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) = check_key_valid(get_session_token(&req.headers), &input_cred.access_key).await?;
let (cred, _owner) = check_key_valid(&req.headers, &input_cred.access_key).await?;
let ak = query.access_key.as_deref().unwrap_or_default();
@@ -247,7 +246,7 @@ impl Operation for RemoveUser {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) = check_key_valid(get_session_token(&req.headers), &input_cred.access_key).await?;
let (cred, _owner) = check_key_valid(&req.headers, &input_cred.access_key).await?;
let sys_cred = get_global_action_cred()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "get_global_action_cred failed"))?;

8
rustfs/src/admin/router.rs
View File

@@ -78,6 +78,14 @@ where
return Err(s3_error!(NotImplemented));
}
// check_access before call
async fn check_access(&self, req: &mut S3Request<Body>) -> S3Result<()> {
match req.credentials {
Some(_) => Ok(()),
None => Err(s3_error!(AccessDenied, "Signature is required")),
}
}
}
#[async_trait::async_trait]

260
rustfs/src/auth.rs
View File

@@ -1,8 +1,19 @@
use std::collections::HashMap;
use http::HeaderMap;
use iam::auth;
use iam::auth::get_claims_from_token_with_secret;
use iam::error::Error as IamError;
use iam::get_global_action_cred;
use iam::sys::SESSION_POLICY_NAME;
use s3s::auth::S3Auth;
use s3s::auth::SecretKey;
use s3s::auth::SimpleAuth;
use s3s::s3_error;
use s3s::S3Error;
use s3s::S3ErrorCode;
use s3s::S3Result;
use serde_json::Value;
pub struct IAMAuth {
simple_auth: SimpleAuth,
@@ -35,3 +46,252 @@ impl S3Auth for IAMAuth {
Err(s3_error!(UnauthorizedAccess, "Your account is not signed up2"))
}
}
// check_key_valid checks the key is valid or not. return the user's credentials and if the user is the owner.
pub async fn check_key_valid(header: &HeaderMap, access_key: &str) -> S3Result<(auth::Credentials, bool)> {
let Some(mut cred) = get_global_action_cred() else {
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
format!("get_global_action_cred {:?}", IamError::IamSysNotInitialized),
));
};
let sys_cred = cred.clone();
if cred.access_key != access_key {
let Ok(iam_store) = iam::get() else {
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
format!("check_key_valid {:?}", IamError::IamSysNotInitialized),
));
};
let (u, ok) = iam_store
.check_key(access_key)
.await
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed1 {}", e)))?;
if !ok {
if let Some(u) = u {
if u.credentials.status == "off" {
return Err(s3_error!(InvalidRequest, "ErrAccessKeyDisabled"));
}
}
return Err(s3_error!(InvalidRequest, "ErrAccessKeyDisabled"));
}
let Some(u) = u else {
return Err(s3_error!(InvalidRequest, "check key failed"));
};
cred = u.credentials;
}
let claims = check_claims_from_token(header, &cred)
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed {}", e)))?;
cred.claims = if !claims.is_empty() { Some(claims) } else { None };
let mut owner = sys_cred.access_key == cred.access_key || cred.parent_user == sys_cred.access_key;
// permitRootAccess
if let Some(claims) = &cred.claims {
if claims.contains_key(SESSION_POLICY_NAME) {
owner = false
}
}
Ok((cred, owner))
}
pub fn check_claims_from_token(header: &HeaderMap, cred: &auth::Credentials) -> S3Result<HashMap<String, Value>> {
let token = get_session_token(header).unwrap_or_default();
if !token.is_empty() && cred.access_key.is_empty() {
return Err(s3_error!(InvalidRequest, "no access key"));
}
if token.is_empty() && cred.is_temp() && !cred.is_service_account() {
return Err(s3_error!(InvalidRequest, "invalid token"));
}
if !token.is_empty() && !cred.is_temp() {
return Err(s3_error!(InvalidRequest, "invalid token"));
}
if !cred.is_service_account() && cred.is_temp() && token != cred.session_token {
return Err(s3_error!(InvalidRequest, "invalid token"));
}
if cred.is_temp() && cred.is_expired() {
return Err(s3_error!(InvalidRequest, "invalid access key is temp and expired"));
}
let Some(sys_cred) = get_global_action_cred() else {
return Err(s3_error!(InternalError, "action cred not init"));
};
// TODO: REPLICATION
let (token, secret) = if cred.is_service_account() {
(cred.session_token.as_str(), cred.secret_key.as_str())
} else {
(token, sys_cred.secret_key.as_str())
};
if !token.is_empty() {
let claims: HashMap<String, Value> =
get_claims_from_token_with_secret(token, secret).map_err(|_e| s3_error!(InvalidRequest, "invalid token"))?;
return Ok(claims);
}
Ok(HashMap::new())
}
pub fn get_session_token(hds: &HeaderMap) -> Option<&str> {
hds.get("x-amz-security-token").map(|v| v.to_str().unwrap_or_default())
}
pub fn get_condition_values(header: &HeaderMap, cred: &auth::Credentials) -> HashMap<String, Vec<String>> {
let username = if cred.is_temp() || cred.is_service_account() {
cred.parent_user.clone()
} else {
cred.access_key.clone()
};
let sys_cred = get_global_action_cred().unwrap_or_default();
let claims = &cred.claims;
let principal_type = if !username.is_empty() {
if claims.is_some() {
"AssumedRole"
} else if sys_cred.access_key == username {
"Account"
} else {
"User"
}
} else {
"Anonymous"
};
let mut args = HashMap::new();
args.insert("userid".to_owned(), vec![username.clone()]);
args.insert("username".to_owned(), vec![username]);
args.insert("principaltype".to_owned(), vec![principal_type.to_string()]);
let mut clone_header = header.clone();
if let Some(v) = clone_header.get("x-amz-signature-age") {
args.insert("signatureAge".to_string(), vec![v.to_str().unwrap_or("").to_string()]);
clone_header.remove("x-amz-signature-age");
}
// TODO: parse_object_tags
// if let Some(_user_tags) = clone_header.get("x-amz-tagging") {
// TODO: parse_object_tags
// if let Ok(tag) = tags::parse_object_tags(user_tags.to_str().unwrap_or("")) {
// let tag_map = tag.to_map();
// let mut keys = Vec::new();
// for (k, v) in tag_map {
// args.insert(format!("ExistingObjectTag/{}", k), vec![v.clone()]);
// args.insert(format!("RequestObjectTag/{}", k), vec![v.clone()]);
// keys.push(k);
// }
// args.insert("RequestObjectTagKeys".to_string(), keys);
// }
// }
for obj_lock in &[
"x-amz-object-lock-mode",
"x-amz-object-lock-legal-hold",
"x-amz-object-lock-retain-until-date",
] {
let values = clone_header
.get_all(*obj_lock)
.iter()
.map(|v| v.to_str().unwrap_or("").to_string())
.collect::<Vec<String>>();
if !values.is_empty() {
args.insert(obj_lock.trim_start_matches("x-amz-").to_string(), values);
}
clone_header.remove(*obj_lock);
}
for (key, _values) in clone_header.iter() {
if key.as_str().eq_ignore_ascii_case("x-amz-tagging") {
continue;
}
if let Some(existing_values) = args.get_mut(key.as_str()) {
existing_values.extend(clone_header.get_all(key).iter().map(|v| v.to_str().unwrap_or("").to_string()));
} else {
args.insert(
key.as_str().to_string(),
header
.get_all(key)
.iter()
.map(|v| v.to_str().unwrap_or("").to_string())
.collect(),
);
}
}
// TODO: add from url query
// let mut clone_url_values = r
// .uri()
// .query()
// .unwrap_or("")
// .split('&')
// .map(|s| {
// let mut split = s.split('=');
// (split.next().unwrap_or("").to_string(), split.next().unwrap_or("").to_string())
// })
// .collect::<HashMap<String, String>>();
// for obj_lock in &[
// "x-amz-object-lock-mode",
// "x-amz-object-lock-legal-hold",
// "x-amz-object-lock-retain-until-date",
// ] {
// if let Some(values) = clone_url_values.get(*obj_lock) {
// args.insert(obj_lock.trim_start_matches("x-amz-").to_string(), vec![values.clone()]);
// }
// clone_url_values.remove(*obj_lock);
// }
// for (key, values) in clone_url_values.iter() {
// if let Some(existing_values) = args.get_mut(key) {
// existing_values.push(values.clone());
// } else {
// args.insert(key.clone(), vec![values.clone()]);
// }
// }
if let Some(claims) = &cred.claims {
for (k, v) in claims {
if let Some(v_str) = v.as_str() {
args.insert(k.trim_start_matches("ldap").to_lowercase(), vec![v_str.to_string()]);
}
}
if let Some(grps_val) = claims.get("groups") {
if let Some(grps_is) = grps_val.as_array() {
let grps = grps_is
.iter()
.filter_map(|g| g.as_str().map(|s| s.to_string()))
.collect::<Vec<String>>();
if !grps.is_empty() {
args.insert("groups".to_string(), grps);
}
}
}
}
if let Some(groups) = &cred.groups {
if !args.contains_key("groups") {
args.insert("groups".to_string(), groups.clone());
}
}
args
}

1
rustfs/src/console.rs
View File

@@ -132,6 +132,7 @@ struct License {
pub(crate) static CONSOLE_CONFIG: OnceLock<Config> = OnceLock::new();
#[allow(clippy::const_is_empty)]
pub(crate) fn init_console_cfg(local_ip: Ipv4Addr, port: u16) {
CONSOLE_CONFIG.get_or_init(|| {
let ver = {

524
rustfs/src/storage/access.rs
View File

@@ -1,28 +1,55 @@
use std::str::FromStr;
use super::ecfs::FS;
use ecstore::bucket::policy::action::Action;
use http::HeaderMap;
use crate::auth::{check_key_valid, get_condition_values};
use iam::auth;
use iam::error::Error as IamError;
use iam::policy::action::{Action, S3Action};
use iam::sys::Args;
use s3s::access::{S3Access, S3AccessContext};
use s3s::auth::Credentials;
use s3s::{dto::*, s3_error, S3Request, S3Result};
use s3s::{dto::*, s3_error, S3Error, S3ErrorCode, S3Request, S3Result};
use std::collections::HashMap;
use tracing::info;
use uuid::Uuid;
#[allow(dead_code)]
#[derive(Default, Clone)]
struct ReqInfo {
pub card: Option<Credentials>,
pub action: Option<Action>,
pub(crate) struct ReqInfo {
pub cred: auth::Credentials,
pub is_owner: bool,
pub bucket: Option<String>,
pub object: Option<String>,
pub version_id: Option<Uuid>,
pub version_id: Option<String>,
}
async fn authorize_request(_req: &ReqInfo, _hs: &HeaderMap, _action: Action) -> S3Result<()> {
// TODO: globalIAMSys.IsAllowed
pub async fn authorize_request<T>(req: &mut S3Request<T>, actions: Vec<Action>) -> S3Result<()> {
let Ok(iam_store) = iam::get() else {
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
format!("check_key_valid {:?}", IamError::IamSysNotInitialized),
));
};
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
let default_claims = HashMap::new();
let claims = req_info.cred.claims.as_ref().unwrap_or(&default_claims);
let conditions = get_condition_values(&req.headers, &req_info.cred);
for action in actions {
let args = &Args {
account: &req_info.cred.access_key,
groups: &req_info.cred.groups,
action,
bucket: req_info.bucket.as_deref().unwrap_or(""),
conditions: &conditions,
is_owner: req_info.is_owner,
object: req_info.object.as_deref().unwrap_or(""),
claims,
deny_only: false,
};
if !iam_store.is_allowed(args).await {
return Err(s3_error!(AccessDenied, "Access Denied"));
}
}
Ok(())
}
@@ -45,31 +72,28 @@ impl S3Access for FS {
async fn check(&self, cx: &mut S3AccessContext<'_>) -> S3Result<()> {
// 上层验证了 ak/sk
info!(
"s3 check path: {:?}, s3_op: {:?}, cred: {:?}",
"s3 check uri: {:?}, method: {:?} path: {:?}, s3_op: {:?}, cred: {:?}, headers:{:?}",
cx.uri(),
cx.method(),
cx.s3_path(),
cx.s3_op().name(),
cx.credentials()
cx.credentials(),
cx.headers(),
// cx.extensions_mut(),
);
if cx.credentials().is_none() {
let Some(input_cred) = cx.credentials() else {
return Err(s3_error!(UnauthorizedAccess, "Signature is required"));
};
// TODO: FIXME: check auth
let action = match Action::from_str(format!("s3:{}", cx.s3_op().name()).as_str()) {
Ok(res) => Some(res),
Err(_) => None,
};
let (cred, is_owner) = check_key_valid(cx.headers(), &input_cred.access_key).await?;
let req_info = ReqInfo {
card: cx.credentials().cloned(),
action,
cred,
is_owner,
..Default::default()
};
// warn!("req_info {:?}", req_info);
let ext = cx.extensions_mut();
ext.insert(req_info);
@@ -82,14 +106,23 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn create_bucket(&self, req: &mut S3Request<CreateBucketInput>) -> S3Result<()> {
if let Some(req_info) = req.extensions.get_mut::<ReqInfo>() {
let CreateBucketInput { bucket, .. } = &req.input;
req_info.bucket = Some(bucket.to_owned());
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req_info, &req.headers, Action::CreateBucket).await
} else {
Err(s3_error!(AccessDenied, "AccessDenied"))
authorize_request(req, vec![Action::S3Action(S3Action::CreateBucketAction)]).await?;
if req.input.object_lock_enabled_for_bucket.is_some_and(|v| v) {
authorize_request(
req,
vec![
Action::S3Action(S3Action::PutBucketObjectLockConfigurationAction),
Action::S3Action(S3Action::PutBucketVersioningAction),
],
)
.await?;
}
Ok(())
}
/// Checks whether the AbortMultipartUpload request has accesses to the resources.
///
@@ -108,8 +141,32 @@ impl S3Access for FS {
/// Checks whether the CopyObject request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn copy_object(&self, _req: &mut S3Request<CopyObjectInput>) -> S3Result<()> {
Ok(())
async fn copy_object(&self, req: &mut S3Request<CopyObjectInput>) -> S3Result<()> {
{
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
let (src_bucket, src_key, version_id) = match &req.input.copy_source {
CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)),
CopySource::Bucket {
ref bucket,
ref key,
version_id,
} => (bucket.to_string(), key.to_string(), version_id.as_ref().map(|v| v.to_string())),
};
req_info.bucket = Some(src_bucket);
req_info.object = Some(src_key);
req_info.version_id = version_id;
authorize_request(req, vec![Action::S3Action(S3Action::GetObjectAction)]).await?;
}
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::PutObjectAction)]).await
}
/// Checks whether the CreateMultipartUpload request has accesses to the resources.
@@ -122,7 +179,15 @@ impl S3Access for FS {
/// Checks whether the DeleteBucket request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_bucket(&self, _req: &mut S3Request<DeleteBucketInput>) -> S3Result<()> {
async fn delete_bucket(&self, req: &mut S3Request<DeleteBucketInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::DeleteBucketAction)]).await?;
if req.input.force_delete.is_some_and(|v| v) {
authorize_request(req, vec![Action::S3Action(S3Action::ForceDeleteBucketAction)]).await?;
}
Ok(())
}
@@ -139,15 +204,21 @@ impl S3Access for FS {
/// Checks whether the DeleteBucketCors request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_bucket_cors(&self, _req: &mut S3Request<DeleteBucketCorsInput>) -> S3Result<()> {
Ok(())
async fn delete_bucket_cors(&self, req: &mut S3Request<DeleteBucketCorsInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketCorsAction)]).await
}
/// Checks whether the DeleteBucketEncryption request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_bucket_encryption(&self, _req: &mut S3Request<DeleteBucketEncryptionInput>) -> S3Result<()> {
Ok(())
async fn delete_bucket_encryption(&self, req: &mut S3Request<DeleteBucketEncryptionInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketEncryptionAction)]).await
}
/// Checks whether the DeleteBucketIntelligentTieringConfiguration request has accesses to the resources.
@@ -173,8 +244,11 @@ impl S3Access for FS {
/// Checks whether the DeleteBucketLifecycle request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_bucket_lifecycle(&self, _req: &mut S3Request<DeleteBucketLifecycleInput>) -> S3Result<()> {
Ok(())
async fn delete_bucket_lifecycle(&self, req: &mut S3Request<DeleteBucketLifecycleInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketLifecycleAction)]).await
}
/// Checks whether the DeleteBucketMetricsConfiguration request has accesses to the resources.
@@ -197,22 +271,31 @@ impl S3Access for FS {
/// Checks whether the DeleteBucketPolicy request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_bucket_policy(&self, _req: &mut S3Request<DeleteBucketPolicyInput>) -> S3Result<()> {
Ok(())
async fn delete_bucket_policy(&self, req: &mut S3Request<DeleteBucketPolicyInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::DeleteBucketPolicyAction)]).await
}
/// Checks whether the DeleteBucketReplication request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_bucket_replication(&self, _req: &mut S3Request<DeleteBucketReplicationInput>) -> S3Result<()> {
Ok(())
async fn delete_bucket_replication(&self, req: &mut S3Request<DeleteBucketReplicationInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutReplicationConfigurationAction)]).await
}
/// Checks whether the DeleteBucketTagging request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_bucket_tagging(&self, _req: &mut S3Request<DeleteBucketTaggingInput>) -> S3Result<()> {
Ok(())
async fn delete_bucket_tagging(&self, req: &mut S3Request<DeleteBucketTaggingInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketTaggingAction)]).await
}
/// Checks whether the DeleteBucketWebsite request has accesses to the resources.
@@ -225,15 +308,25 @@ impl S3Access for FS {
/// Checks whether the DeleteObject request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_object(&self, _req: &mut S3Request<DeleteObjectInput>) -> S3Result<()> {
Ok(())
async fn delete_object(&self, req: &mut S3Request<DeleteObjectInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::DeleteObjectAction)]).await
}
/// Checks whether the DeleteObjectTagging request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn delete_object_tagging(&self, _req: &mut S3Request<DeleteObjectTaggingInput>) -> S3Result<()> {
Ok(())
async fn delete_object_tagging(&self, req: &mut S3Request<DeleteObjectTaggingInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::DeleteObjectTaggingAction)]).await
}
/// Checks whether the DeleteObjects request has accesses to the resources.
@@ -263,8 +356,11 @@ impl S3Access for FS {
/// Checks whether the GetBucketAcl request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_acl(&self, _req: &mut S3Request<GetBucketAclInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_acl(&self, req: &mut S3Request<GetBucketAclInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketPolicyAction)]).await
}
/// Checks whether the GetBucketAnalyticsConfiguration request has accesses to the resources.
@@ -280,15 +376,21 @@ impl S3Access for FS {
/// Checks whether the GetBucketCors request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_cors(&self, _req: &mut S3Request<GetBucketCorsInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_cors(&self, req: &mut S3Request<GetBucketCorsInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketCorsAction)]).await
}
/// Checks whether the GetBucketEncryption request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_encryption(&self, _req: &mut S3Request<GetBucketEncryptionInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_encryption(&self, req: &mut S3Request<GetBucketEncryptionInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketEncryptionAction)]).await
}
/// Checks whether the GetBucketIntelligentTieringConfiguration request has accesses to the resources.
@@ -316,16 +418,22 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn get_bucket_lifecycle_configuration(
&self,
_req: &mut S3Request<GetBucketLifecycleConfigurationInput>,
req: &mut S3Request<GetBucketLifecycleConfigurationInput>,
) -> S3Result<()> {
Ok(())
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketLifecycleAction)]).await
}
/// Checks whether the GetBucketLocation request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_location(&self, _req: &mut S3Request<GetBucketLocationInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_location(&self, req: &mut S3Request<GetBucketLocationInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketLocationAction)]).await
}
/// Checks whether the GetBucketLogging request has accesses to the resources.
@@ -347,9 +455,12 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn get_bucket_notification_configuration(
&self,
_req: &mut S3Request<GetBucketNotificationConfigurationInput>,
req: &mut S3Request<GetBucketNotificationConfigurationInput>,
) -> S3Result<()> {
Ok(())
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketNotificationAction)]).await
}
/// Checks whether the GetBucketOwnershipControls request has accesses to the resources.
@@ -362,22 +473,31 @@ impl S3Access for FS {
/// Checks whether the GetBucketPolicy request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_policy(&self, _req: &mut S3Request<GetBucketPolicyInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_policy(&self, req: &mut S3Request<GetBucketPolicyInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketPolicyAction)]).await
}
/// Checks whether the GetBucketPolicyStatus request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_policy_status(&self, _req: &mut S3Request<GetBucketPolicyStatusInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_policy_status(&self, req: &mut S3Request<GetBucketPolicyStatusInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketPolicyStatusAction)]).await
}
/// Checks whether the GetBucketReplication request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_replication(&self, _req: &mut S3Request<GetBucketReplicationInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_replication(&self, req: &mut S3Request<GetBucketReplicationInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetReplicationConfigurationAction)]).await
}
/// Checks whether the GetBucketRequestPayment request has accesses to the resources.
@@ -390,15 +510,21 @@ impl S3Access for FS {
/// Checks whether the GetBucketTagging request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_tagging(&self, _req: &mut S3Request<GetBucketTaggingInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_tagging(&self, req: &mut S3Request<GetBucketTaggingInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketTaggingAction)]).await
}
/// Checks whether the GetBucketVersioning request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_bucket_versioning(&self, _req: &mut S3Request<GetBucketVersioningInput>) -> S3Result<()> {
Ok(())
async fn get_bucket_versioning(&self, req: &mut S3Request<GetBucketVersioningInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketVersioningAction)]).await
}
/// Checks whether the GetBucketWebsite request has accesses to the resources.
@@ -411,50 +537,92 @@ impl S3Access for FS {
/// Checks whether the GetObject request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_object(&self, _req: &mut S3Request<GetObjectInput>) -> S3Result<()> {
Ok(())
async fn get_object(&self, req: &mut S3Request<GetObjectInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::GetObjectAction)]).await
}
/// Checks whether the GetObjectAcl request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_object_acl(&self, _req: &mut S3Request<GetObjectAclInput>) -> S3Result<()> {
Ok(())
async fn get_object_acl(&self, req: &mut S3Request<GetObjectAclInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketPolicyAction)]).await
}
/// Checks whether the GetObjectAttributes request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_object_attributes(&self, _req: &mut S3Request<GetObjectAttributesInput>) -> S3Result<()> {
Ok(())
async fn get_object_attributes(&self, req: &mut S3Request<GetObjectAttributesInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
let mut actions = Vec::new();
if req.input.version_id.is_some() {
actions.push(Action::S3Action(S3Action::GetObjectVersionAttributesAction));
actions.push(Action::S3Action(S3Action::GetObjectVersionAction));
} else {
actions.push(Action::S3Action(S3Action::GetObjectAttributesAction));
actions.push(Action::S3Action(S3Action::GetObjectAction));
}
authorize_request(req, actions).await
}
/// Checks whether the GetObjectLegalHold request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_object_legal_hold(&self, _req: &mut S3Request<GetObjectLegalHoldInput>) -> S3Result<()> {
Ok(())
async fn get_object_legal_hold(&self, req: &mut S3Request<GetObjectLegalHoldInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::GetObjectLegalHoldAction)]).await
}
/// Checks whether the GetObjectLockConfiguration request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_object_lock_configuration(&self, _req: &mut S3Request<GetObjectLockConfigurationInput>) -> S3Result<()> {
Ok(())
async fn get_object_lock_configuration(&self, req: &mut S3Request<GetObjectLockConfigurationInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetBucketObjectLockConfigurationAction)]).await
}
/// Checks whether the GetObjectRetention request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_object_retention(&self, _req: &mut S3Request<GetObjectRetentionInput>) -> S3Result<()> {
Ok(())
async fn get_object_retention(&self, req: &mut S3Request<GetObjectRetentionInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::GetObjectRetentionAction)]).await
}
/// Checks whether the GetObjectTagging request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn get_object_tagging(&self, _req: &mut S3Request<GetObjectTaggingInput>) -> S3Result<()> {
Ok(())
async fn get_object_tagging(&self, req: &mut S3Request<GetObjectTaggingInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::GetObjectTaggingAction)]).await
}
/// Checks whether the GetObjectTorrent request has accesses to the resources.
@@ -474,15 +642,23 @@ impl S3Access for FS {
/// Checks whether the HeadBucket request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn head_bucket(&self, _req: &mut S3Request<HeadBucketInput>) -> S3Result<()> {
Ok(())
async fn head_bucket(&self, req: &mut S3Request<HeadBucketInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::ListBucketAction)]).await
}
/// Checks whether the HeadObject request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn head_object(&self, _req: &mut S3Request<HeadObjectInput>) -> S3Result<()> {
Ok(())
async fn head_object(&self, req: &mut S3Request<HeadObjectInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::GetObjectAction)]).await
}
/// Checks whether the ListBucketAnalyticsConfigurations request has accesses to the resources.
@@ -529,14 +705,18 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn list_buckets(&self, _req: &mut S3Request<ListBucketsInput>) -> S3Result<()> {
// check inside
Ok(())
}
/// Checks whether the ListMultipartUploads request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn list_multipart_uploads(&self, _req: &mut S3Request<ListMultipartUploadsInput>) -> S3Result<()> {
Ok(())
async fn list_multipart_uploads(&self, req: &mut S3Request<ListMultipartUploadsInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::ListBucketMultipartUploadsAction)]).await
}
/// Checks whether the ListObjectVersions request has accesses to the resources.
@@ -549,15 +729,21 @@ impl S3Access for FS {
/// Checks whether the ListObjects request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn list_objects(&self, _req: &mut S3Request<ListObjectsInput>) -> S3Result<()> {
Ok(())
async fn list_objects(&self, req: &mut S3Request<ListObjectsInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::ListBucketAction)]).await
}
/// Checks whether the ListObjectsV2 request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn list_objects_v2(&self, _req: &mut S3Request<ListObjectsV2Input>) -> S3Result<()> {
Ok(())
async fn list_objects_v2(&self, req: &mut S3Request<ListObjectsV2Input>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::ListBucketAction)]).await
}
/// Checks whether the ListParts request has accesses to the resources.
@@ -580,8 +766,11 @@ impl S3Access for FS {
/// Checks whether the PutBucketAcl request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_bucket_acl(&self, _req: &mut S3Request<PutBucketAclInput>) -> S3Result<()> {
Ok(())
async fn put_bucket_acl(&self, req: &mut S3Request<PutBucketAclInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketPolicyAction)]).await
}
/// Checks whether the PutBucketAnalyticsConfiguration request has accesses to the resources.
@@ -597,15 +786,21 @@ impl S3Access for FS {
/// Checks whether the PutBucketCors request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_bucket_cors(&self, _req: &mut S3Request<PutBucketCorsInput>) -> S3Result<()> {
Ok(())
async fn put_bucket_cors(&self, req: &mut S3Request<PutBucketCorsInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketCorsAction)]).await
}
/// Checks whether the PutBucketEncryption request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_bucket_encryption(&self, _req: &mut S3Request<PutBucketEncryptionInput>) -> S3Result<()> {
Ok(())
async fn put_bucket_encryption(&self, req: &mut S3Request<PutBucketEncryptionInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketEncryptionAction)]).await
}
/// Checks whether the PutBucketIntelligentTieringConfiguration request has accesses to the resources.
@@ -633,9 +828,12 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn put_bucket_lifecycle_configuration(
&self,
_req: &mut S3Request<PutBucketLifecycleConfigurationInput>,
req: &mut S3Request<PutBucketLifecycleConfigurationInput>,
) -> S3Result<()> {
Ok(())
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketLifecycleAction)]).await
}
/// Checks whether the PutBucketLogging request has accesses to the resources.
@@ -657,9 +855,12 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn put_bucket_notification_configuration(
&self,
_req: &mut S3Request<PutBucketNotificationConfigurationInput>,
req: &mut S3Request<PutBucketNotificationConfigurationInput>,
) -> S3Result<()> {
Ok(())
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketNotificationAction)]).await
}
/// Checks whether the PutBucketOwnershipControls request has accesses to the resources.
@@ -672,15 +873,21 @@ impl S3Access for FS {
/// Checks whether the PutBucketPolicy request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_bucket_policy(&self, _req: &mut S3Request<PutBucketPolicyInput>) -> S3Result<()> {
Ok(())
async fn put_bucket_policy(&self, req: &mut S3Request<PutBucketPolicyInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketPolicyAction)]).await
}
/// Checks whether the PutBucketReplication request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_bucket_replication(&self, _req: &mut S3Request<PutBucketReplicationInput>) -> S3Result<()> {
Ok(())
async fn put_bucket_replication(&self, req: &mut S3Request<PutBucketReplicationInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutReplicationConfigurationAction)]).await
}
/// Checks whether the PutBucketRequestPayment request has accesses to the resources.
@@ -693,15 +900,21 @@ impl S3Access for FS {
/// Checks whether the PutBucketTagging request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_bucket_tagging(&self, _req: &mut S3Request<PutBucketTaggingInput>) -> S3Result<()> {
Ok(())
async fn put_bucket_tagging(&self, req: &mut S3Request<PutBucketTaggingInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketTaggingAction)]).await
}
/// Checks whether the PutBucketVersioning request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_bucket_versioning(&self, _req: &mut S3Request<PutBucketVersioningInput>) -> S3Result<()> {
Ok(())
async fn put_bucket_versioning(&self, req: &mut S3Request<PutBucketVersioningInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketVersioningAction)]).await
}
/// Checks whether the PutBucketWebsite request has accesses to the resources.
@@ -714,43 +927,71 @@ impl S3Access for FS {
/// Checks whether the PutObject request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_object(&self, _req: &mut S3Request<PutObjectInput>) -> S3Result<()> {
Ok(())
async fn put_object(&self, req: &mut S3Request<PutObjectInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::PutObjectAction)]).await
}
/// Checks whether the PutObjectAcl request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_object_acl(&self, _req: &mut S3Request<PutObjectAclInput>) -> S3Result<()> {
Ok(())
async fn put_object_acl(&self, req: &mut S3Request<PutObjectAclInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketPolicyAction)]).await
}
/// Checks whether the PutObjectLegalHold request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_object_legal_hold(&self, _req: &mut S3Request<PutObjectLegalHoldInput>) -> S3Result<()> {
Ok(())
async fn put_object_legal_hold(&self, req: &mut S3Request<PutObjectLegalHoldInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::PutObjectLegalHoldAction)]).await
}
/// Checks whether the PutObjectLockConfiguration request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_object_lock_configuration(&self, _req: &mut S3Request<PutObjectLockConfigurationInput>) -> S3Result<()> {
Ok(())
async fn put_object_lock_configuration(&self, req: &mut S3Request<PutObjectLockConfigurationInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutBucketObjectLockConfigurationAction)]).await
}
/// Checks whether the PutObjectRetention request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_object_retention(&self, _req: &mut S3Request<PutObjectRetentionInput>) -> S3Result<()> {
Ok(())
async fn put_object_retention(&self, req: &mut S3Request<PutObjectRetentionInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::PutObjectRetentionAction)]).await
}
/// Checks whether the PutObjectTagging request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn put_object_tagging(&self, _req: &mut S3Request<PutObjectTaggingInput>) -> S3Result<()> {
Ok(())
async fn put_object_tagging(&self, req: &mut S3Request<PutObjectTaggingInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::PutObjectTaggingAction)]).await
}
/// Checks whether the PutPublicAccessBlock request has accesses to the resources.
@@ -763,22 +1004,35 @@ impl S3Access for FS {
/// Checks whether the RestoreObject request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn restore_object(&self, _req: &mut S3Request<RestoreObjectInput>) -> S3Result<()> {
Ok(())
async fn restore_object(&self, req: &mut S3Request<RestoreObjectInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, vec![Action::S3Action(S3Action::RestoreObjectAction)]).await
}
/// Checks whether the SelectObjectContent request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn select_object_content(&self, _req: &mut S3Request<SelectObjectContentInput>) -> S3Result<()> {
Ok(())
async fn select_object_content(&self, req: &mut S3Request<SelectObjectContentInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
authorize_request(req, vec![Action::S3Action(S3Action::GetObjectAction)]).await
}
/// Checks whether the UploadPart request has accesses to the resources.
///
/// This method returns `Ok(())` by default.
async fn upload_part(&self, _req: &mut S3Request<UploadPartInput>) -> S3Result<()> {
Ok(())
async fn upload_part(&self, req: &mut S3Request<UploadPartInput>) -> S3Result<()> {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
authorize_request(req, vec![Action::S3Action(S3Action::PutObjectAction)]).await
}
/// Checks whether the UploadPartCopy request has accesses to the resources.

34
rustfs/src/storage/ecfs.rs
View File

@@ -1,6 +1,8 @@
use super::access::authorize_request;
use super::options::del_opts;
use super::options::extract_metadata;
use super::options::put_opts;
use crate::storage::access::ReqInfo;
use bytes::Bytes;
use common::error::Result;
use ecstore::bucket::error::BucketMetadataError;
@@ -36,6 +38,8 @@ use ecstore::xhttp;
use futures::pin_mut;
use futures::{Stream, StreamExt};
use http::HeaderMap;
use iam::policy::action::Action;
use iam::policy::action::S3Action;
use lazy_static::lazy_static;
use log::warn;
use s3s::dto::*;
@@ -536,14 +540,36 @@ impl S3 for FS {
}
#[tracing::instrument(level = "debug", skip(self))]
async fn list_buckets(&self, _: S3Request<ListBucketsInput>) -> S3Result<S3Response<ListBucketsOutput>> {
async fn list_buckets(&self, req: S3Request<ListBucketsInput>) -> S3Result<S3Response<ListBucketsOutput>> {
// mc ls
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let bucket_infos = store.list_bucket(&BucketOptions::default()).await.map_err(to_s3_error)?;
let mut bucket_infos = store.list_bucket(&BucketOptions::default()).await.map_err(to_s3_error)?;
let mut req = req;
if authorize_request(&mut req, vec![Action::S3Action(S3Action::ListAllMyBucketsAction)])
.await
.is_err()
{
bucket_infos.retain(|info| {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(info.name.clone());
futures::executor::block_on(async {
authorize_request(&mut req, vec![Action::S3Action(S3Action::ListBucketAction)])
.await
.is_ok()
|| authorize_request(&mut req, vec![Action::S3Action(S3Action::GetBucketLocationAction)])
.await
.is_ok()
})
});
}
let buckets: Vec<Bucket> = bucket_infos
.iter()
@@ -911,9 +937,9 @@ impl S3 for FS {
async fn upload_part_copy(&self, req: S3Request<UploadPartCopyInput>) -> S3Result<S3Response<UploadPartCopyOutput>> {
let _input = req.input;
let output = UploadPartCopyOutput { ..Default::default() };
let _output = UploadPartCopyOutput { ..Default::default() };
Ok(S3Response::new(output))
unimplemented!("upload_part_copy");
}
#[tracing::instrument(level = "debug", skip(self, req))]

1
rustfs/src/utils.rs
View File

@@ -1,4 +1,3 @@
use local_ip_address;
use std::net::IpAddr;
pub(crate) fn get_local_ip() -> Option<std::net::Ipv4Addr> {

8
scripts/run.sh
View File

@@ -31,4 +31,12 @@ if [ -n "$1" ]; then
fi
# check ./rustfs/static/index.html not exists
if [ ! -f ./rustfs/static/index.html ]; then
echo "Downloading rustfs-console-latest.zip"
# download rustfs-console-latest.zip do not show log
curl -s -L "https://dl.rustfs.com/console/rustfs-console-latest.zip" -o tempfile.zip && unzip -q -o tempfile.zip -d ./rustfs/static && rm tempfile.zip
fi
cargo run --bin rustfs