git.delta.rocks / fleet / refs/commits / 615754ca0747

--- /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<Utf8PathBuf>,
+	/// Nix secret signing key; its public counterpart must be in trusted-public-keys on the host
+	#[clap(long)]
+	sign_key: Option<Utf8PathBuf>,
+	#[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<NamingSecret> {
+	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<Store>,
+	secret: &NamingSecret,
+	dir: &Utf8Path,
+	hostname: &str,
+	toplevel: &Utf8Path,
+) -> Result<Manifest> {
+	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::<HashSet<_>>();
+	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<Store>,
+	secret: &NamingSecret,
+	dir: &Utf8Path,
+	existing: &mut HashSet<String>,
+	path: &Utf8Path,
+) -> Result<PathEntry> {
+	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::<Vec<_>>();
+		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::<Vec<_>>();
+		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<Vec<Utf8PathBuf>> {
+		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<usize> {
+		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<Self> {
+		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<Vec<u8>> = 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(())
+	}
+}

--- /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<Utf8PathBuf>,
+	/// Already mounted update directory, instead of --device
+	#[clap(long)]
+	mount_point: Option<Utf8PathBuf>,
+	/// 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<String>,
+	/// Executed on update lifecycle events with the event name as the first argument:
+	/// copying, switching, done, noop, failed <error>
+	#[clap(long, verbatim_doc_comment)]
+	hook: Option<Utf8PathBuf>,
+	#[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<Utf8PathBuf>);
+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<age::ssh::Identity> {
+	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<Outcome> {
+	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<Bytes, std::io::Error>;
+
+struct State {
+	dir: Utf8PathBuf,
+	prefix: String,
+	store_dir: String,
+	entries: HashMap<String, PathEntry>,
+}
+
+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<Self> {
+		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<State>) {
+	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<State>, req: Request<Incoming>) -> Result<Response<Body>, 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<State>, path: &str) -> Result<Response<Body>> {
+	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::<Vec<_>>()
+		.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<State>, entry: &PathEntry) -> Result<Response<Body>> {
+	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::<Vec<_>>();
+
+	let (mut tx, rx) = futures::channel::mpsc::channel::<Result<Frame<Bytes>, std::io::Error>>(8);
+	spawn_blocking(move || {
+		let mut stream = || -> std::io::Result<()> {
+			let mut chunks: Box<dyn std::io::Read> = 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<Body> {
+	Response::new(
+		Full::new(Bytes::from(s))
+			.map_err(|never| match never {})
+			.boxed(),
+	)
+}
+
+fn status(code: StatusCode) -> Response<Body> {
+	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<PathEntry>,
+}
+
+#[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<Utf8PathBuf>,
+	pub sigs: Vec<String>,
+	pub compression: String,
+	pub key: String,
+	pub chunks: Vec<String>,
+}
+
+pub fn encrypt_manifest<'r>(
+	manifest: &Manifest,
+	recipients: impl Iterator<Item = &'r dyn Recipient>,
+) -> Result<Vec<u8>> {
+	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<Manifest> {
+	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<Sha256>;
+
+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<W: Write> {
+	inner: W,
+	aead: XChaCha20Poly1305,
+	buf: Vec<u8>,
+	counter: u64,
+}
+
+impl<W: Write> EncryptWriter<W> {
+	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<W> {
+		self.emit(true)?;
+		self.inner.flush()?;
+		Ok(self.inner)
+	}
+}
+
+impl<W: Write> Write for EncryptWriter<W> {
+	fn write(&mut self, data: &[u8]) -> io::Result<usize> {
+		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<R: Read> {
+	inner: R,
+	aead: XChaCha20Poly1305,
+	counter: u64,
+	plain: Vec<u8>,
+	plain_pos: usize,
+	lookahead: Option<u8>,
+	done: bool,
+}
+
+impl<R: Read> DecryptReader<R> {
+	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<R: Read> Read for DecryptReader<R> {
+	fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
+		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::<Vec<_>>();
+		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));
+	}
+}

no changes

--- /dev/null
+++ b/modules/nixos/usbd.nix
@@ -0,0 +1,89 @@
+# Tied to cmds/usbd, see docs/features/usbd.adoc
+{
+  config,
+  lib,
+  pkgs,
+  utils,
+  ...
+}:
+let
+  inherit (lib)
+    mkEnableOption
+    mkIf
+    mkOption
+    mkPackageOption
+    types
+    ;
+  inherit (lib.lists) optionals;
+  cfg = config.services.fleet-usbd;
+  devicePath = "/dev/disk/by-label/${cfg.label}";
+  deviceUnit = "${utils.escapeSystemdPath devicePath}.device";
+in
+{
+  options.services.fleet-usbd = {
+    enable = mkEnableOption "fleet USB update daemon";
+    package = mkPackageOption pkgs "fleet-usbd" { };
+    label = mkOption {
+      type = types.str;
+      default = "FLEETUSBD";
+      description = ''
+        Volume label of the update stick, matching the label passed to
+        `fleet usbd write`. FAT32 labels are at most 11 characters, uppercase.
+      '';
+    };
+    signPublicKeys = mkOption {
+      type = types.listOf types.str;
+      description = ''
+        Nix public keys the update closure is signed with
+        (the counterparts of `fleet usbd write --sign-key`).
+      '';
+    };
+    hook = mkOption {
+      type = types.nullOr types.path;
+      default = null;
+      description = ''
+        User feedback script, executed on update lifecycle events with the
+        event name as the first argument: copying, switching, done, noop,
+        failed <error>.
+      '';
+    };
+    extraArgs = mkOption {
+      type = types.listOf types.str;
+      default = [ ];
+      description = "Extra arguments for the fleet-usbd invocation.";
+    };
+  };
+
+  config = mkIf cfg.enable {
+    nix.settings.trusted-public-keys = cfg.signPublicKeys;
+
+    systemd.services.fleet-usbd = {
+      description = "Fleet USB update daemon";
+      wantedBy = [ deviceUnit ];
+      after = [
+        deviceUnit
+        "local-fs.target"
+      ];
+      serviceConfig = {
+        Type = "oneshot";
+        ExecStartPre = "-${pkgs.dosfstools}/bin/fsck.vfat -a ${devicePath}";
+        ExecStart = utils.escapeSystemdExecArgs (
+          [
+            (lib.getExe cfg.package)
+            "--device"
+            devicePath
+          ]
+          ++ optionals (cfg.hook != null) [
+            "--hook"
+            cfg.hook
+          ]
+          ++ cfg.extraArgs
+        );
+      };
+      unitConfig = {
+        X-RestartIfChanged = false;
+        X-StopIfChanged = false;
+      };
+    };
+  };
+}

--- /dev/null
+++ b/pkgs/fleet-usbd.nix
@@ -0,0 +1,38 @@
+{
+  lib,
+  craneLib,
+  inputs,
+
+  stdenv,
+  pkg-config,
+  rustPlatform,
+}:
+let
+  system = stdenv.hostPlatform.system;
+in
+craneLib.buildPackage rec {
+  pname = "fleet-usbd";
+  src = lib.cleanSourceWith {
+    src = ../.;
+    filter =
+      path: type:
+      (lib.hasSuffix "\.cc" path)
+      || (lib.hasSuffix "\.hh" path)
+      || (craneLib.filterCargoSources path type);
+  };
+  strictDeps = true;
+
+  cargoExtraArgs = "--locked -p ${pname}";
+
+  buildInputs = [
+    inputs.nix.packages.${system}.nix-expr-c
+    inputs.nix.packages.${system}.nix-flake-c
+    inputs.nix.packages.${system}.nix-fetchers-c
+  ];
+  nativeBuildInputs = [
+    pkg-config
+    rustPlatform.bindgenHook
+  ];
+
+  meta.mainProgram = "fleet-usbd";
+}