openmoat_proxy/
post.rs

1//! One outbound HTTP `POST` for live audit delivery (ADR-081).
2//!
3//! The person names the endpoint in their delivery configuration, which the
4//! agent cannot write; the request still keeps the proxy's own rules where
5//! they apply:
6//!
7//! - `https` to any host, verified against the system's roots or a CA the
8//!   person gives ([`Roots`]); plain `http` only to this machine's loopback
9//!   addresses, where nothing crosses a network.
10//! - Every resolved address is checked: link-local, cloud metadata and
11//!   non-unicast addresses are refused whatever the configuration says, so a
12//!   name pointed at the metadata service gets no audit log. Private and
13//!   loopback addresses pass: a collector usually lives there.
14//! - No redirect is followed, no proxy variable is read, and only the
15//!   response's status line and headers are read, up to a limit, and the
16//!   body of an accepted answer, up to [`MAX_BODY`]: an OTLP collector says
17//!   there which records it rejected.
18//!
19//! Plain `std::net` and rustls, one connection per request with deadlines,
20//! as the rest of this crate.
21
22use std::io::{self, Read, Write};
23use std::net::{IpAddr, SocketAddr, TcpStream};
24use std::sync::Arc;
25use std::time::Duration;
26
27use rustls::pki_types::ServerName;
28use rustls::{ClientConfig, ClientConnection, StreamOwned};
29
30mod body;
31
32use body::Framing;
33pub use body::MAX_BODY;
34
35use crate::decide::{self, Forbidden};
36use crate::tls::{Roots, TlsError};
37use crate::upstream::Resolve;
38
39/// The most bytes of a response head read before giving up on it.
40const MAX_HEAD: usize = 16 * 1024;
41
42/// The most response headers read.
43const MAX_RESPONSE_HEADERS: usize = 64;
44
45/// Why a request was not made or not answered.
46#[derive(Debug, thiserror::Error)]
47#[non_exhaustive]
48pub enum PostError {
49    /// The URL is not one this client sends to.
50    #[error("{0}")]
51    Url(String),
52    /// A header name or value that would change the request's framing.
53    #[error("header `{0}` has a character a header cannot carry")]
54    Header(String),
55    /// The name resolved to an address that is never reached.
56    #[error("{host} resolves to {addr}, {}, which audit delivery never reaches", why.describe())]
57    Address {
58        /// The host named.
59        host: String,
60        /// The address refused.
61        addr: IpAddr,
62        /// Why.
63        why: Forbidden,
64    },
65    /// Plain `http` to an address that is not loopback.
66    #[error("{host} resolves to {addr}, not a loopback address: use https")]
67    PlainOffMachine {
68        /// The host named.
69        host: String,
70        /// The address that is not loopback.
71        addr: IpAddr,
72    },
73    /// Nothing answered.
74    #[error("cannot reach {0}")]
75    Unreachable(String),
76    /// TLS could not be set up or failed.
77    #[error("TLS: {0}")]
78    Tls(String),
79    /// Reading or writing failed, or timed out.
80    #[error("{0}")]
81    Io(#[from] io::Error),
82    /// The answer is not an HTTP/1.x response head within the limits.
83    #[error("the answer is not an HTTP response: {0}")]
84    Response(String),
85}
86
87/// `http` or `https`.
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub enum Scheme {
90    /// Loopback only.
91    Http,
92    /// TLS, verified.
93    Https,
94}
95
96/// Where a request goes: scheme, host, port and the path with its query.
97#[derive(Debug, Clone, PartialEq, Eq)]
98pub struct Endpoint {
99    /// `http` or `https`.
100    pub scheme: Scheme,
101    /// The host as written, without brackets for an IPv6 literal.
102    pub host: String,
103    /// The port, the scheme's default when none is written.
104    pub port: u16,
105    /// Path and query, `/` when none is written.
106    pub path: String,
107}
108
109impl Endpoint {
110    /// Read `url`. Refused: another scheme, user information, a fragment,
111    /// whitespace or control characters, and `http` to a name other than
112    /// `localhost` or a loopback literal.
113    pub fn parse(url: &str) -> Result<Self, PostError> {
114        let bad = |why: &str| PostError::Url(format!("`{url}`: {why}"));
115        if url.chars().any(|c| c.is_whitespace() || c.is_control()) {
116            return Err(bad("a URL has no spaces or control characters"));
117        }
118        let (scheme, rest) = if let Some(rest) = url.strip_prefix("https://") {
119            (Scheme::Https, rest)
120        } else if let Some(rest) = url.strip_prefix("http://") {
121            (Scheme::Http, rest)
122        } else {
123            return Err(bad("only https:// (or http:// to this machine) is sent to"));
124        };
125        if rest.contains('#') {
126            return Err(bad("a fragment is never sent"));
127        }
128        let (authority, path) = match rest.find(['/', '?']) {
129            Some(at) if rest[at..].starts_with('/') => (&rest[..at], rest[at..].to_owned()),
130            Some(at) => (&rest[..at], format!("/{}", &rest[at..])),
131            None => (rest, "/".to_owned()),
132        };
133        if authority.contains('@') {
134            return Err(bad("credentials go in a header, never in the URL"));
135        }
136        let default = match scheme {
137            Scheme::Http => 80,
138            Scheme::Https => 443,
139        };
140        let (host, port) = split_port(authority, default).ok_or_else(|| bad("no valid host"))?;
141        if decide::metadata_name(host.trim_end_matches('.')) {
142            return Err(bad("a cloud metadata service is never sent to"));
143        }
144        let endpoint = Self {
145            scheme,
146            host,
147            port,
148            path,
149        };
150        if scheme == Scheme::Http && !endpoint.names_loopback() {
151            return Err(bad(
152                "plain http goes only to this machine (localhost, 127.0.0.1, [::1]); use https",
153            ));
154        }
155        Ok(endpoint)
156    }
157
158    fn names_loopback(&self) -> bool {
159        self.host.eq_ignore_ascii_case("localhost")
160            || self.host.parse::<IpAddr>().is_ok_and(|ip| ip.is_loopback())
161    }
162
163    /// The `Host` header's value.
164    fn authority(&self) -> String {
165        let host = if self.host.contains(':') {
166            format!("[{}]", self.host)
167        } else {
168            self.host.clone()
169        };
170        let default = match self.scheme {
171            Scheme::Http => 80,
172            Scheme::Https => 443,
173        };
174        if self.port == default {
175            host
176        } else {
177            format!("{host}:{}", self.port)
178        }
179    }
180}
181
182fn split_port(authority: &str, default: u16) -> Option<(String, u16)> {
183    let (host, port) = if let Some(rest) = authority.strip_prefix('[') {
184        let (host, after) = rest.split_once(']')?;
185        host.parse::<std::net::Ipv6Addr>().ok()?;
186        (host, after.strip_prefix(':'))
187    } else {
188        match authority.split_once(':') {
189            Some((host, port)) => (host, Some(port)),
190            None => (authority, None),
191        }
192    };
193    let valid = |c: char| c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_' | ':');
194    if host.is_empty() || !host.chars().all(valid) {
195        return None;
196    }
197    let port = match port {
198        Some(p) => p.parse::<u16>().ok().filter(|p| *p != 0)?,
199        None => default,
200    };
201    Some((host.to_ascii_lowercase(), port))
202}
203
204/// What the endpoint answered.
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct Answer {
207    /// The HTTP status code.
208    pub status: u16,
209    /// `Retry-After` in seconds, when it gave one in that form.
210    pub retry_after_s: Option<u64>,
211    /// The body of a `2xx` answer, read whole, at most [`MAX_BODY`] bytes;
212    /// `None` for any other answer and for a body longer, cut off, badly
213    /// chunked or not read in time.
214    pub body: Option<Vec<u8>>,
215}
216
217impl Answer {
218    /// A `2xx` status.
219    #[must_use]
220    pub fn accepted(&self) -> bool {
221        (200..300).contains(&self.status)
222    }
223
224    /// A status that asks to try again later: `408`, `429`, any `5xx` but
225    /// `501` and `505`. The OTLP specification names `429`, `502`, `503` and
226    /// `504` as retryable; webhooks commonly use the others too.
227    #[must_use]
228    pub fn retryable(&self) -> bool {
229        matches!(self.status, 408 | 429)
230            || (500..600).contains(&self.status) && !matches!(self.status, 501 | 505)
231    }
232}
233
234/// Sends requests: the TLS configuration and the resolver, kept for every
235/// request of one delivery run.
236pub struct Poster {
237    tls: Arc<ClientConfig>,
238    resolver: Box<dyn Resolve>,
239}
240
241impl std::fmt::Debug for Poster {
242    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
243        f.debug_struct("Poster").finish_non_exhaustive()
244    }
245}
246
247impl Poster {
248    /// A client verifying servers against `roots`, resolving with `resolver`.
249    pub fn new(roots: &Roots, resolver: Box<dyn Resolve>) -> Result<Self, TlsError> {
250        let provider = Arc::new(rustls::crypto::ring::default_provider());
251        Ok(Self {
252            tls: Arc::new(crate::tls::upstream_config(roots, &provider)?),
253            resolver,
254        })
255    }
256
257    /// `POST` `body` to `endpoint` with `headers` (besides `Host`,
258    /// `Content-Length` and `Connection: close`), each step within
259    /// `timeout`, and read the answer's head.
260    pub fn post(
261        &self,
262        endpoint: &Endpoint,
263        headers: &[(&str, &str)],
264        body: &[u8],
265        timeout: Duration,
266    ) -> Result<Answer, PostError> {
267        let request = request_head(endpoint, headers, body.len())?;
268        let tcp = self.connect(endpoint, timeout)?;
269        tcp.set_read_timeout(Some(timeout))?;
270        tcp.set_write_timeout(Some(timeout))?;
271        match endpoint.scheme {
272            Scheme::Http => exchange(tcp, &request, body),
273            Scheme::Https => {
274                let name = ServerName::try_from(endpoint.host.clone()).map_err(|_| {
275                    PostError::Url(format!("{} is not a server name", endpoint.host))
276                })?;
277                let conn = ClientConnection::new(self.tls.clone(), name)
278                    .map_err(|e| PostError::Tls(e.to_string()))?;
279                exchange(StreamOwned::new(conn, tcp), &request, body)
280            }
281        }
282    }
283
284    fn connect(&self, endpoint: &Endpoint, timeout: Duration) -> Result<TcpStream, PostError> {
285        let host = &endpoint.host;
286        let addrs: Vec<SocketAddr> = match host.parse::<IpAddr>() {
287            Ok(ip) => vec![SocketAddr::new(ip, endpoint.port)],
288            Err(_) => self
289                .resolver
290                .resolve(host, endpoint.port)
291                .map_err(|e| PostError::Unreachable(format!("{host}: {e}")))?,
292        };
293        if addrs.is_empty() {
294            return Err(PostError::Unreachable(format!("{host}: no addresses")));
295        }
296        // One bad address refuses the name, as in the proxy.
297        for addr in &addrs {
298            check_address(endpoint, addr.ip())?;
299        }
300        let mut last = None;
301        for addr in &addrs {
302            match TcpStream::connect_timeout(addr, timeout) {
303                Ok(stream) => return Ok(stream),
304                Err(e) => last = Some(e),
305            }
306        }
307        Err(PostError::Unreachable(match last {
308            Some(e) => format!("{host}:{}: {e}", endpoint.port),
309            None => format!("{host}:{}", endpoint.port),
310        }))
311    }
312}
313
314/// Refuse an address audit delivery never reaches, and plain http off the
315/// machine.
316fn check_address(endpoint: &Endpoint, ip: IpAddr) -> Result<(), PostError> {
317    match decide::forbidden(ip) {
318        Some(why @ (Forbidden::LinkLocal | Forbidden::Metadata | Forbidden::NotUnicast)) => {
319            return Err(PostError::Address {
320                host: endpoint.host.clone(),
321                addr: ip,
322                why,
323            });
324        }
325        Some(Forbidden::Loopback) => return Ok(()),
326        _ => {}
327    }
328    if endpoint.scheme == Scheme::Http {
329        return Err(PostError::PlainOffMachine {
330            host: endpoint.host.clone(),
331            addr: ip,
332        });
333    }
334    Ok(())
335}
336
337/// The request line and headers, refusing a header that would change framing.
338fn request_head(
339    endpoint: &Endpoint,
340    headers: &[(&str, &str)],
341    length: usize,
342) -> Result<Vec<u8>, PostError> {
343    let mut head = format!(
344        "POST {} HTTP/1.1\r\nHost: {}\r\nContent-Length: {length}\r\nConnection: close\r\n",
345        endpoint.path,
346        endpoint.authority()
347    );
348    for (name, value) in headers {
349        let token = |c: char| c.is_ascii_alphanumeric() || "!#$%&'*+-.^_`|~".contains(c);
350        let framing = ["host", "content-length", "connection", "transfer-encoding"];
351        if name.is_empty()
352            || !name.chars().all(token)
353            || framing.iter().any(|f| name.eq_ignore_ascii_case(f))
354            || value.chars().any(|c| c.is_control() && c != '\t')
355        {
356            return Err(PostError::Header((*name).to_owned()));
357        }
358        head.push_str(name);
359        head.push_str(": ");
360        head.push_str(value);
361        head.push_str("\r\n");
362    }
363    head.push_str("\r\n");
364    Ok(head.into_bytes())
365}
366
367/// Write the request and read the answer's head from `stream`, and the body
368/// of an accepted answer ([`Answer::body`]).
369pub(crate) fn exchange(
370    mut stream: impl Read + Write,
371    head: &[u8],
372    body: &[u8],
373) -> Result<Answer, PostError> {
374    stream.write_all(head)?;
375    stream.write_all(body)?;
376    stream.flush()?;
377    let mut buf = Vec::with_capacity(1024);
378    let mut chunk = [0u8; 2048];
379    loop {
380        if let Some((mut answer, end, framing)) = parse_head(&buf)? {
381            if answer.accepted() {
382                let after = buf.split_off(end);
383                answer.body = body::read(&mut stream, after, framing);
384            }
385            return Ok(answer);
386        }
387        if buf.len() >= MAX_HEAD {
388            return Err(PostError::Response(format!(
389                "no end of head in {MAX_HEAD} bytes"
390            )));
391        }
392        let n = stream.read(&mut chunk)?;
393        if n == 0 {
394            return Err(PostError::Response(
395                "closed before the head ended".to_owned(),
396            ));
397        }
398        buf.extend_from_slice(&chunk[..n]);
399    }
400}
401
402/// The answer in `buf` once its head is whole, where the head ends, and how
403/// the body after it is framed.
404fn parse_head(buf: &[u8]) -> Result<Option<(Answer, usize, Framing)>, PostError> {
405    let mut headers = [httparse::EMPTY_HEADER; MAX_RESPONSE_HEADERS];
406    let mut response = httparse::Response::new(&mut headers);
407    match response.parse(buf) {
408        Ok(httparse::Status::Partial) => Ok(None),
409        Ok(httparse::Status::Complete(end)) => {
410            let status = response
411                .code
412                .ok_or_else(|| PostError::Response("no status".to_owned()))?;
413            let value = |name: &str| {
414                response
415                    .headers
416                    .iter()
417                    .find(|h| h.name.eq_ignore_ascii_case(name))
418                    .and_then(|h| std::str::from_utf8(h.value).ok())
419                    .map(str::trim)
420            };
421            let retry_after_s = value("retry-after").and_then(|v| v.parse::<u64>().ok());
422            let chunked = value("transfer-encoding")
423                .is_some_and(|v| v.to_ascii_lowercase().ends_with("chunked"));
424            // Chunked wins over a length (RFC 9112 §6.3); a length that does
425            // not read leaves the body unread.
426            let framing = match (status, chunked, value("content-length")) {
427                (204 | 304, _, _) => Framing::Empty,
428                (_, true, _) => Framing::Chunked,
429                (_, false, Some(length)) => length
430                    .parse::<usize>()
431                    .map_or(Framing::Length(usize::MAX), Framing::Length),
432                (_, false, None) => Framing::Close,
433            };
434            let answer = Answer {
435                status,
436                retry_after_s,
437                body: None,
438            };
439            Ok(Some((answer, end, framing)))
440        }
441        Err(e) => Err(PostError::Response(e.to_string())),
442    }
443}
444
445#[cfg(test)]
446mod tests;
openmoat_proxy/post.rs

openmoat_proxy/
post.rs
