tzap_core/

streaming_writer.rs

use std::io::{self, ErrorKind, Read};

use crate::crypto::{KdfParams, MasterKey};
use crate::format::{ArchiveWriteError, FormatError};
use crate::metadata::{
    normalize_lookup_file_path, validate_directory_path_bytes, validate_file_path_bytes,
};
use crate::writer::{
    write_ordered_parallel_stream_archive_to_sink, ArchiveWriteSink, MemoryArchiveSink,
    RootAuthAuthenticator, RootAuthWriterConfig, StreamingRegularMember, WriterOptions,
    WrittenArchive, WrittenArchiveSummary,
};

const TAR_BLOCK_LEN: usize = 512;
const MAX_TAR_STDIN_METADATA_PAYLOAD_BYTES: usize = 1024 * 1024;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamingTarWriterSummary {
    pub archive: WrittenArchiveSummary,
    pub input_member_count: u64,
    pub input_tar_bytes: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamingRawWriterSummary {
    pub archive: WrittenArchiveSummary,
    pub input_bytes: u64,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct TarStdinInputSummary {
    regular_file_count: u64,
    skipped_directory_count: u64,
    input_tar_bytes: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct TarStdinRegularMember {
    path: Vec<u8>,
    mode: u32,
    mtime: u64,
    logical_size: u64,
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct LocalTarMetadata {
    pending_header: bool,
    pax_path: Option<Vec<u8>>,
    pax_size: Option<u64>,
    pax_mode: Option<u32>,
    pax_mtime: Option<u64>,
    gnu_long_name: Option<Vec<u8>>,
}

impl LocalTarMetadata {
    fn has_pending(&self) -> bool {
        self.pending_header
            || self.pax_path.is_some()
            || self.pax_size.is_some()
            || self.pax_mode.is_some()
            || self.pax_mtime.is_some()
            || self.gnu_long_name.is_some()
    }
}

pub fn write_tar_stream_archive<R: Read>(
    reader: R,
    master_key: &MasterKey,
    options: WriterOptions,
) -> Result<WrittenArchive, FormatError> {
    let mut sink = MemoryArchiveSink::default();
    let summary = write_tar_stream_archive_to_sink(reader, master_key, options, &mut sink)
        .map_err(format_error_from_archive_write_error)?;
    Ok(WrittenArchive {
        bytes: sink
            .volumes
            .first()
            .cloned()
            .ok_or(FormatError::WriterInvariant("no volumes emitted"))?,
        volumes: sink.volumes,
        bootstrap_sidecar: sink.bootstrap_sidecar,
        archive_uuid: summary.archive.archive_uuid,
        session_id: summary.archive.session_id,
        timings: summary.archive.timings,
    })
}

pub fn write_tar_stream_archive_to_sink<R, O>(
    reader: R,
    master_key: &MasterKey,
    options: WriterOptions,
    sink: &mut O,
) -> Result<StreamingTarWriterSummary, ArchiveWriteError>
where
    R: Read,
    O: ArchiveWriteSink,
{
    write_tar_stream_archive_to_sink_with_kdf_and_root_auth(
        reader,
        master_key,
        options,
        &KdfParams::Raw,
        None,
        None,
        sink,
    )
}

#[allow(clippy::too_many_arguments)]
pub fn write_sized_raw_member_archive_to_sink_with_kdf_and_root_auth<R, O>(
    mut reader: R,
    archive_path: &str,
    input_size: u64,
    master_key: &MasterKey,
    options: WriterOptions,
    kdf_params: &KdfParams,
    root_auth: Option<RootAuthWriterConfig<'_>>,
    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
    sink: &mut O,
) -> Result<StreamingRawWriterSummary, ArchiveWriteError>
where
    R: Read,
    O: ArchiveWriteSink,
{
    validate_streaming_create_writer_options(options)?;
    let archive_path = normalize_lookup_file_path(archive_path, options.max_path_length)?;
    let archive = write_ordered_parallel_stream_archive_to_sink(
        master_key,
        options,
        kdf_params,
        root_auth,
        authenticator,
        sink,
        |writer| {
            let mut payload = SizedRawPayloadReader {
                reader: &mut reader,
                remaining: input_size,
            };
            writer.write_regular_member_from_reader(
                StreamingRegularMember {
                    archive_path,
                    file_data_size: input_size,
                    mode: 0o644,
                    mtime: 0,
                },
                &mut payload,
            )?;
            if payload.remaining != 0 {
                return Err(FormatError::WriterInvariant(
                    "raw stdin payload was not fully consumed",
                )
                .into());
            }
            reject_trailing_raw_stdin_bytes(&mut reader)?;
            Ok(())
        },
    )?;
    Ok(StreamingRawWriterSummary {
        archive,
        input_bytes: input_size,
    })
}

#[allow(clippy::too_many_arguments)]
pub fn write_tar_stream_archive_to_sink_with_kdf_and_root_auth<R, O>(
    mut reader: R,
    master_key: &MasterKey,
    options: WriterOptions,
    kdf_params: &KdfParams,
    root_auth: Option<RootAuthWriterConfig<'_>>,
    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
    sink: &mut O,
) -> Result<StreamingTarWriterSummary, ArchiveWriteError>
where
    R: Read,
    O: ArchiveWriteSink,
{
    validate_streaming_create_writer_options(options)?;
    let mut input_summary = None;
    let archive = write_ordered_parallel_stream_archive_to_sink(
        master_key,
        options,
        kdf_params,
        root_auth,
        authenticator,
        sink,
        |writer| {
            let summary = stream_tar_stdin_regulars(
                &mut reader,
                options.max_path_length,
                |member, payload| {
                    writer.write_regular_member_from_reader(
                        StreamingRegularMember {
                            archive_path: member.path,
                            file_data_size: member.logical_size,
                            mode: member.mode,
                            mtime: member.mtime,
                        },
                        payload,
                    )
                },
            )?;
            input_summary = Some(summary);
            Ok(())
        },
    )?;
    let input_summary = input_summary.ok_or(FormatError::WriterInvariant(
        "streaming tar parser did not return a summary",
    ))?;
    Ok(StreamingTarWriterSummary {
        archive,
        input_member_count: input_summary.regular_file_count,
        input_tar_bytes: input_summary.input_tar_bytes,
    })
}

fn format_error_from_archive_write_error(error: ArchiveWriteError) -> FormatError {
    match error {
        ArchiveWriteError::Format(error) => error,
        ArchiveWriteError::Io(_) => FormatError::WriterInvariant("in-memory archive writer failed"),
    }
}

fn validate_streaming_create_writer_options(options: WriterOptions) -> Result<(), FormatError> {
    if options.volume_loss_tolerance != 0 {
        return Err(FormatError::WriterUnsupported(
            "streaming create cannot tolerate volume loss",
        ));
    }
    if options.target_volume_size.is_some() {
        return Err(FormatError::WriterUnsupported(
            "streaming create does not support target volume sizing",
        ));
    }
    Ok(())
}

struct SizedRawPayloadReader<'a, R: Read> {
    reader: &'a mut R,
    remaining: u64,
}

impl<R: Read> Read for SizedRawPayloadReader<'_, R> {
    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
        if out.is_empty() || self.remaining == 0 {
            return Ok(0);
        }
        let max_read = out
            .len()
            .min(usize::try_from(self.remaining).unwrap_or(usize::MAX));
        match self.reader.read(&mut out[..max_read]) {
            Ok(read) => {
                self.remaining -= read as u64;
                Ok(read)
            }
            Err(error) if error.kind() == ErrorKind::Interrupted => self.read(out),
            Err(error) => Err(error),
        }
    }
}

fn reject_trailing_raw_stdin_bytes<R: Read>(reader: &mut R) -> Result<(), ArchiveWriteError> {
    let mut extra = [0u8; 1];
    loop {
        match reader.read(&mut extra) {
            Ok(0) => return Ok(()),
            Ok(_) => {
                return Err(
                    FormatError::InvalidArchive("raw stdin exceeds declared --stdin-size").into(),
                )
            }
            Err(error) if error.kind() == ErrorKind::Interrupted => {}
            Err(error) => return Err(ArchiveWriteError::Io(error)),
        }
    }
}

fn stream_tar_stdin_regulars<R, F>(
    reader: &mut R,
    max_path_length: u32,
    mut on_regular: F,
) -> Result<TarStdinInputSummary, ArchiveWriteError>
where
    R: Read,
    F: for<'a> FnMut(
        TarStdinRegularMember,
        &mut LimitedTarPayloadReader<'a, R>,
    ) -> Result<(), ArchiveWriteError>,
{
    let mut summary = TarStdinInputSummary::default();
    let mut metadata = LocalTarMetadata::default();

    loop {
        let Some(header) = read_tar_block(reader, &mut summary.input_tar_bytes)? else {
            if metadata.has_pending() {
                return Err(
                    FormatError::InvalidArchive("tar stream ended after metadata header").into(),
                );
            }
            return Ok(summary);
        };
        if header.iter().all(|byte| *byte == 0) {
            if metadata.has_pending() {
                return Err(
                    FormatError::InvalidArchive("tar stream ended after metadata header").into(),
                );
            }
            let second = read_tar_block(reader, &mut summary.input_tar_bytes)?.ok_or(
                FormatError::InvalidArchive("tar stream ended inside end-of-archive marker"),
            )?;
            if second.iter().any(|byte| *byte != 0) {
                return Err(FormatError::InvalidArchive(
                    "tar stream has non-zero bytes after end-of-archive marker",
                )
                .into());
            }
            drain_zero_blocks_to_eof(reader, &mut summary.input_tar_bytes)?;
            return Ok(summary);
        }

        verify_tar_checksum(&header)?;
        let typeflag = header[156];
        let header_size = parse_tar_number(&header[124..136])?;
        match typeflag {
            b'x' => {
                let payload =
                    read_metadata_payload(reader, header_size, &mut summary.input_tar_bytes)?;
                metadata.pending_header = true;
                parse_pax_records(&payload, &mut metadata)?;
            }
            b'L' => {
                let payload =
                    read_metadata_payload(reader, header_size, &mut summary.input_tar_bytes)?;
                metadata.pending_header = true;
                metadata.gnu_long_name = Some(trimmed_metadata_payload(&payload));
            }
            b'K' => {
                let _payload =
                    read_metadata_payload(reader, header_size, &mut summary.input_tar_bytes)?;
                metadata.pending_header = true;
            }
            b'g' => {
                return Err(
                    FormatError::InvalidArchive("global PAX headers are not allowed").into(),
                )
            }
            b'V' | b'M' | b'N' => {
                return Err(
                    FormatError::InvalidArchive("global GNU headers are not allowed").into(),
                )
            }
            b'S' => {
                return Err(
                    FormatError::ReaderUnsupported("unsupported GNU sparse tar entry").into(),
                )
            }
            0 | b'0' | b'5' | b'1' | b'2' | b'3' | b'4' | b'6' => {
                let effective_size = metadata.pax_size.unwrap_or(header_size);
                let mode = metadata
                    .pax_mode
                    .unwrap_or(parse_tar_number(&header[100..108])? as u32);
                let mtime = metadata
                    .pax_mtime
                    .unwrap_or(parse_tar_number(&header[136..148])?);
                let path = canonical_main_path(&header, typeflag, &metadata, max_path_length)?;
                metadata = LocalTarMetadata::default();

                match typeflag {
                    0 | b'0' => {
                        let member = TarStdinRegularMember {
                            path,
                            mode,
                            mtime,
                            logical_size: effective_size,
                        };
                        {
                            let mut payload = LimitedTarPayloadReader {
                                reader,
                                remaining: effective_size,
                                input_tar_bytes: &mut summary.input_tar_bytes,
                            };
                            on_regular(member, &mut payload)?;
                            if payload.remaining != 0 {
                                return Err(FormatError::WriterInvariant(
                                    "streaming tar payload was not fully consumed",
                                )
                                .into());
                            }
                        }
                        drain_tar_padding(
                            reader,
                            padding_to_512_u64(effective_size),
                            &mut summary.input_tar_bytes,
                        )?;
                        summary.regular_file_count = checked_input_add(
                            summary.regular_file_count,
                            1,
                            "tar regular file count",
                        )?;
                    }
                    b'5' => {
                        if effective_size != 0 {
                            return Err(FormatError::InvalidArchive(
                                "non-regular tar entry has non-zero payload size",
                            )
                            .into());
                        }
                        summary.skipped_directory_count = checked_input_add(
                            summary.skipped_directory_count,
                            1,
                            "tar directory count",
                        )?;
                    }
                    _ => {
                        return Err(FormatError::WriterUnsupported(
                            "streaming tar stdin supports regular files and directory entries only",
                        )
                        .into())
                    }
                }
            }
            _ => return Err(FormatError::ReaderUnsupported("unsupported tar entry type").into()),
        }
    }
}

