fix(app): drop the PutObjectInput re-export main now provides

RequestCredentials::secret_key now returns Option<&str>: None when the edge kept the secret, and an empty handed-over secret is refused at construction. Secret comparisons go through secret_key_matches, which never matches an absent secret, and readers that need the value fail closed with an internal error.

.config/s3s-footprint-baseline.txt

@@ -13,5 +13,5 @@
 #                 header's rule clears (rustfs_s3_types's macro); fails closed
 # ecstore_files   Rust files referencing s3s paths under crates/ecstore/src
 files=65
-s3_error_lines=219
+s3_error_lines=217
 ecstore_files=0

rustfs/src/admin/handlers/account.rs

@@ -165,7 +165,8 @@ impl Operation for ChangeOwnPasswordHandler {
         let signing_secret = req
             .credentials()
             .ok_or_else(|| s3::error(S3ErrorCode::AccessDenied, "authentication required"))?
-            .secret_key()
+            .require_secret_key()
+            .map_err(|err| s3::error(S3ErrorCode::InternalError, err.to_string()))?
             .to_string();
         let caller = CallerIdentity::resolve_envelope(req.envelope()).await?;
         let audit = AccountAuditContext::from_envelope(req.envelope());

rustfs/src/admin/handlers/bucket_recovery.rs

@@ -63,8 +63,10 @@ impl Operation for RecoverOrphanedBucketHandler {
         authorize_admin_envelope(req.envelope(), vec![Action::AdminAction(AdminAction::RecoverOrphanedBucketAction)]).await?;
         let secret_key = req
             .credentials()
-            .map(|credentials| Zeroizing::new(credentials.secret_key().to_owned()))
-            .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "authentication required"))?;
+            .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "authentication required"))?
+            .require_secret_key()
+            .map(|secret| Zeroizing::new(secret.to_owned()))
+            .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))?;
         let bucket = params
             .get("bucket")
             .filter(|bucket| !bucket.is_empty())

rustfs/src/admin/handlers/object_zip_download.rs

@@ -15,11 +15,12 @@
 use crate::admin::operation::{AdminRequest, AdminResponse, Operation};
 use crate::admin::router::{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH, AdminOperation, S3Router};
 use crate::admin::runtime_sources::{ServerContextSlot, current_action_credentials, current_object_store_handle, current_region};
-use crate::admin::storage_api::access::{ReqInfo, authorize_request};
+use crate::admin::storage_api::access::{ReqInfo, authorize_owned_request, authorize_request};
 use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
 use crate::admin::storage_api::contract::list::ListOperations as _;
 use crate::admin::storage_api::contract::object::{ObjectIO as _, ObjectOperations as _};
 use crate::admin::storage_api::object::StorageObjectOptions as ObjectOptions;
+use crate::app::{RequestEnvelope, RequestHead};
 use crate::auth::{check_key_valid, get_session_token};
 use crate::error::ApiError;
 use crate::license::license_check;
@@ -495,25 +496,32 @@ async fn authorize_object_zip_download_s3_action(
         req_info.version_id = None;
     }
 
-    let mut authorization = s3_authorization_request(
-        req.method().clone(),
-        req.uri().clone(),
-        &authenticated_zip_download_auth_context(req),
-        req.envelope().server_context().cloned(),
-        req.req_info().cloned(),
+    let mut extensions = http::Extensions::new();
+    extensions.insert(req.remote_addr().map(RemoteAddr));
+    if let Some(client_info) = req.client_info() {
+        extensions.insert(client_info.clone());
+    }
+    if let Some(server_context) = req.envelope().server_context() {
+        extensions.insert(server_context.clone());
+    }
+    if let Some(req_info) = req.req_info() {
+        extensions.insert(req_info.clone());
+    }
+    let mut authorization = RequestEnvelope::from_parts(
+        RequestHead {
+            method: req.method().clone(),
+            uri: req.uri().clone(),
+            headers: req.headers().clone(),
+            credentials: None,
+            region: None,
+            service: None,
+        },
+        extensions,
     );
-    authorize_request(&mut authorization, Action::S3Action(action))
-        .await
-        .map_err(Into::into)
+    authorize_owned_request(&mut authorization, Action::S3Action(action), None).await
 }
 
