pithos_lib 0.8.0

use super::reader::{
    AccessKeys, Archive, ArchiveFeature, EntryKind, ExternalBlockAccessPolicy,
    ExternalBlockResolver, OpenLimits, OpenOptions,
};
use crate::archive::{
    ArchivePath, ArchiveWriter, BlockKeyMode, CdcConfig, Chunking, EntryMetadata, PayloadCipher,
    ProcessingOptions, WriteOptions,
};
use crate::crypto::{self, FileKey, PrivateKey, PublicKey};
use crate::error::PithosError;
use crate::format::block::BlockLocation;
use crate::format::directory::{Directory, DirectoryEntries};
use crate::format::encryption::{EncryptionSection, RecipientData, RecipientSection};
use crate::format::file_entry::{BlockDataState, FileEntry, FileType};
use crate::format::limits::{DeserializationError, DeserializationLimits};
use crate::source::{ArchiveSource, MemorySource, SourceError};
use indexmap::IndexMap;
use std::fs::File;
use std::io::{Cursor, Write};
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};
use x25519_dalek::{PublicKey as DalekPublicKey, StaticSecret};
use zeroize::Zeroizing;

fn private(name: &str) -> PrivateKey {
    crate::crypto::parse_private_pem(
        &std::fs::read(format!("tests/data/keys/{name}_private.pem")).unwrap(),
    )
    .unwrap()
}

fn public(name: &str) -> PublicKey {
    crate::crypto::parse_public_pem(
        &std::fs::read(format!("tests/data/keys/{name}_public.pem")).unwrap(),
    )
    .unwrap()
}

fn fixture() -> (tempfile::TempDir, std::path::PathBuf) {
    fixture_with("archive reader private fixture", 0)
}

fn empty_fixture() -> (tempfile::TempDir, PathBuf) {
    let temp = tempfile::tempdir().unwrap();
    let path = temp.path().join("empty.pith");
    let writer = ArchiveWriter::create(
        File::create(&path).unwrap(),
        WriteOptions::new(private("sender"), vec![public("recipient1")]),
    )
    .unwrap();
    writer.finish().unwrap();
    (temp, path)
}

fn fixture_with(content: &str, compression_level: u8) -> (tempfile::TempDir, PathBuf) {
    fixture_with_encryption(content, compression_level, true)
}

fn fixture_with_encryption(
    content: &str,
    compression_level: u8,
    encrypt: bool,
) -> (tempfile::TempDir, PathBuf) {
    fixture_with_options(content, compression_level, encrypt, None)
}

fn fixture_with_options(
    content: &str,
    compression_level: u8,
    encrypt: bool,
    cdc: Option<CdcConfig>,
) -> (tempfile::TempDir, PathBuf) {
    let temp = tempfile::tempdir().unwrap();
    let path = temp.path().join("archive.pith");
    let options = WriteOptions::new(private("sender"), vec![public("recipient1")]);
    let options = match cdc {
        Some(cdc) => options.with_chunking(Chunking::ContentDefined(cdc)),
        None => options,
    };
    let mut writer = ArchiveWriter::create(File::create(&path).unwrap(), options).unwrap();
    writer
        .add_file(
            ArchivePath::new("data").unwrap(),
            EntryMetadata::new(0, 0, 0o644),
            ProcessingOptions::new(encrypt, compression_level).unwrap(),
            Some(content.len() as u64),
            Cursor::new(content.as_bytes()),
        )
        .unwrap();
    writer.finish().unwrap();
    (temp, path)
}

fn fixture_entries(entries: &[(&str, &str)]) -> (tempfile::TempDir, PathBuf) {
    let temp = tempfile::tempdir().unwrap();
    let path = temp.path().join("entries.pith");
    let mut writer = ArchiveWriter::create(
        File::create(&path).unwrap(),
        WriteOptions::new(private("sender"), vec![public("recipient1")]),
    )
    .unwrap();
    for (path, content) in entries {
        writer
            .add_file(
                ArchivePath::new(*path).unwrap(),
                EntryMetadata::new(0, 0, 0o644),
                ProcessingOptions::new(true, 0).unwrap(),
                Some(content.len() as u64),
                Cursor::new(content.as_bytes()),
            )
            .unwrap();
    }
    writer.finish().unwrap();
    (temp, path)
}

fn open_path(path: &Path, keys: AccessKeys) -> Archive<MemorySource> {
    Archive::open(
        MemorySource::new(Arc::<[u8]>::from(std::fs::read(path).unwrap())),
        OpenOptions::default().with_access_keys(keys),
    )
    .unwrap()
}

fn directory_bounds(bytes: &[u8]) -> (usize, usize) {
    let len = u64::from_be_bytes(bytes[bytes.len() - 12..bytes.len() - 4].try_into().unwrap());
    let len = usize::try_from(len).unwrap();
    (bytes.len() - len, len)
}

fn rewrite_terminal_directory(path: &Path, mutate: impl FnOnce(&mut Directory)) {
    let mut archive = std::fs::read(path).unwrap();
    let (start, _) = directory_bounds(&archive);
    let mut directory = crate::format::directory::decode_directory(
        &mut Cursor::new(&archive[start..]),
        &DeserializationLimits::default(),
    )
    .unwrap();
    mutate(&mut directory);
    crate::format::directory::update_directory_len(&mut directory).unwrap();
    crate::format::directory::update_directory_crc(&mut directory).unwrap();
    let mut replacement = Vec::new();
    crate::format::directory::encode_directory(&directory, &mut replacement).unwrap();
    archive.truncate(start);
    archive.extend_from_slice(&replacement);
    std::fs::write(path, archive).unwrap();
}

fn decode_terminal_directory(path: &Path) -> Directory {
    let archive = std::fs::read(path).unwrap();
    let (start, _) = directory_bounds(&archive);
    crate::format::directory::decode_directory(
        &mut Cursor::new(&archive[start..]),
        &DeserializationLimits::default(),
    )
    .unwrap()
}

fn append_empty_directory(path: &Path, relations: Option<Vec<(u64, String)>>) {
    let mut archive = std::fs::read(path).unwrap();
    let (parent_start, parent_len) = directory_bounds(&archive);
    let mut directory = Directory::new(
        Some((parent_start as u64, parent_len as u64)),
        DirectoryEntries::new(),
        IndexMap::new(),
    );
    if let Some(relations) = relations {
        directory.relations = relations;
    }
    crate::format::directory::update_directory_len(&mut directory).unwrap();
    crate::format::directory::update_directory_crc(&mut directory).unwrap();
    crate::format::directory::encode_directory(&directory, &mut archive).unwrap();
    std::fs::write(path, archive).unwrap();
}

fn append_recipient_directory(path: &Path, mutate: impl FnOnce(&mut RecipientData)) {
    let mut archive = std::fs::read(path).unwrap();
    let (parent_start, parent_len) = directory_bounds(&archive);
    let parent = crate::format::directory::decode_directory(
        &mut Cursor::new(&archive[parent_start..]),
        &DeserializationLimits::default(),
    )
    .unwrap();
    let (sender, section) = parent.encryption.first().unwrap();
    let (recipient, recipient_section) = section.recipients.first().unwrap();
    let mut recipient_data = recipient_section.recipient_data.clone();
    mutate(&mut recipient_data);
    let encryption = IndexMap::from_iter([(
        *sender,
        EncryptionSection {
            recipients: IndexMap::from_iter([(*recipient, RecipientSection { recipient_data })]),
        },
    )]);
    let mut child = Directory::new(
        Some((parent_start as u64, parent_len as u64)),
        DirectoryEntries::new(),
        encryption,
    );
    crate::format::directory::update_directory_len(&mut child).unwrap();
    crate::format::directory::update_directory_crc(&mut child).unwrap();
    crate::format::directory::encode_directory(&child, &mut archive).unwrap();
    std::fs::write(path, archive).unwrap();
}

fn set_base_recipient_records(path: &Path, records: Vec<(u64, [u8; 32])>) {
    rewrite_terminal_directory(path, |directory| {
        directory
            .encryption
            .first_mut()
            .unwrap()
            .1
            .recipients
            .first_mut()
            .unwrap()
            .1
            .recipient_data = RecipientData::Decrypted(Zeroizing::new(records));
    });
}

fn make_terminal_recipient_count_and_id_non_minimal(path: &Path, id: u8, key: u8) {
    let mut archive = std::fs::read(path).unwrap();
    let (start, _) = directory_bounds(&archive);
    let pattern = [vec![1, 1, id], vec![key; 32]].concat();
    let relative = archive[start..]
        .windows(pattern.len())
        .position(|window| window == pattern)
        .expect("terminal decrypted recipient record");
    let record = start + relative;
    archive.splice(record + 1..record + 3, [0x81, 0x00, id | 0x80, 0x00]);
    let directory_len = (archive.len() - start) as u64;
    let footer = archive.len() - 12;
    archive[footer..footer + 8].copy_from_slice(&directory_len.to_be_bytes());
    let checksum = crc32fast::hash(&archive[start..archive.len() - 4]);
    let crc_offset = archive.len() - 4;
    archive[crc_offset..].copy_from_slice(&checksum.to_be_bytes());
    std::fs::write(path, archive).unwrap();
}

fn assert_conflicting_recipient_grant(path: &Path) {
    let bytes = Arc::<[u8]>::from(std::fs::read(path).unwrap());
    for with_matching_key in [false, true] {
        let options = if with_matching_key {
            OpenOptions::default()
                .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
        } else {
            OpenOptions::default()
        };
        let error = match Archive::open(MemorySource::new(Arc::clone(&bytes)), options) {
            Ok(_) => panic!("conflicting recipient grant was accepted"),
            Err(error) => error,
        };
        assert_eq!(format!("{error:?}"), "ConflictingRecipientGrant");
    }
}

fn open_without_keys(path: &Path) -> Result<Archive<MemorySource>, PithosError> {
    Archive::open(
        MemorySource::new(Arc::<[u8]>::from(std::fs::read(path).unwrap())),
        OpenOptions::default(),
    )
}

