From 2231633ae11fc19f972ff355fa9bc754a6f9aeb1 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 2 Sep 2026 22:16:15 +0800 Subject: [PATCH] fix: preserve protocol service account claims (#7062) --- crates/protocols/src/common/client/s3.rs | 63 +-- crates/protocols/src/common/dummy_storage.rs | 58 +-- crates/protocols/src/common/gateway.rs | 62 ++- crates/protocols/src/common/session.rs | 5 + crates/protocols/src/ftps/driver.rs | 111 +---- crates/protocols/src/sftp/attrs.rs | 13 +- crates/protocols/src/sftp/dir.rs | 28 +- crates/protocols/src/sftp/driver.rs | 58 +-- crates/protocols/src/sftp/read.rs | 8 +- crates/protocols/src/sftp/write.rs | 36 +- crates/protocols/src/webdav/driver.rs | 385 ++++---------- crates/protocols/src/webdav/server.rs | 64 +-- rustfs/src/protocols/client.rs | 497 +++++++++---------- 13 files changed, 522 insertions(+), 866 deletions(-) diff --git a/crates/protocols/src/common/client/s3.rs b/crates/protocols/src/common/client/s3.rs index 25e4d3d4c..3af886479 100644 --- a/crates/protocols/src/common/client/s3.rs +++ b/crates/protocols/src/common/client/s3.rs @@ -13,6 +13,7 @@ // limitations under the License. use async_trait::async_trait; +use rustfs_credentials::Credentials; use s3s::dto::*; #[cfg(feature = "webdav")] @@ -27,49 +28,33 @@ pub trait StorageBackend: Send + Sync { &self, bucket: &str, key: &str, - access_key: &str, - secret_key: &str, + credentials: &Credentials, start_pos: Option, ) -> Result; async fn get_object_range( &self, bucket: &str, key: &str, - access_key: &str, - secret_key: &str, + credentials: &Credentials, start_pos: u64, length: u64, ) -> Result; /// Put object content with metadata - async fn put_object(&self, input: PutObjectInput, access_key: &str, secret_key: &str) - -> Result; + async fn put_object(&self, input: PutObjectInput, credentials: &Credentials) -> Result; /// Delete an object - async fn delete_object( - &self, - bucket: &str, - key: &str, - access_key: &str, - secret_key: &str, - ) -> Result; + async fn delete_object(&self, bucket: &str, key: &str, credentials: &Credentials) -> Result; /// Get object metadata without content - async fn head_object( - &self, - bucket: &str, - key: &str, - access_key: &str, - secret_key: &str, - ) -> Result; + async fn head_object(&self, bucket: &str, key: &str, credentials: &Credentials) -> Result; /// Check if bucket exists and get metadata - async fn head_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result; + async fn head_bucket(&self, bucket: &str, credentials: &Credentials) -> Result; /// List objects in a bucket with pagination async fn list_objects_v2( &self, input: ListObjectsV2Input, - access_key: &str, - secret_key: &str, + credentials: &Credentials, ) -> Result; /// List all buckets (requires authentication). - async fn list_buckets(&self, access_key: &str, secret_key: &str) -> Result; + async fn list_buckets(&self, credentials: &Credentials) -> Result; /// List buckets visible to the authenticated session. /// /// Backends that implement this must apply per-bucket authorization. The default denies the @@ -87,20 +72,15 @@ pub trait StorageBackend: Send + Sync { )) } /// Create a new bucket - async fn create_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result; + async fn create_bucket(&self, bucket: &str, credentials: &Credentials) -> Result; /// Delete a bucket (must be empty) - async fn delete_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result; + async fn delete_bucket(&self, bucket: &str, credentials: &Credentials) -> Result; /// Server-side copy of an object from one bucket+key to another. /// The input carries the full S3 surface (content type, metadata map, /// metadata directive, storage class, SSE config, conditional-copy /// headers) so protocol drivers can map client-supplied metadata /// onto the destination object. - async fn copy_object( - &self, - input: CopyObjectInput, - access_key: &str, - secret_key: &str, - ) -> Result; + async fn copy_object(&self, input: CopyObjectInput, credentials: &Credentials) -> Result; /// Initiate a multipart upload. Returns an upload_id that identifies /// the in-progress upload for subsequent UploadPart, CompleteMultipartUpload, /// and AbortMultipartUpload calls. The input carries the full S3 surface @@ -110,25 +90,18 @@ pub trait StorageBackend: Send + Sync { async fn create_multipart_upload( &self, input: CreateMultipartUploadInput, - access_key: &str, - secret_key: &str, + credentials: &Credentials, ) -> Result; /// Upload one part of a multipart upload. The part_number must be in /// the range 1 to the 10 000-part S3 limit. The returned ETag /// identifies the part in the subsequent CompleteMultipartUpload call. - async fn upload_part( - &self, - input: UploadPartInput, - access_key: &str, - secret_key: &str, - ) -> Result; + async fn upload_part(&self, input: UploadPartInput, credentials: &Credentials) -> Result; /// Assemble the parts listed in the input into the final object. /// The parts list must be sorted by part_number with no duplicates. async fn complete_multipart_upload( &self, input: CompleteMultipartUploadInput, - access_key: &str, - secret_key: &str, + credentials: &Credentials, ) -> Result; /// Abort an in-progress multipart upload. Releases any storage /// associated with the upload_id. Idempotent: calling abort on an @@ -138,8 +111,7 @@ pub trait StorageBackend: Send + Sync { async fn abort_multipart_upload( &self, input: AbortMultipartUploadInput, - access_key: &str, - secret_key: &str, + credentials: &Credentials, ) -> Result; /// Copy a byte range from an existing object into a part of an /// in-progress multipart upload. Used by rename for objects larger @@ -147,7 +119,6 @@ pub trait StorageBackend: Send + Sync { async fn upload_part_copy( &self, input: UploadPartCopyInput, - access_key: &str, - secret_key: &str, + credentials: &Credentials, ) -> Result; } diff --git a/crates/protocols/src/common/dummy_storage.rs b/crates/protocols/src/common/dummy_storage.rs index 4df113e78..e9cd612e2 100644 --- a/crates/protocols/src/common/dummy_storage.rs +++ b/crates/protocols/src/common/dummy_storage.rs @@ -35,6 +35,7 @@ use crate::common::session::SessionContext; use async_trait::async_trait; use bytes::Bytes; use futures_util::stream::{self, StreamExt}; +use rustfs_credentials::Credentials; use s3s::dto::{ AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput, CopyObjectInput, CopyObjectOutput, CopyPartResult, CreateBucketOutput, CreateMultipartUploadInput, @@ -605,8 +606,7 @@ impl StorageBackend for DummyBackend { &self, bucket: &str, key: &str, - _ak: &str, - _sk: &str, + _credentials: &Credentials, _start_pos: Option, ) -> Result { match self.inner.lock().expect("lock").get_object.pop_front() { @@ -619,8 +619,7 @@ impl StorageBackend for DummyBackend { &self, bucket: &str, key: &str, - _ak: &str, - _sk: &str, + _credentials: &Credentials, _start_pos: u64, _length: u64, ) -> Result { @@ -630,7 +629,7 @@ impl StorageBackend for DummyBackend { } } - async fn put_object(&self, input: PutObjectInput, _ak: &str, _sk: &str) -> Result { + async fn put_object(&self, input: PutObjectInput, _credentials: &Credentials) -> Result { // Decide control flow while holding the lock. Release before // awaiting so the stall path does not hold the Mutex across // an await point. @@ -659,7 +658,12 @@ impl StorageBackend for DummyBackend { } } - async fn delete_object(&self, bucket: &str, key: &str, _ak: &str, _sk: &str) -> Result { + async fn delete_object( + &self, + bucket: &str, + key: &str, + _credentials: &Credentials, + ) -> Result { let mut inner = self.inner.lock().expect("lock"); inner.delete_object_calls.push(DeleteObjectCall { bucket: bucket.to_string(), @@ -671,7 +675,7 @@ impl StorageBackend for DummyBackend { } } - async fn head_object(&self, bucket: &str, key: &str, _ak: &str, _sk: &str) -> Result { + async fn head_object(&self, bucket: &str, key: &str, _credentials: &Credentials) -> Result { { let mut inner = self.inner.lock().expect("lock"); inner.head_object_calls.push(HeadObjectCall { @@ -685,7 +689,7 @@ impl StorageBackend for DummyBackend { } } - async fn head_bucket(&self, bucket: &str, _ak: &str, _sk: &str) -> Result { + async fn head_bucket(&self, bucket: &str, _credentials: &Credentials) -> Result { match self.inner.lock().expect("lock").head_bucket.pop_front() { Some(r) => r, None => Err(DummyError::NoSuchBucket(bucket.to_string())), @@ -695,8 +699,7 @@ impl StorageBackend for DummyBackend { async fn list_objects_v2( &self, _input: ListObjectsV2Input, - _ak: &str, - _sk: &str, + _credentials: &Credentials, ) -> Result { // Decide control flow while holding the lock. Release before // awaiting so the stall path does not hold the Mutex across @@ -721,7 +724,7 @@ impl StorageBackend for DummyBackend { } } - async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result { + async fn list_buckets(&self, _credentials: &Credentials) -> Result { match self.inner.lock().expect("lock").list_buckets.pop_front() { Some(r) => r, None => Ok(ListBucketsOutput::default()), @@ -744,14 +747,14 @@ impl StorageBackend for DummyBackend { .unwrap_or_else(|| Ok(ListBucketsOutput::default())) } - async fn create_bucket(&self, _bucket: &str, _ak: &str, _sk: &str) -> Result { + async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { match self.inner.lock().expect("lock").create_bucket.pop_front() { Some(r) => r, None => Err(DummyError::Unconfigured("create_bucket")), } } - async fn delete_bucket(&self, bucket: &str, _ak: &str, _sk: &str) -> Result { + async fn delete_bucket(&self, bucket: &str, _credentials: &Credentials) -> Result { let mut inner = self.inner.lock().expect("lock"); inner.delete_bucket_calls.push(bucket.to_string()); match inner.delete_bucket.pop_front() { @@ -760,7 +763,7 @@ impl StorageBackend for DummyBackend { } } - async fn copy_object(&self, _input: CopyObjectInput, _ak: &str, _sk: &str) -> Result { + async fn copy_object(&self, _input: CopyObjectInput, _credentials: &Credentials) -> Result { match self.inner.lock().expect("lock").copy_object.pop_front() { Some(r) => r, None => Err(DummyError::Unconfigured("copy_object")), @@ -770,8 +773,7 @@ impl StorageBackend for DummyBackend { async fn create_multipart_upload( &self, input: CreateMultipartUploadInput, - _ak: &str, - _sk: &str, + _credentials: &Credentials, ) -> Result { { let mut inner = self.inner.lock().expect("lock"); @@ -787,7 +789,7 @@ impl StorageBackend for DummyBackend { } } - async fn upload_part(&self, input: UploadPartInput, _ak: &str, _sk: &str) -> Result { + async fn upload_part(&self, input: UploadPartInput, _credentials: &Credentials) -> Result { // Record the call and decide the control flow while holding the // lock. Release the lock before awaiting so the stall path does // not hold the Mutex across an await point. @@ -821,8 +823,7 @@ impl StorageBackend for DummyBackend { async fn complete_multipart_upload( &self, input: CompleteMultipartUploadInput, - _ak: &str, - _sk: &str, + _credentials: &Credentials, ) -> Result { let part_count = input .multipart_upload @@ -847,8 +848,7 @@ impl StorageBackend for DummyBackend { async fn abort_multipart_upload( &self, input: AbortMultipartUploadInput, - _ak: &str, - _sk: &str, + _credentials: &Credentials, ) -> Result { { let mut inner = self.inner.lock().expect("lock"); @@ -867,8 +867,7 @@ impl StorageBackend for DummyBackend { async fn upload_part_copy( &self, _input: UploadPartCopyInput, - _ak: &str, - _sk: &str, + _credentials: &Credentials, ) -> Result { match self.inner.lock().expect("lock").upload_part_copy.pop_front() { Some(r) => r, @@ -884,7 +883,8 @@ mod tests { #[tokio::test] async fn dummy_backend_reports_not_found_by_default() { let backend = DummyBackend::new(); - let result = backend.head_object("b", "k", "ak", "sk").await; + let credentials = Credentials::default(); + let result = backend.head_object("b", "k", &credentials).await; let Err(err) = result else { panic!("default head_object must return an error"); }; @@ -897,21 +897,23 @@ mod tests { #[tokio::test] async fn dummy_backend_returns_queued_head_object_response() { let backend = DummyBackend::new(); + let credentials = Credentials::default(); backend.queue_head_object_ok(42, None); - let out = backend.head_object("b", "k", "ak", "sk").await.expect("queued Ok"); + let out = backend.head_object("b", "k", &credentials).await.expect("queued Ok"); assert_eq!(out.content_length, Some(42)); } #[tokio::test] async fn dummy_backend_logs_abort_multipart_calls() { let backend = Arc::new(DummyBackend::new()); + let credentials = Credentials::default(); let input = AbortMultipartUploadInput::builder() .bucket("b".to_string()) .key("k".to_string()) .upload_id("UP-1".to_string()) .build() .expect("build"); - backend.abort_multipart_upload(input, "ak", "sk").await.expect("Ok"); + backend.abort_multipart_upload(input, &credentials).await.expect("Ok"); let calls = backend.abort_multipart_calls(); assert_eq!(calls.len(), 1); assert_eq!(calls[0].upload_id, "UP-1"); @@ -920,6 +922,7 @@ mod tests { #[tokio::test] async fn dummy_backend_unconfigured_errors_loudly() { let backend = DummyBackend::new(); + let credentials = Credentials::default(); let err = backend .create_multipart_upload( CreateMultipartUploadInput::builder() @@ -927,8 +930,7 @@ mod tests { .key("k".to_string()) .build() .expect("build"), - "ak", - "sk", + &credentials, ) .await .expect_err("default create_multipart_upload must error"); diff --git a/crates/protocols/src/common/gateway.rs b/crates/protocols/src/common/gateway.rs index 28e2db992..30feb1b46 100644 --- a/crates/protocols/src/common/gateway.rs +++ b/crates/protocols/src/common/gateway.rs @@ -288,12 +288,7 @@ pub async fn is_authorized( } }; - // Create policy arguments - let mut claims = HashMap::new(); - claims.insert( - "principal".to_string(), - serde_json::Value::String(session_context.principal.access_key().to_string()), - ); + let claims = policy_claims_for_session(session_context); let policy_action: rustfs_policy::policy::action::Action = action.clone().into(); @@ -315,6 +310,21 @@ pub async fn is_authorized( Ok(iam_sys.is_allowed(&args).await) } +fn policy_claims_for_session(session_context: &SessionContext) -> HashMap { + let mut claims = session_context + .principal + .user_identity + .credentials + .claims + .clone() + .unwrap_or_default(); + claims.insert( + "principal".to_string(), + serde_json::Value::String(session_context.principal.access_key().to_string()), + ); + claims +} + /// Authorize an operation and return an error if not authorized. /// AccessDenied covers both the protocol-not-supported case and the /// policy-denies case. IamUnavailable propagates from is_authorized @@ -457,7 +467,9 @@ pub use test_auth_override::{with_test_auth_override, with_test_iam_unavailable} mod tests { use super::*; use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext}; + use rustfs_credentials::{IAM_POLICY_CLAIM_NAME_SA, INHERITED_POLICY_TYPE}; use rustfs_policy::auth::UserIdentity; + use serde_json::Value; use std::net::{IpAddr, Ipv4Addr}; use std::sync::Arc; @@ -466,6 +478,44 @@ mod tests { SessionContext::new(principal, Protocol::Sftp, IpAddr::V4(Ipv4Addr::LOCALHOST)) } + fn session_with_claims(access_key: &str, claims: HashMap) -> SessionContext { + let identity = UserIdentity::new(rustfs_credentials::Credentials { + access_key: access_key.to_string(), + secret_key: "secret".to_string(), + claims: Some(claims), + ..Default::default() + }); + let principal = ProtocolPrincipal::new(Arc::new(identity)); + SessionContext::new(principal, Protocol::WebDav, IpAddr::V4(Ipv4Addr::LOCALHOST)) + } + + #[test] + fn policy_claims_preserve_authenticated_service_account_claims() { + let parent = "parent-user"; + let mut stored_claims = HashMap::new(); + stored_claims.insert("parent".to_string(), Value::String(parent.to_string())); + stored_claims.insert(IAM_POLICY_CLAIM_NAME_SA.to_string(), Value::String(INHERITED_POLICY_TYPE.to_string())); + let session = session_with_claims("service-account", stored_claims); + + let claims = policy_claims_for_session(&session); + + assert_eq!(claims.get("parent").and_then(Value::as_str), Some(parent)); + assert_eq!(claims.get(IAM_POLICY_CLAIM_NAME_SA).and_then(Value::as_str), Some(INHERITED_POLICY_TYPE)); + assert_eq!(claims.get("principal").and_then(Value::as_str), Some("service-account")); + } + + #[test] + fn policy_claims_overwrite_untrusted_principal_claim() { + let session = session_with_claims( + "authenticated-service-account", + HashMap::from([("principal".to_string(), Value::String("forged-principal".to_string()))]), + ); + + let claims = policy_claims_for_session(&session); + + assert_eq!(claims.get("principal").and_then(Value::as_str), Some("authenticated-service-account")); + } + #[tokio::test] async fn with_test_auth_override_allow_returns_ok() { let session = test_session(); diff --git a/crates/protocols/src/common/session.rs b/crates/protocols/src/common/session.rs index 8c684bed2..d1acba1ce 100644 --- a/crates/protocols/src/common/session.rs +++ b/crates/protocols/src/common/session.rs @@ -84,6 +84,11 @@ impl SessionContext { pub fn access_key(&self) -> &str { self.principal.access_key() } + + /// Get the authenticated credentials for this session. + pub fn credentials(&self) -> &Credentials { + &self.principal.user_identity.credentials + } } /// Build a SessionContext suitable for driver-level unit tests. The diff --git a/crates/protocols/src/ftps/driver.rs b/crates/protocols/src/ftps/driver.rs index 4942db393..5c7431d39 100644 --- a/crates/protocols/src/ftps/driver.rs +++ b/crates/protocols/src/ftps/driver.rs @@ -129,14 +129,7 @@ where } let mut list_result = Vec::new(); - match self - .storage - .list_buckets( - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.list_buckets(session_context.credentials()).await { Ok(output) => { if let Some(buckets) = output.buckets { for bucket in buckets { @@ -190,15 +183,7 @@ where Error::new(ErrorKind::PermanentFileNotAvailable, format!("Failed to build ListObjectsV2Input: {}", e)) })?; - if let Ok(output) = self - .storage - .list_objects_v2( - list_input, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + if let Ok(output) = self.storage.list_objects_v2(list_input, session_context.credentials()).await { // Delete all objects in this page if let Some(objects) = output.contents { for obj in objects { @@ -209,12 +194,7 @@ where let _ = self .storage - .delete_object( - bucket, - &obj_key, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) + .delete_object(bucket, &obj_key, session_context.credentials()) .await; } } @@ -231,15 +211,7 @@ where } // Then delete the bucket - match self - .storage - .delete_bucket( - bucket, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.delete_bucket(bucket, session_context.credentials()).await { Ok(_) => Ok(()), Err(e) if e.to_string().contains("NoSuchBucket") => Ok(()), Err(e) => { @@ -277,16 +249,7 @@ where .await .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?; - match self - .storage - .head_object( - &bucket, - &key, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.head_object(&bucket, &key, session_context.credentials()).await { Ok(output) => { let size = output.content_length.unwrap_or(0) as u64; let modified = output.last_modified.map(|dt| { @@ -323,15 +286,7 @@ where .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?; let bucket_clone = bucket.clone(); - match self - .storage - .head_bucket( - &bucket, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.head_bucket(&bucket, session_context.credentials()).await { Ok(_) => Ok(FtpsMetadata { size: 0, modified: Some(std::time::SystemTime::now()), @@ -390,15 +345,7 @@ where Error::new(ErrorKind::PermanentFileNotAvailable, format!("Failed to build ListObjectsV2Input: {}", e)) })?; - match self - .storage - .list_objects_v2( - list_input, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.list_objects_v2(list_input, session_context.credentials()).await { Ok(output) => { let mut fileinfos = Vec::new(); @@ -515,8 +462,7 @@ where .get_object( &bucket, &key, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, + session_context.credentials(), Some(start_pos), // Pass start_pos for range request ) .await @@ -624,15 +570,7 @@ where .build() .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Failed to build PutObjectInput"))?; - match self - .storage - .put_object( - put_input, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.put_object(put_input, session_context.credentials()).await { Ok(_output) => { Ok(file_size as u64) // Return the size of the uploaded object } @@ -681,16 +619,7 @@ where .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?; // Delete file - match self - .storage - .delete_object( - &bucket, - &key, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.delete_object(&bucket, &key, session_context.credentials()).await { Ok(_) => Ok(()), Err(e) => { error!( @@ -748,15 +677,7 @@ where .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?; // Create bucket for directory - match self - .storage - .create_bucket( - &bucket, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.create_bucket(&bucket, session_context.credentials()).await { Ok(_) => { debug!( event = EVENT_FTPS_DIRECTORY_STATE, @@ -856,15 +777,7 @@ where .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?; // Check if bucket exists - match self - .storage - .head_bucket( - &bucket, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.head_bucket(&bucket, session_context.credentials()).await { Ok(_) => Ok(()), Err(e) => { error!( diff --git a/crates/protocols/src/sftp/attrs.rs b/crates/protocols/src/sftp/attrs.rs index bfc3a830b..aac6089f3 100644 --- a/crates/protocols/src/sftp/attrs.rs +++ b/crates/protocols/src/sftp/attrs.rs @@ -137,7 +137,7 @@ impl SftpDriver { // on success. Size and mtime are not returned by HeadBucket. None => { self.authorize(&S3Action::HeadBucket, &bucket, None).await?; - self.run_backend("head_bucket", self.storage.head_bucket(&bucket, self.access_key(), self.secret_key())) + self.run_backend("head_bucket", self.storage.head_bucket(&bucket, self.credentials())) .await?; Ok(s3_attrs_to_sftp(0, None, true)) } @@ -154,11 +154,7 @@ impl SftpDriver { Some(object_key) => { self.authorize(&S3Action::HeadObject, &bucket, Some(&object_key)).await?; match self - .run_backend_with_err( - "head_object", - self.storage - .head_object(&bucket, &object_key, self.access_key(), self.secret_key()), - ) + .run_backend_with_err("head_object", self.storage.head_object(&bucket, &object_key, self.credentials())) .await? { Ok(out) => { @@ -183,10 +179,7 @@ impl SftpDriver { .build() .map_err(|e| s3_error_to_sftp("build_list_objects", e))?; let out = self - .run_backend( - "list_objects_v2", - self.storage.list_objects_v2(input, self.access_key(), self.secret_key()), - ) + .run_backend("list_objects_v2", self.storage.list_objects_v2(input, self.credentials())) .await?; let has_contents = out.contents.map(|c| !c.is_empty()).unwrap_or(false); diff --git a/crates/protocols/src/sftp/dir.rs b/crates/protocols/src/sftp/dir.rs index 24cd03e58..36d137b1d 100644 --- a/crates/protocols/src/sftp/dir.rs +++ b/crates/protocols/src/sftp/dir.rs @@ -102,10 +102,7 @@ impl SftpDriver { let input = builder.build().map_err(|e| s3_error_to_sftp("build_list_objects", e))?; let out = self - .run_backend( - "list_objects_v2", - self.storage.list_objects_v2(input, self.access_key(), self.secret_key()), - ) + .run_backend("list_objects_v2", self.storage.list_objects_v2(input, self.credentials())) .await?; let mut entries = Vec::new(); @@ -196,10 +193,7 @@ impl SftpDriver { // Issue list_objects_v2. On Err the destructive caller never // runs because validate_directory_empty returns the Err. let out = self - .run_backend( - "list_objects_v2", - self.storage.list_objects_v2(input, self.access_key(), self.secret_key()), - ) + .run_backend("list_objects_v2", self.storage.list_objects_v2(input, self.credentials())) .await?; // Count content entries that are not the directory's own marker. @@ -234,7 +228,7 @@ impl SftpDriver { self.authorize(&S3Action::ListBuckets, "", None).await?; let out = self - .run_backend("list_buckets", self.storage.list_buckets(self.access_key(), self.secret_key())) + .run_backend("list_buckets", self.storage.list_buckets(self.credentials())) .await?; let mut entries = Vec::new(); @@ -280,7 +274,7 @@ impl SftpDriver { /// MKDIR for a bucket-level path: authorise and issue CreateBucket. pub(super) async fn mkdir_bucket(&self, bucket: &str) -> Result<(), SftpError> { self.authorize(&S3Action::CreateBucket, bucket, None).await?; - self.run_backend("create_bucket", self.storage.create_bucket(bucket, self.access_key(), self.secret_key())) + self.run_backend("create_bucket", self.storage.create_bucket(bucket, self.credentials())) .await?; Ok(()) } @@ -302,7 +296,7 @@ impl SftpDriver { .body(Some(streaming)) .build() .map_err(|e| s3_error_to_sftp("build_put_object", e))?; - self.run_backend("put_object", self.storage.put_object(input, self.access_key(), self.secret_key())) + self.run_backend("put_object", self.storage.put_object(input, self.credentials())) .await?; Ok(()) } @@ -312,7 +306,7 @@ impl SftpDriver { pub(super) async fn rmdir_bucket(&self, bucket: &str) -> Result<(), SftpError> { self.validate_directory_empty(bucket, "").await?; self.authorize(&S3Action::DeleteBucket, bucket, None).await?; - self.run_backend("delete_bucket", self.storage.delete_bucket(bucket, self.access_key(), self.secret_key())) + self.run_backend("delete_bucket", self.storage.delete_bucket(bucket, self.credentials())) .await?; Ok(()) } @@ -326,12 +320,8 @@ impl SftpDriver { let marker_key = path::encode_dir_object(&prefix); self.authorize(&S3Action::DeleteObject, bucket, Some(&marker_key)).await?; - self.run_backend( - "delete_object", - self.storage - .delete_object(bucket, &marker_key, self.access_key(), self.secret_key()), - ) - .await?; + self.run_backend("delete_object", self.storage.delete_object(bucket, &marker_key, self.credentials())) + .await?; Ok(()) } @@ -398,7 +388,7 @@ impl SftpDriver { if prefix.is_empty() { None } else { Some(prefix.as_str()) }, ) .await?; - self.run_backend("head_bucket", self.storage.head_bucket(&bucket, self.access_key(), self.secret_key())) + self.run_backend("head_bucket", self.storage.head_bucket(&bucket, self.credentials())) .await?; DirCursor::Listing { bucket, diff --git a/crates/protocols/src/sftp/driver.rs b/crates/protocols/src/sftp/driver.rs index d4c4d1289..dcd5191a9 100644 --- a/crates/protocols/src/sftp/driver.rs +++ b/crates/protocols/src/sftp/driver.rs @@ -34,6 +34,7 @@ use crate::common::client::s3::StorageBackend; use crate::common::gateway::{AuthorizationError, S3Action, authorize_operation}; use crate::common::session::SessionContext; use russh_sftp::protocol::{Attrs, Data, File, FileAttributes, Handle, Name, OpenFlags, Packet, Status, StatusCode, Version}; +use rustfs_credentials::Credentials; use rustfs_utils::MaskedAccessKey; use s3s::dto::{AbortMultipartUploadInput, CopyObjectInput, CopySource}; use std::collections::HashMap; @@ -165,16 +166,14 @@ impl SftpDriver { super::read_cache::ReadCache::new(Arc::clone(&self.read_cache_in_use)) } - /// Borrow the authenticated principal's S3 access key. Each StorageBackend - /// call needs this alongside the secret key for signing. + /// Borrow the authenticated principal's S3 access key for diagnostics. pub(super) fn access_key(&self) -> &str { - &self.session_context.principal.user_identity.credentials.access_key + &self.credentials().access_key } - /// Borrow the authenticated principal's S3 secret key. Used together with - /// access_key for signing every backend call. - pub(super) fn secret_key(&self) -> &str { - &self.session_context.principal.user_identity.credentials.secret_key + /// Borrow the authenticated principal credentials for backend calls. + pub(super) fn credentials(&self) -> &Credentials { + self.session_context.credentials() } /// Returns Err(PermissionDenied) when the driver is read-only, @@ -787,12 +786,8 @@ impl russh_sftp::server::Handler for self.authorize(&S3Action::DeleteObject, &bucket, Some(&object_key)).await?; - self.run_backend( - "delete_object", - self.storage - .delete_object(&bucket, &object_key, self.access_key(), self.secret_key()), - ) - .await?; + self.run_backend("delete_object", self.storage.delete_object(&bucket, &object_key, self.credentials())) + .await?; Ok(ok_status(id)) } @@ -898,11 +893,7 @@ impl russh_sftp::server::Handler for // single-shot vs multipart-copy branch below. self.authorize(&S3Action::HeadObject, &src_bucket, Some(&src_object)).await?; let head = self - .run_backend( - "head_object", - self.storage - .head_object(&src_bucket, &src_object, self.access_key(), self.secret_key()), - ) + .run_backend("head_object", self.storage.head_object(&src_bucket, &src_object, self.credentials())) .await?; let content_length = head.content_length.unwrap_or(0).max(0) as u64; @@ -920,7 +911,7 @@ impl russh_sftp::server::Handler for .key(dst_object.clone()) .build() .map_err(|e| s3_error_to_sftp("build_copy_object", e))?; - self.run_backend("copy_object", self.storage.copy_object(input, self.access_key(), self.secret_key())) + self.run_backend("copy_object", self.storage.copy_object(input, self.credentials())) .await?; } else { self.multipart_copy(&src_bucket, &src_object, &dst_bucket, &dst_object, content_length) @@ -932,12 +923,8 @@ impl russh_sftp::server::Handler for // delete separately. self.authorize(&S3Action::DeleteObject, &src_bucket, Some(&src_object)) .await?; - self.run_backend( - "delete_object", - self.storage - .delete_object(&src_bucket, &src_object, self.access_key(), self.secret_key()), - ) - .await?; + self.run_backend("delete_object", self.storage.delete_object(&src_bucket, &src_object, self.credentials())) + .await?; Ok(ok_status(id)) } @@ -1029,14 +1016,12 @@ impl russh_sftp::server::Handler for impl Drop for SftpDriver { fn drop(&mut self) { // Snapshot credentials, peer IP, and the per-call backend - // timeout before draining the handle table. self.access_key() - // and self.secret_key() borrow self.session_context immutably, - // which conflicts with the mutable borrow of self.handles - // inside the loop. The timeout is copied into each spawned - // abort task so the deadline applies uniformly to inline calls - // and Drop-time aborts. - let access_key = self.session_context.principal.user_identity.credentials.access_key.clone(); - let secret_key = self.session_context.principal.user_identity.credentials.secret_key.clone(); + // timeout before draining the handle table. Borrowing + // self.session_context inside the loop would conflict with the + // mutable borrow of self.handles. The timeout is copied into each + // spawned abort task so the deadline applies uniformly to inline + // calls and Drop-time aborts. + let credentials = self.session_context.credentials().clone(); let peer = self.session_context.source_ip; let backend_op_timeout_secs = self.backend_op_timeout_secs; @@ -1056,7 +1041,7 @@ impl Drop for SftpDriver { key = %key, upload_id = %upload_id, peer = %peer, - access_key = %access_key, + access_key = %MaskedAccessKey(&credentials.access_key), "skipped abort of orphaned multipart upload on session drop, principal lacks s3:AbortMultipartUpload, bucket lifecycle rules must reclaim parts", ); } @@ -1065,8 +1050,7 @@ impl Drop for SftpDriver { }; let storage = Arc::clone(&self.storage); - let access_key = access_key.clone(); - let secret_key = secret_key.clone(); + let credentials = credentials.clone(); let upload_id = upload_id_owned; // Cap the global abort fan-out so a burst of session @@ -1122,7 +1106,7 @@ impl Drop for SftpDriver { }; match tokio::time::timeout( std::time::Duration::from_secs(backend_op_timeout_secs), - storage.abort_multipart_upload(input, &access_key, &secret_key), + storage.abort_multipart_upload(input, &credentials), ) .await { diff --git a/crates/protocols/src/sftp/read.rs b/crates/protocols/src/sftp/read.rs index 00f5e4af5..5d366b68d 100644 --- a/crates/protocols/src/sftp/read.rs +++ b/crates/protocols/src/sftp/read.rs @@ -46,11 +46,7 @@ impl SftpDriver { // the body. These are cached on the handle so READ can detect EOF // and FSTAT can answer without another backend call. let head = self - .run_backend( - "head_object", - self.storage - .head_object(&bucket, &object_key, self.access_key(), self.secret_key()), - ) + .run_backend("head_object", self.storage.head_object(&bucket, &object_key, self.credentials())) .await?; let size = head.content_length.unwrap_or(0).max(0) as u64; let mtime = timestamp_to_mtime(head.last_modified); @@ -166,7 +162,7 @@ impl SftpDriver { .run_backend( "get_object_range", self.storage - .get_object_range(bucket, key, self.access_key(), self.secret_key(), offset, fetch_len), + .get_object_range(bucket, key, self.credentials(), offset, fetch_len), ) .await?; diff --git a/crates/protocols/src/sftp/write.rs b/crates/protocols/src/sftp/write.rs index df92d50fb..0c3c7143f 100644 --- a/crates/protocols/src/sftp/write.rs +++ b/crates/protocols/src/sftp/write.rs @@ -293,11 +293,7 @@ impl SftpDriver { // not-found error means the key is free. Any other error is // propagated rather than misinterpreted as "does not exist". match self - .run_backend_with_err( - "head_object", - self.storage - .head_object(&bucket, &object_key, self.access_key(), self.secret_key()), - ) + .run_backend_with_err("head_object", self.storage.head_object(&bucket, &object_key, self.credentials())) .await? { Ok(_) => return Err(SftpError::code(StatusCode::Failure)), @@ -385,7 +381,7 @@ impl SftpDriver { .map_err(|e| s3_error_to_sftp("build_put_object", e))?; let outcome = self - .run_backend_with_err("put_object", self.storage.put_object(input, self.access_key(), self.secret_key())) + .run_backend_with_err("put_object", self.storage.put_object(input, self.credentials())) .await?; let backend_err = match outcome { @@ -448,7 +444,7 @@ impl SftpDriver { .map_err(|e| s3_error_to_sftp("build_upload_part", e))?; let out = self - .run_backend("upload_part", self.storage.upload_part(input, self.access_key(), self.secret_key())) + .run_backend("upload_part", self.storage.upload_part(input, self.credentials())) .await?; let e_tag = out.e_tag.ok_or_else(|| { @@ -528,11 +524,7 @@ impl SftpDriver { .map_err(|e| s3_error_to_sftp("build_create_multipart_upload", e))?; let out = self - .run_backend( - "create_multipart_upload", - self.storage - .create_multipart_upload(input, self.access_key(), self.secret_key()), - ) + .run_backend("create_multipart_upload", self.storage.create_multipart_upload(input, self.credentials())) .await?; let upload_id = out.upload_id.ok_or_else(|| { @@ -585,8 +577,7 @@ impl SftpDriver { let result = self .run_backend( "complete_multipart_upload", - self.storage - .complete_multipart_upload(input, self.access_key(), self.secret_key()), + self.storage.complete_multipart_upload(input, self.credentials()), ) .await; result?; @@ -852,12 +843,8 @@ impl SftpDriver { .build() .map_err(|e| s3_error_to_sftp("build_abort_multipart_upload", e))?; - self.run_backend( - "abort_multipart_upload", - self.storage - .abort_multipart_upload(input, self.access_key(), self.secret_key()), - ) - .await?; + self.run_backend("abort_multipart_upload", self.storage.abort_multipart_upload(input, self.credentials())) + .await?; Ok(()) } @@ -1066,10 +1053,7 @@ impl SftpDriver { .map_err(|e| s3_error_to_sftp("build_upload_part_copy", e))?; let out = self - .run_backend( - "upload_part_copy", - self.storage.upload_part_copy(input, self.access_key(), self.secret_key()), - ) + .run_backend("upload_part_copy", self.storage.upload_part_copy(input, self.credentials())) .await?; let e_tag = out.copy_part_result.and_then(|r| r.e_tag).ok_or_else(|| { @@ -1125,6 +1109,7 @@ mod tests { use crate::common::dummy_storage::{AbortCall, DummyBackend, DummyError}; use crate::common::gateway::with_test_auth_override; use russh_sftp::protocol::{FileAttributes, OpenFlags, StatusCode}; + use rustfs_credentials::Credentials; use s3s::dto::ETag; use std::sync::Arc; use std::time::Duration; @@ -2324,9 +2309,10 @@ mod tests { let backend = Arc::new(DummyBackend::new()); backend.queue_head_object_err(DummyError::AccessDenied("pinned".to_string())); let driver = build_driver(backend, TEST_PART_SIZE); + let credentials = Credentials::default(); let result = driver - .run_backend_with_err("head_object", driver.storage.head_object("b", "k", "ak", "sk")) + .run_backend_with_err("head_object", driver.storage.head_object("b", "k", &credentials)) .await; match result { diff --git a/crates/protocols/src/webdav/driver.rs b/crates/protocols/src/webdav/driver.rs index f1d7f511c..d6bbffc6b 100644 --- a/crates/protocols/src/webdav/driver.rs +++ b/crates/protocols/src/webdav/driver.rs @@ -22,6 +22,7 @@ use dav_server::fs::{ }; use futures_util::{FutureExt, StreamExt, stream}; use percent_encoding::percent_decode_str; +use rustfs_credentials::Credentials; use rustfs_utils::MaskedAccessKey; use rustfs_utils::path; use s3s::S3ErrorCode; @@ -198,15 +199,7 @@ where let key = self.key.clone(); async move { - match storage - .head_object( - &bucket, - &key, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match storage.head_object(&bucket, &key, session_context.credentials()).await { Ok(output) => { let size = output.content_length.unwrap_or(0) as u64; let modified = output @@ -288,14 +281,7 @@ where async move { let start_pos = *position.read().await; match storage - .get_object_range( - &bucket, - &key, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - start_pos, - count as u64, - ) + .get_object_range(&bucket, &key, session_context.credentials(), start_pos, count as u64) .await { Ok(output) => { @@ -407,14 +393,7 @@ where .build() .map_err(|_| FsError::GeneralFailure)?; - match storage - .put_object( - put_input, - &session_context.principal.user_identity.credentials.access_key, - &session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match storage.put_object(put_input, session_context.credentials()).await { Ok(_) => { debug!( event = EVENT_WEBDAV_OBJECT_WRITE_STATE, @@ -522,11 +501,8 @@ where self } - fn credentials(&self) -> (&str, &str) { - ( - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) + fn credentials(&self) -> &Credentials { + self.session_context.credentials() } fn is_missing_head_object_error(error: &str) -> bool { @@ -538,7 +514,7 @@ where } async fn prefix_has_entries(&self, bucket: &str, prefix: &str) -> FsResult { - let (access_key, secret_key) = self.credentials(); + let credentials = self.credentials(); let list_input = ListObjectsV2Input::builder() .bucket(bucket.to_string()) .prefix(Some(prefix.to_string())) @@ -546,32 +522,28 @@ where .build() .map_err(|_| FsError::GeneralFailure)?; - let output = self - .storage - .list_objects_v2(list_input, access_key, secret_key) - .await - .map_err(|e| { - error!( - event = EVENT_WEBDAV_LIST_FAILED, - component = LOG_COMPONENT_PROTOCOLS, - subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER, - bucket = %bucket, - prefix = %prefix, - error = %e, - "webdav list failed" - ); - FsError::GeneralFailure - })?; + let output = self.storage.list_objects_v2(list_input, credentials).await.map_err(|e| { + error!( + event = EVENT_WEBDAV_LIST_FAILED, + component = LOG_COMPONENT_PROTOCOLS, + subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER, + bucket = %bucket, + prefix = %prefix, + error = %e, + "webdav list failed" + ); + FsError::GeneralFailure + })?; Ok(output.contents.map(|c| !c.is_empty()).unwrap_or(false) || output.common_prefixes.map(|c| !c.is_empty()).unwrap_or(false)) } async fn copy_object_streaming(&self, src_bucket: &str, src_key: &str, dst_bucket: &str, dst_key: &str) -> FsResult<()> { - let (access_key, secret_key) = self.credentials(); + let credentials = self.credentials(); let get_output = self .storage - .get_object(src_bucket, src_key, access_key, secret_key, None) + .get_object(src_bucket, src_key, credentials, None) .await .map_err(|e| { error!( @@ -625,24 +597,21 @@ where let put_input = put_builder.build().map_err(|_| FsError::GeneralFailure)?; - self.storage - .put_object(put_input, access_key, secret_key) - .await - .map_err(|e| { - error!( - event = EVENT_WEBDAV_COPY_FAILED, - component = LOG_COMPONENT_PROTOCOLS, - subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER, - state = "destination_write_failed", - src_bucket = %src_bucket, - src_object = %src_key, - dst_bucket = %dst_bucket, - dst_object = %dst_key, - error = %e, - "webdav copy failed" - ); - FsError::GeneralFailure - })?; + self.storage.put_object(put_input, credentials).await.map_err(|e| { + error!( + event = EVENT_WEBDAV_COPY_FAILED, + component = LOG_COMPONENT_PROTOCOLS, + subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER, + state = "destination_write_failed", + src_bucket = %src_bucket, + src_object = %src_key, + dst_bucket = %dst_bucket, + dst_object = %dst_key, + error = %e, + "webdav copy failed" + ); + FsError::GeneralFailure + })?; Ok(()) } @@ -653,7 +622,7 @@ where dst_bucket: &str, rename_pairs: &[(String, String)], ) -> FsResult<()> { - let (access_key, secret_key) = self.credentials(); + let credentials = self.credentials(); for (src_obj_key, dst_obj_key) in rename_pairs { self.copy_object_streaming(src_bucket, src_obj_key, dst_bucket, dst_obj_key) @@ -662,7 +631,7 @@ where for (src_obj_key, _) in rename_pairs { self.storage - .delete_object(src_bucket, src_obj_key, access_key, secret_key) + .delete_object(src_bucket, src_obj_key, credentials) .await .map_err(|e| { error!( @@ -683,7 +652,7 @@ where } async fn probe_head_object(&self, bucket: &str, key: &str) -> FsResult { - let (access_key, secret_key) = self.credentials(); + let credentials = self.credentials(); if authorize_operation(&self.session_context, &S3Action::HeadObject, bucket, Some(key)) .await @@ -692,7 +661,7 @@ where return Ok(HeadObjectProbe::Forbidden); } - match self.storage.head_object(bucket, key, access_key, secret_key).await { + match self.storage.head_object(bucket, key, credentials).await { Ok(output) => Ok(HeadObjectProbe::Found(Box::new(output))), Err(e) => { let err_msg = e.to_string(); @@ -816,8 +785,8 @@ where async fn list_buckets(&self) -> FsResult> { match authorize_operation(&self.session_context, &S3Action::ListBuckets, "", None).await { Ok(()) => { - let (access_key, secret_key) = self.credentials(); - return match self.storage.list_buckets(access_key, secret_key).await { + let credentials = self.credentials(); + return match self.storage.list_buckets(credentials).await { Ok(output) => Ok(Self::bucket_entries(output)), Err(error) => { error!( @@ -825,7 +794,7 @@ where component = LOG_COMPONENT_PROTOCOLS, subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER, error = %error, - access_key = %MaskedAccessKey(access_key), + access_key = %MaskedAccessKey(credentials.access_key.as_str()), "webdav bucket list failed" ); Err(FsError::GeneralFailure) @@ -908,15 +877,7 @@ where .build() .map_err(|_| FsError::GeneralFailure)?; - match self - .storage - .list_objects_v2( - list_input, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.list_objects_v2(list_input, self.credentials()).await { Ok(output) => { let mut entries = Vec::new(); @@ -1054,15 +1015,7 @@ where let list_input = list_input.build().map_err(|_| FsError::GeneralFailure)?; - if let Ok(output) = self - .storage - .list_objects_v2( - list_input, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + if let Ok(output) = self.storage.list_objects_v2(list_input, self.credentials()).await { // Delete all objects in this page if let Some(objects) = output.contents { for obj in objects { @@ -1071,15 +1024,7 @@ where .await .map_err(|_| FsError::Forbidden)?; - let _ = self - .storage - .delete_object( - bucket, - &obj_key, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await; + let _ = self.storage.delete_object(bucket, &obj_key, self.credentials()).await; } } } @@ -1095,15 +1040,7 @@ where } // Then delete the bucket - match self - .storage - .delete_bucket( - bucket, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.delete_bucket(bucket, self.credentials()).await { Ok(_) => Ok(()), Err(e) if e.to_string().contains("NoSuchBucket") => Ok(()), Err(e) => { @@ -1250,15 +1187,7 @@ where .await .map_err(|_| FsError::Forbidden)?; - match self - .storage - .head_bucket( - &bucket, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.head_bucket(&bucket, self.credentials()).await { Ok(_) => Ok(Box::new(WebDavMetaData { size: 0, modified: SystemTime::now(), @@ -1318,15 +1247,7 @@ where .build() .map_err(|_| FsError::GeneralFailure)?; - match self - .storage - .put_object( - put_input, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.put_object(put_input, self.credentials()).await { Ok(_) => { debug!( event = EVENT_WEBDAV_DIRECTORY_STATE, @@ -1360,15 +1281,7 @@ where .await .map_err(|_| FsError::Forbidden)?; - match self - .storage - .create_bucket( - &bucket, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.create_bucket(&bucket, self.credentials()).await { Ok(_) => { debug!( event = EVENT_WEBDAV_DIRECTORY_STATE, @@ -1438,15 +1351,7 @@ where let list_input = list_input.build().map_err(|_| FsError::GeneralFailure)?; - if let Ok(output) = self - .storage - .list_objects_v2( - list_input, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + if let Ok(output) = self.storage.list_objects_v2(list_input, self.credentials()).await { if let Some(objects) = output.contents { for obj in objects { if let Some(obj_key) = obj.key { @@ -1454,15 +1359,7 @@ where .await .map_err(|_| FsError::Forbidden)?; - let _ = self - .storage - .delete_object( - &bucket, - &obj_key, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await; + let _ = self.storage.delete_object(&bucket, &obj_key, self.credentials()).await; } } } @@ -1479,12 +1376,7 @@ where // Also delete the directory marker itself let _ = self .storage - .delete_object( - &bucket, - &prefix_with_slash, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) + .delete_object(&bucket, &prefix_with_slash, self.credentials()) .await; return Ok(()); @@ -1515,16 +1407,7 @@ where .await .map_err(|_| FsError::Forbidden)?; - match self - .storage - .delete_object( - &bucket, - &key, - &self.session_context.principal.user_identity.credentials.access_key, - &self.session_context.principal.user_identity.credentials.secret_key, - ) - .await - { + match self.storage.delete_object(&bucket, &key, self.credentials()).await { Ok(_) => { debug!( event = EVENT_WEBDAV_OBJECT_DELETE_STATE, @@ -1566,7 +1449,7 @@ where let src_key = src_key.ok_or(FsError::Forbidden)?; let dst_key = dst_key.ok_or(FsError::Forbidden)?; - let (access_key, secret_key) = self.credentials(); + let credentials = self.credentials(); let resolved_src = self.resolve_path(&src_bucket, &src_key).await?; let (src_prefix, include_src_marker) = match resolved_src { ResolvedPath::File(_) => { @@ -1584,7 +1467,7 @@ where .await?; self.storage - .delete_object(&src_bucket, &src_key, access_key, secret_key) + .delete_object(&src_bucket, &src_key, credentials) .await .map_err(|e| { error!( @@ -1656,25 +1539,21 @@ where } let list_input = list_builder.build().map_err(|_| FsError::GeneralFailure)?; - let output = self - .storage - .list_objects_v2(list_input, access_key, secret_key) - .await - .map_err(|e| { - error!( - event = EVENT_WEBDAV_RENAME_STATE, - component = LOG_COMPONENT_PROTOCOLS, - subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER, - state = "directory_list_failed", - src_bucket = %src_bucket, - src_prefix = %src_prefix, - dst_bucket = %dst_bucket, - dst_prefix = %dst_prefix, - error = %e, - "WebDAV rename directory listing failed" - ); - FsError::GeneralFailure - })?; + let output = self.storage.list_objects_v2(list_input, credentials).await.map_err(|e| { + error!( + event = EVENT_WEBDAV_RENAME_STATE, + component = LOG_COMPONENT_PROTOCOLS, + subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER, + state = "directory_list_failed", + src_bucket = %src_bucket, + src_prefix = %src_prefix, + dst_bucket = %dst_bucket, + dst_prefix = %dst_prefix, + error = %e, + "WebDAV rename directory listing failed" + ); + FsError::GeneralFailure + })?; let mut page_pairs: Vec<(String, String)> = Vec::new(); if let Some(objects) = output.contents { @@ -1785,8 +1664,7 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, _start_pos: Option, ) -> Result { unreachable!("parse_path tests should not hit storage") @@ -1796,20 +1674,14 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, _start_pos: u64, _length: u64, ) -> Result { unreachable!("parse_path tests should not hit storage") } - async fn put_object( - &self, - _input: PutObjectInput, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn put_object(&self, _input: PutObjectInput, _credentials: &Credentials) -> Result { unreachable!("parse_path tests should not hit storage") } @@ -1817,8 +1689,7 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } @@ -1827,57 +1698,39 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } - async fn head_bucket( - &self, - _bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn head_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { unreachable!("parse_path tests should not hit storage") } async fn list_objects_v2( &self, _input: ListObjectsV2Input, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } - async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result { + async fn list_buckets(&self, _credentials: &Credentials) -> Result { unreachable!("parse_path tests should not hit storage") } - async fn create_bucket( - &self, - _bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { unreachable!("parse_path tests should not hit storage") } - async fn delete_bucket( - &self, - _bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn delete_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { unreachable!("parse_path tests should not hit storage") } async fn copy_object( &self, _input: CopyObjectInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } @@ -1885,8 +1738,7 @@ mod tests { async fn create_multipart_upload( &self, _input: CreateMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } @@ -1894,8 +1746,7 @@ mod tests { async fn upload_part( &self, _input: UploadPartInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } @@ -1903,8 +1754,7 @@ mod tests { async fn complete_multipart_upload( &self, _input: CompleteMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } @@ -1912,8 +1762,7 @@ mod tests { async fn abort_multipart_upload( &self, _input: AbortMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } @@ -1921,8 +1770,7 @@ mod tests { async fn upload_part_copy( &self, _input: UploadPartCopyInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("parse_path tests should not hit storage") } @@ -2099,8 +1947,7 @@ mod tests { &self, bucket: &str, key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, _start_pos: Option, ) -> Result { let data = self @@ -2127,8 +1974,7 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, _start_pos: u64, _length: u64, ) -> Result { @@ -2138,8 +1984,7 @@ mod tests { async fn put_object( &self, mut input: PutObjectInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { let bucket = input.bucket.clone(); let key = input.key.clone(); @@ -2163,8 +2008,7 @@ mod tests { &self, bucket: &str, key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { let mut state = self.state.lock().expect("recording storage lock poisoned"); state.delete_keys.push(key.to_string()); @@ -2179,26 +2023,19 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("head_object is not used in rename regression tests") } - async fn head_bucket( - &self, - _bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn head_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { unreachable!("head_bucket is not used in rename regression tests") } async fn list_objects_v2( &self, input: ListObjectsV2Input, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { let prefix = input.prefix.unwrap_or_default(); let mut keys: Vec = self @@ -2226,25 +2063,15 @@ mod tests { }) } - async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result { + async fn list_buckets(&self, _credentials: &Credentials) -> Result { unreachable!("list_buckets is not used in rename regression tests") } - async fn create_bucket( - &self, - _bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { unreachable!("create_bucket is not used in rename regression tests") } - async fn delete_bucket( - &self, - bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn delete_bucket(&self, bucket: &str, _credentials: &Credentials) -> Result { self.state .lock() .expect("recording storage lock poisoned") @@ -2256,8 +2083,7 @@ mod tests { async fn copy_object( &self, _input: CopyObjectInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("copy_object is not used in rename regression tests") } @@ -2265,8 +2091,7 @@ mod tests { async fn create_multipart_upload( &self, _input: CreateMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("create_multipart_upload is not used in rename regression tests") } @@ -2274,8 +2099,7 @@ mod tests { async fn upload_part( &self, _input: UploadPartInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("upload_part is not used in rename regression tests") } @@ -2283,8 +2107,7 @@ mod tests { async fn complete_multipart_upload( &self, _input: CompleteMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("complete_multipart_upload is not used in rename regression tests") } @@ -2292,8 +2115,7 @@ mod tests { async fn abort_multipart_upload( &self, _input: AbortMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("abort_multipart_upload is not used in rename regression tests") } @@ -2301,8 +2123,7 @@ mod tests { async fn upload_part_copy( &self, _input: UploadPartCopyInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("upload_part_copy is not used in rename regression tests") } diff --git a/crates/protocols/src/webdav/server.rs b/crates/protocols/src/webdav/server.rs index 8162e8601..69558bd5b 100644 --- a/crates/protocols/src/webdav/server.rs +++ b/crates/protocols/src/webdav/server.rs @@ -687,6 +687,7 @@ mod tests { use futures_util::stream; use http_body_util::StreamBody; use hyper::body::Frame; + use rustfs_credentials::Credentials; use s3s::dto::*; use std::fmt::{Debug, Formatter}; use std::net::{Ipv4Addr, SocketAddr}; @@ -715,8 +716,7 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, _start_pos: Option, ) -> Result { unreachable!("connection tests should not hit storage") @@ -726,20 +726,14 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, _start_pos: u64, _length: u64, ) -> Result { unreachable!("connection tests should not hit storage") } - async fn put_object( - &self, - _input: PutObjectInput, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn put_object(&self, _input: PutObjectInput, _credentials: &Credentials) -> Result { unreachable!("connection tests should not hit storage") } @@ -747,8 +741,7 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } @@ -757,57 +750,39 @@ mod tests { &self, _bucket: &str, _key: &str, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } - async fn head_bucket( - &self, - _bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn head_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { unreachable!("connection tests should not hit storage") } async fn list_objects_v2( &self, _input: ListObjectsV2Input, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } - async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result { + async fn list_buckets(&self, _credentials: &Credentials) -> Result { unreachable!("connection tests should not hit storage") } - async fn create_bucket( - &self, - _bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { unreachable!("connection tests should not hit storage") } - async fn delete_bucket( - &self, - _bucket: &str, - _access_key: &str, - _secret_key: &str, - ) -> Result { + async fn delete_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result { unreachable!("connection tests should not hit storage") } async fn copy_object( &self, _input: CopyObjectInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } @@ -815,8 +790,7 @@ mod tests { async fn create_multipart_upload( &self, _input: CreateMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } @@ -824,8 +798,7 @@ mod tests { async fn upload_part( &self, _input: UploadPartInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } @@ -833,8 +806,7 @@ mod tests { async fn complete_multipart_upload( &self, _input: CompleteMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } @@ -842,8 +814,7 @@ mod tests { async fn abort_multipart_upload( &self, _input: AbortMultipartUploadInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } @@ -851,8 +822,7 @@ mod tests { async fn upload_part_copy( &self, _input: UploadPartCopyInput, - _access_key: &str, - _secret_key: &str, + _credentials: &Credentials, ) -> Result { unreachable!("connection tests should not hit storage") } diff --git a/rustfs/src/protocols/client.rs b/rustfs/src/protocols/client.rs index da8139776..cc77ba5e2 100644 --- a/rustfs/src/protocols/client.rs +++ b/rustfs/src/protocols/client.rs @@ -163,8 +163,7 @@ fn build_object_uri(bucket: &str, key: &str, query: &[(&str, Option<&str>)]) -> struct RequestParams<'a> { bucket: Option, object: Option, - access_key: &'a str, - secret_key: &'a str, + credentials: &'a rustfs_credentials::Credentials, } /// Protocol storage client that implements the StorageBackend trait @@ -181,39 +180,22 @@ impl ProtocolStorageClient { } /// Create a proper S3Request with ReqInfo extension for authorization - async fn create_request( - &self, - input: T, - method: Method, - uri: http::Uri, - params: RequestParams<'_>, - ) -> S3Result> { + fn create_request(input: T, method: Method, uri: http::Uri, params: RequestParams<'_>) -> S3Result> { let mut extensions = http::Extensions::default(); let is_owner = if let Some(global_cred) = current_action_credentials() { - params.access_key == global_cred.access_key + params.credentials.access_key == global_cred.access_key } else { false }; let credentials = Some(s3s::auth::Credentials { - access_key: params.access_key.to_string(), - secret_key: params.secret_key.to_string().into(), + access_key: params.credentials.access_key.clone(), + secret_key: params.credentials.secret_key.clone().into(), }); extensions.insert(ReqInfo { - cred: Some(rustfs_credentials::Credentials { - access_key: params.access_key.to_string(), - secret_key: params.secret_key.to_string(), - session_token: String::new(), - expiration: None, - status: String::new(), - parent_user: String::new(), - groups: None, - claims: None, - name: None, - description: None, - }), + cred: Some(params.credentials.clone()), is_owner, bucket: params.bucket, object: params.object, @@ -247,8 +229,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli &self, bucket: &str, key: &str, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, start_pos: Option, ) -> Result { trace_protocol_request("get_object", Some(bucket), Some(key)); @@ -279,19 +260,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli })?; let uri = build_object_uri(bucket, key, &[])?; - let req = self - .create_request( - input, - Method::GET, - uri, - RequestParams { - bucket: Some(bucket.to_string()), - object: Some(key.to_string()), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::GET, + uri, + RequestParams { + bucket: Some(bucket.to_string()), + object: Some(key.to_string()), + credentials, + }, + )?; match self.fs.get_object(req).await { Ok(response) => Ok(response.output), @@ -302,8 +280,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli async fn put_object( &self, input: PutObjectInput, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace!( event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST, @@ -330,19 +307,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli } } - let req = self - .create_request( - input, - Method::PUT, - uri, - RequestParams { - bucket: Some(bucket), - object: Some(key), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::PUT, + uri, + RequestParams { + bucket: Some(bucket), + object: Some(key), + credentials, + }, + )?; let req = S3Request { headers, ..req }; match self.fs.put_object(req).await { @@ -355,8 +329,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli &self, bucket: &str, key: &str, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace_protocol_request("delete_object", Some(bucket), Some(key)); @@ -369,19 +342,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli })?; let uri = build_object_uri(bucket, key, &[])?; - let req = self - .create_request( - input, - Method::DELETE, - uri, - RequestParams { - bucket: Some(bucket.to_string()), - object: Some(key.to_string()), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::DELETE, + uri, + RequestParams { + bucket: Some(bucket.to_string()), + object: Some(key.to_string()), + credentials, + }, + )?; match self.fs.delete_object(req).await { Ok(response) => Ok(response.output), @@ -393,8 +363,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli &self, bucket: &str, key: &str, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace_protocol_request("head_object", Some(bucket), Some(key)); @@ -407,19 +376,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli })?; let uri = build_object_uri(bucket, key, &[])?; - let req = self - .create_request( - input, - Method::HEAD, - uri, - RequestParams { - bucket: Some(bucket.to_string()), - object: Some(key.to_string()), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::HEAD, + uri, + RequestParams { + bucket: Some(bucket.to_string()), + object: Some(key.to_string()), + credentials, + }, + )?; match self.fs.head_object(req).await { Ok(response) => Ok(response.output), @@ -427,7 +393,11 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli } } - async fn head_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result { + async fn head_bucket( + &self, + bucket: &str, + credentials: &rustfs_credentials::Credentials, + ) -> Result { trace_protocol_request("head_bucket", Some(bucket), None); let input = HeadBucketInput::builder().bucket(bucket.to_string()).build().map_err(|e| { @@ -435,19 +405,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli })?; let uri = build_bucket_uri(bucket, &[])?; - let req = self - .create_request( - input, - Method::HEAD, - uri, - RequestParams { - bucket: Some(bucket.to_string()), - object: None, - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::HEAD, + uri, + RequestParams { + bucket: Some(bucket.to_string()), + object: None, + credentials, + }, + )?; match self.fs.head_bucket(req).await { Ok(response) => Ok(response.output), @@ -458,26 +425,22 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli async fn list_objects_v2( &self, input: ListObjectsV2Input, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace_protocol_request("list_objects_v2", Some(&input.bucket), None); let bucket = input.bucket.clone(); let uri = build_bucket_uri(&bucket, &[("list-type", Some("2"))])?; - let req = self - .create_request( - input, - Method::GET, - uri, - RequestParams { - bucket: Some(bucket), - object: None, - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::GET, + uri, + RequestParams { + bucket: Some(bucket), + object: None, + credentials, + }, + )?; match self.fs.list_objects_v2(req).await { Ok(response) => Ok(response.output), @@ -485,13 +448,13 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli } } - async fn list_buckets(&self, access_key: &str, secret_key: &str) -> Result { + async fn list_buckets(&self, credentials: &rustfs_credentials::Credentials) -> Result { trace!( event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST, component = LOG_COMPONENT_PROTOCOLS, subsystem = LOG_SUBSYSTEM_STORAGE_CLIENT, operation = "list_buckets", - access_key = %MaskedAccessKey(access_key), + access_key = %MaskedAccessKey(&credentials.access_key), "Protocol storage client request" ); @@ -499,19 +462,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli s3s::S3Error::with_message(s3s::S3ErrorCode::InvalidRequest, format!("Failed to build ListBucketsInput: {}", e)) })?; - let req = self - .create_request( - input, - Method::GET, - http::Uri::from_static("/"), - RequestParams { - bucket: None, - object: None, - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::GET, + http::Uri::from_static("/"), + RequestParams { + bucket: None, + object: None, + credentials, + }, + )?; match self.fs.list_buckets(req).await { Ok(response) => Ok(response.output), @@ -542,7 +502,11 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli self.fs.list_buckets(request).await.map(|response| response.output) } - async fn create_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result { + async fn create_bucket( + &self, + bucket: &str, + credentials: &rustfs_credentials::Credentials, + ) -> Result { trace_protocol_request("create_bucket", Some(bucket), None); let input = CreateBucketInput::builder().bucket(bucket.to_string()).build().map_err(|e| { @@ -550,19 +514,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli })?; let uri = build_bucket_uri(bucket, &[])?; - let req = self - .create_request( - input, - Method::PUT, - uri, - RequestParams { - bucket: Some(bucket.to_string()), - object: None, - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::PUT, + uri, + RequestParams { + bucket: Some(bucket.to_string()), + object: None, + credentials, + }, + )?; match self.fs.create_bucket(req).await { Ok(response) => Ok(response.output), @@ -574,8 +535,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli &self, bucket: &str, key: &str, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, start_pos: u64, length: u64, ) -> Result { @@ -607,19 +567,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli })?; let uri = build_object_uri(bucket, key, &[])?; - let req = self - .create_request( - input, - Method::GET, - uri, - RequestParams { - bucket: Some(bucket.to_string()), - object: Some(key.to_string()), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::GET, + uri, + RequestParams { + bucket: Some(bucket.to_string()), + object: Some(key.to_string()), + credentials, + }, + )?; match self.fs.get_object(req).await { Ok(response) => Ok(response.output), @@ -630,8 +587,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli async fn copy_object( &self, input: CopyObjectInput, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace!( event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST, @@ -647,19 +603,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli let key = input.key.clone(); let uri = build_object_uri(&bucket, &key, &[])?; - let req = self - .create_request( - input, - Method::PUT, - uri, - RequestParams { - bucket: Some(bucket), - object: Some(key), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::PUT, + uri, + RequestParams { + bucket: Some(bucket), + object: Some(key), + credentials, + }, + )?; match self.fs.copy_object(req).await { Ok(response) => Ok(response.output), @@ -667,7 +620,11 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli } } - async fn delete_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result { + async fn delete_bucket( + &self, + bucket: &str, + credentials: &rustfs_credentials::Credentials, + ) -> Result { trace_protocol_request("delete_bucket", Some(bucket), None); let input = DeleteBucketInput::builder().bucket(bucket.to_string()).build().map_err(|e| { @@ -675,19 +632,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli })?; let uri = build_bucket_uri(bucket, &[])?; - let req = self - .create_request( - input, - Method::DELETE, - uri, - RequestParams { - bucket: Some(bucket.to_string()), - object: None, - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::DELETE, + uri, + RequestParams { + bucket: Some(bucket.to_string()), + object: None, + credentials, + }, + )?; match self.fs.delete_bucket(req).await { Ok(response) => Ok(response.output), @@ -698,8 +652,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli async fn create_multipart_upload( &self, input: CreateMultipartUploadInput, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace!( event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST, @@ -715,19 +668,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli let key = input.key.clone(); let uri = build_object_uri(&bucket, &key, &[("uploads", None)])?; - let req = self - .create_request( - input, - Method::POST, - uri, - RequestParams { - bucket: Some(bucket), - object: Some(key), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::POST, + uri, + RequestParams { + bucket: Some(bucket), + object: Some(key), + credentials, + }, + )?; match self.fs.create_multipart_upload(req).await { Ok(response) => Ok(response.output), @@ -738,8 +688,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli async fn upload_part( &self, input: UploadPartInput, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace!( event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST, @@ -786,19 +735,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli } } - let req = self - .create_request( - input, - Method::PUT, - uri, - RequestParams { - bucket: Some(bucket), - object: Some(key), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::PUT, + uri, + RequestParams { + bucket: Some(bucket), + object: Some(key), + credentials, + }, + )?; let req = S3Request { headers, ..req }; match self.fs.upload_part(req).await { @@ -810,8 +756,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli async fn complete_multipart_upload( &self, input: CompleteMultipartUploadInput, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace!( event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST, @@ -828,19 +773,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli let upload_id = input.upload_id.clone(); let uri = build_object_uri(&bucket, &key, &[("uploadId", Some(upload_id.as_str()))])?; - let req = self - .create_request( - input, - Method::POST, - uri, - RequestParams { - bucket: Some(bucket), - object: Some(key), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::POST, + uri, + RequestParams { + bucket: Some(bucket), + object: Some(key), + credentials, + }, + )?; match self.fs.complete_multipart_upload(req).await { Ok(response) => Ok(response.output), @@ -851,8 +793,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli async fn abort_multipart_upload( &self, input: AbortMultipartUploadInput, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace!( event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST, @@ -870,19 +811,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli let upload_id = input.upload_id.clone(); let uri = build_object_uri(&bucket, &key, &[("uploadId", Some(upload_id.as_str()))])?; - let req = self - .create_request( - input, - Method::DELETE, - uri, - RequestParams { - bucket: Some(bucket), - object: Some(key), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::DELETE, + uri, + RequestParams { + bucket: Some(bucket), + object: Some(key), + credentials, + }, + )?; match self.fs.abort_multipart_upload(req).await { Ok(response) => Ok(response.output), @@ -893,8 +831,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli async fn upload_part_copy( &self, input: UploadPartCopyInput, - access_key: &str, - secret_key: &str, + credentials: &rustfs_credentials::Credentials, ) -> Result { trace!( event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST, @@ -921,19 +858,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli ], )?; - let req = self - .create_request( - input, - Method::PUT, - uri, - RequestParams { - bucket: Some(bucket), - object: Some(key), - access_key, - secret_key, - }, - ) - .await?; + let req = Self::create_request( + input, + Method::PUT, + uri, + RequestParams { + bucket: Some(bucket), + object: Some(key), + credentials, + }, + )?; match self.fs.upload_part_copy(req).await { Ok(response) => Ok(response.output), @@ -945,6 +879,47 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli #[cfg(test)] mod tests { use super::*; + use rustfs_credentials::{IAM_POLICY_CLAIM_NAME_SA, INHERITED_POLICY_TYPE}; + + #[test] + fn create_request_preserves_authenticated_service_account_identity() { + let claims = std::collections::HashMap::from([ + ("parent".to_string(), serde_json::json!("alice")), + (IAM_POLICY_CLAIM_NAME_SA.to_string(), serde_json::json!(INHERITED_POLICY_TYPE)), + ]); + let credentials = rustfs_credentials::Credentials { + access_key: "service-account".to_string(), + secret_key: "secret".to_string(), + session_token: "signed-service-account-token".to_string(), + parent_user: "alice".to_string(), + groups: Some(vec!["developers".to_string()]), + claims: Some(claims.clone()), + ..Default::default() + }; + + let request = ProtocolStorageClient::create_request( + ListObjectsV2Input::default(), + Method::GET, + http::Uri::from_static("/bucket?list-type=2"), + RequestParams { + bucket: Some("bucket".to_string()), + object: None, + credentials: &credentials, + }, + ) + .expect("request should build"); + let request_info = request.extensions.get::().expect("request info should be present"); + let copied = request_info.cred.as_ref().expect("credentials should be present"); + + assert_eq!(copied.access_key, credentials.access_key); + assert_eq!(copied.secret_key, credentials.secret_key); + assert_eq!(copied.session_token, credentials.session_token); + assert_eq!(copied.parent_user, credentials.parent_user); + assert_eq!(copied.groups, credentials.groups); + assert_eq!(copied.claims, Some(claims)); + assert!(copied.is_service_account()); + } + #[cfg(feature = "webdav")] #[test] fn request_extensions_preserve_authenticated_identity_and_source_ip() {