nautilus_network/http/
client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! HTTP client implementation with rate limiting and timeout support.
17
18use std::{borrow::Cow, collections::HashMap, str::FromStr, sync::Arc, time::Duration};
19
20use nautilus_core::collections::into_ustr_vec;
21use nautilus_cryptography::providers::install_cryptographic_provider;
22use reqwest::{
23    Method, Response, Url,
24    header::{HeaderMap, HeaderName, HeaderValue},
25    redirect::Policy,
26};
27use ustr::Ustr;
28
29use super::{HttpClientError, HttpResponse, HttpStatus};
30use crate::ratelimiter::{RateLimiter, clock::MonotonicClock, quota::Quota};
31
32/// Default maximum idle connections per host.
33const DEFAULT_POOL_MAX_IDLE_PER_HOST: usize = 32;
34
35/// Default idle connection timeout in seconds.
36const DEFAULT_POOL_IDLE_TIMEOUT_SECS: u64 = 60;
37
38/// Default HTTP/2 keep-alive interval in seconds.
39const DEFAULT_HTTP2_KEEP_ALIVE_SECS: u64 = 30;
40
41/// Default maximum HTTP response body size in bytes (100 MiB).
42///
43/// Bounds peak memory per response so a hostile or malfunctioning endpoint
44/// cannot exhaust memory by streaming an arbitrarily large body. Mirrors the
45/// caps already enforced on the WebSocket and raw-socket paths.
46const DEFAULT_MAX_RESPONSE_BYTES: usize = 100 * 1024 * 1024;
47
48/// Controls whether an HTTP client follows redirects.
49#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
50pub enum HttpRedirectPolicy {
51    /// Follow up to ten redirects, matching the existing client behavior.
52    #[default]
53    Follow,
54    /// Reject every redirect response.
55    Reject,
56}
57
58/// An asynchronous HTTP client with rate limiting, timeouts, and custom headers.
59///
60/// The client uses `reqwest` for I/O and supports default and per-key quotas. Multiple clients
61/// can share the same rate limiter when their requests consume one quota budget.
62#[derive(Clone, Debug)]
63pub struct HttpClient {
64    pub(crate) client: InnerHttpClient,
65    pub(crate) rate_limiters: Arc<[Arc<RateLimiter<Ustr, MonotonicClock>>]>,
66}
67
68#[bon::bon]
69impl HttpClient {
70    /// Returns a builder for a new [`HttpClient`] instance.
71    ///
72    /// Set `rate_limiters` to share quota state across clients. When omitted, the client creates
73    /// one rate limiter from `default_quota` and `keyed_quotas`. An explicit empty vector disables
74    /// rate limiting. Each request awaits every configured limiter with the same keys. A limiter
75    /// without a default quota ignores keys it does not own, allowing independent scopes such as
76    /// per-IP and per-account limits to apply to one request.
77    ///
78    /// # Errors
79    ///
80    /// Returns an error if:
81    /// - Shared rate limiters are combined with quota configuration.
82    /// - The proxy URL is malformed.
83    /// - Building the underlying `reqwest::Client` fails.
84    #[builder(finish_fn = build)]
85    pub fn builder(
86        #[builder(default)] headers: HashMap<String, String>,
87        #[builder(default)] header_keys: Vec<String>,
88        #[builder(default)] keyed_quotas: Vec<(String, Quota)>,
89        default_quota: Option<Quota>,
90        timeout_secs: Option<u64>,
91        proxy_url: Option<String>,
92        rate_limiters: Option<Vec<Arc<RateLimiter<Ustr, MonotonicClock>>>>,
93        #[builder(default)] redirect_policy: HttpRedirectPolicy,
94        #[builder(default = true)] use_system_proxy: bool,
95    ) -> Result<Self, HttpClientError> {
96        let rate_limiters = if let Some(rate_limiters) = rate_limiters {
97            if default_quota.is_some() || !keyed_quotas.is_empty() {
98                return Err(HttpClientError::Error(
99                    "Cannot combine shared rate limiters with quota configuration".to_string(),
100                ));
101            }
102            rate_limiters
103        } else {
104            let keyed_quotas = keyed_quotas
105                .into_iter()
106                .map(|(key, quota)| (Ustr::from(&key), quota))
107                .collect();
108            vec![Arc::new(RateLimiter::new_with_quota(
109                default_quota,
110                keyed_quotas,
111            ))]
112        };
113
114        Self::build(
115            headers,
116            header_keys,
117            timeout_secs,
118            proxy_url,
119            rate_limiters,
120            redirect_policy,
121            use_system_proxy,
122        )
123    }
124
125    fn build(
126        headers: HashMap<String, String>,
127        header_keys: Vec<String>,
128        timeout_secs: Option<u64>,
129        proxy_url: Option<String>,
130        rate_limiters: Vec<Arc<RateLimiter<Ustr, MonotonicClock>>>,
131        redirect_policy: HttpRedirectPolicy,
132        use_system_proxy: bool,
133    ) -> Result<Self, HttpClientError> {
134        install_cryptographic_provider();
135
136        // Build default headers
137        let mut header_map = HeaderMap::new();
138
139        for (key, value) in headers {
140            let header_name = HeaderName::from_str(&key)
141                .map_err(|e| HttpClientError::Error(format!("Invalid header name '{key}': {e}")))?;
142            let header_value = HeaderValue::from_str(&value).map_err(|e| {
143                HttpClientError::Error(format!("Invalid header value for '{key}': {e}"))
144            })?;
145            header_map.insert(header_name, header_value);
146        }
147
148        let mut client_builder = reqwest::Client::builder()
149            .default_headers(header_map)
150            .tcp_nodelay(true)
151            .pool_max_idle_per_host(DEFAULT_POOL_MAX_IDLE_PER_HOST)
152            .pool_idle_timeout(Duration::from_secs(DEFAULT_POOL_IDLE_TIMEOUT_SECS))
153            .http2_keep_alive_interval(Duration::from_secs(DEFAULT_HTTP2_KEEP_ALIVE_SECS))
154            .http2_keep_alive_while_idle(true)
155            .http2_adaptive_window(true)
156            .redirect(match redirect_policy {
157                HttpRedirectPolicy::Follow => Policy::limited(10),
158                HttpRedirectPolicy::Reject => Policy::none(),
159            });
160
161        if let Some(timeout_secs) = timeout_secs {
162            client_builder = client_builder.timeout(Duration::from_secs(timeout_secs));
163        }
164
165        // Configure proxy if provided
166        if let Some(proxy_url) = proxy_url {
167            let proxy = reqwest::Proxy::all(&proxy_url)
168                .map_err(|_| HttpClientError::InvalidProxy("proxy URL is malformed".to_string()))?;
169            client_builder = client_builder.proxy(proxy);
170        } else if !use_system_proxy {
171            client_builder = client_builder.no_proxy();
172        }
173
174        let client = client_builder
175            .build()
176            .map_err(|e| HttpClientError::ClientBuildError(e.to_string()))?;
177
178        // Pre-intern header keys as HeaderName. An invalid key is an error: a silent drop would
179        // make response extraction read nothing.
180        let response_headers = header_keys
181            .into_iter()
182            .map(|key| match HeaderName::from_str(&key) {
183                Ok(name) => Ok((key, name)),
184                Err(e) => Err(HttpClientError::Error(format!(
185                    "Invalid header key '{key}': {e}"
186                ))),
187            })
188            .collect::<Result<Vec<_>, _>>()?;
189
190        let client = InnerHttpClient {
191            client,
192            response_headers: Arc::from(response_headers),
193            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
194        };
195
196        Ok(Self {
197            client,
198            rate_limiters: rate_limiters.into(),
199        })
200    }
201
202    /// Sends an HTTP request.
203    ///
204    /// # Errors
205    ///
206    /// Returns an error if unable to send request or times out.
207    ///
208    /// # Examples
209    ///
210    /// If requesting `/foo/bar`, pass rate-limit keys `["foo/bar", "foo"]`.
211    #[expect(clippy::too_many_arguments)]
212    pub async fn request(
213        &self,
214        method: Method,
215        url: String,
216        params: Option<&HashMap<String, Vec<String>>>,
217        headers: Option<HashMap<String, String>>,
218        body: Option<Vec<u8>>,
219        timeout_secs: Option<u64>,
220        keys: Option<Vec<String>>,
221    ) -> Result<HttpResponse, HttpClientError> {
222        let keys = keys.map(into_ustr_vec);
223
224        self.request_with_ustr_keys(method, url, params, headers, body, timeout_secs, keys)
225            .await
226    }
227
228    /// Sends an HTTP request while redacting the URL from logs and transport errors.
229    ///
230    /// Use this for endpoints whose path or other URL components can carry credentials.
231    ///
232    /// # Errors
233    ///
234    /// Returns an error if unable to send request or times out.
235    #[expect(clippy::too_many_arguments)]
236    pub async fn request_with_url_redacted(
237        &self,
238        method: Method,
239        url: String,
240        params: Option<&HashMap<String, Vec<String>>>,
241        headers: Option<HashMap<String, String>>,
242        body: Option<Vec<u8>>,
243        timeout_secs: Option<u64>,
244        keys: Option<Vec<String>>,
245    ) -> Result<HttpResponse, HttpClientError> {
246        let keys = keys.map(into_ustr_vec);
247        self.await_rate_limits(keys.as_deref()).await;
248
249        self.client
250            .send_request_with_url_redacted(method, url, params, headers, body, timeout_secs)
251            .await
252    }
253
254    /// Sends an HTTP request with serializable query parameters.
255    ///
256    /// This method accepts any type implementing `Serialize` for query parameters,
257    /// which will be automatically encoded into the URL query string using reqwest's
258    /// `.query()` method, avoiding unnecessary `HashMap` allocations.
259    ///
260    /// # Errors
261    ///
262    /// Returns an error if unable to send request or times out.
263    #[expect(clippy::too_many_arguments)]
264    pub async fn request_with_params<P: serde::Serialize>(
265        &self,
266        method: Method,
267        url: String,
268        params: Option<&P>,
269        headers: Option<HashMap<String, String>>,
270        body: Option<Vec<u8>>,
271        timeout_secs: Option<u64>,
272        keys: Option<Vec<String>>,
273    ) -> Result<HttpResponse, HttpClientError> {
274        let keys = keys.map(into_ustr_vec);
275        self.await_rate_limits(keys.as_deref()).await;
276
277        self.client
278            .send_request_with_query(method, url, params, headers, body, timeout_secs)
279            .await
280    }
281
282    /// Sends an HTTP request using pre-interned rate limiter keys.
283    ///
284    /// # Errors
285    ///
286    /// Returns an error if unable to send the request or the request times out.
287    #[expect(clippy::too_many_arguments)]
288    pub async fn request_with_ustr_keys(
289        &self,
290        method: Method,
291        url: String,
292        params: Option<&HashMap<String, Vec<String>>>,
293        headers: Option<HashMap<String, String>>,
294        body: Option<Vec<u8>>,
295        timeout_secs: Option<u64>,
296        keys: Option<Vec<Ustr>>,
297    ) -> Result<HttpResponse, HttpClientError> {
298        self.await_rate_limits(keys.as_deref()).await;
299
300        self.client
301            .send_request(method, url, params, headers, body, timeout_secs)
302            .await
303    }
304
305    pub(crate) async fn await_rate_limits(&self, keys: Option<&[Ustr]>) {
306        RateLimiter::await_limiters_ready(&self.rate_limiters, keys).await;
307    }
308
309    /// Sends an HTTP GET request.
310    ///
311    /// # Errors
312    ///
313    /// Returns an error if unable to send request or times out.
314    pub async fn get(
315        &self,
316        url: String,
317        params: Option<&HashMap<String, Vec<String>>>,
318        headers: Option<HashMap<String, String>>,
319        timeout_secs: Option<u64>,
320        keys: Option<Vec<String>>,
321    ) -> Result<HttpResponse, HttpClientError> {
322        self.request(Method::GET, url, params, headers, None, timeout_secs, keys)
323            .await
324    }
325
326    /// Sends an HTTP POST request.
327    ///
328    /// # Errors
329    ///
330    /// Returns an error if unable to send request or times out.
331    pub async fn post(
332        &self,
333        url: String,
334        params: Option<&HashMap<String, Vec<String>>>,
335        headers: Option<HashMap<String, String>>,
336        body: Option<Vec<u8>>,
337        timeout_secs: Option<u64>,
338        keys: Option<Vec<String>>,
339    ) -> Result<HttpResponse, HttpClientError> {
340        self.request(Method::POST, url, params, headers, body, timeout_secs, keys)
341            .await
342    }
343
344    /// Sends an HTTP PATCH request.
345    ///
346    /// # Errors
347    ///
348    /// Returns an error if unable to send request or times out.
349    pub async fn patch(
350        &self,
351        url: String,
352        params: Option<&HashMap<String, Vec<String>>>,
353        headers: Option<HashMap<String, String>>,
354        body: Option<Vec<u8>>,
355        timeout_secs: Option<u64>,
356        keys: Option<Vec<String>>,
357    ) -> Result<HttpResponse, HttpClientError> {
358        self.request(
359            Method::PATCH,
360            url,
361            params,
362            headers,
363            body,
364            timeout_secs,
365            keys,
366        )
367        .await
368    }
369
370    /// Sends an HTTP DELETE request.
371    ///
372    /// # Errors
373    ///
374    /// Returns an error if unable to send request or times out.
375    pub async fn delete(
376        &self,
377        url: String,
378        params: Option<&HashMap<String, Vec<String>>>,
379        headers: Option<HashMap<String, String>>,
380        timeout_secs: Option<u64>,
381        keys: Option<Vec<String>>,
382    ) -> Result<HttpResponse, HttpClientError> {
383        self.request(
384            Method::DELETE,
385            url,
386            params,
387            headers,
388            None,
389            timeout_secs,
390            keys,
391        )
392        .await
393    }
394}
395
396/// Internal implementation backing [`HttpClient`].
397///
398/// The underlying [`reqwest::Client`] reuses pooled connections and is cheap to clone. Responses
399/// retain only configured header fields, and bodies larger than `max_response_bytes` are rejected.
400#[derive(Clone, Debug)]
401pub struct InnerHttpClient {
402    pub(crate) client: reqwest::Client,
403    pub(crate) response_headers: Arc<[(String, HeaderName)]>,
404    pub(crate) max_response_bytes: usize,
405}
406
407impl InnerHttpClient {
408    /// Sends an HTTP request and returns an [`HttpResponse`].
409    ///
410    /// # Errors
411    ///
412    /// Returns an error if unable to send request or times out.
413    pub async fn send_request(
414        &self,
415        method: Method,
416        url: String,
417        params: Option<&HashMap<String, Vec<String>>>,
418        headers: Option<HashMap<String, String>>,
419        body: Option<Vec<u8>>,
420        timeout_secs: Option<u64>,
421    ) -> Result<HttpResponse, HttpClientError> {
422        self.send_request_with_redaction(method, url, params, headers, body, timeout_secs, false)
423            .await
424    }
425
426    async fn send_request_with_url_redacted(
427        &self,
428        method: Method,
429        url: String,
430        params: Option<&HashMap<String, Vec<String>>>,
431        headers: Option<HashMap<String, String>>,
432        body: Option<Vec<u8>>,
433        timeout_secs: Option<u64>,
434    ) -> Result<HttpResponse, HttpClientError> {
435        self.send_request_with_redaction(method, url, params, headers, body, timeout_secs, true)
436            .await
437    }
438
439    #[expect(clippy::too_many_arguments)]
440    async fn send_request_with_redaction(
441        &self,
442        method: Method,
443        url: String,
444        params: Option<&HashMap<String, Vec<String>>>,
445        headers: Option<HashMap<String, String>>,
446        body: Option<Vec<u8>>,
447        timeout_secs: Option<u64>,
448        redact_url: bool,
449    ) -> Result<HttpResponse, HttpClientError> {
450        let full_url = encode_url_params(&url, params)?;
451        self.send_request_internal(
452            method,
453            full_url.as_ref(),
454            None::<&()>,
455            headers,
456            body,
457            timeout_secs,
458            redact_url,
459        )
460        .await
461    }
462
463    /// Sends an HTTP request with query parameters using reqwest's `.query()` method.
464    ///
465    /// This method accepts any type implementing `Serialize` for query parameters,
466    /// avoiding `HashMap` conversion overhead.
467    ///
468    /// # Errors
469    ///
470    /// Returns an error if unable to send request or times out.
471    pub async fn send_request_with_query<Q: serde::Serialize>(
472        &self,
473        method: Method,
474        url: String,
475        query: Option<&Q>,
476        headers: Option<HashMap<String, String>>,
477        body: Option<Vec<u8>>,
478        timeout_secs: Option<u64>,
479    ) -> Result<HttpResponse, HttpClientError> {
480        self.send_request_internal(method, &url, query, headers, body, timeout_secs, false)
481            .await
482    }
483
484    /// Internal implementation for sending HTTP requests.
485    ///
486    /// # Errors
487    ///
488    /// Returns an error if unable to send request or times out.
489    #[expect(clippy::too_many_arguments)]
490    async fn send_request_internal<Q: serde::Serialize>(
491        &self,
492        method: Method,
493        url: &str,
494        query: Option<&Q>,
495        headers: Option<HashMap<String, String>>,
496        body: Option<Vec<u8>>,
497        timeout_secs: Option<u64>,
498        redact_url: bool,
499    ) -> Result<HttpResponse, HttpClientError> {
500        let reqwest_url =
501            Url::parse(url).map_err(|e| HttpClientError::from(format!("URL parse error: {e}")))?;
502
503        let mut request_builder = self.client.request(method, reqwest_url);
504        let extra_header_count = headers.as_ref().map_or(0, HashMap::len);
505        let body_len = body.as_ref().map_or(0, Vec::len);
506
507        if let Some(headers) = headers {
508            let mut header_map = HeaderMap::with_capacity(headers.len());
509            for (header_key, header_value) in &headers {
510                let key = HeaderName::from_bytes(header_key.as_bytes())
511                    .map_err(|e| HttpClientError::from(format!("Invalid header name: {e}")))?;
512
513                if header_map
514                    .insert(
515                        key.clone(),
516                        header_value.parse().map_err(|e| {
517                            HttpClientError::from(format!("Invalid header value: {e}"))
518                        })?,
519                    )
520                    .is_some()
521                {
522                    log::trace!("Replaced duplicate request header '{key}'");
523                }
524            }
525            request_builder = request_builder.headers(header_map);
526        }
527
528        if let Some(q) = query {
529            request_builder = request_builder.query(q);
530        }
531
532        if let Some(timeout_secs) = timeout_secs {
533            request_builder = request_builder.timeout(Duration::new(timeout_secs, 0));
534        }
535
536        let request = match body {
537            Some(b) => request_builder
538                .body(b)
539                .build()
540                .map_err(|e| http_client_error(e, redact_url))?,
541            None => request_builder
542                .build()
543                .map_err(|e| http_client_error(e, redact_url))?,
544        };
545
546        let query_len = request.url().query().map_or(0, str::len);
547        log::trace!(
548            "Sending HTTP request: method={} extra_headers={extra_header_count} \
549             query_bytes={query_len} body_bytes={body_len}",
550            request.method(),
551        );
552
553        let response = self
554            .client
555            .execute(request)
556            .await
557            .map_err(|e| http_client_error(e, redact_url))?;
558
559        self.to_response_internal(response, redact_url).await
560    }
561
562    /// Converts a `reqwest::Response` into an `HttpResponse`.
563    ///
564    /// Uses pre-interned `HeaderName` values to avoid string-to-header parsing per response.
565    ///
566    /// # Errors
567    ///
568    /// Returns an error if unable to send request or times out.
569    pub async fn to_response(&self, response: Response) -> Result<HttpResponse, HttpClientError> {
570        self.to_response_internal(response, false).await
571    }
572
573    async fn to_response_internal(
574        &self,
575        response: Response,
576        redact_url: bool,
577    ) -> Result<HttpResponse, HttpClientError> {
578        let status_code = response.status();
579        let resp_headers = response.headers();
580        let header_count = resp_headers.len();
581        let mut headers = HashMap::with_capacity(std::cmp::min(
582            self.response_headers.len(),
583            resp_headers.len(),
584        ));
585
586        for (key, name) in self.response_headers.iter() {
587            if let Some(val) = resp_headers.get(name)
588                && let Ok(v) = val.to_str()
589            {
590                headers.insert(key.clone(), v.to_owned());
591            }
592        }
593
594        let status = HttpStatus::new(status_code);
595        let body = self.read_body_capped(response, redact_url).await?;
596
597        log::trace!(
598            "Received HTTP response: status={status_code} headers={header_count} body_bytes={}",
599            body.len(),
600        );
601
602        Ok(HttpResponse {
603            status,
604            headers,
605            body,
606        })
607    }
608
609    /// Reads the response body, rejecting any body that exceeds `max_response_bytes`.
610    ///
611    /// A `Content-Length` larger than the cap is rejected up front; otherwise the
612    /// body is streamed chunk-by-chunk and aborted as soon as the accumulated size
613    /// would exceed the cap, so an oversized or unbounded (chunked) body is never
614    /// fully buffered into memory.
615    ///
616    /// # Errors
617    ///
618    /// Returns an error if the body exceeds the configured maximum size, or if
619    /// reading a chunk fails.
620    async fn read_body_capped(
621        &self,
622        mut response: Response,
623        redact_url: bool,
624    ) -> Result<bytes::Bytes, HttpClientError> {
625        let max = self.max_response_bytes;
626
627        // Fast path: reject up front when the advertised length already exceeds the cap.
628        if let Some(len) = response.content_length()
629            && len > max as u64
630        {
631            return Err(HttpClientError::Error(format!(
632                "HTTP response body of {len} bytes exceeds maximum of {max} bytes",
633            )));
634        }
635
636        let mut buf = bytes::BytesMut::new();
637
638        while let Some(chunk) = response
639            .chunk()
640            .await
641            .map_err(|e| http_client_error(e, redact_url))?
642        {
643            if buf.len() + chunk.len() > max {
644                return Err(HttpClientError::Error(format!(
645                    "HTTP response body exceeds maximum of {max} bytes",
646                )));
647            }
648            buf.extend_from_slice(&chunk);
649        }
650
651        Ok(buf.freeze())
652    }
653}
654
655fn http_client_error(error: reqwest::Error, redact_url: bool) -> HttpClientError {
656    if redact_url {
657        HttpClientError::from(error.without_url())
658    } else {
659        HttpClientError::from(error)
660    }
661}
662
663impl Default for InnerHttpClient {
664    /// Creates a new default [`InnerHttpClient`] instance.
665    ///
666    /// The default client is initialized with an empty list of header keys and a new `reqwest::Client`.
667    fn default() -> Self {
668        install_cryptographic_provider();
669        let client = reqwest::Client::new();
670        Self {
671            client,
672            response_headers: Arc::default(),
673            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
674        }
675    }
676}
677
678/// Encodes URL parameters into the query string.
679///
680/// Returns `Cow::Borrowed` when no parameters need appending (zero-alloc fast path).
681/// Parameters can have multiple values per key (for doseq=True behavior).
682/// Preserves existing query strings in the URL by appending with '&' instead of '?'.
683/// The query is inserted before any fragment, which is preserved unchanged.
684fn encode_url_params<'a>(
685    url: &'a str,
686    params: Option<&HashMap<String, Vec<String>>>,
687) -> Result<Cow<'a, str>, HttpClientError> {
688    let Some(params) = params else {
689        return Ok(Cow::Borrowed(url));
690    };
691
692    let pairs: Vec<(&str, &str)> = params
693        .iter()
694        .flat_map(|(key, values)| {
695            values
696                .iter()
697                .map(move |value| (key.as_str(), value.as_str()))
698        })
699        .collect();
700
701    if pairs.is_empty() {
702        return Ok(Cow::Borrowed(url));
703    }
704
705    let query_string = serde_urlencoded::to_string(pairs)
706        .map_err(|e| HttpClientError::Error(format!("Failed to encode params: {e}")))?;
707
708    // The first literal '#' starts the fragment per RFC 3986 section 3.5.
709    // A data '#' in an earlier component must be percent-encoded as "%23".
710    let (base, fragment) = match url.split_once('#') {
711        Some((base, fragment)) => (base, Some(fragment)),
712        None => (url, None),
713    };
714    let separator = if base.contains('?') { '&' } else { '?' };
715
716    Ok(Cow::Owned(match fragment {
717        Some(fragment) => format!("{base}{separator}{query_string}#{fragment}"),
718        None => format!("{base}{separator}{query_string}"),
719    }))
720}
721
722#[cfg(test)]
723mod encode_url_params_tests {
724    use std::{borrow::Cow, collections::HashMap};
725
726    use rstest::rstest;
727
728    use super::encode_url_params;
729
730    fn params(pairs: &[(&str, &str)]) -> HashMap<String, Vec<String>> {
731        let mut map: HashMap<String, Vec<String>> = HashMap::new();
732
733        for (key, value) in pairs {
734            map.entry((*key).to_string())
735                .or_default()
736                .push((*value).to_string());
737        }
738
739        map
740    }
741
742    #[rstest]
743    #[case("https://x/y", "https://x/y?a=b")]
744    #[case("https://x/y?old=1", "https://x/y?old=1&a=b")]
745    #[case("https://x/y#frag", "https://x/y?a=b#frag")]
746    #[case("https://x/y?old=1#frag", "https://x/y?old=1&a=b#frag")]
747    #[case(
748        "https://x/y#section?display=full",
749        "https://x/y?a=b#section?display=full"
750    )]
751    #[case("https://x/y#", "https://x/y?a=b#")]
752    fn test_query_is_inserted_before_the_fragment(#[case] url: &str, #[case] expected: &str) {
753        let params = params(&[("a", "b")]);
754
755        assert_eq!(encode_url_params(url, Some(&params)).unwrap(), expected);
756    }
757
758    #[rstest]
759    fn test_url_is_borrowed_when_no_params_are_supplied() {
760        assert!(matches!(
761            encode_url_params("https://x/y#frag", None).unwrap(),
762            Cow::Borrowed("https://x/y#frag")
763        ));
764    }
765
766    #[rstest]
767    fn test_url_is_borrowed_when_params_are_empty() {
768        let params = HashMap::new();
769
770        assert!(matches!(
771            encode_url_params("https://x/y#frag", Some(&params)).unwrap(),
772            Cow::Borrowed("https://x/y#frag")
773        ));
774    }
775}
776
777#[cfg(test)]
778#[cfg(target_os = "linux")] // Only run network tests on Linux (CI stability)
779mod tests {
780    use std::{net::SocketAddr, num::NonZeroU32};
781
782    use axum::{
783        Router,
784        body::to_bytes,
785        extract::Request,
786        response::IntoResponse,
787        routing::{any, delete, get, patch, post},
788        serve,
789    };
790    use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
791    use http::status::StatusCode;
792    use log::Level;
793    #[cfg(all(feature = "simulation", madsim))]
794    use madsim::task as test_task;
795    use rstest::rstest;
796    #[cfg(not(all(feature = "simulation", madsim)))]
797    use tokio::task as test_task;
798    use tokio::{
799        io::{AsyncReadExt, AsyncWriteExt},
800        sync::oneshot,
801    };
802
803    use super::*;
804    use crate::logging::tests::capture_logs;
805
806    async fn capture_request(request: Request) -> impl IntoResponse {
807        let (parts, body) = request.into_parts();
808        let body = to_bytes(body, usize::MAX).await.unwrap();
809        let default_header = parts.headers.get("x-default").unwrap().to_str().unwrap();
810        let request_header = parts.headers.get("x-request").unwrap().to_str().unwrap();
811        let query = parts.uri.query().unwrap_or_default();
812        let body = String::from_utf8(body.to_vec()).unwrap();
813        let capture = format!(
814            "{}\n{}\n{query}\n{default_header}\n{request_header}\n{body}",
815            parts.method,
816            parts.uri.path(),
817        );
818
819        ([("x-response-id", "response-42")], capture)
820    }
821
822    fn create_router() -> Router {
823        Router::new()
824            .route("/get", get(|| async { "hello-world!" }))
825            .route("/post", post(|| async { StatusCode::OK }))
826            .route("/patch", patch(|| async { StatusCode::OK }))
827            .route("/delete", delete(|| async { StatusCode::OK }))
828            .route("/capture", any(capture_request))
829            .route("/notfound", get(|| async { StatusCode::NOT_FOUND }))
830            .route(
831                "/redirect",
832                get(|| async { (StatusCode::TEMPORARY_REDIRECT, [("location", "/get")]) }),
833            )
834            .route(
835                "/slow",
836                get(|| async {
837                    tokio::time::sleep(Duration::from_secs(2)).await;
838                    "Eventually responded"
839                }),
840            )
841            .route(
842                "/large",
843                // Returns a 1 MiB body to exercise the response size cap.
844                get(|| async { "x".repeat(1024 * 1024) }),
845            )
846    }
847
848    async fn start_test_server() -> Result<SocketAddr, Box<dyn std::error::Error + Send + Sync>> {
849        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
850        let addr = listener.local_addr().unwrap();
851
852        tokio::spawn(async move {
853            serve(listener, create_router()).await.unwrap();
854        });
855
856        Ok(addr)
857    }
858
859    async fn spawn_connection_dropper() -> (SocketAddr, tokio::task::JoinHandle<()>) {
860        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
861        let addr = listener.local_addr().unwrap();
862
863        let task = tokio::spawn(async move {
864            loop {
865                let (stream, _) = listener.accept().await.unwrap();
866                drop(stream);
867            }
868        });
869
870        (addr, task)
871    }
872
873    async fn spawn_chunked_response_server() -> (SocketAddr, tokio::task::JoinHandle<()>) {
874        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
875        let addr = listener.local_addr().unwrap();
876        let task = tokio::spawn(async move {
877            let (mut stream, _) = listener.accept().await.unwrap();
878            let mut request = Vec::new();
879            let mut chunk = [0u8; 1024];
880
881            loop {
882                let read = stream.read(&mut chunk).await.unwrap();
883                if read == 0 {
884                    break;
885                }
886                request.extend_from_slice(&chunk[..read]);
887                if request.windows(4).any(|window| window == b"\r\n\r\n") {
888                    break;
889                }
890            }
891
892            stream
893                .write_all(
894                    b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n\
895                      5\r\nfirst\r\n6\r\nsecond\r\n0\r\n\r\n",
896                )
897                .await
898                .unwrap();
899        });
900
901        (addr, task)
902    }
903
904    async fn spawn_rejecting_connect_proxy() -> (SocketAddr, oneshot::Receiver<String>) {
905        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
906        let addr = listener.local_addr().unwrap();
907        let (request_tx, request_rx) = oneshot::channel();
908
909        tokio::spawn(async move {
910            let (mut stream, _) = listener.accept().await.unwrap();
911            let mut request = Vec::new();
912            let mut chunk = [0u8; 1024];
913            loop {
914                let read = stream.read(&mut chunk).await.unwrap();
915                if read == 0 {
916                    break;
917                }
918                request.extend_from_slice(&chunk[..read]);
919                if request.windows(4).any(|window| window == b"\r\n\r\n") {
920                    break;
921                }
922            }
923            request_tx
924                .send(String::from_utf8(request).unwrap())
925                .unwrap();
926            stream
927                .write_all(
928                    b"HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n",
929                )
930                .await
931                .unwrap();
932        });
933
934        (addr, request_rx)
935    }
936
937    #[tokio::test]
938    async fn test_http_client_awaits_multiple_rate_limiters() {
939        let quota = Quota::per_minute(NonZeroU32::MIN);
940        let request_key = Ustr::from("scope:request");
941        let order_key = Ustr::from("scope:order");
942        let request_limiter = Arc::new(RateLimiter::new_with_quota(
943            None,
944            vec![(request_key, quota)],
945        ));
946        let order_limiter = Arc::new(RateLimiter::new_with_quota(None, vec![(order_key, quota)]));
947        let client = HttpClient::builder()
948            .rate_limiters(vec![
949                Arc::clone(&request_limiter),
950                Arc::clone(&order_limiter),
951            ])
952            .build()
953            .unwrap();
954
955        client
956            .await_rate_limits(Some(&[request_key, order_key]))
957            .await;
958
959        assert!(request_limiter.check_key(&request_key).is_err());
960        assert!(order_limiter.check_key(&order_key).is_err());
961    }
962
963    #[cfg_attr(
964        not(all(feature = "simulation", madsim)),
965        tokio::test(start_paused = true)
966    )]
967    #[cfg_attr(all(feature = "simulation", madsim), madsim::test)]
968    async fn test_http_client_reserves_multiple_rate_limits_together() {
969        let global_key = Ustr::from("scope:global");
970        let order_key = Ustr::from("scope:order");
971        let global_limiter = Arc::new(RateLimiter::new_with_quota(
972            None,
973            vec![(
974                global_key,
975                Quota::with_period(Duration::from_secs(1)).unwrap(),
976            )],
977        ));
978        let order_limiter = Arc::new(RateLimiter::new_with_quota(
979            None,
980            vec![(
981                order_key,
982                Quota::with_period(Duration::from_secs(10)).unwrap(),
983            )],
984        ));
985        order_limiter.check_key(&order_key).unwrap();
986
987        let client = HttpClient::builder()
988            .rate_limiters(vec![
989                Arc::clone(&global_limiter),
990                Arc::clone(&order_limiter),
991            ])
992            .build()
993            .unwrap();
994
995        let request = test_task::spawn(async move {
996            client
997                .await_rate_limits(Some(&[global_key, order_key]))
998                .await;
999        });
1000        test_task::yield_now().await;
1001
1002        global_limiter.check_key(&global_key).unwrap();
1003        assert!(!request.is_finished());
1004
1005        advance_test_clock(Duration::from_millis(9_999)).await;
1006        global_limiter.until_key_ready(&global_key).await;
1007        global_limiter.until_key_ready(&global_key).await;
1008        advance_test_clock(Duration::from_millis(1)).await;
1009        test_task::yield_now().await;
1010        assert!(!request.is_finished());
1011
1012        advance_test_clock(Duration::from_millis(998)).await;
1013        test_task::yield_now().await;
1014        assert!(!request.is_finished());
1015
1016        advance_test_clock(Duration::from_millis(1)).await;
1017        request.await.unwrap();
1018
1019        assert!(global_limiter.check_key(&global_key).is_err());
1020        assert!(order_limiter.check_key(&order_key).is_err());
1021    }
1022
1023    #[cfg(all(feature = "simulation", madsim))]
1024    async fn advance_test_clock(duration: Duration) {
1025        madsim::time::advance(duration);
1026        test_task::yield_now().await;
1027    }
1028
1029    #[cfg(not(all(feature = "simulation", madsim)))]
1030    async fn advance_test_clock(duration: Duration) {
1031        tokio::time::advance(duration).await;
1032    }
1033
1034    #[tokio::test]
1035    async fn test_get() {
1036        let addr = start_test_server().await.unwrap();
1037        let url = format!("http://{addr}");
1038
1039        let client = InnerHttpClient::default();
1040        let response = client
1041            .send_request(
1042                reqwest::Method::GET,
1043                format!("{url}/get"),
1044                None,
1045                None,
1046                None,
1047                None,
1048            )
1049            .await
1050            .unwrap();
1051
1052        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1053        assert_eq!(String::from_utf8_lossy(&response.body), "hello-world!");
1054    }
1055
1056    #[tokio::test]
1057    async fn test_request_preserves_wire_semantics_and_extracts_response_headers() {
1058        let addr = start_test_server().await.unwrap();
1059        let mut default_headers = HashMap::new();
1060        default_headers.insert("x-default".to_string(), "default-a".to_string());
1061        let client = HttpClient::builder()
1062            .headers(default_headers)
1063            .header_keys(vec!["x-response-id".to_string()])
1064            .build()
1065            .unwrap();
1066        let mut params = HashMap::new();
1067        params.insert(
1068            "tag".to_string(),
1069            vec!["A B".to_string(), "C/D".to_string()],
1070        );
1071        let mut request_headers = HashMap::new();
1072        request_headers.insert("x-request".to_string(), "request-b".to_string());
1073
1074        let response = client
1075            .request(
1076                Method::PUT,
1077                format!("http://{addr}/capture?existing=seed"),
1078                Some(&params),
1079                Some(request_headers),
1080                Some(b"payload-c".to_vec()),
1081                None,
1082                None,
1083            )
1084            .await
1085            .unwrap();
1086
1087        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1088        assert_eq!(
1089            response.headers,
1090            HashMap::from([("x-response-id".to_string(), "response-42".to_string())])
1091        );
1092        assert_eq!(
1093            response.body.as_ref(),
1094            b"PUT\n/capture\nexisting=seed&tag=A+B&tag=C%2FD\ndefault-a\nrequest-b\npayload-c"
1095        );
1096    }
1097
1098    #[tokio::test]
1099    async fn test_request_with_params_serializes_query_fields() {
1100        #[derive(serde::Serialize)]
1101        struct Query<'a> {
1102            symbol: &'a str,
1103            limit: u32,
1104        }
1105
1106        let addr = start_test_server().await.unwrap();
1107        let mut default_headers = HashMap::new();
1108        default_headers.insert("x-default".to_string(), "default-d".to_string());
1109        let client = HttpClient::builder()
1110            .headers(default_headers)
1111            .header_keys(vec!["x-response-id".to_string()])
1112            .build()
1113            .unwrap();
1114        let mut request_headers = HashMap::new();
1115        request_headers.insert("x-request".to_string(), "request-e".to_string());
1116        let params = Query {
1117            symbol: "BTC/USDT",
1118            limit: 37,
1119        };
1120
1121        let response = client
1122            .request_with_params(
1123                Method::GET,
1124                format!("http://{addr}/capture"),
1125                Some(&params),
1126                Some(request_headers),
1127                None,
1128                None,
1129                None,
1130            )
1131            .await
1132            .unwrap();
1133
1134        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1135        assert_eq!(
1136            response.headers,
1137            HashMap::from([("x-response-id".to_string(), "response-42".to_string())])
1138        );
1139        assert_eq!(
1140            response.body.as_ref(),
1141            b"GET\n/capture\nsymbol=BTC%2FUSDT&limit=37\ndefault-d\nrequest-e\n"
1142        );
1143    }
1144
1145    #[tokio::test]
1146    async fn test_response_body_within_cap_is_returned() {
1147        let addr = start_test_server().await.unwrap();
1148        let url = format!("http://{addr}");
1149
1150        // Cap above the 1 MiB payload: body should be returned intact.
1151        let client = InnerHttpClient {
1152            max_response_bytes: 4 * 1024 * 1024,
1153            ..Default::default()
1154        };
1155
1156        let response = client
1157            .send_request(
1158                reqwest::Method::GET,
1159                format!("{url}/large"),
1160                None,
1161                None,
1162                None,
1163                None,
1164            )
1165            .await
1166            .unwrap();
1167
1168        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1169        assert_eq!(response.body.len(), 1024 * 1024);
1170    }
1171
1172    #[tokio::test]
1173    async fn test_response_body_exceeding_cap_is_rejected() {
1174        let addr = start_test_server().await.unwrap();
1175        let url = format!("http://{addr}");
1176
1177        // Cap below the 1 MiB payload: the request must fail rather than buffer it.
1178        let client = InnerHttpClient {
1179            max_response_bytes: 16 * 1024,
1180            ..Default::default()
1181        };
1182
1183        let result = client
1184            .send_request(
1185                reqwest::Method::GET,
1186                format!("{url}/large"),
1187                None,
1188                None,
1189                None,
1190                None,
1191            )
1192            .await;
1193
1194        let err = result.expect_err("oversized response body should be rejected");
1195        assert!(
1196            err.to_string().contains("exceeds maximum"),
1197            "unexpected error: {err}",
1198        );
1199    }
1200
1201    #[tokio::test]
1202    async fn test_chunked_response_body_exceeding_cap_is_rejected() {
1203        let (addr, server_task) = spawn_chunked_response_server().await;
1204        let max_response_bytes = 8;
1205        let client = InnerHttpClient {
1206            max_response_bytes,
1207            ..Default::default()
1208        };
1209
1210        let error = client
1211            .send_request(
1212                reqwest::Method::GET,
1213                format!("http://{addr}"),
1214                None,
1215                None,
1216                None,
1217                None,
1218            )
1219            .await
1220            .expect_err("chunked response body should be rejected");
1221        server_task.await.unwrap();
1222
1223        let HttpClientError::Error(message) = error else {
1224            panic!("expected HTTP error, was {error:?}");
1225        };
1226        assert_eq!(
1227            message,
1228            format!("HTTP response body exceeds maximum of {max_response_bytes} bytes")
1229        );
1230    }
1231
1232    #[tokio::test]
1233    async fn test_post() {
1234        let addr = start_test_server().await.unwrap();
1235        let url = format!("http://{addr}");
1236
1237        let client = InnerHttpClient::default();
1238        let response = client
1239            .send_request(
1240                reqwest::Method::POST,
1241                format!("{url}/post"),
1242                None,
1243                None,
1244                None,
1245                None,
1246            )
1247            .await
1248            .unwrap();
1249
1250        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1251    }
1252
1253    #[tokio::test]
1254    async fn test_post_with_body() {
1255        let addr = start_test_server().await.unwrap();
1256        let url = format!("http://{addr}");
1257
1258        let client = InnerHttpClient::default();
1259
1260        let mut body = HashMap::new();
1261        body.insert(
1262            "key1".to_string(),
1263            serde_json::Value::String("value1".to_string()),
1264        );
1265        body.insert(
1266            "key2".to_string(),
1267            serde_json::Value::String("value2".to_string()),
1268        );
1269
1270        let body_string = serde_json::to_string(&body).unwrap();
1271        let body_bytes = body_string.into_bytes();
1272
1273        let response = client
1274            .send_request(
1275                reqwest::Method::POST,
1276                format!("{url}/post"),
1277                None,
1278                None,
1279                Some(body_bytes),
1280                None,
1281            )
1282            .await
1283            .unwrap();
1284
1285        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1286    }
1287
1288    #[tokio::test]
1289    async fn test_patch() {
1290        let addr = start_test_server().await.unwrap();
1291        let url = format!("http://{addr}");
1292
1293        let client = InnerHttpClient::default();
1294        let response = client
1295            .send_request(
1296                reqwest::Method::PATCH,
1297                format!("{url}/patch"),
1298                None,
1299                None,
1300                None,
1301                None,
1302            )
1303            .await
1304            .unwrap();
1305
1306        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1307    }
1308
1309    #[tokio::test]
1310    async fn test_delete() {
1311        let addr = start_test_server().await.unwrap();
1312        let url = format!("http://{addr}");
1313
1314        let client = InnerHttpClient::default();
1315        let response = client
1316            .send_request(
1317                reqwest::Method::DELETE,
1318                format!("{url}/delete"),
1319                None,
1320                None,
1321                None,
1322                None,
1323            )
1324            .await
1325            .unwrap();
1326
1327        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1328    }
1329
1330    #[tokio::test]
1331    async fn test_not_found() {
1332        let addr = start_test_server().await.unwrap();
1333        let url = format!("http://{addr}/notfound");
1334        let client = InnerHttpClient::default();
1335
1336        let response = client
1337            .send_request(reqwest::Method::GET, url, None, None, None, None)
1338            .await
1339            .unwrap();
1340
1341        assert!(response.status.is_client_error());
1342        assert_eq!(response.status.as_u16(), 404);
1343    }
1344
1345    #[tokio::test]
1346    async fn test_timeout() {
1347        let addr = start_test_server().await.unwrap();
1348        let url = format!("http://{addr}/slow");
1349        let client = InnerHttpClient::default();
1350
1351        // We'll set a 1-second timeout for a route that sleeps 2 seconds
1352        let result = client
1353            .send_request(reqwest::Method::GET, url, None, None, None, Some(1))
1354            .await;
1355
1356        assert!(
1357            matches!(&result, Err(HttpClientError::TimeoutError(_))),
1358            "Expected a timeout error, was: {result:?}"
1359        );
1360    }
1361
1362    #[rstest]
1363    fn test_http_client_without_proxy() {
1364        // Create client with no proxy
1365        let result = HttpClient::builder().build();
1366
1367        assert!(result.is_ok());
1368    }
1369
1370    #[rstest]
1371    fn test_http_client_builder_preserves_empty_rate_limiters() {
1372        let client = HttpClient::builder()
1373            .rate_limiters(Vec::new())
1374            .build()
1375            .unwrap();
1376
1377        assert!(client.rate_limiters.is_empty());
1378    }
1379
1380    #[rstest]
1381    fn test_http_client_builder_rejects_shared_rate_limiters_with_quotas() {
1382        let quota = Quota::with_period(Duration::from_secs(1)).unwrap();
1383        let rate_limiter = Arc::new(RateLimiter::new_with_quota(None, Vec::new()));
1384        let result = HttpClient::builder()
1385            .default_quota(quota)
1386            .rate_limiters(vec![rate_limiter])
1387            .build();
1388
1389        assert_eq!(
1390            result.unwrap_err().to_string(),
1391            "HTTP error occurred: Cannot combine shared rate limiters with quota configuration"
1392        );
1393    }
1394
1395    #[tokio::test]
1396    async fn test_http_client_without_proxy_requests_directly() {
1397        let addr = start_test_server().await.unwrap();
1398        let client = HttpClient::builder().timeout_secs(2).build().unwrap();
1399        let response = client
1400            .request(
1401                Method::GET,
1402                format!("http://{addr}/get"),
1403                None,
1404                None,
1405                None,
1406                None,
1407                None,
1408            )
1409            .await
1410            .expect("direct request");
1411
1412        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1413        assert_eq!(response.body.as_ref(), b"hello-world!");
1414    }
1415
1416    #[tokio::test]
1417    async fn test_http_client_redirect_policy() {
1418        let addr = start_test_server().await.unwrap();
1419        let follow = HttpClient::builder().timeout_secs(2).build().unwrap();
1420        let reject = HttpClient::builder()
1421            .timeout_secs(2)
1422            .redirect_policy(HttpRedirectPolicy::Reject)
1423            .build()
1424            .unwrap();
1425
1426        let followed = follow
1427            .request(
1428                Method::GET,
1429                format!("http://{addr}/redirect"),
1430                None,
1431                None,
1432                None,
1433                None,
1434                None,
1435            )
1436            .await
1437            .unwrap();
1438        let rejected = reject
1439            .request(
1440                Method::GET,
1441                format!("http://{addr}/redirect"),
1442                None,
1443                None,
1444                None,
1445                None,
1446                None,
1447            )
1448            .await
1449            .unwrap();
1450
1451        assert_eq!(followed.status.as_u16(), StatusCode::OK.as_u16());
1452        assert_eq!(followed.body.as_ref(), b"hello-world!");
1453        assert_eq!(
1454            rejected.status.as_u16(),
1455            StatusCode::TEMPORARY_REDIRECT.as_u16()
1456        );
1457        assert!(rejected.body.is_empty());
1458    }
1459
1460    #[tokio::test]
1461    async fn test_http_client_redacted_url_request_preserves_response() {
1462        let addr = start_test_server().await.unwrap();
1463        let client = HttpClient::builder().timeout_secs(2).build().unwrap();
1464        let response = client
1465            .request_with_url_redacted(
1466                Method::GET,
1467                format!("http://{addr}/get"),
1468                None,
1469                None,
1470                None,
1471                None,
1472                None,
1473            )
1474            .await
1475            .expect("direct request with URL redaction");
1476
1477        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1478        assert_eq!(response.body.as_ref(), b"hello-world!");
1479    }
1480
1481    #[tokio::test]
1482    async fn test_http_client_redacted_url_request_removes_endpoint_from_error() {
1483        const USERINFO_SECRET: &str = "transport-userinfo-secret";
1484        const PATH_SECRET: &str = "transport-path-secret";
1485        const QUERY_SECRET: &str = "transport-query-secret";
1486        let (addr, drop_task) = spawn_connection_dropper().await;
1487        let url = format!(
1488            "http://rpc-user:{USERINFO_SECRET}@{addr}/{PATH_SECRET}?api_key={QUERY_SECRET}"
1489        );
1490        let client = HttpClient::builder().timeout_secs(1).build().unwrap();
1491
1492        let error = client
1493            .request_with_url_redacted(Method::GET, url.clone(), None, None, None, None, None)
1494            .await
1495            .expect_err("an unreachable endpoint should fail");
1496        drop_task.abort();
1497        let task_error = drop_task
1498            .await
1499            .expect_err("connection dropper should be cancelled");
1500
1501        assert!(task_error.is_cancelled());
1502        for rendered in [error.to_string(), format!("{error:?}")] {
1503            assert!(!rendered.contains(USERINFO_SECRET));
1504            assert!(!rendered.contains(PATH_SECRET));
1505            assert!(!rendered.contains(QUERY_SECRET));
1506            assert!(!rendered.contains(&url));
1507        }
1508    }
1509
1510    #[tokio::test]
1511    async fn test_http_client_redacted_url_request_removes_endpoint_from_trace_logs() {
1512        const USERINFO_SECRET: &str = "trace-userinfo-secret";
1513        const PATH_SECRET: &str = "trace-path-secret";
1514        const QUERY_SECRET: &str = "trace-query-secret";
1515        let capture = capture_logs().await;
1516        let addr = start_test_server().await.unwrap();
1517        let url = format!(
1518            "http://rpc-user:{USERINFO_SECRET}@{addr}/{PATH_SECRET}?api_key={QUERY_SECRET}"
1519        );
1520        let client = HttpClient::builder().timeout_secs(2).build().unwrap();
1521
1522        let response = client
1523            .request_with_url_redacted(Method::GET, url.clone(), None, None, None, None, None)
1524            .await
1525            .expect("credentialized endpoint should return an HTTP response");
1526        let messages = capture.messages();
1527
1528        assert_eq!(response.status.as_u16(), StatusCode::NOT_FOUND.as_u16());
1529        assert!(messages.iter().any(|(level, message)| {
1530            *level == Level::Trace && message.starts_with("Sending HTTP request: method=GET")
1531        }));
1532        assert!(messages.iter().any(|(level, message)| {
1533            *level == Level::Trace
1534                && message.starts_with("Received HTTP response: status=404 Not Found")
1535        }));
1536
1537        for (_, message) in messages {
1538            assert!(!message.contains(USERINFO_SECRET));
1539            assert!(!message.contains(PATH_SECRET));
1540            assert!(!message.contains(QUERY_SECRET));
1541            assert!(!message.contains(&url));
1542        }
1543    }
1544
1545    #[tokio::test]
1546    async fn test_http_client_uses_connect_and_proxy_authorization_for_https() {
1547        const USERNAME: &str = "proxytest";
1548        const PASSWORD: &str = "fixture42";
1549        let (proxy_addr, request_rx) = spawn_rejecting_connect_proxy().await;
1550        let client = HttpClient::builder()
1551            .timeout_secs(2)
1552            .proxy_url(format!("http://{USERNAME}:{PASSWORD}@{proxy_addr}"))
1553            .build()
1554            .unwrap();
1555        let error = client
1556            .request(
1557                Method::GET,
1558                "https://fixture.example.test/path".to_string(),
1559                None,
1560                None,
1561                None,
1562                None,
1563                None,
1564            )
1565            .await
1566            .expect_err("proxy should reject CONNECT");
1567        let request = request_rx.await.expect("captured CONNECT request");
1568        let mut lines = request.split("\r\n");
1569        let request_line = lines.next().expect("CONNECT request line");
1570        let auth_value = lines
1571            .find_map(|line| {
1572                let (name, value) = line.split_once(':')?;
1573                name.eq_ignore_ascii_case("proxy-authorization")
1574                    .then_some(value.trim())
1575            })
1576            .expect("Proxy-Authorization header");
1577        let expected_auth = format!("Basic {}", BASE64.encode(format!("{USERNAME}:{PASSWORD}")));
1578
1579        assert_eq!(request_line, "CONNECT fixture.example.test:443 HTTP/1.1");
1580        assert_eq!(auth_value, expected_auth);
1581        assert!(!error.to_string().contains(PASSWORD));
1582        assert!(!error.to_string().contains(&BASE64.encode(PASSWORD)));
1583        assert!(!error.to_string().contains(&expected_auth));
1584    }
1585
1586    #[tokio::test]
1587    async fn test_http_client_unreachable_proxy_error_redacts_credentials() {
1588        const USERNAME: &str = "proxy-user";
1589        const SECRET: &str = "unreachable-proxy-secret";
1590        let (proxy_addr, drop_task) = spawn_connection_dropper().await;
1591        let client = HttpClient::builder()
1592            .timeout_secs(1)
1593            .proxy_url(format!("http://{USERNAME}:{SECRET}@{proxy_addr}"))
1594            .build()
1595            .unwrap();
1596        let error = client
1597            .request(
1598                Method::GET,
1599                "https://fixture.example.test/".to_string(),
1600                None,
1601                None,
1602                None,
1603                None,
1604                None,
1605            )
1606            .await
1607            .expect_err("unreachable proxy should fail");
1608        drop_task.abort();
1609        let task_error = drop_task
1610            .await
1611            .expect_err("connection dropper should be cancelled");
1612
1613        assert!(task_error.is_cancelled());
1614        assert!(!error.to_string().contains(SECRET));
1615        assert!(!error.to_string().contains(&BASE64.encode(SECRET)));
1616        assert!(
1617            !error
1618                .to_string()
1619                .contains(&BASE64.encode(format!("{USERNAME}:{SECRET}")))
1620        );
1621    }
1622
1623    #[rstest]
1624    fn test_http_client_with_valid_proxy() {
1625        // Create client with a valid proxy URL
1626        let result = HttpClient::builder()
1627            .proxy_url("http://proxy.example.com:8080".to_string())
1628            .build();
1629
1630        assert!(result.is_ok());
1631    }
1632
1633    #[rstest]
1634    fn test_http_client_with_socks5_proxy() {
1635        // Create client with a SOCKS5 proxy URL
1636        let result = HttpClient::builder()
1637            .proxy_url("socks5://127.0.0.1:1080".to_string())
1638            .build();
1639
1640        assert!(result.is_ok());
1641    }
1642
1643    #[rstest]
1644    fn test_http_client_with_malformed_proxy() {
1645        // Note: reqwest::Proxy::all() is lenient and accepts most strings.
1646        // It only fails on obviously malformed URLs like "://invalid" or "http://".
1647        // More subtle issues (like "not-a-valid-url") are caught when connecting.
1648        let result = HttpClient::builder()
1649            .proxy_url("://invalid".to_string())
1650            .build();
1651
1652        assert!(result.is_err());
1653        assert!(matches!(result, Err(HttpClientError::InvalidProxy(_))));
1654    }
1655
1656    #[rstest]
1657    fn test_http_client_invalid_proxy_error_redacts_credentials() {
1658        const SECRET: &str = "unique-proxy-secret";
1659        let result = HttpClient::builder()
1660            .proxy_url(format!("http://proxytest:{SECRET}@[::1"))
1661            .build();
1662        let error = result.expect_err("malformed proxy URL should fail");
1663
1664        assert_eq!(
1665            error.to_string(),
1666            "Invalid proxy URL: proxy URL is malformed"
1667        );
1668        assert!(!error.to_string().contains(SECRET));
1669    }
1670
1671    #[rstest]
1672    fn test_http_client_with_empty_proxy_string() {
1673        // Create client with an empty proxy URL string
1674        let result = HttpClient::builder().proxy_url(String::new()).build();
1675
1676        assert!(result.is_err());
1677        assert!(matches!(result, Err(HttpClientError::InvalidProxy(_))));
1678    }
1679
1680    #[tokio::test]
1681    async fn test_http_client_get() {
1682        let addr = start_test_server().await.unwrap();
1683        let url = format!("http://{addr}/get");
1684
1685        let client = HttpClient::builder().build().unwrap();
1686        let response = client.get(url, None, None, None, None).await.unwrap();
1687
1688        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1689        assert_eq!(String::from_utf8_lossy(&response.body), "hello-world!");
1690    }
1691
1692    #[tokio::test]
1693    async fn test_http_client_post() {
1694        let addr = start_test_server().await.unwrap();
1695        let url = format!("http://{addr}/post");
1696
1697        let client = HttpClient::builder().build().unwrap();
1698        let response = client
1699            .post(url, None, None, None, None, None)
1700            .await
1701            .unwrap();
1702
1703        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1704    }
1705
1706    #[tokio::test]
1707    async fn test_http_client_patch() {
1708        let addr = start_test_server().await.unwrap();
1709        let url = format!("http://{addr}/patch");
1710
1711        let client = HttpClient::builder().build().unwrap();
1712        let response = client
1713            .patch(url, None, None, None, None, None)
1714            .await
1715            .unwrap();
1716
1717        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1718    }
1719
1720    #[tokio::test]
1721    async fn test_http_client_delete() {
1722        let addr = start_test_server().await.unwrap();
1723        let url = format!("http://{addr}/delete");
1724
1725        let client = HttpClient::builder().build().unwrap();
1726        let response = client.delete(url, None, None, None, None).await.unwrap();
1727
1728        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
1729    }
1730}
nautilus_network/http/client.rs

nautilus_network/http/
client.rs
