oci-image 0.1.1

//! One-pass layer unpack.
//!
//! The tar is read once, in order. Whiteouts (`.wh.<name>`) and opaque
//! markers (`.wh..wh..opq`) remove what LOWER layers put there; anything this
//! layer itself writes, before or after the marker, stays. That is the OCI
//! rule, and it makes the result independent of where the markers sit in the
//! tar, so no separate whiteout pass is needed.
//!
//! Directories, links and special files are created in order on the calling
//! thread. Regular files up to [`SMALL_FILE`] are read into memory and written
//! by a pool of worker threads (create, write, chown, chmod, mtime, all on the
//! open descriptor), with at most [`IN_FLIGHT`] bytes waiting. A later entry
//! that touches a path still being written waits for the pool first.

use std::collections::HashSet;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Read};
use std::os::unix::fs::{MetadataExt, PermissionsExt};
use std::path::{Component, Path, PathBuf};
use std::sync::{mpsc, Arc, Condvar, Mutex};
use std::thread;
use std::time::{Duration, SystemTime};

use crate::decode::{open_layer, Decoder};
use crate::error::Context;
use crate::{Error, Result};

/// Regular files up to this size go to the writer pool; larger ones are
/// streamed to disk on the unpacking thread.
pub const SMALL_FILE: u64 = 1 << 20;
/// The most file bytes held in memory for the writer pool.
pub const IN_FLIGHT: usize = 16 << 20;
/// Read-ahead chunk between the decoder thread and the tar reader.
const CHUNK: usize = 1 << 20;

/// How to unpack layers.
#[derive(Debug, Clone)]
pub struct UnpackOptions {
    /// Threads for file writes and parallel gzip decode; 0 means all cores
    /// this process may use. 1 unpacks on the calling thread only.
    pub threads: usize,
    /// Decode gzip layers with `lgz` on `threads` workers (false: one-core
    /// zlib-rs).
    pub parallel_gzip: bool,
    /// Restore the uid/gid of every entry (needs root or CAP_CHOWN).
    pub same_owner: bool,
    /// Apply whiteouts onto `rootfs` (flatten), or keep them in overlayfs
    /// form for a per-layer directory.
    pub layout: Layout,
}

/// What an unpack does with whiteouts.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Layout {
    /// Apply them: the layer lands on top of the layers already in `rootfs`.
    #[default]
    Flatten,
    /// Keep them for overlayfs: `.wh.<name>` becomes a 0/0 character device,
    /// `.wh..wh..opq` an `overlay.opaque` xattr on its directory. For one
    /// layer into its own empty directory.
    Overlay(OverlayXattr),
}

/// The xattr namespace overlayfs reads its markers from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OverlayXattr {
    /// `trusted.overlay.*`: a mount by real root.
    Trusted,
    /// `user.overlay.*`, mounted with `userxattr` (in a user namespace).
    User,
}

impl Default for UnpackOptions {
    fn default() -> Self {
        Self {
            threads: 0,
            parallel_gzip: true,
            same_owner: fs::metadata("/proc/self").is_ok_and(|m| m.uid() == 0),
            layout: Layout::Flatten,
        }
    }
}

impl UnpackOptions {
    /// The thread count to use.
    pub fn threads(&self) -> usize {
        if self.threads > 0 {
            self.threads
        } else {
            thread::available_parallelism().map_or(1, |n| n.get())
        }
    }
}

/// What one layer unpack did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnpackStats {
    pub decoder: Decoder,
    /// Tar entries read.
    pub entries: u64,
    /// Regular files written.
    pub files: u64,
    /// Bytes of regular file content written.
    pub bytes: u64,
    /// Whiteout and opaque markers applied.
    pub whiteouts: u64,
    /// Device nodes that could not be created (in a user namespace) and were skipped.
    pub skipped_devices: u64,
}

/// Apply the layer blob at `blob` (plain, gzip or zstd) onto `rootfs`.
pub fn apply_layer_file(blob: &Path, rootfs: &Path, opts: &UnpackOptions) -> Result<UnpackStats> {
    let (reader, decoder, _) = open_layer(blob, opts, false)?;
    let mut stats = apply_layer(reader, rootfs, opts)?;
    stats.decoder = decoder;
    Ok(stats)
}

