security_client_rs/http/
transport.rs

1use std::{fmt, sync::Arc, time::Duration};
2
3use base64::{
4    Engine,
5    alphabet::STANDARD,
6    engine::{DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig, general_purpose},
7};
8use http::{
9    HeaderMap, HeaderName, HeaderValue, Method,
10    header::{AUTHORIZATION, CONTENT_TYPE, USER_AGENT},
11};
12use percent_encoding::percent_decode_str;
13use serde::Serialize;
14use url::Url;
15
16use super::{Body, Response, body::Content, merge_headers, redacted, response::read_bounded};
17use crate::{Error, Result};
18
19/// The address used by [`Transport::single_node`] examples and local development stacks.
20pub const DEFAULT_ADDRESS: &str = "http://localhost:5601";
21
22pub(crate) const ERROR_LIMIT: usize = 16 * 1024;
23const RESPONSE_LIMIT: usize = 32 * 1024 * 1024;
24
25/// Credentials are redacted from `Debug` output.
26#[derive(Clone)]
27#[non_exhaustive]
28pub enum Credentials {
29    /// Username and password.
30    Basic(String, String),
31    /// An OAuth or Elasticsearch token service bearer token.
32    Bearer(String),
33    /// An API key as its ID and secret.
34    ApiKey(String, String),
35    /// A Base64-encoded API key, as displayed by Kibana when a key is created.
36    EncodedApiKey(String),
37}
38
39impl Credentials {
40    fn header(&self) -> Result<HeaderValue> {
41        let encode =
42            |id: &str, secret: &str| general_purpose::STANDARD.encode(format!("{id}:{secret}"));
43        let value = match self {
44            Self::Basic(username, password) => format!("Basic {}", encode(username, password)),
45            Self::Bearer(token) => format!("Bearer {token}"),
46            Self::ApiKey(id, key) => format!("ApiKey {}", encode(id, key)),
47            Self::EncodedApiKey(key) => format!("ApiKey {key}"),
48        };
49        let mut header = HeaderValue::from_str(&value).map_err(|_| {
50            Error::Configuration("credential contains invalid header characters".into())
51        })?;
52        header.set_sensitive(true);
53        Ok(header)
54    }
55}
56
57impl fmt::Debug for Credentials {
58    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
59        f.write_str(match self {
60            Self::Basic(..) => "Credentials::Basic([redacted])",
61            Self::Bearer(_) => "Credentials::Bearer([redacted])",
62            Self::ApiKey(..) => "Credentials::ApiKey([redacted])",
63            Self::EncodedApiKey(_) => "Credentials::EncodedApiKey([redacted])",
64        })
65    }
66}
67
68/// A trusted root certificate for the Kibana server.
69#[derive(Clone)]
70pub struct Certificate(reqwest::Certificate);
71
72impl Certificate {
73    /// Every certificate in `pem` is trusted, so a CA bundle can be passed as is.
74    /// Fails when `pem` contains no certificate.
75    pub fn from_pem(pem: &[u8]) -> Result<Self> {
76        let invalid = || Error::Configuration("no PEM certificate found".into());
77        let bundle = reqwest::Certificate::from_pem_bundle(pem).map_err(|_| invalid())?;
78        if bundle.is_empty() {
79            return Err(invalid());
80        }
81        reqwest::Certificate::from_pem(pem)
82            .map(Self)
83            .map_err(|_| invalid())
84    }
85
86    /// The DER encoding is checked when the transport is built.
87    pub fn from_der(der: &[u8]) -> Result<Self> {
88        reqwest::Certificate::from_der(der)
89            .map(Self)
90            .map_err(|_| Error::Configuration("invalid DER certificate".into()))
91    }
92}
93
94impl fmt::Debug for Certificate {
95    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
96        f.write_str("Certificate")
97    }
98}
99
100enum Proxy {
101    System,
102    Disabled,
103    Url(Url),
104}
105
106/// Configures a [`Transport`]. Requests have a 60 second timeout, a 10 second
107/// connect timeout and a 32 MiB response limit unless changed here.
108///
109/// `Debug` output omits credentials, header values and proxy passwords.
110pub struct TransportBuilder {
111    url: Url,
112    credentials: Option<Credentials>,
113    headers: HeaderMap,
114    timeout: Duration,
115    connect_timeout: Duration,
116    certificates: Vec<Certificate>,
117    proxy: Proxy,
118    response_limit: usize,
119}
120
121impl TransportBuilder {
122    /// Every request path is appended to `url`, so a reverse-proxy base path is preserved.
123    pub fn new(url: Url) -> Self {
124        Self {
125            url,
126            credentials: None,
127            headers: HeaderMap::new(),
128            timeout: Duration::from_secs(60),
129            connect_timeout: Duration::from_secs(10),
130            certificates: Vec::new(),
131            proxy: Proxy::System,
132            response_limit: RESPONSE_LIMIT,
133        }
134    }
135
136    pub fn auth(mut self, credentials: Credentials) -> Self {
137        self.credentials = Some(credentials);
138        self
139    }
140
141    /// Sends a header with every request, replacing earlier values of the same name.
142    ///
143    /// Headers are combined in this order, each replacing every earlier value
144    /// of a name it sets: built-in defaults and credentials, these transport
145    /// headers, the body's `Content-Type`, then per-request headers.
146    pub fn header(mut self, name: HeaderName, value: HeaderValue) -> Self {
147        self.headers.insert(name, value);
148        self
149    }
150
151    /// Sends these headers with every request. Each name replaces earlier values
152    /// of that name; repeated values of one name in `headers` are all sent.
153    pub fn headers(mut self, headers: HeaderMap) -> Self {
154        merge_headers(&mut self.headers, headers);
155        self
156    }
157
158    /// The total time allowed for a request, including reading the body.
159    pub fn timeout(mut self, timeout: Duration) -> Self {
160        self.timeout = timeout;
161        self
162    }
163
164    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
165        self.connect_timeout = timeout;
166        self
167    }
168
169    /// Trusts an additional root certificate alongside the platform roots.
170    pub fn root_certificate(mut self, certificate: Certificate) -> Self {
171        self.certificates.push(certificate);
172        self
173    }
174
175    /// Routes every request through this proxy. Credentials may be embedded in the URL.
176    pub fn proxy(mut self, url: Url) -> Self {
177        self.proxy = Proxy::Url(url);
178        self
179    }
180
181    /// Ignores proxy environment variables such as `HTTPS_PROXY`, which are honored by default.
182    pub fn disable_proxy(mut self) -> Self {
183        self.proxy = Proxy::Disabled;
184        self
185    }
186
187    /// The largest body `Response::json`, `bytes` and `text` will read.
188    /// Streaming with [`Response::bytes_stream`] bypasses this limit.
189    /// A zero limit fails when [`Self::build`] is called.
190    pub fn response_limit(mut self, bytes: usize) -> Self {
191        self.response_limit = bytes;
192        self
193    }
194
195    /// Builds the connection pool and validates its settings. Invalid base URLs,
196    /// credentials, certificates or proxy settings return [`Error::Configuration`].
197    pub fn build(self) -> Result<Transport> {
198        let url = self.url;
199        if !matches!(url.scheme(), "http" | "https")
200            || url.host_str().is_none()
201            || !url.username().is_empty()
202            || url.password().is_some()
203            || url.query().is_some()
204            || url.fragment().is_some()
205        {
206            return Err(Error::Configuration(
207                "base URL must use HTTP(S), without credentials, query, or fragment".into(),
208            ));
209        }
210        if self.response_limit == 0 {
211            return Err(Error::Configuration(
212                "response limit must be positive".into(),
213            ));
214        }
215
216        let mut headers = HeaderMap::new();
217        headers.insert("kbn-xsrf", HeaderValue::from_static("security-client-rs"));
218        headers.insert(
219            USER_AGENT,
220            HeaderValue::from_static(concat!("security-client-rs/", env!("CARGO_PKG_VERSION"))),
221        );
222        if let Some(credentials) = &self.credentials {
223            headers.insert(AUTHORIZATION, credentials.header()?);
224        }
225        merge_headers(&mut headers, self.headers);
226
227        let mut client = reqwest::Client::builder()
228            .timeout(self.timeout)
229            .connect_timeout(self.connect_timeout)
230            .redirect(reqwest::redirect::Policy::none())
231            .retry(reqwest::retry::never());
232        for certificate in self.certificates {
233            client = client.add_root_certificate(certificate.0);
234        }
235        client = match self.proxy {
236            Proxy::System => client,
237            Proxy::Disabled => client.no_proxy(),
238            Proxy::Url(url) => client.proxy(
239                reqwest::Proxy::all(url)
240                    .map_err(|_| Error::Configuration("invalid proxy URL".into()))?,
241            ),
242        };
243
244        Ok(Transport {
245            inner: Arc::new(Inner {
246                client: client.build().map_err(|_| {
247                    Error::Configuration("invalid TLS configuration or root certificate".into())
248                })?,
249                url,
250                headers,
251                response_limit: self.response_limit,
252            }),
253        })
254    }
255}
256
257impl fmt::Debug for TransportBuilder {
258    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
259        let proxy = match &self.proxy {
260            Proxy::System => "system".to_owned(),
261            Proxy::Disabled => "disabled".to_owned(),
262            Proxy::Url(url) => redacted(url),
263        };
264        f.debug_struct("TransportBuilder")
265            .field("url", &redacted(&self.url))
266            .field("credentials", &self.credentials)
267            .field("headers", &self.headers.keys().collect::<Vec<_>>())
268            .field("timeout", &self.timeout)
269            .field("connect_timeout", &self.connect_timeout)
270            .field("certificates", &self.certificates.len())
271            .field("proxy", &proxy)
272            .field("response_limit", &self.response_limit)
273            .finish()
274    }
275}
276
277struct Inner {
278    client: reqwest::Client,
279    url: Url,
280    headers: HeaderMap,
281    response_limit: usize,
282}
283
284/// Sends requests to one Kibana address. Clones share the connection pool.
285///
286/// Requests are never retried and redirects are never followed. An interrupted
287/// mutation can have an unknown outcome and must be reconciled by the caller.
288#[derive(Clone)]
289pub struct Transport {
290    inner: Arc<Inner>,
291}
292
293impl Transport {
294    pub fn single_node(url: &str) -> Result<Self> {
295        let url = Url::parse(url)
296            .map_err(|_| Error::Configuration("expected an absolute HTTP(S) URL".into()))?;
297        TransportBuilder::new(url).build()
298    }
299
300    /// Connects to the Kibana endpoint encoded in an Elastic Cloud ID.
301    pub fn cloud(cloud_id: &str, credentials: Credentials) -> Result<Self> {
302        TransportBuilder::new(cloud_url(cloud_id)?)
303            .auth(credentials)
304            .build()
305    }
306
307    pub fn url(&self) -> &Url {
308        &self.inner.url
309    }
310
311    /// Sends one request and checks its status.
312    ///
313    /// `path` must already be percent-encoded and is appended to the base URL.
314    /// It cannot contain a query, fragment, backslash, control character or dot
315    /// segment in literal or percent-encoded form, so it always stays under the
316    /// base path. Non-success statuses become [`Error::Api`] with at most 16 KiB of body.
317    pub async fn send<Q>(
318        &self,
319        method: Method,
320        path: &str,
321        headers: HeaderMap,
322        query: Option<&Q>,
323        body: Option<Body>,
324        timeout: Option<Duration>,
325    ) -> Result<Response>
326    where
327        Q: Serialize + ?Sized,
328    {
329        self.send_named("request", method, path, headers, query, body, timeout)
330            .await
331    }
332
333    /// [`send`](Self::send) for a named endpoint, which labels its trace events.
334    #[allow(clippy::too_many_arguments)]
335    pub(crate) async fn send_named<Q>(
336        &self,
337        operation: &'static str,
338        method: Method,
339        path: &str,
340        headers: HeaderMap,
341        query: Option<&Q>,
342        body: Option<Body>,
343        timeout: Option<Duration>,
344    ) -> Result<Response>
345    where
346        Q: Serialize + ?Sized,
347    {
348        #[cfg(feature = "tracing")]
349        let trace =
350            super::trace::Request::start(operation, &method, path, &headers, &self.inner.headers);
351        let result = self
352            .dispatch(method, path, headers, query, body, timeout)
353            .await
354            .map(|response| response.named(operation));
355        #[cfg(feature = "tracing")]
356        let result = {
357            trace.finish(&result);
358            result.map(|response| response.traced(trace))
359        };
360        result
361    }
362
363    async fn dispatch<Q>(
364        &self,
365        method: Method,
366        path: &str,
367        headers: HeaderMap,
368        query: Option<&Q>,
369        body: Option<Body>,
370        timeout: Option<Duration>,
371    ) -> Result<Response>
372    where
373        Q: Serialize + ?Sized,
374    {
375        let url = self.endpoint(path, query)?;
376        // The final headers are prepared once, so a name set at a later stage
377        // replaces every earlier value instead of being sent twice.
378        let mut final_headers = self.inner.headers.clone();
379        let mut request = self.inner.client.request(method, url);
380        request = match body.map(|body| body.0) {
381            None => request,
382            Some(Content::Json(bytes)) => {
383                final_headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
384                request.body(bytes)
385            }
386            Some(Content::File {
387                field,
388                file_name,
389                content_type,
390                bytes,
391            }) => {
392                // Multipart sets its own Content-Type with the boundary.
393                final_headers.remove(CONTENT_TYPE);
394                request.multipart(Content::form(field, file_name, &content_type, bytes)?)
395            }
396        };
397        merge_headers(&mut final_headers, headers);
398        request = request.headers(final_headers);
399        if let Some(timeout) = timeout {
400            request = request.timeout(timeout);
401        }
402
403        let mut response = request.send().await?;
404        let status = response.status();
405        if !status.is_success() {
406            let headers = Box::new(std::mem::take(response.headers_mut()));
407            let body = read_bounded(response, ERROR_LIMIT).await;
408            return Err(Error::Api {
409                status,
410                headers,
411                body: String::from_utf8_lossy(&body.bytes).into_owned(),
412                truncated: body.truncated,
413                body_error: body.error,
414            });
415        }
416        Ok(Response::new(response, self.inner.response_limit))
417    }
418
419    fn endpoint<Q: Serialize + ?Sized>(&self, path: &str, query: Option<&Q>) -> Result<Url> {
420        if path.contains(['?', '#', '\\']) || path.chars().any(char::is_control) {
421            return Err(Error::InvalidRequest(
422                "path must not contain a query, fragment, backslash or control character".into(),
423            ));
424        }
425        // URL parsing resolves `.` and `..` segments, including `%2e` forms.
426        if path.split('/').any(|segment| {
427            let decoded = percent_decode_str(segment).decode_utf8_lossy();
428            decoded == "." || decoded == ".."
429        }) {
430            return Err(Error::InvalidRequest(
431                "path must not contain dot segments".into(),
432            ));
433        }
434        let mut url = self.inner.url.clone();
435        let base = url.path().trim_end_matches('/').to_owned();
436        let joined = format!("{base}/{}", path.trim_start_matches('/'));
437        url.set_path(&joined);
438        if !url.path().starts_with(&format!("{base}/")) {
439            return Err(Error::InvalidRequest(
440                "path must stay under the base URL path".into(),
441            ));
442        }
443        if let Some(query) = query {
444            let query = serde_urlencoded::to_string(query).map_err(Error::serialize)?;
445            if !query.is_empty() {
446                url.set_query(Some(&query));
447            }
448        }
449        Ok(url)
450    }
451}
452
453impl fmt::Debug for Transport {
454    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
455        f.debug_struct("Transport")
456            .field("url", &self.inner.url.as_str())
457            .finish_non_exhaustive()
458    }
459}
460
461/// A Cloud ID is `name:base64(host[:port]$elasticsearch-id$kibana-id)`.
462fn cloud_url(cloud_id: &str) -> Result<Url> {
463    let invalid = || Error::Configuration("invalid Elastic Cloud ID".into());
464    let (_, encoded) = cloud_id.rsplit_once(':').ok_or_else(invalid)?;
465    let engine = GeneralPurpose::new(
466        &STANDARD,
467        GeneralPurposeConfig::new().with_decode_padding_mode(DecodePaddingMode::Indifferent),
468    );
469    let decoded =
470        String::from_utf8(engine.decode(encoded).map_err(|_| invalid())?).map_err(|_| invalid())?;
471    let mut parts = decoded.split('$');
472    let host = parts.next().filter(|h| !h.is_empty()).ok_or_else(invalid)?;
473    let kibana = parts.nth(1).filter(|k| !k.is_empty()).ok_or_else(|| {
474        Error::Configuration("Elastic Cloud ID does not include a Kibana endpoint".into())
475    })?;
476    let (host, port) = host.split_once(':').unwrap_or((host, "443"));
477    Url::parse(&format!("https://{kibana}.{host}:{port}")).map_err(|_| invalid())
478}
479
480#[cfg(test)]
481mod tests {
482    use super::*;
483
484    fn cloud_id(decoded: &str) -> String {
485        format!("deployment:{}", general_purpose::STANDARD.encode(decoded))
486    }
487
488    #[test]
489    fn cloud_ids_resolve_the_kibana_endpoint() {
490        let url = cloud_url(&cloud_id("eu-west-1.aws.found.io$es-uuid$kb-uuid")).unwrap();
491        assert_eq!(url.as_str(), "https://kb-uuid.eu-west-1.aws.found.io/");
492        let url = cloud_url(&cloud_id("example.com:9243$es-uuid$kb-uuid")).unwrap();
493        assert_eq!(url.as_str(), "https://kb-uuid.example.com:9243/");
494        let unpadded = cloud_id("host.io$es$kb").trim_end_matches('=').to_owned();
495        assert_eq!(cloud_url(&unpadded).unwrap().host_str(), Some("kb.host.io"));
496        for invalid in [
497            "no-separator".to_owned(),
498            "name:not base64!".to_owned(),
499            cloud_id("host.io$es-uuid"),
500            cloud_id("host.io$es-uuid$"),
501        ] {
502            assert!(cloud_url(&invalid).is_err(), "{invalid}");
503        }
504    }
505
506    #[test]
507    fn credentials_encode_the_documented_schemes() {
508        let header = |c: Credentials| c.header().unwrap().to_str().unwrap().to_owned();
509        assert_eq!(
510            header(Credentials::Basic("elastic".into(), "changeme".into())),
511            "Basic ZWxhc3RpYzpjaGFuZ2VtZQ=="
512        );
513        assert_eq!(
514            header(Credentials::ApiKey("id".into(), "secret".into())),
515            "ApiKey aWQ6c2VjcmV0"
516        );
517        assert_eq!(
518            header(Credentials::EncodedApiKey("aWQ6c2VjcmV0".into())),
519            "ApiKey aWQ6c2VjcmV0"
520        );
521        assert_eq!(header(Credentials::Bearer("token".into())), "Bearer token");
522        assert!(Credentials::Bearer("line\nbreak".into()).header().is_err());
523        assert!(
524            Credentials::Basic("a".into(), "b".into())
525                .header()
526                .unwrap()
527                .is_sensitive()
528        );
529    }
530}
security_client_rs/http/transport.rs

security_client_rs/http/
transport.rs
