ferrocrypt 0.3.0

//! Replays the frozen `testvectors/wire/` conformance corpus (`FORMAT.md`
//! §12.3) through the public API.
//!
//! This is FerroCrypt's own replay layer for the corpus it publishes. It
//! validates the manifest grammar and referential integrity, checks every
//! committed digest against the bytes it names, and then asserts each case's
//! recorded outcome. Per §12.1 the assertion maps typed [`CryptoError`] values
//! to stable diagnostic-class identifiers; it never parses English display
//! text, because that text is not part of the cross-language contract. It
//! also checks that each cap a limit profile lowers above zero is exactly
//! what an accepted case of that profile needs, and that a cap lowered to zero
//! outside the small-artifact profile is needed by a case. The grammar is
//! `ferrocrypt_test_support::wire_manifest`, which the generator writes the
//! corpus through.
//!
//! The corpus is regenerated by the ignored `wire_vector_gen` unit test in the
//! library crate and committed by hand.

use std::cell::RefCell;
use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::path::{Path, PathBuf};

use ferrocrypt::{
    ArchiveLimits, CryptoError, Decryptor, FormatDefect, HeaderReadLimits, KdfLimit, KeyReadLimits,
    Passphrase, PrivateKey, ProgressEvent, PublicKey, UnsupportedVersion,
    validate_private_key_file,
};
use ferrocrypt_test_support::wire_manifest::{
    CAP_COLUMN_PREFIX, DEFAULT_LIMIT_PROFILE_ID, DIGEST_PAIRS, Row,
    SMALL_ARTIFACT_LIMIT_PROFILE_ID, cases_withdrawn_by_errata, corpus_files, corpus_revision,
    field, is_lower_hex, is_structural_corpus_file, limit_value, read_committed, read_table,
    sha3_hex, table_columns,
};

/// Capabilities this build supports beyond the `0.3.0` baseline. A
/// capability-relative case is asserted only while its capability is absent
/// (`FORMAT.md` §12.2); FerroCrypt implements none of the future versions and
/// types the corpus names, so the set is empty and every stored outcome is
/// asserted.
const DECLARED_CAPABILITIES: &[&str] = &[];

/// The compatibility baseline this build claims (`FORMAT.md` §11.4). Every
/// case must be first required by it or one of its ancestors, because
/// FerroCrypt must pass every case in the corpus it ships.
const CLAIMED_BASELINE: &str = "0.3.0";

fn corpus_root() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("testvectors/wire")
}

/// Whether the corpus is on disk at all. `Cargo.toml` excludes it from the
/// published crate, so these replays skip there; a checkout always has the
/// directory, and a file missing from inside it still fails loudly below. A
/// corpus deleted from a checkout is caught by `tools/verify_manifests.py`,
/// which CI runs as a job of its own.
fn corpus_present() -> bool {
    corpus_root().is_dir()
}

/// Collects one column into a set, refusing a repeated value. §12.3 requires
/// replay to reject duplicate IDs rather than let a later row win.
fn unique_ids(table: &[Row], name: &str, column: &str) -> BTreeSet<String> {
    let mut seen = BTreeSet::new();
    for row in table {
        let value = field(row, column).to_string();
        assert!(seen.insert(value.clone()), "{name}: {value}: duplicate id");
    }
    seen
}

/// The corpus revision that appended `row`.
fn introduced_in(row: &Row) -> u32 {
    field(row, "introduced_in_corpus_revision")
        .parse()
        .expect("introduced_in_corpus_revision is a number")
}

/// The revision each row was appended in, which §12.3 bounds by the corpus
/// revision rather than fixing at it: earlier rows stay as later ones arrive.
fn assert_revisions_within(table: &[Row], name: &str, revision: u32) {
    for row in table {
        let introduced = introduced_in(row);
        assert!(introduced >= 1, "{name}: revision 0 does not exist");
        assert!(
            introduced <= revision,
            "{name}: a row from revision {introduced} is ahead of CORPUS-REVISION {revision}"
        );
    }
}

/// Why the replay leaves a case row unasserted.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Unasserted {
    /// An erratum in effect at the current revision withdrew the case from
    /// replay (§12.3). The row itself is never removed.
    WithdrawnByErratum,
    /// The case depends on a capability this build declares (§12.2), so its
    /// stored outcome does not apply to this build.
    DeclaredCapability,
}

/// Why the replay does not assert `row`, or `None` when it does.
fn unasserted(row: &Row, withdrawn: &BTreeSet<String>) -> Option<Unasserted> {
    if withdrawn.contains(field(row, "case_id")) {
        return Some(Unasserted::WithdrawnByErratum);
    }
    let capability = field(row, "capability_id");
    (capability != "-" && DECLARED_CAPABILITIES.contains(&capability))
        .then_some(Unasserted::DeclaredCapability)
}