struct LimitedTarPayloadReader<'a, R: Read> {
    reader: &'a mut R,
    remaining: u64,
    input_tar_bytes: &'a mut u64,
}

impl<R: Read> Read for LimitedTarPayloadReader<'_, R> {
    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
        if out.is_empty() || self.remaining == 0 {
            return Ok(0);
        }
        let max_read = out
            .len()
            .min(usize::try_from(self.remaining).unwrap_or(usize::MAX));
        match self.reader.read(&mut out[..max_read]) {
            Ok(0) => Err(io::Error::new(
                ErrorKind::UnexpectedEof,
                "tar stream ended inside member payload",
            )),
            Ok(read) => {
                self.remaining -= read as u64;
                *self.input_tar_bytes =
                    checked_input_add(*self.input_tar_bytes, read as u64, "tar input size")
                        .map_err(io_error_from_format)?;
                Ok(read)
            }
            Err(error) if error.kind() == ErrorKind::Interrupted => self.read(out),
            Err(error) => Err(error),
        }
    }
}

fn read_tar_block<R: Read>(
    reader: &mut R,
    input_tar_bytes: &mut u64,
) -> Result<Option<[u8; TAR_BLOCK_LEN]>, ArchiveWriteError> {
    let mut block = [0u8; TAR_BLOCK_LEN];
    let mut filled = 0usize;
    while filled < TAR_BLOCK_LEN {
        match reader.read(&mut block[filled..]) {
            Ok(0) if filled == 0 => return Ok(None),
            Ok(0) => {
                return Err(
                    FormatError::InvalidArchive("tar stream ended inside member group").into(),
                )
            }
            Ok(read) => {
                filled += read;
                *input_tar_bytes =
                    checked_input_add(*input_tar_bytes, read as u64, "tar input size")?;
            }
            Err(error) if error.kind() == ErrorKind::Interrupted => {}
            Err(error) => return Err(ArchiveWriteError::Io(error)),
        }
    }
    Ok(Some(block))
}

