From beacece287e5ca8196cf8f95db063f464bf83364 Mon Sep 17 00:00:00 2001 From: cfal Date: Sun, 25 Oct 2026 21:22:02 +0000 Subject: [PATCH] Preserve download routing and enforce cold-install boundaries Keep per-source pool rotation across attempts and reloads without retaining connectors. Construct URL targets without reparsing IPv6 addresses, resolve dangling local aliases against staged inputs, and enforce aggregate byte limits before downloads. Preserve validation causes and identify failing cache paths. --- src/client_proxy_chain.rs | 19 ++ src/rule_set/mod.rs | 34 ++- src/rule_updater/http.rs | 73 ++++- src/rule_updater/mod.rs | 65 ++++- src/rule_updater/tests.rs | 591 ++++++++++++++++++++++++++++---------- 5 files changed, 595 insertions(+), 187 deletions(-) diff --git a/src/client_proxy_chain.rs b/src/client_proxy_chain.rs index 7b02a9b1..a56d3ac4 100644 --- a/src/client_proxy_chain.rs +++ b/src/client_proxy_chain.rs @@ -144,6 +144,25 @@ impl std::fmt::Debug for ClientProxyChain { } impl ClientProxyChain { + #[cfg(unix)] + pub(crate) fn tcp_selection_indices(&self) -> Vec { + std::iter::once(&self.initial_hop_next_index) + .chain(&self.subsequent_next_indices) + .map(|counter| counter.load(Ordering::Relaxed)) + .collect() + } + + #[cfg(unix)] + pub(crate) fn restore_tcp_selection_indices(&mut self, indices: &[u32]) { + assert_eq!(indices.len(), self.num_hops()); + for (counter, index) in std::iter::once(&mut self.initial_hop_next_index) + .chain(&mut self.subsequent_next_indices) + .zip(indices) + { + *counter.get_mut() = *index; + } + } + fn needs_carrier_dialer(&self) -> bool { self.initial_hop.iter().any(|entry| matches!(entry.as_ref(), InitialHopEntry::Proxy { proxy, .. } if proxy.needs_carrier_dialer())) || self.subsequent_hops.iter().flatten().any(|proxy| proxy.needs_carrier_dialer()) diff --git a/src/rule_set/mod.rs b/src/rule_set/mod.rs index b7345c38..16f55512 100644 --- a/src/rule_set/mod.rs +++ b/src/rule_set/mod.rs @@ -267,17 +267,31 @@ fn cache_paths(path: &Path) -> [PathBuf; 3] { } pub(crate) fn source_identity(path: &Path) -> io::Result { - let absolute = std::path::absolute(path)?; - if let Ok(canonical) = absolute.canonicalize() { - return Ok(canonical); + let mut absolute = std::path::absolute(path)?; + for _ in 0..40 { + if let Ok(canonical) = absolute.canonicalize() { + return Ok(canonical); + } + // A local alias may point at a remote cache that has not been installed yet. + if let Ok(target) = std::fs::read_link(&absolute) { + absolute = if target.is_absolute() { + target + } else { + absolute.parent().unwrap().join(target) + }; + continue; + } + if let Some(parent) = absolute.parent() + && let Ok(parent) = parent.canonicalize() + && let Some(name) = absolute.file_name() + { + return Ok(parent.join(name)); + } + return Ok(absolute); } - if let Some(parent) = absolute.parent() - && let Ok(parent) = parent.canonicalize() - && let Some(name) = absolute.file_name() - { - return Ok(parent.join(name)); - } - Ok(absolute) + Err(invalid( + "rule source symlink chain exceeds 40 links or contains a loop", + )) } pub(crate) fn validate_remote_paths(configs: &mut [Config], extra: &[PathBuf]) -> io::Result<()> { diff --git a/src/rule_updater/http.rs b/src/rule_updater/http.rs index 6d8a9b6c..8ec34697 100644 --- a/src/rule_updater/http.rs +++ b/src/rule_updater/http.rs @@ -18,11 +18,41 @@ use tokio::time::timeout; const MAX_HEADER_COUNT: usize = 64; const MAX_HEADER_BYTES: usize = 32 * 1024; +#[derive(Clone, Default)] +pub(super) struct DownloadState { + indices: Vec, +} + +struct DownloadAttempt<'a> { + chain: crate::client_proxy_chain::ClientProxyChain, + state: &'a mut DownloadState, +} + +impl Drop for DownloadAttempt<'_> { + fn drop(&mut self) { + // Failed and cancelled attempts still advance rotation; connectors remain transient. + self.state.indices = self.chain.tcp_selection_indices(); + } +} + +fn target(url: &url::Url) -> io::Result { + use crate::address::{Address, NetLocation}; + let port = url + .port_or_known_default() + .ok_or_else(|| invalid("missing URL port"))?; + match url.host().ok_or_else(|| invalid("missing URL host"))? { + url::Host::Domain(name) => Ok(NetLocation::new(Address::Hostname(name.to_owned()), port)), + url::Host::Ipv4(ip) => Ok(NetLocation::from_ip_addr(ip.into(), port)), + url::Host::Ipv6(ip) => Ok(NetLocation::from_ip_addr(ip.into(), port)), + } +} + async fn connect( url: &url::Url, download: &Download, dns: &DnsRegistry, global: GlobalLimits, + state: &mut DownloadState, ) -> io::Result> { let host = match url.host().ok_or_else(|| invalid("missing URL host"))? { url::Host::Domain(name) => name.to_owned(), @@ -30,8 +60,7 @@ async fn connect( url::Host::Ipv6(ip) => ip.to_string(), }; let resolver = dns.rule_bootstrap(&download.bootstrap_url)?; - let target = - crate::address::NetLocation::from_str(&host, Some(url.port_or_known_default().unwrap()))?; + let target = target(url)?; let targets = crate::resolver::resolve_addresses(&resolver, &target) .await .map_err(|_| invalid("rule download bootstrap lookup failed"))?; @@ -68,15 +97,19 @@ async fn connect( } } } - let chain = crate::tcp::chain_builder::build_client_chain_group_with_limits( - crate::option_util::NoneOrSome::One(download.client_chain.clone()), + let mut chain = crate::tcp::chain_builder::build_client_proxy_chain_with_limits( + download.client_chain.hops.clone(), resolver.clone(), global, )?; + if !state.indices.is_empty() { + chain.restore_tcp_selection_indices(&state.indices); + } + let attempt = DownloadAttempt { chain, state }; let mut connected = None; for target in targets { let target = crate::address::NetLocation::from_ip_addr(target.ip(), target.port()); - if let Ok(result) = chain.connect_tcp(target.into(), &resolver).await { + if let Ok(result) = attempt.chain.connect_tcp(target.into(), &resolver).await { connected = Some(result); break; } @@ -114,6 +147,7 @@ pub(super) async fn download( global: GlobalLimits, file: &mut File, limits: Limits, + state: &mut DownloadState, ) -> io::Result { let url = source .parsed_url()? @@ -122,7 +156,7 @@ pub(super) async fn download( .download .as_ref() .ok_or_else(|| invalid("download policy required"))?; - let stream = timeout(limits.connect, connect(&url, download, dns, global)) + let stream = timeout(limits.connect, connect(&url, download, dns, global, state)) .await .map_err(|_| invalid("connect/DNS/TLS deadline exceeded"))??; @@ -209,3 +243,30 @@ pub(super) async fn download( _ = connection => receive.await, } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn url_targets_preserve_host_and_port_without_reparsing_authorities() { + for (raw, address, port) in [ + ("http://[::1]:8080/rules", "::1", 8080), + ( + "http://[2001:db8::1:8080]:9090/rules", + "2001:db8::1:8080", + 9090, + ), + ("https://[::ffff:192.0.2.1]/rules", "::ffff:192.0.2.1", 443), + ("https://example.com:8443/rules", "example.com", 8443), + ("http://192.0.2.1/rules", "192.0.2.1", 80), + ] { + let location = target(&url::Url::parse(raw).unwrap()).unwrap(); + assert_eq!( + location.address(), + &crate::address::Address::from(address).unwrap() + ); + assert_eq!(location.port(), port, "{raw}"); + } + } +} diff --git a/src/rule_updater/mod.rs b/src/rule_updater/mod.rs index d793690f..6c5edda3 100644 --- a/src/rule_updater/mod.rs +++ b/src/rule_updater/mod.rs @@ -14,6 +14,7 @@ use install::Install; use tokio::time::{Instant, timeout}; const MAX_BYTES: u64 = 64 * 1024 * 1024; +const MAX_STAGED_BYTES: u64 = 512 * 1024 * 1024; #[derive(Clone, Copy)] struct Limits { @@ -47,6 +48,7 @@ async fn stage( dns: &DnsRegistry, global: GlobalLimits, limits: Limits, + state: &mut http::DownloadState, ) -> io::Result { let mut install = Install::open(&source.path)?; let digest = crate::control::url_test::BACKGROUND @@ -54,7 +56,7 @@ async fn stage( (), timeout( limits.overall, - http::download(source, dns, global, &mut install.file, limits), + http::download(source, dns, global, &mut install.file, limits, state), ), ) .await @@ -80,7 +82,11 @@ async fn validate_staged(configs: Vec, staged: &[Staged]) -> io::Result< .collect::>>()?; let validated = crate::config::compile_configs_with_inputs(configs, inputs) .await - .map_err(|_| invalid("download rejected by whole configuration validation"))?; + .map_err(|error| { + invalid(format!( + "download rejected by whole configuration validation: {error}" + )) + })?; tokio::task::spawn_blocking(move || { if validated.catalog.is_some() { crate::control::validation::validate_resources(&validated)?; @@ -91,6 +97,41 @@ async fn validate_staged(configs: Vec, staged: &[Staged]) -> io::Result< .map_err(io::Error::other)? } +async fn stage_sources( + sources: &[RuleSource], + dns: &DnsRegistry, + global: GlobalLimits, + mut remaining: u64, +) -> io::Result> { + let mut staged = Vec::new(); + for source in sources { + if remaining == 0 { + return Err(invalid("cold rule downloads exceed staging budget")); + } + let limits = Limits { + bytes: MAX_BYTES.min(remaining), + ..Limits::default() + }; + let item = stage( + source, + dns, + global, + limits, + &mut http::DownloadState::default(), + ) + .await + .map_err(|error| { + io::Error::new( + error.kind(), + format!("rule download {}: {error}", source.path.display()), + ) + })?; + remaining -= item.install.file.metadata()?.len(); + staged.push(item); + } + Ok(staged) +} + pub(crate) async fn prepare( mut configs: Vec, previous: Option<&DnsRegistry>, @@ -156,16 +197,7 @@ pub(crate) async fn prepare( ) .await .map_err(|_| invalid("rule bootstrap preparation deadline exceeded"))??; - let mut staged = Vec::new(); - let mut total = 0; - for source in &sources { - let item = stage(source, &dns, global, Limits::default()).await?; - total += item.install.file.metadata()?.len(); - if total > 512 * 1024 * 1024 { - return Err(invalid("cold rule downloads exceed 512 MiB staging budget")); - } - staged.push(item); - } + let staged = stage_sources(&sources, &dns, global, MAX_STAGED_BYTES).await?; let validated = validate_staged(configs, &staged).await?; for mut item in staged { item.install.finish(item.digest)?; @@ -182,6 +214,7 @@ struct Scheduled { source: RuleSource, due: Instant, failures: u32, + download: http::DownloadState, } async fn refresh_source( @@ -189,8 +222,9 @@ async fn refresh_source( configs: &[Config], dns: &DnsRegistry, global: GlobalLimits, + state: &mut http::DownloadState, ) -> io::Result { - let mut staged = stage(source, dns, global, Limits::default()).await?; + let mut staged = stage(source, dns, global, Limits::default(), state).await?; if staged.install.unchanged(staged.digest)? { return Ok(false); } @@ -231,6 +265,8 @@ impl Refresh { source: source.clone(), due: prior.map_or_else(|| Instant::now() + remaining, |item| item.due), failures: prior.map_or(0, |item| item.failures), + download: prior + .map_or_else(http::DownloadState::default, |item| item.download.clone()), }); } Ok(Self { configs, scheduled }) @@ -243,7 +279,8 @@ impl Refresh { return; }; tokio::time::sleep_until(item.due).await; - let result = refresh_source(&item.source, &self.configs, dns, global).await; + let result = + refresh_source(&item.source, &self.configs, dns, global, &mut item.download).await; let interval = item.source.interval().expect("validated interval"); let delay = match &result { Ok(_) => { diff --git a/src/rule_updater/tests.rs b/src/rule_updater/tests.rs index d223d089..40c8952c 100644 --- a/src/rule_updater/tests.rs +++ b/src/rule_updater/tests.rs @@ -54,8 +54,14 @@ fn configs(source: &RuleSource) -> Vec { } async fn update(source: &RuleSource, limits: Limits) -> io::Result<&'static str> { - let mut staged = - super::stage(source, &DnsRegistry::new(), GlobalLimits::default(), limits).await?; + let mut staged = super::stage( + source, + &DnsRegistry::new(), + GlobalLimits::default(), + limits, + &mut http::DownloadState::default(), + ) + .await?; validate_staged(configs(source), std::slice::from_ref(&staged)).await?; Ok(if staged.install.finish(staged.digest)?.changed() { "updated" @@ -718,12 +724,168 @@ async fn cold_start_is_validated_as_a_whole_and_offline_checks_do_not_download() let (url, fixture) = serve(response(200, "", VALID)).await; let mut invalid = configs(&self::source(&url, &path)); invalid.push(serde_json::from_value(serde_json::json!({"rule_group": "bad", "rules": [{"masks": {"ip_cidr": ["not an IP"]}}]})).unwrap()); - assert!(prepare(invalid, None, &mut None).await.is_err()); + let error = prepare(invalid, None, &mut None) + .await + .err() + .unwrap() + .to_string(); + assert!(error.contains("whole configuration validation"), "{error}"); + assert!(error.contains("IP"), "{error}"); + assert!(!error.contains("QUERY_SECRET"), "{error}"); fixture.await.unwrap(); assert!(!path.exists()); assert!(!stage_path(&path).exists()); } +#[tokio::test] +async fn cold_staging_caps_each_download_by_remaining_capacity() { + let dir = directory(); + for budget in [ + OLD.len() as u64, + (OLD.len() * 2 - 1) as u64, + (OLD.len() * 2) as u64, + ] { + let a = dir.path().join("a.json"); + let b = dir.path().join("b.json"); + let (url_a, fixture_a) = serve(response(200, "", OLD)).await; + let trap = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let (url_b, fixture_b) = if budget == OLD.len() as u64 { + (format!("http://{}/rules", trap.local_addr().unwrap()), None) + } else { + let (url, fixture) = serve(response(200, "", OLD)).await; + (url, Some(fixture)) + }; + let sources = [source(&url_a, &a), source(&url_b, &b)]; + let result = stage_sources( + &sources, + &DnsRegistry::new(), + GlobalLimits::default(), + budget, + ) + .await; + fixture_a.await.unwrap(); + if let Some(fixture) = fixture_b { + fixture.await.unwrap(); + } + if budget == (OLD.len() * 2) as u64 { + let staged = result.unwrap(); + assert_eq!( + staged + .iter() + .map(|item| item.install.file.metadata().unwrap().len()) + .sum::(), + budget + ); + drop(staged); + } else { + let error = result.err().unwrap().to_string(); + if budget == OLD.len() as u64 { + assert!(error.contains("staging budget"), "{error}"); + assert!( + timeout(Duration::from_millis(20), trap.accept()) + .await + .is_err() + ); + } else { + assert!(error.contains("size outside permitted range"), "{error}"); + assert!(error.contains(b.to_str().unwrap()), "{error}"); + } + } + for path in [&a, &b] { + assert!(!path.exists()); + assert!(!stage_path(path).exists()); + } + } +} + +#[tokio::test] +async fn cold_download_errors_identify_cache_without_exposing_url() { + let dir = directory(); + let path = dir.path().join("missing-directory/rules.json"); + let source = source("http://127.0.0.1:1/rules?token=QUERY_SECRET", &path); + let error = prepare(configs(&source), None, &mut None) + .await + .err() + .unwrap() + .to_string(); + assert!(error.contains(path.to_str().unwrap()), "{error}"); + assert!(!error.contains("QUERY_SECRET"), "{error}"); + assert!(!stage_path(&path).exists()); +} + +#[tokio::test] +async fn cold_download_validates_dangling_local_aliases_against_staged_bytes() { + let dir = directory(); + let path = dir.path().join("cache.json"); + let alias = dir.path().join("alias.json"); + symlink("second.json", &alias).unwrap(); + symlink("cache.json", dir.path().join("second.json")).unwrap(); + let (url, fixture) = serve(response(200, "", VALID)).await; + let source = source(&url, &path); + let mut config = configs(&source); + config.push( + serde_json::from_value( + serde_json::json!({"rule_source_group": "alias", "rules": [{"path": alias}]}), + ) + .unwrap(), + ); + prepare(config, None, &mut None).await.unwrap(); + fixture.await.unwrap(); + assert_eq!(std::fs::read(&path).unwrap(), VALID); + assert_eq!(std::fs::read(alias).unwrap(), VALID); +} + +#[test] +fn dangling_sidecar_aliases_and_symlink_cycles_are_rejected() { + let dir = directory(); + let path = dir.path().join("cache.json"); + let source = source("https://example.invalid/rules.json", &path); + for suffix in [".shoes-update.lock", ".shoes-update.tmp"] { + let alias = dir.path().join(format!("alias{suffix}.json")); + symlink(format!("cache.json{suffix}"), &alias).unwrap(); + let mut config = configs(&source); + config.push( + serde_json::from_value( + serde_json::json!({"rule_source_group": "alias", "rules": [{"path": alias}]}), + ) + .unwrap(), + ); + let error = crate::rule_set::validate_remote_paths(&mut config, &[]).unwrap_err(); + assert!(error.to_string().contains("sidecar"), "{error}"); + } + let cycle = dir.path().join("cycle.json"); + symlink("./cycle.json", &cycle).unwrap(); + assert!( + crate::rule_set::source_identity(&cycle) + .unwrap_err() + .to_string() + .contains("loop") + ); +} + +#[tokio::test] +async fn ipv6_literal_download_reaches_the_exact_url_port() { + let listener = TcpListener::bind("[::1]:0").await.unwrap(); + let url = format!("http://{}/rules", listener.local_addr().unwrap()); + let fixture = tokio::spawn(async move { + let (mut stream, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + request(&mut stream).await; + stream.write_all(&response(200, "", VALID)).await.unwrap(); + }); + let dir = directory(); + let path = dir.path().join("rules.json"); + assert_eq!( + update(&source(&url, &path), Limits::default()) + .await + .unwrap(), + "updated" + ); + fixture.await.unwrap(); +} + #[tokio::test] async fn named_download_chain_is_used_and_failure_never_falls_back_to_direct() { use crate::control::selection::{SelectionOwner, SelectorDefinition}; @@ -828,6 +990,111 @@ async fn refresh_noop_remembers_check_time_and_removal_drops_download() { assert_eq!(std::fs::read(&path).unwrap(), OLD); } +#[tokio::test] +async fn refresh_pools_rotate_after_success_failure_and_generation_replacement() { + for first_result in ["success", "failure", "cancelled"] { + for replace_generation in [false, true] { + let dir = directory(); + let path = dir.path().join("cache.json"); + write_rule(&path, OLD); + let trap = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let a = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let b = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let mut source = source( + &format!("http://{}/rules", trap.local_addr().unwrap()), + &path, + ); + source.download.as_mut().unwrap().client_chain = + serde_json::from_value(serde_json::json!([{"pool": [ + {"address": a.local_addr().unwrap().to_string(), "protocol": {"type": "socks"}}, + {"address": b.local_addr().unwrap().to_string(), "protocol": {"type": "socks"}} + ]}])) + .unwrap(); + let fixture = tokio::spawn(async move { + let mut selected = Vec::new(); + let mut stalled = None; + for attempt in 0..3 { + let (index, mut stream) = timeout(Duration::from_secs(5), async { + tokio::select! { + result = a.accept() => (0, result.unwrap().0), + result = b.accept() => (1, result.unwrap().0), + } + }) + .await + .unwrap(); + selected.push(index); + if attempt == 0 { + if first_result == "failure" { + continue; + } + if first_result == "cancelled" { + stalled = Some(stream); + continue; + } + } + let mut greeting = [0; 3]; + stream.read_exact(&mut greeting).await.unwrap(); + assert_eq!(greeting, [5, 1, 0]); + stream.write_all(&[5, 0]).await.unwrap(); + let mut connect = [0; 10]; + stream.read_exact(&mut connect).await.unwrap(); + stream + .write_all(&[5, 0, 0, 1, 127, 0, 0, 1, 0, 1]) + .await + .unwrap(); + request(&mut stream).await; + stream.write_all(&response(200, "", VALID)).await.unwrap(); + } + drop(stalled); + selected + }); + let config = configs(&source); + let sources = crate::config::compile_configs(config.clone()) + .await + .unwrap() + .remote_rules; + let mut refresh = Refresh::new(config.clone(), &sources, None).unwrap(); + for attempt in 0..3 { + let item = &mut refresh.scheduled[0]; + let dns = DnsRegistry::new(); + let future = refresh_source( + &item.source, + &config, + &dns, + GlobalLimits::default(), + &mut item.download, + ); + if first_result == "cancelled" && attempt == 0 { + assert!(timeout(Duration::from_millis(100), future).await.is_err()); + assert_eq!(std::fs::read(&path).unwrap(), OLD); + } else if first_result == "failure" && attempt == 0 { + assert!( + future + .await + .unwrap_err() + .to_string() + .contains("configured rule download chain failed") + ); + assert_eq!(std::fs::read(&path).unwrap(), OLD); + } else { + future.await.unwrap(); + } + if replace_generation { + refresh = Refresh::new(config.clone(), &sources, Some(&refresh)).unwrap(); + } + } + assert_eq!(fixture.await.unwrap(), [0, 1, 0]); + assert!( + timeout(Duration::from_millis(20), trap.accept()) + .await + .is_err() + ); + assert_eq!(std::fs::read(&path).unwrap(), VALID); + assert!(!stage_path(&path).exists()); + } + } +} + #[tokio::test] async fn refresh_validates_the_whole_candidate_only_for_changed_bytes() { let dir = directory(); @@ -849,16 +1116,20 @@ async fn refresh_validates_the_whole_candidate_only_for_changed_bytes() { &config, &DnsRegistry::new(), GlobalLimits::default(), + &mut http::DownloadState::default(), ) .await; fixture.await.unwrap(); if body == OLD { assert!(!result.unwrap()); } else { - assert_eq!( - result.unwrap_err().to_string(), - "download rejected by whole configuration validation" + let error = result.unwrap_err().to_string(); + assert!( + error.starts_with("download rejected by whole configuration validation:"), + "{error}" ); + assert!(error.contains("IP"), "{error}"); + assert!(!error.contains("QUERY_SECRET"), "{error}"); } assert_preserved(&path, &original); } @@ -901,9 +1172,15 @@ async fn explicit_bootstrap_resolves_target_without_system_dns() { &path, ); source.download.as_mut().unwrap().bootstrap_url = "bootstrap".into(); - let mut staged = super::stage(&source, &dns, GlobalLimits::default(), Limits::default()) - .await - .unwrap(); + let mut staged = super::stage( + &source, + &dns, + GlobalLimits::default(), + Limits::default(), + &mut http::DownloadState::default(), + ) + .await + .unwrap(); staged.install.finish(staged.digest).unwrap(); fixture.await.unwrap(); assert_eq!(resolver.calls.load(std::sync::atomic::Ordering::Relaxed), 1); @@ -973,154 +1250,6 @@ fn cache_conflicts_and_unknown_download_routes_are_rejected_offline() { assert!(crate::config::create_server_configs(configs(&missing)).is_err()); } -#[test] -fn remote_sources_work_in_source_groups_policies_and_flat_overrides() { - let dir = directory(); - let path = dir.path().join("rules.json"); - write_rule(&path, VALID); - let remote = source("https://example.com/rules.json", &path); - let mut flat = serde_json::to_value(&remote).unwrap(); - flat["action"] = "block".into(); - let config = serde_json::from_value(serde_json::json!([ - {"rule_source_group": "sources", "rules": [remote, {"domain": ["inline.example"]}]}, - {"rule_group": "policies", "rules": [flat]}, - {"address": "127.0.0.1:0", "rules": ["policies", { - "masks": {"rule_source_group": "sources"}, "action": "block" - }], "protocol": {"type": "websocket", "targets": [{ - "protocol": {"type": "socks"}, "override_rules": [flat] - }]}} - ])) - .unwrap(); - let mut validated = crate::config::create_server_configs(config).unwrap(); - assert_eq!(validated.remote_rules.len(), 1); - let mut count = 0; - crate::config::rule_visit::visit( - &mut validated.configs, - &mut |rule| { - assert!(matches!( - rule.action, - crate::config::RuleActionConfig::Block - )); - assert!(rule.criteria.compiled.is_some()); - count += 1; - Ok(()) - }, - &mut |_| panic!("unexpanded policy group"), - ) - .unwrap(); - assert_eq!(count, 3); -} - -#[test] -fn local_source_reuse_does_not_accept_an_empty_remote_cache() { - let dir = directory(); - let path = dir.path().join("rules.json"); - write_rule(&path, br#"{"version":1,"rules":[]}"#); - let remote = source("https://example.com/rules.json", &path); - let local = RuleSource::from_path(path); - let mut compiler = crate::rules::Compiler::default(); - crate::rule_set::load_source(&local, &mut compiler).unwrap(); - assert!( - crate::rule_set::load_source(&remote, &mut compiler) - .err() - .unwrap() - .to_string() - .contains("empty remote") - ); -} - -#[tokio::test] -async fn staged_candidate_uses_normalized_paths_even_before_cache_creation() { - let dir = directory(); - let path = dir.path().join("./rules.json"); - let stage = dir.path().join("staged"); - write_rule(&stage, VALID); - let remote = source("https://example.com/rules.json", &path); - let mut config = configs(&remote); - config.push( - serde_json::from_value(serde_json::json!({ - "rule_source_group": "local-alias", "rules": [{ - "path": dir.path().join("rules.json"), "format": "mihomo_text", "behavior": "ipcidr" - }] - })) - .unwrap(), - ); - let inputs = HashMap::from([(path, File::open(stage).unwrap())]); - let error = crate::config::compile_configs_with_inputs(config, inputs) - .await - .err() - .unwrap(); - assert_eq!(error.kind(), io::ErrorKind::InvalidInput, "{error}"); - assert!(error.to_string().contains("CIDR"), "{error}"); -} - -#[test] -fn source_identity_resolves_existing_paths_and_missing_leaf_parents() { - use crate::rule_set::source_identity; - - let dir = directory(); - let parent = dir.path().join("parent"); - std::fs::create_dir(&parent).unwrap(); - let alias = dir.path().join("alias"); - symlink(&parent, &alias).unwrap(); - let canonical_parent = parent.canonicalize().unwrap(); - - write_rule(&parent.join("existing.json"), VALID); - for name in ["existing.json", "missing.json"] { - assert_eq!( - source_identity(&alias.join(name)).unwrap(), - canonical_parent.join(name) - ); - } - let missing_parent = alias.join("missing/rules.json"); - assert_eq!(source_identity(&missing_parent).unwrap(), missing_parent); -} - -#[tokio::test] -async fn caches_cannot_overwrite_config_pem_state_or_other_source_files() { - let dir = directory(); - let path = dir.path().join("cache.json"); - write_rule(&path, VALID); - let source = source("https://example.com/rules.json", &path); - for extra in [ - serde_json::json!({"cache_file_path": path}), - serde_json::json!({"dns_group": "dns", "dns_servers": "system", "fake_ip": {"state_file": path}}), - serde_json::json!({"path": path, "protocol": {"type": "socks"}}), - serde_json::json!({"rule_source_group": "other", "rules": [{"path": stage_path(&path)}]}), - ] { - let mut config = configs(&source); - config.push(serde_json::from_value(extra).unwrap()); - assert!( - prepare(config, None, &mut None) - .await - .err() - .unwrap() - .to_string() - .contains("overlap") - ); - assert_eq!(std::fs::read(&path).unwrap(), VALID); - } - let mut config = configs(&source); - config.push(serde_json::from_value(serde_json::json!({"pem": "pem", "path": path})).unwrap()); - assert!( - crate::config::convert_cert_paths(config) - .await - .err() - .unwrap() - .to_string() - .contains("overlap") - ); - // Declaring-file protection runs before any rule decoding or networking. - std::fs::write(&path, serde_json::to_vec(&configs(&source)).unwrap()).unwrap(); - assert!( - crate::config::load_configs(&vec![path.to_str().unwrap().into()]) - .await - .err() - .unwrap() - .to_string() - .contains("overlap") - ); -} #[test] fn remote_bootstrap_literals_reject_active_and_retained_synthetic_ranges() { let dir = directory(); @@ -1275,3 +1404,151 @@ async fn remote_bootstrap_group_names_take_precedence_over_url_syntax() { } } +#[test] +fn remote_sources_work_in_source_groups_policies_and_flat_overrides() { + let dir = directory(); + let path = dir.path().join("rules.json"); + write_rule(&path, VALID); + let remote = source("https://example.com/rules.json", &path); + let mut flat = serde_json::to_value(&remote).unwrap(); + flat["action"] = "block".into(); + let config = serde_json::from_value(serde_json::json!([ + {"rule_source_group": "sources", "rules": [remote, {"domain": ["inline.example"]}]}, + {"rule_group": "policies", "rules": [flat]}, + {"address": "127.0.0.1:0", "rules": ["policies", { + "masks": {"rule_source_group": "sources"}, "action": "block" + }], "protocol": {"type": "websocket", "targets": [{ + "protocol": {"type": "socks"}, "override_rules": [flat] + }]}} + ])) + .unwrap(); + let mut validated = crate::config::create_server_configs(config).unwrap(); + assert_eq!(validated.remote_rules.len(), 1); + let mut count = 0; + crate::config::rule_visit::visit( + &mut validated.configs, + &mut |rule| { + assert!(matches!( + rule.action, + crate::config::RuleActionConfig::Block + )); + assert!(rule.criteria.compiled.is_some()); + count += 1; + Ok(()) + }, + &mut |_| panic!("unexpanded policy group"), + ) + .unwrap(); + assert_eq!(count, 3); +} + +#[test] +fn local_source_reuse_does_not_accept_an_empty_remote_cache() { + let dir = directory(); + let path = dir.path().join("rules.json"); + write_rule(&path, br#"{"version":1,"rules":[]}"#); + let remote = source("https://example.com/rules.json", &path); + let local = RuleSource::from_path(path); + let mut compiler = crate::rules::Compiler::default(); + crate::rule_set::load_source(&local, &mut compiler).unwrap(); + assert!( + crate::rule_set::load_source(&remote, &mut compiler) + .err() + .unwrap() + .to_string() + .contains("empty remote") + ); +} + +#[tokio::test] +async fn staged_candidate_uses_normalized_paths_even_before_cache_creation() { + let dir = directory(); + let path = dir.path().join("./rules.json"); + let stage = dir.path().join("staged"); + write_rule(&stage, VALID); + let remote = source("https://example.com/rules.json", &path); + let mut config = configs(&remote); + config.push( + serde_json::from_value(serde_json::json!({ + "rule_source_group": "local-alias", "rules": [{ + "path": dir.path().join("rules.json"), "format": "mihomo_text", "behavior": "ipcidr" + }] + })) + .unwrap(), + ); + let inputs = HashMap::from([(path, File::open(stage).unwrap())]); + let error = crate::config::compile_configs_with_inputs(config, inputs) + .await + .err() + .unwrap(); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput, "{error}"); + assert!(error.to_string().contains("CIDR"), "{error}"); +} + +#[test] +fn source_identity_resolves_existing_paths_and_missing_leaf_parents() { + use crate::rule_set::source_identity; + + let dir = directory(); + let parent = dir.path().join("parent"); + std::fs::create_dir(&parent).unwrap(); + let alias = dir.path().join("alias"); + symlink(&parent, &alias).unwrap(); + let canonical_parent = parent.canonicalize().unwrap(); + + write_rule(&parent.join("existing.json"), VALID); + for name in ["existing.json", "missing.json"] { + assert_eq!( + source_identity(&alias.join(name)).unwrap(), + canonical_parent.join(name) + ); + } + let missing_parent = alias.join("missing/rules.json"); + assert_eq!(source_identity(&missing_parent).unwrap(), missing_parent); +} + +#[tokio::test] +async fn caches_cannot_overwrite_config_pem_state_or_other_source_files() { + let dir = directory(); + let path = dir.path().join("cache.json"); + write_rule(&path, VALID); + let source = source("https://example.com/rules.json", &path); + for extra in [ + serde_json::json!({"cache_file_path": path}), + serde_json::json!({"dns_group": "dns", "dns_servers": "system", "fake_ip": {"state_file": path}}), + serde_json::json!({"path": path, "protocol": {"type": "socks"}}), + serde_json::json!({"rule_source_group": "other", "rules": [{"path": stage_path(&path)}]}), + ] { + let mut config = configs(&source); + config.push(serde_json::from_value(extra).unwrap()); + assert!( + prepare(config, None, &mut None) + .await + .err() + .unwrap() + .to_string() + .contains("overlap") + ); + assert_eq!(std::fs::read(&path).unwrap(), VALID); + } + let mut config = configs(&source); + config.push(serde_json::from_value(serde_json::json!({"pem": "pem", "path": path})).unwrap()); + assert!( + crate::config::convert_cert_paths(config) + .await + .err() + .unwrap() + .to_string() + .contains("overlap") + ); + // Declaring-file protection runs before any rule decoding or networking. + std::fs::write(&path, serde_json::to_vec(&configs(&source)).unwrap()).unwrap(); + assert!( + crate::config::load_configs(&vec![path.to_str().unwrap().into()]) + .await + .err() + .unwrap() + .to_string() + .contains("overlap") + ); +}