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

fix: honor explicit Flussonic HLS catchup-source with bounded archive… (#887)

Browse Source 
* **New Features**
  * Added support for bounded Flussonic HLS catch-up playback when an explicit source is configured, with valid start times and durations of up to seven days.
  * Bounded archive playlist resources now inherit missing query parameters from their source while preserving values provided by child resources. Cross-origin URLs remain unchanged.
* **Bug Fixes**
  * Rewritten Flussonic catch-up output now uses the rewritten mode without retaining the original catch-up type or provider token.
This commit is contained in:
euzu
2026-09-30 10:41:54 +02:00
committed by GitHub
parent 59e0f242ab
commit e703c8dd87
3 changed files with 373 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/parser/src/hls/mod.rs
+176 -6
View File
@@ -103,6 +103,24 @@ fn preserve_archive_start_query(base: &Url, mut target: Url) -> Url {
target
}
fn is_bounded_flussonic_hls_url(url: &Url) -> bool {
url.path_segments().and_then(Iterator::last).is_some_and(|file| {
matches!(crate::m3u_format::parse_flussonic_archive_file(file),
Some(crate::m3u_format::FlussonicArchiveKind::Archive { duration, extension: ".m3u8", .. })
if duration.parse::<u64>().is_ok_and(|duration| duration > 0))
})
}
fn preserve_bounded_archive_query(base: &Url, mut target: Url) -> Url {
// Child URIs can supply their own credentials; inherit only missing query keys.
for (key, value) in base.query_pairs() {
if !target.query_pairs().any(|(target_key, _)| target_key.eq_ignore_ascii_case(&key)) {
target.query_pairs_mut().append_pair(&key, &value);
}
}
target
}
fn preserve_token_query(base: &Url, mut target: Url) -> Url {
let target_has_token = target.query_pairs().any(|(key, _)| key.eq_ignore_ascii_case("token"));
if target_has_token {
@@ -123,9 +141,19 @@ fn has_same_origin(left: &Url, right: &Url) -> bool {
}
/// Rewrites an HLS URI relative to a base playlist URL.
/// Absolute URIs are returned unchanged.
/// Bounded Flussonic archives inherit source queries for resources on the same origin.
/// Other absolute URIs are returned unchanged.
pub fn rewrite_hls_url<'a>(base: &'a str, reference: &'a str) -> Cow<'a, str> {
if Url::parse(reference).is_ok() {
rewrite_hls_url_with_archive_context(base, reference, false)
}
fn rewrite_hls_url_with_archive_context<'a>(base: &'a str, reference: &'a str, bounded_archive: bool) -> Cow<'a, str> {
if let Ok(target) = Url::parse(reference) {
if let Ok(base_url) = Url::parse(base) {
if (bounded_archive || is_bounded_flussonic_hls_url(&base_url)) && has_same_origin(&base_url, &target) {
return Cow::Owned(preserve_bounded_archive_query(&base_url, target).into());
}
}
return Cow::Borrowed(reference);
}
@@ -138,7 +166,11 @@ pub fn rewrite_hls_url<'a>(base: &'a str, reference: &'a str) -> Cow<'a, str> {
|mut target| {
let is_same_origin = has_same_origin(&target, &base_url);
if is_same_origin {
target = preserve_token_query(&base_url, target);
target = if bounded_archive || is_bounded_flussonic_hls_url(&base_url) {
preserve_bounded_archive_query(&base_url, target)
} else {
preserve_token_query(&base_url, target)
};
}
let is_same_origin_child_playlist = is_same_origin
&& extract_extension_from_url(target.as_str())
@@ -152,13 +184,40 @@ pub fn rewrite_hls_url<'a>(base: &'a str, reference: &'a str) -> Cow<'a, str> {
)
}
fn bounded_archive_session_virtual_id(token: &str, username: &str) -> Option<u32> {
let payload = token.strip_prefix("m3u-catchup|").or_else(|| token.strip_prefix("catchup|"))?;
let (fingerprint, subject) = payload.split_once('|')?;
// Match the authenticated username before parsing fields; usernames may contain separators.
let archive = subject.strip_prefix(username)?.strip_prefix('|')?;
let mut fields = archive.split('|');
let virtual_id = fields.next()?.parse::<u32>().ok()?;
if fingerprint.is_empty() || username.is_empty() || fields.next()? != "archive" {
return None;
}
fields.next()?.parse::<u64>().ok()?;
if fields.next()?.parse::<u64>().ok()? == 0 {
return None;
}
match (fields.next(), fields.next()) {
(None | Some(".m3u8" | ".ts"), None) => Some(virtual_id),
_ => None,
}
}
fn rewrite_hls_resource_url<'a>(props: &'a RewriteHlsProps, reference: &'a str, username: &str) -> Cow<'a, str> {
// The range identity survives child playlists whose filenames no longer identify the archive.
let bounded_archive = props.user_token.and_then(|token| bounded_archive_session_virtual_id(token, username))
== Some(props.virtual_id);
rewrite_hls_url_with_archive_context(&props.hls_url, reference, bounded_archive)
}
fn rewrite_uri_attrib<'a>(line: &'a str, props: &RewriteHlsProps, user: &ProxyUserCredentials) -> Cow<'a, str> {
let Some(caps) = CONSTANTS.re_hls_uri.captures(line) else {
return Cow::Borrowed(line);
};
let uri = &caps[1];
let rewritten = rewrite_hls_url(&props.hls_url, uri);
let rewritten = rewrite_hls_resource_url(props, uri, &user.username);
let token = if let Some(user_token) = &props.user_token {
create_hls_session_token_and_url(props.secret, user_token, &rewritten)
@@ -191,7 +250,7 @@ pub fn rewrite_hls(user: &ProxyUserCredentials, props: &RewriteHlsProps) -> Stri
}
// target url
let target_url = rewrite_hls_url(&props.hls_url, line);
let target_url = rewrite_hls_resource_url(props, line, username);
let token = if let Some(user_token) = &props.user_token {
create_hls_session_token_and_url(props.secret, user_token, &target_url)
} else {
@@ -209,7 +268,10 @@ pub fn rewrite_hls(user: &ProxyUserCredentials, props: &RewriteHlsProps) -> Stri
#[cfg(test)]
mod test {
use super::{get_hls_session_token_and_url_from_token, rewrite_hls, rewrite_hls_url, RewriteHlsProps};
use super::{
get_hls_session_token_and_url_from_token, rewrite_hls, rewrite_hls_resource_url, rewrite_hls_url,
RewriteHlsProps,
};
use rand::RngCore;
use shared::{defaults::HLS_PREFIX, utils::u32_to_base64};
use tuliprox_core::model::ProxyUserCredentials;
@@ -254,6 +316,114 @@ mod test {
}
}
#[test]
fn bounded_archive_token_survives_master_variant_and_segment_rewrites() {
let master = "https://provider.example/channel/archive-1717200000-3600.m3u8?token=a%2Fb&auth=secret";
let variant = rewrite_hls_url(master, "tracks-v1a1/archive-1717200000-3600.m3u8");
assert_eq!(
variant,
"https://provider.example/channel/tracks-v1a1/archive-1717200000-3600.m3u8?token=a%2Fb&auth=secret"
);
let segment = rewrite_hls_url(&variant, "dvr/segment.ts");
assert_eq!(segment, "https://provider.example/channel/tracks-v1a1/dvr/segment.ts?token=a%2Fb&auth=secret");
assert_eq!(rewrite_hls_url(&variant, "//elsewhere.example/segment.ts"), "https://elsewhere.example/segment.ts");
}
#[test]
fn bounded_archive_session_preserves_queries_for_generic_variant_names() {
let props = RewriteHlsProps {
secret: &[7; 16],
base_url: "http://proxy",
content: "",
hls_url: "https://provider.example/channel/variant.m3u8?token=secret&auth=value".to_string(),
target_id: 1,
virtual_id: 42,
input_id: 1,
user_token: Some("m3u-catchup|fp|alice|42|archive|1717200000|3600"),
};
assert_eq!(
rewrite_hls_resource_url(&props, "segment.ts", "alice"),
"https://provider.example/channel/segment.ts?token=secret&auth=value"
);
assert_eq!(
rewrite_hls_resource_url(&props, "https://other.example/segment.ts", "alice"),
"https://other.example/segment.ts"
);
let live = RewriteHlsProps { user_token: None, ..props };
assert_eq!(
rewrite_hls_resource_url(&live, "segment.ts", "alice"),
"https://provider.example/channel/segment.ts?token=secret"
);
}
#[test]
fn archive_text_in_username_does_not_enable_bounded_archive_queries() {
let mut props = RewriteHlsProps {
secret: &[7; 16],
base_url: "http://proxy",
content: "",
hls_url: "https://provider.example/channel/variant.m3u8?token=secret&auth=value".to_string(),
target_id: 1,
virtual_id: 42,
input_id: 1,
user_token: None,
};
for (token, username) in [
("m3u-catchup|fp|alice|42|archive|42|3600", "alice|42|archive"),
("fp|alice|archive|name|42", "alice|archive|name"),
("m3u-catchup|fp|alice|archive|name|42|live", "alice|archive|name"),
("m3u-catchup|fp|alice|archive|name|42|opaque", "alice|archive|name"),
("m3u-catchup|fp|alice|42|archive|invalid|3600", "alice"),
("m3u-catchup|fp|alice|42|archive|1717200000|0", "alice"),
("m3u-catchup|fp|alice|43|archive|1717200000|3600", "alice"),
("m3u-catchup|fp|alice|42|archive|1717200000|3600|unexpected", "alice"),
] {
props.user_token = Some(token);
assert_eq!(
rewrite_hls_resource_url(&props, "segment.ts", username),
"https://provider.example/channel/segment.ts?token=secret",
"{token}"
);
}
for (token, username) in [
("m3u-catchup|fp|alice|archive|name|42|archive|1717200000|3600", "alice|archive|name"),
("m3u-catchup|fp|alice|42|archive|1717200000|3600|.m3u8", "alice"),
("catchup|fp|alice|42|archive|1717200000|3600", "alice"),
] {
props.user_token = Some(token);
assert_eq!(
rewrite_hls_resource_url(&props, "segment.ts", username),
"https://provider.example/channel/segment.ts?token=secret&auth=value",
"{token}"
);
}
}
#[test]
fn bounded_archive_inherits_query_only_on_same_origin_and_keeps_child_credentials() {
let base = "https://provider.example/channel/archive-1717200000-3600.m3u8?token=parent&auth=secret";
assert_eq!(
rewrite_hls_url(base, "https://provider.example/channel/segment.ts?token=child"),
"https://provider.example/channel/segment.ts?token=child&auth=secret"
);
assert_eq!(rewrite_hls_url(base, "https://other.example/segment.ts"), "https://other.example/segment.ts");
assert_eq!(
rewrite_hls_url(base, "https://provider.example:444/segment.ts"),
"https://provider.example:444/segment.ts"
);
assert_eq!(
rewrite_hls_url(base, "key.bin?AUTH=child"),
"https://provider.example/channel/key.bin?AUTH=child&token=parent"
);
assert_eq!(
rewrite_hls_url(
"https://provider.example/channel/archive-1717200000-3600.ts?token=parent&auth=secret",
"seg.ts"
),
"https://provider.example/channel/seg.ts?token=parent"
);
}
#[test]
fn rewrite_relative_hls_resource_keeps_its_own_token() {
let base = "https://cdn.example/hls/channel/index.m3u8?TOKEN=base";
backend/parser/src/m3u_format/catchup.rs
+165 -3
View File
@@ -21,6 +21,12 @@ const XC_START_TEMPLATE: &str = "{Y}-{m}-{d}:{H}-{M}-{S}";
const FLUSSONIC_UTC_SENTINEL: &str = "__TULIPROX_M3U_CATCHUP_UTC__";
const XTREAM_BRIDGE_START_FORMATS: [&str; 4] = ["%Y-%m-%d %H:%M", "%Y-%m-%d:%H-%M", "%Y-%m-%d:%H:%M", "%Y-%m-%d-%H-%M"];
const MAX_CATCHUP_FUTURE_SKEW_SECS: i64 = 10 * 60;
const MAX_FLUSSONIC_ARCHIVE_DURATION_SECS: i64 = 7 * 24 * 60 * 60;
fn has_flussonic_hls_source_override(catchup: &CatchupProperties) -> bool {
catchup.native_flussonic_player_mode() == Some("flussonic")
&& catchup.source.as_deref().is_some_and(|source| !source.trim().is_empty())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct M3uCatchupRewrite {
@@ -448,7 +454,7 @@ pub fn build_m3u_catchup_rewrite(
source_url: &str,
catchup: &CatchupProperties,
) -> Result<Option<M3uCatchupRewrite>, TuliproxError> {
if catchup.native_flussonic_player_mode().is_some() {
if catchup.native_flussonic_player_mode().is_some() && !has_flussonic_hls_source_override(catchup) {
return Ok(None);
}
let Some(template) = derived_template_for_mode(source_url, catchup) else {
@@ -562,6 +568,51 @@ pub fn resolve_xtream_m3u_catchup_url(
Ok(ResolvedM3uCatchup { url, discriminator })
}
fn validate_flussonic_archive_collectors(
placeholders: &[&str],
collectors: &mut [(usize, String)],
raw_query: Option<&str>,
) -> Result<Option<String>, TuliproxError> {
let Some(start_idx) = placeholders.iter().position(|p| placeholder_name(p) == "utc") else {
return Ok(None);
};
let Some(duration_idx) = placeholders.iter().position(|p| placeholder_name(p) == "duration") else {
return Ok(None);
};
let invalid_window = || TuliproxError::RepositoryM3u("Invalid bounded Flussonic archive range".to_string());
let start =
parse_catchup_timestamp(&collectors[start_idx].1).filter(|start| *start >= 0).ok_or_else(invalid_window)?;
validate_catchup_start(start)?;
let mut duration = collectors[duration_idx].1.parse::<i64>().map_err(|_| invalid_window())?;
if let Some(query) = raw_query {
let params: Vec<_> = url::form_urlencoded::parse(query.as_bytes()).collect();
let utc = params.iter().find(|(key, _)| key.eq_ignore_ascii_case("utc"));
let lutc = params.iter().find(|(key, _)| key.eq_ignore_ascii_case("lutc"));
if utc.is_some() || lutc.is_some() {
let end = lutc.and_then(|(_, value)| parse_catchup_timestamp(value)).ok_or_else(invalid_window)?;
if utc.and_then(|(_, value)| parse_catchup_timestamp(value)) != Some(start) {
return Err(invalid_window());
}
duration = end.checked_sub(start).ok_or_else(invalid_window)?;
}
}
if !(1..=MAX_FLUSSONIC_ARCHIVE_DURATION_SECS).contains(&duration) {
return Err(TuliproxError::RepositoryM3u(
"Flussonic archive duration must be between 1 second and 7 days".to_string(),
));
}
start.checked_add(duration).ok_or_else(invalid_window)?;
// All repeated placeholders use the same validated range, including indexed player requests.
for ((_, value), placeholder) in collectors.iter_mut().zip(placeholders) {
match placeholder_name(placeholder) {
"utc" => *value = start.to_string(),
"duration" => *value = duration.to_string(),
_ => {}
}
}
Ok(Some(format!("archive|{start}|{duration}")))
}
pub fn resolve_m3u_catchup_url(
source_url: &str,
catchup: &CatchupProperties,
@@ -572,7 +623,7 @@ pub fn resolve_m3u_catchup_url(
};
let segments = parse_template(template.as_ref());
let placeholders = collect_placeholders(&segments);
let collectors = resolve_collectors(raw_query, &placeholders);
let mut collectors = resolve_collectors(raw_query, &placeholders);
if collectors.len() != placeholders.len() {
return Err(TuliproxError::Crypto(format!(
"Catchup collector mismatch: expected {}, got {}",
@@ -581,10 +632,15 @@ pub fn resolve_m3u_catchup_url(
)));
}
let bounded_discriminator = if has_flussonic_hls_source_override(catchup) {
validate_flussonic_archive_collectors(&placeholders, &mut collectors, raw_query)?
} else {
None
};
let url = render_template(&segments, &collectors)?;
// Prefer archive|{utc}|{duration} so Streams/History can recover EPG from the session token
// after append/shift query params are stripped from rewritten HLS segment URLs.
let discriminator = archive_discriminator_from_resolved_url(&url)?.unwrap_or_else(|| {
let discriminator = bounded_discriminator.or(archive_discriminator_from_resolved_url(&url)?).unwrap_or_else(|| {
let mut discriminator = url::form_urlencoded::Serializer::new(String::new());
let mode = catchup.effective_mode().unwrap_or_default();
discriminator.append_pair("mode", if mode.is_empty() { "default" } else { mode });
@@ -604,6 +660,112 @@ mod tests {
};
use shared::{error::TuliproxError, model::CatchupProperties, utils::Internable};
fn bounded_flussonic_properties() -> CatchupProperties {
CatchupProperties {
mode: Some("fs".intern()),
catchup_type: Some("flussonic".intern()),
source: Some(
"https://provider.example/channel/archive-{utc}-{duration}.m3u8?token=a%2Fb&auth=secret".intern(),
),
..CatchupProperties::default()
}
}
#[test]
fn flussonic_explicit_hls_source_is_exported() -> Result<(), TuliproxError> {
let mut catchup = bounded_flussonic_properties();
let rewrite = build_m3u_catchup_rewrite(
&[7; 16],
"http://proxy.example",
"alice",
7,
42,
"https://provider.example/channel/mono.m3u8?token=secret",
&catchup,
)?
.ok_or_else(|| TuliproxError::RepositoryM3u("Missing rewrite".to_string()))?;
assert_eq!(rewrite.mode.as_ref(), "default");
assert!(rewrite.source.contains("v0={utc}&v1={duration}"));
assert!(!rewrite.source.contains("secret"));
catchup.source = None;
assert!(build_m3u_catchup_rewrite(
&[7; 16],
"http://proxy.example",
"alice",
7,
42,
"https://provider.example/channel/mono.m3u8",
&catchup
)?
.is_none());
catchup = bounded_flussonic_properties();
catchup.catchup_type = Some("flussonic-ts".intern());
assert!(build_m3u_catchup_rewrite(
&[7; 16],
"http://proxy.example",
"alice",
7,
42,
"https://provider.example/channel/mpegts",
&catchup
)?
.is_none());
Ok(())
}
#[test]
fn flussonic_bounded_source_uses_validated_window_and_distinct_identity() -> Result<(), TuliproxError> {
let catchup = bounded_flussonic_properties();
let first = resolve_m3u_catchup_url(
"https://provider.example/channel/mono.m3u8",
&catchup,
Some("utc=1717200000&lutc=1717203600&duration=99999999"),
)?
.ok_or_else(|| TuliproxError::RepositoryM3u("Missing resolution".to_string()))?;
assert_eq!(first.url, "https://provider.example/channel/archive-1717200000-3600.m3u8?token=a%2Fb&auth=secret");
assert_eq!(first.discriminator, "archive|1717200000|3600");
let indexed = resolve_m3u_catchup_url(
"https://provider.example/channel/mono.m3u8",
&catchup,
Some("v0=1717200000&v1=3600"),
)?;
assert_eq!(indexed, Some(first.clone()));
let second = resolve_m3u_catchup_url(
"https://provider.example/channel/mono.m3u8",
&catchup,
Some("utc=1717200000&lutc=1717207200"),
)?
.ok_or_else(|| TuliproxError::RepositoryM3u("Missing resolution".to_string()))?;
assert_ne!(first.discriminator, second.discriminator);
assert_ne!(first.url, second.url);
Ok(())
}
#[test]
fn flussonic_bounded_source_rejects_invalid_or_excessive_windows() {
let catchup = bounded_flussonic_properties();
for query in [
"utc=1717200000",
"lutc=1717203600",
"utc=invalid&lutc=1717203600",
"utc=1717200000&lutc=invalid",
"utc=1717200000&lutc=1717200000",
"utc=1717203600&lutc=1717200000",
"utc=1717200000&lutc=1717804801",
"utc=1717200000&duration=3600",
"v0=1717200000&v1=0",
"v0=1717200000&v1=-3600",
"v0=1717200000&v1=604801",
"v0=-1&v1=3600",
"v0=999999999999999999&v1=3600",
] {
assert!(
resolve_m3u_catchup_url("https://provider.example/channel/mono.m3u8", &catchup, Some(query)).is_err(),
"{query}"
);
}
}
#[test]
fn explicit_append_rewrite_uses_live_route_marker() {
let rewrite = build_m3u_catchup_rewrite(
shared/src/model/playlist.rs
+32 -1
View File
@@ -672,7 +672,9 @@ fn append_m3u_catchup_attributes(
}
append_catchup_attribute(line, "catchup-time", catchup.time.as_ref());
append_catchup_attribute(line, "catchup-correction", catchup.correction.as_ref());
append_catchup_attribute(line, "catchup-type", catchup.catchup_type.as_ref());
if !catchup.is_flussonic() || rewritten_mode.is_none_or(|mode| mode.is_empty()) {
append_catchup_attribute(line, "catchup-type", catchup.catchup_type.as_ref());
}
append_extra_catchup_attributes(line, &catchup.extra_attributes);
}
@@ -2462,6 +2464,35 @@ mod tests {
assert!(!output.contains(r#"catchup-source="?offset=-${offset}""#));
}
#[test]
fn m3u_to_m3u_flussonic_source_override_uses_rewritten_mode() {
let mut item = M3uPlaylistItem::from(&PlaylistItem { header: PlaylistItemHeader::default() });
item.url = "https://provider.example/channel/mono.m3u8?token=secret".intern();
item.t_stream_url = "http://proxy.example/m3u-stream/alice/pass/42/index.m3u8".intern();
item.t_catchup_mode = Some("default".intern());
item.t_catchup_source = Some("http://proxy.example/m3u-catchup/token?v0={utc}&v1={duration}".intern());
item.additional_properties = Some(StreamProperties::Live(Box::new(LiveStreamProperties {
catchup: Some(CatchupProperties {
mode: Some("fs".intern()),
catchup_type: Some("flussonic".intern()),
source: Some("https://provider.example/channel/archive-{utc}-{duration}.m3u8?token=secret".intern()),
..CatchupProperties::default()
}),
..LiveStreamProperties::default()
})));
let output = item.to_m3u(None, false);
assert!(output.contains(r#"catchup="default""#));
assert!(output.contains("v0={utc}&v1={duration}"));
assert!(!output.contains("catchup-type="));
assert!(!output.contains("token=secret"));
assert!(output.ends_with("/42/index.m3u8"));
item.t_catchup_mode = None;
item.t_catchup_source = None;
let native = item.to_m3u(None, false);
assert!(native.contains(r#"catchup-type="flussonic""#));
assert!(!native.contains("catchup-source="));
}
#[test]
fn m3u_to_m3u_emits_only_configured_upstream_user_agent() {
let mut item = M3uPlaylistItem::from(&PlaylistItem { header: PlaylistItemHeader::default() });