From 278bf41d2c548d09fa848c9e59837705156dc5c6 Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Sat, 26 Sep 2026 18:56:17 +0200 Subject: [PATCH] Fix/stalker (#884) * **New Features** * Stalker playback resolves channel links on request, using available channel commands and falling back to the stored command when needed. * If a portal rejects a session, playback retries once after reconnecting. * **Bug Fixes** * Reading playlists or resolving playback for a different portal identity no longer replaces the published catalog or clears its refresh checkpoint. * A failed channel playback no longer prevents other published channels from remaining available. * **Documentation** * Updated Stalker playback guidance to describe on-demand link resolution and session recovery. --- CHANGELOG.md | 30 ++ .../app/src/api/endpoints/v1_api_playlist.rs | 115 ++++++ backend/iptv/src/stalker/catalog.rs | 2 + backend/iptv/src/stalker/client.rs | 69 +++- backend/iptv/src/stalker/parser.rs | 21 +- backend/iptv/src/stalker/playback.rs | 62 +-- backend/processing/src/processor/stalker.rs | 179 +++++++-- backend/repository/src/playlist_repository.rs | 11 +- .../src/stalker_generation_repository.rs | 56 +++ docs/src/configuration/source.md | 29 +- docs/src/rest-api-cookbook.md | 10 +- frontend/public/config.json | 1 + .../scenarios/stalker-live-resolution.yml | 61 +++ .../stalker-refused-channel-keeps-catalog.yml | 52 +++ .../stalker-session-refused-once.yml | 46 +++ tools/tuliprox-testkit/src/bootstrap.rs | 107 ++++- tools/tuliprox-testkit/src/config.rs | 190 ++++++++- tools/tuliprox-testkit/src/main.rs | 366 +++++++++++++++++- tools/tuliprox-testkit/src/origin_events.rs | 17 + 19 files changed, 1339 insertions(+), 85 deletions(-) create mode 100644 test/fixtures/testkit/scenarios/stalker-live-resolution.yml create mode 100644 test/fixtures/testkit/scenarios/stalker-refused-channel-keeps-catalog.yml create mode 100644 test/fixtures/testkit/scenarios/stalker-session-refused-once.yml diff --git a/CHANGELOG.md b/CHANGELOG.md index 5400b489e..00d38f6d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1024,6 +1024,23 @@ ## 🐛 Fixes +- **Stalker playback resolution now says why it failed, retries a rejected session, and can fall back to the stored + command.** A playback request that could not be resolved reported a single message naming the requested item's portal + id, while the actual cause — no published catalog for the configured portal identity, an item missing from the active + generation, an item without a playback descriptor, or a `create_link` refusal — was written to the debug log or + nowhere at all, which made a portal refusal look like a mismatched stream id. Each of these stages now logs at warning + level with the input name, stream id, item name and the portal's sanitized error, and the resolution reports the + portal's reason once its candidates are exhausted. A portal that rejects the session (HTTP 204/401/403/456, or a + Ministra `code` of 44 / 440..=449 inside a `200 OK`) no longer ends as "could not be resolved": the cached session is + dropped, a fresh handshake is issued and the request is retried once, which is the recovery the typed token-rejection + error was introduced for but nothing called. Items whose persisted playback descriptor is empty now resolve through + their stored `cmd`, so a row that still carries the portal command stays playable. +- **A playback request no longer discards the published Stalker catalog.** The active-manifest lookup used by playback + resolution and by the disk playlist sources replaced the published manifest with an empty one and deleted the refresh + checkpoint whenever the stored identity did not match the configured one — for example after editing the input's + Stalker block or changing the MAG preset. The catalog files survived but nothing could resolve, and the running + refresh lost its resume point. Read paths now report "no published catalog for this identity" and leave the + publication state to the refresh that owns it. - **Resource URLs no longer expose internal destinations.** Resource URLs are classified before they are written into playlist or EPG output: only a provably public destination is handed to the client, everything else (`tvg-logo`, `tvg-logo-small`, Xtream covers and backdrops, EPG channel and programme icons, Web UI item icons) is replaced by an @@ -1568,6 +1585,19 @@ ## 🛠 Maintenance +- **The testkit can now drive a Stalker/Ministra input.** The fixture origin emulates a portal + (`handshake`, `get_profile`, `get_genres`, `get_ordered_list`, `create_link`), a scenario selects it with + `input_type: stalker` plus a `stalker:` block (MAC, MAG preset, scripted `create_link` refusals), and the + generated `source.yml` points the input at it. Playback resolution is asserted where it is observable — on the + portal: `assert_origin` gained `stalker_handshakes_at_least`, `stalker_create_links_at_least`, + `stalker_token_refusals_at_least` and `stalker_create_link_markers` (set comparison, so a playback that resolved + another catalog item's command fails). Three scenarios use it: `stalker-live-resolution` (each channel resolves + through its own stored `cmd`), `stalker-session-refused-once` (a stale-session refusal is recovered by a + re-handshake and retry; fails without that recovery) and `stalker-refused-channel-keeps-catalog` (a permanently + refused channel is reported and the catalog keeps serving the others). The fixture portal answers with a loopback + stream URL on purpose: the SUT's destination guard refuses it, so the scenarios assert the resolution chain and + the guard instead of pretending a portal-supplied private URL is playable. Asserting a hard upstream failure + needed a new expectation, `expect: { http_error: }`. - **Playlist curation now has a dedicated capability boundary**: matching and ordered membership evaluation live in the source-neutral `tuliprox-curation` kernel, while Trakt HTTP/JSON handling translates records at the edge and the category-scoped compatibility projector remains separate from membership identity. Existing category identity and diff --git a/backend/app/src/api/endpoints/v1_api_playlist.rs b/backend/app/src/api/endpoints/v1_api_playlist.rs index 96ec449e4..3b0a0c277 100644 --- a/backend/app/src/api/endpoints/v1_api_playlist.rs +++ b/backend/app/src/api/endpoints/v1_api_playlist.rs @@ -3553,6 +3553,121 @@ mod tests { server_handle.abort(); } + /// The playback endpoints resolve a Stalker item from the catalog through the item's own + /// provider id. This is the path behind `Failed to resolve initial Stalker playback URL`, + /// so it needs the real id -> item -> `create_link` chain against a portal, not just the + /// Web UI preview route. + #[tokio::test] + async fn initial_stalker_playback_resolution_uses_the_requested_items_own_command() { + let temp_dir = tempdir().expect("temp dir"); + let (base_url, server_handle) = spawn_stalker_mock_server().await; + let input = Arc::new(ConfigInput { + id: 7, + name: "stalker".intern(), + input_type: InputType::Stalker, + url: base_url, + enabled: true, + options: Some(ConfigInputOptions { + flags: crate::model::ConfigInputFlagsSet::new(), + update_quality: ConfigInputUpdateQuality::default(), + resolve_delay: shared::defaults::default_resolve_delay_secs(), + probe_delay: shared::defaults::default_probe_delay_secs(), + probe_live_interval_hours: 120, + resolve_filter: None, + probe_filter: None, + }), + stalker: Some(crate::model::StalkerInputConfig { + device: None, + auth_mode: shared::model::StalkerAuthMode::Auto, + mag_preset: shared::model::StalkerMagPreset::GenericSafe, + endpoint_preference: shared::model::StalkerEndpointPreference::ServerLoad, + size_caps: None, + catalog_max_pages: None, + ..Default::default() + }), + ..Default::default() + }); + let source = ConfigSource { inputs: vec![Arc::clone(&input.name)], targets: vec![] }; + let app_config = test_app_config(Arc::clone(&input), source); + app_config.config.store(Arc::new(Config { + storage_dir: temp_dir.path().to_string_lossy().to_string(), + ..Default::default() + })); + let app_state = test_app_state(Arc::new(app_config)); + let router = super::v1_api_playlist_register_protected(super::v1_api_playlist_register_public(Router::new())) + .with_state(Arc::clone(&app_state)); + + // The catalog import publishes the generation the playback path reads. + let response = router + .clone() + .into_service::() + .oneshot( + Request::builder() + .method("POST") + .uri("/playlist/live") + .header("content-type", "application/json") + .body(Body::from(r#"{"Input":"stalker"}"#)) + .expect("request"), + ) + .await + .expect("catalog response"); + assert_eq!(response.status(), StatusCode::OK); + + let live_input = app_state.app_config.get_input_by_name(&"stalker".intern()).expect("stalker input"); + let unresolved: Arc = "".intern(); + let live = shared::model::PlaylistItemType::Live; + let resolve = |provider_id: u32| { + let app_state = Arc::clone(&app_state); + let live_input = Arc::clone(&live_input); + let unresolved = Arc::clone(&unresolved); + async move { + crate::api::api_utils::resolve_initial_stalker_playback_url( + &app_state, + &live_input, + provider_id, + XtreamCluster::Live, + live, + &unresolved, + ) + .await + } + }; + + // An id that is not in the catalog must not resolve, and must not invalidate the + // published generation the next request reads. + let err = resolve(999_999).await.expect_err("an unknown provider id cannot be resolved"); + assert!(err.to_string().contains("999999"), "{err}"); + + // Each item resolves through its own stored cmd: the mock portal answers a different + // destination per cmd, so a provider id that picked the wrong item would surface here. + assert_eq!( + resolve(101).await.expect("item 101 resolves").as_ref(), + "http://8.8.8.8/live/101", + "the resolved url must come from item 101's own cmd" + ); + assert_eq!( + resolve(102).await.expect("item 102 resolves").as_ref(), + "http://127.0.0.1/live/102", + "the resolved url must come from item 102's own cmd" + ); + + // An item that already carries a url is served as is. + let resolved_url: Arc = "http://stream.example/already-resolved.ts".intern(); + let passthrough = crate::api::api_utils::resolve_initial_stalker_playback_url( + &app_state, + &live_input, + 101, + XtreamCluster::Live, + live, + &resolved_url, + ) + .await + .expect("a resolved url is passed through"); + assert!(Arc::ptr_eq(&passthrough, &resolved_url)); + + server_handle.abort(); + } + #[tokio::test] async fn playlist_epg_input_route_returns_cached_provider_epg() { let temp_dir = tempdir().expect("temp dir"); diff --git a/backend/iptv/src/stalker/catalog.rs b/backend/iptv/src/stalker/catalog.rs index 037d0493f..f73e47fc8 100644 --- a/backend/iptv/src/stalker/catalog.rs +++ b/backend/iptv/src/stalker/catalog.rs @@ -55,6 +55,8 @@ pub struct StalkerRawItem { #[serde(default)] pub cmd: Option, #[serde(default)] + pub cmd_1: Option, + #[serde(default)] pub logo: Option, #[serde(default)] pub stream_icon: Option, diff --git a/backend/iptv/src/stalker/client.rs b/backend/iptv/src/stalker/client.rs index b55256ab7..c63d3aa7f 100644 --- a/backend/iptv/src/stalker/client.rs +++ b/backend/iptv/src/stalker/client.rs @@ -848,7 +848,9 @@ mod transport_tests { session::StalkerSession, transport::testing::{FakeTransport, Reply}, }; - use shared::model::stalker::{StalkerBootstrapRecipe, StalkerPortalCapabilitiesDto}; + use shared::model::stalker::{ + StalkerBootstrapRecipe, StalkerPlaybackMode, StalkerPortalCapabilitiesDto, StalkerStreamKind, + }; use tuliprox_core::utils::ManualClock; const PORTAL: &str = "http://portal.example/stalker_portal/"; @@ -948,6 +950,71 @@ mod transport_tests { ); } + /// Portals invalidate sessions out of band (another device on the same MAC, a short TTL), + /// and then answer `create_link` with a stale-token refusal. Treating that as "this item + /// cannot be resolved" is what made playback fail without a usable reason. + #[tokio::test] + async fn create_link_rehandshakes_once_when_the_portal_rejects_the_session() { + let transport = std::sync::Arc::new(FakeTransport::new([ + // create_link against the stale session. + Reply::ok(r#"{"code": 44, "text": "Authorization failed"}"#), + // Re-handshake: handshake, get_profile, capabilities. + Reply::ok(r#"{"js":{"token":"fresh"}}"#), + Reply::ok(r#"{"js":{"status":"1","max_connections":"1"}}"#), + Reply::ok("{}"), + // create_link retried with the fresh session. + Reply::ok(r#"{"js":{"cmd":"ffmpeg http://line.example/live/7.ts"}}"#), + ])); + let client = client_with(std::sync::Arc::clone(&transport), StalkerInputConfig::default()); + + let resolved = client + .create_link( + &handshake(), + StalkerStreamKind::Live, + StalkerPlaybackMode::DirectUrl, + "ffmpeg http://portal.example/ch/7", + None, + None, + None, + ) + .await + .expect("the item resolves after the session is refreshed"); + + assert_eq!(resolved.stream_url, "http://line.example/live/7.ts"); + assert_eq!(transport.requested().len(), 5, "one refused create_link, one re-handshake, one resolved retry"); + } + + /// A fresh session that is refused as well must be reported, not retried forever. + #[tokio::test] + async fn create_link_reports_a_session_rejected_twice() { + let transport = std::sync::Arc::new(FakeTransport::new([ + Reply::ok(r#"{"code": 44, "text": "Authorization failed"}"#), + Reply::ok(r#"{"js":{"token":"fresh"}}"#), + Reply::ok(r#"{"js":{"status":"1","max_connections":"1"}}"#), + Reply::ok("{}"), + Reply::ok(r#"{"code": 44, "text": "Authorization failed"}"#), + Reply::ok(r#"{"code": 44, "text": "Authorization failed"}"#), + Reply::ok(r#"{"code": 44, "text": "Authorization failed"}"#), + ])); + let client = client_with(std::sync::Arc::clone(&transport), StalkerInputConfig::default()); + + let err = client + .create_link( + &handshake(), + StalkerStreamKind::Live, + StalkerPlaybackMode::DirectUrl, + "ffmpeg http://portal.example/ch/7", + None, + None, + None, + ) + .await + .expect_err("a portal that keeps refusing the session cannot resolve the item"); + + assert!(err.is_token_rejected(), "the refusal must stay classifiable: {err}"); + assert_eq!(transport.requested().len(), 7, "at most one re-handshake per call"); + } + /// Every candidate failing must surface the last real error, not a synthetic one. #[tokio::test] async fn exhausting_every_candidate_reports_the_last_failure() { diff --git a/backend/iptv/src/stalker/parser.rs b/backend/iptv/src/stalker/parser.rs index 54a8f2d61..777f999c0 100644 --- a/backend/iptv/src/stalker/parser.rs +++ b/backend/iptv/src/stalker/parser.rs @@ -154,7 +154,7 @@ where } fn build_descriptor_from_raw(raw: &StalkerRawItem) -> Option { - let cmd = raw.cmd.clone().filter(|s| !s.is_empty())?; + let cmd = raw.cmd_1.as_ref().filter(|s| !s.is_empty()).or(raw.cmd.as_ref().filter(|s| !s.is_empty()))?.clone(); let mode = playback_mode_from_flags(StalkerTempLinkFlags::from(raw)); let candidates = vec![StalkerCommandVariantDto { cmd, playback_mode: mode, source_key: None, priority: 0 }]; Some(StalkerPlaybackDescriptorDto { primary_mode: mode, candidates, capabilities: None }) @@ -649,4 +649,23 @@ mod tests { let desc = build_descriptor_from_raw(&raw).expect("descriptor"); assert_eq!(desc.primary_mode, StalkerPlaybackMode::DirectUrl); } + + #[test] + fn alternate_command_is_persisted_separately_from_the_raw_fallback() { + let raw = StalkerRawItem { + cmd: Some("ffmpeg http://portal/raw/17".to_string()), + cmd_1: Some("ffmpeg http://portal/descriptor/17".to_string()), + ..Default::default() + }; + let item = map_stalker_to_playlist_item(&raw, None, StalkerStreamKind::Live, 0); + + assert_eq!(item.cmd.as_ref(), "ffmpeg http://portal/raw/17"); + assert_eq!( + item.playback_descriptor + .as_ref() + .and_then(|descriptor| descriptor.candidates.first()) + .map(|candidate| candidate.cmd.as_str()), + Some("ffmpeg http://portal/descriptor/17") + ); + } } diff --git a/backend/iptv/src/stalker/playback.rs b/backend/iptv/src/stalker/playback.rs index e03d05168..16aa6ea40 100644 --- a/backend/iptv/src/stalker/playback.rs +++ b/backend/iptv/src/stalker/playback.rs @@ -46,31 +46,49 @@ pub async fn create_link( archive_start: Option<&str>, archive_end: Option<&str>, ) -> StalkerResult { - let spec = recipe_spec_for(handshake.profile.bootstrap_recipe); - let candidates = client.ordered_load_urls(); - let mut last_err: Option = None; - for load_url in candidates { - let builder = build_create_link_builder( - client, - &load_url, - handshake, - &spec, - kind, - cmd, - series_number, - archive_start, - archive_end, - ); - match client.send_json::(builder, StalkerAction::CreateLink).await { - Ok(resp) => { - return resolve_response(resp, kind, requested_mode, cmd); - } - Err(err) => { - last_err = Some(err); + let mut handshake = handshake.clone(); + // A rejected token invalidates every endpoint candidate at once, so the whole chain is + // retried once against a fresh session instead of failing the item. + let mut allow_session_retry = true; + 'session: loop { + let spec = recipe_spec_for(handshake.profile.bootstrap_recipe); + let candidates = client.ordered_load_urls(); + let mut last_err: Option = None; + for load_url in candidates { + let builder = build_create_link_builder( + client, + &load_url, + &handshake, + &spec, + kind, + cmd, + series_number, + archive_start, + archive_end, + ); + match client.send_json::(builder, StalkerAction::CreateLink).await { + Ok(resp) => { + return resolve_response(resp, kind, requested_mode, cmd); + } + Err(err) => { + if allow_session_retry && err.is_token_rejected() { + // Ministra reports a stale token either as 401/403/204/456 or inside a + // `200 OK` body (`code` 44 / 440..=449). The session we handshook with is + // gone, so drop it, handshake again and retry the chain once. + warn!("Stalker create_link token was rejected; re-handshaking and retrying the request"); + client.invalidate_session(); + handshake = client.handshake().await?; + allow_session_retry = false; + continue 'session; + } + last_err = Some(err); + } } } + return Err( + last_err.unwrap_or_else(|| StalkerError::NoEndpoint { portal: safe_stalker_url(client.portal_url()) }) + ); } - Err(last_err.unwrap_or_else(|| StalkerError::NoEndpoint { portal: safe_stalker_url(client.portal_url()) })) } #[allow(clippy::too_many_arguments)] diff --git a/backend/processing/src/processor/stalker.rs b/backend/processing/src/processor/stalker.rs index c61e95439..e4edc956a 100644 --- a/backend/processing/src/processor/stalker.rs +++ b/backend/processing/src/processor/stalker.rs @@ -11,15 +11,17 @@ #![allow(clippy::too_many_lines, clippy::needless_pass_by_value)] use super::stalker_refresh::{advance_stalker_refresh, StalkerRefreshMode, StalkerRefreshOutcome, StalkerRefreshPlan}; -use log::{debug, info, warn}; +use log::{info, warn}; use lru::LruCache; use parking_lot::Mutex; use shared::{ error::TuliproxError, model::{ - stalker::StalkerStreamKind, stalker_item::StalkerPlaylistItem, PlaylistGroup, PlaylistItem, UpdateQualityPolicy, + stalker::{StalkerPlaybackMode, StalkerStreamKind}, + stalker_item::StalkerPlaylistItem, + PlaylistGroup, PlaylistItem, UpdateQualityPolicy, }, - utils::Internable, + utils::{sanitize_sensitive_info, Internable}, }; use std::{ borrow::Cow, @@ -34,7 +36,7 @@ use tuliprox_iptv::{ stalker::{client::StalkerApiClient, error::StalkerError, parser}, }; use tuliprox_repository::{ - stalker_generation_repository::{load_active_manifest, load_checkpoint}, + stalker_generation_repository::{load_active_manifest, load_checkpoint, readable_active_manifest}, stalker_repository::{ensure_stalker_storage_path, load_stalker_items_at, read_stalker_item_at}, }; @@ -467,6 +469,34 @@ fn cached_runtime_stalker_client( Ok(client) } +/// Portal commands a Stalker item can be resolved with, most specific first. +/// +/// The persisted playback descriptor is authoritative. When it is absent or holds no usable +/// command, the item's raw `cmd` column is still enough to call `create_link`, so an item is +/// not declared unplayable while its stored command would have resolved. +fn playback_candidates(item: &StalkerPlaylistItem) -> Vec<(&str, StalkerPlaybackMode)> { + let mut candidates: Vec<(&str, StalkerPlaybackMode)> = item + .playback_descriptor + .as_ref() + .map(|descriptor| { + descriptor + .candidates + .iter() + .filter(|candidate| !candidate.cmd.trim().is_empty()) + .map(|candidate| (candidate.cmd.as_str(), candidate.playback_mode)) + .collect() + }) + .unwrap_or_default(); + if candidates.is_empty() && !item.cmd.trim().is_empty() { + let primary_mode = item + .playback_descriptor + .as_ref() + .map_or(StalkerPlaybackMode::DirectUrl, |descriptor| descriptor.primary_mode); + candidates.push((item.cmd.as_ref(), primary_mode)); + } + candidates +} + pub async fn re_resolve_stalker_url( app_config: &Arc, http_client: &reqwest::Client, @@ -484,7 +514,15 @@ pub async fn re_resolve_stalker_url( let portal_url = resolve_stalker_portal_url(input)?; let identity_fingerprint = stalker_cfg.identity_fingerprint(&portal_url); let storage_path = ensure_stalker_storage_path(app_config, &input.name).await?; - let manifest = load_active_manifest(&storage_path, identity_fingerprint).await?; + // Read-only manifest lookup: a playback request must never start (or destroy) a refresh. + let (manifest, published) = readable_active_manifest(&storage_path, identity_fingerprint).await?; + if !published { + warn!( + "Stalker playback resolution for input '{}' found no published catalog for the configured portal identity; run a playlist update for this input", + input.name + ); + return Ok(None); + } let generation_and_path = match kind { StalkerStreamKind::Live | StalkerStreamKind::Archive => { manifest.live.as_ref().map(|files| (files.generation, &files.data)) @@ -492,53 +530,67 @@ pub async fn re_resolve_stalker_url( StalkerStreamKind::Movie => manifest.vod.as_ref().map(|files| (files.generation, &files.data)), StalkerStreamKind::Episode => manifest.series.as_ref().map(|files| (files.generation, &files.episodes)), }; - let Some((generation, item_path)) = generation_and_path else { return Ok(None) }; + let Some((generation, item_path)) = generation_and_path else { + warn!( + "Stalker playback resolution for input '{}' has no published {kind:?} catalog; run a playlist update with that cluster enabled", + input.name + ); + return Ok(None); + }; let link_key = RuntimeLinkKey { fingerprint: identity_fingerprint, generation, provider_id, kind }; if let Some(url) = cached_resolved_link(link_key, force_refresh) { return Ok(Some(url)); } let Some(item) = read_stalker_item_at(app_config, item_path, provider_id).await? else { + warn!( + "Stalker playback resolution for input '{}' found no {kind:?} item with provider id {provider_id} in generation {generation}", + input.name + ); return Ok(None); }; - let Some(descriptor) = item.playback_descriptor.as_ref() else { - debug!("Stalker re-resolve skipped: stream_id={} has no playback_descriptor", item.stream_id); - return Ok(None); - }; - if descriptor.candidates.is_empty() { - debug!("Stalker re-resolve skipped: stream_id={} descriptor has no candidate", item.stream_id); - return Ok(None); - } - if descriptor.candidates.iter().all(|candidate| candidate.cmd.trim().is_empty()) { - debug!("Stalker re-resolve skipped: stream_id={} descriptor candidates are empty", item.stream_id); + // The descriptor is the canonical candidate list; the raw `cmd` column is the fallback for + // rows that carry no usable descriptor (older generations, partially written selections). + let candidates = playback_candidates(&item); + if candidates.is_empty() { + warn!( + "Stalker runtime re-resolve skipped: stream_id={} ({}) has neither a playback descriptor candidate nor a stored cmd", + item.stream_id, item.name + ); return Ok(None); } let api_client = cached_runtime_stalker_client(http_client, portal_url, stalker_cfg)?; - let handshake = api_client.handshake().await.map_err(stalker_err_to_repo)?; let series_number = (kind == StalkerStreamKind::Episode).then_some(item.number); - for candidate in &descriptor.candidates { - if candidate.cmd.trim().is_empty() { - continue; - } - match api_client - .create_link(&handshake, kind, candidate.playback_mode, &candidate.cmd, series_number, None, None) - .await - { + let mut last_error: Option = None; + for (cmd, mode) in &candidates { + let handshake = api_client.handshake().await.map_err(stalker_err_to_repo)?; + match api_client.create_link(&handshake, kind, *mode, cmd, series_number, None, None).await { Ok(resolved) => { let url = Internable::intern(resolved.stream_url); cache_resolved_link(link_key, Arc::clone(&url)); return Ok(Some(url)); } Err(err) => { - debug!( - "Stalker runtime re-resolve candidate failed for stream_id={}, mode={:?}: {err}", - item.stream_id, candidate.playback_mode + // Playback failing silently is what makes a portal refusal indistinguishable + // from a missing catalog; the reason belongs in the operator's log. + warn!( + "Stalker create_link failed for input '{}' stream_id={} ({}) mode={mode:?}: {}", + input.name, + item.stream_id, + item.name, + sanitize_sensitive_info(&err.to_string()) ); + last_error = Some(sanitize_sensitive_info(&err.to_string()).into_owned()); } } } - warn!("Stalker runtime re-resolve failed for stream_id={}, invalidating stale stream_url", item.stream_id); + warn!( + "Stalker runtime re-resolve failed for stream_id={} ({}); invalidating stale stream_url (last portal error: {})", + item.stream_id, + item.name, + last_error.as_deref().unwrap_or("none") + ); RUNTIME_STALKER_LINKS.lock().pop(&link_key); Ok(None) } @@ -549,6 +601,75 @@ mod tests { fn runtime_cfg() -> StalkerInputConfig { StalkerInputConfig::default() } + fn stalker_item(cmd: &str, descriptor: Option) -> StalkerPlaylistItem { + let mut item = StalkerPlaylistItem { stream_id: 1063, cmd: cmd.into(), ..StalkerPlaylistItem::default() }; + if let Some(mode) = descriptor { + item.playback_descriptor = Some(shared::model::stalker::StalkerPlaybackDescriptorDto { + primary_mode: mode, + candidates: vec![shared::model::stalker::StalkerCommandVariantDto { + cmd: cmd.to_string(), + playback_mode: mode, + ..Default::default() + }], + capabilities: None, + }); + } + item + } + + #[test] + fn playback_candidates_prefers_the_persisted_descriptor() { + let item = stalker_item("ffrt http://portal.example/ch/1063", Some(StalkerPlaybackMode::TempLinkFlussonic)); + + let candidates = playback_candidates(&item); + + assert_eq!(candidates.len(), 1); + assert_eq!(candidates[0].0, "ffrt http://portal.example/ch/1063"); + assert_eq!(candidates[0].1, StalkerPlaybackMode::TempLinkFlussonic); + } + + #[test] + fn playback_candidates_falls_back_to_the_stored_cmd() { + let item = stalker_item("ffrt http://portal.example/ch/1063", None); + + let candidates = playback_candidates(&item); + + assert_eq!(candidates.len(), 1, "an item without a descriptor is still resolvable from its stored cmd"); + assert_eq!(candidates[0].0, "ffrt http://portal.example/ch/1063"); + assert_eq!(candidates[0].1, StalkerPlaybackMode::DirectUrl); + } + + #[test] + fn playback_candidates_skips_blank_candidates() { + let mut item = stalker_item("ffrt http://portal.example/ch/1063", None); + item.playback_descriptor = Some(shared::model::stalker::StalkerPlaybackDescriptorDto { + primary_mode: StalkerPlaybackMode::TempLinkNginx, + candidates: vec![shared::model::stalker::StalkerCommandVariantDto { + cmd: " ".to_string(), + playback_mode: StalkerPlaybackMode::TempLinkNginx, + ..Default::default() + }], + capabilities: None, + }); + + let candidates = playback_candidates(&item); + + assert_eq!(candidates.len(), 1); + assert_eq!( + candidates[0].1, + StalkerPlaybackMode::TempLinkNginx, + "the descriptor still names the mode the portal expects" + ); + assert_eq!(candidates[0].0, "ffrt http://portal.example/ch/1063"); + } + + #[test] + fn playback_candidates_are_empty_without_any_command() { + let item = stalker_item("", None); + + assert!(playback_candidates(&item).is_empty()); + } + #[test] fn raw_group_catalog_is_published_at_the_input_storage_root() { let stalker_path = std::path::Path::new("/storage/input_portal/stalker"); diff --git a/backend/repository/src/playlist_repository.rs b/backend/repository/src/playlist_repository.rs index 521e88427..6c9f415ff 100644 --- a/backend/repository/src/playlist_repository.rs +++ b/backend/repository/src/playlist_repository.rs @@ -1252,11 +1252,20 @@ pub async fn load_input_playlist( TuliproxError::ConfigInput(format!("Stalker input '{}' has no Stalker configuration", input.name)) })?; let portal_url = input.resolve_url(&input.url)?.into_owned(); - let manifest = crate::stalker_generation_repository::load_active_manifest( + // A read path: when the published manifest belongs to a different identity the + // input simply has nothing to serve yet, so fall back to an empty manifest + // instead of replacing the publication state of the refresh that owns it. + let (manifest, published) = crate::stalker_generation_repository::readable_active_manifest( &stalker_path, stalker_config.identity_fingerprint(&portal_url), ) .await?; + if !published { + debug!( + "Stalker input '{}' has no published catalog for its current identity; serving an empty playlist", + input.name + ); + } if disk_based_processing { let source = PlaylistSource::stalker_disk( StalkerDiskPlaylistSource::new(app_config, &stalker_path, Arc::clone(&input.name), manifest) diff --git a/backend/repository/src/stalker_generation_repository.rs b/backend/repository/src/stalker_generation_repository.rs index 1c759f389..10d60ecf3 100644 --- a/backend/repository/src/stalker_generation_repository.rs +++ b/backend/repository/src/stalker_generation_repository.rs @@ -163,6 +163,35 @@ pub async fn save_active_manifest(storage_path: &Path, manifest: &StalkerActiveM atomic_write_json(storage_path.join(ACTIVE_MANIFEST), manifest).await } +/// Read the published manifest for `identity_fingerprint` without changing publication state. +/// +/// [`load_active_manifest`] starts a fresh generation when the stored manifest belongs to a +/// different identity, which is what a refresh needs. Read paths (playback resolution, the +/// disk playlist sources) must not do that: a mismatched identity there only means "this +/// instance cannot serve the published catalog yet", and overwriting the manifest also +/// deletes the resume checkpoint of the refresh that owns it. +pub async fn lookup_active_manifest( + storage_path: &Path, + identity_fingerprint: u64, +) -> Result, TuliproxError> { + let path = storage_path.join(ACTIVE_MANIFEST); + let manifest = load_json::(&path).await?; + Ok(manifest + .filter(|manifest| manifest.schema == SCHEMA_VERSION && manifest.identity_fingerprint == identity_fingerprint)) +} + +/// Manifest for a read path: the published one when it belongs to `identity_fingerprint`, +/// otherwise an empty manifest that is never written to disk. +pub async fn readable_active_manifest( + storage_path: &Path, + identity_fingerprint: u64, +) -> Result<(StalkerActiveManifest, bool), TuliproxError> { + match lookup_active_manifest(storage_path, identity_fingerprint).await? { + Some(manifest) => Ok((manifest, true)), + None => Ok((StalkerActiveManifest::empty(identity_fingerprint), false)), + } +} + pub async fn load_active_manifest( storage_path: &Path, identity_fingerprint: u64, @@ -380,6 +409,33 @@ mod tests { Ok(()) } + #[tokio::test] + async fn a_read_path_lookup_never_replaces_the_published_manifest() -> Result<(), Box> { + let temp = tempfile::tempdir()?; + let mut manifest = StalkerActiveManifest::empty(17); + manifest.live = Some(ClusterFiles { generation: 1, data: "live-1.db".into() }); + save_active_manifest(temp.path(), &manifest).await?; + save_checkpoint(temp.path(), &StalkerCheckpoint::new(17, 2, 1, 123)).await?; + + assert_eq!(lookup_active_manifest(temp.path(), 17).await?.as_ref(), Some(&manifest)); + assert!(lookup_active_manifest(temp.path(), 18).await?.is_none(), "a foreign identity is not this instance's"); + assert_eq!( + load_active_manifest(temp.path(), 17).await?, + manifest, + "a lookup must leave the published manifest untouched" + ); + assert!( + load_checkpoint(temp.path(), 17).await?.is_some(), + "a lookup must leave the owning refresh checkpoint in place" + ); + + let (empty, published) = readable_active_manifest(temp.path(), 18).await?; + assert!(!published); + assert_eq!(empty, StalkerActiveManifest::empty(18)); + assert_eq!(load_active_manifest(temp.path(), 17).await?, manifest, "the fallback manifest must stay in memory"); + Ok(()) + } + #[tokio::test] async fn identity_change_invalidates_active_and_staging() -> Result<(), Box> { let temp = tempfile::tempdir()?; diff --git a/docs/src/configuration/source.md b/docs/src/configuration/source.md index 06a4800c2..d7c4680a7 100644 --- a/docs/src/configuration/source.md +++ b/docs/src/configuration/source.md @@ -159,9 +159,6 @@ inputs: device: mac_address: '00:1A:79:12:34:56' enabled: true - options: - stalker_pre_resolve_playback: false - stalker_runtime_resolve_playback: true ``` Stalker refreshes write pages into an unpublished generation. Live, VOD, series, and EPG selected for one refresh @@ -171,9 +168,9 @@ previous complete snapshot; a first import exposes no partial catalog. A saved c When `process_parallel` is enabled, progress messages include the input name. Targets wait for every enabled input in their source and begin as soon as that source is ready, without waiting for unrelated sources. -Use `stalker_pre_resolve_playback: true` if you want Tuliprox to materialize playback URLs during refresh whenever the portal already grants them. -Keep `stalker_runtime_resolve_playback: true` when the portal uses expiring or session-bound temp links that may need a fresh `create_link` -call later during playback. +Stalker playback is resolved on demand: the catalog is imported without materializing stream URLs, and a playback +request runs `create_link` against the portal when the item has no usable URL yet. A stale portal session is +re-handshaken once and the request is retried. Stalker supports four authentication modes: @@ -260,8 +257,6 @@ specific provider. | `resolve_delay` / `probe_delay` | Int | `2` | **Ban Protection:** Hard wait time (in seconds) between API or Probe requests to the *same* provider! Prevents API spamming. | | `resolve_filter` | String | - | Filter expression to selectively resolve only entries matching the condition. Uses the same Filter syntax. | | `probe_filter` | String | - | Filter expression to selectively probe only entries matching the condition. Uses the same Filter syntax. | -| `stalker_pre_resolve_playback` | Bool | `false` | Stalker-only: resolves `create_link` during playlist processing and persists the returned playback URL when the portal already grants one. Useful when you want the playlist/export step to materialize playable stream URLs up front. | -| `stalker_runtime_resolve_playback` | Bool | `false` | Stalker-only: lets the reverse-proxy retry `create_link` during playback when a persisted Stalker URL has gone stale or is rejected by the portal. This is the recovery path for temp links and expired session-bound stream URLs. | > **Note:** For `resolve_vod` and `resolve_series`, data is cached per input and only new or changed entries are > updated. @@ -335,18 +330,18 @@ again when the process restarts. Shared playback uses the identity of the shared #### Stalker playback notes * Stalker playlist preview in the Web UI now works through the same protected playlist endpoints used for other input types. -* `stalker_pre_resolve_playback` and `stalker_runtime_resolve_playback` are complementary: - * `stalker_pre_resolve_playback: true` tries to turn portal `cmd` values into concrete playback URLs during refresh. - * `stalker_runtime_resolve_playback: true` retries `create_link` later if the stored playback URL is stale, temp-link based, or rejected after processing. +* Stalker items are imported with their portal metadata and playback descriptor but without a materialized stream URL, so the + exported playlist never exposes the raw portal `cmd`. +* Playback resolves the item on demand: the reverse-proxy path runs `create_link` for the item's descriptor candidates and + falls back to the item's stored `cmd` when no descriptor candidate is usable. * Temp-link variants (`nginx_secure_link`, `flussonic_tmp_link`, `wowza_tmp_link`) are persisted as explicit playback modes and reused during runtime refresh, instead of being flattened into a generic direct-URL path. -* When pre-resolve does not materialize a URL, Tuliprox keeps the Stalker item metadata and playback descriptor but does - not leak the raw `cmd` into the exported playlist URL field. -* If pre-resolve is disabled or the portal refuses to resolve a specific item during refresh, the item can still remain playable - later through runtime resolution, assuming the reverse-proxy path is used and runtime resolve is enabled. * Runtime refresh reuses a cached Stalker client per input configuration and treats the session TTL as a soft re-handshake boundary. - If refresh still cannot resolve a playable URL, Tuliprox invalidates the stale persisted URL instead of continuing to serve it indefinitely. -* Stalker EPG import now also consumes the portal bulk-EPG endpoint during processing when Stalker playback pre-resolve is enabled. + A portal that rejects the session (HTTP 401/403/204/456, or a `code` 44 / 440..449 body) triggers one re-handshake and retry; + a portal that refuses the request again is reported with the portal's reason in the log. +* A playback request only ever reads the published catalog generation. If the input has no published catalog for the configured + portal identity, playback fails with a log line naming that input instead of starting or discarding a refresh. +* Stalker EPG import also consumes the portal bulk-EPG endpoint during processing. * The bulk-EPG path is streamed and batch-persisted to reduce peak memory pressure on large portals, but portal-specific tuning for pathological datasets is still a separate follow-up topic. * Supported Stalker playback transports are currently `http` and `https` only. `rtmp://` / `rtsp://` commands are rejected diff --git a/docs/src/rest-api-cookbook.md b/docs/src/rest-api-cookbook.md index 08ea1d001..712fd4932 100644 --- a/docs/src/rest-api-cookbook.md +++ b/docs/src/rest-api-cookbook.md @@ -284,12 +284,12 @@ Typical use: Notes: - Replace `"stalker"` with the configured input name. -- The preview endpoint returns catalog items. Actual playback can still require Stalker `create_link` resolution later, depending - on your `stalker_pre_resolve_playback` / `stalker_runtime_resolve_playback` settings. -- If a Stalker item has not been materialized yet, Tuliprox no longer exposes the raw portal `cmd` as the playlist URL. Runtime +- The preview endpoint returns catalog items. Actual playback resolves the Stalker `create_link` URL when the request reaches + the stream endpoint. +- If a Stalker item has not been materialized yet, Tuliprox does not expose the raw portal `cmd` as the playlist URL. Runtime playback resolves a real media URL later through the reverse-proxy path. -- Expired temp links are refreshed automatically when `stalker_runtime_resolve_playback` is enabled. Only portal-specific extra - header/cookie requirements remain a possible follow-up. +- Expired temp links are refreshed automatically, including one re-handshake when the portal rejects the session. Only + portal-specific extra header/cookie requirements remain a possible follow-up. - Runtime refresh only produces `http`/`https` playback URLs; Stalker `rtmp://` / `rtsp://` commands are rejected explicitly rather than proxied half-supported. diff --git a/frontend/public/config.json b/frontend/public/config.json index 5b6fa1eff..06ccab527 100644 --- a/frontend/public/config.json +++ b/frontend/public/config.json @@ -7,6 +7,7 @@ "authUrl": "/auth/" }, "discord": "https://discord.gg/gkzCmWw9Tf", + "releases": "https://github.com/euzu/tuliprox/pkgs/container/tuliprox-alpine", "documentation": "static/docs/", "wsUrl" : "/ws", "protocolVersion": 1 diff --git a/test/fixtures/testkit/scenarios/stalker-live-resolution.yml b/test/fixtures/testkit/scenarios/stalker-live-resolution.yml new file mode 100644 index 000000000..190a750f4 --- /dev/null +++ b/test/fixtures/testkit/scenarios/stalker-live-resolution.yml @@ -0,0 +1,61 @@ +schema_version: 1 +name: stalker-live-resolution +# A Stalker/Ministra input resolves every playback through the portal's own `create_link`. +# +# The emulated portal answers with the fixture's stream URL, which lives on loopback. The SUT +# refuses that destination on purpose — a portal-supplied URL must never make the proxy talk to +# a private address — so this scenario asserts the resolution chain, not the media: the portal +# has to see one `create_link` per requested channel, carrying its descriptor's `cmd_1`. +# The raw `cmd` resolves the same marker, so the source check catches fallback regressions. +tuliprox: + base_url: http://fixture.invalid + execution_mode: isolated_fixture + input_type: stalker + bootstrap: + command: "${env:TULIPROX_TESTKIT_SUT_BINARY}" + arguments: ["-s", "-c", "{config_file}", "-i", "{source_file}", "-a", "{api_proxy_file}"] + readiness_url: "{base_url}/healthcheck" + stalker: + mac_address: '00:1A:79:00:00:01' + mag_preset: generic_safe + separate_descriptor_command: true +origin: + run_id: stalker-live-resolution +actors: + - id: tv-a + agent: local + username: testkit + user_agent: tuliprox-testkit/tv-a +channels: + live-17: { origin_marker: 17, protocol: xtream_ts } + live-19: { origin_marker: 19, protocol: xtream_ts } +steps: + - command_id: resolve-17 + start: + actor: tv-a + playback_id: playback-17 + session_group: playback-17 + channel: live-17 + # The non-public create_link destination is refused; the m3u endpoint answers 400 for it. + expect: + http_error: 400 + assert_origin: + stalker_create_links_at_least: 1 + stalker_token_refusals_at_least: 0 + stalker_create_link_markers: [17] + stalker_descriptor_markers: [17] + stalker_raw_command_markers: [] + - command_id: resolve-19 + start: + actor: tv-a + playback_id: playback-19 + session_group: playback-19 + channel: live-19 + expect: + http_error: 400 + assert_origin: + stalker_create_links_at_least: 2 + stalker_token_refusals_at_least: 0 + stalker_create_link_markers: [17, 19] + stalker_descriptor_markers: [17, 19] + stalker_raw_command_markers: [] diff --git a/test/fixtures/testkit/scenarios/stalker-refused-channel-keeps-catalog.yml b/test/fixtures/testkit/scenarios/stalker-refused-channel-keeps-catalog.yml new file mode 100644 index 000000000..554484dd0 --- /dev/null +++ b/test/fixtures/testkit/scenarios/stalker-refused-channel-keeps-catalog.yml @@ -0,0 +1,52 @@ +schema_version: 1 +name: stalker-refused-channel-keeps-catalog +# One channel's `create_link` is refused every time, the other resolves. The failing playback +# must be reported (the SUT answers 502 for an unresolvable item) without damaging the +# published catalog: the second channel still resolves through its own stored `cmd`, and the +# portal never answered a stream URL for the refused marker. +tuliprox: + base_url: http://fixture.invalid + execution_mode: isolated_fixture + input_type: stalker + bootstrap: + command: "${env:TULIPROX_TESTKIT_SUT_BINARY}" + arguments: ["-s", "-c", "{config_file}", "-i", "{source_file}", "-a", "{api_proxy_file}"] + readiness_url: "{base_url}/healthcheck" + stalker: + mac_address: '00:1A:79:00:00:01' + mag_preset: generic_safe + refuse_create_link_markers: [19] +origin: + run_id: stalker-refused-channel-keeps-catalog +actors: + - id: tv-a + agent: local + username: testkit + user_agent: tuliprox-testkit/tv-a +channels: + live-17: { origin_marker: 17, protocol: xtream_ts } + live-19: { origin_marker: 19, protocol: xtream_ts } +steps: + - command_id: refused-19 + start: + actor: tv-a + playback_id: playback-19 + session_group: playback-19 + channel: live-19 + # No stream URL for this item: the m3u endpoint reports the resolution failure. + expect: + http_error: 502 + assert_origin: + stalker_token_refusals_at_least: 2 + stalker_create_links_at_least: 2 + stalker_create_link_markers: [] + - command_id: healthy-17 + start: + actor: tv-a + playback_id: playback-17 + session_group: playback-17 + channel: live-17 + expect: + http_error: 400 + assert_origin: + stalker_create_link_markers: [17] diff --git a/test/fixtures/testkit/scenarios/stalker-session-refused-once.yml b/test/fixtures/testkit/scenarios/stalker-session-refused-once.yml new file mode 100644 index 000000000..bf5e6bf15 --- /dev/null +++ b/test/fixtures/testkit/scenarios/stalker-session-refused-once.yml @@ -0,0 +1,46 @@ +schema_version: 1 +name: stalker-session-refused-once +# A portal that invalidates the session out of band answers the first `create_link` with a +# stale-session body (`code` 44 inside a `200 OK`). The playback path has to drop the cached +# session, handshake again and retry, instead of reporting the item as unresolvable. +# +# The portal counters are the assertion: without the retry the fixture sees one `create_link` +# that resolved nothing and no second handshake. The client-visible status is the destination +# guard refusing the fixture's loopback stream URL, which is unrelated to the session and stays +# 400 either way — the recovered resolution is what the portal statistics prove. +tuliprox: + base_url: http://fixture.invalid + execution_mode: isolated_fixture + input_type: stalker + bootstrap: + command: "${env:TULIPROX_TESTKIT_SUT_BINARY}" + arguments: ["-s", "-c", "{config_file}", "-i", "{source_file}", "-a", "{api_proxy_file}"] + readiness_url: "{base_url}/healthcheck" + stalker: + mac_address: '00:1A:79:00:00:01' + mag_preset: ministra_modern + refuse_create_link_once: true +origin: + run_id: stalker-session-refused-once +actors: + - id: tv-a + agent: local + username: testkit + user_agent: tuliprox-testkit/tv-a +channels: + live-17: { origin_marker: 17, protocol: xtream_ts } +steps: + - command_id: recover-17 + start: + actor: tv-a + playback_id: playback-17 + session_group: playback-17 + channel: live-17 + expect: + http_error: 400 + assert_origin: + # One refusal, then a re-handshake and a retried resolution that the portal answered. + stalker_token_refusals_at_least: 1 + stalker_handshakes_at_least: 1 + stalker_create_links_at_least: 2 + stalker_create_link_markers: [17] diff --git a/tools/tuliprox-testkit/src/bootstrap.rs b/tools/tuliprox-testkit/src/bootstrap.rs index cd5f437fa..28f074c34 100644 --- a/tools/tuliprox-testkit/src/bootstrap.rs +++ b/tools/tuliprox-testkit/src/bootstrap.rs @@ -1,5 +1,5 @@ use crate::{ - config::{BootstrapConfig, Channel, PolicyContract}, + config::{BootstrapConfig, Channel, FixtureInputType, FixtureStalker, PolicyContract}, TestkitError, }; use std::{ @@ -14,6 +14,17 @@ use tokio::process::{Child, Command}; const FIXTURE_PASSWORD: &str = "testkit-password"; +/// The provider protocol the generated fixture input uses, plus the portal settings the +/// emulated Stalker portal needs. +pub struct FixtureInputPlan<'a> { + pub input_type: FixtureInputType, + pub stalker: &'a FixtureStalker, +} + +impl FixtureInputPlan<'_> { + const fn is_stalker(&self) -> bool { matches!(self.input_type, FixtureInputType::Stalker) } +} + /// Files owned by one isolated run. They are written below a unique temporary /// directory and must never refer to an operator's configuration or storage. pub struct FixturePaths { @@ -26,6 +37,7 @@ pub struct FixturePaths { } impl FixturePaths { + #[allow(clippy::too_many_arguments)] pub fn create( run_id: &str, origin_address: SocketAddr, @@ -34,6 +46,7 @@ impl FixturePaths { channels: &HashMap, fixture_stream: &crate::config::FixtureStreamOptions, add_xtream_output: bool, + input: &FixtureInputPlan<'_>, ) -> Result { let directory = tempfile::Builder::new().prefix("tuliprox-testkit-").tempdir()?; let root = directory.path().to_path_buf(); @@ -55,7 +68,7 @@ impl FixturePaths { )?; std::fs::write( &source_file, - render_sources(run_id, origin_address, channels, policy, fixture_stream, add_xtream_output), + render_sources(run_id, origin_address, channels, policy, fixture_stream, add_xtream_output, input), )?; std::fs::write(&api_proxy_file, render_api_proxy(tuliprox_address, policy))?; @@ -118,6 +131,7 @@ fn render_sources( policy: Option<&PolicyContract>, fixture_stream: &crate::config::FixtureStreamOptions, add_xtream_output: bool, + input: &FixtureInputPlan<'_>, ) -> String { let mut markers = channels.values().map(|channel| channel.origin_marker).collect::>(); markers.sort_unstable(); @@ -126,7 +140,7 @@ fn render_sources( markers.push(17); } let marker_comment = markers.iter().map(u32::to_string).collect::>().join(","); - let provider_config = render_provider_pool(run_id, origin_address, policy); + let provider_config = render_provider_input(run_id, origin_address, policy, input); let share_hls = fixture_stream.share_live_hls; let share_mpeg_ts = fixture_stream.share_live_mpeg_ts; let extra_output = if add_xtream_output { " - type: xtream\n" } else { "" }; @@ -135,6 +149,31 @@ fn render_sources( ) } +fn render_provider_input( + run_id: &str, + origin_address: SocketAddr, + policy: Option<&PolicyContract>, + input: &FixtureInputPlan<'_>, +) -> String { + if input.is_stalker() { + return render_stalker_input(origin_address, input.stalker); + } + render_provider_pool(run_id, origin_address, policy) +} + +/// The fixture input for the emulated Stalker portal. +/// +/// The portal URL keeps the `/stalker_portal/c/` web-UI path a real Ministra install uses; +/// the SUT derives its load endpoints from it. `endpoint_preference: server_load` pins the +/// endpoint the fixture answers on, so a scenario never depends on the fallback order. +fn render_stalker_input(origin_address: SocketAddr, stalker: &FixtureStalker) -> String { + format!( + " - name: testkit-origin\n type: stalker\n url: 'http://{origin_address}/stalker_portal/c/'\n stalker:\n auth_mode: mac_only\n mag_preset: {}\n endpoint_preference: server_load\n device:\n mac_address: {}\n", + yaml_string(&stalker.mag_preset), + yaml_string(&stalker.mac_address) + ) +} + fn render_provider_pool(run_id: &str, origin_address: SocketAddr, policy: Option<&PolicyContract>) -> String { let accounts = policy.map(|contract| contract.provider_pool.as_slice()).unwrap_or_default(); if let Some((root, aliases)) = accounts.split_first() { @@ -219,6 +258,7 @@ impl IsolatedFixture { channels: &HashMap, fixture_stream: &crate::config::FixtureStreamOptions, add_xtream_output: bool, + input: &FixtureInputPlan<'_>, ) -> Result { const MAX_START_ATTEMPTS: usize = 5; let mut last_error = None; @@ -235,6 +275,7 @@ impl IsolatedFixture { channels, fixture_stream, add_xtream_output, + input, ) .await { @@ -266,6 +307,7 @@ impl IsolatedFixture { channels: &HashMap, fixture_stream: &crate::config::FixtureStreamOptions, add_xtream_output: bool, + input: &FixtureInputPlan<'_>, ) -> Result { let origin_address = reserve_loopback_address()?; let origin_control_address = reserve_loopback_address()?; @@ -278,6 +320,7 @@ impl IsolatedFixture { channels, fixture_stream, add_xtream_output, + input, )?; let markers = fixture_markers(channels); let mut origin_cmd = Command::new(testkit_binary); @@ -304,6 +347,17 @@ impl IsolatedFixture { origin_cmd.arg("--limit-mode").arg(mode_str); } } + if input.stalker.refuse_create_link_once { + origin_cmd.arg("--stalker-refuse-create-link-once"); + } + if input.stalker.separate_descriptor_command { + origin_cmd.arg("--stalker-separate-descriptor-command"); + } + if !input.stalker.refuse_create_link_markers.is_empty() { + let markers = + input.stalker.refuse_create_link_markers.iter().map(u32::to_string).collect::>().join(","); + origin_cmd.arg("--stalker-refuse-create-link-markers").arg(markers); + } let mut origin = origin_cmd.current_dir(working_directory).spawn()?; let origin_health = format!("http://{origin_control_address}/health"); if let Err(error) = wait_for_http_ready(&mut origin, &origin_health, Duration::from_secs(10)).await { @@ -507,6 +561,10 @@ mod tests { &channels, &crate::config::FixtureStreamOptions::default(), false, + &FixtureInputPlan { + input_type: crate::config::FixtureInputType::M3u, + stalker: &crate::config::FixtureStalker::default(), + }, )?; let config = std::fs::read_to_string(&fixture.config_file)?; let source = std::fs::read_to_string(&fixture.source_file)?; @@ -537,10 +595,45 @@ mod tests { Ok(()) } + fn m3u_input() -> FixtureInputPlan<'static> { + FixtureInputPlan { + input_type: crate::config::FixtureInputType::M3u, + stalker: Box::leak(Box::new(crate::config::FixtureStalker::default())), + } + } + + #[test] + fn stalker_input_renders_a_portal_url_and_device_identity() { + let stalker = crate::config::FixtureStalker { + mac_address: "00:1A:79:00:00:09".to_owned(), + mag_preset: "ministra_modern".to_owned(), + refuse_create_link_once: true, + refuse_create_link_markers: vec![19], + separate_descriptor_command: false, + }; + let plan = FixtureInputPlan { input_type: crate::config::FixtureInputType::Stalker, stalker: &stalker }; + let source = render_sources( + "run", + "127.0.0.1:9910".parse().unwrap(), + &HashMap::new(), + None, + &crate::config::FixtureStreamOptions::default(), + false, + &plan, + ); + + assert!(source.contains("type: stalker"), "{source}"); + assert!(source.contains("http://127.0.0.1:9910/stalker_portal/c/"), "{source}"); + assert!(source.contains("mac_address: '00:1A:79:00:00:09'"), "{source}"); + assert!(source.contains("mag_preset: 'ministra_modern'"), "{source}"); + assert!(!source.contains("catalog/input.m3u"), "{source}"); + } + #[test] fn render_sources_with_mpeg_ts_sharing_disabled() { let opts = crate::config::FixtureStreamOptions { share_live_hls: true, share_live_mpeg_ts: false }; - let source = render_sources("run", "127.0.0.1:9910".parse().unwrap(), &HashMap::new(), None, &opts, false); + let source = + render_sources("run", "127.0.0.1:9910".parse().unwrap(), &HashMap::new(), None, &opts, false, &m3u_input()); assert!(source.contains("mpeg_ts: false")); assert!(source.contains("hls: true")); assert!(!source.contains("type: xtream")); @@ -549,14 +642,16 @@ mod tests { #[test] fn render_sources_includes_xtream_output_when_requested() { let opts = crate::config::FixtureStreamOptions::default(); - let source = render_sources("run", "127.0.0.1:9910".parse().unwrap(), &HashMap::new(), None, &opts, true); + let source = + render_sources("run", "127.0.0.1:9910".parse().unwrap(), &HashMap::new(), None, &opts, true, &m3u_input()); assert!(source.contains("type: xtream")); } #[test] fn render_sources_omits_xtream_output_by_default() { let opts = crate::config::FixtureStreamOptions::default(); - let source = render_sources("run", "127.0.0.1:9910".parse().unwrap(), &HashMap::new(), None, &opts, false); + let source = + render_sources("run", "127.0.0.1:9910".parse().unwrap(), &HashMap::new(), None, &opts, false, &m3u_input()); assert!(!source.contains("type: xtream")); assert!(source.contains("mpeg_ts: true")); } diff --git a/tools/tuliprox-testkit/src/config.rs b/tools/tuliprox-testkit/src/config.rs index 29bb299a5..25398fd83 100644 --- a/tools/tuliprox-testkit/src/config.rs +++ b/tools/tuliprox-testkit/src/config.rs @@ -105,6 +105,53 @@ impl PolicyContract { } } +/// Which provider protocol the isolated fixture imports from. +/// +/// `m3u` (default) keeps the generated playlist input. `stalker` points the input at the +/// Ministra portal the fixture origin emulates, so the import, the runtime `create_link` +/// resolution and the destination guard can be exercised against a scripted portal. +#[derive(Debug, Clone, Copy, Default, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum FixtureInputType { + #[default] + M3u, + Stalker, +} + +/// The emulated portal's identity and its scripted refusals. +#[derive(Debug, Clone, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct FixtureStalker { + /// MAC the fixture input authenticates with. The emulated portal accepts any MAC. + pub mac_address: String, + /// MAG preset written into the fixture input. Must be one of the four presets. + pub mag_preset: String, + /// Refuse the first `create_link` with a stale-session body (`code` 44) and answer the + /// retry. A portal that invalidates the session out of band behaves this way; the + /// playback path has to re-handshake instead of reporting the item as unresolvable. + pub refuse_create_link_once: bool, + /// Channels whose `create_link` is always refused with the same stale-session body. + pub refuse_create_link_markers: Vec, + /// Serve a distinct `cmd_1` descriptor command alongside the raw `cmd`. + pub separate_descriptor_command: bool, +} + +impl FixtureStalker { + pub const PRESETS: [&'static str; 4] = ["generic_safe", "mag250_legacy", "mag254_strict", "ministra_modern"]; +} + +impl Default for FixtureStalker { + fn default() -> Self { + Self { + mac_address: "00:1A:79:00:00:01".to_owned(), + mag_preset: "generic_safe".to_owned(), + refuse_create_link_once: false, + refuse_create_link_markers: Vec::new(), + separate_descriptor_command: false, + } + } +} + #[derive(Debug, Clone, Deserialize)] #[serde(deny_unknown_fields)] pub struct TuliproxConfig { @@ -112,6 +159,10 @@ pub struct TuliproxConfig { #[serde(default)] pub api_base_url: Option, #[serde(default)] + pub input_type: FixtureInputType, + #[serde(default)] + pub stalker: FixtureStalker, + #[serde(default)] pub admin_username: Option, #[serde(default)] pub admin_password_env: Option, @@ -252,6 +303,11 @@ pub enum ExpectedPlayback { /// A recently evicted playback retried and the SUT terminated it quietly /// (HTTP 204, no custom error video, no connection-denied event). Suppressed, + /// The SUT answered with a plain HTTP failure instead of a stream — the SUT's own + /// rejection statuses are covered by [`Self::Rejected`]. Used where the stream cannot + /// be served for a reason the SUT reports as an error, such as a Stalker `create_link` + /// target the destination guard refuses. + HttpError(u16), } impl ExpectedPlayback { @@ -259,7 +315,9 @@ impl ExpectedPlayback { pub fn is_streaming(&self) -> bool { matches!(self, Self::Streaming) } #[must_use] - pub fn is_rejected(&self) -> bool { matches!(self, Self::Rejected | Self::RejectedWith(_) | Self::Suppressed) } + pub fn is_rejected(&self) -> bool { + matches!(self, Self::Rejected | Self::RejectedWith(_) | Self::Suppressed | Self::HttpError(_)) + } #[must_use] pub fn matches_outcome(&self, outcome: &crate::protocol::PlaybackOutcome) -> bool { @@ -272,6 +330,7 @@ impl ExpectedPlayback { | (Self::Suppressed, PlaybackOutcome::AdmissionRejected { reason: RejectionReason::HttpStatus(204) }) => { true } + (Self::HttpError(expected), PlaybackOutcome::HttpError { status }) => expected == status, (Self::RejectedWith(expected), PlaybackOutcome::AdmissionRejected { reason }) => match reason { RejectionReason::CustomVideo(kind) => { expected.custom_video.is_none_or(|expected_kind| expected_kind == *kind) @@ -308,6 +367,25 @@ pub struct AssertOrigin { pub no_limit_rejections: Option, #[serde(default)] pub latest_request_account: Option, + /// Portal handshakes the emulated Stalker portal answered at least. + #[serde(default)] + pub stalker_handshakes_at_least: Option, + /// `create_link` requests the portal received at least, refusals included. + #[serde(default)] + pub stalker_create_links_at_least: Option, + /// `create_link` requests the portal refused with a stale-session body, at least. + #[serde(default)] + pub stalker_token_refusals_at_least: Option, + /// Markers the portal answered with a stream URL, compared as a set. Proves that a + /// playback resolved the requested channel's own stored command. + #[serde(default)] + pub stalker_create_link_markers: Option>, + /// Markers resolved through the portal's `cmd_1` descriptor command. + #[serde(default)] + pub stalker_descriptor_markers: Option>, + /// Markers resolved through the raw `cmd` fallback command. + #[serde(default)] + pub stalker_raw_command_markers: Option>, } /// Per-step assertions against the SUT runtime status (`/api/v1/status`). @@ -501,6 +579,7 @@ impl Scenario { } self.validate_policy_contract()?; self.validate_playback_endpoint()?; + self.validate_stalker_fixture()?; Ok(()) } @@ -530,6 +609,49 @@ impl Scenario { Ok(()) } + /// The emulated Stalker portal only serves what the fixture can answer for: the + /// channels the scenario declares, the presets the SUT knows, and no multi-account + /// pool (Stalker aliases carry their own portal identities, which the fixture does + /// not model). + fn validate_stalker_fixture(&self) -> Result<(), TestkitError> { + if self.tuliprox.input_type != FixtureInputType::Stalker { + if self.tuliprox.stalker.refuse_create_link_once + || !self.tuliprox.stalker.refuse_create_link_markers.is_empty() + { + return Err(TestkitError::Configuration( + "stalker portal refusals require input_type: stalker".to_owned(), + )); + } + return Ok(()); + } + if !FixtureStalker::PRESETS.contains(&self.tuliprox.stalker.mag_preset.as_str()) { + return Err(TestkitError::Configuration(format!( + "unknown mag_preset '{}'; expected one of {:?}", + self.tuliprox.stalker.mag_preset, + FixtureStalker::PRESETS + ))); + } + if self.tuliprox.stalker.mac_address.trim().is_empty() { + return Err(TestkitError::Configuration("stalker.mac_address cannot be empty".to_owned())); + } + if self.policy_contract.as_ref().is_some_and(|contract| !contract.provider_pool.is_empty()) { + return Err(TestkitError::Configuration( + "a provider_pool is not supported for input_type: stalker (the fixture portal has one identity)" + .to_owned(), + )); + } + let markers = + self.channels.values().map(|channel| channel.origin_marker).collect::>(); + for marker in &self.tuliprox.stalker.refuse_create_link_markers { + if !markers.contains(marker) { + return Err(TestkitError::Configuration(format!( + "stalker.refuse_create_link_markers names marker {marker}, which no channel of this scenario uses" + ))); + } + } + Ok(()) + } + pub fn expanded_steps(&self) -> Result, TestkitError> { let mut expanded = Vec::new(); expand_steps(&self.steps, "", &mut expanded)?; @@ -650,6 +772,8 @@ steps: execution_mode: ExecutionMode::IsolatedFixture, playback_endpoint: PlaybackEndpoint::default(), fixture_stream: FixtureStreamOptions::default(), + input_type: FixtureInputType::default(), + stalker: FixtureStalker::default(), }, origin: None, actors: vec![Actor { @@ -757,6 +881,70 @@ steps: assert_eq!(expanded[3].stop.as_ref().map(|stop| stop.playback_id.as_str()), Some("repeat:1:playback")); } + #[test] + fn http_error_expectation_matches_only_the_declared_status() { + use crate::protocol::PlaybackOutcome; + + let expected = ExpectedPlayback::HttpError(502); + assert!(expected.matches_outcome(&PlaybackOutcome::HttpError { status: 502 })); + assert!(!expected.matches_outcome(&PlaybackOutcome::HttpError { status: 400 })); + assert!(!expected.matches_outcome(&PlaybackOutcome::Streaming { frames: 1, bytes: 1 })); + // A status the SUT itself reports as a rejection is a different outcome class. + assert!(!expected.matches_outcome(&PlaybackOutcome::AdmissionRejected { + reason: crate::protocol::RejectionReason::HttpStatus(502) + })); + assert!(expected.is_rejected(), "an error step must not await frames"); + } + + #[test] + fn stalker_fixture_validation_rejects_unknown_presets_and_markers() -> Result<(), TestkitError> { + let mut scenario = Scenario { + schema_version: 1, + name: "stalker".to_owned(), + tuliprox: TuliproxConfig { + base_url: "http://example.invalid".to_owned(), + api_base_url: None, + admin_username: None, + admin_password_env: None, + playlist_url: None, + bootstrap: Some(BootstrapConfig { + command: "true".to_owned(), + arguments: Vec::new(), + readiness_url: "http://example.invalid/healthcheck".to_owned(), + readiness_timeout_millis: 1000, + }), + execution_mode: ExecutionMode::IsolatedFixture, + playback_endpoint: PlaybackEndpoint::M3u, + fixture_stream: FixtureStreamOptions::default(), + input_type: FixtureInputType::Stalker, + stalker: FixtureStalker::default(), + }, + origin: None, + actors: Vec::new(), + channels: HashMap::from([( + "live-17".to_owned(), + Channel { origin_marker: 17, protocol: "xtream_ts".to_owned() }, + )]), + agents: AgentsConfig::default(), + policy_contract: None, + steps: Vec::new(), + }; + assert!(scenario.validate().is_ok()); + + scenario.tuliprox.stalker.mag_preset = "mag999".to_owned(); + assert!(scenario.validate().is_err(), "an unknown MAG preset must be rejected"); + + scenario.tuliprox.stalker.mag_preset = "generic_safe".to_owned(); + scenario.tuliprox.stalker.refuse_create_link_markers = vec![42]; + assert!(scenario.validate().is_err(), "a marker no channel uses must be rejected"); + + scenario.tuliprox.input_type = FixtureInputType::M3u; + scenario.tuliprox.stalker.refuse_create_link_markers.clear(); + scenario.tuliprox.stalker.refuse_create_link_once = true; + assert!(scenario.validate().is_err(), "portal refusals require a stalker input"); + Ok(()) + } + #[test] fn rejected_playback_does_not_match_500_corrupt_data_or_early_eof() { use crate::{ diff --git a/tools/tuliprox-testkit/src/main.rs b/tools/tuliprox-testkit/src/main.rs index dd12cd401..9efda1483 100644 --- a/tools/tuliprox-testkit/src/main.rs +++ b/tools/tuliprox-testkit/src/main.rs @@ -11,6 +11,7 @@ use bytes::BytesMut; use clap::{Parser, Subcommand}; use futures::StreamExt; use serde::Serialize; +use serde_json::json; use std::{ collections::{BTreeMap, HashMap, HashSet}, fmt::Write, @@ -38,7 +39,7 @@ use tuliprox_testkit::{ hls::read_segments, observation::TuliproxObserver, oracle::{AdmissionOracle, Decision as AdmissionDecision, Request as AdmissionRequest}, - origin_events::{BodyCloseReason, OriginEvent, OriginEventKind, OriginStats}, + origin_events::{BodyCloseReason, OriginEvent, OriginEventKind, OriginStats, StalkerPortalStats}, origin_transport::{ serve_tracked_media_listener, OriginConnectionMeta, OriginLimitMode, OriginPolicy, OriginTracker, }, @@ -119,6 +120,14 @@ enum Command { account_limit: Option, #[arg(long, default_value = "observe_only")] limit_mode: String, + /// Refuse the first `create_link` the emulated Stalker portal receives. + #[arg(long)] + stalker_refuse_create_link_once: bool, + /// Markers whose `create_link` the emulated Stalker portal always refuses. + #[arg(long, value_delimiter = ',')] + stalker_refuse_create_link_markers: Vec, + #[arg(long)] + stalker_separate_descriptor_command: bool, }, /// Execute a local, sequential scenario against Tuliprox URLs. Controller { @@ -159,9 +168,103 @@ struct OriginState { observations: Arc>>, faults: Arc>, hls_sequences: Arc>>, + stalker: Arc>, tracker: OriginTracker, } +/// The emulated Ministra portal: its identity-independent script plus what it was asked for. +/// +/// The fixture portal accepts any MAC (the input's device identity is not validated), offers +/// every marker of the run as a live channel, and answers `create_link` with the fixture's own +/// stream URL — a loopback destination, which the SUT's Stalker destination guard then refuses. +/// That refusal is part of the contract under test, not an accident: a portal-supplied URL must +/// never make the proxy talk to a private destination. +struct StalkerPortal { + stats: StalkerPortalStats, + refuse_once: bool, + refuse_once_armed: bool, + /// Set when a configured refusal makes the portal treat the SUT's current session as + /// stale: every `create_link` is refused until a new handshake arrives, which is how a + /// portal that invalidated the session out of band behaves. A refusal that only hit one + /// endpoint candidate would be recovered by the endpoint fallback instead of exercising + /// the session recovery this fixture exists for. + stale_session: bool, + refuse_markers: Vec, + command_layout: StalkerCommandLayout, +} + +#[derive(Clone, Copy)] +enum StalkerCommandLayout { + Shared, + Separate, +} + +impl StalkerPortal { + fn new(refuse_once: bool, refuse_markers: Vec, separate_descriptor_command: bool) -> Self { + Self { + stats: StalkerPortalStats::default(), + refuse_once, + refuse_once_armed: refuse_once, + stale_session: false, + refuse_markers, + command_layout: if separate_descriptor_command { + StalkerCommandLayout::Separate + } else { + StalkerCommandLayout::Shared + }, + } + } + + fn reset(&mut self) { + self.stats = StalkerPortalStats::default(); + self.refuse_once_armed = self.refuse_once; + self.stale_session = false; + } + + /// A handshake replaces the session, so a stale one becomes usable again. + fn note_handshake(&mut self) { + self.stats.handshakes += 1; + self.stale_session = false; + } + + /// Whether this `create_link` is answered with a stale-session body instead of a URL. + /// + /// A marker listed in `refuse_create_link_markers` is refused every time; the + /// `refuse_create_link_once` switch makes the session stale for the first resolution that + /// would otherwise succeed, so a scenario can let one channel fail for good and still test + /// the session recovery. + fn refuses(&mut self, marker: u32) -> bool { + let always = self.refuse_markers.contains(&marker); + if !always && self.refuse_once_armed { + self.refuse_once_armed = false; + self.stale_session = true; + } + if !always && !self.stale_session { + return false; + } + self.stats.token_refusals += 1; + true + } + + fn record_resolution(&mut self, marker: u32, command_source: Option<&str>) { + if !self.stats.resolved_markers.contains(&marker) { + self.stats.resolved_markers.push(marker); + self.stats.resolved_markers.sort_unstable(); + } + let markers = match command_source { + Some("descriptor") => Some(&mut self.stats.resolved_descriptor_markers), + Some("raw") => Some(&mut self.stats.resolved_raw_command_markers), + _ => None, + }; + if let Some(markers) = markers { + if !markers.contains(&marker) { + markers.push(marker); + markers.sort_unstable(); + } + } + } +} + #[derive(Clone, Serialize)] struct OriginObservation { origin_stream_id: String, @@ -191,7 +294,18 @@ async fn main() -> ExitCode { async fn run(cli: Cli) -> Result { match cli.command { - Command::Origin { listen, control_listen, run_id, bitrate, mut markers, account_limit, limit_mode } => { + Command::Origin { + listen, + control_listen, + run_id, + bitrate, + mut markers, + account_limit, + limit_mode, + stalker_refuse_create_link_once, + stalker_refuse_create_link_markers, + stalker_separate_descriptor_command, + } => { markers.sort_unstable(); markers.dedup(); if markers.is_empty() { @@ -215,6 +329,11 @@ async fn run(cli: Cli) -> Result { observations: Arc::new(Mutex::new(Vec::new())), faults: Arc::new(Mutex::new(FaultSchedule::default())), hls_sequences: Arc::new(Mutex::new(HashMap::new())), + stalker: Arc::new(Mutex::new(StalkerPortal::new( + stalker_refuse_create_link_once, + stalker_refuse_create_link_markers, + stalker_separate_descriptor_command, + ))), tracker, }, ) @@ -559,6 +678,9 @@ async fn serve_origin( let media_listener = TcpListener::bind(media_listen).await?; let media_app = Router::new() .route("/catalog/input.m3u", get(catalog)) + .route("/stalker_portal/server/load.php", get(stalker_load)) + .route("/stalker_portal/portal.php", get(stalker_load)) + .route("/stalker_portal/c/", get(stalker_load)) .route("/live/{*stream}", get(live)) .route("/hls/{*resource}", get(hls)) .route("/vod/{*object}", get(vod).head(vod_head)) @@ -571,6 +693,7 @@ async fn serve_origin( .route("/v1/runs/{run_id}/connections", get(origin_connections)) .route("/v1/runs/{run_id}/events", get(origin_events)) .route("/v1/runs/{run_id}/stats", get(origin_stats)) + .route("/v1/runs/{run_id}/stalker", get(stalker_stats)) .route("/v1/runs/{run_id}/faults/{fault_id}", put(set_fault).delete(clear_fault)) .with_state(state.clone()); @@ -604,6 +727,100 @@ fn origin_account(query: &HashMap) -> Option<&str> { query.get("account").or_else(|| query.get("token")).map(String::as_str) } +/// The emulated Ministra/Stalker portal endpoint. +/// +/// Answers the actions the SUT's catalog import and playback resolution use: the STB +/// handshake/profile pair, the live genres and ordered list, and `create_link`. Everything +/// else gets an empty `js` object, which is what a portal answers for an action it does not +/// implement. The portal is deliberately stateless apart from its request counters — the +/// SUT is expected to carry the session itself. +async fn stalker_load( + State(state): State, + Query(query): Query>, + headers: HeaderMap, +) -> Response { + let action = query.get("action").map_or("", String::as_str); + let portal_type = query.get("type").map_or("", String::as_str); + let host = headers.get(header::HOST).and_then(|value| value.to_str().ok()).unwrap_or("127.0.0.1"); + match (portal_type, action) { + ("stb", "handshake") => { + let mut portal = state.stalker.lock().await; + portal.note_handshake(); + let token = format!("testkit-token-{}", portal.stats.handshakes); + drop(portal); + Json(json!({ "js": { "token": token, "random": "testkit" } })).into_response() + } + ("stb", "get_profile") => { + Json(json!({ "js": { "id": "testkit", "status": 1, "max_connections": 8, "phone": "0000000000" } })) + .into_response() + } + ("itv", "get_genres") => Json(json!({ "js": [{ "id": "1", "title": "Testkit Live" }] })).into_response(), + // The fixture portal serves live channels only: VOD and series answer an empty, valid + // collection so the catalog refresh completes those clusters instead of pausing on a + // response it cannot parse. + ("vod" | "series", "get_genres") => Json(json!({ "js": [] })).into_response(), + ("vod" | "series", "get_ordered_list") => Json(json!({ + "js": { "total_items": 0, "max_page_items": 0, "data": [] } + })) + .into_response(), + ("itv", "get_ordered_list") => { + let separate_descriptor_command = + matches!(state.stalker.lock().await.command_layout, StalkerCommandLayout::Separate); + let channels = state + .markers + .iter() + .map(|marker| { + let mut channel = json!({ + "id": marker.to_string(), + "name": format!("Test channel {marker}"), + "number": marker.to_string(), + "category_id": "1", + "epg_channel_id": format!("test-{marker}"), + "cmd": format!("ffmpeg http://{host}/ch/{marker}"), + "logo": format!("http://{host}/logo/{marker}.png"), + "tv_archive": 0, + }); + if separate_descriptor_command { + channel["cmd"] = json!(format!("ffmpeg http://{host}/raw/{marker}")); + channel["cmd_1"] = json!(format!("ffmpeg http://{host}/descriptor/{marker}")); + } + channel + }) + .collect::>(); + let count = channels.len(); + Json(json!({ + "js": { "total_items": count, "max_page_items": count, "data": channels } + })) + .into_response() + } + ("itv", "create_link") => { + let command = query.get("cmd"); + let Some(marker) = command + .and_then(|cmd| cmd.split('?').next()) + .and_then(|cmd| cmd.trim_end_matches('/').rsplit('/').next()) + .and_then(|segment| segment.parse::().ok()) + .filter(|marker| state.markers.contains(marker)) + else { + return Json(json!({ "js": { "error": "unknown cmd" } })).into_response(); + }; + let mut portal = state.stalker.lock().await; + portal.stats.create_links += 1; + if portal.refuses(marker) { + drop(portal); + // A stale session: Ministra reports it inside a `200 OK` body. + return Json(json!({ "code": 44, "text": "Authorization failed" })).into_response(); + } + let command_source = command.and_then(|cmd| cmd.trim_end_matches('/').rsplit('/').nth(1)); + portal.record_resolution(marker, command_source); + drop(portal); + let url = format!("http://{host}/live/{marker}.ts?run={}", state.run_id.0); + let cmd = BASE64.encode(format!("ffmpeg {url}")); + Json(json!({ "js": { "id": marker.to_string(), "cmd": cmd, "error": "" } })).into_response() + } + _ => Json(json!({ "js": {} })).into_response(), + } +} + async fn catalog( State(state): State, Query(query): Query>, @@ -779,6 +996,13 @@ async fn origin_stats(Path(run_id): Path, State(state): State, State(state): State) -> Response { + if run_id != state.run_id.0 { + return StatusCode::NOT_FOUND.into_response(); + } + Json(state.stalker.lock().await.stats.clone()).into_response() +} + async fn origin_instance(State(state): State) -> Json { Json(OriginInstance { run_id: state.run_id.0.clone() }) } @@ -790,6 +1014,7 @@ async fn reset_origin_run(Path(run_id): Path, State(state): State Result { + let url = format!("{}/v1/runs/{run_id}/stalker", self.control_base_url); + let resp = self.client.get(&url).send().await?; + if !resp.status().is_success() { + return Err(TestkitError::Protocol(format!("stalker portal stats error: {}", resp.status()))); + } + Ok(resp.json::().await?) + } + pub async fn events(&self, run_id: &str) -> Result, TestkitError> { let url = format!("{}/v1/runs/{run_id}/events", self.control_base_url); let resp = self.client.get(&url).send().await?; @@ -1704,6 +1938,10 @@ async fn run_controller( &scenario.channels, &scenario.tuliprox.fixture_stream, scenario.tuliprox.playback_endpoint == tuliprox_testkit::config::PlaybackEndpoint::Xtream, + &tuliprox_testkit::bootstrap::FixtureInputPlan { + input_type: scenario.tuliprox.input_type, + stalker: &scenario.tuliprox.stalker, + }, ) .await { @@ -2172,6 +2410,16 @@ async fn check_origin_assertions<'a>( events.push((step_id, "origin_events_unavailable")); } } + check_stalker_assertions( + step_id, + assert_origin, + obs, + origin_run_id, + events, + any_failed, + observations_incomplete, + ) + .await; break; } tokio::time::sleep(Duration::from_millis(25)).await; @@ -2186,6 +2434,76 @@ async fn check_origin_assertions<'a>( } } +/// Verifies what the emulated Stalker portal was asked for. +/// +/// Playback resolution is only observable there: the portal sees one `create_link` per +/// resolution attempt and knows which markers were answered, so a scenario can prove that a +/// playback resolved the requested channel's own stored command — and that a stale-session +/// refusal was followed by a re-handshake and a retry. +async fn check_stalker_assertions<'a>( + step_id: &'a str, + assert_origin: &tuliprox_testkit::config::AssertOrigin, + obs: &OriginObserver, + origin_run_id: &str, + events: &mut Vec<(&'a str, &'static str)>, + any_failed: &mut bool, + observations_incomplete: &mut bool, +) { + if assert_origin.stalker_handshakes_at_least.is_none() + && assert_origin.stalker_create_links_at_least.is_none() + && assert_origin.stalker_token_refusals_at_least.is_none() + && assert_origin.stalker_create_link_markers.is_none() + && assert_origin.stalker_descriptor_markers.is_none() + && assert_origin.stalker_raw_command_markers.is_none() + { + return; + } + let Ok(stats) = obs.stalker_stats(origin_run_id).await else { + *observations_incomplete = true; + events.push((step_id, "stalker_portal_stats_unavailable")); + return; + }; + if assert_origin.stalker_handshakes_at_least.is_some_and(|expected| stats.handshakes < expected) { + *any_failed = true; + events.push((step_id, "stalker_handshakes_below_expected")); + } + if assert_origin.stalker_create_links_at_least.is_some_and(|expected| stats.create_links < expected) { + *any_failed = true; + events.push((step_id, "stalker_create_links_below_expected")); + } + if assert_origin.stalker_token_refusals_at_least.is_some_and(|expected| stats.token_refusals < expected) { + *any_failed = true; + events.push((step_id, "stalker_token_refusals_below_expected")); + } + if let Some(expected) = assert_origin.stalker_create_link_markers.as_ref() { + let mut expected = expected.clone(); + expected.sort_unstable(); + expected.dedup(); + if expected != stats.resolved_markers { + *any_failed = true; + events.push((step_id, "stalker_resolved_markers_mismatch")); + } + } + if let Some(expected) = assert_origin.stalker_descriptor_markers.as_ref() { + let mut expected = expected.clone(); + expected.sort_unstable(); + expected.dedup(); + if expected != stats.resolved_descriptor_markers { + *any_failed = true; + events.push((step_id, "stalker_descriptor_markers_mismatch")); + } + } + if let Some(expected) = assert_origin.stalker_raw_command_markers.as_ref() { + let mut expected = expected.clone(); + expected.sort_unstable(); + expected.dedup(); + if expected != stats.resolved_raw_command_markers { + *any_failed = true; + events.push((step_id, "stalker_raw_command_markers_mismatch")); + } + } +} + /// Verifies the SUT's user-side accounting (`active_users`, `active_user_connections`) /// after a step, polling briefly so an in-flight cleanup does not fail the check. async fn check_runtime_assertions<'a>( @@ -3063,6 +3381,49 @@ mod tests { assert!(history_confirms_kick(&kicked, session_id.unwrap_or_default())); } + #[test] + fn a_stale_session_is_refused_until_a_new_handshake_arrives() { + let mut portal = StalkerPortal::new(true, Vec::new(), false); + portal.note_handshake(); + + assert!(portal.refuses(17), "the first resolution meets the stale session"); + assert!(portal.refuses(17), "every endpoint candidate of that session is refused too"); + + portal.note_handshake(); + assert!(!portal.refuses(17), "a fresh handshake makes the session usable again"); + assert_eq!(portal.stats.token_refusals, 2); + assert_eq!(portal.stats.handshakes, 2); + assert!(portal.stats.resolved_markers.is_empty(), "a refusal is not a resolution"); + + portal.reset(); + assert_eq!(portal.stats.token_refusals, 0); + assert!(portal.refuses(17), "a reset run re-arms the configured refusal"); + } + + #[test] + fn a_listed_channel_is_refused_every_time_without_staling_the_session() { + let mut portal = StalkerPortal::new(false, vec![19], false); + + assert!(portal.refuses(19)); + assert!(portal.refuses(19)); + portal.note_handshake(); + assert!(portal.refuses(19), "the refusal is bound to the channel, not to the session"); + assert!(!portal.refuses(17), "other channels resolve normally"); + assert_eq!(portal.stats.token_refusals, 3); + } + + #[test] + fn a_resolved_marker_is_recorded_once() { + let mut portal = StalkerPortal::new(false, Vec::new(), false); + portal.record_resolution(19, Some("descriptor")); + portal.record_resolution(17, Some("raw")); + portal.record_resolution(19, Some("descriptor")); + portal.record_resolution(19, Some("raw")); + assert_eq!(portal.stats.resolved_markers, vec![17, 19]); + assert_eq!(portal.stats.resolved_descriptor_markers, vec![19]); + assert_eq!(portal.stats.resolved_raw_command_markers, vec![17, 19]); + } + fn test_origin_state(run_id: &str) -> OriginState { OriginState { run_id: RunId::new(run_id), @@ -3072,6 +3433,7 @@ mod tests { observations: Arc::new(Mutex::new(Vec::new())), faults: Arc::new(Mutex::new(FaultSchedule::default())), hls_sequences: Arc::new(Mutex::new(HashMap::from([(17, 9)]))), + stalker: Arc::new(Mutex::new(StalkerPortal::new(false, Vec::new(), false))), tracker: OriginTracker::new(run_id, OriginPolicy::default()), } } diff --git a/tools/tuliprox-testkit/src/origin_events.rs b/tools/tuliprox-testkit/src/origin_events.rs index e60aa9db9..2b4033b2e 100644 --- a/tools/tuliprox-testkit/src/origin_events.rs +++ b/tools/tuliprox-testkit/src/origin_events.rs @@ -87,6 +87,23 @@ pub struct OriginStats { pub total_bytes_emitted: u64, } +/// What the emulated Stalker portal was asked for during a run. +/// +/// Playback resolution is only observable through the portal: a scenario asserts that the +/// requested channel's `create_link` was attempted, that a stale-session refusal was +/// followed by a re-handshake and a retry, and which markers actually resolved. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct StalkerPortalStats { + pub handshakes: u64, + /// Every `create_link` request, refusals included. + pub create_links: u64, + pub token_refusals: u64, + /// Markers the portal answered with a stream URL, sorted and deduplicated. + pub resolved_markers: Vec, + pub resolved_descriptor_markers: Vec, + pub resolved_raw_command_markers: Vec, +} + #[must_use] pub fn redact_path_and_query(raw: &str) -> String { let Some((path, query)) = raw.split_once('?') else {