From ec4ba4c7fa5f4bf246bb102396f4553ca280d1cc Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 31 Jul 2026 20:37:19 +0200 Subject: [PATCH 1/7] feat(#2862): fd-carrying line framing for the priv socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First half of the fd-passing work, and deliberately the half with the real failure mode in it. No syscalls here — the caller does the recvmsg and feeds this (bytes, fds); it hands back complete messages paired with the descriptor each one owns. Association is the whole point. A descriptor does not arrive neatly paired with the request that wants it: recvmsg returns whatever bytes happen to be available plus whatever ancillary data rode along, so a descriptor can arrive with a chunk holding only part of its request's line, with a chunk whose bytes finish the previous request, ahead of any of its own bytes, or alongside several complete requests at once. Pairing "the fd from this chunk" with "the request in this chunk" is therefore wrong in the worst way: the types are identical either way, so nothing catches it, and the failure is one request executing against another's descriptor — in this process, writing one agent's state into a different transfer's socket. So descriptors queue on arrival and each message claims the oldest unclaimed one at the moment it completes. Two consequences worth stating: a line that fails to decode does NOT consume a descriptor (closing it there would destroy something belonging to a request nobody processed), and unclaimed descriptors are drainable so the teardown path can close them instead of leaking one per abandoned message in a long-lived helper. Lives in hive-priv-sock, not hive-priv: clippy's dead-code error was right that an unwired module doesn't belong in the binary, and chasing that produced the better home anyway — both ends need this. The daemon sends descriptors and the helper reassembles them, so framing is part of the wire contract rather than one side's implementation detail. --- hive-priv-sock/src/framing.rs | 279 ++++++++++++++++++++++++++++++++++ hive-priv-sock/src/lib.rs | 6 + 2 files changed, 285 insertions(+) create mode 100644 hive-priv-sock/src/framing.rs diff --git a/hive-priv-sock/src/framing.rs b/hive-priv-sock/src/framing.rs new file mode 100644 index 00000000..1670e3fc --- /dev/null +++ b/hive-priv-sock/src/framing.rs @@ -0,0 +1,279 @@ +//! Line framing that also carries passed file descriptors. +//! +//! The privileged helper speaks newline-delimited JSON. Passing a file +//! descriptor (so a caller can hand us a connected socket to write a +//! `btrfs send` stream into, without us ever learning what it connects +//! to) means `recvmsg` instead of a plain `read`, because `SCM_RIGHTS` +//! ancillary data is attached to one specific `recvmsg` call and a +//! buffered line reader cannot surface it. +//! +//! It lives in the protocol crate rather than in `hive-priv` because +//! both ends need it: the helper reassembles what the daemon sends, so +//! framing is part of the wire contract, not an implementation detail +//! of one side. +//! +//! Everything subtle about that lives here, deliberately kept free of +//! syscalls so it can be tested exhaustively: the caller does the +//! `recvmsg` and feeds us `(bytes, fds)`; we hand back complete +//! messages with the descriptor that belongs to each. See [`Framer`] +//! for why that pairing is not the obvious one. +use std::collections::VecDeque; +use std::os::fd::OwnedFd; + +/// One complete request line plus the descriptor it claimed, if any. +pub struct Framed { + /// The message body — a JSON line, without its trailing newline. + pub line: String, + /// The descriptor this message claimed, oldest-first. `None` when + /// no unclaimed descriptor was pending, which is the ordinary case + /// for every request that doesn't pass one. + pub fd: Option, +} + +/// Reassembles `(bytes, fds)` chunks into complete messages. +/// +/// Bytes accumulate until a newline; descriptors queue independently. +/// A message claims the oldest queued descriptor at the moment the +/// message *completes*, never at the moment a chunk arrives. +/// +/// That indirection is the point. A descriptor does *not* arrive +/// neatly paired with the request that wants it — it can arrive with a +/// chunk holding only part of its request's line, with a chunk whose +/// bytes finish the *previous* request, ahead of any of its own bytes, +/// or alongside several complete requests at once. +/// +/// So pairing "the fd from this chunk" with "the request in this +/// chunk" is wrong, and wrong in the worst way: the types are +/// identical either way, so nothing catches it, and the failure is one +/// request executing against another's descriptor — here, writing one +/// agent's state into a different transfer's socket. +#[derive(Default)] +pub struct Framer { + buf: Vec, + pending_fds: VecDeque, +} + +impl Framer { + /// Feed one `recvmsg` result: the bytes it read and any descriptors + /// it carried. + /// + /// Returns every message completed by this chunk, in order. A chunk + /// can complete several (or none). + /// + /// Invalid UTF-8 in a line is surfaced as an error rather than + /// lossily replaced: the body is JSON, and a mangled line should + /// fail loudly instead of parsing into something adjacent. + pub fn push(&mut self, bytes: &[u8], fds: Vec) -> Vec> { + self.pending_fds.extend(fds); + self.buf.extend_from_slice(bytes); + + let mut out = Vec::new(); + while let Some(nl) = self.buf.iter().position(|&b| b == b'\n') { + let line_bytes: Vec = self.buf.drain(..=nl).take(nl).collect(); + match String::from_utf8(line_bytes) { + Ok(line) => out.push(Ok(Framed { + line, + fd: self.pending_fds.pop_front(), + })), + // The descriptor is deliberately NOT claimed for a line + // we can't read: dropping it here would close a + // descriptor that belongs to a request we never + // processed. It stays queued for the next complete + // message, and `drain_pending_fds` closes whatever is + // left when the connection ends. + Err(e) => out.push(Err(FramingError::InvalidUtf8(e.utf8_error().to_string()))), + } + } + out + } + + /// How many descriptors have arrived but not yet been claimed. + /// + /// Exposed for the connection teardown path: anything still here + /// when a connection ends belongs to no request and must be closed, + /// or a long-lived helper leaks a descriptor per abandoned message. + pub fn pending_fd_count(&self) -> usize { + self.pending_fds.len() + } + + /// Take every unclaimed descriptor so the caller can drop (close) + /// them explicitly at teardown. + pub fn drain_pending_fds(&mut self) -> Vec { + self.pending_fds.drain(..).collect() + } + + /// Bytes buffered without a terminating newline yet. + /// + /// A connection that ends here sent a partial request; the caller + /// should treat that as a truncated message rather than silently + /// discarding it. + pub fn partial_len(&self) -> usize { + self.buf.len() + } +} + +/// A message that could not be framed. +#[derive(Debug)] +pub enum FramingError { + /// The line was not valid UTF-8. Carries the underlying reason as + /// a string so the error type stays trivially cloneable/loggable. + InvalidUtf8(String), +} + +impl std::fmt::Display for FramingError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::InvalidUtf8(e) => write!(f, "request line was not valid UTF-8: {e}"), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::fd::{AsRawFd as _, OwnedFd}; + + /// A real, owned descriptor to pass around. `/dev/null` is opened + /// rather than faked so the tests exercise genuine `OwnedFd` move + /// semantics — a fake would not catch a double-close. + fn some_fd() -> OwnedFd { + std::fs::File::open("/dev/null") + .expect("open /dev/null") + .into() + } + + fn ok(res: Result) -> Framed { + res.expect("expected a framed message") + } + + #[test] + fn line_without_fd_claims_nothing() { + let mut f = Framer::default(); + let out = f.push(b"{\"a\":1}\n", vec![]); + assert_eq!(out.len(), 1); + let m = ok(out.into_iter().next().unwrap()); + assert_eq!(m.line, "{\"a\":1}"); + assert!( + m.fd.is_none(), + "no descriptor was sent, none should be claimed" + ); + } + + #[test] + fn fd_arriving_with_a_partial_line_is_claimed_by_that_line() { + // The case the whole type exists for: the descriptor rides in + // on a chunk that does not complete any message. + let mut f = Framer::default(); + let raw = { + let fd = some_fd(); + let raw = fd.as_raw_fd(); + let out = f.push(b"{\"partial\"", vec![fd]); + assert!(out.is_empty(), "no newline yet, so no message yet"); + raw + }; + assert_eq!(f.pending_fd_count(), 1, "descriptor waits for its message"); + + let out = f.push(b":true}\n", vec![]); + assert_eq!(out.len(), 1); + let m = ok(out.into_iter().next().unwrap()); + assert_eq!(m.line, "{\"partial\":true}"); + assert_eq!( + m.fd.expect("the completed message must claim it") + .as_raw_fd(), + raw + ); + } + + #[test] + fn fd_does_not_leak_to_a_neighbouring_message_in_the_same_chunk() { + // Two complete messages arrive together with ONE descriptor. + // It belongs to the first; the second must get nothing rather + // than inheriting it. + let mut f = Framer::default(); + let out = f.push(b"{\"first\":1}\n{\"second\":2}\n", vec![some_fd()]); + assert_eq!(out.len(), 2); + let mut it = out.into_iter(); + let first = ok(it.next().unwrap()); + let second = ok(it.next().unwrap()); + assert_eq!(first.line, "{\"first\":1}"); + assert!(first.fd.is_some(), "oldest message claims the descriptor"); + assert_eq!(second.line, "{\"second\":2}"); + assert!( + second.fd.is_none(), + "a second message must not inherit the first's descriptor" + ); + } + + #[test] + fn descriptors_are_claimed_oldest_first() { + let mut f = Framer::default(); + let (a, b) = (some_fd(), some_fd()); + let (ra, rb) = (a.as_raw_fd(), b.as_raw_fd()); + f.push(b"", vec![a]); + f.push(b"", vec![b]); + let out = f.push(b"one\ntwo\n", vec![]); + assert_eq!(out.len(), 2); + let mut it = out.into_iter(); + assert_eq!(ok(it.next().unwrap()).fd.unwrap().as_raw_fd(), ra); + assert_eq!(ok(it.next().unwrap()).fd.unwrap().as_raw_fd(), rb); + } + + #[test] + fn a_chunk_may_complete_nothing() { + let mut f = Framer::default(); + assert!(f.push(b"no newline here", vec![]).is_empty()); + assert_eq!( + f.partial_len(), + 15, + "bytes stay buffered for the next chunk" + ); + } + + #[test] + fn invalid_utf8_errors_without_consuming_a_descriptor() { + // A line we cannot read must not swallow a descriptor: closing + // it here would destroy something belonging to a request that + // was never processed. + let mut f = Framer::default(); + let out = f.push(b"\xff\xfe\n", vec![some_fd()]); + assert_eq!(out.len(), 1); + assert!(matches!( + out.into_iter().next().unwrap(), + Err(FramingError::InvalidUtf8(_)) + )); + assert_eq!( + f.pending_fd_count(), + 1, + "the descriptor must survive a bad line, not be consumed by it" + ); + } + + #[test] + fn unclaimed_descriptors_are_recoverable_for_teardown() { + // Whatever is still queued when a connection ends belongs to no + // request. The caller has to be able to get it back and close + // it, or a long-lived helper leaks one per abandoned message. + let mut f = Framer::default(); + f.push(b"", vec![some_fd(), some_fd()]); + assert_eq!(f.pending_fd_count(), 2); + assert_eq!(f.drain_pending_fds().len(), 2); + assert_eq!(f.pending_fd_count(), 0); + } + + #[test] + fn empty_line_is_a_message_not_a_skip() { + // A bare newline is a zero-length request. It should surface as + // an (invalid) message for the dispatcher to reject, rather + // than being silently dropped here — silently dropping it would + // desynchronise descriptor claiming. + let mut f = Framer::default(); + let fd_present = { + let out = f.push(b"\n", vec![some_fd()]); + assert_eq!(out.len(), 1); + let m = ok(out.into_iter().next().unwrap()); + assert_eq!(m.line, ""); + m.fd.is_some() + }; + assert!(fd_present, "an empty line still claims its descriptor"); + } +} diff --git a/hive-priv-sock/src/lib.rs b/hive-priv-sock/src/lib.rs index fcb71318..32c85fe0 100644 --- a/hive-priv-sock/src/lib.rs +++ b/hive-priv-sock/src/lib.rs @@ -12,6 +12,12 @@ use serde::{Deserialize, Serialize}; +// Line framing that also carries passed file descriptors. Part of the +// wire contract rather than one side's implementation detail: the +// daemon sends descriptors and the helper reassembles them, so both +// ends have to agree on how a descriptor is bound to a request. +pub mod framing; + /// Default socket path for the privileged helper. pub const PRIV_SOCK: &str = "/run/hive/priv.sock"; From 364bc290dfb77afe54cb14bfc99ab1ff63e36567 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 31 Jul 2026 21:01:17 +0200 Subject: [PATCH 2/7] refactor(#2862): drop the fd framing module, the hazard is unreachable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Framer bound a passed descriptor to the request line it belongs to, on the premise that several requests can be in flight on one connection so a descriptor could arrive with a chunk belonging to a different one. That premise is false. hive-sock-client::try_once connects per request (connect, write one line, read one line, drop) and priv_client's two connect sites each open their own stream, so a connection carries exactly one request: one line, at most one descriptor, nothing to disambiguate. Request and response align by connection. Delete it rather than move it. The recvmsg swap still has to happen — SCM_RIGHTS is attached to a specific recvmsg call and BufReader::lines cannot surface it — but the pairing it needs is "take the descriptor that arrived with this line", not a queue and a claim policy. --- hive-priv-sock/src/framing.rs | 279 ---------------------------------- hive-priv-sock/src/lib.rs | 6 - 2 files changed, 285 deletions(-) delete mode 100644 hive-priv-sock/src/framing.rs diff --git a/hive-priv-sock/src/framing.rs b/hive-priv-sock/src/framing.rs deleted file mode 100644 index 1670e3fc..00000000 --- a/hive-priv-sock/src/framing.rs +++ /dev/null @@ -1,279 +0,0 @@ -//! Line framing that also carries passed file descriptors. -//! -//! The privileged helper speaks newline-delimited JSON. Passing a file -//! descriptor (so a caller can hand us a connected socket to write a -//! `btrfs send` stream into, without us ever learning what it connects -//! to) means `recvmsg` instead of a plain `read`, because `SCM_RIGHTS` -//! ancillary data is attached to one specific `recvmsg` call and a -//! buffered line reader cannot surface it. -//! -//! It lives in the protocol crate rather than in `hive-priv` because -//! both ends need it: the helper reassembles what the daemon sends, so -//! framing is part of the wire contract, not an implementation detail -//! of one side. -//! -//! Everything subtle about that lives here, deliberately kept free of -//! syscalls so it can be tested exhaustively: the caller does the -//! `recvmsg` and feeds us `(bytes, fds)`; we hand back complete -//! messages with the descriptor that belongs to each. See [`Framer`] -//! for why that pairing is not the obvious one. -use std::collections::VecDeque; -use std::os::fd::OwnedFd; - -/// One complete request line plus the descriptor it claimed, if any. -pub struct Framed { - /// The message body — a JSON line, without its trailing newline. - pub line: String, - /// The descriptor this message claimed, oldest-first. `None` when - /// no unclaimed descriptor was pending, which is the ordinary case - /// for every request that doesn't pass one. - pub fd: Option, -} - -/// Reassembles `(bytes, fds)` chunks into complete messages. -/// -/// Bytes accumulate until a newline; descriptors queue independently. -/// A message claims the oldest queued descriptor at the moment the -/// message *completes*, never at the moment a chunk arrives. -/// -/// That indirection is the point. A descriptor does *not* arrive -/// neatly paired with the request that wants it — it can arrive with a -/// chunk holding only part of its request's line, with a chunk whose -/// bytes finish the *previous* request, ahead of any of its own bytes, -/// or alongside several complete requests at once. -/// -/// So pairing "the fd from this chunk" with "the request in this -/// chunk" is wrong, and wrong in the worst way: the types are -/// identical either way, so nothing catches it, and the failure is one -/// request executing against another's descriptor — here, writing one -/// agent's state into a different transfer's socket. -#[derive(Default)] -pub struct Framer { - buf: Vec, - pending_fds: VecDeque, -} - -impl Framer { - /// Feed one `recvmsg` result: the bytes it read and any descriptors - /// it carried. - /// - /// Returns every message completed by this chunk, in order. A chunk - /// can complete several (or none). - /// - /// Invalid UTF-8 in a line is surfaced as an error rather than - /// lossily replaced: the body is JSON, and a mangled line should - /// fail loudly instead of parsing into something adjacent. - pub fn push(&mut self, bytes: &[u8], fds: Vec) -> Vec> { - self.pending_fds.extend(fds); - self.buf.extend_from_slice(bytes); - - let mut out = Vec::new(); - while let Some(nl) = self.buf.iter().position(|&b| b == b'\n') { - let line_bytes: Vec = self.buf.drain(..=nl).take(nl).collect(); - match String::from_utf8(line_bytes) { - Ok(line) => out.push(Ok(Framed { - line, - fd: self.pending_fds.pop_front(), - })), - // The descriptor is deliberately NOT claimed for a line - // we can't read: dropping it here would close a - // descriptor that belongs to a request we never - // processed. It stays queued for the next complete - // message, and `drain_pending_fds` closes whatever is - // left when the connection ends. - Err(e) => out.push(Err(FramingError::InvalidUtf8(e.utf8_error().to_string()))), - } - } - out - } - - /// How many descriptors have arrived but not yet been claimed. - /// - /// Exposed for the connection teardown path: anything still here - /// when a connection ends belongs to no request and must be closed, - /// or a long-lived helper leaks a descriptor per abandoned message. - pub fn pending_fd_count(&self) -> usize { - self.pending_fds.len() - } - - /// Take every unclaimed descriptor so the caller can drop (close) - /// them explicitly at teardown. - pub fn drain_pending_fds(&mut self) -> Vec { - self.pending_fds.drain(..).collect() - } - - /// Bytes buffered without a terminating newline yet. - /// - /// A connection that ends here sent a partial request; the caller - /// should treat that as a truncated message rather than silently - /// discarding it. - pub fn partial_len(&self) -> usize { - self.buf.len() - } -} - -/// A message that could not be framed. -#[derive(Debug)] -pub enum FramingError { - /// The line was not valid UTF-8. Carries the underlying reason as - /// a string so the error type stays trivially cloneable/loggable. - InvalidUtf8(String), -} - -impl std::fmt::Display for FramingError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::InvalidUtf8(e) => write!(f, "request line was not valid UTF-8: {e}"), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::os::fd::{AsRawFd as _, OwnedFd}; - - /// A real, owned descriptor to pass around. `/dev/null` is opened - /// rather than faked so the tests exercise genuine `OwnedFd` move - /// semantics — a fake would not catch a double-close. - fn some_fd() -> OwnedFd { - std::fs::File::open("/dev/null") - .expect("open /dev/null") - .into() - } - - fn ok(res: Result) -> Framed { - res.expect("expected a framed message") - } - - #[test] - fn line_without_fd_claims_nothing() { - let mut f = Framer::default(); - let out = f.push(b"{\"a\":1}\n", vec![]); - assert_eq!(out.len(), 1); - let m = ok(out.into_iter().next().unwrap()); - assert_eq!(m.line, "{\"a\":1}"); - assert!( - m.fd.is_none(), - "no descriptor was sent, none should be claimed" - ); - } - - #[test] - fn fd_arriving_with_a_partial_line_is_claimed_by_that_line() { - // The case the whole type exists for: the descriptor rides in - // on a chunk that does not complete any message. - let mut f = Framer::default(); - let raw = { - let fd = some_fd(); - let raw = fd.as_raw_fd(); - let out = f.push(b"{\"partial\"", vec![fd]); - assert!(out.is_empty(), "no newline yet, so no message yet"); - raw - }; - assert_eq!(f.pending_fd_count(), 1, "descriptor waits for its message"); - - let out = f.push(b":true}\n", vec![]); - assert_eq!(out.len(), 1); - let m = ok(out.into_iter().next().unwrap()); - assert_eq!(m.line, "{\"partial\":true}"); - assert_eq!( - m.fd.expect("the completed message must claim it") - .as_raw_fd(), - raw - ); - } - - #[test] - fn fd_does_not_leak_to_a_neighbouring_message_in_the_same_chunk() { - // Two complete messages arrive together with ONE descriptor. - // It belongs to the first; the second must get nothing rather - // than inheriting it. - let mut f = Framer::default(); - let out = f.push(b"{\"first\":1}\n{\"second\":2}\n", vec![some_fd()]); - assert_eq!(out.len(), 2); - let mut it = out.into_iter(); - let first = ok(it.next().unwrap()); - let second = ok(it.next().unwrap()); - assert_eq!(first.line, "{\"first\":1}"); - assert!(first.fd.is_some(), "oldest message claims the descriptor"); - assert_eq!(second.line, "{\"second\":2}"); - assert!( - second.fd.is_none(), - "a second message must not inherit the first's descriptor" - ); - } - - #[test] - fn descriptors_are_claimed_oldest_first() { - let mut f = Framer::default(); - let (a, b) = (some_fd(), some_fd()); - let (ra, rb) = (a.as_raw_fd(), b.as_raw_fd()); - f.push(b"", vec![a]); - f.push(b"", vec![b]); - let out = f.push(b"one\ntwo\n", vec![]); - assert_eq!(out.len(), 2); - let mut it = out.into_iter(); - assert_eq!(ok(it.next().unwrap()).fd.unwrap().as_raw_fd(), ra); - assert_eq!(ok(it.next().unwrap()).fd.unwrap().as_raw_fd(), rb); - } - - #[test] - fn a_chunk_may_complete_nothing() { - let mut f = Framer::default(); - assert!(f.push(b"no newline here", vec![]).is_empty()); - assert_eq!( - f.partial_len(), - 15, - "bytes stay buffered for the next chunk" - ); - } - - #[test] - fn invalid_utf8_errors_without_consuming_a_descriptor() { - // A line we cannot read must not swallow a descriptor: closing - // it here would destroy something belonging to a request that - // was never processed. - let mut f = Framer::default(); - let out = f.push(b"\xff\xfe\n", vec![some_fd()]); - assert_eq!(out.len(), 1); - assert!(matches!( - out.into_iter().next().unwrap(), - Err(FramingError::InvalidUtf8(_)) - )); - assert_eq!( - f.pending_fd_count(), - 1, - "the descriptor must survive a bad line, not be consumed by it" - ); - } - - #[test] - fn unclaimed_descriptors_are_recoverable_for_teardown() { - // Whatever is still queued when a connection ends belongs to no - // request. The caller has to be able to get it back and close - // it, or a long-lived helper leaks one per abandoned message. - let mut f = Framer::default(); - f.push(b"", vec![some_fd(), some_fd()]); - assert_eq!(f.pending_fd_count(), 2); - assert_eq!(f.drain_pending_fds().len(), 2); - assert_eq!(f.pending_fd_count(), 0); - } - - #[test] - fn empty_line_is_a_message_not_a_skip() { - // A bare newline is a zero-length request. It should surface as - // an (invalid) message for the dispatcher to reject, rather - // than being silently dropped here — silently dropping it would - // desynchronise descriptor claiming. - let mut f = Framer::default(); - let fd_present = { - let out = f.push(b"\n", vec![some_fd()]); - assert_eq!(out.len(), 1); - let m = ok(out.into_iter().next().unwrap()); - assert_eq!(m.line, ""); - m.fd.is_some() - }; - assert!(fd_present, "an empty line still claims its descriptor"); - } -} diff --git a/hive-priv-sock/src/lib.rs b/hive-priv-sock/src/lib.rs index 32c85fe0..fcb71318 100644 --- a/hive-priv-sock/src/lib.rs +++ b/hive-priv-sock/src/lib.rs @@ -12,12 +12,6 @@ use serde::{Deserialize, Serialize}; -// Line framing that also carries passed file descriptors. Part of the -// wire contract rather than one side's implementation detail: the -// daemon sends descriptors and the helper reassembles them, so both -// ends have to agree on how a descriptor is bound to a request. -pub mod framing; - /// Default socket path for the privileged helper. pub const PRIV_SOCK: &str = "/run/hive/priv.sock"; From 51f352f0ca59adc33ed973fa67ebaecaac81f149 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 31 Jul 2026 21:11:51 +0200 Subject: [PATCH 3/7] 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. --- hive-priv-sock/src/lib.rs | 29 ++++ hive-priv/src/main.rs | 311 +++++++++++++++++++++++++++++++++++--- 2 files changed, 319 insertions(+), 21 deletions(-) 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. From 282bbc3709f2c489cfad8c81b0926d3ea47208d9 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 31 Jul 2026 21:40:34 +0200 Subject: [PATCH 4/7] feat(#2862): push a snapshot to a peer hive's store over the mesh Adds the caller the fd-passing machinery existed for: hivectl agent subvol snapshot push --peer resolves the peer, connects to its snapshot store, writes the agent header, and hands the connected socket to hive-priv, which runs btrfs send straight into it. The split keeps the root helper ignorant. Everything that involves knowing where a peer is, what the wire protocol looks like, and which hive to trust happens in the unprivileged daemon; hive-priv only ever receives an already-open descriptor. Once btrfs send starts, neither process is in the data path, so a multi-gigabyte transfer costs no per-byte work and survives a hive-c0re restart. call_with_fd takes the descriptor by value and closes it as soon as the kernel has it. A socket stays open until every copy closes, so holding one back would leave the receiver waiting for an EOF that never comes: btrfs receive blocks and this side reports success for a transfer the peer never committed. Ownership makes that unrepresentable. The peer's store port is a new swarm.peers..snapshotStorePort option rather than a constant matching the module default. A pushing hive cannot read the receiver's configuration, so assuming 51821 would push at a port nobody promised to listen on; absent, the push fails naming the option. swarm_peers parses the mesh address the host module has always rendered into HYPERHIVE_PEERS but nothing read. --- hive-c0re/src/main.rs | 2 + hive-c0re/src/priv_client.rs | 138 ++++++++++++++- hive-c0re/src/server.rs | 21 +++ hive-c0re/src/snapshot_push.rs | 89 ++++++++++ hive-c0re/src/swarm_peers.rs | 188 +++++++++++++++++++++ hive-host-sock/src/lib.rs | 17 ++ hivectl/src/cli.rs | 31 +++- hivectl/src/subvol.rs | 36 ++++ nix/host-modules/hive-c0re/environment.nix | 14 +- nix/host-modules/swarm.nix | 19 +++ 10 files changed, 546 insertions(+), 9 deletions(-) create mode 100644 hive-c0re/src/snapshot_push.rs create mode 100644 hive-c0re/src/swarm_peers.rs diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index c6264986..bc3fd011 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -31,9 +31,11 @@ mod paths; mod priv_client; mod questions; mod server; +mod snapshot_push; mod socket_server; mod stats; mod stores; +mod swarm_peers; mod webhook_secret; mod workers; diff --git a/hive-c0re/src/priv_client.rs b/hive-c0re/src/priv_client.rs index 2ce0020a..4663db86 100644 --- a/hive-c0re/src/priv_client.rs +++ b/hive-c0re/src/priv_client.rs @@ -11,7 +11,9 @@ use hive_priv_sock::{ BindMount, CredentialMount, InfraAction, InfraContainer, JournalQuery, NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream, }; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use std::os::fd::{AsRawFd as _, OwnedFd, RawFd}; + +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Interest}; use tokio::net::UnixStream; /// Send a single request to `hive-priv` and return the response. @@ -40,6 +42,140 @@ pub async fn call(req: &PrivRequest) -> Result { } } +/// 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; the union member supplies the `cmsghdr` +/// alignment `CMSG_FIRSTHDR` requires. +#[repr(C)] +union CmsgSpace { + _align: libc::cmsghdr, + bytes: [u8; 32], +} + +/// `sendmsg` `bytes` with `fd` attached as `SCM_RIGHTS`, returning how +/// many bytes were accepted. +/// +/// The descriptor rides on this one call — ancillary data cannot be +/// sent separately from payload — so the caller must not have written +/// any of `bytes` beforehand. +fn send_with_fd(sock: RawFd, bytes: &[u8], fd: RawFd) -> std::io::Result { + const FD_SIZE: usize = std::mem::size_of::(); + + let mut iov = libc::iovec { + iov_base: bytes.as_ptr().cast::().cast_mut(), + iov_len: bytes.len(), + }; + let mut cmsg = CmsgSpace { bytes: [0; 32] }; + // SAFETY: msghdr is a plain C struct with no invalid bit patterns; + // every field we rely on 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(); + // SAFETY: CMSG_SPACE is a pure size computation. + let space = unsafe { libc::CMSG_SPACE(u32::try_from(FD_SIZE).unwrap_or(4)) }; + msg.msg_controllen = space as _; + + // SAFETY: the control buffer is live, aligned, and long enough for + // the header CMSG_SPACE just sized. + let hdr = unsafe { libc::CMSG_FIRSTHDR(&raw const msg) }; + if hdr.is_null() { + return Err(std::io::Error::other( + "control buffer too small for SCM_RIGHTS", + )); + } + // SAFETY: CMSG_LEN is a pure size computation; `hdr` points into + // our own buffer, and write_unaligned tolerates its alignment. + unsafe { + let len = libc::CMSG_LEN(u32::try_from(FD_SIZE).unwrap_or(4)); + std::ptr::write_unaligned( + hdr, + libc::cmsghdr { + cmsg_len: len as _, + cmsg_level: libc::SOL_SOCKET, + cmsg_type: libc::SCM_RIGHTS, + }, + ); + // Copied in byte-wise: the control buffer is only cmsghdr- + // aligned, so casting CMSG_DATA to a *mut RawFd would be + // unsound even where it happens to work. + std::ptr::copy_nonoverlapping( + std::ptr::from_ref(&fd).cast::(), + libc::CMSG_DATA(hdr), + FD_SIZE, + ); + } + + // SAFETY: `msg` points at a live iovec over `bytes` and the control + // buffer we just filled in. + let n = unsafe { libc::sendmsg(sock, &raw const msg, 0) }; + if n < 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(usize::try_from(n).unwrap_or_default()) +} + +/// Send a request to `hive-priv` with an open file descriptor attached, +/// and return the response. +/// +/// The helper receives the descriptor itself — not a path or an address +/// — so it can act on something it was handed without being told what +/// that thing is or how to reach it. Used for +/// [`PrivRequest::SendAgentSnapshotToFd`], where the descriptor is a +/// socket already connected to a peer hive's snapshot store. +/// +/// ⚠️ Takes the descriptor by value and closes it as soon as the kernel +/// has it, *before* awaiting the response. That is not tidiness: a +/// socket stays open until every copy of it is closed, so a caller +/// holding one back would leave the receiving end waiting for an EOF +/// that never comes — `btrfs receive` blocks, and this side reports +/// success for a transfer the peer has not committed. Passing ownership +/// makes that mistake unrepresentable. +/// +/// # Errors +/// +/// Fails if the socket is unreachable, the descriptor cannot be +/// attached, or hive-priv answers with something other than a terminal +/// event. +pub async fn call_with_fd(req: &PrivRequest, fd: OwnedFd) -> Result { + let mut stream = UnixStream::connect(PRIV_SOCK) + .await + .context("connect to hive-priv socket (fd-passing)")?; + let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n"; + let bytes = line.as_bytes(); + + let sock = stream.as_raw_fd(); + let raw_fd = fd.as_raw_fd(); + let sent = stream + .async_io(Interest::WRITABLE, || send_with_fd(sock, bytes, raw_fd)) + .await + .context("send request + descriptor to hive-priv")?; + // The kernel has duplicated the descriptor into hive-priv's queue, + // so our copy has done its job. Close it now, before waiting on the + // response: see the EOF note on this function. + drop(fd); + + // A short sendmsg is legal; the descriptor went with the first + // call, so the tail is an ordinary write. + if sent < bytes.len() { + stream + .write_all(&bytes[sent..]) + .await + .context("send remainder of request to hive-priv")?; + } + stream.shutdown().await.context("shutdown write half")?; + + let mut resp_line = String::new(); + BufReader::new(stream) + .read_line(&mut resp_line) + .await + .context("read response from hive-priv")?; + match serde_json::from_str::(&resp_line).context("parse PrivResponse")? { + PrivEvent::Done(resp) => Ok(resp), + PrivEvent::Line(_) => bail!("unexpected stream line from non-streaming priv op"), + } +} + /// Send a streaming request to `hive-priv`, calling `on_line` for each /// `PrivEvent::Line` as it arrives, then returning the terminal /// `PrivResponse`. Used for long-running ops (`create` / `update`). diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 93dba9f3..5e19962e 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -264,6 +264,12 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { parent, dest, } => handle_send_snapshot(name.as_str(), label, parent.as_deref(), dest).await?, + HostRequest::PushSnapshot { + name, + label, + parent, + peer, + } => handle_push_snapshot(name.as_str(), label, parent.as_deref(), peer).await?, }) } .await; @@ -661,6 +667,21 @@ async fn handle_send_snapshot( Ok(HostResponse::messages(vec![path])) } +/// Push a snapshot to a peer hive's store. The network sibling of +/// [`handle_send_snapshot`]: nothing lands on this host, so success is +/// bare rather than a path. +async fn handle_push_snapshot( + name: &str, + label: &str, + parent: Option<&str>, + peer: &str, +) -> Result { + crate::snapshot_push::push_agent_snapshot(name, label, parent, peer) + .await + .with_context(|| format!("push {name} snapshot (label {label:?}) to peer {peer:?}"))?; + Ok(HostResponse::success()) +} + async fn handle_matrix_sync_admin() -> Result { require_matrix_present().await?; let register_token = diff --git a/hive-c0re/src/snapshot_push.rs b/hive-c0re/src/snapshot_push.rs new file mode 100644 index 00000000..75a3ef0a --- /dev/null +++ b/hive-c0re/src/snapshot_push.rs @@ -0,0 +1,89 @@ +//! Push an agent snapshot to a peer hive's snapshot store. +//! +//! This is the c0re half of the transport: it resolves the peer, opens +//! the connection, writes the protocol header, and hands the connected +//! socket to `hive-priv`, which runs `btrfs send` straight into it. +//! +//! The split is the point. Everything that requires knowing *where* the +//! peer is and *what* the wire protocol looks like happens here, in the +//! unprivileged daemon. hive-priv only ever receives an already-open +//! descriptor, so the root helper never learns an address, never parses +//! a peer list, and never decides who to trust — and once `btrfs send` +//! starts, neither process is in the data path. + +use anyhow::{Context as _, Result, bail}; +use hive_priv_sock::PrivRequest; +use tokio::io::AsyncWriteExt as _; +use tokio::net::TcpStream; + +use crate::priv_client; +use crate::swarm_peers; + +/// Send `snapshot` of `agent` to `peer_domain`'s snapshot store, +/// optionally as an incremental against `parent`. +/// +/// # Errors +/// +/// Fails when the peer is unknown, declares no mesh address or store +/// port, is unreachable, or when hive-priv reports the `btrfs send` +/// failed. +pub async fn push_agent_snapshot( + agent: &str, + snapshot: &str, + parent: Option<&str>, + peer_domain: &str, +) -> Result<()> { + let peer = swarm_peers::peer(peer_domain)?; + let addr = peer.snapshot_store_addr()?; + + let mut sock = TcpStream::connect(&addr) + .await + .with_context(|| format!("connect to {peer_domain} snapshot store at {addr}"))?; + + // The header names which agent this stream belongs to: a btrfs + // stream carries the sender's own subvolume name, not the hive's + // notion of the agent, so the receiver cannot infer it. One line, + // because the receiver reads it with `read` and must not buffer + // past the newline into the byte stream. + sock.write_all(format!("agent {agent}\n").as_bytes()) + .await + .with_context(|| format!("send header to {peer_domain} snapshot store"))?; + sock.flush() + .await + .with_context(|| format!("flush header to {peer_domain} snapshot store"))?; + + // Hand the connected socket over. `call_with_fd` takes ownership and + // closes our copy as soon as the kernel has it, so the receiver sees + // EOF when `btrfs send` finishes rather than hanging on a descriptor + // this process is still holding. + let fd = sock + .into_std() + .context("detach snapshot-store socket for hand-off")? + .into(); + + let resp = priv_client::call_with_fd( + &PrivRequest::SendAgentSnapshotToFd { + agent_name: agent.to_owned(), + snapshot_name: snapshot.to_owned(), + parent_snapshot_name: parent.map(str::to_owned), + }, + fd, + ) + .await + .with_context(|| format!("stream {agent}/{snapshot} to {peer_domain}"))?; + + if !resp.ok { + bail!( + "push of {agent}/{snapshot} to {peer_domain} failed: {}", + resp.error.unwrap_or_else(|| resp.stderr.clone()) + ); + } + tracing::info!( + agent, + snapshot, + parent, + peer = peer_domain, + "pushed agent snapshot to peer snapshot store" + ); + Ok(()) +} diff --git a/hive-c0re/src/swarm_peers.rs b/hive-c0re/src/swarm_peers.rs new file mode 100644 index 00000000..aa2c8cdc --- /dev/null +++ b/hive-c0re/src/swarm_peers.rs @@ -0,0 +1,188 @@ +//! Peer-hive lookup for intra-swarm connections. +//! +//! Reads the `HYPERHIVE_PEERS` env var the host module renders from +//! `services.hyperhive.swarm.peers` and resolves a peer domain to the +//! address of a service running on that peer. +//! +//! Deliberately separate from the dashboard's `PeerHiveView`, which +//! parses the same variable: that type is shaped for rendering the +//! P33RS tab and drops the fields a connection needs. Nothing here is +//! defaulted — a peer that hasn't declared an address or a port is an +//! error naming what's missing, never a guess at a well-known value. +//! Ports and addresses are deployment facts, so they come from the nix +//! side or not at all. + +use anyhow::{Context as _, Result, bail}; +use serde::Deserialize; + +/// Env var carrying the serialised peer list. Rendered by +/// `nix/host-modules/hive-c0re/environment.nix`. +const PEERS_ENV: &str = "HYPERHIVE_PEERS"; + +/// One peer hive, as serialised into [`PEERS_ENV`]. +/// +/// Only the fields a connection needs are modelled; `cert_fingerprint` +/// and any later additions are ignored by serde rather than duplicated +/// from the dashboard's view type. +#[derive(Debug, Clone, Deserialize)] +pub struct SwarmPeer { + /// The peer's DNS domain — the attrset key on the nix side, and the + /// name an operator refers to the peer by. + pub domain: String, + /// The peer's address on the WireGuard mesh, with prefix length + /// (`10.100.0.2/32`). Absent when the peer isn't in the mesh. + #[serde(default)] + pub wireguard_address: Option, + /// TCP port of the peer's snapshot store. Absent when it runs none. + #[serde(default)] + pub snapshot_store_port: Option, +} + +impl SwarmPeer { + /// `host:port` for this peer's snapshot store. + /// + /// # Errors + /// + /// Fails when the peer is not in the mesh, or runs no snapshot + /// store. Both are configuration facts the pushing hive cannot + /// discover on its own, so they're reported rather than guessed. + pub fn snapshot_store_addr(&self) -> Result { + let Some(addr) = self.wireguard_address.as_deref() else { + bail!( + "peer {} has no wireguardAddress — it is not in the mesh, \ + and the mesh is the only route to a snapshot store", + self.domain + ); + }; + let Some(port) = self.snapshot_store_port else { + bail!( + "peer {} declares no snapshotStorePort — it hosts no snapshot store \ + (set services.hyperhive.swarm.peers.\"{}\".snapshotStorePort on this \ + host to match the receiver's services.hyperhive.snapshotStore.port)", + self.domain, + self.domain + ); + }; + // `wireguardAddress` is CIDR because WireGuard's `allowedIPs` + // wants it that way; a connect() wants the bare address. + let host = addr.split('/').next().unwrap_or(addr); + Ok(format!("{host}:{port}")) + } +} + +/// Parse the peer list out of `HYPERHIVE_PEERS`. +/// +/// # Errors +/// +/// Fails when the variable is unset (this host declares no peers, so it +/// is not in a swarm) or does not parse. +pub fn peers() -> Result> { + let raw = std::env::var(PEERS_ENV).with_context(|| { + format!("{PEERS_ENV} is unset — this host declares no swarm.peers, so it has no peers") + })?; + parse_peers(&raw) +} + +/// Find one peer by domain. +/// +/// # Errors +/// +/// Fails when no peer matches, naming the peers that do exist — an +/// operator typo is the likeliest cause and the list is short. +pub fn peer(domain: &str) -> Result { + let all = peers()?; + if let Some(found) = all.iter().find(|p| p.domain == domain) { + return Ok(found.clone()); + } + let known: Vec<&str> = all.iter().map(|p| p.domain.as_str()).collect(); + bail!("no swarm peer named {domain}; declared peers: {}", { + if known.is_empty() { + "(none)".to_owned() + } else { + known.join(", ") + } + }) +} + +/// Split out of [`peers`] so the parsing is testable without touching +/// process-wide environment, which would race other tests. +fn parse_peers(raw: &str) -> Result> { + serde_json::from_str(raw).with_context(|| format!("{PEERS_ENV} is not valid peer JSON")) +} + +#[cfg(test)] +mod tests { + use super::{SwarmPeer, parse_peers}; + + fn peer_with(addr: Option<&str>, port: Option) -> SwarmPeer { + SwarmPeer { + domain: "lab.example.com".to_owned(), + wireguard_address: addr.map(str::to_owned), + snapshot_store_port: port, + } + } + + #[test] + fn parses_the_rendered_shape_and_ignores_unknown_fields() { + // cert_fingerprint is in the real payload and irrelevant here; + // it must not break parsing. + let raw = r#"[{"domain":"lab.example.com","cert_fingerprint":null, + "wireguard_address":"10.100.0.2/32","snapshot_store_port":51821}]"#; + let peers = parse_peers(raw).expect("parse"); + assert_eq!(peers.len(), 1); + assert_eq!(peers[0].snapshot_store_port, Some(51821)); + } + + #[test] + fn absent_optional_fields_are_none_not_an_error() { + // A peer that is only a federation/dashboard link declares + // neither field; that must parse, and fail later with a + // specific message, rather than fail to parse at all. + let raw = r#"[{"domain":"edge.corp","cert_fingerprint":null}]"#; + let peers = parse_peers(raw).expect("parse"); + assert!(peers[0].wireguard_address.is_none()); + assert!(peers[0].snapshot_store_port.is_none()); + } + + #[test] + fn store_addr_strips_the_cidr_prefix() { + // allowedIPs wants `10.100.0.2/32`; connect() does not. + let addr = peer_with(Some("10.100.0.2/32"), Some(51821)) + .snapshot_store_addr() + .expect("addr"); + assert_eq!(addr, "10.100.0.2:51821"); + } + + #[test] + fn a_bare_address_without_a_prefix_still_works() { + let addr = peer_with(Some("10.100.0.2"), Some(51821)) + .snapshot_store_addr() + .expect("addr"); + assert_eq!(addr, "10.100.0.2:51821"); + } + + #[test] + fn no_mesh_address_names_the_mesh_as_the_problem() { + let err = peer_with(None, Some(51821)) + .snapshot_store_addr() + .expect_err("a peer off the mesh has no route"); + let msg = format!("{err:#}"); + assert!(msg.contains("wireguardAddress"), "{msg}"); + } + + #[test] + fn no_port_is_an_error_rather_than_the_module_default() { + // The receiving module defaults to 51821, but this host cannot + // read the receiver's config: assuming the default here would + // silently push at a port nobody promised to listen on. + let err = peer_with(Some("10.100.0.2/32"), None) + .snapshot_store_addr() + .expect_err("no port must fail"); + let msg = format!("{err:#}"); + assert!(msg.contains("snapshotStorePort"), "{msg}"); + assert!( + !msg.contains("51821"), + "must not suggest a guessed port: {msg}" + ); + } +} diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index b702f10c..f74bee85 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -375,6 +375,23 @@ pub enum HostRequest { parent: Option, dest: String, }, + /// Push a snapshot to a peer hive's snapshot store over the + /// WireGuard mesh (`hivectl agent subvol snapshot push`). + /// The network sibling of [`HostRequest::SendSnapshot`]: same + /// snapshot and optional incremental `parent`, but the stream goes + /// to `peer`'s receiver instead of a local file. + /// + /// `peer` is a domain from `services.hyperhive.swarm.peers`; the + /// daemon resolves its mesh address and store port from there and + /// fails if the peer declares neither. Bare success — nothing is + /// written on this host to report a path for. + PushSnapshot { + name: Ident, + label: String, + #[serde(default)] + parent: Option, + peer: String, + }, } /// One agent's btrfs qgroup usage row — the [`HostRequest::QuotaShow`] diff --git a/hivectl/src/cli.rs b/hivectl/src/cli.rs index d3d5f8b1..34f45582 100644 --- a/hivectl/src/cli.rs +++ b/hivectl/src/cli.rs @@ -620,10 +620,10 @@ pub enum SnapshotCmd { /// Snapshot label passed to `subvol snapshot create --label`. label: String, }, - /// Export a snapshot to a local file via `btrfs send` (the local-file - /// half of inter-hive migration transport; the cross-hive `ssh ... - /// btrfs receive` leg isn't wired up yet). Also useful standalone as a - /// point-in-time backup: a full send with no `--parent` produces a + /// Export a snapshot to a local file via `btrfs send` — the + /// local-file half of the inter-hive migration transport (`push` + /// is the network half). Also useful standalone as a point-in-time + /// backup: a full send with no `--parent` produces a /// self-contained archive of the snapshot. Send { /// Snapshot label passed to `subvol snapshot create --label`. @@ -638,4 +638,27 @@ pub enum SnapshotCmd { #[arg(long)] dest: String, }, + /// Stream a snapshot to a peer hive's snapshot store over the + /// WireGuard mesh — the network half of the migration transport. + /// + /// Nothing is staged locally: `btrfs send` writes straight into the + /// connection, so a multi-gigabyte agent needs no scratch space on + /// this host. The mesh is the authentication (cryptokey routing + /// binds the peer's address to its key), so there is no credential + /// to pass here. + Push { + /// Snapshot label passed to `subvol snapshot create --label`. + label: String, + /// Optional parent snapshot label for an incremental send + /// (`btrfs send -p`) — must be an existing, older snapshot of the + /// same agent, and must already be present on the receiver. + /// Omit for a full send. + #[arg(long)] + parent: Option, + /// Peer hive domain, as declared in + /// `services.hyperhive.swarm.peers`. Its mesh address and + /// snapshot-store port are read from there. + #[arg(long)] + peer: String, + }, } diff --git a/hivectl/src/subvol.rs b/hivectl/src/subvol.rs index 811fe800..2f680176 100644 --- a/hivectl/src/subvol.rs +++ b/hivectl/src/subvol.rs @@ -44,6 +44,11 @@ pub(crate) async fn dispatch_subvol(socket: &Path, name: &str, cmd: SubvolCmd) - parent, dest, } => subvol_snapshot_send(socket, name, &label, parent.as_deref(), &dest).await, + SnapshotCmd::Push { + label, + parent, + peer, + } => subvol_snapshot_push(socket, name, &label, parent.as_deref(), &peer).await, }, } } @@ -217,3 +222,34 @@ async fn subvol_snapshot_send( ) .await } + +/// `subvol snapshot push