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/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));
+	}
+}

--- /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/<label> with the daemon service bound to it (WantedBy=), no udisks dependency.
+
+== Update flow
+
+1. Stick inserted, device unit starts the service, stick is mounted.
+2. Daemon decrypts the manifest, checks the system name matches this host.
+3. If the manifest's system is already the current generation - noop. Guards against reinsertion when the .done rename previously failed.
+4. Closure is copied from the stick. Hash verification is part of the copy; copying already-present paths is a noop.
+5. New generation is set as the boot default (boot, not switch), so the existing on-boot health-check rollback applies.
+6. Manifest is renamed to usbd-update.done.
+7. Reboot.
+
+== Requirements
+
+One file per store path:: it should be possible to rewrite the system update on USB stick in incremental fashion: unchanged store paths keep their exact file name and bytes, only added/removed paths change on the stick.
+Deterministic, non-leaking names and ciphertext:: file names and file contents must be byte-stable across rewrites (otherwise incremental sync degrades to full rewrite), but must not leak store path names/contents. Mix a secret into the derivation: per-file key and file name are derived from (deployment secret, narHash); the deployment secret is an ordinary fleet shared secret owned by the target host. Derivation must use narHash, not the store path: input-addressed paths can be rebuilt nondeterministically, yielding the same store path with different contents, and a same-name-different-bytes file breaks both immutability and size-only synchronization. Note age cannot be used for the path files - its encryption is randomized, so ciphertext would churn on every rewrite.
+Manifest encrypted to the host ssh key:: the manifest is encrypted the same way fleet already performs secret encryption (age, ssh-ed25519 host key recipient). It carries the system name and the per-path metadata listed in the copying section.
+Manifest should include system name:: update daemon should check if the system update is suitable for it.
+User feedback via callbacks:: how the user learns "update running / done / safe to remove" differs per client (screen, beep, LED). The daemon only exposes callback hooks for overriding; no built-in feedback.
+Backwards-compatible updates:: an old usbd must be able to apply updates produced by a newer fleet (forward compatibility from the daemon's point of view). Consequences: the on-stick format core (discovery label, manifest name and encryption, name/key derivation, file cipher, narinfo fields) is frozen from v1; the manifest is self-describing and old daemons ignore unknown fields, so evolution is additive-only. A truly breaking format change is still possible in two hops: every update ships the whole system including usbd itself, so the writer can emit the old format once to upgrade the daemon, then switch.
+
+== Copying store paths
+
+The daemon exposes the stick as a standard nix binary cache on localhost and lets nix substitution perform the copy. This reuses nix machinery for hash verification, signature checking and skipping already-present paths - no diffing logic in the daemon.
+
+1. Manifest carries, per store path: store path, narHash, narSize, references, build-time signature, on-stick file name and per-file key - everything needed to synthesize a narinfo.
+2. Daemon binds an ephemeral HTTP listener on 127.0.0.1 serving the binary cache protocol: /nix-cache-info, synthesized <hash>.narinfo (Sig: from build time, URL: nar/<derived-name>, Compression: zstd) and /nar/<derived-name>, which streaming-decrypts the stick file (chunked AEAD) into the .nar.zst; nix does the decompression and NAR hash check itself.
+3. The copy goes through the nix-eval FFI (Store::open on the localhost cache, copy_to into the local store), not a subprocess. A random token path prefix in the cache URL keeps other local processes from pulling decrypted NARs off the listener.
+4. Signatures are verified by nix against trusted-public-keys; the deployer signing key is pinned there by the fleet module.
+5. The copy holds a temporary GC root on the system path so a concurrent GC cannot race it. The listener is shut down after the copy, then the normal flow continues: set boot generation, rename manifest, reboot.
+
+Stick writing is the mirror image and also goes through the FFI: query path infos of the system closure, sign, dump each path to NAR, zstd-compress, encrypt with the derived per-file key, write under the derived name, write the age-encrypted manifest last.
+
+== Stick format
+
+FAT32 (mkfs.vfat): universally supported and more robust against unclean ejects than exFAT. The 4 GiB file size limit is handled by splitting large encrypted files into fixed-size chunks; chunk names are derived from (deployment secret, narHash, chunk index) and the manifest lists the chunk sequence per path. This is plain size-splitting, unrelated to dedup chunking. Filesystem is case-insensitive; derived file names should be lowercase.
+
+== Synchronization
+
+Stick content should be synchronizable from http with a known tool that works on Windows. Flat content-addressed layout plus one manifest file satisfies this: rclone handles it (single portable exe, can be pre-installed on the stick itself with a .bat wrapper running `rclone sync --progress`; rclone has no native GUI and its experimental web GUI downloads assets from github on first run, so it is unusable here). Files are immutable under their derived names, so size-only comparison suffices. Sync should use --delete-before: rclone does not order transfers and by default deletes extraneous files last, which would require the stick to hold the old and new closure simultaneously. A partially synced stick is safe: the copy is hash-verified and the closure must be complete before the switch happens; the daemon should stat all manifest-referenced files upfront to report a partially synced stick cleanly.
+
+== Usage
+
+On the host, enable the daemon and trust the update signing key:
+
+[source,nix]
+----
+{
+  services.fleet-usbd = {
+    enable = true;
+    signPublicKeys = [ "client-1:..." ];
+    # Optional user feedback script: receives copying/switching/done/noop/failed
+    # hook = ./usbd-hook.sh;
+  };
+}
+----
+
+The daemon is bound to the volume label (FLEETUSBD by default) via a systemd device unit; inserting the stick runs fsck and applies the update.
+
+Writing the stick (expects an inserted stick with the matching label, mounted via udisks):
+
+[source,shell]
+----
+mkfs.vfat -n FLEETUSBD /dev/sdX1
+fleet usbd write HOSTNAME --sign-key ./client-1.secret
+----
+
+The signing keypair is generated once with `nix key generate-secret --key-name client-1`.
+
+== Not a requirement
+
+Rollback prevention:: not a concern for any of the current clients.
+Update authenticity in the daemon:: handled by nix itself via store signatures during the copy; nothing to do on the usbd level.
+Chunking of large paths:: attic-style sub-path chunking would help large paths that change slightly, but conflicts with ease of synchronization; skipped.
+Status/log back-channel on the stick:: postponed, depends on the client.
+GC of old generations:: not a daemon problem.
+Multi-host sticks:: conflicts with incremental stick rewrite; one stick per host for now.

--- /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";
+}