fix(table-catalog): preserve credential vending after rename (#8400)

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

@@ -25,10 +25,19 @@ impl Operation for RestLoadCredentialsHandler {
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableCredentialsAction).await?;
         ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
-        let store = table_catalog_store_from_extensions(&req.extensions)?;
+        let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
+        let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let issuer = IamTableCredentialIssuer::from_request(&req)?;
-        let response =
-            load_credentials_response(&store, &warehouse, &namespace, &table, &issuer, Some(&principal.credentials)).await?;
+        let response = load_credentials_response(
+            &store,
+            &metadata_backend,
+            &warehouse,
+            &namespace,
+            &table,
+            &issuer,
+            Some(&principal.credentials),
+        )
+        .await?;
         build_sensitive_json_response(StatusCode::OK, &response)
     }
 }

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

@@ -2632,7 +2632,10 @@ fn table_credential_ttl_seconds() -> i64 {
         .unwrap_or(DEFAULT_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS)
 }
 
-fn table_credential_scope(entry: &crate::table_catalog::TableEntry) -> S3Result<TableCredentialScope> {
+fn table_credential_scope(
+    entry: &crate::table_catalog::TableEntry,
+    metadata: &serde_json::Value,
+) -> S3Result<TableCredentialScope> {
     let location = entry
         .warehouse_location
         .strip_prefix("s3://")
@@ -2645,17 +2648,13 @@ fn table_credential_scope(entry: &crate::table_catalog::TableEntry) -> S3Result<
     }
     let object_prefix = normalize_table_credential_object_prefix(object_prefix)?;
     validate_persisted_table_metadata_location(entry, &entry.metadata_location)?;
+    // Rename preserves the metadata pointer; ownership follows its UUID and warehouse, not its old path identifier.
+    validate_persisted_table_metadata(entry, metadata, true)?;
     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"));
-    }
+    crate::table_catalog::Namespace::parse(&entry.namespace).map_err(|_| persisted_metadata_error("table"))?;
+    crate::table_catalog::IdentifierSegment::parse(entry.table.clone()).map_err(|_| persisted_metadata_error("table"))?;
     let metadata_scope_prefix = table_metadata_location_for_client(&entry.table_bucket, &entry.metadata_location);
     Ok(TableCredentialScope {
         warehouse_scope_prefix: format!("s3://{bucket}/{object_prefix}"),
@@ -2920,13 +2919,14 @@ fn add_table_credential_scope_config(config: &mut BTreeMap<String, String>, scop
 
 async fn load_credentials_response_from_entry(
     entry: &crate::table_catalog::TableEntry,
+    metadata: &serde_json::Value,
     issuer: &dyn TableCredentialIssuer,
     principal: Option<&rustfs_credentials::Credentials>,
 ) -> S3Result<RestLoadCredentialsResponse> {
     if !issuer.enabled() {
         return Ok(client_provided_credentials_response(CREDENTIAL_VENDING_DISABLED_REASON));
     }
-    let scope = table_credential_scope(entry)?;
+    let scope = table_credential_scope(entry, metadata)?;
     let request = TableCredentialIssueRequest {
         entry,
         principal,
@@ -2976,7 +2976,7 @@ async fn enrich_load_table_response_with_credentials(
     issuer: &dyn TableCredentialIssuer,
     principal: Option<&rustfs_credentials::Credentials>,
 ) -> S3Result<RestLoadTableResponse> {
-    let credential_response = load_credentials_response_from_entry(entry, issuer, principal).await?;
+    let credential_response = load_credentials_response_from_entry(entry, &response.metadata, issuer, principal).await?;
     Ok(apply_credentials_to_load_table_response(response, credential_response))
 }
 
@@ -6176,6 +6176,7 @@ where
 
 async fn load_credentials_response<S>(
     store: &S,
+    metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend,
     bucket: &str,
     namespace: &crate::table_catalog::Namespace,
     table: &str,
@@ -6192,7 +6193,12 @@ where
     else {
         return Err(iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_TABLE, StatusCode::NOT_FOUND, "table not found"));
     };
-    load_credentials_response_from_entry(&entry, issuer, principal).await
+    let metadata = if issuer.enabled() {
+        read_persisted_table_metadata_for_entry(metadata_backend, &entry, &entry.metadata_location, true).await?
+    } else {
+        serde_json::Value::Null
+    };
+    load_credentials_response_from_entry(&entry, &metadata, issuer, principal).await
 }
 
 async fn get_table_metadata_location_response<S>(

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

@@ -10739,7 +10739,7 @@ fn table_entry_for_credentials() -> crate::table_catalog::TableEntry {
 #[tokio::test]
 async fn disabled_table_credential_issuer_keeps_credentials_empty() {
     let issuer = DisabledTableCredentialIssuer;
-    let response = load_credentials_response_from_entry(&table_entry_for_credentials(), &issuer, None)
+    let response = load_credentials_response_from_entry(&table_entry_for_credentials(), &serde_json::Value::Null, &issuer, None)
         .await
         .expect("disabled issuer should build an empty response");
 
@@ -10764,7 +10764,7 @@ async fn disabled_table_credential_issuer_skips_scope_validation() {
     let mut entry = table_entry_for_credentials();
     entry.warehouse_location = "s3://warehouse/".to_string();
 
-    let response = load_credentials_response_from_entry(&entry, &issuer, None)
+    let response = load_credentials_response_from_entry(&entry, &serde_json::Value::Null, &issuer, None)
         .await
         .expect("disabled issuer should not validate credential scopes");
 
@@ -10792,9 +10792,10 @@ impl TableCredentialIssuer for UnavailableTableCredentialIssuer {
 #[tokio::test]
 async fn unavailable_table_credential_issuer_reports_fallback_scope() {
     let issuer = UnavailableTableCredentialIssuer;
-    let response = load_credentials_response_from_entry(&table_entry_for_credentials(), &issuer, None)
-        .await
-        .expect("unavailable issuer should build a fallback response");
+    let response =
+        load_credentials_response_from_entry(&table_entry_for_credentials(), &table_metadata_for_credentials(), &issuer, None)
+            .await
+            .expect("unavailable issuer should build a fallback response");
 
     assert!(response.storage_credentials.is_empty());
     assert_eq!(
@@ -10821,6 +10822,294 @@ async fn unavailable_table_credential_issuer_reports_fallback_scope() {
 
 struct TestTableCredentialIssuer;
 
+fn table_metadata_for_credentials() -> serde_json::Value {
+    test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id")
+}
+
+#[derive(Default)]
+struct RecordingTableCredentialIssuer {
+    calls: std::sync::atomic::AtomicUsize,
+}
+
+#[async_trait::async_trait]
+impl TableCredentialIssuer for RecordingTableCredentialIssuer {
+    async fn issue_table_credentials(
+        &self,
+        request: TableCredentialIssueRequest<'_>,
+    ) -> S3Result<Option<IssuedTableCredentials>> {
+        self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        assert_eq!(
+            request.scope_prefix,
+            format!("{}/", request.entry.warehouse_location.trim_end_matches('/'))
+        );
+        assert_eq!(
+            request.metadata_object,
+            crate::table_catalog::table_catalog_object_key_from_location(
+                &request.entry.table_bucket,
+                &request.entry.metadata_location,
+            )
+            .expect("metadata must belong to the table bucket")
+        );
+        Ok(Some(IssuedTableCredentials {
+            access_key_id: "temporary-access-key".to_string(),
+            secret_access_key: "temporary-secret-key".to_string(),
+            session_token: "temporary-session-token".to_string(),
+            expiration: OffsetDateTime::from_unix_timestamp(1_800_000_000).expect("test timestamp should be valid"),
+        }))
+    }
+}
+
+#[tokio::test]
+async fn renamed_table_vending_preserves_identity_and_isolates_recreated_metadata_across_backings() {
+    for mode in [
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+        crate::table_catalog::TableCatalogBackingMode::DurableStrong,
+    ] {
+        for destination in ["analytics", "curated"] {
+            let backend = TestTableCatalogObjectBackend::content_addressed();
+            let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(backend.clone(), mode);
+            let source = crate::table_catalog::Namespace::parse("analytics").expect("source namespace");
+            let destination_namespace = crate::table_catalog::Namespace::parse(destination).expect("destination namespace");
+            create_standard_events_table(&store, &backend, &source).await;
+            if destination != "analytics" {
+                create_namespace_response(
+                    &store,
+                    "warehouse",
+                    CreateNamespaceRequest {
+                        namespace: vec![destination.to_string()],
+                        properties: BTreeMap::new(),
+                    },
+                    true,
+                )
+                .await
+                .expect("destination namespace should be created");
+            }
+            let before = store
+                .load_table("warehouse", "analytics", "events")
+                .await
+                .expect("lookup")
+                .expect("table");
+            store
+                .rename_table("warehouse", "analytics", "events", destination, "renamed_events")
+                .await
+                .expect("rename should succeed");
+            let renamed = store
+                .load_table("warehouse", destination, "renamed_events")
+                .await
+                .expect("lookup")
+                .expect("table");
+            assert_eq!(renamed.table_id, before.table_id);
+            assert_eq!(renamed.table_uuid, before.table_uuid);
+            assert_eq!(renamed.warehouse_location, before.warehouse_location);
+            assert_eq!(renamed.metadata_location, before.metadata_location);
+            assert_eq!(renamed.version_token, before.version_token);
+            assert_eq!(renamed.generation, before.generation);
+            let issuer = RecordingTableCredentialIssuer::default();
+            let delegated = load_table_response_with_credentials(
+                &store,
+                &backend,
+                "warehouse",
+                &destination_namespace,
+                "renamed_events",
+                &issuer,
+                None,
+            )
+            .await
+            .expect("renamed table should vend credentials for its retained metadata");
+            let credentials =
+                load_credentials_response(&store, &backend, "warehouse", &destination_namespace, "renamed_events", &issuer, None)
+                    .await
+                    .expect("credentials endpoint should accept the retained metadata");
+            assert_eq!(
+                serde_json::to_value(&credentials.storage_credentials).expect("credential JSON"),
+                serde_json::to_value(&delegated.storage_credentials).expect("delegated JSON")
+            );
+            assert_eq!(delegated.metadata["table-uuid"], before.table_uuid);
+            assert_eq!(delegated.storage_credentials.len(), 2);
+            assert_eq!(delegated.storage_credentials[0].prefix, format!("{}/", before.warehouse_location));
+            assert_eq!(
+                delegated.storage_credentials[1].prefix,
+                table_metadata_location_for_client("warehouse", &before.metadata_location)
+            );
+            let missing = load_table_response_with_credentials(&store, &backend, "warehouse", &source, "events", &issuer, None)
+                .await
+                .expect_err("old identifier must not remain loadable");
+            assert_eq!(missing.status_code(), Some(StatusCode::NOT_FOUND));
+            let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+                "name": "events", "schema": {"type": "struct", "schema-id": 0, "fields": []}
+            }))
+            .expect("recreate request");
+            create_table_response(
+                &store,
+                &TableCommitObjectBackend::trusted(backend.clone()),
+                "warehouse",
+                &source,
+                request,
+                true,
+            )
+            .await
+            .expect("old identifier should be reusable");
+            let recreated = store
+                .load_table("warehouse", "analytics", "events")
+                .await
+                .expect("lookup")
+                .expect("table");
+            assert_ne!(recreated.table_id, renamed.table_id);
+            assert_ne!(recreated.table_uuid, renamed.table_uuid);
+            assert_ne!(recreated.metadata_location, renamed.metadata_location);
+            assert_ne!(recreated.warehouse_location, renamed.warehouse_location);
+            let metadata_object =
+                table_metadata_location_for_catalog("warehouse", &renamed.metadata_location).expect("metadata key");
+            let prefix = renamed
+                .warehouse_location
+                .strip_prefix("s3://warehouse/")
+                .expect("warehouse prefix");
+            let policy =
+                table_credential_session_policy(&renamed, &format!("{prefix}/"), &metadata_object).expect("session policy");
+            let conditions = HashMap::new();
+            let claims = HashMap::new();
+            for (object, allowed) in [
+                (metadata_object.clone(), true),
+                (format!("{metadata_object}.neighbor"), false),
+                (
+                    table_metadata_location_for_catalog("warehouse", &recreated.metadata_location).expect("recreated key"),
+                    false,
+                ),
+            ] {
+                assert_eq!(
+                    policy
+                        .is_allowed(&rustfs_policy::policy::Args {
+                            account: "temporary-access-key",
+                            groups: &None,
+                            action: Action::S3Action(rustfs_policy::policy::action::S3Action::GetObjectAction),
+                            bucket: "warehouse",
+                            conditions: &conditions,
+                            is_owner: false,
+                            object: &object,
+                            claims: &claims,
+                            deny_only: false,
+                        })
+                        .await,
+                    allowed,
+                    "{mode:?} {destination} {object}"
+                );
+            }
+            let reloaded = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(backend.clone(), mode);
+            for (namespace, name, expected) in [
+                (&destination_namespace, "renamed_events", &renamed),
+                (&source, "events", &recreated),
+            ] {
+                let response =
+                    load_table_response_with_credentials(&reloaded, &backend, "warehouse", namespace, name, &issuer, None)
+                        .await
+                        .expect("fresh store should vend for the correct persisted identity");
+                assert_eq!(response.metadata["table-uuid"], expected.table_uuid);
+                assert_eq!(response.storage_credentials[0].prefix, format!("{}/", expected.warehouse_location));
+                assert_eq!(
+                    response.storage_credentials[1].prefix,
+                    table_metadata_location_for_client("warehouse", &expected.metadata_location)
+                );
+                let credentials = load_credentials_response(&reloaded, &backend, "warehouse", namespace, name, &issuer, None)
+                    .await
+                    .expect("fresh store credentials endpoint should bind the persisted identity");
+                assert_eq!(
+                    serde_json::to_value(&credentials.storage_credentials).expect("credential JSON"),
+                    serde_json::to_value(&response.storage_credentials).expect("delegated JSON")
+                );
+            }
+        }
+    }
+}
+
+#[tokio::test]
+async fn credential_vending_rejects_corrupt_or_foreign_metadata_before_issuance_across_backings() {
+    for mode in [
+        crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+        crate::table_catalog::TableCatalogBackingMode::DurableStrong,
+    ] {
+        let backend = TestTableCatalogObjectBackend::content_addressed();
+        let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(backend.clone(), mode);
+        let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace");
+        let created = create_standard_events_table(&store, &backend, &namespace).await;
+        let before = store
+            .load_table("warehouse", "analytics", "events")
+            .await
+            .expect("lookup")
+            .expect("table");
+        let issuer = RecordingTableCredentialIssuer::default();
+        let mut foreign_uuid = created.metadata.clone();
+        foreign_uuid["table-uuid"] = serde_json::Value::from("foreign-table-uuid");
+        let mut foreign_warehouse = created.metadata.clone();
+        foreign_warehouse["location"] = serde_json::Value::from("s3://warehouse/tables/another-table");
+        let mut cross_bucket = created.metadata.clone();
+        cross_bucket["location"] = serde_json::Value::from("s3://other/tables/another-table");
+        let mut missing_uuid = created.metadata.clone();
+        missing_uuid.as_object_mut().expect("metadata object").remove("table-uuid");
+        let mut unsupported_format = created.metadata.clone();
+        unsupported_format["format-version"] = serde_json::Value::from(99);
+        for metadata in [
+            foreign_uuid,
+            foreign_warehouse,
+            cross_bucket,
+            missing_uuid,
+            unsupported_format,
+            serde_json::Value::Null,
+        ] {
+            backend.put_json("warehouse", &before.metadata_location, metadata).await;
+            let credentials_error = load_credentials_response(&store, &backend, "warehouse", &namespace, "events", &issuer, None)
+                .await
+                .expect_err("invalid persisted metadata must not produce credentials");
+            let delegated_error =
+                load_table_response_with_credentials(&store, &backend, "warehouse", &namespace, "events", &issuer, None)
+                    .await
+                    .expect_err("delegated load must fail closed");
+            for error in [credentials_error, delegated_error] {
+                assert_eq!(error.status_code(), Some(StatusCode::INTERNAL_SERVER_ERROR));
+                assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_REST.into()));
+                assert_eq!(error.message(), Some("persisted table metadata is invalid"));
+            }
+            assert_eq!(issuer.calls.load(std::sync::atomic::Ordering::SeqCst), 0);
+            let after = store
+                .load_table("warehouse", "analytics", "events")
+                .await
+                .expect("lookup")
+                .expect("table");
+            assert_eq!(after.table_uuid, before.table_uuid);
+            assert_eq!(after.metadata_location, before.metadata_location);
+            assert_eq!(after.version_token, before.version_token);
+            assert_eq!(after.generation, before.generation);
+        }
+        let disabled =
+            load_credentials_response(&store, &backend, "warehouse", &namespace, "events", &DisabledTableCredentialIssuer, None)
+                .await
+                .expect("disabled vending should not read or validate metadata");
+        assert!(disabled.storage_credentials.is_empty());
+        backend
+            .put_json("warehouse", &before.metadata_location, created.metadata)
+            .await;
+        backend.fail_next_read("warehouse", &before.metadata_location).await;
+        let read_error = load_credentials_response(&store, &backend, "warehouse", &namespace, "events", &issuer, None)
+            .await
+            .expect_err("metadata read failure must prevent issuance");
+        assert_eq!(read_error.status_code(), Some(StatusCode::INTERNAL_SERVER_ERROR));
+        assert_eq!(read_error.message(), Some("persisted table metadata is invalid"));
+        backend
+            .delete_object("warehouse", &before.metadata_location)
+            .await
+            .expect("remove metadata fixture");
+        let missing_metadata = load_credentials_response(&store, &backend, "warehouse", &namespace, "events", &issuer, None)
+            .await
+            .expect_err("missing metadata must prevent issuance");
+        assert_eq!(missing_metadata.status_code(), Some(StatusCode::INTERNAL_SERVER_ERROR));
+        assert_eq!(issuer.calls.load(std::sync::atomic::Ordering::SeqCst), 0);
+        let missing =
+            load_credentials_response(&store, &backend, "warehouse", &namespace, "missing", &DisabledTableCredentialIssuer, None)
+                .await
+                .expect_err("disabled vending must still reject missing tables");
+        assert_eq!(missing.status_code(), Some(StatusCode::NOT_FOUND));
+    }
+}
+
 #[async_trait::async_trait]
 impl TableCredentialIssuer for TestTableCredentialIssuer {
     async fn issue_table_credentials(
@@ -10852,9 +11141,14 @@ async fn credential_issuer_returns_temporary_scoped_storage_credentials() {
         ..Default::default()
     };
 
-    let response = load_credentials_response_from_entry(&table_entry_for_credentials(), &issuer, Some(&principal))
-        .await
-        .expect("issuer should build a scoped credential response");
+    let response = load_credentials_response_from_entry(
+        &table_entry_for_credentials(),
+        &table_metadata_for_credentials(),
+        &issuer,
+        Some(&principal),
+    )
+    .await
+    .expect("issuer should build a scoped credential response");
 
     assert_eq!(
         response.config.get(CREDENTIAL_VENDING_CONFIG_KEY),
@@ -10893,11 +11187,7 @@ async fn credential_issuer_returns_temporary_scoped_storage_credentials() {
 #[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 metadata = table_metadata_for_credentials();
     let principal = rustfs_credentials::Credentials {
         access_key: "parent-access-key".to_string(),
         secret_key: "parent-secret-key".to_string(),
@@ -10911,9 +11201,14 @@ async fn load_table_uses_the_shared_credential_vending_result() {
     )
     .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");
+    let credentials = load_credentials_response_from_entry(
+        &entry,
+        &table_metadata_for_credentials(),
+        &TestTableCredentialIssuer,
+        Some(&principal),
+    )
+    .await
+    .expect("credentials endpoint should include vended credentials");
 
     assert_eq!(
         load_table.config.get(CREDENTIAL_VENDING_CONFIG_KEY),
@@ -10952,11 +11247,12 @@ impl TableCredentialIssuer for RefusingTableCredentialIssuer {
 #[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 credentials_error =
+        load_credentials_response_from_entry(&entry, &table_metadata_for_credentials(), &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!({})),
+        load_table_response_from_entry(entry.clone(), table_metadata_for_credentials()),
         &entry,
         &RefusingTableCredentialIssuer,
         None,
@@ -10965,6 +11261,7 @@ async fn load_table_and_credentials_endpoint_propagate_issuer_refusal() {
     .expect_err("load table should propagate issuer refusal");
 
     assert_eq!(load_table_error.code(), credentials_error.code());
+    assert_eq!(credentials_error.code(), &S3ErrorCode::AccessDenied);
     assert_eq!(load_table_error.status_code(), credentials_error.status_code());
     assert_eq!(load_table_error.message(), credentials_error.message());
 }
@@ -10972,9 +11269,10 @@ async fn load_table_and_credentials_endpoint_propagate_issuer_refusal() {
 #[tokio::test]
 async fn credential_response_serializes_sensitive_config_only_inside_storage_credentials() {
     let issuer = TestTableCredentialIssuer;
-    let response = load_credentials_response_from_entry(&table_entry_for_credentials(), &issuer, None)
-        .await
-        .expect("issuer should build a scoped credential response");
+    let response =
+        load_credentials_response_from_entry(&table_entry_for_credentials(), &table_metadata_for_credentials(), &issuer, None)
+            .await
+            .expect("issuer should build a scoped credential response");
 
     let value = serde_json::to_value(&response).expect("credential response should serialize");
 
@@ -11006,9 +11304,14 @@ fn credential_http_response_disables_caching() {
 
 #[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 response = load_credentials_response_from_entry(
+        &table_entry_for_credentials(),
+        &table_metadata_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"));
@@ -11241,24 +11544,25 @@ async fn table_credential_session_policy_includes_table_resource_actions() {
 fn table_credential_scope_rejects_cross_bucket_or_unsafe_prefix() {
     let mut entry = table_entry_for_credentials();
     entry.warehouse_location = "s3://other-warehouse/tables/table-id".to_string();
-    assert!(table_credential_scope(&entry).is_err());
+    assert!(table_credential_scope(&entry, &table_metadata_for_credentials()).is_err());
 
     let mut entry = table_entry_for_credentials();
     entry.warehouse_location = "s3://warehouse/tables/../table-id".to_string();
-    assert!(table_credential_scope(&entry).is_err());
+    assert!(table_credential_scope(&entry, &table_metadata_for_credentials()).is_err());
 
     let mut entry = table_entry_for_credentials();
     entry.warehouse_location = "s3://warehouse/.rustfs-table".to_string();
-    assert!(table_credential_scope(&entry).is_err());
+    assert!(table_credential_scope(&entry, &table_metadata_for_credentials()).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());
+    assert!(table_credential_scope(&entry, &table_metadata_for_credentials()).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());
+    let foreign_metadata = test_table_metadata_json("orders-table-uuid", &entry.warehouse_location);
+    assert!(table_credential_scope(&entry, &foreign_metadata).is_err());
 }
 
 #[test]
@@ -11266,7 +11570,8 @@ 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");
+    let scope = table_credential_scope(&entry, &table_metadata_for_credentials())
+        .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");