znippy-common 0.10.1

//! The tree's **shape**: the directories and the unix modes that the chunk index
//! structurally cannot hold.
//!
//! ## The hole this closes
//!
//! `index::base_index_fields()` is a table of *chunks*, and every chunk belongs
//! to a regular file. Two facts of a real directory tree therefore had no row
//! shape at all and were dropped on the way in:
//!
//! * an **empty directory** owns no file, so it produced no chunk, so it
//!   produced no row. Extract only ever ran `create_dir_all` on a *file's*
//!   parent ([`crate::decompress`]), which recreates exactly the directories
//!   that happen to contain something. MEASURED before this module, on a
//!   synthetic `talosctl image cache-create` tree: `blob/`, `manifests/…` came
//!   back, `empty-dir/` did not.
//! * a **unix mode**. Nothing on the write path ever called `metadata().mode()`
//!   and nothing on the read path ever called `set_permissions`, so an extracted
//!   file got `0666 & ~umask`. MEASURED on the same tree: a `0755` script came
//!   back `0664`.
//!
//! Neither failure reported anything. `verify` was green, every file that *did*
//! come back was byte-exact, and the tree was still wrong — which is the worst
//! shape a fidelity bug can take.
//!
//! ## The representation
//!
//! One reserved section, [`ZNIPPY_FSMETA_MODULE`], holding an Arrow IPC stream of
//! `(relative_path: Utf8, kind: UInt8, mode: UInt32)`. Reserved means
//! `is_reserved_module` already makes every data reader skip it, so the section
//! is additive in both directions: an archive written before it existed reads
//! unchanged, and one written with it opens in an older znippy — which simply
//! does not learn the empty dirs and the modes. No [`ZNIPPY_FORMAT_VERSION`]
//! bump, no new column for the iceberg/append/gunnar writers to supply. The
//! alternatives were both worse and are recorded on [`ZNIPPY_FSMETA_MODULE`].
//!
//! Rows are only what the chunk index does not already carry plus the one thing
//! it carries without the mode: **every** directory (so an empty one survives)
//! and **every** file whose mode is not the plain default. Skipping the default
//! is what keeps the section proportional to the interesting part of a tree —
//! on a registry mirror, thousands of `0644` blobs contribute nothing and the
//! section is the directory list.
//!
//! ## The restore order is load-bearing
//!
//! [`restore_fsmeta`] applies in three phases, and each ordering is forced:
//!
//! 1. **directories first, before extraction** — an empty directory has no file
//!    to trigger `create_dir_all`, so nothing else will ever create it.
//! 2. **file modes after extraction** — a `0444` file must be written before it
//!    is made read-only, or the write fails.
//! 3. **directory modes last, deepest first** — a `0555` directory cannot have
//!    children created inside it, so its own mode is applied only once
//!    everything below it exists.
//!
//! Getting phase 3's order wrong is not a cosmetic bug: it turns a restore of a
//! tree containing one read-only directory into a hard failure.

use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;

use anyhow::{Result, anyhow};
use arrow::array::{StringArray, StringBuilder, UInt8Array, UInt8Builder, UInt32Array, UInt32Builder};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;

use crate::index::{
    FORMAT_VERSION_KEY, ZNIPPY_FORMAT_VERSION, ZNIPPY_FSMETA_MODULE, check_format_version,
    read_reserved_section_bytes,
};

