euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-26 16:52:25 +01:00
Code Issues Packages Projects Releases 5 Wiki Activity

config refactoring

Browse Source
This commit is contained in:
euzu
2025-06-24 17:34:09 +02:00
parent f38d4e32b3
commit c0eccb68ab
29 changed files with 594 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/api/api_utils.rs
+11 -11
View File
@@ -120,7 +120,7 @@ pub async fn serve_file(file_path: &Path, mime_type: mime::Mime) -> impl axum::r
pub fn get_user_target_by_username(username: &str, app_state: &AppState) -> Option<(ProxyUserCredentials, Arc<ConfigTarget>)> {
if !username.is_empty() {
return app_state.config.get_target_for_username(username);
return app_state.app_config.get_target_for_username(username);
}
None
}
@@ -128,13 +128,13 @@ pub fn get_user_target_by_username(username: &str, app_state: &AppState) -> Opti
pub fn get_user_target_by_credentials<'a>(username: &str, password: &str, api_req: &'a UserApiRequest,
app_state: &'a AppState) -> Option<(ProxyUserCredentials, Arc<ConfigTarget>)> {
if !username.is_empty() && !password.is_empty() {
app_state.config.get_target_for_user(username, password)
app_state.app_config.get_target_for_user(username, password)
} else {
let token = api_req.token.as_str().trim();
if token.is_empty() {
None
} else {
app_state.config.get_target_for_user_by_token(token)
app_state.app_config.get_target_for_user_by_token(token)
}
}
}
@@ -174,7 +174,7 @@ pub struct StreamOptions {
/// Returns a `StreamOptions` instance with the resolved configuration.
fn get_stream_options(app_state: &AppState) -> StreamOptions {
let (stream_retry, stream_force_retry_secs, buffer_enabled, buffer_size) = app_state
.config.config.load()
.app_config.config.load()
.reverse_proxy
.as_ref()
.and_then(|reverse_proxy| reverse_proxy.stream.as_ref())
@@ -298,7 +298,7 @@ async fn resolve_streaming_strategy(app_state: &AppState, stream_url: &str, inpu
let stream_response_params = match &*provider_connection_guard {
ProviderAllocation::Exhausted => {
debug!("Input {} is exhausted. No connections allowed.", input.name);
let stream = create_provider_connections_exhausted_stream(&app_state.config, &[]);
let stream = create_provider_connections_exhausted_stream(&app_state.app_config, &[]);
ProviderStreamState::Custom(stream)
}
ProviderAllocation::Available(ref provider)
@@ -345,7 +345,7 @@ async fn create_stream_response_details(app_state: &AppState,
force_provider: Option<&str>) -> StreamDetails {
let mut streaming_strategy =
resolve_streaming_strategy(app_state, stream_url, input, force_provider).await;
let config_grace_period_millis = app_state.config.config.load().reverse_proxy.as_ref()
let config_grace_period_millis = app_state.app_config.config.load().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, &streaming_strategy.provider_stream_state, config_grace_period_millis);
match streaming_strategy.provider_stream_state {
@@ -367,7 +367,7 @@ async fn create_stream_response_details(app_state: &AppState,
let ((stream, stream_info), reconnect_flag) = if let Ok(url) = parsed_url {
let provider_stream_factory_options = ProviderStreamFactoryOptions::new(item_type, share_stream, stream_options, &url, req_headers, streaming_strategy.input_headers.as_ref());
let reconnect_flag = provider_stream_factory_options.get_reconnect_flag_clone();
let provider_stream = match create_provider_stream(Arc::clone(&app_state.config), Arc::clone(&app_state.http_client), provider_stream_factory_options).await {
let provider_stream = match create_provider_stream(Arc::clone(&app_state.app_config), Arc::clone(&app_state.http_client), provider_stream_factory_options).await {
None => (None, None),
Some((stream, info)) => {
(Some(stream), info)
@@ -555,7 +555,7 @@ pub async fn force_provider_stream_response(app_state: &AppState,
}
drop(stream_details.provider_connection_guard.take());
if let (Some(stream), _stream_info) =
create_channel_unavailable_stream(&app_state.config, &[], StatusCode::BAD_GATEWAY)
create_channel_unavailable_stream(&app_state.app_config, &[], StatusCode::BAD_GATEWAY)
{
debug!("Streaming custom stream");
axum::response::Response::builder().status(StatusCode::OK).body(Body::from_stream(stream)).unwrap().into_response()
@@ -579,7 +579,7 @@ pub async fn stream_response(app_state: &AppState,
if log_enabled!(log::Level::Trace) { trace!("Try to open stream {}", sanitize_sensitive_info(stream_url)); }
if connection_permission == UserConnectionPermission::Exhausted {
return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserConnectionsExhausted).into_response();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response();
}
let share_stream = is_stream_share_enabled(item_type, target);
@@ -646,7 +646,7 @@ pub async fn stream_response(app_state: &AppState,
}
fn get_stream_throttle(app_state: &AppState) -> u64 {
app_state.config.config.load()
app_state.app_config.config.load()
.reverse_proxy
.as_ref()
.and_then(|reverse_proxy| reverse_proxy.stream.as_ref())
@@ -775,7 +775,7 @@ pub fn get_username_from_auth_header(
token: &str,
app_state: &Arc<AppState>,
) -> Option<String> {
if let Some(web_auth_config) = &app_state.config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) {
if let Some(web_auth_config) = &app_state.app_config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) {
let secret_key: &str = web_auth_config.secret.as_ref();
if let Ok(token_data) = decode::<Claims>(
token,
backend/src/api/config_watch.rs
+58 -23
View File
@@ -1,13 +1,14 @@
use crate::api::model::app_state::AppState;
use shared::error::{TuliproxError, TuliproxErrorKind};
use crate::utils;
use crate::utils::is_directory;
use crate::utils::{is_directory, read_config_file, read_sources_file};
use log::{debug, error, info};
use notify::event::{AccessKind, AccessMode};
use notify::{recommended_watcher, Event, EventKind, RecursiveMode, Watcher};
use shared::error::{TuliproxError, TuliproxErrorKind};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{mpsc, Arc};
use crate::model::{Config, SourcesConfig};
enum ConfigFile {
Config,
@@ -18,17 +19,20 @@ enum ConfigFile {
impl ConfigFile {
fn load_mappping(app_state: &Arc<AppState>) -> Result<(), TuliproxError> {
match utils::read_mappings(app_state.config.t_mapping_file_path.as_str(), true) {
Ok(Some(mappings_cfg)) => {
app_state.config.set_mappings(&mappings_cfg);
info!("Loaded mapping file {}", app_state.config.t_mapping_file_path.as_str());
}
Ok(None) => {
info!("No mapping file loaded {}", app_state.config.t_mapping_file_path.as_str());
}
Err(err) => {
error!("Failed to load mapping file {err}");
return Err(err);
let paths = app_state.app_config.paths.load();
if let Some(mapping_file_path) = paths.mapping_file_path.as_ref() {
match utils::read_mappings(mapping_file_path, true) {
Ok(Some(mappings_cfg)) => {
app_state.app_config.set_mappings(&mappings_cfg);
info!("Loaded mapping file {mapping_file_path}");
}
Ok(None) => {
info!("No mapping file loaded {mapping_file_path}");
}
Err(err) => {
error!("Failed to load mapping file {err}");
return Err(err);
}
}
}
@@ -36,9 +40,15 @@ impl ConfigFile {
}
fn load_api_proxy(app_state: &Arc<AppState>) -> Result<(), TuliproxError> {
match utils::read_api_proxy_config(&app_state.config) {
Ok(()) => {
info!("Api Proxy File: {:?}", &app_state.config.t_api_proxy_file_path);
match utils::read_api_proxy_config(&app_state.app_config, true) {
Ok(Some(api_proxy)) => {
app_state.app_config.set_api_proxy(api_proxy)?;
let paths = app_state.app_config.paths.load();
info!("Loaded Api Proxy File: {:?}", &paths.api_proxy_file_path);
}
Ok(None) => {
let paths = app_state.app_config.paths.load();
info!("Coul dnot load Api Proxy File: {:?}", &paths.api_proxy_file_path);
}
Err(err) => {
error!("Failed to load api-proxy file {err}");
@@ -47,12 +57,35 @@ impl ConfigFile {
}
Ok(())
}
fn load_config(app_state: &Arc<AppState>) -> Result<(), TuliproxError> {
let paths = app_state.app_config.paths.load();
let config_file = paths.config_file_path.as_str();
let config_dto = read_config_file(config_file, true)?;
let mut config: Config = Config::from(config_dto);
config.prepare(paths.config_path.as_str())?;
info!("Loaded config file {config_file}");
app_state.set_config(config)?;
Ok(())
}
fn load_sources(app_state: &Arc<AppState>) -> Result<(), TuliproxError> {
let paths = app_state.app_config.paths.load();
let sources_file = paths.sources_file_path.as_str();
let sources_dto = read_sources_file(sources_file, true, true)?;
let sources: SourcesConfig = sources_dto.into();
info!("Loaded sources file {sources_file}");
app_state.app_config.set_sources(sources)?;
Ok(())
}
pub(crate) fn reload(&self, file_path: &Path, app_state: &Arc<AppState>) -> Result<(), TuliproxError> {
debug!("File change detected {}", file_path.display());
match self {
ConfigFile::ApiProxy => ConfigFile::load_api_proxy(app_state),
ConfigFile::Mapping => ConfigFile::load_mappping(app_state),
ConfigFile::Config | ConfigFile::Sources => { Ok(()) }
ConfigFile::Config => ConfigFile::load_config(app_state),
ConfigFile::Sources => ConfigFile::load_sources(app_state),
}
}
}
@@ -60,11 +93,13 @@ impl ConfigFile {
pub async fn exec_config_watch(app_state: &Arc<AppState>) -> Result<(), TuliproxError> {
let (tx, rx) = mpsc::channel::<notify::Result<Event>>();
let paths = &app_state.app_config.paths.load();
let mapping_file_path = paths.mapping_file_path.as_ref().map_or_else(String::new, ToString::to_string);
let mut files = HashMap::new();
[(&app_state.config.t_config_file_path, ConfigFile::Config),
(&app_state.config.t_api_proxy_file_path, ConfigFile::ApiProxy),
(&app_state.config.t_mapping_file_path, ConfigFile::Mapping),
(&app_state.config.t_sources_file_path, ConfigFile::Sources)
[(&paths.config_file_path, ConfigFile::Config),
(&paths.api_proxy_file_path, ConfigFile::ApiProxy),
(&mapping_file_path, ConfigFile::Mapping),
(&paths.sources_file_path, ConfigFile::Sources)
].into_iter()
.filter(|(path, _)| !path.is_empty())
.for_each(|(path, config_file)| { files.insert(PathBuf::from(path), (config_file, is_directory(path))); });
@@ -77,8 +112,8 @@ pub async fn exec_config_watch(app_state: &Arc<AppState>) -> Result<(), Tuliprox
// Add a path to be watched. All files and directories at that path and
// below will be monitored for changes.
let path = Path::new(app_state.config.t_config_path.as_str());
let recursive_mode = if utils::is_directory(&app_state.config.t_mapping_file_path) { RecursiveMode::Recursive } else { RecursiveMode::NonRecursive };
let path = Path::new(paths.config_path.as_str());
let recursive_mode = if !mapping_file_path.is_empty() && utils::is_directory(&mapping_file_path) { RecursiveMode::Recursive } else { RecursiveMode::NonRecursive };
watcher.watch(path, recursive_mode).map_err(|err| TuliproxError::new(TuliproxErrorKind::Info, format!("Failed to start config file watcher {err}")))?;
info!("Watching config file changes {}", path.display());
backend/src/api/endpoints/download_api.rs
+2 -2
View File
@@ -118,7 +118,7 @@ pub async fn queue_download_file(
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
axum::extract::Json(req): axum::extract::Json<FileDownloadRequest>,
) -> impl axum::response::IntoResponse + Send {
let app_config = &*app_state.config;
let app_config = &*app_state.app_config;
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&app_config.config);
if let Some(download_cfg) = config.video.as_ref().unwrap().download.as_ref() {
@@ -129,7 +129,7 @@ pub async fn queue_download_file(
Some(file_download) => {
app_state.downloads.queue.lock().await.push_back(file_download.clone());
if app_state.downloads.active.read().await.is_none() {
match run_download_queue(&app_state.config, download_cfg, &app_state.downloads).await {
match run_download_queue(&app_state.app_config, download_cfg, &app_state.downloads).await {
Ok(()) => {}
Err(err) => return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err}))).into_response(),
}
backend/src/api/endpoints/hdhomerun_api.rs
+5 -5
View File
@@ -157,8 +157,8 @@ where
}
fn create_device(app_state: &Arc<HdHomerunAppState>) -> Option<Device> {
if let Some(credentials) = app_state.app_state.config.get_user_credentials(&app_state.device.t_username) {
let server_info = app_state.app_state.config.get_user_server_info(&credentials);
if let Some(credentials) = app_state.app_state.app_config.get_user_credentials(&app_state.device.t_username) {
let server_info = app_state.app_state.app_config.get_user_server_info(&credentials);
let device = &app_state.device;
let device_url = format!("{}://{}:{}", server_info.protocol, server_info.host, device.port);
Some(Device {
@@ -237,7 +237,7 @@ async fn lineup(app_state: &Arc<HdHomerunAppState>, cfg: &Arc<AppConfig>, creden
.body(axum::body::Body::from_stream(body_stream))
.unwrap().into_response();
} else if (use_all || use_xtream) && target.has_output(&TargetType::Xtream) {
let server_info = app_state.app_state.config.get_user_server_info(credentials);
let server_info = app_state.app_state.app_config.get_user_server_info(credentials);
let base_url = server_info.get_base_url();
let base_url_live = if credentials.proxy.is_redirect(PlaylistItemType::Live) || target.is_force_redirect(PlaylistItemType::Live) { None } else { Some(base_url.clone()) };
@@ -273,7 +273,7 @@ async fn lineup(app_state: &Arc<HdHomerunAppState>, cfg: &Arc<AppConfig>, creden
}
async fn auth_lineup_json(AuthBasic((username, password)): AuthBasic, axum::extract::State(app_state): axum::extract::State<Arc<HdHomerunAppState>>) -> impl IntoResponse {
let cfg = Arc::clone(&app_state.app_state.config);
let cfg = Arc::clone(&app_state.app_state.app_config);
if let Some((credentials, target)) = cfg.get_target_for_username(&app_state.device.t_username) {
if !username.eq(&credentials.username) || !password.eq(&credentials.password) {
return axum::http::StatusCode::UNAUTHORIZED.into_response();
@@ -285,7 +285,7 @@ async fn auth_lineup_json(AuthBasic((username, password)): AuthBasic, axum::extr
}
async fn lineup_json(axum::extract::State(app_state): axum::extract::State<Arc<HdHomerunAppState>>) -> impl IntoResponse {
let cfg = Arc::clone(&app_state.app_state.config);
let cfg = Arc::clone(&app_state.app_state.app_config);
if let Some((credentials, target)) = cfg.get_target_for_username(&app_state.device.t_username) {
let user_credentials = Arc::new(credentials);
return lineup(&app_state, &cfg, &user_credentials, &target).await.into_response();
backend/src/api/endpoints/hls_api.rs
+10 -10
View File
@@ -44,7 +44,7 @@ pub(in crate::api) async fn handle_hls_stream_request(
input: &ConfigInput,
connection_permission: UserConnectionPermission) -> impl IntoResponse + Send {
let url = replace_url_extension(hls_url, HLS_EXT);
let server_info = app_state.config.get_user_server_info(user);
let server_info = app_state.app_config.get_user_server_info(user);
let (request_url, session_token) = match user_session {
Some(session) => {
@@ -72,7 +72,7 @@ pub(in crate::api) async fn handle_hls_stream_request(
match request::download_text_content(Arc::clone(&app_state.http_client), input, &request_url, None).await {
Ok((content, response_url)) => {
let rewrite_hls_props = RewriteHlsProps {
secret: &app_state.config.t_encrypt_secret,
secret: &app_state.app_config.encrypt_secret,
base_url: &server_info.get_base_url(),
content: &content,
hls_url: response_url,
@@ -85,7 +85,7 @@ pub(in crate::api) async fn handle_hls_stream_request(
}
Err(err) => {
error!("Failed to download m3u8 {}", sanitize_sensitive_info(err.to_string().as_str()));
create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::ChannelUnavailable).into_response()
create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::ChannelUnavailable).into_response()
}
}
}
@@ -97,29 +97,29 @@ async fn hls_api_stream(
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl axum::response::IntoResponse + Send {
let (user, target) = try_option_bad_request!(
app_state.config.get_target_for_user(&params.username, &params.password), false,
app_state.app_config.get_target_for_user(&params.username, &params.password), false,
format!("Could not find any user {}", params.username));
if user.permission_denied(&app_state) {
return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserAccountExpired).into_response();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserAccountExpired).into_response();
}
let target_name = &target.name;
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 input = try_option_bad_request!(app_state.app_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 user_session_token = format!("{fingerprint}{virtual_id}");
let mut user_session = app_state.active_users.get_user_session(&user.username, &user_session_token).await;
if let Some(session) = &mut 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.app_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();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response();
}
let hls_url = match get_hls_session_token_and_url_from_token(&app_state.config.t_encrypt_secret, &params.token) {
let hls_url = match get_hls_session_token_and_url_from_token(&app_state.app_config.encrypt_secret, &params.token) {
Some((Some(session_token), hls_url)) if session.token.eq(&session_token) => hls_url,
_ => return axum::http::StatusCode::BAD_REQUEST.into_response(),
};
@@ -136,7 +136,7 @@ async fn hls_api_stream(
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();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response();
}
if is_hls_url(&session.stream_url) {
backend/src/api/endpoints/m3u_api.rs
+8 -8
View File
@@ -22,7 +22,7 @@ async fn m3u_api(
) -> impl axum::response::IntoResponse + Send {
match get_user_target(api_req, app_state) {
Some((user, target)) => {
match m3u_load_rewrite_playlist(&app_state.config, &target, &user).await {
match m3u_load_rewrite_playlist(&app_state.app_config, &target, &user).await {
Ok(m3u_iter) => {
// Convert the iterator into a stream of `Bytes`
let content_stream = stream::iter(m3u_iter.map(|line| Ok::<Bytes, String>(Bytes::from([line.to_string().as_bytes(), b"\n"].concat()))));
@@ -69,7 +69,7 @@ async fn m3u_api_stream(
) -> impl axum::response::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) {
return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserAccountExpired).into_response();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserAccountExpired).into_response();
}
let target_name = &target.name;
@@ -80,8 +80,8 @@ async fn m3u_api_stream(
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 pli = try_result_bad_request!(m3u_get_item_for_stream_id(virtual_id, &app_state.app_config, &target).await, true, format!("Failed to read m3u item for stream id {}", virtual_id));
let input = try_option_bad_request!(app_state.app_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);
@@ -90,11 +90,11 @@ async fn m3u_api_stream(
let session_url = 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.app_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();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response();
}
if session.virtual_id == virtual_id && is_seek_request(cluster, req_headers).await {
// partial request means we are in reverse proxy mode, seek happened
@@ -107,7 +107,7 @@ async fn m3u_api_stream(
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();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response();
}
let context = ApiStreamContext::try_from(cluster).unwrap_or(ApiStreamContext::Live);
@@ -159,7 +159,7 @@ async fn m3u_api_resource(
debug!("Target has no m3u playlist {target_name}");
return StatusCode::BAD_REQUEST.into_response();
}
let m3u_item = match m3u_get_item_for_stream_id(m3u_stream_id, &app_state.config, &target).await {
let m3u_item = match m3u_get_item_for_stream_id(m3u_stream_id, &app_state.app_config, &target).await {
Ok(item) => item,
Err(err) => {
error!("Failed to get m3u url: {}", sanitize_sensitive_info(err.to_string().as_str()));
backend/src/api/endpoints/user_api.rs
+4 -4
View File
@@ -52,7 +52,7 @@ async fn playlist_categories(
}
let target_name = &target.name;
let xtream_stream = if target.has_output(&TargetType::Xtream) {
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
let live_categories = get_categories_from_xtream(xtream_get_playlist_categories(config, target_name, XtreamCluster::Live).await);
let vod_categories = get_categories_from_xtream(xtream_get_playlist_categories(config, target_name, XtreamCluster::Video).await);
let series_categories = get_categories_from_xtream(xtream_get_playlist_categories(config, target_name, XtreamCluster::Series).await);
@@ -70,7 +70,7 @@ async fn playlist_categories(
};
let m3u_stream = if target.has_output(&TargetType::M3u) {
let live_categories = get_categories_from_m3u_playlist(&target, &app_state.config).await;
let live_categories = get_categories_from_m3u_playlist(&target, &app_state.app_config).await;
stream::iter(vec![
Ok::<Bytes, String>(Bytes::from(r#"{"live": "#)),
Ok::<Bytes, String>(Bytes::from(serde_json::to_string(&live_categories).unwrap_or("[]".to_string()))),
@@ -108,7 +108,7 @@ async fn save_playlist_bouquet(
if user.permission_denied(&app_state) {
return axum::http::StatusCode::FORBIDDEN.into_response();
}
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
match save_user_bouquet(config, &target.name, &username, &bouquet).await {
Ok(()) => {
return axum::http::StatusCode::OK.into_response();
@@ -131,7 +131,7 @@ async fn playlist_bouquet(
if user.permission_denied(&app_state) {
return axum::http::StatusCode::FORBIDDEN.into_response();
}
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
let xtream = load_user_bouquet_as_json(config, &username, TargetType::Xtream).await;
let m3u = load_user_bouquet_as_json(config, &username, TargetType::M3u).await;
return axum::response::Response::builder()
backend/src/api/endpoints/v1_api.rs
+33 -98
View File
@@ -5,7 +5,7 @@ use crate::api::model::app_state::AppState;
use crate::api::model::request::{PlaylistRequest, PlaylistRequestType};
use crate::auth::create_access_token;
use crate::auth::validator_admin;
use crate::model::{AppConfig, ConfigSource, ConfigTarget, StatusCheck, TargetUser};
use crate::model::{StatusCheck, TargetUser};
use crate::model::{ConfigInput, ConfigInputOptions};
use crate::processing::processor::playlist;
use crate::repository::user_repository::store_api_user;
@@ -15,11 +15,10 @@ use axum::response::IntoResponse;
use log::error;
use serde_json::json;
use shared::error::TuliproxError;
use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, ConfigApiDto, ConfigDto, ConfigRenameDto, ConfigSortDto, InputType, TargetOutputDto, TargetUserDto, XtreamPlaylistItem};
use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, ConfigDto, InputType, TargetUserDto, XtreamPlaylistItem};
use shared::utils::sanitize_sensitive_info;
use std::collections::{BTreeMap, HashSet};
use std::sync::Arc;
use crate::api::model::config::{ServerConfig, ServerInputConfig, ServerSourceConfig, ServerTargetConfig};
fn intern_save_config_api_proxy(backup_dir: &str, api_proxy: &ApiProxyConfigDto, file_path: &str) -> Option<TuliproxError> {
match utils::save_api_proxy(file_path, backup_dir, api_proxy) {
@@ -68,20 +67,21 @@ async fn save_config_api_proxy_user(
}
}
if let Some(old_api_proxy) = app_state.config.t_api_proxy.load().clone() {
if let Some(old_api_proxy) = app_state.app_config.api_proxy.load().clone() {
let mut api_proxy = (*old_api_proxy).clone();
api_proxy.user = users.iter().map(TargetUser::from).collect();
let new_api_proxy = Arc::new(api_proxy);
app_state.config.t_api_proxy.store(Some(Arc::clone(&new_api_proxy)));
app_state.app_config.api_proxy.store(Some(Arc::clone(&new_api_proxy)));
if new_api_proxy.use_user_db {
if let Err(err) = store_api_user(&app_state.config, &new_api_proxy.user) {
if let Err(err) = store_api_user(&app_state.app_config, &new_api_proxy.user) {
return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response();
}
} else {
let config = app_state.config.config.load();
let config = app_state.app_config.config.load();
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(&*new_api_proxy), app_state.config.t_api_proxy_file_path.as_str()) {
let paths = app_state.app_config.paths.load();
if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(&*new_api_proxy), paths.api_proxy_file_path.as_str()) {
return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response();
}
}
@@ -94,8 +94,9 @@ async fn save_config_main(
axum::extract::Json(cfg): axum::extract::Json<ConfigDto>,
) -> impl axum::response::IntoResponse + Send {
if cfg.is_valid() {
let file_path = app_state.config.t_config_file_path.as_str();
let config = app_state.config.config.load();
let paths = app_state.app_config.paths.load();
let file_path = paths.config_file_path.as_str();
let config = app_state.app_config.config.load();
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
if let Some(err) = intern_save_config_main(file_path, backup_dir, &cfg) {
return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response();
@@ -117,14 +118,15 @@ async fn save_config_api_proxy_config(
}
// TODO wenn hot reload an ist wird doppelt geladen
if let Some(old_api_proxy) = app_state.config.t_api_proxy.load().clone() {
if let Some(old_api_proxy) = app_state.app_config.api_proxy.load().clone() {
let mut api_proxy = (*old_api_proxy).clone();
api_proxy.server = req_api_proxy.iter().map(Into::into).collect();
let new_api_proxy = Arc::new(api_proxy);
app_state.config.t_api_proxy.store(Some(Arc::clone(&new_api_proxy)));
let config = app_state.config.config.load();
app_state.app_config.api_proxy.store(Some(Arc::clone(&new_api_proxy)));
let config = app_state.app_config.config.load();
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(new_api_proxy.as_ref()), app_state.config.t_api_proxy_file_path.as_str()) {
let paths = app_state.app_config.paths.load();
if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(new_api_proxy.as_ref()), paths.api_proxy_file_path.as_str()) {
return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response();
}
}
@@ -136,10 +138,10 @@ async fn playlist_update(
axum::extract::Json(targets): axum::extract::Json<Vec<String>>,
) -> impl axum::response::IntoResponse + Send {
let user_targets = if targets.is_empty() { None } else { Some(targets) };
let process_targets = app_state.config.sources.load().validate_targets(user_targets.as_ref());
let process_targets = app_state.app_config.sources.load().validate_targets(user_targets.as_ref());
match process_targets {
Ok(valid_targets) => {
tokio::spawn(playlist::exec_processing(Arc::clone(&app_state.http_client), Arc::clone(&app_state.config), Arc::new(valid_targets)));
tokio::spawn(playlist::exec_processing(Arc::clone(&app_state.http_client), Arc::clone(&app_state.app_config), Arc::new(valid_targets)));
axum::http::StatusCode::OK.into_response()
}
Err(err) => {
@@ -192,18 +194,18 @@ async fn playlist_content(
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
axum::extract::Json(playlist_req): axum::extract::Json<PlaylistRequest>,
) -> impl IntoResponse + Send {
let config = app_state.config.config.load();
let config = app_state.app_config.config.load();
match playlist_req.rtype {
PlaylistRequestType::Input => {
if let Some(source_id) = playlist_req.source_id {
get_playlist(Arc::clone(&app_state.http_client), app_state.config.get_input_by_id(source_id).as_deref(), &config).await.into_response()
get_playlist(Arc::clone(&app_state.http_client), app_state.app_config.get_input_by_id(source_id).as_deref(), &config).await.into_response()
} else {
(axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid input"}))).into_response()
}
}
PlaylistRequestType::Target => {
if let Some(source_id) = playlist_req.source_id {
get_playlist_for_target(app_state.config.get_target_by_id(source_id).as_deref(), &app_state.config).await.into_response()
get_playlist_for_target(app_state.app_config.get_target_by_id(source_id).as_deref(), &app_state.app_config).await.into_response()
} else {
(axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid target"}))).into_response()
}
@@ -232,10 +234,10 @@ async fn playlist_webplayer(
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
axum::extract::Json(playlist_item): axum::extract::Json<XtreamPlaylistItem>,
) -> impl axum::response::IntoResponse + Send {
let access_token = create_access_token(&app_state.config.t_access_token_secret, 5);
let config = app_state.config.config.load();
let access_token = create_access_token(&app_state.app_config.access_token_secret, 5);
let config = app_state.app_config.config.load();
let server_name = config.web_ui.as_ref().and_then(|web_ui| web_ui.player_server.as_ref()).map_or("default", |server_name| server_name.as_str());
let server_info = app_state.config.get_server_info(server_name);
let server_info = app_state.app_config.get_server_info(server_name);
let base_url = server_info.get_base_url();
format!("{base_url}/token/{access_token}/{target_id}/{}/{}", playlist_item.xtream_cluster.as_stream_type(), playlist_item.virtual_id).into_response()
}
@@ -243,82 +245,15 @@ async fn playlist_webplayer(
async fn config(
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl axum::response::IntoResponse + Send {
let map_input = |i: &Arc<ConfigInput>| ServerInputConfig {
id: i.id,
name: i.name.clone(),
input_type: i.input_type,
url: i.url.clone(),
username: i.username.clone(),
password: i.password.clone(),
persist: i.persist.clone(),
enabled: i.enabled,
};
let map_target = |t: &Arc<ConfigTarget>| {
let mapping = t.mapping.load();
ServerTargetConfig {
id: t.id,
enabled: t.enabled,
name: t.name.clone(),
options: t.options.clone(),
sort: t.sort.as_ref().map(ConfigSortDto::from),
filter: t.filter.to_string(),
output: t.output.iter().map(TargetOutputDto::from).collect(),
rename: t.rename.as_ref().map(|l| l.iter().map(ConfigRenameDto::from).collect()),
mapping: mapping.as_ref().map(|m| m.iter().map(|m| m.id.to_string()).collect()),
processing_order: t.processing_order,
watch: t.watch.as_ref().map(|w| w.iter().map(std::string::ToString::to_string).collect()),
let paths = app_state.app_config.paths.load();
match utils::read_app_config_dto(&paths,
true, false) {
Ok(app_config) => axum::response::Json(app_config).into_response(),
Err(err) => {
error!("Failed to read config files: {err}");
axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
};
let map_source = |s: &ConfigSource| ServerSourceConfig {
inputs: s.inputs.iter().map(map_input).collect(),
targets: s.targets.iter().map(map_target).collect(),
};
let map_config = |app_config: &AppConfig| {
let sources = app_config.sources.load();
let config = app_config.config.load();
ServerConfig {
api: ConfigApiDto::from(&config.api),
threads: config.threads,
working_dir: config.working_dir.clone(),
backup_dir: config.backup_dir.clone(),
user_config_dir: config.user_config_dir.clone(),
log: config.log.as_ref().map(Into::into),
update_on_boot: config.update_on_boot,
web_ui: config.web_ui.as_ref().map(Into::into),
schedules: config.schedules.as_ref().map(|l| l.iter().map(Into::into).collect()),
reverse_proxy: config.reverse_proxy.as_ref().map(Into::into),
messaging: config.messaging.as_ref().map(Into::into),
video: config.video.as_ref().map(Into::into),
sources: sources.sources.iter().map(map_source).collect(),
proxy: config.proxy.as_ref().map(Into::into),
ipcheck: config.ipcheck.as_ref().map(Into::into),
api_proxy: utils::read_api_proxy_dto(&app_state.config, false),
}
};
let mut result = match utils::read_config(app_state.config.t_config_path.as_str(),
app_state.config.t_config_file_path.as_str(),
app_state.config.t_sources_file_path.as_str(),
app_state.config.t_api_proxy_file_path.as_str(),
Some(app_state.config.t_mapping_file_path.to_string()),
false) {
Ok(mut cfg) => {
let _ = cfg.prepare(false);
map_config(&cfg)
}
Err(_) => map_config(&app_state.config)
};
// if we didn't read it from file then we should use it from app_state
if result.api_proxy.is_none() {
let value = app_state.config.t_api_proxy.load();
result.api_proxy = value.as_ref().map(|a| ApiProxyConfigDto::from(a.as_ref()));
}
axum::response::Json(result).into_response()
}
@@ -331,7 +266,7 @@ struct IpCheck {
}
async fn create_ipinfo_check(app_state: &Arc<AppState>) -> Option<(Option<String>, Option<String>)> {
let config = app_state.config.config.load();
let config = app_state.app_config.config.load();
if let Some(ipcheck) = config.ipcheck.as_ref() {
if let Ok(check) = get_ips(&app_state.http_client, ipcheck).await {
return Some(check);
@@ -404,7 +339,7 @@ pub fn v1_api_register(web_auth_enabled: bool, app_state: Arc<AppState>, web_ui_
.route("/playlist", axum::routing::post(playlist_content))
.route("/file/download", axum::routing::post(download_api::queue_download_file))
.route("/file/download/info", axum::routing::get(download_api::download_file_info));
let config = app_state.config.config.load();
let config = app_state.app_config.config.load();
if config.ipcheck.is_some() {
router = router.route("/ipinfo", axum::routing::get(ipinfo));
}
backend/src/api/endpoints/web_index.rs
+5 -5
View File
@@ -18,7 +18,7 @@ async fn token(
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
axum::extract::Json(mut req): axum::extract::Json<UserCredential>,
) -> impl axum::response::IntoResponse + Send {
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
match config.web_ui.as_ref().and_then(|c| c.auth.as_ref()) {
None => no_web_auth_token().into_response(),
Some(web_auth) => {
@@ -37,7 +37,7 @@ async fn token(
}
}
}
if let Some(credentials) = app_state.config.get_user_credentials(username) {
if let Some(credentials) = app_state.app_config.get_user_credentials(username) {
if credentials.password == password {
if let Ok(token) = create_jwt_user(web_auth, username) {
req.zeroize();
@@ -57,7 +57,7 @@ async fn token_refresh(
AuthBearer(token): AuthBearer,
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl axum::response::IntoResponse + Send {
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
match &config.web_ui.as_ref().and_then(|c| c.auth.as_ref()) {
None => no_web_auth_token().into_response(),
Some(web_auth) => {
@@ -85,7 +85,7 @@ async fn token_refresh(
async fn index(
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl axum::response::IntoResponse + Send {
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
let path: PathBuf = [&config.api.web_root, "index.html"].iter().collect();
if let Some(web_ui_path) = &config.web_ui.as_ref().and_then(|c| c.path.as_ref()) {
match tokio::fs::read_to_string(&path).await {
@@ -115,7 +115,7 @@ async fn index(
async fn index_config(
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
) -> impl axum::response::IntoResponse + Send {
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
let path: PathBuf = [&config.api.web_root, "config.json"].iter().collect();
if let Some(web_ui_path) = &config.web_ui.as_ref().and_then(|c| c.path.as_ref()) {
match tokio::fs::read_to_string(&path).await {
backend/src/api/endpoints/xmltv_api.rs
+1 -1
View File
@@ -192,7 +192,7 @@ async fn xmltv_api(
return axum::http::StatusCode::FORBIDDEN.into_response();
}
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
let Some(epg_path) = get_epg_path_for_target(config, &target) else {
// No epg configured, No processing or timeshift, epg can't be mapped to the channels.
// we do not deliver epg
backend/src/api/endpoints/xtream_api.rs
+29 -29
View File
@@ -166,9 +166,9 @@ pub(in crate::api) fn get_xtream_player_api_stream_url(
}
async fn get_user_info(user: &ProxyUserCredentials, app_state: &AppState) -> XtreamAuthorizationResponse {
let server_info = app_state.config.get_user_server_info(user);
let server_info = app_state.app_config.get_user_server_info(user);
let active_connections = app_state.get_active_connections_for_user(&user.username).await;
XtreamAuthorizationResponse::new(&server_info, user, active_connections, app_state.config.config.load().user_access_control)
XtreamAuthorizationResponse::new(&server_info, user, active_connections, app_state.app_config.config.load().user_access_control)
}
async fn xtream_player_api_stream(
@@ -180,7 +180,7 @@ async fn xtream_player_api_stream(
) -> 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) {
return create_custom_video_stream_response(&app_state.config, CustomVideoStreamType::UserAccountExpired).into_response();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserAccountExpired).into_response();
}
let target_name = &target.name;
@@ -191,8 +191,8 @@ async fn xtream_player_api_stream(
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));
let (pli, mapping) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id));
let input = try_option_bad_request!(app_state.app_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 item_type = if stream_req.context == ApiStreamContext::Timeshift { PlaylistItemType::Catchup } else { pli.item_type };
@@ -202,11 +202,11 @@ async fn xtream_player_api_stream(
let session_url = 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.app_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();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::ProviderConnectionsExhausted).into_response();
}
if session.virtual_id == virtual_id && is_seek_request(cluster, req_headers).await {
@@ -221,7 +221,7 @@ async fn xtream_player_api_stream(
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();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response();
}
let context = stream_req.context;
@@ -271,7 +271,7 @@ async fn xtream_player_api_stream_with_token(
target_id: u16,
stream_req: ApiStreamRequest<'_>,
) -> impl IntoResponse + Send {
if let Some(target) = app_state.config.get_target_by_id(target_id) {
if let Some(target) = app_state.app_config.get_target_by_id(target_id) {
let target_name = &target.name;
if !target.has_output(&TargetType::Xtream) {
debug!("Target has no xtream output {target_name}");
@@ -279,14 +279,14 @@ async fn xtream_player_api_stream_with_token(
}
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));
let (pli, _mapping) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id));
let input = try_option_bad_request!(app_state.app_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 session_key = format!("{fingerprint}{virtual_id}");
let is_hls_request = pli.item_type == PlaylistItemType::LiveHls || stream_ext.as_deref() == Some(HLS_EXT);
let config = app_state.config.config.load();
let config = app_state.app_config.config.load();
let server = config.web_ui.as_ref().and_then(|web_ui| web_ui.player_server.as_ref()).map_or("default", |server_name| server_name.as_str());
let user = ProxyUserCredentials {
@@ -475,11 +475,11 @@ async fn xtream_player_api_resource(
}
let virtual_id: u32 = try_result_bad_request!(resource_req.stream_id.trim().parse());
let resource = resource_req.action_path.trim();
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));
let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, &target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id));
let stream_url = if resource.starts_with(crate::model::XC_INFO_RESOURCE_PREFIX) {
try_result_bad_request!(xtream_get_info_resource_url(&app_state.config, &pli, &target, resource))
try_result_bad_request!(xtream_get_info_resource_url(&app_state.app_config, &pli, &target, resource))
} else if resource.starts_with(crate::model::XC_SEASON_RESOURCE_PREFIX) {
try_result_bad_request!(xtream_get_season_resource_url(&app_state.config, &pli, &target, resource))
try_result_bad_request!(xtream_get_season_resource_url(&app_state.app_config, &pli, &target, resource))
} else {
pli.get_field(resource)
};
@@ -612,15 +612,15 @@ async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserC
Err(_) => return axum::http::StatusCode::BAD_REQUEST.into_response()
};
if let Ok((pli, virtual_record)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, Some(cluster)) {
if let Ok((pli, virtual_record)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, target, Some(cluster)) {
if pli.provider_id > 0 {
let input_name = &pli.input_name;
if let Some(input) = app_state.config.get_input_by_name(input_name.as_str()) {
if let Some(input) = app_state.app_config.get_input_by_name(input_name.as_str()) {
if let Some(info_url) = xtream::get_xtream_player_api_info_url(&input, cluster, pli.provider_id) {
// Redirect is only possible for live streams, vod and series info needs to be modified
if user.proxy == ProxyType::Redirect && cluster == XtreamCluster::Live {
return redirect(&info_url).into_response();
} else if let Ok(content) = xtream::get_xtream_stream_info(Arc::clone(&app_state.http_client), &app_state.config, user, &input, target, &pli, info_url.as_str(), cluster).await {
} else if let Ok(content) = xtream::get_xtream_stream_info(Arc::clone(&app_state.http_client), &app_state.app_config, user, &input, target, &pli, info_url.as_str(), cluster).await {
return axum::response::Response::builder()
.status(StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
@@ -666,10 +666,10 @@ async fn xtream_get_short_epg(app_state: &AppState, user: &ProxyUserCredentials,
Err(_) => return axum::http::StatusCode::BAD_REQUEST.into_response()
};
if let Ok((pli, _)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, None) {
if let Ok((pli, _)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, target, None) {
if pli.provider_id > 0 {
let input_name = &pli.input_name;
if let Some(input) = app_state.config.get_input_by_name(input_name.as_str()) {
if let Some(input) = app_state.app_config.get_input_by_name(input_name.as_str()) {
if let Some(action_url) = xtream::get_xtream_player_api_action_url(&input, crate::model::XC_ACTION_GET_SHORT_EPG) {
let mut info_url = format!("{action_url}&{}={}", crate::model::XC_TAG_STREAM_ID, pli.provider_id);
if !(limit.is_empty() || limit.eq("0")) {
@@ -727,16 +727,16 @@ async fn xtream_player_api_handle_content_action(config: &Config, target_name: &
async fn xtream_get_catchup_response(app_state: &AppState, target: &ConfigTarget, stream_id: &str, start: &str, end: &str) -> impl IntoResponse + Send {
let virtual_id: u32 = try_result_bad_request!(FromStr::from_str(stream_id));
let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, Some(XtreamCluster::Live)));
let input = try_option_bad_request!(app_state.config.get_input_by_name(pli.input_name.as_str()));
let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.app_config, target, Some(XtreamCluster::Live)));
let input = try_option_bad_request!(app_state.app_config.get_input_by_name(pli.input_name.as_str()));
let info_url = try_option_bad_request!(xtream::get_xtream_player_api_action_url(&input, crate::model::XC_ACTION_GET_CATCHUP_TABLE)
.map(|action_url| format!("{action_url}&{}={}&start={start}&end={end}", crate::model::XC_TAG_STREAM_ID, pli.provider_id)));
let content = try_result_bad_request!(xtream::get_xtream_stream_info_content(Arc::clone(&app_state.http_client), info_url.as_str(), &input).await);
let mut doc: Map<String, Value> = try_result_bad_request!(serde_json::from_str(&content));
let epg_listings = try_option_bad_request!(doc.get_mut(crate::model::XC_TAG_EPG_LISTINGS).and_then(Value::as_array_mut));
let config = &app_state.config.config.load();
let config = &app_state.app_config.config.load();
let target_path = try_option_bad_request!(get_target_storage_path(config, target.name.as_str()));
let (mut target_id_mapping, file_lock) = get_target_id_mapping(&app_state.config, &target_path).await;
let (mut target_id_mapping, file_lock) = get_target_id_mapping(&app_state.app_config, &target_path).await;
for epg_list_item in epg_listings.iter_mut().filter_map(Value::as_object_mut) {
// TODO epg_id
if let Some(catchup_provider_id) = epg_list_item.get(crate::model::XC_TAG_ID).and_then(Value::as_str).and_then(|id| id.parse::<u32>().ok()) {
@@ -798,7 +798,7 @@ async fn xtream_player_api(
}
// Process specific playlist actions
let (skip_live, skip_vod, skip_series) = if let Some(inputs) = app_state.config.get_inputs_for_target(&target.name) {
let (skip_live, skip_vod, skip_series) = if let Some(inputs) = app_state.app_config.get_inputs_for_target(&target.name) {
inputs.iter().fold((true, true, true), |acc, i| {
let (l, v, s) = acc;
i.options.as_ref().map_or((false, false, false), |o| (l && o.xtream_skip_live, v && o.xtream_skip_vod, s && o.xtream_skip_series))
@@ -831,18 +831,18 @@ async fn xtream_player_api(
let category_id = api_req.category_id.trim().parse::<u32>().ok();
// Handle general content actions
if let Some(response) = xtream_player_api_handle_content_action(
&app_state.config.config.load(), &target.name, action, category_id, &user,
&app_state.app_config.config.load(), &target.name, action, category_id, &user,
).await {
return response.into_response();
}
let result = match action {
crate::model::XC_ACTION_GET_LIVE_STREAMS =>
skip_flag_optional!(skip_live, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Live, &app_state.config, &target, category_id, &user).await),
skip_flag_optional!(skip_live, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Live, &app_state.app_config, &target, category_id, &user).await),
crate::model::XC_ACTION_GET_VOD_STREAMS =>
skip_flag_optional!(skip_vod, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Video, &app_state.config, &target, category_id, &user).await),
skip_flag_optional!(skip_vod, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Video, &app_state.app_config, &target, category_id, &user).await),
crate::model::XC_ACTION_GET_SERIES =>
skip_flag_optional!(skip_series, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Series, &app_state.config, &target, category_id, &user).await),
skip_flag_optional!(skip_series, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Series, &app_state.app_config, &target, category_id, &user).await),
_ => Some(Err(info_err!(format!("Cant find action: {action} for target: {}", &target.name))
)),
};
backend/src/api/main_api.rs
+3 -3
View File
@@ -81,7 +81,7 @@ async fn create_shared_data(app_config: &Arc<AppConfig>) -> AppState {
let client = builder.build().unwrap_or_else(|_| Client::new());
AppState {
config: Arc::clone(app_config),
app_config: Arc::clone(app_config),
http_client: Arc::new(client),
downloads: Arc::new(DownloadQueue::new()),
cache,
@@ -175,7 +175,7 @@ fn create_compression_layer() -> tower_http::compression::CompressionLayer {
fn start_hdhomerun(app_config: &Arc<AppConfig>, app_state: &Arc<AppState>, infos: &mut Vec<String>) {
let config = app_config.config.load();
let host = config.api.host.to_string();
let guard = app_config.t_hdhomerun.load();
let guard = app_config.hdhomerun.load();
if let Some(hdhomerun) = &*guard {
if hdhomerun.enabled {
for device in &hdhomerun.devices {
@@ -247,7 +247,7 @@ pub async fn start_server(app_config: Arc<AppConfig>, targets: Arc<ProcessTarget
let web_auth_enabled = is_web_auth_enabled(&cfg, web_ui_enabled);
if app_config.t_api_proxy.load().is_some() {
if app_config.api_proxy.load().is_some() {
start_hdhomerun(&app_config, &app_state, &mut infos);
}
backend/src/api/model/active_provider_manager.rs
+24 -22
View File
@@ -2,21 +2,19 @@ use crate::model::{AppConfig, ConfigInput};
use log::{debug, log_enabled};
use std::collections::HashMap;
use std::ops::Deref;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};
use crate::api::model::provider_config::{ProviderConfig, ProviderConfigWrapper};
use shared::utils::{default_grace_period_millis, default_grace_period_timeout_secs};
pub struct ProviderConnectionGuard {
// manager: Arc<ActiveProviderManager>,
allocation: ProviderAllocation,
}
impl ProviderConnectionGuard {
pub fn new(_manager: Arc<ActiveProviderManager>, allocation: ProviderAllocation) -> Self {
pub fn new(allocation: ProviderAllocation) -> Self {
Self {
// manager,
allocation,
}
}
@@ -395,20 +393,18 @@ impl MultiProviderLineup {
}
pub struct ActiveProviderManager {
grace_period_millis: u64,
grace_period_timeout_secs: u64,
grace_period_millis: AtomicU64,
grace_period_timeout_secs: AtomicU64,
providers: Arc<RwLock<Vec<ProviderLineup>>>,
}
impl ActiveProviderManager {
pub async fn new(cfg: &AppConfig) -> Self {
let (grace_period_millis, grace_period_timeout_secs) = cfg.config.load().reverse_proxy.as_ref()
.and_then(|r| r.stream.as_ref())
.map_or_else(|| (default_grace_period_millis(), default_grace_period_timeout_secs()), |s| (s.grace_period_millis, s.grace_period_timeout_secs));
let (grace_period_millis, grace_period_timeout_secs) = Self::get_grace_options(cfg);
let mut this = Self {
grace_period_millis,
grace_period_timeout_secs,
grace_period_millis: AtomicU64::new(grace_period_millis),
grace_period_timeout_secs: AtomicU64::new(grace_period_timeout_secs),
providers: Arc::new(RwLock::new(Vec::new())),
};
for source in &cfg.sources.load().sources {
@@ -419,12 +415,17 @@ impl ActiveProviderManager {
this
}
fn clone_inner(&self) -> Self {
Self {
grace_period_millis: self.grace_period_millis,
grace_period_timeout_secs: self.grace_period_timeout_secs,
providers: Arc::clone(&self.providers),
}
fn get_grace_options(cfg: &AppConfig) -> (u64, u64) {
let (grace_period_millis, grace_period_timeout_secs) = cfg.config.load().reverse_proxy.as_ref()
.and_then(|r| r.stream.as_ref())
.map_or_else(|| (default_grace_period_millis(), default_grace_period_timeout_secs()), |s| (s.grace_period_millis, s.grace_period_timeout_secs));
(grace_period_millis, grace_period_timeout_secs)
}
pub fn update_config(&self, cfg: &AppConfig) {
let (grace_period_millis, grace_period_timeout_secs) = Self::get_grace_options(cfg);
self.grace_period_millis.store(grace_period_millis, Ordering::Relaxed);
self.grace_period_timeout_secs.store(grace_period_timeout_secs, Ordering::Relaxed);
}
pub async fn add_provider(&mut self, input: &ConfigInput) {
@@ -474,7 +475,7 @@ impl ActiveProviderManager {
Some((_lineup, config)) => config.force_allocate().await,
};
ProviderConnectionGuard::new(Arc::new(self.clone_inner()), allocation)
ProviderConnectionGuard::new(allocation)
}
// Returns the next available provider connection
@@ -482,7 +483,8 @@ impl ActiveProviderManager {
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)) => lineup.acquire(self.grace_period_millis > 0, self.grace_period_timeout_secs).await
Some((lineup, _config)) => lineup.acquire(self.grace_period_millis.load(Ordering::Relaxed) > 0,
self.grace_period_timeout_secs.load(Ordering::Relaxed)).await
};
if log_enabled!(log::Level::Debug) {
@@ -495,7 +497,7 @@ impl ActiveProviderManager {
}
}
ProviderConnectionGuard::new(Arc::new(self.clone_inner()), allocation)
ProviderConnectionGuard::new(allocation)
}
// This method is used for redirects to cycle through provider
@@ -505,7 +507,7 @@ impl ActiveProviderManager {
match Self::get_provider_config(input_name, &providers) {
None => None,
Some((lineup, _config)) => {
let cfg = lineup.get_next(self.grace_period_timeout_secs).await;
let cfg = lineup.get_next(self.grace_period_timeout_secs.load(Ordering::Relaxed)).await;
if log_enabled!(log::Level::Debug) {
if let Some(ref c) = cfg {
debug!("Using provider {}", c.name);
@@ -564,7 +566,7 @@ impl ActiveProviderManager {
pub async fn is_over_limit(&self, provider_name: &str) -> bool {
let providers = self.providers.read().await;
if let Some((_, config)) = Self::get_provider_config(provider_name, &providers) {
config.is_over_limit(self.grace_period_timeout_secs).await
config.is_over_limit(self.grace_period_timeout_secs.load(Ordering::Relaxed)).await
} else {
false
}
backend/src/api/model/active_user_manager.rs
+85 -53
View File
@@ -4,16 +4,76 @@ use shared::utils::{current_time_secs, default_grace_period_millis, default_grac
use jsonwebtoken::get_current_timestamp;
use log::{debug, info};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use tokio::sync::RwLock;
use shared::model::UserConnectionPermission;
macro_rules! active_user_manager_shared_impl {
() => {
#[inline]
async fn get_active_connections(user: &Arc<RwLock<HashMap<String, UserConnectionData>>>) -> usize {
user.read().await.values().map(|c| c.connections as usize).sum()
}
fn log_active_user(&self) {
if self.is_log_user_enabled() {
let user = Arc::clone(&self.user);
tokio::spawn(async move {
let user_count = user.read().await.len();
let user_connection_count = Self::get_active_connections(&user).await;
info!("Active Users: {user_count}, Active User Connections: {user_connection_count}");
});
}
}
};
}
const USER_CON_TTL: u64 = 10_800; // 3 hours
const USER_SESSION_LIMIT: usize = 50;
fn get_grace_options(config: &Config) -> (u64, u64) {
let (grace_period_millis, grace_period_timeout_secs) = config.reverse_proxy.as_ref()
.and_then(|r| r.stream.as_ref())
.map_or_else(|| (default_grace_period_millis(), default_grace_period_timeout_secs()), |s| (s.grace_period_millis, s.grace_period_timeout_secs));
(grace_period_millis, grace_period_timeout_secs)
}
struct ConnectionGuardUserManager {
log_active_user: bool,
user: Arc<RwLock<HashMap<String, UserConnectionData>>>,
}
impl ConnectionGuardUserManager {
active_user_manager_shared_impl!();
fn is_log_user_enabled(&self)-> bool {
self.log_active_user
}
async fn remove_connection(&self, username: &str) {
let mut lock = self.user.write().await;
if let Some(connection_data) = lock.get_mut(username) {
if connection_data.connections > 0 {
connection_data.connections -= 1;
}
if connection_data.connections == 0 {
lock.remove(username);
} else if connection_data.connections < connection_data.max_connections {
// Grace timeout expired, reset grace counters
connection_data.granted_grace = false;
connection_data.grace_ts = 0;
}
}
drop(lock);
self.log_active_user();
}
}
pub struct UserConnectionGuard {
manager: Arc<ActiveUserManager>,
manager: Arc<ConnectionGuardUserManager>,
username: String,
}
impl Drop for UserConnectionGuard {
@@ -68,9 +128,9 @@ impl UserConnectionData {
}
pub struct ActiveUserManager {
grace_period_millis: u64,
grace_period_timeout_secs: u64,
log_active_user: bool,
grace_period_millis: AtomicU64,
grace_period_timeout_secs: AtomicU64,
log_active_user: AtomicBool,
user: Arc<RwLock<HashMap<String, UserConnectionData>>>,
gc_ts: Option<AtomicU64>,
}
@@ -78,26 +138,29 @@ pub struct ActiveUserManager {
impl ActiveUserManager {
pub fn new(config: &Config) -> Self {
let log_active_user = config.log.as_ref().is_some_and(|l| l.log_active_user);
let (grace_period_millis, grace_period_timeout_secs) = config.reverse_proxy.as_ref()
.and_then(|r| r.stream.as_ref())
.map_or_else(|| (default_grace_period_millis(), default_grace_period_timeout_secs()), |s| (s.grace_period_millis, s.grace_period_timeout_secs));
let (grace_period_millis, grace_period_timeout_secs) = get_grace_options(config);
Self {
grace_period_millis,
grace_period_timeout_secs,
log_active_user,
grace_period_millis : AtomicU64::new(grace_period_millis),
grace_period_timeout_secs: AtomicU64::new(grace_period_timeout_secs),
log_active_user: AtomicBool::new(log_active_user),
user: Arc::new(RwLock::new(HashMap::new())),
gc_ts: Some(AtomicU64::new(current_time_secs())),
}
}
fn clone_inner(&self) -> Self {
Self {
grace_period_millis: self.grace_period_millis,
grace_period_timeout_secs: self.grace_period_timeout_secs,
log_active_user: self.log_active_user,
pub fn update_config(&self, config: &Config) {
let log_active_user = config.log.as_ref().is_some_and(|l| l.log_active_user);
let (grace_period_millis, grace_period_timeout_secs) = get_grace_options(config);
self.grace_period_millis.store(grace_period_millis, Ordering::Relaxed);
self.grace_period_timeout_secs.store(grace_period_timeout_secs, Ordering::Relaxed);
self.log_active_user.store(log_active_user, Ordering::Relaxed);
}
fn clone_inner(&self) -> ConnectionGuardUserManager {
ConnectionGuardUserManager {
log_active_user: self.log_active_user.load(Ordering::Relaxed),
user: Arc::clone(&self.user),
gc_ts: None,
}
}
@@ -121,7 +184,7 @@ impl ActiveUserManager {
let now = get_current_timestamp();
// Check if user already used grace period
if connection_data.granted_grace {
if current_connections > connection_data.max_connections && now - connection_data.grace_ts <= self.grace_period_timeout_secs {
if current_connections > connection_data.max_connections && now - connection_data.grace_ts <= self.grace_period_timeout_secs.load(Ordering::Relaxed) {
// Grace timeout still active, deny connection
debug!("User access denied, grace exhausted, too many connections: {username}");
return UserConnectionPermission::Exhausted;
@@ -131,7 +194,7 @@ impl ActiveUserManager {
connection_data.grace_ts = 0;
}
if self.grace_period_millis > 0 && current_connections == connection_data.max_connections {
if self.grace_period_millis.load(Ordering::Relaxed) > 0 && current_connections == connection_data.max_connections {
// Allow grace period once
connection_data.granted_grace = true;
connection_data.grace_ts = now;
@@ -166,11 +229,6 @@ impl ActiveUserManager {
Self::get_active_connections(&self.user).await
}
#[inline]
async fn get_active_connections(user: &Arc<RwLock<HashMap<String, UserConnectionData>>>) -> usize {
user.read().await.values().map(|c| c.connections as usize).sum()
}
pub async fn add_connection(&self, username: &str, max_connections: u32) -> UserConnectionGuard {
let mut lock = self.user.write().await;
if let Some(connection_data) = lock.get_mut(username) {
@@ -189,24 +247,9 @@ impl ActiveUserManager {
}
}
async fn remove_connection(&self, username: &str) {
let mut lock = self.user.write().await;
if let Some(connection_data) = lock.get_mut(username) {
if connection_data.connections > 0 {
connection_data.connections -= 1;
}
if connection_data.connections == 0 {
lock.remove(username);
} else if connection_data.connections < connection_data.max_connections {
// Grace timeout expired, reset grace counters
connection_data.granted_grace = false;
connection_data.grace_ts = 0;
}
}
drop(lock);
self.log_active_user();
active_user_manager_shared_impl!();
fn is_log_user_enabled(&self)-> bool {
self.log_active_user.load(Ordering::Relaxed)
}
fn find_user_session<'a>(token: &'a str, sessions: &'a [UserSession]) -> Option<&'a UserSession> {
@@ -294,17 +337,6 @@ impl ActiveUserManager {
None
}
fn log_active_user(&self) {
if self.log_active_user {
let user = Arc::clone(&self.user);
tokio::spawn(async move {
let user_count = user.read().await.len();
let user_connection_count = Self::get_active_connections(&user).await;
info!("Active Users: {user_count}, Active User Connections: {user_connection_count}");
});
}
}
async fn gc(&self) {
if let Some(gc_ts) = &self.gc_ts {
let ts = gc_ts.load(Ordering::Acquire);
backend/src/api/model/app_state.rs
+10 -2
View File
@@ -1,16 +1,17 @@
use tokio::sync::{Mutex};
use std::sync::Arc;
use shared::error::TuliproxError;
use shared::model::UserConnectionPermission;
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::streams::shared_stream_manager::SharedStreamManager;
use crate::model::{AppConfig, HdHomeRunDeviceConfig};
use crate::model::{AppConfig, Config, HdHomeRunDeviceConfig};
use crate::tools::lru_cache::LRUResourceCache;
#[derive(Clone)]
pub struct AppState {
pub config: Arc<AppConfig>,
pub app_config: Arc<AppConfig>,
pub http_client: Arc<reqwest::Client>,
pub downloads: Arc<DownloadQueue>,
pub cache: Arc<Option<Mutex<LRUResourceCache>>>,
@@ -20,6 +21,13 @@ pub struct AppState {
}
impl AppState {
pub fn set_config(&self, config: Config) -> Result<(), TuliproxError> {
self.active_users.update_config(&config);
self.active_provider.update_config(&self.app_config);
self.app_config.set_config(config)
}
pub async fn get_active_connections_for_user(&self, username: &str) -> u32 {
self.active_users.user_connections(username).await
}
backend/src/api/model/streams/active_client_stream.rs
+4 -3
View File
@@ -47,11 +47,12 @@ impl ActiveClientStream {
let grant_user_grace_period = connection_permission == UserConnectionPermission::GracePeriod;
let username = user.username.as_str();
let user_connection_guard = Some(active_user.add_connection(username, user.max_connections).await);
let cfg = &app_state.config;
let cfg = &app_state.app_config;
let waker = Arc::new(Mutex::new(None));
let waker_clone = Arc::clone(&waker);
let grace_stop_flag = Self::stream_grace_period(&stream_details, grant_user_grace_period, user, &active_user, &active_provider, &waker_clone);
let custom_video = cfg.t_custom_stream_response.as_ref()
let custom_response = cfg.custom_stream_response.load();
let custom_video = custom_response.as_ref()
.map_or((None, None), |c|
(
c.user_connections_exhausted.clone(),
@@ -59,7 +60,7 @@ impl ActiveClientStream {
));
let stream = stream_details.stream.take().unwrap();
let stream = match app_state.config.config.load().sleep_timer_mins {
let stream = match app_state.app_config.config.load().sleep_timer_mins {
None => stream,
Some(mins) => {
let secs = u32::try_from((u64::from(mins) * 60).min(u64::from(u32::MAX))).unwrap_or(0);
backend/src/api/model/streams/provider_stream.rs
+8 -4
View File
@@ -30,22 +30,26 @@ fn create_video_stream(video_buffer: Option<&TransportStreamBuffer>, headers: &[
}
pub fn create_channel_unavailable_stream(cfg: &AppConfig, headers: &[(String, String)], status: StatusCode) -> ProviderStreamResponse {
let video = cfg.t_custom_stream_response.as_ref().and_then(|c| c.channel_unavailable.as_ref());
let custom_stream_response = cfg.custom_stream_response.load();
let video = custom_stream_response.as_ref().and_then(|c| c.channel_unavailable.as_ref());
create_video_stream(video, headers, &format!("Streaming response channel unavailable for status {status}"))
}
pub fn create_user_connections_exhausted_stream(cfg: &AppConfig, headers: &[(String, String)]) -> ProviderStreamResponse {
let video = cfg.t_custom_stream_response.as_ref().and_then(|c| c.user_connections_exhausted.as_ref());
let custom_stream_response = cfg.custom_stream_response.load();
let video = custom_stream_response.as_ref().and_then(|c| c.user_connections_exhausted.as_ref());
create_video_stream(video, headers, "Streaming response user connections exhausted")
}
pub fn create_provider_connections_exhausted_stream(cfg: &AppConfig, headers: &[(String, String)]) -> ProviderStreamResponse {
let video = cfg.t_custom_stream_response.as_ref().and_then(|c| c.provider_connections_exhausted.as_ref());
let custom_stream_response = cfg.custom_stream_response.load();
let video = custom_stream_response.as_ref().and_then(|c| c.provider_connections_exhausted.as_ref());
create_video_stream(video, headers, "Streaming response provider connections exhausted")
}
pub fn create_user_account_expired_stream(cfg: &AppConfig, headers: &[(String, String)]) -> ProviderStreamResponse {
let video = cfg.t_custom_stream_response.as_ref().and_then(|c| c.user_account_expired.as_ref());
let custom_stream_response = cfg.custom_stream_response.load();
let video = custom_stream_response.as_ref().and_then(|c| c.user_account_expired.as_ref());
create_video_stream(video, headers, "Streaming response user account expired")
}
backend/src/auth/authenticator.rs
+2 -2
View File
@@ -82,7 +82,7 @@ fn validate_request(
token: &str,
verify_fn: fn(&str, &[u8]) -> bool,
) -> Result<(), ()> {
if let Some(web_auth_config) = &app_state.config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) {
if let Some(web_auth_config) = &app_state.app_config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) {
let secret_key = web_auth_config.secret.as_ref();
if verify_fn(token, secret_key) {
return Ok(());
@@ -111,7 +111,7 @@ pub async fn validator_user(
next: axum::middleware::Next,
) -> Result<axum::response::Response, axum::http::StatusCode> {
if let Some(username) = get_username_from_auth_header(&token, &app_state) {
if let Some(user) = app_state.config.get_user_credentials(&username) {
if let Some(user) = app_state.app_config.get_user_credentials(&username) {
if !user.ui_enabled {
return Err(axum::http::StatusCode::FORBIDDEN);
}
backend/src/main.rs
+30 -29
View File
@@ -20,6 +20,7 @@ use log::{error, info};
use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use shared::model::ConfigPaths;
use shared::utils::set_sanitize_sensitive_info;
use crate::utils::init_logger;
@@ -92,12 +93,9 @@ fn main() {
return;
}
let config_path: String = utils::resolve_directory_path(&resolve_env_var(&args.config_path.unwrap_or_else(utils::get_default_config_path)));
let config_file: String = resolve_env_var(&args.config_file.unwrap_or_else(|| utils::get_default_config_file_path(&config_path)));
let api_proxy_file = resolve_env_var(&args.api_proxy.unwrap_or_else(|| utils::get_default_api_proxy_config_path(config_path.as_str())));
let mappings_file = args.mapping_file.as_ref();
let mut config_paths = get_file_paths(&args);
init_logger(args.log_level.as_ref(), config_file.as_str());
init_logger(args.log_level.as_ref(), config_paths.config_file_path.as_str());
info!("Version: {VERSION}");
if let Some(bts) = BUILD_TIMESTAMP.to_string().parse::<DateTime<Utc>>().ok().map(|datetime| datetime.format("%Y-%m-%d %H:%M:%S %Z").to_string()) {
@@ -105,13 +103,10 @@ fn main() {
}
if args.healthcheck {
healthcheck(config_file.as_str());
healthcheck(config_paths.config_file_path.as_str());
}
let sources_file: String = args.source_file.unwrap_or_else(|| utils::get_default_sources_file_path(&config_path));
let app_config = utils::read_config(config_path.as_str(), config_file.as_str(),
sources_file.as_str(), api_proxy_file.as_str(),
mappings_file.cloned(), true).unwrap_or_else(|err| exit!("{}", err));
let app_config = utils::read_app_config(&mut config_paths, true, true, args.server).unwrap_or_else(|err| exit!("{}", err));
let config = app_config.config.load();
set_sanitize_sensitive_info(config.log.as_ref().is_none_or(|l| l.sanitize_sensitive_info));
@@ -123,39 +118,28 @@ fn main() {
let sources = app_config.sources.load();
let targets = sources.validate_targets(args.target.as_ref()).unwrap_or_else(|err| exit!("{}", err));
let paths = app_config.paths.load();
info!("Current time: {}", chrono::offset::Local::now().format("%Y-%m-%d %H:%M:%S"));
info!("Temp dir: {}", temp_path.display());
info!("Working dir: {:?}", &config.working_dir);
info!("Config dir: {:?}", &app_config.t_config_path);
info!("Config file: {:?}", &app_config.t_config_file_path);
info!("Source file: {:?}", &app_config.t_sources_file_path);
info!("Api Proxy File: {:?}", &app_config.t_api_proxy_file_path);
match utils::read_mappings(&app_config.t_mapping_file_path, true) {
Ok(Some(mappings)) => {
info!("Mapping file: {:?}", &app_config.t_mapping_file_path);
app_config.set_mappings(&mappings);
}
Ok(None) => {
info!("Mapping file: not used");
},
Err(err) => exit!("{err}"),
}
info!("Config dir: {:?}", &paths.config_path);
info!("Config file: {:?}", &paths.config_file_path);
info!("Source file: {:?}", &paths.sources_file_path);
info!("Api Proxy File: {:?}", &paths.api_proxy_file_path);
info!("Mapping file: {:?}", &paths.mapping_file_path.as_ref().map_or_else(|| "not used", |v| v.as_str()));
if let Some(cache) = config.reverse_proxy.as_ref().and_then(|r| r.cache.as_ref()) {
if cache.enabled {
info!("Cache dir: {}", cache.dir);
}
}
if let Some(resource_path) = app_config.t_custom_stream_response_path.as_ref() {
if let Some(resource_path) = paths.custom_stream_response_path.as_ref() {
info!("Resource path: {resource_path}");
}
let rt = tokio::runtime::Runtime::new().unwrap();
let () = rt.block_on(async {
if args.server {
match utils::read_api_proxy_config(&app_config) {
Ok(()) => {}
Err(err) => exit!("{err}"),
}
start_in_server_mode(Arc::new(app_config), Arc::new(targets)).await;
} else {
start_in_cli_mode(Arc::new(app_config), Arc::new(targets)).await;
@@ -163,6 +147,23 @@ fn main() {
});
}
fn get_file_paths(args: &Args) -> ConfigPaths {
let config_path: String = utils::resolve_directory_path(&resolve_env_var(&args.config_path.as_ref().map_or_else(utils::get_default_config_path, ToString::to_string)));
let config_file: String = resolve_env_var(&args.config_file.as_ref().map_or_else(|| utils::get_default_config_file_path(&config_path), ToString::to_string));
let api_proxy_file = resolve_env_var(&args.api_proxy.as_ref().map_or_else(|| utils::get_default_api_proxy_config_path(config_path.as_str()), ToString::to_string));
let sources_file: String = resolve_env_var(&args.source_file.as_ref().map_or_else(|| utils::get_default_sources_file_path(&config_path), ToString::to_string));
let mappings_file = args.mapping_file.as_ref().map(|p| resolve_env_var(p));
ConfigPaths {
config_path,
config_file_path: config_file,
sources_file_path: sources_file,
mapping_file_path: mappings_file, // need to be set after config read
api_proxy_file_path: api_proxy_file,
custom_stream_response_path: None,
}
}
fn create_directories(cfg: &Config, temp_path: &Path) {
// Collect the paths into a vector.
let paths_strings = [
backend/src/model/config/api_proxy.rs
+3 -3
View File
@@ -7,7 +7,7 @@ use std::fs;
use std::sync::Arc;
use arc_swap::access::Access;
use arc_swap::ArcSwap;
use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, TargetUserDto};
use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, ConfigPaths, TargetUserDto};
use crate::{utils};
#[derive(Debug, Clone)]
@@ -98,6 +98,8 @@ impl ApiProxyConfig {
// When we switch from one to other we need to migrate the existing data.
/// # Panics
pub fn migrate_api_user(&mut self, cfg: &AppConfig, errors: &mut Vec<String>) {
let paths = <Arc<ArcSwap<ConfigPaths>> as Access<ConfigPaths>>::load(&cfg.paths);
let api_proxy_file = paths.api_proxy_file_path.as_str();
if self.use_user_db {
// we have user defined in config file.
// we migrate them to the db and delete them from the config file
@@ -105,7 +107,6 @@ impl ApiProxyConfig {
if let Err(err) = merge_api_user(cfg, &self.user) {
errors.push(err.to_string());
} else {
let api_proxy_file = cfg.t_api_proxy_file_path.as_str();
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&cfg.config);
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
self.user = vec![];
@@ -141,7 +142,6 @@ impl ApiProxyConfig {
}
}
}
let api_proxy_file = cfg.t_api_proxy_file_path.as_str();
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&cfg.config);
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
backend/src/model/config/api_user.rs
+2 -2
View File
@@ -77,7 +77,7 @@ impl ProxyUserCredentials {
}
pub fn has_permissions(&self, app_state: &AppState) -> bool {
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&app_state.config.config);
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&app_state.app_config.config);
if config.user_access_control {
if let Some(exp_date) = self.exp_date.as_ref() {
let now = Local::now();
@@ -103,7 +103,7 @@ impl ProxyUserCredentials {
}
pub async fn connection_permission(&self, app_state: &AppState) -> UserConnectionPermission {
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&app_state.config.config);
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&app_state.app_config.config);
if self.max_connections > 0 && config.user_access_control {
// we allow requests with max connection reached, but we should block streaming after grace period
return app_state.get_connection_permission(&self.username, self.max_connections).await;
backend/src/model/config/app.rs
+69 -51
View File
@@ -10,6 +10,7 @@ use log::{debug, error};
use rand::Rng;
use shared::create_tuliprox_error_result;
use shared::error::{TuliproxError, TuliproxErrorKind};
use shared::model::ConfigPaths;
use crate::api::model::streams::transport_stream_buffer::TransportStreamBuffer;
use crate::model::{ApiProxyConfig, ApiProxyServerInfo, Config, ConfigInput, ConfigInputOptions, ConfigTarget, CustomStreamResponse, HdHomeRunConfig, Mappings, ProxyUserCredentials, SourcesConfig, TargetOutput};
use crate::utils;
@@ -31,26 +32,52 @@ fn generate_secret() -> [u8; 32] {
pub struct AppConfig {
pub config: Arc<ArcSwap<Config>>,
pub sources: Arc<ArcSwap<SourcesConfig>>,
pub t_hdhomerun: Arc<ArcSwapOption<HdHomeRunConfig>>,
pub t_api_proxy: Arc<ArcSwapOption<ApiProxyConfig>>,
pub t_config_path: String,
pub t_config_file_path: String,
pub t_sources_file_path: String,
pub t_mapping_file_path: String,
pub t_api_proxy_file_path: String,
pub t_custom_stream_response_path: Option<String>,
pub hdhomerun: Arc<ArcSwapOption<HdHomeRunConfig>>,
pub api_proxy: Arc<ArcSwapOption<ApiProxyConfig>>,
pub file_locks: Arc<utils::FileLockManager>,
pub t_custom_stream_response: Option<CustomStreamResponse>,
pub t_access_token_secret: [u8; 32],
pub t_encrypt_secret: [u8; 16],
pub paths: Arc<ArcSwap<ConfigPaths>>,
pub custom_stream_response: Arc<ArcSwapOption<CustomStreamResponse>>,
pub access_token_secret: [u8; 32],
pub encrypt_secret: [u8; 16],
}
impl AppConfig {
pub fn set_api_proxy(&self, api_proxy: Option<Arc<ApiProxyConfig>>) -> Result<(), TuliproxError> {
self.t_api_proxy.store(api_proxy);
pub fn set_config(&self, config: Config) -> Result<(), TuliproxError> {
self.config.store(Arc::new(config));
self.prepare_custom_stream_response();
Ok(())
}
pub fn set_sources(&self, sources: SourcesConfig) -> Result<(), TuliproxError> {
self.sources.store(Arc::new(sources));
self.prepare_sources()?;
Ok(())
}
pub fn set_api_proxy(&self, api_proxy: ApiProxyConfig) -> Result<(), TuliproxError> {
self.api_proxy.store(Some(Arc::new(api_proxy)));
self.check_target_user()
}
pub fn set_mappings(&self, mappings_cfg: &Mappings) {
let sources = <Arc<ArcSwap<SourcesConfig>> as Access<SourcesConfig>>::load(&self.sources);
for source in &sources.sources {
for target in &source.targets {
if let Some(mapping_ids) = &target.mapping_ids {
let mut target_mappings = Vec::with_capacity(128);
for mapping_id in mapping_ids {
let mapping = mappings_cfg.get_mapping(mapping_id);
if let Some(mappings) = mapping {
target_mappings.push(mappings);
}
}
target.mapping.store(if target_mappings.is_empty() { None } else { Some(Arc::new(target_mappings)) });
}
}
}
}
fn check_username(&self, output_username: Option<&str>, target_name: &str) -> Result<(), TuliproxError> {
if let Some(username) = output_username {
if let Some((_, config_target)) = self.get_target_for_username(username) {
@@ -66,7 +93,7 @@ impl AppConfig {
}
}
fn check_target_user(&self) -> Result<(), TuliproxError> {
let check_homerun = self.t_hdhomerun.load().as_ref().is_some_and(|h| h.enabled);
let check_homerun = self.hdhomerun.load().as_ref().is_some_and(|h| h.enabled);
let sources = <Arc<ArcSwap<SourcesConfig>> as Access<SourcesConfig>>::load(&self.sources);
for source in &sources.sources {
for target in &source.targets {
@@ -80,7 +107,7 @@ impl AppConfig {
if check_homerun {
let hdhr_name = &hdhomerun_output.device;
self.check_username(Some(&hdhomerun_output.username), &target.name)?;
if let Some(old_hdhomerun) = self.t_hdhomerun.load().clone() {
if let Some(old_hdhomerun) = self.hdhomerun.load().clone() {
let mut hdhomerun = (*old_hdhomerun).clone();
for device in &mut hdhomerun.devices {
if &device.name == hdhr_name {
@@ -88,7 +115,7 @@ impl AppConfig {
device.t_enabled = true;
}
}
self.t_hdhomerun.store(Some(Arc::new(hdhomerun)));
self.hdhomerun.store(Some(Arc::new(hdhomerun)));
}
}
}
@@ -97,7 +124,7 @@ impl AppConfig {
}
}
let guard = self.t_hdhomerun.load();
let guard = self.hdhomerun.load();
if let Some(hdhomerun) = &*guard {
for device in &hdhomerun.devices {
if !device.t_enabled {
@@ -142,22 +169,22 @@ impl AppConfig {
pub fn get_target_for_username(&self, username: &str) -> Option<(ProxyUserCredentials, Arc<ConfigTarget>)> {
if let Some(credentials) = self.get_user_credentials(username) {
return self.t_api_proxy.load().as_ref()
return self.api_proxy.load().as_ref()
.and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name(&credentials.username, &credentials.password)));
}
None
}
pub fn get_target_for_user(&self, username: &str, password: &str) -> Option<(ProxyUserCredentials, Arc<ConfigTarget>)> {
self.t_api_proxy.load().as_ref().and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name(username, password)))
self.api_proxy.load().as_ref().and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name(username, password)))
}
pub fn get_target_for_user_by_token(&self, token: &str) -> Option<(ProxyUserCredentials, Arc<ConfigTarget>)> {
self.t_api_proxy.load().as_ref().as_ref().and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name_by_token(token)))
self.api_proxy.load().as_ref().as_ref().and_then(|api_proxy| self.intern_get_target_for_user(api_proxy.get_target_name_by_token(token)))
}
pub fn get_user_credentials(&self, username: &str) -> Option<ProxyUserCredentials> {
self.t_api_proxy.load().as_ref().as_ref().and_then(|api_proxy| api_proxy.get_user_credentials(username))
self.api_proxy.load().as_ref().as_ref().and_then(|api_proxy| api_proxy.get_user_credentials(username))
}
pub fn get_input_by_name(&self, input_name: &str) -> Option<Arc<ConfigInput>> {
@@ -201,24 +228,6 @@ impl AppConfig {
sources.get_target_by_id(target_id)
}
pub fn set_mappings(&self, mappings_cfg: &Mappings) {
let sources = <Arc<ArcSwap<SourcesConfig>> as Access<SourcesConfig>>::load(&self.sources);
for source in &sources.sources {
for target in &source.targets {
if let Some(mapping_ids) = &target.mapping_ids {
let mut target_mappings = Vec::with_capacity(128);
for mapping_id in mapping_ids {
let mapping = mappings_cfg.get_mapping(mapping_id);
if let Some(mappings) = mapping {
target_mappings.push(mappings);
}
}
target.mapping.store(if target_mappings.is_empty() { None } else { Some(Arc::new(target_mappings)) });
}
}
}
}
fn check_unique_input_names(&self) -> Result<(), TuliproxError> {
let mut seen_names = HashSet::new();
let sources = <Arc<ArcSwap<SourcesConfig>> as Access<SourcesConfig>>::load(&self.sources);
@@ -250,7 +259,7 @@ impl AppConfig {
}
fn check_scheduled_targets(&mut self, target_names: &HashSet<Cow<str>>) -> Result<(), TuliproxError> {
fn check_scheduled_targets(&self, target_names: &HashSet<Cow<str>>) -> Result<(), TuliproxError> {
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&self.config);
if let Some(schedules) = &config.schedules {
for schedule in schedules {
@@ -272,21 +281,25 @@ impl AppConfig {
pub fn prepare(&mut self, include_computed: bool) -> Result<(), TuliproxError> {
if include_computed {
self.t_access_token_secret = generate_secret();
self.t_encrypt_secret = <&[u8] as TryInto<[u8; 16]>>::try_into(&generate_secret()[0..16]).map_err(|err| TuliproxError::new(TuliproxErrorKind::Info, err.to_string()))?;
self.access_token_secret = generate_secret();
self.encrypt_secret = <&[u8] as TryInto<[u8; 16]>>::try_into(&generate_secret()[0..16]).map_err(|err| TuliproxError::new(TuliproxErrorKind::Info, err.to_string()))?;
self.prepare_custom_stream_response();
}
let sources = <Arc<ArcSwap<SourcesConfig>> as Access<SourcesConfig>>::load(&self.sources);
let target_names = sources.get_unique_target_names();
self.check_scheduled_targets(&target_names)?;
self.check_unique_input_names()?;
self.prepare_sources()?;
Ok(())
}
fn prepare_sources(&self) -> Result<(), TuliproxError> {
let sources = <Arc<ArcSwap<SourcesConfig>> as Access<SourcesConfig>>::load(&self.sources);
let target_names = sources.get_unique_target_names();
self.check_scheduled_targets(&target_names)?;
self.check_unique_input_names()?;
Ok(())
}
fn prepare_custom_stream_response(&mut self) {
fn prepare_custom_stream_response(&self) {
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&self.config);
if let Some(custom_stream_response_path) = config.custom_stream_response_path.as_ref() {
fn load_and_set_file(file_path: &Path) -> Option<TransportStreamBuffer> {
@@ -323,17 +336,22 @@ impl AppConfig {
let path = PathBuf::from(custom_stream_response_path);
let path = utils::make_path_absolute(&path, &config.working_dir);
self.t_custom_stream_response_path = Some(path.to_string_lossy().to_string());
let paths = self.paths.load_full();
let mut new_paths = paths.as_ref().clone();
new_paths.custom_stream_response_path = Some(path.to_string_lossy().to_string());
self.paths.store(Arc::new(new_paths));
let channel_unavailable = load_and_set_file(&path.join(CHANNEL_UNAVAILABLE));
let user_connections_exhausted = load_and_set_file(&path.join(USER_CONNECTIONS_EXHAUSTED));
let provider_connections_exhausted = load_and_set_file(&path.join(PROVIDER_CONNECTIONS_EXHAUSTED));
let user_account_expired = load_and_set_file(&path.join(USER_ACCOUNT_EXPIRED));
self.t_custom_stream_response = Some(CustomStreamResponse {
self.custom_stream_response.store(Some(Arc::new(CustomStreamResponse {
channel_unavailable,
user_connections_exhausted,
provider_connections_exhausted,
user_account_expired,
});
})));
}
}
@@ -342,7 +360,7 @@ impl AppConfig {
///
/// Will panic if default server invalid
pub fn get_server_info(&self, server_info_name: &str) -> ApiProxyServerInfo {
let guard = self.t_api_proxy.load();
let guard = self.api_proxy.load();
if let Ok(api_proxy) = guard.as_ref().ok_or_else(|| {
TuliproxError::new(TuliproxErrorKind::Info, "API proxy config not loaded".to_string())
}) {
backend/src/repository/user_repository.rs
+17 -15
View File
@@ -109,7 +109,8 @@ impl StoredProxyUserCredentials {
pub fn get_api_user_db_path(cfg: &AppConfig) -> PathBuf {
PathBuf::from(&cfg.t_config_path).join(storage_const::API_USER_DB_FILE)
let paths = cfg.paths.load();
PathBuf::from(&paths.config_path).join(storage_const::API_USER_DB_FILE)
}
@@ -397,10 +398,10 @@ pub async fn user_get_bouquet_filter(config: &Config, username: &str, category_i
#[cfg(test)]
mod tests {
use super::*;
use shared::model::{ProxyType, ProxyUserStatus};
use shared::model::{ConfigPaths, ProxyType, ProxyUserStatus};
use std::env::temp_dir;
use std::sync::Arc;
use arc_swap::ArcSwapAny;
use arc_swap::{ArcSwap, ArcSwapAny, ArcSwapOption};
use crate::utils::FileLockManager;
#[test]
@@ -471,21 +472,22 @@ mod tests {
let mut cfg = AppConfig {
config: Arc::new(ArcSwapAny::default()),
sources: Arc::new(ArcSwapAny::default()),
t_hdhomerun: Arc::new(ArcSwapAny::default()),
t_api_proxy: Arc::new(ArcSwapAny::default()),
t_config_path: String::new(),
t_config_file_path: String::new(),
t_sources_file_path: String::new(),
t_mapping_file_path: String::new(),
t_api_proxy_file_path: String::new(),
t_custom_stream_response_path: None,
hdhomerun: Arc::new(ArcSwapAny::default()),
api_proxy: Arc::new(ArcSwapAny::default()),
paths: Arc::new(ArcSwap::from(Arc::new(ConfigPaths {
config_path: temp_dir().to_string_lossy().to_string(),
config_file_path: "".to_string(),
sources_file_path: "".to_string(),
mapping_file_path: None,
api_proxy_file_path: "".to_string(),
custom_stream_response_path: None,
}))),
file_locks: Arc::new(FileLockManager::default()),
t_custom_stream_response: None,
t_access_token_secret: Default::default(),
t_encrypt_secret: Default::default(),
custom_stream_response: Arc::new(ArcSwapAny::default()),
access_token_secret: Default::default(),
encrypt_secret: Default::default(),
};
let target_user = vec![user];
cfg.t_config_path = temp_dir().to_string_lossy().to_string();
let _ = store_api_user(&cfg, &target_user);
let user_list = load_api_user(&cfg);
backend/src/utils/file/config_reader.rs
+132 -74
View File
@@ -1,21 +1,22 @@
use crate::model::Config;
use crate::model::{ApiProxyConfig, AppConfig, SourcesConfig};
use crate::model::{Config};
use shared::error::{create_tuliprox_error, info_err, to_io_error, TuliproxError, TuliproxErrorKind};
use crate::utils::{open_file, EnvResolvingReader, FileLockManager};
use crate::utils::{file_reader};
use crate::utils;
use crate::utils::file_reader;
use crate::utils::sys_utils::exit;
use shared::utils::CONSTANTS;
use crate::utils::{open_file, read_mappings_file, EnvResolvingReader, FileLockManager};
use arc_swap::{ArcSwap, ArcSwapAny};
use chrono::Local;
use log::{error, info, warn};
use serde::Serialize;
use shared::error::{create_tuliprox_error, info_err, to_io_error, TuliproxError, TuliproxErrorKind};
use shared::model::{ApiProxyConfigDto, AppConfigDto, ConfigDto, ConfigPaths, SourcesConfigDto};
use shared::utils::CONSTANTS;
use std::env;
use std::fs::File;
use std::io::{self, BufReader, Read};
use std::path::{PathBuf};
use std::path::PathBuf;
use std::sync::Arc;
use arc_swap::{ArcSwap, ArcSwapAny};
use shared::model::{ApiProxyConfigDto, ConfigDto, SourcesConfigDto};
use crate::utils;
use arc_swap::access::{Access};
enum EitherReader<L, R> {
Left(L),
@@ -41,32 +42,37 @@ pub fn config_file_reader(file: File, resolve_env: bool) -> impl Read
}
}
pub fn read_api_proxy_config(cfg: &AppConfig) -> Result<(), TuliproxError> {
let api_proxy_config = read_api_proxy(cfg, true);
match api_proxy_config {
None => {
warn!("cant read api_proxy_config file: {}", cfg.t_api_proxy_file_path.as_str());
Ok(())
}
Some(config) => {
cfg.set_api_proxy(Some(Arc::new(config)))?;
Ok(())
pub fn read_api_proxy_config(config: &AppConfig, resolve_env: bool) -> Result<Option<ApiProxyConfig>, TuliproxError> {
let paths = <Arc<ArcSwap<ConfigPaths>> as Access<ConfigPaths>>::load(&config.paths);
let api_proxy_file_path = paths.api_proxy_file_path.as_str();
if let Some(api_proxy_dto) = read_api_proxy_file(api_proxy_file_path, resolve_env)? {
let mut errors = vec![];
let mut api_proxy: ApiProxyConfig = ApiProxyConfig::from(&api_proxy_dto);
api_proxy.migrate_api_user(config, &mut errors);
if !errors.is_empty() {
for error in errors {
error!("{error}");
}
}
Ok(Some(api_proxy))
} else {
warn!("cant read api_proxy_config file: {api_proxy_file_path}");
Ok(None)
}
}
pub fn read_sources(sources_file: &str, resolve_env: bool, include_computed: bool) -> Result<SourcesConfigDto, TuliproxError> {
pub fn read_sources_file(sources_file: &str, resolve_env: bool, include_computed: bool) -> Result<SourcesConfigDto, TuliproxError> {
match open_file(&std::path::PathBuf::from(sources_file)) {
Ok(file) => {
let maybe_sources: Result<SourcesConfigDto, _> = serde_yaml::from_reader(config_file_reader(file, resolve_env));
match maybe_sources {
Ok(mut sources) => {
if let Err(err) = sources.prepare(include_computed) {
Err(info_err!(format!("Can't read the sources-config file: {sources_file}: {err}")))
} else {
Ok(sources)
if resolve_env {
if let Err(err) = sources.prepare(include_computed) {
return Err(info_err!(format!("Can't read the sources-config file: {sources_file}: {err}")));
}
}
Ok(sources)
}
Err(err) => Err(info_err!(format!("Can't read the sources-config file: {sources_file}: {err}")))
}
@@ -75,90 +81,142 @@ pub fn read_sources(sources_file: &str, resolve_env: bool, include_computed: boo
}
}
pub fn read_config(config_path: &str, config_file: &str, sources_file: &str, api_proxy_file: &str, mappings_file: Option<String>, include_computed: bool) -> Result<AppConfig, TuliproxError> {
let resolve_env = true;
let sources_dto = read_sources(sources_file, resolve_env, include_computed)?;
let config_dto = match open_file(&std::path::PathBuf::from(config_file)) {
pub fn read_config_file(config_file: &str, resolve_env: bool) -> Result<ConfigDto, TuliproxError> {
match open_file(&std::path::PathBuf::from(config_file)) {
Ok(file) => {
let maybe_config: Result<ConfigDto, _> = serde_yaml::from_reader(config_file_reader(file, resolve_env));
match maybe_config {
Ok(mut config) => {
config.prepare()?;
config
},
Err(err) => return Err(info_err!(format!("Can't read the config file: {config_file}: {err}")))
Ok(mut config) => {
if resolve_env {
config.prepare()?;
}
Ok(config)
}
Err(err) => Err(info_err!(format!("Can't read the config file: {config_file}: {err}")))
}
}
Err(err) => return Err(info_err!(format!("Can't read the config file: {config_file}: {err}")))
Err(err) => Err(info_err!(format!("Can't read the config file: {config_file}: {err}")))
}
}
pub fn read_app_config_dto(paths: &ConfigPaths,
resolve_env: bool,
include_computed: bool) -> Result<AppConfigDto, TuliproxError> {
let config_file = paths.config_file_path.as_str();
let sources_file = paths.sources_file_path.as_str();
let api_proxy_file = paths.api_proxy_file_path.as_str();
let sources = read_sources_file(sources_file, resolve_env, include_computed)?;
let config = read_config_file(config_file, resolve_env)?;
let mappings = if let Some(mappings_file) = paths.mapping_file_path.as_ref() {
read_mappings_file(mappings_file, resolve_env).unwrap_or(None)
} else {
None
};
let api_proxy = read_api_proxy_file(api_proxy_file, resolve_env).unwrap_or(None);
Ok(AppConfigDto {
config,
sources,
mappings,
api_proxy,
})
}
pub fn read_app_config(paths: &mut ConfigPaths,
resolve_env: bool,
include_computed: bool,
server_mode: bool) -> Result<AppConfig, TuliproxError> {
let config_path = paths.config_path.as_str();
let config_file = paths.config_file_path.as_str();
let sources_file = paths.sources_file_path.as_str();
let sources_dto = read_sources_file(sources_file, resolve_env, include_computed)?;
let sources: SourcesConfig = sources_dto.into();
let config_dto = read_config_file(config_file, resolve_env)?;
let mut config: Config = Config::from(config_dto);
config.prepare(config_path)?;
let mappings_file_path = config.mapping_path.as_ref().map_or_else(String::new, |mp| mp.trim().to_string());
let mappings_file_path = if mappings_file_path.is_empty() {
resolve_env_var(&mappings_file.unwrap_or_else(|| utils::get_default_mappings_path(config_path)))
} else {
mappings_file_path
};
if paths.mapping_file_path.is_none() {
let mut path = config.mapping_path.as_ref().map_or_else(|| utils::get_default_mappings_path(config_path), ToString::to_string);
if resolve_env {
path = resolve_env_var(&path);
}
paths.mapping_file_path.replace(path);
}
let mut app_config = AppConfig {
config: Arc::new(ArcSwap::from_pointee(config)),
sources: Arc::new(ArcSwap::from_pointee(sources)),
t_hdhomerun: Arc::new(ArcSwapAny::default()),
t_api_proxy: Arc::new(ArcSwapAny::default()),
t_config_path: config_path.to_string(),
t_config_file_path: config_file.to_string(),
t_sources_file_path: sources_file.to_string(),
t_mapping_file_path: mappings_file_path,
t_api_proxy_file_path: api_proxy_file.to_string(),
t_custom_stream_response_path: None,
hdhomerun: Arc::new(ArcSwapAny::default()),
api_proxy: Arc::new(ArcSwapAny::default()),
paths: Arc::new(ArcSwap::from_pointee(paths.clone())),
file_locks: Arc::new(FileLockManager::default()),
t_custom_stream_response: None,
t_access_token_secret: Default::default(),
t_encrypt_secret: Default::default(),
custom_stream_response: Arc::new(ArcSwapAny::default()),
access_token_secret: Default::default(),
encrypt_secret: Default::default(),
};
app_config.prepare(include_computed)?;
if let Some(mappings_file) = &paths.mapping_file_path {
match utils::read_mappings(mappings_file.as_str(), resolve_env) {
Ok(Some(mappings)) => app_config.set_mappings(&mappings),
Ok(None) => info!("Mapping file: not used"),
Err(err) => exit!("{err}"),
}
}
if server_mode {
match read_api_proxy_config(&app_config, resolve_env) {
Ok(Some(api_proxy)) => app_config.set_api_proxy(api_proxy)?,
Ok(None) => info!("Api-Proxy file: not used"),
Err(err) => exit!("{err}"),
}
}
Ok(app_config)
}
pub fn read_api_proxy_dto(config: &AppConfig, resolve_env: bool) -> Option<ApiProxyConfigDto> {
let api_proxy_file = config.t_api_proxy_file_path.as_str();
open_file(&std::path::PathBuf::from(api_proxy_file)).map_or(None, |file| {
pub fn read_api_proxy_file(api_proxy_file: &str, resolve_env: bool) -> Result<Option<ApiProxyConfigDto>, TuliproxError> {
open_file(&std::path::PathBuf::from(api_proxy_file)).map_or(Ok(None), |file| {
let maybe_api_proxy: Result<ApiProxyConfigDto, _> = serde_yaml::from_reader(config_file_reader(file, resolve_env));
match maybe_api_proxy {
Ok(mut api_proxy_dto) => {
if let Err(err) = api_proxy_dto.prepare() {
exit!("cant read api-proxy-config file: {err}");
} else {
Some(api_proxy_dto)
if resolve_env {
if let Err(err) = api_proxy_dto.prepare() {
exit!("cant read api-proxy-config file: {err}");
}
}
Ok(Some(api_proxy_dto))
}
Err(err) => {
error!("cant read api-proxy-config file: {err}");
None
Err(info_err!(format!("cant read api-proxy-config file: {err}")))
}
}
})
}
pub fn read_api_proxy(config: &AppConfig, resolve_env: bool) -> Option<ApiProxyConfig> {
if let Some(api_proxy_dto) = read_api_proxy_dto(config, resolve_env) {
let mut errors = vec![];
let mut api_proxy: ApiProxyConfig = api_proxy_dto.into();
api_proxy.migrate_api_user(config, &mut errors);
if !errors.is_empty() {
for error in errors {
error!("{error}");
let paths = <Arc<ArcSwap<ConfigPaths>> as Access<ConfigPaths>>::load(&config.paths);
match read_api_proxy_file(paths.api_proxy_file_path.as_str(), resolve_env) {
Ok(Some(api_proxy_dto)) => {
let mut errors = vec![];
let mut api_proxy: ApiProxyConfig = api_proxy_dto.into();
api_proxy.migrate_api_user(config, &mut errors);
if !errors.is_empty() {
for error in errors {
error!("{error}");
}
}
Some(api_proxy)
}
Ok(None) => None,
Err(err) => {
error!("Failed to read Api-Proxy file {err}");
None
}
Some(api_proxy)
} else {
None
}
}
backend/src/utils/file/mapping_reader.rs
+10 -3
View File
@@ -113,14 +113,14 @@ fn read_mappings_from_directory(path: &Path, resolve_env: bool) -> Result<Option
merge_mapping_definitions(mappings)
}
pub fn read_mappings(mappings_file: &str, resolve_env: bool) -> Result<Option<Mappings>, TuliproxError> {
pub fn read_mappings_file(mappings_file: &str, resolve_env: bool) -> Result<Option<MappingsDto>, TuliproxError> {
let path = PathBuf::from(mappings_file);
match std::fs::metadata(&path) {
Ok(metadata) => {
if metadata.is_file() {
read_mappings_from_file(&path, resolve_env).map(|s| s.map(Into::into))
read_mappings_from_file(&path, resolve_env)
} else if metadata.is_dir() {
read_mappings_from_directory(&path, resolve_env).map(|s| s.map(Into::into))
read_mappings_from_directory(&path, resolve_env)
} else {
Ok(None)
}
@@ -129,4 +129,11 @@ pub fn read_mappings(mappings_file: &str, resolve_env: bool) -> Result<Option<Ma
Ok(None)
}
}
}
pub fn read_mappings(mappings_file: &str, resolve_env: bool) -> Result<Option<Mappings>, TuliproxError> {
match read_mappings_file(mappings_file, resolve_env)? {
Some(dto) => Ok(Some(Mappings::from(&dto))),
None => Ok(None),
}
}
shared/src/foundation/mapper.rs
+4 -4
View File
@@ -597,7 +597,7 @@ impl<'a> MapperContext<'a> {
Expression::Identifier(ident)
| Expression::VarAccess(ident, _) => {
if !identifiers.contains(ident.as_str()) {
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Identifier unknown {}", ident);
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Identifier unknown {}, {:?}", ident, expr);
}
}
Expression::NullValue
@@ -608,7 +608,7 @@ impl<'a> MapperContext<'a> {
match field {
RegexSource::Identifier(ident) => {
if !identifiers.contains(ident.as_str()) {
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Identifier unknown {}", ident);
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Regex identifier unknown {}, {:?}", ident, expr);
}
}
RegexSource::Field(_) => {}
@@ -665,7 +665,7 @@ impl<'a> MapperContext<'a> {
match identifier {
MatchCaseKey::Identifier(ident) => {
if !identifiers.contains(ident.as_str()) {
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Identifier unknown {}", ident);
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Match case identifier unknown {}", ident);
}
identifier_key.push_str(ident.as_str());
identifier_key.push_str(", ");
@@ -693,7 +693,7 @@ impl<'a> MapperContext<'a> {
MapKey::Identifier(ident)
| MapKey::VarAccess(ident, _) => {
if !identifiers.contains(ident.as_str()) {
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Identifier unknown {}", ident);
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Map key identifier unknown {}", ident);
}
}
MapKey::FieldAccess(_) => {}
shared/src/model/config/app_config.rs
+10
View File
@@ -0,0 +1,10 @@
use crate::model::{ApiProxyConfigDto, ConfigDto, MappingsDto, SourcesConfigDto};
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
#[serde(deny_unknown_fields)]
pub struct AppConfigDto {
pub config: ConfigDto,
pub sources: SourcesConfigDto,
pub mappings: Option<MappingsDto>,
pub api_proxy: Option<ApiProxyConfigDto>,
}
shared/src/model/config/mod.rs
+5 -1
View File
@@ -25,6 +25,8 @@ mod api;
mod epg_smart_match;
mod pattern_template;
pub mod macros;
mod paths;
mod app_config;
pub use base::*;
pub use api::*;
@@ -51,4 +53,6 @@ pub use reverse_proxy::*;
pub use proxy::*;
pub use trakt::*;
pub use rename::*;
pub use pattern_template::*;
pub use pattern_template::*;
pub use paths::*;
pub use app_config::*;
shared/src/model/config/paths.rs
+10
View File
@@ -0,0 +1,10 @@
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ConfigPaths {
pub config_path: String,
pub config_file_path: String,
pub sources_file_path: String,
pub mapping_file_path: Option<String>,
pub api_proxy_file_path: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub custom_stream_response_path: Option<String>,
}