/// Maps a typed error to its `FORMAT.md` §12.1 diagnostic class. Structured
/// values only: the corpus contract excludes English display text.
fn diagnostic_class(error: &CryptoError) -> &'static str {
    match error {
        CryptoError::InvalidFormat(defect) => match defect {
            FormatDefect::BadMagic => "bad_magic",
            FormatDefect::Truncated => "truncated",
            FormatDefect::NotAKeyFile => "not_a_key_file",
            FormatDefect::WrongKind { .. } => "wrong_kind",
            FormatDefect::WrongKeyFileType => "wrong_key_file_type",
            FormatDefect::UnsupportedArchiveVersion { .. } => "unsupported_fca_version",
            FormatDefect::OversizedHeader { .. } => "oversized_header",
            FormatDefect::MalformedHeader => "malformed_header",
            FormatDefect::ExtTooLarge { .. } => "extension_region_too_large",
            FormatDefect::MalformedTlv => "malformed_tlv",
            FormatDefect::UnknownCriticalTag { .. } => "unknown_critical_tlv",
            FormatDefect::RecipientCountOutOfRange { .. } => "recipient_count_out_of_range",
            FormatDefect::MalformedTypeName => "malformed_type_name",
            FormatDefect::MalformedRecipientEntry => "malformed_recipient_entry",
            FormatDefect::RecipientFlagsReserved => "recipient_flags_reserved",
            FormatDefect::MalformedPayloadStream => "malformed_payload_stream",
            FormatDefect::MalformedPublicKey => "malformed_public_key",
            FormatDefect::MalformedPrivateKey => "malformed_private_key",
            other => panic!("no diagnostic class mapped for {other:?}"),
        },
        CryptoError::UnsupportedVersion(version) => match version {
            UnsupportedVersion::OlderFile { .. } | UnsupportedVersion::NewerFile { .. } => {
                "unsupported_outer_version"
            }
            UnsupportedVersion::OlderKey { .. } | UnsupportedVersion::NewerKey { .. } => {
                "unsupported_private_key_version"
            }
            UnsupportedVersion::OlderPublicKey { .. }
            | UnsupportedVersion::NewerPublicKey { .. } => "unsupported_public_key_version",
            other => panic!("no diagnostic class mapped for {other:?}"),
        },
        CryptoError::UnknownCriticalRecipient { .. } => "unknown_critical_recipient",
        CryptoError::NoSupportedRecipient => "no_supported_recipient",
        CryptoError::IncompatibleRecipients { .. } => "incompatible_recipients",
        CryptoError::RecipientUnwrapFailed { .. } => "recipient_unwrap_failed",
        CryptoError::HeaderTampered | CryptoError::HeaderMacFailedAfterUnwrap { .. } => {
            "header_authentication_failed"
        }
        CryptoError::InvalidKdfParams(_) => "invalid_kdf_parameters",
        CryptoError::KdfResourceCapExceeded { .. }
        | CryptoError::KdfTimeCostCapExceeded { .. }
        | CryptoError::KdfLanesCapExceeded { .. }
        | CryptoError::KdfWorkCapExceeded { .. }
        | CryptoError::HeaderLenCapExceeded { .. }
        | CryptoError::RecipientCountCapExceeded { .. }
        | CryptoError::RecipientBodyCapExceeded { .. }
        | CryptoError::HeaderMacWorkCapExceeded { .. }
        | CryptoError::RecipientStringCapExceeded { .. }
        | CryptoError::PrivateKeyWrappedSecretCapExceeded { .. }
        | CryptoError::ArchiveEntryCountCapExceeded { .. }
        | CryptoError::ArchiveTotalBytesCapExceeded { .. }
        | CryptoError::ArchiveManifestLenCapExceeded { .. }
        | CryptoError::ArchivePathBytesCapExceeded { .. }
        | CryptoError::ArchivePathDepthCapExceeded { .. }
        | CryptoError::ArchiveExtLenCapExceeded { .. }
        | CryptoError::ArchiveEntryExtLenCapExceeded { .. }
        | CryptoError::ArchiveTotalEntryExtCapExceeded { .. } => "resource_cap_exceeded",
        CryptoError::PayloadTampered => "payload_authentication_failed",
        CryptoError::PayloadTruncated => "payload_truncated",
        // No artifact reaches these two: one needs a reader that yields
        // bytes after reporting the end of the payload, the other a
        // payload of about 281 TB. Mapped so that a reader defect
        // reports a class rather than a panic (`FORMAT.md` §12.1).
        CryptoError::ExtraDataAfterPayload => "extra_data_after_payload",
        CryptoError::PayloadChunkCountExceeded => "payload_chunk_count_exceeded",
        CryptoError::MalformedArchive { .. } => "malformed_archive",
        CryptoError::UnsafeArchivePath { .. } => "unsafe_archive_path",
        CryptoError::InvalidArchiveTree { .. } => "invalid_archive_tree",
        CryptoError::UnsupportedKeyType { .. } => "unsupported_key_type",
        CryptoError::KeyFileUnlockFailed => "private_key_unlock_failed",
        other => panic!("no diagnostic class mapped for {other:?}"),
    }
}

/// A credential resolved from `credentials.tsv`.
enum Credential {
    Passphrase(String),
    PrivateKey { key_path: PathBuf, unlock: String },
    None,
}

/// Resolves every `credentials.tsv` row, keyed by its ID, refusing a
/// duplicate ID. Each committed file is checked against its digest as it is
/// read.
fn read_credentials(root: &Path, table: &[Row]) -> BTreeMap<String, Credential> {
    let passphrase = |row: &Row, ref_column: &str, digest_column: &str| {
        String::from_utf8(read_committed(root, row, ref_column, digest_column))
            .expect("a passphrase credential is UTF-8")
    };
    let mut credentials = BTreeMap::new();
    for row in table {
        let (id, credential) = {
            let id = field(row, "credential_id").to_string();
            let credential = match field(row, "kind") {
                "passphrase" => {
                    Credential::Passphrase(passphrase(row, "primary_ref", "primary_sha3_256"))
                }
                "private_key" => {
                    // The API opens the key file by path; its bytes are checked
                    // against the committed digest first.
                    read_committed(root, row, "primary_ref", "primary_sha3_256");
                    Credential::PrivateKey {
                        key_path: root.join(field(row, "primary_ref")),
                        unlock: passphrase(row, "secret_ref", "secret_sha3_256"),
                    }
                }
                "none" => Credential::None,
                other => panic!("{id}: unknown credential kind {other}"),
            };
            (id, credential)
        };
        assert!(
            credentials.insert(id.clone(), credential).is_none(),
            "credentials.tsv: {id}: duplicate id"
        );
    }
    credentials
}

