fix(s3): persist the ACL, object lock and ownership a bucket is created with

crates/fakecloud-persistence/src/lib.rs

@@ -11,6 +11,6 @@ pub use s3::{
     AclGrantSnapshot, AclSnapshot, AnnotationSnapshot, BodyRef, BodySource, BucketMeta,
     BucketSnapshot, BucketSubresource, InventorySnapshot, LoadedMpu, LoadedObject, LoadedPart,
     MemoryS3Store, MpuInit, ObjectMeta, S3State as S3StateSnapshot, S3Store, StoreError,
-    StoreResult, TagsSnapshot, UploadPartMeta,
+    StoreResult, TagsSnapshot, UploadPartMeta, ALL_SUBRESOURCES,
 };
 pub use snapshot::{DiskSnapshotStore, MemorySnapshotStore, SnapshotHook, SnapshotStore};

crates/fakecloud-persistence/src/s3.rs

@@ -982,11 +982,29 @@ impl S3Store for DiskS3Store {
 
     fn delete_object(&self, bucket: &str, key: &str, version: Option<&str>) -> StoreResult<()> {
         let (dir, bin_path, toml_path) = self.object_paths(bucket, key, version);
-        for p in [&bin_path, &toml_path] {
-            match std::fs::remove_file(p) {
-                Ok(_) => {}
-                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
-                Err(e) => return Err(e.into()),
+        // Sidecar first, body second. `load` iterates `*.toml` and treats a
+        // sidecar whose `.bin` is gone as a hard error that costs the whole
+        // bucket, while an orphan `.bin` is never looked at. So if this is
+        // interrupted -- a signal, or the second `remove_file` failing -- the
+        // residue left behind has to be the body, not the sidecar.
+        // The sidecar is the durable record, so its removal is the one that has
+        // to succeed. Once it is gone the object is deleted as far as every
+        // reader is concerned, and failing the call over a body that could not
+        // be unlinked would report a delete that did happen as an error --
+        // while still leaving the caller no way to reclaim the bytes.
+        match std::fs::remove_file(&toml_path) {
+            Ok(_) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+            Err(e) => return Err(e.into()),
+        }
+        if let Err(e) = std::fs::remove_file(&bin_path) {
+            if e.kind() != std::io::ErrorKind::NotFound {
+                tracing::warn!(
+                    path = %bin_path.display(),
+                    error = %e,
+                    "deleted object's body could not be removed; it is orphaned until the \
+                     bucket is deleted",
+                );
             }
         }
         Self::cleanup_empty(&dir);
@@ -1504,6 +1522,56 @@ mod disk_tests {
         assert_eq!(obj.meta.tags.get("x").map(String::as_str), Some("y"));
     }
 
+    #[test]
+    fn delete_object_removes_the_sidecar_before_the_body() {
+        // `load` iterates `*.toml` and treats a sidecar whose `.bin` is gone as
+        // a hard error for the whole bucket, while an orphan `.bin` is never
+        // read. So the sidecar has to be the file the delete attempts first.
+        //
+        // Forces the SIDECAR removal to fail, by making that path a non-empty
+        // directory `remove_file` cannot take. Only a sidecar-first delete
+        // reports the failure with the body still on disk; a body-first one
+        // would have removed the body before reaching it.
+        let tmp = TempDir::new().unwrap();
+        let store = new_store(&tmp);
+        store
+            .put_bucket_meta(
+                "b",
+                &BucketMeta {
+                    name: "b".to_string(),
+                    ..Default::default()
+                },
+            )
+            .unwrap();
+        store
+            .put_object(
+                "b",
+                "k.txt",
+                None,
+                BodySource::Bytes(Bytes::from_static(b"v")),
+                &ObjectMeta {
+                    key: "k.txt".to_string(),
+                    size: 1,
+                    ..Default::default()
+                },
+            )
+            .unwrap();
+
+        let (_dir, bin, toml) = store.object_paths("b", "k.txt", None);
+        std::fs::remove_file(&toml).unwrap();
+        std::fs::create_dir(&toml).unwrap();
+        std::fs::write(toml.join("blocker"), b"x").unwrap();
+
+        assert!(
+            store.delete_object("b", "k.txt", None).is_err(),
+            "a sidecar that cannot be removed must fail the delete: it is the durable record"
+        );
+        assert!(
+            bin.exists(),
+            "the body must still be there, which is only true if the sidecar was attempted first"
+        );
+    }
+
     #[test]
     fn delete_object_cleans_up_files_and_cache() {
         let tmp = TempDir::new().unwrap();

crates/fakecloud-s3/src/persistence.rs

@@ -112,6 +112,24 @@ pub fn upload_part_meta_snapshot(p: &UploadPart) -> UploadPartMeta {
     }
 }
 
+/// Whether a stored grant names someone it could ever match.
+///
+/// An older build read an email grantee's absent `<ID>` as an empty string, so
+/// a legacy `acl.toml` can hold a grant that matches nobody. Dropping those at
+/// load keeps memory, disk and `GetBucketAcl` telling the same story -- hiding
+/// them only when rendering would make the response disagree with the ACL the
+/// bucket actually has, and a read-modify-write would then persist the omission.
+fn acl_grant_names_a_grantee(g: &AclGrantSnapshot) -> bool {
+    // Trimmed, like the request-side checks: a stored grantee of whitespace
+    // names nobody just as surely as an empty one.
+    let named = |v: &Option<String>| v.as_deref().is_some_and(|s| !s.trim().is_empty());
+    match g.grantee_type.as_str() {
+        "Group" => named(&g.grantee_uri),
+        "AmazonCustomerByEmail" => named(&g.grantee_display_name),
+        _ => named(&g.grantee_id),
+    }
+}
+
 fn acl_grant_from_snapshot(g: &AclGrantSnapshot) -> AclGrant {
     AclGrant {
         grantee_type: g.grantee_type.clone(),
@@ -247,7 +265,13 @@ pub fn s3_bucket_from_snapshot(
         grantee_uri: None,
         permission: "FULL_CONTROL".to_string(),
     };
-    let has_acl_sidecar = subresources.contains_key("acl.toml");
+    // An `acl.toml` written by an older build could be empty (its writer used
+    // `unwrap_or_default()` on serialization failure). Treat that as no sidecar
+    // at all, so the bucket keeps the default owner grant instead of loading
+    // with no grants and silently stripping the owner's FULL_CONTROL.
+    let has_acl_sidecar = subresources
+        .get("acl.toml")
+        .is_some_and(|text| !text.trim().is_empty());
     let mut b = S3Bucket {
         name: name.to_string(),
         creation_date: meta.creation_date,
@@ -330,7 +354,30 @@ pub fn s3_bucket_from_snapshot(
                 if !snap.owner_id.is_empty() {
                     b.acl_owner_id = snap.owner_id;
                 }
-                b.acl_grants = snap.grants.iter().map(acl_grant_from_snapshot).collect();
+                // Drop grants that name nobody, so the loaded ACL is one the
+                // server would also accept back from a client.
+                let usable: Vec<AclGrant> = snap
+                    .grants
+                    .iter()
+                    .filter(|g| acl_grant_names_a_grantee(g))
+                    .map(acl_grant_from_snapshot)
+                    .collect();
+                // ...but never leave the bucket with no grants at all. A
+                // sidecar whose every grant is unusable is indistinguishable
+                // from one that was never written, so fall back to the same
+                // default owner grant a bucket with no `acl.toml` gets, rather
+                // than loading a bucket whose owner has lost FULL_CONTROL.
+                b.acl_grants = if usable.is_empty() {
+                    vec![AclGrant {
+                        grantee_type: "CanonicalUser".to_string(),
+                        grantee_id: Some(b.acl_owner_id.clone()),
+                        grantee_display_name: Some(b.acl_owner_id.clone()),
+                        grantee_uri: None,
+                        permission: "FULL_CONTROL".to_string(),
+                    }]
+                } else {
+                    usable
+                };
             }
             "inventory.toml" => {
                 if text.trim().is_empty() {
@@ -345,19 +392,58 @@ pub fn s3_bucket_from_snapshot(
                 if text.trim().is_empty() {
                     continue;
                 }
-                b.analytics_configs = toml::from_str(&text).unwrap_or_default();
+                // Neither `unwrap_or_default` nor `?`. Defaulting would load the
+                // bucket with no configurations and no word, and the next Put
+                // would overwrite the file and make the loss permanent; failing
+                // would hide every object in the bucket behind an unparseable
+                // reporting config. Skip the sidecar, keep the bucket, and say so,
+                // so the file is still there to look at.
+                match toml::from_str(&text) {
+                    Ok(parsed) => b.analytics_configs = parsed,
+                    Err(e) => tracing::warn!(
+                        bucket = %name,
+                        error = %e,
+                        "ignoring unreadable analytics.toml; its configurations are not loaded",
+                    ),
+                }
             }
             "intelligent_tiering.toml" => {
                 if text.trim().is_empty() {
                     continue;
                 }
-                b.intelligent_tiering_configs = toml::from_str(&text).unwrap_or_default();
+                // Neither `unwrap_or_default` nor `?`. Defaulting would load the
+                // bucket with no configurations and no word, and the next Put
+                // would overwrite the file and make the loss permanent; failing
+                // would hide every object in the bucket behind an unparseable
+                // reporting config. Skip the sidecar, keep the bucket, and say so,
+                // so the file is still there to look at.
+                match toml::from_str(&text) {
+                    Ok(parsed) => b.intelligent_tiering_configs = parsed,
+                    Err(e) => tracing::warn!(
+                        bucket = %name,
+                        error = %e,
+                        "ignoring unreadable intelligent_tiering.toml; its configurations are not loaded",
+                    ),
+                }
             }
             "metrics.toml" => {
                 if text.trim().is_empty() {
                     continue;
                 }
-                b.metrics_configs = toml::from_str(&text).unwrap_or_default();
+                // Neither `unwrap_or_default` nor `?`. Defaulting would load the
+                // bucket with no configurations and no word, and the next Put
+                // would overwrite the file and make the loss permanent; failing
+                // would hide every object in the bucket behind an unparseable
+                // reporting config. Skip the sidecar, keep the bucket, and say so,
+                // so the file is still there to look at.
+                match toml::from_str(&text) {
+                    Ok(parsed) => b.metrics_configs = parsed,
+                    Err(e) => tracing::warn!(
+                        bucket = %name,
+                        error = %e,
+                        "ignoring unreadable metrics.toml; its configurations are not loaded",
+                    ),
+                }
             }
             "request_payment.toml" => {
                 b.request_payment = Some(text);
@@ -381,11 +467,41 @@ pub fn hydrate_s3_state(
     snapshot: S3StateSnapshot,
     account_id: &str,
     region: &str,
+) -> Result<S3State, String> {
+    hydrate_s3_state_reporting(snapshot, account_id, region, &mut |_, _| {})
+}
+
+/// Hydrate a loaded snapshot, reporting each bucket that cannot be used instead
+/// of failing the whole load.
+///
+/// The store already isolates a bucket whose objects it cannot read: it warns,
+/// records the refusal and carries on, so one bad file costs one bucket rather
+/// than the server. Sidecars are parsed here, a layer later, and a single
+/// malformed `acl.toml` or `tags.toml` used to abort startup -- every other
+/// bucket inaccessible because of one file. Such a bucket is skipped and handed
+/// to `refused` so it is reported exactly like a store-level refusal: absent
+/// from memory, its name refused by CreateBucket, and clearable by DeleteBucket.
+pub fn hydrate_s3_state_reporting(
+    snapshot: S3StateSnapshot,
+    account_id: &str,
+    region: &str,
+    refused: &mut dyn FnMut(&str, &str),
 ) -> Result<S3State, String> {
     let mut state = S3State::new(account_id, region);
     for (name, snap) in snapshot.buckets {
-        let bucket = s3_bucket_from_snapshot(&name, snap, region)?;
-        state.buckets.insert(name, bucket);
+        match s3_bucket_from_snapshot(&name, snap, region) {
+            Ok(bucket) => {
+                state.buckets.insert(name, bucket);
+            }
+            Err(e) => {
+                tracing::warn!(
+                    bucket = %name,
+                    error = %e,
+                    "skipping S3 bucket whose stored configuration could not be read",
+                );
+                refused(&name, &e);
+            }
+        }
     }
     Ok(state)
 }

crates/fakecloud-s3/src/service/acl.rs

@@ -6,8 +6,7 @@ use fakecloud_core::service::{AwsRequest, AwsResponse, AwsServiceError};
 use crate::persistence::object_meta_snapshot;
 
 use super::{
-    build_acl_xml, canned_acl_grants_for_object, no_such_key, parse_acl_xml, parse_grant_headers,
-    s3_xml, S3Service,
+    build_acl_xml, canned_acl_grants_for_object, no_such_key, parse_acl_xml, s3_xml, S3Service,
 };
 
 impl S3Service {
@@ -43,6 +42,12 @@ impl S3Service {
             .get("x-amz-acl")
             .and_then(|v| v.to_str().ok())
             .map(|s| s.to_string());
+        // Validated before the key is resolved, as S3 does: a bad canned value
+        // is an InvalidArgument whether or not the key exists.
+        if let Some(acl) = canned.as_deref() {
+            super::validate_object_canned_acl(acl)?;
+        }
+        super::reject_conflicting_acl_sources(canned.as_deref(), &req.headers, &req.body)?;
 
         if self.bucket_owner_enforced(account_id, bucket) {
             return Err(AwsServiceError::aws_error(
@@ -70,19 +75,35 @@ impl S3Service {
         let proposed_grants = if let Some(acl) = &canned {
             canned_acl_grants_for_object(acl, &owner_id)
         } else {
-            let has_grant_headers = req.headers.keys().any(|k| {
-                let name = k.as_str();
-                name.starts_with("x-amz-grant-")
-            });
-            if has_grant_headers {
-                parse_grant_headers(&req.headers)
+            if super::has_grant_headers(&req.headers) {
+                super::resolved_grant_headers(&req.headers)?
             } else {
+                // No canned header, no grant header, no body names no ACL at
+                // all. Re-persisting the object's current grants would answer
+                // 200 for a request that asked for nothing, which the caller
+                // cannot tell from an applied change -- the same rule
+                // PutBucketAcl applies.
                 let body_str = std::str::from_utf8(&req.body).unwrap_or("");
-                if !body_str.is_empty() {
-                    parse_acl_xml(body_str)?
-                } else {
-                    obj.acl_grants.clone()
+                if body_str.trim().is_empty() {
+                    return Err(AwsServiceError::aws_error(
+                        StatusCode::BAD_REQUEST,
+                        "MalformedACLError",
+                        "The XML you provided was not well-formed or did not validate against our published schema",
+                    ));
+                }
+                // A body that is not an AccessControlPolicy at all (JSON, a
+                // misspelled root) parses to zero grants, and taking that as
+                // "remove every grant" strips the object's owner FULL_CONTROL
+                // and writes the empty list into its meta. Same rule the bucket
+                // path applies.
+                if !body_str.contains("<AccessControlPolicy") {
+                    return Err(AwsServiceError::aws_error(
+                        StatusCode::BAD_REQUEST,
+                        "MalformedXML",
+                        "The XML you provided was not well-formed or did not validate against our published schema",
+                    ));
                 }
+                parse_acl_xml(body_str)?
             }
         };
 
@@ -98,12 +119,37 @@ impl S3Service {
                 ));
             }
         }
-        obj.acl_grants = proposed_grants;
-
-        let meta = object_meta_snapshot(obj);
+        // Persist before touching memory, as put_bucket_acl does: assigning
+        // first would leave the in-memory ACL ahead of the stored one, and a
+        // restart would silently revert what the caller was told had failed.
+        //
+        // The snapshot has to carry the PROPOSED grants, not the object's
+        // current ones -- `object_meta_snapshot` copies `acl_grants`, so
+        // snapshotting before the assignment would persist the old ACL and
+        // leave the new one memory-only.
+        let mut meta = object_meta_snapshot(obj);
+        meta.acl_grants = proposed_grants
+            .iter()
+            .map(fakecloud_persistence::AclGrantSnapshot::from)
+            .collect();
         self.store
             .put_object_meta(bucket, key, meta.version_id.as_deref(), &meta)
             .map_err(super::persistence_error)?;
+        obj.acl_grants = proposed_grants.clone();
+        // A versioned bucket keeps a second copy of this version in
+        // `object_versions`, and `resolve_object` reads THAT one for a
+        // versionId request. Leaving it stale made GetObjectAcl answer
+        // differently depending on whether a versionId was passed, and a later
+        // delete re-derived the current object from the stale copy, silently
+        // reverting the change that had already been persisted.
+        let version_id = obj.version_id.clone();
+        if let Some(versions) = b.object_versions.get_mut(key) {
+            for v in versions.iter_mut() {
+                if v.version_id == version_id {
+                    v.acl_grants = proposed_grants.clone();
+                }
+            }
+        }
         Ok(AwsResponse {
             status: StatusCode::OK,
             content_type: "application/xml".to_string(),