feat(table-catalog): vend credentials from LoadTable (#6878)

* feat(table-catalog): vend credentials from LoadTable

* fix(table-catalog): preserve entry-relative metadata paths

docs/architecture/s3-tables-support-matrix.md

@@ -30,7 +30,7 @@ catalog extension.
 | `/_iceberg/v1` | Supported compatibility alias | MinIO AIStor-style alias. The smoke profile defaults to REST signing name `s3tables`. |
 | S3 object data plane | Supported | Data, metadata, manifest, and delete files remain ordinary S3 objects, with table-aware policy checks for table warehouse paths. |
 | Table bucket enablement | Supported | A regular RustFS bucket can be enabled for table catalog use and then addressed as the REST catalog warehouse. |
-| Catalog-vended table credentials | Automated when enabled | Disabled by default. When enabled, the credentials endpoint returns short-lived table-scoped S3 credentials. |
+| Catalog-vended table credentials | Automated when enabled | Disabled by default. When enabled, LoadTable vends credentials only when `X-Iceberg-Access-Delegation` contains the exact `vended-credentials` token; the dedicated credentials endpoint uses the same issuer path. |
 | AWS S3 Tables endpoint shape | Profile generator | Generates the AWS catalog URI and S3 Tables warehouse ARN shape for migration docs. Full AWS S3 Tables API parity is not claimed. |
 | MinIO AIStor Tables profile | Profile generator plus RustFS alias smoke | RustFS exposes the alias shape, but does not claim all AIStor private extensions. |
 | Cloudflare R2 Data Catalog profile | Profile generator | Generates the catalog URI and warehouse-name shape for migration docs. Live RustFS interoperability is not claimed. |
@@ -69,7 +69,7 @@ catalog extension.
 | Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose staged/finalization gaps and repair safe idempotency gaps without moving the table pointer. |
 | Snapshot refs | Supported | Refs can be listed, created or replaced, and deleted through catalog commits. `main` is protected and refs with explicit retention require forced delete. |
 | Iceberg views | Supported | Basic create, list, load, replace, existence check, and drop routes persist view metadata with view-scoped authorization. Replace identifiers must match the URL resource, `schema-id: -1` resolves to the last added schema, one commit timestamp is used consistently, and only Iceberg view format version 1 is accepted. |
-| Table credentials endpoint | Supported | Returns an empty `storage-credentials` list by default. Returns table-scoped temporary credentials only when credential vending is enabled. Credential responses set `Cache-Control: no-store, private`, `Pragma: no-cache`, and `Expires: 0`. |
+| LoadTable and table credentials endpoint | Supported | LoadTable keeps the client-provided mode unless the request negotiates `vended-credentials`. Successful vending returns one temporary session for both the table warehouse prefix and the exact current metadata location. Missing credential permission falls back to metadata-only LoadTable with an explicit reason; issuer failures remain errors. Negotiated and dedicated credential responses set `Cache-Control: no-store, private`, `Pragma: no-cache`, and `Expires: 0`. |
 | Catalog diagnostics and export | Supported | Exposes recovery state, consistency state, backing manifest, recoverable commit-log WAL state, strong backing migration target, single-active-writer policy, and scale validation matrix. |
 | Catalog import and rollback | Supported | Import/register and online rollback use catalog validation and commit paths rather than direct pointer mutation. Online rollback accepts only a forward-safe metadata target that preserves assignment watermarks and retained definitions. Restoring an older target that lowers those watermarks is an offline disaster-recovery operation and requires every writer to be stopped. |
 | External catalog bridge | Supported operator path | Operator-supplied metadata pointer sync/import is supported for external catalog identity boundaries. Online vendor SDK polling and policy mirroring are not claimed. |

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

@@ -73,6 +73,8 @@ pub use view::*;
 const JSON_CONTENT_TYPE: &str = "application/json";
 const ENV_TABLE_CATALOG_CREDENTIAL_VENDING: &str = "RUSTFS_TABLE_CATALOG_CREDENTIAL_VENDING";
 const ENV_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: &str = "RUSTFS_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS";
+const ICEBERG_ACCESS_DELEGATION_HEADER: &str = "x-iceberg-access-delegation";
+const ICEBERG_VENDED_CREDENTIALS_DELEGATION: &[u8] = b"vended-credentials";
 const DEFAULT_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 15 * 60;
 const MIN_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 60;
 const MAX_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 60 * 60;
@@ -123,6 +125,7 @@ const CREDENTIAL_VENDING_UNSUPPORTED: &str = "unsupported";
 const CREDENTIAL_VENDING_SUPPORTED: &str = "supported";
 const CREDENTIAL_VENDING_UNSUPPORTED_REASON: &str = "temporary-credentials-not-implemented";
 const CREDENTIAL_VENDING_DISABLED_REASON: &str = "credential-vending-disabled";
+const CREDENTIAL_VENDING_NOT_AUTHORIZED_REASON: &str = "credential-vending-not-authorized";
 const CREDENTIAL_SCOPE_WAREHOUSE_PREFIX: &str = "warehouse-prefix";
 const CREDENTIAL_SCOPE_TABLE_PREFIX: &str = "table-prefix";
 const CREDENTIAL_MODE_CLIENT_PROVIDED: &str = "client-provided-s3-credentials-required";
@@ -717,16 +720,28 @@ struct RestLoadViewResponse {
     config: BTreeMap<String, String>,
 }
 
-#[derive(Debug, Serialize)]
+#[derive(Serialize)]
 struct RestStorageCredential {
     prefix: String,
     config: BTreeMap<String, String>,
 }
 
+impl std::fmt::Debug for RestStorageCredential {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("RestStorageCredential")
+            .field("prefix", &self.prefix)
+            .field("config", &"[REDACTED]")
+            .finish()
+    }
+}
+
 #[derive(Debug, Clone)]
 struct TableCredentialScope {
-    scope_prefix: String,
-    object_prefix: String,
+    warehouse_scope_prefix: String,
+    warehouse_object_prefix: String,
+    metadata_scope_prefix: String,
+    metadata_object: String,
 }
 
 #[derive(Debug, Clone)]
@@ -735,9 +750,10 @@ struct TableCredentialIssueRequest<'a> {
     principal: Option<&'a rustfs_credentials::Credentials>,
     scope_prefix: String,
     object_prefix: String,
+    metadata_object: String,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 struct IssuedTableCredentials {
     access_key_id: String,
     secret_access_key: String,
@@ -745,6 +761,18 @@ struct IssuedTableCredentials {
     expiration: OffsetDateTime,
 }
 
+impl std::fmt::Debug for IssuedTableCredentials {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("IssuedTableCredentials")
+            .field("access_key_id", &"[REDACTED]")
+            .field("secret_access_key", &"[REDACTED]")
+            .field("session_token", &"[REDACTED]")
+            .field("expiration", &self.expiration)
+            .finish()
+    }
+}
+
 #[async_trait::async_trait]
 trait TableCredentialIssuer: Sync {
     fn enabled(&self) -> bool {
@@ -822,7 +850,7 @@ impl TableCredentialIssuer for IamTableCredentialIssuer {
             ));
         }
 
-        let policy = table_credential_session_policy(request.entry, &request.object_prefix)?;
+        let policy = table_credential_session_policy(request.entry, &request.object_prefix, &request.metadata_object)?;
         let policy_buf = serde_json::to_vec(&policy)
             .map_err(|err| s3_error!(InternalError, "failed to serialize table credential session policy: {}", err))?;
         let expiration = OffsetDateTime::now_utc().saturating_add(Duration::seconds(self.ttl_seconds));
@@ -2461,9 +2489,24 @@ fn table_credential_scope(entry: &crate::table_catalog::TableEntry) -> S3Result<
         return Err(s3_error!(InvalidRequest, "table warehouse location must be inside the table bucket"));
     }
     let object_prefix = normalize_table_credential_object_prefix(object_prefix)?;
+    validate_persisted_table_metadata_location(entry, &entry.metadata_location)?;
+    let metadata_object =
+        crate::table_catalog::table_catalog_object_key_from_location(&entry.table_bucket, &entry.metadata_location)
+            .ok_or_else(|| persisted_metadata_error("table"))?;
+    let namespace = crate::table_catalog::Namespace::parse(&entry.namespace).map_err(|_| persisted_metadata_error("table"))?;
+    let table =
+        crate::table_catalog::IdentifierSegment::parse(entry.table.clone()).map_err(|_| persisted_metadata_error("table"))?;
+    if crate::table_catalog::is_reserved_table_object_key(&metadata_object)
+        && !crate::table_catalog::is_valid_table_metadata_location(&namespace, &table, &metadata_object)
+    {
+        return Err(persisted_metadata_error("table"));
+    }
+    let metadata_scope_prefix = table_metadata_location_for_client(&entry.table_bucket, &entry.metadata_location);
     Ok(TableCredentialScope {
-        scope_prefix: format!("s3://{bucket}/{object_prefix}"),
-        object_prefix,
+        warehouse_scope_prefix: format!("s3://{bucket}/{object_prefix}"),
+        warehouse_object_prefix: object_prefix,
+        metadata_scope_prefix,
+        metadata_object,
     })
 }
 
@@ -2501,9 +2544,16 @@ fn table_credential_catalog_resource(entry: &crate::table_catalog::TableEntry) -
     Ok(format!("namespaces/{}/tables/{}", namespace.storage_id(), table.as_str()))
 }
 
