diff --git a/frontend/src/model/server-config.ts b/frontend/src/model/server-config.ts index 70cb5c3af..52a1e1266 100644 --- a/frontend/src/model/server-config.ts +++ b/frontend/src/model/server-config.ts @@ -140,6 +140,12 @@ export interface Credentials { password: string; token: string; proxy: 'redirect' | 'reverse'; + created_at: number, + exp_date: number, + is_trial: String, + max_connections: String, + message: String, + status: String, } export interface TargetUser { diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index 424507489..84dcf8b62 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -18,7 +18,7 @@ use actix_files::NamedFile; use actix_web::body::{BodyStream, SizedStream}; use actix_web::http::header::{HeaderValue, CACHE_CONTROL}; use actix_web::{HttpRequest, HttpResponse}; -use parking_lot::FairMutex; +use parking_lot::{Mutex, RwLock}; use futures::{TryStreamExt}; use log::{error, log_enabled, trace}; use reqwest::StatusCode; @@ -132,12 +132,14 @@ fn get_stream_content_length(provider_response: Option<&(Vec<(String, String)>, pub async fn stream_response(app_state: &AppState, stream_url: &str, req: &HttpRequest, input: Option<&ConfigInput>, - item_type: PlaylistItemType, target: &ConfigTarget) -> HttpResponse { + item_type: PlaylistItemType, target: &ConfigTarget, + user: &ProxyUserCredentials) -> HttpResponse { if log_enabled!(log::Level::Trace) { trace!("Try to open stream {}", sanitize_sensitive_info(stream_url)); } + let log_active_clients = app_state.config.log.as_ref().is_some_and(|l| l.active_clients); let share_stream = is_stream_share_enabled(item_type, target); if share_stream { - if let Some(value) = shared_stream_response(app_state, stream_url) { + if let Some(value) = shared_stream_response(app_state, stream_url, log_active_clients, user) { return value; } } @@ -146,7 +148,7 @@ pub async fn stream_response(app_state: &AppState, stream_url: &str, get_stream_options(app_state); if let Ok(url) = Url::parse(stream_url) { - let active_clients = Arc::clone(&app_state.active_clients); + let active_clients = Arc::clone(&app_state.active_users); let (stream_opt, provider_response) = if direct_pipe_provider_stream { get_provider_pipe_stream(&app_state.http_client, &url, req, input, item_type).await } else { @@ -155,8 +157,7 @@ pub async fn stream_response(app_state: &AppState, stream_url: &str, }; if let Some(stream) = stream_opt { let content_length = get_stream_content_length(provider_response.as_ref()); - let log_active_clients = app_state.config.log.as_ref().is_some_and(|l| l.active_clients); - let stream = ActiveClientStream::new(stream, active_clients, log_active_clients); + let stream = ActiveClientStream::new(stream, active_clients, user, log_active_clients); let stream_resp = if share_stream { let shared_headers = provider_response.as_ref().map_or_else(Vec::new, |(h, _)| h.clone()); SharedStreamManager::subscribe(app_state, stream_url, stream, shared_stream_use_own_buffer, shared_headers); @@ -182,11 +183,13 @@ pub async fn stream_response(app_state: &AppState, stream_url: &str, HttpResponse::BadRequest().finish() } -fn shared_stream_response(app_state: &AppState, stream_url: &str) -> Option { +fn shared_stream_response(app_state: &AppState, stream_url: &str, log_active_clients: bool, user: &ProxyUserCredentials) -> Option { if let Some(stream) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url) { debug_if_enabled!("Using shared channel {}", sanitize_sensitive_info(stream_url)); if let Some(headers) = app_state.shared_stream_manager.lock().get_shared_state_headers(stream_url) { let mut response_builder = get_stream_response_with_headers(Some((headers.clone(), StatusCode::OK)), stream_url); + let active_clients = Arc::clone(&app_state.active_users); + let stream = ActiveClientStream::new(stream, active_clients, user, log_active_clients); return Some(response_builder.body(BodyStream::new(stream))); } } @@ -209,7 +212,7 @@ pub fn get_headers_from_request(req: &HttpRequest, filter: &HeaderFilter) -> Has .collect() } -fn get_add_cache_content(res_url: &str, cache: &Arc>>) -> Box { +fn get_add_cache_content(res_url: &str, cache: &Arc>>) -> Box { let resource_url = String::from(res_url); let cache = Arc::clone(cache); let add_cache_content: Box = Box::new(move |size| { @@ -255,8 +258,7 @@ pub async fn resource_response(app_state: &AppState, resource_url: &str, req: &H let byte_stream = response.bytes_stream().map_err(|err| StreamError::reqwest(&err)); if let Some(cache) = app_state.cache.as_ref() { let resource_path = { - let guard = cache.lock(); - guard.store_path(resource_url) + cache.lock().store_path(resource_url) }; if let Ok(file) = create_new_file_for_write(&resource_path) { let writer = Arc::new(file); diff --git a/src/api/endpoints/hls_api.rs b/src/api/endpoints/hls_api.rs index f4ca2c479..12b535c73 100644 --- a/src/api/endpoints/hls_api.rs +++ b/src/api/endpoints/hls_api.rs @@ -37,10 +37,14 @@ async fn hls_api_stream( target_type: TargetType ) -> HttpResponse { let (_token, username, password, channel, _hash, _chunk) = path.into_inner(); - let (_user, target) = try_option_bad_request!( + let (user, target) = try_option_bad_request!( get_user_target_by_credentials(&username, &password, api_req, app_state).await, false, format!("Could not find any user {username}")); + if !user.is_active() { + debug!("User access denied: {user:?}"); + return HttpResponse::Forbidden().finish(); + } let target_name = &target.name; let virtual_id: u32 = try_result_bad_request!(channel.parse()); @@ -59,7 +63,7 @@ async fn hls_api_stream( // we don't respond as hlsr, we take the original stream, because the location could be different and then it does not work // The next problem is, different url to same channel causes to fail stream share. // let stream_url = format!("{input_url}/hlsr/{token}/{input_username}/{input_password}/{}/{hash}/{chunk}", pli.provider_id); - stream_response(app_state, &pli_url, req, Some(input), PlaylistItemType::Live, target).await + stream_response(app_state, &pli_url, req, Some(input), PlaylistItemType::Live, target, &user).await } async fn hls_api_stream_xtream( diff --git a/src/api/endpoints/m3u_api.rs b/src/api/endpoints/m3u_api.rs index 6ec41528e..a812a5395 100644 --- a/src/api/endpoints/m3u_api.rs +++ b/src/api/endpoints/m3u_api.rs @@ -68,6 +68,10 @@ async fn m3u_api_stream( let virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse()); let Some((user, target)) = get_user_target_by_credentials(&username, &password, &api_req, &app_state).await else { return HttpResponse::BadRequest().finish() }; + if !user.is_active() { + debug!("User access denied: {user:?}"); + return HttpResponse::Forbidden().finish(); + } if !target.has_output(&TargetType::M3u) { return HttpResponse::BadRequest().finish(); @@ -96,7 +100,7 @@ async fn m3u_api_stream( return handle_hls_stream_request(&app_state, &user, &m3u_item, input, TargetType::M3u).await; } - stream_response(&app_state, m3u_item.url.as_str(), &req, None, m3u_item.item_type, target).await + stream_response(&app_state, m3u_item.url.as_str(), &req, None, m3u_item.item_type, target, &user).await } async fn m3u_api_resource( @@ -109,6 +113,10 @@ async fn m3u_api_resource( let Ok(m3u_stream_id) = stream_id.parse::() else { return HttpResponse::BadRequest().finish() }; let Some((user, target)) = get_user_target_by_credentials(&username, &password, &api_req, &app_state).await else { return HttpResponse::BadRequest().finish() }; + if !user.is_active() { + debug!("User access denied: {user:?}"); + return HttpResponse::Forbidden().finish(); + } if !target.has_output(&TargetType::M3u) { return HttpResponse::BadRequest().finish(); diff --git a/src/api/endpoints/xmltv_api.rs b/src/api/endpoints/xmltv_api.rs index ef7bc9264..36e88f151 100644 --- a/src/api/endpoints/xmltv_api.rs +++ b/src/api/endpoints/xmltv_api.rs @@ -2,7 +2,7 @@ use std::fs::File; use std::path::{Path, PathBuf}; use actix_web::{http::header, web, HttpRequest, HttpResponse}; -use log::{error, trace}; +use log::{debug, error, trace}; use quick_xml::{Reader, Writer}; use flate2::write::GzEncoder; use flate2::Compression; @@ -164,6 +164,10 @@ async fn xmltv_api( app_state: web::Data, ) -> HttpResponse { if let Some((user, target)) = get_user_target(&api_req, &app_state).await { + if !user.is_active() { + debug!("User access denied: {user:?}"); + return HttpResponse::Forbidden().finish(); + } match get_epg_path_for_target(&app_state.config, target) { None => { // No epg configured, No processing or timeshift, epg can't be mapped to the channels. diff --git a/src/api/endpoints/xtream_api.rs b/src/api/endpoints/xtream_api.rs index 0daef24e7..7e34d458e 100644 --- a/src/api/endpoints/xtream_api.rs +++ b/src/api/endpoints/xtream_api.rs @@ -144,6 +144,10 @@ async fn xtream_player_api_stream( stream_req: XtreamApiStreamRequest<'_>, ) -> HttpResponse { let (user, target) = try_option_bad_request!(get_user_target_by_credentials(stream_req.username, stream_req.password, api_req, app_state).await, false, format!("Could not find any user {}", stream_req.username)); + if !user.is_active() { + debug!("User access denied: {user:?}"); + return HttpResponse::Forbidden().finish(); + } let target_name = &target.name; if !target.has_output(&TargetType::Xtream) { debug!("Target has no xtream output {}", target_name); @@ -196,7 +200,7 @@ async fn xtream_player_api_stream( stream_req.context)); trace_if_enabled!("Streaming stream request from {}", sanitize_sensitive_info(&stream_url)); - stream_response(app_state, &stream_url, req, Some(input), pli.item_type, target).await + stream_response(app_state, &stream_url, req, Some(input), pli.item_type, target, &user).await } @@ -332,6 +336,10 @@ async fn xtream_player_api_resource( resource_req: XtreamApiStreamRequest<'_>, ) -> HttpResponse { let (user, target) = try_option_bad_request!(get_user_target_by_credentials(resource_req.username, resource_req.password, api_req, app_state).await, false, format!("Could not find any user {}", resource_req.username)); + if !user.is_active() { + debug!("User access denied: {user:?}"); + return HttpResponse::Forbidden().finish(); + } let target_name = &target.name; if !target.has_output(&TargetType::Xtream) { debug!("Target has no xtream output {}", target_name); @@ -584,6 +592,11 @@ async fn xtream_player_api( return HttpResponse::Ok().json(get_user_info(&user, &app_state.config)); } + if !user.is_active() { + debug!("User access denied: {user:?}"); + return HttpResponse::Forbidden().finish(); + } + // Process specific playlist actions let (skip_live, skip_vod, skip_series) = if let Some(inputs) = app_state.config.get_inputs_for_target(&target.name) { inputs.iter().fold((true, true, true), |acc, i| { diff --git a/src/api/main_api.rs b/src/api/main_api.rs index 94a4c1f15..cb89be28a 100644 --- a/src/api/main_api.rs +++ b/src/api/main_api.rs @@ -2,14 +2,13 @@ use actix_cors::Cors; use actix_web::middleware::Logger; use actix_web::web::Data; use actix_web::{web, App, HttpResponse, HttpServer}; -use parking_lot::{FairMutex}; +use parking_lot::{Mutex as PlMutex, RwLock as PlRwLock}; use tokio::sync::{RwLock, Mutex}; use log::{error, info}; use std::collections::{VecDeque}; use std::io::ErrorKind; use std::path::PathBuf; use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; use crate::api::endpoints::hls_api::hls_api_register; use crate::api::endpoints::m3u_api::m3u_api_register; use crate::api::model::app_state::AppState; @@ -20,6 +19,7 @@ use crate::api::endpoints::v1_api::v1_api_register; use crate::api::endpoints::web_index::index_register; use crate::api::endpoints::xmltv_api::xmltv_api_register; use crate::api::endpoints::xtream_api::xtream_api_register; +use crate::api::model::active_user_manager::ActiveUserManager; use crate::model::config::{validate_targets, Config, ProcessTargets, ScheduleConfig}; use crate::model::healthcheck::Healthcheck; use crate::processing::processor::playlist; @@ -40,18 +40,23 @@ fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result,) -> HttpResponse { let ts = chrono::offset::Local::now().format("%Y-%m-%d %H:%M:%S").to_string(); + let (active_clients, active_connections) = { + let active_user = app_state.active_users.read(); + (active_user.active_users(), active_user.active_connections()) + }; HttpResponse::Ok().json(Healthcheck { status: "ok".to_string(), version: VERSION.to_string(), time: ts, mem: sys_utils::get_memory_usage().map_or(String::from("?"), human_readable_byte_size), - active_clients: app_state.active_clients.as_ref().load(Ordering::Relaxed) + active_clients, + active_connections, }) } fn create_shared_data(cfg: &Arc) -> Data { let lru_cache = cfg.reverse_proxy.as_ref().and_then(|r| r.cache.as_ref()).and_then(|c| if c.enabled { - Some(FairMutex::new(LRUResourceCache::new(c.t_size, &PathBuf::from(c.dir.as_ref().unwrap())))) + Some(PlMutex::new(LRUResourceCache::new(c.t_size, &PathBuf::from(c.dir.as_ref().unwrap())))) } else { None} ); let cache = Arc::new(lru_cache); let cache_scanner = Arc::clone(&cache); @@ -70,8 +75,8 @@ fn create_shared_data(cfg: &Arc) -> Data { active: Arc::from(RwLock::new(None)), finished: Arc::from(RwLock::new(Vec::new())), }), - active_clients: Arc::new(AtomicUsize::new(0)), - shared_stream_manager: Arc::new(FairMutex::new(SharedStreamManager::new())), + shared_stream_manager: Arc::new(PlMutex::new(SharedStreamManager::new())), + active_users: Arc::new(PlRwLock::new(ActiveUserManager::new())), http_client: Arc::new(reqwest::Client::new()), cache, }) diff --git a/src/api/model/active_user_manager.rs b/src/api/model/active_user_manager.rs new file mode 100644 index 000000000..5ae9675b6 --- /dev/null +++ b/src/api/model/active_user_manager.rs @@ -0,0 +1,51 @@ +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, Ordering}; + +pub struct ActiveUserManager { + pub user: HashMap, +} + +impl Default for ActiveUserManager { + fn default() -> Self { + Self::new() + } +} + +impl ActiveUserManager { + pub fn new() -> Self { + Self { + user: HashMap::new(), + } + } + + pub fn user_connections(&self, username: &str) -> u32 { + if let Some(counter) = self.user.get(username) { + return counter.load(std::sync::atomic::Ordering::Relaxed); + } + 0 + } + + pub fn active_users(&self) -> usize { + self.user.len() + } + + pub fn active_connections(&self) -> usize { + self.user.values().map(|c| c.load(Ordering::Relaxed) as usize).sum() + } + + pub fn add_connection(&mut self, username: &str) { + if let Some(counter) = self.user.get(username) { + counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } else { + self.user.insert(username.to_string(), AtomicU32::new(1)); + } + } + + pub fn remove_connection(&mut self, username: &str) { + if let Some(counter) = self.user.get(username) { + if counter.fetch_sub(1, std::sync::atomic::Ordering::Relaxed) == 1 { + self.user.remove(username); + } + } + } +} \ No newline at end of file diff --git a/src/api/model/app_state.rs b/src/api/model/app_state.rs index 9cadd91f1..a0bf58b4f 100644 --- a/src/api/model/app_state.rs +++ b/src/api/model/app_state.rs @@ -1,6 +1,6 @@ -use std::sync::Arc; -use std::sync::atomic::AtomicUsize; -use parking_lot::FairMutex; +use std::sync::{Arc}; +use parking_lot::{Mutex, RwLock}; +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::config::{Config}; @@ -9,8 +9,8 @@ use crate::tools::lru_cache::LRUResourceCache; pub struct AppState { pub config: Arc, pub downloads: Arc, - pub shared_stream_manager: Arc>, - pub active_clients: Arc, + pub shared_stream_manager: Arc>, pub http_client: Arc, - pub cache: Arc>> + pub cache: Arc>>, + pub active_users: Arc>, } diff --git a/src/api/model/mod.rs b/src/api/model/mod.rs index 8bb96a2e8..08e2f76ec 100644 --- a/src/api/model/mod.rs +++ b/src/api/model/mod.rs @@ -6,3 +6,4 @@ pub(in crate::api) mod xtream; pub(in crate::api) mod model_utils; pub(in crate::api) mod stream_error; pub(crate) mod streams; +pub(in crate::api) mod active_user_manager; diff --git a/src/api/model/streams/active_client_stream.rs b/src/api/model/streams/active_client_stream.rs index 67a658035..9e58fb7e2 100644 --- a/src/api/model/streams/active_client_stream.rs +++ b/src/api/model/streams/active_client_stream.rs @@ -1,26 +1,33 @@ -use crate::api::model::streams::provider_stream_factory::ResponseStream; -use bytes::Bytes; -use std::pin::Pin; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Arc; -use std::task::{Poll}; -use futures::{Stream}; -use log::info; +use crate::api::model::active_user_manager::ActiveUserManager; use crate::api::model::stream_error::StreamError; +use crate::api::model::streams::provider_stream_factory::ResponseStream; +use crate::model::api_proxy::ProxyUserCredentials; +use bytes::Bytes; +use futures::Stream; +use log::info; +use parking_lot::{RwLock}; +use std::pin::Pin; +use std::sync::Arc; +use std::task::Poll; pub(in crate::api) struct ActiveClientStream { inner: ResponseStream, - active_clients: Arc, + active_clients: Arc>, log_active_clients: bool, + username: String, } impl ActiveClientStream { - pub(crate) fn new(inner: ResponseStream, active_clients: Arc, log_active_clients: bool) -> Self { - let client_count = active_clients.fetch_add(1, Ordering::Relaxed) + 1; + pub(crate) fn new(inner: ResponseStream, active_clients: Arc>, user: &ProxyUserCredentials, log_active_clients: bool) -> Self { + let client_count = { + let mut clients = active_clients.write(); + clients.add_connection(&user.username); + clients.active_users() + }; if log_active_clients { info!("Active clients: {client_count}"); } - Self { inner, active_clients, log_active_clients } + Self { inner, active_clients, log_active_clients, username: user.username.clone() } } } impl Stream for ActiveClientStream { @@ -30,14 +37,18 @@ impl Stream for ActiveClientStream { mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> Poll> { - Pin::as_mut(&mut self.inner).poll_next(cx) + Pin::as_mut(&mut self.inner).poll_next(cx) } } impl Drop for ActiveClientStream { fn drop(&mut self) { - let client_count = self.active_clients.fetch_sub(1, Ordering::Relaxed) -1; + let client_count = { + let mut clients = self.active_clients.write(); + clients.remove_connection(&self.username); + clients.active_users() + }; if self.log_active_clients { info!("Active clients: {client_count}"); } diff --git a/src/api/model/streams/shared_stream_manager.rs b/src/api/model/streams/shared_stream_manager.rs index bab976c77..71aa527ad 100644 --- a/src/api/model/streams/shared_stream_manager.rs +++ b/src/api/model/streams/shared_stream_manager.rs @@ -3,7 +3,7 @@ use crate::api::model::streams::provider_stream_factory::STREAM_QUEUE_SIZE; use crate::api::model::stream_error::StreamError; use crate::utils::debug_if_enabled; use crate::utils::network::request::sanitize_sensitive_info; -use parking_lot::FairMutex; +use parking_lot::{FairMutex, Mutex}; use bytes::Bytes; use futures::stream::BoxStream; use futures::{Stream, StreamExt}; @@ -71,7 +71,7 @@ impl SharedStreamState { convert_stream(ReceiverStream::new(rx).boxed()) } - fn broadcast(&self, stream_url: &str, bytes_stream: S, shared_streams: Arc>) + fn broadcast(&self, stream_url: &str, bytes_stream: S, shared_streams: Arc>) where S: Stream> + Unpin + 'static, { diff --git a/src/api/model/xtream.rs b/src/api/model/xtream.rs index a4bd01d06..0447d7531 100644 --- a/src/api/model/xtream.rs +++ b/src/api/model/xtream.rs @@ -1,20 +1,15 @@ -use crate::model::api_proxy::{ApiProxyServerInfo, ProxyUserCredentials}; -use serde::Serialize; +use crate::model::api_proxy::{ApiProxyServerInfo, ProxyUserCredentials, ProxyUserStatus}; use chrono::{Duration, Local}; +use serde::Serialize; #[derive(Serialize)] pub struct XtreamUserInfo { pub active_cons: String, - pub allowed_output_formats: Vec, - //["ts"], - pub auth: u16, - // 0 | 1 - pub created_at: i64, - //1623429679, - pub exp_date: i64, - //1628755200, - pub is_trial: String, - // 0 | 1 + pub allowed_output_formats: Vec, //["ts"], + pub auth: u16, // 0 | 1 + pub created_at: i64, //1623429679, + pub exp_date: i64, //1628755200, + pub is_trial: String, // 0 | 1 pub max_connections: String, pub message: String, pub password: String, @@ -54,19 +49,26 @@ pub struct XtreamServerInfoDto { impl XtreamAuthorizationResponse { pub fn new(server_info: &ApiProxyServerInfo, user: &ProxyUserCredentials) -> Self { let now = Local::now(); + let created_at = user.created_at.as_ref().map_or_else(|| (now - Duration::days(365)).timestamp(), |d| *d); + let exp_date = user.exp_date.as_ref().map_or_else(|| (now + Duration::days(365)).timestamp(), |d| *d); + + let is_expired = (exp_date - now.timestamp()) < 0; + let is_trial = user.status.as_ref().map_or("0", |s| if *s == ProxyUserStatus::Trial { "1" } else { "0" }).to_string(); + let max_connections = user.max_connections.as_ref().map_or("1", |s| s).to_string(); + let user_status = if is_expired { &ProxyUserStatus::Expired } else { user.status.as_ref().unwrap_or(&ProxyUserStatus::Active) }; Self { user_info: XtreamUserInfo { active_cons: "0".to_string(), allowed_output_formats: Vec::from(["ts".to_string()]), auth: 1, - created_at: (now - Duration::days(365)).timestamp(), // fake - exp_date: (now + Duration::days(365)).timestamp(), // fake - is_trial: "0".to_string(), - max_connections: "1".to_string(), + created_at, + exp_date, + is_trial, + max_connections, message: server_info.message.to_string(), password: user.password.to_string(), username: user.username.to_string(), - status: "Active".to_string(), + status: user_status.to_string(), }, server_info: XtreamServerInfo { url: server_info.host.clone(), diff --git a/src/model/api_proxy.rs b/src/model/api_proxy.rs index bcb53e452..0adfa5f0b 100644 --- a/src/model/api_proxy.rs +++ b/src/model/api_proxy.rs @@ -1,11 +1,11 @@ use std::collections::HashSet; use std::fmt::Display; use std::str::FromStr; - +use chrono::{Local}; +use crate::m3u_filter_error::{create_m3u_filter_error_result, info_err, M3uFilterError, M3uFilterErrorKind}; +use crate::utils::file::config_reader; use enum_iterator::Sequence; use log::debug; -use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind, create_m3u_filter_error_result, info_err}; -use crate::utils::file::config_reader; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Sequence, PartialEq, Eq)] pub enum ProxyType { @@ -29,9 +29,9 @@ impl ProxyType { impl Display for ProxyType { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{}", match self { - Self::Reverse => Self::REVERSE, - Self::Redirect => Self::REDIRECT, - } + Self::Reverse => Self::REVERSE, + Self::Redirect => Self::REDIRECT, + } ) } } @@ -40,12 +40,64 @@ impl FromStr for ProxyType { type Err = M3uFilterError; fn from_str(s: &str) -> Result { - if s.eq("reverse") { - Ok(Self::Reverse) - } else if s.eq("redirect") { - Ok(Self::Redirect) - } else { - create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "Unknown ProxyType: {}", s) + match s { + Self::REVERSE => Ok(Self::Reverse), + Self::REDIRECT => Ok(Self::Redirect), + _ => create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "Unknown ProxyType: {}", s) + } + } +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Sequence, PartialEq, Eq)] +pub enum ProxyUserStatus { + Active, // The account is in good standing and can stream content + Expired, // The account can no longer access content unless it is renewed. + Banned, // The account is temporarily or permanently disabled. Typically used for users who violate terms of service or abuse the system. + Trial, // The account is marked as a trial account. + Disabled, // The account is inactive or deliberately disabled by the administrator. + Pending, +} + + +impl Default for ProxyUserStatus { + fn default() -> Self { + Self::Active + } +} + +impl ProxyUserStatus { + const ACTIVE: &'static str = "Active"; + const EXPIRED: &'static str = "Expired"; + const BANNED: &'static str = "Banned"; + const TRIAL: &'static str = "Trial"; + const DISABLED: &'static str = "Disabled"; + const PENDING: &'static str = "Pending"; +} + +impl Display for ProxyUserStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", match self { + Self::Active => Self::ACTIVE, + Self::Expired => Self::EXPIRED, + Self::Banned => Self::BANNED, + Self::Trial => Self::TRIAL, + Self::Disabled => Self::DISABLED, + Self::Pending => Self::PENDING, + }) + } +} + +impl FromStr for ProxyUserStatus { + type Err = M3uFilterError; + + fn from_str(s: &str) -> Result { + match s { + Self::EXPIRED => Ok(Self::Expired), + Self::BANNED => Ok(Self::Banned), + Self::TRIAL => Ok(Self::Trial), + Self::DISABLED => Ok(Self::Disabled), + Self::PENDING => Ok(Self::Pending), + _ => create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "Unknown ProxyType: {}", s) } } } @@ -62,6 +114,14 @@ pub struct ProxyUserCredentials { pub server: Option, #[serde(skip_serializing_if = "Option::is_none")] pub epg_timeshift: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub created_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub exp_date: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_connections: Option, // int as string + #[serde(skip_serializing_if = "Option::is_none")] + pub status: Option, } impl ProxyUserCredentials { @@ -97,6 +157,19 @@ impl ProxyUserCredentials { } } } + + pub fn is_active(&self) -> bool { + if let Some(status) = &self.status { + let now = Local::now(); + if let Some(exp_date) = self.exp_date.as_ref() { + if (exp_date - now.timestamp()) < 0 { + return false; + } + } + return matches!(status, ProxyUserStatus::Active | ProxyUserStatus::Trial); + } + true + } } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -274,7 +347,7 @@ impl ApiProxyConfig { None } - pub fn get_user_credentials(&self,username: &str) -> Option { + pub fn get_user_credentials(&self, username: &str) -> Option { let result = self.user.iter() .flat_map(|target_user| &target_user.credentials) .find(|credential| credential.username == username) diff --git a/src/model/healthcheck.rs b/src/model/healthcheck.rs index c5ff5e113..537cd09db 100644 --- a/src/model/healthcheck.rs +++ b/src/model/healthcheck.rs @@ -5,4 +5,5 @@ pub struct Healthcheck { pub time: String, pub mem: String, pub active_clients: usize, + pub active_connections: usize, } \ No newline at end of file