fn first_block(path: &Path) -> (u64, u64) {
    let bytes = std::fs::read(path).unwrap();
    let (start, _) = directory_bounds(&bytes);
    let directory = crate::format::directory::decode_directory(
        &mut Cursor::new(&bytes[start..]),
        &DeserializationLimits::default(),
    )
    .unwrap();
    let block = directory.blocks.first().unwrap().1;
    (block.offset, block.stored_size)
}

fn make_block_lists_direct(path: &Path, count: usize) {
    rewrite_terminal_directory(path, |directory| {
        let blocks = directory
            .blocks
            .iter()
            .map(|(hash, descriptor)| (*hash, descriptor.original_size))
            .collect::<Vec<_>>();
        let mut index = 0usize;
        directory
            .files
            .try_for_each_mut(|_, file| {
                if index < count {
                    let (hash, original_size) = blocks[index];
                    file.block_data = BlockDataState::Decrypted(Zeroizing::new(vec![(
                        hash,
                        [index as u8 + 1; 32],
                    )]));
                    file.file_size = original_size;
                    index += 1;
                }
                Ok::<_, ()>(())
            })
            .unwrap();
        assert_eq!(index, count);
    });
}

fn recover_first_file_key(path: &Path) -> FileKey {
    let directory = decode_terminal_directory(path);
    let recipient = private("recipient1").into_dalek_static_secret();
    let recipient_public = DalekPublicKey::from(&recipient).to_bytes();
    let (sender, section) = directory.encryption.first().unwrap();
    let data = &section.recipients[&recipient_public].recipient_data;
    let RecipientData::Encrypted(data) = data else {
        panic!("writer recipient data was not encrypted");
    };
    let shared = crypto::derive_shared(recipient.as_bytes(), sender).unwrap();
    let shared = crypto::grant_wrapping_key(
        crate::format::header::FormatVersion::V1_1,
        shared,
        sender,
        &recipient_public,
        &crypto::sealed_nonce(data).unwrap(),
    );
    let plaintext = crypto::unwrap_recipient_list(&shared, data).unwrap();
    let records = crate::format::encryption::decode_decrypted_recipient_list(
        &plaintext,
        &DeserializationLimits::default(),
    )
    .unwrap();
    FileKey::from_protocol(&records[0].1)
}

fn assert_block_reference_limit<T>(result: Result<T, PithosError>) {
    assert!(matches!(
        result,
        Err(PithosError::Deserialization(
            DeserializationError::LimitExceeded {
                field: "block references",
                limit: 0,
                actual: 1,
            }
        ))
    ));
}

fn corrupt_payload(path: &Path, byte: usize) {
    let (offset, stored) = first_block(path);
    assert!(byte < stored as usize);
    let mut bytes = std::fs::read(path).unwrap();
    bytes[offset as usize + 4 + byte] ^= 1;
    std::fs::write(path, bytes).unwrap();
}

fn assert_copy_failure_without_sink(path: &Path) {
    let archive = open_path(path, AccessKeys::new().with_key(private("recipient1")));
    let mut sink = RecordingSink(Vec::new());
    assert!(archive.copy_to("data", &mut sink).is_err());
    assert!(sink.0.is_empty());
}

struct RecordingSink(Vec<u8>);

impl Write for RecordingSink {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.0.extend_from_slice(bytes);
        Ok(bytes.len())
    }
    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

struct RecordingSource {
    bytes: Arc<[u8]>,
    reads: Arc<Mutex<Vec<(u64, usize)>>>,
}

impl ArchiveSource for RecordingSource {
    fn len(&self) -> Result<u64, SourceError> {
        Ok(self.bytes.len() as u64)
    }

    fn read_exact_at(&self, offset: u64, output: &mut [u8]) -> Result<(), SourceError> {
        self.reads.lock().unwrap().push((offset, output.len()));
        MemorySource::new(Arc::clone(&self.bytes)).read_exact_at(offset, output)
    }
}

#[test]
fn writers_emit_only_the_required_standard_relationship_table() {
    let (_temporary, path) = fixture();
    let expected = [
        (0, "DESCRIBES"),
        (1, "ANNOTATES"),
        (2, "DERIVED_FROM"),
        (3, "SOURCE_OF"),
        (4, "PREVIOUS_VERSION"),
        (5, "NEXT_VERSION"),
        (6, "PART_OF"),
        (7, "CONTAINS"),
        (8, "INPUT_TO"),
        (9, "OUTPUT_FROM"),
    ];
    assert_eq!(
        decode_terminal_directory(&path)
            .relations
            .iter()
            .map(|(id, name)| (*id, name.as_str()))
            .collect::<Vec<_>>(),
        expected
    );

    append_empty_directory(&path, None);
    assert!(decode_terminal_directory(&path).relations.is_empty());
    open_without_keys(&path).unwrap();
}

#[test]
fn archive_rejects_invalid_base_standard_relationship_tables() {
    let mutations: [fn(&mut Directory); 3] = [
        |directory| directory.relations.clear(),
        |directory| directory.relations[0].1 = "describes".into(),
        |directory| directory.relations.swap(0, 1),
    ];
    for mutate in mutations {
        let (_temporary, path) = fixture();
        rewrite_terminal_directory(&path, mutate);
        assert!(open_without_keys(&path).is_err());
    }
}

#[test]
fn archive_validates_custom_relationship_ids_and_names() {
    for relation in [(10, "CUSTOM"), (999, "CUSTOM"), (1000, "")] {
        let (_temporary, path) = fixture();
        rewrite_terminal_directory(&path, |directory| {
            directory.relations.push((relation.0, relation.1.into()));
        });
        assert!(open_without_keys(&path).is_err(), "{relation:?}");
    }

    let (_temporary, path) = fixture();
    rewrite_terminal_directory(&path, |directory| {
        directory.relations.push((1000, "CUSTOM".into()));
    });
    open_without_keys(&path).unwrap();
}

#[test]
fn appended_standard_relationship_repeats_must_match() {
    let (_temporary, exact) = fixture();
    append_empty_directory(&exact, Some(vec![(0, "DESCRIBES".into())]));
    open_without_keys(&exact).unwrap();

    let (_temporary, conflicting) = fixture();
    append_empty_directory(&conflicting, Some(vec![(0, "describes".into())]));
    assert!(open_without_keys(&conflicting).is_err());
}

#[test]
fn archive_rejects_no_content_sizes_and_undefined_permission_bits() {
    let mutations: [fn(&mut crate::format::file_entry::FileEntry); 3] = [
        |entry| {
            entry.file_type = FileType::Directory;
            entry.block_data = BlockDataState::Decrypted(Vec::new().into());
            entry.file_size = 1;
            entry.symlink_target = None;
        },
        |entry| {
            entry.file_type = FileType::Symlink;
            entry.block_data = BlockDataState::Decrypted(Vec::new().into());
            entry.file_size = 1;
            entry.symlink_target = Some("target".into());
        },
        |entry| entry.permissions = 0x1000,
    ];
    for mutate in mutations {
        let (_temporary, path) = fixture();
        rewrite_terminal_directory(&path, |directory| {
            directory
                .files
                .try_for_each_mut(|_, entry| {
                    mutate(entry);
                    Ok::<_, ()>(())
                })
                .unwrap();
        });
        assert!(open_without_keys(&path).is_err());
    }
}

#[derive(Clone)]
struct CountingResolver {
    response: Arc<[u8]>,
    calls: Arc<AtomicUsize>,
    expected: Arc<Mutex<Vec<(u64, u64)>>>,
}

const INITIAL_TARGET: &str = "https://storage.test/initial";
const REDIRECT_TARGET: &str = "https://storage.test/redirect";

#[derive(Clone)]
struct RecordingPolicy {
    checks: Arc<Mutex<Vec<String>>>,
    denied: Option<&'static str>,
}

impl ExternalBlockAccessPolicy for RecordingPolicy {
    fn allows(&self, target: &str) -> bool {
        self.checks.lock().unwrap().push(target.to_owned());
        self.denied != Some(target)
    }
}

fn allowing_policy() -> Arc<dyn ExternalBlockAccessPolicy> {
    Arc::new(RecordingPolicy {
        checks: Arc::new(Mutex::new(Vec::new())),
        denied: None,
    })
}

impl ExternalBlockResolver for CountingResolver {
    fn resolve(
        &self,
        policy: &dyn ExternalBlockAccessPolicy,
        _location: &super::types::ExternalLocation,
        expected_len: u64,
        max_response_size: u64,
    ) -> Result<Vec<u8>, PithosError> {
        if !policy.allows(INITIAL_TARGET) {
            return Err(PithosError::ExternalBlockAccessDenied);
        }
        self.calls.fetch_add(1, Ordering::Relaxed);
        self.expected
            .lock()
            .unwrap()
            .push((expected_len, max_response_size));
        Ok(self.response.to_vec())
    }
}

#[derive(Clone)]
struct RedirectingResolver {
    response: Arc<[u8]>,
    accessed: Arc<Mutex<Vec<String>>>,
}

impl ExternalBlockResolver for RedirectingResolver {
    fn resolve(
        &self,
        policy: &dyn ExternalBlockAccessPolicy,
        _location: &super::types::ExternalLocation,
        _expected_len: u64,
        _max_response_size: u64,
    ) -> Result<Vec<u8>, PithosError> {
        for target in [INITIAL_TARGET, REDIRECT_TARGET] {
            if !policy.allows(target) {
                return Err(PithosError::ExternalBlockAccessDenied);
            }
            self.accessed.lock().unwrap().push(target.to_owned());
        }
        Ok(self.response.to_vec())
    }
}

fn as_external(path: &Path) -> Vec<u8> {
    let bytes = std::fs::read(path).unwrap();
    let (offset, stored) = first_block(path);
    let response = bytes[offset as usize..offset as usize + 4 + stored as usize].to_vec();
    rewrite_terminal_directory(path, |directory| {
        let block = directory.blocks.first_mut().unwrap().1;
        block.location = BlockLocation::External {
            url: "test:external".into(),
        };
        block.offset = u64::MAX;
    });
    response
}

