spetz commented on code in PR #4615: URL: https://github.com/apache/iggy/pull/4615#discussion_r4340261593
########## core/server/src/stream_key_rewrapper.rs: ########## @@ -0,0 +1,538 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::shell::{ServerShard, ShellBus, ShellShard}; +use bytes::Bytes; +use consensus::MetadataHandle; +use iggy_binary_protocol::PrepareHeader; +use iggy_binary_protocol::requests::streams::{RewrapStreamKeyEntry, RewrapStreamKeysRequest}; +use iggy_common::{HeaderKey, IggyError, OptionKey, ServerKeyContext, stream_wrap_set}; +use journal::superblock::SuperblockStore; +use journal::{Journal, JournalHandle}; +use metadata::impls::metadata::StreamsFrontend; +use metadata::stm::stream::{Stream, StreamsInner}; +use server_common::Message; +use shard::Receiver; +use std::collections::BTreeSet; +use std::rc::Rc; +use std::time::Duration; +use tracing::{info, trace, warn}; + +/// How often the metadata primary checks whether a sweep is due. A tick with +/// an unchanged view and applied op does nothing else. +pub const REWRAP_INTERVAL: Duration = Duration::from_secs(5); +/// Upper bound on streams rewrapped by one replicated operation, so one +/// rotation never monopolizes the metadata pipeline. +const MAX_ENTRIES_PER_OPERATION: usize = 64; + +/// What the rewrapper keeps between ticks. +#[derive(Debug, Default)] +struct Rewrapper { + /// Metadata view and last applied op of the last finished sweep. + swept: Option<(u32, u64)>, + /// Stream incarnations already reported as stuck, so each warns once. + stuck: BTreeSet<(usize, u64)>, + /// Whether the last finished sweep found every set settled. + settled: bool, +} + +/// Outcome of one tick. +#[derive(Debug, PartialEq, Eq)] +enum Sweep { + Stopped, + NotPrimary, + Unchanged, + Interrupted, + Finished, +} + +/// A stream that needs a rewrap this node cannot make. +#[derive(Debug)] +struct StuckStream { + stream_id: usize, + created_op: u64, + kek_fingerprints: String, + error: IggyError, +} + +/// Run the rewrapper until `stop` fires, planning with `context`, this node's +/// keys plus the retired fingerprints. Spawned only when that context is +/// rotating. Only the caught-up metadata primary proposes, so a rotation is +/// replicated once and every replica stores the same wrap sets. A node never +/// rewrites them locally. +pub async fn run_stream_key_rewrapper( + shard: Rc<ServerShard>, + context: ServerKeyContext, + stop: Receiver<()>, +) { + trace!(shard = shard.id, "stream key rewrapper started"); + let mut rewrapper = Rewrapper::default(); + loop { + if compio::time::timeout(REWRAP_INTERVAL, stop.recv()) + .await + .is_ok() + || rewrapper.sweep(&shard, &context, &stop).await == Sweep::Stopped + { + break; + } + } + trace!(shard = shard.id, "stream key rewrapper exited"); +} + +impl Rewrapper { + /// Scan every stream once when this node is the caught-up primary and + /// the metadata moved since the last finished sweep, then propose the + /// unsettled ones in bounded operations. + async fn sweep<B, MJ, S, SB>( + &mut self, + shard: &Rc<ShellShard<B, MJ, S, SB>>, + context: &ServerKeyContext, + stop: &Receiver<()>, + ) -> Sweep + where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, + S: 'static, + SB: SuperblockStore + 'static, + { + if stop.try_recv().is_ok() { + return Sweep::Stopped; + } + let metadata = shard.plane.metadata(); + let Some(consensus) = metadata + .consensus + .as_ref() + .filter(|_| metadata.is_caught_up_primary()) + else { + self.swept = None; + return Sweep::NotPrimary; + }; + // Read with the scan below and no await between, so the mark names + // exactly the state the scan saw. + let mark = (consensus.view(), consensus.commit_min()); + if self.swept == Some(mark) { + return Sweep::Unchanged; + } + let pending = metadata + .mux_stm + .streams() + .read(|streams| unsettled_streams(streams, context)); + let mut rewrapped = 0; + let mut stuck = BTreeSet::new(); + for ids in pending.chunks(MAX_ENTRIES_PER_OPERATION) { Review Comment: Added `given_more_stale_streams_than_one_operation_when_sweeping_should_commit_every_chunk` at `core/server/src/stream_key_rewrapper.rs:403`. It seeds `2 * 64 + 8` stale streams and requires one `sweep()` to commit `ceil(n / 64)` rewraps and settle every set. The `chunks(64).take(1)` mutant fails it. ########## core/server/src/boot/recovery.rs: ########## @@ -204,8 +235,35 @@ pub(in crate::boot) async fn build_shard_for_thread( // bundle was broadcast (see `MetadataHandoff::Owner`). All shards // here only add their per-partition deltas, so the shared // `Arc<TopicStats>` atomics race only against other atomic adds. - for (stream_id, topic_id, partition_stats, partition_metadata, topic_runtime) in owned { + let mut deferred_streams = BTreeSet::new(); + for OwnedPartition { + stream_id, + topic_id, + stats: partition_stats, + metadata: partition_metadata, + runtime: topic_runtime, + identity, + encryption, + } in owned + { + if deferred_streams.contains(&stream_id) { + continue; Review Comment: Boot now hands the reconciler the namespaces it did not build (deferred keys and refused offset reservations), and the removals pass tears down any owned namespace that never built once it leaves the target, also when it came from a failed `Add` (`core/server/src/partition_reconciler.rs:1055`). Covered by `given_unbuilt_encrypted_partition_when_its_stream_is_deleted_should_remove_its_files`. ########## core/common/src/utils/stream_key.rs: ########## @@ -0,0 +1,1410 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::IggyError; +use crate::utils::crypto::{Aes256GcmSivEncryptor, EncryptorKind}; +use crate::utils::text; +use aes_gcm::aead::{Aead, Generate, Payload}; +use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; +use base64::Engine; +use base64::engine::general_purpose::STANDARD; +use ring::rand::{SecureRandom, SystemRandom}; +use ring::{digest, hkdf}; +use secrecy::zeroize::Zeroizing; +use secrecy::{ExposeSecret, ExposeSecretMut, SecretBox}; +use std::fmt::{Debug, Formatter}; +use std::str::FromStr; +use std::sync::Arc; + +pub use iggy_binary_protocol::requests::streams::{ + MAX_WRAP_SET_ENVELOPES, WRAPPED_STREAM_KEY_LENGTH, +}; + +pub const STREAM_KEY_LENGTH: usize = 32; +/// Most retired config keys a node keeps for unwrapping during a rotation. +pub const MAX_PREVIOUS_KEYS: usize = MAX_WRAP_SET_ENVELOPES - 1; + +const WRAPPED_STREAM_KEY_VERSION: u8 = 1; +const FINGERPRINT_LENGTH: usize = 8; +const NONCE_LENGTH: usize = 12; +const WRAPPED_KEY_CIPHERTEXT_LENGTH: usize = STREAM_KEY_LENGTH + 16; +const ORIGIN_OFFSET: usize = 1; +const KEK_FINGERPRINT_OFFSET: usize = ORIGIN_OFFSET + 1; +const NONCE_OFFSET: usize = KEK_FINGERPRINT_OFFSET + FINGERPRINT_LENGTH; +const CIPHERTEXT_OFFSET: usize = NONCE_OFFSET + NONCE_LENGTH; +const KEY_FINGERPRINT_OFFSET: usize = CIPHERTEXT_OFFSET + WRAPPED_KEY_CIPHERTEXT_LENGTH; +const STREAM_KEY_WRAP_INFO: &[u8] = b"iggy/stream-key-wrap/v1"; +const STREAM_KEY_AAD_LABEL: &[u8] = b"iggy/stream-key/v1"; +const STREAM_KEY_FINGERPRINT_LABEL: &[u8] = b"iggy/stream-key-fingerprint/v1"; +const KEK_FINGERPRINT_LABEL: &[u8] = b"iggy/stream-kek-fingerprint/v1"; + +/// A stream's own data encryption key: 32 uniformly random bytes. +/// +/// Never printed and never serialized. The bytes leave this type only to be +/// wrapped, encoded into a create request, or turned into an encryptor. A key +/// derived from a password turns its public fingerprint into an offline +/// guessing oracle, so callers must draw it from a secure random generator. +pub struct StreamEncryptionKey(SecretBox<[u8; STREAM_KEY_LENGTH]>); + +/// Who chose a stream's key, recorded and authenticated in its envelope. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum StreamKeyOrigin { + /// The client sent the key with the create. + Client = 1, + /// The server generated the key because the client sent none. + Server = 2, +} + +/// A stream key wrapped under one key-encryption key. +/// +/// Layout: `[version:1][origin:1][kek_fingerprint:8][nonce:12][ciphertext:32][tag:16][key_fingerprint:8]`. +/// The version, the origin and both fingerprints are authenticated as +/// associated data. +#[derive(Clone, PartialEq, Eq)] +pub struct WrappedStreamKey([u8; WRAPPED_STREAM_KEY_LENGTH]); + +/// One stream key wrapped under one or more key-encryption keys: one envelope +/// outside a rotation, one per configured key during it. All envelopes hold +/// the same key from the same origin, sorted by KEK fingerprint, none twice. +#[derive(Clone, PartialEq, Eq)] +pub struct WrapSet { + envelopes: Vec<WrappedStreamKey>, + bytes: Vec<u8>, +} + +/// The key-encryption keys derived from the server config: the current one, +/// the previous ones a rotation still unwraps with, and the fingerprints of +/// retired ones whose wraps a rewrap drops. +#[derive(Clone)] +pub struct ServerKeyContext { + current: Option<KeyEncryptionKey>, + previous: Vec<KeyEncryptionKey>, + retired: Vec<KekFingerprint>, + generates: bool, +} + +#[derive(Clone)] +struct KeyEncryptionKey { + cipher: Aes256Gcm, + fingerprint: [u8; FINGERPRINT_LENGTH], +} + +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +struct KekFingerprint([u8; FINGERPRINT_LENGTH]); + +struct DerivedKeyLength; + +impl StreamEncryptionKey { + /// A new random key from the operating system's secure generator. + /// + /// # Errors + /// + /// `CannotEncryptData` when the generator fails. + pub fn generate() -> Result<Self, IggyError> { + let mut key = SecretBox::new(Box::new([0u8; STREAM_KEY_LENGTH])); + SystemRandom::new() + .fill(key.expose_secret_mut()) + .map_err(|_| IggyError::CannotEncryptData)?; + Ok(Self(key)) + } + + /// 16 lowercase hex characters identifying this key without revealing it, + /// as long as the key is uniformly random. + #[must_use] + pub fn fingerprint(&self) -> String { + hex(&self.fingerprint_bytes()) + } + + /// The raw key, for the request encoder and the stream encryptor. + #[must_use] + pub fn as_bytes(&self) -> &[u8; STREAM_KEY_LENGTH] { + self.0.expose_secret() + } + + /// The encryptor for this stream's message payloads and user headers. + /// + /// # Errors + /// + /// `InvalidEncryptionKey` when the cipher refuses the key. + pub fn encryptor(&self) -> Result<Arc<EncryptorKind>, IggyError> { + Ok(Arc::new(EncryptorKind::Aes256GcmSiv( + Aes256GcmSivEncryptor::new(self.as_bytes())?, + ))) + } + + fn fingerprint_bytes(&self) -> [u8; FINGERPRINT_LENGTH] { + labelled_fingerprint(STREAM_KEY_FINGERPRINT_LABEL, self.as_bytes()) + } +} + +impl WrappedStreamKey { + /// Who chose the key inside this envelope. + #[must_use] + pub fn origin(&self) -> StreamKeyOrigin { + // Construction refuses any other origin byte. + if self.0[ORIGIN_OFFSET] == StreamKeyOrigin::Server as u8 { + StreamKeyOrigin::Server + } else { + StreamKeyOrigin::Client + } + } + + /// Fingerprint of the stream key inside this envelope. + #[must_use] + pub fn key_fingerprint(&self) -> String { + hex(self.key_fingerprint_bytes()) + } + + /// Fingerprint of the key-encryption key that wrapped this envelope. + #[must_use] + pub fn kek_fingerprint(&self) -> String { + hex(self.kek_fingerprint_bytes()) + } + + fn key_fingerprint_bytes(&self) -> &[u8] { + &self.0[KEY_FINGERPRINT_OFFSET..] + } + + fn kek_fingerprint_bytes(&self) -> &[u8] { + &self.0[KEK_FINGERPRINT_OFFSET..NONCE_OFFSET] + } + + fn associated_data(&self) -> Vec<u8> { + associated_data( + self.0[ORIGIN_OFFSET], + self.kek_fingerprint_bytes(), + self.key_fingerprint_bytes(), + ) + } +} + +impl WrapSet { + /// A set of `envelopes` in canonical order. + /// + /// # Errors + /// + /// `UnsupportedCriticalOption` for an empty set, more than three + /// envelopes, a KEK fingerprint twice, or envelopes of different keys or + /// origins. + pub fn new(mut envelopes: Vec<WrappedStreamKey>) -> Result<Self, IggyError> { + envelopes.sort_by(|left, right| { + left.kek_fingerprint_bytes() + .cmp(right.kek_fingerprint_bytes()) + }); + let Some(first) = envelopes.first() else { + return Err(unsupported_wrapped_key()); + }; + let consistent = envelopes.len() <= MAX_WRAP_SET_ENVELOPES + && envelopes.iter().all(|envelope| { + envelope.origin() == first.origin() + && envelope.key_fingerprint_bytes() == first.key_fingerprint_bytes() + }) + && envelopes + .windows(2) + .all(|pair| pair[0].kek_fingerprint_bytes() < pair[1].kek_fingerprint_bytes()); + if !consistent { + return Err(unsupported_wrapped_key()); + } + let bytes = envelopes.iter().flat_map(|envelope| envelope.0).collect(); + Ok(Self { envelopes, bytes }) + } + + #[must_use] + pub fn envelopes(&self) -> &[WrappedStreamKey] { + &self.envelopes + } + + /// Who chose the key every envelope holds. + #[must_use] + pub fn origin(&self) -> StreamKeyOrigin { + self.envelopes[0].origin() + } + + /// Fingerprint of the key every envelope holds. + #[must_use] + pub fn key_fingerprint(&self) -> String { + self.envelopes[0].key_fingerprint() + } + + /// The raw fingerprint of the key every envelope holds. + #[must_use] + pub fn key_fingerprint_bytes(&self) -> &[u8] { + self.envelopes[0].key_fingerprint_bytes() + } + + /// Fingerprints of the key-encryption keys, in canonical order. + #[must_use] + pub fn kek_fingerprints(&self) -> Vec<String> { + self.envelopes + .iter() + .map(WrappedStreamKey::kek_fingerprint) + .collect() + } +} + +impl ServerKeyContext { + /// No config key: streams stay plaintext and stream keys are refused. + #[must_use] + pub const fn disabled() -> Self { + Self { + current: None, + previous: Vec::new(), + retired: Vec::new(), + generates: false, + } + } + + /// Derive the key-encryption keys from the base64 config keys. + /// + /// An empty `key` is valid only when encryption is disabled and no + /// previous key is set, and then nothing is derived. A supplied `key` + /// always derives the key-encryption key, so a client may bring its own + /// stream key while keyless streams stay plaintext. `enabled` decides + /// whether a keyless stream gets a generated key. + /// + /// # Errors + /// + /// `InvalidEncryptionKey` when a key is not 32 bytes of base64, when + /// `key` is empty while `enabled` is true or previous keys are set, for + /// more than two previous keys, or for a key configured twice. + pub fn from_config( + key: &str, + previous_keys: &[String], + enabled: bool, + ) -> Result<Self, IggyError> { + if key.is_empty() { + return if enabled || !previous_keys.is_empty() { + Err(IggyError::InvalidEncryptionKey) + } else { + Ok(Self::disabled()) + }; + } + if previous_keys.len() > MAX_PREVIOUS_KEYS { + return Err(IggyError::InvalidEncryptionKey); + } + let current = KeyEncryptionKey::from_config_key(key)?; + let previous = previous_keys + .iter() + .map(|previous| KeyEncryptionKey::from_config_key(previous)) + .collect::<Result<Vec<_>, _>>()?; + let mut fingerprints: Vec<_> = previous.iter().map(|kek| kek.fingerprint).collect(); + fingerprints.push(current.fingerprint); + fingerprints.sort_unstable(); + if fingerprints.windows(2).any(|pair| pair[0] == pair[1]) { + return Err(IggyError::InvalidEncryptionKey); + } + Ok(Self { + current: Some(current), + previous, + retired: Vec::new(), + generates: enabled, + }) + } + + /// This context with the key-encryption key fingerprints whose wraps a + /// rewrap drops, as reads print them. + /// + /// # Errors + /// + /// `InvalidEncryptionKey` without a config key, for an entry that is not + /// 16 lowercase hex characters, one naming a configured key, or one given + /// twice. + pub fn with_retired_kek_fingerprints( + mut self, + fingerprints: &[String], + ) -> Result<Self, IggyError> { + if fingerprints.is_empty() { + return Ok(self); + } + if self.current.is_none() { + return Err(IggyError::InvalidEncryptionKey); + } + let mut retired = fingerprints + .iter() + .map(|fingerprint| fingerprint.parse::<KekFingerprint>()) + .collect::<Result<Vec<_>, _>>()?; + retired.sort_unstable(); + let repeated = retired.windows(2).any(|pair| pair[0] == pair[1]); + let configured = retired + .iter() + .any(|retired| self.keks().any(|kek| kek.fingerprint == retired.0)); + if repeated || configured { + return Err(IggyError::InvalidEncryptionKey); + } + self.retired = retired; + Ok(self) + } + + /// Whether a stream may bring its own key on this node. + #[must_use] + pub const fn accepts_stream_keys(&self) -> bool { + self.current.is_some() + } + + /// Whether this node gives a keyless stream a generated key. + #[must_use] + pub const fn generates_stream_keys(&self) -> bool { + self.current.is_some() && self.generates + } + + /// Fingerprint of the key-encryption key that wraps, when a key is set. + #[must_use] + pub fn kek_fingerprint(&self) -> Option<String> { + self.current.as_ref().map(|kek| hex(&kek.fingerprint)) + } + + /// Whether this node's keys can call for a rewrap: a previous key, whose + /// streams need a wrap under the current key, or a retired key to drop. + /// Without either no stored set can differ from its target. + #[must_use] + pub const fn rotating(&self) -> bool { + self.current.is_some() && !(self.previous.is_empty() && self.retired.is_empty()) + } + + /// Fingerprints of every configured key-encryption key, current first. + #[must_use] + pub fn kek_fingerprints(&self) -> Vec<String> { + self.keks().map(|kek| hex(&kek.fingerprint)).collect() + } + + /// Wrap a stream key under the current key-encryption key. + /// + /// # Errors + /// + /// `StreamEncryptionUnavailable` without a config key, `CannotEncryptData` + /// when the cipher fails. + pub fn wrap( + &self, + key: &StreamEncryptionKey, + origin: StreamKeyOrigin, + ) -> Result<WrappedStreamKey, IggyError> { + self.current + .as_ref() + .ok_or(IggyError::StreamEncryptionUnavailable)? + .wrap(key, origin) + } + + /// The wrap set a new stream stores: its key wrapped under every + /// configured key-encryption key, so it also opens on nodes that still + /// hold only a previous key while a rotation rolls out. + /// + /// # Errors + /// + /// As [`Self::wrap`]. + pub fn wrap_set( + &self, + key: &StreamEncryptionKey, + origin: StreamKeyOrigin, + ) -> Result<WrapSet, IggyError> { + if self.current.is_none() { + return Err(IggyError::StreamEncryptionUnavailable); + } + let envelopes = self + .keks() + .map(|kek| kek.wrap(key, origin)) + .collect::<Result<Vec<_>, _>>()?; + WrapSet::new(envelopes) + } + + /// Authenticate and unwrap one envelope with the configured key-encryption + /// key it names. + /// + /// # Errors + /// + /// `StreamEncryptionUnavailable` without a config key, `CannotDecryptData` + /// when no configured key matches, the envelope does not authenticate, or + /// its key fingerprint does not match the key inside. + pub fn unwrap(&self, wrapped: &WrappedStreamKey) -> Result<StreamEncryptionKey, IggyError> { + if self.current.is_none() { + return Err(IggyError::StreamEncryptionUnavailable); + } + let kek = self + .keks() + .find(|kek| kek.fingerprint == wrapped.kek_fingerprint_bytes()) + .ok_or(IggyError::CannotDecryptData)?; + let nonce = Nonce::try_from(&wrapped.0[NONCE_OFFSET..CIPHERTEXT_OFFSET]) + .map_err(|_| IggyError::CannotDecryptData)?; + let plaintext = Zeroizing::new( + kek.cipher + .decrypt( + &nonce, + Payload { + msg: &wrapped.0[CIPHERTEXT_OFFSET..KEY_FINGERPRINT_OFFSET], + aad: &wrapped.associated_data(), + }, + ) + .map_err(|_| IggyError::CannotDecryptData)?, + ); + let key = StreamEncryptionKey::try_from(plaintext.as_slice()) + .map_err(|_| IggyError::CannotDecryptData)?; + if key.fingerprint_bytes() != wrapped.key_fingerprint_bytes() { + return Err(IggyError::CannotDecryptData); + } + Ok(key) + } + + /// The stream key of a wrap set, from the first envelope that opens. + /// + /// # Errors + /// + /// As [`Self::unwrap`], from the last envelope tried. + pub fn open(&self, set: &WrapSet) -> Result<StreamEncryptionKey, IggyError> { + let mut last = IggyError::CannotDecryptData; + for envelope in set.envelopes() { + match self.unwrap(envelope) { + Ok(key) => return Ok(key), + Err(error) => last = error, + } + } + Err(last) + } + + /// Whether a stored wrap set already is what a rewrap on this node would + /// store: one envelope under the current key-encryption key and none + /// under a retired one. Reads only the fingerprints in the raw stored + /// value, so a sweep over every stream allocates nothing. Without a + /// config key nothing can change, so every set is settled. + #[must_use] + pub fn is_settled(&self, stored: &[u8]) -> bool { + let Some(current) = &self.current else { + return true; + }; + let (envelopes, rest) = stored.as_chunks::<WRAPPED_STREAM_KEY_LENGTH>(); + rest.is_empty() + && envelopes.iter().any(|envelope| { + envelope[KEK_FINGERPRINT_OFFSET..NONCE_OFFSET] == current.fingerprint + }) + && !envelopes + .iter() + .any(|envelope| self.retires(&envelope[KEK_FINGERPRINT_OFFSET..NONCE_OFFSET])) + } + + /// The set a rewrap should store instead of `stored`: every envelope not + /// under a retired key-encryption key, plus one under the current key + /// when missing. Envelopes under keys this node does not hold stay, + /// because a node still being rolled out may need them. `None` when + /// nothing changes. + /// + /// # Errors + /// + /// When the set must change but this node cannot do it, and the set is + /// left alone: `CannotDecryptData` when no envelope opens here, + /// `UnsupportedCriticalOption` when the set already holds the most + /// envelopes and none is retired, `CannotEncryptData` when wrapping + /// fails. + pub fn target_set(&self, stored: &WrapSet) -> Result<Option<WrapSet>, IggyError> { + let Some(current) = &self.current else { + return Ok(None); + }; + if self.is_settled(stored.as_ref()) { + return Ok(None); + } + let key = self.open(stored)?; + let mut envelopes: Vec<WrappedStreamKey> = stored + .envelopes() + .iter() + .filter(|envelope| !self.retires(envelope.kek_fingerprint_bytes())) + .cloned() + .collect(); + if !envelopes + .iter() + .any(|envelope| envelope.kek_fingerprint_bytes() == current.fingerprint) + { + envelopes.push(current.wrap(&key, stored.origin())?); + } + WrapSet::new(envelopes).map(Some) + } Review Comment: `target_set` now returns `RewrapPlanError::WrapSetFull` for a full set with no current or retired wrap (`core/common/src/utils/stream_key.rs:555`), and the rewrapper logs it with a hint to add a fingerprint to `retired_kek_fingerprints`. The three-wrap cap is documented with the rotation steps in `core/server/config.toml:445` and the server README. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
