Creepso/fluxa-core
1
0
Fork 0
mirror of https://github.com/FluxaMedia/fluxa-core.git synced 2026-08-16 12:13:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Harden telemetry and plugin URL validation

Browse source
This commit is contained in:
KhooLy 2026-08-01 21:09:47 +03:00
parent 46ad60811e
commit ba39d577d2
4 changed files with 91 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1344,6 +1344,7 @@ dependencies = [
"sha2 0.11.0",
"tokio",
"uniffi",
"url",
"wasm-bindgen",
"web-time",
]

1
Cargo.toml
View file

@ -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 }

68
fluxa-streaming-engine/src/torrent_engine.rs
View file

@ -108,6 +108,12 @@ struct ActiveTelemetrySession {
generation: u64,
}
#[derive(Default)]
struct TelemetryState {
active_sessions: HashMap<usize, ActiveTelemetrySession>,
records: HashMap<(usize, String), PlaybackTelemetry>,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct TelemetryEvent {
@ -182,8 +188,10 @@ struct EngineState {
prioritized_files: Arc<Mutex<HashMap<usize, TorrentFileFocus>>>,
playback_windows: Arc<Mutex<HashMap<(usize, usize), PlaybackWindow>>>,
playback_sessions: Arc<Mutex<HashMap<(usize, usize), PlaybackSession>>>,
playback_telemetry: Arc<Mutex<HashMap<(usize, String), PlaybackTelemetry>>>,
active_telemetry_sessions: Arc<Mutex<HashMap<usize, ActiveTelemetrySession>>>,
// 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<Mutex<TelemetryState>>,
/// Root cancellation per torrent. Probe readers use child tokens too, so
/// they cannot outlive playback deactivation or cache eviction.
torrent_cancellations: Arc<Mutex<HashMap<usize, CancellationToken>>>,
@ -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);
}
}

82
src/plugin_runtime/mod.rs
View file

@ -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::<u8>().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::<std::net::Ipv4Addr>()
.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(),