/// [`apply_layer_file`], also returning the sha256 of the uncompressed tar
/// (the layer's diff_id). On the lgz path it is computed on one more thread
/// from the decoder's in-order output, beside the tar reader: no extra pass.
pub(crate) fn apply_layer_file_digest(
    blob: &Path,
    rootfs: &Path,
    opts: &UnpackOptions,
) -> Result<(UnpackStats, crate::Digest)> {
    let (reader, decoder, digest) = open_layer(blob, opts, true)?;
    let mut stats = apply_layer(reader, rootfs, opts)?;
    stats.decoder = decoder;
    let digest = digest
        .finish()
        .ok_or_else(|| Error::Invalid(format!("hashing layer {} failed", blob.display())))?;
    Ok((stats, digest))
}

/// Apply one uncompressed layer tar onto `rootfs`.
pub fn apply_layer<R: Read + Send>(
    tar: R,
    rootfs: &Path,
    opts: &UnpackOptions,
) -> Result<UnpackStats> {
    let threads = opts.threads();
    let root = rootfs
        .canonicalize()
        .ctx(|| format!("rootfs {}", rootfs.display()))?;
    let mut stats = UnpackStats {
        decoder: Decoder::Plain,
        entries: 0,
        files: 0,
        bytes: 0,
        whiteouts: 0,
        skipped_devices: 0,
    };
    let mut dirs: Vec<(PathBuf, u64)> = Vec::new();

    thread::scope(|s| -> Result<()> {
        let reader: Box<dyn Read> = if threads > 1 {
            Box::new(read_ahead(s, tar))
        } else {
            Box::new(tar)
        };
        let mut pool = Pool::new(s, threads.saturating_sub(1), opts.same_owner);
        let mut archive = tar::Archive::new(reader);
        archive.set_preserve_permissions(true);
        archive.set_preserve_ownerships(opts.same_owner);
        archive.set_preserve_mtime(true);
        archive.set_overwrite(true);
        archive.set_unpack_xattrs(false);

        // Paths this layer has put down (and their parents): whiteouts only
        // hide lower layers.
        let mut ours: HashSet<PathBuf> = HashSet::new();
        // Parent directories already checked to resolve inside the rootfs.
        let mut checked: HashSet<PathBuf> = HashSet::new();

        for entry in archive.entries().ctx(|| "reading layer".into())? {
            let mut entry = entry.ctx(|| "reading layer".into())?;
            stats.entries += 1;
            let raw = entry.path().ctx(|| "entry path".into())?.into_owned();
            let Some(rel) = entry_path(&raw) else {
                continue;
            };
            let name = rel.file_name().and_then(|n| n.to_str()).unwrap_or("");
            let parent = rel.parent().unwrap_or(Path::new("")).to_path_buf();

            if let Layout::Overlay(xattr) = opts.layout {
                let marker = if name == ".wh..wh..opq" {
                    Some(None)
                } else {
                    name.strip_prefix(".wh.").map(Some)
                };
                if let Some(marker) = marker {
                    stats.whiteouts += 1;
                    check_parent(&root, &parent, &mut checked)?;
                    let at = match marker {
                        None => {
                            set_opaque(&root.join(&parent), xattr)
                                .ctx(|| format!("marking {} opaque", parent.display()))?;
                            parent
                        }
                        Some(hidden) => {
                            let hidden = parent.join(hidden);
                            if pool.inflight.contains(&hidden) {
                                pool.drain()?;
                            }
                            remove_any(&root.join(&hidden));
                            whiteout(&root.join(&hidden))
                                .ctx(|| format!("whiteout {}", hidden.display()))?;
                            hidden
                        }
                    };
                    add_with_parents(&mut ours, &at);
                    continue;
                }
            }
            if name == ".wh..wh..opq" {
                stats.whiteouts += 1;
                checked.clear();
                if let Ok(children) = fs::read_dir(root.join(&parent)) {
                    for child in children.flatten() {
                        hide(&root, &parent.join(child.file_name()), &ours);
                    }
                }
                continue;
            }
            if let Some(hidden) = name.strip_prefix(".wh.") {
                stats.whiteouts += 1;
                checked.clear();
                hide(&root, &parent.join(hidden), &ours);
                continue;
            }

            let kind = entry.header().entry_type();
            let target = root.join(&rel);
            if pool.inflight.contains(&rel) {
                pool.drain()?;
            }
            if kind.is_hard_link() {
                if let Some(src) = entry
                    .link_name()
                    .ok()
                    .flatten()
                    .and_then(|l| entry_path(&l))
                {
                    if pool.inflight.contains(&src) {
                        pool.drain()?;
                    }
                }
            }
            // A file replacing a directory (or the reverse) needs the old one gone.
            if let Ok(meta) = fs::symlink_metadata(&target) {
                if !(meta.is_dir() && kind.is_dir()) {
                    if meta.is_dir() && ours.contains(&rel) {
                        pool.drain()?;
                    }
                    checked.clear();
                    remove_any(&target);
                }
            }

            if kind.is_file() || kind == tar::EntryType::Continuous {
                check_parent(&root, &parent, &mut checked)?;
                let h = entry.header();
                let meta = FileMeta {
                    mode: h.mode().unwrap_or(0o644) & 0o7777,
                    uid: h.uid().unwrap_or(0),
                    gid: h.gid().unwrap_or(0),
                    mtime: h.mtime().unwrap_or(0),
                };
                let size = entry.size();
                if pool.active() && size <= SMALL_FILE {
                    let mut data = Vec::with_capacity(size as usize);
                    entry
                        .read_to_end(&mut data)
                        .ctx(|| format!("reading {}", rel.display()))?;
                    pool.submit(Job { target, data, meta })?;
                    pool.inflight.insert(rel.clone());
                } else {
                    write_file(&target, &mut entry, &meta, opts.same_owner)
                        .ctx(|| format!("writing {}", rel.display()))?;
                }
                stats.files += 1;
                stats.bytes += size;
            } else {
                if kind.is_dir() {
                    check_parent(&root, &parent, &mut checked)?;
                }
                if let Err(e) = entry.unpack_in(&root) {
                    if kind.is_character_special() || kind.is_block_special() {
                        // Device nodes cannot be created in a user namespace; a
                        // container gets its own /dev anyway.
                        stats.skipped_devices += 1;
                        continue;
                    }
                    if e.to_string().contains("outside of") {
                        return Err(Error::Escape(rel));
                    }
                    return Err(e).ctx(|| format!("unpacking {}", rel.display()));
                }
                if kind.is_dir() {
                    dirs.push((target, entry.header().mtime().unwrap_or(0)));
                }
            }
            add_with_parents(&mut ours, &rel);
        }
        pool.finish()?;
        // Read to the end so the gzip/zstd trailer is checked.
        let mut rest = archive.into_inner();
        io::copy(&mut rest, &mut io::sink()).ctx(|| "reading the layer to its end".into())?;
        Ok(())
    })?;

    // Directory mtimes last, deepest first: writing into a directory moves it.
    for (dir, mtime) in dirs.iter().rev() {
        if let Ok(f) = File::open(dir) {
            let _ = f.set_modified(SystemTime::UNIX_EPOCH + Duration::from_secs(*mtime));
        }
    }
    Ok(stats)
}