fn last_block_as_external(path: &Path) -> Vec<u8> {
    let bytes = std::fs::read(path).unwrap();
    let directory = decode_terminal_directory(path);
    let descriptor = directory.blocks.last().unwrap().1;
    let start = descriptor.offset as usize;
    let response = bytes[start..start + 4 + descriptor.stored_size as usize].to_vec();
    rewrite_terminal_directory(path, |directory| {
        let block = directory.blocks.last_mut().unwrap().1;
        block.location = BlockLocation::External {
            url: "test:external".into(),
        };
        block.offset = u64::MAX;
    });
    response
}

struct CountingSource {
    bytes: Arc<[u8]>,
    reads: Arc<AtomicUsize>,
}
impl ArchiveSource for CountingSource {
    fn len(&self) -> Result<u64, SourceError> {
        Ok(self.bytes.len() as u64)
    }
    fn read_exact_at(&self, offset: u64, output: &mut [u8]) -> Result<(), SourceError> {
        self.reads.fetch_add(1, Ordering::Relaxed);
        let start = usize::try_from(offset).map_err(|_| SourceError::RangeOverflow {
            offset,
            length: output.len(),
        })?;
        let input =
            self.bytes
                .get(start..start + output.len())
                .ok_or(SourceError::UnexpectedEof {
                    offset,
                    expected: output.len(),
                    actual: self.bytes.len().saturating_sub(start),
                })?;
        output.copy_from_slice(input);
        Ok(())
    }
}

#[test]
fn archive_open_copy_range_unavailable_and_limits_are_boundary_checked() {
    let (_temp, path) = fixture();
    let bytes = Arc::<[u8]>::from(std::fs::read(path).unwrap());
    let reads = Arc::new(AtomicUsize::new(0));
    let archive = Archive::open(
        CountingSource {
            bytes: Arc::clone(&bytes),
            reads: Arc::clone(&reads),
        },
        OpenOptions::default().with_access_keys(AccessKeys::new().with_key(private("recipient1"))),
    )
    .unwrap();
    let mut output = Vec::new();
    archive.copy_to("data", &mut output).unwrap();
    assert_eq!(output, b"archive reader private fixture");
    let before = reads.load(Ordering::Relaxed);
    archive
        .copy_range_to("data", 0..0, &mut Vec::new())
        .unwrap();
    assert_eq!(reads.load(Ordering::Relaxed), before);
    assert!(
        archive
            .copy_range_to("data", 99..100, &mut Vec::new())
            .is_err()
    );
    assert_eq!(reads.load(Ordering::Relaxed), before);
    let unavailable = Archive::open(MemorySource::new(bytes), OpenOptions::default()).unwrap();
    assert!(matches!(
        unavailable.entries().next().unwrap().kind,
        EntryKind::File {
            available: false,
            ..
        }
    ));
}

#[test]
fn archive_opens_terminal_entries_preserving_entry_order_and_unavailable_entries() {
    let (_temporary, path) = fixture_entries(&[("first", "one"), ("second", "two")]);
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let entries = archive.entries().collect::<Vec<_>>();
    assert_eq!(
        entries
            .iter()
            .map(|entry| entry.path.as_str())
            .collect::<Vec<_>>(),
        ["first", "second"]
    );
    assert_eq!(entries[0].permissions, 0o644);
    assert_eq!(entries[0].created, 0);
    assert_eq!(entries[0].modified, 0);
    assert!(matches!(
        entries[0].kind,
        EntryKind::File {
            available: true,
            ..
        }
    ));

    let unavailable = open_path(&path, AccessKeys::new());
    assert!(matches!(
        unavailable.entries().next().unwrap().kind,
        EntryKind::File {
            available: false,
            ..
        }
    ));
}

#[test]
fn archive_distinguishes_missing_and_wrong_keys_from_matching_metadata_corruption() {
    let (_temporary, path) = fixture();
    for keys in [
        AccessKeys::new(),
        AccessKeys::new().with_key(private("recipient2")),
    ] {
        let archive = open_path(&path, keys);
        assert!(matches!(
            archive.copy_to("data", &mut Vec::new()),
            Err(PithosError::ContentUnavailable)
        ));
    }

    let (_temporary, envelope) = fixture();
    rewrite_terminal_directory(&envelope, |directory| {
        let recipient = directory
            .encryption
            .values_mut()
            .next()
            .unwrap()
            .recipients
            .values_mut()
            .next()
            .unwrap();
        if let RecipientData::Encrypted(bytes) = &mut recipient.recipient_data {
            bytes[0] ^= 1;
        }
    });
    assert!(matches!(
        Archive::open(
            MemorySource::new(Arc::<[u8]>::from(std::fs::read(envelope).unwrap())),
            OpenOptions::default()
                .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
        ),
        Err(PithosError::Crypt(_))
    ));

    let (_temporary, blocks) = fixture();
    rewrite_terminal_directory(&blocks, |directory| {
        directory
            .files
            .try_for_each_mut(|_, file| {
                if let BlockDataState::Encrypted(bytes) = &mut file.block_data {
                    bytes[0] ^= 1;
                }
                Ok::<_, ()>(())
            })
            .unwrap();
    });
    assert!(matches!(
        Archive::open(
            MemorySource::new(Arc::<[u8]>::from(std::fs::read(blocks).unwrap())),
            OpenOptions::default()
                .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
        ),
        Err(PithosError::Crypt(_))
    ));
}

#[test]
fn archive_rejects_conflicting_recovered_file_keys_at_open() {
    let (_temporary, path) = fixture();
    let reader = private("recipient1").into_dalek_static_secret();
    rewrite_terminal_directory(&path, |directory| {
        let sender = StaticSecret::from([9; 32]);
        let recipient = DalekPublicKey::from(&reader);
        let shared = crypto::derive_shared(sender.as_bytes(), recipient.as_bytes()).unwrap();
        let nonce = [5; 12];
        let key = crypto::grant_wrapping_key(
            crate::format::header::FormatVersion::V1_1,
            shared,
            &DalekPublicKey::from(&sender).to_bytes(),
            recipient.as_bytes(),
            &nonce,
        );
        let mut plaintext = vec![1, 0];
        plaintext.extend_from_slice(&[0x77; 32]);
        let encrypted = crypto::wrap_recipient_list_with_nonce(&key, &plaintext, nonce).unwrap();
        directory.encryption.insert(
            DalekPublicKey::from(&sender).to_bytes(),
            EncryptionSection {
                recipients: IndexMap::from_iter([(
                    recipient.to_bytes(),
                    RecipientSection {
                        recipient_data: RecipientData::Encrypted(encrypted),
                    },
                )]),
            },
        );
    });
    assert!(matches!(
        Archive::open(
            MemorySource::new(Arc::<[u8]>::from(std::fs::read(path).unwrap())),
            OpenOptions::default()
                .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
        ),
        Err(PithosError::ConflictingRecoveredFileKey)
    ));
}

#[test]
fn archive_validates_every_encoded_sender_and_recipient_key_without_access_keys() {
    for sender_is_invalid in [true, false] {
        let (_temporary, path) = empty_fixture();
        rewrite_terminal_directory(&path, |directory| {
            if sender_is_invalid {
                let section = directory.encryption.first().unwrap().1.clone();
                directory.encryption.clear();
                directory.encryption.insert([0; 32], section);
            } else {
                let section = directory.encryption.first_mut().unwrap().1;
                let recipient = section.recipients.first().unwrap().1.clone();
                section.recipients.clear();
                section.recipients.insert([0; 32], recipient);
            }
        });

        assert!(matches!(
            open_without_keys(&path),
            Err(PithosError::Crypt(
                crypto::CryptoError::NonContributoryPublicKey
            ))
        ));
    }
}

#[test]
fn structurally_equal_encrypted_recipient_grants_repeat_with_or_without_access_keys() {
    let (_temporary, path) = empty_fixture();
    append_recipient_directory(&path, |_| {});
    open_without_keys(&path).unwrap();
    open_path(&path, AccessKeys::new().with_key(private("recipient1")));
}

#[test]
fn encrypted_recipient_grant_conflicts_are_key_independent_and_compare_all_bytes() {
    for mutate_last_byte in [false, true] {
        let (_temporary, path) = empty_fixture();
        append_recipient_directory(&path, |data| {
            let RecipientData::Encrypted(bytes) = data else {
                panic!("writer recipient grant was not sealed");
            };
            let index = if mutate_last_byte { bytes.len() - 1 } else { 0 };
            bytes[index] ^= 1;
        });
        assert_conflicting_recipient_grant(&path);
    }
}

#[test]
fn decrypted_recipient_grants_compare_decoded_order_and_accept_non_minimal_uleb128() {
    let first = (7, [0x5a; 32]);
    let second = (8, [0xa5; 32]);

    let (_temporary, equal) = empty_fixture();
    set_base_recipient_records(&equal, vec![first]);
    append_recipient_directory(&equal, |_| {});
    make_terminal_recipient_count_and_id_non_minimal(&equal, 7, 0x5a);
    open_without_keys(&equal).unwrap();
    open_path(&equal, AccessKeys::new().with_key(private("recipient1")));

    let (_temporary, conflicting) = empty_fixture();
    set_base_recipient_records(&conflicting, vec![first, second]);
    append_recipient_directory(&conflicting, |data| {
        let RecipientData::Decrypted(records) = data else {
            panic!("test grant was unexpectedly encrypted");
        };
        records.reverse();
    });
    assert_conflicting_recipient_grant(&conflicting);
}

#[test]
fn recipient_grant_variants_must_match_across_directories() {
    let (_temporary, path) = empty_fixture();
    set_base_recipient_records(&path, Vec::new());
    append_recipient_directory(&path, |data| {
        *data = RecipientData::Encrypted(Vec::new());
    });
    assert_conflicting_recipient_grant(&path);
}

