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    static HTTPS_CERTIFICATE: std::sync::LazyLock<rcgen::CertifiedKey<rcgen::KeyPair>> =
241        std::sync::LazyLock::new(|| {
242            let now = std::time::SystemTime::now();
243            let mut params =
244                rcgen::CertificateParams::new(vec!["localhost".into(), "127.0.0.1".into()])
245                    .unwrap();
246            params.not_before = (now - Duration::from_secs(60)).into();
247            params.not_after = (now + Duration::from_secs(86400)).into();
248            params.key_usages = vec![rcgen::KeyUsagePurpose::DigitalSignature];
249            params.extended_key_usages = vec![rcgen::ExtendedKeyUsagePurpose::ServerAuth];
250            let signing_key = rcgen::KeyPair::generate().unwrap();
251            let cert = params.self_signed(&signing_key).unwrap();
252            rcgen::CertifiedKey { cert, signing_key }
253        });
254
255    fn https_only_client() -> Client {
256        HttpClientManager::base_builder(None)
257            .unwrap()
258            .https_only(true)
259            .no_proxy()
260            .add_root_certificate(
261                reqwest::Certificate::from_der(HTTPS_CERTIFICATE.cert.der()).unwrap(),
262            )
263            .timeout(Duration::from_secs(2))
264            .build()
265            .unwrap()
266    }
267
268    async fn tls_stub(response: String) -> (String, tokio::task::JoinHandle<String>) {
269        use std::io::{
270            Read,
271            Write,
272        };
273
274        let _ = rustls::crypto::ring::default_provider().install_default();
275        let certificate = &*HTTPS_CERTIFICATE;
276        let cert = certificate.cert.der().clone();
277        let key =
278            rustls::pki_types::PrivateKeyDer::Pkcs8(certificate.signing_key.serialize_der().into());
279        let config = Arc::new(
280            rustls::ServerConfig::builder()
281                .with_no_client_auth()
282                .with_single_cert(vec![cert], key)
283                .unwrap(),
284        );
285        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
286        let url = format!(
287            "https://127.0.0.1:{}",
288            listener.local_addr().unwrap().port()
289        );
290        let server = tokio::spawn(async move {
291            let (stream, _) = tokio::time::timeout(Duration::from_secs(2), listener.accept())
292                .await
293                .unwrap()
294                .unwrap();
295            let stream = stream.into_std().unwrap();
296            stream.set_nonblocking(false).unwrap();
297            stream
298                .set_read_timeout(Some(Duration::from_secs(2)))
299                .unwrap();
300            stream
301                .set_write_timeout(Some(Duration::from_secs(2)))
302                .unwrap();
303            tokio::task::spawn_blocking(move || {
304                let connection = rustls::ServerConnection::new(config).unwrap();
305                let mut stream = rustls::StreamOwned::new(connection, stream);
306                let mut request = Vec::new();
307                let mut buffer = [0; 4096];
308                while !request.windows(4).any(|part| part == b"\r\n\r\n") {
309                    let count = stream.read(&mut buffer).unwrap();
310                    assert!(count > 0 && request.len() + count <= 8192);
311                    request.extend_from_slice(&buffer[..count]);
312                }
313                stream.write_all(response.as_bytes()).unwrap();
314                stream.flush().unwrap();
315                String::from_utf8(request).unwrap()
316            })
317            .await
318            .unwrap()
319        });
320        (url, server)
321    }
322
323    #[tokio::test]
324    async fn https_only_rejects_http_before_sending_credentials() {
325        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
326        let error = https_only_client()
327            .get(format!("http://{}", listener.local_addr().unwrap()))
328            .header("Authorization", "token fixture")
329            .send()
330            .await
331            .unwrap_err();
332
333        assert!(error.is_builder(), "{error}");
334        assert!(
335            tokio::time::timeout(Duration::from_millis(50), listener.accept())
336                .await
337                .is_err()
338        );
339    }
340
341    #[tokio::test]
342    async fn https_only_allows_https_with_credentials() {
343        let (url, server) = tls_stub(
344            "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_string(),
345        )
346        .await;
347        let response = https_only_client()
348            .get(url)
349            .header("Authorization", "token fixture")
350            .send()
351            .await
352            .unwrap();
353
354        assert_eq!(response.text().await.unwrap(), "ok");
355        assert!(server
356            .await
357            .unwrap()
358            .to_ascii_lowercase()
359            .contains("authorization: token fixture"));
360    }
361
362    #[tokio::test]
363    async fn https_only_rejects_http_downgrade_redirect_before_traffic() {
364        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
365        let (url, server) = tls_stub(format!(
366            "HTTP/1.1 302 Found\r\nLocation: http://{}/token\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
367            listener.local_addr().unwrap()
368        ))
369        .await;
370        let error = https_only_client()
371            .get(url)
372            .header("Authorization", "token fixture")
373            .send()
374            .await
375            .unwrap_err();
376
377        assert!(error.is_redirect(), "{error}");
378        assert!(server
379            .await
380            .unwrap()
381            .to_ascii_lowercase()
382            .contains("authorization: token fixture"));
383        assert!(
384            tokio::time::timeout(Duration::from_millis(50), listener.accept())
385                .await
386                .is_err()
387        );
388    }
389
390    #[cfg(target_os = "ios")]
391    #[tokio::test]
392    async fn ios_shared_clients_reject_http_in_every_cache_and_proxy_configuration() {
393        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
394        let url = format!("http://{}", listener.local_addr().unwrap());
395        let manager = HttpClientManager::new().unwrap();
396        for proxy in ["", url.as_str()] {
397            manager.set_proxy(proxy).unwrap();
398            let clients = [
399                manager.client_for(&ClientOptions::default()).unwrap(),
400                manager.client_for(&options(&url, false)).unwrap(),
401                manager.client_for(&options(&url, true)).unwrap(),
402                manager.status_client().unwrap(),
403            ];
404            for client in clients {
405                let error = client
406                    .get(&url)
407                    .header("Authorization", "token fixture")
408                    .send()
409                    .await
410                    .unwrap_err();
411                assert!(error.is_builder(), "{error}");
412            }
413        }
414        assert!(
415            tokio::time::timeout(Duration::from_millis(50), listener.accept())
416                .await
417                .is_err()
418        );
419    }
420
421    #[test]
422    fn default_options_share_the_default_client() {
423        let manager = HttpClientManager::new().unwrap();
424        let a = manager.client_for(&ClientOptions::default()).unwrap();
425        let b = manager.client_for(&ClientOptions::default()).unwrap();
426        assert!(Arc::ptr_eq(&a, &b));
427    }
428
429    #[test]
430    fn equivalent_origins_share_one_client() {
431        let manager = HttpClientManager::new().unwrap();
432        let a = manager.client_for(&options("https://h", false)).unwrap();
433        let b = manager
434            .client_for(&options("https://H:443/", false))
435            .unwrap();
436        assert!(Arc::ptr_eq(&a, &b));
437        assert!(!Arc::ptr_eq(&a, &manager.current().default_client));
438    }
439
440    #[test]
441    fn insecure_and_secure_clients_differ_for_same_origin() {
442        let manager = HttpClientManager::new().unwrap();
443        let secure = manager
444            .client_for(&options("https://argo.local", false))
445            .unwrap();
446        let insecure = manager
447            .client_for(&options("https://argo.local", true))
448            .unwrap();
449        assert!(!Arc::ptr_eq(&secure, &insecure));
450    }
451
452    #[test]
453    fn different_origins_get_different_clients() {
454        let manager = HttpClientManager::new().unwrap();
455        let a = manager
456            .client_for(&options("https://a.example", false))
457            .unwrap();
458        let b = manager
459            .client_for(&options("https://a.example:8443", false))
460            .unwrap();
461        assert!(!Arc::ptr_eq(&a, &b));
462    }
463
464    #[test]
465    fn status_client_is_cached_and_separate() {
466        let manager = HttpClientManager::new().unwrap();
467        let a = manager.status_client().unwrap();
468        let b = manager.status_client().unwrap();
469        assert!(Arc::ptr_eq(&a, &b));
470        let insecure = manager
471            .client_for(&options("https://a.example", true))
472            .unwrap();
473        assert!(!Arc::ptr_eq(&a, &insecure));
474        assert!(!Arc::ptr_eq(&a, &manager.current().default_client));
475    }
476
477    #[test]
478    fn redirect_targets_must_be_public_https() {
479        for allowed in [
480            "https://www.githubstatus.com/api/v2/summary.json",
481            "https://8.8.8.8/",
482        ] {
483            assert!(
484                is_safe_redirect_target(&Url::parse(allowed).unwrap()),
485                "{allowed}"
486            );
487        }
488        for refused in [
489            "http://www.githubstatus.com/",
490            "http://127.0.0.1:8765/",
491            "https://127.0.0.1/",
492            "https://10.1.2.3/",
493            "https://192.168.0.1/",
494            "https://169.254.169.254/",
495            "https://0.0.0.0/",
496            "https://[::1]/",
497            "https://[fe80::1]/",
498            "https://[fd00::1]/",
499            "https://[::ffff:127.0.0.1]/",
500        ] {
501            assert!(
502                !is_safe_redirect_target(&Url::parse(refused).unwrap()),
503                "{refused}"
504            );
505        }
506    }
507
508    #[cfg(not(target_os = "ios"))]
509    async fn proxy_stub() -> (String, Arc<std::sync::Mutex<Vec<String>>>) {
510        use tokio::io::{
511            AsyncReadExt,
512            AsyncWriteExt,
513        };
514
515        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
516        let url = format!("http://{}", listener.local_addr().unwrap());
517        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
518        let log = Arc::clone(&seen);
519        tokio::spawn(async move {
520            loop {
521                let (mut stream, _) = listener.accept().await.unwrap();
522                let mut buf = vec![0; 4096];
523                let n = stream.read(&mut buf).await.unwrap();
524                let head = String::from_utf8_lossy(&buf[..n]);
525                let line = head.lines().next().unwrap_or_default().to_string();
526                log.lock().unwrap().push(line);
527                let _ = stream
528                    .write_all(
529                        b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok",
530                    )
531                    .await;
532            }
533        });
534        (url, seen)
535    }
536
537    #[cfg(not(target_os = "ios"))]
538    #[tokio::test]
539    async fn every_client_goes_through_the_proxy_until_it_is_cleared() {
540        let (proxy, seen) = proxy_stub().await;
541        let manager = HttpClientManager::new().unwrap();
542
543        assert!(manager.set_proxy(&format!(" {proxy} ")).unwrap());
544        assert!(!manager.set_proxy(&proxy).unwrap());
545        let clients = [
546            manager.client_for(&ClientOptions::default()).unwrap(),
547            manager
548                .client_for(&options("http://argo.pipedash.test", true))
549                .unwrap(),
550            manager.status_client().unwrap(),
551        ];
552        for (i, client) in clients.iter().enumerate() {
553            let body = client
554                .get(format!("http://host{i}.pipedash.test/api"))
555                .send()
556                .await
557                .unwrap()
558                .text()
559                .await
560                .unwrap();
561            assert_eq!(body, "ok");
562        }
563
564        assert!(manager.set_proxy("").unwrap());
565        let direct = manager
566            .client_for(&ClientOptions::default())
567            .unwrap()
568            .get("http://127.0.0.1:9/never-proxied")
569            .send()
570            .await;
571
572        assert!(direct.is_err());
573        assert_eq!(
574            *seen.lock().unwrap(),
575            [
576                "GET http://host0.pipedash.test/api HTTP/1.1",
577                "GET http://host1.pipedash.test/api HTTP/1.1",
578                "GET http://host2.pipedash.test/api HTTP/1.1",
579            ]
580        );
581    }
582
583    #[test]
584    fn proxy_url_accepts_http_and_https_without_credentials() {
585        assert_eq!(parse_proxy_url("  ").unwrap(), None);
586        assert_eq!(
587            parse_proxy_url(" http://proxy.corp:3128 ")
588                .unwrap()
589                .unwrap()
590                .as_str(),
591            "http://proxy.corp:3128/"
592        );
593        assert!(parse_proxy_url("https://proxy.corp").unwrap().is_some());
594
595        for refused in [
596            "proxy.corp:3128",
597            "socks5://proxy.corp:1080",
598            "http://",
599            "not a url",
600            "http://alice@proxy.corp:3128",
601            "http://alice:secret@proxy.corp:3128",
602        ] {
603            let error = parse_proxy_url(refused).unwrap_err().to_string();
604            assert!(!error.contains("secret"), "{error}");
605            assert!(matches!(
606                parse_proxy_url(refused),
607                Err(DomainError::InvalidConfig(_))
608            ));
609        }
610    }
611}
pipedash_core/infrastructure/http_client.rs

pipedash_core/infrastructure/
http_client.rs
