m4a_agent/engine/
http.rs

1//! HTTP execution of the engine's requests against one server: URL building (the spec prefix is
2//! kept only when the server was found to serve it), discovery, and the bearer.
3
4use std::time::Duration;
5
6use mail4agent_messenger::{HttpResponseDescriptor, OutgoingRequest};
7use reqwest::Url;
8
9use crate::error::{AgentError, Result};
10
11/// Same clipping the shell uses for text that may reach logs: no long opaque tokens.
12pub fn clip_public(text: String) -> String {
13    let mut out = String::new();
14    for token in text.split_whitespace() {
15        if token.len() > 80 {
16            out.push_str("[omitted]");
17        } else {
18            out.push_str(token);
19        }
20        out.push(' ');
21        if out.len() > 240 {
22            break;
23        }
24    }
25    out
26}
27
28/// A reqwest error with its causes (connect refused, dns, timed out, tls...), URL left out:
29/// "error sending request" alone hid why the box client stayed down for hours.
30pub fn error_chain(e: &reqwest::Error) -> String {
31    let mut out = if e.is_timeout() { "timed out".to_string() } else if e.is_connect() { "connect failed".to_string() } else { "request failed".to_string() };
32    let mut src = std::error::Error::source(e);
33    while let Some(cause) = src {
34        out.push_str(": ");
35        out.push_str(&cause.to_string());
36        src = cause.source();
37    }
38    out
39}
40
41pub fn percent_encode(value: &str) -> String {
42    let mut out = String::new();
43    for byte in value.bytes() {
44        match byte {
45            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => out.push(byte as char),
46            _ => out.push_str(&format!("%{byte:02X}")),
47        }
48    }
49    out
50}
51
52/// A server origin plus the decision whether it serves the Matrix paths under `/_matrix`.
53#[derive(Clone)]
54pub struct HttpExec {
55    client: reqwest::blocking::Client,
56    base: Url,
57    keep_prefix: bool,
58}
59
60pub(crate) fn parse_origin(raw: &str) -> Result<Url> {
61    let url = Url::parse(raw).map_err(|_| AgentError::Protocol("not a valid server URL".into()))?;
62    if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() || url.query().is_some() || !url.username().is_empty() || url.password().is_some() {
63        return Err(AgentError::Protocol("not a valid server URL".into()));
64    }
65    let mut url = url;
66    url.set_fragment(None);
67    Ok(url)
68}
69
70impl HttpExec {
71    pub fn new(base: &str) -> Result<Self> {
72        let client = reqwest::blocking::Client::builder()
73            .timeout(Duration::from_secs(60))
74            // A pooled connection that outlives a server restart answers every request with a
75            // transport failure; short idle life and keepalive make the pool drop it on its own.
76            .pool_idle_timeout(Duration::from_secs(15))
77            .tcp_keepalive(Duration::from_secs(30))
78            .http1_only().build().map_err(|e| AgentError::Transport(clip_public(error_chain(&e))))?;
79        Ok(Self { client, base: parse_origin(base)?, keep_prefix: false })
80    }
81
82    /// Our own mounts serve `/client/v3` without the spec prefix. Anything else is asked, never
83    /// assumed: `GET /_matrix/client/versions` answering 2xx means the prefix is served.
84    pub fn probe_prefix(&mut self) -> bool {
85        self.keep_prefix = true; // ask with the prefix on
86        self.keep_prefix = self.get_status("/_matrix/client/versions").is_some_and(|s| (200..300).contains(&s));
87        self.keep_prefix
88    }
89
90    pub fn set_keep_prefix(&mut self, keep: bool) {
91        self.keep_prefix = keep;
92    }
93
94    pub fn keep_prefix(&self) -> bool {
95        self.keep_prefix
96    }
97
98    pub fn base(&self) -> &Url {
99        &self.base
100    }
101
102    pub fn client(&self) -> &reqwest::blocking::Client {
103        &self.client
104    }
105
106    /// Replaces the origin (`.well-known` told us the real one).
107    pub fn rebase(&mut self, origin: &str) -> Result<()> {
108        self.base = parse_origin(origin)?;
109        Ok(())
110    }
111
112    fn get_status(&self, path: &str) -> Option<u16> {
113        let url = self.url(path, &[]).ok()?;
114        self.client.get(url).send().ok().map(|r| r.status().as_u16())
115    }
116
117    /// `GET path` as JSON (no bearer): `(status, json)`.
118    pub fn get_json(&self, path: &str) -> Result<(u16, serde_json::Value)> {
119        let url = self.url(path, &[])?;
120        let r = self.client.get(url).send().map_err(|e| AgentError::Transport(clip_public(error_chain(&e))))?;
121        let st = r.status().as_u16();
122        Ok((st, r.json().unwrap_or(serde_json::Value::Null)))
123    }
124
125    /// `GET path` as JSON with a bearer: `(status, json)`.
126    pub fn get_json_as(&self, bearer: &str, path: &str) -> Result<(u16, serde_json::Value)> {
127        let url = self.url(path, &[])?;
128        let r = self.client.get(url).bearer_auth(bearer).send().map_err(|e| AgentError::Transport(clip_public(error_chain(&e))))?;
129        let st = r.status().as_u16();
130        Ok((st, r.json().unwrap_or(serde_json::Value::Null)))
131    }
132
133    /// `/_matrix/client/v3/x` becomes `/client/v3/x` unless the server serves the prefix. A path
134    /// already without it is left alone.
135    pub fn url(&self, path: &str, query: &[(String, String)]) -> Result<Url> {
136        let path = if self.keep_prefix { path } else { path.strip_prefix("/_matrix").unwrap_or(path) };
137        if !path.starts_with('/') {
138            return Err(AgentError::Protocol("path must start with a slash".into()));
139        }
140        let mut raw = self.base.as_str().trim_end_matches('/').to_string();
141        raw.push_str(path);
142        if !query.is_empty() {
143            raw.push('?');
144            for (i, (n, v)) in query.iter().enumerate() {
145                if i > 0 {
146                    raw.push('&');
147                }
148                raw.push_str(&percent_encode(n));
149                raw.push('=');
150                raw.push_str(&percent_encode(v));
151            }
152        }
153        Url::parse(&raw).map_err(|_| AgentError::Protocol("not a valid request URL".into()))
154    }
155
156    /// One engine request with `bearer`.
157    pub fn perform(&self, bearer: &str, request: &OutgoingRequest) -> Result<HttpResponseDescriptor> {
158        let url = self.url(&request.path, &request.query)?;
159        let method = reqwest::Method::from_bytes(request.method.as_str().as_bytes()).map_err(|_| AgentError::Protocol("unsupported method".into()))?;
160        let mut header = reqwest::header::HeaderValue::from_str(&format!("Bearer {bearer}")).map_err(|_| AgentError::Protocol("token is not a header value".into()))?;
161        header.set_sensitive(true);
162        let mut builder = self.client.request(method, url).header(reqwest::header::AUTHORIZATION, header);
163        if let Some(body) = &request.body {
164            let bytes = serde_json::to_vec(body).map_err(|e| AgentError::Protocol(e.to_string()))?;
165            builder = builder.header(reqwest::header::CONTENT_TYPE, "application/json").body(bytes);
166        }
167        let response = builder.send().map_err(|e| AgentError::Transport(clip_public(error_chain(&e))))?;
168        let status = response.status().as_u16();
169        let body = response.bytes().map_err(|e| AgentError::Transport(clip_public(error_chain(&e))))?.to_vec();
170        Ok(HttpResponseDescriptor { status, body })
171    }
172}
173
174#[cfg(test)]
175mod tests {
176    use super::*;
177
178    #[test]
179    fn the_spec_prefix_is_kept_only_when_asked() {
180        let mut h = HttpExec::new("http://127.0.0.1:9").expect("base");
181        let q = [("timeout".to_string(), "5".to_string())];
182        assert_eq!(h.url("/_matrix/client/v3/sync", &q).unwrap().as_str(), "http://127.0.0.1:9/client/v3/sync?timeout=5");
183        assert_eq!(h.url("/client/v3/sync", &[]).unwrap().path(), "/client/v3/sync");
184        h.set_keep_prefix(true);
185        assert_eq!(h.url("/_matrix/client/v3/sync", &q).unwrap().as_str(), "http://127.0.0.1:9/_matrix/client/v3/sync?timeout=5");
186        assert!(HttpExec::new("ftp://x").is_err());
187        assert!(HttpExec::new("http://u:p@x").is_err());
188    }
189}
m4a_agent/engine/http.rs

m4a_agent/engine/
http.rs
