runner_manager_agent/docker_cleanup/
http.rs

1//! Small bounded HTTP/1 framing for local Docker JSON requests. Streaming
2//! endpoints are relayed by the proxy without buffering their payloads.
3
4use std::io::{self, Read};
5
6use super::invalid;
7
8pub const MAX_JSON: usize = 8 * 1024 * 1024;
9const MAX_HEADERS: usize = 32 * 1024;
10
11#[derive(Debug)]
12pub struct Head {
13    pub first: String,
14    pub headers: Vec<(String, String)>,
15}
16
17impl Head {
18    pub fn read(reader: &mut impl Read) -> io::Result<Self> {
19        let mut bytes = Vec::new();
20        loop {
21            let mut byte = [0];
22            reader.read_exact(&mut byte)?;
23            bytes.push(byte[0]);
24            if bytes.len() > MAX_HEADERS {
25                return Err(invalid("Docker HTTP headers too large"));
26            }
27            if bytes.ends_with(b"\r\n\r\n") {
28                break;
29            }
30        }
31        let text = std::str::from_utf8(&bytes).map_err(invalid)?;
32        let mut lines = text[..text.len() - 4].split("\r\n");
33        let first = lines
34            .next()
35            .ok_or_else(|| invalid("missing HTTP start line"))?
36            .to_owned();
37        let mut headers = Vec::new();
38        for line in lines {
39            let (name, value) = line
40                .split_once(':')
41                .ok_or_else(|| invalid("invalid HTTP header"))?;
42            if name.is_empty()
43                || name
44                    .bytes()
45                    .any(|byte| !byte.is_ascii_alphanumeric() && byte != b'-')
46            {
47                return Err(invalid("invalid HTTP header name"));
48            }
49            headers.push((name.into(), value.trim().into()));
50        }
51        Ok(Self { first, headers })
52    }
53
54    pub fn get(&self, name: &str) -> Option<&str> {
55        self.headers
56            .iter()
57            .find(|(key, _)| key.eq_ignore_ascii_case(name))
58            .map(|(_, value)| value.as_str())
59    }
60
61    pub fn encode_for_relay(&self) -> Vec<u8> {
62        let upgrade = self.get("Upgrade").is_some();
63        let mut text = format!("{}\r\n", self.first);
64        for (name, value) in &self.headers {
65            if !upgrade && name.eq_ignore_ascii_case("Connection") {
66                continue;
67            }
68            text.push_str(&format!("{name}: {value}\r\n"));
69        }
70        if !upgrade {
71            text.push_str("Connection: close\r\n");
72        }
73        text.push_str("\r\n");
74        text.into_bytes()
75    }
76
77    pub fn body(&self, reader: &mut impl Read) -> io::Result<Vec<u8>> {
78        let lengths: Vec<_> = self
79            .headers
80            .iter()
81            .filter(|(name, _)| name.eq_ignore_ascii_case("Content-Length"))
82            .collect();
83        if lengths.len() > 1 || (!lengths.is_empty() && self.get("Transfer-Encoding").is_some()) {
84            return Err(invalid("ambiguous HTTP framing"));
85        }
86        if let Some(encoding) = self.get("Transfer-Encoding") {
87            if !encoding.eq_ignore_ascii_case("chunked") {
88                return Err(invalid("unsupported HTTP encoding"));
89            }
90            let mut body = Vec::new();
91            loop {
92                let line = read_line(reader, 128)?;
93                let size = usize::from_str_radix(line.split(';').next().unwrap_or_default(), 16)
94                    .map_err(invalid)?;
95                if size > MAX_JSON.saturating_sub(body.len()) {
96                    return Err(invalid("Docker JSON too large"));
97                }
98                if size == 0 {
99                    let mut total = 0;
100                    loop {
101                        let trailer = read_line(reader, MAX_HEADERS)?;
102                        total += trailer.len() + 2;
103                        if total > MAX_HEADERS {
104                            return Err(invalid("HTTP trailers too large"));
105                        }
106                        if trailer.is_empty() {
107                            return Ok(body);
108                        }
109                    }
110                }
111                let start = body.len();
112                body.resize(start + size, 0);
113                reader.read_exact(&mut body[start..])?;
114                let mut ending = [0; 2];
115                reader.read_exact(&mut ending)?;
116                if ending != *b"\r\n" {
117                    return Err(invalid("invalid HTTP chunk ending"));
118                }
119            }
120        }
121        let size: usize = self
122            .get("Content-Length")
123            .unwrap_or("0")
124            .parse()
125            .map_err(invalid)?;
126        if size > MAX_JSON {
127            return Err(invalid("Docker JSON too large"));
128        }
129        let mut body = vec![0; size];
130        reader.read_exact(&mut body)?;
131        Ok(body)
132    }
133}
134
135fn read_line(reader: &mut impl Read, limit: usize) -> io::Result<String> {
136    let mut bytes = Vec::new();
137    loop {
138        let mut byte = [0];
139        reader.read_exact(&mut byte)?;
140        bytes.push(byte[0]);
141        if bytes.len() > limit {
142            return Err(invalid("HTTP line too large"));
143        }
144        if bytes.ends_with(b"\r\n") {
145            bytes.truncate(bytes.len() - 2);
146            return String::from_utf8(bytes).map_err(invalid);
147        }
148    }
149}
150
151#[cfg(test)]
152mod tests {
153    use super::*;
154
155    #[test]
156    fn reads_chunked_json_and_consumes_bounded_trailers() {
157        let mut wire = &b"POST /containers/create HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n4\r\n{\"a\"\r\n3\r\n:1}\r\n0\r\nX-Fixture: trailer\r\n\r\nnext"[..];
158        let head = Head::read(&mut wire).unwrap();
159        assert_eq!(head.body(&mut wire).unwrap(), b"{\"a\":1}");
160        assert_eq!(wire, b"next");
161    }
162
163    #[test]
164    fn rejects_ambiguous_framing_and_oversized_payloads() {
165        for headers in [
166            "Content-Length: 1\r\nContent-Length: 1",
167            "Content-Length: 1\r\nTransfer-Encoding: chunked",
168            "Content-Length: 9000000",
169            "Transfer-Encoding: gzip",
170        ] {
171            let bytes = format!("POST /containers/create HTTP/1.1\r\n{headers}\r\n\r\nx");
172            let mut wire = bytes.as_bytes();
173            let head = Head::read(&mut wire).unwrap();
174            assert!(head.body(&mut wire).is_err(), "{headers}");
175        }
176    }
177
178    #[test]
179    fn ordinary_relays_close_the_connection_but_attach_upgrades_remain_bidirectional() {
180        let mut ordinary =
181            &b"GET /info HTTP/1.1\r\nConnection: keep-alive\r\nHost: docker\r\n\r\n"[..];
182        let head = Head::read(&mut ordinary).unwrap();
183        let relayed = String::from_utf8(head.encode_for_relay()).unwrap();
184        assert!(relayed.contains("Connection: close\r\n"));
185        assert!(!relayed.contains("keep-alive"));
186        let wire = b"POST /containers/fixture/attach HTTP/1.1\r\nConnection: Upgrade\r\nUpgrade: tcp\r\n\r\n";
187        let mut input = &wire[..];
188        let head = Head::read(&mut input).unwrap();
189        assert_eq!(head.encode_for_relay(), wire);
190    }
191}
runner_manager_agent/docker_cleanup/http.rs

runner_manager_agent/docker_cleanup/
http.rs
