1pub mod batch;
34pub mod local;
35mod local_fs;
36pub mod process;
37pub mod projects;
38pub mod remote;
39pub mod sftp;
40pub mod winps;
41
42use std::collections::{BTreeMap, BTreeSet};
43use std::path::Path;
44use std::sync::Arc;
45
46use serde::Serialize;
47
48use super::apply::host::ApplyHost;
49use super::layout::{RelPath, TAG_TYPES};
50use super::manifest::Manifest;
51use super::rules::GATEWAY_OWNED_FILES;
52use super::target::{NodeTarget, TargetOs};
53use super::tree::RenderedTree;
54use crate::error::CoreError;
55
56pub const CORE_COLLECTION: &str = "core";
58
59pub const READ_MAX_FILE_BYTES: u64 = 67_108_864;
61pub const READ_MAX_TOTAL_BYTES: u64 = 1_073_741_824;
63pub const READ_MAX_ENTRIES: usize = 200_000;
65
66#[derive(Debug, Clone, PartialEq, Eq)]
68pub struct FileBlob {
69 pub bytes: Vec<u8>,
71 pub sha256: String,
73}
74
75#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct Scope {
84 pub mode: String,
86 pub types: Vec<(String, String)>,
88 pub tag_providers: Vec<String>,
90}
91
92impl Scope {
93 pub fn from_manifest(manifest: &Manifest, mode: &str) -> Scope {
98 let types = manifest
99 .types
100 .iter()
101 .filter_map(|t| t.split_once('/'))
102 .filter(|(module, ty)| !(*module == "ignition" && TAG_TYPES.contains(ty)))
103 .map(|(module, ty)| (module.to_string(), ty.to_string()))
104 .collect();
105 Scope {
106 mode: mode.to_string(),
107 types,
108 tag_providers: manifest.tag_providers.clone(),
109 }
110 }
111}
112
113#[derive(Debug, Clone, Default, PartialEq, Eq)]
116pub struct ObservedFiles {
117 pub core: BTreeMap<RelPath, FileBlob>,
119 pub mode: BTreeMap<RelPath, FileBlob>,
121 pub local_units: Vec<String>,
125}
126
127#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
132pub struct DeleteSet {
133 pub files: Vec<RelPath>,
135}
136
137#[async_trait::async_trait]
139pub trait TreeReader: Send + Sync {
140 async fn list_types(&self, collection: &str) -> Result<Vec<(String, String)>, CoreError>;
143
144 async fn read(&self, scope: &Scope) -> Result<ObservedFiles, CoreError>;
147
148 async fn read_ledger(&self, gateway: &str, mode: &str) -> Result<Option<Vec<u8>>, CoreError>;
157
158 async fn read_journal(&self, gateway: &str, mode: &str) -> Result<Option<Vec<u8>>, CoreError>;
169}
170
171enum NodeAccess {
173 Local(std::path::PathBuf),
175 Ssh {
177 shell: Box<remote::RemoteShell>,
178 data_dir: String,
181 windows: bool,
183 },
184}
185
186fn node_access(
188 repo: &Path,
189 target_file: &str,
190 label: &str,
191 node: &NodeTarget,
192) -> Result<NodeAccess, CoreError> {
193 match (&node.ssh, &node.local) {
194 (Some(_), Some(_)) => Err(CoreError::deploy_tree_invalid(
195 target_file,
196 None,
197 None,
198 format!("{label}: set ssh or local, not both"),
199 )),
200 (None, Some(local)) => {
201 let configured = Path::new(&local.data_dir);
202 let data_dir = if configured.is_absolute() {
206 configured.to_path_buf()
207 } else {
208 std::path::absolute(repo.join(configured)).map_err(|e| {
209 CoreError::deploy_tree_invalid(
210 target_file,
211 None,
212 None,
213 format!(
214 "{label}.local.data_dir {:?}: cannot be made absolute ({:?})",
215 local.data_dir,
216 e.kind()
217 ),
218 )
219 })?
220 };
221 Ok(NodeAccess::Local(data_dir))
222 }
223 (Some(ssh), None) => {
224 let windows = ssh.os == TargetOs::Windows;
225 let data_dir = if windows {
226 sftp::windows_sftp_root(&ssh.data_dir)
227 } else {
228 sftp::posix_root(&ssh.data_dir)
229 }
230 .map_err(|why| {
231 CoreError::deploy_tree_invalid(
232 target_file,
233 None,
234 None,
235 format!("{label}.ssh.data_dir: {why}"),
236 )
237 })?;
238 let shell = remote::RemoteShell::from_ssh(
239 repo,
240 target_file,
241 label,
242 ssh,
243 Arc::new(process::TokioProcessRunner),
244 )?;
245 Ok(NodeAccess::Ssh {
246 shell: Box::new(shell),
247 data_dir,
248 windows,
249 })
250 }
251 (None, None) => Err(CoreError::deploy_tree_invalid(
252 target_file,
253 None,
254 None,
255 format!("{label}: set ssh or local; this verb needs host access"),
256 )),
257 }
258}
259
260fn sftp_transport(
262 shell: remote::RemoteShell,
263 data_dir: String,
264 label: &str,
265 windows: bool,
266) -> sftp::SftpTransport {
267 if windows {
268 sftp::SftpTransport::new_windows(shell, data_dir, label)
269 } else {
270 sftp::SftpTransport::new(shell, data_dir, label)
271 }
272}
273
274pub fn for_node(
281 repo: &Path,
282 target_file: &str,
283 label: &str,
284 node: &NodeTarget,
285) -> Result<Box<dyn TreeReader>, CoreError> {
286 Ok(match node_access(repo, target_file, label, node)? {
287 NodeAccess::Local(data_dir) => Box::new(local::LocalFsTransport::new(data_dir, label)),
288 NodeAccess::Ssh {
289 shell,
290 data_dir,
291 windows,
292 } => Box::new(sftp_transport(*shell, data_dir, label, windows)),
293 })
294}
295
296#[async_trait::async_trait]
341pub trait TreeMirror: Send + Sync {
342 async fn stage(&self, tree: &RenderedTree) -> Result<Staged, CoreError>;
344
345 async fn commit(
349 &self,
350 staged: &Staged,
351 delete_set: &DeleteSet,
352 ) -> Result<MirrorReport, CoreError>;
353}
354
355#[derive(Debug, Clone, PartialEq, Eq)]
357pub enum StagingMode {
358 Separate {
361 root: String,
364 },
365 InPlace,
367}
368
369#[derive(Debug)]
371pub struct Staged {
372 pub files: Vec<(RelPath, String)>,
374 pub(crate) local: tempfile::TempDir,
378 pub(crate) remote_root: Option<String>,
381 pub(crate) owner: String,
383}
384
385#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
387pub struct MirrorReport {
388 pub written: Vec<String>,
390 pub deleted: Vec<String>,
392 pub removed_dirs: Vec<String>,
394}
395
396pub fn mirror_for_node(
403 repo: &Path,
404 target_file: &str,
405 label: &str,
406 node: &NodeTarget,
407 mode: &str,
408 staging: StagingMode,
409) -> Result<Box<dyn TreeMirror>, CoreError> {
410 let invalid = |detail: String| CoreError::deploy_tree_invalid(target_file, None, None, detail);
411 Ok(match node_access(repo, target_file, label, node)? {
412 NodeAccess::Local(data_dir) => Box::new(
413 local::LocalFsTransport::new(data_dir, label)
414 .with_mirror(mode, staging)
415 .map_err(|why| invalid(format!("{label}: {why}")))?,
416 ),
417 NodeAccess::Ssh {
418 shell,
419 data_dir,
420 windows,
421 } => Box::new(
422 sftp_transport(*shell, data_dir, label, windows)
423 .with_mirror(mode, staging)
424 .map_err(|why| invalid(format!("{label}: {why}")))?,
425 ),
426 })
427}
428
429pub fn apply_host_for_node(
439 repo: &Path,
440 target_file: &str,
441 label: &str,
442 node: &NodeTarget,
443 gateway: &str,
444 mode: &str,
445) -> Result<Box<dyn ApplyHost>, CoreError> {
446 let invalid = |detail: String| CoreError::deploy_tree_invalid(target_file, None, None, detail);
447 if let Some(why) = mirror_mode_problem(mode) {
448 return Err(invalid(format!("{label}: {why}")));
449 }
450 let access = node_access(repo, target_file, label, node)?;
451 Ok(
452 match apply_host_kind(access, label, gateway, mode).map_err(invalid)? {
453 ApplyHostKind::Local(host) => Box::new(host),
454 ApplyHostKind::Sftp(host) => host,
455 },
456 )
457}
458
459enum ApplyHostKind {
461 Local(local::LocalFsTransport),
462 Sftp(Box<sftp::SftpTransport>),
463}
464
465fn apply_host_kind(
466 access: NodeAccess,
467 label: &str,
468 gateway: &str,
469 mode: &str,
470) -> Result<ApplyHostKind, String> {
471 Ok(match access {
472 NodeAccess::Local(data_dir) => ApplyHostKind::Local(
473 local::LocalFsTransport::new(data_dir, label)
474 .with_apply(gateway, mode)
475 .map_err(|why| format!("{label}: {why}"))?,
476 ),
477 NodeAccess::Ssh {
478 shell,
479 data_dir,
480 windows,
481 } => ApplyHostKind::Sftp(Box::new(
482 sftp_transport(*shell, data_dir, label, windows)
483 .with_apply(gateway, mode)
484 .map_err(|why| format!("{label}: {why}"))?,
485 )),
486 })
487}
488
489pub async fn default_staging_mode(
504 repo: &Path,
505 target_file: &str,
506 label: &str,
507 node: &NodeTarget,
508 gateway: &str,
509 mode: &str,
510 run_id: &str,
511) -> Result<StagingMode, CoreError> {
512 let invalid = |detail: String| CoreError::deploy_tree_invalid(target_file, None, None, detail);
513 for (what, value) in [("gateway", gateway), ("mode", mode)] {
514 if let Some(why) = super::component_problem(value) {
515 return Err(invalid(format!("{label}: the staging {what} {why}")));
516 }
517 }
518 let run_ok = (1..=128).contains(&run_id.len())
519 && !run_id.starts_with('.')
520 && run_id
521 .bytes()
522 .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-'));
523 if !run_ok {
524 return Err(invalid(format!(
525 "{label}: the run id must be 1 to 128 characters of [A-Za-z0-9._-] and must not start with a dot"
526 )));
527 }
528 let rel = super::ledger::staging_rel(gateway, mode, run_id);
529 Ok(match node_access(repo, target_file, label, node)? {
530 NodeAccess::Local(data_dir) => {
531 let root = data_dir.join(&rel);
532 let root = root.to_str().ok_or_else(|| {
533 invalid(format!("{label}: the data directory is not valid UTF-8"))
534 })?;
535 StagingMode::Separate {
536 root: root.to_string(),
537 }
538 }
539 NodeAccess::Ssh {
540 shell,
541 data_dir,
542 windows,
543 } => {
544 sftp_transport(*shell, data_dir, label, windows)
545 .default_staging(gateway, mode, run_id)
546 .await?
547 }
548 })
549}
550
551pub(crate) fn mirror_mode_problem(mode: &str) -> Option<String> {
554 if let Some(why) = super::component_problem(mode) {
555 return Some(format!("the mode {why}"));
556 }
557 if mode == CORE_COLLECTION || mode == "local" {
558 return Some(format!(
559 "the mode {mode:?} names a collection a mirror never writes"
560 ));
561 }
562 None
563}
564
565pub(crate) fn mirror_path_problem(path: &RelPath) -> Option<String> {
571 let parts: Vec<&str> = path.components().collect();
572 if parts.len() < MIRROR_MIN_COMPONENTS {
573 return Some(format!(
574 "{path} is not a collection path (<module>/<type>/<name>..., at least {MIRROR_MIN_COMPONENTS} components)"
575 ));
576 }
577 for part in parts {
578 if part.starts_with('.') {
579 return Some(format!("{path} has a dot-named component {part:?}"));
580 }
581 if GATEWAY_OWNED_FILES.contains(&part) {
582 return Some(format!("{path} names the gateway-owned file {part:?}"));
583 }
584 if part.chars().any(char::is_control) {
585 return Some(format!("{path} has a control character in a name"));
586 }
587 }
588 None
589}
590
591pub(crate) const MIRROR_MIN_COMPONENTS: usize = 3;
593
594pub(crate) fn validate_mirror_inputs(
599 tree: &RenderedTree,
600 delete_set: &DeleteSet,
601 staging: &StagingMode,
602 resources_root: &str,
603) -> Result<(), String> {
604 let mut total: u64 = 0;
605 for (path, bytes) in &tree.files {
606 let len = bytes.len() as u64;
607 if len > READ_MAX_FILE_BYTES {
608 return Err(format!(
609 "{path} is {len} bytes, over the {READ_MAX_FILE_BYTES} byte file cap"
610 ));
611 }
612 total = total.saturating_add(len);
613 if total > READ_MAX_TOTAL_BYTES {
614 return Err(format!(
615 "the staged tree exceeds {READ_MAX_TOTAL_BYTES} bytes in total (at {path})"
616 ));
617 }
618 }
619 let staged: Vec<&RelPath> = tree.files.keys().collect();
620 validate_mirror_paths(&staged, delete_set)?;
621 validate_staging_root(staging, resources_root)
622}
623
624pub(crate) fn validate_mirror_paths(
627 staged: &[&RelPath],
628 delete_set: &DeleteSet,
629) -> Result<(), String> {
630 for path in staged {
631 if let Some(why) = mirror_path_problem(path) {
632 return Err(format!("staged path refused: {why}"));
633 }
634 }
635 let mut deleting: BTreeSet<&str> = BTreeSet::new();
636 for path in &delete_set.files {
637 if let Some(why) = mirror_path_problem(path) {
638 return Err(format!("delete-set path refused: {why}"));
639 }
640 if !deleting.insert(path.as_str()) {
641 return Err(format!("delete-set path {path} is listed twice"));
642 }
643 }
644 let staging: BTreeSet<&str> = staged.iter().map(|p| p.as_str()).collect();
645 if staging.len() != staged.len() {
646 return Err("a path is staged twice".to_string());
647 }
648 if let Some(both) = deleting.iter().find(|p| staging.contains(**p)) {
649 return Err(format!("{both} is both staged and in the delete set"));
650 }
651 let mut dirs: BTreeSet<&str> = BTreeSet::new();
653 for path in staging.iter().chain(deleting.iter()) {
654 for (at, _) in path.match_indices('/') {
655 dirs.insert(&path[..at]);
656 }
657 }
658 if let Some(clash) = staging
659 .iter()
660 .chain(deleting.iter())
661 .find(|p| dirs.contains(**p))
662 {
663 return Err(format!(
664 "{clash} is a file and also the directory of another path"
665 ));
666 }
667 if staging.len() + deleting.len() + dirs.len() > READ_MAX_ENTRIES {
668 return Err(format!(
669 "more than {READ_MAX_ENTRIES} entries (files, deletes and directories together)"
670 ));
671 }
672 Ok(())
673}
674
675pub(crate) fn absolute_components(text: &str) -> Result<Vec<String>, String> {
678 if text.is_empty() || text.chars().any(char::is_control) {
679 return Err("must be a non-empty path without control characters".to_string());
680 }
681 let normal = text.replace('\\', "/");
682 let bytes = normal.as_bytes();
683 let absolute = normal.starts_with('/')
684 || (bytes.len() >= 3
685 && bytes[1] == b':'
686 && bytes[2] == b'/'
687 && bytes[0].is_ascii_alphabetic());
688 if !absolute {
689 return Err("must be an absolute path".to_string());
690 }
691 let mut parts = Vec::new();
692 for part in normal.split('/') {
693 match part {
694 "" => {}
695 "." | ".." => return Err("must not contain `.` or `..` components".to_string()),
696 other => parts.push(other.to_string()),
697 }
698 }
699 Ok(parts)
700}
701
702pub(crate) fn validate_staging_root(
705 staging: &StagingMode,
706 resources_root: &str,
707) -> Result<(), String> {
708 validate_staging_root_for(staging, resources_root, cfg!(windows))
709}
710
711pub(crate) fn validate_staging_root_for(
716 staging: &StagingMode,
717 resources_root: &str,
718 case_insensitive: bool,
719) -> Result<(), String> {
720 let StagingMode::Separate { root } = staging else {
721 return Ok(());
722 };
723 let staging_parts =
724 absolute_components(root).map_err(|why| format!("the staging root {root:?} {why}"))?;
725 if staging_parts.len() < 2 {
726 return Err(format!(
727 "the staging root {root:?} must be at least two directories deep"
728 ));
729 }
730 let resources_parts = absolute_components(resources_root)
731 .map_err(|why| format!("the resources root {resources_root:?} {why}"))?;
732 let same = |a: &String, b: &String| {
733 if case_insensitive {
734 a.eq_ignore_ascii_case(b)
735 } else {
736 a == b
737 }
738 };
739 let starts_with = |long: &[String], short: &[String]| {
740 long.len() >= short.len() && long.iter().zip(short).all(|(a, b)| same(a, b))
741 };
742 if starts_with(&staging_parts, &resources_parts) {
743 return Err(format!(
744 "the staging root {root:?} is inside config/resources; staging entries there would be scanned by the gateway"
745 ));
746 }
747 if starts_with(&resources_parts, &staging_parts) {
748 return Err(format!(
749 "the staging root {root:?} contains config/resources"
750 ));
751 }
752 Ok(())
753}
754
755pub(crate) fn copy_to_temp(
758 tree: &RenderedTree,
759) -> Result<(tempfile::TempDir, Vec<(RelPath, String)>), String> {
760 use std::io::Write;
761 let temp = tempfile::Builder::new()
762 .prefix("ign-stage-")
763 .tempdir()
764 .map_err(|e| format!("cannot create a staging dir ({})", e.kind()))?;
765 let mut files = Vec::with_capacity(tree.files.len());
766 for (path, bytes) in &tree.files {
767 let target = temp.path().join(path.as_str());
768 if let Some(parent) = target.parent() {
769 std::fs::create_dir_all(parent)
770 .map_err(|e| format!("cannot create the staging dir for {path} ({})", e.kind()))?;
771 }
772 let mut file = std::fs::OpenOptions::new()
773 .write(true)
774 .create_new(true)
775 .open(&target)
776 .map_err(|e| format!("cannot stage {path} ({})", e.kind()))?;
777 file.write_all(bytes)
778 .map_err(|e| format!("cannot stage {path} ({})", e.kind()))?;
779 files.push((path.clone(), super::tree::sha256_hex(bytes)));
780 }
781 Ok((temp, files))
782}
783
784pub(crate) fn read_staged_copy(
789 base: &Path,
790 rel: &RelPath,
791 expect_sha: &str,
792) -> Result<Vec<u8>, String> {
793 use super::tree::{OpenError, open_regular_file, sha256_hex};
794 use std::io::Read;
795 let path = base.join(rel.as_str());
796 let before = std::fs::symlink_metadata(&path)
797 .map_err(|e| format!("the staged copy of {rel} is gone ({})", e.kind()))?;
798 if !before.file_type().is_file() {
799 return Err(format!("the staged copy of {rel} is not a regular file"));
800 }
801 let (file, held) =
802 open_regular_file(base, Path::new(rel.as_str()), &before).map_err(|e| match e {
803 OpenError::Changed => format!("the staged copy of {rel} changed while it was read"),
804 OpenError::Io(e) => format!("cannot read the staged copy of {rel} ({})", e.kind()),
805 })?;
806 if held.len() > READ_MAX_FILE_BYTES {
807 return Err(format!(
808 "the staged copy of {rel} is over the {READ_MAX_FILE_BYTES} byte file cap"
809 ));
810 }
811 let mut bytes = Vec::with_capacity(held.len() as usize);
812 file.take(READ_MAX_FILE_BYTES + 1)
813 .read_to_end(&mut bytes)
814 .map_err(|e| format!("cannot read the staged copy of {rel} ({})", e.kind()))?;
815 if bytes.len() as u64 > READ_MAX_FILE_BYTES || sha256_hex(&bytes) != expect_sha {
816 return Err(format!(
817 "the staged copy of {rel} changed after it was staged"
818 ));
819 }
820 Ok(bytes)
821}
822
823#[cfg(test)]
824mod tests {
825 use super::*;
826 use crate::deploy::target::{Ssh, TargetOs};
827
828 fn ssh_node(os: TargetOs, data_dir: &str) -> NodeTarget {
829 NodeTarget {
830 profile: "p".to_string(),
831 ssh: Some(Ssh {
832 host: "gw.example.test".to_string(),
833 port: None,
834 user: "svc".to_string(),
835 key_file: None,
836 key_env: None,
837 os,
838 data_dir: data_dir.to_string(),
839 known_hosts: "ssh/prod.known_hosts".to_string(),
840 connect_timeout_s: None,
841 op_timeout_s: None,
842 }),
843 local: None,
844 gateway_name: None,
845 }
846 }
847
848 fn repo() -> tempfile::TempDir {
849 let repo = tempfile::tempdir().expect("repo");
850 std::fs::create_dir_all(repo.path().join("ssh")).expect("ssh dir");
851 std::fs::write(
852 repo.path().join("ssh/prod.known_hosts"),
853 "gw.example.test ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIPinnedPinnedPinnedPinnedPinnedPinnedPinned\n",
854 )
855 .expect("known_hosts");
856 repo
857 }
858
859 #[test]
862 fn apply_host_for_node_linux_ssh() {
863 let repo = repo();
864 let linux = ssh_node(TargetOs::Linux, "/usr/local/bin/ignition/data");
865 assert!(
866 apply_host_for_node(
867 repo.path(),
868 "targets/prod.yaml",
869 "nodes.master",
870 &linux,
871 "gw",
872 "prod"
873 )
874 .is_ok(),
875 "a Linux ssh node builds an apply host"
876 );
877 let access =
878 node_access(repo.path(), "targets/prod.yaml", "nodes.master", &linux).expect("access");
879 let ApplyHostKind::Sftp(host) =
880 apply_host_kind(access, "nodes.master", "gw", "prod").expect("host")
881 else {
882 panic!("an ssh node gets the sftp host");
883 };
884 assert!(!host.is_windows_for_tests());
885 let local = NodeTarget {
886 profile: "p".to_string(),
887 ssh: None,
888 local: Some(crate::deploy::target::Local {
889 data_dir: repo.path().join("data").display().to_string(),
890 }),
891 gateway_name: None,
892 };
893 let access =
894 node_access(repo.path(), "targets/prod.yaml", "nodes.master", &local).expect("access");
895 assert!(matches!(
896 apply_host_kind(access, "nodes.master", "gw", "prod").expect("host"),
897 ApplyHostKind::Local(_)
898 ));
899 for (gateway, mode) in [
900 (".hidden", "prod"),
901 ("gw", "core"),
902 ("gw", "local"),
903 ("g/w", "prod"),
904 ] {
905 let err = apply_host_for_node(
906 repo.path(),
907 "targets/prod.yaml",
908 "nodes.master",
909 &linux,
910 gateway,
911 mode,
912 )
913 .err()
914 .unwrap_or_else(|| panic!("{gateway}/{mode} is refused"));
915 assert_eq!(err.code(), "deploy_tree_invalid", "{gateway}/{mode}");
916 }
917 }
918
919 #[test]
923 fn apply_host_for_node_windows_ssh() {
924 let repo = repo();
925 let windows = ssh_node(TargetOs::Windows, "C:\\Ignition\\data");
926 assert!(
927 apply_host_for_node(
928 repo.path(),
929 "targets/prod.yaml",
930 "nodes.master",
931 &windows,
932 "gw",
933 "prod"
934 )
935 .is_ok(),
936 "a Windows ssh node builds an apply host"
937 );
938 let access = node_access(repo.path(), "targets/prod.yaml", "nodes.master", &windows)
939 .expect("access");
940 let ApplyHostKind::Sftp(host) =
941 apply_host_kind(access, "nodes.master", "gw", "prod").expect("host")
942 else {
943 panic!("a Windows ssh node gets the sftp host");
944 };
945 assert!(host.is_windows_for_tests());
946 let err = apply_host_for_node(
948 repo.path(),
949 "targets/prod.yaml",
950 "nodes.master",
951 &windows,
952 "gw",
953 "core",
954 )
955 .err()
956 .expect("core is refused");
957 assert_eq!(err.code(), "deploy_tree_invalid");
958 }
959}