mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-12 23:52:45 +00:00
fix(tier): drop reserved internal headers from transition PUT (#8363)
* fix(tier): drop reserved internal headers from transition PUT * test(tier): transition PUT against a target rejecting reserved headers The fake S3 target gains a mode that answers 400 InvalidArgument to any PutObject or UploadPart carrying an x-minio-internal-* request header, the rule MinIO and the Storj gateway enforce. RT-14 transitions an inline and a non-inline object through a real server to that target and asserts the stored bytes, both transition identity pairs and read-back. The e2e-full linux selection digest is refreshed for the new test. --------- Co-authored-by: Vector <vector@wcomp.io> Co-authored-by: cxymds <cxymds@gmail.com>
This commit is contained in:
shurkus
@@ -1,2 +1,2 @@
|
||||
sha256-darwin=98dff0961c6eb21d6128bfd2d4a6100a624ddf1fb7db05dba360a5d3d0a310c2
|
||||
sha256-linux=9ff0da7406de9b89c0e848ff29f2238b30ba9da89c7e9df97c246ce86631350f
|
||||
sha256-linux=890f5925b87e1bf3710c356253f3b2276d901e156df9ccc04ec9315374de6d22
|
||||
|
||||
@@ -511,6 +511,10 @@ struct StoreState {
|
||||
/// version id the target never had answers 404 `NoSuchVersion` instead of
|
||||
/// the idempotent 204 RustFS/MinIO give.
|
||||
reject_unknown_version_deletes: bool,
|
||||
/// Models MinIO and its derivatives (Storj): `X-Minio-Internal-` is a reserved
|
||||
/// metadata prefix, so a PutObject or UploadPart carrying any such request
|
||||
/// header is refused with `InvalidArgument` (rustfs#8362).
|
||||
reject_reserved_internal_headers: bool,
|
||||
limits: StoreLimits,
|
||||
buckets: HashMap<String, BucketState>,
|
||||
uploads: HashMap<String, MultipartState>,
|
||||
@@ -975,6 +979,43 @@ impl FakeS3Target {
|
||||
lock(&self.backend.store).reject_unknown_version_deletes = enabled;
|
||||
}
|
||||
|
||||
/// Refuse every PutObject / UploadPart that carries an `x-minio-internal-*`
|
||||
/// request header, the way MinIO and Storj do.
|
||||
pub fn reject_reserved_internal_headers(&self, enabled: bool) {
|
||||
lock(&self.backend.store).reject_reserved_internal_headers = enabled;
|
||||
}
|
||||
|
||||
/// Keys of every stored, non-delete object in the bucket, sorted.
|
||||
pub fn stored_keys(&self, bucket: &str) -> Vec<String> {
|
||||
let state = lock(&self.backend.store);
|
||||
let mut keys: Vec<String> = state
|
||||
.buckets
|
||||
.get(bucket)
|
||||
.map(|bucket| {
|
||||
bucket
|
||||
.objects
|
||||
.iter()
|
||||
.filter(|(_, versions)| versions.last().is_some_and(|version| !version.delete_marker))
|
||||
.map(|(key, _)| key.clone())
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
keys.sort();
|
||||
keys
|
||||
}
|
||||
|
||||
/// Body and user metadata (names as stored, without the `x-amz-meta-` prefix)
|
||||
/// of the latest non-delete version of an object.
|
||||
pub fn stored_object(&self, bucket: &str, key: &str) -> Option<(Bytes, HashMap<String, String>)> {
|
||||
lock(&self.backend.store)
|
||||
.buckets
|
||||
.get(bucket)
|
||||
.and_then(|bucket| bucket.objects.get(key))
|
||||
.and_then(|versions| versions.last())
|
||||
.filter(|version| !version.delete_marker)
|
||||
.map(|version| (version.body.clone(), version.metadata.clone().unwrap_or_default()))
|
||||
}
|
||||
|
||||
pub fn require_checksum_for_object_lock(&self, enabled: bool) {
|
||||
lock(&self.backend.store).require_checksum_for_object_lock = enabled;
|
||||
}
|
||||
@@ -1174,6 +1215,15 @@ impl S3Access for FaultAccess {
|
||||
"this target does not decode aws-chunked request bodies; send a plain signed payload with an exact Content-Length"
|
||||
));
|
||||
}
|
||||
if matches!(operation, Operation::PutObject | Operation::UploadPart)
|
||||
&& lock(&self.store).reject_reserved_internal_headers
|
||||
&& context
|
||||
.headers()
|
||||
.keys()
|
||||
.any(|name| name.as_str().starts_with("x-minio-internal-"))
|
||||
{
|
||||
return Err(s3s::s3_error!(InvalidArgument, "Your metadata headers are not supported."));
|
||||
}
|
||||
let prebody_permit = if operation == Operation::CompleteMultipartUpload {
|
||||
Some(
|
||||
timeout(
|
||||
|
||||
@@ -165,4 +165,210 @@ mod tests {
|
||||
info!("RT-13c PASS: scanner/config persists across restart");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// RT-14: a transition PUT must not carry reserved `x-minio-internal-*` headers.
|
||||
///
|
||||
/// Regression pattern: transition forwarded the object's internal metadata as raw
|
||||
/// request headers, so MinIO-derived remotes (Storj) refused every transition with
|
||||
/// `400 InvalidArgument` while `tier add` still succeeded (rustfs#8362). The fake
|
||||
/// target refuses any PutObject carrying such a header, like MinIO and Storj do.
|
||||
#[tokio::test]
|
||||
async fn test_transition_put_omits_reserved_internal_headers() -> TestResult {
|
||||
use crate::fake_s3_target::{BucketMode, FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, Operation};
|
||||
use aws_sdk_s3::primitives::ByteStream;
|
||||
use aws_sdk_s3::types::{
|
||||
BucketLifecycleConfiguration, ExpirationStatus, LifecycleRule, LifecycleRuleFilter, Transition,
|
||||
TransitionStorageClass,
|
||||
};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
init_logging();
|
||||
info!("RT-14: transition PUT omits reserved internal headers");
|
||||
|
||||
const TIER: &str = "RT14COLD";
|
||||
const TARGET_BUCKET: &str = "rt14-cold";
|
||||
const SOURCE_BUCKET: &str = "rt14-hot";
|
||||
|
||||
let target = FakeS3Target::start().await.expect("start fake target");
|
||||
target.create_bucket_with_mode(TARGET_BUCKET, BucketMode::Unversioned);
|
||||
target.reject_reserved_internal_headers(true);
|
||||
|
||||
let mut env = RustFSTestEnvironment::new().await.expect("create test environment");
|
||||
env.start_rustfs_server_with_env(
|
||||
vec![],
|
||||
&[
|
||||
("RUSTFS_CONSOLE_ENABLE", "false"),
|
||||
("RUSTFS_TIER_RUSTFS_ALLOW_LOOPBACK_ENDPOINT", "true"),
|
||||
],
|
||||
)
|
||||
.await
|
||||
.expect("start RustFS");
|
||||
|
||||
let tier_body = serde_json::json!({
|
||||
"type": "rustfs",
|
||||
"rustfs": {
|
||||
"name": TIER,
|
||||
"endpoint": target.endpoint(),
|
||||
"accessKey": FAKE_ACCESS_KEY,
|
||||
"secretKey": FAKE_SECRET_KEY,
|
||||
"bucket": TARGET_BUCKET,
|
||||
"prefix": "cold/",
|
||||
"region": "us-east-1",
|
||||
"storageClass": ""
|
||||
}
|
||||
})
|
||||
.to_string();
|
||||
let deadline = Instant::now() + Duration::from_secs(30);
|
||||
loop {
|
||||
match admin_ok(&env, http::Method::PUT, "/rustfs/admin/v3/tier", Some(tier_body.clone())).await {
|
||||
Ok(_) => break,
|
||||
Err(err) if Instant::now() < deadline => {
|
||||
info!(" add tier not ready yet: {err}");
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
Err(err) => panic!("RT-14 FAIL: add tier: {err}"),
|
||||
}
|
||||
}
|
||||
|
||||
let client = env.create_s3_client();
|
||||
client
|
||||
.create_bucket()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.send()
|
||||
.await
|
||||
.expect("create bucket");
|
||||
|
||||
let objects: Vec<(&str, Vec<u8>)> = vec![
|
||||
("small.bin", (0..1024u32).map(|i| (i % 251) as u8).collect()),
|
||||
("medium.bin", (0..200_000u32).map(|i| (i.wrapping_mul(31) % 253) as u8).collect()),
|
||||
];
|
||||
for (key, body) in &objects {
|
||||
client
|
||||
.put_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key(*key)
|
||||
.body(ByteStream::from(body.clone()))
|
||||
.send()
|
||||
.await
|
||||
.expect("put source object");
|
||||
}
|
||||
|
||||
let rule = LifecycleRule::builder()
|
||||
.id("to-cold")
|
||||
.status(ExpirationStatus::Enabled)
|
||||
.filter(LifecycleRuleFilter::builder().prefix("").build())
|
||||
.transitions(
|
||||
Transition::builder()
|
||||
.days(0)
|
||||
.storage_class(TransitionStorageClass::from(TIER))
|
||||
.build(),
|
||||
)
|
||||
.build()
|
||||
.expect("build rule");
|
||||
client
|
||||
.put_bucket_lifecycle_configuration()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.lifecycle_configuration(
|
||||
BucketLifecycleConfiguration::builder()
|
||||
.rules(rule)
|
||||
.build()
|
||||
.expect("build config"),
|
||||
)
|
||||
.send()
|
||||
.await
|
||||
.expect("put lifecycle");
|
||||
|
||||
admin_ok(
|
||||
&env,
|
||||
http::Method::POST,
|
||||
&format!("/rustfs/admin/v3/ilm/transition/run?bucket={SOURCE_BUCKET}&prefix=&tier={TIER}&dryRun=false"),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("run transition");
|
||||
|
||||
let deadline = Instant::now() + Duration::from_secs(60);
|
||||
while target.stored_keys(TARGET_BUCKET).len() < objects.len() {
|
||||
assert!(
|
||||
Instant::now() < deadline,
|
||||
"RT-14 FAIL: transition did not reach the target; stored keys: {:?}",
|
||||
target.stored_keys(TARGET_BUCKET)
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
|
||||
let keys = target.stored_keys(TARGET_BUCKET);
|
||||
assert_eq!(keys.len(), objects.len(), "RT-14 FAIL: unexpected remote objects: {keys:?}");
|
||||
|
||||
let mut stored_bodies = Vec::new();
|
||||
for key in &keys {
|
||||
let (body, metadata) = target.stored_object(TARGET_BUCKET, key).expect("remote object");
|
||||
for name in [
|
||||
"x-rustfs-internal-transition-transaction-id",
|
||||
"x-minio-internal-transition-transaction-id",
|
||||
"x-rustfs-internal-transition-tier-destination-id",
|
||||
"x-minio-internal-transition-tier-destination-id",
|
||||
] {
|
||||
assert!(
|
||||
metadata.contains_key(name),
|
||||
"RT-14 FAIL: remote object {key} lost transition identity {name}: {:?}",
|
||||
metadata.keys().collect::<Vec<_>>()
|
||||
);
|
||||
}
|
||||
stored_bodies.push(body.to_vec());
|
||||
}
|
||||
let mut expected_bodies: Vec<Vec<u8>> = objects.iter().map(|(_, body)| body.clone()).collect();
|
||||
expected_bodies.sort();
|
||||
stored_bodies.sort();
|
||||
assert_eq!(stored_bodies, expected_bodies, "RT-14 FAIL: remote bytes differ from the source objects");
|
||||
|
||||
let transition_puts = target
|
||||
.requests()
|
||||
.iter()
|
||||
.filter(|record| {
|
||||
record.operation == Operation::PutObject
|
||||
&& record
|
||||
.key
|
||||
.as_deref()
|
||||
.is_some_and(|key| key.contains("transition-transactions"))
|
||||
})
|
||||
.count();
|
||||
assert_eq!(
|
||||
transition_puts,
|
||||
objects.len(),
|
||||
"RT-14 FAIL: transition PUTs were retried or refused by the target"
|
||||
);
|
||||
|
||||
for (key, body) in &objects {
|
||||
let head = client
|
||||
.head_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key(*key)
|
||||
.send()
|
||||
.await
|
||||
.expect("head transitioned");
|
||||
assert_eq!(
|
||||
head.storage_class().map(|class| class.as_str()),
|
||||
Some(TIER),
|
||||
"RT-14 FAIL: {key} is not reported as transitioned"
|
||||
);
|
||||
let got = client
|
||||
.get_object()
|
||||
.bucket(SOURCE_BUCKET)
|
||||
.key(*key)
|
||||
.send()
|
||||
.await
|
||||
.expect("get transitioned")
|
||||
.body
|
||||
.collect()
|
||||
.await
|
||||
.expect("read body")
|
||||
.into_bytes();
|
||||
assert_eq!(got.as_ref(), body.as_slice(), "RT-14 FAIL: read-back of {key} differs");
|
||||
}
|
||||
|
||||
target.shutdown().await;
|
||||
info!("RT-14 PASS: transition PUT omits reserved internal headers");
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -298,6 +298,11 @@ pub fn build_transition_put_options(storage_class: String, mut metadata: HashMap
|
||||
}
|
||||
}
|
||||
|
||||
// x-minio-internal-* is reserved by MinIO and its derivatives; they reject
|
||||
// any request carrying it with 400 InvalidArgument. The transition ids above
|
||||
// are the only internal keys the remote needs, and they travel as x-amz-meta-*.
|
||||
metadata.retain(|key, _| !rustfs_utils::http::is_internal_key(key));
|
||||
|
||||
opts.user_metadata = metadata;
|
||||
opts
|
||||
}
|
||||
@@ -2244,6 +2249,38 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_transition_put_options_drops_reserved_internal_headers() {
|
||||
let transaction_suffix = rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TRANSACTION_ID;
|
||||
let mut metadata = HashMap::new();
|
||||
for suffix in [
|
||||
rustfs_utils::http::SUFFIX_INLINE_DATA,
|
||||
rustfs_utils::http::SUFFIX_ACTUAL_SIZE,
|
||||
rustfs_utils::http::SUFFIX_PART_CHECKSUMS,
|
||||
transaction_suffix,
|
||||
] {
|
||||
rustfs_utils::http::metadata_compat::insert_str(&mut metadata, suffix, "x".to_string());
|
||||
}
|
||||
metadata.insert("X-Minio-Internal-Compression".to_string(), "klauspost/compress/s2".to_string());
|
||||
metadata.insert("x-amz-meta-foo".to_string(), "bar".to_string());
|
||||
metadata.insert("name".to_string(), "object".to_string());
|
||||
|
||||
let opts = build_transition_put_options("COLD".to_string(), metadata);
|
||||
|
||||
let leaked: Vec<_> = opts
|
||||
.user_metadata
|
||||
.keys()
|
||||
.filter(|key| rustfs_utils::http::is_internal_key(key))
|
||||
.collect();
|
||||
assert!(leaked.is_empty(), "reserved internal keys must not reach the remote: {leaked:?}");
|
||||
assert_eq!(opts.user_metadata.get("x-amz-meta-foo").map(String::as_str), Some("bar"));
|
||||
assert_eq!(opts.user_metadata.get("name").map(String::as_str), Some("object"));
|
||||
assert!(opts.user_metadata.contains_key(&format!(
|
||||
"x-amz-meta-{}",
|
||||
rustfs_utils::http::metadata_compat::internal_key_rustfs(transaction_suffix)
|
||||
)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_transition_put_options_requests_no_checksum_and_content_md5() {
|
||||
// Regression for rustfs/rustfs#4811: transition uploads must leave the
|
||||
|
||||
Reference in New Issue
Block a user