#[test]
fn archive_ranges_are_half_open_and_invalid_ranges_do_not_acquire_payload() {
    let (_temporary, path) = fixture_with("0123456789abcdef", 0);
    let bytes = Arc::<[u8]>::from(std::fs::read(path).unwrap());
    let reads = Arc::new(AtomicUsize::new(0));
    let archive = Archive::open(
        CountingSource {
            bytes,
            reads: Arc::clone(&reads),
        },
        OpenOptions::default().with_access_keys(AccessKeys::new().with_key(private("recipient1"))),
    )
    .unwrap();
    for (range, expected) in [
        (0..4, b"0123".as_slice()),
        (4..12, b"456789ab".as_slice()),
        (12..16, b"cdef".as_slice()),
        (16..16, b"".as_slice()),
    ] {
        let mut sink = Vec::new();
        archive.copy_range_to("data", range, &mut sink).unwrap();
        assert_eq!(sink, expected);
    }
    let before = reads.load(Ordering::Relaxed);
    for range in [Range { start: 8, end: 7 }, 17..17, 0..17] {
        assert!(matches!(
            archive.copy_range_to("data", range, &mut Vec::new()),
            Err(PithosError::InvalidReadRange { .. })
        ));
    }
    assert_eq!(reads.load(Ordering::Relaxed), before);
}

#[test]
fn archive_ranges_verify_and_slice_every_intersecting_multiblock_payload() {
    let content = (0..512)
        .map(|index| format!("{index:08x}-unique-cdc-payload-"))
        .collect::<String>();
    let (_temporary, path) = fixture_with_options(
        &content,
        0,
        true,
        Some(CdcConfig::new(64, 256, 1024).unwrap()),
    );
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let ranges = [
        0..31,
        91..511,
        (content.len() - 71) as u64..content.len() as u64,
    ];
    for range in ranges {
        let mut output = Vec::new();
        archive
            .copy_range_to("data", range.clone(), &mut output)
            .unwrap();
        assert_eq!(
            output,
            content.as_bytes()[range.start as usize..range.end as usize]
        );
    }
}

#[test]
fn archive_rejects_bad_nonce_tag_and_ciphertext_before_sink_output() {
    for byte in [0, 12, 28] {
        let (_temporary, path) = fixture_with(
            "encrypted payload long enough for every corruption position",
            0,
        );
        corrupt_payload(&path, byte);
        assert_copy_failure_without_sink(&path);
    }
}

#[test]
fn archive_rejects_bad_compressed_payload_before_sink_output() {
    let (_temporary, path) =
        fixture_with_encryption(&"repeated block payload ".repeat(512), 3, false);
    corrupt_payload(&path, 0);
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let mut sink = RecordingSink(Vec::new());
    let error = archive.copy_to("data", &mut sink).unwrap_err();
    assert!(sink.0.is_empty());
    assert!(error.to_string().contains("decompress block"));
    assert!(std::error::Error::source(&error).is_some());
}

#[test]
fn archive_rejects_descriptor_size_mismatches_before_sink_output() {
    let (_temporary, path) = fixture();
    rewrite_terminal_directory(&path, |directory| {
        let block = directory.blocks.first_mut().unwrap().1;
        block.original_size += 1;
        directory
            .files
            .try_for_each_mut(|_, file| {
                file.file_size += 1;
                Ok::<_, ()>(())
            })
            .unwrap();
    });
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let mut sink = RecordingSink(Vec::new());
    assert!(archive.copy_to("data", &mut sink).is_err());
    assert!(sink.0.is_empty());
}

#[test]
fn archive_rejects_plaintext_hash_mismatches_before_sink_output() {
    let (_temporary, hash) = fixture_with_encryption("plaintext hash verification", 0, false);
    corrupt_payload(&hash, 0);
    let archive = open_path(&hash, AccessKeys::new().with_key(private("recipient1")));
    let mut sink = RecordingSink(Vec::new());
    assert!(matches!(
        archive.copy_to("data", &mut sink),
        Err(PithosError::BlockHashMismatch { .. })
    ));
    assert!(sink.0.is_empty());
}

#[test]
fn archive_rejects_stored_size_failures_before_sink_output() {
    let (_temporary, stored) = fixture();
    rewrite_terminal_directory(&stored, |directory| {
        directory.blocks.first_mut().unwrap().1.stored_size -= 1
    });
    assert_copy_failure_without_sink(&stored);
}

#[test]
fn direct_block_list_keys_are_used_without_file_or_recipient_keys() {
    let (_temporary, path) = fixture_with("encrypted payload with a wrong direct key", 0);
    rewrite_terminal_directory(&path, |directory| {
        let hash = *directory.blocks.first().unwrap().0;
        directory
            .files
            .try_for_each_mut(|_, file| {
                file.block_data = BlockDataState::Decrypted(Zeroizing::new(vec![(hash, [0; 32])]));
                Ok::<_, ()>(())
            })
            .unwrap();
    });

    let archive = open_without_keys(&path).unwrap();
    let mut sink = RecordingSink(Vec::new());
    assert!(matches!(
        archive.copy_to("data", &mut sink),
        Err(PithosError::Crypt(_))
    ));
    assert!(sink.0.is_empty());
}

#[test]
fn direct_block_lists_share_one_decode_budget_within_and_across_directories() {
    let limits = OpenLimits {
        max_accessible_block_references: 1,
        ..OpenLimits::default()
    };

    let (_temporary, same_directory) = fixture_entries(&[("first", "one"), ("second", "two")]);
    make_block_lists_direct(&same_directory, 2);
    assert_block_reference_limit(Archive::open(
        MemorySource::new(Arc::<[u8]>::from(std::fs::read(&same_directory).unwrap())),
        OpenOptions::default().with_limits(limits),
    ));

    let (_temporary, appended) = fixture_with("shared direct block", 0);
    make_block_lists_direct(&appended, 1);
    let mut bytes = std::fs::read(&appended).unwrap();
    let (parent_start, parent_len) = directory_bounds(&bytes);
    let parent = decode_terminal_directory(&appended);
    let (hash, descriptor) = parent.blocks.first().unwrap();
    let mut files = DirectoryEntries::with_maximum_id(0);
    files
        .insert(
            1,
            "child",
            FileEntry {
                file_type: FileType::Data,
                block_data: BlockDataState::Decrypted(Zeroizing::new(vec![(*hash, [1; 32])])),
                created: 0,
                modified: 0,
                file_size: descriptor.original_size,
                permissions: 0o644,
                references: Vec::new(),
                symlink_target: None,
            },
        )
        .unwrap();
    let mut child = Directory::new(
        Some((parent_start as u64, parent_len as u64)),
        files,
        IndexMap::new(),
    );
    crate::format::directory::update_directory_len(&mut child).unwrap();
    crate::format::directory::update_directory_crc(&mut child).unwrap();
    crate::format::directory::encode_directory(&child, &mut bytes).unwrap();
    assert_block_reference_limit(Archive::open(
        MemorySource::new(Arc::<[u8]>::from(bytes)),
        OpenOptions::default().with_limits(limits),
    ));
}

#[test]
fn direct_and_decrypted_encrypted_block_lists_share_one_open_budget() {
    let (_temporary, path) = fixture_entries(&[("direct", "one"), ("encrypted", "two")]);
    make_block_lists_direct(&path, 1);
    let limits = OpenLimits {
        max_accessible_block_references: 1,
        ..OpenLimits::default()
    };
    assert_block_reference_limit(Archive::open(
        MemorySource::new(Arc::<[u8]>::from(std::fs::read(path).unwrap())),
        OpenOptions::default()
            .with_limits(limits)
            .with_access_keys(AccessKeys::new().with_key(private("recipient1"))),
    ));
}

#[test]
fn decrypted_encrypted_block_list_accepts_a_non_minimal_count() {
    let (_temporary, path) = fixture_with("non-minimal encrypted block list", 0);
    let file_key = recover_first_file_key(&path);
    rewrite_terminal_directory(&path, |directory| {
        directory
            .files
            .try_for_each_mut(|_, file| {
                let BlockDataState::Encrypted(encrypted) = &mut file.block_data else {
                    panic!("writer block list was not encrypted");
                };
                let plaintext = crypto::open_file_block_list(&file_key, encrypted.clone()).unwrap();
                assert_eq!(plaintext[0], 1);
                let mut non_minimal = vec![0x81, 0x00];
                non_minimal.extend_from_slice(&plaintext[1..]);
                *encrypted =
                    crypto::seal_file_block_list_with_nonce(&file_key, &non_minimal, [0x5a; 12])
                        .unwrap();
                Ok::<_, ()>(())
            })
            .unwrap();
    });

    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let mut output = Vec::new();
    archive.copy_to("data", &mut output).unwrap();
    assert_eq!(output, b"non-minimal encrypted block list");
}

#[test]
fn archive_open_reads_no_block_bytes() {
    let (_temporary, path) = fixture_with("marker validation payload", 0);
    let (offset, stored_size) = first_block(&path);
    let bytes = Arc::<[u8]>::from(std::fs::read(&path).unwrap());
    let reads = Arc::new(Mutex::new(Vec::new()));
    Archive::open(
        RecordingSource {
            bytes,
            reads: Arc::clone(&reads),
        },
        OpenOptions::default().with_access_keys(AccessKeys::new().with_key(private("recipient1"))),
    )
    .unwrap();
    let block_end = offset + 4 + stored_size;
    assert!(
        reads.lock().unwrap().iter().all(|(start, len)| {
            let end = start.saturating_add(*len as u64);
            end <= offset || *start >= block_end
        }),
        "archive open read block bytes"
    );
}

#[test]
fn missing_or_changed_local_block_markers_fail_when_the_block_is_read() {
    for missing in [false, true] {
        let (_temporary, path) = fixture_with("invalid marker payload", 0);
        if missing {
            rewrite_terminal_directory(&path, |directory| {
                let descriptor = directory.blocks.first_mut().unwrap().1;
                descriptor.offset += 1;
                descriptor.stored_size -= 1;
            });
        } else {
            let (offset, _) = first_block(&path);
            let mut bytes = std::fs::read(&path).unwrap();
            bytes[offset as usize..offset as usize + 4].copy_from_slice(b"NOPE");
            std::fs::write(&path, bytes).unwrap();
        }

        open_without_keys(&path).unwrap();
        let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
        let mut sink = RecordingSink(Vec::new());
        let error = archive.copy_to("data", &mut sink).unwrap_err();
        assert!(error.to_string().contains("block marker"), "{error}");
        assert!(sink.0.is_empty());
    }
}

