vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#2862): drop the fd framing module, the hazard is unreachable

Browse source 
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.
This commit is contained in:
atlas 2026-07-31 21:01:17 +02:00 • committed by mara
commit 364bc290df
2 changed files with 0 additions and 285 deletions
Download patch file Download diff file Expand all files Collapse all files

279
hive-priv-sock/src/framing.rs
View file

@ -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<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");
}
}

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

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