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

nautilus_network/http/
client.rs
