conmonrs/streaming_server/
stream_protocol.rs

1//! What the streaming protocols of Kubernetes share between exec, attach and
2//! port forwarding, whatever the transport.
3
4use std::{fmt, time::Duration};
5
6/// The header which tells what a stream of a Kubernetes client is for.
7pub(super) const STREAM_TYPE_HEADER: &str = "streamType";
8
9/// How long a Kubernetes client may take to open the streams which belong
10/// together, the default of the kubelet and CRI-O.
11pub(super) const STREAM_CREATION_TIMEOUT: Duration = Duration::from_secs(30);
12
13/// The first of the protocols a client requested which the server supports,
14/// in the order of the client, with its name.
15pub(super) fn first_supported<'a, T: Copy>(
16    requested: impl IntoIterator<Item = &'a str>,
17    supported: &[(&'static str, T)],
18) -> Option<(T, &'static str)> {
19    requested.into_iter().find_map(|protocol| {
20        supported
21            .iter()
22            .find_map(|(name, value)| (*name == protocol).then_some((*value, *name)))
23    })
24}
25
26/// How much of what a client sent a log shows.
27const MAX_LOGGED_LEN: usize = 64;
28
29/// A string of a client for the logs: quoted and escaped, so that it cannot
30/// add lines to the log, and cut off after [`MAX_LOGGED_LEN`] characters.
31pub(super) struct ClientStr<'a>(pub &'a str);
32
33impl fmt::Debug for ClientStr<'_> {
34    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
35        match self.0.char_indices().nth(MAX_LOGGED_LEN) {
36            Some((end, _)) => write!(f, "{:?}...", &self.0[..end]),
37            None => write!(f, "{:?}", self.0),
38        }
39    }
40}
41
42/// Bytes of a client for the logs, like [`ClientStr`].
43pub(super) struct ClientBytes<'a>(pub &'a [u8]);
44
45impl fmt::Debug for ClientBytes<'_> {
46    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47        let end = self.0.len().min(MAX_LOGGED_LEN);
48        write!(f, "\"{}\"", self.0[..end].escape_ascii())?;
49        if end < self.0.len() {
50            write!(f, "...")?;
51        }
52        Ok(())
53    }
54}
55
56#[cfg(test)]
57mod tests {
58    use super::*;
59
60    #[test]
61    fn client_strings_get_escaped_and_cut_off() {
62        assert_eq!(format!("{:?}", ClientStr("a\nb")), r#""a\nb""#);
63        let long = "x".repeat(MAX_LOGGED_LEN + 1);
64        assert_eq!(
65            format!("{:?}", ClientStr(&long)),
66            format!("{:?}...", &long[..MAX_LOGGED_LEN])
67        );
68        assert_eq!(
69            format!("{:?}", ClientStr(&long[1..])),
70            format!("{:?}", &long[1..])
71        );
72        // Not within a character.
73        let wide = "\u{e9}".repeat(MAX_LOGGED_LEN + 1);
74        assert!(format!("{:?}", ClientStr(&wide)).ends_with("\"..."));
75
76        assert_eq!(format!("{:?}", ClientBytes(b"a\n\xff")), r#""a\n\xff""#);
77        assert_eq!(
78            format!("{:?}", ClientBytes(long.as_bytes())),
79            format!("\"{}\"...", &long[..MAX_LOGGED_LEN])
80        );
81    }
82}
conmonrs/streaming_server/stream_protocol.rs

conmonrs/streaming_server/
stream_protocol.rs
