mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-06 20:19:14 +00:00
cxymds
655f6ae452
* fix(s3): harden Snowball extract error boundaries * fix(s3): close Snowball extract compatibility gaps * fix(s3): verify Snowball request body completion * test(s3): reject forged Snowball streaming signatures * build(deps): pin Snowball archive parser limits * fix(s3): preserve Snowball trailer and member errors * docs(architecture): register Snowball tar fork cleanup * refactor(s3): route Snowball errors through object boundary * ci(deps): allow pinned tokio-tar source * fix: align Snowball archive codec detection * fix(s3): harden Snowball codec compatibility * fix(s3): preserve Snowball codec compatibility * test(zip): align yield wake assertion with Tokio * fix(rio): preserve legacy large-block reads * fix(zip): accept blank tar numeric fields
1182 lines
46 KiB
Rust
1182 lines
46 KiB
Rust
// Copyright 2024 RustFS Team
|
|
//
|
|
// Licensed 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 async_compression::{
|
|
tokio::bufread::{BzDecoder, GzipDecoder, Lz4Decoder, XzDecoder, ZlibDecoder, ZstdDecoder},
|
|
zstd::DParameter,
|
|
};
|
|
use rustfs_rio::S2Decoder;
|
|
use std::io;
|
|
use std::pin::Pin;
|
|
use std::task::{Context, Poll};
|
|
use thiserror::Error;
|
|
use tokio::io::{AsyncRead, AsyncReadExt, BufReader, ReadBuf};
|
|
|
|
const MAGIC_SNIFF_LEN: usize = 6;
|
|
const TAR_HEADER_LEN: usize = 512;
|
|
const TAR_CHECKSUM_START: usize = 148;
|
|
const TAR_CHECKSUM_END: usize = 156;
|
|
const SHARED_SKIPPABLE_FRAME_HEADER_LEN: usize = 8;
|
|
const SNIFF_DISCARD_BUFFER_LEN: usize = 8 * 1024;
|
|
const SNIFF_YIELD_AFTER_BYTES: usize = 64 * 1024;
|
|
const SNIFF_YIELD_AFTER_FRAMES: usize = 64;
|
|
const SNIFF_YIELD_AFTER_READY_READS: usize = 64;
|
|
|
|
// XZ declares its LZMA2 dictionary size before producing decoded bytes, so the
|
|
// decoded-size guard cannot prevent that allocation. This accepts every
|
|
// standard xz preset (the largest needs about 65 MiB to decode) while keeping
|
|
// one Snowball decoder well below the multi-gigabyte archive budget.
|
|
const XZ_DECODER_MEMORY_LIMIT_BYTES: u64 = 128_u64 * 1024 * 1024;
|
|
// Match MinIO's Snowball decoder boundary so a frame cannot reserve an
|
|
// oversized history window before the decoded-size guard observes any bytes.
|
|
const ZSTD_DECODER_MAX_WINDOW_LOG: u32 = 24;
|
|
|
|
pub type Result<T> = std::result::Result<T, ZipError>;
|
|
|
|
#[derive(Debug, Error)]
|
|
pub enum ZipError {
|
|
#[error("unsupported {operation} for format {format:?}")]
|
|
UnsupportedFormat {
|
|
format: CompressionFormat,
|
|
operation: &'static str,
|
|
},
|
|
#[error("failed to inspect archive compression magic")]
|
|
InspectStream(#[source] io::Error),
|
|
}
|
|
|
|
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
|
|
pub enum CompressionFormat {
|
|
Gzip,
|
|
Bzip2,
|
|
Zip,
|
|
Xz,
|
|
Zlib,
|
|
Zstd,
|
|
Lz4,
|
|
S2,
|
|
Tar,
|
|
Unknown,
|
|
}
|
|
|
|
/// Reader returned by [`CompressionFormat::sniff`]. It replays the bounded
|
|
/// prefix that identified the codec before continuing with the original
|
|
/// stream. Codec-neutral leading skippable frames are consumed by `sniff` and
|
|
/// intentionally omitted from the replayed stream.
|
|
#[derive(Debug)]
|
|
pub struct SniffedReader<R> {
|
|
inner: R,
|
|
prefix: Vec<u8>,
|
|
prefix_pos: usize,
|
|
}
|
|
|
|
impl<R> AsyncRead for SniffedReader<R>
|
|
where
|
|
R: AsyncRead + Unpin,
|
|
{
|
|
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
|
if self.prefix_pos < self.prefix.len() && buf.remaining() > 0 {
|
|
let to_copy = buf.remaining().min(self.prefix.len() - self.prefix_pos);
|
|
buf.put_slice(&self.prefix[self.prefix_pos..self.prefix_pos + to_copy]);
|
|
self.prefix_pos += to_copy;
|
|
return Poll::Ready(Ok(()));
|
|
}
|
|
|
|
Pin::new(&mut self.inner).poll_read(cx, buf)
|
|
}
|
|
}
|
|
|
|
fn is_shared_skippable_magic(prefix: &[u8]) -> bool {
|
|
prefix.len() >= 4 && (0x50..=0x5f).contains(&prefix[0]) && prefix[1..4] == [0x2a, 0x4d, 0x18]
|
|
}
|
|
|
|
#[derive(Debug, Default)]
|
|
struct SniffYieldBudget {
|
|
ready_reads: usize,
|
|
input_bytes: usize,
|
|
frames: usize,
|
|
}
|
|
|
|
impl SniffYieldBudget {
|
|
async fn record_read(&mut self, read: usize) {
|
|
self.ready_reads = self.ready_reads.saturating_add(1);
|
|
self.input_bytes = self.input_bytes.saturating_add(read);
|
|
if self.ready_reads >= SNIFF_YIELD_AFTER_READY_READS || self.input_bytes >= SNIFF_YIELD_AFTER_BYTES {
|
|
self.yield_now().await;
|
|
}
|
|
}
|
|
|
|
async fn record_frame(&mut self) {
|
|
self.frames = self.frames.saturating_add(1);
|
|
if self.frames >= SNIFF_YIELD_AFTER_FRAMES {
|
|
self.yield_now().await;
|
|
}
|
|
}
|
|
|
|
async fn yield_now(&mut self) {
|
|
tokio::task::yield_now().await;
|
|
self.ready_reads = 0;
|
|
self.input_bytes = 0;
|
|
self.frames = 0;
|
|
}
|
|
}
|
|
|
|
async fn inspect_fill_to<R>(input: &mut R, output: &mut Vec<u8>, target_len: usize, budget: &mut SniffYieldBudget) -> Result<bool>
|
|
where
|
|
R: AsyncRead + Unpin,
|
|
{
|
|
let mut scratch = [0u8; TAR_HEADER_LEN];
|
|
while output.len() < target_len {
|
|
let remaining = target_len - output.len();
|
|
let to_read = remaining.min(scratch.len());
|
|
let read = input.read(&mut scratch[..to_read]).await.map_err(ZipError::InspectStream)?;
|
|
if read == 0 {
|
|
return Ok(false);
|
|
}
|
|
output.extend_from_slice(&scratch[..read]);
|
|
budget.record_read(read).await;
|
|
}
|
|
Ok(true)
|
|
}
|
|
|
|
async fn inspect_discard_fully<R>(
|
|
input: &mut R,
|
|
mut remaining: u64,
|
|
budget: &mut SniffYieldBudget,
|
|
truncated_message: &'static str,
|
|
) -> Result<()>
|
|
where
|
|
R: AsyncRead + Unpin,
|
|
{
|
|
let mut discard = [0u8; SNIFF_DISCARD_BUFFER_LEN];
|
|
while remaining > 0 {
|
|
let to_read = usize::try_from(remaining).map_or(discard.len(), |remaining| remaining.min(discard.len()));
|
|
let read = input.read(&mut discard[..to_read]).await.map_err(ZipError::InspectStream)?;
|
|
if read == 0 {
|
|
return Err(ZipError::InspectStream(io::Error::new(io::ErrorKind::UnexpectedEof, truncated_message)));
|
|
}
|
|
let read_u64 =
|
|
u64::try_from(read).map_err(|_| ZipError::InspectStream(io::Error::other("skippable frame read size overflowed")))?;
|
|
remaining -= read_u64;
|
|
budget.record_read(read).await;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn tar_octal_field(field: &[u8]) -> Option<u64> {
|
|
let mut value = 0u64;
|
|
let mut saw_digit = false;
|
|
let mut terminated = false;
|
|
|
|
for byte in field {
|
|
match *byte {
|
|
b'0'..=b'7' if !terminated => {
|
|
saw_digit = true;
|
|
value = value.checked_mul(8)?.checked_add(u64::from(*byte - b'0'))?;
|
|
}
|
|
b' ' if !saw_digit && !terminated => {}
|
|
b' ' | 0 if saw_digit || terminated => terminated = true,
|
|
0 => terminated = true,
|
|
_ => return None,
|
|
}
|
|
}
|
|
|
|
saw_digit.then_some(value)
|
|
}
|
|
|
|
fn tar_numeric_field_has_shape(field: &[u8]) -> bool {
|
|
if field.first().is_some_and(|byte| byte & 0x80 != 0) {
|
|
return true;
|
|
}
|
|
tar_octal_field(field).is_some() || field.iter().all(|byte| matches!(*byte, b' ' | 0))
|
|
}
|
|
|
|
fn is_tar_header(prefix: &[u8]) -> bool {
|
|
let Some(header) = prefix.get(..TAR_HEADER_LEN) else {
|
|
return false;
|
|
};
|
|
if header.iter().all(|byte| *byte == 0) {
|
|
return true;
|
|
}
|
|
|
|
let has_path = header[..100].iter().any(|byte| *byte != 0);
|
|
if !has_path
|
|
|| ![(100, 108), (108, 116), (116, 124), (124, 136), (136, 148)]
|
|
.iter()
|
|
.all(|(start, end)| tar_numeric_field_has_shape(&header[*start..*end]))
|
|
{
|
|
return false;
|
|
}
|
|
|
|
let Some(expected) = tar_octal_field(&header[TAR_CHECKSUM_START..TAR_CHECKSUM_END]) else {
|
|
return false;
|
|
};
|
|
let actual = header[..TAR_CHECKSUM_START]
|
|
.iter()
|
|
.chain(std::iter::repeat_n(&b' ', TAR_CHECKSUM_END - TAR_CHECKSUM_START))
|
|
.chain(&header[TAR_CHECKSUM_END..])
|
|
.fold(0u64, |sum, byte| sum + u64::from(*byte));
|
|
actual == expected
|
|
}
|
|
|
|
/// Archive guardrails. The values are carried here so every archive caller
|
|
/// shares one default policy; enforcement belongs to the caller, which maps a
|
|
/// breach onto its own protocol error.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub struct ArchiveLimits {
|
|
pub max_entries: usize,
|
|
pub max_entry_size: u64,
|
|
pub max_total_unpacked_size: u64,
|
|
pub max_decoded_size: u64,
|
|
pub max_path_length: usize,
|
|
pub max_pax_metadata_size: u64,
|
|
pub max_total_pax_metadata_size: u64,
|
|
pub max_pax_metadata_records: usize,
|
|
pub max_total_pax_metadata_records: usize,
|
|
pub validate_entry_paths: bool,
|
|
}
|
|
|
|
impl Default for ArchiveLimits {
|
|
fn default() -> Self {
|
|
Self {
|
|
max_entries: 100_000,
|
|
max_entry_size: 1_073_741_824,
|
|
max_total_unpacked_size: 10_737_418_240,
|
|
max_decoded_size: 11_811_160_064,
|
|
max_path_length: 1024,
|
|
max_pax_metadata_size: 1_048_576,
|
|
max_total_pax_metadata_size: 67_108_864,
|
|
max_pax_metadata_records: 4_096,
|
|
max_total_pax_metadata_records: 100_000,
|
|
validate_entry_paths: true,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl CompressionFormat {
|
|
/// Map an archive extension onto the stream codec needed to read it.
|
|
/// Tar-family suffixes (`tgz`, `tbz2`, `txz`, `tzst`, ...) resolve to their
|
|
/// codec because the tar container itself is read from the decoded stream.
|
|
pub fn from_extension(ext: &str) -> Self {
|
|
match ext.to_ascii_lowercase().as_str() {
|
|
"gz" | "gzip" | "tgz" => CompressionFormat::Gzip,
|
|
"bz2" | "bzip2" | "tbz" | "tbz2" => CompressionFormat::Bzip2,
|
|
"xz" | "txz" => CompressionFormat::Xz,
|
|
"zlib" | "zz" => CompressionFormat::Zlib,
|
|
"zst" | "zstd" | "tzst" => CompressionFormat::Zstd,
|
|
"lz4" | "tlz4" => CompressionFormat::Lz4,
|
|
"s2" | "snappy" => CompressionFormat::S2,
|
|
"tar" => CompressionFormat::Tar,
|
|
"zip" => CompressionFormat::Zip,
|
|
_ => CompressionFormat::Unknown,
|
|
}
|
|
}
|
|
|
|
pub fn extension(&self) -> &'static str {
|
|
match self {
|
|
CompressionFormat::Gzip => "gz",
|
|
CompressionFormat::Bzip2 => "bz2",
|
|
CompressionFormat::Zip => "zip",
|
|
CompressionFormat::Xz => "xz",
|
|
CompressionFormat::Zlib => "zlib",
|
|
CompressionFormat::Zstd => "zst",
|
|
CompressionFormat::Lz4 => "lz4",
|
|
CompressionFormat::S2 => "s2",
|
|
CompressionFormat::Tar => "tar",
|
|
CompressionFormat::Unknown => "",
|
|
}
|
|
}
|
|
|
|
/// Detect a stream codec from an unambiguous bounded prefix. Ordinary
|
|
/// unknown bytes are a raw TAR stream, matching MinIO Snowball behavior;
|
|
/// the skippable-frame magic shared by Zstd and LZ4 remains
|
|
/// [`CompressionFormat::Unknown`] until [`Self::sniff`] sees a later frame.
|
|
/// Object names and suffixes do not participate in detection.
|
|
///
|
|
/// Zlib deliberately has no magic match here: its two-byte header can also
|
|
/// be the start of a valid TAR member name. Callers preserving historical
|
|
/// zlib-by-extension behavior must apply that compatibility fallback only
|
|
/// after this method returns [`CompressionFormat::Tar`].
|
|
/// ZIP signatures are also left as TAR because stream ZIP decoding is not
|
|
/// supported and those bytes are valid at the start of a TAR member name.
|
|
pub fn from_magic(prefix: &[u8]) -> Self {
|
|
if prefix.starts_with(&[0x1f, 0x8b, 0x08]) {
|
|
return Self::Gzip;
|
|
}
|
|
if prefix.starts_with(&[0x28, 0xb5, 0x2f, 0xfd]) {
|
|
return Self::Zstd;
|
|
}
|
|
if is_shared_skippable_magic(prefix) {
|
|
return Self::Unknown;
|
|
}
|
|
if prefix.starts_with(&[0x04, 0x22, 0x4d, 0x18]) {
|
|
return Self::Lz4;
|
|
}
|
|
if prefix.starts_with(&[0xff, 0x06, 0x00, 0x00]) {
|
|
return Self::S2;
|
|
}
|
|
if prefix.starts_with(b"BZh") {
|
|
return Self::Bzip2;
|
|
}
|
|
if prefix.starts_with(&[0xfd, 0x37, 0x7a, 0x58, 0x5a, 0x00]) {
|
|
return Self::Xz;
|
|
}
|
|
Self::Tar
|
|
}
|
|
|
|
/// Identify the codec and return a reader that replays the deciding prefix.
|
|
/// A complete, checksum-valid TAR header wins over a codec-like member-name
|
|
/// prefix. Leading skippable frames shared by the Zstd and LZ4 frame formats
|
|
/// are discarded with bounded memory until a non-skippable frame identifies
|
|
/// the decoder. The underlying reader still observes every discarded byte,
|
|
/// so callers can keep transport length and checksum accounting below this
|
|
/// boundary.
|
|
pub async fn sniff<R>(mut input: R) -> Result<(Self, SniffedReader<R>)>
|
|
where
|
|
R: AsyncRead + Unpin,
|
|
{
|
|
let mut prefix = Vec::with_capacity(TAR_HEADER_LEN);
|
|
let mut budget = SniffYieldBudget::default();
|
|
let _ = inspect_fill_to(&mut input, &mut prefix, MAGIC_SNIFF_LEN, &mut budget).await?;
|
|
let deciding_prefix_len = prefix.len().min(MAGIC_SNIFF_LEN);
|
|
let initial_format = Self::from_magic(&prefix[..deciding_prefix_len]);
|
|
let needs_tar_disambiguation =
|
|
matches!(initial_format, Self::Bzip2 | Self::Lz4) || is_shared_skippable_magic(&prefix[..deciding_prefix_len]);
|
|
if !needs_tar_disambiguation {
|
|
return Ok((
|
|
initial_format,
|
|
SniffedReader {
|
|
inner: input,
|
|
prefix,
|
|
prefix_pos: 0,
|
|
},
|
|
));
|
|
}
|
|
|
|
let _ = inspect_fill_to(&mut input, &mut prefix, TAR_HEADER_LEN, &mut budget).await?;
|
|
if is_tar_header(&prefix) {
|
|
return Ok((
|
|
Self::Tar,
|
|
SniffedReader {
|
|
inner: input,
|
|
prefix,
|
|
prefix_pos: 0,
|
|
},
|
|
));
|
|
}
|
|
|
|
let mut skipped_shared_frame = false;
|
|
|
|
loop {
|
|
if prefix.len() < MAGIC_SNIFF_LEN {
|
|
let _ = inspect_fill_to(&mut input, &mut prefix, MAGIC_SNIFF_LEN, &mut budget).await?;
|
|
}
|
|
let deciding_prefix_len = prefix.len().min(MAGIC_SNIFF_LEN);
|
|
if !is_shared_skippable_magic(&prefix[..deciding_prefix_len]) {
|
|
let mut format = Self::from_magic(&prefix[..deciding_prefix_len]);
|
|
if skipped_shared_frame && !matches!(format, Self::Zstd | Self::Lz4) {
|
|
format = Self::Unknown;
|
|
}
|
|
return Ok((
|
|
format,
|
|
SniffedReader {
|
|
inner: input,
|
|
prefix,
|
|
prefix_pos: 0,
|
|
},
|
|
));
|
|
}
|
|
|
|
skipped_shared_frame = true;
|
|
let complete_header =
|
|
inspect_fill_to(&mut input, &mut prefix, SHARED_SKIPPABLE_FRAME_HEADER_LEN, &mut budget).await?;
|
|
if !complete_header {
|
|
return Err(ZipError::InspectStream(io::Error::new(
|
|
io::ErrorKind::UnexpectedEof,
|
|
"truncated shared Zstd/LZ4 skippable frame header",
|
|
)));
|
|
}
|
|
let payload_len =
|
|
usize::try_from(u32::from_le_bytes([prefix[4], prefix[5], prefix[6], prefix[7]])).map_err(|_| {
|
|
ZipError::InspectStream(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
"shared Zstd/LZ4 skippable frame length does not fit usize",
|
|
))
|
|
})?;
|
|
let buffered_payload_len = (prefix.len() - SHARED_SKIPPABLE_FRAME_HEADER_LEN).min(payload_len);
|
|
let buffered_frame_len = SHARED_SKIPPABLE_FRAME_HEADER_LEN
|
|
.checked_add(buffered_payload_len)
|
|
.ok_or_else(|| ZipError::InspectStream(io::Error::other("skippable frame buffered length overflowed")))?;
|
|
prefix.drain(..buffered_frame_len);
|
|
let remaining_payload_len = payload_len - buffered_payload_len;
|
|
inspect_discard_fully(
|
|
&mut input,
|
|
u64::try_from(remaining_payload_len)
|
|
.map_err(|_| ZipError::InspectStream(io::Error::other("skippable frame remaining length exceeds u64")))?,
|
|
&mut budget,
|
|
"truncated shared Zstd/LZ4 skippable frame payload",
|
|
)
|
|
.await?;
|
|
budget.record_frame().await;
|
|
}
|
|
}
|
|
|
|
pub fn get_decoder<R>(&self, input: R) -> Result<Box<dyn AsyncRead + Send + Unpin>>
|
|
where
|
|
R: AsyncRead + Send + Unpin + 'static,
|
|
{
|
|
let reader = BufReader::new(input);
|
|
|
|
let decoder: Box<dyn AsyncRead + Send + Unpin + 'static> = match self {
|
|
CompressionFormat::Gzip => {
|
|
let mut decoder = GzipDecoder::new(reader);
|
|
decoder.multiple_members(true);
|
|
Box::new(decoder)
|
|
}
|
|
CompressionFormat::Bzip2 => {
|
|
let mut decoder = BzDecoder::new(reader);
|
|
decoder.multiple_members(true);
|
|
Box::new(decoder)
|
|
}
|
|
CompressionFormat::Zlib => {
|
|
let mut decoder = ZlibDecoder::new(reader);
|
|
decoder.multiple_members(true);
|
|
Box::new(decoder)
|
|
}
|
|
CompressionFormat::Xz => {
|
|
let mut decoder = XzDecoder::with_mem_limit(reader, XZ_DECODER_MEMORY_LIMIT_BYTES);
|
|
decoder.multiple_members(true);
|
|
Box::new(decoder)
|
|
}
|
|
CompressionFormat::Zstd => {
|
|
let mut decoder = ZstdDecoder::with_params(reader, &[DParameter::window_log_max(ZSTD_DECODER_MAX_WINDOW_LOG)]);
|
|
decoder.multiple_members(true);
|
|
Box::new(decoder)
|
|
}
|
|
CompressionFormat::Lz4 => {
|
|
let mut decoder = Lz4Decoder::new(reader);
|
|
decoder.multiple_members(true);
|
|
Box::new(decoder)
|
|
}
|
|
CompressionFormat::S2 => Box::new(S2Decoder::new(reader)),
|
|
CompressionFormat::Tar => Box::new(reader),
|
|
CompressionFormat::Zip => {
|
|
return Err(ZipError::UnsupportedFormat {
|
|
format: *self,
|
|
operation: "stream decoding",
|
|
});
|
|
}
|
|
CompressionFormat::Unknown => {
|
|
return Err(ZipError::UnsupportedFormat {
|
|
format: *self,
|
|
operation: "decoding",
|
|
});
|
|
}
|
|
};
|
|
|
|
Ok(decoder)
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use async_compression::{
|
|
Level,
|
|
tokio::write::{BzEncoder, GzipEncoder, Lz4Encoder, XzEncoder, ZstdEncoder},
|
|
zstd::CParameter,
|
|
};
|
|
use std::future::Future;
|
|
use std::mem::size_of;
|
|
use std::sync::{
|
|
Arc,
|
|
atomic::{AtomicUsize, Ordering},
|
|
};
|
|
use std::task::{Wake, Waker};
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
|
|
|
#[derive(Default)]
|
|
struct WakeCounter(AtomicUsize);
|
|
|
|
impl Wake for WakeCounter {
|
|
fn wake(self: Arc<Self>) {
|
|
self.0.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
}
|
|
|
|
async fn encode_xz(payload: &[u8]) -> Vec<u8> {
|
|
let mut encoder = XzEncoder::with_quality(Vec::new(), Level::Fastest);
|
|
encoder.write_all(payload).await.expect("XZ encode should succeed");
|
|
encoder.shutdown().await.expect("XZ encoder shutdown should succeed");
|
|
encoder.into_inner()
|
|
}
|
|
|
|
async fn encode_lz4(payload: &[u8]) -> Vec<u8> {
|
|
let mut encoder = Lz4Encoder::new(Vec::new());
|
|
encoder.write_all(payload).await.expect("LZ4 encode should succeed");
|
|
encoder.shutdown().await.expect("LZ4 encoder shutdown should succeed");
|
|
encoder.into_inner()
|
|
}
|
|
|
|
async fn encode_zstd_with_window(payload: &[u8], window_log: u32) -> Vec<u8> {
|
|
let mut encoder = ZstdEncoder::with_quality_and_params(
|
|
Vec::new(),
|
|
Level::Default,
|
|
&[CParameter::window_log(window_log), CParameter::content_size_flag(false)],
|
|
);
|
|
encoder.write_all(payload).await.expect("Zstd encode should succeed");
|
|
encoder.shutdown().await.expect("Zstd encoder shutdown should succeed");
|
|
encoder.into_inner()
|
|
}
|
|
|
|
fn shared_skippable_frame(variant: u8, payload: &[u8]) -> Vec<u8> {
|
|
assert!(variant <= 0x0f);
|
|
let payload_len = u32::try_from(payload.len()).expect("test skippable payload should fit u32");
|
|
let mut frame = Vec::with_capacity(SHARED_SKIPPABLE_FRAME_HEADER_LEN + payload.len());
|
|
frame.extend_from_slice(&[0x50 + variant, 0x2a, 0x4d, 0x18]);
|
|
frame.extend_from_slice(&payload_len.to_le_bytes());
|
|
frame.extend_from_slice(payload);
|
|
frame
|
|
}
|
|
|
|
fn raw_tar_with_name(name: &[u8]) -> Vec<u8> {
|
|
assert!(!name.is_empty() && name.len() <= 100, "test TAR name must fit the legacy name field");
|
|
|
|
let mut header = [0u8; TAR_HEADER_LEN];
|
|
header[..name.len()].copy_from_slice(name);
|
|
for (start, end) in [(100, 108), (108, 116), (116, 124), (124, 136), (136, 148)] {
|
|
header[start..end].fill(b'0');
|
|
header[end - 1] = 0;
|
|
}
|
|
header[156] = b'0';
|
|
header[257..263].copy_from_slice(b"ustar\0");
|
|
header[263..265].copy_from_slice(b"00");
|
|
header[TAR_CHECKSUM_START..TAR_CHECKSUM_END].fill(b' ');
|
|
let checksum = header.iter().fold(0u64, |sum, byte| sum + u64::from(*byte));
|
|
let checksum_field = format!("{checksum:06o}\0 ");
|
|
assert_eq!(checksum_field.len(), TAR_CHECKSUM_END - TAR_CHECKSUM_START);
|
|
header[TAR_CHECKSUM_START..TAR_CHECKSUM_END].copy_from_slice(checksum_field.as_bytes());
|
|
|
|
let mut archive = header.to_vec();
|
|
archive.resize(TAR_HEADER_LEN * 3, 0);
|
|
archive
|
|
}
|
|
|
|
struct FragmentedReader {
|
|
bytes: Vec<u8>,
|
|
position: usize,
|
|
return_pending: bool,
|
|
}
|
|
|
|
impl FragmentedReader {
|
|
fn new(bytes: Vec<u8>) -> Self {
|
|
Self {
|
|
bytes,
|
|
position: 0,
|
|
return_pending: true,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl AsyncRead for FragmentedReader {
|
|
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, output: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
|
if self.return_pending {
|
|
self.return_pending = false;
|
|
cx.waker().wake_by_ref();
|
|
return Poll::Pending;
|
|
}
|
|
if self.position >= self.bytes.len() || output.remaining() == 0 {
|
|
return Poll::Ready(Ok(()));
|
|
}
|
|
|
|
let byte = self.bytes[self.position];
|
|
self.position += 1;
|
|
self.return_pending = true;
|
|
output.put_slice(&[byte]);
|
|
Poll::Ready(Ok(()))
|
|
}
|
|
}
|
|
|
|
struct AlwaysReadyOneByteReader {
|
|
bytes: Vec<u8>,
|
|
position: usize,
|
|
ready_reads: Arc<AtomicUsize>,
|
|
bytes_read: Arc<AtomicUsize>,
|
|
}
|
|
|
|
impl AsyncRead for AlwaysReadyOneByteReader {
|
|
fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, output: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
|
self.ready_reads.fetch_add(1, Ordering::Relaxed);
|
|
if self.position >= self.bytes.len() || output.remaining() == 0 {
|
|
return Poll::Ready(Ok(()));
|
|
}
|
|
|
|
let byte = self.bytes[self.position];
|
|
self.position += 1;
|
|
self.bytes_read.fetch_add(1, Ordering::Relaxed);
|
|
output.put_slice(&[byte]);
|
|
Poll::Ready(Ok(()))
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
struct ErrorAfterBytes {
|
|
bytes: Vec<u8>,
|
|
position: usize,
|
|
}
|
|
|
|
impl AsyncRead for ErrorAfterBytes {
|
|
fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, output: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
|
if self.position >= self.bytes.len() {
|
|
return Poll::Ready(Err(io::Error::new(io::ErrorKind::ConnectionReset, "sentinel inspect failure")));
|
|
}
|
|
let available = self.bytes.len() - self.position;
|
|
let to_copy = available.min(output.remaining());
|
|
output.put_slice(&self.bytes[self.position..self.position + to_copy]);
|
|
self.position += to_copy;
|
|
Poll::Ready(Ok(()))
|
|
}
|
|
}
|
|
|
|
async fn assert_shared_skippable_round_trip(format: CompressionFormat, encoded: Vec<u8>, frames: &[Vec<u8>]) {
|
|
let mut stream = Vec::new();
|
|
for frame in frames {
|
|
stream.extend_from_slice(frame);
|
|
}
|
|
stream.extend_from_slice(&encoded);
|
|
|
|
let (detected, mut sniffed) = CompressionFormat::sniff(std::io::Cursor::new(stream.clone()))
|
|
.await
|
|
.expect("shared skippable prefix should be inspected");
|
|
assert_eq!(detected, format);
|
|
let mut replayed_prefix = [0u8; MAGIC_SNIFF_LEN];
|
|
sniffed
|
|
.read_exact(&mut replayed_prefix)
|
|
.await
|
|
.expect("deciding codec prefix should replay completely");
|
|
assert_eq!(replayed_prefix, encoded[..MAGIC_SNIFF_LEN]);
|
|
|
|
let (detected, sniffed) = CompressionFormat::sniff(std::io::Cursor::new(stream))
|
|
.await
|
|
.expect("shared skippable prefix should be inspected for decoding");
|
|
let mut decoder = detected.get_decoder(sniffed).expect("detected decoder should be created");
|
|
let mut decoded = Vec::new();
|
|
decoder.read_to_end(&mut decoded).await.expect("framed stream should decode");
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
async fn assert_ready_sniff_yields(input: Vec<u8>, encoded_prefix: &[u8]) {
|
|
let wake_counter = Arc::new(WakeCounter::default());
|
|
let waker = Waker::from(wake_counter.clone());
|
|
let mut sniff = Box::pin(CompressionFormat::sniff(std::io::Cursor::new(input)));
|
|
{
|
|
let mut cx = Context::from_waker(&waker);
|
|
assert!(sniff.as_mut().poll(&mut cx).is_pending(), "bounded inspection must yield");
|
|
}
|
|
tokio::task::yield_now().await;
|
|
assert_eq!(wake_counter.0.load(Ordering::Relaxed), 1, "yield must arrange another poll");
|
|
|
|
let (format, mut sniffed) = sniff.await.expect("inspection should resume after yielding");
|
|
assert_eq!(format, CompressionFormat::Lz4);
|
|
let mut replayed_prefix = [0u8; MAGIC_SNIFF_LEN];
|
|
sniffed
|
|
.read_exact(&mut replayed_prefix)
|
|
.await
|
|
.expect("deciding codec prefix should replay after yielding");
|
|
assert_eq!(replayed_prefix, encoded_prefix);
|
|
}
|
|
|
|
fn xz_crc32(bytes: &[u8]) -> u32 {
|
|
let mut crc = u32::MAX;
|
|
for byte in bytes {
|
|
crc ^= u32::from(*byte);
|
|
for _ in 0..8 {
|
|
let mask = 0_u32.wrapping_sub(crc & 1);
|
|
crc = (crc >> 1) ^ (0xedb8_8320 & mask);
|
|
}
|
|
}
|
|
!crc
|
|
}
|
|
|
|
async fn xz_with_dictionary_property(payload: &[u8], dictionary_property: u8) -> Vec<u8> {
|
|
const XZ_STREAM_HEADER_LEN: usize = 12;
|
|
|
|
let mut encoded = encode_xz(payload).await;
|
|
assert_eq!(&encoded[..MAGIC_SNIFF_LEN], b"\xfd7zXZ\0");
|
|
|
|
let block_header_start = XZ_STREAM_HEADER_LEN;
|
|
let block_header_len = (usize::from(encoded[block_header_start]) + 1) * 4;
|
|
let block_header_crc_start = block_header_start + block_header_len - 4;
|
|
assert_eq!(block_header_len, 12, "default XZ fixture should use one compact LZMA2 filter header");
|
|
assert_eq!(
|
|
&encoded[block_header_start + 1..block_header_start + 4],
|
|
&[0x00, 0x21, 0x01],
|
|
"default XZ fixture should contain one LZMA2 filter with one property byte"
|
|
);
|
|
|
|
encoded[block_header_start + 4] = dictionary_property;
|
|
let block_header_crc = xz_crc32(&encoded[block_header_start..block_header_crc_start]).to_le_bytes();
|
|
encoded[block_header_crc_start..block_header_crc_start + 4].copy_from_slice(&block_header_crc);
|
|
encoded
|
|
}
|
|
|
|
#[test]
|
|
fn test_compression_format_from_extension() {
|
|
assert_eq!(CompressionFormat::from_extension("gz"), CompressionFormat::Gzip);
|
|
assert_eq!(CompressionFormat::from_extension("ZIP"), CompressionFormat::Zip);
|
|
assert_eq!(CompressionFormat::from_extension("tzst"), CompressionFormat::Zstd);
|
|
assert_eq!(CompressionFormat::from_extension("s2"), CompressionFormat::S2);
|
|
assert_eq!(CompressionFormat::from_extension("txt"), CompressionFormat::Unknown);
|
|
}
|
|
|
|
#[test]
|
|
fn test_compression_format_from_magic() {
|
|
let cases: &[(&[u8], CompressionFormat)] = &[
|
|
(&[0x1f, 0x8b, 0x08, 0x00], CompressionFormat::Gzip),
|
|
(b"BZh9", CompressionFormat::Bzip2),
|
|
(&[0x28, 0xb5, 0x2f, 0xfd], CompressionFormat::Zstd),
|
|
(&[0x50, 0x2a, 0x4d, 0x18], CompressionFormat::Unknown),
|
|
(&[0x04, 0x22, 0x4d, 0x18], CompressionFormat::Lz4),
|
|
(&[0xff, 0x06, 0x00, 0x00], CompressionFormat::S2),
|
|
(&[0xfd, 0x37, 0x7a, 0x58, 0x5a, 0x00], CompressionFormat::Xz),
|
|
(&[0x78, 0x9c], CompressionFormat::Tar),
|
|
(&[0x78, 0x5e, b'o', b'b'], CompressionFormat::Tar),
|
|
(b"PK\x03\x04", CompressionFormat::Tar),
|
|
(b"plain tar bytes", CompressionFormat::Tar),
|
|
];
|
|
|
|
for (magic, expected) in cases {
|
|
assert_eq!(CompressionFormat::from_magic(magic), *expected, "magic={magic:02x?}");
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_prefers_checksum_valid_tar_over_codec_like_member_names() {
|
|
let cases: &[(&[u8], CompressionFormat)] = &[
|
|
(b"BZh9-report.txt", CompressionFormat::Bzip2),
|
|
(b"\x04\x22\x4d\x18-report.txt", CompressionFormat::Lz4),
|
|
];
|
|
|
|
for (name, prefix_format) in cases {
|
|
assert_eq!(CompressionFormat::from_magic(name), *prefix_format);
|
|
let archive = raw_tar_with_name(name);
|
|
let (format, mut sniffed) = CompressionFormat::sniff(std::io::Cursor::new(archive.clone()))
|
|
.await
|
|
.expect("checksum-valid TAR header should be inspectable");
|
|
assert_eq!(format, CompressionFormat::Tar, "member name prefix={name:02x?}");
|
|
|
|
let mut replayed = Vec::new();
|
|
sniffed
|
|
.read_to_end(&mut replayed)
|
|
.await
|
|
.expect("TAR lookahead should replay without loss");
|
|
assert_eq!(replayed, archive);
|
|
}
|
|
|
|
let mut bad_checksum = raw_tar_with_name(b"BZh9-bad-checksum.txt");
|
|
bad_checksum[99] = b'x';
|
|
let (format, _) = CompressionFormat::sniff(std::io::Cursor::new(bad_checksum))
|
|
.await
|
|
.expect("malformed TAR lookahead should still permit codec detection");
|
|
assert_eq!(format, CompressionFormat::Bzip2, "TAR priority must require a valid checksum");
|
|
}
|
|
|
|
#[test]
|
|
fn test_compression_format_size_is_small() {
|
|
assert!(size_of::<CompressionFormat>() <= 8);
|
|
assert!(size_of::<Option<CompressionFormat>>() <= 16);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_round_trips_gzip_stream() {
|
|
let mut encoder = GzipEncoder::new(Vec::new());
|
|
encoder.write_all(b"payload").await.expect("gzip encode should succeed");
|
|
encoder.shutdown().await.expect("gzip encoder shutdown should succeed");
|
|
|
|
let (format, sniffed) = CompressionFormat::sniff(std::io::Cursor::new(encoder.into_inner()))
|
|
.await
|
|
.expect("gzip magic should be inspected");
|
|
assert_eq!(format, CompressionFormat::Gzip);
|
|
let mut decoder = format.get_decoder(sniffed).expect("gzip decoder should be created");
|
|
let mut decoded = Vec::new();
|
|
decoder.read_to_end(&mut decoded).await.expect("gzip decode should succeed");
|
|
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_still_recognizes_a_real_bzip2_stream() {
|
|
let mut encoder = BzEncoder::new(Vec::new());
|
|
encoder.write_all(b"payload").await.expect("Bzip2 encode should succeed");
|
|
encoder.shutdown().await.expect("Bzip2 encoder shutdown should succeed");
|
|
|
|
let (format, sniffed) = CompressionFormat::sniff(std::io::Cursor::new(encoder.into_inner()))
|
|
.await
|
|
.expect("Bzip2 magic should be inspected after TAR lookahead");
|
|
assert_eq!(format, CompressionFormat::Bzip2);
|
|
let mut decoder = format.get_decoder(sniffed).expect("Bzip2 decoder should be created");
|
|
let mut decoded = Vec::new();
|
|
decoder.read_to_end(&mut decoded).await.expect("Bzip2 decode should succeed");
|
|
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_consumes_concatenated_gzip_members() {
|
|
async fn gzip_member(payload: &[u8]) -> Vec<u8> {
|
|
let mut encoder = GzipEncoder::new(Vec::new());
|
|
encoder.write_all(payload).await.expect("gzip encode should succeed");
|
|
encoder.shutdown().await.expect("gzip encoder shutdown should succeed");
|
|
encoder.into_inner()
|
|
}
|
|
|
|
let mut encoded = gzip_member(b"first-").await;
|
|
encoded.extend(gzip_member(b"second").await);
|
|
let mut decoder = CompressionFormat::Gzip
|
|
.get_decoder(std::io::Cursor::new(encoded))
|
|
.expect("gzip decoder should be created");
|
|
let mut decoded = Vec::new();
|
|
|
|
decoder
|
|
.read_to_end(&mut decoded)
|
|
.await
|
|
.expect("concatenated gzip members should decode");
|
|
|
|
assert_eq!(decoded, b"first-second");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_round_trips_lz4_stream() {
|
|
let mut encoder = Lz4Encoder::new(Vec::new());
|
|
encoder.write_all(b"payload").await.expect("LZ4 encode should succeed");
|
|
encoder.shutdown().await.expect("LZ4 encoder shutdown should succeed");
|
|
|
|
let (format, sniffed) = CompressionFormat::sniff(std::io::Cursor::new(encoder.into_inner()))
|
|
.await
|
|
.expect("LZ4 magic should be inspected");
|
|
assert_eq!(format, CompressionFormat::Lz4);
|
|
let mut decoder = format.get_decoder(sniffed).expect("LZ4 decoder should be created");
|
|
let mut decoded = Vec::new();
|
|
decoder.read_to_end(&mut decoded).await.expect("LZ4 decode should succeed");
|
|
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_skips_shared_frames_before_lz4_and_zstd() {
|
|
let lz4 = encode_lz4(b"payload").await;
|
|
assert_shared_skippable_round_trip(
|
|
CompressionFormat::Lz4,
|
|
lz4,
|
|
&[shared_skippable_frame(0, b""), shared_skippable_frame(15, b"lz4 metadata")],
|
|
)
|
|
.await;
|
|
|
|
let zstd = encode_zstd_with_window(b"payload", ZSTD_DECODER_MAX_WINDOW_LOG).await;
|
|
let large_metadata = vec![0x5a; SNIFF_YIELD_AFTER_BYTES + 1];
|
|
assert_shared_skippable_round_trip(
|
|
CompressionFormat::Zstd,
|
|
zstd,
|
|
&[shared_skippable_frame(3, &large_metadata), shared_skippable_frame(4, b"")],
|
|
)
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_rejects_truncated_shared_skippable_frames() {
|
|
let truncated_header = [0x50, 0x2a, 0x4d, 0x18, 0x04, 0x00];
|
|
let err = CompressionFormat::sniff(std::io::Cursor::new(truncated_header))
|
|
.await
|
|
.expect_err("truncated shared frame header must fail");
|
|
assert!(matches!(
|
|
err,
|
|
ZipError::InspectStream(ref source) if source.kind() == io::ErrorKind::UnexpectedEof
|
|
));
|
|
|
|
let mut truncated_payload = shared_skippable_frame(0, b"abcd");
|
|
truncated_payload.truncate(truncated_payload.len() - 2);
|
|
let err = CompressionFormat::sniff(std::io::Cursor::new(truncated_payload))
|
|
.await
|
|
.expect_err("truncated shared frame payload must fail");
|
|
assert!(matches!(
|
|
err,
|
|
ZipError::InspectStream(ref source) if source.kind() == io::ErrorKind::UnexpectedEof
|
|
));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_shared_frames_handles_fragmented_pending_reads() {
|
|
let encoded = encode_lz4(b"payload").await;
|
|
let mut input = Vec::new();
|
|
for variant in 0..=SNIFF_YIELD_AFTER_FRAMES {
|
|
input.extend_from_slice(&shared_skippable_frame(u8::try_from(variant % 16).expect("variant should fit u8"), b""));
|
|
}
|
|
input.extend_from_slice(&encoded);
|
|
|
|
let (format, sniffed) = CompressionFormat::sniff(FragmentedReader::new(input))
|
|
.await
|
|
.expect("fragmented shared frames should be inspected");
|
|
assert_eq!(format, CompressionFormat::Lz4);
|
|
let mut decoder = format.get_decoder(sniffed).expect("LZ4 decoder should be created");
|
|
let mut decoded = Vec::new();
|
|
decoder
|
|
.read_to_end(&mut decoded)
|
|
.await
|
|
.expect("fragmented stream should decode");
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_shared_frames_yields_at_frame_budget_with_ready_reader() {
|
|
let encoded = encode_lz4(b"payload").await;
|
|
let mut input = Vec::new();
|
|
for variant in 0..SNIFF_YIELD_AFTER_FRAMES {
|
|
let variant = u8::try_from(variant % 16).expect("variant should fit u8");
|
|
input.extend_from_slice(&shared_skippable_frame(variant, b""));
|
|
}
|
|
input.extend_from_slice(&encoded);
|
|
|
|
assert_ready_sniff_yields(input, &encoded[..MAGIC_SNIFF_LEN]).await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_shared_frame_yields_at_byte_budget_with_ready_reader() {
|
|
let encoded = encode_lz4(b"payload").await;
|
|
let metadata = vec![0x5a; SNIFF_YIELD_AFTER_BYTES];
|
|
let mut input = shared_skippable_frame(0, &metadata);
|
|
input.extend_from_slice(&encoded);
|
|
|
|
assert_ready_sniff_yields(input, &encoded[..MAGIC_SNIFF_LEN]).await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_shared_frame_bounds_always_ready_one_byte_reads_per_poll() {
|
|
let encoded = encode_lz4(b"payload").await;
|
|
let metadata = vec![0x5a; TAR_HEADER_LEN * 4];
|
|
let mut input = shared_skippable_frame(0, &metadata);
|
|
input.extend_from_slice(&encoded);
|
|
|
|
let ready_reads = Arc::new(AtomicUsize::new(0));
|
|
let bytes_read = Arc::new(AtomicUsize::new(0));
|
|
let reader = AlwaysReadyOneByteReader {
|
|
bytes: input,
|
|
position: 0,
|
|
ready_reads: ready_reads.clone(),
|
|
bytes_read: bytes_read.clone(),
|
|
};
|
|
let wake_counter = Arc::new(WakeCounter::default());
|
|
let waker = Waker::from(wake_counter.clone());
|
|
let mut sniff = Box::pin(CompressionFormat::sniff(reader));
|
|
|
|
for poll_number in 0..10 {
|
|
let before = ready_reads.load(Ordering::Relaxed);
|
|
let mut cx = Context::from_waker(&waker);
|
|
assert!(
|
|
sniff.as_mut().poll(&mut cx).is_pending(),
|
|
"poll {poll_number} must yield at the ready-read budget"
|
|
);
|
|
let reads_this_poll = ready_reads.load(Ordering::Relaxed) - before;
|
|
assert_eq!(
|
|
reads_this_poll, SNIFF_YIELD_AFTER_READY_READS,
|
|
"poll {poll_number} exceeded the ready-read budget"
|
|
);
|
|
}
|
|
assert!(
|
|
bytes_read.load(Ordering::Relaxed) > TAR_HEADER_LEN,
|
|
"manual polls must progress beyond TAR lookahead into the shared-frame payload"
|
|
);
|
|
assert_eq!(
|
|
wake_counter.0.load(Ordering::Relaxed),
|
|
0,
|
|
"Tokio defers the wake until the scheduler runs"
|
|
);
|
|
tokio::task::yield_now().await;
|
|
assert_eq!(
|
|
wake_counter.0.load(Ordering::Relaxed),
|
|
1,
|
|
"repeated voluntary yields with the same waker must coalesce into one scheduled repoll"
|
|
);
|
|
|
|
let (format, sniffed) = sniff.await.expect("bounded inspection should eventually complete");
|
|
assert_eq!(format, CompressionFormat::Lz4);
|
|
let mut decoder = format.get_decoder(sniffed).expect("LZ4 decoder should be created");
|
|
let mut decoded = Vec::new();
|
|
decoder
|
|
.read_to_end(&mut decoded)
|
|
.await
|
|
.expect("LZ4 stream should survive bounded inspection");
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_shared_frame_preserves_underlying_read_error() {
|
|
let mut bytes = shared_skippable_frame(0, b"abcd");
|
|
bytes.truncate(bytes.len() - 2);
|
|
let err = CompressionFormat::sniff(ErrorAfterBytes { bytes, position: 0 })
|
|
.await
|
|
.expect_err("underlying read failure must escape inspection");
|
|
|
|
assert!(matches!(
|
|
err,
|
|
ZipError::InspectStream(ref source)
|
|
if source.kind() == io::ErrorKind::ConnectionReset && source.to_string() == "sentinel inspect failure"
|
|
));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_shared_frame_without_lz4_or_zstd_successor_is_unknown() {
|
|
let mut input = shared_skippable_frame(0, b"metadata");
|
|
input.extend_from_slice(b"raw tar bytes");
|
|
|
|
let (format, mut sniffed) = CompressionFormat::sniff(std::io::Cursor::new(input))
|
|
.await
|
|
.expect("complete shared frame should be inspectable");
|
|
assert_eq!(format, CompressionFormat::Unknown);
|
|
let mut replayed = Vec::new();
|
|
sniffed
|
|
.read_to_end(&mut replayed)
|
|
.await
|
|
.expect("successor bytes should replay");
|
|
assert_eq!(replayed, b"raw tar bytes");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_round_trips_xz_stream_with_memory_limit() {
|
|
let encoded = encode_xz(b"payload").await;
|
|
let (format, sniffed) = CompressionFormat::sniff(std::io::Cursor::new(encoded))
|
|
.await
|
|
.expect("XZ magic should be inspected");
|
|
assert_eq!(format, CompressionFormat::Xz);
|
|
|
|
let mut decoder = format.get_decoder(sniffed).expect("XZ decoder should be created");
|
|
let mut decoded = Vec::new();
|
|
decoder
|
|
.read_to_end(&mut decoded)
|
|
.await
|
|
.expect("XZ decode should succeed within the memory limit");
|
|
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_rejects_xz_dictionary_over_memory_limit() {
|
|
const LZMA2_MAX_DICTIONARY_PROPERTY: u8 = 40;
|
|
|
|
let encoded = xz_with_dictionary_property(b"payload", LZMA2_MAX_DICTIONARY_PROPERTY).await;
|
|
let mut decoder = CompressionFormat::Xz
|
|
.get_decoder(std::io::Cursor::new(encoded))
|
|
.expect("XZ decoder should be created before inspecting the stream header");
|
|
let mut decoded = Vec::new();
|
|
let err = decoder
|
|
.read_to_end(&mut decoded)
|
|
.await
|
|
.expect_err("hostile XZ dictionary request must exceed the decoder memory limit");
|
|
|
|
assert!(decoded.is_empty(), "decoder must reject the hostile dictionary before producing output");
|
|
assert!(
|
|
err.to_string().contains("memory limit"),
|
|
"hostile XZ dictionary should fail at the decoder memory boundary: {err}"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_accepts_xz_64_mib_dictionary() {
|
|
const LZMA2_64_MIB_DICTIONARY_PROPERTY: u8 = 28;
|
|
|
|
let encoded = xz_with_dictionary_property(b"payload", LZMA2_64_MIB_DICTIONARY_PROPERTY).await;
|
|
let mut decoder = CompressionFormat::Xz
|
|
.get_decoder(std::io::Cursor::new(encoded))
|
|
.expect("XZ decoder should be created before inspecting the stream header");
|
|
let mut decoded = Vec::new();
|
|
decoder
|
|
.read_to_end(&mut decoded)
|
|
.await
|
|
.expect("the memory limit must retain preset-9-compatible dictionary headroom");
|
|
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_accepts_zstd_window_at_limit() {
|
|
let encoded = encode_zstd_with_window(b"payload", ZSTD_DECODER_MAX_WINDOW_LOG).await;
|
|
let mut decoder = CompressionFormat::Zstd
|
|
.get_decoder(std::io::Cursor::new(encoded))
|
|
.expect("Zstd decoder should be created before inspecting the frame header");
|
|
let mut decoded = Vec::new();
|
|
decoder
|
|
.read_to_end(&mut decoded)
|
|
.await
|
|
.expect("16 MiB Zstd windows must remain compatible");
|
|
|
|
assert_eq!(decoded, b"payload");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_rejects_zstd_window_over_limit() {
|
|
let encoded = encode_zstd_with_window(b"payload", ZSTD_DECODER_MAX_WINDOW_LOG + 1).await;
|
|
let mut decoder = CompressionFormat::Zstd
|
|
.get_decoder(std::io::Cursor::new(encoded))
|
|
.expect("Zstd decoder should be created before inspecting the frame header");
|
|
let mut decoded = Vec::new();
|
|
decoder
|
|
.read_to_end(&mut decoded)
|
|
.await
|
|
.expect_err("Zstd windows larger than 16 MiB must be rejected");
|
|
|
|
assert!(decoded.is_empty(), "decoder must reject the oversized window before producing output");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_sniff_replays_raw_tar_prefix() {
|
|
let input = b"ustar-prefix-and-payload".to_vec();
|
|
let (format, mut sniffed) = CompressionFormat::sniff(std::io::Cursor::new(input.clone()))
|
|
.await
|
|
.expect("raw TAR prefix should be inspected");
|
|
let mut replayed = Vec::new();
|
|
sniffed
|
|
.read_to_end(&mut replayed)
|
|
.await
|
|
.expect("sniffed prefix should replay");
|
|
|
|
assert_eq!(format, CompressionFormat::Tar);
|
|
assert_eq!(replayed, input);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_get_decoder_rejects_zip_and_unknown_formats() {
|
|
let zip_err = CompressionFormat::Zip
|
|
.get_decoder(std::io::Cursor::new(Vec::<u8>::new()))
|
|
.err()
|
|
.expect("zip stream decoding should be rejected");
|
|
assert!(matches!(
|
|
zip_err,
|
|
ZipError::UnsupportedFormat {
|
|
format: CompressionFormat::Zip,
|
|
operation: "stream decoding",
|
|
}
|
|
));
|
|
|
|
let unknown_err = CompressionFormat::Unknown
|
|
.get_decoder(std::io::Cursor::new(Vec::<u8>::new()))
|
|
.err()
|
|
.expect("unknown format decoding should be rejected");
|
|
assert!(matches!(
|
|
unknown_err,
|
|
ZipError::UnsupportedFormat {
|
|
format: CompressionFormat::Unknown,
|
|
operation: "decoding",
|
|
}
|
|
));
|
|
}
|
|
}
|