fix(s3): align snowball member semantics (#6944)

* fix(s3): harden Snowball extract error boundaries

* fix(s3): close Snowball extract compatibility gaps

* fix(s3): verify Snowball request body completion

* test(s3): reject forged Snowball streaming signatures

* build(deps): pin Snowball archive parser limits

* fix(s3): preserve Snowball trailer and member errors

* docs(architecture): register Snowball tar fork cleanup

* refactor(s3): route Snowball errors through object boundary

* ci(deps): allow pinned tokio-tar source

* fix: align Snowball archive codec detection

* fix(s3): harden Snowball codec compatibility

* fix(s3): preserve Snowball codec compatibility

* test(zip): align yield wake assertion with Tokio

* fix(rio): preserve legacy large-block reads

* fix(zip): accept blank tar numeric fields

* fix(s3): align Snowball member import semantics

* fix(s3): authorize PAX legal-hold conditions

* refactor(s3): preserve Snowball error boundary

* fix(iam): support legal-hold policy conditions

crates/e2e_test/src/multipart_auth_test.rs

@@ -4467,9 +4467,15 @@ async fn test_signed_put_object_extract_authorizes_each_pax_privilege_and_retent
     let context_archive_resources = [
         format!("arn:aws:s3:::{bucket}/tag-context.tar"),
         format!("arn:aws:s3:::{bucket}/lock-context.tar"),
+        format!("arn:aws:s3:::{bucket}/legal-hold-context.tar"),
+        format!("arn:aws:s3:::{bucket}/user-agent-bypass.tar"),
+        format!("arn:aws:s3:::{bucket}/sse-bypass.tar"),
     ];
     let tag_entry_resource = format!("arn:aws:s3:::{bucket}/tag-context-entry.txt");
     let lock_entry_resource = format!("arn:aws:s3:::{bucket}/lock-context-entry.txt");
+    let legal_hold_entry_resource = format!("arn:aws:s3:::{bucket}/legal-hold-context-entry.txt");
+    let user_agent_entry_resource = format!("arn:aws:s3:::{bucket}/user-agent-bypass-entry.txt");
+    let sse_entry_resource = format!("arn:aws:s3:::{bucket}/sse-bypass-entry.txt");
     let policy = serde_json::json!({
         "Version": "2012-10-17",
         "Statement": [
@@ -4529,7 +4535,7 @@ async fn test_signed_put_object_extract_authorizes_each_pax_privilege_and_retent
                 "Sid": "PaxContextArchives",
                 "Effect": "Allow",
                 "Principal": { "AWS": [pax_context_user] },
-                "Action": ["s3:PutObject", "s3:PutObjectRetention", "s3:PutObjectTagging"],
+                "Action": ["s3:PutObject", "s3:PutObjectRetention", "s3:PutObjectLegalHold", "s3:PutObjectTagging"],
                 "Resource": context_archive_resources
             },
             {
@@ -4569,6 +4575,49 @@ async fn test_signed_put_object_extract_authorizes_each_pax_privilege_and_retent
                 "Principal": { "AWS": [pax_context_user] },
                 "Action": ["s3:PutObjectRetention"],
                 "Resource": [lock_entry_resource]
+            },
+            {
+                "Sid": "PaxLegalHoldContextPut",
+                "Effect": "Allow",
+                "Principal": { "AWS": [pax_context_user] },
+                "Action": ["s3:PutObject"],
+                "Resource": [legal_hold_entry_resource.clone()]
+            },
+            {
+                "Sid": "PaxLegalHoldContextAction",
+                "Effect": "Allow",
+                "Principal": { "AWS": [pax_context_user] },
+                "Action": ["s3:PutObjectLegalHold"],
+                "Resource": [legal_hold_entry_resource],
+                "Condition": {
+                    "StringEquals": {
+                        "s3:object-lock-legal-hold": "OFF"
+                    }
+                }
+            },
+            {
+                "Sid": "MemberUserAgentCondition",
+                "Effect": "Allow",
+                "Principal": { "AWS": [pax_context_user] },
+                "Action": ["s3:PutObject"],
+                "Resource": [user_agent_entry_resource],
+                "Condition": {
+                    "StringEquals": {
+                        "aws:UserAgent": "trusted"
+                    }
+                }
+            },
+            {
+                "Sid": "MemberSseCondition",
+                "Effect": "Allow",
+                "Principal": { "AWS": [pax_context_user] },
+                "Action": ["s3:PutObject"],
+                "Resource": [sse_entry_resource],
+                "Condition": {
+                    "StringEquals": {
+                        "s3:x-amz-server-side-encryption": "AES256"
+                    }
+                }
             }
         ]
     })
