From 2fdcc980b5a6f6dff5df34232994216b24db76b6 Mon Sep 17 00:00:00 2001 From: Shurik <10572093+shurkus@users.noreply.github.com> Date: Mon, 19 Oct 2026 19:03:09 +0300 Subject: [PATCH 1/3] feat(provisioning): declare default bucket encryption Buckets accept `encryption` with `mode: SSE-S3` (S3 `AES256`) or `mode: SSE-KMS` (`aws:kms`, optional `kmsKeyId`), and `state: Absent` to remove the configuration. Ownership follows objectLockConfiguration and lifecycle: omission makes no encryption calls, a matching live configuration is adopted without a write, and a different one is replaced or removed only while its hash matches the configuration last recorded in status; otherwise the bucket reports BucketEncryptionConflict. SSE-KMS requires `spec.encryption.enabled`; without it RustFS answers PutBucketEncryption with 500 "KMS service not initialized", which would otherwise keep the bucket Pending forever. RustFS records its default KMS key when none is sent, so ownership is the hash of the verified live configuration, and SSE-KMS without `kmsKeyId` matches any key. RustFS answers GetBucketEncryption without a configuration with 400 ServerSideEncryptionConfigurationNotFoundError rather than 404; the client treats that code as absent. Closes #284 --- CHANGELOG.md | 2 + console-web/types/api.ts | 9 + crates/rustfs-admin/src/helpers.rs | 1 + crates/rustfs-admin/src/lib.rs | 6 + crates/rustfs-admin/src/s3_ops.rs | 294 ++++++++ deploy/rustfs-operator/README.md | 2 +- deploy/rustfs-operator/crds/tenant-crd.yaml | 43 ++ docs/operator-user-guide.md | 4 + docs/operator-user-guide.zh-CN.md | 4 + src/console/openapi.rs | 27 +- src/reconcile/provisioning.rs | 733 +++++++++++++++++++- src/sts/rustfs_client.rs | 1 + src/types/v1alpha1.rs | 26 + src/types/v1alpha1/provisioning.rs | 51 ++ src/types/v1alpha1/status.rs | 20 + src/types/v1alpha1/status/provisioning.rs | 8 + 16 files changed, 1218 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a2b823c..35ca8a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,8 @@ All notable changes to RustFS Operator are documented in this file. The format i - Tenant `spec.tiers` for declarative S3, RustFS, and MinIO remote tiers with Secret-sourced credentials, rotation on Secret updates, and conflict reporting for immutable tier settings. - Declarative Tenant bucket versioning and Object Lock default retention with drift-aware ownership. +- Declarative default bucket encryption (`SSE-S3` or `SSE-KMS`) with drift-aware ownership and + explicit removal. - Helm chart `cosiDriver.enabled` to deploy the RustFS COSI (`rustfs.objectstorage.k8s.io`) driver alongside the upstream provisioner sidecar, with dedicated ServiceAccount/RBAC. - Seven built-in RustFS Grafana dashboards, published by default as sidecar-discoverable diff --git a/console-web/types/api.ts b/console-web/types/api.ts index 7c611e3..4866b9a 100644 --- a/console-web/types/api.ts +++ b/console-web/types/api.ts @@ -130,6 +130,7 @@ export interface ProvisioningBucket { objectLockConfiguration?: BucketObjectLockConfiguration anonymous?: "Private" | "Download" | "Upload" | "Public" policy?: PolicyDocumentSource + encryption?: BucketEncryptionSpec lifecycle?: BucketLifecycleSpec deletionPolicy?: ProvisioningDeletionPolicy } @@ -154,6 +155,12 @@ export interface BucketObjectLockConfiguration { days?: number } +export interface BucketEncryptionSpec { + state?: "Present" | "Absent" + mode?: "SSE-S3" | "SSE-KMS" + kmsKeyId?: string +} + export interface BucketLifecycleSpec { state: "Present" | "Absent" rules?: BucketLifecycleRule[] @@ -204,6 +211,8 @@ export interface ProvisioningBucketStatus extends ProvisioningItemStatus { versioning?: "Unversioned" | "Enabled" | "Suspended" | null objectLockConfigurationDesiredHash?: string | null objectLockConfigurationLastAppliedHash?: string | null + encryptionDesiredHash?: string | null + encryptionLastAppliedHash?: string | null } export interface ProvisioningStatus { diff --git a/crates/rustfs-admin/src/helpers.rs b/crates/rustfs-admin/src/helpers.rs index 624a956..2e740fa 100644 --- a/crates/rustfs-admin/src/helpers.rs +++ b/crates/rustfs-admin/src/helpers.rs @@ -101,6 +101,7 @@ pub(super) fn body_mentions_not_found(body: &str) -> bool { || body.contains("nosuchpolicy") || body.contains("no such policy") || body.contains("objectlockconfigurationnotfound") + || body.contains("serversideencryptionconfigurationnotfound") || body.contains("nosuchbucketpolicy") || body.contains("no such bucket policy") || body.contains("not found") diff --git a/crates/rustfs-admin/src/lib.rs b/crates/rustfs-admin/src/lib.rs index 352a6fa..f067484 100644 --- a/crates/rustfs-admin/src/lib.rs +++ b/crates/rustfs-admin/src/lib.rs @@ -33,6 +33,7 @@ mod pool_ops; #[path = "s3_ops.rs"] mod s3_ops; pub use s3_ops::{ + BucketEncryptionAlgorithm, BucketEncryptionConfiguration, BucketLifecycleAbortIncompleteMultipartUpload, BucketLifecycleConfiguration, BucketLifecycleExpiration, BucketLifecycleNoncurrentVersionTransition, BucketLifecycleRule, BucketLifecycleRuleStatus, BucketLifecycleTransition, BucketObjectLockConfiguration, @@ -257,6 +258,7 @@ pub enum RustfsClientError { }, InvalidBucketVersioningResponse, InvalidObjectLockConfigurationResponse, + InvalidBucketEncryptionResponse, InvalidLifecycleConfigurationResponse, ParseResponseFailed, SigningFailed, @@ -309,6 +311,9 @@ impl std::fmt::Display for RustfsClientError { Self::InvalidObjectLockConfigurationResponse => { write!(f, "invalid bucket Object Lock configuration response") } + Self::InvalidBucketEncryptionResponse => { + write!(f, "invalid bucket encryption configuration response") + } Self::ParseResponseFailed => write!(f, "failed to parse AssumeRole response"), Self::SigningFailed => write!(f, "failed to compute request signature"), } @@ -361,6 +366,7 @@ impl RustfsClientError { | Self::RequestBuildFailed | Self::InvalidBucketVersioningResponse | Self::InvalidObjectLockConfigurationResponse + | Self::InvalidBucketEncryptionResponse | Self::InvalidLifecycleConfigurationResponse | Self::SigningFailed => false, } diff --git a/crates/rustfs-admin/src/s3_ops.rs b/crates/rustfs-admin/src/s3_ops.rs index ff43c3e..0ad6ec8 100644 --- a/crates/rustfs-admin/src/s3_ops.rs +++ b/crates/rustfs-admin/src/s3_ops.rs @@ -14,6 +14,7 @@ //! S3 boundary: //! - bucket lifecycle methods (create/lookup features) +//! - default bucket encryption (`?encryption`) //! - request semantics for S3-style object storage operations. use super::helpers::{ @@ -140,6 +141,52 @@ impl BucketObjectLockConfiguration { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BucketEncryptionAlgorithm { + /// SSE-S3 (`AES256`). + Aes256, + /// SSE-KMS (`aws:kms`). + AwsKms, +} + +impl BucketEncryptionAlgorithm { + const fn as_s3_str(self) -> &'static str { + match self { + Self::Aes256 => "AES256", + Self::AwsKms => "aws:kms", + } + } +} + +/// Default bucket encryption (`ServerSideEncryptionConfiguration` with a single rule). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BucketEncryptionConfiguration { + pub algorithm: BucketEncryptionAlgorithm, + pub kms_master_key_id: Option, + pub bucket_key_enabled: bool, +} + +impl BucketEncryptionConfiguration { + pub fn to_xml(&self) -> String { + let mut xml = format!( + "" + ); + xml.push_str(self.algorithm.as_s3_str()); + xml.push_str(""); + if let Some(key_id) = &self.kms_master_key_id { + xml.push_str(""); + xml.push_str(&escape_xml(key_id)); + xml.push_str(""); + } + xml.push_str(""); + if self.bucket_key_enabled { + xml.push_str("true"); + } + xml.push_str(""); + xml + } +} + #[derive(Debug)] struct StrictXmlElement { name: String, @@ -257,6 +304,58 @@ fn parse_bucket_object_lock_xml( }) } +fn parse_bucket_encryption_xml( + xml: &str, +) -> Result { + let invalid = || RustfsClientError::InvalidBucketEncryptionResponse; + let root = parse_strict_configuration_xml(xml, "ServerSideEncryptionConfiguration") + .map_err(|()| invalid())?; + root.validate_children(&["Rule"]).map_err(|()| invalid())?; + let rule = root + .child("Rule") + .map_err(|()| invalid())? + .ok_or_else(invalid)?; + rule.validate_children(&["ApplyServerSideEncryptionByDefault", "BucketKeyEnabled"]) + .map_err(|()| invalid())?; + let by_default = rule + .child("ApplyServerSideEncryptionByDefault") + .map_err(|()| invalid())? + .ok_or_else(invalid)?; + by_default + .validate_children(&["SSEAlgorithm", "KMSMasterKeyID"]) + .map_err(|()| invalid())?; + let algorithm = match by_default + .child("SSEAlgorithm") + .map_err(|()| invalid())? + .ok_or_else(invalid)? + .leaf_text() + .map_err(|()| invalid())? + { + "AES256" => BucketEncryptionAlgorithm::Aes256, + "aws:kms" => BucketEncryptionAlgorithm::AwsKms, + _ => return Err(invalid()), + }; + let kms_master_key_id = match by_default.child("KMSMasterKeyID").map_err(|()| invalid())? { + Some(key_id) if key_id.text.trim().is_empty() && key_id.children.is_empty() => None, + Some(key_id) => Some(key_id.leaf_text().map_err(|()| invalid())?.to_string()), + None => None, + }; + let bucket_key_enabled = match rule.child("BucketKeyEnabled").map_err(|()| invalid())? { + Some(enabled) => match enabled.leaf_text().map_err(|()| invalid())? { + "true" => true, + "false" => false, + _ => return Err(invalid()), + }, + None => false, + }; + + Ok(BucketEncryptionConfiguration { + algorithm, + kms_master_key_id, + bucket_key_enabled, + }) +} + fn parse_positive_i32(value: &str) -> Option { value.parse::().ok().filter(|value| *value > 0) } @@ -1034,6 +1133,139 @@ impl RustfsAdminClient { .map(|configuration| configuration.is_some()) } + pub async fn get_bucket_encryption( + &self, + bucket: &str, + ) -> Result, RustfsClientError> { + if bucket.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let query = build_canonical_query(&[("encryption", "")]); + let signed = self.sign_request("GET", &path, &query, "", None, ADMIN_SIGNING_SERVICE)?; + let host = self.host()?; + let response = self + .http_client + .get(format!( + "{}{}?{query}", + self.base_url.trim_end_matches('/'), + path + )) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + let body = read_bucket_configuration_response(response, || { + RustfsClientError::InvalidBucketEncryptionResponse + }) + .await?; + return parse_bucket_encryption_xml(&body).map(Some); + } + + // RustFS answers a missing configuration with 400 ServerSideEncryptionConfigurationNotFoundError. + let status = response.status(); + let (body, truncated) = RustfsClientError::limited_response_body(response).await; + if is_absent_resource(status, &body) { + return Ok(None); + } + Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )) + } + + pub async fn put_bucket_encryption( + &self, + bucket: &str, + configuration: &BucketEncryptionConfiguration, + ) -> Result<(), RustfsClientError> { + if bucket.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let query = build_canonical_query(&[("encryption", "")]); + let body = configuration.to_xml(); + let content_md5 = content_md5(&body); + let signed = self.sign_request_with_extra_headers( + "PUT", + &path, + &query, + &body, + ADMIN_SIGNING_SERVICE, + &[ + ("content-md5", content_md5.as_str()), + ("content-type", "application/xml"), + ], + )?; + let host = self.host()?; + let response = self + .http_client + .put(format!( + "{}{}?{query}", + self.base_url.trim_end_matches('/'), + path + )) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .header("content-md5", content_md5) + .header("content-type", "application/xml") + .body(body) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + Ok(()) + } else { + Err(RustfsClientError::unexpected_response(response).await) + } + } + + pub async fn delete_bucket_encryption(&self, bucket: &str) -> Result<(), RustfsClientError> { + if bucket.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let query = build_canonical_query(&[("encryption", "")]); + let signed = self.sign_request("DELETE", &path, &query, "", None, ADMIN_SIGNING_SERVICE)?; + let host = self.host()?; + let response = self + .http_client + .delete(format!( + "{}{}?{query}", + self.base_url.trim_end_matches('/'), + path + )) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + return Ok(()); + } + let status = response.status(); + let (body, truncated) = RustfsClientError::limited_response_body(response).await; + if is_absent_resource(status, &body) { + return Ok(()); + } + Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )) + } + pub async fn put_bucket_policy( &self, bucket: &str, @@ -1415,6 +1647,68 @@ mod tests { )); } + #[test] + fn bucket_encryption_xml_matches_rustfs_get_response_shape() { + // GetBucketEncryption body from RustFS 1.0.2-preview.2 after `PUT ?encryption` with AES256. + let rustfs_get = "AES256"; + let sse_s3 = BucketEncryptionConfiguration { + algorithm: BucketEncryptionAlgorithm::Aes256, + kms_master_key_id: None, + bucket_key_enabled: false, + }; + assert_eq!(parse_bucket_encryption_xml(rustfs_get).unwrap(), sse_s3); + assert_eq!( + sse_s3.to_xml(), + "AES256" + ); + assert_eq!( + parse_bucket_encryption_xml(&sse_s3.to_xml()).unwrap(), + sse_s3 + ); + + let sse_kms = BucketEncryptionConfiguration { + algorithm: BucketEncryptionAlgorithm::AwsKms, + kms_master_key_id: Some("app-key".to_string()), + bucket_key_enabled: true, + }; + assert!( + sse_kms + .to_xml() + .contains("app-key") + ); + assert_eq!( + parse_bucket_encryption_xml(&sse_kms.to_xml()).unwrap(), + sse_kms + ); + } + + #[test] + fn bucket_encryption_xml_rejects_unknown_ambiguous_or_missing_fields() { + for xml in [ + "", + "", + "AES128", + "AES256AES256", + "AES256", + "AES256yes", + "", + ] { + assert!( + matches!( + parse_bucket_encryption_xml(xml), + Err(RustfsClientError::InvalidBucketEncryptionResponse) + ), + "{xml} must be rejected" + ); + } + let empty_key = parse_bucket_encryption_xml( + "aws:kmsfalse", + ) + .unwrap(); + assert_eq!(empty_key.kms_master_key_id, None); + assert!(!empty_key.bucket_key_enabled); + } + #[test] fn lifecycle_xml_is_stable_and_escapes_values() { let configuration = BucketLifecycleConfiguration { diff --git a/deploy/rustfs-operator/README.md b/deploy/rustfs-operator/README.md index 43a8634..68c90aa 100755 --- a/deploy/rustfs-operator/README.md +++ b/deploy/rustfs-operator/README.md @@ -336,7 +336,7 @@ spec: days: 30 ``` -Policy ConfigMaps, user Secrets, and tier Secrets must live in the Tenant namespace. `users[].credsSecret.name` selects the credentials Secret; when omitted, the operator falls back to a Secret named after `users[].name` for compatibility with existing manifests. Bucket versioning can be enabled or suspended. Object Lock automatically requires enabled versioning and cannot be disabled after activation; `objectLockConfiguration` manages an optional default `Governance` or `Compliance` retention period in days. Omitting the configuration leaves an existing rule unmanaged, while `state: Absent` removes only an operator-owned default rule. Bucket lifecycle supports expiration, incomplete multipart upload cleanup, and transition of current or noncurrent versions to a remote tier registered in RustFS. `spec.tiers` registers such tiers before buckets are reconciled; tier credentials come from `tiers[].credsSecret` and are rotated when the Secret changes, while a live tier with different immutable settings is reported as a conflict. Omission leaves lifecycle unmanaged; `state: Absent` deletes only a configuration whose live hash still matches the Operator's ownership record. The operator indexes references from Tenant specs, so creating or updating a referenced object enqueues every referencing Tenant without requiring or mutating labels or requiring write access to that object. Provisioned resources are retained when removed from the Tenant spec. +Policy ConfigMaps, user Secrets, and tier Secrets must live in the Tenant namespace. `users[].credsSecret.name` selects the credentials Secret; when omitted, the operator falls back to a Secret named after `users[].name` for compatibility with existing manifests. Bucket versioning can be enabled or suspended. Object Lock automatically requires enabled versioning and cannot be disabled after activation; `objectLockConfiguration` manages an optional default `Governance` or `Compliance` retention period in days. Omitting the configuration leaves an existing rule unmanaged, while `state: Absent` removes only an operator-owned default rule. `encryption` sets the bucket default encryption (`SSE-S3`, or `SSE-KMS` with an optional `kmsKeyId` when the Tenant enables KMS) with the same ownership rules. Bucket lifecycle supports expiration, incomplete multipart upload cleanup, and transition of current or noncurrent versions to a remote tier registered in RustFS. `spec.tiers` registers such tiers before buckets are reconciled; tier credentials come from `tiers[].credsSecret` and are rotated when the Secret changes, while a live tier with different immutable settings is reported as a conflict. Omission leaves lifecycle unmanaged; `state: Absent` deletes only a configuration whose live hash still matches the Operator's ownership record. The operator indexes references from Tenant specs, so creating or updating a referenced object enqueues every referencing Tenant without requiring or mutating labels or requiring write access to that object. Provisioned resources are retained when removed from the Tenant spec. ### RBAC Configuration diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml index 74a2934..ee4f55c 100644 --- a/deploy/rustfs-operator/crds/tenant-crd.yaml +++ b/deploy/rustfs-operator/crds/tenant-crd.yaml @@ -1235,6 +1235,43 @@ spec: enum: - Retain type: string + encryption: + description: |- + Default bucket encryption. Omission leaves the live configuration unmanaged; `Absent` + removes only a configuration previously owned by the operator. + nullable: true + properties: + kmsKeyId: + description: KMS key for `SSE-KMS`. Omission uses the RustFS default KMS key. + maxLength: 1024 + minLength: 1 + nullable: true + pattern: ^[A-Za-z0-9._:/+=@-]+$ + type: string + mode: + description: |- + `SSE-S3` stores S3 `AES256`; `SSE-KMS` stores `aws:kms` and requires + `spec.encryption.enabled` on the Tenant. + enum: + - SSE-S3 + - SSE-KMS + - null + nullable: true + type: string + state: + description: Omission is equivalent to `Present`. + enum: + - Present + - Absent + - null + nullable: true + type: string + type: object + x-kubernetes-validations: + - message: Present bucket encryption requires mode; Absent forbids mode and kmsKeyId + rule: '(!has(self.state) || self.state == ''Present'') ? has(self.mode) : !has(self.mode) && !has(self.kmsKeyId)' + - message: kmsKeyId requires mode SSE-KMS + rule: '!has(self.kmsKeyId) || (has(self.mode) && self.mode == ''SSE-KMS'')' lifecycle: description: |- Declarative S3 bucket lifecycle configuration. Omission leaves the live lifecycle @@ -3997,6 +4034,12 @@ spec: desiredHash: nullable: true type: string + encryptionDesiredHash: + nullable: true + type: string + encryptionLastAppliedHash: + nullable: true + type: string lastAppliedAccessKeyHash: nullable: true type: string diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md index cdeab96..78010dd 100644 --- a/docs/operator-user-guide.md +++ b/docs/operator-user-guide.md @@ -962,6 +962,8 @@ The operator can create RustFS policies, users, remote tiers, and buckets after Bucket versioning and Object Lock are declarative but conservative. `versioning: true` enables versioning. `versioning: false` keeps a never-versioned bucket unchanged or requests S3 `Suspended` after versioning has been enabled; omitting the field leaves versioning unmanaged unless Object Lock requires it. `objectLock: true` enables Object Lock after versioning is enabled. Object Lock cannot be disabled, and the Operator refuses `objectLock: false` or versioning suspension when the live bucket has Object Lock enabled. `objectLockConfiguration` manages the default retention rule: omit `state` for `Present`, specify `mode: Governance` or `Compliance` with `days`, or use `state: Absent` without `mode` and `days` to remove only the default rule. Omitting the entire field leaves a live default-retention rule unmanaged. The Operator adopts an identical live rule, but does not replace or remove a different rule unless its hash matches the last configuration recorded in status. Default retention applies to new object versions; it does not retroactively lock existing versions. +Bucket default encryption is declared with `encryption`. `mode: SSE-S3` stores S3 `AES256`; `mode: SSE-KMS` stores `aws:kms` with an optional `kmsKeyId` (RustFS uses its default KMS key when omitted) and requires `spec.encryption.enabled` on the Tenant. Default encryption applies to objects written after it is set; it does not re-encrypt existing objects. `state: Absent` without `mode` and `kmsKeyId` removes the configuration. The Operator adopts a matching live configuration, but never replaces or removes a different one unless its hash matches the configuration last recorded in `status.provisioning.buckets`. Omitting `encryption` performs no encryption API calls and leaves the live configuration untouched. + Each bucket may also declare `lifecycle.state: Present` with one or more S3 lifecycle rules. The current schema supports expiration after a positive number of days, aborting incomplete multipart uploads after a positive number of days, `transition` of current versions after zero or more days, and `noncurrentVersionTransition` of noncurrent versions after zero or more days; every rule has a unique ID, an `Enabled` or `Disabled` status, and an explicit prefix filter (use `prefix: ""` for all objects). `lifecycle.state: Absent` requests deletion and cannot include rules. The Operator adopts an identical live configuration, but never replaces or deletes a different configuration without a matching ownership hash in `status.provisioning.buckets`. Omitting `lifecycle` performs no lifecycle API calls and leaves the live configuration untouched. Transition actions move data to a remote tier. `storageClass` is a tier name: a tier declared in `spec.tiers`, or one registered in RustFS by other means (for example with `rc ilm tier add`). Tier names use only uppercase ASCII letters, digits, `_`, and `-`, and `STANDARD` and `REDUCED_REDUNDANCY` are reserved. Before applying a lifecycle with transition actions the Operator checks that every referenced tier exists; while one is missing, the bucket stays `Pending` with a message naming the tier and the Operator retries, so the tier can be registered after the rule is added. A referenced tier declared in `spec.tiers` must also be `Ready`: while it is pending the bucket stays `Pending`, and when it has failed, for example with `TierConflict`, the bucket fails with a message naming the tier. @@ -1081,6 +1083,8 @@ spec: objectLockConfiguration: mode: Compliance days: 30 + encryption: + mode: SSE-S3 anonymous: Download lifecycle: state: Present diff --git a/docs/operator-user-guide.zh-CN.md b/docs/operator-user-guide.zh-CN.md index 7336b4f..963db5c 100644 --- a/docs/operator-user-guide.zh-CN.md +++ b/docs/operator-user-guide.zh-CN.md @@ -906,6 +906,8 @@ Operator 可以在 Tenant workload Ready 后自动创建 RustFS policy、user、 Bucket 版本控制和 Object Lock 采用声明式但保守的调和策略。`versioning: true` 启用版本控制。`versioning: false` 会让从未启用版本控制的 bucket 保持原状，或在版本控制已启用后请求 S3 `Suspended`；省略该字段时保持现状，除非 Object Lock 要求先启用版本控制。`objectLock: true` 会在版本控制启用后开启 Object Lock。Object Lock 不可关闭；当线上 bucket 已启用 Object Lock 时，Operator 会拒绝 `objectLock: false` 或暂停版本控制。`objectLockConfiguration` 管理默认保留规则：省略 `state` 等同于 `Present`，此时必须设置 `mode: Governance` 或 `Compliance` 以及 `days`；`state: Absent` 必须省略 `mode` 和 `days`，仅移除默认规则。完全省略 `objectLockConfiguration` 时，不管理已有默认保留规则。Operator 可以接管与声明完全相同的线上规则，但只有线上配置哈希仍与 status 中最后一次应用的配置一致时，才会替换或移除不同规则。默认保留策略只作用于之后创建的对象版本，不会追溯锁定已有版本。 +Bucket 默认加密通过 `encryption` 声明。`mode: SSE-S3` 写入 S3 `AES256`；`mode: SSE-KMS` 写入 `aws:kms`，可选 `kmsKeyId`（省略时 RustFS 使用默认 KMS key），并要求 Tenant 设置 `spec.encryption.enabled`。默认加密只作用于设置之后写入的对象，不会重新加密已有对象。`state: Absent`（不能包含 `mode` 和 `kmsKeyId`）表示移除该配置。Operator 可以接管与声明一致的线上配置，但只有线上配置哈希与 `status.provisioning.buckets` 中最后记录的配置一致时，才会替换或移除不同的配置。省略 `encryption` 时不会调用加密 API，也不会改动现有配置。 + 每个 bucket 还可以通过 `lifecycle.state: Present` 声明一个或多个 S3 生命周期规则。当前 schema 支持按正整数天数过期对象、按正整数天数清理未完成的分片上传、通过 `transition` 在零天或更多天后迁移当前版本，以及通过 `noncurrentVersionTransition` 在零天或更多天后迁移非当前版本；每条规则必须有唯一 ID、`Enabled` 或 `Disabled` 状态，并显式指定 prefix filter（`prefix: ""` 表示全部对象）。`lifecycle.state: Absent` 表示请求删除，且不能包含规则。Operator 可以接管与声明完全相同的现有配置，但没有 `status.provisioning.buckets` 中匹配的所有权哈希时，不会覆盖或删除不同的现有配置。省略 `lifecycle` 时不会调用 Lifecycle API，也不会改动现有配置。 迁移类操作会把数据移动到远程层（tier）。`storageClass` 是 tier 名称：可以是 `spec.tiers` 中声明的 tier，也可以是通过其他方式（例如 `rc ilm tier add`）在 RustFS 中注册的 tier。tier 名称只能包含大写 ASCII 字母、数字、`_` 和 `-`，`STANDARD` 和 `REDUCED_REDUNDANCY` 为保留名称。在应用包含迁移操作的生命周期配置之前，Operator 会检查所有被引用的 tier 是否存在；如果有 tier 尚未注册，bucket 会保持 `Pending` 状态并在消息中列出该 tier，Operator 会持续重试，因此可以在添加规则之后再注册 tier。被引用且在 `spec.tiers` 中声明的 tier 还必须处于 `Ready` 状态：该 tier 处于等待状态时 bucket 保持 `Pending`；该 tier 失败（例如 `TierConflict`）时 bucket 也会失败，并在消息中列出该 tier。 @@ -1025,6 +1027,8 @@ spec: objectLockConfiguration: mode: Compliance days: 30 + encryption: + mode: SSE-S3 anonymous: Download lifecycle: state: Present diff --git a/src/console/openapi.rs b/src/console/openapi.rs index 191b5d2..6574f68 100644 --- a/src/console/openapi.rs +++ b/src/console/openapi.rs @@ -55,14 +55,14 @@ use crate::console::models::topology::{ TopologyOverviewResponse, TopologyPod, TopologyPool, TopologyTenant, TopologyTenantSummary, }; use crate::types::v1alpha1::provisioning::{ - BucketAnonymousAccess, BucketLifecycleAbortIncompleteMultipartUpload, - BucketLifecycleExpiration, BucketLifecycleNoncurrentVersionTransition, BucketLifecycleRule, - BucketLifecycleRuleFilter, BucketLifecycleRuleStatus, BucketLifecycleSpec, - BucketLifecycleState, BucketLifecycleTransition, BucketObjectLockConfiguration, - BucketObjectLockConfigurationState, BucketObjectLockRetentionMode, ConfigMapKeyReference, - PolicyDocumentSource, ProvisioningBucket, ProvisioningDeletionPolicy, ProvisioningPolicy, - ProvisioningTier, ProvisioningTierType, ProvisioningUser, TierCredentialsSecretRef, - UserCredentialsSecretRef, + BucketAnonymousAccess, BucketEncryptionMode, BucketEncryptionSpec, BucketEncryptionState, + BucketLifecycleAbortIncompleteMultipartUpload, BucketLifecycleExpiration, + BucketLifecycleNoncurrentVersionTransition, BucketLifecycleRule, BucketLifecycleRuleFilter, + BucketLifecycleRuleStatus, BucketLifecycleSpec, BucketLifecycleState, + BucketLifecycleTransition, BucketObjectLockConfiguration, BucketObjectLockConfigurationState, + BucketObjectLockRetentionMode, ConfigMapKeyReference, PolicyDocumentSource, ProvisioningBucket, + ProvisioningDeletionPolicy, ProvisioningPolicy, ProvisioningTier, ProvisioningTierType, + ProvisioningUser, TierCredentialsSecretRef, UserCredentialsSecretRef, }; use crate::types::v1alpha1::status::provisioning::{ BucketVersioningStatus, ProvisioningBucketStatus, ProvisioningItemState, @@ -141,6 +141,9 @@ use crate::types::v1alpha1::status::provisioning::{ BucketObjectLockConfiguration, BucketObjectLockConfigurationState, BucketObjectLockRetentionMode, + BucketEncryptionSpec, + BucketEncryptionState, + BucketEncryptionMode, BucketLifecycleSpec, BucketLifecycleState, BucketLifecycleRule, @@ -802,6 +805,7 @@ mod tests { assert!(schemas.contains_key("BucketLifecycleTransition")); assert!(schemas.contains_key("BucketLifecycleNoncurrentVersionTransition")); assert!(schemas.contains_key("BucketObjectLockConfiguration")); + assert!(schemas.contains_key("BucketEncryptionSpec")); assert!(schemas.contains_key("UserCredentialsSecretRef")); assert!(schemas.contains_key("ProvisioningTier")); assert!(schemas.contains_key("ProvisioningTierType")); @@ -852,6 +856,13 @@ mod tests { .and_then(Value::as_str), Some("#/components/schemas/BucketObjectLockConfiguration") ); + assert_eq!( + spec.pointer( + "/components/schemas/ProvisioningBucket/properties/encryption/oneOf/1/$ref", + ) + .and_then(Value::as_str), + Some("#/components/schemas/BucketEncryptionSpec") + ); assert_eq!( spec.pointer("/components/schemas/ProvisioningStatus/properties/buckets/items/$ref") .and_then(Value::as_str), diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index 354a749..6a43479 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -14,6 +14,8 @@ use crate::context::{self, Context}; use crate::sts::rustfs_client::{ + BucketEncryptionAlgorithm as AdminEncryptionAlgorithm, + BucketEncryptionConfiguration as AdminEncryptionConfiguration, BucketLifecycleAbortIncompleteMultipartUpload as AdminLifecycleAbort, BucketLifecycleConfiguration as AdminLifecycleConfiguration, BucketLifecycleExpiration as AdminLifecycleExpiration, @@ -29,11 +31,12 @@ use crate::sts::rustfs_client::{ RustfsTierConfig, canonicalize_bucket_lifecycle_xml, }; use crate::types::v1alpha1::provisioning::{ - BucketAnonymousAccess, BucketLifecycleRuleStatus, BucketLifecycleSpec, BucketLifecycleState, + BucketAnonymousAccess, BucketEncryptionMode, BucketEncryptionSpec, BucketEncryptionState, + BucketLifecycleRuleStatus, BucketLifecycleSpec, BucketLifecycleState, BucketObjectLockConfigurationState, BucketObjectLockRetentionMode, - MAX_BUCKET_LIFECYCLE_RULE_ID_LENGTH, MAX_BUCKET_LIFECYCLE_RULES, - MAX_BUCKET_OBJECT_LOCK_RETENTION_DAYS, PolicyDocumentSource, ProvisioningBucket, - ProvisioningPolicy, ProvisioningTier, ProvisioningUser, + MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH, MAX_BUCKET_LIFECYCLE_RULE_ID_LENGTH, + MAX_BUCKET_LIFECYCLE_RULES, MAX_BUCKET_OBJECT_LOCK_RETENTION_DAYS, PolicyDocumentSource, + ProvisioningBucket, ProvisioningPolicy, ProvisioningTier, ProvisioningUser, duplicate_user_credentials_secret_names, }; use crate::types::v1alpha1::status::Reason; @@ -2462,6 +2465,10 @@ async fn reconcile_bucket( .object_lock_configuration .as_ref() .and_then(object_lock_configuration_desired_hash); + status.encryption_desired_hash = bucket + .encryption + .as_ref() + .and_then(bucket_encryption_desired_hash); if status.state != ProvisioningItemState::Ready.as_str() { status.lifecycle_desired_hash = bucket.lifecycle.as_ref().and_then(lifecycle_desired_hash); return status; @@ -2487,6 +2494,12 @@ async fn reconcile_bucket( return status; } + status = reconcile_bucket_encryption(run, client, bucket, previous.as_ref(), status).await; + if status.state != ProvisioningItemState::Ready.as_str() { + status.lifecycle_desired_hash = bucket.lifecycle.as_ref().and_then(lifecycle_desired_hash); + return status; + } + let mut completed = reconcile_bucket_lifecycle(run, client, bucket, previous.as_ref(), status.item.clone()) .await; @@ -2998,6 +3011,11 @@ fn validate_bucket_configuration(bucket: &ProvisioningBucket) -> Result<(), (Rea { return Err((Reason::BucketLifecycleApplyFailed, message)); } + if let Some(spec) = bucket.encryption.as_ref() + && let Err(message) = bucket_encryption_configuration(spec) + { + return Err((Reason::BucketEncryptionApplyFailed, message)); + } Ok(()) } @@ -3094,6 +3112,262 @@ fn copy_bucket_control_status( source.object_lock_configuration_desired_hash.clone(); target.object_lock_configuration_last_applied_hash = source.object_lock_configuration_last_applied_hash.clone(); + target.encryption_desired_hash = source.encryption_desired_hash.clone(); + target.encryption_last_applied_hash = source.encryption_last_applied_hash.clone(); +} + +async fn reconcile_bucket_encryption( + run: &mut ProvisioningRun<'_>, + client: &RustfsAdminClient, + bucket: &ProvisioningBucket, + previous: Option<&ProvisioningBucketStatus>, + mut status: ProvisioningBucketStatus, +) -> ProvisioningBucketStatus { + let Some(spec) = bucket.encryption.as_ref() else { + return status; + }; + let desired = match bucket_encryption_configuration(spec) { + Ok(desired) => desired, + Err(message) => { + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionApplyFailed, + message, + ); + return status; + } + }; + if desired + .as_ref() + .is_some_and(|desired| desired.algorithm == AdminEncryptionAlgorithm::AwsKms) + && !run + .tenant + .spec + .encryption + .as_ref() + .is_some_and(|encryption| encryption.enabled) + { + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionApplyFailed, + "SSE-KMS bucket encryption requires spec.encryption.enabled on the Tenant", + ); + return status; + } + + let live = match client.get_bucket_encryption(&bucket.name).await { + Ok(live) => live, + Err(error) => { + status.item = bucket_item_from_admin_error( + run, + previous, + &status.item, + Reason::BucketEncryptionApplyFailed, + error, + "failed to read RustFS bucket encryption", + ); + return status; + } + }; + let live_hash = live + .as_ref() + .map(|configuration| hash_document(&configuration.to_xml())); + let last_applied_hash = previous.and_then(|item| item.encryption_last_applied_hash.as_deref()); + + let Some(desired) = desired else { + status.encryption_desired_hash = None; + let Some(live_hash) = live_hash else { + status.encryption_last_applied_hash = None; + status.item.message = Some(append_bucket_message( + &status.item, + "RustFS bucket encryption is absent", + )); + return status; + }; + if last_applied_hash != Some(live_hash.as_str()) { + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionConflict, + bucket_encryption_conflict_message(last_applied_hash, "delete"), + ); + return status; + } + if let Err(error) = client.delete_bucket_encryption(&bucket.name).await { + status.item = bucket_item_from_admin_error( + run, + previous, + &status.item, + Reason::BucketEncryptionApplyFailed, + error, + "failed to remove RustFS bucket encryption", + ); + return status; + } + status.encryption_last_applied_hash = None; + status.item.message = Some(append_bucket_message( + &status.item, + "RustFS bucket encryption was removed", + )); + return status; + }; + + if let Some(live) = live.as_ref() + && bucket_encryption_matches(&desired, live) + { + status.encryption_last_applied_hash = live_hash; + status.item.message = Some(append_bucket_message( + &status.item, + "RustFS bucket encryption matches spec", + )); + return status; + } + if let Some(live_hash) = live_hash.as_deref() + && last_applied_hash != Some(live_hash) + { + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionConflict, + bucket_encryption_conflict_message(last_applied_hash, "replace"), + ); + return status; + } + + if let Err(error) = client.put_bucket_encryption(&bucket.name, &desired).await { + status.item = bucket_item_from_admin_error( + run, + previous, + &status.item, + Reason::BucketEncryptionApplyFailed, + error, + "failed to apply RustFS bucket encryption", + ); + return status; + } + let verified = match client.get_bucket_encryption(&bucket.name).await { + Ok(Some(verified)) if bucket_encryption_matches(&desired, &verified) => verified, + Ok(_) => { + status.item = replace_bucket_item_state( + run, + previous, + &status.item, + ProvisioningItemState::Failed, + Reason::BucketEncryptionApplyFailed, + "RustFS bucket encryption did not match spec after update", + ); + return status; + } + Err(error) => { + status.item = bucket_item_from_admin_error( + run, + previous, + &status.item, + Reason::BucketEncryptionApplyFailed, + error, + "failed to verify RustFS bucket encryption after update", + ); + return status; + } + }; + // Ownership is the verified live document: RustFS fills in the default KMS key for SSE-KMS. + status.encryption_last_applied_hash = Some(hash_document(&verified.to_xml())); + status.item.message = Some(append_bucket_message( + &status.item, + "RustFS bucket encryption was applied", + )); + status +} + +fn bucket_encryption_conflict_message(last_applied_hash: Option<&str>, action: &str) -> String { + if last_applied_hash.is_some() { + format!( + "Live RustFS bucket encryption changed since the operator last applied it; refusing to {action} it" + ) + } else { + format!( + "Live RustFS bucket encryption is not owned by this status; refusing to {action} it" + ) + } +} + +/// `Ok(None)` is a valid `Absent` spec. +fn bucket_encryption_configuration( + spec: &BucketEncryptionSpec, +) -> Result, String> { + match spec.state() { + BucketEncryptionState::Absent => { + if spec.mode.is_some() || spec.kms_key_id.is_some() { + return Err( + "Absent bucket encryption must not contain mode or kmsKeyId".to_string() + ); + } + Ok(None) + } + BucketEncryptionState::Present => { + let algorithm = match spec.mode { + Some(BucketEncryptionMode::SseS3) => AdminEncryptionAlgorithm::Aes256, + Some(BucketEncryptionMode::SseKms) => AdminEncryptionAlgorithm::AwsKms, + None => return Err("Present bucket encryption must contain a mode".to_string()), + }; + if let Some(key_id) = spec.kms_key_id.as_deref() { + if algorithm != AdminEncryptionAlgorithm::AwsKms { + return Err("bucket encryption kmsKeyId requires mode SSE-KMS".to_string()); + } + if !is_valid_kms_key_id(key_id) { + return Err(format!( + "bucket encryption kmsKeyId must contain between 1 and {MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH} characters from [A-Za-z0-9._:/+=@-]" + )); + } + } + Ok(Some(AdminEncryptionConfiguration { + algorithm, + kms_master_key_id: spec.kms_key_id.clone(), + bucket_key_enabled: false, + })) + } + } +} + +fn is_valid_kms_key_id(key_id: &str) -> bool { + !key_id.is_empty() + && key_id.len() <= MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH as usize + && key_id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"._:/+=@-".contains(&byte)) +} + +fn bucket_encryption_desired_hash(spec: &BucketEncryptionSpec) -> Option { + bucket_encryption_configuration(spec) + .ok() + .flatten() + .map(|configuration| hash_document(&configuration.to_xml())) +} + +/// SSE-KMS without `kmsKeyId` accepts whatever default key RustFS recorded. +fn bucket_encryption_matches( + desired: &AdminEncryptionConfiguration, + live: &AdminEncryptionConfiguration, +) -> bool { + desired.algorithm == live.algorithm + && desired.bucket_key_enabled == live.bucket_key_enabled + && match desired.kms_master_key_id.as_deref() { + Some(key_id) => live.kms_master_key_id.as_deref() == Some(key_id), + None => { + desired.algorithm == AdminEncryptionAlgorithm::AwsKms + || live.kms_master_key_id.is_none() + } + } } async fn reconcile_bucket_lifecycle( @@ -4204,6 +4478,8 @@ fn reason_from_str(reason: &str) -> Reason { "BucketObjectLockConfigurationConflict" => Reason::BucketObjectLockConfigurationConflict, "BucketPolicyApplyFailed" => Reason::BucketPolicyApplyFailed, "BucketPolicyConflict" => Reason::BucketPolicyConflict, + "BucketEncryptionApplyFailed" => Reason::BucketEncryptionApplyFailed, + "BucketEncryptionConflict" => Reason::BucketEncryptionConflict, "BucketLifecycleApplyFailed" => Reason::BucketLifecycleApplyFailed, "BucketLifecycleConflict" => Reason::BucketLifecycleConflict, "TierSecretInvalid" => Reason::TierSecretInvalid, @@ -4266,8 +4542,10 @@ mod tests { existing: bool, ignore_versioning_put: bool, ignore_object_lock_put: bool, + ignore_encryption_put: bool, versioning: Arc>, object_lock: Arc>>, + encryption: Arc>>, requests: Arc>>, put_requests: Arc>>, create_object_lock_header: Arc>>, @@ -4287,8 +4565,10 @@ mod tests { existing: false, ignore_versioning_put: false, ignore_object_lock_put: false, + ignore_encryption_put: false, versioning: Arc::new(Mutex::new(BucketVersioningState::Unversioned)), object_lock: Arc::new(Mutex::new(None)), + encryption: Arc::new(Mutex::new(None)), requests: Arc::new(Mutex::new(Vec::new())), put_requests: Arc::new(Mutex::new(Vec::new())), create_object_lock_header: Arc::new(Mutex::new(None)), @@ -4445,6 +4725,49 @@ mod tests { } StatusCode::OK.into_response() } + (axum::http::Method::GET, "encryption=") => { + match capture.encryption.lock().await.clone() { + // RustFS omits the namespace on GetBucketEncryption. + Some(configuration) => ( + StatusCode::OK, + configuration + .to_xml() + .replace(" xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"", ""), + ) + .into_response(), + // RustFS answers a missing configuration with 400, not 404. + None => ( + StatusCode::BAD_REQUEST, + "ServerSideEncryptionConfigurationNotFoundErrorThe server-side encryption configuration was not found.", + ) + .into_response(), + } + } + (axum::http::Method::PUT, "encryption=") => { + let body = capture_bucket_control_put(&capture, &query, request).await; + let kms = body.contains("aws:kms"); + let kms_master_key_id = body + .split_once("") + .and_then(|(_, tail)| tail.split_once("")) + .map(|(key, _)| key.to_string()) + .or_else(|| kms.then(|| "rustfs-default-key".to_string())); + if !capture.ignore_encryption_put { + *capture.encryption.lock().await = Some(AdminEncryptionConfiguration { + algorithm: if kms { + AdminEncryptionAlgorithm::AwsKms + } else { + AdminEncryptionAlgorithm::Aes256 + }, + kms_master_key_id, + bucket_key_enabled: false, + }); + } + StatusCode::OK.into_response() + } + (axum::http::Method::DELETE, "encryption=") => { + *capture.encryption.lock().await = None; + StatusCode::NO_CONTENT.into_response() + } _ => StatusCode::METHOD_NOT_ALLOWED.into_response(), } } @@ -6699,6 +7022,408 @@ mod tests { server.abort(); } + fn sse_s3() -> AdminEncryptionConfiguration { + AdminEncryptionConfiguration { + algorithm: AdminEncryptionAlgorithm::Aes256, + kms_master_key_id: None, + bucket_key_enabled: false, + } + } + + fn encryption_bucket(encryption: serde_json::Value) -> ProvisioningBucket { + serde_json::from_value(serde_json::json!({ + "name": "secure-bucket", + "encryption": encryption + })) + .expect("encryption bucket should deserialize") + } + + fn previous_encryption_owner(hash: String) -> ProvisioningBucketStatus { + let mut previous = ProvisioningBucketStatus::new(ProvisioningItemStatus::new( + "secure-bucket", + ProvisioningItemState::Ready, + Reason::ProvisioningConfigured.as_str(), + )); + previous.encryption_last_applied_hash = Some(hash); + previous + } + + async fn requests_for(capture: &BucketControlCapture, request: &str) -> usize { + capture + .requests + .lock() + .await + .iter() + .filter(|logged| logged.as_str() == request) + .count() + } + + #[tokio::test] + async fn sse_s3_bucket_encryption_is_applied_to_a_new_bucket_and_owned() { + let capture = BucketControlCapture::default(); + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("RustFS bucket encryption was applied")) + ); + assert_eq!(capture.encryption.lock().await.clone(), Some(sse_s3())); + let owned = Some(hash_document(&sse_s3().to_xml())); + assert_eq!(item.encryption_desired_hash, owned); + assert_eq!(item.encryption_last_applied_hash, owned); + let put_requests = capture.put_requests.lock().await; + let put = put_requests + .iter() + .find(|request| request.query == "encryption=") + .expect("encryption PUT is captured"); + assert_eq!(put.body, sse_s3().to_xml()); + assert!(put.content_md5.is_some()); + assert!(put.authorization.as_deref().is_some_and(|authorization| { + authorization.contains( + "SignedHeaders=content-md5;content-type;host;x-amz-content-sha256;x-amz-date", + ) + })); + server.abort(); + } + + #[tokio::test] + async fn matching_live_bucket_encryption_is_adopted_without_a_write() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_s3()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!( + item.encryption_last_applied_hash, + Some(hash_document(&sse_s3().to_xml())) + ); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 0); + server.abort(); + } + + #[tokio::test] + async fn omitted_bucket_encryption_makes_no_encryption_calls() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_s3()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + let bucket: ProvisioningBucket = + serde_json::from_value(serde_json::json!({"name": "secure-bucket"})) + .expect("bucket should deserialize"); + + let item = reconcile_bucket(&mut run, &client, &bucket).await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert!( + !capture + .requests + .lock() + .await + .iter() + .any(|request| request.ends_with("encryption=")) + ); + assert_eq!(capture.encryption.lock().await.clone(), Some(sse_s3())); + server.abort(); + } + + #[tokio::test] + async fn unowned_live_bucket_encryption_is_not_replaced_or_removed() { + let foreign = AdminEncryptionConfiguration { + algorithm: AdminEncryptionAlgorithm::AwsKms, + kms_master_key_id: Some("foreign-key".to_string()), + bucket_key_enabled: false, + }; + for spec in [ + serde_json::json!({"mode": "SSE-S3"}), + serde_json::json!({"state": "Absent"}), + ] { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(foreign.clone()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket(&mut run, &client, &encryption_bucket(spec.clone())).await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str(), "{spec}"); + assert_eq!(item.reason, Reason::BucketEncryptionConflict.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("not owned by this status")) + ); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 0); + assert_eq!(requests_for(&capture, "DELETE encryption=").await, 0); + assert_eq!( + capture.encryption.lock().await.clone(), + Some(foreign.clone()) + ); + server.abort(); + } + } + + #[tokio::test] + async fn owned_bucket_encryption_follows_a_spec_change_to_sse_kms() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_s3()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let mut tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + tenant.spec.encryption = Some(crate::types::v1alpha1::encryption::EncryptionConfig { + enabled: true, + ..Default::default() + }); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(hash_document(&sse_s3().to_xml()))); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + let applied = AdminEncryptionConfiguration { + algorithm: AdminEncryptionAlgorithm::AwsKms, + kms_master_key_id: Some("rustfs-default-key".to_string()), + bucket_key_enabled: false, + }; + assert_eq!( + capture.encryption.lock().await.clone(), + Some(applied.clone()) + ); + assert_eq!( + item.encryption_last_applied_hash, + Some(hash_document(&applied.to_xml())) + ); + assert_ne!( + item.encryption_desired_hash, + item.encryption_last_applied_hash + ); + drop(run); + + // The RustFS-chosen default key is adopted on the next pass without another write. + let mut run = empty_run(&ctx, &tenant); + run.previous.buckets.push(item.clone()); + let again = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS"})), + ) + .await; + assert_eq!(again.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 1); + server.abort(); + } + + #[tokio::test] + async fn absent_bucket_encryption_removes_only_an_owned_configuration() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_s3()))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(hash_document(&sse_s3().to_xml()))); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"state": "Absent"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(capture.encryption.lock().await.clone(), None); + assert_eq!(item.encryption_desired_hash, None); + assert_eq!(item.encryption_last_applied_hash, None); + assert_eq!(requests_for(&capture, "DELETE encryption=").await, 1); + server.abort(); + } + + #[tokio::test] + async fn changed_owned_bucket_encryption_is_a_conflict() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(AdminEncryptionConfiguration { + bucket_key_enabled: true, + ..sse_s3() + }))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(hash_document(&sse_s3().to_xml()))); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionConflict.as_str()); + assert!( + item.message.as_deref().is_some_and( + |message| message.contains("changed since the operator last applied it") + ) + ); + assert_eq!( + item.encryption_last_applied_hash, + Some(hash_document(&sse_s3().to_xml())) + ); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 0); + server.abort(); + } + + #[tokio::test] + async fn sse_kms_bucket_encryption_requires_tenant_kms() { + let capture = BucketControlCapture { + existing: true, + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS", "kmsKeyId": "app-key"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionApplyFailed.as_str()); + assert!( + !capture + .requests + .lock() + .await + .iter() + .any(|request| request.ends_with("encryption=")) + ); + server.abort(); + } + + #[tokio::test] + async fn bucket_encryption_update_is_verified_after_a_success_response() { + let capture = BucketControlCapture { + existing: true, + ignore_encryption_put: true, + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionApplyFailed.as_str()); + assert_eq!(item.encryption_last_applied_hash, None); + server.abort(); + } + + #[test] + fn bucket_encryption_spec_is_validated_before_any_call() { + let spec = |value: serde_json::Value| -> BucketEncryptionSpec { + serde_json::from_value(value).expect("encryption spec deserializes") + }; + assert_eq!( + bucket_encryption_configuration(&spec(serde_json::json!({"mode": "SSE-S3"}))), + Ok(Some(sse_s3())) + ); + assert_eq!( + bucket_encryption_configuration(&spec(serde_json::json!({"state": "Absent"}))), + Ok(None) + ); + for invalid in [ + serde_json::json!({}), + serde_json::json!({"state": "Absent", "mode": "SSE-S3"}), + serde_json::json!({"mode": "SSE-S3", "kmsKeyId": "app-key"}), + serde_json::json!({"mode": "SSE-KMS", "kmsKeyId": "app key"}), + serde_json::json!({"mode": "SSE-KMS", "kmsKeyId": "app"}), + serde_json::json!({"mode": "SSE-KMS", "kmsKeyId": ""}), + ] { + assert!( + bucket_encryption_configuration(&spec(invalid.clone())).is_err(), + "{invalid} must be rejected" + ); + } + let bucket = encryption_bucket(serde_json::json!({"state": "Absent", "mode": "SSE-S3"})); + assert!(matches!( + validate_bucket_configuration(&bucket), + Err((Reason::BucketEncryptionApplyFailed, _)) + )); + } + #[tokio::test] async fn matching_live_object_lock_configuration_is_adopted_without_a_write() { let live = AdminObjectLockConfiguration { diff --git a/src/sts/rustfs_client.rs b/src/sts/rustfs_client.rs index f07f5cc..a99b1e8 100644 --- a/src/sts/rustfs_client.rs +++ b/src/sts/rustfs_client.rs @@ -21,6 +21,7 @@ use crate::Tenant; use crate::cluster_dns; pub use rustfs_admin::{ + BucketEncryptionAlgorithm, BucketEncryptionConfiguration, BucketLifecycleAbortIncompleteMultipartUpload, BucketLifecycleConfiguration, BucketLifecycleExpiration, BucketLifecycleNoncurrentVersionTransition, BucketLifecycleRule, BucketLifecycleRuleStatus, BucketLifecycleTransition, BucketObjectLockConfiguration, diff --git a/src/types/v1alpha1.rs b/src/types/v1alpha1.rs index 9d1c8df..f659238 100755 --- a/src/types/v1alpha1.rs +++ b/src/types/v1alpha1.rs @@ -506,6 +506,28 @@ mod tenant_provisioning_crd_tests { object_lock["x-kubernetes-validations"][0]["message"], json!("Present object lock configuration requires mode and days; Absent forbids them") ); + let encryption = &spec["properties"]["buckets"]["items"]["properties"]["encryption"]; + assert_eq!( + encryption["properties"]["mode"]["enum"], + json!(["SSE-S3", "SSE-KMS", null]) + ); + assert_eq!( + encryption["properties"]["kmsKeyId"]["pattern"], + json!("^[A-Za-z0-9._:/+=@-]+$") + ); + let encryption_messages: Vec<_> = encryption["x-kubernetes-validations"] + .as_array() + .expect("encryption validations are present") + .iter() + .map(|validation| validation["message"].clone()) + .collect(); + assert_eq!( + encryption_messages, + vec![ + json!("Present bucket encryption requires mode; Absent forbids mode and kmsKeyId"), + json!("kmsKeyId requires mode SSE-KMS"), + ] + ); let bucket_validations = spec["properties"]["buckets"]["items"]["x-kubernetes-validations"] .as_array() .expect("bucket validations are present"); @@ -533,6 +555,10 @@ mod tenant_provisioning_crd_tests { bucket_status["properties"]["objectLockConfigurationLastAppliedHash"]["nullable"], json!(true) ); + assert_eq!( + bucket_status["properties"]["encryptionLastAppliedHash"]["nullable"], + json!(true) + ); } } diff --git a/src/types/v1alpha1/provisioning.rs b/src/types/v1alpha1/provisioning.rs index 3cd702d..f132279 100644 --- a/src/types/v1alpha1/provisioning.rs +++ b/src/types/v1alpha1/provisioning.rs @@ -30,6 +30,7 @@ pub(crate) const MAX_BUCKET_NAME_LENGTH: u32 = 63; pub(crate) const MAX_BUCKET_LIFECYCLE_RULES: u32 = 1_000; pub(crate) const MAX_BUCKET_LIFECYCLE_RULE_ID_LENGTH: u32 = 255; pub(crate) const MAX_BUCKET_OBJECT_LOCK_RETENTION_DAYS: u32 = 36_500; +pub(crate) const MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH: u32 = 1024; pub(crate) const MAX_PROVISIONING_TIER_NAME_LENGTH: u32 = 253; pub(crate) const MAX_TIER_CREDENTIALS_SECRET_NAME_LENGTH: u32 = 253; @@ -266,6 +267,51 @@ impl BucketObjectLockConfiguration { } } +#[derive(Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +#[serde(rename_all = "PascalCase")] +pub enum BucketEncryptionState { + Present, + Absent, +} + +#[derive(Deserialize, Serialize, Clone, Copy, Debug, JsonSchema, ToSchema, PartialEq, Eq)] +pub enum BucketEncryptionMode { + #[serde(rename = "SSE-S3")] + SseS3, + #[serde(rename = "SSE-KMS")] + SseKms, +} + +/// Default server-side encryption applied by RustFS to new objects in the bucket. +#[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[x_kube(validation = Rule::new("(!has(self.state) || self.state == 'Present') ? has(self.mode) : !has(self.mode) && !has(self.kmsKeyId)").message("Present bucket encryption requires mode; Absent forbids mode and kmsKeyId"))] +#[x_kube(validation = Rule::new("!has(self.kmsKeyId) || (has(self.mode) && self.mode == 'SSE-KMS')").message("kmsKeyId requires mode SSE-KMS"))] +pub struct BucketEncryptionSpec { + /// Omission is equivalent to `Present`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub state: Option, + + /// `SSE-S3` stores S3 `AES256`; `SSE-KMS` stores `aws:kms` and requires + /// `spec.encryption.enabled` on the Tenant. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mode: Option, + + /// KMS key for `SSE-KMS`. Omission uses the RustFS default KMS key. + #[schemars( + length(min = 1, max = MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH), + regex(pattern = r"^[A-Za-z0-9._:/+=@-]+$") + )] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub kms_key_id: Option, +} + +impl BucketEncryptionSpec { + pub(crate) fn state(&self) -> BucketEncryptionState { + self.state.unwrap_or(BucketEncryptionState::Present) + } +} + #[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)] #[serde(rename_all = "camelCase")] pub struct BucketLifecycleRuleFilter { @@ -396,6 +442,11 @@ pub struct ProvisioningBucket { #[serde(default, skip_serializing_if = "Option::is_none")] pub policy: Option, + /// Default bucket encryption. Omission leaves the live configuration unmanaged; `Absent` + /// removes only a configuration previously owned by the operator. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub encryption: Option, + /// Declarative S3 bucket lifecycle configuration. Omission leaves the live lifecycle /// configuration unmanaged; `Absent` removes only a configuration previously owned by the /// operator. diff --git a/src/types/v1alpha1/status.rs b/src/types/v1alpha1/status.rs index 0ee061a..8401376 100755 --- a/src/types/v1alpha1/status.rs +++ b/src/types/v1alpha1/status.rs @@ -189,6 +189,8 @@ pub enum Reason { BucketObjectLockConfigurationConflict, BucketPolicyApplyFailed, BucketPolicyConflict, + BucketEncryptionApplyFailed, + BucketEncryptionConflict, BucketLifecycleApplyFailed, BucketLifecycleConflict, TierSecretInvalid, @@ -276,6 +278,8 @@ impl Reason { Self::BucketObjectLockConfigurationConflict => "BucketObjectLockConfigurationConflict", Self::BucketPolicyApplyFailed => "BucketPolicyApplyFailed", Self::BucketPolicyConflict => "BucketPolicyConflict", + Self::BucketEncryptionApplyFailed => "BucketEncryptionApplyFailed", + Self::BucketEncryptionConflict => "BucketEncryptionConflict", Self::BucketLifecycleApplyFailed => "BucketLifecycleApplyFailed", Self::BucketLifecycleConflict => "BucketLifecycleConflict", Self::TierSecretInvalid => "TierSecretInvalid", @@ -566,6 +570,8 @@ pub fn is_blocked_reason(reason: &str) -> bool { | "BucketObjectLockConfigurationConflict" | "BucketPolicyApplyFailed" | "BucketPolicyConflict" + | "BucketEncryptionApplyFailed" + | "BucketEncryptionConflict" | "BucketLifecycleApplyFailed" | "BucketLifecycleConflict" | "TierSecretInvalid" @@ -688,6 +694,10 @@ pub fn next_actions_for_reason(reason: &str) -> Vec<&'static str> { } "BucketPolicyApplyFailed" => vec!["fixBucketPolicy", "inspectOperatorLogs"], "BucketPolicyConflict" => vec!["inspectLiveBucketPolicy", "updateBucketSpec"], + "BucketEncryptionApplyFailed" => vec!["fixBucketEncryption", "inspectOperatorLogs"], + "BucketEncryptionConflict" => { + vec!["inspectLiveBucketEncryption", "updateBucketSpec"] + } "BucketLifecycleApplyFailed" => vec!["fixBucketLifecycle", "inspectOperatorLogs"], "BucketLifecycleConflict" => { vec!["inspectLiveBucketLifecycle", "updateBucketSpec"] @@ -812,6 +822,16 @@ mod tests { next_actions_for_reason("BucketLifecycleConflict"), vec!["inspectLiveBucketLifecycle", "updateBucketSpec"] ); + assert!(is_blocked_reason("BucketEncryptionApplyFailed")); + assert!(is_blocked_reason("BucketEncryptionConflict")); + assert_eq!( + next_actions_for_reason("BucketEncryptionApplyFailed"), + vec!["fixBucketEncryption", "inspectOperatorLogs"] + ); + assert_eq!( + next_actions_for_reason("BucketEncryptionConflict"), + vec!["inspectLiveBucketEncryption", "updateBucketSpec"] + ); for reason in ["TierSecretInvalid", "TierApplyFailed", "TierConflict"] { assert!(is_blocked_reason(reason), "{reason} must block"); } diff --git a/src/types/v1alpha1/status/provisioning.rs b/src/types/v1alpha1/status/provisioning.rs index 661ce5b..070c646 100644 --- a/src/types/v1alpha1/status/provisioning.rs +++ b/src/types/v1alpha1/status/provisioning.rs @@ -153,6 +153,12 @@ pub struct ProvisioningBucketStatus { #[serde(default, skip_serializing_if = "Option::is_none")] pub object_lock_configuration_last_applied_hash: Option, + + #[serde(default, skip_serializing_if = "Option::is_none")] + pub encryption_desired_hash: Option, + + #[serde(default, skip_serializing_if = "Option::is_none")] + pub encryption_last_applied_hash: Option, } impl ProvisioningBucketStatus { @@ -164,6 +170,8 @@ impl ProvisioningBucketStatus { versioning: None, object_lock_configuration_desired_hash: None, object_lock_configuration_last_applied_hash: None, + encryption_desired_hash: None, + encryption_last_applied_hash: None, } } } From b426df0216e63bd524ca347ffd377f0766504f19 Mon Sep 17 00:00:00 2001 From: Shurik <10572093+shurkus@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:58:27 +0200 Subject: [PATCH 2/3] fix(provisioning): validate bucket encryption providers and default KMS key SSE-S3 bucket defaults are applied only when RustFS can encrypt the objects: Tenant KMS is enabled, or spec.env sets RUSTFS_SSE_S3_MASTER_KEY to a base64 32-byte key by value or from a Secret key, which the Operator reads and validates. Otherwise the bucket fails instead of reporting Ready while every upload under that default is rejected. SSE-KMS without kmsKeyId resolves to the Tenant spec.encryption.defaultKeyId, the key RustFS records for such a request, and requires it. Live and desired configurations are compared exactly, so removing an explicit key or changing the Tenant default key re-applies an owned configuration, and an external key change is a BucketEncryptionConflict that keeps the ownership record. --- CHANGELOG.md | 4 +- deploy/rustfs-operator/crds/tenant-crd.yaml | 6 +- docs/operator-user-guide.md | 17 +- docs/operator-user-guide.zh-CN.md | 17 +- src/reconcile/provisioning.rs | 550 +++++++++++++++++--- src/types/v1alpha1/provisioning.rs | 6 +- 6 files changed, 532 insertions(+), 68 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 35ca8a2..6f01f3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,8 +42,8 @@ All notable changes to RustFS Operator are documented in this file. The format i - Tenant `spec.tiers` for declarative S3, RustFS, and MinIO remote tiers with Secret-sourced credentials, rotation on Secret updates, and conflict reporting for immutable tier settings. - Declarative Tenant bucket versioning and Object Lock default retention with drift-aware ownership. -- Declarative default bucket encryption (`SSE-S3` or `SSE-KMS`) with drift-aware ownership and - explicit removal. +- Declarative default bucket encryption (`SSE-S3` or `SSE-KMS`) with drift-aware ownership, + explicit removal, and validation of the Tenant SSE-S3 master key or KMS default key. - Helm chart `cosiDriver.enabled` to deploy the RustFS COSI (`rustfs.objectstorage.k8s.io`) driver alongside the upstream provisioner sidecar, with dedicated ServiceAccount/RBAC. - Seven built-in RustFS Grafana dashboards, published by default as sidecar-discoverable diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml index ee4f55c..8ce920f 100644 --- a/deploy/rustfs-operator/crds/tenant-crd.yaml +++ b/deploy/rustfs-operator/crds/tenant-crd.yaml @@ -1242,7 +1242,7 @@ spec: nullable: true properties: kmsKeyId: - description: KMS key for `SSE-KMS`. Omission uses the RustFS default KMS key. + description: KMS key for `SSE-KMS`. Omission uses the Tenant `spec.encryption.defaultKeyId`. maxLength: 1024 minLength: 1 nullable: true @@ -1250,8 +1250,8 @@ spec: type: string mode: description: |- - `SSE-S3` stores S3 `AES256`; `SSE-KMS` stores `aws:kms` and requires - `spec.encryption.enabled` on the Tenant. + `SSE-S3` stores S3 `AES256` and requires Tenant KMS or `RUSTFS_SSE_S3_MASTER_KEY` in + `spec.env`; `SSE-KMS` stores `aws:kms` and requires `spec.encryption.enabled` on the Tenant. enum: - SSE-S3 - SSE-KMS diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md index 78010dd..62ed2c0 100644 --- a/docs/operator-user-guide.md +++ b/docs/operator-user-guide.md @@ -962,7 +962,7 @@ The operator can create RustFS policies, users, remote tiers, and buckets after Bucket versioning and Object Lock are declarative but conservative. `versioning: true` enables versioning. `versioning: false` keeps a never-versioned bucket unchanged or requests S3 `Suspended` after versioning has been enabled; omitting the field leaves versioning unmanaged unless Object Lock requires it. `objectLock: true` enables Object Lock after versioning is enabled. Object Lock cannot be disabled, and the Operator refuses `objectLock: false` or versioning suspension when the live bucket has Object Lock enabled. `objectLockConfiguration` manages the default retention rule: omit `state` for `Present`, specify `mode: Governance` or `Compliance` with `days`, or use `state: Absent` without `mode` and `days` to remove only the default rule. Omitting the entire field leaves a live default-retention rule unmanaged. The Operator adopts an identical live rule, but does not replace or remove a different rule unless its hash matches the last configuration recorded in status. Default retention applies to new object versions; it does not retroactively lock existing versions. -Bucket default encryption is declared with `encryption`. `mode: SSE-S3` stores S3 `AES256`; `mode: SSE-KMS` stores `aws:kms` with an optional `kmsKeyId` (RustFS uses its default KMS key when omitted) and requires `spec.encryption.enabled` on the Tenant. Default encryption applies to objects written after it is set; it does not re-encrypt existing objects. `state: Absent` without `mode` and `kmsKeyId` removes the configuration. The Operator adopts a matching live configuration, but never replaces or removes a different one unless its hash matches the configuration last recorded in `status.provisioning.buckets`. Omitting `encryption` performs no encryption API calls and leaves the live configuration untouched. +Bucket default encryption is declared with `encryption`. `mode: SSE-S3` stores S3 `AES256` and requires an SSE-S3 key for RustFS: either `spec.encryption.enabled` on the Tenant, or `RUSTFS_SSE_S3_MASTER_KEY` in `spec.env`, a base64-encoded 32-byte key set by value or from a Secret key, which the Operator validates before applying the bucket default. `mode: SSE-KMS` stores `aws:kms`, requires `spec.encryption.enabled`, and uses `kmsKeyId` or, when it is omitted, the Tenant `spec.encryption.defaultKeyId`; changing either key re-applies an owned configuration. Default encryption applies to objects written after it is set; it does not re-encrypt existing objects. `state: Absent` without `mode` and `kmsKeyId` removes the configuration. The Operator adopts a matching live configuration, but never replaces or removes a different one unless its hash matches the configuration last recorded in `status.provisioning.buckets`. Omitting `encryption` performs no encryption API calls and leaves the live configuration untouched. Each bucket may also declare `lifecycle.state: Present` with one or more S3 lifecycle rules. The current schema supports expiration after a positive number of days, aborting incomplete multipart uploads after a positive number of days, `transition` of current versions after zero or more days, and `noncurrentVersionTransition` of noncurrent versions after zero or more days; every rule has a unique ID, an `Enabled` or `Disabled` status, and an explicit prefix filter (use `prefix: ""` for all objects). `lifecycle.state: Absent` requests deletion and cannot include rules. The Operator adopts an identical live configuration, but never replaces or deletes a different configuration without a matching ownership hash in `status.provisioning.buckets`. Omitting `lifecycle` performs no lifecycle API calls and leaves the live configuration untouched. @@ -1051,6 +1051,15 @@ stringData: accesskey: appuser01 secretkey: appuser01secret --- +apiVersion: v1 +kind: Secret +metadata: + name: rustfs-sse-s3 + namespace: storage +stringData: + # openssl rand -base64 32 + master-key: REPLACE_WITH_BASE64_32_BYTE_KEY +--- apiVersion: rustfs.com/v1alpha1 kind: Tenant metadata: @@ -1064,6 +1073,12 @@ spec: servers: 1 persistence: volumesPerServer: 4 + env: + - name: RUSTFS_SSE_S3_MASTER_KEY + valueFrom: + secretKeyRef: + name: rustfs-sse-s3 + key: master-key policies: - name: app-readwrite document: diff --git a/docs/operator-user-guide.zh-CN.md b/docs/operator-user-guide.zh-CN.md index 963db5c..5add73f 100644 --- a/docs/operator-user-guide.zh-CN.md +++ b/docs/operator-user-guide.zh-CN.md @@ -906,7 +906,7 @@ Operator 可以在 Tenant workload Ready 后自动创建 RustFS policy、user、 Bucket 版本控制和 Object Lock 采用声明式但保守的调和策略。`versioning: true` 启用版本控制。`versioning: false` 会让从未启用版本控制的 bucket 保持原状，或在版本控制已启用后请求 S3 `Suspended`；省略该字段时保持现状，除非 Object Lock 要求先启用版本控制。`objectLock: true` 会在版本控制启用后开启 Object Lock。Object Lock 不可关闭；当线上 bucket 已启用 Object Lock 时，Operator 会拒绝 `objectLock: false` 或暂停版本控制。`objectLockConfiguration` 管理默认保留规则：省略 `state` 等同于 `Present`，此时必须设置 `mode: Governance` 或 `Compliance` 以及 `days`；`state: Absent` 必须省略 `mode` 和 `days`，仅移除默认规则。完全省略 `objectLockConfiguration` 时，不管理已有默认保留规则。Operator 可以接管与声明完全相同的线上规则，但只有线上配置哈希仍与 status 中最后一次应用的配置一致时，才会替换或移除不同规则。默认保留策略只作用于之后创建的对象版本，不会追溯锁定已有版本。 -Bucket 默认加密通过 `encryption` 声明。`mode: SSE-S3` 写入 S3 `AES256`；`mode: SSE-KMS` 写入 `aws:kms`，可选 `kmsKeyId`（省略时 RustFS 使用默认 KMS key），并要求 Tenant 设置 `spec.encryption.enabled`。默认加密只作用于设置之后写入的对象，不会重新加密已有对象。`state: Absent`（不能包含 `mode` 和 `kmsKeyId`）表示移除该配置。Operator 可以接管与声明一致的线上配置，但只有线上配置哈希与 `status.provisioning.buckets` 中最后记录的配置一致时，才会替换或移除不同的配置。省略 `encryption` 时不会调用加密 API，也不会改动现有配置。 +Bucket 默认加密通过 `encryption` 声明。`mode: SSE-S3` 写入 S3 `AES256`，并要求 RustFS 具备 SSE-S3 密钥：Tenant 设置 `spec.encryption.enabled`，或在 `spec.env` 中提供 `RUSTFS_SSE_S3_MASTER_KEY`（base64 编码的 32 字节密钥，直接设置或来自 Secret key），Operator 会在应用 bucket 默认加密前校验该密钥。`mode: SSE-KMS` 写入 `aws:kms`，要求设置 `spec.encryption.enabled`，并使用 `kmsKeyId`；省略时使用 Tenant 的 `spec.encryption.defaultKeyId`。任一 key 变更时，Operator 会重新应用其拥有的配置。默认加密只作用于设置之后写入的对象，不会重新加密已有对象。`state: Absent`（不能包含 `mode` 和 `kmsKeyId`）表示移除该配置。Operator 可以接管与声明一致的线上配置，但只有线上配置哈希与 `status.provisioning.buckets` 中最后记录的配置一致时，才会替换或移除不同的配置。省略 `encryption` 时不会调用加密 API，也不会改动现有配置。 每个 bucket 还可以通过 `lifecycle.state: Present` 声明一个或多个 S3 生命周期规则。当前 schema 支持按正整数天数过期对象、按正整数天数清理未完成的分片上传、通过 `transition` 在零天或更多天后迁移当前版本，以及通过 `noncurrentVersionTransition` 在零天或更多天后迁移非当前版本；每条规则必须有唯一 ID、`Enabled` 或 `Disabled` 状态，并显式指定 prefix filter（`prefix: ""` 表示全部对象）。`lifecycle.state: Absent` 表示请求删除，且不能包含规则。Operator 可以接管与声明完全相同的现有配置，但没有 `status.provisioning.buckets` 中匹配的所有权哈希时，不会覆盖或删除不同的现有配置。省略 `lifecycle` 时不会调用 Lifecycle API，也不会改动现有配置。 @@ -995,6 +995,15 @@ stringData: accesskey: appuser01 secretkey: appuser01secret --- +apiVersion: v1 +kind: Secret +metadata: + name: rustfs-sse-s3 + namespace: storage +stringData: + # openssl rand -base64 32 + master-key: REPLACE_WITH_BASE64_32_BYTE_KEY +--- apiVersion: rustfs.com/v1alpha1 kind: Tenant metadata: @@ -1008,6 +1017,12 @@ spec: servers: 1 persistence: volumesPerServer: 4 + env: + - name: RUSTFS_SSE_S3_MASTER_KEY + valueFrom: + secretKeyRef: + name: rustfs-sse-s3 + key: master-key policies: - name: app-readwrite document: diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs index 6a43479..c033a5d 100644 --- a/src/reconcile/provisioning.rs +++ b/src/reconcile/provisioning.rs @@ -46,8 +46,9 @@ use crate::types::v1alpha1::status::provisioning::{ ProvisioningUserOwnershipStatus, ProvisioningUserStatus, }; use crate::types::v1alpha1::tenant::Tenant; +use base64::Engine as _; use k8s_openapi::ByteString; -use k8s_openapi::api::core::v1::{ConfigMap, Secret}; +use k8s_openapi::api::core::v1::{ConfigMap, EnvVarSource, Secret}; use kube::Api; use kube::api::{Patch, PatchParams}; use serde_json::Value; @@ -58,6 +59,7 @@ use tracing::{info, warn}; const CHECKPOINT_CONFLICT_RETRY: Duration = Duration::from_secs(2); const CHECKPOINT_TRANSIENT_RETRY: Duration = Duration::from_secs(10); +const SSE_S3_MASTER_KEY_ENV: &str = "RUSTFS_SSE_S3_MASTER_KEY"; pub(super) struct ProvisioningReconcileResult { pub status: ProvisioningStatus, @@ -2468,7 +2470,7 @@ async fn reconcile_bucket( status.encryption_desired_hash = bucket .encryption .as_ref() - .and_then(bucket_encryption_desired_hash); + .and_then(|spec| bucket_encryption_desired_hash(run.tenant, spec)); if status.state != ProvisioningItemState::Ready.as_str() { status.lifecycle_desired_hash = bucket.lifecycle.as_ref().and_then(lifecycle_desired_hash); return status; @@ -3126,7 +3128,9 @@ async fn reconcile_bucket_encryption( let Some(spec) = bucket.encryption.as_ref() else { return status; }; - let desired = match bucket_encryption_configuration(spec) { + let desired = match bucket_encryption_configuration(spec) + .and_then(|desired| resolve_bucket_encryption(run.tenant, desired)) + { Ok(desired) => desired, Err(message) => { status.item = replace_bucket_item_state( @@ -3142,22 +3146,12 @@ async fn reconcile_bucket_encryption( }; if desired .as_ref() - .is_some_and(|desired| desired.algorithm == AdminEncryptionAlgorithm::AwsKms) - && !run - .tenant - .spec - .encryption - .as_ref() - .is_some_and(|encryption| encryption.enabled) + .is_some_and(|desired| desired.algorithm == AdminEncryptionAlgorithm::Aes256) + && let Err(error) = check_sse_s3_provider(run).await { - status.item = replace_bucket_item_state( - run, - previous, - &status.item, - ProvisioningItemState::Failed, - Reason::BucketEncryptionApplyFailed, - "SSE-KMS bucket encryption requires spec.encryption.enabled on the Tenant", - ); + let mut item = run.item_from_spec_error(previous, &status.item.name, error); + copy_bucket_configuration_status(&mut item, &status.item); + status.item = item; return status; } @@ -3220,9 +3214,7 @@ async fn reconcile_bucket_encryption( return status; }; - if let Some(live) = live.as_ref() - && bucket_encryption_matches(&desired, live) - { + if live.as_ref() == Some(&desired) { status.encryption_last_applied_hash = live_hash; status.item.message = Some(append_bucket_message( &status.item, @@ -3256,7 +3248,7 @@ async fn reconcile_bucket_encryption( return status; } let verified = match client.get_bucket_encryption(&bucket.name).await { - Ok(Some(verified)) if bucket_encryption_matches(&desired, &verified) => verified, + Ok(Some(verified)) if verified == desired => verified, Ok(_) => { status.item = replace_bucket_item_state( run, @@ -3280,7 +3272,6 @@ async fn reconcile_bucket_encryption( return status; } }; - // Ownership is the verified live document: RustFS fills in the default KMS key for SSE-KMS. status.encryption_last_applied_hash = Some(hash_document(&verified.to_xml())); status.item.message = Some(append_bucket_message( &status.item, @@ -3347,29 +3338,153 @@ fn is_valid_kms_key_id(key_id: &str) -> bool { .all(|byte| byte.is_ascii_alphanumeric() || b"._:/+=@-".contains(&byte)) } -fn bucket_encryption_desired_hash(spec: &BucketEncryptionSpec) -> Option { +/// SSE-KMS requires Tenant KMS. Without `kmsKeyId` it names the Tenant's `defaultKeyId`, the key +/// RustFS records for such a request, so a changed default key is a spec change. +fn resolve_bucket_encryption( + tenant: &Tenant, + desired: Option, +) -> Result, String> { + let Some(mut desired) = desired else { + return Ok(None); + }; + if desired.algorithm != AdminEncryptionAlgorithm::AwsKms { + return Ok(Some(desired)); + } + let Some(encryption) = tenant + .spec + .encryption + .as_ref() + .filter(|encryption| encryption.enabled) + else { + return Err( + "SSE-KMS bucket encryption requires spec.encryption.enabled on the Tenant".to_string(), + ); + }; + if desired.kms_master_key_id.is_none() { + let Some(default_key_id) = encryption + .default_key_id + .as_deref() + .filter(|key_id| !key_id.is_empty()) + else { + return Err( + "SSE-KMS bucket encryption without kmsKeyId requires spec.encryption.defaultKeyId on the Tenant" + .to_string(), + ); + }; + if !is_valid_kms_key_id(default_key_id) { + return Err(format!( + "Tenant spec.encryption.defaultKeyId must contain between 1 and {MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH} characters from [A-Za-z0-9._:/+=@-] to be used as a bucket default" + )); + } + desired.kms_master_key_id = Some(default_key_id.to_string()); + } + Ok(Some(desired)) +} + +/// Without Tenant KMS, RustFS encrypts SSE-S3 objects with `RUSTFS_SSE_S3_MASTER_KEY` and +/// rejects every upload under an SSE-S3 bucket default when that key is missing or invalid. +async fn check_sse_s3_provider(run: &ProvisioningRun<'_>) -> Result<(), SpecLoadError> { + if run + .tenant + .spec + .encryption + .as_ref() + .is_some_and(|encryption| encryption.enabled) + { + return Ok(()); + } + let invalid = |message: String| SpecLoadError { + reason: Reason::BucketEncryptionApplyFailed, + message, + transient: false, + }; + let Some(env) = run + .tenant + .spec + .env + .iter() + .find(|env| env.name == SSE_S3_MASTER_KEY_ENV) + else { + return Err(invalid(format!( + "SSE-S3 bucket encryption requires spec.encryption.enabled or {SSE_S3_MASTER_KEY_ENV} in spec.env on the Tenant" + ))); + }; + let value = match (env.value.as_deref(), env.value_from.as_ref()) { + (Some(value), None) => value.to_string(), + ( + None, + Some(EnvVarSource { + secret_key_ref: Some(selector), + .. + }), + ) => read_sse_s3_master_key_secret(run, &selector.name, &selector.key).await?, + _ => { + return Err(invalid(format!( + "{SSE_S3_MASTER_KEY_ENV} in spec.env must be set by value or from a Secret key" + ))); + } + }; + match base64::engine::general_purpose::STANDARD.decode(value.trim()) { + Ok(key) if key.len() == 32 => Ok(()), + _ => Err(invalid(format!( + "{SSE_S3_MASTER_KEY_ENV} must be a base64-encoded 32-byte key for SSE-S3 bucket encryption" + ))), + } +} + +async fn read_sse_s3_master_key_secret( + run: &ProvisioningRun<'_>, + secret_name: &str, + key: &str, +) -> Result { + let invalid = |message: String| SpecLoadError { + reason: Reason::BucketEncryptionApplyFailed, + message, + transient: false, + }; + let secret: Secret = run + .ctx + .get(secret_name, run.namespace) + .await + .map_err(|error| { + if context::is_kube_not_found(&error) { + invalid(format!( + "{SSE_S3_MASTER_KEY_ENV} Secret '{secret_name}' was not found" + )) + } else { + SpecLoadError { + reason: Reason::BucketEncryptionApplyFailed, + message: format!( + "failed to read {SSE_S3_MASTER_KEY_ENV} Secret '{secret_name}': {error}" + ), + transient: context::is_transient_kube_error(&error), + } + } + })?; + let raw = secret + .data + .as_ref() + .and_then(|data| data.get(key)) + .ok_or_else(|| { + invalid(format!( + "{SSE_S3_MASTER_KEY_ENV} Secret '{secret_name}' has no key '{key}'" + )) + })?; + String::from_utf8(raw.0.clone()).map_err(|_| { + invalid(format!( + "{SSE_S3_MASTER_KEY_ENV} Secret '{secret_name}' key '{key}' must be valid UTF-8" + )) + }) +} + +fn bucket_encryption_desired_hash(tenant: &Tenant, spec: &BucketEncryptionSpec) -> Option { bucket_encryption_configuration(spec) + .and_then(|desired| resolve_bucket_encryption(tenant, desired)) .ok() .flatten() .map(|configuration| hash_document(&configuration.to_xml())) } -/// SSE-KMS without `kmsKeyId` accepts whatever default key RustFS recorded. -fn bucket_encryption_matches( - desired: &AdminEncryptionConfiguration, - live: &AdminEncryptionConfiguration, -) -> bool { - desired.algorithm == live.algorithm - && desired.bucket_key_enabled == live.bucket_key_enabled - && match desired.kms_master_key_id.as_deref() { - Some(key_id) => live.kms_master_key_id.as_deref() == Some(key_id), - None => { - desired.algorithm == AdminEncryptionAlgorithm::AwsKms - || live.kms_master_key_id.is_none() - } - } -} - async fn reconcile_bucket_lifecycle( run: &mut ProvisioningRun<'_>, client: &RustfsAdminClient, @@ -4505,6 +4620,7 @@ mod tests { }; use http_body_util::BodyExt; use k8s_openapi::ByteString; + use k8s_openapi::api::core::v1::{EnvVar, SecretKeySelector}; use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta; use kube::{Client, client::Body as KubeBody}; use std::convert::Infallible; @@ -7038,6 +7154,94 @@ mod tests { .expect("encryption bucket should deserialize") } + // base64 of the 32 bytes "0123456789abcdef0123456789abcdef". + const SSE_S3_TEST_MASTER_KEY: &str = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY="; + + fn sse_s3_master_key_env(value: Option<&str>, secret_key: Option<&str>) -> EnvVar { + EnvVar { + name: "RUSTFS_SSE_S3_MASTER_KEY".to_string(), + value: value.map(str::to_string), + value_from: secret_key.map(|key| EnvVarSource { + secret_key_ref: Some(SecretKeySelector { + name: "sse-s3".to_string(), + key: key.to_string(), + optional: Some(false), + }), + ..Default::default() + }), + } + } + + fn sse_s3_tenant() -> Tenant { + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let mut tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + tenant + .spec + .env + .push(sse_s3_master_key_env(Some(SSE_S3_TEST_MASTER_KEY), None)); + tenant + } + + fn kms_tenant(default_key_id: Option<&str>) -> Tenant { + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let mut tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + tenant.spec.encryption = Some(crate::types::v1alpha1::encryption::EncryptionConfig { + enabled: true, + default_key_id: default_key_id.map(str::to_string), + ..Default::default() + }); + tenant + } + + fn sse_kms(key_id: &str) -> AdminEncryptionConfiguration { + AdminEncryptionConfiguration { + algorithm: AdminEncryptionAlgorithm::AwsKms, + kms_master_key_id: Some(key_id.to_string()), + bucket_key_enabled: false, + } + } + + fn sse_s3_secret_kube_context(key: &'static str, value: &'static str) -> Context { + let kube_service = service_fn(move |request: http::Request| async move { + assert_eq!(request.method(), http::Method::GET); + assert_eq!( + request.uri().path(), + "/api/v1/namespaces/storage/secrets/sse-s3" + ); + let secret = Secret { + metadata: ObjectMeta { + name: Some("sse-s3".to_string()), + namespace: Some("storage".to_string()), + ..Default::default() + }, + data: Some(BTreeMap::from([( + key.to_string(), + ByteString(value.as_bytes().to_vec()), + )])), + ..Default::default() + }; + Ok::<_, Infallible>( + http::Response::builder() + .header("content-type", "application/json") + .body(KubeBody::from( + serde_json::to_vec(&secret).expect("Secret should serialize"), + )) + .expect("response should build"), + ) + }); + Context::new(Client::new(kube_service, "default")) + } + + async fn encryption_requests(capture: &BucketControlCapture) -> usize { + capture + .requests + .lock() + .await + .iter() + .filter(|request| request.ends_with("encryption=")) + .count() + } + fn previous_encryption_owner(hash: String) -> ProvisioningBucketStatus { let mut previous = ProvisioningBucketStatus::new(ProvisioningItemStatus::new( "secure-bucket", @@ -7063,8 +7267,7 @@ mod tests { let capture = BucketControlCapture::default(); let (client, server) = bucket_control_test_client(capture.clone()).await; let ctx = empty_kube_context(); - let user = provisioning_user("app-user", "app-user-secret", "readwrite"); - let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let tenant = sse_s3_tenant(); let mut run = empty_run(&ctx, &tenant); let item = reconcile_bucket( @@ -7108,8 +7311,7 @@ mod tests { }; let (client, server) = bucket_control_test_client(capture.clone()).await; let ctx = empty_kube_context(); - let user = provisioning_user("app-user", "app-user-secret", "readwrite"); - let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let tenant = sse_s3_tenant(); let mut run = empty_run(&ctx, &tenant); let item = reconcile_bucket( @@ -7177,8 +7379,7 @@ mod tests { }; let (client, server) = bucket_control_test_client(capture.clone()).await; let ctx = empty_kube_context(); - let user = provisioning_user("app-user", "app-user-secret", "readwrite"); - let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let tenant = sse_s3_tenant(); let mut run = empty_run(&ctx, &tenant); let item = reconcile_bucket(&mut run, &client, &encryption_bucket(spec.clone())).await; @@ -7209,12 +7410,7 @@ mod tests { }; let (client, server) = bucket_control_test_client(capture.clone()).await; let ctx = empty_kube_context(); - let user = provisioning_user("app-user", "app-user-secret", "readwrite"); - let mut tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); - tenant.spec.encryption = Some(crate::types::v1alpha1::encryption::EncryptionConfig { - enabled: true, - ..Default::default() - }); + let tenant = kms_tenant(Some("rustfs-default-key")); let mut run = empty_run(&ctx, &tenant); run.previous .buckets @@ -7241,13 +7437,13 @@ mod tests { item.encryption_last_applied_hash, Some(hash_document(&applied.to_xml())) ); - assert_ne!( + assert_eq!( item.encryption_desired_hash, item.encryption_last_applied_hash ); drop(run); - // The RustFS-chosen default key is adopted on the next pass without another write. + // The default key is not written again on the next pass. let mut run = empty_run(&ctx, &tenant); run.previous.buckets.push(item.clone()); let again = reconcile_bucket( @@ -7304,8 +7500,7 @@ mod tests { }; let (client, server) = bucket_control_test_client(capture.clone()).await; let ctx = empty_kube_context(); - let user = provisioning_user("app-user", "app-user-secret", "readwrite"); - let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let tenant = sse_s3_tenant(); let mut run = empty_run(&ctx, &tenant); run.previous .buckets @@ -7374,8 +7569,7 @@ mod tests { }; let (client, server) = bucket_control_test_client(capture.clone()).await; let ctx = empty_kube_context(); - let user = provisioning_user("app-user", "app-user-secret", "readwrite"); - let tenant = provisioning_test_tenant(user, ProvisioningStatus::default()); + let tenant = sse_s3_tenant(); let mut run = empty_run(&ctx, &tenant); let item = reconcile_bucket( @@ -7391,6 +7585,246 @@ mod tests { server.abort(); } + #[tokio::test] + async fn sse_s3_bucket_encryption_requires_an_encryption_provider() { + let user = provisioning_user("app-user", "app-user-secret", "readwrite"); + let without_provider = provisioning_test_tenant(user, ProvisioningStatus::default()); + let mut invalid_key = sse_s3_tenant(); + invalid_key.spec.env = vec![sse_s3_master_key_env(Some("c2hvcnQ="), None)]; + for (tenant, expected) in [ + ( + without_provider, + "requires spec.encryption.enabled or RUSTFS_SSE_S3_MASTER_KEY", + ), + (invalid_key, "must be a base64-encoded 32-byte key"), + ] { + let capture = BucketControlCapture { + existing: true, + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionApplyFailed.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains(expected)), + "{:?}", + item.message + ); + assert_eq!(encryption_requests(&capture).await, 0); + server.abort(); + } + } + + #[tokio::test] + async fn sse_s3_bucket_encryption_accepts_tenant_kms_or_a_secret_master_key() { + let mut from_secret = sse_s3_tenant(); + from_secret.spec.env = vec![sse_s3_master_key_env(None, Some("master-key"))]; + for (tenant, ctx) in [ + (kms_tenant(None), empty_kube_context()), + ( + from_secret, + sse_s3_secret_kube_context("master-key", SSE_S3_TEST_MASTER_KEY), + ), + ] { + let capture = BucketControlCapture { + existing: true, + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!(capture.encryption.lock().await.clone(), Some(sse_s3())); + server.abort(); + } + } + + #[tokio::test] + async fn sse_s3_master_key_secret_without_the_key_fails_before_any_call() { + let capture = BucketControlCapture { + existing: true, + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = sse_s3_secret_kube_context("other-key", SSE_S3_TEST_MASTER_KEY); + let mut tenant = sse_s3_tenant(); + tenant.spec.env = vec![sse_s3_master_key_env(None, Some("master-key"))]; + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-S3"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("has no key 'master-key'")) + ); + assert_eq!(encryption_requests(&capture).await, 0); + server.abort(); + } + + #[tokio::test] + async fn removing_an_explicit_kms_key_applies_the_tenant_default_key() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_kms("old-explicit-key")))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let tenant = kms_tenant(Some("rustfs-default-key")); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(hash_document( + &sse_kms("old-explicit-key").to_xml(), + ))); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!( + capture.encryption.lock().await.clone(), + Some(sse_kms("rustfs-default-key")) + ); + let put_requests = capture.put_requests.lock().await; + let put = put_requests + .iter() + .find(|request| request.query == "encryption=") + .expect("encryption PUT is captured"); + assert_eq!(put.body, sse_kms("rustfs-default-key").to_xml()); + assert_eq!( + item.encryption_last_applied_hash, + Some(hash_document(&sse_kms("rustfs-default-key").to_xml())) + ); + server.abort(); + } + + #[tokio::test] + async fn tenant_default_kms_key_change_is_applied_to_an_owned_bucket() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_kms("old-default-key")))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let tenant = kms_tenant(Some("new-default-key")); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(hash_document( + &sse_kms("old-default-key").to_xml(), + ))); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Ready.as_str()); + assert_eq!( + capture.encryption.lock().await.clone(), + Some(sse_kms("new-default-key")) + ); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 1); + server.abort(); + } + + #[tokio::test] + async fn external_kms_key_change_is_a_conflict_that_keeps_the_ownership_record() { + let capture = BucketControlCapture { + existing: true, + encryption: Arc::new(Mutex::new(Some(sse_kms("foreign-key")))), + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let tenant = kms_tenant(Some("rustfs-default-key")); + let owned = hash_document(&sse_kms("rustfs-default-key").to_xml()); + let mut run = empty_run(&ctx, &tenant); + run.previous + .buckets + .push(previous_encryption_owner(owned.clone())); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionConflict.as_str()); + assert_eq!(item.encryption_last_applied_hash, Some(owned)); + assert_eq!(requests_for(&capture, "PUT encryption=").await, 0); + assert_eq!( + capture.encryption.lock().await.clone(), + Some(sse_kms("foreign-key")) + ); + server.abort(); + } + + #[tokio::test] + async fn sse_kms_without_a_key_requires_the_tenant_default_key() { + let capture = BucketControlCapture { + existing: true, + ..Default::default() + }; + let (client, server) = bucket_control_test_client(capture.clone()).await; + let ctx = empty_kube_context(); + let tenant = kms_tenant(None); + let mut run = empty_run(&ctx, &tenant); + + let item = reconcile_bucket( + &mut run, + &client, + &encryption_bucket(serde_json::json!({"mode": "SSE-KMS"})), + ) + .await; + + assert_eq!(item.state, ProvisioningItemState::Failed.as_str()); + assert_eq!(item.reason, Reason::BucketEncryptionApplyFailed.as_str()); + assert!( + item.message + .as_deref() + .is_some_and(|message| message.contains("spec.encryption.defaultKeyId")) + ); + assert_eq!(encryption_requests(&capture).await, 0); + server.abort(); + } + #[test] fn bucket_encryption_spec_is_validated_before_any_call() { let spec = |value: serde_json::Value| -> BucketEncryptionSpec { diff --git a/src/types/v1alpha1/provisioning.rs b/src/types/v1alpha1/provisioning.rs index f132279..ac014e2 100644 --- a/src/types/v1alpha1/provisioning.rs +++ b/src/types/v1alpha1/provisioning.rs @@ -292,12 +292,12 @@ pub struct BucketEncryptionSpec { #[serde(default, skip_serializing_if = "Option::is_none")] pub state: Option, - /// `SSE-S3` stores S3 `AES256`; `SSE-KMS` stores `aws:kms` and requires - /// `spec.encryption.enabled` on the Tenant. + /// `SSE-S3` stores S3 `AES256` and requires Tenant KMS or `RUSTFS_SSE_S3_MASTER_KEY` in + /// `spec.env`; `SSE-KMS` stores `aws:kms` and requires `spec.encryption.enabled` on the Tenant. #[serde(default, skip_serializing_if = "Option::is_none")] pub mode: Option, - /// KMS key for `SSE-KMS`. Omission uses the RustFS default KMS key. + /// KMS key for `SSE-KMS`. Omission uses the Tenant `spec.encryption.defaultKeyId`. #[schemars( length(min = 1, max = MAX_BUCKET_ENCRYPTION_KMS_KEY_ID_LENGTH), regex(pattern = r"^[A-Za-z0-9._:/+=@-]+$") From 06ab0a89eda8665ed6727d776af5b45ce807c4b3 Mon Sep 17 00:00:00 2001 From: Shurik <10572093+shurkus@users.noreply.github.com> Date: Mon, 26 Oct 2026 17:41:30 +0200 Subject: [PATCH 3/3] test(e2e): verify SSE-S3 bucket default encryption on a live Tenant --- e2e/src/cases/mod.rs | 2 +- e2e/src/cases/operator.rs | 8 ++ e2e/tests/operator.rs | 224 +++++++++++++++++++++++++++++++++++++- 3 files changed, 232 insertions(+), 2 deletions(-) diff --git a/e2e/src/cases/mod.rs b/e2e/src/cases/mod.rs index 19e55c8..99d012a 100644 --- a/e2e/src/cases/mod.rs +++ b/e2e/src/cases/mod.rs @@ -145,7 +145,7 @@ mod tests { }); assert_eq!(counts.get(&Suite::Smoke).copied().unwrap_or_default(), 3); - assert_eq!(counts.get(&Suite::Operator).copied().unwrap_or_default(), 2); + assert_eq!(counts.get(&Suite::Operator).copied().unwrap_or_default(), 3); assert_eq!(counts.get(&Suite::Sts).copied().unwrap_or_default(), 2); assert_eq!(counts.get(&Suite::Console).copied().unwrap_or_default(), 1); assert_eq!( diff --git a/e2e/src/cases/operator.rs b/e2e/src/cases/operator.rs index 1fc097d..fbd4e2a 100644 --- a/e2e/src/cases/operator.rs +++ b/e2e/src/cases/operator.rs @@ -30,6 +30,13 @@ pub fn cases() -> Vec { "operator/status", "operator", ), + CaseSpec::new( + Suite::Operator, + "operator_live_sse_s3_bucket_encrypts_new_objects", + "Provision an SSE-S3 bucket with the master key from a Secret and verify new objects are stored AES256-encrypted and read back intact.", + "operator/provisioning", + "operator", + ), ] } @@ -49,6 +56,7 @@ mod tests { vec![ "operator_live_tenant_is_ready_and_observed", "operator_live_status_subresource_enforces_cas_and_pruning", + "operator_live_sse_s3_bucket_encrypts_new_objects", ] ); } diff --git a/e2e/tests/operator.rs b/e2e/tests/operator.rs index 266ffb9..8796751 100644 --- a/e2e/tests/operator.rs +++ b/e2e/tests/operator.rs @@ -12,18 +12,29 @@ // See the License for the specific language governing permissions and // limitations under the License. -use anyhow::{Result, ensure}; +use anyhow::{Context, Result, ensure}; +use aws_config::BehaviorVersion; +use aws_credential_types::Credentials; +use aws_sdk_s3::{Client, config::Region, primitives::ByteStream, types::ServerSideEncryption}; use kube::Api; use rustfs_operator_e2e::framework::{ assertions, config::E2eConfig, kube_client, kubectl::Kubectl, live, + port_forward::PortForwardSpec, resources, wait, }; use serde_json::{Value, json}; +use std::time::Duration; +use uuid::Uuid; use operator::types::v1alpha1::tenant::Tenant; const CHECKPOINT_TEST_CRD: &str = "ownershipcheckpointtests.e2e.rustfs.com"; const CHECKPOINT_TEST_RESOURCE: &str = "ownershipcheckpointtests.e2e.rustfs.com"; const CHECKPOINT_TEST_NAME: &str = "ownership-checkpoint-contract"; +const SSE_S3_MASTER_KEY_ENV: &str = "RUSTFS_SSE_S3_MASTER_KEY"; +const SSE_S3_MASTER_KEY_SECRET_KEY: &str = "master-key"; +// base64 of the 32 ASCII bytes "0123456789abcdef0123456789abcdef". +const SSE_S3_TEST_MASTER_KEY: &str = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY="; +const SSE_S3_PROBE_OBJECT: &str = "sse-s3-probe.bin"; const CHECKPOINT_TEST_CRD_YAML: &str = r#" apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition @@ -123,6 +134,217 @@ async fn operator_live_status_subresource_enforces_cas_and_pruning() -> Result<( Ok(()) } +#[tokio::test] +#[ignore = "requires a dedicated live cluster; run through `make e2e-live-run`"] +async fn operator_live_sse_s3_bucket_encrypts_new_objects() -> Result<()> { + let config = E2eConfig::from_env(); + live::require_live_enabled(&config)?; + live::ensure_dedicated_context(&config)?; + + let kubectl = Kubectl::new(&config); + let secret_name = format!("{}-sse-s3", config.tenant_name); + let bucket = format!("e2e-sse-s3-{}", &Uuid::new_v4().simple().to_string()[..8]); + + let result = verify_sse_s3_bucket_encryption(&config, &kubectl, &secret_name, &bucket).await; + let restore_result = restore_smoke_tenant(&config, &kubectl, &secret_name).await; + + result?; + restore_result?; + Ok(()) +} + +async fn verify_sse_s3_bucket_encryption( + config: &E2eConfig, + kubectl: &Kubectl, + secret_name: &str, + bucket: &str, +) -> Result<()> { + kubectl + .apply_yaml_command(sse_s3_secret_manifest(config, secret_name)) + .run_checked()?; + kubectl + .apply_yaml_command(sse_s3_tenant_manifest(config, secret_name, bucket)?) + .run_checked()?; + + let client = kube_client::default_client().await?; + let tenants: Api = kube_client::tenant_api(client, &config.test_namespace); + wait::wait_for_tenant_ready(tenants.clone(), &config.tenant_name, config.timeout).await?; + wait_for_bucket_encryption_applied(&tenants, &config.tenant_name, bucket, config.timeout) + .await?; + + let spec = + PortForwardSpec::tenant_io_on_available_port(&config.test_namespace, &config.tenant_name)?; + let mut port_forward = spec + .start_with_temp_log(kubectl) + .context("start RustFS tenant IO port-forward")?; + let s3 = s3_client(&spec.local_base_url()).await; + + let payload = format!("sse-s3 probe {bucket}\n").repeat(4096).into_bytes(); + wait::wait_until( + "S3 PutObject through the tenant IO port-forward", + Duration::from_secs(60), + Duration::from_secs(2), + || { + let s3 = s3.clone(); + let payload = payload.clone(); + async move { + Ok(s3 + .put_object() + .bucket(bucket) + .key(SSE_S3_PROBE_OBJECT) + .body(ByteStream::from(payload)) + .send() + .await + .ok()) + } + }, + ) + .await?; + port_forward.ensure_running()?; + + let head = s3 + .head_object() + .bucket(bucket) + .key(SSE_S3_PROBE_OBJECT) + .send() + .await + .context("HeadObject on the SSE-S3 probe object")?; + ensure!( + head.server_side_encryption() == Some(&ServerSideEncryption::Aes256), + "object written without SSE headers is not AES256-encrypted: {:?}", + head.server_side_encryption() + ); + + let body = s3 + .get_object() + .bucket(bucket) + .key(SSE_S3_PROBE_OBJECT) + .send() + .await + .context("GetObject on the SSE-S3 probe object")? + .body + .collect() + .await? + .into_bytes(); + ensure!( + body.as_ref() == payload.as_slice(), + "decrypted object differs from the uploaded payload ({} vs {} bytes)", + body.len(), + payload.len() + ); + + s3.delete_object() + .bucket(bucket) + .key(SSE_S3_PROBE_OBJECT) + .send() + .await + .context("delete the SSE-S3 probe object")?; + s3.delete_bucket() + .bucket(bucket) + .send() + .await + .context("delete the SSE-S3 probe bucket")?; + Ok(()) +} + +async fn restore_smoke_tenant( + config: &E2eConfig, + kubectl: &Kubectl, + secret_name: &str, +) -> Result<()> { + resources::apply_smoke_tenant_resources(config)?; + let client = kube_client::default_client().await?; + let tenants: Api = kube_client::tenant_api(client, &config.test_namespace); + wait::wait_for_tenant_ready(tenants, &config.tenant_name, config.timeout).await?; + kubectl + .clone() + .namespaced(&config.test_namespace) + .command(["delete", "secret", secret_name, "--ignore-not-found=true"]) + .run_checked()?; + Ok(()) +} + +fn sse_s3_secret_manifest(config: &E2eConfig, secret_name: &str) -> String { + format!( + r#"apiVersion: v1 +kind: Secret +metadata: + name: {secret_name} + namespace: {namespace} +type: Opaque +stringData: + {key}: {value} +"#, + namespace = config.test_namespace, + key = SSE_S3_MASTER_KEY_SECRET_KEY, + value = SSE_S3_TEST_MASTER_KEY, + ) +} + +fn sse_s3_tenant_manifest(config: &E2eConfig, secret_name: &str, bucket: &str) -> Result { + let mut tenant = serde_json::to_value(resources::smoke_tenant_template(config).build())?; + let spec = tenant["spec"] + .as_object_mut() + .context("smoke Tenant has no spec")?; + spec.entry("env") + .or_insert_with(|| json!([])) + .as_array_mut() + .context("smoke Tenant spec.env is not a list")? + .push(json!({ + "name": SSE_S3_MASTER_KEY_ENV, + "valueFrom": { + "secretKeyRef": {"name": secret_name, "key": SSE_S3_MASTER_KEY_SECRET_KEY} + } + })); + spec.insert( + "buckets".to_string(), + json!([{"name": bucket, "encryption": {"mode": "SSE-S3"}}]), + ); + Ok(serde_yaml_ng::to_string(&tenant)?) +} + +async fn wait_for_bucket_encryption_applied( + tenants: &Api, + tenant_name: &str, + bucket: &str, + timeout: Duration, +) -> Result<()> { + wait::wait_until( + &format!("bucket {bucket} SSE-S3 encryption to be applied"), + timeout, + Duration::from_secs(5), + || async move { + let tenant = tenants.get(tenant_name).await?; + let applied = tenant.status.as_ref().is_some_and(|status| { + status.provisioning.buckets.iter().any(|item| { + item.item.name == bucket + && item.item.state == "Ready" + && item.encryption_desired_hash.is_some() + && item.encryption_desired_hash == item.encryption_last_applied_hash + }) + }); + Ok(applied.then_some(())) + }, + ) + .await +} + +async fn s3_client(endpoint: &str) -> Client { + let (access_key, secret_key) = resources::test_credentials(); + let credentials = Credentials::new(access_key, secret_key, None, None, "rustfs-e2e-sse-s3"); + let shared_config = aws_config::defaults(BehaviorVersion::latest()) + .region(Region::new("us-east-1")) + .credentials_provider(credentials) + .endpoint_url(endpoint) + .load() + .await; + Client::from_conf( + aws_sdk_s3::config::Builder::from(&shared_config) + .force_path_style(true) + .build(), + ) +} + fn verify_status_subresource_contract(kubectl: &Kubectl, namespace: &str) -> Result<()> { kubectl .apply_yaml_command(CHECKPOINT_TEST_CRD_YAML)