/// What a row describes. Encoded as a `UInt8` rather than a string so the column
/// is fixed-width; the variants are open-ended on purpose — an unknown kind read
/// by an older/newer reader is skipped rather than being an error, which is what
/// keeps the section additive.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum FsKind {
    /// A directory. Written for every directory in the tree, empty or not.
    Dir = 0,
    /// A regular file whose mode is NOT the tree's [`FileDefaultMode`](Self::FileDefaultMode).
    /// The file's *bytes* remain the chunk index's job; this row carries only the mode.
    File = 1,
    /// The mode every file **not** carrying an explicit [`File`](Self::File) row has.
    /// `relative_path` is empty — the row is archive-scoped, not path-scoped.
    ///
    /// # Why this row exists instead of a per-file row
    ///
    /// The first cut compared each file's mode against a hardcoded `0o644` and
    /// wrote a row only on a difference. That is wrong, and it was caught by its
    /// own test: `0o644` is `0o666 & ~umask` for `umask 022`, and the box this
    /// was measured on runs `umask 002`, so every one of a 69-file tree's plain
    /// files "deviated" and earned a row. A constant cannot stand in for a
    /// property of whichever process happens to be extracting.
    ///
    /// The alternative — a row per file, unconditionally — is exact but pays a
    /// second copy of every path in the archive (the lookup sub-index already
    /// holds them all), so the section would grow with the payload. On a registry
    /// mirror of ~50 000 blobs whose paths average 80 bytes that is ~4 MB to say
    /// one thing 50 000 times.
    ///
    /// One archive-scoped row saying "the rest are 0o644" is both: exact,
    /// independent of anyone's umask, and O(distinct modes) rather than O(files).
    /// A registry mirror seals two file rows total — this one, and the one script.
    FileDefaultMode = 2,
}

impl FsKind {
    fn from_u8(v: u8) -> Option<Self> {
        match v {
            0 => Some(Self::Dir),
            1 => Some(Self::File),
            2 => Some(Self::FileDefaultMode),
            _ => None,
        }
    }
}

/// One tree-shape fact: this path is a directory / has this mode.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FsEntry {
    /// Relative to the archive root, derived exactly as the chunk index's
    /// `relative_path` is (`strip_prefix(root)`), so the two name the same tree.
    pub relative_path: String,
    pub kind: FsKind,
    /// Unix permission bits (`mode & 0o7777`). `0` means "not recorded" — what a
    /// non-unix writer produces — and is skipped on restore rather than applied.
    pub mode: u32,
}

/// Arrow schema of the [`ZNIPPY_FSMETA_MODULE`] section. Stamped with
/// [`FORMAT_VERSION_KEY`] like every other Arrow section so
/// [`check_format_version`] can refuse a section from a newer znippy instead of
/// mis-reading its columns.
pub fn fsmeta_schema() -> Arc<Schema> {
    let fields = vec![
        Field::new("relative_path", DataType::Utf8, false),
        Field::new("kind", DataType::UInt8, false),
        Field::new("mode", DataType::UInt32, false),
    ];
    let mut md = HashMap::new();
    md.insert(FORMAT_VERSION_KEY.to_string(), ZNIPPY_FORMAT_VERSION.to_string());
    Arc::new(Schema::new_with_metadata(fields, md))
}

/// Serialize the shape rows into the one `RecordBatch` the section holds.
///
/// Sorted by path so the section is deterministic — the same tree walked on two
/// machines (readdir order is not stable) seals the same bytes, which is the
/// same property Law 3 gives the checksums.
pub fn build_fsmeta_batch(entries: &[FsEntry]) -> Result<RecordBatch> {
    let mut order: Vec<&FsEntry> = entries.iter().collect();
    order.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));

    let n = order.len();
    // PERF (Law 2): all three builders sized once from the row count; the string
    // builder gets a byte estimate too, so a deep mirror tree's long paths do not
    // walk the Vec through a dozen reallocations.
    let mut path_b = StringBuilder::with_capacity(n, n * 48);
    let mut kind_b = UInt8Builder::with_capacity(n);
    let mut mode_b = UInt32Builder::with_capacity(n);
    for e in order {
        path_b.append_value(&e.relative_path);
        kind_b.append_value(e.kind as u8);
        mode_b.append_value(e.mode);
    }
    RecordBatch::try_new(
        fsmeta_schema(),
        vec![Arc::new(path_b.finish()), Arc::new(kind_b.finish()), Arc::new(mode_b.finish())],
    )
    .map_err(|e| anyhow!("fsmeta batch: {e}"))
}

