pipedash_core/infrastructure/
http_client.rs

1use std::net::{
2    IpAddr,
3    Ipv4Addr,
4    Ipv6Addr,
5};
6use std::sync::{
7    Arc,
8    OnceLock,
9    PoisonError,
10    RwLock,
11};
12use std::time::Duration;
13
14use dashmap::DashMap;
15use pipedash_plugin_api::ClientOptions;
16use reqwest::{
17    redirect,
18    Client,
19    ClientBuilder,
20    NoProxy,
21    Proxy,
22    Url,
23};
24
25use crate::domain::{
26    DomainError,
27    DomainResult,
28};
29
30const USER_AGENT: &str = concat!("pipedash/", env!("CARGO_PKG_VERSION"));
31const STATUS_TIMEOUT: Duration = Duration::from_secs(10);
32const STATUS_MAX_REDIRECTS: usize = 3;
33
34type ClientKey = (Option<String>, bool);
35
36pub struct HttpClientManager {
37    clients: RwLock<Arc<Clients>>,
38}
39
40struct Clients {
41    proxy: Option<Url>,
42    default_client: Arc<Client>,
43    by_origin: DashMap<ClientKey, Arc<Client>>,
44    status_client: OnceLock<Arc<Client>>,
45}
46
47impl Clients {
48    fn new(proxy: Option<Url>) -> DomainResult<Self> {
49        let default_client = HttpClientManager::build_client(false, proxy.as_ref())?;
50        Ok(Self {
51            proxy,
52            default_client: Arc::new(default_client),
53            by_origin: DashMap::new(),
54            status_client: OnceLock::new(),
55        })
56    }
57}
58
59impl HttpClientManager {
60    pub fn new() -> DomainResult<Self> {
61        Ok(Self {
62            clients: RwLock::new(Arc::new(Clients::new(None)?)),
63        })
64    }
65
66    pub fn set_proxy(&self, proxy_url: &str) -> DomainResult<bool> {
67        let proxy = parse_proxy_url(proxy_url)?;
68        if self.current().proxy == proxy {
69            return Ok(false);
70        }
71
72        let clients = Arc::new(Clients::new(proxy)?);
73        *self.clients.write().unwrap_or_else(PoisonError::into_inner) = clients;
74        Ok(true)
75    }
76
77    fn current(&self) -> Arc<Clients> {
78        Arc::clone(&self.clients.read().unwrap_or_else(PoisonError::into_inner))
79    }
80
81    pub fn client_for(&self, options: &ClientOptions) -> DomainResult<Arc<Client>> {
82        let clients = self.current();
83        if *options == ClientOptions::default() {
84            return Ok(Arc::clone(&clients.default_client));
85        }
86
87        let key = (
88            options.origin.as_ref().map(normalize_origin),
89            options.accept_invalid_certs,
90        );
91        let entry = clients.by_origin.entry(key).or_try_insert_with(|| {
92            Self::build_client(options.accept_invalid_certs, clients.proxy.as_ref()).map(Arc::new)
93        })?;
94        Ok(Arc::clone(entry.value()))
95    }
96
97    pub fn status_client(&self) -> DomainResult<Arc<Client>> {
98        let clients = self.current();
99        if let Some(client) = clients.status_client.get() {
100            return Ok(Arc::clone(client));
101        }
102        let client = Arc::new(Self::build_status_client(clients.proxy.as_ref())?);
103        Ok(Arc::clone(clients.status_client.get_or_init(|| client)))
104    }
105
106    fn base_builder(proxy: Option<&Url>) -> DomainResult<ClientBuilder> {
107        let _ = rustls::crypto::ring::default_provider().install_default();
108        let builder = Client::builder()
109            .tls_backend_rustls()
110            .https_only(cfg!(target_os = "ios"))
111            .user_agent(USER_AGENT);
112        let Some(proxy) = proxy else {
113            return Ok(builder);
114        };
115
116        let proxy = Proxy::all(proxy.clone())
117            .map_err(|e| DomainError::InvalidConfig(format!("Failed to use the proxy URL: {e}")))?
118            .no_proxy(NoProxy::from_env());
119        Ok(builder.proxy(proxy))
120    }
121
122    fn build_client(accept_invalid_certs: bool, proxy: Option<&Url>) -> DomainResult<Client> {
123        let pool_size = std::env::var("PIPEDASH_HTTP_POOL_SIZE")
124            .ok()
125            .and_then(|s| s.parse().ok())
126            .unwrap_or(10);
127
128        Self::base_builder(proxy)?
129            .tls_danger_accept_invalid_certs(accept_invalid_certs)
130            .pool_max_idle_per_host(pool_size)
131            .pool_idle_timeout(Duration::from_secs(90))
132            .tcp_keepalive(Duration::from_secs(60))
133            .timeout(Duration::from_secs(30))
134            .connect_timeout(Duration::from_secs(10))
135            .build()
136            .map_err(|e| DomainError::InternalError(format!("Failed to create HTTP client: {}", e)))
137    }
138
139    fn build_status_client(proxy: Option<&Url>) -> DomainResult<Client> {
140        let policy = redirect::Policy::custom(|attempt| {
141            if attempt.previous().len() > STATUS_MAX_REDIRECTS {
142                attempt.error("too many redirects")
143            } else if is_safe_redirect_target(attempt.url()) {
144                attempt.follow()
145            } else {
146                attempt.error("redirect target refused")
147            }
148        });
149
150        Self::base_builder(proxy)?
151            .redirect(policy)
152            .timeout(STATUS_TIMEOUT)
153            .connect_timeout(STATUS_TIMEOUT)
154            .build()
155            .map_err(|e| {
156                DomainError::InternalError(format!("Failed to create status HTTP client: {}", e))
157            })
158    }
159}
160
161pub fn parse_proxy_url(value: &str) -> DomainResult<Option<Url>> {
162    let value = value.trim();
163    if value.is_empty() {
164        return Ok(None);
165    }
166
167    let url = Url::parse(value)
168        .ok()
169        .filter(|url| matches!(url.scheme(), "http" | "https") && url.host_str().is_some())
170        .ok_or_else(|| {
171            DomainError::InvalidConfig(
172                "Proxy URL must look like http://proxy.example.com:3128 or https://...".to_string(),
173            )
174        })?;
175    if !url.username().is_empty() || url.password().is_some() {
176        return Err(DomainError::InvalidConfig(
177            "Proxy URL can't contain a username or password. Pipedash supports proxies without \
178             authentication only."
179                .to_string(),
180        ));
181    }
182
183    Ok(Some(url))
184}
185
186fn normalize_origin(url: &Url) -> String {
187    format!(
188        "{}://{}:{}",
189        url.scheme().to_ascii_lowercase(),
190        url.host_str().unwrap_or_default().to_ascii_lowercase(),
191        url.port_or_known_default().unwrap_or_default()
192    )
193}
194
195fn is_safe_redirect_target(url: &Url) -> bool {
196    if url.scheme() != "https" {
197        return false;
198    }
199    let Some(host) = url.host_str() else {
200        return false;
201    };
202    match host
203        .trim_start_matches('[')
204        .trim_end_matches(']')
205        .parse::<IpAddr>()
206    {
207        Ok(ip) => !is_internal_ip(ip),
208        Err(_) => true,
209    }
210}
211
212fn is_internal_ip(ip: IpAddr) -> bool {
213    match ip {
214        IpAddr::V4(v4) => is_internal_v4(v4),
215        IpAddr::V6(v6) => v6
216            .to_ipv4_mapped()
217            .map_or_else(|| is_internal_v6(v6), is_internal_v4),
218    }
219}
220
221fn is_internal_v4(ip: Ipv4Addr) -> bool {
222    ip.is_loopback() || ip.is_private() || ip.is_link_local() || ip.is_unspecified()
223}
224
225fn is_internal_v6(ip: Ipv6Addr) -> bool {
226    ip.is_loopback() || ip.is_unique_local() || ip.is_unicast_link_local() || ip.is_unspecified()
227}
228
229#[cfg(test)]
230mod tests {
231    use super::*;
232
233    fn options(origin: &str, accept_invalid_certs: bool) -> ClientOptions {
234        ClientOptions {
235            origin: Some(Url::parse(origin).unwrap()),
236            accept_invalid_certs,
237        }
238    }
239
240    fn https_only_client() -> Client {
241        HttpClientManager::base_builder(None)
242            .unwrap()
243            .https_only(true)
244            .no_proxy()
245            .add_root_certificate(
246                reqwest::Certificate::from_der(include_bytes!(
247                    "../../tests/fixtures/https_server_cert.der"
248                ))
249                .unwrap(),
250            )
251            .timeout(Duration::from_secs(2))
252            .build()
253            .unwrap()
254    }
255
256    async fn tls_stub(response: String) -> (String, tokio::task::JoinHandle<String>) {
257        use std::io::{
258            Read,
259            Write,
260        };
261
262        let _ = rustls::crypto::ring::default_provider().install_default();
263        let cert = rustls::pki_types::CertificateDer::from(
264            include_bytes!("../../tests/fixtures/https_server_cert.der").to_vec(),
265        );
266        let key = rustls::pki_types::PrivateKeyDer::Pkcs8(
267            include_bytes!("../../tests/fixtures/self_signed_key.der")
268                .to_vec()
269                .into(),
270        );
271        let config = Arc::new(
272            rustls::ServerConfig::builder()
273                .with_no_client_auth()
274                .with_single_cert(vec![cert], key)
275                .unwrap(),
276        );
277        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
278        let url = format!(
279            "https://127.0.0.1:{}",
280            listener.local_addr().unwrap().port()
281        );
282        let server = tokio::spawn(async move {
283            let (stream, _) = tokio::time::timeout(Duration::from_secs(2), listener.accept())
284                .await
285                .unwrap()
286                .unwrap();
287            let stream = stream.into_std().unwrap();
288            stream.set_nonblocking(false).unwrap();
289            stream
290                .set_read_timeout(Some(Duration::from_secs(2)))
291                .unwrap();
292            stream
293                .set_write_timeout(Some(Duration::from_secs(2)))
294                .unwrap();
295            tokio::task::spawn_blocking(move || {
296                let connection = rustls::ServerConnection::new(config).unwrap();
297                let mut stream = rustls::StreamOwned::new(connection, stream);
298                let mut request = Vec::new();
299                let mut buffer = [0; 4096];
300                while !request.windows(4).any(|part| part == b"\r\n\r\n") {
301                    let count = stream.read(&mut buffer).unwrap();
302                    assert!(count > 0 && request.len() + count <= 8192);
303                    request.extend_from_slice(&buffer[..count]);
304                }
305                stream.write_all(response.as_bytes()).unwrap();
306                stream.flush().unwrap();
307                String::from_utf8(request).unwrap()
308            })
309            .await
310            .unwrap()
311        });
312        (url, server)
313    }
314
315    #[tokio::test]
316    async fn https_only_rejects_http_before_sending_credentials() {
317        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
318        let error = https_only_client()
319            .get(format!("http://{}", listener.local_addr().unwrap()))
320            .header("Authorization", "token fixture")
321            .send()
322            .await
323            .unwrap_err();
324
325        assert!(error.is_builder(), "{error}");
326        assert!(
327            tokio::time::timeout(Duration::from_millis(50), listener.accept())
328                .await
329                .is_err()
330        );
331    }
332
333    #[tokio::test]
334    async fn https_only_allows_https_with_credentials() {
335        let (url, server) = tls_stub(
336            "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_string(),
337        )
338        .await;
339        let response = https_only_client()
340            .get(url)
341            .header("Authorization", "token fixture")
342            .send()
343            .await
344            .unwrap();
345
346        assert_eq!(response.text().await.unwrap(), "ok");
347        assert!(server
348            .await
349            .unwrap()
350            .to_ascii_lowercase()
351            .contains("authorization: token fixture"));
352    }
353
354    #[tokio::test]
355    async fn https_only_rejects_http_downgrade_redirect_before_traffic() {
356        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
357        let (url, server) = tls_stub(format!(
358            "HTTP/1.1 302 Found\r\nLocation: http://{}/token\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
359            listener.local_addr().unwrap()
360        ))
361        .await;
362        let error = https_only_client()
363            .get(url)
364            .header("Authorization", "token fixture")
365            .send()
366            .await
367            .unwrap_err();
368
369        assert!(error.is_redirect(), "{error}");
370        assert!(server
371            .await
372            .unwrap()
373            .to_ascii_lowercase()
374            .contains("authorization: token fixture"));
375        assert!(
376            tokio::time::timeout(Duration::from_millis(50), listener.accept())
377                .await
378                .is_err()
379        );
380    }
381
382    #[cfg(target_os = "ios")]
383    #[tokio::test]
384    async fn ios_shared_clients_reject_http_in_every_cache_and_proxy_configuration() {
385        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
386        let url = format!("http://{}", listener.local_addr().unwrap());
387        let manager = HttpClientManager::new().unwrap();
388        for proxy in ["", url.as_str()] {
389            manager.set_proxy(proxy).unwrap();
390            let clients = [
391                manager.client_for(&ClientOptions::default()).unwrap(),
392                manager.client_for(&options(&url, false)).unwrap(),
393                manager.client_for(&options(&url, true)).unwrap(),
394                manager.status_client().unwrap(),
395            ];
396            for client in clients {
397                let error = client
398                    .get(&url)
399                    .header("Authorization", "token fixture")
400                    .send()
401                    .await
402                    .unwrap_err();
403                assert!(error.is_builder(), "{error}");
404            }
405        }
406        assert!(
407            tokio::time::timeout(Duration::from_millis(50), listener.accept())
408                .await
409                .is_err()
410        );
411    }
412
413    #[test]
414    fn default_options_share_the_default_client() {
415        let manager = HttpClientManager::new().unwrap();
416        let a = manager.client_for(&ClientOptions::default()).unwrap();
417        let b = manager.client_for(&ClientOptions::default()).unwrap();
418        assert!(Arc::ptr_eq(&a, &b));
419    }
420
421    #[test]
422    fn equivalent_origins_share_one_client() {
423        let manager = HttpClientManager::new().unwrap();
424        let a = manager.client_for(&options("https://h", false)).unwrap();
425        let b = manager
426            .client_for(&options("https://H:443/", false))
427            .unwrap();
428        assert!(Arc::ptr_eq(&a, &b));
429        assert!(!Arc::ptr_eq(&a, &manager.current().default_client));
430    }
431
432    #[test]
433    fn insecure_and_secure_clients_differ_for_same_origin() {
434        let manager = HttpClientManager::new().unwrap();
435        let secure = manager
436            .client_for(&options("https://argo.local", false))
437            .unwrap();
438        let insecure = manager
439            .client_for(&options("https://argo.local", true))
440            .unwrap();
441        assert!(!Arc::ptr_eq(&secure, &insecure));
442    }
443
444    #[test]
445    fn different_origins_get_different_clients() {
446        let manager = HttpClientManager::new().unwrap();
447        let a = manager
448            .client_for(&options("https://a.example", false))
449            .unwrap();
450        let b = manager
451            .client_for(&options("https://a.example:8443", false))
452            .unwrap();
453        assert!(!Arc::ptr_eq(&a, &b));
454    }
455
456    #[test]
457    fn status_client_is_cached_and_separate() {
458        let manager = HttpClientManager::new().unwrap();
459        let a = manager.status_client().unwrap();
460        let b = manager.status_client().unwrap();
461        assert!(Arc::ptr_eq(&a, &b));
462        let insecure = manager
463            .client_for(&options("https://a.example", true))
464            .unwrap();
465        assert!(!Arc::ptr_eq(&a, &insecure));
466        assert!(!Arc::ptr_eq(&a, &manager.current().default_client));
467    }
468
469    #[test]
470    fn redirect_targets_must_be_public_https() {
471        for allowed in [
472            "https://www.githubstatus.com/api/v2/summary.json",
473            "https://8.8.8.8/",
474        ] {
475            assert!(
476                is_safe_redirect_target(&Url::parse(allowed).unwrap()),
477                "{allowed}"
478            );
479        }
480        for refused in [
481            "http://www.githubstatus.com/",
482            "http://127.0.0.1:8765/",
483            "https://127.0.0.1/",
484            "https://10.1.2.3/",
485            "https://192.168.0.1/",
486            "https://169.254.169.254/",
487            "https://0.0.0.0/",
488            "https://[::1]/",
489            "https://[fe80::1]/",
490            "https://[fd00::1]/",
491            "https://[::ffff:127.0.0.1]/",
492        ] {
493            assert!(
494                !is_safe_redirect_target(&Url::parse(refused).unwrap()),
495                "{refused}"
496            );
497        }
498    }
499
500    #[cfg(not(target_os = "ios"))]
501    async fn proxy_stub() -> (String, Arc<std::sync::Mutex<Vec<String>>>) {
502        use tokio::io::{
503            AsyncReadExt,
504            AsyncWriteExt,
505        };
506
507        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
508        let url = format!("http://{}", listener.local_addr().unwrap());
509        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
510        let log = Arc::clone(&seen);
511        tokio::spawn(async move {
512            loop {
513                let (mut stream, _) = listener.accept().await.unwrap();
514                let mut buf = vec![0; 4096];
515                let n = stream.read(&mut buf).await.unwrap();
516                let head = String::from_utf8_lossy(&buf[..n]);
517                let line = head.lines().next().unwrap_or_default().to_string();
518                log.lock().unwrap().push(line);
519                let _ = stream
520                    .write_all(
521                        b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok",
522                    )
523                    .await;
524            }
525        });
526        (url, seen)
527    }
528
529    #[cfg(not(target_os = "ios"))]
530    #[tokio::test]
531    async fn every_client_goes_through_the_proxy_until_it_is_cleared() {
532        let (proxy, seen) = proxy_stub().await;
533        let manager = HttpClientManager::new().unwrap();
534
535        assert!(manager.set_proxy(&format!(" {proxy} ")).unwrap());
536        assert!(!manager.set_proxy(&proxy).unwrap());
537        let clients = [
538            manager.client_for(&ClientOptions::default()).unwrap(),
539            manager
540                .client_for(&options("http://argo.pipedash.test", true))
541                .unwrap(),
542            manager.status_client().unwrap(),
543        ];
544        for (i, client) in clients.iter().enumerate() {
545            let body = client
546                .get(format!("http://host{i}.pipedash.test/api"))
547                .send()
548                .await
549                .unwrap()
550                .text()
551                .await
552                .unwrap();
553            assert_eq!(body, "ok");
554        }
555
556        assert!(manager.set_proxy("").unwrap());
557        let direct = manager
558            .client_for(&ClientOptions::default())
559            .unwrap()
560            .get("http://127.0.0.1:9/never-proxied")
561            .send()
562            .await;
563
564        assert!(direct.is_err());
565        assert_eq!(
566            *seen.lock().unwrap(),
567            [
568                "GET http://host0.pipedash.test/api HTTP/1.1",
569                "GET http://host1.pipedash.test/api HTTP/1.1",
570                "GET http://host2.pipedash.test/api HTTP/1.1",
571            ]
572        );
573    }
574
575    #[test]
576    fn proxy_url_accepts_http_and_https_without_credentials() {
577        assert_eq!(parse_proxy_url("  ").unwrap(), None);
578        assert_eq!(
579            parse_proxy_url(" http://proxy.corp:3128 ")
580                .unwrap()
581                .unwrap()
582                .as_str(),
583            "http://proxy.corp:3128/"
584        );
585        assert!(parse_proxy_url("https://proxy.corp").unwrap().is_some());
586
587        for refused in [
588            "proxy.corp:3128",
589            "socks5://proxy.corp:1080",
590            "http://",
591            "not a url",
592            "http://alice@proxy.corp:3128",
593            "http://alice:secret@proxy.corp:3128",
594        ] {
595            let error = parse_proxy_url(refused).unwrap_err().to_string();
596            assert!(!error.contains("secret"), "{error}");
597            assert!(matches!(
598                parse_proxy_url(refused),
599                Err(DomainError::InvalidConfig(_))
600            ));
601        }
602    }
603}
pipedash_core/infrastructure/http_client.rs

pipedash_core/infrastructure/
http_client.rs
