diff --git a/crates/e2e_test/src/kms/kms_rekey_sweep_test.rs b/crates/e2e_test/src/kms/kms_rekey_sweep_test.rs new file mode 100644 index 000000000..cb0da2992 --- /dev/null +++ b/crates/e2e_test/src/kms/kms_rekey_sweep_test.rs @@ -0,0 +1,191 @@ +// Copyright 2024 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. + +//! Bulk DEK rekey sweep over stored objects. +//! +//! The full loop — rotate the master key, sweep, prove convergence — runs +//! against Vault Transit, whose context-bound envelopes exercise the +//! decrypt + re-encrypt rewrap route end to end. The capability refusal runs +//! against the Local backend, which supports no rewrap at all. + +use super::common::{ + LocalKMSTestEnvironment, VAULT_KEY_NAME, VaultTestEnvironment, kms_admin_request, start_kms, wait_for_kms_ready, +}; +use crate::common::{TEST_BUCKET, init_logging}; +use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::types::ServerSideEncryption; +use std::time::Duration; +use tracing::info; + +async fn rekey_status( + base_url: &str, + access_key: &str, + secret_key: &str, +) -> Result> { + let body = kms_admin_request( + base_url, + http::Method::GET, + "/rustfs/admin/v3/kms/keys/rekey/status", + None, + access_key, + secret_key, + ) + .await?; + Ok(serde_json::from_str(&body)?) +} + +/// Start a sweep and poll it to a terminal state. +async fn run_rekey_to_completion( + base_url: &str, + access_key: &str, + secret_key: &str, + request_body: &str, +) -> Result> { + kms_admin_request( + base_url, + http::Method::POST, + "/rustfs/admin/v3/kms/keys/rekey", + Some(request_body), + access_key, + secret_key, + ) + .await?; + + for _ in 0..120 { + let status = rekey_status(base_url, access_key, secret_key).await?; + if status["state"] != "running" { + return Ok(status); + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + Err("rekey sweep did not reach a terminal state in time".into()) +} + +#[tokio::test] +async fn kms_rekey_sweep_rewraps_rotated_envelopes_and_converges() -> Result<(), Box> { + init_logging(); + info!("Testing the bulk rekey sweep against Vault Transit"); + + let mut env = VaultTestEnvironment::new().await?; + env.start_vault().await?; + env.setup_vault_transit().await?; + env.start_rustfs_for_vault().await?; + env.configure_vault_transit_kms().await?; + start_kms(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?; + wait_for_kms_ready(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?; + + let base_url = env.base_env.url.clone(); + let access_key = env.base_env.access_key.clone(); + let secret_key = env.base_env.secret_key.clone(); + + let s3_client = env.base_env.create_s3_client(); + env.base_env.create_test_bucket(TEST_BUCKET).await?; + + // Three encrypted objects the sweep must rewrap, one plaintext object it + // must leave alone. + let encrypted_keys = ["rekey/alpha", "rekey/beta", "rekey/gamma"]; + let mut bodies = Vec::new(); + for (index, key) in encrypted_keys.iter().enumerate() { + let body: Vec = (0..2048).map(|i| ((i + index * 7) % 251) as u8).collect(); + s3_client + .put_object() + .bucket(TEST_BUCKET) + .key(*key) + .server_side_encryption(ServerSideEncryption::AwsKms) + .ssekms_key_id(VAULT_KEY_NAME) + .body(ByteStream::from(body.clone())) + .send() + .await?; + bodies.push(body); + } + s3_client + .put_object() + .bucket(TEST_BUCKET) + .key("rekey/plaintext") + .body(ByteStream::from(b"unencrypted".to_vec())) + .send() + .await?; + + // Rotate the master key so the stored envelopes fall behind Vault's + // latest version. + kms_admin_request( + &base_url, + http::Method::POST, + "/rustfs/admin/v3/kms/keys/rotate", + Some(&format!(r#"{{"key_id":"{VAULT_KEY_NAME}"}}"#)), + &access_key, + &secret_key, + ) + .await?; + + let status = + run_rekey_to_completion(&base_url, &access_key, &secret_key, &format!(r#"{{"buckets":["{TEST_BUCKET}"]}}"#)).await?; + assert_eq!(status["state"], "completed", "first sweep must complete: {status}"); + assert_eq!(status["failed"], 0, "no object may fail: {status}"); + assert_eq!( + status["rewrapped"], + encrypted_keys.len(), + "every rotated envelope must be rewrapped: {status}" + ); + assert!( + status["not_applicable"].as_u64().unwrap_or(0) >= 1, + "the plaintext object must be reported not applicable: {status}" + ); + + // The rewrapped objects still serve their exact bytes. + for (key, expected) in encrypted_keys.iter().zip(&bodies) { + let response = s3_client.get_object().bucket(TEST_BUCKET).key(*key).send().await?; + let data = response.body.collect().await?.into_bytes(); + assert_eq!(data.as_ref(), expected.as_slice(), "object {key} must be byte-exact after the rewrap"); + } + + // Convergence: a second sweep finds everything current and writes nothing. + let status = + run_rekey_to_completion(&base_url, &access_key, &secret_key, &format!(r#"{{"buckets":["{TEST_BUCKET}"]}}"#)).await?; + assert_eq!(status["state"], "completed", "second sweep must complete: {status}"); + assert_eq!(status["rewrapped"], 0, "a converged sweep must write nothing: {status}"); + assert_eq!(status["failed"], 0, "{status}"); + assert_eq!( + status["already_current"], + encrypted_keys.len(), + "every envelope must now be current: {status}" + ); + + env.base_env.delete_test_bucket(TEST_BUCKET).await?; + Ok(()) +} + +#[tokio::test] +async fn kms_rekey_refuses_a_backend_without_rewrap_support() -> Result<(), Box> { + init_logging(); + info!("Testing that the rekey sweep refuses the Local backend up front"); + + let mut kms_env = LocalKMSTestEnvironment::new().await?; + let _default_key_id = kms_env.start_rustfs_for_local_kms().await?; + kms_env.wait_for_kms_ready().await?; + + let error = kms_admin_request( + &kms_env.base_env.url, + http::Method::POST, + "/rustfs/admin/v3/kms/keys/rekey", + Some("{}"), + &kms_env.base_env.access_key, + &kms_env.base_env.secret_key, + ) + .await + .expect_err("a backend without rewrap support must be refused up front"); + assert!(error.to_string().contains("501"), "the refusal must be 501 Not Implemented, got: {error}"); + + Ok(()) +} diff --git a/crates/e2e_test/src/kms/mod.rs b/crates/e2e_test/src/kms/mod.rs index 5d0e045e5..23bc2ff3d 100644 --- a/crates/e2e_test/src/kms/mod.rs +++ b/crates/e2e_test/src/kms/mod.rs @@ -62,3 +62,6 @@ mod kms_authorization_negative_matrix_test; #[cfg(test)] mod kms_ilm_sse_kms_test; + +#[cfg(test)] +mod kms_rekey_sweep_test; diff --git a/crates/policy/src/policy/action.rs b/crates/policy/src/policy/action.rs index 6f663406d..1a3438219 100644 --- a/crates/policy/src/policy/action.rs +++ b/crates/policy/src/policy/action.rs @@ -712,6 +712,12 @@ pub enum KmsAction { /// Preflight or execute a KMS restore. #[strum(serialize = "kms:Restore")] RestoreAction, + /// Run the bulk rekey sweep that rewraps stored object DEK envelopes onto + /// current master key versions. Cluster-scoped: it walks and rewrites + /// object metadata across buckets, so it is never conferred by a per-key + /// role template. + #[strum(serialize = "kms:Rekey")] + RekeyAction, } #[cfg(test)] @@ -754,6 +760,7 @@ mod tests { ("kms:Decrypt", KmsAction::DecryptAction), ("kms:Backup", KmsAction::BackupAction), ("kms:Restore", KmsAction::RestoreAction), + ("kms:Rekey", KmsAction::RekeyAction), ] { let action = Action::try_from(raw).expect("Should parse KMS action"); assert_eq!(action, Action::KmsAction(expected)); diff --git a/crates/policy/src/policy/policy.rs b/crates/policy/src/policy/policy.rs index 946f930a8..b0eb4d236 100644 --- a/crates/policy/src/policy/policy.rs +++ b/crates/policy/src/policy/policy.rs @@ -854,13 +854,14 @@ mod test { /// Actions that act on the service or on every key's material at once. No role /// template may confer them. - const KMS_CLUSTER_ADMIN_ACTIONS: [KmsAction; 6] = [ + const KMS_CLUSTER_ADMIN_ACTIONS: [KmsAction; 7] = [ KmsAction::AllActions, KmsAction::ConfigureAction, KmsAction::ServiceControlAction, KmsAction::ClearCacheAction, KmsAction::BackupAction, KmsAction::RestoreAction, + KmsAction::RekeyAction, ]; const KMS_ROLE_TEMPLATES: [&str; 3] = [default::KMS_KEY_ADMINISTRATOR, default::KMS_KEY_USER, default::KMS_AUDITOR]; diff --git a/docs/operations/kms-backend-security.md b/docs/operations/kms-backend-security.md index cdf275027..9d3c2307b 100644 --- a/docs/operations/kms-backend-security.md +++ b/docs/operations/kms-backend-security.md @@ -133,7 +133,7 @@ Decryption loads exactly the version recorded in the envelope and fails closed w ### Retention and destruction preconditions -- Every version record that any stored DEK envelope references must remain readable. Until an object rewrap/migration capability exists, assume **every** version of a rotated key is referenced: destroying a version record permanently orphans all objects whose DEKs it wrapped. +- Every version record that any stored DEK envelope references must remain readable. The bulk rekey sweep (`POST /rustfs/admin/v3/kms/keys/rekey`, gated on `kms:Rekey`) rewraps stored envelopes onto the current version; until a sweep has completed with zero failures after the last rotation, assume **every** version of a rotated key is referenced: destroying a version record permanently orphans all objects whose DEKs it wrapped. A completed sweep is evidence, not authority — the deletion gate stays the decision point. Replication strips encryption metadata in transit, so a sweep never propagates to a replica site: each site runs its own. - Version records are ordinary KV v2 secrets under the key subtree. Never run `kv metadata delete` or `kv destroy` against `{prefix}/{key_id}/versions/*`, and do not apply `delete-version-after` or retention tooling to that subtree. RustFS-managed retention does not rely on KV2's own secret versioning (each version record has a single KV revision), so KV `max-versions` settings do not protect or endanger history — but metadata deletion always removes a record entirely. - Permanent key deletion through RustFS (`force_immediate` after `PendingDeletion`) purges the key's version records together with the key record; that is the only supported way to remove them. It is refused by default: the server must set `RUSTFS_KMS_ALLOW_IMMEDIATE_DELETION=true`, and the request must be a `DELETE` with a JSON body that sets `force_immediate` and echoes the key id back as `confirm_key_id` — the query-parameter form (`?force_immediate=true`) is refused outright, whatever the gate is set to. Leave the gate off unless you are actively destroying keys, and turn it off again afterwards — the pending-deletion window plus `CancelKeyDeletion` is the only recovery path for objects encrypted under the key. - For Vault Transit, retention is governed by the Transit key's `min_decryption_version`: never raise it above the oldest version that may still protect live ciphertext. diff --git a/docs/operations/kms-cryptographic-compliance.md b/docs/operations/kms-cryptographic-compliance.md index 08dd593e3..e8c19e21a 100644 --- a/docs/operations/kms-cryptographic-compliance.md +++ b/docs/operations/kms-cryptographic-compliance.md @@ -147,4 +147,4 @@ Retirement moves an algorithm through these states, never skipping one: ### Known gap -Step 3 is currently unreachable for object data. There is no object rewrap or re-encryption capability, so there is no supported way to migrate already-written objects off an algorithm or off a master key version — the same gap that forces the rotation retention rule in [KMS backend security properties](kms-backend-security.md#retention-and-destruction-preconditions). Until a rewrap capability exists, treat every algorithm that has ever been written as permanently read-required, and confine deprecation to step 1. +Step 3 is partially reachable for object data: the bulk rekey sweep (`POST /rustfs/admin/v3/kms/keys/rekey`) migrates stored DEK envelopes off superseded **master key versions** without touching object bodies. It does not re-encrypt object data, so migrating off a data-encryption **algorithm** still has no supported path — treat every algorithm that has ever been written as permanently read-required, and confine algorithm deprecation to step 1. diff --git a/rustfs/src/admin/handlers/kms.rs b/rustfs/src/admin/handlers/kms.rs index 00714af6f..ef7cb2e50 100644 --- a/rustfs/src/admin/handlers/kms.rs +++ b/rustfs/src/admin/handlers/kms.rs @@ -14,7 +14,7 @@ //! KMS admin handlers for HTTP API -use super::{kms_backup, kms_dynamic, kms_key_lifecycle, kms_key_metadata, kms_keys, kms_management}; +use super::{kms_backup, kms_dynamic, kms_key_lifecycle, kms_key_metadata, kms_keys, kms_management, kms_rekey}; use crate::admin::router::{AdminOperation, S3Router}; pub fn register_kms_route(r: &mut S3Router) -> std::io::Result<()> { @@ -24,5 +24,6 @@ pub fn register_kms_route(r: &mut S3Router) -> std::io::Result<( kms_key_lifecycle::register_kms_key_lifecycle_route(r)?; kms_backup::register_kms_backup_route(r)?; kms_key_metadata::register_kms_key_metadata_route(r)?; + kms_rekey::register_kms_rekey_route(r)?; Ok(()) } diff --git a/rustfs/src/admin/handlers/kms_rekey.rs b/rustfs/src/admin/handlers/kms_rekey.rs new file mode 100644 index 000000000..32186d781 --- /dev/null +++ b/rustfs/src/admin/handlers/kms_rekey.rs @@ -0,0 +1,209 @@ +// Copyright 2024 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. + +//! Admin API for the bulk DEK rekey sweep: start, status and cancel. +//! +//! All three endpoints require `kms:Rekey`, a cluster-scoped action: the sweep +//! walks and rewrites object metadata across buckets, so no per-key role +//! template confers it. + +use crate::admin::auth::authorize_admin_request; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::runtime_sources::{current_kms_runtime_service_manager, current_object_store_handle}; +use crate::admin::storage_api::s3::{self, Body, S3ErrorCode, S3Request, S3Response, S3Result}; +use crate::kms_rekey::{self, RekeyStartError}; +use crate::server::ADMIN_PREFIX; +use hyper::{HeaderMap, Method, StatusCode}; +use matchit::Params; +use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; +use rustfs_policy::policy::action::{Action, KmsAction}; +use serde::Deserialize; + +fn kms_rekey_actions() -> Vec { + vec![Action::KmsAction(KmsAction::RekeyAction)] +} + +async fn authorize_kms_rekey_request(req: &S3Request) -> S3Result<()> { + if req.credentials.is_none() { + return Err(s3::error(S3ErrorCode::InvalidRequest, "authentication required")); + } + authorize_admin_request(req, kms_rekey_actions()).await?; + Ok(()) +} + +fn json_response(status: StatusCode, body: Vec) -> S3Response<(StatusCode, Body)> { + let mut headers = HeaderMap::new(); + headers.insert(s3::header::CONTENT_TYPE, "application/json".parse().expect("static content type")); + S3Response::with_headers((status, Body::from(body)), headers) +} + +fn snapshot_response(status: StatusCode, snapshot: &kms_rekey::RekeyJobSnapshot) -> S3Result> { + let body = serde_json::to_vec(snapshot) + .map_err(|e| s3::error(S3ErrorCode::InternalError, format!("failed to serialize rekey status: {e}")))?; + Ok(json_response(status, body)) +} + +/// Body of `POST /v3/kms/keys/rekey`. An empty body sweeps every bucket. +#[derive(Debug, Default, Deserialize)] +#[serde(deny_unknown_fields)] +struct StartKmsRekeyRequest { + /// Buckets to sweep; every bucket when absent or empty. + #[serde(default)] + buckets: Option>, + /// Object key prefix to restrict the sweep to. + #[serde(default)] + prefix: Option, +} + +/// `POST /v3/kms/keys/rekey` — start a sweep. +pub struct StartKmsRekeyHandler; + +#[async_trait::async_trait] +impl Operation for StartKmsRekeyHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_kms_rekey_request(&req).await?; + + let mut req = req; + let body = req + .input + .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE) + .await + .map_err(|e| s3::error(S3ErrorCode::InvalidRequest, format!("failed to read request body: {e}")))?; + let request: StartKmsRekeyRequest = if body.is_empty() { + StartKmsRekeyRequest::default() + } else { + serde_json::from_slice(&body) + .map_err(|e| s3::error(S3ErrorCode::InvalidRequest, format!("invalid rekey request body: {e}")))? + }; + + // Refuse up front when the backend cannot rewrap at all: a sweep would + // count every encrypted object as failed while changing nothing. + // `UnsupportedCapability` is a permanent gap of the configured backend, + // so it surfaces as 501, mirroring the key lifecycle handlers. + let Some(manager) = current_kms_runtime_service_manager() else { + return Err(s3::error(S3ErrorCode::InternalError, "KMS service not initialized")); + }; + let Some(service) = manager.get_encryption_service().await else { + return Err(s3::error(S3ErrorCode::InternalError, "KMS service not running")); + }; + if !service.backend_capabilities().rewrap { + return Ok(json_response( + StatusCode::NOT_IMPLEMENTED, + serde_json::json!({ + "error": "the configured KMS backend does not support rewrapping data-key envelopes" + }) + .to_string() + .into_bytes(), + )); + } + + let Some(store) = current_object_store_handle() else { + return Err(s3::error(S3ErrorCode::InternalError, "object store is not ready")); + }; + + match kms_rekey::start(store, request.buckets, request.prefix.unwrap_or_default()).await { + Ok(snapshot) => snapshot_response(StatusCode::OK, &snapshot), + Err(RekeyStartError::AlreadyRunning(job_id)) => Ok(json_response( + StatusCode::CONFLICT, + serde_json::json!({ "error": "a rekey sweep is already running", "job_id": job_id }) + .to_string() + .into_bytes(), + )), + Err(RekeyStartError::Storage(message)) => Err(s3::error(S3ErrorCode::InternalError, message)), + } + } +} + +/// `GET /v3/kms/keys/rekey/status` — progress of the current or last sweep. +pub struct KmsRekeyStatusHandler; + +#[async_trait::async_trait] +impl Operation for KmsRekeyStatusHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_kms_rekey_request(&req).await?; + + match kms_rekey::status() { + Some(snapshot) => snapshot_response(StatusCode::OK, &snapshot), + None => Ok(json_response( + StatusCode::NOT_FOUND, + serde_json::json!({ "error": "no rekey sweep has run" }) + .to_string() + .into_bytes(), + )), + } + } +} + +/// `POST /v3/kms/keys/rekey/cancel` — request cancellation of a running sweep. +pub struct CancelKmsRekeyHandler; + +#[async_trait::async_trait] +impl Operation for CancelKmsRekeyHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_kms_rekey_request(&req).await?; + + match kms_rekey::cancel() { + Some(snapshot) => snapshot_response(StatusCode::OK, &snapshot), + None => Ok(json_response( + StatusCode::NOT_FOUND, + serde_json::json!({ "error": "no rekey sweep has run" }) + .to_string() + .into_bytes(), + )), + } + } +} + +pub fn register_kms_rekey_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/rekey").as_str(), + AdminOperation(&StartKmsRekeyHandler {}), + )?; + + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/rekey/status").as_str(), + AdminOperation(&KmsRekeyStatusHandler {}), + )?; + + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/rekey/cancel").as_str(), + AdminOperation(&CancelKmsRekeyHandler {}), + )?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rekey_endpoints_require_the_cluster_scoped_action() { + assert_eq!(kms_rekey_actions(), vec![Action::KmsAction(KmsAction::RekeyAction)]); + } + + #[test] + fn start_request_parses_and_rejects_unknown_fields() { + let parsed: StartKmsRekeyRequest = + serde_json::from_slice(br#"{"buckets": ["a", "b"], "prefix": "photos/"}"#).expect("valid body must parse"); + assert_eq!(parsed.buckets.as_deref(), Some(["a".to_string(), "b".to_string()].as_slice())); + assert_eq!(parsed.prefix.as_deref(), Some("photos/")); + + serde_json::from_slice::(br#"{"bucket": "typo"}"#) + .expect_err("unknown fields must be rejected, not silently ignored"); + } +} diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 1495d2508..a18b579d5 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -41,6 +41,7 @@ pub mod kms_key_lifecycle; pub mod kms_key_metadata; pub mod kms_keys; pub mod kms_management; +pub mod kms_rekey; pub mod metrics; pub mod mfa; pub mod module_switch; diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index ea5ed00b6..4ac40e5b7 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -65,6 +65,7 @@ const KMS_DISABLE_KEY: AdminActionRef = AdminActionRef::new("kms:DisableKey"); const KMS_ENABLE_KEY: AdminActionRef = AdminActionRef::new("kms:EnableKey"); const KMS_GENERATE_DATA_KEY: AdminActionRef = AdminActionRef::new("kms:GenerateDataKey"); const KMS_LIST_KEYS: AdminActionRef = AdminActionRef::new("kms:ListKeys"); +const KMS_REKEY: AdminActionRef = AdminActionRef::new("kms:Rekey"); const KMS_RESTORE: AdminActionRef = AdminActionRef::new("kms:Restore"); const KMS_ROTATE_KEY: AdminActionRef = AdminActionRef::new("kms:RotateKey"); const KMS_SERVICE_CONTROL: AdminActionRef = AdminActionRef::new("kms:ServiceControl"); @@ -821,6 +822,21 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ RouteRiskLevel::High, ), admin(HttpMethod::Post, "/rustfs/admin/v3/kms/keys/rotate", KMS_ROTATE_KEY, RouteRiskLevel::High), + // The rekey sweep walks and rewrites object metadata across buckets, so it + // carries its own cluster-scoped action rather than reusing a per-key one. + admin(HttpMethod::Post, "/rustfs/admin/v3/kms/keys/rekey", KMS_REKEY, RouteRiskLevel::High), + admin( + HttpMethod::Get, + "/rustfs/admin/v3/kms/keys/rekey/status", + KMS_REKEY, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/rustfs/admin/v3/kms/keys/rekey/cancel", + KMS_REKEY, + RouteRiskLevel::High, + ), // Backup and restore act on the material of every key at once, so they // carry their own actions rather than reusing any per-key one. admin(HttpMethod::Get, "/rustfs/admin/v3/kms/backup", KMS_BACKUP, RouteRiskLevel::Sensitive), diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index e9d1ce39b..8508e06ce 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -357,6 +357,9 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::POST, "/v3/kms/keys/enable"), admin_route(Method::POST, "/v3/kms/keys/disable"), admin_route(Method::POST, "/v3/kms/keys/rotate"), + admin_route(Method::POST, "/v3/kms/keys/rekey"), + admin_route(Method::GET, "/v3/kms/keys/rekey/status"), + admin_route(Method::POST, "/v3/kms/keys/rekey/cancel"), admin_route(Method::GET, "/v3/kms/backup"), admin_route(Method::POST, "/v3/kms/backup"), admin_route(Method::POST, "/v3/kms/restore/dry-run"), diff --git a/rustfs/src/admin/snapshots/rustfs__admin__kms_contract__tests__kms_admin_route_contract.snap b/rustfs/src/admin/snapshots/rustfs__admin__kms_contract__tests__kms_admin_route_contract.snap index 1b64b8595..c321ea885 100644 --- a/rustfs/src/admin/snapshots/rustfs__admin__kms_contract__tests__kms_admin_route_contract.snap +++ b/rustfs/src/admin/snapshots/rustfs__admin__kms_contract__tests__kms_admin_route_contract.snap @@ -39,6 +39,12 @@ expression: kms_route_contract() "action": "kms:ListKeys", "risk": "Sensitive" }, + { + "method": "GET", + "path": "/rustfs/admin/v3/kms/keys/rekey/status", + "action": "kms:Rekey", + "risk": "Sensitive" + }, { "method": "GET", "path": "/rustfs/admin/v3/kms/keys/{key_id}", @@ -123,6 +129,18 @@ expression: kms_route_contract() "action": "kms:EnableKey", "risk": "High" }, + { + "method": "POST", + "path": "/rustfs/admin/v3/kms/keys/rekey", + "action": "kms:Rekey", + "risk": "High" + }, + { + "method": "POST", + "path": "/rustfs/admin/v3/kms/keys/rekey/cancel", + "action": "kms:Rekey", + "risk": "High" + }, { "method": "POST", "path": "/rustfs/admin/v3/kms/keys/rotate", diff --git a/rustfs/src/kms_rekey.rs b/rustfs/src/kms_rekey.rs new file mode 100644 index 000000000..d253e541b --- /dev/null +++ b/rustfs/src/kms_rekey.rs @@ -0,0 +1,354 @@ +// Copyright 2024 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. + +//! Bulk DEK rekey sweep: walk stored objects and rewrap their KMS-wrapped +//! data keys onto current master key versions, touching only `xl.meta`. +//! +//! The sweep is convergent rather than checkpointed: every step is idempotent +//! (an already-current envelope costs one describe-shaped KMS call and no +//! write), so recovery from a crash, a cancel, or partial failure is simply +//! running the sweep again. Failures are counted and logged per object, never +//! silently dropped, and never abort the sweep — a sweep that stopped at the +//! first unreadable object would hide every object behind it. +//! +//! One sweep runs at a time. Concurrent sweeps would double every KMS +//! round-trip for zero extra coverage and interleave their metadata writes. +//! +//! Cross-site note: replication strips encryption metadata in transit, so a +//! rewrap here never propagates to a replica site — each site runs its own +//! sweep. + +use crate::storage_api::kms::contract::bucket::{BucketOperations, BucketOptions}; +use crate::storage_api::kms::contract::list::ListOperations; +use crate::storage_api::kms::contract::object::ObjectOperations; +use crate::storage_api::kms::{ECStore, ObjectDekRewrapOutcome, StorageObjectOptions, rewrap_object_encryption_metadata}; +use serde::Serialize; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, LazyLock, Mutex}; +use tokio_util::sync::CancellationToken; +use tracing::{info, warn}; + +const LOG_COMPONENT: &str = "kms"; +const LOG_SUBSYSTEM: &str = "rekey"; + +/// Terminal and non-terminal states of a sweep. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum RekeyJobState { + Running, + Completed, + Cancelled, +} + +/// Wire shape of the status response; also the internal progress record. +#[derive(Debug, Clone, Serialize)] +pub(crate) struct RekeyJobSnapshot { + pub job_id: String, + pub state: RekeyJobState, + /// Buckets the sweep covers, in sweep order. + pub buckets: Vec, + /// Bucket currently being walked; `None` before the first and after the last. + pub current_bucket: Option, + /// Object versions the walk yielded. + pub scanned: u64, + /// Envelopes rewrapped and persisted. + pub rewrapped: u64, + /// Envelopes already on the current version and format. + pub already_current: u64, + /// Versions with nothing to rewrap (plaintext, SSE-C, MinIO-sealed). + pub not_applicable: u64, + /// Versions whose rewrap or metadata write failed; details are in the log. + /// A re-run retries exactly these, because everything else converges to + /// `already_current`. + pub failed: u64, +} + +struct RekeyJob { + id: String, + cancel: CancellationToken, + buckets: Vec, + scanned: AtomicU64, + rewrapped: AtomicU64, + already_current: AtomicU64, + not_applicable: AtomicU64, + failed: AtomicU64, + /// `(state, current_bucket)` under one lock so a snapshot never pairs a + /// terminal state with a bucket still marked in progress. + progress: Mutex<(RekeyJobState, Option)>, +} + +impl RekeyJob { + fn snapshot(&self) -> RekeyJobSnapshot { + let (state, current_bucket) = self.progress.lock().expect("rekey progress lock").clone(); + RekeyJobSnapshot { + job_id: self.id.clone(), + state, + buckets: self.buckets.clone(), + current_bucket, + scanned: self.scanned.load(Ordering::Relaxed), + rewrapped: self.rewrapped.load(Ordering::Relaxed), + already_current: self.already_current.load(Ordering::Relaxed), + not_applicable: self.not_applicable.load(Ordering::Relaxed), + failed: self.failed.load(Ordering::Relaxed), + } + } + + fn set_progress(&self, state: RekeyJobState, current_bucket: Option) { + *self.progress.lock().expect("rekey progress lock") = (state, current_bucket); + } +} + +/// The single sweep slot. Holding the finished job (rather than clearing it) +/// keeps the final counters queryable until the next sweep starts. +static ACTIVE_JOB: LazyLock>>> = LazyLock::new(|| Mutex::new(None)); + +/// Why a sweep could not be started. +#[derive(Debug)] +pub(crate) enum RekeyStartError { + /// A sweep is already running; its id is carried for the error message. + AlreadyRunning(String), + /// The requested bucket list could not be resolved. + Storage(String), +} + +/// Start a sweep over `buckets` (all buckets when `None`) under `prefix`. +/// +/// Returns the job id. The sweep runs on a background task; progress and the +/// terminal state are read through [`status`]. +pub(crate) async fn start( + store: Arc, + buckets: Option>, + prefix: String, +) -> Result { + let buckets = match buckets { + Some(buckets) if !buckets.is_empty() => buckets, + _ => store + .list_bucket(&BucketOptions::default()) + .await + .map_err(|e| RekeyStartError::Storage(format!("failed to list buckets: {e}")))? + .into_iter() + .map(|bucket| bucket.name) + .collect(), + }; + + let job = { + let mut slot = ACTIVE_JOB.lock().expect("rekey job slot lock"); + if let Some(existing) = slot.as_ref() + && existing.snapshot().state == RekeyJobState::Running + { + return Err(RekeyStartError::AlreadyRunning(existing.id.clone())); + } + let job = Arc::new(RekeyJob { + id: uuid::Uuid::new_v4().to_string(), + cancel: CancellationToken::new(), + buckets, + scanned: AtomicU64::new(0), + rewrapped: AtomicU64::new(0), + already_current: AtomicU64::new(0), + not_applicable: AtomicU64::new(0), + failed: AtomicU64::new(0), + progress: Mutex::new((RekeyJobState::Running, None)), + }); + *slot = Some(job.clone()); + job + }; + + info!( + component = LOG_COMPONENT, + subsystem = LOG_SUBSYSTEM, + event = "kms_rekey_started", + job_id = %job.id, + buckets = job.buckets.len(), + "KMS rekey sweep started" + ); + let task_job = job.clone(); + tokio::spawn(async move { run_sweep(store, task_job, prefix).await }); + + Ok(job.snapshot()) +} + +/// Progress of the current or most recent sweep, if any. +pub(crate) fn status() -> Option { + ACTIVE_JOB + .lock() + .expect("rekey job slot lock") + .as_ref() + .map(|job| job.snapshot()) +} + +/// Request cancellation of the running sweep. Returns the snapshot the caller +/// can report, or `None` when no sweep exists. +pub(crate) fn cancel() -> Option { + let slot = ACTIVE_JOB.lock().expect("rekey job slot lock"); + let job = slot.as_ref()?; + job.cancel.cancel(); + Some(job.snapshot()) +} + +async fn run_sweep(store: Arc, job: Arc, prefix: String) { + for bucket in job.buckets.clone() { + if job.cancel.is_cancelled() { + break; + } + job.set_progress(RekeyJobState::Running, Some(bucket.clone())); + sweep_bucket(&store, &job, &bucket, &prefix).await; + } + + let state = if job.cancel.is_cancelled() { + RekeyJobState::Cancelled + } else { + RekeyJobState::Completed + }; + job.set_progress(state, None); + let snapshot = job.snapshot(); + info!( + component = LOG_COMPONENT, + subsystem = LOG_SUBSYSTEM, + event = "kms_rekey_finished", + job_id = %job.id, + state = ?snapshot.state, + scanned = snapshot.scanned, + rewrapped = snapshot.rewrapped, + already_current = snapshot.already_current, + not_applicable = snapshot.not_applicable, + failed = snapshot.failed, + "KMS rekey sweep finished" + ); +} + +async fn sweep_bucket(store: &Arc, job: &Arc, bucket: &str, prefix: &str) { + let (tx, mut rx) = tokio::sync::mpsc::channel(256); + let walk_cancel = job.cancel.child_token(); + // Every version's envelope needs rewrapping, not just the latest. + type WalkOptionsOf = ::WalkOptions; + let walk_options = WalkOptionsOf { + latest_only: false, + ..Default::default() + }; + let walk_store = store.clone(); + let walk_bucket = bucket.to_string(); + let walk_prefix = prefix.to_string(); + let walk_task = tokio::spawn(async move { + walk_store + .walk(walk_cancel, &walk_bucket, &walk_prefix, tx, walk_options) + .await + }); + + while let Some(entry) = rx.recv().await { + if job.cancel.is_cancelled() { + break; + } + if let Some(error) = entry.err { + job.failed.fetch_add(1, Ordering::Relaxed); + warn!( + component = LOG_COMPONENT, + subsystem = LOG_SUBSYSTEM, + event = "kms_rekey_walk_entry_failed", + job_id = %job.id, + bucket, + error = %error, + "KMS rekey sweep could not list an object" + ); + continue; + } + let Some(object_info) = entry.item else { + continue; + }; + if object_info.delete_marker || object_info.is_dir { + continue; + } + job.scanned.fetch_add(1, Ordering::Relaxed); + + let object = object_info.name.clone(); + match rewrap_object_encryption_metadata(bucket, &object, object_info.user_defined.as_ref()).await { + Ok(ObjectDekRewrapOutcome::NotApplicable) => { + job.not_applicable.fetch_add(1, Ordering::Relaxed); + } + Ok(ObjectDekRewrapOutcome::AlreadyCurrent) => { + job.already_current.fetch_add(1, Ordering::Relaxed); + } + Ok(ObjectDekRewrapOutcome::Rewrapped { metadata }) => { + let options = StorageObjectOptions { + version_id: object_info.version_id.map(|version| version.to_string()), + eval_metadata: Some(metadata), + ..Default::default() + }; + match store.put_object_metadata(bucket, &object, &options).await { + Ok(_) => { + job.rewrapped.fetch_add(1, Ordering::Relaxed); + } + Err(error) => { + job.failed.fetch_add(1, Ordering::Relaxed); + warn!( + component = LOG_COMPONENT, + subsystem = LOG_SUBSYSTEM, + event = "kms_rekey_persist_failed", + job_id = %job.id, + bucket, + object = %object, + error = %error, + "KMS rekey sweep rewrapped an envelope but could not persist it; the object keeps its \ + previous (still valid) wrapping and a re-run retries it" + ); + } + } + } + Err(error) => { + job.failed.fetch_add(1, Ordering::Relaxed); + warn!( + component = LOG_COMPONENT, + subsystem = LOG_SUBSYSTEM, + event = "kms_rekey_object_failed", + job_id = %job.id, + bucket, + object = %object, + error = %error, + "KMS rekey sweep could not rewrap an object's data key; a re-run retries it" + ); + } + } + } + // Dropping the receiver ends the walk on early break; a cancelled walk is + // not a sweep failure. + drop(rx); + match walk_task.await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + if !job.cancel.is_cancelled() { + job.failed.fetch_add(1, Ordering::Relaxed); + warn!( + component = LOG_COMPONENT, + subsystem = LOG_SUBSYSTEM, + event = "kms_rekey_walk_failed", + job_id = %job.id, + bucket, + error = %error, + "KMS rekey sweep walk ended with an error; coverage of this bucket is incomplete" + ); + } + } + Err(error) => { + job.failed.fetch_add(1, Ordering::Relaxed); + warn!( + component = LOG_COMPONENT, + subsystem = LOG_SUBSYSTEM, + event = "kms_rekey_walk_failed", + job_id = %job.id, + bucket, + error = %error, + "KMS rekey sweep walk task failed; coverage of this bucket is incomplete" + ); + } + } +} diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index 64c4dbf05..1b6fac1fa 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -89,6 +89,7 @@ pub mod error; pub mod init; pub mod inspect; pub(crate) mod kms_deletion_gate; +pub(crate) mod kms_rekey; pub mod license; pub mod memory_observability; pub mod module_switches; diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index c33a18f24..5e85dc080 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -109,10 +109,10 @@ pub(crate) use super::ecfs_extend::{ validate_list_object_unordered_with_delimiter, validate_object_key, wrap_response_with_cors, }; pub(crate) use super::sse::{ - DecryptionRequest, EncryptionRequest, PrepareEncryptionRequest, SseKmsPrincipal, authorize_sse_kms_object_read, - classify_sse_read_response, extract_server_side_encryption_from_headers, sse_decryption, sse_encryption, - sse_prepare_encryption, strip_managed_encryption_metadata, validate_sse_headers_for_read, validate_sse_headers_for_write, - validate_ssec_for_read, + DecryptionRequest, EncryptionRequest, ObjectDekRewrapOutcome, PrepareEncryptionRequest, SseKmsPrincipal, + authorize_sse_kms_object_read, classify_sse_read_response, extract_server_side_encryption_from_headers, + rewrap_object_encryption_metadata, sse_decryption, sse_encryption, sse_prepare_encryption, strip_managed_encryption_metadata, + validate_sse_headers_for_read, validate_sse_headers_for_write, validate_ssec_for_read, }; pub(crate) mod access_consumer { diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 6c12018b6..cbd2ab165 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -87,9 +87,19 @@ pub(crate) mod kms { pub(crate) mod bucket { pub(crate) use super::super::super::storage_contracts::{BucketOperations, BucketOptions}; } + + pub(crate) mod list { + pub(crate) use super::super::super::storage_contracts::ListOperations; + } + + pub(crate) mod object { + pub(crate) use super::super::super::storage_contracts::ObjectOperations; + } } pub(crate) use crate::storage::storage_api::{ECStore, StorageError}; + pub(crate) type StorageObjectOptions = crate::storage::storage_api::StorageObjectOptions; + pub(crate) use crate::storage::storage_api::{ObjectDekRewrapOutcome, rewrap_object_encryption_metadata}; /// Bucket SSE configuration for the KMS deletion reference gate; /// `Err(StorageError::ConfigNotFound)` when the bucket has none.