ignition-core 1.6.6

//! How a verb reads a gateway node's resource collections (Phase 22).
//!
//! A transport reaches the gateway HOST's `data/config/resources/<collection>/`
//! directories. This module defines the **read half** the verbs receive:
//! [`TreeReader`]. The write half (a mirror driven by an explicit delete set,
//! [`TreeMirror`]) is a separate trait that is never reachable from a verb,
//! so "plan, pull and verify never write to the host" is a compile-time
//! fact for `actions::deploy`, not a convention.
//!
//! Two properties every reader keeps:
//!
//! - **A failed read is never an empty read.** Only an absent collection
//!   directory reads as empty. A missing or unreadable `data_dir`, a symlink,
//!   a special file, a non-UTF-8 name, a file over [`READ_MAX_FILE_BYTES`],
//!   more than [`READ_MAX_ENTRIES`] entries or more than
//!   [`READ_MAX_TOTAL_BYTES`] is [`CoreError::TransportFailed`]. A plan that
//!   mistook an unreadable tree for deletions would be worse than no plan.
//! - **Scoped.** Only the directories the manifest allowlists are opened
//!   (see [`Scope`]); nothing outside them is listed or read.
//!
//! The write half is [`TreeMirror`]; [`mirror_for_node`] is the only way to
//! obtain one, and a source-scan test (`write_half_unreachable`) pins that no
//! verb or surface names it.
//!
//! [`for_node`] chooses the reader for a target node: `local: { data_dir }`
//! selects [`local::LocalFsTransport`] (D-02); `ssh` selects
//! [`sftp::SftpTransport`] over the hardened [`remote::RemoteShell`], for a
//! Linux host with a POSIX `data_dir` and for a Windows OpenSSH host with a
//! drive-letter `data_dir` ([`sftp::windows_sftp_root`], 22-13). A Windows read
//! is cross-checked against a PowerShell enumeration ([`winps`]) before it is
//! reported complete.

pub mod batch;
pub mod local;
mod local_fs;
pub mod process;
pub mod projects;
pub mod remote;
pub mod sftp;
pub mod winps;

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::sync::Arc;

use serde::Serialize;

use super::apply::host::ApplyHost;
use super::layout::{RelPath, TAG_TYPES};
use super::manifest::Manifest;
use super::rules::GATEWAY_OWNED_FILES;
use super::target::{NodeTarget, TargetOs};
use super::tree::RenderedTree;
use crate::error::CoreError;

/// The literal name of the base resource collection on a gateway.
pub const CORE_COLLECTION: &str = "core";

/// A single file larger than this is refused (64 MiB).
pub const READ_MAX_FILE_BYTES: u64 = 67_108_864;
/// All files read in one call together may not exceed this (1 GiB).
pub const READ_MAX_TOTAL_BYTES: u64 = 1_073_741_824;
/// More directory entries than this in one read is refused.
pub const READ_MAX_ENTRIES: usize = 200_000;

/// One live file: its bytes and their lowercase hex SHA-256.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FileBlob {
    /// Raw file bytes.
    pub bytes: Vec<u8>,
    /// Lowercase hex SHA-256 of `bytes`.
    pub sha256: String,
}

/// What a read covers.
///
/// Every `types` entry is a `<module>/<type>` directory read whole. Every
/// `tag_providers` entry `P` adds the four [`TAG_TYPES`] directories under
/// `ignition`, each restricted to its `<P>/` subtree. No production scope
/// reads a tag type whole: [`Scope::from_manifest`] keeps the tag types out
/// of `types`, and pull's provider discovery lists provider directory names
/// with [`TreeReader::list_tag_providers`] instead of reading
/// `ignition/tag-provider`. An excluded tag provider
/// ([`super::exclude`]) is therefore never listed, read or downloaded.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Scope {
    /// The target's deployment mode: the name of the mode collection.
    pub mode: String,
    /// `(module, type)` directories read whole.
    pub types: Vec<(String, String)>,
    /// Tag provider directory names.
    pub tag_providers: Vec<String>,
}

impl Scope {
    /// The scope of a manifest for `mode`: the manifest's non-tag types and
    /// its tag providers minus every provider of `excluded` (the effective
    /// exclusion set, [`super::exclude::effective`]; matched by ASCII case
    /// fold). A `types` entry naming one of the four tag types is covered by
    /// `tag_providers` (restricted to those providers), so it is not read
    /// whole. Render already refuses an excluded provider that is declared;
    /// the filter here keeps it out of every read regardless.
    pub fn from_manifest(manifest: &Manifest, mode: &str, excluded: &[String]) -> Scope {
        let types = manifest
            .types
            .iter()
            .filter_map(|t| t.split_once('/'))
            .filter(|(module, ty)| !(*module == "ignition" && TAG_TYPES.contains(ty)))
            .map(|(module, ty)| (module.to_string(), ty.to_string()))
            .collect();
        Scope {
            mode: mode.to_string(),
            types,
            tag_providers: manifest
                .tag_providers
                .iter()
                .filter(|p| !super::exclude::is_excluded(p, excluded))
                .cloned()
                .collect(),
        }
    }
}

/// The live files a read found, keyed by collection-relative path
/// (`<module>/<type>/...`).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ObservedFiles {
    /// Files of the `core` collection.
    pub core: BTreeMap<RelPath, FileBlob>,
    /// Files of the mode collection (`Scope::mode`).
    pub mode: BTreeMap<RelPath, FileBlob>,
    /// Unit keys present in the `local` collection for the same scope. The
    /// gateway lets `local` override the mode collection; plan warns about
    /// them. Names only, no bytes.
    pub local_units: Vec<String>,
    /// Files of the `core` and mode collections the host holds but cannot
    /// serve: on a Windows sftp host, a file whose host path reaches
    /// `LONG_PATH_LIMIT` ([`winps::LONG_PATH_LIMIT`], 240 characters). Seen
    /// only by the PowerShell enumeration, never downloaded, and never read
    /// as absent: plan, pull, verify and apply treat the whole unit as
    /// unreadable. Byte-sorted by `(collection, path)`. Always empty on a
    /// Linux sftp host and on a local host.
    pub unreadable: Vec<Unreadable>,
}

