This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch fix_encryption_2 in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 35c80e17c62899288c25c6ef26673440e53b58b7 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Fri Jul 10 10:30:21 2026 +0200 fix encryption --- .../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 6b6a0f2b7..32dd18b18 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, @@ -1030,8 +1030,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);
