Fix/policy trusted destinations (#883)

* fixing resource url problem

* fix(resources): classify resource destinations per hop and keep the proxy in the fetch path
A resource URL is classified (public / non-public / local) before it reaches a client, and every
redirect hop is classified and fetched again: non-public hops connect directly, public hops go
through the configured proxy, local hops are refused. Adds the public no-redirect resource client,
a process-wide destination memo, and regression tests for the redirect and Web UI paths.

* fix(resources): classify resource destinations per hop and keep the proxy in the fetch path
A resource URL is classified (public / non-public / local) before it reaches a client, and every
redirect hop is classified and fetched again: non-public hops connect directly, public hops go
through the configured proxy, local hops are refused. Adds the public no-redirect resource client,
a process-wide destination memo, and regression tests for the redirect and Web UI paths.

---------

Co-authored-by: euzu <euzu@proton.me>

backend/processing/src/processor/epg.rs

@@ -952,7 +952,6 @@ mod tests {
                 match_names,
                 config: Box::new(IcsEpgSourceConfig::default()),
             },
-            input_name: None,
         }])
     }
 
@@ -1041,7 +1040,6 @@ mod tests {
             priority: 0,
             logo_override: false,
             kind: PersistedEpgSourceKind::Xmltv,
-            input_name: None,
         }]);
         let mut playlist = FetchedPlaylist {
             input: &input,
@@ -1256,7 +1254,6 @@ mod tests {
                 priority: 0,
                 logo_override: true,
                 kind: PersistedEpgSourceKind::Xmltv,
-                input_name: Some("epg-input".into()),
             }]);
             let mut playlist = FetchedPlaylist {
                 input: &input,
@@ -1269,12 +1266,8 @@ mod tests {
 
             let updated = playlist.items_mut().next().unwrap();
             assert_eq!(updated.header.epg_channel_id.as_deref(), Some("demo.channel"));
-            let logo = shared::model::ResourceLocator::decode(&updated.header.logo).expect("EPG logo locator");
-            let logo_small =
-                shared::model::ResourceLocator::decode(&updated.header.logo_small).expect("EPG small logo locator");
-            assert_eq!(logo.input_name.as_ref(), "epg-input");
-            assert_eq!(logo.url.as_ref(), "http://guide/icon.png");
-            assert_eq!(logo_small, logo);
+            assert_eq!(updated.header.logo.as_ref(), "http://guide/icon.png");
+            assert_eq!(updated.header.logo_small.as_ref(), "http://guide/icon.png");
             assert_eq!(epg[0].children[0].id.as_ref(), "Demo.Channel");
         });
     }
