This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new f318b79a7 fix(server-ng): add messages encryption (#3644)
f318b79a7 is described below
commit f318b79a7632ed0e31ee71be9246b0bb0938af52
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Fri Jul 10 13:56:46 2026 +0200
fix(server-ng): add messages encryption (#3644)
This PR adds support for `messages` encryption in `server-ng` binary.
---
.../tests/server/scenarios/encryption_scenario.rs | 23 +++++++-
core/integration/tests/server/scenarios/mod.rs | 1 -
core/partitions/src/iggy_partitions.rs | 29 +++++++++-
core/partitions/src/types.rs | 10 +++-
core/server-ng/src/bootstrap.rs | 13 ++++-
core/server-ng/src/dispatch.rs | 23 +++++---
core/server-ng/src/http/handlers.rs | 11 +++-
core/server-ng/src/http/wire.rs | 10 +++-
core/server-ng/src/partition_reconciler.rs | 1 +
core/server-ng/src/responses.rs | 67 +++++++++++++++++-----
core/server_common/src/send_messages2.rs | 61 +++++++++++++++++++-
core/simulator/src/replica.rs | 1 +
12 files changed, 213 insertions(+), 37 deletions(-)
diff --git a/core/integration/tests/server/scenarios/encryption_scenario.rs
b/core/integration/tests/server/scenarios/encryption_scenario.rs
index 3405b1b02..79c1c2e99 100644
--- a/core/integration/tests/server/scenarios/encryption_scenario.rs
+++ b/core/integration/tests/server/scenarios/encryption_scenario.rs
@@ -93,6 +93,10 @@ async fn
should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.await
.unwrap();
+ // server-ng has no flush primitive (FLUSH_UNSAVED_BUFFER denies typed);
+ // the eager-flush envs in `build_server_config` make every committed
+ // batch hit disk instead.
+ #[cfg(not(feature = "vsr"))]
client
.flush_unsaved_buffer(
&Identifier::named(stream_name).unwrap(),
@@ -103,7 +107,7 @@ async fn
should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.await
.unwrap();
- tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+ tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
// Verify on-disk encryption of headers and payload
let data_path = harness.server().data_path();
@@ -247,6 +251,10 @@ async fn
should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.await
.unwrap();
+ // server-ng has no flush primitive (FLUSH_UNSAVED_BUFFER denies typed);
+ // the eager-flush envs in `build_server_config` make every committed
+ // batch hit disk instead.
+ #[cfg(not(feature = "vsr"))]
client
.flush_unsaved_buffer(
&Identifier::named(stream_name).unwrap(),
@@ -257,7 +265,7 @@ async fn
should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.await
.unwrap();
- tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+ tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
let polled = client
.poll_messages(
@@ -490,6 +498,17 @@ fn encryption_disabled() -> bool {
fn build_server_config(encryption: bool) -> TestServerConfig {
let mut extra_envs = HashMap::new();
+ // server-ng flushes on the journal thresholds (no flush primitive), so
+ // force every committed batch straight to disk for the on-disk asserts.
+ extra_envs.insert(
+ "IGGY_SYSTEM_PARTITION_MESSAGES_REQUIRED_TO_SAVE".to_string(),
+ "1".to_string(),
+ );
+ extra_envs.insert(
+ "IGGY_SYSTEM_PARTITION_ENFORCE_FSYNC".to_string(),
+ "true".to_string(),
+ );
+
if encryption {
extra_envs.insert(
"IGGY_SYSTEM_ENCRYPTION_ENABLED".to_string(),
diff --git a/core/integration/tests/server/scenarios/mod.rs
b/core/integration/tests/server/scenarios/mod.rs
index b0ad38133..d69f61c90 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -34,7 +34,6 @@ pub mod create_message_payload;
// shard-0 HTTP listener and the create/delete commit through the metadata STM,
// so the token replicates to every shard a TCP client may land on.
pub mod cross_protocol_pat_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod encryption_scenario;
pub mod invalid_consumer_offset_scenario;
// Asserts server log-file rotation/archival policies; server-ng's file
diff --git a/core/partitions/src/iggy_partitions.rs
b/core/partitions/src/iggy_partitions.rs
index 069918489..38955759b 100644
--- a/core/partitions/src/iggy_partitions.rs
+++ b/core/partitions/src/iggy_partitions.rs
@@ -23,9 +23,10 @@ use crate::{IggyPartition, Partition, PollingArgs,
PollingConsumer};
use ahash::AHashSet;
use consensus::{Consensus, Plane, PlaneIdentity, VsrConsensus};
use iggy_binary_protocol::{
- Command2, ConsensusHeader, PrepareHeader, PrepareOkHeader, RequestHeader,
+ Command2, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader,
RequestHeader,
};
use message_bus::MessageBus;
+use server_common::send_messages2::{convert_request_message,
encrypt_batch_request};
use server_common::sharding::{IggyNamespace, LocalIdx, ShardId};
#[cfg(debug_assertions)]
use std::cell::Cell;
@@ -468,6 +469,32 @@ where
);
return;
}
+ // At-rest encryption happens HERE, once, before the op enters
+ // consensus: canonicalize the wire form first so both the legacy and
+ // v2 request encodings encrypt identically, then encrypt payload +
+ // user headers per message. The ciphertext is what gets journaled,
+ // replicated, checksummed, and persisted -- every replica stores
+ // identical bytes -- and the poll reply is the single decrypt point.
+ let message = if message.header().operation == Operation::SendMessages
+ && let Some(encryptor) = &self.config().encryptor
+ {
+ let canonical = convert_request_message(namespace, message)
+ .and_then(|message| encrypt_batch_request(message, encryptor));
+ match canonical {
+ Ok(message) => message,
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ namespace_raw = namespace.inner(),
+ %error,
+ "dropping send_messages: failed to encrypt batch at
ingestion"
+ );
+ return;
+ }
+ }
+ } else {
+ message
+ };
let Some(partition) = self.get_mut_by_ns(&namespace) else {
warn!(
target: "iggy.partitions.diag",
diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs
index 9d599488d..3f2caa8a9 100644
--- a/core/partitions/src/types.rs
+++ b/core/partitions/src/types.rs
@@ -15,9 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_common::{IggyByteSize, PollingStrategy};
+use iggy_common::{EncryptorKind, IggyByteSize, PollingStrategy};
use server_common::iobuf::Frozen;
use smallvec::SmallVec;
+use std::sync::Arc;
#[derive(Debug, Clone)]
pub struct Fragment<const ALIGN: usize = 4096> {
@@ -216,6 +217,13 @@ pub struct PartitionsConfig {
pub enforce_fsync: bool,
/// Maximum size of a single segment before rotation.
pub segment_size: IggyByteSize,
+ /// Server-side at-rest encryption. Applied ONCE, on the primary at
+ /// ingestion, so the ciphertext replicates verbatim: every replica
+ /// journals, acks, and persists identical bytes (checksums and the
+ /// deterministic segment rolls both depend on that), and the poll path
+ /// decrypts uniformly whether a fragment came from the resident journal
+ /// or from disk.
+ pub encryptor: Option<Arc<EncryptorKind>>,
}
impl PartitionsConfig {
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 65e41d447..ab2fd4c1d 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -46,7 +46,7 @@ use iggy_common::defaults::{
DEFAULT_ROOT_USERNAME, MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH,
MIN_PASSWORD_LENGTH,
MIN_USERNAME_LENGTH,
};
-use iggy_common::{IggyByteSize, PartitionStats, variadic};
+use iggy_common::{Aes256GcmEncryptor, EncryptorKind, IggyByteSize,
PartitionStats, variadic};
use journal::Journal;
use journal::prepare_journal::PrepareJournal;
use message_bus::client_listener::{self, RequestHandler};
@@ -1414,6 +1414,16 @@ async fn build_shard_for_thread(
let owned_partitions_capacity = total_partitions
.div_ceil(usize::from(total_shards).max(1))
.saturating_mul(2);
+ // At-rest encryption: built once per shard from the shared config; the
+ // ingestion path encrypts on the primary and the poll reply decrypts.
+ // A bad key fails the boot rather than silently serving plaintext.
+ let encryptor = if config.system.encryption.enabled {
+ let aes =
Aes256GcmEncryptor::from_base64_key(&config.system.encryption.key)
+ .map_err(|error| ServerNgError::Iggy(Box::new(error)))?;
+ Some(Arc::new(EncryptorKind::Aes256Gcm(aes)))
+ } else {
+ None
+ };
let partitions = IggyPartitions::with_capacity(
shard_local_id,
PartitionsConfig {
@@ -1424,6 +1434,7 @@ async fn build_shard_for_thread(
.size_of_messages_required_to_save,
enforce_fsync: config.system.partition.enforce_fsync,
segment_size: config.system.segment.size,
+ encryptor,
},
owned_partitions_capacity,
);
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 2ca621933..88e59d831 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -1512,15 +1512,20 @@ async fn handle_poll_messages(
Some(PartitionReadReply::Poll {
fragments,
current_offset,
- }) => build_polled_messages_body(partition_id, current_offset,
fragments)
- .unwrap_or_else(|error| {
- warn!(
- transport_client_id,
- error = %error,
- "failed to re-encode polled batches; replying
empty poll"
- );
- empty_polled_messages_body(partition_id)
- }),
+ }) => build_polled_messages_body(
+ partition_id,
+ current_offset,
+ fragments,
+ shard.plane.partitions().config().encryptor.as_deref(),
+ )
+ .unwrap_or_else(|error| {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to re-encode polled batches; replying empty
poll"
+ );
+ empty_polled_messages_body(partition_id)
+ }),
other => {
warn!(
transport_client_id,
diff --git a/core/server-ng/src/http/handlers.rs
b/core/server-ng/src/http/handlers.rs
index a796ff7c2..6ca70039d 100644
--- a/core/server-ng/src/http/handlers.rs
+++ b/core/server-ng/src/http/handlers.rs
@@ -27,7 +27,7 @@ use axum::extract::{Path, Query, State};
use axum::http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header};
use axum::response::{IntoResponse, Response};
use chrono::Local;
-use consensus::MetadataHandle;
+use consensus::{MetadataHandle, PartitionsHandle};
use iggy_binary_protocol::codes::{
GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE,
GET_PERSONAL_ACCESS_TOKENS_CODE,
GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE,
GET_TOPICS_CODE,
@@ -1033,8 +1033,13 @@ pub(in crate::http) async fn poll_messages(
fragments,
current_offset,
}) => {
- let body = build_polled_messages_body(partition_id,
current_offset, fragments)
- .map_err(ReadError::Rejected)?;
+ let body = build_polled_messages_body(
+ partition_id,
+ current_offset,
+ fragments,
+ state.shard.plane.partitions().config().encryptor.as_deref(),
+ )
+ .map_err(ReadError::Rejected)?;
Ok(Json(
PolledMessages::from_bytes(body).map_err(ReadError::Rejected)?,
))
diff --git a/core/server-ng/src/http/wire.rs b/core/server-ng/src/http/wire.rs
index 69b65b741..e96866f80 100644
--- a/core/server-ng/src/http/wire.rs
+++ b/core/server-ng/src/http/wire.rs
@@ -509,9 +509,13 @@ mod tests {
header.base_offset = 41;
header.base_timestamp = 999_999;
- let body =
- build_polled_messages_body(3, 42,
fragment_from_stored_batch(&header, &stored.blob))
- .expect("re-encodes wire body");
+ let body = build_polled_messages_body(
+ 3,
+ 42,
+ fragment_from_stored_batch(&header, &stored.blob),
+ None,
+ )
+ .expect("re-encodes wire body");
let polled = PolledMessages::from_bytes(body).expect("decodes as the
SDK does");
assert_eq!(polled.partition_id, 3);
diff --git a/core/server-ng/src/partition_reconciler.rs
b/core/server-ng/src/partition_reconciler.rs
index 37d3f5d3b..5e8b71bc7 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -1054,6 +1054,7 @@ mod tests {
size_of_messages_required_to_save:
iggy_common::IggyByteSize::from(1024_u64),
enforce_fsync: false,
segment_size: config.system.segment.size,
+ encryptor: None,
},
);
let shards_table = PapayaShardsTable::new();
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index 11e111708..1fd86e0a7 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -77,7 +77,7 @@ use iggy_binary_protocol::{
Command2, GenericHeader, IGGY_PROTOCOL_VERSION, KIND_CONSUMER_GROUP,
Operation, ReplyHeader,
RequestHeader, WireDecode, WireEncode, WireIdentifier, WireName,
WirePartitioning,
};
-use iggy_common::{Identifier, IggyError, IggyTimestamp, MaxTopicSize};
+use iggy_common::{EncryptorKind, Identifier, IggyError, IggyTimestamp,
MaxTopicSize};
use metadata::impls::metadata::StreamsFrontend;
use partitions::PollFragments;
use server_common::Message;
@@ -1307,6 +1307,7 @@ pub(crate) fn build_polled_messages_body(
partition_id: u32,
current_offset: u64,
fragments: PollFragments,
+ encryptor: Option<&EncryptorKind>,
) -> Result<Bytes, IggyError> {
// Body head: [partition_id:4][current_offset:8][count:4]. `count` sits at
// COUNT_OFFSET and is backpatched once the walk below knows it.
@@ -1374,20 +1375,56 @@ pub(crate) fn build_polled_messages_body(
body.extend_from_slice(&offset.to_le_bytes());
body.extend_from_slice(×tamp.to_le_bytes());
body.extend_from_slice(&origin_timestamp.to_le_bytes());
- body.extend_from_slice(
- &u32::try_from(user_headers_length)
- .expect("length came from u32")
- .to_le_bytes(),
- );
- body.extend_from_slice(
- &u32::try_from(payload_length)
- .expect("length came from u32")
- .to_le_bytes(),
- );
- body.extend_from_slice(&0u64.to_le_bytes()); // reserved
- // Stored sections are already in legacy order
- // (`[payload][user_headers]`): copy through contiguously.
- body.extend_from_slice(&stream[sections_start..sections_end]);
+ if let Some(encryptor) = encryptor {
+ // At-rest encryption: stored sections are ciphertext
(encrypted
+ // once at ingestion, replicated verbatim); this reply is the
+ // single decrypt point, so lengths are rewritten to the
+ // plaintext sizes. The stored per-message checksum still
covers
+ // the ciphertext and is passed through untouched (the SDK does
+ // not re-validate it against the reply layout).
+ let payload_end = sections_start + payload_length;
+ let payload = encryptor
+ .decrypt(&stream[sections_start..payload_end])
+ .map_err(|_| IggyError::CannotDecryptData)?;
+ let user_headers = if user_headers_length > 0 {
+ Some(
+ encryptor
+ .decrypt(&stream[payload_end..sections_end])
+ .map_err(|_| IggyError::CannotDecryptData)?,
+ )
+ } else {
+ None
+ };
+ let user_headers_bytes: &[u8] =
user_headers.as_deref().unwrap_or_default();
+ body.extend_from_slice(
+ &u32::try_from(user_headers_bytes.len())
+ .map_err(|_| IggyError::InvalidCommand)?
+ .to_le_bytes(),
+ );
+ body.extend_from_slice(
+ &u32::try_from(payload.len())
+ .map_err(|_| IggyError::InvalidCommand)?
+ .to_le_bytes(),
+ );
+ body.extend_from_slice(&0u64.to_le_bytes()); // reserved
+ body.extend_from_slice(&payload);
+ body.extend_from_slice(user_headers_bytes);
+ } else {
+ body.extend_from_slice(
+ &u32::try_from(user_headers_length)
+ .expect("length came from u32")
+ .to_le_bytes(),
+ );
+ body.extend_from_slice(
+ &u32::try_from(payload_length)
+ .expect("length came from u32")
+ .to_le_bytes(),
+ );
+ body.extend_from_slice(&0u64.to_le_bytes()); // reserved
+ // Stored sections are already in legacy order
+ // (`[payload][user_headers]`): copy through contiguously.
+ body.extend_from_slice(&stream[sections_start..sections_end]);
+ }
count += 1;
cursor = sections_end;
diff --git a/core/server_common/src/send_messages2.rs
b/core/server_common/src/send_messages2.rs
index 71321c726..a64cd0664 100644
--- a/core/server_common/src/send_messages2.rs
+++ b/core/server_common/src/send_messages2.rs
@@ -20,7 +20,7 @@ use crate::iobuf::Owned;
use crate::sharding::IggyNamespace;
use bytes::{Bytes, BytesMut};
use iggy_binary_protocol::{PrepareHeader, RequestHeader};
-use iggy_common::{INDEX_SIZE, IggyError, random_id};
+use iggy_common::{EncryptorKind, INDEX_SIZE, IggyError, random_id};
use std::hash::Hasher;
use twox_hash::XxHash3_64;
@@ -531,6 +531,65 @@ impl<'a> Iterator for SendMessages2IteratorWithOffsets<'a>
{
pub(crate) type FrozenBatchHeader = crate::iobuf::Frozen<MESSAGE_ALIGN>;
+/// Re-encode a canonical `SendMessages2` request with every message's payload
+/// and user headers encrypted, per-message checksums and lengths recomputed,
+/// and the batch header (length + checksum) restamped.
+///
+/// Runs ONCE, on the primary at ingestion (after [`convert_request_message`]
+/// canonicalized the wire form), so the ciphertext is what replicates: every
+/// replica journals and persists identical bytes, and the poll path decrypts
+/// uniformly regardless of which replica or tier served the fragment.
+///
+/// # Errors
+///
+/// [`IggyError::InvalidCommand`] on an undecodable batch; encryption errors
+/// propagate from the encryptor.
+pub fn encrypt_batch_request(
+ message: Message<RequestHeader>,
+ encryptor: &EncryptorKind,
+) -> Result<Message<RequestHeader>, IggyError> {
+ let request_header = *message.header();
+ let total_size = request_header.size as usize;
+ let body =
&message.as_slice()[std::mem::size_of::<RequestHeader>()..total_size];
+ let batch = decode_batch_slice(body)?;
+
+ let mut blob = BytesMut::with_capacity(batch.blob().len() * 2);
+ for view in batch.iter() {
+ let encrypted_payload = encryptor.encrypt(view.payload)?;
+ let encrypted_user_headers = if view.user_headers.is_empty() {
+ None
+ } else {
+ Some(encryptor.encrypt(view.user_headers)?)
+ };
+ let user_headers: &[u8] =
encrypted_user_headers.as_deref().unwrap_or_default();
+ let payload_length =
+ u32::try_from(encrypted_payload.len()).map_err(|_|
IggyError::InvalidCommand)?;
+ let user_headers_length =
+ u32::try_from(user_headers.len()).map_err(|_|
IggyError::InvalidCommand)?;
+
+ let mut header = [0u8; MESSAGE_HEADER_SIZE];
+ header[8..24].copy_from_slice(&view.header.id.to_le_bytes());
+
header[24..28].copy_from_slice(&view.header.offset_delta.to_le_bytes());
+
header[28..32].copy_from_slice(&view.header.timestamp_delta.to_le_bytes());
+ header[32..36].copy_from_slice(&user_headers_length.to_le_bytes());
+ header[36..40].copy_from_slice(&payload_length.to_le_bytes());
+ let checksum = calculate_checksum_parts(&header[8..],
&encrypted_payload, user_headers);
+ header[0..8].copy_from_slice(&checksum.to_le_bytes());
+
+ blob.extend_from_slice(&header);
+ blob.extend_from_slice(&encrypted_payload);
+ blob.extend_from_slice(user_headers);
+ }
+
+ let blob = blob.freeze();
+ let mut header = batch.header;
+ header.batch_length =
+ u64::try_from(COMMAND_HEADER_SIZE + blob.len()).map_err(|_|
IggyError::InvalidCommand)?;
+ header.batch_checksum = calculate_batch_checksum(&header, &blob);
+
+ SendMessages2Owned { header, blob }.encode_request(request_header)
+}
+
pub fn convert_request_message(
namespace: IggyNamespace,
message: Message<RequestHeader>,
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index aea179ac1..ea243a88f 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -78,6 +78,7 @@ pub fn new_replica(id: u8, name: String, bus:
&Arc<SimOutbox>, replica_count: u8
size_of_messages_required_to_save: IggyByteSize::from(4 * 1024 * 1024),
enforce_fsync: false, //Disable fsync for simulation
segment_size: IggyByteSize::from(1024 * 1024 * 1024),
+ encryptor: None,
};
let partitions = IggyPartitions::new(ShardId::new(u16::from(id)),
partitions_config);
