euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-30 02:32:24 +01:00
Code Issues Packages Projects Releases 5 Wiki Activity

hls reverse proxy implemented

Browse Source
This commit is contained in:
euzu
2025-03-14 13:15:26 +01:00
parent b806ed8166
commit f7eb465bd8
22 changed files with 308 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/endpoints/hls_api.rs
+37 -55
View File
@@ -1,13 +1,11 @@
use crate::api::api_utils::{get_user_target_by_credentials, stream_response};
use crate::api::api_utils::{try_option_bad_request, try_result_bad_request};
use crate::api::api_utils::{stream_response};
use crate::api::api_utils::{try_option_bad_request};
use crate::api::model::app_state::AppState;
use crate::api::model::request::UserApiRequest;
use crate::model::api_proxy::ProxyUserCredentials;
use crate::model::config::{ConfigInput, TargetType};
use crate::model::playlist::{PlaylistEntry, PlaylistItemType, XtreamCluster};
use crate::processing::parser::hls::{rewrite_hls, M3U_HLSR_PREFIX};
use crate::model::playlist::{PlaylistItemType, XtreamCluster};
use crate::processing::parser::hls::rewrite_hls;
use crate::repository::playlist_repository::HLS_EXT;
use crate::repository::{m3u_repository, xtream_repository};
use crate::utils::network::request;
use crate::utils::network::request::{replace_extension, sanitize_sensitive_info};
use axum::response::IntoResponse;
@@ -21,16 +19,22 @@ struct HlsApiPathParams {
token: String,
username: String,
password: String,
channel: String,
hash: String,
chunk: String,
stream_id: u32,
chunk: u32,
}
pub(in crate::api) async fn handle_hls_stream_request(app_state: &Arc<AppState>, user: &ProxyUserCredentials, pli: &dyn PlaylistEntry, input: &ConfigInput, target_type: TargetType) -> impl axum::response::IntoResponse + Send {
let url = replace_extension(&pli.get_provider_url(), HLS_EXT);
pub(in crate::api) async fn handle_hls_stream_request(app_state: &Arc<AppState>,
user: &ProxyUserCredentials,
hls_url: &str,
virtual_id: u32,
input: &ConfigInput,
target_type: TargetType) -> impl axum::response::IntoResponse + Send {
let url = replace_extension(hls_url, HLS_EXT);
let server_info = app_state.config.get_user_server_info(user).await;
match request::download_text_content(Arc::clone(&app_state.http_client), input, &url, None).await {
Ok(content) => {
let hls_content = rewrite_hls(&content, pli.get_virtual_id(), user, &target_type);
let (hls_entry, hls_content) = rewrite_hls(&server_info.get_base_url(), &content, hls_url, virtual_id, user, &target_type, &input.name);
app_state.hls_cache.add_entry(hls_entry).await;
axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, "application/x-mpegurl")
@@ -46,62 +50,40 @@ pub(in crate::api) async fn handle_hls_stream_request(app_state: &Arc<AppState>,
}
async fn hls_api_stream(
req_headers: &axum::http::HeaderMap,
api_req: &UserApiRequest,
params: HlsApiPathParams,
app_state: &Arc<AppState>,
target_type: TargetType,
req_headers: axum::http::HeaderMap,
axum::extract::Path(params): axum::extract::Path<HlsApiPathParams>,
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl axum::response::IntoResponse + Send {
let (user, target) = try_option_bad_request!(
get_user_target_by_credentials(&params.username, &params.password, api_req, app_state).await,
false,
app_state.config.get_target_for_user(&params.username, &params.password).await, false,
format!("Could not find any user {}", params.username));
if !user.has_permissions(app_state).await {
if !user.has_permissions(&app_state).await {
return axum::http::StatusCode::FORBIDDEN.into_response();
}
let Some(hls_entry) = app_state.hls_cache.get_entry(&params.token).await else { return axum::http::StatusCode::BAD_REQUEST.into_response(); };
let Some(hls_url) = hls_entry.get_chunk_url(params.chunk) else { return axum::http::StatusCode::BAD_REQUEST.into_response(); };
let target_name = &target.name;
let virtual_id: u32 = try_result_bad_request!(params.channel.parse());
let (pli_url, input_name) = if target_type == TargetType::Xtream {
let (pli, _) = 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));
(pli.url, pli.input_name)
} else {
let pli = try_result_bad_request!(m3u_repository::m3u_get_item_for_stream_id(virtual_id, &app_state.config, target).await, true, format!("Failed to read xtream item for stream id {}", virtual_id));
(pli.url, pli.input_name)
};
let input = try_option_bad_request!(app_state.config.get_input_by_name(&input_name), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live));
// let input_username = input.username.as_ref().map_or("", |v| v);
// let input_password = input.password.as_ref().map_or("", |v| v);
// let input_url = input.url.as_str();
let virtual_id = params.stream_id;
let input = try_option_bad_request!(app_state.config.get_input_by_name(&hls_entry.input_name), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live));
// we don't respond as hlsr, we take the original stream, because the location could be different and then it does not work
// The next problem is, different url to same channel causes to fail stream share.
// let stream_url = format!("{input_url}/hlsr/{token}/{input_username}/{input_password}/{}/{hash}/{chunk}", pli.provider_id);
stream_response(app_state, &pli_url, req_headers, Some(input), PlaylistItemType::Live, target, &user).await.into_response()
}
if hls_url.ends_with(HLS_EXT) {
return handle_hls_stream_request(&app_state, &user, hls_url, virtual_id, input, hls_entry.target_type.clone()).await.into_response();
}
async fn hls_api_stream_xtream(
req_headers: axum::http::HeaderMap,
axum::extract::Query(api_req): axum::extract::Query<UserApiRequest>,
axum::extract::Path(params): axum::extract::Path<HlsApiPathParams>,
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl axum::response::IntoResponse + Send {
hls_api_stream(&req_headers, &api_req, params, &app_state, TargetType::Xtream).await.into_response()
}
async fn hls_api_stream_m3u(
req_headers: axum::http::HeaderMap,
axum::extract::Query(api_req): axum::extract::Query<UserApiRequest>,
axum::extract::Path(params): axum::extract::Path<HlsApiPathParams>,
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl axum::response::IntoResponse + Send {
hls_api_stream(&req_headers, &api_req, params, &app_state, TargetType::M3u).await.into_response()
// let (pli_url, input_name) = if hls_entry.target_type == TargetType::Xtream {
// let (pli, _) = 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));
// (pli.url, pli.input_name)
// } else {
// let pli = try_result_bad_request!(m3u_repository::m3u_get_item_for_stream_id(virtual_id, &app_state.config, target).await, true, format!("Failed to read xtream item for stream id {}", virtual_id));
// (pli.url, pli.input_name)
// };
stream_response(&app_state, hls_url, &req_headers, Some(input), PlaylistItemType::Live, target, &user).await.into_response()
}
pub fn hls_api_register() -> axum::Router<Arc<AppState>> {
axum::Router::new()
.route("/hlsr/{token}/{username}/{password}/{channel}/{hash}/{chunk}", axum::routing::get(hls_api_stream_xtream))
.route(&format!("/{M3U_HLSR_PREFIX}/{{token}}/{{username}}/{{password}}/{{channel}}/{{hash}}/{{chunk}}"), axum::routing::get(hls_api_stream_m3u))
.route("/hls/{token}/{username}/{password}/{stream_id}/{chunk}", axum::routing::get(hls_api_stream))
//cfg.service(web::resource("/hls/{token}/{stream}").route(web::get().to(xtream_player_api_hls_stream)));
//cfg.service(web::resource("/play/{token}/{type}").route(web::get().to(xtream_player_api_play_stream)));
}
src/api/endpoints/m3u_api.rs
+4 -5
View File
@@ -4,7 +4,7 @@ use crate::api::model::app_state::AppState;
use crate::api::model::request::UserApiRequest;
use crate::model::api_proxy::ProxyType;
use crate::model::config::TargetType;
use crate::model::playlist::{FieldGetAccessor, XtreamCluster};
use crate::model::playlist::{FieldGetAccessor, PlaylistItemType, XtreamCluster};
use crate::repository::m3u_playlist_iterator::{M3U_RESOURCE_PATH, M3U_STREAM_PATH};
use crate::repository::m3u_repository::{m3u_get_item_for_stream_id, m3u_load_rewrite_playlist};
use crate::repository::playlist_repository::HLS_EXT;
@@ -87,7 +87,7 @@ async fn m3u_api_stream(
let input = app_state.config.get_input_by_name(m3u_item.input_name.as_str());
let is_hls_request = stream_ext.as_deref() == Some(HLS_EXT);
let is_hls_request = m3u_item.item_type == PlaylistItemType::LiveHls || stream_ext.as_deref() == Some(HLS_EXT);
if user.proxy == ProxyType::Redirect {
let redirect_url = if is_hls_request { &replace_extension(&m3u_item.url, "m3u8") } else { &m3u_item.url };
@@ -98,9 +98,8 @@ async fn m3u_api_stream(
// Reverse proxy mode
if is_hls_request {
let target_name = &target.name;
let hls_input = try_option_bad_request!(input, true,
format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live));
return handle_hls_stream_request(&app_state, &user, &m3u_item, hls_input, TargetType::M3u).await.into_response();
let hls_input = try_option_bad_request!(input, true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live));
return handle_hls_stream_request(&app_state, &user, &m3u_item.url, m3u_item.virtual_id, hls_input, TargetType::M3u).await.into_response();
}
stream_response(&app_state, m3u_item.url.as_str(), &req_headers, input, m3u_item.item_type, target, &user).await.into_response()
src/api/endpoints/web_index.rs
-10
View File
@@ -96,13 +96,3 @@ pub fn index_register(web_dir_path: &Path) -> axum::Router<Arc<AppState>> {
.route("/", axum::routing::get(index))
.fallback(axum::routing::get_service(tower_http::services::ServeDir::new(web_dir_path))))
}
// pub fn index_register(web_dir_path: &Path) -> impl Fn(&mut web::ServiceConfig) + '_ {
// move |cfg: &mut web::ServiceConfig| {
// cfg.service(web::scope("/auth")
// .route("/token", web::post().to(token))
// .route("/refresh", web::post().to(token_refresh)));
// cfg.service(web::scope("")
// .route("/", web::get().to(index))
// .service(actix_files::Files::new("", web_dir_path)));
// }
// }
src/api/endpoints/xmltv_api.rs
-1
View File
@@ -5,7 +5,6 @@ use axum::response::IntoResponse;
use chrono::{Duration, NaiveDateTime, TimeDelta};
use flate2::write::GzEncoder;
use flate2::Compression;
// use actix_web::{http::header, web, HttpRequest, HttpResponse};
use log::{error, trace};
use quick_xml::events::{BytesStart, Event};
use quick_xml::{Reader, Writer};
src/api/endpoints/xtream_api.rs
+11 -11
View File
@@ -34,7 +34,7 @@ use crate::repository::{user_repository, xtream_repository};
use crate::repository::xtream_repository::{TAG_EPISODES, TAG_INFO_DATA, TAG_SEASONS_DATA};
use crate::utils::hash_utils::generate_playlist_uuid;
use crate::utils::json_utils::get_u32_from_serde_value;
use crate::utils::network::request::{extract_extension_from_url, replace_extension, sanitize_sensitive_info};
use crate::utils::network::request::{extract_extension_from_url, sanitize_sensitive_info};
use crate::utils::network::xtream::{create_vod_info_from_item, ACTION_GET_LIVE_CATEGORIES, ACTION_GET_LIVE_STREAMS, ACTION_GET_SERIES, ACTION_GET_SERIES_CATEGORIES, ACTION_GET_SERIES_INFO, ACTION_GET_VOD_CATEGORIES, ACTION_GET_VOD_INFO, ACTION_GET_VOD_STREAMS};
use crate::utils::json_utils;
use crate::utils::debug_if_enabled;
@@ -159,12 +159,7 @@ async fn xtream_player_api_stream(
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 pli.item_type == PlaylistItemType::LiveHls {
debug_if_enabled!("Redirecting stream request to {}", sanitize_sensitive_info(&pli.url));
return redirect(&pli.url).into_response();
}
let is_hls_request = stream_ext.as_deref() == Some(HLS_EXT);
let is_hls_request = pli.item_type == PlaylistItemType::LiveHls || stream_ext.as_deref() == Some(HLS_EXT);
if user.proxy == ProxyType::Redirect {
if pli.xtream_cluster == XtreamCluster::Series {
@@ -176,14 +171,19 @@ async fn xtream_player_api_stream(
return redirect(&stream_url).into_response();
}
let redirect_url = if is_hls_request { &replace_extension(&pli.url, "m3u8") } else { &pli.url };
debug_if_enabled!("Redirecting stream request to {}", sanitize_sensitive_info(redirect_url));
return redirect(redirect_url.as_str()).into_response();
// if pli.item_type == PlaylistItemType::LiveHls {
// let redirect_url = &replace_extension(&pli.url, "m3u8");
// debug_if_enabled!("Redirecting stream request to {}", sanitize_sensitive_info(redirect_url));
// return redirect(redirect_url).into_response();
// }
debug_if_enabled!("Redirecting stream request to {}", sanitize_sensitive_info(&pli.url));
return redirect(&pli.url).into_response();
}
// Reverse proxy mode
if is_hls_request {
return handle_hls_stream_request(app_state, &user, &pli, input, TargetType::Xtream).await.into_response();
return handle_hls_stream_request(app_state, &user, &pli.url, pli.virtual_id, input, TargetType::Xtream).await.into_response();
}
let extension = stream_ext.unwrap_or_else(
src/api/main_api.rs
+3 -31
View File
@@ -27,6 +27,7 @@ use std::sync::Arc;
use tokio::sync::Mutex;
use std::future::IntoFuture;
use crate::api::model::event_manager::EventManager;
use crate::api::model::hls_cache::HlsCache;
fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result<PathBuf, std::io::Error> {
let web_dir = web_root.to_string();
@@ -124,8 +125,9 @@ fn create_shared_data(cfg: &Arc<Config>) -> AppState {
AppState {
config: Arc::clone(cfg),
http_client: Arc::new(reqwest::Client::new()),
downloads: Arc::from(DownloadQueue::new()),
downloads: Arc::new(DownloadQueue::new()),
cache,
hls_cache: HlsCache::garbage_collected(),
shared_stream_manager: Arc::new(SharedStreamManager::new()),
active_users,
active_provider,
@@ -303,34 +305,4 @@ pub async fn start_server(cfg: Arc<Config>, targets: Arc<ProcessTargets>) -> fut
let router: axum::Router<()> = router.with_state(shared_data.clone());
let listener = tokio::net::TcpListener::bind(format!("{host}:{port}")).await?;
axum::serve(listener, router).into_future().await
// HttpServer::new(move || {
// App::new()
// .wrap(Logger::default())
// .wrap(Cors::default()
// .supports_credentials()
// .allow_any_origin()
// .allowed_methods(vec!["GET", "POST", "OPTIONS", "HEAD"])
// .allow_any_header()
// .max_age(3600))
// .app_data(shared_data.clone())
// // .wrap(Condition::new(web_auth_enabled, ErrorHandlers::new().handler(StatusCode::UNAUTHORIZED, handle_unauthorized)))
// .configure(|srvcfg| {
// if web_ui_enabled {
// srvcfg.service(actix_files::Files::new("/static", web_dir_path.join("static")));
// srvcfg.configure(v1_api_register(web_auth_enabled));
// }
// srvcfg.service(web::resource("/healthcheck").route(web::get().to(healthcheck)));
// srvcfg.service(web::resource("/status").route(web::get().to(status)));
// })
// .configure(xtream_api_register)
// .configure(m3u_api_register)
// .configure(xmltv_api_register)
// .configure(hls_api_register)
// .configure(|srvcfg| {
// if web_ui_enabled {
// srvcfg.configure(index_register(&web_dir_path));
// }
// })
// }).bind(format!("{host}:{port}"))?.run().await
}
src/api/model/app_state.rs
+2
View File
@@ -4,6 +4,7 @@ use crate::api::model::active_provider_manager::ActiveProviderManager;
use crate::api::model::active_user_manager::ActiveUserManager;
use crate::api::model::download::DownloadQueue;
use crate::api::model::event_manager::EventManager;
use crate::api::model::hls_cache::HlsCache;
use crate::api::model::streams::shared_stream_manager::SharedStreamManager;
use crate::model::config::{Config};
use crate::model::hdhomerun_config::HdHomeRunDeviceConfig;
@@ -16,6 +17,7 @@ pub struct AppState {
pub http_client: Arc<reqwest::Client>,
pub downloads: Arc<DownloadQueue>,
pub cache: Arc<Option<Mutex<LRUResourceCache>>>,
pub hls_cache: Arc<HlsCache>,
pub shared_stream_manager: Arc<SharedStreamManager>,
pub active_users: Arc<ActiveUserManager>,
pub active_provider: Arc<ActiveProviderManager>,
src/api/model/hls_cache.rs
+61
View File
@@ -0,0 +1,61 @@
use log::error;
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use chrono::Local;
use cron::Schedule;
use tokio::sync::RwLock;
use tokio::time::Instant;
use crate::utils::sys_utils::exit;
use crate::model::hls::HlsEntry;
const EXPIRE_DURATION: u64 = 600; // 10 minutes
fn start_garbage_collector(cache: &Arc<HlsCache>) {
let cache_clone = cache.clone();
tokio::spawn(async move {
match Schedule::from_str("0 */15 * * * * *") {
Ok(schedule) => {
let offset = *Local::now().offset();
loop {
let mut upcoming = schedule.upcoming(offset).take(1);
if let Some(datetime) = upcoming.next() {
tokio::time::sleep_until(tokio::time::Instant::from(crate::api::scheduler::datetime_to_instant(datetime))).await;
cache_clone.gc().await;
}
}
}
Err(err) => exit!("Failed to start scheduler: {}", err)
}
});
}
pub struct HlsCache {
pub entries: RwLock<HashMap<String, HlsEntry>>,
}
impl HlsCache {
pub fn garbage_collected() -> Arc<Self> {
let cache = Arc::new(Self {
entries: RwLock::new(HashMap::new()),
});
start_garbage_collector(&cache);
cache
}
pub async fn add_entry(&self, entry: HlsEntry) {
self.entries.write().await.insert(entry.token.to_string(), entry);
}
pub async fn get_entry(&self, token: &str) -> Option<HlsEntry>{
self.entries.read().await.get(token).cloned()
}
pub async fn gc(&self) {
let threshold = Instant::now() - Duration::from_secs(EXPIRE_DURATION);
// Remove all expired elements
self.entries.write().await.retain(|_, entry| entry.ts > threshold);
}
}
src/api/model/mod.rs
+1
View File
@@ -9,3 +9,4 @@ pub(crate) mod streams;
pub(in crate::api) mod active_user_manager;
pub(in crate::api) mod active_provider_manager;
pub(in crate::api) mod event_manager;
pub(in crate::api) mod hls_cache;
src/api/model/streams/provider_stream_factory.rs
+1 -1
View File
@@ -421,7 +421,7 @@ pub async fn create_provider_stream(cfg: &Config,
// use std::sync::Arc;
// use crate::model::config::Config;
//
// #[actix_rt::test]
// #[tokio::test]
// async fn test_stream() {
// let app = App::new().route("/test", web::get().to(test_stream_handler));
// let server = test::init_service(app).await;
src/api/scheduler.rs
+4 -3
View File
@@ -8,7 +8,7 @@ use crate::utils::sys_utils::exit;
use crate::model::config::{Config, ProcessTargets};
use crate::processing::processor::playlist::exec_processing;
fn datetime_to_instant(datetime: DateTime<FixedOffset>) -> Instant {
pub fn datetime_to_instant(datetime: DateTime<FixedOffset>) -> Instant {
// Convert DateTime<FixedOffset> to SystemTime
let target_system_time: SystemTime = datetime.into();
@@ -44,11 +44,12 @@ pub async fn start_scheduler(client: Arc<reqwest::Client>, expression: &str, con
mod tests {
use std::str::FromStr;
use std::sync::atomic::{AtomicU8, Ordering};
use std::time::Instant;
use chrono::Local;
use cron::Schedule;
use crate::api::scheduler::datetime_to_instant;
#[actix_rt::test]
#[tokio::test]
async fn test_run_scheduler() {
// Define a cron expression that runs every second
let expression = "0/1 * * * * * *"; // every second
@@ -63,7 +64,7 @@ mod tests {
loop {
let mut upcoming = schedule.upcoming(offset).take(1);
if let Some(datetime) = upcoming.next() {
tokio::time::sleep_until(actix_rt::time::Instant::from(datetime_to_instant(datetime))).await;
tokio::time::sleep_until(tokio::time::Instant::from(datetime_to_instant(datetime))).await;
run_me();
}
if runs.load(Ordering::SeqCst) == 6 {