fn read_metadata_payload<R: Read>(
    reader: &mut R,
    size: u64,
    input_tar_bytes: &mut u64,
) -> Result<Vec<u8>, ArchiveWriteError> {
    let len = usize::try_from(size)
        .map_err(|_| FormatError::ReaderUnsupported("tar metadata payload is too large"))?;
    if len > MAX_TAR_STDIN_METADATA_PAYLOAD_BYTES {
        return Err(
            FormatError::ReaderUnsupported("tar metadata payload exceeds streaming cap").into(),
        );
    }
    let mut payload = vec![0u8; len];
    read_exact_counted(reader, &mut payload, input_tar_bytes)?;
    drain_tar_padding(reader, padding_to_512_u64(size), input_tar_bytes)?;
    Ok(payload)
}

fn read_exact_counted<R: Read>(
    reader: &mut R,
    mut out: &mut [u8],
    input_tar_bytes: &mut u64,
) -> Result<(), ArchiveWriteError> {
    while !out.is_empty() {
        match reader.read(out) {
            Ok(0) => {
                return Err(
                    FormatError::InvalidArchive("tar stream ended inside member group").into(),
                )
            }
            Ok(read) => {
                *input_tar_bytes =
                    checked_input_add(*input_tar_bytes, read as u64, "tar input size")?;
                let remaining = out;
                out = &mut remaining[read..];
            }
            Err(error) if error.kind() == ErrorKind::Interrupted => {}
            Err(error) => return Err(ArchiveWriteError::Io(error)),
        }
    }
    Ok(())
}