@@ -1334,7 +1327,6 @@ mod tests {
                 priority: 0,
                 logo_override: false,
                 kind: PersistedEpgSourceKind::Xmltv,
-                input_name: None,
             }]);
             let mut playlist = FetchedPlaylist {
                 input: &input,
@@ -1442,7 +1434,6 @@ mod tests {
                 priority: 0,
                 logo_override: false,
                 kind: PersistedEpgSourceKind::Xmltv,
-                input_name: None,
             }]);
             let mut playlist = FetchedPlaylist {
                 input: &input,
@@ -1477,7 +1468,6 @@ mod tests {
                 priority: 0,
                 logo_override: false,
                 kind: PersistedEpgSourceKind::Xmltv,
-                input_name: None,
             }]);
             let mut playlist = FetchedPlaylist {
                 input: &input,

backend/processing/src/processor/library.rs

@@ -139,21 +139,14 @@ fn to_playlist_item(
 ) {
     match &entry.metadata {
         MediaMetadata::Movie(_) => {
-            let mut item = build_movie_playlist_item(entry, input_name, group_name, api_base_path);
-            item.header.ingest_resource_values(input_name);
-            channels.push(item);
+            channels.push(build_movie_playlist_item(entry, input_name, group_name, api_base_path));
         }
         MediaMetadata::Series(_) => {
             if let Some((series_info, episodes)) =
                 build_series_playlist_items(entry, input_name, group_name, api_base_path)
             {
-                let mut series_info = series_info;
-                series_info.header.ingest_resource_values(input_name);
                 channels.push(series_info);
-                channels.extend(episodes.into_iter().map(|mut episode| {
-                    episode.header.ingest_resource_values(input_name);
-                    episode
-                }));
+                channels.extend(episodes);
             }
         }
     }

backend/processing/src/processor/playlist/tests.rs

@@ -2425,7 +2425,6 @@ mod mapping_stage {
                         match_names: vec!["BBC One".intern()],
                         config: Box::new(IcsEpgSourceConfig::default()),
                     },
-                    input_name: None,
                 }]);
 
                 let mut playlist = FetchedPlaylist {
@@ -2521,7 +2520,6 @@ match {
                         match_names: vec![],
                         config: Box::new(IcsEpgSourceConfig::default()),
                     },
-                    input_name: None,
                 }]);
                 let mut playlist = FetchedPlaylist {
                     input: &input,

backend/processing/src/processor/playlist/transform.rs

@@ -164,12 +164,6 @@ impl MappingStageOutcome {
 }
 
 pub(crate) fn map_channel(mut channel: PlaylistItem, mapping: &CompiledMapping) -> ChannelMappingOutcome {
-    let mut trusted_locators = HashSet::new();
-    channel.header.visit_resource_values(&mut |value| {
-        if shared::model::resolve_resource_value(value).is_ok_and(|locator| locator.is_some()) {
-            trusted_locators.insert(Arc::clone(value));
-        }
-    });
     let mut matched_rules = 0;
     let mut virtual_items = vec![];
     let mut changed_fields = HashSet::new();
@@ -204,16 +198,6 @@ pub(crate) fn map_channel(mut channel: PlaylistItem, mapping: &CompiledMapping)
             }
         }
     }
-    let normalize_mapped_item = |item: &mut PlaylistItem| {
-        item.header.visit_resource_values_mut(&mut |value| {
-            if shared::model::has_resource_scheme(value) && !trusted_locators.contains(value) {
-                *value = Arc::from("");
-            }
-        });
-        item.header.normalize_internal_resource_values();
-    };
-    normalize_mapped_item(&mut channel);
-    virtual_items.iter_mut().for_each(normalize_mapped_item);
     ChannelMappingOutcome { channel, virtual_items, matched_rules, changed_fields, diagnostics }
 }
 

backend/processing/src/processor/stalker.rs