/// One file a read could not deliver (see [`ObservedFiles::unreadable`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Unreadable {
    /// `core` or `mode`, the words of a plan entry's `live_in`.
    pub collection: String,
    /// Collection-relative path (`<module>/<type>/...`).
    pub path: RelPath,
    /// The host path length in characters (without the `\\?\` prefix).
    pub length: usize,
}

/// One entry of a [`TreeListing`]: a file or a folder, by collection-relative
/// path (`<module>/<type>/...`). Names and kinds only.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TreeEntry {
    /// Collection-relative path.
    pub path: RelPath,
    /// Whether the entry is a folder.
    pub dir: bool,
}

/// Every file and folder below one `<module>/<type>` folder of a collection
/// ([`TreeReader::list_tree`]), byte-sorted by path. The folder itself is not
/// an entry.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TreeListing {
    /// The module folder name (`ignition`).
    pub module: String,
    /// The type folder name (`tag-definition`).
    pub ty: String,
    /// The entries below `<module>/<type>/`, byte-sorted by path.
    pub entries: Vec<TreeEntry>,
}

/// The files plan proposes to delete from the mode collection: the
/// declared-names ledger's files minus the rendered files (D-05, 22-03).
/// Collection-relative paths, byte-sorted and unique. Serialized as
/// `{"files": [...]}`.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct DeleteSet {
    /// The paths, byte-sorted.
    pub files: Vec<RelPath>,
}

/// The read half of a transport.
#[async_trait::async_trait]
pub trait TreeReader: Send + Sync {
    /// The `<module>/<type>` directory pairs of `collection`, byte-sorted;
    /// empty when the collection directory is absent.
    async fn list_types(&self, collection: &str) -> Result<Vec<(String, String)>, CoreError>;

    /// The tag provider directory names under
    /// `config/resources/<collection>/ignition/tag-provider/`, byte-sorted;
    /// empty when the directory is absent. Names only: no file is downloaded
    /// and nothing below a provider directory is listed, so a provider with
    /// no file in it is named too. A symlink is refused.
    async fn list_tag_providers(&self, collection: &str) -> Result<Vec<String>, CoreError>;

    /// Every file and folder below each `<module>/<type>` folder of `dirs` in
    /// `collection`: one [`TreeListing`] per folder that exists, in `dirs`
    /// order; an absent folder is left out. Names and kinds only: no byte is
    /// downloaded. A symlink, a special file or (on Windows) a reparse point
    /// anywhere below is refused, and the entries count against
    /// [`READ_MAX_ENTRIES`]. On a Windows sftp host the listing is held to the
    /// PowerShell walk, and an entry whose host path reaches
    /// [`winps::LONG_PATH_LIMIT`] is listed, never dropped. A pair's mode tag
    /// type folders are listed whole this way, so content the plan does not
    /// read is seen (quick 261104-7cy, C3).
    async fn list_tree(
        &self,
        collection: &str,
        dirs: &[(String, String)],
    ) -> Result<Vec<TreeListing>, CoreError>;

    /// Read the `core` and mode collections within `scope`, and list the
    /// `local` units in scope. Fails closed on any error.
    async fn read(&self, scope: &Scope) -> Result<ObservedFiles, CoreError>;

    /// The declared-names ledger bytes of `gateway` and `mode` at
    /// [`ledger_location`](super::ledger::ledger_location) below the node's
    /// data directory. `None` means the ledger is absent (a first deploy). A
    /// ledger that exists but cannot be read (a symlink, a non-regular file,
    /// over [`LEDGER_MAX_BYTES`](super::ledger::LEDGER_MAX_BYTES), a read
    /// error) is [`CoreError::DeployLedgerInvalid`], never `None` (D-07). The
    /// bytes are not validated here; see
    /// [`Ledger::parse`](super::ledger::Ledger::parse).
    async fn read_ledger(&self, gateway: &str, mode: &str) -> Result<Option<Vec<u8>>, CoreError>;

    /// The apply journal bytes of `gateway` and `mode` at
    /// [`journal_location`](super::ledger::journal_location), under the same
    /// root as the ledger. `None` means there is no journal at any level of
    /// its path: no interrupted run. A journal that exists but cannot be read
    /// (a symlink, a non-regular file, over
    /// [`JOURNAL_MAX_BYTES`](super::apply::journal::JOURNAL_MAX_BYTES), a read
    /// error) is [`CoreError::DeployRunInvalid`] naming the location, never
    /// `None`: `plan`, `apply` and `rollback` must not proceed over a run
    /// they cannot see (DAPL-09 fails closed). The bytes are not validated
    /// here; see [`Journal::parse`](super::apply::journal::Journal::parse).
    async fn read_journal(&self, gateway: &str, mode: &str) -> Result<Option<Vec<u8>>, CoreError>;
}

/// How a node's data directory is reached, resolved once for both halves.
enum NodeAccess {
    /// A directory on the runner's own filesystem.
    Local(std::path::PathBuf),
    /// A host over the hardened ssh shell.
    Ssh {
        shell: Box<remote::RemoteShell>,
        /// The data directory as sftp spells it (`/opt/ignition/data`, or
        /// `/C:/Program Files/.../data` on Windows).
        data_dir: String,
        /// A Windows OpenSSH host.
        windows: bool,
    },
}