/// Decode the section's Arrow IPC bytes back into rows.
fn decode_fsmeta_section(bytes: &[u8]) -> Result<Vec<FsEntry>> {
    use arrow::ipc::reader::StreamReader;

    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)
        .map_err(|e| anyhow!("fsmeta section: not a readable Arrow stream: {e}"))?;
    check_format_version(reader.schema().metadata())?;

    let mut out = Vec::new();
    for batch in reader {
        let batch = batch.map_err(|e| anyhow!("fsmeta section read: {e}"))?;
        let paths = batch
            .column_by_name("relative_path")
            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
            .ok_or_else(|| anyhow!("fsmeta section: missing/!Utf8 relative_path"))?;
        let kinds = batch
            .column_by_name("kind")
            .and_then(|c| c.as_any().downcast_ref::<UInt8Array>())
            .ok_or_else(|| anyhow!("fsmeta section: missing/!UInt8 kind"))?;
        let modes = batch
            .column_by_name("mode")
            .and_then(|c| c.as_any().downcast_ref::<UInt32Array>())
            .ok_or_else(|| anyhow!("fsmeta section: missing/!UInt32 mode"))?;
        out.reserve(batch.num_rows());
        for i in 0..batch.num_rows() {
            // An unrecognised kind is SKIPPED, not an error: that is the whole
            // additivity contract. A future znippy recording symlinks here must
            // not make its archives unreadable by this one.
            let Some(kind) = FsKind::from_u8(kinds.value(i)) else { continue };
            out.push(FsEntry {
                relative_path: paths.value(i).to_string(),
                kind,
                mode: modes.value(i),
            });
        }
    }
    Ok(out)
}

/// Read an archive's tree shape. `Ok(None)` — not an empty vec — when the archive
/// carries no such section, so an archive sealed before this module existed says
/// "I do not know" rather than "the tree had no directories".
pub fn read_fsmeta(archive: &Path) -> Result<Option<Vec<FsEntry>>> {
    match read_reserved_section_bytes(archive, ZNIPPY_FSMETA_MODULE)? {
        None => Ok(None),
        Some(bytes) => Ok(Some(decode_fsmeta_section(&bytes)?)),
    }
}

/// What [`restore_fsmeta`] actually did — reported rather than assumed, so a
/// caller can tell "no section" from "a section that restored nothing".
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct FsRestoreReport {
    pub dirs_created: u64,
    pub modes_applied: u64,
}

/// Phase 1: create every recorded directory, **before** extraction.
///
/// This is the phase that makes an empty directory survive at all — nothing else
/// on the read path will ever create one. Runs before the chunk loop so a
/// directory is in place ahead of the files that land in it, which also spares
/// the extract loop a `create_dir_all` per file for paths already covered here.
///
/// Every path goes through `safe_output_path`, the same lexical guard the file
/// rows use: a `..` or absolute component in the section is refused, not
/// sanitized. The section is as attacker-reachable as the index is.
pub fn create_recorded_dirs(out_dir: &Path, entries: &[FsEntry]) -> Result<u64> {
    let mut created = 0u64;
    for e in entries.iter().filter(|e| e.kind == FsKind::Dir) {
        let full = crate::decompress::safe_output_path(out_dir, &e.relative_path)?;
        if !full.is_dir() {
            std::fs::create_dir_all(&full)?;
            created += 1;
        }
    }
    Ok(created)
}