fn drain_tar_padding<R: Read>(
    reader: &mut R,
    padding: u64,
    input_tar_bytes: &mut u64,
) -> Result<(), ArchiveWriteError> {
    let mut remaining = padding;
    let mut buf = [0u8; TAR_BLOCK_LEN];
    while remaining > 0 {
        let take = buf
            .len()
            .min(usize::try_from(remaining).unwrap_or(usize::MAX));
        read_exact_counted(reader, &mut buf[..take], input_tar_bytes)?;
        if buf[..take].iter().any(|byte| *byte != 0) {
            return Err(FormatError::InvalidArchive("tar member padding is non-zero").into());
        }
        remaining -= take as u64;
    }
    Ok(())
}

fn drain_zero_blocks_to_eof<R: Read>(
    reader: &mut R,
    input_tar_bytes: &mut u64,
) -> Result<(), ArchiveWriteError> {
    while let Some(block) = read_tar_block(reader, input_tar_bytes)? {
        if block.iter().any(|byte| *byte != 0) {
            return Err(FormatError::InvalidArchive(
                "tar stream has non-zero bytes after end-of-archive marker",
            )
            .into());
        }
    }
    Ok(())
}

fn canonical_main_path(
    header: &[u8],
    typeflag: u8,
    metadata: &LocalTarMetadata,
    max_path_length: u32,
) -> Result<Vec<u8>, FormatError> {
    let mut path = metadata
        .pax_path
        .clone()
        .or_else(|| metadata.gnu_long_name.clone())
        .unwrap_or_else(|| ustar_path(header));
    while path.starts_with(b"./") {
        path.drain(..2);
    }
    if typeflag == b'5' && path.ends_with(b"/") && !path.ends_with(b"//") {
        path.pop();
    }
    if typeflag == b'5' {
        validate_directory_path_bytes(&path, max_path_length)?;
    } else {
        validate_file_path_bytes(&path, max_path_length)?;
    }
    Ok(path)
}

fn ustar_path(header: &[u8]) -> Vec<u8> {
    let name = nul_trimmed(&header[0..100]);
    let prefix = nul_trimmed(&header[345..500]);
    if prefix.is_empty() {
        return name.to_vec();
    }
    let mut path = Vec::with_capacity(prefix.len() + 1 + name.len());
    path.extend_from_slice(prefix);
    path.push(b'/');
    path.extend_from_slice(name);
    path
}

fn parse_pax_records(payload: &[u8], metadata: &mut LocalTarMetadata) -> Result<(), FormatError> {
    let mut cursor = 0usize;
    while cursor < payload.len() {
        let len_digits_start = cursor;
        while cursor < payload.len() && payload[cursor].is_ascii_digit() {
            cursor += 1;
        }
        if cursor == len_digits_start || cursor >= payload.len() || payload[cursor] != b' ' {
            return Err(FormatError::InvalidArchive("malformed PAX record"));
        }
        let len = parse_decimal_usize(&payload[len_digits_start..cursor])?;
        let record_start = len_digits_start;
        let record_end = record_start
            .checked_add(len)
            .ok_or(FormatError::InvalidArchive("malformed PAX record"))?;
        if record_end > payload.len() || len < 4 {
            return Err(FormatError::InvalidArchive("malformed PAX record"));
        }
        let body_start = cursor + 1;
        let record = &payload[body_start..record_end];
        if record.last().copied() != Some(b'\n') {
            return Err(FormatError::InvalidArchive("malformed PAX record"));
        }
        let body = &record[..record.len() - 1];
        let eq = body
            .iter()
            .position(|byte| *byte == b'=')
            .ok_or(FormatError::InvalidArchive("malformed PAX record"))?;
        let key = std::str::from_utf8(&body[..eq])
            .map_err(|_| FormatError::InvalidArchive("malformed PAX key"))?;
        let value = &body[eq + 1..];
        match key {
            "path" => metadata.pax_path = Some(value.to_vec()),
            "size" => metadata.pax_size = Some(parse_decimal_u64(value)?),
            "mode" => {
                metadata.pax_mode = Some(
                    u32::try_from(parse_decimal_u64(value)?)
                        .map_err(|_| FormatError::WriterUnsupported("tar mode overflow"))?,
                )
            }
            "mtime" => metadata.pax_mtime = Some(parse_pax_mtime(value)?),
            key if key.starts_with("GNU.sparse.") => {
                return Err(FormatError::ReaderUnsupported(
                    "unsupported GNU sparse tar entry",
                ))
            }
            _ => {}
        }
        cursor = record_end;
    }
    Ok(())
}

fn verify_tar_checksum(header: &[u8]) -> Result<(), FormatError> {
    let expected = parse_tar_number(&header[148..156])?;
    let actual = header[..148]
        .iter()
        .chain(header[156..].iter())
        .fold(8u64 * 32, |sum, byte| sum + u64::from(*byte));
    if actual != expected {
        return Err(FormatError::InvalidArchive("tar header checksum mismatch"));
    }
    Ok(())
}

fn parse_tar_number(field: &[u8]) -> Result<u64, FormatError> {
    if field.first().is_some_and(|byte| byte & 0x80 != 0) {
        return parse_tar_base256(field);
    }
    parse_tar_octal(field)
}

fn parse_tar_base256(field: &[u8]) -> Result<u64, FormatError> {
    let Some(first) = field.first() else {
        return Err(FormatError::InvalidArchive("empty tar numeric field"));
    };
    if first & 0x40 != 0 {
        return Err(FormatError::ReaderUnsupported(
            "negative tar base-256 numeric fields are not supported",
        ));
    }
    let mut value = u128::from(first & 0x7f);
    for byte in &field[1..] {
        value = value
            .checked_mul(256)
            .and_then(|current| current.checked_add(u128::from(*byte)))
            .ok_or(FormatError::InvalidArchive("tar base-256 field overflow"))?;
    }
    u64::try_from(value).map_err(|_| FormatError::ReaderUnsupported("tar numeric field too large"))
}

