ez_ffmpeg/http_input/
mod.rs

1//! Experimental single-resource HTTP(S) input via rustls.
2//!
3//! This module is compiled only with the default-off `http-input` feature.
4//! It never hijacks [`crate::Input::from`] URL routing: FFmpeg still handles
5//! `https://` URLs when the linked build has an HTTPS protocol. Only the
6//! explicit [`HttpInput`](crate::http_input::HttpInput) /
7//! [`HttpClient`](crate::http_input::HttpClient) API uses the Rust HTTP stack.
8//!
9//! # Scope
10//!
11//! One HTTP(S) response body is one media resource (MP4, MPEG-TS, FLV,
12//! Matroska, or a single live connection). HLS and DASH are **not**
13//! supported: they are rejected by URL / Content-Type / prefix sniff and by
14//! a deny-all `io_open` callback. Full HLS needs a demuxer-specific adapter
15//! (FFmpeg 8.1 `hls.c` calls `avio_find_protocol_name` before `io_open`).
16//!
17//! # Experimental
18//!
19//! The types in this module are experimental. Errors, builder fields, and
20//! reconnect policy may still change.
21
22pub(crate) mod client;
23pub(crate) mod config;
24pub(crate) mod error;
25pub(crate) mod runtime;
26pub(crate) mod sniff;
27pub(crate) mod stream;
28pub(crate) mod urlutil;
29
30pub use client::{HttpClient, HttpClientBuilder};
31pub use config::{
32    HttpTimeouts, ProxyConfig, ProxyPolicy, ReconnectPolicy, RootPolicy, STOP_POLL_TICK,
33};
34pub use error::{HttpInputError, ManifestKind};
35
36use crate::core::context::http_avio;
37use crate::core::context::input::Input;
38use crate::core::context::InterruptState;
39use crate::http_input::client::exclusive_client;
40use crate::http_input::stream::{wait_reply, RequestSpec, StreamJob};
41use crate::http_input::urlutil::parse_input_url;
42use crossbeam_channel::bounded;
43use std::ffi::CString;
44use std::fmt;
45use std::sync::atomic::{AtomicBool, Ordering};
46use std::sync::{Arc, Mutex};
47use std::time::Duration;
48
49const CHANNEL_CAP: usize = 8;
50
51/// Custom-IO read callback handed to the shared AVIO bridge: fills the
52/// buffer and returns the byte count, `0` at EOF, or a negative `AVERROR`.
53pub(crate) type ReadCallback = Box<dyn FnMut(&mut [u8]) -> i32 + Send>;
54
55/// Prepared custom-IO callbacks plus the sanitized filename for FFmpeg.
56pub(crate) struct PreparedHttpInput {
57    pub read: ReadCallback,
58    pub seek: Box<dyn FnMut(i64, i32) -> i64 + Send>,
59    pub display_url: CString,
60    pub seekable: bool,
61    pub io_buffer_size: usize,
62    pub failure: Arc<Mutex<Option<HttpInputError>>>,
63}
64
65/// Extra state installed on [`Input`] after [`prepare_for_open`].
66pub(crate) struct HttpAvioAttach {
67    pub display_url: CString,
68    pub seekable: bool,
69    pub failure: Arc<Mutex<Option<HttpInputError>>>,
70}
71
72/// One explicit HTTP(S) media input.
73///
74/// Convert to [`Input`] with [`From`] and pass it to
75/// [`FfmpegContext::builder`](crate::FfmpegContext::builder). Existing
76/// `Input` setters (`set_format`, codec options, …) apply after the
77/// conversion. Conversion sets `exit_on_error` so a truncated body or
78/// `ResourceChanged` fails the job instead of finishing as a short read.
79///
80/// Share a connection pool with [`HttpClient::input`]. Reconnect is off by
81/// default (FFmpeg `reconnect=0`); for seekable VOD resume use
82/// [`ReconnectPolicy::seekable_default`].
83#[derive(Clone)]
84pub struct HttpInput {
85    url: reqwest::Url,
86    client: Option<HttpClient>,
87    extra_headers: Vec<(String, String)>,
88    user_agent: Option<String>,
89    disable_ua: bool,
90    timeouts: HttpTimeouts,
91    reconnect: ReconnectPolicy,
92    io_buffer_size: usize,
93}
94
95impl fmt::Debug for HttpInput {
96    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
97        f.debug_struct("HttpInput")
98            .field("url", &urlutil::redact_url(&self.url))
99            .finish_non_exhaustive()
100    }
101}
102
103impl HttpInput {
104    /// Start a one-shot builder that creates an exclusive [`HttpClient`].
105    pub fn builder(url: impl Into<String>) -> HttpInputBuilder {
106        HttpInputBuilder::new(url.into())
107    }
108}
109
110/// Builder for [`HttpInput`].
111pub struct HttpInputBuilder {
112    url: String,
113    client: Option<HttpClient>,
114    extra_headers: Vec<(String, String)>,
115    user_agent: Option<String>,
116    disable_ua: bool,
117    timeouts: HttpTimeouts,
118    reconnect: ReconnectPolicy,
119    io_buffer_size: usize,
120}
121
122impl fmt::Debug for HttpInputBuilder {
123    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
124        let header_names: Vec<&str> = self
125            .extra_headers
126            .iter()
127            .map(|(name, _)| name.as_str())
128            .collect();
129        let url = parse_input_url(&self.url)
130            .map(|u| urlutil::redact_url(&u))
131            .unwrap_or_else(|_| "<invalid-url>".into());
132        f.debug_struct("HttpInputBuilder")
133            .field("url", &url)
134            .field("headers", &header_names)
135            .finish_non_exhaustive()
136    }
137}
138
139impl HttpInputBuilder {
140    pub(crate) fn new(url: String) -> Self {
141        Self {
142            url,
143            client: None,
144            extra_headers: Vec::new(),
145            user_agent: None,
146            disable_ua: false,
147            timeouts: HttpTimeouts::default(),
148            reconnect: ReconnectPolicy::default(),
149            io_buffer_size: crate::core::context::DEFAULT_CUSTOM_IO_BUFFER_SIZE,
150        }
151    }
152
153    pub(crate) fn client(mut self, client: HttpClient) -> Self {
154        self.client = Some(client);
155        self
156    }
157
158    /// Extra origin header. Reserved names (`Range`, `Accept-Encoding`, …) fail.
159    pub fn header(
160        mut self,
161        name: impl Into<String>,
162        value: impl Into<String>,
163    ) -> Result<Self, HttpInputError> {
164        let name = name.into();
165        let value = value.into();
166        config::validate_header(&name, &value)?;
167        self.extra_headers.push((name, value));
168        Ok(self)
169    }
170
171    /// Override User-Agent for this input.
172    pub fn user_agent(mut self, ua: impl Into<String>) -> Self {
173        self.user_agent = Some(ua.into());
174        self.disable_ua = false;
175        self
176    }
177
178    /// Send no User-Agent on this input.
179    pub fn disable_user_agent(mut self) -> Self {
180        self.disable_ua = true;
181        self.user_agent = None;
182        self
183    }
184
185    /// Per-input response-header and body-idle timeouts. Zero durations are
186    /// rejected. Connect timeout is a client-level setting: exclusive builders
187    /// apply it when the hidden client is created; a shared [`HttpClient`]
188    /// keeps the connect budget from [`HttpClientBuilder::timeouts`].
189    pub fn timeouts(mut self, timeouts: HttpTimeouts) -> Self {
190        self.timeouts = timeouts;
191        self
192    }
193
194    /// Body idle timeout convenience (see [`HttpTimeouts::read_idle`]).
195    pub fn read_idle_timeout(mut self, idle: Duration) -> Self {
196        self.timeouts.read_idle = Some(idle);
197        self
198    }
199
200    /// Reconnect policy. Default is off (FFmpeg `reconnect=0`).
201    ///
202    /// For a seekable file, [`ReconnectPolicy::seekable_default`] enables a
203    /// conservative retry budget and still requires an ETag or Last-Modified
204    /// unless `require_validator` is set false. For a non-seekable live
205    /// stream of unknown length, use [`ReconnectPolicy::streamed_default`]
206    /// (a hand-built policy with only `reconnect_streamed` set never
207    /// matches a resumable case and is a no-op).
208    pub fn reconnect(mut self, policy: ReconnectPolicy) -> Self {
209        self.reconnect = policy;
210        self
211    }
212
213    /// AVIO buffer size (same contract as [`Input::set_io_buffer_size`](crate::Input::set_io_buffer_size)).
214    ///
215    /// This sizes FFmpeg's read buffer only. The network read-ahead between
216    /// the HTTP worker and AVIO is a fixed bounded queue (8 events of at
217    /// most 64 KiB each, about 512 KiB) and is not configurable.
218    pub fn io_buffer_size(mut self, size: usize) -> Self {
219        self.io_buffer_size = size;
220        self
221    }
222
223    /// Validate configuration. No network I/O.
224    pub fn build(self) -> Result<HttpInput, HttpInputError> {
225        self.timeouts.validate()?;
226        config::validate_header_set(&self.extra_headers)?;
227        let url = parse_input_url(&self.url)?;
228        if let Some(kind) = sniff::manifest_from_url(url.as_str()) {
229            return Err(HttpInputError::ManifestUnsupported { kind });
230        }
231        if self.io_buffer_size == 0 || self.io_buffer_size > i32::MAX as usize {
232            return Err(HttpInputError::InvalidUrl {
233                reason: "io_buffer_size must be in 1..=i32::MAX",
234            });
235        }
236        Ok(HttpInput {
237            url,
238            client: self.client,
239            extra_headers: self.extra_headers,
240            user_agent: self.user_agent,
241            disable_ua: self.disable_ua,
242            timeouts: self.timeouts,
243            reconnect: self.reconnect,
244            io_buffer_size: self.io_buffer_size,
245        })
246    }
247}
248
249impl From<HttpInput> for Input {
250    fn from(http: HttpInput) -> Self {
251        let io_buffer_size = http.io_buffer_size;
252        let mut input = Input::from(String::new());
253        input.url = None;
254        input.http_input = Some(http);
255        input.io_buffer_size = io_buffer_size;
256        // TruncatedBody / ResourceChanged map to EIO. FFmpeg's default
257        // exit_on_error=0 would otherwise finish the job after demux has
258        // started. Callers can still `.set_exit_on_error(false)` after this.
259        input.exit_on_error = Some(true);
260        input
261    }
262}
263
264pub(crate) fn attach_http_input(
265    input: &mut Input,
266    interrupt: &Arc<InterruptState>,
267) -> crate::error::Result<()> {
268    if let Some(format) = input.format.as_deref() {
269        if let Some(kind) = sniff::manifest_from_format(format) {
270            return Err(HttpInputError::ManifestUnsupported { kind }.into());
271        }
272    }
273    let http = input
274        .http_input
275        .take()
276        .expect("attach_http_input without http_input");
277    let prepared = prepare_for_open(http, interrupt)?;
278    input.read_callback = Some(prepared.read);
279    input.seek_callback = Some(prepared.seek);
280    input.io_buffer_size = prepared.io_buffer_size;
281    input.http_avio = Some(HttpAvioAttach {
282        display_url: prepared.display_url,
283        seekable: prepared.seekable,
284        failure: prepared.failure,
285    });
286    Ok(())
287}
288
289pub(crate) fn prepare_for_open(
290    input: HttpInput,
291    interrupt: &Arc<InterruptState>,
292) -> Result<PreparedHttpInput, HttpInputError> {
293    let client = match input.client {
294        Some(client) => client,
295        None => exclusive_client(
296            input.timeouts.clone(),
297            input.user_agent.clone(),
298            input.disable_ua,
299        )?,
300    };
301    let runtime = client.runtime()?;
302    let cancel = Arc::new(AtomicBool::new(false));
303    let (event_tx, event_rx) = bounded(CHANNEL_CAP);
304    let (reply_tx, reply_rx) = std::sync::mpsc::channel();
305    let user_agent = if input.disable_ua {
306        None
307    } else {
308        input
309            .user_agent
310            .clone()
311            .or_else(|| client.inner.user_agent.clone())
312    };
313    let spec = RequestSpec {
314        url: input.url.clone(),
315        extra_headers: input.extra_headers.clone(),
316        user_agent,
317        header_timeout: input.timeouts.response_headers,
318        read_idle: input.timeouts.read_idle,
319        redirect_limit: client.inner.redirect_limit,
320        range_start: 0,
321        send_range: true,
322        if_range: None,
323        generation: 0,
324    };
325    // Shared with every StreamJob of this input: the opening response may
326    // omit ETag / Last-Modified while a later 206 window carries one, and a
327    // later seek needs that learned validator for its If-Range.
328    let learned_validator: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
329    runtime.submit(StreamJob {
330        client: client.inner.client.clone(),
331        spec: spec.clone(),
332        event_tx,
333        reply_tx: Mutex::new(Some(reply_tx)),
334        cancel: Arc::clone(&cancel),
335        reconnect: input.reconnect.clone(),
336        prior: None,
337        learned_validator: Arc::clone(&learned_validator),
338    })?;
339    let meta = wait_reply(&reply_rx, interrupt, &cancel)?;
340    if let Some(kind) = sniff::sniff_manifest(
341        meta.final_url.as_str(),
342        meta.content_type.as_deref(),
343        &meta.prefix,
344    ) {
345        cancel.store(true, Ordering::Relaxed);
346        return Err(HttpInputError::ManifestUnsupported { kind });
347    }
348    let mut spec = spec;
349    spec.url = meta.final_url.clone();
350    spec.if_range = meta.validator.clone();
351    stream::drop_cross_origin_secrets(&mut spec, &input.url);
352    let cancel = Arc::clone(&cancel);
353    let failure = Arc::new(Mutex::new(None));
354    let state = Arc::new(Mutex::new(http_avio::new_state(
355        event_rx,
356        if meta.prefix.is_empty() {
357            None
358        } else {
359            Some(meta.prefix)
360        },
361        meta.size,
362        meta.seekable,
363        cancel,
364        client,
365        runtime,
366        spec,
367        Arc::clone(interrupt),
368        input.reconnect,
369        Arc::clone(&failure),
370        learned_validator,
371    )));
372    let read_state = Arc::clone(&state);
373    let seek_state = Arc::clone(&state);
374    let display = urlutil::sanitized_display_url(&meta.final_url);
375    let display_url = CString::new(display)
376        .unwrap_or_else(|_| CString::new("https://http-input.invalid/resource").expect("static"));
377    Ok(PreparedHttpInput {
378        read: Box::new(move |buf| http_avio::read(&read_state, buf)),
379        seek: Box::new(move |offset, whence| http_avio::seek(&seek_state, offset, whence)),
380        display_url,
381        seekable: meta.seekable,
382        io_buffer_size: input.io_buffer_size,
383        failure,
384    })
385}
386
387pub(crate) unsafe fn reject_manifest_demuxer(
388    ctx: *mut ffmpeg_sys_next::AVFormatContext,
389) -> Result<(), HttpInputError> {
390    if ctx.is_null() || (*ctx).iformat.is_null() {
391        return Ok(());
392    }
393    let name = std::ffi::CStr::from_ptr((*(*ctx).iformat).name).to_string_lossy();
394    if name
395        .split(',')
396        .any(|part| part == "hls" || part == "applehttp")
397    {
398        return Err(HttpInputError::ManifestUnsupported {
399            kind: ManifestKind::Hls,
400        });
401    }
402    if name.split(',').any(|part| part == "dash") {
403        return Err(HttpInputError::ManifestUnsupported {
404            kind: ManifestKind::Dash,
405        });
406    }
407    Ok(())
408}
409
410#[cfg(test)]
411mod tests {
412    use super::*;
413
414    #[test]
415    fn builder_rejects_hls_url() {
416        let err = HttpInput::builder("https://example.com/live.m3u8")
417            .build()
418            .unwrap_err();
419        assert!(
420            matches!(
421                err,
422                HttpInputError::ManifestUnsupported {
423                    kind: ManifestKind::Hls
424                }
425            ),
426            "{err}"
427        );
428        let msg = err.to_string();
429        assert!(msg.contains("HLS manifests"), "{msg}");
430        assert!(msg.contains("single media stream"), "{msg}");
431    }
432
433    #[test]
434    fn builder_rejects_dash_url() {
435        let err = HttpInput::builder("https://example.com/manifest.mpd")
436            .build()
437            .unwrap_err();
438        assert!(matches!(
439            err,
440            HttpInputError::ManifestUnsupported {
441                kind: ManifestKind::Dash
442            }
443        ));
444        let msg = err.to_string();
445        assert!(msg.contains("DASH manifests"), "{msg}");
446    }
447
448    #[test]
449    fn builder_rejects_userinfo() {
450        let err = HttpInput::builder("https://u:p@example.com/v.mp4")
451            .build()
452            .unwrap_err();
453        assert!(matches!(err, HttpInputError::UserinfoForbidden));
454    }
455
456    #[test]
457    fn builder_rejects_reserved_header() {
458        let err = HttpInput::builder("https://example.com/v.mp4")
459            .header("Accept-Encoding", "gzip")
460            .unwrap_err();
461        assert!(matches!(err, HttpInputError::HeaderReserved { .. }));
462    }
463
464    #[test]
465    fn from_http_input_does_not_set_url() {
466        let http = HttpInput::builder("https://example.com/v.mp4")
467            .build()
468            .unwrap();
469        let input = Input::from(http);
470        assert!(input.url.is_none());
471        assert!(input.http_input.is_some());
472        assert!(input.read_callback.is_none());
473        assert_eq!(input.exit_on_error, Some(true));
474    }
475
476    #[test]
477    fn url_from_does_not_install_http_input() {
478        let input = Input::from("https://example.com/v.mp4");
479        assert_eq!(input.url.as_deref(), Some("https://example.com/v.mp4"));
480        assert!(input.http_input.is_none());
481        assert!(input.read_callback.is_none());
482    }
483
484    #[test]
485    fn shared_client_input_inherits_timeouts() {
486        let timeouts = HttpTimeouts {
487            connect: Duration::from_secs(3),
488            response_headers: Duration::from_secs(4),
489            read_idle: Some(Duration::from_secs(5)),
490        };
491        let client = HttpClient::builder().timeouts(timeouts).build().unwrap();
492        let http = client.input("https://example.com/v.mp4").build().unwrap();
493        assert_eq!(http.timeouts.connect, Duration::from_secs(3));
494        assert_eq!(http.timeouts.response_headers, Duration::from_secs(4));
495        assert_eq!(http.timeouts.read_idle, Some(Duration::from_secs(5)));
496    }
497}
ez_ffmpeg/http_input/mod.rs

ez_ffmpeg/http_input/
mod.rs
