From fe5690ca70ba24a81289f762429ced6333846b70 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Tue, 12 May 2026 21:34:52 +0800 Subject: [PATCH] fix(sftp): preserve open attrs metadata (#2929) Co-authored-by: houseme --- crates/protocols/src/common/dummy_storage.rs | 49 ++++- crates/protocols/src/sftp/attrs.rs | 89 +++++++- crates/protocols/src/sftp/driver.rs | 16 +- crates/protocols/src/sftp/read.rs | 7 +- crates/protocols/src/sftp/state.rs | 4 + crates/protocols/src/sftp/test_support.rs | 1 + crates/protocols/src/sftp/write.rs | 208 ++++++++++++++++--- 7 files changed, 330 insertions(+), 44 deletions(-) diff --git a/crates/protocols/src/common/dummy_storage.rs b/crates/protocols/src/common/dummy_storage.rs index 4917021a1..aa21bd2b1 100644 --- a/crates/protocols/src/common/dummy_storage.rs +++ b/crates/protocols/src/common/dummy_storage.rs @@ -39,7 +39,7 @@ use s3s::dto::{ ListObjectsV2Input, ListObjectsV2Output, PutObjectInput, PutObjectOutput, StreamingBlob, Timestamp, UploadPartCopyInput, UploadPartCopyOutput, UploadPartInput, UploadPartOutput, }; -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex}; use thiserror::Error; use tokio::sync::Notify; @@ -88,6 +88,22 @@ pub struct UploadPartCall { pub content_length: Option, } +/// Recorded invocation of put_object. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PutObjectCall { + pub bucket: String, + pub key: String, + pub metadata: Option>, +} + +/// Recorded invocation of create_multipart_upload. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CreateMultipartCall { + pub bucket: String, + pub key: String, + pub metadata: Option>, +} + /// Recorded invocation of complete_multipart_upload. #[derive(Debug, Clone, PartialEq, Eq)] pub struct CompleteCall { @@ -126,6 +142,8 @@ struct Inner { // Observation logs. abort_multipart_calls: Vec, + put_object_calls: Vec, + create_multipart_calls: Vec, upload_part_calls: Vec, complete_multipart_calls: Vec, head_object_calls: Vec, @@ -173,6 +191,8 @@ impl Inner { abort_multipart_upload: VecDeque::new(), upload_part_copy: VecDeque::new(), abort_multipart_calls: Vec::new(), + put_object_calls: Vec::new(), + create_multipart_calls: Vec::new(), upload_part_calls: Vec::new(), complete_multipart_calls: Vec::new(), head_object_calls: Vec::new(), @@ -422,6 +442,16 @@ impl DummyBackend { self.inner.lock().expect("lock").abort_multipart_calls.clone() } + /// Snapshot the put_object call log. + pub fn put_object_calls(&self) -> Vec { + self.inner.lock().expect("lock").put_object_calls.clone() + } + + /// Snapshot the create_multipart_upload call log. + pub fn create_multipart_calls(&self) -> Vec { + self.inner.lock().expect("lock").create_multipart_calls.clone() + } + /// Snapshot the upload_part call log. pub fn upload_part_calls(&self) -> Vec { self.inner.lock().expect("lock").upload_part_calls.clone() @@ -471,12 +501,17 @@ impl StorageBackend for DummyBackend { } } - async fn put_object(&self, _input: PutObjectInput, _ak: &str, _sk: &str) -> Result { + async fn put_object(&self, input: PutObjectInput, _ak: &str, _sk: &str) -> Result { // Decide control flow while holding the lock. Release before // awaiting so the stall path does not hold the Mutex across // an await point. let (stall, entered, popped) = { let mut inner = self.inner.lock().expect("lock"); + inner.put_object_calls.push(PutObjectCall { + bucket: input.bucket.to_string(), + key: input.key.to_string(), + metadata: input.metadata.clone(), + }); let stall = inner.stall_put_object; let entered = inner.put_object_entered.clone(); let popped = if stall { None } else { inner.put_object.pop_front() }; @@ -582,10 +617,18 @@ impl StorageBackend for DummyBackend { async fn create_multipart_upload( &self, - _input: CreateMultipartUploadInput, + input: CreateMultipartUploadInput, _ak: &str, _sk: &str, ) -> Result { + { + let mut inner = self.inner.lock().expect("lock"); + inner.create_multipart_calls.push(CreateMultipartCall { + bucket: input.bucket.to_string(), + key: input.key.to_string(), + metadata: input.metadata.clone(), + }); + } match self.inner.lock().expect("lock").create_multipart_upload.pop_front() { Some(r) => r, None => Err(DummyError::Unconfigured("create_multipart_upload")), diff --git a/crates/protocols/src/sftp/attrs.rs b/crates/protocols/src/sftp/attrs.rs index 4e7c27535..bfc3a830b 100644 --- a/crates/protocols/src/sftp/attrs.rs +++ b/crates/protocols/src/sftp/attrs.rs @@ -24,6 +24,12 @@ use crate::common::client::s3::StorageBackend; use crate::common::gateway::S3Action; use russh_sftp::protocol::{File, FileAttributes, StatusCode}; use s3s::dto::ListObjectsV2Input; +use std::collections::HashMap; + +const SFTP_META_MTIME: &str = "mtime"; +const SFTP_META_MODE: &str = "mode"; +const SFTP_META_UID: &str = "uid"; +const SFTP_META_GID: &str = "gid"; /// Build the SFTP FileAttributes struct returned by STAT, LSTAT, and /// FSTAT. Callers are responsible for any clamping or conversion of the @@ -43,6 +49,44 @@ pub(super) fn s3_attrs_to_sftp(size: u64, mtime: Option, is_dir: bool) -> F } } +fn parse_u32_metadata(metadata: &HashMap, key: &str) -> Option { + metadata.get(key).and_then(|value| value.parse::().ok()) +} + +pub(super) fn sftp_attrs_to_user_metadata(attrs: &FileAttributes) -> Option> { + let mut metadata = HashMap::new(); + if let Some(mtime) = attrs.mtime { + metadata.insert(SFTP_META_MTIME.to_string(), mtime.to_string()); + } + if let Some(mode) = attrs.permissions { + metadata.insert(SFTP_META_MODE.to_string(), mode.to_string()); + } + if let Some(uid) = attrs.uid { + metadata.insert(SFTP_META_UID.to_string(), uid.to_string()); + } + if let Some(gid) = attrs.gid { + metadata.insert(SFTP_META_GID.to_string(), gid.to_string()); + } + + if metadata.is_empty() { None } else { Some(metadata) } +} + +pub(super) fn apply_user_metadata_to_sftp_attrs(attrs: &mut FileAttributes, metadata: &HashMap) { + if let Some(mtime) = parse_u32_metadata(metadata, SFTP_META_MTIME) { + attrs.mtime = Some(mtime); + attrs.atime = Some(mtime); + } + if let Some(mode) = parse_u32_metadata(metadata, SFTP_META_MODE) { + attrs.permissions = Some(mode); + } + if let Some(uid) = parse_u32_metadata(metadata, SFTP_META_UID) { + attrs.uid = Some(uid); + } + if let Some(gid) = parse_u32_metadata(metadata, SFTP_META_GID) { + attrs.gid = Some(gid); + } +} + /// Convert an s3s Timestamp into the u32 seconds field SFTPv3 expects. /// Pre-1970 values clamp to 0. Post-2106 values clamp to u32::MAX. The /// clamps prevent the i64-to-u32 cast from wrapping. @@ -120,7 +164,11 @@ impl SftpDriver { Ok(out) => { let size = out.content_length.unwrap_or(0).max(0) as u64; let mtime = timestamp_to_mtime(out.last_modified); - Ok(s3_attrs_to_sftp(size, mtime, false)) + let mut attrs = s3_attrs_to_sftp(size, mtime, false); + if let Some(metadata) = out.metadata { + apply_user_metadata_to_sftp_attrs(&mut attrs, &metadata); + } + Ok(attrs) } Err(e) if is_not_found_error(&e) => { // No object at this key. Check whether it is a @@ -188,6 +236,45 @@ mod tests { assert!(attrs.is_regular()); } + #[test] + fn sftp_attrs_to_user_metadata_maps_only_present_open_attrs() { + let attrs = FileAttributes { + size: None, + uid: Some(1000), + gid: Some(1001), + user: None, + group: None, + permissions: Some(0o100640), + atime: None, + mtime: Some(1_777_992_333), + }; + + let metadata = sftp_attrs_to_user_metadata(&attrs).expect("present attrs produce metadata"); + assert_eq!(metadata.get("mtime").map(String::as_str), Some("1777992333")); + assert_eq!(metadata.get("mode").map(String::as_str), Some("33184")); + assert_eq!(metadata.get("uid").map(String::as_str), Some("1000")); + assert_eq!(metadata.get("gid").map(String::as_str), Some("1001")); + assert!(!metadata.contains_key("size"), "object size is data-path state, not OPEN metadata"); + } + + #[test] + fn apply_user_metadata_to_sftp_attrs_overrides_defaults() { + let mut attrs = s3_attrs_to_sftp(42, Some(10), false); + let metadata = HashMap::from([ + ("mtime".to_string(), "1777992348".to_string()), + ("mode".to_string(), "33152".to_string()), + ("uid".to_string(), "501".to_string()), + ("gid".to_string(), "20".to_string()), + ]); + + apply_user_metadata_to_sftp_attrs(&mut attrs, &metadata); + assert_eq!(attrs.mtime, Some(1_777_992_348)); + assert_eq!(attrs.atime, Some(1_777_992_348)); + assert_eq!(attrs.permissions, Some(0o100600)); + assert_eq!(attrs.uid, Some(501)); + assert_eq!(attrs.gid, Some(20)); + } + #[test] fn generate_longname_prefixes_d_for_directory() { let attrs = s3_attrs_to_sftp(0, Some(0), true); diff --git a/crates/protocols/src/sftp/driver.rs b/crates/protocols/src/sftp/driver.rs index 90acff2e8..fae030b56 100644 --- a/crates/protocols/src/sftp/driver.rs +++ b/crates/protocols/src/sftp/driver.rs @@ -473,14 +473,8 @@ impl russh_sftp::server::Handler for /// in-place edit cycle (download, modify, upload). Clients that need /// that pattern (rare for SFTP) get a clear protocol error rather /// than a data loss path. - #[tracing::instrument(level = "info", skip(self, _attrs), fields(id, path = %sanitise_control_bytes(&filename), pflags = ?pflags), err(Debug))] - async fn open( - &mut self, - id: u32, - filename: String, - pflags: OpenFlags, - _attrs: FileAttributes, - ) -> Result { + #[tracing::instrument(level = "info", skip(self, attrs), fields(id, path = %sanitise_control_bytes(&filename), pflags = ?pflags), err(Debug))] + async fn open(&mut self, id: u32, filename: String, pflags: OpenFlags, attrs: FileAttributes) -> Result { if pflags.contains(OpenFlags::APPEND) { return Err(SftpError::code(StatusCode::OpUnsupported)); } @@ -502,7 +496,7 @@ impl russh_sftp::server::Handler for return Err(SftpError::code(StatusCode::OpUnsupported)); } if is_write { - return self.open_write(id, &filename, pflags).await; + return self.open_write(id, &filename, pflags, attrs).await; } if is_read { return self.open_read(id, &filename).await; @@ -564,6 +558,7 @@ impl russh_sftp::server::Handler for bucket, key, attrs, + open_attrs, phase, }) = removed else { @@ -574,7 +569,7 @@ impl russh_sftp::server::Handler for WritePhase::Buffering { part_buffer } => { // Small-file path. No multipart state exists so nothing // to abort on failure. - self.commit_write(&bucket, &key, part_buffer).await?; + self.commit_write(&bucket, &key, &open_attrs, part_buffer).await?; } WritePhase::Streaming { upload_id, @@ -686,6 +681,7 @@ impl russh_sftp::server::Handler for bucket, key, attrs, + open_attrs: _, phase: WritePhase::Streaming { upload_id, diff --git a/crates/protocols/src/sftp/read.rs b/crates/protocols/src/sftp/read.rs index f8136078e..00f5e4af5 100644 --- a/crates/protocols/src/sftp/read.rs +++ b/crates/protocols/src/sftp/read.rs @@ -15,7 +15,7 @@ //! Read-side operation handlers: open_read and the body of the read() //! Handler trait method. -use super::attrs::{s3_attrs_to_sftp, timestamp_to_mtime}; +use super::attrs::{apply_user_metadata_to_sftp_attrs, s3_attrs_to_sftp, timestamp_to_mtime}; use super::constants::limits::{MAX_READ_LEN, READ_CACHE_DISABLED}; use super::driver::SftpDriver; use super::errors::{SftpError, s3_error_to_sftp}; @@ -54,7 +54,10 @@ impl SftpDriver { .await?; let size = head.content_length.unwrap_or(0).max(0) as u64; let mtime = timestamp_to_mtime(head.last_modified); - let attrs = s3_attrs_to_sftp(size, mtime, false); + let mut attrs = s3_attrs_to_sftp(size, mtime, false); + if let Some(metadata) = head.metadata { + apply_user_metadata_to_sftp_attrs(&mut attrs, &metadata); + } let read_cache = self.new_read_cache(); let handle = self.allocate_handle(HandleState::File { diff --git a/crates/protocols/src/sftp/state.rs b/crates/protocols/src/sftp/state.rs index 54fbf4b0c..b3b915a79 100644 --- a/crates/protocols/src/sftp/state.rs +++ b/crates/protocols/src/sftp/state.rs @@ -59,6 +59,10 @@ pub(super) enum HandleState { /// The size field tracks the running total of bytes received so /// a client polling FSTAT during a transfer sees the progress. attrs: FileAttributes, + /// Raw attributes supplied on OPEN. Only fields explicitly set + /// by the client are copied into S3 user metadata at object + /// creation time. + open_attrs: FileAttributes, /// Multipart upload lifecycle state. See WritePhase. phase: WritePhase, }, diff --git a/crates/protocols/src/sftp/test_support.rs b/crates/protocols/src/sftp/test_support.rs index 37bf7cbb3..65f0176ae 100644 --- a/crates/protocols/src/sftp/test_support.rs +++ b/crates/protocols/src/sftp/test_support.rs @@ -71,6 +71,7 @@ pub(super) fn write_handle(bucket: &str, key: &str, phase: WritePhase) -> Handle bucket: bucket.to_string(), key: key.to_string(), attrs: FileAttributes::default(), + open_attrs: FileAttributes::default(), phase, } } diff --git a/crates/protocols/src/sftp/write.rs b/crates/protocols/src/sftp/write.rs index 134269409..fcf8b78ac 100644 --- a/crates/protocols/src/sftp/write.rs +++ b/crates/protocols/src/sftp/write.rs @@ -19,7 +19,7 @@ //! (build_write_tombstone, should_abort_on_drop) that the Drop impl //! in driver.rs consumes. -use super::attrs::s3_attrs_to_sftp; +use super::attrs::{s3_attrs_to_sftp, sftp_attrs_to_user_metadata}; use super::constants::limits::{ COMMIT_WRITE_BACKOFF_MS, COMMIT_WRITE_MAX_RETRIES, S3_MAX_MULTIPART_PARTS, S3_MAX_PART_SIZE, S3_MIN_PART_SIZE, }; @@ -173,6 +173,7 @@ pub(super) fn build_write_tombstone( bucket: bucket.to_string(), key: key.to_string(), attrs: attrs.clone(), + open_attrs: FileAttributes::default(), phase: WritePhase::Failed { upload_id, abort_authorized, @@ -232,7 +233,13 @@ impl SftpDriver { /// between this HEAD and the eventual CLOSE. The SFTPv3 draft /// does not guarantee atomicity here and S3 has no native CAS /// primitive, so the race is accepted. - pub(super) async fn open_write(&mut self, id: u32, filename: &str, pflags: OpenFlags) -> Result { + pub(super) async fn open_write( + &mut self, + id: u32, + filename: &str, + pflags: OpenFlags, + open_attrs: FileAttributes, + ) -> Result { self.enforce_server_readonly()?; let (bucket, key) = parse_s3_path(filename)?; @@ -275,11 +282,25 @@ impl SftpDriver { } } - let attrs = s3_attrs_to_sftp(0, None, false); + let mut attrs = s3_attrs_to_sftp(0, None, false); + if open_attrs.mtime.is_some() { + attrs.mtime = open_attrs.mtime; + attrs.atime = open_attrs.mtime; + } + if open_attrs.permissions.is_some() { + attrs.permissions = open_attrs.permissions; + } + if open_attrs.uid.is_some() { + attrs.uid = open_attrs.uid; + } + if open_attrs.gid.is_some() { + attrs.gid = open_attrs.gid; + } let handle = self.allocate_handle(HandleState::Write { bucket, key: object_key, attrs, + open_attrs, phase: WritePhase::Buffering { part_buffer: Vec::new() }, })?; Ok(Handle { id, handle }) @@ -302,7 +323,13 @@ impl SftpDriver { /// run_backend deadline still propagates immediately as Failure /// rather than retrying, because a stuck backend is not a /// transient classification the retry set targets. - pub(super) async fn commit_write(&self, bucket: &str, key: &str, buffer: Vec) -> Result<(), SftpError> { + pub(super) async fn commit_write( + &self, + bucket: &str, + key: &str, + attrs: &FileAttributes, + buffer: Vec, + ) -> Result<(), SftpError> { let size = buffer.len() as i64; let body_bytes = Bytes::from(buffer); @@ -324,6 +351,7 @@ impl SftpDriver { .bucket(bucket.to_string()) .key(key.to_string()) .content_length(Some(size)) + .metadata(sftp_attrs_to_user_metadata(attrs)) .body(Some(streaming)) .build() .map_err(|e| s3_error_to_sftp("build_put_object", e))?; @@ -429,7 +457,12 @@ impl SftpDriver { /// passes an empty conditions map. Only unconditional Allow/Deny is /// honoured. This is a gateway-wide limitation, not specific to /// this cache. - pub(super) async fn start_multipart_upload(&self, bucket: &str, key: &str) -> Result { + pub(super) async fn start_multipart_upload( + &self, + bucket: &str, + key: &str, + attrs: &FileAttributes, + ) -> Result { self.authorize(&S3Action::CreateMultipartUpload, bucket, Some(key)).await?; // Probe AbortMultipartUpload authorisation immediately after @@ -450,6 +483,7 @@ impl SftpDriver { let input = CreateMultipartUploadInput::builder() .bucket(bucket.to_string()) .key(key.to_string()) + .metadata(sftp_attrs_to_user_metadata(attrs)) .build() .map_err(|e| s3_error_to_sftp("build_create_multipart_upload", e))?; @@ -553,6 +587,7 @@ impl SftpDriver { bucket, key, attrs, + open_attrs, phase, } = state else { @@ -575,7 +610,7 @@ impl SftpDriver { while write_dispatch_has_full_part(phase, part_size) { if matches!(phase, WritePhase::Buffering { .. }) { - self.write_dispatch_begin_streaming(handle, phase, &bucket_owned, &key_owned) + self.write_dispatch_begin_streaming(handle, phase, &bucket_owned, &key_owned, open_attrs) .await?; } self.write_dispatch_flush_one_part(phase, &bucket_owned, &key_owned, part_size) @@ -602,6 +637,7 @@ impl SftpDriver { phase: &mut WritePhase, bucket: &str, key: &str, + attrs: &FileAttributes, ) -> Result<(), SftpError> { if !matches!(phase, WritePhase::Buffering { .. }) { return Ok(()); @@ -611,7 +647,7 @@ impl SftpDriver { // authorize_operation(AbortMultipartUpload) result. The pair is // stored on the Streaming variant so Drop (which cannot await) // has a pre-decided policy answer. - let mp = self.start_multipart_upload(bucket, key).await?; + let mp = self.start_multipart_upload(bucket, key, attrs).await?; let existing_buffer = match phase { WritePhase::Buffering { part_buffer } => std::mem::take(part_buffer), _ => { @@ -635,7 +671,7 @@ impl SftpDriver { // synchronous window between start_multipart_upload returning // Ok and this insert contains no await, so cancellation cannot // fire in it. - let tombstone = build_write_tombstone(bucket, key, &FileAttributes::default(), mp.upload_id, mp.abort_authorized); + let tombstone = build_write_tombstone(bucket, key, attrs, mp.upload_id, mp.abort_authorized); self.handles.insert(handle.to_string(), tombstone); Ok(()) } @@ -915,7 +951,9 @@ impl SftpDriver { // abort_authorized flag is carried on the MultipartUpload so // close_abort_or_skip can honour a Deny-Abort policy without a // second IAM probe per error path. - let mp = self.start_multipart_upload(dst_bucket, dst_key).await?; + let mp = self + .start_multipart_upload(dst_bucket, dst_key, &FileAttributes::default()) + .await?; let result: Result, SftpError> = async { let mut uploaded_parts = Vec::new(); @@ -1550,9 +1588,12 @@ mod tests { part_buffer: vec![1, 2, 3, 4], }; - with_test_auth_override(|_, _, _| true, driver.write_dispatch_begin_streaming(&handle_id, &mut phase, "b", "k")) - .await - .expect("begin_streaming must succeed on queued Create Ok"); + with_test_auth_override( + |_, _, _| true, + driver.write_dispatch_begin_streaming(&handle_id, &mut phase, "b", "k", &FileAttributes::default()), + ) + .await + .expect("begin_streaming must succeed on queued Create Ok"); // Tombstone invariant: the driver.handles entry under handle_id // is now a Failed-variant HandleState carrying the upload_id. @@ -1589,13 +1630,57 @@ mod tests { backend.queue_create_multipart_upload_ok("UP-ALLOW"); let driver = build_driver(backend, TEST_PART_SIZE); - let mp = with_test_auth_override(|_, _, _| true, driver.start_multipart_upload("b", "k")) + let mp = with_test_auth_override(|_, _, _| true, driver.start_multipart_upload("b", "k", &FileAttributes::default())) .await .expect("start_multipart_upload must succeed on Allow"); assert_eq!(mp.upload_id, "UP-ALLOW"); assert!(mp.abort_authorized, "Allow on AbortMultipartUpload probe must cache as true"); } + #[tokio::test] + async fn start_multipart_upload_preserves_open_attrs_as_metadata() { + let backend = Arc::new(DummyBackend::new()); + backend.queue_create_multipart_upload_ok("UP-ATTRS"); + let driver = build_driver(backend.clone(), TEST_PART_SIZE); + let attrs = FileAttributes { + size: None, + uid: Some(1000), + gid: Some(1001), + user: None, + group: None, + permissions: Some(0o100640), + atime: None, + mtime: Some(1_777_992_333), + }; + + with_test_auth_override(|_, _, _| true, driver.start_multipart_upload("b", "k", &attrs)) + .await + .expect("start_multipart_upload must succeed"); + + let calls = backend.create_multipart_calls(); + assert_eq!(calls.len(), 1); + let metadata = calls[0].metadata.as_ref().expect("OPEN attrs must become S3 user metadata"); + assert_eq!(metadata.get("mtime").map(String::as_str), Some("1777992333")); + assert_eq!(metadata.get("mode").map(String::as_str), Some("33184")); + assert_eq!(metadata.get("uid").map(String::as_str), Some("1000")); + assert_eq!(metadata.get("gid").map(String::as_str), Some("1001")); + } + + #[tokio::test] + async fn start_multipart_upload_omits_metadata_when_open_attrs_empty() { + let backend = Arc::new(DummyBackend::new()); + backend.queue_create_multipart_upload_ok("UP-NO-ATTRS"); + let driver = build_driver(backend.clone(), TEST_PART_SIZE); + + with_test_auth_override(|_, _, _| true, driver.start_multipart_upload("b", "k", &FileAttributes::default())) + .await + .expect("start_multipart_upload must succeed"); + + let calls = backend.create_multipart_calls(); + assert_eq!(calls.len(), 1); + assert!(calls[0].metadata.is_none(), "empty OPEN attrs must not write default metadata"); + } + #[tokio::test] async fn start_multipart_upload_caches_abort_authorized_false_on_deny_abort() { let backend = Arc::new(DummyBackend::new()); @@ -1606,7 +1691,7 @@ mod tests { // a WORM-shaped IAM policy a principal can meet in production. let mp = with_test_auth_override( |action, _bucket, _object| !matches!(action, S3Action::AbortMultipartUpload), - driver.start_multipart_upload("b", "k"), + driver.start_multipart_upload("b", "k", &FileAttributes::default()), ) .await .expect("Create Allow must succeed even when Abort is Deny"); @@ -1628,7 +1713,7 @@ mod tests { let err = with_test_auth_override( |action, _, _| !matches!(action, S3Action::CreateMultipartUpload), - driver.start_multipart_upload("b", "k"), + driver.start_multipart_upload("b", "k", &FileAttributes::default()), ) .await .expect_err("Deny on CreateMultipartUpload must fail fast"); @@ -1790,6 +1875,52 @@ mod tests { ); } + #[tokio::test] + async fn commit_write_preserves_open_attrs_as_metadata() { + let backend = Arc::new(DummyBackend::new()); + backend.queue_put_object_ok(); + let driver = build_driver(backend.clone(), TEST_PART_SIZE); + let attrs = FileAttributes { + size: None, + uid: Some(501), + gid: Some(20), + user: None, + group: None, + permissions: Some(0o100600), + atime: None, + mtime: Some(1_777_992_348), + }; + + driver + .commit_write("b", "k", &attrs, b"hello".to_vec()) + .await + .expect("commit_write must succeed"); + + let calls = backend.put_object_calls(); + assert_eq!(calls.len(), 1); + let metadata = calls[0].metadata.as_ref().expect("OPEN attrs must become S3 user metadata"); + assert_eq!(metadata.get("mtime").map(String::as_str), Some("1777992348")); + assert_eq!(metadata.get("mode").map(String::as_str), Some("33152")); + assert_eq!(metadata.get("uid").map(String::as_str), Some("501")); + assert_eq!(metadata.get("gid").map(String::as_str), Some("20")); + } + + #[tokio::test] + async fn commit_write_omits_metadata_when_open_attrs_empty() { + let backend = Arc::new(DummyBackend::new()); + backend.queue_put_object_ok(); + let driver = build_driver(backend.clone(), TEST_PART_SIZE); + + driver + .commit_write("b", "k", &FileAttributes::default(), b"hello".to_vec()) + .await + .expect("commit_write must succeed"); + + let calls = backend.put_object_calls(); + assert_eq!(calls.len(), 1); + assert!(calls[0].metadata.is_none(), "empty OPEN attrs must not write default metadata"); + } + // --- open_write strict-flag gate --- #[tokio::test] @@ -1799,9 +1930,12 @@ mod tests { let backend = Arc::new(DummyBackend::new()); let mut driver = build_driver(backend.clone(), TEST_PART_SIZE); - let err = with_test_auth_override(|_, _, _| true, driver.open_write(7, "/bucket/key", OpenFlags::WRITE)) - .await - .expect_err("WRITE without CREATE or TRUNCATE must be rejected"); + let err = with_test_auth_override( + |_, _, _| true, + driver.open_write(7, "/bucket/key", OpenFlags::WRITE, FileAttributes::default()), + ) + .await + .expect_err("WRITE without CREATE or TRUNCATE must be rejected"); assert!(matches!(err.0, StatusCode::OpUnsupported)); assert!( backend.head_object_calls().is_empty(), @@ -1818,10 +1952,12 @@ mod tests { let backend = Arc::new(DummyBackend::new()); let mut driver = build_driver(backend.clone(), TEST_PART_SIZE); - let err = - with_test_auth_override(|_, _, _| true, driver.open_write(7, "/bucket/key", OpenFlags::WRITE | OpenFlags::CREATE)) - .await - .expect_err("WRITE | CREATE without TRUNCATE must be rejected"); + let err = with_test_auth_override( + |_, _, _| true, + driver.open_write(7, "/bucket/key", OpenFlags::WRITE | OpenFlags::CREATE, FileAttributes::default()), + ) + .await + .expect_err("WRITE | CREATE without TRUNCATE must be rejected"); assert!(matches!(err.0, StatusCode::OpUnsupported)); assert!( backend.head_object_calls().is_empty(), @@ -1841,7 +1977,12 @@ mod tests { let handle = with_test_auth_override( |_, _, _| true, - driver.open_write(9, "/bucket/missing_key", OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE), + driver.open_write( + 9, + "/bucket/missing_key", + OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE, + FileAttributes::default(), + ), ) .await .expect("WRITE | CREATE | TRUNCATE on a missing file must allocate a handle"); @@ -1861,7 +2002,12 @@ mod tests { let handle = with_test_auth_override( |_, _, _| true, - driver.open_write(10, "/bucket/existing_key", OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE), + driver.open_write( + 10, + "/bucket/existing_key", + OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE, + FileAttributes::default(), + ), ) .await .expect("WRITE | CREATE | TRUNCATE on an existing file must allocate a handle"); @@ -1883,6 +2029,7 @@ mod tests { 7, "/bucket/key", OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::EXCLUDE | OpenFlags::WRITE, + FileAttributes::default(), ), ) .await @@ -1906,6 +2053,7 @@ mod tests { 9, "/bucket/key2", OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::EXCLUDE | OpenFlags::WRITE, + FileAttributes::default(), ), ) .await @@ -2028,7 +2176,11 @@ mod tests { // so the assertion failure mode distinguishes "driver did not // honour the deadline" (outer fires) from "deadline fired but // mapped to the wrong status" (Ok(Err) with non-Failure). - let outcome = tokio::time::timeout(Duration::from_secs(10), driver.commit_write("b", "k", b"hello".to_vec())).await; + let outcome = tokio::time::timeout( + Duration::from_secs(10), + driver.commit_write("b", "k", &FileAttributes::default(), b"hello".to_vec()), + ) + .await; let inner = outcome.expect("driver deadline must fire before the outer 10 s guard"); let err = inner.expect_err("stalling backend must surface as Err"); @@ -2079,7 +2231,7 @@ mod tests { let driver = build_driver(backend.clone(), TEST_PART_SIZE); driver - .commit_write("b", "k", b"hello".to_vec()) + .commit_write("b", "k", &FileAttributes::default(), b"hello".to_vec()) .await .expect("commit_write must succeed once a retry returns Ok"); assert_eq!( @@ -2102,7 +2254,7 @@ mod tests { let driver = build_driver(backend.clone(), TEST_PART_SIZE); let err = driver - .commit_write("b", "k", b"hello".to_vec()) + .commit_write("b", "k", &FileAttributes::default(), b"hello".to_vec()) .await .expect_err("commit_write must surface the final retryable error after the cap"); assert!(matches!(err.0, StatusCode::Failure)); @@ -2130,7 +2282,7 @@ mod tests { let driver = build_driver(backend.clone(), TEST_PART_SIZE); let err = driver - .commit_write("b", "k", b"hello".to_vec()) + .commit_write("b", "k", &FileAttributes::default(), b"hello".to_vec()) .await .expect_err("commit_write must surface AccessDenied without retrying"); assert!(matches!(err.0, StatusCode::PermissionDenied));