/// One step of a layer, as [`changes`] reports it.
pub enum Change<'a, 'b, R: Read> {
    /// `.wh..wh..opq` in this directory: lower layers' contents are hidden.
    Opaque(PathBuf),
    /// `.wh.<name>`: this path from lower layers is hidden.
    Whiteout(PathBuf),
    /// Any other entry, with its clean relative path.
    Entry(PathBuf, &'a mut tar::Entry<'b, R>),
}

/// Walk an uncompressed layer tar in order, classifying whiteouts, for
/// readers that keep their own view of the filesystem (scanners, catalogers).
/// Entries whose path would escape the root are skipped.
pub fn changes<R: Read>(
    tar: R,
    mut f: impl FnMut(Change<'_, '_, R>) -> io::Result<()>,
) -> Result<()> {
    let mut archive = tar::Archive::new(tar);
    for entry in archive.entries().ctx(|| "reading layer".into())? {
        let mut entry = entry.ctx(|| "reading layer".into())?;
        let raw = entry.path().ctx(|| "entry path".into())?.into_owned();
        let Some(rel) = entry_path(&raw) else {
            continue;
        };
        let name = rel.file_name().and_then(|n| n.to_str()).unwrap_or("");
        let parent = rel.parent().unwrap_or(Path::new("")).to_path_buf();
        let change = if name == ".wh..wh..opq" {
            Change::Opaque(parent)
        } else if let Some(hidden) = name.strip_prefix(".wh.") {
            Change::Whiteout(parent.join(hidden))
        } else {
            Change::Entry(rel, &mut entry)
        };
        f(change).ctx(|| "layer visitor".into())?;
    }
    Ok(())
}

