git.delta.rocks / fleet / refs/commits / a369041a95eb

--- a/cmds/fleet/src/better_nix_eval.rs
+++ b/cmds/fleet/src/better_nix_eval.rs
@@ -472,7 +472,7 @@
 	($field:ident $($tt:tt)*) => {{
 		use $crate::{better_nix_eval::NixExprBuilder, nix_expr_inner};
 		#[allow(unused_mut, reason = "might be used if indexed")]
-		let mut out = NixExprBuilder::field($field);
+		let mut out = NixExprBuilder::field($field.clone());
 		nix_expr_inner!(@field(out) $($tt)*);
 		out
 	}};

--- a/cmds/fleet/src/cmds/build_systems.rs
+++ b/cmds/fleet/src/cmds/build_systems.rs
@@ -291,9 +291,11 @@
 		info!("building");
 		let action = Action::from(self.subcommand.clone());
 		let fleet_field = &config.fleet_field;
-		let drv = nix_go!(fleet_field.buildSystems(Obj {
-			localSystem: { config.local_system.clone() }
-		}));
+		let drv = nix_go!(
+			fleet_field.buildSystems(Obj {
+				localSystem: { config.local_system.clone() }
+			})[{ action.build_attr() }][{ host }]
+		);
 		let outputs = drv.build().await.map_err(|e| {
 			if action.build_attr() == "sdImage" {
 				info!("sd-image build failed");

1use crate::{1use crate::{
2	command::MyCommand,
2	fleetdata::{FleetSecret, FleetSharedSecret},3	fleetdata::{FleetSecret, FleetSharedSecret},
3	host::Config,4	host::Config,
4	nix_go, nix_go_json,5	nix_go, nix_go_json,
12	collections::HashSet,13	collections::HashSet,
13	io::{self, Cursor, Read},14	io::{self, Cursor, Read},
14	path::PathBuf,15	path::PathBuf,
16	sync::Arc,
15};17};
16use tabled::{Table, Tabled};18use tabled::{Table, Tabled};
17use tokio::fs::read_to_string;19use tokio::fs::read_to_string;
97			Secret::InvokeGenerator => {99			Secret::InvokeGenerator => {
98				let config_field = &config.config_unchecked_field;100				let config_field = &config.config_unchecked_field;
99101
100				let generate_impure =102				let secret =
101					nix_go!(config_field.sharedSecrets["kube-apiserver.pem"].generateImpure);103					nix_go!(config_field.configUnchecked.sharedSecrets["kube-apiserver.pem"]);
104				let generate_impure = nix_go!(secret.generateImpure);
102				let on = nix_go!(generate_impure.on);105				let on = nix_go!(generate_impure.on);
103				let call_package = nix_go!(106				let call_package = nix_go!(
104					config_field.buildableSystems(Obj {107					config_field.buildableSystems(Obj {
105						localSystem: { config.local_system.clone() }108						localSystem: { config.local_system.clone() }
106					})[on]109					})[on]
107						.config110						.config
108						.nixpkgs111						.nixpkgs
109						.pkgs112						.resolvedPkgs
110						.callPackage113						.callPackage
111				);114				);
112				let generator = nix_go!(call_package(generate_impure.generator));115				let generator = nix_go!(call_package(generate_impure.generator)(Obj {}));
113				let built = generator.build().await?;116				let built = &generator.build().await?["out"];
114				// .as_json().await?;117				let mut nix = MyCommand::new("nix");
118				let on: String = on.as_json().await?;
119				nix.arg("copy")
120					.arg("--substitute-on-destination")
121					.comparg("--to", format!("ssh-ng://{on}"))
122					.arg(built);
123				nix.run_nix().await?;
124
125				let session = config.host(&on).await?;
126
127				let owners: Vec<String> = nix_go_json!(secret.expectedOwners);
128				dbg!(&owners);
129
130				let mut recipients = String::new();
131				for owner in owners {
132					let key = config.key(&owner).await?;
133					recipients.push_str(&format!("-r \"{key}\" "));
134				}
135				recipients.push_str("-e");
136
137				// FIXME: security: created directory might be accessible to other users
138				// This shouldn't be much of a concern, as data is encrypted right after creation, yet
139				// still better to have.
140				let tempdir = session.mktemp_dir().await?;
141
142				let mut gen = session.cmd(built).await?;
143				gen.env("rageArgs", recipients).env("out", &tempdir);
144				gen.run().await?;
145
146				{
147					let marker = session.read_file_text(format!("{tempdir}/marker")).await?;
148					ensure!(marker == "SUCCESS", "generation not succeeded");
149				}
150
151				let public = session
152					.read_file_bin(format!("{tempdir}/public"))
153					.await
154					.ok();
155				let secret = session
156					.read_file_bin(format!("{tempdir}/secret"))
157					.await
158					.ok();
159				if let Some(secret) = &secret {
160					ensure!(
161						age::Decryptor::new(Cursor::new(&secret)).is_ok(),
162						"builder produced non-encrypted value as secret, this is highly insecure"
163					);
164				}
115				dbg!(&built);165				dbg!(&secret);
166				// // .as_json().await?;
167				// dbg!(&built);
116			}168			}
117			Secret::ForceKeys => {169			Secret::ForceKeys => {
118				for host in config.list_hosts().await? {170				for host in config.list_hosts().await? {
249				if secret.secret.is_empty() {301				if secret.secret.is_empty() {
250					bail!("no secret {name}");302					bail!("no secret {name}");
251				}303				}
304				let host = config.host(&machine).await?;
252				let data = config.decrypt_on_host(&machine, secret.secret).await?;305				let data = host.decrypt(secret.secret).await?;
253				if plaintext {306				if plaintext {
254					let s = String::from_utf8(data).context("output is not utf8")?;307					let s = String::from_utf8(data).context("output is not utf8")?;
255					print!("{s}");308					print!("{s}");

--- a/cmds/fleet/src/command.rs
+++ b/cmds/fleet/src/command.rs
@@ -1,6 +1,7 @@
 use std::{
 	collections::HashMap,
 	ffi::OsStr,
+	pin,
 	process::Stdio,
 	sync::{Arc, Mutex},
 	task::Poll,
@@ -10,7 +11,7 @@
 use futures::StreamExt;
 use itertools::Either;
 use once_cell::sync::Lazy;
-use openssh::{OverSsh, Session};
+use openssh::{OverSsh, OwningCommand, Session};
 use regex::Regex;
 use serde::{de::Visitor, Deserialize};
 use tokio::{io::AsyncRead, process::Command, select};
@@ -44,6 +45,15 @@
 	ssh_session: Option<Arc<Session>>,
 }
 impl MyCommand {
+	pub fn new_on(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),
+		}
+	}
 	pub fn new(cmd: impl AsRef<OsStr>) -> Self {
 		assert!(!cmd.as_ref().is_empty());
 		Self {
@@ -66,6 +76,29 @@
 		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("env");
+		if let Some(session) = self.ssh_session {
+			out = out.ssh_session(session);
+		}
+		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() {
@@ -98,7 +131,7 @@
 	}
 	fn into_command_new(self) -> Result<Either<Command, openssh::OwningCommand<Arc<Session>>>> {
 		Ok(if let Some(session) = self.ssh_session.clone() {
-			let cmd = self.into_command();
+			let cmd = self.translate_env_into_env().into_command();
 			Either::Right(
 				cmd.over_ssh(session)
 					.map_err(|e| anyhow!("ssh error: {e}"))?,
@@ -126,6 +159,11 @@
 		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();
@@ -133,9 +171,10 @@
 		}
 		self
 	}
-	pub fn sudo(self) -> Self {
+	pub fn sudo(mut self) -> Self {
 		if std::env::var_os("NO_SUDO").is_some() {
 			let mut out = Self::new("su");
+			out.ssh_session = self.ssh_session.take();
 			out.arg("-c").arg(self.into_string());
 			out
 		} else {
@@ -144,27 +183,38 @@
 			out
 		}
 	}
-	pub fn ssh(self, on: impl AsRef<OsStr>) -> Self {
+	pub fn ssh_session(mut self, on: Arc<Session>) -> Self {
+		self.ssh_session = Some(on);
+		self
+	}
+	pub fn ssh(mut self, on: impl AsRef<OsStr>) -> Self {
 		let mut out = Self::new("ssh");
+		out.ssh_session = self.ssh_session.take();
 		out.arg(on).arg("--");
 		out.arg(self.into_string());
 		out
 	}
-	pub fn over_ssh(mut self, session: Arc<Session>) -> Self {
-		self.ssh_session = Some(session);
-		self
-	}
 
 	pub async fn run(self) -> Result<()> {
 		let str = self.clone().into_string();
-		let cmd = self.into_command();
-		run_nix_inner(str, cmd, &mut PlainHandler).await?;
+		let cmd = self.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.into_command();
-		let v = run_nix_inner_stdout(str, cmd, &mut PlainHandler).await?;
+		let cmd = self.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)
 	}
 
@@ -172,7 +222,8 @@
 		let str = self.clone().into_string();
 		let mut cmd = self.into_command();
 		cmd.arg("--log-format").arg("internal-json");
-		run_nix_inner_stdout(str, cmd, &mut NixHandler::default()).await
+		let bytes = run_nix_inner_stdout(str, cmd, &mut NixHandler::default()).await?;
+		Ok(String::from_utf8(bytes)?)
 	}
 	pub async fn run_nix(self) -> Result<()> {
 		let str = self.clone().into_string();
@@ -198,7 +249,7 @@
 	str: String,
 	cmd: Command,
 	handler: &mut dyn Handler,
-) -> Result<String> {
+) -> Result<Vec<u8>> {
 	Ok(run_nix_inner_raw(str, cmd, true, handler, None)
 		.await?
 		.expect("has out"))
@@ -208,6 +259,24 @@
 	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(())
+}
 
 pub trait Handler: Send {
 	fn handle_line(&mut self, e: &str);
@@ -468,7 +537,7 @@
 	want_stdout: bool,
 	err_handler: &mut dyn Handler,
 	mut out_handler: Option<&mut dyn Handler>,
-) -> Result<Option<String>> {
+) -> Result<Option<Vec<u8>>> {
 	cmd.stderr(Stdio::piped());
 	cmd.stdout(Stdio::piped());
 	let mut child = cmd.spawn()?;
@@ -522,7 +591,71 @@
 		}
 	}
 
-	Ok(out_buf.map(String::from_utf8).transpose()?)
+	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>>> {
+	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)
 }
 
 pub trait ErrorRecorder: Send {

--- a/cmds/fleet/src/host.rs
+++ b/cmds/fleet/src/host.rs
@@ -1,10 +1,10 @@
 use std::{
 	env::current_dir,
-	ffi::OsString,
+	ffi::{OsStr, OsString},
 	io::Write,
 	ops::Deref,
 	path::PathBuf,
-	sync::{Arc, Mutex, MutexGuard},
+	sync::{Arc, Mutex, MutexGuard, OnceLock},
 };
 
 use anyhow::{anyhow, bail, Context, Result};
@@ -46,16 +46,55 @@
 
 pub struct ConfigHost {
 	pub name: String,
+	pub session: OnceLock<Arc<openssh::Session>>,
 }
 impl ConfigHost {
-	async fn open_session(&self) -> Result<openssh::Session> {
-		let mut session = SessionBuilder::default();
+	pub async fn open_session(&self) -> Result<Arc<openssh::Session>> {
+		// FIXME: TOCTOU
+		if let Some(session) = &self.session.get() {
+			return Ok((*session).clone());
+		};
+		let session = SessionBuilder::default();
 
-		session
+		let session = session
 			.connect(&self.name)
 			.await
-			.map_err(|e| anyhow!("ssh error: {e}"))
+			.map_err(|e| anyhow!("ssh error: {e}"))?;
+		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 cmd(&self, cmd: impl AsRef<OsStr>) -> Result<MyCommand> {
+		let session = self.open_session().await?;
+		Ok(MyCommand::new_on(cmd, session))
+	}
+
+	pub async fn decrypt(&self, data: Vec<u8>) -> Result<Vec<u8>> {
+		let mut cmd = self.cmd("fleet-install-secrets").await?;
+		cmd.arg("decrypt").eqarg("--secret", z85::encode(&data));
+		let encoded = cmd
+			.sudo()
+			.run_string()
+			.await
+			.context("failed to call remote host for decrypt")?;
+		z85::decode(encoded.trim_end()).context("bad encoded data? outdated host?")
+	}
 }
 
 impl Config {
@@ -96,12 +135,21 @@
 		command.run_string().await
 	}
 
+	pub async fn host(&self, name: &str) -> Result<ConfigHost> {
+		Ok(ConfigHost {
+			name: name.to_owned(),
+			session: OnceLock::new(),
+		})
+	}
 	pub async fn list_hosts(&self) -> Result<Vec<ConfigHost>> {
 		let fleet_field = &self.fleet_field;
 		let names = nix_go!(fleet_field.configuredHosts).list_fields().await?;
 		let mut out = vec![];
 		for name in names {
-			out.push(ConfigHost { name })
+			out.push(ConfigHost {
+				name,
+				session: OnceLock::new(),
+			})
 		}
 		Ok(out)
 	}
@@ -152,19 +200,6 @@
 		host_secrets.insert(secret, value);
 	}
 
-	pub async fn decrypt_on_host(&self, host: &str, data: Vec<u8>) -> Result<Vec<u8>> {
-		let data = z85::encode(&data);
-		let mut cmd = MyCommand::new("fleet-install-secrets");
-		cmd.arg("decrypt").eqarg("--secret", data);
-		cmd = cmd.sudo().ssh(host);
-		let encoded = cmd
-			.run_string()
-			.await
-			.context("failed to call remote host for decrypt")?
-			.trim()
-			.to_owned();
-		z85::decode(encoded).context("bad encoded data? outdated host?")
-	}
 	pub async fn reencrypt_on_host(
 		&self,
 		host: &str,

--- a/nixos/meta.nix
+++ b/nixos/meta.nix
@@ -1,11 +1,18 @@
-{ lib, ... }:
-with lib;
 {
+  lib,
+  pkgs,
+  ...
+}:
+with lib; {
   options = with types; {
+    nixpkgs.resolvedPkgs = mkOption {
+      type = types.pkgs // {description = "nixpkgs.pkgs";};
+      description = "Value of pkgs";
+    };
     tags = mkOption {
       type = listOf str;
       description = "Host tags";
-      default = [ ];
+      default = [];
     };
     network = mkOption {
       type = submodule {
@@ -13,12 +20,12 @@
           internalIps = mkOption {
             type = listOf str;
             description = "Internal ips";
-            default = [ ];
+            default = [];
           };
           externalIps = mkOption {
             type = listOf str;
             description = "External ips";
-            default = [ ];
+            default = [];
           };
         };
       };
@@ -29,7 +36,8 @@
     };
   };
   config = {
-    tags = [ "all" ];
-    network = { };
+    tags = ["all"];
+    network = {};
+    nixpkgs.resolvedPkgs = pkgs;
   };
 }