gsadmin/Firelink
1
0
Fork 0
mirror of https://github.com/nimbold/Firelink.git synced 2026-08-21 16:36:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(security): resolve all medium severity vulnerabilities

Browse Source 
- Ensure yt-dlp partial files are cleaned up on error exits.
- Handle Semaphore closure gracefully without panicking.
- Replace busy-polling in wait_permit_released with Notify.
- Fix extension server silent failure with retry loop and logging.
- Ignore malformed existing scheduler settings instead of failing.
- Add final symlink checks before revealing/opening downloads.
- Delay listener registration in App.tsx until core is ready.
- Sanitize Referer header to prevent injection.
This commit is contained in:
NimBold
2026-06-25 00:19:18 +03:30
parent 187d70f348
commit 8a258981ce
6 changed files with 221 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands.rs
+8
View File
@@ -15,6 +15,10 @@ pub async fn reveal_in_file_manager(
) )
})?; })?;
if std::fs::symlink_metadata(&path).is_ok_and(|metadata| metadata.file_type().is_symlink()) {
return Err("Download path was replaced by a symlink before reveal".to_string());
}
app_handle app_handle
.opener() .opener()
.reveal_item_in_dir(path.to_string_lossy().as_ref()) .reveal_item_in_dir(path.to_string_lossy().as_ref())
@@ -33,6 +37,10 @@ pub async fn open_downloaded_file(
return Err(format!("Downloaded file is missing: {}", path.display())); return Err(format!("Downloaded file is missing: {}", path.display()));
} }
if std::fs::symlink_metadata(&path).is_ok_and(|metadata| metadata.file_type().is_symlink()) {
return Err("Download path was replaced by a symlink before open".to_string());
}
app_handle app_handle
.opener() .opener()
.open_path(path.to_string_lossy().as_ref(), None::<String>) .open_path(path.to_string_lossy().as_ref(), None::<String>)
src-tauri/src/lib.rs
+174 -131
View File
@@ -745,6 +745,9 @@ async fn validate_url_ssrf(url: &str) -> Result<Option<(String, std::net::Socket
if (ipv6.segments()[0] & 0xfe00) == 0xfc00 { // ULA check if (ipv6.segments()[0] & 0xfe00) == 0xfc00 { // ULA check
return Err("SSRF blocked: Private/local IP not allowed".to_string()); return Err("SSRF blocked: Private/local IP not allowed".to_string());
} }
if (ipv6.segments()[0] & 0xffc0) == 0xfe80 { // Link-local check
return Err("SSRF blocked: Private/local IP not allowed".to_string());
}
} }
} }
Ok(Some((host.to_string(), addr))) Ok(Some((host.to_string(), addr)))
@@ -1317,7 +1320,7 @@ pub struct AppState {
pub extension_frontend_ready: extension_server::SharedFrontendReady, pub extension_frontend_ready: extension_server::SharedFrontendReady,
pub extension_server_port: extension_server::SharedServerPort, pub extension_server_port: extension_server::SharedServerPort,
pub extension_server_shutdown: tokio::sync::watch::Sender<bool>, pub extension_server_shutdown: tokio::sync::watch::Sender<bool>,
pub aria2_port: u16, pub aria2_port: std::sync::Arc<std::sync::atomic::AtomicU16>,
pub aria2_secret: String, pub aria2_secret: String,
pub media_semaphore: Arc<tokio::sync::Semaphore>, pub media_semaphore: Arc<tokio::sync::Semaphore>,
pub sleep_preventer: Arc<Mutex<Option<keepawake::KeepAwake>>>, pub sleep_preventer: Arc<Mutex<Option<keepawake::KeepAwake>>>,
@@ -1551,7 +1554,7 @@ async fn test_aria2c(app_handle: tauri::AppHandle, state: tauri::State<'_, AppSt
} }
let result = rpc_call( let result = rpc_call(
state.aria2_port, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret, &state.aria2_secret,
"aria2.getVersion", "aria2.getVersion",
serde_json::json!([]), serde_json::json!([]),
@@ -1926,7 +1929,7 @@ async fn get_engine_status(
app_handle: tauri::AppHandle, app_handle: tauri::AppHandle,
state: tauri::State<'_, AppState>, state: tauri::State<'_, AppState>,
) -> Result<EngineStatusResult, String> { ) -> Result<EngineStatusResult, String> {
let port = state.aria2_port; let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed);
let secret = state.aria2_secret.clone(); let secret = state.aria2_secret.clone();
let (aria2, ytdlp, ffmpeg, deno) = tokio::join!( let (aria2, ytdlp, ffmpeg, deno) = tokio::join!(
@@ -1946,7 +1949,7 @@ async fn get_aria2_engine_status(
app_handle: tauri::AppHandle, app_handle: tauri::AppHandle,
state: tauri::State<'_, AppState>, state: tauri::State<'_, AppState>,
) -> Result<EngineStatusItem, String> { ) -> Result<EngineStatusItem, String> {
Ok(check_aria2(&app_handle, state.aria2_port, &state.aria2_secret).await) Ok(check_aria2(&app_handle, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret).await)
} }
#[tauri::command] #[tauri::command]
@@ -2072,7 +2075,7 @@ pub(crate) async fn start_media_download_internal(
.arg("--progress-delta").arg("0.2") .arg("--progress-delta").arg("0.2")
.arg("--progress-template") .arg("--progress-template")
.arg(format!("download:{MEDIA_PROGRESS_PREFIX}%(progress)j")) .arg(format!("download:{MEDIA_PROGRESS_PREFIX}%(progress)j"))
.arg("--no-check-formats")
.arg("--socket-timeout").arg("20") .arg("--socket-timeout").arg("20")
.arg("--retries").arg("3") .arg("--retries").arg("3")
.arg("--extractor-retries").arg("3") .arg("--extractor-retries").arg("3")
@@ -2242,6 +2245,7 @@ pub(crate) async fn start_media_download_internal(
} }
Some(tauri_plugin_shell::process::CommandEvent::Error(err)) => { Some(tauri_plugin_shell::process::CommandEvent::Error(err)) => {
log::error!("yt-dlp shell error [{}]: {}", id, err); log::error!("yt-dlp shell error [{}]: {}", id, err);
cleanup_media_artifacts(&out_path, false).await;
break err; break err;
} }
Some(tauri_plugin_shell::process::CommandEvent::Terminated(payload)) => { Some(tauri_plugin_shell::process::CommandEvent::Terminated(payload)) => {
@@ -2262,6 +2266,7 @@ pub(crate) async fn start_media_download_internal(
return Ok(()); return Ok(());
} }
log::error!("yt-dlp exited with non-zero code {:?} for id: {}", payload.code, id); log::error!("yt-dlp exited with non-zero code {:?} for id: {}", payload.code, id);
cleanup_media_artifacts(&out_path, false).await;
break if stderr_tail.is_empty() { break if stderr_tail.is_empty() {
format!("yt-dlp exited with code {:?}", payload.code) format!("yt-dlp exited with code {:?}", payload.code)
} else { } else {
@@ -2270,6 +2275,7 @@ pub(crate) async fn start_media_download_internal(
} }
Some(_) => {} Some(_) => {}
None => { None => {
cleanup_media_artifacts(&out_path, false).await;
break if stderr_tail.is_empty() { break if stderr_tail.is_empty() {
"yt-dlp process ended unexpectedly".to_string() "yt-dlp process ended unexpectedly".to_string()
} else { } else {
@@ -2325,7 +2331,7 @@ async fn pause_download(
let gid = state.queue_manager.aria2_gid_for_download(&id); let gid = state.queue_manager.aria2_gid_for_download(&id);
if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) { if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) {
let status = aria2_download_status( let status = aria2_download_status(
state.aria2_port, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret, &state.aria2_secret,
gid, gid,
) )
@@ -2336,7 +2342,7 @@ async fn pause_download(
} }
"active" | "waiting" => { "active" | "waiting" => {
let result = rpc_call( let result = rpc_call(
state.aria2_port, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret, &state.aria2_secret,
"aria2.forcePause", "aria2.forcePause",
serde_json::json!([gid]), serde_json::json!([gid]),
@@ -2412,7 +2418,7 @@ async fn resume_download(
return Ok(false); return Ok(false);
} }
let status = aria2_download_status(state.aria2_port, &state.aria2_secret, &gid).await?; let status = aria2_download_status(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, &gid).await?;
match status.as_str() { match status.as_str() {
"paused" => { "paused" => {
use tauri::Emitter; use tauri::Emitter;
@@ -2422,11 +2428,12 @@ async fn resume_download(
); );
let queue_manager = state.queue_manager.clone(); let queue_manager = state.queue_manager.clone();
let aria2_port = state.aria2_port; let aria2_port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed);
let aria2_secret = state.aria2_secret.clone(); let aria2_secret = state.aria2_secret.clone();
let id_clone = id.clone(); let id_clone = id.clone();
let gid_clone = gid.clone(); let gid_clone = gid.clone();
let app_handle_clone = app_handle.clone();
tauri::async_runtime::spawn(async move { tauri::async_runtime::spawn(async move {
let acquired = queue_manager.ensure_aria2_permit(&id_clone).await; let acquired = queue_manager.ensure_aria2_permit(&id_clone).await;
let result = match rpc_call( let result = match rpc_call(
@@ -2443,6 +2450,13 @@ async fn resume_download(
queue_manager.release_permit(&id_clone).await; queue_manager.release_permit(&id_clone).await;
} }
log::error!("failed to resume aria2 gid {}: {}", gid_clone, error); log::error!("failed to resume aria2 gid {}: {}", gid_clone, error);
let _ = app_handle_clone.emit(
"download-state",
crate::ipc::DownloadStateEvent::new(
&id_clone,
crate::ipc::DownloadStatus::Failed,
),
);
return; return;
} }
}; };
@@ -2451,6 +2465,13 @@ async fn resume_download(
queue_manager.release_permit(&id_clone).await; queue_manager.release_permit(&id_clone).await;
} }
log::error!("failed to resume aria2 gid {}: {}", gid_clone, error); log::error!("failed to resume aria2 gid {}: {}", gid_clone, error);
let _ = app_handle_clone.emit(
"download-state",
crate::ipc::DownloadStateEvent::new(
&id_clone,
crate::ipc::DownloadStatus::Failed,
),
);
return; return;
} }
log::info!("aria2 resume [{}]: unpaused gid {}", id_clone, gid_clone); log::info!("aria2 resume [{}]: unpaused gid {}", id_clone, gid_clone);
@@ -2510,8 +2531,8 @@ async fn remove_download(
let gid = state.queue_manager.aria2_gid_for_download(&id); let gid = state.queue_manager.aria2_gid_for_download(&id);
if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) { if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) {
let removal_result = async { let removal_result = async {
force_remove_aria2_gid(state.aria2_port, &state.aria2_secret, gid).await?; force_remove_aria2_gid(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid).await?;
wait_for_aria2_stopped(state.aria2_port, &state.aria2_secret, gid).await wait_for_aria2_stopped(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid).await
} }
.await; .await;
if let Err(error) = removal_result { if let Err(error) = removal_result {
@@ -2605,13 +2626,13 @@ async fn detach_download_for_reconfigure(
if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) { if let Some(gid) = gid.as_deref().filter(|gid| !gid.starts_with("native:")) {
let removal_result = async { let removal_result = async {
rpc_call( rpc_call(
state.aria2_port, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret, &state.aria2_secret,
"aria2.forcePause", "aria2.forcePause",
serde_json::json!([gid]), serde_json::json!([gid]),
) )
.await?; .await?;
wait_for_aria2_stopped(state.aria2_port, &state.aria2_secret, gid).await wait_for_aria2_stopped(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, gid).await
} }
.await; .await;
if let Err(error) = removal_result { if let Err(error) = removal_result {
@@ -2956,7 +2977,7 @@ async fn set_global_speed_limit(state: tauri::State<'_, AppState>, limit: Option
.and_then(normalize_speed_limit_for_aria2) .and_then(normalize_speed_limit_for_aria2)
.unwrap_or_else(|| "0".to_string()); .unwrap_or_else(|| "0".to_string());
rpc_call( rpc_call(
state.aria2_port, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret, &state.aria2_secret,
"aria2.changeGlobalOption", "aria2.changeGlobalOption",
serde_json::json!([{"max-overall-download-limit": limit_str}]) serde_json::json!([{"max-overall-download-limit": limit_str}])
@@ -3313,7 +3334,7 @@ async fn read_logs(app_handle: tauri::AppHandle, limit: usize) -> Result<Vec<Str
#[tauri::command] #[tauri::command]
async fn clear_logs(app_handle: tauri::AppHandle) -> Result<(), String> { async fn clear_logs(app_handle: tauri::AppHandle) -> Result<(), String> {
for file in log_files(&app_handle).await? { for file in log_files(&app_handle).await? {
let _ = tokio::fs::write(&file, "").await; tokio::fs::write(&file, "").await.map_err(|e| format!("Failed to clear log file {:?}: {}", file, e))?;
} }
Ok(()) Ok(())
} }
@@ -3332,7 +3353,7 @@ async fn export_logs(
chrono::Utc::now().to_rfc3339(), chrono::Utc::now().to_rfc3339(),
); );
let (aria2, ytdlp, ffmpeg, deno) = tokio::join!( let (aria2, ytdlp, ffmpeg, deno) = tokio::join!(
check_aria2(&app_handle, state.aria2_port, &state.aria2_secret), check_aria2(&app_handle, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret),
check_ytdlp(&app_handle), check_ytdlp(&app_handle),
check_ffmpeg(&app_handle), check_ffmpeg(&app_handle),
check_deno(&app_handle), check_deno(&app_handle),
@@ -3883,10 +3904,9 @@ pub fn run() {
let server_extension_port = extension_server_port.clone(); let server_extension_port = extension_server_port.clone();
let (extension_server_shutdown_tx, extension_server_shutdown_rx) = tokio::sync::watch::channel(false); let (extension_server_shutdown_tx, extension_server_shutdown_rx) = tokio::sync::watch::channel(false);
let aria2_port = std::net::TcpListener::bind("127.0.0.1:0") let initial_aria2_port = 6800; // Will be determined dynamically in background
.and_then(|listener| listener.local_addr()) let aria2_port = Arc::new(std::sync::atomic::AtomicU16::new(initial_aria2_port));
.map(|addr| addr.port()) let aria2_port_clone = Arc::clone(&aria2_port);
.unwrap_or(6800);
let aria2_secret = uuid::Uuid::new_v4().to_string(); let aria2_secret = uuid::Uuid::new_v4().to_string();
tauri::Builder::default() tauri::Builder::default()
.plugin(tauri_plugin_single_instance::init(|app, _args, _cwd| { .plugin(tauri_plugin_single_instance::init(|app, _args, _cwd| {
@@ -3956,7 +3976,7 @@ pub fn run() {
extension_frontend_ready, extension_frontend_ready,
extension_server_port, extension_server_port,
extension_server_shutdown: extension_server_shutdown_tx.clone(), extension_server_shutdown: extension_server_shutdown_tx.clone(),
aria2_port, aria2_port: aria2_port.clone(),
aria2_secret: aria2_secret.clone(), aria2_secret: aria2_secret.clone(),
media_semaphore: Arc::new(tokio::sync::Semaphore::new(3)), media_semaphore: Arc::new(tokio::sync::Semaphore::new(3)),
sleep_preventer: Arc::new(Mutex::new(None)), sleep_preventer: Arc::new(Mutex::new(None)),
@@ -4016,123 +4036,141 @@ pub fn run() {
.map(|settings| settings.global_speed_limit) .map(|settings| settings.global_speed_limit)
.unwrap_or_default(); .unwrap_or_default();
match resolve_bundled_binary_path(app.handle(), "aria2c") { let aria2_secret_clone = aria2_secret.clone();
Ok(binary_path) => { let app_handle_bg = app.handle().clone();
let mut cmd = std::process::Command::new(&binary_path); tauri::async_runtime::spawn(async move {
let mut config_file = tempfile::Builder::new().prefix("aria2-").suffix(".conf").tempfile().expect("failed to create aria2 config file");
use std::io::Write;
let config_content = format!("rpc-secret={}\n", aria2_secret);
config_file.write_all(config_content.as_bytes()).expect("failed to write aria2 config file");
let config_path = config_file.into_temp_path();
cmd.arg("--enable-rpc=true") let mut ws_port = 6800;
.arg(format!("--conf-path={}", config_path.display())) match resolve_bundled_binary_path(&app_handle_bg, "aria2c") {
.arg(format!("--rpc-listen-port={}", aria2_port)) Ok(binary_path) => {
.arg("--rpc-listen-all=false") let mut success = false;
.arg("--continue=true") for attempt_port in 6800..6900 {
.arg("--retry-wait=2") let mut cmd = std::process::Command::new(&binary_path);
.arg("--allow-overwrite=false")
.arg("--summary-interval=1") let mut config_file = tempfile::Builder::new().prefix("aria2-").suffix(".conf").tempfile().expect("failed to create aria2 config file");
.arg("--console-log-level=warn") use std::io::Write;
.arg("--download-result=hide") let config_content = format!("rpc-secret={}\n", aria2_secret_clone);
.arg("--max-concurrent-downloads=9999") config_file.write_all(config_content.as_bytes()).expect("failed to write aria2 config file");
.arg("--check-certificate=true"); let config_path = config_file.into_temp_path();
if let Some(global_speed_limit) = normalize_speed_limit_for_aria2(&global_speed_limit) { cmd.arg("--enable-rpc=true")
cmd.arg(format!("--max-overall-download-limit={}", global_speed_limit)); .arg(format!("--conf-path={}", config_path.display()))
} .arg(format!("--rpc-listen-port={}", attempt_port))
.arg("--rpc-listen-all=false")
cmd.stdout(std::process::Stdio::null()); .arg("--continue=true")
cmd.stderr(std::process::Stdio::piped()); .arg("--retry-wait=2")
.arg("--allow-overwrite=false")
match cmd.spawn() { .arg("--summary-interval=1")
Ok(mut child) => { .arg("--console-log-level=warn")
log::info!("aria2c spawned successfully on port {}", aria2_port); .arg("--download-result=hide")
.arg("--max-concurrent-downloads=9999")
let daemon_app = app.handle().clone(); .arg("--check-certificate=true");
if let Some(stderr) = child.stderr.take() {
std::thread::spawn(move || { if let Some(limit) = normalize_speed_limit_for_aria2(&global_speed_limit) {
use std::io::BufRead; cmd.arg(format!("--max-overall-download-limit={}", limit));
let reader = std::io::BufReader::new(stderr);
for line in reader.lines().map_while(Result::ok) {
let trimmed = line.trim().to_string();
if let Ok(mut stderr_lock) = daemon_app.state::<Aria2DaemonGuard>().last_stderr.lock() {
stderr_lock.push_str(&trimmed);
stderr_lock.push('\n');
let excess = stderr_lock.len().saturating_sub(8192);
if excess > 0 {
let _ = stderr_lock.drain(..excess);
}
}
let lower = trimmed.to_lowercase();
if lower.contains("error") || lower.contains("critical") {
log::error!("aria2c stderr: {}", trimmed);
}
}
});
} }
cmd.stdout(std::process::Stdio::null());
cmd.stderr(std::process::Stdio::piped());
match cmd.spawn() {
Ok(mut child) => {
// Give it a moment to fail if port is in use
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
if let Ok(Some(_)) = child.try_wait() {
// Process exited, likely port collision, try next
continue;
}
let guard = app.state::<Aria2DaemonGuard>(); log::info!("aria2c spawned successfully on port {}", attempt_port);
*guard.child.lock().unwrap() = Some(child);
*guard.config_path.lock().unwrap() = Some(config_path);
let port = aria2_port; aria2_port_clone.store(attempt_port, std::sync::atomic::Ordering::Relaxed);
let secret = aria2_secret.clone(); ws_port = attempt_port;
let start = std::time::Instant::now(); success = true;
let ready = tauri::async_runtime::block_on(async {
let mut last_err = String::new(); let daemon_app = app_handle_bg.clone();
loop { if let Some(stderr) = child.stderr.take() {
if start.elapsed() > std::time::Duration::from_secs(5) { std::thread::spawn(move || {
return Err(if last_err.is_empty() { use std::io::BufRead;
"aria2 daemon did not become ready within 5 seconds".to_string() let reader = std::io::BufReader::new(stderr);
} else { for line in reader.lines().map_while(Result::ok) {
format!("aria2 did not become ready: {last_err}") let trimmed = line.trim().to_string();
if let Ok(mut stderr_lock) = daemon_app.state::<Aria2DaemonGuard>().last_stderr.lock() {
stderr_lock.push_str(&trimmed);
stderr_lock.push('\n');
let excess = stderr_lock.len().saturating_sub(8192);
if excess > 0 {
let _ = stderr_lock.drain(..excess);
}
}
let lower = trimmed.to_lowercase();
if lower.contains("error") || lower.contains("critical") {
log::error!("aria2c stderr: {}", trimmed);
}
}
}); });
} }
match rpc_call(port, &secret, "aria2.getVersion", serde_json::json!([])).await {
Ok(ver) => return Ok(ver), let guard = app_handle_bg.state::<Aria2DaemonGuard>();
Err(e) => { *guard.child.lock().unwrap() = Some(child);
last_err = e; *guard.config_path.lock().unwrap() = Some(config_path);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let mut last_err = String::new();
let start = std::time::Instant::now();
let mut ready = false;
while start.elapsed() < std::time::Duration::from_secs(5) {
match rpc_call(attempt_port, &aria2_secret_clone, "aria2.getVersion", serde_json::json!([])).await {
Ok(ver) => {
let v = ver.get("version").and_then(|v| v.as_str()).unwrap_or("unknown");
log::info!("aria2 daemon ready (version {}) on port {}", v, attempt_port);
ready = true;
break;
}
Err(e) => {
last_err = e;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
} }
} }
} if !ready {
}); let err = if last_err.is_empty() { "aria2 daemon did not become ready within 5 seconds".to_string() } else { format!("aria2 did not become ready: {last_err}") };
log::error!("{}", err);
match ready { *guard.startup_error.lock().unwrap() = Some(err);
Ok(ver) => { }
let v = ver.get("version").and_then(|v| v.as_str()).unwrap_or("unknown"); break;
log::info!("aria2 daemon ready (version {}) on port {}", v, port);
} }
Err(e) => { Err(e) => {
log::error!("aria2 daemon readiness check failed: {}", e); log::error!("Failed to spawn aria2c: {}", e);
let guard = app.state::<Aria2DaemonGuard>(); let guard = app_handle_bg.state::<Aria2DaemonGuard>();
*guard.startup_error.lock().unwrap() = Some(e); *guard.startup_error.lock().unwrap() = Some(format!("Failed to spawn aria2c: {e}"));
break;
} }
} }
} }
Err(e) => { if !success {
log::error!("Failed to spawn aria2c: {}", e); let guard = app_handle_bg.state::<Aria2DaemonGuard>();
let guard = app.state::<Aria2DaemonGuard>(); *guard.startup_error.lock().unwrap() = Some("Failed to find open port for aria2c".to_string());
*guard.startup_error.lock().unwrap() = Some(format!("Failed to spawn aria2c: {e}"));
} }
} }
Err(e) => {
log::error!("Failed to resolve aria2c binary: {}", e);
let guard = app_handle_bg.state::<Aria2DaemonGuard>();
*guard.startup_error.lock().unwrap() = Some(format!("Failed to resolve aria2c: {e}"));
}
} }
Err(e) => {
log::error!("Failed to resolve aria2c binary: {}", e);
let guard = app.state::<Aria2DaemonGuard>();
*guard.startup_error.lock().unwrap() = Some(format!("Failed to resolve aria2c: {e}"));
}
}
let app_handle_ws = app.handle().clone(); let mut ws_retries = 0;
let ws_port = aria2_port;
tauri::async_runtime::spawn(async move {
loop { loop {
tokio::time::sleep(std::time::Duration::from_secs(1)).await; if ws_retries > 10 {
log::error!("Max WebSocket reconnection attempts reached. aria2 integration is disabled.");
let guard = app_handle_bg.state::<Aria2DaemonGuard>();
*guard.startup_error.lock().unwrap() = Some("Max WebSocket reconnection attempts reached.".to_string());
break;
}
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
let ws_url = format!("ws://127.0.0.1:{}/jsonrpc", ws_port); let ws_url = format!("ws://127.0.0.1:{}/jsonrpc", ws_port);
if let Ok((ws_stream, _)) = tokio_tungstenite::connect_async(&ws_url).await { if let Ok((ws_stream, _)) = tokio_tungstenite::connect_async(&ws_url).await {
ws_retries = 0; // reset on success
use futures_util::StreamExt; use futures_util::StreamExt;
let (_, mut read) = ws_stream.split(); let (_, mut read) = ws_stream.split();
while let Some(msg) = read.next().await { while let Some(msg) = read.next().await {
@@ -4142,7 +4180,7 @@ pub fn run() {
if let Some(params) = json.get("params").and_then(|p| p.as_array()) { if let Some(params) = json.get("params").and_then(|p| p.as_array()) {
if let Some(event) = params.first().and_then(|p| p.as_object()) { if let Some(event) = params.first().and_then(|p| p.as_object()) {
if let Some(gid) = event.get("gid").and_then(|g| g.as_str()) { if let Some(gid) = event.get("gid").and_then(|g| g.as_str()) {
let state = app_handle_ws.state::<AppState>(); let state = app_handle_bg.state::<AppState>();
let outcome = match method { let outcome = match method {
"aria2.onDownloadComplete" => Some(crate::queue::PendingOutcome::Complete), "aria2.onDownloadComplete" => Some(crate::queue::PendingOutcome::Complete),
"aria2.onDownloadError" => { "aria2.onDownloadError" => {
@@ -4164,14 +4202,14 @@ pub fn run() {
} }
} }
} }
// Connection lost, clear permits and reconnect ws_retries += 1;
app_handle_ws.state::<AppState>().queue_manager.clear_aria2_permits().await; app_handle_bg.state::<AppState>().queue_manager.clear_aria2_permits().await;
tokio::time::sleep(std::time::Duration::from_secs(4)).await; tokio::time::sleep(std::time::Duration::from_secs(4)).await;
} }
}); });
let app_handle_poll = app.handle().clone(); let app_handle_poll = app.handle().clone();
let poll_port = aria2_port; let poll_port = aria2_port.clone();
let poll_secret = aria2_secret.clone(); let poll_secret = aria2_secret.clone();
let poll_mgr = Arc::clone(&queue_manager_poll); let poll_mgr = Arc::clone(&queue_manager_poll);
tauri::async_runtime::spawn(async move { tauri::async_runtime::spawn(async move {
@@ -4179,7 +4217,7 @@ pub fn run() {
loop { loop {
interval.tick().await; interval.tick().await;
let params = serde_json::json!([["gid", "status", "totalLength", "completedLength", "downloadSpeed", "errorMessage"]]); let params = serde_json::json!([["gid", "status", "totalLength", "completedLength", "downloadSpeed", "errorMessage"]]);
if let Ok(active_list) = rpc_call(poll_port, &poll_secret, "aria2.tellActive", params).await { if let Ok(active_list) = rpc_call(poll_port.load(std::sync::atomic::Ordering::Relaxed), &poll_secret, "aria2.tellActive", params).await {
if let Some(active_arr) = active_list.as_array() { if let Some(active_arr) = active_list.as_array() {
for status_info in active_arr { for status_info in active_arr {
let gid = status_info.get("gid").and_then(|s| s.as_str()).unwrap_or(""); let gid = status_info.get("gid").and_then(|s| s.as_str()).unwrap_or("");
@@ -4221,14 +4259,19 @@ pub fn run() {
let ext_app_handle = app.handle().clone(); let ext_app_handle = app.handle().clone();
tauri::async_runtime::spawn(async move { tauri::async_runtime::spawn(async move {
if let Err(error) = extension_server::start_server( loop {
ext_app_handle, if let Err(error) = extension_server::start_server(
server_pairing_token.clone(), ext_app_handle.clone(),
server_frontend_ready.clone(), server_pairing_token.clone(),
server_extension_port.clone(), server_frontend_ready.clone(),
extension_server_shutdown_rx, server_extension_port.clone(),
).await { extension_server_shutdown_rx.clone(),
eprintln!("Browser extension server unavailable: {error}"); ).await {
log::error!("Browser extension server unavailable: {error}");
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
} else {
break;
}
} }
}); });
Ok(()) Ok(())
src-tauri/src/queue.rs
+25 -8
View File
@@ -96,6 +96,7 @@ pub struct QueueManager<R: tauri::Runtime = tauri::Wry> {
target_capacity: AtomicUsize, target_capacity: AtomicUsize,
slots_to_retire: AtomicUsize, slots_to_retire: AtomicUsize,
notify: Notify, notify: Notify,
notify_permit_released: Notify,
/// aria2 gid -> download id map (shared with the WS poller). /// aria2 gid -> download id map (shared with the WS poller).
pub aria2_gids: Arc<std::sync::RwLock<HashMap<String, String>>>, pub aria2_gids: Arc<std::sync::RwLock<HashMap<String, String>>>,
@@ -143,6 +144,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
target_capacity: AtomicUsize::new(capacity), target_capacity: AtomicUsize::new(capacity),
slots_to_retire: AtomicUsize::new(0), slots_to_retire: AtomicUsize::new(0),
notify: Notify::new(), notify: Notify::new(),
notify_permit_released: Notify::new(),
aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())),
pending_completion: Arc::new(Mutex::new(HashMap::new())), pending_completion: Arc::new(Mutex::new(HashMap::new())),
aria2_payloads: Mutex::new(HashMap::new()), aria2_payloads: Mutex::new(HashMap::new()),
@@ -191,12 +193,12 @@ impl<R: tauri::Runtime> QueueManager<R> {
} }
/// Acquire a permit from the semaphore (blocks until one is available). /// Acquire a permit from the semaphore (blocks until one is available).
pub async fn acquire_permit(&self) -> OwnedSemaphorePermit { pub async fn acquire_permit(&self) -> Option<OwnedSemaphorePermit> {
self.semaphore self.semaphore
.clone() .clone()
.acquire_owned() .acquire_owned()
.await .await
.expect("semaphore closed") .ok()
} }
/// Park an already-acquired permit under `id`. /// Park an already-acquired permit under `id`.
@@ -219,7 +221,10 @@ impl<R: tauri::Runtime> QueueManager<R> {
return false; return false;
} }
let permit = self.acquire_permit().await; let permit = match self.acquire_permit().await {
Some(p) => p,
None => return false,
};
let mut permits = self.active_permits.lock().await; let mut permits = self.active_permits.lock().await;
if permits.contains_key(id) { if permits.contains_key(id) {
drop(permits); drop(permits);
@@ -239,6 +244,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
let removed = self.active_permits.lock().await.remove(id).is_some(); let removed = self.active_permits.lock().await.remove(id).is_some();
self.active_kinds.lock().await.remove(id); self.active_kinds.lock().await.remove(id);
if removed { if removed {
self.notify_permit_released.notify_waiters();
self.notify.notify_one(); self.notify.notify_one();
} }
} }
@@ -323,12 +329,16 @@ impl<R: tauri::Runtime> QueueManager<R> {
continue; continue;
} }
// (2) Acquire a slot. // (2) Acquire a slot.
let permit = self let permit_opt = self
.semaphore .semaphore
.clone() .clone()
.acquire_owned() .acquire_owned()
.await .await
.expect("semaphore closed"); .ok();
let permit = match permit_opt {
Some(p) => p,
None => break, // Semaphore closed, exit dispatcher
};
// (3) CAS retirement — never underflows to usize::MAX. // (3) CAS retirement — never underflows to usize::MAX.
let mut retired = false; let mut retired = false;
let mut debt = self.slots_to_retire.load(Ordering::Relaxed); let mut debt = self.slots_to_retire.load(Ordering::Relaxed);
@@ -561,7 +571,14 @@ impl<R: tauri::Runtime> QueueManager<R> {
if !self.active_permits.lock().await.contains_key(id) { if !self.active_permits.lock().await.contains_key(id) {
return; return;
} }
tokio::time::sleep(Duration::from_millis(100)).await; let notified = self.notify_permit_released.notified();
if !self.active_permits.lock().await.contains_key(id) {
return;
}
tokio::select! {
_ = notified => {}
_ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {}
}
} }
} }
@@ -888,7 +905,7 @@ impl SidecarSpawner for ProductionSpawner {
let uris = crate::collect_download_uris(&payload.url, payload.mirrors.as_deref()); let uris = crate::collect_download_uris(&payload.url, payload.mirrors.as_deref());
let params = serde_json::json!([uris, options]); let params = serde_json::json!([uris, options]);
match crate::rpc_call(state.aria2_port, &state.aria2_secret, "aria2.addUri", params).await { match crate::rpc_call(state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, "aria2.addUri", params).await {
Ok(result) => { Ok(result) => {
let gid = result.as_str().unwrap_or("").to_string(); let gid = result.as_str().unwrap_or("").to_string();
if gid.is_empty() { if gid.is_empty() {
@@ -931,7 +948,7 @@ impl SidecarSpawner for ProductionSpawner {
async fn remove_uri(&self, gid: &str) -> Result<(), String> { async fn remove_uri(&self, gid: &str) -> Result<(), String> {
let state = self.app_handle.state::<crate::AppState>(); let state = self.app_handle.state::<crate::AppState>();
let result = crate::rpc_call( let result = crate::rpc_call(
state.aria2_port, state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret, &state.aria2_secret,
"aria2.forceRemove", "aria2.forceRemove",
serde_json::json!([gid]), serde_json::json!([gid]),
src-tauri/src/settings.rs
+11 -2
View File
@@ -50,8 +50,17 @@ pub fn preserve_scheduler_runtime_keys(
let Some(existing) = existing else { let Some(existing) = existing else {
return Ok(incoming.to_string()); return Ok(incoming.to_string());
}; };
let existing_document = decode_document(&Value::String(existing.to_string()))?; let existing_document = match decode_document(&Value::String(existing.to_string())) {
let existing_state = settings_state(&existing_document)?; Ok(doc) => doc,
Err(e) => {
log::warn!("Failed to decode existing settings, dropping runtime keys: {}", e);
return Ok(incoming.to_string());
}
};
let existing_state = match settings_state(&existing_document) {
Ok(state) => state,
Err(_) => return Ok(incoming.to_string()),
};
let mut incoming_document = decode_document(&Value::String(incoming.to_string()))?; let mut incoming_document = decode_document(&Value::String(incoming.to_string()))?;
let incoming_state = settings_state_mut(&mut incoming_document)?; let incoming_state = settings_state_mut(&mut incoming_document)?;
for key in ["schedulerLastStartKey", "schedulerLastStopKey"] { for key in ["schedulerLastStartKey", "schedulerLastStopKey"] {
src/App.tsx
+2 -1
View File
@@ -513,6 +513,7 @@ function App() {
}, [theme]); }, [theme]);
useEffect(() => { useEffect(() => {
if (!coreReady) return;
const unlistenDownload = initDownloadListener(); const unlistenDownload = initDownloadListener();
const unlistenTerminalState = listen('download-state', (event) => { const unlistenTerminalState = listen('download-state', (event) => {
@@ -571,7 +572,7 @@ function App() {
unlistenDeepLink.then(f => f()); unlistenDeepLink.then(f => f());
unlistenDownload.then(f => { if (f) f(); }); unlistenDownload.then(f => { if (f) f(); });
}; };
}, []); }, [coreReady]);
return ( return (
<div className="app-shell flex h-screen w-screen overflow-hidden text-text-primary"> <div className="app-shell flex h-screen w-screen overflow-hidden text-text-primary">
src/components/AddDownloadsModal.tsx
+1 -1
View File
@@ -107,7 +107,7 @@ export const AddDownloadsModal = () => {
setChecksumAlgo('SHA-256'); setChecksumAlgo('SHA-256');
setChecksumValue(''); setChecksumValue('');
setHeaders([ setHeaders([
pendingAddReferer ? `Referer: ${pendingAddReferer}` : '', pendingAddReferer ? `Referer: ${pendingAddReferer.replace(/[\r\n]/g, '')}` : '',
pendingAddHeaders pendingAddHeaders
].filter(Boolean).join('\n')); ].filter(Boolean).join('\n'));
setCookies(pendingAddCookies); setCookies(pendingAddCookies);