#[test]
fn an_expected_digest_mismatch_fails_before_any_metadata_is_decrypted() {
    let (_temporary, path) = fixture();
    rewrite_terminal_directory(&path, |directory| {
        for section in directory.encryption.values_mut() {
            for recipient in section.recipients.values_mut() {
                let RecipientData::Encrypted(bytes) = &mut recipient.recipient_data else {
                    panic!("fixture grants are encrypted");
                };
                *bytes.last_mut().unwrap() ^= 1;
            }
        }
    });
    let digest = open_without_keys(&path).unwrap().metadata_digest();
    let open = |digest| {
        Archive::open(
            MemorySource::new(Arc::<[u8]>::from(std::fs::read(&path).unwrap())),
            OpenOptions::default()
                .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
                .with_expected_metadata_digest(digest),
        )
    };
    assert!(matches!(open(digest), Err(PithosError::Crypt(_))));
    let mut wrong = digest;
    wrong[0] ^= 1;
    assert!(matches!(
        open(wrong),
        Err(PithosError::MetadataDigestMismatch)
    ));
}

#[test]
fn encrypted_local_descriptor_stored_size_has_a_28_byte_minimum() {
    for stored_size in [0, 27] {
        let (_temporary, path) = fixture_with("encrypted descriptor size boundary", 0);
        rewrite_terminal_directory(&path, |directory| {
            let descriptor = directory.blocks.first_mut().unwrap().1;
            assert!(descriptor.flags.is_encrypted());
            descriptor.stored_size = stored_size;
        });
        let error = match open_without_keys(&path) {
            Ok(_) => panic!("encrypted descriptor with stored size {stored_size} opened"),
            Err(error) => error,
        };
        assert!(error.to_string().contains("block descriptor"));
    }

    let (_temporary, path) = fixture_with("encrypted descriptor size boundary", 0);
    rewrite_terminal_directory(&path, |directory| {
        let descriptor = directory.blocks.first_mut().unwrap().1;
        assert!(descriptor.flags.is_encrypted());
        descriptor.stored_size = 28;
    });
    open_without_keys(&path).unwrap();
}

#[test]
fn archive_external_blocks_validate_exact_framing_and_share_read_paths() {
    let (temporary, path) = fixture();
    let response = as_external(&path);
    let calls = Arc::new(AtomicUsize::new(0));
    let expected = Arc::new(Mutex::new(Vec::new()));
    let resolver = CountingResolver {
        response: Arc::from(response.clone()),
        calls: Arc::clone(&calls),
        expected: Arc::clone(&expected),
    };
    let archive = Archive::open(
        MemorySource::new(Arc::<[u8]>::from(std::fs::read(&path).unwrap())),
        OpenOptions::default()
            .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
            .with_external_resolver(resolver)
            .with_external_access_policy(allowing_policy()),
    )
    .unwrap();
    let mut full = Vec::new();
    archive.copy_to("data", &mut full).unwrap();
    let mut range = Vec::new();
    archive.copy_range_to("data", 1..4, &mut range).unwrap();
    let output = temporary.path().join("external-output");
    extract(&archive, "data", &output).unwrap();
    assert_eq!(std::fs::read(output.join("data")).unwrap(), full);
    assert_eq!(range, &full[1..4]);
    #[cfg(feature = "crypt4gh")]
    {
        let mut crypt4gh = Vec::new();
        crypt4gh::export(&archive, "data", vec![public("recipient2")], &mut crypt4gh).unwrap();
        assert!(!crypt4gh.is_empty());
        assert_eq!(calls.load(Ordering::Relaxed), 4);
    }
    let (expected_len, policy) = expected.lock().unwrap()[0];
    assert_eq!(expected_len, response.len() as u64);
    assert_eq!(policy, OpenLimits::default().max_stored_block_bytes + 4);

    let missing = Archive::open(
        MemorySource::new(Arc::<[u8]>::from(std::fs::read(&path).unwrap())),
        OpenOptions::default().with_access_keys(AccessKeys::new().with_key(private("recipient1"))),
    )
    .unwrap();
    assert!(matches!(
        missing.copy_to("data", &mut Vec::new()),
        Err(PithosError::UnsupportedFeature(
            ArchiveFeature::ExternalStorage
        ))
    ));
    let mut corrupt = response.clone();
    corrupt[4] ^= 1;
    for response in [
        vec![b'B'; 3],
        [b"BLCK".as_slice(), &response, b"x"].concat(),
        [b"NOPE".as_slice(), &response[4..]].concat(),
        corrupt,
    ] {
        let resolver = CountingResolver {
            response: Arc::from(response),
            calls: Arc::new(AtomicUsize::new(0)),
            expected: Arc::new(Mutex::new(Vec::new())),
        };
        let archive = Archive::open(
            MemorySource::new(Arc::<[u8]>::from(std::fs::read(&path).unwrap())),
            OpenOptions::default()
                .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
                .with_external_resolver(resolver)
                .with_external_access_policy(allowing_policy()),
        )
        .unwrap();
        assert!(archive.copy_to("data", &mut Vec::new()).is_err());
    }
}

#[test]
fn archive_block_limits_reject_before_local_or_external_acquisition() {
    let (_temporary, path) = fixture();
    let bytes = Arc::<[u8]>::from(std::fs::read(&path).unwrap());
    let reads = Arc::new(AtomicUsize::new(0));
    let limits = OpenLimits {
        max_stored_block_bytes: 0,
        ..OpenLimits::default()
    };
    let archive = Archive::open(
        CountingSource {
            bytes: Arc::clone(&bytes),
            reads: Arc::clone(&reads),
        },
        OpenOptions::default()
            .with_limits(limits)
            .with_access_keys(AccessKeys::new().with_key(private("recipient1"))),
    )
    .unwrap();
    let before = reads.load(Ordering::Relaxed);
    assert!(matches!(
        archive.copy_to("data", &mut Vec::new()),
        Err(PithosError::LimitExceeded {
            field: "stored block",
            ..
        })
    ));
    assert_eq!(reads.load(Ordering::Relaxed), before);

    let response = as_external(&path);
    let calls = Arc::new(AtomicUsize::new(0));
    let resolver = CountingResolver {
        response: Arc::from(response),
        calls: Arc::clone(&calls),
        expected: Arc::new(Mutex::new(Vec::new())),
    };
    let archive = Archive::open(
        MemorySource::new(Arc::<[u8]>::from(std::fs::read(path).unwrap())),
        OpenOptions::default()
            .with_limits(limits)
            .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
            .with_external_resolver(resolver)
            .with_external_access_policy(allowing_policy()),
    )
    .unwrap();
    assert!(matches!(
        archive.copy_to("data", &mut Vec::new()),
        Err(PithosError::LimitExceeded {
            field: "stored block",
            ..
        })
    ));
    assert_eq!(calls.load(Ordering::Relaxed), 0);

    let reads = Arc::new(AtomicUsize::new(0));
    let limits = OpenLimits {
        max_decoded_block_bytes: 0,
        ..OpenLimits::default()
    };
    let archive = Archive::open(
        CountingSource {
            bytes,
            reads: Arc::clone(&reads),
        },
        OpenOptions::default()
            .with_limits(limits)
            .with_access_keys(AccessKeys::new().with_key(private("recipient1"))),
    )
    .unwrap();
    let before = reads.load(Ordering::Relaxed);
    assert!(matches!(
        archive.copy_to("data", &mut Vec::new()),
        Err(PithosError::LimitExceeded {
            field: "decoded block",
            ..
        })
    ));
    assert_eq!(reads.load(Ordering::Relaxed), before);
}

#[test]
fn external_availability_requires_both_resolver_and_policy() {
    let (_temporary, path) = fixture();
    let response = as_external(&path);
    let bytes = Arc::<[u8]>::from(std::fs::read(path).unwrap());
    let calls = Arc::new(AtomicUsize::new(0));
    let resolver = CountingResolver {
        response: Arc::from(response.clone()),
        calls: Arc::clone(&calls),
        expected: Arc::new(Mutex::new(Vec::new())),
    };
    let keys = || AccessKeys::new().with_key(private("recipient1"));

    let default = Archive::open(
        MemorySource::new(Arc::clone(&bytes)),
        OpenOptions::default().with_access_keys(keys()),
    )
    .unwrap();
    assert!(matches!(
        default.entries().next().unwrap().kind,
        EntryKind::File {
            available: false,
            ..
        }
    ));
    assert!(matches!(
        default.copy_range_to("data", 0..0, &mut Vec::new()),
        Err(PithosError::UnsupportedFeature(
            ArchiveFeature::ExternalStorage
        ))
    ));

    let resolver_only = Archive::open(
        MemorySource::new(Arc::clone(&bytes)),
        OpenOptions::default()
            .with_access_keys(keys())
            .with_external_resolver(resolver.clone()),
    )
    .unwrap();
    assert!(matches!(
        resolver_only.copy_to("data", &mut Vec::new()),
        Err(PithosError::UnsupportedFeature(
            ArchiveFeature::ExternalStorage
        ))
    ));
    assert_eq!(calls.load(Ordering::Relaxed), 0);

    let policy_only = Archive::open(
        MemorySource::new(Arc::clone(&bytes)),
        OpenOptions::default()
            .with_access_keys(keys())
            .with_external_access_policy(allowing_policy()),
    )
    .unwrap();
    assert!(matches!(
        policy_only.copy_to("data", &mut Vec::new()),
        Err(PithosError::UnsupportedFeature(
            ArchiveFeature::ExternalStorage
        ))
    ));

    for options in [
        OpenOptions::default()
            .with_access_keys(keys())
            .with_external_resolver(resolver.clone())
            .with_external_access_policy(allowing_policy()),
        OpenOptions::default()
            .with_access_keys(keys())
            .with_external_access_policy(allowing_policy())
            .with_external_resolver(resolver.clone()),
    ] {
        let archive = Archive::open(MemorySource::new(Arc::clone(&bytes)), options).unwrap();
        assert!(matches!(
            archive.entries().next().unwrap().kind,
            EntryKind::File {
                available: true,
                ..
            }
        ));
        let mut output = Vec::new();
        archive.copy_to("data", &mut output).unwrap();
        assert_eq!(output, b"archive reader private fixture");
    }
}

