--- a/Cargo.lock +++ b/Cargo.lock @@ -1912,14 +1912,18 @@ version = "0.2.0" dependencies = [ "anyhow", + "base64 0.22.1", "camino", "clap", "clap_complete", "fleet-base", + "fleet-usb", "futures", "goodlog-subscriber", "itertools 0.15.0", + "nix", "nix-eval", + "rand 0.10.1", "remowt-fleet", "serde", "serde_json", @@ -1928,6 +1932,8 @@ "tempfile", "tokio", "tracing", + "zbus", + "zstd", ] [[package]] @@ -2006,6 +2012,47 @@ ] [[package]] +name = "fleet-usb" +version = "0.1.9" +dependencies = [ + "age", + "anyhow", + "base64 0.22.1", + "camino", + "chacha20poly1305", + "hex", + "hmac 0.13.0", + "serde", + "serde_json", + "sha2 0.11.0", +] + +[[package]] +name = "fleet-usbd" +version = "0.1.9" +dependencies = [ + "age", + "anyhow", + "base64 0.22.1", + "bytes", + "camino", + "clap", + "fleet-usb", + "futures", + "goodlog-subscriber", + "hex", + "hostname", + "http-body-util", + "hyper", + "hyper-util", + "nix", + "nix-eval", + "rand 0.10.1", + "tokio", + "tracing", +] + +[[package]] name = "fluent" version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -7841,6 +7888,34 @@ checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" [[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.0.16+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" +dependencies = [ + "cc", + "pkg-config", +] + +[[package]] name = "zvariant" version = "5.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ [workspace.dependencies] fleet-base = { path = "./crates/fleet-base" } fleet-shared = { path = "./crates/fleet-shared" } +fleet-usb = { path = "./crates/fleet-usb" } nix-eval = { path = "./crates/nix-eval" } nixlike = { path = "./crates/nixlike" } opentelemetry-exporter-env = { path = "./crates/opentelemetry-exporter-env" } @@ -36,6 +37,7 @@ bindgen = "0.72.0" bytes = "1.11.0" camino = "1.2.2" +chacha20poly1305 = "0.10" chrono = { version = "0.4.41", features = ["serde"] } clap = { version = "4.5", features = ["derive", "env", "unicode", "wrap_help"] } clap_complete = "4.5" @@ -47,6 +49,9 @@ hex = "0.4.3" hmac = "0.13.0" hostname = "0.4.1" +http-body-util = "0.1" +hyper = { version = "1", features = ["http1", "server"] } +hyper-util = { version = "0.1", features = ["tokio"] } human-repr = "1.1" indicatif = "0.18" indoc = "2.0.6" @@ -85,6 +90,7 @@ unicode_categories = "0.1.1" vte = { version = "0.15.0", features = ["ansi"] } x25519-dalek = { version = "2.0.1", features = ["getrandom"] } +zstd = "0.13" zbus = "5.16.0" zbus_polkit = "5.0.0" goodlog-subscriber = { version = "0.1.9", path = "crates/goodlog-subscriber" } --- a/cmds/fleet/Cargo.toml +++ b/cmds/fleet/Cargo.toml @@ -9,17 +9,23 @@ [dependencies] anyhow.workspace = true +base64.workspace = true camino.workspace = true clap.workspace = true clap_complete.workspace = true fleet-base.workspace = true +fleet-usb.workspace = true +nix.workspace = true nix-eval.workspace = true +rand.workspace = true remowt-fleet.workspace = true serde.workspace = true serde_json.workspace = true tempfile.workspace = true tokio.workspace = true tracing.workspace = true +zbus.workspace = true +zstd.workspace = true futures.workspace = true itertools.workspace = true --- a/cmds/fleet/src/cmds/build_systems.rs +++ b/cmds/fleet/src/cmds/build_systems.rs @@ -33,7 +33,11 @@ build_attr: String, } -async fn build_task(config: Config, hostname: String, build_attr: &str) -> Result { +pub(crate) async fn build_task( + config: Config, + hostname: String, + build_attr: &str, +) -> Result { info!(target: TARGET, "building"); let host = config.host(&hostname)?; // let action = Action::from(self.subcommand.clone()); --- a/cmds/fleet/src/cmds/mod.rs +++ b/cmds/fleet/src/cmds/mod.rs @@ -4,3 +4,4 @@ pub mod rollback; pub mod secrets; pub mod tf; +pub mod usbd; --- /dev/null +++ b/cmds/fleet/src/cmds/usbd.rs @@ -0,0 +1,396 @@ +use std::{ + collections::HashSet, + fs::File, + io::{self, Write as _}, + sync::Arc, +}; + +use anyhow::{Context as _, Result, anyhow, bail}; +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD as BASE64; +use camino::{Utf8Path, Utf8PathBuf}; +use clap::Parser; +use fleet_base::{fleetdata::SecretOwner, host::Config}; +use fleet_usb::{ + MANIFEST_DONE_NAME, MANIFEST_NAME, MANIFEST_VERSION, + manifest::{Manifest, PathEntry, encrypt_manifest}, + names::{self, NAMING_SECRET_SIZE, NamingSecret}, + stream::EncryptWriter, +}; +use nix_eval::{Store, eval_store}; +use rand::Rng as _; +use tokio::task::spawn_blocking; +use tracing::{info, warn}; +use zbus::zvariant::{self, OwnedObjectPath}; + +use super::build_systems::build_task; + +const TARGET: &str = "usbd"; + +const TMP_PREFIX: &str = "usbd-tmp-"; +const MAX_CHUNK_SIZE: u64 = 1 << 30; +const MAX_CHUNKS: u32 = 1024; + +#[derive(Parser)] +pub enum Usbd { + /// Write a system update for a host onto an inserted USB stick + Write(UsbdWrite), +} + +impl Usbd { + pub async fn run(self, config: &Config) -> Result<()> { + match self { + Usbd::Write(w) => w.run(config).await, + } + } +} + +#[derive(Parser)] +pub struct UsbdWrite { + hostname: String, + /// Volume label of the update stick + #[clap(long, default_value = "FLEETUSBD")] + label: String, + /// Write into this directory instead of discovering and mounting the stick via udisks + #[clap(long)] + mount_point: Option, + /// Nix secret signing key; its public counterpart must be in trusted-public-keys on the host + #[clap(long)] + sign_key: Option, + #[clap(long, default_value = "toplevel-fleet")] + build_attr: String, +} + +impl UsbdWrite { + pub async fn run(self, config: &Config) -> Result<()> { + let host = config.host(&self.hostname)?; + let hostname = host.name.clone(); + let recipient = config.recipient(&SecretOwner::host(&hostname)).await?; + let secret = naming_secret(config, &hostname)?; + + let built = build_task(config.clone(), hostname.clone(), &self.build_attr).await?; + + if let Some(key) = self.sign_key.clone() { + let store = eval_store(); + let path = built.clone(); + spawn_blocking(move || store.sign_closure(&path, &key)).await??; + } else { + warn!(target: TARGET, "closure is not signed (no --sign-key), the host will refuse it unless paths are signed by other means"); + } + + let (stick, dir) = match &self.mount_point { + Some(dir) => (None, dir.clone()), + None => { + let stick = Stick::discover(&self.label).await?; + let dir = stick.mount_point.clone(); + (Some(stick), dir) + } + }; + info!(target: TARGET, "writing update to {dir}"); + + let store = eval_store(); + let manifest = { + let secret = secret.clone(); + let hostname = hostname.clone(); + let built = built.clone(); + let dir = dir.clone(); + spawn_blocking(move || write_closure(&store, &secret, &dir, &hostname, &built)) + .await?? + }; + + let encrypted = encrypt_manifest(&manifest, std::iter::once(&*recipient))?; + { + let dir = dir.clone(); + spawn_blocking(move || write_manifest(&dir, &encrypted)).await??; + } + + if let Some(stick) = stick { + stick.unmount().await?; + } + info!(target: TARGET, "update for {hostname} written, the stick can be removed"); + Ok(()) + } +} + +fn naming_secret(config: &Config, host: &str) -> Result { + let mut extra = config.data.extra.write().expect("no poisoning"); + let usbd = extra + .entry("usbd".to_owned()) + .or_insert_with(|| serde_json::json!({})); + let usbd = usbd + .as_object_mut() + .context("fleet.nix extra.usbd should be an object")?; + let secrets = usbd + .entry("namingSecrets") + .or_insert_with(|| serde_json::json!({})); + let secrets = secrets + .as_object_mut() + .context("fleet.nix extra.usbd.namingSecrets should be an object")?; + if let Some(value) = secrets.get(host) { + let value = value + .as_str() + .context("naming secret should be a base64 string")?; + let bytes = BASE64.decode(value).context("naming secret base64")?; + let bytes: [u8; NAMING_SECRET_SIZE] = bytes + .try_into() + .map_err(|_| anyhow!("naming secret should be {NAMING_SECRET_SIZE} bytes"))?; + Ok(NamingSecret(bytes)) + } else { + let mut bytes = [0u8; NAMING_SECRET_SIZE]; + rand::rng().fill_bytes(&mut bytes); + secrets.insert( + host.to_owned(), + serde_json::Value::String(BASE64.encode(bytes)), + ); + Ok(NamingSecret(bytes)) + } +} + +fn write_closure( + store: &Arc, + secret: &NamingSecret, + dir: &Utf8Path, + hostname: &str, + toplevel: &Utf8Path, +) -> Result { + let mut closure = store.compute_closure(toplevel)?; + closure.sort(); + info!(target: TARGET, "update closure contains {} paths", closure.len()); + + let mut existing = HashSet::new(); + for entry in dir.read_dir_utf8().context("reading stick")? { + existing.insert(entry?.file_name().to_owned()); + } + + let total = closure.len(); + let mut entries = Vec::with_capacity(total); + for (i, path) in closure.iter().enumerate() { + let entry = write_path(store, secret, dir, &mut existing, path) + .with_context(|| format!("writing {path}"))?; + info!(target: TARGET, "[{}/{total}] {path}", i + 1); + entries.push(entry); + } + + let referenced = entries + .iter() + .flat_map(|e| e.chunks.iter()) + .map(String::as_str) + .collect::>(); + for name in &existing { + let stale_data = names::is_data_name(name) && !referenced.contains(name.as_str()); + let stale_other = name.starts_with(TMP_PREFIX) || name == MANIFEST_DONE_NAME; + if stale_data || stale_other { + std::fs::remove_file(dir.join(name)) + .with_context(|| format!("deleting stale {name}"))?; + } + } + + Ok(Manifest { + version: MANIFEST_VERSION, + host: hostname.to_owned(), + toplevel: toplevel.to_owned(), + paths: entries, + }) +} + +fn write_path( + store: &Arc, + secret: &NamingSecret, + dir: &Utf8Path, + existing: &mut HashSet, + path: &Utf8Path, +) -> Result { + let info = store.query_path_info(path)?; + let key = names::file_key(secret, &info.nar_hash); + + let mut chunks = None; + for count in 1..=MAX_CHUNKS { + if !existing.contains(&names::chunk_name(secret, &info.nar_hash, 0, count)) { + continue; + } + let all = (0..count) + .map(|i| names::chunk_name(secret, &info.nar_hash, i, count)) + .collect::>(); + if all.iter().all(|c| existing.contains(c)) { + chunks = Some(all); + break; + } + } + + let chunks = if let Some(chunks) = chunks { + chunks + } else { + let chunk_writer = ChunkWriter::new(dir.to_owned()); + let encrypt = EncryptWriter::new(&key, chunk_writer); + let mut compress = + zstd::stream::write::Encoder::new(encrypt, zstd::DEFAULT_COMPRESSION_LEVEL)?; + store.nar_from_path(path, &mut compress)?; + let tmp_files = compress.finish()?.finish()?.finish()?; + let count = u32::try_from(tmp_files.len()).expect("chunk count"); + if count > MAX_CHUNKS { + bail!("{path} does not fit into {MAX_CHUNKS} chunks"); + } + let chunks = (0..count) + .map(|i| names::chunk_name(secret, &info.nar_hash, i, count)) + .collect::>(); + for (tmp, name) in tmp_files.iter().zip(&chunks) { + std::fs::rename(tmp, dir.join(name))?; + } + existing.extend(chunks.iter().cloned()); + chunks + }; + + Ok(PathEntry { + store_path: path.to_owned(), + nar_hash: info.nar_hash, + nar_size: info.nar_size, + references: info.references, + sigs: info.sigs, + compression: "zstd".to_owned(), + key: BASE64.encode(key), + chunks, + }) +} + +fn write_manifest(dir: &Utf8Path, encrypted: &[u8]) -> Result<()> { + let tmp = dir.join(format!("{TMP_PREFIX}manifest")); + let mut file = File::create(&tmp)?; + file.write_all(encrypted)?; + file.sync_all()?; + drop(file); + std::fs::rename(&tmp, dir.join(MANIFEST_NAME))?; + nix::unistd::sync(); + Ok(()) +} + +struct ChunkWriter { + dir: Utf8PathBuf, + files: Vec<(Utf8PathBuf, File)>, + current_len: u64, +} + +impl ChunkWriter { + fn new(dir: Utf8PathBuf) -> Self { + Self { + dir, + files: Vec::new(), + current_len: 0, + } + } + + fn finish(self) -> io::Result> { + for (_, file) in &self.files { + file.sync_all()?; + } + Ok(self.files.into_iter().map(|(path, _)| path).collect()) + } +} + +impl io::Write for ChunkWriter { + fn write(&mut self, data: &[u8]) -> io::Result { + if self.files.is_empty() || self.current_len == MAX_CHUNK_SIZE { + let path = self.dir.join(format!("{TMP_PREFIX}{}", self.files.len())); + let file = File::create(&path)?; + self.files.push((path, file)); + self.current_len = 0; + } + let take = data.len().min((MAX_CHUNK_SIZE - self.current_len) as usize); + let file = &mut self.files.last_mut().expect("just ensured").1; + file.write_all(&data[..take])?; + self.current_len += take as u64; + Ok(take) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +const UDISKS_DEST: &str = "org.freedesktop.UDisks2"; +const IFACE_BLOCK: &str = "org.freedesktop.UDisks2.Block"; +const IFACE_FILESYSTEM: &str = "org.freedesktop.UDisks2.Filesystem"; + +struct Stick { + conn: zbus::Connection, + object: OwnedObjectPath, + mount_point: Utf8PathBuf, + mounted_by_us: bool, +} + +impl Stick { + async fn discover(label: &str) -> Result { + let conn = zbus::Connection::system() + .await + .context("connecting to system dbus for udisks")?; + let om = zbus::fdo::ObjectManagerProxy::builder(&conn) + .destination(UDISKS_DEST)? + .path("/org/freedesktop/UDisks2")? + .build() + .await?; + let objects = om + .get_managed_objects() + .await + .context("is udisks2 running?")?; + + let mut found = Vec::new(); + for (object, ifaces) in objects { + let Some(block) = ifaces.iter().find(|(k, _)| k.as_str() == IFACE_BLOCK) else { + continue; + }; + if !ifaces.iter().any(|(k, _)| k.as_str() == IFACE_FILESYSTEM) { + continue; + } + let Some(id_label) = block.1.get("IdLabel") else { + continue; + }; + let Ok(id_label) = String::try_from(id_label.clone()) else { + continue; + }; + if id_label == label { + found.push(object); + } + } + let object = match found.len() { + 0 => bail!("no inserted filesystem with label {label} found, is the stick inserted?"), + 1 => found.remove(0), + _ => bail!("multiple filesystems with label {label} found: {found:?}"), + }; + + let fs = zbus::Proxy::new(&conn, UDISKS_DEST, &object, IFACE_FILESYSTEM).await?; + let mount_points: Vec> = fs.get_property("MountPoints").await?; + if let Some(mp) = mount_points.first() { + let mp = mp.strip_suffix(&[0]).unwrap_or(mp); + let mount_point = Utf8PathBuf::from(String::from_utf8(mp.to_vec())?); + info!(target: TARGET, "stick is already mounted at {mount_point}"); + return Ok(Self { + conn, + object, + mount_point, + mounted_by_us: false, + }); + } + + let options = std::collections::HashMap::<&str, zvariant::Value>::new(); + let mount_point: String = fs + .call("Mount", &(options,)) + .await + .context("mounting the stick via udisks")?; + Ok(Self { + conn, + object, + mount_point: Utf8PathBuf::from(mount_point), + mounted_by_us: true, + }) + } + + async fn unmount(self) -> Result<()> { + if !self.mounted_by_us { + return Ok(()); + } + let fs = zbus::Proxy::new(&self.conn, UDISKS_DEST, &self.object, IFACE_FILESYSTEM).await?; + let options = std::collections::HashMap::<&str, zvariant::Value>::new(); + fs.call::<_, _, ()>("Unmount", &(options,)) + .await + .context("unmounting the stick")?; + Ok(()) + } +} --- a/cmds/fleet/src/main.rs +++ b/cmds/fleet/src/main.rs @@ -14,6 +14,7 @@ rollback::RollbackSingle, secrets::Secret, tf::Tf, + usbd::Usbd, }; use fleet_base::{host::Config, opts::FleetOpts}; use futures::{TryStreamExt, stream::FuturesUnordered}; @@ -74,6 +75,9 @@ Prefetch(Prefetch), /// Config parsing Info(Info), + /// Airgapped updates over USB stick + #[clap(subcommand)] + Usbd(Usbd), /// Command completions #[clap(hide(true))] Complete(Complete), @@ -100,6 +104,7 @@ Opts::RollbackSingle(r) => r.run(config, &opts).await?, Opts::Secret(s) => s.run(config, &opts).await?, Opts::Info(i) => i.run(config).await?, + Opts::Usbd(u) => u.run(config).await?, Opts::Prefetch(p) => p.run(config).await?, Opts::Tf(t) => t.run(config).await?, // TODO: actually parse commands before starting the async runtime --- /dev/null +++ b/cmds/usbd/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "fleet-usbd" +description = "Airgapped NixOS update daemon, applies fleet updates from a USB stick" +version.workspace = true +edition.workspace = true +rust-version.workspace = true + +[dependencies] +age.workspace = true +anyhow.workspace = true +base64.workspace = true +bytes.workspace = true +camino.workspace = true +clap.workspace = true +fleet-usb.workspace = true +futures.workspace = true +goodlog-subscriber.workspace = true +hostname.workspace = true +hex.workspace = true +http-body-util.workspace = true +hyper.workspace = true +hyper-util.workspace = true +nix = { workspace = true, features = ["mount"] } +nix-eval.workspace = true +rand.workspace = true +tokio = { workspace = true, features = ["net", "process"] } +tracing.workspace = true --- /dev/null +++ b/cmds/usbd/src/main.rs @@ -0,0 +1,279 @@ +mod server; + +use std::{process::ExitCode, sync::Arc}; + +use anyhow::{Context as _, Result, bail}; +use camino::{Utf8Path, Utf8PathBuf}; +use clap::Parser; +use fleet_usb::{MANIFEST_DONE_NAME, MANIFEST_NAME, manifest::decrypt_manifest}; +use goodlog_subscriber::{LogOpts, setup_logging}; +use nix::mount::MsFlags; +use nix_eval::{ + Store, gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix, +}; +use tokio::task::spawn_blocking; +use tracing::{error, info, warn}; + +const MOUNT_TARGET: &str = "/run/fleet-usbd/mnt"; + +#[derive(Parser)] +#[clap(version, author)] +struct Opts { + /// Block device with the update filesystem, e.g. /dev/disk/by-label/FLEETUSBD + #[clap( + long, + required_unless_present = "mount_point", + conflicts_with = "mount_point" + )] + device: Option, + /// Already mounted update directory, instead of --device + #[clap(long)] + mount_point: Option, + /// Host ssh key the manifest is encrypted to + #[clap(long, default_value = "/etc/ssh/ssh_host_ed25519_key")] + host_key: Utf8PathBuf, + /// Expected manifest host name, defaults to the system hostname + #[clap(long)] + hostname: Option, + /// Executed on update lifecycle events with the event name as the first argument: + /// copying, switching, done, noop, failed + #[clap(long, verbatim_doc_comment)] + hook: Option, + #[clap(long, default_value = "/nix/var/nix/profiles/system")] + profile: String, + /// Do not reboot after the update is applied + #[clap(long)] + no_reboot: bool, + #[clap(flatten)] + log: LogOpts, +} + +enum Outcome { + Applied, + UpToDate, + Nothing, +} + +struct Hook(Option); +impl Hook { + async fn call(&self, event: &str, detail: Option<&str>) { + let Some(script) = &self.0 else { + return; + }; + let mut cmd = tokio::process::Command::new(script); + cmd.arg(event); + if let Some(detail) = detail { + cmd.arg(detail); + } + match cmd.status().await { + Ok(status) if status.success() => {} + Ok(status) => warn!("hook {event}: {status}"), + Err(e) => warn!("hook {event}: {e}"), + } + } +} + +fn host_identity(path: &Utf8Path) -> Result { + let data = std::fs::read(path).with_context(|| format!("reading host key {path}"))?; + age::ssh::Identity::from_buffer(std::io::Cursor::new(data), Some(path.to_string())) + .with_context(|| format!("parsing host key {path}")) +} + +fn mount_stick(device: &Utf8Path) -> Result<()> { + std::fs::create_dir_all(MOUNT_TARGET)?; + nix::mount::mount( + Some(device.as_std_path()), + MOUNT_TARGET, + Some("vfat"), + MsFlags::MS_NOSUID | MsFlags::MS_NODEV | MsFlags::MS_NOEXEC, + None::<&str>, + ) + .with_context(|| format!("mounting {device} at {MOUNT_TARGET}"))?; + Ok(()) +} + +fn mark_done(dir: &Utf8Path) -> Result<()> { + std::fs::rename(dir.join(MANIFEST_NAME), dir.join(MANIFEST_DONE_NAME)) + .context("marking update as done")?; + nix::unistd::sync(); + Ok(()) +} + +async fn apply(opts: &Opts, dir: &Utf8Path, hook: &Hook) -> Result { + let manifest_path = dir.join(MANIFEST_NAME); + if !manifest_path.exists() { + if dir.join(MANIFEST_DONE_NAME).exists() { + info!("update on this stick is already applied"); + } else { + info!("no update manifest on the stick"); + } + return Ok(Outcome::Nothing); + } + + let identity = host_identity(&opts.host_key)?; + let data = std::fs::read(&manifest_path).context("reading manifest")?; + let manifest = decrypt_manifest(&data, &identity) + .context("decrypting manifest, is this stick meant for this machine?")?; + + let host = match &opts.hostname { + Some(host) => host.clone(), + None => hostname::get() + .context("querying hostname")? + .to_string_lossy() + .into_owned(), + }; + if manifest.host != host { + bail!( + "update is for host {}, but this host is {host}", + manifest.host + ); + } + + let current = std::fs::read_link("/run/current-system").ok(); + if current.as_deref() == Some(manifest.toplevel.as_std_path()) { + info!("system is already up to date"); + mark_done(dir)?; + return Ok(Outcome::UpToDate); + } + + let missing = manifest + .paths + .iter() + .flat_map(|e| e.chunks.iter()) + .filter(|c| !dir.join(c.as_str()).exists()) + .count(); + if missing > 0 { + bail!("stick is missing {missing} update files, was it synchronized fully?"); + } + + info!( + "applying update to {} ({} paths)", + manifest.toplevel, + manifest.paths.len() + ); + hook.call("copying", None).await; + let cache = server::CacheServer::start(dir.to_owned(), &manifest).await?; + { + let url = cache.url.clone(); + let toplevel = manifest.toplevel.clone(); + spawn_blocking(move || -> Result<()> { + let src = Store::open(&url)?; + let dst = Store::open("auto")?; + src.copy_to(&dst, &toplevel) + }) + .await? + .context("copying closure from the stick")?; + } + drop(cache); + + hook.call("switching", None).await; + { + let profile = opts.profile.clone(); + let toplevel = manifest.toplevel.clone(); + spawn_blocking(move || -> Result<()> { + let store = Store::open("auto")?; + store.switch_profile(&profile, &toplevel) + }) + .await??; + } + let switch = manifest.toplevel.join("bin/switch-to-configuration"); + let status = tokio::process::Command::new(&switch) + .arg("boot") + .status() + .await + .with_context(|| format!("running {switch}"))?; + if !status.success() { + bail!("{switch} boot failed: {status}"); + } + + mark_done(dir)?; + Ok(Outcome::Applied) +} + +async fn run(opts: &Opts) -> Result<()> { + let hook = Hook(opts.hook.clone()); + + let mounted = match &opts.device { + Some(device) => { + mount_stick(device)?; + true + } + None => false, + }; + let dir = opts + .mount_point + .clone() + .unwrap_or_else(|| Utf8PathBuf::from(MOUNT_TARGET)); + + let outcome = apply(opts, &dir, &hook).await; + + nix::unistd::sync(); + if mounted && let Err(e) = nix::mount::umount(MOUNT_TARGET) { + warn!("unmounting the stick: {e}"); + } + + match outcome { + Ok(Outcome::Applied) => { + hook.call("done", None).await; + if opts.no_reboot { + info!("update applied, reboot to activate"); + } else { + info!("update applied, rebooting"); + let status = tokio::process::Command::new("systemctl") + .arg("reboot") + .status() + .await + .context("requesting reboot")?; + if !status.success() { + bail!("systemctl reboot failed: {status}"); + } + } + Ok(()) + } + Ok(Outcome::UpToDate) => { + hook.call("done", None).await; + Ok(()) + } + Ok(Outcome::Nothing) => { + hook.call("noop", None).await; + Ok(()) + } + Err(e) => { + hook.call("failed", Some(&format!("{e:#}"))).await; + Err(e) + } + } +} + +fn main() -> ExitCode { + let opts = Opts::parse(); + if let Err(e) = setup_logging(&opts.log) { + eprintln!("{e:#}"); + return ExitCode::FAILURE; + } + + init_libraries(); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .on_thread_start(|| { + gc_register_my_thread(); + }) + .on_thread_stop(|| { + gc_unregister_my_thread(); + }) + .build() + .expect("failed to build runtime"); + let runtime = Arc::new(runtime); + + init_tokio_for_nix(runtime.clone()); + + runtime.block_on(async { + if let Err(e) = run(&opts).await { + error!("{e:#}"); + ExitCode::FAILURE + } else { + ExitCode::SUCCESS + } + }) +} --- /dev/null +++ b/cmds/usbd/src/server.rs @@ -0,0 +1,218 @@ +use std::{collections::HashMap, convert::Infallible, fs::File, io::Read as _, sync::Arc}; + +use anyhow::{Context as _, Result, anyhow}; +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD as BASE64; +use bytes::Bytes; +use camino::Utf8PathBuf; +use fleet_usb::{manifest::Manifest, manifest::PathEntry, stream::DecryptReader}; +use futures::SinkExt as _; +use http_body_util::{BodyExt as _, Full, StreamBody, combinators::BoxBody}; +use hyper::{ + Method, Request, Response, StatusCode, + body::{Frame, Incoming}, + service::service_fn, +}; +use hyper_util::rt::TokioIo; +use rand::Rng as _; +use tokio::{net::TcpListener, task::spawn_blocking}; +use tracing::debug; + +type Body = BoxBody; + +struct State { + dir: Utf8PathBuf, + prefix: String, + store_dir: String, + entries: HashMap, +} + +pub struct CacheServer { + pub url: String, + accept_loop: tokio::task::JoinHandle<()>, +} + +impl Drop for CacheServer { + fn drop(&mut self) { + self.accept_loop.abort(); + } +} + +impl CacheServer { + pub async fn start(dir: Utf8PathBuf, manifest: &Manifest) -> Result { + let mut token = [0u8; 16]; + rand::rng().fill_bytes(&mut token); + let token = hex::encode(token); + + let store_dir = manifest + .toplevel + .parent() + .context("toplevel should be a store path")? + .to_string(); + let mut entries = HashMap::new(); + for entry in &manifest.paths { + let base = entry + .store_path + .file_name() + .context("store path should have a file name")?; + let hash = base + .split('-') + .next() + .expect("split returns at least one part"); + entries.insert(hash.to_owned(), entry.clone()); + } + + let listener = TcpListener::bind(("127.0.0.1", 0)) + .await + .context("binding local cache server")?; + let port = listener.local_addr()?.port(); + let state = Arc::new(State { + dir, + prefix: format!("/{token}/"), + store_dir, + entries, + }); + let accept_loop = tokio::spawn(accept_loop(listener, state)); + Ok(Self { + url: format!("http://127.0.0.1:{port}/{token}"), + accept_loop, + }) + } +} + +async fn accept_loop(listener: TcpListener, state: Arc) { + loop { + let stream = match listener.accept().await { + Ok((stream, _)) => stream, + Err(e) => { + debug!("cache server accept: {e}"); + continue; + } + }; + let state = state.clone(); + tokio::spawn(async move { + let service = service_fn(move |req| handle(state.clone(), req)); + if let Err(e) = hyper::server::conn::http1::Builder::new() + .serve_connection(TokioIo::new(stream), service) + .await + { + debug!("cache server connection: {e}"); + } + }); + } +} + +async fn handle(state: Arc, req: Request) -> Result, Infallible> { + let head = req.method() == Method::HEAD; + if !head && req.method() != Method::GET { + return Ok(status(StatusCode::METHOD_NOT_ALLOWED)); + } + let mut resp = route(&state, req.uri().path()).unwrap_or_else(|e| { + debug!("cache server request {} failed: {e}", req.uri().path()); + status(StatusCode::INTERNAL_SERVER_ERROR) + }); + if head { + *resp.body_mut() = empty(); + } + Ok(resp) +} + +fn route(state: &Arc, path: &str) -> Result> { + let Some(rest) = path.strip_prefix(&state.prefix) else { + return Ok(status(StatusCode::NOT_FOUND)); + }; + if rest == "nix-cache-info" { + return Ok(text(format!( + "StoreDir: {}\nWantMassQuery: 1\nPriority: 30\n", + state.store_dir + ))); + } + if let Some(hash) = rest.strip_suffix(".narinfo") { + let Some(entry) = state.entries.get(hash) else { + return Ok(status(StatusCode::NOT_FOUND)); + }; + return Ok(text(narinfo(hash, entry))); + } + if let Some(hash) = rest.strip_prefix("nar/") { + let Some(entry) = state.entries.get(hash) else { + return Ok(status(StatusCode::NOT_FOUND)); + }; + return nar(state, entry); + } + Ok(status(StatusCode::NOT_FOUND)) +} + +fn narinfo(hash: &str, entry: &PathEntry) -> String { + let references = entry + .references + .iter() + .filter_map(|r| r.file_name()) + .collect::>() + .join(" "); + let mut out = format!( + "StorePath: {}\nURL: nar/{hash}\nCompression: {}\nNarHash: {}\nNarSize: {}\nReferences: {references}\n", + entry.store_path, entry.compression, entry.nar_hash, entry.nar_size, + ); + for sig in &entry.sigs { + out.push_str(&format!("Sig: {sig}\n")); + } + out +} + +fn nar(state: &Arc, entry: &PathEntry) -> Result> { + let key: [u8; 32] = BASE64 + .decode(&entry.key)? + .try_into() + .map_err(|_| anyhow!("invalid key size in manifest"))?; + let files = entry + .chunks + .iter() + .map(|c| state.dir.join(c)) + .collect::>(); + + let (mut tx, rx) = futures::channel::mpsc::channel::, std::io::Error>>(8); + spawn_blocking(move || { + let mut stream = || -> std::io::Result<()> { + let mut chunks: Box = Box::new(std::io::empty()); + for file in &files { + chunks = Box::new(chunks.chain(File::open(file)?)); + } + let mut reader = DecryptReader::new(&key, chunks); + let mut buf = vec![0u8; 64 * 1024]; + loop { + let n = reader.read(&mut buf)?; + if n == 0 { + return Ok(()); + } + let frame = Frame::data(Bytes::copy_from_slice(&buf[..n])); + if futures::executor::block_on(tx.send(Ok(frame))).is_err() { + return Ok(()); + } + } + }; + if let Err(e) = stream() { + let _ = futures::executor::block_on(tx.send(Err(e))); + } + }); + Ok(Response::new(StreamBody::new(rx).boxed())) +} + +fn empty() -> Body { + Full::new(Bytes::new()) + .map_err(|never| match never {}) + .boxed() +} + +fn text(s: String) -> Response { + Response::new( + Full::new(Bytes::from(s)) + .map_err(|never| match never {}) + .boxed(), + ) +} + +fn status(code: StatusCode) -> Response { + let mut resp = Response::new(empty()); + *resp.status_mut() = code; + resp +} --- /dev/null +++ b/crates/fleet-usb/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "fleet-usb" +version.workspace = true +edition.workspace = true +rust-version.workspace = true + +[dependencies] +age.workspace = true +anyhow.workspace = true +base64.workspace = true +camino = { workspace = true, features = ["serde1"] } +chacha20poly1305.workspace = true +hex.workspace = true +hmac.workspace = true +serde.workspace = true +serde_json.workspace = true +sha2.workspace = true --- /dev/null +++ b/crates/fleet-usb/src/lib.rs @@ -0,0 +1,7 @@ +pub mod manifest; +pub mod names; +pub mod stream; + +pub const MANIFEST_NAME: &str = "usbd-update"; +pub const MANIFEST_DONE_NAME: &str = "usbd-update.done"; +pub const MANIFEST_VERSION: u32 = 1; --- /dev/null +++ b/crates/fleet-usb/src/manifest.rs @@ -0,0 +1,62 @@ +use std::io::{Read as _, Write as _}; + +use age::{Identity, Recipient}; +use anyhow::{Context as _, Result, anyhow, bail}; +use camino::Utf8PathBuf; +use serde::{Deserialize, Serialize}; + +use crate::MANIFEST_VERSION; + +#[derive(Serialize, Deserialize, Debug, Clone)] +#[serde(rename_all = "camelCase")] +pub struct Manifest { + pub version: u32, + pub host: String, + pub toplevel: Utf8PathBuf, + pub paths: Vec, +} + +#[derive(Serialize, Deserialize, Debug, Clone)] +#[serde(rename_all = "camelCase")] +pub struct PathEntry { + pub store_path: Utf8PathBuf, + pub nar_hash: String, + pub nar_size: u64, + pub references: Vec, + pub sigs: Vec, + pub compression: String, + pub key: String, + pub chunks: Vec, +} + +pub fn encrypt_manifest<'r>( + manifest: &Manifest, + recipients: impl Iterator, +) -> Result> { + let json = serde_json::to_vec(manifest)?; + let mut out = Vec::new(); + let mut encryptor = age::Encryptor::with_recipients(recipients) + .map_err(|e| anyhow!("no recipients: {e}"))? + .wrap_output(&mut out) + .context("age encrypt")?; + encryptor.write_all(&json)?; + encryptor.finish()?; + Ok(out) +} + +pub fn decrypt_manifest(data: &[u8], identity: &dyn Identity) -> Result { + let decryptor = age::Decryptor::new(data).context("age decrypt")?; + let mut reader = decryptor + .decrypt(std::iter::once(identity)) + .context("manifest is not encrypted for this identity")?; + let mut json = Vec::new(); + reader.read_to_end(&mut json)?; + let manifest: Manifest = serde_json::from_slice(&json).context("manifest json")?; + if manifest.version != MANIFEST_VERSION { + bail!( + "unsupported manifest version {}, this fleet-usbd supports {MANIFEST_VERSION}", + manifest.version + ); + } + Ok(manifest) +} --- /dev/null +++ b/crates/fleet-usb/src/names.rs @@ -0,0 +1,42 @@ +use hmac::{Hmac, KeyInit as _, Mac as _}; +use sha2::Sha256; + +type HmacSha256 = Hmac; + +pub const NAMING_SECRET_SIZE: usize = 32; + +#[derive(Clone)] +pub struct NamingSecret(pub [u8; NAMING_SECRET_SIZE]); + +fn derive(secret: &NamingSecret, parts: &[&[u8]]) -> [u8; 32] { + let mut mac = HmacSha256::new_from_slice(&secret.0).expect("any key size is valid"); + for part in parts { + mac.update(&u64::to_be_bytes(part.len() as u64)); + mac.update(part); + } + mac.finalize().into_bytes().into() +} + +pub fn file_key(secret: &NamingSecret, nar_hash: &str) -> [u8; 32] { + derive(secret, &[b"fleet-usb.v1.key\0", nar_hash.as_bytes()]) +} + +pub fn chunk_name(secret: &NamingSecret, nar_hash: &str, index: u32, count: u32) -> String { + let h = derive( + secret, + &[ + b"fleet-usb.v1.name\0", + nar_hash.as_bytes(), + &index.to_le_bytes(), + &count.to_le_bytes(), + ], + ); + hex::encode(&h[..16]) +} + +pub fn is_data_name(name: &str) -> bool { + name.len() == 32 + && name + .bytes() + .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) +} --- /dev/null +++ b/crates/fleet-usb/src/stream.rs @@ -0,0 +1,179 @@ +use std::io::{self, Read, Write}; + +use chacha20poly1305::{KeyInit as _, XChaCha20Poly1305, XNonce, aead::Aead as _}; + +pub const FRAME_SIZE: usize = 256 * 1024; +pub const TAG_SIZE: usize = 16; + +fn nonce(counter: u64, last: bool) -> XNonce { + let mut n = [0u8; 24]; + n[..8].copy_from_slice(&counter.to_le_bytes()); + n[23] = last as u8; + n.into() +} + +pub struct EncryptWriter { + inner: W, + aead: XChaCha20Poly1305, + buf: Vec, + counter: u64, +} + +impl EncryptWriter { + pub fn new(key: &[u8; 32], inner: W) -> Self { + Self { + inner, + aead: XChaCha20Poly1305::new(key.into()), + buf: Vec::with_capacity(FRAME_SIZE), + counter: 0, + } + } + + fn emit(&mut self, last: bool) -> io::Result<()> { + let ct = self + .aead + .encrypt(&nonce(self.counter, last), self.buf.as_slice()) + .map_err(|_| io::Error::other("aead encrypt failed"))?; + self.inner.write_all(&ct)?; + self.counter += 1; + self.buf.clear(); + Ok(()) + } + + pub fn finish(mut self) -> io::Result { + self.emit(true)?; + self.inner.flush()?; + Ok(self.inner) + } +} + +impl Write for EncryptWriter { + fn write(&mut self, data: &[u8]) -> io::Result { + let take = data.len().min(FRAME_SIZE - self.buf.len()); + self.buf.extend_from_slice(&data[..take]); + if self.buf.len() == FRAME_SIZE { + self.emit(false)?; + } + Ok(take) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +pub struct DecryptReader { + inner: R, + aead: XChaCha20Poly1305, + counter: u64, + plain: Vec, + plain_pos: usize, + lookahead: Option, + done: bool, +} + +impl DecryptReader { + pub fn new(key: &[u8; 32], inner: R) -> Self { + Self { + inner, + aead: XChaCha20Poly1305::new(key.into()), + counter: 0, + plain: Vec::new(), + plain_pos: 0, + lookahead: None, + done: false, + } + } + + fn read_frame(&mut self) -> io::Result<()> { + let mut frame = Vec::with_capacity(FRAME_SIZE + TAG_SIZE + 1); + if let Some(b) = self.lookahead.take() { + frame.push(b); + } + (&mut self.inner) + .take((FRAME_SIZE + TAG_SIZE + 1 - frame.len()) as u64) + .read_to_end(&mut frame)?; + let last = if frame.len() == FRAME_SIZE + TAG_SIZE + 1 { + self.lookahead = frame.pop(); + false + } else { + true + }; + if frame.len() < TAG_SIZE { + return Err(io::Error::other("truncated stream: incomplete frame")); + } + self.plain = self + .aead + .decrypt(&nonce(self.counter, last), frame.as_slice()) + .map_err(|_| io::Error::other("aead decrypt failed: corrupted or truncated stream"))?; + self.counter += 1; + self.plain_pos = 0; + self.done = last; + Ok(()) + } +} + +impl Read for DecryptReader { + fn read(&mut self, out: &mut [u8]) -> io::Result { + while self.plain_pos == self.plain.len() { + if self.done { + return Ok(0); + } + self.read_frame()?; + } + let take = out.len().min(self.plain.len() - self.plain_pos); + out[..take].copy_from_slice(&self.plain[self.plain_pos..self.plain_pos + take]); + self.plain_pos += take; + Ok(take) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn roundtrip(len: usize) { + let key = [7u8; 32]; + let data = (0..len).map(|i| (i * 31 % 256) as u8).collect::>(); + let mut enc = EncryptWriter::new(&key, Vec::new()); + enc.write_all(&data).unwrap(); + let ct = enc.finish().unwrap(); + let mut out = Vec::new(); + DecryptReader::new(&key, ct.as_slice()) + .read_to_end(&mut out) + .unwrap(); + assert_eq!(out, data); + } + + #[test] + fn roundtrips() { + roundtrip(0); + roundtrip(1); + roundtrip(FRAME_SIZE - 1); + roundtrip(FRAME_SIZE); + roundtrip(FRAME_SIZE + 1); + roundtrip(FRAME_SIZE * 3 + 17); + } + + #[test] + fn truncation_detected() { + let key = [7u8; 32]; + let mut enc = EncryptWriter::new(&key, Vec::new()); + enc.write_all(&vec![1u8; FRAME_SIZE * 2]).unwrap(); + let ct = enc.finish().unwrap(); + let mut out = Vec::new(); + let res = DecryptReader::new(&key, &ct[..ct.len() - TAG_SIZE - 1]).read_to_end(&mut out); + assert!(res.is_err()); + } + + #[test] + fn deterministic() { + let key = [7u8; 32]; + let data = vec![3u8; FRAME_SIZE + 100]; + let ct = |d: &[u8]| { + let mut enc = EncryptWriter::new(&key, Vec::new()); + enc.write_all(d).unwrap(); + enc.finish().unwrap() + }; + assert_eq!(ct(&data), ct(&data)); + } +} --- a/crates/nix-eval/src/lib.cc +++ b/crates/nix-eval/src/lib.cc @@ -14,6 +14,7 @@ #include #include #include +#include #include #include #include @@ -184,6 +185,70 @@ } } +CxxPathInfo query_path_info(Store *store, rust::Str path) { + CxxPathInfo out{rust::String(), rust::String(), 0, {}, {}}; + try { + auto nixStore = store->ptr; + auto sp = nixStore->parseStorePath(std::string(path)); + auto info = nixStore->queryPathInfo(sp); + out.nar_hash = + rust::String(info->narHash.to_string(nix::HashFormat::Nix32, true)); + out.nar_size = info->narSize; + for (auto &r : info->references) + out.references.push_back(rust::String(nixStore->printStorePath(r))); + for (auto &s : info->sigs) + out.sigs.push_back(rust::String(s.to_string())); + } catch (const std::exception &e) { + out.error = rust::String(e.what()); + } + return out; +} + +CxxBuildResult compute_closure(Store *store, rust::Str path) { + CxxBuildResult res{rust::String(), {}}; + try { + auto nixStore = store->ptr; + auto root = nixStore->parseStorePath(std::string(path)); + nix::StorePathSet closure; + nixStore->computeFSClosure(root, closure); + for (auto &p : closure) + res.outputs.push_back(rust::String(nixStore->printStorePath(p))); + } catch (const std::exception &e) { + res.error = rust::String(e.what()); + } + return res; +} + +namespace { +struct RustFnSink : nix::Sink { + size_t data; + rust::Fn)> sink; + RustFnSink(size_t data, + rust::Fn)> sink) + : data(data), sink(sink) {} + void operator()(std::string_view chunk) override { + bool ok = sink(data, rust::Slice( + reinterpret_cast(chunk.data()), + chunk.size())); + if (!ok) + throw nix::Error("nar sink aborted"); + } +}; +} // namespace + +rust::String nar_from_path(Store *store, rust::Str path, size_t sink_data, + rust::Fn)> sink) { + try { + auto nixStore = store->ptr; + auto sp = nixStore->parseStorePath(std::string(path)); + RustFnSink s(sink_data, sink); + nixStore->narFromPath(sp, s); + return rust::String(); + } catch (const std::exception &e) { + return rust::String(e.what()); + } +} + AddFileToStoreResult add_file_to_store(Store *store, rust::Str name, rust::Str path) { AddFileToStoreResult out{rust::String(), rust::String(), rust::String()}; --- a/crates/nix-eval/src/lib.hh +++ b/crates/nix-eval/src/lib.hh @@ -30,3 +30,11 @@ CxxBuildResult substitute_paths(Store *store, rust::Str paths_joined); bool is_valid_path(Store *store, rust::Str path); + +struct CxxPathInfo; +CxxPathInfo query_path_info(Store *store, rust::Str path); + +CxxBuildResult compute_closure(Store *store, rust::Str path); + +rust::String nar_from_path(Store *store, rust::Str path, size_t sink_data, + rust::Fn)> sink); --- a/crates/nix-eval/src/lib.rs +++ b/crates/nix-eval/src/lib.rs @@ -90,6 +90,13 @@ error: String, outputs: Vec, } + struct CxxPathInfo { + error: String, + nar_hash: String, + nar_size: u64, + references: Vec, + sigs: Vec, + } unsafe extern "C++" { type nix_fetchers_settings; type Store; @@ -129,6 +136,20 @@ #[allow(clippy::missing_safety_doc)] unsafe fn is_valid_path(store: *mut Store, path: &str) -> bool; + + #[allow(clippy::missing_safety_doc)] + unsafe fn query_path_info(store: *mut Store, path: &str) -> CxxPathInfo; + + #[allow(clippy::missing_safety_doc)] + unsafe fn compute_closure(store: *mut Store, path: &str) -> CxxBuildResult; + + #[allow(clippy::missing_safety_doc)] + unsafe fn nar_from_path( + store: *mut Store, + path: &str, + sink_data: usize, + sink: fn(usize, &[u8]) -> bool, + ) -> String; } } @@ -535,6 +556,14 @@ unsafe impl Send for Store {} unsafe impl Sync for Store {} +#[derive(Debug, Clone)] +pub struct PathInfo { + pub nar_hash: String, + pub nar_size: u64, + pub references: Vec, + pub sigs: Vec, +} + pub fn eval_store() -> Arc { GLOBAL_STATE.store.clone() } @@ -623,6 +652,63 @@ } #[instrument(skip(self))] + pub fn query_path_info(&self, path: &Utf8Path) -> Result { + let res = unsafe { nix_cxx::query_path_info(self.as_ptr().cast(), path.as_str()) }; + if !res.error.is_empty() { + bail!("failed to query path info for {path}: {}", res.error); + } + Ok(PathInfo { + nar_hash: res.nar_hash, + nar_size: res.nar_size, + references: res.references.into_iter().map(Utf8PathBuf::from).collect(), + sigs: res.sigs, + }) + } + + #[instrument(skip(self))] + pub fn compute_closure(&self, path: &Utf8Path) -> Result> { + let res = unsafe { nix_cxx::compute_closure(self.as_ptr().cast(), path.as_str()) }; + if !res.error.is_empty() { + bail!("failed to compute closure of {path}: {}", res.error); + } + Ok(res.outputs.into_iter().map(Utf8PathBuf::from).collect()) + } + + #[instrument(skip(self, out))] + pub fn nar_from_path(&self, path: &Utf8Path, out: &mut dyn std::io::Write) -> Result<()> { + struct SinkState<'a> { + out: &'a mut dyn std::io::Write, + error: Option, + } + fn sink(state: usize, data: &[u8]) -> bool { + let state = unsafe { &mut *(state as *mut SinkState) }; + match state.out.write_all(data) { + Ok(()) => true, + Err(e) => { + state.error = Some(e); + false + } + } + } + let mut state = SinkState { out, error: None }; + let msg = unsafe { + nix_cxx::nar_from_path( + self.as_ptr().cast(), + path.as_str(), + (&raw mut state) as usize, + sink, + ) + }; + if let Some(e) = state.error { + return Err(anyhow!(e).context(format!("nar sink failed for {path}"))); + } + if !msg.is_empty() { + bail!("failed to dump nar of {path}: {msg}"); + } + Ok(()) + } + + #[instrument(skip(self))] pub fn build_drv_outputs( &self, drv_path: &Utf8Path, --- /dev/null +++ b/docs/features/usbd.adoc @@ -0,0 +1,83 @@ +== Fleet-usbd + +For airgapped deployments/system updates for low-tech users with limited internet access, it is better to have something more manual than pusher. + +fleet-usbd is a simple update daemon, which watches for a USB stick with the system update to be inserted into the machine, reads the manifest (usbd-update) from it, copies the closure from the USB stick, switches the current system generation, moves the manifest (usbd-update.done) and reboots. + +The stick is discovered by volume label; the label is only a discovery hint, not a trust anchor. Detection is a systemd device unit on /dev/disk/by-label/