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") + ); +}