/// One `limit-profiles.tsv` row as the public limit types a replay applies
/// (`FORMAT.md` §12.3).
#[derive(Clone, Copy)]
struct LimitProfile {
    header: HeaderReadLimits,
    kdf: KdfLimit,
    key: KeyReadLimits,
    archive: ArchiveLimits,
}

/// Builds every `limit-profiles.tsv` row, keyed by its ID, refusing a
/// duplicate ID. Each limit type is built in one expression from the
/// profile's own values, so no default survives into a replay, and every
/// limit column must have been read into one of them. A value above a
/// structural maximum was refused when the table was parsed, so no builder
/// clamps what the profile states.
fn read_limit_profiles(table: &[Row]) -> BTreeMap<String, LimitProfile> {
    let limit_columns: BTreeSet<&str> = table_columns("limit-profiles.tsv")
        .iter()
        .copied()
        .filter(|c| c.starts_with(CAP_COLUMN_PREFIX))
        .collect();
    let mut profiles = BTreeMap::new();
    for row in table {
        let id = field(row, "limit_profile_id").to_string();
        let applied = RefCell::new(BTreeSet::new());
        let value = |column: &'static str| -> u64 {
            applied.borrow_mut().insert(column);
            limit_value(row, column)
        };
        let narrow = |column: &'static str| -> u32 {
            u32::try_from(value(column))
                .unwrap_or_else(|_| panic!("{id}: {column} does not fit this build's limit type"))
        };
        let profile = LimitProfile {
            header: HeaderReadLimits::default()
                .max_header_len(narrow("max_header_len"))
                .max_recipient_count(u16::try_from(value("max_recipient_count")).unwrap_or_else(
                    |_| panic!("{id}: max_recipient_count does not fit this build's limit type"),
                ))
                .max_recipient_body_len(narrow("max_recipient_body_len"))
                .max_header_mac_work_bytes(value("max_header_mac_work_bytes")),
            kdf: KdfLimit::new(narrow("max_kdf_mem_kib"))
                .max_time_cost(narrow("max_kdf_time"))
                .max_lanes(narrow("max_kdf_lanes"))
                .max_work(value("max_kdf_work")),
            key: KeyReadLimits::default()
                .max_recipient_string_chars(narrow("max_recipient_string_chars"))
                .max_private_key_wrapped_secret_len(narrow("max_private_key_wrapped_secret_len")),
            archive: ArchiveLimits::default()
                .max_entry_count(narrow("max_entry_count"))
                .max_total_plaintext_bytes(value("max_total_plaintext_bytes"))
                .max_path_depth(narrow("max_path_depth"))
                .max_path_bytes(narrow("max_path_bytes"))
                .max_manifest_bytes(narrow("max_manifest_bytes"))
                .max_archive_ext_bytes(narrow("max_archive_ext_bytes"))
                .max_entry_ext_bytes(narrow("max_entry_ext_bytes"))
                .max_total_entry_ext_bytes(value("max_total_entry_ext_bytes"))
                .max_tlv_value_bytes(narrow("max_tlv_value_bytes")),
        };
        assert_eq!(
            *applied.borrow(),
            limit_columns,
            "{id}: every limit column must reach a limit type"
        );
        assert!(
            profiles.insert(id.clone(), profile).is_none(),
            "limit-profiles.tsv: {id}: duplicate id"
        );
    }
    profiles
}

/// The class a reader reports for a file one unit past limit `column`: a TLV
/// value over its cap is a TLV defect (`FORMAT.md` §9.12), and every other
/// cap is a resource-cap refusal.
fn cap_refusal_class(column: &str) -> &'static str {
    if column == "max_tlv_value_bytes" {
        "malformed_tlv"
    } else {
        "resource_cap_exceeded"
    }
}

/// The case types whose action applies no local cap, so their rows name no
/// limit profile (`FORMAT.md` §12.3).
fn applies_no_local_cap(case_type: &str) -> bool {
    matches!(case_type, "private_key_validate" | "stream_encrypt_kat")
}

/// `baseline` and every ancestor `baselines.tsv` records for it.
fn baseline_lineage(root: &Path, baseline: &str) -> BTreeSet<String> {
    let parents: BTreeMap<String, String> = read_table(root, "baselines.tsv")
        .iter()
        .map(|row| {
            (
                field(row, "baseline_id").to_string(),
                field(row, "parent_baseline_id").to_string(),
            )
        })
        .collect();
    let mut lineage = BTreeSet::new();
    let mut current = baseline.to_string();
    while current != "-" {
        let parent = parents
            .get(&current)
            .unwrap_or_else(|| panic!("baseline {current} is not declared"))
            .clone();
        assert!(
            lineage.insert(current),
            "the baseline lineage of {baseline} forms a cycle"
        );
        current = parent;
    }
    lineage
}

