diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index c05b236bd..28bd45bde 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=3a7f52dd884ed269746029e983006ce69b62f260db2ad1acdcffdb2196c65d11 -sha256-linux=4af4bb607919b96e63e2a3fb7a78bd18fcb08e265f4ca4ce6afb09a8be798305 +sha256-darwin=331393a79b7c1fbccd2a6f6324361c2aa2ab60c818176ef56fbf7fa15bfc5af0 +sha256-linux=f85f0021153264f00eba8fa45384678f5ae396d7d74995c789dbbca9cd6d7ea1 diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 939544ed7..426bba873 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -189,6 +189,9 @@ mod conditional_delete_objects_test; #[cfg(test)] mod delete_authorization_test; +#[cfg(test)] +mod table_location_authorization_test; + // Regression test for signed DELETE Object?versionId requests without Content-Length. #[cfg(test)] mod delete_object_no_content_length_test; diff --git a/crates/e2e_test/src/table_location_authorization_test.rs b/crates/e2e_test/src/table_location_authorization_test.rs new file mode 100644 index 000000000..f2d8b5a23 --- /dev/null +++ b/crates/e2e_test/src/table_location_authorization_test.rs @@ -0,0 +1,417 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::common::{ + AdminTransport, RustFSTestEnvironment, admin_add_canned_policy_via, admin_attach_user_policy_via, admin_create_user, + init_logging, signed_request, +}; +use aws_sdk_s3::primitives::ByteStream; +use http::{Method, StatusCode}; +use serde_json::{Value, json}; +use uuid::Uuid; + +type TestResult = Result>; + +async fn set_policy(env: &RustFSTestEnvironment, user: &str, policy: Value) -> TestResult { + admin_add_canned_policy_via( + AdminTransport::Signed, + &env.url, + &env.access_key, + &env.secret_key, + user, + &policy.to_string(), + ) + .await?; + admin_attach_user_policy_via(AdminTransport::Signed, &env.url, &env.access_key, &env.secret_key, user, user).await +} + +async fn catalog_request( + env: &RustFSTestEnvironment, + credentials: &(String, String), + method: Method, + path: &str, + body: Option, + expected: StatusCode, +) -> TestResult { + let response = signed_request( + method.clone(), + &format!("{}{path}", env.url), + &credentials.0, + &credentials.1, + body.map(|value| serde_json::to_vec(&value)).transpose()?, + Some("application/json"), + ) + .await?; + let status = response.status(); + let text = response.text().await?; + assert_eq!(status, expected, "{method} {path}: {text}"); + if text.is_empty() || !status.is_success() { + Ok(Value::Null) + } else { + Ok(serde_json::from_str(&text)?) + } +} + +async fn get_object( + env: &RustFSTestEnvironment, + credentials: &(String, String), + bucket: &str, + key: &str, + expected: StatusCode, +) -> TestResult { + let response = signed_request( + Method::GET, + &format!("{}/{bucket}/{key}", env.url), + &credentials.0, + &credentials.1, + None, + None, + ) + .await?; + assert_eq!(response.status(), expected, "GET {bucket}/{key}"); + if expected == StatusCode::OK { + assert_eq!(response.bytes().await?.as_ref(), b"original object"); + } + Ok(()) +} + +fn create_table(name: &str) -> Value { + json!({ + "name": name, + "schema": { + "type": "struct", + "fields": [{"id": 1, "name": "id", "required": false, "type": "long"}] + }, + "stage-create": false + }) +} + +fn staged_commit(metadata: &Value, location: &str) -> Value { + json!({ + "requirements": [{"type": "assert-create"}], + "updates": [ + {"action": "assign-uuid", "uuid": metadata["table-uuid"]}, + {"action": "upgrade-format-version", "format-version": metadata["format-version"]}, + {"action": "add-schema", "schema": metadata["schemas"][0]}, + {"action": "set-current-schema", "schema-id": -1}, + {"action": "add-spec", "spec": metadata["partition-specs"][0]}, + {"action": "set-default-spec", "spec-id": -1}, + {"action": "add-sort-order", "sort-order": metadata["sort-orders"][0]}, + {"action": "set-default-sort-order", "sort-order-id": -1}, + {"action": "set-location", "location": location} + ] + }) +} + +async fn location_claim_preserves_other_users_objects_after_drop_and_restart(backing: &str) -> TestResult { + init_logging(); + let mut env = RustFSTestEnvironment::new().await?; + let server_env = [("RUSTFS_TABLE_CATALOG_BACKING", backing)]; + env.start_rustfs_server_without_cleanup_with_env(&[("RUSTFS_TABLE_CATALOG_BACKING", "object")]) + .await?; + env.wait_for_server_ready().await?; + let root = (env.access_key.clone(), env.secret_key.clone()); + let creator = (format!("creator-{}", Uuid::new_v4().simple()), Uuid::new_v4().simple().to_string()); + let reader = (format!("reader-{}", Uuid::new_v4().simple()), Uuid::new_v4().simple().to_string()); + let bucket = "table-location-authorization"; + let root_s3 = env.create_s3_client(); + root_s3.create_bucket().bucket(bucket).send().await?; + for key in ["shared/known.txt", "shared2/known.txt", "authorized/known.txt"] { + root_s3 + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from_static(b"original object")) + .send() + .await?; + } + let catalog = format!("/iceberg/v1/{bucket}/namespaces/lab/tables"); + catalog_request(&env, &root, Method::PUT, &format!("/iceberg/v1/buckets/{bucket}"), None, StatusCode::OK).await?; + catalog_request( + &env, + &root, + Method::POST, + &format!("/iceberg/v1/{bucket}/namespaces"), + Some(json!({"namespace": ["lab"]})), + StatusCode::OK, + ) + .await?; + if backing == "durable-strong" { + catalog_request( + &env, + &root, + Method::POST, + &format!("/iceberg/v1/{bucket}/catalog/warehouse-index/backfill"), + None, + StatusCode::NO_CONTENT, + ) + .await?; + let preflight = catalog_request( + &env, + &root, + Method::GET, + &format!("/iceberg/v1/{bucket}/catalog/migration"), + None, + StatusCode::OK, + ) + .await?; + assert_eq!(preflight["blockers"], json!([]), "{preflight}"); + let migration = catalog_request( + &env, + &root, + Method::POST, + &format!("/iceberg/v1/{bucket}/catalog/migration"), + None, + StatusCode::OK, + ) + .await?; + assert_eq!(migration["ready_to_enable_durable_strong"], json!(true), "{migration}"); + env.restart_server_preserving_data(vec![], &server_env).await?; + env.wait_for_server_ready().await?; + } + admin_create_user(&env, &creator.0, &creator.1).await?; + admin_create_user(&env, &reader.0, &reader.1).await?; + let create_statement = json!({ + "Effect": "Allow", + "Action": "admin:CreateTable", + "Resource": format!("arn:aws:s3:::{bucket}/namespaces/lab") + }); + set_policy( + &env, + &creator.0, + json!({"Version": "2012-10-17", "Statement": [create_statement.clone()]}), + ) + .await?; + set_policy( + &env, + &reader.0, + json!({"Version": "2012-10-17", "Statement": [{ + "Effect": "Allow", "Action": "s3:GetObject", "Resource": format!("arn:aws:s3:::{bucket}/*") + }]}), + ) + .await?; + for key in ["shared/known.txt", "shared2/known.txt"] { + get_object(&env, &reader, bucket, key, StatusCode::OK).await?; + get_object(&env, &creator, bucket, key, StatusCode::FORBIDDEN).await?; + } + for (method, path) in [ + (Method::PUT, format!("/{bucket}/shared/new.txt")), + (Method::GET, format!("/{bucket}?list-type=2")), + ] { + let response = signed_request(method, &format!("{}{path}", env.url), &creator.0, &creator.1, None, None).await?; + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + // Both direct and staged requests must reject a caller-selected foreign prefix. + for staged in [false, true] { + for location in [format!("s3://{bucket}/shared"), format!("s3://{bucket}/shared/")] { + let mut request = create_table("claim"); + request["location"] = json!(location); + request["stage-create"] = json!(staged); + catalog_request(&env, &creator, Method::POST, &catalog, Some(request), StatusCode::FORBIDDEN).await?; + } + } + catalog_request(&env, &root, Method::GET, &format!("{catalog}/claim"), None, StatusCode::NOT_FOUND).await?; + + // Server-assigned locations still need only CreateTable, and retain protection after drop. + let created = catalog_request(&env, &creator, Method::POST, &catalog, Some(create_table("managed")), StatusCode::OK).await?; + let location = created["metadata"]["location"].as_str().ok_or("missing table location")?; + let managed_key = format!( + "{}/known.txt", + location + .strip_prefix(&format!("s3://{bucket}/")) + .ok_or("table location has wrong bucket")? + ); + root_s3 + .put_object() + .bucket(bucket) + .key(&managed_key) + .body(ByteStream::from_static(b"original object")) + .send() + .await?; + get_object(&env, &reader, bucket, &managed_key, StatusCode::FORBIDDEN).await?; + catalog_request(&env, &root, Method::DELETE, &format!("{catalog}/managed"), None, StatusCode::NO_CONTENT).await?; + env.restart_server_preserving_data(vec![], &server_env).await?; + env.wait_for_server_ready().await?; + for table in ["claim", "managed"] { + catalog_request(&env, &root, Method::GET, &format!("{catalog}/{table}"), None, StatusCode::NOT_FOUND).await?; + } + for key in ["shared/known.txt", "shared2/known.txt", "authorized/known.txt"] { + get_object(&env, &reader, bucket, key, StatusCode::OK).await?; + } + get_object(&env, &reader, bucket, &managed_key, StatusCode::FORBIDDEN).await?; + get_object(&env, &root, bucket, &managed_key, StatusCode::OK).await?; + + // Explicit adoption is scoped to a physical prefix and does not grant S3 access. + set_policy( + &env, + &creator.0, + json!({"Version": "2012-10-17", "Statement": [ + create_statement.clone(), + {"Effect": "Allow", "Action": "admin:ClaimTableLocation", "Resource": format!("arn:aws:s3:::{bucket}/authorized")} + ]}), + ) + .await?; + let mut request = create_table("authorized"); + request["location"] = json!(format!("s3://{bucket}/shared")); + catalog_request(&env, &creator, Method::POST, &catalog, Some(request.clone()), StatusCode::FORBIDDEN).await?; + request["location"] = json!(format!("s3://{bucket}/authorized")); + let adopted = catalog_request(&env, &creator, Method::POST, &catalog, Some(request), StatusCode::OK).await?; + get_object(&env, &creator, bucket, "authorized/known.txt", StatusCode::FORBIDDEN).await?; + get_object(&env, &reader, bucket, "authorized/known.txt", StatusCode::FORBIDDEN).await?; + // Existing updates retain their permissions; relocation and re-registration + // require a prefix grant even when the caller can access every S3 object. + set_policy( + &env, + &creator.0, + json!({"Version": "2012-10-17", "Statement": [ + create_statement, + {"Effect": "Allow", "Action": "admin:RegisterTable", "Resource": [ + format!("arn:aws:s3:::{bucket}/namespaces/lab"), + format!("arn:aws:s3:::{bucket}/namespaces/lab/tables/*") + ]}, + {"Effect": "Allow", "Action": "admin:CommitTable", "Resource": format!("arn:aws:s3:::{bucket}/namespaces/lab/tables/*")}, + {"Effect": "Allow", "Action": "admin:SetTableMetadataLocation", "Resource": format!("arn:aws:s3:::{bucket}/namespaces/lab/tables/authorized")}, + {"Effect": "Allow", "Action": ["s3:GetObject", "s3:PutObject"], "Resource": format!("arn:aws:s3:::{bucket}/*")} + ]}), + ) + .await?; + let requirements = json!([{"type": "assert-table-uuid", "uuid": adopted["metadata"]["table-uuid"]}]); + let committed = catalog_request( + &env, + &creator, + Method::POST, + &format!("{catalog}/authorized"), + Some(json!({ + "requirements": requirements, + "updates": [{"action": "set-properties", "updates": {"test.location-authorization": "retained"}}] + })), + StatusCode::OK, + ) + .await?; + catalog_request( + &env, + &creator, + Method::POST, + &format!("{catalog}/authorized"), + Some(json!({ + "requirements": requirements, + "updates": [{"action": "set-location", "location": format!("s3://{bucket}/shared")}] + })), + StatusCode::FORBIDDEN, + ) + .await?; + let current = catalog_request(&env, &root, Method::GET, &format!("{catalog}/authorized"), None, StatusCode::OK).await?; + assert_eq!(current["metadata"]["location"], json!(format!("s3://{bucket}/authorized"))); + assert_eq!(current["metadata-location"], committed["metadata-location"]); + catalog_request( + &env, + &root, + Method::DELETE, + &format!("{catalog}/authorized"), + None, + StatusCode::NO_CONTENT, + ) + .await?; + catalog_request( + &env, + &creator, + Method::POST, + &format!("/iceberg/v1/{bucket}/namespaces/lab/register"), + Some(json!({"name": "authorized", "metadata-location": committed["metadata-location"]})), + StatusCode::FORBIDDEN, + ) + .await?; + catalog_request( + &env, + &creator, + Method::POST, + &format!("{catalog}/authorized/catalog/import"), + Some(json!({"metadata-location": committed["metadata-location"]})), + StatusCode::FORBIDDEN, + ) + .await?; + if backing == "object" { + catalog_request( + &env, + &creator, + Method::POST, + &format!("{catalog}/authorized/catalog/external/sync"), + Some(json!({ + "catalog": "polaris", + "external-namespace": "lab", + "external-table": "authorized", + "metadata-location": committed["metadata-location"] + })), + StatusCode::FORBIDDEN, + ) + .await?; + } + + // Staged metadata is client-controlled and must be authorized at publication. + let mut request = create_table("staged"); + request["stage-create"] = json!(true); + let staged = catalog_request(&env, &creator, Method::POST, &catalog, Some(request), StatusCode::OK).await?; + catalog_request( + &env, + &creator, + Method::POST, + &format!("{catalog}/staged"), + Some(staged_commit(&staged["metadata"], &format!("s3://{bucket}/shared"))), + StatusCode::FORBIDDEN, + ) + .await?; + catalog_request(&env, &root, Method::GET, &format!("{catalog}/staged"), None, StatusCode::NOT_FOUND).await?; + set_policy( + &env, + &creator.0, + json!({"Version": "2012-10-17", "Statement": [ + {"Effect": "Allow", "Action": "admin:CreateTable", "Resource": format!("arn:aws:s3:::{bucket}/namespaces/lab")}, + {"Effect": "Allow", "Action": "admin:CommitTable", "Resource": format!("arn:aws:s3:::{bucket}/namespaces/lab/tables/staged")}, + {"Effect": "Allow", "Action": "admin:ClaimTableLocation", "Resource": format!("arn:aws:s3:::{bucket}/stage-owned")}, + {"Effect": "Allow", "Action": ["s3:GetObject", "s3:PutObject"], "Resource": format!("arn:aws:s3:::{bucket}/*")} + ]}), + ) + .await?; + catalog_request( + &env, + &creator, + Method::POST, + &format!("{catalog}/staged"), + Some(staged_commit(&staged["metadata"], &format!("s3://{bucket}/stage-owned"))), + StatusCode::OK, + ) + .await?; + catalog_request(&env, &root, Method::DELETE, &format!("{catalog}/staged"), None, StatusCode::NO_CONTENT).await?; + env.restart_server_preserving_data(vec![], &server_env).await?; + env.wait_for_server_ready().await?; + for table in ["authorized", "staged"] { + catalog_request(&env, &root, Method::GET, &format!("{catalog}/{table}"), None, StatusCode::NOT_FOUND).await?; + } + get_object(&env, &reader, bucket, "authorized/known.txt", StatusCode::FORBIDDEN).await?; + get_object(&env, &root, bucket, "authorized/known.txt", StatusCode::OK).await?; + for key in ["shared/known.txt", "shared2/known.txt"] { + get_object(&env, &reader, bucket, key, StatusCode::OK).await?; + } + Ok(()) +} + +#[tokio::test] +async fn object_catalog_location_claim_preserves_objects_after_drop_and_restart() -> TestResult { + location_claim_preserves_other_users_objects_after_drop_and_restart("object").await +} + +#[tokio::test] +async fn durable_strong_catalog_location_claim_preserves_objects_after_drop_and_restart() -> TestResult { + location_claim_preserves_other_users_objects_after_drop_and_restart("durable-strong").await +} diff --git a/crates/policy/src/policy/action.rs b/crates/policy/src/policy/action.rs index d86952566..67ad9fe0a 100644 --- a/crates/policy/src/policy/action.rs +++ b/crates/policy/src/policy/action.rs @@ -556,6 +556,8 @@ pub enum AdminAction { SetTableAction, #[strum(serialize = "admin:CreateTable")] CreateTableAction, + #[strum(serialize = "admin:ClaimTableLocation")] + ClaimTableLocationAction, #[strum(serialize = "admin:RegisterTable")] RegisterTableAction, #[strum(serialize = "admin:CommitTable")] @@ -611,6 +613,7 @@ impl AdminAction { | AdminAction::GetTableAction | AdminAction::SetTableAction | AdminAction::CreateTableAction + | AdminAction::ClaimTableLocationAction | AdminAction::RegisterTableAction | AdminAction::CommitTableAction | AdminAction::DeleteTableAction @@ -703,6 +706,7 @@ impl AdminAction { | AdminAction::GetTableAction | AdminAction::SetTableAction | AdminAction::CreateTableAction + | AdminAction::ClaimTableLocationAction | AdminAction::RegisterTableAction | AdminAction::CommitTableAction | AdminAction::DeleteTableAction diff --git a/crates/policy/src/policy/policy.rs b/crates/policy/src/policy/policy.rs index 9290c5a30..70cf7a2cd 100644 --- a/crates/policy/src/policy/policy.rs +++ b/crates/policy/src/policy/policy.rs @@ -1748,6 +1748,53 @@ mod test { ); } + #[tokio::test] + async fn test_table_location_claim_is_limited_to_the_physical_prefix() -> Result<()> { + use crate::policy::action::{Action, AdminAction}; + + let policy = Policy::parse_config( + br#"{ + "Version": "2012-10-17", + "Statement": [ + {"Effect": "Allow", "Action": "admin:CreateTable", "Resource": "arn:aws:s3:::warehouse/namespaces/lab"}, + {"Effect": "Allow", "Action": "admin:ClaimTableLocation", "Resource": [ + "arn:aws:s3:::warehouse/owned", "arn:aws:s3:::warehouse/owned/*" + ]}, + {"Effect": "Deny", "Action": "admin:ClaimTableLocation", "Resource": [ + "arn:aws:s3:::warehouse/owned/private", "arn:aws:s3:::warehouse/owned/private/*" + ]} + ] + }"#, + )?; + let conditions = HashMap::new(); + let claims = HashMap::new(); + let groups = None; + for (bucket, object, allowed) in [ + ("warehouse", "owned", true), + ("warehouse", "owned/", true), + ("warehouse", "owned/child/", true), + ("warehouse", "owned/private/", false), + ("warehouse", "owned2/", false), + ("warehouse", "shared/", false), + ("warehouse", "namespaces/lab", false), + ("another-warehouse", "owned/", false), + ] { + let args = Args { + account: "creator", + groups: &groups, + action: Action::AdminAction(AdminAction::ClaimTableLocationAction), + bucket, + conditions: &conditions, + is_owner: false, + object, + claims: &claims, + deny_only: false, + }; + assert_eq!(policy.is_allowed(&args).await, allowed, "{bucket}/{object}"); + } + Ok(()) + } + #[tokio::test] async fn test_table_admin_action_with_resource_is_limited_to_bucket() -> Result<()> { use crate::policy::action::{Action, AdminAction}; diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 05522badb..40b1a4a21 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -79,12 +79,20 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t | Area | Status | Covered behavior | |---|---|---| | Table-aware S3 policy bridge | Supported | Ordinary S3 actions on warehouse paths are checked through the table bridge; table policy cannot be bypassed by direct object access. | +| Warehouse location authorization | Supported | Caller-selected locations require `admin:ClaimTableLocation` on the normalized physical prefix without a trailing slash, for example `arn:aws:s3:::warehouse/shared`. To also allow claims on descendant prefixes, include `arn:aws:s3:::warehouse/shared/*`. This applies to explicit create locations, staged publication, register/import, external catalog registration, and warehouse relocation. Server-assigned create locations retain the namespace `admin:CreateTable` requirement. Existing table operations that retain the warehouse prefix keep their existing permissions. | | Reserved catalog protection | Supported | Catalog-reserved prefixes are protected from ordinary object mutation. | | Static S3 credentials | Automated | Default PyIceberg smoke path. | | Catalog-vended credentials | Automated when enabled | `rustfs-vended-credentials` verifies the returned prefix, then checks Put/Head/Get/DeleteObject inside it and denies access outside it. | | Credential lifetime | Supported | Server-side TTL clamped to a short-lived range. | | No-long-term-data-credential bootstrap | Not claimed | Catalog setup still uses the configured principal before table-scoped credentials are requested. | +Granting `admin:ClaimTableLocation` authorizes changing the catalog ownership and +access requirements of all existing objects and versions under that prefix. It +does not grant S3 read or write access. Upgrade every server before adding the +action to policies; older servers do not recognize it. Applications using custom +locations, staged publication, or table registration need an explicit grant for +their warehouse prefixes. + ## Maintenance Matrix Object-backed and durable-strong catalogs support the configuration, scheduler, diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs index 130cea31d..0c02f2087 100644 --- a/rustfs/src/admin/handlers/table_catalog/mod.rs +++ b/rustfs/src/admin/handlers/table_catalog/mod.rs @@ -1224,6 +1224,31 @@ async fn authorize_table_catalog_resource_for_principal( action: AdminAction, ) -> S3Result<()> { let object_path = resource.object_path(); + authorize_table_catalog_object_for_principal(req, principal, resource.warehouse, object_path.as_deref().unwrap_or(""), action) + .await +} + +async fn authorize_table_warehouse_location_request(req: &S3Request, bucket: &str, location: &str) -> S3Result<()> { + let prefix = + crate::table_catalog::table_warehouse_object_prefix_from_location(bucket, location).map_err(catalog_store_error)?; + let principal = table_catalog_request_principal(req).await?; + authorize_table_catalog_object_for_principal( + req, + &principal, + bucket, + prefix.trim_end_matches('/'), + AdminAction::ClaimTableLocationAction, + ) + .await +} + +async fn authorize_table_catalog_object_for_principal( + req: &S3Request, + principal: &TableCatalogRequestPrincipal, + bucket: &str, + object: &str, + action: AdminAction, +) -> S3Result<()> { validate_admin_action_with_bucket_object_for_iam( principal.iam_store.clone(), &req.headers, @@ -1231,7 +1256,7 @@ async fn authorize_table_catalog_resource_for_principal( principal.owner, Action::AdminAction(action), req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)), - AdminResourceScope::bucket_object(resource.warehouse, object_path.as_deref().unwrap_or("")), + AdminResourceScope::bucket_object(bucket, object), ) .await } @@ -1443,6 +1468,10 @@ where } } }; + self.record_authorization_result(result).await + } + + async fn record_authorization_result(&self, result: S3Result<()>) -> crate::table_catalog::TableCatalogStoreResult<()> { if let Err(err) = result { let mut authorization_error = self.authorization_error.lock().await; if authorization_error.is_none() { @@ -1868,6 +1897,25 @@ where self.begin_bucket_publication(table_bucket).await } + async fn authorize_table_warehouse_location( + &self, + table_bucket: &str, + location: &str, + ) -> crate::table_catalog::TableCatalogStoreResult<()> { + let result = match &self.authorization { + TableCommitObjectAuthorization::Request(req) => { + let req = req.lock().await; + authorize_table_warehouse_location_request(&req, table_bucket, location).await + } + TableCommitObjectAuthorization::Preauthorized => Ok(()), + #[cfg(test)] + TableCommitObjectAuthorization::Test { .. } => { + return self.backend.authorize_table_warehouse_location(table_bucket, location).await; + } + }; + self.record_authorization_result(result).await + } + fn table_bucket_commit_publication_is_held(&self, table_bucket: &str) -> bool { let publication = self.publication.lock(); publication.bucket_fence.as_deref() == Some(table_bucket) && publication.guards.iter().all(|guard| !guard.is_lock_lost()) @@ -5764,6 +5812,10 @@ where let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; validate_metadata_table_location_in_bucket(bucket, &metadata)?; adopt_registered_metadata_identity(&mut entry, &metadata)?; + metadata_backend + .authorize_table_warehouse_location(bucket, &entry.warehouse_location) + .await + .map_err(catalog_store_error)?; validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?; store .register_table_with_publication(entry.clone(), metadata_backend) @@ -5953,6 +6005,13 @@ where { let mut target_entry = entry.clone(); target_entry.warehouse_location = crate::table_catalog::table_metadata_location(metadata)?.to_string(); + if crate::table_catalog::table_warehouse_object_prefix(&target_entry)? + != crate::table_catalog::table_warehouse_object_prefix(entry)? + { + metadata_backend + .authorize_table_warehouse_location(bucket, &target_entry.warehouse_location) + .await?; + } let context = crate::table_catalog::TableSnapshotGraphValidationContext::new(metadata_backend, bucket, &target_entry); crate::table_catalog::validate_table_snapshot_changes(&context, current_metadata, metadata).await } @@ -6639,6 +6698,10 @@ where updated_at: None, }; adopt_registered_metadata_identity(&mut entry, &metadata)?; + metadata_backend + .authorize_table_warehouse_location(bucket, &entry.warehouse_location) + .await + .map_err(catalog_store_error)?; validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?; crate::table_catalog::TableCommitPublication::begin_table_bucket(metadata_backend, bucket) @@ -7637,6 +7700,10 @@ where }, )?; adopt_registered_metadata_identity(&mut entry, &target_metadata)?; + metadata_backend + .authorize_table_warehouse_location(bucket, &entry.warehouse_location) + .await + .map_err(catalog_store_error)?; validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &target_metadata).await?; store .register_table_with_publication(entry.clone(), metadata_backend) @@ -7680,6 +7747,10 @@ where let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; validate_metadata_table_location_in_bucket(bucket, &metadata)?; adopt_registered_metadata_identity(&mut entry, &metadata)?; + metadata_backend + .authorize_table_warehouse_location(bucket, &entry.warehouse_location) + .await + .map_err(catalog_store_error)?; validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?; if let Some(existing) = store .load_table(bucket, &namespace.public_name(), table) diff --git a/rustfs/src/admin/handlers/table_catalog/table.rs b/rustfs/src/admin/handlers/table_catalog/table.rs index 5b048e7ef..66ea2f404 100644 --- a/rustfs/src/admin/handlers/table_catalog/table.rs +++ b/rustfs/src/admin/handlers/table_catalog/table.rs @@ -74,12 +74,15 @@ pub struct RestCreateTableHandler {} #[async_trait::async_trait] impl Operation for RestCreateTableHandler { - async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + async fn call(&self, mut req: S3Request, params: Params<'_, '_>) -> S3Result> { let warehouse = warehouse_from_params(¶ms)?; let namespace = namespace_from_params(¶ms)?; let resource = TableCatalogResource::namespace(&warehouse, &namespace); authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?; - let request = read_json_body::(req.input).await?; + let request = read_json_body::(std::mem::take(&mut req.input)).await?; + if let Some(location) = request.location.as_deref() { + authorize_table_warehouse_location_request(&req, &warehouse, location).await?; + } let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?; let store = table_catalog_store_from_backend(metadata_backend.clone())?; let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?; diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index 7580f3ac5..52af41cdb 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -279,6 +279,113 @@ async fn table_catalog_authentication_and_credentials_use_the_request_context() assert_eq!(err.code(), &S3ErrorCode::AccessDenied); } +#[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; + 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 IAM format"); + let iam = rustfs_iam::build_iam_sys(store.clone()).await.expect("initialize IAM"); + let access_key = "location-creator"; + let secret_key = "location-creator-secret"; + iam.create_user( + access_key, + &AddOrUpdateUserReq { + secret_key: secret_key.to_string(), + policy: None, + status: AccountStatus::Enabled, + }, + ) + .await + .expect("create restricted user"); + iam.set_policy( + "location-creator", + Policy::parse_config( + br#"{"Version":"2012-10-17","Statement":[ + {"Effect":"Allow","Action":"admin:CreateTable","Resource":"arn:aws:s3:::warehouse/namespaces/analytics"}, + {"Effect":"Allow","Action":"admin:ClaimTableLocation","Resource":[ + "arn:aws:s3:::warehouse/owned", "arn:aws:s3:::warehouse/owned/*" + ]} + ]}"#, + ) + .expect("parse location policy"), + ) + .await + .expect("store location policy"); + iam.policy_db_set(access_key, UserType::Reg, false, "location-creator") + .await + .expect("attach location policy"); + let context = Arc::new(AppContext::new(store, Arc::new(RequestIam { handle: iam }), Arc::new(RequestKms))); + assert!(context.publish_action_credentials(rustfs_credentials::Credentials { + access_key: "location-root".to_string(), + secret_key: "location-root-secret".to_string(), + status: "on".to_string(), + ..Default::default() + })); + let request = || table_catalog_handler_request(context.clone(), access_key, secret_key); + for location in ["s3://warehouse/owned", "s3://warehouse/owned/", "s3://warehouse/owned/child"] { + authorize_table_warehouse_location_request(&request(), "warehouse", location) + .await + .expect("explicit prefix grant permits adoption"); + } + for (bucket, location) in [ + ("warehouse", "s3://warehouse/shared"), + ("warehouse", "s3://warehouse/owned2"), + ("other-bucket", "s3://other-bucket/owned"), + ] { + let error = authorize_table_warehouse_location_request(&request(), bucket, location) + .await + .expect_err("namespace create grant cannot authorize another prefix"); + assert_eq!(error.code(), &S3ErrorCode::AccessDenied); + } + let mut anonymous = request(); + anonymous.credentials = None; + assert!( + authorize_table_warehouse_location_request(&anonymous, "warehouse", "s3://warehouse/owned") + .await + .is_err() + ); + let root = table_catalog_handler_request(context.clone(), "location-root", "location-root-secret"); + authorize_table_warehouse_location_request(&root, "warehouse", "s3://warehouse/shared") + .await + .expect("root can explicitly adopt a location"); + + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace"); + let (entry, metadata) = table_entry_from_create_table_request( + "warehouse", + &namespace, + serde_json::from_value(serde_json::json!({ + "name": "events", + "schema": {"type":"struct","fields":[{"id":1,"name":"id","required":false,"type":"long"}]} + })) + .expect("create request"), + ) + .expect("initial metadata"); + for (location, allowed) in [ + (entry.warehouse_location.as_str(), true), + ("s3://warehouse/owned", true), + ("s3://warehouse/shared", false), + ] { + let backend = TableCommitObjectBackend::for_request(TestTableCatalogObjectBackend::default(), request()); + let mut target = metadata.clone(); + target["location"] = serde_json::json!(location); + let result = validate_table_metadata_snapshot_graph(&backend, "warehouse", &entry, Some(&metadata), &target).await; + let result = backend.finish(result).await; + if allowed { + result.expect("unchanged or explicitly authorized warehouse location"); + } else { + assert_eq!( + result.expect_err("relocation requires prefix authorization").code(), + &S3ErrorCode::AccessDenied + ); + } + } +} + #[tokio::test] #[serial_test::serial] async fn table_maintenance_worker_handler_enforces_conditions_after_authorization() { diff --git a/rustfs/src/table_catalog/iceberg/validation.rs b/rustfs/src/table_catalog/iceberg/validation.rs index e2066b696..97c1d03e9 100644 --- a/rustfs/src/table_catalog/iceberg/validation.rs +++ b/rustfs/src/table_catalog/iceberg/validation.rs @@ -77,7 +77,10 @@ fn warehouse_object_prefix_from_location( normalize_warehouse_object_prefix(object_prefix, max_prefix_depth) } -fn table_warehouse_object_prefix_from_location(table_bucket: &str, warehouse_location: &str) -> TableCatalogStoreResult { +pub(crate) fn table_warehouse_object_prefix_from_location( + table_bucket: &str, + warehouse_location: &str, +) -> TableCatalogStoreResult { warehouse_object_prefix_from_location(table_bucket, warehouse_location, Some(WAREHOUSE_INDEX_MAX_PREFIX_DEPTH)) } diff --git a/rustfs/src/table_catalog/store/mod.rs b/rustfs/src/table_catalog/store/mod.rs index 91cc264cc..c51f50728 100644 --- a/rustfs/src/table_catalog/store/mod.rs +++ b/rustfs/src/table_catalog/store/mod.rs @@ -966,6 +966,12 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static { async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult; + /// Authorize adopting a caller-selected warehouse prefix. Request adapters + /// enforce the caller's policy; trusted storage backends have no IAM identity. + async fn authorize_table_warehouse_location(&self, _table_bucket: &str, _location: &str) -> TableCatalogStoreResult<()> { + Ok(()) + } + async fn begin_table_bucket_commit_publication(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> { Ok(()) }