1use std::{2 cell::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, 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;2829use crate::{30 command::MyCommand,31 fleetdata::{FleetData, FleetSecret, FleetSharedSecret},32};3334pub struct FleetConfigInternals {35 pub local_system: String,36 pub directory: PathBuf,37 pub opts: FleetOpts,38 pub data: Mutex<FleetData>,39 pub nix_args: Vec<OsString>,40 41 pub config_field: Value,42 43 pub config_unchecked_field: Value,4445 46 pub default_pkgs: Value,47}4849#[derive(Clone)]50pub struct Config(Arc<FleetConfigInternals>);5152impl Deref for Config {53 type Target = FleetConfigInternals;5455 fn deref(&self) -> &Self::Target {56 &self.057 }58}5960pub struct ConfigHost {61 config: Config,62 pub name: String,63 pub local: bool,64 pub session: OnceLock<Arc<openssh::Session>>,65 groups: OnceCell<Vec<String>>,6667 pub nixos_config: Option<Value>,68}69impl ConfigHost {70 pub async fn tags(&self) -> Result<Vec<String>> {71 if let Some(v) = self.groups.get() {72 return Ok(v.clone());73 }74 75 76 let Some(nixos_config) = &self.nixos_config else {77 return Ok(vec![]);78 };79 let tags: Vec<String> = nix_go_json!(nixos_config.tags);8081 let _ = self.groups.set(tags.clone());8283 Ok(tags)84 }85 async fn open_session(&self) -> Result<Arc<openssh::Session>> {86 assert!(!self.local, "do not open ssh connection to local session");87 88 if let Some(session) = &self.session.get() {89 return Ok((*session).clone());90 };91 let session = SessionBuilder::default();9293 let session = session94 .connect(&self.name)95 .await96 .map_err(|e| anyhow!("ssh error while connecting to {}: {e}", self.name))?;97 let session = Arc::new(session);98 self.session.set(session.clone()).expect("TOCTOU happened");99 Ok(session)100 }101 pub async fn mktemp_dir(&self) -> Result<String> {102 let mut cmd = self.cmd("mktemp").await?;103 cmd.arg("-d");104 let path = cmd.run_string().await?;105 Ok(path.trim_end().to_owned())106 }107 pub async fn read_file_bin(&self, path: impl AsRef<OsStr>) -> Result<Vec<u8>> {108 let mut cmd = self.cmd("cat").await?;109 cmd.arg(path);110 cmd.run_bytes().await111 }112 pub async fn read_file_text(&self, path: impl AsRef<OsStr>) -> Result<String> {113 let mut cmd = self.cmd("cat").await?;114 cmd.arg(path);115 cmd.run_string().await116 }117 pub async fn read_dir(&self, path: impl AsRef<OsStr>) -> Result<Vec<String>> {118 let mut cmd = self.cmd("ls").await?;119 cmd.arg(path);120 let out = cmd.run_string().await?;121 let mut lines = out.split('\n');122 if let Some(last) = lines.next_back() {123 ensure!(last.is_empty(), "output of ls should end with newline");124 }125 Ok(lines.map(ToOwned::to_owned).collect())126 }127 #[allow(dead_code)]128 pub async fn read_file_json<D: DeserializeOwned>(&self, path: impl AsRef<OsStr>) -> Result<D> {129 let text = self.read_file_text(path).await?;130 Ok(serde_json::from_str(&text)?)131 }132 pub async fn read_file_value<D: FromStr>(&self, path: impl AsRef<OsStr>) -> Result<D>133 where134 <D as FromStr>::Err: Display,135 {136 let text = self.read_file_text(path).await?;137 D::from_str(&text).map_err(|e| anyhow!("failed to parse value: {e}"))138 }139 pub async fn cmd(&self, cmd: impl AsRef<OsStr>) -> Result<MyCommand> {140 if self.local {141 Ok(MyCommand::new(cmd))142 } else {143 let session = self.open_session().await?;144 Ok(MyCommand::new_on(cmd, session))145 }146 }147148 pub async fn decrypt(&self, data: SecretData) -> Result<Vec<u8>> {149 ensure!(data.encrypted, "secret is not encrypted");150 let mut cmd = self.cmd("fleet-install-secrets").await?;151 cmd.arg("decrypt").eqarg("--secret", data.to_string());152 let encoded = cmd153 .sudo()154 .run_string()155 .await156 .context("failed to call remote host for decrypt")?;157 let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;158 ensure!(!data.encrypted, "secret came out encrypted");159 Ok(data.data)160 }161 pub async fn reencrypt(&self, data: SecretData, targets: Vec<String>) -> Result<SecretData> {162 ensure!(data.encrypted, "secret is not encrypted");163 let mut cmd = self.cmd("fleet-install-secrets").await?;164 cmd.arg("reencrypt").eqarg("--secret", data.to_string());165 for target in targets {166 let key = self.config.key(&target).await?;167 cmd.eqarg("--targets", key);168 }169 let encoded = cmd170 .sudo()171 .run_string()172 .await173 .context("failed to call remote host for decrypt")?;174 let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;175 ensure!(data.encrypted, "secret came out not encrypted");176 Ok(data)177 }178 179 pub async fn remote_derivation(&self, path: &PathBuf) -> Result<PathBuf> {180 if self.local {181 182 return Ok(path.to_owned());183 }184 let mut nix = MyCommand::new("nix");185 nix.arg("copy")186 .arg("--substitute-on-destination")187 .comparg("--to", format!("ssh-ng://{}", self.name))188 .arg(path);189 nix.run_nix().await.context("nix copy")?;190 Ok(path.to_owned())191 }192 pub async fn systemctl_stop(&self, name: &str) -> Result<()> {193 let mut cmd = self.cmd("systemctl").await?;194 cmd.arg("stop").arg(name);195 cmd.sudo().run().await196 }197 pub async fn systemctl_start(&self, name: &str) -> Result<()> {198 let mut cmd = self.cmd("systemctl").await?;199 cmd.arg("start").arg(name);200 cmd.sudo().run().await201 }202203 pub async fn rm_file(&self, path: impl AsRef<OsStr>, sudo: bool) -> Result<()> {204 let mut cmd = self.cmd("rm").await?;205 cmd.arg("-f").arg(path);206 if sudo {207 cmd = cmd.sudo()208 }209 cmd.run().await210 }211212 pub async fn list_configured_secrets(&self) -> Result<Vec<String>> {213 let Some(nixos) = &self.nixos_config else {214 return Ok(vec![]);215 };216 let secrets = nix_go!(nixos.secrets);217 let mut out = Vec::new();218 for name in secrets.list_fields().await? {219 let secret = nix_go!(secrets[{ name }]);220 let is_shared: bool = nix_go_json!(secret.shared);221 if is_shared {222 continue;223 }224 out.push(name);225 }226 Ok(out)227 }228 pub async fn secret_field(&self, name: &str) -> Result<Value> {229 let Some(nixos) = &self.nixos_config else {230 bail!("host is virtual and has no secrets");231 };232 Ok(nix_go!(nixos.secrets[{ name }]))233 }234235 236 pub async fn pkgs(&self) -> Result<Value> {237 let Some(nixos) = &self.nixos_config else {238 return Ok(self.config.default_pkgs.clone());239 };240 Ok(nix_go!(nixos.nixpkgs.resolvedPkgs))241 }242}243244impl Config {245 pub async fn should_skip(&self, host: &ConfigHost) -> Result<bool> {246 if !self.opts.skip.is_empty() && self.opts.skip.iter().any(|h| h as &str == host.name) {247 return Ok(true);248 }249 if self.opts.only.is_empty() {250 return Ok(false);251 }252 let mut have_group_matches = false;253 for item in self.opts.only.iter() {254 match item {255 HostItem::Host { name, .. } if *name == host.name => {256 return Ok(false);257 }258 HostItem::Tag { .. } => {259 have_group_matches = true;260 }261 _ => {}262 }263 }264 if have_group_matches {265 let host_tags = host.tags().await?;266 for item in self.opts.only.iter() {267 match item {268 HostItem::Tag { name, .. } if host_tags.contains(name) => {269 return Ok(false);270 }271 _ => {}272 }273 }274 }275 Ok(true)276 }277 pub async fn action_attr(&self, host: &ConfigHost, attr: &str) -> Result<Option<String>> {278 if self.opts.only.is_empty() {279 return Ok(None);280 }281 let mut have_group_matches = false;282 for item in self.opts.only.iter() {283 match item {284 HostItem::Host { name, attrs }285 if *name == host.name && attrs.contains_key(attr) =>286 {287 return Ok(attrs.get(attr).cloned());288 }289 HostItem::Tag { attrs, .. } if attrs.contains_key(attr) => {290 have_group_matches = true;291 }292 _ => {}293 }294 }295 if have_group_matches {296 let host_tags = host.tags().await?;297 for item in self.opts.only.iter() {298 match item {299 HostItem::Tag { name, attrs }300 if host_tags.contains(name) && attrs.contains_key(attr) =>301 {302 return Ok(attrs.get(attr).cloned());303 }304 _ => {}305 }306 }307 }308 Ok(None)309 }310 pub fn is_local(&self, host: &str) -> bool {311 self.opts.localhost.as_ref().map(|s| s as &str) == Some(host)312 }313314 pub fn local_host(&self) -> ConfigHost {315 ConfigHost {316 config: self.clone(),317 name: "<virtual localhost>".to_owned(),318 local: true,319 session: OnceLock::new(),320 nixos_config: None,321 groups: {322 let cell = OnceCell::new();323 let _ = cell.set(vec![]);324 cell325 },326 }327 }328329 pub async fn host(&self, name: &str) -> Result<ConfigHost> {330 let config = &self.config_unchecked_field;331 let nixos_config = nix_go!(config.hosts[{ name }].nixosSystem.config);332 Ok(ConfigHost {333 config: self.clone(),334 name: name.to_owned(),335 local: self.is_local(name),336 session: OnceLock::new(),337 nixos_config: Some(nixos_config),338 groups: OnceCell::new(),339 })340 }341 pub async fn list_hosts(&self) -> Result<Vec<ConfigHost>> {342 let config = &self.config_unchecked_field;343 let names = nix_go!(config.hosts).list_fields().await?;344 let mut out = vec![];345 for name in names {346 out.push(self.host(&name).await?);347 }348 Ok(out)349 }350 pub async fn system_config(&self, host: &str) -> Result<Value> {351 let fleet_field = &self.config_unchecked_field;352 Ok(nix_go!(fleet_field.hosts[{ host }].nixosSystem.config))353 }354355 pub(super) fn data(&self) -> MutexGuard<FleetData> {356 self.data.lock().unwrap()357 }358 pub(super) fn data_mut(&self) -> MutexGuard<FleetData> {359 self.data.lock().unwrap()360 }361 362 pub async fn list_configured_shared(&self) -> Result<Vec<String>> {363 let config_field = &self.config_unchecked_field;364 Ok(nix_go!(config_field.sharedSecrets).list_fields().await?)365 }366 367 pub fn list_shared(&self) -> Vec<String> {368 let data = self.data();369 data.shared_secrets.keys().cloned().collect()370 }371 pub fn has_shared(&self, name: &str) -> bool {372 let data = self.data();373 data.shared_secrets.contains_key(name)374 }375 pub fn replace_shared(&self, name: String, shared: FleetSharedSecret) {376 let mut data = self.data_mut();377 data.shared_secrets.insert(name.to_owned(), shared);378 }379 pub fn remove_shared(&self, secret: &str) {380 let mut data = self.data_mut();381 data.shared_secrets.remove(secret);382 }383384 pub fn list_secrets(&self, host: &str) -> Vec<String> {385 let data = self.data();386 let Some(secrets) = data.host_secrets.get(host) else {387 return Vec::new();388 };389 secrets.keys().cloned().collect()390 }391392 pub fn has_secret(&self, host: &str, secret: &str) -> bool {393 let data = self.data();394 let Some(host_secrets) = data.host_secrets.get(host) else {395 return false;396 };397 host_secrets.contains_key(secret)398 }399 pub fn insert_secret(&self, host: &str, secret: String, value: FleetSecret) {400 let mut data = self.data_mut();401 let host_secrets = data.host_secrets.entry(host.to_owned()).or_default();402 host_secrets.insert(secret, value);403 }404405 pub fn host_secret(&self, host: &str, secret: &str) -> Result<FleetSecret> {406 let data = self.data();407 let Some(host_secrets) = data.host_secrets.get(host) else {408 bail!("no secrets for machine {host}");409 };410 let Some(secret) = host_secrets.get(secret) else {411 bail!("machine {host} has no secret {secret}");412 };413 Ok(secret.clone())414 }415 pub fn shared_secret(&self, secret: &str) -> Result<FleetSharedSecret> {416 let data = self.data();417 let Some(secret) = data.shared_secrets.get(secret) else {418 bail!("no shared secret {secret}");419 };420 Ok(secret.clone())421 }422 pub async fn shared_secret_expected_owners(&self, secret: &str) -> Result<Vec<String>> {423 let config_field = &self.config_unchecked_field;424 Ok(nix_go_json!(425 config_field.sharedSecrets[{ secret }].expectedOwners426 ))427 }428429 pub fn save(&self) -> Result<()> {430 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.")?;431 let data = nixlike::serialize(&self.data() as &FleetData)?;432 tempfile.write_all(433 format!(434 "# This file contains fleet state and shouldn't be edited by hand\n\n{}\n\n# vim: ts=2 et nowrap\n",435 data436 )437 .as_bytes(),438 )?;439 let mut fleet_data_path = self.directory.clone();440 fleet_data_path.push("fleet.nix");441 tempfile.persist(fleet_data_path)?;442 Ok(())443 }444}445446#[derive(Clone)]447enum HostItem {448 Host {449 name: String,450 attrs: BTreeMap<String, String>,451 },452 Tag {453 name: String,454 attrs: BTreeMap<String, String>,455 },456}457fn host_item_parser(input: &str) -> Result<HostItem, String> {458 fn err_to_string(err: nom::Err<nom::error::Error<&str>>) -> String {459 err.to_string()460 }461462 let (input, is_tag) = map(opt(char('@')), |c| c.is_some())(input).map_err(err_to_string)?;463 let (input, name) = map(464 take_while1(|v| v != ',' && v != '?' && v != '@'),465 str::to_owned,466 )(input)467 .map_err(err_to_string)?;468469 let kw_item = separated_pair(470 map(take_while1(|v| v != '&' && v != '='), str::to_owned),471 char('='),472 map(take_while1(|v| v != '&'), str::to_owned),473 );474 let kw = map(separated_list1(char('&'), kw_item), |vec| {475 vec.into_iter().collect::<BTreeMap<_, _>>()476 });477 let mut opt_kw = map(opt(preceded(char('?'), kw)), Option::unwrap_or_default);478479 let (input, attrs) = opt_kw(input).map_err(err_to_string)?;480481 if !input.is_empty() {482 return Err(format!("unexpected trailing input: {input:?}"));483 }484 Ok(if is_tag {485 HostItem::Tag { name, attrs }486 } else {487 HostItem::Host { name, attrs }488 })489}490491#[derive(Parser, Clone)]492pub struct FleetOpts {493 494 #[clap(long, number_of_values = 1, value_parser = host_item_parser)]495 only: Vec<HostItem>,496497 498 #[clap(long, number_of_values = 1)]499 skip: Vec<String>,500501 502 #[clap(long)]503 pub localhost: Option<String>,504505 506 507 #[clap(long, default_value = "detect")]508 pub local_system: String,509}510511impl FleetOpts {512 pub async fn build(mut self, nix_args: Vec<OsString>) -> Result<Config> {513 if self.localhost.is_none() {514 self.localhost515 .replace(hostname::get().unwrap().to_str().unwrap().to_owned());516 }517 let directory = current_dir()?;518519 let pool = NixSessionPool::new(directory.as_os_str().to_owned(), nix_args.clone()).await?;520 let root_field = pool.get().await?;521522 let builtins_field = Value::binding(root_field.clone(), "builtins").await?;523 if self.local_system == "detect" {524 self.local_system = nix_go_json!(builtins_field.currentSystem);525 }526 let local_system = self.local_system.clone();527528 let fleet_root = Value::binding(root_field, "fleetConfigurations").await?;529 let fleet_field = nix_go!(fleet_root.default);530531 let config_field = nix_go!(fleet_field.config);532 let config_unchecked_field = nix_go!(fleet_field.unchecked.config);533534 let import = nix_go!(builtins_field.import);535 let overlays = nix_go!(config_unchecked_field.overlays);536 let nixpkgs = nix_go!(fleet_field.nixpkgs | import);537538 let default_pkgs = nix_go!(nixpkgs(Obj {539 overlays,540 system: { self.local_system.clone() },541 }));542543 let mut fleet_data_path = directory.clone();544 fleet_data_path.push("fleet.nix");545 let bytes = std::fs::read_to_string(fleet_data_path)?;546 let data = nixlike::parse_str(&bytes)?;547548 Ok(Config(Arc::new(FleetConfigInternals {549 opts: self,550 directory,551 data,552 local_system,553 nix_args,554 config_field,555 config_unchecked_field,556 default_pkgs,557 })))558 }559}