diff --git a/CHANGELOG.md b/CHANGELOG.md index a2b823c..6f01f3d 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, + 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/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..8ce920f 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 Tenant `spec.encryption.defaultKeyId`. + maxLength: 1024 + minLength: 1 + nullable: true + pattern: ^[A-Za-z0-9._:/+=@-]+$ + type: string + mode: + description: |- + `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 + - 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..62ed2c0 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` 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. 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. @@ -1049,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: @@ -1062,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: @@ -1081,6 +1098,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..5add73f 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`，并要求 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，也不会改动现有配置。 迁移类操作会把数据移动到远程层（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。 @@ -993,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: @@ -1006,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: @@ -1025,6 +1042,8 @@ spec: objectLockConfiguration: mode: Compliance days: 30 + encryption: + mode: SSE-S3 anonymous: Download lifecycle: state: Present 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) 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..c033a5d 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; @@ -43,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; @@ -55,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, @@ -2462,6 +2467,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(|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; @@ -2487,6 +2496,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 +3013,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 +3114,375 @@ 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) + .and_then(|desired| resolve_bucket_encryption(run.tenant, desired)) + { + 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::Aes256) + && let Err(error) = check_sse_s3_provider(run).await + { + 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; + } + + 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 live.as_ref() == Some(&desired) { + 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 verified == desired => 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; + } + }; + 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)) +} + +/// 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())) } async fn reconcile_bucket_lifecycle( @@ -4204,6 +4593,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, @@ -4229,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; @@ -4266,8 +4658,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 +4681,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 +4841,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 +7138,726 @@ 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") + } + + // 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", + 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 tenant = sse_s3_tenant(); + 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 tenant = sse_s3_tenant(); + 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 tenant = sse_s3_tenant(); + 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 tenant = kms_tenant(Some("rustfs-default-key")); + 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_eq!( + item.encryption_desired_hash, + item.encryption_last_applied_hash + ); + drop(run); + + // 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( + &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 tenant = sse_s3_tenant(); + 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 tenant = sse_s3_tenant(); + 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(); + } + + #[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 { + 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..ac014e2 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` 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 Tenant `spec.encryption.defaultKeyId`. + #[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, } } }