conmonrs/streaming_server/
channel.rs

1//! The channel framing of the Kubernetes WebSocket protocols, which exec,
2//! attach and port forwarding share: every message starts with its channel,
3//! base64 protocols send text messages with the channel as an ASCII digit
4//! offset and the payload base64 encoded.
5
6use crate::streaming_server::web::Message;
7use bytes::{Buf, BufMut, Bytes, BytesMut};
8use data_encoding::{BASE64, Encoding};
9use std::sync::LazyLock;
10use tracing::debug;
11
12/// Decodes the base64 payloads of clients like `base64.StdEncoding` of Go,
13/// which the reference implementation of the protocols uses: line breaks
14/// get ignored and the bits of the last symbol which are no data may be
15/// set. Clients which wrap their base64 or encode it less strictly work
16/// with the kubelet, so they have to work here.
17static BASE64_LENIENT: LazyLock<Encoding> = LazyLock::new(|| {
18    let mut spec = BASE64.specification();
19    spec.ignore.push_str("\r\n");
20    spec.check_trailing_bits = false;
21    // The specification of the standard encoding with two options changed
22    // stays valid, the tests would notice the strict one.
23    spec.encoding().unwrap_or(BASE64)
24});
25
26/// A message of the server on a channel.
27pub(super) fn message(base64: bool, channel: u8, data: &[u8]) -> Message {
28    if base64 {
29        let mut text = String::with_capacity(1 + data.len().div_ceil(3) * 4);
30        text.push(char::from(b'0' + channel));
31        BASE64.encode_append(data, &mut text);
32        Message::Text(text.into())
33    } else {
34        let mut frame = BytesMut::with_capacity(1 + data.len());
35        frame.put_u8(channel);
36        frame.put_slice(data);
37        Message::Binary(frame.freeze())
38    }
39}
40
41/// The channel and the payload of a message of the client, none for an
42/// empty one or invalid base64. Binary payloads do not get copied.
43pub(super) fn decode(base64: bool, mut message: Bytes) -> Option<(u8, Bytes)> {
44    let &channel = message.first()?;
45    if !base64 {
46        message.advance(1);
47        return Some((channel, message));
48    }
49    match BASE64_LENIENT.decode(&message[1..]) {
50        Ok(payload) => Some((channel.wrapping_sub(b'0'), payload.into())),
51        Err(e) => {
52            debug!("Ignoring message with invalid base64: {e}");
53            None
54        }
55    }
56}
57
58/// The entry points of the fuzz targets.
59#[cfg(any(test, feature = "fuzzing"))]
60pub(super) mod fuzzing {
61    use super::*;
62
63    /// Decode the input as a message of a client. A message of the server
64    /// with its first byte as the channel and the rest as the payload decodes
65    /// to them again, for channels which base64 messages can carry as a
66    /// single ASCII digit offset. Returns whether the input decoded.
67    pub fn channel(base64: bool, input: &[u8]) -> bool {
68        let decoded = decode(base64, Bytes::copy_from_slice(input)).is_some();
69        let Some((&channel, payload)) = input.split_first() else {
70            return decoded;
71        };
72        if base64 && channel > 0x7f - b'0' {
73            return decoded;
74        }
75        let message = match message(base64, channel, payload) {
76            Message::Binary(data) => data,
77            Message::Text(text) => Bytes::from(text),
78            message => panic!("unexpected message {message:?}"),
79        };
80        assert_eq!(
81            decode(base64, message),
82            Some((channel, Bytes::copy_from_slice(payload)))
83        );
84        decoded
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91
92    #[test]
93    fn messages_get_framed() {
94        assert_eq!(
95            message(false, 1, b"out"),
96            Message::Binary(Bytes::from_static(b"\x01out"))
97        );
98        assert_eq!(message(true, 2, b"err"), Message::Text("2ZXJy".into()));
99        assert_eq!(message(true, 1, b""), Message::Text("1".into()));
100    }
101
102    #[test]
103    fn messages_get_decoded() {
104        assert_eq!(
105            decode(false, Bytes::from_static(b"\x00ls")),
106            Some((0, Bytes::from_static(b"ls")))
107        );
108        assert_eq!(
109            decode(true, Bytes::from_static(b"0bHM=")),
110            Some((0, Bytes::from_static(b"ls")))
111        );
112        assert_eq!(decode(false, Bytes::new()), None);
113        assert_eq!(decode(true, Bytes::from_static(b"0!")), None);
114    }
115
116    /// What `base64.StdEncoding.Decode` of Go takes and rejects.
117    #[test]
118    fn base64_gets_decoded_like_go() {
119        let ls = Some((0, Bytes::from_static(b"ls")));
120        // Line breaks anywhere.
121        assert_eq!(decode(true, Bytes::from_static(b"0bH\nM=")), ls);
122        assert_eq!(decode(true, Bytes::from_static(b"0b\r\nHM=\r\n")), ls);
123        assert_eq!(
124            decode(true, Bytes::from_static(b"0\n")),
125            Some((0, Bytes::new()))
126        );
127        // Trailing bits which are not zero.
128        assert_eq!(decode(true, Bytes::from_static(b"0bHN=")), ls);
129        assert_eq!(
130            decode(true, Bytes::from_static(b"0bB==")),
131            Some((0, Bytes::from_static(b"l")))
132        );
133        // Still invalid: other white space, missing padding.
134        assert_eq!(decode(true, Bytes::from_static(b"0bH M=")), None);
135        assert_eq!(decode(true, Bytes::from_static(b"0bHM")), None);
136    }
137}
conmonrs/streaming_server/channel.rs

conmonrs/streaming_server/
channel.rs
