fix: preserve protocol service account claims (#7062)

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<u64>,
     ) -> Result<GetObjectOutput, Self::Error>;
     async fn get_object_range(
         &self,
         bucket: &str,
         key: &str,
-        access_key: &str,
-        secret_key: &str,
+        credentials: &Credentials,
         start_pos: u64,
         length: u64,
     ) -> Result<GetObjectOutput, Self::Error>;
     /// Put object content with metadata
-    async fn put_object(&self, input: PutObjectInput, access_key: &str, secret_key: &str)
-    -> Result<PutObjectOutput, Self::Error>;
+    async fn put_object(&self, input: PutObjectInput, credentials: &Credentials) -> Result<PutObjectOutput, Self::Error>;
     /// Delete an object
-    async fn delete_object(
-        &self,
-        bucket: &str,
-        key: &str,
-        access_key: &str,
-        secret_key: &str,
-    ) -> Result<DeleteObjectOutput, Self::Error>;
+    async fn delete_object(&self, bucket: &str, key: &str, credentials: &Credentials) -> Result<DeleteObjectOutput, Self::Error>;
     /// Get object metadata without content
-    async fn head_object(
-        &self,
-        bucket: &str,
-        key: &str,
-        access_key: &str,
-        secret_key: &str,
-    ) -> Result<HeadObjectOutput, Self::Error>;
+    async fn head_object(&self, bucket: &str, key: &str, credentials: &Credentials) -> Result<HeadObjectOutput, Self::Error>;
     /// Check if bucket exists and get metadata
-    async fn head_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<HeadBucketOutput, Self::Error>;
+    async fn head_bucket(&self, bucket: &str, credentials: &Credentials) -> Result<HeadBucketOutput, Self::Error>;
     /// List objects in a bucket with pagination
     async fn list_objects_v2(
         &self,
         input: ListObjectsV2Input,
-        access_key: &str,
-        secret_key: &str,
+        credentials: &Credentials,
     ) -> Result<ListObjectsV2Output, Self::Error>;
     /// List all buckets (requires authentication).
-    async fn list_buckets(&self, access_key: &str, secret_key: &str) -> Result<ListBucketsOutput, Self::Error>;
+    async fn list_buckets(&self, credentials: &Credentials) -> Result<ListBucketsOutput, Self::Error>;
     /// 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<CreateBucketOutput, Self::Error>;
+    async fn create_bucket(&self, bucket: &str, credentials: &Credentials) -> Result<CreateBucketOutput, Self::Error>;
     /// Delete a bucket (must be empty)
-    async fn delete_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<DeleteBucketOutput, Self::Error>;
+    async fn delete_bucket(&self, bucket: &str, credentials: &Credentials) -> Result<DeleteBucketOutput, Self::Error>;
     /// 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<CopyObjectOutput, Self::Error>;
+    async fn copy_object(&self, input: CopyObjectInput, credentials: &Credentials) -> Result<CopyObjectOutput, Self::Error>;
     /// 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<CreateMultipartUploadOutput, Self::Error>;
     /// 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<UploadPartOutput, Self::Error>;
+    async fn upload_part(&self, input: UploadPartInput, credentials: &Credentials) -> Result<UploadPartOutput, Self::Error>;
     /// 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<CompleteMultipartUploadOutput, Self::Error>;
     /// 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<AbortMultipartUploadOutput, Self::Error>;
     /// 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<UploadPartCopyOutput, Self::Error>;
 }

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<u64>,
     ) -> Result<GetObjectOutput, Self::Error> {
         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<GetObjectOutput, Self::Error> {
@@ -630,7 +629,7 @@ impl StorageBackend for DummyBackend {
         }
     }
 
-    async fn put_object(&self, input: PutObjectInput, _ak: &str, _sk: &str) -> Result<PutObjectOutput, Self::Error> {
+    async fn put_object(&self, input: PutObjectInput, _credentials: &Credentials) -> Result<PutObjectOutput, Self::Error> {
         // 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<DeleteObjectOutput, Self::Error> {
+    async fn delete_object(
+        &self,
+        bucket: &str,
+        key: &str,
+        _credentials: &Credentials,
+    ) -> Result<DeleteObjectOutput, Self::Error> {
         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<HeadObjectOutput, Self::Error> {
+    async fn head_object(&self, bucket: &str, key: &str, _credentials: &Credentials) -> Result<HeadObjectOutput, Self::Error> {
         {
             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<HeadBucketOutput, Self::Error> {
+    async fn head_bucket(&self, bucket: &str, _credentials: &Credentials) -> Result<HeadBucketOutput, Self::Error> {
         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<ListObjectsV2Output, Self::Error> {
         // 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<ListBucketsOutput, Self::Error> {
+    async fn list_buckets(&self, _credentials: &Credentials) -> Result<ListBucketsOutput, Self::Error> {
         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<CreateBucketOutput, Self::Error> {
+    async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<CreateBucketOutput, Self::Error> {
         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<DeleteBucketOutput, Self::Error> {
+    async fn delete_bucket(&self, bucket: &str, _credentials: &Credentials) -> Result<DeleteBucketOutput, Self::Error> {
         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<CopyObjectOutput, Self::Error> {
+    async fn copy_object(&self, _input: CopyObjectInput, _credentials: &Credentials) -> Result<CopyObjectOutput, Self::Error> {
         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<CreateMultipartUploadOutput, Self::Error> {
         {
             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<UploadPartOutput, Self::Error> {
+    async fn upload_part(&self, input: UploadPartInput, _credentials: &Credentials) -> Result<UploadPartOutput, Self::Error> {
         // 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<CompleteMultipartUploadOutput, Self::Error> {
         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<AbortMultipartUploadOutput, Self::Error> {
         {
             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<UploadPartCopyOutput, Self::Error> {
         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");

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<String, serde_json::Value> {
+    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<String, Value>) -> 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();

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