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

merge main

Browse Source
This commit is contained in:
weisd
2025-04-14 22:38:58 +08:00
parent b9fb5e7e84 08fe5192fb
commit 38b7bbbc49
161 changed files with 12615 additions and 4296 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/Cargo.toml
+35 -32
View File
@@ -16,66 +16,70 @@ workspace = true
[dependencies]
madmin.workspace = true
log.workspace = true
api = { path = "../s3select/api" }
appauth = { version = "0.0.1", path = "../appauth" }
atoi = { workspace = true }
atomic_enum = { workspace = true }
axum.workspace = true
axum-extra = { workspace = true }
axum-server = { workspace = true }
async-trait.workspace = true
bytes.workspace = true
chrono = { workspace = true }
clap.workspace = true
crypto = { path = "../crypto" }
datafusion = { workspace = true }
common.workspace = true
const-str = { version = "0.6.1", features = ["std", "proc"] }
ecstore.workspace = true
policy.workspace =true
policy.workspace = true
flatbuffers.workspace = true
futures.workspace = true
futures-util.workspace = true
h2 = "0.4.7"
hyper.workspace = true
hyper-util.workspace = true
http.workspace = true
http-body.workspace = true
iam = { path = "../iam" }
lock.workspace = true
local-ip-address = { workspace = true }
matchit = { workspace = true }
mime.workspace = true
netif.workspace = true
mime_guess = "2.0.5"
pin-project-lite.workspace = true
prost.workspace = true
prost-types.workspace = true
protos.workspace = true
protobuf.workspace = true
query = { path = "../s3select/query" }
rmp-serde.workspace = true
rustfs-obs = { workspace = true }
rustls.workspace = true
rustls-pemfile.workspace = true
rustls-pki-types.workspace = true
rust-embed = { workspace = true, features = ["interpolate-folder-path"] }
s3s.workspace = true
serde.workspace = true
serde_json.workspace = true
serde_urlencoded = { workspace = true }
shadow-rs.workspace = true
tracing.workspace = true
time = { workspace = true, features = ["parsing", "formatting", "serde"] }
tokio-util = { version = "0.7.13", features = ["io", "compat"] }
tokio-util.workspace = true
tokio = { workspace = true, features = [
"rt-multi-thread",
"macros",
"net",
"signal",
] }
tokio-rustls.workspace = true
lazy_static.workspace = true
tokio-stream.workspace = true
tonic = { version = "0.12.3", features = ["gzip"] }
tonic-reflection.workspace = true
tonic = { workspace = true }
tower.workspace = true
tracing-error.workspace = true
tracing-subscriber.workspace = true
transform-stream.workspace = true
uuid = "1.15.1"
url.workspace = true
axum.workspace = true
matchit = "0.8.6"
shadow-rs.workspace = true
const-str = { version = "0.6.1", features = ["std", "proc"] }
atoi = "2.0.0"
serde_urlencoded = "0.7.1"
crypto = { path = "../crypto" }
iam = { path = "../iam" }
jsonwebtoken = "9.3.0"
tower-http = { version = "0.6.2", features = ["cors"] }
mime_guess = "2.0.5"
rust-embed = { workspace = true, features = ["interpolate-folder-path"] }
local-ip-address = { workspace = true }
chrono = { workspace = true }
tower-http.workspace = true
uuid = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
[build-dependencies]
prost-build.workspace = true
@@ -87,14 +91,13 @@ futures-util.workspace = true
# uuid = { version = "1.8.0", features = ["v4", "fast-rng", "serde"] }
ecstore = { path = "../ecstore" }
s3s.workspace = true
clap = { version = "4.5.31", features = ["derive", "env"] }
tracing-subscriber = { version = "0.3.19", features = ["env-filter", "time"] }
hyper-util = { version = "0.1.10", features = [
clap = { workspace = true }
hyper-util = { workspace = true, features = [
"tokio",
"server-auto",
"server-graceful",
] }
transform-stream = "0.3.1"
transform-stream = { workspace = true }
netif = "0.1.6"
shadow-rs.workspace = true
# pin-utils = "0.1.0"
rustfs/README.md
+30
View File
@@ -0,0 +1,30 @@
rustfs/
├── Cargo.toml
├── src/
│ ├── main.rs # 主入口
│ ├── admin/
│ │ └── mod.rs # 管理接口
│ ├── auth/
│ │ └── mod.rs # 认证模块
│ ├── config/
│ │ ├── mod.rs # 配置模块
│ │ └── options.rs # 命令行参数
│ ├── console/
│ │ ├── mod.rs # 控制台模块
│ │ └── server.rs # 控制台服务器
│ ├── grpc/
│ │ └── mod.rs # gRPC 服务
│ ├── license/
│ │ └── mod.rs # 许可证管理
│ ├── logging/
│ │ └── mod.rs # 日志管理
│ ├── server/
│ │ ├── mod.rs # 服务器实现
│ │ ├── connection.rs # 连接处理
│ │ ├── service.rs # 服务实现
│ │ └── state.rs # 状态管理
│ ├── storage/
│ │ ├── mod.rs # 存储模块
│ │ └── fs.rs # 文件系统实现
│ └── utils/
│ └── mod.rs # 工具函数
rustfs/src/admin/handlers.rs
+184 -40
View File
@@ -1,11 +1,11 @@
use super::router::Operation;
use ::policy::policy::action::{Action, S3Action};
use ::policy::policy::resource::Resource;
use ::policy::policy::statement::BPStatement;
use ::policy::policy::{ActionSet, BucketPolicy, Effect, ResourceSet};
use crate::auth::check_key_valid;
use crate::auth::get_condition_values;
use crate::auth::get_session_token;
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;
@@ -14,14 +14,22 @@ 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 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};
@@ -40,7 +48,7 @@ use tokio_stream::wrappers::ReceiverStream;
use tracing::{error, info, warn};
pub mod group;
pub mod policy;
pub mod policys;
pub mod pools;
pub mod service_account;
pub mod sts;
@@ -59,49 +67,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();
@@ -126,8 +272,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/pools.rs
+5 -1
View File
@@ -4,6 +4,7 @@ use matchit::Params;
use s3s::{header::CONTENT_TYPE, s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
use serde::Deserialize;
use serde_urlencoded::from_bytes;
use tokio::sync::broadcast;
use tracing::warn;
use crate::{admin::router::Operation, storage::error::to_s3_error};
@@ -159,6 +160,9 @@ impl Operation for StartDecommission {
let pools: Vec<&str> = query.pool.split(",").collect();
let mut pools_indices = Vec::with_capacity(pools.len());
// TODO: ctx
let (_ctx_tx, ctx_rx) = broadcast::channel::<bool>(1);
for pool in pools.iter() {
let idx = {
if is_byid {
@@ -188,7 +192,7 @@ impl Operation for StartDecommission {
}
if !pools_indices.is_empty() {
store.decommission(pools_indices).await.map_err(to_s3_error)?;
store.decommission(ctx_rx, pools_indices).await.map_err(to_s3_error)?;
}
Ok(S3Response::new((StatusCode::OK, Body::default())))
rustfs/src/admin/handlers/service_account.rs
+153 -24
View File
@@ -1,4 +1,5 @@
use crate::admin::utils::has_space_be;
use crate::auth::{get_condition_values, get_session_token};
use crate::{admin::router::Operation, auth::check_key_valid};
use http::HeaderMap;
use hyper::StatusCode;
@@ -12,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;
@@ -29,7 +31,8 @@ impl Operation for AddServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) = check_key_valid(&req.headers, &req_cred.access_key).await?;
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;
let body = match input.store_all_unlimited().await {
@@ -89,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 {
@@ -210,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| {
@@ -278,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()
@@ -357,27 +432,55 @@ impl Operation for ListServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred 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 (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &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 {
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")
@@ -415,10 +518,13 @@ impl Operation for DeleteServiceAccount {
return Err(s3_error!(InvalidRequest, "get cred 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 (cred, owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &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() {
@@ -436,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) {
@@ -447,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/sts.rs
+3 -2
View File
@@ -42,12 +42,13 @@ impl Operation for AssumeRoleHandle {
let Some(user) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) };
let session_token = get_session_token(&req.headers);
let session_token = get_session_token(&req.uri, &req.headers);
if session_token.is_some() {
return Err(s3_error!(InvalidRequest, "AccessDenied1"));
}
let (cred, _owner) = check_key_valid(&req.headers, &user.access_key).await?;
let (cred, _owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &user.access_key).await?;
// // TODO: 判断权限, 不允许sts访问
if cred.is_temp() || cred.is_service_account() {
rustfs/src/admin/handlers/user.rs
+54 -21
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,
auth::{check_key_valid, get_condition_values, get_session_token},
};
#[derive(Debug, Deserialize, Default)]
@@ -39,7 +43,8 @@ impl Operation for AddUser {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) = check_key_valid(&req.headers, &input_cred.access_key).await?;
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();
@@ -62,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"));
}
@@ -88,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)
@@ -112,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 =
@@ -164,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 =
@@ -213,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 =
@@ -246,7 +254,8 @@ impl Operation for RemoveUser {
return Err(s3_error!(InvalidRequest, "get cred failed"));
};
let (cred, _owner) = check_key_valid(&req.headers, &input_cred.access_key).await?;
let (cred, _owner) =
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &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"))?;
@@ -275,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 =
@@ -295,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, pools,
group, policys, pools,
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/auth.rs
+25 -10
View File
@@ -1,6 +1,7 @@
use std::collections::HashMap;
use http::HeaderMap;
use http::Uri;
use iam::error::Error as IamError;
use iam::get_global_action_cred;
use iam::sys::SESSION_POLICY_NAME;
@@ -48,7 +49,7 @@ impl S3Auth for IAMAuth {
}
// 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)> {
pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result<(auth::Credentials, bool)> {
let Some(mut cred) = get_global_action_cred() else {
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
@@ -88,7 +89,7 @@ pub async fn check_key_valid(header: &HeaderMap, access_key: &str) -> S3Result<(
cred = u.credentials;
}
let claims = check_claims_from_token(header, &cred)
let claims = check_claims_from_token(session_token, &cred)
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed {}", e)))?;
cred.claims = if !claims.is_empty() { Some(claims) } else { None };
@@ -105,23 +106,21 @@ pub async fn check_key_valid(header: &HeaderMap, access_key: &str) -> S3Result<(
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();
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"));
return Err(s3_error!(InvalidRequest, "invalid token1"));
}
if !token.is_empty() && !cred.is_temp() {
return Err(s3_error!(InvalidRequest, "invalid token"));
return Err(s3_error!(InvalidRequest, "invalid token2"));
}
if !cred.is_service_account() && cred.is_temp() && token != cred.session_token {
return Err(s3_error!(InvalidRequest, "invalid token"));
return Err(s3_error!(InvalidRequest, "invalid token3"));
}
if cred.is_temp() && cred.is_expired() {
@@ -149,8 +148,10 @@ pub fn check_claims_from_token(header: &HeaderMap, cred: &auth::Credentials) ->
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_session_token<'a>(uri: &'a Uri, hds: &'a HeaderMap) -> Option<&'a str> {
hds.get("x-amz-security-token")
.map(|v| v.to_str().unwrap_or_default())
.or_else(|| get_query_param(uri.query().unwrap_or_default(), "x-amz-security-token"))
}
pub fn get_condition_values(header: &HeaderMap, cred: &auth::Credentials) -> HashMap<String, Vec<String>> {
@@ -295,3 +296,17 @@ pub fn get_condition_values(header: &HeaderMap, cred: &auth::Credentials) -> Has
args
}
pub fn get_query_param<'a>(query: &'a str, param_name: &str) -> Option<&'a str> {
let param_name = param_name.to_lowercase();
for pair in query.split('&') {
let mut parts = pair.splitn(2, '=');
if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
if key.to_lowercase() == param_name {
return Some(value);
}
}
}
None
}
rustfs/src/config/mod.rs
+59 -6
View File
@@ -1,11 +1,39 @@
use clap::Parser;
use const_str::concat;
use ecstore::global::DEFAULT_PORT;
use std::string::ToString;
shadow_rs::shadow!(build);
/// Default Access Key
/// Default value: rustfsadmin
/// Environment variable: RUSTFS_ACCESS_KEY
/// Command line argument: --access-key
/// Example: RUSTFS_ACCESS_KEY=rustfsadmin
/// Example: --access-key rustfsadmin
pub const DEFAULT_ACCESS_KEY: &str = "rustfsadmin";
/// Default Secret Key
/// Default value: rustfsadmin
/// Environment variable: RUSTFS_SECRET_KEY
/// Command line argument: --secret-key
/// Example: RUSTFS_SECRET_KEY=rustfsadmin
/// Example: --secret-key rustfsadmin
pub const DEFAULT_SECRET_KEY: &str = "rustfsadmin";
/// Default configuration file for observability
/// Default value: config/obs.toml
/// Environment variable: RUSTFS_OBS_CONFIG
/// Command line argument: --obs-config
/// Example: RUSTFS_OBS_CONFIG=config/obs.toml
/// Example: --obs-config config/obs.toml
/// Example: --obs-config /etc/rustfs/obs.toml
pub const DEFAULT_OBS_CONFIG: &str = "config/obs.toml";
/// Default TLS key for rustfs
/// This is the default key for TLS.
pub(crate) const RUSTFS_TLS_KEY: &str = "rustfs_private.key";
/// Default TLS cert for rustfs
/// This is the default cert for TLS.
pub(crate) const RUSTFS_TLS_CERT: &str = "rustfs_public.crt";
#[allow(clippy::const_is_empty)]
const SHORT_VERSION: &str = {
@@ -31,7 +59,7 @@ const LONG_VERSION: &str = concat!(
concat!("git status :\n", build::GIT_STATUS_FILE),
);
#[derive(Debug, Parser)]
#[derive(Debug, Parser, Clone)]
#[command(version = SHORT_VERSION, long_version = LONG_VERSION)]
pub struct Opt {
/// DIR points to a directory on a filesystem.
@@ -42,6 +70,7 @@ pub struct Opt {
#[arg(long, default_value_t = format!("0.0.0.0:{}", DEFAULT_PORT), env = "RUSTFS_ADDRESS")]
pub address: String,
/// Domain name used for virtual-hosted-style requests.
#[arg(long, env = "RUSTFS_SERVER_DOMAINS")]
pub server_domains: Vec<String>,
@@ -53,13 +82,37 @@ pub struct Opt {
#[arg(long, default_value_t = DEFAULT_SECRET_KEY.to_string(), env = "RUSTFS_SECRET_KEY")]
pub secret_key: String,
/// Domain name used for virtual-hosted-style requests.
#[arg(long, env = "RUSTFS_DOMAIN_NAME")]
pub domain_name: Option<String>,
#[arg(long, default_value_t = false, env = "RUSTFS_CONSOLE_ENABLE")]
pub console_enable: bool,
#[arg(long, default_value_t = format!("127.0.0.1:{}", 9002), env = "RUSTFS_CONSOLE_ADDRESS")]
pub console_address: String,
/// rustfs endpoint for console
#[arg(long, env = "RUSTFS_CONSOLE_FS_ENDPOINT")]
pub console_fs_endpoint: Option<String>,
/// Observability configuration file
/// Default value: config/obs.toml
#[arg(long, default_value_t = DEFAULT_OBS_CONFIG.to_string(), env = "RUSTFS_OBS_CONFIG")]
pub obs_config: String,
/// tls path for rustfs api and console.
#[arg(long, env = "RUSTFS_TLS_PATH")]
pub tls_path: Option<String>,
#[arg(long, env = "RUSTFS_LICENSE")]
pub license: Option<String>,
}
// lazy_static::lazy_static! {
// pub(crate) static ref OPT: OnceLock<Opt> = OnceLock::new();
// }
// pub fn init_config(opt: Opt) {
// OPT.set(opt).expect("Failed to set global config");
// }
// pub fn get_config() -> &'static Opt {
// OPT.get().expect("Global config not initialized")
// }
rustfs/src/console.rs
+178 -31
View File
@@ -1,19 +1,25 @@
use crate::config::{RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
use crate::license::get_license;
use axum::{
body::Body,
extract::Host,
http::{Response, StatusCode},
response::IntoResponse,
routing::get,
Router,
};
use axum_extra::extract::Host;
use axum_server::tls_rustls::RustlsConfig;
use http::Uri;
use mime_guess::from_path;
use rust_embed::RustEmbed;
use serde::Serialize;
use shadow_rs::shadow;
use std::net::{Ipv4Addr, SocketAddr, ToSocketAddrs};
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::OnceLock;
use tracing::info;
use std::time::Duration;
use tokio::signal;
use tracing::{debug, error, info, instrument};
shadow!(build);
@@ -153,31 +159,85 @@ pub(crate) fn init_console_cfg(local_ip: Ipv4Addr, port: u16) {
// host.parse::<SocketAddr>().is_ok() || host.parse::<IpAddr>().is_ok()
// }
async fn license_handler() -> impl IntoResponse {
let license = get_license().unwrap_or_default();
Response::builder()
.header("content-type", "application/json")
.status(StatusCode::OK)
.body(Body::from(serde_json::to_string(&license).unwrap_or_default()))
.unwrap()
}
fn _is_private_ip(ip: std::net::IpAddr) -> bool {
match ip {
std::net::IpAddr::V4(ip) => {
let octets = ip.octets();
// 10.0.0.0/8
octets[0] == 10 ||
// 172.16.0.0/12
(octets[0] == 172 && (octets[1] >= 16 && octets[1] <= 31)) ||
// 192.168.0.0/16
(octets[0] == 192 && octets[1] == 168)
}
std::net::IpAddr::V6(_) => false,
}
}
#[allow(clippy::const_is_empty)]
async fn config_handler(Host(host): Host) -> impl IntoResponse {
let host_with_port = if host.contains(':') { host } else { format!("{}:80", host) };
#[instrument(fields(host))]
async fn config_handler(uri: Uri, Host(host): Host) -> impl IntoResponse {
let scheme = uri.scheme().map(|s| s.as_str()).unwrap_or("http");
let is_addr = host_with_port
.to_socket_addrs()
.map(|addrs| {
addrs.into_iter().find(|v| {
if let SocketAddr::V4(ipv4) = v {
!ipv4.ip().is_private() && !ipv4.ip().is_loopback() && !ipv4.ip().is_unspecified()
} else {
false
}
})
})
.unwrap_or_default();
// 从 uri 中获取 host，如果没有则使用 Host extractor 的值
let host = uri.host().unwrap_or(host.as_str());
let host = if host.contains(':') {
let (host, _) = host.split_once(':').unwrap_or((host, "80"));
host
} else {
host
};
// 将当前配置复制一份
let mut cfg = CONSOLE_CONFIG.get().unwrap().clone();
let url = if let Some(addr) = is_addr {
format!("http://{}:{}", addr.ip(), cfg.port)
} else {
let (host, _) = host_with_port.split_once(':').unwrap_or_default();
format!("http://{}:{}", host, cfg.port)
};
let url = format!("{}://{}:{}", scheme, host, cfg.port);
// // 如果指定入口, 直接使用
// let url = if let Some(endpoint) = &config::get_config().console_fs_endpoint {
// debug!("axum Using rustfs endpoint address: {}", endpoint);
// endpoint.clone()
// } else {
// let host_with_port = if host.contains(':') {
// host.clone()
// } else {
// format!("{}:80", host)
// };
// // 尝试解析为 socket address，但不强制要求一定要是 IP 地址
// let socket_addr = host_with_port.to_socket_addrs().ok().and_then(|mut addrs| addrs.next());
// debug!("axum Using host with port: {}, Socket address: {:?}", host_with_port, socket_addr);
// match socket_addr {
// Some(addr) if addr.ip().is_ipv4() => {
// let ipv4 = addr.ip().to_string();
// // 如果是私有 IP、环回地址或未指定地址，保留原始域名
// if is_private_ip(addr.ip()) || addr.ip().is_loopback() || addr.ip().is_unspecified() {
// let (host, _) = host_with_port.split_once(':').unwrap_or((&host, "80"));
// debug!("axum Using private IPv4 address: {}", host);
// format!("http://{}:{}", host, cfg.port)
// } else {
// debug!("axum Using public IPv4 address");
// format!("http://{}:{}", ipv4, cfg.port)
// }
// }
// _ => {
// // 如果不是有效的 IPv4 地址，保留原始域名
// let (host, _) = host_with_port.split_once(':').unwrap_or((&host, "80"));
// debug!("axum Using domain address: {}", host);
// format!("http://{}:{}", host, cfg.port)
// }
// }
// };
cfg.api.base_url = format!("{}{}", url, RUSTFS_ADMIN_PREFIX);
cfg.s3.endpoint = url;
@@ -189,18 +249,105 @@ async fn config_handler(Host(host): Host) -> impl IntoResponse {
.unwrap()
}
pub async fn start_static_file_server(addrs: &str, local_ip: Ipv4Addr, access_key: &str, secret_key: &str) {
// 创建路由
pub async fn start_static_file_server(
addrs: &str,
local_ip: Ipv4Addr,
access_key: &str,
secret_key: &str,
tls_path: Option<String>,
) {
// Create a route
let app = Router::new()
.route("/license", get(license_handler))
.route("/config.json", get(config_handler))
.nest_service("/", get(static_handler));
let listener = tokio::net::TcpListener::bind(addrs).await.unwrap();
let local_addr = listener.local_addr().unwrap();
.fallback_service(get(static_handler));
let local_addr: SocketAddr = addrs.parse().expect("Failed to parse socket address");
info!("WebUI: http://{}:{} http://127.0.0.1:{}", local_ip, local_addr.port(), local_addr.port());
info!(" RootUser: {}", access_key);
info!(" RootPass: {}", secret_key);
axum::serve(listener, app).await.unwrap();
// Check and start the HTTPS/HTTP server
match start_server(addrs, local_addr, tls_path, app.clone()).await {
Ok(_) => info!("Server shutdown gracefully"),
Err(e) => error!("Server error: {}", e),
}
}
async fn start_server(addrs: &str, local_addr: SocketAddr, tls_path: Option<String>, app: Router) -> std::io::Result<()> {
let tls_path = tls_path.unwrap_or_default();
let key_path = format!("{}/{}", tls_path, RUSTFS_TLS_KEY);
let cert_path = format!("{}/{}", tls_path, RUSTFS_TLS_CERT);
let has_tls_certs = tokio::try_join!(tokio::fs::metadata(&key_path), tokio::fs::metadata(&cert_path)).is_ok();
debug!("Console TLS certs: {:?}", has_tls_certs);
if has_tls_certs {
debug!("Found TLS certificates, starting with HTTPS");
match tokio::try_join!(tokio::fs::read(&key_path), tokio::fs::read(&cert_path)) {
Ok((key_data, cert_data)) => {
match RustlsConfig::from_pem(cert_data, key_data).await {
Ok(config) => {
let handle = axum_server::Handle::new();
// create a signal off listening task
let handle_clone = handle.clone();
tokio::spawn(async move {
shutdown_signal().await;
handle_clone.graceful_shutdown(Some(Duration::from_secs(10)));
});
info!("Starting HTTPS server...");
axum_server::bind_rustls(local_addr, config)
.handle(handle.clone())
.serve(app.into_make_service())
.await
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
Ok(())
}
Err(e) => {
error!("Failed to create TLS config: {}", e);
start_http_server(addrs, app).await
}
}
}
Err(e) => {
error!("Failed to read TLS certificates: {}", e);
start_http_server(addrs, app).await
}
}
} else {
debug!("TLS certificates not found at {} and {}", key_path, cert_path);
start_http_server(addrs, app).await
}
}
async fn start_http_server(addrs: &str, app: Router) -> std::io::Result<()> {
debug!("Starting HTTP server...");
let listener = tokio::net::TcpListener::bind(addrs).await?;
axum::serve(listener, app)
.with_graceful_shutdown(shutdown_signal())
.await
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
}
async fn shutdown_signal() {
let ctrl_c = async {
signal::ctrl_c().await.expect("failed to install Ctrl+C handler");
};
#[cfg(unix)]
let terminate = async {
signal::unix::signal(signal::unix::SignalKind::terminate())
.expect("failed to install signal handler")
.recv()
.await;
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {
info!("shutdown_signal ctrl_c")
},
_ = terminate => {
info!("shutdown_signal terminate")
},
}
}
rustfs/src/license.rs
+62
View File
@@ -0,0 +1,62 @@
use appauth::token::Token;
use common::error::{Error, Result};
use std::sync::OnceLock;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
use tracing::error;
use tracing::info;
lazy_static::lazy_static! {
static ref LICENSE: OnceLock<Token> = OnceLock::new();
}
pub fn init_license(license: Option<String>) {
if license.is_none() {
error!("License is None");
return;
}
let license = license.unwrap();
let token = appauth::token::parse_license(&license).unwrap_or_default();
LICENSE.set(token).unwrap_or_else(|_| {
error!("Failed to set license");
});
}
pub fn get_license() -> Option<Token> {
LICENSE.get().cloned()
}
#[allow(unreachable_code)]
pub fn license_check() -> Result<()> {
return Ok(());
let inval_license = LICENSE.get().map(|token| {
if token.expired < SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() {
error!("License expired");
return Err(Error::from_string("Incorrect license, please contact RustFS.".to_string()));
}
info!("License is valid ! expired at {}", token.expired);
Ok(())
});
// let inval_license = config::get_config().license.as_ref().map(|license| {
// if license.is_empty() {
// error!("License is empty");
// return Err(Error::from_string("Incorrect license, please contact RustFS.".to_string()));
// }
// let token = appauth::token::parse_license(license)?;
// if token.expired < SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs() {
// error!("License expired");
// return Err(Error::from_string("Incorrect license, please contact RustFS.".to_string()));
// }
// info!("License is valid ! expired at {}", token.expired);
// Ok(())
// });
if inval_license.is_none() || inval_license.is_some_and(|v| v.is_err()) {
return Err(Error::from_string("Incorrect license, please contact RustFS.".to_string()));
}
Ok(())
}
rustfs/src/logging/mod.rs
+1
View File
@@ -0,0 +1 @@
rustfs/src/main.rs
+196 -65
View File
@@ -3,19 +3,24 @@ mod auth;
mod config;
mod console;
mod grpc;
pub mod license;
mod logging;
mod server;
mod service;
mod storage;
mod utils;
use crate::auth::IAMAuth;
use crate::console::{init_console_cfg, CONSOLE_CONFIG};
// Ensure the correct path for parse_license is imported
use crate::server::{wait_for_shutdown, ServiceState, ServiceStateManager, ShutdownSignal, SHUTDOWN_TIMEOUT};
use crate::utils::error;
use chrono::Datelike;
use clap::Parser;
use common::{
error::{Error, Result},
globals::set_global_addr,
};
use config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY};
use config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
use ecstore::heal::background_heal_ops::init_auto_heal;
use ecstore::utils::net::{self, get_available_port};
use ecstore::{
@@ -27,40 +32,27 @@ use ecstore::{
};
use ecstore::{global::set_global_rustfs_port, notification_sys::new_global_notification_sys};
use grpc::make_server;
use hyper_util::server::graceful::GracefulShutdown;
use hyper_util::{
rt::{TokioExecutor, TokioIo},
server::conn::auto::Builder as ConnBuilder,
service::TowerToHyperService,
};
use iam::init_iam_sys;
use license::init_license;
use protos::proto_gen::node_service::node_service_server::NodeServiceServer;
use rustfs_obs::{init_obs, load_config, set_global_guard, InitLogStatus};
use rustls::ServerConfig;
use s3s::{host::MultiDomain, service::S3ServiceBuilder};
use service::hybrid;
use std::{io::IsTerminal, net::SocketAddr};
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio::signal::unix::{signal, SignalKind};
use tokio_rustls::TlsAcceptor;
use tonic::{metadata::MetadataValue, Request, Status};
use tower_http::cors::CorsLayer;
use tracing::{debug, error, info, warn};
use tracing_error::ErrorLayer;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
fn setup_tracing() {
use tracing_subscriber::EnvFilter;
let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
let enable_color = std::io::stdout().is_terminal();
let subscriber = tracing_subscriber::fmt::fmt()
.pretty()
.with_env_filter(env_filter)
.with_ansi(enable_color)
.with_file(true)
.with_line_number(true)
.finish()
.with(ErrorLayer::default());
subscriber.try_init().expect("failed to set global default subscriber");
}
use tracing::{debug, error, info, info_span, warn};
fn check_auth(req: Request<()>) -> Result<Request<()>, Status> {
let token: MetadataValue<_> = "rustfs rpc".parse().unwrap();
@@ -83,22 +75,39 @@ fn print_server_info() {
info!("Docs: {}", cfg.doc());
}
fn main() -> Result<()> {
//解析获得到的参数
#[tokio::main]
async fn main() -> Result<()> {
// Parse the obtained parameters
let opt = config::Opt::parse();
//设置trace
setup_tracing();
// config::init_config(opt.clone());
//运行参数
run(opt)
init_license(opt.license.clone());
// Load the configuration file
let config = load_config(Some(opt.clone().obs_config));
// Initialize Observability
let (_logger, guard) = init_obs(config.clone()).await;
// Store in global storage
set_global_guard(guard)?;
// Log initialization status
InitLogStatus::init_start_log(&config.observability).await?;
// Run parameters
run(opt).await
}
#[tokio::main]
// #[tokio::main]
async fn run(opt: config::Opt) -> Result<()> {
let span = info_span!("trace-main-run");
let _enter = span.enter();
debug!("opt: {:?}", &opt);
let mut server_addr = net::check_local_server_addr(opt.address.as_str()).unwrap();
let mut server_addr = net::check_local_server_addr(opt.address.as_str())?;
if server_addr.port() == 0 {
server_addr.set_port(get_available_port());
@@ -110,18 +119,18 @@ async fn run(opt: config::Opt) -> Result<()> {
debug!("server_address {}", &server_address);
//设置AK和SK
//设置 AK 和 SK
iam::init_global_action_cred(Some(opt.access_key.clone()), Some(opt.secret_key.clone()))?;
iam::init_global_action_cred(Some(opt.access_key.clone()), Some(opt.secret_key.clone())).unwrap();
set_global_rustfs_port(server_port);
//监听地址,端口从参数中获取
//监听地址，端口从参数中获取
let listener = TcpListener::bind(server_address.clone()).await?;
//获取监听地址
let local_addr: SocketAddr = listener.local_addr()?;
let local_ip = utils::get_local_ip().ok_or(local_addr.ip()).unwrap();
// 用于rpc
// 用于 rpc
let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_address.clone().as_str(), opt.volumes.clone())
.map_err(|err| Error::from_string(err.to_string()))?;
@@ -172,15 +181,15 @@ async fn run(opt: config::Opt) -> Result<()> {
.map_err(|err| Error::from_string(err.to_string()))?;
// Setup S3 service
// 本项目使用s3s库来实现s3服务
let service = {
// 本项目使用 s3s 库来实现 s3 服务
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());
let access_key = opt.access_key.clone();
let secret_key = opt.secret_key.clone();
//显示info信息
//显示 info 信息
debug!("authentication is enabled {}, {}", &access_key, &secret_key);
b.set_auth(IAMAuth::new(access_key, secret_key));
@@ -213,50 +222,153 @@ async fn run(opt: config::Opt) -> Result<()> {
let rpc_service = NodeServiceServer::with_interceptor(make_server(), check_auth);
tokio::spawn(async move {
let hyper_service = service.into_shared();
let tls_path = opt.tls_path.clone().unwrap_or_default();
let key_path = format!("{}/{}", tls_path, RUSTFS_TLS_KEY);
let cert_path = format!("{}/{}", tls_path, RUSTFS_TLS_CERT);
let has_tls_certs = tokio::try_join!(tokio::fs::metadata(key_path.clone()), tokio::fs::metadata(cert_path.clone())).is_ok();
debug!("Main TLS certs: {:?}", has_tls_certs);
let tls_acceptor = if has_tls_certs {
debug!("Found TLS certificates, starting with HTTPS");
let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
let certs = utils::load_certs(cert_path.as_str()).map_err(|e| error(e.to_string()))?;
let key = utils::load_private_key(key_path.as_str()).map_err(|e| error(e.to_string()))?;
let mut server_config = ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certs, key)
.map_err(|e| error(e.to_string()))?;
server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
Some(TlsAcceptor::from(Arc::new(server_config)))
} else {
debug!("TLS certificates not found, starting with HTTP");
None
};
let state_manager = ServiceStateManager::new();
let worker_state_manager = state_manager.clone();
// 更新服务状态为启动中
state_manager.update(ServiceState::Starting);
// Create shutdown channel
let (shutdown_tx, mut shutdown_rx) = tokio::sync::broadcast::channel(1);
let shutdown_tx_clone = shutdown_tx.clone();
tokio::spawn(async move {
// 错误处理改进
let sigterm_inner = match signal(SignalKind::terminate()) {
Ok(signal) => signal,
Err(e) => {
error!("Failed to create SIGTERM signal handler: {}", e);
return;
}
};
let sigint_inner = match signal(SignalKind::interrupt()) {
Ok(signal) => signal,
Err(e) => {
error!("Failed to create SIGINT signal handler: {}", e);
return;
}
};
let mut sigterm_inner = sigterm_inner;
let mut sigint_inner = sigint_inner;
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());
let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
let graceful = hyper_util::server::graceful::GracefulShutdown::new();
let graceful = GracefulShutdown::new();
debug!("graceful initiated");
// 服务准备就绪
worker_state_manager.update(ServiceState::Ready);
loop {
debug!("waiting for SIGINT or SIGTERM has_tls_certs: {}", has_tls_certs);
// Wait for a connection
let (socket, _) = tokio::select! {
res = listener.accept() => {
match res {
Ok(conn) => conn,
Err(err) => {
tracing::error!("error accepting connection: {err}");
error!("error accepting connection: {err}");
continue;
}
}
}
_ = ctrl_c.as_mut() => {
info!("Ctrl-C received in worker thread");
let _ = shutdown_tx_clone.send(());
break;
}
_ = sigint_inner.recv() => {
info!("SIGINT received in worker thread");
let _ = shutdown_tx_clone.send(());
break;
}
_ = sigterm_inner.recv() => {
info!("SIGTERM received in worker thread");
let _ = shutdown_tx_clone.send(());
break;
}
_ = shutdown_rx.recv() => {
info!("Shutdown signal received in worker thread");
break;
}
};
let conn = http_server.serve_connection(TokioIo::new(socket), hybrid_service.clone());
let conn = graceful.watch(conn.into_owned());
tokio::spawn(async move {
let _ = conn.await;
});
}
tokio::select! {
() = graceful.shutdown() => {
tracing::debug!("Gracefully shutdown!");
},
() = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
tracing::debug!("Waited 10 seconds for graceful shutdown, aborting...");
if has_tls_certs {
debug!("TLS certificates found, starting with SIGINT");
let tls_socket = match tls_acceptor
.as_ref()
.ok_or_else(|| error("TLS not configured".to_string()))
.unwrap()
.accept(socket)
.await
{
Ok(tls_socket) => tls_socket,
Err(err) => {
error!("TLS handshake failed {}", err);
continue;
}
};
let conn = http_server.serve_connection(TokioIo::new(tls_socket), hybrid_service.clone());
let conn = graceful.watch(conn.into_owned());
tokio::task::spawn_blocking(move || {
tokio::runtime::Runtime::new()
.expect("Failed to create runtime")
.block_on(async move {
if let Err(err) = conn.await {
error!("Https Connection error: {}", err);
}
});
});
debug!("TLS handshake success");
} else {
debug!("Http handshake start");
let conn = http_server.serve_connection(TokioIo::new(socket), hybrid_service.clone());
let conn = graceful.watch(conn.into_owned());
tokio::spawn(async move {
if let Err(err) = conn.await {
error!("Http Connection error: {}", err);
}
});
debug!("Http handshake success");
}
}
worker_state_manager.update(ServiceState::Stopping);
tokio::select! {
() = graceful.shutdown() => {
debug!("Gracefully shutdown!");
},
() = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
debug!("Waited 10 seconds for graceful shutdown, aborting...");
}
}
worker_state_manager.update(ServiceState::Stopped);
});
// init store
@@ -268,15 +380,15 @@ async fn run(opt: config::Opt) -> Result<()> {
})?;
ECStore::init(store.clone()).await.map_err(|err| {
error!("ECStore init faild {:?}", &err);
error!("ECStore init failed {:?}", &err);
Error::from_string(err.to_string())
})?;
debug!("init store success!");
init_iam_sys(store.clone()).await.unwrap();
init_iam_sys(store.clone()).await?;
new_global_notification_sys(endpoint_pools.clone()).await.map_err(|err| {
error!("new_global_notification_sys faild {:?}", &err);
error!("new_global_notification_sys failed {:?}", &err);
Error::from_string(err.to_string())
})?;
@@ -294,17 +406,36 @@ async fn run(opt: config::Opt) -> Result<()> {
let access_key = opt.access_key.clone();
let secret_key = opt.secret_key.clone();
let console_address = opt.console_address.clone();
let tls_path = opt.tls_path.clone();
if console_address.is_empty() {
error!("console_address is empty");
return Err(Error::from_string("console_address is empty".to_string()));
}
tokio::spawn(async move {
console::start_static_file_server(&console_address, local_ip, &access_key, &secret_key).await;
console::start_static_file_server(&console_address, local_ip, &access_key, &secret_key, tls_path).await;
});
}
tokio::select! {
_ = tokio::signal::ctrl_c() => {
// Perform hibernation for 1 second
tokio::time::sleep(SHUTDOWN_TIMEOUT).await;
// listen to the shutdown signal
match wait_for_shutdown().await {
ShutdownSignal::CtrlC | ShutdownSignal::Sigint | ShutdownSignal::Sigterm => {
info!("Shutdown signal received in main thread");
// update the status to stopping first
state_manager.update(ServiceState::Stopping);
info!("Server is stopping...");
let _ = shutdown_tx.send(());
// Wait for the worker thread to complete the cleaning work
tokio::time::sleep(SHUTDOWN_TIMEOUT).await;
// the last updated status is stopped
state_manager.update(ServiceState::Stopped);
info!("Server stopped current ");
}
}
info!("server is stopped");
info!("server is stopped state: {:?}", state_manager.current_state());
Ok(())
}
rustfs/src/server/mod.rs
+6
View File
@@ -0,0 +1,6 @@
mod service_state;
pub(crate) use service_state::wait_for_shutdown;
pub(crate) use service_state::ServiceState;
pub(crate) use service_state::ServiceStateManager;
pub(crate) use service_state::ShutdownSignal;
pub(crate) use service_state::SHUTDOWN_TIMEOUT;
rustfs/src/server/service_state.rs
+152
View File
@@ -0,0 +1,152 @@
use atomic_enum::atomic_enum;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use tokio::signal::unix::{signal, SignalKind};
use tracing::info;
// a configurable shutdown timeout
pub(crate) const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1);
#[cfg(target_os = "linux")]
fn notify_systemd(state: &str) {
use libsystemd::daemon::{notify, NotifyState};
use tracing::{debug, error};
let notify_state = match state {
"ready" => NotifyState::Ready,
"stopping" => NotifyState::Stopping,
_ => {
info!("Unsupported state passed to notify_systemd: {}", state);
return;
}
};
if let Err(e) = notify(false, &[notify_state]) {
error!("Failed to notify systemd: {}", e);
} else {
debug!("Successfully notified systemd: {}", state);
}
info!("Systemd notifications are enabled on linux (state: {})", state);
}
#[cfg(not(target_os = "linux"))]
fn notify_systemd(state: &str) {
info!("Systemd notifications are not available on this platform not linux (state: {})", state);
}
#[derive(Debug)]
pub enum ShutdownSignal {
CtrlC,
Sigterm,
Sigint,
}
#[atomic_enum]
#[derive(PartialEq)]
pub(crate) enum ServiceState {
Starting,
Ready,
Stopping,
Stopped,
}
pub(crate) async fn wait_for_shutdown() -> ShutdownSignal {
let mut sigterm = signal(SignalKind::terminate()).expect("failed to create SIGTERM signal handler");
let mut sigint = signal(SignalKind::interrupt()).expect("failed to create SIGINT signal handler");
tokio::select! {
_ = tokio::signal::ctrl_c() => {
info!("Received Ctrl-C signal");
ShutdownSignal::CtrlC
}
_ = sigint.recv() => {
info!("Received SIGINT signal");
ShutdownSignal::Sigint
}
_ = sigterm.recv() => {
info!("Received SIGTERM signal");
ShutdownSignal::Sigterm
}
}
}
#[derive(Clone)]
pub(crate) struct ServiceStateManager {
state: Arc<AtomicServiceState>,
}
impl ServiceStateManager {
pub fn new() -> Self {
Self {
state: Arc::new(AtomicServiceState::new(ServiceState::Starting)),
}
}
pub fn update(&self, new_state: ServiceState) {
self.state.store(new_state, Ordering::SeqCst);
self.notify_systemd(&new_state);
}
pub fn current_state(&self) -> ServiceState {
self.state.load(Ordering::SeqCst)
}
fn notify_systemd(&self, state: &ServiceState) {
match state {
ServiceState::Starting => {
info!("Service is starting...");
#[cfg(target_os = "linux")]
if let Err(e) = libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Starting...".to_string())]) {
tracing::error!("Failed to notify systemd of starting state: {}", e);
}
}
ServiceState::Ready => {
info!("Service is ready");
notify_systemd("ready");
}
ServiceState::Stopping => {
info!("Service is stopping...");
notify_systemd("stopping");
}
ServiceState::Stopped => {
info!("Service has stopped");
#[cfg(target_os = "linux")]
if let Err(e) = libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Stopped".to_string())]) {
tracing::error!("Failed to notify systemd of stopped state: {}", e);
}
}
}
}
}
impl Default for ServiceStateManager {
fn default() -> Self {
Self::new()
}
}
// 使用示例
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_service_state_manager() {
let manager = ServiceStateManager::new();
// 初始状态应该是 Starting
assert_eq!(manager.current_state(), ServiceState::Starting);
// 更新状态到 Ready
manager.update(ServiceState::Ready);
assert_eq!(manager.current_state(), ServiceState::Ready);
// 更新状态到 Stopping
manager.update(ServiceState::Stopping);
assert_eq!(manager.current_state(), ServiceState::Stopping);
// 更新状态到 Stopped
manager.update(ServiceState::Stopped);
assert_eq!(manager.current_state(), ServiceState::Stopped);
}
}
rustfs/src/service.rs
+1 -2
View File
@@ -80,8 +80,7 @@ pin_project! {
impl<RestBody, GrpcBody> Default for HybridBody<RestBody, GrpcBody>
where
RestBody: Default,
GrpcBody: Default,
// GrpcBody: Default,
{
fn default() -> Self {
Self::Rest {
rustfs/src/storage/access.rs
+9 -2
View File
@@ -1,5 +1,6 @@
use super::ecfs::FS;
use crate::auth::{check_key_valid, get_condition_values};
use crate::auth::{check_key_valid, get_condition_values, get_session_token};
use crate::license::license_check;
use ecstore::bucket::policy_sys::PolicySys;
use iam::error::Error as IamError;
use policy::auth;
@@ -155,7 +156,8 @@ impl S3Access for FS {
// );
let (cred, is_owner) = if let Some(input_cred) = cx.credentials() {
let (cred, is_owner) = check_key_valid(cx.headers(), &input_cred.access_key).await?;
let (cred, is_owner) =
check_key_valid(get_session_token(cx.uri(), cx.headers()).unwrap_or_default(), &input_cred.access_key).await?;
(Some(cred), is_owner)
} else {
(None, false)
@@ -179,6 +181,8 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn create_bucket(&self, req: &mut S3Request<CreateBucketInput>) -> S3Result<()> {
license_check().map_err(|er| s3_error!(AccessDenied, "{:?}", er.to_string()))?;
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
req_info.bucket = Some(req.input.bucket.clone());
@@ -240,6 +244,7 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn create_multipart_upload(&self, _req: &mut S3Request<CreateMultipartUploadInput>) -> S3Result<()> {
license_check().map_err(|er| s3_error!(AccessDenied, "{:?}", er.to_string()))?;
Ok(())
}
@@ -994,6 +999,8 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn put_object(&self, req: &mut S3Request<PutObjectInput>) -> S3Result<()> {
license_check().map_err(|er| s3_error!(AccessDenied, "{:?}", er.to_string()))?;
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());
rustfs/src/storage/ecfs.rs
+117 -30
View File
@@ -4,8 +4,14 @@ use super::options::extract_metadata;
use super::options::put_opts;
use crate::auth::get_condition_values;
use crate::storage::access::ReqInfo;
use api::query::Context;
use api::query::Query;
use api::server::dbms::DatabaseManagerSystem;
use bytes::Bytes;
use common::error::Result;
use datafusion::arrow::csv::WriterBuilder as CsvWriterBuilder;
use datafusion::arrow::json::writer::JsonArray;
use datafusion::arrow::json::WriterBuilder as JsonWriterBuilder;
use ecstore::bucket::error::BucketMetadataError;
use ecstore::bucket::metadata::BUCKET_LIFECYCLE_CONFIG;
use ecstore::bucket::metadata::BUCKET_NOTIFICATION_CONFIG;
@@ -40,13 +46,13 @@ use futures::pin_mut;
use futures::{Stream, StreamExt};
use http::HeaderMap;
use lazy_static::lazy_static;
use log::warn;
use policy::auth;
use policy::policy::action::Action;
use policy::policy::action::S3Action;
use policy::policy::BucketPolicy;
use policy::policy::BucketPolicyArgs;
use policy::policy::Validator;
use query::instance::make_rustfsms;
use s3s::dto::*;
use s3s::s3_error;
use s3s::S3Error;
@@ -56,11 +62,14 @@ use s3s::S3;
use s3s::{S3Request, S3Response};
use std::fmt::Debug;
use std::str::FromStr;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tokio_util::io::ReaderStream;
use tokio_util::io::StreamReader;
use tracing::debug;
use tracing::error;
use tracing::info;
use tracing::warn;
use transform_stream::AsyncTryStream;
use uuid::Uuid;
@@ -221,7 +230,7 @@ impl S3 for FS {
#[tracing::instrument(level = "debug", skip(self, req))]
async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
let input = req.input;
// TODO: DeleteBucketInput 没有force参数？
// TODO: DeleteBucketInput 没有 force 参数？
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
@@ -252,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,
@@ -316,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,
@@ -398,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,
@@ -438,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)?;
@@ -507,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);
@@ -1045,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?;
@@ -1859,6 +1883,69 @@ impl S3 for FS {
}
Ok(S3Response::new(PutObjectAclOutput::default()))
}
async fn select_object_content(
&self,
req: S3Request<SelectObjectContentInput>,
) -> S3Result<S3Response<SelectObjectContentOutput>> {
info!("handle select_object_content");
let input = req.input;
info!("{:?}", input);
let db = make_rustfsms(input.clone(), false).await.map_err(|e| {
error!("make db failed, {}", e.to_string());
s3_error!(InternalError)
})?;
let query = Query::new(Context { input: input.clone() }, input.request.expression);
let result = db.execute(&query).await.map_err(|_| s3_error!(InternalError))?;
let results = result.result().chunk_result().await.unwrap().to_vec();
let mut buffer = Vec::new();
if input.request.output_serialization.csv.is_some() {
let mut csv_writer = CsvWriterBuilder::new().with_header(false).build(&mut buffer);
for batch in results {
csv_writer
.write(&batch)
.map_err(|e| s3_error!(InternalError, "cann't encode output to csv. e: {}", e.to_string()))?;
}
} else if input.request.output_serialization.json.is_some() {
let mut json_writer = JsonWriterBuilder::new()
.with_explicit_nulls(true)
.build::<_, JsonArray>(&mut buffer);
for batch in results {
json_writer
.write(&batch)
.map_err(|e| s3_error!(InternalError, "cann't encode output to json. e: {}", e.to_string()))?;
}
json_writer
.finish()
.map_err(|e| s3_error!(InternalError, "writer output into json error, e: {}", e.to_string()))?;
} else {
return Err(s3_error!(InvalidArgument, "unknow output format"));
}
let (tx, rx) = mpsc::channel::<S3Result<SelectObjectContentEvent>>(2);
let stream = ReceiverStream::new(rx);
tokio::spawn(async move {
let _ = tx
.send(Ok(SelectObjectContentEvent::Cont(ContinuationEvent::default())))
.await;
let _ = tx
.send(Ok(SelectObjectContentEvent::Records(RecordsEvent {
payload: Some(Bytes::from(buffer)),
})))
.await;
let _ = tx.send(Ok(SelectObjectContentEvent::End(EndEvent::default()))).await;
drop(tx);
});
Ok(S3Response::new(SelectObjectContentOutput {
payload: Some(SelectObjectContentEventStream::new(stream)),
}))
}
}
#[allow(dead_code)]
rustfs/src/utils.rs
-9
View File
@@ -1,9 +0,0 @@
use std::net::IpAddr;
pub(crate) fn get_local_ip() -> Option<std::net::Ipv4Addr> {
match local_ip_address::local_ip() {
Ok(IpAddr::V4(ip)) => Some(ip),
Err(_) => None,
Ok(IpAddr::V6(_)) => todo!(),
}
}
rustfs/src/utils/mod.rs
+37
View File
@@ -0,0 +1,37 @@
use rustls_pemfile::{certs, private_key};
use rustls_pki_types::{CertificateDer, PrivateKeyDer};
use std::net::IpAddr;
use std::{fs, io};
pub(crate) fn get_local_ip() -> Option<std::net::Ipv4Addr> {
match local_ip_address::local_ip() {
Ok(IpAddr::V4(ip)) => Some(ip),
Err(_) => None,
Ok(IpAddr::V6(_)) => todo!(),
}
}
/// Load public certificate from file.
pub(crate) fn load_certs(filename: &str) -> io::Result<Vec<CertificateDer<'static>>> {
// Open certificate file.
let cert_file = fs::File::open(filename).map_err(|e| error(format!("failed to open {}: {}", filename, e)))?;
let mut reader = io::BufReader::new(cert_file);
// Load and return certificate.
let certs = certs(&mut reader).collect::<Result<Vec<_>, _>>()?;
Ok(certs)
}
/// Load private key from file.
pub(crate) fn load_private_key(filename: &str) -> io::Result<PrivateKeyDer<'static>> {
// Open keyfile.
let keyfile = fs::File::open(filename).map_err(|e| error(format!("failed to open {}: {}", filename, e)))?;
let mut reader = io::BufReader::new(keyfile);
// Load and return a single private key.
private_key(&mut reader)?.ok_or_else(|| error(format!("no private key found in {}", filename)))
}
pub(crate) fn error(err: String) -> io::Error {
io::Error::new(io::ErrorKind::Other, err)
}
rustfs/static/readme.md
-1
View File
@@ -1 +0,0 @@
console static path, do not delete