git.delta.rocks / jrsonnet / refs/commits / cc4ecd613a27

1use std::{2	cell::{LazyCell, OnceCell},3	collections::BTreeMap,4	env::current_dir,5	ffi::{OsStr, OsString},6	fmt::Display,7	io::Write,8	ops::Deref,9	path::PathBuf,10	str::FromStr,11	sync::{Arc, Mutex, MutexGuard, OnceLock},12};1314use anyhow::{anyhow, bail, ensure, Context, Result};15use clap::Parser;16use fleet_shared::SecretData;17use nix_eval::{nix_go, nix_go_json, util::assert_warn, NixSessionPool, Value};18use nom::{19	bytes::complete::take_while1,20	character::complete::char,21	combinator::{map, opt},22	multi::separated_list1,23	sequence::{preceded, separated_pair},24};25use openssh::SessionBuilder;26use serde::de::DeserializeOwned;27use tempfile::NamedTempFile;28use tracing::error;2930use crate::{31	command::MyCommand,32	fleetdata::{FleetData, FleetSecret, FleetSharedSecret},33};3435pub struct FleetConfigInternals {36	pub local_system: String,37	pub directory: PathBuf,38	pub opts: FleetOpts,39	pub data: Mutex<FleetData>,40	pub nix_args: Vec<OsString>,41	/// fleet_config.config42	pub config_field: Value,4344	/// import nixpkgs {system = local};45	pub default_pkgs: Value,46}4748#[derive(Clone)]49pub struct Config(Arc<FleetConfigInternals>);5051impl Deref for Config {52	type Target = FleetConfigInternals;5354	fn deref(&self) -> &Self::Target {55		&self.056	}57}5859#[derive(Clone, Copy, Debug)]60pub enum EscalationStrategy {61	Sudo,62	Run0,63	Su,64}6566pub struct ConfigHost {67	config: Config,68	pub name: String,69	pub local: bool,70	pub session: OnceLock<Arc<openssh::Session>>,71	groups: OnceCell<Vec<String>>,7273	pub host_config: Option<Value>,74	pub nixos_config: OnceCell<Value>,75}76impl ConfigHost {77	pub async fn escalation_strategy(&self) -> Result<EscalationStrategy> {78		// Prefer sudo, as run0 has some gotchas with polkit79		// and too many repeating prompts.80		if let Ok(_) = self.find_in_path("sudo").await {81			return Ok(EscalationStrategy::Sudo);82		}83		if let Ok(_) = self.find_in_path("run0").await {84			return Ok(EscalationStrategy::Run0);85		}86		Ok(EscalationStrategy::Su)87	}88	// TOCTOU is possible here in case if config is changed, but this case is not handled anywhere anyway,89	// assuming getting tags always returns the same value.90	pub async fn tags(&self) -> Result<Vec<String>> {91		if let Some(v) = self.groups.get() {92			return Ok(v.clone());93		}94		let Some(host_config) = &self.host_config else {95			return Ok(vec![]);96		};97		let tags: Vec<String> = nix_go_json!(host_config.tags);9899		let _ = self.groups.set(tags.clone());100101		Ok(tags)102	}103	pub async fn nixos_config(&self) -> Result<Value> {104		if let Some(v) = self.nixos_config.get() {105			return Ok(v.clone());106		}107		let Some(host_config) = &self.host_config else {108			bail!("local host has no nixos_config");109		};110		let nixos_config = nix_go!(host_config.nixos.config);111		assert_warn("nixos config evaluation", &nixos_config).await?;112113		let _ = self.nixos_config.set(nixos_config.clone());114115		Ok(nixos_config)116	}117	async fn open_session(&self) -> Result<Arc<openssh::Session>> {118		assert!(!self.local, "do not open ssh connection to local session");119		// FIXME: TOCTOU120		if let Some(session) = &self.session.get() {121			return Ok((*session).clone());122		};123		let mut session = SessionBuilder::default();124		let session = session125			.connect(&self.name)126			.await127			.map_err(|e| anyhow!("ssh error while connecting to {}: {e}", self.name))?;128		let session = Arc::new(session);129		self.session.set(session.clone()).expect("TOCTOU happened");130		Ok(session)131	}132	pub async fn mktemp_dir(&self) -> Result<String> {133		let mut cmd = self.cmd("mktemp").await?;134		cmd.arg("-d");135		let path = cmd.run_string().await?;136		Ok(path.trim_end().to_owned())137	}138	pub async fn read_file_bin(&self, path: impl AsRef<OsStr>) -> Result<Vec<u8>> {139		let mut cmd = self.cmd("cat").await?;140		cmd.arg(path);141		cmd.run_bytes().await142	}143	pub async fn read_file_text(&self, path: impl AsRef<OsStr>) -> Result<String> {144		let mut cmd = self.cmd("cat").await?;145		cmd.arg(path);146		cmd.run_string().await147	}148	pub async fn read_dir(&self, path: impl AsRef<OsStr>) -> Result<Vec<String>> {149		let mut cmd = self.cmd("ls").await?;150		cmd.arg(path);151		let out = cmd.run_string().await?;152		let mut lines = out.split('\n');153		if let Some(last) = lines.next_back() {154			ensure!(last.is_empty(), "output of ls should end with newline");155		}156		Ok(lines.map(ToOwned::to_owned).collect())157	}158	#[allow(dead_code)]159	pub async fn read_file_json<D: DeserializeOwned>(&self, path: impl AsRef<OsStr>) -> Result<D> {160		let text = self.read_file_text(path).await?;161		Ok(serde_json::from_str(&text)?)162	}163	pub async fn read_env(&self, env: &str) -> Result<String> {164		let mut cmd = self.cmd("printenv").await?;165		cmd.arg(env);166		Ok(cmd.run_string().await?)167	}168	pub async fn find_in_path(&self, command: &str) -> Result<String> {169		// // `which` is not a part of coreutils, and it might not exist on machine.170		// let path = self.read_env("PATH").await?;171		// // Assuming delimiter is :, we don't work with windows host, this check will be much172		// // more sophisticated in remowt backend (and quicker, since actual PATH search will be done on remote machine)173		// for ele in path.split(':') {174		// 	let test_path = format!("{ele}/{cmd}");175		// 	test -x etc176		// }177		// let mut cmd = self.cmd("printenv").await?;178		// cmd.arg(env);179		// Ok(cmd.run_string().await?)180		// Assuming this is an environment issue if which doesn't exist, will be fixed with remowt.181		let mut cmd = self182			.cmd_escalation(183				// Not used184				EscalationStrategy::Su,185				"which",186			)187			.await?;188		cmd.arg(command);189		cmd.run_string().await190	}191	pub async fn read_file_value<D: FromStr>(&self, path: impl AsRef<OsStr>) -> Result<D>192	where193		<D as FromStr>::Err: Display,194	{195		let text = self.read_file_text(path).await?;196		D::from_str(&text).map_err(|e| anyhow!("failed to parse value: {e}"))197	}198	pub async fn cmd(&self, cmd: impl AsRef<OsStr>) -> Result<MyCommand> {199		self.cmd_escalation(self.escalation_strategy().await?, cmd)200			.await201	}202	pub async fn cmd_escalation(203		&self,204		escalation: EscalationStrategy,205		cmd: impl AsRef<OsStr>,206	) -> Result<MyCommand> {207		if self.local {208			Ok(MyCommand::new(escalation, cmd))209		} else {210			let session = self.open_session().await?;211			Ok(MyCommand::new_on(escalation, cmd, session))212		}213	}214215	pub async fn decrypt(&self, data: SecretData) -> Result<Vec<u8>> {216		ensure!(data.encrypted, "secret is not encrypted");217		let mut cmd = self.cmd("fleet-install-secrets").await?;218		cmd.arg("decrypt").eqarg("--secret", data.to_string());219		let encoded = cmd220			.sudo()221			.run_string()222			.await223			.context("failed to call remote host for decrypt")?;224		let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;225		ensure!(!data.encrypted, "secret came out encrypted");226		Ok(data.data)227	}228	pub async fn reencrypt(&self, data: SecretData, targets: Vec<String>) -> Result<SecretData> {229		ensure!(data.encrypted, "secret is not encrypted");230		let mut cmd = self.cmd("fleet-install-secrets").await?;231		cmd.arg("reencrypt").eqarg("--secret", data.to_string());232		for target in targets {233			let key = self.config.key(&target).await?;234			cmd.eqarg("--targets", key);235		}236		let encoded = cmd237			.sudo()238			.run_string()239			.await240			.context("failed to call remote host for decrypt")?;241		let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;242		ensure!(data.encrypted, "secret came out not encrypted");243		Ok(data)244	}245	/// Returns path for futureproofing, as path might change i.e on conversion to CA246	pub async fn remote_derivation(&self, path: &PathBuf) -> Result<PathBuf> {247		if self.local {248			// Path is located locally, thus already trusted.249			return Ok(path.to_owned());250		}251		let mut nix = MyCommand::new(252			// Not used253			EscalationStrategy::Su,254			"nix",255		);256		nix.arg("copy")257			.arg("--substitute-on-destination")258			.comparg("--to", format!("ssh-ng://{}", self.name))259			.arg(path);260		nix.run_nix().await.context("nix copy")?;261		Ok(path.to_owned())262	}263	pub async fn systemctl_stop(&self, name: &str) -> Result<()> {264		let mut cmd = self.cmd("systemctl").await?;265		cmd.arg("stop").arg(name);266		cmd.sudo().run().await267	}268	pub async fn systemctl_start(&self, name: &str) -> Result<()> {269		let mut cmd = self.cmd("systemctl").await?;270		cmd.arg("start").arg(name);271		cmd.sudo().run().await272	}273274	pub async fn rm_file(&self, path: impl AsRef<OsStr>, sudo: bool) -> Result<()> {275		let mut cmd = self.cmd("rm").await?;276		cmd.arg("-f").arg(path);277		if sudo {278			cmd = cmd.sudo()279		}280		cmd.run().await281	}282283	pub async fn list_configured_secrets(&self) -> Result<Vec<String>> {284		let nixos = self.nixos_config().await?;285		let secrets = nix_go!(nixos.secrets);286		let mut out = Vec::new();287		for name in secrets.list_fields().await? {288			let secret = nix_go!(secrets[{ name }]);289			let is_shared: bool = nix_go_json!(secret.shared);290			if is_shared {291				continue;292			}293			out.push(name);294		}295		Ok(out)296	}297	pub async fn secret_field(&self, name: &str) -> Result<Value> {298		let nixos = self.nixos_config().await?;299		Ok(nix_go!(nixos.secrets[{ name }]))300	}301302	/// Packages for this host, resolved with nixpkgs overlays303	pub async fn pkgs(&self) -> Result<Value> {304		let nixos = self.nixos_config().await?;305		Ok(nix_go!(nixos._resolvedPkgs))306	}307}308309impl Config {310	pub async fn should_skip(&self, host: &ConfigHost) -> Result<bool> {311		if !self.opts.skip.is_empty() && self.opts.skip.iter().any(|h| h as &str == host.name) {312			return Ok(true);313		}314		if self.opts.only.is_empty() {315			return Ok(false);316		}317		let mut have_group_matches = false;318		for item in self.opts.only.iter() {319			match item {320				HostItem::Host { name, .. } if *name == host.name => {321					return Ok(false);322				}323				HostItem::Tag { .. } => {324					have_group_matches = true;325				}326				_ => {}327			}328		}329		if have_group_matches {330			let host_tags = host.tags().await?;331			for item in self.opts.only.iter() {332				match item {333					HostItem::Tag { name, .. } if host_tags.contains(name) => {334						return Ok(false);335					}336					_ => {}337				}338			}339		}340		Ok(true)341	}342	pub async fn action_attr(&self, host: &ConfigHost, attr: &str) -> Result<Option<String>> {343		if self.opts.only.is_empty() {344			return Ok(None);345		}346		let mut have_group_matches = false;347		for item in self.opts.only.iter() {348			match item {349				HostItem::Host { name, attrs }350					if *name == host.name && attrs.contains_key(attr) =>351				{352					return Ok(attrs.get(attr).cloned());353				}354				HostItem::Tag { attrs, .. } if attrs.contains_key(attr) => {355					have_group_matches = true;356				}357				_ => {}358			}359		}360		if have_group_matches {361			let host_tags = host.tags().await?;362			for item in self.opts.only.iter() {363				match item {364					HostItem::Tag { name, attrs }365						if host_tags.contains(name) && attrs.contains_key(attr) =>366					{367						return Ok(attrs.get(attr).cloned());368					}369					_ => {}370				}371			}372		}373		Ok(None)374	}375	pub fn is_local(&self, host: &str) -> bool {376		self.opts.localhost.as_ref().map(|s| s as &str) == Some(host)377	}378379	pub fn local_host(&self) -> ConfigHost {380		ConfigHost {381			config: self.clone(),382			name: "<virtual localhost>".to_owned(),383			local: true,384			session: OnceLock::new(),385			host_config: None,386			nixos_config: OnceCell::new(),387			groups: {388				let cell = OnceCell::new();389				let _ = cell.set(vec![]);390				cell391			},392		}393	}394395	pub async fn host(&self, name: &str) -> Result<ConfigHost> {396		let config = &self.config_field;397		let host_config = nix_go!(config.hosts[{ name }]);398399400		Ok(ConfigHost {401			config: self.clone(),402			name: name.to_owned(),403			local: self.is_local(name),404			session: OnceLock::new(),405			host_config: Some(host_config),406			nixos_config: OnceCell::new(),407			groups: OnceCell::new(),408		})409	}410	pub async fn list_hosts(&self) -> Result<Vec<ConfigHost>> {411		let config = &self.config_field;412		let names = nix_go!(config.hosts).list_fields().await?;413		let mut out = vec![];414		for name in names {415			out.push(self.host(&name).await?);416		}417		Ok(out)418	}419	pub async fn system_config(&self, host: &str) -> Result<Value> {420		let fleet_field = &self.config_field;421		Ok(nix_go!(fleet_field.hosts[{ host }].nixos.config))422	}423424	pub(super) fn data(&self) -> MutexGuard<FleetData> {425		self.data.lock().unwrap()426	}427	pub(super) fn data_mut(&self) -> MutexGuard<FleetData> {428		self.data.lock().unwrap()429	}430	/// Shared secrets configured in fleet.nix or in flake431	pub async fn list_configured_shared(&self) -> Result<Vec<String>> {432		let config_field = &self.config_field;433		Ok(nix_go!(config_field.sharedSecrets).list_fields().await?)434	}435	/// Shared secrets configured in fleet.nix436	pub fn list_shared(&self) -> Vec<String> {437		let data = self.data();438		data.shared_secrets.keys().cloned().collect()439	}440	pub fn has_shared(&self, name: &str) -> bool {441		let data = self.data();442		data.shared_secrets.contains_key(name)443	}444	pub fn replace_shared(&self, name: String, shared: FleetSharedSecret) {445		let mut data = self.data_mut();446		data.shared_secrets.insert(name.to_owned(), shared);447	}448	pub fn remove_shared(&self, secret: &str) {449		let mut data = self.data_mut();450		data.shared_secrets.remove(secret);451	}452453	pub fn list_secrets(&self, host: &str) -> Vec<String> {454		let data = self.data();455		let Some(secrets) = data.host_secrets.get(host) else {456			return Vec::new();457		};458		secrets.keys().cloned().collect()459	}460461	pub fn has_secret(&self, host: &str, secret: &str) -> bool {462		let data = self.data();463		let Some(host_secrets) = data.host_secrets.get(host) else {464			return false;465		};466		host_secrets.contains_key(secret)467	}468	pub fn insert_secret(&self, host: &str, secret: String, value: FleetSecret) {469		let mut data = self.data_mut();470		let host_secrets = data.host_secrets.entry(host.to_owned()).or_default();471		host_secrets.insert(secret, value);472	}473474	pub fn host_secret(&self, host: &str, secret: &str) -> Result<FleetSecret> {475		let data = self.data();476		let Some(host_secrets) = data.host_secrets.get(host) else {477			bail!("no secrets for machine {host}");478		};479		let Some(secret) = host_secrets.get(secret) else {480			bail!("machine {host} has no secret {secret}");481		};482		Ok(secret.clone())483	}484	pub fn shared_secret(&self, secret: &str) -> Result<FleetSharedSecret> {485		let data = self.data();486		let Some(secret) = data.shared_secrets.get(secret) else {487			bail!("no shared secret {secret}");488		};489		Ok(secret.clone())490	}491	pub async fn shared_secret_expected_owners(&self, secret: &str) -> Result<Vec<String>> {492		let config_field = &self.config_field;493		Ok(nix_go_json!(494			config_field.sharedSecrets[{ secret }].expectedOwners495		))496	}497498	pub fn save(&self) -> Result<()> {499		let mut tempfile = NamedTempFile::new_in(self.directory.clone()).context("failed to create updated version of fleet.nix in the same directory as original.\nDo you have write access to it? Access only to the fleet.nix won't be enough, the directory is used for atomic overwrite operation.\nIt is not recommended to use fleet by root anyway, move fleet project to your home directory.")?;500		let data = nixlike::serialize(&self.data() as &FleetData)?;501		tempfile.write_all(502			format!(503				"# This file contains fleet state and shouldn't be edited by hand\n\n{}\n\n# vim: ts=2 et nowrap\n",504				data505			)506			.as_bytes(),507		)?;508		let mut fleet_data_path = self.directory.clone();509		fleet_data_path.push("fleet.nix");510		tempfile.persist(fleet_data_path)?;511		Ok(())512	}513}514515#[derive(Clone)]516enum HostItem {517	Host {518		name: String,519		attrs: BTreeMap<String, String>,520	},521	Tag {522		name: String,523		attrs: BTreeMap<String, String>,524	},525}526fn host_item_parser(input: &str) -> Result<HostItem, String> {527	fn err_to_string(err: nom::Err<nom::error::Error<&str>>) -> String {528		err.to_string()529	}530531	let (input, is_tag) = map(opt(char('@')), |c| c.is_some())(input).map_err(err_to_string)?;532	let (input, name) = map(533		take_while1(|v| v != ',' && v != '?' && v != '@'),534		str::to_owned,535	)(input)536	.map_err(err_to_string)?;537538	let kw_item = separated_pair(539		map(take_while1(|v| v != '&' && v != '='), str::to_owned),540		char('='),541		map(take_while1(|v| v != '&'), str::to_owned),542	);543	let kw = map(separated_list1(char('&'), kw_item), |vec| {544		vec.into_iter().collect::<BTreeMap<_, _>>()545	});546	let mut opt_kw = map(opt(preceded(char('?'), kw)), Option::unwrap_or_default);547548	let (input, attrs) = opt_kw(input).map_err(err_to_string)?;549550	if !input.is_empty() {551		return Err(format!("unexpected trailing input: {input:?}"));552	}553	Ok(if is_tag {554		HostItem::Tag { name, attrs }555	} else {556		HostItem::Host { name, attrs }557	})558}559560#[derive(Parser, Clone)]561pub struct FleetOpts {562	/// All hosts except those would be skipped563	#[clap(long, number_of_values = 1, value_parser = host_item_parser)]564	only: Vec<HostItem>,565566	/// Hosts to skip567	#[clap(long, number_of_values = 1)]568	skip: Vec<String>,569570	/// Host, which should be threaten as current machine571	#[clap(long)]572	pub localhost: Option<String>,573574	/// Override detected system for host, to perform builds via575	/// binfmt-declared qemu instead of trying to crosscompile576	#[clap(long, default_value = "detect")]577	pub local_system: String,578}579580impl FleetOpts {581	pub async fn build(mut self, nix_args: Vec<OsString>) -> Result<Config> {582		if self.localhost.is_none() {583			self.localhost584				.replace(hostname::get().unwrap().to_str().unwrap().to_owned());585		}586		let directory = current_dir()?;587588		let pool = NixSessionPool::new(directory.as_os_str().to_owned(), nix_args.clone()).await?;589		let root_field = pool.get().await?;590591		let builtins_field = Value::binding(root_field.clone(), "builtins").await?;592		if self.local_system == "detect" {593			self.local_system = nix_go_json!(builtins_field.currentSystem);594		}595		let local_system = self.local_system.clone();596597		let mut fleet_data_path = directory.clone();598		fleet_data_path.push("fleet.nix");599		let bytes = std::fs::read_to_string(fleet_data_path)?;600		let data: Mutex<FleetData> = nixlike::parse_str(&bytes)?;601602		let fleet_root = Value::binding(root_field, "fleetConfigurations").await?;603		let fleet_field = nix_go!(fleet_root.default({ data }));604605		let config_field = nix_go!(fleet_field.config);606607		assert_warn("fleet config evaluation", &config_field).await?;608609		let import = nix_go!(builtins_field.import);610		let overlays = nix_go!(config_field.nixpkgs.overlays);611		let nixpkgs = nix_go!(fleet_field.nixpkgs.buildUsing | import);612613		let default_pkgs = nix_go!(nixpkgs(Obj {614			overlays,615			system: { self.local_system.clone() },616		}));617618		Ok(Config(Arc::new(FleetConfigInternals {619			opts: self,620			directory,621			data,622			local_system,623			nix_args,624			config_field,625			default_pkgs,626		})))627	}628}