diff --git a/backend/src/api/model/metadata_update_manager.rs b/backend/src/api/model/metadata_update_manager.rs index 27e415bca..1b1b43610 100644 --- a/backend/src/api/model/metadata_update_manager.rs +++ b/backend/src/api/model/metadata_update_manager.rs @@ -3484,6 +3484,7 @@ impl InputWorker { match update_live_stream_metadata( &app_state.app_config, + client, input, id.clone(), false, @@ -3512,6 +3513,7 @@ impl InputWorker { let outcome = update_generic_stream_metadata( &app_state.app_config, + client, input.as_ref(), unique_id, url, diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index c26be46e3..279bb74b1 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -27,7 +27,8 @@ use crate::{ PlaylistSource, }, utils::{ - debug_if_enabled, epg, log_memory_snapshot, m3u, trace_if_enabled, xtream, StepMeasure, StepMeasureCallback, + debug_if_enabled, epg, log_memory_snapshot, m3u, trace_if_enabled, xtream, + StepMeasure, StepMeasureCallback, }, }; use futures::{FutureExt, StreamExt}; diff --git a/backend/src/processing/processor/stream_probe.rs b/backend/src/processing/processor/stream_probe.rs index 380503910..e3f6f83d6 100644 --- a/backend/src/processing/processor/stream_probe.rs +++ b/backend/src/processing/processor/stream_probe.rs @@ -4,8 +4,9 @@ use crate::model::{AppConfig}; use crate::processing::processor::{select_cancel_token, ProbeHandleGuard}; use crate::repository::{get_input_m3u_playlist_file_path, get_input_storage_path, get_input_local_library_playlist_file_path, xtream_get_file_path, BPlusTreeUpdate}; use crate::utils::debug_if_enabled; -use crate::utils::ffmpeg::{FfmpegExecutor, ProbeFailureKind, ProbeStreamStats, ProbeUrlOutcome}; -use log::{info, warn}; +use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeStreamStats, ProbeUrlOutcome}; +use shared::utils::sanitize_sensitive_info; +use log::{debug, info, warn}; use shared::error::TuliproxError; use shared::model::{EpisodeStreamProperties, InputType, PlaylistItemType, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, LiveStreamProperties, M3uPlaylistItem, XtreamCluster, XtreamPlaylistItem}; use std::sync::Arc; @@ -34,6 +35,7 @@ fn requires_provider_connection_for_generic_probe(input_type: InputType) -> bool #[allow(clippy::too_many_arguments, clippy::too_many_lines)] pub async fn update_generic_stream_metadata( app_config: &Arc, + client: &reqwest::Client, input: &ConfigInput, unique_id: &str, stream_url: &str, @@ -106,6 +108,14 @@ pub async fn update_generic_stream_metadata( } let probe_url = input.resolve_url(stream_url)?.into_owned(); + if !is_supported_probe_url(&probe_url) { + let safe_probe_url = sanitize_sensitive_info(&probe_url).into_owned(); + debug!("Skipping unsupported generic stream probe for {unique_id}: {safe_probe_url}"); + return Ok(GenericProbeOutcome::Noop); + } + let is_remote_probe = reqwest::Url::parse(&probe_url) + .ok() + .is_some_and(|url| matches!(url.scheme(), "http" | "https")); let config = app_config.config.load(); let metadata_update = config.metadata_update.clone().unwrap_or_default(); let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60); @@ -128,16 +138,31 @@ pub async fn update_generic_stream_metadata( acquired_handle.as_ref().and_then(ProbeHandleGuard::handle), active_handle, ); - let probe_data = FfmpegExecutor::new().probe_url_with_cancel( - &probe_url, - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - config.proxy.as_ref(), - cancel_token, - ) - .await; + let probe_data = if is_remote_probe { + FfmpegExecutor::new() + .probe_remote_url_with_cancel( + client, + &probe_url, + user_agent.as_deref(), + analyze_duration, + probe_size, + ffprobe_timeout, + cancel_token, + ) + .await + } else { + FfmpegExecutor::new() + .probe_url_with_cancel( + &probe_url, + user_agent.as_deref(), + analyze_duration, + probe_size, + ffprobe_timeout, + config.proxy.as_ref(), + cancel_token, + ) + .await + }; if let Some(handle) = acquired_handle { handle.release().await; diff --git a/backend/src/processing/processor/xtream.rs b/backend/src/processing/processor/xtream.rs index ac78df2fb..377805026 100644 --- a/backend/src/processing/processor/xtream.rs +++ b/backend/src/processing/processor/xtream.rs @@ -5,15 +5,16 @@ use crate::model::{AppConfig, ConfigInput, ConfigInputFlags}; use shared::model::{LiveStreamProperties, StreamProperties, XtreamCluster, XtreamPlaylistItem}; use crate::repository::{get_input_storage_path, persist_input_live_info, BPlusTreeQuery, xtream_get_file_path}; use crate::utils::{debug_if_enabled}; -use crate::utils::ffmpeg::{FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome}; +use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome}; use log::{debug, warn}; use crate::processing::parser::xtream::create_xtream_url; use crate::api::model::{ActiveProviderManager, ProviderHandle, ProviderIdType}; /// Updates metadata for a single Live stream (primarily probing) -#[allow(clippy::too_many_lines)] +#[allow(clippy::too_many_arguments, clippy::too_many_lines)] pub async fn update_live_stream_metadata( app_config: &Arc, + client: &reqwest::Client, input: &ConfigInput, id: ProviderIdType, save: bool, @@ -117,6 +118,14 @@ pub async fn update_live_stream_metadata( use_prefix, no_ext ); let probe_url_cow = input.resolve_url(&stream_url)?; + if !is_supported_probe_url(probe_url_cow.as_ref()) { + debug!( + "Skipping unsupported live probe for input {}: {}", + input.name, + probe_url_cow.as_ref() + ); + return Ok(None); + } let config = app_config.config.load(); let metadata_update = config.metadata_update.clone().unwrap_or_default(); let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60); @@ -134,16 +143,31 @@ pub async fn update_live_stream_metadata( let mut success = false; let mut not_found = false; - match FfmpegExecutor::new().probe_url( - probe_url_cow.as_ref(), - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - config.proxy.as_ref(), - ) - .await - { + let is_remote_probe = reqwest::Url::parse(probe_url_cow.as_ref()) + .ok() + .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + let probe_result = if is_remote_probe { + FfmpegExecutor::new().probe_remote_url( + client, + probe_url_cow.as_ref(), + user_agent.as_deref(), + analyze_duration, + probe_size, + ffprobe_timeout, + ) + .await + } else { + FfmpegExecutor::new().probe_url( + probe_url_cow.as_ref(), + user_agent.as_deref(), + analyze_duration, + probe_size, + ffprobe_timeout, + config.proxy.as_ref(), + ) + .await + }; + match probe_result { ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, _stats) => { // 3. Update properties on success if let Some(v) = raw_video { diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index c4e3c2645..908767cd3 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -19,7 +19,7 @@ use crate::repository::{ get_input_storage_path, persist_input_series_info_batch, MemoryPlaylistSource, PlaylistSource, }; use crate::repository::{xtream_get_file_path, BPlusTreeQuery}; -use crate::utils::ffmpeg::{FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome}; +use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome}; use crate::utils::{debug_if_enabled, xtream}; use log::{debug, error, info, log_enabled, trace, warn, Level}; use parking_lot::Mutex; @@ -928,6 +928,46 @@ pub async fn update_series_metadata( if active_handle.is_some() || temp_handle.is_some() { let episode_url = create_xtream_series_episode_url(input_url, input_username, input_password, ep); + let probe_url = match input.resolve_url(&episode_url) { + Ok(url) => url, + Err(err) => { + if matches!(err, shared::error::TuliproxError::ConfigInput(_)) { + debug!( + "Provider config resolution failed for series episode '{}' (S{}E{}): {err}", + ep.title, ep.season, ep.episode_num + ); + if probe_failure.is_none() { + probe_failure = Some(ProbeFailureKind::Other); + } + } else { + warn!( + "Failed to resolve probe URL for series episode '{}' (S{}E{}): {err}", + ep.title, ep.season, ep.episode_num + ); + probe_pending = true; + if probe_failure.is_none() { + probe_failure = Some(ProbeFailureKind::Other); + } + } + if let Some(guard) = temp_handle { + guard.release().await; + } + continue; + } + }; + if !is_supported_probe_url(probe_url.as_ref()) { + debug!( + "Skipping unsupported series probe for episode '{}' (S{}E{}): {}", + ep.title, + ep.season, + ep.episode_num, + probe_url.as_ref() + ); + if let Some(guard) = temp_handle { + guard.release().await; + } + continue; + } // Specific logging for the user to follow let missing_reason = if missing_video && missing_audio { @@ -946,17 +986,33 @@ pub async fn update_series_metadata( temp_handle.as_ref().and_then(ProbeHandleGuard::handle), active_handle, ); - match FfmpegExecutor::new().probe_url_with_cancel( - &episode_url, - user_agent.as_deref(), - probe_settings.analyze_duration_micros, - probe_settings.probe_size_bytes, - probe_settings.timeout_secs, - config.proxy.as_ref(), - cancel_token, - ) - .await - { + let is_remote_probe = reqwest::Url::parse(probe_url.as_ref()) + .ok() + .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + let probe_result = if is_remote_probe { + FfmpegExecutor::new().probe_remote_url_with_cancel( + client, + probe_url.as_ref(), + user_agent.as_deref(), + probe_settings.analyze_duration_micros, + probe_settings.probe_size_bytes, + probe_settings.timeout_secs, + cancel_token, + ) + .await + } else { + FfmpegExecutor::new().probe_url_with_cancel( + probe_url.as_ref(), + user_agent.as_deref(), + probe_settings.analyze_duration_micros, + probe_settings.probe_size_bytes, + probe_settings.timeout_secs, + config.proxy.as_ref(), + cancel_token, + ) + .await + }; + match probe_result { ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, _stats) => { if let Some(v) = raw_video { ep.video = Some(v.to_string().into()); diff --git a/backend/src/processing/processor/xtream_vod.rs b/backend/src/processing/processor/xtream_vod.rs index f2e1dd51c..a9f45e89e 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -17,7 +17,7 @@ use crate::ptt::ptt_parse_title; use crate::repository::persist_input_vod_info; use crate::repository::persist_input_vod_info_batch; use crate::repository::{xtream_get_file_path, BPlusTreeQuery}; -use crate::utils::ffmpeg::{FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome}; +use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome}; use crate::utils::{debug_if_enabled, trace_if_enabled, xtream}; use log::{debug, error, info, log_enabled, trace, warn, Level}; use parking_lot::Mutex; @@ -837,86 +837,124 @@ pub async fn update_vod_metadata( true, true, ); + let probe_url = input.resolve_url(&stream_url).map_err(|err| { + if matches!(err, shared::error::TuliproxError::ConfigInput(_)) { + shared::error::TuliproxError::ConfigInput(format!( + "Provider config resolution failed for VOD probe URL '{stream_url}': {err}" + )) + } else { + shared::error::TuliproxError::Config(format!( + "Failed to resolve probe URL for VOD {display_id}: {err}" + )) + } + })?; - let config = app_config.config.load(); - let metadata_update = config.metadata_update.clone().unwrap_or_default(); - let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60); - let user_agent = config.default_user_agent.clone(); - let analyze_duration = metadata_update.ffprobe.analyze_duration_micros; - let probe_size = metadata_update.ffprobe.probe_size_bytes; + if is_supported_probe_url(probe_url.as_ref()) { - let probe_priority = config.metadata_update.as_ref().map_or(default_probe_user_priority(), |cfg| cfg.probe.user_priority); + let config = app_config.config.load(); + let metadata_update = config.metadata_update.clone().unwrap_or_default(); + let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60); + let user_agent = config.default_user_agent.clone(); + let analyze_duration = metadata_update.ffprobe.analyze_duration_micros; + let probe_size = metadata_update.ffprobe.probe_size_bytes; - // Acquire Connection logic - let temp_handle = if active_handle.is_some() { - None - } else { - active_provider - .acquire_connection_for_probe(&input.name, probe_priority) - .await - .map(|handle| ProbeHandleGuard::new(active_provider, handle)) - }; + let probe_priority = config.metadata_update.as_ref().map_or(default_probe_user_priority(), |cfg| cfg.probe.user_priority); - if active_handle.is_some() || temp_handle.is_some() { - debug_if_enabled!("Probing VOD '{}' (ID: {})", display_title, display_id); - let cancel_token = select_cancel_token( - temp_handle.as_ref().and_then(ProbeHandleGuard::handle), - active_handle, - ); - match FfmpegExecutor::new().probe_url_with_cancel( - &stream_url, - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - config.proxy.as_ref(), - cancel_token, - ) - .await - { - ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, stats) => { - if let Some(details) = properties.details.as_mut() { - if let Some(v) = raw_video { - details.video = Some(v.to_string().into()); - properties_updated = true; + // Acquire Connection logic + let temp_handle = if active_handle.is_some() { + None + } else { + active_provider + .acquire_connection_for_probe(&input.name, probe_priority) + .await + .map(|handle| ProbeHandleGuard::new(active_provider, handle)) + }; + + if active_handle.is_some() || temp_handle.is_some() { + debug_if_enabled!("Probing VOD '{}' (ID: {})", display_title, display_id); + let cancel_token = select_cancel_token( + temp_handle.as_ref().and_then(ProbeHandleGuard::handle), + active_handle, + ); + let is_remote_probe = reqwest::Url::parse(probe_url.as_ref()) + .ok() + .is_some_and(|u| matches!(u.scheme(), "http" | "https")); + let probe_result = if is_remote_probe { + FfmpegExecutor::new().probe_remote_url_with_cancel( + client, + probe_url.as_ref(), + user_agent.as_deref(), + analyze_duration, + probe_size, + ffprobe_timeout, + cancel_token, + ) + .await + } else { + FfmpegExecutor::new().probe_url_with_cancel( + probe_url.as_ref(), + user_agent.as_deref(), + analyze_duration, + probe_size, + ffprobe_timeout, + config.proxy.as_ref(), + cancel_token, + ) + .await + }; + match probe_result { + ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, stats) => { + if let Some(details) = properties.details.as_mut() { + if let Some(v) = raw_video { + details.video = Some(v.to_string().into()); + properties_updated = true; + } + if let Some(a) = raw_audio { + details.audio = Some(a.to_string().into()); + properties_updated = true; + } + if let Some(duration_secs) = stats.duration_secs { + details.duration_secs = Some(duration_secs.to_string().into()); + properties_updated = true; + } + if let Some(bitrate) = stats.bitrate { + details.bitrate = bitrate; + properties_updated = true; + } } - if let Some(a) = raw_audio { - details.audio = Some(a.to_string().into()); - properties_updated = true; + } + ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => { + probe_failure = Some(ProbeFailureKind::NotFound); + } + ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => { + probe_pending = true; + if probe_failure.is_none() { + probe_failure = Some(ProbeFailureKind::Other); } - if let Some(duration_secs) = stats.duration_secs { - details.duration_secs = Some(duration_secs.to_string().into()); - properties_updated = true; - } - if let Some(bitrate) = stats.bitrate { - details.bitrate = bitrate; - properties_updated = true; + } + ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => { + probe_pending = true; + if probe_failure.is_none() { + probe_failure = Some(ProbeFailureKind::Cancelled); } } } - ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => { - probe_failure = Some(ProbeFailureKind::NotFound); - } - ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => { - probe_pending = true; - if probe_failure.is_none() { - probe_failure = Some(ProbeFailureKind::Other); - } - } - ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => { - probe_pending = true; - if probe_failure.is_none() { - probe_failure = Some(ProbeFailureKind::Cancelled); - } - } + } else { + probe_pending = true; + warn!("Skipping probe for VOD {display_id} due to connection limits"); + } + + if let Some(guard) = temp_handle { + guard.release().await; } } else { - probe_pending = true; - warn!("Skipping probe for VOD {display_id} due to connection limits"); - } - if let Some(guard) = temp_handle { - guard.release().await; + debug_if_enabled!( + "Skipping unsupported VOD probe for '{}' (ID: {}): {}", + display_title, + display_id, + probe_url.as_ref() + ); } } } diff --git a/backend/src/utils/ffmpeg.rs b/backend/src/utils/ffmpeg.rs index fddf266d5..26410db4d 100644 --- a/backend/src/utils/ffmpeg.rs +++ b/backend/src/utils/ffmpeg.rs @@ -1,11 +1,17 @@ use crate::model::ProxyConfig; use log::{debug, warn}; +use reqwest::{ + header::{RANGE, USER_AGENT}, + Client, StatusCode, +}; use serde_json::Value; use shared::model::MediaQuality; -use shared::utils::{default_thumbnail_height, default_thumbnail_width, sanitize_sensitive_info}; +use shared::utils::{default_thumbnail_height, default_thumbnail_width, is_dash_url, is_hls_url, sanitize_sensitive_info}; use std::path::Path; +use std::io::ErrorKind; +use std::process::{Output, Stdio}; use std::time::Duration; -use tokio::process::Command; +use tokio::{io::AsyncWriteExt, process::Command}; use url::Url; const FFMPEG_TIMEOUT: Duration = Duration::from_secs(60); @@ -118,62 +124,7 @@ impl FfmpegExecutor { let output_result = tokio::time::timeout(timeout_val, command.output()).await; match output_result { - Ok(Ok(output)) => { - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - debug!("ffprobe failed for {}: {}", sanitize_sensitive_info(url), sanitize_sensitive_info(&stderr)); - if is_not_found_probe_error(&stderr) { - return ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound); - } - return ProbeUrlOutcome::Failed(ProbeFailureKind::Other); - } - - if let Ok(json) = serde_json::from_slice::(&output.stdout) { - if let Some(stream_list) = json.get("streams").and_then(Value::as_array) { - let mut video_stream: Option<&Value> = None; - let mut audio_stream: Option<&Value> = None; - - for stream in stream_list { - let codec_type = stream.get("codec_type").and_then(Value::as_str); - if video_stream.is_none() - && (codec_type == Some("video") - || (codec_type.is_none() - && (stream.get("width").is_some() || stream.get("height").is_some()))) - && !is_attached_pic(stream) - { - video_stream = Some(stream); - } else if audio_stream.is_none() - && (codec_type == Some("audio") - || (codec_type.is_none() - && (stream.get("channels").is_some() - || stream.get("channel_layout").is_some()))) - { - audio_stream = Some(stream); - } - if video_stream.is_some() && audio_stream.is_some() { - break; - } - } - - if video_stream.is_some() || audio_stream.is_some() { - let video_str = video_stream.map(Value::to_string); - let audio_str = audio_stream.map(Value::to_string); - let mq = MediaQuality::from_ffprobe_info(audio_str.as_deref(), video_str.as_deref()); - if let Some(quality) = mq { - let stats = extract_probe_stream_stats(&json, video_stream, audio_stream); - return ProbeUrlOutcome::Success( - quality, - video_stream.cloned(), - audio_stream.cloned(), - stats, - ); - } - } - } - } else { - warn!("Failed to parse ffprobe json output for {}", sanitize_sensitive_info(url)); - } - } + Ok(Ok(output)) => return parse_ffprobe_output(url, &output), Ok(Err(e)) => { warn!("ffprobe execution failed for {}: {}", sanitize_sensitive_info(url), e); } @@ -185,6 +136,38 @@ impl FfmpegExecutor { ProbeUrlOutcome::Failed(ProbeFailureKind::Other) } + #[allow(clippy::too_many_arguments)] + pub async fn probe_remote_url( + &self, + client: &Client, + url: &str, + user_agent: Option<&str>, + analyze_duration: u64, + probe_size: u64, + timeout_secs: u64, + ) -> ProbeUrlOutcome { + let analyze_overhead = Duration::from_micros(analyze_duration) + Duration::from_secs(5); + let config_timeout = Duration::from_secs(timeout_secs); + let timeout_val = std::cmp::max(analyze_overhead, config_timeout); + + let probe_result = tokio::time::timeout( + timeout_val, + self.probe_remote_url_inner(client, url, user_agent, analyze_duration, probe_size), + ) + .await; + + if let Ok(outcome) = probe_result { + outcome + } else { + warn!( + "ffprobe remote stdin probe timed out after {:?} for {}", + timeout_val, + sanitize_sensitive_info(url) + ); + ProbeUrlOutcome::Failed(ProbeFailureKind::Other) + } + } + /// Wrapper around [`Self::probe_url`] that races the probe against an optional cancellation token. #[allow(clippy::too_many_arguments)] pub async fn probe_url_with_cancel( @@ -211,6 +194,31 @@ impl FfmpegExecutor { } } + #[allow(clippy::too_many_arguments)] + pub async fn probe_remote_url_with_cancel( + &self, + client: &Client, + url: &str, + user_agent: Option<&str>, + analyze_duration: u64, + probe_size: u64, + timeout_secs: u64, + cancel_token: Option<&tokio_util::sync::CancellationToken>, + ) -> ProbeUrlOutcome { + if let Some(token) = cancel_token { + tokio::select! { + biased; + () = token.cancelled() => { + warn!("Probe preempted for {}", shared::utils::sanitize_sensitive_info(url)); + ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) + } + result = self.probe_remote_url(client, url, user_agent, analyze_duration, probe_size, timeout_secs) => result, + } + } else { + self.probe_remote_url(client, url, user_agent, analyze_duration, probe_size, timeout_secs).await + } + } + async fn check_binary_availability(&self, binary: &str) -> bool { let mut command = Command::new(binary); command @@ -237,6 +245,98 @@ impl FfmpegExecutor { .map_err(|_| format_ffmpeg_timeout_error(args))? .map_err(|e| e.to_string()) } + + async fn probe_remote_url_inner( + &self, + client: &Client, + url: &str, + user_agent: Option<&str>, + analyze_duration: u64, + probe_size: u64, + ) -> ProbeUrlOutcome { + let mut command = Command::new("ffprobe"); + command + .kill_on_drop(true) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .arg("-v") + .arg("error") + .arg("-show_streams") + .arg("-show_format") + .arg("-of") + .arg("json") + .arg("-analyzeduration") + .arg(analyze_duration.to_string()) + .arg("-probesize") + .arg(probe_size.to_string()) + .arg("-i") + .arg("pipe:0"); + + let mut child = match command.spawn() { + Ok(child) => child, + Err(err) => { + warn!("ffprobe execution failed for {}: {}", sanitize_sensitive_info(url), err); + return ProbeUrlOutcome::Failed(ProbeFailureKind::Other); + } + }; + + let Some(mut stdin) = child.stdin.take() else { + warn!("ffprobe stdin unavailable for {}", sanitize_sensitive_info(url)); + return ProbeUrlOutcome::Failed(ProbeFailureKind::Other); + }; + + let max_bytes = if probe_size == 0 { usize::MAX } else { usize::try_from(probe_size).unwrap_or(usize::MAX) }; + let fetch_result = stream_probe_bytes_to_stdin(&mut stdin, client, url, user_agent, probe_size, max_bytes).await; + + let stdin_err = match fetch_result { + Ok(()) => None, + Err(ProbeFailureKind::NotFound) => { + let _ = child.kill().await; + return ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound); + } + Err(kind) => Some(kind), + }; + + if let Err(err) = stdin.shutdown().await { + if err.kind() != ErrorKind::BrokenPipe { + debug!( + "ffprobe stdin shutdown reported {} for {}, continuing with child output", + err, + sanitize_sensitive_info(url) + ); + } + } + drop(stdin); + + match child.wait_with_output().await { + Ok(output) => { + if let Some(kind) = stdin_err { + debug!("ffprobe remote stdin probe had fetch error, checking child output for {}", sanitize_sensitive_info(url)); + let outcome = parse_ffprobe_output(url, &output); + if matches!(outcome, ProbeUrlOutcome::Success(..)) { + return outcome; + } + ProbeUrlOutcome::Failed(kind) + } else { + parse_ffprobe_output(url, &output) + } + } + Err(err) => { + warn!("ffprobe execution failed for {}: {}", sanitize_sensitive_info(url), err); + ProbeUrlOutcome::Failed(stdin_err.unwrap_or(ProbeFailureKind::Other)) + } + } + } +} + +#[must_use] +pub fn is_supported_probe_url(url: &str) -> bool { + // TODO: HLS manifests are intentionally unsupported for ffprobe-based metadata extraction for now. + // Probing them correctly requires explicit variant selection and base-url aware follow-up fetching, + // which we do not model yet. Revisit this once HLS probing semantics are defined. + // DASH manifests are also excluded for the same reason. + !is_hls_url(url) && !is_dash_url(url) } fn format_ffmpeg_timeout_error(args: &[String]) -> String { @@ -247,6 +347,200 @@ fn format_ffmpeg_timeout_error(args: &[String]) -> String { ) } +async fn stream_probe_bytes_to_stdin( + stdin: &mut W, + client: &Client, + url: &str, + user_agent: Option<&str>, + probe_size_header: u64, + max_bytes: usize, +) -> Result<(), ProbeFailureKind> +where + W: tokio::io::AsyncWrite + Unpin, +{ + let mut request = client.get(url); + if let Some(ua) = user_agent { + request = request.header(USER_AGENT, ua); + } + if probe_size_header > 0 { + request = request.header(RANGE, format!("bytes=0-{}", probe_size_header.saturating_sub(1))); + } + + let mut response = match request.send().await { + Ok(response) => response, + Err(err) => { + warn!("ffprobe fetch failed for {}: {}", sanitize_sensitive_info(url), err); + return Err(ProbeFailureKind::Other); + } + }; + + if response.status() == StatusCode::NOT_FOUND { + return Err(ProbeFailureKind::NotFound); + } + if !response.status().is_success() && response.status() != StatusCode::PARTIAL_CONTENT { + warn!( + "ffprobe fetch returned {} for {}", + response.status(), + sanitize_sensitive_info(url) + ); + return Err(ProbeFailureKind::Other); + } + + let mut total_written: usize = 0; + loop { + let chunk = match response.chunk().await { + Ok(Some(chunk)) => chunk, + Ok(None) => break, + Err(err) => { + warn!("ffprobe fetch body failed for {}: {}", sanitize_sensitive_info(url), err); + return Err(ProbeFailureKind::Other); + } + }; + + let remaining = max_bytes.saturating_sub(total_written); + let to_write = if chunk.len() <= remaining { + chunk.len() + } else { + remaining + }; + + if to_write > 0 { + match stdin.write_all(&chunk[..to_write]).await { + Ok(()) => {} + Err(err) if err.kind() == ErrorKind::BrokenPipe => { + debug!( + "ffprobe closed stdin early for {}, stopping fetch", + sanitize_sensitive_info(url) + ); + return Ok(()); + } + Err(err) => { + warn!("ffprobe stdin write failed for {}: {}", sanitize_sensitive_info(url), err); + return Err(ProbeFailureKind::Other); + } + } + total_written += to_write; + } + + if total_written >= max_bytes { + break; + } + } + + Ok(()) +} + +#[cfg(test)] +async fn fetch_probe_bytes( + client: &Client, + url: &str, + user_agent: Option<&str>, + probe_size: u64, +) -> Result, ProbeFailureKind> { + let max_bytes = if probe_size == 0 { usize::MAX } else { usize::try_from(probe_size).unwrap_or(usize::MAX) }; + let mut request = client.get(url); + if let Some(ua) = user_agent { + request = request.header(USER_AGENT, ua); + } + if probe_size > 0 { + request = request.header(RANGE, format!("bytes=0-{}", probe_size.saturating_sub(1))); + } + + let mut response = match request.send().await { + Ok(response) => response, + Err(err) => { + warn!("ffprobe fetch failed for {}: {}", sanitize_sensitive_info(url), err); + return Err(ProbeFailureKind::Other); + } + }; + + if response.status() == StatusCode::NOT_FOUND { + return Err(ProbeFailureKind::NotFound); + } + if !response.status().is_success() && response.status() != StatusCode::PARTIAL_CONTENT { + warn!( + "ffprobe fetch returned {} for {}", + response.status(), + sanitize_sensitive_info(url) + ); + return Err(ProbeFailureKind::Other); + } + + let mut bytes = Vec::with_capacity(max_bytes.min(64 * 1024)); + while bytes.len() < max_bytes { + let chunk = match response.chunk().await { + Ok(Some(chunk)) => chunk, + Ok(None) => break, + Err(err) => { + warn!("ffprobe fetch body failed for {}: {}", sanitize_sensitive_info(url), err); + return Err(ProbeFailureKind::Other); + } + }; + + let remaining = max_bytes.saturating_sub(bytes.len()); + if chunk.len() <= remaining { + bytes.extend_from_slice(&chunk); + } else { + bytes.extend_from_slice(&chunk[..remaining]); + break; + } + } + + Ok(bytes) +} + +#[cfg(test)] +async fn write_probe_bytes_to_stdin(stdin: &mut W, probe_bytes: &[u8], url: &str) -> Result<(), ProbeFailureKind> +where + W: tokio::io::AsyncWrite + Unpin, +{ + match stdin.write_all(probe_bytes).await { + Ok(()) => Ok(()), + Err(err) if err.kind() == ErrorKind::BrokenPipe => { + debug!( + "ffprobe closed stdin early for {}, continuing with child output", + sanitize_sensitive_info(url) + ); + Ok(()) + } + Err(err) => { + warn!("ffprobe stdin write failed for {}: {}", sanitize_sensitive_info(url), err); + Err(ProbeFailureKind::Other) + } + } +} + +#[cfg(test)] +async fn run_ffprobe_with_stdin(mut child: tokio::process::Child, probe_bytes: &[u8], url: &str) -> ProbeUrlOutcome { + let Some(mut stdin) = child.stdin.take() else { + warn!("ffprobe stdin unavailable for {}", sanitize_sensitive_info(url)); + return ProbeUrlOutcome::Failed(ProbeFailureKind::Other); + }; + + if let Err(kind) = write_probe_bytes_to_stdin(&mut stdin, probe_bytes, url).await { + return ProbeUrlOutcome::Failed(kind); + } + + if let Err(err) = stdin.shutdown().await { + if err.kind() != ErrorKind::BrokenPipe { + debug!( + "ffprobe stdin shutdown reported {} for {}, continuing with child output", + err, + sanitize_sensitive_info(url) + ); + } + } + drop(stdin); + + match child.wait_with_output().await { + Ok(output) => parse_ffprobe_output(url, &output), + Err(err) => { + warn!("ffprobe execution failed for {}: {}", sanitize_sensitive_info(url), err); + ProbeUrlOutcome::Failed(ProbeFailureKind::Other) + } + } +} + fn build_ffprobe_proxy_url(proxy_cfg: &ProxyConfig) -> Option { let mut proxy_url = Url::parse(proxy_cfg.url.as_str()).ok()?; if let Some(username) = proxy_cfg.username.as_deref() { @@ -285,6 +579,58 @@ fn apply_proxy_to_ffprobe(command: &mut Command, proxy_cfg: Option<&ProxyConfig> } } +fn parse_ffprobe_output(url: &str, output: &Output) -> ProbeUrlOutcome { + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + debug!("ffprobe failed for {}: {}", sanitize_sensitive_info(url), sanitize_sensitive_info(&stderr)); + if is_not_found_probe_error(&stderr) { + return ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound); + } + return ProbeUrlOutcome::Failed(ProbeFailureKind::Other); + } + + if let Ok(json) = serde_json::from_slice::(&output.stdout) { + if let Some(stream_list) = json.get("streams").and_then(Value::as_array) { + let mut video_stream: Option<&Value> = None; + let mut audio_stream: Option<&Value> = None; + + for stream in stream_list { + let codec_type = stream.get("codec_type").and_then(Value::as_str); + if video_stream.is_none() + && (codec_type == Some("video") + || (codec_type.is_none() && (stream.get("width").is_some() || stream.get("height").is_some()))) + && !is_attached_pic(stream) + { + video_stream = Some(stream); + } else if audio_stream.is_none() + && (codec_type == Some("audio") + || (codec_type.is_none() + && (stream.get("channels").is_some() || stream.get("channel_layout").is_some()))) + { + audio_stream = Some(stream); + } + if video_stream.is_some() && audio_stream.is_some() { + break; + } + } + + if video_stream.is_some() || audio_stream.is_some() { + let video_str = video_stream.map(Value::to_string); + let audio_str = audio_stream.map(Value::to_string); + let mq = MediaQuality::from_ffprobe_info(audio_str.as_deref(), video_str.as_deref()); + if let Some(quality) = mq { + let stats = extract_probe_stream_stats(&json, video_stream, audio_stream); + return ProbeUrlOutcome::Success(quality, video_stream.cloned(), audio_stream.cloned(), stats); + } + } + } + } else { + warn!("Failed to parse ffprobe json output for {}", sanitize_sensitive_info(url)); + } + + ProbeUrlOutcome::Failed(ProbeFailureKind::Other) +} + /// Returns `true` when the stream is an embedded thumbnail / cover art /// (e.g. PNG or MJPEG poster images inside MKV containers). /// ffprobe reports these as `codec_type: "video"` but with @@ -387,12 +733,93 @@ fn build_thumbnail_args(input_path: &str, output_path: &Path, scale_filter: &str mod tests { use super::{ build_ffprobe_proxy_url, build_thumbnail_args, build_thumbnail_scale_filter, extract_probe_stream_stats, - format_ffmpeg_timeout_error, FFMPEG_TIMEOUT, ProbeStreamStats, + fetch_probe_bytes, format_ffmpeg_timeout_error, run_ffprobe_with_stdin, write_probe_bytes_to_stdin, FFMPEG_TIMEOUT, + ProbeFailureKind, ProbeStreamStats, ProbeUrlOutcome, }; use crate::model::ProxyConfig; use serde_json::json; use shared::utils::{default_thumbnail_height, default_thumbnail_width}; - use std::path::Path; + use std::{io, path::Path, pin::Pin, task::{Context, Poll}}; + use std::sync::{atomic::{AtomicUsize, Ordering}, Arc, Mutex}; + use tokio::{ + io::{AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::TcpListener, + process::Command, + }; + + async fn start_test_http_server( + status_line: &str, + body: &'static [u8], + ) -> std::io::Result<(String, Arc>>, Arc, tokio::task::JoinHandle<()>)> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + let request_capture = Arc::new(Mutex::new(Vec::new())); + let accepted = Arc::new(AtomicUsize::new(0)); + let request_capture_clone = Arc::clone(&request_capture); + let accepted_clone = Arc::clone(&accepted); + let content_length = body.len(); + let status_line = status_line.to_string(); + + let handle = tokio::spawn(async move { + loop { + let Ok((mut socket, _)) = listener.accept().await else { + continue; + }; + accepted_clone.fetch_add(1, Ordering::SeqCst); + let request_capture = Arc::clone(&request_capture_clone); + let status_line = status_line.clone(); + tokio::spawn(async move { + let mut buf = vec![0_u8; 4096]; + if let Ok(read) = socket.read(&mut buf).await { + if read > 0 { + let mut guard = request_capture.lock().expect("request capture mutex poisoned"); + guard.extend_from_slice(&buf[..read]); + } + } + let response_head = format!( + "{status_line}\r\nContent-Length: {content_length}\r\nConnection: close\r\n\r\n" + ); + let _ = socket.write_all(response_head.as_bytes()).await; + let _ = socket.write_all(body).await; + let _ = socket.shutdown().await; + }); + } + }); + + Ok((format!("http://127.0.0.1:{}/probe.ts", addr.port()), request_capture, accepted, handle)) + } + + fn spawn_stdin_test_child(script: &str) -> tokio::process::Child { + Command::new("sh") + .arg("-c") + .arg(script) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .kill_on_drop(true) + .spawn() + .expect("test child should spawn") + } + + struct ErroringWriter; + + impl AsyncWrite for ErroringWriter { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &[u8], + ) -> Poll> { + Poll::Ready(Err(io::Error::new(io::ErrorKind::ConnectionReset, "synthetic write failure"))) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } #[test] fn build_ffprobe_proxy_url_injects_credentials() { @@ -472,6 +899,14 @@ mod tests { assert!(!super::is_not_found_probe_error("protocol handler not found")); } + #[test] + fn probe_support_rejects_hls_manifest_urls() { + assert!(!super::is_supported_probe_url("http://provider.example/live/master.m3u8")); + assert!(!super::is_supported_probe_url("http://provider.example/live/master.m3u8?token=abc")); + assert!(super::is_supported_probe_url("http://provider.example/live/stream.ts")); + assert!(super::is_supported_probe_url("file:///var/media/movie.mkv")); + } + #[test] fn extract_probe_stream_stats_prefers_format_section() { let payload = json!({ @@ -507,4 +942,93 @@ mod tests { } ); } + + #[tokio::test] + async fn fetch_probe_bytes_caps_response_and_sets_headers() { + let (url, request_capture, accepted, handle) = match start_test_http_server("HTTP/1.1 200 OK", b"abcdefghij").await + { + Ok(server) => server, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => { + eprintln!("skipping fetch_probe_bytes_caps_response_and_sets_headers: {err}"); + return; + } + Err(err) => panic!("failed to start test http server: {err}"), + }; + + let client = reqwest::Client::builder() + .no_proxy() + .build() + .expect("http client should build"); + let bytes = fetch_probe_bytes(&client, &url, Some("test-agent/1.0"), 5) + .await + .expect("fetch should succeed"); + + assert_eq!(bytes, b"abcde"); + assert_eq!(accepted.load(Ordering::SeqCst), 1); + + let request_text = + String::from_utf8_lossy(&request_capture.lock().expect("request capture mutex poisoned")).to_ascii_lowercase(); + assert!(request_text.contains("range: bytes=0-4")); + assert!(request_text.contains("user-agent: test-agent/1.0")); + + handle.abort(); + } + + #[tokio::test] + async fn fetch_probe_bytes_maps_404_to_not_found() { + let (url, _request_capture, _accepted, handle) = + match start_test_http_server("HTTP/1.1 404 Not Found", b"missing").await { + Ok(server) => server, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => { + eprintln!("skipping fetch_probe_bytes_maps_404_to_not_found: {err}"); + return; + } + Err(err) => panic!("failed to start test http server: {err}"), + }; + + let client = reqwest::Client::builder() + .no_proxy() + .build() + .expect("http client should build"); + let result = fetch_probe_bytes(&client, &url, None, 16).await; + + assert!(matches!(result, Err(ProbeFailureKind::NotFound))); + + handle.abort(); + } + + #[tokio::test] + async fn run_ffprobe_with_stdin_tolerates_early_child_exit_on_success() { + let child = spawn_stdin_test_child( + "dd bs=1 count=1 of=/dev/null 2>/dev/null; \ + printf '%s' '{\"streams\":[{\"codec_type\":\"video\",\"codec_name\":\"h264\",\"width\":1920,\"height\":1080},{\"codec_type\":\"audio\",\"codec_name\":\"aac\",\"channels\":2}],\"format\":{\"duration\":\"10\",\"bit_rate\":\"1000\"}}'", + ); + let probe_bytes = vec![b'x'; 1024 * 1024]; + + let outcome = run_ffprobe_with_stdin(child, &probe_bytes, "https://example.com/live.ts").await; + + assert!(matches!(outcome, ProbeUrlOutcome::Success(..))); + } + + #[tokio::test] + async fn run_ffprobe_with_stdin_uses_child_failure_after_early_close() { + let child = spawn_stdin_test_child( + "dd bs=1 count=1 of=/dev/null 2>/dev/null; \ + printf '%s' 'Invalid data found when processing input' >&2; \ + exit 1", + ); + let probe_bytes = vec![b'x'; 1024 * 1024]; + + let outcome = run_ffprobe_with_stdin(child, &probe_bytes, "https://example.com/live.ts").await; + + assert!(matches!(outcome, ProbeUrlOutcome::Failed(ProbeFailureKind::Other))); + } + + #[tokio::test] + async fn write_probe_bytes_to_stdin_keeps_non_broken_pipe_failures_fatal() { + let mut writer = ErroringWriter; + let result = write_probe_bytes_to_stdin(&mut writer, b"abcdef", "https://example.com/live.ts").await; + + assert!(matches!(result, Err(ProbeFailureKind::Other))); + } }