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

stream table ip fixed

Browse Source
This commit is contained in:
euzu
2025-10-30 13:00:25 +01:00
parent 5bf90fbd39
commit d71e8080f5
16 changed files with 452 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/api/api_utils.rs
+20 -19
View File
@@ -123,6 +123,7 @@ pub use try_option_bad_request;
pub use try_result_bad_request;
pub use try_result_not_found;
pub use try_unwrap_body;
use crate::auth::Fingerprint;
pub fn get_server_time() -> String {
chrono::offset::Local::now()
@@ -382,7 +383,7 @@ struct StreamingStrategy {
async fn resolve_streaming_strategy(
app_state: &AppState,
stream_url: &str,
addr: &str,
fingerprint: &Fingerprint,
input: &ConfigInput,
force_provider: Option<&str>,
) -> StreamingStrategy {
@@ -391,13 +392,13 @@ async fn resolve_streaming_strategy(
Some(provider) => {
app_state
.active_provider
.force_exact_acquire_connection(provider, addr)
.force_exact_acquire_connection(provider, &fingerprint.addr)
.await
}
None => {
app_state
.active_provider
.acquire_connection(&input.name, addr)
.acquire_connection(&input.name, &fingerprint.addr)
.await
}
};
@@ -461,7 +462,7 @@ async fn create_stream_response_details(
app_state: &AppState,
stream_options: &StreamOptions,
stream_url: &str,
addr: &str,
fingerprint: &Fingerprint,
req_headers: &HeaderMap,
input: &ConfigInput,
item_type: PlaylistItemType,
@@ -469,7 +470,7 @@ async fn create_stream_response_details(
connection_permission: UserConnectionPermission,
force_provider: Option<&str>,
) -> StreamDetails {
let mut streaming_strategy = resolve_streaming_strategy(app_state, stream_url, addr, input, force_provider).await;
let mut streaming_strategy = resolve_streaming_strategy(app_state, stream_url, fingerprint, input, force_provider).await;
let config_grace_period_millis = app_state
.app_config
.config
@@ -736,7 +737,7 @@ fn prepare_body_stream(
/// # Panics
pub async fn force_provider_stream_response(
addr: &str,
fingerprint: &Fingerprint,
app_state: &AppState,
user_session: &UserSession,
mut stream_channel: StreamChannel,
@@ -753,7 +754,7 @@ pub async fn force_provider_stream_response(
app_state,
&stream_options,
&user_session.stream_url,
addr,
fingerprint,
req_headers,
input,
item_type,
@@ -770,11 +771,11 @@ pub async fn force_provider_stream_response(
.map(|(h, sc, url)| (h.clone(), *sc, url.clone()));
app_state
.active_users
.update_session_addr(&user.username, &user_session.token, addr)
.update_session_addr(&user.username, &user_session.token, &fingerprint.addr)
.await;
stream_channel.shared = share_stream;
let stream =
ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel, req_headers)
ActiveClientStream::new(stream_details, app_state, user, connection_permission, fingerprint, stream_channel, req_headers)
.await;
let (status_code, header_map) =
@@ -809,7 +810,7 @@ pub async fn force_provider_stream_response(
/// # Panics
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub async fn stream_response(
addr: &str,
fingerprint: &Fingerprint,
app_state: &AppState,
session_token: &str,
mut stream_channel: StreamChannel,
@@ -838,7 +839,7 @@ pub async fn stream_response(
let share_stream = is_stream_share_enabled(item_type, target);
if share_stream {
if let Some(value) =
shared_stream_response(app_state, stream_url, addr, user, connection_permission, stream_channel.clone(), req_headers).await
shared_stream_response(app_state, stream_url, fingerprint, user, connection_permission, stream_channel.clone(), req_headers).await
{
return value.into_response();
}
@@ -849,7 +850,7 @@ pub async fn stream_response(
app_state,
&stream_options,
stream_url,
addr,
fingerprint,
req_headers,
input,
item_type,
@@ -873,7 +874,7 @@ pub async fn stream_response(
};
stream_channel.shared = share_stream;
let stream =
ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel, req_headers)
ActiveClientStream::new(stream_details, app_state, user, connection_permission, fingerprint, stream_channel, req_headers)
.await;
let stream_resp = if share_stream {
debug_if_enabled!(
@@ -889,7 +890,7 @@ pub async fn stream_response(
app_state,
stream_url,
stream,
Some(addr),
Some(&fingerprint.addr),
shared_headers,
stream_options.buffer_size,
provider_guard,
@@ -952,7 +953,7 @@ pub async fn stream_response(
virtual_id,
&provider,
&session_url,
addr,
&fingerprint.addr,
connection_permission,
)
.await;
@@ -984,14 +985,14 @@ fn get_stream_throttle(app_state: &AppState) -> u64 {
async fn shared_stream_response(
app_state: &AppState,
stream_url: &str,
addr: &str,
fingerprint: &Fingerprint,
user: &ProxyUserCredentials,
connect_permission: UserConnectionPermission,
mut stream_channel: StreamChannel,
req_headers: &HeaderMap,
) -> Option<impl IntoResponse> {
if let Some((stream, provider)) =
SharedStreamManager::subscribe_shared_stream(app_state, stream_url, Some(addr)).await
SharedStreamManager::subscribe_shared_stream(app_state, stream_url, Some(&fingerprint.addr)).await
{
debug_if_enabled!(
"Using shared stream {}",
@@ -1010,7 +1011,7 @@ async fn shared_stream_response(
stream_details.provider_name = provider;
stream_channel.shared = true;
let stream =
ActiveClientStream::new(stream_details, app_state, user, connect_permission, addr, stream_channel, req_headers)
ActiveClientStream::new(stream_details, app_state, user, connect_permission, fingerprint, stream_channel, req_headers)
.await
.boxed();
let mut response = axum::response::Response::builder().status(status_code);
@@ -1326,6 +1327,6 @@ pub fn json_or_bin_response<T: Serialize>(accept: Option<&String>, data: &T) ->
json_response(data).into_response()
}
pub fn create_fingerprint(fingerprint: &str, username: &str, virtual_id: u32) -> String {
pub fn create_session_fingerprint(fingerprint: &str, username: &str, virtual_id: u32) -> String {
format!("{fingerprint}|{username}|{virtual_id}")
}
backend/src/api/endpoints/hls_api.rs
+9 -11
View File
@@ -1,4 +1,4 @@
use crate::api::api_utils::{create_fingerprint, try_unwrap_body};
use crate::api::api_utils::{create_session_fingerprint, try_unwrap_body};
use crate::api::api_utils::{
force_provider_stream_response, get_stream_alternative_url, is_seek_request,
};
@@ -41,8 +41,7 @@ fn hls_response(hls_content: String) -> impl IntoResponse + Send {
#[allow(clippy::too_many_arguments)]
pub(in crate::api) async fn handle_hls_stream_request(
fingerprint: &str,
addr: &str,
fingerprint: &Fingerprint,
app_state: &Arc<AppState>,
user: &ProxyUserCredentials,
user_session: Option<&UserSession>,
@@ -59,7 +58,7 @@ pub(in crate::api) async fn handle_hls_stream_request(
Some(session) => {
match app_state
.active_provider
.force_exact_acquire_connection(&session.provider, addr)
.force_exact_acquire_connection(&session.provider, &fingerprint.addr)
.await
.get_provider_config()
{
@@ -78,14 +77,14 @@ pub(in crate::api) async fn handle_hls_stream_request(
{
Some(provider_cfg) => {
let stream_url = get_stream_alternative_url(&url, input, &provider_cfg);
let user_session_token = create_fingerprint(fingerprint, &user.username, virtual_id);
let user_session_token = create_session_fingerprint(&fingerprint.key, &user.username, virtual_id);
let session_token = app_state.active_users.create_user_session(
user,
&user_session_token,
virtual_id,
&provider_cfg.name,
&stream_url,
addr,
&fingerprint.addr,
connection_permission,
).await;
(stream_url, Some(session_token))
@@ -188,7 +187,7 @@ async fn resolve_stream_channel(
#[allow(clippy::too_many_lines)]
async fn hls_api_stream(
Fingerprint(fingerprint, addr): Fingerprint,
fingerprint: Fingerprint,
req_headers: axum::http::HeaderMap,
axum::extract::Path(params): axum::extract::Path<HlsApiPathParams>,
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
@@ -218,7 +217,7 @@ async fn hls_api_stream(
)
);
let user_session_token = create_fingerprint(&fingerprint, &user.username, virtual_id);
let user_session_token = create_session_fingerprint(&fingerprint.key, &user.username, virtual_id);
let mut user_session = app_state
.active_users
.get_and_update_user_session(&user.username, &user_session_token).await;
@@ -258,7 +257,7 @@ async fn hls_api_stream(
if is_seek_request(stream_channel.cluster, &req_headers).await {
// partial request means we are in reverse proxy mode, seek happened
return force_provider_stream_response(
&addr,
&fingerprint,
&app_state,
session,
stream_channel,
@@ -285,7 +284,6 @@ async fn hls_api_stream(
if is_hls_url(&session.stream_url) {
return handle_hls_stream_request(
&fingerprint,
&addr,
&app_state,
&user,
Some(session),
@@ -301,7 +299,7 @@ async fn hls_api_stream(
let stream_channel = resolve_stream_channel(&app_state, &target, virtual_id, &hls_url).await;
force_provider_stream_response(
&addr,
&fingerprint,
&app_state,
session,
stream_channel,
backend/src/api/endpoints/m3u_api.rs
+6 -9
View File
@@ -1,4 +1,4 @@
use crate::api::api_utils::{create_fingerprint, try_unwrap_body};
use crate::api::api_utils::{create_session_fingerprint, try_unwrap_body};
use crate::api::api_utils::{
force_provider_stream_response, get_user_target, get_user_target_by_credentials,
is_seek_request, redirect, redirect_response, resource_response, separate_number_and_remainder,
@@ -72,8 +72,7 @@ async fn m3u_api_post(
#[allow(clippy::too_many_lines)]
async fn m3u_api_stream(
fingerprint: &str,
addr: &str,
fingerprint: &Fingerprint,
req_headers: &axum::http::HeaderMap,
app_state: &Arc<AppState>,
api_req: &UserApiRequest,
@@ -124,7 +123,7 @@ async fn m3u_api_stream(
);
let cluster = XtreamCluster::try_from(pli.item_type).unwrap_or(XtreamCluster::Live);
let session_key = create_fingerprint(fingerprint, &user.username, virtual_id);
let session_key = create_session_fingerprint(&fingerprint.key, &user.username, virtual_id);
let user_session = app_state
.active_users
.get_and_update_user_session(&user.username, &session_key).await;
@@ -152,7 +151,7 @@ async fn m3u_api_stream(
if session.virtual_id == virtual_id && is_seek_request(cluster, req_headers).await {
// partial request means we are in reverse proxy mode, seek happened
return force_provider_stream_response(
addr,
fingerprint,
app_state,
session,
pli.to_stream_channel(),
@@ -208,7 +207,6 @@ async fn m3u_api_stream(
if is_hls_request {
return handle_hls_stream_request(
fingerprint,
addr,
app_state,
&user,
user_session.as_ref(),
@@ -223,7 +221,7 @@ async fn m3u_api_stream(
}
stream_response(
addr,
fingerprint,
app_state,
&session_key,
pli.to_stream_channel(),
@@ -302,7 +300,7 @@ async fn m3u_api_resource(
macro_rules! create_m3u_api_stream {
($fn_name:ident, $context:expr) => {
async fn $fn_name(
Fingerprint(fingerprint, addr): Fingerprint,
fingerprint: Fingerprint,
req_headers: axum::http::HeaderMap,
axum::extract::Query(api_req): axum::extract::Query<UserApiRequest>,
axum::extract::Path((username, password, stream_id)): axum::extract::Path<(
@@ -315,7 +313,6 @@ macro_rules! create_m3u_api_stream {
) -> impl IntoResponse + Send {
m3u_api_stream(
&fingerprint,
&addr,
&req_headers,
&app_state,
&api_req,
backend/src/api/endpoints/v1_api.rs
+3 -2
View File
@@ -8,7 +8,7 @@ use crate::auth::validator_admin;
use crate::utils::ip_checker::get_ips;
use crate::{VERSION};
use axum::response::IntoResponse;
use shared::model::{InputFetchMethod, IpCheckDto, StatusCheck};
use shared::model::{default_geoip_url, InputFetchMethod, IpCheckDto, StatusCheck};
use shared::utils::{concat_path_leading_slash};
use std::collections::{BTreeMap, HashMap};
use std::io::{Cursor};
@@ -80,8 +80,9 @@ async fn geoip_update(axum::extract::State(app_state): axum::extract::State<Arc<
let geoip_db_path = &*get_geoip_path(&config.working_dir);
let _file_lock = app_state.app_config.file_locks.write_lock(geoip_db_path);
let url = if geoip.url.trim().is_empty() { default_geoip_url() } else { geoip.url.clone() };
let input_source = InputSource {
url: geoip.url.clone(),
url,
username: None,
password: None,
method: InputFetchMethod::GET,
backend/src/api/endpoints/xtream_api.rs
+20 -22
View File
@@ -1,7 +1,7 @@
// https://github.com/tellytv/go.xtream-codes/blob/master/structs.go
use crate::api::api_utils;
use crate::api::api_utils::{create_fingerprint, try_unwrap_body};
use crate::api::api_utils::{create_session_fingerprint, try_unwrap_body};
use crate::api::api_utils::{
force_provider_stream_response, get_user_target, get_user_target_by_credentials,
is_seek_request, redirect_response, resource_response, separate_number_and_remainder,
@@ -211,8 +211,7 @@ async fn get_user_info(user: &ProxyUserCredentials, app_state: &AppState) -> Xtr
#[allow(clippy::too_many_lines)]
async fn xtream_player_api_stream(
fingerprint: &str,
addr: &str,
fingerprint: &Fingerprint,
req_headers: &HeaderMap,
app_state: &Arc<AppState>,
api_req: &UserApiRequest,
@@ -269,7 +268,7 @@ async fn xtream_player_api_stream(
(pli.xtream_cluster, pli.item_type)
};
let session_key = create_fingerprint(fingerprint, &user.username, virtual_id);
let session_key = create_session_fingerprint(&fingerprint.key, &user.username, virtual_id);
let user_session = app_state
.active_users
.get_and_update_user_session(&user.username, &session_key).await;
@@ -295,13 +294,16 @@ async fn xtream_player_api_stream(
.into_response();
}
let mut stream_channel = pli.to_stream_channel();
stream_channel.item_type = item_type;
if session.virtual_id == virtual_id && is_seek_request(cluster, req_headers).await {
// partial request means we are in reverse proxy mode, seek happened
return force_provider_stream_response(
addr,
fingerprint,
app_state,
session,
pli.to_stream_channel(),
stream_channel,
req_headers,
&input,
&user,
@@ -369,7 +371,6 @@ async fn xtream_player_api_stream(
if is_hls_request {
return handle_hls_stream_request(
fingerprint,
addr,
app_state,
&user,
user_session.as_ref(),
@@ -383,11 +384,14 @@ async fn xtream_player_api_stream(
.into_response();
}
let mut stream_channel = pli.to_stream_channel();
stream_channel.item_type = item_type;
stream_response(
addr,
fingerprint,
app_state,
session_key.as_str(),
pli.to_stream_channel(),
stream_channel,
&stream_url,
req_headers,
&input,
@@ -402,8 +406,7 @@ async fn xtream_player_api_stream(
#[allow(clippy::too_many_lines)]
// Used by webui
async fn xtream_player_api_stream_with_token(
fingerprint: &str,
addr: &str,
fingerprint: &Fingerprint,
req_headers: &HeaderMap,
app_state: &Arc<AppState>,
target_id: u16,
@@ -439,7 +442,7 @@ async fn xtream_player_api_stream_with_token(
)
);
let session_key = create_fingerprint(fingerprint, "webui", virtual_id);
let session_key = create_session_fingerprint(&fingerprint.key, "webui", virtual_id);
let is_hls_request =
pli.item_type == PlaylistItemType::LiveHls || stream_ext.as_deref() == Some(HLS_EXT);
@@ -472,7 +475,6 @@ async fn xtream_player_api_stream_with_token(
if is_hls_request {
return handle_hls_stream_request(
fingerprint,
addr,
app_state,
&user,
None,
@@ -516,7 +518,7 @@ async fn xtream_player_api_stream_with_token(
sanitize_sensitive_info(&stream_url)
);
stream_response(
addr,
fingerprint,
app_state,
session_key.as_str(),
pli.to_stream_channel(),
@@ -770,7 +772,7 @@ async fn xtream_player_api_resource(
macro_rules! create_xtream_player_api_stream {
($fn_name:ident, $context:expr) => {
async fn $fn_name(
Fingerprint(fingerprint, addr): Fingerprint,
fingerprint: Fingerprint,
req_headers: HeaderMap,
axum::extract::Path((username, password, stream_id)): axum::extract::Path<(
String,
@@ -782,7 +784,6 @@ macro_rules! create_xtream_player_api_stream {
) -> impl IntoResponse + Send {
xtream_player_api_stream(
&fingerprint,
&addr,
&req_headers,
&app_state,
&api_req,
@@ -848,7 +849,7 @@ struct XtreamApiTimeShiftRequest {
}
async fn xtream_player_api_timeshift_stream(
Fingerprint(fingerprint, addr): Fingerprint,
fingerprint: Fingerprint,
req_headers: HeaderMap,
axum::extract::Query(mut api_req): axum::extract::Query<UserApiRequest>,
axum::extract::Path(timeshift_request): axum::extract::Path<XtreamApiTimeShiftRequest>,
@@ -891,7 +892,6 @@ async fn xtream_player_api_timeshift_stream(
xtream_player_api_stream(
&fingerprint,
&addr,
&req_headers,
&app_state,
&api_req,
@@ -908,7 +908,7 @@ async fn xtream_player_api_timeshift_stream(
}
async fn xtream_player_api_timeshift_query_stream(
Fingerprint(fingerprint, addr): Fingerprint,
fingerprint: Fingerprint,
req_headers: HeaderMap,
axum::extract::Query(api_query_req): axum::extract::Query<UserApiRequest>,
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
@@ -933,7 +933,6 @@ async fn xtream_player_api_timeshift_query_stream(
}
xtream_player_api_stream(
&fingerprint,
&addr,
&req_headers,
&app_state,
&api_query_req,
@@ -1585,7 +1584,7 @@ macro_rules! register_xtream_api_timeshift {
}
async fn xtream_player_token_stream(
Fingerprint(fingerprint, addr): Fingerprint,
fingerprint: Fingerprint,
axum::extract::Path((token, target_id, cluster, stream_id)): axum::extract::Path<(
String,
u16,
@@ -1598,7 +1597,6 @@ async fn xtream_player_token_stream(
let ctxt = try_result_bad_request!(ApiStreamContext::from_str(cluster.as_str()));
xtream_player_api_stream_with_token(
&fingerprint,
&addr,
&req_headers,
&app_state,
target_id,
backend/src/api/model/active_user_manager.rs
+7 -5
View File
@@ -14,6 +14,7 @@ use std::sync::Arc;
use arc_swap::ArcSwapOption;
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
use tokio::sync::RwLock;
use crate::auth::Fingerprint;
use crate::utils::GeoIp;
const USER_GC_TTL: u64 = 900; // 15 Min
@@ -331,11 +332,11 @@ impl ActiveUserManager {
Self::get_active_connections(&self.user).await
}
pub async fn add_connection(&self, username: &str, max_connections: u32, addr: &str, provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>) -> UserConnectionGuard {
pub async fn add_connection(&self, username: &str, max_connections: u32, fingerprint: &Fingerprint, provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>) -> UserConnectionGuard {
let country = {
let geoip = self.geo_ip.load();
if let Some(geoip_db) = (*geoip).as_ref() {
geoip_db.lookup(&strip_port(addr))
geoip_db.lookup(&strip_port(&fingerprint.client_ip))
} else {
None
}
@@ -343,7 +344,8 @@ impl ActiveUserManager {
let stream_info = StreamInfo::new(
username,
addr,
&fingerprint.addr,
&fingerprint.client_ip,
provider,
stream_channel,
user_agent.to_string(),
@@ -364,7 +366,7 @@ impl ActiveUserManager {
{
let mut user_by_addr = self.user_by_addr.write().await;
user_by_addr.insert(addr.to_string(), username.to_string());
user_by_addr.insert(fingerprint.addr.to_string(), username.to_string());
}
if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Connected(stream_info)) {
@@ -372,7 +374,7 @@ impl ActiveUserManager {
}
self.log_active_user().await;
UserConnectionGuard::new(Arc::new(self.clone_inner()), addr, self.release_sender())
UserConnectionGuard::new(Arc::new(self.clone_inner()), &fingerprint.addr, self.release_sender())
}
fn is_log_user_enabled(&self) -> bool {
backend/src/api/model/streams/active_client_stream.rs
+4 -3
View File
@@ -18,6 +18,7 @@ use std::task::{Poll};
use axum::http::header::USER_AGENT;
use axum::http::HeaderMap;
use futures::task::AtomicWaker;
use crate::auth::Fingerprint;
const INNER_STREAM: u8 = 0_u8;
const GRACE_BLOCK_STREAM: u8 = 1_u8;
@@ -40,7 +41,7 @@ impl ActiveClientStream {
app_state: &AppState,
user: &ProxyUserCredentials,
connection_permission: UserConnectionPermission,
addr: &str,
fingerprint: &Fingerprint,
stream_channel: StreamChannel,
req_headers: &HeaderMap) -> Self {
if connection_permission == UserConnectionPermission::Exhausted {
@@ -51,11 +52,11 @@ impl ActiveClientStream {
let provider_name = stream_details.provider_name.as_ref().map_or_else(String::new, ToString::to_string);
let user_agent = req_headers.get(USER_AGENT).map(|h| String::from_utf8_lossy(h.as_bytes())).unwrap_or_default();
let user_connection_guard = Some(app_state.active_users.add_connection(username, user.max_connections, addr, &provider_name, stream_channel, user_agent).await);
let user_connection_guard = Some(app_state.active_users.add_connection(username, user.max_connections, fingerprint, &provider_name, stream_channel, user_agent).await);
let cfg = &app_state.app_config;
let waker = Some(Arc::new(AtomicWaker::new()));
let waker_clone = waker.clone();
let grace_stop_flag = Self::stream_grace_period(app_state, &stream_details, grant_user_grace_period, user, addr, waker_clone.clone());
let grace_stop_flag = Self::stream_grace_period(app_state, &stream_details, grant_user_grace_period, user, &fingerprint.addr, waker_clone.clone());
let custom_response = cfg.custom_stream_response.load();
let custom_video = custom_response.as_ref()
.map_or((None, None), |c|