git.delta.rocks / fleet / refs/heads / push-kyumtlkprzyo

1mod server;23use std::{process::ExitCode, sync::Arc};45use anyhow::{Context as _, Result, bail};6use camino::{Utf8Path, Utf8PathBuf};7use clap::Parser;8use fleet_usb::{9	DATA_DIR, LOGS_DIR, MANIFEST_DONE_SUFFIX, MANIFEST_PREFIX, manifest::decrypt_manifest, names,10};11use goodlog_subscriber::{LogOpts, setup_logging};12use nix::mount::MsFlags;13use nix_eval::{14	Store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,15};16use rand::RngExt as _;17use tokio::task::spawn_blocking;18use tracing::{debug, error, info, warn};1920const MOUNT_TARGET: &str = "/run/fleet-usbd/mnt";2122#[derive(Parser)]23#[clap(version, author)]24struct Opts {25	/// Block device with the update filesystem, e.g. /dev/disk/by-label/FLEETUSBD26	#[clap(27		long,28		required_unless_present = "mount_point",29		conflicts_with = "mount_point"30	)]31	device: Option<Utf8PathBuf>,32	/// Already mounted update directory, instead of --device33	#[clap(long)]34	mount_point: Option<Utf8PathBuf>,35	/// Host ssh key the manifest is encrypted to36	#[clap(long, default_value = "/etc/ssh/ssh_host_ed25519_key")]37	host_key: Utf8PathBuf,38	/// Expected manifest host name, defaults to the system hostname39	#[clap(long)]40	hostname: Option<String>,41	/// Executed on update lifecycle events with the event name as the first argument:42	/// copying, switching, done, noop, failed <error>43	#[clap(long, verbatim_doc_comment)]44	hook: Option<Utf8PathBuf>,45	/// Blink connected Nitrokey 3 devices while the daemon is working:46	/// blinking stopped and the machine rebooted - update succeeded,47	/// blinking stopped without a reboot - update failed48	#[clap(long, verbatim_doc_comment)]49	nk3_wink: bool,50	/// After every run, dump the journal onto the stick, encrypted to this armored OpenPGP public key51	#[clap(long)]52	log_recipient_file: Option<Utf8PathBuf>,53	/// journalctl --since window of the log dump54	#[clap(long, default_value = "-48h")]55	log_since: String,56	#[clap(long, default_value = "/nix/var/nix/profiles/system")]57	profile: String,58	/// Do not reboot after the update is applied59	#[clap(long)]60	no_reboot: bool,61	#[clap(flatten)]62	log: LogOpts,63}6465enum Outcome {66	Applied,67	UpToDate,68	Nothing,69}7071struct Hook(Option<Utf8PathBuf>);72impl Hook {73	async fn call(&self, event: &str, detail: Option<&str>) {74		let Some(script) = &self.0 else {75			return;76		};77		let mut cmd = tokio::process::Command::new(script);78		cmd.arg(event);79		if let Some(detail) = detail {80			cmd.arg(detail);81		}82		match cmd.status().await {83			Ok(status) if status.success() => {}84			Ok(status) => warn!("hook {event}: {status}"),85			Err(e) => warn!("hook {event}: {e}"),86		}87	}88}8990fn host_identity(path: &Utf8Path) -> Result<age::ssh::Identity> {91	let data = std::fs::read(path).with_context(|| format!("reading host key {path}"))?;92	age::ssh::Identity::from_buffer(std::io::Cursor::new(data), Some(path.to_string()))93		.with_context(|| format!("parsing host key {path}"))94}9596fn mount_stick(device: &Utf8Path) -> Result<()> {97	std::fs::create_dir_all(MOUNT_TARGET)?;98	nix::mount::mount(99		Some(device.as_std_path()),100		MOUNT_TARGET,101		Some("vfat"),102		MsFlags::MS_NOSUID | MsFlags::MS_NODEV | MsFlags::MS_NOEXEC,103		None::<&str>,104	)105	.with_context(|| format!("mounting {device} at {MOUNT_TARGET}"))?;106	Ok(())107}108109fn mark_done(manifest_path: &Utf8Path) -> Result<()> {110	std::fs::rename(111		manifest_path,112		format!("{manifest_path}{MANIFEST_DONE_SUFFIX}"),113	)114	.context("marking update as done")?;115	nix::unistd::sync();116	Ok(())117}118119fn dump_logs(dir: &Utf8Path, recipient: &Utf8Path, since: &str) -> Result<()> {120	use std::process::{Command, Stdio};121	let logs_dir = dir.join(LOGS_DIR);122	std::fs::create_dir_all(&logs_dir)?;123	let tmp = logs_dir.join("usbd-tmp-log");124	let _ = std::fs::remove_file(&tmp);125126	let _ = Command::new("journalctl").arg("--sync").status();127128	let gpg_home = tempfile::tempdir().context("creating gpg home")?;129	let mut journalctl = Command::new("journalctl")130		.args(["-o", "export"])131		.arg(format!("--since={since}"))132		.stdout(Stdio::piped())133		.spawn()134		.context("spawning journalctl")?;135	let gpg = Command::new("gpg")136		.env("GNUPGHOME", gpg_home.path())137		.args([138			"--batch",139			"--no-tty",140			"--yes",141			"--trust-model",142			"always",143			"--encrypt",144			"--recipient-file",145		])146		.arg(recipient)147		.arg("--output")148		.arg(&tmp)149		.stdin(journalctl.stdout.take().expect("stdout is piped"))150		.status()151		.context("running gpg")?;152	let journalctl = journalctl.wait()?;153	if !journalctl.success() {154		bail!("journalctl failed: {journalctl}");155	}156	if !gpg.success() {157		bail!("gpg failed: {gpg}");158	}159160	let file = std::fs::File::open(&tmp)?;161	file.sync_all()?;162	drop(file);163	let name = format!(164		"{}-{:04x}.export.gpg",165		chrono::Local::now().format("%Y-%m-%d-%H-%M-%S"),166		rand::rng().random::<u16>(),167	);168	std::fs::rename(&tmp, logs_dir.join(&name))?;169	info!("journal dumped to {LOGS_DIR}/{name}");170	Ok(())171}172173async fn apply(opts: &Opts, dir: &Utf8Path, hook: &Hook) -> Result<Outcome> {174	let identity = host_identity(&opts.host_key)?;175176	let mut manifests = Vec::new();177	let mut done_files = Vec::new();178	for entry in dir.read_dir_utf8().context("reading stick")? {179		let entry = entry?;180		let name = entry.file_name();181		if !name.starts_with(MANIFEST_PREFIX) {182			continue;183		}184		if name.ends_with(MANIFEST_DONE_SUFFIX) {185			done_files.push(entry.path().to_owned());186		} else {187			manifests.push(entry.path().to_owned());188		}189	}190191	let mut found = None;192	for path in manifests {193		let data = std::fs::read(&path).with_context(|| format!("reading manifest {path}"))?;194		match decrypt_manifest(&data, &identity).with_context(|| format!("manifest {path}"))? {195			Some(manifest) => {196				found = Some((path, manifest));197				break;198			}199			None => debug!("manifest {path} is for a different host"),200		}201	}202	let Some((manifest_path, manifest)) = found else {203		for path in done_files {204			let data = std::fs::read(&path)?;205			if decrypt_manifest(&data, &identity)?.is_some() {206				info!("update on this stick is already applied");207				return Ok(Outcome::Nothing);208			}209		}210		info!("no update for this host on the stick");211		return Ok(Outcome::Nothing);212	};213214	let host = match &opts.hostname {215		Some(host) => host.clone(),216		None => hostname::get()217			.context("querying hostname")?218			.to_string_lossy()219			.into_owned(),220	};221	if manifest.host != host {222		bail!(223			"update is for host {}, but this host is {host}",224			manifest.host225		);226	}227228	let current = std::fs::read_link("/run/current-system").ok();229	if current.as_deref() == Some(manifest.toplevel.as_std_path()) {230		info!("system is already up to date");231		mark_done(&manifest_path)?;232		return Ok(Outcome::UpToDate);233	}234235	let data_dir = dir.join(DATA_DIR);236	let missing = manifest237		.paths238		.iter()239		.flat_map(|e| e.chunks.iter())240		.filter(|c| !data_dir.join(names::data_rel_path(c)).exists())241		.count();242	if missing > 0 {243		bail!("stick is missing {missing} update files, was it synchronized fully?");244	}245246	info!(247		"applying update to {} ({} paths)",248		manifest.toplevel,249		manifest.paths.len()250	);251	hook.call("copying", None).await;252	let cache = server::CacheServer::start(dir, &manifest).await?;253	{254		let url = cache.url.clone();255		let toplevel = manifest.toplevel.clone();256		let paths = manifest257			.paths258			.iter()259			.map(|e| e.store_path.clone())260			.collect::<Vec<_>>();261		spawn_blocking(move || -> Result<()> {262			let src = Store::open(&url)?;263			let dst = Store::open("auto")?;264			for path in &paths {265				dst.add_temp_root(path)?;266			}267			src.copy_to(&dst, &toplevel)268		})269		.await?270		.context("copying closure from the stick")?;271	}272	drop(cache);273274	hook.call("switching", None).await;275	{276		let profile = opts.profile.clone();277		let toplevel = manifest.toplevel.clone();278		spawn_blocking(move || -> Result<()> {279			let store = Store::open("auto")?;280			store.switch_profile(&profile, &toplevel)281		})282		.await??;283	}284	let switch = manifest.toplevel.join("bin/switch-to-configuration");285	let status = tokio::process::Command::new(&switch)286		.arg("boot")287		.status()288		.await289		.with_context(|| format!("running {switch}"))?;290	if !status.success() {291		bail!("{switch} boot failed: {status}");292	}293294	mark_done(&manifest_path)?;295	Ok(Outcome::Applied)296}297298fn wink_all() -> Result<()> {299	let hidapi = hidapi::HidApi::new().context("initializing hidapi")?;300	for info in hidapi.device_list() {301		let nk3 = info.vendor_id() == 0x20a0 && info.product_id() == 0x42b2;302		let fido = info.usage_page() == 0xf1d0;303		if !nk3 && !fido {304			continue;305		}306		let winked = info307			.open_device(&hidapi)308			.map_err(anyhow::Error::from)309			.and_then(|device| Ok(ctaphid::Device::new(device, info.clone())?))310			.and_then(|device| Ok(device.wink()?));311		if let Err(e) = winked {312			debug!("winking {:?}: {e:#}", info.path());313		}314	}315	Ok(())316}317318async fn run(opts: &Opts) -> Result<()> {319	let hook = Hook(opts.hook.clone());320	let winker = opts.nk3_wink.then(|| {321		tokio::spawn(async {322			loop {323				match spawn_blocking(wink_all).await {324					Ok(Ok(())) => {}325					Ok(Err(e)) => debug!("nk3 wink: {e:#}"),326					Err(e) => debug!("nk3 wink task: {e}"),327				}328				tokio::time::sleep(std::time::Duration::from_secs(5)).await;329			}330		})331	});332333	let mounted = match &opts.device {334		Some(device) => {335			mount_stick(device)?;336			true337		}338		None => false,339	};340	let dir = opts341		.mount_point342		.clone()343		.unwrap_or_else(|| Utf8PathBuf::from(MOUNT_TARGET));344345	let outcome = apply(opts, &dir, &hook).await;346347	if let Err(e) = &outcome {348		error!("update failed: {e:#}");349	}350	if let Some(recipient) = &opts.log_recipient_file {351		let recipient = recipient.clone();352		let dir = dir.clone();353		let since = opts.log_since.clone();354		let dumped = spawn_blocking(move || dump_logs(&dir, &recipient, &since))355			.await356			.expect("dump_logs should not panic");357		if let Err(e) = dumped {358			warn!("failed to dump logs onto the stick: {e:#}");359		}360	}361362	nix::unistd::sync();363	if mounted && let Err(e) = nix::mount::umount(MOUNT_TARGET) {364		warn!("unmounting the stick: {e}");365	}366	if let Some(winker) = winker {367		winker.abort();368	}369370	match outcome {371		Ok(Outcome::Applied) => {372			hook.call("done", None).await;373			if opts.no_reboot {374				info!("update applied, reboot to activate");375			} else {376				info!("update applied, rebooting");377				let status = tokio::process::Command::new("systemctl")378					.arg("reboot")379					.status()380					.await381					.context("requesting reboot")?;382				if !status.success() {383					bail!("systemctl reboot failed: {status}");384				}385			}386			Ok(())387		}388		Ok(Outcome::UpToDate) => {389			hook.call("done", None).await;390			Ok(())391		}392		Ok(Outcome::Nothing) => {393			hook.call("noop", None).await;394			Ok(())395		}396		Err(e) => {397			hook.call("failed", Some(&format!("{e:#}"))).await;398			Err(e)399		}400	}401}402403fn main() -> ExitCode {404	let opts = Opts::parse();405	if let Err(e) = setup_logging(&opts.log) {406		eprintln!("{e:#}");407		return ExitCode::FAILURE;408	}409410	init_libraries();411412	let runtime = tokio::runtime::Builder::new_multi_thread()413		.enable_all()414		.on_thread_start(|| {415			gc_register_my_thread();416		})417		.on_thread_stop(|| {418			gc_unregister_my_thread();419		})420		.build()421		.expect("failed to build runtime");422	let runtime = Arc::new(runtime);423424	init_tokio_for_nix(runtime.clone());425426	runtime.block_on(async {427		if let Err(e) = run(&opts).await {428			error!("{e:#}");429			ExitCode::FAILURE430		} else {431			ExitCode::SUCCESS432		}433	})434}