gitea-admin/clash-verge-rev
1
0
Fork 0
mirror of https://github.com/clash-verge-rev/clash-verge-rev.git synced 2026-11-13 22:32:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add launch guard and secure multi-user service ownership (#7569)

Browse Source 
* feat(service): isolate macOS launch and owner runtime

* ci(windows): run owner security tests

* chore(deps): use kode-bridge v0.4.1 tag

* fix(macos): classify installed service state

* fix(macos): classify marked unavailable service

* fix(macos): gate core startup on service migration

* fix(macos): scope service init to supported platforms

* feat(macos): prompt before migrating installed service

* fix(macos): shorten sidecar IPC path

* build(service): pin helper protocol and artifacts

* fix(core): require confirmed service and sidecar readiness

* fix(macos): harden launch roots and singleton state

* fix(ui): refresh migration state before restart

* fix(macos): linearize service migration lifecycle

* fix(ui): keep service migration state fresh

* feat(service): track authenticated owner sessions

* fix(service): preserve newer owner sessions

* feat(macos): route service proxy changes through helper

* fix(macos): serialize service proxy guard

* fix(macos): close service proxy owner lifecycle

* fix(macos): close remaining proxy stop races

* fix(app): do not cancel controlled service stop

* fix(lifecycle): keep controlled updates cancellation-safe

* fix(macos): stage dev core for service

* test(macos): exercise staged core cleanup

* fix(dev): await controlled macOS exit

* chore: update changelog

* fix(macos): recover sidecar after service install failure

Allow the existing sidecar to continue when service installation leaves the manager in InstallRequired, then settle the session state as SidecarAllowed. Show neutral status text while the migration dialog waits for its final state.

* fix(macos): stage dev service tools before elevation

* fix(macos): localize launch guard prompts

* fix(service): address cross-platform integration issues

* feat: support seamless service migration

* fix: pass Windows service installer path safely

* fix(prebuild): derive service release from Cargo dependency

* fix(dev): preserve service installation state

