conmonrs/streaming_server/
handshake.rs

1//! The upgrade of a request to a WebSocket or to SPDY, by the protocols the
2//! client asks for.
3
4use super::{MAX_MESSAGE_SIZE, PORT_FORWARD_MAX_MESSAGE_SIZE, Session, State, StreamingServer};
5use crate::{
6    mutex::lock,
7    random_id::RandomId,
8    streaming_server::{
9        port_forward::{self, PortForward, WebSocketProtocol},
10        remotecommand::{Framing, SPDY_PROTOCOLS},
11        stream_protocol::first_supported,
12        web::{self, Response, WebSocket, WebSocketUpgrade},
13    },
14};
15use anyhow::Result;
16use hyper::{
17    HeaderMap, Method, Request, StatusCode,
18    body::Incoming,
19    header::{self, HeaderName, HeaderValue},
20    upgrade::{OnUpgrade, Upgraded},
21};
22use hyper_util::rt::TokioIo;
23use std::{future::Future, sync::Mutex};
24use tokio::task;
25use tracing::{Instrument, Span, debug, debug_span};
26
27/// The upgrade SPDY clients ask for, of exec, attach and port forwarding.
28const SPDY_31: &str = "SPDY/3.1";
29
30/// The streaming protocols of a SPDY upgrade request, and the one the server
31/// picked in the response.
32const PROTOCOL_VERSION_HEADER: HeaderName = HeaderName::from_static("x-stream-protocol-version");
33
34/// The protocols the server supports, if it supports none of the client.
35const ACCEPTED_PROTOCOL_VERSIONS_HEADER: HeaderName =
36    HeaderName::from_static("x-accepted-stream-protocol-versions");
37
38impl StreamingServer {
39    /// Handle an exec or attach request, `kind` is the path of its URL, which
40    /// the client upgrades to a WebSocket or SPDY.
41    pub(super) fn handle(
42        kind: &str,
43        token: RandomId,
44        state: &Mutex<State>,
45        request: Request<Incoming>,
46    ) -> Response {
47        debug_span!("handle_common", ?token).in_scope(|| {
48            debug!("Got request");
49            // An unknown token comes first, the request cannot succeed. The
50            // token of a session of the other kind is unknown on this path,
51            // the session stays for its own.
52            let typ = lock(state).sessions.get(&token).map(Session::typ);
53            if typ != Some(kind) {
54                debug!("Unable to find session");
55                return web::status(StatusCode::NOT_FOUND);
56            }
57            if Self::is_websocket_request(request.headers()) {
58                let requested = Self::websocket_protocols(request.headers());
59                let Some((framing, subprotocol)) =
60                    Framing::negotiate(requested.iter().map(String::as_str))
61                else {
62                    return Self::unsupported_protocols(&requested);
63                };
64                return Self::upgrade_websocket(
65                    request,
66                    subprotocol,
67                    MAX_MESSAGE_SIZE,
68                    || Self::take_session(state, token),
69                    move |socket, session| {
70                        Self::handle_websocket(socket, session, framing)
71                            .instrument(debug_span!("websocket", ?token))
72                    },
73                );
74            }
75
76            Self::upgrade_spdy(
77                request,
78                &SPDY_PROTOCOLS,
79                || Self::take_session(state, token),
80                Self::handle_spdy,
81                debug_span!("spdy", ?token),
82            )
83        })
84    }
85
86    /// Upgrade a request to a WebSocket with the `subprotocol` the server
87    /// picked, and serve it with the session once it got upgraded. The
88    /// session gets taken only for a valid request. The client may send
89    /// messages of up to `max_message_size` bytes.
90    fn upgrade_websocket<S, F>(
91        mut request: Request<Incoming>,
92        subprotocol: Option<&'static str>,
93        max_message_size: usize,
94        take_session: impl FnOnce() -> Option<S>,
95        serve: impl FnOnce(WebSocket, S) -> F + Send + 'static,
96    ) -> Response
97    where
98        S: Send + 'static,
99        F: Future<Output = ()> + Send + 'static,
100    {
101        let ws = match WebSocketUpgrade::from_request(&mut request) {
102            Ok(ws) => ws,
103            Err(rejection) => return *rejection,
104        };
105        let Some(session) = take_session() else {
106            return web::status(StatusCode::NOT_FOUND);
107        };
108        ws.protocol(subprotocol)
109            .max_message_size(max_message_size)
110            .on_upgrade(move |socket| serve(socket, session))
111    }
112
113    /// Answer a SPDY upgrade request with one of the supported protocols and
114    /// serve the connection with the session and what the protocol stands
115    /// for once it got upgraded. The session gets taken only for a valid
116    /// request.
117    fn upgrade_spdy<S, T, F>(
118        mut request: Request<Incoming>,
119        supported: &[(&'static str, T)],
120        take_session: impl FnOnce() -> Option<S>,
121        serve: impl FnOnce(TokioIo<Upgraded>, S, T) -> F + Send + 'static,
122        span: Span,
123    ) -> Response
124    where
125        S: Send + 'static,
126        T: Copy + Send + 'static,
127        F: Future<Output = ()> + Send + 'static,
128    {
129        let (value, protocol) = match Self::spdy_handshake(request.headers(), supported) {
130            Ok(protocol) => protocol,
131            Err(response) => return *response,
132        };
133        // A HEAD gets routed like a GET, but must not take the session.
134        if !matches!(*request.method(), Method::GET | Method::POST) {
135            let mut response = web::status(StatusCode::METHOD_NOT_ALLOWED);
136            response
137                .headers_mut()
138                .insert(header::ALLOW, HeaderValue::from_static("GET,POST"));
139            return response;
140        }
141        let Some(on_upgrade) = request.extensions_mut().remove::<OnUpgrade>() else {
142            return web::text(
143                StatusCode::INTERNAL_SERVER_ERROR,
144                "unable to upgrade: connection does not support upgrades",
145            );
146        };
147        let Some(session) = take_session() else {
148            return web::status(StatusCode::NOT_FOUND);
149        };
150        task::spawn(
151            async move {
152                match on_upgrade.await {
153                    Ok(upgraded) => serve(TokioIo::new(upgraded), session, value).await,
154                    Err(e) => debug!("Unable to upgrade connection: {e}"),
155                }
156            }
157            .instrument(span),
158        );
159        Self::switching_protocols(protocol)
160    }
161
162    /// The response of a successful SPDY upgrade.
163    fn switching_protocols(protocol: &'static str) -> Response {
164        let mut response = web::status(StatusCode::SWITCHING_PROTOCOLS);
165        let headers = response.headers_mut();
166        headers.insert(header::CONNECTION, HeaderValue::from_static("Upgrade"));
167        headers.insert(header::UPGRADE, HeaderValue::from_static(SPDY_31));
168        headers.insert(PROTOCOL_VERSION_HEADER, HeaderValue::from_static(protocol));
169        response
170    }
171
172    /// The WebSocket subprotocols a client requested.
173    fn websocket_protocols(headers: &HeaderMap) -> Vec<String> {
174        web::header_tokens(headers, header::SEC_WEBSOCKET_PROTOCOL)
175            .map(str::to_owned)
176            .collect()
177    }
178
179    /// The answer to a WebSocket upgrade without a supported subprotocol.
180    fn unsupported_protocols(requested: &[String]) -> Response {
181        web::text(
182            StatusCode::FORBIDDEN,
183            format!("unable to upgrade: requested protocol(s) are not supported: {requested:?}"),
184        )
185    }
186
187    /// Handle a port forward request, which the client upgrades to SPDY.
188    pub(super) fn handle_port_forward(
189        token: RandomId,
190        state: &Mutex<State>,
191        request: Request<Incoming>,
192    ) -> Response {
193        debug_span!("handle_port_forward", ?token).in_scope(|| {
194            debug!("Got port forward request");
195            // An unknown token comes first, the request cannot succeed.
196            if !lock(state).port_forwards.contains_key(&token) {
197                debug!("Unable to find port forward session");
198                return web::status(StatusCode::NOT_FOUND);
199            }
200            if Self::is_websocket_request(request.headers()) {
201                return Self::upgrade_port_forward_websocket(token, state, request);
202            }
203            Self::upgrade_spdy(
204                request,
205                &[(port_forward::PROTOCOL, ())],
206                || Self::take_port_forward(state, token),
207                |io, session: PortForward, ()| session.serve_spdy(io),
208                debug_span!("port_forward", ?token),
209            )
210        })
211    }
212
213    fn is_websocket_request(headers: &HeaderMap) -> bool {
214        web::header_has_token(headers, header::UPGRADE, "websocket")
215    }
216
217    /// Upgrade a port forward request to a WebSocket, which tunnels SPDY or
218    /// speaks the channel protocol, by the subprotocols of the client.
219    fn upgrade_port_forward_websocket(
220        token: RandomId,
221        state: &Mutex<State>,
222        request: Request<Incoming>,
223    ) -> Response {
224        let requested = Self::websocket_protocols(request.headers());
225        let Some((protocol, subprotocol)) =
226            WebSocketProtocol::negotiate(requested.iter().map(String::as_str))
227        else {
228            return Self::unsupported_protocols(&requested);
229        };
230        let ports = match protocol {
231            WebSocketProtocol::SpdyTunnel => vec![],
232            WebSocketProtocol::Channels { .. } => {
233                match PortForward::query_ports(request.uri().query()) {
234                    Ok(ports) => ports,
235                    Err(e) => return web::text(StatusCode::BAD_REQUEST, format!("{e:#}")),
236                }
237            }
238        };
239
240        Self::upgrade_websocket(
241            request,
242            subprotocol,
243            PORT_FORWARD_MAX_MESSAGE_SIZE,
244            || Self::take_port_forward(state, token),
245            move |socket, session: PortForward| {
246                session
247                    .serve_websocket(socket, protocol, ports)
248                    .instrument(debug_span!("port_forward", ?token))
249            },
250        )
251    }
252
253    /// Check that the request asks for a SPDY/3.1 upgrade and negotiate one
254    /// of the supported protocols, by the order of the client. The rejections
255    /// are the ones of the SPDY server of Kubernetes.
256    fn spdy_handshake<T: Copy>(
257        headers: &HeaderMap,
258        supported: &[(&'static str, T)],
259    ) -> Result<(T, &'static str), Box<Response>> {
260        if !web::header_has_token(headers, header::CONNECTION, "upgrade")
261            || !web::header_has_token(headers, header::UPGRADE, SPDY_31)
262        {
263            return Err(Box::new(web::text(
264                StatusCode::BAD_REQUEST,
265                "unable to upgrade: missing WebSocket or SPDY/3.1 upgrade headers in request",
266            )));
267        }
268
269        let protocols: Vec<_> = web::header_tokens(headers, PROTOCOL_VERSION_HEADER).collect();
270        if protocols.is_empty() {
271            return Err(Box::new(web::text(
272                StatusCode::BAD_REQUEST,
273                format!("unable to upgrade: header {PROTOCOL_VERSION_HEADER} is required"),
274            )));
275        }
276        first_supported(protocols.iter().copied(), supported).ok_or_else(|| {
277            let supported: Vec<_> = supported.iter().map(|(protocol, _)| *protocol).collect();
278            let mut response = web::text(
279                StatusCode::FORBIDDEN,
280                format!(
281                    "unable to upgrade: unable to negotiate protocol: client supports {protocols:?}, server accepts {supported:?}"
282                ),
283            );
284            for protocol in supported {
285                response.headers_mut().append(
286                    ACCEPTED_PROTOCOL_VERSIONS_HEADER,
287                    HeaderValue::from_static(protocol),
288                );
289            }
290            Box::new(response)
291        })
292    }
293}
294
295/// The entry points of the fuzz targets.
296#[cfg(any(test, feature = "fuzzing"))]
297pub(super) mod fuzzing;
298
299#[cfg(test)]
300mod tests;
conmonrs/streaming_server/handshake.rs

conmonrs/streaming_server/
handshake.rs