/// Resolve `node` to an access path.
fn node_access(
    repo: &Path,
    target_file: &str,
    label: &str,
    node: &NodeTarget,
) -> Result<NodeAccess, CoreError> {
    match (&node.ssh, &node.local) {
        (Some(_), Some(_)) => Err(CoreError::deploy_tree_invalid(
            target_file,
            None,
            None,
            format!("{label}: set ssh or local, not both"),
        )),
        (None, Some(local)) => {
            let configured = Path::new(&local.data_dir);
            // A relative data_dir is relative to the repo, and `--repo` is
            // `.` by default. The mirror validates its staging and resources
            // roots as absolute paths, so the join is absolutized here.
            let data_dir = if configured.is_absolute() {
                configured.to_path_buf()
            } else {
                std::path::absolute(repo.join(configured)).map_err(|e| {
                    CoreError::deploy_tree_invalid(
                        target_file,
                        None,
                        None,
                        format!(
                            "{label}.local.data_dir {:?}: cannot be made absolute ({:?})",
                            local.data_dir,
                            e.kind()
                        ),
                    )
                })?
            };
            Ok(NodeAccess::Local(data_dir))
        }
        (Some(ssh), None) => {
            let windows = ssh.os == TargetOs::Windows;
            let data_dir = if windows {
                sftp::windows_sftp_root(&ssh.data_dir)
            } else {
                sftp::posix_root(&ssh.data_dir)
            }
            .map_err(|why| {
                CoreError::deploy_tree_invalid(
                    target_file,
                    None,
                    None,
                    format!("{label}.ssh.data_dir: {why}"),
                )
            })?;
            let shell = remote::RemoteShell::from_ssh(
                repo,
                target_file,
                label,
                ssh,
                Arc::new(process::TokioProcessRunner),
            )?;
            Ok(NodeAccess::Ssh {
                shell: Box::new(shell),
                data_dir,
                windows,
            })
        }
        (None, None) => Err(CoreError::deploy_tree_invalid(
            target_file,
            None,
            None,
            format!("{label}: set ssh or local; this verb needs host access"),
        )),
    }
}

/// An sftp transport over `shell` for a Linux or a Windows host.
fn sftp_transport(
    shell: remote::RemoteShell,
    data_dir: String,
    label: &str,
    windows: bool,
) -> sftp::SftpTransport {
    if windows {
        sftp::SftpTransport::new_windows(shell, data_dir, label)
    } else {
        sftp::SftpTransport::new(shell, data_dir, label)
    }
}

/// The reader for `node`, from its target block.
///
/// `local` builds the local-filesystem transport (a relative `data_dir`
/// resolves against `repo`). A node with neither `ssh` nor `local` is
/// refused as `deploy_tree_invalid` naming `label`, because a verb that
/// reads the host needs host access (D-02).
pub fn for_node(
    repo: &Path,
    target_file: &str,
    label: &str,
    node: &NodeTarget,
) -> Result<Box<dyn TreeReader>, CoreError> {
    Ok(match node_access(repo, target_file, label, node)? {
        NodeAccess::Local(data_dir) => Box::new(local::LocalFsTransport::new(data_dir, label)),
        NodeAccess::Ssh {
            shell,
            data_dir,
            windows,
        } => Box::new(sftp_transport(*shell, data_dir, label, windows)),
    })
}

/// The write half of a transport: stage a rendered tree, then commit it with
/// an explicit delete set.
///
/// **No Phase 22 verb may import this trait.** `render`, `plan`, `pull` and
/// `verify` receive a [`TreeReader`] from [`for_node`] and cannot reach it;
/// the `write_half_unreachable` test scans the verb sources for
/// `TreeMirror`, `mirror_for_node`, `StagingMode`, `.stage(` and `.commit(`.
/// Phase 23 is the first caller: it calls [`stage`](TreeMirror::stage) before
/// the scan-lock and [`commit`](TreeMirror::commit) inside it.
///
/// What a mirror guarantees, on both transports:
///
/// - **Bounded writes.** [`commit`](TreeMirror::commit) writes exactly the
///   staged files and removes exactly the delete-set files. Nothing else in
///   the collection is touched (never `config-mode.json`, never an
///   undeclared resource, no mirror-style removal of extras). A directory
///   emptied by a delete is removed; the `<module>/<type>` directory and
///   everything above it never is.
/// - **Nothing foreign in the collection.** No temp, backup, staging or
///   dot-named entry is created below `config/resources` (the Designer
///   rejects such names). Staging happens in a private temp dir and, for
///   [`StagingMode::Separate`], in a root outside `config/resources`.
/// - **Fail closed.** Every path is validated before any I/O. The expected
///   state (delete targets are regular files, write targets are absent or
///   regular files, no component is a symlink, staged copies are intact) is
///   checked before the first write; any mismatch is `transport_failed` with
///   nothing written. Symlinks are never followed and never created.
/// - **Bounded size.** The staged tree obeys the read side's
///   [`READ_MAX_FILE_BYTES`], [`READ_MAX_TOTAL_BYTES`] and
///   [`READ_MAX_ENTRIES`].
/// - **Atomic where the platform allows.** `Separate` moves each staged file
///   into place with a rename (a replace on the same filesystem); `InPlace`
///   writes the target directly and is not atomic per file. A Linux `ssh`
///   node (23-06, D-01) does both inside ONE call of an uploaded POSIX script:
///   `Separate` moves the copies staged on the host, and `InPlace` uploads the
///   copies with the script into a throw-away directory below the account
///   home and moves them from there, so on that host `InPlace` is a `mv` per
///   file too. A commit is not atomic across files: a failure part-way leaves
///   the files already moved in place, which is why Phase 23 wraps it in a
///   snapshot.
///
/// A [`Staged`] belongs to the mirror that made it and is consumed by its
/// commit in `Separate` mode (the staged files are moved); stage again to
/// commit again.
#[async_trait::async_trait]
pub trait TreeMirror: Send + Sync {
    /// Copy `tree` into staging. Writes nothing into the collection.
    async fn stage(&self, tree: &RenderedTree) -> Result<Staged, CoreError>;

    /// Write the staged files into the mode collection and remove
    /// `delete_set`. Refuses, with nothing written, anything that is not a
    /// valid, in-bounds, expected-state operation.
    async fn commit(
        &self,
        staged: &Staged,
        delete_set: &DeleteSet,
    ) -> Result<MirrorReport, CoreError>;
}

/// Where staged files wait before they are moved into the collection.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StagingMode {
    /// Staged files are uploaded to `root`, a directory OUTSIDE
    /// `config/resources`, and moved in with a rename at commit.
    Separate {
        /// The staging root, as the transport spells paths (a local path, or
        /// an absolute POSIX path on the host).
        root: String,
    },
    /// Files are written straight into the collection at commit.
    InPlace,
}

