mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

hive-priv: drop the descriptor-taking snapshot send and its fd-receiving loop

Browse source
This commit is contained in:
atlas 2026-11-09 23:59:57 +01:00 • committed by mara
commit 418b50980c
4 changed files with 39 additions and 750 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/4-implementation/trust-boundary/hive-c0re-privilege.md
View file

@ -57,7 +57,6 @@ known operations; there is no arbitrary command pass-through:
| `SnapshotAgentSubvolume` | `btrfs subvolume snapshot -r <agent_root> <snapshot_path>` — the frozen point-in-time copy a migration streams from |
| `DeleteAgentSnapshot` | `btrfs subvolume delete <snapshot_path>` |
| `SendAgentSnapshotToFile` | `btrfs send [-p <parent>] <snapshot>` into a bare filename under `MIGRATE_STAGING_ROOT`; refuses to overwrite an existing export |
| `SendAgentSnapshotToFd` | `btrfs send [-p <parent>] <snapshot>` 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

28
hive-priv-sock/src/lib.rs
View file

@ -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 <parent>]
/// <snapshot> >&<passed fd>`.
///
/// 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<String>,
},
/// 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

586
hive-priv/src/main.rs
View file

@ -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<UnixListener> {
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<OwnedFd>)> {
const FD_SIZE: usize = std::mem::size_of::<RawFd>();
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<u8>,
fd: Option<OwnedFd>,
}
impl Requests<'_> {
/// Next complete request line and its descriptor, or `None` at EOF.
async fn next(&mut self) -> Result<Option<(String, Option<OwnedFd>)>> {
loop {
if let Some(nl) = self.buf.iter().position(|&b| b == b'\n') {
let line: Vec<u8> = 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<OwnedFd>, 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<OwnedFd>,
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::<PrivRequest>(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<OwnedFd>,
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::<PrivRequest>(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::<RawFd>();
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::<u8>(),
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<OwnedFd>) -> 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")
);
}
}

174
hive-priv/src/storage.rs
View file

@ -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<OwnedFd>,
) -> 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 <bytes|none> <…/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}");
}
}