#[test]
fn wire_corpus_manifests_are_well_formed() {
    if !corpus_present() {
        return;
    }
    let root = corpus_root();
    assert_eq!(
        fs::read_to_string(root.join("SCHEMA-VERSION"))
            .expect("read SCHEMA-VERSION")
            .trim(),
        "1"
    );
    let revision = corpus_revision(&root);
    assert!(revision >= 1, "revision 0 does not exist");

    // Every committed digest must equal the digest of the bytes its reference
    // names, so a row can never describe a file the corpus does not hold.
    for (table_name, pairs) in DIGEST_PAIRS {
        for row in read_table(&root, table_name) {
            for (ref_column, digest_column) in *pairs {
                if field(&row, ref_column) == "-" {
                    assert_eq!(
                        field(&row, digest_column),
                        "-",
                        "{table_name}: '-' reference needs a '-' digest"
                    );
                    continue;
                }
                read_committed(&root, &row, ref_column, digest_column);
            }
        }
    }

    // The initial baseline row FORMAT.md §12.3 fixes for revision 1.
    let baselines = read_table(&root, "baselines.tsv");
    assert_eq!(baselines.len(), 1, "revision 1 declares one baseline");
    let baseline = &baselines[0];
    assert_eq!(field(baseline, "baseline_id"), "0.3.0");
    assert_eq!(field(baseline, "established_by_release"), "0.3.0");
    assert_eq!(field(baseline, "parent_baseline_id"), "-");

    let class_table = read_table(&root, "diagnostic-classes.tsv");
    let credential_table = read_table(&root, "credentials.tsv");
    let origin_table = read_table(&root, "origins.tsv");
    let case_table = read_table(&root, "cases.tsv");
    let errata_table = read_table(&root, "errata.tsv");
    let profile_table = read_table(&root, "limit-profiles.tsv");

    // Every identifier is declared once, and every row was appended by this
    // revision or an earlier one. The credential and profile readers refuse
    // a duplicate ID themselves; building the profiles also checks that every
    // value reaches a limit type.
    let credentials = read_credentials(&root, &credential_table);
    read_limit_profiles(&profile_table);
    let classes = unique_ids(&class_table, "diagnostic-classes.tsv", "class_id");
    let origins = unique_ids(&origin_table, "origins.tsv", "origin_id");
    let case_ids = unique_ids(&case_table, "cases.tsv", "case_id");
    unique_ids(&errata_table, "errata.tsv", "erratum_id");
    for (name, table) in [
        ("baselines.tsv", &baselines),
        ("diagnostic-classes.tsv", &class_table),
        ("credentials.tsv", &credential_table),
        ("limit-profiles.tsv", &profile_table),
        ("origins.tsv", &origin_table),
        ("cases.tsv", &case_table),
    ] {
        assert_revisions_within(table, name, revision);
    }

    // A case names only a baseline or a limit profile declared by its own
    // revision or an earlier one.
    let declared_in = |table: &[Row], id_column: &str| -> BTreeMap<String, u32> {
        table
            .iter()
            .map(|row| (field(row, id_column).to_string(), introduced_in(row)))
            .collect()
    };
    let baseline_revisions = declared_in(&baselines, "baseline_id");
    let profile_revisions = declared_in(&profile_table, "limit_profile_id");
    let mut profiles_named = BTreeSet::new();

    for row in &case_table {
        let case_id = field(row, "case_id");
        assert_ne!(
            field(row, "artifact_ref"),
            "-",
            "{case_id}: a case names its artifact"
        );
        let case_revision = introduced_in(row);
        let baseline = field(row, "first_required_by_baseline");
        let baseline_revision = baseline_revisions
            .get(baseline)
            .unwrap_or_else(|| panic!("{case_id}: baseline {baseline} undeclared"));
        assert!(
            *baseline_revision <= case_revision,
            "{case_id}: first required by a baseline declared after the case"
        );
        let profile = field(row, "limit_profile_id");
        if applies_no_local_cap(field(row, "case_type")) {
            assert_eq!(profile, "-", "{case_id}: its action applies no local cap");
        } else {
            let profile_revision = profile_revisions
                .get(profile)
                .unwrap_or_else(|| panic!("{case_id}: limit profile {profile} undeclared"));
            assert!(
                *profile_revision <= case_revision,
                "{case_id}: names a limit profile declared after the case"
            );
            profiles_named.insert(profile.to_string());
        }

        let outcome = field(row, "outcome");
        let class = field(row, "diagnostic_class");
        let condition = field(row, "condition_id");
        let expected = field(row, "expected_ref");
        match outcome {
            "reject" => {
                assert_ne!(condition, "-", "{case_id}: reject needs a condition");
                assert!(
                    classes.contains(class),
                    "{case_id}: class {class} undeclared"
                );
                assert_eq!(
                    expected, "-",
                    "{case_id}: reject carries no expected result"
                );
            }
            "accept" | "transcript_equal" => {
                assert_ne!(
                    expected, "-",
                    "{case_id}: non-reject needs an expected result"
                );
                assert_eq!(class, "-", "{case_id}: non-reject uses class '-'");
            }
            other => panic!("{case_id}: unknown outcome {other}"),
        }

        match field(row, "expectation_scope") {
            "invariant" => assert_eq!(field(row, "capability_id"), "-"),
            "capability_relative" => assert_ne!(field(row, "capability_id"), "-"),
            other => panic!("{case_id}: unknown expectation scope {other}"),
        }

        let transcript = field(row, "payload_transcript_kind");
        if field(row, "case_type") == "fcr_decrypt" {
            assert!(
                matches!(transcript, "origins" | "none"),
                "{case_id}: transcript kind"
            );
        } else {
            assert_eq!(transcript, "not_applicable", "{case_id}: transcript kind");
        }
        if transcript == "origins" {
            for origin in field(row, "payload_origin_ids").split(',') {
                assert!(
                    origins.contains(origin),
                    "{case_id}: origin {origin} undeclared"
                );
            }
        } else {
            assert_eq!(
                field(row, "payload_origin_ids"),
                "-",
                "{case_id}: origins list"
            );
        }

        if field(row, "construction") == "mutation" {
            assert_ne!(
                field(row, "parent_case_id"),
                "-",
                "{case_id}: mutation parent"
            );
        } else {
            assert_eq!(
                field(row, "parent_case_id"),
                "-",
                "{case_id}: parent only for mutations"
            );
        }

        let credential = field(row, "credential_id");
        if credential != "-" {
            assert!(
                credentials.contains_key(credential),
                "{case_id}: credential undeclared"
            );
        }
    }

    // Every origin's anchor case exists, every origin commits to its payload
    // key, a KAT origin names the key file behind that commitment, and
    // independent origins use distinct nonce prefixes.
    let mut nonces = BTreeSet::new();
    for row in &origin_table {
        let origin_id = field(row, "origin_id");
        assert!(
            case_ids.contains(field(row, "anchor_case_id")),
            "{origin_id}: anchor case is not declared"
        );
        match field(row, "origin_kind") {
            "stream_kat" => assert_ne!(
                field(row, "payload_key_ref"),
                "-",
                "{origin_id}: a KAT origin names its payload key"
            ),
            "fcr_payload" => {}
            other => panic!("{origin_id}: unknown origin kind {other}"),
        }
        assert_ne!(
            field(row, "payload_key_sha3_256"),
            "-",
            "{origin_id}: an origin commits to its payload key"
        );
        if field(row, "payload_key_ref") != "-" {
            read_committed(&root, row, "payload_key_ref", "payload_key_sha3_256");
        }
        let nonce = field(row, "stream_nonce_hex");
        assert!(
            nonce.len() == 38 && is_lower_hex(nonce),
            "{origin_id}: stream_nonce_hex is a 19-byte prefix in lowercase hexadecimal"
        );
        assert!(
            nonces.insert(nonce.to_string()),
            "{origin_id}: duplicate nonce prefix"
        );
    }

    // An erratum names an existing case, states a rationale, and takes effect
    // from a revision. §12.3 keeps the affected row and its history in place.
    for row in &errata_table {
        let erratum_id = field(row, "erratum_id");
        assert!(
            case_ids.contains(field(row, "affected_case_id")),
            "{erratum_id}: affected case is not declared"
        );
        let replacement = field(row, "replacement_case_id");
        assert!(
            replacement == "-" || case_ids.contains(replacement),
            "{erratum_id}: replacement case is not declared"
        );
        assert_ne!(
            field(row, "rationale_ref"),
            "-",
            "{erratum_id}: an erratum states its rationale"
        );
        let effective: u32 = field(row, "effective_corpus_revision")
            .parse()
            .unwrap_or_else(|_| panic!("{erratum_id}: effective_corpus_revision is not a number"));
        assert!(effective >= 1, "{erratum_id}: revision 0 does not exist");
    }

    // A profile no case names is configuration nothing is evaluated under.
    for profile in profile_revisions.keys() {
        assert!(
            profiles_named.contains(profile),
            "limit-profiles.tsv: {profile}: no case names this profile"
        );
    }

    assert_every_file_is_referenced(&root, &origin_table);
}

