Merge pull request #391 from euzu/feature/stream_table

## Summary by CodeRabbit

* **New Features**
  * Added GeoIP support for geolocation resolution of client IPs with configurable database URL
  * Added GeoIP database update functionality accessible from the UI
  * Enhanced SSDP discovery protocol validation and handling
  * Added stream monitoring data: duration tracking, player user agent, and country information

* **Bug Fixes**
  * Fixed protocol encoding for HDHomeRun proprietary TLV strings
  * Tightened TCP request validation minimum size

* **Documentation**
  * Added GeoIP configuration section with setup and CSV format details
  * Enhanced reverse proxy guidance for header forwarding
This commit is contained in:
euzu
2025-10-28 18:53:02 +01:00
committed by GitHub
60 changed files with 1040 additions and 440 deletions
+1
View File
@@ -14,6 +14,7 @@ Output filters are applied after all transformations have been performed, theref
- Telegram supports markdown generation for structured json messages. simply set `markdown: true` in telegram config.
- Added User-Stream-Connections Table to WebUI
- Enhanced STRM output filenames to include detailed media quality info (e.g., 4K, HDR, x265, 5.1) for easy version distinction.
- Added standardized SSDP (Simple Service Discovery Protocol) and the Proprietary HDHomeRun UDP Discovery Protocol (Port 65001)
# 3.1.7 (2025-10-10)
- Added Dark/Bright theme switch
Generated
+15 -72
View File
@@ -2,15 +2,6 @@
# It is not intended for manual editing.
version = 4
[[package]]
name = "addr2line"
version = "0.25.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b5d307320b3181d6d7954e663bd7c774a838b8220fe0593c86d9fb09f498b4b"
dependencies = [
"gimli",
]
[[package]]
name = "adler2"
version = "2.0.1"
@@ -239,21 +230,6 @@ dependencies = [
"syn 2.0.106",
]
[[package]]
name = "backtrace"
version = "0.3.76"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bb531853791a215d7c62a30daf0dde835f381ab5de4589cfe7c649d2cbe92bd6"
dependencies = [
"addr2line",
"cfg-if",
"libc",
"miniz_oxide",
"object",
"rustc-demangle",
"windows-link 0.2.1",
]
[[package]]
name = "base16ct"
version = "0.2.0"
@@ -1022,7 +998,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -1337,12 +1313,6 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "gimli"
version = "0.32.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7"
[[package]]
name = "gloo"
version = "0.8.1"
@@ -2248,17 +2218,6 @@ dependencies = [
"libc",
]
[[package]]
name = "io-uring"
version = "0.7.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "046fa2d4d00aea763528b4950358d0ead425372445dc8ff86312b3c69ff7727b"
dependencies = [
"bitflags 2.9.4",
"cfg-if",
"libc",
]
[[package]]
name = "ipnet"
version = "2.11.0"
@@ -2342,9 +2301,9 @@ dependencies = [
[[package]]
name = "jsonwebtoken"
version = "10.0.0"
version = "10.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1417155a38e99d7704ddb3ea7445fe57fdbd5d756d727740a9ed8b9ebaed6e1"
checksum = "3d119c6924272d16f0ab9ce41f7aa0bfef9340c00b0bb7ca3dd3b263d4a9150b"
dependencies = [
"base64",
"ed25519-dalek",
@@ -2682,15 +2641,6 @@ dependencies = [
"libc",
]
[[package]]
name = "object"
version = "0.37.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe"
dependencies = [
"memchr",
]
[[package]]
name = "once_cell"
version = "1.21.3"
@@ -3365,9 +3315,9 @@ dependencies = [
[[package]]
name = "regex"
version = "1.12.1"
version = "1.12.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a52d8d02cacdb176ef4678de6c052efb4b3da14b78e4db683a4252762be5433"
checksum = "843bc0191f75f3e22651ae5f1e72939ab2f72a4bc30fa80a066bd66edefc24d4"
dependencies = [
"aho-corasick",
"memchr",
@@ -3547,12 +3497,6 @@ dependencies = [
"crossbeam-utils",
]
[[package]]
name = "rustc-demangle"
version = "0.1.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "56f7d92ca342cea22a06f2121d944b4fd82af56988c270852495420f961d4ace"
[[package]]
name = "rustc-hash"
version = "2.1.1"
@@ -3578,7 +3522,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -3868,6 +3812,7 @@ dependencies = [
"enum-iterator",
"fastrand",
"indexmap",
"js-sys",
"log",
"path-clean",
"pest",
@@ -4056,7 +4001,7 @@ dependencies = [
"getrandom 0.3.3",
"once_cell",
"rustix",
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
@@ -4159,28 +4104,25 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
[[package]]
name = "tokio"
version = "1.47.1"
version = "1.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "89e49afdadebb872d3145a5638b59eb0691ea23e46ca484037cfab3b76b95038"
checksum = "ff360e02eab121e0bc37a2d3b4d4dc622e6eda3a8e5253d5435ecf5bd4c68408"
dependencies = [
"backtrace",
"bytes",
"io-uring",
"libc",
"mio",
"parking_lot",
"pin-project-lite",
"slab",
"socket2",
"tokio-macros",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
name = "tokio-macros"
version = "2.5.0"
version = "2.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8"
checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5"
dependencies = [
"proc-macro2",
"quote",
@@ -4426,6 +4368,7 @@ dependencies = [
"deunicode",
"enum-iterator",
"env_logger",
"fastrand",
"filetime",
"flate2",
"futures",
@@ -4840,7 +4783,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.52.0",
]
[[package]]
+27 -1
View File
@@ -217,7 +217,8 @@ Attributes:
- `buffer`
- `throttle` Allowed units are `KB/s`,`MB/s`,`KiB/s`,`MiB/s`,`kbps`,`mbps`,`Mibps`. Default unit is `kbps`
- `grace_period_millis` default set to 300 milliseconds.
- `grace_period_timeout_secs` efault set to 2 seconds.
- `grace_period_timeout_secs` default set to 2 seconds.
- `geopip` is for resolving ip addresses to country names.
##### 1.6.1.1 `retry`
If set to `true` on connection loss to provider, the stream will be reconnected.
@@ -268,6 +269,26 @@ If the connection is not throttled, the player will play its buffered content lo
##### 1.6.1.4 `grace_period_timeout_secs`
How long the grace grant will last, until another grace grant can made.
##### 1.6.1.5 `geoip`
Disabled by default.
Is used to resolve ip addresses to location.
It has 2 attributes:
```yaml
geoip:
enabled: true
url: <the url>
```
The `url` is optional and default vaue is: `https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/asn-country/asn-country-ipv4.csv`
The format is csv with 3 columns `range_start,range_end,country_code`
Example:
```csv
1.0.0.0,1.0.0.255,AU
1.0.1.0,1.0.3.255,CN
1.0.4.0,1.0.7.255,AU
```
#### 1.6.2 `cache`
LRU-Cache is for resources. If it is `enabled`, the resources/images are persisted in the given `dir`. If the cache size exceeds `size`,
In an LRU cache, the least recently used items are evicted to make room for new items if the cache `size`is exceeded.
@@ -1497,6 +1518,11 @@ user:
status: Active
```
If yu use a reverse proxy in fron of Tuliprox, dont forget to forward
- `X-Real-IP`
- `X-Forwarded-For`
Now you can do `nginx` configuration like
```config
location /tuliprox {
+3 -2
View File
@@ -22,7 +22,7 @@ axum = { version = "0.8" , features = ["macros", "default", "ws"]}
tower = "0.5"
tower-http = { version = "0.6", features = ["cors", "auth", "fs", "compression-full", "trace"] }
tower_governor = { version = "0.8", features = ["axum"] }
jsonwebtoken = { version = "10.0", features = ["rust_crypto"] }
jsonwebtoken = { version = "10.1", features = ["rust_crypto"] }
rust-argon2 = "3"
futures = "0.3"
path-clean = "1.0"
@@ -36,12 +36,13 @@ log = "0.4"
env_logger = "0.11"
bincode = { version = "2.0", features = ["std", "serde"] }
rand = "0.9"
fastrand = "2"
rpassword = "7.4"
flate2 = "1.1"
blake3 = "1.8"
bytes = "1.10"
tokio-stream = { version = "0.1", features = ["sync"] }
tokio = { version = "1.47", features = ["rt-multi-thread", "parking_lot", "fs"] }
tokio = { version = "1.48", features = ["rt-multi-thread", "parking_lot", "fs"] }
#tokio = { version = "1.46", features = ["rt-multi-thread", "parking_lot", "fs", "tracing"] }
#console-subscriber = "0"
#tracing = "0.1"
+37 -26
View File
@@ -329,7 +329,7 @@ enum ProviderStreamState {
pub struct StreamDetails {
pub stream: Option<BoxedProviderStream>,
stream_info: ProviderStreamInfo,
pub input_name: Option<String>,
pub provider_name: Option<String>,
pub grace_period_millis: u64,
pub reconnect_flag: Option<Arc<AtomicOnceFlag>>,
pub provider_connection_guard: Option<Arc<ProviderConnectionGuard>>,
@@ -340,7 +340,7 @@ impl StreamDetails {
Self {
stream: Some(stream),
stream_info: None,
input_name: None,
provider_name: None,
grace_period_millis: default_grace_period_millis(),
reconnect_flag: None,
provider_connection_guard: None,
@@ -469,8 +469,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, addr, input, force_provider).await;
let config_grace_period_millis = app_state
.app_config
.config
@@ -484,6 +483,13 @@ async fn create_stream_response_details(
&streaming_strategy.provider_stream_state,
config_grace_period_millis,
);
let provider_name = streaming_strategy
.provider_connection_guard
.as_ref()
.and_then(|guard| guard.get_provider_name());
debug!("🍄🍄🍄 using provider {provider_name:?}");
match streaming_strategy.provider_stream_state {
// custom stream means we display our own stream like connection exhausted, channel-unavailable...
ProviderStreamState::Custom(provider_stream) => {
@@ -491,7 +497,7 @@ async fn create_stream_response_details(
StreamDetails {
stream,
stream_info,
input_name: None,
provider_name: provider_name.clone(),
grace_period_millis,
reconnect_flag: None,
provider_connection_guard: streaming_strategy.provider_connection_guard.clone(),
@@ -525,14 +531,6 @@ async fn create_stream_response_details(
((None, None), None)
};
// if we have no stream, we should release the provider
if stream.is_none() {
if let Some(guard) = streaming_strategy.provider_connection_guard.take() {
drop(guard);
}
error!("Cant open stream {}", sanitize_sensitive_info(&request_url));
}
if log_enabled!(log::Level::Debug) {
if let Some((headers, status_code, response_url)) = stream_info.as_ref() {
debug!(
@@ -546,10 +544,18 @@ async fn create_stream_response_details(
}
}
// if we have no stream, we should release the provider
if stream.is_none() {
if let Some(guard) = streaming_strategy.provider_connection_guard.take() {
drop(guard);
}
error!("Cant open stream {}", sanitize_sensitive_info(&request_url));
}
StreamDetails {
stream,
stream_info,
input_name: provider_name,
provider_name,
grace_period_millis,
reconnect_flag,
provider_connection_guard: streaming_strategy.provider_connection_guard.take(),
@@ -770,7 +776,7 @@ pub async fn force_provider_stream_response(
.await;
stream_channel.shared = share_stream;
let stream =
ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel)
ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel, req_headers)
.await;
let (status_code, header_map) =
@@ -834,7 +840,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()).await
shared_stream_response(app_state, stream_url, addr, user, connection_permission, stream_channel.clone(), req_headers).await
{
return value.into_response();
}
@@ -860,10 +866,7 @@ pub async fn stream_response(
.stream_info
.as_ref()
.map(|(h, sc, response_url)| (h.clone(), *sc, response_url.clone()));
let provider_name = stream_details
.provider_connection_guard
.as_ref()
.and_then(|guard| guard.get_provider_name());
let provider_name = stream_details.provider_name.clone();
let provider_guard = if share_stream {
stream_details.provider_connection_guard.take()
@@ -872,7 +875,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)
ActiveClientStream::new(stream_details, app_state, user, connection_permission, addr, stream_channel, req_headers)
.await;
let stream_resp = if share_stream {
debug_if_enabled!(
@@ -884,7 +887,7 @@ pub async fn stream_response(
.as_ref()
.map_or_else(Vec::new, |(h, _, _)| h.clone());
if let Some(broadcast_stream) = SharedStreamManager::register_shared_stream(
if let Some((broadcast_stream, _provider)) = SharedStreamManager::register_shared_stream(
app_state,
stream_url,
stream,
@@ -986,9 +989,10 @@ async fn shared_stream_response(
addr: &str,
user: &ProxyUserCredentials,
connect_permission: UserConnectionPermission,
mut stream_channel: StreamChannel
mut stream_channel: StreamChannel,
req_headers: &HeaderMap,
) -> Option<impl IntoResponse> {
if let Some(stream) =
if let Some((stream, provider)) =
SharedStreamManager::subscribe_shared_stream(app_state, stream_url, Some(addr)).await
{
debug_if_enabled!(
@@ -1004,10 +1008,11 @@ async fn shared_stream_response(
headers.clone(),
axum::http::StatusCode::OK,
)));
let stream_details = StreamDetails::from_stream(stream);
let mut stream_details = StreamDetails::from_stream(stream);
stream_details.provider_name = provider;
stream_channel.shared = true;
let stream =
ActiveClientStream::new(stream_details, app_state, user, connect_permission, addr, stream_channel)
ActiveClientStream::new(stream_details, app_state, user, connect_permission, addr, stream_channel, req_headers)
.await
.boxed();
let mut response = axum::response::Response::builder().status(status_code);
@@ -1051,6 +1056,8 @@ fn get_add_cache_content(
let cache = Arc::clone(cache);
let add_cache_content: Arc<dyn Fn(usize) + Send + Sync> = Arc::new(move |size| {
let res_url = resource_url.clone();
// todo spawn, replace with unboundchannel
let cache = Arc::clone(&cache);
tokio::spawn(async move {
if let Some(cache) = cache.load().as_ref() {
@@ -1319,4 +1326,8 @@ pub fn json_or_bin_response<T: Serialize>(accept: Option<&String>, data: &T) ->
return bin_response(data).into_response();
}
json_response(data).into_response()
}
pub fn create_fingerprint(fingerprint: &str, username: &str, virtual_id: u32) -> String {
format!("{fingerprint}{username}{virtual_id}")
}
+1 -1
View File
@@ -65,7 +65,7 @@ impl ConfigFile {
async fn load_config(app_state: &Arc<AppState>) -> Result<(), TuliproxError> {
let paths = <Arc<ArcSwap<ConfigPaths>> as Access<ConfigPaths>>::load(&app_state.app_config.paths);
let config_file = paths.config_file_path.as_str();
let config_dto = read_config_file(config_file, true)?;
let config_dto = read_config_file(config_file, true, true)?;
let mapping_changed = paths.mapping_file_path.as_ref() != config_dto.mapping_path.as_ref();
let mut config: Config = Config::from(config_dto);
config.prepare(paths.config_path.as_str())?;
+3 -3
View File
@@ -1,4 +1,4 @@
use crate::api::api_utils::try_unwrap_body;
use crate::api::api_utils::{create_fingerprint, try_unwrap_body};
use crate::api::api_utils::{
force_provider_stream_response, get_stream_alternative_url, is_seek_request,
};
@@ -78,7 +78,7 @@ 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 = format!("{fingerprint}{virtual_id}");
let user_session_token = create_fingerprint(fingerprint, &user.username, virtual_id);
let session_token = app_state.active_users.create_user_session(
user,
&user_session_token,
@@ -218,7 +218,7 @@ async fn hls_api_stream(
)
);
let user_session_token = format!("{fingerprint}{virtual_id}");
let user_session_token = create_fingerprint(&fingerprint, &user.username, virtual_id);
let mut user_session = app_state
.active_users
.get_and_update_user_session(&user.username, &user_session_token).await;
+2 -2
View File
@@ -1,4 +1,4 @@
use crate::api::api_utils::try_unwrap_body;
use crate::api::api_utils::{create_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,
@@ -123,7 +123,7 @@ async fn m3u_api_stream(
);
let cluster = XtreamCluster::try_from(pli.item_type).unwrap_or(XtreamCluster::Live);
let session_key = format!("{fingerprint}{virtual_id}");
let session_key = create_fingerprint(fingerprint, &user.username, virtual_id);
let user_session = app_state
.active_users
.get_and_update_user_session(&user.username, &session_key).await;
+61 -2
View File
@@ -8,12 +8,18 @@ use crate::auth::validator_admin;
use crate::utils::ip_checker::get_ips;
use crate::{VERSION};
use axum::response::IntoResponse;
use shared::model::{IpCheckDto,StatusCheck};
use shared::model::{InputFetchMethod, IpCheckDto, StatusCheck};
use shared::utils::{concat_path_leading_slash};
use std::collections::BTreeMap;
use std::collections::{BTreeMap, HashMap};
use std::io::{Cursor};
use std::sync::Arc;
use log::error;
use crate::api::endpoints::extract_accept_header::ExtractAcceptHeader;
use crate::api::endpoints::v1_api_config::v1_api_config_register;
use crate::model::InputSource;
use crate::repository::storage::get_geoip_path;
use crate::utils::GeoIp;
use crate::utils::request::download_text_content;
async fn create_ipinfo_check(app_state: &Arc<AppState>) -> Option<(Option<String>, Option<String>)> {
let config = app_state.app_config.config.load();
@@ -67,6 +73,58 @@ async fn streams(ExtractAcceptHeader(accept): ExtractAcceptHeader,
json_or_bin_response(accept.as_ref(), &streams).into_response()
}
async fn geoip_update(axum::extract::State(app_state): axum::extract::State<Arc<AppState>>) -> axum::response::Response {
let config = app_state.app_config.config.load();
if let Some(geoip) = config.reverse_proxy.as_ref().and_then(|r| r.geoip.as_ref()) {
if geoip.enabled {
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 input_source = InputSource {
url: geoip.url.clone(),
username: None,
password: None,
method: InputFetchMethod::GET,
headers: HashMap::default(),
};
return match download_text_content(Arc::clone(&app_state.http_client.load()), &input_source, None, None).await {
Ok((content, _)) => {
let reader = Cursor::new(content);
let mut geoip = GeoIp::new();
let result = {
match geoip.import_ipv4_from_csv(reader, geoip_db_path) {
Ok(size) => {
(Some(size), None)
}
Err(err) => (None, Some(err))
}
};
return match result {
(Some(_), None) => {
app_state.geoip.store(Some(Arc::new(geoip)));
axum::http::StatusCode::OK.into_response()
},
(None, Some(err)) => {
error!("Failed to process geoip db: {err}");
axum::http::StatusCode::NOT_FOUND.into_response()
},
_ => {
axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}
}
Err(err) => {
error!("Failed to download geoip db: {err}");
axum::http::StatusCode::BAD_REQUEST.into_response()
}
}
}
}
axum::http::StatusCode::BAD_REQUEST.into_response()
}
async fn ipinfo(axum::extract::State(app_state): axum::extract::State<Arc<AppState>>) -> axum::response::Response {
if let Some((ipv4, ipv6)) = create_ipinfo_check(&app_state).await {
let ipcheck = IpCheckDto {
@@ -87,6 +145,7 @@ pub fn v1_api_register(web_auth_enabled: bool, app_state: Arc<AppState>, web_ui_
router = router
.route("/status", axum::routing::get(status))
.route("/streams", axum::routing::get(streams))
.route("/geoip/update", axum::routing::get(geoip_update))
.route("/file/download", axum::routing::post(download_api::queue_download_file))
.route("/file/download/info", axum::routing::get(download_api::download_file_info))
.route("/ipinfo", axum::routing::get(ipinfo));
+1 -1
View File
@@ -63,7 +63,7 @@ async fn playlist_update(
let app_config = Arc::clone(&app_state.app_config);
let event_manager = Arc::clone(&app_state.event_manager);
let playlist_state = Arc::clone(&app_state.playlists);
tokio::spawn(playlist::exec_processing(Arc::clone(&app_state.http_client.load()), app_config, Arc::new(valid_targets), Some(event_manager), Some(playlist_state)));
playlist::exec_processing(Arc::clone(&app_state.http_client.load()), app_config, Arc::new(valid_targets), Some(event_manager), Some(playlist_state)).await;
axum::http::StatusCode::ACCEPTED.into_response()
}
Err(err) => {
+20 -33
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::try_unwrap_body;
use crate::api::api_utils::{create_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,
@@ -29,7 +29,7 @@ use axum::response::IntoResponse;
use bytes::Bytes;
use futures::stream::{self, StreamExt};
use futures::Stream;
use log::{debug, error, warn};
use log::{debug, error, log_enabled, warn};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use shared::error::create_tuliprox_error_result;
@@ -218,25 +218,22 @@ async fn xtream_player_api_stream(
api_req: &UserApiRequest,
stream_req: ApiStreamRequest<'_>,
) -> impl IntoResponse + Send {
debug!("Stream Request {stream_req:?} - {req_headers:?}");
if log_enabled!(log::Level::Debug) {
let message = format!("Client Request headers {req_headers:?}");
debug!("{}", sanitize_sensitive_info(&message));
}
let (user, target) = try_option_bad_request!(
get_user_target_by_credentials(
stream_req.username,
stream_req.password,
api_req,
app_state
),
get_user_target_by_credentials( stream_req.username, stream_req.password, api_req, app_state),
false,
format!(
"Could not find any user for xc stream {}",
stream_req.username
)
format!("Could not find any user for xc stream {}", stream_req.username)
);
if user.permission_denied(app_state) {
return create_custom_video_stream_response(
&app_state.app_config,
CustomVideoStreamType::UserAccountExpired,
)
.into_response();
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserAccountExpired,).into_response();
}
let target_name = &target.name;
@@ -248,24 +245,14 @@ 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_not_found!(
xtream_repository::xtream_get_item_for_stream_id(
virtual_id,
app_state,
&target,
None
).await,
xtream_repository::xtream_get_item_for_stream_id(virtual_id, app_state, &target, None).await,
true,
format!("Failed to read xtream item for stream id {}", virtual_id)
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()),
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
)
format!( "Cant find input for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context)
);
let (cluster, item_type) = if stream_req.context == ApiStreamContext::Timeshift {
@@ -274,7 +261,7 @@ async fn xtream_player_api_stream(
(pli.xtream_cluster, pli.item_type)
};
let session_key = format!("{fingerprint}{virtual_id}");
let session_key = create_fingerprint(fingerprint, &user.username, virtual_id);
let user_session = app_state
.active_users
.get_and_update_user_session(&user.username, &session_key).await;
@@ -443,7 +430,7 @@ async fn xtream_player_api_stream_with_token(
)
);
let session_key = format!("{fingerprint}{virtual_id}");
let session_key = create_fingerprint(fingerprint, "webui", virtual_id);
let is_hls_request =
pli.item_type == PlaylistItemType::LiveHls || stream_ext.as_deref() == Some(HLS_EXT);
+10 -8
View File
@@ -52,8 +52,11 @@ fn write_tlv_str(buf: &mut BytesMut, tag: u8, value: &str) {
if bytes.len() < 0x80 {
buf.put_u8(u8::try_from(bytes.len()).unwrap_or(0));
} else {
buf.put_u8(0x82);
buf.put_u16(u16::try_from(bytes.len()).unwrap_or(0));
let len = u16::try_from(bytes.len()).unwrap_or(0);
let byte_first = 0x80 | ((len & 0x7F) as u8);
let byte_second = ((len >> 7) & 0xFF) as u8;
buf.put_u8(byte_first);
buf.put_u8(byte_second);
}
buf.put_slice(bytes);
}
@@ -118,12 +121,11 @@ fn parse_tlv(cursor: &mut Cursor<&[u8]>) -> HashMap<u8, Vec<u8>> {
let len = if (len_buf[0] & 0x80) == 0 {
len_buf[0] as usize
} else {
let ext_len_bytes = (len_buf[0] & 0x7F) as usize;
let mut ext = vec![0u8; ext_len_bytes];
if Read::read_exact(cursor, &mut ext).is_err() {
let mut second_byte = [0u8; 1];
if Read::read_exact(cursor, &mut second_byte).is_err() {
break;
}
ext.iter().fold(0usize, |acc, b| (acc << 8) | (*b as usize))
((second_byte[0] as usize) << 7) + ((len_buf[0] & 0x7F) as usize)
};
@@ -212,7 +214,7 @@ async fn handle_tcp_connection(
Ok(0) => return, // Connection closed
Ok(n) => {
let request_data = &buf[..n];
if request_data.len() < 4 {
if request_data.len() < 8 {
continue;
}
@@ -221,7 +223,7 @@ async fn handle_tcp_connection(
if msg_type == packet::HDHOMERUN_TYPE_GETSET_REQ {
let response = process_getset_request(request_data, &app_state).await;
if response.is_empty() {
error!("Protocol error or oinvalid request");
error!("Protocol error or invalid request");
return;
}
if let Err(e) = stream.write_all(&response).await {
+33 -13
View File
@@ -36,19 +36,39 @@ async fn ssdp_task_loop(socket: UdpSocket, app_config: Arc<AppConfig>, server_ho
};
let request = String::from_utf8_lossy(&buf[..len]);
if request.starts_with("M-SEARCH") && (request.contains("urn:schemas-upnp-org:device:MediaServer:1") || request.contains("ssdp:all")) {
trace!("Received HDHomeRun M-SEARCH from {remote_addr}");
let hdhomerun_guard = app_config.hdhomerun.load();
if let Some(hd_config) = &*hdhomerun_guard {
if hd_config.enabled {
for device in &hd_config.devices {
if device.t_enabled {
let response = create_ssdp_response(device, &server_host);
if let Err(e) = socket.send_to(response.as_bytes(), remote_addr).await {
error!("Failed to send SSDP response to {remote_addr}: {e}");
} else {
trace!("Sent SSDP response for device '{}' to {remote_addr}", device.name);
}
if !request.starts_with("M-SEARCH") { continue; }
let req = request.to_ascii_lowercase();
if !req.contains(r#"man: "ssdp:discover""#) { continue; }
// Extract ST and MX (defaults)
let st = req.lines()
.find_map(|l| l.strip_prefix("st:").map(|v| v.trim().to_string()))
.unwrap_or_else(|| "ssdp:all".to_string());
let mx: u64 = req.lines()
.find_map(|l| l.strip_prefix("mx:").and_then(|v| v.trim().parse().ok()))
.unwrap_or(1);
// Normalize to the set we support
let supported = [
"urn:schemas-upnp-org:device:mediaserver:1",
"upnp:rootdevice",
"ssdp:all",
];
if !supported.contains(&st.as_str()) && st != "ssdp:all" { continue; }
// Randomized delay per MX
let delay_ms = (fastrand::u64(0..=mx*1000)).min(2000);
if delay_ms > 0 { tokio::time::sleep(Duration::from_millis(delay_ms)).await; }
trace!("Received HDHomeRun M-SEARCH from {remote_addr}");
let hdhomerun_guard = app_config.hdhomerun.load();
if let Some(hd_config) = &*hdhomerun_guard {
if hd_config.enabled {
for device in &hd_config.devices {
if device.t_enabled {
let response = create_ssdp_response(device, &server_host);
if let Err(e) = socket.send_to(response.as_bytes(), remote_addr).await {
error!("Failed to send SSDP response to {remote_addr}: {e}");
} else {
trace!("Sent SSDP response for device '{}' to {remote_addr}", device.name);
}
}
}
+23 -8
View File
@@ -32,6 +32,8 @@ use std::sync::atomic::AtomicI8;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tower_governor::key_extractor::SmartIpKeyExtractor;
use crate::repository::storage::get_geoip_path;
use crate::utils::GeoIp;
fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result<PathBuf, std::io::Error> {
let web_dir = web_root.to_string();
@@ -66,21 +68,33 @@ fn create_shared_data(
forced_targets: &Arc<ProcessTargets>,
) -> AppState {
let config = app_config.config.load();
let use_geoip = config.is_geoip_enabled();
let geoip = if use_geoip {
let path = get_geoip_path(&config.working_dir);
let _file_lock = app_config.file_locks.read_lock(&path);
let geoip = GeoIp::load(&path).ok();
if geoip.is_some() {
info!("GeoIp db loaded");
}
Arc::new(ArcSwapOption::from_pointee(geoip))
} else {
Arc::new(ArcSwapOption::from(None))
};
let cache = create_cache(&config);
let shared_stream_manager = Arc::new(SharedStreamManager::new());
let (provider_change_tx, provider_change_rx) = tokio::sync::mpsc::channel(10);
let (provider_change_tx, provider_change_rx) = tokio::sync::mpsc::unbounded_channel();
let active_provider = Arc::new(ActiveProviderManager::new(app_config, provider_change_tx));
let (active_user_change_tx, active_user_change_rx) = tokio::sync::mpsc::channel(10);
let active_users = Arc::new(ActiveUserManager::new(
let (active_user_change_tx, active_user_change_rx) = tokio::sync::mpsc::unbounded_channel();
let active_users = ActiveUserManager::new(
&config,
&shared_stream_manager,
&active_provider,
&geoip,
active_user_change_tx,
));
let event_manager = Arc::new(EventManager::new(
active_user_change_rx,
provider_change_rx,
));
);
let event_manager = Arc::new(EventManager::new(active_user_change_rx, provider_change_rx, ));
let client = create_http_client(app_config);
AppState {
@@ -95,6 +109,7 @@ fn create_shared_data(
event_manager,
cancel_tokens: Arc::new(ArcSwap::from_pointee(CancelTokens::default())),
playlists: Arc::new(PlaylistStorageState::new()),
geoip,
}
}
@@ -1,7 +1,7 @@
use crate::api::model::{ProviderConfig, ProviderConfigConnection, ProviderConfigWrapper, ProviderConnectionChangeCallback, ProviderConnectionChangeSender};
use crate::model::{AppConfig, ConfigInput};
use arc_swap::ArcSwap;
use log::{debug, log_enabled, trace};
use log::{debug, error, log_enabled, trace};
use shared::utils::{default_grace_period_millis, default_grace_period_timeout_secs, display_vec, sanitize_sensitive_info};
use std::collections::HashMap;
use std::fmt;
@@ -10,6 +10,7 @@ use std::sync::atomic::{AtomicU64, AtomicU8, AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::RwLock;
use crate::utils::debug_if_enabled;
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
const CONNECTION_STATE_ACTIVE: u8 = 0;
const CONNECTION_STATE_SHARED: u8 = 1;
@@ -17,9 +18,14 @@ const CONNECTION_STATE_RELEASED: u8 = 2;
pub struct ProviderConnectionGuard {
allocation: ProviderAllocation,
release_tx: UnboundedSender<Arc<ProviderConfig>>,
}
impl ProviderConnectionGuard {
pub fn new(allocation: ProviderAllocation, release_tx: UnboundedSender<Arc<ProviderConfig>>) -> Self {
Self { allocation, release_tx }
}
// for shared streams, we need to disable release
// The connection should be released when all shared streams close!
pub(crate) fn disable_release(&self) {
@@ -31,6 +37,17 @@ impl ProviderConnectionGuard {
}
}
}
fn send_release(&self, config: &Arc<ProviderConfig>) {
let provider_config = Arc::clone(config);
if let Err(_err) = &self.release_tx.send(Arc::clone(config)) {
// Fallback
tokio::spawn(async move {
provider_config.release().await;
});
}
}
pub(crate) fn release(&self) {
match &self.allocation {
ProviderAllocation::Exhausted => {}
@@ -38,11 +55,7 @@ impl ProviderConnectionGuard {
ProviderAllocation::GracePeriod(state, config) => {
// we can't release shared state
if state.compare_exchange(CONNECTION_STATE_ACTIVE, CONNECTION_STATE_RELEASED, Ordering::SeqCst, Ordering::SeqCst).is_ok() {
let provider_config = Arc::clone(config);
trace!("Releasing provider connection {:?}", provider_config.name);
tokio::spawn(async move {
provider_config.release().await;
});
self.send_release(config);
}
}
}
@@ -56,23 +69,12 @@ impl ProviderConnectionGuard {
| ProviderAllocation::GracePeriod(state, config) => {
if state.load(Ordering::SeqCst) < CONNECTION_STATE_RELEASED {
state.store(CONNECTION_STATE_RELEASED, Ordering::SeqCst);
let provider_config = Arc::clone(config);
trace!("Forced releasing provider connection {:?}", provider_config.name);
tokio::spawn(async move {
provider_config.release().await;
});
trace!("Forced releasing provider connection {:?}", config.name);
self.send_release(config);
}
}
}
}
}
impl ProviderConnectionGuard {
pub fn new(allocation: ProviderAllocation) -> Self {
Self {
allocation,
}
}
pub fn get_provider_name(&self) -> Option<String> {
match self.allocation {
@@ -533,10 +535,10 @@ impl ProviderLineupManager {
let cfg_name = cfg_input.name.clone();
let on_connection_change: ProviderConnectionChangeCallback = Arc::new(move |_name: &str, connections: usize| {
let connection_change_sender = connection_change_sender.clone();
let cfg_name = cfg_name.clone();
tokio::spawn(async move {
let _ = connection_change_sender.send((cfg_name, connections)).await;
});
let provider_cfg_name = cfg_name.clone();
if let Err(err) = connection_change_sender.send((provider_cfg_name, connections)) {
error!("Failed to send connection change: {cfg_name}: {connections}, {err}");
}
});
let on_connection_change = Arc::new(on_connection_change);
@@ -682,18 +684,18 @@ impl ProviderLineupManager {
None
}
async fn force_exact_acquire_connection(&self, provider_name: &str) -> Arc<ProviderConnectionGuard> {
async fn force_exact_acquire_connection(&self, provider_name: &str, release_tx: UnboundedSender<Arc<ProviderConfig>>) -> Arc<ProviderConnectionGuard> {
let providers = self.providers.load();
let allocation = match Self::get_provider_config(provider_name, &providers) {
None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider
Some((_lineup, config)) => config.force_allocate().await,
};
Arc::new(ProviderConnectionGuard::new(allocation))
Arc::new(ProviderConnectionGuard::new(allocation, release_tx))
}
// Returns the next available provider connection
async fn acquire_connection(&self, input_name: &str) -> Arc<ProviderConnectionGuard> {
async fn acquire_connection(&self, input_name: &str, release_tx: UnboundedSender<Arc<ProviderConfig>>) -> Arc<ProviderConnectionGuard> {
let providers = self.providers.load();
let allocation = match Self::get_provider_config(input_name, &providers) {
None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider
@@ -711,7 +713,7 @@ impl ProviderLineupManager {
}
}
Arc::new(ProviderConnectionGuard::new(allocation))
Arc::new(ProviderConnectionGuard::new(allocation, release_tx))
}
// This method is used for redirects to cycle through provider
@@ -780,19 +782,36 @@ impl ProviderLineupManager {
pub struct ActiveProviderManager {
providers: ProviderLineupManager,
connections: RwLock<HashMap<String, Arc<ProviderConnectionGuard>>>,
release_tx: UnboundedSender<Arc<ProviderConfig>>,
}
impl ActiveProviderManager {
pub fn new(cfg: &AppConfig, connection_change_sender: ProviderConnectionChangeSender) -> Self {
let (grace_period_millis, grace_period_timeout_secs) = Self::get_grace_options(cfg);
let inputs = Self::get_config_inputs(cfg);
// Create the cleanup channel
let (cleanup_tx, mut cleanup_rx) = unbounded_channel::<Arc<ProviderConfig>>();
// Spawn the async cleanup worker
tokio::spawn(async move {
while let Some(cfg) = cleanup_rx.recv().await {
debug!("🧹 Provider connection releasing {:?}", cfg.name);
cfg.release().await;
}
debug!("Provider cleanup worker terminated");
});
Self {
providers: ProviderLineupManager::new(inputs, grace_period_millis, grace_period_timeout_secs, connection_change_sender),
connections: RwLock::new(HashMap::new()),
release_tx: cleanup_tx,
}
}
fn release_sender(&self) -> UnboundedSender<Arc<ProviderConfig>> {
self.release_tx.clone()
}
fn get_config_inputs(cfg: &AppConfig) -> Vec<Arc<ConfigInput>> {
cfg.sources.load().sources
.iter().flat_map(|s| s.inputs.iter().map(Arc::clone)).collect()
@@ -812,14 +831,14 @@ impl ActiveProviderManager {
}
pub async fn force_exact_acquire_connection(&self, provider_name: &str, addr: &str) -> Arc<ProviderConnectionGuard> {
let guard = self.providers.force_exact_acquire_connection(provider_name).await;
let guard = self.providers.force_exact_acquire_connection(provider_name, self.release_sender()).await;
self.register_connection(addr, &guard).await;
guard
}
// Returns the next available provider connection
pub async fn acquire_connection(&self, input_name: &str, addr: &str) -> Arc<ProviderConnectionGuard> {
let guard = self.providers.acquire_connection(input_name).await;
let guard = self.providers.acquire_connection(input_name, self.release_sender()).await;
self.register_connection(addr, &guard).await;
guard
}
@@ -844,7 +863,7 @@ impl ActiveProviderManager {
async fn register_connection(&self, addr: &str, guard: &Arc<ProviderConnectionGuard>) {
if !matches!(guard.allocation, ProviderAllocation::Exhausted) {
trace!("Added provider connection {:?}", guard.get_provider_name().unwrap_or_default());
debug!("🌻🌻🌻 Added provider connection {:?} for {addr}", guard.get_provider_name().unwrap_or_default());
self.connections.write().await.insert(addr.to_string(), Arc::clone(guard));
}
}
@@ -852,6 +871,7 @@ impl ActiveProviderManager {
pub async fn release_connection(&self, addr: &str) {
let guard = self.connections.write().await.remove(addr);
if let Some(guard) = guard {
debug!("🌸🌸🌸 Released provider connection {:?} for {addr}", guard.get_provider_name().unwrap_or_default());
guard.release();
}
}
+95 -37
View File
@@ -1,23 +1,26 @@
use crate::api::model::ActiveProviderManager;
use std::borrow::Cow;
use crate::api::model::{ActiveProviderManager};
use crate::api::model::SharedStreamManager;
use crate::model::Config;
use crate::model::ProxyUserCredentials;
use jsonwebtoken::get_current_timestamp;
use log::{debug, error, info};
use shared::model::{ActiveUserConnectionChange, StreamChannel, StreamInfo, UserConnectionPermission};
use shared::utils::{current_time_secs, default_grace_period_millis, default_grace_period_timeout_secs, sanitize_sensitive_info};
use shared::utils::{current_time_secs, default_grace_period_millis, default_grace_period_timeout_secs, sanitize_sensitive_info, strip_port};
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use arc_swap::ArcSwapOption;
use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
use tokio::sync::RwLock;
use crate::utils::GeoIp;
const USER_GC_TTL: u64 = 900; // 15 Min
const USER_CON_TTL: u64 = 10_800; // 3 hours
const USER_SESSION_LIMIT: usize = 50;
type ActiveUserConnectionChangeSender = tokio::sync::mpsc::Sender<ActiveUserConnectionChange>;
pub type ActiveUserConnectionChangeReceiver = tokio::sync::mpsc::Receiver<ActiveUserConnectionChange>;
type ActiveUserConnectionChangeSender = tokio::sync::mpsc::UnboundedSender<ActiveUserConnectionChange>;
pub type ActiveUserConnectionChangeReceiver = tokio::sync::mpsc::UnboundedReceiver<ActiveUserConnectionChange>;
macro_rules! active_user_manager_shared_impl {
() => {
@@ -38,7 +41,9 @@ macro_rules! active_user_manager_shared_impl {
let is_log_user_enabled = self.is_log_user_enabled();
let user_connection_count = Self::get_active_connections(&user).await;
let user_count = user.read().await.iter().filter(|(_, c)| c.connections > 0).count();
let _= self.connection_change_tx.try_send(ActiveUserConnectionChange::Connections(user_count, user_connection_count));
if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Connections(user_count, user_connection_count)) {
error!("Failed to send active user connection change: user-count: {user_count}, user-connection-count: {user_connection_count} {err:?}");
}
if is_log_user_enabled {
info!("Active Users: {user_count}, Active User Connections: {user_connection_count}");
}
@@ -50,24 +55,29 @@ macro_rules! active_user_manager_shared_impl {
};
if let Some(username) = username_opt {
let mut user = self.user.write().await;
if let Some(connection_data) = user.get_mut(&username) {
if connection_data.connections > 0 {
connection_data.connections -= 1;
}
{
let mut user = self.user.write().await;
if let Some(connection_data) = user.get_mut(&username) {
if connection_data.connections > 0 {
connection_data.connections -= 1;
}
if connection_data.connections < connection_data.max_connections {
connection_data.granted_grace = false;
connection_data.grace_ts = 0;
if connection_data.connections < connection_data.max_connections {
connection_data.granted_grace = false;
connection_data.grace_ts = 0;
}
connection_data.streams.retain(|c| c.addr != addr);
}
connection_data.streams.retain(|c| c.addr != addr);
}
}
self.drop_connection(&addr);
self.shared_stream_manager.release_connection(addr, true).await;
self.provider_manager.release_connection(addr).await;
let _= self.connection_change_tx.try_send(ActiveUserConnectionChange::Disconnected(addr.to_string()));
self.log_active_user().await;
self.drop_connection(&addr);
self.shared_stream_manager.release_connection(addr, true).await;
self.provider_manager.release_connection(addr).await;
if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Disconnected(addr.to_string())) {
error!("Failed to send active user connection change: {err:?}");
}
self.log_active_user().await;
}
}
};
}
@@ -79,7 +89,7 @@ fn get_grace_options(config: &Config) -> (u64, u64) {
(grace_period_millis, grace_period_timeout_secs)
}
struct ConnectionGuardUserManager {
pub struct ConnectionGuardUserManager {
log_active_user: bool,
user: Arc<RwLock<HashMap<String, UserConnectionData>>>,
user_by_addr: Arc<RwLock<HashMap<String, String>>>,
@@ -100,18 +110,28 @@ pub struct UserConnectionGuard {
manager: Arc<ConnectionGuardUserManager>,
// username: String,
addr: String,
release_tx: UnboundedSender<String>,
}
impl UserConnectionGuard {
pub fn new(manager: Arc<ConnectionGuardUserManager>, addr: &str, release_tx: UnboundedSender<String>) -> Self {
Self {
manager,
addr: addr.to_string(),
release_tx,
}
}
}
impl Drop for UserConnectionGuard {
fn drop(&mut self) {
let manager = self.manager.clone();
let addr = self.addr.clone();
if let Ok(rt) = tokio::runtime::Handle::try_current() {
rt.spawn(async move {
if let Err(_err) = self.release_tx.send(addr.clone()) {
// fallback
tokio::spawn(async move {
manager.remove_connection(&addr).await;
});
} else {
// Fallback: no runtime
error!("Runtime not available, cannot cleanly remove connection for {addr}");
}
}
}
@@ -172,15 +192,24 @@ pub struct ActiveUserManager {
close_signal_tx: tokio::sync::broadcast::Sender<String>,
shared_stream_manager: Arc<SharedStreamManager>,
provider_manager: Arc<ActiveProviderManager>,
geo_ip: Arc<ArcSwapOption<GeoIp>>,
connection_change_tx: ActiveUserConnectionChangeSender,
release_tx: UnboundedSender<String>,
}
impl ActiveUserManager {
pub fn new(config: &Config, shared_stream_manager: &Arc<SharedStreamManager>, provider_manager: &Arc<ActiveProviderManager>, connection_change_tx: ActiveUserConnectionChangeSender) -> Self {
let log_active_user = config.log.as_ref().is_some_and(|l| l.log_active_user);
pub fn new(config: &Config, shared_stream_manager: &Arc<SharedStreamManager>,
provider_manager: &Arc<ActiveProviderManager>,
geoip: &Arc<ArcSwapOption<GeoIp>>,
connection_change_tx: ActiveUserConnectionChangeSender) -> Arc<Self> {
let log_active_user: bool = config.log.as_ref().is_some_and(|l| l.log_active_user);
let (grace_period_millis, grace_period_timeout_secs) = get_grace_options(config);
let (close_signal_tx, _) = tokio::sync::broadcast::channel(10);
Self {
// Create the cleanup channel
let (cleanup_tx, mut cleanup_rx) = unbounded_channel::<String>();
let active_user_manager = Arc::new(Self {
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),
@@ -190,12 +219,31 @@ impl ActiveUserManager {
close_signal_tx,
shared_stream_manager: Arc::clone(shared_stream_manager),
provider_manager: Arc::clone(provider_manager),
geo_ip: Arc::clone(geoip),
connection_change_tx,
}
release_tx: cleanup_tx,
});
let active_user_manager_clone = Arc::clone(&active_user_manager);
// Spawn the async cleanup worker
tokio::spawn(async move {
while let Some(addr) = cleanup_rx.recv().await {
debug!("🧹 User manager - connection releasing {addr:?}");
active_user_manager_clone.remove_connection(&addr).await;
}
debug!("User manager - cleanup worker terminated");
});
active_user_manager
}
active_user_manager_shared_impl!();
pub fn release_sender(&self) -> UnboundedSender<String> {
self.release_tx.clone()
}
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);
@@ -280,12 +328,23 @@ 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) -> UserConnectionGuard {
pub async fn add_connection(&self, username: &str, max_connections: u32, addr: &str, 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))
} else {
None
}
};
let stream_info = StreamInfo::new(
username,
addr,
provider,
stream_channel,
user_agent.to_string(),
country,
);
{
let mut user_map = self.user.write().await;
@@ -305,13 +364,12 @@ impl ActiveUserManager {
user_by_addr.insert(addr.to_owned(), username.to_owned());
}
let _= self.connection_change_tx.try_send(ActiveUserConnectionChange::Connected(stream_info));
if let Err(err) = self.connection_change_tx.send(ActiveUserConnectionChange::Connected(stream_info)) {
error!("Failed to send connection change: {err}");
}
self.log_active_user().await;
UserConnectionGuard {
manager: Arc::new(self.clone_inner()),
addr: addr.to_owned(),
}
UserConnectionGuard::new(Arc::new(self.clone_inner()), addr, self.release_sender())
}
fn is_log_user_enabled(&self) -> bool {
+28 -1
View File
@@ -24,6 +24,8 @@ use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;
use crate::repository::storage::get_geoip_path;
use crate::utils::GeoIp;
macro_rules! cancel_service {
($field: ident, $changes:expr, $cancel_tokens:expr) => {
@@ -58,16 +60,18 @@ struct TargetChanges {
target: Arc<ConfigTarget>,
}
#[allow(clippy::struct_excessive_bools)]
pub(in crate::api) struct UpdateChanges {
scheduler: bool,
hdhomerun: bool,
file_watch: bool,
geoip: bool,
targets: Option<HashMap<String, TargetChanges>>,
}
impl UpdateChanges {
pub(in crate::api) fn modified(&self) -> bool {
self.scheduler || self.hdhomerun || self.file_watch
self.scheduler || self.hdhomerun || self.file_watch || self.geoip
}
}
@@ -258,6 +262,7 @@ pub struct AppState {
pub event_manager: Arc<EventManager>,
pub cancel_tokens: Arc<ArcSwap<CancelTokens>>,
pub playlists: Arc<PlaylistStorageState>,
pub geoip: Arc<ArcSwapOption<GeoIp>>
}
impl AppState {
@@ -267,12 +272,29 @@ impl AppState {
) -> Result<UpdateChanges, TuliproxError> {
let changes = self.detect_changes_for_config(&config);
config.update_runtime();
let use_geoip = config.is_geoip_enabled();
let working_dir = config.working_dir.clone();
self.active_users.update_config(&config);
self.app_config.set_config(config)?;
self.active_provider
.update_config(&self.app_config)
.await;
self.update_config().await;
if changes.geoip {
let new_geoip = if use_geoip {
let path = get_geoip_path(&working_dir);
let _file_lock = self.app_config.file_locks.read_lock(&path);
GeoIp::load(&path).ok().map(Arc::new)
} else {
None
};
self.geoip.store(new_geoip);
}
Ok(changes)
}
@@ -340,11 +362,15 @@ impl AppState {
config.mapping_path.as_ref()
);
let geoip_enabled = config.is_geoip_enabled();
let geoip_enabled_old = old_config.is_geoip_enabled();
UpdateChanges {
scheduler: changed_schedules,
hdhomerun: changed_hdhomerun,
file_watch: changed_file_watch,
targets: None,
geoip: geoip_enabled != geoip_enabled_old,
}
}
@@ -412,6 +438,7 @@ impl AppState {
scheduler: false,
hdhomerun: false,
file_watch: file_watch_changed,
geoip: false,
targets: Some(target_changes),
}
}
+1 -2
View File
@@ -1,5 +1,4 @@
use log::{info, trace};
use tokio::task;
use shared::model::{ActiveUserConnectionChange, ConfigType, PlaylistUpdateState};
use crate::api::model::{ActiveUserConnectionChangeReceiver};
use crate::api::model::{ProviderConnectionChangeReceiver};
@@ -27,7 +26,7 @@ impl EventManager {
let (channel_tx, _channel_rx) = tokio::sync::broadcast::channel(10);
let channel_tx_clone = channel_tx.clone();
task::spawn(async move {
tokio::spawn(async move {
loop {
tokio::select! {
Some(event) = active_user_change_rx.recv() => {
+6 -7
View File
@@ -9,8 +9,8 @@ use tokio::sync::RwLock;
use shared::model::InputType;
use shared::write_if_some;
pub type ProviderConnectionChangeSender = tokio::sync::mpsc::Sender<(String, usize)>;
pub type ProviderConnectionChangeReceiver = tokio::sync::mpsc::Receiver<(String, usize)>;
pub type ProviderConnectionChangeSender = tokio::sync::mpsc::UnboundedSender<(String, usize)>;
pub type ProviderConnectionChangeReceiver = tokio::sync::mpsc::UnboundedReceiver<(String, usize)>;
pub type ProviderConnectionChangeCallback = Arc<dyn Fn(&str, usize) + Send + Sync>;
@@ -261,13 +261,12 @@ impl ProviderConfig {
pub async fn release(&self) {
let mut guard = self.connection.write().await;
if guard.current_connections > 0 {
if guard.current_connections == 1 && self.max_connections > 1 {
guard.granted_grace = false;
guard.grace_ts = 0;
}
modify_connections!(self, guard, -1);
}
if guard.current_connections == 0 || guard.current_connections < self.max_connections {
guard.granted_grace = false;
guard.grace_ts = 0;
}
}
#[inline]
@@ -15,6 +15,8 @@ use std::pin::Pin;
use std::sync::atomic::AtomicU8;
use std::sync::{Arc};
use std::task::{Poll};
use axum::http::header::USER_AGENT;
use axum::http::HeaderMap;
use futures::task::AtomicWaker;
const INNER_STREAM: u8 = 0_u8;
@@ -39,19 +41,16 @@ impl ActiveClientStream {
user: &ProxyUserCredentials,
connection_permission: UserConnectionPermission,
addr: &str,
stream_channel: StreamChannel) -> Self {
stream_channel: StreamChannel,
req_headers: &HeaderMap) -> Self {
if connection_permission == UserConnectionPermission::Exhausted {
error!("Something is wrong this should not happen");
}
let grant_user_grace_period = connection_permission == UserConnectionPermission::GracePeriod;
let username = user.username.as_str();
let provider_name = stream_details
.provider_connection_guard
.as_ref()
.and_then(|guard| guard.get_provider_name())
.as_deref()
.map_or_else(String::new, ToString::to_string);
let user_connection_guard = Some(app_state.active_users.add_connection(username, user.max_connections, addr, &provider_name, stream_channel).await);
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 cfg = &app_state.app_config;
let waker = Some(Arc::new(AtomicWaker::new()));
let waker_clone = waker.clone();
@@ -106,8 +105,8 @@ impl ActiveClientStream {
let active_provider = Arc::clone(&app_state.active_provider);
let shared_stream_manager = Arc::clone(&app_state.shared_stream_manager);
let provider_grace_check = if stream_details.has_grace_period() && stream_details.input_name.is_some() {
let provider_name = stream_details.input_name.as_deref().unwrap_or_default().to_string();
let provider_grace_check = if stream_details.has_grace_period() && stream_details.provider_name.is_some() {
let provider_name = stream_details.provider_name.as_ref().map_or_else(String::new, ToString::to_string);
Some(provider_name)
} else {
None
@@ -128,7 +128,7 @@ impl SharedStreamState {
}
}
async fn subscribe(&self, addr: &str, manager: Arc<SharedStreamManager>) -> BoxedProviderStream {
async fn subscribe(&self, addr: &str, manager: Arc<SharedStreamManager>) -> (BoxedProviderStream, Option<String>) {
let (client_tx, client_rx) = mpsc::channel(self.buf_size);
let mut broadcast_rx = self.broadcaster.subscribe();
let cancel_token = CancellationToken::new();
@@ -147,8 +147,8 @@ impl SharedStreamState {
let mut loop_cnt = 0;
loop {
loop_cnt += 1;
tokio::select! {
loop_cnt += 1;
tokio::select! {
biased;
() = cancel_token.cancelled() => {
@@ -181,7 +181,13 @@ impl SharedStreamState {
}
manager.release_connection(&address, false).await;
});
convert_stream(ReceiverStream::new(client_rx).boxed())
let provider = match &self.provider_guard {
None => None,
Some(connection_guard) => connection_guard.get_provider_name()
};
(convert_stream(ReceiverStream::new(client_rx).boxed()), provider)
}
fn broadcast<S, E>(
@@ -300,6 +306,7 @@ impl SharedStreamManager {
}
pub async fn release_connection(&self, addr: &str, send_stop_signal: bool) {
let stream_url = {
self.shared_streams_by_addr.write().await.remove(addr)
};
@@ -336,7 +343,7 @@ impl SharedStreamManager {
}
async fn subscribe_stream(&self, stream_url: &str, addr: Option<&str>, manager: Arc<SharedStreamManager>) -> Option<BoxedProviderStream> {
async fn subscribe_stream(&self, stream_url: &str, addr: Option<&str>, manager: Arc<SharedStreamManager>) -> Option<(BoxedProviderStream, Option<String>)> {
let shared_stream_state = self.shared_streams.read().await.get(stream_url).map(Arc::clone);
match shared_stream_state {
None => None,
@@ -344,8 +351,7 @@ impl SharedStreamManager {
if let Some(address) = addr {
debug_if_enabled!("Responding to existing shared client stream {}", sanitize_sensitive_info(stream_url));
self.shared_streams_by_addr.write().await.insert(address.to_string(), stream_url.to_owned());
let stream = stream_state.subscribe(address, manager).await;
Some(stream)
Some(stream_state.subscribe(address, manager).await)
} else {
None
}
@@ -364,7 +370,7 @@ impl SharedStreamManager {
addr: Option<&str>,
headers: Vec<(String, String)>,
buffer_size: usize,
provider_guard: Option<Arc<ProviderConnectionGuard>>) -> Option<BoxedProviderStream>
provider_guard: Option<Arc<ProviderConnectionGuard>>) -> Option<(BoxedProviderStream, Option<String>)>
where
S: Stream<Item=Result<Bytes, E>> + Unpin + 'static + Send,
E: std::fmt::Debug + Send,
@@ -383,7 +389,7 @@ impl SharedStreamManager {
app_state: &AppState,
stream_url: &str,
addr: Option<&str>,
) -> Option<BoxedProviderStream> {
) -> Option<(BoxedProviderStream, Option<String>)> {
let manager = Arc::clone(&app_state.shared_stream_manager);
app_state.shared_stream_manager.subscribe_stream(stream_url, addr, manager).await
}
+9 -7
View File
@@ -134,11 +134,7 @@ where
let user_manager_clone = Arc::clone(&user_manager);
let mut addr_close_rx = user_manager_clone.get_close_connection_channel();
let connection_closed = async move || {
debug!("Connection closed: {remote_addr}");
let addr = remote_addr.to_string();
user_manager_clone.remove_connection(&addr).await;
};
let connection_release = user_manager.release_sender();
debug!("Connection opened: {addr_str}");
@@ -148,11 +144,17 @@ where
if let Err(err) = result {
trace!("failed to serve connection: {err:#}");
}
connection_closed().await;
if let Err(_err) = connection_release.send(remote_addr.to_string()) {
let addr = remote_addr.to_string();
user_manager_clone.remove_connection(&addr).await;
}
break;
}
() = &mut signal_closed => {
connection_closed().await;
if let Err(_err) = connection_release.send(remote_addr.to_string()) {
let addr = remote_addr.to_string();
user_manager_clone.remove_connection(&addr).await;
}
debug!("Connection gracefully closed: {remote_addr}");
conn.as_mut().graceful_shutdown();
}
+6 -8
View File
@@ -59,18 +59,16 @@ impl Fingerprint {
}
}
let client_ip = format!("{}:{}", real_ip.as_ref()
let client_ip = real_ip.as_ref()
.map(ToString::to_string)
.or(forwarded_for.as_ref().map(ToString::to_string))
.unwrap_or_else(|| addr.ip().to_string()),
addr.port());
.unwrap_or_else(|| addr.ip().to_string());
let client_ip_port =format!("{client_ip}:{}", addr.port());
let ua = user_agent.unwrap_or_else(String::new);
let key = match real_ip.or(forwarded_for) {
Some(xff) => format!("{}{xff}{ua}", addr.ip()),
None => format!("{}{ua}", addr.ip()),
};
let key = format!("{client_ip }{ua}");
Ok(Fingerprint(key, client_ip))
Ok(Fingerprint(key, client_ip_port))
}
}
+42 -50
View File
@@ -11,27 +11,23 @@ fn is_enabled(kind: MsgKind, cfg: &MessagingConfig) -> bool {
cfg.notify_on.contains(&kind)
}
fn send_http_post_request(client: &Arc<reqwest::Client>, msg: &str, messaging: &MessagingConfig) {
async fn send_http_post_request(client: &Arc<reqwest::Client>, msg: &str, messaging: &MessagingConfig) {
if let Some(rest) = &messaging.rest {
let url = rest.url.clone();
let data = msg.to_owned();
let the_client = Arc::clone(client);
tokio::spawn(async move {
match the_client
.post(&url)
.header(header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(data)
.send()
.await
{
Ok(_) => debug!("Text message sent successfully to rest api"),
Err(e) => error!("Text message wasn't sent to rest api because of: {e}"),
}
});
let data = msg.to_owned();
match client
.post(&rest.url)
.header(header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(data)
.send()
.await
{
Ok(_) => debug!("Text message sent successfully to rest api"),
Err(e) => error!("Text message wasn't sent to rest api because of: {e}"),
}
}
}
fn send_telegram_message(client: &Arc<reqwest::Client>, msg: &str, messaging: &MessagingConfig, json: bool) {
async fn send_telegram_message(client: &Arc<reqwest::Client>, msg: &str, messaging: &MessagingConfig, json: bool) {
// TODO use proxy settings
if let Some(telegram) = &messaging.telegram {
let (message, options) = {
@@ -48,55 +44,51 @@ fn send_telegram_message(client: &Arc<reqwest::Client>, msg: &str, messaging: &M
for chat_id in &telegram.chat_ids {
let bot = telegram_create_instance(&telegram.bot_token, chat_id);
telegram_send_message(client, &bot, &message, options.as_ref());
telegram_send_message(client, &bot, &message, options.as_ref()).await;
}
}
}
fn send_pushover_message(client: &Arc<reqwest::Client>, msg: &str, messaging: &MessagingConfig) {
async fn send_pushover_message(client: &Arc<reqwest::Client>, msg: &str, messaging: &MessagingConfig) {
if let Some(pushover) = &messaging.pushover {
let encoded_message: String = url::form_urlencoded::Serializer::new(String::new())
.append_pair("token", pushover.token.as_str())
.append_pair("user", pushover.user.as_str())
.append_pair("message", msg)
.finish();
let the_client = Arc::clone(client);
let pushover_url = pushover.url.clone();
tokio::spawn(async move {
match the_client
.post(pushover_url)
.header(header::CONTENT_TYPE, mime::APPLICATION_WWW_FORM_URLENCODED.to_string())
.body(encoded_message)
.send()
.await
{
Ok(response) => {
if response.status().is_success() {
debug!("Text message sent successfully to PUSHOVER, status code {}", response.status());
} else {
error!("Failed to send text message to PUSHOVER, status code {}", response.status());
}
match client
.post(&pushover.url)
.header(header::CONTENT_TYPE, mime::APPLICATION_WWW_FORM_URLENCODED.to_string())
.body(encoded_message)
.send()
.await
{
Ok(response) => {
if response.status().is_success() {
debug!("Text message sent successfully to PUSHOVER, status code {}", response.status());
} else {
error!("Failed to send text message to PUSHOVER, status code {}", response.status());
}
Err(e) => error!("Text message wasn't sent to PUSHOVER api because of: {e}"),
}
});
}
}
fn dispatch_send_message(client: &Arc<reqwest::Client>, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str, json: bool) {
if let Some(messaging) = cfg {
if is_enabled(kind, messaging) {
send_telegram_message(client, msg, messaging, json);
send_http_post_request(client, msg, messaging);
send_pushover_message(client, msg, messaging);
Err(e) => error!("Text message wasn't sent to PUSHOVER api because of: {e}"),
}
}
}
pub fn send_message_json(client: &Arc<reqwest::Client>, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str) {
dispatch_send_message(client, kind, cfg, msg, true);
async fn dispatch_send_message(client: &Arc<reqwest::Client>, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str, json: bool) {
if let Some(messaging) = cfg {
if is_enabled(kind, messaging) {
send_telegram_message(client, msg, messaging, json).await;
send_http_post_request(client, msg, messaging).await;
send_pushover_message(client, msg, messaging).await;
}
}
}
pub fn send_message(client: &Arc<reqwest::Client>, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str) {
dispatch_send_message(client, kind, cfg, msg, false);
pub async fn send_message_json(client: &Arc<reqwest::Client>, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str) {
dispatch_send_message(client, kind, cfg, msg, true).await;
}
pub async fn send_message(client: &Arc<reqwest::Client>, kind: MsgKind, cfg: Option<&MessagingConfig>, msg: &str) {
dispatch_send_message(client, kind, cfg, msg, false).await;
}
+4
View File
@@ -117,6 +117,10 @@ impl Config {
devices: hdhr.devices.iter().map(|d| d.name.clone()).collect::<Vec<String>>(),
})
}
pub fn is_geoip_enabled(&self) -> bool {
self.reverse_proxy.as_ref().is_some_and(|r| r.geoip.as_ref().is_some_and(|g| g.enabled))
}
}
macros::from_impl!(Config);
+28
View File
@@ -0,0 +1,28 @@
use shared::model::GeoIpConfigDto;
use crate::model::macros;
#[derive(Debug, Clone)]
pub struct GeoIpConfig {
pub(crate) enabled: bool,
pub(crate) url: String,
}
macros::from_impl!(GeoIpConfig);
impl From<&GeoIpConfigDto> for GeoIpConfig {
fn from(dto: &GeoIpConfigDto) -> Self {
Self {
enabled: dto.enabled,
url: dto.url.clone(),
}
}
}
impl From<&GeoIpConfig> for GeoIpConfigDto {
fn from(instance: &GeoIpConfig) -> Self {
Self {
enabled: instance.enabled,
url: instance.url.clone(),
}
}
}
+2
View File
@@ -28,6 +28,7 @@ mod api_user;
mod trakt_api;
pub mod macros;
mod favourites;
mod geoip;
pub use api::*;
pub use api_proxy::*;
@@ -46,6 +47,7 @@ pub use proxy::*;
pub use rate_limit::*;
pub use rename::*;
pub use reverse_proxy::*;
pub use geoip::*;
pub use schedule::*;
pub use sort::*;
pub use source::*;
+4 -1
View File
@@ -1,6 +1,6 @@
use shared::model::ReverseProxyConfigDto;
use crate::model::config::cache::CacheConfig;
use crate::model::{macros, RateLimitConfig, StreamConfig};
use crate::model::{macros, GeoIpConfig, RateLimitConfig, StreamConfig};
#[derive(Debug, Clone)]
pub struct ReverseProxyConfig {
@@ -9,6 +9,7 @@ pub struct ReverseProxyConfig {
pub stream: Option<StreamConfig>,
pub cache: Option<CacheConfig>,
pub rate_limit: Option<RateLimitConfig>,
pub geoip: Option<GeoIpConfig>,
}
macros::from_impl!(ReverseProxyConfig);
@@ -21,6 +22,7 @@ impl From<&ReverseProxyConfigDto> for ReverseProxyConfig {
stream: dto.stream.as_ref().map(Into::into),
cache: dto.cache.as_ref().map(Into::into),
rate_limit: dto.rate_limit.as_ref().map(Into::into),
geoip: dto.geoip.as_ref().map(Into::into),
}
}
}
@@ -33,6 +35,7 @@ impl From<&ReverseProxyConfig> for ReverseProxyConfigDto {
stream: instance.stream.as_ref().map(Into::into),
cache: instance.cache.as_ref().map(Into::into),
rate_limit: instance.rate_limit.as_ref().map(Into::into),
geoip: instance.geoip.as_ref().map(Into::into),
}
}
}
+4 -4
View File
@@ -8,7 +8,7 @@ use crate::model::Config;
use crate::utils;
use crate::utils::{bincode_deserialize, bincode_serialize};
pub fn process_group_watch(client: &Arc<reqwest::Client>, cfg: &Config, target_name: &str, pl: &PlaylistGroup) {
pub async fn process_group_watch(client: &Arc<reqwest::Client>, cfg: &Config, target_name: &str, pl: &PlaylistGroup) {
let mut new_tree = BTreeSet::new();
pl.channels.iter().for_each(|chan| {
let header = &chan.header;
@@ -28,7 +28,7 @@ pub fn process_group_watch(client: &Arc<reqwest::Client>, cfg: &Config, target_n
let removed_difference: BTreeSet<String> = loaded_tree.difference(&new_tree).cloned().collect();
if !added_difference.is_empty() || !removed_difference.is_empty() {
changed = true;
handle_watch_notification(client, cfg, &added_difference, &removed_difference, target_name, &pl.title);
handle_watch_notification(client, cfg, &added_difference, &removed_difference, target_name, &pl.title).await;
}
} else {
error!("failed to load watch_file {}", &path.to_str().unwrap_or_default());
@@ -60,7 +60,7 @@ struct WatchChanges {
pub removed: Vec<String>,
}
fn handle_watch_notification(client: &Arc<reqwest::Client>, cfg: &Config, added: &BTreeSet<String>, removed: &BTreeSet<String>, target_name: &str, group_name: &str) {
async fn handle_watch_notification(client: &Arc<reqwest::Client>, cfg: &Config, added: &BTreeSet<String>, removed: &BTreeSet<String>, target_name: &str, group_name: &str) {
let added = added.iter().map(std::string::ToString::to_string).collect::<Vec<String>>();
let removed = removed.iter().map(std::string::ToString::to_string).collect::<Vec<String>>();
if !added.is_empty() || !removed.is_empty() {
@@ -73,7 +73,7 @@ fn handle_watch_notification(client: &Arc<reqwest::Client>, cfg: &Config, added:
let msg = serde_json::to_string_pretty(&changes).unwrap_or_else(|_| "Error: Failed to serialize watch changes".to_string());
info!("{}", &msg);
send_message(client, MsgKind::Watch, cfg.messaging.as_ref(), &msg);
send_message(client, MsgKind::Watch, cfg.messaging.as_ref(), &msg).await;
}
}
+5 -5
View File
@@ -579,7 +579,7 @@ async fn process_playlist_for_target(app_config: &AppConfig,
step.tick("assigning channel counter");
let config = app_config.config.load();
if process_watch(&config, &client, target, &flat_new_playlist) {
if process_watch(&config, &client, target, &flat_new_playlist).await {
step.tick("group watches");
}
let result = persist_playlist(app_config, &mut flat_new_playlist, flatten_tvguide(&new_epg).as_ref(), target, playlist_state).await;
@@ -620,14 +620,14 @@ async fn process_epg(processed_fetched_playlists: &mut Vec<FetchedPlaylist<'_>>)
(new_epg, new_playlist)
}
fn process_watch(cfg: &Config, client: &Arc<reqwest::Client>, target: &ConfigTarget, new_playlist: &Vec<PlaylistGroup>) -> bool {
async fn process_watch(cfg: &Config, client: &Arc<reqwest::Client>, target: &ConfigTarget, new_playlist: &Vec<PlaylistGroup>) -> bool {
if let Some(watches) = &target.watch {
if default_as_default().eq_ignore_ascii_case(&target.name) {
error!("cant watch a target with no unique name");
} else {
for pl in new_playlist {
if watches.iter().any(|r| r.is_match(&pl.title)) {
process_group_watch(client, cfg, &target.name, pl);
process_group_watch(client, cfg, &target.name, pl).await;
}
}
}
@@ -657,7 +657,7 @@ pub async fn exec_processing(client: Arc<reqwest::Client>, app_config: Arc<AppCo
// print stats
info!("{stats_msg}");
// send stats
send_message_json(&client, MsgKind::Stats, messaging, stats_msg.as_str());
send_message_json(&client, MsgKind::Stats, messaging, stats_msg.as_str()).await;
}
Err(err) => error!("Failed to serialize playlist stats {err}"),
}
@@ -672,7 +672,7 @@ pub async fn exec_processing(client: Arc<reqwest::Client>, app_config: Arc<AppCo
events.send_event(EventMessage::PlaylistUpdate(PlaylistUpdateState::Failure));
}
if let Ok(error_msg) = serde_json::to_string(&serde_json::Value::Object(serde_json::map::Map::from_iter([("errors".to_string(), serde_json::Value::String(message))]))) {
send_message_json(&client, MsgKind::Error, messaging, error_msg.as_str());
send_message_json(&client, MsgKind::Error, messaging, error_msg.as_str()).await;
}
} else if let Some(events) = event_manager {
events.send_event(EventMessage::PlaylistUpdate(PlaylistUpdateState::Success));
+97
View File
@@ -58,6 +58,49 @@ where
}
fn query_tree_le<K, V, R: Read + Seek>(file: &mut R, key: &K) -> Option<V>
where
K: Ord + Serialize + for<'de> Deserialize<'de> + Clone,
V: Serialize + for<'de> Deserialize<'de> + Clone,
{
let mut offset = 0;
let mut buffer = vec![0u8; BLOCK_SIZE];
loop {
match BPlusTreeNode::<K, V>::deserialize_from_block(file, &mut buffer, offset, false) {
Ok((node, pointers)) => {
if node.is_leaf {
let idx = get_entry_index_upper_bound::<K>(&node.keys, key);
if idx == 0 {
return None;
}
return node.values.get(idx - 1).cloned();
}
let child_idx = get_entry_index_upper_bound::<K>(&node.keys, key);
if let Some(child_offsets) = pointers {
if let Some(child_offset) = child_offsets.get(child_idx) {
offset = *child_offset;
} else {
// defensive: if out of bounds try last pointer
if let Some(last) = child_offsets.last() {
offset = *last;
} else {
return None;
}
}
} else {
return None;
}
}
Err(err) => {
error!("Failed to read id tree from file {err}");
return None;
}
}
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
struct BPlusTreeNode<K, V> {
keys: Vec<K>,
@@ -193,6 +236,32 @@ where
}
}
/// Find the largest key <= `key` in this subtree.
/// Returns a reference to (key, value) if found (only valid for leaf entries).
fn find_le(&self, key: &K) -> Option<(&K, &V)> {
if self.is_leaf {
// find index of first key > key, then step one back
let idx = self.get_entry_index_upper_bound(key);
if idx == 0 {
None
} else {
let i = idx - 1;
// safe: leaf guarantees values.len() == keys.len()
Some((&self.keys[i], &self.values[i]))
}
} else {
// descend into the appropriate child (child index = upper_bound)
let child_idx = self.get_entry_index_upper_bound(key);
// child_idx can be equal to children.len() if key > all keys; children.get handles that
if let Some(child) = self.children.get(child_idx) {
child.find_le(key)
} else {
// fallback: if child_idx is out of bounds, try last child (defensive)
self.children.last().and_then(|c| c.find_le(key))
}
}
}
pub fn traverse<F>(&self, visit: &mut F)
where
F: FnMut(&Vec<K>, &Vec<V>),
@@ -506,6 +575,15 @@ where
Ok(Self::new_with_root(root))
}
/// Find the largest key <= `key` in the in-memory tree and return references to (key, value).
pub fn find_le(&self, key: &K) -> Option<(&K, &V)> {
// empty tree
if self.root.keys.is_empty() && self.root.is_leaf && self.root.values.is_empty() {
return None;
}
self.root.find_le(key)
}
pub fn traverse<F>(&self, mut visit: F)
where
F: FnMut(&Vec<K>, &Vec<V>),
@@ -604,6 +682,19 @@ where
query_tree(&mut self.file, key)
}
/// On-disk: find largest key <= `key` and return owned V (cloned/deserialized)
pub fn query_le(&mut self, key: &K) -> Option<V> {
// use the same buffer/reader pattern as query()
// we need a mutable reader over the inner BufReader<File>
let file = &mut self.file;
// Seek to start to be safe
if file.seek(SeekFrom::Start(0)).is_err() {
// if seek fails, still try to query — but bail out with None
return None;
}
query_tree_le(file, key)
}
// pub fn traverse<F>(&mut self, mut visit: F)
// where
// F: FnMut(&Vec<K>, &Vec<V>),
@@ -681,6 +772,12 @@ where
}
}
}
/// On-disk update helper: find largest key <= `key`.
pub fn query_le(&mut self, key: &K) -> Option<V> {
let mut reader = utils::file_reader(&mut self.file);
query_tree_le(&mut reader, key)
}
}
pub struct BPlusTreeIterator<'a, K, V> {
+4
View File
@@ -32,3 +32,7 @@ pub fn get_input_storage_path(input_name: &str, working_dir: &str) -> std::io::R
// Create the directory and return the path or propagate the error
std::fs::create_dir_all(&path).map(|()| path)
}
pub fn get_geoip_path(working_dir: &str) -> PathBuf {
Path::new(working_dir).join("geoip.db")
}
@@ -440,6 +440,8 @@ pub async fn xtream_get_item_for_stream_id(
PlaylistItemType::Series => {
if let Ok(mut item) = xtream_read_series_item_for_stream_id(app_config, mapping.parent_virtual_id, &storage_path) {
item.provider_id = mapping.provider_id;
Ok(item)
} else {
xtream_read_item_for_stream_id(app_config, virtual_id, &storage_path, XtreamCluster::Series)
+4 -4
View File
@@ -82,14 +82,14 @@ pub fn read_sources_file(sources_file: &str, resolve_env: bool, include_computed
}
}
pub fn read_config_file(config_file: &str, resolve_env: bool) -> Result<ConfigDto, TuliproxError> {
pub fn read_config_file(config_file: &str, resolve_env: bool, include_computed: 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) => {
if resolve_env {
config.prepare()?;
config.prepare(include_computed)?;
}
Ok(config)
}
@@ -107,7 +107,7 @@ pub fn read_app_config_dto(paths: &ConfigPaths,
let sources_file = paths.sources_file_path.as_str();
let api_proxy_file = paths.api_proxy_file_path.as_str();
let config = read_config_file(config_file, resolve_env)?;
let config = read_config_file(config_file, resolve_env, include_computed)?;
let sources = read_sources_file(sources_file, resolve_env, include_computed, config.get_hdhr_device_overview().as_ref())?;
let mappings = if let Some(mappings_file) = paths.mapping_file_path.as_ref() {
read_mappings_file(mappings_file, resolve_env).unwrap_or(None)
@@ -211,7 +211,7 @@ pub fn read_initial_app_config(paths: &mut ConfigPaths,
let config_file = paths.config_file_path.as_str();
let sources_file = paths.sources_file_path.as_str();
let config_dto = read_config_file(config_file, resolve_env)?;
let config_dto = read_config_file(config_file, resolve_env, include_computed)?;
let mut sources_dto = read_sources_file(sources_file, resolve_env, include_computed, config_dto.get_hdhr_device_overview().as_ref())?;
prepare_sources_batch(&mut sources_dto, include_computed)?;
let sources: SourcesConfig = SourcesConfig::try_from(sources_dto)?;
+91
View File
@@ -0,0 +1,91 @@
use std::io;
use std::io::{BufRead};
use std::net::Ipv4Addr;
use std::path::Path;
use serde::{Serialize, Deserialize};
use crate::repository::bplustree::BPlusTree;
fn ipv4_to_u32(ip: &str) -> Option<u32> {
ip.parse::<Ipv4Addr>().ok().map(u32::from)
}
#[derive(Serialize, Deserialize)]
pub struct GeoIp {
tree: BPlusTree<u32, (u32, String)>,
}
impl GeoIp {
pub fn load(path: &Path) -> io::Result<Self> {
let tree = BPlusTree::load(path)?;
Ok(Self { tree })
}
pub fn new() -> Self {
Self { tree: BPlusTree::new() }
}
pub fn import_ipv4_from_csv(&mut self, mut reader: impl BufRead, db_path: &Path) -> std::io::Result<u64> {
let mut buf = String::new();
while reader.read_line(&mut buf)? > 0 {
let line = buf.trim();
if line.is_empty() || line.starts_with('#') { continue; }
let parts: Vec<&str> = line.split(',').collect();
if parts.len() != 3 { continue; }
if let (Some(start), Some(end)) = (ipv4_to_u32(parts[0]), ipv4_to_u32(parts[1])) {
let cc = parts[2].trim().to_string();
self.tree.insert(start, (end, cc));
}
buf.clear();
}
self.tree.store(db_path)
}
pub fn lookup(&self, ip_str: &str) -> Option<String> {
let ip = ipv4_to_u32(ip_str)?;
if let Some((_, (end, cc))) = self.tree.find_le(&ip) {
if ip <= *end {
return Some(cc.to_string());
}
}
None
}
}
impl Default for GeoIp {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod test {
// https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/asn-country/asn-country-ipv4.csv
use std::fs::File;
use std::io::BufReader;
use std::path::PathBuf;
use crate::utils::geoip::GeoIp;
#[test]
pub fn test_csv() {
let db_file = PathBuf::from("/projects/m3u-test/asn-country-ipv4.db");
let source = PathBuf::from("/projects/m3u-test/asn-country-ipv4.csv");
let file = File::open(source).expect("Could not open csv file");
let reader = BufReader::new(file);
let mut geo_ip = GeoIp::new();
let _ = geo_ip.import_ipv4_from_csv(reader, &db_file).expect("Could not import csv");
let geo_ip = GeoIp::load(&db_file).expect("Failed to load geoip db");
if let Some(cc) = geo_ip.lookup("72.13.24.23") {
assert_eq!(cc, "US");
} else {
assert!(false);
}
}
}
+2
View File
@@ -9,11 +9,13 @@ mod trakt;
mod json_utils;
mod bincode_utils;
mod telegram;
mod geoip;
pub use self::bincode_utils::*;
pub use self::logging::*;
pub use self::trakt::*;
pub use self::telegram::*;
pub use self::geoip::*;
#[macro_export]
macro_rules! debug_if_enabled {
+3 -3
View File
@@ -157,7 +157,7 @@ async fn xtream_login(cfg: &Config, client: &Arc<reqwest::Client>, input: &Input
if let Ok(cur_status) = ProxyUserStatus::from_str(&status) {
if !matches!(cur_status, ProxyUserStatus::Active | ProxyUserStatus::Trial) {
warn!("User status for user {username} is {cur_status:?}");
send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User status for user {username} is {cur_status:?}"));
send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User status for user {username} is {cur_status:?}")).await;
}
}
}
@@ -175,11 +175,11 @@ async fn xtream_login(cfg: &Config, client: &Arc<reqwest::Client>, input: &Input
let datetime = DateTime::from_timestamp(expiration_timestamp, 0).unwrap();
let formatted = datetime.format("%Y-%m-%d %H:%M:%S").to_string();
warn!("User account for user {username} expires {formatted}");
send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} expires {formatted}"));
send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} expires {formatted}")).await;
}
} else {
warn!("User account for user {username} is expired");
send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} is expired"));
send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} is expired")).await;
}
}
}
+19 -21
View File
@@ -62,7 +62,7 @@ pub fn telegram_create_instance(bot_token: &str, chat_id: &str) -> BotInstance {
}
}
pub fn telegram_send_message(
pub async fn telegram_send_message(
client: &Arc<reqwest::Client>,
instance: &BotInstance,
msg: &str,
@@ -87,27 +87,25 @@ pub fn telegram_send_message(
.map(ToString::to_string),
};
let the_client = Arc::clone(client);
tokio::spawn(async move {
let result = the_client
.post(url)
.json(&request_json_obj)
.send()
.await;
let result = client
.post(url)
.json(&request_json_obj)
.send()
.await;
match result {
Ok(response) => {
if response.status().is_success() {
debug!("Message sent successfully to {chat_id} telegram api");
} else {
match response.json::<TelegramErrorResult>().await {
Ok(json) => error!("Message wasn't sent to {chat_id} telegram api because of: {}", json.description),
Err(_) => error!("Message wasn't sent to {chat_id} telegram api. Telegram response could not be parsed!"),
}
match result {
Ok(response) => {
if response.status().is_success() {
debug!("Message sent successfully to {chat_id} telegram api");
} else {
match response.json::<TelegramErrorResult>().await {
Ok(json) => error!("Message wasn't sent to {chat_id} telegram api because of: {}", json.description),
Err(_) => error!("Message wasn't sent to {chat_id} telegram api. Telegram response could not be parsed!"),
}
},
Err(e) => error!("Message wasn't sent to {chat_id} telegram api because of: {e}"),
}
});
}
},
Err(e) => error!("Message wasn't sent to {chat_id} telegram api because of: {e}"),
}
}
+1 -1
View File
@@ -28,7 +28,7 @@ futures = "0.3"
prost = "0"
wasm-bindgen-futures = "0"
bytes = "1"
regex = "1.12.1"
regex = "1.12.2"
base64 = "0.22.1"
cron = "0.15"
fastrand = "2.3.0"
+12 -1
View File
@@ -291,9 +291,16 @@
"CLIENT_IP": "Client IP",
"STREAM_ID": "Stream Id",
"SHARED": "Shared",
"CLUSTER": "Type",
"USER_AGENT": "Player",
"FORMAT": "Format",
"DURATION": "Duration",
"ADD_QUALITY_TO_FILENAME": "Quality",
"SSDP_DISCOVERY": "SSDP Discovery",
"PROPRIETARY_DISCOVERY": "Proprietary Discovery"
"PROPRIETARY_DISCOVERY": "Proprietary Discovery",
"COUNTRY": "Country",
"GEOIP": "Geo-IP",
"UPDATE_GEOIP_DB": "Update Geo-IP db"
},
"TITLE": {
"USER_BOUQUET_EDITOR": "User group editor"
@@ -317,6 +324,10 @@
},
"SERVER_CONFIG": {
"FAIL": "Failed to download server config!"
},
"GEOIP": {
"SUCCESS": "Successfully downloaded Geo-IP db",
"FAIL": "Failed to download Geo-IP db!"
}
},
"LOGIN": {
@@ -18,4 +18,8 @@
gap: var(--gap-larger);
overflow: auto;
}
}
.tp__stream-table__duration {
letter-spacing: 2px;
}
@@ -28,6 +28,8 @@ const LABEL_CONFIG: &str = "LABEL.CONFIG";
const LABEL_EDIT: &str = "LABEL.EDIT";
const LABEL_VIEW: &str = "LABEL.VIEW";
const LABEL_SAVE: &str = "LABEL.SAVE";
const LABEL_UPDATE_GEOIP: &str = "LABEL.UPDATE_GEOIP_DB";
// ==========================
macro_rules! collect_modified {
@@ -162,7 +164,7 @@ pub fn ConfigView() -> Html {
let mut config_dto = config_ctx.config.as_ref().map_or_else(ConfigDto::default,
|app_cfg| app_cfg.config.clone());
update_config(&mut config_dto, modified_forms);
match config_dto.prepare() {
match config_dto.prepare(false) {
Ok(_) => {
let services = services.clone();
let translate = translate.clone();
@@ -214,6 +216,22 @@ pub fn ConfigView() -> Html {
};
let handle_update_geoip = {
let services = services_ctx.clone();
let translate = translate.clone();
Callback::from(move |_| {
let services = services.clone();
let translate = translate.clone();
wasm_bindgen_futures::spawn_local(async move {
match services.config.update_geoip().await {
Ok(_) => services.toastr.success(translate.t("MESSAGES.DOWNLOAD.GEOIP.SUCCESS")),
Err(_err) => services.toastr.error(translate.t("MESSAGES.DOWNLOAD.GEOIP.FAIL")),
}
});
})
};
let context = ConfigViewContext {
edit_mode: edit_mode.clone(),
on_form_change: on_form_change.clone(),
@@ -224,11 +242,18 @@ pub fn ConfigView() -> Html {
<div class="tp__config-view">
<div class="tp__config-view__header">
<h1>{ translate.t(LABEL_CONFIG) } </h1>
<TextButton name="config_edit"
{html_if!(config_ctx.config.is_some_and(|c| c.config.is_geoip_enabled()), {
<TextButton class="tertiary" name="save_config"
icon="Refresh"
title={ translate.t(LABEL_UPDATE_GEOIP)}
onclick={handle_update_geoip}></TextButton>
})}
<TextButton name="config_edit"
class={ if *edit_mode { "secondary" } else { "primary" }}
icon={ if *edit_mode { "Unlocked" } else { "Locked" }}
title={ if *edit_mode { translate.t(LABEL_EDIT) } else { translate.t(LABEL_VIEW) }}
onclick={handle_config_edit}></TextButton>
</div>
<div class="tp__config-view__body">
<Card>
@@ -86,7 +86,7 @@ pub fn HdHomerunConfigView() -> Html {
.unwrap_or(8901) + 1;
new_device.port = next_port;
new_device.name = format!("hdhr_{next_port}");
if let Err(err) = new_device.prepare(devices.len() as u8) {
if let Err(err) = new_device.prepare(devices.len() as u8, false) {
error!("Failed to prepare hdhr device: {err}");
}
devices.push(new_device);
@@ -46,7 +46,7 @@ pub fn HdHomerunDeviceView(props: &HdHomerunDeviceViewProps) -> Html {
let device_id = props.device_id;
let deps = (form_state.clone(), form_state.modified);
use_effect_with(deps, move |(state, modified)| {
on_form_change.emit((device_id, *modified, *state.form.clone()));
on_form_change.emit((device_id, *modified, (*state.form).clone()));
});
}
@@ -1,15 +1,12 @@
use yew::prelude::*;
use yew_i18n::use_translation;
use shared::model::{
CacheConfigDto, RateLimitConfigDto, StreamConfigDto, ReverseProxyConfigDto,
};
use shared::model::{CacheConfigDto, RateLimitConfigDto, StreamConfigDto, ReverseProxyConfigDto, GeoIpConfigDto};
use crate::app::context::ConfigContext;
use crate::app::components::config::config_view_context::ConfigViewContext;
use crate::app::components::config::config_page::ConfigForm;
use crate::app::components::config::macros::HasFormData;
use crate::app::components::Card;
use crate::{config_field, config_field_bool, config_field_optional,
edit_field_bool, edit_field_number, edit_field_number_u64, edit_field_text_option, generate_form_reducer};
use crate::app::components::{Card};
use crate::{config_field, config_field_bool, config_field_optional, edit_field_bool, edit_field_number, edit_field_number_u64, edit_field_text, edit_field_text_option, generate_form_reducer};
const LABEL_CACHE: &str = "LABEL.CACHE";
const LABEL_ENABLED: &str = "LABEL.ENABLED";
@@ -30,6 +27,8 @@ const LABEL_BURST_SIZE: &str = "LABEL.BURST_SIZE";
const LABEL_RESOURCE_REWRITE_DISABLED: &str = "LABEL.RESOURCE_REWRITE_DISABLED";
const LABEL_DISABLE_REFERER_HEADER: &str = "LABEL.DISABLE_REFERER_HEADER";
const LABEL_GEOIP: &str = "LABEL.GEOIP";
const LABEL_URL: &str = "LABEL.URL";
generate_form_reducer!(
state: CacheConfigFormState { form: CacheConfigDto },
@@ -64,6 +63,15 @@ generate_form_reducer!(
}
);
generate_form_reducer!(
state: GeoIpConfigFormState { form: GeoIpConfigDto },
action_name: GeoIpConfigFormAction,
fields {
Enabled => enabled: bool,
Url => url: String,
}
);
generate_form_reducer!(
state: ReverseProxyConfigFormState { form: ReverseProxyConfigDto },
action_name: ReverseProxyConfigFormAction,
@@ -92,22 +100,28 @@ pub fn ReverseProxyConfigView() -> Html {
StreamConfigFormState { form: StreamConfigDto::default(), modified: false }
});
let geoip_state: UseReducerHandle<GeoIpConfigFormState> = use_reducer(|| {
GeoIpConfigFormState { form: GeoIpConfigDto::default(), modified: false }
});
{
let on_form_change = config_view_ctx.on_form_change.clone();
let reverse_proxy_state = reverse_proxy_state.clone();
let cache_state = cache_state.clone();
let rate_limit_state = rate_limit_state.clone();
let stream_state = stream_state.clone();
let geoip_state = geoip_state.clone();
use_effect_with(
(reverse_proxy_state, cache_state, rate_limit_state, stream_state),
move |(rp, cache, rl, stream)| {
(reverse_proxy_state, cache_state, rate_limit_state, stream_state, geoip_state),
move |(rp, cache, rl, stream, geoip)| {
let mut form = rp.form.clone();
form.cache = Some(cache.form.clone());
form.rate_limit = Some(rl.form.clone());
form.stream = Some(stream.form.clone());
form.geoip = Some(geoip.form.clone());
let modified = rp.modified || cache.modified || rl.modified || stream.modified;
let modified = rp.modified || cache.modified || rl.modified || stream.modified || geoip.modified;
on_form_change.emit(ConfigForm::ReverseProxy(modified, form));
},
);
@@ -118,6 +132,7 @@ pub fn ReverseProxyConfigView() -> Html {
let cache_state = cache_state.clone();
let rate_limit_state = rate_limit_state.clone();
let stream_state = stream_state.clone();
let geoip_state = geoip_state.clone();
let reverse_proxy_cfg = config_ctx.config.as_ref().and_then(|c| c.config.reverse_proxy.clone());
use_effect_with((reverse_proxy_cfg, config_view_ctx.edit_mode.clone()), move |(cfg, _mode)| {
@@ -126,11 +141,13 @@ pub fn ReverseProxyConfigView() -> Html {
cache_state.dispatch(CacheConfigFormAction::SetAll(rp.cache.as_ref().map_or_else(CacheConfigDto::default, |c| c.clone())));
rate_limit_state.dispatch(RateLimitConfigFormAction::SetAll(rp.rate_limit.as_ref().map_or_else(RateLimitConfigDto::default, |rl| rl.clone())));
stream_state.dispatch(StreamConfigFormAction::SetAll(rp.stream.as_ref().map_or_else(StreamConfigDto::default, |s| s.clone())));
geoip_state.dispatch(GeoIpConfigFormAction::SetAll(rp.geoip.as_ref().map_or_else(GeoIpConfigDto::default, |s| s.clone())));
} else {
reverse_proxy_state.dispatch(ReverseProxyConfigFormAction::SetAll(ReverseProxyConfigDto::default()));
cache_state.dispatch(CacheConfigFormAction::SetAll(CacheConfigDto::default()));
rate_limit_state.dispatch(RateLimitConfigFormAction::SetAll(RateLimitConfigDto::default()));
stream_state.dispatch(StreamConfigFormAction::SetAll(StreamConfigDto::default()));
geoip_state.dispatch(GeoIpConfigFormAction::SetAll(GeoIpConfigDto::default()));
}
|| ()
});
@@ -171,6 +188,16 @@ pub fn ReverseProxyConfigView() -> Html {
}
};
let render_geoip = || {
html! {
<Card class="tp__config-view__card">
<h1>{translate.t(LABEL_GEOIP)}</h1>
{ config_field_bool!(geoip_state.form, translate.t(LABEL_ENABLED), enabled) }
{ config_field!(geoip_state.form, translate.t(LABEL_URL), url) }
</Card>
}
};
let render_view_mode = || {
html! {
<>
@@ -182,6 +209,7 @@ pub fn ReverseProxyConfigView() -> Html {
{ render_cache() }
{ render_rate_limit() }
{ render_stream() }
{ render_geoip() }
</div>
</>
}
@@ -215,6 +243,11 @@ pub fn ReverseProxyConfigView() -> Html {
{ edit_field_number!(stream_state, translate.t(LABEL_FORCED_RETRY_INTERVAL_SECS), forced_retry_interval_secs, StreamConfigFormAction::ForcedRetryIntervalSecs) }
{ edit_field_number_u64!(stream_state, translate.t(LABEL_THROTTLE_KBPS), throttle_kbps, StreamConfigFormAction::ThrottleKbps) }
</Card>
<Card class="tp__config-view__card">
<h1>{translate.t(LABEL_GEOIP)}</h1>
{ edit_field_bool!(geoip_state, translate.t(LABEL_ENABLED), enabled, GeoIpConfigFormAction::Enabled) }
{ edit_field_text!(geoip_state, translate.t(LABEL_URL), url, GeoIpConfigFormAction::Url) }
</Card>
</div>
</>
};
@@ -7,20 +7,52 @@ use shared::model::{SortOrder, StreamInfo};
use std::fmt::Display;
use std::rc::Rc;
use std::str::FromStr;
use gloo_timers::callback::Interval;
use gloo_utils::window;
use wasm_bindgen::JsCast;
use web_sys::Element;
use yew::prelude::*;
use yew_i18n::use_translation;
use shared::utils::{current_time_secs, strip_port};
const HEADERS: [&str; 8] = [
const HEADERS: [&str; 12] = [
"LABEL.EMPTY",
"LABEL.USERNAME",
"LABEL.STREAM_ID",
"LABEL.CLUSTER",
"LABEL.CHANNEL",
"LABEL.GROUP",
"LABEL.CLIENT_IP",
"LABEL.COUNTRY",
"LABEL.PROVIDER",
"LABEL.SHARED"
"LABEL.SHARED",
"LABEL.USER_AGENT",
"LABEL.DURATION"
];
pub fn format_duration(seconds: u64) -> String {
let hours = seconds / 3600;
let minutes = (seconds % 3600) / 60;
let seconds = seconds % 60;
format!("{hours:02}:{minutes:02}:{seconds:02}")
}
fn update_timestamps() {
let window = window();
let document = window.document().unwrap();
let spans = document.query_selector_all("span[data-ts]").unwrap();
for i in 0..spans.length() {
if let Some(node) = spans.item(i) {
let el: Element = node.dyn_into().unwrap();
if let Some(ts_str) = el.get_attribute("data-ts") {
if let Ok(ts) = ts_str.parse::<u64>() {
el.set_inner_html(&format_duration(current_time_secs() - ts));
}
}
}
}
}
#[derive(Properties, PartialEq, Clone)]
pub struct StreamsTableProps {
pub streams: Option<Vec<Rc<StreamInfo>>>,
@@ -34,6 +66,14 @@ pub fn StreamsTable(props: &StreamsTableProps) -> Html {
let popup_is_open = use_state(|| false);
let selected_dto = use_state(|| None::<Rc<StreamInfo>>);
use_effect_with((), move |_| {
Interval::new(1000, || {
update_timestamps();
}).forget();
});
let handle_popup_close = {
let set_is_open = popup_is_open.clone();
Callback::from(move |()| {
@@ -91,11 +131,15 @@ pub fn StreamsTable(props: &StreamsTableProps) -> Html {
{ dto.channel.provider_id.to_string() }
{")"}
</>},
3 => html! {dto.channel.title.as_str()},
4 => html! {dto.channel.group.as_str()},
5 => html! {dto.addr.as_str()},
6 => html! {dto.provider.as_str()},
7 => html! { <ToggleSwitch value={dto.channel.shared} readonly={true} /> },
3 => html! {dto.channel.cluster},
4 => html! {dto.channel.title.as_str()},
5 => html! {dto.channel.group.as_str()},
6 => html! { strip_port(&dto.addr)},
7 => html! { dto.country.as_ref().map_or_else(String::new, |c| c.clone()) },
8 => html! {dto.provider.as_str()},
9 => html! { <ToggleSwitch value={dto.channel.shared} readonly={true} /> },
10 => html! { dto.user_agent.as_str() },
11 => html! { <span class="tp__stream-table__duration" data-ts={dto.ts.to_string()}>{format_duration(dto.ts)}</span> },
_ => html! {""},
}
})
+6
View File
@@ -21,6 +21,7 @@ pub struct ConfigService {
config_path: String,
ip_check_path: String,
batch_input_content_path: String,
geoip_path: String,
event_service: Rc<EventService>
}
@@ -35,6 +36,7 @@ impl ConfigService {
config_path: concat_path_leading_slash(&base_href, "api/v1/config"),
ip_check_path: concat_path_leading_slash(&base_href, "api/v1/ipinfo"),
batch_input_content_path: concat_path_leading_slash(&base_href, "api/v1/config/batchContent"),
geoip_path: concat_path_leading_slash(&base_href, "api/v1/geoip/update"),
event_service
}
}
@@ -142,4 +144,8 @@ impl ConfigService {
}
}
pub async fn update_geoip(&self) -> Result<Option<()>, Error> {
request_get::<()>(&self.geoip_path, None, None).await
}
}
+3 -1
View File
@@ -21,4 +21,6 @@ fastrand = { version = "2", features = ["js"] }
zeroize = "1"
chrono = "0.4.42"
bytes = "1"
ciborium = "0.2.2"
ciborium = "0.2.2"
#[cfg(target_arch = "wasm32")]
js-sys = "0.3.81"
+9 -4
View File
@@ -148,7 +148,7 @@ pub struct HdHomeRunDeviceOverview {
}
impl ConfigDto {
pub fn prepare(&mut self) -> Result<(), TuliproxError> {
pub fn prepare(&mut self, include_computed: bool) -> Result<(), TuliproxError> {
if let Some(mins) = self.sleep_timer_mins {
if mins == 0 {
return Err(TuliproxError::new(TuliproxErrorKind::Info, "`sleep_timer_mins` must be > 0 when specified".to_string()));
@@ -157,7 +157,7 @@ impl ConfigDto {
self.api.prepare();
self.prepare_web()?;
self.prepare_hdhomerun()?;
self.prepare_hdhomerun(include_computed)?;
self.prepare_video_config()?;
if let Some(reverse_proxy) = self.reverse_proxy.as_mut() {
@@ -180,10 +180,10 @@ impl ConfigDto {
Ok(())
}
fn prepare_hdhomerun(&mut self) -> Result<(), TuliproxError> {
fn prepare_hdhomerun(&mut self, include_computed: bool) -> Result<(), TuliproxError> {
if let Some(hdhomerun) = &mut self.hdhomerun {
if hdhomerun.enabled {
hdhomerun.prepare(self.api.port)?;
hdhomerun.prepare(self.api.port, include_computed)?;
}
}
Ok(())
@@ -251,4 +251,9 @@ impl ConfigDto {
self.accept_insecure_ssl_certificates = main_config.accept_insecure_ssl_certificates;
}
pub fn is_geoip_enabled(&self) -> bool {
self.reverse_proxy.as_ref().is_some_and(|r| r.geoip.as_ref().is_some_and(|g| g.enabled))
}
}
+11
View File
@@ -0,0 +1,11 @@
fn default_geoip_url() -> String { String::from("https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/asn-country/asn-country-ipv4.csv") }
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct GeoIpConfigDto {
#[serde(default)]
pub enabled: bool,
#[serde(default = "default_geoip_url")]
pub url: String,
}
+22 -14
View File
@@ -1,8 +1,8 @@
use std::collections::HashSet;
use log::warn;
use crate::create_tuliprox_error_result;
use crate::error::{TuliproxError, TuliproxErrorKind};
use crate::utils::{default_as_true, generate_hdhr_device_id, generate_hdhr_device_id_from_base, validate_hdhr_device_id, hash_string, hex_encode};
use crate::utils::{default_as_true, generate_hdhr_device_id, generate_hdhr_device_id_from_base, hash_string, hex_encode, validate_hdhr_device_id};
use log::warn;
use std::collections::HashSet;
fn default_friendly_name() -> String { String::from("TuliproxTV") }
fn default_manufacturer() -> String { String::from("Silicondust") }
@@ -60,11 +60,13 @@ impl Default for HdHomeRunDeviceConfigDto {
}
impl HdHomeRunDeviceConfigDto {
pub fn prepare(&mut self, device_num: u8) -> Result<(), TuliproxError> {
pub fn prepare(&mut self, device_num: u8, include_computed: bool) -> Result<(), TuliproxError> {
self.name = self.name.trim().to_string();
if self.name.is_empty() {
self.name = format!("device{device_num}");
warn!("Device name empty, assigned new name: {}", self.name);
if include_computed {
warn!("Device name empty, assigned new name: {}", self.name);
}
}
if self.tuner_count == 0 {
@@ -86,7 +88,9 @@ impl HdHomeRunDeviceConfigDto {
let p4 = hex_encode(&hash[8..10]);
let p5 = hex_encode(&hash[10..16]);
self.device_udn = format!("{p1}-{p2}-{p3}-{p4}-{p5}");
warn!("HDHomeRun device '{}' is missing a unique device_udn. A new one has been generated: {}", self.name, self.device_udn);
if include_computed {
warn!("HDHomeRun device '{}' is missing a unique device_udn. A new one has been generated: {}", self.name, self.device_udn);
}
} else {
// Ensure only the UUID part is stored.
if let Some(uuid_part) = self.device_udn.strip_prefix("uuid:") {
@@ -97,11 +101,15 @@ impl HdHomeRunDeviceConfigDto {
// --- Device ID Logic ---
if self.device_id.is_empty() {
self.device_id = generate_hdhr_device_id();
warn!("HDHomeRun device '{}' is missing a device_id. A new one has been generated: {}", self.name, self.device_id);
if include_computed {
warn!("HDHomeRun device '{}' is missing a device_id. A new one has been generated: {}", self.name, self.device_id);
}
} else if !validate_hdhr_device_id(&self.device_id) {
let old_id = self.device_id.clone();
self.device_id = generate_hdhr_device_id_from_base(&self.device_id);
warn!("HDHomeRun device '{}' has an invalid device_id '{}'. A valid one has been generated: {}", self.name, old_id, self.device_id);
if include_computed {
warn!("HDHomeRun device '{}' has an invalid device_id '{}'. A valid one has been generated: {}", self.name, old_id, self.device_id);
}
}
Ok(())
}
@@ -132,13 +140,13 @@ impl HdHomeRunConfigDto {
// It's only included to satisfy the frontend compiler.
}
pub fn prepare(&mut self, api_port: u16) -> Result<(), TuliproxError> {
pub fn prepare(&mut self, api_port: u16, include_computed: bool) -> Result<(), TuliproxError> {
let mut names = HashSet::new();
let mut ports = HashSet::new();
let mut device_ids = HashSet::new();
ports.insert(api_port);
for (device_num, device) in (0_u8..).zip(self.devices.iter_mut()) {
device.prepare(device_num)?;
device.prepare(device_num, include_computed)?;
if !names.insert(device.name.clone()) {
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "HdHomeRun duplicate device name {}", device.name);
}
@@ -153,10 +161,10 @@ impl HdHomeRunConfigDto {
for device in &mut self.devices {
if device.port == 0 {
while ports.contains(&current_port) || current_port == 0 {
current_port = current_port.wrapping_add(1);
if current_port == api_port { // full cycle guard
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "No free port available for HdHomeRun devices");
}
current_port = current_port.wrapping_add(1);
if current_port == api_port { // full cycle guard
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "No free port available for HdHomeRun devices");
}
}
device.port = current_port;
+2
View File
@@ -33,6 +33,7 @@ mod proxy_type;
mod proxy_user_status;
mod favourites;
mod geoip;
pub use proxy_type::*;
pub use proxy_user_status::*;
@@ -58,6 +59,7 @@ pub use epg::*;
pub use cache::*;
pub use rate_limit::*;
pub use reverse_proxy::*;
pub use geoip::*;
pub use proxy::*;
pub use trakt::*;
pub use rename::*;
+4 -1
View File
@@ -1,5 +1,5 @@
use crate::error::TuliproxError;
use crate::model::{CacheConfigDto, RateLimitConfigDto, StreamConfigDto};
use crate::model::{CacheConfigDto, GeoIpConfigDto, RateLimitConfigDto, StreamConfigDto};
use log::warn;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)]
@@ -15,6 +15,9 @@ pub struct ReverseProxyConfigDto {
pub cache: Option<CacheConfigDto>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub rate_limit: Option<RateLimitConfigDto>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub geoip: Option<GeoIpConfigDto>,
}
impl ReverseProxyConfigDto {
+1
View File
@@ -3,6 +3,7 @@ use crate::model::WebAuthConfigDto;
use crate::utils::{default_as_true, is_blank_optional_string};
const RESERVED_PATHS: &[&str] = &[
"cvs",
"live",
"movie",
"series",
+10 -3
View File
@@ -1,5 +1,6 @@
use serde::{Deserialize, Serialize};
use crate::model::{M3uPlaylistItem, PlaylistEntry, PlaylistItemType, XtreamCluster, XtreamPlaylistItem};
use crate::utils::{current_time_secs, StringExt};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct StreamChannel {
@@ -20,7 +21,7 @@ impl XtreamPlaylistItem {
item_type: self.item_type,
cluster: self.xtream_cluster,
group: self.group.clone(),
title: self.title.clone(),
title: String::longest(self.title.as_str(), self.name.as_str()).to_string(),
url: self.url.clone(),
shared: false,
}
@@ -35,7 +36,7 @@ impl M3uPlaylistItem {
item_type: self.item_type,
cluster: XtreamCluster::try_from(self.item_type).unwrap_or(XtreamCluster::Live),
group: self.group.clone(),
title: self.title.clone(),
title: String::longest(self.title.as_str(), self.name.as_str()).to_string(),
url: self.url.clone(),
shared: false,
}
@@ -48,15 +49,21 @@ pub struct StreamInfo {
pub channel: StreamChannel,
pub provider: String,
pub addr: String,
pub user_agent: String,
pub ts: u64,
pub country: Option<String>,
}
impl StreamInfo {
pub fn new(username: &str, addr: &str, provider: &str, stream_channel: StreamChannel) -> Self {
pub fn new(username: &str, addr: &str, provider: &str, stream_channel: StreamChannel, user_agent: String, country: Option<String>) -> Self {
Self {
username: username.to_string(),
channel: stream_channel,
provider: provider.to_string(),
addr: addr.to_string(),
user_agent,
ts: current_time_secs(),
country,
}
}
}
+18 -16
View File
@@ -24,16 +24,20 @@ pub fn validate_hdhr_device_id(device_id: &str) -> bool {
}
pub fn generate_hdhr_device_id_from_base(base_id: &str) -> String {
let base_sanitized = base_id.chars().filter(|c| c.is_ascii_hexdigit()).collect::<String>();
let base_padded = if base_sanitized.is_empty() {
return generate_hdhr_device_id();
} else {
format!("{:0<7}", &base_sanitized[..base_sanitized.len().min(7)])
};
if let Ok(device_id_int_base) = u32::from_str_radix(&base_padded, 16) {
let checksum = calculate_checksum(device_id_int_base);
let final_id = (device_id_int_base & 0xFFFFFFF0) | u32::from(checksum);
let base_sanitized: String = base_id
.chars()
.filter(|c| c.is_ascii_hexdigit())
.collect::<String>()
.to_uppercase();
if base_sanitized.is_empty() {
return generate_hdhr_device_id();
}
// Keep at most 7 hex digits, pad-left with zeros to 7
let base7 = format!("{:0>7}", &base_sanitized[..base_sanitized.len().min(7)]);
if let Ok(base7_int) = u32::from_str_radix(&base7, 16) {
let base_shifted = base7_int << 4; // bits 4-31 for base, bits 0-3 for checksum
let checksum = calculate_checksum(base_shifted);
let final_id = base_shifted | u32::from(checksum);
format!("{:08X}", final_id)
} else {
generate_hdhr_device_id()
@@ -41,10 +45,8 @@ pub fn generate_hdhr_device_id_from_base(base_id: &str) -> String {
}
pub fn generate_hdhr_device_id() -> String {
let random_part: String = (0..4)
.map(|_| format!("{:X}", fastrand::u8(0..16)))
.collect();
let base_id = format!("105{}0", random_part);
generate_hdhr_device_id_from_base(&base_id)
// 3 fixed + 4 random = 7 hex digits base
let rnd = (0..4).map(|_| format!("{:X}", fastrand::u8(0..16))).collect::<String>();
let base7 = format!("105{rnd}");
generate_hdhr_device_id_from_base(&base7)
}
+4 -2
View File
@@ -8,7 +8,8 @@ mod directed_graph;
mod hash_utils;
mod json_utils;
mod serde_utils;
pub mod hdhomerun_utils;
mod hdhomerun_utils;
mod net_utils;
use std::fmt::Display;
pub use self::default_utils::*;
@@ -22,6 +23,7 @@ pub use self::hash_utils::*;
pub use self::json_utils::*;
pub use self::serde_utils::*;
pub use self::hdhomerun_utils::*;
pub use self::net_utils::*;
#[macro_export]
macro_rules! write_if_some {
@@ -42,4 +44,4 @@ pub fn display_vec<T: Display>(vec: &[T]) -> String {
.collect::<Vec<_>>()
.join(", ");
format!("[{inner}]")
}
}
+25
View File
@@ -0,0 +1,25 @@
use std::borrow::Cow;
pub fn strip_port<'a>(input: &'a str) -> Cow<'a, str> {
// IPv6 with port: [2001:db8::1]:8080
if let Some(stripped) = input.strip_prefix('[') {
if let Some(end) = stripped.find(']') {
return Cow::Owned(stripped[..end].to_string());
}
// Invalid IPv6
return Cow::Borrowed(input);
}
// IPv4 or IPv6 without bracket
if let Some((left, _right)) = input.rsplit_once(':') {
// If `left` has a colon then its IPv6 without port.
if left.contains(':') {
Cow::Borrowed(input)
} else {
// IPv4:Port
Cow::Owned(left.to_string())
}
} else {
Cow::Borrowed(input)
}
}
+10
View File
@@ -127,6 +127,16 @@ pub fn humanize_snake_case(s: &str) -> String {
result
}
pub trait StringExt {
fn longest<'a>(a: &'a str, b: &'a str) -> &'a str;
}
impl StringExt for String {
fn longest<'a>(a: &'a str, b: &'a str) -> &'a str {
if a.len() >= b.len() { a } else { b }
}
}
#[cfg(test)]
mod test {
use std::collections::HashSet;
+9 -6
View File
@@ -1,9 +1,12 @@
use std::time::{SystemTime, UNIX_EPOCH};
use chrono::{DateTime};
#[cfg(target_arch = "wasm32")]
pub fn current_time_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
(js_sys::Date::now() / 1000.0) as u64
}
#[cfg(not(target_arch = "wasm32"))]
pub fn current_time_secs() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
@@ -15,7 +18,7 @@ pub fn unix_ts_to_str(ts: i64) -> Option<String> {
} else {
ts
};
DateTime::from_timestamp(normalized_ts, 0).map(|dt| dt.format("%d.%m.%Y").to_string())
chrono::DateTime::from_timestamp(normalized_ts, 0).map(|dt| dt.format("%d.%m.%Y").to_string())
} else {
None
}