/// Clean relative path of a tar entry, or `None` if it would escape (or is
/// the root itself).
pub fn entry_path(path: &Path) -> Option<PathBuf> {
    let mut out = PathBuf::new();
    for c in path.components() {
        match c {
            Component::Normal(p) => out.push(p),
            Component::CurDir | Component::RootDir => {}
            Component::ParentDir | Component::Prefix(_) => return None,
        }
    }
    (!out.as_os_str().is_empty()).then_some(out)
}

fn add_with_parents(set: &mut HashSet<PathBuf>, rel: &Path) {
    let mut p = Some(rel);
    while let Some(q) = p.filter(|q| !q.as_os_str().is_empty()) {
        if !set.insert(q.to_path_buf()) {
            break;
        }
        p = q.parent();
    }
}

/// Hide `rel` from lower layers: remove it, unless this layer wrote it; in a
/// directory this layer wrote into, hide what is inside instead.
fn hide(root: &Path, rel: &Path, ours: &HashSet<PathBuf>) {
    let full = root.join(rel);
    if !ours.contains(rel) {
        remove_any(&full);
        return;
    }
    if fs::symlink_metadata(&full).is_ok_and(|m| m.is_dir()) {
        if let Ok(children) = fs::read_dir(&full) {
            for c in children.flatten() {
                hide(root, &rel.join(c.file_name()), ours);
            }
        }
    }
}

fn cpath(p: &Path) -> io::Result<std::ffi::CString> {
    std::ffi::CString::new(p.as_os_str().as_encoded_bytes())
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))
}

/// An overlayfs whiteout: a character device 0/0.
fn whiteout(path: &Path) -> io::Result<()> {
    let c = cpath(path)?;
    if unsafe { libc::mknod(c.as_ptr(), libc::S_IFCHR, 0) } != 0 {
        return Err(io::Error::last_os_error());
    }
    Ok(())
}

/// Mark a directory opaque for overlayfs.
fn set_opaque(dir: &Path, xattr: OverlayXattr) -> io::Result<()> {
    let c = cpath(dir)?;
    let name: &[u8] = match xattr {
        OverlayXattr::Trusted => b"trusted.overlay.opaque\0",
        OverlayXattr::User => b"user.overlay.opaque\0",
    };
    let r =
        unsafe { libc::lsetxattr(c.as_ptr(), name.as_ptr().cast(), b"y".as_ptr().cast(), 1, 0) };
    if r != 0 {
        return Err(io::Error::last_os_error());
    }
    Ok(())
}

fn remove_any(path: &Path) {
    match fs::symlink_metadata(path) {
        Ok(m) if m.is_dir() => {
            let _ = fs::remove_dir_all(path);
        }
        Ok(_) => {
            let _ = fs::remove_file(path);
        }
        Err(_) => {}
    }
}

/// Make sure `parent` (relative) exists and resolves inside `root`.
fn check_parent(root: &Path, parent: &Path, checked: &mut HashSet<PathBuf>) -> Result<()> {
    if parent.as_os_str().is_empty() || checked.contains(parent) {
        return Ok(());
    }
    let full = root.join(parent);
    let mut existing = full.as_path();
    while fs::symlink_metadata(existing).is_err() {
        existing = existing.parent().unwrap_or(root);
    }
    let inside = |p: &Path| p.canonicalize().is_ok_and(|c| c.starts_with(root));
    if !inside(existing) {
        return Err(Error::Escape(parent.to_path_buf()));
    }
    if existing != full {
        fs::create_dir_all(&full).ctx(|| format!("creating {}", parent.display()))?;
        if !inside(&full) {
            return Err(Error::Escape(parent.to_path_buf()));
        }
    }
    checked.insert(parent.to_path_buf());
    Ok(())
}

struct FileMeta {
    mode: u32,
    uid: u64,
    gid: u64,
    mtime: u64,
}

/// Create `target` (never through a symlink), fill it, then set owner, mode
/// and mtime on the open descriptor.
fn write_file(
    target: &Path,
    data: &mut dyn Read,
    meta: &FileMeta,
    same_owner: bool,
) -> io::Result<()> {
    let open = || OpenOptions::new().write(true).create_new(true).open(target);
    let mut f = match open() {
        Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
            fs::remove_file(target)?;
            open()?
        }
        r => r?,
    };
    io::copy(data, &mut f)?;
    if same_owner {
        std::os::unix::fs::fchown(&f, Some(meta.uid as u32), Some(meta.gid as u32))?;
    }
    f.set_permissions(fs::Permissions::from_mode(meta.mode))?;
    f.set_modified(SystemTime::UNIX_EPOCH + Duration::from_secs(meta.mtime))?;
    Ok(())
}