/// Every other file must be reachable from a manifest row. The corpus is
/// frozen and append-only, so a file no row names would be committed forever
/// with no digest behind it and no way to notice a change.
fn assert_every_file_is_referenced(root: &Path, origins: &[Row]) {
    let mut referenced = BTreeSet::new();
    for (table_name, pairs) in DIGEST_PAIRS {
        for row in read_table(root, table_name) {
            for (ref_column, _) in *pairs {
                let reference = field(&row, ref_column);
                if reference != "-" {
                    referenced.insert(reference.to_string());
                }
            }
        }
    }
    for row in origins {
        let reference = field(row, "payload_key_ref");
        if reference != "-" {
            referenced.insert(reference.to_string());
        }
    }

    for path in corpus_files(root) {
        let relative = path
            .strip_prefix(root)
            .expect("corpus file is under the corpus root")
            .to_str()
            .expect("corpus path is UTF-8")
            .replace('\\', "/");
        assert!(
            is_structural_corpus_file(&relative) || referenced.contains(&relative),
            "{relative}: no manifest row references this file"
        );
    }
}

#[test]
fn wire_corpus_cases_replay() {
    if !corpus_present() {
        return;
    }
    let root = corpus_root();
    let credentials = read_credentials(&root, &read_table(&root, "credentials.tsv"));
    let profiles = read_limit_profiles(&read_table(&root, "limit-profiles.tsv"));
    let claimed_baselines = baseline_lineage(&root, CLAIMED_BASELINE);
    let withdrawn = cases_withdrawn_by_errata(&root, corpus_revision(&root));
    let case_table = read_table(&root, "cases.tsv");
    // Every row lands in exactly one of these buckets. Counting them all is
    // what turns "some cases ran" into "no case row went unasserted".
    let mut replayed = 0usize;
    let mut deferred_to_kat_replay = 0usize;
    let mut withdrawn_by_errata = 0usize;
    let mut skipped_as_declared = 0usize;

    for row in &case_table {
        let case_id = field(row, "case_id");
        let baseline = field(row, "first_required_by_baseline");
        assert!(
            claimed_baselines.contains(baseline),
            "{case_id}: first required by baseline {baseline}, which this build does not claim"
        );
        match unasserted(row, &withdrawn) {
            Some(Unasserted::WithdrawnByErratum) => {
                withdrawn_by_errata += 1;
                continue;
            }
            Some(Unasserted::DeclaredCapability) => {
                skipped_as_declared += 1;
                continue;
            }
            None => {}
        }
        // The API reads the artifact by path; its bytes are checked against
        // the committed digest first.
        read_committed(&root, row, "artifact_ref", "artifact_sha3_256");
        let case_type = field(row, "case_type");
        let profile_id = field(row, "limit_profile_id");
        assert_eq!(
            profile_id == "-",
            applies_no_local_cap(case_type),
            "{case_id}: names a limit profile exactly when its action applies local caps"
        );
        // §12.3: every local cap is set to the profile's value, whatever the
        // case expects.
        let limits = || {
            *profiles
                .get(profile_id)
                .unwrap_or_else(|| panic!("{case_id}: limit profile {profile_id} undeclared"))
        };
        let Some(result) = replay_row(&root, row, &credentials, limits) else {
            deferred_to_kat_replay += 1;
            continue;
        };

        match field(row, "outcome") {
            "accept" => {
                let produced =
                    result.unwrap_or_else(|e| panic!("{case_id}: must be accepted, got {e:?}"));
                let expected = read_committed(&root, row, "expected_ref", "expected_sha3_256");
                assert_eq!(
                    produced, expected,
                    "{case_id}: result differs from the committed expectation"
                );
            }
            "reject" => {
                let error = result.expect_err(&format!("{case_id}: must be rejected"));
                assert_eq!(
                    diagnostic_class(&error),
                    field(row, "diagnostic_class"),
                    "{case_id}: rejected with the wrong diagnostic class ({error:?})"
                );
            }
            other => panic!("{case_id}: unexpected outcome {other}"),
        }
        replayed += 1;
    }

    // Accounting, not a floor. A row that reaches none of the four buckets —
    // a `continue` added without one, a filter that drops rows — fails here
    // instead of going quietly unasserted.
    let accounted = replayed + deferred_to_kat_replay + withdrawn_by_errata + skipped_as_declared;
    assert_eq!(
        accounted,
        case_table.len(),
        "every case row must be replayed or skipped for a reason §12.2 or §12.3 states"
    );
    // Only the STREAM known-answer rows may be deferred: raw payload
    // encryption is crate-internal, so `replay_stream_kats` covers exactly
    // this set and asserts the same count from its side. A row already
    // withdrawn or skipped above is owed to neither half, so it is not
    // counted here either.
    let stream_rows = case_table
        .iter()
        .filter(|row| field(row, "case_type") == "stream_encrypt_kat")
        .filter(|row| unasserted(row, &withdrawn).is_none())
        .count();
    assert_eq!(
        deferred_to_kat_replay, stream_rows,
        "only STREAM known-answer rows may be deferred to the crate-internal replay"
    );
    assert!(replayed > 0, "the corpus replayed no cases");
    println!(
        "wire corpus: {} case row(s) — {replayed} replayed here, {deferred_to_kat_replay} deferred to replay_stream_kats, {withdrawn_by_errata} withdrawn by errata, {skipped_as_declared} skipped as declared capabilities",
        case_table.len()
    );
}