/// A rendered tree held in staging, ready to commit.
#[derive(Debug)]
pub struct Staged {
    /// Each staged file with the lowercase hex SHA-256 of its bytes.
    pub files: Vec<(RelPath, String)>,
    /// A private temp dir holding a copy of every staged file. Commit
    /// re-reads and re-verifies it through
    /// [`open_regular_file`](super::tree::open_regular_file).
    pub(crate) local: tempfile::TempDir,
    /// The `Separate` staging root the files were copied to, `None` for
    /// `InPlace`.
    pub(crate) remote_root: Option<String>,
    /// Which mirror (node and mode) staged this.
    pub(crate) owner: String,
}

/// What a commit did, collection-relative and byte-sorted.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct MirrorReport {
    /// Files written (staged files).
    pub written: Vec<String>,
    /// Files removed (the delete set).
    pub deleted: Vec<String>,
    /// Directories removed because a delete emptied them.
    pub removed_dirs: Vec<String>,
}

/// The write half for `node`'s `mode` collection.
///
/// `local` builds the local-filesystem mirror; an `ssh` node (Linux or
/// Windows) builds the sftp mirror over the same hardened shell as
/// [`for_node`]. `mode` is the mode collection name; `core` and `local` are
/// never written.
pub fn mirror_for_node(
    repo: &Path,
    target_file: &str,
    label: &str,
    node: &NodeTarget,
    mode: &str,
    staging: StagingMode,
) -> Result<Box<dyn TreeMirror>, CoreError> {
    let invalid = |detail: String| CoreError::deploy_tree_invalid(target_file, None, None, detail);
    Ok(match node_access(repo, target_file, label, node)? {
        NodeAccess::Local(data_dir) => Box::new(
            local::LocalFsTransport::new(data_dir, label)
                .with_mirror(mode, staging)
                .map_err(|why| invalid(format!("{label}: {why}")))?,
        ),
        NodeAccess::Ssh {
            shell,
            data_dir,
            windows,
        } => Box::new(
            sftp_transport(*shell, data_dir, label, windows)
                .with_mirror(mode, staging)
                .map_err(|why| invalid(format!("{label}: {why}")))?,
        ),
    })
}

/// The apply half ([`ApplyHost`](super::apply::host::ApplyHost)) for `node`'s
/// `mode` collection, for the target `gateway`.
///
/// `local` builds the local-filesystem host. An `ssh` node builds the sftp host
/// that runs every operation as one call of the uploaded script, through a
/// fixed stub that checks the script's and the work list's SHA-256 first
/// (D-01): the POSIX `sh` script for a Linux node (23-06), the PowerShell
/// script for a Windows node (23-08, 23-20). `gateway` and `mode` must be names
/// a run may write under.
pub fn apply_host_for_node(
    repo: &Path,
    target_file: &str,
    label: &str,
    node: &NodeTarget,
    gateway: &str,
    mode: &str,
) -> Result<Box<dyn ApplyHost>, CoreError> {
    let invalid = |detail: String| CoreError::deploy_tree_invalid(target_file, None, None, detail);
    if let Some(why) = mirror_mode_problem(mode) {
        return Err(invalid(format!("{label}: {why}")));
    }
    let access = node_access(repo, target_file, label, node)?;
    Ok(
        match apply_host_kind(access, label, gateway, mode).map_err(invalid)? {
            ApplyHostKind::Local(host) => Box::new(host),
            ApplyHostKind::Sftp(host) => host,
        },
    )
}

/// The concrete host an access path resolves to, before it is boxed.
enum ApplyHostKind {
    Local(local::LocalFsTransport),
    Sftp(Box<sftp::SftpTransport>),
}

fn apply_host_kind(
    access: NodeAccess,
    label: &str,
    gateway: &str,
    mode: &str,
) -> Result<ApplyHostKind, String> {
    Ok(match access {
        NodeAccess::Local(data_dir) => ApplyHostKind::Local(
            local::LocalFsTransport::new(data_dir, label)
                .with_apply(gateway, mode)
                .map_err(|why| format!("{label}: {why}"))?,
        ),
        NodeAccess::Ssh {
            shell,
            data_dir,
            windows,
        } => ApplyHostKind::Sftp(Box::new(
            sftp_transport(*shell, data_dir, label, windows)
                .with_apply(gateway, mode)
                .map_err(|why| format!("{label}: {why}"))?,
        )),
    })
}

/// The staging mode Phase 23 uses by default (Decision B(i), D-05): a
/// [`StagingMode::Separate`] root at `ign-deploy/<gateway>/<mode>/staging/<run_id>`
/// below the ledger root of the node's transport, outside every collection:
///
/// - `ssh` nodes (Linux and Windows): below the service account's home, which
///   the server reports (`/C:/Users/svc-ignition-deploy/...` on the Windows
///   capture host). The account can write there and not under the data
///   directory (C6), and the commit is a rename from the home into the mode
///   collection on the same volume.
/// - `local` nodes: below `<data_dir>`.
///
/// `gateway` and `mode` are single path components and `run_id` is
/// `[A-Za-z0-9._-]{1,128}` not starting with a dot, or the call is refused
/// before any I/O. Only an `ssh` node touches the network (one `pwd`).
pub async fn default_staging_mode(
    repo: &Path,
    target_file: &str,
    label: &str,
    node: &NodeTarget,
    gateway: &str,
    mode: &str,
    run_id: &str,
) -> Result<StagingMode, CoreError> {
    let invalid = |detail: String| CoreError::deploy_tree_invalid(target_file, None, None, detail);
    for (what, value) in [("gateway", gateway), ("mode", mode)] {
        if let Some(why) = super::component_problem(value) {
            return Err(invalid(format!("{label}: the staging {what} {why}")));
        }
    }
    let run_ok = (1..=128).contains(&run_id.len())
        && !run_id.starts_with('.')
        && run_id
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-'));
    if !run_ok {
        return Err(invalid(format!(
            "{label}: the run id must be 1 to 128 characters of [A-Za-z0-9._-] and must not start with a dot"
        )));
    }
    let rel = super::ledger::staging_rel(gateway, mode, run_id);
    Ok(match node_access(repo, target_file, label, node)? {
        NodeAccess::Local(data_dir) => {
            let root = data_dir.join(&rel);
            let root = root.to_str().ok_or_else(|| {
                invalid(format!("{label}: the data directory is not valid UTF-8"))
            })?;
            StagingMode::Separate {
                root: root.to_string(),
            }
        }
        NodeAccess::Ssh {
            shell,
            data_dir,
            windows,
        } => {
            sftp_transport(*shell, data_dir, label, windows)
                .default_staging(gateway, mode, run_id)
                .await?
        }
    })
}

