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

m4a_agent/engine/
http.rs