/// Every cap a profile lowers below its default, and above zero, is one that
/// an accepted case of that profile sits exactly on: the case is accepted
/// under the profile, and lowering that cap by one more refuses it with the
/// class a reader reports for that cap. The corpus relies on this to evidence
/// each cap it lowers from the accepting side (`FORMAT.md` §12.3). A cap left
/// above what its cases need, or an artifact that no longer reaches its cap,
/// lets every accepted case through. Only the cases the replay asserts count,
/// because a withdrawn or capability-dependent case proves nothing. A cap
/// lowered to zero cannot go lower, so the next paragraph covers it.
///
/// Every cap a profile other than the small-artifact one lowers to zero is
/// one that a case of that profile needs: the case is refused with that cap's
/// class, and with the default restored it is still refused, with another
/// class. Such a cap lets a file break it together with a check that no file
/// over the default cap can break, so that the case fixes the order of the two
/// (`FORMAT.md` §12.3). In the small-artifact profile, zero is what a case
/// without that region needs.
///
/// Every cap a profile raises above its default is one that a case of that
/// profile needs: restoring the default refuses the case with the class a
/// reader reports for that cap. A profile raises a cap only so that an
/// artifact reaches a structural limit the default cap would refuse first
/// (`FORMAT.md` §12.3), so a raise that no case needs is left over from a case
/// that moved to another profile or shrank.
///
/// The small-artifact profile sets every cap to what its cases need, so it
/// must lower each one: a limit column added without a value for it keeps the
/// default and fails here.
#[test]
fn every_cap_a_profile_changes_is_what_a_case_needs() {
    if !corpus_present() {
        return;
    }
    let root = corpus_root();
    let credentials = read_credentials(&root, &read_table(&root, "credentials.tsv"));
    let withdrawn = cases_withdrawn_by_errata(&root, corpus_revision(&root));
    let cases = read_table(&root, "cases.tsv");
    let profile_table = read_table(&root, "limit-profiles.tsv");
    let profiles = read_limit_profiles(&profile_table);
    let profile_row = |id: &str| -> &Row {
        profile_table
            .iter()
            .find(|row| field(row, "limit_profile_id") == id)
            .unwrap_or_else(|| panic!("limit-profiles.tsv declares no {id}"))
    };
    let default = profile_row(DEFAULT_LIMIT_PROFILE_ID);
    let columns: Vec<&str> = table_columns("limit-profiles.tsv")
        .iter()
        .copied()
        .filter(|column| column.starts_with(CAP_COLUMN_PREFIX))
        .collect();

    let small = profile_row(SMALL_ARTIFACT_LIMIT_PROFILE_ID);
    for column in &columns {
        assert!(
            limit_value(small, column) < limit_value(default, column),
            "{SMALL_ARTIFACT_LIMIT_PROFILE_ID}: {column} keeps its default"
        );
    }

    let mut lowered_caps = 0usize;
    for profile in &profile_table {
        let id = field(profile, "limit_profile_id");
        let lowered: Vec<(&str, u64)> = columns
            .iter()
            .map(|column| (*column, limit_value(profile, column)))
            .filter(|(column, value)| *value > 0 && *value < limit_value(default, column))
            .collect();
        if lowered.is_empty() {
            continue;
        }
        let accepted: Vec<&Row> = cases
            .iter()
            .filter(|row| field(row, "limit_profile_id") == id)
            .filter(|row| field(row, "outcome") == "accept")
            .filter(|row| unasserted(row, &withdrawn).is_none())
            .collect();
        for row in &accepted {
            let case_id = field(row, "case_id");
            read_committed(&root, row, "artifact_ref", "artifact_sha3_256");
            let result = replay_row(&root, row, &credentials, || profiles[id])
                .unwrap_or_else(|| panic!("{case_id}: not replayed through the public API"));
            if let Err(error) = result {
                panic!("{case_id}: must be accepted under {id}, got {error:?}");
            }
        }
        for (column, value) in lowered {
            let mut tighter = profile.clone();
            tighter.insert(column.to_string(), (value - 1).to_string());
            let tighter = read_limit_profiles(std::slice::from_ref(&tighter))[id];
            let class = cap_refusal_class(column);
            let refused = accepted.iter().any(|row| {
                matches!(
                    replay_row(&root, row, &credentials, || tighter),
                    Some(Err(error)) if diagnostic_class(&error) == class
                )
            });
            assert!(
                refused,
                "{id}: no accepted case needs {column} as high as {value}"
            );
            lowered_caps += 1;
        }
    }
    assert!(lowered_caps > 0, "no profile lowers a cap");

    // The limits of `profile` with the cap `column` back at its default.
    let with_default_cap = |profile: &Row, column: &str| {
        let mut restored = profile.clone();
        restored.insert(column.to_string(), field(default, column).to_string());
        read_limit_profiles(std::slice::from_ref(&restored))[field(profile, "limit_profile_id")]
    };
    // The class `row` is refused with under `limits`, if it is refused.
    let refused_with =
        |row: &Row, limits: LimitProfile| match replay_row(&root, row, &credentials, || limits) {
            Some(Err(error)) => Some(diagnostic_class(&error)),
            _ => None,
        };

    // Caps whose default a case of the profile must not survive: those raised
    // above it, and those lowered to zero outside the small-artifact profile.
    let mut raised_caps = 0usize;
    let mut zeroed_caps = 0usize;
    for profile in &profile_table {
        let id = field(profile, "limit_profile_id");
        let restorable: Vec<(&str, bool)> = columns
            .iter()
            .copied()
            .filter_map(|column| {
                let value = limit_value(profile, column);
                let default_value = limit_value(default, column);
                if value > default_value {
                    Some((column, true))
                } else if value == 0 && default_value > 0 && id != SMALL_ARTIFACT_LIMIT_PROFILE_ID {
                    Some((column, false))
                } else {
                    None
                }
            })
            .collect();
        if restorable.is_empty() {
            continue;
        }
        let asserted: Vec<&Row> = cases
            .iter()
            .filter(|row| field(row, "limit_profile_id") == id)
            .filter(|row| unasserted(row, &withdrawn).is_none())
            .collect();
        for row in &asserted {
            read_committed(&root, row, "artifact_ref", "artifact_sha3_256");
        }
        for (column, raised) in restorable {
            let restored = with_default_cap(profile, column);
            let class = cap_refusal_class(column);
            if raised {
                let needed = asserted.iter().any(|row| {
                    field(row, "diagnostic_class") != class
                        && refused_with(row, restored) == Some(class)
                });
                assert!(needed, "{id}: no case needs {column} above its default");
                raised_caps += 1;
            } else {
                let needed = asserted.iter().any(|row| {
                    refused_with(row, profiles[id]) == Some(class)
                        && refused_with(row, restored).is_some_and(|other| other != class)
                });
                assert!(needed, "{id}: no case needs {column} at zero");
                zeroed_caps += 1;
            }
        }
    }
    assert!(raised_caps > 0, "no profile raises a cap");
    assert!(zeroed_caps > 0, "no profile lowers a cap to zero");
    println!(
        "wire corpus: {lowered_caps} cap(s) lowered above zero, each exactly what an accepted case needs; {zeroed_caps} lowered to zero, each needed by a case; {raised_caps} raised above the default, each needed by a case"
    );
}