-fn table_credential_session_policy(entry: &crate::table_catalog::TableEntry, object_prefix: &str) -> S3Result<Policy> {
+fn table_credential_session_policy(
+    entry: &crate::table_catalog::TableEntry,
+    object_prefix: &str,
+    metadata_object: &str,
+) -> S3Result<Policy> {
     let bucket = &entry.table_bucket;
     let object_prefix = normalize_table_credential_object_prefix(object_prefix)?;
+    validate_persisted_table_metadata_location(entry, metadata_object)?;
+    let metadata_object = crate::table_catalog::table_catalog_object_key_from_location(bucket, metadata_object)
+        .ok_or_else(|| persisted_metadata_error("table"))?;
     let catalog_resource = table_credential_catalog_resource(entry)?;
     let policy = serde_json::json!({
         "Version": "2012-10-17",
@@ -2521,6 +2571,15 @@ fn table_credential_session_policy(entry: &crate::table_catalog::TableEntry, obj
                     format!("arn:aws:s3:::{bucket}/{object_prefix}*")
                 ]
             },
+            {
+                "Effect": "Allow",
+                "Action": [
+                    "s3:GetObject"
+                ],
+                "Resource": [
+                    format!("arn:aws:s3:::{bucket}/{metadata_object}")
+                ]
+            },
             {
                 "Effect": "Allow",
                 "Action": [
@@ -2547,23 +2606,44 @@ fn table_credential_session_policy(entry: &crate::table_catalog::TableEntry, obj
     Policy::parse_config(&data).map_err(|err| s3_error!(InvalidRequest, "invalid table credential policy: {}", err))
 }
 
-fn storage_credential_from_issued(scope: TableCredentialScope, issued: IssuedTableCredentials) -> RestStorageCredential {
+fn storage_credential_config(scope_prefix: &str, issued: &IssuedTableCredentials) -> BTreeMap<String, String> {
     let mut config = BTreeMap::new();
-    config.insert(S3_ACCESS_KEY_ID_CONFIG_KEY.to_string(), issued.access_key_id);
-    config.insert(S3_SECRET_ACCESS_KEY_CONFIG_KEY.to_string(), issued.secret_access_key);
-    config.insert(S3_SESSION_TOKEN_CONFIG_KEY.to_string(), issued.session_token);
+    config.insert(S3_ACCESS_KEY_ID_CONFIG_KEY.to_string(), issued.access_key_id.clone());
+    config.insert(S3_SECRET_ACCESS_KEY_CONFIG_KEY.to_string(), issued.secret_access_key.clone());
+    config.insert(S3_SESSION_TOKEN_CONFIG_KEY.to_string(), issued.session_token.clone());
     config.insert(CREDENTIAL_VENDING_CONFIG_KEY.to_string(), CREDENTIAL_VENDING_SUPPORTED.to_string());
     config.insert(CREDENTIAL_MODE_CONFIG_KEY.to_string(), CREDENTIAL_MODE_CATALOG_VENDED.to_string());
     config.insert(CREDENTIAL_SCOPE_CONFIG_KEY.to_string(), CREDENTIAL_SCOPE_TABLE_PREFIX.to_string());
-    config.insert(CREDENTIAL_SCOPE_PREFIX_CONFIG_KEY.to_string(), scope.scope_prefix.clone());
+    config.insert(CREDENTIAL_SCOPE_PREFIX_CONFIG_KEY.to_string(), scope_prefix.to_string());
     config.insert(
         CREDENTIAL_EXPIRATION_CONFIG_KEY.to_string(),
         issued.expiration.unix_timestamp().to_string(),
     );
-    RestStorageCredential {
-        prefix: scope.scope_prefix,
-        config,
-    }
+    config
+}
+
+fn storage_credentials_from_issued(scope: TableCredentialScope, issued: IssuedTableCredentials) -> Vec<RestStorageCredential> {
+    let warehouse_config = storage_credential_config(&scope.warehouse_scope_prefix, &issued);
+    let metadata_config = storage_credential_config(&scope.metadata_scope_prefix, &issued);
+    vec![
+        RestStorageCredential {
+            prefix: scope.warehouse_scope_prefix,
+            config: warehouse_config,
+        },
+        RestStorageCredential {
+            prefix: scope.metadata_scope_prefix,
+            config: metadata_config,
+        },
+    ]
+}
+
+fn requests_vended_credentials(headers: &HeaderMap) -> bool {
+    headers
+        .get_all(ICEBERG_ACCESS_DELEGATION_HEADER)
+        .iter()
+        .flat_map(|value| value.as_bytes().split(|byte| *byte == b','))
+        .map(|token| token.trim_ascii())
+        .any(|token| token == ICEBERG_VENDED_CREDENTIALS_DELEGATION)
 }
 
 fn table_metadata_location_for_client(table_bucket: &str, metadata_location: &str) -> String {
@@ -2647,6 +2727,13 @@ fn load_credentials_response_config(vending: &str, mode: &str, reason: Option<&s
     config
 }
 
+fn client_provided_credentials_response(reason: &str) -> RestLoadCredentialsResponse {
+    RestLoadCredentialsResponse {
+        config: load_credentials_response_config(CREDENTIAL_VENDING_UNSUPPORTED, CREDENTIAL_MODE_CLIENT_PROVIDED, Some(reason)),
+        storage_credentials: Vec::new(),
+    }
+}
+
 fn add_table_credential_scope_config(config: &mut BTreeMap<String, String>, scope_prefix: &str) {
     config.insert(CREDENTIAL_SCOPE_CONFIG_KEY.to_string(), CREDENTIAL_SCOPE_TABLE_PREFIX.to_string());
     config.insert(CREDENTIAL_SCOPE_PREFIX_CONFIG_KEY.to_string(), scope_prefix.to_string());
@@ -2658,25 +2745,19 @@ async fn load_credentials_response_from_entry(
     principal: Option<&rustfs_credentials::Credentials>,
 ) -> S3Result<RestLoadCredentialsResponse> {
     if !issuer.enabled() {
-        return Ok(RestLoadCredentialsResponse {
-            config: load_credentials_response_config(
-                CREDENTIAL_VENDING_UNSUPPORTED,
-                CREDENTIAL_MODE_CLIENT_PROVIDED,
-                Some(CREDENTIAL_VENDING_DISABLED_REASON),
-            ),
-            storage_credentials: Vec::new(),
-        });
+        return Ok(client_provided_credentials_response(CREDENTIAL_VENDING_DISABLED_REASON));
     }
     let scope = table_credential_scope(entry)?;
     let request = TableCredentialIssueRequest {
         entry,
         principal,
-        scope_prefix: scope.scope_prefix.clone(),
-        object_prefix: scope.object_prefix.clone(),
+        scope_prefix: scope.warehouse_scope_prefix.clone(),
+        object_prefix: scope.warehouse_object_prefix.clone(),
+        metadata_object: scope.metadata_object.clone(),
     };
-    let scope_prefix = scope.scope_prefix.clone();
+    let scope_prefix = scope.warehouse_scope_prefix.clone();
     let storage_credentials = match issuer.issue_table_credentials(request).await? {
-        Some(issued) => vec![storage_credential_from_issued(scope, issued)],
+        Some(issued) => storage_credentials_from_issued(scope, issued),
         None => {
             let mut config = load_credentials_response_config(
                 CREDENTIAL_VENDING_UNSUPPORTED,
@@ -2698,6 +2779,28 @@ async fn load_credentials_response_from_entry(
     })
 }
 
+fn apply_credentials_to_load_table_response(
+    mut response: RestLoadTableResponse,
+    credential_response: RestLoadCredentialsResponse,
+) -> RestLoadTableResponse {
+    response.config.remove(CREDENTIAL_VENDING_CONFIG_KEY);
+    response.config.remove(CREDENTIAL_VENDING_REASON_CONFIG_KEY);
+    response.config.remove(CREDENTIAL_MODE_CONFIG_KEY);
+    response.config.extend(credential_response.config);
+    response.storage_credentials = credential_response.storage_credentials;
+    response
+}
+
+async fn enrich_load_table_response_with_credentials(
+    response: RestLoadTableResponse,
+    entry: &crate::table_catalog::TableEntry,
+    issuer: &dyn TableCredentialIssuer,
+    principal: Option<&rustfs_credentials::Credentials>,
+) -> S3Result<RestLoadTableResponse> {
+    let credential_response = load_credentials_response_from_entry(entry, issuer, principal).await?;
+    Ok(apply_credentials_to_load_table_response(response, credential_response))
+}
+
 fn commit_table_response_from_result(
     result: crate::table_catalog::TableCommitResult,
     metadata: serde_json::Value,
@@ -5510,6 +5613,37 @@ async fn load_table_response<S>(
     namespace: &crate::table_catalog::Namespace,
     table: &str,
 ) -> S3Result<RestLoadTableResponse>
+where
+    S: crate::table_catalog::TableCatalogStore + ?Sized,
+{
+    let (entry, metadata) = load_table_entry_and_metadata(store, metadata_backend, bucket, namespace, table).await?;
+    Ok(load_table_response_from_entry(entry, metadata))
+}
+
+async fn load_table_response_with_credentials<S>(
+    store: &S,
+    metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend,
+    bucket: &str,
+    namespace: &crate::table_catalog::Namespace,
+    table: &str,
+    issuer: &dyn TableCredentialIssuer,
+    principal: Option<&rustfs_credentials::Credentials>,
+) -> S3Result<RestLoadTableResponse>
+where
+    S: crate::table_catalog::TableCatalogStore + ?Sized,
+{
+    let (entry, metadata) = load_table_entry_and_metadata(store, metadata_backend, bucket, namespace, table).await?;
+    let response = load_table_response_from_entry(entry.clone(), metadata);
+    enrich_load_table_response_with_credentials(response, &entry, issuer, principal).await
+}
+
+async fn load_table_entry_and_metadata<S>(
+    store: &S,
+    metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend,
+    bucket: &str,
+    namespace: &crate::table_catalog::Namespace,
+    table: &str,
+) -> S3Result<(crate::table_catalog::TableEntry, serde_json::Value)>
 where
     S: crate::table_catalog::TableCatalogStore + ?Sized,
 {
@@ -5521,7 +5655,7 @@ where
         return Err(iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_TABLE, StatusCode::NOT_FOUND, "table not found"));
     };
     let metadata = read_persisted_table_metadata_for_entry(metadata_backend, &entry, &entry.metadata_location, true).await?;
-    Ok(load_table_response_from_entry(entry, metadata))
+    Ok((entry, metadata))
 }
 
 async fn list_views_response<S>(

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

@@ -121,14 +121,56 @@ impl Operation for RestLoadTableHandler {
         let namespace = namespace_from_params(&params)?;
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?;
         ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let snapshot_selection = rest_table_snapshot_selection_from_query(&req.uri)?;
-        let mut response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
+        let vended_credentials_requested = requests_vended_credentials(&req.headers);
+        let mut response = if !vended_credentials_requested {
+            load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?
+        } else {
+            let issuer = IamTableCredentialIssuer::from_request(&req)?;
+            let credential_permission = if issuer.enabled() {
+                authorize_table_catalog_resource_for_principal(
+                    &req,
+                    &principal,
+                    &resource,
+                    AdminAction::GetTableCredentialsAction,
+                )
+                .await
+            } else {
+                Ok(())
+            };
+            match credential_permission {
+                Ok(()) => {
+                    load_table_response_with_credentials(
+                        &store,
+                        &metadata_backend,
+                        &warehouse,
+                        &namespace,
+                        &table,
+                        &issuer,
+                        Some(&principal.credentials),
+                    )
+                    .await?
+                }
+                Err(err) if err.code() == &S3ErrorCode::AccessDenied => {
+                    let response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
+                    apply_credentials_to_load_table_response(
+                        response,
+                        client_provided_credentials_response(CREDENTIAL_VENDING_NOT_AUTHORIZED_REASON),
+                    )
+                }
+                Err(err) => return Err(err),
+            }
+        };
         apply_rest_table_snapshot_selection(&mut response.metadata, snapshot_selection);
-        build_json_response(StatusCode::OK, &response)
+        if vended_credentials_requested {
+            build_sensitive_json_response(StatusCode::OK, &response)
+        } else {
+            build_json_response(StatusCode::OK, &response)
+        }
     }
 }
 

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

@@ -1,5 +1,6 @@
 use super::*;
 use crate::admin::runtime_sources::{AppContext, IamInterface, KmsInterface, ServerContextSlot};
+use crate::storage::storage_api::contract::bucket::{BucketOperations as _, MakeBucketOptions};
 use crate::table_catalog::{TableCatalogObjectBackend, TableCatalogStore};
 use datafusion::{
     arrow::{
@@ -43,6 +44,49 @@ impl KmsInterface for RequestKms {
     }
 }
 
+fn table_catalog_handler_request(context: Arc<AppContext>, access_key: &str, secret_key: &str) -> S3Request<Body> {
+    let slot = ServerContextSlot::new();
+    assert!(slot.install(context));
+    let mut extensions = http::Extensions::new();
+    extensions.insert(slot);
+    S3Request {
+        input: Body::empty(),
+        method: Method::GET,
+        uri: "/iceberg/v1/warehouse/namespaces/analytics/tables/events"
+            .parse()
+            .expect("load table URI"),
+        headers: HeaderMap::new(),
+        extensions,
+        credentials: Some(s3s::auth::Credentials {
+            access_key: access_key.to_string(),
+            secret_key: s3s::auth::SecretKey::from(secret_key.to_string()),
+        }),
+        region: None,
+        service: None,
+        trailing_headers: None,
+    }
+}
+
+fn with_access_delegation(mut request: S3Request<Body>, values: &[&str]) -> S3Request<Body> {
+    for value in values {
+        request.headers.append(
+            ICEBERG_ACCESS_DELEGATION_HEADER,
+            HeaderValue::from_str(value).expect("access delegation header should be valid"),
+        );
+    }
+    request
+}
+
+async fn call_load_table_handler(request: S3Request<Body>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let mut router = matchit::Router::new();
+    router
+        .insert("/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}", ())
+        .expect("load table test route should insert");
+    let path = request.uri.path().to_string();
+    let matched = router.at(&path).expect("load table test route should match");
+    RestLoadTableHandler {}.call(request, matched.params).await
+}
+
 #[tokio::test]
 async fn table_catalog_authentication_and_credentials_use_the_request_context() {
     let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
@@ -130,6 +174,238 @@ async fn table_catalog_authentication_and_credentials_use_the_request_context()
     assert!(Arc::ptr_eq(&resolved_store, &store));
 }
 
+#[tokio::test]
+#[serial_test::serial]
+async fn load_table_handler_negotiates_vended_credentials_without_breaking_metadata_only_callers() {
+    let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    object_store
+        .make_bucket("warehouse", &MakeBucketOptions::default())
+        .await
+        .expect("table bucket should be created");
+    rustfs_iam::store::object::ObjectStore::new(object_store.clone())
+        .save_iam_config(
+            serde_json::json!({"version": 1}),
+            format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+        )
+        .await
+        .expect("request IAM format should be seeded");
+    let iam = rustfs_iam::build_iam_sys(object_store.clone())
+        .await
+        .expect("request IAM should initialize");
+    let metadata_access_key = "load-table-metadata-only";
+    let metadata_secret_key = "load-table-metadata-only-secret";
+    let vended_access_key = "load-table-vended";
+    let vended_secret_key = "load-table-vended-secret";
+    for (access_key, secret_key) in [
+        (metadata_access_key, metadata_secret_key),
+        (vended_access_key, vended_secret_key),
+    ] {
+        iam.create_user(
+            access_key,
+            &AddOrUpdateUserReq {
+                secret_key: secret_key.to_string(),
+                policy: None,
+                status: AccountStatus::Enabled,
+            },
+        )
+        .await
+        .expect("load table user should be created");
+    }
+    iam.set_policy(
+        "load-table-metadata-only-policy",
+        Policy::parse_config(br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:GetTableMetadata"]}]}"#)
+            .expect("metadata-only policy should parse"),
+    )
+    .await
+    .expect("metadata-only policy should be stored");
+    iam.policy_db_set(metadata_access_key, UserType::Reg, false, "load-table-metadata-only-policy")
+        .await
+        .expect("metadata-only policy should be attached");
+    iam.set_policy(
+        "load-table-vended-policy",
+        Policy::parse_config(br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:GetTableMetadata","admin:GetTableCredentials"]}]}"#)
+            .expect("vended credential policy should parse"),
+    )
+    .await
+    .expect("vended credential policy should be stored");
+    iam.policy_db_set(vended_access_key, UserType::Reg, false, "load-table-vended-policy")
+        .await
+        .expect("vended credential policy should be attached");
+
+    let action_credentials = rustfs_credentials::Credentials {
+        access_key: "load-table-root-access-key".to_string(),
+        secret_key: "load-table-root-secret-key".to_string(),
+        status: "on".to_string(),
+        ..Default::default()
+    };
+    let context = Arc::new(AppContext::new(
+        object_store.clone(),
+        Arc::new(RequestIam { handle: iam }),
+        Arc::new(RequestKms),
+    ));
+    assert!(context.publish_action_credentials(action_credentials));
+
+    let setup_request = table_catalog_handler_request(context.clone(), vended_access_key, vended_secret_key);
+    let metadata_backend =
+        table_catalog_backend_from_extensions(&setup_request.extensions).expect("table catalog backend should resolve");
+    let catalog_store = table_catalog_store_from_backend(metadata_backend.clone()).expect("table catalog store should resolve");
+    enable_table_bucket_marker(object_store.as_ref(), "warehouse")
+        .await
+        .expect("table bucket should be enabled");
+    ensure_table_bucket_entry(&catalog_store, "warehouse", true)
+        .await
+        .expect("table bucket entry should be created");
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    create_namespace_response(
+        &catalog_store,
+        "warehouse",
+        CreateNamespaceRequest {
+            namespace: vec!["analytics".to_string()],
+            properties: BTreeMap::new(),
+        },
+        true,
+    )
+    .await
+    .expect("namespace should be created");
+    let create_request = serde_json::from_value::<CreateTableRequest>(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        }
+    }))
+    .expect("create table request should parse");
+    create_table_response(
+        &catalog_store,
+        &TableCommitObjectBackend::trusted(metadata_backend),
+        "warehouse",
+        &namespace,
+        create_request,
+        true,
+    )
+    .await
+    .expect("table should be created");
+
+    let absent = temp_env::async_with_vars([(ENV_TABLE_CATALOG_CREDENTIAL_VENDING, Some("true"))], async {
+        call_load_table_handler(table_catalog_handler_request(context.clone(), metadata_access_key, metadata_secret_key)).await
+    })
+    .await
+    .expect("metadata-only caller should load without requesting delegation");
+    assert!(absent.headers.get(http::header::CACHE_CONTROL).is_none());
+    let absent_json: serde_json::Value =
+        serde_json::from_slice(&absent.output.1.bytes().expect("absent delegation body should be buffered"))
+            .expect("absent delegation response should parse");
+    assert_eq!(absent_json["storage-credentials"], serde_json::json!([]));
+
+    let disabled = temp_env::async_with_vars([(ENV_TABLE_CATALOG_CREDENTIAL_VENDING, None::<&str>)], async {
+        call_load_table_handler(with_access_delegation(
+            table_catalog_handler_request(context.clone(), metadata_access_key, metadata_secret_key),
+            &["vended-credentials"],
+        ))
+        .await
+    })
+    .await
+    .expect("disabled vending should not add a credential permission requirement");
+    assert_eq!(
+        disabled.headers.get(http::header::CACHE_CONTROL),
+        Some(&HeaderValue::from_static("no-store, private"))
+    );
+    let disabled_json: serde_json::Value =
+        serde_json::from_slice(&disabled.output.1.bytes().expect("disabled vending body should be buffered"))
+            .expect("disabled vending response should parse");
+    assert_eq!(
+        disabled_json["config"][CREDENTIAL_VENDING_REASON_CONFIG_KEY],
+        serde_json::Value::String(CREDENTIAL_VENDING_DISABLED_REASON.to_string())
+    );
+    assert_eq!(disabled_json["storage-credentials"], serde_json::json!([]));
+
+    let remote_signing = temp_env::async_with_vars([(ENV_TABLE_CATALOG_CREDENTIAL_VENDING, Some("true"))], async {
+        call_load_table_handler(with_access_delegation(
+            table_catalog_handler_request(context.clone(), metadata_access_key, metadata_secret_key),
+            &["remote-signing"],
+        ))
+        .await
+    })
+    .await
+    .expect("unrequested vending should preserve metadata-only access");
+    assert!(remote_signing.headers.get(http::header::CACHE_CONTROL).is_none());
+
+    let not_authorized = temp_env::async_with_vars([(ENV_TABLE_CATALOG_CREDENTIAL_VENDING, Some("true"))], async {
+        call_load_table_handler(with_access_delegation(
+            table_catalog_handler_request(context.clone(), metadata_access_key, metadata_secret_key),
+            &["vended-credentials"],
+        ))
+        .await
+    })
+    .await
+    .expect("credential denial should fall back to the authorized metadata response");
+    let not_authorized_json: serde_json::Value = serde_json::from_slice(
+        &not_authorized
+            .output
+            .1
+            .bytes()
+            .expect("credential denial body should be buffered"),
+    )
+    .expect("credential denial response should parse");
+    assert_eq!(
+        not_authorized_json["config"][CREDENTIAL_VENDING_REASON_CONFIG_KEY],
+        serde_json::Value::String(CREDENTIAL_VENDING_NOT_AUTHORIZED_REASON.to_string())
+    );
+    assert_eq!(not_authorized_json["storage-credentials"], serde_json::json!([]));
+
+    let issued = temp_env::async_with_vars([(ENV_TABLE_CATALOG_CREDENTIAL_VENDING, Some("true"))], async {
+        call_load_table_handler(with_access_delegation(
+            table_catalog_handler_request(context.clone(), vended_access_key, vended_secret_key),
+            &["remote-signing", "unknown, vended-credentials"],
+        ))
+        .await
+    })
+    .await
+    .expect("credential-authorized caller should receive vended credentials");
+    assert_eq!(issued.output.0, StatusCode::OK);
+    assert_eq!(
+        issued.headers.get(http::header::CACHE_CONTROL),
+        Some(&HeaderValue::from_static("no-store, private"))
+    );
+    assert_eq!(issued.headers.get(http::header::PRAGMA), Some(&HeaderValue::from_static("no-cache")));
+    assert_eq!(issued.headers.get(http::header::EXPIRES), Some(&HeaderValue::from_static("0")));
+    let issued_json: serde_json::Value =
+        serde_json::from_slice(&issued.output.1.bytes().expect("issued credential body should be buffered"))
+            .expect("issued credential response should parse");
+    assert_eq!(
+        issued_json["config"][CREDENTIAL_VENDING_CONFIG_KEY],
+        serde_json::Value::String(CREDENTIAL_VENDING_SUPPORTED.to_string())
+    );
+    assert_eq!(
+        issued_json["config"][CREDENTIAL_MODE_CONFIG_KEY],
+        serde_json::Value::String(CREDENTIAL_MODE_CATALOG_VENDED.to_string())
+    );
+    assert_eq!(issued_json["storage-credentials"].as_array().map(Vec::len), Some(2));
+    assert_eq!(issued_json["storage-credentials"][1]["prefix"], issued_json["metadata-location"]);
+    assert_eq!(
+        issued_json["storage-credentials"][0]["config"][S3_ACCESS_KEY_ID_CONFIG_KEY],
+        issued_json["storage-credentials"][1]["config"][S3_ACCESS_KEY_ID_CONFIG_KEY]
+    );
+    for required_key in [
+        S3_ACCESS_KEY_ID_CONFIG_KEY,
+        S3_SECRET_ACCESS_KEY_CONFIG_KEY,
+        S3_SESSION_TOKEN_CONFIG_KEY,
+    ] {
+        for credential in issued_json["storage-credentials"]
+            .as_array()
+            .expect("storage credentials should be an array")
+        {
+            assert!(
+                credential["config"][required_key]
+                    .as_str()
+                    .is_some_and(|value| !value.is_empty()),
+                "LoadTable should include {required_key}"
+            );
+        }
+    }
+}
+
 #[test]
 #[serial_test::serial]
 fn catalog_config_response_lists_standard_rest_endpoints() {
@@ -9924,6 +10200,10 @@ impl TableCredentialIssuer for TestTableCredentialIssuer {
         assert_eq!(request.entry.table_bucket, "warehouse");
         assert_eq!(request.scope_prefix, "s3://warehouse/tables/table-id/");
         assert_eq!(request.object_prefix, "tables/table-id/");
+        assert_eq!(
+            request.metadata_object,
+            ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json"
+        );
         Ok(Some(IssuedTableCredentials {
             access_key_id: "temporary-access-key".to_string(),
             secret_access_key: "temporary-secret-key".to_string(),
@@ -9957,25 +10237,106 @@ async fn credential_issuer_returns_temporary_scoped_storage_credentials() {
     assert!(!response.config.contains_key(S3_ACCESS_KEY_ID_CONFIG_KEY));
     assert!(!response.config.contains_key(S3_SECRET_ACCESS_KEY_CONFIG_KEY));
     assert!(!response.config.contains_key(S3_SESSION_TOKEN_CONFIG_KEY));
-    assert_eq!(response.storage_credentials.len(), 1);
-    let credential = &response.storage_credentials[0];
-    assert_eq!(credential.prefix, "s3://warehouse/tables/table-id/");
-    assert_eq!(credential.config.get("s3.access-key-id"), Some(&"temporary-access-key".to_string()));
-    assert_eq!(credential.config.get("s3.secret-access-key"), Some(&"temporary-secret-key".to_string()));
-    assert_eq!(credential.config.get("s3.session-token"), Some(&"temporary-session-token".to_string()));
+    assert_eq!(response.storage_credentials.len(), 2);
+    assert_eq!(response.storage_credentials[0].prefix, "s3://warehouse/tables/table-id/");
     assert_eq!(
-        credential.config.get("rustfs.credential-mode"),
-        Some(&"catalog-vended-temporary-credentials".to_string())
+        response.storage_credentials[1].prefix,
+        "s3://warehouse/.rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json"
+    );
+    for credential in &response.storage_credentials {
+        assert_eq!(credential.config.get("s3.access-key-id"), Some(&"temporary-access-key".to_string()));
+        assert_eq!(credential.config.get("s3.secret-access-key"), Some(&"temporary-secret-key".to_string()));
+        assert_eq!(credential.config.get("s3.session-token"), Some(&"temporary-session-token".to_string()));
+        assert_eq!(
+            credential.config.get("rustfs.credential-mode"),
+            Some(&"catalog-vended-temporary-credentials".to_string())
+        );
+        assert_eq!(credential.config.get("rustfs.credential-scope-prefix"), Some(&credential.prefix));
+        assert_eq!(
+            credential.config.get("rustfs.credential-expiration-unix-seconds"),
+            Some(&"1800000000".to_string())
+        );
+        assert!(!credential.config.contains_key("rustfs.credential-vending-reason"));
+    }
+}
+
+#[tokio::test]
+async fn load_table_uses_the_shared_credential_vending_result() {
+    let entry = table_entry_for_credentials();
+    let metadata = serde_json::json!({
+        "format-version": 2,
+        "table-uuid": "table-uuid",
+        "location": "s3://warehouse/tables/table-id"
+    });
+    let principal = rustfs_credentials::Credentials {
+        access_key: "parent-access-key".to_string(),
+        secret_key: "parent-secret-key".to_string(),
+        ..Default::default()
+    };
+    let load_table = enrich_load_table_response_with_credentials(
+        load_table_response_from_entry(entry.clone(), metadata),
+        &entry,
+        &TestTableCredentialIssuer,
+        Some(&principal),
+    )
+    .await
+    .expect("load table should include vended credentials");
+    let credentials = load_credentials_response_from_entry(&entry, &TestTableCredentialIssuer, Some(&principal))
+        .await
+        .expect("credentials endpoint should include vended credentials");
+
+    assert_eq!(
+        load_table.config.get(CREDENTIAL_VENDING_CONFIG_KEY),
+        Some(&CREDENTIAL_VENDING_SUPPORTED.to_string())
     );
     assert_eq!(
-        credential.config.get("rustfs.credential-scope-prefix"),
-        Some(&"s3://warehouse/tables/table-id/".to_string())
+        load_table.config.get(CREDENTIAL_MODE_CONFIG_KEY),
+        Some(&CREDENTIAL_MODE_CATALOG_VENDED.to_string())
+    );
+    assert!(!load_table.config.contains_key(CREDENTIAL_VENDING_REASON_CONFIG_KEY));
+    assert_eq!(
+        load_table.config.get(CREDENTIAL_SCOPE_PREFIX_CONFIG_KEY),
+        credentials.config.get(CREDENTIAL_SCOPE_PREFIX_CONFIG_KEY)
     );
     assert_eq!(
-        credential.config.get("rustfs.credential-expiration-unix-seconds"),
-        Some(&"1800000000".to_string())
+        serde_json::to_value(&load_table.storage_credentials).expect("load table credentials should serialize"),
+        serde_json::to_value(&credentials.storage_credentials).expect("endpoint credentials should serialize")
     );
-    assert!(!credential.config.contains_key("rustfs.credential-vending-reason"));
+}
+
+struct RefusingTableCredentialIssuer;
+
+#[async_trait::async_trait]
+impl TableCredentialIssuer for RefusingTableCredentialIssuer {
+    async fn issue_table_credentials(
+        &self,
+        _request: TableCredentialIssueRequest<'_>,
+    ) -> S3Result<Option<IssuedTableCredentials>> {
+        Err(S3Error::with_message(
+            S3ErrorCode::AccessDenied,
+            "table credential issuer refused request",
+        ))
+    }
+}
+
+#[tokio::test]
+async fn load_table_and_credentials_endpoint_propagate_issuer_refusal() {
+    let entry = table_entry_for_credentials();
+    let credentials_error = load_credentials_response_from_entry(&entry, &RefusingTableCredentialIssuer, None)
+        .await
+        .expect_err("credentials endpoint should propagate issuer refusal");
+    let load_table_error = enrich_load_table_response_with_credentials(
+        load_table_response_from_entry(entry.clone(), serde_json::json!({})),
+        &entry,
+        &RefusingTableCredentialIssuer,
+        None,
+    )
+    .await
+    .expect_err("load table should propagate issuer refusal");
+
+    assert_eq!(load_table_error.code(), credentials_error.code());
+    assert_eq!(load_table_error.status_code(), credentials_error.status_code());
+    assert_eq!(load_table_error.message(), credentials_error.message());
 }
 
 #[tokio::test]
@@ -10013,6 +10374,19 @@ fn credential_http_response_disables_caching() {
     assert_eq!(response.headers.get(http::header::EXPIRES), Some(&HeaderValue::from_static("0")));
 }
 
+#[tokio::test]
+async fn credential_debug_output_redacts_secrets_and_tokens() {
+    let response = load_credentials_response_from_entry(&table_entry_for_credentials(), &TestTableCredentialIssuer, None)
+        .await
+        .expect("issuer should build a scoped credential response");
+    let debug_output = format!("{response:?}");
+
+    assert!(!debug_output.contains("temporary-access-key"));
+    assert!(!debug_output.contains("temporary-secret-key"));
+    assert!(!debug_output.contains("temporary-session-token"));
+    assert!(debug_output.contains("[REDACTED]"));
+}
+
 #[test]
 fn table_credentials_do_not_snapshot_parent_groups() {
     let principal = rustfs_credentials::Credentials {
@@ -10030,7 +10404,8 @@ fn table_credentials_do_not_snapshot_parent_groups() {
 
 #[tokio::test]
 async fn table_credential_session_policy_is_limited_to_table_prefix() {
-    let policy = table_credential_session_policy(&table_entry_for_credentials(), "tables/table-id/")
+    let metadata_object = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json";
+    let policy = table_credential_session_policy(&table_entry_for_credentials(), "tables/table-id/", metadata_object)
         .expect("table credential policy should build");
     let groups = None;
     let conditions = std::collections::HashMap::new();
@@ -10081,6 +10456,66 @@ async fn table_credential_session_policy_is_limited_to_table_prefix() {
             })
             .await
     );
+    assert!(
+        policy
+            .is_allowed(&rustfs_policy::policy::Args {
+                account: "temporary-access-key",
+                groups: &groups,
+                action: Action::S3Action(rustfs_policy::policy::action::S3Action::GetObjectAction),
+                bucket: "warehouse",
+                conditions: &conditions,
+                is_owner: false,
+                object: metadata_object,
+                claims: &claims,
+                deny_only: false,
+            })
+            .await
+    );
+    assert!(
+        !policy
+            .is_allowed(&rustfs_policy::policy::Args {
+                account: "temporary-access-key",
+                groups: &groups,
+                action: Action::S3Action(rustfs_policy::policy::action::S3Action::PutObjectAction),
+                bucket: "warehouse",
+                conditions: &conditions,
+                is_owner: false,
+                object: metadata_object,
+                claims: &claims,
+                deny_only: false,
+            })
+            .await
+    );
+    assert!(
+        !policy
+            .is_allowed(&rustfs_policy::policy::Args {
+                account: "temporary-access-key",
+                groups: &groups,
+                action: Action::S3Action(rustfs_policy::policy::action::S3Action::DeleteObjectAction),
+                bucket: "warehouse",
+                conditions: &conditions,
+                is_owner: false,
+                object: metadata_object,
+                claims: &claims,
+                deny_only: false,
+            })
+            .await
+    );
+    assert!(
+        !policy
+            .is_allowed(&rustfs_policy::policy::Args {
+                account: "temporary-access-key",
+                groups: &groups,
+                action: Action::S3Action(rustfs_policy::policy::action::S3Action::GetObjectAction),
+                bucket: "warehouse",
+                conditions: &conditions,
+                is_owner: false,
+                object: ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00002.metadata.json",
+                claims: &claims,
+                deny_only: false,
+            })
+            .await
+    );
     assert!(
         !policy
             .is_allowed(&rustfs_policy::policy::Args {
@@ -10115,8 +10550,12 @@ async fn table_credential_session_policy_is_limited_to_table_prefix() {
 
 #[tokio::test]
 async fn table_credential_session_policy_includes_table_resource_actions() {
-    let policy = table_credential_session_policy(&table_entry_for_credentials(), "tables/table-id/")
-        .expect("table credential policy should build");
+    let policy = table_credential_session_policy(
+        &table_entry_for_credentials(),
+        "tables/table-id/",
+        ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json",
+    )
+    .expect("table credential policy should build");
     let groups = None;
     let conditions = std::collections::HashMap::new();
     let claims = std::collections::HashMap::new();
@@ -10177,6 +10616,45 @@ fn table_credential_scope_rejects_cross_bucket_or_unsafe_prefix() {
     let mut entry = table_entry_for_credentials();
     entry.warehouse_location = "s3://warehouse/tables/../table-id".to_string();
     assert!(table_credential_scope(&entry).is_err());
+
+    let mut entry = table_entry_for_credentials();
+    entry.metadata_location = "s3://other/.rustfs-table/metadata/00001.metadata.json".to_string();
+    assert!(table_credential_scope(&entry).is_err());
+
+    let mut entry = table_entry_for_credentials();
+    entry.metadata_location =
+        ".rustfs-table/warehouses/default/namespaces/analytics/tables/orders/metadata/00001.metadata.json".to_string();
+    assert!(table_credential_scope(&entry).is_err());
+}
+
+#[test]
+fn table_credential_scope_accepts_entry_relative_metadata_location() {
+    let mut entry = table_entry_for_credentials();
+    entry.metadata_location = "s3://warehouse/tables/table-id/metadata/v1.metadata.json".to_string();
+
+    let scope = table_credential_scope(&entry).expect("entry-relative metadata should remain vendable");
+
+    assert_eq!(scope.metadata_object, "tables/table-id/metadata/v1.metadata.json");
+    assert_eq!(scope.metadata_scope_prefix, "s3://warehouse/tables/table-id/metadata/v1.metadata.json");
+    table_credential_session_policy(&entry, &scope.warehouse_object_prefix, &scope.metadata_object)
+        .expect("entry-relative metadata should produce a credential policy");
+}
+
+#[test]
+fn vended_credential_delegation_requires_an_exact_comma_separated_token() {
+    let mut headers = HeaderMap::new();
+    assert!(!requests_vended_credentials(&headers));
+
+    headers.append(ICEBERG_ACCESS_DELEGATION_HEADER, HeaderValue::from_static("remote-signing"));
+    headers.append(ICEBERG_ACCESS_DELEGATION_HEADER, HeaderValue::from_static("unknown, vended-credentials"));
+    assert!(requests_vended_credentials(&headers));
+
+    let mut headers = HeaderMap::new();
+    headers.insert(
+        ICEBERG_ACCESS_DELEGATION_HEADER,
+        HeaderValue::from_static("not-vended-credentials, VENDED-CREDENTIALS"),
+    );
+    assert!(!requests_vended_credentials(&headers));
 }
 
 #[test]

scripts/table-catalog/README.md

@@ -475,25 +475,32 @@ current unsupported inventory is:
 ## Credential Boundary
 
 RustFS advertises table credential scope metadata without returning reusable
-storage secrets by default. `loadTable` includes the table warehouse prefix in
-the response config, and the standard credentials endpoint is registered:
+storage secrets by default. The standard credentials endpoint is registered:
 
 ```text
 GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/credentials
 ```
 
 The endpoint returns an empty `storage-credentials` list unless table catalog
-credential vending is explicitly enabled. When enabled, RustFS issues temporary
-table-scoped S3 credentials through the credentials endpoint. Those credentials
-are constrained to the table warehouse prefix and include a session token and
-expiration.
+credential vending is explicitly enabled. LoadTable uses the same issuer when
+the request includes `X-Iceberg-Access-Delegation: vended-credentials`. The
+response advertises the issued session for the table warehouse prefix and for
+the exact current metadata object; the session policy keeps table data access
+inside the warehouse and grants only `GetObject` to that metadata object.
 
-The `rustfs-vended-credentials` profile verifies the client handoff from the
-catalog principal to the table-scoped temporary credentials. It still uses the
-configured principal for setup and REST request signing; the vended credentials
-are first checked against the created table warehouse location, then checked
-with a direct S3 scope probe, and finally applied to PyIceberg S3 data-plane
-access after the table has been created.
+LoadTable remains metadata-only when delegation is absent, vending is disabled,
+or the caller lacks the separate table-credentials permission. Disabled and
+not-authorized fallbacks include an explicit reason. Issuer errors, including
+disallowed chained temporary credentials, are returned as request errors rather
+than silently falling back.
+
+The `rustfs-vended-credentials` profile verifies the client handoff through the
+dedicated credentials endpoint. It still uses the configured principal for
+setup and REST request signing; the vended credentials are checked against the
+created table warehouse location, probed directly against S3 scope boundaries,
+and then applied to PyIceberg data-plane access. Stable PyIceberg releases up to
+0.11 do not consume LoadTable `storage-credentials`; native LoadTable coverage
+requires a client release with that support.
 
 Enablement is server-side and fail-closed: