feat: adopt a host-created listening socket on every Unix backend

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<dyn Write>) {
+    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);
+}

crates/turnloop/src/backend/ipc.rs

@@ -101,6 +101,28 @@ pub(super) fn stdio(raw: RawFd) -> Result<Detached> {
 pub(super) fn classify(fd: OwnedFd) -> Result<Detached> {
     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<Detached> {
+    #[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<Detached> {
     // SAFETY: stat is plain output storage and all-zero is a valid initial value.
     let mut stat: libc::stat = unsafe { zeroed() };

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<Self> {
         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<Self> {
+        super::ipc::classify_listener(fd)
+    }
 }
 impl Drop for Detached {
     fn drop(&mut self) {