fn parse_tar_octal(field: &[u8]) -> Result<u64, FormatError> {
    let field = field
        .split(|byte| *byte == 0)
        .next()
        .unwrap_or(field)
        .iter()
        .copied()
        .skip_while(|byte| *byte == b' ' || *byte == b'0')
        .take_while(|byte| *byte != b' ')
        .collect::<Vec<_>>();
    if field.is_empty() {
        return Ok(0);
    }
    let mut value = 0u64;
    for byte in field {
        if !(b'0'..=b'7').contains(&byte) {
            return Err(FormatError::InvalidArchive("invalid tar octal field"));
        }
        value = value
            .checked_mul(8)
            .and_then(|current| current.checked_add(u64::from(byte - b'0')))
            .ok_or(FormatError::InvalidArchive("tar octal field overflow"))?;
    }
    Ok(value)
}

fn parse_decimal_usize(bytes: &[u8]) -> Result<usize, FormatError> {
    usize::try_from(parse_decimal_u64(bytes)?)
        .map_err(|_| FormatError::InvalidArchive("decimal field overflow"))
}

fn parse_decimal_u64(bytes: &[u8]) -> Result<u64, FormatError> {
    if bytes.is_empty() {
        return Err(FormatError::InvalidArchive("malformed decimal field"));
    }
    let mut value = 0u64;
    for byte in bytes {
        if !byte.is_ascii_digit() {
            return Err(FormatError::InvalidArchive("malformed decimal field"));
        }
        value = value
            .checked_mul(10)
            .and_then(|current| current.checked_add(u64::from(byte - b'0')))
            .ok_or(FormatError::InvalidArchive("decimal field overflow"))?;
    }
    Ok(value)
}

fn parse_pax_mtime(bytes: &[u8]) -> Result<u64, FormatError> {
    let whole = bytes.split(|byte| *byte == b'.').next().unwrap_or(bytes);
    parse_decimal_u64(whole)
}

fn trimmed_metadata_payload(payload: &[u8]) -> Vec<u8> {
    let mut end = payload.len();
    while end > 0 && payload[end - 1] == 0 {
        end -= 1;
    }
    payload[..end].to_vec()
}

fn nul_trimmed(bytes: &[u8]) -> &[u8] {
    let end = bytes
        .iter()
        .position(|byte| *byte == 0)
        .unwrap_or(bytes.len());
    &bytes[..end]
}

fn padding_to_512_u64(len: u64) -> u64 {
    let remainder = len % TAR_BLOCK_LEN as u64;
    if remainder == 0 {
        0
    } else {
        TAR_BLOCK_LEN as u64 - remainder
    }
}

fn checked_input_add(lhs: u64, rhs: u64, field: &'static str) -> Result<u64, FormatError> {
    lhs.checked_add(rhs)
        .ok_or(FormatError::WriterUnsupported(field))
}

fn io_error_from_format(error: FormatError) -> io::Error {
    io::Error::new(ErrorKind::InvalidData, error.to_string())
}

#[cfg(test)]
mod tests {
    use std::io;

    use crate::crypto::MasterKey;
    use crate::format::{FormatError, BLOCK_RECORD_FRAMING_LEN, MASTER_KEY_LEN, VOLUME_HEADER_LEN};
    use crate::reader::{open_archive, open_archive_volumes};
    use crate::root_auth::data_block_merkle_leaf_hash;
    use crate::wire::{BlockRecord, CryptoHeader, VolumeHeader};
    use crate::writer::{write_archive, RegularFile};

    use super::*;

    fn master_key() -> MasterKey {
        MasterKey::from_raw_key(&[0x31; MASTER_KEY_LEN]).unwrap()
    }

    fn options() -> WriterOptions {
        WriterOptions {
            archive_uuid: Some([0x41; 16]),
            session_id: Some([0x42; 16]),
            closed_at_ns: 987_654_321,
            stripe_width: 1,
            volume_loss_tolerance: 0,
            bit_rot_buffer_pct: 0,
            index_root_fec_parity_shards: 0,
            ..WriterOptions::default()
        }
    }

    fn single_pass_equivalence_options() -> WriterOptions {
        WriterOptions {
            // The single-pass writer must commit the CryptoHeader before stdin is
            // consumed, so it predeclares the largest supported IndexRoot class.
            // Compare byte identity against the legacy writer with that same
            // declared class, not against the legacy default that can be raised
            // after payload planning.
            index_root_fec_data_shards: u16::MAX,
            ..options()
        }
    }

    fn multi_volume_options(stripe_width: u32) -> WriterOptions {
        WriterOptions {
            stripe_width,
            volume_loss_tolerance: 0,
            ..options()
        }
    }

    fn tar_stream(entries: &[(&str, &[u8])]) -> Vec<u8> {
        let mut out = Vec::new();
        for (path, data) in entries {
            out.extend_from_slice(&tar_header(path.as_bytes(), b'0', data.len() as u64));
            out.extend_from_slice(data);
            out.resize(out.len() + padding_to_512(data.len()), 0);
        }
        out.extend_from_slice(&[0u8; TAR_BLOCK_LEN * 2]);
        out
    }

