diff --git a/Cargo.lock b/Cargo.lock index a3bd910..2668e2a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1344,6 +1344,7 @@ dependencies = [ "sha2 0.11.0", "tokio", "uniffi", + "url", "wasm-bindgen", "web-time", ] diff --git a/Cargo.toml b/Cargo.toml index 479b8d0..28d5831 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -96,6 +96,7 @@ p256 = { version = "0.14.0", optional = true } pbkdf2 = { version = "0.13.0", optional = true } rand = { version = "0.10.2", optional = true } regex = "1" +url = "2" rquickjs = { version = "0.12.1", default-features = false, features = ["std", "chrono", "loader", "dyn-load", "either", "indexmap", "futures"], optional = true } rsa = { version = "0.9.10", features = ["sha2"], optional = true } scraper = { version = "0.27.0", optional = true } diff --git a/fluxa-streaming-engine/src/torrent_engine.rs b/fluxa-streaming-engine/src/torrent_engine.rs index 5b4fa4f..a0393a6 100644 --- a/fluxa-streaming-engine/src/torrent_engine.rs +++ b/fluxa-streaming-engine/src/torrent_engine.rs @@ -108,6 +108,12 @@ struct ActiveTelemetrySession { generation: u64, } +#[derive(Default)] +struct TelemetryState { + active_sessions: HashMap, + records: HashMap<(usize, String), PlaybackTelemetry>, +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct TelemetryEvent { @@ -182,8 +188,10 @@ struct EngineState { prioritized_files: Arc>>, playback_windows: Arc>>, playback_sessions: Arc>>, - playback_telemetry: Arc>>, - active_telemetry_sessions: Arc>>, + // Session ownership and its measurements must be changed atomically. Keeping + // them together also prevents lock-order inversions between telemetry writes + // and torrent teardown. + telemetry: Arc>, /// Root cancellation per torrent. Probe readers use child tokens too, so /// they cannot outlive playback deactivation or cache eviction. torrent_cancellations: Arc>>, @@ -346,8 +354,7 @@ pub fn start_torrent_server( prioritized_files: Arc::new(Mutex::new(HashMap::new())), playback_windows: Arc::new(Mutex::new(HashMap::new())), playback_sessions: Arc::new(Mutex::new(HashMap::new())), - playback_telemetry: Arc::new(Mutex::new(HashMap::new())), - active_telemetry_sessions: Arc::new(Mutex::new(HashMap::new())), + telemetry: Arc::new(Mutex::new(TelemetryState::default())), torrent_cancellations: Arc::new(Mutex::new(HashMap::new())), lifecycle: Arc::new(Mutex::new(HashMap::new())), active_torrent: Arc::new(Mutex::new(None)), @@ -653,13 +660,19 @@ async fn record_telemetry( if event.session_id.is_empty() || event.session_id.len() > 128 { return error_response(StatusCode::BAD_REQUEST, "invalid telemetry session"); } - let mut active_sessions = match state.active_telemetry_sessions.lock() { - Ok(sessions) => sessions, + if !matches!( + event.event.as_str(), + "firstFrame" | "stallStarted" | "stallEnded" + ) { + return error_response(StatusCode::BAD_REQUEST, "unsupported telemetry event"); + } + let mut telemetry = match state.telemetry.lock() { + Ok(telemetry) => telemetry, Err(_) => { return error_response(StatusCode::INTERNAL_SERVER_ERROR, "telemetry unavailable"); } }; - match active_sessions.get(&id) { + match telemetry.active_sessions.get(&id) { Some(active) if event.session_generation < active.generation => { return error_response(StatusCode::CONFLICT, "stale telemetry session"); } @@ -669,19 +682,19 @@ async fn record_telemetry( return error_response(StatusCode::CONFLICT, "telemetry session mismatch"); } Some(active) if event.session_generation > active.generation => { - active_sessions.insert( + telemetry.active_sessions.insert( id, ActiveTelemetrySession { id: event.session_id.clone(), generation: event.session_generation, }, ); - if let Ok(mut telemetry) = state.playback_telemetry.lock() { - telemetry.retain(|(stored_id, _), _| *stored_id != id); - } + telemetry + .records + .retain(|(stored_id, _), _| *stored_id != id); } None => { - active_sessions.insert( + telemetry.active_sessions.insert( id, ActiveTelemetrySession { id: event.session_id.clone(), @@ -691,14 +704,7 @@ async fn record_telemetry( } _ => {} } - drop(active_sessions); - let mut telemetry = match state.playback_telemetry.lock() { - Ok(telemetry) => telemetry, - Err(_) => { - return error_response(StatusCode::INTERNAL_SERVER_ERROR, "telemetry unavailable"); - } - }; - let entry = telemetry.entry((id, event.session_id)).or_default(); + let entry = telemetry.records.entry((id, event.session_id)).or_default(); match event.event.as_str() { "firstFrame" => entry.first_frame_ms = event.elapsed_ms.or(entry.first_frame_ms), "stallStarted" => entry.stall_count = entry.stall_count.saturating_add(1), @@ -707,7 +713,7 @@ async fn record_telemetry( .stall_duration_ms .saturating_add(event.elapsed_ms.unwrap_or_default()) } - _ => return error_response(StatusCode::BAD_REQUEST, "unsupported telemetry event"), + _ => unreachable!("telemetry event was validated before mutating state"), } (StatusCode::OK, Json(json!({}))).into_response() } @@ -1104,17 +1110,15 @@ async fn status_response( window.smoothed_download_bps * 8.0 / window.estimated_bitrate_bps.max(1) as f64 }) .unwrap_or(0.0); - let active_session = state - .active_telemetry_sessions - .lock() - .ok() - .and_then(|sessions| sessions.get(&id).cloned()); let playback_telemetry = state - .playback_telemetry + .telemetry .lock() .ok() .and_then(|telemetry| { - active_session.and_then(|session| telemetry.get(&(id, session.id)).copied()) + telemetry + .active_sessions + .get(&id) + .and_then(|session| telemetry.records.get(&(id, session.id.clone())).copied()) }) .unwrap_or_default(); let scheduler = window.map(|window| { @@ -1335,11 +1339,9 @@ async fn deactivate_torrent(state: &EngineState, torrent_id: usize) { } fn clear_playback_telemetry(state: &EngineState, torrent_id: usize) { - if let Ok(mut telemetry) = state.playback_telemetry.lock() { - telemetry.retain(|(id, _), _| *id != torrent_id); - } - if let Ok(mut sessions) = state.active_telemetry_sessions.lock() { - sessions.remove(&torrent_id); + if let Ok(mut telemetry) = state.telemetry.lock() { + telemetry.records.retain(|(id, _), _| *id != torrent_id); + telemetry.active_sessions.remove(&torrent_id); } } diff --git a/src/plugin_runtime/mod.rs b/src/plugin_runtime/mod.rs index 46e9677..464a539 100644 --- a/src/plugin_runtime/mod.rs +++ b/src/plugin_runtime/mod.rs @@ -10,6 +10,7 @@ use settings_layout::run_settings_layout; use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; +use url::{Host, Url}; pub(super) const PLUGIN_TIMEOUT_SECS: u64 = 60; pub(super) const PLUGIN_MEMORY_LIMIT: usize = 256 * 1024 * 1024; @@ -23,36 +24,17 @@ pub fn plugin_http_request_error(request: &PluginHttpRequest) -> Option<&'static ) { return Some("plugin request method is not allowed"); } - let lower = request.url.to_ascii_lowercase(); - if !(lower.starts_with("http://") || lower.starts_with("https://")) { + let parsed = match Url::parse(&request.url) { + Ok(parsed) => parsed, + Err(_) => return Some("invalid plugin URL"), + }; + if !matches!(parsed.scheme(), "http" | "https") { return Some("only http and https plugin URLs are allowed"); } - let authority = lower - .split("//") - .nth(1)? - .split('/') - .next()? - .split('@') - .next_back()?; - let host = authority.split(':').next().unwrap_or(authority); - let private_172 = host - .strip_prefix("172.") - .and_then(|rest| rest.split('.').next()) - .and_then(|part| part.parse::().ok()) - .is_some_and(|second| (16..=31).contains(&second)); - if host == "localhost" - || host.ends_with(".localhost") - || host == "::1" - || host.starts_with("127.") - || host.starts_with("10.") - || host.starts_with("192.168.") - || host.starts_with("169.254.") - || host.starts_with("0.") - || private_172 - || host.starts_with("fc") - || host.starts_with("fd") - || host.starts_with("fe80:") - { + let Some(host) = parsed.host() else { + return Some("plugin URL is missing a host"); + }; + if !is_public_plugin_host(host) { return Some("private or loopback plugin URL is not allowed"); } if let Some(body) = &request.body @@ -71,6 +53,29 @@ pub fn plugin_http_request_error(request: &PluginHttpRequest) -> Option<&'static None } +fn is_public_plugin_host(host: Host<&str>) -> bool { + match host { + Host::Ipv4(ip) => { + !ip.is_private() && !ip.is_loopback() && !ip.is_link_local() && !ip.is_unspecified() + } + Host::Ipv6(ip) => { + !ip.is_loopback() + && !ip.is_unspecified() + && !ip.is_unique_local() + && !ip.is_unicast_link_local() + } + Host::Domain(domain) => { + let domain = domain.trim_end_matches('.').to_ascii_lowercase(); + domain != "localhost" + && !domain.ends_with(".localhost") + && domain + .parse::() + .map(|ip| is_public_plugin_host(Host::Ipv4(ip))) + .unwrap_or(true) + } + } +} + #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))] #[serde(rename_all = "camelCase")] @@ -95,6 +100,10 @@ pub struct PluginHttpResponse { #[cfg_attr(feature = "uniffi-bindings", uniffi::export(callback_interface))] pub trait PluginHttpClient: Send + Sync { + /// Before every connection, resolve the requested host and reject every + /// loopback, private, link-local, or unspecified A/AAAA result. Apply the + /// same check to every redirect target to prevent DNS rebinding and redirect + /// based SSRF; this core-side policy can only inspect the original URL. fn fetch(&self, request: PluginHttpRequest) -> PluginHttpResponse; } @@ -190,6 +199,23 @@ mod tests { assert!(plugin_http_request_error(&request).is_some()); } + #[test] + fn plugin_request_policy_rejects_bracketed_private_ipv6() { + for url in [ + "http://[::1]/", + "http://[fc00::1]/", + "http://[fe80::1]/", + "http://localhost./", + "http://127.0.0.1./", + ] { + let request = PluginHttpRequest { + url: url.to_string(), + ..PluginHttpRequest::default() + }; + assert!(plugin_http_request_error(&request).is_some(), "{url}"); + } + } + fn run_scraper(code: &str, tmdb_id: &str, media_type: &str) -> String { execute_scraper( mock_client(),