conmonrs/
streaming_server.rs

1//! The streaming server of exec, attach and port forward sessions: the URLs
2//! with their one time tokens and the HTTP server their clients connect to.
3
4use crate::{listener, mutex::lock, random_id::RandomId};
5use anyhow::{Context, Result, bail};
6use bounded_map::BoundedMap;
7use hyper::{
8    Method, Request, StatusCode,
9    body::Incoming,
10    header::{self, HeaderValue},
11    server::conn::http1,
12    service::service_fn,
13};
14use hyper_util::rt::{TokioIo, TokioTimer};
15use port_forward::PortForward;
16use std::{
17    convert::Infallible,
18    io,
19    net::Ipv4Addr,
20    path::PathBuf,
21    sync::{Arc, Mutex, Weak},
22};
23use stream_protocol::ClientStr;
24use tokio::{
25    net::{TcpListener, TcpStream},
26    sync::{OnceCell, Semaphore},
27    task,
28    time::{self, Duration},
29};
30use tracing::{Level, debug, info, trace};
31use web::Response;
32
33mod attach;
34mod bounded_map;
35mod budget;
36mod channel;
37mod exec;
38mod handshake;
39mod idle;
40mod port_forward;
41mod remotecommand;
42mod session;
43mod spdy;
44mod stream_protocol;
45#[cfg(test)]
46mod testing;
47mod web;
48
49pub(crate) use attach::AttachSession;
50#[cfg(test)]
51pub(crate) use bounded_map::DEFAULT_CAPACITY;
52pub(crate) use exec::ExecSession;
53
54/// The entry points of the fuzz targets.
55#[cfg(any(test, feature = "fuzzing"))]
56pub(crate) mod fuzzing {
57    pub use super::{
58        channel::fuzzing::channel as channel_decode,
59        handshake::fuzzing::upgrade_headers,
60        port_forward::fuzzing::{query_ports as port_forward_query, spdy as port_forward_spdy},
61        remotecommand::fuzzing::{resize_events, spdy as remotecommand_spdy},
62        session::fuzzing::{client_messages, pending_stdin},
63        spdy::fuzzing::{
64            connection as spdy_connection, frames as spdy_frames, header_block as spdy_header_block,
65        },
66    };
67}
68
69const ADDR: Ipv4Addr = Ipv4Addr::LOCALHOST;
70
71/// How long a client may take to send the head of its request, which only
72/// applies with a timer. A connection which never sends one would stay open
73/// otherwise. Clients send it right after they connected, and every
74/// connection serves one request, see [`MAX_CONNECTIONS`]. Far below the 30
75/// seconds of hyper, so that a connection without a request holds its
76/// permit for a short time only. That does not bound how long a local
77/// process can hold the server off: one which keeps opening connections,
78/// holds all permits and keeps the backlog of the listener full makes the
79/// connection of another client wait behind its own ones, a few times
80/// [`UPGRADE_TIMEOUT`], or get its handshake dropped and retried while the
81/// backlog is full, for as long as it keeps that up. It cannot use up the
82/// file descriptors of the server.
83const HEADER_READ_TIMEOUT: Duration = if cfg!(test) {
84    Duration::from_secs(1)
85} else {
86    Duration::from_secs(5)
87};
88
89/// How long a connection may take until it got upgraded, also one which
90/// does not read its response. Clients make one upgrade request per
91/// connection, hyper has no timeout for writing the response.
92const UPGRADE_TIMEOUT: Duration = HEADER_READ_TIMEOUT.saturating_mul(2);
93
94/// The most connections which did not get upgraded yet. More wait to be
95/// accepted, so that local processes which open connections without a
96/// request cannot use up the descriptors the whole pod needs. Upgraded
97/// connections do not count, their tokens bound them. There is no limit per
98/// peer, the server listens on the loopback address only, where the kubelet
99/// and a local process which holds connections look the same.
100const MAX_CONNECTIONS: usize = if cfg!(test) { 8 } else { 256 };
101
102/// The most a connection buffers of the head of its request, the request
103/// line and the headers, see [`MAX_CONNECTIONS`]. A longer one gets a 431
104/// response. hyper takes about 400 KiB otherwise, which lets local processes
105/// make the server hold about 100 MiB without a token. The kubelet sends
106/// the headers of the client and its own, a few KiB, Go servers take up to
107/// 1 MiB by default, nginx 32 KiB.
108const MAX_HEADER_BUF_SIZE: usize = 64 * 1024;
109
110const EXEC_PATH: &str = "exec";
111const ATTACH_PATH: &str = "attach";
112const PORT_FORWARD_PATH: &str = "portforward";
113
114/// The largest WebSocket message and frame a client of exec or attach may
115/// send. Clients send stdin, which kubectl reads in 32 KiB chunks and the
116/// attach sockets take in packets of up to 33 KiB, and small resize events. The
117/// Python client sends all stdin of a write in one message, 1 MiB at a time
118/// for `k8s_cp` of Ansible, a third more in base64. The defaults of 64 MiB
119/// per message, or the 32 MiB of Kubernetes, would let a client make the
120/// server buffer that much per message. A client which sends more fails its
121/// session with a status and a close which tell why.
122const MAX_MESSAGE_SIZE: usize = 4 * 1024 * 1024;
123
124/// The largest WebSocket message and frame a client of port forwarding may
125/// send. A single message must not carry more than a connection takes while
126/// it gets established, the `CONNECT_BUFFER` of [`port_forward`], nor all
127/// that the ports of a WebSocket hold together, its `CHANNEL_BUFFER`. A
128/// client which sends more gets the WebSocket closed with a code which tells
129/// why.
130const PORT_FORWARD_MAX_MESSAGE_SIZE: usize = 1024 * 1024;
131
132/// The error of a request for a URL while too many of its kind wait for
133/// their client. Clients match it, see `ErrTooManyRequests` of the Go client,
134/// so it has to stay the start of the message.
135const TOO_MANY_REQUESTS: &str = "maximum number of in-flight requests exceeded";
136
137/// How often the sessions whose client did not come in time get dropped,
138/// while sessions wait.
139const SWEEP_INTERVAL: Duration = Duration::from_secs(10);
140
141/// Serves the exec, attach and port forward sessions over HTTP on the
142/// loopback address, each under a one time URL, see
143/// [`StreamingServer::exec_url`].
144#[derive(Debug, Default)]
145pub(crate) struct StreamingServer {
146    /// The port the server listens on, once it got started for the first
147    /// session.
148    port: OnceCell<u16>,
149    state: Arc<Mutex<State>>,
150}
151
152/// State handled by the streaming server, the sessions waiting for their
153/// client by token. Never locked across an await.
154#[derive(Debug, Default)]
155struct State {
156    sessions: BoundedMap<RandomId, Session>,
157    port_forwards: BoundedMap<RandomId, PortForward>,
158
159    /// Whether a task drops the sessions whose client did not come in time,
160    /// see [`StreamingServer::sweep`].
161    sweeping: bool,
162}
163
164/// A dedicated session for each provided functionality. Not `Debug`, the
165/// command of an exec may carry secrets.
166enum Session {
167    Exec(Box<ExecSession>),
168    Attach(AttachSession),
169}
170
171/// What the exec and attach sessions have in common.
172#[derive(Debug)]
173pub(crate) struct SessionCommon {
174    pub(crate) container_id: Box<str>,
175    pub(crate) stdin: bool,
176    pub(crate) stdout: bool,
177    pub(crate) stderr: bool,
178    pub(crate) idle_timeout: Option<Duration>,
179}
180
181impl StreamingServer {
182    /// Start the streaming server if not already running and return its
183    /// port.
184    async fn start_if_required(&self) -> Result<u16> {
185        let port = self.port.get_or_try_init(|| async {
186            let listener = TcpListener::bind((ADDR, 0))
187                .await
188                .context("bind streaming server")?;
189
190            let local_addr = listener
191                .local_addr()
192                .context("get listeners local address")?;
193
194            info!("Starting streaming server on {local_addr}");
195            task::spawn_local(Self::serve(listener, self.state.clone()));
196            Ok::<_, anyhow::Error>(local_addr.port())
197        });
198        port.await.copied().context("start streaming server")
199    }
200
201    /// The port of the started server.
202    #[cfg(test)]
203    fn port(&self) -> u16 {
204        *self.port.get().expect("streaming server started")
205    }
206
207    /// Serve the main streaming server, every connection in a task of its
208    /// own, at most [`MAX_CONNECTIONS`] at once, each for one request.
209    async fn serve(listener: TcpListener, state: Arc<Mutex<State>>) {
210        let connections = Arc::new(Semaphore::new(MAX_CONNECTIONS));
211        let mut accept_errors = listener::AcceptErrors::default();
212        loop {
213            let Ok(permit) = connections.clone().acquire_owned().await else {
214                return;
215            };
216            let stream = match Self::accept(&listener).await {
217                Ok(stream) => stream,
218                Err(e) => {
219                    accept_errors.failed(&e).await;
220                    continue;
221                }
222            };
223            let state = state.clone();
224            task::spawn(async move {
225                // Until the connection ended or got upgraded.
226                let _permit = permit;
227                let service = service_fn(move |request| {
228                    let response = Self::route(request, &state);
229                    async move { Ok::<_, Infallible>(response) }
230                });
231                let connection = http1::Builder::new()
232                    .timer(TokioTimer::new())
233                    .header_read_timeout(HEADER_READ_TIMEOUT)
234                    .max_buf_size(MAX_HEADER_BUF_SIZE)
235                    .serve_connection(TokioIo::new(stream), service)
236                    .with_upgrades();
237                match time::timeout(UPGRADE_TIMEOUT, connection).await {
238                    Ok(Err(e)) => trace!("Unable to serve connection: {e}"),
239                    Ok(Ok(())) => {}
240                    Err(_) => debug!("Closing connection which did not end in time"),
241                }
242            });
243        }
244    }
245
246    /// Accept the next connection. Like the Go server of CRI-O without
247    /// Nagle's algorithm, so that small writes in a row, like the last output
248    /// and the status of a session, do not wait for a delayed ACK.
249    async fn accept(listener: &TcpListener) -> io::Result<TcpStream> {
250        let (stream, _) = listener.accept().await?;
251        if let Err(e) = stream.set_nodelay(true) {
252            debug!("Unable to disable Nagle's algorithm: {e}");
253        }
254        Ok(stream)
255    }
256
257    /// Drop the sessions whose client did not come in time, otherwise they
258    /// would stay until the next request. The task ends once no session
259    /// waits or the server is gone, so that an idle server does not wake up.
260    async fn sweep(state: Weak<Mutex<State>>) {
261        loop {
262            time::sleep(SWEEP_INTERVAL).await;
263            let Some(state) = state.upgrade() else {
264                return;
265            };
266            let mut state = lock(&state);
267            state.sessions.sweep();
268            state.port_forwards.sweep();
269            if state.sessions.is_empty() && state.port_forwards.is_empty() {
270                state.sweeping = false;
271                return;
272            }
273        }
274    }
275
276    /// Route a request and log it with the status of its response.
277    fn route(request: Request<Incoming>, state: &Mutex<State>) -> Response {
278        let logged = tracing::enabled!(Level::DEBUG).then(|| {
279            (
280                request.method().clone(),
281                Self::logged_path(request.uri().path()),
282            )
283        });
284        let mut response = Self::dispatch(request, state);
285        if let Some((method, path)) = logged {
286            debug!("{method} {path}: {}", response.status());
287        }
288        // One request per connection, clients make one upgrade request each.
289        // A client which keeps sending requests without reading the
290        // responses would hold its connection otherwise, see
291        // [`MAX_CONNECTIONS`]. Turning keep-alive off for the connection
292        // would add the close to the upgrades as well.
293        if response.status() != StatusCode::SWITCHING_PROTOCOLS {
294            response
295                .headers_mut()
296                .insert(header::CONNECTION, HeaderValue::from_static("close"));
297        }
298        response
299    }
300
301    /// The path of a request for the logs, which shows only the start of its
302    /// token, see [`RandomId`].
303    fn logged_path(path: &str) -> String {
304        match path.strip_prefix('/').and_then(|path| path.split_once('/')) {
305            Some((kind, token)) => {
306                let start: String = token.chars().take(8).collect();
307                format!("/{:?}/{:?}...", ClientStr(kind), ClientStr(&start))
308            }
309            None => format!("{:?}", ClientStr(path)),
310        }
311    }
312
313    /// Hand a request to the handler of its path, `/<kind>/<token>`. Like a
314    /// GET, a HEAD gets handled, the response has no body then.
315    fn dispatch(request: Request<Incoming>, state: &Mutex<State>) -> Response {
316        let Some((kind, token)) = request
317            .uri()
318            .path()
319            .strip_prefix('/')
320            .and_then(|path| path.split_once('/'))
321        else {
322            return web::status(StatusCode::NOT_FOUND);
323        };
324        // The path of the kind, which outlives the request.
325        let Some(kind) = [EXEC_PATH, ATTACH_PATH, PORT_FORWARD_PATH]
326            .into_iter()
327            .find(|path| *path == kind)
328        else {
329            return web::status(StatusCode::NOT_FOUND);
330        };
331        if token.is_empty() || token.contains('/') {
332            return web::status(StatusCode::NOT_FOUND);
333        }
334        if !matches!(*request.method(), Method::GET | Method::HEAD | Method::POST) {
335            let mut response = web::status(StatusCode::METHOD_NOT_ALLOWED);
336            response
337                .headers_mut()
338                .insert(header::ALLOW, HeaderValue::from_static("GET,HEAD,POST"));
339            return response;
340        }
341        let token = match token.parse() {
342            Ok(token) => token,
343            Err(e) => return web::text(StatusCode::BAD_REQUEST, format!("Invalid URL: {e}")),
344        };
345        if kind == PORT_FORWARD_PATH {
346            Self::handle_port_forward(token, state, request)
347        } else {
348            Self::handle(kind, token, state, request)
349        }
350    }
351
352    /// Start the server if required and return the URL of the exec session.
353    pub(crate) async fn exec_url(&self, session: ExecSession) -> Result<String> {
354        self.wait_for_client(
355            EXEC_PATH,
356            |state| &mut state.sessions,
357            Session::Exec(session.into()),
358        )
359        .await
360    }
361
362    /// Start the server if required and return the URL of the attach
363    /// session.
364    pub(crate) async fn attach_url(&self, session: AttachSession) -> Result<String> {
365        self.wait_for_client(
366            ATTACH_PATH,
367            |state| &mut state.sessions,
368            Session::Attach(session),
369        )
370        .await
371    }
372
373    /// Start the server if required and return the URL for forwarding ports
374    /// into the network namespace.
375    pub(crate) async fn port_forward_url(
376        &self,
377        netns_path: PathBuf,
378        idle_timeout: Option<Duration>,
379    ) -> Result<String> {
380        self.wait_for_client(
381            PORT_FORWARD_PATH,
382            |state| &mut state.port_forwards,
383            PortForward::new(netns_path, idle_timeout),
384        )
385        .await
386    }
387
388    /// Start the server if required and keep a session in its map until its
389    /// client takes it, return its URL under `path`. Too many sessions which
390    /// wait for their client fail a new one, like for the kubelet, instead
391    /// of dropping one whose client would come.
392    async fn wait_for_client<V>(
393        &self,
394        path: &str,
395        sessions: impl FnOnce(&mut State) -> &mut BoundedMap<RandomId, V>,
396        session: V,
397    ) -> Result<String> {
398        let port = self.start_if_required().await?;
399        let token = RandomId::new();
400        let mut state = lock(&self.state);
401        if sessions(&mut state).insert(token, session).is_err() {
402            bail!(TOO_MANY_REQUESTS);
403        }
404        if !state.sweeping {
405            state.sweeping = true;
406            task::spawn(Self::sweep(Arc::downgrade(&self.state)));
407        }
408        Ok(format!("http://{ADDR}:{port}/{path}/{token}"))
409    }
410
411    /// Take the exec or attach session of a valid request. Requests which get
412    /// rejected before leave the token, so that the client can retry.
413    fn take_session(state: &Mutex<State>, token: RandomId) -> Option<Session> {
414        let Some(session) = lock(state).sessions.remove(&token) else {
415            debug!("Session got taken by another request or expired");
416            return None;
417        };
418        // Only what identifies the session, the command of an exec may carry
419        // secrets.
420        info!(
421            "Got valid {} session for container {} and token {token:?}",
422            session.typ(),
423            session.common().container_id
424        );
425        Some(session)
426    }
427
428    /// Take the session of a valid request. Requests which get rejected
429    /// before leave the token, so that the client can retry. Another request
430    /// may have taken it meanwhile.
431    fn take_port_forward(state: &Mutex<State>, token: RandomId) -> Option<PortForward> {
432        let session = lock(state).port_forwards.remove(&token);
433        if session.is_none() {
434            debug!("Port forward session got taken by another request or expired");
435        }
436        session
437    }
438}
439
440impl Session {
441    /// The kind of the session for the logs, the path of its URL.
442    fn typ(&self) -> &'static str {
443        match self {
444            Session::Exec(_) => EXEC_PATH,
445            Session::Attach(_) => ATTACH_PATH,
446        }
447    }
448
449    /// What both kinds of sessions have.
450    fn common(&self) -> &SessionCommon {
451        match self {
452            Session::Exec(s) => &s.common,
453            Session::Attach(s) => &s.common,
454        }
455    }
456
457    /// Whether the process of the session has a terminal.
458    fn terminal(&self) -> bool {
459        match self {
460            Session::Exec(s) => s.tty,
461            Session::Attach(s) => s.child.io().attach().is_terminal(),
462        }
463    }
464}
465
466#[cfg(test)]
467mod tests;
conmonrs/streaming_server.rs

conmonrs/
streaming_server.rs
