From e4057b6746b68e04f57ab2e87901ff6bcfaecfeb Mon Sep 17 00:00:00 2001
From: Henry Guo <marshawcoco@gmail.com>
Date: Sat, 7 Nov 2026 08:07:58 +0800
Subject: [PATCH] fix(table-catalog): reject storage credentials for catalog
 access (#9185)

* fix(table-catalog): reject storage credentials for catalog access

* fix(table-catalog): avoid growing the s3s error macro footprint

Use the existing S3Error constructor for the storage-credential rejection
while preserving its AccessDenied code and message. Keep the merged PR
within the current s3s footprint baseline without relaxing the guard.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: Hauser Zhi <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
Co-authored-by: Chris An <anzhengchao@gmail.com>
---
 docs/architecture/s3-tables-support-matrix.md |   2 +-
 .../src/admin/handlers/table_catalog/mod.rs   |  10 ++
 .../src/admin/handlers/table_catalog/tests.rs | 137 ++++++++++++++++++
 3 files changed, 148 insertions(+), 1 deletion(-)

diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md
index bb3e7391e..b5cd06f1d 100644
--- a/docs/architecture/s3-tables-support-matrix.md
+++ b/docs/architecture/s3-tables-support-matrix.md
@@ -67,7 +67,7 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t
 | Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. Object-backed recovery reports a staged request as SUPERSEDED only when a reachable finalized same-base winner proves it lost CAS and its idempotency binding is intact. The original staged receipt is retained; its old request still conflicts. The winning history must be retained with the losing receipt; missing or contradictory history requires manual review. |
 | Snapshot refs | Supported | List, create/replace, delete via commits; `main` is protected; refs with explicit retention need forced delete. |
 | Iceberg views | Supported | Create, list, load, replace, exists, drop with view-scoped authorization; only view format version 1. |
-| LoadTable and table credentials endpoint | Supported | Vending only on negotiated `vended-credentials`; one temporary session scoped to the warehouse prefix and current metadata location; missing credential permission falls back to metadata-only with an explicit reason; responses carry `Cache-Control: no-store, private`. |
+| LoadTable and table credentials endpoint | Supported | Vending only on negotiated `vended-credentials`; one temporary session scoped to the warehouse prefix and current metadata location; missing credential permission falls back to metadata-only with an explicit reason; vended storage sessions cannot authenticate Catalog requests; responses carry `Cache-Control: no-store, private`. |
 | Catalog diagnostics and export | Supported | Object-backed and durable-strong stores expose recovery state, consistency, backing manifest, WAL state, migration target, single-active-writer policy, and scale validation matrix. Durable-strong responses also identify the snapshot path, ETag, and format version; diagnostics re-check the snapshot observation before returning and use zero-file retention until strong maintenance configuration is persisted. |
 | Durable catalog backup and restore | Preview / controlled | Durable-strong only. `POST /iceberg/v1/{warehouse}/catalog/backup` seals a table-bucket snapshot behind the publication and migration fences, records catalog/object watermarks, and stores an immutable backup artifact. `POST /iceberg/v1/{warehouse}/catalog/restore` verifies those watermarks, requires an explicit replacement decision for an existing target, applies the catalog snapshot with ETag-CAS, and leaves a recoverable restore intent if maintenance-state finalization fails. It does not copy data files or claim automatic cross-region failover. |
 | Catalog import and rollback | Supported | Import/register and online rollback go through validation and commit paths. Online rollback accepts only forward-safe targets; restoring an older target that lowers watermarks is an offline disaster-recovery operation with all writers stopped. |
diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs
index 8bb617d4a..87fc3e5f1 100644
--- a/rustfs/src/admin/handlers/table_catalog/mod.rs
+++ b/rustfs/src/admin/handlers/table_catalog/mod.rs
@@ -1326,6 +1326,16 @@ async fn table_catalog_request_principal(req: &S3Request<Body>) -> S3Result<Tabl
         Some(context.as_ref()),
     )
     .await?;
+    // Storage credentials are a data-plane delegation, never a catalog identity.
+    if ["rustfs:table-bucket", "rustfs:table-id", "rustfs:credential-scope-prefix"]
+        .iter()
+        .any(|claim| credentials.claims_or_empty().contains_key(*claim))
+    {
+        return Err(S3Error::with_message(
+            S3ErrorCode::AccessDenied,
+            "table-scoped storage credentials cannot access the catalog control plane",
+        ));
+    }
     let iam = context.iam();
     if !iam.is_ready() {
         return Err(table_catalog_internal_error("iam not init"));
diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs
index 1ebe0d8d1..77084b63b 100644
--- a/rustfs/src/admin/handlers/table_catalog/tests.rs
+++ b/rustfs/src/admin/handlers/table_catalog/tests.rs
@@ -323,6 +323,119 @@ async fn table_catalog_authentication_and_credentials_use_the_request_context()
     assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
 }
 
