conmonrs/attach/
socket.rs

1//! The attach sockets themselves: accepting the clients, reading their
2//! stdin and writing the output to them.
3
4use super::{AttachListener, ListenerCleanup, SharedContainerAttach};
5use crate::{
6    async_fd,
7    container_io::{DRAIN_MAX_WAIT, Drain, Lag, Message, READ_BUFFER_SIZE},
8    fs_util::FileId,
9    listener,
10    mutex::lock,
11    util::with_scratch,
12};
13use anyhow::{Context, Result, bail};
14use bytes::Bytes;
15use nix::{
16    errno::Errno,
17    poll::{PollFd, PollFlags, PollTimeout, poll},
18    sys::epoll::{Epoll, EpollCreateFlags, EpollEvent, EpollFlags},
19    sys::socket::{Backlog, SockType, UnixAddr, bind, listen},
20};
21use std::{
22    fs, future,
23    io::{self, IoSlice},
24    os::{
25        fd::{AsFd, AsRawFd},
26        unix::fs::FileTypeExt,
27    },
28    path::Path,
29    pin::Pin,
30    sync::{Arc, atomic::Ordering},
31    task::{Poll, ready},
32};
33use tokio::{
34    io::{ErrorKind, Interest},
35    select,
36    sync::broadcast::{Receiver, error::RecvError},
37    task, time,
38};
39use tokio_seqpacket::{UnixSeqpacket, UnixSeqpacketListener};
40use tokio_util::sync::CancellationToken;
41use tracing::{Instrument, debug, debug_span, error, trace, warn};
42
43/// Check if no more input will arrive after a zero byte read. That is the
44/// case once the peer closed its write side, which a SOCK_SEQPACKET socket
45/// reports as POLLRDHUP, and no data is queued anymore. Checking the flag
46/// first makes sure that no data arrives in between.
47#[expect(unsafe_code, reason = "the FIONREAD ioctl")]
48fn input_ended(socket: &UnixSeqpacket) -> io::Result<bool> {
49    // Besides POLLRDHUP, poll can only report a hangup or an error, which
50    // end the reading as well. The number of ready fds tells that, nix
51    // does not decode the POLLRDHUP bit of the returned events.
52    let events = PollFlags::from_bits_retain(libc::POLLRDHUP);
53    let read_closed = loop {
54        match poll(
55            &mut [PollFd::new(socket.as_fd(), events)],
56            PollTimeout::ZERO,
57        ) {
58            Err(Errno::EINTR) => {}
59            res => break res? > 0,
60        }
61    };
62    if !read_closed {
63        return Ok(false);
64    }
65
66    // For a SOCK_SEQPACKET socket, FIONREAD sums up all queued packets.
67    let mut queued: libc::c_int = 0;
68    // SAFETY: FIONREAD writes a single c_int.
69    if unsafe { libc::ioctl(socket.as_raw_fd(), libc::FIONREAD, &mut queued) } < 0 {
70        return Err(io::Error::last_os_error());
71    }
72    Ok(queued == 0)
73}
74
75/// The most data of an output packet, after its pipe byte. Like for conmon,
76/// CRI-O and the Go client read one byte more than this, so that a full read
77/// of a pipe goes out as one packet.
78pub(super) const PACKET_BUF_SIZE: usize = 8192;
79
80// The write loop sends each read of a pipe as one packet.
81const _: () = assert!(READ_BUFFER_SIZE <= PACKET_BUF_SIZE);
82
83/// The maximum size of a stdin packet from an attach client, a larger one
84/// gets truncated. The Go client writes at most 32 KiB at once, like CRI-O
85/// does for conmon, plus the start of a detach key sequence which it held
86/// back, a few bytes which the extra KiB leaves room for. The scratch buffer
87/// of the thread keeps this size after the first stdin, see [`with_scratch`].
88pub(super) const STDIN_PACKET_MAX_SIZE: usize = 33 * 1024;
89
90/// The packet indicating that we're done writing.
91pub(super) const DONE_PACKET: &[u8; 1] = &[0];
92
93/// Listen on the attach socket at `path`, unless a listener of this server
94/// does already.
95pub(super) fn create_listener(
96    path: &Path,
97    attach: SharedContainerAttach,
98    token: CancellationToken,
99) -> Result<()> {
100    // Held until the new listener got registered, so that a listener which
101    // stops meanwhile does not remove its socket.
102    let mut listeners = lock(&attach.listeners);
103    if let Some(listener) = listeners.remove(path) {
104        if !listener.stop.is_cancelled() && listener.is_bound_at(path) {
105            debug!("Already listening on attach socket {}", path.display());
106            listeners.insert(path.to_path_buf(), listener);
107            return Ok(());
108        }
109        // Nobody can connect to a socket which got removed, for example by
110        // an older client, or replaced. Stop accepting on it, otherwise
111        // the listener would wait until the container exits.
112        debug!(
113            "Attach socket {} got removed or replaced, listening again",
114            path.display()
115        );
116        listener.stop.cancel();
117    }
118
119    // A socket of this path which no listener of this server accepts on
120    // is left over, for example by a previous server.
121    match fs::symlink_metadata(path) {
122        Ok(m) if m.file_type().is_socket() => {
123            debug!("Removing stale attach socket {}", path.display());
124            fs::remove_file(path).context("remove stale attach socket")?;
125        }
126        Ok(_) => bail!(
127            "attach socket path {} exists and is no socket",
128            path.display()
129        ),
130        Err(e) if e.kind() == ErrorKind::NotFound => {}
131        Err(e) => return Err(e).context("check attach socket path"),
132    }
133
134    debug!("Creating attach socket: {}", path.display());
135    let fd = listener::private_socket(SockType::SeqPacket).context("create attach socket")?;
136
137    // keep parent_fd in scope until the bind, or else the socket will not work
138    let (shortened_path, _parent_dir) = listener::shorten_socket_path(path)?;
139    let addr = UnixAddr::new(&shortened_path).context("create socket addr")?;
140    bind(fd.as_raw_fd(), &addr).context("bind socket fd")?;
141
142    // Nothing listens on the socket if the setup fails, remove it. The
143    // socket stays open until then, so the file is still the own one.
144    let remove = |e: anyhow::Error| {
145        if let Err(e) = fs::remove_file(path) {
146            debug!("Unable to remove attach socket {}: {e:#}", path.display());
147        }
148        e
149    };
150    let held = fd
151        .try_clone()
152        .context("duplicate attach socket")
153        .map_err(remove)?;
154    let setup = || -> Result<(fs::Metadata, UnixSeqpacketListener)> {
155        // Without following a symlink, like `FileId::is_at`.
156        let metadata = fs::symlink_metadata(path).context("get attach socket metadata")?;
157        listen(&fd, Backlog::MAXCONN).context("listen on socket fd")?;
158        let listener = UnixSeqpacketListener::try_from(fd).context("create seqpacket listener")?;
159        Ok((metadata, listener))
160    };
161    let (metadata, mut listener) = setup().map_err(remove)?;
162    drop(held);
163
164    let stop = CancellationToken::new();
165    let id = AttachListener::next_id();
166    listeners.insert(
167        path.to_path_buf(),
168        AttachListener {
169            id,
170            file: FileId::new(&metadata),
171            stop: stop.clone(),
172        },
173    );
174    drop(listeners);
175
176    let cleanup = ListenerCleanup {
177        attach,
178        path: path.to_path_buf(),
179        id,
180    };
181    task::spawn(
182        async move {
183            accept_connections(&mut listener, &cleanup.path, &cleanup.attach, token, stop).await;
184            // The cleanup runs before the listener gets closed.
185            drop(cleanup);
186            drop(listener);
187        }
188        .instrument(debug_span!("attach"))
189        .in_current_span(),
190    );
191
192    Ok(())
193}
194
195/// Accept connections until the container exits, which cancels `token`,
196/// or another listener replaces this one, which cancels `stop`.
197async fn accept_connections(
198    listener: &mut UnixSeqpacketListener,
199    path: &Path,
200    attach: &SharedContainerAttach,
201    token: CancellationToken,
202    stop: CancellationToken,
203) {
204    debug!("Start listening on attach socket {}", path.display());
205    let mut accept_errors = listener::AcceptErrors::default();
206    loop {
207        let accepted = select! {
208            accepted = listener.accept() => accepted,
209            () = token.cancelled() => {
210                debug!("Stopping attach listener because token cancelled");
211                return;
212            }
213            () = stop.cancelled() => {
214                debug!("Stopping attach listener because it got replaced");
215                return;
216            }
217        };
218        match accepted {
219            Ok(socket) => {
220                debug!("Got new attach stream connection");
221                let read = Arc::new(socket);
222                let write = read.clone();
223
224                let attach_clone = attach.clone();
225                let token_clone = token.clone();
226                task::spawn(
227                    async move {
228                        if let Err(e) = read_loop(read, attach_clone, token_clone).await {
229                            error!("Attach read loop failure: {e:#}");
230                        }
231                    }
232                    .instrument(debug_span!("read_loop"))
233                    .in_current_span(),
234                );
235
236                let (write_half_rx, output_ended) = attach.subscribe();
237                let attach_clone = attach.clone();
238                let token_clone = token.clone();
239                task::spawn(
240                    async move {
241                        if let Err(e) = write_loop(
242                            write,
243                            attach_clone,
244                            write_half_rx,
245                            output_ended,
246                            token_clone,
247                        )
248                        .await
249                        {
250                            error!("Attach write loop failure: {e:#}");
251                        }
252                    }
253                    .instrument(debug_span!("write_loop"))
254                    .in_current_span(),
255                );
256            }
257            Err(e) => accept_errors.failed(&e).await,
258        }
259    }
260}
261
262/// Forward the stdin of a client until it closes, the stdin of the
263/// container got closed or the container exited, which cancels `token`.
264pub(super) async fn read_loop(
265    socket: Arc<UnixSeqpacket>,
266    attach: SharedContainerAttach,
267    token: CancellationToken,
268) -> Result<()> {
269    let stop_after_stdin_eof = attach.stop_after_stdin_eof.load(Ordering::Acquire);
270    let stopped = async || {
271        select! {
272            () = token.cancelled() => debug!("Exiting because token cancelled"),
273            () = attach.stdin_closed.cancelled() => debug!("Exiting because stdin got closed"),
274        }
275    };
276
277    loop {
278        // A client which stays connected must not keep the loop going once
279        // the container exited or its stdin got closed.
280        let read = select! {
281            read = recv_stdin(&socket) => read,
282            () = stopped() => return Ok(()),
283        };
284
285        match read {
286            Ok((data, truncated)) if !data.is_empty() => {
287                if truncated {
288                    warn!("Truncated attach stdin packet to {} bytes", data.len());
289                }
290                debug!("Read {} stdin bytes from client", data.len());
291                // Waiting for the container to take the data must not
292                // outlive the container or its stdin.
293                select! {
294                    () = attach.send_stdin(data) => {}
295                    () = stopped() => return Ok(()),
296                }
297            }
298            Ok(_) => {
299                // A zero byte read on a SOCK_SEQPACKET socket is either an
300                // empty packet or the peer closing its write side, so
301                // check the socket rather than guessing from another read.
302                if !input_ended(&socket).context("check attach socket for more input")? {
303                    trace!("Ignoring empty packet");
304                    continue;
305                }
306                // The client closed its write half, no more input will
307                // ever arrive on this connection. The write loop keeps
308                // delivering output independently.
309                debug!("Stopping read loop because there is nothing more to read");
310                if stop_after_stdin_eof {
311                    attach.close_container_stdin();
312                }
313                return Ok(());
314            }
315            Err(e) if e.kind() == ErrorKind::ConnectionReset => {
316                // The client closed the connection without reading all of
317                // its output. The error gets reported only once, the
318                // stdin queued before is still readable.
319                debug!("Attach client reset the connection");
320            }
321            Err(e) => return Err(e).context("read from attach socket"),
322        }
323    }
324}
325
326/// Receive a stdin packet into the scratch buffer of the thread, see
327/// [`with_scratch`], and copy it out, so that a connection holds no
328/// buffer while it waits and the queued stdin only the data of the
329/// packets. Stdin is binary, null bytes are part of the data. Returns
330/// whether the packet got truncated as well.
331async fn recv_stdin(socket: &UnixSeqpacket) -> io::Result<(Bytes, bool)> {
332    future::poll_fn(|cx| {
333        with_scratch(STDIN_PACKET_MAX_SIZE, |buf| {
334            let info = ready!(socket.poll_recv(cx, buf))?;
335            let data = Bytes::copy_from_slice(&buf[..info.bytes_read()]);
336            Poll::Ready(Ok((data, info.truncated())))
337        })
338    })
339    .await
340}
341
342/// Whether a send failed because the client closed its end. A client
343/// which leaves output unread gets the connection reset.
344fn client_gone(e: &io::Error) -> bool {
345    matches!(e.kind(), ErrorKind::BrokenPipe | ErrorKind::ConnectionReset)
346}
347
348/// Forward the container output of `attach` to a client until the output
349/// ended or the client went away. Once the container exited, which
350/// cancels `token`, a process left behind may keep the output open, so
351/// forward only what arrives shortly after the exit, see
352/// [`Drain::recv_attach`].
353pub(super) async fn write_loop(
354    socket: Arc<UnixSeqpacket>,
355    attach: SharedContainerAttach,
356    mut rx: Receiver<Message>,
357    output_ended: bool,
358    token: CancellationToken,
359) -> Result<()> {
360    let hangup = hangup(&socket);
361    tokio::pin!(hangup);
362    let mut drain: Option<Drain> = None;
363    let mut lag = Lag::default();
364    loop {
365        // The done message may have been sent before subscribing, so write
366        // it once nothing is queued anymore.
367        let message = if output_ended {
368            rx.try_recv().unwrap_or(Message::Done)
369        } else {
370            let res = select! {
371                // The exit starts the drain once, also while the output
372                // keeps coming.
373                biased;
374                () = token.cancelled(), if drain.is_none() => {
375                    debug!("Draining the output because token cancelled");
376                    drain = Some(Drain::default());
377                    continue;
378                }
379                () = &mut hangup => {
380                    debug!("Exiting because the client hung up");
381                    return Ok(());
382                }
383                res = next_message(&attach, &mut rx, drain.as_mut(), &mut lag) => res,
384            };
385            match res {
386                Ok(message) => message,
387                Err(RecvError::Lagged(n)) => {
388                    lag.skipped(n);
389                    continue;
390                }
391                Err(RecvError::Closed) => {
392                    debug!("Exiting because container I/O is gone");
393                    return Ok(());
394                }
395            }
396        };
397        let draining = drain.is_some();
398        match message {
399            Message::Done => {
400                debug!("Exiting because the container output ended");
401                let packet = [IoSlice::new(DONE_PACKET)];
402                match send_packet(&socket, &packet, hangup.as_mut(), &token, draining).await {
403                    Ok(true) => debug!("Wrote done packet to client"),
404                    Ok(false) => {}
405                    Err(e) => return Err(e).context("write done packet"),
406                }
407                return Ok(());
408            }
409            Message::Data(buf, pipe) => {
410                // A read of a pipe fits into one packet, see
411                // `PACKET_BUF_SIZE`. Send the pipe header and the data as
412                // one packet without copying the data into a separate
413                // buffer.
414                debug_assert!(buf.len() <= PACKET_BUF_SIZE);
415                let header = [pipe.attach_packet_byte()];
416                let packet = [IoSlice::new(&header), IoSlice::new(&buf)];
417                match send_packet(&socket, &packet, hangup.as_mut(), &token, draining).await {
418                    Ok(true) => debug!("Wrote {pipe} packet to client"),
419                    // The client is gone, so stop receiving output for it
420                    // as well.
421                    Ok(false) => return Ok(()),
422                    Err(e) => return Err(e).context("write packet"),
423                }
424            }
425        }
426    }
427}
428
429/// Send a packet, false if the client went away or does not take it in
430/// time. A client which does not read keeps a send waiting, until it
431/// hangs up or, once the container exited, for at most
432/// [`DRAIN_MAX_WAIT`]. Each packet gets that long, so that a slow client
433/// still gets the output which was there already.
434async fn send_packet(
435    socket: &UnixSeqpacket,
436    packet: &[IoSlice<'_>],
437    hangup: Pin<&mut impl Future<Output = ()>>,
438    token: &CancellationToken,
439    draining: bool,
440) -> io::Result<bool> {
441    let stuck = async {
442        if !draining {
443            token.cancelled().await;
444        }
445        time::sleep(DRAIN_MAX_WAIT).await;
446    };
447    select! {
448        res = socket.send_vectored(packet) => match res {
449            Ok(_) => Ok(true),
450            Err(ref e) if client_gone(e) => {
451                debug!("Exiting because the client closed the connection");
452                Ok(false)
453            }
454            Err(e) => Err(e),
455        },
456        () = hangup => {
457            debug!("Exiting because the client hung up");
458            Ok(false)
459        }
460        () = stuck => {
461            debug!("Exiting because the client does not take the output after the exit");
462            Ok(false)
463        }
464    }
465}
466
467/// The next output message, bounded by the drain once the container
468/// exited. If the output did not end after the exit, the client gets
469/// the done packet anyway.
470async fn next_message(
471    attach: &SharedContainerAttach,
472    rx: &mut Receiver<Message>,
473    drain: Option<&mut Drain>,
474    lag: &mut Lag,
475) -> Result<Message, RecvError> {
476    match drain {
477        Some(drain) => Ok(drain
478            .recv_attach(attach, rx, Some(lag))
479            .await
480            .unwrap_or(Message::Done)),
481        None => {
482            let res = rx.recv().await;
483            if res.is_ok() {
484                lag.received(rx);
485            }
486            res
487        }
488    }
489}
490
491/// Wait until the client closed the connection. Without output, a send
492/// never notices, and a client which only closed its write side still
493/// receives the output. The hangup gets reported by an epoll instance of
494/// its own, which watches the socket for nothing but the hangups and
495/// errors epoll always reports. Neither input nor output wakes it up,
496/// and its readiness is separate from the one of the socket, which the
497/// read and write loops use. If that fails, only a failing send ends
498/// the loop.
499async fn hangup(socket: &UnixSeqpacket) {
500    let wait = async || -> Result<()> {
501        let epoll = Epoll::new(EpollCreateFlags::EPOLL_CLOEXEC).context("create epoll")?;
502        epoll
503            .add(socket.as_fd(), EpollEvent::new(EpollFlags::EPOLLET, 0))
504            .context("watch attach socket")?;
505        let ready =
506            async_fd::register(epoll.0.as_fd(), Interest::READABLE).context("register epoll")?;
507        let mut events = [EpollEvent::empty()];
508        loop {
509            let mut guard = ready.readable().await.context("wait for epoll")?;
510            let n = epoll
511                .wait(&mut events, PollTimeout::ZERO)
512                .context("read epoll events")?;
513            if n > 0 {
514                return Ok(());
515            }
516            guard.clear_ready();
517        }
518    };
519    if let Err(e) = wait().await {
520        warn!("Unable to watch for the attach client to hang up: {e:#}");
521        future::pending::<()>().await;
522    }
523}
conmonrs/attach/socket.rs

conmonrs/attach/
socket.rs
