fix(s3select): enforce SSE-KMS read authorization (#5698)

* fix(s3select): enforce SSE-KMS read authorization

* fix(app): route select SSE auth through facade

rustfs/src/app/select_object.rs

@@ -1,6 +1,7 @@
 use super::storage_api::select_object::contract::object::ObjectOperations as _;
 use super::storage_api::select_object::options::get_opts;
 use super::storage_api::select_object::request_context::spawn_traced;
+use super::storage_api::select_object::sse::{SseKmsPrincipal, authorize_sse_kms_object_read};
 use super::storage_api::select_object::{get_validated_store, validate_sse_headers_for_read, validate_ssec_for_read};
 use crate::app::runtime_sources::current_s3select_db;
 use crate::error::ApiError;
@@ -62,10 +63,11 @@ enum SelectProducerOutcome {
 pub async fn execute_select_object_content(
     req: S3Request<SelectObjectContentInput>,
 ) -> S3Result<S3Response<SelectObjectContentOutput>> {
+    let read_principal = SseKmsPrincipal::from_request(&req);
     let mut input = req.input;
     let validation = validate_select_request(&req.headers, &mut input)?;
     log_select_request_summary(&input, &validation);
-    let metadata = preflight_select_object(&req.headers, &input).await?;
+    let metadata = preflight_select_object(&req.headers, &input, read_principal.as_ref()).await?;
     validate_scan_range_for_object_size(&input.request, metadata.size)?;
 
     let input = Arc::new(input);
@@ -386,7 +388,11 @@ fn validate_input_delimiter_pair(field_delimiter: Option<&str>, record_delimiter
     Ok(())
 }
 
-async fn preflight_select_object(headers: &http::HeaderMap, input: &SelectObjectContentInput) -> S3Result<SelectObjectMetadata> {
+async fn preflight_select_object(
+    headers: &http::HeaderMap,
+    input: &SelectObjectContentInput,
+    read_principal: Option<&SseKmsPrincipal>,
+) -> S3Result<SelectObjectMetadata> {
     let opts = get_opts(&input.bucket, &input.key, None, None, headers)
         .await
         .map_err(ApiError::from)?;
@@ -397,6 +403,7 @@ async fn preflight_select_object(headers: &http::HeaderMap, input: &SelectObject
         .map_err(ApiError::from)?;
     validate_sse_headers_for_read(&info.user_defined, headers)?;
     validate_ssec_for_read(&info.user_defined, input.sse_customer_key.as_ref(), input.sse_customer_key_md5.as_ref())?;
+    authorize_sse_kms_object_read(read_principal, &info.user_defined).await?;
     Ok(SelectObjectMetadata {
         size: info.size.max(0) as u64,
     })

rustfs/src/app/storage_api.rs

@@ -1128,7 +1128,7 @@ pub(crate) mod select_object {
         }
     }
 
-    pub(crate) use super::{options, request_context};
+    pub(crate) use super::{options, request_context, sse};
     pub(crate) use crate::storage::storage_api::{get_validated_store, validate_sse_headers_for_read, validate_ssec_for_read};
 }
 

rustfs/src/storage/sse.rs

@@ -845,21 +845,21 @@ pub struct SseKmsPrincipal {
 }
 
 impl SseKmsPrincipal {
-    /// Build a principal from an authenticated S3 request.
+    /// Build a principal from an S3 request.
     ///
-    /// Returns `None` for unauthenticated requests: an anonymous caller has no identity
-    /// policy to evaluate, and denying it outright would break public buckets holding
-    /// SSE-KMS objects. Such requests remain governed by bucket policy alone.
+    /// Anonymous requests use an empty account, matching the bucket-policy authorization
+    /// path. This keeps every request that crossed the S3 boundary distinct from trusted
+    /// internal callers, which are represented by `None`.
     pub(crate) fn from_request<T>(req: &S3Request<T>) -> Option<Self> {
         let req_info = req.extensions.get::<ReqInfo>()?;
-        let cred = req_info.cred.as_ref()?;
+        let cred = req_info.cred.as_ref();
 
         Some(Self {
-            account: cred.access_key.clone(),
-            groups: cred.groups.clone(),
+            account: cred.map(|cred| cred.access_key.clone()).unwrap_or_default(),
+            groups: cred.and_then(|cred| cred.groups.clone()),
             is_owner: req_info.is_owner,
-            claims: cred.claims_or_empty().clone(),
-            conditions: resource_free_condition_values(req, cred),
+            claims: cred.map(|cred| cred.claims_or_empty().clone()).unwrap_or_default(),
+            conditions: cred.map(|cred| resource_free_condition_values(req, cred)).unwrap_or_default(),
             request_audit: request_context_from_req(req).and_then(|context| kms_request_audit(&context.request_id)),
             #[cfg(test)]
             test_hooks: None,
@@ -3432,6 +3432,28 @@ mod tests {
     };
     use rustfs_utils::http::headers::SSEC_ALGORITHM_HEADER;
 
+    #[test]
+    fn anonymous_s3_request_builds_kms_principal() {
+        let mut request = s3s::S3Request {
+            input: (),
+            method: http::Method::GET,
+            uri: http::Uri::from_static("/bucket/object"),
+            headers: http::HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+        request.extensions.insert(crate::storage::access::ReqInfo::default());
+
+        let principal = SseKmsPrincipal::from_request(&request).expect("S3-boundary request should build a principal");
+
+        assert!(principal.account.is_empty());
+        assert!(principal.groups.is_none());
+        assert!(principal.claims.is_empty());
+    }
+
     #[test]
     fn parse_simple_sse_cmk_rejects_bad_keys_without_crashing() {
         // Empty / whitespace-only.