CalvinBackup/donutbrowser
1
0
Fork 0
mirror of https://github.com/zhom/donutbrowser.git synced 2026-07-06 04:57:48 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: more robust vpn handling

Browse Source
This commit is contained in:
zhom
2026-04-04 03:16:04 +04:00
parent ac5d975e5b
commit 48883ddd03
19 changed files with 1936 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/ip_utils.rs
+16 -7
View File
@@ -55,6 +55,7 @@ pub async fn fetch_public_ip(proxy: Option<&str>) -> Result<String, IpError> {
let proxy = reqwest::Proxy::all(proxy_url)
.map_err(|e| IpError::Network(format!("Invalid proxy: {}", e)))?;
client_builder
.no_proxy()
.proxy(proxy)
.build()
.map_err(|e| IpError::Network(e.to_string()))?
@@ -64,7 +65,7 @@ pub async fn fetch_public_ip(proxy: Option<&str>) -> Result<String, IpError> {
.map_err(|e| IpError::Network(e.to_string()))?
};
let mut last_error = None;
let mut errors = Vec::new();
for url in &urls {
match client.get(*url).send().await {
@@ -76,21 +77,29 @@ pub async fn fetch_public_ip(proxy: Option<&str>) -> Result<String, IpError> {
}
}
Err(e) => {
last_error = Some(format!("Failed to read response from {}: {}", url, e));
errors.push(format!("{}: {}", url, e));
}
},
Ok(response) => {
last_error = Some(format!("HTTP {} from {}", response.status(), url));
errors.push(format!("{}: HTTP {}", url, response.status()));
}
Err(e) => {
last_error = Some(format!("Request to {} failed: {}", url, e));
errors.push(format!("{}: {}", url, e));
}
}
}
Err(IpError::Network(last_error.unwrap_or_else(|| {
"Failed to fetch public IP from any endpoint".to_string()
})))
if errors.is_empty() {
Err(IpError::Network(
"Failed to fetch public IP from any endpoint".to_string(),
))
} else {
Err(IpError::Network(format!(
"All {} endpoints failed: {}",
errors.len(),
errors.join("; ")
)))
}
}
#[cfg(test)]
src-tauri/src/lib.rs
+103 -30
View File
@@ -813,6 +813,42 @@ async fn download_geoip_database(app_handle: tauri::AppHandle) -> Result<(), Str
}
// VPN commands
#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct VpnDependencyStatus {
is_available: bool,
requires_external_install: bool,
missing_binary: bool,
missing_windows_adapter: bool,
dependency_check_failed: bool,
}
#[tauri::command]
async fn get_vpn_dependency_status(vpn_type: vpn::VpnType) -> Result<VpnDependencyStatus, String> {
match vpn_type {
vpn::VpnType::WireGuard => Ok(VpnDependencyStatus {
is_available: true,
requires_external_install: false,
missing_binary: false,
missing_windows_adapter: false,
dependency_check_failed: false,
}),
vpn::VpnType::OpenVPN => {
let status = crate::vpn::openvpn_socks5::OpenVpnSocks5Server::dependency_status();
let is_available =
status.binary_found && !status.missing_windows_adapter && !status.dependency_check_failed;
Ok(VpnDependencyStatus {
is_available,
requires_external_install: true,
missing_binary: !status.binary_found,
missing_windows_adapter: status.missing_windows_adapter,
dependency_check_failed: status.dependency_check_failed,
})
}
}
}
#[tauri::command]
async fn import_vpn_config(
content: String,
@@ -986,45 +1022,81 @@ async fn check_vpn_validity(
.unwrap_or_default()
.as_secs();
// Start a temporary VPN worker to send real traffic
let had_existing_worker = vpn_worker_storage::find_vpn_worker_by_vpn_id(&vpn_id).is_some();
let vpn_worker = vpn_worker_runner::start_vpn_worker(&vpn_id)
.await
.map_err(|e| format!("Failed to start VPN worker: {e}"))?;
let socks_url = format!("socks5://127.0.0.1:{}", vpn_worker.local_port.unwrap_or(0));
let socks_url = format!(
"socks5://127.0.0.1:{}",
vpn_worker.local_port.unwrap_or_default()
);
// Fetch public IP through the VPN SOCKS5 proxy
let result = match ip_utils::fetch_public_ip(Some(&socks_url)).await {
Ok(ip) => {
let (city, country, country_code) =
crate::proxy_manager::ProxyManager::get_ip_geolocation(&ip)
.await
.unwrap_or_default();
crate::proxy_manager::ProxyCheckResult {
ip,
city,
country,
country_code,
timestamp: now,
is_valid: true,
}
}
Err(e) => {
log::warn!("VPN check failed to fetch public IP: {e}");
crate::proxy_manager::ProxyCheckResult {
ip: String::new(),
city: None,
country: None,
country_code: None,
timestamp: now,
is_valid: false,
let local_proxy = crate::proxy_runner::start_proxy_process(Some(socks_url), None)
.await
.map_err(|error| error.to_string());
let local_proxy = match local_proxy {
Ok(proxy) => proxy,
Err(error_message) => {
if !had_existing_worker {
let _ = vpn_worker_runner::stop_vpn_worker(&vpn_worker.id).await;
}
return Err(format!("Failed to start validation proxy: {error_message}"));
}
};
// Stop the temporary VPN worker
let _ = vpn_worker_runner::stop_vpn_worker(&vpn_worker.id).await;
let local_proxy_url = format!(
"http://127.0.0.1:{}",
local_proxy.local_port.unwrap_or_default()
);
let mut result = None;
for attempt in 0..3 {
if attempt > 0 {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
match ip_utils::fetch_public_ip(Some(&local_proxy_url)).await {
Ok(ip) => {
let (city, country, country_code) =
crate::proxy_manager::ProxyManager::get_ip_geolocation(&ip)
.await
.unwrap_or_default();
result = Some(crate::proxy_manager::ProxyCheckResult {
ip,
city,
country,
country_code,
timestamp: now,
is_valid: true,
});
break;
}
Err(error) => {
log::warn!(
"VPN validation attempt {} failed to fetch public IP through donut-proxy: {}",
attempt + 1,
error
);
}
}
}
let _ = crate::proxy_runner::stop_proxy_process(&local_proxy.id).await;
if !had_existing_worker {
let _ = vpn_worker_runner::stop_vpn_worker(&vpn_worker.id).await;
}
let result = result.unwrap_or(crate::proxy_manager::ProxyCheckResult {
ip: String::new(),
city: None,
country: None,
country_code: None,
timestamp: now,
is_valid: false,
});
Ok(result)
}
@@ -1932,6 +2004,7 @@ pub fn run() {
add_mcp_to_claude_code,
remove_mcp_from_claude_code,
// VPN commands
get_vpn_dependency_status,
import_vpn_config,
list_vpn_configs,
get_vpn_config,
src-tauri/src/proxy_runner.rs
+150 -1
View File
@@ -2,12 +2,161 @@ use crate::proxy_storage::{
delete_proxy_config, generate_proxy_id, get_proxy_config, is_process_running, list_proxy_configs,
save_proxy_config, ProxyConfig,
};
use std::path::{Path, PathBuf};
use std::process::Stdio;
lazy_static::lazy_static! {
static ref PROXY_PROCESSES: std::sync::Mutex<std::collections::HashMap<String, u32>> =
std::sync::Mutex::new(std::collections::HashMap::new());
}
fn target_binary_name(base_name: &str) -> Option<String> {
let target = std::env::var("TARGET").ok()?;
#[cfg(windows)]
{
Some(format!("{base_name}-{target}.exe"))
}
#[cfg(not(windows))]
{
Some(format!("{base_name}-{target}"))
}
}
fn unsuffixed_binary_name(base_name: &str) -> String {
#[cfg(windows)]
{
match base_name {
"donut-proxy" => "donut-proxy.exe".to_string(),
"donut-daemon" => "donut-daemon.exe".to_string(),
_ => String::new(),
}
}
#[cfg(not(windows))]
{
base_name.to_string()
}
}
fn binary_matches_prefix(path: &Path, base_name: &str) -> bool {
let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
return false;
};
#[cfg(windows)]
{
file_name.starts_with(&format!("{base_name}-")) && file_name.ends_with(".exe")
}
#[cfg(not(windows))]
{
file_name.starts_with(&format!("{base_name}-"))
}
}
fn push_candidate_dir(dirs: &mut Vec<PathBuf>, dir: Option<PathBuf>) {
if let Some(dir) = dir {
if !dirs.iter().any(|existing| existing == &dir) {
dirs.push(dir);
}
}
}
pub(crate) fn find_sidecar_executable(
base_name: &str,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
let current_exe = std::env::current_exe()?;
let current_dir = current_exe
.parent()
.ok_or("Failed to get parent directory of current executable")?;
if current_exe
.file_stem()
.and_then(|stem| stem.to_str())
.is_some_and(|stem| stem == base_name)
{
return Ok(current_exe);
}
let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
let mut search_dirs = Vec::new();
push_candidate_dir(&mut search_dirs, Some(current_dir.to_path_buf()));
push_candidate_dir(
&mut search_dirs,
current_dir.parent().map(std::path::Path::to_path_buf),
);
push_candidate_dir(
&mut search_dirs,
current_dir
.parent()
.and_then(|parent| parent.parent())
.map(Path::to_path_buf),
);
push_candidate_dir(&mut search_dirs, Some(current_dir.join("binaries")));
push_candidate_dir(
&mut search_dirs,
current_dir.parent().map(|parent| parent.join("binaries")),
);
push_candidate_dir(
&mut search_dirs,
current_dir
.parent()
.and_then(|parent| parent.parent())
.map(|parent| parent.join("binaries")),
);
push_candidate_dir(&mut search_dirs, Some(manifest_dir.join("binaries")));
push_candidate_dir(
&mut search_dirs,
Some(manifest_dir.join("target").join("debug")),
);
push_candidate_dir(
&mut search_dirs,
Some(manifest_dir.join("target").join("release")),
);
let mut exact_names = vec![unsuffixed_binary_name(base_name)];
if let Some(target_name) = target_binary_name(base_name) {
exact_names.push(target_name);
}
for dir in &search_dirs {
for name in &exact_names {
if name.is_empty() {
continue;
}
let candidate = dir.join(name);
if candidate.exists() {
return Ok(candidate);
}
}
if let Ok(entries) = std::fs::read_dir(dir) {
for entry in entries.flatten() {
let path = entry.path();
if path.is_file() && binary_matches_prefix(&path, base_name) {
return Ok(path);
}
}
}
}
Err(
format!(
"Failed to locate '{}' executable. Searched in: {}",
base_name,
search_dirs
.iter()
.map(|dir| dir.display().to_string())
.collect::<Vec<_>>()
.join(", ")
)
.into(),
)
}
pub async fn start_proxy_process(
upstream_url: Option<String>,
port: Option<u16>,
@@ -47,7 +196,7 @@ pub async fn start_proxy_process_with_profile(
// Spawn proxy worker process in the background using std::process::Command
// This ensures proper process detachment on Unix systems
let exe = std::env::current_exe()?;
let exe = find_sidecar_executable("donut-proxy")?;
#[cfg(unix)]
{
src-tauri/src/traffic_stats.rs
+6 -2
View File
@@ -580,7 +580,9 @@ impl LiveTrafficTracker {
.profile_id
.clone()
.unwrap_or_else(|| self.proxy_id.clone());
let session_file = get_traffic_stats_dir().join(format!("{}.session.json", storage_key));
let storage_dir = get_traffic_stats_dir();
fs::create_dir_all(&storage_dir)?;
let session_file = storage_dir.join(format!("{}.session.json", storage_key));
// Write atomically using a temp file
let temp_file = session_file.with_extension("tmp");
@@ -761,9 +763,11 @@ impl LiveTrafficTracker {
.profile_id
.clone()
.unwrap_or_else(|| self.proxy_id.clone());
let storage_dir = get_traffic_stats_dir();
fs::create_dir_all(&storage_dir)?;
// Use file locking to prevent concurrent writes from multiple proxy processes
let lock_path = get_traffic_stats_dir().join(format!("{}.lock", storage_key));
let lock_path = storage_dir.join(format!("{}.lock", storage_key));
let _lock = match acquire_file_lock(&lock_path) {
Ok(lock) => lock,
Err(e) => {
src-tauri/src/vpn/openvpn_socks5.rs
+635 -64
View File
@@ -1,8 +1,24 @@
use super::config::{OpenVpnConfig, VpnError};
use std::path::PathBuf;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::{lookup_host, TcpListener, TcpSocket, TcpStream};
const OPENVPN_CONNECT_TIMEOUT_SECS: u64 = 90;
enum SocksTarget {
Address(SocketAddr),
Domain(String, u16),
}
#[derive(Debug, Clone, Copy)]
pub(crate) struct OpenVpnDependencyStatus {
pub binary_found: bool,
pub missing_windows_adapter: bool,
pub dependency_check_failed: bool,
}
pub struct OpenVpnSocks5Server {
config: OpenVpnConfig,
@@ -14,7 +30,167 @@ impl OpenVpnSocks5Server {
Self { config, port }
}
fn find_openvpn_binary() -> Result<PathBuf, VpnError> {
fn read_log_tail(path: &Path, lines: usize) -> String {
std::fs::read_to_string(path)
.unwrap_or_default()
.lines()
.rev()
.take(lines)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect::<Vec<_>>()
.join("\n")
}
fn extract_vpn_ip(line: &str) -> Option<Ipv4Addr> {
for field in line.split(',') {
let trimmed = field.trim();
if let Ok(ip) = trimmed.parse::<Ipv4Addr>() {
if ip.is_private() && !ip.is_loopback() {
return Some(ip);
}
}
}
None
}
fn log_indicates_connected(log_content: &str) -> bool {
log_content.contains("Initialization Sequence Completed")
}
fn log_indicates_failure(log_content: &str) -> bool {
log_content.contains("AUTH_FAILED")
|| log_content.contains("Exiting due to fatal error")
|| log_content.contains("Fatal error")
|| log_content.contains("Options error")
|| log_content.contains("Exiting")
}
fn has_config_directive(config: &str, directive: &str) -> bool {
config.lines().any(|line| {
let trimmed = line.trim();
!trimmed.is_empty()
&& !trimmed.starts_with('#')
&& !trimmed.starts_with(';')
&& trimmed.starts_with(directive)
})
}
fn strip_config_directive(config: &str, directive: &str) -> String {
config
.lines()
.filter(|line| {
let trimmed = line.trim();
trimmed.is_empty()
|| trimmed.starts_with('#')
|| trimmed.starts_with(';')
|| !trimmed.starts_with(directive)
})
.collect::<Vec<_>>()
.join("\n")
}
fn build_runtime_config(&self) -> String {
let mut runtime_config = self.config.raw_config.clone();
runtime_config = Self::strip_config_directive(&runtime_config, "redirect-gateway");
runtime_config = Self::strip_config_directive(&runtime_config, "block-outside-dns");
runtime_config = Self::strip_config_directive(&runtime_config, "dhcp-option");
if !runtime_config.contains("pull-filter ignore \"redirect-gateway\"") {
runtime_config.push_str("\npull-filter ignore \"redirect-gateway\"\n");
}
if !runtime_config.contains("pull-filter ignore \"block-outside-dns\"") {
runtime_config.push_str("pull-filter ignore \"block-outside-dns\"\n");
}
if !runtime_config.contains("pull-filter ignore \"dhcp-option\"") {
runtime_config.push_str("pull-filter ignore \"dhcp-option\"\n");
}
if !Self::has_config_directive(&runtime_config, "route 0.0.0.0") {
runtime_config.push_str("\nroute 0.0.0.0 0.0.0.0 vpn_gateway 9999\n");
}
#[cfg(windows)]
{
if Self::has_config_directive(&runtime_config, "dev-node") {
runtime_config = runtime_config
.lines()
.filter(|line| {
let trimmed = line.trim();
trimmed.is_empty()
|| trimmed.starts_with('#')
|| trimmed.starts_with(';')
|| !trimmed.starts_with("dev-node")
})
.collect::<Vec<_>>()
.join("\n");
}
if !Self::has_config_directive(&runtime_config, "disable-dco") {
runtime_config.push_str("\ndisable-dco\n");
}
if self.config.dev_type.starts_with("tun")
&& !Self::has_config_directive(&runtime_config, "windows-driver")
{
runtime_config.push_str("\nwindows-driver wintun\n");
}
}
runtime_config
}
pub(crate) fn dependency_status() -> OpenVpnDependencyStatus {
let Ok(openvpn_bin) = Self::find_openvpn_binary() else {
return OpenVpnDependencyStatus {
binary_found: false,
missing_windows_adapter: false,
dependency_check_failed: false,
};
};
#[cfg(windows)]
{
match Self::windows_openvpn_has_adapter(&openvpn_bin) {
Ok(has_adapter) => OpenVpnDependencyStatus {
binary_found: true,
missing_windows_adapter: !has_adapter,
dependency_check_failed: false,
},
Err(_) => OpenVpnDependencyStatus {
binary_found: true,
missing_windows_adapter: false,
dependency_check_failed: true,
},
}
}
#[cfg(not(windows))]
{
OpenVpnDependencyStatus {
binary_found: true,
missing_windows_adapter: false,
dependency_check_failed: false,
}
}
}
pub(crate) fn find_openvpn_binary() -> Result<PathBuf, VpnError> {
if let Ok(path) = std::env::var("DONUTBROWSER_OPENVPN_BIN") {
let path = PathBuf::from(path);
if path.exists() {
return Ok(path);
}
return Err(VpnError::Connection(format!(
"Configured OpenVPN binary does not exist: {}",
path.display()
)));
}
let locations = [
"/usr/sbin/openvpn",
"/usr/local/sbin/openvpn",
@@ -71,12 +247,300 @@ impl OpenVpnSocks5Server {
))
}
fn openvpn_supports_management(openvpn_bin: &Path) -> bool {
let mut command = Command::new(openvpn_bin);
command.arg("--version");
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
const CREATE_NO_WINDOW: u32 = 0x08000000;
command.creation_flags(CREATE_NO_WINDOW);
}
let Ok(output) = command.output() else {
return true;
};
let version_text = format!(
"{}{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
!version_text.contains("enable_management=no")
}
#[cfg(windows)]
pub(crate) fn windows_openvpn_has_adapter(openvpn_bin: &Path) -> Result<bool, VpnError> {
use std::os::windows::process::CommandExt;
const CREATE_NO_WINDOW: u32 = 0x08000000;
let output = Command::new(openvpn_bin)
.arg("--show-adapters")
.creation_flags(CREATE_NO_WINDOW)
.output()
.map_err(|e| VpnError::Connection(format!("Failed to inspect OpenVPN adapters: {e}")))?;
let text = format!(
"{}{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
Ok(
text
.lines()
.map(str::trim)
.any(|line| !line.is_empty() && !line.starts_with("Available adapters")),
)
}
fn extract_vpn_ip_from_log(log_content: &str) -> Option<Ipv4Addr> {
for line in log_content.lines() {
if let Some(ip) = Self::extract_vpn_ip(line) {
return Some(ip);
}
if let Some(position) = line.find("ifconfig ") {
let after = &line[position + "ifconfig ".len()..];
if let Some(ip_str) = after
.split_whitespace()
.next()
.or_else(|| after.split(',').next())
{
if let Ok(ip) = ip_str.parse::<Ipv4Addr>() {
if ip.is_private() && !ip.is_loopback() {
return Some(ip);
}
}
}
}
}
None
}
async fn wait_for_openvpn_ready_via_management(
child: &mut std::process::Child,
mgmt_port: u16,
log_path: &Path,
) -> Result<Option<Ipv4Addr>, VpnError> {
let deadline =
tokio::time::Instant::now() + tokio::time::Duration::from_secs(OPENVPN_CONNECT_TIMEOUT_SECS);
let mgmt_stream = loop {
if tokio::time::Instant::now() >= deadline {
return Err(VpnError::Connection(format!(
"Timed out connecting to OpenVPN management interface. Last OpenVPN output:\n{}",
Self::read_log_tail(log_path, 20)
)));
}
if let Ok(Some(status)) = child.try_wait() {
return Err(VpnError::Connection(format!(
"OpenVPN exited (status: {}) before the tunnel was established. Last output:\n{}",
status,
Self::read_log_tail(log_path, 20)
)));
}
match TcpStream::connect(("127.0.0.1", mgmt_port)).await {
Ok(stream) => break stream,
Err(_) => tokio::time::sleep(tokio::time::Duration::from_millis(500)).await,
}
};
let (mgmt_reader, mut mgmt_writer) = mgmt_stream.into_split();
let _ = mgmt_writer.write_all(b"state on\nstate\n").await;
let mut lines = BufReader::new(mgmt_reader).lines();
let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(1));
interval.tick().await;
let mut vpn_ip = None;
loop {
if tokio::time::Instant::now() >= deadline {
return Err(VpnError::Connection(format!(
"Timed out waiting for OpenVPN to reach CONNECTED state. Last OpenVPN output:\n{}",
Self::read_log_tail(log_path, 20)
)));
}
if let Ok(Some(status)) = child.try_wait() {
return Err(VpnError::Connection(format!(
"OpenVPN exited (status: {}) before connecting. Last output:\n{}",
status,
Self::read_log_tail(log_path, 20)
)));
}
tokio::select! {
line_result = lines.next_line() => {
match line_result {
Ok(Some(line)) => {
if let Some(ip) = Self::extract_vpn_ip(&line) {
vpn_ip = Some(ip);
}
if line.contains(",CONNECTED,") {
break;
}
if line.contains("AUTH_FAILED") {
return Err(VpnError::Connection(format!(
"OpenVPN authentication failed. Last output:\n{}",
Self::read_log_tail(log_path, 20)
)));
}
if line.contains(",EXITING,") || line.contains(">FATAL:") {
return Err(VpnError::Connection(format!(
"OpenVPN is exiting. Last output:\n{}",
Self::read_log_tail(log_path, 20)
)));
}
}
Ok(None) => {
return Err(VpnError::Connection(format!(
"OpenVPN management connection closed before CONNECTED state. Last output:\n{}",
Self::read_log_tail(log_path, 20)
)));
}
Err(_) => {}
}
}
_ = interval.tick() => {
let _ = mgmt_writer.write_all(b"state\n").await;
let log_path = log_path.to_path_buf();
let log_content = tokio::task::spawn_blocking(move || std::fs::read_to_string(log_path))
.await
.ok()
.and_then(Result::ok);
if let Some(content) = log_content {
if Self::log_indicates_connected(&content) {
break;
}
}
}
}
}
if vpn_ip.is_none() {
if let Ok(log_content) = std::fs::read_to_string(log_path) {
vpn_ip = Self::extract_vpn_ip_from_log(&log_content);
}
}
Ok(vpn_ip)
}
async fn wait_for_openvpn_ready_via_log(
child: &mut std::process::Child,
log_path: &Path,
) -> Result<Option<Ipv4Addr>, VpnError> {
let deadline =
tokio::time::Instant::now() + tokio::time::Duration::from_secs(OPENVPN_CONNECT_TIMEOUT_SECS);
loop {
if tokio::time::Instant::now() >= deadline {
return Err(VpnError::Connection(format!(
"Timed out waiting for OpenVPN to connect. Last OpenVPN output:\n{}",
Self::read_log_tail(log_path, 40)
)));
}
if let Ok(Some(status)) = child.try_wait() {
return Err(VpnError::Connection(format!(
"OpenVPN exited (status: {}) before connecting. Last output:\n{}",
status,
Self::read_log_tail(log_path, 40)
)));
}
let log_path_buf = log_path.to_path_buf();
let log_content = tokio::task::spawn_blocking(move || std::fs::read_to_string(log_path_buf))
.await
.ok()
.and_then(Result::ok)
.unwrap_or_default();
if Self::log_indicates_connected(&log_content) {
return Ok(Self::extract_vpn_ip_from_log(&log_content));
}
if Self::log_indicates_failure(&log_content) {
return Err(VpnError::Connection(format!(
"OpenVPN reported a fatal error while connecting. Last output:\n{}",
Self::read_log_tail(log_path, 40)
)));
}
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
}
}
async fn connect_target(
target: SocksTarget,
vpn_bind_ip: Ipv4Addr,
) -> Result<(TcpStream, SocketAddr), Box<dyn std::error::Error + Send + Sync>> {
let mut addresses = match target {
SocksTarget::Address(addr) => vec![addr],
SocksTarget::Domain(host, port) => {
let mut resolved = lookup_host((host.as_str(), port))
.await?
.collect::<Vec<_>>();
resolved.sort_by_key(|addr| if addr.is_ipv4() { 0 } else { 1 });
resolved
}
};
if addresses.is_empty() {
return Err("No addresses resolved for SOCKS5 target".into());
}
let mut last_error = None;
for address in addresses.drain(..) {
let socket = if address.is_ipv4() {
let socket = TcpSocket::new_v4()?;
if !vpn_bind_ip.is_unspecified() {
socket.bind(SocketAddr::new(IpAddr::V4(vpn_bind_ip), 0))?;
}
socket
} else {
TcpSocket::new_v6()?
};
match socket.connect(address).await {
Ok(stream) => return Ok((stream, address)),
Err(error) => last_error = Some(error),
}
}
Err(
last_error
.map(|error| error.into())
.unwrap_or_else(|| "Failed to connect to any resolved SOCKS5 target".into()),
)
}
pub async fn run(self, config_id: String) -> Result<(), VpnError> {
let openvpn_bin = Self::find_openvpn_binary()?;
let supports_management = Self::openvpn_supports_management(&openvpn_bin);
#[cfg(windows)]
if !Self::windows_openvpn_has_adapter(&openvpn_bin)? {
return Err(VpnError::Connection(
"OpenVPN requires a TAP/Wintun/ovpn-dco adapter on Windows, but none were found. Install or provision an adapter before connecting.".to_string(),
));
}
// Write config to temp file
let config_path = std::env::temp_dir().join(format!("openvpn_{}.ovpn", config_id));
std::fs::write(&config_path, &self.config.raw_config).map_err(VpnError::Io)?;
std::fs::write(&config_path, self.build_runtime_config()).map_err(VpnError::Io)?;
#[cfg(unix)]
{
@@ -84,43 +548,74 @@ impl OpenVpnSocks5Server {
let _ = std::fs::set_permissions(&config_path, std::fs::Permissions::from_mode(0o600));
}
// Find a management port
let mgmt_listener = std::net::TcpListener::bind("127.0.0.1:0")
.map_err(|e| VpnError::Connection(format!("Failed to bind management port: {e}")))?;
let mgmt_port = mgmt_listener
.local_addr()
.map_err(|e| VpnError::Connection(format!("Failed to get management port: {e}")))?
.port();
drop(mgmt_listener);
let mgmt_port = if supports_management {
let mgmt_listener = std::net::TcpListener::bind("127.0.0.1:0")
.map_err(|e| VpnError::Connection(format!("Failed to bind management port: {e}")))?;
let port = mgmt_listener
.local_addr()
.map_err(|e| VpnError::Connection(format!("Failed to get management port: {e}")))?
.port();
drop(mgmt_listener);
Some(port)
} else {
log::info!(
"[vpn-worker] OpenVPN build does not support management; using log-based readiness"
);
None
};
let openvpn_log_path = std::env::temp_dir().join(format!("openvpn-{}.log", config_id));
let log_file = std::fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(&openvpn_log_path)
.map_err(VpnError::Io)?;
// Start OpenVPN with SOCKS proxy mode
let mut cmd = Command::new(&openvpn_bin);
cmd.arg("--config").arg(&config_path);
if let Some(mgmt_port) = mgmt_port {
cmd
.arg("--management")
.arg("127.0.0.1")
.arg(mgmt_port.to_string());
}
cmd
.arg("--config")
.arg(&config_path)
.arg("--management")
.arg("127.0.0.1")
.arg(mgmt_port.to_string())
.arg("--socks-proxy")
.arg("127.0.0.1")
.arg(self.port.to_string())
.arg("--verb")
.arg("3")
.stdout(Stdio::piped())
.stderr(Stdio::piped());
.stdout(
log_file
.try_clone()
.map(Stdio::from)
.map_err(VpnError::Io)?,
)
.stderr(Stdio::from(log_file));
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
const CREATE_NO_WINDOW: u32 = 0x08000000;
cmd.arg("--disable-dco");
if self.config.dev_type.starts_with("tun") {
cmd.arg("--windows-driver").arg("wintun");
}
cmd.creation_flags(CREATE_NO_WINDOW);
}
let mut child = cmd
.spawn()
.map_err(|e| VpnError::Connection(format!("Failed to start OpenVPN: {e}")))?;
// Wait for OpenVPN to start
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
match child.try_wait() {
Ok(Some(status)) => {
let _ = std::fs::remove_file(&config_path);
return Err(VpnError::Connection(format!(
"OpenVPN exited early with status: {status}. OpenVPN requires elevated privileges (sudo/admin)."
"OpenVPN exited immediately (status: {}). Last output:\n{}",
status,
Self::read_log_tail(&openvpn_log_path, 20)
)));
}
Ok(None) => {}
@@ -132,8 +627,15 @@ impl OpenVpnSocks5Server {
}
}
// Start a basic SOCKS5 proxy that tunnels through the OpenVPN TUN interface
let listener = TcpListener::bind(format!("127.0.0.1:{}", self.port))
let vpn_bind_ip = if let Some(mgmt_port) = mgmt_port {
Self::wait_for_openvpn_ready_via_management(&mut child, mgmt_port, &openvpn_log_path).await?
} else {
Self::wait_for_openvpn_ready_via_log(&mut child, &openvpn_log_path).await?
}
.unwrap_or(Ipv4Addr::UNSPECIFIED);
let vpn_bind_ip = Arc::new(vpn_bind_ip);
let listener = TcpListener::bind(("127.0.0.1", self.port))
.await
.map_err(|e| VpnError::Connection(format!("Failed to bind SOCKS5: {e}")))?;
@@ -142,10 +644,10 @@ impl OpenVpnSocks5Server {
.map_err(|e| VpnError::Connection(format!("Failed to get local addr: {e}")))?
.port();
if let Some(mut wc) = crate::vpn_worker_storage::get_vpn_worker_config(&config_id) {
wc.local_port = Some(actual_port);
wc.local_url = Some(format!("socks5://127.0.0.1:{}", actual_port));
let _ = crate::vpn_worker_storage::save_vpn_worker_config(&wc);
if let Some(mut worker_config) = crate::vpn_worker_storage::get_vpn_worker_config(&config_id) {
worker_config.local_port = Some(actual_port);
worker_config.local_url = Some(format!("socks5://127.0.0.1:{}", actual_port));
let _ = crate::vpn_worker_storage::save_vpn_worker_config(&worker_config);
}
log::info!(
@@ -156,10 +658,13 @@ impl OpenVpnSocks5Server {
loop {
match listener.accept().await {
Ok((client, _)) => {
tokio::spawn(Self::handle_socks5_client(client));
let bind_ip = vpn_bind_ip.clone();
tokio::spawn(async move {
let _ = Self::handle_socks5_client(client, bind_ip).await;
});
}
Err(e) => {
log::warn!("[vpn-worker] Accept error: {e}");
Err(error) => {
log::warn!("[vpn-worker] Accept error: {error}");
}
}
}
@@ -167,53 +672,119 @@ impl OpenVpnSocks5Server {
async fn handle_socks5_client(
mut client: TcpStream,
vpn_bind_ip: Arc<Ipv4Addr>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// SOCKS5 greeting
let mut buf = [0u8; 256];
let n = client.read(&mut buf).await?;
if n < 3 || buf[0] != 0x05 {
let mut greeting = [0u8; 2];
if let Err(error) = client.read_exact(&mut greeting).await {
if error.kind() != std::io::ErrorKind::UnexpectedEof {
log::debug!("[socks5] Failed to read greeting header: {}", error);
}
return Ok(());
}
if greeting[0] != 0x05 {
return Ok(());
}
let mut methods = vec![0u8; greeting[1] as usize];
if let Err(error) = client.read_exact(&mut methods).await {
if error.kind() != std::io::ErrorKind::UnexpectedEof {
log::debug!("[socks5] Failed to read methods list: {}", error);
}
return Ok(());
}
client.write_all(&[0x05, 0x00]).await?;
// SOCKS5 connect request
let n = client.read(&mut buf).await?;
if n < 10 || buf[0] != 0x05 || buf[1] != 0x01 {
let mut request_header = [0u8; 4];
if let Err(error) = client.read_exact(&mut request_header).await {
if error.kind() != std::io::ErrorKind::UnexpectedEof {
log::debug!("[socks5] Failed to read request header: {}", error);
}
return Ok(());
}
let dest_addr = match buf[3] {
if request_header[0] != 0x05 {
return Ok(());
}
if request_header[1] != 0x01 {
let _ = client
.write_all(&[0x05, 0x07, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
.await;
return Ok(());
}
let target = match request_header[3] {
0x01 => {
let ip = std::net::Ipv4Addr::new(buf[4], buf[5], buf[6], buf[7]);
let port = u16::from_be_bytes([buf[8], buf[9]]);
format!("{}:{}", ip, port)
let mut addr_port = [0u8; 6];
client.read_exact(&mut addr_port).await?;
SocksTarget::Address(SocketAddr::new(
IpAddr::V4(Ipv4Addr::new(
addr_port[0],
addr_port[1],
addr_port[2],
addr_port[3],
)),
u16::from_be_bytes([addr_port[4], addr_port[5]]),
))
}
0x03 => {
let domain_len = buf[4] as usize;
let domain = String::from_utf8_lossy(&buf[5..5 + domain_len]).to_string();
let port_start = 5 + domain_len;
let port = u16::from_be_bytes([buf[port_start], buf[port_start + 1]]);
format!("{}:{}", domain, port)
let mut len = [0u8; 1];
client.read_exact(&mut len).await?;
if len[0] == 0 {
return Ok(());
}
let mut domain = vec![0u8; len[0] as usize];
client.read_exact(&mut domain).await?;
let mut port = [0u8; 2];
client.read_exact(&mut port).await?;
SocksTarget::Domain(
String::from_utf8_lossy(&domain).to_string(),
u16::from_be_bytes(port),
)
}
0x04 => {
let mut addr_port = [0u8; 18];
client.read_exact(&mut addr_port).await?;
let mut octets = [0u8; 16];
octets.copy_from_slice(&addr_port[..16]);
SocksTarget::Address(SocketAddr::new(
IpAddr::V6(std::net::Ipv6Addr::from(octets)),
u16::from_be_bytes([addr_port[16], addr_port[17]]),
))
}
_ => {
let _ = client
.write_all(&[0x05, 0x08, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
.await;
return Ok(());
}
_ => return Ok(()),
};
// Connect to destination through OpenVPN tunnel (OS routing handles it)
match TcpStream::connect(&dest_addr).await {
Ok(upstream) => {
match Self::connect_target(target, *vpn_bind_ip).await {
Ok((upstream, _address)) => {
client
.write_all(&[0x05, 0x00, 0x00, 0x01, 127, 0, 0, 1, 0, 0])
.await?;
let (mut cr, mut cw) = client.into_split();
let (mut ur, mut uw) = upstream.into_split();
let (mut client_read, mut client_write) = client.into_split();
let (mut upstream_read, mut upstream_write) = upstream.into_split();
let c2u = tokio::io::copy(&mut cr, &mut uw);
let u2c = tokio::io::copy(&mut ur, &mut cw);
let _ = tokio::try_join!(c2u, u2c);
let client_to_upstream = tokio::io::copy(&mut client_read, &mut upstream_write);
let upstream_to_client = tokio::io::copy(&mut upstream_read, &mut client_write);
let _ = tokio::try_join!(client_to_upstream, upstream_to_client)?;
}
Err(_) => {
Err(error) => {
log::debug!(
"[socks5] Failed to connect through OpenVPN tunnel: {}",
error
);
client
.write_all(&[0x05, 0x05, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
.await?;
src-tauri/src/vpn/socks5_server.rs
+40 -19
View File
@@ -370,6 +370,8 @@ impl WireGuardSocks5Server {
smol_handle: SocketHandle,
tcp_stream: TcpStream,
socks_done: bool,
connecting: bool,
greeting_done: bool,
read_buf: Vec<u8>,
dest_addr: Option<SocketAddr>,
}
@@ -391,6 +393,8 @@ impl WireGuardSocks5Server {
smol_handle: handle,
tcp_stream: stream,
socks_done: false,
connecting: false,
greeting_done: false,
read_buf: Vec::new(),
dest_addr: None,
});
@@ -409,7 +413,30 @@ impl WireGuardSocks5Server {
// Process each connection
let mut completed = Vec::new();
for (idx, conn) in connections.iter_mut().enumerate() {
if !conn.socks_done {
if conn.connecting {
let socket = sockets.get_mut::<TcpSocket>(conn.smol_handle);
if socket.may_send() {
let _ = conn.tcp_stream.try_write(&[
0x05,
0x00,
0x00,
0x01,
127,
0,
0,
1,
(actual_port >> 8) as u8,
(actual_port & 0xff) as u8,
]);
conn.connecting = false;
conn.socks_done = true;
} else if !socket.is_open() {
let _ = conn
.tcp_stream
.try_write(&[0x05, 0x05, 0x00, 0x01, 0, 0, 0, 0, 0, 0]);
completed.push(idx);
}
} else if !conn.socks_done {
// Handle SOCKS5 handshake
let mut buf = [0u8; 512];
match conn.tcp_stream.try_read(&mut buf) {
@@ -427,19 +454,26 @@ impl WireGuardSocks5Server {
}
}
if conn.dest_addr.is_none() && conn.read_buf.len() >= 3 {
if !conn.greeting_done && conn.read_buf.len() >= 3 {
// SOCKS5 greeting: version, nmethods, methods
if conn.read_buf[0] != 0x05 {
completed.push(idx);
continue;
}
// Reply: no auth required
let _ = conn.tcp_stream.try_write(&[0x05, 0x00]);
let nmethods = conn.read_buf[1] as usize;
if conn.read_buf.len() < 2 + nmethods {
continue;
}
// Reply: no auth required
if conn.tcp_stream.try_write(&[0x05, 0x00]).is_err() {
completed.push(idx);
continue;
}
conn.read_buf.drain(..2 + nmethods);
conn.greeting_done = true;
}
if conn.dest_addr.is_none() && conn.read_buf.len() >= 10 {
if conn.greeting_done && conn.dest_addr.is_none() && conn.read_buf.len() >= 10 {
// SOCKS5 connect request
if conn.read_buf[0] != 0x05 || conn.read_buf[1] != 0x01 {
completed.push(idx);
@@ -539,20 +573,7 @@ impl WireGuardSocks5Server {
continue;
}
// Send SOCKS5 success reply
let _ = conn.tcp_stream.try_write(&[
0x05,
0x00,
0x00,
0x01,
127,
0,
0,
1,
(actual_port >> 8) as u8,
(actual_port & 0xff) as u8,
]);
conn.socks_done = true;
conn.connecting = true;
}
} else {
// Data relay between SOCKS5 client and smoltcp socket
src-tauri/src/vpn_worker_runner.rs
+116 -46
View File
@@ -1,3 +1,4 @@
use crate::proxy_runner::find_sidecar_executable;
use crate::proxy_storage::is_process_running;
use crate::vpn_worker_storage::{
delete_vpn_worker_config, find_vpn_worker_by_vpn_id, generate_vpn_worker_id,
@@ -5,12 +6,124 @@ use crate::vpn_worker_storage::{
};
use std::process::Stdio;
const VPN_WORKER_POLL_INTERVAL_MS: u64 = 100;
const VPN_WORKER_STARTUP_TIMEOUT_MS: u64 = 10_000;
const OPENVPN_WORKER_STARTUP_TIMEOUT_MS: u64 = 100_000;
async fn vpn_worker_accepting_connections(config: &VpnWorkerConfig) -> bool {
let Some(port) = config.local_port else {
return false;
};
if config
.local_url
.as_ref()
.is_none_or(|local_url| local_url.is_empty())
{
return false;
}
matches!(
tokio::time::timeout(
tokio::time::Duration::from_millis(VPN_WORKER_POLL_INTERVAL_MS),
tokio::net::TcpStream::connect(("127.0.0.1", port)),
)
.await,
Ok(Ok(_))
)
}
fn worker_log_path(id: &str) -> std::path::PathBuf {
std::env::temp_dir().join(format!("donut-vpn-{}.log", id))
}
fn read_worker_log(id: &str) -> String {
std::fs::read_to_string(worker_log_path(id)).unwrap_or_else(|_| "No log available".to_string())
}
async fn wait_for_vpn_worker_ready(
id: &str,
vpn_type: &str,
) -> Result<VpnWorkerConfig, Box<dyn std::error::Error>> {
let startup_timeout = if vpn_type == "openvpn" {
tokio::time::Duration::from_millis(OPENVPN_WORKER_STARTUP_TIMEOUT_MS)
} else {
tokio::time::Duration::from_millis(VPN_WORKER_STARTUP_TIMEOUT_MS)
};
let startup_deadline = tokio::time::Instant::now() + startup_timeout;
tokio::time::sleep(tokio::time::Duration::from_millis(
VPN_WORKER_POLL_INTERVAL_MS,
))
.await;
let mut attempts = 0u32;
loop {
tokio::time::sleep(tokio::time::Duration::from_millis(
VPN_WORKER_POLL_INTERVAL_MS,
))
.await;
if let Some(updated_config) = get_vpn_worker_config(id) {
let process_running = updated_config.pid.map(is_process_running).unwrap_or(false);
if !process_running && attempts > 2 {
let log_output = read_worker_log(id);
delete_vpn_worker_config(id);
return Err(format!("VPN worker process crashed. Log output:\n{}", log_output).into());
}
if vpn_worker_accepting_connections(&updated_config).await {
return Ok(updated_config);
}
}
attempts += 1;
if tokio::time::Instant::now() >= startup_deadline {
if let Some(config) = get_vpn_worker_config(id) {
let process_running = config.pid.map(is_process_running).unwrap_or(false);
let log_output = read_worker_log(id);
delete_vpn_worker_config(id);
return Err(
format!(
"VPN worker failed to start within {:.1}s. pid={:?}, process_running={}, local_url={:?}\n\nVPN worker log:\n{}",
startup_timeout.as_secs_f32(),
config.pid,
process_running,
config.local_url,
log_output
)
.into(),
);
}
delete_vpn_worker_config(id);
return Err("VPN worker config not found after spawn".into());
}
}
}
pub async fn start_vpn_worker(vpn_id: &str) -> Result<VpnWorkerConfig, Box<dyn std::error::Error>> {
for config in list_vpn_worker_configs() {
if let Some(pid) = config.pid {
if !is_process_running(pid) {
delete_vpn_worker_config(&config.id);
}
} else {
delete_vpn_worker_config(&config.id);
}
}
// Check if a VPN worker for this vpn_id already exists and is running
if let Some(existing) = find_vpn_worker_by_vpn_id(vpn_id) {
if let Some(pid) = existing.pid {
if is_process_running(pid) {
return Ok(existing);
if vpn_worker_accepting_connections(&existing).await {
return Ok(existing);
}
return wait_for_vpn_worker_ready(&existing.id, &existing.vpn_type).await;
}
}
// Worker config exists but process is dead, clean up
@@ -63,7 +176,7 @@ pub async fn start_vpn_worker(vpn_id: &str) -> Result<VpnWorkerConfig, Box<dyn s
save_vpn_worker_config(&config)?;
// Spawn detached VPN worker process
let exe = std::env::current_exe()?;
let exe = find_sidecar_executable("donut-proxy")?;
#[cfg(unix)]
{
@@ -149,50 +262,7 @@ pub async fn start_vpn_worker(vpn_id: &str) -> Result<VpnWorkerConfig, Box<dyn s
drop(child);
}
// Wait for the worker to update config with local_url
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
let mut attempts = 0;
let max_attempts = 100; // 10 seconds max
loop {
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
if let Some(updated_config) = get_vpn_worker_config(&id) {
if let Some(ref local_url) = updated_config.local_url {
if !local_url.is_empty() {
if let Some(port) = updated_config.local_port {
if let Ok(Ok(_)) = tokio::time::timeout(
tokio::time::Duration::from_millis(100),
tokio::net::TcpStream::connect(("127.0.0.1", port)),
)
.await
{
return Ok(updated_config);
}
}
}
}
}
attempts += 1;
if attempts >= max_attempts {
if let Some(config) = get_vpn_worker_config(&id) {
let process_running = config.pid.map(is_process_running).unwrap_or(false);
// Clean up on failure
delete_vpn_worker_config(&id);
return Err(
format!(
"VPN worker failed to start in time. pid={:?}, process_running={}, local_url={:?}",
config.pid, process_running, config.local_url
)
.into(),
);
}
delete_vpn_worker_config(&id);
return Err("VPN worker config not found after spawn".into());
}
}
wait_for_vpn_worker_ready(&id, vpn_type_str).await
}
pub async fn stop_vpn_worker(id: &str) -> Result<bool, Box<dyn std::error::Error>> {