diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 063fd3a41..1dfd412a3 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/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. | diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs index 5a8e94033..d958c8c9f 100644 --- a/rustfs/src/admin/handlers/table_catalog/mod.rs +++ b/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, } -#[derive(Debug, Serialize)] +#[derive(Serialize)] struct RestStorageCredential { prefix: String, config: BTreeMap, } +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 { +fn table_credential_session_policy( + entry: &crate::table_catalog::TableEntry, + object_prefix: &str, + metadata_object: &str, +) -> S3Result { 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 { 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 { + 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, 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 { 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 { + 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( namespace: &crate::table_catalog::Namespace, table: &str, ) -> S3Result +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( + 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 +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( + 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( diff --git a/rustfs/src/admin/handlers/table_catalog/table.rs b/rustfs/src/admin/handlers/table_catalog/table.rs index b98b2fbfd..95e1773be 100644 --- a/rustfs/src/admin/handlers/table_catalog/table.rs +++ b/rustfs/src/admin/handlers/table_catalog/table.rs @@ -121,14 +121,56 @@ impl Operation for RestLoadTableHandler { let namespace = namespace_from_params(¶ms)?; let table = table_name_from_params(¶ms)?; 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) + } } } diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index 05a70d0f2..7e1520b87 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/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, access_key: &str, secret_key: &str) -> S3Request { + 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, values: &[&str]) -> S3Request { + 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) -> S3Result> { + 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::(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( + ¬_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> { + 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] diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 771a9ea40..2a11d17e8 100644 --- a/scripts/table-catalog/README.md +++ b/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: