git.delta.rocks / jrsonnet / refs/commits / 87c4900cde09

--- a/cmds/fleet/src/command.rs
+++ /dev/null
@@ -1,430 +0,0 @@
-use std::{ffi::OsStr, pin, process::Stdio, sync::Arc, task::Poll};
-
-use anyhow::{anyhow, Result};
-use better_command::{Handler, NixHandler, PlainHandler};
-use futures::StreamExt;
-use itertools::Either;
-use openssh::{OverSsh, OwningCommand, Session};
-use tokio::{io::AsyncRead, process::Command, select};
-use tokio_util::codec::{BytesCodec, FramedRead, LinesCodec};
-use tracing::debug;
-
-use crate::host::EscalationStrategy;
-
-fn escape_bash(input: &str, out: &mut String) {
-	const TO_ESCAPE: &str = "$ !\"#&'()*,;<>?[\\]^`{|}";
-	if input.chars().all(|c| !TO_ESCAPE.contains(c)) {
-		out.push_str(input);
-		return;
-	}
-	out.push('\'');
-	for (i, v) in input.split('\'').enumerate() {
-		if i != 0 {
-			out.push_str("'\"'\"'");
-		}
-		out.push_str(v);
-	}
-	out.push('\'');
-}
-fn ostoutf8(os: impl AsRef<OsStr>) -> String {
-	os.as_ref().to_str().expect("non-utf8 data").to_owned()
-}
-
-#[derive(Clone, Debug)]
-pub struct MyCommand {
-	command: String,
-	args: Vec<String>,
-	env: Vec<(String, String)>,
-	ssh_session: Option<Arc<Session>>,
-	escalation: EscalationStrategy,
-	escalate: bool,
-}
-impl MyCommand {
-	pub fn new_on(
-		escalation: EscalationStrategy,
-		cmd: impl AsRef<OsStr>,
-		session: Arc<Session>,
-	) -> Self {
-		assert!(!cmd.as_ref().is_empty());
-		Self {
-			command: ostoutf8(cmd),
-			args: vec![],
-			env: vec![],
-			ssh_session: Some(session),
-			escalation,
-			escalate: false,
-		}
-	}
-	pub fn new(escalation: EscalationStrategy, cmd: impl AsRef<OsStr>) -> Self {
-		assert!(!cmd.as_ref().is_empty());
-		Self {
-			command: ostoutf8(cmd),
-			args: vec![],
-			env: vec![],
-			ssh_session: None,
-			escalation,
-			escalate: false,
-		}
-	}
-	fn new_here(&self, cmd: impl AsRef<OsStr>) -> Self {
-		if let Some(ssh_session) = self.ssh_session.clone() {
-			Self::new_on(self.escalation, cmd, ssh_session)
-		} else {
-			Self::new(self.escalation, cmd)
-		}
-	}
-
-	fn into_args(self) -> Vec<String> {
-		let mut out = Vec::new();
-		if !self.env.is_empty() {
-			out.push("env".to_owned());
-			for (k, v) in self.env {
-				assert!(!k.contains('='));
-				out.push(format!("{k}={v}"));
-			}
-		}
-		out.push(self.command);
-		out.extend(self.args);
-		out
-	}
-
-	/// Translates environment variables into env command execution.
-	/// Required for ssh, as ssh don't allow to send environment variables (at least by default).
-	///
-	/// FIXME: Insecure, as arguments might be seen by other users on the same machine.
-	/// Figure out some way to transfer environment using stdio?
-	fn translate_env_into_env(self) -> Self {
-		if self.env.is_empty() {
-			return self;
-		}
-		let mut out = self.new_here("env");
-		for (k, v) in self.env {
-			assert!(!k.contains('='));
-			out.arg(format!("{k}={v}"));
-		}
-		out.arg(self.command);
-		out.args(self.args);
-
-		out
-	}
-	fn into_string(self) -> String {
-		let mut out = String::new();
-		if !self.env.is_empty() {
-			out.push_str("env");
-			for (k, v) in self.env {
-				out.push(' ');
-				assert!(!k.contains('='));
-				escape_bash(&k, &mut out);
-				out.push('=');
-				escape_bash(&v, &mut out);
-			}
-		}
-		if !out.is_empty() {
-			out.push(' ');
-		}
-		escape_bash(&self.command, &mut out);
-		for arg in self.args {
-			out.push(' ');
-			escape_bash(&arg, &mut out);
-		}
-		out
-	}
-	fn into_command(self) -> Command {
-		let mut out = Command::new(self.command);
-		out.args(self.args);
-		for (k, v) in self.env {
-			out.env(k, v);
-		}
-		out
-	}
-	fn into_command_new(self) -> Result<Either<Command, openssh::OwningCommand<Arc<Session>>>> {
-		Ok(if let Some(session) = self.ssh_session.clone() {
-			let cmd = self.translate_env_into_env().into_command();
-			Either::Right(
-				cmd.over_ssh(session)
-					.map_err(|e| anyhow!("ssh error: {e}"))?,
-			)
-		} else {
-			let cmd = self.into_command();
-			Either::Left(cmd)
-		})
-	}
-	pub fn arg(&mut self, arg: impl AsRef<OsStr>) -> &mut Self {
-		let arg = arg.as_ref();
-		self.args.push(ostoutf8(arg));
-		self
-	}
-	pub fn eqarg(&mut self, arg: impl AsRef<OsStr>, value: impl AsRef<OsStr>) -> &mut Self {
-		let arg = arg.as_ref();
-		let value = value.as_ref();
-		let arg = ostoutf8(arg);
-		let value = ostoutf8(value);
-		self.arg(format!("{arg}={value}"));
-		self
-	}
-	pub fn comparg(&mut self, arg: impl AsRef<OsStr>, value: impl AsRef<OsStr>) -> &mut Self {
-		self.arg(arg);
-		self.arg(value);
-		self
-	}
-	pub fn env(&mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> &mut Self {
-		self.env
-			.push((name.as_ref().to_owned(), value.as_ref().to_owned()));
-		self
-	}
-	pub fn args<V: AsRef<OsStr>>(&mut self, args: impl IntoIterator<Item = V>) -> &mut Self {
-		for arg in args.into_iter() {
-			let arg = arg.as_ref();
-			self.args.push(ostoutf8(arg));
-		}
-		self
-	}
-	pub fn sudo(mut self) -> Self {
-		self.escalate = true;
-		self
-	}
-	fn wrap_sudo_if_needed(self) -> Self {
-		if !self.escalate {
-			return self;
-		}
-		match self.escalation {
-			EscalationStrategy::Su => {
-				let mut out = self.new_here("su");
-				out.arg("-c").arg(self.into_string());
-				out
-			}
-			EscalationStrategy::Sudo => {
-				let mut out = self.new_here("sudo");
-				out.args(self.into_args());
-				out
-			}
-			EscalationStrategy::Run0 => {
-				// run0 wants interactive authentication by default.
-				let mut run0 = self.new_here("run0");
-				let mut out = self.new_here("script");
-
-				// Red backgrounds messes with fleet formatting
-				run0.arg("--background=");
-				run0.args(self.into_args());
-
-				out.arg("-q");
-				out.arg("/dev/null");
-				out.arg("-c");
-				out.arg(run0.into_string());
-				dbg!(&out);
-				out
-			}
-		}
-	}
-
-	pub async fn run(self) -> Result<()> {
-		let str = self.clone().into_string();
-		let cmd = self.wrap_sudo_if_needed().into_command_new()?;
-		match cmd {
-			Either::Left(cmd) => run_nix_inner(str, cmd, &mut PlainHandler).await?,
-			Either::Right(cmd) => run_nix_inner_ssh(str, cmd, &mut PlainHandler).await?,
-		};
-		Ok(())
-	}
-	pub async fn run_string(self) -> Result<String> {
-		let bytes = self.run_bytes().await?;
-		Ok(String::from_utf8(bytes)?)
-	}
-	pub async fn run_bytes(self) -> Result<Vec<u8>> {
-		let str = self.clone().into_string();
-		let cmd = self.wrap_sudo_if_needed().into_command_new()?;
-		let v = match cmd {
-			Either::Left(cmd) => run_nix_inner_stdout(str, cmd, &mut PlainHandler).await?,
-			Either::Right(cmd) => run_nix_inner_stdout_ssh(str, cmd, &mut PlainHandler).await?,
-		};
-		Ok(v)
-	}
-
-	pub async fn run_nix_string(mut self) -> Result<String> {
-		let str = self.clone().into_string();
-		self.arg("--log-format").arg("internal-json");
-		let mut cmd = self.wrap_sudo_if_needed().into_command();
-		let bytes = run_nix_inner_stdout(str, cmd, &mut NixHandler::default()).await?;
-		Ok(String::from_utf8(bytes)?)
-	}
-	pub async fn run_nix(mut self) -> Result<()> {
-		let str = self.clone().into_string();
-		self.arg("--log-format").arg("internal-json");
-		let mut cmd = self.wrap_sudo_if_needed().into_command();
-		cmd.stdout(Stdio::inherit());
-		run_nix_inner(str, cmd, &mut NixHandler::default()).await
-	}
-}
-
-struct EmptyAsyncRead;
-impl AsyncRead for EmptyAsyncRead {
-	fn poll_read(
-		self: std::pin::Pin<&mut Self>,
-		_cx: &mut std::task::Context<'_>,
-		_buf: &mut tokio::io::ReadBuf<'_>,
-	) -> Poll<std::io::Result<()>> {
-		Poll::Pending
-	}
-}
-
-async fn run_nix_inner_stdout(
-	str: String,
-	cmd: Command,
-	handler: &mut dyn Handler,
-) -> Result<Vec<u8>> {
-	Ok(run_nix_inner_raw(str, cmd, true, handler, None)
-		.await?
-		.expect("has out"))
-}
-async fn run_nix_inner(str: String, cmd: Command, handler: &mut dyn Handler) -> Result<()> {
-	let v = run_nix_inner_raw(str, cmd, false, handler, None).await?;
-	assert!(v.is_none());
-	Ok(())
-}
-async fn run_nix_inner_stdout_ssh(
-	str: String,
-	cmd: OwningCommand<Arc<Session>>,
-	handler: &mut dyn Handler,
-) -> Result<Vec<u8>> {
-	Ok(run_nix_inner_raw_ssh(str, cmd, true, handler, None)
-		.await?
-		.expect("has out"))
-}
-async fn run_nix_inner_ssh(
-	str: String,
-	cmd: OwningCommand<Arc<Session>>,
-	handler: &mut dyn Handler,
-) -> Result<()> {
-	let v = run_nix_inner_raw_ssh(str, cmd, false, handler, None).await?;
-	assert!(v.is_none());
-	Ok(())
-}
-
-async fn run_nix_inner_raw(
-	str: String,
-	mut cmd: Command,
-	want_stdout: bool,
-	err_handler: &mut dyn Handler,
-	mut out_handler: Option<&mut dyn Handler>,
-) -> Result<Option<Vec<u8>>> {
-	cmd.stderr(Stdio::piped());
-	cmd.stdout(Stdio::piped());
-	debug!("running command {str:?} on local");
-	let mut child = cmd.spawn()?;
-	let mut stderr = child.stderr.take().unwrap();
-	let stdout = child.stdout.take().unwrap();
-	let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
-	let mut out: Option<Box<dyn AsyncRead + Unpin>> = Some(Box::new(stdout));
-	let mut ob = want_stdout
-		.then(|| out.take().unwrap())
-		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut ol = (!want_stdout)
-		.then(|| out.take().unwrap())
-		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut ob = FramedRead::new(&mut ob, BytesCodec::new());
-	let mut ol = FramedRead::new(&mut ol, LinesCodec::new());
-
-	// while let Some(line) = read.next().await? {}
-
-	let mut out_buf = if want_stdout { Some(vec![]) } else { None };
-	loop {
-		select! {
-			e = err.next() => {
-				if let Some(e) = e {
-					let e = e?;
-					err_handler.handle_line(&e);
-				}
-			},
-			o = ob.next() => {
-				if let Some(o) = o {
-					out_buf.as_mut().expect("stdout == wants_stdout").extend_from_slice(&o?);
-				}
-			},
-			o = ol.next() => {
-				if let Some(o) = o {
-					let o = o?;
-					if let Some(out) = out_handler.as_mut() {
-						out.handle_line(&o)
-					} else {
-						err_handler.handle_line(&o)
-					}
-					// out_handler.handle_info(&o);
-				}
-			},
-			code = child.wait() => {
-				let code = code?;
-				if !code.success() {
-					anyhow::bail!("command '{str}' failed with status {}", code);
-				}
-				break;
-			}
-		}
-	}
-
-	Ok(out_buf)
-}
-async fn run_nix_inner_raw_ssh(
-	str: String,
-	mut cmd: OwningCommand<Arc<Session>>,
-	want_stdout: bool,
-	err_handler: &mut dyn Handler,
-	mut out_handler: Option<&mut dyn Handler>,
-) -> Result<Option<Vec<u8>>> {
-	debug!("running command {str:?} over ssh");
-	cmd.stderr(openssh::Stdio::piped());
-	cmd.stdout(openssh::Stdio::piped());
-	let mut child = cmd.spawn().await?;
-	let mut stderr = child.stderr().take().unwrap();
-	let stdout = child.stdout().take().unwrap();
-	let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
-	let mut out: Option<Box<dyn AsyncRead + Unpin>> = Some(Box::new(stdout));
-	let mut ob = want_stdout
-		.then(|| out.take().unwrap())
-		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut ol = (!want_stdout)
-		.then(|| out.take().unwrap())
-		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut ob = FramedRead::new(&mut ob, BytesCodec::new());
-	let mut ol = FramedRead::new(&mut ol, LinesCodec::new());
-
-	// while let Some(line) = read.next().await? {}
-
-	let mut out_buf = if want_stdout { Some(vec![]) } else { None };
-
-	let mut wait_future = pin::pin!(child.wait());
-	loop {
-		select! {
-			e = err.next() => {
-				if let Some(e) = e {
-					let e = e?;
-					err_handler.handle_line(&e);
-				}
-			},
-			o = ob.next() => {
-				if let Some(o) = o {
-					out_buf.as_mut().expect("stdout == wants_stdout").extend_from_slice(&o?);
-				}
-			},
-			o = ol.next() => {
-				if let Some(o) = o {
-					let o = o?;
-					if let Some(out) = out_handler.as_mut() {
-						out.handle_line(&o)
-					} else {
-						err_handler.handle_line(&o)
-					}
-					// out_handler.handle_info(&o);
-				}
-			},
-			code = &mut wait_future => {
-				let code = code?;
-				if !code.success() {
-					anyhow::bail!("command '{str}' failed with status {}", code);
-				}
-				break;
-			}
-		}
-	}
-
-	Ok(out_buf)
-}

--- a/cmds/fleet/src/fleetdata.rs
+++ /dev/null
@@ -1,107 +0,0 @@
-use std::{
-	collections::BTreeMap,
-	io::{self, Cursor},
-};
-
-use age::Recipient;
-use chrono::{DateTime, Utc};
-use fleet_shared::SecretData;
-use itertools::Itertools;
-use serde::{de::Error, Deserialize, Serialize};
-
-#[derive(Serialize, Deserialize, Default)]
-#[serde(rename_all = "camelCase")]
-pub struct HostData {
-	#[serde(default)]
-	#[serde(skip_serializing_if = "String::is_empty")]
-	pub encryption_key: String,
-}
-
-const VERSION: &str = "0.1.0";
-pub struct FleetDataVersion;
-impl Serialize for FleetDataVersion {
-	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-	where
-		S: serde::Serializer,
-	{
-		VERSION.serialize(serializer)
-	}
-}
-impl<'de> Deserialize<'de> for FleetDataVersion {
-	fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-	where
-		D: serde::Deserializer<'de>,
-	{
-		let version = String::deserialize(deserializer)?;
-		if version != VERSION {
-			return Err(D::Error::custom(format!(
-				"fleet.nix data version mismatch, expected {VERSION}, got {version}.\nFollow the docs for migration instruction"
-			)));
-		}
-		Ok(Self)
-	}
-}
-
-#[derive(Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct FleetData {
-	pub version: FleetDataVersion,
-
-	#[serde(default)]
-	pub hosts: BTreeMap<String, HostData>,
-	#[serde(default)]
-	#[serde(skip_serializing_if = "BTreeMap::is_empty")]
-	pub shared_secrets: BTreeMap<String, FleetSharedSecret>,
-	#[serde(default)]
-	#[serde(skip_serializing_if = "BTreeMap::is_empty")]
-	pub host_secrets: BTreeMap<String, BTreeMap<String, FleetSecret>>,
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-#[serde(rename_all = "camelCase")]
-#[must_use]
-pub struct FleetSharedSecret {
-	pub owners: Vec<String>,
-	#[serde(flatten)]
-	pub secret: FleetSecret,
-}
-
-/// Returns None if recipients.is_empty()
-pub fn encrypt_secret_data(
-	recipients: impl IntoIterator<Item = impl Recipient + Send + 'static>,
-	data: Vec<u8>,
-) -> Option<SecretData> {
-	let mut encrypted = vec![];
-	let recipients = recipients
-		.into_iter()
-		.map(|v| Box::new(v) as Box<dyn Recipient + Send>)
-		.collect_vec();
-	let mut encryptor = age::Encryptor::with_recipients(recipients)?
-		.wrap_output(&mut encrypted)
-		.expect("in memory write");
-	io::copy(&mut Cursor::new(data), &mut encryptor).expect("in memory copy");
-	encryptor.finish().expect("in memory flush");
-	Some(SecretData {
-		data: encrypted,
-		encrypted: true,
-	})
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct FleetSecretPart {
-	pub raw: SecretData,
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-#[serde(rename_all = "camelCase")]
-#[must_use]
-pub struct FleetSecret {
-	#[serde(default = "Utc::now")]
-	pub created_at: DateTime<Utc>,
-	#[serde(default)]
-	#[serde(skip_serializing_if = "Option::is_none", alias = "expire_at")]
-	pub expires_at: Option<DateTime<Utc>>,
-
-	#[serde(flatten)]
-	pub parts: BTreeMap<String, FleetSecretPart>,
-}

--- a/cmds/fleet/src/host.rs
+++ /dev/null
@@ -1,630 +0,0 @@
-use std::{
-	cell::{LazyCell, OnceCell},
-	collections::BTreeMap,
-	env::current_dir,
-	ffi::{OsStr, OsString},
-	fmt::Display,
-	io::Write,
-	ops::Deref,
-	path::PathBuf,
-	str::FromStr,
-	sync::{Arc, Mutex, MutexGuard, OnceLock},
-};
-
-use anyhow::{anyhow, bail, ensure, Context, Result};
-use clap::Parser;
-use fleet_shared::SecretData;
-use nix_eval::{nix_go, nix_go_json, util::assert_warn, NixSessionPool, Value};
-use nom::{
-	bytes::complete::take_while1,
-	character::complete::char,
-	combinator::{map, opt},
-	multi::separated_list1,
-	sequence::{preceded, separated_pair},
-};
-use openssh::SessionBuilder;
-use serde::de::DeserializeOwned;
-use tempfile::NamedTempFile;
-use tracing::error;
-
-use crate::{
-	command::MyCommand,
-	fleetdata::{FleetData, FleetSecret, FleetSharedSecret},
-};
-
-pub struct FleetConfigInternals {
-	pub local_system: String,
-	pub directory: PathBuf,
-	pub opts: FleetOpts,
-	pub data: Mutex<FleetData>,
-	pub nix_args: Vec<OsString>,
-	/// fleet_config.config
-	pub config_field: Value,
-
-	/// import nixpkgs {system = local};
-	pub default_pkgs: Value,
-}
-
-#[derive(Clone)]
-pub struct Config(Arc<FleetConfigInternals>);
-
-impl Deref for Config {
-	type Target = FleetConfigInternals;
-
-	fn deref(&self) -> &Self::Target {
-		&self.0
-	}
-}
-
-#[derive(Clone, Copy, Debug)]
-pub enum EscalationStrategy {
-	Sudo,
-	Run0,
-	Su,
-}
-
-pub struct ConfigHost {
-	config: Config,
-	pub name: String,
-	pub local: bool,
-	pub session: OnceLock<Arc<openssh::Session>>,
-	groups: OnceCell<Vec<String>>,
-
-	pub host_config: Option<Value>,
-	pub nixos_config: OnceCell<Value>,
-}
-impl ConfigHost {
-	pub async fn escalation_strategy(&self) -> Result<EscalationStrategy> {
-		// Prefer sudo, as run0 has some gotchas with polkit
-		// and too many repeating prompts.
-		if let Ok(_) = self.find_in_path("sudo").await {
-			return Ok(EscalationStrategy::Sudo);
-		}
-		if let Ok(_) = self.find_in_path("run0").await {
-			return Ok(EscalationStrategy::Run0);
-		}
-		Ok(EscalationStrategy::Su)
-	}
-	// TOCTOU is possible here in case if config is changed, but this case is not handled anywhere anyway,
-	// assuming getting tags always returns the same value.
-	pub async fn tags(&self) -> Result<Vec<String>> {
-		if let Some(v) = self.groups.get() {
-			return Ok(v.clone());
-		}
-		let Some(host_config) = &self.host_config else {
-			return Ok(vec![]);
-		};
-		let tags: Vec<String> = nix_go_json!(host_config.tags);
-
-		let _ = self.groups.set(tags.clone());
-
-		Ok(tags)
-	}
-	pub async fn nixos_config(&self) -> Result<Value> {
-		if let Some(v) = self.nixos_config.get() {
-			return Ok(v.clone());
-		}
-		let Some(host_config) = &self.host_config else {
-			bail!("local host has no nixos_config");
-		};
-		let nixos_config = nix_go!(host_config.nixos.config);
-		assert_warn("nixos config evaluation", &nixos_config).await?;
-
-		let _ = self.nixos_config.set(nixos_config.clone());
-
-		Ok(nixos_config)
-	}
-	async fn open_session(&self) -> Result<Arc<openssh::Session>> {
-		assert!(!self.local, "do not open ssh connection to local session");
-		// FIXME: TOCTOU
-		if let Some(session) = &self.session.get() {
-			return Ok((*session).clone());
-		};
-		let mut session = SessionBuilder::default();
-		let session = session
-			.connect(&self.name)
-			.await
-			.map_err(|e| anyhow!("ssh error while connecting to {}: {e}", self.name))?;
-		let session = Arc::new(session);
-		self.session.set(session.clone()).expect("TOCTOU happened");
-		Ok(session)
-	}
-	pub async fn mktemp_dir(&self) -> Result<String> {
-		let mut cmd = self.cmd("mktemp").await?;
-		cmd.arg("-d");
-		let path = cmd.run_string().await?;
-		Ok(path.trim_end().to_owned())
-	}
-	pub async fn read_file_bin(&self, path: impl AsRef<OsStr>) -> Result<Vec<u8>> {
-		let mut cmd = self.cmd("cat").await?;
-		cmd.arg(path);
-		cmd.run_bytes().await
-	}
-	pub async fn read_file_text(&self, path: impl AsRef<OsStr>) -> Result<String> {
-		let mut cmd = self.cmd("cat").await?;
-		cmd.arg(path);
-		cmd.run_string().await
-	}
-	pub async fn read_dir(&self, path: impl AsRef<OsStr>) -> Result<Vec<String>> {
-		let mut cmd = self.cmd("ls").await?;
-		cmd.arg(path);
-		let out = cmd.run_string().await?;
-		let mut lines = out.split('\n');
-		if let Some(last) = lines.next_back() {
-			ensure!(last.is_empty(), "output of ls should end with newline");
-		}
-		Ok(lines.map(ToOwned::to_owned).collect())
-	}
-	#[allow(dead_code)]
-	pub async fn read_file_json<D: DeserializeOwned>(&self, path: impl AsRef<OsStr>) -> Result<D> {
-		let text = self.read_file_text(path).await?;
-		Ok(serde_json::from_str(&text)?)
-	}
-	pub async fn read_env(&self, env: &str) -> Result<String> {
-		let mut cmd = self.cmd("printenv").await?;
-		cmd.arg(env);
-		Ok(cmd.run_string().await?)
-	}
-	pub async fn find_in_path(&self, command: &str) -> Result<String> {
-		// // `which` is not a part of coreutils, and it might not exist on machine.
-		// let path = self.read_env("PATH").await?;
-		// // Assuming delimiter is :, we don't work with windows host, this check will be much
-		// // more sophisticated in remowt backend (and quicker, since actual PATH search will be done on remote machine)
-		// for ele in path.split(':') {
-		// 	let test_path = format!("{ele}/{cmd}");
-		// 	test -x etc
-		// }
-		// let mut cmd = self.cmd("printenv").await?;
-		// cmd.arg(env);
-		// Ok(cmd.run_string().await?)
-		// Assuming this is an environment issue if which doesn't exist, will be fixed with remowt.
-		let mut cmd = self
-			.cmd_escalation(
-				// Not used
-				EscalationStrategy::Su,
-				"which",
-			)
-			.await?;
-		cmd.arg(command);
-		cmd.run_string().await
-	}
-	pub async fn read_file_value<D: FromStr>(&self, path: impl AsRef<OsStr>) -> Result<D>
-	where
-		<D as FromStr>::Err: Display,
-	{
-		let text = self.read_file_text(path).await?;
-		D::from_str(&text).map_err(|e| anyhow!("failed to parse value: {e}"))
-	}
-	pub async fn cmd(&self, cmd: impl AsRef<OsStr>) -> Result<MyCommand> {
-		self.cmd_escalation(self.escalation_strategy().await?, cmd)
-			.await
-	}
-	pub async fn cmd_escalation(
-		&self,
-		escalation: EscalationStrategy,
-		cmd: impl AsRef<OsStr>,
-	) -> Result<MyCommand> {
-		if self.local {
-			Ok(MyCommand::new(escalation, cmd))
-		} else {
-			let session = self.open_session().await?;
-			Ok(MyCommand::new_on(escalation, cmd, session))
-		}
-	}
-
-	pub async fn decrypt(&self, data: SecretData) -> Result<Vec<u8>> {
-		ensure!(data.encrypted, "secret is not encrypted");
-		let mut cmd = self.cmd("fleet-install-secrets").await?;
-		cmd.arg("decrypt").eqarg("--secret", data.to_string());
-		let encoded = cmd
-			.sudo()
-			.run_string()
-			.await
-			.context("failed to call remote host for decrypt")?;
-		let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;
-		ensure!(!data.encrypted, "secret came out encrypted");
-		Ok(data.data)
-	}
-	pub async fn reencrypt(&self, data: SecretData, targets: Vec<String>) -> Result<SecretData> {
-		ensure!(data.encrypted, "secret is not encrypted");
-		let mut cmd = self.cmd("fleet-install-secrets").await?;
-		cmd.arg("reencrypt").eqarg("--secret", data.to_string());
-		for target in targets {
-			let key = self.config.key(&target).await?;
-			cmd.eqarg("--targets", key);
-		}
-		let encoded = cmd
-			.sudo()
-			.run_string()
-			.await
-			.context("failed to call remote host for decrypt")?;
-		let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;
-		ensure!(data.encrypted, "secret came out not encrypted");
-		Ok(data)
-	}
-	/// Returns path for futureproofing, as path might change i.e on conversion to CA
-	pub async fn remote_derivation(&self, path: &PathBuf) -> Result<PathBuf> {
-		if self.local {
-			// Path is located locally, thus already trusted.
-			return Ok(path.to_owned());
-		}
-		let mut nix = MyCommand::new(
-			// Not used
-			EscalationStrategy::Su,
-			"nix",
-		);
-		nix.arg("copy")
-			.arg("--substitute-on-destination")
-			.comparg("--to", format!("ssh-ng://{}", self.name))
-			.arg(path);
-		nix.run_nix().await.context("nix copy")?;
-		Ok(path.to_owned())
-	}
-	pub async fn systemctl_stop(&self, name: &str) -> Result<()> {
-		let mut cmd = self.cmd("systemctl").await?;
-		cmd.arg("stop").arg(name);
-		cmd.sudo().run().await
-	}
-	pub async fn systemctl_start(&self, name: &str) -> Result<()> {
-		let mut cmd = self.cmd("systemctl").await?;
-		cmd.arg("start").arg(name);
-		cmd.sudo().run().await
-	}
-
-	pub async fn rm_file(&self, path: impl AsRef<OsStr>, sudo: bool) -> Result<()> {
-		let mut cmd = self.cmd("rm").await?;
-		cmd.arg("-f").arg(path);
-		if sudo {
-			cmd = cmd.sudo()
-		}
-		cmd.run().await
-	}
-
-	pub async fn list_configured_secrets(&self) -> Result<Vec<String>> {
-		let nixos = self.nixos_config().await?;
-		let secrets = nix_go!(nixos.secrets);
-		let mut out = Vec::new();
-		for name in secrets.list_fields().await? {
-			let secret = nix_go!(secrets[{ name }]);
-			let is_shared: bool = nix_go_json!(secret.shared);
-			if is_shared {
-				continue;
-			}
-			out.push(name);
-		}
-		Ok(out)
-	}
-	pub async fn secret_field(&self, name: &str) -> Result<Value> {
-		let nixos = self.nixos_config().await?;
-		Ok(nix_go!(nixos.secrets[{ name }]))
-	}
-
-	/// Packages for this host, resolved with nixpkgs overlays
-	pub async fn pkgs(&self) -> Result<Value> {
-		let Some(host_config) = &self.host_config else {
-			bail!("local host has no host_config");
-		};
-		// TODO: Should nixos.options be cached?
-		Ok(nix_go!(host_config.nixos.options._module.args.value.pkgs))
-	}
-}
-
-impl Config {
-	pub async fn should_skip(&self, host: &ConfigHost) -> Result<bool> {
-		if !self.opts.skip.is_empty() && self.opts.skip.iter().any(|h| h as &str == host.name) {
-			return Ok(true);
-		}
-		if self.opts.only.is_empty() {
-			return Ok(false);
-		}
-		let mut have_group_matches = false;
-		for item in self.opts.only.iter() {
-			match item {
-				HostItem::Host { name, .. } if *name == host.name => {
-					return Ok(false);
-				}
-				HostItem::Tag { .. } => {
-					have_group_matches = true;
-				}
-				_ => {}
-			}
-		}
-		if have_group_matches {
-			let host_tags = host.tags().await?;
-			for item in self.opts.only.iter() {
-				match item {
-					HostItem::Tag { name, .. } if host_tags.contains(name) => {
-						return Ok(false);
-					}
-					_ => {}
-				}
-			}
-		}
-		Ok(true)
-	}
-	pub async fn action_attr(&self, host: &ConfigHost, attr: &str) -> Result<Option<String>> {
-		if self.opts.only.is_empty() {
-			return Ok(None);
-		}
-		let mut have_group_matches = false;
-		for item in self.opts.only.iter() {
-			match item {
-				HostItem::Host { name, attrs }
-					if *name == host.name && attrs.contains_key(attr) =>
-				{
-					return Ok(attrs.get(attr).cloned());
-				}
-				HostItem::Tag { attrs, .. } if attrs.contains_key(attr) => {
-					have_group_matches = true;
-				}
-				_ => {}
-			}
-		}
-		if have_group_matches {
-			let host_tags = host.tags().await?;
-			for item in self.opts.only.iter() {
-				match item {
-					HostItem::Tag { name, attrs }
-						if host_tags.contains(name) && attrs.contains_key(attr) =>
-					{
-						return Ok(attrs.get(attr).cloned());
-					}
-					_ => {}
-				}
-			}
-		}
-		Ok(None)
-	}
-	pub fn is_local(&self, host: &str) -> bool {
-		self.opts.localhost.as_ref().map(|s| s as &str) == Some(host)
-	}
-
-	pub fn local_host(&self) -> ConfigHost {
-		ConfigHost {
-			config: self.clone(),
-			name: "<virtual localhost>".to_owned(),
-			local: true,
-			session: OnceLock::new(),
-			host_config: None,
-			nixos_config: OnceCell::new(),
-			groups: {
-				let cell = OnceCell::new();
-				let _ = cell.set(vec![]);
-				cell
-			},
-		}
-	}
-
-	pub async fn host(&self, name: &str) -> Result<ConfigHost> {
-		let config = &self.config_field;
-		let host_config = nix_go!(config.hosts[{ name }]);
-
-		Ok(ConfigHost {
-			config: self.clone(),
-			name: name.to_owned(),
-			local: self.is_local(name),
-			session: OnceLock::new(),
-			host_config: Some(host_config),
-			nixos_config: OnceCell::new(),
-			groups: OnceCell::new(),
-		})
-	}
-	pub async fn list_hosts(&self) -> Result<Vec<ConfigHost>> {
-		let config = &self.config_field;
-		let names = nix_go!(config.hosts).list_fields().await?;
-		let mut out = vec![];
-		for name in names {
-			out.push(self.host(&name).await?);
-		}
-		Ok(out)
-	}
-	pub async fn system_config(&self, host: &str) -> Result<Value> {
-		let fleet_field = &self.config_field;
-		Ok(nix_go!(fleet_field.hosts[{ host }].nixos.config))
-	}
-
-	pub(super) fn data(&self) -> MutexGuard<FleetData> {
-		self.data.lock().unwrap()
-	}
-	pub(super) fn data_mut(&self) -> MutexGuard<FleetData> {
-		self.data.lock().unwrap()
-	}
-	/// Shared secrets configured in fleet.nix or in flake
-	pub async fn list_configured_shared(&self) -> Result<Vec<String>> {
-		let config_field = &self.config_field;
-		Ok(nix_go!(config_field.sharedSecrets).list_fields().await?)
-	}
-	/// Shared secrets configured in fleet.nix
-	pub fn list_shared(&self) -> Vec<String> {
-		let data = self.data();
-		data.shared_secrets.keys().cloned().collect()
-	}
-	pub fn has_shared(&self, name: &str) -> bool {
-		let data = self.data();
-		data.shared_secrets.contains_key(name)
-	}
-	pub fn replace_shared(&self, name: String, shared: FleetSharedSecret) {
-		let mut data = self.data_mut();
-		data.shared_secrets.insert(name.to_owned(), shared);
-	}
-	pub fn remove_shared(&self, secret: &str) {
-		let mut data = self.data_mut();
-		data.shared_secrets.remove(secret);
-	}
-
-	pub fn list_secrets(&self, host: &str) -> Vec<String> {
-		let data = self.data();
-		let Some(secrets) = data.host_secrets.get(host) else {
-			return Vec::new();
-		};
-		secrets.keys().cloned().collect()
-	}
-
-	pub fn has_secret(&self, host: &str, secret: &str) -> bool {
-		let data = self.data();
-		let Some(host_secrets) = data.host_secrets.get(host) else {
-			return false;
-		};
-		host_secrets.contains_key(secret)
-	}
-	pub fn insert_secret(&self, host: &str, secret: String, value: FleetSecret) {
-		let mut data = self.data_mut();
-		let host_secrets = data.host_secrets.entry(host.to_owned()).or_default();
-		host_secrets.insert(secret, value);
-	}
-
-	pub fn host_secret(&self, host: &str, secret: &str) -> Result<FleetSecret> {
-		let data = self.data();
-		let Some(host_secrets) = data.host_secrets.get(host) else {
-			bail!("no secrets for machine {host}");
-		};
-		let Some(secret) = host_secrets.get(secret) else {
-			bail!("machine {host} has no secret {secret}");
-		};
-		Ok(secret.clone())
-	}
-	pub fn shared_secret(&self, secret: &str) -> Result<FleetSharedSecret> {
-		let data = self.data();
-		let Some(secret) = data.shared_secrets.get(secret) else {
-			bail!("no shared secret {secret}");
-		};
-		Ok(secret.clone())
-	}
-	pub async fn shared_secret_expected_owners(&self, secret: &str) -> Result<Vec<String>> {
-		let config_field = &self.config_field;
-		Ok(nix_go_json!(
-			config_field.sharedSecrets[{ secret }].expectedOwners
-		))
-	}
-
-	pub fn save(&self) -> Result<()> {
-		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.")?;
-		let data = nixlike::serialize(&self.data() as &FleetData)?;
-		tempfile.write_all(
-			format!(
-				"# This file contains fleet state and shouldn't be edited by hand\n\n{}\n\n# vim: ts=2 et nowrap\n",
-				data
-			)
-			.as_bytes(),
-		)?;
-		let mut fleet_data_path = self.directory.clone();
-		fleet_data_path.push("fleet.nix");
-		tempfile.persist(fleet_data_path)?;
-		Ok(())
-	}
-}
-
-#[derive(Clone)]
-enum HostItem {
-	Host {
-		name: String,
-		attrs: BTreeMap<String, String>,
-	},
-	Tag {
-		name: String,
-		attrs: BTreeMap<String, String>,
-	},
-}
-fn host_item_parser(input: &str) -> Result<HostItem, String> {
-	fn err_to_string(err: nom::Err<nom::error::Error<&str>>) -> String {
-		err.to_string()
-	}
-
-	let (input, is_tag) = map(opt(char('@')), |c| c.is_some())(input).map_err(err_to_string)?;
-	let (input, name) = map(
-		take_while1(|v| v != ',' && v != '?' && v != '@'),
-		str::to_owned,
-	)(input)
-	.map_err(err_to_string)?;
-
-	let kw_item = separated_pair(
-		map(take_while1(|v| v != '&' && v != '='), str::to_owned),
-		char('='),
-		map(take_while1(|v| v != '&'), str::to_owned),
-	);
-	let kw = map(separated_list1(char('&'), kw_item), |vec| {
-		vec.into_iter().collect::<BTreeMap<_, _>>()
-	});
-	let mut opt_kw = map(opt(preceded(char('?'), kw)), Option::unwrap_or_default);
-
-	let (input, attrs) = opt_kw(input).map_err(err_to_string)?;
-
-	if !input.is_empty() {
-		return Err(format!("unexpected trailing input: {input:?}"));
-	}
-	Ok(if is_tag {
-		HostItem::Tag { name, attrs }
-	} else {
-		HostItem::Host { name, attrs }
-	})
-}
-
-#[derive(Parser, Clone)]
-pub struct FleetOpts {
-	/// All hosts except those would be skipped
-	#[clap(long, number_of_values = 1, value_parser = host_item_parser)]
-	only: Vec<HostItem>,
-
-	/// Hosts to skip
-	#[clap(long, number_of_values = 1)]
-	skip: Vec<String>,
-
-	/// Host, which should be threaten as current machine
-	#[clap(long)]
-	pub localhost: Option<String>,
-
-	/// Override detected system for host, to perform builds via
-	/// binfmt-declared qemu instead of trying to crosscompile
-	#[clap(long, default_value = "detect")]
-	pub local_system: String,
-}
-
-impl FleetOpts {
-	pub async fn build(mut self, nix_args: Vec<OsString>) -> Result<Config> {
-		if self.localhost.is_none() {
-			self.localhost
-				.replace(hostname::get().unwrap().to_str().unwrap().to_owned());
-		}
-		let directory = current_dir()?;
-
-		let pool = NixSessionPool::new(directory.as_os_str().to_owned(), nix_args.clone()).await?;
-		let root_field = pool.get().await?;
-
-		let builtins_field = Value::binding(root_field.clone(), "builtins").await?;
-		if self.local_system == "detect" {
-			self.local_system = nix_go_json!(builtins_field.currentSystem);
-		}
-		let local_system = self.local_system.clone();
-
-		let mut fleet_data_path = directory.clone();
-		fleet_data_path.push("fleet.nix");
-		let bytes = std::fs::read_to_string(fleet_data_path)?;
-		let data: Mutex<FleetData> = nixlike::parse_str(&bytes)?;
-
-		let fleet_root = Value::binding(root_field, "fleetConfigurations").await?;
-		let fleet_field = nix_go!(fleet_root.default({ data }));
-
-		let config_field = nix_go!(fleet_field.config);
-
-		assert_warn("fleet config evaluation", &config_field).await?;
-
-		let import = nix_go!(builtins_field.import);
-		let overlays = nix_go!(config_field.nixpkgs.overlays);
-		let nixpkgs = nix_go!(fleet_field.nixpkgs.buildUsing | import);
-
-		let default_pkgs = nix_go!(nixpkgs(Obj {
-			overlays,
-			system: { self.local_system.clone() },
-		}));
-
-		Ok(Config(Arc::new(FleetConfigInternals {
-			opts: self,
-			directory,
-			data,
-			local_system,
-			nix_args,
-			config_field,
-			default_pkgs,
-		})))
-	}
-}