/// Phases 2 and 3: apply the recorded modes, **after** extraction.
///
/// `extracted` is the set of file paths the extract actually restored — the
/// [`FsKind::FileDefaultMode`] row applies to every one of them that has no
/// explicit row of its own, which is what makes the section O(distinct modes)
/// rather than O(files).
///
/// Files first, then directories deepest-first. Both orderings are forced and
/// neither is cosmetic — see the module docs. A `0` mode is "not recorded" and is
/// left alone.
#[cfg(unix)]
pub fn apply_recorded_modes<'p, I>(out_dir: &Path, entries: &[FsEntry], extracted: I) -> Result<u64>
where
    I: IntoIterator<Item = &'p str>,
{
    use std::collections::HashSet;
    use std::os::unix::fs::PermissionsExt;

    let mut applied = 0u64;
    let set = |rel: &str, mode: u32| -> Result<bool> {
        let full = crate::decompress::safe_output_path(out_dir, rel)?;
        // A filtered extract restores a subset, and a mode row for a path that was
        // not restored is expected rather than an error. `symlink_metadata` so a
        // dangling entry is still seen as present.
        if std::fs::symlink_metadata(&full).is_err() {
            return Ok(false);
        }
        std::fs::set_permissions(&full, std::fs::Permissions::from_mode(mode))?;
        Ok(true)
    };

    // Explicit per-file rows first, and remember which paths they covered so the
    // default does not then overwrite them.
    let mut explicit: HashSet<&str> = HashSet::new();
    for e in entries.iter().filter(|e| e.kind == FsKind::File && e.mode != 0) {
        explicit.insert(e.relative_path.as_str());
        if set(&e.relative_path, e.mode)? {
            applied += 1;
        }
    }

    // Then the archive-scoped default over everything else that came out.
    if let Some(def) = entries
        .iter()
        .find(|e| e.kind == FsKind::FileDefaultMode && e.mode != 0)
        .map(|e| e.mode)
    {
        for rel in extracted {
            if explicit.contains(rel) {
                continue;
            }
            if set(rel, def)? {
                applied += 1;
            }
        }
    }

    // Deepest first: a directory whose recorded mode denies write (0o555, or the
    // 0o700-minus-w shapes a cache tree uses) must not be sealed while anything
    // below it still needs creating or chmod-ing. Depth by component count, not
    // string length — `a/bbbb` is shallower than `a/b/c` however the bytes fall.
    let mut dirs: Vec<&FsEntry> =
        entries.iter().filter(|e| e.kind == FsKind::Dir && e.mode != 0).collect();
    dirs.sort_by_key(|e| std::cmp::Reverse(e.relative_path.matches('/').count()));
    for e in dirs {
        if set(&e.relative_path, e.mode)? {
            applied += 1;
        }
    }
    Ok(applied)
}

/// Non-unix: mode bits do not translate, so nothing is applied. The directory
/// phase still runs — an empty directory is not a unix concept.
#[cfg(not(unix))]
pub fn apply_recorded_modes<'p, I>(
    _out_dir: &Path,
    _entries: &[FsEntry],
    _extracted: I,
) -> Result<u64>
where
    I: IntoIterator<Item = &'p str>,
{
    Ok(0)
}

/// A file's permission bits, or `None` off unix where they do not translate.
#[cfg(unix)]
pub fn file_mode(meta: &std::fs::Metadata) -> Option<u32> {
    use std::os::unix::fs::MetadataExt;
    Some(meta.mode() & 0o7777)
}

#[cfg(not(unix))]
pub fn file_mode(_meta: &std::fs::Metadata) -> Option<u32> {
    None
}

/// Reduce per-file modes to the rows worth sealing: one
/// [`FsKind::FileDefaultMode`] row carrying the tree's **most common** file mode,
/// plus one [`FsKind::File`] row for each file that deviates from it.
///
/// Taking the modal mode rather than a constant is what makes the elision exact:
/// whatever `umask` the tree was built under, the bulk case is described once and
/// the deviations — the executable script, the read-only blob — are named.
///
/// `modes` is consumed in walk order; ties on frequency resolve to the lowest
/// mode, so the section is deterministic for the same reason
/// [`build_fsmeta_batch`] sorts.
pub fn summarise_file_modes(modes: &[(String, u32)]) -> Vec<FsEntry> {
    if modes.is_empty() {
        return Vec::new();
    }
    let mut counts: HashMap<u32, usize> = HashMap::new();
    for (_, m) in modes {
        *counts.entry(*m).or_default() += 1;
    }
    // Most frequent wins; lowest mode breaks a tie so two machines agree.
    let default = counts
        .iter()
        .max_by_key(|(mode, n)| (**n, std::cmp::Reverse(**mode)))
        .map(|(mode, _)| *mode)
        .unwrap_or(0);

    let mut out = Vec::with_capacity(1 + modes.len() - counts[&default]);
    out.push(FsEntry {
        relative_path: String::new(),
        kind: FsKind::FileDefaultMode,
        mode: default,
    });
    for (path, mode) in modes {
        if *mode != default {
            out.push(FsEntry {
                relative_path: path.clone(),
                kind: FsKind::File,
                mode: *mode,
            });
        }
    }
    out
}