/// A mode name that may be written: a plain path component that is not one of
/// the collections a mirror must never touch.
pub(crate) fn mirror_mode_problem(mode: &str) -> Option<String> {
    if let Some(why) = super::component_problem(mode) {
        return Some(format!("the mode {why}"));
    }
    if mode == CORE_COLLECTION || mode == "local" {
        return Some(format!(
            "the mode {mode:?} names a collection a mirror never writes"
        ));
    }
    None
}

/// Why `path` may not be written or deleted by a mirror, or `None`.
///
/// A mirror path is `<module>/<type>/<name>...`: at least three components,
/// no dot-named component (the Designer rejects them), no gateway-owned file
/// name and no control character.
pub(crate) fn mirror_path_problem(path: &RelPath) -> Option<String> {
    let parts: Vec<&str> = path.components().collect();
    if parts.len() < MIRROR_MIN_COMPONENTS {
        return Some(format!(
            "{path} is not a collection path (<module>/<type>/<name>..., at least {MIRROR_MIN_COMPONENTS} components)"
        ));
    }
    for part in parts {
        if part.starts_with('.') {
            return Some(format!("{path} has a dot-named component {part:?}"));
        }
        if GATEWAY_OWNED_FILES.contains(&part) {
            return Some(format!("{path} names the gateway-owned file {part:?}"));
        }
        if part.chars().any(char::is_control) {
            return Some(format!("{path} has a control character in a name"));
        }
    }
    None
}

/// Why the apply key `key` may not be written, deleted, restored or
/// fingerprinted, or `None` (quick 261102-cef, P1).
///
/// A bare key is a mode-collection path and gets [`mirror_path_problem`]. A
/// [`CORE_KEY_PREFIX`](crate::deploy::layout::CORE_KEY_PREFIX) key names the
/// `core` collection: its collection path must be a
/// [`PAIR_UNSYNCED_TAG_TYPES`](crate::deploy::layout::PAIR_UNSYNCED_TAG_TYPES)
/// path, the only ones a pair deploy routes to `core`, and then gets the same
/// [`mirror_path_problem`]. The qualifier is split once, so `.core/.core/...`
/// is a dot-named component of the rest and refused. The legacy mirror stays
/// on [`mirror_path_problem`] alone and never meets a core key.
pub(crate) fn apply_key_problem(key: &RelPath) -> Option<String> {
    use crate::deploy::layout::{KeyCollection, is_pair_unsynced_tag_path, split_key};
    match split_key(key.as_str()) {
        (KeyCollection::Mode, _) => mirror_path_problem(key),
        (KeyCollection::Core, rest) => {
            let Ok(path) = RelPath::new(rest) else {
                return Some(format!("{key} names no core collection path"));
            };
            if !is_pair_unsynced_tag_path(rest) {
                return Some(format!(
                    "{key} names the core collection outside ignition/tag-definition and ignition/tag-type-definition"
                ));
            }
            mirror_path_problem(&path).map(|why| format!("core collection: {why}"))
        }
    }
}

/// The routed tag types the apply keys `touched` write or delete in `core`
/// (quick 261102-cef, P10), in [`PAIR_UNSYNCED_TAG_TYPES`] order, or none.
///
/// [`PAIR_UNSYNCED_TAG_TYPES`]: crate::deploy::layout::PAIR_UNSYNCED_TAG_TYPES
pub(crate) fn core_tag_types(touched: &[String]) -> Vec<&'static str> {
    use crate::deploy::layout::{KeyCollection, PAIR_UNSYNCED_TAG_TYPES, split_key};
    PAIR_UNSYNCED_TAG_TYPES
        .iter()
        .copied()
        .filter(|ty| {
            touched.iter().any(|key| match split_key(key) {
                (KeyCollection::Core, rest) => rest
                    .strip_prefix("ignition/")
                    .and_then(|r| r.split('/').next())
                    .is_some_and(|t| t == *ty),
                (KeyCollection::Mode, _) => false,
            })
        })
        .collect()
}

/// The text of a `write_host` probe problem code about `core` (quick
/// 261102-cef, P10), or `None` for any other code. Every host reports the
/// same codes: `core_missing`, `core_symlink`, `core_not_dir`,
/// `core_tags_unwritable:<tag type>` and `core_tags_unsafe:<tag type>` (a
/// `core/ignition/<tag type>` or `core/ignition` that is a symlink or not a
/// directory, quick 261103-ell, F5, or a folder below the type folder on a
/// path the run writes, quick 261104-7cy, C5); each text names the folder and
/// points at the README grant.
pub(crate) fn core_probe_problem(code: &str) -> Option<String> {
    use crate::deploy::layout::PAIR_UNSYNCED_TAG_TYPES;
    const SEE: &str = "see README Tags and redundant pairs";
    let text = match code {
        "core_missing" => format!(
            "config/resources/core does not exist; a redundant pair's tag units are written there and it is never created ({SEE})"
        ),
        "core_symlink" => format!("config/resources/core is a symlink; refused ({SEE})"),
        "core_not_dir" => format!("config/resources/core is not a directory ({SEE})"),
        _ => {
            if let Some(ty) = code
                .strip_prefix("core_tags_unsafe:")
                .filter(|ty| PAIR_UNSYNCED_TAG_TYPES.contains(ty))
            {
                format!(
                    "config/resources/core/ignition/{ty} (or the folder above it, config/resources/core/ignition, or a folder below it on a path this run writes) is a symlink or not a directory; refused ({SEE})"
                )
            } else {
                let ty = code
                    .strip_prefix("core_tags_unwritable:")
                    .filter(|ty| PAIR_UNSYNCED_TAG_TYPES.contains(ty))?;
                format!(
                    "config/resources/core/ignition/{ty} (or a folder below it on a path this run writes, or, where a folder does not exist yet, the nearest existing one) is not writable or not searchable by the deploy account, which needs write access there on both nodes ({SEE})"
                )
            }
        }
    };
    Some(text)
}