@@ -4581,9 +4630,14 @@ async fn test_signed_put_object_extract_authorizes_each_pax_privilege_and_retent
     let cases = [
         (
             "legal-hold.tar",
-            put_only_client,
+            put_only_client.clone(),
             HashMap::from([("minio.metadata.x-amz-object-lock-legal-hold", "ON".to_string())]),
         ),
+        (
+            "tagging.tar",
+            put_only_client,
+            HashMap::from([("minio.metadata.x-amz-tagging", "classification=restricted".to_string())]),
+        ),
         (
             "retention-condition.tar",
             conditional_client,
@@ -4670,6 +4724,57 @@ async fn test_signed_put_object_extract_authorizes_each_pax_privilege_and_retent
     assert_eq!(stored.body.collect().await?.into_bytes().as_ref(), b"condition-body");
 
     let pax_context_client = restricted_user_client(&env, pax_context_user, pax_context_secret);
+    for (archive_key, entry_key, pax_key, injected_value, outer_user_agent) in [
+        (
+            "user-agent-bypass.tar",
+            "user-agent-bypass-entry.txt",
+            "minio.metadata.user-agent",
+            "trusted",
+            Some("untrusted"),
+        ),
+        (
+            "sse-bypass.tar",
+            "sse-bypass-entry.txt",
+            "minio.metadata.x-amz-server-side-encryption",
+            "AES256",
+            None,
+        ),
+    ] {
+        let pax = HashMap::from([(pax_key, injected_value.to_string())]);
+        let archive = make_tar_with_pax_entry(entry_key, b"must-not-write", None, &pax).await;
+        let err = pax_context_client
+            .put_object()
+            .bucket(bucket)
+            .key(archive_key)
+            .body(ByteStream::from(archive))
+            .customize()
+            .mutate_request(move |req| {
+                req.headers_mut().insert("x-amz-meta-snowball-auto-extract", "true");
+                if let Some(user_agent) = outer_user_agent {
+                    req.headers_mut().insert("user-agent", user_agent);
+                }
+            })
+            .send()
+            .await
+            .expect_err("PAX metadata must not satisfy unrelated IAM request conditions");
+        assert_eq!(
+            err.as_service_error().and_then(|error| error.meta().code()),
+            Some("AccessDenied"),
+            "{archive_key}"
+        );
+        let err = admin_client
+            .head_object()
+            .bucket(bucket)
+            .key(entry_key)
+            .send()
+            .await
+            .expect_err("a denied PAX member must not be written");
+        assert!(matches!(
+            err.as_service_error().and_then(|error| error.meta().code()),
+            Some("NoSuchKey" | "NotFound")
+        ));
+    }
+
     let tag_pax = HashMap::from([("minio.metadata.x-amz-tagging", "classification=public".to_string())]);
     let archive = make_tar_with_pax_entry("tag-context-entry.txt", b"tag-context-body", None, &tag_pax).await;
     pax_context_client
@@ -4733,6 +4838,34 @@ async fn test_signed_put_object_extract_authorizes_each_pax_privilege_and_retent
         pax_retain_until
     );
 
+    let legal_hold_pax = HashMap::from([("minio.metadata.x-amz-object-lock-legal-hold", "ON".to_string())]);
+    let archive = make_tar_with_pax_entry("legal-hold-context-entry.txt", b"must-not-write", None, &legal_hold_pax).await;
+    let err = pax_context_client
+        .put_object()
+        .bucket(bucket)
+        .key("legal-hold-context.tar")
+        .object_lock_legal_hold_status(aws_sdk_s3::types::ObjectLockLegalHoldStatus::Off)
+        .body(ByteStream::from(archive))
+        .customize()
+        .mutate_request(|req| {
+            req.headers_mut().insert("x-amz-meta-snowball-auto-extract", "true");
+        })
+        .send()
+        .await
+        .expect_err("PAX legal hold must replace the outer value in the member IAM condition context");
+    assert_eq!(err.as_service_error().and_then(|error| error.meta().code()), Some("AccessDenied"));
+    let err = admin_client
+        .head_object()
+        .bucket(bucket)
+        .key("legal-hold-context-entry.txt")
+        .send()
+        .await
+        .expect_err("a denied PAX legal-hold member must not be written");
+    assert!(matches!(
+        err.as_service_error().and_then(|error| error.meta().code()),
+        Some("NoSuchKey" | "NotFound")
+    ));
+
     Ok(())
 }
 

