haematite_cli/install/

bundle.rs

use std::collections::BTreeSet;
use std::fs::{self, DirBuilder};
use std::os::unix::fs::{DirBuilderExt, PermissionsExt};
use std::path::{Path, PathBuf};

use super::agent::Agent;
use super::files::{Payload, bounded, entry};
#[cfg(test)]
use super::fixtures::Hooks;
use super::plan::{Plan, destination, json, text};
use super::publication::{Stage, stale_stages, sync_path};
use super::types::{Error, Manifest, Report};

/// Install the operator, configured workers and plugin, or verify an identical bundle.
///
/// # Errors
/// Refuses invalid inputs, partial or different bundles and every write or sync failure.
pub fn install(binary: &Path, config: &Path, directory: &Path) -> Result<Report, Error> {
    install_with(
        binary,
        config,
        directory,
        #[cfg(test)]
        &mut Hooks::default(),
    )
}

pub(super) fn install_with(
    binary: &Path,
    config: &Path,
    directory: &Path,
    #[cfg(test)] hooks: &mut Hooks,
) -> Result<Report, Error> {
    install_from(
        || Ok(binary),
        config,
        directory,
        None,
        #[cfg(test)]
        hooks,
    )
}

pub(super) fn install_current(
    config: &Path,
    directory: &Path,
    agent: Option<&Agent>,
    source: impl FnOnce() -> Result<PathBuf, Error>,
) -> Result<Report, Error> {
    install_from(
        source,
        config,
        directory,
        agent,
        #[cfg(test)]
        &mut Hooks::default(),
    )
}

fn install_from<B: AsRef<Path>>(
    source: impl FnOnce() -> Result<B, Error>,
    config: &Path,
    directory: &Path,
    agent: Option<&Agent>,
    #[cfg(test)] hooks: &mut Hooks,
) -> Result<Report, Error> {
    let directory = destination(directory)?;
    let stages = stale_stages(&directory)?;
    let outcome = source().and_then(|binary| {
        install_to(
            binary.as_ref(),
            config,
            &directory,
            agent,
            #[cfg(test)]
            hooks,
        )
    });
    match outcome {
        Ok(mut report) => {
            report.stale_stages = stages;
            Ok(report)
        }
        Err(error) => Err(error.with_stale_stages(stages)),
    }
}