/// The shallowest path a mirror touches: `<module>/<type>/<name>`.
pub(crate) const MIRROR_MIN_COMPONENTS: usize = 3;

/// The fewest components a directory a delete emptied may have and still be
/// removed, for the apply key `path` the delete named (a commit's delete, a
/// restore's removal of a created file, a mirror's delete), counted on its
/// collection path. It is [`MIRROR_MIN_COMPONENTS`], so `<module>/<type>`
/// stays, except below the type folder of a
/// [`PAIR_UNSYNCED_TAG_TYPES`](crate::deploy::layout::PAIR_UNSYNCED_TAG_TYPES)
/// type in the mode collection, where the emptied type folder
/// `ignition/<type>` is removed too: on a redundant pair an empty one in the
/// mode collection alone stops the backup from syncing (debug
/// deploy-tag-redundancy-sync, T6), and a standalone gateway does not need
/// it. A [`CORE_KEY_PREFIX`](crate::deploy::layout::CORE_KEY_PREFIX) key keeps
/// [`MIRROR_MIN_COMPONENTS`] on its collection path, so `core/ignition/<type>`
/// is never removed: the gateway owns it (quick 261102-cef, P7). The module
/// directory is never removed. The script hosts (`commit.sh`, `commit.ps1`)
/// apply the same rule, and their reports are checked against it.
pub(crate) fn emptied_dir_floor(path: &str) -> usize {
    use crate::deploy::layout::{KeyCollection, is_pair_unsynced_tag_path, split_key};
    match split_key(path) {
        (KeyCollection::Core, _) => MIRROR_MIN_COMPONENTS,
        (KeyCollection::Mode, bare) if is_pair_unsynced_tag_path(bare) => 2,
        (KeyCollection::Mode, _) => MIRROR_MIN_COMPONENTS,
    }
}

/// Refuse a staged tree and delete set before any I/O: every path valid, no
/// path both staged and deleted, no path a file and a directory of another,
/// bytes and entries within the read side's caps, and a `Separate` staging
/// root that is not inside (or around) `resources_root`.
pub(crate) fn validate_mirror_inputs(
    tree: &RenderedTree,
    delete_set: &DeleteSet,
    staging: &StagingMode,
    resources_root: &str,
) -> Result<(), String> {
    let mut total: u64 = 0;
    for (path, bytes) in &tree.files {
        let len = bytes.len() as u64;
        if len > READ_MAX_FILE_BYTES {
            return Err(format!(
                "{path} is {len} bytes, over the {READ_MAX_FILE_BYTES} byte file cap"
            ));
        }
        total = total.saturating_add(len);
        if total > READ_MAX_TOTAL_BYTES {
            return Err(format!(
                "the staged tree exceeds {READ_MAX_TOTAL_BYTES} bytes in total (at {path})"
            ));
        }
    }
    let staged: Vec<&RelPath> = tree.files.keys().collect();
    validate_mirror_paths(&staged, delete_set)?;
    validate_staging_root(staging, resources_root)
}

/// The path half of [`validate_mirror_inputs`], also run by commit on the
/// paths of a [`Staged`].
pub(crate) fn validate_mirror_paths(
    staged: &[&RelPath],
    delete_set: &DeleteSet,
) -> Result<(), String> {
    for path in staged {
        if let Some(why) = mirror_path_problem(path) {
            return Err(format!("staged path refused: {why}"));
        }
    }
    let mut deleting: BTreeSet<&str> = BTreeSet::new();
    for path in &delete_set.files {
        if let Some(why) = mirror_path_problem(path) {
            return Err(format!("delete-set path refused: {why}"));
        }
        if !deleting.insert(path.as_str()) {
            return Err(format!("delete-set path {path} is listed twice"));
        }
    }
    let staging: BTreeSet<&str> = staged.iter().map(|p| p.as_str()).collect();
    if staging.len() != staged.len() {
        return Err("a path is staged twice".to_string());
    }
    if let Some(both) = deleting.iter().find(|p| staging.contains(**p)) {
        return Err(format!("{both} is both staged and in the delete set"));
    }
    // A path may not be a file and the directory of another path.
    let mut dirs: BTreeSet<&str> = BTreeSet::new();
    for path in staging.iter().chain(deleting.iter()) {
        for (at, _) in path.match_indices('/') {
            dirs.insert(&path[..at]);
        }
    }
    if let Some(clash) = staging
        .iter()
        .chain(deleting.iter())
        .find(|p| dirs.contains(**p))
    {
        return Err(format!(
            "{clash} is a file and also the directory of another path"
        ));
    }
    if staging.len() + deleting.len() + dirs.len() > READ_MAX_ENTRIES {
        return Err(format!(
            "more than {READ_MAX_ENTRIES} entries (files, deletes and directories together)"
        ));
    }
    Ok(())
}

/// The components of an absolute path spelled with `/` or `\`, or why it is
/// not usable as a staging root.
pub(crate) fn absolute_components(text: &str) -> Result<Vec<String>, String> {
    if text.is_empty() || text.chars().any(char::is_control) {
        return Err("must be a non-empty path without control characters".to_string());
    }
    let normal = text.replace('\\', "/");
    let bytes = normal.as_bytes();
    let absolute = normal.starts_with('/')
        || (bytes.len() >= 3
            && bytes[1] == b':'
            && bytes[2] == b'/'
            && bytes[0].is_ascii_alphabetic());
    if !absolute {
        return Err("must be an absolute path".to_string());
    }
    let mut parts = Vec::new();
    for part in normal.split('/') {
        match part {
            "" => {}
            "." | ".." => return Err("must not contain `.` or `..` components".to_string()),
            other => parts.push(other.to_string()),
        }
    }
    Ok(parts)
}