crates/e2e_test/src/snowball_auto_extract_test.rs

@@ -21,6 +21,53 @@ mod tests {
     use std::error::Error;
     use std::io::{Cursor, Write};
 
+    fn pax_record(key: &str, value: &str) -> Vec<u8> {
+        let payload = format!("{key}={value}\n");
+        let mut len = payload.len() + 3;
+        loop {
+            let record = format!("{len} {payload}");
+            if record.len() == len {
+                return record.into_bytes();
+            }
+            len = record.len();
+        }
+    }
+
+    async fn append_pax_header(
+        builder: &mut tokio_tar::Builder<Cursor<Vec<u8>>>,
+        entry_type: tokio_tar::EntryType,
+        records: &[(&str, &str)],
+    ) -> Result<(), Box<dyn Error + Send + Sync>> {
+        let mut payload = Vec::new();
+        for (key, value) in records {
+            payload.extend(pax_record(key, value));
+        }
+        let mut header = tokio_tar::Header::new_ustar();
+        header.set_entry_type(entry_type);
+        header.set_size(u64::try_from(payload.len()).expect("PAX payload length should fit in u64"));
+        header.set_mode(0o644);
+        header.set_cksum();
+        builder
+            .append_data(&mut header, "PaxHeaders.X/snowball", Cursor::new(payload))
+            .await?;
+        Ok(())
+    }
+
+    async fn append_typed_entry(
+        builder: &mut tokio_tar::Builder<Cursor<Vec<u8>>>,
+        path: &str,
+        entry_type: tokio_tar::EntryType,
+        body: &[u8],
+    ) -> Result<(), Box<dyn Error + Send + Sync>> {
+        let mut header = tokio_tar::Header::new_gnu();
+        header.set_entry_type(entry_type);
+        header.set_size(u64::try_from(body.len()).expect("TAR member length should fit in u64"));
+        header.set_mode(0o644);
+        header.set_cksum();
+        builder.append_data(&mut header, path, Cursor::new(body)).await?;
+        Ok(())
+    }
+
     async fn build_test_archive() -> Result<Vec<u8>, Box<dyn Error + Send + Sync>> {
         let mut builder = tokio_tar::Builder::new(Cursor::new(Vec::new()));
 
@@ -147,6 +194,57 @@ mod tests {
         archive
     }
 
+    async fn build_member_semantics_archive() -> Result<Vec<u8>, Box<dyn Error + Send + Sync>> {
+        let mut builder = tokio_tar::Builder::new(Cursor::new(Vec::new()));
+        append_pax_header(
+            &mut builder,
+            tokio_tar::EntryType::XGlobalHeader,
+            &[
+                ("minio.metadata.x-amz-meta-owner", "global"),
+                ("minio.metadata.x-amz-meta-snowball-auto-extract", "true"),
+            ],
+        )
+        .await?;
+        append_pax_header(
+            &mut builder,
+            tokio_tar::EntryType::XHeader,
+            &[("minio.metadata.x-amz-meta-owner", "local")],
+        )
+        .await?;
+        append_typed_entry(&mut builder, "regular.txt", tokio_tar::EntryType::Regular, b"regular-body").await?;
+        for (path, entry_type) in [
+            ("char", tokio_tar::EntryType::Char),
+            ("block", tokio_tar::EntryType::Block),
+            ("fifo", tokio_tar::EntryType::Fifo),
+        ] {
+            append_typed_entry(&mut builder, path, entry_type, b"").await?;
+        }
+        let mut directory = tokio_tar::Header::new_gnu();
+        directory.set_entry_type(tokio_tar::EntryType::Directory);
+        directory.set_size(0);
+        directory.set_mode(0o755);
+        directory.set_cksum();
+        builder
+            .append_data(&mut directory, "directory/", Cursor::new(Vec::new()))
+            .await?;
+        for (path, entry_type) in [
+            ("hard-link", tokio_tar::EntryType::Link),
+            ("symlink", tokio_tar::EntryType::Symlink),
+            ("continuous", tokio_tar::EntryType::Continuous),
+            ("unknown", tokio_tar::EntryType::Other(b'9')),
+        ] {
+            append_typed_entry(&mut builder, path, entry_type, b"").await?;
+        }
+        Ok(builder.into_inner().await?.into_inner())
+    }
+
+    async fn build_versioned_member_archive(path: &str, version_id: &str) -> Result<Vec<u8>, Box<dyn Error + Send + Sync>> {
+        let mut builder = tokio_tar::Builder::new(Cursor::new(Vec::new()));
+        append_pax_header(&mut builder, tokio_tar::EntryType::XHeader, &[("minio.versionId", version_id)]).await?;
+        append_typed_entry(&mut builder, path, tokio_tar::EntryType::Regular, b"versioned-body").await?;
+        Ok(builder.into_inner().await?.into_inner())
+    }
+
     fn build_archive_with_invalid_utf8_entry() -> Vec<u8> {
         let mut archive = Vec::new();
         append_raw_tar_entry(&mut archive, b"invalid-\xff.txt", b"ignored-body");
@@ -199,6 +297,147 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn snowball_auto_extract_applies_member_semantics_and_metadata_precedence() -> Result<(), Box<dyn Error + Send + Sync>>
+    {
+        init_logging();
+
+        let mut env = RustFSTestEnvironment::new().await?;
+        env.start_rustfs_server(vec![]).await?;
+
+        let client = env.create_s3_client();
+        let bucket = "snowball-member-semantics";
+        client.create_bucket().bucket(bucket).send().await?;
+        client
+            .put_object()
+            .bucket(bucket)
+            .key("fixture.tar")
+            .metadata("Snowball-Auto-Extract", "true")
+            .metadata("Minio-Snowball-Prefix", "members")
+            .metadata("owner", "outer")
+            .body(ByteStream::from(build_member_semantics_archive().await?))
+            .send()
+            .await?;
+
+        let regular = client.head_object().bucket(bucket).key("members/regular.txt").send().await?;
+        let regular_metadata = regular.metadata().expect("regular member should expose metadata");
+        assert_eq!(regular_metadata.get("owner").map(String::as_str), Some("local"));
+        assert!(!regular_metadata.contains_key("snowball-auto-extract"));
+        assert!(!regular_metadata.contains_key("minio-snowball-prefix"));
+
+        for key in ["char", "block", "fifo"] {
+            let head = client
+                .head_object()
+                .bucket(bucket)
+                .key(format!("members/{key}"))
+                .send()
+                .await?;
+            assert_eq!(head.content_length(), Some(0), "{key} should be materialized as an empty object");
+            assert_eq!(
+                head.metadata().and_then(|metadata| metadata.get("owner")).map(String::as_str),
+                Some("outer"),
+                "{key} should not inherit global PAX metadata"
+            );
+        }
+        let directory = client.head_object().bucket(bucket).key("members/directory/").send().await?;
+        assert_eq!(directory.content_length(), Some(0));
+
+        for key in ["hard-link", "symlink", "continuous", "unknown"] {
+            let error = client
+                .head_object()
+                .bucket(bucket)
+                .key(format!("members/{key}"))
+                .send()
+                .await
+                .expect_err("unsupported TAR entry type must be skipped");
+            assert_eq!(error.into_service_error().code(), Some("NotFound"), "{key}");
+        }
+
+        env.stop_server();
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn snowball_auto_extract_validates_pax_version_id_against_bucket_state() -> Result<(), Box<dyn Error + Send + Sync>> {
+        init_logging();
+
+        let mut env = RustFSTestEnvironment::new().await?;
+        env.start_rustfs_server(vec![]).await?;
+
+        let client = env.create_s3_client();
+        let bucket = "snowball-version-semantics";
+        client.create_bucket().bucket(bucket).send().await?;
+
+        client
+            .put_object()
+            .bucket(bucket)
+            .key("null.tar")
+            .metadata("Snowball-Auto-Extract", "true")
+            .body(ByteStream::from(build_versioned_member_archive("null.txt", "null").await?))
+            .send()
+            .await?;
+        let null_member = client.get_object().bucket(bucket).key("null.txt").send().await?;
+        assert_eq!(null_member.body.collect().await?.into_bytes().as_ref(), b"versioned-body");
+
+        for (archive_key, member_key, version_id) in [
+            ("uuid.tar", "uuid.txt", uuid::Uuid::new_v4().to_string()),
+            ("uppercase-null.tar", "uppercase-null.txt", "NULL".to_string()),
+        ] {
+            let error = client
+                .put_object()
+                .bucket(bucket)
+                .key(archive_key)
+                .metadata("Snowball-Auto-Extract", "true")
+                .body(ByteStream::from(build_versioned_member_archive(member_key, &version_id).await?))
+                .send()
+                .await
+                .expect_err("invalid or unversioned UUID import must be rejected");
+            assert_eq!(error.into_service_error().code(), Some("InvalidArgument"), "{archive_key}");
+            let missing = client
+                .head_object()
+                .bucket(bucket)
+                .key(member_key)
+                .send()
+                .await
+                .expect_err("rejected version import must not create an object");
+            assert_eq!(missing.into_service_error().code(), Some("NotFound"), "{member_key}");
+        }
+
+        client
+            .put_bucket_versioning()
+            .bucket(bucket)
+            .versioning_configuration(
+                aws_sdk_s3::types::VersioningConfiguration::builder()
+                    .status(aws_sdk_s3::types::BucketVersioningStatus::Enabled)
+                    .build(),
+            )
+            .send()
+            .await?;
+        let imported_version_id = uuid::Uuid::new_v4().to_string();
+        client
+            .put_object()
+            .bucket(bucket)
+            .key("versioned-uuid.tar")
+            .metadata("Snowball-Auto-Extract", "true")
+            .body(ByteStream::from(
+                build_versioned_member_archive("versioned-uuid.txt", &imported_version_id).await?,
+            ))
+            .send()
+            .await?;
+        let imported = client
+            .get_object()
+            .bucket(bucket)
+            .key("versioned-uuid.txt")
+            .version_id(&imported_version_id)
+            .send()
+            .await?;
+        assert_eq!(imported.version_id(), Some(imported_version_id.as_str()));
+        assert_eq!(imported.body.collect().await?.into_bytes().as_ref(), b"versioned-body");
+
+        env.stop_server();
+        Ok(())
+    }
+
     #[tokio::test]
     async fn snowball_auto_extract_supports_standard_headers_with_combined_extract_options()
     -> Result<(), Box<dyn Error + Send + Sync>> {

crates/policy/src/policy/function/key_name.rs

@@ -198,6 +198,9 @@ pub enum S3KeyName {
     #[strum(serialize = "s3:object-lock-retain-until-date")]
     S3ObjectLockRetainUntilDate,
 
+    #[strum(serialize = "s3:object-lock-legal-hold")]
+    S3ObjectLockLegalHold,
+
     #[strum(serialize = "s3:object-lock-mode")]
     S3ObjectLockMode,
 
@@ -389,6 +392,7 @@ mod tests {
     #[test_case("s3:VersionId", KeyName::S3(S3KeyName::S3VersionId) ; "aws_version_id")]
     #[test_case("s3:versionid", KeyName::S3(S3KeyName::S3VersionId) ; "minio_version_id")]
     #[test_case("s3:object-lock-mode", KeyName::S3(S3KeyName::S3ObjectLockMode))]
+    #[test_case("s3:object-lock-legal-hold", KeyName::S3(S3KeyName::S3ObjectLockLegalHold))]
     #[test_case("aws:SecureTransport", KeyName::Aws(AwsKeyName::AWSSecureTransport))]
     #[test_case("jwt:sub", KeyName::Jwt(JwtKeyName::JWTSub))]
     #[test_case("ldap:user", KeyName::Ldap(LdapKeyName::User))]
@@ -412,6 +416,7 @@ mod tests {
     #[test_case("s3:VersionId", KeyName::S3(S3KeyName::S3VersionId) ; "aws_version_id")]
     #[test_case("s3:versionid", KeyName::S3(S3KeyName::S3VersionId) ; "minio_version_id")]
     #[test_case("s3:object-lock-mode", KeyName::S3(S3KeyName::S3ObjectLockMode))]
+    #[test_case("s3:object-lock-legal-hold", KeyName::S3(S3KeyName::S3ObjectLockLegalHold))]
     #[test_case("aws:SecureTransport", KeyName::Aws(AwsKeyName::AWSSecureTransport))]
     #[test_case("jwt:sub", KeyName::Jwt(JwtKeyName::JWTSub))]
     #[test_case("ldap:user", KeyName::Ldap(LdapKeyName::User))]
@@ -431,6 +436,7 @@ mod tests {
     #[test_case("s3:x-amz-copy-source", KeyName::S3(S3KeyName::S3XAmzCopySource))]
     #[test_case("s3:versionid", KeyName::S3(S3KeyName::S3VersionId))]
     #[test_case("s3:object-lock-mode", KeyName::S3(S3KeyName::S3ObjectLockMode))]
+    #[test_case("s3:object-lock-legal-hold", KeyName::S3(S3KeyName::S3ObjectLockLegalHold))]
     #[test_case("aws:SecureTransport", KeyName::Aws(AwsKeyName::AWSSecureTransport))]
     #[test_case("jwt:sub", KeyName::Jwt(JwtKeyName::JWTSub))]
     #[test_case("ldap:user", KeyName::Ldap(LdapKeyName::User))]

crates/policy/src/policy/function/string.rs

@@ -287,7 +287,7 @@ mod tests {
     };
     use std::collections::HashMap;
 
-    use crate::policy::function::key_name::S3KeyName::{S3LocationConstraint, S3ObjectLockMode};
+    use crate::policy::function::key_name::S3KeyName::{S3LocationConstraint, S3ObjectLockLegalHold, S3ObjectLockMode};
     use test_case::test_case;
 
     fn new_func(name: KeyName, variable: Option<String>, values: Vec<&str>) -> StringFunc {
@@ -309,6 +309,7 @@ mod tests {
     #[test_case(r#"{"aws:username/value": ["johndoe", "aaa"]}"#, new_func(Aws(AWSUsername), Some("value".into()), vec!["johndoe", "aaa"]
     ))]
     #[test_case(r#"{"s3:object-lock-mode": "COMPLIANCE"}"#, new_func(S3(S3ObjectLockMode), None, vec!["COMPLIANCE"]))]
+    #[test_case(r#"{"s3:object-lock-legal-hold": "ON"}"#, new_func(S3(S3ObjectLockLegalHold), None, vec!["ON"]))]
     fn test_deser(input: &str, expect: StringFunc) -> Result<(), serde_json::Error> {
         let v: StringFunc = serde_json::from_str(input)?;
         assert_eq!(v, expect);