diff --git a/frontend/src/component/main-config-view/main-config-view.tsx b/frontend/src/component/main-config-view/main-config-view.tsx
index 93fcba0e6..014d96491 100644
--- a/frontend/src/component/main-config-view/main-config-view.tsx
+++ b/frontend/src/component/main-config-view/main-config-view.tsx
@@ -38,6 +38,16 @@ const isNumber = (value: string): boolean => {
return !isNaN(value as any);
}
+const getObject = (value: any): any => {
+ if (!value) {
+ return undefined;
+ }
+ if (Object.keys(value).length === 0) {
+ return undefined;
+ }
+ return value;
+}
+
const CONFIG_API_FIELDS = [
{name: 'host', label: 'LABEL.HOST', fieldType: FormFieldType.TEXT, validator: undefined},
{name: 'port', label: 'LABEL.PORT', fieldType: FormFieldType.NUMBER, validator: isNumber},
@@ -240,28 +250,34 @@ export default function MainConfigView(props: MainConfigViewProps) {
let pushoverConfigAvailable = pushoverConfig?.user?.trim().length && pushoverConfig?.token?.length;
const cfgMessaging: MessagingConfig = {
notify_on: messagingConfig.notify_on,
- telegram: telegramConfigAvailable ? telegramConfig : undefined,
- rest: restConfigAvailable ? restConfig : undefined,
- pushover: pushoverConfigAvailable ? pushoverConfig : undefined,
+ telegram: telegramConfigAvailable ? getObject(telegramConfig) : undefined,
+ rest: restConfigAvailable ? getObject(restConfig) : undefined,
+ pushover: pushoverConfigAvailable ? getObject(pushoverConfig) : undefined,
};
- let cfgReverseProxyStreamBuffer: StreamConfig = {
+
+ let reverseProxyStreamConfigAvailable = getObject(reverseProxyStreamConfig) != undefined;
+ let cfgReverseProxyStreamBuffer: StreamConfig = reverseProxyStreamConfigAvailable ? ({
retry: reverseProxyStreamConfig.retry,
- throttle: reverseProxyStreamConfig.throttle,
+ throttle: reverseProxyStreamConfig.throttle?.trim().length ? reverseProxyStreamConfig.throttle : undefined,
grace_period_millis: reverseProxyStreamConfig.grace_period_millis,
grace_period_timeout_secs: reverseProxyStreamConfig.grace_period_timeout_secs,
buffer: {
enabled: reverseProxyStreamBufferConfig.enabled,
size: reverseProxyStreamBufferConfig.size ?? undefined,
}
- }
+ }): undefined;
- const cfgReverseProxy: ReverseProxyConfig = {
+ const reverseProxyConfigAvailable = getObject(reverseProxyConfig) ||
+ getObject(cfgReverseProxyStreamBuffer) ||
+ getObject(reverseProxyCacheConfig) ||
+ getObject(reverseProxyRateLimitConfig);
+ const cfgReverseProxy: ReverseProxyConfig = reverseProxyConfigAvailable ? {
resource_rewrite_disabled: reverseProxyConfig.resource_rewrite_disabled,
- stream: cfgReverseProxyStreamBuffer ?? undefined,
- cache: reverseProxyCacheConfig ?? undefined,
- rate_limit: reverseProxyRateLimitConfig ?? undefined,
- };
+ stream: getObject(cfgReverseProxyStreamBuffer),
+ cache: getObject(reverseProxyCacheConfig),
+ rate_limit: getObject(reverseProxyRateLimitConfig),
+ } : undefined;
const cfgWebUiConfig: WebUiConfig = {
enabled: webUiConfig.enabled ?? false,
@@ -270,10 +286,10 @@ export default function MainConfigView(props: MainConfigViewProps) {
auth: webAuthConfig,
}
- const cfgVideo = {
+ const cfgVideo = getObject(videoConfig) && getObject(videoDownloadConfig) ? {
...videoConfig,
- download: videoDownloadConfig
- };
+ download: getObject(videoDownloadConfig)
+ }: undefined;
const cfgLog = {
...logConfig
@@ -285,12 +301,12 @@ export default function MainConfigView(props: MainConfigViewProps) {
backup_dir: mainConfig.backup_dir,
schedules: mainConfig.schedules?.filter(s => s.schedule?.trim().length),
threads: mainConfig.threads,
- messaging: cfgMessaging,
- reverse_proxy: cfgReverseProxy,
- video: cfgVideo,
- log: cfgLog,
+ messaging: getObject(cfgMessaging),
+ reverse_proxy: getObject(cfgReverseProxy),
+ video: getObject(cfgVideo),
+ log: getObject(cfgLog),
update_on_boot: mainConfig.update_on_boot,
- web_ui: cfgWebUiConfig,
+ web_ui: getObject(cfgWebUiConfig),
};
services.config().saveMainConfig(cfgMain).subscribe({
diff --git a/frontend/src/model/server-config.ts b/frontend/src/model/server-config.ts
index 4226cbc60..192d6659c 100644
--- a/frontend/src/model/server-config.ts
+++ b/frontend/src/model/server-config.ts
@@ -135,7 +135,7 @@ export interface PushoverConfig {
}
export interface MessagingConfig {
- notify_on?: string[];
+ notify_on: string[];
telegram?: TelegramConfig;
rest?: RestConfig;
pushover?: PushoverConfig;
@@ -205,7 +205,7 @@ export interface StreamBufferConfig {
export interface StreamConfig {
retry: boolean,
buffer?: StreamBufferConfig,
- throttle?: number,
+ throttle?: string,
grace_period_millis?: number,
grace_period_timeout_secs?: number,
}
diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs
index d337a89ef..f66f6dd29 100644
--- a/src/api/api_utils.rs
+++ b/src/api/api_utils.rs
@@ -1,8 +1,9 @@
-use std::borrow::Cow;
+use crate::api::endpoints::xtream_api::{get_xtream_player_api_stream_url, XtreamApiStreamContext};
use crate::api::model::active_provider_manager::{ProviderAllocation, ProviderConfig, ProviderConnectionGuard};
use crate::api::model::app_state::AppState;
use crate::api::model::model_utils::get_stream_response_with_headers;
use crate::api::model::request::UserApiRequest;
+use crate::api::model::stream::{BoxedProviderStream, ProviderStreamInfo, ProviderStreamResponse};
use crate::api::model::stream_error::StreamError;
use crate::api::model::streams::active_client_stream::ActiveClientStream;
use crate::api::model::streams::persist_pipe_stream::PersistPipeStream;
@@ -10,32 +11,35 @@ use crate::api::model::streams::provider_stream;
use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, create_provider_connections_exhausted_stream, CustomVideoStreamType};
use crate::api::model::streams::provider_stream_factory::BufferStreamOptions;
use crate::api::model::streams::shared_stream_manager::SharedStreamManager;
-use crate::api::endpoints::xtream_api::{get_xtream_player_api_stream_url, XtreamApiStreamContext};
-use crate::api::model::stream::{BoxedProviderStream, ProviderStreamInfo, ProviderStreamResponse};
use crate::api::model::streams::throttled_stream::ThrottledStream;
-use crate::tools::atomic_once_flag::AtomicOnceFlag;
-use crate::utils::constants::{DASH_EXT, HLS_EXT};
-use crate::utils::default_utils::default_grace_period_millis;
use crate::auth::authenticator::Claims;
use crate::model::api_proxy::{ProxyUserCredentials, UserConnectionPermission};
use crate::model::config::{ConfigInput, ConfigTarget, InputFetchMethod, TargetType};
use crate::model::playlist::{PlaylistEntry, PlaylistItemType, XtreamCluster};
+use crate::tools::atomic_once_flag::AtomicOnceFlag;
use crate::tools::lru_cache::LRUResourceCache;
+use crate::utils::constants::{DASH_EXT, HLS_EXT};
+use crate::utils::crypto_utils::{deobfuscate_text, obfuscate_text};
+use crate::utils::default_utils::default_grace_period_millis;
use crate::utils::file::file_utils::create_new_file_for_write;
use crate::utils::network::request;
use crate::utils::network::request::{extract_extension_from_url, replace_url_extension, sanitize_sensitive_info};
+use crate::utils::size_utils::human_readable_byte_size;
use crate::utils::{debug_if_enabled, sys_utils, trace_if_enabled};
+use crate::BUILD_TIMESTAMP;
use axum::http::HeaderMap;
use axum::response::IntoResponse;
+use chrono::{DateTime, Utc};
use futures::{StreamExt, TryStreamExt};
use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation};
use log::{debug, error, log_enabled, trace};
use reqwest::StatusCode;
+use std::borrow::Cow;
use std::collections::HashMap;
use std::io::BufWriter;
use std::path::Path;
use std::sync::Arc;
-use chrono::{DateTime, Utc};
+use axum::body::Body;
use tokio::sync::Mutex;
use url::Url;
@@ -79,8 +83,6 @@ macro_rules! try_result_bad_request {
pub use try_option_bad_request;
pub use try_result_bad_request;
-use crate::BUILD_TIMESTAMP;
-use crate::utils::size_utils::human_readable_byte_size;
pub fn get_server_time() -> String {
chrono::offset::Local::now().with_timezone(&chrono::Local).format("%Y-%m-%d %H:%M:%S %Z").to_string()
@@ -250,10 +252,13 @@ impl StreamDetails {
/**
* If successfully a provider connection is used, do not forget to release if unsuccessfully
*/
-async fn get_streaming_options(app_state: &AppState, stream_url: &str, input_opt: Option<&ConfigInput>)
+async fn get_streaming_options(app_state: &AppState, stream_url: &str, input_opt: Option<&ConfigInput>, force_provider: Option<&str>)
-> (Option
, StreamingOption, Option>) {
if let Some(input) = input_opt {
- let provider_connection_guard = app_state.active_provider.acquire_connection(&input.name).await;
+ let provider_connection_guard = match force_provider {
+ Some(provider) => app_state.active_provider.force_exact_acquire_connection(provider).await,
+ None => app_state.active_provider.acquire_connection(&input.name).await
+ };
let stream_response_params = match &*provider_connection_guard {
ProviderAllocation::Exhausted => {
let stream = create_provider_connections_exhausted_stream(&app_state.config, &[]);
@@ -261,7 +266,9 @@ async fn get_streaming_options(app_state: &AppState, stream_url: &str, input_opt
}
ProviderAllocation::Available(ref provider)
| ProviderAllocation::GracePeriod(ref provider) => {
- let (provider, url) = if provider.id == input.id {
+ // force_stream_provider means we keep the url and the provider.
+ // If force_stream_provider or the input is the same as the config we dont need to get new url
+ let (provider, url) = if force_provider.is_some() || provider.id == input.id {
(input.name.to_string(), stream_url.to_string())
} else {
(provider.name.to_string(), get_stream_alternative_url(stream_url, input, provider))
@@ -292,9 +299,12 @@ fn get_grace_period_millis(connection_permission: &UserConnectionPermission, str
async fn create_stream_response_details(app_state: &AppState, stream_options: &StreamOptions, stream_url: &str,
req_headers: &HeaderMap, input_opt: Option<&ConfigInput>,
item_type: PlaylistItemType, share_stream: bool,
- connection_permission: UserConnectionPermission) -> StreamDetails {
- let (mut provider_connection_guard, stream_response_params, input_headers) = get_streaming_options(app_state, stream_url, input_opt).await;
- let config_grace_period_millis = app_state.config.reverse_proxy.as_ref().and_then(|r| r.stream.as_ref()).map_or_else(default_grace_period_millis, |s| s.grace_period_millis);
+ connection_permission: UserConnectionPermission,
+ force_provider: Option<&str>) -> StreamDetails {
+ let (mut provider_connection_guard, stream_response_params, input_headers) =
+ get_streaming_options(app_state, stream_url, input_opt, force_provider).await;
+ let config_grace_period_millis = app_state.config.reverse_proxy.as_ref()
+ .and_then(|r| r.stream.as_ref()).map_or_else(default_grace_period_millis, |s| s.grace_period_millis);
let grace_period_millis = get_grace_period_millis(&connection_permission, &stream_response_params, config_grace_period_millis);
match stream_response_params {
StreamingOption::Custom(provider_stream) => {
@@ -327,6 +337,7 @@ async fn create_stream_response_details(app_state: &AppState, stream_options: &S
// if we have no stream we should release the provider
if stream.is_none() {
drop(provider_connection_guard.take());
+ error!("Cant open stream {}", sanitize_sensitive_info(&request_url));
}
if log_enabled!(log::Level::Debug) {
@@ -461,6 +472,80 @@ where
None
}
+fn is_throttled_stream(item_type: PlaylistItemType, throttle_kbps: usize) -> bool {
+ throttle_kbps > 0 && matches!(item_type, PlaylistItemType::Video | PlaylistItemType::Series | PlaylistItemType::SeriesInfo)
+}
+
+const SESSION_COOKIE: &str = "m3uflt_session=";
+
+pub fn read_session_cookie(headers: &HeaderMap) -> Option {
+ if let Some(cookie_header) = headers.get(axum::http::header::COOKIE) {
+ let cookie_value = cookie_header.to_str().unwrap_or_default();
+ if let Some(cookie) = cookie_value.split(';').map(str::trim).find(|&part| part.starts_with(SESSION_COOKIE)).and_then(|p| p.strip_prefix(SESSION_COOKIE)) {
+ return Some(cookie.to_string());
+ }
+ }
+ None
+}
+
+fn get_session_cookie(cookie: &str) -> String {
+ format!("{SESSION_COOKIE}{cookie}; HttpOnly; Secure.")
+}
+
+/// # Panics
+pub async fn seek_stream_response(app_state: &AppState,
+ cookie: &str,
+ req_headers: &HeaderMap,
+ input: Option<&ConfigInput>,
+ item_type: PlaylistItemType,
+ _target: &ConfigTarget,
+ user: &ProxyUserCredentials) -> impl axum::response::IntoResponse + Send {
+ let stream_options = get_stream_options(app_state);
+ let share_stream = false;
+ let connection_permission = UserConnectionPermission::Allowed;
+
+ if let Ok(provider_and_stream_url) = deobfuscate_text(&app_state.config.t_encrypt_secret, cookie) {
+ let mut parts = provider_and_stream_url.splitn(2, '@');
+ let provider_name = parts.next().unwrap_or("");
+ let stream_url = parts.next().unwrap_or("");
+ if !provider_name.is_empty() && !stream_url.is_empty() {
+ let mut stream_details =
+ create_stream_response_details(app_state, &stream_options, stream_url, req_headers, input, item_type, share_stream, connection_permission.clone(), Some(provider_name)).await;
+
+ if stream_details.has_stream() {
+ let provider_response = stream_details.stream_info.as_ref().map(|(h, sc)| (h.clone(), *sc));
+ let stream = ActiveClientStream::new(stream_details, app_state, user, connection_permission).await;
+
+ let (status_code, header_map) = get_stream_response_with_headers(provider_response);
+ let mut response = axum::response::Response::builder()
+ .status(status_code);
+ for (key, value) in &header_map {
+ response = response.header(key, value);
+ }
+
+ response = response.header(axum::http::header::SET_COOKIE, get_session_cookie(cookie));
+
+ let body_stream = prepare_body_stream(app_state, item_type, stream);
+ debug_if_enabled!("Streaming partial stream request from {}", sanitize_sensitive_info(stream_url));
+ return response.body(body_stream).unwrap().into_response();
+ }
+ drop(stream_details.provider_connection_guard.take());
+ }
+ }
+ error!("Cant open partial stream for cookie {cookie}");
+ axum::http::StatusCode::BAD_REQUEST.into_response()
+}
+
+fn prepare_body_stream(app_state: &AppState, item_type: PlaylistItemType, stream: ActiveClientStream) -> Body {
+ let throttle_kbps = usize::try_from(get_stream_throttle(app_state)).unwrap_or_default();
+ let body_stream = if is_throttled_stream(item_type, throttle_kbps) {
+ axum::body::Body::from_stream(ThrottledStream::new(stream.boxed(), throttle_kbps))
+ } else {
+ axum::body::Body::from_stream(stream)
+ };
+ body_stream
+}
+
/// # Panics
#[allow(clippy::too_many_arguments)]
pub async fn stream_response(app_state: &AppState,
@@ -486,11 +571,12 @@ pub async fn stream_response(app_state: &AppState,
let stream_options = get_stream_options(app_state);
let mut stream_details =
- create_stream_response_details(app_state, &stream_options, stream_url, req_headers, input, item_type, share_stream, connection_permission.clone()).await;
-
+ create_stream_response_details(app_state, &stream_options, stream_url, req_headers, input, item_type, share_stream, connection_permission.clone(), None).await;
if stream_details.has_stream() {
// let content_length = get_stream_content_length(provider_response.as_ref());
let provider_response = stream_details.stream_info.as_ref().map(|(h, sc)| (h.clone(), *sc));
+ let provider_name = stream_details.provider_connection_guard.as_ref().and_then(ProviderConnectionGuard::get_provider_name);
+
let stream = ActiveClientStream::new(stream_details, app_state, user, connection_permission).await;
let stream_resp = if share_stream {
// Shared Stream response
@@ -515,12 +601,13 @@ pub async fn stream_response(app_state: &AppState,
response = response.header(key, value);
}
- let throttle_kbps = usize::try_from(get_stream_throttle(app_state)).unwrap_or_default();
- let body_stream = if throttle_kbps > 0 && matches!(item_type, PlaylistItemType::Video | PlaylistItemType::Series | PlaylistItemType::SeriesInfo) {
- axum::body::Body::from_stream(ThrottledStream::new(stream.boxed(), throttle_kbps as usize))
- } else {
- axum::body::Body::from_stream(stream)
- };
+ if let Some(provider) = provider_name {
+ if let Ok(cookie_value) = obfuscate_text(&app_state.config.t_encrypt_secret, &format!("{provider}@{stream_url}")) {
+ response = response.header(axum::http::header::SET_COOKIE, get_session_cookie(&cookie_value));
+ }
+ }
+
+ let body_stream = prepare_body_stream(app_state, item_type, stream);
response.body(body_stream).unwrap().into_response()
// if content_length > 0 { response_builder.body(SizedStream::new(content_length, stream)) } else { response_builder.streaming(stream) }
};
@@ -528,7 +615,6 @@ pub async fn stream_response(app_state: &AppState,
return stream_resp.into_response();
}
drop(stream_details.provider_connection_guard.take());
- error!("Cant open stream {}", sanitize_sensitive_info(stream_url));
axum::http::StatusCode::BAD_REQUEST.into_response()
}
@@ -683,3 +769,16 @@ pub fn redirect(url: &str) -> impl IntoResponse {
.body(axum::body::Body::empty())
.unwrap()
}
+
+pub fn is_seek_response(req_headers: &HeaderMap) -> Option {
+ if let Some(cookie) = read_session_cookie(req_headers) {
+ if let Some(maybe_value) = req_headers.get("range") {
+ if let Ok(value) = maybe_value.to_str() {
+ if !value.starts_with("bytes=0-") {
+ return Some(cookie);
+ }
+ }
+ }
+ }
+ None
+}
diff --git a/src/api/endpoints/hls_api.rs b/src/api/endpoints/hls_api.rs
index 83f1af619..9fda6f11e 100644
--- a/src/api/endpoints/hls_api.rs
+++ b/src/api/endpoints/hls_api.rs
@@ -77,7 +77,7 @@ async fn hls_api_stream(
return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response();
}
- let Ok(hls_url) = crypto_utils::decrypt_text(&app_state.config.t_encrypt_secret, ¶ms.token) else { return axum::http::StatusCode::BAD_REQUEST.into_response(); };
+ let Ok(hls_url) = crypto_utils::deobfuscate_text(&app_state.config.t_encrypt_secret, ¶ms.token) else { return axum::http::StatusCode::BAD_REQUEST.into_response(); };
let target_name = &target.name;
let virtual_id = params.stream_id;
diff --git a/src/api/endpoints/m3u_api.rs b/src/api/endpoints/m3u_api.rs
index 453c3cabb..ec6b8b55c 100644
--- a/src/api/endpoints/m3u_api.rs
+++ b/src/api/endpoints/m3u_api.rs
@@ -1,4 +1,4 @@
-use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, redirect, redirect_response, resource_response, separate_number_and_remainder, stream_response, try_option_bad_request, try_result_bad_request, RedirectParams};
+use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, is_seek_response, redirect, redirect_response, resource_response, seek_stream_response, separate_number_and_remainder, stream_response, try_option_bad_request, try_result_bad_request, RedirectParams};
use crate::api::endpoints::hls_api::handle_hls_stream_request;
use crate::api::model::app_state::AppState;
use crate::api::model::request::UserApiRequest;
@@ -73,10 +73,6 @@ async fn m3u_api_stream(
if user.permission_denied(&app_state) {
return axum::http::StatusCode::FORBIDDEN.into_response();
}
- let connection_permission = user.connection_permission(&app_state).await;
- if connection_permission == UserConnectionPermission::Exhausted {
- return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response();
- }
if !target.has_output(&TargetType::M3u) {
return axum::http::StatusCode::BAD_REQUEST.into_response();
@@ -92,6 +88,17 @@ async fn m3u_api_stream(
let input = app_state.config.get_input_by_name(m3u_item.input_name.as_str());
+
+ if let Some(cookie) = is_seek_response(&req_headers) {
+ // partial request means we are in reverse proxy mode, seek happened
+ return seek_stream_response(&app_state, &cookie, &req_headers, input, m3u_item.item_type, target, &user).await.into_response()
+ }
+
+ let connection_permission = user.connection_permission(&app_state).await;
+ if connection_permission == UserConnectionPermission::Exhausted {
+ return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response();
+ }
+
let cluster = XtreamCluster::try_from(m3u_item.item_type).unwrap_or(XtreamCluster::Live);
let context = XtreamApiStreamContext::try_from(cluster).unwrap_or(XtreamApiStreamContext::Live);
diff --git a/src/api/endpoints/xtream_api.rs b/src/api/endpoints/xtream_api.rs
index 785f3122a..4faabc3b0 100644
--- a/src/api/endpoints/xtream_api.rs
+++ b/src/api/endpoints/xtream_api.rs
@@ -1,7 +1,7 @@
// https://github.com/tellytv/go.xtream-codes/blob/master/structs.go
use crate::api::api_utils;
-use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, redirect_response, resource_response, separate_number_and_remainder, serve_file, stream_response, RedirectParams};
+use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, is_seek_response, redirect_response, resource_response, seek_stream_response, separate_number_and_remainder, serve_file, stream_response, RedirectParams};
use crate::api::api_utils::{redirect, try_option_bad_request, try_result_bad_request};
use crate::api::endpoints::hls_api::handle_hls_stream_request;
use crate::api::endpoints::xmltv_api::get_empty_epg_response;
@@ -185,10 +185,6 @@ async fn xtream_player_api_stream(
if user.permission_denied(app_state) {
return StatusCode::FORBIDDEN.into_response();
}
- let connection_permission = user.connection_permission(app_state).await;
- if connection_permission == UserConnectionPermission::Exhausted {
- return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response();
- }
let target_name = &target.name;
if !target.has_output(&TargetType::Xtream) {
@@ -196,11 +192,25 @@ async fn xtream_player_api_stream(
return StatusCode::BAD_REQUEST.into_response();
}
+
let (action_stream_id, stream_ext) = separate_number_and_remainder(stream_req.stream_id);
let virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse());
let (pli, mapping) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id));
let input = try_option_bad_request!(app_state.config.get_input_by_name(pli.input_name.as_str()), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context));
+
+ if let Some(cookie) = is_seek_response(req_headers) {
+ // partial request means we are in reverse proxy mode, seek happened
+ return seek_stream_response(app_state, &cookie, req_headers, Some(input), pli.item_type, target, &user).await.into_response()
+
+ }
+
+ let connection_permission = user.connection_permission(app_state).await;
+ if connection_permission == UserConnectionPermission::Exhausted {
+ return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response();
+ }
+
+
let redirect_params = RedirectParams {
item: &pli,
provider_id: Some(mapping.provider_id),
@@ -241,6 +251,7 @@ async fn xtream_player_api_stream(
stream_response(app_state, &stream_url, req_headers, Some(input), pli.item_type, target, &user, connection_permission).await.into_response()
}
+
async fn xtream_player_api_stream_with_token(
req_headers: &HeaderMap,
app_state: &Arc,
diff --git a/src/api/main_api.rs b/src/api/main_api.rs
index 0df5d5f8c..cd629da5e 100644
--- a/src/api/main_api.rs
+++ b/src/api/main_api.rs
@@ -71,10 +71,10 @@ async fn create_shared_data(cfg: &Arc) -> AppState {
let active_users = Arc::new(ActiveUserManager::new(cfg.log.as_ref().is_some_and(|l| l.log_active_user)));
let active_provider = Arc::new(ActiveProviderManager::new(cfg).await);
- let mut builder = Client::builder();
+ let mut builder = Client::builder().http1_only();
if cfg.connect_timeout_secs > 0 {
- builder = builder.connect_timeout(Duration::from_secs(u64::from(cfg.connect_timeout_secs)))
- };
+ builder = builder.connect_timeout(Duration::from_secs(u64::from(cfg.connect_timeout_secs)));
+ }
let client = builder.build().unwrap_or_else(|_| Client::new());
diff --git a/src/api/model/active_provider_manager.rs b/src/api/model/active_provider_manager.rs
index 7eea896e3..ed539c408 100644
--- a/src/api/model/active_provider_manager.rs
+++ b/src/api/model/active_provider_manager.rs
@@ -1,9 +1,9 @@
use crate::model::config::{Config, ConfigInput, ConfigInputAlias, InputType, InputUserInfo};
+use log::{debug, log_enabled};
use std::collections::HashMap;
use std::ops::Deref;
-use std::sync::{Arc};
use std::sync::atomic::{AtomicU16, AtomicUsize, Ordering};
-use log::{debug, log_enabled};
+use std::sync::Arc;
use tokio::sync::RwLock;
pub struct ProviderConnectionGuard {
@@ -11,6 +11,18 @@ pub struct ProviderConnectionGuard {
allocation: ProviderAllocation,
}
+impl ProviderConnectionGuard {
+ pub fn get_provider_name(&self) -> Option {
+ match self.allocation {
+ ProviderAllocation::Exhausted => None,
+ ProviderAllocation::Available(ref cfg) |
+ ProviderAllocation::GracePeriod(ref cfg) => {
+ Some(cfg.name.clone())
+ }
+ }
+ }
+}
+
impl Deref for ProviderConnectionGuard {
type Target = ProviderAllocation;
fn deref(&self) -> &Self::Target {
@@ -22,7 +34,7 @@ impl Drop for ProviderConnectionGuard {
fn drop(&mut self) {
match &self.allocation {
ProviderAllocation::Exhausted => {}
- ProviderAllocation::Available(config)|
+ ProviderAllocation::Available(config) |
ProviderAllocation::GracePeriod(config) => {
let manager = self.manager.clone();
let provider_config = Arc::clone(config);
@@ -116,20 +128,24 @@ impl ProviderConfig {
self.current_connections.fetch_add(1, Ordering::SeqCst);
return 1;
}
- if (!grace && connections < self.max_connections) || (grace && connections <= self.max_connections) {
+ if (!grace && connections < self.max_connections) || (grace && connections <= self.max_connections) {
self.current_connections.fetch_add(1, Ordering::SeqCst);
return if connections < self.max_connections { 1 } else { 2 };
}
3
}
+ fn force_allocate(&self) {
+ self.current_connections.fetch_add(1, Ordering::SeqCst);
+ }
+
// is intended to use with redirects, to cycle through provider
fn get_next(&self, grace: bool) -> bool {
let connections = self.current_connections.load(Ordering::SeqCst);
if self.max_connections == 0 {
return true;
}
- if (!grace && connections < self.max_connections) || (grace && connections <= self.max_connections) {
+ if (!grace && connections < self.max_connections) || (grace && connections <= self.max_connections) {
return true;
}
false
@@ -149,7 +165,7 @@ impl ProviderConfig {
#[derive(Clone, Debug)]
struct ProviderConfigWrapper {
- inner: Arc
+ inner: Arc,
}
@@ -160,6 +176,11 @@ impl ProviderConfigWrapper {
}
}
+ pub fn force_allocate(&self) -> ProviderAllocation {
+ self.inner.force_allocate();
+ ProviderAllocation::Available(Arc::clone(&self.inner))
+ }
+
pub fn try_allocate(&self, grace: bool) -> ProviderAllocation {
match self.inner.try_allocate(grace) {
1 => ProviderAllocation::Available(Arc::clone(&self.inner)),
@@ -194,7 +215,6 @@ enum ProviderLineup {
}
impl ProviderLineup {
-
fn get_next(&self) -> Option> {
match self {
ProviderLineup::Single(lineup) => lineup.get_next(),
@@ -533,7 +553,7 @@ impl ActiveProviderManager {
self.providers.write().await.push(lineup);
}
- fn get_provider_config<'a>(name: &str, providers: &'a Vec) -> Option<(&'a ProviderLineup, &'a ProviderConfig)> {
+ fn get_provider_config<'a>(name: &str, providers: &'a Vec) -> Option<(&'a ProviderLineup, &'a ProviderConfigWrapper)> {
for lineup in providers {
match lineup {
ProviderLineup::Single(single) => {
@@ -562,10 +582,24 @@ impl ActiveProviderManager {
None
}
+ pub async fn force_exact_acquire_connection(&self, input_name: &str) -> ProviderConnectionGuard {
+ let providers = self.providers.read().await;
+ let allocation = match Self::get_provider_config(input_name, &providers) {
+ None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider
+ Some((_lineup, config)) => config.force_allocate(),
+ };
+
+ ProviderConnectionGuard {
+ manager: Arc::new(self.clone_inner()),
+ allocation,
+ }
+ }
+
+ // Returns the next available provider connection
pub async fn acquire_connection(&self, input_name: &str) -> ProviderConnectionGuard {
let providers = self.providers.read().await;
let allocation = match Self::get_provider_config(input_name, &providers) {
- None => ProviderAllocation::Exhausted,
+ None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider
Some((lineup, _config)) => lineup.acquire()
};
@@ -661,9 +695,9 @@ impl ActiveProviderManager {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::model::config::InputFetchMethod;
use crate::Arc;
use std::thread;
- use crate::model::config::InputFetchMethod;
macro_rules! should_available {
($lineup:expr, $provider_id:expr) => {
@@ -796,7 +830,6 @@ mod tests {
should_grace_period!(lineup, 2);
should_exhausted!(lineup);
-
}
@@ -909,9 +942,9 @@ mod tests {
let mut handles = vec![];
- let available_count= Arc::new(AtomicU16::new(2));
- let grace_period_count= Arc::new(AtomicU16::new(1));
- let exhausted_count= Arc::new(AtomicU16::new(2));
+ let available_count = Arc::new(AtomicU16::new(2));
+ let grace_period_count = Arc::new(AtomicU16::new(1));
+ let exhausted_count = Arc::new(AtomicU16::new(2));
for _ in 0..5 {
let lineup_clone = Arc::clone(&lineup);
@@ -925,7 +958,6 @@ mod tests {
ProviderAllocation::Available(_) => available.fetch_sub(1, Ordering::SeqCst),
ProviderAllocation::GracePeriod(_) => grace_period.fetch_sub(1, Ordering::SeqCst),
}
-
});
handles.push(handle);
}
diff --git a/src/api/model/streams/provider_stream_factory.rs b/src/api/model/streams/provider_stream_factory.rs
index d9fc59787..e3eecc5d4 100644
--- a/src/api/model/streams/provider_stream_factory.rs
+++ b/src/api/model/streams/provider_stream_factory.rs
@@ -222,6 +222,7 @@ async fn provider_initial_request(cfg: &Config, request_client: Arc = get_response_headers(response.headers());
+ //let url = stream_options.get_url();
// debug!("First headers {headers:?} {} {}", sanitize_sensitive_info(url.as_str()));
Some((response_headers, response.status()))
};
diff --git a/src/model/api_proxy.rs b/src/model/api_proxy.rs
index c576f0a7d..7b9364fca 100644
--- a/src/model/api_proxy.rs
+++ b/src/model/api_proxy.rs
@@ -185,6 +185,7 @@ impl FromStr for ProxyUserStatus {
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
pub struct ProxyUserCredentials {
pub username: String,
pub password: String,
@@ -363,6 +364,7 @@ impl ApiProxyServerInfo {
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
pub struct ApiProxyConfig {
pub server: Vec,
pub user: Vec,
diff --git a/src/model/config.rs b/src/model/config.rs
index 479b94e2a..039e8f285 100644
--- a/src/model/config.rs
+++ b/src/model/config.rs
@@ -28,7 +28,7 @@ use crate::messaging::MsgKind;
use crate::model::api_proxy::{ApiProxyConfig, ApiProxyServerInfo, ProxyUserCredentials};
use crate::model::mapping::Mapping;
use crate::model::mapping::Mappings;
-use crate::utils::default_utils::{default_as_default, default_as_true, default_as_two_u16, default_grace_period_millis, default_grace_period_timeout_secs};
+use crate::utils::default_utils::{default_as_default, default_as_true, default_resolve_delay_secs, default_grace_period_millis, default_grace_period_timeout_secs, default_connect_timeout_secs};
use crate::utils::file::file_lock_manager::FileLockManager;
use crate::utils::file::file_utils;
use crate::utils::file::file_utils::file_reader;
@@ -335,11 +335,11 @@ pub struct XtreamTargetOutput {
pub skip_series_direct_source: bool,
#[serde(default)]
pub resolve_series: bool,
- #[serde(default = "default_as_two_u16")]
+ #[serde(default = "default_resolve_delay_secs")]
pub resolve_series_delay: u16,
#[serde(default)]
pub resolve_vod: bool,
- #[serde(default = "default_as_two_u16")]
+ #[serde(default = "default_resolve_delay_secs")]
pub resolve_vod_delay: u16,
}
@@ -1330,6 +1330,7 @@ pub struct LogConfig {
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
+#[serde(deny_unknown_fields)]
pub struct LogLevelConfig {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub log: Option,
@@ -1737,7 +1738,7 @@ pub struct Config {
pub log: Option,
#[serde(default)]
pub user_access_control: bool,
- #[serde(default)]
+ #[serde(default = "default_connect_timeout_secs")]
pub connect_timeout_secs: u32,
#[serde(default)]
pub update_on_boot: bool,
diff --git a/src/processing/parser/hls.rs b/src/processing/parser/hls.rs
index 74be4da8b..955816f81 100644
--- a/src/processing/parser/hls.rs
+++ b/src/processing/parser/hls.rs
@@ -1,5 +1,5 @@
use crate::model::api_proxy::ProxyUserCredentials;
-use crate::utils::crypto_utils::encrypt_text;
+use crate::utils::crypto_utils::{obfuscate_text};
use std::str;
use crate::utils::constants::{CONSTANTS, HLS_PREFIX};
@@ -31,7 +31,7 @@ fn rewrite_hls_url(input: &str, replacement: &str) -> String {
fn rewrite_uri_attrib(line: &str, props: &RewriteHlsProps) -> String {
if let Some(caps) = CONSTANTS.re_memory_usage.captures(line) {
let uri = &caps[1];
- if let Ok(encrypted_uri) = encrypt_text(props.secret, &rewrite_hls_url(&props.hls_url, uri)) {
+ if let Ok(encrypted_uri) = obfuscate_text(props.secret, &rewrite_hls_url(&props.hls_url, uri)) {
return CONSTANTS.re_hls_uri.replace(line, format!(r#"URI="{encrypted_uri}""#)).to_string();
}
}
@@ -46,9 +46,9 @@ pub fn rewrite_hls(user: &ProxyUserCredentials, props: &RewriteHlsProps) -> Stri
if line.starts_with('#') {
result.push(rewrite_uri_attrib(line, props));
} else if let Ok(token) = if line.starts_with("http") {
- encrypt_text(props.secret, line)
+ obfuscate_text(props.secret, line)
} else {
- encrypt_text(props.secret, &rewrite_hls_url(&props.hls_url, line))
+ obfuscate_text(props.secret, &rewrite_hls_url(&props.hls_url, line))
} {
result.push(format!("{}/{HLS_PREFIX}/{username}/{password}/{}/{}/{token}", props.base_url, props.input_id, props.virtual_id));
}
diff --git a/src/utils/crypto_utils.rs b/src/utils/crypto_utils.rs
index 700d0f0e0..b1d231437 100644
--- a/src/utils/crypto_utils.rs
+++ b/src/utils/crypto_utils.rs
@@ -1,9 +1,38 @@
-use openssl::symm::{Cipher, Crypter, Mode};
-use base64::{engine::general_purpose, Engine as _};
-use rand::Rng;
use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind};
+use base64::{engine::general_purpose, Engine as _};
+use openssl::symm::{Cipher, Crypter, Mode};
+use rand::Rng;
-pub fn encrypt_text(secret: &[u8;16], text: &str) -> Result {
+
+fn encode_base64_string(input: &[u8]) -> String {
+ general_purpose::URL_SAFE_NO_PAD.encode(input)
+}
+
+fn decode_base64_string(input: &str) -> Vec {
+ general_purpose::URL_SAFE_NO_PAD.decode(input).unwrap_or_else(|_| input.as_bytes().to_vec())
+}
+
+pub fn xor_bytes(secret: &[u8], data: &[u8]) -> Vec {
+ data.iter()
+ .enumerate()
+ .map(|(i, &b)| b ^ secret[i % secret.len()])
+ .collect()
+}
+
+pub fn obfuscate_text(secret: &[u8], text: &str) -> Result {
+ Ok(encode_base64_string(&xor_bytes(secret, text.as_bytes())))
+}
+
+pub fn deobfuscate_text(secret: &[u8], text: &str) -> Result {
+ let data = xor_bytes(secret, &decode_base64_string(text));
+ if let Ok(result) = String::from_utf8(data) {
+ Ok(result)
+ } else {
+ Err(text.to_string())
+ }
+}
+
+pub fn encrypt_text(secret: &[u8; 16], text: &str) -> Result {
let iv: [u8; 16] = rand::rng().random(); // Random IV (AES-CBC 16 Bytes)
let cipher = Cipher::aes_128_cbc();
let mut crypter = Crypter::new(cipher, Mode::Encrypt, secret, Some(&iv)).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Info, err.to_string()))?;
@@ -18,7 +47,7 @@ pub fn encrypt_text(secret: &[u8;16], text: &str) -> Result Result {
+pub fn decrypt_text(secret: &[u8; 16], encrypted_text: &str) -> Result {
let data = general_purpose::URL_SAFE_NO_PAD.decode(encrypted_text).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Info, err.to_string()))?;
let (iv, ciphertext) = data.split_at(16); // first 16 bytes IV
let cipher = Cipher::aes_128_cbc();
@@ -33,8 +62,8 @@ pub fn decrypt_text(secret: &[u8;16], encrypted_text: &str) -> Result bool { true }
pub fn default_as_default() -> String { String::from("default") }
-pub const fn default_as_two_u16() -> u16 { 2 }
+// Default delay values for resolving VOD or Series requests,
+// used to prevent frequent requests that could trigger a provider ban.
+pub const fn default_resolve_delay_secs() -> u16 { 2 }
-pub const fn default_grace_period_millis() -> u64 { 0 }
-pub const fn default_grace_period_timeout_secs() -> u64 { 0 }
\ No newline at end of file
+// Default grace values to accommodate rapid channel changes and seek requests,
+// helping avoid triggering hard max_connection enforcement.
+pub const fn default_grace_period_millis() -> u64 { 2000 }
+pub const fn default_grace_period_timeout_secs() -> u64 { 5 }
+pub const fn default_connect_timeout_secs() -> u32 { 10 }
\ No newline at end of file
diff --git a/src/utils/size_utils.rs b/src/utils/size_utils.rs
index f81b74ef5..ca85621ef 100644
--- a/src/utils/size_utils.rs
+++ b/src/utils/size_utils.rs
@@ -80,6 +80,9 @@ pub fn parse_to_kbps(input: &str) -> Result {
];
let speed_str = input.trim();
+ if speed_str.is_empty() {
+ return Ok(0);
+ }
for (unit, multiplier) in units {
if let Some(speed_unit) = speed_str.strip_suffix(unit) {
let number_part = speed_unit.trim();