From b1fd8de2d30dcd92fe07314981c3595292657dd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Wed, 23 Sep 2026 05:11:19 +0200 Subject: [PATCH] feat: adopt a host-created listening socket on every Unix backend Detached::from_fd recognises a listener only where the kernel answers SO_ACCEPTCONN. Linux does; macOS and the BSDs do not (macOS returns ENOPROTOOPT), so on kqueue an adopted listener was classified as a TCP stream and could never accept. A host that binds its own listener, or receives one from another process (a cluster primary handing over its listening socket), had no way to use it on a Mac. Detached::from_listener_fd takes the caller's word that the descriptor is a listener, through the existing classify_hint(fd, true) path that turnloop already uses for listeners it receives over IPC. On epoll the kernel still decides. A descriptor that is not a stream socket, or that has a peer (a listener never does), is refused with InvalidInput. from_fd's docs now point kqueue callers at the new constructor. The contract test binds TCP and Unix-domain listeners outside turnloop, adopts them, and asserts the loop accepts a connection and reads its bytes; a third case checks a connected socket and a file are refused. With from_listener_fd routed back to from_fd, all three fail on macOS. --- .../turnloop-contract/tests/adopt_listener.rs | 107 ++++++++++++++++++ crates/turnloop/src/backend/ipc.rs | 22 ++++ crates/turnloop/src/backend/unix.rs | 17 +++ 3 files changed, 146 insertions(+) create mode 100644 crates/turnloop-contract/tests/adopt_listener.rs diff --git a/crates/turnloop-contract/tests/adopt_listener.rs b/crates/turnloop-contract/tests/adopt_listener.rs new file mode 100644 index 0000000..8ae055e --- /dev/null +++ b/crates/turnloop-contract/tests/adopt_listener.rs @@ -0,0 +1,107 @@ +//! Adopting a listening socket the host created, on every Unix backend. +//! +//! `Detached::from_fd` recognises a listener only where the kernel answers +//! `SO_ACCEPTCONN`; macOS and the BSDs do not, so there an adopted listener used +//! to become a stream that could never accept. `Detached::from_listener_fd` says +//! what the descriptor is. These tests bind outside turnloop, adopt, and prove the +//! loop accepts a real connection and reads its bytes. +#![deny(unsafe_op_in_unsafe_fn)] +#![cfg(all( + not(loom), + any( + target_vendor = "apple", + target_os = "linux", + target_os = "android", + target_os = "freebsd" + ) +))] +use std::{ + io::Write, + net::{Ipv4Addr, TcpListener, TcpStream}, + os::{ + fd::OwnedFd, + unix::net::{UnixListener, UnixStream}, + }, + time::Duration, +}; +use turnloop::*; + +/// Accept one connection on an adopted listener, then read the byte the peer +/// wrote before the accept was even submitted. +fn accept_and_read(listener: OwnedFd, connect: impl FnOnce() -> Box) { + let mut l = Loop::new(Config::default()).expect("loop"); + let server = l + .attach( + Detached::from_listener_fd(listener).expect("adopt listener"), + Token(1), + ) + .expect("attach"); + let mut client = connect(); + client.write_all(b"x").expect("client write"); + l.accept(server, Token(2)).expect("accept"); + let mut out = Completions::default(); + let until = l.now() + Duration::from_secs(5); + let mut conn = None; + let mut read = false; + while !read { + assert!(l.now() < until, "adopted listener never accepted and read"); + l.turn(Timeout::Until(until), &mut out).expect("turn"); + for c in out.drain() { + match c.result { + OpResult::Accepted { conn: h, .. } | OpResult::PipeAccepted { conn: h } => { + conn = Some(h); + l.read(h, ReadBuf::Pooled, Token(3)).expect("read"); + } + OpResult::Read { n, lease: Some(b) } => { + assert_eq!((n, b.as_slice()), (1, &b"x"[..])); + read = true; + } + other => panic!("unexpected {other:?}"), + } + } + } + let conn = conn.expect("accepted"); + for (i, h) in [conn, server].into_iter().enumerate() { + l.close(h, Token(10 + i as u64)).expect("close"); + } + while l.alive() { + assert!(l.now() < until, "close never completed"); + l.turn(Timeout::Until(until), &mut out).expect("turn"); + out.drain(); + } +} + +#[test] +fn adopted_tcp_listener_accepts() { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).expect("bind"); + let address = listener.local_addr().expect("address"); + accept_and_read(listener.into(), move || { + Box::new(TcpStream::connect(address).expect("connect")) + }); +} + +#[test] +fn adopted_unix_listener_accepts() { + let path = std::env::temp_dir().join(format!("turnloop-adopt-{}.sock", std::process::id())); + let _ = std::fs::remove_file(&path); + let listener = UnixListener::bind(&path).expect("bind"); + let peer = path.clone(); + accept_and_read(listener.into(), move || { + Box::new(UnixStream::connect(&peer).expect("connect")) + }); + let _ = std::fs::remove_file(&path); +} + +/// What is provably not a listener is refused rather than adopted as one: a +/// connected socket (it has a peer, which a listener never has) and a file. +#[test] +fn from_listener_fd_refuses_what_is_not_a_listener() { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).expect("bind"); + let connected = TcpStream::connect(listener.local_addr().expect("address")).expect("connect"); + let error = Detached::from_listener_fd(connected.into()).expect_err("connected socket"); + assert_eq!(error.kind, ErrorKind::InvalidInput); + + let file = std::fs::File::open(std::env::current_exe().expect("exe")).expect("open"); + let error = Detached::from_listener_fd(file.into()).expect_err("file"); + assert_eq!(error.kind, ErrorKind::InvalidInput); +} diff --git a/crates/turnloop/src/backend/ipc.rs b/crates/turnloop/src/backend/ipc.rs index 1580dff..a9a6b2b 100644 --- a/crates/turnloop/src/backend/ipc.rs +++ b/crates/turnloop/src/backend/ipc.rs @@ -101,6 +101,28 @@ pub(super) fn stdio(raw: RawFd) -> Result { pub(super) fn classify(fd: OwnedFd) -> Result { classify_hint(fd, false) } +/// Adopt a descriptor the caller says is a listening stream socket, and refuse +/// anything that is provably not one. +/// +/// epoll asks the kernel with `SO_ACCEPTCONN`. kqueue platforms do not support +/// that option (macOS answers `ENOPROTOOPT`), so there the caller's word is +/// taken - except that a socket with a peer is connected, and a listener never +/// has one, so that case is still refused. +pub(super) fn classify_listener(fd: OwnedFd) -> Result { + #[cfg(turnloop_backend = "kqueue")] + { + let mut peer = Addr::empty(); + // SAFETY: initialized sockaddr output storage and length on a live fd. + if unsafe { libc::getpeername(fd.as_raw_fd(), peer.mut_ptr(), &mut peer.len) } == 0 { + return Err(Error::new(ErrorKind::InvalidInput)); + } + } + let transport = classify_hint(fd, true)?; + if !matches!(transport.kind, Kind::Listener | Kind::PipeListener) { + return Err(Error::new(ErrorKind::InvalidInput)); + } + Ok(transport) +} fn classify_hint(fd: OwnedFd, listener: bool) -> Result { // SAFETY: stat is plain output storage and all-zero is a valid initial value. let mut stat: libc::stat = unsafe { zeroed() }; diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs index be0c62a..fe78884 100644 --- a/crates/turnloop/src/backend/unix.rs +++ b/crates/turnloop/src/backend/unix.rs @@ -108,9 +108,26 @@ impl Detached { /// Adopt an owned Unix descriptor, classifying stream/file/TTY or socket. /// The descriptor must have no concurrent I/O users. Status flags and terminal /// settings are restored when this transport is closed or dropped. + /// + /// On kqueue platforms (macOS, the BSDs) a *listening* socket cannot be told + /// apart from a connected one here, and is adopted as a stream; adopt a + /// listener with [`from_listener_fd`](Self::from_listener_fd) instead. pub fn from_fd(fd: OwnedFd) -> Result { super::ipc::classify(fd) } + /// Adopt an owned, already listening TCP or Unix stream socket, so the loop + /// can `accept` on it - a listener a host bound itself or received from + /// another process. + /// + /// This works on every Unix backend. [`from_fd`](Self::from_fd) recognises a + /// listener only where the kernel reports `SO_ACCEPTCONN`, which macOS and + /// the BSDs do not. A descriptor that is not a stream socket, or that has a + /// peer, is refused with `InvalidInput` and closed. On kqueue platforms a + /// socket that was bound but never `listen`ed cannot be detected; its first + /// `accept` fails instead. + pub fn from_listener_fd(fd: OwnedFd) -> Result { + super::ipc::classify_listener(fd) + } } impl Drop for Detached { fn drop(&mut self) {