feat: broker validated sources and bounded media probing

enginefs/src/hls.rs

@@ -235,7 +235,7 @@ impl HlsEngine {
             "Spawning ffmpeg probe command with analyzeduration={} probesize={} path={}",
             analyzeduration,
             probesize,
-            file_path
+            redact_source_capabilities(file_path)
         );
         let mut child = cmd.spawn().context("Failed to spawn ffmpeg")?;
         let stderr = child.stderr.take().context("Failed to capture stderr")?;
@@ -694,14 +694,14 @@ impl HlsEngine {
         // Output format: MPEG-TS for HLS with copyts to preserve timestamps
         cmd.args(["-mpegts_copyts", "1", "-f", "mpegts", "pipe:1"]);
 
-        tracing::debug!("FFmpeg video command (HLS V2): {:?}", cmd);
+        tracing::debug!("Starting FFmpeg video command for HLS V2");
 
         #[allow(clippy::zombie_processes)]
         let mut child = cmd
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
             .spawn()
-            .with_context(|| format!("Failed to spawn ffmpeg for video segment: {:?}", cmd))?;
+            .context("Failed to spawn ffmpeg for video segment")?;
 
         // Spawn a task to log stderr in background (for debugging)
         if let Some(stderr) = child.stderr.take() {
@@ -711,7 +711,10 @@ impl HlsEngine {
                 let mut line = String::new();
                 while reader.read_line(&mut line).await.unwrap_or(0) > 0 {
                     if !line.trim().is_empty() {
-                        tracing::warn!("FFmpeg video stderr: {}", line.trim());
+                        tracing::warn!(
+                            "FFmpeg video stderr: {}",
+                            redact_source_capabilities(line.trim())
+                        );
                     }
                     line.clear();
                 }
@@ -795,14 +798,14 @@ impl HlsEngine {
         // Output format: MPEG-TS for HLS
         cmd.args(["-mpegts_copyts", "1", "-f", "mpegts", "pipe:1"]);
 
-        tracing::debug!("FFmpeg audio command (HLS V2): {:?}", cmd);
+        tracing::debug!("Starting FFmpeg audio command for HLS V2");
 
         #[allow(clippy::zombie_processes)]
         let mut child = cmd
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
             .spawn()
-            .with_context(|| format!("Failed to spawn ffmpeg for audio segment: {:?}", cmd))?;
+            .context("Failed to spawn ffmpeg for audio segment")?;
 
         if let Some(stderr) = child.stderr.take() {
             tokio::spawn(async move {
@@ -811,7 +814,10 @@ impl HlsEngine {
                 let mut line = String::new();
                 while reader.read_line(&mut line).await.unwrap_or(0) > 0 {
                     if !line.trim().is_empty() {
-                        tracing::warn!("FFmpeg audio stderr: {}", line.trim());
+                        tracing::warn!(
+                            "FFmpeg audio stderr: {}",
+                            redact_source_capabilities(line.trim())
+                        );
                     }
                     line.clear();
                 }
@@ -937,6 +943,36 @@ fn text_mentions_high_bit_depth(value: &str) -> bool {
         || value.contains("yuv444p12")
 }
 
+fn redact_source_capabilities(input: &str) -> std::borrow::Cow<'_, str> {
+    const MARKER: &str = "cap=";
+    const CREDENTIAL_LENGTH: usize = 64;
+
+    let mut remainder = input;
+    let mut output = String::new();
+    let mut redacted = false;
+    while let Some(index) = remainder.find(MARKER) {
+        let credential_start = index + MARKER.len();
+        let credential = remainder
+            .as_bytes()
+            .get(credential_start..credential_start + CREDENTIAL_LENGTH);
+        if credential.is_some_and(|bytes| bytes.iter().all(u8::is_ascii_hexdigit)) {
+            output.push_str(&remainder[..credential_start]);
+            output.push_str("<redacted>");
+            remainder = &remainder[credential_start + CREDENTIAL_LENGTH..];
+            redacted = true;
+        } else {
+            output.push_str(&remainder[..credential_start]);
+            remainder = &remainder[credential_start..];
+        }
+    }
+    if redacted {
+        output.push_str(remainder);
+        std::borrow::Cow::Owned(output)
+    } else {
+        std::borrow::Cow::Borrowed(input)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -994,4 +1030,20 @@ mod tests {
             }
         }
     }
+
+    #[test]
+    fn ffmpeg_diagnostics_redact_transcoding_source_capabilities() {
+        let credential = "a".repeat(64);
+        let diagnostic = format!(
+            "unable to open http://127.0.0.1:11470/_transcoding/source?cap={credential}: denied"
+        );
+        let redacted = redact_source_capabilities(&diagnostic);
+
+        assert!(redacted.contains("cap=<redacted>"));
+        assert!(!redacted.contains(&credential));
+        assert_eq!(
+            redact_source_capabilities("ordinary diagnostic"),
+            "ordinary diagnostic"
+        );
+    }
 }

enginefs/tests/hls_playlist_allocations.rs

@@ -1,16 +1,29 @@
 use enginefs::hls::{HlsEngine, ProbeResult};
 use std::alloc::{GlobalAlloc, Layout, System};
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::cell::Cell;
+use std::sync::{
+    Arc, Barrier,
+    atomic::{AtomicUsize, Ordering},
+};
 
 struct CountingAllocator;
 
-static COUNTING: AtomicBool = AtomicBool::new(false);
+thread_local! {
+    static COUNTING: Cell<bool> = const { Cell::new(false) };
+}
+
 static ALLOCATIONS: AtomicUsize = AtomicUsize::new(0);
 static REALLOCATIONS: AtomicUsize = AtomicUsize::new(0);
 
+fn is_counting() -> bool {
+    COUNTING
+        .try_with(|counting| counting.get())
+        .unwrap_or(false)
+}
+
 unsafe impl GlobalAlloc for CountingAllocator {
     unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
-        if COUNTING.load(Ordering::Relaxed) {
+        if is_counting() {
             ALLOCATIONS.fetch_add(1, Ordering::Relaxed);
         }
         unsafe { System.alloc(layout) }
@@ -21,7 +34,7 @@ unsafe impl GlobalAlloc for CountingAllocator {
     }
 
     unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
-        if COUNTING.load(Ordering::Relaxed) {
+        if is_counting() {
             REALLOCATIONS.fetch_add(1, Ordering::Relaxed);
         }
         unsafe { System.realloc(ptr, layout, new_size) }
@@ -32,14 +45,14 @@ unsafe impl GlobalAlloc for CountingAllocator {
 static GLOBAL: CountingAllocator = CountingAllocator;
 
 fn start_counting() {
-    COUNTING.store(false, Ordering::SeqCst);
+    COUNTING.with(|counting| counting.set(false));
     ALLOCATIONS.store(0, Ordering::SeqCst);
     REALLOCATIONS.store(0, Ordering::SeqCst);
-    COUNTING.store(true, Ordering::SeqCst);
+    COUNTING.with(|counting| counting.set(true));
 }
 
 fn stop_counting() -> (usize, usize) {
-    COUNTING.store(false, Ordering::SeqCst);
+    COUNTING.with(|counting| counting.set(false));
     (
         ALLOCATIONS.load(Ordering::SeqCst),
         REALLOCATIONS.load(Ordering::SeqCst),
@@ -48,6 +61,30 @@ fn stop_counting() -> (usize, usize) {
 
 #[test]
 fn two_hour_playlist_avoids_per_segment_temporary_allocations() {
+    let allocation_started = Arc::new(Barrier::new(2));
+    let allocation_finished = Arc::new(Barrier::new(2));
+    let worker = {
+        let allocation_started = Arc::clone(&allocation_started);
+        let allocation_finished = Arc::clone(&allocation_finished);
+        std::thread::spawn(move || {
+            allocation_started.wait();
+            let unrelated = Box::new([0_u8; 1_024]);
+            std::hint::black_box(&unrelated);
+            allocation_finished.wait();
+        })
+    };
+
+    start_counting();
+    allocation_started.wait();
+    allocation_finished.wait();
+    let unrelated_counts = stop_counting();
+    worker.join().expect("allocation worker must finish");
+    assert_eq!(
+        unrelated_counts,
+        (0, 0),
+        "allocator measurements must ignore unrelated test-harness threads"
+    );
+
     let probe = ProbeResult {
         duration: 7_200.0,
         container: "test".to_string(),

server/Cargo.toml

@@ -31,6 +31,7 @@ serde_json = "1.0.151"
 anyhow = "1.0.104"
 tokio-util = { version = "0.7.19", features = ["io", "compat", "rt"] }
 hex = "0.4.3"
+getrandom = "0.4.3"
 librqbit = { version = "9.0.0", optional = true }
 if-addrs = "0.15.0"
 ipnet = "2.12.1"

server/src/diagnostics/logging.rs

@@ -79,6 +79,13 @@ pub(crate) struct SanitizedRequestTarget {
 }
 
 pub(crate) fn sanitize_request_target(uri: &axum::http::Uri) -> SanitizedRequestTarget {
+    if uri.path() == "/_transcoding/source" {
+        return SanitizedRequestTarget {
+            uri: "/_transcoding/source?<redacted>".to_owned(),
+            path: "/_transcoding/source".to_owned(),
+            query: String::new(),
+        };
+    }
     if uri.path().starts_with("/proxy") {
         return SanitizedRequestTarget {
             uri: "/proxy/<redacted>".to_owned(),
@@ -446,6 +453,18 @@ mod tests {
         assert_eq!(target.query, "");
     }
 
+    #[test]
+    fn transcoding_source_capabilities_are_redacted_for_every_logging_path() {
+        let uri: Uri = "/_transcoding/source?cap=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+            .parse()
+            .unwrap();
+        let target = sanitize_request_target(&uri);
+        assert_eq!(target.uri, "/_transcoding/source?<redacted>");
+        assert_eq!(target.path, "/_transcoding/source");
+        assert_eq!(target.query, "");
+        assert!(!target.uri.contains("aaaaaaaa"));
+    }
+
     #[test]
     fn ordinary_request_targets_keep_diagnostic_context() {
         let uri: Uri = "/heartbeat?probe=1".parse().unwrap();

server/src/lib.rs

@@ -575,13 +575,19 @@ async fn run_inner(
         Arc::new(network_security::SystemClock),
         listeners,
     ));
-    state.proxy_runtime = Arc::new(network_security::ProxyRuntime::new(
+    let proxy_runtime = Arc::new(network_security::ProxyRuntime::new(
         network_security::ProxyPolicySettings {
             allow_private_network_sources: settings.allow_private_network_sources,
             allow_invalid_proxy_tls_certificates: settings.allow_invalid_proxy_tls_certificates,
         },
         validator,
     ));
+    state.proxy_runtime = proxy_runtime.clone();
+    state.source_broker = Arc::new(transcoding::source::SourceBroker::new(
+        state.stream_engine(),
+        bound_http_addr,
+        proxy_runtime,
+    ));
 
     #[cfg(any(target_os = "windows", target_os = "macos", target_os = "linux"))]
     {
@@ -971,6 +977,10 @@ pub fn build_router(state: AppState) -> Router {
             "/stream/{infoHash}/{fileIdx}",
             get(routes::stream::stream_video).head(routes::stream::head_stream_video),
         )
+        .route(
+            "/_transcoding/source",
+            get(transcoding::source::route_source).head(transcoding::source::route_source),
+        )
         .route(
             "/{infoHash}/{fileIdx}",
             get(routes::stream::stream_video).head(routes::stream::head_stream_video),
@@ -1084,6 +1094,7 @@ pub fn build_router_with_listeners(mut state: AppState, listeners: Vec<SocketAdd
         .try_read()
         .expect("settings must be uncontended during router construction")
         .clone();
+    let source_listener = listeners[0];
     let validator = Arc::new(network_security::DestinationValidator::new(
         Arc::new(network_security::SystemDnsResolver),
         Arc::new(network_security::SystemLocalNetworkProvider),
@@ -1093,13 +1104,19 @@ pub fn build_router_with_listeners(mut state: AppState, listeners: Vec<SocketAdd
             .map(|socket| network_security::ListenerBinding { socket })
             .collect(),
     ));
-    state.proxy_runtime = Arc::new(network_security::ProxyRuntime::new(
+    let proxy_runtime = Arc::new(network_security::ProxyRuntime::new(
         network_security::ProxyPolicySettings {
             allow_private_network_sources: settings.allow_private_network_sources,
             allow_invalid_proxy_tls_certificates: settings.allow_invalid_proxy_tls_certificates,
         },
         validator,
     ));
+    state.proxy_runtime = proxy_runtime.clone();
+    state.source_broker = Arc::new(transcoding::source::SourceBroker::new(
+        state.stream_engine(),
+        source_listener,
+        proxy_runtime,
+    ));
     build_router(state)
 }