#[test]
fn external_policy_denial_precedes_initial_and_redirect_access() {
    let (_temporary, path) = fixture();
    let response = as_external(&path);
    let bytes = Arc::<[u8]>::from(std::fs::read(path).unwrap());
    let checks = Arc::new(Mutex::new(Vec::new()));
    let calls = Arc::new(AtomicUsize::new(0));
    let archive = Archive::open(
        MemorySource::new(Arc::clone(&bytes)),
        OpenOptions::default()
            .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
            .with_external_resolver(CountingResolver {
                response: Arc::from(response.clone()),
                calls: Arc::clone(&calls),
                expected: Arc::new(Mutex::new(Vec::new())),
            })
            .with_external_access_policy(Arc::new(RecordingPolicy {
                checks: Arc::clone(&checks),
                denied: Some(INITIAL_TARGET),
            })),
    )
    .unwrap();
    assert!(matches!(
        archive.entries().next().unwrap().kind,
        EntryKind::File {
            available: true,
            ..
        }
    ));
    assert!(matches!(
        archive.copy_to("data", &mut Vec::new()),
        Err(PithosError::ExternalBlockAccessDenied)
    ));
    assert_eq!(calls.load(Ordering::Relaxed), 0);
    assert_eq!(&*checks.lock().unwrap(), &[INITIAL_TARGET]);

    let checks = Arc::new(Mutex::new(Vec::new()));
    let accessed = Arc::new(Mutex::new(Vec::new()));
    let archive = Archive::open(
        MemorySource::new(bytes),
        OpenOptions::default()
            .with_access_keys(AccessKeys::new().with_key(private("recipient1")))
            .with_external_resolver(RedirectingResolver {
                response: Arc::from(response),
                accessed: Arc::clone(&accessed),
            })
            .with_external_access_policy(Arc::new(RecordingPolicy {
                checks: Arc::clone(&checks),
                denied: Some(REDIRECT_TARGET),
            })),
    )
    .unwrap();
    assert!(matches!(
        archive.copy_to("data", &mut Vec::new()),
        Err(PithosError::ExternalBlockAccessDenied)
    ));
    assert_eq!(&*checks.lock().unwrap(), &[INITIAL_TARGET, REDIRECT_TARGET]);
    assert_eq!(&*accessed.lock().unwrap(), &[INITIAL_TARGET]);
    assert!(
        !checks
            .lock()
            .unwrap()
            .iter()
            .any(|target| target == "test:external")
    );
}

#[test]
fn unsupported_external_content_preflights_the_whole_file_and_adapters() {
    let content = format!("{}{}", "x".repeat(1024), "y".repeat(1024));
    let (_temporary, path) = fixture_with_options(
        &content,
        0,
        true,
        Some(CdcConfig::new(64, 256, 1024).unwrap()),
    );
    assert!(decode_terminal_directory(&path).blocks.len() > 1);
    let _response = last_block_as_external(&path);
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let mut sink = RecordingSink(Vec::new());
    assert!(matches!(
        archive.copy_to("data", &mut sink),
        Err(PithosError::UnsupportedFeature(
            ArchiveFeature::ExternalStorage
        ))
    ));
    assert!(sink.0.is_empty());

    #[cfg(feature = "crypt4gh")]
    {
        let mut exported = Vec::new();
        assert!(matches!(
            crypt4gh::export(&archive, "data", vec![public("recipient2")], &mut exported,),
            Err(crate::adapters::crypt4gh::Crypt4GHError::Archive {
                source: PithosError::UnsupportedFeature(ArchiveFeature::ExternalStorage),
                ..
            })
        ));
        assert!(exported.is_empty());
    }
}

#[cfg(feature = "crypt4gh")]
#[test]
fn crypt4gh_blocks_read_a_repeated_adjacent_block_once() {
    let blocks = [[0u8; 16], [1; 16], [1; 16], [1; 16], [2; 16]];
    let sender = private("sender");
    let options = WriteOptions::new(sender.duplicate(), vec![sender.public_key()])
        .with_chunking(Chunking::Fixed(16));
    let mut writer = ArchiveWriter::create(Vec::new(), options).unwrap();
    writer
        .add_file(
            ArchivePath::new("data").unwrap(),
            EntryMetadata::new(0, 0, 0o644),
            ProcessingOptions::new(true, 0).unwrap(),
            Some(80),
            Cursor::new(blocks.concat()),
        )
        .unwrap();
    let reads = Arc::new(Mutex::new(Vec::new()));
    let archive = Archive::open(
        RecordingSource {
            bytes: Arc::from(writer.finish().unwrap()),
            reads: Arc::clone(&reads),
        },
        OpenOptions::default().with_access_keys(AccessKeys::new().with_key(sender)),
    )
    .unwrap();
    let opened = reads.lock().unwrap().len();
    let (id, _) = archive.view().content_id("data").unwrap();
    let mut output = Vec::new();
    let result = archive.for_each_verified_file_block(id, |plaintext| {
        output.push(plaintext.to_vec());
        Ok::<_, ()>(())
    });
    assert!(result.is_ok());
    assert_eq!(output, blocks);
    // A, B and C are each read once.
    assert_eq!(reads.lock().unwrap().len(), opened + 3);
}

#[cfg(feature = "crypt4gh")]
use crate::adapters::crypt4gh;
use crate::fs::extract;

/// Splits the fixture's block list into pieces with the given key ids and grants the
/// `granted` piece keys to recipient1 under the version 1.1 grant rules.
fn split_into_pieces(path: &Path, key_ids: &[u64], granted: &[u64]) {
    let file_key = recover_first_file_key(path);
    rewrite_terminal_directory(path, |directory| {
        let entry = directory.files.iter().next().unwrap().2.clone();
        let BlockDataState::Encrypted(sealed) = &entry.block_data else {
            panic!("fixture block list was not encrypted");
        };
        let plaintext = crypto::open_file_block_list(&file_key, sealed.clone()).unwrap();
        let entries = crate::format::file_entry::decode_decrypted_block_list_with_budget(
            &plaintext,
            &DeserializationLimits::default(),
            &mut u64::MAX.clone(),
        )
        .unwrap();
        let chunk = entries.len().div_ceil(key_ids.len()).max(1);
        let mut pieces = Vec::new();
        let mut grants = Vec::new();
        for (index, key_id) in key_ids.iter().enumerate() {
            let start = (index * chunk).min(entries.len());
            let end = ((index + 1) * chunk).min(entries.len());
            let key = [index as u8 + 1; 32];
            let mut list = Vec::new();
            crate::format::file_entry::encode_decrypted_block_list(&entries[start..end], &mut list)
                .unwrap();
            let sealed = crypto::seal_file_block_list_with_nonce(
                &FileKey::from_protocol(&key),
                &list,
                [index as u8; 12],
            )
            .unwrap();
            pieces.push(crate::format::file_entry::BlockListPiece {
                key_id: *key_id,
                sealed,
            });
            if granted.contains(key_id) {
                grants.push((*key_id, key));
            }
        }
        directory
            .files
            .try_for_each_mut(|_, file| {
                file.block_data = BlockDataState::Pieces(pieces.clone());
                Ok::<_, ()>(())
            })
            .unwrap();

        let sender = StaticSecret::from([9; 32]);
        let sender_public = DalekPublicKey::from(&sender).to_bytes();
        let recipient = public("recipient1").into_dalek_public_key().to_bytes();
        let nonce = [6; 12];
        let key = crypto::grant_wrapping_key(
            crate::format::header::FormatVersion::V1_1,
            crypto::derive_shared(sender.as_bytes(), &recipient).unwrap(),
            &sender_public,
            &recipient,
            &nonce,
        );
        let mut records = Vec::new();
        crate::format::encryption::encode_decrypted_recipient_list(&grants, &mut records).unwrap();
        directory.encryption.insert(
            sender_public,
            EncryptionSection {
                recipients: IndexMap::from_iter([(
                    recipient,
                    RecipientSection {
                        recipient_data: RecipientData::Encrypted(
                            crypto::wrap_recipient_list_with_nonce(&key, &records, nonce).unwrap(),
                        ),
                    },
                )]),
            },
        );
    });
}

fn pieces_fixture() -> (tempfile::TempDir, PathBuf, String) {
    let content = "piece content ".repeat(300);
    let (temporary, path) = fixture_with_options(
        &content,
        0,
        true,
        Some(CdcConfig::new(64, 256, 1024).unwrap()),
    );
    (temporary, path, content)
}

fn open_pieces(path: &Path) -> Result<Archive<MemorySource>, PithosError> {
    Archive::open(
        MemorySource::new(Arc::<[u8]>::from(std::fs::read(path).unwrap())),
        OpenOptions::default().with_access_keys(AccessKeys::new().with_key(private("recipient1"))),
    )
}

#[test]
fn block_list_pieces_concatenate_in_stored_order() {
    let (_temporary, path, content) = pieces_fixture();
    split_into_pieces(&path, &[5, 6, 9], &[5, 6, 9]);
    let mut output = Vec::new();
    let archive = open_pieces(&path).unwrap();
    archive.copy_to("data", &mut output).unwrap();
    assert_eq!(output, content.as_bytes());
    // New file ids must stay above every piece key id.
    assert_eq!(
        archive.into_append_snapshot().maximum_id(),
        Some(crate::archive::FileId(9))
    );
}

