diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index b1583a9a2..83d19d887 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -1,6 +1,5 @@ -use crate::api::endpoints::xtream_api::{get_xtream_player_api_stream_url, XtreamApiStreamContext}; +use crate::api::endpoints::xtream_api::{get_xtream_player_api_stream_url, ApiStreamContext}; use crate::api::model::active_provider_manager::{ProviderAllocation, ProviderConnectionGuard}; -use crate::tuliprox_error::{info_err}; use crate::api::model::app_state::AppState; use crate::api::model::model_utils::{ get_stream_response_with_headers}; use crate::api::model::request::UserApiRequest; @@ -27,7 +26,7 @@ use crate::utils::human_readable_byte_size; use crate::utils::{debug_if_enabled, trace_if_enabled}; use crate::{BUILD_TIMESTAMP}; use axum::body::Body; -use axum::http::{HeaderMap, HeaderValue}; +use axum::http::{HeaderMap}; use axum::response::IntoResponse; use chrono::{DateTime, Utc}; use futures::{StreamExt, TryStreamExt}; @@ -84,7 +83,6 @@ pub use try_option_bad_request; pub use try_result_bad_request; use crate::api::model::active_user_manager::UserSession; use crate::api::model::provider_config::ProviderConfig; -use crate::utils::base64_to_u32; pub fn get_server_time() -> String { chrono::offset::Local::now().with_timezone(&chrono::Local).format("%Y-%m-%d %H:%M:%S %Z").to_string() @@ -391,10 +389,10 @@ async fn create_stream_response_details(app_state: &AppState, } if log_enabled!(log::Level::Debug) { - if let Some((headers, status_code)) = stream_info.as_ref() { + if let Some((headers, status_code, response_url)) = stream_info.as_ref() { debug!( "Responding stream request {} with status {}, headers {:?}", - sanitize_sensitive_info(&request_url), + sanitize_sensitive_info(response_url.as_ref().map_or(stream_url, |s| s.as_str())), status_code, headers ); @@ -425,7 +423,7 @@ where pub input: &'a ConfigInput, pub user: &'a ProxyUserCredentials, pub stream_ext: Option<&'a str>, - pub req_context: XtreamApiStreamContext, + pub req_context: ApiStreamContext, pub action_path: &'a str, } @@ -516,33 +514,7 @@ where } 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_NAME: &str = "tuliprox_session="; - -// fn create_delete_session_cookie() -> String { -// format!("{SESSION_COOKIE_NAME}; Max-Age=0; Path=/; HttpOnly") -// } - -pub fn create_session_cookie(token: u32) -> String { - let cookie = crate::utils::u32_to_base64(token); - format!("{SESSION_COOKIE_NAME}{cookie}; Path=/; HttpOnly; SameSite=None") -} - -pub fn read_session_token(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_NAME)) - .and_then(|p| { - p.strip_prefix(SESSION_COOKIE_NAME).map(str::trim) - }) { - return base64_to_u32(cookie); - } - } - None + throttle_kbps > 0 && matches!(item_type, PlaylistItemType::Video | PlaylistItemType::Series | PlaylistItemType::SeriesInfo | PlaylistItemType::Catchup) } fn prepare_body_stream(app_state: &AppState, item_type: PlaylistItemType, stream: ActiveClientStream) -> Body { @@ -555,7 +527,6 @@ fn prepare_body_stream(app_state: &AppState, item_type: PlaylistItemType, stream body_stream } - /// # Panics pub async fn force_provider_stream_response(app_state: &AppState, user_session: &UserSession, @@ -571,10 +542,10 @@ pub async fn force_provider_stream_response(app_state: &AppState, create_stream_response_details(app_state, &stream_options, &user_session.stream_url, req_headers, input, item_type, share_stream, connection_permission, Some(&user_session.provider)).await; if stream_details.has_stream() { - let provider_response = stream_details.stream_info.as_ref().map(|(h, sc)| (h.clone(), *sc)); + let provider_response = stream_details.stream_info.as_ref().map(|(h, sc,url)| (h.clone(), *sc, url.clone())); 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 (status_code, header_map) = get_stream_response_with_headers(provider_response.map(|(h,s,_)| (h, s))); let mut response = axum::response::Response::builder().status(status_code); for (key, value) in &header_map { response = response.header(key, value); @@ -598,6 +569,7 @@ pub async fn force_provider_stream_response(app_state: &AppState, /// # Panics #[allow(clippy::too_many_arguments)] pub async fn stream_response(app_state: &AppState, + session_token: u32, virtual_id: u32, item_type: PlaylistItemType, stream_url: &str, @@ -624,17 +596,17 @@ pub async fn stream_response(app_state: &AppState, create_stream_response_details(app_state, &stream_options, stream_url, req_headers, input, item_type, share_stream, connection_permission, 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_response = stream_details.stream_info.as_ref().map(|(h, sc, response_url)| (h.clone(), *sc, response_url.clone())); 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 { debug_if_enabled!("Streaming shared stream request from {}", sanitize_sensitive_info(stream_url)); // Shared Stream response - let shared_headers = provider_response.as_ref().map_or_else(Vec::new, |(h, _)| h.clone()); + let shared_headers = provider_response.as_ref().map_or_else(Vec::new, |(h, _, _)| h.clone()); SharedStreamManager::subscribe(app_state, stream_url, stream, shared_headers, stream_options.buffer_size).await; if let Some(broadcast_stream) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url).await { - let (status_code, header_map) = get_stream_response_with_headers(provider_response); + let (status_code, header_map) = get_stream_response_with_headers(provider_response.map(|(h,s,_)| (h, s))); let mut response = axum::response::Response::builder() .status(status_code); for (key, value) in &header_map { @@ -645,17 +617,18 @@ pub async fn stream_response(app_state: &AppState, axum::http::StatusCode::BAD_REQUEST.into_response() } } else { + let session_url = provider_response.as_ref().and_then(|(_,_,u)| u.as_ref()).map_or_else(|| Cow::Borrowed(stream_url), |url| Cow::Owned(url.to_string())); debug_if_enabled!("Streaming stream request from {}", sanitize_sensitive_info(stream_url)); - let (status_code, header_map) = get_stream_response_with_headers(provider_response); + let (status_code, header_map) = get_stream_response_with_headers(provider_response.map(|(h,s,_)| (h, s))); let mut response = axum::response::Response::builder().status(status_code); for (key, value) in &header_map { response = response.header(key, value); } if let Some(provider) = provider_name { - if matches!(item_type, PlaylistItemType::LiveHls | PlaylistItemType::LiveDash | PlaylistItemType::Video | PlaylistItemType::Series) { - if let Some(token) = app_state.active_users.create_user_session(user, virtual_id, &provider, stream_url, connection_permission).await { - response = response.header(axum::http::header::SET_COOKIE, create_session_cookie(token)); + if matches!(item_type, PlaylistItemType::LiveHls | PlaylistItemType::LiveDash | PlaylistItemType::Video | PlaylistItemType::Series | PlaylistItemType::Catchup) { + if let Some(token) = app_state.active_users.create_user_session(user, session_token, virtual_id, &provider, &session_url, connection_permission).await { + debug!("Creating session for user {} with token {token} {session_url}", user.username); } } } diff --git a/src/api/endpoints/api_playlist_utils.rs b/src/api/endpoints/api_playlist_utils.rs index 20744c540..5dab9e45e 100644 --- a/src/api/endpoints/api_playlist_utils.rs +++ b/src/api/endpoints/api_playlist_utils.rs @@ -67,13 +67,13 @@ fn group_playlist_items_by_cluster(params: Option<(utils::FileReadGuard, for (item, _) in iter { match item.item_type { PlaylistItemType::Live - | PlaylistItemType::Catchup | PlaylistItemType::LiveUnknown | PlaylistItemType::LiveHls | PlaylistItemType::LiveDash => { live.push(item); } - PlaylistItemType::Video => { + PlaylistItemType::Catchup + | PlaylistItemType::Video => { video.push(item); } PlaylistItemType::Series diff --git a/src/api/endpoints/hls_api.rs b/src/api/endpoints/hls_api.rs index 8d6a3b109..a8f1496e8 100644 --- a/src/api/endpoints/hls_api.rs +++ b/src/api/endpoints/hls_api.rs @@ -1,4 +1,5 @@ -use crate::api::api_utils::{create_session_cookie, force_provider_stream_response, get_stream_alternative_url, is_seek_request, read_session_token}; +use rand::RngCore; +use crate::api::api_utils::{force_provider_stream_response, get_stream_alternative_url, is_seek_request}; use crate::api::api_utils::{try_option_bad_request}; use crate::api::model::app_state::AppState; use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType}; @@ -60,7 +61,9 @@ pub(in crate::api) async fn handle_hls_stream_request(app_state: &Arc, match app_state.active_provider.get_next_provider(&input.name).await { Some(provider_cfg) => { let stream_url = get_stream_alternative_url(&url, input, &provider_cfg); - let session_token= app_state.active_users.create_user_session(user, virtual_id, &provider_cfg.name, &stream_url, connection_permission).await; + // TODO avoid same tokens twice, this is a little bit hacky + let user_session_token = rand::rng().next_u32(); + let session_token= app_state.active_users.create_user_session(user, user_session_token, virtual_id, &provider_cfg.name, &stream_url, connection_permission).await; (stream_url, session_token) }, None => (url, None), @@ -80,7 +83,7 @@ pub(in crate::api) async fn handle_hls_stream_request(app_state: &Arc, user_token: session_token, }; let hls_content = rewrite_hls(user, &rewrite_hls_props); - hls_response(hls_content, session_token.map(create_session_cookie)).into_response() + hls_response(hls_content, session_token.map(|t: u32| std::string::ToString::to_string(&t))).into_response() } Err(err) => { error!("Failed to download m3u8 {}", sanitize_sensitive_info(err.to_string().as_str())); @@ -105,9 +108,12 @@ async fn hls_api_stream( let virtual_id = params.stream_id; let input = try_option_bad_request!(app_state.config.get_input_by_id(params.input_id), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live)); - let mut user_session = match read_session_token(&req_headers) { - None => None, - Some(token) => app_state.active_users.get_user_session(&user.username, token).await, + // TODO SESSION TOKEN GENRATION !!! + let user_session_token = 0; + let mut user_session = if user_session_token == 0 { + None + } else { + app_state.active_users.get_user_session(&user.username, user_session_token).await }; if let Some(session) = &mut user_session { diff --git a/src/api/endpoints/m3u_api.rs b/src/api/endpoints/m3u_api.rs index ff9e52910..647860a1d 100644 --- a/src/api/endpoints/m3u_api.rs +++ b/src/api/endpoints/m3u_api.rs @@ -1,22 +1,23 @@ -use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, is_seek_request, redirect, redirect_response, resource_response, force_provider_stream_response, separate_number_and_remainder, stream_response, try_option_bad_request, try_result_bad_request, RedirectParams, read_session_token}; +use crate::api::api_utils::{force_provider_stream_response, get_user_target, get_user_target_by_credentials, is_seek_request, redirect, redirect_response, resource_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::endpoints::xtream_api::{ApiStreamContext, ApiStreamRequest}; use crate::api::model::app_state::AppState; use crate::api::model::request::UserApiRequest; -use crate::model::{UserConnectionPermission}; -use crate::model::{TargetType}; +use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType}; +use crate::model::TargetType; +use crate::model::UserConnectionPermission; use crate::model::{FieldGetAccessor, PlaylistEntry, PlaylistItemType, XtreamCluster}; use crate::repository::m3u_repository::{m3u_get_item_for_stream_id, m3u_load_rewrite_playlist}; +use crate::repository::storage_const; use crate::utils::request::{extract_extension_from_url, sanitize_sensitive_info}; +use crate::utils::HLS_EXT; +use axum::http::{HeaderMap, StatusCode}; use axum::response::IntoResponse; use bytes::Bytes; use futures::stream; use log::{debug, error}; +use rand::RngCore; use std::sync::Arc; -use axum::http::StatusCode; -use crate::api::endpoints::xtream_api::XtreamApiStreamContext; -use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType}; -use crate::repository::storage_const; -use crate::utils::{HLS_EXT}; async fn m3u_api( api_req: &UserApiRequest, @@ -62,13 +63,14 @@ async fn m3u_api_post( } async fn m3u_api_stream( - req_headers: axum::http::HeaderMap, - axum::extract::Query(api_req): axum::extract::Query, - axum::extract::Path((username, password, stream_id)): axum::extract::Path<(String, String, String)>, - axum::extract::State(app_state): axum::extract::State>, + req_headers: &HeaderMap, + api_req: &UserApiRequest, + app_state: &Arc, + stream_req: ApiStreamRequest<'_>, + // _addr: &std::net::SocketAddr, ) -> impl axum::response::IntoResponse + Send { - let (user, target) = try_option_bad_request!(get_user_target_by_credentials(&username, &password, &api_req, &app_state), false, format!("Could not find any user {username}")); - if user.permission_denied(&app_state) { + let (user, target) = try_option_bad_request!(get_user_target_by_credentials(stream_req.username, stream_req.password, api_req, app_state), false, format!("Could not find any user {}", stream_req.username)); + if user.permission_denied(app_state) { return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserAccountExpired).into_response(); } @@ -78,19 +80,19 @@ async fn m3u_api_stream( return StatusCode::BAD_REQUEST.into_response(); } - let (action_stream_id, stream_ext) = separate_number_and_remainder(&stream_id); + 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 = try_result_bad_request!(m3u_get_item_for_stream_id(virtual_id, &app_state.config, target).await, true, format!("Failed to read m3u 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}, stream_id {virtual_id}")); - let cluster = XtreamCluster::try_from(pli.item_type).unwrap_or(XtreamCluster::Live); - let user_session = match read_session_token(&req_headers) { - None => None, - Some(token) => app_state.active_users.get_user_session(&user.username, token).await, + let user_session = if api_req.session == 0 { + None + } else { + app_state.active_users.get_user_session(&user.username, api_req.session).await }; - if let Some(session) = &user_session { + if let Some(session) = &user_session { if session.permission == UserConnectionPermission::Exhausted { return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } @@ -98,9 +100,9 @@ async fn m3u_api_stream( if app_state.active_provider.is_over_limit(&session.provider).await { return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response(); } - if session.virtual_id == virtual_id && is_seek_request(cluster, &req_headers).await { + if session.virtual_id == virtual_id && is_seek_request(cluster, &req_headers).await { // partial request means we are in reverse proxy mode, seek happened - return force_provider_stream_response(&app_state, session, pli.item_type, &req_headers, input, &user).await.into_response() + return force_provider_stream_response(&app_state, session, pli.item_type, &req_headers, input, &user).await.into_response(); } } @@ -109,7 +111,7 @@ async fn m3u_api_stream( return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } - let context = XtreamApiStreamContext::try_from(cluster).unwrap_or(XtreamApiStreamContext::Live); + let context = ApiStreamContext::try_from(cluster).unwrap_or(ApiStreamContext::Live); let redirect_params = RedirectParams { item: &pli, @@ -121,7 +123,7 @@ async fn m3u_api_stream( user: &user, stream_ext: stream_ext.as_deref(), req_context: context, - action_path: "" // TODO is there timeshoft or something like that ? + action_path: "", // TODO is there timeshoft or something like that ? }; if let Some(response) = redirect_response(&app_state, &redirect_params).await { @@ -137,7 +139,7 @@ async fn m3u_api_stream( return handle_hls_stream_request(&app_state, &user, user_session.as_ref(), &pli.url, pli.virtual_id, input, connection_permission).await.into_response(); } - stream_response(&app_state, pli.virtual_id, pli.item_type, pli.url.as_str(), &req_headers, input, target, &user, connection_permission).await.into_response() + stream_response(&app_state, api_req.session, pli.virtual_id, pli.item_type, pli.url.as_str(), &req_headers, input, target, &user, connection_permission).await.into_response() } async fn m3u_api_resource( @@ -170,7 +172,7 @@ async fn m3u_api_resource( match stream_url { None => axum::http::StatusCode::NOT_FOUND.into_response(), Some(url) => { - if user.proxy.is_redirect(m3u_item.item_type) || target.is_force_redirect(m3u_item.item_type) { + if user.proxy.is_redirect(m3u_item.item_type) || target.is_force_redirect(m3u_item.item_type) { debug!("Redirecting stream request to {}", sanitize_sensitive_info(&url)); redirect(&url).into_response() } else { @@ -180,11 +182,70 @@ async fn m3u_api_resource( } } -macro_rules! register_m3u_stream_routes { - ($router:expr, [$($path:expr),*]) => {{ - $router - $( - .route(&format!("/{}/{}/{{username}}/{{password}}/{{stream_id}}", storage_const::M3U_STREAM_PATH, $path), axum::routing::get(m3u_api_stream)) + +async fn m3u_api_stream_with_session( + // addr: &SocketAddr, + app_state: &Arc, + api_req: &UserApiRequest, + stream_req: ApiStreamRequest<'_>, + req_headers: &HeaderMap, +) -> impl IntoResponse + Send { + if api_req.session > 0 { + return m3u_api_stream(&req_headers, &api_req, &app_state, stream_req/*, &addr*/).await.into_response() + } + if let Some(credentials) = app_state.config.get_user_credentials(stream_req.username) { + // create new session and redirect + let context = match stream_req.context { + ApiStreamContext::Live => "live", + ApiStreamContext::Movie => "movie", + ApiStreamContext::Series => "series", + _ => "", + }; + + let username = stream_req.username; + let password = stream_req.password; + let stream_id = stream_req.stream_id; + + // TODO avoid same tokens twice, this is a little bit hacky + let session_token = rand::rng().next_u32(); + let server_info = app_state.config.get_user_server_info(&credentials); + let redirect_url = format!("{}/{}/{context}/{username}/{password}/{stream_id}?session={session_token}", + storage_const::M3U_STREAM_PATH, server_info.get_base_url()); + return redirect(&redirect_url).into_response(); + } + StatusCode::BAD_REQUEST.into_response() +} +macro_rules! create_m3u_api_stream { + ($fn_name:ident, $context:expr) => { + async fn $fn_name( + req_headers: axum::http::HeaderMap, + axum::extract::Query(api_req): axum::extract::Query, + axum::extract::Path((username, password, stream_id)): axum::extract::Path<(String, String, String)>, + axum::extract::State(app_state): axum::extract::State>, + // axum::extract::ConnectInfo(addr): axum::extract::ConnectInfo, + ) -> impl IntoResponse + Send { + m3u_api_stream_with_session( + // &addr, + &app_state, + &api_req, + ApiStreamRequest::from($context, &username, &password, &stream_id, ""), + &req_headers, + ).await.into_response() + } + } +} + +create_m3u_api_stream!(m3u_api_live_stream_alt, ApiStreamContext::LiveAlt); +create_m3u_api_stream!(m3u_api_live_stream, ApiStreamContext::Live); +create_m3u_api_stream!(m3u_api_series_stream, ApiStreamContext::Series); +create_m3u_api_stream!(m3u_api_movie_stream, ApiStreamContext::Movie); + + +macro_rules! register_m3u_api_stream { + ($router:expr, [$(($path:expr, $fn_name:ident)),*]) => {{ + $router + $( + .route(&format!("/{}/{{username}}/{{password}}/{{stream_id}}", $path), axum::routing::get($fn_name)) // $cfg.service(web::resource(format!("/{M3U_STREAM_PATH}/{}/{{username}}/{{password}}/{{stream_id}}", $path)).route(web::get().to(m3u_api_stream))); )* }}; @@ -204,7 +265,12 @@ macro_rules! register_m3u_api_routes { pub fn m3u_api_register() -> axum::Router> { let mut router = axum::Router::new(); router = register_m3u_api_routes!(router, ["get.php", "apiget", "m3u"]); - register_m3u_stream_routes!(router, ["live", "movie", "series"]) - .route(&format!("/{}/{{username}}/{{password}}/{{stream_id}}", storage_const::M3U_STREAM_PATH), axum::routing::get(m3u_api_stream)) - .route(&format!("/{}/{{username}}/{{password}}/{{stream_id}}/{{resource}}", storage_const::M3U_RESOURCE_PATH), axum::routing::get(m3u_api_resource)) + router = register_m3u_api_stream!(router, [ + (storage_const::M3U_STREAM_PATH, m3u_api_live_stream_alt), + (format!("{}/live", storage_const::M3U_STREAM_PATH), m3u_api_live_stream), + (format!("{}/movie", storage_const::M3U_STREAM_PATH), m3u_api_movie_stream), + (format!("{}/series", storage_const::M3U_STREAM_PATH), m3u_api_series_stream)]); + + router + .route(&format!("/{}/{{username}}/{{password}}/{{stream_id}}/{{resource}}", storage_const::M3U_RESOURCE_PATH), axum::routing::get(m3u_api_resource)) } \ No newline at end of file diff --git a/src/api/endpoints/xtream_api.rs b/src/api/endpoints/xtream_api.rs index 302951371..af08d3803 100644 --- a/src/api/endpoints/xtream_api.rs +++ b/src/api/endpoints/xtream_api.rs @@ -1,41 +1,41 @@ // https://github.com/tellytv/go.xtream-codes/blob/master/structs.go -use std::borrow::Cow; use crate::api::api_utils; -use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, is_seek_request, redirect_response, resource_response, force_provider_stream_response, separate_number_and_remainder, serve_file, stream_response, RedirectParams, read_session_token}; +use crate::api::api_utils::{force_provider_stream_response, get_user_target, get_user_target_by_credentials, is_seek_request, redirect_response, resource_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; use crate::api::model::app_state::AppState; use crate::api::model::request::UserApiRequest; -use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType} -; +use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType}; use crate::api::model::xtream::XtreamAuthorizationResponse; -use crate::tuliprox_error::create_tuliprox_error_result; -use crate::tuliprox_error::info_err; -use crate::tuliprox_error::{str_to_io_error, TuliproxError, TuliproxErrorKind}; -use crate::model::{ProxyType, ProxyUserCredentials, UserConnectionPermission}; use crate::model::TargetType; -use crate::model::{Config, ConfigInput, ConfigTarget}; use crate::model::{get_backdrop_path_value, FieldGetAccessor, PlaylistEntry, PlaylistItemType, XtreamCluster, XtreamPlaylistItem}; +use crate::model::{Config, ConfigInput, ConfigTarget}; +use crate::model::{ProxyType, ProxyUserCredentials, UserConnectionPermission}; use crate::repository::playlist_repository::get_target_id_mapping; use crate::repository::storage::{get_target_storage_path, hex_encode}; use crate::repository::{storage_const, user_repository, xtream_repository}; -use crate::utils::HLS_EXT; +use crate::tuliprox_error::create_tuliprox_error_result; +use crate::tuliprox_error::info_err; +use crate::tuliprox_error::{str_to_io_error, TuliproxError, TuliproxErrorKind}; use crate::utils::generate_playlist_uuid; use crate::utils::get_u32_from_serde_value; use crate::utils::request::{extract_extension_from_url, sanitize_sensitive_info}; -use crate::utils::xtream::create_vod_info_from_item; -use crate::utils::{request, xtream}; use crate::utils::trace_if_enabled; +use crate::utils::xtream::create_vod_info_from_item; +use crate::utils::HLS_EXT; +use crate::utils::{request, xtream}; use axum::http::{HeaderMap, StatusCode}; use axum::response::IntoResponse; use bytes::Bytes; use futures::stream::{self, StreamExt}; use futures::Stream; use log::{debug, error, warn}; +use rand::RngCore; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; +use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::fmt::{Display, Formatter}; use std::path::Path; @@ -43,7 +43,7 @@ use std::str::FromStr; use std::sync::Arc; #[derive(Serialize, Deserialize, Debug, Copy, Clone, Eq, PartialEq)] -pub enum XtreamApiStreamContext { +pub enum ApiStreamContext { LiveAlt, Live, Movie, @@ -51,14 +51,14 @@ pub enum XtreamApiStreamContext { Timeshift, } -impl XtreamApiStreamContext { +impl ApiStreamContext { const LIVE: &'static str = "live"; const MOVIE: &'static str = "movie"; const SERIES: &'static str = "series"; const TIMESHIFT: &'static str = "timeshift"; } -impl Display for XtreamApiStreamContext { +impl Display for ApiStreamContext { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!(f, "{}", match self { Self::Live | Self::LiveAlt => Self::LIVE, @@ -69,7 +69,7 @@ impl Display for XtreamApiStreamContext { } } -impl TryFrom for XtreamApiStreamContext { +impl TryFrom for ApiStreamContext { type Error = String; fn try_from(cluster: XtreamCluster) -> Result { match cluster { @@ -80,7 +80,7 @@ impl TryFrom for XtreamApiStreamContext { } } -impl FromStr for XtreamApiStreamContext { +impl FromStr for ApiStreamContext { type Err = TuliproxError; fn from_str(s: &str) -> Result { @@ -95,17 +95,17 @@ impl FromStr for XtreamApiStreamContext { } #[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)] -struct XtreamApiStreamRequest<'a> { - context: XtreamApiStreamContext, - access_token: bool, - username: &'a str, - password: &'a str, - stream_id: &'a str, - action_path: &'a str, +pub struct ApiStreamRequest<'a> { + pub context: ApiStreamContext, + pub access_token: bool, + pub username: &'a str, + pub password: &'a str, + pub stream_id: &'a str, + pub action_path: &'a str, } -impl<'a> XtreamApiStreamRequest<'a> { - pub const fn from(context: XtreamApiStreamContext, +impl<'a> ApiStreamRequest<'a> { + pub const fn from(context: ApiStreamContext, username: &'a str, password: &'a str, stream_id: &'a str, @@ -119,7 +119,7 @@ impl<'a> XtreamApiStreamRequest<'a> { action_path, } } - pub const fn from_access_token(context: XtreamApiStreamContext, + pub const fn from_access_token(context: ApiStreamContext, password: &'a str, stream_id: &'a str, action_path: &'a str) -> Self { @@ -140,18 +140,18 @@ pub fn serve_query(file_path: &Path, filter: &HashMap<&str, HashSet>) -> } pub(in crate::api) fn get_xtream_player_api_stream_url( - input: &ConfigInput, context: XtreamApiStreamContext, action_path: &str, fallback_url: &str, + input: &ConfigInput, context: ApiStreamContext, action_path: &str, fallback_url: &str, ) -> Option { if let Some(input_user_info) = input.get_user_info() { let ctx = match context { - XtreamApiStreamContext::LiveAlt | - XtreamApiStreamContext::Live => { + ApiStreamContext::LiveAlt | + ApiStreamContext::Live => { let use_prefix = input.options.as_ref().is_none_or(|o| o.xtream_live_stream_use_prefix); String::from(if use_prefix { "live" } else { "" }) } - XtreamApiStreamContext::Movie - | XtreamApiStreamContext::Series - | XtreamApiStreamContext::Timeshift => context.to_string() + ApiStreamContext::Movie + | ApiStreamContext::Series + | ApiStreamContext::Timeshift => context.to_string() }; let ctx_path = if ctx.is_empty() { String::new() } else { format!("{ctx}/") }; Some(format!("{}/{}{}/{}/{}", @@ -178,7 +178,8 @@ async fn xtream_player_api_stream( req_headers: &HeaderMap, api_req: &UserApiRequest, app_state: &Arc, - stream_req: XtreamApiStreamRequest<'_>, + stream_req: ApiStreamRequest<'_>, + // _addr: &SocketAddr, ) -> impl IntoResponse + Send { let (user, target) = try_option_bad_request!(get_user_target_by_credentials(stream_req.username, stream_req.password, api_req, app_state), false, format!("Could not find any user {}", stream_req.username)); if user.permission_denied(app_state) { @@ -197,23 +198,24 @@ async fn xtream_player_api_stream( 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)); let cluster = pli.xtream_cluster; - let user_session = match read_session_token(req_headers) { - None => None, - Some(token) => app_state.active_users.get_user_session(&user.username, token).await, + let user_session = if api_req.session == 0 { + None + } else { + app_state.active_users.get_user_session(&user.username, api_req.session).await }; - if let Some(session) = &user_session { + if let Some(session) = &user_session { if session.permission == UserConnectionPermission::Exhausted { - return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); + return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } if app_state.active_provider.is_over_limit(&session.provider).await { return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response(); } - if session.virtual_id == virtual_id && is_seek_request(cluster, req_headers).await { + if session.virtual_id == virtual_id && is_seek_request(cluster, req_headers).await { // partial request means we are in reverse proxy mode, seek happened - return force_provider_stream_response(app_state, session, pli.item_type, req_headers, input, &user).await.into_response() + return force_provider_stream_response(app_state, session, pli.item_type, req_headers, input, &user).await.into_response(); } } @@ -258,14 +260,15 @@ async fn xtream_player_api_stream( return handle_hls_stream_request(app_state, &user, user_session.as_ref(), &stream_url, pli.virtual_id, input, connection_permission).await.into_response(); } - stream_response(app_state, pli.virtual_id, pli.item_type, &stream_url, req_headers, input, target, &user, connection_permission).await.into_response() + stream_response(app_state, api_req.session, pli.virtual_id, pli.item_type, &stream_url, req_headers, input, target, &user, connection_permission).await.into_response() } +// Used by webui async fn xtream_player_api_stream_with_token( req_headers: &HeaderMap, app_state: &Arc, target_id: u16, - stream_req: XtreamApiStreamRequest<'_>, + stream_req: ApiStreamRequest<'_>, ) -> impl IntoResponse + Send { if let Some(target) = app_state.config.get_target_by_id(target_id) { let target_name = &target.name; @@ -319,7 +322,7 @@ async fn xtream_player_api_stream_with_token( stream_req.context)); trace_if_enabled!("Streaming stream request from {}", sanitize_sensitive_info(&stream_url)); - stream_response(app_state, pli.virtual_id, pli.item_type, &stream_url, req_headers, input, target, &user, UserConnectionPermission::Allowed).await.into_response() + stream_response(app_state, 0, pli.virtual_id, pli.item_type, &stream_url, req_headers, input, target, &user, UserConnectionPermission::Allowed).await.into_response() } else { axum::http::StatusCode::BAD_REQUEST.into_response() } @@ -455,7 +458,7 @@ async fn xtream_player_api_resource( req_headers: &HeaderMap, api_req: &UserApiRequest, app_state: &Arc, - resource_req: XtreamApiStreamRequest<'_>, + resource_req: ApiStreamRequest<'_>, ) -> impl IntoResponse { let (user, target) = try_option_bad_request!(get_user_target_by_credentials(resource_req.username, resource_req.password, api_req, app_state), false, format!("Could not find any user {}", resource_req.username)); if user.permission_denied(app_state) { @@ -480,7 +483,7 @@ async fn xtream_player_api_resource( match stream_url { None => axum::http::StatusCode::NOT_FOUND.into_response(), Some(url) => { - if user.proxy.is_redirect(pli.item_type) || target.is_force_redirect(pli.item_type) { + if user.proxy.is_redirect(pli.item_type) || target.is_force_redirect(pli.item_type) { trace_if_enabled!("Redirecting resource request to {}", sanitize_sensitive_info(&url)); redirect(&url).into_response() } else { @@ -491,6 +494,59 @@ async fn xtream_player_api_resource( } } +async fn xtream_player_api_stream_with_session( + // addr: &SocketAddr, + app_state: &Arc, + api_req: &UserApiRequest, + stream_req: ApiStreamRequest<'_>, + req_headers: &HeaderMap, +) -> impl IntoResponse + Send { + if api_req.session > 0 { + return xtream_player_api_stream(req_headers, api_req, app_state, stream_req/*, addr*/).await.into_response() + } + if let Some(credentials) = app_state.config.get_user_credentials(stream_req.username) { + // create new session and redirect + let context = match stream_req.context { + ApiStreamContext::LiveAlt => "", + ApiStreamContext::Live => "live", + ApiStreamContext::Movie => "movie", + ApiStreamContext::Series => "series", + ApiStreamContext::Timeshift => "timeshift", + }; + + let username = stream_req.username; + let password = stream_req.password; + let stream_id = stream_req.stream_id; + + // TODO avoid same tokens twice, this is a little bit hacky + let session_token = rand::rng().next_u32(); + let server_info = app_state.config.get_user_server_info(&credentials); + let redirect_url = format!("{}/{context}/{username}/{password}/{stream_id}?session={session_token}", server_info.get_base_url()); + return redirect(&redirect_url).into_response(); + } + StatusCode::BAD_REQUEST.into_response() +} + +macro_rules! create_xtream_player_api_stream { + ($fn_name:ident, $context:expr) => { + async fn $fn_name( + // axum::extract::ConnectInfo(addr): axum::extract::ConnectInfo, + req_headers: HeaderMap, + axum::extract::Path((username, password, stream_id)): axum::extract::Path<(String, String, String)>, + axum::extract::State(app_state): axum::extract::State>, + axum::extract::Query(api_req): axum::extract::Query, + ) -> impl IntoResponse + Send { + xtream_player_api_stream_with_session( + // &addr, + &app_state, + &api_req, + ApiStreamRequest::from($context, &username, &password, &stream_id, ""), + &req_headers, + ).await.into_response() + } + } +} + macro_rules! create_xtream_player_api_resource { ($fn_name:ident, $context:expr) => { async fn $fn_name( @@ -499,32 +555,19 @@ macro_rules! create_xtream_player_api_resource { axum::extract::Query(api_req): axum::extract::Query, req_headers: HeaderMap, ) -> impl IntoResponse { - xtream_player_api_resource(&req_headers, &api_req, &app_state, XtreamApiStreamRequest::from($context, &username, &password, &stream_id, &resource)).await.into_response() + xtream_player_api_resource(&req_headers, &api_req, &app_state, ApiStreamRequest::from($context, &username, &password, &stream_id, &resource)).await.into_response() } } } -macro_rules! create_xtream_player_api_stream { - ($fn_name:ident, $context:expr) => { - async fn $fn_name( - axum::extract::Path((username, password, stream_id)): axum::extract::Path<(String, String, String)>, - axum::extract::State(app_state): axum::extract::State>, - axum::extract::Query(api_req): axum::extract::Query, - req_headers: HeaderMap, - ) -> impl IntoResponse + Send { - xtream_player_api_stream(&req_headers, &api_req, &app_state, XtreamApiStreamRequest::from($context, &username, &password, &stream_id, "")).await.into_response() - } - } -} +create_xtream_player_api_stream!(xtream_player_api_live_stream, ApiStreamContext::Live); +create_xtream_player_api_stream!(xtream_player_api_live_stream_alt, ApiStreamContext::LiveAlt); +create_xtream_player_api_stream!(xtream_player_api_series_stream, ApiStreamContext::Series); +create_xtream_player_api_stream!(xtream_player_api_movie_stream, ApiStreamContext::Movie); -create_xtream_player_api_stream!(xtream_player_api_live_stream, XtreamApiStreamContext::Live); -create_xtream_player_api_stream!(xtream_player_api_live_stream_alt, XtreamApiStreamContext::LiveAlt); -create_xtream_player_api_stream!(xtream_player_api_series_stream, XtreamApiStreamContext::Series); -create_xtream_player_api_stream!(xtream_player_api_movie_stream, XtreamApiStreamContext::Movie); - -create_xtream_player_api_resource!(xtream_player_api_live_resource, XtreamApiStreamContext::Live); -create_xtream_player_api_resource!(xtream_player_api_series_resource, XtreamApiStreamContext::Series); -create_xtream_player_api_resource!(xtream_player_api_movie_resource, XtreamApiStreamContext::Movie); +create_xtream_player_api_resource!(xtream_player_api_live_resource, ApiStreamContext::Live); +create_xtream_player_api_resource!(xtream_player_api_series_resource, ApiStreamContext::Series); +create_xtream_player_api_resource!(xtream_player_api_movie_resource, ApiStreamContext::Movie); fn get_non_empty<'a>(first: &'a str, second: &'a str, third: &'a str) -> &'a str { if !first.is_empty() { @@ -546,21 +589,63 @@ struct XtreamApiTimeShiftRequest { } async fn xtream_player_api_timeshift_stream( + // ConnectInfo(addr): ConnectInfo, req_headers: HeaderMap, - axum::extract::Query(api_query_req): axum::extract::Query, + axum::extract::Query(mut api_req): axum::extract::Query, axum::extract::Path(timeshift_request): axum::extract::Path, axum::extract::State(app_state): axum::extract::State>, axum::extract::Form(api_form_req): axum::extract::Form, ) -> impl IntoResponse + Send { - let username = get_non_empty(×hift_request.username, &api_query_req.username, &api_form_req.username); - let password = get_non_empty(×hift_request.password, &api_query_req.password, &api_form_req.password); - let stream_id = get_non_empty(×hift_request.stream_id, &api_query_req.stream, &api_form_req.stream); - let duration = get_non_empty(×hift_request.duration, &api_query_req.duration, &api_form_req.duration); - let start = get_non_empty(×hift_request.start, &api_query_req.start, &api_form_req.start); + let username = get_non_empty(×hift_request.username, ×hift_request.username, &api_form_req.username); + let password = get_non_empty(×hift_request.password, ×hift_request.password, &api_form_req.password); + let stream_id = get_non_empty(×hift_request.stream_id, &api_req.stream_id, &api_form_req.stream_id).to_string(); + let duration = get_non_empty(×hift_request.duration, ×hift_request.duration, &api_form_req.duration); + let start = get_non_empty(×hift_request.start, ×hift_request.start, &api_form_req.start); + + if api_req.session == 0 { + if let Some(credentials) = app_state.config.get_user_credentials(×hift_request.username) { + // create new session and redirect + // TODO avoid same tokens twice, this is a little bit hacky + let session_token = rand::rng().next_u32(); + let server_info = app_state.config.get_user_server_info(&credentials); + + let redirect_url = format!("{}/timeshift/{username}/{password}/{duration}/{start}/{stream_id}?session={session_token}", + server_info.get_base_url()); + return redirect(&redirect_url).into_response(); + } + return StatusCode::BAD_REQUEST.into_response(); + } + let action_path = format!("{duration}/{start}"); - xtream_player_api_stream(&req_headers, &api_query_req, &app_state, XtreamApiStreamRequest::from(XtreamApiStreamContext::Timeshift, username, password, stream_id, &action_path)).await + api_req.username = username.to_string(); + api_req.password = password.to_string(); + api_req.stream_id = stream_id.to_string(); + xtream_player_api_stream(&req_headers, &api_req, &app_state, ApiStreamRequest::from(ApiStreamContext::Timeshift, username, password, &stream_id, &action_path), /*&addr*/).await.into_response() } +async fn xtream_player_api_timeshift_query_stream( + req_headers: HeaderMap, + axum::extract::Query(api_query_req): axum::extract::Query, + axum::extract::State(app_state): axum::extract::State>, + axum::extract::Form(api_form_req): axum::extract::Form, +) -> impl IntoResponse + Send { + let username = get_non_empty(&api_query_req.username, &api_form_req.username, ""); + let password = get_non_empty(&api_query_req.password, &api_form_req.password, ""); + let stream_id = get_non_empty(&api_query_req.stream, &api_form_req.stream, ""); + let duration = get_non_empty(&api_query_req.duration, &api_form_req.duration, ""); + let start = get_non_empty(&api_query_req.start, &api_form_req.start, ""); + let action_path = format!("{duration}/{start}"); + if username.is_empty() || password.is_empty() || stream_id.is_empty() || duration.is_empty() || start.is_empty() { + // if token.is_empty() { + return axum::http::StatusCode::BAD_REQUEST.into_response(); + // } + // xtream_player_api_stream(&req_headers, &api_query_req, &app_state, ApiStreamRequest::from_access_token(ApiStreamContext::Timeshift, token, stream_id, &action_path)/*, &addr*/).await.into_response() + } else { + xtream_player_api_stream(&req_headers, &api_query_req, &app_state, ApiStreamRequest::from(ApiStreamContext::Timeshift, username, password, stream_id, &action_path)).await.into_response() + } +} + + async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserCredentials, target: &ConfigTarget, stream_id: &str, cluster: XtreamCluster) -> impl IntoResponse + Send { @@ -632,7 +717,7 @@ async fn xtream_get_short_epg(app_state: &AppState, user: &ProxyUserCredentials, if !(limit.is_empty() || limit.eq("0")) { info_url = format!("{info_url}&limit={limit}"); } - if user.proxy.is_redirect(pli.item_type) || target.is_force_redirect(pli.item_type) { + if user.proxy.is_redirect(pli.item_type) || target.is_force_redirect(pli.item_type) { return redirect(&info_url).into_response(); } @@ -897,8 +982,8 @@ macro_rules! register_xtream_api_timeshift { ($router:expr, [$($path:expr),*]) => {{ $router $( - .route($path, axum::routing::get(xtream_player_api_timeshift_stream)) - .route($path, axum::routing::post(xtream_player_api_timeshift_stream)) + .route($path, axum::routing::get(xtream_player_api_timeshift_query_stream)) + .route($path, axum::routing::post(xtream_player_api_timeshift_query_stream)) //$cfg.service(web::resource($path).route(web::get().to(xtream_player_api_timeshift_stream)).route(web::post().to(xtream_player_api_timeshift_stream))); )* }}; @@ -909,8 +994,8 @@ async fn xtream_player_token_stream( axum::extract::State(app_state): axum::extract::State>, req_headers: HeaderMap, ) -> impl IntoResponse + Send { - let ctxt = try_result_bad_request!(XtreamApiStreamContext::from_str(cluster.as_str())); - xtream_player_api_stream_with_token(&req_headers, &app_state, target_id, XtreamApiStreamRequest::from_access_token(ctxt, &token, &stream_id, "")).await.into_response() + let ctxt = try_result_bad_request!(ApiStreamContext::from_str(cluster.as_str())); + xtream_player_api_stream_with_token(&req_headers, &app_state, target_id, ApiStreamRequest::from_access_token(ctxt, &token, &stream_id, "")).await.into_response() } pub fn xtream_api_register() -> axum::Router> { @@ -922,8 +1007,8 @@ pub fn xtream_api_register() -> axum::Router> { ("/live", xtream_player_api_live_stream), ("/movie", xtream_player_api_movie_stream), ("/series", xtream_player_api_series_stream)]); + router = router.route("/timeshift/{username}/{password}/{duration}/{start}/{stream_id}", axum::routing::get(xtream_player_api_timeshift_stream)); router = register_xtream_api_timeshift!(router, [ - "/timeshift/{username}/{password}/{duration}/{start}/{stream_id}", "/timeshift.php", "/streaming/timeshift.php"]); register_xtream_api_resource!(router, [ diff --git a/src/api/model/active_user_manager.rs b/src/api/model/active_user_manager.rs index 54040240e..cdd39eaf3 100644 --- a/src/api/model/active_user_manager.rs +++ b/src/api/model/active_user_manager.rs @@ -3,7 +3,6 @@ use crate::model::Config; use crate::utils::{current_time_secs, default_grace_period_millis, default_grace_period_timeout_secs}; use jsonwebtoken::get_current_timestamp; use log::{debug, info}; -use rand::RngCore; use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -199,8 +198,7 @@ impl ActiveUserManager { sessions.iter().find(|&session| session.token == token) } - fn new_user_session(virtual_id: u32, provider: &str, stream_url: &str, connection_permission: UserConnectionPermission) -> UserSession { - let session_token = rand::rng().next_u32(); + fn new_user_session(session_token: u32, virtual_id: u32, provider: &str, stream_url: &str, connection_permission: UserConnectionPermission) -> UserSession { UserSession { token: session_token, virtual_id, @@ -211,17 +209,17 @@ impl ActiveUserManager { } } - pub async fn create_user_session(&self, user: &ProxyUserCredentials, virtual_id: u32, provider: &str, stream_url: &str, connection_permission: UserConnectionPermission) -> Option { + pub async fn create_user_session(&self, user: &ProxyUserCredentials, session_token: u32, virtual_id: u32, provider: &str, stream_url: &str, connection_permission: UserConnectionPermission) -> Option { self.gc().await; let mut lock = self.user.write().await; if let Some(connection_data) = lock.get_mut(&user.username) { - let session = Self::new_user_session(virtual_id, provider, stream_url, connection_permission); + let session = Self::new_user_session(session_token, virtual_id, provider, stream_url, connection_permission); let token = session.token; connection_data.sessions.push(session); Some(token) } else { let mut connection_data = UserConnectionData::new(0, user.max_connections); - let session = Self::new_user_session(virtual_id, provider, stream_url, connection_permission); + let session = Self::new_user_session(session_token, virtual_id, provider, stream_url, connection_permission); let token = session.token; connection_data.sessions.push(session); lock.insert(user.username.to_string(), connection_data); diff --git a/src/api/model/request.rs b/src/api/model/request.rs index ef746358c..77776098d 100644 --- a/src/api/model/request.rs +++ b/src/api/model/request.rs @@ -72,4 +72,6 @@ pub struct UserApiRequest { pub duration: String, #[serde(default, alias = "type")] pub content_type: String, + #[serde(default)] + pub session: u32, } \ No newline at end of file diff --git a/src/api/model/stream.rs b/src/api/model/stream.rs index 7bd82e502..ee1918521 100644 --- a/src/api/model/stream.rs +++ b/src/api/model/stream.rs @@ -2,10 +2,11 @@ use crate::api::model::stream_error::StreamError; use axum::http::StatusCode; use bytes::Bytes; use futures::stream::BoxStream; +use url::Url; pub type BoxedProviderStream = BoxStream<'static, Result>; pub type ProviderStreamHeader = Vec<(String, String)>; -pub type ProviderStreamInfo = Option<(ProviderStreamHeader, StatusCode)>; +pub type ProviderStreamInfo = Option<(ProviderStreamHeader, StatusCode, Option)>; pub type ProviderStreamResponse = (Option, ProviderStreamInfo); diff --git a/src/api/model/streams/provider_stream.rs b/src/api/model/streams/provider_stream.rs index dd9ed616e..9f724928c 100644 --- a/src/api/model/streams/provider_stream.rs +++ b/src/api/model/streams/provider_stream.rs @@ -23,7 +23,7 @@ fn create_video_stream(video: Option<&Arc>>, headers: &[(String, String) .filter(|(key, _)| !(key.eq("content-type") || key.eq("content-length") || key.contains("range"))) .map(|(key, value)| (key.to_string(), value.to_string())).collect(); response_headers.push(("content-type".to_string(), "video/mp2t".to_string())); - (Some(Box::pin(CustomVideoStream::new(Arc::clone(video)))), Some((response_headers, StatusCode::OK))) + (Some(Box::pin(CustomVideoStream::new(Arc::clone(video)))), Some((response_headers, StatusCode::OK, None))) } else { (None, None) } @@ -50,7 +50,7 @@ pub fn create_user_account_expired_stream(cfg: &Config, headers: &[(String, Stri } pub fn create_custom_video_stream_response(config: &Config, video_response: CustomVideoStreamType) -> impl axum::response::IntoResponse + Send { - if let (Some(stream), Some((headers, status_code))) = match video_response { + if let (Some(stream), Some((headers, status_code, _))) = match video_response { CustomVideoStreamType::ChannelUnavailable => create_channel_unavailable_stream(config, &[], StatusCode::BAD_REQUEST), CustomVideoStreamType::UserConnectionsExhausted => create_user_connections_exhausted_stream(config, &[]), CustomVideoStreamType::ProviderConnectionsExhausted => create_provider_connections_exhausted_stream(config, &[]), diff --git a/src/api/model/streams/provider_stream_factory.rs b/src/api/model/streams/provider_stream_factory.rs index 2d1671238..e1a6e5954 100644 --- a/src/api/model/streams/provider_stream_factory.rs +++ b/src/api/model/streams/provider_stream_factory.rs @@ -169,8 +169,7 @@ fn get_request_range_start_bytes(req_headers: &HashMap>) -> Opti } } } - Some(0) // catchup workaround. without range catchup not working. - //None + None } // fn get_host_and_optional_port(url: &Url) -> Option { @@ -253,7 +252,7 @@ async fn provider_stream_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())) + Some((response_headers, response.status(), Some(response.url().clone()))) }; let provider_stream = response.bytes_stream().map_err(|err| { @@ -288,14 +287,21 @@ async fn provider_stream_request(cfg: &Config, request_client: Arc {} - _ => { - debug!("Server error status response : {status}"); + StatusCode::GATEWAY_TIMEOUT => { + if let (Some(boxed_provider_stream), response_info) = + create_channel_unavailable_stream(cfg, &get_response_headers(stream_options.get_headers()), StatusCode::BAD_GATEWAY) + { + Ok(Some((boxed_provider_stream, response_info))) + } else { + Err(StatusCode::SERVICE_UNAVAILABLE) + } } + _ => Err(status) } } Err(status) @@ -374,7 +380,7 @@ pub async fn create_provider_stream(cfg: Arc, match get_provider_stream(&cfg, Arc::clone(&client), &stream_options).await { Ok(Some((init_stream, info))) => { - let is_media_stream_or_not_piped = if let Some((headers, _)) = &info { + let is_media_stream_or_not_piped = if let Some((headers, _, _)) = &info { // if it is piped or no video stream, then we don't reconnect !stream_options.pipe_stream && classify_content_type(headers) == MimeCategory::Video } else { diff --git a/src/model/playlist.rs b/src/model/playlist.rs index 967409fec..20cb51ccc 100644 --- a/src/model/playlist.rs +++ b/src/model/playlist.rs @@ -79,10 +79,8 @@ impl TryFrom for XtreamCluster { fn try_from(item_type: PlaylistItemType) -> Result { match item_type { PlaylistItemType::Live | PlaylistItemType::LiveHls | PlaylistItemType::LiveDash | PlaylistItemType::LiveUnknown => Ok(Self::Live), - PlaylistItemType::Video => Ok(Self::Video), + PlaylistItemType::Catchup | PlaylistItemType::Video => Ok(Self::Video), PlaylistItemType::Series | PlaylistItemType::SeriesInfo => Ok(Self::Series), - // TODO is catchup video or live ? - PlaylistItemType::Catchup => Err(format!("Cant convert {item_type}")), } } } diff --git a/src/repository/kodi_repository.rs b/src/repository/kodi_repository.rs index 7a4207fc1..4210bf12d 100644 --- a/src/repository/kodi_repository.rs +++ b/src/repository/kodi_repository.rs @@ -482,9 +482,9 @@ async fn cleanup_strm_output_directory( fn filter_strm_item(pli: &PlaylistItem) -> bool { let item_type = pli.header.item_type; - item_type == PlaylistItemType::Series - || item_type == PlaylistItemType::Live + item_type == PlaylistItemType::Live || item_type == PlaylistItemType::Video + || item_type == PlaylistItemType::Series } fn get_relative_path_str(full_path: &Path, root_path: &Path) -> String { diff --git a/src/utils/network/request.rs b/src/utils/network/request.rs index cd9bfbce5..3b24f32ea 100644 --- a/src/utils/network/request.rs +++ b/src/utils/network/request.rs @@ -506,7 +506,7 @@ pub fn get_base_url_from_str(url: &str) -> Option { } pub fn create_client(cfg: &Config) -> reqwest::ClientBuilder { - let mut client = reqwest::Client::builder(); + let mut client = reqwest::Client::builder().redirect(reqwest::redirect::Policy::limited(10)); if let Some(proxy_cfg) = cfg.proxy.as_ref() { let proxy = match reqwest::Proxy::all(&proxy_cfg.url) {