feat(#2862): fd-carrying line framing for the priv socket

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

@@ -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<OwnedFd>,
+}
+
+/// 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<u8>,
+    pending_fds: VecDeque<OwnedFd>,
+}
+
+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<OwnedFd>) -> Vec<Result<Framed, FramingError>> {
+        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<u8> = 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<OwnedFd> {
+        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, FramingError>) -> 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");
+    }
+}

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";