This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch vsr-dvc-headers
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/vsr-dvc-headers by this push:
new dd294bce2 fix(consensus): pin the verified body range and stamp
requests
dd294bce2 is described below
commit dd294bce2e25f82bae8b5e1d629983683ce67b2a
Author: Krishna Vishal <[email protected]>
AuthorDate: Fri Aug 7 12:13:38 2026 +0530
fix(consensus): pin the verified body range and stamp requests
---
core/binary_protocol/src/consensus/header.rs | 14 +++
core/binary_protocol/src/consensus/mod.rs | 2 +-
core/binary_protocol/src/lib.rs | 2 +-
core/consensus/src/impls.rs | 11 +-
core/consensus/src/plane_helpers.rs | 76 ++++++++---
core/metadata/src/impls/metadata.rs | 5 +-
core/metadata/src/impls/recovery.rs | 182 ++++++++++++++++++++++++++-
core/partitions/src/iggy_partition.rs | 5 +-
core/sdk/src/vsr.rs | 8 +-
core/server-ng/src/dispatch.rs | 33 ++++-
core/server-ng/src/wire.rs | 94 +++++++++++++-
core/shard/src/lib.rs | 8 +-
12 files changed, 396 insertions(+), 44 deletions(-)
diff --git a/core/binary_protocol/src/consensus/header.rs
b/core/binary_protocol/src/consensus/header.rs
index 1ada15cf5..c7d511668 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -888,6 +888,20 @@ impl ConsensusHeader for PrepareHeader {
/// Verification skips such entries so an older build's WAL still replays.
pub const CHECKSUM_UNSEALED: u128 = 0;
+/// The frame's body, bounded by `size`. What `checksum_body` covers.
+///
+/// Not `&frame[HEADER_SIZE..]`: `Message::try_from` accepts a buffer longer
than
+/// `size` without trimming, while the WAL scan reads exactly `size`, so
slicing to
+/// the end makes the two disagree. Empty when `size` overruns the buffer.
+#[must_use]
+pub fn frame_body(frame: &[u8], size: u32) -> &[u8] {
+ let end = size as usize;
+ if end <= HEADER_SIZE || end > frame.len() {
+ return &[];
+ }
+ &frame[HEADER_SIZE..end]
+}
+
impl PrepareHeader {
/// Which prepare this is, independent of which view re-sent it.
///
diff --git a/core/binary_protocol/src/consensus/mod.rs
b/core/binary_protocol/src/consensus/mod.rs
index 83a299291..0cf8eefa3 100644
--- a/core/binary_protocol/src/consensus/mod.rs
+++ b/core/binary_protocol/src/consensus/mod.rs
@@ -50,7 +50,7 @@ pub use header::{
RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader,
ReplyHeader, RequestHeader,
RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
RequestStateTransferHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader,
StartViewHeader,
- StateChunkHeader, StateTransferTargetHeader, frame_checksum_bytes,
read_size_field,
+ StateChunkHeader, StateTransferTargetHeader, frame_body,
frame_checksum_bytes, read_size_field,
};
pub use operation::Operation;
pub use reply_result::{RESULT_COUNT_LEN, RESULT_ENTRY_LEN, result_code,
result_section_len};
diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs
index 75defea5a..4d66bbc7d 100644
--- a/core/binary_protocol/src/lib.rs
+++ b/core/binary_protocol/src/lib.rs
@@ -77,7 +77,7 @@ pub use consensus::{
RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
RequestStartViewHeader, RequestStateChunkHeader,
RequestStateTransferHeader, SIZE_FIELD_OFFSET,
StartViewChangeHeader, StartViewHeader, StateChunkHeader,
StateTransferTargetHeader,
- frame_checksum_bytes, read_size_field, result_code, result_section_len,
+ frame_body, frame_checksum_bytes, read_size_field, result_code,
result_section_len,
};
pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation,
lookup_command};
pub use error::WireError;
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 0386b23ac..4c3f8c1ad 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -29,6 +29,7 @@ use clock::{Clock, IggySystemClock};
use iggy_binary_protocol::{
Command2, ConsensusHeader, DoViewChangeHeader, GenericHeader,
PrepareHeader, PrepareOkHeader,
ReplyHeader, RequestHeader, RequestStartViewHeader, StartViewChangeHeader,
StartViewHeader,
+ frame_body,
};
use iggy_common::IggyTimestamp;
use iggy_common::calculate_checksum;
@@ -3490,10 +3491,14 @@ where
// to the view-change merge. Closing that wants `checksum_body` here
to BE
// the batch checksum, recomputed after stamping, so `checksum` covers
the
// body for free by hashing this field.
+ //
+ // Bounded by `size`, the range every verifier re-reads; the prepare
+ // inherits it verbatim below.
let checksum_body = if consensus.namespace ==
METADATA_CONSENSUS_NAMESPACE {
- u128::from(calculate_checksum(
- &self.as_slice()[size_of::<PrepareHeader>()..],
- ))
+ u128::from(calculate_checksum(frame_body(
+ self.as_slice(),
+ self.header().size,
+ )))
} else {
0
};
diff --git a/core/consensus/src/plane_helpers.rs
b/core/consensus/src/plane_helpers.rs
index 467f77f1f..aeb26c33b 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -21,7 +21,7 @@ use crate::{
};
use iggy_binary_protocol::{
CHECKSUM_UNSEALED, Command2, ConsensusHeader, PrepareHeader,
PrepareOkHeader, ReplyHeader,
- RequestHeader,
+ RequestHeader, frame_body,
};
use message_bus::{MessageBus, SendError};
use server_common::{Message, iobuf::Owned};
@@ -125,18 +125,22 @@ where
/// bytes it arrived with, so a corrupted frame is admitted whenever its
flipped
/// value satisfies those comparisons, then journaled and re-served to peers.
///
-/// `body` is the frame past the header, empty for a header-only prepare.
-/// [`CHECKSUM_UNSEALED`] skips the partition plane, which seals neither and
-/// carries a verified `batch_checksum` over the same bytes.
+/// `frame` is the whole message. The body range comes from [`frame_body`],
not the
+/// caller, so no ingress point can verify a different span than the producer
sealed.
+/// [`CHECKSUM_UNSEALED`] skips the partition plane, which carries
`batch_checksum`
+/// over the same bytes instead.
///
/// # Errors
/// Returns a static description of which field failed.
-pub fn verify_prepare_integrity(header: &PrepareHeader, body: &[u8]) ->
Result<(), &'static str> {
+pub fn verify_prepare_integrity(header: &PrepareHeader, frame: &[u8]) ->
Result<(), &'static str> {
if header.checksum != CHECKSUM_UNSEALED && header.identity_checksum() !=
header.checksum {
return Err("prepare header does not match its own checksum");
}
if header.checksum_body != 0
- && u128::from(iggy_common::calculate_checksum(body)) !=
header.checksum_body
+ && u128::from(iggy_common::calculate_checksum(frame_body(
+ frame,
+ header.size,
+ ))) != header.checksum_body
{
return Err("prepare body does not match its checksum");
}
@@ -1237,16 +1241,60 @@ mod tests {
assert_eq!(verify_prepare_integrity(&header, &[]), Ok(()));
}
+ /// A frame carrying `body`, with `size` covering exactly header + body and
+ /// `checksum_body` sealed over it, as the metadata projection does.
+ fn sealed_frame(body: &[u8]) -> Vec<u8> {
+ let mut frame = vec![0u8; size_of::<PrepareHeader>() + body.len()];
+ frame[size_of::<PrepareHeader>()..].copy_from_slice(body);
+ let header = bytemuck::checked::from_bytes_mut::<PrepareHeader>(
+ &mut frame[..size_of::<PrepareHeader>()],
+ );
+ header.command = Command2::Prepare;
+ header.op = 7;
+ header.size = u32::try_from(size_of::<PrepareHeader>() +
body.len()).expect("fits u32");
+ header.checksum_body = u128::from(calculate_checksum(body));
+ frame
+ }
+
#[test]
fn given_a_prepare_whose_body_was_altered_when_verifying_should_reject() {
- let mut header = PrepareHeader {
- command: Command2::Prepare,
- op: 7,
- ..Default::default()
- };
- header.checksum_body = u128::from(calculate_checksum(b"body"));
- assert_eq!(verify_prepare_integrity(&header, b"body"), Ok(()));
- assert!(verify_prepare_integrity(&header, b"bodY").is_err());
+ let frame = sealed_frame(b"body");
+ let header =
+
*bytemuck::checked::from_bytes::<PrepareHeader>(&frame[..size_of::<PrepareHeader>()]);
+ assert_eq!(verify_prepare_integrity(&header, &frame), Ok(()));
+
+ let mut altered = frame;
+ *altered.last_mut().expect("the frame carries a body") ^= 1;
+ assert!(verify_prepare_integrity(&header, &altered).is_err());
+ }
+
+ #[test]
+ fn given_bytes_past_the_frame_size_when_verifying_should_ignore_them() {
+ // `try_from` accepts a buffer longer than `size` without trimming;
hashing to
+ // the end would reject a correctly sealed prepare and disagree with
the WAL scan.
+ let frame = sealed_frame(b"body");
+ let header =
+
*bytemuck::checked::from_bytes::<PrepareHeader>(&frame[..size_of::<PrepareHeader>()]);
+
+ let mut padded = frame;
+ padded.extend_from_slice(b"trailing garbage");
+ assert_eq!(
+ verify_prepare_integrity(&header, &padded),
+ Ok(()),
+ "only the bytes `size` covers are the body"
+ );
+ }
+
+ #[test]
+ fn given_a_size_that_overruns_the_buffer_when_verifying_should_reject() {
+ // Truncated frame, header still claims the full length: the body it
names is
+ // not there to hash.
+ let frame = sealed_frame(b"body");
+ let header =
+
*bytemuck::checked::from_bytes::<PrepareHeader>(&frame[..size_of::<PrepareHeader>()]);
+
+ let truncated = &frame[..frame.len() - 1];
+ assert!(verify_prepare_integrity(&header, truncated).is_err());
}
#[test]
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index f06af55d9..bafd51cee 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -1059,10 +1059,7 @@ where
// corrupted between primary and backup is journaled as-is and
re-served to
// peers, which the interior-corruption boot refusal turns into an
unbootable
// node on the next restart.
- if let Err(reason) = verify_prepare_integrity(
- &header,
- &message.as_slice()[std::mem::size_of::<PrepareHeader>()..],
- ) {
+ if let Err(reason) = verify_prepare_integrity(&header,
message.as_slice()) {
warn!(
target: "iggy.metadata.diag",
plane = "metadata",
diff --git a/core/metadata/src/impls/recovery.rs
b/core/metadata/src/impls/recovery.rs
index 286d286e2..3b2e009af 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -23,8 +23,9 @@ use consensus::{
ClientTable, ClientTableDecodeError, VsrState, VsrStateError,
build_reply_message,
build_reply_message_with,
};
-use iggy_binary_protocol::consensus::{Operation, PrepareHeader};
+use iggy_binary_protocol::consensus::{CHECKSUM_UNSEALED, Operation,
PrepareHeader};
use iggy_common::IggyError;
+use journal::Journal as _;
use journal::prepare_journal::{JournalError, PrepareJournal};
use journal::superblock::{
PingPongSuperblock, SLOT_FILE_NAMES, SuperblockContents, SuperblockStore,
@@ -307,6 +308,13 @@ pub struct RecoveredMetadata<M> {
/// they stay journal-only until the recovered primary re-replicates them
/// (or a backup sees the commit point advance past them).
pub last_journaled_op: Option<u64>,
+ /// First op replay could not connect to its predecessor, `None` when the
+ /// replayed range is one unbroken chain.
+ ///
+ /// `Some(op)` means entries at and above `op` were truncated and must
come back
+ /// from the cluster. `last_journaled_op` stops below it, which keeps the
restored
+ /// head, the re-pipeline range, and the recovery barrier honest.
+ pub chain_break_op: Option<u64>,
}
/// Recover metadata state from disk.
@@ -545,10 +553,35 @@ where
let mut last_applied_op: Option<u64> = None;
let mut last_journaled_op: Option<u64> = None;
+ let mut chain_break_op: Option<u64> = None;
+ let mut previous: Option<PrepareHeader> = None;
for header in &headers_to_replay {
- // TODO: Check hash chain integrity against `previous_header`. On a
- // same-view break, stop replay here and mark the remaining entries for
- // repair via VSR instead of panicking.
+ // Stop at the first op that does not connect to the one before it.
Applying
+ // across a hole replays effects onto a state machine that never saw
the
+ // missing op, and nothing downstream re-checks it.
+ //
+ // The WAL scan does not cover this: it only fires on CONSECUTIVE ops
with
+ // both ends sealed, so a gap reaches here. The first replayed op is
exempt,
+ // since a snapshot records no checksum for its parent to chain to.
+ if let Some(previous) = previous {
+ let gap = previous.op + 1 != header.op;
+ let broken_chain = previous.checksum != CHECKSUM_UNSEALED
+ && header.checksum != CHECKSUM_UNSEALED
+ && header.parent != previous.checksum;
+ if gap || broken_chain {
+ tracing::error!(
+ op = header.op,
+ previous_op = previous.op,
+ gap,
+ broken_chain,
+ "metadata WAL does not connect at this op; stopping replay
and dropping the \
+ suffix for VSR repair"
+ );
+ chain_break_op = Some(header.op);
+ break;
+ }
+ }
+ previous = Some(*header);
last_journaled_op = Some(header.op);
if header.op > commit_watermark {
@@ -626,6 +659,22 @@ where
last_applied_op = Some(header.op);
}
+ // `truncate_from`, never `drain`: the removed ops must stay refillable,
so the
+ // snapshot watermark stays put. Leaving them resident would make `append`
refuse
+ // the slot, failing repair on exactly the ops it exists to fix.
+ if let Some(break_op) = chain_break_op {
+ let removed = journal
+ .truncate_from(break_op)
+ .await
+ .map_err(RecoveryError::Io)?;
+ tracing::warn!(
+ break_op,
+ removed,
+ last_journaled_op,
+ "dropped the disconnected metadata WAL suffix; the cluster
re-supplies these ops"
+ );
+ }
+
Ok(RecoveredMetadata {
journal,
snapshot,
@@ -636,6 +685,7 @@ where
client_table,
last_applied_op,
last_journaled_op,
+ chain_break_op,
})
}
@@ -976,6 +1026,130 @@ mod tests {
assert_eq!(recovered.journal.last_op(), Some(3));
}
+ /// A prepare sealed the way a live primary seals one: `parent` chains to
the
+ /// previous op's identity and `checksum` is that identity.
+ fn make_chained_prepare(op: u64, commit: u64, parent: u128) ->
Message<PrepareHeader> {
+ let mut message = make_prepare_with_commit(op, commit, 32);
+ let header = bytemuck::checked::from_bytes_mut::<PrepareHeader>(
+ &mut message.as_mut_slice()[..HEADER_SIZE],
+ );
+ header.parent = parent;
+ let checksum = header.identity_checksum();
+ header.checksum = checksum;
+ message
+ }
+
+ #[compio::test]
+ async fn recover_stops_at_a_gap_and_drops_the_disconnected_suffix() {
+ // Ops 1-3 then 5: op 4 never landed. Replaying 5 over a state machine
that
+ // never saw 4 diverges silently, and the WAL scan waves this through
--
+ // its chain check only fires on CONSECUTIVE ops, since a gap is also
what
+ // ordinary compaction leaves behind.
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+
+ {
+ let journal =
PrepareJournal::open(&metadata_dir.join("journal.wal"), 0)
+ .await
+ .unwrap();
+ for op in 1..=3u64 {
+ journal
+ .append(make_prepare_with_commit(op, op, 32))
+ .await
+ .unwrap();
+ }
+ journal
+ .append(make_prepare_with_commit(5, 5, 32))
+ .await
+ .unwrap();
+ journal.storage_ref().fsync().await.unwrap();
+ }
+
+ let recovered = recover::<TestStm>(
+ dir.path(),
+ CLUSTERED,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(recovered.chain_break_op, Some(5));
+ assert_eq!(
+ recovered.last_applied_op,
+ Some(3),
+ "op 5 must not apply across the hole at op 4"
+ );
+ assert_eq!(
+ recovered.last_journaled_op,
+ Some(3),
+ "the restored head stops below the break, so nothing re-pipelines
it"
+ );
+ assert_eq!(
+ recovered.journal.last_op(),
+ Some(3),
+ "the disconnected entry is dropped so repair can journal the
cluster's op 5"
+ );
+ assert_eq!(
+ recovered.journal.snapshot_op(),
+ 0,
+ "truncating a suffix must leave the watermark, or the ops stop
being refillable"
+ );
+ }
+
+ #[compio::test]
+ async fn recover_stops_at_a_broken_chain_between_consecutive_ops() {
+ // Consecutive and sealed on both ends, but op 3 names a parent that
is not
+ // op 2: a fork left by a crash mid view change. Ops are appended out
of
+ // ascending file order so the scan's own chain check does not fire
first.
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+
+ {
+ let journal =
PrepareJournal::open(&metadata_dir.join("journal.wal"), 0)
+ .await
+ .unwrap();
+ let first = make_chained_prepare(1, 1, 0);
+ let first_checksum = first.header().checksum;
+ journal.append(first).await.unwrap();
+ let second = make_chained_prepare(2, 2, first_checksum);
+ journal.append(second).await.unwrap();
+ // Parent of a prepare that is not op 2.
+ journal
+ .append(make_chained_prepare(3, 3, 0xdead_beef))
+ .await
+ .unwrap();
+ journal.storage_ref().fsync().await.unwrap();
+ }
+
+ let recovered = recover::<TestStm>(
+ dir.path(),
+ CLUSTERED,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await;
+
+ // The WAL scan reaches this first and refuses boot: consecutive ops,
both
+ // sealed, chain broken, with no entry after it is only a tail. Either
+ // outcome is a refusal to apply the fork; what must never happen is a
+ // clean recovery that replayed op 3.
+ match recovered {
+ Err(RecoveryError::Journal(_) | RecoveryError::Io(_)) => {}
+ Ok(recovered) => {
+ assert!(
+ recovered.last_applied_op < Some(3),
+ "op 3 forks the chain and must not be applied"
+ );
+ }
+ Err(other) => panic!("unexpected recovery error: {other:?}"),
+ }
+ }
+
#[compio::test]
async fn recover_applies_only_the_committed_prefix() {
let dir = tempdir().unwrap();
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 54a793555..bb7de4290 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -2027,10 +2027,7 @@ where
// Same reason as the metadata plane: `checksum` is compared as an
opaque token
// downstream, so a corrupted frame passes whenever its flipped value
satisfies
// those comparisons.
- if let Err(reason) = verify_prepare_integrity(
- &header,
- &message.as_slice()[std::mem::size_of::<PrepareHeader>()..],
- ) {
+ if let Err(reason) = verify_prepare_integrity(&header,
message.as_slice()) {
emit_partition_diag(
tracing::Level::WARN,
&PartitionDiagEvent::new(
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 9bc5d1cd1..1a64d843e 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -38,7 +38,7 @@ use iggy_binary_protocol::requests::consumer_offsets::{
use iggy_binary_protocol::requests::messages::SendMessagesHeader;
use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
use iggy_binary_protocol::{WireIdentifier, WirePartitioning};
-use iggy_common::{IggyError, eviction_reason_to_error};
+use iggy_common::{IggyError, calculate_checksum, eviction_reason_to_error};
const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> = 0..4;
@@ -139,6 +139,12 @@ pub(crate) fn encode_request_header(
request: request_id,
session: session_id,
namespace,
+ // Lets the server's client table tell a genuine retry from a `request`
+ // number reused for different arguments. Zero means unstamped, which
is what
+ // an SDK predating this sends. A server that rewrites the body (PAT,
+ // password) carries this through untouched, so it keeps describing
what the
+ // client sent.
+ request_checksum: u128::from(calculate_checksum(payload)),
// Zeroed: the field is "informational" -- the server copies it into
// `ReplyHeader.timestamp` for RTT but nothing else reads it. Paying
// a `clock_gettime` syscall per encoded request (formerly held the
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 8ac2c8d59..cd0ed34ed 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -49,7 +49,7 @@ use crate::responses::{
use crate::session_manager::SessionManager;
use crate::snapshot;
use crate::users::maybe_rewrite_user_password_request;
-use crate::wire::{request_body, usize_to_u32};
+use crate::wire::{request_body, usize_to_u32, verify_request_checksum};
use bytes::Bytes;
use configs::server_ng::NgSystemConfig;
use consensus::{
@@ -761,6 +761,37 @@ async fn handle_client_request<B, MJ, S, SB>(
}
};
+ // The last point that still sees the body the CLIENT sent; every rewrite
below
+ // substitutes server-chosen bytes and carries the stamp through unchanged.
+ if let Err(error) = verify_request_checksum(&request) {
+ warn!(
+ transport_client_id,
+ operation = ?request.header().operation,
+ request = request.header().request,
+ "dropping client request whose body does not match its own
checksum"
+ );
+ let commit = current_metadata_commit(shard);
+ let reply = build_deny_reply(
+ request.header(),
+ transport_client_id,
+ 0,
+ commit,
+ error.as_code(),
+ );
+ if let Err(send_error) = shard
+ .bus
+ .send_to_client(transport_client_id,
reply.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %send_error,
+ "failed to send request-checksum deny reply"
+ );
+ }
+ return;
+ }
+
ensure_transport_connection(shard, sessions, transport_client_id);
// Any request is liveness proof, not just PING: an idle-but-active client
diff --git a/core/server-ng/src/wire.rs b/core/server-ng/src/wire.rs
index e7a047089..639819efc 100644
--- a/core/server-ng/src/wire.rs
+++ b/core/server-ng/src/wire.rs
@@ -30,6 +30,23 @@ pub(crate) fn request_body(request: &Message<RequestHeader>)
-> &[u8] {
&request.as_slice()[std::mem::size_of::<RequestHeader>()..request.header().size
as usize]
}
+/// Check a client's `request_checksum` against the body it stamps.
+///
+/// Must run BEFORE any body rewrite: PAT / password / consumer-group paths
+/// substitute server-chosen bytes. Zero is "unstamped" and skips the check,
so an
+/// SDK predating the stamp still works.
+///
+/// # Errors
+/// [`IggyError::InvalidFormat`] when the stamp disagrees with the body.
+pub(crate) fn verify_request_checksum(request: &Message<RequestHeader>) ->
Result<(), IggyError> {
+ let stamped = request.header().request_checksum;
+ if stamped == 0 ||
u128::from(iggy_common::calculate_checksum(request_body(request))) == stamped
+ {
+ return Ok(());
+ }
+ Err(IggyError::InvalidFormat)
+}
+
/// Map the transport kind to the legacy wire discriminant
/// (`1=TCP, 2=QUIC, 4=WebSocket`); TLS variants report their base
/// transport. `ClientTransportKind` is `#[non_exhaustive]`, so any other
@@ -65,11 +82,78 @@ pub(crate) fn rewrite_request_body(
.expect("zeroed bytes are a valid request header");
*header = *request.header();
header.size = size;
+ // Both describe the body just replaced, and nothing recomputes them for a
+ // `RequestHeader` -- the prepare projection derives its own
`checksum_body`
+ // downstream. Clear rather than recompute; carrying them forward is a
stale claim.
+ header.checksum = 0;
+ header.checksum_body = 0;
+ // `request_checksum` is deliberately NOT touched: it stamps what the
CLIENT sent,
+ // already validated at admission. Re-stamping it over the substituted
body would
+ // make the client-table reuse check compare a value no client ever
produced.
rewritten.as_mut_slice()[std::mem::size_of::<RequestHeader>()..].copy_from_slice(body);
- // TODO(vsr): the body changed but `request_checksum` / `checksum` /
- // `checksum_body` were copied verbatim from the original header. Safe
- // today because the SDK initializes `request_checksum` to 0 and the
- // server does not validate it; the moment integrity checking lands,
- // recompute these here (or zero them and re-sign in a follow-up step).
Ok(rewritten)
}
+
+#[cfg(test)]
+mod tests {
+ use super::{request_body, rewrite_request_body};
+ use bytes::Bytes;
+ use iggy_binary_protocol::{Command2, Operation, RequestHeader};
+ use server_common::Message;
+ use std::mem::size_of;
+
+ fn request(body: &[u8], request_checksum: u128) -> Message<RequestHeader> {
+ let total_size = size_of::<RequestHeader>() + body.len();
+ let mut message =
Message::<RequestHeader>::new(total_size).transmute_header(
+ |_, header: &mut RequestHeader| {
+ header.command = Command2::Request;
+ header.operation = Operation::CreateStream;
+ header.client = 1;
+ header.session = 1;
+ header.request = 9;
+ header.size = u32::try_from(total_size).expect("fits u32");
+ header.request_checksum = request_checksum;
+ header.checksum = 0xdead;
+ header.checksum_body = 0xbeef;
+ },
+ );
+
message.as_mut_slice()[size_of::<RequestHeader>()..].copy_from_slice(body);
+ message
+ }
+
+ #[test]
+ fn
given_a_body_rewrite_should_keep_the_client_stamp_and_clear_the_stale_seals() {
+ // The secret-bearing wire body is swapped for the hash-carrying
replicated
+ // one. `request_checksum` describes what the client sent and
admission has
+ // already checked it, so it must survive; the other two describe the
body
+ // that just went away.
+ let original = request(b"plaintext-secret", 0x1234);
+ let rewritten = rewrite_request_body(&original,
&Bytes::from_static(b"argon2-hash"))
+ .expect("the rewritten body fits a request message");
+
+ assert_eq!(
+ rewritten.header().request_checksum,
+ 0x1234,
+ "the client's stamp must not be re-signed over server-substituted
bytes"
+ );
+ assert_eq!(rewritten.header().checksum, 0);
+ assert_eq!(rewritten.header().checksum_body, 0);
+ assert_eq!(request_body(&rewritten), b"argon2-hash");
+ assert_eq!(
+ rewritten.header().size as usize,
+ size_of::<RequestHeader>() + b"argon2-hash".len(),
+ "`size` follows the new body, so `request_body` bounds it
correctly"
+ );
+ }
+
+ #[test]
+ fn given_an_unstamped_request_when_rewriting_should_stay_unstamped() {
+ // Zero means "unstamped" all the way through the client table, so a
rewrite
+ // must not manufacture a stamp for a client that sent none.
+ let original = request(b"plaintext-secret", 0);
+ let rewritten = rewrite_request_body(&original,
&Bytes::from_static(b"argon2-hash"))
+ .expect("the rewritten body fits a request message");
+
+ assert_eq!(rewritten.header().request_checksum, 0);
+ }
+}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index c2b6e2c99..f5c5cf837 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -4022,9 +4022,7 @@ where
// Recompute both integrity fields before durable storage:
everything
// above treats `header.checksum` as an opaque token, so a
corrupted
// frame passes whenever its flipped value satisfies the
comparisons.
- if let Err(reason) =
- verify_prepare_integrity(&header,
&msg.as_slice()[size_of::<PrepareHeader>()..])
- {
+ if let Err(reason) = verify_prepare_integrity(&header,
msg.as_slice()) {
tracing::warn!(
shard = self.id,
op = header.op,
@@ -4093,9 +4091,7 @@ where
// The partition arm reaches the WAL via `apply_repaired_prepare` with
no
// view fence and no ack, so this is its only integrity gate. Without
it a
// repaired partition prepare is journaled on the serving peer's word
alone.
- if let Err(reason) =
- verify_prepare_integrity(&header,
&msg.as_slice()[size_of::<PrepareHeader>()..])
- {
+ if let Err(reason) = verify_prepare_integrity(&header, msg.as_slice())
{
tracing::warn!(
shard = self.id,
op = header.op,