+#[tokio::test]
+async fn table_storage_credentials_are_rejected_by_catalog_authentication() {
+    let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    rustfs_iam::store::object::ObjectStore::new(store.clone())
+        .save_iam_config(
+            serde_json::json!({"version": 1}),
+            format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+        )
+        .await
+        .expect("seed request IAM format");
+    let iam = rustfs_iam::build_iam_sys(store.clone())
+        .await
+        .expect("initialize request IAM");
+    let parent_key = "storage-session-parent";
+    let parent_secret = "storage-session-parent-secret";
+    iam.create_user(
+        parent_key,
+        &AddOrUpdateUserReq {
+            secret_key: parent_secret.to_string(),
+            policy: None,
+            status: AccountStatus::Enabled,
+        },
+    )
+    .await
+    .expect("create session parent");
+    iam.set_policy(
+        "storage-session-parent-policy",
+        Policy::parse_config(br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"admin:GetTable"}]}"#)
+            .expect("parse catalog read policy"),
+    )
+    .await
+    .expect("store parent policy");
+    iam.policy_db_set(parent_key, UserType::Reg, false, "storage-session-parent-policy")
+        .await
+        .expect("attach parent policy");
+    let root_key = "storage-session-root";
+    let root_secret = "storage-session-root-secret";
+    let context = Arc::new(AppContext::new(store, Arc::new(RequestIam { handle: iam.clone() }), Arc::new(RequestKms)));
+    assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
+        access_key: root_key.to_string(),
+        secret_key: root_secret.to_string(),
+        status: "on".to_string(),
+        ..Default::default()
+    }));
+    for (key, secret) in [(root_key, root_secret), (parent_key, parent_secret)] {
+        authorize_table_catalog_request(
+            &table_catalog_handler_request(context.clone(), key, secret),
+            AdminAction::GetTableAction,
+        )
+        .await
+        .expect("ordinary catalog identities remain authorized");
+    }
+
+    for parent in [root_key, parent_key] {
+        let mut cases = vec![(None, serde_json::Value::Null)];
+        for marker in ["rustfs:table-bucket", "rustfs:table-id", "rustfs:credential-scope-prefix"] {
+            for value in [
+                serde_json::json!("table-scope"),
+                serde_json::json!(""),
+                serde_json::Value::Null,
+                serde_json::json!(7),
+                serde_json::json!([]),
+                serde_json::json!({}),
+            ] {
+                cases.push((Some(marker), value));
+            }
+        }
+        for (marker, value) in cases {
+            let expiration = OffsetDateTime::now_utc().saturating_add(Duration::minutes(15));
+            let mut claims = HashMap::from([
+                ("exp".to_string(), serde_json::json!(expiration.unix_timestamp())),
+                ("parent".to_string(), serde_json::json!(parent)),
+            ]);
+            if let Some(marker) = marker {
+                claims.insert(marker.to_string(), value);
+            }
+            let mut credential = get_new_credentials_with_metadata(&claims, root_secret).expect("sign a temporary credential");
+            credential.parent_user = parent.to_string();
+            iam.set_temp_user(&credential.access_key, &credential, None)
+                .await
+                .expect("store temporary credential");
+            let mut request = table_catalog_handler_request(context.clone(), &credential.access_key, &credential.secret_key);
+            request.headers.insert(
+                "x-amz-security-token",
+                HeaderValue::from_str(&credential.session_token).expect("session token header"),
+            );
+            let unscoped = authorize_table_catalog_request(&request, AdminAction::GetTableAction).await;
+            let scoped = authorize_table_catalog_resource_request(
+                &request,
+                &TableCatalogResource::warehouse("warehouse"),
+                AdminAction::GetTableAction,
+            )
+            .await;
+            if marker.is_some() {
+                assert_eq!(
+                    unscoped.expect_err("any storage marker must deny catalog access").code(),
+                    &S3ErrorCode::AccessDenied
+                );
+                assert_eq!(
+                    scoped
+                        .err()
+                        .expect("partial or malformed storage claims must fail closed")
+                        .code(),
+                    &S3ErrorCode::AccessDenied
+                );
+            } else {
+                unscoped.expect("ordinary temporary credentials retain catalog authorization");
+                scoped.expect("ordinary temporary credentials retain resource authorization");
+            }
+        }
+    }
+}
+
 #[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;
@@ -1182,6 +1295,30 @@ async fn load_table_handler_negotiates_vended_credentials_without_breaking_metad
             );
         }
     }
+    let delegated = &issued_json["storage-credentials"][0]["config"];
+    let mut storage_request = table_catalog_handler_request(
+        context.clone(),
+        delegated[S3_ACCESS_KEY_ID_CONFIG_KEY].as_str().expect("vended access key"),
+        delegated[S3_SECRET_ACCESS_KEY_CONFIG_KEY]
+            .as_str()
+            .expect("vended secret key"),
+    );
+    storage_request.headers.insert(
+        "x-amz-security-token",
+        HeaderValue::from_str(delegated[S3_SESSION_TOKEN_CONFIG_KEY].as_str().expect("vended session token"))
+            .expect("session token header"),
+    );
+    let error = call_load_table_handler(storage_request)
+        .await
+        .expect_err("vended storage credentials must not enter the catalog handler");
+    assert_eq!(error.code(), &S3ErrorCode::AccessDenied);
+    let reloaded = call_load_table_handler(table_catalog_handler_request(context, vended_access_key, vended_secret_key))
+        .await
+        .expect("the parent catalog identity remains authorized");
+    let reloaded_json: serde_json::Value =
+        serde_json::from_slice(&reloaded.output.1.bytes().expect("reloaded table body")).expect("parse reloaded table");
+    assert_eq!(reloaded_json["metadata-location"], issued_json["metadata-location"]);
+    assert_eq!(reloaded_json["metadata"], issued_json["metadata"]);
 }
 
 #[test]