#[test]
fn a_missing_piece_key_leaves_the_content_unavailable() {
    let (_temporary, path, _) = pieces_fixture();
    split_into_pieces(&path, &[5, 6], &[5]);
    let archive = open_pieces(&path).unwrap();
    assert!(matches!(
        archive.entry("data").unwrap().unwrap().kind,
        EntryKind::File {
            available: false,
            ..
        }
    ));
    assert!(archive.copy_to("data", &mut Vec::new()).is_err());
}

#[test]
fn piece_rules_reject_version_1_0_archives_and_shared_key_ids() {
    let (_temporary, path, _) = pieces_fixture();
    split_into_pieces(&path, &[5, 6], &[5, 6]);
    let mut legacy = std::fs::read(&path).unwrap();
    legacy[5] = 0x00;
    assert!(matches!(
        Archive::open(MemorySource::new(legacy), OpenOptions::default()),
        Err(PithosError::UnsupportedBlockListPieces)
    ));

    let (_temporary, path, _) = pieces_fixture();
    split_into_pieces(&path, &[0, 6], &[0, 6]);
    assert!(matches!(
        open_pieces(&path),
        Err(PithosError::PieceKeyIdConflict(0))
    ));
}

#[test]
fn a_tampered_piece_fails_before_any_content_is_indexed() {
    let (_temporary, path, _) = pieces_fixture();
    split_into_pieces(&path, &[5, 6], &[5, 6]);
    rewrite_terminal_directory(&path, |directory| {
        directory
            .files
            .try_for_each_mut(|_, file| {
                if let BlockDataState::Pieces(pieces) = &mut file.block_data {
                    let last = pieces[1].sealed.len() - 1;
                    pieces[1].sealed[last] ^= 1;
                }
                Ok::<_, ()>(())
            })
            .unwrap();
    });
    assert!(matches!(open_pieces(&path), Err(PithosError::Crypt(_))));
}

#[test]
fn an_append_changes_the_metadata_digest() {
    let (_temporary, path) = fixture();
    let before = open_path(&path, AccessKeys::new()).metadata_digest();
    append_empty_directory(&path, None);
    let after = open_path(&path, AccessKeys::new()).metadata_digest();
    assert_ne!(before, after);
    assert!(matches!(
        Archive::open(
            MemorySource::new(Arc::<[u8]>::from(std::fs::read(&path).unwrap())),
            OpenOptions::default().with_expected_metadata_digest(before),
        ),
        Err(PithosError::MetadataDigestMismatch)
    ));
}

fn fixture_with_processing(
    content: &[u8],
    processing: ProcessingOptions,
) -> (tempfile::TempDir, PathBuf) {
    let temp = tempfile::tempdir().unwrap();
    let path = temp.path().join("processing.pith");
    let options = WriteOptions::new(private("sender"), vec![public("recipient1")]).with_chunking(
        Chunking::ContentDefined(CdcConfig::new(64, 256, 1024).unwrap()),
    );
    let mut writer = ArchiveWriter::create(File::create(&path).unwrap(), options).unwrap();
    writer
        .add_file(
            ArchivePath::new("data").unwrap(),
            EntryMetadata::new(0, 0, 0o644),
            processing,
            Some(content.len() as u64),
            Cursor::new(content),
        )
        .unwrap();
    writer.finish().unwrap();
    (temp, path)
}

fn unique_processing() -> ProcessingOptions {
    ProcessingOptions::new(true, 0)
        .unwrap()
        .with_key_mode(BlockKeyMode::Unique)
        .unwrap()
}

fn aes_processing() -> ProcessingOptions {
    ProcessingOptions::new(true, 0)
        .unwrap()
        .with_cipher(PayloadCipher::Aes256Gcm)
        .unwrap()
}

#[test]
fn unique_key_blocks_round_trip_and_store_every_repeat() {
    // Zero bytes split into two equal 1024-byte chunks.
    let content = vec![0u8; 2048];
    let (_convergent_dir, convergent) =
        fixture_with_processing(&content, ProcessingOptions::new(true, 0).unwrap());
    assert_eq!(decode_terminal_directory(&convergent).blocks.len(), 1);

    let (_temporary, path) = fixture_with_processing(&content, unique_processing());
    let blocks = decode_terminal_directory(&path).blocks;
    assert_eq!(blocks.len(), 2);
    assert!(blocks.values().all(|block| block.flags.0 == 0x18));
    assert!(
        !blocks
            .keys()
            .any(|hash| *hash == *blake3::hash(&content[..1024]).as_bytes())
    );
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let mut output = Vec::new();
    archive.copy_to("data", &mut output).unwrap();
    assert_eq!(output, content);
}

#[test]
fn version_1_1_flags_are_rejected_in_version_1_0_archives() {
    for (processing, expected) in [
        (ProcessingOptions::new(true, 0).unwrap(), None),
        (unique_processing(), Some(0x18)),
        (aes_processing(), Some(0x28)),
        (
            unique_processing()
                .with_cipher(PayloadCipher::Aes256Gcm)
                .unwrap(),
            Some(0x38),
        ),
    ] {
        let (_temporary, path) = fixture_with_processing(b"version gated block", processing);
        let mut legacy = std::fs::read(&path).unwrap();
        legacy[5] = 0x00;
        let result = Archive::open(MemorySource::new(legacy), OpenOptions::default());
        match expected {
            None => assert!(result.is_ok()),
            Some(flags) => assert!(matches!(
                result,
                Err(PithosError::UnsupportedProcessingFlags(actual)) if actual == flags
            )),
        }
    }
}

#[test]
fn version_1_1_flags_require_encryption() {
    for (processing, expected) in [(unique_processing(), 0x10), (aes_processing(), 0x20)] {
        let (_temporary, path) = fixture_with_processing(b"plain block", processing);
        rewrite_terminal_directory(&path, |directory| {
            directory
                .blocks
                .first_mut()
                .unwrap()
                .1
                .flags
                .set_encryption(false);
        });
        assert!(matches!(
            open_without_keys(&path),
            Err(PithosError::ProcessingRequiresEncryption(actual)) if actual == expected
        ));
    }
}

#[test]
fn unique_key_blocks_verify_payload_and_identity_before_output() {
    let content = b"unique key payload long enough for every corruption position";
    for byte in [0, 12, 40] {
        let (_temporary, path) = fixture_with_processing(content, unique_processing());
        corrupt_payload(&path, byte);
        assert_copy_failure_without_sink(&path);
    }
    let (_temporary, path) = fixture_with_processing(content, unique_processing());
    rewrite_terminal_directory(&path, |directory| {
        directory
            .blocks
            .first_mut()
            .unwrap()
            .1
            .flags
            .set_unique_key(false);
    });
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let mut sink = RecordingSink(Vec::new());
    assert!(matches!(
        archive.copy_to("data", &mut sink),
        Err(PithosError::BlockHashMismatch { .. })
    ));
    assert!(sink.0.is_empty());
}

#[test]
fn aes_256_gcm_blocks_verify_payload_and_cipher_before_output() {
    let content = b"AES-256-GCM payload long enough for every corruption position";
    for byte in [0, 12, 40] {
        let (_temporary, path) = fixture_with_processing(content, aes_processing());
        corrupt_payload(&path, byte);
        assert_copy_failure_without_sink(&path);
    }
    let (_temporary, path) = fixture_with_processing(content, aes_processing());
    rewrite_terminal_directory(&path, |directory| {
        directory
            .blocks
            .first_mut()
            .unwrap()
            .1
            .flags
            .set_aes_256_gcm(false);
    });
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    let mut sink = RecordingSink(Vec::new());
    assert!(matches!(
        archive.copy_to("data", &mut sink),
        Err(PithosError::Crypt(_))
    ));
    assert!(sink.0.is_empty());
}

#[test]
fn ciphers_mix_per_descriptor_and_reused_blocks_keep_their_cipher() {
    let temp = tempfile::tempdir().unwrap();
    let path = temp.path().join("mixed.pith");
    let options = WriteOptions::new(private("sender"), vec![public("recipient1")]);
    let mut writer = ArchiveWriter::create(File::create(&path).unwrap(), options).unwrap();
    // Each content is below the minimum chunk size, so every file is one block.
    let shared = b"block first written with ChaCha20-Poly1305".as_slice();
    let other = b"block written with AES-256-GCM".as_slice();
    let files = [
        ("chacha", shared, ProcessingOptions::new(true, 0).unwrap()),
        ("aes-reused", shared, aes_processing()),
        ("aes", other, aes_processing()),
        (
            "unique-aes",
            shared,
            aes_processing()
                .with_key_mode(BlockKeyMode::Unique)
                .unwrap(),
        ),
    ];
    for (name, content, processing) in files {
        writer
            .add_file(
                ArchivePath::new(name).unwrap(),
                EntryMetadata::new(0, 0, 0o644),
                processing,
                None,
                Cursor::new(content),
            )
            .unwrap();
    }
    writer.finish().unwrap();

    // The AES request for the shared block reuses its ChaCha20-Poly1305 descriptor.
    let flags = decode_terminal_directory(&path)
        .blocks
        .values()
        .map(|block| block.flags.0)
        .collect::<Vec<_>>();
    assert_eq!(flags, [0x08, 0x28, 0x38]);
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    for (name, expected, _) in files {
        let mut output = Vec::new();
        archive.copy_to(name, &mut output).unwrap();
        assert_eq!(output, expected, "{name}");
    }
}

#[test]
fn version_1_1_processing_round_trips_compressed_blocks() {
    let content = b"compressible version 1.1 payload ".repeat(64);
    for (processing, extra) in [
        (aes_processing(), 0x20),
        (unique_processing(), 0x10),
        (
            unique_processing()
                .with_cipher(PayloadCipher::Aes256Gcm)
                .unwrap(),
            0x30,
        ),
    ] {
        let processing = ProcessingOptions::new(true, 3)
            .unwrap()
            .with_key_mode(processing.key_mode())
            .unwrap()
            .with_cipher(processing.cipher())
            .unwrap();
        let (_temporary, path) = fixture_with_processing(&content, processing);
        let blocks = decode_terminal_directory(&path).blocks;
        assert!(!blocks.is_empty());
        for block in blocks.values() {
            assert_eq!(block.flags.0, 0x08 | extra | 3);
            assert!(block.stored_size < block.original_size);
        }
        let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
        let mut output = Vec::new();
        archive.copy_to("data", &mut output).unwrap();
        assert_eq!(output, content);
    }
}

