conmonrs/streaming_server/spdy/
frame.rs

1//! The SPDY/3.1 frames and their encoding, see
2//! <https://www.chromium.org/spdy/spdy-protocol/spdy-protocol-draft3-1/>.
3//!
4//! Header blocks are compressed with zlib and a preset dictionary. Each
5//! direction of a connection keeps one compression context for all its header
6//! blocks, so every frame which carries one has to pass the same [`Decoder`]
7//! or [`Encoder`], also the ones which are otherwise ignored. The server only
8//! sends empty header blocks, the [`Encoder`] stores them uncompressed in the
9//! zlib stream instead of compressing them.
10
11use anyhow::{Context, Result, ensure};
12use bytes::{Buf, BufMut, Bytes, BytesMut};
13use flate2::{Decompress, FlushDecompress, Status};
14use std::{fmt, io};
15use tokio::io::{AsyncRead, AsyncReadExt};
16
17/// The protocol version of the control frames.
18const VERSION: u16 = 3;
19
20/// The length of the common header of all frames.
21const HEADER_LEN: usize = 8;
22
23const TYPE_SYN_STREAM: u16 = 1;
24const TYPE_SYN_REPLY: u16 = 2;
25const TYPE_RST_STREAM: u16 = 3;
26const TYPE_PING: u16 = 6;
27const TYPE_GOAWAY: u16 = 7;
28const TYPE_HEADERS: u16 = 8;
29
30/// The flag which ends the sending side of a stream, on SYN_STREAM, SYN_REPLY,
31/// HEADERS and data frames.
32const FLAG_FIN: u8 = 0x01;
33
34/// The largest control frame a client may send. The ones Kubernetes clients
35/// send carry a few short headers.
36const MAX_CONTROL_FRAME_LEN: usize = 64 * 1024;
37
38/// The largest frame the protocol allows, its length has 24 bits.
39pub(super) const MAX_DATA_FRAME_LEN: usize = 0xff_ffff;
40
41/// How much of a data frame the server reads at once. Larger frames get
42/// handed to their stream in parts, so that the server does not buffer them
43/// as a whole. Go clients send 32 KiB per frame.
44const DATA_CHUNK_LEN: usize = 64 * 1024;
45
46/// The largest decompressed header block, which bounds what a small
47/// compressed block may expand to. Kubernetes clients send around a hundred
48/// bytes, a stream holds its headers as long as it is open. A larger block
49/// fails the connection, like any header block which does not decode, the
50/// compression context of the connection would be out of sync otherwise.
51const MAX_HEADER_BLOCK_LEN: usize = 8 * 1024;
52
53/// The most headers of a stream, Kubernetes clients send up to three. Each
54/// takes more memory than the bytes of its block.
55const MAX_HEADERS: usize = 32;
56
57/// RST_STREAM status: the frame violates the protocol.
58pub(super) const STATUS_PROTOCOL_ERROR: u32 = 1;
59
60/// RST_STREAM status: the stream got refused before any processing.
61pub(super) const STATUS_REFUSED_STREAM: u32 = 3;
62
63/// RST_STREAM status: the stream is not needed anymore. spdystream resets
64/// with it as well, also for streams it rejects.
65pub(super) const STATUS_CANCEL: u32 = 5;
66
67/// GOAWAY status: the connection ends normally.
68pub(super) const GOAWAY_OK: u32 = 0;
69
70/// GOAWAY status: the client violated the protocol.
71pub(super) const GOAWAY_PROTOCOL_ERROR: u32 = 1;
72
73/// A single SPDY frame. Only the frames and fields the server acts on are
74/// represented, see [`Frame::Ignored`]. Its debug output shows only the
75/// length of data, which may carry secrets, like typed passwords or
76/// forwarded traffic.
77#[derive(PartialEq)]
78pub(crate) enum Frame {
79    /// The client opens a stream.
80    SynStream {
81        stream_id: u32,
82        fin: bool,
83        headers: Headers,
84    },
85
86    /// The server accepts a stream.
87    SynReply { stream_id: u32, headers: Headers },
88
89    /// Either side aborts a stream.
90    RstStream { stream_id: u32, status: u32 },
91
92    /// A ping, which the receiver echoes if the other side started it.
93    Ping { id: u32 },
94
95    /// Either side ends the connection.
96    GoAway { last_stream_id: u32, status: u32 },
97
98    /// Additional headers, which only matter for their FIN flag.
99    Headers { stream_id: u32, fin: bool },
100
101    /// Data of a stream, an empty one with FIN closes its sending side.
102    Data {
103        stream_id: u32,
104        fin: bool,
105        data: Bytes,
106    },
107
108    /// A frame which needs no handling: SETTINGS, WINDOW_UPDATE or an unknown
109    /// control frame. spdystream does not do flow control, so window updates
110    /// are not needed either.
111    Ignored,
112}
113
114impl fmt::Debug for Frame {
115    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
116        match self {
117            Self::SynStream {
118                stream_id,
119                fin,
120                headers,
121            } => f
122                .debug_struct("SynStream")
123                .field("stream_id", stream_id)
124                .field("fin", fin)
125                .field("headers", headers)
126                .finish(),
127            Self::SynReply { stream_id, headers } => f
128                .debug_struct("SynReply")
129                .field("stream_id", stream_id)
130                .field("headers", headers)
131                .finish(),
132            Self::RstStream { stream_id, status } => f
133                .debug_struct("RstStream")
134                .field("stream_id", stream_id)
135                .field("status", status)
136                .finish(),
137            Self::Ping { id } => f.debug_struct("Ping").field("id", id).finish(),
138            Self::GoAway {
139                last_stream_id,
140                status,
141            } => f
142                .debug_struct("GoAway")
143                .field("last_stream_id", last_stream_id)
144                .field("status", status)
145                .finish(),
146            Self::Headers { stream_id, fin } => f
147                .debug_struct("Headers")
148                .field("stream_id", stream_id)
149                .field("fin", fin)
150                .finish(),
151            Self::Data {
152                stream_id,
153                fin,
154                data,
155            } => f
156                .debug_struct("Data")
157                .field("stream_id", stream_id)
158                .field("fin", fin)
159                .field("len", &data.len())
160                .finish(),
161            Self::Ignored => f.write_str("Ignored"),
162        }
163    }
164}
165
166/// The headers of a stream. SPDY requires lowercase names, multiple values
167/// of a name are joined with NUL bytes.
168#[derive(Debug, Default, PartialEq)]
169pub(crate) struct Headers(Vec<(String, String)>);
170
171impl Headers {
172    /// The first value of a header, the name is matched case insensitively.
173    pub(crate) fn get(&self, name: &str) -> Option<&str> {
174        self.0
175            .iter()
176            .find(|(n, _)| n.eq_ignore_ascii_case(name))
177            .and_then(|(_, v)| v.split('\0').next())
178    }
179}
180
181impl<const N: usize> From<[(&str, &str); N]> for Headers {
182    fn from(headers: [(&str, &str); N]) -> Self {
183        Self(
184            headers
185                .into_iter()
186                .map(|(n, v)| (n.to_ascii_lowercase(), v.to_owned()))
187                .collect(),
188        )
189    }
190}
191
192/// Reads the frames of one direction of a connection.
193pub(super) struct Decoder {
194    decompress: Decompress,
195
196    /// The data frame which is handed out in parts: its stream, the bytes
197    /// still to read and its FIN flag.
198    partial_data: Option<(u32, usize, bool)>,
199}
200
201impl Default for Decoder {
202    fn default() -> Self {
203        Self {
204            decompress: Decompress::new(true),
205            partial_data: None,
206        }
207    }
208}
209
210impl Decoder {
211    /// Read the next frame, `None` if the connection ended in between frames.
212    /// A data frame larger than [`DATA_CHUNK_LEN`] comes as several ones of
213    /// the same stream, only the last one with its FIN flag.
214    pub(super) async fn read<R>(&mut self, reader: &mut R) -> Result<Option<Frame>>
215    where
216        R: AsyncRead + Unpin,
217    {
218        if let Some((stream_id, remaining, fin)) = self.partial_data {
219            return self
220                .read_data(reader, stream_id, remaining, fin)
221                .await
222                .map(Some);
223        }
224
225        let mut header = [0; HEADER_LEN];
226        let mut filled = 0;
227        while filled < HEADER_LEN {
228            match reader
229                .read(&mut header[filled..])
230                .await
231                .context("read frame header")?
232            {
233                0 if filled == 0 => return Ok(None),
234                // An I/O error like the end within a payload, which tells
235                // a client which went away from a protocol violation.
236                0 => {
237                    return Err(io::Error::new(
238                        io::ErrorKind::UnexpectedEof,
239                        "connection ended within a frame header",
240                    )
241                    .into());
242                }
243                n => filled += n,
244            }
245        }
246
247        let control = header[0] & 0x80 != 0;
248        let flags = header[4];
249        let len =
250            usize::from(header[5]) << 16 | usize::from(header[6]) << 8 | usize::from(header[7]);
251        // Data frames do not get buffered as a whole, independent of the
252        // limit of the control frames.
253        if !control {
254            let stream_id = u32::from_be_bytes([header[0], header[1], header[2], header[3]]);
255            return self
256                .read_data(reader, stream_id, len, flags & FLAG_FIN != 0)
257                .await
258                .map(Some);
259        }
260        ensure!(
261            len <= MAX_CONTROL_FRAME_LEN,
262            "frame of {len} bytes exceeds {MAX_CONTROL_FRAME_LEN} bytes"
263        );
264
265        let mut payload = BytesMut::zeroed(len);
266        reader
267            .read_exact(&mut payload)
268            .await
269            .context("read frame payload")?;
270        self.decode(header, flags, payload.freeze()).map(Some)
271    }
272
273    /// Read a data frame, or the next part of a large one.
274    async fn read_data<R>(
275        &mut self,
276        reader: &mut R,
277        stream_id: u32,
278        remaining: usize,
279        fin: bool,
280    ) -> Result<Frame>
281    where
282        R: AsyncRead + Unpin,
283    {
284        let len = remaining.min(DATA_CHUNK_LEN);
285        let mut data = BytesMut::zeroed(len);
286        reader
287            .read_exact(&mut data)
288            .await
289            .context("read frame payload")?;
290        let remaining = remaining - len;
291        self.partial_data = (remaining > 0).then_some((stream_id, remaining, fin));
292        Ok(Frame::Data {
293            stream_id,
294            fin: fin && remaining == 0,
295            data: data.freeze(),
296        })
297    }
298
299    /// Decode a control frame from its header and complete payload.
300    fn decode(&mut self, header: [u8; HEADER_LEN], flags: u8, mut payload: Bytes) -> Result<Frame> {
301        let fin = flags & FLAG_FIN != 0;
302        let first = u32::from_be_bytes([header[0], header[1], header[2], header[3]]);
303        let version = (first >> 16) as u16 & 0x7fff;
304        ensure!(version == VERSION, "unsupported SPDY version {version}");
305
306        #[expect(
307            clippy::cast_possible_truncation,
308            reason = "the type is the lower half of the first word"
309        )]
310        let typ = first as u16;
311        let frame = match typ {
312            TYPE_SYN_STREAM => {
313                // The stream ID, the associated stream ID, the priority and
314                // the slot precede the header block.
315                ensure!(payload.len() >= 10, "SYN_STREAM too short");
316                let stream_id = payload.get_u32() & 0x7fff_ffff;
317                payload.advance(6);
318                let headers = self.header_block(&payload)?;
319                Frame::SynStream {
320                    stream_id,
321                    fin,
322                    headers,
323                }
324            }
325            TYPE_SYN_REPLY => {
326                ensure!(payload.len() >= 4, "SYN_REPLY too short");
327                let stream_id = payload.get_u32() & 0x7fff_ffff;
328                let headers = self.header_block(&payload)?;
329                Frame::SynReply { stream_id, headers }
330            }
331            TYPE_RST_STREAM => {
332                ensure!(payload.len() == 8, "RST_STREAM has the wrong length");
333                Frame::RstStream {
334                    stream_id: payload.get_u32() & 0x7fff_ffff,
335                    status: payload.get_u32(),
336                }
337            }
338            TYPE_PING => {
339                ensure!(payload.len() == 4, "PING has the wrong length");
340                Frame::Ping {
341                    id: payload.get_u32(),
342                }
343            }
344            TYPE_GOAWAY => {
345                ensure!(payload.len() == 8, "GOAWAY has the wrong length");
346                Frame::GoAway {
347                    last_stream_id: payload.get_u32() & 0x7fff_ffff,
348                    status: payload.get_u32(),
349                }
350            }
351            TYPE_HEADERS => {
352                ensure!(payload.len() >= 4, "HEADERS too short");
353                let stream_id = payload.get_u32() & 0x7fff_ffff;
354                self.header_block(&payload)?;
355                Frame::Headers { stream_id, fin }
356            }
357            _ => Frame::Ignored,
358        };
359        Ok(frame)
360    }
361
362    /// Decompress and parse a header block.
363    fn header_block(&mut self, mut input: &[u8]) -> Result<Headers> {
364        // One byte more than the limit, so that a block of exactly the limit
365        // still leaves room, which tells that it is complete.
366        let max_capacity = MAX_HEADER_BLOCK_LEN + 1;
367        let mut block = Vec::with_capacity((input.len() * 4).min(max_capacity));
368        loop {
369            if block.len() == block.capacity() {
370                ensure!(
371                    block.len() <= MAX_HEADER_BLOCK_LEN,
372                    "header block exceeds {MAX_HEADER_BLOCK_LEN} bytes"
373                );
374                // Exactly, a larger capacity would let the block grow past
375                // the limit.
376                block.reserve_exact((max_capacity - block.len()).min(4096));
377            }
378
379            let (total_in, total_out) = (self.decompress.total_in(), self.decompress.total_out());
380            let status =
381                match self
382                    .decompress
383                    .decompress_vec(input, &mut block, FlushDecompress::Sync)
384                {
385                    Ok(status) => status,
386                    Err(e) if e.needs_dictionary().is_some() => {
387                        self.decompress
388                            .set_dictionary(DICTIONARY)
389                            .context("set header dictionary")?;
390                        Status::Ok
391                    }
392                    Err(e) => return Err(e).context("decompress header block"),
393                };
394            #[expect(
395                clippy::cast_possible_truncation,
396                reason = "at most the length of the input"
397            )]
398            let read = (self.decompress.total_in() - total_in) as usize;
399            input = &input[read..];
400            // The allocator may still have given more room than asked for.
401            ensure!(
402                block.len() <= MAX_HEADER_BLOCK_LEN,
403                "header block exceeds {MAX_HEADER_BLOCK_LEN} bytes"
404            );
405
406            // The sender flushes after each block, so all of it is there once
407            // the input is consumed and the output has room left.
408            if status == Status::StreamEnd || input.is_empty() && block.len() < block.capacity() {
409                break;
410            }
411            ensure!(
412                block.len() == block.capacity()
413                    || self.decompress.total_in() > total_in
414                    || self.decompress.total_out() > total_out,
415                "header block does not decompress"
416            );
417        }
418
419        Self::parse_header_block(&block)
420    }
421
422    fn parse_header_block(mut block: &[u8]) -> Result<Headers> {
423        fn string(block: &mut &[u8]) -> Result<String> {
424            ensure!(block.len() >= 4, "header block truncated");
425            let len = block.get_u32() as usize;
426            ensure!(block.len() >= len, "header block truncated");
427            let s = String::from_utf8_lossy(&block[..len]).into_owned();
428            block.advance(len);
429            Ok(s)
430        }
431
432        ensure!(block.len() >= 4, "header block truncated");
433        let count = block.get_u32() as usize;
434        ensure!(
435            count <= MAX_HEADERS,
436            "header block has more than {MAX_HEADERS} headers"
437        );
438        // Each pair takes at least the two lengths.
439        ensure!(count <= block.len() / 8, "header block truncated");
440        let mut headers = Vec::with_capacity(count);
441        for _ in 0..count {
442            let name = string(&mut block)?;
443            let value = string(&mut block)?;
444            headers.push((name, value));
445        }
446        Ok(Headers(headers))
447    }
448}
449
450/// Writes the frames of one direction of a connection.
451#[derive(Default)]
452pub(super) struct Encoder {
453    /// Whether the zlib header went out with the first header block.
454    started: bool,
455}
456
457impl Encoder {
458    /// Append the encoded frame to the buffer.
459    pub(super) fn encode(&mut self, frame: &Frame, buf: &mut BytesMut) -> Result<()> {
460        match frame {
461            Frame::SynStream {
462                stream_id,
463                fin,
464                headers,
465            } => {
466                let block = self.header_block(headers);
467                Self::control(buf, TYPE_SYN_STREAM, *fin, 10 + block.len());
468                buf.put_u32(*stream_id);
469                // No associated stream, the lowest priority and no slot.
470                buf.put_u32(0);
471                buf.put_u16(0);
472                buf.put_slice(&block);
473            }
474            Frame::SynReply { stream_id, headers } => {
475                let block = self.header_block(headers);
476                Self::control(buf, TYPE_SYN_REPLY, false, 4 + block.len());
477                buf.put_u32(*stream_id);
478                buf.put_slice(&block);
479            }
480            Frame::RstStream { stream_id, status } => {
481                Self::control(buf, TYPE_RST_STREAM, false, 8);
482                buf.put_u32(*stream_id);
483                buf.put_u32(*status);
484            }
485            Frame::Ping { id } => {
486                Self::control(buf, TYPE_PING, false, 4);
487                buf.put_u32(*id);
488            }
489            Frame::GoAway {
490                last_stream_id,
491                status,
492            } => {
493                Self::control(buf, TYPE_GOAWAY, false, 8);
494                buf.put_u32(*last_stream_id);
495                buf.put_u32(*status);
496            }
497            Frame::Headers { stream_id, fin } => {
498                let block = self.header_block(&Headers::default());
499                Self::control(buf, TYPE_HEADERS, *fin, 4 + block.len());
500                buf.put_u32(*stream_id);
501                buf.put_slice(&block);
502            }
503            Frame::Data {
504                stream_id,
505                fin,
506                data,
507            } => {
508                ensure!(data.len() <= MAX_DATA_FRAME_LEN, "data frame too large");
509                buf.put_u32(*stream_id);
510                buf.put_u8(if *fin { FLAG_FIN } else { 0 });
511                Self::put_len(buf, data.len());
512                buf.put_slice(data);
513            }
514            Frame::Ignored => {}
515        }
516        Ok(())
517    }
518
519    fn control(buf: &mut BytesMut, typ: u16, fin: bool, len: usize) {
520        buf.put_u16(0x8000 | VERSION);
521        buf.put_u16(typ);
522        buf.put_u8(if fin { FLAG_FIN } else { 0 });
523        Self::put_len(buf, len);
524    }
525
526    #[expect(
527        clippy::cast_possible_truncation,
528        reason = "data frames are checked, control frames carry the few headers of the server"
529    )]
530    fn put_len(buf: &mut BytesMut, len: usize) {
531        buf.put_slice(&(len as u32).to_be_bytes()[1..]);
532    }
533
534    /// Serialize a header block into stored deflate blocks, which every
535    /// zlib decompressor reads like compressed ones. Compressing the blocks
536    /// of the server would gain nothing, they are empty.
537    fn header_block(&mut self, headers: &Headers) -> Vec<u8> {
538        self.deflate(&Self::serialize(headers))
539    }
540
541    /// The uncompressed header block.
542    #[expect(
543        clippy::cast_possible_truncation,
544        reason = "the few short headers of the server"
545    )]
546    fn serialize(headers: &Headers) -> Vec<u8> {
547        let mut raw = Vec::new();
548        raw.put_u32(headers.0.len() as u32);
549        for (name, value) in &headers.0 {
550            raw.put_u32(name.len() as u32);
551            raw.put_slice(name.as_bytes());
552            raw.put_u32(value.len() as u32);
553            raw.put_slice(value.as_bytes());
554        }
555        raw
556    }
557
558    /// Wrap a serialized header block into stored deflate blocks, the first
559    /// one of the connection with the zlib header.
560    fn deflate(&mut self, raw: &[u8]) -> Vec<u8> {
561        let mut block = Vec::with_capacity(raw.len() + 16);
562        if !self.started {
563            // Deflate with a 32 KiB window and the preset dictionary.
564            block.put_slice(&[0x78, 0x20]);
565            block.put_u32(DICTIONARY_ID);
566            self.started = true;
567        }
568        for chunk in raw.chunks(usize::from(u16::MAX)) {
569            Self::stored_block(&mut block, chunk);
570        }
571        // The empty block of a sync flush, Go hands out the data before it
572        // only once it read one. Without it, the decompressor reads beyond
573        // the frame and fails all later header blocks.
574        Self::stored_block(&mut block, &[]);
575        block
576    }
577
578    /// A stored deflate block which is not the last one.
579    #[expect(
580        clippy::cast_possible_truncation,
581        reason = "the callers pass at most u16::MAX bytes"
582    )]
583    fn stored_block(block: &mut Vec<u8>, data: &[u8]) {
584        let len = data.len() as u16;
585        block.put_u8(0);
586        block.put_u16_le(len);
587        block.put_u16_le(!len);
588        block.put_slice(data);
589    }
590}
591
592/// The entry points of the fuzz targets.
593#[cfg(any(test, feature = "fuzzing"))]
594pub(super) mod fuzzing;
595
596/// The Adler-32 checksum of the dictionary, which the zlib header names.
597const DICTIONARY_ID: u32 = 0xe3c6a7c2;
598
599/// The preset dictionary for the header compression, from the specification.
600const DICTIONARY: &[u8; 1423] = b"\
601    \0\0\0\x07options\0\0\0\x04head\0\0\0\x04post\0\0\0\x03put\0\0\0\x06delete\
602    \0\0\0\x05trace\0\0\0\x06accept\0\0\0\x0eaccept-charset\
603    \0\0\0\x0faccept-encoding\0\0\0\x0faccept-language\0\0\0\x0daccept-ranges\
604    \0\0\0\x03age\0\0\0\x05allow\0\0\0\x0dauthorization\0\0\0\x0dcache-control\
605    \0\0\0\x0aconnection\0\0\0\x0ccontent-base\0\0\0\x10content-encoding\
606    \0\0\0\x10content-language\0\0\0\x0econtent-length\0\0\0\x10content-location\
607    \0\0\0\x0bcontent-md5\0\0\0\x0dcontent-range\0\0\0\x0ccontent-type\
608    \0\0\0\x04date\0\0\0\x04etag\0\0\0\x06expect\0\0\0\x07expires\0\0\0\x04from\
609    \0\0\0\x04host\0\0\0\x08if-match\0\0\0\x11if-modified-since\
610    \0\0\0\x0dif-none-match\0\0\0\x08if-range\0\0\0\x13if-unmodified-since\
611    \0\0\0\x0dlast-modified\0\0\0\x08location\0\0\0\x0cmax-forwards\
612    \0\0\0\x06pragma\0\0\0\x12proxy-authenticate\0\0\0\x13proxy-authorization\
613    \0\0\0\x05range\0\0\0\x07referer\0\0\0\x0bretry-after\0\0\0\x06server\
614    \0\0\0\x02te\0\0\0\x07trailer\0\0\0\x11transfer-encoding\0\0\0\x07upgrade\
615    \0\0\0\x0auser-agent\0\0\0\x04vary\0\0\0\x03via\0\0\0\x07warning\
616    \0\0\0\x10www-authenticate\0\0\0\x06method\0\0\0\x03get\0\0\0\x06status\
617    \0\0\0\x06200 OK\0\0\0\x07version\0\0\0\x08HTTP/1.1\0\0\0\x03url\
618    \0\0\0\x06public\0\0\0\x0aset-cookie\0\0\0\x0akeep-alive\0\0\0\x06origin100\
619    101201202205206300302303304305306307402405406407408409410411412413414415416\
620    417502504505203 Non-Authoritative Information204 No Content301 Moved \
621    Permanently400 Bad Request401 Unauthorized403 Forbidden404 Not Found500 \
622    Internal Server Error501 Not Implemented503 Service UnavailableJan Feb Mar \
623    Apr May Jun Jul Aug Sept Oct Nov Dec 00:00:00 Mon, Tue, Wed, Thu, Fri, Sat, \
624    Sun, GMTchunked,text/html,image/png,image/jpg,image/gif,application/xml,\
625    application/xhtml+xml,text/plain,text/javascript,publicprivatemax-age=gzip,\
626    deflate,sdchcharset=utf-8charset=iso-8859-1,utf-,*,enq=0.";
627
628#[cfg(test)]
629mod tests;
conmonrs/streaming_server/spdy/frame.rs

conmonrs/streaming_server/spdy/
frame.rs
