diff --git a/hive-priv-sock/src/lib.rs b/hive-priv-sock/src/lib.rs index fcb71318..0c006a4b 100644 --- a/hive-priv-sock/src/lib.rs +++ b/hive-priv-sock/src/lib.rs @@ -715,6 +715,35 @@ pub enum PrivRequest { dest_file_name: String, }, + /// Stream a previously-created read-only snapshot into a file + /// descriptor the caller passes alongside this request (`SCM_RIGHTS` + /// ancillary data on the same socket): `btrfs send [-p ] + /// >&`. + /// + /// The network half of the inter-hive migration transport. hive-c0re + /// connects to the peer hive's snapshot store, writes the header + /// itself, and hands the **connected socket** over — so hive-priv + /// never learns an address, a protocol, or that a network is + /// involved, and nobody sits in the data path once the send starts + /// (which is what makes a multi-gigabyte transfer survive a + /// hive-c0re restart). + /// + /// Exactly one descriptor must accompany this request. hive-priv + /// rejects the request if none arrived, if more than one did, or if a + /// descriptor arrives alongside any *other* operation — no guessing + /// when the caller didn't say. Requires root. + SendAgentSnapshotToFd { + /// Logical agent name (validated by `validate_agent_name`). + agent_name: String, + /// Snapshot label to send, same validation as `SnapshotAgentSubvolume`. + snapshot_name: String, + /// Optional parent snapshot label for an incremental + /// (`btrfs send -p`) send — must be an older read-only snapshot of + /// the same agent, still present on disk. `None` sends the full + /// snapshot. + parent_snapshot_name: Option, + }, + /// Write `/etc/tmpfiles.d/hyperhive-agents.conf` for the given agent set /// and immediately apply it with `systemd-tmpfiles --create`. Each entry /// declares the per-agent runtime dirs (`/run/hyperhive/agents/` and diff --git a/hive-priv/src/main.rs b/hive-priv/src/main.rs index e6c9b019..e7fc424e 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -17,6 +17,7 @@ //! `LISTEN_FDS=1` + `LISTEN_PID=`, the inherited fd 3 is used //! instead of binding a fresh socket. +use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd}; use std::path::{Path, PathBuf}; use anyhow::{Context as _, Result, bail}; @@ -26,7 +27,7 @@ use hive_priv_sock::{ PAUSED_MARKER_FILE, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream, PrivStreamLine, SIBLING_CONTAINERS, }; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::io::{AsyncWriteExt, BufReader}; use tokio::net::unix::OwnedWriteHalf; use tokio::net::{UnixListener, UnixStream}; use tokio::process::Command; @@ -92,11 +93,162 @@ fn socket_listener() -> Result { Ok(listener) } +/// Ancillary-data buffer sized and aligned for one `SCM_RIGHTS` message. +/// +/// `CMSG_SPACE` is not a `const fn`, so the size is a literal with room +/// to spare (24 bytes are needed for a single descriptor on x86-64). The +/// union member gives the `cmsghdr` alignment `CMSG_FIRSTHDR` requires — +/// a bare `[u8; N]` is only byte-aligned and would be undefined behaviour +/// to walk. +#[repr(C)] +union CmsgSpace { + _align: libc::cmsghdr, + bytes: [u8; 32], +} + +/// One `recvmsg` into `buf`, returning the bytes read plus any file +/// descriptors that rode along as `SCM_RIGHTS`. +/// +/// Why not a plain read: ancillary data is attached to a *specific* +/// `recvmsg` call, so a buffered line reader cannot surface it — it +/// reads the bytes and silently drops the descriptor. +/// +/// `MSG_CMSG_CLOEXEC` is not optional: without it a received descriptor +/// is inherited by every `btrfs` / `nixos-container` child this helper +/// later spawns. +fn recv_with_fds(sock: RawFd, buf: &mut [u8]) -> std::io::Result<(usize, Vec)> { + const FD_SIZE: usize = std::mem::size_of::(); + + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr().cast(), + iov_len: buf.len(), + }; + let mut cmsg = CmsgSpace { bytes: [0; 32] }; + // SAFETY: msghdr is a plain C struct with no invalid bit patterns; + // every field we care about is set immediately below. + let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; + msg.msg_iov = &raw mut iov; + msg.msg_iovlen = 1; + msg.msg_control = std::ptr::addr_of_mut!(cmsg.bytes).cast(); + msg.msg_controllen = 32; + + // SAFETY: `msg` points at a live iovec covering `buf` and a live, + // correctly aligned control buffer of the length we just declared. + let n = unsafe { libc::recvmsg(sock, &raw mut msg, libc::MSG_CMSG_CLOEXEC) }; + if n < 0 { + return Err(std::io::Error::last_os_error()); + } + + // Take ownership of every descriptor the kernel attached, even ones + // this protocol never expects: an `OwnedFd` we drop is closed, an + // fd we fail to claim is leaked for the lifetime of the process. + let mut fds = Vec::new(); + // SAFETY: `msg` was just filled in by a successful `recvmsg`. + let mut cmsgp = unsafe { libc::CMSG_FIRSTHDR(&raw const msg) }; + while !cmsgp.is_null() { + // SAFETY: CMSG_FIRSTHDR / CMSG_NXTHDR only ever return a pointer + // to a complete header inside the control buffer. + let hdr = unsafe { std::ptr::read_unaligned(cmsgp) }; + if hdr.cmsg_level == libc::SOL_SOCKET && hdr.cmsg_type == libc::SCM_RIGHTS { + // SAFETY: same, and CMSG_LEN(0) is the header's own length. + let payload = hdr.cmsg_len as usize - unsafe { libc::CMSG_LEN(0) } as usize; + let count = payload / FD_SIZE; + // SAFETY: CMSG_DATA points at `payload` bytes of descriptors. + let data = unsafe { libc::CMSG_DATA(cmsgp) }; + for i in 0..count { + // Copied out byte-wise rather than read through a + // `*const RawFd`: the control buffer is only guaranteed + // `cmsghdr`-aligned, so casting to a more strictly + // aligned pointer would be unsound even where it happens + // to work. + let mut raw = [0u8; FD_SIZE]; + // SAFETY: i < count, so this reads inside the payload. + unsafe { + std::ptr::copy_nonoverlapping(data.add(i * FD_SIZE), raw.as_mut_ptr(), FD_SIZE); + } + // SAFETY: the kernel just created this descriptor for + // us — we are its only owner. + fds.push(unsafe { OwnedFd::from_raw_fd(RawFd::from_ne_bytes(raw)) }); + } + } + // SAFETY: `cmsgp` came from this same message. + cmsgp = unsafe { libc::CMSG_NXTHDR(&raw const msg, cmsgp) }; + } + + // `n >= 0` was checked above, so the conversion cannot fail; going + // through `try_from` keeps it a cast-free, lint-clean widening. + let read = usize::try_from(n).unwrap_or_default(); + Ok((read, fds)) +} + +/// Reads newline-delimited requests off one connection, pairing each +/// with the descriptor that arrived with it. +/// +/// The pairing is deliberately trivial, because the protocol is: +/// `hive-sock-client` connects per request, so a connection carries one +/// line and at most one descriptor. The loop below still handles several +/// sequential requests (the server always has), but it refuses to guess +/// — a second descriptor arriving before its line is a protocol error, +/// not something to queue and hope about. +struct Requests<'a> { + sock: &'a UnixStream, + buf: Vec, + fd: Option, +} + +impl Requests<'_> { + /// Next complete request line and its descriptor, or `None` at EOF. + async fn next(&mut self) -> Result)>> { + loop { + if let Some(nl) = self.buf.iter().position(|&b| b == b'\n') { + let line: Vec = self.buf.drain(..=nl).take(nl).collect(); + let line = String::from_utf8(line).context("request line was not valid UTF-8")?; + return Ok(Some((line, self.fd.take()))); + } + + let mut chunk = [0u8; 8192]; + let raw = self.sock.as_raw_fd(); + let (n, fds) = self + .sock + .async_io(tokio::io::Interest::READABLE, || { + recv_with_fds(raw, &mut chunk) + }) + .await + .context("recvmsg on the priv socket")?; + + for fd in fds { + if self.fd.replace(fd).is_some() { + bail!("more than one file descriptor passed for a single request"); + } + } + if n == 0 { + if !self.buf.is_empty() { + bail!("connection closed mid-request ({} bytes)", self.buf.len()); + } + return Ok(None); + } + self.buf.extend_from_slice(&chunk[..n]); + } + } +} + async fn handle(stream: UnixStream) { let (reader, mut writer) = stream.into_split(); - let mut lines = BufReader::new(reader).lines(); - while let Ok(Some(line)) = lines.next_line().await { - let resp = dispatch(&line, &mut writer).await; + let mut requests = Requests { + sock: reader.as_ref(), + buf: Vec::new(), + fd: None, + }; + loop { + let (line, fd) = match requests.next().await { + Ok(Some(req)) => req, + Ok(None) => break, + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "reading request failed"); + break; + } + }; + let resp = dispatch(&line, fd, &mut writer).await; // Write the terminal PrivResponse as a PrivEvent::Done. Wire-identical // to a bare PrivResponse (untagged), so old hive-c0re callers that // deserialise directly to PrivResponse continue to work. @@ -112,31 +264,54 @@ async fn handle(stream: UnixStream) { } } -async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse { - match serde_json::from_str::(line) { - Ok(req) => match exec(req, writer).await { - Ok((stdout, stderr)) => PrivResponse { - ok: true, - stdout, - stderr, - error: None, - }, - Err(e) => PrivResponse { - ok: false, - stdout: String::new(), - stderr: String::new(), - error: Some(format!("{e:#}")), - }, +/// Reject a request whose descriptor and operation disagree, in either +/// direction. +/// +/// No guessing when the caller didn't say: an op that streams into a +/// passed descriptor cannot invent one, and an op that takes none must +/// not silently accept one. Returning the `Err` here drops the +/// `OwnedFd`, which closes it. +fn check_fd_agreement(req: &PrivRequest, fd: Option<&OwnedFd>) -> Result<()> { + let wants_fd = matches!(req, PrivRequest::SendAgentSnapshotToFd { .. }); + match (wants_fd, fd.is_some()) { + (true, false) => bail!("this operation requires a passed file descriptor, none arrived"), + (false, true) => bail!("this operation does not take a passed file descriptor"), + _ => Ok(()), + } +} + +async fn dispatch(line: &str, fd: Option, writer: &mut OwnedWriteHalf) -> PrivResponse { + match run(line, fd, writer).await { + Ok((stdout, stderr)) => PrivResponse { + ok: true, + stdout, + stderr, + error: None, }, Err(e) => PrivResponse { ok: false, stdout: String::new(), stderr: String::new(), - error: Some(format!("parse request: {e}")), + error: Some(format!("{e:#}")), }, } } +/// Parse one request line, check it agrees with the descriptor that +/// arrived with it, and execute it. +/// +/// Split out of [`dispatch`] so the three failure modes collapse into one +/// `Result` instead of three nested matches building the same struct. +async fn run( + line: &str, + fd: Option, + writer: &mut OwnedWriteHalf, +) -> Result<(String, String)> { + let req = serde_json::from_str::(line).context("parse request")?; + check_fd_agreement(&req, fd.as_ref())?; + exec(req, fd, writer).await +} + /// Write one `PrivEvent::Line` to the client. Best-effort: a write /// failure is logged but doesn't abort the running subprocess. async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) { @@ -159,7 +334,17 @@ async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: // One match arm per priv op — a flat 1:1 dispatch table. The length tracks // the op count, not complexity; splitting it would just scatter the mapping. #[allow(clippy::too_many_lines)] -async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> { +/// Execute one validated request. +/// +/// `fd` is the descriptor that arrived with this request, already checked +/// against the operation by [`check_fd_agreement`]: `Some` exactly for +/// the variants that stream into a caller-supplied descriptor, `None` +/// for every other operation. +async fn exec( + req: PrivRequest, + fd: Option, + writer: &mut OwnedWriteHalf, +) -> Result<(String, String)> { match req { PrivRequest::StartContainer { ref name } => { validate_container_name(name)?; @@ -417,6 +602,26 @@ async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, .await } + PrivRequest::SendAgentSnapshotToFd { + ref agent_name, + ref snapshot_name, + ref parent_snapshot_name, + } => { + validate_agent_name(agent_name)?; + validate_snapshot_name(snapshot_name)?; + if let Some(parent) = parent_snapshot_name { + validate_snapshot_name(parent)?; + } + let dest = fd.context("no descriptor to stream into")?; + send_agent_snapshot_to_fd( + agent_name, + snapshot_name, + parent_snapshot_name.as_deref(), + dest, + ) + .await + } + PrivRequest::SyncAgentTmpfiles { ref agents } => sync_agent_tmpfiles(agents).await, } } @@ -1424,6 +1629,70 @@ async fn send_agent_snapshot_to_file( Ok((dest.display().to_string(), String::new())) } +/// `SendAgentSnapshotToFd` — stream a read-only snapshot (optionally +/// incremental against `parent_name`) straight into a descriptor the +/// caller passed us. +/// +/// The network half of the inter-hive migration transport, arranged so +/// this helper never learns there *is* a network: hive-c0re connects to +/// the peer's snapshot store, writes the header itself, and hands the +/// connected socket over. We only ever see "a thing to write bytes into", +/// which keeps a root process out of any address, protocol or trust +/// decision — and keeps everyone out of the data path once `btrfs send` +/// starts, which matters at multi-gigabyte sizes. +async fn send_agent_snapshot_to_fd( + agent_name: &str, + snapshot_name: &str, + parent_name: Option<&str>, + dest: OwnedFd, +) -> Result<(String, String)> { + let snap = snapshot_path(agent_name, snapshot_name); + if !snap.exists() { + bail!( + "snapshot {} does not exist — create it with `subvol snapshot create` first", + snap.display() + ); + } + + let mut cmd = Command::new("btrfs"); + cmd.arg("send"); + if let Some(parent) = parent_name { + let parent_path = snapshot_path(agent_name, parent); + if !parent_path.exists() { + bail!( + "parent snapshot {} does not exist — pick an existing parent or omit it for a full send", + parent_path.display() + ); + } + cmd.arg("-p").arg(&parent_path); + } + cmd.arg(&snap); + cmd.stdout(std::process::Stdio::from(dest)); + cmd.stderr(std::process::Stdio::piped()); + + let out = cmd + .spawn() + .with_context(|| format!("spawn btrfs send {}", snap.display()))? + .wait_with_output() + .await + .with_context(|| format!("wait on btrfs send {}", snap.display()))?; + if !out.status.success() { + // Nothing to clean up: the destination isn't ours. A partial + // stream is the receiving end's problem, and `btrfs receive` + // refuses to commit an incomplete subvolume anyway. + bail!( + "btrfs send {} failed: {}", + snap.display(), + String::from_utf8_lossy(&out.stderr).trim() + ); + } + tracing::info!( + agent = %agent_name, snapshot = %snap.display(), parent = ?parent_name, + "streamed agent snapshot into a passed descriptor" + ); + Ok((String::new(), String::new())) +} + /// `SetSubvolumeQuota` — set or clear a qgroup size limit on an agent /// subvolume (`btrfs qgroup limit <…/agent_name>`). See the /// wire doc.