* fix: resolve clippy warnings in service lifecycle
This commit is contained in:
wonfen
2026-07-24 09:01:49 +08:00
committed by GitHub
parent c079da6c99
commit 23dda47641
90 changed files with 9711 additions and 900 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/core/handle.rs
+4
View File
@@ -80,6 +80,10 @@ impl Handle {
self.is_exiting.store(true, Ordering::Release);
}
pub fn clear_is_exiting(&self) {
self.is_exiting.store(false, Ordering::Release);
}
pub fn is_exiting(&self) -> bool {
self.is_exiting.load(Ordering::Acquire)
}
src-tauri/src/core/logger.rs
+4 -19
View File
@@ -8,7 +8,6 @@ use std::{
use anyhow::{Result, bail};
use clash_verge_logging::{Type, logging};
use clash_verge_service_ipc::WriterConfig;
use compact_str::CompactString;
use flexi_logger::{
Cleanup, Criterion, DeferredNow, FileSpec, LogSpecBuilder, LogSpecification, LoggerHandle,
@@ -20,7 +19,7 @@ use parking_lot::{Mutex, RwLock};
use crate::{
core::service,
singleton,
utils::dirs::{self, service_log_dir, sidecar_log_dir},
utils::dirs::{self, sidecar_log_dir},
};
pub struct Logger {
@@ -180,10 +179,9 @@ impl Logger {
*self.sidecar_file_writer.write() = Some(sidecar_writer);
// update service writer config
if service::is_service_ipc_path_exists() && service::is_service_available().await.is_ok() {
let service_log_dir = dirs::path_to_str(&service_log_dir()?)?.into();
clash_verge_service_ipc::update_writer(&WriterConfig {
directory: service_log_dir,
if service::is_service_available().await {
service::update_writer_by_service(&clash_verge_service_ipc::WriterConfig {
directory: String::new(),
max_log_size: log_max_size * 1024,
max_log_files: log_max_count,
})
@@ -225,17 +223,4 @@ impl Logger {
logging!(error, Type::System, "failed to get sidecar file log writer");
}
}
pub fn service_writer_config(&self) -> Result<WriterConfig> {
let service_log_dir = dirs::path_to_str(&service_log_dir()?)?.into();
let log_max_size = self.log_max_size.load(Ordering::SeqCst);
let log_max_count = self.log_max_count.load(Ordering::SeqCst);
let writer_config = WriterConfig {
directory: service_log_dir,
max_log_size: log_max_size * 1024,
max_log_files: log_max_count,
};
Ok(writer_config)
}
}
src-tauri/src/core/manager/config.rs
+22 -1
View File
@@ -1,4 +1,4 @@
use super::CoreManager;
use super::{CoreManager, RunningMode};
use crate::{
config::{Config, ConfigType, runtime::IRuntime},
constants::timing,
@@ -136,6 +136,27 @@ impl CoreManager {
}
async fn apply_config(&self, path: PathBuf) -> Result<()> {
if matches!(*self.get_running_mode(), RunningMode::Service) {
let _lifecycle = self.lifecycle_lock.lock().await;
if !matches!(*self.get_running_mode(), RunningMode::Service) {
Config::runtime().await.discard();
return Err(anyhow!("core mode changed while applying service configuration"));
}
let result = self.replace_service_core_with_config(&path).await;
return match result {
Ok(()) => {
Config::runtime().await.apply();
logging!(info, Type::Core, "Configuration materialized and applied by service");
Ok(())
}
Err(error) => {
Config::runtime().await.discard();
Err(error)
}
};
}
let path = dirs::path_to_str(&path)?;
match self.reload_config(path).await {
Ok(_) => {
src-tauri/src/core/manager/lifecycle.rs
+1044 -43
View File
File diff suppressed because it is too large Load Diff
src-tauri/src/core/manager/mod.rs
+74 -5
View File
@@ -10,7 +10,7 @@ use std::{
fmt,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
atomic::{AtomicBool, AtomicU64, Ordering},
},
time::Instant,
};
@@ -22,7 +22,37 @@ use std::os::windows::io::OwnedHandle;
pub(crate) static CLASH_LOGGER: Lazy<Arc<AsyncLogger>> = Lazy::new(|| Arc::new(AsyncLogger::new()));
#[derive(Debug, serde::Serialize, PartialEq, Eq)]
const CORE_READINESS_ACTIVE_BIT: u64 = 1;
const CORE_READINESS_GENERATION_STEP: u64 = 1 << 1;
const fn next_active_core_readiness_state(state: u64) -> u64 {
(state.wrapping_add(CORE_READINESS_GENERATION_STEP) & !CORE_READINESS_ACTIVE_BIT) | CORE_READINESS_ACTIVE_BIT
}
const fn inactive_core_readiness_state(state: u64) -> u64 {
state.wrapping_add(CORE_READINESS_GENERATION_STEP) & !CORE_READINESS_ACTIVE_BIT
}
const fn active_core_readiness_generation(state: u64) -> Option<u64> {
if state & CORE_READINESS_ACTIVE_BIT != 0 {
Some(state)
} else {
None
}
}
fn claim_core_readiness_generation(state: &AtomicU64, captured_generation: u64) -> bool {
state
.compare_exchange(
captured_generation,
inactive_core_readiness_state(captured_generation),
Ordering::AcqRel,
Ordering::Acquire,
)
.is_ok()
}
#[derive(Clone, Copy, Debug, serde::Serialize, PartialEq, Eq)]
pub enum RunningMode {
Service,
Sidecar,
@@ -46,9 +76,10 @@ pub struct CoreManager {
#[cfg(target_os = "windows")]
job_handle: ArcSwapOption<OwnedHandle>,
config_update_in_progress: AtomicBool,
core_readiness_state: AtomicU64,
// 串行化 start/stop/restart 和 sidecar→service 交接。
// 锁序固定为 config_update_in_progress → lifecycle_lock。
lifecycle_lock: tokio::sync::Mutex<()>,
pub(crate) lifecycle_lock: tokio::sync::Mutex<()>,
// sidecar→service 交接 watcher 单实例标志。
#[cfg(target_os = "windows")]
handoff_watcher_running: AtomicBool,
@@ -77,6 +108,7 @@ impl Default for CoreManager {
#[cfg(target_os = "windows")]
job_handle: ArcSwapOption::new(None),
config_update_in_progress: AtomicBool::new(false),
core_readiness_state: AtomicU64::new(0),
lifecycle_lock: tokio::sync::Mutex::new(()),
#[cfg(target_os = "windows")]
handoff_watcher_running: AtomicBool::new(false),
@@ -101,11 +133,18 @@ impl CoreManager {
.and_then(|arc| Arc::try_unwrap(arc).ok())
}
pub fn get_running_sidecar_pid(&self) -> Option<u32> {
self.state.load().child_sidecar.load_full().map(|child| child.pid())
}
pub fn get_last_update(&self) -> Option<Arc<Instant>> {
self.last_update.load_full()
}
pub fn set_running_mode(&self, mode: RunningMode) {
if matches!(mode, RunningMode::NotRunning) {
self.invalidate_core_readiness();
}
let state = self.state.load();
state.running_mode.store(Arc::new(mode));
}
@@ -115,6 +154,36 @@ impl CoreManager {
state.child_sidecar.store(Some(Arc::new(child)));
}
fn mark_core_ready(&self) -> u64 {
let mut current = self.core_readiness_state.load(Ordering::Acquire);
loop {
let next = next_active_core_readiness_state(current);
match self
.core_readiness_state
.compare_exchange(current, next, Ordering::AcqRel, Ordering::Acquire)
{
Ok(_) => return next,
Err(changed) => current = changed,
}
}
}
fn current_core_readiness_generation(&self) -> Option<u64> {
active_core_readiness_generation(self.core_readiness_state.load(Ordering::Acquire))
}
pub(crate) fn invalidate_core_readiness(&self) {
let _ = self
.core_readiness_state
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
active_core_readiness_generation(current).map(inactive_core_readiness_state)
});
}
fn invalidate_core_readiness_if(&self, captured_generation: u64) -> bool {
claim_core_readiness_generation(&self.core_readiness_state, captured_generation)
}
pub fn set_last_update(&self, time: Instant) {
self.last_update.store(Some(Arc::new(time)));
}
@@ -138,9 +207,9 @@ impl CoreManager {
self.config_update_in_progress.store(false, Ordering::Release);
}
pub async fn init(&self) -> Result<()> {
pub async fn init(&self) -> Result<bool> {
self.start_core().await?;
Ok(())
Ok(!matches!(*self.get_running_mode(), RunningMode::NotRunning))
}
}
src-tauri/src/core/manager/state.rs
+213 -35
View File
@@ -1,18 +1,54 @@
#[cfg(test)]
use super::claim_core_readiness_generation;
use super::{CoreManager, RunningMode};
use crate::{
AsyncHandler,
config::{Config, IClashTemp},
core::{handle, logger::Logger, manager::CLASH_LOGGER, service},
config::Config,
core::{handle, logger::Logger, manager::CLASH_LOGGER, proxy_control, service},
logging,
utils::dirs,
utils::{dirs, server},
};
use anyhow::Result;
use anyhow::{Context as _, Result};
use clash_verge_logging::Type;
use compact_str::CompactString;
use log::Level;
use scopeguard::defer;
use std::path::Path;
use tauri_plugin_mihomo::MihomoExt as _;
use tauri_plugin_shell::ShellExt as _;
const SIDECAR_READINESS_ATTEMPTS: usize = 30;
const SIDECAR_READINESS_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100);
const SIDECAR_READINESS_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(400);
async fn poll_sidecar_readiness<F, Fut>(
max_attempts: usize,
retry_delay: std::time::Duration,
mut probe: F,
) -> Result<()>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<()>>,
{
let mut last_error = None;
for attempt in 0..max_attempts {
match probe().await {
Ok(()) => return Ok(()),
Err(error) => last_error = Some(error),
}
if attempt + 1 < max_attempts {
tokio::time::sleep(retry_delay).await;
}
}
Err(last_error
.unwrap_or_else(|| anyhow::anyhow!("sidecar readiness was configured with no attempts"))
.context("Mihomo API did not become ready"))
}
fn should_clear_terminated_sidecar(running_mode: &RunningMode, current_pid: Option<u32>, terminated_pid: u32) -> bool {
matches!(running_mode, RunningMode::Sidecar) && current_pid == Some(terminated_pid)
}
#[cfg(target_os = "windows")]
use {
std::os::windows::io::{AsRawHandle as _, FromRawHandle as _, OwnedHandle},
@@ -39,33 +75,43 @@ impl CoreManager {
pub(super) async fn start_core_by_sidecar(&self) -> Result<()> {
logging!(info, Type::Core, "Starting core in sidecar mode");
server::set_pac_available(false);
self.set_running_mode(RunningMode::NotRunning);
let sidecar_ipc = dirs::sidecar_ipc_path()?;
handle::Handle::app_handle()
.mihomo()
.write()
.await
.update_socket_path(dirs::path_to_str(&sidecar_ipc)?.to_owned())?;
let config_file = Config::generate_file(crate::config::ConfigType::Run).await?;
let app_handle = handle::Handle::app_handle();
let clash_core = Config::verge().await.latest_arc().get_valid_clash_core();
let config_dir = dirs::app_home_dir()?;
#[cfg(unix)]
let previous_mask = unsafe { tauri_plugin_clash_verge_sysinfo::libc::umask(0o007) };
let (mut rx, child) = app_handle
.shell()
.sidecar(clash_core.as_str())?
.args([
"-d",
dirs::path_to_str(&config_dir)?,
"-f",
dirs::path_to_str(&config_file)?,
if cfg!(windows) {
"-ext-ctl-pipe"
} else {
"-ext-ctl-unix"
},
&IClashTemp::guard_external_controller_ipc(),
])
.spawn()?;
let previous_mask = unsafe { tauri_plugin_clash_verge_sysinfo::libc::umask(0o077) };
let command = app_handle.shell().sidecar(clash_core.as_str())?.args([
"-d",
dirs::path_to_str(&config_dir)?,
"-f",
dirs::path_to_str(&config_file)?,
if cfg!(windows) {
"-ext-ctl-pipe"
} else {
"-ext-ctl-unix"
},
dirs::path_to_str(&sidecar_ipc)?,
]);
#[cfg(windows)]
let command = command.env(
"LISTEN_NAMEDPIPE_SDDL",
crate::core::owner_identity::current_user_pipe_sddl()?,
);
let (mut rx, child) = command.spawn()?;
#[cfg(target_os = "windows")]
{
let job = match create_and_assign_sidecar_job(child.pid()) {
let job = {
match create_and_assign_sidecar_job(child.pid()) {
Ok(job) => job,
Err(job_error) => {
let pid = child.pid();
@@ -81,9 +127,8 @@ impl CoreManager {
logging!(error, Type::Core, "Failed to start sidecar: {error:#}");
return Err(error);
}
};
self.set_job_handle(Some(job));
}
}
};
#[cfg(unix)]
unsafe {
@@ -93,10 +138,36 @@ impl CoreManager {
let pid = child.pid();
logging!(trace, Type::Core, "Sidecar started with PID: {}", pid);
self.set_running_child_sidecar(child);
self.set_running_mode(RunningMode::Sidecar);
let readiness = poll_sidecar_readiness(SIDECAR_READINESS_ATTEMPTS, SIDECAR_READINESS_INTERVAL, || async {
tokio::time::timeout(SIDECAR_READINESS_PROBE_TIMEOUT, async {
handle::Handle::mihomo().await.get_version().await
})
.await
.context("Mihomo readiness probe timed out")??;
Ok(())
})
.await;
if let Err(readiness_error) = readiness {
proxy_control::stop_guard().await;
self.invalidate_core_readiness();
self.set_running_mode(RunningMode::NotRunning);
server::set_pac_available(false);
return match child.kill() {
Ok(()) => Err(readiness_error),
Err(kill_error) => Err(anyhow::anyhow!(
"{readiness_error:#}; failed to terminate unready sidecar PID {pid}: {kill_error:#}"
)),
};
}
AsyncHandler::spawn(|| async move {
#[cfg(target_os = "windows")]
self.set_job_handle(Some(job));
self.set_running_child_sidecar(child);
let core_readiness_generation = self.mark_core_ready();
self.set_running_mode(RunningMode::Sidecar);
server::set_pac_available(true);
AsyncHandler::spawn(move || async move {
while let Some(event) = rx.recv().await {
match event {
tauri_plugin_shell::process::CommandEvent::Stdout(line)
@@ -106,6 +177,8 @@ impl CoreManager {
CLASH_LOGGER.append_log(message).await;
}
tauri_plugin_shell::process::CommandEvent::Terminated(term) => {
let manager = Self::global();
let _ = manager.invalidate_core_readiness_if(core_readiness_generation);
let message = if let Some(code) = term.code {
CompactString::from(format!("Process terminated with code: {}", code))
} else if let Some(signal) = term.signal {
@@ -115,6 +188,7 @@ impl CoreManager {
};
Logger::global().writer_sidecar_log(Level::Info, &message);
CLASH_LOGGER.clear_logs().await;
manager.clear_terminated_sidecar(pid).await;
break;
}
_ => {}
@@ -125,9 +199,12 @@ impl CoreManager {
Ok(())
}
pub(super) fn stop_core_by_sidecar(&self) {
/// Terminates the sidecar after its caller has successfully cleared the
/// system proxy.
pub(super) fn stop_core_by_sidecar_unprepared(&self) {
logging!(info, Type::Core, "Stopping sidecar");
defer! {
server::set_pac_available(false);
self.set_running_mode(RunningMode::NotRunning);
}
if let Some(child) = self.take_child_sidecar() {
@@ -159,16 +236,28 @@ impl CoreManager {
pub(super) async fn start_core_by_service(&self) -> Result<()> {
logging!(info, Type::Core, "Starting core in service mode");
server::set_pac_available(false);
let service_ipc = dirs::ipc_path()?;
let config_file = Config::generate_file(crate::config::ConfigType::Run).await?;
handle::Handle::app_handle()
.mihomo()
.write()
.await
.update_socket_path(dirs::path_to_str(&service_ipc)?.to_owned())?;
self.start_core_by_service_with_config(&config_file).await
}
pub(super) async fn start_core_by_service_with_config(&self, config_file: &Path) -> Result<()> {
// 交接时等待 sidecar 释放 ext-controller 通道。
#[cfg(target_os = "windows")]
{
use crate::constants::timing;
let mut last_err = None;
for attempt in 0..timing::SERVICE_START_RETRIES {
match service::run_core_by_service(&config_file).await {
match service::run_core_by_service(config_file).await {
Ok(()) => {
self.mark_core_ready();
self.set_running_mode(RunningMode::Service);
return Ok(());
}
@@ -191,7 +280,8 @@ impl CoreManager {
#[cfg(not(target_os = "windows"))]
{
service::run_core_by_service(&config_file).await?;
service::run_core_by_service(config_file).await?;
self.mark_core_ready();
self.set_running_mode(RunningMode::Service);
Ok(())
}
@@ -199,12 +289,100 @@ impl CoreManager {
pub(super) async fn stop_core_by_service(&self) -> Result<()> {
logging!(info, Type::Core, "Stopping service");
defer! {
self.set_running_mode(RunningMode::NotRunning);
}
service::stop_core_by_service().await?;
server::set_pac_available(false);
self.set_running_mode(RunningMode::NotRunning);
Ok(())
}
async fn clear_terminated_sidecar(&self, terminated_pid: u32) {
let _life = self.lifecycle_lock.lock().await;
if !should_clear_terminated_sidecar(&self.get_running_mode(), self.get_running_sidecar_pid(), terminated_pid) {
return;
}
let _ = self.take_child_sidecar();
#[cfg(target_os = "windows")]
self.set_job_handle(None);
proxy_control::stop_guard().await;
server::set_pac_available(false);
self.invalidate_core_readiness();
self.set_running_mode(RunningMode::NotRunning);
self.after_core_process();
}
}
#[cfg(test)]
mod readiness_tests {
use super::{claim_core_readiness_generation, poll_sidecar_readiness, should_clear_terminated_sidecar};
use crate::core::manager::{CoreManager, RunningMode};
use std::{
sync::{
Arc,
atomic::{AtomicU64, AtomicUsize, Ordering},
},
time::Duration,
};
#[tokio::test]
async fn sidecar_readiness_poll_is_bounded_and_accepts_a_real_api_response() -> anyhow::Result<()> {
let attempts = Arc::new(AtomicUsize::new(0));
let probe_attempts = Arc::clone(&attempts);
poll_sidecar_readiness(3, Duration::ZERO, move || {
let attempt = probe_attempts.fetch_add(1, Ordering::SeqCst) + 1;
async move {
if attempt == 3 {
Ok(())
} else {
Err(anyhow::anyhow!("not ready"))
}
}
})
.await?;
assert_eq!(attempts.load(Ordering::SeqCst), 3);
let failed_attempts = Arc::new(AtomicUsize::new(0));
let probe_attempts = Arc::clone(&failed_attempts);
assert!(
poll_sidecar_readiness(3, Duration::ZERO, move || {
probe_attempts.fetch_add(1, Ordering::SeqCst);
async { Err(anyhow::anyhow!("still unavailable")) }
})
.await
.is_err()
);
assert_eq!(failed_attempts.load(Ordering::SeqCst), 3);
Ok(())
}
#[test]
fn only_the_current_sidecar_termination_clears_local_state() {
assert!(should_clear_terminated_sidecar(&RunningMode::Sidecar, Some(42), 42));
assert!(!should_clear_terminated_sidecar(&RunningMode::Sidecar, Some(43), 42));
assert!(!should_clear_terminated_sidecar(&RunningMode::Service, Some(42), 42));
assert!(!should_clear_terminated_sidecar(&RunningMode::NotRunning, None, 42));
}
#[test]
fn core_readiness_generation_can_only_be_claimed_once() {
let generation = AtomicU64::new(7);
assert!(claim_core_readiness_generation(&generation, 7));
assert_eq!(generation.load(Ordering::Acquire), 8);
assert!(!claim_core_readiness_generation(&generation, 7));
}
#[test]
fn invalidated_core_readiness_cannot_be_recaptured_from_stale_mode() {
let manager = CoreManager::default();
manager.mark_core_ready();
manager.set_running_mode(RunningMode::Service);
manager.invalidate_core_readiness();
assert_eq!(*manager.get_running_mode(), RunningMode::Service);
assert_eq!(manager.current_core_readiness_generation(), None);
}
}
#[cfg(target_os = "windows")]
src-tauri/src/core/mod.rs
+3
View File
@@ -5,7 +5,10 @@ pub mod hotkey;
pub mod logger;
pub mod manager;
mod notification;
pub(crate) mod owner_identity;
pub mod proxy_control;
pub mod proxy_view;
mod runtime_bundle;
pub mod service;
pub mod sysopt;
pub mod timer;
src-tauri/src/core/owner_identity.rs
+575
View File
@@ -0,0 +1,575 @@
use crate::utils::dirs;
use anyhow::{Context as _, Result};
use clash_verge_service_ipc::{OwnerCredentials, OwnerIdentity};
use std::path::Path;
pub(crate) fn current_owner_credentials() -> Result<OwnerCredentials> {
current_owner_credentials_for_root(&dirs::app_home_dir()?)
}
#[allow(clippy::unnecessary_wraps)] // Windows SID discovery is fallible; Unix keeps the shared API.
pub(crate) fn current_owner_identity() -> Result<OwnerIdentity> {
#[cfg(unix)]
return Ok(OwnerIdentity::Unix {
uid: unsafe { libc::geteuid() },
gid: unsafe { libc::getegid() },
});
#[cfg(windows)]
{
windows_current_identity()
}
}
pub(crate) fn current_owner_credentials_for_root(app_root: &Path) -> Result<OwnerCredentials> {
let app_data_root = std::fs::canonicalize(app_root)
.with_context(|| format!("failed to canonicalize application data root {app_root:?}"))?;
#[cfg(unix)]
let (identity, token) = (current_owner_identity()?, None);
#[cfg(windows)]
let (identity, token) = windows_owner_credentials(&app_data_root)?;
Ok(OwnerCredentials {
identity,
app_data_dir: app_data_root.to_string_lossy().into_owned(),
token,
})
}
#[cfg(windows)]
fn windows_owner_credentials(app_data_root: &Path) -> Result<(OwnerIdentity, Option<String>)> {
let sid = windows_owner::current_sid()?;
let token = windows_owner::load_or_create_token(app_data_root, &sid)?;
Ok((OwnerIdentity::Windows { sid }, Some(token)))
}
#[cfg(windows)]
fn windows_current_identity() -> Result<OwnerIdentity> {
Ok(OwnerIdentity::Windows {
sid: windows_owner::current_sid()?,
})
}
#[cfg(windows)]
pub(crate) fn create_private_current_user_file(path: &Path) -> Result<std::fs::File> {
windows_owner::create_private_file(path)
}
#[cfg(windows)]
pub(crate) fn open_private_current_user_file(path: &Path) -> Result<std::fs::File> {
windows_owner::open_private_file(path)
}
#[cfg(windows)]
pub(crate) fn open_or_create_private_current_user_file(path: &Path) -> Result<std::fs::File> {
windows_owner::open_or_create_private_file(path)
}
#[cfg(windows)]
pub(crate) fn current_user_pipe_sddl() -> Result<String> {
let sid = windows_owner::current_sid()?;
Ok(format!("D:P(A;;GA;;;{sid})(A;;GA;;;SY)(A;;GA;;;BA)"))
}
#[cfg(windows)]
mod windows_owner {
use anyhow::{Context as _, Result, bail};
use clash_verge_service_ipc::OWNER_TOKEN_FILE_NAME;
use std::ffi::c_void;
use std::io::{Read as _, Write as _};
use std::os::windows::ffi::OsStrExt as _;
use std::os::windows::io::{AsRawHandle as _, FromRawHandle as _};
use std::path::Path;
use windows_sys::Win32::Foundation::{
CloseHandle, ERROR_FILE_EXISTS, GENERIC_READ, GENERIC_WRITE, GetLastError, INVALID_HANDLE_VALUE, LocalFree,
};
use windows_sys::Win32::Security::Authorization::{
ConvertSidToStringSidW, ConvertStringSecurityDescriptorToSecurityDescriptorW, GetSecurityInfo, SDDL_REVISION_1,
SE_FILE_OBJECT, SetSecurityInfo,
};
use windows_sys::Win32::Security::{
ACCESS_ALLOWED_ACE, ACL, DACL_SECURITY_INFORMATION, EqualSid, GetAce, GetSecurityDescriptorControl,
GetSecurityDescriptorDacl, GetSecurityDescriptorOwner, GetTokenInformation, IsValidSid, IsWellKnownSid,
OWNER_SECURITY_INFORMATION, PROTECTED_DACL_SECURITY_INFORMATION, PSECURITY_DESCRIPTOR, PSID, SE_DACL_PROTECTED,
SECURITY_ATTRIBUTES, TOKEN_QUERY, TOKEN_USER, TokenUser, WinBuiltinAdministratorsSid, WinLocalSystemSid,
};
use windows_sys::Win32::Storage::FileSystem::{
BY_HANDLE_FILE_INFORMATION, CREATE_NEW, CreateFileW, FILE_ALL_ACCESS, FILE_ATTRIBUTE_DIRECTORY,
FILE_ATTRIBUTE_HIDDEN, FILE_ATTRIBUTE_NORMAL, FILE_ATTRIBUTE_REPARSE_POINT, FILE_FLAG_OPEN_REPARSE_POINT,
FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE, FILE_TYPE_DISK, GetFileInformationByHandle, GetFileType,
OPEN_EXISTING, READ_CONTROL, WRITE_DAC,
};
use windows_sys::Win32::System::Threading::{GetCurrentProcess, OpenProcessToken};
const TOKEN_BYTES: usize = 32;
const ACCESS_ALLOWED_ACE_TYPE: u8 = 0;
pub(super) fn current_sid() -> Result<String> {
let mut token = std::ptr::null_mut();
if unsafe { OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &mut token) } == 0 {
return Err(std::io::Error::last_os_error()).context("failed to open process token");
}
let token = OwnedHandle(token);
let mut required = 0_u32;
unsafe { GetTokenInformation(token.0, TokenUser, std::ptr::null_mut(), 0, &mut required) };
if required == 0 {
return Err(std::io::Error::last_os_error()).context("failed to size process SID buffer");
}
let words = (required as usize).div_ceil(std::mem::size_of::<usize>());
let mut buffer = vec![0_usize; words];
if unsafe { GetTokenInformation(token.0, TokenUser, buffer.as_mut_ptr().cast(), required, &mut required) } == 0
{
return Err(std::io::Error::last_os_error()).context("failed to read process SID");
}
let token_user = unsafe { &*buffer.as_ptr().cast::<TOKEN_USER>() };
sid_to_string(token_user.User.Sid)
}
pub(super) fn load_or_create_token(app_data_root: &Path, sid: &str) -> Result<String> {
let token_path = app_data_root.join(OWNER_TOKEN_FILE_NAME);
let descriptor = LocalSecurityDescriptor::from_sid(sid)?;
let attributes = SECURITY_ATTRIBUTES {
nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
lpSecurityDescriptor: descriptor.0,
bInheritHandle: 0,
};
let wide = wide_path(&token_path)?;
let handle = unsafe {
CreateFileW(
wide.as_ptr(),
GENERIC_READ | GENERIC_WRITE | READ_CONTROL | WRITE_DAC,
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
&attributes,
CREATE_NEW,
FILE_ATTRIBUTE_HIDDEN | FILE_FLAG_OPEN_REPARSE_POINT,
std::ptr::null_mut(),
)
};
if handle != INVALID_HANDLE_VALUE {
let mut file = unsafe { std::fs::File::from_raw_handle(handle) };
let mut token = [0_u8; TOKEN_BYTES];
getrandom::fill(&mut token).context("failed to generate owner token")?;
file.write_all(&token).context("failed to write owner token")?;
file.sync_all().context("failed to flush owner token")?;
return Ok(encode_token(&token));
}
if unsafe { GetLastError() } != ERROR_FILE_EXISTS {
return Err(std::io::Error::last_os_error()).context("failed to create owner token");
}
let handle = unsafe {
CreateFileW(
wide.as_ptr(),
GENERIC_READ | READ_CONTROL | WRITE_DAC,
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
std::ptr::null(),
OPEN_EXISTING,
FILE_FLAG_OPEN_REPARSE_POINT,
std::ptr::null_mut(),
)
};
if handle == INVALID_HANDLE_VALUE {
return Err(std::io::Error::last_os_error()).context("failed to open owner token");
}
let mut file = unsafe { std::fs::File::from_raw_handle(handle) };
validate_token_file(&file, descriptor.owner()?)?;
descriptor.apply_dacl(file.as_raw_handle())?;
let mut token = [0_u8; TOKEN_BYTES];
file.read_exact(&mut token).context("failed to read owner token")?;
Ok(encode_token(&token))
}
pub(super) fn create_private_file(path: &Path) -> Result<std::fs::File> {
let sid = current_sid()?;
let descriptor = LocalSecurityDescriptor::from_sid(&sid)?;
create_file(path, &descriptor)
}
pub(super) fn open_private_file(path: &Path) -> Result<std::fs::File> {
let sid = current_sid()?;
let descriptor = LocalSecurityDescriptor::from_sid(&sid)?;
let file = open_file(path)?;
validate_private_file(&file, descriptor.owner()?)?;
Ok(file)
}
pub(super) fn open_or_create_private_file(path: &Path) -> Result<std::fs::File> {
let sid = current_sid()?;
let descriptor = LocalSecurityDescriptor::from_sid(&sid)?;
match create_file(path, &descriptor) {
Ok(file) => Ok(file),
Err(error)
if error
.downcast_ref::<std::io::Error>()
.is_some_and(|error| error.raw_os_error() == Some(ERROR_FILE_EXISTS as i32)) =>
{
let file = open_file(path)?;
validate_private_file(&file, descriptor.owner()?)?;
Ok(file)
}
Err(error) => Err(error),
}
}
fn create_file(path: &Path, descriptor: &LocalSecurityDescriptor) -> Result<std::fs::File> {
let attributes = SECURITY_ATTRIBUTES {
nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
lpSecurityDescriptor: descriptor.0,
bInheritHandle: 0,
};
let wide = wide_path(path)?;
let handle = unsafe {
CreateFileW(
wide.as_ptr(),
GENERIC_READ | GENERIC_WRITE | READ_CONTROL | WRITE_DAC,
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
&attributes,
CREATE_NEW,
FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OPEN_REPARSE_POINT,
std::ptr::null_mut(),
)
};
if handle == INVALID_HANDLE_VALUE {
return Err(std::io::Error::last_os_error()).context("failed to create private current-user file");
}
let file = unsafe { std::fs::File::from_raw_handle(handle) };
validate_private_file(&file, descriptor.owner()?)?;
Ok(file)
}
fn open_file(path: &Path) -> Result<std::fs::File> {
let wide = wide_path(path)?;
let handle = unsafe {
CreateFileW(
wide.as_ptr(),
GENERIC_READ | GENERIC_WRITE | READ_CONTROL | WRITE_DAC,
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
std::ptr::null(),
OPEN_EXISTING,
FILE_FLAG_OPEN_REPARSE_POINT,
std::ptr::null_mut(),
)
};
if handle == INVALID_HANDLE_VALUE {
return Err(std::io::Error::last_os_error()).context("failed to open private current-user file");
}
Ok(unsafe { std::fs::File::from_raw_handle(handle) })
}
fn validate_private_file(file: &std::fs::File, expected_owner: PSID) -> Result<()> {
let handle = file.as_raw_handle();
let mut information = BY_HANDLE_FILE_INFORMATION::default();
if unsafe { GetFileInformationByHandle(handle, &mut information) } == 0 {
return Err(std::io::Error::last_os_error()).context("failed to inspect private file metadata");
}
if information.dwFileAttributes & (FILE_ATTRIBUTE_DIRECTORY | FILE_ATTRIBUTE_REPARSE_POINT) != 0
|| unsafe { GetFileType(handle) } != FILE_TYPE_DISK
{
bail!("current-user file is not an ordinary file");
}
let mut owner = std::ptr::null_mut();
let mut dacl: *mut ACL = std::ptr::null_mut();
let mut security = std::ptr::null_mut();
let status = unsafe {
GetSecurityInfo(
handle,
SE_FILE_OBJECT,
OWNER_SECURITY_INFORMATION | DACL_SECURITY_INFORMATION,
&mut owner,
std::ptr::null_mut(),
&mut dacl,
std::ptr::null_mut(),
&mut security,
)
};
if status != 0 || security.is_null() {
bail!("failed to inspect private file security: Windows error {status}");
}
let security = LocalSecurityDescriptor(security);
if owner.is_null() || unsafe { EqualSid(owner, expected_owner) } == 0 || dacl.is_null() {
bail!("current-user file owner or DACL is invalid");
}
let mut control = 0_u16;
let mut revision = 0_u32;
if unsafe { GetSecurityDescriptorControl(security.0, &mut control, &mut revision) } == 0
|| control & SE_DACL_PROTECTED == 0
{
bail!("current-user file DACL is not protected");
}
let mut owner_ace = false;
let mut system_ace = false;
let mut administrators_ace = false;
for index in 0..unsafe { (*dacl).AceCount } as u32 {
let mut ace = std::ptr::null_mut();
if unsafe { GetAce(dacl, index, &mut ace) } == 0 || ace.is_null() {
bail!("current-user file DACL could not be inspected");
}
let allowed = unsafe { &*ace.cast::<ACCESS_ALLOWED_ACE>() };
if allowed.Header.AceType != ACCESS_ALLOWED_ACE_TYPE || allowed.Mask & FILE_ALL_ACCESS != FILE_ALL_ACCESS {
bail!("current-user file DACL contains an unexpected ACE");
}
let ace_sid = std::ptr::addr_of!(allowed.SidStart).cast_mut().cast::<c_void>();
if unsafe { IsValidSid(ace_sid) } == 0 {
bail!("current-user file DACL contains an invalid SID");
}
if unsafe { EqualSid(ace_sid, expected_owner) } != 0 {
owner_ace = true;
} else if unsafe { IsWellKnownSid(ace_sid, WinLocalSystemSid) } != 0 {
system_ace = true;
} else if unsafe { IsWellKnownSid(ace_sid, WinBuiltinAdministratorsSid) } != 0 {
administrators_ace = true;
} else {
bail!("current-user file DACL grants access to another SID");
}
}
if unsafe { (*dacl).AceCount } != 3 || !owner_ace || !system_ace || !administrators_ace {
bail!("current-user file DACL is missing a required principal");
}
Ok(())
}
fn validate_token_file(file: &std::fs::File, expected_owner: PSID) -> Result<()> {
let handle = file.as_raw_handle();
let mut information = BY_HANDLE_FILE_INFORMATION::default();
if unsafe { GetFileInformationByHandle(handle, &mut information) } == 0 {
return Err(std::io::Error::last_os_error()).context("failed to inspect owner token metadata");
}
if information.dwFileAttributes & (FILE_ATTRIBUTE_DIRECTORY | FILE_ATTRIBUTE_REPARSE_POINT) != 0
|| unsafe { GetFileType(handle) } != FILE_TYPE_DISK
|| information.nFileSizeHigh != 0
|| information.nFileSizeLow != TOKEN_BYTES as u32
{
bail!("owner token is not an ordinary 32-byte file");
}
let mut owner = std::ptr::null_mut();
let mut security = std::ptr::null_mut();
let status = unsafe {
GetSecurityInfo(
handle,
SE_FILE_OBJECT,
OWNER_SECURITY_INFORMATION,
&mut owner,
std::ptr::null_mut(),
std::ptr::null_mut(),
std::ptr::null_mut(),
&mut security,
)
};
if status != 0 || security.is_null() {
bail!("failed to inspect owner token security: Windows error {status}");
}
let security = LocalSecurityDescriptor(security);
if owner.is_null() || unsafe { EqualSid(owner, expected_owner) } == 0 {
bail!("owner token belongs to a different Windows user");
}
drop(security);
Ok(())
}
fn sid_to_string(sid: PSID) -> Result<String> {
let mut value = std::ptr::null_mut();
if unsafe { ConvertSidToStringSidW(sid, &mut value) } == 0 || value.is_null() {
return Err(std::io::Error::last_os_error()).context("failed to format process SID");
}
let value = LocalWideString(value);
let length = (0..).take_while(|index| unsafe { *value.0.add(*index) } != 0).count();
String::from_utf16(unsafe { std::slice::from_raw_parts(value.0, length) })
.context("process SID is not valid UTF-16")
}
fn encode_token(token: &[u8; TOKEN_BYTES]) -> String {
let mut encoded = String::with_capacity(TOKEN_BYTES * 2);
for byte in token {
use std::fmt::Write as _;
let _ = write!(encoded, "{byte:02x}");
}
encoded
}
fn wide_path(path: &Path) -> Result<Vec<u16>> {
let mut wide: Vec<u16> = path.as_os_str().encode_wide().collect();
if wide.contains(&0) {
bail!("owner token path contains NUL");
}
wide.push(0);
Ok(wide)
}
struct OwnedHandle(*mut c_void);
impl Drop for OwnedHandle {
fn drop(&mut self) {
if !self.0.is_null() && self.0 != INVALID_HANDLE_VALUE {
unsafe { CloseHandle(self.0) };
}
}
}
struct LocalWideString(*mut u16);
impl Drop for LocalWideString {
fn drop(&mut self) {
if !self.0.is_null() {
unsafe { LocalFree(self.0.cast()) };
}
}
}
struct LocalSecurityDescriptor(PSECURITY_DESCRIPTOR);
impl LocalSecurityDescriptor {
fn from_sid(sid: &str) -> Result<Self> {
let sddl = format!("O:{sid}D:P(A;;FA;;;{sid})(A;;FA;;;SY)(A;;FA;;;BA)");
let mut wide: Vec<u16> = sddl.encode_utf16().collect();
wide.push(0);
let mut descriptor = std::ptr::null_mut();
if unsafe {
ConvertStringSecurityDescriptorToSecurityDescriptorW(
wide.as_ptr(),
SDDL_REVISION_1,
&mut descriptor,
std::ptr::null_mut(),
)
} == 0
|| descriptor.is_null()
{
return Err(std::io::Error::last_os_error()).context("failed to build owner token security descriptor");
}
Ok(Self(descriptor))
}
fn owner(&self) -> Result<PSID> {
let mut owner = std::ptr::null_mut();
let mut defaulted = 0;
if unsafe { GetSecurityDescriptorOwner(self.0, &mut owner, &mut defaulted) } == 0 || owner.is_null() {
return Err(std::io::Error::last_os_error()).context("failed to read token descriptor owner");
}
Ok(owner)
}
fn apply_dacl(&self, handle: *mut c_void) -> Result<()> {
let mut present = 0;
let mut defaulted = 0;
let mut dacl = std::ptr::null_mut();
if unsafe { GetSecurityDescriptorDacl(self.0, &mut present, &mut dacl, &mut defaulted) } == 0
|| present == 0
|| dacl.is_null()
{
return Err(std::io::Error::last_os_error()).context("failed to read token descriptor DACL");
}
let status = unsafe {
SetSecurityInfo(
handle,
SE_FILE_OBJECT,
DACL_SECURITY_INFORMATION | PROTECTED_DACL_SECURITY_INFORMATION,
std::ptr::null_mut(),
std::ptr::null_mut(),
dacl,
std::ptr::null(),
)
};
if status != 0 {
bail!("failed to restrict owner token DACL: Windows error {status}");
}
Ok(())
}
}
impl Drop for LocalSecurityDescriptor {
fn drop(&mut self) {
if !self.0.is_null() {
unsafe { LocalFree(self.0) };
}
}
}
}
#[cfg(test)]
mod tests {
use super::current_owner_credentials_for_root;
use clash_verge_service_ipc::OwnerIdentity;
#[cfg(unix)]
#[test]
fn unix_credentials_come_from_process_and_have_no_token() -> anyhow::Result<()> {
let app_root = std::env::temp_dir();
let credentials = current_owner_credentials_for_root(&app_root)?;
assert_eq!(
credentials.identity,
OwnerIdentity::Unix {
uid: unsafe { libc::geteuid() },
gid: unsafe { libc::getegid() },
}
);
assert_eq!(credentials.token, None);
assert_eq!(
std::path::PathBuf::from(credentials.app_data_dir),
std::fs::canonicalize(app_root)?
);
Ok(())
}
#[cfg(windows)]
#[test]
fn windows_credentials_use_stable_sid_and_private_token() -> anyhow::Result<()> {
let app_root = std::env::temp_dir().join(format!("cvr-owner-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&app_root);
std::fs::create_dir(&app_root)?;
let first = current_owner_credentials_for_root(&app_root)?;
let second = current_owner_credentials_for_root(&app_root)?;
let OwnerIdentity::Windows { sid } = &first.identity else {
anyhow::bail!("expected Windows owner identity");
};
assert!(sid.starts_with("S-1-5-"));
assert_eq!(first.identity, second.identity);
assert_eq!(first.token, second.token);
assert_eq!(first.token.as_deref().map(str::len), Some(64));
std::fs::remove_dir_all(app_root)?;
Ok(())
}
#[cfg(windows)]
#[test]
fn private_file_open_rejects_directories_and_reparse_points() -> anyhow::Result<()> {
use super::{create_private_current_user_file, open_private_current_user_file};
let root = std::env::temp_dir().join(format!("cvr-private-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir(&root)?;
let private_file = root.join("record.json");
drop(create_private_current_user_file(&private_file)?);
drop(open_private_current_user_file(&private_file)?);
let directory = root.join("directory");
std::fs::create_dir(&directory)?;
assert!(open_private_current_user_file(&directory).is_err());
let link = root.join("record-link.json");
if let Err(error) = std::os::windows::fs::symlink_file(&private_file, &link) {
if error.raw_os_error() == Some(1314) {
std::fs::remove_dir_all(root)?;
return Ok(());
}
return Err(error.into());
}
assert!(open_private_current_user_file(&link).is_err());
std::fs::remove_file(link)?;
std::fs::remove_dir_all(root)?;
Ok(())
}
}
src-tauri/src/core/proxy_control.rs
+503
View File
@@ -0,0 +1,503 @@
use crate::{
config::{Config, IVerge},
core::{CoreManager, manager::RunningMode, service, sysopt::Sysopt},
process::AsyncHandler,
utils::server,
};
use anyhow::{Result, bail, ensure};
use clash_verge_logging::{Type, logging};
use clash_verge_service_ipc::{MacosProxyConfig, OwnerSessionProof, ProxyApplyOutcome};
use std::{
future::Future,
sync::atomic::{AtomicU64, Ordering},
time::Duration,
};
use tokio::sync::Mutex;
const LOOPBACK_HOST: &str = "127.0.0.1";
const MACOS_DEFAULT_BYPASS: &str =
"127.0.0.1,192.168.0.0/16,10.0.0.0/8,172.16.0.0/12,localhost,*.local,*.crashlytics.com,<local>";
const MAX_SERVICE_BYPASS_LEN: usize = 8192;
static SERVICE_PROXY_OPERATIONS: ServiceProxyOperations = ServiceProxyOperations::new();
struct ServiceProxyOperations {
guard_generation: AtomicU64,
operation_lock: Mutex<()>,
}
impl ServiceProxyOperations {
const fn new() -> Self {
Self {
guard_generation: AtomicU64::new(0),
operation_lock: Mutex::const_new(()),
}
}
fn invalidate_guard(&self) -> u64 {
self.guard_generation.fetch_add(1, Ordering::AcqRel).wrapping_add(1)
}
async fn run_service_operation<F, Fut, T>(&self, operation: F) -> T
where
F: FnOnce() -> Fut,
Fut: Future<Output = T>,
{
let _operation = self.operation_lock.lock().await;
operation().await
}
async fn run_final_service_operation<F, Fut, T>(&self, operation: F) -> T
where
F: FnOnce() -> Fut,
Fut: Future<Output = T>,
{
self.invalidate_guard();
self.run_service_operation(operation).await
}
async fn cancel_and_drain<F, Fut>(&self, cancel: F) -> u64
where
F: FnOnce() -> Fut,
Fut: Future<Output = ()>,
{
let generation = self.invalidate_guard();
self.run_service_operation(cancel).await;
generation
}
async fn run_guard_request<Mode, ActiveProof, Request, RequestFuture>(
&self,
captured_generation: u64,
captured_proof: &OwnerSessionProof,
is_service_mode: Mode,
active_proof: ActiveProof,
request: Request,
) -> Result<bool>
where
Mode: FnOnce() -> bool,
ActiveProof: FnOnce() -> Result<OwnerSessionProof>,
Request: FnOnce() -> RequestFuture,
RequestFuture: Future<Output = Result<ProxyApplyOutcome>>,
{
self.run_service_operation(|| async move {
if !guard_generation_is_current(&self.guard_generation, captured_generation)
|| !is_service_mode()
|| !active_proof().is_ok_and(|active| active == *captured_proof)
{
return Ok(false);
}
service_apply_result(request().await?)?;
Ok(true)
})
.await
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProxyBackendRoute {
Local,
Service,
}
const fn proxy_backend_route(is_macos: bool, running_mode: &RunningMode) -> ProxyBackendRoute {
if is_macos && matches!(running_mode, RunningMode::Service) {
ProxyBackendRoute::Service
} else {
ProxyBackendRoute::Local
}
}
fn guard_generation_is_current(generation: &AtomicU64, captured_generation: u64) -> bool {
generation.load(Ordering::Acquire) == captured_generation
}
fn truncate_utf8(value: &mut String, max_bytes: usize) {
if value.len() <= max_bytes {
return;
}
let mut boundary = max_bytes;
while !value.is_char_boundary(boundary) {
boundary -= 1;
}
value.truncate(boundary);
}
fn service_bypass(verge: &IVerge) -> Result<String> {
let custom = verge.system_proxy_bypass.as_deref().unwrap_or("");
ensure!(!custom.contains('\0'), "system proxy bypass contains NUL");
let mut bypass = if custom.is_empty() {
MACOS_DEFAULT_BYPASS.to_owned()
} else if verge.use_default_bypass.unwrap_or(true) {
format!("{MACOS_DEFAULT_BYPASS},{custom}")
} else {
custom.to_owned()
};
truncate_utf8(&mut bypass, MAX_SERVICE_BYPASS_LEN);
Ok(bypass)
}
fn service_proxy_config(verge: &IVerge, mixed_port: u16, pac_port: u16) -> Result<MacosProxyConfig> {
if !verge.enable_system_proxy.unwrap_or_default() {
return Ok(MacosProxyConfig::Disabled);
}
if verge.proxy_auto_config.unwrap_or_default() {
ensure!(pac_port != 0, "embedded PAC server port must not be zero");
Ok(MacosProxyConfig::Pac {
url: format!("http://{LOOPBACK_HOST}:{pac_port}/commands/pac"),
})
} else {
ensure!(mixed_port != 0, "system proxy port must not be zero");
Ok(MacosProxyConfig::Global {
host: LOOPBACK_HOST.to_owned(),
port: mixed_port,
bypass: service_bypass(verge)?,
})
}
}
fn service_apply_result(outcome: ProxyApplyOutcome) -> Result<()> {
match outcome {
ProxyApplyOutcome::Applied | ProxyApplyOutcome::NotRequested => Ok(()),
ProxyApplyOutcome::DirectFallback { message } => bail!(message),
}
}
async fn current_service_proxy_config(verge: &IVerge) -> Result<MacosProxyConfig> {
if !verge.enable_system_proxy.unwrap_or_default() {
return Ok(MacosProxyConfig::Disabled);
}
if verge.proxy_auto_config.unwrap_or_default() {
return service_proxy_config(verge, 0, server::embedded_server_port()?);
}
let mixed_port = match verge.verge_mixed_port {
Some(port) => port,
None => Config::clash().await.latest_arc().get_mixed_port(),
};
service_proxy_config(verge, mixed_port, 0)
}
pub async fn apply() -> Result<()> {
let running_mode = CoreManager::global().get_running_mode();
match proxy_backend_route(cfg!(target_os = "macos"), &running_mode) {
ProxyBackendRoute::Local => Sysopt::global().update_sysproxy().await,
ProxyBackendRoute::Service => {
let verge = Config::verge().await.latest_arc();
let proxy = current_service_proxy_config(&verge).await?;
SERVICE_PROXY_OPERATIONS
.run_final_service_operation(|| async {
service_apply_result(service::set_system_proxy_by_service(&proxy).await?)
})
.await
}
}
}
pub async fn clear() -> Result<()> {
let running_mode = CoreManager::global().get_running_mode();
match proxy_backend_route(cfg!(target_os = "macos"), &running_mode) {
ProxyBackendRoute::Local => Sysopt::global().reset_sysproxy().await,
ProxyBackendRoute::Service => {
SERVICE_PROXY_OPERATIONS
.run_final_service_operation(|| async {
service_apply_result(service::set_system_proxy_by_service(&MacosProxyConfig::Disabled).await?)
})
.await
}
}
}
pub async fn refresh_guard() -> Result<()> {
let generation = SERVICE_PROXY_OPERATIONS
.cancel_and_drain(|| Sysopt::global().stop_proxy_guard())
.await;
let running_mode = CoreManager::global().get_running_mode();
if matches!(
proxy_backend_route(cfg!(target_os = "macos"), &running_mode),
ProxyBackendRoute::Local
) {
Sysopt::global().refresh_guard().await;
return Ok(());
}
let verge = Config::verge().await.latest_arc();
if !verge.enable_system_proxy.unwrap_or_default() || !verge.enable_proxy_guard.unwrap_or_default() {
return Ok(());
}
let proxy = current_service_proxy_config(&verge).await?;
let proof = service::active_service_session()?;
let interval = Duration::from_secs(verge.proxy_guard_duration.unwrap_or(30).max(1));
AsyncHandler::spawn(move || async move {
loop {
tokio::time::sleep(interval).await;
match SERVICE_PROXY_OPERATIONS
.run_guard_request(
generation,
&proof,
|| matches!(*CoreManager::global().get_running_mode(), RunningMode::Service),
service::active_service_session,
|| service::set_system_proxy_by_service_with_session(&proxy, &proof),
)
.await
{
Ok(true) => {}
Ok(false) => break,
Err(_) => logging!(warn, Type::Core, "failed to refresh system proxy through Service"),
}
}
});
Ok(())
}
pub async fn stop_guard() {
SERVICE_PROXY_OPERATIONS
.cancel_and_drain(|| Sysopt::global().stop_proxy_guard())
.await;
}
#[cfg(test)]
mod tests {
use super::{
ProxyBackendRoute, ServiceProxyOperations, guard_generation_is_current, proxy_backend_route,
service_proxy_config,
};
use crate::{config::IVerge, core::manager::RunningMode};
use clash_verge_service_ipc::{MacosProxyConfig, OwnerSessionProof, ProxyApplyOutcome};
use parking_lot::Mutex;
use std::sync::{
Arc,
atomic::{AtomicBool, AtomicU64, Ordering},
};
use std::task::Poll;
use tokio::sync::Barrier;
fn proof(generation: u64, token: &str) -> OwnerSessionProof {
OwnerSessionProof {
generation,
token: token.to_owned(),
}
}
#[test]
fn only_macos_service_routes_proxy_to_helper() {
assert_eq!(
proxy_backend_route(true, &RunningMode::Service),
ProxyBackendRoute::Service
);
assert_eq!(
proxy_backend_route(true, &RunningMode::Sidecar),
ProxyBackendRoute::Local
);
assert_eq!(
proxy_backend_route(false, &RunningMode::Service),
ProxyBackendRoute::Local
);
}
#[test]
fn stale_periodic_refresh_stops_after_owner_loss() {
let generation = AtomicU64::new(7);
let captured_generation = generation.load(Ordering::Acquire);
assert!(guard_generation_is_current(&generation, captured_generation));
generation.fetch_add(1, Ordering::AcqRel);
assert!(!guard_generation_is_current(&generation, captured_generation));
}
#[tokio::test]
async fn final_service_mutation_finishes_after_in_flight_guard_request() {
let operations = Arc::new(ServiceProxyOperations::new());
let generation = operations.invalidate_guard();
let captured_proof = proof(1, "captured");
let active_proof = Arc::new(Mutex::new(captured_proof.clone()));
let guard_started = Arc::new(Barrier::new(2));
let release_guard = Arc::new(Barrier::new(2));
let events = Arc::new(Mutex::new(Vec::new()));
let guard_task = {
let operations = Arc::clone(&operations);
let captured_proof = captured_proof.clone();
let active_proof = Arc::clone(&active_proof);
let guard_started = Arc::clone(&guard_started);
let release_guard = Arc::clone(&release_guard);
let events = Arc::clone(&events);
tokio::spawn(async move {
operations
.run_guard_request(
generation,
&captured_proof,
|| true,
|| Ok(active_proof.lock().clone()),
|| async {
events.lock().push("guard-started");
guard_started.wait().await;
release_guard.wait().await;
events.lock().push("guard-finished");
Ok(ProxyApplyOutcome::Applied)
},
)
.await
})
};
guard_started.wait().await;
let mut final_operation = Box::pin(operations.run_final_service_operation(|| async {
events.lock().push("final");
Ok::<_, anyhow::Error>(())
}));
assert!(matches!(futures::poll!(final_operation.as_mut()), Poll::Pending));
assert!(!guard_generation_is_current(&operations.guard_generation, generation));
release_guard.wait().await;
assert!(matches!(guard_task.await, Ok(Ok(true))));
assert!(matches!(final_operation.await, Ok(())));
assert_eq!(&*events.lock(), &["guard-started", "guard-finished", "final"]);
}
#[tokio::test]
async fn generation_invalidation_while_guard_waits_prevents_rpc() {
let operations = Arc::new(ServiceProxyOperations::new());
let generation = operations.invalidate_guard();
let captured_proof = proof(1, "captured");
let blocker_started = Arc::new(Barrier::new(2));
let release_blocker = Arc::new(Barrier::new(2));
let blocker = {
let operations = Arc::clone(&operations);
let blocker_started = Arc::clone(&blocker_started);
let release_blocker = Arc::clone(&release_blocker);
tokio::spawn(async move {
operations
.run_service_operation(|| async {
blocker_started.wait().await;
release_blocker.wait().await;
})
.await
})
};
blocker_started.wait().await;
let rpc_called = AtomicBool::new(false);
let mut guard_request = Box::pin(operations.run_guard_request(
generation,
&captured_proof,
|| true,
|| Ok(captured_proof.clone()),
|| async {
rpc_called.store(true, Ordering::Release);
Ok(ProxyApplyOutcome::Applied)
},
));
assert!(matches!(futures::poll!(guard_request.as_mut()), Poll::Pending));
operations.invalidate_guard();
release_blocker.wait().await;
assert!(blocker.await.is_ok());
assert!(matches!(guard_request.await, Ok(false)));
assert!(!rpc_called.load(Ordering::Acquire));
}
#[tokio::test]
async fn old_guard_loop_does_not_adopt_replacement_proof() {
let operations = Arc::new(ServiceProxyOperations::new());
let generation = operations.invalidate_guard();
let captured_proof = proof(1, "captured");
let replacement_proof = proof(2, "replacement");
let active_proof = Mutex::new(captured_proof.clone());
let rpc_called = AtomicBool::new(false);
let blocker_started = Arc::new(Barrier::new(2));
let release_blocker = Arc::new(Barrier::new(2));
let blocker = {
let operations = Arc::clone(&operations);
let blocker_started = Arc::clone(&blocker_started);
let release_blocker = Arc::clone(&release_blocker);
tokio::spawn(async move {
operations
.run_service_operation(|| async {
blocker_started.wait().await;
release_blocker.wait().await;
})
.await
})
};
blocker_started.wait().await;
let mut guard_request = Box::pin(operations.run_guard_request(
generation,
&captured_proof,
|| true,
|| Ok(active_proof.lock().clone()),
|| async {
rpc_called.store(true, Ordering::Release);
Ok(ProxyApplyOutcome::Applied)
},
));
assert!(matches!(futures::poll!(guard_request.as_mut()), Poll::Pending));
*active_proof.lock() = replacement_proof;
release_blocker.wait().await;
assert!(blocker.await.is_ok());
assert!(matches!(guard_request.await, Ok(false)));
assert!(!rpc_called.load(Ordering::Acquire));
}
#[tokio::test]
async fn guard_refresh_rejects_direct_fallback() {
let operations = ServiceProxyOperations::new();
let generation = operations.invalidate_guard();
let captured_proof = proof(1, "captured");
let refreshed = operations
.run_guard_request(
generation,
&captured_proof,
|| true,
|| Ok(captured_proof.clone()),
|| async {
Ok(ProxyApplyOutcome::DirectFallback {
message: "fallback detail".to_owned(),
})
},
)
.await;
assert!(refreshed.is_err());
}
#[test]
fn service_proxy_config_forces_loopback_targets_and_bounds_bypass() {
let verge = IVerge {
enable_system_proxy: Some(true),
proxy_auto_config: Some(false),
proxy_host: Some("192.0.2.1".into()),
system_proxy_bypass: Some(format!("{}界", "x".repeat(8191)).into()),
use_default_bypass: Some(false),
..IVerge::default()
};
let proxy = service_proxy_config(&verge, 7897, 3333).unwrap_or_else(|_| unreachable!());
let MacosProxyConfig::Global { host, port, bypass } = proxy else {
unreachable!();
};
assert_eq!(host, "127.0.0.1");
assert_eq!(port, 7897);
assert!(bypass.len() <= 8192);
assert!(bypass.is_char_boundary(bypass.len()));
let pac_verge = IVerge {
proxy_auto_config: Some(true),
..verge
};
assert_eq!(
service_proxy_config(&pac_verge, 7897, 3333).unwrap_or_else(|_| unreachable!()),
MacosProxyConfig::Pac {
url: "http://127.0.0.1:3333/commands/pac".to_owned()
}
);
}
}
src-tauri/src/core/runtime_bundle.rs
+250
View File
@@ -0,0 +1,250 @@
use anyhow::{Context as _, Result, bail};
use clash_verge_service_ipc::{RuntimeAsset, RuntimeBundle};
use serde_yaml_ng::Value;
use std::collections::HashSet;
use std::path::{Component, Path, PathBuf};
const GEO_ASSETS: &[&str] = &[
"Country.mmdb",
"geoip.dat",
"geosite.dat",
"geoip.metadb",
"GeoSite.dat",
];
pub(crate) async fn collect_runtime_bundle(config_file: &Path, core_path: &Path) -> Result<RuntimeBundle> {
let yaml = tokio::fs::read_to_string(config_file)
.await
.with_context(|| format!("failed to read runtime config {config_file:?}"))?;
let mut config: Value =
serde_yaml_ng::from_str(&yaml).with_context(|| format!("failed to parse runtime config {config_file:?}"))?;
let config_root = config_file
.parent()
.ok_or_else(|| anyhow::anyhow!("runtime config has no parent directory"))?;
let config_root = std::fs::canonicalize(config_root)?;
let mut assets = Vec::new();
let mut destinations = HashSet::new();
collect_provider_assets(
&mut config,
"proxy-providers",
&config_root,
&mut destinations,
&mut assets,
)?;
collect_provider_assets(
&mut config,
"rule-providers",
&config_root,
&mut destinations,
&mut assets,
)?;
for filename in GEO_ASSETS {
let source = config_root.join(filename);
if source.is_file() && destinations.insert((*filename).to_string()) {
assets.push(RuntimeAsset {
source: std::fs::canonicalize(&source)?.to_string_lossy().into_owned(),
destination: (*filename).to_string(),
});
}
}
Ok(RuntimeBundle {
yaml: serde_yaml_ng::to_string(&config).context("failed to serialize service runtime config")?,
assets,
core_path: core_path.to_string_lossy().into_owned(),
})
}
fn collect_provider_assets(
config: &mut Value,
section: &str,
config_root: &Path,
destinations: &mut HashSet<String>,
assets: &mut Vec<RuntimeAsset>,
) -> Result<()> {
let Some(providers) = config
.as_mapping_mut()
.and_then(|mapping| mapping.get_mut(section))
.and_then(Value::as_mapping_mut)
else {
return Ok(());
};
for provider in providers.values_mut().filter_map(Value::as_mapping_mut) {
let Some(raw_path) = provider.get("path").and_then(Value::as_str) else {
continue;
};
let is_remote = provider.get("url").and_then(Value::as_str).is_some();
let destination = if is_remote {
provider_destination(config_root, raw_path)?
} else {
let source = local_provider_source(config_root, raw_path)?;
let destination = destination_below_root(config_root, &source)?;
if destinations.insert(destination.clone()) {
assets.push(RuntimeAsset {
source: source.to_string_lossy().into_owned(),
destination: destination.clone(),
});
}
destination
};
provider.insert(Value::String("path".to_owned()), Value::String(destination));
}
Ok(())
}
fn local_provider_source(config_root: &Path, raw_path: &str) -> Result<PathBuf> {
let path = Path::new(raw_path);
let source = if path.is_absolute() {
path.to_path_buf()
} else {
config_root.join(path)
};
let canonical_source =
std::fs::canonicalize(&source).with_context(|| format!("local runtime provider is unavailable: {source:?}"))?;
canonical_source
.strip_prefix(config_root)
.map_err(|_| anyhow::anyhow!("local runtime provider is outside the config root: {canonical_source:?}"))?;
Ok(canonical_source)
}
fn provider_destination(config_root: &Path, raw_path: &str) -> Result<String> {
let path = Path::new(raw_path);
if path
.components()
.any(|component| matches!(component, Component::ParentDir))
{
bail!("runtime provider destination traverses outside the runtime");
}
if path.is_absolute() {
let normalized = canonicalize_with_missing_tail(path)?;
destination_below_root(config_root, &normalized)
} else {
normalized_destination(path)
}
}
fn canonicalize_with_missing_tail(path: &Path) -> Result<PathBuf> {
let mut ancestor = path;
let mut tail = Vec::new();
while !ancestor.exists() {
let name = ancestor
.file_name()
.ok_or_else(|| anyhow::anyhow!("runtime provider path has no existing ancestor: {path:?}"))?;
tail.push(name.to_owned());
ancestor = ancestor
.parent()
.ok_or_else(|| anyhow::anyhow!("runtime provider path has no existing ancestor: {path:?}"))?;
}
let mut normalized = std::fs::canonicalize(ancestor)?;
for component in tail.iter().rev() {
normalized.push(component);
}
Ok(normalized)
}
fn destination_below_root(config_root: &Path, path: &Path) -> Result<String> {
let relative = path
.strip_prefix(config_root)
.map_err(|_| anyhow::anyhow!("runtime provider path is outside the config root: {path:?}"))?;
normalized_destination(relative)
}
fn normalized_destination(relative: &Path) -> Result<String> {
let mut destination = PathBuf::new();
for component in relative.components() {
match component {
Component::Normal(component) => destination.push(component),
Component::CurDir => {}
_ => bail!("local runtime provider destination traverses outside the runtime"),
}
}
if destination.as_os_str().is_empty() {
bail!("local runtime provider destination is empty");
}
Ok(destination.to_string_lossy().replace('\\', "/"))
}
#[cfg(test)]
mod tests {
use super::collect_runtime_bundle;
#[tokio::test]
async fn collects_only_local_providers_and_existing_geo_assets() -> anyhow::Result<()> {
let root = std::env::temp_dir().join(format!("clash-verge-runtime-bundle-{}", std::process::id()));
std::fs::create_dir_all(root.join("providers"))?;
std::fs::write(root.join("providers/local.yaml"), b"proxies: []\n")?;
std::fs::write(root.join("Country.mmdb"), b"geo")?;
let config = root.join("config.yaml");
let absolute_local = root.join("providers/local.yaml");
let absolute_remote = root.join("providers/remote.yaml");
std::fs::write(
&config,
format!(
"proxy-providers:\n local:\n type: file\n path: {}\n remote:\n type: http\n url: https://example.com/p.yaml\n path: {}\nrule-providers: {{}}\n",
absolute_local.display(),
absolute_remote.display()
),
)?;
let core = root.join("mihomo");
std::fs::write(&core, b"core")?;
let bundle = collect_runtime_bundle(&config, &core).await?;
let bundled_config: serde_yaml_ng::Value = serde_yaml_ng::from_str(&bundle.yaml)?;
assert_eq!(
bundled_config["proxy-providers"]["local"]["path"].as_str(),
Some("providers/local.yaml")
);
assert_eq!(
bundled_config["proxy-providers"]["remote"]["path"].as_str(),
Some("providers/remote.yaml")
);
assert!(!bundle.yaml.contains(&root.to_string_lossy().into_owned()));
assert!(
bundle
.assets
.iter()
.any(|asset| asset.destination == "providers/local.yaml")
);
assert!(bundle.assets.iter().any(|asset| asset.destination == "Country.mmdb"));
assert!(!bundle.assets.iter().any(|asset| asset.destination.contains("remote")));
assert!(
!bundle
.assets
.iter()
.any(|asset| asset.destination.contains("geoip.dat"))
);
std::fs::remove_dir_all(root)?;
Ok(())
}
#[tokio::test]
async fn rejects_provider_paths_outside_the_config_root() -> anyhow::Result<()> {
let root = std::env::temp_dir().join(format!("clash-verge-runtime-bundle-outside-{}", std::process::id()));
std::fs::create_dir_all(&root)?;
let config = root.join("config.yaml");
let outside = root
.parent()
.ok_or_else(|| anyhow::anyhow!("test root has no parent"))?
.join(format!("outside-{}.yaml", std::process::id()));
std::fs::write(
&config,
format!(
"proxy-providers:\n remote:\n type: http\n url: https://example.com/p.yaml\n path: {}\n",
outside.display()
),
)?;
let core = root.join("mihomo");
std::fs::write(&core, b"core")?;
let Err(error) = collect_runtime_bundle(&config, &core).await else {
anyhow::bail!("outside provider path must be rejected");
};
assert!(error.to_string().contains("outside the config root"));
std::fs::remove_dir_all(root)?;
Ok(())
}
}
src-tauri/src/core/service.rs
+1866 -263
View File
File diff suppressed because it is too large Load Diff
src-tauri/src/core/sysopt.rs
+36 -16
View File
@@ -1,7 +1,4 @@
use crate::{
config::{Config, IVerge},
singleton,
};
use crate::{config::Config, singleton, utils::server};
use anyhow::Result;
use clash_verge_logging::{Type, logging};
use parking_lot::RwLock;
@@ -33,8 +30,9 @@ const fn proxy_apply_steps(sys_enabled: bool, auto_enabled: bool) -> [ProxyApply
}
}
pub struct Sysopt {
pub(crate) struct Sysopt {
update_lock: TokioMutex<()>,
guard_operation_lock: TokioMutex<()>,
reset_sysproxy: AtomicBool,
inner_proxy: Arc<RwLock<(Sysproxy, Autoproxy)>>,
guard: Arc<RwLock<GuardMonitor>>,
@@ -44,6 +42,7 @@ impl Default for Sysopt {
fn default() -> Self {
Self {
update_lock: TokioMutex::new(()),
guard_operation_lock: TokioMutex::new(()),
reset_sysproxy: AtomicBool::new(false),
inner_proxy: Arc::new(RwLock::new((Sysproxy::default(), Autoproxy::default()))),
guard: Arc::new(RwLock::new(GuardMonitor::new(GuardType::None, Duration::from_secs(30)))),
@@ -84,17 +83,35 @@ impl Sysopt {
Arc::clone(&self.guard)
}
pub async fn refresh_guard(&self) {
async fn stop_proxy_guard_locked(&self) {
loop {
let state = self.access_guard().read().get_state();
if state.is_pendding() {
tokio::task::yield_now().await;
continue;
}
self.access_guard().write().stop();
return;
}
}
pub(super) async fn stop_proxy_guard(&self) {
let _operation = self.guard_operation_lock.lock().await;
self.stop_proxy_guard_locked().await;
}
pub(super) async fn refresh_guard(&self) {
logging!(info, Type::Core, "Refreshing system proxy guard...");
let verge = Config::verge().await.latest_arc();
let _operation = self.guard_operation_lock.lock().await;
if !verge.enable_system_proxy.unwrap_or_default() {
logging!(info, Type::Core, "System proxy is disabled.");
self.access_guard().write().stop();
self.stop_proxy_guard_locked().await;
return;
}
if !verge.enable_proxy_guard.unwrap_or_default() {
logging!(info, Type::Core, "System proxy guard is disabled.");
self.access_guard().write().stop();
self.stop_proxy_guard_locked().await;
return;
}
logging!(
@@ -114,26 +131,28 @@ impl Sysopt {
let guard = self.access_guard();
guard.write().start();
}
while self.access_guard().read().get_state().is_pendding() {
tokio::task::yield_now().await;
}
}
/// Wait for any in-progress `update_sysproxy` to finish, so that a
/// subsequent read of OS-level sysproxy state sees a fully applied
/// configuration instead of a partially-applied one (e.g. SOCKS already
/// disabled but HTTP still enabled mid-transition).
pub async fn wait_idle(&self) {
pub(crate) async fn wait_idle(&self) {
let _ = self.update_lock.lock().await;
}
/// init the sysproxy
pub async fn update_sysproxy(&self) -> Result<()> {
pub(super) async fn update_sysproxy(&self) -> Result<()> {
let _lock = self.update_lock.lock().await;
let verge = Config::verge().await.latest_arc();
let port = match verge.verge_mixed_port {
Some(port) => port,
None => Config::clash().await.latest_arc().get_mixed_port(),
};
let pac_port = IVerge::get_singleton_port();
let pac_port = server::embedded_server_port()?;
// 先 await, 避免持有锁导致的 Send 问题
let bypass = get_bypass().await;
@@ -178,6 +197,7 @@ impl Sysopt {
(sys.clone(), auto.clone(), guard_type)
};
let _guard_operation = self.guard_operation_lock.lock().await;
self.access_guard().write().set_guard_type(guard_type);
let apply_steps = proxy_apply_steps(sys.enable, auto.enable);
@@ -197,7 +217,7 @@ impl Sysopt {
}
/// reset the sysproxy
pub async fn reset_sysproxy(&self) -> Result<()> {
pub(super) async fn reset_sysproxy(&self) -> Result<()> {
if self
.reset_sysproxy
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
@@ -208,9 +228,9 @@ impl Sysopt {
defer! {
self.reset_sysproxy.store(false, Ordering::SeqCst);
}
// close proxy guard
self.access_guard().write().set_guard_type(GuardType::None);
let _lock = self.update_lock.lock().await;
let _guard_operation = self.guard_operation_lock.lock().await;
self.stop_proxy_guard_locked().await;
// 直接关闭所有代理
let (sys, auto) = {
src-tauri/src/core/timer.rs
+127 -26
View File
@@ -36,6 +36,24 @@ struct TaskState {
running: bool,
}
async fn run_timer_profile_update_transition<Started, Update, UpdateFuture, Finished, Completed, Output>(
update_started: Started,
update_profile: Update,
update_finished: Finished,
update_completed: Completed,
) where
Started: FnOnce(),
Update: FnOnce() -> UpdateFuture,
UpdateFuture: std::future::Future<Output = Output>,
Finished: FnOnce(Output),
Completed: FnOnce(),
{
update_started();
let result = update_profile().await;
update_finished(result);
update_completed();
}
impl TaskState {
const fn new(key: Key, interval_minutes: u64) -> Self {
Self {
@@ -386,36 +404,35 @@ impl Timer {
let task_start = std::time::Instant::now();
logging!(debug, Type::Timer, "Running timer task for profile: {}", uid);
match tokio::time::timeout(std::time::Duration::from_secs(40), async {
Self::emit_update_event(uid, true);
let is_current = Config::profiles().await.latest_arc().current.as_ref() == Some(uid);
logging!(
debug,
Type::Timer,
"Profile {} is current active profile: {}",
uid,
is_current
);
feat::update_profile(uid, None, is_current, false, false).await
})
.await
{
Ok(Ok(_)) => {
run_timer_profile_update_transition(
|| Self::emit_update_event(uid, true),
|| async {
let is_current = Config::profiles().await.latest_arc().current.as_ref() == Some(uid);
logging!(
info,
debug,
Type::Timer,
"Timer task completed for uid: {} (took {}ms)",
"Profile {} is current active profile: {}",
uid,
task_start.elapsed().as_millis()
is_current
);
}
Ok(Err(e)) => logging_error!(Type::Timer, "Failed to update profile uid {}: {}", uid, e),
Err(_) => logging_error!(Type::Timer, "Timer task timed out for uid: {}", uid),
}
Self::emit_update_event(uid, false);
feat::update_profile(uid, None, is_current, false, false).await
},
|result| match result {
Ok(_) => {
logging!(
info,
Type::Timer,
"Timer task completed for uid: {} (took {}ms)",
uid,
task_start.elapsed().as_millis()
);
}
Err(e) => logging_error!(Type::Timer, "Failed to update profile uid {}: {}", uid, e),
},
|| Self::emit_update_event(uid, false),
)
.await;
}
async fn wait_until_resolve_done(max_wait: Duration) {
@@ -431,8 +448,31 @@ impl Timer {
#[cfg(test)]
mod tests {
use super::Timer;
use super::{Timer, run_timer_profile_update_transition};
use crate::config::{PrfItem, PrfOption};
use parking_lot::Mutex;
use std::{
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
task::Poll,
time::Duration,
};
use tokio::sync::Barrier;
struct CancellationProbe {
cancelled: Arc<AtomicBool>,
completed: Arc<AtomicBool>,
}
impl Drop for CancellationProbe {
fn drop(&mut self) {
if !self.completed.load(Ordering::Acquire) {
self.cancelled.store(true, Ordering::Release);
}
}
}
fn remote_profile(uid: &str, allow_auto_update: Option<bool>, update_interval: Option<u64>) -> PrfItem {
PrfItem {
@@ -463,4 +503,65 @@ mod tests {
assert_eq!(map.get("enabled"), Some(&30));
assert_eq!(map.get("missing-flag"), Some(&30));
}
#[tokio::test(start_paused = true)]
async fn timer_profile_update_runs_past_former_deadline_and_pairs_events() -> anyhow::Result<()> {
let calls = Arc::new(Mutex::new(Vec::new()));
let update_started = Arc::new(Barrier::new(2));
let release_update = Arc::new(Barrier::new(2));
let update_cancelled = Arc::new(AtomicBool::new(false));
let update_completed = Arc::new(AtomicBool::new(false));
let mut update = Box::pin(run_timer_profile_update_transition(
{
let calls = Arc::clone(&calls);
move || calls.lock().push("update-started")
},
{
let update_started = Arc::clone(&update_started);
let release_update = Arc::clone(&release_update);
let update_cancelled = Arc::clone(&update_cancelled);
let update_completed = Arc::clone(&update_completed);
move || async move {
let _probe = CancellationProbe {
cancelled: update_cancelled,
completed: Arc::clone(&update_completed),
};
update_started.wait().await;
release_update.wait().await;
update_completed.store(true, Ordering::Release);
Ok::<(), anyhow::Error>(())
}
},
{
let calls = Arc::clone(&calls);
move |result: anyhow::Result<()>| {
assert!(result.is_ok());
calls.lock().push("terminal-result");
}
},
{
let calls = Arc::clone(&calls);
move || calls.lock().push("update-completed")
},
));
assert!(matches!(futures::poll!(update.as_mut()), Poll::Pending));
update_started.wait().await;
tokio::time::advance(Duration::from_secs(41)).await;
assert!(matches!(futures::poll!(update.as_mut()), Poll::Pending));
assert!(!update_cancelled.load(Ordering::Acquire));
assert_eq!(&*calls.lock(), &["update-started"]);
release_update.wait().await;
update.await;
assert!(update_completed.load(Ordering::Acquire));
assert!(!update_cancelled.load(Ordering::Acquire));
assert_eq!(
&*calls.lock(),
&["update-started", "terminal-result", "update-completed"]
);
Ok(())
}
}
src-tauri/src/core/tray/mod.rs
+2 -3
View File
@@ -208,8 +208,7 @@ impl Tray {
let verge = Config::verge().await.latest_arc();
let system_proxy = verge.enable_system_proxy.as_ref().unwrap_or(&false);
let tun_mode = verge.enable_tun_mode.as_ref().unwrap_or(&false);
let tun_mode_available =
is_current_app_handle_admin(app_handle) || service::is_service_available().await.is_ok();
let tun_mode_available = is_current_app_handle_admin(app_handle) || service::is_service_available().await;
let mode = {
Config::clash()
.await
@@ -994,7 +993,7 @@ fn on_menu_event(_: &AppHandle, event: MenuEvent) {
let _ = help::open_app_latest_log();
}
MenuIds::CORE_LOG => {
let _ = help::open_core_latest_log();
let _ = help::open_core_latest_log().await;
}
MenuIds::RESTART_CLASH => feat::restart_clash_core().await,
MenuIds::RESTART_APP => feat::restart_app().await,