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 HTTP client that supports rate limiting and timeouts.
48///
49/// Built on `reqwest` for async I/O. Allows per-endpoint and default quotas
50/// through a rate limiter.
51///
52/// This struct is designed to handle HTTP requests efficiently, providing
53/// support for rate limiting, timeouts, and custom headers. The client is
54/// built on top of `reqwest` and can be used for both synchronous and
55/// asynchronous HTTP requests.
56#[derive(Clone, Debug)]
57#[cfg_attr(
58    feature = "python",
59    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.network", from_py_object)
60)]
61#[cfg_attr(
62    feature = "python",
63    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.network")
64)]
65pub struct HttpClient {
66    /// The underlying HTTP client used to make requests.
67    pub(crate) client: InnerHttpClient,
68    /// The rate limiters that control the request rate.
69    pub(crate) rate_limiters: Arc<[Arc<RateLimiter<Ustr, MonotonicClock>>]>,
70}
71
72impl HttpClient {
73    /// Creates a new [`HttpClient`] instance.
74    ///
75    /// # Errors
76    ///
77    /// - Returns `InvalidProxy` if the proxy URL is malformed.
78    /// - Returns `ClientBuildError` if building the underlying `reqwest::Client` fails.
79    pub fn new(
80        headers: HashMap<String, String>,
81        header_keys: Vec<String>,
82        keyed_quotas: Vec<(String, Quota)>,
83        default_quota: Option<Quota>,
84        timeout_secs: Option<u64>,
85        proxy_url: Option<String>,
86    ) -> Result<Self, HttpClientError> {
87        let keyed_quotas = keyed_quotas
88            .into_iter()
89            .map(|(key, quota)| (Ustr::from(&key), quota))
90            .collect();
91
92        let rate_limiter = Arc::new(RateLimiter::new_with_quota(default_quota, keyed_quotas));
93
94        Self::new_with_rate_limiter(headers, header_keys, timeout_secs, proxy_url, rate_limiter)
95    }
96
97    /// Creates a new [`HttpClient`] instance sharing an externally-owned rate limiter.
98    ///
99    /// Use this constructor to share a single [`RateLimiter`] across multiple
100    /// [`HttpClient`] instances (for example, the HTTP clients owned by an
101    /// exchange adapter's data and execution clients). All quota state lives
102    /// inside the limiter, so passing the same `Arc` produces a single shared
103    /// bucket.
104    ///
105    /// # Errors
106    ///
107    /// - Returns `InvalidProxy` if the proxy URL is malformed.
108    /// - Returns `ClientBuildError` if building the underlying `reqwest::Client` fails.
109    pub fn new_with_rate_limiter(
110        headers: HashMap<String, String>,
111        header_keys: Vec<String>,
112        timeout_secs: Option<u64>,
113        proxy_url: Option<String>,
114        rate_limiter: Arc<RateLimiter<Ustr, MonotonicClock>>,
115    ) -> Result<Self, HttpClientError> {
116        Self::new_with_rate_limiters(
117            headers,
118            header_keys,
119            timeout_secs,
120            proxy_url,
121            vec![rate_limiter],
122        )
123    }
124
125    /// Creates a new [`HttpClient`] instance sharing multiple externally-owned rate limiters.
126    ///
127    /// Each request awaits every limiter with the same keys. A limiter with no default quota
128    /// ignores keys it does not own, allowing independent quota scopes such as per-IP and
129    /// per-account limits to apply to one request.
130    ///
131    /// # Errors
132    ///
133    /// - Returns `InvalidProxy` if the proxy URL is malformed.
134    /// - Returns `ClientBuildError` if building the underlying `reqwest::Client` fails.
135    pub fn new_with_rate_limiters(
136        headers: HashMap<String, String>,
137        header_keys: Vec<String>,
138        timeout_secs: Option<u64>,
139        proxy_url: Option<String>,
140        rate_limiters: Vec<Arc<RateLimiter<Ustr, MonotonicClock>>>,
141    ) -> Result<Self, HttpClientError> {
142        install_cryptographic_provider();
143
144        // Build default headers
145        let mut header_map = HeaderMap::new();
146
147        for (key, value) in headers {
148            let header_name = HeaderName::from_str(&key)
149                .map_err(|e| HttpClientError::Error(format!("Invalid header name '{key}': {e}")))?;
150            let header_value = HeaderValue::from_str(&value).map_err(|e| {
151                HttpClientError::Error(format!("Invalid header value '{value}': {e}"))
152            })?;
153            header_map.insert(header_name, header_value);
154        }
155
156        let mut client_builder = reqwest::Client::builder()
157            .default_headers(header_map)
158            .tcp_nodelay(true)
159            .pool_max_idle_per_host(DEFAULT_POOL_MAX_IDLE_PER_HOST)
160            .pool_idle_timeout(Duration::from_secs(DEFAULT_POOL_IDLE_TIMEOUT_SECS))
161            .http2_keep_alive_interval(Duration::from_secs(DEFAULT_HTTP2_KEEP_ALIVE_SECS))
162            .http2_keep_alive_while_idle(true)
163            .http2_adaptive_window(true);
164
165        if let Some(timeout_secs) = timeout_secs {
166            client_builder = client_builder.timeout(Duration::from_secs(timeout_secs));
167        }
168
169        // Configure proxy if provided
170        if let Some(proxy_url) = proxy_url {
171            let proxy = reqwest::Proxy::all(&proxy_url)
172                .map_err(|_| HttpClientError::InvalidProxy("proxy URL is malformed".to_string()))?;
173            client_builder = client_builder.proxy(proxy);
174        }
175
176        let client = client_builder
177            .build()
178            .map_err(|e| HttpClientError::ClientBuildError(e.to_string()))?;
179
180        // Pre-intern header keys as HeaderName, keeping both vectors aligned,
181        // an invalid key is an error: a silent drop would make response extraction read nothing.
182        let (valid_keys, header_names): (Vec<String>, Vec<HeaderName>) = header_keys
183            .into_iter()
184            .map(|k| {
185                HeaderName::from_str(&k)
186                    .map(|name| (k.clone(), name))
187                    .map_err(|e| HttpClientError::Error(format!("Invalid header key '{k}': {e}")))
188            })
189            .collect::<Result<Vec<_>, _>>()?
190            .into_iter()
191            .unzip();
192
193        let client = InnerHttpClient {
194            client,
195            header_keys: Arc::from(valid_keys),
196            header_names: Arc::from(header_names),
197            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
198        };
199
200        Ok(Self {
201            client,
202            rate_limiters: rate_limiters.into(),
203        })
204    }
205
206    /// Sends an HTTP request.
207    ///
208    /// # Errors
209    ///
210    /// Returns an error if unable to send request or times out.
211    ///
212    /// # Examples
213    ///
214    /// If requesting `/foo/bar`, pass rate-limit keys `["foo/bar", "foo"]`.
215    #[expect(clippy::too_many_arguments)]
216    pub async fn request(
217        &self,
218        method: Method,
219        url: String,
220        params: Option<&HashMap<String, Vec<String>>>,
221        headers: Option<HashMap<String, String>>,
222        body: Option<Vec<u8>>,
223        timeout_secs: Option<u64>,
224        keys: Option<Vec<String>>,
225    ) -> Result<HttpResponse, HttpClientError> {
226        let keys = keys.map(into_ustr_vec);
227
228        self.request_with_ustr_keys(method, url, params, headers, body, timeout_secs, keys)
229            .await
230    }
231
232    /// Sends an HTTP request with serializable query parameters.
233    ///
234    /// This method accepts any type implementing `Serialize` for query parameters,
235    /// which will be automatically encoded into the URL query string using reqwest's
236    /// `.query()` method, avoiding unnecessary `HashMap` allocations.
237    ///
238    /// # Errors
239    ///
240    /// Returns an error if unable to send request or times out.
241    #[expect(clippy::too_many_arguments)]
242    pub async fn request_with_params<P: serde::Serialize>(
243        &self,
244        method: Method,
245        url: String,
246        params: Option<&P>,
247        headers: Option<HashMap<String, String>>,
248        body: Option<Vec<u8>>,
249        timeout_secs: Option<u64>,
250        keys: Option<Vec<String>>,
251    ) -> Result<HttpResponse, HttpClientError> {
252        let keys = keys.map(into_ustr_vec);
253        self.await_rate_limits(keys.as_deref()).await;
254
255        self.client
256            .send_request_with_query(method, url, params, headers, body, timeout_secs)
257            .await
258    }
259
260    /// Sends an HTTP request using pre-interned rate limiter keys.
261    ///
262    /// # Errors
263    ///
264    /// Returns an error if unable to send the request or the request times out.
265    #[expect(clippy::too_many_arguments)]
266    pub async fn request_with_ustr_keys(
267        &self,
268        method: Method,
269        url: String,
270        params: Option<&HashMap<String, Vec<String>>>,
271        headers: Option<HashMap<String, String>>,
272        body: Option<Vec<u8>>,
273        timeout_secs: Option<u64>,
274        keys: Option<Vec<Ustr>>,
275    ) -> Result<HttpResponse, HttpClientError> {
276        self.await_rate_limits(keys.as_deref()).await;
277
278        self.client
279            .send_request(method, url, params, headers, body, timeout_secs)
280            .await
281    }
282
283    pub(crate) async fn await_rate_limits(&self, keys: Option<&[Ustr]>) {
284        for rate_limiter in self.rate_limiters.iter() {
285            rate_limiter.await_keys_ready(keys).await;
286        }
287    }
288
289    /// Sends an HTTP GET request.
290    ///
291    /// # Errors
292    ///
293    /// Returns an error if unable to send request or times out.
294    pub async fn get(
295        &self,
296        url: String,
297        params: Option<&HashMap<String, Vec<String>>>,
298        headers: Option<HashMap<String, String>>,
299        timeout_secs: Option<u64>,
300        keys: Option<Vec<String>>,
301    ) -> Result<HttpResponse, HttpClientError> {
302        self.request(Method::GET, url, params, headers, None, timeout_secs, keys)
303            .await
304    }
305
306    /// Sends an HTTP POST request.
307    ///
308    /// # Errors
309    ///
310    /// Returns an error if unable to send request or times out.
311    pub async fn post(
312        &self,
313        url: String,
314        params: Option<&HashMap<String, Vec<String>>>,
315        headers: Option<HashMap<String, String>>,
316        body: Option<Vec<u8>>,
317        timeout_secs: Option<u64>,
318        keys: Option<Vec<String>>,
319    ) -> Result<HttpResponse, HttpClientError> {
320        self.request(Method::POST, url, params, headers, body, timeout_secs, keys)
321            .await
322    }
323
324    /// Sends an HTTP PATCH request.
325    ///
326    /// # Errors
327    ///
328    /// Returns an error if unable to send request or times out.
329    pub async fn patch(
330        &self,
331        url: String,
332        params: Option<&HashMap<String, Vec<String>>>,
333        headers: Option<HashMap<String, String>>,
334        body: Option<Vec<u8>>,
335        timeout_secs: Option<u64>,
336        keys: Option<Vec<String>>,
337    ) -> Result<HttpResponse, HttpClientError> {
338        self.request(
339            Method::PATCH,
340            url,
341            params,
342            headers,
343            body,
344            timeout_secs,
345            keys,
346        )
347        .await
348    }
349
350    /// Sends an HTTP DELETE request.
351    ///
352    /// # Errors
353    ///
354    /// Returns an error if unable to send request or times out.
355    pub async fn delete(
356        &self,
357        url: String,
358        params: Option<&HashMap<String, Vec<String>>>,
359        headers: Option<HashMap<String, String>>,
360        timeout_secs: Option<u64>,
361        keys: Option<Vec<String>>,
362    ) -> Result<HttpResponse, HttpClientError> {
363        self.request(
364            Method::DELETE,
365            url,
366            params,
367            headers,
368            None,
369            timeout_secs,
370            keys,
371        )
372        .await
373    }
374}
375
376/// Internal implementation backing [`HttpClient`].
377///
378/// The client is backed by a [`reqwest::Client`] which keeps connections alive and
379/// can be cloned cheaply. The client also has a list of header fields to
380/// extract from the response.
381///
382/// The client returns an [`HttpResponse`]. The client filters only the key value
383/// for the give `header_keys`.
384#[derive(Clone, Debug)]
385pub struct InnerHttpClient {
386    pub(crate) client: reqwest::Client,
387    pub(crate) header_keys: Arc<[String]>,
388    pub(crate) header_names: Arc<[HeaderName]>,
389    /// Maximum response body size in bytes; bodies exceeding this are rejected.
390    pub(crate) max_response_bytes: usize,
391}
392
393impl InnerHttpClient {
394    /// Sends an HTTP request and returns an [`HttpResponse`].
395    ///
396    /// # Errors
397    ///
398    /// Returns an error if unable to send request or times out.
399    pub async fn send_request(
400        &self,
401        method: Method,
402        url: String,
403        params: Option<&HashMap<String, Vec<String>>>,
404        headers: Option<HashMap<String, String>>,
405        body: Option<Vec<u8>>,
406        timeout_secs: Option<u64>,
407    ) -> Result<HttpResponse, HttpClientError> {
408        let full_url = encode_url_params(&url, params)?;
409        self.send_request_internal(
410            method,
411            full_url.as_ref(),
412            None::<&()>,
413            headers,
414            body,
415            timeout_secs,
416        )
417        .await
418    }
419
420    /// Sends an HTTP request with query parameters using reqwest's `.query()` method.
421    ///
422    /// This method accepts any type implementing `Serialize` for query parameters,
423    /// avoiding `HashMap` conversion overhead.
424    ///
425    /// # Errors
426    ///
427    /// Returns an error if unable to send request or times out.
428    pub async fn send_request_with_query<Q: serde::Serialize>(
429        &self,
430        method: Method,
431        url: String,
432        query: Option<&Q>,
433        headers: Option<HashMap<String, String>>,
434        body: Option<Vec<u8>>,
435        timeout_secs: Option<u64>,
436    ) -> Result<HttpResponse, HttpClientError> {
437        self.send_request_internal(method, &url, query, headers, body, timeout_secs)
438            .await
439    }
440
441    /// Internal implementation for sending HTTP requests.
442    ///
443    /// # Errors
444    ///
445    /// Returns an error if unable to send request or times out.
446    async fn send_request_internal<Q: serde::Serialize>(
447        &self,
448        method: Method,
449        url: &str,
450        query: Option<&Q>,
451        headers: Option<HashMap<String, String>>,
452        body: Option<Vec<u8>>,
453        timeout_secs: Option<u64>,
454    ) -> Result<HttpResponse, HttpClientError> {
455        let reqwest_url =
456            Url::parse(url).map_err(|e| HttpClientError::from(format!("URL parse error: {e}")))?;
457
458        let mut request_builder = self.client.request(method, reqwest_url);
459
460        if let Some(headers) = headers {
461            let mut header_map = HeaderMap::with_capacity(headers.len());
462            for (header_key, header_value) in &headers {
463                let key = HeaderName::from_bytes(header_key.as_bytes())
464                    .map_err(|e| HttpClientError::from(format!("Invalid header name: {e}")))?;
465
466                if let Some(old_value) = header_map.insert(
467                    key.clone(),
468                    header_value
469                        .parse()
470                        .map_err(|e| HttpClientError::from(format!("Invalid header value: {e}")))?,
471                ) {
472                    log::trace!("Replaced header '{key}': old={old_value:?}, new={header_value}");
473                }
474            }
475            request_builder = request_builder.headers(header_map);
476        }
477
478        if let Some(q) = query {
479            request_builder = request_builder.query(q);
480        }
481
482        if let Some(timeout_secs) = timeout_secs {
483            request_builder = request_builder.timeout(Duration::new(timeout_secs, 0));
484        }
485
486        let request = match body {
487            Some(b) => request_builder
488                .body(b)
489                .build()
490                .map_err(HttpClientError::from)?,
491            None => request_builder.build().map_err(HttpClientError::from)?,
492        };
493
494        log::trace!("{} {}", request.method(), request.url());
495
496        let response = self
497            .client
498            .execute(request)
499            .await
500            .map_err(HttpClientError::from)?;
501
502        self.to_response(response).await
503    }
504
505    /// Converts a `reqwest::Response` into an `HttpResponse`.
506    ///
507    /// Uses pre-interned `HeaderName` values to avoid string-to-header parsing per response.
508    ///
509    /// # Errors
510    ///
511    /// Returns an error if unable to send request or times out.
512    pub async fn to_response(&self, response: Response) -> Result<HttpResponse, HttpClientError> {
513        log::trace!("{response:?}");
514
515        let resp_headers = response.headers();
516        let mut headers =
517            HashMap::with_capacity(std::cmp::min(self.header_names.len(), resp_headers.len()));
518
519        for (name, key_str) in self.header_names.iter().zip(self.header_keys.iter()) {
520            if let Some(val) = resp_headers.get(name)
521                && let Ok(v) = val.to_str()
522            {
523                headers.insert(key_str.clone(), v.to_owned());
524            }
525        }
526
527        let status = HttpStatus::new(response.status());
528        let body = self.read_body_capped(response).await?;
529
530        Ok(HttpResponse {
531            status,
532            headers,
533            body,
534        })
535    }
536
537    /// Reads the response body, rejecting any body that exceeds `max_response_bytes`.
538    ///
539    /// A `Content-Length` larger than the cap is rejected up front; otherwise the
540    /// body is streamed chunk-by-chunk and aborted as soon as the accumulated size
541    /// would exceed the cap, so an oversized or unbounded (chunked) body is never
542    /// fully buffered into memory.
543    ///
544    /// # Errors
545    ///
546    /// Returns an error if the body exceeds the configured maximum size, or if
547    /// reading a chunk fails.
548    async fn read_body_capped(
549        &self,
550        mut response: Response,
551    ) -> Result<bytes::Bytes, HttpClientError> {
552        let max = self.max_response_bytes;
553
554        // Fast path: reject up front when the advertised length already exceeds the cap.
555        if let Some(len) = response.content_length()
556            && len > max as u64
557        {
558            return Err(HttpClientError::Error(format!(
559                "HTTP response body of {len} bytes exceeds maximum of {max} bytes",
560            )));
561        }
562
563        let mut buf = bytes::BytesMut::new();
564        while let Some(chunk) = response.chunk().await.map_err(HttpClientError::from)? {
565            if buf.len() + chunk.len() > max {
566                return Err(HttpClientError::Error(format!(
567                    "HTTP response body exceeds maximum of {max} bytes",
568                )));
569            }
570            buf.extend_from_slice(&chunk);
571        }
572
573        Ok(buf.freeze())
574    }
575}
576
577impl Default for InnerHttpClient {
578    /// Creates a new default [`InnerHttpClient`] instance.
579    ///
580    /// The default client is initialized with an empty list of header keys and a new `reqwest::Client`.
581    fn default() -> Self {
582        install_cryptographic_provider();
583        let client = reqwest::Client::new();
584        Self {
585            client,
586            header_keys: Arc::default(),
587            header_names: Arc::default(),
588            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
589        }
590    }
591}
592
593/// Encodes URL parameters into the query string.
594///
595/// Returns `Cow::Borrowed` when no parameters need appending (zero-alloc fast path).
596/// Parameters can have multiple values per key (for doseq=True behavior).
597/// Preserves existing query strings in the URL by appending with '&' instead of '?'.
598fn encode_url_params<'a>(
599    url: &'a str,
600    params: Option<&HashMap<String, Vec<String>>>,
601) -> Result<Cow<'a, str>, HttpClientError> {
602    let Some(params) = params else {
603        return Ok(Cow::Borrowed(url));
604    };
605
606    let pairs: Vec<(&str, &str)> = params
607        .iter()
608        .flat_map(|(key, values)| {
609            values
610                .iter()
611                .map(move |value| (key.as_str(), value.as_str()))
612        })
613        .collect();
614
615    if pairs.is_empty() {
616        return Ok(Cow::Borrowed(url));
617    }
618
619    let query_string = serde_urlencoded::to_string(pairs)
620        .map_err(|e| HttpClientError::Error(format!("Failed to encode params: {e}")))?;
621
622    let separator = if url.contains('?') { '&' } else { '?' };
623    Ok(Cow::Owned(format!("{url}{separator}{query_string}")))
624}
625
626#[cfg(test)]
627#[cfg(target_os = "linux")] // Only run network tests on Linux (CI stability)
628mod tests {
629    use std::{net::SocketAddr, num::NonZeroU32};
630
631    use axum::{
632        Router,
633        routing::{delete, get, patch, post},
634        serve,
635    };
636    use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
637    use http::status::StatusCode;
638    use rstest::rstest;
639    use tokio::{
640        io::{AsyncReadExt, AsyncWriteExt},
641        sync::oneshot,
642    };
643
644    use super::*;
645
646    fn create_router() -> Router {
647        Router::new()
648            .route("/get", get(|| async { "hello-world!" }))
649            .route("/post", post(|| async { StatusCode::OK }))
650            .route("/patch", patch(|| async { StatusCode::OK }))
651            .route("/delete", delete(|| async { StatusCode::OK }))
652            .route("/notfound", get(|| async { StatusCode::NOT_FOUND }))
653            .route(
654                "/slow",
655                get(|| async {
656                    tokio::time::sleep(Duration::from_secs(2)).await;
657                    "Eventually responded"
658                }),
659            )
660            .route(
661                "/large",
662                // Returns a 1 MiB body to exercise the response size cap.
663                get(|| async { "x".repeat(1024 * 1024) }),
664            )
665    }
666
667    async fn start_test_server() -> Result<SocketAddr, Box<dyn std::error::Error + Send + Sync>> {
668        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
669        let addr = listener.local_addr().unwrap();
670
671        tokio::spawn(async move {
672            serve(listener, create_router()).await.unwrap();
673        });
674
675        Ok(addr)
676    }
677
678    async fn spawn_rejecting_connect_proxy() -> (SocketAddr, oneshot::Receiver<String>) {
679        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
680        let addr = listener.local_addr().unwrap();
681        let (request_tx, request_rx) = oneshot::channel();
682
683        tokio::spawn(async move {
684            let (mut stream, _) = listener.accept().await.unwrap();
685            let mut request = Vec::new();
686            let mut chunk = [0u8; 1024];
687            loop {
688                let read = stream.read(&mut chunk).await.unwrap();
689                if read == 0 {
690                    break;
691                }
692                request.extend_from_slice(&chunk[..read]);
693                if request.windows(4).any(|window| window == b"\r\n\r\n") {
694                    break;
695                }
696            }
697            request_tx
698                .send(String::from_utf8(request).unwrap())
699                .unwrap();
700            stream
701                .write_all(
702                    b"HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n",
703                )
704                .await
705                .unwrap();
706        });
707
708        (addr, request_rx)
709    }
710
711    #[tokio::test]
712    async fn test_http_client_awaits_multiple_rate_limiters() {
713        let quota = Quota::per_minute(NonZeroU32::MIN);
714        let request_key = Ustr::from("scope:request");
715        let order_key = Ustr::from("scope:order");
716        let request_limiter = Arc::new(RateLimiter::new_with_quota(
717            None,
718            vec![(request_key, quota)],
719        ));
720        let order_limiter = Arc::new(RateLimiter::new_with_quota(None, vec![(order_key, quota)]));
721        let client = HttpClient::new_with_rate_limiters(
722            HashMap::new(),
723            Vec::new(),
724            None,
725            None,
726            vec![Arc::clone(&request_limiter), Arc::clone(&order_limiter)],
727        )
728        .unwrap();
729
730        client
731            .await_rate_limits(Some(&[request_key, order_key]))
732            .await;
733
734        assert!(request_limiter.check_key(&request_key).is_err());
735        assert!(order_limiter.check_key(&order_key).is_err());
736    }
737
738    #[tokio::test]
739    async fn test_get() {
740        let addr = start_test_server().await.unwrap();
741        let url = format!("http://{addr}");
742
743        let client = InnerHttpClient::default();
744        let response = client
745            .send_request(
746                reqwest::Method::GET,
747                format!("{url}/get"),
748                None,
749                None,
750                None,
751                None,
752            )
753            .await
754            .unwrap();
755
756        assert!(response.status.is_success());
757        assert_eq!(String::from_utf8_lossy(&response.body), "hello-world!");
758    }
759
760    #[tokio::test]
761    async fn test_response_body_within_cap_is_returned() {
762        let addr = start_test_server().await.unwrap();
763        let url = format!("http://{addr}");
764
765        // Cap above the 1 MiB payload: body should be returned intact.
766        let client = InnerHttpClient {
767            max_response_bytes: 4 * 1024 * 1024,
768            ..Default::default()
769        };
770
771        let response = client
772            .send_request(
773                reqwest::Method::GET,
774                format!("{url}/large"),
775                None,
776                None,
777                None,
778                None,
779            )
780            .await
781            .unwrap();
782
783        assert!(response.status.is_success());
784        assert_eq!(response.body.len(), 1024 * 1024);
785    }
786
787    #[tokio::test]
788    async fn test_response_body_exceeding_cap_is_rejected() {
789        let addr = start_test_server().await.unwrap();
790        let url = format!("http://{addr}");
791
792        // Cap below the 1 MiB payload: the request must fail rather than buffer it.
793        let client = InnerHttpClient {
794            max_response_bytes: 16 * 1024,
795            ..Default::default()
796        };
797
798        let result = client
799            .send_request(
800                reqwest::Method::GET,
801                format!("{url}/large"),
802                None,
803                None,
804                None,
805                None,
806            )
807            .await;
808
809        let err = result.expect_err("oversized response body should be rejected");
810        assert!(
811            err.to_string().contains("exceeds maximum"),
812            "unexpected error: {err}",
813        );
814    }
815
816    #[tokio::test]
817    async fn test_post() {
818        let addr = start_test_server().await.unwrap();
819        let url = format!("http://{addr}");
820
821        let client = InnerHttpClient::default();
822        let response = client
823            .send_request(
824                reqwest::Method::POST,
825                format!("{url}/post"),
826                None,
827                None,
828                None,
829                None,
830            )
831            .await
832            .unwrap();
833
834        assert!(response.status.is_success());
835    }
836
837    #[tokio::test]
838    async fn test_post_with_body() {
839        let addr = start_test_server().await.unwrap();
840        let url = format!("http://{addr}");
841
842        let client = InnerHttpClient::default();
843
844        let mut body = HashMap::new();
845        body.insert(
846            "key1".to_string(),
847            serde_json::Value::String("value1".to_string()),
848        );
849        body.insert(
850            "key2".to_string(),
851            serde_json::Value::String("value2".to_string()),
852        );
853
854        let body_string = serde_json::to_string(&body).unwrap();
855        let body_bytes = body_string.into_bytes();
856
857        let response = client
858            .send_request(
859                reqwest::Method::POST,
860                format!("{url}/post"),
861                None,
862                None,
863                Some(body_bytes),
864                None,
865            )
866            .await
867            .unwrap();
868
869        assert!(response.status.is_success());
870    }
871
872    #[tokio::test]
873    async fn test_patch() {
874        let addr = start_test_server().await.unwrap();
875        let url = format!("http://{addr}");
876
877        let client = InnerHttpClient::default();
878        let response = client
879            .send_request(
880                reqwest::Method::PATCH,
881                format!("{url}/patch"),
882                None,
883                None,
884                None,
885                None,
886            )
887            .await
888            .unwrap();
889
890        assert!(response.status.is_success());
891    }
892
893    #[tokio::test]
894    async fn test_delete() {
895        let addr = start_test_server().await.unwrap();
896        let url = format!("http://{addr}");
897
898        let client = InnerHttpClient::default();
899        let response = client
900            .send_request(
901                reqwest::Method::DELETE,
902                format!("{url}/delete"),
903                None,
904                None,
905                None,
906                None,
907            )
908            .await
909            .unwrap();
910
911        assert!(response.status.is_success());
912    }
913
914    #[tokio::test]
915    async fn test_not_found() {
916        let addr = start_test_server().await.unwrap();
917        let url = format!("http://{addr}/notfound");
918        let client = InnerHttpClient::default();
919
920        let response = client
921            .send_request(reqwest::Method::GET, url, None, None, None, None)
922            .await
923            .unwrap();
924
925        assert!(response.status.is_client_error());
926        assert_eq!(response.status.as_u16(), 404);
927    }
928
929    #[tokio::test]
930    async fn test_timeout() {
931        let addr = start_test_server().await.unwrap();
932        let url = format!("http://{addr}/slow");
933        let client = InnerHttpClient::default();
934
935        // We'll set a 1-second timeout for a route that sleeps 2 seconds
936        let result = client
937            .send_request(reqwest::Method::GET, url, None, None, None, Some(1))
938            .await;
939
940        match result {
941            Err(HttpClientError::TimeoutError(msg)) => {
942                println!("Got expected timeout error: {msg}");
943            }
944            Err(e) => panic!("Expected a timeout error, was: {e:?}"),
945            Ok(resp) => panic!("Expected a timeout error, but was a successful response: {resp:?}"),
946        }
947    }
948
949    #[rstest]
950    fn test_http_client_without_proxy() {
951        // Create client with no proxy
952        let result = HttpClient::new(
953            HashMap::new(),
954            vec![],
955            vec![],
956            None,
957            None,
958            None, // No proxy
959        );
960
961        assert!(result.is_ok());
962    }
963
964    #[tokio::test]
965    async fn test_http_client_without_proxy_requests_directly() {
966        let addr = start_test_server().await.unwrap();
967        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, Some(2), None).unwrap();
968        let response = client
969            .request(
970                Method::GET,
971                format!("http://{addr}/get"),
972                None,
973                None,
974                None,
975                None,
976                None,
977            )
978            .await
979            .expect("direct request");
980
981        assert_eq!(response.status.as_u16(), StatusCode::OK.as_u16());
982        assert_eq!(response.body.as_ref(), b"hello-world!");
983    }
984
985    #[tokio::test]
986    async fn test_http_client_uses_connect_and_proxy_authorization_for_https() {
987        const USERNAME: &str = "proxytest";
988        const PASSWORD: &str = "fixture42";
989        let (proxy_addr, request_rx) = spawn_rejecting_connect_proxy().await;
990        let client = HttpClient::new(
991            HashMap::new(),
992            vec![],
993            vec![],
994            None,
995            Some(2),
996            Some(format!("http://{USERNAME}:{PASSWORD}@{proxy_addr}")),
997        )
998        .unwrap();
999        let error = client
1000            .request(
1001                Method::GET,
1002                "https://fixture.example.test/path".to_string(),
1003                None,
1004                None,
1005                None,
1006                None,
1007                None,
1008            )
1009            .await
1010            .expect_err("proxy should reject CONNECT");
1011        let request = request_rx.await.expect("captured CONNECT request");
1012        let mut lines = request.split("\r\n");
1013        let request_line = lines.next().expect("CONNECT request line");
1014        let auth_value = lines
1015            .find_map(|line| {
1016                let (name, value) = line.split_once(':')?;
1017                name.eq_ignore_ascii_case("proxy-authorization")
1018                    .then_some(value.trim())
1019            })
1020            .expect("Proxy-Authorization header");
1021        let expected_auth = format!("Basic {}", BASE64.encode(format!("{USERNAME}:{PASSWORD}")));
1022
1023        assert_eq!(request_line, "CONNECT fixture.example.test:443 HTTP/1.1");
1024        assert_eq!(auth_value, expected_auth);
1025        assert!(!error.to_string().contains(PASSWORD));
1026        assert!(!error.to_string().contains(&BASE64.encode(PASSWORD)));
1027        assert!(!error.to_string().contains(&expected_auth));
1028    }
1029
1030    #[tokio::test]
1031    async fn test_http_client_unreachable_proxy_error_redacts_credentials() {
1032        const USERNAME: &str = "proxy-user";
1033        const SECRET: &str = "unreachable-proxy-secret";
1034        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1035        let proxy_addr = listener.local_addr().unwrap();
1036        drop(listener);
1037        let client = HttpClient::new(
1038            HashMap::new(),
1039            vec![],
1040            vec![],
1041            None,
1042            Some(1),
1043            Some(format!("http://{USERNAME}:{SECRET}@{proxy_addr}")),
1044        )
1045        .unwrap();
1046        let error = client
1047            .request(
1048                Method::GET,
1049                "https://fixture.example.test/".to_string(),
1050                None,
1051                None,
1052                None,
1053                None,
1054                None,
1055            )
1056            .await
1057            .expect_err("unreachable proxy should fail");
1058
1059        assert!(!error.to_string().contains(SECRET));
1060        assert!(!error.to_string().contains(&BASE64.encode(SECRET)));
1061        assert!(
1062            !error
1063                .to_string()
1064                .contains(&BASE64.encode(format!("{USERNAME}:{SECRET}")))
1065        );
1066    }
1067
1068    #[rstest]
1069    fn test_http_client_with_valid_proxy() {
1070        // Create client with a valid proxy URL
1071        let result = HttpClient::new(
1072            HashMap::new(),
1073            vec![],
1074            vec![],
1075            None,
1076            None,
1077            Some("http://proxy.example.com:8080".to_string()),
1078        );
1079
1080        assert!(result.is_ok());
1081    }
1082
1083    #[rstest]
1084    fn test_http_client_with_socks5_proxy() {
1085        // Create client with a SOCKS5 proxy URL
1086        let result = HttpClient::new(
1087            HashMap::new(),
1088            vec![],
1089            vec![],
1090            None,
1091            None,
1092            Some("socks5://127.0.0.1:1080".to_string()),
1093        );
1094
1095        assert!(result.is_ok());
1096    }
1097
1098    #[rstest]
1099    fn test_http_client_with_malformed_proxy() {
1100        // Note: reqwest::Proxy::all() is lenient and accepts most strings.
1101        // It only fails on obviously malformed URLs like "://invalid" or "http://".
1102        // More subtle issues (like "not-a-valid-url") are caught when connecting.
1103        let result = HttpClient::new(
1104            HashMap::new(),
1105            vec![],
1106            vec![],
1107            None,
1108            None,
1109            Some("://invalid".to_string()),
1110        );
1111
1112        assert!(result.is_err());
1113        assert!(matches!(result, Err(HttpClientError::InvalidProxy(_))));
1114    }
1115
1116    #[rstest]
1117    fn test_http_client_invalid_proxy_error_redacts_credentials() {
1118        const SECRET: &str = "unique-proxy-secret";
1119        let result = HttpClient::new(
1120            HashMap::new(),
1121            vec![],
1122            vec![],
1123            None,
1124            None,
1125            Some(format!("http://proxytest:{SECRET}@[::1")),
1126        );
1127        let error = result.expect_err("malformed proxy URL should fail");
1128
1129        assert_eq!(
1130            error.to_string(),
1131            "Invalid proxy URL: proxy URL is malformed"
1132        );
1133        assert!(!error.to_string().contains(SECRET));
1134    }
1135
1136    #[rstest]
1137    fn test_http_client_with_empty_proxy_string() {
1138        // Create client with an empty proxy URL string
1139        let result = HttpClient::new(
1140            HashMap::new(),
1141            vec![],
1142            vec![],
1143            None,
1144            None,
1145            Some(String::new()),
1146        );
1147
1148        assert!(result.is_err());
1149        assert!(matches!(result, Err(HttpClientError::InvalidProxy(_))));
1150    }
1151
1152    #[tokio::test]
1153    async fn test_http_client_get() {
1154        let addr = start_test_server().await.unwrap();
1155        let url = format!("http://{addr}/get");
1156
1157        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, None, None).unwrap();
1158        let response = client.get(url, None, None, None, None).await.unwrap();
1159
1160        assert!(response.status.is_success());
1161        assert_eq!(String::from_utf8_lossy(&response.body), "hello-world!");
1162    }
1163
1164    #[tokio::test]
1165    async fn test_http_client_post() {
1166        let addr = start_test_server().await.unwrap();
1167        let url = format!("http://{addr}/post");
1168
1169        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, None, None).unwrap();
1170        let response = client
1171            .post(url, None, None, None, None, None)
1172            .await
1173            .unwrap();
1174
1175        assert!(response.status.is_success());
1176    }
1177
1178    #[tokio::test]
1179    async fn test_http_client_patch() {
1180        let addr = start_test_server().await.unwrap();
1181        let url = format!("http://{addr}/patch");
1182
1183        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, None, None).unwrap();
1184        let response = client
1185            .patch(url, None, None, None, None, None)
1186            .await
1187            .unwrap();
1188
1189        assert!(response.status.is_success());
1190    }
1191
1192    #[tokio::test]
1193    async fn test_http_client_delete() {
1194        let addr = start_test_server().await.unwrap();
1195        let url = format!("http://{addr}/delete");
1196
1197        let client = HttpClient::new(HashMap::new(), vec![], vec![], None, None, None).unwrap();
1198        let response = client.delete(url, None, None, None, None).await.unwrap();
1199
1200        assert!(response.status.is_success());
1201    }
1202}
nautilus_network/http/client.rs

nautilus_network/http/
client.rs