/// Builds the extraction listing `FORMAT.md` §12.3 defines for a directory
/// root: one line per extracted object, ordered by path depth and then by path
/// bytes, holding kind, size, content digest, and the archive path.
fn extraction_listing(root: &Path) -> Vec<u8> {
    let root_name = root
        .file_name()
        .expect("extracted root has a name")
        .to_str()
        .expect("extracted root name is UTF-8")
        .to_string();
    let mut objects = Vec::new();
    collect_objects(root, &root_name, &mut objects);
    objects.sort_by(|(a, _), (b, _)| {
        a.split('/')
            .count()
            .cmp(&b.split('/').count())
            .then_with(|| a.cmp(b))
    });
    let mut out = String::new();
    for (path, content) in objects {
        match content {
            Some(bytes) => {
                out.push_str(&format!("f {} {} {path}\n", bytes.len(), sha3_hex(&bytes)))
            }
            None => out.push_str(&format!("d - - {path}\n")),
        }
    }
    out.into_bytes()
}

/// Walks an extracted tree, pairing each archive path with its file content,
/// or with `None` for a directory. Recursion is bounded by the FCA path-depth
/// cap the extraction already enforced.
fn collect_objects(path: &Path, archive_path: &str, out: &mut Vec<(String, Option<Vec<u8>>)>) {
    if path.is_dir() {
        out.push((archive_path.to_string(), None));
        let mut children: Vec<PathBuf> = fs::read_dir(path)
            .expect("read extracted directory")
            .map(|entry| entry.expect("read directory entry").path())
            .collect();
        children.sort();
        for child in children {
            let name = child
                .file_name()
                .expect("extracted entry has a name")
                .to_str()
                .expect("extracted entry name is UTF-8")
                .to_string();
            collect_objects(&child, &format!("{archive_path}/{name}"), out);
        }
    } else {
        out.push((
            archive_path.to_string(),
            Some(fs::read(path).expect("read extracted file")),
        ));
    }
}

