secure_exec_sidecar_core/
vm_fetch.rs

1use crate::SidecarCoreError;
2use base64::Engine as _;
3use serde_json::{json, Value};
4use std::collections::BTreeMap;
5
6// Keep raw loopback fetch buffers inside the default sidecar wire frame budget.
7pub const VM_FETCH_BUFFER_LIMIT_BYTES: usize = 1024 * 1024;
8
9pub fn parse_kernel_http_fetch_response(
10    buffer: &[u8],
11    peer_closed: bool,
12    url: &str,
13) -> Result<Option<String>, SidecarCoreError> {
14    let Some(header_end) = find_http_header_end(buffer) else {
15        return Ok(None);
16    };
17    let header_bytes = &buffer[..header_end];
18    let head = String::from_utf8_lossy(header_bytes);
19    let mut lines = head.split("\r\n");
20    let status_line = lines.next().unwrap_or_default();
21    let mut status_parts = status_line.splitn(3, ' ');
22    let version = status_parts.next().unwrap_or_default();
23    if !version.starts_with("HTTP/") {
24        return Err(SidecarCoreError::new(format!(
25            "invalid vm.fetch HTTP response status line: {status_line}"
26        )));
27    }
28    let status = status_parts
29        .next()
30        .ok_or_else(|| {
31            SidecarCoreError::new(format!(
32                "invalid vm.fetch HTTP response status line: {status_line}"
33            ))
34        })?
35        .parse::<u16>()
36        .map_err(|error| {
37            SidecarCoreError::new(format!(
38                "invalid vm.fetch HTTP response status code in {status_line:?}: {error}"
39            ))
40        })?;
41    let status_text = status_parts.next().unwrap_or_default();
42    let mut headers = Vec::new();
43    let mut raw_headers = Vec::new();
44    let mut content_length = None;
45    let mut transfer_encoding_values = Vec::new();
46    for line in lines {
47        if line.is_empty() {
48            continue;
49        }
50        let Some((name, value)) = line.split_once(':') else {
51            return Err(SidecarCoreError::new(format!(
52                "invalid vm.fetch HTTP response header line: {line}"
53            )));
54        };
55        let value = value.trim().to_owned();
56        let normalized = name.to_ascii_lowercase();
57        if normalized == "content-length" {
58            content_length = Some(value.parse::<usize>().map_err(|error| {
59                SidecarCoreError::new(format!(
60                    "invalid vm.fetch Content-Length header {value:?}: {error}"
61                ))
62            })?);
63        } else if normalized == "transfer-encoding" {
64            transfer_encoding_values.push(value.clone());
65        }
66        headers.push(json!([normalized, value.clone()]));
67        raw_headers.push(Value::String(name.to_owned()));
68        raw_headers.push(Value::String(value));
69    }
70
71    let body_start = header_end + 4;
72    let transfer_encoding = transfer_encoding_tokens(&transfer_encoding_values);
73    let is_chunked = transfer_encoding.iter().any(|token| token == "chunked");
74    let body = if is_chunked {
75        if content_length.is_some() {
76            return Err(SidecarCoreError::new(
77                "vm.fetch HTTP response cannot include both Transfer-Encoding: chunked and Content-Length",
78            ));
79        }
80        if transfer_encoding.len() != 1 {
81            return Err(SidecarCoreError::new(format!(
82                "unsupported vm.fetch Transfer-Encoding: {}",
83                transfer_encoding.join(", ")
84            )));
85        }
86        let Some(decoded) = decode_kernel_http_chunked_body(&buffer[body_start..])? else {
87            return Ok(None);
88        };
89        decoded
90    } else if !transfer_encoding.is_empty() {
91        return Err(SidecarCoreError::new(format!(
92            "unsupported vm.fetch Transfer-Encoding: {}",
93            transfer_encoding.join(", ")
94        )));
95    } else if let Some(content_length) = content_length {
96        let body_end = body_start.saturating_add(content_length);
97        if buffer.len() < body_end {
98            return Ok(None);
99        }
100        buffer[body_start..body_end].to_vec()
101    } else if peer_closed {
102        buffer[body_start..].to_vec()
103    } else {
104        return Ok(None);
105    };
106
107    serde_json::to_string(&json!({
108        "status": status,
109        "statusText": status_text,
110        "headers": headers,
111        "rawHeaders": raw_headers,
112        "body": base64::engine::general_purpose::STANDARD.encode(&body),
113        "bodyEncoding": "base64",
114        "url": url,
115    }))
116    .map(Some)
117    .map_err(|error| SidecarCoreError::new(format!("ERR_AGENTOS_NODE_SYNC_RPC: {error}")))
118}
119
120pub fn serialize_kernel_http_fetch_request(
121    port: u16,
122    path: &str,
123    method: &str,
124    headers_json: &str,
125    body: Option<&str>,
126) -> Result<Vec<u8>, SidecarCoreError> {
127    let headers = parse_vm_fetch_headers(headers_json)?;
128    let method = if method.is_empty() { "GET" } else { method };
129    let target_path = if path.starts_with('/') {
130        path.to_owned()
131    } else {
132        format!("/{path}")
133    };
134    let mut lines = vec![format!("{method} {target_path} HTTP/1.1")];
135    let mut has_host = false;
136    let mut has_connection = false;
137    let mut has_content_length = false;
138
139    for (name, values) in &headers {
140        match name.as_str() {
141            "host" => has_host = true,
142            "connection" => has_connection = true,
143            "content-length" => has_content_length = true,
144            _ => {}
145        }
146        lines.push(format!("{name}: {}", values.join(", ")));
147    }
148    if !has_host {
149        lines.push(format!("Host: 127.0.0.1:{port}"));
150    }
151    if !has_connection {
152        lines.push(String::from("Connection: close"));
153    }
154    let body = body.unwrap_or("").as_bytes();
155    if !has_content_length && !body.is_empty() {
156        lines.push(format!("Content-Length: {}", body.len()));
157    }
158    lines.push(String::new());
159    lines.push(String::new());
160
161    let mut request = lines.join("\r\n").into_bytes();
162    request.extend_from_slice(body);
163    Ok(request)
164}
165
166pub fn ensure_vm_fetch_response_within_limit(
167    response_json: &str,
168    operation: &str,
169    limit: usize,
170) -> Result<(), SidecarCoreError> {
171    let size = response_json.len();
172    if size > limit {
173        return Err(SidecarCoreError::new(format!(
174            "{operation} payload is {size} bytes, limit is {limit}"
175        )));
176    }
177    Ok(())
178}
179
180pub fn ensure_vm_fetch_raw_response_buffer_within_limit(
181    size: usize,
182    operation: &str,
183) -> Result<(), SidecarCoreError> {
184    if size > VM_FETCH_BUFFER_LIMIT_BYTES {
185        return Err(SidecarCoreError::new(format!(
186            "{operation} raw response buffer is {size} bytes, limit is {VM_FETCH_BUFFER_LIMIT_BYTES}"
187        )));
188    }
189    Ok(())
190}
191
192fn parse_vm_fetch_headers(
193    headers_json: &str,
194) -> Result<BTreeMap<String, Vec<String>>, SidecarCoreError> {
195    let headers: BTreeMap<String, Value> = serde_json::from_str(headers_json).map_err(|error| {
196        SidecarCoreError::new(format!("vm.fetch headers_json must be valid JSON: {error}"))
197    })?;
198    let mut normalized = BTreeMap::<String, Vec<String>>::new();
199    for (raw_name, value) in headers {
200        let values = match value {
201            Value::String(text) => vec![text],
202            Value::Array(values) => values
203                .into_iter()
204                .map(|entry| {
205                    entry.as_str().map(str::to_owned).ok_or_else(|| {
206                        SidecarCoreError::new(format!(
207                            "vm.fetch header {raw_name} must contain only strings"
208                        ))
209                    })
210                })
211                .collect::<Result<Vec<_>, _>>()?,
212            other => {
213                return Err(SidecarCoreError::new(format!(
214                    "vm.fetch header {raw_name} must be a string or string array, received {other}"
215                )));
216            }
217        };
218        normalized
219            .entry(raw_name.to_ascii_lowercase())
220            .or_default()
221            .extend(values);
222    }
223    Ok(normalized)
224}
225
226fn find_http_header_end(buffer: &[u8]) -> Option<usize> {
227    buffer.windows(4).position(|window| window == b"\r\n\r\n")
228}
229
230fn find_crlf(buffer: &[u8], start: usize) -> Option<usize> {
231    buffer
232        .get(start..)?
233        .windows(2)
234        .position(|window| window == b"\r\n")
235        .map(|offset| start + offset)
236}
237
238fn transfer_encoding_tokens(values: &[String]) -> Vec<String> {
239    values
240        .iter()
241        .flat_map(|value| value.split(','))
242        .map(|token| token.trim().to_ascii_lowercase())
243        .filter(|token| !token.is_empty())
244        .collect()
245}
246
247fn decode_kernel_http_chunked_body(buffer: &[u8]) -> Result<Option<Vec<u8>>, SidecarCoreError> {
248    let mut offset = 0;
249    let mut body = Vec::new();
250    loop {
251        let Some(line_end) = find_crlf(buffer, offset) else {
252            return Ok(None);
253        };
254        let size_line = std::str::from_utf8(&buffer[offset..line_end]).map_err(|error| {
255            SidecarCoreError::new(format!(
256                "invalid vm.fetch chunk size line encoding: {error}"
257            ))
258        })?;
259        let size_part = size_line.split(';').next().unwrap_or_default();
260        if size_part.is_empty() || !size_part.bytes().all(|byte| byte.is_ascii_hexdigit()) {
261            return Err(SidecarCoreError::new(format!(
262                "invalid vm.fetch chunk size line: {size_line:?}"
263            )));
264        }
265        let chunk_size = usize::from_str_radix(size_part, 16).map_err(|error| {
266            SidecarCoreError::new(format!(
267                "invalid vm.fetch chunk size {size_part:?}: {error}"
268            ))
269        })?;
270        let chunk_start = line_end + 2;
271        let chunk_end = chunk_start
272            .checked_add(chunk_size)
273            .ok_or_else(|| SidecarCoreError::new("vm.fetch chunk size overflow"))?;
274        if chunk_size > 0 {
275            let chunk_terminator_end = chunk_end
276                .checked_add(2)
277                .ok_or_else(|| SidecarCoreError::new("vm.fetch chunk terminator overflow"))?;
278            if chunk_terminator_end > buffer.len() {
279                return Ok(None);
280            }
281            if buffer.get(chunk_end..chunk_terminator_end) != Some(b"\r\n") {
282                return Err(SidecarCoreError::new("invalid vm.fetch chunk terminator"));
283            }
284            body.extend_from_slice(&buffer[chunk_start..chunk_end]);
285            offset = chunk_terminator_end;
286            continue;
287        }
288
289        if buffer.get(chunk_start..chunk_start + 2) == Some(b"\r\n") {
290            return Ok(Some(body));
291        }
292        let Some(trailer_end) = find_http_header_end(&buffer[chunk_start..]) else {
293            return Ok(None);
294        };
295        let trailer_bytes = &buffer[chunk_start..chunk_start + trailer_end];
296        let trailers = String::from_utf8_lossy(trailer_bytes);
297        for line in trailers.split("\r\n") {
298            if line.is_empty() {
299                continue;
300            }
301            if line.starts_with(' ') || line.starts_with('\t') || !line.contains(':') {
302                return Err(SidecarCoreError::new(format!(
303                    "invalid vm.fetch chunk trailer line: {line}"
304                )));
305            }
306        }
307        return Ok(Some(body));
308    }
309}
310
311#[cfg(test)]
312mod tests {
313    use super::*;
314    use serde_json::Value;
315
316    #[test]
317    fn parses_content_length_response() {
318        let response = parse_kernel_http_fetch_response(
319            b"HTTP/1.1 201 Created\r\nContent-Length: 5\r\nX-Test: ok\r\n\r\nhello",
320            false,
321            "http://127.0.0.1:8080/hello",
322        )
323        .expect("parse response")
324        .expect("complete response");
325        let value: Value = serde_json::from_str(&response).expect("response json");
326
327        assert_eq!(value["status"], 201);
328        assert_eq!(value["statusText"], "Created");
329        assert_eq!(value["body"], "aGVsbG8=");
330        assert_eq!(value["url"], "http://127.0.0.1:8080/hello");
331    }
332
333    #[test]
334    fn serializes_loopback_fetch_request() {
335        let request = serialize_kernel_http_fetch_request(
336            3000,
337            "submit",
338            "POST",
339            r#"{"x-test":["a","b"]}"#,
340            Some("hello"),
341        )
342        .expect("serialize request");
343
344        assert_eq!(
345            String::from_utf8(request).expect("utf8 request"),
346            "POST /submit HTTP/1.1\r\nx-test: a, b\r\nHost: 127.0.0.1:3000\r\nConnection: close\r\nContent-Length: 5\r\n\r\nhello"
347        );
348    }
349
350    #[test]
351    fn decodes_chunked_response_body() {
352        let response = parse_kernel_http_fetch_response(
353            b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n",
354            false,
355            "http://127.0.0.1:8080/chunked",
356        )
357        .expect("parse response")
358        .expect("complete response");
359        let value: Value = serde_json::from_str(&response).expect("response json");
360
361        assert_eq!(value["body"], "aGVsbG8=");
362    }
363
364    #[test]
365    fn waits_for_incomplete_body() {
366        let response = parse_kernel_http_fetch_response(
367            b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhe",
368            false,
369            "http://127.0.0.1:8080/partial",
370        )
371        .expect("parse response");
372
373        assert!(response.is_none());
374    }
375
376    #[test]
377    fn rejects_invalid_chunked_content_length_combination() {
378        let error = parse_kernel_http_fetch_response(
379            b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n",
380            false,
381            "http://127.0.0.1:8080/bad",
382        )
383        .expect_err("response should be invalid");
384
385        assert!(error
386            .to_string()
387            .contains("cannot include both Transfer-Encoding: chunked and Content-Length"));
388    }
389}
secure_exec_sidecar_core/vm_fetch.rs

secure_exec_sidecar_core/
vm_fetch.rs
