fix(table-catalog): authorize warehouse location claims (#8736)

This commit is contained in:
GatewayJ
2026-10-20 22:05:29 +08:00
committed by GitHub
parent 7cc4a9dbd0
commit 4dba635d48
11 changed files with 675 additions and 6 deletions
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=3a7f52dd884ed269746029e983006ce69b62f260db2ad1acdcffdb2196c65d11
sha256-linux=4af4bb607919b96e63e2a3fb7a78bd18fcb08e265f4ca4ce6afb09a8be798305
sha256-darwin=331393a79b7c1fbccd2a6f6324361c2aa2ab60c818176ef56fbf7fa15bfc5af0
sha256-linux=f85f0021153264f00eba8fa45384678f5ae396d7d74995c789dbbca9cd6d7ea1
+3
View File
@@ -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;
@@ -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<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
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<Value>,
expected: StatusCode,
) -> TestResult<Value> {
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
}
+4
View File
@@ -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
+47
View File
@@ -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};
@@ -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,
+72 -1
View File
@@ -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<Body>, 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<Body>,
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::<Option<RemoteAddr>>().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)
@@ -74,12 +74,15 @@ pub struct RestCreateTableHandler {}
#[async_trait::async_trait]
impl Operation for RestCreateTableHandler {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let resource = TableCatalogResource::namespace(&warehouse, &namespace);
authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?;
let request = read_json_body::<CreateTableRequest>(req.input).await?;
let request = read_json_body::<CreateTableRequest>(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?;
@@ -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() {
@@ -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<String> {
pub(crate) fn table_warehouse_object_prefix_from_location(
table_bucket: &str,
warehouse_location: &str,
) -> TableCatalogStoreResult<String> {
warehouse_object_prefix_from_location(table_bucket, warehouse_location, Some(WAREHOUSE_INDEX_MAX_PREFIX_DEPTH))
}
+6
View File
@@ -966,6 +966,12 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard>;
/// 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(())
}