netget/server/icap/
wire.rs

1//! RFC 3507 framing shared by both roles: ICAP heads, `Encapsulated` offsets, embedded HTTP
2//! heads and chunked bodies with preview.
3use anyhow::{bail, ensure, Context, Result};
4use serde::{Deserialize, Serialize};
5use serde_json::{json, Value};
6use std::time::Duration;
7use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt};
8
9pub const MAX_HEAD_BYTES: usize = 64 * 1024;
10pub const MAX_HEADERS: usize = 100;
11pub const MAX_BODY_BYTES: usize = 1024 * 1024;
12pub const MAX_CHUNKS: usize = 4096;
13pub const IO_TIMEOUT: Duration = Duration::from_secs(30);
14
15/// An HTTP head (request or response) carried inside an ICAP message.
16#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
17pub struct HttpHead {
18    /// "GET /path HTTP/1.1" or "HTTP/1.1 200 OK", split into three parts.
19    pub start: [String; 3],
20    pub headers: Vec<(String, String)>,
21}
22
23fn token(name: &str) -> bool {
24    !name.is_empty()
25        && name
26            .bytes()
27            .all(|b| b.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|~".contains(&b))
28}
29
30fn header_value_ok(v: &str) -> bool {
31    !v.bytes()
32        .any(|b| b == b'\r' || b == b'\n' || b == 0 || (b < 0x20 && b != b'\t') || b == 0x7f)
33}
34
35/// Parse a CRLF-separated head (start line + headers, without the blank line).
36pub fn parse_head(text: &str) -> Result<HttpHead> {
37    let mut lines = text.split("\r\n");
38    let start = lines.next().context("empty head")?;
39    let parts: Vec<&str> = start.splitn(3, ' ').collect();
40    ensure!(
41        parts.len() == 3 && parts.iter().take(2).all(|p| !p.is_empty()),
42        "malformed start line '{start}'"
43    );
44    let mut headers = Vec::new();
45    for line in lines.filter(|l| !l.is_empty()) {
46        ensure!(headers.len() < MAX_HEADERS, "too many headers");
47        let (name, value) = line
48            .split_once(':')
49            .with_context(|| format!("header without ':' — '{line}'"))?;
50        ensure!(token(name), "invalid header name '{name}'");
51        let value = value.trim();
52        ensure!(
53            header_value_ok(value),
54            "header {name} has a control character"
55        );
56        headers.push((name.to_owned(), value.to_owned()));
57    }
58    Ok(HttpHead {
59        start: [parts[0].into(), parts[1].into(), parts[2].into()],
60        headers,
61    })
62}
63
64pub fn render_head(head: &HttpHead) -> Result<String> {
65    for p in &head.start {
66        ensure!(
67            header_value_ok(p) && !p.contains('\t'),
68            "start line part has a control character"
69        );
70    }
71    ensure!(
72        head.start
73            .iter()
74            .take(2)
75            .all(|p| !p.is_empty() && !p.contains(' ')),
76        "start line parts 1 and 2 must be single tokens"
77    );
78    ensure!(head.headers.len() <= MAX_HEADERS, "too many headers");
79    let mut out = format!("{} {} {}\r\n", head.start[0], head.start[1], head.start[2]);
80    for (k, v) in &head.headers {
81        ensure!(token(k), "invalid header name '{k}'");
82        ensure!(header_value_ok(v), "header {k} has a control character");
83        out.push_str(&format!("{k}: {v}\r\n"));
84    }
85    out.push_str("\r\n");
86    ensure!(
87        out.len() <= MAX_HEAD_BYTES,
88        "head exceeds {MAX_HEAD_BYTES} bytes"
89    );
90    Ok(out)
91}
92
93pub fn header<'a>(headers: &'a [(String, String)], name: &str) -> Option<&'a str> {
94    headers
95        .iter()
96        .find(|(k, _)| k.eq_ignore_ascii_case(name))
97        .map(|(_, v)| v.as_str())
98}
99
100/// `Encapsulated: req-hdr=0, res-hdr=137, res-body=296` in order.
101pub fn parse_encapsulated(value: &str) -> Result<Vec<(String, usize)>> {
102    let mut out: Vec<(String, usize)> = Vec::new();
103    for part in value.split(',') {
104        let (k, v) = part
105            .trim()
106            .split_once('=')
107            .context("Encapsulated entry without '='")?;
108        ensure!(
109            matches!(
110                k,
111                "req-hdr" | "res-hdr" | "req-body" | "res-body" | "opt-body" | "null-body"
112            ),
113            "unknown Encapsulated entity '{k}'"
114        );
115        let offset: usize = v
116            .trim()
117            .parse()
118            .context("Encapsulated offset is not a number")?;
119        ensure!(
120            out.last().is_none_or(|(_, prev)| offset >= *prev),
121            "Encapsulated offsets must not decrease"
122        );
123        ensure!(
124            offset <= MAX_HEAD_BYTES,
125            "Encapsulated offset beyond the head bound"
126        );
127        out.push((k.to_owned(), offset));
128        ensure!(out.len() <= 4, "too many Encapsulated entities");
129    }
130    ensure!(!out.is_empty(), "empty Encapsulated header");
131    let last = &out.last().expect("non-empty").0;
132    ensure!(
133        last.ends_with("-body"),
134        "the last Encapsulated entity must be a body or null-body"
135    );
136    ensure!(
137        out[..out.len() - 1]
138            .iter()
139            .all(|(k, _)| k.ends_with("-hdr")),
140        "only the last Encapsulated entity may be a body"
141    );
142    Ok(out)
143}
144
145/// Read one CRLF-terminated head (ending in an empty line), bounded.
146pub async fn read_head<R: AsyncBufRead + Unpin>(
147    reader: &mut R,
148    limit: usize,
149) -> Result<Option<String>> {
150    let mut head = Vec::new();
151    loop {
152        let mut line = Vec::new();
153        let n = (&mut *reader)
154            .take((limit + 2 - head.len().min(limit)) as u64)
155            .read_until(b'\n', &mut line)
156            .await?;
157        if n == 0 {
158            ensure!(head.is_empty(), "peer closed inside a head");
159            return Ok(None);
160        }
161        ensure!(line.ends_with(b"\r\n"), "head lines must end in CRLF");
162        if line == b"\r\n" {
163            if head.is_empty() {
164                continue; // tolerate a stray CRLF between messages
165            }
166            break;
167        }
168        head.extend_from_slice(&line);
169        ensure!(head.len() <= limit, "head exceeds {limit} bytes");
170    }
171    head.truncate(head.len() - 2);
172    String::from_utf8(head)
173        .map(Some)
174        .context("head is not UTF-8")
175}
176
177/// Read exactly `len` bytes of an embedded HTTP head.
178pub async fn read_exact_head<R: AsyncBufRead + Unpin>(
179    reader: &mut R,
180    len: usize,
181) -> Result<HttpHead> {
182    ensure!(
183        len <= MAX_HEAD_BYTES && len >= 4,
184        "embedded head length out of bounds"
185    );
186    let mut bytes = vec![0u8; len];
187    reader.read_exact(&mut bytes).await?;
188    ensure!(
189        bytes.ends_with(b"\r\n\r\n"),
190        "embedded HTTP head must end with an empty line"
191    );
192    let text = std::str::from_utf8(&bytes[..len - 4]).context("embedded head is not UTF-8")?;
193    parse_head(text)
194}
195
196/// Outcome of reading chunks: the data and whether the sender marked end of preview (`ieof`).
197pub struct Chunks {
198    pub data: Vec<u8>,
199    pub ieof: bool,
200}
201
202/// Read chunks up to the zero chunk. `total` is what earlier reads already accepted.
203pub async fn read_chunks<R: AsyncBufRead + Unpin>(reader: &mut R, total: usize) -> Result<Chunks> {
204    let mut data = Vec::new();
205    for _ in 0..MAX_CHUNKS {
206        let mut line = Vec::new();
207        (&mut *reader)
208            .take(258)
209            .read_until(b'\n', &mut line)
210            .await?;
211        ensure!(line.ends_with(b"\r\n"), "chunk size line must end in CRLF");
212        let text =
213            std::str::from_utf8(&line[..line.len() - 2]).context("chunk size is not ASCII")?;
214        let (size, ext) = match text.split_once(';') {
215            Some((s, e)) => (s.trim(), Some(e.trim())),
216            None => (text.trim(), None),
217        };
218        ensure!(
219            !size.is_empty() && size.len() <= 8 && size.bytes().all(|b| b.is_ascii_hexdigit()),
220            "invalid chunk size '{size}'"
221        );
222        let size = usize::from_str_radix(size, 16)?;
223        if size == 0 {
224            let mut end = [0u8; 2];
225            reader.read_exact(&mut end).await?;
226            ensure!(&end == b"\r\n", "zero chunk must be followed by CRLF");
227            return Ok(Chunks {
228                data,
229                ieof: ext == Some("ieof"),
230            });
231        }
232        ensure!(ext.is_none(), "only the zero chunk may carry an extension");
233        ensure!(
234            size <= MAX_BODY_BYTES.saturating_sub(total + data.len()),
235            "body exceeds {MAX_BODY_BYTES} bytes"
236        );
237        let start = data.len();
238        data.resize(start + size, 0);
239        reader.read_exact(&mut data[start..]).await?;
240        let mut crlf = [0u8; 2];
241        reader.read_exact(&mut crlf).await?;
242        ensure!(&crlf == b"\r\n", "chunk data must be followed by CRLF");
243    }
244    bail!("body exceeds {MAX_CHUNKS} chunks")
245}
246
247pub fn chunked(body: &[u8]) -> Vec<u8> {
248    let mut out = Vec::with_capacity(body.len() + 16);
249    if !body.is_empty() {
250        out.extend_from_slice(format!("{:x}\r\n", body.len()).as_bytes());
251        out.extend_from_slice(body);
252        out.extend_from_slice(b"\r\n");
253    }
254    out.extend_from_slice(b"0\r\n\r\n");
255    out
256}
257
258/// Assemble an ICAP message: head, then the embedded HTTP heads and body, with the
259/// `Encapsulated` header computed from their sizes.
260pub fn message(
261    start: &str,
262    mut icap_headers: Vec<(String, String)>,
263    req: Option<&HttpHead>,
264    res: Option<&HttpHead>,
265    body: Option<&[u8]>,
266    body_kind: &str,
267) -> Result<Vec<u8>> {
268    let mut encapsulated = Vec::new();
269    let mut section = Vec::new();
270    if let Some(h) = req {
271        encapsulated.push(format!("req-hdr={}", section.len()));
272        section.extend_from_slice(render_head(h)?.as_bytes());
273    }
274    if let Some(h) = res {
275        encapsulated.push(format!("res-hdr={}", section.len()));
276        section.extend_from_slice(render_head(h)?.as_bytes());
277    }
278    match body {
279        Some(b) => {
280            encapsulated.push(format!("{body_kind}={}", section.len()));
281            section.extend_from_slice(&chunked(b));
282        }
283        None => encapsulated.push(format!("null-body={}", section.len())),
284    }
285    icap_headers.retain(|(k, _)| !k.eq_ignore_ascii_case("Encapsulated"));
286    icap_headers.push(("Encapsulated".into(), encapsulated.join(", ")));
287    let mut out = format!("{start}\r\n");
288    for (k, v) in &icap_headers {
289        ensure!(token(k) && header_value_ok(v), "invalid ICAP header {k}");
290        out.push_str(&format!("{k}: {v}\r\n"));
291    }
292    out.push_str("\r\n");
293    let mut bytes = out.into_bytes();
294    bytes.extend_from_slice(&section);
295    Ok(bytes)
296}
297
298pub async fn write_all<W: AsyncWrite + Unpin>(writer: &mut W, bytes: &[u8]) -> Result<()> {
299    tokio::time::timeout(IO_TIMEOUT, async {
300        writer.write_all(bytes).await?;
301        writer.flush().await
302    })
303    .await
304    .context("ICAP write deadline")??;
305    Ok(())
306}
307
308pub fn head_json(h: &HttpHead, request: bool) -> Value {
309    let headers: Vec<Value> = h.headers.iter().map(|(k, v)| json!([k, v])).collect();
310    if request {
311        json!({"method": h.start[0], "uri": h.start[1], "version": h.start[2], "headers": headers})
312    } else {
313        json!({"version": h.start[0], "status": h.start[1].parse::<u16>().unwrap_or(0), "reason": h.start[2], "headers": headers})
314    }
315}
316
317fn headers_from(v: &Value) -> Result<Vec<(String, String)>> {
318    match v {
319        Value::Null => Ok(Vec::new()),
320        Value::Array(items) => items
321            .iter()
322            .map(|i| {
323                let pair = i
324                    .as_array()
325                    .filter(|p| p.len() == 2)
326                    .context("headers must be [[name, value], ...]")?;
327                Ok((
328                    pair[0]
329                        .as_str()
330                        .context("header name must be a string")?
331                        .to_owned(),
332                    pair[1]
333                        .as_str()
334                        .context("header value must be a string")?
335                        .to_owned(),
336                ))
337            })
338            .collect(),
339        Value::Object(map) => map
340            .iter()
341            .map(|(k, v)| {
342                Ok((
343                    k.clone(),
344                    v.as_str()
345                        .context("header value must be a string")?
346                        .to_owned(),
347                ))
348            })
349            .collect(),
350        _ => bail!("headers must be an array of [name, value] or an object"),
351    }
352}
353
354/// Structured HTTP request → head.
355pub fn request_head(v: &Value) -> Result<HttpHead> {
356    let method = v["method"].as_str().unwrap_or("GET");
357    ensure!(token(method), "invalid HTTP method");
358    let uri = v["uri"].as_str().context("http_request.uri is required")?;
359    let version = v["version"].as_str().unwrap_or("HTTP/1.1");
360    let head = HttpHead {
361        start: [method.into(), uri.into(), version.into()],
362        headers: headers_from(&v["headers"])?,
363    };
364    render_head(&head)?;
365    Ok(head)
366}
367
368/// Structured HTTP response → head.
369pub fn response_head(v: &Value) -> Result<HttpHead> {
370    let status = v["status"]
371        .as_u64()
372        .context("http_response.status is required")?;
373    ensure!((100..=599).contains(&status), "HTTP status out of range");
374    let reason = v["reason"].as_str().unwrap_or("");
375    let version = v["version"].as_str().unwrap_or("HTTP/1.1");
376    let head = HttpHead {
377        start: [version.into(), status.to_string(), reason.into()],
378        headers: headers_from(&v["headers"])?,
379    };
380    render_head(&head)?;
381    Ok(head)
382}
383
384/// Body as event data: text when it is UTF-8, otherwise its size only.
385pub fn body_json(body: &[u8]) -> Value {
386    match std::str::from_utf8(body) {
387        Ok(t) => json!({"body_text": t, "body_bytes": body.len()}),
388        Err(_) => json!({"body_binary": true, "body_bytes": body.len()}),
389    }
390}
391
392/// Replace or set Content-Length on a head carrying a body we re-encode.
393pub fn set_length(head: &mut HttpHead, len: usize) {
394    head.headers.retain(|(k, _)| {
395        !k.eq_ignore_ascii_case("Content-Length") && !k.eq_ignore_ascii_case("Transfer-Encoding")
396    });
397    head.headers
398        .push(("Content-Length".into(), len.to_string()));
399}
netget/server/icap/wire.rs

netget/server/icap/
wire.rs
