--- a/cmds/fleet/src/command.rs
+++ b/cmds/fleet/src/command.rs
@@ -1,4 +1,4 @@
-use std::{ffi::OsStr, process::Stdio};
+use std::{ffi::OsStr, process::Stdio, task::Poll};
use anyhow::{Context, Result};
use async_trait::async_trait;
@@ -7,20 +7,292 @@
de::{DeserializeOwned, Visitor},
Deserialize,
};
-use tokio::{process::Command, select};
+use tokio::{io::AsyncRead, process::Command, select};
use tokio_util::codec::{BytesCodec, FramedRead, LinesCodec};
use tracing::{info, warn};
+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)]
+pub struct MyCommand {
+ command: String,
+ args: Vec<String>,
+ env: Vec<(String, String)>,
+}
+impl MyCommand {
+ pub fn new(cmd: impl AsRef<OsStr>) -> Self {
+ assert!(!cmd.as_ref().is_empty());
+ Self {
+ command: ostoutf8(cmd),
+ args: vec![],
+ env: vec![],
+ }
+ }
+ 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.into_iter());
+ 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
+ }
+ 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 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(self) -> Self {
+ let mut out = Self::new("sudo");
+ out.args(self.into_args());
+ out
+ }
+ pub fn ssh(self, on: impl AsRef<OsStr>) -> Self {
+ let mut out = Self::new("ssh");
+ out.arg(on).arg("--");
+ out.arg(self.into_string());
+ out
+ }
+
+ pub async fn run(self) -> Result<()> {
+ let str = self.clone().into_string();
+ info!("running {str}");
+ let mut cmd = self.into_command();
+ cmd.inherit_stdio();
+ let out = cmd.spawn()?.wait_with_output().await?;
+ if !out.status.success() {
+ anyhow::bail!("command '{}' failed with status {}", str, out.status);
+ }
+ Ok(())
+ }
+ pub async fn run_string(self) -> Result<String> {
+ let str = self.clone().into_string();
+ info!("running {str}");
+ let mut cmd = self.into_command();
+ cmd.inherit_stdio();
+ cmd.stdout(Stdio::piped());
+ let out = cmd.spawn()?.wait_with_output().await?;
+ if !out.status.success() {
+ anyhow::bail!("command '{}' failed with status {}", str, out.status);
+ }
+ Ok(String::from_utf8(out.stdout)?)
+ }
+ pub async fn run_nix_json<T: DeserializeOwned>(self) -> Result<T> {
+ let str = self.run_nix_string().await?;
+ serde_json::from_str(&str).with_context(|| format!("{:?}", str))
+ }
+
+ pub async fn run_nix_string(self) -> Result<String> {
+ let str = self.clone().into_string();
+ let mut cmd = self.into_command();
+ cmd.stdout(Stdio::piped());
+ run_nix_inner(str, cmd).await.map(|v| v.unwrap())
+ }
+ pub async fn run_nix(self) -> Result<()> {
+ let str = self.clone().into_string();
+ let mut cmd = self.into_command();
+ cmd.stdout(Stdio::inherit());
+ run_nix_inner(str, cmd).await.map(|v| {
+ assert!(v.is_none());
+ })
+ }
+}
+
+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(str: String, mut cmd: Command) -> Result<Option<String>> {
+ info!("running {str}");
+ cmd.arg("--log-format").arg("internal-json");
+ cmd.stderr(Stdio::piped());
+ let mut child = cmd.spawn()?;
+ let mut stderr = child.stderr.take().unwrap();
+ let stdout = child.stdout.take();
+ let wants_stdout = stdout.is_some();
+ let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
+ let mut out: Box<dyn AsyncRead + Unpin> = stdout
+ .map(|s| Box::new(s) as Box<dyn AsyncRead + Unpin>)
+ .unwrap_or_else(|| Box::new(EmptyAsyncRead));
+ let mut out = FramedRead::new(&mut out, BytesCodec::new());
+
+ // while let Some(line) = read.next().await? {}
+
+ let mut out_buf = if wants_stdout { Some(vec![]) } else { None };
+ loop {
+ select! {
+ e = err.next() => {
+ if let Some(e) = e {
+ let e = e?;
+ if let Some(e) = e.strip_prefix("@nix ") {
+
+ let log: NixLog = match serde_json::from_str(e) {
+ Ok(l) => l,
+ Err(err) => {
+ warn!("failed to parse nix log line {:?}: {}", e, err);
+ continue;
+ },
+ };
+ match log {
+ NixLog::Msg { msg, raw_msg, .. } => {
+ if !(msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m Git tree '") && msg.ends_with("' is dirty"))
+ && !msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m not writing modified lock file of flake")
+ && msg != "\u{1b}[35;1mwarning:\u{1b}[0m \u{1b}[31;1merror:\u{1b}[0m SQLite database '\u{1b}[35;1m/nix/var/nix/db/db.sqlite\u{1b}[0m' is busy" {
+ if let Some(raw_msg) = raw_msg {
+ info!(target: "nix", "{raw_msg}\n{msg}")
+ }else {
+ info!(target: "nix", "{msg}")
+
+ }
+ }
+ },
+ NixLog::Start { ref fields, typ, .. } if typ == 105 && !fields.is_empty() => {
+ if let [LogField::String(drv), ..] = &fields[..] {
+ info!(target: "nix","building {}", drv)
+ } else {
+ warn!("bad build log: {:?}", log)
+ }
+ },
+ NixLog::Start { ref fields, typ, .. } if typ == 100 && fields.len() >= 3 => {
+ if let [LogField::String(drv), LogField::String(from), LogField::String(to), ..] = &fields[..] {
+ info!(target: "nix","copying {} {} -> {}", drv, from, to)
+ } else {
+ warn!("bad copy log: {:?}", log)
+ }
+ },
+ NixLog::Start { text, typ, .. } if typ == 0 || typ == 102 || typ == 103 || typ == 104 => {
+ if !text.is_empty() && text != "querying info about missing paths" && text != "copying 0 paths" {
+ info!(target: "nix", "{}", text)
+ }
+ },
+ NixLog::Start { text, level: 0, typ: 108, .. } if text.is_empty() => {
+ // Cache lookup? Coupled with copy log
+ },
+ NixLog::Start { text, level: 4, typ: 109, .. } if text.starts_with("querying info about ") => {
+ // Cache lookup
+ }
+ NixLog::Start { text, level: 4, typ: 101, .. } if text.starts_with("downloading ") => {
+ // NAR downloading, coupled with copy log
+ }
+ NixLog::Start { text, level: 1, typ: 111, .. } if text.starts_with("waiting for a machine to build ") => {
+ // Useless repeating notification about build
+ }
+ NixLog::Start { text, level: 3, typ: 111, .. } if text.starts_with("resolved derivation: ") => {
+ // CA resolved
+ }
+ NixLog::Stop { .. } => {},
+ NixLog::Result { .. } => {},
+ _ => warn!("unknown log: {:?}", log)
+ };
+ } else {
+ warn!(target="nix","unknown: {}", e)
+ }
+ }
+ },
+ o = out.next() => {
+ if let Some(o) = o {
+ out_buf.as_mut().expect("stdout == wants_stdout").extend_from_slice(&o?);
+ }
+ },
+ code = child.wait() => {
+ let code = code?;
+ if !code.success() {
+ anyhow::bail!("command '{str}' failed with status {}", code);
+ }
+ break;
+ }
+ }
+ }
+
+ Ok(out_buf.map(String::from_utf8).transpose()?)
+}
+
#[async_trait]
pub trait CommandExt {
- async fn run_nix(&mut self) -> Result<()>;
- async fn run_nix_json<T: DeserializeOwned>(&mut self) -> Result<T>;
- async fn run_nix_string(&mut self) -> Result<String>;
- async fn run(&mut self) -> Result<()>;
- async fn run_json<T: DeserializeOwned>(&mut self) -> Result<T>;
- async fn run_string(&mut self) -> Result<String>;
+ // async fn run_nix(&mut self) -> Result<()>;
+ // async fn run_nix_json<T: DeserializeOwned>(&mut self) -> Result<T>;
+ // async fn run_nix_string(&mut self) -> Result<String>;
+ // async fn run(&mut self) -> Result<()>;
+ // async fn run_json<T: DeserializeOwned>(&mut self) -> Result<T>;
+ // async fn run_string(&mut self) -> Result<String>;
fn inherit_stdio(&mut self) -> &mut Self;
- fn ssh_on(host: impl AsRef<OsStr>, command: impl AsRef<OsStr>) -> Self;
}
#[derive(Debug)]
@@ -91,170 +363,9 @@
#[async_trait]
impl CommandExt for Command {
- async fn run_nix(&mut self) -> Result<()> {
- self.run_nix_string().await.map(|_| ())
- }
- async fn run_nix_json<T: DeserializeOwned>(&mut self) -> Result<T> {
- let str = self.run_nix_string().await?;
- serde_json::from_str(&str).with_context(|| format!("{:?}", str))
- }
-
- async fn run_nix_string(&mut self) -> Result<String> {
- self.arg("--log-format").arg("internal-json");
- self.stderr(Stdio::piped());
- self.stdout(Stdio::piped());
- let mut child = self.spawn()?;
- let mut stderr = child.stderr.take().unwrap();
- let mut stdout = child.stdout.take().unwrap();
- let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
- let mut out = FramedRead::new(&mut stdout, BytesCodec::new());
-
- // while let Some(line) = read.next().await? {}
-
- let mut out_buf = vec![];
- loop {
- select! {
- e = err.next() => {
- if let Some(e) = e {
- let e = e?;
- if let Some(e) = e.strip_prefix("@nix ") {
-
- let log: NixLog = match serde_json::from_str(e) {
- Ok(l) => l,
- Err(err) => {
- warn!("failed to parse nix log line {:?}: {}", e, err);
- continue;
- },
- };
- match log {
- NixLog::Msg { msg, raw_msg, .. } => {
- if !(msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m Git tree '") && msg.ends_with("' is dirty"))
- && !msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m not writing modified lock file of flake")
- && msg != "\u{1b}[35;1mwarning:\u{1b}[0m \u{1b}[31;1merror:\u{1b}[0m SQLite database '\u{1b}[35;1m/nix/var/nix/db/db.sqlite\u{1b}[0m' is busy" {
- if let Some(raw_msg) = raw_msg {
- info!(target: "nix", "{raw_msg}\n{msg}")
- }else {
- info!(target: "nix", "{msg}")
-
- }
- }
- },
- NixLog::Start { ref fields, typ, .. } if typ == 105 && !fields.is_empty() => {
- if let [LogField::String(drv), ..] = &fields[..] {
- info!(target: "nix","building {}", drv)
- } else {
- warn!("bad build log: {:?}", log)
- }
- },
- NixLog::Start { ref fields, typ, .. } if typ == 100 && fields.len() >= 3 => {
- if let [LogField::String(drv), LogField::String(from), LogField::String(to), ..] = &fields[..] {
- info!(target: "nix","copying {} {} -> {}", drv, from, to)
- } else {
- warn!("bad copy log: {:?}", log)
- }
- },
- NixLog::Start { text, typ, .. } if typ == 0 || typ == 102 || typ == 103 || typ == 104 => {
- if !text.is_empty() && text != "querying info about missing paths" && text != "copying 0 paths" {
- info!(target: "nix", "{}", text)
- }
- },
- NixLog::Start { text, level: 0, typ: 108, .. } if text.is_empty() => {
- // Cache lookup? Coupled with copy log
- },
- NixLog::Start { text, level: 4, typ: 109, .. } if text.starts_with("querying info about ") => {
- // Cache lookup
- }
- NixLog::Start { text, level: 4, typ: 101, .. } if text.starts_with("downloading ") => {
- // NAR downloading, coupled with copy log
- }
- NixLog::Start { text, level: 1, typ: 111, .. } if text.starts_with("waiting for a machine to build ") => {
- // Useless repeating notification about build
- }
- NixLog::Start { text, level: 3, typ: 111, .. } if text.starts_with("resolved derivation: ") => {
- // CA resolved
- }
- NixLog::Stop { .. } => {},
- NixLog::Result { .. } => {},
- _ => warn!("unknown log: {:?}", log)
- };
- } else {
- warn!(target="nix","unknown: {}", e)
- }
- }
- },
- o = out.next() => {
- if let Some(o) = o {
- out_buf.extend_from_slice(&o?);
- }
- },
- code = child.wait() => {
- let code = code?;
- if !code.success() {
- anyhow::bail!("command ({:?}) failed with status {}", self, code);
- }
- break;
- }
- }
- }
-
- Ok(String::from_utf8(out_buf)?)
- }
-
fn inherit_stdio(&mut self) -> &mut Self {
self.stderr(Stdio::inherit());
+ self.stdout(Stdio::inherit());
self
- }
-
- async fn run(&mut self) -> Result<()> {
- self.stderr(Stdio::piped());
- self.stdout(Stdio::piped());
- let mut child = self.spawn()?;
- let mut stderr = child.stderr.take().unwrap();
- let mut stdout = child.stdout.take().unwrap();
- let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
- let mut out = FramedRead::new(&mut stdout, LinesCodec::new());
- loop {
- select! {
- e = err.next() => {
- if let Some(e) = e {
- warn!("{}", e?);
- }
- },
- o = out.next() => {
- if let Some(o) = o {
- info!("{}", o?);
- }
- },
- code = child.wait() => {
- let code = code?;
- if !code.success() {
- anyhow::bail!("command ({:?}) failed with status {}", self, code);
- }
- break;
- }
- }
- }
- Ok(())
- }
-
- async fn run_json<T: DeserializeOwned>(&mut self) -> Result<T> {
- let str = self.run_string().await?;
- serde_json::from_str(&str).with_context(|| format!("{:?}", str))
- }
-
- async fn run_string(&mut self) -> Result<String> {
- self.inherit_stdio();
- self.stdout(Stdio::piped());
- let out = self.spawn()?.wait_with_output().await?;
- if !out.status.success() {
- anyhow::bail!("command ({:?}) failed with status {}", self, out.status);
- }
- Ok(String::from_utf8(out.stdout)?)
- }
-
- fn ssh_on(host: impl AsRef<OsStr>, command: impl AsRef<OsStr>) -> Self {
- let mut cmd = Command::new("ssh");
- cmd.arg(host).arg("--").arg(command);
- cmd
}
}
--- a/cmds/fleet/src/host.rs
+++ b/cmds/fleet/src/host.rs
@@ -1,7 +1,7 @@
use std::{
cell::{Ref, RefCell, RefMut},
env::current_dir,
- ffi::{OsStr, OsString},
+ ffi::OsString,
io::Write,
ops::Deref,
path::PathBuf,
@@ -12,10 +12,9 @@
use clap::{ArgGroup, Parser};
use serde::de::DeserializeOwned;
use tempfile::NamedTempFile;
-use tokio::process::Command;
use crate::{
- command::CommandExt,
+ command::MyCommand,
fleetdata::{FleetData, FleetSecret, FleetSharedSecret},
};
@@ -52,24 +51,24 @@
self.opts.localhost.as_ref().map(|s| s as &str) == Some(host)
}
- pub fn command_on(&self, host: &str, program: impl AsRef<OsStr>, sudo: bool) -> Command {
- if self.is_local(host) {
- if sudo {
- let mut cmd = Command::new("sudo");
- cmd.arg(program);
- cmd
- } else {
- Command::new(program)
- }
- } else {
- let mut cmd = Command::new("ssh");
- cmd.arg(host).arg("--");
- if sudo {
- cmd.arg("sudo");
- }
- cmd.arg(program);
- cmd
+ pub async fn run_on(&self, host: &str, mut command: MyCommand, sudo: bool) -> Result<()> {
+ if sudo {
+ command = command.sudo();
+ }
+ if !self.is_local(host) {
+ command = command.ssh(host);
+ }
+ command.run().await
+ }
+ #[must_use]
+ pub async fn run_string_on(&self, host: &str, mut command: MyCommand, sudo: bool) -> Result<String> {
+ if sudo {
+ command = command.sudo();
+ }
+ if !self.is_local(host) {
+ command = command.ssh(host);
}
+ command.run_string().await
}
pub fn configuration_attr_name(&self, name: &str) -> OsString {
@@ -83,36 +82,36 @@
}
pub async fn list_hosts(&self) -> Result<Vec<String>> {
- Command::new("nix")
- .arg("eval")
+ let mut cmd = MyCommand::new("nix");
+ cmd.arg("eval")
.arg(self.configuration_attr_name("configuredHosts"))
.args(["--apply", "builtins.attrNames", "--json", "--show-trace"])
- .args(&self.nix_args)
- .run_nix_json()
+ .args(&self.nix_args);
+ cmd.run_nix_json()
.await
}
pub async fn shared_config_attr<T: DeserializeOwned>(&self, attr: &str) -> Result<T> {
- Command::new("nix")
- .arg("eval")
+ let mut cmd = MyCommand::new("nix");
+ cmd.arg("eval")
.arg(self.configuration_attr_name(&format!("configUnchecked.{}", attr)))
.args(["--json", "--show-trace"])
- .args(&self.nix_args)
- .run_nix_json()
+ .args(&self.nix_args);
+ cmd.run_nix_json()
.await
}
pub async fn shared_config_attr_names(&self, attr: &str) -> Result<Vec<String>> {
- Command::new("nix")
- .arg("eval")
+ let mut cmd = MyCommand::new("nix");
+ cmd.arg("eval")
.arg(self.configuration_attr_name(&format!("configUnchecked.{}", attr)))
.args(["--apply", "builtins.attrNames"])
.args(["--json", "--show-trace"])
- .args(&self.nix_args)
- .run_nix_json()
+ .args(&self.nix_args);
+ cmd.run_nix_json()
.await
}
pub async fn config_attr<T: DeserializeOwned>(&self, host: &str, attr: &str) -> Result<T> {
- Command::new("nix")
- .arg("eval")
+ let mut cmd = MyCommand::new("nix");
+ cmd.arg("eval")
.arg(
self.configuration_attr_name(&format!(
"configuredSystems.{}.config.{}",
@@ -120,8 +119,8 @@
)),
)
.args(["--json", "--show-trace"])
- .args(&self.nix_args)
- .run_nix_json()
+ .args(&self.nix_args);
+ cmd.run_nix_json()
.await
}
@@ -171,23 +170,20 @@
pub async fn decrypt_on_host(&self, host: &str, data: Vec<u8>) -> Result<Vec<u8>>{
let data = z85::encode(&data);
- let encoded = self.command_on(host, "fleet-install-secrets", true)
- .arg("decrypt")
- .arg("--secret")
- .arg(data).run_string().await.context("failed to call remote host for decrypt")?.trim().to_owned();
+ 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();
Ok(z85::decode(encoded).context("bad encoded data? outdated host?")?)
}
pub async fn reencrypt_on_host(&self, host: &str, data: Vec<u8>, targets: Vec<String>) -> Result<Vec<u8>>{
let data = z85::encode(&data);
- let mut recmd = self.command_on(host, "fleet-install-secrets", true);
- recmd
- .arg("reencrypt")
- .arg("--secret")
- .arg(format!("\"{}\"", data.replace('$', "\\$")));
+ let mut recmd = MyCommand::new("fleet-install-secrets");
+ recmd.arg("reencrypt").eqarg("--secret",data);
for target in targets {
- recmd.arg("--targets");
- recmd.arg(format!("\"{target}\""));
+ recmd.eqarg("--targets", target);
}
+ recmd = recmd.sudo().ssh(host);
let encoded = recmd.run_string().await.context("failed to call remote host for decrypt")?.trim().to_owned();
Ok(z85::decode(encoded).context("bad encoded data? outdated host?")?)
}