-/// The S3 gate (`storage::access::authorize_request`) still evaluates the s3s
-/// request shape, so the facts it reads are copied into one for the duration
-/// of the call: the authenticated `ReqInfo`, the peer and proxy-reported client
-/// addresses, the headers and query the policy conditions come from, and the
-/// server slot that selects the store. The gate reads neither the presented
-/// credentials nor the region or service, so those stay unset, and nothing it
-/// attaches to the request is read afterwards.
+/// Builds the legacy authorization request used when a download token is redeemed.
 fn s3_authorization_request(
     method: Method,
     uri: Uri,
@@ -943,6 +951,103 @@ mod tests {
     use super::*;
     use rustfs_s3_types::S3ErrorCode;
 
+    #[tokio::test]
+    async fn zip_scope_authorizes_real_iam_and_restores_prefix_query() {
+        use crate::admin::operation::test_support::TestRequest;
+        use crate::admin::runtime_sources::{AppContext, IamInterface, KmsInterface};
+        use crate::storage::storage_api::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+        use rustfs_iam::store::{Store as _, UserType};
+        use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
+        use rustfs_policy::policy::Policy;
+
+        struct ReadyIam(Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>);
+        impl IamInterface for ReadyIam {
+            fn handle(&self) -> Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>> {
+                Arc::clone(&self.0)
+            }
+            fn is_ready(&self) -> bool {
+                true
+            }
+        }
+        struct TestKms;
+        impl KmsInterface for TestKms {
+            fn handle(&self) -> Arc<rustfs_kms::KmsServiceManager> {
+                Arc::new(rustfs_kms::KmsServiceManager::new())
+            }
+        }
+
+        let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+        store
+            .make_bucket("photos", &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        rustfs_iam::store::object::ObjectStore::new(Arc::clone(&store))
+            .save_iam_config(
+                serde_json::json!({"version": 1}),
+                format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+            )
+            .await
+            .expect("seed IAM format");
+        let iam = rustfs_iam::build_iam_sys(Arc::clone(&store)).await.expect("initialize IAM");
+        let access_key = "zip-scope-user";
+        iam.create_user(
+            access_key,
+            &AddOrUpdateUserReq {
+                secret_key: "zip-scope-secret-key".to_owned(),
+                policy: None,
+                status: AccountStatus::Enabled,
+            },
+        )
+        .await
+        .expect("create user");
+        iam.set_policy(
+            "zip-scope-policy",
+            Policy::parse_config(br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:ListBucket","Resource":"arn:aws:s3:::photos","Condition":{"StringEquals":{"s3:prefix":"allowed/"}}},{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/allowed/*"}]}"#)
+                .expect("parse policy"),
+        )
+        .await
+        .expect("store policy");
+        iam.policy_db_set(access_key, UserType::Reg, false, "zip-scope-policy")
+            .await
+            .expect("attach policy");
+        let credentials = iam.get_user(access_key).await.expect("load user").credentials;
+        let context = Arc::new(AppContext::new(store, Arc::new(ReadyIam(iam)), Arc::new(TestKms)));
+        let slot = ServerContextSlot::new();
+        assert!(slot.install(context));
+        let original_uri = "/rustfs/admin/v3/object-zip-downloads?prefix=old&marker=1";
+        let mut request = TestRequest::new(Method::POST, original_uri)
+            .extension(slot)
+            .extension(ReqInfo {
+                cred: Some(credentials),
+                ..Default::default()
+            })
+            .build();
+
+        let mut selection = valid_request();
+        selection.prefix = None;
+        selection.prefixes = vec!["allowed/".to_owned()];
+        selection.objects = vec!["allowed/file".to_owned()];
+        authorize_object_zip_download_scope(&mut request, &selection)
+            .await
+            .expect("allowed prefix and object");
+        assert_eq!(request.uri().to_string(), original_uri);
+
+        selection.prefixes = vec!["denied/".to_owned()];
+        let err = authorize_object_zip_download_scope(&mut request, &selection)
+            .await
+            .expect_err("another prefix is denied");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+        assert_eq!(request.uri().to_string(), original_uri);
+
+        selection.prefixes = vec!["allowed/".to_owned()];
+        selection.objects = vec!["denied/file".to_owned()];
+        let err = authorize_object_zip_download_scope(&mut request, &selection)
+            .await
+            .expect_err("another object is denied");
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+        assert_eq!(request.uri().to_string(), original_uri);
+    }
+
     fn valid_request() -> CreateObjectZipDownloadRequest {
         CreateObjectZipDownloadRequest {
             bucket: "photos".to_string(),

rustfs/src/admin/handlers/service_account.rs

@@ -343,7 +343,7 @@ impl Operation for AddServiceAccount {
         let (cred, owner) =
             check_key_valid(get_session_token(req.uri(), req.headers()).unwrap_or_default(), req_cred.access_key()).await?;
 
-        if !constant_time_eq(&cred.secret_key, req_cred.secret_key()) {
+        if !req_cred.secret_key_matches(&cred.secret_key) {
             return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "access denied"));
         }
 

rustfs/src/admin/handlers/sts.rs

@@ -222,7 +222,7 @@ async fn handle_assume_role(
 
     // The signature was verified before the request body was consumed. A fresh
     // IAM lookup must not silently replace the secret that authenticated it.
-    if !crate::auth::constant_time_eq(&cred.secret_key, user.secret_key()) {
+    if !user.secret_key_matches(&cred.secret_key) {
         return Err(S3Error::with_message(S3ErrorCode::AccessDenied, "Access Denied"));
     }
 

rustfs/src/admin/handlers/table_catalog/mod.rs

@@ -15,7 +15,7 @@
 use crate::admin::operation::{AdminRequest, AdminResponse};
 use crate::admin::runtime_sources;
 use crate::admin::runtime_sources::default_admin_usecase;
-use crate::admin::storage_api::access::{ReqInfo, authorize_internal_object_request};
+use crate::admin::storage_api::access::{ReqInfo, authorize_owned_internal_object_request};
 use crate::admin::storage_api::bucket::metadata::table_catalog_path_hash;
 use crate::admin::storage_api::runtime::ECStore;
 use crate::admin::utils::owned_empty_response;
@@ -1088,12 +1088,8 @@ fn build_sensitive_json_response<T: Serialize>(status: StatusCode, body: &T) ->
     Ok(AdminResponse::with_headers(status, body, headers))
 }
 
-// The commit-path handlers (`TableCommitObjectBackend::for_request`) authorize
-// exact object keys through `storage::access::authorize_internal_object_request`,
-// which still takes the s3s request. Until that path has an owned form they keep
-// the s3s request and read every other request fact through this owned view.
-// TODO(T1.12-codemod): remove once the generic object authorization path moves
-// to the request envelope.
+// These handlers still use the legacy router while their request bodies migrate.
+// TODO(T1.12-codemod): move them to the owned operation boundary.
 type LegacyRequest = s3s::S3Request<s3s::Body>;
 type LegacyResponse = s3s::S3Response<(StatusCode, s3s::Body)>;
 
@@ -1318,15 +1314,14 @@ fn install_table_catalog_s3_request_info(req: &mut LegacyRequest, principal: &Ta
 }
 
 async fn authorize_table_catalog_s3_actions(
-    req: &mut LegacyRequest,
+    req: &mut AdminRequest,
     bucket: &str,
     object: &str,
     actions: &[S3Action],
 ) -> S3Result<()> {
     let original = {
         let req_info = req
-            .extensions
-            .get_mut::<ReqInfo>()
+            .req_info_mut()
             .ok_or_else(|| s3_error!(AccessDenied, "authentication required"))?;
         (
             req_info.bucket.replace(bucket.to_string()),
@@ -1337,15 +1332,15 @@ async fn authorize_table_catalog_s3_actions(
 
     let mut result = Ok(());
     for action in actions {
-        if let Err(err) = authorize_internal_object_request(req, Action::S3Action(*action)).await {
+        if let Err(err) = authorize_owned_internal_object_request(req.envelope_mut(), Action::S3Action(*action)).await {
             result = Err(err);
             break;
         }
     }
-    if let Some(req_info) = req.extensions.get_mut::<ReqInfo>() {
+    if let Some(req_info) = req.req_info_mut() {
         (req_info.bucket, req_info.object, req_info.version_id) = original;
     }
-    result.map_err(Into::into)
+    result
 }
 
 async fn table_catalog_request_principal(req: &AdminRequest) -> S3Result<TableCatalogRequestPrincipal> {
@@ -1383,7 +1378,7 @@ async fn table_catalog_request_principal(req: &AdminRequest) -> S3Result<TableCa
 
 #[derive(Clone)]
 enum TableCommitObjectAuthorization {
-    Request(Arc<tokio::sync::Mutex<LegacyRequest>>),
+    Request(Arc<tokio::sync::Mutex<AdminRequest>>),
     Preauthorized,
     #[cfg(test)]
     Test {
@@ -1459,7 +1454,7 @@ where
     fn for_request(backend: B, req: LegacyRequest) -> Self {
         Self::new(
             backend,
-            TableCommitObjectAuthorization::Request(Arc::new(tokio::sync::Mutex::new(req))),
+            TableCommitObjectAuthorization::Request(Arc::new(tokio::sync::Mutex::new(legacy_request_view(&req)))),
             rustfs_utils::get_env_bool(crate::table_catalog::ENV_TABLE_CATALOG_PUBLICATION_FENCE_FLEET_CONFIRMED, false),
         )
     }
@@ -2093,7 +2088,7 @@ where
         let result = match &self.authorization {
             TableCommitObjectAuthorization::Request(req) => {
                 let req = req.lock().await;
-                authorize_table_warehouse_location_request(&legacy_request_view(&req), table_bucket, location).await
+                authorize_table_warehouse_location_request(&req, table_bucket, location).await
             }
             TableCommitObjectAuthorization::Preauthorized => Ok(()),
             #[cfg(test)]

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -77,8 +77,7 @@ fn table_catalog_handler_request(context: Arc<AppContext>, access_key: &str, sec
     load_table_request_builder(context, access_key, secret_key).build()
 }
 
-/// The s3s form of [`table_catalog_handler_request`] for the commit backend,
-/// which still holds the legacy request (see `legacy_request_view`).
+/// Builds the legacy request accepted by the commit backend, which derives an owned view.
 // TODO(T1.12-codemod): remove with `TableCommitObjectBackend::for_request`'s s3s parameter.
 fn legacy_table_catalog_handler_request(
     context: Arc<AppContext>,
@@ -1295,6 +1294,10 @@ async fn table_storage_credentials_are_rejected_by_catalog_authentication() {
 #[tokio::test]
 async fn warehouse_location_claim_checks_request_identity_and_relocation() {
     let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    store
+        .make_bucket("warehouse", &MakeBucketOptions::default())
+        .await
+        .expect("create warehouse bucket");
     rustfs_iam::store::object::ObjectStore::new(store.clone())
         .save_iam_config(
             serde_json::json!({"version": 1}),
@@ -1332,6 +1335,18 @@ async fn warehouse_location_claim_checks_request_identity_and_relocation() {
     iam.policy_db_set(access_key, UserType::Reg, false, "location-creator")
         .await
         .expect("attach location policy");
+    iam.set_policy(
+        "owned-object-policy",
+        Policy::parse_config(
+            br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::warehouse/owned/*"},{"Effect":"Allow","Action":"s3:PutObject","Resource":"arn:aws:s3:::warehouse/owned/writable"}]}"#,
+        )
+        .expect("parse object policy"),
+    )
+    .await
+    .expect("store object policy");
+    iam.policy_db_set(access_key, UserType::Reg, false, "location-creator,owned-object-policy")
+        .await
+        .expect("attach object policy");
     let context = Arc::new(AppContext::new(store, Arc::new(RequestIam { handle: iam }), Arc::new(RequestKms)));
     assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
         access_key: "location-root".to_string(),
@@ -1339,6 +1354,125 @@ async fn warehouse_location_claim_checks_request_identity_and_relocation() {
         status: "on".to_string(),
         ..Default::default()
     }));
+    // Exercise the actual owned internal authorization path, including its temporary target.
+    let mut legacy = legacy_table_catalog_handler_request(context.clone(), access_key, secret_key);
+    let principal = table_catalog_request_principal(&legacy_request_view(&legacy))
+        .await
+        .expect("resolve principal");
+    install_table_catalog_s3_request_info(&mut legacy, &principal).expect("install authorization context");
+    let info = legacy.extensions.get_mut::<ReqInfo>().expect("request info");
+    info.bucket = Some("original-bucket".to_owned());
+    info.object = Some("original-object".to_owned());
+    info.version_id = Some("original-version".to_owned());
+    let mut owned = legacy_request_view(&legacy);
+    for (object, allowed) in [("owned/allowed", true), ("shared/denied", false)] {
+        let result = authorize_table_catalog_s3_actions(&mut owned, "warehouse", object, &[S3Action::GetObjectAction]).await;
+        if allowed {
+            result.expect("the exact owned object should be allowed");
+        } else {
+            assert_eq!(result.expect_err("the other object should be denied").code(), &S3ErrorCode::AccessDenied);
+        }
+        let info = owned.envelope().req_info().expect("request info remains installed");
+        assert_eq!(info.bucket.as_deref(), Some("original-bucket"));
+        assert_eq!(info.object.as_deref(), Some("original-object"));
+        assert_eq!(info.version_id.as_deref(), Some("original-version"));
+    }
+    // The exact-key check selects internal-object mode: a request header the object
+    // write does not apply (a malformed x-amz-tagging) must not reach the decision,
+    // while the ordinary owned entry point still evaluates it and rejects it.
+    let mut tagged = legacy_table_catalog_handler_request(context.clone(), access_key, secret_key);
+    install_table_catalog_s3_request_info(&mut tagged, &principal).expect("install authorization context");
+    tagged
+        .headers
+        .insert("x-amz-tagging", http::HeaderValue::from_static("classification=%ZZ"));
+    let mut tagged = legacy_request_view(&tagged);
+    authorize_table_catalog_s3_actions(&mut tagged, "warehouse", "owned/writable", &[S3Action::PutObjectAction])
+        .await
+        .expect("internal-object mode ignores request options the write does not apply");
+    let info = tagged.req_info_mut().expect("request info");
+    info.bucket = Some("warehouse".to_owned());
+    info.object = Some("owned/writable".to_owned());
+    crate::app::authorize_owned_request(
+        tagged.envelope_mut(),
+        rustfs_policy::policy::action::Action::S3Action(S3Action::PutObjectAction),
+        None,
+    )
+    .await
+    .expect_err("ordinary mode evaluates the malformed tagging header");
+    let objects = TestTableCatalogObjectBackend::default();
+    objects
+        .seed_object("warehouse", "owned/read-only", b"readable".to_vec())
+        .await;
+    objects.seed_object("warehouse", "shared/denied", b"unchanged".to_vec()).await;
+    let backend = || {
+        let mut request = legacy_table_catalog_handler_request(context.clone(), access_key, secret_key);
+        install_table_catalog_s3_request_info(&mut request, &principal).expect("install authorization context");
+        TableCommitObjectBackend::for_request(objects.clone(), request)
+    };
+    let allowed = backend();
+    let object = allowed
+        .read_object("warehouse", "owned/read-only")
+        .await
+        .expect("authorized read")
+        .expect("seeded object");
+    assert_eq!(object.data, b"readable");
+    allowed
+        .put_object(
+            "warehouse",
+            "owned/writable",
+            b"written".to_vec(),
+            crate::table_catalog::TableCatalogPutPrecondition::Any,
+        )
+        .await
+        .expect("authorized write");
+    assert_eq!(
+        objects
+            .read_object("warehouse", "owned/writable")
+            .await
+            .expect("read stored object")
+            .expect("written object")
+            .data,
+        b"written"
+    );
+    for (object, write) in [("shared/denied", false), ("owned/read-only", true), ("shared/denied", true)] {
+        let denied = backend();
+        let result = if write {
+            denied
+                .put_object(
+                    "warehouse",
+                    object,
+                    b"modified".to_vec(),
+                    crate::table_catalog::TableCatalogPutPrecondition::Any,
+                )
+                .await
+                .map(|_| ())
+        } else {
+            denied.read_object("warehouse", object).await.map(|_| ())
+        };
+        assert!(result.is_err(), "unauthorized {object} operation must fail");
+        assert_eq!(
+            denied.finish(Ok(())).await.expect_err("retain authorization error").code(),
+            &S3ErrorCode::AccessDenied
+        );
+    }
+    assert_eq!(
+        objects
+            .read_object("warehouse", "owned/read-only")
+            .await
+            .expect("read object")
+            .expect("seeded object")
+            .data,
+        b"readable"
+    );
+    assert_eq!(
+        objects
+            .read_object("warehouse", "shared/denied")
+            .await
+            .expect("read object")
+            .expect("seeded object")
+            .data,
+        b"unchanged"
+    );
     let request = || table_catalog_handler_request(context.clone(), access_key, secret_key);
     for location in ["s3://warehouse/owned", "s3://warehouse/owned/", "s3://warehouse/owned/child"] {
         authorize_table_warehouse_location_request(&request(), "warehouse", location)

rustfs/src/admin/handlers/user.rs

@@ -243,7 +243,10 @@ impl Operation for AddUser {
         let (cred, owner) =
             check_key_valid(get_session_token(req.uri(), req.headers()).unwrap_or_default(), input_cred.access_key()).await?;
         // The caller's current secret outlives the body read below.
-        let input_secret = input_cred.secret_key().to_owned();
+        let input_secret = input_cred
+            .require_secret_key()
+            .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))?
+            .to_owned();
 
         let ak = query.access_key.as_deref().unwrap_or_default();
 

rustfs/src/admin/operation.rs

@@ -210,11 +210,10 @@ pub(crate) mod test_support {
         /// token is read from the request like the s3s edge does, so a test
         /// that sets `x-amz-security-token` keeps working unchanged.
         pub(crate) fn credentials(mut self, access_key: &str, secret_key: &str, session_token: Option<&str>) -> Self {
-            self.credentials = Some(RequestCredentials::new(
-                access_key.to_owned(),
-                secret_key.to_owned(),
-                session_token.map(str::to_owned),
-            ));
+            self.credentials = Some(
+                RequestCredentials::new(access_key.to_owned(), secret_key.to_owned(), session_token.map(str::to_owned))
+                    .expect("test credentials carry a non-empty secret"),
+            );
             self
         }
 
@@ -246,11 +245,7 @@ pub(crate) mod test_support {
         pub(crate) fn build(self) -> AdminRequest {
             let credentials = self.credentials.map(|credentials| match credentials.session_token() {
                 Some(_) => credentials,
-                None => RequestCredentials::new(
-                    credentials.access_key().to_owned(),
-                    credentials.secret_key().to_owned(),
-                    get_session_token(&self.uri, &self.headers).map(str::to_owned),
-                ),
+                None => credentials.with_session_token(get_session_token(&self.uri, &self.headers).map(str::to_owned)),
             });
             let head = RequestHead {
                 method: self.method,
@@ -298,7 +293,7 @@ mod tests {
             .build();
         let credentials = req.credentials().expect("credentials were supplied");
         assert_eq!(credentials.access_key(), "AKIATEST");
-        assert_eq!(credentials.secret_key(), "secret");
+        assert_eq!(credentials.secret_key(), Some("secret"));
         assert_eq!(credentials.session_token(), Some("from-header"));
 
         let explicit = TestRequest::new(Method::GET, "/rustfs/admin/v3/info")

rustfs/src/admin/router.rs

@@ -3882,7 +3882,7 @@ mod tests {
                 .credentials()
                 .ok_or_else(|| OwnedS3Error::with_message(OwnedS3ErrorCode::AccessDenied, "owned admin identity is required"))?;
             let identity_matches = credentials.access_key() == "AKIAOWNED"
-                && constant_time_eq(credentials.secret_key(), "owned-secret")
+                && credentials.secret_key_matches("owned-secret")
                 && credentials
                     .session_token()
                     .is_some_and(|token| constant_time_eq(token, "owned-session"));

rustfs/src/admin/storage_api.rs

@@ -953,9 +953,8 @@ pub(crate) mod data_usage {
 }
 
 pub(crate) mod access {
-    pub(crate) use crate::storage::storage_api::access_consumer::{
-        ReqInfo, authorize_internal_object_request, authorize_request,
-    };
+    pub(crate) use crate::app::{authorize_owned_internal_object_request, authorize_owned_request};
+    pub(crate) use crate::storage::storage_api::access_consumer::{ReqInfo, authorize_request};
     pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced};
 }
 

rustfs/src/app/context.rs

@@ -26,6 +26,7 @@ mod startup;
 
 pub(crate) use envelope::RequestHead;
 pub use envelope::{RequestCredentials, RequestEnvelope};
+pub(crate) use envelope::{authorize_owned_internal_object_request, authorize_owned_request};
 pub use global::*;
 pub use handles::*;
 pub use interfaces::*;

rustfs/src/app/context/envelope.rs

@@ -29,13 +29,19 @@ use super::ServerContextSlot;
 use crate::app::metadata_route::WebsiteRead;
 use crate::app::object::request_body::BodyReadControl;
 use crate::app::storage_api::context::{
-    BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, ObjectTagConditions,
-    OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo, RequestContext,
+    AuditRequestView, AuthorizationRequest, DenialAuditor, DenialScope, authorize_request_core,
+};
+use crate::app::storage_api::context::{
+    BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, MultipartStagingOperation,
+    ObjectTagConditions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectPolicyHeaders,
+    PostObjectRequestMarker, ReqInfo, RequestContext, RequestedCopySource, RequestedObjectLegalHold,
+    RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags, StagedMultipartPartAuthorization,
     TableDataPlaneListAccess, TableDataPlanePublicationGuards,
 };
 use crate::auth::{VerifiedPresignedRequest, VerifiedSigV4Request};
 use crate::shared_types::{NativeTlsTransport, RemoteAddr};
 use http::{Extensions, HeaderMap, Method, Uri};
+use rustfs_policy::policy::action::{Action, S3Action};
 use rustfs_trusted_proxies::ClientInfo;
 use std::fmt;
 use std::net::SocketAddr;
@@ -45,42 +51,109 @@ use zeroize::Zeroizing;
 /// The identity a request was authenticated with.
 ///
 /// The secret key and the session token are bearer material: the type has no
-/// `PartialEq`, and its `Debug` output redacts both.
+/// `PartialEq`, and its `Debug` output redacts both. An edge that keeps the
+/// caller's secret (it verified the signature itself) builds the value with
+/// [`Self::without_secret_key`]; the absence is then `None`, never a
+/// substitute string a reader could compare, seal or derive with.
 #[derive(Clone)]
 pub struct RequestCredentials {
     access_key: String,
-    secret_key: Zeroizing<String>,
+    secret_key: Option<Zeroizing<String>>,
     session_token: Option<Zeroizing<String>>,
 }
 
+/// An edge handed over an empty secret key. Absence is expressed with
+/// [`RequestCredentials::without_secret_key`], so an empty secret is refused
+/// rather than read as "absent" or used as a key.
+#[derive(Debug)]
+pub(crate) struct EmptySecretKey;
+
+impl fmt::Display for EmptySecretKey {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("handed-over secret key is empty")
+    }
+}
+
+impl std::error::Error for EmptySecretKey {}
+
 impl RequestCredentials {
-    pub(crate) fn new(access_key: String, secret_key: String, session_token: Option<String>) -> Self {
+    /// Credentials whose secret the edge handed over.
+    ///
+    /// # Errors
+    ///
+    /// [`EmptySecretKey`] when `secret_key` is empty.
+    pub(crate) fn new(access_key: String, secret_key: String, session_token: Option<String>) -> Result<Self, EmptySecretKey> {
+        if secret_key.is_empty() {
+            return Err(EmptySecretKey);
+        }
+        Ok(Self {
+            access_key,
+            secret_key: Some(Zeroizing::new(secret_key)),
+            session_token: session_token.map(Zeroizing::new),
+        })
+    }
+
+    /// Credentials whose secret the edge kept.
+    pub(crate) fn without_secret_key(access_key: String, session_token: Option<String>) -> Self {
         Self {
             access_key,
-            secret_key: Zeroizing::new(secret_key),
+            secret_key: None,
             session_token: session_token.map(Zeroizing::new),
         }
     }
 
+    /// The same identity with `session_token` as its token.
+    #[cfg(test)]
+    pub(crate) fn with_session_token(mut self, session_token: Option<String>) -> Self {
+        self.session_token = session_token.map(Zeroizing::new);
+        self
+    }
+
     pub fn access_key(&self) -> &str {
         &self.access_key
     }
 
-    /// Never log this value or compare it with `==`.
-    pub fn secret_key(&self) -> &str {
-        &self.secret_key
+    /// The caller's secret, or `None` when the edge did not hand it over.
+    /// A reader must fail closed on `None`. Never log this value or compare it
+    /// with `==`.
+    pub fn secret_key(&self) -> Option<&str> {
+        self.secret_key.as_deref().map(String::as_str)
     }
 
     pub fn session_token(&self) -> Option<&str> {
         self.session_token.as_deref().map(String::as_str)
     }
+
+    /// The handed-over secret, or [`MissingSecretKey`] when the edge kept it.
+    pub(crate) fn require_secret_key(&self) -> Result<&str, MissingSecretKey> {
+        self.secret_key().ok_or(MissingSecretKey)
+    }
+
+    /// Whether the handed-over secret equals `stored`, compared in constant
+    /// time. An absent secret matches nothing, not even an empty `stored`.
+    pub(crate) fn secret_key_matches(&self, stored: &str) -> bool {
+        self.secret_key()
+            .is_some_and(|secret| crate::auth::constant_time_eq(secret, stored))
+    }
 }
 
+/// A reader needed the caller's secret, but the edge did not hand it over.
+#[derive(Debug)]
+pub(crate) struct MissingSecretKey;
+
+impl fmt::Display for MissingSecretKey {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("the caller's secret key was not handed over for this request")
+    }
+}
+
+impl std::error::Error for MissingSecretKey {}
+
 impl fmt::Debug for RequestCredentials {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("RequestCredentials")
             .field("access_key", &self.access_key)
-            .field("secret_key", &"<redacted>")
+            .field("secret_key", &self.secret_key.as_ref().map(|_| "<redacted>"))
             .field("session_token", &self.session_token.as_ref().map(|_| "<redacted>"))
             .finish()
     }
@@ -128,6 +201,7 @@ pub struct RequestEnvelope {
     verified_presigned: Option<VerifiedPresignedRequest>,
     website_read: Option<WebsiteRead>,
     post_object: Option<PostObjectRequestMarker>,
+    post_object_policy_headers: Option<PostObjectPolicyHeaders>,
     // Attached by the authorization phase (`storage/access.rs`) for the use case.
     req_info: Option<ReqInfo>,
     bucket_generation: Option<BucketGenerationGuard>,
@@ -138,6 +212,13 @@ pub struct RequestEnvelope {
     table_publication_guards: Option<TableDataPlanePublicationGuards>,
     table_list_access: Option<TableDataPlaneListAccess>,
     object_tag_conditions: Option<ObjectTagConditions>,
+    multipart_staging_operation: Option<MultipartStagingOperation>,
+    staged_multipart_part_authorization: Option<StagedMultipartPartAuthorization>,
+    requested_object_lock_retention: Option<RequestedObjectLockRetention>,
+    requested_object_lock_retain_until: Option<RequestedObjectLockRetainUntil>,
+    requested_object_legal_hold: Option<RequestedObjectLegalHold>,
+    requested_object_tags: Option<RequestedObjectTags>,
+    requested_copy_source: Option<RequestedCopySource>,
     residual_extensions: Extensions,
 }
 
@@ -167,6 +248,7 @@ impl RequestEnvelope {
             verified_presigned: extensions.remove(),
             website_read: extensions.remove(),
             post_object: extensions.remove(),
+            post_object_policy_headers: extensions.remove(),
             req_info: extensions.remove(),
             bucket_generation: extensions.remove(),
             pending_delete_bucket_generation: extensions.remove(),
@@ -176,6 +258,13 @@ impl RequestEnvelope {
             table_publication_guards: extensions.remove(),
             table_list_access: extensions.remove(),
             object_tag_conditions: extensions.remove(),
+            multipart_staging_operation: extensions.remove(),
+            staged_multipart_part_authorization: extensions.remove(),
+            requested_object_lock_retention: extensions.remove(),
+            requested_object_lock_retain_until: extensions.remove(),
+            requested_object_legal_hold: extensions.remove(),
+            requested_object_tags: extensions.remove(),
+            requested_copy_source: extensions.remove(),
             residual_extensions: extensions,
         }
     }
@@ -312,6 +401,10 @@ impl RequestEnvelope {
         self.table_publication_guards.as_ref()
     }
 
+    pub(crate) fn install_table_publication_guards(&mut self, guards: TableDataPlanePublicationGuards) {
+        self.table_publication_guards = Some(guards);
+    }
+
     pub(crate) fn table_list_access(&self) -> Option<&TableDataPlaneListAccess> {
         self.table_list_access.as_ref()
     }
@@ -319,19 +412,337 @@ impl RequestEnvelope {
     pub(crate) fn object_tag_conditions(&self) -> Option<&ObjectTagConditions> {
         self.object_tag_conditions.as_ref()
     }
+
+    pub(crate) fn install_object_tag_conditions(&mut self, conditions: ObjectTagConditions) {
+        self.object_tag_conditions = Some(conditions);
+    }
+
+    pub(crate) fn multipart_staging_operation(&self) -> Option<&MultipartStagingOperation> {
+        self.multipart_staging_operation.as_ref()
+    }
+
+    pub(crate) fn staged_multipart_part_authorization(&self) -> Option<&StagedMultipartPartAuthorization> {
+        self.staged_multipart_part_authorization.as_ref()
+    }
+
+    pub(crate) fn requested_object_lock_retention(&self) -> Option<&RequestedObjectLockRetention> {
+        self.requested_object_lock_retention.as_ref()
+    }
+
+    pub(crate) fn requested_object_lock_retain_until(&self) -> Option<&RequestedObjectLockRetainUntil> {
+        self.requested_object_lock_retain_until.as_ref()
+    }
+
+    pub(crate) fn post_object_policy_headers(&self) -> Option<&PostObjectPolicyHeaders> {
+        self.post_object_policy_headers.as_ref()
+    }
+
+    pub(crate) fn requested_object_legal_hold(&self) -> Option<&RequestedObjectLegalHold> {
+        self.requested_object_legal_hold.as_ref()
+    }
+
+    pub(crate) fn requested_object_tags(&self) -> Option<&RequestedObjectTags> {
+        self.requested_object_tags.as_ref()
+    }
+
+    pub(crate) fn requested_copy_source(&self) -> Option<&RequestedCopySource> {
+        self.requested_copy_source.as_ref()
+    }
+}
+
+/// The authorization core in `storage` reads requests only through
+/// [`AuthorizationRequest`]; this impl hands it the envelope's named fields.
+impl AuthorizationRequest for RequestEnvelope {
+    fn headers(&self) -> &HeaderMap {
+        self.headers()
+    }
+    fn uri(&self) -> &http::Uri {
+        self.uri()
+    }
+    fn remote_addr(&self) -> Option<std::net::SocketAddr> {
+        self.remote_addr()
+    }
+    fn client_info(&self) -> Option<&ClientInfo> {
+        self.client_info()
+    }
+    fn server_context(&self) -> Option<&Arc<ServerContextSlot>> {
+        self.server_context()
+    }
+    fn req_info(&self) -> Option<&ReqInfo> {
+        self.req_info()
+    }
+    fn internal_object_authorization(&self) -> bool {
+        false
+    }
+    fn multipart_staging(&self) -> bool {
+        self.multipart_staging_operation().is_some()
+    }
+    fn staged_multipart_part(&self) -> bool {
+        self.staged_multipart_part_authorization().is_some()
+    }
+    fn requested_retention(&self) -> Option<&RequestedObjectLockRetention> {
+        self.requested_object_lock_retention()
+    }
+    fn requested_retain_until(&self) -> Option<&RequestedObjectLockRetainUntil> {
+        self.requested_object_lock_retain_until()
+    }
+    fn requested_legal_hold(&self) -> Option<&RequestedObjectLegalHold> {
+        self.requested_object_legal_hold()
+    }
+    fn requested_object_tags(&self) -> Option<&RequestedObjectTags> {
+        self.requested_object_tags()
+    }
+    fn requested_copy_source(&self) -> Option<&RequestedCopySource> {
+        self.requested_copy_source()
+    }
+    fn policy_headers(&self) -> &HeaderMap {
+        self.post_object_policy_headers().map_or(self.headers(), |headers| &headers.0)
+    }
+    fn object_tag_conditions(&self) -> Option<&ObjectTagConditions> {
+        self.object_tag_conditions()
+    }
+    fn install_object_tag_conditions(&mut self, conditions: ObjectTagConditions) {
+        self.install_object_tag_conditions(conditions);
+    }
+    fn publication_guards(&self) -> Option<&TableDataPlanePublicationGuards> {
+        self.table_publication_guards()
+    }
+    fn install_publication_guards(&mut self, guards: TableDataPlanePublicationGuards) {
+        self.install_table_publication_guards(guards);
+    }
+    fn audit_denial(&self, auditor: &DenialAuditor, action: &Action, reason: &'static str) {
+        auditor.audit_owned(&self.audit_view(), action, reason, DenialScope::Request);
+    }
+}
+
+/// Authorizes an owned request through the same IAM, bucket and table policies as the legacy edge.
+///
+/// Like the legacy `authorize_request`, it first restores the signed `x-amz-meta-*`
+/// query parameters into headers, so policy conditions and duplicate rejection see
+/// the same request shape on both edges.
+///
+/// This is only the policy decision. It does **not** perform the composite checks
+/// the legacy `S3Access` hooks run around their decision, such as the
+/// `s3:BypassGovernanceRetention` authorization for a governance-bypass header or
+/// the copy-source read authorization; a caller that needs them runs them itself.
+///
+/// # Errors
+///
+/// `InvalidArgument` when the signed query repeats a user metadata key;
+/// `AccessDenied` when the decision for `action` depends on a request-body input
+/// the envelope does not carry (see [`require_body_policy_inputs`]) or when the
+/// policies deny the request.
+pub(crate) async fn authorize_owned_request(
+    req: &mut RequestEnvelope,
+    action: Action,
+    location_constraint: Option<String>,
+) -> rustfs_s3_types::S3Result<()> {
+    prepare_owned_authorization(req, action)?;
+    authorize_request_core(req, action, location_constraint, false)
+        .await
+        .map_err(Into::into)
+}
+
+/// Authorizes one exact object key after its enclosing table-catalog operation has been authorized.
+///
+/// It applies the same preparation and has the same limits as [`authorize_owned_request`].
+pub(crate) async fn authorize_owned_internal_object_request(
+    req: &mut RequestEnvelope,
+    action: Action,
+) -> rustfs_s3_types::S3Result<()> {
+    prepare_owned_authorization(req, action)?;
+    authorize_request_core(req, action, None, true).await.map_err(Into::into)
+}
+
+fn prepare_owned_authorization(req: &mut RequestEnvelope, action: Action) -> rustfs_s3_types::S3Result<()> {
+    crate::auth::hoist_signed_query_params(&mut req.headers, req.uri.query()).map_err(rustfs_s3_types::S3Error::from)?;
+    require_body_policy_inputs(req, action)
+}
+
+/// Fails closed when the decision for `action` depends on a value the request
+/// body supplies but the envelope does not carry it. Policy conditions would
+/// otherwise fall back to client headers, which the stored value need not match
+/// (header `OFF` with body `ON` would pass a legal-hold Deny).
+fn require_body_policy_inputs(req: &RequestEnvelope, action: Action) -> rustfs_s3_types::S3Result<()> {
+    let missing = match action {
+        Action::S3Action(S3Action::PutObjectLegalHoldAction) => req.requested_object_legal_hold.is_none(),
+        Action::S3Action(S3Action::PutObjectTaggingAction | S3Action::PutObjectVersionTaggingAction) => {
+            req.requested_object_tags.is_none()
+        }
+        Action::S3Action(S3Action::PutObjectRetentionAction) => req.requested_object_lock_retention.is_none(),
+        _ => false,
+    } || (req.post_object.is_some() && req.post_object_policy_headers.is_none())
+        || (req.headers.contains_key(COPY_SOURCE_HEADER) && req.requested_copy_source.is_none());
+    if missing {
+        return Err(rustfs_s3_types::s3_error!(
+            AccessDenied,
+            "request body policy input is missing for this action"
+        ));
+    }
+    Ok(())
+}
+
+const COPY_SOURCE_HEADER: &str = "x-amz-copy-source";
+
+impl RequestEnvelope {
+    /// The request fields an audit entry for this request is built from.
+    pub(crate) fn audit_view(&self) -> AuditRequestView<'_> {
+        AuditRequestView::from_facts(
+            self.headers(),
+            self.uri(),
+            self.req_info(),
+            self.client_info(),
+            self.remote_addr(),
+            self.request_context(),
+        )
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn authorization_payloads_move_from_extensions_into_named_fields() {
+        let mut extensions = Extensions::new();
+        extensions.insert(MultipartStagingOperation);
+        extensions.insert(StagedMultipartPartAuthorization);
+        let mut retention = RequestedObjectLockRetention(Default::default());
+        retention.0.mode = Some("COMPLIANCE".to_owned().into());
+        extensions.insert(retention);
+        let retain_until = time::OffsetDateTime::from_unix_timestamp(2_000_000_000).expect("valid timestamp");
+        extensions.insert(RequestedObjectLockRetainUntil(retain_until));
+        let head = || RequestHead {
+            method: Method::PUT,
+            uri: "/bucket/key".parse().expect("valid URI"),
+            headers: HeaderMap::new(),
+            credentials: None,
+            region: None,
+            service: None,
+        };
+
+        let envelope = RequestEnvelope::from_parts(head(), extensions);
+        let empty = RequestEnvelope::from_parts(head(), Extensions::new());
+
+        assert_eq!(
+            (
+                envelope.multipart_staging_operation().is_some(),
+                envelope.staged_multipart_part_authorization().is_some(),
+                envelope
+                    .requested_object_lock_retention()
+                    .and_then(|value| value.0.mode.as_ref().map(|mode| mode.as_str())),
+                envelope.requested_object_lock_retain_until().map(|value| value.0),
+                envelope.residual_extensions().is_empty(),
+                empty.multipart_staging_operation().is_none(),
+                empty.staged_multipart_part_authorization().is_none(),
+                empty.requested_object_lock_retention().is_none(),
+                empty.requested_object_lock_retain_until().is_none(),
+            ),
+            (true, true, Some("COMPLIANCE"), Some(retain_until), true, true, true, true, true),
+        );
+    }
+
+    #[test]
+    fn request_body_policy_payloads_move_from_extensions_into_named_fields() {
+        let mut form_headers = HeaderMap::new();
+        form_headers.insert("x-amz-tagging", http::HeaderValue::from_static("from=form"));
+        let mut legal_hold = RequestedObjectLegalHold(Default::default());
+        legal_hold.0.status = Some("ON".to_owned().into());
+        let mut tags = RequestedObjectTags(Default::default());
+        tags.0.tag_set.push(Default::default());
+        tags.0.tag_set[0].key = Some("body".to_owned());
+        let mut extensions = Extensions::new();
+        extensions.insert(PostObjectPolicyHeaders(form_headers));
+        extensions.insert(legal_hold);
+        extensions.insert(tags);
+        extensions.insert(RequestedCopySource("source-bucket/source-key".to_owned()));
+        let head = || RequestHead {
+            method: Method::PUT,
+            uri: "/bucket/key".parse().expect("valid URI"),
+            headers: HeaderMap::new(),
+            credentials: None,
+            region: None,
+            service: None,
+        };
+
+        let envelope = RequestEnvelope::from_parts(head(), extensions);
+        let empty = RequestEnvelope::from_parts(head(), Extensions::new());
+
+        assert_eq!(
+            (
+                envelope
+                    .post_object_policy_headers()
+                    .and_then(|headers| headers.0.get("x-amz-tagging"))
+                    .and_then(|value| value.to_str().ok()),
+                envelope
+                    .requested_object_legal_hold()
+                    .and_then(|hold| hold.0.status.as_ref().map(|status| status.as_str())),
+                envelope
+                    .requested_object_tags()
+                    .and_then(|tags| tags.0.tag_set.first())
+                    .and_then(|tag| tag.key.as_deref()),
+                envelope.requested_copy_source().map(|source| source.0.as_str()),
+                envelope.residual_extensions().is_empty(),
+                empty.post_object_policy_headers().is_none(),
+                empty.requested_object_legal_hold().is_none(),
+                empty.requested_object_tags().is_none(),
+                empty.requested_copy_source().is_none(),
+            ),
+            (
+                Some("from=form"),
+                Some("ON"),
+                Some("body"),
+                Some("source-bucket/source-key"),
+                true,
+                true,
+                true,
+                true,
+                true
+            ),
+        );
+    }
+
+    #[test]
+    fn an_empty_handed_over_secret_is_refused() {
+        assert!(RequestCredentials::new("AKIAENVELOPE".to_owned(), String::new(), None).is_err());
+    }
+
+    #[test]
+    fn an_absent_secret_matches_no_stored_secret() {
+        let kept = RequestCredentials::without_secret_key("AKIAENVELOPE".to_owned(), None);
+        assert_eq!(kept.secret_key(), None);
+        assert!(!kept.secret_key_matches(""), "absent must not equal an empty stored secret");
+        assert!(!kept.secret_key_matches("stored-secret"));
+        let handed = RequestCredentials::new("AKIAENVELOPE".to_owned(), "stored-secret".to_owned(), None).expect("secret");
+        assert!(!handed.secret_key_matches(""));
+        assert!(!handed.secret_key_matches("other-secret"));
+        assert!(handed.secret_key_matches("stored-secret"));
+    }
+
+    #[test]
+    fn a_reader_of_an_absent_secret_fails_closed() {
+        let kept = RequestCredentials::without_secret_key("AKIAENVELOPE".to_owned(), Some("token".to_owned()));
+        assert!(kept.require_secret_key().is_err());
+        let handed = RequestCredentials::new("AKIAENVELOPE".to_owned(), "s".to_owned(), None).expect("secret");
+        assert_eq!(handed.require_secret_key().ok(), Some("s"));
+    }
+
+    #[test]
+    fn kept_secret_debug_prints_neither_secret_nor_token() {
+        let kept = RequestCredentials::without_secret_key("AKIAENVELOPE".to_owned(), Some("envelope-session-token".to_owned()));
+        let debug = format!("{kept:?}");
+        assert!(debug.contains("AKIAENVELOPE"), "{debug}");
+        assert!(!debug.contains("envelope-session-token"), "{debug}");
+    }
+
     #[test]
     fn request_credentials_debug_redacts_secret_and_session_token() {
         let credentials = RequestCredentials::new(
             "AKIAENVELOPE".to_owned(),
             "envelope-secret-value".to_owned(),
             Some("envelope-session-token".to_owned()),
-        );
+        )
+        .expect("non-empty secret");
 
         let debug = format!("{credentials:?}");
 

rustfs/src/app/mod.rs

@@ -36,6 +36,7 @@ pub(crate) mod trailer_adapter;
 // from here rather than through `app::context`.
 pub(crate) use context::RequestHead;
 pub use context::{RequestCredentials, RequestEnvelope};
+pub(crate) use context::{authorize_owned_internal_object_request, authorize_owned_request};
 
 #[cfg(test)]
 mod capacity_dirty_scope_test;
@@ -48,6 +49,8 @@ pub(crate) mod gating_test_env;
 #[cfg(test)]
 mod lifecycle_transition_api_test;
 #[cfg(test)]
+mod owned_authorization_test;
+#[cfg(test)]
 mod put_prelookup_gating_test;
 #[cfg(test)]
 pub(crate) mod test_seams;

rustfs/src/app/storage_api.rs

@@ -53,10 +53,17 @@ pub(crate) mod s3 {
         PersistedReplicationFilter, PersistedReplicationRule, PersistedReplicationTag,
     };
     #[cfg(test)]
+    pub(crate) use s3s::access::S3Access;
+    #[cfg(test)]
     pub(crate) use s3s::auth::{Credentials, SimpleAuth};
     #[cfg(test)]
     pub(crate) use s3s::config::{S3Config, StaticConfigProvider};
     #[cfg(test)]
+    pub(crate) use s3s::dto::{
+        BucketLocationConstraint, CreateBucketConfiguration, CreateBucketInput, CreateMultipartUploadInput, DeleteObjectInput,
+        ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, Tag, Tagging,
+    };
+    #[cfg(test)]
     pub(crate) use s3s::dto::{GetObjectInput, HeadObjectInput, ListObjectsV2Input, ListObjectsV2Output};
     #[cfg(test)]
     pub(crate) use s3s::dto::{GetObjectOutput, HeadObjectOutput, PutObjectInput, PutObjectOutput};
@@ -1817,14 +1824,30 @@ pub(crate) mod select_object {
     };
 }
 
+/// The storage items the owned/legacy authorization parity tests drive.
+#[cfg(test)]
+pub(crate) mod owned_parity {
+    pub(crate) use crate::storage::storage_api::owned_parity_consumer::{
+        DenialAuditor, DenialRateLimiter, DenialScope, FS, ObjectTagConditions, PostObjectPolicyHeaders, PostObjectRequestMarker,
+        ReqInfo, RequestContext, RequestedCopySource, RequestedObjectLegalHold, RequestedObjectLockRetainUntil,
+        RequestedObjectLockRetention, RequestedObjectTags, SUPPRESSED_DENIALS_TAG, TableDataPlanePublicationGuards,
+        authorization_denial_entry, authorization_denial_entry_for_view, authorize_request, parity_test_support,
+    };
+}
+
 pub(crate) mod context {
     #[cfg(test)]
     pub(crate) use super::EndpointServerPools;
     pub(crate) use super::bucket;
     pub(crate) use super::runtime;
     pub(crate) use crate::storage::storage_api::access_consumer::{
-        BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, ObjectTagConditions,
-        OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo, TableDataPlaneListAccess,
+        AuditRequestView, AuthorizationRequest, DenialAuditor, DenialScope, authorize_request_core,
+    };
+    pub(crate) use crate::storage::storage_api::access_consumer::{
+        BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, MultipartStagingOperation,
+        ObjectTagConditions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectPolicyHeaders,
+        PostObjectRequestMarker, ReqInfo, RequestedCopySource, RequestedObjectLegalHold, RequestedObjectLockRetainUntil,
+        RequestedObjectLockRetention, RequestedObjectTags, StagedMultipartPartAuthorization, TableDataPlaneListAccess,
         TableDataPlanePublicationGuards,
     };
     pub(crate) use crate::storage::storage_api::request_context_consumer::RequestContext;

rustfs/src/storage/denial_audit.rs

@@ -20,8 +20,10 @@
 //! counted and the count is carried on the next entry that is let through.
 
 use crate::module_switches::is_audit_module_enabled;
-use crate::storage::access::{ReqInfo, request_context_from_req};
-use crate::storage::helper::{DenialRequest, DenialScope, authorization_denial_entry, spawn_background_with_context};
+use crate::storage::helper::{
+    AuditRequestView, DenialRequest, DenialScope, authorization_denial_entry, authorization_denial_entry_for_view,
+    spawn_background_with_context,
+};
 use crate::storage::request_context::RequestContext;
 use rustfs_audit::{AuditEntry, global::AuditLogger};
 use rustfs_policy::policy::action::Action;
@@ -184,21 +186,32 @@ impl DenialAuditor {
         scope: DenialScope,
         now: Instant,
     ) {
+        self.audit_view_at(&AuditRequestView::legacy(req), now, || {
+            authorization_denial_entry(req, action, reason, scope)
+        });
+    }
+
+    /// Audits a denial for a request an owned edge decoded into `view`.
+    pub(crate) fn audit_owned(&self, view: &AuditRequestView<'_>, action: &Action, reason: &'static str, scope: DenialScope) {
+        self.audit_view_at(view, (self.clock)(), || authorization_denial_entry_for_view(view, action, reason, scope));
+    }
+
+    fn audit_view_at(&self, req: &AuditRequestView<'_>, now: Instant, build_entry: impl FnOnce() -> AuditEntry) {
         if !(self.enabled)() {
             return;
         }
-        let anonymous = req.extensions.get::<ReqInfo>().is_none_or(|info| info.cred.is_none());
+        let anonymous = req.req_info.is_none_or(|info| info.cred.is_none());
         let Admission::Emit { suppressed } = self.limiter.admit(anonymous, now) else {
             return;
         };
-        let mut entry = authorization_denial_entry(req, action, reason, scope);
+        let mut entry = build_entry();
         if suppressed > 0 {
             entry
                 .tags
                 .get_or_insert_with(Default::default)
                 .insert(SUPPRESSED_DENIALS_TAG.to_string(), Value::from(suppressed));
         }
-        (self.emit)(request_context_from_req(req), entry);
+        (self.emit)(req.request_context(), entry);
     }
 }
 

rustfs/src/storage/helper.rs

@@ -19,7 +19,7 @@ use crate::storage::request_context::RequestContext;
 use crate::storage::sse::KmsRequestAuditScope;
 use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
 use hashbrown::HashMap;
-use http::StatusCode;
+use http::{HeaderMap, StatusCode, Uri};
 use metrics::counter;
 use rustfs_audit::{
     ApiDetails, AuditEntry, ObjectVersion,
@@ -38,12 +38,64 @@ use rustfs_utils::http::headers::AMZ_REQUEST_ID;
 use s3s::{S3Request, S3Response, S3Result};
 use serde_json::Value;
 use std::future::Future;
+use std::net::SocketAddr;
 use tokio::runtime::{Builder, Handle};
 
 /// Request type seen by the denial auditor, so `denial_audit` stays off the direct s3s surface.
 pub(crate) type DenialRequest<T> = S3Request<T>;
 use tracing::{Instrument, info_span, warn};
 
+/// The request fields used to build an audit entry, shared by both request edges.
+pub(crate) struct AuditRequestView<'a> {
+    headers: &'a HeaderMap,
+    uri: &'a Uri,
+    pub(crate) req_info: Option<&'a ReqInfo>,
+    client_info: Option<&'a ClientInfo>,
+    remote_addr: Option<SocketAddr>,
+    request_context: Option<&'a RequestContext>,
+}
+
+impl<'a> AuditRequestView<'a> {
+    pub(crate) fn legacy<T>(req: &'a S3Request<T>) -> Self {
+        Self {
+            headers: &req.headers,
+            uri: &req.uri,
+            req_info: req.extensions.get::<ReqInfo>(),
+            client_info: req.extensions.get::<ClientInfo>(),
+            remote_addr: req
+                .extensions
+                .get::<Option<RemoteAddr>>()
+                .and_then(|addr| addr.map(|addr| addr.0)),
+            request_context: req.extensions.get::<RequestContext>(),
+        }
+    }
+
+    /// The view of a request whose facts an owned edge already decoded.
+    pub(crate) fn from_facts(
+        headers: &'a HeaderMap,
+        uri: &'a Uri,
+        req_info: Option<&'a ReqInfo>,
+        client_info: Option<&'a ClientInfo>,
+        remote_addr: Option<SocketAddr>,
+        request_context: Option<&'a RequestContext>,
+    ) -> Self {
+        Self {
+            headers,
+            uri,
+            req_info,
+            client_info,
+            remote_addr,
+            request_context,
+        }
+    }
+
+    pub(crate) fn request_context(&self) -> Option<RequestContext> {
+        self.request_context
+            .cloned()
+            .or_else(|| self.req_info.and_then(|info| info.request_context.clone()))
+    }
+}
+
 #[cfg(feature = "diagnostic-read-version-caller")]
 use super::storage_api::ecstore_diagnostic;
 use crate::storage::storage_api::helper_consumer::StorageObjectInfo as ObjectInfo;
@@ -127,32 +179,27 @@ pub(crate) fn build_event_resp_elements<T>(response: &S3Response<T>, request_id:
 
 /// Builds the audit entry shared by every request-scoped audit source: caller identity,
 /// client address and the request line, without any outcome.
-fn request_audit_builder<T>(
-    req: &S3Request<T>,
-    req_info: Option<&ReqInfo>,
+fn request_audit_builder(
+    req: &AuditRequestView<'_>,
     event: EventName,
     trigger: &str,
     api: ApiDetails,
     request_id: &str,
 ) -> AuditEntryBuilder {
     let remote_host = req
-        .extensions
-        .get::<ClientInfo>()
+        .client_info
         .map(|info| info.real_ip.to_string())
-        .or_else(|| {
-            req.extensions
-                .get::<Option<RemoteAddr>>()
-                .and_then(|addr| addr.map(|addr| addr.0.ip().to_string()))
-        })
+        .or_else(|| req.remote_addr.map(|addr| addr.ip().to_string()))
         .unwrap_or_default();
     let mut builder = AuditEntryBuilder::new("1.0", event, trigger, api)
         .remote_host(remote_host)
-        .user_agent(get_request_user_agent(&req.headers))
-        .req_host(get_request_host(&req.headers))
+        .user_agent(get_request_user_agent(req.headers))
+        .req_host(get_request_host(req.headers))
         .req_path(req.uri.path().to_string())
-        .req_query(extract_params_header(&req.headers))
+        .req_query(extract_params_header(req.headers))
         .request_id(request_id);
-    if let Some(cred) = req_info
+    if let Some(cred) = req
+        .req_info
         .and_then(|info| info.cred.as_ref())
         .filter(|cred| !cred.access_key.is_empty())
     {
@@ -374,8 +421,9 @@ impl OperationHelper {
         let request_context = request_context.unwrap_or_else(|| RequestContext::from_external_headers(&req.headers));
         let request_id = request_context.request_id.clone();
 
-        let audit_builder =
-            audit_enabled.then(|| request_audit_builder(req, req_info, event, trigger, api_builder.clone().build(), &request_id));
+        let audit_builder = audit_enabled.then(|| {
+            request_audit_builder(&AuditRequestView::legacy(req), event, trigger, api_builder.clone().build(), &request_id)
+        });
 
         let event_object = ObjectInfo {
             bucket: bucket.clone(),
@@ -602,7 +650,16 @@ pub(crate) fn authorization_denial_entry<T>(
     reason: &'static str,
     scope: DenialScope,
 ) -> AuditEntry {
-    let req_info = req.extensions.get::<ReqInfo>();
+    authorization_denial_entry_for_view(&AuditRequestView::legacy(req), action, reason, scope)
+}
+
+pub(crate) fn authorization_denial_entry_for_view(
+    req: &AuditRequestView<'_>,
+    action: &Action,
+    reason: &'static str,
+    scope: DenialScope,
+) -> AuditEntry {
+    let req_info = req.req_info;
     let path = req.uri.path().trim_start_matches('/');
     let mut segs = path.splitn(2, '/');
     let path_bucket = segs.next().unwrap_or("");
@@ -631,23 +688,18 @@ pub(crate) fn authorization_denial_entry<T>(
         api = api.object(object);
     }
 
-    let request_context = request_context_from_req(req).unwrap_or_else(|| RequestContext::from_external_headers(&req.headers));
+    let request_context = req
+        .request_context()
+        .unwrap_or_else(|| RequestContext::from_external_headers(req.headers));
     let mut tags = HashMap::new();
     tags.insert("denialReason".to_string(), Value::String(reason.to_string()));
     if let DenialScope::Keys(denied) = scope {
         tags.insert("deniedKeys".to_string(), Value::from(denied));
     }
-    request_audit_builder(
-        req,
-        req_info,
-        EventName::AuditS3Operation,
-        action_name,
-        api.build(),
-        &request_context.request_id,
-    )
-    .error("Access Denied")
-    .tags(tags)
-    .build()
+    request_audit_builder(req, EventName::AuditS3Operation, action_name, api.build(), &request_context.request_id)
+        .error("Access Denied")
+        .tags(tags)
+        .build()
 }
 
 /// Runs `handler` on `req` and records the outcome as an audit-only entry for `op`.
@@ -1435,8 +1487,7 @@ mod tests {
                 ..Default::default()
             });
             let entry = super::request_audit_builder(
-                &req,
-                req.extensions.get::<ReqInfo>(),
+                &super::AuditRequestView::legacy(&req),
                 EventName::ObjectAccessedGet,
                 "GetObject",
                 rustfs_audit::ApiDetails::default(),

rustfs/src/storage/s3_compat/envelope.rs

@@ -25,11 +25,14 @@ impl RequestEnvelope {
     /// Copies every request fact of `req` the app layer reads.
     pub fn from_s3s<T>(req: &S3Request<T>) -> Self {
         let credentials = req.credentials.as_ref().map(|credentials| {
+            let session_token = get_session_token(&req.uri, &req.headers).map(str::to_owned);
+            // An empty secret is not a key: carry it as absent so every reader fails closed.
             RequestCredentials::new(
                 credentials.access_key.clone(),
                 credentials.secret_key.expose().to_owned(),
-                get_session_token(&req.uri, &req.headers).map(str::to_owned),
+                session_token.clone(),
             )
+            .unwrap_or_else(|_| RequestCredentials::without_secret_key(credentials.access_key.clone(), session_token))
         });
         let head = RequestHead {
             method: req.method.clone(),
@@ -178,7 +181,7 @@ mod tests {
         assert_eq!(envelope.headers(), &req.headers);
         let credentials = envelope.credentials().expect("credentials");
         assert_eq!(credentials.access_key(), "AKIAENVELOPE");
-        assert_eq!(credentials.secret_key(), "envelope-secret");
+        assert_eq!(credentials.secret_key(), Some("envelope-secret"));
         assert_eq!(credentials.session_token(), Some("header-token"));
         assert_eq!(envelope.region(), Some("eu-west-3"));
         assert_eq!(envelope.service(), Some("s3"));

rustfs/src/storage/storage_api.rs

@@ -147,17 +147,34 @@ pub(crate) use super::sse::{
 
 pub(crate) mod access_consumer {
     pub(crate) use super::super::access::{
-        BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, MultipartStagingOperation,
-        ObjectTagConditions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo,
+        AuthorizationRequest, BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard,
+        MultipartStagingOperation, ObjectTagConditions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard,
+        PostObjectPolicyHeaders, PostObjectRequestMarker, ReqInfo, RequestedCopySource, RequestedObjectLegalHold,
+        RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags, StagedMultipartPartAuthorization,
         TABLE_DATA_PLANE_LIST_CURSOR_PREFIX, TableDataPlaneListAccess, TableDataPlaneListCursorPosition,
         TableDataPlanePublicationGuards, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard,
-        authorize_copy_object_tags, authorize_internal_object_request, authorize_request, bucket_config_mutation_incarnation,
+        authorize_copy_object_tags, authorize_request, authorize_request_core, bucket_config_mutation_incarnation,
         delete_object_authorize_action, has_bypass_governance_header, load_bucket_generation_from_store,
         log_list_buckets_iam_implicit_deny, odm_read_generation, prepare_list_buckets_iam_authorization,
         prepare_odm_read_generation, recursive_force_delete_has_authenticated_caller, replication_proxy_allowed,
         replication_request_authorized, req_info_mut, req_info_ref, retained_table_data_plane_publication_guards,
         set_requested_object_lock_retain_until,
     };
+    pub(crate) use super::super::denial_audit::DenialAuditor;
+    pub(crate) use super::super::helper::{AuditRequestView, DenialScope};
+}
+
+#[cfg(test)]
+pub(crate) mod owned_parity_consumer {
+    pub(crate) use super::super::access::{
+        ObjectTagConditions, PostObjectPolicyHeaders, PostObjectRequestMarker, ReqInfo, RequestedCopySource,
+        RequestedObjectLegalHold, RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags,
+        TableDataPlanePublicationGuards, authorize_request, parity_test_support,
+    };
+    pub(crate) use super::super::denial_audit::{DenialAuditor, DenialRateLimiter, SUPPRESSED_DENIALS_TAG};
+    pub(crate) use super::super::ecfs::FS;
+    pub(crate) use super::super::helper::{DenialScope, authorization_denial_entry, authorization_denial_entry_for_view};
+    pub(crate) use super::super::request_context::RequestContext;
 }
 
 pub(crate) mod concurrency_consumer {