fn install_to(
    binary: &Path,
    config: &Path,
    directory: &Path,
    agent: Option<&Agent>,
    #[cfg(test)] hooks: &mut Hooks,
) -> Result<Report, Error> {
    let mut plan = match agent {
        Some(agent) => Plan::with_agent(binary, config, directory, Some(agent))?,
        None => Plan::new(binary, config, directory)?,
    };
    match fs::symlink_metadata(directory) {
        Ok(metadata) => {
            if !metadata.is_dir() || metadata.file_type().is_symlink() {
                return Err(Error::new(
                    "bundle_destination_invalid",
                    "bundle destination is not a real directory",
                ));
            }
            verify(directory, &plan.manifest)?;
            durable(
                directory,
                &plan.manifest,
                #[cfg(test)]
                hooks,
            )?;
            return report(directory, plan.manifest.files.len(), true);
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => return Err(Error::new("bundle_destination_invalid", &error)),
    }
    let mut stage = Stage::create(
        directory,
        #[cfg(test)]
        hooks,
    )?;
    let mut completed = Vec::new();
    let result = write(
        stage.path(),
        &mut plan,
        &mut completed,
        #[cfg(test)]
        hooks,
    )
    .and_then(|()| verify(stage.path(), &plan.manifest))
    .and_then(|()| {
        stage.publish(
            directory,
            #[cfg(test)]
            hooks,
        )
    });
    if let Err(error) = result {
        let mut error = error.with_partial(&completed, stage.published);
        if let Err(cleanup) = stage.cleanup(
            #[cfg(test)]
            hooks,
        ) {
            error = error.with_cleanup(&cleanup);
        }
        return Err(error);
    }
    report(directory, plan.manifest.files.len(), false)
}

fn report(directory: &Path, files: usize, already_installed: bool) -> Result<Report, Error> {
    Ok(Report {
        directory: text(directory)?.to_owned(),
        files,
        already_installed,
        stale_stages: Vec::new(),
    })
}

fn directories(manifest: &Manifest) -> BTreeSet<String> {
    let mut directories = BTreeSet::new();
    for name in manifest.files.keys() {
        let mut path = Path::new(name).parent();
        while let Some(parent) = path {
            if let Some(name) = parent.to_str().filter(|name| !name.is_empty()) {
                directories.insert(name.to_owned());
            }
            path = parent.parent();
        }
    }
    directories
}

fn write(
    directory: &Path,
    plan: &mut Plan,
    completed: &mut Vec<String>,
    #[cfg(test)] hooks: &mut Hooks,
) -> Result<(), Error> {
    let directories = directories(&plan.manifest);
    for name in &directories {
        DirBuilder::new()
            .mode(0o700)
            .create(directory.join(name))
            .map_err(|error| Error::new("bundle_write_failed", &error))?;
    }
    for (name, payload) in &mut plan.payloads {
        let entry =
            plan.manifest.files.get(name).ok_or_else(|| {
                Error::new("bundle_plan_invalid", "payload has no manifest entry")
            })?;
        let path = directory.join(name);
        #[cfg(test)]
        hooks.payload(completed.len() + 1, &path)?;
        payload.write(
            &path,
            entry,
            #[cfg(test)]
            hooks,
        )?;
        completed.push(name.clone());
    }
    let mut manifest = Payload::Bytes(json(&plan.manifest)?);
    let manifest_entry = manifest.entry(0o600)?;
    manifest.write(
        &directory.join("bundle.json"),
        &manifest_entry,
        #[cfg(test)]
        hooks,
    )?;
    sync_directories(
        directory,
        &plan.manifest,
        #[cfg(test)]
        hooks,
    )
}

fn durable(
    directory: &Path,
    manifest: &Manifest,
    #[cfg(test)] hooks: &mut Hooks,
) -> Result<(), Error> {
    for name in manifest
        .files
        .keys()
        .map(String::as_str)
        .chain(std::iter::once("bundle.json"))
    {
        sync_path(
            &directory.join(name),
            #[cfg(test)]
            hooks,
        )?;
    }
    sync_directories(
        directory,
        manifest,
        #[cfg(test)]
        hooks,
    )?;
    sync_path(
        directory
            .parent()
            .ok_or_else(|| Error::new("bundle_path_invalid", "bundle parent missing"))?,
        #[cfg(test)]
        hooks,
    )
}

fn sync_directories(
    directory: &Path,
    manifest: &Manifest,
    #[cfg(test)] hooks: &mut Hooks,
) -> Result<(), Error> {
    for name in directories(manifest).iter().rev() {
        sync_path(
            &directory.join(name),
            #[cfg(test)]
            hooks,
        )?;
    }
    sync_path(
        directory,
        #[cfg(test)]
        hooks,
    )
}

fn verify(directory: &Path, expected: &Manifest) -> Result<(), Error> {
    let bytes = bounded(&directory.join("bundle.json"), "bundle_manifest_invalid")?;
    let actual: Manifest = serde_json::from_slice(&bytes)
        .map_err(|error| Error::new("bundle_manifest_invalid", &error))?;
    if actual != *expected {
        return Err(Error::new(
            "bundle_conflict",
            "bundle manifest differs from the requested inputs",
        ));
    }
    for (name, expected) in &actual.files {
        if entry(&directory.join(name), "bundle_payload_invalid")? != *expected {
            return Err(Error::new(
                "bundle_payload_invalid",
                &format!("payload does not match manifest: {name}"),
            ));
        }
    }
    let directories = directories(&actual);
    let mut pending = vec![String::new()];
    let mut seen = BTreeSet::new();
    while let Some(relative) = pending.pop() {
        let path = directory.join(&relative);
        let metadata = fs::symlink_metadata(&path)
            .map_err(|error| Error::new("bundle_inventory_invalid", &error))?;
        if !metadata.is_dir() || metadata.permissions().mode() & 0o7777 != 0o700 {
            return Err(Error::new(
                "bundle_inventory_invalid",
                "bundle directory type or permissions changed",
            ));
        }
        for child in
            fs::read_dir(path).map_err(|error| Error::new("bundle_inventory_invalid", &error))?
        {
            let child = child.map_err(|error| Error::new("bundle_inventory_invalid", &error))?;
            let name = child
                .file_name()
                .into_string()
                .map_err(|_| Error::new("bundle_inventory_invalid", "bundle entry is not UTF-8"))?;
            let name = if relative.is_empty() {
                name
            } else {
                format!("{relative}/{name}")
            };
            if directories.contains(&name) {
                pending.push(name);
            } else if actual.files.contains_key(&name) {
                seen.insert(name);
            } else if name == "bundle.json" {
                if entry(&child.path(), "bundle_manifest_invalid")?.mode != 0o600 {
                    return Err(Error::new(
                        "bundle_manifest_invalid",
                        "manifest permissions changed",
                    ));
                }
            } else {
                return Err(Error::new(
                    "bundle_inventory_invalid",
                    "unexpected bundle entry",
                ));
            }
        }
    }
    if seen.len() != actual.files.len() {
        return Err(Error::new(
            "bundle_inventory_invalid",
            "bundle inventory is incomplete",
        ));
    }
    Ok(())
}