gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into feat/kms-vault-transit2

Browse Source
This commit is contained in:
安正超
2026-04-07 19:16:28 +08:00
committed by GitHub
parent 72928a43fa 0b99e02891
commit c4b5b00d9c
39 changed files with 2609 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+112 -50
View File
@@ -203,7 +203,7 @@ version = "1.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.60.2",
]
[[package]]
@@ -214,7 +214,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
dependencies = [
"anstyle",
"once_cell_polyfill",
"windows-sys 0.61.2",
"windows-sys 0.60.2",
]
[[package]]
@@ -600,6 +600,26 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "async-tungstenite"
version = "0.34.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eb26fbd2a93308b1c1b74ac4e494a11ac10db57c476882240573bdf961463520"
dependencies = [
"atomic-waker",
"futures-core",
"futures-io",
"futures-task",
"futures-util",
"log",
"pin-project-lite",
"rustls-native-certs",
"rustls-pki-types",
"tokio",
"tokio-rustls",
"tungstenite",
]
[[package]]
name = "atoi"
version = "2.0.0"
@@ -3133,7 +3153,7 @@ dependencies = [
"libc",
"option-ext",
"redox_users",
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -3215,6 +3235,7 @@ dependencies = [
"urlencoding",
"uuid",
"walkdir",
"zip",
"zstd",
]
@@ -3412,7 +3433,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -3542,9 +3563,9 @@ dependencies = [
[[package]]
name = "flume"
version = "0.11.1"
version = "0.12.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095"
checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be"
dependencies = [
"futures-core",
"futures-sink",
@@ -4667,7 +4688,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
dependencies = [
"hermit-abi",
"libc",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -4744,7 +4765,7 @@ dependencies = [
"portable-atomic",
"portable-atomic-util",
"serde_core",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -5511,6 +5532,16 @@ dependencies = [
"uuid",
]
[[package]]
name = "mqttbytes-core-next"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4f2068d9ea771ce85a7b8caa23d9e542b10b118e99b2b16dbcac626211c9903c"
dependencies = [
"bytes",
"thiserror 2.0.18",
]
[[package]]
name = "multimap"
version = "0.10.1"
@@ -5644,7 +5675,7 @@ version = "0.50.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -7508,23 +7539,54 @@ dependencies = [
]
[[package]]
name = "rumqttc"
version = "0.25.1"
name = "rumqttc-core-next"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0feff8d882bff0b2fddaf99355a10336d43dd3ed44204f85ece28cf9626ab519"
checksum = "bad77a37d1cf72bf95973879c3b6542530889a27ced6861d98de80145ea0e871"
dependencies = [
"bytes",
"fixedbitset",
"flume",
"async-tungstenite",
"futures-io",
"futures-util",
"log",
"http 1.4.0",
"rustls-native-certs",
"rustls-pemfile",
"rustls-webpki 0.102.8",
"rustls-pki-types",
"rustls-webpki",
"thiserror 2.0.18",
"tokio",
"tokio-rustls",
]
[[package]]
name = "rumqttc-next"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "70844faf7766027294c088fdc456096cb0877eb4a346cc8119b7fc444eeb6afa"
dependencies = [
"rumqttc-v5-next",
]
[[package]]
name = "rumqttc-v5-next"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c8418a883899469a505b30076738ddf344f8c625690fb10602ea965786a094b0"
dependencies = [
"async-tungstenite",
"bytes",
"fixedbitset",
"flume",
"futures-io",
"futures-util",
"http 1.4.0",
"log",
"mqttbytes-core-next",
"rumqttc-core-next",
"rustls-native-certs",
"rustls-pki-types",
"rustls-webpki",
"thiserror 2.0.18",
"tokio",
"tokio-rustls",
"tokio-stream",
"tokio-util",
]
@@ -7722,7 +7784,7 @@ dependencies = [
"futures",
"hashbrown 0.16.1",
"metrics",
"rumqttc",
"rumqttc-next",
"rustfs-config",
"rustfs-ecstore",
"rustfs-s3-common",
@@ -8150,7 +8212,7 @@ dependencies = [
"hashbrown 0.16.1",
"quick-xml 0.39.2",
"rayon",
"rumqttc",
"rumqttc-next",
"rustc-hash",
"rustfs-config",
"rustfs-ecstore",
@@ -8454,11 +8516,13 @@ version = "0.0.5"
dependencies = [
"async-trait",
"criterion",
"hyper-rustls",
"reqwest 0.13.2",
"rumqttc",
"rumqttc-next",
"rustfs-config",
"rustfs-s3-common",
"rustfs-utils",
"rustls",
"serde",
"serde_json",
"snap",
@@ -8625,7 +8689,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys 0.12.1",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -8639,7 +8703,7 @@ dependencies = [
"once_cell",
"ring",
"rustls-pki-types",
"rustls-webpki 0.103.10",
"rustls-webpki",
"subtle",
"zeroize",
]
@@ -8656,15 +8720,6 @@ dependencies = [
"security-framework",
]
[[package]]
name = "rustls-pemfile"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50"
dependencies = [
"rustls-pki-types",
]
[[package]]
name = "rustls-pki-types"
version = "1.14.0"
@@ -8689,11 +8744,11 @@ dependencies = [
"rustls",
"rustls-native-certs",
"rustls-platform-verifier-android",
"rustls-webpki 0.103.10",
"rustls-webpki",
"security-framework",
"security-framework-sys",
"webpki-root-certs",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -8702,17 +8757,6 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
[[package]]
name = "rustls-webpki"
version = "0.102.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9"
dependencies = [
"ring",
"rustls-pki-types",
"untrusted 0.9.0",
]
[[package]]
name = "rustls-webpki"
version = "0.103.10"
@@ -8740,7 +8784,7 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
[[package]]
name = "s3s"
version = "0.14.0-dev"
source = "git+https://github.com/rustfs/s3s?rev=79bb9abe6b353025ef4572a25db8eb64f1409faf#79bb9abe6b353025ef4572a25db8eb64f1409faf"
source = "git+https://github.com/rustfs/s3s?rev=010ae03bdd8c93a9c046108190987bb0e6cb0692#010ae03bdd8c93a9c046108190987bb0e6cb0692"
dependencies = [
"arc-swap",
"arrayvec",
@@ -9369,7 +9413,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e"
dependencies = [
"libc",
"windows-sys 0.61.2",
"windows-sys 0.60.2",
]
[[package]]
@@ -9698,7 +9742,7 @@ dependencies = [
"getrandom 0.4.2",
"once_cell",
"rustix 1.1.4",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -10234,6 +10278,24 @@ version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
[[package]]
name = "tungstenite"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8"
dependencies = [
"bytes",
"data-encoding",
"http 1.4.0",
"httparse",
"log",
"rand 0.9.2",
"rustls",
"rustls-pki-types",
"sha1 0.10.6",
"thiserror 2.0.18",
]
[[package]]
name = "twox-hash"
version = "2.1.2"
@@ -10653,7 +10715,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
Cargo.toml
+2 -2
View File
@@ -248,11 +248,11 @@ rayon = "1.11.0"
reed-solomon-erasure = { version = "6.0", default-features = false, features = ["std", "simd-accel"] }
reed-solomon-simd = "3.1.0"
regex = { version = "1.12.3" }
rumqttc = { version = "0.25.1" }
rumqttc = { package = "rumqttc-next", version = "0.29.0", features = ["websocket"] }
rustix = { version = "1.1.4", features = ["fs"] }
rust-embed = { version = "8.11.0" }
rustc-hash = { version = "2.1.2" }
s3s = { git = "https://github.com/rustfs/s3s", rev = "79bb9abe6b353025ef4572a25db8eb64f1409faf", features = ["minio"] }
s3s = { git = "https://github.com/rustfs/s3s", rev = "010ae03bdd8c93a9c046108190987bb0e6cb0692", features = ["minio"] }
serial_test = "3.4.0"
shadow-rs = { version = "1.7.1", default-features = false }
siphasher = "1.0.2"
crates/audit/src/factory.rs
+23 -8
View File
@@ -19,7 +19,8 @@ use rumqttc::QoS;
use rustfs_config::audit::{AUDIT_MQTT_KEYS, AUDIT_WEBHOOK_KEYS, ENV_AUDIT_MQTT_KEYS, ENV_AUDIT_WEBHOOK_KEYS};
use rustfs_config::{
AUDIT_DEFAULT_DIR, DEFAULT_LIMIT, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR,
MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT,
MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_POLICY,
MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT,
WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR,
WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY,
};
@@ -27,7 +28,10 @@ use rustfs_ecstore::config::KVS;
use rustfs_targets::{
Target,
error::TargetError,
target::{mqtt::MQTTArgs, webhook::WebhookArgs},
target::{
mqtt::{MQTTArgs, MQTTTlsConfig, validate_mqtt_broker_url},
webhook::WebhookArgs,
},
};
use std::time::Duration;
use tracing::{debug, warn};
@@ -166,6 +170,14 @@ impl TargetFactory for MQTTTargetFactory {
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs)
.unwrap_or_else(|| Duration::from_secs(30)),
tls: MQTTTlsConfig::from_values(
config.lookup(MQTT_TLS_POLICY).as_deref(),
config.lookup(MQTT_TLS_CA).as_deref(),
config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(),
config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(),
config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(),
config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(),
)?,
queue_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or(AUDIT_DEFAULT_DIR.to_string()),
queue_limit: config
.lookup(MQTT_QUEUE_LIMIT)
@@ -185,12 +197,15 @@ impl TargetFactory for MQTTTargetFactory {
let url = Url::parse(&broker)
.map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?;
match url.scheme() {
"tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" => {}
_ => {
return Err(TargetError::Configuration("Unsupported broker URL scheme".to_string()));
}
}
let tls = MQTTTlsConfig::from_values(
config.lookup(MQTT_TLS_POLICY).as_deref(),
config.lookup(MQTT_TLS_CA).as_deref(),
config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(),
config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(),
config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(),
config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(),
)?;
validate_mqtt_broker_url(&url, &tls)?;
if config.lookup(MQTT_TOPIC).is_none() {
return Err(TargetError::Configuration("Missing MQTT topic".to_string()));
crates/audit/tests/config_parsing_test.rs
+6
View File
@@ -53,6 +53,12 @@ fn test_mqtt_valid_fields() {
"reconnect_interval",
"queue_dir",
"queue_limit",
"tls_policy",
"tls_ca",
"tls_client_cert",
"tls_client_key",
"tls_trust_leaf_as_ca",
"ws_path_allowlist",
];
// This tests the MQTT configuration fields we support
crates/config/src/audit/mqtt.rs
+19 -1
View File
@@ -23,9 +23,15 @@ pub const ENV_AUDIT_MQTT_RECONNECT_INTERVAL: &str = "RUSTFS_AUDIT_MQTT_RECONNECT
pub const ENV_AUDIT_MQTT_KEEP_ALIVE_INTERVAL: &str = "RUSTFS_AUDIT_MQTT_KEEP_ALIVE_INTERVAL";
pub const ENV_AUDIT_MQTT_QUEUE_DIR: &str = "RUSTFS_AUDIT_MQTT_QUEUE_DIR";
pub const ENV_AUDIT_MQTT_QUEUE_LIMIT: &str = "RUSTFS_AUDIT_MQTT_QUEUE_LIMIT";
pub const ENV_AUDIT_MQTT_TLS_POLICY: &str = "RUSTFS_AUDIT_MQTT_TLS_POLICY";
pub const ENV_AUDIT_MQTT_TLS_CA: &str = "RUSTFS_AUDIT_MQTT_TLS_CA";
pub const ENV_AUDIT_MQTT_TLS_CLIENT_CERT: &str = "RUSTFS_AUDIT_MQTT_TLS_CLIENT_CERT";
pub const ENV_AUDIT_MQTT_TLS_CLIENT_KEY: &str = "RUSTFS_AUDIT_MQTT_TLS_CLIENT_KEY";
pub const ENV_AUDIT_MQTT_TLS_TRUST_LEAF_AS_CA: &str = "RUSTFS_AUDIT_MQTT_TLS_TRUST_LEAF_AS_CA";
pub const ENV_AUDIT_MQTT_WS_PATH_ALLOWLIST: &str = "RUSTFS_AUDIT_MQTT_WS_PATH_ALLOWLIST";
/// A list of all valid configuration keys for an MQTT target.
pub const ENV_AUDIT_MQTT_KEYS: &[&str; 10] = &[
pub const ENV_AUDIT_MQTT_KEYS: &[&str; 16] = &[
ENV_AUDIT_MQTT_ENABLE,
ENV_AUDIT_MQTT_BROKER,
ENV_AUDIT_MQTT_TOPIC,
@@ -36,6 +42,12 @@ pub const ENV_AUDIT_MQTT_KEYS: &[&str; 10] = &[
ENV_AUDIT_MQTT_KEEP_ALIVE_INTERVAL,
ENV_AUDIT_MQTT_QUEUE_DIR,
ENV_AUDIT_MQTT_QUEUE_LIMIT,
ENV_AUDIT_MQTT_TLS_POLICY,
ENV_AUDIT_MQTT_TLS_CA,
ENV_AUDIT_MQTT_TLS_CLIENT_CERT,
ENV_AUDIT_MQTT_TLS_CLIENT_KEY,
ENV_AUDIT_MQTT_TLS_TRUST_LEAF_AS_CA,
ENV_AUDIT_MQTT_WS_PATH_ALLOWLIST,
];
/// A list of all valid configuration keys for an MQTT target.
@@ -50,5 +62,11 @@ pub const AUDIT_MQTT_KEYS: &[&str] = &[
crate::MQTT_KEEP_ALIVE_INTERVAL,
crate::MQTT_QUEUE_DIR,
crate::MQTT_QUEUE_LIMIT,
crate::MQTT_TLS_POLICY,
crate::MQTT_TLS_CA,
crate::MQTT_TLS_CLIENT_CERT,
crate::MQTT_TLS_CLIENT_KEY,
crate::MQTT_TLS_TRUST_LEAF_AS_CA,
crate::MQTT_WS_PATH_ALLOWLIST,
crate::COMMENT_KEY,
];
crates/config/src/constants/targets.rs
+6
View File
@@ -34,6 +34,12 @@ pub const MQTT_RECONNECT_INTERVAL: &str = "reconnect_interval";
pub const MQTT_KEEP_ALIVE_INTERVAL: &str = "keep_alive_interval";
pub const MQTT_QUEUE_DIR: &str = "queue_dir";
pub const MQTT_QUEUE_LIMIT: &str = "queue_limit";
pub const MQTT_TLS_POLICY: &str = "tls_policy";
pub const MQTT_TLS_CA: &str = "tls_ca";
pub const MQTT_TLS_CLIENT_CERT: &str = "tls_client_cert";
pub const MQTT_TLS_CLIENT_KEY: &str = "tls_client_key";
pub const MQTT_TLS_TRUST_LEAF_AS_CA: &str = "tls_trust_leaf_as_ca";
pub const MQTT_WS_PATH_ALLOWLIST: &str = "ws_path_allowlist";
/// Environment variable controlling whether target queue files are Snappy-compressed.
/// Applies to both notify and audit target queue stores.
crates/config/src/notify/mqtt.rs
+19 -1
View File
@@ -24,6 +24,12 @@ pub const NOTIFY_MQTT_KEYS: &[&str] = &[
crate::MQTT_KEEP_ALIVE_INTERVAL,
crate::MQTT_QUEUE_DIR,
crate::MQTT_QUEUE_LIMIT,
crate::MQTT_TLS_POLICY,
crate::MQTT_TLS_CA,
crate::MQTT_TLS_CLIENT_CERT,
crate::MQTT_TLS_CLIENT_KEY,
crate::MQTT_TLS_TRUST_LEAF_AS_CA,
crate::MQTT_WS_PATH_ALLOWLIST,
crate::COMMENT_KEY,
];
@@ -38,8 +44,14 @@ pub const ENV_NOTIFY_MQTT_RECONNECT_INTERVAL: &str = "RUSTFS_NOTIFY_MQTT_RECONNE
pub const ENV_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL: &str = "RUSTFS_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL";
pub const ENV_NOTIFY_MQTT_QUEUE_DIR: &str = "RUSTFS_NOTIFY_MQTT_QUEUE_DIR";
pub const ENV_NOTIFY_MQTT_QUEUE_LIMIT: &str = "RUSTFS_NOTIFY_MQTT_QUEUE_LIMIT";
pub const ENV_NOTIFY_MQTT_TLS_POLICY: &str = "RUSTFS_NOTIFY_MQTT_TLS_POLICY";
pub const ENV_NOTIFY_MQTT_TLS_CA: &str = "RUSTFS_NOTIFY_MQTT_TLS_CA";
pub const ENV_NOTIFY_MQTT_TLS_CLIENT_CERT: &str = "RUSTFS_NOTIFY_MQTT_TLS_CLIENT_CERT";
pub const ENV_NOTIFY_MQTT_TLS_CLIENT_KEY: &str = "RUSTFS_NOTIFY_MQTT_TLS_CLIENT_KEY";
pub const ENV_NOTIFY_MQTT_TLS_TRUST_LEAF_AS_CA: &str = "RUSTFS_NOTIFY_MQTT_TLS_TRUST_LEAF_AS_CA";
pub const ENV_NOTIFY_MQTT_WS_PATH_ALLOWLIST: &str = "RUSTFS_NOTIFY_MQTT_WS_PATH_ALLOWLIST";
pub const ENV_NOTIFY_MQTT_KEYS: &[&str; 10] = &[
pub const ENV_NOTIFY_MQTT_KEYS: &[&str; 16] = &[
ENV_NOTIFY_MQTT_ENABLE,
ENV_NOTIFY_MQTT_BROKER,
ENV_NOTIFY_MQTT_TOPIC,
@@ -50,4 +62,10 @@ pub const ENV_NOTIFY_MQTT_KEYS: &[&str; 10] = &[
ENV_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL,
ENV_NOTIFY_MQTT_QUEUE_DIR,
ENV_NOTIFY_MQTT_QUEUE_LIMIT,
ENV_NOTIFY_MQTT_TLS_POLICY,
ENV_NOTIFY_MQTT_TLS_CA,
ENV_NOTIFY_MQTT_TLS_CLIENT_CERT,
ENV_NOTIFY_MQTT_TLS_CLIENT_KEY,
ENV_NOTIFY_MQTT_TLS_TRUST_LEAF_AS_CA,
ENV_NOTIFY_MQTT_WS_PATH_ALLOWLIST,
];
crates/e2e_test/Cargo.toml
+1
View File
@@ -80,3 +80,4 @@ suppaftp = { workspace = true, features = ["tokio", "rustls-aws-lc-rs"] }
rcgen.workspace = true
anyhow.workspace = true
rustls.workspace = true
zip.workspace = true
crates/e2e_test/src/archive_download_integrity_test.rs
+329
View File
@@ -0,0 +1,329 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#[cfg(test)]
mod tests {
use crate::common::{RustFSTestEnvironment, init_logging, rustfs_binary_path};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use http::header::{CONTENT_TYPE, HOST};
use reqwest::StatusCode;
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
use rustfs_signer::{pre_sign_v4, sign_v4};
use s3s::Body;
use serial_test::serial;
use sha2::{Digest, Sha256};
use std::error::Error;
use std::io::{Cursor, Write};
use std::process::Command;
use time::OffsetDateTime;
use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions};
const ARCHIVE_TEST_BUCKET: &str = "archive-download-integrity";
const MULTIPART_ARCHIVE_TEST_BUCKET: &str = "archive-multipart-integrity";
const MULTIPART_PART_SIZE: usize = 5 * 1024 * 1024;
fn build_zip_bytes(files: &[(&str, &[u8])]) -> Result<Vec<u8>, Box<dyn Error + Send + Sync>> {
let cursor = Cursor::new(Vec::new());
let mut zip = ZipWriter::new(cursor);
let options = SimpleFileOptions::default().compression_method(CompressionMethod::Stored);
for (name, content) in files {
zip.start_file(*name, options)?;
zip.write_all(content)?;
}
Ok(zip.finish()?.into_inner())
}
fn random_bytes(size: usize) -> Vec<u8> {
(0..size).map(|idx| (idx % 251) as u8).collect()
}
async fn start_rustfs_server_with_env(
env: &mut RustFSTestEnvironment,
extra_env: &[(&str, &str)],
) -> Result<(), Box<dyn Error + Send + Sync>> {
let binary_path = rustfs_binary_path();
let mut command = Command::new(&binary_path);
command.env("RUST_LOG", "rustfs=info,rustfs_notify=debug");
for (key, value) in extra_env {
command.env(key, value);
}
let process = command
.args([
"--address",
&env.address,
"--access-key",
&env.access_key,
"--secret-key",
&env.secret_key,
&env.temp_dir,
])
.spawn()?;
env.process = Some(process);
env.wait_for_server_ready().await?;
Ok(())
}
async fn presigned_get_request_with_accept_encoding(
url: &str,
access_key: &str,
secret_key: &str,
accept_encoding: &str,
) -> Result<reqwest::Response, Box<dyn Error + Send + Sync>> {
let uri = url.parse::<http::Uri>()?;
let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
let signed = pre_sign_v4(
http::Request::builder()
.method(http::Method::GET)
.uri(uri)
.header(HOST, authority)
.body(Body::empty())?,
access_key,
secret_key,
"",
"us-east-1",
600,
OffsetDateTime::now_utc(),
);
let client = reqwest::Client::builder()
.no_proxy()
.no_gzip()
.no_brotli()
.no_zstd()
.no_deflate()
.build()?;
Ok(client
.get(signed.uri().to_string())
.header("Accept-Encoding", accept_encoding)
.send()
.await?)
}
async fn signed_put_request_with_headers(
url: &str,
access_key: &str,
secret_key: &str,
body: Vec<u8>,
content_type: &str,
content_encoding: &str,
) -> Result<reqwest::Response, Box<dyn Error + Send + Sync>> {
let uri = url.parse::<http::Uri>()?;
let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
let request = http::Request::builder()
.method(http::Method::PUT)
.uri(uri)
.header(HOST, authority)
.header(CONTENT_TYPE, content_type)
.header("content-encoding", content_encoding)
.header("x-amz-content-sha256", UNSIGNED_PAYLOAD)
.body(Body::empty())?;
let signed = sign_v4(request, body.len() as i64, access_key, secret_key, "", "us-east-1");
let client = reqwest::Client::builder().no_proxy().build()?;
let mut builder = client.put(url).body(body);
for (name, value) in signed.headers() {
builder = builder.header(name, value);
}
Ok(builder.send().await?)
}
#[tokio::test]
#[serial]
async fn test_archive_put_rejects_content_encoding_by_default() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server_without_cleanup(vec![]).await?;
env.create_test_bucket(ARCHIVE_TEST_BUCKET).await?;
let zip_bytes = build_zip_bytes(&[("alpha.txt", b"archive-body")])?;
let object_url = format!("{}/{}/{}", env.url, ARCHIVE_TEST_BUCKET, "bundle.zip");
let response =
signed_put_request_with_headers(&object_url, &env.access_key, &env.secret_key, zip_bytes, "application/zip", "gzip")
.await?;
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let body = response.text().await?;
assert!(
body.contains("InvalidArgument") || body.contains("Content-Encoding"),
"unexpected error body: {body}"
);
env.stop_server();
Ok(())
}
#[tokio::test]
#[serial]
async fn test_archive_download_roundtrip_with_http_compression_enabled() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
start_rustfs_server_with_env(
&mut env,
&[
("RUSTFS_COMPRESS_ENABLE", "on"),
("RUSTFS_COMPRESS_MIME_TYPES", "text/*,application/json,application/zip"),
("RUSTFS_COMPRESS_MIN_SIZE", "1"),
],
)
.await?;
env.create_test_bucket(ARCHIVE_TEST_BUCKET).await?;
let client = env.create_s3_client();
let zip_bytes = build_zip_bytes(&[
("docs/readme.txt", b"archive-download-integrity"),
("docs/notes.txt", b"response-compression-must-not-alter-zip-bytes"),
])?;
let expected_sha256 = Sha256::digest(&zip_bytes);
client
.put_object()
.bucket(ARCHIVE_TEST_BUCKET)
.key("bundle.zip")
.content_type("application/zip")
.body(ByteStream::from(zip_bytes.clone()))
.send()
.await?;
let object_url = format!("{}/{}/{}", env.url, ARCHIVE_TEST_BUCKET, "bundle.zip");
let response =
presigned_get_request_with_accept_encoding(&object_url, &env.access_key, &env.secret_key, "gzip, br, zstd").await?;
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
.get("content-encoding")
.and_then(|value| value.to_str().ok()),
None,
"archive download must not be HTTP-compressed"
);
let downloaded = response.bytes().await?;
assert_eq!(
downloaded.as_ref(),
zip_bytes.as_slice(),
"downloaded archive bytes must match uploaded bytes"
);
assert_eq!(
Sha256::digest(downloaded.as_ref()).as_slice(),
expected_sha256.as_slice(),
"archive SHA256 mismatch"
);
env.stop_server();
Ok(())
}
#[tokio::test]
#[serial]
async fn test_archive_multipart_roundtrip_preserves_bytes() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server_without_cleanup(vec![]).await?;
env.create_test_bucket(MULTIPART_ARCHIVE_TEST_BUCKET).await?;
let client = env.create_s3_client();
let payload = random_bytes(MULTIPART_PART_SIZE + 512 * 1024);
let zip_bytes = build_zip_bytes(&[("payload.bin", payload.as_slice())])?;
assert!(zip_bytes.len() > MULTIPART_PART_SIZE, "zip payload must exceed multipart threshold");
let expected_sha256 = Sha256::digest(&zip_bytes);
let create_output = client
.create_multipart_upload()
.bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
.key("multipart-bundle.zip")
.content_type("application/zip")
.send()
.await?;
let upload_id = create_output.upload_id().expect("multipart upload id");
let first_part = zip_bytes[..MULTIPART_PART_SIZE].to_vec();
let second_part = zip_bytes[MULTIPART_PART_SIZE..].to_vec();
let upload_part_1 = client
.upload_part()
.bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
.key("multipart-bundle.zip")
.upload_id(upload_id)
.part_number(1)
.body(ByteStream::from(first_part))
.send()
.await?;
let upload_part_2 = client
.upload_part()
.bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
.key("multipart-bundle.zip")
.upload_id(upload_id)
.part_number(2)
.body(ByteStream::from(second_part))
.send()
.await?;
let completed_upload = CompletedMultipartUpload::builder()
.parts(
CompletedPart::builder()
.part_number(1)
.e_tag(upload_part_1.e_tag().unwrap_or_default())
.build(),
)
.parts(
CompletedPart::builder()
.part_number(2)
.e_tag(upload_part_2.e_tag().unwrap_or_default())
.build(),
)
.build();
client
.complete_multipart_upload()
.bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
.key("multipart-bundle.zip")
.upload_id(upload_id)
.multipart_upload(completed_upload)
.send()
.await?;
let downloaded = client
.get_object()
.bucket(MULTIPART_ARCHIVE_TEST_BUCKET)
.key("multipart-bundle.zip")
.send()
.await?
.body
.collect()
.await?
.into_bytes();
assert_eq!(
downloaded.as_ref(),
zip_bytes.as_slice(),
"multipart archive bytes must match uploaded bytes"
);
assert_eq!(
Sha256::digest(downloaded.as_ref()).as_slice(),
expected_sha256.as_slice(),
"multipart archive SHA256 mismatch"
);
env.stop_server();
Ok(())
}
}
crates/e2e_test/src/common.rs
+20 -4
View File
@@ -622,6 +622,7 @@ pub struct RustFSTestClusterEnvironment {
pub temp_dir: String,
pub access_key: String,
pub secret_key: String,
pub extra_env: Vec<(String, String)>,
}
impl RustFSTestClusterEnvironment {
@@ -670,9 +671,19 @@ impl RustFSTestClusterEnvironment {
temp_dir,
access_key: DEFAULT_ACCESS_KEY.to_string(),
secret_key: DEFAULT_SECRET_KEY.to_string(),
extra_env: Vec::new(),
})
}
/// Add an extra environment variable applied to every cluster node process.
pub fn set_env<K, V>(&mut self, key: K, value: V)
where
K: Into<String>,
V: Into<String>,
{
self.extra_env.push((key.into(), value.into()));
}
/// Build the volumes argument string for RustFS binary (internal helper method).
///
/// Concatenates the address and data directory of all cluster nodes into a single string
@@ -703,15 +714,20 @@ impl RustFSTestClusterEnvironment {
for (i, node) in self.nodes.iter_mut().enumerate() {
info!("Starting cluster node {} on {}", i, node.address);
let process = Command::new(&binary_path)
let mut command = Command::new(&binary_path);
command
.env("RUSTFS_VOLUMES", &volumes_arg)
.env("RUSTFS_ADDRESS", &node.address)
.env("RUSTFS_ACCESS_KEY", &self.access_key)
.env("RUSTFS_SECRET_KEY", &self.secret_key)
.env("RUSTFS_CONSOLE_ENABLE", "false")
.env("RUST_LOG", "rustfs=info,rustfs_notify=debug")
.current_dir(&node.data_dir)
.spawn()?;
.env("RUST_LOG", "rustfs=info,rustfs_notify=debug");
for (key, value) in &self.extra_env {
command.env(key, value);
}
let process = command.current_dir(&node.data_dir).spawn()?;
node.process = Some(process);
}
crates/e2e_test/src/lib.rs
+6
View File
@@ -56,6 +56,9 @@ mod special_chars_test;
#[cfg(test)]
mod content_encoding_test;
#[cfg(test)]
mod archive_download_integrity_test;
// ListObjectsV2 pagination test (Issue #1596)
#[cfg(test)]
mod list_objects_v2_pagination_test;
@@ -113,6 +116,9 @@ mod bucket_logging_test;
#[cfg(test)]
mod multipart_auth_test;
#[cfg(test)]
mod stale_multipart_cleanup_cluster_test;
// Object lambda end-to-end regression tests
#[cfg(test)]
mod object_lambda_test;
crates/e2e_test/src/stale_multipart_cleanup_cluster_test.rs
+155
View File
@@ -0,0 +1,155 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::common::{RustFSTestClusterEnvironment, init_logging};
use aws_sdk_s3::error::SdkError;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::CompletedMultipartUpload;
use serial_test::serial;
use tokio::time::{Duration, sleep};
use tracing::info;
use uuid::Uuid;
const CLEANUP_BUCKET: &str = "stale-multipart-cleanup-cluster";
async fn list_parts_reports_missing_upload(
client: &aws_sdk_s3::Client,
bucket: &str,
key: &str,
upload_id: &str,
) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
let result = client.list_parts().bucket(bucket).key(key).upload_id(upload_id).send().await;
match result {
Ok(_) => Ok(false),
Err(SdkError::ServiceError(err)) => {
let code = err.err().meta().code().unwrap_or("");
if code == "NoSuchUpload" {
Ok(true)
} else {
Err(format!("unexpected list_parts service error: code={code}, err={err:?}").into())
}
}
Err(err) => Err(format!("unexpected list_parts error: {err:?}").into()),
}
}
async fn complete_reports_missing_upload(
client: &aws_sdk_s3::Client,
bucket: &str,
key: &str,
upload_id: &str,
) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
let result = client
.complete_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(upload_id)
.multipart_upload(CompletedMultipartUpload::builder().build())
.send()
.await;
match result {
Ok(_) => Ok(false),
Err(SdkError::ServiceError(err)) => {
let code = err.err().meta().code().unwrap_or("");
if code == "NoSuchUpload" {
Ok(true)
} else {
Err(format!("unexpected complete_multipart_upload service error: code={code}, err={err:?}").into())
}
}
Err(err) => Err(format!("unexpected complete_multipart_upload error: {err:?}").into()),
}
}
async fn wait_for_cleanup_on_all_nodes(
clients: &[aws_sdk_s3::Client],
bucket: &str,
key: &str,
upload_id: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
for attempt in 0..30 {
let mut all_cleaned = true;
for (idx, client) in clients.iter().enumerate() {
let list_parts_missing = list_parts_reports_missing_upload(client, bucket, key, upload_id).await?;
let complete_missing = complete_reports_missing_upload(client, bucket, key, upload_id).await?;
if !(list_parts_missing && complete_missing) {
info!("stale multipart still visible on node {} at attempt {}", idx, attempt + 1);
all_cleaned = false;
break;
}
}
if all_cleaned {
return Ok(());
}
sleep(Duration::from_secs(1)).await;
}
Err("stale multipart upload was not cleaned up on all nodes within timeout".into())
}
#[tokio::test]
#[serial]
async fn test_stale_multipart_cleanup_removes_incomplete_upload_across_cluster()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
cluster.set_env("RUSTFS_API_STALE_UPLOADS_EXPIRY", "5s");
cluster.set_env("RUSTFS_API_STALE_UPLOADS_CLEANUP_INTERVAL", "1s");
cluster.start().await?;
cluster.create_test_bucket(CLEANUP_BUCKET).await?;
let clients = cluster.create_all_clients()?;
let key = format!("multipart/stale-{}.txt", Uuid::new_v4().simple());
let create_output = clients[0]
.create_multipart_upload()
.bucket(CLEANUP_BUCKET)
.key(&key)
.send()
.await?;
let upload_id = create_output
.upload_id()
.ok_or("create_multipart_upload response missing upload_id")?
.to_string();
clients[1]
.upload_part()
.bucket(CLEANUP_BUCKET)
.key(&key)
.upload_id(&upload_id)
.part_number(1)
.body(ByteStream::from_static(b"stale multipart part"))
.send()
.await?;
let parts_before_cleanup = clients[2]
.list_parts()
.bucket(CLEANUP_BUCKET)
.key(&key)
.upload_id(&upload_id)
.send()
.await?;
assert_eq!(
parts_before_cleanup.parts().len(),
1,
"multipart upload should be visible before background cleanup"
);
wait_for_cleanup_on_all_nodes(&clients, CLEANUP_BUCKET, &key, &upload_id).await?;
Ok(())
}
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+699 -7
View File
@@ -19,22 +19,28 @@
#![allow(clippy::all)]
use crate::bucket::lifecycle::bucket_lifecycle_audit::{LcAuditEvent, LcEventSrc};
use crate::bucket::lifecycle::lifecycle::{self, ExpirationOptions, Lifecycle, TransitionOptions};
use crate::bucket::lifecycle::lifecycle::{
self, ExpirationOptions, Lifecycle, ObjectOpts, TransitionOptions, abort_incomplete_multipart_upload_due,
};
use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats};
use crate::bucket::lifecycle::tier_sweeper::{Jentry, delete_object_from_remote_tier};
use crate::bucket::object_lock::objectlock_sys::check_object_lock_for_deletion;
use crate::bucket::{metadata_sys::get_lifecycle_config, versioning_sys::BucketVersioningSys};
use crate::bucket::{metadata_sys, metadata_sys::get_lifecycle_config, versioning_sys::BucketVersioningSys};
use crate::client::object_api_utils::new_getobjectreader;
use crate::disk::error::DiskError;
use crate::disk::{DeleteOptions, Disk, DiskAPI, RUSTFS_META_MULTIPART_BUCKET, STORAGE_FORMAT_FILE};
use crate::error::Error;
use crate::error::StorageError;
use crate::error::{error_resp_to_object_err, is_err_object_not_found, is_err_version_not_found, is_network_or_host_down};
use crate::event_notification::{EventArgs, send_event};
use crate::global::GLOBAL_LocalNodeName;
use crate::global::{GLOBAL_LifecycleSys, GLOBAL_TierConfigMgr, get_global_deployment_id};
use crate::set_disk::{MAX_PARTS_COUNT, RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY, SetDisks};
use crate::store::ECStore;
use crate::store_api::StorageAPI;
use crate::store_api::{
GetObjectReader, HTTPRangeSpec, ListOperations, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete,
GetObjectReader, HTTPRangeSpec, ListOperations, MultipartOperations, ObjectInfo, ObjectOperations, ObjectOptions,
ObjectToDelete,
};
use crate::tier::warm_backend::WarmBackendGetOpts;
use async_channel::{Receiver as A_Receiver, Sender as A_Sender, bounded};
@@ -45,7 +51,7 @@ use lazy_static::lazy_static;
use rustfs_common::data_usage::TierStats;
use rustfs_common::heal_channel::rep_has_active_rules;
use rustfs_common::metrics::{IlmAction, Metrics};
use rustfs_filemeta::{FileInfo, NULL_VERSION_ID, RestoreStatusOps, is_restored_object_on_disk};
use rustfs_filemeta::{FileInfo, FileInfoOpts, NULL_VERSION_ID, RestoreStatusOps, get_file_info, is_restored_object_on_disk};
use rustfs_s3_common::EventName;
use rustfs_utils::{get_env_i64, get_env_usize, path::encode_dir_object, string::strings_has_prefix_fold};
use s3s::Body;
@@ -61,7 +67,8 @@ use std::env;
use std::io::Write;
use std::pin::Pin;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, Mutex};
use std::sync::{Arc, Mutex, Weak};
use std::time::Duration as StdDuration;
use time::OffsetDateTime;
use tokio::select;
use tokio::sync::mpsc::{Receiver, Sender};
@@ -84,6 +91,10 @@ pub const AMZ_ENCRYPTION_AES: &str = "AES256";
pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms";
pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier.";
const ENV_STALE_UPLOADS_EXPIRY: &str = "RUSTFS_API_STALE_UPLOADS_EXPIRY";
const ENV_STALE_UPLOADS_CLEANUP_INTERVAL: &str = "RUSTFS_API_STALE_UPLOADS_CLEANUP_INTERVAL";
const DEFAULT_STALE_UPLOADS_EXPIRY: StdDuration = StdDuration::from_secs(24 * 60 * 60);
const DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL: StdDuration = StdDuration::from_secs(6 * 60 * 60);
lazy_static! {
pub static ref GLOBAL_ExpiryState: Arc<RwLock<ExpiryState>> = ExpiryState::new();
@@ -686,6 +697,334 @@ pub async fn init_background_expiry(api: Arc<ECStore>) {
ExpiryState::resize_workers(workers, api).await;
}
#[derive(Debug, Clone)]
struct StaleMultipartUploadCandidate {
path: String,
initiated: OffsetDateTime,
metadata: Option<HashMap<String, String>>,
}
fn parse_stale_uploads_duration(env_key: &str, default: StdDuration) -> StdDuration {
env::var(env_key)
.ok()
.and_then(|value| rustfs_madmin::utils::parse_duration(&value).ok())
.filter(|duration| !duration.is_zero())
.unwrap_or(default)
}
fn stale_uploads_expiry() -> StdDuration {
parse_stale_uploads_duration(ENV_STALE_UPLOADS_EXPIRY, DEFAULT_STALE_UPLOADS_EXPIRY)
}
fn stale_uploads_cleanup_interval() -> StdDuration {
parse_stale_uploads_duration(ENV_STALE_UPLOADS_CLEANUP_INTERVAL, DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL)
}
fn encode_stale_upload_id(upload_uuid: &str) -> String {
base64_simd::URL_SAFE_NO_PAD
.encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_uuid).as_bytes())
}
fn initiated_from_upload_dir(upload_dir: &str, fallback: Option<OffsetDateTime>) -> OffsetDateTime {
upload_dir
.split_once('x')
.and_then(|(_, nanos)| nanos.parse::<i128>().ok())
.and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(nanos).ok())
.or(fallback)
.unwrap_or_else(OffsetDateTime::now_utc)
}
fn stale_upload_default_due(initiated: OffsetDateTime, default_expiry: StdDuration) -> OffsetDateTime {
initiated + time::Duration::seconds(default_expiry.as_secs() as i64)
}
async fn stale_upload_current_size(set: &Arc<SetDisks>, metadata: &HashMap<String, String>, upload_dir: &str) -> Option<usize> {
let bucket = metadata.get(RUSTFS_MULTIPART_BUCKET_KEY)?;
let object = metadata.get(RUSTFS_MULTIPART_OBJECT_KEY)?;
let upload_id = encode_stale_upload_id(upload_dir);
let parts = set
.list_object_parts(bucket, object, &upload_id, None, MAX_PARTS_COUNT, &ObjectOptions::default())
.await
.ok()?;
Some(
parts
.parts
.iter()
.map(|part| part.actual_size.max(part.size as i64).max(0) as usize)
.sum(),
)
}
async fn stale_upload_lifecycle_due(
set: &Arc<SetDisks>,
metadata: &HashMap<String, String>,
initiated: OffsetDateTime,
upload_dir: &str,
) -> Option<OffsetDateTime> {
let bucket = metadata.get(RUSTFS_MULTIPART_BUCKET_KEY)?;
let object = metadata.get(RUSTFS_MULTIPART_OBJECT_KEY)?;
let lifecycle = match metadata_sys::get_lifecycle_config(bucket).await {
Ok((lifecycle, _)) => lifecycle,
Err(_) => return None,
};
let object_opts = ObjectOpts {
name: object.clone(),
user_tags: metadata.get(AMZ_OBJECT_TAGGING).cloned().unwrap_or_default(),
mod_time: Some(initiated),
size: stale_upload_current_size(set, metadata, upload_dir).await.unwrap_or_default(),
is_latest: true,
delete_marker: false,
user_defined: metadata.clone(),
..Default::default()
};
abort_incomplete_multipart_upload_due(&lifecycle, &object_opts)
.await
.map(|(due, _)| due)
}
async fn read_stale_multipart_candidate(
disk: &Disk,
sha_dir: &str,
upload_dir: &str,
) -> Result<StaleMultipartUploadCandidate, DiskError> {
let metadata_path = format!("{sha_dir}/{upload_dir}/{STORAGE_FORMAT_FILE}");
let metadata_bytes = disk.read_metadata(RUSTFS_META_MULTIPART_BUCKET, &metadata_path).await?;
let (metadata, mod_time) = match get_file_info(
&metadata_bytes,
RUSTFS_META_MULTIPART_BUCKET,
&metadata_path,
"",
FileInfoOpts {
data: false,
include_free_versions: false,
},
) {
Ok(file_info) => (Some(file_info.metadata), file_info.mod_time),
Err(err) => {
warn!(path = %metadata_path, error = ?err, "failed to parse multipart metadata during stale cleanup");
(None, None)
}
};
let initiated = initiated_from_upload_dir(upload_dir, mod_time);
Ok(StaleMultipartUploadCandidate {
path: format!("{sha_dir}/{upload_dir}"),
initiated,
metadata,
})
}
fn merge_stale_multipart_candidate(
candidates: &mut HashMap<String, StaleMultipartUploadCandidate>,
candidate: StaleMultipartUploadCandidate,
) {
match candidates.get(&candidate.path) {
Some(existing) if existing.metadata.is_some() => {}
Some(existing) if existing.metadata.is_none() && candidate.metadata.is_none() => {}
_ => {
candidates.insert(candidate.path.clone(), candidate);
}
}
}
async fn cleanup_empty_multipart_sha_dirs_on_local_disks(set: &Arc<SetDisks>) {
for disk in set.get_local_disks().await.into_iter().flatten() {
if !disk.is_online().await {
continue;
}
let sha_dirs = match disk
.list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, "", -1)
.await
{
Ok(entries) => entries,
Err(err) => {
if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound {
warn!(error = ?err, "failed to list multipart root during empty sha cleanup");
}
continue;
}
};
for sha_dir in sha_dirs {
let sha_dir = sha_dir.trim_end_matches('/').to_string();
let upload_dirs = match disk
.list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, &sha_dir, -1)
.await
{
Ok(entries) => entries,
Err(err) => {
if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound {
warn!(sha_dir = %sha_dir, error = ?err, "failed to list multipart sha dir during empty sha cleanup");
}
continue;
}
};
if !upload_dirs.is_empty() {
continue;
}
if let Err(err) = disk
.delete(RUSTFS_META_MULTIPART_BUCKET, &sha_dir, DeleteOptions::default())
.await
&& err != DiskError::FileNotFound
&& err != DiskError::VolumeNotFound
{
warn!(sha_dir = %sha_dir, error = ?err, "failed to remove empty multipart sha dir");
}
}
}
}
async fn cleanup_stale_multipart_uploads_in_set(set: &Arc<SetDisks>, now: OffsetDateTime, default_expiry: StdDuration) -> usize {
let mut deleted = 0usize;
let mut candidates = HashMap::new();
for disk in set.get_local_disks().await.into_iter().flatten() {
if !disk.is_online().await {
continue;
}
let sha_dirs = match disk
.list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, "", -1)
.await
{
Ok(entries) => entries,
Err(err) => {
if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound {
warn!(error = ?err, "failed to list multipart root during stale cleanup");
}
continue;
}
};
for sha_dir in sha_dirs {
let sha_dir = sha_dir.trim_end_matches('/').to_string();
let upload_dirs = match disk
.list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, &sha_dir, -1)
.await
{
Ok(entries) => entries,
Err(err) => {
if err != DiskError::FileNotFound && err != DiskError::VolumeNotFound {
warn!(sha_dir = %sha_dir, error = ?err, "failed to list multipart sha dir during stale cleanup");
}
continue;
}
};
for upload_dir in upload_dirs {
let upload_dir = upload_dir.trim_end_matches('/').to_string();
let candidate_path = format!("{sha_dir}/{upload_dir}");
if candidates
.get(&candidate_path)
.is_some_and(|existing: &StaleMultipartUploadCandidate| existing.metadata.is_some())
{
continue;
}
let candidate = match read_stale_multipart_candidate(disk.as_ref(), &sha_dir, &upload_dir).await {
Ok(candidate) => candidate,
Err(err) => {
if err != DiskError::FileNotFound {
warn!(path = %candidate_path, error = ?err, "failed to read multipart metadata during stale cleanup");
}
let initiated = initiated_from_upload_dir(&upload_dir, None);
StaleMultipartUploadCandidate {
path: candidate_path,
initiated,
metadata: None,
}
}
};
merge_stale_multipart_candidate(&mut candidates, candidate);
}
}
}
for candidate in candidates.into_values() {
let upload_dir = candidate.path.rsplit('/').next().unwrap_or_default().to_string();
let mut due = stale_upload_default_due(candidate.initiated, default_expiry);
if let Some(metadata) = candidate.metadata.as_ref()
&& let Some(lifecycle_due) = stale_upload_lifecycle_due(set, metadata, candidate.initiated, &upload_dir).await
&& lifecycle_due < due
{
due = lifecycle_due;
}
if now < due {
continue;
}
match set.delete_all(RUSTFS_META_MULTIPART_BUCKET, &candidate.path).await {
Ok(()) => {
deleted += 1;
let upload_id = encode_stale_upload_id(&upload_dir);
if let Some(metadata) = candidate.metadata.as_ref() {
info!(
bucket = metadata.get(RUSTFS_MULTIPART_BUCKET_KEY).cloned().unwrap_or_default(),
object = metadata.get(RUSTFS_MULTIPART_OBJECT_KEY).cloned().unwrap_or_default(),
upload_id = %upload_id,
due = ?due,
"removed stale multipart upload"
);
} else {
info!(path = %candidate.path, upload_id = %upload_id, due = ?due, "removed stale multipart upload");
}
}
Err(err) => warn!(path = %candidate.path, error = ?err, "failed to remove stale multipart upload"),
}
}
cleanup_empty_multipart_sha_dirs_on_local_disks(set).await;
deleted
}
async fn cleanup_stale_multipart_uploads_once_at(api: Arc<ECStore>, now: OffsetDateTime, default_expiry: StdDuration) -> usize {
let mut deleted = 0usize;
for pool in &api.pools {
for set in &pool.disk_set {
deleted += cleanup_stale_multipart_uploads_in_set(set, now, default_expiry).await;
}
}
deleted
}
pub async fn run_stale_multipart_upload_cleanup_once(api: Arc<ECStore>) -> usize {
cleanup_stale_multipart_uploads_once_at(api, OffsetDateTime::now_utc(), stale_uploads_expiry()).await
}
pub fn init_background_stale_multipart_upload_cleanup(api: Arc<ECStore>) {
let cleanup_interval = stale_uploads_cleanup_interval();
let default_expiry = stale_uploads_expiry();
let api = Arc::downgrade(&api);
tokio::spawn(async move {
let mut interval = tokio::time::interval(cleanup_interval);
loop {
interval.tick().await;
let Some(api) = Weak::upgrade(&api) else {
return;
};
let deleted = cleanup_stale_multipart_uploads_once_at(api, OffsetDateTime::now_utc(), default_expiry).await;
if deleted > 0 {
info!(deleted, "completed stale multipart cleanup pass");
}
}
});
}
pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Result<(), std::io::Error> {
for rule in &lc.rules {
if let Some(transitions) = &rule.transitions {
@@ -1288,8 +1627,31 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc,
#[cfg(test)]
mod tests {
use super::mark_delete_opts_skip_decommissioned_on_remote_success;
use crate::store_api::ObjectOptions;
use super::{
StaleMultipartUploadCandidate, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at,
mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate,
};
use crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG;
use crate::bucket::metadata_sys;
use crate::disk::RUSTFS_META_MULTIPART_BUCKET;
use crate::disk::endpoint::Endpoint;
use crate::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
use crate::error::is_err_invalid_upload_id;
use crate::set_disk::{RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY};
use crate::store::ECStore;
use crate::store_api::{
BucketOperations, BucketOptions, MakeBucketOptions, MultipartOperations, ObjectOptions, PutObjReader,
};
use serial_test::serial;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock};
use std::time::Duration as StdDuration;
use time::OffsetDateTime;
use tokio::fs;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
#[test]
fn mark_delete_opts_skip_decommissioned_on_remote_success_sets_flag_on_success() {
@@ -1320,4 +1682,334 @@ mod tests {
assert!(opts.skip_decommissioned);
}
static STALE_MULTIPART_TEST_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>) {
if let Some((paths, ecstore)) = STALE_MULTIPART_TEST_ENV.get() {
return (paths.clone(), ecstore.clone());
}
let test_base_dir = format!("/tmp/rustfs_stale_multipart_test_{}", Uuid::new_v4());
let temp_dir = PathBuf::from(&test_base_dir);
if temp_dir.exists() {
fs::remove_dir_all(&temp_dir).await.ok();
}
fs::create_dir_all(&temp_dir).await.unwrap();
let disk_paths = vec![
temp_dir.join("disk1"),
temp_dir.join("disk2"),
temp_dir.join("disk3"),
temp_dir.join("disk4"),
];
for disk_path in &disk_paths {
fs::create_dir_all(disk_path).await.unwrap();
}
let mut endpoints = Vec::new();
for (i, disk_path) in disk_paths.iter().enumerate() {
let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(i);
endpoints.push(endpoint);
}
let endpoint_pools = EndpointServerPools(vec![PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: 4,
endpoints: Endpoints::from(endpoints),
cmd_line: "stale-multipart-test".to_string(),
platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
}]);
crate::store::init_local_disks(endpoint_pools.clone()).await.unwrap();
let ecstore = ECStore::new("127.0.0.1:0".parse().unwrap(), endpoint_pools, CancellationToken::new())
.await
.unwrap();
let buckets = ecstore
.list_bucket(&BucketOptions {
no_metadata: true,
..Default::default()
})
.await
.unwrap()
.into_iter()
.map(|bucket| bucket.name)
.collect();
metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await;
let _ = STALE_MULTIPART_TEST_ENV.set((disk_paths.clone(), ecstore.clone()));
(disk_paths, ecstore)
}
async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket: &str) {
ecstore
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("bucket should be created");
}
async fn set_abort_incomplete_lifecycle(bucket: &str, prefix: &str, days_after_initiation: i32) {
let lifecycle_xml = format!(
r#"<?xml version="1.0" encoding="UTF-8"?>
<LifecycleConfiguration>
<Rule>
<ID>abort-multipart</ID>
<Status>Enabled</Status>
<Filter>
<Prefix>{prefix}</Prefix>
</Filter>
<AbortIncompleteMultipartUpload>
<DaysAfterInitiation>{days_after_initiation}</DaysAfterInitiation>
</AbortIncompleteMultipartUpload>
</Rule>
</LifecycleConfiguration>"#
);
metadata_sys::update(bucket, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes())
.await
.expect("lifecycle metadata should be stored");
}
async fn set_abort_incomplete_lifecycle_with_size(
bucket: &str,
prefix: &str,
days_after_initiation: i32,
object_size_greater_than: usize,
) {
let lifecycle_xml = format!(
r#"<?xml version="1.0" encoding="UTF-8"?>
<LifecycleConfiguration>
<Rule>
<ID>abort-multipart-size</ID>
<Status>Enabled</Status>
<Filter>
<And>
<Prefix>{prefix}</Prefix>
<ObjectSizeGreaterThan>{object_size_greater_than}</ObjectSizeGreaterThan>
</And>
</Filter>
<AbortIncompleteMultipartUpload>
<DaysAfterInitiation>{days_after_initiation}</DaysAfterInitiation>
</AbortIncompleteMultipartUpload>
</Rule>
</LifecycleConfiguration>"#
);
metadata_sys::update(bucket, BUCKET_LIFECYCLE_CONFIG, lifecycle_xml.into_bytes())
.await
.expect("lifecycle metadata should be stored");
}
fn multipart_sha_dir(bucket: &str, object: &str) -> String {
hex_simd::encode_to_string(Sha256::digest(format!("{bucket}/{object}").as_bytes()), hex_simd::AsciiCase::Lower)
}
#[test]
fn merge_stale_multipart_candidate_prefers_metadata_over_fallback() {
let mut candidates = HashMap::new();
merge_stale_multipart_candidate(
&mut candidates,
StaleMultipartUploadCandidate {
path: "sha/upload".to_string(),
initiated: OffsetDateTime::UNIX_EPOCH,
metadata: None,
},
);
merge_stale_multipart_candidate(
&mut candidates,
StaleMultipartUploadCandidate {
path: "sha/upload".to_string(),
initiated: OffsetDateTime::UNIX_EPOCH,
metadata: Some(HashMap::from([("k".to_string(), "v".to_string())])),
},
);
assert_eq!(
candidates
.get("sha/upload")
.and_then(|candidate| candidate.metadata.as_ref())
.and_then(|metadata| metadata.get("k")),
Some(&"v".to_string())
);
}
#[tokio::test]
#[serial]
async fn stale_multipart_cleanup_uses_default_expiry_without_lifecycle() {
let (_paths, ecstore) = setup_test_env().await;
let bucket = format!("stale-default-{}", Uuid::new_v4().simple());
let object = "default-cleanup/object.txt";
create_test_bucket(&ecstore, &bucket).await;
let initiated = OffsetDateTime::now_utc() - time::Duration::hours(30);
let upload = ecstore
.new_multipart_upload(
&bucket,
object,
&ObjectOptions {
mod_time: Some(initiated),
..Default::default()
},
)
.await
.expect("multipart upload should be created");
let deleted = cleanup_stale_multipart_uploads_once_at(
ecstore.clone(),
OffsetDateTime::now_utc(),
StdDuration::from_secs(24 * 60 * 60),
)
.await;
assert!(deleted >= 1, "expected at least one stale multipart upload to be removed");
let err = ecstore
.get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
.await
.expect_err("stale multipart upload should be removed");
assert!(is_err_invalid_upload_id(&err));
}
#[tokio::test]
#[serial]
async fn stale_multipart_cleanup_applies_abort_incomplete_lifecycle_before_default_expiry() {
let (_paths, ecstore) = setup_test_env().await;
let bucket = format!("stale-lifecycle-{}", Uuid::new_v4().simple());
let object = "logs/prefix/object.txt";
create_test_bucket(&ecstore, &bucket).await;
set_abort_incomplete_lifecycle(&bucket, "logs/", 1).await;
let initiated = OffsetDateTime::now_utc() - time::Duration::hours(48);
let upload = ecstore
.new_multipart_upload(
&bucket,
object,
&ObjectOptions {
mod_time: Some(initiated),
..Default::default()
},
)
.await
.expect("multipart upload should be created");
let deleted = cleanup_stale_multipart_uploads_once_at(
ecstore.clone(),
OffsetDateTime::now_utc(),
StdDuration::from_secs(7 * 24 * 60 * 60),
)
.await;
assert!(deleted >= 1, "expected lifecycle-driven stale multipart cleanup to run");
let err = ecstore
.get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
.await
.expect_err("multipart upload should be removed by lifecycle abort rule");
assert!(is_err_invalid_upload_id(&err));
}
#[tokio::test]
#[serial]
async fn stale_multipart_cleanup_applies_abort_lifecycle_with_size_filter() {
let (_paths, ecstore) = setup_test_env().await;
let bucket = format!("stale-size-{}", Uuid::new_v4().simple());
let object = "logs/sized/object.txt";
create_test_bucket(&ecstore, &bucket).await;
set_abort_incomplete_lifecycle_with_size(&bucket, "logs/", 1, 5).await;
let initiated = OffsetDateTime::now_utc() - time::Duration::hours(48);
let upload = ecstore
.new_multipart_upload(
&bucket,
object,
&ObjectOptions {
mod_time: Some(initiated),
..Default::default()
},
)
.await
.expect("multipart upload should be created");
let mut data = PutObjReader::from_vec(vec![1, 2, 3, 4, 5, 6]);
ecstore
.put_object_part(&bucket, object, &upload.upload_id, 1, &mut data, &ObjectOptions::default())
.await
.expect("multipart part should be uploaded");
let deleted = cleanup_stale_multipart_uploads_once_at(
ecstore.clone(),
OffsetDateTime::now_utc(),
StdDuration::from_secs(7 * 24 * 60 * 60),
)
.await;
assert!(deleted >= 1, "expected lifecycle-driven stale multipart cleanup to run");
let err = ecstore
.get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
.await
.expect_err("multipart upload should be removed by size-qualified lifecycle abort rule");
assert!(is_err_invalid_upload_id(&err));
}
#[tokio::test]
#[serial]
async fn multipart_info_and_list_parts_do_not_expose_internal_metadata_keys() {
let (_paths, ecstore) = setup_test_env().await;
let bucket = format!("stale-sanitize-{}", Uuid::new_v4().simple());
let object = "sanitize/object.txt";
create_test_bucket(&ecstore, &bucket).await;
let upload = ecstore
.new_multipart_upload(&bucket, object, &ObjectOptions::default())
.await
.expect("multipart upload should be created");
let multipart_info = ecstore
.get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
.await
.expect("multipart info should be readable");
assert!(!multipart_info.user_defined.contains_key(RUSTFS_MULTIPART_BUCKET_KEY));
assert!(!multipart_info.user_defined.contains_key(RUSTFS_MULTIPART_OBJECT_KEY));
let parts = ecstore
.list_object_parts(&bucket, object, &upload.upload_id, None, 0, &ObjectOptions::default())
.await
.expect("multipart parts should be readable");
assert!(!parts.user_defined.contains_key(RUSTFS_MULTIPART_BUCKET_KEY));
assert!(!parts.user_defined.contains_key(RUSTFS_MULTIPART_OBJECT_KEY));
}
#[tokio::test]
#[serial]
async fn cleanup_removes_empty_multipart_sha_dirs() {
let (paths, ecstore) = setup_test_env().await;
let bucket = format!("stale-empty-sha-{}", Uuid::new_v4().simple());
let object = "empty-sha/object.txt";
let sha_dir = multipart_sha_dir(&bucket, object);
for path in &paths {
fs::create_dir_all(path.join(RUSTFS_META_MULTIPART_BUCKET).join(&sha_dir))
.await
.expect("empty multipart sha dir should be created for cleanup");
assert!(
path.join(RUSTFS_META_MULTIPART_BUCKET).join(&sha_dir).exists(),
"empty multipart sha dir should exist before cleanup"
);
}
cleanup_empty_multipart_sha_dirs_on_local_disks(&ecstore.pools[0].disk_set[0]).await;
for path in &paths {
assert!(
!path.join(RUSTFS_META_MULTIPART_BUCKET).join(&sha_dir).exists(),
"empty multipart sha dir should be removed"
);
}
}
}
crates/ecstore/src/bucket/lifecycle/lifecycle.rs
+20
View File
@@ -806,6 +806,26 @@ pub fn expected_expiry_time(mod_time: OffsetDateTime, days: i32) -> OffsetDateTi
t
}
pub async fn abort_incomplete_multipart_upload_due(
lc: &BucketLifecycleConfiguration,
obj: &ObjectOpts,
) -> Option<(OffsetDateTime, String)> {
let initiated = obj.mod_time?;
let rules = lc.filter_rules(obj).await?;
rules
.into_iter()
.filter_map(|rule| {
let days = rule
.abort_incomplete_multipart_upload
.as_ref()?
.days_after_initiation
.filter(|days| *days > 0)?;
Some((expected_expiry_time(initiated, days), rule.id.clone().unwrap_or_default()))
})
.min_by_key(|(due, _)| due.unix_timestamp_nanos())
}
#[derive(Default)]
pub struct ObjectOpts {
pub name: String,
crates/ecstore/src/compress.rs
+10
View File
@@ -82,16 +82,26 @@ pub const STANDARD_EXCLUDE_COMPRESS_CONTENT_TYPES: &[&str] = &[
"video/*",
"audio/*",
"image/*",
// Archive formats (compressed)
"application/zip",
"application/gzip",
"application/x-gzip",
"application/x-zip-compressed",
"application/x-compress",
"application/x-spoon",
"application/x-rar-compressed",
"application/x-7z-compressed",
"application/x-bzip",
"application/x-bzip2",
"application/x-xz",
"application/x-lzip",
"application/x-lzma",
"application/x-lzop",
"application/zstd",
"application/x-zstd",
// Archive formats (uncompressed containers that are typically not further compressible)
"application/x-tar",
"application/tar",
"application/pdf",
"application/wasm",
"font/*",
crates/ecstore/src/config/audit.rs
+32 -1
View File
@@ -15,7 +15,8 @@
use crate::config::{KV, KVS};
use rustfs_config::{
COMMENT_KEY, DEFAULT_LIMIT, ENABLE_KEY, EVENT_DEFAULT_DIR, EnableState, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD,
MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, WEBHOOK_AUTH_TOKEN,
MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY,
MQTT_TLS_POLICY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, WEBHOOK_AUTH_TOKEN,
WEBHOOK_BATCH_SIZE, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT,
WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL, WEBHOOK_SKIP_TLS_VERIFY,
};
@@ -154,6 +155,36 @@ pub static DEFAULT_AUDIT_MQTT_KVS: LazyLock<KVS> = LazyLock::new(|| {
value: DEFAULT_LIMIT.to_string(),
hidden_if_empty: false,
},
KV {
key: MQTT_TLS_POLICY.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_TLS_CA.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_TLS_CLIENT_CERT.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_TLS_CLIENT_KEY.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_TLS_TRUST_LEAF_AS_CA.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_WS_PATH_ALLOWLIST.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: COMMENT_KEY.to_owned(),
value: "".to_owned(),
crates/ecstore/src/config/notify.rs
+32 -1
View File
@@ -15,7 +15,8 @@
use crate::config::{KV, KVS};
use rustfs_config::{
COMMENT_KEY, DEFAULT_LIMIT, ENABLE_KEY, EVENT_DEFAULT_DIR, EnableState, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD,
MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, WEBHOOK_AUTH_TOKEN,
MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY,
MQTT_TLS_POLICY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, WEBHOOK_AUTH_TOKEN,
WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT,
WEBHOOK_SKIP_TLS_VERIFY,
};
@@ -133,6 +134,36 @@ pub static DEFAULT_NOTIFY_MQTT_KVS: LazyLock<KVS> = LazyLock::new(|| {
value: DEFAULT_LIMIT.to_string(),
hidden_if_empty: false,
},
KV {
key: MQTT_TLS_POLICY.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_TLS_CA.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_TLS_CLIENT_CERT.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_TLS_CLIENT_KEY.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_TLS_TRUST_LEAF_AS_CA.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: MQTT_WS_PATH_ALLOWLIST.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
},
KV {
key: COMMENT_KEY.to_owned(),
value: "".to_owned(),
crates/ecstore/src/set_disk.rs
+21 -3
View File
@@ -184,6 +184,13 @@ use uuid::Uuid;
pub const DEFAULT_READ_BUFFER_SIZE: usize = MI_B; // 1 MiB = 1024 * 1024;
pub const MAX_PARTS_COUNT: usize = 10000;
pub(crate) const RUSTFS_MULTIPART_BUCKET_KEY: &str = "x-rustfs-internal-multipart-bucket";
pub(crate) const RUSTFS_MULTIPART_OBJECT_KEY: &str = "x-rustfs-internal-multipart-object";
pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap<String, String>) {
metadata.remove(RUSTFS_MULTIPART_BUCKET_KEY);
metadata.remove(RUSTFS_MULTIPART_OBJECT_KEY);
}
/// Get the duplex buffer size from environment variable or use default.
///
@@ -2691,7 +2698,11 @@ impl MultipartOperations for SetDisks {
storage_class,
max_parts,
part_number_marker,
user_defined: fi.metadata.clone(),
user_defined: {
let mut metadata = fi.metadata.clone();
strip_internal_multipart_metadata(&mut metadata);
metadata
},
..Default::default()
};
@@ -3024,6 +3035,9 @@ impl MultipartOperations for SetDisks {
);
}
user_defined.insert(RUSTFS_MULTIPART_BUCKET_KEY.to_string(), bucket.to_string());
user_defined.insert(RUSTFS_MULTIPART_OBJECT_KEY.to_string(), object.to_string());
let (shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi);
let mod_time = opts.mod_time.unwrap_or(OffsetDateTime::now_utc());
@@ -3072,7 +3086,7 @@ impl MultipartOperations for SetDisks {
_opts: &ObjectOptions,
) -> Result<MultipartInfo> {
// TODO: nslock
let (fi, _) = self
let (mut fi, _) = self
.check_upload_id_exists(bucket, object, upload_id, false)
.await
.map_err(|e| to_object_err(e, vec![bucket, object, upload_id]))?;
@@ -3081,7 +3095,10 @@ impl MultipartOperations for SetDisks {
bucket: bucket.to_owned(),
object: object.to_owned(),
upload_id: upload_id.to_owned(),
user_defined: fi.metadata.clone(),
user_defined: {
strip_internal_multipart_metadata(&mut fi.metadata);
fi.metadata.clone()
},
..Default::default()
})
}
@@ -3372,6 +3389,7 @@ impl MultipartOperations for SetDisks {
fi.metadata.remove(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM);
fi.metadata.remove(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE);
strip_internal_multipart_metadata(&mut fi.metadata);
fi.size = object_size as i64;
fi.mod_time = opts.mod_time;
crates/ecstore/src/store/init.rs
+1
View File
@@ -344,6 +344,7 @@ impl ECStore {
init_global_bucket_monitor(num_nodes);
init_background_expiry(self.clone()).await;
crate::bucket::lifecycle::bucket_lifecycle_ops::init_background_stale_multipart_upload_cleanup(self.clone());
TransitionState::init(self.clone()).await;
crate::tier::tier::try_migrate_tiering_config(self.clone()).await;
crates/notify/src/factory.rs
+23 -8
View File
@@ -19,7 +19,8 @@ use rumqttc::QoS;
use rustfs_config::notify::{ENV_NOTIFY_MQTT_KEYS, ENV_NOTIFY_WEBHOOK_KEYS, NOTIFY_MQTT_KEYS, NOTIFY_WEBHOOK_KEYS};
use rustfs_config::{
DEFAULT_LIMIT, EVENT_DEFAULT_DIR, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR,
MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT,
MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_POLICY,
MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT,
WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR,
WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY,
};
@@ -27,7 +28,10 @@ use rustfs_ecstore::config::KVS;
use rustfs_targets::{
Target,
error::TargetError,
target::{mqtt::MQTTArgs, webhook::WebhookArgs},
target::{
mqtt::{MQTTArgs, MQTTTlsConfig, validate_mqtt_broker_url},
webhook::WebhookArgs,
},
};
use std::time::Duration;
use tracing::{debug, warn};
@@ -166,6 +170,14 @@ impl TargetFactory for MQTTTargetFactory {
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs)
.unwrap_or_else(|| Duration::from_secs(30)),
tls: MQTTTlsConfig::from_values(
config.lookup(MQTT_TLS_POLICY).as_deref(),
config.lookup(MQTT_TLS_CA).as_deref(),
config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(),
config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(),
config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(),
config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(),
)?,
queue_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or(EVENT_DEFAULT_DIR.to_string()),
queue_limit: config
.lookup(MQTT_QUEUE_LIMIT)
@@ -185,12 +197,15 @@ impl TargetFactory for MQTTTargetFactory {
let url = Url::parse(&broker)
.map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?;
match url.scheme() {
"tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" => {}
_ => {
return Err(TargetError::Configuration("Unsupported broker URL scheme".to_string()));
}
}
let tls = MQTTTlsConfig::from_values(
config.lookup(MQTT_TLS_POLICY).as_deref(),
config.lookup(MQTT_TLS_CA).as_deref(),
config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(),
config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(),
config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(),
config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(),
)?;
validate_mqtt_broker_url(&url, &tls)?;
if config.lookup(MQTT_TOPIC).is_none() {
return Err(TargetError::Configuration("Missing MQTT topic".to_string()));
crates/object-io/src/get.rs
+56
View File
@@ -170,6 +170,7 @@ pub trait CachedGetObjectSource {
fn content_type(&self) -> Option<&str>;
fn e_tag(&self) -> Option<&str>;
fn last_modified(&self) -> Option<&str>;
fn expires(&self) -> Option<&str>;
fn cache_control(&self) -> Option<&str>;
fn content_disposition(&self) -> Option<&str>;
fn content_encoding(&self) -> Option<&str>;
@@ -405,6 +406,10 @@ where
.last_modified()
.and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok())
.map(Timestamp::from);
let expires = cached
.expires()
.and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok())
.map(Timestamp::from);
let content_type = cached.content_type().and_then(|ct| ContentType::from_str(ct).ok());
@@ -416,6 +421,7 @@ where
accept_ranges: Some("bytes".to_string()),
e_tag: cached.e_tag().map(to_s3s_etag),
last_modified,
expires,
content_type,
cache_control: cached.cache_control().map(str::to_string),
content_disposition: cached.content_disposition().map(str::to_string),
@@ -890,8 +896,12 @@ pub fn build_get_object_output(
body,
content_length: Some(response_content_length),
last_modified,
expires: info.expires.map(Timestamp::from),
content_type,
cache_control: info.user_defined.get(CACHE_CONTROL.as_str()).cloned(),
content_disposition: info.user_defined.get(CONTENT_DISPOSITION.as_str()).cloned(),
content_encoding: info.content_encoding.clone(),
content_language: info.user_defined.get(CONTENT_LANGUAGE.as_str()).cloned(),
accept_ranges: Some("bytes".to_string()),
content_range,
e_tag: info.etag.as_ref().map(|etag| to_s3s_etag(etag)),
@@ -1052,6 +1062,7 @@ mod tests {
content_type: Option<String>,
e_tag: Option<String>,
last_modified: Option<String>,
expires: Option<String>,
cache_control: Option<String>,
content_disposition: Option<String>,
content_encoding: Option<String>,
@@ -1090,6 +1101,10 @@ mod tests {
self.last_modified.as_deref()
}
fn expires(&self) -> Option<&str> {
self.expires.as_deref()
}
fn cache_control(&self) -> Option<&str> {
self.cache_control.as_deref()
}
@@ -1624,6 +1639,7 @@ mod tests {
content_type: None,
e_tag: None,
last_modified: None,
expires: None,
cache_control: None,
content_disposition: None,
content_encoding: None,
@@ -1651,6 +1667,46 @@ mod tests {
assert_eq!(result.output.checksum_type, Some(ChecksumType::from_static(ChecksumType::FULL_OBJECT)));
}
#[test]
fn build_get_object_output_preserves_http_metadata_like_cached_path() {
let mut info = ObjectInfo {
content_type: Some("application/octet-stream".to_string()),
content_encoding: Some("zstd".to_string()),
etag: Some("etag".to_string()),
expires: Some(OffsetDateTime::UNIX_EPOCH),
..Default::default()
};
info.user_defined
.insert("cache-control".to_string(), "max-age=3600".to_string());
info.user_defined
.insert("content-disposition".to_string(), "attachment; filename=\"bundle.zip\"".to_string());
info.user_defined.insert("content-language".to_string(), "en-US".to_string());
let output = build_get_object_output(
None,
&info,
info.content_type
.as_ref()
.and_then(|content_type| ContentType::from_str(content_type).ok()),
info.mod_time.map(Timestamp::from),
8,
None,
None,
None,
None,
None,
&GetObjectChecksums::default(),
None,
None,
);
assert_eq!(output.cache_control.as_deref(), Some("max-age=3600"));
assert_eq!(output.content_disposition.as_deref(), Some("attachment; filename=\"bundle.zip\""));
assert_eq!(output.content_language.as_deref(), Some("en-US"));
assert_eq!(output.content_encoding.as_deref(), Some("zstd"));
assert_eq!(output.expires, Some(Timestamp::from(OffsetDateTime::UNIX_EPOCH)));
}
#[test]
fn build_cors_wrapped_get_object_flow_result_uses_wrapped_mode() {
let result = build_cors_wrapped_get_object_flow_result(
crates/scanner/src/scanner_folder.rs
+138 -12
View File
@@ -14,6 +14,7 @@
use std::collections::HashSet;
use std::fs::FileType;
use std::io::ErrorKind;
use std::sync::Arc;
use std::time::{Duration, SystemTime};
@@ -608,15 +609,29 @@ impl FolderScanner {
debug!("scan_folder: dir_path: {:?}", dir_path);
let mut dir_reader = tokio::fs::read_dir(&dir_path)
.await
.map_err(|e| ScannerError::Other(e.to_string()))?;
let mut dir_reader = match tokio::fs::read_dir(&dir_path).await {
Ok(dir_reader) => dir_reader,
Err(e) if e.kind() == ErrorKind::NotFound => {
warn!("scan_folder: directory disappeared before read {}: {}", dir_path, e);
return Ok(());
}
Err(e) => return Err(ScannerError::Io(e)),
};
while let Some(entry) = dir_reader
.next_entry()
.await
.map_err(|e| ScannerError::Other(e.to_string()))?
{
loop {
let entry = match dir_reader.next_entry().await {
Ok(Some(entry)) => entry,
Ok(None) => break,
Err(e) if e.kind() == ErrorKind::NotFound => {
warn!("scan_folder: directory disappeared during iteration {}: {}", dir_path, e);
break;
}
Err(e) if e.kind() == ErrorKind::NotADirectory => {
warn!("scan_folder: path became non-directory during iteration {}: {}", dir_path, e);
break;
}
Err(e) => return Err(ScannerError::Io(e)),
};
let file_name = entry.file_name().to_string_lossy().to_string();
if file_name.is_empty() || file_name == "." || file_name == ".." {
continue;
@@ -632,7 +647,42 @@ impl FolderScanner {
continue;
}
let entry_type = entry.file_type().await.map_err(|e| ScannerError::Other(e.to_string()))?;
// Ignore entries that disappeared during traversal or hit symlink
// loops, but propagate other walk errors.
let mut entry_type = match entry.file_type().await {
Ok(entry_type) => entry_type,
Err(e) if e.kind() == ErrorKind::NotFound => {
warn!("scan_folder: entry disappeared before type lookup {}: {}", entry_name, e);
continue;
}
Err(e) if e.kind() == ErrorKind::TooManyLinks => {
warn!("scan_folder: entry hit symlink loop before type lookup {}: {}", entry_name, e);
continue;
}
Err(e) => return Err(ScannerError::Io(e)),
};
if entry_type.is_symlink() {
let metadata = match tokio::fs::metadata(&file_path).await {
Ok(metadata) => metadata,
Err(e) if e.kind() == ErrorKind::NotFound => {
warn!("scan_folder: symlink target disappeared before metadata lookup {}: {}", file_path, e);
continue;
}
Err(e) if e.kind() == ErrorKind::TooManyLinks => {
warn!("scan_folder: symlink target hit loop before metadata lookup {}: {}", file_path, e);
continue;
}
Err(e) => return Err(ScannerError::Io(e)),
};
if metadata.is_dir() {
warn!("scan_folder: ignoring symlinked directory {}", file_path);
continue;
}
entry_type = metadata.file_type();
}
// ok
@@ -816,7 +866,10 @@ impl FolderScanner {
// Use Box::pin for recursive async call
let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst));
fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
if let Err(e) = fut.await {
warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e);
continue;
}
tokio::task::yield_now().await;
if !into.compacted {
@@ -865,7 +918,10 @@ impl FolderScanner {
// Use Box::pin for recursive async call
let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst));
fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
if let Err(e) = fut.await {
warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e);
continue;
}
tokio::task::yield_now().await;
if !into.compacted {
@@ -1076,7 +1132,10 @@ impl FolderScanner {
// Use Box::pin for recursive async call
let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst));
fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
if let Err(e) = fut.await {
warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e);
continue;
}
tokio::task::yield_now().await;
if !into.compacted {
@@ -1264,6 +1323,8 @@ mod tests {
use super::*;
use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk};
use serial_test::serial;
#[cfg(unix)]
use std::os::unix::fs::{PermissionsExt, symlink};
use std::sync::atomic::AtomicBool;
use uuid::Uuid;
@@ -1453,4 +1514,69 @@ mod tests {
assert!(scanner.new_cache.info.failed_objects.contains_key("fresh2"));
assert!(!scanner.new_cache.info.failed_objects.contains_key("expired"));
}
#[tokio::test]
#[serial]
#[cfg(unix)]
async fn test_scan_folder_skips_unreadable_child_directory() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir.clone());
let bucket_dir = temp_dir.join("bucket");
let good_dir = bucket_dir.join("good");
let bad_dir = bucket_dir.join("bad");
std::fs::create_dir_all(&good_dir).expect("failed to create good dir");
std::fs::create_dir_all(&bad_dir).expect("failed to create bad dir");
std::fs::set_permissions(&bad_dir, std::fs::Permissions::from_mode(0o000)).expect("failed to remove bad dir permissions");
scanner.old_cache.info.name = "bucket".to_string();
scanner.new_cache.info.name = "bucket".to_string();
scanner.update_cache.info.name = "bucket".to_string();
let folder = CachedFolder {
name: "bucket".to_string(),
parent: None,
object_heal_prob_div: 1,
};
let mut into = DataUsageEntry::default();
let result = scanner.scan_folder(CancellationToken::new(), folder, &mut into).await;
std::fs::set_permissions(&bad_dir, std::fs::Permissions::from_mode(0o755))
.expect("failed to restore bad dir permissions");
assert!(result.is_ok(), "expected unreadable child directory to be skipped");
}
#[tokio::test]
#[serial]
#[cfg(unix)]
async fn test_scan_folder_ignores_symlinked_child_directory() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir.clone());
let bucket_dir = temp_dir.join("bucket");
let target_dir = bucket_dir.join("target");
let link_dir = bucket_dir.join("link");
std::fs::create_dir_all(&target_dir).expect("failed to create target dir");
symlink(&target_dir, &link_dir).expect("failed to create symlinked dir");
scanner.old_cache.info.name = "bucket".to_string();
scanner.new_cache.info.name = "bucket".to_string();
scanner.update_cache.info.name = "bucket".to_string();
let folder = CachedFolder {
name: "bucket".to_string(),
parent: None,
object_heal_prob_div: 1,
};
let mut into = DataUsageEntry::default();
let result = scanner.scan_folder(CancellationToken::new(), folder, &mut into).await;
assert!(result.is_ok(), "expected symlinked child directory to be ignored");
assert_eq!(into.failed_objects, 0, "expected ignored symlink not to count as a failed object");
}
}
crates/targets/Cargo.toml
+3 -1
View File
@@ -13,11 +13,13 @@ documentation = "https://docs.rs/rustfs-target/latest/rustfs_target/"
[dependencies]
rustfs-config = { workspace = true, features = ["notify", "constants", "audit"] }
rustfs-utils = { workspace = true, features = ["sys", "notify"] }
rustfs-utils = { workspace = true, features = ["sys", "notify", "tls"] }
rustfs-s3-common = { workspace = true }
async-trait = { workspace = true }
hyper-rustls = { workspace = true }
reqwest = { workspace = true }
rumqttc = { workspace = true }
rustls = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
snap = { workspace = true }
crates/targets/src/check.rs
+34 -33
View File
@@ -21,7 +21,9 @@
/// * `password` - Optional password for authentication
/// # Returns
/// * `Ok(())` - If the connection is successful
/// * `Err(String)` - If the connection fails, contains an error message
/// * `Err(TargetError)` - If the check fails.
/// `TargetError::Configuration` indicates a bad configuration (invalid URL, TLS settings, etc.).
/// Other variants indicate a connectivity or runtime failure.
///
/// # Example
/// ```rust,no_run
@@ -40,54 +42,53 @@
/// }
/// }
/// ```
/// # Note
/// Need to add `rumqttc` and `url` dependencies in `Cargo.toml`
/// ```toml
/// [dependencies]
/// rumqttc = "0.25.0"
/// url = "2.5.7"
/// tokio = { version = "1", features = ["full"] }
/// ```
///
pub async fn check_mqtt_broker_available(
broker_url: &str,
topic: &str,
username: Option<&str>,
password: Option<&str>,
) -> Result<(), String> {
use rumqttc::{AsyncClient, MqttOptions, QoS};
let url = rustfs_utils::parse_url(broker_url).map_err(|e| format!("Broker URL parsing failed:{e}"))?;
) -> Result<(), crate::TargetError> {
use crate::target::mqtt::MQTTTlsConfig;
check_mqtt_broker_available_with_tls(broker_url, topic, username, password, &MQTTTlsConfig::default()).await
}
pub async fn check_mqtt_broker_available_with_tls(
broker_url: &str,
topic: &str,
username: Option<&str>,
password: Option<&str>,
tls: &crate::target::mqtt::MQTTTlsConfig,
) -> Result<(), crate::TargetError> {
use crate::target::mqtt::build_mqtt_options;
use rumqttc::{AsyncClient, QoS};
let url = rustfs_utils::parse_url(broker_url)
.map_err(|e| crate::TargetError::Configuration(format!("Broker URL parsing failed: {e}")))?;
let url = url.url();
match url.scheme() {
"tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" | "tls" | "tcps" => {}
_ => return Err("unsupported broker url scheme".to_string()),
}
let host = url.host_str().ok_or("Broker is missing host")?;
let port = url.port().unwrap_or(1883);
let mut mqtt_options = MqttOptions::new("rustfs_check", host, port);
// Set credentials if provided
if let Some(user) = username
&& !user.is_empty()
{
let pass = password.unwrap_or("");
mqtt_options.set_credentials(user, pass);
}
mqtt_options.set_keep_alive(std::time::Duration::from_secs(5));
// build_mqtt_options returns TargetError directly; Configuration variants propagate as-is.
let mqtt_options = build_mqtt_options(
"rustfs_check".to_string(),
url,
username,
password,
tls,
std::time::Duration::from_secs(5),
None,
)?;
let (client, mut eventloop) = AsyncClient::new(mqtt_options, 1);
// Try to connect and subscribe
client
.subscribe(topic, QoS::AtLeastOnce)
.await
.map_err(|e| format!("MQTT subscription failed:{e}"))?;
.map_err(|e| crate::TargetError::Network(format!("MQTT subscription failed: {e}")))?;
// Wait for eventloop to receive at least one event
match tokio::time::timeout(std::time::Duration::from_secs(3), eventloop.poll()).await {
Ok(Ok(_)) => Ok(()),
Ok(Err(e)) => Err(format!("MQTT connection failed:{e}")),
Err(_) => Err("MQTT connection timeout".to_string()),
Ok(Err(e)) => Err(crate::TargetError::Network(format!("MQTT connection failed: {e}"))),
Err(_) => Err(crate::TargetError::Timeout("MQTT connection timed out".to_string())),
}
}
crates/targets/src/lib.rs
+1 -1
View File
@@ -18,7 +18,7 @@ pub mod error;
pub mod store;
pub mod target;
pub use check::check_mqtt_broker_available;
pub use check::{check_mqtt_broker_available, check_mqtt_broker_available_with_tls};
pub use error::{StoreError, TargetError};
pub use rustfs_s3_common::EventName;
use serde::{Deserialize, Serialize};
crates/targets/src/target/mqtt.rs
+482 -39
View File
@@ -20,13 +20,21 @@ use crate::{
target::{ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetType},
};
use async_trait::async_trait;
use rumqttc::{AsyncClient, ConnectionError, EventLoop, MqttOptions, Outgoing, Packet, QoS, mqttbytes::Error as MqttBytesError};
use hyper_rustls::ConfigBuilderExt;
use rumqttc::{
AsyncClient, Broker, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, QoS, Transport,
mqttbytes::Error as MqttBytesError,
};
use rustfs_config::{
EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST,
};
use rustls::ClientConfig;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::sync::Arc;
use std::{
marker::PhantomData,
path::PathBuf,
path::{Path, PathBuf},
sync::atomic::{AtomicBool, Ordering},
time::Duration,
};
@@ -36,6 +44,381 @@ use url::Url;
const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15);
const EVENT_LOOP_POLL_TIMEOUT: Duration = Duration::from_secs(10); // For initial connection check in task
const DEFAULT_MQTT_TCP_PORT: u16 = 1883;
const DEFAULT_MQTT_TLS_PORT: u16 = 8883;
const DEFAULT_MQTT_WSS_PORT: u16 = 443;
const MAX_MQTT_PACKET_SIZE_BYTES: u32 = 100 * 1024 * 1024;
const DEFAULT_MQTT_WS_PATH_ALLOWLIST: &[&str] = &["/", "/mqtt"];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MQTTTlsPolicy {
SystemCa,
CustomCa,
}
impl MQTTTlsPolicy {
fn parse(value: &str) -> Result<Self, TargetError> {
match value.trim() {
value if value.eq_ignore_ascii_case("system_ca") => Ok(Self::SystemCa),
value if value.eq_ignore_ascii_case("custom_ca") => Ok(Self::CustomCa),
_ => Err(TargetError::Configuration(
"MQTT tls_policy must be one of: system_ca, custom_ca".to_string(),
)),
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MQTTTlsConfig {
pub policy: Option<MQTTTlsPolicy>,
pub ca_path: String,
pub client_cert_path: String,
pub client_key_path: String,
pub trust_leaf_as_ca: bool,
pub ws_path_allowlist: Vec<String>,
}
impl MQTTTlsConfig {
pub fn from_values(
policy: Option<&str>,
ca_path: Option<&str>,
client_cert_path: Option<&str>,
client_key_path: Option<&str>,
trust_leaf_as_ca: Option<&str>,
ws_path_allowlist: Option<&str>,
) -> Result<Self, TargetError> {
let policy = match policy.map(str::trim).filter(|value| !value.is_empty()) {
Some(value) => Some(MQTTTlsPolicy::parse(value)?),
None => None,
};
let trust_leaf_as_ca = match trust_leaf_as_ca.map(str::trim).filter(|value| !value.is_empty()) {
Some(value) => value
.parse::<EnableState>()
.map(EnableState::is_enabled)
.map_err(|_| TargetError::Configuration(format!("Invalid value for {MQTT_TLS_TRUST_LEAF_AS_CA}")))?,
None => false,
};
let ws_path_allowlist = match ws_path_allowlist.map(str::trim).filter(|value| !value.is_empty()) {
Some(value) => parse_ws_path_allowlist(value)?,
None => Vec::new(),
};
Ok(Self {
policy,
ca_path: ca_path.unwrap_or_default().trim().to_string(),
client_cert_path: client_cert_path.unwrap_or_default().trim().to_string(),
client_key_path: client_key_path.unwrap_or_default().trim().to_string(),
trust_leaf_as_ca,
ws_path_allowlist,
})
}
fn effective_ws_path_allowlist(&self) -> Vec<&str> {
if self.ws_path_allowlist.is_empty() {
DEFAULT_MQTT_WS_PATH_ALLOWLIST.to_vec()
} else {
self.ws_path_allowlist.iter().map(String::as_str).collect()
}
}
}
fn parse_ws_path_allowlist(value: &str) -> Result<Vec<String>, TargetError> {
let mut allowlist = Vec::new();
for raw in value.split(',') {
let path = raw.trim();
if path.is_empty() {
continue;
}
if !path.starts_with('/') || path.contains('?') || path.contains('#') {
return Err(TargetError::Configuration(format!(
"{MQTT_WS_PATH_ALLOWLIST} entries must be absolute paths without query or fragment"
)));
}
allowlist.push(path.to_string());
}
if allowlist.is_empty() {
return Err(TargetError::Configuration(format!(
"{MQTT_WS_PATH_ALLOWLIST} must contain at least one websocket path"
)));
}
Ok(allowlist)
}
fn keep_alive_seconds(duration: Duration) -> u16 {
duration.as_secs().min(u64::from(u16::MAX)) as u16
}
fn default_broker_port(scheme: &str) -> u16 {
match scheme {
"ssl" | "tls" | "tcps" | "mqtts" => DEFAULT_MQTT_TLS_PORT,
"wss" => DEFAULT_MQTT_WSS_PORT,
_ => DEFAULT_MQTT_TCP_PORT,
}
}
fn websocket_broker_url(broker: &Url, secure: bool) -> Result<String, TargetError> {
let mut url = broker.clone();
url.set_scheme("ws")
.map_err(|_| TargetError::Configuration("Failed to normalize websocket broker URL scheme".to_string()))?;
if secure && url.port().is_none() {
url.set_port(Some(DEFAULT_MQTT_WSS_PORT))
.map_err(|_| TargetError::Configuration("Failed to set default secure websocket broker port".to_string()))?;
}
Ok(url.to_string())
}
fn ensure_rustls_provider_installed() {
if rustls::crypto::CryptoProvider::get_default().is_none()
&& rustls::crypto::aws_lc_rs::default_provider().install_default().is_err()
{
debug!("rustls crypto provider was installed concurrently, skipping aws-lc-rs install");
}
}
fn validate_path_is_absolute(path: &str, field: &str) -> Result<(), TargetError> {
if !Path::new(path).is_absolute() {
return Err(TargetError::Configuration(format!("{field} must be an absolute path")));
}
Ok(())
}
fn build_root_store(ca_path: &str, trust_leaf_as_ca: bool) -> Result<rustls::RootCertStore, TargetError> {
let certs =
rustfs_utils::load_certs(ca_path).map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_ca: {e}")))?;
let mut store = rustls::RootCertStore::empty();
if trust_leaf_as_ca {
let (valid, invalid) = store.add_parsable_certificates(certs);
if valid == 0 {
return Err(TargetError::Configuration(format!(
"MQTT tls_ca did not contain any parsable trust anchors (ignored {invalid} entries)"
)));
}
} else {
for cert in certs {
store
.add(cert)
.map_err(|e| TargetError::Configuration(format!("Failed to add MQTT tls_ca to root store: {e}")))?;
}
}
Ok(store)
}
fn build_mqtt_tls_transport(broker: &Url, tls: &MQTTTlsConfig) -> Result<Transport, TargetError> {
ensure_rustls_provider_installed();
let client_config = match tls
.policy
.ok_or_else(|| TargetError::Configuration("Secure MQTT schemes require an explicit tls_policy".to_string()))?
{
MQTTTlsPolicy::SystemCa => {
let builder = ClientConfig::builder()
.with_native_roots()
.map_err(|e| TargetError::Configuration(format!("Failed to load native root certificates: {e}")))?;
if tls.client_cert_path.is_empty() {
builder.with_no_client_auth()
} else {
let certs = rustfs_utils::load_certs(&tls.client_cert_path)
.map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_cert: {e}")))?;
let key = rustfs_utils::load_private_key(&tls.client_key_path)
.map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_key: {e}")))?;
builder
.with_client_auth_cert(certs, key)
.map_err(|e| TargetError::Configuration(format!("Failed to build MQTT client mTLS identity: {e}")))?
}
}
MQTTTlsPolicy::CustomCa => {
let builder = ClientConfig::builder().with_root_certificates(build_root_store(&tls.ca_path, tls.trust_leaf_as_ca)?);
if tls.client_cert_path.is_empty() {
builder.with_no_client_auth()
} else {
let certs = rustfs_utils::load_certs(&tls.client_cert_path)
.map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_cert: {e}")))?;
let key = rustfs_utils::load_private_key(&tls.client_key_path)
.map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_key: {e}")))?;
builder
.with_client_auth_cert(certs, key)
.map_err(|e| TargetError::Configuration(format!("Failed to build MQTT client mTLS identity: {e}")))?
}
}
};
if matches!(broker.scheme(), "wss") {
Ok(Transport::wss_with_config(client_config.into()))
} else {
Ok(Transport::tls_with_config(client_config.into()))
}
}
pub fn validate_mqtt_broker_url(broker: &Url, tls: &MQTTTlsConfig) -> Result<(), TargetError> {
match broker.scheme() {
"ws" | "wss" | "tcp" | "ssl" | "tls" | "tcps" | "mqtt" | "mqtts" => {}
_ => {
return Err(TargetError::Configuration("unknown protocol in broker address".to_string()));
}
}
if !broker.username().is_empty() || broker.password().is_some() {
return Err(TargetError::Configuration("Broker URL must not embed username or password".to_string()));
}
broker
.host_str()
.ok_or_else(|| TargetError::Configuration("Broker is missing host".to_string()))?;
let secure_scheme = matches!(broker.scheme(), "wss" | "ssl" | "tls" | "tcps" | "mqtts");
let websocket_scheme = matches!(broker.scheme(), "ws" | "wss");
if !websocket_scheme {
if !matches!(broker.path(), "" | "/") {
return Err(TargetError::Configuration(
"Broker URL path is only supported for ws/wss schemes".to_string(),
));
}
if broker.query().is_some() {
return Err(TargetError::Configuration(
"Broker URL query is only supported for ws/wss schemes".to_string(),
));
}
if broker.fragment().is_some() {
return Err(TargetError::Configuration(
"Broker URL fragment is only supported for ws/wss schemes".to_string(),
));
}
if !tls.ws_path_allowlist.is_empty() {
return Err(TargetError::Configuration(format!(
"{MQTT_WS_PATH_ALLOWLIST} is only supported for ws/wss schemes"
)));
}
} else if !tls
.effective_ws_path_allowlist()
.iter()
.any(|allowed_path| *allowed_path == broker.path())
{
return Err(TargetError::Configuration(format!(
"Websocket broker path '{}' is not in the {MQTT_WS_PATH_ALLOWLIST} allowlist",
broker.path()
)));
}
if secure_scheme {
let policy = tls
.policy
.ok_or_else(|| TargetError::Configuration("Secure MQTT schemes require an explicit tls_policy".to_string()))?;
if !tls.client_cert_path.is_empty() {
validate_path_is_absolute(&tls.client_cert_path, MQTT_TLS_CLIENT_CERT)?;
}
if !tls.client_key_path.is_empty() {
validate_path_is_absolute(&tls.client_key_path, MQTT_TLS_CLIENT_KEY)?;
}
if tls.client_cert_path.is_empty() != tls.client_key_path.is_empty() {
return Err(TargetError::Configuration(
"MQTT tls_client_cert and tls_client_key must be specified together".to_string(),
));
}
match policy {
MQTTTlsPolicy::SystemCa => {
if !tls.ca_path.is_empty() {
return Err(TargetError::Configuration(format!(
"{MQTT_TLS_CA} is not allowed when tls_policy=system_ca"
)));
}
if tls.trust_leaf_as_ca {
return Err(TargetError::Configuration(format!(
"{MQTT_TLS_TRUST_LEAF_AS_CA} requires tls_policy=custom_ca"
)));
}
}
MQTTTlsPolicy::CustomCa => {
if tls.ca_path.is_empty() {
return Err(TargetError::Configuration(format!("{MQTT_TLS_CA} is required when tls_policy=custom_ca")));
}
validate_path_is_absolute(&tls.ca_path, MQTT_TLS_CA)?;
}
}
} else if tls.policy.is_some()
|| !tls.ca_path.is_empty()
|| !tls.client_cert_path.is_empty()
|| !tls.client_key_path.is_empty()
|| tls.trust_leaf_as_ca
{
return Err(TargetError::Configuration(
"TLS settings are only allowed for mqtts/ssl/tls/tcps/wss schemes".to_string(),
));
}
Ok(())
}
pub(crate) fn build_mqtt_options(
client_id: String,
broker: &Url,
username: Option<&str>,
password: Option<&str>,
tls: &MQTTTlsConfig,
keep_alive: Duration,
max_packet_size: Option<u32>,
) -> Result<MqttOptions, TargetError> {
validate_mqtt_broker_url(broker, tls)?;
let host = broker
.host_str()
.ok_or_else(|| TargetError::Configuration("Broker is missing host".to_string()))?;
let port = broker.port().unwrap_or_else(|| default_broker_port(broker.scheme()));
let mut mqtt_options = match broker.scheme() {
"tcp" | "mqtt" => MqttOptions::new(client_id, (host, port)),
"ssl" | "tls" | "tcps" | "mqtts" => {
let mut options = MqttOptions::new(client_id, (host, port));
options.set_transport(build_mqtt_tls_transport(broker, tls)?);
options
}
"ws" => {
let websocket_broker = Broker::websocket(broker.as_str().to_string())
.map_err(|e| TargetError::Configuration(format!("Invalid websocket broker URL: {e}")))?;
MqttOptions::new(client_id, websocket_broker)
}
"wss" => {
let websocket_broker = Broker::websocket(websocket_broker_url(broker, true)?)
.map_err(|e| TargetError::Configuration(format!("Invalid secure websocket broker URL: {e}")))?;
let mut options = MqttOptions::new(client_id, websocket_broker);
options.set_transport(build_mqtt_tls_transport(broker, tls)?);
options
}
_ => {
return Err(TargetError::Configuration("unknown protocol in broker address".to_string()));
}
};
mqtt_options.set_keep_alive(keep_alive_seconds(keep_alive));
if let Some(max_packet_size) = max_packet_size {
mqtt_options.set_max_packet_size(Some(max_packet_size));
}
if let Some(user) = username
&& !user.is_empty()
{
mqtt_options.set_credentials(user.to_string(), password.unwrap_or("").to_string());
}
Ok(mqtt_options)
}
/// Arguments for configuring an MQTT target
#[derive(Debug, Clone)]
@@ -52,6 +435,8 @@ pub struct MQTTArgs {
pub username: String,
/// The password for the broker
pub password: String,
/// Explicit TLS configuration for secure MQTT transports
pub tls: MQTTTlsConfig,
/// The maximum interval for reconnection attempts (Note: rumqttc has internal strategy)
pub max_reconnect_interval: Duration,
/// The keep alive interval
@@ -70,12 +455,7 @@ impl MQTTArgs {
return Ok(());
}
match self.broker.scheme() {
"ws" | "wss" | "tcp" | "ssl" | "tls" | "tcps" | "mqtt" | "mqtts" => {}
_ => {
return Err(TargetError::Configuration("unknown protocol in broker address".to_string()));
}
}
validate_mqtt_broker_url(&self.broker, &self.tls)?;
if self.topic.is_empty() {
return Err(TargetError::Configuration("MQTT topic cannot be empty".to_string()));
@@ -187,16 +567,15 @@ where
.init_cell
.get_or_try_init(|| async {
debug!(target_id = %target_id_clone, "Initializing MQTT background task.");
let host = args_clone.broker.host_str().unwrap_or("localhost");
let port = args_clone.broker.port().unwrap_or(1883);
let mut mqtt_options = MqttOptions::new(format!("rustfs_notify_{}", uuid::Uuid::new_v4()), host, port);
mqtt_options
.set_keep_alive(args_clone.keep_alive)
.set_max_packet_size(100 * 1024 * 1024, 100 * 1024 * 1024); // 100MB
if !args_clone.username.is_empty() {
mqtt_options.set_credentials(args_clone.username.clone(), args_clone.password.clone());
}
let mqtt_options = build_mqtt_options(
format!("rustfs_notify_{}", uuid::Uuid::new_v4()),
&args_clone.broker,
Some(args_clone.username.as_str()),
Some(args_clone.password.as_str()),
&args_clone.tls,
args_clone.keep_alive,
Some(MAX_MQTT_PACKET_SIZE_BYTES),
)?;
let (new_client, eventloop) = AsyncClient::new(mqtt_options, 10);
@@ -341,40 +720,40 @@ async fn run_mqtt_event_loop(
polled_event_result = async {
if !initial_connection_established || !connected_status.load(Ordering::SeqCst) {
match tokio::time::timeout(EVENT_LOOP_POLL_TIMEOUT, eventloop.poll()).await {
Ok(Ok(event)) => Ok(event),
Ok(Err(e)) => Err(e),
Ok(result) => Some(result),
Err(_) => {
debug!(target_id = %target_id, "MQTT poll timed out (EVENT_LOOP_POLL_TIMEOUT) while not connected or status pending.");
Err(ConnectionError::NetworkTimeout)
connected_status.store(false, Ordering::SeqCst);
None
}
}
} else {
eventloop.poll().await
Some(eventloop.poll().await)
}
} => {
match polled_event_result {
Ok(notification) => {
Some(Ok(notification)) => {
trace!(target_id = %target_id, event = ?notification, "Received MQTT event");
match notification {
rumqttc::Event::Incoming(Packet::ConnAck(_conn_ack)) => {
rumqttc::Event::Incoming(Incoming::ConnAck(_conn_ack)) => {
info!(target_id = %target_id, "MQTT connected (ConnAck).");
connected_status.store(true, Ordering::SeqCst);
initial_connection_established = true;
}
rumqttc::Event::Incoming(Packet::Publish(publish)) => {
debug!(target_id = %target_id, topic = %publish.topic, payload_len = publish.payload.len(), "Received message on subscribed topic.");
rumqttc::Event::Incoming(Incoming::Publish(publish)) => {
debug!(target_id = %target_id, topic = ?publish.topic, payload_len = publish.payload.len(), "Received message on subscribed topic.");
}
rumqttc::Event::Incoming(Packet::Disconnect) => {
rumqttc::Event::Incoming(Incoming::Disconnect(_)) => {
info!(target_id = %target_id, "Received Disconnect packet from broker. MQTT connection lost.");
connected_status.store(false, Ordering::SeqCst);
}
rumqttc::Event::Incoming(Packet::PingResp) => {
rumqttc::Event::Incoming(Incoming::PingResp(_)) => {
trace!(target_id = %target_id, "Received PingResp from broker. Connection is alive.");
}
rumqttc::Event::Incoming(Packet::SubAck(suback)) => {
rumqttc::Event::Incoming(Incoming::SubAck(suback)) => {
trace!(target_id = %target_id, "Received SubAck for pkid: {}", suback.pkid);
}
rumqttc::Event::Incoming(Packet::PubAck(puback)) => {
rumqttc::Event::Incoming(Incoming::PubAck(puback)) => {
trace!(target_id = %target_id, "Received PubAck for pkid: {}", puback.pkid);
}
// Process other incoming packet types as needed (PubRec, PubRel, PubComp, UnsubAck)
@@ -393,18 +772,13 @@ async fn run_mqtt_event_loop(
}
}
}
Err(e) => {
Some(Err(e)) => {
connected_status.store(false, Ordering::SeqCst);
error!(target_id = %target_id, error = %e, "Error from MQTT event loop poll");
if matches!(e, ConnectionError::NetworkTimeout) && (!initial_connection_established || !connected_status.load(Ordering::SeqCst)) {
warn!(target_id = %target_id, "Timeout during initial poll or pending state, will retry.");
continue;
}
if matches!(e,
ConnectionError::Io(_) |
ConnectionError::NetworkTimeout |
ConnectionError::Timeout(_) |
ConnectionError::ConnectionRefused(_) |
ConnectionError::Tls(_)
) {
@@ -422,6 +796,10 @@ async fn run_mqtt_event_loop(
// Sleep briefly to avoid busy cycles in case of rapid failure.
tokio::time::sleep(Duration::from_secs(1)).await;
}
None => {
warn!(target_id = %target_id, "Timeout during initial poll or pending state, will retry.");
continue;
}
}
}
}
@@ -443,7 +821,7 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool {
match state_err {
// If StateError is caused by deserialization issues, check the underlying MqttBytesError
rumqttc::StateError::Deserialization(mqtt_bytes_err) => { // The type of mqtt_bytes_err is &rumqttc::mqttbytes::Error
matches!(
matches!(
mqtt_bytes_err,
MqttBytesError::InvalidProtocol // Invalid agreement
| MqttBytesError::InvalidProtocolLevel(_) // Invalid protocol level
@@ -451,7 +829,7 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool {
| MqttBytesError::InvalidPacketType(_) // Invalid package type
| MqttBytesError::MalformedPacket // Package format error
| MqttBytesError::PayloadTooLong // Too long load
| MqttBytesError::PayloadSizeLimitExceeded(_) // Load size limit exceeded
| MqttBytesError::PayloadSizeLimitExceeded { .. } // Load size limit exceeded
| MqttBytesError::TopicNotUtf8 // Topic Non-UTF-8 (Serious Agreement Violation)
)
}
@@ -640,3 +1018,68 @@ where
self.args.enable
}
}
#[cfg(test)]
mod tests {
use super::{MQTTTlsConfig, validate_mqtt_broker_url};
use url::Url;
#[test]
fn validate_mqtt_broker_url_rejects_non_websocket_path() {
let url = Url::parse("mqtt://broker.example.com:1883/custom").expect("valid url");
let err = validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect_err("non-websocket path should be rejected");
assert!(err.to_string().contains("path is only supported"));
}
#[test]
fn validate_mqtt_broker_url_rejects_non_websocket_query() {
let url = Url::parse("mqtt://broker.example.com:1883?client_id=test").expect("valid url");
let err = validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect_err("non-websocket query should be rejected");
assert!(err.to_string().contains("query is only supported"));
}
#[test]
fn validate_mqtt_broker_url_rejects_non_websocket_fragment() {
let url = Url::parse("mqtt://broker.example.com:1883/#section").expect("valid url");
let err =
validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect_err("non-websocket fragment should be rejected");
assert!(err.to_string().contains("fragment is only supported"));
}
#[test]
fn validate_mqtt_broker_url_allows_websocket_path_and_query() {
let url = Url::parse("ws://broker.example.com:8080/mqtt?client_id=test").expect("valid url");
validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect("websocket path and query should be allowed");
}
#[test]
fn validate_mqtt_broker_url_rejects_url_embedded_credentials() {
let url = Url::parse("mqtt://user:pass@broker.example.com:1883").expect("valid url");
let err = validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect_err("url credentials should be rejected");
assert!(err.to_string().contains("must not embed username or password"));
}
#[test]
fn validate_mqtt_broker_url_requires_explicit_tls_policy_for_secure_scheme() {
let url = Url::parse("mqtts://broker.example.com:8883").expect("valid url");
let err = validate_mqtt_broker_url(&url, &MQTTTlsConfig::default())
.expect_err("secure scheme should require explicit tls policy");
assert!(err.to_string().contains("explicit tls_policy"));
}
#[test]
fn validate_mqtt_broker_url_rejects_disallowed_websocket_path() {
let url = Url::parse("wss://broker.example.com/private").expect("valid url");
let tls = MQTTTlsConfig::from_values(Some("system_ca"), None, None, None, None, Some("/mqtt")).expect("valid tls config");
let err = validate_mqtt_broker_url(&url, &tls).expect_err("path outside allowlist should be rejected");
assert!(err.to_string().contains("allowlist"));
}
#[test]
fn validate_mqtt_broker_url_requires_tls_ca_for_custom_ca_policy() {
let url = Url::parse("mqtts://broker.example.com:8883").expect("valid url");
let tls = MQTTTlsConfig::from_values(Some("custom_ca"), None, None, None, None, None).expect("valid tls config");
let err = validate_mqtt_broker_url(&url, &tls).expect_err("custom_ca policy without path should be rejected");
assert!(err.to_string().contains("tls_ca"));
}
}
helm/README.md
+2 -1
View File
@@ -47,7 +47,7 @@ RustFS helm chart supports **standalone and distributed mode**. For standalone m
| config.rustfs.obs_endpoint.logs.endpoint | string | `""` | Remote endpoint url for logs. |
| config.rustfs.obs_endpoint.profiling.enabled | bool | `false` | Whether to send profiling to remote endpoint. |
| config.rustfs.obs_endpoint.profiling.endpoint | string | `""` | Remote endpoint url for profiling. |
| extraEnv | list | `[]` | Extra environment variables for RustFS container. |
| extraEnv | map | `[]` | Extra environment variables for RustFS container. |
| containerSecurityContext.capabilities.drop[0] | string | `"ALL"` | |
| containerSecurityContext.readOnlyRootFilesystem | bool | `true` | |
| containerSecurityContext.runAsNonRoot | bool | `true` | |
@@ -133,6 +133,7 @@ RustFS helm chart supports **standalone and distributed mode**. For standalone m
| storageclass.dataStorageSize | string | `"256Mi"` | The storage size for data PVC. |
| storageclass.logStorageSize | string | `"256Mi"` | The storage size for logs PVC. |
| storageclass.name | string | `"local-path"` | The name for StorageClass. |
| storageclass.pvcAnnotations | map | `{}` | PVC customized annotations. |
| tolerations | list | `[]` | |
| gatewayApi.enabled | bool | `false` | To enable/disable gateway api support. |
| gatewayApi.gatewayClass | string | `traefik` | Gateway class implementation. |
helm/rustfs/templates/pvc.yaml
+6
View File
@@ -5,6 +5,9 @@ kind: PersistentVolumeClaim
metadata:
annotations:
helm.sh/resource-policy: keep
{{- with .pvcAnnotations }}
{{- toYaml .| nindent 4 }}
{{- end }}
name: {{ include "rustfs.fullname" $ }}-data
namespace: {{ $.Release.Namespace }}
labels:
@@ -28,6 +31,9 @@ kind: PersistentVolumeClaim
metadata:
annotations:
helm.sh/resource-policy: keep
{{- with .pvcAnnotations }}
{{- toYaml .| nindent 4 }}
{{- end }}
name: {{ include "rustfs.fullname" $ }}-logs
namespace: {{ $.Release.Namespace }}
labels:
helm/rustfs/templates/statefulset.yaml
+6
View File
@@ -192,6 +192,8 @@ spec:
name: logs
labels:
{{- toYaml .Values.commonLabels | nindent 10 }}
annotations:
{{- toYaml .Values.storageclass.pvcAnnotations | nindent 10 }}
spec:
accessModes: ["ReadWriteOnce"]
storageClassName: {{ .Values.storageclass.name }}
@@ -204,6 +206,8 @@ spec:
name: data-rustfs-{{ $i }}
labels:
{{- toYaml $.Values.commonLabels | nindent 10 }}
annotations:
{{- toYaml $.Values.storageclass.pvcAnnotations | nindent 10 }}
spec:
accessModes: ["ReadWriteOnce"]
storageClassName: {{ $.Values.storageclass.name }}
@@ -216,6 +220,8 @@ spec:
name: data
labels:
{{- toYaml .Values.commonLabels | nindent 10 }}
annotations:
{{- toYaml .Values.storageclass.pvcAnnotations | nindent 10 }}
spec:
accessModes: ["ReadWriteOnce"]
storageClassName: {{ .Values.storageclass.name }}
helm/rustfs/values.yaml
+4
View File
@@ -264,6 +264,10 @@ storageclass:
name: local-path
dataStorageSize: 256Mi
logStorageSize: 256Mi
pvcAnnotations: {}
# pvcAnnotations:
# - name: team
# value: "RustFS"
pdb:
create: false
rustfs/src/admin/handlers/audit.rs
+18 -3
View File
@@ -24,7 +24,7 @@ use rustfs_audit::{audit_system, start_audit_system as start_global_audit_system
use rustfs_config::audit::{AUDIT_MQTT_KEYS, AUDIT_MQTT_SUB_SYS, AUDIT_ROUTE_PREFIX, AUDIT_WEBHOOK_KEYS, AUDIT_WEBHOOK_SUB_SYS};
use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, EnableState, MAX_ADMIN_REQUEST_BODY_SIZE};
use rustfs_ecstore::config::Config;
use rustfs_targets::check_mqtt_broker_available;
use rustfs_targets::{TargetError, check_mqtt_broker_available_with_tls, target::mqtt::MQTTTlsConfig};
use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
@@ -521,9 +521,24 @@ impl Operation for AuditTargetConfig {
.ok_or_else(|| s3_error!(InvalidArgument, "topic is required"))?;
let username = kv_map.get(rustfs_config::MQTT_USERNAME).map(String::as_str);
let password = kv_map.get(rustfs_config::MQTT_PASSWORD).map(String::as_str);
check_mqtt_broker_available(endpoint, topic, username, password)
let tls = MQTTTlsConfig::from_values(
kv_map.get(rustfs_config::MQTT_TLS_POLICY).map(String::as_str),
kv_map.get(rustfs_config::MQTT_TLS_CA).map(String::as_str),
kv_map.get(rustfs_config::MQTT_TLS_CLIENT_CERT).map(String::as_str),
kv_map.get(rustfs_config::MQTT_TLS_CLIENT_KEY).map(String::as_str),
kv_map.get(rustfs_config::MQTT_TLS_TRUST_LEAF_AS_CA).map(String::as_str),
kv_map.get(rustfs_config::MQTT_WS_PATH_ALLOWLIST).map(String::as_str),
)
.map_err(|e| s3_error!(InvalidArgument, "invalid MQTT TLS settings: {}", e))?;
let parsed_broker = Url::parse(endpoint).map_err(|e| s3_error!(InvalidArgument, "invalid broker URL: {}", e))?;
rustfs_targets::target::mqtt::validate_mqtt_broker_url(&parsed_broker, &tls)
.map_err(|e| s3_error!(InvalidArgument, "{}", e))?;
check_mqtt_broker_available_with_tls(parsed_broker.as_str(), topic, username, password, &tls)
.await
.map_err(|e| s3_error!(InvalidArgument, "MQTT Broker unavailable: {}", e))?;
.map_err(|e| match e {
TargetError::Configuration(_) => s3_error!(InvalidArgument, "{}", e),
_ => s3_error!(InvalidArgument, "MQTT broker check failed: {}", e),
})?;
if let Some(queue_dir) = kv_map.get("queue_dir") {
validate_queue_dir(queue_dir.as_str()).await?;
rustfs/src/admin/handlers/event.rs
+18 -3
View File
@@ -25,7 +25,7 @@ use rustfs_config::notify::{
};
use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, EnableState, MAX_ADMIN_REQUEST_BODY_SIZE};
use rustfs_ecstore::config::Config;
use rustfs_targets::check_mqtt_broker_available;
use rustfs_targets::{TargetError, check_mqtt_broker_available_with_tls, target::mqtt::MQTTTlsConfig};
use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
@@ -450,9 +450,24 @@ impl Operation for NotificationTarget {
.ok_or_else(|| s3_error!(InvalidArgument, "topic is required"))?;
let username = kv_map.get(rustfs_config::MQTT_USERNAME).map(String::as_str);
let password = kv_map.get(rustfs_config::MQTT_PASSWORD).map(String::as_str);
check_mqtt_broker_available(endpoint, topic, username, password)
let tls = MQTTTlsConfig::from_values(
kv_map.get(rustfs_config::MQTT_TLS_POLICY).map(String::as_str),
kv_map.get(rustfs_config::MQTT_TLS_CA).map(String::as_str),
kv_map.get(rustfs_config::MQTT_TLS_CLIENT_CERT).map(String::as_str),
kv_map.get(rustfs_config::MQTT_TLS_CLIENT_KEY).map(String::as_str),
kv_map.get(rustfs_config::MQTT_TLS_TRUST_LEAF_AS_CA).map(String::as_str),
kv_map.get(rustfs_config::MQTT_WS_PATH_ALLOWLIST).map(String::as_str),
)
.map_err(|e| s3_error!(InvalidArgument, "invalid MQTT TLS settings: {}", e))?;
let parsed_broker = Url::parse(endpoint).map_err(|e| s3_error!(InvalidArgument, "invalid broker URL: {}", e))?;
rustfs_targets::target::mqtt::validate_mqtt_broker_url(&parsed_broker, &tls)
.map_err(|e| s3_error!(InvalidArgument, "{}", e))?;
check_mqtt_broker_available_with_tls(parsed_broker.as_str(), topic, username, password, &tls)
.await
.map_err(|e| s3_error!(InvalidArgument, "MQTT Broker unavailable: {}", e))?;
.map_err(|e| match e {
TargetError::Configuration(_) => s3_error!(InvalidArgument, "{}", e),
_ => s3_error!(InvalidArgument, "MQTT broker check failed: {}", e),
})?;
if let Some(queue_dir) = kv_map.get("queue_dir") {
validate_queue_dir(queue_dir.as_str()).await?;
rustfs/src/app/bucket_usecase.rs
+6 -6
View File
@@ -170,18 +170,18 @@ fn encode_list_objects_v2_value(value: &str, encoding_type: Option<&EncodingType
}
}
fn build_metadata_extension_user_metadata(user_defined: &HashMap<String, String>) -> Option<MinioUserMetadata> {
fn build_metadata_extension_user_metadata(user_defined: &HashMap<String, String>) -> Option<UserMetadataCollection> {
let mut items = extract_user_defined_metadata(user_defined)
.into_iter()
.filter(|(key, _)| !key.is_empty())
.map(|(key, value)| MinioMetadataEntry { key, value })
.map(|(key, value)| UserMetadataEntry { key, value })
.collect::<Vec<_>>();
items.sort_by(|left, right| left.key.cmp(&right.key));
if items.is_empty() {
None
} else {
Some(MinioUserMetadata { items })
Some(UserMetadataCollection { items })
}
}
@@ -2729,7 +2729,7 @@ mod tests {
assert_eq!(version.internal, Some(ObjectInternalInfo { k: 4, m: 2 }));
assert_eq!(
version.user_metadata.as_ref().map(|metadata| metadata.items.clone()),
Some(vec![MinioMetadataEntry {
Some(vec![UserMetadataEntry {
key: "project".to_string(),
value: "alpha".to_string(),
}])
@@ -2745,7 +2745,7 @@ mod tests {
assert!(marker.user_tags.is_none());
assert_eq!(
marker.user_metadata.as_ref().map(|metadata| metadata.items.clone()),
Some(vec![MinioMetadataEntry {
Some(vec![UserMetadataEntry {
key: "marker".to_string(),
value: "true".to_string(),
}])
@@ -2840,7 +2840,7 @@ mod tests {
assert!(object.owner.is_some());
assert_eq!(
object.user_metadata.as_ref().map(|metadata| metadata.items.clone()),
Some(vec![MinioMetadataEntry {
Some(vec![UserMetadataEntry {
key: "project".to_string(),
value: "alpha".to_string(),
}])
rustfs/src/app/multipart_usecase.rs
+7 -1
View File
@@ -23,7 +23,7 @@ use crate::storage::entity;
use crate::storage::helper::OperationHelper;
use crate::storage::options::{
copy_src_opts, extract_metadata, get_complete_multipart_upload_opts, get_content_sha256_with_query, get_opts,
parse_copy_source_range, put_opts,
parse_copy_source_range, put_opts, validate_archive_content_encoding,
};
use crate::storage::request_context::spawn_traced;
use crate::storage::s3_api::multipart::build_list_parts_output;
@@ -559,6 +559,12 @@ impl DefaultMultipartUsecase {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
validate_archive_content_encoding(
&key,
req.headers.get("content-type").and_then(|value| value.to_str().ok()),
req.headers.get("content-encoding").and_then(|value| value.to_str().ok()),
)?;
let mut metadata = extract_metadata(&req.headers);
if let Some(tags) = tagging {
rustfs/src/app/object_usecase.rs
+15
View File
@@ -38,6 +38,7 @@ use crate::storage::helper::OperationHelper;
use crate::storage::options::{
copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime_with_object_name,
filter_object_metadata, get_content_sha256_with_query, get_opts, normalize_content_encoding_for_storage, put_opts,
validate_archive_content_encoding,
};
use crate::storage::s3_api::multipart::parse_list_parts_params;
use crate::storage::s3_api::{acl, restore, select};
@@ -315,6 +316,20 @@ fn apply_put_request_metadata(
tagging: Option<TaggingHeader>,
storage_class: Option<StorageClass>,
) -> S3Result<()> {
let request_content_type = content_type.as_ref().map(ToString::to_string).or_else(|| {
headers
.get("content-type")
.and_then(|value| value.to_str().ok())
.map(ToOwned::to_owned)
});
let request_content_encoding = content_encoding.as_ref().map(ToString::to_string).or_else(|| {
headers
.get("content-encoding")
.and_then(|value| value.to_str().ok())
.map(ToOwned::to_owned)
});
validate_archive_content_encoding(object_name, request_content_type.as_deref(), request_content_encoding.as_deref())?;
if let Some(cache_control) = cache_control {
metadata.insert("cache-control".to_string(), cache_control.to_string());
}
rustfs/src/app/object_usecase/app_adapters.rs
+4
View File
@@ -104,6 +104,10 @@ impl ObjectIoCachedGetObjectSource for CachedGetObject {
self.last_modified.as_deref()
}
fn expires(&self) -> Option<&str> {
self.expires.as_deref()
}
fn cache_control(&self) -> Option<&str> {
self.cache_control.as_deref()
}
rustfs/src/server/compress.rs
+83 -13
View File
@@ -46,6 +46,8 @@ use rustfs_config::{
DEFAULT_COMPRESS_ENABLE, DEFAULT_COMPRESS_EXTENSIONS, DEFAULT_COMPRESS_MIME_TYPES, DEFAULT_COMPRESS_MIN_SIZE,
ENV_COMPRESS_ENABLE, ENV_COMPRESS_EXTENSIONS, ENV_COMPRESS_MIME_TYPES, ENV_COMPRESS_MIN_SIZE, EnableState,
};
use rustfs_ecstore::compress::{STANDARD_EXCLUDE_COMPRESS_CONTENT_TYPES, STANDARD_EXCLUDE_COMPRESS_EXTENSIONS};
use rustfs_utils::string::{has_pattern, has_string_suffix_in_slice};
use std::str::FromStr;
use tower_http::compression::predicate::Predicate;
use tracing::debug;
@@ -196,6 +198,20 @@ impl CompressionConfig {
None
}
pub(crate) fn is_excluded_filename(filename: &str) -> bool {
has_string_suffix_in_slice(&filename.to_ascii_lowercase(), STANDARD_EXCLUDE_COMPRESS_EXTENSIONS)
}
pub(crate) fn is_excluded_mime_type(content_type: &str) -> bool {
let main_type = content_type
.split(';')
.next()
.unwrap_or(content_type)
.trim()
.to_ascii_lowercase();
!main_type.is_empty() && has_pattern(STANDARD_EXCLUDE_COMPRESS_CONTENT_TYPES, &main_type)
}
}
impl Default for CompressionConfig {
@@ -299,23 +315,37 @@ impl Predicate for CompressionPredicate {
return false;
}
// Check if the response matches configured extension via Content-Disposition
// Hard-stop archive/media/package MIME types even if the whitelist matches.
// This includes tar, gzip, bzip2, xz, zstd, zip, rar, 7z, lzip, lzma, lzop variants,
// plus video/*, audio/*, image/*, font/*, application/pdf, and application/wasm.
if let Some(content_type) = response.headers().get(http::header::CONTENT_TYPE)
&& let Ok(ct) = content_type.to_str()
{
if CompressionConfig::is_excluded_mime_type(ct) {
debug!("Skipping compression for excluded Content-Type '{}'", ct);
return false;
}
if self.config.matches_mime_type(ct) {
debug!("Compressing response: Content-Type '{}' matches configured MIME pattern", ct);
return true;
}
}
// Hard-stop archive-like attachment downloads even if the whitelist matches.
if let Some(content_disposition) = response.headers().get(http::header::CONTENT_DISPOSITION)
&& let Ok(cd) = content_disposition.to_str()
&& let Some(filename) = CompressionConfig::extract_filename_from_content_disposition(cd)
&& self.config.matches_extension(&filename)
{
debug!("Compressing response: filename '{}' matches configured extension", filename);
return true;
}
if CompressionConfig::is_excluded_filename(&filename) {
debug!("Skipping compression for excluded filename '{}'", filename);
return false;
}
// Check if the response matches configured MIME type
if let Some(content_type) = response.headers().get(http::header::CONTENT_TYPE)
&& let Ok(ct) = content_type.to_str()
&& self.config.matches_mime_type(ct)
{
debug!("Compressing response: Content-Type '{}' matches configured MIME pattern", ct);
return true;
if self.config.matches_extension(&filename) {
debug!("Compressing response: filename '{}' matches configured extension", filename);
return true;
}
}
// Default: don't compress (whitelist approach)
@@ -405,7 +435,11 @@ use std::task::{Context, Poll};
use tower::{Layer, Service};
/// Tower layer that injects `RequestPathCategory` into each response's extensions
/// based on the incoming request URI path. Must be placed before `CompressionLayer`.
/// based on the incoming request URI path.
///
/// It must be placed inside `CompressionLayer` so the category is available when
/// the outer compression middleware evaluates its response predicate. With
/// `tower::ServiceBuilder`, that means adding this layer after `CompressionLayer`.
#[derive(Clone, Copy, Debug)]
pub(crate) struct PathCategoryInjectionLayer;
@@ -627,6 +661,42 @@ mod tests {
assert_eq!(predicate.config.min_size, 1000);
}
#[test]
fn test_compression_predicate_skips_archive_mime_type_even_when_whitelisted() {
let predicate = CompressionPredicate::new(CompressionConfig {
enabled: true,
extensions: vec![],
mime_patterns: vec!["application/zip".to_string()],
min_size: 0,
});
let response = Response::builder()
.header(http::header::CONTENT_TYPE, "application/zip")
.header(http::header::CONTENT_LENGTH, "4096")
.body(http_body_util::Empty::<bytes::Bytes>::new())
.expect("response");
assert!(!predicate.should_compress(&response));
}
#[test]
fn test_compression_predicate_skips_archive_filename_even_when_whitelisted() {
let predicate = CompressionPredicate::new(CompressionConfig {
enabled: true,
extensions: vec![".zip".to_string()],
mime_patterns: vec![],
min_size: 0,
});
let response = Response::builder()
.header(http::header::CONTENT_DISPOSITION, r#"attachment; filename="bundle.zip""#)
.header(http::header::CONTENT_LENGTH, "4096")
.body(http_body_util::Empty::<bytes::Bytes>::new())
.expect("response");
assert!(!predicate.should_compress(&response));
}
#[test]
fn test_path_category_classify_s3() {
assert_eq!(PathCategory::classify("/"), PathCategory::S3DataPlane);
rustfs/src/server/http.rs
+96 -3
View File
@@ -603,8 +603,8 @@ fn process_connection(
// 9. KeystoneAuthLayer — X-Auth-Token validation
// 10. TraceLayer — request/response tracing + metrics
// 11. PropagateRequestIdLayer — X-Request-ID → response
// 12. PathCategoryInjectionLayer — injects path category for compression
// 13. CompressionLayer — response compression (whitelist, path-aware)
// 12. CompressionLayer — response compression (whitelist, path-aware)
// 13. PathCategoryInjectionLayer — injects path category for compression predicate
// 14. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes
// 15. ConditionalCorsLayer — S3 API CORS
// 16. RedirectLayer — console redirect (conditional)
@@ -739,8 +739,8 @@ fn process_connection(
.layer(PropagateRequestIdLayer::x_request_id())
// Compress responses based on whitelist configuration
// Only compresses when enabled and matches configured extensions/MIME types
.layer(PathCategoryInjectionLayer)
.layer(CompressionLayer::new().compress_when(PathAwareCompressionPredicate::new(compression_config)))
.layer(PathCategoryInjectionLayer)
.layer(ObjectAttributesEtagFixLayer)
// Conditional CORS layer: only applies to S3 API requests (not Admin, not Console)
// Admin has its own CORS handling in router.rs
@@ -927,8 +927,16 @@ fn get_default_tcp_keepalive() -> TcpKeepalive {
#[cfg(test)]
mod tests {
use super::*;
use crate::server::compress::RequestPathCategory;
use bytes::Bytes;
use http::HeaderMap;
use http::Request as HttpRequest;
use http_body_util::Empty;
use opentelemetry::propagation::Extractor;
use std::convert::Infallible;
use std::future::Ready;
use std::task::{Context, Poll};
use tower::{Layer, Service, ServiceBuilder};
/// Baseline constants — reference the authoritative config defaults.
/// If a config default changes, tests automatically follow.
@@ -1065,4 +1073,89 @@ mod tests {
assert_eq!(carrier.get("Content-Type"), Some("application/json"));
assert_eq!(carrier.get("CONTENT-TYPE"), Some("application/json"));
}
#[derive(Clone, Copy)]
struct ObserveCategoryLayer;
#[derive(Clone)]
struct ObserveCategoryService<S> {
inner: S,
}
impl<S> Layer<S> for ObserveCategoryLayer {
type Service = ObserveCategoryService<S>;
fn layer(&self, inner: S) -> Self::Service {
ObserveCategoryService { inner }
}
}
impl<S, ReqBody, ResBody> Service<HttpRequest<ReqBody>> for ObserveCategoryService<S>
where
S: Service<HttpRequest<ReqBody>, Response = Response<ResBody>, Error = Infallible>,
{
type Response = Response<ResBody>;
type Error = Infallible;
type Future = Ready<std::result::Result<Response<ResBody>, Infallible>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
let response = futures::executor::block_on(self.inner.call(req)).expect("infallible");
let mut response = response;
let seen = response.extensions().get::<RequestPathCategory>().is_some();
response
.headers_mut()
.insert("x-category-seen", if seen { "true" } else { "false" }.parse().expect("header"));
std::future::ready(Ok(response))
}
}
#[derive(Clone, Copy)]
struct OkService;
impl<ReqBody> Service<HttpRequest<ReqBody>> for OkService {
type Response = Response<Empty<Bytes>>;
type Error = Infallible;
type Future = Ready<std::result::Result<Response<Empty<Bytes>>, Infallible>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _req: HttpRequest<ReqBody>) -> Self::Future {
std::future::ready(Ok(Response::new(Empty::new())))
}
}
#[test]
fn test_service_builder_order_regression_for_response_extensions() {
let request = HttpRequest::builder().uri("/bucket/archive.zip").body(()).expect("request");
let mut broken_order = ServiceBuilder::new()
.layer(PathCategoryInjectionLayer)
.layer(ObserveCategoryLayer)
.service(OkService);
let broken_response = futures::executor::block_on(broken_order.call(request)).expect("response");
assert_eq!(
broken_response.headers().get("x-category-seen").and_then(|v| v.to_str().ok()),
Some("false")
);
let request = HttpRequest::builder().uri("/bucket/archive.zip").body(()).expect("request");
let mut fixed_order = ServiceBuilder::new()
.layer(ObserveCategoryLayer)
.layer(PathCategoryInjectionLayer)
.service(OkService);
let fixed_response = futures::executor::block_on(fixed_order.call(request)).expect("response");
assert_eq!(
fixed_response.headers().get("x-category-seen").and_then(|v| v.to_str().ok()),
Some("true")
);
}
}
rustfs/src/storage/options.rs
+94 -1
View File
@@ -33,7 +33,7 @@ use rustfs_policy::service_type::ServiceType;
use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
use rustfs_utils::http::AMZ_CONTENT_SHA256;
use rustfs_utils::path::is_dir_object;
use s3s::{S3Result, s3_error};
use s3s::{S3Error, S3ErrorCode, S3Result, s3_error};
use std::collections::HashMap;
use std::sync::LazyLock;
use tracing::error;
@@ -349,6 +349,73 @@ pub(crate) fn normalize_content_encoding_for_storage(value: &str) -> Option<Stri
if normalized.is_empty() { None } else { Some(normalized) }
}
const ENV_ALLOW_ARCHIVE_CONTENT_ENCODING: &str = "RUSTFS_ALLOW_ARCHIVE_CONTENT_ENCODING";
const ARCHIVE_CONTENT_ENCODING_BLOCKED_SUFFIXES: &[&str] = &[
".zip",
".tar",
".tar.gz",
".tgz",
".tar.bz2",
".tbz",
".tbz2",
".tar.xz",
".txz",
".tar.zst",
".tar.zstd",
".tzst",
];
const ARCHIVE_CONTENT_ENCODING_BLOCKED_CONTENT_TYPES: &[&str] =
&["application/zip", "application/x-zip-compressed", "application/x-tar"];
fn is_archive_object_name_for_content_encoding(object_name: &str) -> bool {
let object_name = object_name.to_ascii_lowercase();
ARCHIVE_CONTENT_ENCODING_BLOCKED_SUFFIXES
.iter()
.any(|suffix| object_name.ends_with(suffix))
}
fn is_archive_content_type_for_content_encoding(content_type: &str) -> bool {
let main_type = content_type
.split(';')
.next()
.unwrap_or(content_type)
.trim()
.to_ascii_lowercase();
ARCHIVE_CONTENT_ENCODING_BLOCKED_CONTENT_TYPES
.iter()
.any(|candidate| main_type == *candidate)
}
pub(crate) fn validate_archive_content_encoding(
object_name: &str,
content_type: Option<&str>,
content_encoding: Option<&str>,
) -> S3Result<()> {
if rustfs_utils::get_env_bool(ENV_ALLOW_ARCHIVE_CONTENT_ENCODING, false) {
return Ok(());
}
let Some(content_encoding) = content_encoding.map(str::trim).filter(|value| !value.is_empty()) else {
return Ok(());
};
let is_archive_like = is_archive_object_name_for_content_encoding(object_name)
|| content_type.is_some_and(is_archive_content_type_for_content_encoding);
if !is_archive_like {
return Ok(());
}
Err(S3Error::with_message(
S3ErrorCode::InvalidArgument,
format!(
"Content-Encoding '{content_encoding}' is not allowed for archive objects by default; set {ENV_ALLOW_ARCHIVE_CONTENT_ENCODING}=true to allow legacy behavior"
),
))
}
/// Extracts metadata from headers and returns it as a HashMap with object name for MIME type detection.
pub fn extract_metadata_from_mime_with_object_name(
headers: &HeaderMap<HeaderValue>,
@@ -692,6 +759,8 @@ fn get_content_sha256_cksum(headers: &HeaderMap<HeaderValue>, service_type: Serv
#[cfg(test)]
mod tests {
use temp_env;
use super::*;
use http::{HeaderMap, HeaderValue};
use std::collections::HashMap;
@@ -1358,6 +1427,30 @@ mod tests {
assert_eq!(detect_content_type_from_object_name("noextension"), "application/octet-stream");
}
#[test]
fn test_validate_archive_content_encoding_rejects_archive_suffix_by_default() {
let err = validate_archive_content_encoding("bundle.tar.gz", Some("application/gzip"), Some("gzip")).unwrap_err();
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
}
#[test]
fn test_validate_archive_content_encoding_rejects_archive_mime_by_default() {
let err = validate_archive_content_encoding("bundle", Some("application/zip"), Some("gzip")).unwrap_err();
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
}
#[test]
fn test_validate_archive_content_encoding_allows_non_archive_precompressed_object() {
validate_archive_content_encoding("logs/app.log.zst", Some("text/plain"), Some("zstd")).expect("non-archive");
}
#[test]
fn test_validate_archive_content_encoding_allows_legacy_opt_in() {
temp_env::with_var(ENV_ALLOW_ARCHIVE_CONTENT_ENCODING, Some("true"), || {
validate_archive_content_encoding("bundle.zip", Some("application/zip"), Some("gzip")).expect("legacy opt-in");
});
}
#[test]
fn test_parse_copy_source_range() {
// Test complete range: bytes=0-1023