struct Job {
    target: PathBuf,
    data: Vec<u8>,
    meta: FileMeta,
}

#[derive(Default)]
struct State {
    pending: usize,
    bytes: usize,
    error: Option<Error>,
}

#[derive(Default)]
struct Shared {
    state: Mutex<State>,
    cv: Condvar,
}

/// File writer threads. `inflight` holds the relative paths submitted since
/// the last drain.
struct Pool {
    tx: Option<mpsc::Sender<Job>>,
    shared: Arc<Shared>,
    inflight: HashSet<PathBuf>,
}

impl Pool {
    fn new<'s>(s: &'s thread::Scope<'s, '_>, workers: usize, same_owner: bool) -> Self {
        let shared = Arc::new(Shared::default());
        if workers == 0 {
            return Self {
                tx: None,
                shared,
                inflight: HashSet::new(),
            };
        }
        let (tx, rx) = mpsc::channel::<Job>();
        let rx = Arc::new(Mutex::new(rx));
        for _ in 0..workers {
            let (rx, shared) = (rx.clone(), shared.clone());
            s.spawn(move || loop {
                let job = rx.lock().unwrap().recv();
                let Ok(job) = job else { return };
                let len = job.data.len();
                let r = write_file(&job.target, &mut &job.data[..], &job.meta, same_owner)
                    .ctx(|| format!("writing {}", job.target.display()));
                let mut st = shared.state.lock().unwrap();
                st.pending -= 1;
                st.bytes -= len;
                if let Err(e) = r {
                    st.error.get_or_insert(e);
                }
                shared.cv.notify_all();
            });
        }
        Self {
            tx: Some(tx),
            shared,
            inflight: HashSet::new(),
        }
    }

    fn active(&self) -> bool {
        self.tx.is_some()
    }

    fn submit(&mut self, job: Job) -> Result<()> {
        let len = job.data.len();
        let mut st = self.shared.state.lock().unwrap();
        while st.pending > 0 && st.bytes + len > IN_FLIGHT && st.error.is_none() {
            st = self.shared.cv.wait(st).unwrap();
        }
        if let Some(e) = st.error.take() {
            return Err(e);
        }
        st.pending += 1;
        st.bytes += len;
        drop(st);
        self.tx
            .as_ref()
            .unwrap()
            .send(job)
            .expect("writer threads live");
        Ok(())
    }

    /// Wait for every submitted write.
    fn drain(&mut self) -> Result<()> {
        let mut st = self.shared.state.lock().unwrap();
        while st.pending > 0 {
            st = self.shared.cv.wait(st).unwrap();
        }
        self.inflight.clear();
        st.error.take().map_or(Ok(()), Err)
    }

    fn finish(mut self) -> Result<()> {
        let r = self.drain();
        self.tx = None;
        r
    }
}

/// Decode on its own thread, a few chunks ahead of the tar reader.
fn read_ahead<'s, R: Read + Send + 's>(
    s: &'s thread::Scope<'s, '_>,
    mut inner: R,
) -> ChannelReader {
    let (tx, rx) = mpsc::sync_channel::<io::Result<Vec<u8>>>(4);
    s.spawn(move || loop {
        let mut buf = vec![0u8; CHUNK];
        let mut n = 0;
        while n < CHUNK {
            match inner.read(&mut buf[n..]) {
                Ok(0) => break,
                Ok(k) => n += k,
                Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
                Err(e) => {
                    let _ = tx.send(Err(e));
                    return;
                }
            }
        }
        buf.truncate(n);
        let done = n < CHUNK;
        if (n > 0 && tx.send(Ok(buf)).is_err()) || done {
            return;
        }
    });
    ChannelReader {
        rx,
        buf: Vec::new(),
        pos: 0,
    }
}

struct ChannelReader {
    rx: mpsc::Receiver<io::Result<Vec<u8>>>,
    buf: Vec<u8>,
    pos: usize,
}

impl Read for ChannelReader {
    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
        while self.pos == self.buf.len() {
            match self.rx.recv() {
                Ok(chunk) => {
                    self.buf = chunk?;
                    self.pos = 0;
                }
                Err(_) => return Ok(0),
            }
        }
        let n = out.len().min(self.buf.len() - self.pos);
        out[..n].copy_from_slice(&self.buf[self.pos..self.pos + n]);
        self.pos += n;
        Ok(n)
    }
}