    fn tar_header(path: &[u8], kind: u8, size: u64) -> [u8; TAR_BLOCK_LEN] {
        let mut header = [0u8; TAR_BLOCK_LEN];
        header[..path.len()].copy_from_slice(path);
        write_octal(&mut header[100..108], 0o644);
        write_octal(&mut header[108..116], 0);
        write_octal(&mut header[116..124], 0);
        write_octal(&mut header[124..136], size);
        write_octal(&mut header[136..148], 0);
        header[148..156].fill(b' ');
        header[156] = kind;
        header[257..263].copy_from_slice(b"ustar\0");
        header[263..265].copy_from_slice(b"00");
        let checksum = header.iter().map(|byte| *byte as u64).sum::<u64>();
        write_checksum(&mut header[148..156], checksum);
        header
    }

    fn write_octal(field: &mut [u8], value: u64) {
        let digits = format!("{value:o}");
        field.fill(0);
        let start = field.len() - 1 - digits.len();
        field[..start].fill(b'0');
        field[start..start + digits.len()].copy_from_slice(digits.as_bytes());
    }

    fn write_checksum(field: &mut [u8], value: u64) {
        let digits = format!("{value:06o}");
        field[0..6].copy_from_slice(digits.as_bytes());
        field[6] = 0;
        field[7] = b' ';
    }

    fn write_base256(field: &mut [u8], value: u64) {
        field.fill(0);
        let mut value = value;
        for byte in field.iter_mut().rev() {
            *byte = (value & 0xff) as u8;
            value >>= 8;
        }
        field[0] |= 0x80;
    }

    fn padding_to_512(len: usize) -> usize {
        let remainder = len % TAR_BLOCK_LEN;
        if remainder == 0 {
            0
        } else {
            TAR_BLOCK_LEN - remainder
        }
    }

    fn data_leaf_hash_sequence(bytes: &[u8]) -> Vec<(u64, [u8; 32])> {
        let volume_header = VolumeHeader::parse(&bytes[..VOLUME_HEADER_LEN]).unwrap();
        let crypto_start = volume_header.crypto_header_offset as usize;
        let crypto_end = crypto_start + volume_header.crypto_header_length as usize;
        let crypto_header = CryptoHeader::parse(
            &bytes[crypto_start..crypto_end],
            volume_header.crypto_header_length,
        )
        .unwrap();
        let block_size = crypto_header.fixed.block_size as usize;
        let record_len = block_size + BLOCK_RECORD_FRAMING_LEN;
        let mut offset = crypto_end;
        let mut hashes = Vec::new();
        while bytes.get(offset..offset + 4) == Some(b"TZBK") {
            let record =
                BlockRecord::parse(&bytes[offset..offset + record_len], block_size).unwrap();
            if record.kind.is_data() {
                hashes.push((
                    record.block_index,
                    data_block_merkle_leaf_hash(
                        record.block_index,
                        record.kind,
                        record.flags,
                        &record.payload,
                    ),
                ));
            }
            offset += record_len;
        }
        hashes
    }

    struct TinyReadCursor {
        data: Vec<u8>,
        cursor: usize,
        max_chunk: usize,
        reads: usize,
    }

    impl TinyReadCursor {
        fn new(data: Vec<u8>, max_chunk: usize) -> Self {
            Self {
                data,
                cursor: 0,
                max_chunk,
                reads: 0,
            }
        }
    }