/// Refuse a `Separate` staging root that is relative, too shallow, inside
/// `resources_root` (`config/resources`) or around it. `InPlace` has no root.
pub(crate) fn validate_staging_root(
    staging: &StagingMode,
    resources_root: &str,
) -> Result<(), String> {
    validate_staging_root_for(staging, resources_root, cfg!(windows))
}

/// [`validate_staging_root`] with the comparison's case rule chosen by the
/// HOST the paths name, not by the runner: a Windows OpenSSH host is
/// case-insensitive, so `/c:/program files/.../staging` is inside
/// `/C:/Program Files/.../resources` whatever OS `ign` runs on.
pub(crate) fn validate_staging_root_for(
    staging: &StagingMode,
    resources_root: &str,
    case_insensitive: bool,
) -> Result<(), String> {
    let StagingMode::Separate { root } = staging else {
        return Ok(());
    };
    let staging_parts =
        absolute_components(root).map_err(|why| format!("the staging root {root:?} {why}"))?;
    if staging_parts.len() < 2 {
        return Err(format!(
            "the staging root {root:?} must be at least two directories deep"
        ));
    }
    let resources_parts = absolute_components(resources_root)
        .map_err(|why| format!("the resources root {resources_root:?} {why}"))?;
    let same = |a: &String, b: &String| {
        if case_insensitive {
            a.eq_ignore_ascii_case(b)
        } else {
            a == b
        }
    };
    let starts_with = |long: &[String], short: &[String]| {
        long.len() >= short.len() && long.iter().zip(short).all(|(a, b)| same(a, b))
    };
    if starts_with(&staging_parts, &resources_parts) {
        return Err(format!(
            "the staging root {root:?} is inside config/resources; staging entries there would be scanned by the gateway"
        ));
    }
    if starts_with(&resources_parts, &staging_parts) {
        return Err(format!(
            "the staging root {root:?} contains config/resources"
        ));
    }
    Ok(())
}

/// Copy every file of `tree` into a fresh private temp dir and hash it. The
/// dir is new and ours, so a plain `create_new` write is safe.
pub(crate) fn copy_to_temp(
    tree: &RenderedTree,
) -> Result<(tempfile::TempDir, Vec<(RelPath, String)>), String> {
    use std::io::Write;
    let temp = tempfile::Builder::new()
        .prefix("ign-stage-")
        .tempdir()
        .map_err(|e| format!("cannot create a staging dir ({})", e.kind()))?;
    let mut files = Vec::with_capacity(tree.files.len());
    for (path, bytes) in &tree.files {
        let target = temp.path().join(path.as_str());
        if let Some(parent) = target.parent() {
            std::fs::create_dir_all(parent)
                .map_err(|e| format!("cannot create the staging dir for {path} ({})", e.kind()))?;
        }
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&target)
            .map_err(|e| format!("cannot stage {path} ({})", e.kind()))?;
        file.write_all(bytes)
            .map_err(|e| format!("cannot stage {path} ({})", e.kind()))?;
        files.push((path.clone(), super::tree::sha256_hex(bytes)));
    }
    Ok((temp, files))
}

