1use std::{2 collections::{BTreeMap, BTreeSet, HashSet},3 future::Future,4 io::Write,5 ops::Deref,6 path::PathBuf,7 pin::Pin,8 str::FromStr,9 sync::{Arc, OnceLock},10};1112use anyhow::{Context, Result, anyhow, bail, ensure};13use camino::{Utf8Path, Utf8PathBuf};14use chrono::{DateTime, Utc};15use fleet_shared::SecretData;16use nix_eval::{Store, Value, eval_store, nix_go, nix_go_json, util::assert_warn};17use remowt_client::{AgentBundle, Remowt};18use remowt_endpoints::fs::FsClient;19use remowt_fleet::NixClient;20use remowt_link_shared::{Address, BifConfig};21use remowt_ui_prompt::auto::AutoPrompter;22use remowt_ui_prompt::bifrost::PromptEndpoints;23use remowt_ui_prompt::{PrependSourcePrompter, Source};24use tempfile::NamedTempFile;25use tokio::{process::Command, sync::OnceCell, task::spawn_blocking};26use tracing::{info, instrument, warn};2728use crate::fleetdata::{29 FleetData, FleetSecretData, FleetSecretDistribution, FleetSecretPart, SecretOwner,30};3132pub struct FleetConfigInternals {33 pub prefer_identities: BTreeSet<SecretOwner>,34 pub now: DateTime<Utc>,3536 37 pub directory: PathBuf,38 39 pub local_system: String,40 pub data: Arc<FleetData>,41 42 pub config_field: Value,43 44 pub flake_outputs: Value,45 46 pub localhost: String,4748 49 pub default_pkgs: Value,50 51 pub nixpkgs: Value,5253 pub local_host: OnceLock<Arc<ConfigHost>>,54}555657#[derive(Clone)]58pub struct Config(pub Arc<FleetConfigInternals>);5960impl Deref for Config {61 type Target = FleetConfigInternals;6263 fn deref(&self) -> &Self::Target {64 &self.065 }66}6768#[derive(Clone, PartialEq, Copy, Debug)]69pub enum DeployKind {70 71 UpgradeToFleet,72 73 Fleet,74 75 76 NixosInstall,77 78 79 80 NixosLustrate,81}8283impl FromStr for DeployKind {84 type Err = anyhow::Error;85 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {86 match s {87 "upgrade-to-fleet" => Ok(Self::UpgradeToFleet),88 "fleet" => Ok(Self::Fleet),89 "nixos-install" => Ok(Self::NixosInstall),90 "nixos-lustrate" => Ok(Self::NixosLustrate),91 v => bail!(92 "unknown deploy_kind: {v}; expected on of \"upgrade-to-fleet\", \"fleet\", \"nixos-install\", \"nixos-lustrate\""93 ),94 }95 }96}97pub struct ConfigHost {98 config: Config,99 pub name: String,100 groups: OnceLock<Vec<String>>,101102 103 deploy_kind: OnceCell<DeployKind>,104 session_destination: OnceLock<String>,105 legacy_ssh_store: OnceLock<bool>,106107 108 pub host_config: Option<Value>,109 pub nixos_config: OnceLock<Value>,110 pub nixos_unchecked_config: OnceLock<Value>,111 pub pkgs_override: Option<Value>,112113 binary_cache: OnceCell<Option<BinaryCache>>,114115 116 pub local: bool,117 pub remowt: OnceCell<Remowt>,118 nix_store: OnceCell<Arc<Store>>,119 nix_plugin: OnceCell<()>,120}121122const NIX_PLUGIN_ID: u16 = 2;123124fn agents_dir() -> Result<PathBuf> {125 std::env::var_os("REMOWT_AGENTS_DIR")126 .map(PathBuf::from)127 .or_else(|| option_env!("REMOWT_AGENTS_DIR").map(PathBuf::from))128 .ok_or_else(|| {129 anyhow!("no remowt-agents bundle; set REMOWT_AGENTS_DIR to a remowt-agents output")130 })131}132133fn agent_bundle() -> Result<AgentBundle> {134 AgentBundle::from_dir(agents_dir()?)135}136137#[derive(Clone)]138pub enum BinaryCache {139 Attic { name: String, pin: bool },140}141impl BinaryCache {142 async fn upload(&self, local: &Remowt, path: &Utf8Path, pin_name: Option<&str>) -> Result<()> {143 match self {144 BinaryCache::Attic { name, pin } => {145 let mut cmd = local.cmd("attic");146 cmd.arg("push").arg(name).arg(path.as_str());147 cmd.run()148 .await149 .context("failed to push path to attic (is attic-cli on PATH?)")?;150151 if *pin && let Some(pin_name) = pin_name {152 let mut cmd = local.cmd("attic");153 cmd.arg("pin")154 .arg("create")155 .arg(name)156 .arg(pin_name)157 .arg(path.as_str());158 cmd.run()159 .await160 .context("failed to pin path in attic (is your attic-cli compatible?)")?;161 }162 Ok(())163 }164 }165 }166}167168impl ConfigHost {169 pub fn set_session_destination(&self, dest: String) {170 self.session_destination171 .set(dest)172 .expect("session destination is already set")173 }174 pub fn set_deploy_kind(&self, kind: DeployKind) {175 self.deploy_kind176 .set(kind)177 .expect("deploy kind is already set");178 }179 pub fn set_legacy_ssh_store(&self, legacy: bool) {180 self.legacy_ssh_store181 .set(legacy)182 .expect("legacy ssh store is already set")183 }184 pub fn binary_cache(&self) -> Result<Option<BinaryCache>> {185 if let Some(v) = self.binary_cache.get() {186 return Ok(v.clone());187 }188 let cache = if let Some(host_config) = &self.host_config {189 let cache = nix_go!(host_config.cache);190 let enable: bool = nix_go_json!(cache.enable);191 if enable {192 let kind: String = nix_go_json!(cache.kind);193 let name: String = nix_go_json!(cache.name);194 match kind.as_str() {195 "attic" => {196 let pin: bool = nix_go_json!(cache.pin);197 Some(BinaryCache::Attic { name, pin })198 }199 v => bail!("unknown binary cache kind: {v}"),200 }201 } else {202 None203 }204 } else {205 None206 };207 let _ = self.binary_cache.set(cache.clone());208 Ok(cache)209 }210 pub async fn deploy_kind(&self) -> Result<DeployKind> {211 self.deploy_kind.get_or_try_init(|| async {212 let remowt = self.remowt().await?;213 let fs = remowt.endpoints::<FsClient<_>>();214 let is_fleet_managed = match fs.file_exists(Utf8PathBuf::from("/etc/FLEET_HOST")).await {215 Ok(v) => v,216 Err(e) => {217 bail!("failed to query remote system kind: {e}");218 }219 };220 if !is_fleet_managed {221 bail!(222 "{}",223 indoc::indoc! {"224 host is not marked as managed by fleet225 if you're not trying to lustrate/install system from scratch,226 you should either227 1. manually create /etc/FLEET_HOST file on the target host,228 2. use ?deploy_kind=fleet host argument if you're upgrading from older version of fleet229 3. use ?deploy_kind=upgrade_to_fleet if you're upgrading from plain nixos to fleet-managed nixos230 for installation use ?deploy_kind=nixos_install / ?deploy_kind=nixos_lustrate 231 "}232 );233 }234 Ok(DeployKind::Fleet)235 }).await.copied()236 }237 async fn new_remowt(&self) -> Result<Remowt> {238 let bundle = agent_bundle()?;239 let conn = if self.local {240 Remowt::connect_local(&bundle, "remowt-fleet".to_owned())241 .await242 .context("starting local remowt agent")?243 } else {244 let dest = self245 .session_destination246 .get()247 .cloned()248 .unwrap_or_else(|| self.name.clone());249 Remowt::connect(&dest, &bundle, "remowt-fleet".to_owned())250 .await251 .map_err(|e| anyhow!("remowt error while connecting to {}: {e:#?}", self.name))?252 };253 PromptEndpoints(PrependSourcePrompter {254 prompter: AutoPrompter::new().await,255 source: if self.local {256 vec![]257 } else {258 vec![Source(std::borrow::Cow::Owned(format!(259 "ssh host: {}",260 self.name261 )))]262 },263 description: "".to_owned(),264 })265 .register_endpoints(&mut conn.rpc());266 Ok(conn)267 }268269 270 pub async fn remowt(&self) -> Result<Remowt> {271 self.remowt272 .get_or_try_init(|| self.new_remowt())273 .await274 .cloned()275 }276277 pub async fn nix_client(&self) -> Result<NixClient<BifConfig>> {278 let remowt = self.remowt().await?;279 let plugin_id = self.ensure_nix_plugin().await?;280 Ok(remowt.plugin_endpoints(plugin_id))281 }282 pub fn ensure_nix_plugin(&self) -> Pin<Box<dyn Future<Output = Result<u16>> + Send + '_>> {283 Box::pin(async {284 self.nix_plugin285 .get_or_try_init(|| async {286 let pkgs = self.pkgs()?;287 let name = "remowt-plugin-fleet";288 let plugin = nix_go!(pkgs[{ name }]);289 let built = plugin290 .build("out")291 .context("failed to build the fleet nix plugin")?;292 let pin = format!("{}-remowt-plugin-fleet", self.gc_root_prefix());293 let copied = self294 295 .remote_derivation(&built, Some(&pin))296 .await297 .context("failed to copy the fleet nix plugin to the host store")?;298 let bin = copied.join("bin/remowt-plugin-fleet");299 self.remowt()300 .await?301 .run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())302 .await303 .context("failed to load the fleet nix plugin")?;304 305 self.remowt()306 .await?307 .rpc()308 .wait_for_connection_to(Address::Plugin(NIX_PLUGIN_ID))309 .await310 .map_err(|_| anyhow!("failed to wait"))?;311 anyhow::Ok(())312 })313 .await?;314 Ok(NIX_PLUGIN_ID)315 })316 }317318 async fn new_nix_store(&self) -> Result<Store> {319 let conn = self.remowt().await?;320 let socket = match self.deploy_kind().await? {321 DeployKind::NixosInstall => {322 remowt_fleet::nix_store_socket(conn, "/mnt?require-sigs=false").await?323 }324 _ => remowt_fleet::nix_store_socket(conn, "auto").await?,325 };326 327 let uri = format!("unix://{}", socket.display());328 Store::open(&uri)329 }330 async fn nix_store(&self) -> Result<Arc<Store>> {331 self.nix_store332 .get_or_try_init(async || Ok(Arc::new(self.new_nix_store().await?)))333 .await334 .cloned()335 }336337 pub async fn decrypt(&self, data: SecretData) -> Result<Vec<u8>> {338 ensure!(data.encrypted, "secret is not encrypted");339 let remowt = self.remowt().await?;340 let mut cmd = remowt.cmd("fleet-install-secrets");341 cmd.arg("decrypt").eqarg("--secret", data.to_string());342 let encoded = cmd343 .sudo()344 .run_string()345 .await346 .context("failed to call remote host for decrypt")?;347 let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;348 ensure!(!data.encrypted, "secret came out encrypted");349 Ok(data.data)350 }351 pub async fn reencrypt_distribution(352 &self,353 data: &FleetSecretDistribution,354 targets: BTreeSet<SecretOwner>,355 now: DateTime<Utc>,356 ) -> Result<FleetSecretDistribution> {357 let mut parts = BTreeMap::new();358 for (part_name, part) in &data.secret.parts {359 parts.insert(360 part_name.clone(),361 if part.raw.encrypted {362 FleetSecretPart {363 raw: self.reencrypt(part.raw.clone(), targets.clone()).await?,364 }365 } else {366 part.clone()367 },368 );369 }370 let secret = FleetSecretData {371 created_at: data.secret.created_at,372 expires_at: data.secret.expires_at,373 generation_data: data.secret.generation_data.clone(),374 parts,375 };376 Ok(FleetSecretDistribution::new(targets, secret, now))377 }378 pub async fn reencrypt(379 &self,380 data: SecretData,381 targets: BTreeSet<SecretOwner>,382 ) -> Result<SecretData> {383 let remowt = self.remowt().await?;384 ensure!(data.encrypted, "secret is not encrypted");385 let mut cmd = remowt.cmd("fleet-install-secrets");386 cmd.arg("reencrypt").eqarg("--secret", data.to_string());387 for target in targets {388 let key = self.config.key(&target).await?;389 cmd.eqarg("--targets", key);390 }391 let encoded = cmd392 .sudo()393 .run_string()394 .await395 .context("failed to call remote host for decrypt")?;396 let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;397 ensure!(data.encrypted, "secret came out not encrypted");398 Ok(data)399 }400 401 #[instrument(skip(self))]402 pub async fn remote_derivation(403 &self,404 path: &Utf8Path,405 pin_name: Option<&str>,406 ) -> Result<Utf8PathBuf> {407 let path = path.to_owned();408 if self.local {409 410 return Ok(path);411 }412 let cache = self.binary_cache()?;413 let mut cache = if let Some(cache) = cache {414 let local = self.config.local_host().remowt().await?;415 let path = path.to_owned();416 let pin_name = pin_name.map(str::to_owned);417 Some(tokio::spawn(async move {418 if let Err(e) = cache.upload(&local, &path, pin_name.as_deref()).await {419 warn!("failed to upload closure {path} to cache: {e:?}");420 }421 }))422 } else {423 None424 };425 let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {426 let path = path.clone();427428 Box::pin(async move {429 let local = self.config.local_host();430 let plugin_id = local.ensure_nix_plugin().await?;431 let nix = local432 .remowt()433 .await?434 .plugin_endpoints::<remowt_fleet::NixClient<_>>(plugin_id);435 nix.sign_closure(path, Utf8PathBuf::from("/etc/nix/private-key"))436 .await437 .map_err(|e| anyhow!("{e:?}"))?438 .map_err(|e| anyhow!("{e}"))?;439 Ok(())440 })441 };442 if let Err(e) = sign.await {443 warn!("failed to sign store paths: {e}");444 }445 let store = self.nix_store().await?;446 let eval_store = eval_store();447 {448 let path = path.clone();449 spawn_blocking(move || eval_store.copy_to(&store, path.as_ref()))450 .await451 .expect("copy_to panicked")452 .context("copying closure to remote store")?;453 }454455 if let Some(cache) = cache.take() {456 let _ = cache.await;457 }458459 Ok(path)460 }461462 463 464 465 466 #[instrument(skip(self))]467 pub async fn remote_derivation_attic(468 &self,469 path: &Utf8Path,470 cache_name: &str,471 ) -> Result<Utf8PathBuf> {472 let path = path.to_owned();473 if self.local {474 475 return Ok(path);476 }477478 info!("uploading {path} closure to attic cache {cache_name}");479 let status = Command::new("attic")480 .arg("push")481 .arg(cache_name)482 .arg(path.as_str())483 .status()484 .await485 .context("failed to spawn attic-cli, is it on PATH?")?;486 ensure!(status.success(), "attic push failed: {status}");487488 info!("substituting {path} on the remote host");489 let nix = self.nix_client().await?;490 let substituted = nix491 .substitute(vec![path.clone()])492 .await493 .map_err(|e| anyhow!("{e:?}"))?494 .map_err(|e| anyhow!("{e}"))?;495 ensure!(496 substituted.contains(&path),497 "remote host failed to substitute {path} from attic cache {cache_name}"498 );499 Ok(path)500 }501}502503struct HostSecretDefinition(Value);504505impl ConfigHost {506 pub fn gc_root_prefix(&self) -> String {507 format!("{}-{}", self.config.gc_root_prefix(), self.name)508 }509510 511 512 pub fn tags(&self) -> Result<Vec<String>> {513 if let Some(v) = self.groups.get() {514 return Ok(v.clone());515 }516 let Some(host_config) = &self.host_config else {517 return Ok(vec![]);518 };519 let tags: Vec<String> = nix_go_json!(host_config.tags);520521 let _ = self.groups.set(tags.clone());522523 Ok(tags)524 }525 pub fn nixos_config(&self) -> Result<Value> {526 if let Some(v) = self.nixos_config.get() {527 return Ok(v.clone());528 }529 let Some(host_config) = &self.host_config else {530 bail!("local host has no nixos_config");531 };532 let nixos_config = nix_go!(host_config.nixos.config);533 assert_warn("nixos config evaluation", &nixos_config)?;534535 let _ = self.nixos_config.set(nixos_config.clone());536537 Ok(nixos_config)538 }539 pub fn nixos_unchecked_config(&self) -> Result<Value> {540 if let Some(v) = self.nixos_unchecked_config.get() {541 return Ok(v.clone());542 }543 let Some(host_config) = &self.host_config else {544 bail!("local host has no nixos_config");545 };546 let nixos_config = nix_go!(host_config.nixos_unchecked.config);547548 let _ = self.nixos_unchecked_config.set(nixos_config.clone());549550 Ok(nixos_config)551 }552553 pub fn list_defined_secrets(&self) -> Result<Vec<String>> {554 let nixos = self.nixos_unchecked_config()?;555 let secrets = nix_go!(nixos.secrets);556 secrets.list_fields()557 }558559 560 pub fn pkgs(&self) -> Result<Value> {561 if let Some(value) = &self.pkgs_override {562 return Ok(value.clone());563 }564 let Some(host_config) = &self.host_config else {565 bail!("local host has no host_config");566 };567 568 Ok(nix_go!(host_config.nixos.options._module.args.value.pkgs))569 }570}571572#[derive(Clone)]573pub struct SharedSecretDefinition(Value);574impl SharedSecretDefinition {575 pub fn expected_owners(&self) -> Result<BTreeSet<SecretOwner>> {576 let secret = &self.0;577 Ok(nix_go_json!(secret.expectedOwners))578 }579 pub fn allow_different(&self) -> Result<bool> {580 let secret = &self.0;581 Ok(nix_go_json!(secret.allowDifferent))582 }583 pub fn regenerate_on_owner_added(&self) -> Result<bool> {584 let secret = &self.0;585 Ok(nix_go_json!(secret.regenerateOnOwnerAdded))586 }587 pub fn regenerate_on_owner_removed(&self) -> Result<bool> {588 let secret = &self.0;589 Ok(nix_go_json!(secret.regenerateOnOwnerRemoved))590 }591 pub fn generator(&self) -> Result<Value> {592 let secret = &self.0;593 Ok(nix_go!(secret.generator))594 }595}596597impl Config {598 pub fn tagged_hostnames(&self, tag: &str) -> Result<Vec<String>> {599 let config = &self.config_field;600 let tagged: Vec<String> = nix_go_json!(config.taggedWith[{ tag }]);601 Ok(tagged)602 }603 pub fn expand_owner_set(&self, owners: Vec<String>) -> Result<BTreeSet<String>> {604 let mut out = BTreeSet::new();605 for owner in owners {606 if let Some(tag) = owner.strip_prefix('@') {607 let hosts = self.tagged_hostnames(tag)?;608 out.extend(hosts);609 } else {610 out.insert(owner);611 }612 }613 Ok(out)614 }615 pub fn local_host(&self) -> Arc<ConfigHost> {616 self.local_host617 .get_or_init(|| {618 Arc::new(ConfigHost {619 config: self.clone(),620 name: "<virtual localhost>".to_owned(),621 host_config: None,622 nixos_config: OnceLock::new(),623 nixos_unchecked_config: OnceLock::new(),624 groups: {625 let cell = OnceLock::new();626 let _ = cell.set(vec![]);627 cell628 },629 pkgs_override: Some(self.default_pkgs.clone()),630 binary_cache: OnceCell::new(),631632 local: true,633 remowt: OnceCell::new(),634 nix_store: OnceCell::new(),635 nix_plugin: OnceCell::new(),636 deploy_kind: OnceCell::new(),637 session_destination: OnceLock::new(),638 legacy_ssh_store: OnceLock::new(),639 })640 })641 .clone()642 }643644 pub fn preferred_hosts(645 &self,646 filter: impl Fn(&str) -> bool,647 ) -> Result<impl Iterator<Item = Result<ConfigHost>>> {648 let prefer = self649 .prefer_identities650 .iter()651 .filter_map(|v| v.as_host())652 .collect::<HashSet<_>>();653 let config = &self.config_field;654 let mut names = nix_go!(config.hosts).list_fields()?;655 names.retain(|s| filter(s));656 names.sort_by_key(|h| prefer.contains(h.as_str()));657658 Ok(names.into_iter().map(|h| self.host(&h)))659 }660661 pub fn host(&self, name: &str) -> Result<ConfigHost> {662 let config = &self.config_field;663 let host_config = nix_go!(config.hosts[{ name }]);664665 Ok(ConfigHost {666 config: self.clone(),667 name: name.to_owned(),668 host_config: Some(host_config),669 nixos_config: OnceLock::new(),670 nixos_unchecked_config: OnceLock::new(),671 groups: OnceLock::new(),672 pkgs_override: None,673 binary_cache: OnceCell::new(),674675 676 local: self.localhost == name,677 remowt: OnceCell::new(),678 nix_store: OnceCell::new(),679 nix_plugin: OnceCell::new(),680 deploy_kind: OnceCell::new(),681 session_destination: OnceLock::new(),682 legacy_ssh_store: OnceLock::new(),683 })684 }685 pub fn list_hosts(&self) -> Result<Vec<ConfigHost>> {686 let config = &self.config_field;687 let names = nix_go!(config.hosts).list_fields()?;688 let mut out = vec![];689 for name in names {690 out.push(self.host(&name)?);691 }692 Ok(out)693 }694 695 pub fn system_config(&self, host: &str) -> Result<Value> {696 let fleet_field = &self.config_field;697 Ok(nix_go!(fleet_field.hosts[{ host }].nixos.config))698 }699700 pub fn secret_definition(&self, secret: &str) -> Result<Option<SharedSecretDefinition>> {701 let config = &self.config_field;702 let shared_secrets = nix_go!(config.secrets);703 if !shared_secrets.has_field(secret)? {704 return Ok(None);705 }706 Ok(Some(SharedSecretDefinition(nix_go!(707 shared_secrets[secret]708 ))))709 }710711 pub fn save(&self) -> Result<()> {712 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.")?;713 let data = nixlike::serialize(&*self.data)?;714 tempfile.write_all(715 format!(716 "# This file contains fleet state and shouldn't be edited by hand\n\n{data}\n\n# vim: ts=2 et nowrap\n"717 )718 .as_bytes(),719 )?;720 let mut fleet_data_path = self.directory.clone();721 fleet_data_path.push("fleet.nix");722 tempfile.persist(fleet_data_path)?;723 Ok(())724 }725}