    impl Read for TinyReadCursor {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            self.reads += 1;
            if self.cursor >= self.data.len() {
                return Ok(0);
            }
            let len = self
                .max_chunk
                .min(buf.len())
                .min(self.data.len() - self.cursor);
            buf[..len].copy_from_slice(&self.data[self.cursor..self.cursor + len]);
            self.cursor += len;
            Ok(len)
        }
    }

    #[test]
    fn tar_stdin_single_volume_round_trips_list_verify_and_extract() {
        let input = tar_stream(&[
            ("alpha.txt", b"alpha payload".as_slice()),
            ("dir/beta.txt", b"beta payload".as_slice()),
        ]);
        let mut reader = TinyReadCursor::new(input, 17);

        let archive = write_tar_stream_archive(&mut reader, &master_key(), options()).unwrap();

        assert!(reader.reads > 10);
        let opened = open_archive(&archive.bytes, &master_key()).unwrap();
        opened.verify().unwrap();
        let listed = opened
            .list_files()
            .unwrap()
            .into_iter()
            .map(|entry| entry.path)
            .collect::<Vec<_>>();
        assert_eq!(listed, vec!["alpha.txt", "dir/beta.txt"]);
        assert_eq!(
            opened.extract_file("dir/beta.txt").unwrap(),
            Some(b"beta payload".to_vec())
        );
    }

    #[test]
    fn sized_raw_stdin_round_trips_as_regular_tar_member_archive() {
        let input = b"raw bytes from stdin";
        let mut sink = MemoryArchiveSink::default();

        let summary = write_sized_raw_member_archive_to_sink_with_kdf_and_root_auth(
            input.as_slice(),
            "raw/data.bin",
            input.len() as u64,
            &master_key(),
            options(),
            &KdfParams::Raw,
            None,
            None,
            &mut sink,
        )
        .unwrap();

        assert_eq!(summary.input_bytes, input.len() as u64);
        let opened = open_archive(&sink.volumes[0], &master_key()).unwrap();
        opened.verify().unwrap();
        assert_eq!(
            opened.extract_file("raw/data.bin").unwrap(),
            Some(input.to_vec())
        );
    }

    #[test]
    fn sized_raw_stdin_known_size_multi_volume_round_trips() {
        let input = (0..150_000)
            .map(|index| (index % 251) as u8)
            .collect::<Vec<_>>();
        let mut sink = MemoryArchiveSink::default();

        let summary = write_sized_raw_member_archive_to_sink_with_kdf_and_root_auth(
            input.as_slice(),
            "raw/data.bin",
            input.len() as u64,
            &master_key(),
            multi_volume_options(3),
            &KdfParams::Raw,
            None,
            None,
            &mut sink,
        )
        .unwrap();

        assert_eq!(summary.archive.volume_count, 3);
        assert_eq!(sink.volumes.len(), 3);
        let refs = sink.volumes.iter().map(Vec::as_slice).collect::<Vec<_>>();
        let opened = open_archive_volumes(&refs, &master_key()).unwrap();
        opened.verify().unwrap();
        assert_eq!(
            opened.extract_file("raw/data.bin").unwrap(),
            Some(input.to_vec())
        );
    }

    #[test]
    fn sized_raw_stdin_rejects_short_input() {
        let mut sink = MemoryArchiveSink::default();

        let error = write_sized_raw_member_archive_to_sink_with_kdf_and_root_auth(
            b"short".as_slice(),
            "raw/data.bin",
            6,
            &master_key(),
            options(),
            &KdfParams::Raw,
            None,
            None,
            &mut sink,
        )
        .unwrap_err();

        assert!(matches!(error, ArchiveWriteError::Io(_)));
    }

    #[test]
    fn sized_raw_stdin_rejects_trailing_input() {
        let mut sink = MemoryArchiveSink::default();

        let error = write_sized_raw_member_archive_to_sink_with_kdf_and_root_auth(
            b"toolong".as_slice(),
            "raw/data.bin",
            3,
            &master_key(),
            options(),
            &KdfParams::Raw,
            None,
            None,
            &mut sink,
        )
        .unwrap_err();

        assert!(matches!(
            error,
            ArchiveWriteError::Format(FormatError::InvalidArchive(
                "raw stdin exceeds declared --stdin-size"
            ))
        ));
    }

    #[test]
    fn tar_stdin_sink_summary_reports_input_not_output_tar_size() {
        let input = tar_stream(&[("one.txt", b"one".as_slice())]);
        let mut sink = MemoryArchiveSink::default();

        let summary =
            write_tar_stream_archive_to_sink(&input[..], &master_key(), options(), &mut sink)
                .unwrap();

        assert_eq!(summary.input_member_count, 1);
        assert_eq!(summary.input_tar_bytes, input.len() as u64);
        assert_eq!(summary.archive.volume_count, 1);
        assert_eq!(sink.volumes.len(), 1);
        open_archive(&sink.volumes[0], &master_key())
            .unwrap()
            .verify()
            .unwrap();
    }

    #[test]
    fn tar_stdin_large_single_volume_round_trips() {
        let beta = (0..150_000)
            .map(|index| (index % 251) as u8)
            .collect::<Vec<_>>();
        let input = tar_stream(&[
            ("alpha.txt", b"alpha payload".as_slice()),
            ("dir/beta.bin", beta.as_slice()),
        ]);

        let archive = write_tar_stream_archive(&input[..], &master_key(), options()).unwrap();
        let opened = open_archive(&archive.bytes, &master_key()).unwrap();

        opened.verify().unwrap();
        assert_eq!(opened.extract_file("dir/beta.bin").unwrap(), Some(beta));
    }

    #[test]
    fn tar_stdin_multi_volume_round_trips_list_verify_and_extract() {
        let beta = (0..150_000)
            .map(|index| (index % 251) as u8)
            .collect::<Vec<_>>();
        let input = tar_stream(&[
            ("alpha.txt", b"alpha payload".as_slice()),
            ("dir/beta.bin", beta.as_slice()),
        ]);
        let mut reader = TinyReadCursor::new(input, 31);
        let mut sink = MemoryArchiveSink::default();

        let summary = write_tar_stream_archive_to_sink(
            &mut reader,
            &master_key(),
            multi_volume_options(4),
            &mut sink,
        )
        .unwrap();

        assert!(reader.reads > 10);
        assert_eq!(summary.input_member_count, 2);
        assert_eq!(summary.archive.volume_count, 4);
        assert_eq!(sink.volumes.len(), 4);
        let refs = sink.volumes.iter().map(Vec::as_slice).collect::<Vec<_>>();
        let opened = open_archive_volumes(&refs, &master_key()).unwrap();
        assert_eq!(
            opened.extract_file("alpha.txt").unwrap(),
            Some(b"alpha payload".to_vec())
        );
        assert_eq!(
            opened.extract_file("dir/beta.bin").unwrap(),
            Some(beta.clone())
        );
        opened.verify().unwrap();
        let listed = opened
            .list_files()
            .unwrap()
            .into_iter()
            .map(|entry| entry.path)
            .collect::<Vec<_>>();
        assert_eq!(listed, vec!["alpha.txt", "dir/beta.bin"]);
    }

    #[test]
    fn tar_stdin_emits_identical_bytes_to_file_list_create_with_streaming_metadata_class() {
        let input = tar_stream(&[
            ("alpha.txt", b"alpha payload".as_slice()),
            ("dir/beta.txt", b"beta payload".as_slice()),
        ]);
        let options = single_pass_equivalence_options();

        let streaming = write_tar_stream_archive(&input[..], &master_key(), options).unwrap();
        let legacy = write_archive(
            &[
                RegularFile::new("alpha.txt", b"alpha payload"),
                RegularFile::new("dir/beta.txt", b"beta payload"),
            ],
            &master_key(),
            options,
        )
        .unwrap();

        assert_eq!(streaming.bytes, legacy.bytes);
        assert_eq!(streaming.bootstrap_sidecar, legacy.bootstrap_sidecar);
    }

    #[test]
    fn tar_stdin_data_leaf_hashes_match_file_list_writer_sequence() {
        let input = tar_stream(&[
            ("alpha.txt", b"alpha payload".as_slice()),
            ("dir/beta.txt", b"beta payload".as_slice()),
        ]);
        let options = single_pass_equivalence_options();

        let streaming = write_tar_stream_archive(&input[..], &master_key(), options).unwrap();
        let legacy = write_archive(
            &[
                RegularFile::new("alpha.txt", b"alpha payload"),
                RegularFile::new("dir/beta.txt", b"beta payload"),
            ],
            &master_key(),
            options,
        )
        .unwrap();

        assert_eq!(
            data_leaf_hash_sequence(&streaming.bytes),
            data_leaf_hash_sequence(&legacy.bytes)
        );
    }

    #[test]
    fn tar_stdin_empty_stream_accepts_two_zero_eof_blocks() {
        let input = vec![0u8; TAR_BLOCK_LEN * 2];
        let mut sink = MemoryArchiveSink::default();

        let summary =
            write_tar_stream_archive_to_sink(&input[..], &master_key(), options(), &mut sink)
                .unwrap();

        assert_eq!(summary.input_member_count, 0);
        assert_eq!(summary.input_tar_bytes, input.len() as u64);
        let opened = open_archive(&sink.volumes[0], &master_key()).unwrap();
        opened.verify().unwrap();
        assert!(opened.list_files().unwrap().is_empty());
    }

    #[test]
    fn tar_stdin_accepts_base256_numeric_size_fields() {
        let mut header = tar_header(b"large-format.txt", b'0', 0);
        write_base256(&mut header[124..136], 4);
        header[148..156].fill(b' ');
        let checksum = header.iter().map(|byte| *byte as u64).sum::<u64>();
        write_checksum(&mut header[148..156], checksum);
        let mut input = Vec::new();
        input.extend_from_slice(&header);
        input.extend_from_slice(b"data");
        input.resize(input.len() + padding_to_512(4), 0);
        input.extend_from_slice(&[0u8; TAR_BLOCK_LEN * 2]);

        let archive = write_tar_stream_archive(&input[..], &master_key(), options()).unwrap();
        let opened = open_archive(&archive.bytes, &master_key()).unwrap();

        assert_eq!(
            opened.extract_file("large-format.txt").unwrap(),
            Some(b"data".to_vec())
        );
    }

    #[test]
    fn tar_stdin_rejects_metadata_header_without_following_member() {
        let payload = b"13 comment=x\n";
        let mut input = Vec::new();
        input.extend_from_slice(&tar_header(
            b"PaxHeaders/unused",
            b'x',
            payload.len() as u64,
        ));
        input.extend_from_slice(payload);
        input.resize(input.len() + padding_to_512(payload.len()), 0);
        input.extend_from_slice(&[0u8; TAR_BLOCK_LEN * 2]);

        let error = write_tar_stream_archive(&input[..], &master_key(), options()).unwrap_err();

        assert_eq!(
            error,
            FormatError::InvalidArchive("tar stream ended after metadata header")
        );
    }

    #[test]
    fn tar_stdin_skips_directory_entries_and_keeps_regular_children() {
        let mut input = Vec::new();
        input.extend_from_slice(&tar_header(b"dir/", b'5', 0));
        input.extend_from_slice(&tar_header(b"dir/file.txt", b'0', 4));
        input.extend_from_slice(b"data");
        input.resize(input.len() + padding_to_512(4), 0);
        input.extend_from_slice(&[0u8; TAR_BLOCK_LEN * 2]);
        let mut sink = MemoryArchiveSink::default();

        let summary =
            write_tar_stream_archive_to_sink(&input[..], &master_key(), options(), &mut sink)
                .unwrap();

        assert_eq!(summary.input_member_count, 1);
        let opened = open_archive(&sink.volumes[0], &master_key()).unwrap();
        assert_eq!(
            opened
                .list_files()
                .unwrap()
                .into_iter()
                .map(|entry| entry.path)
                .collect::<Vec<_>>(),
            vec!["dir/file.txt"]
        );
    }

    #[test]
    fn tar_stdin_rejects_unsupported_symlink_entries() {
        let mut input = Vec::new();
        input.extend_from_slice(&tar_header(b"link", b'2', 0));
        input.extend_from_slice(&[0u8; TAR_BLOCK_LEN * 2]);
        let mut sink = MemoryArchiveSink::default();

        let error =
            write_tar_stream_archive_to_sink(&input[..], &master_key(), options(), &mut sink)
                .unwrap_err();

        match error {
            ArchiveWriteError::Format(error) => assert_eq!(
                error,
                FormatError::WriterUnsupported(
                    "streaming tar stdin supports regular files and directory entries only"
                )
            ),
            ArchiveWriteError::Io(error) => panic!("unexpected I/O error: {error}"),
        }
        assert!(!sink.volumes.is_empty());
    }

    #[test]
    fn tar_stdin_rejects_volume_loss_tolerance() {
        let mut bad = options();
        bad.volume_loss_tolerance = 1;

        let error = write_tar_stream_archive(
            &tar_stream(&[("x", b"x".as_slice())])[..],
            &master_key(),
            bad,
        )
        .unwrap_err();

        assert_eq!(
            error,
            FormatError::WriterUnsupported("streaming create cannot tolerate volume loss")
        );
    }
}