/// The bytes of the staged copy `rel` below `base`, read through
/// [`open_regular_file`](super::tree::open_regular_file) (descriptor-relative,
/// no symlink followed) and required to be a regular file within the file cap
/// whose SHA-256 is `expect_sha`.
pub(crate) fn read_staged_copy(
    base: &Path,
    rel: &RelPath,
    expect_sha: &str,
) -> Result<Vec<u8>, String> {
    use super::tree::{OpenError, open_regular_file, sha256_hex};
    use std::io::Read;
    let path = base.join(rel.as_str());
    let before = std::fs::symlink_metadata(&path)
        .map_err(|e| format!("the staged copy of {rel} is gone ({})", e.kind()))?;
    if !before.file_type().is_file() {
        return Err(format!("the staged copy of {rel} is not a regular file"));
    }
    let (file, held) =
        open_regular_file(base, Path::new(rel.as_str()), &before).map_err(|e| match e {
            OpenError::Changed => format!("the staged copy of {rel} changed while it was read"),
            OpenError::Io(e) => format!("cannot read the staged copy of {rel} ({})", e.kind()),
        })?;
    if held.len() > READ_MAX_FILE_BYTES {
        return Err(format!(
            "the staged copy of {rel} is over the {READ_MAX_FILE_BYTES} byte file cap"
        ));
    }
    let mut bytes = Vec::with_capacity(held.len() as usize);
    file.take(READ_MAX_FILE_BYTES + 1)
        .read_to_end(&mut bytes)
        .map_err(|e| format!("cannot read the staged copy of {rel} ({})", e.kind()))?;
    if bytes.len() as u64 > READ_MAX_FILE_BYTES || sha256_hex(&bytes) != expect_sha {
        return Err(format!(
            "the staged copy of {rel} changed after it was staged"
        ));
    }
    Ok(bytes)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::deploy::target::{Ssh, TargetOs};

    fn ssh_node(os: TargetOs, data_dir: &str) -> NodeTarget {
        NodeTarget {
            profile: "p".to_string(),
            ssh: Some(Ssh {
                host: "gw.example.test".to_string(),
                port: None,
                user: "svc".to_string(),
                key_file: None,
                key_env: None,
                os,
                data_dir: data_dir.to_string(),
                known_hosts: "ssh/prod.known_hosts".to_string(),
                connect_timeout_s: None,
                op_timeout_s: None,
            }),
            local: None,
            gateway_name: None,
        }
    }

    fn repo() -> tempfile::TempDir {
        let repo = tempfile::tempdir().expect("repo");
        std::fs::create_dir_all(repo.path().join("ssh")).expect("ssh dir");
        std::fs::write(
            repo.path().join("ssh/prod.known_hosts"),
            "gw.example.test ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIPinnedPinnedPinnedPinnedPinnedPinnedPinned\n",
        )
        .expect("known_hosts");
        repo
    }

    /// A Linux ssh node and a `local` node get their hosts; names a run may not
    /// write under are refused before any host is built.
    /// quick 261103-ell, F5: a symlinked or non-directory core tag folder has
    /// a text naming it; only the two tag types have one.
    #[test]
    fn core_probe_problem_names_an_unsafe_type_folder() {
        let text = core_probe_problem("core_tags_unsafe:tag-definition").expect("a text");
        assert!(
            text.contains("config/resources/core/ignition/tag-definition"),
            "{text}"
        );
        assert!(text.contains("symlink or not a directory"), "{text}");
        assert_eq!(core_probe_problem("core_tags_unsafe:tag-group"), None);
    }

    #[test]
    fn apply_host_for_node_linux_ssh() {
        let repo = repo();
        let linux = ssh_node(TargetOs::Linux, "/usr/local/bin/ignition/data");
        assert!(
            apply_host_for_node(
                repo.path(),
                "targets/prod.yaml",
                "nodes.master",
                &linux,
                "gw",
                "prod"
            )
            .is_ok(),
            "a Linux ssh node builds an apply host"
        );
        let access =
            node_access(repo.path(), "targets/prod.yaml", "nodes.master", &linux).expect("access");
        let ApplyHostKind::Sftp(host) =
            apply_host_kind(access, "nodes.master", "gw", "prod").expect("host")
        else {
            panic!("an ssh node gets the sftp host");
        };
        assert!(!host.is_windows_for_tests());
        let local = NodeTarget {
            profile: "p".to_string(),
            ssh: None,
            local: Some(crate::deploy::target::Local {
                data_dir: repo.path().join("data").display().to_string(),
            }),
            gateway_name: None,
        };
        let access =
            node_access(repo.path(), "targets/prod.yaml", "nodes.master", &local).expect("access");
        assert!(matches!(
            apply_host_kind(access, "nodes.master", "gw", "prod").expect("host"),
            ApplyHostKind::Local(_)
        ));
        for (gateway, mode) in [
            (".hidden", "prod"),
            ("gw", "core"),
            ("gw", "local"),
            ("g/w", "prod"),
        ] {
            let err = apply_host_for_node(
                repo.path(),
                "targets/prod.yaml",
                "nodes.master",
                &linux,
                gateway,
                mode,
            )
            .err()
            .unwrap_or_else(|| panic!("{gateway}/{mode} is refused"));
            assert_eq!(err.code(), "deploy_tree_invalid", "{gateway}/{mode}");
        }
    }

    /// 23-20: a Windows ssh node gets the sftp host too, the one that runs the
    /// PowerShell script. Until the PowerShell host was complete this arm was
    /// the 23-04 `deploy_tree_invalid` refusal.
    #[test]
    fn apply_host_for_node_windows_ssh() {
        let repo = repo();
        let windows = ssh_node(TargetOs::Windows, "C:\\Ignition\\data");
        assert!(
            apply_host_for_node(
                repo.path(),
                "targets/prod.yaml",
                "nodes.master",
                &windows,
                "gw",
                "prod"
            )
            .is_ok(),
            "a Windows ssh node builds an apply host"
        );
        let access = node_access(repo.path(), "targets/prod.yaml", "nodes.master", &windows)
            .expect("access");
        let ApplyHostKind::Sftp(host) =
            apply_host_kind(access, "nodes.master", "gw", "prod").expect("host")
        else {
            panic!("a Windows ssh node gets the sftp host");
        };
        assert!(host.is_windows_for_tests());
        // a name a run may not write under is refused for a Windows node too
        let err = apply_host_for_node(
            repo.path(),
            "targets/prod.yaml",
            "nodes.master",
            &windows,
            "gw",
            "core",
        )
        .err()
        .expect("core is refused");
        assert_eq!(err.code(), "deploy_tree_invalid");
    }

    fn key(s: &str) -> RelPath {
        RelPath::new(s).expect("a RelPath")
    }

    /// quick 261102-cef (P1): a core key is split once, admitted only below
    /// the two tag types, and its rest gets the mirror rules; a bare key is
    /// the mirror path it always was.
    #[test]
    fn core_key_apply_key_problem() {
        for ok in [
            ".core/ignition/tag-type-definition/default/T/udts.json",
            ".core/ignition/tag-definition/default/Deploy/tags.json",
            "ignition/database-translator/X/config.json",
            "ignition/tag-definition/default/Deploy/tags.json",
        ] {
            assert_eq!(apply_key_problem(&key(ok)), None, "{ok}");
        }
        for bad in [
            ".core/ignition/database-translator/X/config.json",
            ".core/ignition/tag-provider/p/config.json",
            ".core/.core/ignition/tag-definition/a/b",
            ".core/ignition/tag-definition/a/.hidden",
            ".core/ignition/tag-definition",
            ".core",
        ] {
            assert!(apply_key_problem(&key(bad)).is_some(), "{bad}");
        }
        // `.core/` alone is not even a path
        assert!(RelPath::new(".core/").is_err());
        // the downgrade guarantee: what an older build checks still refuses
        assert!(
            mirror_path_problem(&key(
                ".core/ignition/tag-definition/default/Deploy/tags.json"
            ))
            .is_some()
        );
    }

    /// quick 261102-cef (P7): a core key keeps `<module>/<type>` of `core`;
    /// a bare tag path may lose its emptied type folder (T6).
    #[test]
    fn core_key_emptied_dir_floor() {
        assert_eq!(
            emptied_dir_floor("ignition/tag-definition/default/D/tags.json"),
            2
        );
        assert_eq!(
            emptied_dir_floor(".core/ignition/tag-definition/default/D/tags.json"),
            MIRROR_MIN_COMPONENTS
        );
        assert_eq!(
            emptied_dir_floor("ignition/database-translator/X/config.json"),
            MIRROR_MIN_COMPONENTS
        );
    }
}