nuclei/net/
unix.rs

1//! Unix domain sockets: listeners, streams and datagrams, and [`UCred`], the
2//! credentials of a stream's peer; and Unix pipes ([`pipe`]).
3
4pub mod pipe;
5use core::future::poll_fn;
6use core::pin::Pin;
7use core::task::{Context, Poll};
8use std::io;
9use std::os::fd::OwnedFd;
10use std::os::unix::net::{SocketAddr as UnixSocketAddr, UnixDatagram, UnixListener, UnixStream};
11use std::path::Path;
12use std::time::Duration;
13
14use futures_core::Stream;
15
16use crate::buf::owned::{IoBuf, IoBufMut};
17use crate::io::source::write::{Dest, Peer};
18use crate::io::source::{Addr, Kind};
19use crate::io::sys::UnixAddr;
20use crate::io::{Handle, Interest, IoObject, Ready, sys};
21
22/// The credentials of a Unix stream's peer: its user and group, and its
23/// process where the system tells it (Linux, Android, macOS, iOS).
24#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
25pub struct UCred {
26    uid: u32,
27    gid: u32,
28    pid: Option<i32>,
29}
30
31impl UCred {
32    /// The peer's user id.
33    pub fn uid(&self) -> u32 {
34        self.uid
35    }
36
37    /// The peer's group id.
38    pub fn gid(&self) -> u32 {
39        self.gid
40    }
41
42    /// The peer's process id, where the system tells it.
43    pub fn pid(&self) -> Option<i32> {
44        self.pid
45    }
46}
47
48#[cfg(any(target_os = "linux", target_os = "android"))]
49fn peer_cred(fd: std::os::fd::RawFd) -> io::Result<UCred> {
50    let mut cred = libc::ucred {
51        pid: 0,
52        uid: 0,
53        gid: 0,
54    };
55    let mut len = size_of::<libc::ucred>() as libc::socklen_t;
56    // SAFETY: SO_PEERCRED answers a `ucred`, written into our own.
57    if unsafe {
58        libc::getsockopt(
59            fd,
60            libc::SOL_SOCKET,
61            libc::SO_PEERCRED,
62            core::ptr::from_mut(&mut cred).cast(),
63            &mut len,
64        )
65    } != 0
66    {
67        return Err(io::Error::last_os_error());
68    }
69    Ok(UCred {
70        uid: cred.uid,
71        gid: cred.gid,
72        pid: Some(cred.pid),
73    })
74}
75
76#[cfg(not(any(target_os = "linux", target_os = "android")))]
77fn peer_cred(fd: std::os::fd::RawFd) -> io::Result<UCred> {
78    let (mut uid, mut gid) = (0, 0);
79    // SAFETY: the call writes the two ids into our own.
80    if unsafe { libc::getpeereid(fd, &mut uid, &mut gid) } != 0 {
81        return Err(io::Error::last_os_error());
82    }
83    #[cfg(any(target_os = "macos", target_os = "ios"))]
84    let pid = {
85        let mut pid: libc::pid_t = 0;
86        let mut len = size_of::<libc::pid_t>() as libc::socklen_t;
87        // SAFETY: LOCAL_PEERPID answers a `pid_t`, written into our own.
88        let r = unsafe {
89            libc::getsockopt(
90                fd,
91                libc::SOL_LOCAL,
92                libc::LOCAL_PEERPID,
93                core::ptr::from_mut(&mut pid).cast(),
94                &mut len,
95            )
96        };
97        (r == 0).then_some(pid)
98    };
99    #[cfg(not(any(target_os = "macos", target_os = "ios")))]
100    let pid = None;
101    Ok(UCred { uid, gid, pid })
102}
103
104fn unix(addr: Addr) -> io::Result<UnixSocketAddr> {
105    match addr {
106        Addr::Unix(addr) => Ok(addr.into_inner()),
107        Addr::Inet(_) => Err(io::Error::new(
108            io::ErrorKind::InvalidData,
109            "a Unix socket's peer has an internet address",
110        )),
111    }
112}
113
114impl Handle<UnixListener> {
115    /// A listener bound to `path`.
116    ///
117    /// # Errors
118    ///
119    /// As [`UnixListener::bind`].
120    pub fn bind<P: AsRef<Path>>(path: P) -> io::Result<Handle<UnixListener>> {
121        Handle::adopt(UnixListener::bind(path)?, Kind::UnixListener)
122    }
123
124    /// A listener bound to `addr`: a path, or on Linux a name in the
125    /// abstract namespace.
126    ///
127    /// # Errors
128    ///
129    /// As [`UnixListener::bind_addr`].
130    pub fn bind_addr(addr: &UnixSocketAddr) -> io::Result<Handle<UnixListener>> {
131        Handle::adopt(UnixListener::bind_addr(addr)?, Kind::UnixListener)
132    }
133
134    /// The next connection, with its peer's address. On macOS an accept
135    /// that finds the process out of descriptors closes the connection it
136    /// took, and the next ones take none until a descriptor is free, as a
137    /// TCP listener's do.
138    ///
139    /// # Errors
140    ///
141    /// As `accept(2)`.
142    pub async fn accept(&self) -> io::Result<(Handle<UnixStream>, UnixSocketAddr)> {
143        poll_fn(|cx| self.poll_accept(cx)).await
144    }
145
146    fn poll_accept(
147        &self,
148        cx: &mut Context<'_>,
149    ) -> Poll<io::Result<(Handle<UnixStream>, UnixSocketAddr)>> {
150        self.source().poll_accept(cx).map(|accepted| {
151            let (fd, addr) = accepted?;
152            Ok((stream(fd), unix(addr)?))
153        })
154    }
155
156    /// The connections as they come, without their addresses; the stream
157    /// never ends, and an error is one of its items.
158    pub fn incoming(
159        &self,
160    ) -> impl Stream<Item = io::Result<Handle<UnixStream>>> + Send + Unpin + '_ {
161        Incoming { listener: self }
162    }
163}
164
165fn stream(fd: OwnedFd) -> Handle<UnixStream> {
166    Handle::from_nonblocking(UnixStream::from(fd), Kind::Stream)
167}
168
169struct Incoming<'a> {
170    listener: &'a Handle<UnixListener>,
171}
172
173impl Stream for Incoming<'_> {
174    type Item = io::Result<Handle<UnixStream>>;
175
176    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
177        self.listener
178            .poll_accept(cx)
179            .map(|accepted| Some(accepted.map(|(stream, _)| stream)))
180    }
181}
182
183/// The first pause of a connect that found the listener's backlog full,
184/// doubled up to [`FULL_BACKLOG_PAUSE_MAX`] while it stays full.
185const FULL_BACKLOG_PAUSE: Duration = Duration::from_millis(1);
186const FULL_BACKLOG_PAUSE_MAX: Duration = Duration::from_millis(64);
187
188/// The connect of the non-blocking Unix stream `socket` to `addr`.
189///
190/// Linux answers a connect to a listener whose backlog is full `EAGAIN`, and
191/// nothing tells a socket when a place frees: the connect is made again on
192/// the same socket (so its options stay) after a pause on a timer, doubled
193/// up to [`FULL_BACKLOG_PAUSE_MAX`], until the listener takes it or answers
194/// otherwise, as std's blocking connect waits. The future holds nothing but
195/// its socket and its timer, so a drop of it frees everything
196/// (`proofs/tla/connect`).
197pub(super) async fn connect(socket: OwnedFd, addr: &UnixAddr) -> io::Result<Handle<UnixStream>> {
198    let stream = Handle::from_nonblocking(UnixStream::from(socket), Kind::Stream);
199    let mut pause = FULL_BACKLOG_PAUSE;
200    loop {
201        match poll_fn(|cx| stream.source().poll_connect(cx, Peer::Unix(addr))).await {
202            Ok(()) => return Ok(stream),
203            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
204                crate::time::sleep(pause).await;
205                pause = (pause * 2).min(FULL_BACKLOG_PAUSE_MAX);
206            }
207            Err(err) => return Err(err),
208        }
209    }
210}
211
212impl Handle<UnixStream> {
213    /// A stream connected to the listener at `path`.
214    ///
215    /// # Errors
216    ///
217    /// As `connect(2)`.
218    pub async fn connect<P: AsRef<Path>>(path: P) -> io::Result<Handle<UnixStream>> {
219        connect(
220            sys::socket(libc::AF_UNIX, libc::SOCK_STREAM)?,
221            &UnixAddr::path(path.as_ref())?,
222        )
223        .await
224    }
225
226    /// A stream connected to the listener at `addr`: a path, or on Linux a
227    /// name in the abstract namespace.
228    ///
229    /// # Errors
230    ///
231    /// As `connect(2)`; an unnamed address names no listener.
232    pub async fn connect_addr(addr: &UnixSocketAddr) -> io::Result<Handle<UnixStream>> {
233        connect(
234            sys::socket(libc::AF_UNIX, libc::SOCK_STREAM)?,
235            &UnixAddr::of(addr)?,
236        )
237        .await
238    }
239
240    /// The credentials of the process at the other end, as the connection
241    /// was made.
242    ///
243    /// # Errors
244    ///
245    /// When the system cannot tell them.
246    pub fn peer_cred(&self) -> io::Result<UCred> {
247        peer_cred(self.raw_io())
248    }
249
250    /// Two streams connected to each other.
251    ///
252    /// # Errors
253    ///
254    /// As `socketpair(2)`.
255    pub fn pair() -> io::Result<(Handle<UnixStream>, Handle<UnixStream>)> {
256        let (a, b) = UnixStream::pair()?;
257        Ok((
258            Handle::adopt(a, Kind::Stream)?,
259            Handle::adopt(b, Kind::Stream)?,
260        ))
261    }
262
263    /// Sends bytes; returns how many were taken.
264    ///
265    /// # Errors
266    ///
267    /// As `send(2)`, or the error that stopped earlier bytes.
268    pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
269        poll_fn(|cx| self.source().poll_write(cx, buf)).await
270    }
271
272    /// Receives bytes; zero at the end of the stream.
273    ///
274    /// # Errors
275    ///
276    /// As `recv(2)`.
277    pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
278        poll_fn(|cx| self.source().poll_read(cx, buf, false)).await
279    }
280
281    /// Looks at the bytes the next receive returns, without taking them.
282    ///
283    /// # Errors
284    ///
285    /// As `recv(2)` with `MSG_PEEK`.
286    pub async fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
287        poll_fn(|cx| self.source().poll_read(cx, buf, true)).await
288    }
289}
290
291impl Handle<UnixDatagram> {
292    /// A socket bound to `path`.
293    ///
294    /// # Errors
295    ///
296    /// As [`UnixDatagram::bind`].
297    pub fn bind<P: AsRef<Path>>(path: P) -> io::Result<Handle<UnixDatagram>> {
298        Handle::adopt(UnixDatagram::bind(path)?, Kind::UnixDatagram)
299    }
300
301    /// A socket bound to `addr`: a path, or on Linux a name in the abstract
302    /// namespace.
303    ///
304    /// # Errors
305    ///
306    /// As [`UnixDatagram::bind_addr`].
307    pub fn bind_addr(addr: &UnixSocketAddr) -> io::Result<Handle<UnixDatagram>> {
308        Handle::adopt(UnixDatagram::bind_addr(addr)?, Kind::UnixDatagram)
309    }
310
311    /// A socket bound to no address: it sends with `send_to`, or to the peer
312    /// its `connect` names.
313    ///
314    /// # Errors
315    ///
316    /// When the system has no socket to give.
317    pub fn unbound() -> io::Result<Handle<UnixDatagram>> {
318        Ok(Handle::from_nonblocking(
319            UnixDatagram::from(sys::socket(libc::AF_UNIX, libc::SOCK_DGRAM)?),
320            Kind::UnixDatagram,
321        ))
322    }
323
324    /// Two sockets connected to each other.
325    ///
326    /// # Errors
327    ///
328    /// As `socketpair(2)`.
329    pub fn pair() -> io::Result<(Handle<UnixDatagram>, Handle<UnixDatagram>)> {
330        let (a, b) = UnixDatagram::pair()?;
331        Ok((
332            Handle::adopt(a, Kind::UnixDatagram)?,
333            Handle::adopt(b, Kind::UnixDatagram)?,
334        ))
335    }
336
337    /// Sends a datagram to the connected peer.
338    ///
339    /// # Errors
340    ///
341    /// As `send(2)` of this datagram.
342    pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
343        poll_fn(|cx| self.source().poll_send(cx, buf, Dest::Connected)).await
344    }
345
346    /// Receives a datagram from the connected peer; what does not fit `buf`
347    /// is cut.
348    ///
349    /// # Errors
350    ///
351    /// As `recv(2)`.
352    pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
353        self.receive(buf, false).await.map(|(n, _)| n)
354    }
355
356    /// Sends all of `buf` as one datagram to the connected peer and gives
357    /// it back with its length. The driver sends from `buf` itself, so the
358    /// bytes are never copied.
359    ///
360    /// If the future is dropped before it completes, the datagram is sent
361    /// whole or not at all: one the driver has not handed to the kernel is
362    /// withdrawn, and one already handed over goes out. `buf` is any
363    /// [`IoBuf`](crate::io::IoBuf).
364    pub async fn send_owned<B>(&self, buf: B) -> (io::Result<usize>, B)
365    where
366        B: IoBuf,
367    {
368        self.source().send_owned(buf, Dest::Connected).await
369    }
370
371    /// Sends all of `buf` as one datagram to the socket at `path` and gives
372    /// it back with its length, as [`send_owned`](Self::send_owned) sends to
373    /// the connected peer.
374    pub async fn send_to_owned<P, B>(&self, buf: B, path: P) -> (io::Result<usize>, B)
375    where
376        P: AsRef<Path>,
377        B: IoBuf,
378    {
379        match UnixAddr::path(path.as_ref()) {
380            Ok(addr) => self.source().send_owned(buf, Dest::Unix(&addr)).await,
381            Err(err) => (Err(err), buf),
382        }
383    }
384
385    /// Receives a datagram from the connected peer into the spare capacity
386    /// of `buf`, after the bytes it holds, and gives `buf` back with the
387    /// datagram's length. The driver receives into `buf` itself, so the bytes
388    /// are never copied; what does not fit is cut.
389    ///
390    /// If the future is dropped before it completes, a datagram the driver
391    /// received into `buf` stays with the socket for its next receive, so
392    /// none is lost. `buf` is any [`IoBufMut`](crate::io::IoBufMut).
393    pub async fn recv_owned<B>(&self, buf: B) -> (io::Result<usize>, B)
394    where
395        B: IoBufMut,
396    {
397        let (result, buf) = self.recv_from_owned(buf).await;
398        (result.map(|(n, _)| n), buf)
399    }
400
401    /// Receives a datagram with its sender's address into the spare capacity
402    /// of `buf`, as [`recv_owned`](Self::recv_owned).
403    pub async fn recv_from_owned<B>(&self, buf: B) -> (io::Result<(usize, UnixSocketAddr)>, B)
404    where
405        B: IoBufMut,
406    {
407        let (result, buf) = self.source().recv_from_owned(buf).await;
408        (
409            result.and_then(|(n, addr)| unix(addr).map(|addr| (n, addr))),
410            buf,
411        )
412    }
413
414    /// Looks at the next datagram without taking it.
415    ///
416    /// # Errors
417    ///
418    /// As `recv(2)` with `MSG_PEEK`.
419    pub async fn peek(&self, buf: &mut [u8]) -> io::Result<usize> {
420        self.receive(buf, true).await.map(|(n, _)| n)
421    }
422
423    /// Sends a datagram to the socket at `path`.
424    ///
425    /// # Errors
426    ///
427    /// As `sendto(2)` of this datagram.
428    pub async fn send_to<P: AsRef<Path>>(&self, buf: &[u8], path: P) -> io::Result<usize> {
429        let addr = UnixAddr::path(path.as_ref())?;
430        poll_fn(|cx| self.source().poll_send(cx, buf, Dest::Unix(&addr))).await
431    }
432
433    /// Receives a datagram with its sender's address.
434    ///
435    /// # Errors
436    ///
437    /// As `recvfrom(2)`.
438    pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, UnixSocketAddr)> {
439        self.receive(buf, false).await
440    }
441
442    /// Looks at the next datagram and its sender without taking it.
443    ///
444    /// # Errors
445    ///
446    /// As `recvfrom(2)` with `MSG_PEEK`.
447    pub async fn peek_from(&self, buf: &mut [u8]) -> io::Result<(usize, UnixSocketAddr)> {
448        self.receive(buf, true).await
449    }
450
451    async fn receive(&self, buf: &mut [u8], peek: bool) -> io::Result<(usize, UnixSocketAddr)> {
452        poll_fn(|cx| self.poll_receive(cx, buf, peek)).await
453    }
454
455    #[inline(always)]
456    fn poll_receive(
457        &self,
458        cx: &mut Context<'_>,
459        buf: &mut [u8],
460        peek: bool,
461    ) -> Poll<io::Result<(usize, UnixSocketAddr)>> {
462        self.source().poll_receive_as(cx, buf, peek)
463    }
464
465    /// Waits until a receive would end at once: a datagram came, or an
466    /// error (which the receive reports).
467    ///
468    /// # Errors
469    ///
470    /// When the socket cannot be watched.
471    pub async fn readable(&self) -> io::Result<()> {
472        self.ready(Interest::READABLE).await.map(drop)
473    }
474
475    /// Waits until a send would be taken at once.
476    ///
477    /// # Errors
478    ///
479    /// When the socket cannot be watched.
480    pub async fn writable(&self) -> io::Result<()> {
481        self.ready(Interest::WRITABLE).await.map(drop)
482    }
483
484    /// Waits until some of `interest` is ready, and says what is.
485    ///
486    /// # Errors
487    ///
488    /// When the socket cannot be watched.
489    pub async fn ready(&self, interest: Interest) -> io::Result<Ready> {
490        poll_fn(|cx| super::poll_ready(self.source(), cx, interest)).await
491    }
492
493    /// Polls until a receive would end at once.
494    ///
495    /// # Errors
496    ///
497    /// When the socket cannot be watched.
498    pub fn poll_recv_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
499        self.source().poll_read_ready(cx).map(|r| r.map(drop))
500    }
501
502    /// Polls until a send would be taken at once.
503    ///
504    /// # Errors
505    ///
506    /// When the socket cannot be watched.
507    pub fn poll_send_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
508        self.source().poll_write_ready(cx).map(|r| r.map(drop))
509    }
510
511    /// Sends a datagram to the connected peer without waiting.
512    ///
513    /// # Errors
514    ///
515    /// `WouldBlock` when a datagram is still queued, and as `send(2)`.
516    pub fn try_send(&self, buf: &[u8]) -> io::Result<usize> {
517        super::at_once(|cx| self.source().poll_send(cx, buf, Dest::Connected))
518    }
519
520    /// Sends a datagram to the socket at `path` without waiting.
521    ///
522    /// # Errors
523    ///
524    /// `WouldBlock` when a datagram is still queued, and as `sendto(2)`.
525    pub fn try_send_to<P: AsRef<Path>>(&self, buf: &[u8], path: P) -> io::Result<usize> {
526        let addr = UnixAddr::path(path.as_ref())?;
527        super::at_once(|cx| self.source().poll_send(cx, buf, Dest::Unix(&addr)))
528    }
529
530    /// Receives a datagram from the connected peer without waiting; what
531    /// does not fit `buf` is cut.
532    ///
533    /// # Errors
534    ///
535    /// `WouldBlock` when none came yet, and as `recv(2)`.
536    pub fn try_recv(&self, buf: &mut [u8]) -> io::Result<usize> {
537        super::at_once(|cx| self.poll_receive(cx, buf, false)).map(|(n, _)| n)
538    }
539
540    /// Receives a datagram and its sender's address without waiting.
541    ///
542    /// # Errors
543    ///
544    /// `WouldBlock` when none came yet, and as `recvfrom(2)`.
545    pub fn try_recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, UnixSocketAddr)> {
546        super::at_once(|cx| self.poll_receive(cx, buf, false))
547    }
548}
nuclei/net/unix.rs

nuclei/net/
unix.rs