/// The mode to record for a directory. Always `Some` on unix — the row is being
/// written for the directory's *existence* regardless, so the mode rides along
/// free.
#[cfg(unix)]
pub fn dir_mode(meta: &std::fs::Metadata) -> u32 {
    use std::os::unix::fs::MetadataExt;
    meta.mode() & 0o7777
}

#[cfg(not(unix))]
pub fn dir_mode(_meta: &std::fs::Metadata) -> u32 {
    0
}

#[cfg(test)]
mod tests {
    use super::*;

    fn sample() -> Vec<FsEntry> {
        vec![
            FsEntry { relative_path: "blob".into(), kind: FsKind::Dir, mode: 0o755 },
            FsEntry { relative_path: "empty-dir".into(), kind: FsKind::Dir, mode: 0o755 },
            FsEntry { relative_path: "blob/run.sh".into(), kind: FsKind::File, mode: 0o755 },
            FsEntry {
                // The registry-mirror shapes: ':' in a host:port, '@' in a repo
                // segment, '+' in a tag, and a non-ASCII tag.
                relative_path: "manifests/registry.k8s.io:5000/x@v1/ref/v1.30+b1".into(),
                kind: FsKind::Dir,
                mode: 0o700,
            },
            FsEntry { relative_path: "manifests/ünïcödé".into(), kind: FsKind::Dir, mode: 0o755 },
        ]
    }

    /// The section is only useful if it survives the encode; every column and
    /// every awkward path shape must come back identical.
    #[test]
    fn shape_rows_round_trip_through_arrow() {
        let batch = build_fsmeta_batch(&sample()).unwrap();
        let mut bytes = Vec::new();
        {
            let mut w =
                arrow::ipc::writer::StreamWriter::try_new(&mut bytes, &fsmeta_schema()).unwrap();
            w.write(&batch).unwrap();
            w.finish().unwrap();
        }
        let back = decode_fsmeta_section(&bytes).unwrap();

        let mut want = sample();
        want.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
        assert_eq!(back, want, "every row, in the deterministic sorted order");
    }

    /// Determinism: readdir order is not stable across machines, so the section
    /// must not be either. Two shuffles of one tree seal identical bytes.
    #[test]
    fn the_section_is_independent_of_walk_order() {
        let a = build_fsmeta_batch(&sample()).unwrap();
        let mut shuffled = sample();
        shuffled.reverse();
        let b = build_fsmeta_batch(&shuffled).unwrap();
        assert_eq!(a, b, "sorted by path, so walk order cannot leak into the archive");
    }

