ez_ffmpeg/http_input/
config.rs

1//! Timeouts, reconnect policy, proxy, and header validation.
2
3use crate::http_input::error::HttpInputError;
4use reqwest::header::{HeaderName, HeaderValue};
5use std::fmt;
6use std::time::Duration;
7
8/// Default connect timeout (design §8.7).
9pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
10/// Default response-header timeout.
11pub const DEFAULT_HEADER_TIMEOUT: Duration = Duration::from_secs(15);
12/// Default body idle timeout; `None` disables it.
13pub const DEFAULT_READ_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
14/// AVIO / runtime stop poll tick. Not a network idle timeout.
15pub const STOP_POLL_TICK: Duration = Duration::from_millis(25);
16
17const MAX_HEADER_NAME_BYTES: usize = 8 * 1024;
18const MAX_HEADER_VALUE_BYTES: usize = 8 * 1024;
19const MAX_HEADER_TOTAL_BYTES: usize = 32 * 1024;
20
21/// Connect / header / body-idle deadlines for one input.
22#[derive(Debug, Clone)]
23#[non_exhaustive]
24pub struct HttpTimeouts {
25    /// TCP + TLS connect budget. Applied at
26    /// [`crate::http_input::HttpClientBuilder::build`] to the reqwest client;
27    /// [`crate::http_input::HttpInputBuilder::timeouts`] cannot change it.
28    pub connect: Duration,
29    /// Budget from request start until response headers.
30    pub response_headers: Duration,
31    /// Reset after every body chunk. `None` means no idle limit.
32    pub read_idle: Option<Duration>,
33}
34
35impl Default for HttpTimeouts {
36    fn default() -> Self {
37        Self {
38            connect: DEFAULT_CONNECT_TIMEOUT,
39            response_headers: DEFAULT_HEADER_TIMEOUT,
40            read_idle: Some(DEFAULT_READ_IDLE_TIMEOUT),
41        }
42    }
43}
44
45impl HttpTimeouts {
46    pub(crate) fn validate(&self) -> Result<(), HttpInputError> {
47        if self.connect.is_zero() || self.response_headers.is_zero() {
48            return Err(HttpInputError::InvalidTimeout);
49        }
50        if let Some(idle) = self.read_idle {
51            if idle.is_zero() {
52                return Err(HttpInputError::InvalidTimeout);
53            }
54        }
55        Ok(())
56    }
57}
58
59/// Application-level reconnect. Default matches FFmpeg `reconnect=0`.
60#[derive(Debug, Clone)]
61#[non_exhaustive]
62pub struct ReconnectPolicy {
63    /// When false, no application-level retry after a response has started.
64    pub enabled: bool,
65    /// Allow reconnect on unknown-length non-seekable / live streams.
66    /// Seekable resources never use this path (Range resume or
67    /// `TruncatedBody`). A declared `Content-Length` without
68    /// `Accept-Ranges` is treated as VOD and also refuses a restart from
69    /// byte 0 — that would duplicate prefix bytes already given to FFmpeg.
70    pub reconnect_streamed: bool,
71    /// Reconnect after a clean EOF. For unknown-length live streams this
72    /// re-GETs the resource; for known-length VOD it is a no-op (the body
73    /// already ended).
74    pub reconnect_at_eof: bool,
75    /// Maximum application retries.
76    pub max_retries: u32,
77    /// Cap on a single backoff delay.
78    pub max_delay: Duration,
79    /// Cap on summed backoff. Zero means no total cap.
80    pub max_total_delay: Duration,
81    /// Honor a valid `Retry-After` when retrying.
82    pub respect_retry_after: bool,
83    /// HTTP statuses eligible for retry after `enabled`.
84    pub retry_http_statuses: Vec<u16>,
85    /// Seekable reconnect / Range continuation requires ETag or Last-Modified.
86    /// Default is true; set false only when the caller accepts an unverified splice.
87    pub require_validator: bool,
88}
89
90impl Default for ReconnectPolicy {
91    fn default() -> Self {
92        Self {
93            enabled: false,
94            reconnect_streamed: false,
95            reconnect_at_eof: false,
96            max_retries: 0,
97            max_delay: Duration::from_secs(30),
98            max_total_delay: Duration::ZERO,
99            respect_retry_after: true,
100            retry_http_statuses: vec![408, 429, 500, 502, 503, 504],
101            require_validator: true,
102        }
103    }
104}
105
106impl ReconnectPolicy {
107    /// Conservative seekable-resource retry (still opt-in via `enabled`).
108    pub fn seekable_default() -> Self {
109        Self {
110            enabled: true,
111            reconnect_streamed: false,
112            reconnect_at_eof: false,
113            max_retries: 5,
114            max_delay: Duration::from_secs(30),
115            max_total_delay: Duration::from_secs(60),
116            respect_retry_after: true,
117            retry_http_statuses: vec![408, 429, 500, 502, 503, 504],
118            require_validator: true,
119        }
120    }
121
122    /// Live / unknown-length streamed retry (restart from the live edge).
123    ///
124    /// Sets `reconnect_streamed` together with `enabled` and a non-zero
125    /// `max_retries` (5, matching [`seekable_default`](Self::seekable_default)).
126    /// Setting `reconnect_streamed = true` on a hand-built policy without also
127    /// raising `max_retries` above the default 0 used to be a silent no-op:
128    /// every retry check consults the retry budget first. `reconnect_at_eof`
129    /// is on because live unknown-length streams typically want to re-GET
130    /// after a clean EOF.
131    pub fn streamed_default() -> Self {
132        Self {
133            enabled: true,
134            reconnect_streamed: true,
135            reconnect_at_eof: true,
136            max_retries: 5,
137            max_delay: Duration::from_secs(30),
138            max_total_delay: Duration::from_secs(60),
139            respect_retry_after: true,
140            retry_http_statuses: vec![408, 429, 500, 502, 503, 504],
141            require_validator: true,
142        }
143    }
144}
145
146/// Where TLS trust anchors come from.
147#[derive(Debug, Clone, Copy, PartialEq, Eq)]
148#[non_exhaustive]
149pub enum RootPolicy {
150    /// System store plus any extra PEMs. Default.
151    System,
152    /// Only extra PEMs; system store is not consulted.
153    CustomOnly,
154}
155
156/// Proxy selection. Credentials are never printed by [`Debug`].
157#[derive(Clone, Default)]
158#[non_exhaustive]
159pub enum ProxyPolicy {
160    /// Snapshot `HTTP_PROXY` / `HTTPS_PROXY` / `ALL_PROXY` / `NO_PROXY` at client build.
161    #[default]
162    Environment,
163    /// Direct connect.
164    Disabled,
165    /// Explicit proxy URL (and optional credentials).
166    Explicit(ProxyConfig),
167}
168
169impl fmt::Debug for ProxyPolicy {
170    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
171        match self {
172            Self::Environment => f.write_str("Environment"),
173            Self::Disabled => f.write_str("Disabled"),
174            Self::Explicit(cfg) => f.debug_tuple("Explicit").field(cfg).finish(),
175        }
176    }
177}
178
179/// Explicit proxy endpoint. Password is redacted in [`Debug`].
180#[derive(Clone)]
181pub struct ProxyConfig {
182    url: String,
183    username: Option<String>,
184    password: Option<String>,
185}
186
187impl fmt::Debug for ProxyConfig {
188    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
189        f.debug_struct("ProxyConfig")
190            .field("url", &redact_proxy_url(&self.url))
191            .field("username", &self.username.as_ref().map(|_| "***"))
192            .field("password", &self.password.as_ref().map(|_| "***"))
193            .finish()
194    }
195}
196
197fn redact_proxy_url(raw: &str) -> String {
198    match reqwest::Url::parse(raw) {
199        Ok(mut url) => {
200            let _ = url.set_username("");
201            let _ = url.set_password(None);
202            url.set_query(None);
203            url.set_fragment(None);
204            url.to_string()
205        }
206        Err(_) => "[invalid-proxy-url]".into(),
207    }
208}
209
210impl ProxyConfig {
211    /// `url` is the proxy origin (`http://127.0.0.1:8080`). Credentials are separate.
212    pub fn new(url: impl Into<String>) -> Self {
213        Self {
214            url: url.into(),
215            username: None,
216            password: None,
217        }
218    }
219
220    /// Basic-auth username for the proxy. Not sent to the origin.
221    pub fn username(mut self, username: impl Into<String>) -> Self {
222        self.username = Some(username.into());
223        self
224    }
225
226    /// Basic-auth password for the proxy. Not sent to the origin.
227    pub fn password(mut self, password: impl Into<String>) -> Self {
228        self.password = Some(password.into());
229        self
230    }
231
232    pub(crate) fn url(&self) -> &str {
233        &self.url
234    }
235
236    pub(crate) fn username_ref(&self) -> Option<&str> {
237        self.username.as_deref()
238    }
239
240    pub(crate) fn password_ref(&self) -> Option<&str> {
241        self.password.as_deref()
242    }
243}
244
245const RESERVED_HEADERS: &[&str] = &[
246    "host",
247    "content-length",
248    "content-range",
249    "transfer-encoding",
250    "connection",
251    "range",
252    "if-range",
253    "accept-encoding",
254    "proxy-authorization",
255];
256
257pub(crate) fn is_reserved_header(name: &str) -> bool {
258    RESERVED_HEADERS
259        .iter()
260        .any(|reserved| name.eq_ignore_ascii_case(reserved))
261}
262
263pub(crate) fn validate_header(name: &str, value: &str) -> Result<(), HttpInputError> {
264    if name.len() > MAX_HEADER_NAME_BYTES || value.len() > MAX_HEADER_VALUE_BYTES {
265        return Err(HttpInputError::HeaderInvalid {
266            name: name.to_string(),
267        });
268    }
269    if is_reserved_header(name) {
270        return Err(HttpInputError::HeaderReserved {
271            name: name.to_string(),
272        });
273    }
274    if HeaderName::from_bytes(name.as_bytes()).is_err()
275        || HeaderValue::from_bytes(value.as_bytes()).is_err()
276    {
277        return Err(HttpInputError::HeaderInvalid {
278            name: name.to_string(),
279        });
280    }
281    Ok(())
282}
283
284pub(crate) fn validate_header_set(headers: &[(String, String)]) -> Result<(), HttpInputError> {
285    let mut total = 0usize;
286    for (name, value) in headers {
287        validate_header(name, value)?;
288        total = total.saturating_add(name.len()).saturating_add(value.len());
289        if total > MAX_HEADER_TOTAL_BYTES {
290            return Err(HttpInputError::HeaderInvalid { name: name.clone() });
291        }
292    }
293    Ok(())
294}
295
296#[cfg(test)]
297mod tests {
298    use super::*;
299
300    #[test]
301    fn reserved_headers_are_rejected() {
302        for name in ["Accept-Encoding", "RANGE", "host", "Proxy-Authorization"] {
303            let err = validate_header(name, "x").unwrap_err();
304            assert!(
305                matches!(err, HttpInputError::HeaderReserved { .. }),
306                "{err}"
307            );
308        }
309    }
310
311    #[test]
312    fn authorization_is_allowed() {
313        validate_header("Authorization", "Bearer abc").unwrap();
314        validate_header("Cookie", "a=b").unwrap();
315    }
316
317    #[test]
318    fn zero_timeout_is_rejected() {
319        let t = HttpTimeouts {
320            connect: Duration::ZERO,
321            ..Default::default()
322        };
323        assert!(matches!(t.validate(), Err(HttpInputError::InvalidTimeout)));
324    }
325
326    #[test]
327    fn proxy_debug_redacts_password() {
328        let cfg = ProxyConfig::new("http://127.0.0.1:8080")
329            .username("u")
330            .password("super-secret");
331        let rendered = format!("{cfg:?}");
332        assert!(!rendered.contains("super-secret"), "{rendered}");
333        assert!(rendered.contains("***"), "{rendered}");
334    }
335
336    #[test]
337    fn proxy_debug_redacts_userinfo() {
338        let cfg = ProxyConfig::new("http://user:hunter2@127.0.0.1:8080").username("u");
339        let rendered = format!("{cfg:?}");
340        assert!(!rendered.contains("hunter2"), "{rendered}");
341        assert!(!rendered.contains("user:"), "{rendered}");
342        assert!(!rendered.contains("username: Some(\"u\")"), "{rendered}");
343        assert!(rendered.contains("127.0.0.1:8080"), "{rendered}");
344    }
345
346    #[test]
347    fn require_validator_defaults_true() {
348        assert!(ReconnectPolicy::default().require_validator);
349        assert!(ReconnectPolicy::seekable_default().require_validator);
350        assert!(!ReconnectPolicy::default().enabled);
351        assert!(ReconnectPolicy::seekable_default().enabled);
352    }
353
354    #[test]
355    fn streamed_default_has_a_usable_retry_budget() {
356        let policy = ReconnectPolicy::streamed_default();
357        assert!(policy.enabled);
358        assert!(policy.reconnect_streamed);
359        assert!(policy.reconnect_at_eof);
360        assert!(
361            policy.max_retries > 0,
362            "reconnect_streamed without max_retries is a silent no-op"
363        );
364        assert!(policy.require_validator);
365    }
366}
ez_ffmpeg/http_input/config.rs

ez_ffmpeg/http_input/
config.rs
