numinnex commented on code in PR #4615:
URL: https://github.com/apache/iggy/pull/4615#discussion_r4339589269
##########
core/server/src/boot/recovery.rs:
##########
@@ -398,6 +458,98 @@ const _: () =
assert!(consensus::DVC_HEADERS_MAX ==
iggy_binary_protocol::consensus::DVC_HEADERS_MAX);
const _: () = assert!(consensus::DVC_HEADERS_MAX == u128::BITS as usize);
+/// The boot check of encrypted streams that own no partition, whose keys no
+/// shard opens while building its partitions.
+fn refuse_partitionless_stream_keys(
+ streams: &Streams,
+ key_context: &ServerKeyContext,
+ solo: bool,
+) -> Result<(), ServerError> {
+ let unopenable: Vec<(usize, StreamEncryption, IggyError)> =
streams.read(|inner| {
+ inner
+ .items
+ .iter()
+ .filter(|(_, stream)| {
+ stream
+ .topics
+ .iter()
+ .all(|(_, topic)| topic.partitions.is_empty())
+ })
+ .filter_map(|(_, stream)| {
+ let encryption = StreamEncryption::from(&stream.options);
+ let source = match &encryption {
+ StreamEncryption::Plaintext => None,
+ StreamEncryption::Unavailable(error) =>
Some(error.clone()),
+ StreamEncryption::Wrapped(set) =>
key_context.open(set).err(),
+ }?;
+ Some((stream.id, encryption, source))
+ })
+ .collect()
+ });
+ for (stream_id, encryption, source) in unopenable {
+ defer_or_refuse_stream_key(stream_id, &encryption, key_context,
source, solo)?;
+ }
+ Ok(())
+}
+
+/// A stream key that does not open at boot. A cluster replica's metadata can
+/// trail the commit point until catch-up or state transfer, so the stream
waits
+/// for the reconciler, which builds its partitions once a wrap this node opens
+/// commits. Catch-up heals neither a node without a key nor unreadable
critical
+/// state, and a solo replay is the whole journal, so these refuse.
+fn defer_or_refuse_stream_key(
+ stream_id: usize,
+ encryption: &StreamEncryption,
+ key_context: &ServerKeyContext,
+ source: IggyError,
+ solo: bool,
+) -> Result<(), ServerError> {
+ let stored_kek_fingerprints = match encryption {
+ StreamEncryption::Wrapped(set) => set.kek_fingerprints().join(","),
+ StreamEncryption::Plaintext | StreamEncryption::Unavailable(_) =>
String::new(),
+ };
+ let error = ServerError::StreamKeyUnavailable {
+ stream_id,
+ stored_kek_fingerprints,
+ node_kek_fingerprints: key_context.kek_fingerprints().join(","),
+ source: Box::new(source),
+ };
+ if solo
Review Comment:
**warning**: deferral turns a node holding a KEK no peer holds (key set to
the new key only, before phase 1 finished) from a boot failure into a healthy
replica that can win a view and act as edge. Its creates wrap only under its
own KEKs (`stream_keys.rs:122`), and the `CreateStream` prepare arm
(`core/metadata/src/impls/metadata.rs:4384`) checks only for Secret keys, never
the wrap KEKs.
Peers then leave those partitions unbuilt
(`partition_reconciler.rs:801-817`): with a minority on the wrong key new
encrypted streams never reach quorum, with a majority data commits under a wrap
the correct nodes cannot open.
Fix: a node still deferring streams after catch-up refuses to wrap new
creates (`StreamEncryptionUnavailable`), and prepare refuses a wrap set with no
envelope the primary can open. KEK agreement in the replica handshake can come
later.
##########
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;
+ }
let namespace = IggyNamespace::new(stream_id, topic_id,
partition_metadata.id);
+ let encryptor = match partitions.resolve_stream_encryptor(identity,
&encryption) {
+ Ok(encryptor) => encryptor,
+ Err(source) => {
+ defer_or_refuse_stream_key(
+ stream_id,
+ &encryption,
+ &partitions.key_context(),
+ source,
+ solo,
+ )?;
+ deferred_streams.insert(stream_id);
+ continue;
Review Comment:
**warning**: this `continue` is the only thing keeping a deferred encrypted
partition unbuilt, and no test notices if it goes. Replacing it with "build the
partition with encryptor `None`" passes the server lib suite (571/571) and all
11 `cluster::stream_encryption` / `cluster::stream_key_rotation` tests, which
check log markers and payloads polled through the leader.
Under that mutant a lagging or wrong-key replica hosts the partition keyless
for good (`set_encryptor` runs once at build), and as partition primary it
stores new sends in plaintext (`iggy_partitions.rs:859-860` encrypts only with
`Some`).
Fix: extract the owned-partition key decision into a function and unit-test
that a deferred stream gets no partition and no shards_table row, or have a
cluster test assert the deferring node hosts nothing.
##########
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());
Review Comment:
**warning**: no test pins this rescan trigger. `let mark =
(consensus.view(), 0)` passes all 16 server lib `stream_key` tests, and the
cluster rotation tests create every stream before the rollout or on the rotated
primary, so they cannot see it either.
If it breaks, a stream created mid phase 1 through an old-config edge after
the rotated primary's first sweep is never rewrapped in a stable view, and
stays deferred on phase 2 nodes.
Fix: one `sweep()`-level test on a journal-backed primary: settle all,
create a stale stream, assert the next sweep is not `Unchanged` and rewraps it.
##########
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:
**warning**: the one-scan sweep is linear now (50k streams: 1.8 s -> 41 ms
CPU), but only the planning helper is tested. A mutant that proposes just the
first 64-entry chunk per sweep (`chunks(64).take(1)`) passes all 16 rewrapper
tests, since the cited test never calls `sweep`. Under it a 1M-stream rotation
needs one tick per 64 streams, about 22 h.
Fix: the same `sweep()` test with `2 * 64 + 8` stale streams, asserting
`ceil(n / 64)` commits in one sweep and every set settled.
##########
core/server/config.toml:
##########
@@ -421,15 +421,37 @@ sysinfo_print_interval = "10 s"
# Encryption configuration
[encryption]
# Encrypt message payloads and user headers. Metadata and structural headers
remain unencrypted.
-# `true` enables encryption for stored data using AES-256-GCM.
+# `true` enables encryption for stored payloads and user headers using
AES-256-GCM-SIV.
# `false` means data is stored without encryption.
enabled = false
-# The encryption key used when encryption is enabled (string).
-# Should be a 32 bytes length key, provided as a base64 encoded string.
-# This key is required and used only if encryption is enabled.
+# The server encryption key (string), 32 bytes as standard base64.
+# Every encrypted stream has its own data key, stored wrapped under a key
derived from this one.
+# Required when encryption is enabled. With encryption disabled a key may
still be set:
+# streams created with their own key are then encrypted, keyless streams stay
plaintext.
+# A stream created without encryption stays plaintext for life.
key = ""
+# Earlier server keys (list of strings) a key rotation still needs, at most
two.
+# A node unwraps stream keys with them and wraps every new stream key under
them too.
+# Rotation runs in two phases:
+# 1. Set `key` to the new key and `previous_keys` to the old one on every
node. The metadata
+# primary adds a wrap under the new key to every encrypted stream and keeps
the old wraps,
+# so nodes still on the old key keep serving and booting, new streams
included.
+# 2. Once every node runs the phase 1 config and every encrypted stream
reports the new
+# key's fingerprint in `encryption_kek_fingerprint`, empty `previous_keys`
and put the
+# old key's fingerprint in `retired_kek_fingerprints`. The primary drops
the wraps under it.
+# Messages are never re-encrypted. Losing a key that still wraps a stream
makes it unreadable.
+# As an environment variable: IGGY_ENCRYPTION_PREVIOUS_KEYS='["<old base64
key>"]', a single key may also be given bare.
+previous_keys = []
+
+# Fingerprints of retired server keys (list of strings), 16 lowercase hex
characters each, as
Review Comment:
**nit**: nothing says to clear `retired_kek_fingerprints` once phase 2 is
done. While it is set `rotating()` stays true (`stream_key.rs:375-377`), so the
rewrapper rescans every stream on each tick where metadata moved. Logins and
logouts are metadata ops, so that can be every 5 s: a 61 ms shard-0 stall at 1M
encrypted streams. Clearing it after the drop is safe.
Fix: add that step here, in `core/server/README.md` and in the helm README.
##########
foreign/php/src/sensitive_parameters.rs:
##########
@@ -0,0 +1,134 @@
+// 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 std::{
+ ffi::{CStr, c_char, c_void},
+ mem,
+ sync::{Mutex, PoisonError},
+};
+
+use ext_php_rs::{
+ ffi::{HashTable, zend_function, zend_hash_str_find_ptr_lc, zend_string},
+ types::ZendStr,
+ zend::ClassEntry,
+};
+
+type AddAttribute = unsafe extern "C" fn(
+ attributes: *mut *mut HashTable,
+ name: *mut zend_string,
+ argc: u32,
+ flags: u32,
+ offset: u32,
+ lineno: u32,
+) -> *mut c_void;
+type GetParameterAttribute = unsafe extern "C" fn(
+ attributes: *mut HashTable,
+ name: *const c_char,
+ length: usize,
+ offset: u32,
+) -> *mut c_void;
+
+const SENSITIVE_PARAMETER: &str = "SensitiveParameter";
+const SENSITIVE_PARAMETER_LOWERCASE: &str = "sensitiveparameter";
+const ZEND_ATTRIBUTE_PERSISTENT: u32 = 1;
+const SUCCESS: i32 = 0;
+
+/// Class, method and zero-based position of every parameter that carries a
secret.
+// Every `raw` array is included, since this build cannot tell which unknown
keys are secret.
+const SENSITIVE_PARAMETERS: &[(&str, &str, u32)] = &[
Review Comment:
**nit**: the table leaves out `sendBinaryRequest(int $code, string
$payload)`. That raw payload carries LoginUser/CreateUser passwords and
CreateStream plaintext keys (`tests/IggySdkTest.php:88` sends CreateUser
through it), and with `zend.exception_ignore_args=Off` its first 15 bytes reach
traces.
Fix: add `("Iggy\\Client", "sendBinaryRequest", 1)` and a test row.
##########
core/server/src/stream_keys.rs:
##########
@@ -0,0 +1,355 @@
+// 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::wire::rewrite_request_body;
+use bytes::Bytes;
+use iggy_binary_protocol::requests::streams::CreateStreamRequest;
+use iggy_binary_protocol::{RoutedRequestHeader, WireEncode};
+use iggy_common::wire_conversions::resource_options_to_wire;
+use iggy_common::{
+ HeaderKey, HeaderValue, IggyError, OptionKey, OptionValue,
OptionsProvenance, ResourceOptions,
+ ServerKeyContext, StreamEncryptionKey, StreamKeyOrigin,
critical_keys_marker, erase_buffer,
+};
+use secrecy::zeroize::Zeroize;
+use server_common::Message;
+
+/// A request past static bounds. A `CreateStream` keeps its decoded body and
+/// admitted options for [`maybe_rewrite_secret_options_request`], so a client
+/// key is decoded once and never encoded into a request in the clear again.
+pub enum Bounded {
+ Request(Message<RoutedRequestHeader>),
+ StreamCreate {
+ request: Message<RoutedRequestHeader>,
+ create: CreateStreamRequest,
+ /// The admitted canonical options, a client key among them.
+ options: ResourceOptions,
+ },
+}
+
+/// Wrap a `CreateStream`'s key before the request leaves this node: the
+/// client's key when it sent one, or a key generated here when it did not and
+/// this node encrypts new streams. The key is wrapped under every configured
+/// server key, so the stream also opens on nodes a rotation has not reached.
+/// Only the wrap set and the critical marker travel on, so no plaintext key
Review Comment:
**nit**: "only the wrap set and the critical marker travel on" does not hold
for HTTP. A follower relays `POST /streams` with its raw body
(`core/server/src/http/forward.rs:465-501`) before any handler runs, so the
plaintext key reaches the primary and is wrapped there. The relay uses HTTPS
exactly when the node's own HTTP TLS is on and nothing logs the body, so it is
plaintext only behind a TLS-terminating proxy (as passwords already are).
Fix: correct the comment, or wrap the key on the follower before relaying.
##########
core/common/src/http/options.rs:
##########
@@ -0,0 +1,676 @@
+// 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::types::message::{HeaderKey, HeaderKind, HeaderValue};
+use crate::types::options::{
+ OptionKey, OptionValue, OptionsPatch, OptionsProvenance, ResourceOptions,
erase_options,
+ key_preview, wire_reset_names,
+};
+use crate::wire_conversions::resource_options_to_wire;
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD;
+use iggy_binary_protocol::WireOptions;
+use secrecy::zeroize::Zeroize;
+use serde::de::{Error as _, MapAccess, Visitor};
+use serde::ser::{Error as _, SerializeMap};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use std::collections::{BTreeMap, BTreeSet};
+use std::fmt::{Debug, Formatter};
+use std::str::FromStr;
+
+/// The update body member that lists the keys to reset. No option is ever
+/// named `reset`.
+pub const RESET_FIELD: &str = "reset";
+
+/// The largest integer a JSON number carries exactly in every client, 2^53 -
1.
+const MAX_EXACT_JSON_INTEGER: u64 = (1 << 53) - 1;
+
+/// One option value in an HTTP body: a JSON boolean, a JSON integer, or a
+/// string in the same form a config file uses (`"7 days"`, `"1 GiB"`). A
+/// 64-bit value above 2^53 travels as a decimal string so JavaScript clients
+/// keep its precision.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum OptionInputValue {
+ Bool(bool),
+ U64(u64),
+ String(String),
+}
+
+/// Option values for a create body, keyed by option name. A name repeated in
+/// the JSON object is refused rather than silently collapsed.
+#[derive(Clone, Default, PartialEq, Eq)]
+pub struct OptionInputMap(pub BTreeMap<String, OptionInputValue>);
+
+/// An HTTP update body: one flat object of option keys plus an optional
+/// `reset` array. A value sets the key, an absent key or `null` changes
+/// nothing, a key named in `reset` returns to its default. `null` never
resets,
+/// because many JSON serializers write unset optional fields as `null`. A body
+/// that changes nothing, with no keys, only nulls or only zero sentinels such
+/// as `"message_expiry": 0`, is refused with `InvalidCommand` on every
+/// transport.
+#[derive(Clone, Default, PartialEq, Eq)]
+pub struct OptionsUpdate {
+ pub set: BTreeMap<String, OptionInputValue>,
+ pub reset: BTreeSet<String>,
+}
+
+struct UniqueKeysVisitor;
+
+struct OptionsUpdateVisitor;
+
+impl OptionInputMap {
+ /// Encode the values into a create options block, each in the kind its
+ /// catalog entry expects where the JSON value fits it. Admission validates
+ /// the keys and values, so a mismatch fails in the same category as on
+ /// the binary transports. The map built on the way may hold the stream
+ /// key, so it is erased once encoded.
+ ///
+ /// # Errors
+ ///
+ /// `UnsupportedOptionKey` for a key that is not a valid option name.
+ pub fn to_wire(&self) -> Result<WireOptions, IggyError> {
+ let options = input_options(&self.0)?;
+ let wire = resource_options_to_wire(&options, OptionsProvenance::All);
+ erase_options(options);
+ wire
+ }
+
+ #[must_use]
+ pub fn is_empty(&self) -> bool {
+ self.0.is_empty()
+ }
+}
+
+impl OptionsUpdate {
+ /// Split the body into the keys to set and the keys to reset, in wire
+ /// form. Admission validates the keys and values.
+ ///
+ /// # Errors
+ ///
+ /// As [`OptionInputMap::to_wire`], and `UnsupportedOptionKey` for a reset
+ /// name that is not a valid option name or a reset list longer than the
+ /// wire carries.
+ pub fn to_patch(&self) -> Result<OptionsPatch, IggyError> {
+ let set = input_options(&self.set)?;
+ let patch =
wire_reset_names(self.reset.iter().map(String::as_str)).and_then(|reset| {
+ Ok(OptionsPatch {
+ set: resource_options_to_wire(&set, OptionsProvenance::All)?,
+ reset,
+ })
+ });
+ erase_options(set);
+ patch
+ }
+}
+
+impl OptionInputValue {
+ /// Zero text in place, since it may be the base64 stream key.
+ fn erase(&mut self) {
+ if let Self::String(text) = self {
+ text.as_mut_str().zeroize();
+ }
+ }
+}
+
+impl Drop for OptionInputMap {
Review Comment:
**nit**: `impl Drop` on the public `OptionInputMap` and `OptionsUpdate`
(re-exported from `lib.rs`, returned by `to_update_body`) turns field moves,
destructuring and struct-update syntax into E0509; the test at `:419` had to
change for it. The zeroing is partial anyway: values moved or cloned out escape
it.
Fix: zeroize at the value (e.g.
`OptionInputValue::String(Zeroizing<String>)`) and keep the containers plain
data.
##########
core/binary_protocol/src/consensus/operation.rs:
##########
@@ -64,14 +64,26 @@ pub enum Operation {
TruncatePartition = 68,
FinalizeSession = 70,
CompleteLifecycle = 71,
+ /// Prepare-resolved `UpdateResource` on a stream: column changes
extracted,
+ /// resets resolved to their defaults. Claimed by the streams state
machine.
+ UpdateStreamResolved = 72,
+ /// Prepare-resolved `UpdateResource` on a topic.
+ UpdateTopicResolved = 73,
+ /// Prepare-resolved `UpdateResource` on a user. Claimed by the users state
+ /// machine.
+ UpdateUserResolved = 74,
+ /// Server-originated: replace encrypted streams' wrap sets during a
+ /// key-encryption key rotation. Proposed by the metadata primary, no
+ /// client wire code.
+ RewrapStreamKeys = 75,
Review Comment:
**nit**: the thread reply says both labels were added, but the PR carries
only `breaking:client-to-replica-protocol`.
`breaking:replica-to-replica-protocol` applies too: consensus ops 72-75 and 150
are new here and 129/133/142 are retired.
Fix: add the label.
##########
foreign/node/src/wire/stream/create-stream.command.ts:
##########
@@ -28,28 +34,48 @@ export type CreateStream = {
streamId?: number,
/** Stream name (1-255 bytes) */
name: string
-};
+} & StreamOptions;
/**
* Create stream command definition.
* Creates a new stream with the specified name.
+ * Layout: `[name_len:u8][name]` followed by the options block running to the
+ * end of the payload.
*/
export const CREATE_STREAM = {
code: COMMAND_CODE.CreateStream,
- serialize: ({name}: CreateStream) => {
+ serialize: ({ name, encryptionKey, options: extraOptions = [] }:
CreateStream) => {
// Stream ID is now auto-assigned by the server, not sent in the protocol
const bName = Buffer.from(name);
if (bName.length < 1 || bName.length > 255)
throw new Error('Stream name should be between 1 and 255 bytes');
+ // A plain entry prints its value wherever the caller logs the arguments.
+ if (extraOptions.some(({ key }) => key === StreamOptionKey.EncryptionKey))
Review Comment:
**nit**: a raw `encryption_key` options entry is refused with 70
`InvalidEncryptionKey` even when the key is valid. The server admits it
(`core/common/src/types/options/catalog.rs:806-815`) and every other SDK's raw
path sends it (redacted in logs), so code 70 says "invalid key" for a valid one
and the escape hatch behaves differently in Node only.
Fix: refuse with a non-key category (`InvalidCommand`), or accept and redact
like the others.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -748,11 +768,54 @@ async fn reconcile_additions(
// Resolved only for namespaces actually built, not once per committed
// partition every pass. A stream, topic or partition that vanished
// between the target snapshot and this read defers to the next pass.
- let Some((partition_stats, topic_runtime, partition_metadata)) =
- fetch_partition_build_inputs(ctx, ns)
+ let Some(PartitionBuildInputs {
+ stats: partition_stats,
+ runtime: topic_runtime,
+ metadata: partition_metadata,
+ stream_key,
+ encryption,
+ }) = fetch_partition_build_inputs(ctx, ns)
else {
continue;
};
+ let topic_runtime = match topic_runtime {
+ Ok(topic_runtime) => topic_runtime,
+ Err(error) => {
+ ctx.record_failure(ns, FailureCause::Add, now);
+ ctx.shard.metrics().record_partition_reconcile_failure();
+ error!(
+ shard = shard_id,
+ stream_id = ns.stream_id(),
+ topic_id = ns.topic_id(),
+ partition_id = ns.partition_id(),
+ %error,
+ "Partition of topic {} cannot be built, its options hold
critical state this node cannot read",
+ ns.topic_id()
+ );
+ continue;
+ }
+ };
+ // An encrypted partition is built only with its own stream's key. A
+ // key that does not open here leaves the partition unbuilt, never
+ // served as plaintext or under another key.
+ let encryptor = match partitions.resolve_stream_encryptor(stream_key,
&encryption) {
+ Ok(encryptor) => encryptor,
+ Err(error) => {
+ ctx.record_failure(ns, FailureCause::Add, now);
Review Comment:
**nit**: a stream whose key does not open keeps this Add failure record, and
the fast-skip only arms with an empty `failure_state` (`:517-520`). So while
any stream stays deferred, every wake (at least once a second per shard, plus
each `MetadataCommitTick`) runs a full diff over all committed partitions under
the metadata guard.
Fix: arm the fast-skip when every remaining record is still backing off.
##########
core/server/src/http/extractor.rs:
##########
@@ -101,6 +131,87 @@ impl FromRequestParts<HttpState> for Authenticated {
}
}
+impl<T, S> FromRequest<S> for JsonMembers<T>
+where
+ T: DeserializeOwned,
+ S: Send + Sync,
+{
+ type Rejection = Response;
+
+ /// Runs axum's own `Json` extraction, so content type checks, body limits
+ /// and rejections stay what every other route answers.
+ async fn from_request(request: Request, state: &S) -> Result<Self,
Self::Rejection> {
+ let (parts, body) = request.into_parts();
+ let bytes = Bytes::from_request(Request::from_parts(parts.clone(),
body), state)
Review Comment:
**nit**: some buffers holding the plaintext key still drop without zeroing:
the `JsonMembers` body `Bytes` plus its clone into `Body` here (and
`JsonUpdate` at `:168-176`), the HTTP follower relay body kept for retries, and
the Rust SDK request bytes
(`core/common/src/traits/binary_impls/streams.rs:73-80`,
`core/sdk/src/http/streams.rs:64-71`). Best effort, since hyper/TLS buffers
keep copies too.
Fix: zeroize these buffers once the request is built, or document which
copies are not erased.
##########
core/simulator/src/workload/ops/update_user.rs:
##########
@@ -51,17 +54,22 @@ pub fn sample(
Outcome::Ok => {
let user = shadow.pick_user_name(prng)?;
let current_password = shadow.password_for(&user)?.to_string();
+ let status = if prng.random_bool(0.5) {
Review Comment:
**nit**: deactivation is simulated now, but three paths still are not:
`CreateStream` with a key, `RewrapStreamKeys`, and deactivation -> session end
-> client-table exemption (simulator clients log in as root only). The RNG
reason does not hold for raw mode, which can send fixed precomputed wrap-set
bytes and rewrap entries (tests already build seeded keys in
`stream_key_rewrapper.rs`).
Fix: add those ops in raw mode.
##########
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:
**nit**: a deferred namespace is neither materialized nor tombstoned. When
catch-up then commits `DeleteStream`/`DeleteTopic`, `reconcile_removals`
(`partition_reconciler.rs:1016-1046`) finds no owned or fenced ghost, so the
partition directory (ciphertext) and its initialization record stay on disk
until a same-id recreate wipes them (probe-confirmed). The boot
offset-reservation skip and reconciler load errors leak the same way.
Fix: have the removals pass tear down on-disk dirs of deleted namespaces
this shard owns but never built.
##########
.github/actions/utils/server-start/action.yml:
##########
@@ -53,6 +53,14 @@ inputs:
description: "--replica-id <u8> value (leave empty to omit)"
required: false
default: ""
+ encryption-enabled:
+ description: "Give streams created without their own key a generated key,
which the SDK server-generated key tests need"
+ required: false
+ default: "false"
Review Comment:
**nit**: the default is false now, but every main lane opts back in
(cpp-bazel, go x4, java, node, php, python, swift x4, coverage-baseline x4),
and this action exports `IGGY_ENCRYPTION_KEY` whenever `encryption-key` is
non-empty, which it is by default. So no foreign-SDK lane runs the shipped
default (`key = ""`, client keys refused), and Go, Swift, C++ and Python have
no plaintext lane.
Fix: give each SDK one lane on the shipped default.
##########
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:
**nit**: with three unretired envelopes and none under the current KEK,
`WrapSet::new` here fails as
`UnsupportedCriticalOption("encryption_key_wrapped")`
(`unsupported_wrapped_key`), and the rewrapper logs it as "needs a key rewrap
this node cannot make". That reads as corrupt state and gives no hint to set
`retired_kek_fingerprints`, and the 3-wrap cap is not documented. Reachable
when `previous_keys` is emptied without retiring over two rotations.
Fix: a distinct error and log ("wrap set full, retire a KEK fingerprint"),
and document the cap with the rotation steps.
--
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]
