mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-11 15:46:53 +00:00
fix(sse). Resolving Nonce Overwriting Issues in Multi-Package Scenarios (#2582)
Signed-off-by: 唐小鸭 <tangtang1251@qq.com> Co-authored-by: houseme <housemecn@gmail.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
This commit is contained in:
唐小鸭
@@ -174,6 +174,8 @@ pin_project! {
|
||||
base_nonce: [u8; 12], // Base nonce recorded in object metadata
|
||||
current_nonce_base: [u8; 12], // Active base nonce for the current encrypted segment
|
||||
multipart_mode: bool,
|
||||
multipart_parts: Vec<usize>,
|
||||
current_part_index: usize,
|
||||
current_part: usize,
|
||||
block_index: usize,
|
||||
buffer: Vec<u8>,
|
||||
@@ -200,6 +202,8 @@ where
|
||||
base_nonce: nonce,
|
||||
current_nonce_base: nonce,
|
||||
multipart_mode: false,
|
||||
multipart_parts: Vec::new(),
|
||||
current_part_index: 0,
|
||||
current_part: 0,
|
||||
block_index: 0,
|
||||
buffer: Vec::new(),
|
||||
@@ -214,8 +218,8 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_multipart(inner: R, key: [u8; 32], base_nonce: [u8; 12]) -> Self {
|
||||
let first_part = 1;
|
||||
pub fn new_multipart(inner: R, key: [u8; 32], base_nonce: [u8; 12], multipart_parts: Vec<usize>) -> Self {
|
||||
let first_part = multipart_parts.first().copied().unwrap_or(1);
|
||||
let initial_nonce = derive_part_nonce(&base_nonce, first_part);
|
||||
|
||||
debug!("decrypt_reader: initialized multipart mode");
|
||||
@@ -226,6 +230,8 @@ where
|
||||
base_nonce,
|
||||
current_nonce_base: initial_nonce,
|
||||
multipart_mode: true,
|
||||
multipart_parts,
|
||||
current_part_index: 0,
|
||||
current_part: first_part,
|
||||
block_index: 0,
|
||||
buffer: Vec::new(),
|
||||
@@ -265,91 +271,95 @@ where
|
||||
return Poll::Ready(Ok(()));
|
||||
}
|
||||
|
||||
// Read header (8 bytes)
|
||||
while !*this.header_done && *this.header_read < 8 {
|
||||
let mut temp = [0u8; 8];
|
||||
let mut temp_buf = ReadBuf::new(&mut temp[0..8 - *this.header_read]);
|
||||
match this.inner.as_mut().poll_read(cx, &mut temp_buf) {
|
||||
Poll::Pending => return Poll::Pending,
|
||||
Poll::Ready(Ok(())) => {
|
||||
let n = temp_buf.filled().len();
|
||||
if n == 0 {
|
||||
if *this.header_read == 0 {
|
||||
*this.finished = true;
|
||||
return Poll::Ready(Ok(()));
|
||||
if *this.ciphertext_len == 0 {
|
||||
// Read header (8 bytes) only when there is no in-flight payload.
|
||||
while !*this.header_done && *this.header_read < 8 {
|
||||
let mut temp = [0u8; 8];
|
||||
let mut temp_buf = ReadBuf::new(&mut temp[0..8 - *this.header_read]);
|
||||
match this.inner.as_mut().poll_read(cx, &mut temp_buf) {
|
||||
Poll::Pending => return Poll::Pending,
|
||||
Poll::Ready(Ok(())) => {
|
||||
let n = temp_buf.filled().len();
|
||||
if n == 0 {
|
||||
if *this.header_read == 0 {
|
||||
*this.finished = true;
|
||||
return Poll::Ready(Ok(()));
|
||||
}
|
||||
return Poll::Ready(Err(Error::new(
|
||||
std::io::ErrorKind::UnexpectedEof,
|
||||
"unexpected EOF while reading encrypted block header",
|
||||
)));
|
||||
}
|
||||
return Poll::Ready(Err(Error::new(
|
||||
std::io::ErrorKind::UnexpectedEof,
|
||||
"unexpected EOF while reading encrypted block header",
|
||||
)));
|
||||
this.header_buf[*this.header_read..*this.header_read + n].copy_from_slice(&temp_buf.filled()[..n]);
|
||||
*this.header_read += n;
|
||||
}
|
||||
this.header_buf[*this.header_read..*this.header_read + n].copy_from_slice(&temp_buf.filled()[..n]);
|
||||
*this.header_read += n;
|
||||
Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
|
||||
}
|
||||
Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
|
||||
if !*this.header_done && *this.header_read == 8 {
|
||||
*this.header_done = true;
|
||||
}
|
||||
if !*this.header_done && *this.header_read == 8 {
|
||||
*this.header_done = true;
|
||||
}
|
||||
|
||||
if !*this.header_done {
|
||||
return Poll::Pending;
|
||||
}
|
||||
if !*this.header_done {
|
||||
return Poll::Pending;
|
||||
}
|
||||
|
||||
let typ = this.header_buf[0];
|
||||
let len =
|
||||
(this.header_buf[1] as usize) | ((this.header_buf[2] as usize) << 8) | ((this.header_buf[3] as usize) << 16);
|
||||
let crc = (this.header_buf[4] as u32)
|
||||
| ((this.header_buf[5] as u32) << 8)
|
||||
| ((this.header_buf[6] as u32) << 16)
|
||||
| ((this.header_buf[7] as u32) << 24);
|
||||
let typ = this.header_buf[0];
|
||||
let len =
|
||||
(this.header_buf[1] as usize) | ((this.header_buf[2] as usize) << 8) | ((this.header_buf[3] as usize) << 16);
|
||||
*this.header_read = 0;
|
||||
*this.header_done = false;
|
||||
|
||||
*this.header_read = 0;
|
||||
*this.header_done = false;
|
||||
if typ == 0xFF {
|
||||
if *this.multipart_mode {
|
||||
let next_part = if *this.current_part_index + 1 < this.multipart_parts.len() {
|
||||
*this.current_part_index += 1;
|
||||
this.multipart_parts[*this.current_part_index]
|
||||
} else {
|
||||
*this.current_part + 1
|
||||
};
|
||||
debug!(
|
||||
next_part = next_part,
|
||||
"decrypt_reader: reached segment terminator, advancing to next part"
|
||||
);
|
||||
*this.current_part = next_part;
|
||||
*this.current_nonce_base = derive_part_nonce(this.base_nonce, *this.current_part);
|
||||
*this.block_index = 0;
|
||||
*this.ciphertext_read = 0;
|
||||
*this.ciphertext_len = 0;
|
||||
continue;
|
||||
}
|
||||
|
||||
if typ == 0xFF {
|
||||
if *this.multipart_mode {
|
||||
debug!(
|
||||
next_part = *this.current_part + 1,
|
||||
"decrypt_reader: reached segment terminator, advancing to next part"
|
||||
);
|
||||
*this.current_part += 1;
|
||||
*this.current_nonce_base = derive_part_nonce(this.base_nonce, *this.current_part);
|
||||
*this.finished = true;
|
||||
*this.block_index = 0;
|
||||
*this.ciphertext_read = 0;
|
||||
*this.ciphertext_len = 0;
|
||||
continue;
|
||||
}
|
||||
|
||||
*this.finished = true;
|
||||
tracing::debug!(typ = typ, len = len, "decrypt block header");
|
||||
|
||||
if len == 0 {
|
||||
tracing::warn!("encountered zero-length encrypted block, treating as end of stream");
|
||||
*this.finished = true;
|
||||
*this.ciphertext_read = 0;
|
||||
*this.ciphertext_len = 0;
|
||||
continue;
|
||||
}
|
||||
|
||||
let Some(payload_len) = len.checked_sub(4) else {
|
||||
tracing::error!("invalid encrypted block length: typ={} len={} header={:?}", typ, len, this.header_buf);
|
||||
return Poll::Ready(Err(Error::other("Invalid encrypted block length")));
|
||||
};
|
||||
|
||||
if this.ciphertext_buf.len() < payload_len {
|
||||
this.ciphertext_buf.resize(payload_len, 0);
|
||||
}
|
||||
*this.ciphertext_len = payload_len;
|
||||
*this.ciphertext_read = 0;
|
||||
*this.ciphertext_len = 0;
|
||||
continue;
|
||||
}
|
||||
|
||||
tracing::debug!(typ = typ, len = len, "decrypt block header");
|
||||
|
||||
if len == 0 {
|
||||
tracing::warn!("encountered zero-length encrypted block, treating as end of stream");
|
||||
*this.finished = true;
|
||||
*this.ciphertext_read = 0;
|
||||
*this.ciphertext_len = 0;
|
||||
continue;
|
||||
}
|
||||
|
||||
let Some(payload_len) = len.checked_sub(4) else {
|
||||
tracing::error!("invalid encrypted block length: typ={} len={} header={:?}", typ, len, this.header_buf);
|
||||
return Poll::Ready(Err(Error::other("Invalid encrypted block length")));
|
||||
};
|
||||
|
||||
if this.ciphertext_buf.len() < payload_len {
|
||||
this.ciphertext_buf.resize(payload_len, 0);
|
||||
}
|
||||
*this.ciphertext_len = payload_len;
|
||||
*this.ciphertext_read = 0;
|
||||
|
||||
while *this.ciphertext_read < *this.ciphertext_len {
|
||||
let mut temp_buf = ReadBuf::new(&mut this.ciphertext_buf[*this.ciphertext_read..*this.ciphertext_len]);
|
||||
match this.inner.as_mut().poll_read(cx, &mut temp_buf) {
|
||||
@@ -420,12 +430,16 @@ where
|
||||
return Poll::Ready(Err(Error::other("Plaintext length mismatch")));
|
||||
}
|
||||
|
||||
let expected_crc = (this.header_buf[4] as u32)
|
||||
| ((this.header_buf[5] as u32) << 8)
|
||||
| ((this.header_buf[6] as u32) << 16)
|
||||
| ((this.header_buf[7] as u32) << 24);
|
||||
let actual_crc = {
|
||||
let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
|
||||
hasher.update(&plaintext);
|
||||
hasher.finalize() as u32
|
||||
};
|
||||
if actual_crc != crc {
|
||||
if actual_crc != expected_crc {
|
||||
*this.ciphertext_read = 0;
|
||||
*this.ciphertext_len = 0;
|
||||
return Poll::Ready(Err(Error::other("CRC32 mismatch")));
|
||||
@@ -528,6 +542,49 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
struct PendingChunkedCursor {
|
||||
inner: Cursor<Vec<u8>>,
|
||||
max_chunk: usize,
|
||||
should_pending: bool,
|
||||
}
|
||||
|
||||
impl PendingChunkedCursor {
|
||||
fn new(data: Vec<u8>, max_chunk: usize) -> Self {
|
||||
Self {
|
||||
inner: Cursor::new(data),
|
||||
max_chunk,
|
||||
should_pending: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncRead for PendingChunkedCursor {
|
||||
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
|
||||
if self.should_pending {
|
||||
self.should_pending = false;
|
||||
cx.waker().wake_by_ref();
|
||||
return Poll::Pending;
|
||||
}
|
||||
|
||||
if self.max_chunk == 0 || buf.remaining() == 0 {
|
||||
return Poll::Ready(Ok(()));
|
||||
}
|
||||
|
||||
let remaining = self.inner.get_ref().len() as u64 - self.inner.position();
|
||||
if remaining == 0 {
|
||||
return Poll::Ready(Ok(()));
|
||||
}
|
||||
|
||||
let to_read = remaining.min(self.max_chunk as u64).min(buf.remaining() as u64) as usize;
|
||||
let start = self.inner.position() as usize;
|
||||
let end = start + to_read;
|
||||
buf.put_slice(&self.inner.get_ref()[start..end]);
|
||||
self.inner.set_position(end as u64);
|
||||
self.should_pending = true;
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
}
|
||||
|
||||
fn encrypt_with_legacy_nonce_reuse(data: &[u8], key: [u8; 32], nonce: [u8; 12]) -> Vec<u8> {
|
||||
let cipher = Aes256Gcm::new_from_slice(&key).expect("valid key");
|
||||
let nonce = Nonce::try_from(nonce.as_slice()).expect("valid nonce");
|
||||
@@ -697,6 +754,29 @@ mod tests {
|
||||
assert_eq!(decrypted, data);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_decrypt_reader_large_with_pending_chunks() {
|
||||
let size = 1024 * 1024;
|
||||
let mut data = vec![0u8; size];
|
||||
rand::rng().fill(&mut data[..]);
|
||||
let mut key = [0u8; 32];
|
||||
let mut nonce = [0u8; 12];
|
||||
rand::rng().fill_bytes(&mut key);
|
||||
rand::rng().fill_bytes(&mut nonce);
|
||||
|
||||
let reader = Cursor::new(data.clone());
|
||||
let mut encrypt_reader = EncryptReader::new(reader, key, nonce);
|
||||
let mut encrypted = Vec::new();
|
||||
encrypt_reader.read_to_end(&mut encrypted).await.unwrap();
|
||||
|
||||
let reader = PendingChunkedCursor::new(encrypted, 3);
|
||||
let mut decrypt_reader = DecryptReader::new(reader, key, nonce);
|
||||
let mut decrypted = Vec::new();
|
||||
decrypt_reader.read_to_end(&mut decrypted).await.unwrap();
|
||||
|
||||
assert_eq!(decrypted, data);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_decrypt_reader_large_through_reader_stream() {
|
||||
let size = 1024 * 1024;
|
||||
@@ -781,7 +861,7 @@ mod tests {
|
||||
combined.extend_from_slice(&encrypted_two);
|
||||
|
||||
let reader = BufReader::new(Cursor::new(combined));
|
||||
let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce);
|
||||
let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce, vec![1, 2]);
|
||||
let mut decrypted = Vec::new();
|
||||
decrypt_reader.read_to_end(&mut decrypted).await.unwrap();
|
||||
|
||||
@@ -855,7 +935,7 @@ mod tests {
|
||||
combined.extend_from_slice(&encrypted_two);
|
||||
|
||||
let reader = BufReader::new(Cursor::new(combined));
|
||||
let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce);
|
||||
let mut decrypt_reader = DecryptReader::new_multipart(reader, key, base_nonce, vec![1, 2]);
|
||||
let mut decrypted = Vec::new();
|
||||
decrypt_reader.read_to_end(&mut decrypted).await.unwrap();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user