From 418b50980c187aecd4c8211c977a1be4049832f2 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 9 Nov 2026 23:59:57 +0100 Subject: [PATCH] hive-priv: drop the descriptor-taking snapshot send and its fd-receiving loop --- .../trust-boundary/hive-c0re-privilege.md | 1 - hive-priv-sock/src/lib.rs | 28 - hive-priv/src/main.rs | 586 +----------------- hive-priv/src/storage.rs | 174 +----- 4 files changed, 39 insertions(+), 750 deletions(-) diff --git a/docs/4-implementation/trust-boundary/hive-c0re-privilege.md b/docs/4-implementation/trust-boundary/hive-c0re-privilege.md index d29007b33..b6a1dfbf2 100644 --- a/docs/4-implementation/trust-boundary/hive-c0re-privilege.md +++ b/docs/4-implementation/trust-boundary/hive-c0re-privilege.md @@ -57,7 +57,6 @@ known operations; there is no arbitrary command pass-through: | `SnapshotAgentSubvolume` | `btrfs subvolume snapshot -r ` — the frozen point-in-time copy a migration streams from | | `DeleteAgentSnapshot` | `btrfs subvolume delete ` | | `SendAgentSnapshotToFile` | `btrfs send [-p ] ` into a bare filename under `MIGRATE_STAGING_ROOT`; refuses to overwrite an existing export | -| `SendAgentSnapshotToFd` | `btrfs send [-p ] ` into a **file descriptor passed with the request** (`SCM_RIGHTS`). The caller connects to the destination and hands over the connected socket, so hive-priv never learns an address or protocol; hive-c0re sends no such request today. hive-priv requires exactly one descriptor here, and refuses one that arrives alongside any other operation | **Strings that reach root-loaded config** — `WriteResourceLimits` and `WriteNspawnFlags` copy request strings into a systemd drop-in and an env diff --git a/hive-priv-sock/src/lib.rs b/hive-priv-sock/src/lib.rs index 6f0139f5c..7ce8b7d5a 100644 --- a/hive-priv-sock/src/lib.rs +++ b/hive-priv-sock/src/lib.rs @@ -645,34 +645,6 @@ 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 caller connects to the destination, writes any 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). hive-c0re sends no such request today. - /// - /// 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, - }, - /// Legacy: an older hive-c0re still sends this. hive-priv only unlinks /// `/etc/tmpfiles.d/hyperhive-agents.conf` and returns `Ok`; the `agents` /// payload that request carries is ignored on decode. Remove once no diff --git a/hive-priv/src/main.rs b/hive-priv/src/main.rs index 7d72e46c7..8148b7fbf 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -39,11 +39,9 @@ mod shared; mod storage; mod userns; -use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd}; - use anyhow::{Context as _, Result, bail}; use hive_priv_sock::{PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse}; -use tokio::io::AsyncWriteExt; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::unix::OwnedWriteHalf; use tokio::net::{UnixListener, UnixStream}; @@ -130,162 +128,11 @@ 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 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; + let mut lines = BufReader::new(reader).lines(); + while let Ok(Some(line)) = lines.next_line().await { + let resp = dispatch(&line, &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. @@ -301,24 +148,8 @@ async fn handle(stream: UnixStream) { } } -/// 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 { - response_of(run(line, fd, writer).await) +async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse { + response_of(run(line, writer).await) } /// The wire answer for one request's outcome: the handler's output on @@ -340,40 +171,25 @@ fn response_of(outcome: Result<(String, String)>) -> PrivResponse { } } -/// Parse one request line, check it agrees with the descriptor that -/// arrived with it, and execute it. +/// Parse one request line 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)> { +/// Split out of [`dispatch`] so the two failure modes collapse into one +/// `Result` instead of nested matches building the same struct. +async fn run(line: &str, 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 + exec(req, writer).await } /// Execute a validated `PrivRequest`. Returns `(stdout, stderr)` on success. /// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`) /// output lines are forwarded to `writer` as `PrivEvent::Line` messages and /// the returned strings are empty. -/// -/// `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. // One match arm per priv op — a flat 1:1 dispatch table. Every arm either // delegates directly or validates then delegates; an op whose handling is // more than that gets its own named function instead, so the match's length // tracks the op count, not complexity. #[allow(clippy::too_many_lines)] -async fn exec( - req: PrivRequest, - fd: Option, - writer: &mut OwnedWriteHalf, -) -> Result<(String, String)> { +async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> { match req { PrivRequest::StartContainer { ref name } => container::start_container(name).await, @@ -514,96 +330,20 @@ async fn exec( .await } - PrivRequest::SendAgentSnapshotToFd { - ref agent_name, - ref snapshot_name, - ref parent_snapshot_name, - } => { - storage::exec_send_agent_snapshot_to_fd( - agent_name, - snapshot_name, - parent_snapshot_name.as_deref(), - fd, - ) - .await - } - PrivRequest::SyncAgentTmpfiles => { files::remove_legacy_tmpfiles()?; Ok((String::new(), String::new())) } } } + #[cfg(test)] mod tests { - use std::io::Write as _; - use std::os::fd::{AsRawFd as _, OwnedFd, RawFd}; - use std::os::unix::fs::MetadataExt as _; - use hive_priv_sock::{PrivEvent, PrivRequest, PrivResponse}; use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; use tokio::net::UnixStream; - use super::{ - Requests, check_fd_agreement, check_socket_activation, dispatch, handle, recv_with_fds, - response_of, - }; - - /// A request that streams into a caller-supplied descriptor. - fn fd_taking_request() -> PrivRequest { - PrivRequest::SendAgentSnapshotToFd { - agent_name: "atlas".to_owned(), - snapshot_name: "hive-migrate".to_owned(), - parent_snapshot_name: None, - } - } - - /// A real descriptor — `/dev/null` rather than a fake, so the drop - /// that closes it on a rejection path is genuinely exercised. - fn some_fd() -> OwnedFd { - std::fs::File::open("/dev/null") - .expect("open /dev/null") - .into() - } - - /// An op that streams into a passed descriptor cannot invent one: - /// falling back to anything (a temp file, the response socket) would - /// send an agent's state somewhere the caller never asked for. - #[test] - fn an_fd_taking_op_without_a_descriptor_is_rejected() { - let err = check_fd_agreement(&fd_taking_request(), None) - .expect_err("no descriptor arrived, so this must not proceed"); - let msg = format!("{err:#}"); - assert!(msg.contains("requires a passed file descriptor"), "{msg}"); - } - - /// The mirror case: a descriptor sent alongside an op that takes - /// none is a protocol error, not something to ignore. Returning the - /// error drops the `OwnedFd`, which closes it — the alternative - /// leaks one descriptor per stray request in a long-lived root - /// process. - #[test] - fn a_descriptor_sent_to_an_op_that_takes_none_is_rejected() { - let fd = some_fd(); - let err = check_fd_agreement(&PrivRequest::DaemonReload, Some(&fd)) - .expect_err("an unexpected descriptor must not be silently ignored"); - let msg = format!("{err:#}"); - assert!( - msg.contains("does not take a passed file descriptor"), - "{msg}" - ); - } - - /// Both agreeing combinations pass, so the check rejects mismatches - /// rather than descriptors in general. - #[test] - fn agreeing_combinations_are_accepted() { - let fd = some_fd(); - check_fd_agreement(&fd_taking_request(), Some(&fd)) - .expect("an fd-taking op with its descriptor is the normal case"); - check_fd_agreement(&PrivRequest::DaemonReload, None) - .expect("every ordinary request arrives without a descriptor"); - } + use super::{check_socket_activation, dispatch, handle, response_of}; /// An older hive-c0re's `SyncAgentTmpfiles` still decodes, payload and /// all, so it gets the legacy cleanup rather than a parse error. Control: @@ -619,57 +359,6 @@ mod tests { assert!(serde_json::from_str::(r#"{"op":"sync_agent_tmpfilez"}"#).is_err()); } - /// `sendmsg` `bytes` with every descriptor of `fds` attached as one - /// `SCM_RIGHTS` message: the client side of what `recv_with_fds` reads. - fn send_with_fds(sock: RawFd, bytes: &[u8], fds: &[RawFd]) { - const FD_SIZE: usize = std::mem::size_of::(); - let payload = u32::try_from(FD_SIZE * fds.len()).unwrap(); - let mut ctrl = [0u64; 8]; - let mut iov = libc::iovec { - iov_base: bytes.as_ptr().cast_mut().cast(), - iov_len: bytes.len(), - }; - // SAFETY: msghdr is a plain C struct with no invalid bit patterns. - let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; - msg.msg_iov = &raw mut iov; - msg.msg_iovlen = 1; - // SAFETY: CMSG_SPACE is a pure size computation. - msg.msg_controllen = unsafe { libc::CMSG_SPACE(payload) } as _; - msg.msg_control = ctrl.as_mut_ptr().cast(); - // SAFETY: `ctrl` is aligned for `cmsghdr` and large enough for the - // header and `fds.len()` descriptors. - unsafe { - let hdr = libc::CMSG_FIRSTHDR(&raw const msg); - (*hdr).cmsg_level = libc::SOL_SOCKET; - (*hdr).cmsg_type = libc::SCM_RIGHTS; - (*hdr).cmsg_len = libc::CMSG_LEN(payload) as _; - std::ptr::copy_nonoverlapping( - fds.as_ptr().cast::(), - libc::CMSG_DATA(hdr), - payload as usize, - ); - assert!(libc::sendmsg(sock, &raw const msg, 0) >= 0, "sendmsg"); - } - } - - fn is_cloexec(fd: &OwnedFd) -> bool { - // SAFETY: `fd` is an open descriptor. - let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFD) }; - flags >= 0 && flags & libc::FD_CLOEXEC != 0 - } - - fn same_file(fd: &OwnedFd, other: &std::fs::File) -> bool { - let ours = std::fs::File::from(fd.try_clone().unwrap()) - .metadata() - .unwrap(); - let theirs = other.metadata().unwrap(); - (ours.dev(), ours.ino()) == (theirs.dev(), theirs.ino()) - } - - fn dev_null() -> std::fs::File { - std::fs::File::open("/dev/null").expect("open /dev/null") - } - // ---- socket activation ---- /// Only systemd's own handover starts the daemon: at least one fd, and @@ -702,161 +391,7 @@ mod tests { } } - // ---- recv_with_fds ---- - - /// Plain bytes come back with no descriptors, and a closed peer reads as - /// zero bytes. - #[test] - fn recv_reads_plain_bytes_and_end_of_stream() { - let (a, b) = std::os::unix::net::UnixStream::pair().unwrap(); - (&a).write_all(b"hello").unwrap(); - drop(a); - let mut buf = [0u8; 16]; - let (n, fds) = recv_with_fds(b.as_raw_fd(), &mut buf).unwrap(); - assert_eq!((&buf[..n], fds.len()), (&b"hello"[..], 0)); - let (n, fds) = recv_with_fds(b.as_raw_fd(), &mut buf).unwrap(); - assert_eq!((n, fds.len()), (0, 0)); - } - - /// A descriptor sent with the bytes arrives as the same open file, and - /// close-on-exec: without that every `btrfs` child the helper spawns - /// would inherit it. - #[test] - fn recv_returns_the_passed_descriptor_close_on_exec() { - let (a, b) = std::os::unix::net::UnixStream::pair().unwrap(); - let file = dev_null(); - send_with_fds(a.as_raw_fd(), b"x", &[file.as_raw_fd()]); - let mut buf = [0u8; 16]; - let (n, mut fds) = recv_with_fds(b.as_raw_fd(), &mut buf).unwrap(); - assert_eq!(n, 1); - assert_eq!(fds.len(), 1); - let fd = fds.pop().unwrap(); - assert!(is_cloexec(&fd), "a received descriptor must be CLOEXEC"); - assert!(same_file(&fd, &file)); - } - - /// Every descriptor of the message is claimed, even though the protocol - /// takes one: an unclaimed one would leak for the life of the process. - #[test] - fn recv_claims_every_descriptor_of_a_message() { - let (a, b) = std::os::unix::net::UnixStream::pair().unwrap(); - let (f1, f2) = (dev_null(), dev_null()); - send_with_fds(a.as_raw_fd(), b"x", &[f1.as_raw_fd(), f2.as_raw_fd()]); - let mut buf = [0u8; 16]; - let (_, fds) = recv_with_fds(b.as_raw_fd(), &mut buf).unwrap(); - assert_eq!(fds.len(), 2); - assert!(fds.iter().all(is_cloexec)); - } - - /// A read on a descriptor that is not a socket is the OS error, not a - /// panic or an empty read. - #[test] - fn recv_on_a_bad_descriptor_is_an_error() { - let mut buf = [0u8; 4]; - assert!(recv_with_fds(-1, &mut buf).is_err()); - } - - // ---- Requests: line framing and descriptor pairing ---- - - fn requests(sock: &UnixStream) -> Requests<'_> { - Requests { - sock, - buf: Vec::new(), - fd: None, - } - } - - /// Lines are split at the newline, several in one read come out in - /// order, and end of stream is `None`. - #[tokio::test] - async fn requests_split_lines_in_order_and_end_at_eof() { - let (mut client, server) = UnixStream::pair().unwrap(); - client.write_all(b"one\ntwo\n\nthree\n").await.unwrap(); - drop(client); - let mut reqs = requests(&server); - for want in ["one", "two", "", "three"] { - let (line, fd) = reqs.next().await.unwrap().unwrap(); - assert_eq!(line, want); - assert!(fd.is_none()); - } - assert!(reqs.next().await.unwrap().is_none()); - } - - /// A line that arrives in pieces is one request. - #[tokio::test] - async fn requests_join_a_line_split_across_reads() { - let (mut client, server) = UnixStream::pair().unwrap(); - let writer = tokio::spawn(async move { - client.write_all(b"{\"op\":\"daemon").await.unwrap(); - client.flush().await.unwrap(); - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - client.write_all(b"_reload\"}\n").await.unwrap(); - }); - let mut reqs = requests(&server); - let (line, _) = reqs.next().await.unwrap().unwrap(); - assert_eq!(line, r#"{"op":"daemon_reload"}"#); - writer.await.unwrap(); - } - - /// A connection that closes before the newline is an error naming how - /// much was cut off, never a request. - #[tokio::test] - async fn requests_refuse_a_line_cut_off_by_eof() { - let (mut client, server) = UnixStream::pair().unwrap(); - client.write_all(b"{\"op\":").await.unwrap(); - drop(client); - let err = requests(&server).next().await.unwrap_err(); - assert!( - format!("{err:#}").contains("closed mid-request (6 bytes)"), - "{err:#}" - ); - } - - /// A line that is not UTF-8 is refused outright. - #[tokio::test] - async fn requests_refuse_a_line_that_is_not_utf8() { - let (mut client, server) = UnixStream::pair().unwrap(); - client.write_all(b"\xff\xfe\n").await.unwrap(); - let err = requests(&server).next().await.unwrap_err(); - assert!(format!("{err:#}").contains("not valid UTF-8"), "{err:#}"); - } - - /// The descriptor that came with a line is handed out with that line - /// and with no other. - #[tokio::test] - async fn requests_pair_a_descriptor_with_its_line_only() { - let (client, server) = UnixStream::pair().unwrap(); - let file = dev_null(); - send_with_fds(client.as_raw_fd(), b"first\n", &[file.as_raw_fd()]); - send_with_fds(client.as_raw_fd(), b"second\n", &[]); - let mut reqs = requests(&server); - let (line, fd) = reqs.next().await.unwrap().unwrap(); - assert_eq!(line, "first"); - assert!(same_file( - &fd.expect("the descriptor rides with its line"), - &file - )); - let (line, fd) = reqs.next().await.unwrap().unwrap(); - assert_eq!(line, "second"); - assert!(fd.is_none(), "the descriptor must not carry over"); - } - - /// A second descriptor before the first one's line is complete is a - /// protocol error, not something to queue. - #[tokio::test] - async fn requests_refuse_a_second_descriptor_for_one_line() { - let (client, server) = UnixStream::pair().unwrap(); - let file = dev_null(); - send_with_fds(client.as_raw_fd(), b"par", &[file.as_raw_fd()]); - send_with_fds(client.as_raw_fd(), b"tial\n", &[file.as_raw_fd()]); - let err = requests(&server).next().await.unwrap_err(); - assert!( - format!("{err:#}").contains("more than one file descriptor"), - "{err:#}" - ); - } - - // ---- dispatch: parsing, fd agreement, validation, error mapping ---- + // ---- dispatch: parsing, validation, error mapping ---- /// Success carries the handler's output and no error; failure carries the /// whole context chain and no output. @@ -880,9 +415,9 @@ mod tests { } /// The failed response for `line`, asserting nothing else came back. - async fn refused(line: &str, fd: Option) -> String { + async fn refused(line: &str) -> String { let (mut w, _keep) = sink(); - let resp = dispatch(line, fd, &mut w).await; + let resp = dispatch(line, &mut w).await; assert!(!resp.ok, "{line} was accepted"); assert!(resp.stdout.is_empty() && resp.stderr.is_empty(), "{line}"); resp.error.expect("a failure carries its error") @@ -904,35 +439,11 @@ mod tests { r#"{"op":"start_container","name":7}"#, r#"{"op":"control_infra_container","container":"Forge","action":"stop"}"#, ] { - let msg = refused(line, None).await; + let msg = refused(line).await; assert!(msg.starts_with("parse request"), "{line}: {msg}"); } } - /// A descriptor on an op that takes none, and a missing one on an op that - /// needs it, are answered with the refusal before the op runs. - #[tokio::test] - async fn a_fd_sent_to_an_op_that_takes_none_is_refused_before_it_runs() { - // The name is bad too: only the descriptor check can produce this - // message, so the op's own validation was not reached. - let msg = refused( - r#"{"op":"set_agent_paused","agent_name":"../x","paused":true}"#, - Some(dev_null().into()), - ) - .await; - assert!( - msg.contains("does not take a passed file descriptor"), - "{msg}" - ); - - let msg = refused( - r#"{"op":"send_agent_snapshot_to_fd","agent_name":"atlas","snapshot_name":"hive-x","parent_snapshot_name":null}"#, - None, - ) - .await; - assert!(msg.contains("requires a passed file descriptor"), "{msg}"); - } - /// Every op that names an agent or container refuses a bad name from the /// request line itself, before any command runs or path is built. A /// control (a valid name on the same op) is in the lifecycle and storage @@ -1016,20 +527,9 @@ mod tests { )); for (line, want) in &cases { - let msg = refused(line, None).await; + let msg = refused(line).await; assert!(msg.contains(want), "{line}: {msg}"); } - // The fd op validates before it looks for its descriptor. - let msg = refused( - &named( - "send_agent_snapshot_to_fd", - "agent_name", - r#","snapshot_name":"hive-x","parent_snapshot_name":null"#, - ), - Some(dev_null().into()), - ) - .await; - assert!(msg.contains("invalid name"), "{msg}"); } /// Snapshot labels and the export file name are validated from the wire @@ -1043,7 +543,7 @@ mod tests { }; for op in ["snapshot_agent_subvolume", "delete_agent_snapshot"] { for label in ["", "x", "../hive-x", "hive-../x", "hive-a/b", "hive-a.b"] { - let msg = refused(&snap(op, label), None).await; + let msg = refused(&snap(op, label)).await; assert!(msg.contains("invalid"), "{op} {label:?}: {msg}"); } } @@ -1060,19 +560,16 @@ mod tests { ("hive-x", "null", ""), ("hive-x", "null", "out.img"), ] { - let msg = refused(&export(label, parent, dest), None).await; + let msg = refused(&export(label, parent, dest)).await; assert!(msg.contains("invalid"), "{label} {parent} {dest}: {msg}"); } - let fd_op = r#"{"op":"send_agent_snapshot_to_fd","agent_name":"atlas","snapshot_name":"hive-x","parent_snapshot_name":"../x"}"#; - let msg = refused(fd_op, Some(dev_null().into())).await; - assert!(msg.contains("invalid"), "{msg}"); } /// The error is the whole anyhow chain, so a caller sees the cause and /// the context in one string. #[tokio::test] async fn the_error_carries_the_whole_context_chain() { - let msg = refused(r#"{"op":"start_container","name":7}"#, None).await; + let msg = refused(r#"{"op":"start_container","name":7}"#).await; assert!(msg.starts_with("parse request: "), "{msg}"); assert!(msg.contains("invalid type"), "{msg}"); } @@ -1117,39 +614,22 @@ mod tests { assert!(resps.iter().all(|r| !r.ok)); } - /// A connection that dies mid-line or sends non-UTF-8 gets no response - /// and nothing runs. + /// A connection that sends non-UTF-8 or nothing gets no response and + /// nothing runs. A line cut off by EOF is still one line: it fails to + /// parse and is answered with the parse error, never executed. #[tokio::test] - async fn a_broken_request_closes_the_connection_without_a_response() { - assert!(serve(b"{\"op\":\"start_cont").await.is_empty()); + async fn a_broken_request_closes_the_connection_without_running_anything() { assert!(serve(b"\xff\xfe\n").await.is_empty()); assert!(serve(b"").await.is_empty()); - } - - /// The descriptor check holds end to end: a stray descriptor on a - /// request that takes none is answered with the refusal. - #[tokio::test] - async fn a_stray_descriptor_is_refused_over_the_wire() { - let (client, server) = UnixStream::pair().unwrap(); - let task = tokio::spawn(handle(server)); - let file = dev_null(); - send_with_fds( - client.as_raw_fd(), - b"{\"op\":\"set_agent_paused\",\"agent_name\":\"../x\",\"paused\":true}\n", - &[file.as_raw_fd()], - ); - let (mut rd, mut wr) = client.into_split(); - wr.shutdown().await.unwrap(); - let mut raw = String::new(); - rd.read_to_string(&mut raw).await.unwrap(); - task.await.unwrap(); - let resp: PrivResponse = serde_json::from_str(raw.trim()).unwrap(); - assert!(!resp.ok); + let cut = serve(b"{\"op\":\"start_cont").await; + assert_eq!(cut.len(), 1); + assert!(!cut[0].ok); assert!( - resp.error + cut[0] + .error + .as_deref() .unwrap() - .contains("does not take a passed file descriptor"), - "{raw}" + .starts_with("parse request") ); } } diff --git a/hive-priv/src/storage.rs b/hive-priv/src/storage.rs index 186f19f46..efeb2affd 100644 --- a/hive-priv/src/storage.rs +++ b/hive-priv/src/storage.rs @@ -2,13 +2,12 @@ //! `DeleteAgentSubvolume`, `UpgradeAgentSubvolume`), qgroup accounting and //! quotas (`EnsureBtrfsQuota`, `ReadSubvolumeUsage`, `SetSubvolumeQuota`), //! read-only snapshots and their export (`SnapshotAgentSubvolume`, -//! `DeleteAgentSnapshot`, `SendAgentSnapshotToFile`, `SendAgentSnapshotToFd`) +//! `DeleteAgentSnapshot`, `SendAgentSnapshotToFile`) //! and the `du`-based `MeasureAgentDisk`. //! //! Every path is derived from a validated agent name (and, for snapshots, a //! `hive-`-prefixed label) under `AGENT_STATE_ROOT`. -use std::os::fd::OwnedFd; use std::path::{Path, PathBuf}; use anyhow::{Context as _, Result, anyhow, bail}; @@ -101,29 +100,6 @@ pub(crate) async fn exec_send_agent_snapshot_to_file( .await } -/// `SendAgentSnapshotToFd`. -pub(crate) async fn exec_send_agent_snapshot_to_fd( - agent_name: &str, - snapshot_name: &str, - parent_snapshot_name: Option<&str>, - fd: Option, -) -> Result<(String, String)> { - 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( - &Env::real(), - agent_name, - snapshot_name, - parent_snapshot_name, - dest, - ) - .await -} - /// A btrfs snapshot label must start with `hive-` — this doubles as an /// allow-list: only names hivectl itself constructs (or an operator who /// knows the convention) can reach the `btrfs subvolume snapshot`/`delete` @@ -953,70 +929,6 @@ 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. -/// -/// Arranged so this helper never learns there *is* a network: the -/// caller connects to the destination, writes any 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( - env: &Env<'_, impl Spawn, impl Probes>, - agent_name: &str, - snapshot_name: &str, - parent_name: Option<&str>, - dest: OwnedFd, -) -> Result<(String, String)> { - let snap = snapshot_path(env.state_root, agent_name, snapshot_name); - if !snap.exists() { - bail!( - "snapshot {} does not exist — create it with `subvol snapshot create` first", - snap.display() - ); - } - - let mut cmd = env.btrfs(); - cmd.arg("send"); - if let Some(parent) = parent_name { - let parent_path = snapshot_path(env.state_root, 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. @@ -1051,12 +963,11 @@ mod tests { delete_agent_subvolume, du_command, du_total, ensure_agent_subvolume, ensure_btrfs_quota_in, exec_delete_agent_snapshot, exec_delete_agent_subvolume, exec_ensure_agent_subvolume, exec_measure_agent_disk, exec_read_subvolume_usage, - exec_send_agent_snapshot_to_fd, exec_send_agent_snapshot_to_file, exec_set_subvolume_quota, - exec_snapshot_agent_subvolume, exec_upgrade_agent_subvolume, existing_paths, - is_btrfs_subvolume, is_on_btrfs, is_subvolume_root, open_export_dest, parse_du_total, - read_subvolume_usage, send_agent_snapshot_to_fd, send_agent_snapshot_to_file, - set_subvolume_quota, snapshot_agent_subvolume, snapshot_path, upgrade_agent_subvolume, - validate_snapshot_name, + exec_send_agent_snapshot_to_file, exec_set_subvolume_quota, exec_snapshot_agent_subvolume, + exec_upgrade_agent_subvolume, existing_paths, is_btrfs_subvolume, is_on_btrfs, + is_subvolume_root, open_export_dest, parse_du_total, read_subvolume_usage, + send_agent_snapshot_to_file, set_subvolume_quota, snapshot_agent_subvolume, snapshot_path, + upgrade_agent_subvolume, validate_snapshot_name, }; use crate::shared::Spawn; use crate::shared::test_support::{MissingTool, ScriptSpawn, scratch}; @@ -1343,15 +1254,7 @@ mod tests { exec_send_agent_snapshot_to_file(bad, "hive-x", None, "out").await, "send-file", ); - want( - exec_send_agent_snapshot_to_fd(bad, "hive-x", None, None).await, - "send-fd", - ); } - let err = exec_send_agent_snapshot_to_fd("iris", "hive-x", None, None) - .await - .unwrap_err(); - assert!(err.to_string().contains("no descriptor"), "{err}"); } #[tokio::test] @@ -1379,14 +1282,6 @@ mod tests { exec_send_agent_snapshot_to_file("iris", "hive-x", Some(bad), "out").await, "send-file parent", ); - want( - exec_send_agent_snapshot_to_fd("iris", bad, None, None).await, - "send-fd snapshot", - ); - want( - exec_send_agent_snapshot_to_fd("iris", "hive-x", Some(bad), None).await, - "send-fd parent", - ); } for bad in ["", ".", "..", "../x", "a/b", "a.b", "/etc/passwd"] { let err = exec_send_agent_snapshot_to_file("iris", "hive-x", None, bad) @@ -2052,61 +1947,4 @@ esac"#; assert!(err.to_string().contains("does not exist"), "{err}"); assert!(!fx.staging.join("none").exists()); } - - fn fd_for(path: &Path) -> std::os::fd::OwnedFd { - std::os::fd::OwnedFd::from(std::fs::File::create(path).unwrap()) - } - - #[tokio::test] - async fn export_to_fd_streams_into_the_descriptor() { - let fx = Fx::new(); - snapshots(&fx, &["hive-a", "hive-p"]); - let btrfs = fx.btrfs(LIKE_BTRFS); - let env = fx.env(&btrfs, on_btrfs, not_subvol); - let sink = fx.state.join("sink"); - - let (out, err) = - send_agent_snapshot_to_fd(&env, "iris", "hive-a", Some("hive-p"), fd_for(&sink)) - .await - .unwrap(); - assert!(out.is_empty() && err.is_empty()); - assert_eq!(std::fs::read(&sink).unwrap(), b"STREAM"); - send_agent_snapshot_to_fd(&env, "iris", "hive-a", None, fd_for(&sink)) - .await - .unwrap(); - let (a, p) = (fx.snap("iris", "hive-a"), fx.snap("iris", "hive-p")); - assert_eq!( - fx.calls(), - [ - format!("send -p {} {}", p.display(), a.display()), - format!("send {}", a.display()), - ] - ); - } - - #[tokio::test] - async fn export_to_fd_needs_both_snapshots_and_reports_a_failed_send() { - let fx = Fx::new(); - snapshots(&fx, &["hive-a"]); - let sink = fx.state.join("sink"); - let btrfs = fx.btrfs(LIKE_BTRFS); - let env = fx.env(&btrfs, on_btrfs, not_subvol); - - let err = send_agent_snapshot_to_fd(&env, "iris", "hive-no", None, fd_for(&sink)) - .await - .unwrap_err(); - assert!(err.to_string().contains("does not exist"), "{err}"); - let err = send_agent_snapshot_to_fd(&env, "iris", "hive-a", Some("hive-no"), fd_for(&sink)) - .await - .unwrap_err(); - assert!(err.to_string().contains("parent snapshot"), "{err}"); - assert!(fx.calls().is_empty()); - - let failing = fx.btrfs(FAIL); - let env = fx.env(&failing, on_btrfs, not_subvol); - let err = send_agent_snapshot_to_fd(&env, "iris", "hive-a", None, fd_for(&sink)) - .await - .unwrap_err(); - assert!(err.to_string().contains("boom"), "{err}"); - } }