@@ -202,7 +202,7 @@ pub async fn download_stalker_playlist(
                 app_config,
                 api_client.as_ref(),
                 &handshake,
-                refresh_plan.clone(),
+                refresh_plan,
                 &storage_path,
                 identity_fingerprint,
                 refresh_mode.budget(),

backend/processing/src/processor/stalker_refresh.rs

@@ -128,12 +128,11 @@ impl StalkerClusterSelection {
 }
 
 /// Requested clusters and the publication policy applied once their generation is complete.
-#[derive(Clone)]
+#[derive(Clone, Copy)]
 pub struct StalkerRefreshPlan {
     selection: StalkerClusterSelection,
     update_quality: ConfigInputUpdateQuality,
     quality_bypass_mask: u8,
-    input_name: Arc<str>,
 }
 
 impl StalkerRefreshPlan {
@@ -144,7 +143,7 @@ impl StalkerRefreshPlan {
         let quality_bypass_mask = matches!(quality_policy, UpdateQualityPolicy::Bypass)
             .then_some(selection.mask() & MEDIA_SELECTION)
             .unwrap_or(0);
-        Self { selection, update_quality, quality_bypass_mask, input_name: Arc::clone(&input.name) }
+        Self { selection, update_quality, quality_bypass_mask }
     }
 }
 
@@ -560,16 +559,13 @@ fn map_items(
     categories: &HashMap<u32, StalkerCategory>,
     kind: StalkerStreamKind,
     added_at: i64,
-    input_name: &Arc<str>,
 ) -> Vec<StalkerPlaylistItem> {
     raw_items
         .iter()
         .map(|raw| {
             let category =
                 raw.category_id().and_then(|value| value.parse::<u32>().ok()).and_then(|id| categories.get(&id));
-            let mut item = parser::map_stalker_to_playlist_item(raw, category, kind, added_at);
-            item.ingest_resource_values(input_name);
-            item
+            parser::map_stalker_to_playlist_item(raw, category, kind, added_at)
         })
         .collect()
 }
@@ -640,7 +636,7 @@ pub async fn advance_stalker_refresh(
     identity_fingerprint: u64,
     mut budget: StalkerRefreshBudget,
 ) -> Result<StalkerRefreshOutcome, TuliproxError> {
-    let StalkerRefreshPlan { selection, update_quality, quality_bypass_mask, input_name } = refresh_plan;
+    let StalkerRefreshPlan { selection, update_quality, quality_bypass_mask } = refresh_plan;
     let mut checkpoint =
         load_or_start_checkpoint(storage_path, identity_fingerprint, selection, quality_bypass_mask).await?;
     if checkpoint.phase == StalkerRefreshPhase::Terminal {
@@ -677,7 +673,7 @@ pub async fn advance_stalker_refresh(
                         checkpoint.retry_count = 0;
                     }
                     Ok(raw) => {
-                        let items = map_items(&raw, categories, StalkerStreamKind::Live, added_at, &input_name);
+                        let items = map_items(&raw, categories, StalkerStreamKind::Live, added_at);
                         let path =
                             generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::Live);
                         snapshot_stalker_items_at(app_config, path.clone(), &items).await?;
@@ -711,7 +707,7 @@ pub async fn advance_stalker_refresh(
                 {
                     return yield_after_error(storage_path, checkpoint, err).await;
                 }
-                let items = map_items(&response.items, categories, StalkerStreamKind::Live, added_at, &input_name);
+                let items = map_items(&response.items, categories, StalkerStreamKind::Live, added_at);
                 let path = generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::Live);
                 upsert_stalker_items_at(app_config, &path, &items).await?;
                 checkpoint.processed = checkpoint.processed.saturating_add(items.len() as u64);
@@ -743,7 +739,7 @@ pub async fn advance_stalker_refresh(
                 {
                     return yield_after_error(storage_path, checkpoint, err).await;
                 }
-                let items = map_items(&response.items, categories, StalkerStreamKind::Movie, added_at, &input_name);
+                let items = map_items(&response.items, categories, StalkerStreamKind::Movie, added_at);
                 let path = generation_data_path(storage_path, checkpoint.generation, StalkerGenerationData::Vod);
                 upsert_stalker_items_at(app_config, &path, &items).await?;
                 checkpoint.processed = checkpoint.processed.saturating_add(items.len() as u64);
@@ -790,7 +786,6 @@ pub async fn advance_stalker_refresh(
                         root.flussonic_tmp_link = capabilities.flussonic_temporary_link;
                         root.wowza_tmp_link = capabilities.wowza_temporary_link;
                         root.use_http_tmp_link = capabilities.use_http_temporary_link;
-                        root.ingest_resource_values(&input_name);
                         root
                     })
                     .collect();
@@ -835,10 +830,7 @@ pub async fn advance_stalker_refresh(
                             used_episode_ids
                                 .insert(prepare_stalker_episode_series_at(app_config, &path, series_id).await?)
                         };
-                        let mut episodes = parser::map_stalker_series_details(&details, &root, added_at, used);
-                        for episode in &mut episodes {
-                            episode.ingest_resource_values(&input_name);
-                        }
+                        let episodes = parser::map_stalker_series_details(&details, &root, added_at, used);
                         upsert_stalker_items_at(app_config, &path, &episodes).await?;
                         checkpoint.processed = checkpoint.processed.saturating_add(episodes.len() as u64);
                         checkpoint.phase = StalkerRefreshPhase::SeriesDetails { provider_id: Some(root.stream_id) };
@@ -1529,7 +1521,7 @@ mod tests {
             StalkerCategory { id: "10".to_string(), title: "News".to_string(), alias: None, number: 1 },
         )]);
 
-        let items = map_items(&[raw], &categories, StalkerStreamKind::Live, 0, &Arc::from("stalker-input"));
+        let items = map_items(&[raw], &categories, StalkerStreamKind::Live, 0);
 
         assert_eq!(items.len(), 1);
         assert_eq!(items[0].category_id, 10);