feat(#2862): receive a passed descriptor and stream a snapshot into it
hive-priv read requests with BufReader::lines, which cannot surface SCM_RIGHTS: ancillary data is attached to one specific recvmsg call, so a buffered line reader takes the bytes and silently drops the descriptor. Replace it with a recvmsg loop. The pairing is deliberately trivial. hive-sock-client connects per request, so a connection carries one line and at most one descriptor; a second descriptor arriving before its line is a protocol error rather than something to queue. check_fd_agreement rejects both mismatches -- an fd-taking op that got none, and a descriptor sent to an op that takes none -- and dropping the OwnedFd on that path closes it. recv_with_fds claims every descriptor the kernel attaches, including ones this protocol never expects, because an fd we fail to claim leaks for the life of the process. MSG_CMSG_CLOEXEC keeps a received descriptor out of every btrfs and nixos-container child. The control buffer is only cmsghdr-aligned, so descriptors are copied out byte-wise instead of read through a more strictly aligned pointer. SendAgentSnapshotToFd is SendAgentSnapshotToFile without the staging file: same validation and -p parent handling, stdout wired to the passed descriptor. It exists so hive-c0re can connect to a peer hive's snapshot store, write the header itself, and hand over the connected socket -- leaving this helper with no address, no protocol, and nobody in the data path once the send starts.
This commit is contained in:
atlas
mara
parent
364bc290df
commit
51f352f0ca
2 changed files with 319 additions and 21 deletions
|
|
@ -17,6 +17,7 @@
|
|||
//! `LISTEN_FDS=1` + `LISTEN_PID=<self>`, 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<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 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::<PrivRequest>(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<OwnedFd>, 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<OwnedFd>,
|
||||
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
|
||||
}
|
||||
|
||||
/// 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<OwnedFd>,
|
||||
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 <bytes|none> <…/agent_name>`). See the
|
||||
/// wire doc.
|
||||
|
|
|
|||
Loading…
Reference in a new issue