#[test]
fn version_1_1_appends_reuse_ciphers_and_give_unique_blocks_fresh_identities() {
    let content = b"block shared by the base archive and every append";
    let (temporary, path) =
        fixture_with_processing(content, ProcessingOptions::new(true, 0).unwrap());
    let base_blocks = decode_terminal_directory(&path).blocks;
    assert_eq!(base_blocks.len(), 1);
    assert_eq!(base_blocks[0].flags.0, 0x08);
    let mut identities = std::collections::HashSet::from([*base_blocks.keys().next().unwrap()]);
    let names = ["aes", "unique", "unique-aes"];
    for (name, processing, new_flags) in [
        (names[0], aes_processing(), None),
        (names[1], unique_processing(), Some(0x18)),
        (
            names[2],
            unique_processing()
                .with_cipher(PayloadCipher::Aes256Gcm)
                .unwrap(),
            Some(0x38),
        ),
    ] {
        let source = temporary.path().join(name);
        std::fs::write(&source, content).unwrap();
        crate::fs::append_files(
            &path,
            crate::archive::AppendOptions::new(private("recipient1"), vec![public("recipient1")])
                .with_processing(processing),
            &[source],
        )
        .unwrap();
        let blocks = decode_terminal_directory(&path).blocks;
        match new_flags {
            // The content-derived block reuses the ancestor's ChaCha20-Poly1305 descriptor.
            None => assert!(blocks.is_empty()),
            Some(flags) => {
                assert_eq!(blocks.len(), 1);
                assert_eq!(blocks[0].flags.0, flags);
                assert!(identities.insert(*blocks.keys().next().unwrap()));
            }
        }
    }
    let archive = open_path(&path, AccessKeys::new().with_key(private("recipient1")));
    for name in ["data"].into_iter().chain(names) {
        let mut output = Vec::new();
        archive.copy_to(name, &mut output).unwrap();
        assert_eq!(output, content, "{name}");
    }
}

mod read_wiping {
    use super::*;
    use crate::format::zeroing_tests::{SECRET, watch};
    use std::sync::atomic::AtomicBool;

    /// A plain archive whose file "data" has four 64-byte blocks that each hold the secret.
    pub(super) fn secret_blocks() -> Vec<u8> {
        let content: Vec<u8> = (0..4)
            .flat_map(|block| [SECRET, [0xB0 + block; 32]])
            .flatten()
            .collect();
        let options = WriteOptions::base().with_chunking(Chunking::Fixed(64));
        let mut writer = ArchiveWriter::create(Vec::new(), options).unwrap();
        writer
            .add_file(
                ArchivePath::new("data").unwrap(),
                EntryMetadata::new(0, 0, 0o644),
                ProcessingOptions::new(false, 0).unwrap(),
                Some(content.len() as u64),
                Cursor::new(content),
            )
            .unwrap();
        writer.finish().unwrap()
    }

    /// Serves `bytes`. Once `failing` is set, a read fills all but one byte and fails.
    struct Failing {
        bytes: Vec<u8>,
        failing: Arc<AtomicBool>,
    }

    impl ArchiveSource for Failing {
        fn len(&self) -> Result<u64, SourceError> {
            Ok(self.bytes.len() as u64)
        }

        fn read_exact_at(&self, offset: u64, output: &mut [u8]) -> Result<(), SourceError> {
            let start = offset as usize;
            let filled = output.len() - usize::from(self.failing.load(Ordering::Relaxed));
            output[..filled].copy_from_slice(&self.bytes[start..start + filled]);
            if filled < output.len() {
                return Err(SourceError::UnexpectedEof {
                    offset,
                    expected: output.len(),
                    actual: filled,
                });
            }
            Ok(())
        }
    }

    #[test]
    fn failed_block_read_is_wiped() {
        let failing = Arc::new(AtomicBool::new(false));
        let source = Failing {
            bytes: secret_blocks(),
            failing: Arc::clone(&failing),
        };
        let archive = Archive::open(source, OpenOptions::default()).unwrap();
        failing.store(true, Ordering::Relaxed);
        let (result, _, leaked) = watch(0, &[], || archive.copy_to("data", &mut Vec::new()));
        assert!(matches!(
            result,
            Err(PithosError::Source(SourceError::UnexpectedEof { .. }))
        ));
        assert_eq!(leaked, 0);
    }

    #[test]
    fn rejected_resolver_response_is_wiped() {
        let temporary = tempfile::tempdir().unwrap();
        let path = temporary.path().join("external.pith");
        std::fs::write(&path, secret_blocks()).unwrap();
        let mut response = as_external(&path);
        response.pop();
        let resolver = CountingResolver {
            response: Arc::from(response),
            calls: Arc::default(),
            expected: Arc::default(),
        };
        let options = OpenOptions::default()
            .with_external_resolver(resolver)
            .with_external_access_policy(allowing_policy());
        let source = MemorySource::new(std::fs::read(&path).unwrap());
        let archive = Archive::open(source, options).unwrap();
        let (result, _, leaked) = watch(0, &[], || archive.copy_to("data", &mut Vec::new()));
        assert!(matches!(result, Err(PithosError::ExternalBlockFraming(_))));
        assert_eq!(leaked, 0);
    }
}

#[cfg(feature = "async")]
mod async_wiping {
    use super::read_wiping::secret_blocks;
    use super::*;
    use crate::archive::{AsyncArchive, AsyncExternalBlockResolver, ExternalLocation};
    use crate::format::zeroing_tests::watch;
    use crate::source::AsyncArchiveSource;
    use futures_core::Stream;
    use std::future::Future;
    use std::pin::{Pin, pin};
    use std::sync::atomic::AtomicBool;
    use std::task::{Context, Poll, Waker};

    /// Serves `bytes`, but one byte short once `short` is set.
    struct Bytes {
        bytes: Vec<u8>,
        short: Arc<AtomicBool>,
    }

    impl AsyncArchiveSource for Bytes {
        async fn len(&self) -> Result<u64, SourceError> {
            Ok(self.bytes.len() as u64)
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, SourceError> {
            let end = offset + len - u64::from(self.short.load(Ordering::Relaxed));
            Ok(self.bytes[offset as usize..end as usize].to_vec())
        }
    }

    fn ready<F: Future>(future: F) -> F::Output {
        match pin!(future).poll(&mut Context::from_waker(Waker::noop())) {
            Poll::Ready(output) => output,
            Poll::Pending => panic!("an in-memory read is pending"),
        }
    }

    fn open(bytes: Vec<u8>, short: &Arc<AtomicBool>) -> AsyncArchive<Bytes> {
        let short = Arc::clone(short);
        let source = Bytes { bytes, short };
        ready(AsyncArchive::open(source, OpenOptions::default(), None)).unwrap()
    }

    /// Takes the first item of a stream over all blocks, then drops the stream.
    /// Returns the item length and the number of freed buffers that still held the secret.
    fn first_then_drop<E: AsyncExternalBlockResolver>(
        archive: &AsyncArchive<Bytes, E>,
    ) -> (Result<usize, PithosError>, usize) {
        let (first, _, leaked) = watch(64, &[], || {
            let mut stream = archive.read_range("data", 0..256).unwrap();
            let mut cx = Context::from_waker(Waker::noop());
            let Poll::Ready(Some(first)) = Pin::new(&mut stream).poll_next(&mut cx) else {
                panic!("an in-memory stream has no first item");
            };
            drop(stream);
            first.map(|chunk| Zeroizing::new(chunk).len())
        });
        (first, leaked)
    }

    #[test]
    fn dropped_stream_wipes_chunks() {
        let (first, leaked) = first_then_drop(&open(secret_blocks(), &Arc::default()));
        assert_eq!(first.unwrap(), 64);
        assert_eq!(leaked, 0);
    }

    #[test]
    fn decode_failure_wipes_chunks() {
        let mut bytes = secret_blocks();
        let second = bytes.windows(32).position(|window| window == [0xB1; 32]);
        bytes[second.unwrap()] ^= 1;
        let (first, leaked) = first_then_drop(&open(bytes, &Arc::default()));
        assert!(matches!(first, Err(PithosError::BlockHashMismatch { .. })));
        assert_eq!(leaked, 0);
    }

    #[test]
    fn short_response_is_wiped() {
        let short = Arc::new(AtomicBool::new(false));
        let archive = open(secret_blocks(), &short);
        short.store(true, Ordering::Relaxed);
        let (first, leaked) = first_then_drop(&archive);
        assert!(matches!(
            first,
            Err(PithosError::Source(SourceError::ResponseLength { .. }))
        ));
        assert_eq!(leaked, 0);
    }

    /// Answers every external block with the same response.
    struct Answer(Vec<u8>);

    impl AsyncExternalBlockResolver for Answer {
        fn resolve(
            &self,
            _policy: &dyn ExternalBlockAccessPolicy,
            _location: &ExternalLocation,
            _expected_len: u64,
            _max_response_size: u64,
        ) -> impl Future<Output = Result<Vec<u8>, PithosError>> + Send {
            std::future::ready(Ok(self.0.clone()))
        }
    }

    #[test]
    fn short_resolver_response_is_wiped() {
        let temporary = tempfile::tempdir().unwrap();
        let path = temporary.path().join("external.pith");
        std::fs::write(&path, secret_blocks()).unwrap();
        let mut response = as_external(&path);
        response.pop();
        let bytes = std::fs::read(&path).unwrap();
        let source = Bytes {
            bytes,
            short: Arc::default(),
        };
        let options = OpenOptions::default()
            .with_external_resolver(Answer(response))
            .with_external_access_policy(allowing_policy());
        let archive = ready(AsyncArchive::open(source, options, None)).unwrap();
        let (first, leaked) = first_then_drop(&archive);
        assert!(matches!(first, Err(PithosError::ExternalBlockFraming(_))));
        assert_eq!(leaked, 0);
    }
}