ez_ffmpeg/http_input/
client.rs

1//! Shared `HttpClient`: rustls roots, reqwest builder, per-client runtime.
2
3use crate::http_input::config::{HttpTimeouts, ProxyPolicy, RootPolicy};
4use crate::http_input::error::HttpInputError;
5use crate::http_input::runtime::RuntimeHandle;
6use crate::http_input::HttpInputBuilder;
7use std::fmt;
8use std::sync::{Arc, Mutex};
9
10/// Reusable HTTP stack: TLS config, proxy, and a dedicated current-thread Tokio runtime.
11///
12/// Several [`HttpInput`](crate::http_input::HttpInput)s created from the same
13/// client share the connection pool
14/// and the runtime thread. The last drop shuts the runtime down.
15#[derive(Clone)]
16pub struct HttpClient {
17    pub(crate) inner: Arc<HttpClientInner>,
18}
19
20pub(crate) struct HttpClientInner {
21    pub(crate) client: reqwest::Client,
22    pub(crate) timeouts: HttpTimeouts,
23    pub(crate) user_agent: Option<String>,
24    pub(crate) redirect_limit: u32,
25    runtime: Mutex<Option<RuntimeHandle>>,
26}
27
28impl fmt::Debug for HttpClient {
29    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
30        f.debug_struct("HttpClient")
31            .field("user_agent", &self.inner.user_agent)
32            .finish_non_exhaustive()
33    }
34}
35
36impl HttpClient {
37    /// Start a client builder (system roots, environment proxy).
38    pub fn builder() -> HttpClientBuilder {
39        HttpClientBuilder::default()
40    }
41
42    /// Open an input that shares this client's pool and runtime.
43    ///
44    /// The input starts with this client's response-header and body-idle
45    /// timeouts. [`HttpInputBuilder::timeouts`] can override those two per
46    /// input. Connect timeout is fixed when this client is built
47    /// ([`HttpClientBuilder::timeouts`]); reqwest applies it at client
48    /// construction, not per request.
49    pub fn input(&self, url: impl Into<String>) -> HttpInputBuilder {
50        HttpInputBuilder::new(url.into())
51            .client(self.clone())
52            .timeouts(self.inner.timeouts.clone())
53    }
54
55    pub(crate) fn ensure_runtime(&self) -> Result<(), HttpInputError> {
56        let mut guard = self.inner.runtime.lock().unwrap_or_else(|e| e.into_inner());
57        if guard.is_none() {
58            *guard = Some(RuntimeHandle::start()?);
59        }
60        Ok(())
61    }
62
63    pub(crate) fn runtime(&self) -> Result<RuntimeHandle, HttpInputError> {
64        self.ensure_runtime()?;
65        let guard = self.inner.runtime.lock().unwrap_or_else(|e| e.into_inner());
66        guard
67            .as_ref()
68            .cloned()
69            .ok_or_else(|| HttpInputError::Transport {
70                message: "http runtime failed to start".into(),
71            })
72    }
73}
74
75/// Builder for [`HttpClient`]. Certificate and identity bytes are parsed at
76/// [`build`](Self::build); failures do not wait for the first request.
77pub struct HttpClientBuilder {
78    root_policy: RootPolicy,
79    extra_roots_pem: Vec<Vec<u8>>,
80    identity_pem: Option<Vec<u8>>,
81    user_agent: Option<String>,
82    disable_ua: bool,
83    proxy: ProxyPolicy,
84    timeouts: HttpTimeouts,
85    redirect_limit: u32,
86}
87
88impl fmt::Debug for HttpClientBuilder {
89    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
90        f.debug_struct("HttpClientBuilder")
91            .field("root_policy", &self.root_policy)
92            .field("extra_roots", &self.extra_roots_pem.len())
93            .field("has_identity", &self.identity_pem.is_some())
94            .field("user_agent", &self.user_agent)
95            .field("proxy", &self.proxy)
96            .finish_non_exhaustive()
97    }
98}
99
100impl Default for HttpClientBuilder {
101    fn default() -> Self {
102        Self {
103            root_policy: RootPolicy::System,
104            extra_roots_pem: Vec::new(),
105            identity_pem: None,
106            user_agent: None,
107            disable_ua: false,
108            proxy: ProxyPolicy::Environment,
109            timeouts: HttpTimeouts::default(),
110            redirect_limit: 10,
111        }
112    }
113}
114
115impl HttpClientBuilder {
116    /// System store (default) or custom PEMs only.
117    pub fn root_policy(mut self, policy: RootPolicy) -> Self {
118        self.root_policy = policy;
119        self
120    }
121
122    /// Append a PEM certificate to the trust store. Parsed immediately.
123    pub fn add_root_certificate_pem(
124        mut self,
125        pem: impl AsRef<[u8]>,
126    ) -> Result<Self, HttpInputError> {
127        let pem = pem.as_ref();
128        if !pem
129            .windows(27)
130            .any(|window| window.eq_ignore_ascii_case(b"-----BEGIN CERTIFICATE-----"))
131        {
132            return Err(HttpInputError::InvalidCertificate);
133        }
134        reqwest::Certificate::from_pem(pem).map_err(|_| HttpInputError::InvalidCertificate)?;
135        self.extra_roots_pem.push(pem.to_vec());
136        Ok(self)
137    }
138
139    /// PEM client identity (cert chain + unencrypted private key) for mTLS.
140    pub fn client_identity_pem(mut self, pem: impl Into<Vec<u8>>) -> Self {
141        self.identity_pem = Some(pem.into());
142        self
143    }
144
145    /// Override the default `ez-ffmpeg/<version>` User-Agent.
146    pub fn user_agent(mut self, ua: impl Into<String>) -> Self {
147        self.user_agent = Some(ua.into());
148        self.disable_ua = false;
149        self
150    }
151
152    /// Send no User-Agent header.
153    pub fn disable_user_agent(mut self) -> Self {
154        self.disable_ua = true;
155        self.user_agent = None;
156        self
157    }
158
159    /// Proxy policy. Default is environment variables snapshotted at build.
160    pub fn proxy(mut self, policy: ProxyPolicy) -> Self {
161        self.proxy = policy;
162        self
163    }
164
165    /// Default timeouts inherited by [`HttpClient::input`] unless the input
166    /// overrides them. `response_headers` and `read_idle` can be overridden
167    /// per input. `connect` is applied once to the reqwest client at
168    /// [`build`](Self::build) and cannot be changed per input afterwards.
169    pub fn timeouts(mut self, timeouts: HttpTimeouts) -> Self {
170        self.timeouts = timeouts;
171        self
172    }
173
174    /// Maximum redirect hops (default 10). `0` follows no redirect: the
175    /// original request is sent once and a 3xx response is an error.
176    pub fn redirect_limit(mut self, limit: u32) -> Self {
177        self.redirect_limit = limit;
178        self
179    }
180
181    /// Build the reqwest client and load trust anchors. No network I/O.
182    pub fn build(self) -> Result<HttpClient, HttpInputError> {
183        self.timeouts.validate()?;
184        let mut roots: Vec<reqwest::Certificate> = Vec::new();
185        if self.root_policy == RootPolicy::System {
186            let loaded = rustls_native_certs::load_native_certs();
187            for cert in loaded.certs {
188                if let Ok(parsed) = reqwest::Certificate::from_der(&cert) {
189                    roots.push(parsed);
190                }
191            }
192        }
193        for pem in &self.extra_roots_pem {
194            roots.push(
195                reqwest::Certificate::from_pem(pem)
196                    .map_err(|_| HttpInputError::InvalidCertificate)?,
197            );
198        }
199        if roots.is_empty() {
200            return Err(HttpInputError::NoTrustAnchors);
201        }
202
203        let mut builder = reqwest::Client::builder()
204            .use_rustls_tls()
205            .tls_built_in_root_certs(false)
206            .http1_only()
207            .redirect(reqwest::redirect::Policy::none())
208            .connect_timeout(self.timeouts.connect)
209            .pool_max_idle_per_host(4)
210            .https_only(false);
211
212        for cert in roots {
213            builder = builder.add_root_certificate(cert);
214        }
215
216        if let Some(mut pem) = self.identity_pem {
217            let identity = reqwest::Identity::from_pem(&pem).map_err(|_| {
218                pem.fill(0);
219                HttpInputError::IdentityInvalid
220            })?;
221            pem.fill(0);
222            builder = builder.identity(identity);
223        }
224
225        builder = match self.proxy {
226            ProxyPolicy::Environment => builder,
227            ProxyPolicy::Disabled => builder.no_proxy(),
228            ProxyPolicy::Explicit(cfg) => {
229                let mut proxy =
230                    reqwest::Proxy::all(cfg.url()).map_err(|_| HttpInputError::InvalidProxy)?;
231                if let (Some(user), Some(pass)) = (cfg.username_ref(), cfg.password_ref()) {
232                    proxy = proxy.basic_auth(user, pass);
233                }
234                builder.proxy(proxy)
235            }
236        };
237
238        let client = builder.build().map_err(|e| HttpInputError::Transport {
239            message: sanitize_transport(&e.to_string()),
240        })?;
241
242        let user_agent = if self.disable_ua {
243            None
244        } else {
245            Some(self.user_agent.unwrap_or_else(default_user_agent))
246        };
247
248        Ok(HttpClient {
249            inner: Arc::new(HttpClientInner {
250                client,
251                timeouts: self.timeouts,
252                user_agent,
253                redirect_limit: self.redirect_limit,
254                runtime: Mutex::new(None),
255            }),
256        })
257    }
258}
259
260pub(crate) fn default_user_agent() -> String {
261    format!("ez-ffmpeg/{}", env!("CARGO_PKG_VERSION"))
262}
263
264pub(crate) fn sanitize_transport(msg: &str) -> String {
265    // reqwest includes the full URL in many errors. Replace URL-like
266    // tokens so scheme/host/path/userinfo never leave the crate.
267    let mut out = String::new();
268    let mut rest = msg;
269    while let Some(idx) = rest.find("://") {
270        let prefix = &rest[..idx];
271        let scheme_start = prefix
272            .rfind(|c: char| !(c.is_ascii_alphabetic() || c == '+' || c == '.' || c == '-'))
273            .map(|i| i + 1)
274            .unwrap_or(0);
275        out.push_str(&prefix[..scheme_start]);
276        out.push_str("[url]");
277        let after = &rest[idx + 3..];
278        let end = after
279            .find(|c: char| c.is_whitespace() || matches!(c, '\'' | '"' | ')' | ']' | ',' | ';'))
280            .unwrap_or(after.len());
281        rest = &after[end..];
282        if out.len() > 240 {
283            break;
284        }
285    }
286    for ch in rest.chars() {
287        if ch == '?' || ch == '#' {
288            break;
289        }
290        out.push(ch);
291        if out.len() > 240 {
292            break;
293        }
294    }
295    out
296}
297
298/// Shared defaults used when [`HttpInput::builder`] creates an exclusive client.
299pub(crate) fn exclusive_client(
300    timeouts: HttpTimeouts,
301    user_agent: Option<String>,
302    disable_ua: bool,
303) -> Result<HttpClient, HttpInputError> {
304    let mut b = HttpClient::builder().timeouts(timeouts);
305    if disable_ua {
306        b = b.disable_user_agent();
307    } else if let Some(ua) = user_agent {
308        b = b.user_agent(ua);
309    }
310    b.build()
311}
312
313#[cfg(test)]
314mod tests {
315    use super::*;
316
317    #[test]
318    fn redirect_limit_zero_survives_build() {
319        let client = HttpClient::builder().redirect_limit(0).build().unwrap();
320        assert_eq!(
321            client.inner.redirect_limit, 0,
322            "an explicit zero-redirect policy must not be coerced to 1"
323        );
324    }
325
326    #[test]
327    fn custom_only_without_roots_fails() {
328        let err = HttpClient::builder()
329            .root_policy(RootPolicy::CustomOnly)
330            .build()
331            .unwrap_err();
332        assert!(matches!(err, HttpInputError::NoTrustAnchors));
333    }
334
335    #[test]
336    fn invalid_pem_is_rejected() {
337        let err = HttpClient::builder()
338            .add_root_certificate_pem("not-a-cert")
339            .unwrap_err();
340        assert!(matches!(err, HttpInputError::InvalidCertificate));
341    }
342
343    #[test]
344    fn default_ua_uses_crate_version() {
345        let ua = default_user_agent();
346        assert!(ua.starts_with("ez-ffmpeg/"), "{ua}");
347        assert!(!ua.contains("reqwest"), "{ua}");
348    }
349
350    #[test]
351    fn sanitize_transport_redacts_urls() {
352        let raw = "error sending request for url (https://user:hunter2@cdn.example/signed/video.mp4?token=secret): connection reset";
353        let clean = sanitize_transport(raw);
354        assert!(!clean.contains("cdn.example"), "{clean}");
355        assert!(!clean.contains("hunter2"), "{clean}");
356        assert!(!clean.contains("token"), "{clean}");
357        assert!(!clean.contains("secret"), "{clean}");
358        assert!(clean.contains("[url]"), "{clean}");
359        assert!(clean.contains("connection reset"), "{clean}");
360    }
361}
ez_ffmpeg/http_input/client.rs

ez_ffmpeg/http_input/
client.rs
