git.delta.rocks / fleet / refs/commits / 210037fd620b

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_link_shared::Address;20use remowt_ui_prompt::auto::AutoPrompter;21use remowt_ui_prompt::bifrost::PromptEndpoints;22use remowt_ui_prompt::{PrependSourcePrompter, Source};23use tabled::Tabled;24use tempfile::NamedTempFile;25use time::UtcDateTime;26use tokio::task::spawn_blocking;27use tracing::{info, warn};2829use crate::fleetdata::{30	FleetData, FleetSecretData, FleetSecretDistribution, FleetSecretPart, SecretOwner,31};3233pub struct FleetConfigInternals {34	pub prefer_identities: BTreeSet<SecretOwner>,35	pub now: DateTime<Utc>,3637	/// Fleet project directory, containing fleet.nix file.38	pub directory: PathBuf,39	/// builtins.currentSystem40	pub local_system: String,41	pub data: Arc<FleetData>,42	/// fleet_config.config43	pub config_field: Value,44	/// flake.output45	pub flake_outputs: Value,46	// TODO: Remove with connectivity refactor47	pub localhost: String,4849	/// import nixpkgs {system = local};50	pub default_pkgs: Value,51	/// inputs.nixpkgs52	pub nixpkgs: Value,5354	pub local_host: OnceLock<Arc<ConfigHost>>,55}5657// TODO: Make field not pub58#[derive(Clone)]59pub struct Config(pub Arc<FleetConfigInternals>);6061impl Deref for Config {62	type Target = FleetConfigInternals;6364	fn deref(&self) -> &Self::Target {65		&self.066	}67}6869#[derive(Clone, PartialEq, Copy, Debug)]70pub enum DeployKind {71	/// NixOS => NixOS managed by fleet72	UpgradeToFleet,73	/// NixOS managed by fleet => NixOS managed by fleet74	Fleet,75	/// Remote host has /mnt, /mnt/boot mounted,76	/// generated config is added to fleet configuration.77	NixosInstall,78	/// Remote host has some system and nix installed in multi-user mode (/nix is owned by root),79	/// generated config is added to fleet configuration,80	/// and /etc/NIXOS_LUSTRATE exists, fleet will perform the rest.81	NixosLustrate,82}8384impl FromStr for DeployKind {85	type Err = anyhow::Error;86	fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {87		match s {88			"upgrade-to-fleet" => Ok(Self::UpgradeToFleet),89			"fleet" => Ok(Self::Fleet),90			"nixos-install" => Ok(Self::NixosInstall),91			"nixos-lustrate" => Ok(Self::NixosLustrate),92			v => bail!(93				"unknown deploy_kind: {v}; expected on of \"upgrade-to-fleet\", \"fleet\", \"nixos-install\", \"nixos-lustrate\""94			),95		}96	}97}98pub struct ConfigHost {99	config: Config,100	pub name: String,101	groups: OnceLock<Vec<String>>,102103	// TODO: Both of those values are taken from host opts, there should be a cleaner way to specify it104	deploy_kind: OnceLock<DeployKind>,105	session_destination: OnceLock<String>,106	legacy_ssh_store: OnceLock<bool>,107108	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	// TODO: Move command helpers away with connectivity refactor114	pub local: bool,115	pub remowt: OnceLock<Remowt>,116	nix_store: OnceLock<Arc<Store>>,117	nix_plugin: tokio::sync::OnceCell<()>,118}119120const NIX_PLUGIN_ID: u16 = 2;121122fn agents_dir() -> Result<PathBuf> {123	std::env::var_os("REMOWT_AGENTS_DIR")124		.map(PathBuf::from)125		.or_else(|| option_env!("REMOWT_AGENTS_DIR").map(PathBuf::from))126		.ok_or_else(|| {127			anyhow!("no remowt-agents bundle; set REMOWT_AGENTS_DIR to a remowt-agents output")128		})129}130131fn agent_bundle() -> Result<AgentBundle> {132	AgentBundle::from_dir(agents_dir()?)133}134135#[derive(Debug, Clone, Copy)]136pub enum GenerationStorage {137	Deployer,138	Machine,139	Pusher,140}141impl GenerationStorage {142	fn prefix(&self) -> &'static str {143		match self {144			GenerationStorage::Deployer => "deployer.",145			GenerationStorage::Machine => "",146			GenerationStorage::Pusher => "pusher.",147		}148	}149}150151#[derive(Tabled, Debug)]152pub struct Generation {153	#[tabled(rename = "ID", format("{}", self.rollback_id()))]154	pub id: u32,155	#[tabled(rename = "Current")]156	pub current: bool,157	#[tabled(rename = "Created at")]158	pub datetime: UtcDateTime,159	#[tabled(format = "{:?}")]160	pub store_path: Utf8PathBuf,161	#[tabled(skip)]162	pub location: GenerationStorage,163}164impl Generation {165	pub fn rollback_id(&self) -> String {166		format!("{}{}", self.location.prefix(), self.id)167	}168}169170impl ConfigHost {171	pub async fn list_generations(&self, profile: &str) -> Result<Vec<Generation>> {172		let plugin_id = self.ensure_nix_plugin().await?;173		let nix = self174			.remowt()175			.await?176			.plugin_endpoints::<remowt_fleet::NixClient<_>>(plugin_id);177		let raw = nix178			.list_generations(profile.to_owned())179			.await180			.map_err(|e| anyhow!("{e:?}"))?181			.map_err(|e| anyhow!("{e}"))?;182		raw.into_iter()183			.map(|g| {184				let id: u32 =185					g.id.try_into()186						.with_context(|| format!("generation id {} doesn't fit in u32", g.id))?;187				let datetime = UtcDateTime::from_unix_timestamp(g.creation_time_unix)188					.with_context(|| {189						format!("invalid generation timestamp {}", g.creation_time_unix)190					})?;191				Ok(Generation {192					id,193					current: g.current,194					datetime,195					store_path: g.store_path,196					location: GenerationStorage::Machine,197				})198			})199			.collect()200	}201202	pub fn set_session_destination(&self, dest: String) {203		self.session_destination204			.set(dest)205			.expect("session destination is already set")206	}207	pub fn set_deploy_kind(&self, kind: DeployKind) {208		self.deploy_kind209			.set(kind)210			.expect("deploy kind is already set");211	}212	pub fn set_legacy_ssh_store(&self, legacy: bool) {213		self.legacy_ssh_store214			.set(legacy)215			.expect("legacy ssh store is already set")216	}217	pub async fn deploy_kind(&self) -> Result<DeployKind> {218		if let Some(kind) = self.deploy_kind.get() {219			return Ok(*kind);220		}221		let remowt = self.remowt().await?;222		let fs = remowt.endpoints::<FsClient<_>>();223		let is_fleet_managed = match fs.file_exists(Utf8PathBuf::from("/etc/FLEET_HOST")).await {224			Ok(v) => v,225			Err(e) => {226				bail!("failed to query remote system kind: {e}");227			}228		};229		if !is_fleet_managed {230			bail!(231				"{}",232				indoc::indoc! {"233				host is not marked as managed by fleet234				if you're not trying to lustrate/install system from scratch,235				you should either236					1. manually create /etc/FLEET_HOST file on the target host,237					2. use ?deploy_kind=fleet host argument if you're upgrading from older version of fleet238					3. use ?deploy_kind=upgrade_to_fleet if you're upgrading from plain nixos to fleet-managed nixos239				for installation use ?deploy_kind=nixos_install / ?deploy_kind=nixos_lustrate 240			"}241			);242		}243		// TOCTOU is possible244		let _ = self.deploy_kind.set(DeployKind::Fleet);245		Ok(*self.deploy_kind.get().expect("deploy kind is just set"))246	}247	async fn connection(&self) -> Result<Remowt> {248		if let Some(conn) = self.remowt.get() {249			return Ok(conn.clone());250		}251		let bundle = agent_bundle()?;252		let conn = if self.local {253			Remowt::connect_local(&bundle)254				.await255				.context("starting local remowt agent")?256		} else {257			let dest = self258				.session_destination259				.get()260				.cloned()261				.unwrap_or_else(|| self.name.clone());262			Remowt::connect(&dest, &bundle)263				.await264				.map_err(|e| anyhow!("remowt error while connecting to {}: {e:#?}", self.name))?265		};266		PromptEndpoints(PrependSourcePrompter {267			prompter: AutoPrompter::new().await,268			source: if self.local {269				vec![]270			} else {271				vec![Source(std::borrow::Cow::Owned(format!(272					"ssh host: {}",273					self.name274				)))]275			},276			description: "".to_owned(),277		})278		.register_endpoints(&mut conn.rpc());279		let _ = self.remowt.set(conn);280		Ok(self.remowt.get().expect("just set").clone())281	}282283	/// Client for this host's unprivileged agent.284	pub async fn remowt(&self) -> Result<Remowt> {285		Ok(self.connection().await?)286	}287288	pub fn ensure_nix_plugin(&self) -> Pin<Box<dyn Future<Output = Result<u16>> + Send + '_>> {289		Box::pin(async {290			self.nix_plugin291				.get_or_try_init(|| async {292					let pkgs = self.pkgs()?;293					let name = "remowt-plugin-fleet";294					let plugin = nix_go!(pkgs[{ name }]);295					let built = plugin296						.build("out")297						.context("failed to build the fleet nix plugin")?;298					let copied = self299						.remote_derivation(&built)300						.await301						.context("failed to copy the fleet nix plugin to the host store")?;302					let bin = copied.join("bin/remowt-plugin-fleet");303					self.remowt()304						.await?305						.run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())306						.await307						.context("failed to load the fleet nix plugin")?;308					self.remowt()309						.await?310						.rpc()311						.wait_for_connection_to(Address::Plugin(NIX_PLUGIN_ID))312						.await313						.map_err(|e| anyhow!("failed to wait for plugin"))?;314					anyhow::Ok(())315				})316				.await?;317			Ok(NIX_PLUGIN_ID)318		})319	}320321	async fn nix_store(&self) -> Result<Arc<Store>> {322		if let Some(store) = self.nix_store.get() {323			return Ok(store.clone());324		}325		let conn = self.connection().await?;326		let socket = match self.deploy_kind().await? {327			DeployKind::NixosInstall => {328				remowt_fleet::nix_store_socket(conn, "/mnt?require-sigs=false").await?329			}330			_ => remowt_fleet::nix_store_socket(conn, "auto").await?,331		};332		let uri = format!("unix://{}", socket.display());333		let store = Arc::new(Store::open(&uri)?);334		let _ = self.nix_store.set(store);335		Ok(self.nix_store.get().expect("just set").clone())336	}337338	pub async fn decrypt(&self, data: SecretData) -> Result<Vec<u8>> {339		ensure!(data.encrypted, "secret is not encrypted");340		let remowt = self.remowt().await?;341		let mut cmd = remowt.cmd("fleet-install-secrets");342		cmd.arg("decrypt").eqarg("--secret", data.to_string());343		let encoded = cmd344			.sudo()345			.run_string()346			.await347			.context("failed to call remote host for decrypt")?;348		let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;349		ensure!(!data.encrypted, "secret came out encrypted");350		Ok(data.data)351	}352	pub async fn reencrypt_distribution(353		&self,354		data: &FleetSecretDistribution,355		targets: BTreeSet<SecretOwner>,356		now: DateTime<Utc>,357	) -> Result<FleetSecretDistribution> {358		let mut parts = BTreeMap::new();359		for (part_name, part) in &data.secret.parts {360			parts.insert(361				part_name.clone(),362				if part.raw.encrypted {363					FleetSecretPart {364						raw: self.reencrypt(part.raw.clone(), targets.clone()).await?,365					}366				} else {367					part.clone()368				},369			);370		}371		let secret = FleetSecretData {372			created_at: data.secret.created_at,373			expires_at: data.secret.expires_at,374			generation_data: data.secret.generation_data.clone(),375			parts,376		};377		Ok(FleetSecretDistribution::new(targets, secret, now))378	}379	pub async fn reencrypt(380		&self,381		data: SecretData,382		targets: BTreeSet<SecretOwner>,383	) -> Result<SecretData> {384		let remowt = self.remowt().await?;385		ensure!(data.encrypted, "secret is not encrypted");386		let mut cmd = remowt.cmd("fleet-install-secrets");387		cmd.arg("reencrypt").eqarg("--secret", data.to_string());388		for target in targets {389			let key = self.config.key(&target).await?;390			cmd.eqarg("--targets", key);391		}392		let encoded = cmd393			.sudo()394			.run_string()395			.await396			.context("failed to call remote host for decrypt")?;397		let data: SecretData = encoded.parse().map_err(|e| anyhow!("{e}"))?;398		ensure!(data.encrypted, "secret came out not encrypted");399		Ok(data)400	}401	/// Returns path for futureproofing, as path might change i.e on conversion to CA402	pub async fn remote_derivation(&self, path: impl AsRef<Utf8Path>) -> Result<Utf8PathBuf> {403		let path = path.as_ref().to_owned();404		if self.local {405			// Path is located locally, thus already trusted.406			return Ok(path);407		}408		let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {409			let path = path.clone();410			Box::pin(async move {411				let local = self.config.local_host();412				let plugin_id = local.ensure_nix_plugin().await?;413				let nix = local414					.remowt()415					.await?416					.plugin_endpoints::<remowt_fleet::NixClient<_>>(plugin_id);417				nix.sign_closure(path, Utf8PathBuf::from("/etc/nix/private-key"))418					.await419					.map_err(|e| anyhow!("{e:?}"))?420					.map_err(|e| anyhow!("{e}"))?;421				Ok(())422			})423		};424		if let Err(e) = sign.await {425			warn!("failed to sign store paths: {e}");426		}427		let store = self.nix_store().await?;428		{429			let path = path.clone();430			let store = eval_store();431			spawn_blocking(move || store.copy_to(&store, path.as_ref()))432				.await433				.expect("copy_to panicked")434				.context("copying closure to remote store")?;435		}436		Ok(path)437	}438}439440struct HostSecretDefinition(Value);441442impl ConfigHost {443	// TOCTOU is possible here in case if config is changed, but this case is not handled anywhere anyway,444	// assuming getting tags always returns the same value.445	pub fn tags(&self) -> Result<Vec<String>> {446		if let Some(v) = self.groups.get() {447			return Ok(v.clone());448		}449		let Some(host_config) = &self.host_config else {450			return Ok(vec![]);451		};452		let tags: Vec<String> = nix_go_json!(host_config.tags);453454		let _ = self.groups.set(tags.clone());455456		Ok(tags)457	}458	pub fn nixos_config(&self) -> Result<Value> {459		if let Some(v) = self.nixos_config.get() {460			return Ok(v.clone());461		}462		let Some(host_config) = &self.host_config else {463			bail!("local host has no nixos_config");464		};465		let nixos_config = nix_go!(host_config.nixos.config);466		assert_warn("nixos config evaluation", &nixos_config)?;467468		let _ = self.nixos_config.set(nixos_config.clone());469470		Ok(nixos_config)471	}472	pub fn nixos_unchecked_config(&self) -> Result<Value> {473		if let Some(v) = self.nixos_unchecked_config.get() {474			return Ok(v.clone());475		}476		let Some(host_config) = &self.host_config else {477			bail!("local host has no nixos_config");478		};479		let nixos_config = nix_go!(host_config.nixos_unchecked.config);480481		let _ = self.nixos_unchecked_config.set(nixos_config.clone());482483		Ok(nixos_config)484	}485486	pub fn list_defined_secrets(&self) -> Result<Vec<String>> {487		let nixos = self.nixos_unchecked_config()?;488		let secrets = nix_go!(nixos.secrets);489		secrets.list_fields()490	}491492	/// Packages for this host, resolved with nixpkgs overlays493	pub fn pkgs(&self) -> Result<Value> {494		if let Some(value) = &self.pkgs_override {495			return Ok(value.clone());496		}497		let Some(host_config) = &self.host_config else {498			bail!("local host has no host_config");499		};500		// TODO: Should nixos.options be cached?501		Ok(nix_go!(host_config.nixos.options._module.args.value.pkgs))502	}503}504505#[derive(Clone)]506pub struct SharedSecretDefinition(Value);507impl SharedSecretDefinition {508	pub fn expected_owners(&self) -> Result<BTreeSet<SecretOwner>> {509		let secret = &self.0;510		Ok(nix_go_json!(secret.expectedOwners))511	}512	pub fn allow_different(&self) -> Result<bool> {513		let secret = &self.0;514		Ok(nix_go_json!(secret.allowDifferent))515	}516	pub fn regenerate_on_owner_added(&self) -> Result<bool> {517		let secret = &self.0;518		Ok(nix_go_json!(secret.regenerateOnOwnerAdded))519	}520	pub fn regenerate_on_owner_removed(&self) -> Result<bool> {521		let secret = &self.0;522		Ok(nix_go_json!(secret.regenerateOnOwnerRemoved))523	}524	pub fn generator(&self) -> Result<Value> {525		let secret = &self.0;526		Ok(nix_go!(secret.generator))527	}528}529530impl Config {531	pub fn tagged_hostnames(&self, tag: &str) -> Result<Vec<String>> {532		let config = &self.config_field;533		let tagged: Vec<String> = nix_go_json!(config.taggedWith[{ tag }]);534		Ok(tagged)535	}536	pub fn expand_owner_set(&self, owners: Vec<String>) -> Result<BTreeSet<String>> {537		let mut out = BTreeSet::new();538		for owner in owners {539			if let Some(tag) = owner.strip_prefix('@') {540				let hosts = self.tagged_hostnames(tag)?;541				out.extend(hosts);542			} else {543				out.insert(owner);544			}545		}546		Ok(out)547	}548	pub fn local_host(&self) -> Arc<ConfigHost> {549		self.local_host550			.get_or_init(|| {551				Arc::new(ConfigHost {552					config: self.clone(),553					name: "<virtual localhost>".to_owned(),554					host_config: None,555					nixos_config: OnceLock::new(),556					nixos_unchecked_config: OnceLock::new(),557					groups: {558						let cell = OnceLock::new();559						let _ = cell.set(vec![]);560						cell561					},562					pkgs_override: Some(self.default_pkgs.clone()),563564					local: true,565					remowt: OnceLock::new(),566					nix_store: OnceLock::new(),567					nix_plugin: tokio::sync::OnceCell::new(),568					deploy_kind: OnceLock::new(),569					session_destination: OnceLock::new(),570					legacy_ssh_store: OnceLock::new(),571				})572			})573			.clone()574	}575576	pub fn preferred_hosts(577		&self,578		filter: impl Fn(&str) -> bool,579	) -> Result<impl Iterator<Item = Result<ConfigHost>>> {580		let prefer = self581			.prefer_identities582			.iter()583			.filter_map(|v| v.as_host())584			.collect::<HashSet<_>>();585		let config = &self.config_field;586		let mut names = nix_go!(config.hosts).list_fields()?;587		names.retain(|s| filter(s));588		names.sort_by_key(|h| prefer.contains(h.as_str()));589590		Ok(names.into_iter().map(|h| self.host(&h)))591	}592593	pub fn host(&self, name: &str) -> Result<ConfigHost> {594		let config = &self.config_field;595		let host_config = nix_go!(config.hosts[{ name }]);596597		Ok(ConfigHost {598			config: self.clone(),599			name: name.to_owned(),600			host_config: Some(host_config),601			nixos_config: OnceLock::new(),602			nixos_unchecked_config: OnceLock::new(),603			groups: OnceLock::new(),604			pkgs_override: None,605606			// TODO: Remove with connectivit refactor607			local: self.localhost == name,608			remowt: OnceLock::new(),609			nix_store: OnceLock::new(),610			nix_plugin: tokio::sync::OnceCell::new(),611			deploy_kind: OnceLock::new(),612			session_destination: OnceLock::new(),613			legacy_ssh_store: OnceLock::new(),614		})615	}616	pub fn list_hosts(&self) -> Result<Vec<ConfigHost>> {617		let config = &self.config_field;618		let names = nix_go!(config.hosts).list_fields()?;619		let mut out = vec![];620		for name in names {621			out.push(self.host(&name)?);622		}623		Ok(out)624	}625	// TODO: Replace usages with .host().nixos_config626	pub fn system_config(&self, host: &str) -> Result<Value> {627		let fleet_field = &self.config_field;628		Ok(nix_go!(fleet_field.hosts[{ host }].nixos.config))629	}630631	pub fn secret_definition(&self, secret: &str) -> Result<Option<SharedSecretDefinition>> {632		let config = &self.config_field;633		let shared_secrets = nix_go!(config.secrets);634		if !shared_secrets.has_field(secret)? {635			return Ok(None);636		}637		Ok(Some(SharedSecretDefinition(nix_go!(638			shared_secrets[secret]639		))))640	}641642	pub fn save(&self) -> Result<()> {643		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.")?;644		let data = nixlike::serialize(&*self.data)?;645		tempfile.write_all(646			format!(647				"# This file contains fleet state and shouldn't be edited by hand\n\n{data}\n\n# vim: ts=2 et nowrap\n"648			)649			.as_bytes(),650		)?;651		let mut fleet_data_path = self.directory.clone();652		fleet_data_path.push("fleet.nix");653		tempfile.persist(fleet_data_path)?;654		Ok(())655	}656}