diff --git a/CHANGELOG.md b/CHANGELOG.md index 4124d97b6..3452ba438 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,15 @@ - MVP supports `movies/shows` with `trending` and `popular`. - User-owned Trakt lists remain configured separately under `trakt.lists[]`. +- **Update Log In Playlist Update View**: + - The Playlists → Update view now shows a terminal-style log that accumulates `PlaylistUpdateProgress` and + `LibraryScanProgress` events in real time, prefixed with `[playlist]` / `[library]` and a local `HH:MM:SS` + timestamp, with auto-scroll to the latest line and a FIFO cap of 500 entries. + - The log is cleared synchronously when the user clicks either the playlist Update or the library Update button, + so each run starts with a fresh view. + - Styled to match the dark monospace look of a console (uses existing theme CSS variables for background, border, + and text color) and honors the same touch / overflow behavior as the rest of the view. + - **Session Expiry Handling**: - The Web UI now schedules a client-side logout when the JWT expires, showing a notification and returning the user to the login screen instead of silently failing with 401 errors. diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index d0fb541d8..c58b0aa51 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -506,7 +506,7 @@ pub(crate) fn mark_response_as_uncompressed(response: &mut Response) { response.extensions_mut().insert(DisableResponseCompression); } -#[cfg_attr(not(test), allow(dead_code))] +#[cfg(test)] pub(crate) fn should_compress_response(response: &Response) -> bool { should_compress_response_extensions(response.extensions()) } @@ -724,6 +724,7 @@ struct StreamingAcquireOptions<'a> { user_priority: i8, connection_kind: crate::api::model::ConnectionKind, session_owner: Option<&'a str>, + accept_requested_stream_url: bool, } pub(crate) fn connection_priority_for_kind(user: &ProxyUserCredentials, kind: crate::api::model::ConnectionKind) -> i8 { @@ -784,9 +785,9 @@ pub(crate) struct GraceResolutionContext { /// The original `ConnectionKind` from the admission decision that led to this grace. /// Preserved so that the remaining-strategy fallback can return the correct kind /// (e.g., `Soft`) even when the grace itself hardcoded `Normal`. - #[allow(dead_code)] // Stored so the original admission kind remains available when follow-up // grace fallback reconstruction starts using it again. + #[allow(dead_code)] pub(crate) kind: Option, } @@ -967,6 +968,8 @@ pub(in crate::api) async fn resolve_admission_with_strategies( kind: admission.kind, }; + let _admission_guard = app_state.active_users.acquire_user_admission(username).await; + if let Some(resolution) = evaluate_admission_strategy_loop( app_state, username, @@ -1048,6 +1051,8 @@ pub(in crate::api) async fn evaluate_remaining_strategies_after_grace( kind: original_kind, }; + let _admission_guard = app_state.active_users.acquire_user_admission(username).await; + if let Some(resolution) = evaluate_admission_strategy_loop( app_state, username, @@ -1125,10 +1130,11 @@ async fn activate_session_before_stream_open( // If caller passes FollowUp, verify the session is still counted under the guard. // A stale FollowUp would bypass admission — reclassify to catch this. let effective_request_class = if let Some(request_class) = request_class { - if request_class == PlaybackRequestClass::FollowUp { - // Re-read session under the guard to ensure the counted lease is still held. + if matches!(request_class, PlaybackRequestClass::FollowUp | PlaybackRequestClass::Activate) { + // Re-read session under the guard to ensure the counted lease is still held or acquired. // If it is no longer counted, classify it from the current lifecycle so // stale FollowUp requests cannot bypass admission. + // If it became counted, classify it so stale Activate requests don't double count. let current_session = app_state .active_users .get_and_update_user_session(&user.username, session_token) @@ -1193,8 +1199,13 @@ async fn activate_session_before_stream_open( }, Some(crate::api::model::GraceMode::Hold)) } Some(crate::api::model::PlaybackLifecycle::GraceActive) => { - // Already in GraceActive — nothing to refresh. - (crate::api::model::PlaybackLifecycle::GraceActive, Some(crate::api::model::GraceMode::Instant)) + // Already in GraceActive — infer mode from item_type. + let mode = if item_type.is_live() || item_type.is_live_adaptive() { + crate::api::model::GraceMode::Hold + } else { + crate::api::model::GraceMode::Instant + }; + (crate::api::model::PlaybackLifecycle::GraceActive, Some(mode)) } _ => { // Session not yet in grace state — infer from item_type defaults. @@ -1349,10 +1360,10 @@ fn get_stream_alternative_url_m3u( input: &ConfigInput, alias_input: &Arc, ) -> Option { + let alt_input_user_info = alias_input.get_user_info()?; if let Some((source_base_url, source_username, source_password)) = find_input_account_by_signature(stream_url, input) { - let alt_input_user_info = alias_input.get_user_info()?; let modified = stream_url.replacen(&source_base_url, &alt_input_user_info.base_url, 1); let mut url = Url::parse(&modified).ok()?; @@ -1371,7 +1382,7 @@ fn get_stream_alternative_url_m3u( return Some(url.to_string()); } - if stream_url_has_account_signature(stream_url) { + if stream_url_has_account_signature(stream_url, &alt_input_user_info) { return None; } Some(stream_url.to_string()) @@ -1461,6 +1472,18 @@ fn rewrite_query_auth_fields(url: &mut Url, new_username: &str, new_password: &s true } +fn collect_path_segments(url: &Url) -> Option> { + url.path_segments() + .map(|segments| segments.map(ToOwned::to_owned).collect::>()) +} + +fn find_path_auth_segment_index(segments: &[String], username: &str, password: &str) -> Option { + segments.windows(2).position(|pair| { + pair.first().is_some_and(|segment| segment == username) + && pair.get(1).is_some_and(|segment| segment == password) + }) +} + fn rewrite_path_auth_fields( url: &mut Url, old_username: &str, @@ -1468,28 +1491,11 @@ fn rewrite_path_auth_fields( new_username: &str, new_password: &str, ) -> bool { - let Some(mut segments) = url - .path_segments() - .map(|segments| segments.map(ToOwned::to_owned).collect::>()) - else { + let Some(mut segments) = collect_path_segments(url) else { return false; }; - let credential_index = if segments.len() >= 3 - && matches!(segments.first().map(String::as_str), Some("live" | "movie" | "series")) - && segments.get(1).is_some_and(|segment| segment == old_username) - && segments.get(2).is_some_and(|segment| segment == old_password) - { - Some(1) - } else if segments.len() >= 2 - && segments.first().is_some_and(|segment| segment == old_username) - && segments.get(1).is_some_and(|segment| segment == old_password) - { - Some(0) - } else { - None - }; - + let credential_index = find_path_auth_segment_index(&segments, old_username, old_password); let Some(credential_index) = credential_index else { return false; }; @@ -1518,7 +1524,7 @@ fn stream_url_matches_provider(stream_url: &str, provider_cfg: &ProviderConfig) // For M3U inputs, the stored playlist entry itself is the trust anchor. // Open external URLs are therefore allowed, but external URLs that carry // explicit account markers must still match the selected provider account. - if stream_url_has_account_signature(stream_url) { + if stream_url_has_account_signature(stream_url, &user_info) { return stream_url_account_matches(stream_url, &user_info); } true @@ -1553,18 +1559,14 @@ fn stream_url_account_matches(stream_url: &str, user_info: &crate::model::InputU return has_query_username && has_query_password; } - let mut segments = url.path_segments().into_iter().flatten(); - let first = segments.next(); - let (username, password) = match first { - Some("live" | "movie" | "series") => (segments.next(), segments.next()), - Some(username) => (Some(username), segments.next()), - None => return false, + let Some(segments) = collect_path_segments(&url) else { + return false; }; - username == Some(user_info.username.as_str()) && password == Some(user_info.password.as_str()) + find_path_auth_segment_index(&segments, &user_info.username, &user_info.password).is_some() } -fn stream_url_has_account_signature(stream_url: &str) -> bool { +fn stream_url_has_account_signature(stream_url: &str, user_info: &crate::model::InputUserInfo) -> bool { let Ok(url) = Url::parse(stream_url) else { return false; }; @@ -1586,6 +1588,17 @@ fn stream_url_has_account_signature(stream_url: &str) -> bool { if has_query_username || has_query_password { return has_query_username && has_query_password; } + + // Path-based credentials: some Xtream endpoints embed the account in the URL + // path (e.g. /live///...). Only flag a signature when the + // consecutive segments actually match the configured user/pass — arbitrary + // open paths must not be treated as account signatures. + if let Some(segments) = collect_path_segments(&url) { + if find_path_auth_segment_index(&segments, &user_info.username, &user_info.password).is_some() { + return true; + } + } + false } @@ -1593,7 +1606,11 @@ fn select_provider_stream_url( stream_url: &str, input: &ConfigInput, provider_cfg: &Arc, + accept_requested_stream_url: bool, ) -> Option<(Arc, String)> { + if accept_requested_stream_url { + return Some((provider_cfg.name.clone(), stream_url.to_string())); + } if stream_url_matches_provider(stream_url, provider_cfg) { Some((provider_cfg.name.clone(), stream_url.to_string())) } else { @@ -1729,6 +1746,7 @@ async fn resolve_streaming_strategy( options: StreamingAcquireOptions<'_>, ) -> StreamingStrategy { // allocate a provider connection + let accept_requested_stream_url = options.accept_requested_stream_url; let mut provider_connection_handle = acquire_stream_provider_handle(app_state, input, fingerprint, options).await; // panel_api provisioning/loading is handled later in the stream creation flow @@ -1744,7 +1762,12 @@ async fn resolve_streaming_strategy( ProviderAllocation::Available(ref provider_cfg) | ProviderAllocation::GracePeriod(ref provider_cfg) => { // Keep the URL only when it already targets the selected provider account. Hot reload can leave old // alias URLs in persisted playlists until the next processing run. - if let Some((selected_provider_name, url)) = select_provider_stream_url(stream_url, input, provider_cfg) { + if let Some((selected_provider_name, url)) = select_provider_stream_url( + stream_url, + input, + provider_cfg, + accept_requested_stream_url, + ) { debug_if_enabled!( "provider session: input={} provider_cfg={} user={} allocation={} stream_url={}", sanitize_sensitive_info(&input.name), @@ -1782,7 +1805,10 @@ async fn resolve_streaming_strategy( if release_failed_mapping { if let Some(handle) = provider_connection_handle.take() { - app_state.connection_manager.release_provider_handle(Some(handle)).await; + let connection_manager = Arc::clone(&app_state.connection_manager); + tokio::spawn(async move { + connection_manager.release_provider_handle(Some(handle)).await; + }); } } @@ -1850,6 +1876,8 @@ async fn create_stream_response_details( connection_kind: crate::api::model::ConnectionKind, is_reopen: bool, session_owner: Option<&str>, + session_headers: Option<&HashMap>, + accept_requested_stream_url: bool, grace_hold_override: Option, grace_resolution_context: Option, ) -> Result { @@ -1865,6 +1893,7 @@ async fn create_stream_response_details( user_priority, connection_kind, session_owner, + accept_requested_stream_url, }, ) .await; @@ -1915,6 +1944,7 @@ async fn create_stream_response_details( stream_info, provider_name: Some(provider_name), request_url: None, + session_headers: session_headers.cloned(), grace_period: grace_period_options, provider_grace_active: false, disable_provider_grace: false, @@ -1972,6 +2002,7 @@ async fn create_stream_response_details( stream_url: &url, req_headers, input_headers: streaming_strategy.input_headers.as_ref(), + session_headers, disabled_headers: disabled_headers.as_ref(), default_user_agent: default_user_agent.as_deref(), username: Some(username), @@ -2012,12 +2043,15 @@ async fn create_stream_response_details( } } - // If no upstream stream is ready, release the provider unless provider grace - // intentionally deferred the open until the grace check resolves. - let provider_handle = if stream.is_none() && !defer_provider_stream_until_grace_check { + // If no upstream stream is ready, release the provider. + // Even if provider grace intentionally deferred the open, we must release the handle + // here because the deferred open context will acquire a fresh slot when it resumes. + let provider_handle = if stream.is_none() { let provider_handle = streaming_strategy.provider_handle.take(); app_state.connection_manager.release_provider_handle(provider_handle).await; - error!("Can't open stream {}", sanitize_sensitive_info(&request_url)); + if !defer_provider_stream_until_grace_check { + error!("Can't open stream {}", sanitize_sensitive_info(&request_url)); + } None } else { streaming_strategy.provider_handle.take() @@ -2028,6 +2062,7 @@ async fn create_stream_response_details( stream_info, provider_name: guard_provider_name.clone(), request_url: Some(request_url.clone()), + session_headers: session_headers.cloned(), grace_period: grace_period_options, provider_grace_active, disable_provider_grace: false, @@ -2347,6 +2382,8 @@ pub async fn force_provider_stream_response( connection_kind, true, Some(user_session.token.as_str()), + Some(&user_session.provider_session_headers), + true, grace_mode.map(|mode| matches!(mode, crate::api::model::GraceMode::Hold)), None, ) @@ -2577,6 +2614,10 @@ pub(crate) async fn stream_response( } let stream_options = get_stream_options(app_state); + let session_state = app_state + .active_users + .get_and_update_user_session(&user.username, session_token) + .await; let mut stream_details = match create_stream_response_details( app_state, &stream_options, @@ -2597,6 +2638,8 @@ pub(crate) async fn stream_response( connection_kind, false, Some(session_token), + session_state.as_ref().map(|session| &session.provider_session_headers), + pinned_provider.is_some(), grace_mode.map(|m| matches!(m, crate::api::model::GraceMode::Hold)), activation.grace_context.clone(), ) @@ -2656,6 +2699,11 @@ pub(crate) async fn stream_response( ) .await; + // Captured before `stream_details` is moved into `create_active_client_stream`. + // The pinning rule is centralized in `should_pin_provider_for_session` so it stays + // testable in isolation and in sync with the call site below. + let should_pin_provider = should_pin_provider_for_session(&stream_details, app_state, item_type); + let mut is_stream_shared = share_stream && !stream_details.has_deferred_provider_open(); if let Some((_header, _status_code, _url, Some(_custom_video))) = stream_details.stream_info.as_ref() { if stream_details.stream.is_some() { @@ -2797,12 +2845,14 @@ pub(crate) async fn stream_response( socket_bound, }) .await; - let reservation_ttl_secs = get_session_reservation_ttl_secs(app_state, item_type); - if reservation_ttl_secs > 0 { - app_state - .active_provider - .refresh_provider_reservation(&provider, session_token, reservation_ttl_secs) - .await; + if should_pin_provider { + let reservation_ttl_secs = get_session_reservation_ttl_secs(app_state, item_type); + if reservation_ttl_secs > 0 { + app_state + .active_provider + .refresh_provider_reservation(&provider, session_token, reservation_ttl_secs) + .await; + } } } } @@ -2862,18 +2912,17 @@ async fn prepare_stream_metering( return StreamMeteringConfig::default(); } - if share_stream && !has_stream && !has_deferred_provider_open { + if share_stream { + let meter_uid = app_state + .shared_stream_manager + .get_or_register_meter_uid(stream_url, || app_state.connection_manager.next_stream_uid()) + .await; return StreamMeteringConfig { - meter_uid: app_state.shared_stream_manager.get_meter_uid(stream_url).await.unwrap_or(0), - meter_stream: false, + meter_uid, + meter_stream: has_stream || has_deferred_provider_open, }; - } - - if has_stream || has_deferred_provider_open { + } else if has_stream || has_deferred_provider_open { let meter_uid = app_state.connection_manager.next_stream_uid(); - if share_stream { - app_state.shared_stream_manager.register_meter_uid(stream_url, meter_uid).await; - } return StreamMeteringConfig { meter_uid, meter_stream: true }; } @@ -2928,6 +2977,29 @@ pub(crate) fn get_session_reservation_ttl_secs(app_state: &Arc, item_t } } +/// Whether the session should pin the provider account via `refresh_provider_reservation`. +/// +/// A non-Provisioning custom video (`ChannelUnavailable`, `ProviderConnectionsExhausted`, …) means +/// the upstream open already failed. The provider connection slot was released by +/// `create_provider_stream`, and the custom video is a local fallback served to the client. +/// Pinning the provider via `refresh_provider_reservation` would hold the provider account for +/// the configured session TTL (e.g. `catchup_session_ttl_secs`), blocking other sessions of +/// the same family from using it even though the slot is already free. +/// +/// Only `Provisioning` custom videos represent a real provider handoff that benefits from +/// keeping the same provider pinned, and real provider streams (`stream_info` carries no +/// `CustomVideoStreamType`) obviously qualify. +pub(crate) fn should_pin_provider_for_session( + stream_details: &StreamDetails, + _app_state: &Arc, + _item_type: PlaylistItemType, +) -> bool { + !matches!( + stream_details.stream_info.as_ref(), + Some((_, _, _, Some(cv))) if *cv != CustomVideoStreamType::Provisioning + ) +} + #[allow(clippy::too_many_arguments)] async fn try_shared_stream_response_if_any( app_state: &Arc, @@ -2940,10 +3012,6 @@ async fn try_shared_stream_response_if_any( session_token: &str, req_headers: &HeaderMap, ) -> Option { - if connect_permission == UserConnectionPermission::GracePeriod { - return None; - } - if let Some((stream, provider)) = SharedStreamManager::subscribe_shared_stream( app_state, stream_url, @@ -2983,10 +3051,13 @@ async fn try_shared_stream_response_if_any( } stream_channel.shared = true; stream_channel.shared_joined_existing = Some(true); - stream_channel.shared_stream_id = - app_state.shared_stream_manager.get_meter_uid(stream_url).await.map(u64::from); + let meter_uid = app_state + .shared_stream_manager + .get_or_register_meter_uid(stream_url, || app_state.connection_manager.next_stream_uid()) + .await; + stream_channel.shared_stream_id = Some(u64::from(meter_uid)); let metering = StreamMeteringConfig { - meter_uid: app_state.shared_stream_manager.get_meter_uid(stream_url).await.unwrap_or(0), + meter_uid, meter_stream: false, }; let stream = create_active_client_stream(crate::api::model::ActiveClientStreamParams { @@ -3087,8 +3158,11 @@ pub(crate) async fn local_stream_response( let path = PathBuf::from(pli.url.strip_prefix("file://").unwrap_or(&pli.url)); + let Ok(mut file) = tokio::fs::File::open(&path).await else { return StatusCode::NOT_FOUND.into_response() }; + let Ok(opened_metadata) = file.metadata().await else { return internal_server_error!() }; + // Canonicalize and validate the path - let path = match tokio::fs::canonicalize(&path).await { + let canonical = match tokio::fs::canonicalize(&path).await { Ok(canonical) => canonical, Err(err) => { error!("Local file path is corrupt {}: {err}", path.display()); @@ -3097,6 +3171,17 @@ pub(crate) async fn local_stream_response( }; if check_path { + let Ok(canonical_metadata) = tokio::fs::metadata(&canonical).await else { return internal_server_error!() }; + + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt; + if opened_metadata.dev() != canonical_metadata.dev() || opened_metadata.ino() != canonical_metadata.ino() { + error!("TOCTOU race detected: file swapped during local_stream_response"); + return StatusCode::FORBIDDEN.into_response(); + } + } + let Some(library_paths) = app_state .app_config .config @@ -3110,14 +3195,12 @@ pub(crate) async fn local_stream_response( // Verify path is within allowed media directories // (requires configuration of allowed base paths) - if !is_path_within_allowed_directories(&path, &library_paths) { + if !is_path_within_allowed_directories(&canonical, &library_paths) { return StatusCode::FORBIDDEN.into_response(); } } - let Ok(mut file) = tokio::fs::File::open(&path).await else { return StatusCode::NOT_FOUND.into_response() }; - let Ok(metadata) = file.metadata().await else { return internal_server_error!() }; - let file_size = metadata.len(); + let file_size = opened_metadata.len(); let range = req_headers.get("range").and_then(|v| v.to_str().ok()).and_then(parse_range); @@ -3716,8 +3799,8 @@ pub(crate) fn create_playback_session_fingerprint( // Adaptive playlist starts need a per-initial-socket token so two players behind // the same IP/UA can watch the same HLS/DASH stream independently. The created // UserSession itself can still be non-socket-bound. - let socket_bound = - is_socket_bound_playback_session(item_type, extension) || is_session_based_playback(item_type, extension); + let session_bound = is_session_based_playback(item_type, extension); + let socket_bound = !session_bound && is_socket_bound_playback_session(item_type, extension); create_session_fingerprint(fingerprint, username, virtual_id, socket_bound) } @@ -3945,7 +4028,7 @@ pub fn empty_json_response_as_array() -> axum::http::Result allowed (account matches). + assert!(stream_url_matches_provider( + "http://cdn.example/live/selected-user/selected-pass/123.ts", + &provider + )); + } + #[test] fn stream_url_matches_provider_rejects_open_external_cdn_url_without_account_signature_for_xtream() { let provider = test_runtime_provider("http://provider.example", "selected-user", "selected-pass"); @@ -4248,6 +4366,46 @@ mod tests { ); } + #[test] + fn get_stream_alternative_url_rewrites_timeshift_path_credentials_for_alias_account() { + let input = ConfigInput { + name: "source".intern(), + url: "http://provider.example".to_string(), + username: Some("source-user".to_string()), + password: Some("source-pass".to_string()), + input_type: InputType::Xtream, + ..ConfigInput::default() + }; + let alias = test_runtime_provider("http://alias.example", "alias-user", "alias-pass"); + let stream_url = "http://provider.example/timeshift/source-user/source-pass/30/2026-06-15:20-00/123.ts"; + + let rewritten = get_stream_alternative_url(stream_url, &input, &alias); + assert_eq!( + rewritten, + Some("http://alias.example/timeshift/alias-user/alias-pass/30/2026-06-15:20-00/123.ts".to_string()) + ); + } + + #[test] + fn get_stream_alternative_url_rewrites_future_route_path_credentials_for_alias_account() { + let input = ConfigInput { + name: "source".intern(), + url: "http://provider.example".to_string(), + username: Some("source-user".to_string()), + password: Some("source-pass".to_string()), + input_type: InputType::Xtream, + ..ConfigInput::default() + }; + let alias = test_runtime_provider("http://alias.example", "alias-user", "alias-pass"); + let stream_url = "http://provider.example/future-route/source-user/source-pass/opaque/123.ts"; + + let rewritten = get_stream_alternative_url(stream_url, &input, &alias); + assert_eq!( + rewritten, + Some("http://alias.example/future-route/alias-user/alias-pass/opaque/123.ts".to_string()) + ); + } + #[test] fn get_stream_alternative_url_keeps_open_external_playlist_url_for_m3u() { let input = ConfigInput { @@ -4483,6 +4641,7 @@ mod tests { user_priority: 0, connection_kind: crate::api::model::ConnectionKind::Normal, session_owner: Some("vod-session"), + accept_requested_stream_url: false, }, ) .await; @@ -4504,6 +4663,7 @@ mod tests { user_priority: 0, connection_kind: crate::api::model::ConnectionKind::Normal, session_owner: Some("live-session"), + accept_requested_stream_url: false, }, ) .await; @@ -4544,6 +4704,7 @@ mod tests { user_priority: 0, connection_kind: crate::api::model::ConnectionKind::Normal, session_owner: Some("live-session"), + accept_requested_stream_url: false, }, ) .await; @@ -4582,6 +4743,7 @@ mod tests { user_priority: 0, connection_kind: crate::api::model::ConnectionKind::Normal, session_owner: Some("live-session"), + accept_requested_stream_url: false, }, ) .await; @@ -4592,6 +4754,45 @@ mod tests { app_state.active_provider.release_connection(&addr).await; } + #[tokio::test] + async fn resolve_streaming_strategy_accepts_session_requested_stream_url() { + let app_state = create_test_dual_provider_app_state(); + let input_name = "provider_1".intern(); + let input = app_state + .app_config + .sources + .load() + .get_input_by_name(&input_name) + .cloned() + .unwrap_or_else(|| unreachable!()); + let addr: SocketAddr = "127.0.0.1:55306".parse().unwrap_or_else(|_| unreachable!()); + let trusted_url = "http://unmapped.example/live/user1/pass1/100.ts"; + let strategy = resolve_streaming_strategy( + &app_state, + trusted_url, + &create_test_fingerprint(addr), + &input, + StreamingAcquireOptions { + force_provider: None, + allow_forced_provider_fallback: false, + allow_provider_grace: false, + user_priority: 0, + connection_kind: crate::api::model::ConnectionKind::Normal, + session_owner: Some("live-session"), + accept_requested_stream_url: true, + }, + ) + .await; + + let ProviderStreamState::Available(Some(provider), url) = strategy.provider_stream_state else { + panic!("session-requested URL should be accepted for the pinned provider") + }; + assert_eq!(provider.as_ref(), "provider_1"); + assert_eq!(url.as_ref(), trusted_url); + + app_state.active_provider.release_connection(&addr).await; + } + #[test] fn test_should_allow_exhausted_shared_reconnect_only_for_matching_shared_session() { let session = UserSession { @@ -4601,6 +4802,7 @@ mod tests { virtual_id: 282, provider: Arc::::from("provider"), stream_url: Arc::::from("http://provider/live/449924.ts"), + provider_session_headers: HashMap::new(), addr: "127.0.0.1:1234".parse().unwrap_or_else(|_| unreachable!()), socket_bound: false, active_addrs: vec!["127.0.0.1:1234".parse().unwrap_or_else(|_| unreachable!())], @@ -4795,7 +4997,7 @@ mod tests { let tokens = CancelTokens::default(); let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone())); - let (manual_update_sender, _) = mpsc::channel::>(1); + let (manual_update_sender, _) = mpsc::channel::(1); Arc::new(AppState { forced_targets: Arc::new(ArcSwap::from_pointee(ProcessTargets { @@ -4908,6 +5110,7 @@ mod tests { PlaylistItemType::LiveHls => "http://provider-1.example/live/42.m3u8", _ => "http://provider-1.example/live/42.ts", }), + provider_session_headers: HashMap::new(), addr: "127.0.0.1:55555".parse().unwrap_or_else(|_| unreachable!()), socket_bound: item_type.uses_socket_bound_session(), active_addrs: Vec::new(), @@ -7447,6 +7650,85 @@ mod tests { ); } + /// Regression test for: when a catchup request fails upstream (e.g. provider returns + /// 4xx/5xx) the connection-slot is released, but the provider account was being + /// pinned via `refresh_provider_reservation` for `catchup_session_ttl_secs`. This + /// blocked other sessions of the same family from acquiring the same provider even + /// though the slot was already free. The fix delegates the pinning decision to + /// `should_pin_provider_for_session` and skips the reservation when the response + /// is a non-Provisioning custom video (failure fallback). Provisioning custom videos + /// must keep their reservation since they represent a successful provider handoff. + #[tokio::test] + async fn should_pin_provider_for_session_skips_reservation_on_failure_custom_video() { + let app_state = create_test_app_state(); + let no_video_details = StreamDetails { + stream: None, + stream_info: Some((Vec::new(), StatusCode::OK, None, None)), + provider_name: Some("provider_1".intern()), + request_url: None, + session_headers: None, + grace_period: GracePeriodOptions::default(), + provider_grace_active: false, + disable_provider_grace: false, + reconnect_flag: None, + provider_handle: None, + grace_resolution_context: None, + }; + assert!( + should_pin_provider_for_session(&no_video_details, &app_state, PlaylistItemType::Catchup), + "a real provider stream (no CustomVideoStreamType) must pin the provider" + ); + + let provisioning_details = StreamDetails { + stream: None, + stream_info: Some(( + Vec::new(), + StatusCode::OK, + None, + Some(CustomVideoStreamType::Provisioning), + )), + provider_name: Some("provider_1".intern()), + request_url: None, + session_headers: None, + grace_period: GracePeriodOptions::default(), + provider_grace_active: false, + disable_provider_grace: false, + reconnect_flag: None, + provider_handle: None, + grace_resolution_context: None, + }; + assert!( + should_pin_provider_for_session(&provisioning_details, &app_state, PlaylistItemType::Catchup), + "a Provisioning custom video represents a successful provider handoff and must pin" + ); + + for failure_type in [ + CustomVideoStreamType::ChannelUnavailable, + CustomVideoStreamType::ProviderConnectionsExhausted, + CustomVideoStreamType::UserConnectionsExhausted, + CustomVideoStreamType::UserAccountExpired, + CustomVideoStreamType::LowPriorityPreempted, + ] { + let failure_details = StreamDetails { + stream: None, + stream_info: Some((Vec::new(), StatusCode::BAD_REQUEST, None, Some(failure_type))), + provider_name: Some("provider_1".intern()), + request_url: None, + session_headers: None, + grace_period: GracePeriodOptions::default(), + provider_grace_active: false, + disable_provider_grace: false, + reconnect_flag: None, + provider_handle: None, + grace_resolution_context: None, + }; + assert!( + !should_pin_provider_for_session(&failure_details, &app_state, PlaylistItemType::Catchup), + "{failure_type:?} is a failure fallback — must NOT pin the provider" + ); + } + } + #[tokio::test] async fn local_stream_response_disables_response_compression() { let app_state = create_test_app_state(); @@ -8102,7 +8384,7 @@ mod tests { } #[test] - fn adaptive_playback_session_fingerprint_is_unique_per_initial_socket() { + fn adaptive_playback_session_fingerprint_is_logical_across_initial_sockets() { let Some(first_addr) = "127.0.0.1:55177".parse().ok() else { return; }; @@ -8115,9 +8397,10 @@ mod tests { let first_token = create_playback_session_fingerprint(&first, "user1", 7002, PlaylistItemType::Live, Some(HLS_EXT)); let second_token = create_playback_session_fingerprint(&second, "user1", 7002, PlaylistItemType::Live, Some(HLS_EXT)); - assert_ne!(first_token, second_token); - assert!(first_token.contains(&first.addr.to_string())); - assert!(second_token.contains(&second.addr.to_string())); + assert_eq!(first_token, second_token); + assert!(first_token.contains(&first.key)); + assert!(!first_token.contains(&first.addr.to_string())); + assert!(!second_token.contains(&second.addr.to_string())); } #[test] @@ -8355,6 +8638,7 @@ mod tests { virtual_id: 9001, provider: "provider-a".intern(), stream_url: "http://localhost/movie.mkv".intern(), + provider_session_headers: HashMap::new(), addr: seek, socket_bound: false, active_addrs: vec![primary, overlap, seek, overlap], diff --git a/backend/src/api/endpoints/custom_video_stream_api.rs b/backend/src/api/endpoints/custom_video_stream_api.rs index 5b65b2781..be0bc4eb7 100644 --- a/backend/src/api/endpoints/custom_video_stream_api.rs +++ b/backend/src/api/endpoints/custom_video_stream_api.rs @@ -155,7 +155,7 @@ mod tests { downloads: CancellationToken::new(), }; let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone())); - let (manual_update_sender, _) = mpsc::channel::>(1); + let (manual_update_sender, _) = mpsc::channel::(1); Arc::new(AppState { forced_targets: Arc::new(ArcSwap::from_pointee(crate::model::ProcessTargets { diff --git a/backend/src/api/endpoints/download_api.rs b/backend/src/api/endpoints/download_api.rs index 414efb2f9..c92b79e08 100644 --- a/backend/src/api/endpoints/download_api.rs +++ b/backend/src/api/endpoints/download_api.rs @@ -1988,7 +1988,7 @@ mod tests { let tokens = CancelTokens::default(); let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone())); - let (manual_update_sender, _) = mpsc::channel::>(1); + let (manual_update_sender, _) = mpsc::channel::(1); Arc::new(AppState { forced_targets: Arc::new(ArcSwap::from_pointee(ProcessTargets { diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index b529a2f2a..d2678c8a2 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -24,7 +24,7 @@ use shared::{ model::{PlaylistItemType, StreamChannel, TargetType, UserConnectionPermission, XtreamCluster}, utils::{is_hls_url, replace_url_extension, sanitize_sensitive_info, Internable, CUSTOM_VIDEO_PREFIX, HLS_EXT}, }; -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc}; use url::Url; use shared::model::ConnectFailureReason; use crate::auth::check_network_access_only; @@ -100,6 +100,22 @@ fn hls_response(hls_content: String) -> impl IntoResponse + Send { .body(hls_content)) } +fn extract_hls_provider_session_headers(headers: &HeaderMap) -> HashMap { + let cookies = headers + .get_all(axum::http::header::SET_COOKIE) + .iter() + .filter_map(|value| value.to_str().ok()) + .filter_map(|value| value.split(';').next().map(str::trim)) + .filter(|value| !value.is_empty()) + .collect::>(); + + let mut session_headers = HashMap::new(); + if !cookies.is_empty() { + session_headers.insert(String::from("cookie"), cookies.join("; ")); + } + session_headers +} + async fn release_prepared_hls_manifest_session( app_state: &Arc, username: &str, @@ -321,29 +337,27 @@ pub(in crate::api) async fn handle_hls_stream_request( let input_source = InputSource::from(input).with_url(request_url); let use_manual_redirects = app_state.should_use_manual_redirects(); let download_result = if use_manual_redirects { - request::download_text_content_with_manual_redirects( + request::download_text_content_with_manual_redirects_and_headers( &app_state.app_config, &app_state.http_client_no_redirect.load(), &input_source, Some(&headers), - None, false, MAX_MANUAL_REDIRECTS, ) .await } else { - request::download_text_content( + request::download_text_content_with_headers( &app_state.app_config, &app_state.http_client.load(), &input_source, Some(&headers), - None, false, ) .await }; match download_result { - Ok((content, response_url)) => { + Ok((content, response_url, response_headers)) => { let encrypt_secret = app_state.get_encrypt_secret(); let base_url = server_info.get_base_url(); let rewrite_hls_props = RewriteHlsProps { @@ -358,6 +372,13 @@ pub(in crate::api) async fn handle_hls_stream_request( }; let hls_content = rewrite_hls(user, &rewrite_hls_props); if let Some(session_token) = session_token.as_deref() { + let session_headers = extract_hls_provider_session_headers(&response_headers); + if !session_headers.is_empty() { + app_state + .active_users + .update_session_provider_headers(&user.username, session_token, &session_headers) + .await; + } release_prepared_hls_manifest_session(app_state, &user.username, session_token, &fingerprint.addr).await; } hls_response(hls_content).into_response() @@ -714,7 +735,8 @@ pub fn hls_api_register() -> axum::Router> { #[cfg(test)] mod tests { - use super::m3u_archive_epg_reference_ts; + use super::{extract_hls_provider_session_headers, m3u_archive_epg_reference_ts}; + use axum::http::HeaderMap; #[test] fn archive_epg_reference_supports_query_and_path_formats() { @@ -740,4 +762,15 @@ mod tests { fn archive_epg_reference_rejects_plain_start_queries() { assert_eq!(m3u_archive_epg_reference_ts("http://provider/live/42.m3u8?start=1700000000"), None); } + + #[test] + fn extract_hls_provider_session_headers_converts_set_cookie_to_cookie_header() { + let mut headers = HeaderMap::new(); + headers.append("set-cookie", "sid=abc; Path=/; HttpOnly".parse().expect("valid cookie")); + headers.append("set-cookie", "pref=1; Secure".parse().expect("valid cookie")); + + let session_headers = extract_hls_provider_session_headers(&headers); + + assert_eq!(session_headers.get("cookie").map(String::as_str), Some("sid=abc; pref=1")); + } } diff --git a/backend/src/api/endpoints/library_api.rs b/backend/src/api/endpoints/library_api.rs index ae682e478..3aa002184 100644 --- a/backend/src/api/endpoints/library_api.rs +++ b/backend/src/api/endpoints/library_api.rs @@ -1,11 +1,13 @@ use crate::{api::{ library_scan::{spawn_library_scan, LibraryScanTaskOptions}, - model::{AppState, EventMessage}, + model::AppState, }, auth::permission_layer, library::{resolve_metadata_storage_path, LibraryProcessor, MetadataStorage}}; use axum::response::IntoResponse; use log::{debug, warn}; use serde_json::json; -use shared::model::{permission::Permission, LibraryScanRequest, LibraryScanSummary, LibraryScanSummaryStatus, LibraryStatus}; +use shared::model::{ + permission::Permission, LibraryScanRequest, LibraryStatus, OperationRunAccepted, +}; use std::sync::Arc; // Triggers a library scan @@ -17,12 +19,6 @@ async fn scan_library( let Some(permit) = app_state.update_guard.try_library() else { warn!("Library update already in progress; update skipped."); - let response = LibraryScanSummary { - status: LibraryScanSummaryStatus::Error, - message: "Library update already in progress.".to_string(), - result: None, - }; - let _ = app_state.event_manager.send_event(EventMessage::LibraryScanProgress(response)); return ( axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Library update already in progress.".to_string()})), @@ -36,12 +32,6 @@ async fn scan_library( match config.library.as_ref() { Some(lib) if lib.enabled => (lib.clone(), config.metadata_update.clone(), config.storage_dir.clone()), _ => { - let response = LibraryScanSummary { - status: LibraryScanSummaryStatus::Error, - message: "Library is not enabled".to_string(), - result: None, - }; - let _ = app_state.event_manager.send_event(EventMessage::LibraryScanProgress(response)); return ( axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Library is not enabled".to_string()})), @@ -61,7 +51,7 @@ async fn scan_library( permit, ); - axum::http::StatusCode::ACCEPTED.into_response() + (axum::http::StatusCode::ACCEPTED, axum::Json(OperationRunAccepted {})).into_response() } /// Gets Library status @@ -184,4 +174,4 @@ pub fn library_api_register( .route("/library/status", axum::routing::get(get_library_status)) .route("/library/thumbnail/{uuid}", axum::routing::get(get_thumbnail)), } -} \ No newline at end of file +} diff --git a/backend/src/api/endpoints/v1_api_playlist.rs b/backend/src/api/endpoints/v1_api_playlist.rs index 7069a7f9a..8bb69e8dd 100644 --- a/backend/src/api/endpoints/v1_api_playlist.rs +++ b/backend/src/api/endpoints/v1_api_playlist.rs @@ -16,7 +16,7 @@ use serde_json::json; use shared::utils::deobfuscate_text; use shared::{ model::{ - permission::Permission, EpgChannel, + permission::Permission, EpgChannel, OperationRunAccepted, InputType, PlaylistEpgRequest, PlaylistRequest, PlaylistUrlResolveRequest, ProxyType, TargetType, UiPlaylistItem, XtreamCluster, }, @@ -213,11 +213,14 @@ async fn playlist_update( // Deduplicate rapid clicks: the channel has capacity 1, so at most one // update is queued at any time. Additional requests while the channel // is full are silently dropped — the pending run already covers them. - match app_state.manual_update_sender.try_send(valid_targets) { - Ok(()) => axum::http::StatusCode::ACCEPTED.into_response(), + match app_state + .manual_update_sender + .try_send(crate::api::model::ManualPlaylistUpdateRequest { targets: valid_targets }) + { + Ok(()) => (axum::http::StatusCode::ACCEPTED, axum::Json(OperationRunAccepted {})).into_response(), Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { debug!("Manual playlist update deduplicated: an update is already pending or running"); - axum::http::StatusCode::ACCEPTED.into_response() + (axum::http::StatusCode::ACCEPTED, axum::Json(OperationRunAccepted {})).into_response() } Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { debug!("Manual playlist update rejected: worker channel closed (server shutting down)"); @@ -765,7 +768,7 @@ mod tests { downloads: CancellationToken::new(), }; let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone())); - let (manual_update_sender, _) = mpsc::channel::>(1); + let (manual_update_sender, _) = mpsc::channel::(1); Arc::new(AppState { forced_targets: Arc::new(ArcSwap::from_pointee(crate::model::ProcessTargets { diff --git a/backend/src/api/endpoints/websocket_api.rs b/backend/src/api/endpoints/websocket_api.rs index 32e892dac..9dc47573b 100644 --- a/backend/src/api/endpoints/websocket_api.rs +++ b/backend/src/api/endpoints/websocket_api.rs @@ -110,6 +110,10 @@ fn websocket_can_receive_runtime_events(mem: &ProtocolHandlerMemory, event: &Eve EventMessage::DownloadsUpdate(_) | EventMessage::DownloadsDeltaUpdate(_) => { mem.permissions.contains(Permission::DownloadRead) } + EventMessage::PlaylistUpdateProgress(_) | EventMessage::PlaylistUpdate(_) => { + mem.permissions.contains(Permission::PlaylistWrite) + } + EventMessage::LibraryScanProgress(_) => mem.permissions.contains(Permission::LibraryWrite), _ => mem.permissions.contains(Permission::SystemRead), } } @@ -364,10 +368,10 @@ async fn handle_event_message( ) .await?; } - EventMessage::PlaylistUpdateProgress(target, msg) => { + EventMessage::PlaylistUpdateProgress(progress) => { send_event_response( socket, - ProtocolMessage::PlaylistUpdateProgressResponse(target, msg), + ProtocolMessage::PlaylistUpdateProgressResponse(progress), "Playlist update progress event", ) .await?; @@ -380,10 +384,10 @@ async fn handle_event_message( ) .await?; } - EventMessage::LibraryScanProgress(summary) => { + EventMessage::LibraryScanProgress(progress) => { send_event_response( socket, - ProtocolMessage::LibraryScanProgressResponse(summary), + ProtocolMessage::LibraryScanProgressResponse(progress), "Library scan progress event", ) .await?; @@ -511,7 +515,11 @@ async fn handle_user_action(app_state: &Arc, cmd: UserCommand) -> bool mod tests { use super::websocket_can_receive_runtime_events; use crate::api::model::EventMessage; - use shared::model::{DownloadsDelta, DownloadsResponse, FileDownloadDto, Permission, ProtocolHandlerMemory, TaskKindDto, TaskPriorityDto, TransferStatusDto, UserRole}; + use shared::model::{ + DownloadsDelta, DownloadsResponse, FileDownloadDto, LibraryScanProgressEvent, LibraryScanSummary, + LibraryScanSummaryStatus, Permission, PlaylistUpdateProgressEvent, ProtocolHandlerMemory, TaskKindDto, + TaskPriorityDto, TransferStatusDto, UserRole, + }; #[test] fn test_websocket_runtime_events_allowed_for_admin() { @@ -568,6 +576,43 @@ mod tests { )); } + #[test] + fn test_websocket_playlist_progress_allowed_for_playlist_write_without_system_read() { + let mut mem = ProtocolHandlerMemory { + permissions: Permission::PlaylistWrite.into(), + ..ProtocolHandlerMemory::default() + }; + mem.role = UserRole::User; + + assert!(websocket_can_receive_runtime_events( + &mem, + &EventMessage::PlaylistUpdateProgress(PlaylistUpdateProgressEvent { + target: "target".to_string(), + message: "step".to_string(), + }) + )); + } + + #[test] + fn test_websocket_library_progress_allowed_for_library_write_without_system_read() { + let mut mem = ProtocolHandlerMemory { + permissions: Permission::LibraryWrite.into(), + ..ProtocolHandlerMemory::default() + }; + mem.role = UserRole::User; + + assert!(websocket_can_receive_runtime_events( + &mem, + &EventMessage::LibraryScanProgress(LibraryScanProgressEvent { + summary: LibraryScanSummary { + status: LibraryScanSummaryStatus::Success, + message: "done".to_string(), + result: None, + }, + }) + )); + } + #[test] fn test_websocket_download_updates_allowed_for_download_read_user() { let mut mem = ProtocolHandlerMemory { diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index 04f58848b..470496ea9 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -677,9 +677,7 @@ pub(crate) async fn stream_epg_api( return stream_epg_bad_request(&format!("unknown target_id: {target_id}")); }; - let epg_path = get_epg_path_for_target_by_type(config.as_ref(), &target, TargetType::M3u) - .or_else(|| get_epg_path_for_target_by_type(config.as_ref(), &target, TargetType::Xtream)); - let Some(epg_path) = epg_path else { + let Some(epg_path) = get_epg_path_for_target(config.as_ref(), &target) else { entries.extend(empty_stream_epg_entries(target_id, &items)); continue; }; @@ -842,6 +840,29 @@ mod tests { } } + fn test_target_with_xtream_only() -> ConfigTarget { + ConfigTarget { + id: 1, + enabled: true, + name: "xtream-only".to_string(), + options: None, + sort: None, + filter: Filter::default(), + output: vec![TargetOutput::Xtream(XtreamTargetOutput { + flags: XtreamTargetFlagsSet::new(), + trakt: None, + filter: None, + })], + rename: None, + mapping_ids: None, + mapping: Arc::new(ArcSwapOption::new(None)), + favourites: None, + processing_order: ProcessingOrder::default(), + watch: None, + use_memory_cache: false, + } + } + fn test_config_with_storage(storage_dir: &str) -> Config { Config { storage_dir: storage_dir.to_string(), @@ -892,6 +913,26 @@ mod tests { assert_eq!(picked, xtream_epg); } + #[test] + fn get_epg_path_for_target_xtream_only_target_uses_xtream_path() { + // Regression: stream_epg_api used to probe m3u first regardless of the target's + // configured outputs, producing a misleading "Can't find epg file" TRACE for + // xtream-only targets. After the fix, get_epg_path_for_target iterates + // target.output and must not look for an m3u epg at all. + let dir = tempdir().expect("temp dir"); + let config = test_config_with_storage(dir.path().to_string_lossy().as_ref()); + let target = test_target_with_xtream_only(); + + let xtream_storage = crate::repository::xtream_get_storage_path(&config, &target.name).expect("xtream storage"); + let xtream_epg = crate::repository::xtream_get_epg_file_path_for_target(&xtream_storage); + + fs::create_dir_all(&xtream_storage).expect("create xtream dir"); + fs::write(&xtream_epg, b"xtream").expect("write xtream epg"); + + let picked = get_epg_path_for_target(&config, &target).expect("xtream path"); + assert_eq!(picked, xtream_epg); + } + #[test] fn test_group_stream_epg_items_rejects_empty_requests() { let err = group_stream_epg_items(Vec::new()).unwrap_err(); diff --git a/backend/src/api/library_scan.rs b/backend/src/api/library_scan.rs index 79874d698..38c2e5e69 100644 --- a/backend/src/api/library_scan.rs +++ b/backend/src/api/library_scan.rs @@ -4,7 +4,7 @@ use crate::{ model::{LibraryConfig, MetadataUpdateConfig}, }; use log::{error, info}; -use shared::model::{LibraryScanSummary, LibraryScanSummaryStatus}; +use shared::model::{LibraryScanProgressEvent, LibraryScanSummary, LibraryScanSummaryStatus}; use std::sync::Arc; pub(crate) struct LibraryScanTaskOptions { @@ -37,7 +37,8 @@ pub(crate) fn spawn_library_scan( ), result: Some(result), }; - let _ = event_manager.send_event(EventMessage::LibraryScanProgress(response)); + let _ = event_manager + .send_event(EventMessage::LibraryScanProgress(LibraryScanProgressEvent { summary: response })); } Err(err) => { error!("{prefix}Library scan failed: {err}"); @@ -46,7 +47,8 @@ pub(crate) fn spawn_library_scan( message: format!("{prefix}Scan failed: {err}"), result: None, }; - let _ = event_manager.send_event(EventMessage::LibraryScanProgress(response)); + let _ = event_manager + .send_event(EventMessage::LibraryScanProgress(LibraryScanProgressEvent { summary: response })); } } }); diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index 9c5e8a055..0d884098d 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -21,7 +21,7 @@ use crate::{ model::{ create_cache, create_http_client, create_http_client_no_redirect, exec_provider_dns, ActiveProviderManager, ActiveUserManager, AppState, CancelTokens, ConnectionManager, DownloadQueue, - EventManager, EventMessage, HdHomerunAppState, MetadataUpdateManager, PlaylistStorageState, + EventManager, EventMessage, HdHomerunAppState, ManualPlaylistUpdateRequest, MetadataUpdateManager, PlaylistStorageState, SharedStreamManager, UpdateGuard, exec_qos_aggregation, }, panel_api::sync_panel_api_exp_dates_on_boot, @@ -258,7 +258,7 @@ async fn healthcheck() -> impl axum::response::IntoResponse { axum::Json(create_ async fn create_shared_data( app_config: &Arc, forced_targets: &Arc, -) -> Result<(AppState, mpsc::Receiver>), TuliproxError> { +) -> Result<(AppState, mpsc::Receiver), TuliproxError> { let config = app_config.config.load(); let downloads_state_file = std::path::PathBuf::from(&config.storage_dir).join("downloads_state.json"); @@ -304,7 +304,7 @@ async fn create_shared_data( let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone())); let cancel_tokens = Arc::new(ArcSwap::from_pointee(tokens)); - let (manual_update_sender, manual_update_rx) = mpsc::channel::>(1); + let (manual_update_sender, manual_update_rx) = mpsc::channel::(1); let app_state = AppState { forced_targets: Arc::new(ArcSwap::new(Arc::clone(forced_targets))), @@ -334,13 +334,13 @@ async fn create_shared_data( async fn run_manual_update_worker( client: reqwest::Client, app_state: Arc, - mut rx: mpsc::Receiver>, + mut rx: mpsc::Receiver, ) { - while let Some(targets) = rx.recv().await { + while let Some(request) = rx.recv().await { exec_processing( &client, Arc::clone(&app_state.app_config), - targets, + request.targets, Some(Arc::clone(&app_state.event_manager)), Some(Arc::clone(&app_state)), Some(Arc::clone(&app_state.playlists)), diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index eb642c2e1..ad3a26e6d 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -128,6 +128,7 @@ pub struct UserSession { pub virtual_id: u32, pub provider: Arc, pub stream_url: Arc, + pub provider_session_headers: HashMap, pub addr: SocketAddr, pub socket_bound: bool, pub active_addrs: Vec, @@ -662,6 +663,14 @@ impl ActiveUserManager { key } + fn admission_gate_key(username: &str) -> String { + let mut key = String::with_capacity(username.len() + 11); + key.push_str("admission"); + key.push('\0'); + key.push_str(username); + key + } + fn cleanup_idle_transition_gates(transition_gates: &mut HashMap>>) { transition_gates.retain(|_, gate| Arc::strong_count(gate) > 1); } @@ -680,6 +689,20 @@ impl ActiveUserManager { gate.lock_owned().await } + pub(crate) async fn acquire_user_admission(&self, username: &str) -> tokio::sync::OwnedMutexGuard<()> { + let key = Self::admission_gate_key(username); + let gate = { + let mut transition_gates = self.transition_gates.lock().await; + Self::cleanup_idle_transition_gates(&mut transition_gates); + Arc::clone( + transition_gates + .entry(key) + .or_insert_with(|| Arc::new(Mutex::new(()))), + ) + }; + gate.lock_owned().await + } + fn should_reuse_stream_for_session(existing_stream: &StreamInfo, incoming_channel: &StreamChannel) -> bool { existing_stream.channel.item_type.requires_provider_affinity() || incoming_channel.item_type.requires_provider_affinity() } @@ -1586,6 +1609,7 @@ impl ActiveUserManager { virtual_id: params.virtual_id, provider: params.provider.intern(), stream_url: params.stream_url.intern(), + provider_session_headers: HashMap::new(), addr: *params.addr, socket_bound: params.socket_bound, active_addrs: vec![*params.addr], @@ -1961,11 +1985,17 @@ impl ActiveUserManager { session.socket_bound = socket_bound; remember_session_addr(session, *addr); Self::bump_session_transition_version(session); + let mut reset_provider_session_headers = false; if &*session.stream_url != stream_url { session.stream_url = stream_url.intern(); + reset_provider_session_headers = true; } if &*session.provider != provider { session.provider = provider.intern(); + reset_provider_session_headers = true; + } + if reset_provider_session_headers { + session.provider_session_headers.clear(); } // Normalize stale lifecycle states on session refresh. // Expired, PendingProvider, and Preserved sessions cannot stay in those states @@ -2673,6 +2703,24 @@ impl ActiveUserManager { self.update_user_session(username, token).await } + pub async fn update_session_provider_headers( + &self, + username: &str, + token: &str, + provider_session_headers: &HashMap, + ) -> bool { + let mut user_connections = self.connections.write().await; + let Some(connection_data) = user_connections.by_key.get_mut(username) else { + return false; + }; + let Some(session) = connection_data.sessions.iter_mut().find(|session| session.token == token) else { + return false; + }; + session.provider_session_headers.clone_from(provider_session_headers); + session.ts = current_time_secs(); + true + } + pub async fn pending_provider_version(&self, username: &str, token: &str) -> Option { let user_connections = self.connections.read().await; let connection_data = user_connections.by_key.get(username)?; @@ -3258,7 +3306,7 @@ mod tests { model::{PlaylistItemType, StreamChannel, StreamInfo, XtreamCluster}, utils::Internable, }; - use std::{borrow::Cow, sync::Arc}; + use std::{borrow::Cow, collections::HashMap, sync::Arc}; use shared::model::ProxyType; fn test_channel(virtual_id: u32) -> StreamChannel { @@ -3326,6 +3374,7 @@ mod tests { virtual_id: 7001, provider: "provider-a".intern(), stream_url: "http://localhost/live.m3u8".intern(), + provider_session_headers: HashMap::new(), addr, socket_bound: false, active_addrs: vec![addr], @@ -3387,6 +3436,7 @@ mod tests { virtual_id: 7002, provider: "provider-a".intern(), stream_url: "http://localhost/live.m3u8".intern(), + provider_session_headers: HashMap::new(), addr, socket_bound: false, active_addrs: vec![addr], @@ -3430,6 +3480,134 @@ mod tests { ); } + #[tokio::test] + async fn update_session_provider_headers_updates_existing_session_and_timestamp() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let addr: SocketAddr = "127.0.0.1:55402".parse().unwrap_or_else(|_| unreachable!()); + let mut user = ProxyUserCredentials::default(); + user.username = "user-provider-headers".to_string(); + + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: "tok-provider-headers", + virtual_id: 7003, + provider: "provider-a", + stream_url: "http://localhost/live.m3u8", + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, + }) + .await; + + let before = manager + .get_and_update_user_session(&user.username, "tok-provider-headers") + .await + .expect("session should exist"); + let previous_ts = before.ts; + let headers = HashMap::from([(String::from("cookie"), String::from("sid=abc"))]); + + assert!(manager + .update_session_provider_headers(&user.username, "tok-provider-headers", &headers) + .await); + + let after = manager + .get_and_update_user_session(&user.username, "tok-provider-headers") + .await + .expect("session should exist"); + assert_eq!(after.provider_session_headers, headers); + assert!(after.ts >= previous_ts); + } + + #[tokio::test] + async fn update_session_provider_headers_returns_false_for_missing_user_or_token() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + let headers = HashMap::from([(String::from("cookie"), String::from("sid=abc"))]); + + assert!(!manager + .update_session_provider_headers("missing-user", "missing-token", &headers) + .await); + + let addr: SocketAddr = "127.0.0.1:55403".parse().unwrap_or_else(|_| unreachable!()); + let mut user = ProxyUserCredentials::default(); + user.username = "user-missing-token".to_string(); + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: "tok-existing", + virtual_id: 7004, + provider: "provider-a", + stream_url: "http://localhost/live.m3u8", + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, + }) + .await; + + assert!(!manager + .update_session_provider_headers(&user.username, "tok-missing", &headers) + .await); + } + + #[tokio::test] + async fn create_user_session_clears_provider_headers_when_provider_or_stream_url_changes() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let addr: SocketAddr = "127.0.0.1:55404".parse().unwrap_or_else(|_| unreachable!()); + let mut user = ProxyUserCredentials::default(); + user.username = "user-provider-header-reset".to_string(); + let headers = HashMap::from([(String::from("cookie"), String::from("sid=abc"))]); + + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: "tok-reset", + virtual_id: 7005, + provider: "provider-a", + stream_url: "http://localhost/live-a.m3u8", + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, + }) + .await; + assert!(manager + .update_session_provider_headers(&user.username, "tok-reset", &headers) + .await); + + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: "tok-reset", + virtual_id: 7005, + provider: "provider-b", + stream_url: "http://localhost/live-b.m3u8", + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, + }) + .await; + + let session = manager + .get_and_update_user_session(&user.username, "tok-reset") + .await + .expect("session should exist"); + assert!(session.provider_session_headers.is_empty()); + } + /// `terminate_session` expires a session and removes it. #[tokio::test] async fn terminate_session_expires_and_removes_session() { @@ -7787,6 +7965,7 @@ mod tests { virtual_id: 9002, provider: "provider-a".intern(), stream_url: "http://localhost/stream.ts".intern(), + provider_session_headers: HashMap::new(), addr, socket_bound: false, active_addrs: vec![addr], @@ -7828,6 +8007,7 @@ mod tests { virtual_id: 9003, provider: "provider-a".intern(), stream_url: "http://localhost/stream.ts".intern(), + provider_session_headers: HashMap::new(), addr, socket_bound: false, active_addrs: vec![addr], @@ -7902,6 +8082,7 @@ mod tests { virtual_id: 9004, provider: "provider-a".intern(), stream_url: "http://localhost/stream.ts".intern(), + provider_session_headers: HashMap::new(), addr, socket_bound: false, active_addrs: vec![addr], diff --git a/backend/src/api/model/app_state.rs b/backend/src/api/model/app_state.rs index 7f7e00b38..f39ea2538 100644 --- a/backend/src/api/model/app_state.rs +++ b/backend/src/api/model/app_state.rs @@ -390,6 +390,11 @@ fn video_download_changed(a: &crate::model::VideoDownloadConfig, b: &crate::mode VideoDownloadConfigDto::from(a) != VideoDownloadConfigDto::from(b) } +#[derive(Clone)] +pub struct ManualPlaylistUpdateRequest { + pub targets: Arc, +} + #[derive(Clone)] pub struct AppState { pub forced_targets: Arc>, // as program arguments @@ -412,7 +417,7 @@ pub struct AppState { /// `try_send` deduplicates rapid clicks: if an update is already pending /// or the channel is full, the request is silently dropped so at most one /// update is queued at any time regardless of how many times the button is clicked. - pub manual_update_sender: mpsc::Sender>, + pub manual_update_sender: mpsc::Sender, } impl AppState { diff --git a/backend/src/api/model/event_manager.rs b/backend/src/api/model/event_manager.rs index ebbcdd163..4df0ef180 100644 --- a/backend/src/api/model/event_manager.rs +++ b/backend/src/api/model/event_manager.rs @@ -1,15 +1,12 @@ use crate::api::model::streams::{MeterReading, StreamMeterHandle}; use log::trace; use shared::model::{ - ActiveUserConnectionChange, ConfigType, DownloadsDelta, DownloadsResponse, LibraryScanSummary, PlaylistUpdateState, - StreamMeterEntry, SystemInfo, + ActiveUserConnectionChange, ConfigType, DownloadsDelta, DownloadsResponse, LibraryScanProgressEvent, + PlaylistUpdateProgressEvent, PlaylistUpdateState, StreamMeterEntry, SystemInfo, }; use std::{ collections::HashMap, - sync::{ - atomic::{AtomicUsize, Ordering}, - Arc, - }, + sync::{atomic::AtomicUsize, atomic::Ordering, Arc}, time::Duration, }; use tokio::sync::RwLock; @@ -26,9 +23,9 @@ pub enum EventMessage { ActiveProvider(Arc, usize), ConfigChange(ConfigType), PlaylistUpdate(PlaylistUpdateState), - PlaylistUpdateProgress(String, String), + PlaylistUpdateProgress(PlaylistUpdateProgressEvent), SystemInfoUpdate(SystemInfo), - LibraryScanProgress(LibraryScanSummary), + LibraryScanProgress(LibraryScanProgressEvent), DownloadsUpdate(DownloadsResponse), DownloadsDeltaUpdate(DownloadsDelta), InputMetadataUpdatesCompleted(Arc), diff --git a/backend/src/api/model/stream.rs b/backend/src/api/model/stream.rs index 1cee3473a..46384275f 100644 --- a/backend/src/api/model/stream.rs +++ b/backend/src/api/model/stream.rs @@ -31,6 +31,7 @@ pub struct StreamDetails { pub(crate) stream_info: ProviderStreamInfo, pub provider_name: Option>, pub request_url: Option>, + pub session_headers: Option>, pub grace_period: GracePeriodOptions, pub provider_grace_active: bool, pub disable_provider_grace: bool, @@ -51,6 +52,7 @@ impl Clone for StreamDetails { stream_info: self.stream_info.clone(), provider_name: self.provider_name.clone(), request_url: self.request_url.clone(), + session_headers: self.session_headers.clone(), grace_period: self.grace_period, provider_grace_active: self.provider_grace_active, disable_provider_grace: self.disable_provider_grace, @@ -68,6 +70,7 @@ impl StreamDetails { stream_info: None, provider_name: None, request_url: None, + session_headers: None, grace_period: grace_period_options, provider_grace_active: false, disable_provider_grace: false, diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index 0b2a664e7..2720c91e9 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/backend/src/api/model/streams/active_client_stream.rs @@ -462,6 +462,7 @@ fn create_deferred_provider_open_future( stream_url: &stream_url, req_headers, input_headers: Some(&input.headers), + session_headers: stream_details.session_headers.as_ref(), disabled_headers: disabled_headers.as_ref(), default_user_agent: default_user_agent.as_deref(), username: None, @@ -1370,7 +1371,7 @@ mod tests { let tokens = CancelTokens::default(); let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone())); - let (manual_update_sender, _) = mpsc::channel::>(1); + let (manual_update_sender, _) = mpsc::channel::(1); Arc::new(AppState { forced_targets: Arc::new(ArcSwap::from_pointee(ProcessTargets { @@ -1437,7 +1438,7 @@ mod tests { let tokens = CancelTokens::default(); let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone())); - let (manual_update_sender, _) = mpsc::channel::>(1); + let (manual_update_sender, _) = mpsc::channel::(1); Arc::new(AppState { forced_targets: Arc::new(ArcSwap::from_pointee(ProcessTargets { @@ -1511,6 +1512,7 @@ mod tests { stream_info: None, provider_name: Some(Arc::clone(provider_name)), request_url: Some("http://provider-1.example/live/1".intern()), + session_headers: None, grace_period: GracePeriodOptions { period_millis: 100, timeout_secs: 0, @@ -2361,6 +2363,7 @@ mod tests { stream_info: None, provider_name: Some(provider_name), request_url: Some("http://provider-1.example/live/2.ts".intern()), + session_headers: None, grace_period: GracePeriodOptions { period_millis: 100, timeout_secs: 0, diff --git a/backend/src/api/model/streams/provider_stream_factory.rs b/backend/src/api/model/streams/provider_stream_factory.rs index efe5cff1b..c90919c22 100644 --- a/backend/src/api/model/streams/provider_stream_factory.rs +++ b/backend/src/api/model/streams/provider_stream_factory.rs @@ -56,7 +56,7 @@ create_bitset!( #[derive(Debug, Clone)] pub struct ProviderStreamFactoryOptions { addr: SocketAddr, - // item_type: PlaylistItemType, + item_type: PlaylistItemType, flags: ProviderStreamFactoryFlagsSet, buffer_size: usize, url: Url, @@ -80,6 +80,7 @@ pub(crate) struct ProviderStreamFactoryParams<'a> { pub stream_url: &'a Url, pub req_headers: &'a HeaderMap, pub input_headers: Option<&'a HashMap>, + pub session_headers: Option<&'a HashMap>, pub disabled_headers: Option<&'a ReverseProxyDisabledHeaderConfig>, pub default_user_agent: Option<&'a str>, pub username: Option<&'a str>, @@ -98,6 +99,7 @@ impl ProviderStreamFactoryOptions { stream_url, req_headers, input_headers, + session_headers, disabled_headers, default_user_agent, username, @@ -115,8 +117,15 @@ impl ProviderStreamFactoryOptions { let requested_range = get_request_range_start_bytes(&req_headers); req_headers.remove("range"); + let merged_input_headers = merge_provider_request_headers(*input_headers, *session_headers); + // We merge configured input headers with the headers from the request. - let headers = get_request_headers(*input_headers, Some(&req_headers), *disabled_headers, *default_user_agent); + let headers = get_request_headers( + merged_input_headers.as_ref(), + Some(&req_headers), + *disabled_headers, + *default_user_agent, + ); let default_user_agent = default_user_agent .and_then(|ua| { @@ -151,7 +160,7 @@ impl ProviderStreamFactoryOptions { } Self { - // item_type, + item_type: *item_type, addr: *addr, flags, buffer_size, @@ -194,6 +203,9 @@ impl ProviderStreamFactoryOptions { #[inline] pub fn get_url_as_str(&self) -> &str { self.url.as_str() } + #[inline] + fn get_item_type(&self) -> PlaylistItemType { self.item_type } + #[inline] pub fn should_retry_provider_request(&self) -> bool { self.flags.contains(ProviderStreamFactoryFlags::RetryEnabled) } @@ -244,6 +256,23 @@ impl ProviderStreamFactoryOptions { } +fn merge_provider_request_headers( + input_headers: Option<&HashMap>, + session_headers: Option<&HashMap>, +) -> Option> { + match (input_headers, session_headers) { + (None, None) => None, + (Some(headers), None) | (None, Some(headers)) => Some(headers.clone()), + (Some(input), Some(session)) => { + let mut merged = input.clone(); + for (key, value) in session { + merged.insert(key.clone(), value.clone()); + } + Some(merged) + } + } +} + fn record_provider_open_failure( app_state: &Arc, stream_options: &ProviderStreamFactoryOptions, @@ -286,6 +315,17 @@ fn classify_provider_status_error(status: StatusCode) -> &'static str { } } +fn provider_content_type_looks_like_html(headers: &HeaderMap) -> bool { + headers + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value.split(';').next().unwrap_or_default().trim().eq_ignore_ascii_case("text/html")) +} + +fn should_reject_success_response_content_type(item_type: PlaylistItemType, headers: &HeaderMap) -> bool { + !item_type.is_live_adaptive() && provider_content_type_looks_like_html(headers) +} + #[derive(Clone, Copy, Debug)] enum ProviderStreamRequestFailure { Status { @@ -588,6 +628,17 @@ async fn provider_stream_request( debug!("{}", sanitize_sensitive_info(&message)); } if status.is_success() { + if should_reject_success_response_content_type(stream_options.get_item_type(), response.headers()) { + debug!( + "Provider returned HTML content for non-adaptive stream {}", + sanitize_sensitive_info(stream_options.get_log_url().as_ref()) + ); + return Err(ProviderStreamRequestFailure::Status { + status: StatusCode::BAD_GATEWAY, + provider_error_class: "unexpected_content_type", + serve_channel_unavailable: true, + }); + } let response_info = { // Unfortunately, the HEAD request does not work, so we need this workaround. // We need some header information from the provider, we extract the necessary headers and forward them to the client @@ -846,6 +897,7 @@ mod tests { stream_url: &stream_url, req_headers: &req_headers, input_headers: None, + session_headers: None, disabled_headers, default_user_agent: None, username: None, @@ -866,6 +918,7 @@ mod tests { stream_url: &stream_url, req_headers: &req_headers, input_headers: None, + session_headers: None, disabled_headers, default_user_agent: None, username: None, @@ -886,6 +939,7 @@ mod tests { stream_url: &stream_url, req_headers: &req_headers, input_headers: None, + session_headers: None, disabled_headers, default_user_agent: None, username: None, @@ -907,6 +961,7 @@ mod tests { stream_url: &stream_url, req_headers: &req_headers, input_headers: None, + session_headers: None, disabled_headers, default_user_agent: None, username: None, @@ -934,6 +989,7 @@ mod tests { stream_url: &stream_url, req_headers: &req_headers, input_headers: None, + session_headers: None, disabled_headers: None, default_user_agent: None, username: None, @@ -950,6 +1006,7 @@ mod tests { stream_url: &stream_url, req_headers: &req_headers, input_headers: None, + session_headers: None, disabled_headers: None, default_user_agent: None, username: None, @@ -980,6 +1037,7 @@ mod tests { stream_url: &stream_url, req_headers: &req_headers, input_headers: None, + session_headers: None, disabled_headers: None, default_user_agent: None, username: None, @@ -1010,6 +1068,7 @@ mod tests { stream_url: &stream_url, req_headers: &req_headers, input_headers: None, + session_headers: None, disabled_headers: None, default_user_agent: None, username: Some("alice"), @@ -1042,4 +1101,56 @@ mod tests { assert_eq!(info.channel.input_name.as_ref(), "input-a"); assert_eq!(info.channel.virtual_id, 77); } + + #[test] + fn html_content_type_is_rejected_for_catchup_streams() { + let mut headers = HeaderMap::new(); + headers.insert(reqwest::header::CONTENT_TYPE, "text/html; charset=UTF-8".parse().unwrap()); + + assert!(should_reject_success_response_content_type(PlaylistItemType::Catchup, &headers)); + assert!(should_reject_success_response_content_type(PlaylistItemType::Video, &headers)); + assert!(should_reject_success_response_content_type(PlaylistItemType::Live, &headers)); + } + + #[test] + fn html_content_type_is_allowed_for_live_adaptive_streams() { + let mut headers = HeaderMap::new(); + headers.insert(reqwest::header::CONTENT_TYPE, "text/html; charset=UTF-8".parse().unwrap()); + + assert!(!should_reject_success_response_content_type(PlaylistItemType::LiveHls, &headers)); + assert!(!should_reject_success_response_content_type(PlaylistItemType::LiveDash, &headers)); + } + + #[test] + fn session_headers_are_forwarded_to_provider_requests() { + let addr = "127.0.0.1:8080".parse().unwrap(); + let stream_url = Url::parse("http://example.com/live/segment.ts").unwrap(); + let req_headers = HeaderMap::new(); + let mut session_headers = HashMap::new(); + session_headers.insert(String::from("cookie"), String::from("sid=abc; pref=1")); + let stream_options = + StreamOptions { stream_retry: true, buffer_enabled: true, buffer_size: 1024, pipe_provider_stream: false }; + + let options = ProviderStreamFactoryOptions::new(&ProviderStreamFactoryParams { + addr, + item_type: PlaylistItemType::LiveHls, + share_stream: false, + stream_options: &stream_options, + stream_url: &stream_url, + req_headers: &req_headers, + input_headers: None, + session_headers: Some(&session_headers), + disabled_headers: None, + default_user_agent: None, + username: None, + client_ip: None, + stream_channel: None, + connect_failure_stage: None, + }); + + assert_eq!( + options.get_headers().get(axum::http::header::COOKIE).and_then(|value| value.to_str().ok()), + Some("sid=abc; pref=1") + ); + } } diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs index 739093f7d..2cb3dafdd 100644 --- a/backend/src/api/model/streams/shared_stream_manager.rs +++ b/backend/src/api/model/streams/shared_stream_manager.rs @@ -696,12 +696,9 @@ impl SharedStreamManager { self.get_shared_state(stream_url).await.map(|s| s.headers.clone()) } - pub async fn register_meter_uid(&self, stream_url: &str, meter_uid: u32) { - self.meter_uids.write().await.insert(stream_url.to_string(), meter_uid); - } - - pub async fn get_meter_uid(&self, stream_url: &str) -> Option { - self.meter_uids.read().await.get(stream_url).copied() + pub async fn get_or_register_meter_uid(&self, stream_url: &str, uid_factory: impl FnOnce() -> u32) -> u32 { + let mut uids = self.meter_uids.write().await; + *uids.entry(stream_url.to_string()).or_insert_with(uid_factory) } async fn forget_subscriber_addr(&self, addr: &SocketAddr) -> Option> { diff --git a/backend/src/api/panel_api.rs b/backend/src/api/panel_api.rs index 53197d03f..34ecbca9b 100644 --- a/backend/src/api/panel_api.rs +++ b/backend/src/api/panel_api.rs @@ -3754,6 +3754,7 @@ pub fn create_panel_api_provisioning_stream_details( stream_info, provider_name, request_url: None, + session_headers: None, grace_period: *grace_period_options, provider_grace_active: false, disable_provider_grace: true, @@ -3779,6 +3780,7 @@ pub fn create_panel_api_provisioning_stream_details( stream_info, provider_name, request_url: None, + session_headers: None, grace_period: *grace_period_options, provider_grace_active: false, disable_provider_grace: true, diff --git a/backend/src/model/config/web_ui.rs b/backend/src/model/config/web_ui.rs index 4dc157dc3..e4af3994b 100644 --- a/backend/src/model/config/web_ui.rs +++ b/backend/src/model/config/web_ui.rs @@ -77,7 +77,7 @@ pub struct WebUiConfig { pub player_server: Option, pub kick_secs: u64, pub combine_views_stats_streams: bool, - pub landing_page: ViewType, + pub landing_page: Option, pub stream_info: Option, } diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 37fc1d478..a525361cc 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -44,8 +44,8 @@ use shared::{ foundation::{get_field_value, set_field_value, Filter, ValueAccessor, ValueProvider}, model::{ ClusterSource, CounterModifier, FieldGetAccessor, FieldSetAccessor, InputStats, InputType, ItemField, - PlaylistGroup, PlaylistItem, PlaylistItemType, PlaylistStats, ProcessingOrder, SourceStats, StreamProperties, - TargetStats, UUIDType, XtreamCluster, + PlaylistGroup, PlaylistItem, PlaylistItemType, PlaylistUpdateProgressEvent, PlaylistStats, ProcessingOrder, + SourceStats, StreamProperties, TargetStats, UUIDType, XtreamCluster, }, utils::{ create_alias_uuid, default_as_default, default_probe_delay_secs, default_probe_live_interval, interner_gc, @@ -888,7 +888,10 @@ fn create_broadcast_callback(event_manager: Option<&Arc>) -> StepM if let Some(event_mgr) = event_manager { let events = event_mgr.clone(); Box::new(move |context: &str, msg: &str| { - events.send_event(EventMessage::PlaylistUpdateProgress(context.to_owned(), msg.to_owned())); + events.send_event(EventMessage::PlaylistUpdateProgress(PlaylistUpdateProgressEvent { + target: context.to_owned(), + message: msg.to_owned(), + })); }) } else { Box::new(move |_context: &str, _msg: &str| { /* noop */ }) @@ -1664,10 +1667,10 @@ pub async fn exec_processing( let update_finished_message = format!("🌷 Update process finished! Took {elapsed} secs."); if let Some(events) = event_manager.as_deref() { - events.send_event(EventMessage::PlaylistUpdateProgress( - "Playlist Update".to_string(), - update_finished_message.clone(), - )); + events.send_event(EventMessage::PlaylistUpdateProgress(PlaylistUpdateProgressEvent { + target: "Playlist Update".to_string(), + message: update_finished_message.clone(), + })); } log_memory_snapshot("exec_processing before_interner_gc"); debug!("StringInterner GC removed {} strings", interner_gc()); diff --git a/backend/src/utils/m3u_catchup.rs b/backend/src/utils/m3u_catchup.rs index 5bc2db6bc..c766443c7 100644 --- a/backend/src/utils/m3u_catchup.rs +++ b/backend/src/utils/m3u_catchup.rs @@ -200,10 +200,14 @@ fn extract_suffix_from_filename(file_name: &str) -> &str { .unwrap_or_default() } +fn is_append_like_query_source(mode: &str, source: &str) -> bool { + mode_alias(mode) == "append" || source.starts_with('?') +} + fn derived_template_for_mode<'a>(source_url: &'a str, catchup: &'a CatchupProperties) -> Option> { let mode = catchup.mode.as_deref().unwrap_or_default(); if let Some(source) = catchup.source.as_deref().filter(|source| !source.is_empty()) { - return Some(if mode_alias(mode) == "append" { + return Some(if is_append_like_query_source(mode, source) { append_query_template(source_url, source).map(Cow::Owned)? } else { Cow::Borrowed(source) @@ -290,9 +294,10 @@ pub fn build_m3u_catchup_rewrite( let segments = parse_template(template.as_ref()); let placeholders = collect_placeholders(&segments); let append_mode = catchup - .mode + .source .as_deref() - .is_some_and(|mode| mode_alias(mode) == "append" && catchup.source.as_ref().is_some_and(|source| !source.is_empty())); + .filter(|source| !source.is_empty()) + .is_some_and(|source| is_append_like_query_source(catchup.mode.as_deref().unwrap_or_default(), source)); let source = build_local_source(base_url, "", &token, &placeholders, append_mode); let mode = if append_mode { "append" } else { "default" }; @@ -429,6 +434,50 @@ mod tests { ); } + #[test] + fn resolve_default_mode_query_template_keeps_base_stream_url() { + let resolved = resolve_m3u_catchup_url( + "http://provider.example/live/42.ts", + &CatchupProperties { + mode: Some("default".intern()), + source: Some("?playseek=${timestamp}&duration=${duration}".intern()), + ..CatchupProperties::default() + }, + Some("v0=1717200000&v1=120"), + ) + .unwrap() + .unwrap(); + + assert_eq!( + resolved.url, + "http://provider.example/live/42.ts?playseek=1717200000&duration=120" + ); + } + + #[test] + fn default_mode_query_template_rewrite_uses_append_route_marker() { + let rewrite = build_m3u_catchup_rewrite( + &[7u8; 16], + "http://proxy.example", + "alice", + 7, + 42, + "http://provider.example/live/42.ts", + &CatchupProperties { + mode: Some("default".intern()), + source: Some("?playseek=${timestamp}&duration=${duration}".intern()), + ..CatchupProperties::default() + }, + ) + .unwrap() + .unwrap(); + + assert_eq!(rewrite.mode.as_ref(), "append"); + assert!(rewrite.source.starts_with("?tuliprox-catchup=")); + assert!(rewrite.source.contains("&v0=${timestamp}")); + assert!(rewrite.source.contains("&v1=${duration}")); + } + #[test] fn resolve_shift_template_roundtrips_collectors() { let resolved = resolve_m3u_catchup_url( diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs index 9a5129edf..916de24df 100644 --- a/backend/src/utils/network/request.rs +++ b/backend/src/utils/network/request.rs @@ -1298,6 +1298,57 @@ async fn get_remote_content( Ok((content, response_url)) } +async fn get_remote_content_with_headers( + app_config: &Arc, + client: &reqwest::Client, + input: &InputSource, + headers: Option<&HeaderMap>, + url: &Url, +) -> Result<(String, String, HeaderMap), Error> { + let custom_headers = headers + .map(|h| h.iter().map(|(k, v)| (k.as_str().to_string(), v.as_bytes().to_vec())).collect::>()); + + let config = app_config.config.load(); + let default_user_agent = config.default_user_agent.clone(); + let disabled_headers = config.get_disabled_headers(); + drop(config); + + let merged = get_request_headers( + Some(&input.headers), + custom_headers.as_ref(), + disabled_headers.as_ref(), + default_user_agent.as_deref(), + ); + + let headers: HashMap = merged + .iter() + .map(|(k, v)| (k.as_str().to_string(), String::from_utf8_lossy(v.as_bytes()).to_string())) + .collect(); + + let response = send_with_retry_and_provider(app_config, url, input.get_provider(), false, |resolved_url| { + get_client_request( + client, + input.method, + Some(&headers), + resolved_url, + None, + None, + default_user_agent.as_deref(), + ) + }) + .await?; + + let response_url = response.url().to_string(); + let response_headers = response.headers().clone(); + let mut stream = build_decoded_stream_reader(response).await?; + let mut content = String::new(); + stream + .read_to_string(&mut content) + .await + .map_err(|e| string_to_io_error(format!("Failed to read content: {e}")))?; + Ok((content, response_url, response_headers)) +} + async fn get_remote_content_with_manual_redirects( app_config: &Arc, client: &reqwest::Client, @@ -1392,19 +1443,119 @@ async fn get_remote_content_with_manual_redirects( } } +async fn get_remote_content_with_manual_redirects_and_headers( + app_config: &Arc, + client: &reqwest::Client, + input: &InputSource, + headers: Option<&HeaderMap>, + url: &Url, + max_redirects: usize, +) -> Result<(String, String, HeaderMap), Error> { + let custom_headers = headers + .map(|h| h.iter().map(|(k, v)| (k.as_str().to_string(), v.as_bytes().to_vec())).collect::>()); + + let config = app_config.config.load(); + let default_user_agent = config.default_user_agent.clone(); + let disabled_headers = config.get_disabled_headers(); + drop(config); + + let merged = get_request_headers( + Some(&input.headers), + custom_headers.as_ref(), + disabled_headers.as_ref(), + default_user_agent.as_deref(), + ); + + let headers: HashMap = merged + .iter() + .map(|(k, v)| (k.as_str().to_string(), String::from_utf8_lossy(v.as_bytes()).to_string())) + .collect(); + + let mut current_url = url.clone(); + let mut current_headers = headers; + let mut remaining_redirects = max_redirects; + loop { + let response = + send_with_retry_and_provider(app_config, ¤t_url, input.get_provider(), true, |resolved_url| { + get_client_request( + client, + input.method, + Some(¤t_headers), + resolved_url, + None, + None, + default_user_agent.as_deref(), + ) + }) + .await?; + let response_base_url = response.url().clone(); + + if response.status().is_redirection() { + if remaining_redirects == 0 { + return Err(string_to_io_error(format!( + "Too many redirects while requesting {}", + sanitize_sensitive_info(url.as_str()) + ))); + } + + let Some(location) = response.headers().get(reqwest::header::LOCATION) else { + return Err(string_to_io_error(format!( + "Redirect response missing location header for {}", + sanitize_sensitive_info(current_url.as_str()) + ))); + }; + let Ok(location_str) = location.to_str() else { + return Err(string_to_io_error(format!( + "Redirect response contains invalid location header for {}", + sanitize_sensitive_info(current_url.as_str()) + ))); + }; + let next_url = + response_base_url.join(location_str).or_else(|_| Url::parse(location_str)).map_err(|_| { + string_to_io_error(format!( + "Redirect response contains invalid location URL for {}", + sanitize_sensitive_info(current_url.as_str()) + )) + })?; + + if !same_origin(&response_base_url, &next_url) { + strip_sensitive_headers_for_cross_origin_redirect(&mut current_headers); + } + current_url = next_url; + remaining_redirects = remaining_redirects.saturating_sub(1); + continue; + } + + let response_url = response.url().to_string(); + let response_headers = response.headers().clone(); + let mut stream = build_decoded_stream_reader(response).await?; + let mut content = String::new(); + stream + .read_to_string(&mut content) + .await + .map_err(|e| string_to_io_error(format!("Failed to read content: {e}")))?; + return Ok((content, response_url, response_headers)); + } +} + fn same_origin(lhs: &Url, rhs: &Url) -> bool { lhs.scheme().eq_ignore_ascii_case(rhs.scheme()) && lhs.host_str() == rhs.host_str() && lhs.port_or_known_default() == rhs.port_or_known_default() } +fn is_safe_cross_origin_redirect_header(key: &str) -> bool { + key.eq_ignore_ascii_case("accept") + || key.eq_ignore_ascii_case("accept-encoding") + || key.eq_ignore_ascii_case("accept-language") + || key.eq_ignore_ascii_case("user-agent") + || key.eq_ignore_ascii_case("range") + || key.eq_ignore_ascii_case("if-range") + || key.eq_ignore_ascii_case("icy-metadata") +} + fn strip_sensitive_headers_for_cross_origin_redirect(headers: &mut HashMap) { - headers.retain(|key, _| { - !key.eq_ignore_ascii_case("authorization") - && !key.eq_ignore_ascii_case("cookie") - && !key.eq_ignore_ascii_case("proxy-authorization") - && !key.eq_ignore_ascii_case("host") - }); + headers.retain(|key, _| is_safe_cross_origin_redirect_header(key)); } async fn download_epg_content_as_file( @@ -1486,6 +1637,43 @@ pub async fn download_text_content( result } +pub async fn download_text_content_with_headers( + app_config: &Arc, + client: &reqwest::Client, + input: &InputSource, + headers: Option<&HeaderMap>, + trace_log: bool, +) -> Result<(String, String, HeaderMap), Error> { + let start_time = tokio::time::Instant::now(); + let result = if let Ok(url) = input.url.parse::() { + let result = if url.scheme() == "file" { + match url.to_file_path() { + Ok(file_path) => get_local_file_content(&file_path).await.map(|content| (content, url.to_string(), HeaderMap::new())), + Err(()) => Err(string_to_io_error(format!("Unknown file {}", sanitize_sensitive_info(&input.url)))), + } + } else { + get_remote_content_with_headers(app_config, client, input, headers, &url).await + }; + result + } else { + Err(string_to_io_error(format!("Malformed URL {}", sanitize_sensitive_info(&input.url)))) + }; + + let level = if trace_log { log::Level::Trace } else { log::Level::Debug }; + if log_enabled!(level) { + if let Ok((_, response_url, _)) = result.as_ref() { + log::log!( + level, + "Request took: {} {}", + format_elapsed_time(start_time.elapsed().as_secs()), + sanitize_sensitive_info(response_url.as_str()) + ); + } + } + + result +} + pub async fn download_text_content_with_manual_redirects( app_config: &Arc, client: &reqwest::Client, @@ -1533,6 +1721,44 @@ pub async fn download_text_content_with_manual_redirects( result } +pub async fn download_text_content_with_manual_redirects_and_headers( + app_config: &Arc, + client: &reqwest::Client, + input: &InputSource, + headers: Option<&HeaderMap>, + trace_log: bool, + max_redirects: usize, +) -> Result<(String, String, HeaderMap), Error> { + let start_time = tokio::time::Instant::now(); + let result = if let Ok(url) = input.url.parse::() { + let result = if url.scheme() == "file" { + match url.to_file_path() { + Ok(file_path) => get_local_file_content(&file_path).await.map(|content| (content, url.to_string(), HeaderMap::new())), + Err(()) => Err(string_to_io_error(format!("Unknown file {}", sanitize_sensitive_info(&input.url)))), + } + } else { + get_remote_content_with_manual_redirects_and_headers(app_config, client, input, headers, &url, max_redirects).await + }; + result + } else { + Err(string_to_io_error(format!("Malformed URL {}", sanitize_sensitive_info(&input.url)))) + }; + + let level = if trace_log { log::Level::Trace } else { log::Level::Debug }; + if log_enabled!(level) { + if let Ok((_, response_url, _)) = result.as_ref() { + log::log!( + level, + "Request took: {} {}", + format_elapsed_time(start_time.elapsed().as_secs()), + sanitize_sensitive_info(response_url.as_str()) + ); + } + } + + result +} + pub async fn download_text_content_as_stream( app_config: &Arc, client: &reqwest::Client, @@ -1755,6 +1981,7 @@ pub fn should_trigger_failover(status: StatusCode) -> bool { #[cfg(test)] mod tests { use super::{ + is_safe_cross_origin_redirect_header, next_provider_url_index, preview_request_diagnostics_for_logging, preview_request_target_for_logging, resolve_attempt_target, same_origin, send_with_retry_and_provider, send_with_retry_and_provider_policy, should_try_next_ip_on_connect_error, strip_sensitive_headers_for_cross_origin_redirect, @@ -1916,7 +2143,9 @@ mod tests { headers.insert("Cookie".to_string(), "sid=123".to_string()); headers.insert("Proxy-Authorization".to_string(), "Basic abc".to_string()); headers.insert("Host".to_string(), "old.host".to_string()); - headers.insert("X-Test".to_string(), "ok".to_string()); + headers.insert("X-API-Key".to_string(), "secret".to_string()); + headers.insert("Accept".to_string(), "application/x-mpegurl".to_string()); + headers.insert("User-Agent".to_string(), "mpv".to_string()); strip_sensitive_headers_for_cross_origin_redirect(&mut headers); @@ -1924,7 +2153,20 @@ mod tests { assert!(!headers.contains_key("Cookie")); assert!(!headers.contains_key("Proxy-Authorization")); assert!(!headers.contains_key("Host")); - assert_eq!(headers.get("X-Test").map(String::as_str), Some("ok")); + assert!(!headers.contains_key("X-API-Key")); + assert_eq!(headers.get("Accept").map(String::as_str), Some("application/x-mpegurl")); + assert_eq!(headers.get("User-Agent").map(String::as_str), Some("mpv")); + } + + #[test] + fn test_cross_origin_redirect_header_allowlist_is_minimal() { + assert!(is_safe_cross_origin_redirect_header("accept")); + assert!(is_safe_cross_origin_redirect_header("user-agent")); + assert!(is_safe_cross_origin_redirect_header("icy-metadata")); + assert!(!is_safe_cross_origin_redirect_header("authorization")); + assert!(!is_safe_cross_origin_redirect_header("cookie")); + assert!(!is_safe_cross_origin_redirect_header("x-api-key")); + assert!(!is_safe_cross_origin_redirect_header("x-auth-token")); } #[test] diff --git a/frontend/scss/app/components/_card.scss b/frontend/scss/app/components/_card.scss index d3520cb46..337040d0a 100644 --- a/frontend/scss/app/components/_card.scss +++ b/frontend/scss/app/components/_card.scss @@ -18,7 +18,7 @@ @media (hover: hover) { &:hover { - box-shadow: 0 8px 24px var(--theme-palette-neutral-surface-660, rgba(0, 0, 0, 0.25)); + box-shadow: 0 8px 8px var(--theme-palette-neutral-surface-660, rgba(0, 0, 0, 0.2)); background-color: var(--card-hover-background-color); } } diff --git a/frontend/scss/app/components/dashboard/_dashboard_view.scss b/frontend/scss/app/components/dashboard/_dashboard_view.scss index f903f4351..50dae6963 100644 --- a/frontend/scss/app/components/dashboard/_dashboard_view.scss +++ b/frontend/scss/app/components/dashboard/_dashboard_view.scss @@ -26,6 +26,8 @@ display: grid; grid-template-columns: var(--action-card-width) var(--action-card-width); gap: var(--gap-large); + padding-right: var(--padding-default); + padding-bottom: var(--padding-default); @media (max-width: size.$dashboard-mobile-breakpoint) { grid-template-columns: var(--action-card-width); diff --git a/frontend/scss/app/components/dashboard/_stats_view.scss b/frontend/scss/app/components/dashboard/_stats_view.scss index 0dd37b3a5..add44e1a7 100644 --- a/frontend/scss/app/components/dashboard/_stats_view.scss +++ b/frontend/scss/app/components/dashboard/_stats_view.scss @@ -18,6 +18,8 @@ display: flex; flex-flow: column; gap: var(--gap-default); + padding-right: var(--padding-default); + padding-bottom: var(--padding-default); overflow: auto; pointer-events: auto; touch-action: pan-x pan-y; diff --git a/frontend/scss/app/components/dashboard/_streams_view.scss b/frontend/scss/app/components/dashboard/_streams_view.scss index c62d977df..c4adc82f0 100644 --- a/frontend/scss/app/components/dashboard/_streams_view.scss +++ b/frontend/scss/app/components/dashboard/_streams_view.scss @@ -31,6 +31,8 @@ .tp__stats { &__body { + padding-right: var(--padding-default); + padding-bottom: var(--padding-default); &-group-mobile { display: none; width: 100%; diff --git a/frontend/scss/app/components/playlist/_playlist_update_view.scss b/frontend/scss/app/components/playlist/_playlist_update_view.scss index f7025d78c..597c15b7d 100644 --- a/frontend/scss/app/components/playlist/_playlist_update_view.scss +++ b/frontend/scss/app/components/playlist/_playlist_update_view.scss @@ -27,4 +27,28 @@ touch-action: pan-y; overscroll-behavior: contain; } + + &__log { + width: 100%; + box-sizing: border-box; + max-height: 320px; + min-height: 160px; + overflow-y: auto; + background-color: var(--card-background-color); + color: var(--modest-text-color); + font-family: ui-monospace, SFMono-Regular, "SF Mono", Menlo, Consolas, "Liberation Mono", monospace; + font-size: 0.85rem; + line-height: 1.35; + padding: var(--gap-default); + border-radius: var(--border-radius); + border: 1px solid var(--border-color); + -webkit-overflow-scrolling: touch; + touch-action: pan-y; + overscroll-behavior: contain; + } + + &__log-line { + white-space: pre-wrap; + overflow-wrap: anywhere; + } } \ No newline at end of file diff --git a/frontend/src/app/components/authentication.rs b/frontend/src/app/components/authentication.rs index 2f500d17b..9fcb2c57a 100644 --- a/frontend/src/app/components/authentication.rs +++ b/frontend/src/app/components/authentication.rs @@ -16,6 +16,12 @@ fn should_connect_websocket(success: bool, setup_mode: bool, can_read_system: bo const SESSION_EXPIRY_SKEW_SECS: i64 = 30; +fn compute_session_expiry_delay_ms(exp: i64, now_secs: i64, skew_secs: i64) -> u32 { + let remaining_secs = exp.saturating_sub(skew_secs).saturating_sub(now_secs).max(0); + let remaining_ms = remaining_secs.saturating_mul(1000); + remaining_ms.min(i64::from(i32::MAX)) as u32 +} + #[derive(Properties, Clone, PartialEq)] pub struct AuthenticationProps { pub children: Children, @@ -87,8 +93,7 @@ pub fn Authentication(props: &AuthenticationProps) -> Html { if *authenticated { if let Some(exp) = *token_exp { let now_secs = js_sys::Date::now() / 1000.0; - let remaining_secs = exp - SESSION_EXPIRY_SKEW_SECS - now_secs as i64; - let delay_ms = u32::try_from(remaining_secs.max(0) * 1000).unwrap_or(u32::MAX); + let delay_ms = compute_session_expiry_delay_ms(exp, now_secs as i64, SESSION_EXPIRY_SKEW_SECS); timeout = Some(Timeout::new(delay_ms, move || { services_ctx.auth.logout(); services_ctx.toastr.warning(translate.t("MESSAGES.SESSION.EXPIRED")); @@ -114,7 +119,7 @@ pub fn Authentication(props: &AuthenticationProps) -> Html { #[cfg(test)] mod tests { - use super::should_connect_websocket; + use super::{compute_session_expiry_delay_ms, should_connect_websocket, SESSION_EXPIRY_SKEW_SECS}; #[test] fn websocket_connects_only_for_authenticated_non_setup_users_with_system_read() { @@ -123,4 +128,13 @@ mod tests { assert!(!should_connect_websocket(true, true, true)); assert!(!should_connect_websocket(true, false, false)); } + + #[test] + fn session_expiry_delay_clamps_large_token_ttls_to_browser_safe_max() { + let now_secs = 1_700_000_000; + let hundred_years_secs = 100_i64 * 365 * 24 * 60 * 60; + let exp = now_secs + hundred_years_secs; + + assert_eq!(compute_session_expiry_delay_ms(exp, now_secs, SESSION_EXPIRY_SKEW_SECS), i32::MAX as u32); + } } diff --git a/frontend/src/app/components/config/config_update.rs b/frontend/src/app/components/config/config_update.rs index 320416c23..cee2ca98c 100644 --- a/frontend/src/app/components/config/config_update.rs +++ b/frontend/src/app/components/config/config_update.rs @@ -357,7 +357,7 @@ mod tests { let mut config = ConfigDto { web_ui: Some(WebUiConfigDto { enabled: true, - landing_page: ViewType::Streams, // existing is non-default + landing_page: Some(ViewType::Streams), // existing is non-default ..Default::default() }), ..ConfigDto::default() @@ -369,7 +369,7 @@ mod tests { true, // modified=true: explicit edit WebUiConfigDto { enabled: false, - landing_page: ViewType::Dashboard, // reset to default + landing_page: None, // reset to last-page default ..Default::default() }, )], @@ -377,7 +377,7 @@ mod tests { let web_ui = config.web_ui.expect("webui config should be present"); assert!(!web_ui.enabled); - assert_eq!(web_ui.landing_page, ViewType::Dashboard, "landing_page reset to default must be applied"); + assert_eq!(web_ui.landing_page, None, "landing_page reset to default must be applied"); } #[test] @@ -418,7 +418,7 @@ mod tests { let mut config = ConfigDto { web_ui: Some(WebUiConfigDto { enabled: true, - landing_page: ViewType::Streams, // existing is non-default + landing_page: Some(ViewType::Streams), // existing is non-default ..Default::default() }), ..ConfigDto::default() @@ -430,7 +430,7 @@ mod tests { false, // modified=false: toggle-only, landing_page unchanged in form WebUiConfigDto { enabled: false, - landing_page: ViewType::Streams, // form syncs from existing on open + landing_page: Some(ViewType::Streams), // form syncs from existing on open ..Default::default() }, )], @@ -438,7 +438,7 @@ mod tests { let web_ui = config.web_ui.expect("webui config should be present"); assert!(!web_ui.enabled); - assert_eq!(web_ui.landing_page, ViewType::Streams, "landing_page should be preserved on toggle-only"); + assert_eq!(web_ui.landing_page, Some(ViewType::Streams), "landing_page should be preserved on toggle-only"); } #[test] @@ -448,7 +448,7 @@ mod tests { let mut config = ConfigDto { web_ui: Some(WebUiConfigDto { enabled: true, - landing_page: ViewType::Streams, + landing_page: Some(ViewType::Streams), auth: Some(WebAuthConfigDto { issuer: "test".to_string(), ..Default::default() }), ..Default::default() }), @@ -461,7 +461,7 @@ mod tests { true, // explicit edit: nested values must be applied WebUiConfigDto { enabled: false, - landing_page: ViewType::Streams, // form syncs from existing + landing_page: Some(ViewType::Streams), // form syncs from existing auth: Some(WebAuthConfigDto { issuer: "other".to_string(), ..Default::default() }), ..Default::default() }, @@ -471,7 +471,7 @@ mod tests { // Selective update: scalars and landing_page applied, auth preserved (nested change). let web_ui = config.web_ui.expect("webui config should be present"); assert!(!web_ui.enabled); - assert_eq!(web_ui.landing_page, ViewType::Streams); + assert_eq!(web_ui.landing_page, Some(ViewType::Streams)); assert_eq!(web_ui.auth.as_ref().map(|a| a.issuer.as_str()), Some("other")); } diff --git a/frontend/src/app/components/config/webui_config_view.rs b/frontend/src/app/components/config/webui_config_view.rs index 91c568c94..b67ed5ee9 100644 --- a/frontend/src/app/components/config/webui_config_view.rs +++ b/frontend/src/app/components/config/webui_config_view.rs @@ -38,6 +38,7 @@ const LABEL_CONTENT_SECURITY_POLICY_CUSTOM_ATTRIBUTES: &str = "LABEL.CUSTOM_ATTR const LABEL_PATH: &str = "LABEL.PATH"; const LABEL_COMBINE_VIEWS_STATS_STREAMS: &str = "LABEL.COMBINE_VIEWS_STATS_STREAMS"; const LABEL_LANDING_PAGE: &str = "LABEL.LANDING_PAGE"; +const LANDING_PAGE_LAST_PAGE: &str = "last_page"; const LABEL_STREAM_INFO: &str = "LABEL.STREAM_INFO"; const LABEL_HIDE_GROUP: &str = "LABEL.HIDE_GROUP"; const LABEL_HIDE_IP: &str = "LABEL.HIDE_IP"; @@ -61,7 +62,7 @@ generate_form_reducer!( PlayerServer => player_server: Option, KickSecs => kick_secs: u64, CombineViewsStatsStreams => combine_views_stats_streams: bool, - LandingPage => landing_page: shared::model::view_type::ViewType, + LandingPage => landing_page: Option, } ); @@ -121,15 +122,17 @@ pub fn WebUiConfigView() -> Html { use_default_form_reducer!(StreamInfoConfigFormState { form: StreamInfoConfigDto::default() }); let view_types = use_memo(webui_state.data().landing_page, |landing_page| { - ViewType::iter() - .collect::>() - .iter() - .map(|view_type| DropDownOption { - id: view_type.to_string(), - label: html! { translate.t(&format!("LABEL.VIEW_TYPE_{}", view_type.to_string().to_uppercase()))}, - selected: landing_page == view_type, - }) - .collect::>() + let mut options = vec![DropDownOption { + id: LANDING_PAGE_LAST_PAGE.to_string(), + label: html! { translate.t("LABEL.LAST_PAGE") }, + selected: landing_page.is_none(), + }]; + options.extend(ViewType::iter().map(|view_type| DropDownOption { + id: view_type.to_string(), + label: html! { translate.t(&format!("LABEL.VIEW_TYPE_{}", view_type.to_string().to_uppercase()))}, + selected: *landing_page == Some(view_type), + })); + options }); // Notify parent when form changes @@ -209,7 +212,13 @@ pub fn WebUiConfigView() -> Html { <> { config_field_bool!(webui_state.form, translate.t(LABEL_ENABLED), enabled) } - { config_field_custom!(translate.t(LABEL_LANDING_PAGE), translate.t(&format!("LABEL.VIEW_TYPE_{}", webui_state.form.landing_page.to_string().to_uppercase()))) } + { config_field_custom!(translate.t(LABEL_LANDING_PAGE), { + if let Some(landing_page) = webui_state.form.landing_page { + translate.t(&format!("LABEL.VIEW_TYPE_{}", landing_page.to_string().to_uppercase())) + } else { + translate.t("LABEL.LAST_PAGE") + } + }) } { config_field_bool!(webui_state.form, translate.t(LABEL_USER_UI_ENABLED), user_ui_enabled) } { config_field_bool!(webui_state.form, translate.t(LABEL_COMBINE_VIEWS_STATS_STREAMS), combine_views_stats_streams) } { config_field_optional!(webui_state.form, translate.t(LABEL_PATH), path) } @@ -271,12 +280,24 @@ pub fn WebUiConfigView() -> Html { html! { {services.config().getUiConfig().languages.map(l => )} @@ -320,13 +357,16 @@ pub fn Home() -> Html { { use_effect_with( - (*view_visible, configured_fallback, home_access), - move |(current, configured_fallback, access)| { + (*view_visible, configured_landing_page, stored_last_home_view, home_access), + move |(current, configured_landing_page, stored_last_home_view, access)| { if !setup_mode { - let resolved = resolve_home_view(*current, *configured_fallback, *access); - let want = resolved.as_str(); - if get_location_hash().as_deref() != Some(want) { - set_location_hash(want); + if let Some(resolved) = + resolve_visible_home_view(*current, *configured_landing_page, *stored_last_home_view, *access) + { + let want = resolved.as_str(); + if get_location_hash().as_deref() != Some(want) { + set_location_hash(want); + } } } || () @@ -336,35 +376,59 @@ pub fn Home() -> Html { { let view_vis = view_visible.clone(); - use_effect_with((configured_fallback, home_access), move |(configured_fallback, access)| { - let configured_fallback = *configured_fallback; - let access = *access; - let closure = Closure::::wrap(Box::new(move |_event: Event| { - let requested = get_location_hash().and_then(|hash| ViewType::from_str(&hash).ok()); - let resolved = resolve_home_view(requested, configured_fallback, access); - view_vis.set(Some(resolved)); - })); - let win = window(); - if let Some(win) = win.as_ref() { - if win.add_event_listener_with_callback("hashchange", closure.as_ref().unchecked_ref()).is_err() { - error!("failed to register hashchange listener"); + use_effect_with( + (configured_landing_page, stored_last_home_view, home_access), + move |(configured_landing_page, stored_last_home_view, access)| { + let configured_landing_page = *configured_landing_page; + let stored_last_home_view = *stored_last_home_view; + let access = *access; + let closure = Closure::::wrap(Box::new(move |_event: Event| { + let requested = get_location_hash().and_then(|hash| ViewType::from_str(&hash).ok()); + let resolved = + resolve_visible_home_view(requested, configured_landing_page, stored_last_home_view, access); + view_vis.set(resolved); + })); + let win = window(); + if let Some(win) = win.as_ref() { + if win.add_event_listener_with_callback("hashchange", closure.as_ref().unchecked_ref()).is_err() { + error!("failed to register hashchange listener"); + } } - } - move || { - if let Some(win) = window() { - let _ = win.remove_event_listener_with_callback("hashchange", closure.as_ref().unchecked_ref()); + move || { + if let Some(win) = window() { + let _ = win.remove_event_listener_with_callback("hashchange", closure.as_ref().unchecked_ref()); + } } - } - }); + }, + ); } { let view_vis = view_visible.clone(); - use_effect_with((configured_fallback, home_access), move |(configured_fallback, access)| { - let current = *view_vis; - let resolved = resolve_home_view(current, *configured_fallback, *access); - if current != Some(resolved) { - view_vis.set(Some(resolved)); + use_effect_with( + (configured_landing_page, stored_last_home_view, home_access), + move |(configured_landing_page, stored_last_home_view, access)| { + let current = *view_vis; + let resolved = + resolve_visible_home_view(current, *configured_landing_page, *stored_last_home_view, *access); + if current != resolved { + view_vis.set(resolved); + } + || () + }, + ); + } + + { + use_effect_with((configured_landing_page, *view_visible), move |(configured_landing_page, view_visible)| { + match configured_landing_page { + Some(None) => { + if let Some(view) = view_visible { + set_session_storage_item(LAST_HOME_VIEW_STORAGE_KEY, view.as_str()); + } + } + Some(Some(_)) => remove_session_storage_item(LAST_HOME_VIEW_STORAGE_KEY), + None => {} } || () }); @@ -405,7 +469,9 @@ pub fn Home() -> Html { }; } - let resolved_view = resolve_home_view(*view_visible, configured_fallback, home_access); + let resolved_view = + resolve_visible_home_view(*view_visible, configured_landing_page, stored_last_home_view, home_access) + .unwrap_or(ViewType::Dashboard); let view_page = resolved_view.intern(); html! { context={config_context}> @@ -571,7 +637,7 @@ pub fn Home() -> Html { #[cfg(test)] mod tests { - use super::{is_allowed_home_view, resolve_home_view}; + use super::{is_allowed_home_view, resolve_home_view, resolve_visible_home_view}; use crate::model::ViewType; fn full_access() -> super::HomeViewAccess { @@ -597,6 +663,27 @@ mod tests { assert_eq!(resolved, ViewType::Dashboard); } + #[test] + fn resolve_visible_home_view_waits_for_configured_landing_page_without_hash() { + let resolved = resolve_visible_home_view(None, None, None, full_access()); + + assert_eq!(resolved, None); + } + + #[test] + fn resolve_visible_home_view_uses_configured_landing_page_when_hash_missing() { + let resolved = resolve_visible_home_view(None, Some(Some(ViewType::Streams)), None, full_access()); + + assert_eq!(resolved, Some(ViewType::Streams)); + } + + #[test] + fn resolve_visible_home_view_uses_last_page_when_landing_page_is_none() { + let resolved = resolve_visible_home_view(None, Some(None), Some(ViewType::Streams), full_access()); + + assert_eq!(resolved, Some(ViewType::Streams)); + } + #[test] fn resolve_home_view_rejects_disallowed_hash() { let resolved = resolve_home_view( diff --git a/frontend/src/app/components/playlist/playlist_update_view.rs b/frontend/src/app/components/playlist/playlist_update_view.rs index eefbec9f0..b39e669f3 100644 --- a/frontend/src/app/components/playlist/playlist_update_view.rs +++ b/frontend/src/app/components/playlist/playlist_update_view.rs @@ -6,9 +6,12 @@ use crate::{ hooks::use_service_context, html_if, i18n::use_translation, + model::EventMessage, }; -use shared::model::{permission::Permission, ConfigTargetDto}; +use shared::model::{permission::Permission, ConfigTargetDto, LibraryScanSummary}; use std::rc::Rc; +use wasm_bindgen::JsCast; +use web_sys::HtmlElement; use yew::{platform::spawn_local, prelude::*}; use yew_hooks::use_list; @@ -16,6 +19,50 @@ const LABEL_UPDATE_LOCAL_LIBRARY: &str = "LABEL.UPDATE_LOCAL_LIBRARY"; const LABEL_FORCE: &str = "LABEL.FORCE"; const ACTION_UPDATE_LIBRARY: &str = "update_library"; const ACTION_UPDATE_LIBRARY_FORCE: &str = "update_library_force"; +const MAX_LOG_LINES: usize = 500; + +fn format_hms_now() -> String { + let now = js_sys::Date::new_0(); + let pad = |n: u32| if n < 10 { format!("0{n}") } else { n.to_string() }; + format!("{}:{}:{}", pad(now.get_hours()), pad(now.get_minutes()), pad(now.get_seconds())) +} + +fn format_library_log_line(summary: &LibraryScanSummary) -> String { summary.message.clone() } + +fn append_log_line_entries(current: &[AttrValue], line: String) -> Vec { + let mut updated = current.to_vec(); + updated.push(AttrValue::from(line)); + if updated.len() > MAX_LOG_LINES { + let drop_count = updated.len() - MAX_LOG_LINES; + updated.drain(0..drop_count); + } + updated +} + +#[derive(Clone, PartialEq)] +struct LogLinesState { + lines: Vec, +} + +enum LogLinesAction { + Clear, + Append(String), +} + +impl Reducible for LogLinesState { + type Action = LogLinesAction; + + fn reduce(self: Rc, action: Self::Action) -> Rc { + match action { + LogLinesAction::Clear => Rc::new(Self { lines: Vec::new() }), + LogLinesAction::Append(line) => Rc::new(Self { lines: append_log_line_entries(&self.lines, line) }), + } + } +} + +fn push_log_line(lines: &UseReducerHandle, line: String) { + lines.dispatch(LogLinesAction::Append(line)); +} #[component] pub fn PlaylistUpdateView() -> Html { @@ -27,6 +74,46 @@ pub fn PlaylistUpdateView() -> Html { let can_write_library = services_ctx.auth.has_permission(Permission::LibraryWrite); let breadcrumbs = use_state(|| Rc::new(vec![translate.t("LABEL.PLAYLISTS"), translate.t("LABEL.UPDATE")])); let selected_targets = use_list::>(vec![]); + let log_lines = use_reducer(|| LogLinesState { lines: Vec::new() }); + let log_container_ref = use_node_ref(); + + // Subscribe to playlist and library progress events. The subscription lives for + // the component lifetime; cleanup unsubscribes on unmount to avoid leaks when + // the user navigates away mid-update. + { + let services = services_ctx.clone(); + let log_lines = log_lines.clone(); + use_effect_with((), move |_| { + let services_for_cleanup = services.clone(); + let sub_id = services.event.subscribe(move |msg| match msg { + EventMessage::PlaylistUpdateProgress(progress) => { + push_log_line(&log_lines, format!("{} [playlist] {}", format_hms_now(), progress.message)); + } + EventMessage::LibraryScanProgress(progress) => { + push_log_line( + &log_lines, + format!("{} [library] {}", format_hms_now(), format_library_log_line(&progress.summary)), + ); + } + _ => {} + }); + move || { + services_for_cleanup.event.unsubscribe(sub_id); + } + }); + } + + // Auto-scroll the log container to the bottom whenever a new line is appended. + { + let log_container_ref = log_container_ref.clone(); + let log_snapshot = log_lines.lines.clone(); + use_effect_with(log_snapshot, move |_| { + if let Some(el) = log_container_ref.get().and_then(|n| n.dyn_into::().ok()) { + el.set_scroll_top(el.scroll_height()); + } + || () + }); + } let handle_all_select = { let selected_targets = selected_targets.clone(); @@ -51,10 +138,12 @@ pub fn PlaylistUpdateView() -> Html { let translate = translate.clone(); let services = services_ctx.clone(); let selected_targets = selected_targets.clone(); + let log_lines = log_lines.clone(); Callback::from(move |_| { if !can_write_playlist { return; } + log_lines.dispatch(LogLinesAction::Clear); let selected_targets = selected_targets.clone(); let services = services.clone(); let translate = translate.clone(); @@ -64,13 +153,10 @@ pub fn PlaylistUpdateView() -> Html { targets.iter().map(|t| t.name.clone()).collect::>() }; let update_target_names = target_names.iter().map(|t| t.as_str()).collect::>(); - match services.playlist.update_targets(&update_target_names).await { - true => { - services.toastr.success(translate.t("MESSAGES.PLAYLIST_UPDATE.SUCCESS")); - } - false => { - services.toastr.error(translate.t("MESSAGES.PLAYLIST_UPDATE.FAIL")); - } + if services.playlist.update_targets(&update_target_names).await { + services.toastr.success(translate.t("MESSAGES.PLAYLIST_UPDATE.SUCCESS")); + } else { + services.toastr.error(translate.t("MESSAGES.PLAYLIST_UPDATE.FAIL")); } }); }) @@ -79,12 +165,14 @@ pub fn PlaylistUpdateView() -> Html { let handle_update_content = { let services = services_ctx.clone(); let translate = translate.clone(); + let log_lines = log_lines.clone(); Callback::from(move |name: String| { if !can_write_library { return; } let services = services.clone(); let translate = translate.clone(); + let log_lines = log_lines.clone(); wasm_bindgen_futures::spawn_local(async move { let mode = match name.as_str() { ACTION_UPDATE_LIBRARY => 1, @@ -92,8 +180,9 @@ pub fn PlaylistUpdateView() -> Html { _ => 0, }; if mode > 0 { + log_lines.dispatch(LogLinesAction::Clear); match services.config.update_library(mode == 2).await { - Ok(_) => services.toastr.success(translate.t("MESSAGES.LIBRARY_UPDATE.SUCCESS")), + Ok(()) => services.toastr.success(translate.t("MESSAGES.LIBRARY_UPDATE.SUCCESS")), Err(_err) => services.toastr.error(translate.t("MESSAGES.LIBRARY_UPDATE.FAIL")), } } @@ -102,6 +191,15 @@ pub fn PlaylistUpdateView() -> Html { }; let library_enabled = config_ctx.config.as_ref().is_some_and(|c| c.config.is_library_enabled()); + let log_lines_render = { + let log_lines = log_lines.clone(); + use_memo(log_lines.lines.clone(), |lines| { + lines + .iter() + .map(|line| html! {
{ line }
}) + .collect::>() + }) + }; html! {
@@ -154,6 +252,38 @@ pub fn PlaylistUpdateView() -> Html { }
+
+ { for log_lines_render.iter().cloned() } +
} } + +#[cfg(test)] +mod tests { + use super::{append_log_line_entries, format_library_log_line, MAX_LOG_LINES}; + use shared::model::{LibraryScanSummary, LibraryScanSummaryStatus}; + use yew::AttrValue; + + #[test] + fn append_log_line_entries_keeps_latest_entries_when_log_is_capped() { + let current = (0..MAX_LOG_LINES).map(|idx| AttrValue::from(format!("line-{idx}"))).collect::>(); + + let updated = append_log_line_entries(¤t, "line-new".to_string()); + + assert_eq!(updated.len(), MAX_LOG_LINES); + assert_eq!(updated.first().map(AttrValue::as_str), Some("line-1")); + assert_eq!(updated.last().map(AttrValue::as_str), Some("line-new")); + } + + #[test] + fn format_library_log_line_uses_human_readable_message() { + let summary = LibraryScanSummary { + status: LibraryScanSummaryStatus::Success, + message: "Scan completed".to_string(), + result: None, + }; + + assert_eq!(format_library_log_line(&summary), "Scan completed"); + } +} diff --git a/frontend/src/app/components/playlist/target_table.rs b/frontend/src/app/components/playlist/target_table.rs index 492b196d2..964d31ee0 100644 --- a/frontend/src/app/components/playlist/target_table.rs +++ b/frontend/src/app/components/playlist/target_table.rs @@ -148,13 +148,10 @@ pub fn TargetTable(props: &TargetTableProps) -> Html { let dto_name = selected_dto.as_ref().map_or_else(String::new, |d| d.name.to_string()); spawn_local(async move { let targets = vec![dto_name.as_str()]; - match services_ctx.playlist.update_targets(&targets).await { - true => { - services_ctx.toastr.success(translate.t("MESSAGES.PLAYLIST_UPDATE.SUCCESS")); - } - false => { - services_ctx.toastr.error(translate.t("MESSAGES.PLAYLIST_UPDATE.FAIL")); - } + if services_ctx.playlist.update_targets(&targets).await { + services_ctx.toastr.success(translate.t("MESSAGES.PLAYLIST_UPDATE.SUCCESS")); + } else { + services_ctx.toastr.error(translate.t("MESSAGES.PLAYLIST_UPDATE.FAIL")); } }); } diff --git a/frontend/src/app/components/setup/setup_helpers.rs b/frontend/src/app/components/setup/setup_helpers.rs index b4581d9af..e741ce10b 100644 --- a/frontend/src/app/components/setup/setup_helpers.rs +++ b/frontend/src/app/components/setup/setup_helpers.rs @@ -692,7 +692,7 @@ mod tests { let mut app_config = AppConfigDto::default(); app_config.config.web_ui = Some(WebUiConfigDto { enabled: true, - landing_page: ViewType::Streams, // existing is non-default + landing_page: Some(ViewType::Streams), // existing is non-default ..Default::default() }); let config_ctx = ConfigContext { config: Some(Rc::new(app_config)), api_proxy: None }; @@ -701,7 +701,7 @@ mod tests { true, // modified=true: explicit edit WebUiConfigDto { enabled: false, - landing_page: ViewType::Dashboard, // reset to default + landing_page: None, // reset to last-page default ..Default::default() }, )); @@ -709,7 +709,7 @@ mod tests { let app_cfg = build_setup_app_config(&config_ctx, &form_state, SourcesConfigDto::default()); let web_ui = app_cfg.config.web_ui.expect("webui config should be present"); assert!(!web_ui.enabled); - assert_eq!(web_ui.landing_page, ViewType::Dashboard, "landing_page reset to default must be applied"); + assert_eq!(web_ui.landing_page, None, "landing_page reset to default must be applied"); } #[test] @@ -747,7 +747,7 @@ mod tests { let mut app_config = AppConfigDto::default(); app_config.config.web_ui = Some(WebUiConfigDto { enabled: true, - landing_page: ViewType::Streams, // existing is non-default + landing_page: Some(ViewType::Streams), // existing is non-default ..Default::default() }); let config_ctx = ConfigContext { config: Some(Rc::new(app_config)), api_proxy: None }; @@ -756,7 +756,7 @@ mod tests { true, // modified=true: form IS applied WebUiConfigDto { enabled: false, - landing_page: ViewType::Streams, // form syncs from existing on open + landing_page: Some(ViewType::Streams), // form syncs from existing on open ..Default::default() }, )); @@ -764,7 +764,7 @@ mod tests { let app_cfg = build_setup_app_config(&config_ctx, &form_state, SourcesConfigDto::default()); let web_ui = app_cfg.config.web_ui.expect("webui config should be present"); assert!(!web_ui.enabled); - assert_eq!(web_ui.landing_page, ViewType::Streams, "landing_page should be preserved on toggle-only"); + assert_eq!(web_ui.landing_page, Some(ViewType::Streams), "landing_page should be preserved on toggle-only"); } #[test] @@ -774,7 +774,7 @@ mod tests { let mut app_config = AppConfigDto::default(); app_config.config.web_ui = Some(WebUiConfigDto { enabled: true, - landing_page: ViewType::Streams, + landing_page: Some(ViewType::Streams), auth: Some(WebAuthConfigDto { issuer: "test".to_string(), ..Default::default() }), ..Default::default() }); @@ -784,7 +784,7 @@ mod tests { true, // explicit edit: nested values must be applied WebUiConfigDto { enabled: false, - landing_page: ViewType::Streams, // form syncs from existing + landing_page: Some(ViewType::Streams), // form syncs from existing auth: Some(WebAuthConfigDto { issuer: "other".to_string(), ..Default::default() }), ..Default::default() }, @@ -793,7 +793,7 @@ mod tests { let app_cfg = build_setup_app_config(&config_ctx, &form_state, SourcesConfigDto::default()); let web_ui = app_cfg.config.web_ui.expect("webui config should be present"); assert!(!web_ui.enabled); - assert_eq!(web_ui.landing_page, ViewType::Streams); + assert_eq!(web_ui.landing_page, Some(ViewType::Streams)); assert_eq!(web_ui.auth.as_ref().map(|a| a.issuer.as_str()), Some("other")); } diff --git a/frontend/src/hooks/use_server_status.rs b/frontend/src/hooks/use_server_status.rs index 7d6e4e1e4..90d0411f4 100644 --- a/frontend/src/hooks/use_server_status.rs +++ b/frontend/src/hooks/use_server_status.rs @@ -52,6 +52,14 @@ fn dedupe_streams_by_identity(streams: &mut Vec) { streams.retain(|stream| seen.insert(stream_identity_key(stream))); } +fn should_keep_preserved_stream_visible(stream: &StreamInfo) -> bool { + stream.session_token.is_some() && stream.channel.item_type.is_live_adaptive() +} + +fn should_keep_stream_when_connections_drop_to_zero(stream: &StreamInfo) -> bool { + stream.preserved && should_keep_preserved_stream_visible(stream) +} + fn is_running_download(download: &FileDownloadDto) -> bool { download.status == TransferStatusDto::Running } fn download_stream_uid(id: &str) -> u32 { @@ -177,7 +185,14 @@ fn apply_active_user_change(server_status: &mut StatusCheck, event: ActiveUserCo match event { ActiveUserConnectionChange::Updated(stream_info) => { if stream_info.preserved { - if let Some(pos) = find_stream_update_index(&server_status.active_user_streams, &stream_info) { + if should_keep_preserved_stream_visible(&stream_info) { + if let Some(pos) = find_stream_update_index(&server_status.active_user_streams, &stream_info) { + server_status.active_user_streams[pos] = stream_info; + } else { + server_status.active_user_streams.push(stream_info); + } + dedupe_streams_by_identity(&mut server_status.active_user_streams); + } else if let Some(pos) = find_stream_update_index(&server_status.active_user_streams, &stream_info) { server_status.active_user_streams.remove(pos); } return; @@ -199,7 +214,7 @@ fn apply_active_user_change(server_status: &mut StatusCheck, event: ActiveUserCo server_status.active_users = user_count; server_status.active_user_connections = connections; if connections == 0 { - server_status.active_user_streams.clear(); + server_status.active_user_streams.retain(should_keep_stream_when_connections_drop_to_zero); } } } @@ -444,7 +459,26 @@ mod tests { } #[test] - fn test_preserved_update_removes_active_stream_without_clearing_other_rows() { + fn test_connections_zero_keeps_preserved_adaptive_rows_for_ttl_cleanup() { + let mut preserved = test_stream(1, "127.0.0.1:1234", Some("tok-hls"), PlaylistItemType::LiveHls); + preserved.preserved = true; + let non_adaptive = test_stream(2, "127.0.0.1:5678", Some("tok-vod"), PlaylistItemType::Video); + let mut status = shared::model::StatusCheck { + active_users: 1, + active_user_connections: 1, + active_user_streams: vec![preserved.clone(), non_adaptive], + ..Default::default() + }; + + apply_active_user_change(&mut status, ActiveUserConnectionChange::Connections(1, 0)); + + assert_eq!(status.active_users, 1); + assert_eq!(status.active_user_connections, 0); + assert_eq!(status.active_user_streams, vec![preserved]); + } + + #[test] + fn test_preserved_adaptive_update_stays_visible_for_frontend_ttl_cleanup() { let mut preserved = test_stream(1, "127.0.0.1:1234", Some("tok-hls"), PlaylistItemType::LiveHls); preserved.preserved = true; let other = test_stream(2, "127.0.0.1:5678", Some("tok-other"), PlaylistItemType::LiveHls); @@ -460,6 +494,34 @@ mod tests { apply_active_user_change(&mut status, ActiveUserConnectionChange::Updated(preserved)); + assert_eq!(status.active_user_streams.len(), 2); + assert!(status + .active_user_streams + .iter() + .any(|stream| stream.addr == "127.0.0.1:1234".parse::().unwrap_or_else(|_| unreachable!()) + && stream.preserved)); + assert!(status.active_user_streams.iter().any(|stream| stream == &other)); + assert_eq!(status.active_users, 2); + assert_eq!(status.active_user_connections, 2); + } + + #[test] + fn test_preserved_non_adaptive_update_removes_active_stream_without_clearing_other_rows() { + let mut preserved = test_stream(1, "127.0.0.1:1234", Some("tok-vod"), PlaylistItemType::Video); + preserved.preserved = true; + let other = test_stream(2, "127.0.0.1:5678", Some("tok-other"), PlaylistItemType::LiveHls); + let mut status = shared::model::StatusCheck { + active_users: 2, + active_user_connections: 2, + active_user_streams: vec![ + test_stream(1, "127.0.0.1:1234", Some("tok-vod"), PlaylistItemType::Video), + other.clone(), + ], + ..Default::default() + }; + + apply_active_user_change(&mut status, ActiveUserConnectionChange::Updated(preserved)); + assert_eq!(status.active_user_streams, vec![other]); assert_eq!(status.active_users, 2); assert_eq!(status.active_user_connections, 2); diff --git a/frontend/src/model/event_message.rs b/frontend/src/model/event_message.rs index 1dd683b9e..1502c3208 100644 --- a/frontend/src/model/event_message.rs +++ b/frontend/src/model/event_message.rs @@ -1,7 +1,7 @@ use crate::model::BusyStatus; use shared::model::{ - ActiveUserConnectionChange, ConfigType, DownloadsDelta, DownloadsResponse, LibraryScanSummary, PlaylistUpdateState, - StatusCheck, StreamMeterEntry, SystemInfo, + ActiveUserConnectionChange, ConfigType, DownloadsDelta, DownloadsResponse, LibraryScanProgressEvent, + PlaylistUpdateProgressEvent, PlaylistUpdateState, StatusCheck, StreamMeterEntry, SystemInfo, }; use std::{rc::Rc, sync::Arc}; @@ -17,10 +17,10 @@ pub enum EventMessage { ConfigChange(ConfigType), Busy(BusyStatus), PlaylistUpdate(PlaylistUpdateState), - PlaylistUpdateProgress(String, String), + PlaylistUpdateProgress(PlaylistUpdateProgressEvent), WebSocketStatus(bool), SystemInfoUpdate(SystemInfo), - LibraryScanProgress(LibraryScanSummary), + LibraryScanProgress(LibraryScanProgressEvent), StreamMeterBatch(Vec), DownloadsUpdate(Rc), DownloadsDeltaUpdate(Rc), diff --git a/frontend/src/services/config_service.rs b/frontend/src/services/config_service.rs index b2582ec33..48de4bb77 100644 --- a/frontend/src/services/config_service.rs +++ b/frontend/src/services/config_service.rs @@ -12,7 +12,8 @@ use shared::{ foundation::{get_filter, prepare_templates, MapperScript}, model::{ permission::Permission, ApiProxyConfigDto, AppConfigDto, ConfigDto, ConfigInputDto, IpCheckDto, - LibraryScanRequest, SourcesConfigDto, TargetOutputDto, XtreamLoginInfo, XtreamLoginRequest, + LibraryScanRequest, OperationRunAccepted, SourcesConfigDto, TargetOutputDto, XtreamLoginInfo, + XtreamLoginRequest, }, utils::{ concat_path, concat_path_leading_slash, HEADER_CONFIG_API_PROXY_REVISION, HEADER_CONFIG_MAIN_REVISION, @@ -432,10 +433,10 @@ impl ConfigService { request_get::<()>(&self.geoip_path, None, None).await } - pub async fn update_library(&self, force_rescan: bool) -> Result, Error> { + pub async fn update_library(&self, force_rescan: bool) -> Result<(), Error> { let path = concat_path(&self.library_path, "scan"); let params = LibraryScanRequest { force_rescan }; - request_post::(&path, params, None, None).await + request_post::(&path, params, None, None).await.map(|_| ()) } pub async fn complete_setup(&self, payload: SetupCompleteRequestDto) -> Result<(), Error> { diff --git a/frontend/src/services/playlist_service.rs b/frontend/src/services/playlist_service.rs index 80e94861b..527185f14 100644 --- a/frontend/src/services/playlist_service.rs +++ b/frontend/src/services/playlist_service.rs @@ -4,9 +4,9 @@ use indexmap::IndexMap; use log::error; use shared::{ model::{ - EpgChannel, EpgTv, PlaylistEpgRequest, PlaylistRequest, PlaylistUrlResolveRequest, SeriesStreamProperties, - StreamEpgItemRequest, StreamEpgRequest, StreamEpgResponse, UiPlaylistCategories, UiPlaylistGroup, - UiPlaylistItem, XtreamCluster, XtreamSeriesInfoDoc, + EpgChannel, EpgTv, OperationRunAccepted, PlaylistEpgRequest, PlaylistRequest, PlaylistUrlResolveRequest, + SeriesStreamProperties, StreamEpgItemRequest, StreamEpgRequest, StreamEpgResponse, UiPlaylistCategories, + UiPlaylistGroup, UiPlaylistItem, XtreamCluster, XtreamSeriesInfoDoc, }, utils::concat_path_leading_slash, }; @@ -45,9 +45,7 @@ impl PlaylistService { } } pub async fn update_targets(&self, targets: &[&str]) -> bool { - request_post::<&[&str], ()>(&self.target_update_api_path, targets, None, None) - .await - .map_or_else(|_err| false, |_| true) + request_post::<&[&str], OperationRunAccepted>(&self.target_update_api_path, targets, None, None).await.is_ok() } pub async fn get_playlist_categories( diff --git a/frontend/src/services/requests.rs b/frontend/src/services/requests.rs index f11c5becc..c2d1f58fb 100644 --- a/frontend/src/services/requests.rs +++ b/frontend/src/services/requests.rs @@ -83,6 +83,15 @@ fn encoding_from_content_type(content_type: &str) -> Option { } } +fn should_decode_success_body(status: u16, content_type: &str) -> bool +where + T: 'static, +{ + status != 204 + && std::any::TypeId::of::() != std::any::TypeId::of::<()>() + && encoding_from_content_type(content_type).is_some() +} + async fn decode_response_body(response: Response, encoding: Encoding) -> Result where T: DeserializeOwned + 'static + std::fmt::Debug, @@ -201,10 +210,9 @@ where } } match status { - 200 | 205 | 206 => { + 200 | 201 | 202 | 205 | 206 => { let content_type = response.headers().get("content-type").unwrap_or_default().to_ascii_lowercase(); - if std::any::TypeId::of::() == std::any::TypeId::of::<()>() { - // `T = ()` valid + if !should_decode_success_body::(status, &content_type) { let _ = response.binary().await; return Ok(ResponseMeta { body: None, headers: response_headers }); } @@ -212,7 +220,7 @@ where let decoded = decode_response_body::(response, decode_encoding).await?; Ok(ResponseMeta { body: Some(decoded), headers: response_headers }) } - 201 | 202 | 204 => Ok(ResponseMeta { body: None, headers: response_headers }), + 204 => Ok(ResponseMeta { body: None, headers: response_headers }), 400 => { let message = extract_error_message(response).await; if message.trim().is_empty() { @@ -474,3 +482,24 @@ pub fn get_base_href() -> String { }) .clone() } + +#[cfg(test)] +mod tests { + use super::should_decode_success_body; + + #[test] + fn accepted_json_response_with_body_is_decoded() { + assert!(should_decode_success_body::(202, "application/json")); + } + + #[test] + fn accepted_response_without_decodable_content_type_is_not_decoded() { + assert!(!should_decode_success_body::(202, "")); + } + + #[test] + fn no_content_and_unit_responses_are_not_decoded() { + assert!(!should_decode_success_body::(204, "application/json")); + assert!(!should_decode_success_body::<()>(202, "application/json")); + } +} diff --git a/frontend/src/services/websocket_service.rs b/frontend/src/services/websocket_service.rs index 069134ae2..791b31bd7 100644 --- a/frontend/src/services/websocket_service.rs +++ b/frontend/src/services/websocket_service.rs @@ -364,14 +364,14 @@ fn handle_socket_protocol_msg( ProtocolMessage::PlaylistUpdateResponse(update_state) => { event_service.broadcast(EventMessage::PlaylistUpdate(update_state)); } - ProtocolMessage::PlaylistUpdateProgressResponse(target, msg) => { - event_service.broadcast(EventMessage::PlaylistUpdateProgress(target, msg)); + ProtocolMessage::PlaylistUpdateProgressResponse(progress) => { + event_service.broadcast(EventMessage::PlaylistUpdateProgress(progress)); } ProtocolMessage::SystemInfoResponse(system_info) => { event_service.broadcast(EventMessage::SystemInfoUpdate(system_info)); } - ProtocolMessage::LibraryScanProgressResponse(msg) => { - event_service.broadcast(EventMessage::LibraryScanProgress(msg)); + ProtocolMessage::LibraryScanProgressResponse(progress) => { + event_service.broadcast(EventMessage::LibraryScanProgress(progress)); } ProtocolMessage::DownloadsResponse(downloads) => { event_service.broadcast(EventMessage::DownloadsUpdate(Rc::new(downloads))); diff --git a/frontend/src/utils/storage.rs b/frontend/src/utils/storage.rs index 781955833..a032ae555 100644 --- a/frontend/src/utils/storage.rs +++ b/frontend/src/utils/storage.rs @@ -13,6 +13,26 @@ pub fn get_local_storage_item(key: &str) -> Option { window().and_then(|w| w.local_storage().ok()).flatten().and_then(|storage| storage.get_item(key).ok().flatten()) } +pub fn set_session_storage_item(key: &str, value: &str) { + if let Some(storage) = window().and_then(|w| w.session_storage().ok()).flatten() { + if let Err(err) = storage.set_item(key, value) { + error!("failed to write to sessionStorage: {err:?}"); + } + } +} + +pub fn get_session_storage_item(key: &str) -> Option { + window().and_then(|w| w.session_storage().ok()).flatten().and_then(|storage| storage.get_item(key).ok().flatten()) +} + +pub fn remove_session_storage_item(key: &str) { + if let Some(storage) = window().and_then(|w| w.session_storage().ok()).flatten() { + if let Err(err) = storage.remove_item(key) { + error!("failed to write to sessionStorage: {err:?}"); + } + } +} + pub fn remove_local_storage_item(key: &str) { if let Some(storage) = window().and_then(|w| w.local_storage().ok()).flatten() { if let Err(err) = storage.remove_item(key) { diff --git a/shared/src/model/config/web_ui.rs b/shared/src/model/config/web_ui.rs index 4730b4394..8360e3c15 100644 --- a/shared/src/model/config/web_ui.rs +++ b/shared/src/model/config/web_ui.rs @@ -167,8 +167,8 @@ pub struct WebUiConfigDto { pub kick_secs: u64, #[serde(default, skip_serializing_if = "is_false")] pub combine_views_stats_streams: bool, - #[serde(default, skip_serializing_if = "ViewType::is_default")] - pub landing_page: ViewType, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub landing_page: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub stream_info: Option, } @@ -184,7 +184,7 @@ impl Default for WebUiConfigDto { player_server: None, kick_secs: default_kick_secs(), combine_views_stats_streams: false, - landing_page: ViewType::default(), + landing_page: None, stream_info: None, } } @@ -196,7 +196,7 @@ impl WebUiConfigDto { self.enabled == empty.enabled && self.user_ui_enabled == empty.user_ui_enabled && !self.combine_views_stats_streams - && self.landing_page == ViewType::default() + && self.landing_page.is_none() && is_blank_or_default_web_ui_path(&self.path) && is_blank_optional_str(self.player_server.as_deref()) && self.kick_secs == default_kick_secs() @@ -302,4 +302,18 @@ mod tests { assert!(dto.stream_info.is_none()); assert!(dto.is_empty()); } + + #[test] + fn web_ui_config_dto_default_uses_last_page_landing_page() { + let dto = WebUiConfigDto::default(); + + assert_eq!(dto.landing_page, None); + } + + #[test] + fn web_ui_config_dto_is_not_empty_when_explicit_landing_page_is_set() { + let dto = WebUiConfigDto { landing_page: Some(ViewType::Streams), ..WebUiConfigDto::default() }; + + assert!(!dto.is_empty()); + } } diff --git a/shared/src/model/mod.rs b/shared/src/model/mod.rs index b4e430643..f5eb760fe 100644 --- a/shared/src/model/mod.rs +++ b/shared/src/model/mod.rs @@ -19,6 +19,7 @@ mod playlist_document; mod playlist_info_document; mod playlist_request; mod processing_order; +mod progress; mod regex_cache; mod search_request; mod short_epg; @@ -45,7 +46,7 @@ pub use self::{ active_user_connection_change::*, auth::*, cluster_flags::*, config::*, download::*, epg::*, epg_request::*, ip_check::*, item_field::*, library_request::*, mapping::*, media_properties::*, messaging::*, pagination::*, playlist::*, playlist_categories::*, playlist_info_document::*, playlist_request::*, processing_order::*, - regex_cache::*, search_request::*, short_epg::*, stats::*, status_check::*, stream_history::*, + progress::*, regex_cache::*, search_request::*, short_epg::*, stats::*, status_check::*, stream_history::*, stream_history_record::*, stream_info::*, stream_meter::*, stream_properties::*, strm_export_style::*, system_info::*, target_type::*, transfer::*, ui_playlist_item::*, user_command::*, uuidtype::*, web_socket::*, xtream::*, diff --git a/shared/src/model/progress.rs b/shared/src/model/progress.rs new file mode 100644 index 000000000..d07d1ce97 --- /dev/null +++ b/shared/src/model/progress.rs @@ -0,0 +1,18 @@ +use crate::model::LibraryScanSummary; +use serde::{Deserialize, Serialize}; + +/// Marker response for accepted long-running operations. Carries no payload: +/// progress is delivered over the event channel, not through this response. +#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq, Eq)] +pub struct OperationRunAccepted {} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +pub struct PlaylistUpdateProgressEvent { + pub target: String, + pub message: String, +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] +pub struct LibraryScanProgressEvent { + pub summary: LibraryScanSummary, +} diff --git a/shared/src/model/web_socket.rs b/shared/src/model/web_socket.rs index 439442d57..59435d3be 100644 --- a/shared/src/model/web_socket.rs +++ b/shared/src/model/web_socket.rs @@ -1,6 +1,7 @@ use crate::model::{ user_command::UserCommand, ActiveUserConnectionChange, ConfigType, DownloadsDelta, DownloadsResponse, - LibraryScanSummary, PermissionSet, PlaylistUpdateState, StatusCheck, StreamMeterEntry, SystemInfo, + LibraryScanProgressEvent, PermissionSet, PlaylistUpdateProgressEvent, PlaylistUpdateState, StatusCheck, + StreamMeterEntry, SystemInfo, }; use bytes::Bytes; use serde::{Deserialize, Serialize}; @@ -92,10 +93,10 @@ pub enum ProtocolMessage { ActiveProviderCountResponse(usize), ConfigChangeResponse(ConfigType), PlaylistUpdateResponse(PlaylistUpdateState), - PlaylistUpdateProgressResponse(String, String), + PlaylistUpdateProgressResponse(PlaylistUpdateProgressEvent), UserActionResponse(bool), SystemInfoResponse(SystemInfo), - LibraryScanProgressResponse(LibraryScanSummary), + LibraryScanProgressResponse(LibraryScanProgressEvent), StreamMeterBatchResponse(Vec), DownloadsResponse(DownloadsResponse), DownloadsDeltaResponse(DownloadsDelta), diff --git a/shared/src/utils/constants.rs b/shared/src/utils/constants.rs index ce6987a25..64b9aeed9 100644 --- a/shared/src/utils/constants.rs +++ b/shared/src/utils/constants.rs @@ -135,7 +135,7 @@ pub static CONSTANTS: LazyLock = LazyLock::new(|| { re_ipv4: Regex::new(r"\b((25[0-5]|2[0-4][0-9]|1[0-9]{2}|[1-9]?[0-9])\.){3}(25[0-5]|2[0-4][0-9]|1[0-9]{2}|[1-9]?[0-9])\b").unwrap(), re_ipv6: Regex::new(r"\b([0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}\b|\b::([0-9a-fA-F]{1,4}:){0,6}[0-9a-fA-F]{1,4}\b|\b([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}\b|\b::ffff:(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b").unwrap(), re_stream_url: Regex::new( - r"(?i)^(?P(?:https?|batch|provider)://)[^/]+/(?Plive|video|movie|series|m3u-stream|resource)/[^/]+/[^/]+/" + r"(?i)^(?P(?:https?|batch|provider)://)[^/]+/(?Plive|video|movie|series|m3u-stream|resource|timeshift|streaming|xtream|timeshift\.php)/[^/]+/[^/]+/" ) .unwrap(), re_url: Regex::new(r"(.*://).*?/(.*)").unwrap(), diff --git a/shared/src/utils/request.rs b/shared/src/utils/request.rs index 208b71500..05b92a19e 100644 --- a/shared/src/utils/request.rs +++ b/shared/src/utils/request.rs @@ -197,4 +197,39 @@ mod tests { set_sanitize_sensitive_info(previous); } + + #[test] + fn sanitize_sensitive_info_masks_path_credentials_for_all_xtream_contexts() { + // Regression: only `live|video|movie|series|m3u-stream|resource` were originally + // listed in `re_stream_url`, so URLs like `/timeshift/{user}/{pass}/...` leaked + // the username and password into log lines. The regex was extended to also + // cover `timeshift`, `streaming`, `xtream`, and `timeshift.php`. + let previous = super::is_sanitize_sensitive_info_enabled(); + set_sanitize_sensitive_info(true); + + let cases: &[(&str, &str)] = &[ + ( + "http://example/timeshift/myuser/mypass/3/2026-06-17:14-00/449.ts", + "http://***/timeshift/***/3/2026-06-17:14-00/449.ts", + ), + ("http://example/streaming/myuser/mypass/123.ts", "http://***/streaming/***/123.ts"), + ("http://example/xtream/myuser/mypass/123.ts", "http://***/xtream/***/123.ts"), + ("http://example/timeshift.php/myuser/mypass/123.ts", "http://***/timeshift.php/***/123.ts"), + ("https://provider.example/movie/myuser/mypass/456.ts", "https://***/movie/***/456.ts"), + ("https://provider.example/series/myuser/mypass/789.ts", "https://***/series/***/789.ts"), + ]; + + for (input, expected) in cases { + assert_eq!(sanitize_sensitive_info(input).as_ref(), *expected, "input: {input}"); + } + + // Query-string credentials are masked by `re_credentials`, not by the + // stream-URL rewriter — the host is still masked by `re_url`. + assert_eq!( + sanitize_sensitive_info("http://example/player_api.php?username=foo&password=bar"), + "http://***/player_api.php?username=***&password=***" + ); + + set_sanitize_sensitive_info(previous); + } }