    /// The additivity contract, driven rather than asserted in prose: a row whose
    /// `kind` this reader does not know must be skipped, never an error — or the
    /// first znippy to record symlinks here makes its archives unreadable by
    /// every znippy shipped before it.
    #[test]
    fn an_unknown_kind_is_skipped_not_an_error() {
        let schema = fsmeta_schema();
        let batch = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(StringArray::from(vec!["a", "future/thing"])),
                Arc::new(UInt8Array::from(vec![FsKind::Dir as u8, 200])),
                Arc::new(UInt32Array::from(vec![0o755u32, 0o777])),
            ],
        )
        .unwrap();
        let mut bytes = Vec::new();
        {
            let mut w = arrow::ipc::writer::StreamWriter::try_new(&mut bytes, &schema).unwrap();
            w.write(&batch).unwrap();
            w.finish().unwrap();
        }
        let back = decode_fsmeta_section(&bytes).unwrap();
        assert_eq!(back.len(), 1, "the known row survives");
        assert_eq!(back[0].relative_path, "a");
    }

    /// A path escaping the output root is REFUSED, not sanitized — the section is
    /// exactly as attacker-reachable as the chunk index, and gets the same guard.
    #[test]
    fn a_traversing_dir_row_is_refused() {
        let out = tempfile::tempdir().unwrap();
        let rows = vec![FsEntry {
            relative_path: "../escaped".into(),
            kind: FsKind::Dir,
            mode: 0o755,
        }];
        assert!(create_recorded_dirs(out.path(), &rows).is_err());
        assert!(!out.path().parent().unwrap().join("escaped").exists());
    }

    /// Phase 3's ordering, driven: a read-only directory with content under it
    /// must restore, which it only does if the child is chmod-ed before the
    /// parent is sealed.
    #[cfg(unix)]
    #[test]
    fn a_read_only_directory_does_not_block_its_children() {
        use std::os::unix::fs::PermissionsExt;

        let out = tempfile::tempdir().unwrap();
        let rows = vec![
            FsEntry { relative_path: "locked".into(), kind: FsKind::Dir, mode: 0o555 },
            FsEntry { relative_path: "locked/inner".into(), kind: FsKind::Dir, mode: 0o755 },
            FsEntry { relative_path: "locked/inner/x".into(), kind: FsKind::File, mode: 0o750 },
        ];
        create_recorded_dirs(out.path(), &rows).unwrap();
        std::fs::write(out.path().join("locked/inner/x"), b"payload").unwrap();
        apply_recorded_modes(out.path(), &rows, std::iter::empty()).unwrap();

        let m = |p: &str| {
            std::fs::metadata(out.path().join(p)).unwrap().permissions().mode() & 0o7777
        };
        assert_eq!(m("locked/inner/x"), 0o750);
        assert_eq!(m("locked/inner"), 0o755);
        assert_eq!(m("locked"), 0o555);

        // tempdir cleanup needs the write bit back.
        std::fs::set_permissions(out.path().join("locked"), std::fs::Permissions::from_mode(0o755))
            .unwrap();
    }

    /// The elision, and the bug that shaped it: the bulk mode is described ONCE
    /// and only the deviations are named. Driven with a bulk mode of `0o664` —
    /// deliberately NOT the `0o644` the first cut hardcoded — because a constant
    /// is exactly what this rule replaces.
    #[test]
    fn the_bulk_mode_is_described_once_and_only_deviations_are_named() {
        let modes: Vec<(String, u32)> = (0..40)
            .map(|i| (format!("blob/{i:064x}"), 0o664))
            .chain([("blob/run.sh".to_string(), 0o775), ("blob/ro".to_string(), 0o444)])
            .collect();

        let rows = summarise_file_modes(&modes);
        assert_eq!(rows.len(), 3, "one default + two deviations, not 42 rows: {rows:?}");
        assert_eq!(rows[0].kind, FsKind::FileDefaultMode);
        assert_eq!(rows[0].mode, 0o664, "the MODAL mode, whatever the umask was");
        assert!(rows[0].relative_path.is_empty(), "the default row is archive-scoped");
        let named: Vec<_> = rows[1..].iter().map(|r| (r.relative_path.as_str(), r.mode)).collect();
        assert_eq!(named, vec![("blob/run.sh", 0o775), ("blob/ro", 0o444)]);
    }

    /// The default row must be applied to files that have no row of their own,
    /// and must NOT overwrite one that does. Both halves are the point: without
    /// the first, eliding loses the bulk mode; without the second, eliding
    /// destroys the executable bit it was supposed to preserve.
    #[cfg(unix)]
    #[test]
    fn the_default_reaches_unnamed_files_and_never_overrides_a_named_one() {
        use std::os::unix::fs::PermissionsExt;

        let out = tempfile::tempdir().unwrap();
        for f in ["plain", "other", "run.sh"] {
            std::fs::write(out.path().join(f), b"x").unwrap();
            std::fs::set_permissions(out.path().join(f), std::fs::Permissions::from_mode(0o600))
                .unwrap();
        }
        let rows = summarise_file_modes(&[
            ("plain".into(), 0o664),
            ("other".into(), 0o664),
            ("run.sh".into(), 0o775),
        ]);
        let extracted = ["plain", "other", "run.sh"];
        let n = apply_recorded_modes(out.path(), &rows, extracted).unwrap();
        assert_eq!(n, 3, "all three files got a mode");

        let m = |p: &str| {
            std::fs::metadata(out.path().join(p)).unwrap().permissions().mode() & 0o7777
        };
        assert_eq!(m("plain"), 0o664, "the default reached it");
        assert_eq!(m("other"), 0o664);
        assert_eq!(m("run.sh"), 0o775, "the named row won over the default");
    }
}