From 806dd574d391cc4f212e839fe13f01e2bc5d0fdd Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Sat, 5 Sep 2026 17:14:49 +0200 Subject: [PATCH] User Agent (#861) New Features Added an option to append a stable, numeric stream index to upstream User-Agent headers. Stream indexes remain consistent throughout a playback session, including manifests, segments, reconnects, seeks, and provider failover. Added a configuration checkbox for enabling the feature. Documentation Documented the new configuration option, including session behavior and restart handling. Added English, Arabic, and Russian translations for the setting. Co-authored-by: euzu --- backend/app/src/api/api_utils/mod.rs | 62 ++++++++++++++++++- backend/app/src/api/api_utils/tests.rs | 7 +++ .../app/src/api/endpoints/hls_api/manifest.rs | 30 ++++++++- .../app/src/api/endpoints/hls_api/segment.rs | 20 +++++- .../app/src/api/endpoints/hls_api/tests.rs | 31 +++++++++- .../api/model/streams/active_client_stream.rs | 7 +++ .../model/streams/provider_stream_factory.rs | 4 ++ backend/app/src/api/panel_api/mod.rs | 2 + backend/core/src/model/config/input.rs | 11 ++++ backend/core/src/utils/network/request.rs | 54 +++++++++++++++- backend/hls/src/session.rs | 4 ++ .../session/src/active_user_manager/mod.rs | 36 +++++++++++ .../session/src/active_user_manager/tests.rs | 41 ++++++++++++ backend/session/src/stream.rs | 3 + docs/src/configuration/source.md | 20 ++++++ frontend/public/assets/i18n/ar.json | 2 + frontend/public/assets/i18n/en.json | 2 + frontend/public/assets/i18n/ru.json | 2 + .../components/source_editor/input_form.rs | 2 + .../source_editor/input_form/common.rs | 15 +++-- shared/src/model/config/input.rs | 19 ++++++ 21 files changed, 364 insertions(+), 10 deletions(-) diff --git a/backend/app/src/api/api_utils/mod.rs b/backend/app/src/api/api_utils/mod.rs index 9c5008d57..5ba652a99 100644 --- a/backend/app/src/api/api_utils/mod.rs +++ b/backend/app/src/api/api_utils/mod.rs @@ -25,7 +25,7 @@ use crate::{ }, MediaServerError, MediaServerErrorKind, MediaServerHttpClient, MediaServerImageRef, }, - model::{AppConfig, ConfigInput, ConfigTarget, InputUserInfo, ProxyUserCredentials}, + model::{AppConfig, ConfigInput, ConfigInputFlags, ConfigTarget, InputUserInfo, ProxyUserCredentials}, processing::{ parser::hls::{rewrite_hls, RewriteHlsProps}, processor::re_resolve_stalker_url, @@ -1429,6 +1429,28 @@ pub(crate) async fn resolve_initial_stalker_playback_url( }) } +pub(crate) async fn resolve_stream_user_agent_index( + app_state: &AppState, + input: &ConfigInput, + has_provider_handle: bool, + username: &str, + session_owner: Option<&str>, +) -> Option { + if input.has_flag(ConfigInputFlags::UserAgentStreamIndex) && has_provider_handle { + if let Some(session_token) = session_owner { + app_state + .active_users + .get_or_assign_user_agent_stream_index(username, session_token) + .await + .or_else(|| Some(app_state.active_users.next_user_agent_stream_index())) + } else { + Some(app_state.active_users.next_user_agent_stream_index()) + } + } else { + None + } +} + #[allow(clippy::too_many_arguments, clippy::too_many_lines, clippy::fn_params_excessive_bools)] async fn create_stream_response_details( app_state: &Arc, @@ -1472,6 +1494,14 @@ async fn create_stream_response_details( }, ) .await; + let user_agent_stream_index = resolve_stream_user_agent_index( + app_state, + input, + streaming_strategy.provider_handle.is_some(), + username, + session_owner, + ) + .await; let mut grace_period_options = app_state.get_grace_options(); grace_period_options.period_millis = get_grace_period_millis( connection_permission, @@ -1525,6 +1555,7 @@ async fn create_stream_response_details( request_url: None, session_headers: session_headers.cloned(), provider_session_headers: HashMap::new(), + user_agent_stream_index, grace_period: grace_period_options, provider_grace_active: false, disable_provider_grace: false, @@ -1598,6 +1629,10 @@ async fn create_stream_response_details( content_representation, }); + if let Some(stream_index) = user_agent_stream_index { + provider_stream_factory_options.apply_user_agent_stream_index(stream_index); + } + let provider_config = input.get_resolve_provider(url.as_ref()); provider_stream_factory_options.set_provider(provider_config); if input.input_type.is_stalker() { @@ -1663,6 +1698,9 @@ async fn create_stream_response_details( content_representation, }, ); + if let Some(stream_index) = user_agent_stream_index { + options.apply_user_agent_stream_index(stream_index); + } options.set_provider(input.get_resolve_provider(url.as_ref())); options.require_public_destination(); let retry_reconnect_flag = options.get_reconnect_flag_clone(); @@ -1726,6 +1764,7 @@ async fn create_stream_response_details( request_url: Some(request_url.clone()), session_headers: session_headers.cloned(), provider_session_headers, + user_agent_stream_index, grace_period: grace_period_options, provider_grace_active, disable_provider_grace: false, @@ -2414,6 +2453,15 @@ pub(crate) async fn stream_response( // 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 user_agent_stream_index = stream_details.user_agent_stream_index; + // Persist this before response construction because some item types skip the later + // create_user_session path, including placeholder sessions created by ensure_user_session_placeholder. + if let Some(stream_index) = user_agent_stream_index { + app_state + .active_users + .set_user_agent_stream_index_if_absent(&user.username, session_token, stream_index) + .await; + } 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() { @@ -2561,6 +2609,12 @@ pub(crate) async fn stream_response( socket_bound, }) .await; + if let Some(stream_index) = user_agent_stream_index { + app_state + .active_users + .set_user_agent_stream_index_if_absent(&user.username, session_token, stream_index) + .await; + } if should_pin_provider { let reservation_ttl_secs = get_session_reservation_ttl_secs(app_state, item_type); if reservation_ttl_secs > 0 { @@ -2679,6 +2733,12 @@ async fn detected_catchup_hls_response(params: DetectedCatchupHlsResponseParams< socket_bound: false, }) .await; + if let Some(stream_index) = stream_details.user_agent_stream_index { + app_state + .active_users + .set_user_agent_stream_index_if_absent(&user.username, &created_session_token, stream_index) + .await; + } if !stream_details.provider_session_headers.is_empty() { app_state .active_users diff --git a/backend/app/src/api/api_utils/tests.rs b/backend/app/src/api/api_utils/tests.rs index b7d5ffa85..e66877c5c 100644 --- a/backend/app/src/api/api_utils/tests.rs +++ b/backend/app/src/api/api_utils/tests.rs @@ -916,6 +916,7 @@ async fn forced_legacy_hls_test_response( provider: Arc::clone(&input.name), stream_url: origin_url.as_str().intern(), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr: client_addr, socket_bound: false, active_addrs: vec![client_addr], @@ -1345,6 +1346,7 @@ fn test_should_allow_exhausted_shared_reconnect_only_for_matching_shared_session provider: Arc::::from("provider"), stream_url: Arc::::from("http://provider/live/449924.ts"), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, 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!())], @@ -1721,6 +1723,7 @@ fn create_test_session( _ => "http://provider-1.example/live/42.ts", }), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr: "127.0.0.1:55555".parse().unwrap_or_else(|_| unreachable!()), socket_bound: item_type.uses_socket_bound_session(), active_addrs: Vec::new(), @@ -4292,6 +4295,7 @@ async fn should_pin_provider_for_session_skips_reservation_on_failure_custom_vid request_url: None, session_headers: None, provider_session_headers: HashMap::new(), + user_agent_stream_index: None, grace_period: GracePeriodOptions::default(), provider_grace_active: false, disable_provider_grace: false, @@ -4312,6 +4316,7 @@ async fn should_pin_provider_for_session_skips_reservation_on_failure_custom_vid request_url: None, session_headers: None, provider_session_headers: HashMap::new(), + user_agent_stream_index: None, grace_period: GracePeriodOptions::default(), provider_grace_active: false, disable_provider_grace: false, @@ -4339,6 +4344,7 @@ async fn should_pin_provider_for_session_skips_reservation_on_failure_custom_vid request_url: None, session_headers: None, provider_session_headers: HashMap::new(), + user_agent_stream_index: None, grace_period: GracePeriodOptions::default(), provider_grace_active: false, disable_provider_grace: false, @@ -5265,6 +5271,7 @@ fn session_reacquire_cleanup_addrs_excludes_current_and_deduplicates() { provider: "provider-a".intern(), stream_url: "http://localhost/movie.mkv".intern(), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr: seek, socket_bound: false, active_addrs: vec![primary, overlap, seek, overlap], diff --git a/backend/app/src/api/endpoints/hls_api/manifest.rs b/backend/app/src/api/endpoints/hls_api/manifest.rs index 28536c91c..80db2035d 100644 --- a/backend/app/src/api/endpoints/hls_api/manifest.rs +++ b/backend/app/src/api/endpoints/hls_api/manifest.rs @@ -1462,6 +1462,27 @@ pub(super) async fn prepare_hls_canonical_manifest_origin_runtime( } } +pub(super) async fn apply_hls_user_agent_stream_index( + session: &HlsSessionHandle, + headers: &mut HeaderMap, + enabled: bool, + active_users: &crate::api::model::ActiveUserManager, +) { + if !enabled { + return; + } + let stream_index = { + let mut session = session.write().await; + if session.user_agent_stream_index.is_none() { + session.user_agent_stream_index = Some(active_users.next_user_agent_stream_index()); + } + session.user_agent_stream_index + }; + if let Some(stream_index) = stream_index { + request::append_user_agent_stream_index(headers, stream_index); + } +} + #[allow(clippy::too_many_arguments, clippy::too_many_lines)] pub(super) async fn try_hls_cache_canonical_manifest_response( app_state: &Arc, @@ -1471,7 +1492,7 @@ pub(super) async fn try_hls_cache_canonical_manifest_response( access_lease_id: &HlsAccessLeaseId, access_lease_state: HlsAccessLeaseState, origin: HlsCacheManifestOrigin<'_>, - headers: HeaderMap, + mut headers: HeaderMap, server_path: Option<&str>, _original_hls_entry_path: &str, refresh_ordering: HlsManifestRefreshOrdering, @@ -1559,6 +1580,13 @@ pub(super) async fn try_hls_cache_canonical_manifest_response( Ok(prepared) => prepared, Err(response) => return Some(*response), }; + apply_hls_user_agent_stream_index( + &session, + &mut headers, + origin.input.has_flag(ConfigInputFlags::UserAgentStreamIndex), + &app_state.active_users, + ) + .await; let url_failover_provider = effective_hls_url_failover_provider_for_fetch_url( &prepared_origin.fetch_url, prepared_origin.url_failover_provider.clone(), diff --git a/backend/app/src/api/endpoints/hls_api/segment.rs b/backend/app/src/api/endpoints/hls_api/segment.rs index 4ec80f562..180881e67 100644 --- a/backend/app/src/api/endpoints/hls_api/segment.rs +++ b/backend/app/src/api/endpoints/hls_api/segment.rs @@ -2297,7 +2297,7 @@ pub(in crate::api) async fn handle_hls_stream_request( let disabled_headers = app_state.get_disabled_headers(); let default_user_agent = app_state.app_config.config.load().default_user_agent.clone(); - let headers = build_hls_manifest_request_headers( + let mut headers = build_hls_manifest_request_headers( &input.headers, req_headers, disabled_headers.as_ref(), @@ -2464,6 +2464,24 @@ pub(in crate::api) async fn handle_hls_stream_request( ) }; + let user_agent_stream_index = crate::api::api_utils::resolve_stream_user_agent_index( + app_state, + input, + provider_handle.is_some(), + &user.username, + session_token.as_deref(), + ) + .await; + if let Some(stream_index) = user_agent_stream_index { + request::append_user_agent_stream_index(&mut headers, stream_index); + if let Some(session_token) = session_token.as_deref() { + app_state + .active_users + .set_user_agent_stream_index_if_absent(&user.username, session_token, stream_index) + .await; + } + } + // Playlist requests only need the chosen provider account to derive the URL and pin the session. // Holding the provider slot until the first segment request causes stale active connections and // breaks forced same-account reuse on the next HLS/Catchup stream request. diff --git a/backend/app/src/api/endpoints/hls_api/tests.rs b/backend/app/src/api/endpoints/hls_api/tests.rs index 94d603c21..99a2ab224 100644 --- a/backend/app/src/api/endpoints/hls_api/tests.rs +++ b/backend/app/src/api/endpoints/hls_api/tests.rs @@ -3,8 +3,8 @@ use super::{ hls_manifest_terminal_preflight, hls_terminal_endpoint_action, hls_terminal_failed_closed_response, HlsManifestTerminalPreflight, HlsTerminalEndpointAction, }, - build_hls_manifest_request_headers, extract_hls_provider_session_headers, hls_api_register, - hls_availability_reevaluation_registration_failure_response, hls_canonical_owner_registration, + apply_hls_user_agent_stream_index, build_hls_manifest_request_headers, extract_hls_provider_session_headers, + hls_api_register, hls_availability_reevaluation_registration_failure_response, hls_canonical_owner_registration, hls_temporary_resource_unavailable_response, m3u_archive_epg_reference_ts, m3u_catchup_epg_reference_from_session_token, resolve_leaked_hls_relative_origin, HlsCanonicalOwnerRegistration, HlsCanonicalOwnerRegistrationFailure, HlsCanonicalOwnerRegistrationKind, MAX_HLS_MANIFEST_BYTES, @@ -3151,6 +3151,32 @@ fn hls_manifest_headers_apply_disabled_headers_and_default_user_agent_policy() { assert!(!headers.contains_key("cf-ray")); } +#[tokio::test] +async fn hls_user_agent_stream_index_is_stable_for_the_session() { + let app_state = test_app_state(); + let session = Arc::new(tokio::sync::RwLock::new(HlsSession::new(HlsSessionKey::new(1, "stream-1"), b"secret", 0))); + let mut manifest_headers = HeaderMap::new(); + manifest_headers.insert(header::USER_AGENT, HeaderValue::from_static("VLC/3.0")); + + apply_hls_user_agent_stream_index(&session, &mut manifest_headers, true, &app_state.active_users).await; + let stream_index = session.read().await.user_agent_stream_index.unwrap_or_default(); + assert_ne!(stream_index, 0); + assert_eq!( + manifest_headers.get(header::USER_AGENT).and_then(|value| value.to_str().ok()), + Some(format!("VLC/3.0 {stream_index}").as_str()) + ); + + let mut segment_headers = HeaderMap::new(); + segment_headers.insert(header::USER_AGENT, HeaderValue::from_static("VLC/3.0")); + apply_hls_user_agent_stream_index(&session, &mut segment_headers, true, &app_state.active_users).await; + + assert_eq!( + segment_headers.get(header::USER_AGENT).and_then(|value| value.to_str().ok()), + Some(format!("VLC/3.0 {stream_index}").as_str()) + ); + assert_eq!(session.read().await.user_agent_stream_index, Some(stream_index)); +} + #[tokio::test] async fn legacy_hls_manifest_decodes_supported_origin_codings_and_enforces_identity() { const MANIFEST: &str = "#EXTM3U\n#EXT-X-TARGETDURATION:4\n#EXTINF:4.0,\nsegment.ts\n"; @@ -10779,6 +10805,7 @@ fn stats_provider_test_user_session(provider: &str) -> UserSession { provider: Arc::from(provider), stream_url: Arc::from("http://origin.example.com/live/12345.m3u8"), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr: test_addr(), socket_bound: false, active_addrs: Vec::new(), diff --git a/backend/app/src/api/model/streams/active_client_stream.rs b/backend/app/src/api/model/streams/active_client_stream.rs index c2c661e37..d7fc36041 100644 --- a/backend/app/src/api/model/streams/active_client_stream.rs +++ b/backend/app/src/api/model/streams/active_client_stream.rs @@ -591,6 +591,11 @@ fn create_deferred_provider_open_future( content_representation: stream_details.content_representation, }) .for_deferred_open(); + if stream_details.provider_handle.is_some() { + if let Some(stream_index) = stream_details.user_agent_stream_index { + provider_stream_factory_options.apply_user_agent_stream_index(stream_index); + } + } provider_stream_factory_options.set_provider(input.get_resolve_provider(stream_url.as_ref())); Some(DeferredProviderOpenState::Pending(Box::new(DeferredProviderOpenContext { @@ -1839,6 +1844,7 @@ mod tests { request_url: Some("http://provider-1.example/live/1".intern()), session_headers: None, provider_session_headers: HashMap::new(), + user_agent_stream_index: None, grace_period: GracePeriodOptions { period_millis: 100, timeout_secs: 0, hold_stream: true }, provider_grace_active: true, disable_provider_grace: false, @@ -2749,6 +2755,7 @@ mod tests { request_url: Some("http://provider-1.example/live/2.ts".intern()), session_headers: None, provider_session_headers: HashMap::new(), + user_agent_stream_index: None, grace_period: GracePeriodOptions { period_millis: 100, timeout_secs: 0, hold_stream: true }, provider_grace_active: false, disable_provider_grace: false, diff --git a/backend/app/src/api/model/streams/provider_stream_factory.rs b/backend/app/src/api/model/streams/provider_stream_factory.rs index 496db11c9..73d3acd12 100644 --- a/backend/app/src/api/model/streams/provider_stream_factory.rs +++ b/backend/app/src/api/model/streams/provider_stream_factory.rs @@ -222,6 +222,10 @@ impl ProviderStreamFactoryOptions { pub fn set_provider(&mut self, provider: Option>) { self.provider = provider; } + pub fn apply_user_agent_stream_index(&mut self, stream_index: u64) { + crate::utils::request::append_user_agent_stream_index(&mut self.headers, stream_index); + } + pub fn require_public_destination(&mut self) { self.flags.set(ProviderStreamFactoryFlags::PublicDestinationRequired); } diff --git a/backend/app/src/api/panel_api/mod.rs b/backend/app/src/api/panel_api/mod.rs index e1a1edddd..727ba9d59 100644 --- a/backend/app/src/api/panel_api/mod.rs +++ b/backend/app/src/api/panel_api/mod.rs @@ -3683,6 +3683,7 @@ pub fn create_panel_api_provisioning_stream_details( request_url: None, session_headers: None, provider_session_headers: HashMap::new(), + user_agent_stream_index: None, grace_period: *grace_period_options, provider_grace_active: false, disable_provider_grace: true, @@ -3711,6 +3712,7 @@ pub fn create_panel_api_provisioning_stream_details( request_url: None, session_headers: None, provider_session_headers: HashMap::new(), + user_agent_stream_index: None, grace_period: *grace_period_options, provider_grace_active: false, disable_provider_grace: true, diff --git a/backend/core/src/model/config/input.rs b/backend/core/src/model/config/input.rs index 96ff8d4b0..ce5933658 100644 --- a/backend/core/src/model/config/input.rs +++ b/backend/core/src/model/config/input.rs @@ -44,6 +44,7 @@ create_bitset!( XtreamLiveStreamUsePrefix, XtreamLiveStreamWithoutExtension, DisableHlsStreaming, + UserAgentStreamIndex, ResolveTmdb, ResolveBackground, ResolveSeries, @@ -89,6 +90,7 @@ impl From<&ConfigInputOptionsDto> for ConfigInputOptions { (xtream_live_stream_use_prefix, XtreamLiveStreamUsePrefix), (xtream_live_stream_without_extension, XtreamLiveStreamWithoutExtension), (disable_hls_streaming, DisableHlsStreaming), + (user_agent_stream_index, UserAgentStreamIndex), (resolve_tmdb, ResolveTmdb), (resolve_background, ResolveBackground), (resolve_series, ResolveSeries), @@ -1097,6 +1099,15 @@ mod tests { assert!(options.has_flag(ConfigInputFlags::DisableHlsStreaming)); } + #[test] + fn input_options_conversion_sets_user_agent_stream_index_flag() { + let dto = ConfigInputOptionsDto { user_agent_stream_index: true, ..ConfigInputOptionsDto::default() }; + + let options = ConfigInputOptions::from(&dto); + + assert!(options.has_flag(ConfigInputFlags::UserAgentStreamIndex)); + } + #[test] fn media_server_runtime_mapping_preserves_safe_defaults() { let dto = ConfigInputDto { diff --git a/backend/core/src/utils/network/request.rs b/backend/core/src/utils/network/request.rs index ccb59eb0c..25340d6b8 100644 --- a/backend/core/src/utils/network/request.rs +++ b/backend/core/src/utils/network/request.rs @@ -1796,6 +1796,42 @@ pub fn overlay_upstream_user_agent( } } +/// Appends a stable playback-session index to an already resolved User-Agent. +pub fn append_user_agent_stream_index(headers: &mut HeaderMap, stream_index: u64) { + let Some(user_agent) = headers.get(axum::http::header::USER_AGENT).map(HeaderValue::as_bytes) else { + return; + }; + let Some(first) = user_agent.iter().position(|byte| !byte.is_ascii_whitespace()) else { + return; + }; + let last = user_agent.iter().rposition(|byte| !byte.is_ascii_whitespace()).map_or(first, |index| index + 1); + let user_agent = &user_agent[first..last]; + + let mut digits = [0_u8; 20]; + let mut cursor = digits.len(); + let mut remaining = stream_index; + loop { + cursor -= 1; + digits[cursor] = b'0' + (remaining % 10) as u8; + remaining /= 10; + if remaining == 0 { + break; + } + } + let suffix = &digits[cursor..]; + if user_agent.strip_suffix(suffix).is_some_and(|prefix| prefix.ends_with(b" ")) { + return; + } + + let mut indexed_user_agent = Vec::with_capacity(user_agent.len() + 1 + suffix.len()); + indexed_user_agent.extend_from_slice(user_agent); + indexed_user_agent.push(b' '); + indexed_user_agent.extend_from_slice(suffix); + if let Ok(value) = HeaderValue::from_bytes(&indexed_user_agent) { + headers.insert(axum::http::header::USER_AGENT, value); + } +} + // read local file content and return it as a string. // Gzipped file content is supported. pub async fn get_local_file_content(file_path: &Path) -> Result { @@ -3134,7 +3170,7 @@ mod tests { #[test] fn test_get_request_headers_prioritization() { - use super::{get_request_headers, overlay_upstream_user_agent}; + use super::{append_user_agent_stream_index, get_request_headers, overlay_upstream_user_agent}; use axum::http::header::USER_AGENT; // Case 1: No headers provided -> Default UA @@ -3186,6 +3222,22 @@ mod tests { ); overlay_upstream_user_agent(&mut headers, Some("Blocked-Channel-UA"), Some(&disabled)); assert!(!headers.contains_key(USER_AGENT)); + + let mut headers = + get_request_headers::(None, None, None, Some("VLC/3.0")); + append_user_agent_stream_index(&mut headers, 42); + assert_eq!(headers.get(USER_AGENT).and_then(|value| value.to_str().ok()), Some("VLC/3.0 42")); + append_user_agent_stream_index(&mut headers, 42); + assert_eq!(headers.get(USER_AGENT).and_then(|value| value.to_str().ok()), Some("VLC/3.0 42")); + + let mut headers = HeaderMap::new(); + let extended_user_agent = HeaderValue::from_bytes(b"Receiver/1.0 \xE4"); + assert!(extended_user_agent.is_ok()); + if let Ok(value) = extended_user_agent { + headers.insert(USER_AGENT, value); + } + append_user_agent_stream_index(&mut headers, 7); + assert_eq!(headers.get(USER_AGENT).map(HeaderValue::as_bytes), Some(b"Receiver/1.0 \xE4 7".as_slice())); } #[test] diff --git a/backend/hls/src/session.rs b/backend/hls/src/session.rs index 6a969665d..a6bad6654 100644 --- a/backend/hls/src/session.rs +++ b/backend/hls/src/session.rs @@ -269,6 +269,8 @@ pub struct HlsSession { pub active_segment_fetches: usize, pub segment_fetch_notifiers: HashMap>, pub origin_request_headers: HeaderMap, + /// Stable playback-session suffix used by all origin requests in this session. + pub user_agent_stream_index: Option, pub origin_provider_session_headers: HeaderMap, pub activity: HlsSessionActivity, pub origin_epoch: u64, @@ -338,6 +340,7 @@ impl HlsSession { active_segment_fetches: 0, segment_fetch_notifiers: HashMap::new(), origin_request_headers: HeaderMap::new(), + user_agent_stream_index: None, origin_provider_session_headers: HeaderMap::new(), activity: HlsSessionActivity::default(), origin_epoch: 0, @@ -816,6 +819,7 @@ impl fmt::Debug for HlsSession { .field("active_segment_fetches", &self.active_segment_fetches) .field("segment_fetch_notifiers_len", &self.segment_fetch_notifiers.len()) .field("origin_request_headers_len", &self.origin_request_headers.len()) + .field("user_agent_stream_index", &self.user_agent_stream_index) .field("origin_provider_session_headers_len", &self.origin_provider_session_headers.len()) .field("activity", &self.activity) .field("origin_control", &self.origin_control) diff --git a/backend/session/src/active_user_manager/mod.rs b/backend/session/src/active_user_manager/mod.rs index bd4a1ac30..97a4d9148 100644 --- a/backend/session/src/active_user_manager/mod.rs +++ b/backend/session/src/active_user_manager/mod.rs @@ -134,6 +134,8 @@ pub struct UserSession { pub provider: Arc, pub stream_url: Arc, pub provider_session_headers: HashMap, + /// Stable suffix appended to upstream User-Agent headers for this playback session. + pub user_agent_stream_index: Option, pub addr: SocketAddr, pub socket_bound: bool, pub active_addrs: Vec, @@ -537,6 +539,7 @@ pub struct ActiveUserManager { grace_period_timeout_secs: AtomicU64, adaptive_session_ttl_secs: AtomicU64, log_active_user: AtomicBool, + next_user_agent_stream_index: AtomicU64, gc_ts: Option, connections: RwLock, adaptive_expiry_queue: Arc>>>, @@ -636,6 +639,7 @@ impl ActiveUserManager { grace_period_timeout_secs: AtomicU64::new(grace_period_timeout_secs), adaptive_session_ttl_secs: AtomicU64::new(get_adaptive_session_ttl_secs(config)), log_active_user: AtomicBool::new(log_active_user), + next_user_agent_stream_index: AtomicU64::new(1), connections: RwLock::new(UserConnections::default()), adaptive_expiry_queue: Arc::new(Mutex::new(BinaryHeap::new())), adaptive_expiry_index: Arc::new(Mutex::new(HashMap::new())), @@ -1691,6 +1695,7 @@ impl ActiveUserManager { provider: params.provider.intern(), stream_url: params.stream_url.intern(), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr: *params.addr, socket_bound: params.socket_bound, active_addrs: vec![*params.addr], @@ -2837,6 +2842,37 @@ impl ActiveUserManager { true } + /// Returns the stable User-Agent suffix for a playback session, assigning one on first use. + pub async fn get_or_assign_user_agent_stream_index(&self, username: &str, token: &str) -> Option { + let mut user_connections = self.connections.write().await; + let session = + user_connections.by_key.get_mut(username)?.sessions.iter_mut().find(|session| session.token == token)?; + Some(*session.user_agent_stream_index.get_or_insert_with(|| self.next_user_agent_stream_index())) + } + + /// Allocates the next process-local, globally unique User-Agent stream index. + pub fn next_user_agent_stream_index(&self) -> u64 { + self.next_user_agent_stream_index + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + Some(if current == 0 || current == u64::MAX { 1 } else { current + 1 }) + }) + .map_or(1, |val| if val == 0 { 1 } else { val }) + } + + /// Persists a previously allocated index without replacing an existing session identity. + pub async fn set_user_agent_stream_index_if_absent(&self, username: &str, token: &str, index: u64) -> bool { + let mut user_connections = self.connections.write().await; + let Some(session) = user_connections + .by_key + .get_mut(username) + .and_then(|connection| connection.sessions.iter_mut().find(|session| session.token == token)) + else { + return false; + }; + session.user_agent_stream_index.get_or_insert(index); + 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)?; diff --git a/backend/session/src/active_user_manager/tests.rs b/backend/session/src/active_user_manager/tests.rs index 7593b7197..17473b82c 100644 --- a/backend/session/src/active_user_manager/tests.rs +++ b/backend/session/src/active_user_manager/tests.rs @@ -149,6 +149,42 @@ async fn target_scoped_session_lookup_does_not_use_same_virtual_id_from_other_ta .is_none()); } +#[tokio::test] +async fn user_agent_stream_index_is_stable_per_session_and_unique_between_sessions() { + 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::from(([127, 0, 0, 1], 55_500)); + let mut user = ProxyUserCredentials::default(); + user.username = "indexed-user".to_string(); + + for token in ["indexed-session-a", "indexed-session-b"] { + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: token, + virtual_id: 42, + 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 first = manager.get_or_assign_user_agent_stream_index(&user.username, "indexed-session-a").await; + let repeated = manager.get_or_assign_user_agent_stream_index(&user.username, "indexed-session-a").await; + let second = manager.get_or_assign_user_agent_stream_index(&user.username, "indexed-session-b").await; + + assert!(first.is_some()); + assert_eq!(repeated, first); + assert!(second.is_some()); + assert_ne!(second, first); +} + /// Session refresh normalizes Expired -> Prepared. /// When a new request arrives on an expired session, the lifecycle should be /// reset to Prepared so that full activation evaluation happens. @@ -174,6 +210,7 @@ async fn create_user_session_normalizes_expired_lifecycle() { provider: "provider-a".intern(), stream_url: "http://localhost/live.m3u8".intern(), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr, socket_bound: false, active_addrs: vec![addr], @@ -235,6 +272,7 @@ async fn create_user_session_does_not_normalize_pending_provider_lifecycle() { provider: "provider-a".intern(), stream_url: "http://localhost/live.m3u8".intern(), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr, socket_bound: false, active_addrs: vec![addr], @@ -5505,6 +5543,7 @@ async fn check_divergence_detects_connection_count_mismatch() { provider: "provider-a".intern(), stream_url: "http://localhost/stream.ts".intern(), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr, socket_bound: false, active_addrs: vec![addr], @@ -5548,6 +5587,7 @@ async fn check_divergence_detects_stream_without_counted_session() { provider: "provider-a".intern(), stream_url: "http://localhost/stream.ts".intern(), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr, socket_bound: false, active_addrs: vec![addr], @@ -5632,6 +5672,7 @@ async fn divergence_log_rate_limited_within_cooldown_window() { provider: "provider-a".intern(), stream_url: "http://localhost/stream.ts".intern(), provider_session_headers: HashMap::new(), + user_agent_stream_index: None, addr, socket_bound: false, active_addrs: vec![addr], diff --git a/backend/session/src/stream.rs b/backend/session/src/stream.rs index 11c9cc49b..3e81444d6 100644 --- a/backend/session/src/stream.rs +++ b/backend/session/src/stream.rs @@ -73,6 +73,7 @@ pub struct StreamDetails { pub request_url: Option>, pub session_headers: Option>, pub provider_session_headers: HashMap, + pub user_agent_stream_index: Option, pub grace_period: GracePeriodOptions, pub provider_grace_active: bool, pub disable_provider_grace: bool, @@ -96,6 +97,7 @@ impl Clone for StreamDetails { request_url: self.request_url.clone(), session_headers: self.session_headers.clone(), provider_session_headers: self.provider_session_headers.clone(), + user_agent_stream_index: self.user_agent_stream_index, grace_period: self.grace_period, provider_grace_active: self.provider_grace_active, disable_provider_grace: self.disable_provider_grace, @@ -116,6 +118,7 @@ impl StreamDetails { request_url: None, session_headers: None, provider_session_headers: HashMap::new(), + user_agent_stream_index: None, grace_period: grace_period_options, provider_grace_active: false, disable_provider_grace: false, diff --git a/docs/src/configuration/source.md b/docs/src/configuration/source.md index c444a4b64..faefc873e 100644 --- a/docs/src/configuration/source.md +++ b/docs/src/configuration/source.md @@ -248,6 +248,7 @@ specific provider. | `xtream_live_stream_without_extension` | Bool | `false` | Strips `.ts` from generated stream URLs. | | `xtream_live_stream_use_prefix` | Bool | `true` | Injects the `/live/` prefix into URLs. | | `disable_hls_streaming` | Bool | `false` | Rewrites live `.m3u8` requests to `.ts` and bypasses Tuliprox HLS handling. | +| `user_agent_stream_index` | Bool | `false` | Appends a process-local stream index to upstream `User-Agent` requests (e.g. `VLC/3.0 42`), keeping it stable for the session. | | `resolve_tmdb` | Bool | `false` | Enables TMDB queries for this specific input based on parsed titles to fill missing posters and release years. | | `probe_stream` | Bool | `false` | Uses FFprobe to read A/V details (HDR, 4K). Respects `max_connections`. | | `resolve_background` | Bool | `true` | Metadata scans run asynchronously in the background so the general playlist update (which blocks clients) finishes instantly. | @@ -277,6 +278,25 @@ inputs: disable_hls_streaming: true ``` +#### Stable User-Agent Stream Index Example + +```yaml +inputs: + - name: indexed-provider + type: xtream + url: http://provider.example + username: user + password: pass + headers: + User-Agent: VLC/3.0 + options: + user_agent_stream_index: true +``` + +The resulting upstream requests use a value such as `User-Agent: VLC/3.0 42`. The counter is global to the running +Tuliprox process and uses a 64-bit value to avoid reusing an index while an older session is still active. It starts +again when the process restarts. Shared playback uses the identity of the shared upstream session. + #### Stalker playback notes * Stalker playlist preview in the Web UI now works through the same protected playlist endpoints used for other input types. diff --git a/frontend/public/assets/i18n/ar.json b/frontend/public/assets/i18n/ar.json index 6c55d507c..07487eb11 100644 --- a/frontend/public/assets/i18n/ar.json +++ b/frontend/public/assets/i18n/ar.json @@ -367,6 +367,7 @@ "XTREAM_LIVE_STREAM_USE_PREFIX": "إذا كان صحيحًا، يُضيف بادئة /live/ إلى روابط بث Xtream (الافتراضي صحيح).", "XTREAM_LIVE_STREAM_WITHOUT_EXTENSION": "إذا كان صحيحًا، يزيل امتداد .ts من روابط بث Xtream (الافتراضي خطأ).", "DISABLE_HLS_STREAMING": "يستبدل طلبات البث المباشر من Xtream بصيغة .m3u8 بصيغة .ts. فعّله فقط عندما يدعم المزوّد نقاط نهاية MPEG-TS؛ هذا الخيار لا يقوم بتحويل ترميز HLS.", + "USER_AGENT_STREAM_INDEX": "يضيف فهرسًا رقميًا محليًا للعملية إلى وكيل المستخدم المضبوط. يبقى الفهرس ثابتًا لجلسة تشغيل واحدة عبر القوائم والمقاطع والتنقل وإعادة الاتصال وتبديل المزوّد.", "XTREAM_SKIP_LIVE": "إذا كان صحيحًا، يتم تخطي قسم البث المباشر لمزود Xtream أثناء التحديث.", "XTREAM_SKIP_SERIES": "إذا كان صحيحًا، يتم تخطي قسم المسلسلات لمزود Xtream أثناء التحديث.", "XTREAM_SKIP_VOD": "إذا كان صحيحًا، يتم تخطي قسم الفيديو حسب الطلب لمزود Xtream أثناء التحديث." @@ -1170,6 +1171,7 @@ "LIVE_STREAM_USE_PREFIX": "استخدام بادئة", "LIVE_STREAM_WITHOUT_EXTENSION": "بدون امتداد", "DISABLE_HLS_STREAMING": "إجبار MPEG-TS", + "USER_AGENT_STREAM_INDEX": "إضافة فهرس البث إلى وكيل المستخدم", "LOG": "سجل", "LOGS": "السجلات", "LIVE_LOGS": "السجلات المباشرة", diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index d46253d31..1047def7e 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -367,6 +367,7 @@ "XTREAM_LIVE_STREAM_USE_PREFIX": "If true, adds /live/ prefix to Xtream stream links (default true).", "XTREAM_LIVE_STREAM_WITHOUT_EXTENSION": "If true, removes .ts extension from Xtream stream links (default false).", "DISABLE_HLS_STREAMING": "Replaces live Xtream .m3u8 requests with .ts. Enable only when the provider supports MPEG-TS endpoints; this does not transcode HLS.", + "USER_AGENT_STREAM_INDEX": "Appends a process-local numeric index to the configured User-Agent. The index stays stable for one playback session across manifests, segments, seeks, reconnects, and provider failover.", "XTREAM_SKIP_LIVE": "If true, the live section of the Xtream provider is skipped during update.", "XTREAM_SKIP_SERIES": "If true, the series section of the Xtream provider is skipped during update.", "XTREAM_SKIP_VOD": "If true, the VOD section of the Xtream provider is skipped during update." @@ -1177,6 +1178,7 @@ "LIVE_STREAM_USE_PREFIX": "Use prefix", "LIVE_STREAM_WITHOUT_EXTENSION": "Without ext.", "DISABLE_HLS_STREAMING": "Force MPEG-TS", + "USER_AGENT_STREAM_INDEX": "Append stream index to User-Agent", "LOG": "Log", "LOGS": "Logs", "LIVE_LOGS": "Live Logs", diff --git a/frontend/public/assets/i18n/ru.json b/frontend/public/assets/i18n/ru.json index 15f20d8f4..723d2724b 100644 --- a/frontend/public/assets/i18n/ru.json +++ b/frontend/public/assets/i18n/ru.json @@ -344,6 +344,7 @@ "XTREAM_LIVE_STREAM_USE_PREFIX": "Если true, добавляет префикс /live/ к ссылкам на прямые трансляции Xtream (по умолчанию true).", "XTREAM_LIVE_STREAM_WITHOUT_EXTENSION": "Если true, удаляет расширение .ts из ссылок на прямые трансляции Xtream (по умолчанию false).", "DISABLE_HLS_STREAMING": "Заменяет запросы прямых трансляций Xtream в формате .m3u8 на .ts. Включайте, только если провайдер поддерживает конечные точки MPEG-TS; перекодирование HLS не выполняется.", + "USER_AGENT_STREAM_INDEX": "Добавляет локальный числовой индекс процесса к настроенному User-Agent. Индекс остается постоянным для одного сеанса воспроизведения в манифестах, сегментах, перемотках, переподключениях и при переключении провайдера.", "XTREAM_SKIP_LIVE": "Если true, раздел прямых трансляций провайдера Xtream пропускается при обновлении.", "XTREAM_SKIP_SERIES": "Если true, раздел сериалов провайдера Xtream пропускается при обновлении.", "XTREAM_SKIP_VOD": "Если true, раздел VOD провайдера Xtream пропускается при обновлении." @@ -1119,6 +1120,7 @@ "LIVE_STREAM_USE_PREFIX": "Использовать префикс", "LIVE_STREAM_WITHOUT_EXTENSION": "Без расширения", "DISABLE_HLS_STREAMING": "Принудительный MPEG-TS", + "USER_AGENT_STREAM_INDEX": "Добавлять индекс потока в User-Agent", "LOG": "Лог", "LOGS": "Логи", "LIVE_LOGS": "Живые логи", diff --git a/frontend/src/app/components/source_editor/input_form.rs b/frontend/src/app/components/source_editor/input_form.rs index cd2428ee6..d371f03fd 100644 --- a/frontend/src/app/components/source_editor/input_form.rs +++ b/frontend/src/app/components/source_editor/input_form.rs @@ -86,6 +86,7 @@ const LABEL_SKIP_SERIES: &str = "LABEL.SERIES"; const LABEL_XTREAM_LIVE_STREAM_USE_PREFIX: &str = "LABEL.LIVE_STREAM_USE_PREFIX"; const LABEL_XTREAM_LIVE_STREAM_WITHOUT_EXTENSION: &str = "LABEL.LIVE_STREAM_WITHOUT_EXTENSION"; const LABEL_DISABLE_HLS_STREAMING: &str = "LABEL.DISABLE_HLS_STREAMING"; +const LABEL_USER_AGENT_STREAM_INDEX: &str = "LABEL.USER_AGENT_STREAM_INDEX"; const LABEL_RESOLVE_TMDB: &str = "LABEL.RESOLVE_TMDB"; const LABEL_RESOLVE: &str = "LABEL.RESOLVE"; const LABEL_PROBE: &str = "LABEL.PROBE"; @@ -350,6 +351,7 @@ generate_form_reducer!( XtreamLiveStreamUsePrefix => xtream_live_stream_use_prefix: bool, XtreamLiveStreamWithoutExtension => xtream_live_stream_without_extension: bool, DisableHlsStreaming => disable_hls_streaming: bool, + UserAgentStreamIndex => user_agent_stream_index: bool, ResolveTmdb => resolve_tmdb: bool, ResolveBackground => resolve_background: bool, ResolveSeries => resolve_series: bool, diff --git a/frontend/src/app/components/source_editor/input_form/common.rs b/frontend/src/app/components/source_editor/input_form/common.rs index cdd6b2357..79da84dfd 100644 --- a/frontend/src/app/components/source_editor/input_form/common.rs +++ b/frontend/src/app/components/source_editor/input_form/common.rs @@ -6,7 +6,7 @@ use super::{ LABEL_PROBE_FILTER, LABEL_PROBE_LIVE, LABEL_PROBE_LIVE_INTERVAL_HOURS, LABEL_PROBE_SERIES, LABEL_PROBE_VOD, LABEL_RESOLVE, LABEL_RESOLVE_BACKGROUND, LABEL_RESOLVE_DELAY_SEC, LABEL_RESOLVE_FILTER, LABEL_RESOLVE_SERIES, LABEL_RESOLVE_TMDB, LABEL_RESOLVE_VOD, LABEL_SEQUENTIAL_GROUP, LABEL_SKIP, LABEL_SKIP_LIVE, LABEL_SKIP_SERIES, - LABEL_SKIP_VOD, LABEL_URL, LABEL_USERNAME, LABEL_XTREAM_LIVE_STREAM_USE_PREFIX, + LABEL_SKIP_VOD, LABEL_URL, LABEL_USERNAME, LABEL_USER_AGENT_STREAM_INDEX, LABEL_XTREAM_LIVE_STREAM_USE_PREFIX, LABEL_XTREAM_LIVE_STREAM_WITHOUT_EXTENSION, }; use crate::{ @@ -236,9 +236,16 @@ pub(super) fn InputOptionsForm(props: &InputOptionsFormProps) -> Html { { config_field_child!(translate.t(LABEL_HEADERS), "INPUT_FORM.HEADERS", { let headers = headers.clone(); - html! { } + html! { <> + if props.allow_write { + { edit_field_bool!(state, translate.t(LABEL_USER_AGENT_STREAM_INDEX), user_agent_stream_index, ConfigInputOptionsFormAction::UserAgentStreamIndex) } + } else { + { config_field_bool!(state.form, translate.t(LABEL_USER_AGENT_STREAM_INDEX), user_agent_stream_index) } + } + + } })} } diff --git a/shared/src/model/config/input.rs b/shared/src/model/config/input.rs index 6142b0fbf..1dc19d341 100644 --- a/shared/src/model/config/input.rs +++ b/shared/src/model/config/input.rs @@ -282,6 +282,8 @@ pub struct ConfigInputOptionsDto { #[serde(default, skip_serializing_if = "is_false")] pub disable_hls_streaming: bool, #[serde(default, skip_serializing_if = "is_false")] + pub user_agent_stream_index: bool, + #[serde(default, skip_serializing_if = "is_false")] pub resolve_tmdb: bool, #[serde(default = "default_resolve_background", skip_serializing_if = "is_true")] pub resolve_background: bool, @@ -327,6 +329,7 @@ impl Default for ConfigInputOptionsDto { xtream_live_stream_use_prefix: default_xtream_live_stream_use_prefix(), xtream_live_stream_without_extension: false, disable_hls_streaming: false, + user_agent_stream_index: false, resolve_tmdb: false, resolve_background: default_resolve_background(), resolve_series: false, @@ -354,6 +357,7 @@ impl ConfigInputOptionsDto { && self.xtream_live_stream_use_prefix && !self.xtream_live_stream_without_extension && !self.disable_hls_streaming + && !self.user_agent_stream_index && !self.resolve_tmdb && self.resolve_background && !self.resolve_series @@ -376,6 +380,7 @@ impl ConfigInputOptionsDto { self.xtream_live_stream_use_prefix = default_as_true(); self.xtream_live_stream_without_extension = false; self.disable_hls_streaming = false; + self.user_agent_stream_index = false; self.resolve_tmdb = false; self.resolve_background = default_as_true(); self.resolve_series = false; @@ -2016,6 +2021,20 @@ mod tests { Ok(()) } + #[test] + fn user_agent_stream_index_round_trips_and_keeps_options_non_empty() -> Result<(), serde_json::Error> { + let mut options = ConfigInputOptionsDto { user_agent_stream_index: true, ..ConfigInputOptionsDto::default() }; + let json = serde_json::to_string(&options)?; + let restored: ConfigInputOptionsDto = serde_json::from_str(&json)?; + + assert!(restored.user_agent_stream_index); + assert!(!options.is_empty()); + options.clean(); + assert!(!options.user_agent_stream_index); + assert!(options.is_empty()); + Ok(()) + } + #[test] fn new_stalker_input_has_configuration_block() { let input = ConfigInputDto::new_with_type(InputType::Stalker);