/// Named rather than a closure: a `&|_| {}` literal infers a single concrete
/// lifetime, which does not satisfy the callback's higher-ranked bound.
fn ignore_progress(_event: &ProgressEvent) {}

/// Runs the action of a `case_type` case on `artifact` and returns the
/// byte-exact record the corpus commits for an accepted case, so acceptance
/// proves what was produced rather than only that the operation succeeded.
/// `limits` supplies the local caps to the actions that apply them;
/// `applies_no_local_cap` names the ones that apply none. Returns `None` for
/// a STREAM known-answer case: raw STREAM encryption is not public API, so
/// the library's own `replay_stream_kats` unit test replays those.
fn run_action(
    case_id: &str,
    case_type: &str,
    artifact: &Path,
    credential: &Credential,
    limits: impl Fn() -> LimitProfile,
    out: &Path,
) -> Option<Result<Vec<u8>, CryptoError>> {
    Some(match case_type {
        "fcr_decrypt" => decrypt_case(artifact, credential, limits(), out).map(|path| {
            // §12.3: the extracted root kind chooses the comparison. A file
            // root is compared as content; a directory root has no single
            // plaintext, so it is compared as an extraction listing.
            if path.is_dir() {
                extraction_listing(&path)
            } else {
                fs::read(path).expect("read decrypted output")
            }
        }),
        "public_key_decode" => PublicKey::from_key_file_with_limits(artifact, limits().key)
            .and_then(|key| key.to_x25519_bytes())
            .map(|material| material.to_vec()),
        "private_key_validate" => validate_private_key_file(artifact).map(|()| Vec::new()),
        "private_key_open" => {
            let Credential::PrivateKey { unlock, .. } = credential else {
                panic!("{case_id}: a private_key_open case needs a private-key credential")
            };
            let limits = limits();
            PrivateKey::from_key_file(artifact, Passphrase::new(unlock.clone()))
                .into_public_key_with_limits(limits.kdf, limits.key, ignore_progress)
                .and_then(|key| key.to_x25519_bytes())
                .map(|material| material.to_vec())
        }
        "stream_encrypt_kat" => return None,
        other => panic!("{case_id}: replay does not cover case type {other}"),
    })
}

/// Runs the action of case `row` on its committed artifact under `limits`,
/// with the credential the row names. The caller checks the artifact's digest
/// before the first run.
fn replay_row(
    root: &Path,
    row: &Row,
    credentials: &BTreeMap<String, Credential>,
    limits: impl Fn() -> LimitProfile,
) -> Option<Result<Vec<u8>, CryptoError>> {
    let case_id = field(row, "case_id");
    let credential = credentials
        .get(field(row, "credential_id"))
        .unwrap_or_else(|| panic!("{case_id}: credential not declared"));
    let out = tempfile::tempdir().expect("replay output dir");
    run_action(
        case_id,
        field(row, "case_type"),
        &root.join(field(row, "artifact_ref")),
        credential,
        limits,
        out.path(),
    )
}

/// Decrypts `artifact` with every local cap set from `limits`: the header
/// caps from the open onward, and the KDF, key-file, and archive caps for the
/// decryption and any private-key unlock it performs.
fn decrypt_case(
    artifact: &Path,
    credential: &Credential,
    limits: LimitProfile,
    out: &Path,
) -> Result<PathBuf, CryptoError> {
    match (
        Decryptor::open_with_limits(artifact, limits.header)?,
        credential,
    ) {
        (Decryptor::Passphrase(decryptor), Credential::Passphrase(text)) => decryptor
            .kdf_limit(limits.kdf)
            .archive_limits(limits.archive)
            .decrypt(Passphrase::new(text.clone()), out, ignore_progress)
            .map(|outcome| outcome.output_path),
        (Decryptor::PrivateKey(decryptor), Credential::PrivateKey { key_path, unlock }) => {
            decryptor
                .kdf_limit(limits.kdf)
                .key_read_limits(limits.key)
                .archive_limits(limits.archive)
                .decrypt(
                    PrivateKey::from_key_file(key_path, Passphrase::new(unlock.clone())),
                    out,
                    ignore_progress,
                )
                .map(|outcome| outcome.output_path)
        }
        (opened, _) => panic!("credential does not match the classified mode {opened:?}"),
    }
}