diff --git a/backend/parsers/windmill-parser-yaml/src/lib.rs b/backend/parsers/windmill-parser-yaml/src/lib.rs index c2d417eb6c..06adac1ee3 100644 --- a/backend/parsers/windmill-parser-yaml/src/lib.rs +++ b/backend/parsers/windmill-parser-yaml/src/lib.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use anyhow::anyhow; use serde_json::json; use windmill_parser::{Arg, MainArgSignature, ObjectProperty, Typ}; @@ -208,15 +210,52 @@ pub struct AnsibleInventory { resource_type: Option, pub pinned_resource: Option, } + +#[derive(Debug, Clone)] +pub struct GitRepo { + pub url: String, + pub commit: Option, + pub branch: Option, + pub target_path: String, +} + #[derive(Debug, Clone)] pub struct AnsibleRequirements { pub python_reqs: Vec, - pub collections: Option, + pub roles_and_collections: Option, pub file_resources: Vec, pub inventories: Vec, pub vars: Vec<(String, String)>, pub resources: Vec<(String, String)>, pub options: AnsiblePlaybookOptions, + pub vault_password: Option, + pub vault_id: Vec, + pub git_repos: Vec, + pub git_ssh_identity: Vec, +} + +impl Default for AnsibleRequirements { + fn default() -> Self { + Self { + python_reqs: vec![], + roles_and_collections: None, + file_resources: vec![], + inventories: vec![], + vars: vec![], + resources: vec![], + options: AnsiblePlaybookOptions { + verbosity: None, + forks: None, + timeout: None, + flush_cache: None, + force_handlers: None, + }, + vault_password: None, + vault_id: vec![], + git_repos: vec![], + git_ssh_identity: vec![], + } + } } fn parse_inventories(inventory_yaml: &Yaml) -> anyhow::Result> { @@ -275,22 +314,7 @@ pub fn parse_ansible_reqs( return Ok((logs, None, inner_content.to_string())); } - let opts = AnsiblePlaybookOptions { - verbosity: None, - forks: None, - timeout: None, - flush_cache: None, - force_handlers: None, - }; - let mut ret = AnsibleRequirements { - python_reqs: vec![], - collections: None, - file_resources: vec![], - inventories: vec![], - vars: vec![], - resources: vec![], - options: opts, - }; + let mut ret = AnsibleRequirements::default(); if let Yaml::Hash(doc) = &docs[0] { for (key, value) in doc { @@ -303,7 +327,7 @@ pub fn parse_ansible_reqs( let mut out_str = String::new(); let mut emitter = YamlEmitter::new(&mut out_str); emitter.dump(galaxy_requirements)?; - ret.collections = Some(out_str); + ret.roles_and_collections = Some(out_str); } if let Some(Yaml::Array(py_reqs)) = deps.get(&Yaml::String("python".to_string())) @@ -345,11 +369,60 @@ pub fn parse_ansible_reqs( Yaml::String(key) if key == "inventory" => { ret.inventories = parse_inventories(value)?; } + Yaml::String(key) if key == "vault_password" => { + let Yaml::String(filename) = value else { + return Err(anyhow!( + "Vault Password File expects a String containing the file name" + )); + }; + ret.vault_password = Some(filename.to_string()); + } + Yaml::String(key) if key == "vault_id" => { + let Yaml::Array(filenames) = value else { + return Err(anyhow!("Vault ID field expects an array of strings in the format: `label@filename`")); + }; + + for f in filenames { + let Yaml::String(filename) = f else { + return Err(anyhow!("The elements of the vault_id field should be strings in the format: `label@filename`")); + }; + ret.vault_id.push(filename.to_string()); + } + } Yaml::String(key) if key == "options" => { if let Yaml::Array(opts) = &value { ret.options = parse_ansible_options(opts); } } + Yaml::String(key) if key == "git_repos" => { + let Yaml::Array(repos) = &value else { + return Err(anyhow!("git_repos field expects an array of repos")); + }; + + for r in repos { + ret.git_repos.push( + parse_git_repo(r) + .map_err(|e| anyhow!("Failed to parse git repo: {e}"))?, + ); + } + } + Yaml::String(key) if key == "git_ssh_identity" => { + let Yaml::Array(indentities) = &value else { + return Err(anyhow!( + "git_ssh_identity expects an array of windmill variables (or secrets) containing ssh IDs" + )); + }; + + for r in indentities { + let Yaml::String(file_name) = r else { + return Err(anyhow!( + "Git ssh identity file must be a string path to a Windmill variable/secret" + )); + }; + + ret.git_ssh_identity.push(file_name.clone()); + } + } Yaml::String(key) => logs.push_str(&format!("\nUnknown field `{}`. Ignoring", key)), _ => (), } @@ -364,6 +437,38 @@ pub fn parse_ansible_reqs( Ok((logs, Some(ret), out_str)) } +fn parse_git_repo(r: &Yaml) -> anyhow::Result { + let Yaml::Hash(repo) = r else { + return Err(anyhow!("Should be a Map")); + }; + + let url = repo + .get(&Yaml::String("url".to_string())) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + .ok_or(anyhow!("Expected `url` field"))?; + + let target_path = repo + .get(&Yaml::String("target".to_string())) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + .ok_or(anyhow!( + "Expected `target` field (target directory for cloning the repo)" + ))?; + + let branch = repo + .get(&Yaml::String("branch".to_string())) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + let commit = repo + .get(&Yaml::String("commit".to_string())) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + Ok(GitRepo { url, commit, branch, target_path }) +} + fn parse_ansible_options(opts: &Vec) -> AnsiblePlaybookOptions { let mut ret = AnsiblePlaybookOptions { verbosity: None, @@ -529,3 +634,78 @@ fn yaml_to_json(yaml: &Yaml) -> serde_json::Value { _ => serde_json::Value::Null, } } + +fn update_versions( + section: &str, + yaml: &mut Yaml, + versions: &HashMap, +) -> anyhow::Result { + let mut logs = String::new(); + + let Yaml::Hash(ref mut m) = yaml else { + return Err(anyhow!("{section} dependency should be a map")); + }; + + if let Some(Yaml::Array(elements)) = m.get_mut(&Yaml::String(section.to_string())) { + for el in elements { + let Yaml::Hash(ref mut h) = el else { + return Err(anyhow!("{section} dependency element should be a map")); + }; + + if let Some(name) = h + .get(&Yaml::String("name".to_string())) + .and_then(|n| n.as_str()) + { + if let Some(version) = versions.get(name) { + h.insert( + Yaml::String("version".to_string()), + Yaml::String(version.to_string()), + ); + } else { + logs.push_str(&format!("WARNING: {section} dependency `{name}` has no locked version, using the latest or system installed version.\n")); + } + } else { + return Err(anyhow!( + "{section} dependency element: missing or invalid `name` field" + )); + } + } + } + + Ok(logs) +} + +pub fn add_versions_to_requirements_yaml( + input: &str, + role_versions: &HashMap, + collection_versions: &HashMap, +) -> anyhow::Result<(String,String)> { + let mut docs = + YamlLoader::load_from_str(input).map_err(|e| anyhow!("YAML parse error: {}", e))?; + let doc = &mut docs[0]; + + let mut logs = String::new(); + + logs.push_str( + &update_versions("roles", doc, role_versions) + .map_err(|e| anyhow!("Error updating role versions: {e}"))?, + ); + logs.push_str( + &update_versions("collections", doc, collection_versions) + .map_err(|e| anyhow!("Error updating collection versions: {e}"))?, + ); + + if !logs.is_empty() { + logs.push_str("WARNING: You might want to try adding manual versions for these, otherwise there could be breaking changes on deployed scripts\n"); + } + + let mut out_str = String::new(); + { + let mut emitter = YamlEmitter::new(&mut out_str); + emitter + .dump(doc) + .map_err(|e| anyhow!("YAML emit error: {}", e))?; + } + + Ok((out_str, logs)) +} diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index cdffe6f6a1..b65b50a33a 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -381,11 +381,10 @@ fn normalize_path(path: &Path) -> PathBuf { } ret } -pub fn write_file_at_user_defined_location( + +pub fn is_allowed_file_location( job_dir: &str, user_defined_path: &str, - content: &str, - mode: Option, ) -> error::Result { let job_dir = Path::new(job_dir); let user_path = PathBuf::from(user_defined_path); @@ -405,6 +404,17 @@ pub fn write_file_at_user_defined_location( .into()); } + Ok(normalized_full_path) +} + +pub fn write_file_at_user_defined_location( + job_dir: &str, + user_defined_path: &str, + content: &str, + mode: Option, +) -> error::Result { + let normalized_full_path = is_allowed_file_location(job_dir, user_defined_path)?; + let full_path = normalized_full_path.as_path(); if let Some(parent_dir) = full_path.parent() { std::fs::create_dir_all(parent_dir)?; diff --git a/backend/windmill-worker/src/ansible_executor.rs b/backend/windmill-worker/src/ansible_executor.rs index 7e0d1bd09c..785d1fc868 100644 --- a/backend/windmill-worker/src/ansible_executor.rs +++ b/backend/windmill-worker/src/ansible_executor.rs @@ -5,19 +5,22 @@ use std::{collections::HashMap, os::unix::fs::PermissionsExt, path::PathBuf, pro use std::{collections::HashMap, path::PathBuf, process::Stdio}; use anyhow::anyhow; +use futures::future::try_join_all; use itertools::Itertools; +use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use tokio::process::Command; use uuid::Uuid; use windmill_common::{ error, worker::{ - to_raw_value, write_file, write_file_at_user_defined_location, Connection, WORKER_CONFIG, + is_allowed_file_location, to_raw_value, write_file, write_file_at_user_defined_location, + Connection, WORKER_CONFIG, }, }; use windmill_queue::MiniPulledJob; -use windmill_parser_yaml::{AnsibleRequirements, ResourceOrVariablePath}; +use windmill_parser_yaml::{AnsibleRequirements, GitRepo, ResourceOrVariablePath}; use windmill_queue::{append_logs, CanceledBy}; use crate::{ @@ -28,8 +31,8 @@ use crate::{ }, handle_child::handle_child, python_executor::{create_dependencies_dir, handle_python_reqs, uv_pip_compile, PyVersion}, - AuthedClient, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV, PROXY_ENVS, - PY_INSTALL_DIR, TZ_ENV, + AuthedClient, DISABLE_NSJAIL, DISABLE_NUSER, GIT_PATH, HOME_ENV, NSJAIL_PATH, PATH_ENV, + PROXY_ENVS, PY_INSTALL_DIR, TZ_ENV, }; lazy_static::lazy_static! { @@ -41,7 +44,294 @@ lazy_static::lazy_static! { } const NSJAIL_CONFIG_RUN_ANSIBLE_CONTENT: &str = include_str!("../nsjail/run.ansible.config.proto"); +const WINDMILL_ANSIBLE_PASSWORD_FILENAME: &str = ".windmill.ansible_vault_password_file"; +async fn clone_repo( + repo: &GitRepo, + job_dir: &str, + job_id: &Uuid, + worker_name: &str, + conn: &Connection, + mem_peak: &mut i32, + canceled_by: &mut Option, + w_id: &str, + occupancy_metrics: &mut OccupancyMetrics, + git_ssh_cmd: &str, +) -> error::Result { + let target_path = is_allowed_file_location(job_dir, &repo.target_path)?; + + let mut clone_cmd = Command::new(GIT_PATH.as_str()); + clone_cmd + .current_dir(job_dir) + .env_clear() + .envs(PROXY_ENVS.clone()) + .env("PATH", PATH_ENV.as_str()) + .env("TZ", TZ_ENV.as_str()) + .env("GIT_SSH_COMMAND", git_ssh_cmd) + .args(["clone", "--quiet"]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + if let Some(branch) = &repo.branch { + clone_cmd.args(["--branch", branch]); + } + clone_cmd.arg(&repo.url); + clone_cmd.arg(&target_path); + + let clone_cmd_child = start_child_process(clone_cmd, GIT_PATH.as_str()).await?; + handle_child( + job_id, + conn, + mem_peak, + canceled_by, + clone_cmd_child, + false, + worker_name, + w_id, + "git clone", + None, + false, + &mut Some(occupancy_metrics), + None, + ) + .await?; + + // Checkout specific commit if provided + if let Some(commit) = &repo.commit { + let mut checkout_cmd = Command::new(GIT_PATH.as_str()); + checkout_cmd + .current_dir(job_dir) + .env_clear() + .envs(PROXY_ENVS.clone()) + .env("PATH", PATH_ENV.as_str()) + .env("TZ", TZ_ENV.as_str()) + .env("GIT_SSH_COMMAND", git_ssh_cmd) + .arg("-C") + .arg(&target_path) + .args(["checkout", "--quiet", commit]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let checkout_cmd_child = start_child_process(checkout_cmd, GIT_PATH.as_str()).await?; + handle_child( + job_id, + conn, + mem_peak, + canceled_by, + checkout_cmd_child, + false, + worker_name, + w_id, + "git checkout", + None, + false, + &mut Some(occupancy_metrics), + None, + ) + .await?; + } + + let mut rev_parse_cmd = Command::new(GIT_PATH.as_str()); + + let commit_hash_output = rev_parse_cmd + .current_dir(job_dir) + .env_clear() + .envs(PROXY_ENVS.clone()) + .env("PATH", PATH_ENV.as_str()) + .env("TZ", TZ_ENV.as_str()) + .env("GIT_SSH_COMMAND", git_ssh_cmd) + .arg("-C") + .arg(&target_path) + .args(["rev-parse", "HEAD"]) + .stderr(Stdio::piped()) + .output() + .await?; + + if !commit_hash_output.status.success() { + let stderr = String::from_utf8(commit_hash_output.stderr)?; + return Err(anyhow!("Error getting git repo commit hash: {stderr}").into()); + } + + let commit_hash = String::from_utf8(commit_hash_output.stdout)? + .trim() + .to_string(); + + Ok(commit_hash) +} + +pub fn create_empty_dir(path: &PathBuf) -> std::io::Result<()> { + if path.exists() { + if path.is_dir() { + let mut entries = std::fs::read_dir(&path)?; + if entries.next().is_some() { + return Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!( + "Directory '{}' already exists and is not empty", + path.display() + ), + )); + } + Ok(()) + } else { + Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!("Path '{}' exists and is not a directory", path.display()), + )) + } + } else { + std::fs::create_dir_all(path) + } +} + +async fn clone_repo_without_history( + repo: &GitRepo, + full_commit: &str, + job_dir: &str, + job_id: &Uuid, + worker_name: &str, + conn: &Connection, + mem_peak: &mut i32, + canceled_by: &mut Option, + w_id: &str, + occupancy_metrics: &mut OccupancyMetrics, + git_ssh_cmd: &str, +) -> error::Result<()> { + let target_path = is_allowed_file_location(job_dir, &repo.target_path)?; + + create_empty_dir(&target_path)?; + + let mut init_cmd = Command::new(GIT_PATH.as_str()); + init_cmd + .current_dir(job_dir) + .env_clear() + .envs(PROXY_ENVS.clone()) + .env("PATH", PATH_ENV.as_str()) + .env("TZ", TZ_ENV.as_str()) + .arg("-C") + .arg(&target_path) + .args(["init", "--quiet"]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + if let Some(branch) = &repo.branch { + init_cmd.args(["--initial-branch", branch]); + } + + let init_cmd_child = start_child_process(init_cmd, GIT_PATH.as_str()).await?; + handle_child( + job_id, + conn, + mem_peak, + canceled_by, + init_cmd_child, + false, + worker_name, + w_id, + "git init", + None, + false, + &mut Some(occupancy_metrics), + None, + ) + .await?; + + let mut add_remote_cmd = Command::new(GIT_PATH.as_str()); + add_remote_cmd + .current_dir(job_dir) + .env_clear() + .envs(PROXY_ENVS.clone()) + .env("PATH", PATH_ENV.as_str()) + .env("TZ", TZ_ENV.as_str()) + .env("GIT_SSH_COMMAND", git_ssh_cmd) + .arg("-C") + .arg(&target_path) + .args(vec!["remote", "add", "origin", &repo.url]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let add_remote_cmd_child = start_child_process(add_remote_cmd, GIT_PATH.as_str()).await?; + handle_child( + job_id, + conn, + mem_peak, + canceled_by, + add_remote_cmd_child, + false, + worker_name, + w_id, + "git add remote", + None, + false, + &mut Some(occupancy_metrics), + None, + ) + .await?; + + let mut fetch_cmd = Command::new(GIT_PATH.as_str()); + fetch_cmd + .current_dir(job_dir) + .env_clear() + .envs(PROXY_ENVS.clone()) + .env("PATH", PATH_ENV.as_str()) + .env("TZ", TZ_ENV.as_str()) + .env("GIT_SSH_COMMAND", git_ssh_cmd) + .arg("-C") + .arg(&target_path) + .args(vec!["fetch", "--depth=1", "--quiet", "origin", full_commit]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let fetch_cmd_child = start_child_process(fetch_cmd, GIT_PATH.as_str()).await?; + handle_child( + job_id, + conn, + mem_peak, + canceled_by, + fetch_cmd_child, + false, + worker_name, + w_id, + "git fetch", + None, + false, + &mut Some(occupancy_metrics), + None, + ) + .await?; + + let mut checkout_cmd = Command::new(GIT_PATH.as_str()); + checkout_cmd + .current_dir(job_dir) + .env_clear() + .envs(PROXY_ENVS.clone()) + .env("PATH", PATH_ENV.as_str()) + .env("TZ", TZ_ENV.as_str()) + .env("GIT_SSH_COMMAND", git_ssh_cmd) + .arg("-C") + .arg(&target_path) + .args(["checkout", "--quiet", "FETCH_HEAD"]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let checkout_cmd_child = start_child_process(checkout_cmd, GIT_PATH.as_str()).await?; + handle_child( + job_id, + conn, + mem_peak, + canceled_by, + checkout_cmd_child, + false, + worker_name, + w_id, + "git checkout", + None, + false, + &mut Some(occupancy_metrics), + None, + ) + .await?; + + Ok(()) +} async fn handle_ansible_python_deps( job_dir: &str, requirements_o: Option<&String>, @@ -118,7 +408,7 @@ async fn handle_ansible_python_deps( Ok(additional_python_paths) } -async fn install_galaxy_collections( +pub async fn install_galaxy_collections( collections_yml: &str, job_dir: &str, job_id: &Uuid, @@ -128,6 +418,7 @@ async fn install_galaxy_collections( canceled_by: &mut Option, conn: &Connection, occupancy_metrics: &mut OccupancyMetrics, + git_ssh_cmd: &str, ) -> anyhow::Result<()> { write_file(job_dir, "requirements.yml", collections_yml)?; @@ -138,15 +429,52 @@ async fn install_galaxy_collections( conn, ) .await; - let mut galaxy_command = Command::new(ANSIBLE_GALAXY_PATH.as_str()); - galaxy_command + + let mut galaxy_roles_cmd = Command::new(ANSIBLE_GALAXY_PATH.as_str()); + galaxy_roles_cmd .current_dir(job_dir) .env_clear() .envs(PROXY_ENVS.clone()) .env("PATH", PATH_ENV.as_str()) .env("TZ", TZ_ENV.as_str()) - // .env("BASE_INTERNAL_URL", base_internal_url) - // .env("HOME", HOME_ENV.as_str()) + .env("GIT_SSH_COMMAND", git_ssh_cmd) + .args(vec![ + "role", + "install", + "-r", + "requirements.yml", + "-p", + "./roles", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + let child = start_child_process(galaxy_roles_cmd, ANSIBLE_GALAXY_PATH.as_str()).await?; + handle_child( + job_id, + conn, + mem_peak, + canceled_by, + child, + !*DISABLE_NSJAIL, + worker_name, + w_id, + "ansible-galaxy role install", + None, + false, + &mut Some(occupancy_metrics), + None, + ) + .await?; + + let mut galaxy_collections_cmd = Command::new(ANSIBLE_GALAXY_PATH.as_str()); + galaxy_collections_cmd + .current_dir(job_dir) + .env_clear() + .envs(PROXY_ENVS.clone()) + .env("PATH", PATH_ENV.as_str()) + .env("TZ", TZ_ENV.as_str()) + .env("GIT_SSH_COMMAND", git_ssh_cmd) .args(vec![ "collection", "install", @@ -158,7 +486,7 @@ async fn install_galaxy_collections( .stdout(Stdio::piped()) .stderr(Stdio::piped()); - let child = start_child_process(galaxy_command, ANSIBLE_GALAXY_PATH.as_str()).await?; + let child = start_child_process(galaxy_collections_cmd, ANSIBLE_GALAXY_PATH.as_str()).await?; handle_child( job_id, conn, @@ -168,7 +496,7 @@ async fn install_galaxy_collections( !*DISABLE_NSJAIL, worker_name, w_id, - "ansible galaxy install", + "ansible-galaxy collection install", None, false, &mut Some(occupancy_metrics), @@ -179,6 +507,258 @@ async fn install_galaxy_collections( Ok(()) } +#[derive(Serialize, Deserialize)] +pub struct AnsibleDependencyLocks { + pub python_lockfile: String, + pub git_repos: HashMap, // URL to full commit hash + pub collections_and_roles: String, + pub collections_and_roles_logs: String, + // pub collection_versions: HashMap, // + // pub role_versions: HashMap, +} + +pub async fn get_collection_locks( + job_dir: &str, +) -> anyhow::Result<(HashMap, String)> { + let mut ansible_cmd = Command::new(ANSIBLE_GALAXY_PATH.as_str()); + + ansible_cmd + .current_dir(job_dir) + .args(["collection", "list", "--format", "json", "-p", "./"]); + + let output = ansible_cmd.output().await?; + + let mut ret = HashMap::new(); + let mut logs = String::new(); + + if !output.status.success() { + let stderr = String::from_utf8(output.stderr)?; + return Err(anyhow!( + "Error getting ansible collection versions: {stderr}" + )); + } + + let stdout = String::from_utf8(output.stdout)?; + + let val: serde_json::Value = serde_json::from_str(&stdout)?; + + let Some(own_collections) = val.get(format!("{}/ansible_collections", job_dir)) else { + return Ok((ret, logs)); + }; + + let collections = own_collections.as_object().ok_or(anyhow!( + "Expected an object (map) for the `ansible-galaxy collection list` command output and got {}", + own_collections + ))?; + + for (c_name, c) in collections.iter() { + if let Some(v) = c.get("version").and_then(|v| v.as_str()) { + // TODO: Check if version is not something like `(undefined)` + ret.insert(c_name.clone(), v.to_string()); + } else { + logs.push_str(&format!("Failed to get version for collection `{}`. Expected an object with a string in the `version` field but received {}\n", c_name, c)); + } + } + + Ok((ret, logs)) +} + +pub async fn get_role_locks(job_dir: &str) -> anyhow::Result<(HashMap, String)> { + let mut ansible_cmd = Command::new(ANSIBLE_GALAXY_PATH.as_str()); + + ansible_cmd + .current_dir(job_dir) + .args(["role", "list", "-p", "./roles"]); + + let output = ansible_cmd.output().await?; + let mut ret = HashMap::new(); + let mut logs = String::new(); + + if !output.status.success() { + let stderr = String::from_utf8(output.stderr)?; + logs.push_str(&format!("Error getting ansible role versions: {stderr}")); + return Ok((ret, logs)); + } + + let stdout = String::from_utf8(output.stdout)?; + + let mut lines = stdout.lines(); + + while let Some(line) = lines.next() { + if line == format!("# {}/roles", job_dir) { + break; + } + } + + for line in lines { + let line = line.strip_prefix("-").unwrap_or(line); + let mut cols = line.split(","); + + if let Some(name) = cols.next().map(|n| n.trim()) { + if let Some(version) = cols.next().map(|v| v.trim()) { + // TODO: Check if version is not something like `(undefined)` + ret.insert(name.to_string(), version.to_string()); + } else { + logs.push_str(&format!("Failed to get version for role `{}`.", name)); + } + } + } + + Ok((ret, logs)) +} + +pub async fn get_git_repo_full_head_commit_hash( + repo: &GitRepo, + git_ssh_cmd: &str, +) -> anyhow::Result { + let mut git_cmd = Command::new(GIT_PATH.as_str()); + + git_cmd + .env("GIT_SSH_COMMAND", git_ssh_cmd) + .args(["ls-remote", &repo.url, "HEAD"]); + + let output = git_cmd.stderr(Stdio::piped()).output().await?; + + if !output.status.success() { + let stderr = String::from_utf8(output.stderr)?; + return Err(anyhow!("Error getting git repo commit hash: {stderr}")); + } + + let stdout = String::from_utf8(output.stdout)?; + + let lines: Vec<&str> = stdout.lines().collect(); + + if lines.len() != 1 { + return Err(anyhow!("Unexpected output format for git ls-remote",)); + } + + Ok(lines + .first() + .ok_or(anyhow!( + "The HEAD commit hash was not found for repo `{}`", + &repo.url + ))? + .split_whitespace() + .next() + .map(|s| s.to_string()) + .ok_or(anyhow!("Unexpected output format for git ls-remote"))?) +} + +pub async fn get_git_repos_lock( + repos: &Vec, + job_dir: &str, + job_id: &Uuid, + worker_name: &str, + conn: &Connection, + mem_peak: &mut i32, + canceled_by: &mut Option, + w_id: &str, + occupancy_metrics: &mut OccupancyMetrics, + git_ssh_cmd: &str, +) -> anyhow::Result> { + let mut ret = HashMap::new(); + + for repo in repos { + if repo.commit.is_some() { + ret.insert( + repo.url.to_string(), + clone_repo( + repo, + job_dir, + job_id, + worker_name, + conn, + mem_peak, + canceled_by, + w_id, + occupancy_metrics, + git_ssh_cmd, + ) + .await?, + ); + } else { + ret.insert( + repo.url.to_string(), + get_git_repo_full_head_commit_hash(repo, git_ssh_cmd).await?, + ); + } + } + + Ok(ret) +} + +pub fn create_ansible_cfg( + reqs: Option<&AnsibleRequirements>, + job_dir: &str, + vault_password_file_exists: bool, +) -> error::Result<()> { + let mut passwords_cfg = String::new(); + if vault_password_file_exists { + passwords_cfg.push_str(&format!( + "vault_password_file = {WINDMILL_ANSIBLE_PASSWORD_FILENAME}\n" + )); + } + if let Some(vault_ids) = reqs.as_ref().map(|r| &r.vault_id) { + if !vault_ids.is_empty() { + let password_files = vault_ids.join(","); + + passwords_cfg.push_str(&format!("vault_identity_list = {password_files}\n")); + } + } + let ansible_cfg_content = format!( + r#" +[defaults] +collections_path = ./ +roles_path = ./roles +home={job_dir}/.ansible +local_tmp={job_dir}/.ansible/tmp +remote_tmp={job_dir}/.ansible/tmp +{passwords_cfg} +"# + ); + + write_file(job_dir, "ansible.cfg", &ansible_cfg_content)?; + + Ok(()) +} + +pub async fn get_git_ssh_cmd( + reqs: &AnsibleRequirements, + job_dir: &str, + client: &AuthedClient, +) -> error::Result { + let ssh_id_files = try_join_all(reqs.git_ssh_identity.iter().enumerate().map( + async |(i, var_path)| -> error::Result { + let id_file_name = format!(".ssh_id_priv_{}", i); + let loc = is_allowed_file_location(job_dir, &id_file_name)?; + + let mut content = client.get_variable_value(var_path).await.map_err(|e| { + error::Error::NotFound(format!( + "Variable {var_path} not found for git ssh identity: {e:#}" + )) + })?; + content.push_str("\n"); + + let file = write_file(job_dir, &id_file_name, &content)?; + + #[cfg(unix)] + { + let perm = std::os::unix::fs::PermissionsExt::from_mode(0o600); + file.set_permissions(perm)?; + } + + Ok(format!( + " -i '{}'", + loc.to_string_lossy().replace('\'', r"'\''") + )) + }, + )) + .await?; + + let git_ssh_cmd = format!("ssh -o StrictHostKeyChecking=no{}", ssh_id_files.join("")); + Ok(git_ssh_cmd) +} + pub async fn handle_ansible_job( requirements_o: Option<&String>, job_dir: &str, @@ -202,13 +782,17 @@ pub async fn handle_ansible_job( "ansible", )?; + let req_lockfiles: Option = requirements_o + .map(|s| serde_json::from_str(s)) + .transpose()?; + let (logs, reqs, playbook) = windmill_parser_yaml::parse_ansible_reqs(inner_content)?; append_logs(&job.id, &job.workspace_id, logs, conn).await; write_file(job_dir, "main.yml", &playbook)?; let additional_python_paths = handle_ansible_python_deps( job_dir, - requirements_o, + req_lockfiles.as_ref().map(|r| &r.python_lockfile), reqs.as_ref(), &job.workspace_id, &job.id, @@ -221,6 +805,11 @@ pub async fn handle_ansible_job( ) .await?; + let git_ssh_cmd = &match &reqs { + Some(r) => get_git_ssh_cmd(r, job_dir, client).await?, + None => "ssh".to_string(), + }; + let interpolated_args; if let Some(args) = &job.args { let mut args = args.0.clone(); @@ -268,7 +857,7 @@ pub async fn handle_ansible_job( .unwrap_or_else(|| vec![]); let mut nsjail_extra_mounts = vec![]; - if let Some(r) = reqs { + if let Some(r) = reqs.as_ref() { if let Some(db) = conn.as_sql() { nsjail_extra_mounts = create_file_resources( &job.id, @@ -282,9 +871,82 @@ pub async fn handle_ansible_job( .await?; } - if let Some(collections) = r.collections { + for repo in &r.git_repos { + append_logs( + &job.id, + &job.workspace_id, + format!("\nCloning {}...\n", &repo.url), + conn, + ) + .await; + if let Some(full_commit_hash) = req_lockfiles + .as_ref() + .and_then(|r| r.git_repos.get(&repo.url)) + { + clone_repo_without_history( + repo, + full_commit_hash, + job_dir, + &job.id, + worker_name, + conn, + mem_peak, + canceled_by, + &job.workspace_id, + occupancy_metrics, + git_ssh_cmd, + ) + .await + .map_err(|e| anyhow!("Failed to clone git repo `{}`: {e}", repo.url))?; + } else { + if req_lockfiles.is_some() { + append_logs( + &job.id, + &job.workspace_id, + format!("Warning: `{}` is using latest commit because the lockfile didn't store a commit hash for this repo. Updates to the repo could break the deployed playbook.\n", &repo.url), + conn, + ) + .await; + } + clone_repo( + repo, + job_dir, + &job.id, + worker_name, + conn, + mem_peak, + canceled_by, + &job.workspace_id, + occupancy_metrics, + git_ssh_cmd, + ) + .await + .map_err(|e| anyhow!("Failed to clone git repo `{}`: {e}", repo.url))?; + } + + append_logs( + &job.id, + &job.workspace_id, + format!("Cloned {} into {}\n", &repo.url, &repo.target_path), + conn, + ) + .await; + } + + if let Some(collections) = r.roles_and_collections.as_ref() { + let empty = String::new(); + let (lockfile, logs) = + req_lockfiles + .as_ref() + .map(|r| (&r.collections_and_roles, &r.collections_and_roles_logs)) + .unwrap_or((collections, &empty)); + + if !logs.is_empty() { + append_logs(&job.id, &job.workspace_id, logs, conn).await; + } + install_galaxy_collections( - collections.as_str(), + lockfile, job_dir, &job.id, worker_name, @@ -293,10 +955,12 @@ pub async fn handle_ansible_job( canceled_by, conn, occupancy_metrics, + git_ssh_cmd, ) .await?; } } + append_logs( &job.id, &job.workspace_id, @@ -304,17 +968,17 @@ pub async fn handle_ansible_job( conn, ) .await; - let ansible_cfg_content = format!( - r#" -[defaults] -collections_path = ./ -roles_path = ./roles -home={job_dir}/.ansible -local_tmp={job_dir}/.ansible/tmp -remote_tmp={job_dir}/.ansible/tmp -"# - ); - write_file(job_dir, "ansible.cfg", &ansible_cfg_content)?; + + let vault_password_file_exists = match reqs.as_ref().and_then(|x| x.vault_password.as_ref()) { + Some(var_path) => { + let password = client.get_variable_value(&var_path).await?; + write_file(job_dir, WINDMILL_ANSIBLE_PASSWORD_FILENAME, &password)?; + true + } + None => false, + }; + + create_ansible_cfg(reqs.as_ref(), job_dir, vault_password_file_exists)?; let mut reserved_variables = get_reserved_variables(job, &client.token, conn, parent_runnable_path).await?; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index fed83e07e0..df80b2cb36 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -306,6 +306,7 @@ lazy_static::lazy_static! { pub static ref NSJAIL_PATH: String = std::env::var("NSJAIL_PATH").unwrap_or_else(|_| "nsjail".to_string()); pub static ref PATH_ENV: String = std::env::var("PATH").unwrap_or_else(|_| String::new()); pub static ref HOME_ENV: String = std::env::var("HOME").unwrap_or_else(|_| "/tmp".to_string()); + pub static ref GIT_PATH: String = std::env::var("GIT_PATH").unwrap_or_else(|_| "/usr/bin/git".to_string()); pub static ref NODE_PATH: Option = std::env::var("NODE_PATH").ok(); diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 93cc355eb0..80dec9a70a 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -3,6 +3,8 @@ use std::collections::HashMap; use std::fs::{create_dir_all, remove_dir_all}; use std::path::{Component, Path, PathBuf}; +#[cfg(feature = "python")] +use crate::ansible_executor::{get_git_repos_lock, AnsibleDependencyLocks}; use async_recursion::async_recursion; use serde_json::value::RawValue; use serde_json::{json, Value}; @@ -18,6 +20,8 @@ use windmill_common::scripts::ScriptHash; #[cfg(feature = "python")] use windmill_common::worker::PythonAnnotations; use windmill_common::worker::{to_raw_value, to_raw_value_owned, write_file, Connection}; +#[cfg(feature = "python")] +use windmill_parser_yaml::AnsibleRequirements; use windmill_common::{ apps::AppScriptId, @@ -1917,6 +1921,123 @@ async fn python_dep( req } +#[cfg(feature = "python")] +async fn ansible_dep( + reqs: AnsibleRequirements, + job_id: &Uuid, + mem_peak: &mut i32, + canceled_by: &mut Option, + job_dir: &str, + db: &sqlx::Pool, + worker_name: &str, + w_id: &str, + worker_dir: &str, + occupancy_metrics: &mut OccupancyMetrics, + token: &str, + base_internal_url: &str, +) -> std::result::Result { + use windmill_parser_yaml::add_versions_to_requirements_yaml; + + use crate::{ + ansible_executor::{ + create_ansible_cfg, get_collection_locks, get_git_ssh_cmd, get_role_locks, + install_galaxy_collections, + }, + AuthedClient, + }; + + let python_lockfile = python_dep( + reqs.python_reqs.join("\n").to_string(), + job_id, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + w_id, + worker_dir, + &mut Some(occupancy_metrics), + None, + PythonAnnotations::default(), + ) + .await?; + + let conn = &Connection::Sql(db.clone()); + + let authed_client = AuthedClient { + base_internal_url: base_internal_url.to_string(), + token: token.to_string(), + workspace: w_id.to_string(), + force_client: None, + }; + + let git_ssh_cmd = get_git_ssh_cmd(&reqs, job_dir, &authed_client).await?; + + let git_repos = get_git_repos_lock( + &reqs.git_repos, + job_dir, + job_id, + worker_name, + conn, + mem_peak, + canceled_by, + w_id, + occupancy_metrics, + &git_ssh_cmd, + ) + .await?; + + let ansible_lockfile; + + create_ansible_cfg(Some(&reqs), job_dir, false)?; + + if let Some(collections) = reqs.roles_and_collections.as_ref() { + install_galaxy_collections( + collections, + job_dir, + job_id, + worker_name, + w_id, + mem_peak, + canceled_by, + conn, + occupancy_metrics, + &git_ssh_cmd, + ) + .await?; + + let (collection_versions, logs1) = get_collection_locks(job_dir).await?; + + let (role_versions, logs2) = if collections.contains("roles:") { + get_role_locks(job_dir).await? + } else { + (HashMap::new(), String::new()) + }; + + let (reqs_yaml, logs3) = add_versions_to_requirements_yaml(&collections, &role_versions, &collection_versions)?; + + let logs = format!("\n{logs1}\n{logs2}\n{logs3}\n"); + + append_logs(job_id, w_id, &logs, conn).await; + + ansible_lockfile = AnsibleDependencyLocks { + python_lockfile, + git_repos, + collections_and_roles: reqs_yaml, + collections_and_roles_logs: logs, + }; + } else { + ansible_lockfile = AnsibleDependencyLocks { + python_lockfile, + git_repos, + collections_and_roles: String::new(), + collections_and_roles_logs: String::new(), + }; + } + + serde_json::to_string(&ansible_lockfile).map_err(|e| e.into()) +} + async fn capture_dependency_job( job_id: &Uuid, job_language: &ScriptLang, @@ -2001,10 +2122,9 @@ async fn capture_dependency_job( )); } let (_logs, reqs, _) = windmill_parser_yaml::parse_ansible_reqs(job_raw_code)?; - let reqs = reqs.map(|r| r.python_reqs.join("\n")).unwrap_or_default(); - python_dep( - reqs, + ansible_dep( + reqs.unwrap_or_default(), job_id, mem_peak, canceled_by, @@ -2013,9 +2133,9 @@ async fn capture_dependency_job( worker_name, w_id, worker_dir, - &mut Some(occupancy_metrics), - None, - PythonAnnotations::default(), + occupancy_metrics, + token, + base_internal_url, ) .await } diff --git a/flake.nix b/flake.nix index f821279959..5e50c9fc0e 100644 --- a/flake.nix +++ b/flake.nix @@ -94,6 +94,7 @@ oracle-instantclient # LSP/Local dev svelte-language-server + ansible taplo ]); packages = [ @@ -158,6 +159,7 @@ ]; inherit PKG_CONFIG_PATH RUSTY_V8_ARCHIVE; + GIT_PATH = "${pkgs.git}/bin/git"; NODE_ENV = "development"; NODE_OPTIONS = "--max-old-space-size=16384"; DATABASE_URL = "postgres://postgres:changeme@127.0.0.1:5432/"; @@ -172,12 +174,14 @@ JAVA_PATH = "${pkgs.jdk21}/bin/java"; JAVAC_PATH = "${pkgs.jdk21}/bin/javac"; COURSIER_PATH = "${coursier}/coursier"; - # for related places search: ADD_NEW_LANG + # for related places search: ADD_NEW_LANG FLOCK_PATH = "${pkgs.flock}/bin/flock"; CARGO_PATH = "${rust}/bin/cargo"; DOTNET_PATH = "${pkgs.dotnet-sdk_9}/bin/dotnet"; DOTNET_ROOT = "${pkgs.dotnet-sdk_9}/share/dotnet"; ORACLE_LIB_DIR = "${pkgs.oracle-instantclient.lib}/lib"; + ANSIBLE_PLAYBOOK_PATH = "${pkgs.ansible}/bin/ansible-playbook"; + ANSIBLE_GALAXY_PATH = "${pkgs.ansible}/bin/ansible-galaxy"; RUST_LOG = "debug"; }; packages.default = self.packages.${system}.windmill; diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts index 9e90ebe74c..a9a0ceb4de 100644 --- a/frontend/src/lib/script_helpers.ts +++ b/frontend/src/lib/script_helpers.ts @@ -931,6 +931,12 @@ inventory: - resource_type: ansible_inventory # You can pin an inventory to this script by hardcoding the resource path: # resource: u/user/your_resource +# - name: hcloud.yml +# resource_type: dynamic_inventory + +options: + - verbosity: vvv + # File resources will be written in the relative \`target\` location before # running the playbook @@ -946,11 +952,15 @@ extra_vars: world_qualifier: type: string +# If using Ansible Vault: +# vault_password: u/user/ansible_vault_password + dependencies: galaxy: collections: - name: community.general - name: community.vmware + roles: python: - jmespath ---