From a5e260f16cf38aa8b3889f9d8d44df2f62af266c Mon Sep 17 00:00:00 2001 From: forkwright Date: Thu, 2 Jul 2026 09:07:57 -0500 Subject: [PATCH] fix(ergasia): download filesystem safety (reconciliation, disk guard, zip-slip) - #360: reconcile_persisted_torrents rebuilds torrent_map from live torrents and a persisted download_id<->librqbit_id side-table, so persisted downloads stay manageable after restart. - #365: get_available_space returns a Result and propagates errors instead of returning u64::MAX and silently bypassing the disk-space guard. - #367: replace the blocking df subprocess and synchronous extraction with fs2::available_space and off-executor (spawn_blocking) extraction. - #366: the reported ExtractedFile.path is derived from the sanitized actual write location. - #452: zip extraction pre-scans and atomically refuses the whole archive on any symlink entry, absolute path, or path-traversal (zip-slip) before writing anything. - #453: add the Seeding->Failed and Completed->Deleted state transitions. - #454: enforce a configurable max_decompression_ratio (default 100x) against the declared uncompressed size to stop decompression bombs. Closes #360 Closes #365 Closes #366 Closes #367 Closes #452 Closes #453 Closes #454 Gate-Passed: kanon 0.1.5 +stages:fmt,check,clippy,nextest,lint sha:49f3752052b311213d980fccb7ee6b1d313bc337 --- Cargo.lock | 1 + crates/archon/src/serve.rs | 7 +- .../archon/tests/acquisition_integration.rs | 2 +- crates/ergasia/Cargo.toml | 3 +- crates/ergasia/src/error.rs | 45 ++ crates/ergasia/src/extract/fs_walk.rs | 49 ++ crates/ergasia/src/extract/mod.rs | 3 +- crates/ergasia/src/extract/pipeline.rs | 449 +++++++++++++++--- crates/ergasia/src/extract/rar.rs | 79 ++- crates/ergasia/src/extract/seven_zip.rs | 37 +- crates/ergasia/src/extract/zip_extract.rs | 250 ++++++++-- crates/ergasia/src/lib.rs | 9 +- crates/ergasia/src/session.rs | 299 +++++++++++- crates/ergasia/src/state.rs | 4 + crates/horismos/src/subsystems.rs | 2 + crates/syntaxis/src/pipeline.rs | 6 +- docs/download/archive.md | 4 + 17 files changed, 1093 insertions(+), 156 deletions(-) create mode 100644 crates/ergasia/src/extract/fs_walk.rs diff --git a/Cargo.lock b/Cargo.lock index 2a9cc233..467f7272 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1528,6 +1528,7 @@ dependencies = [ "librqbit", "regex", "rstest", + "rustix", "serde", "serde_json", "sevenz-rust2", diff --git a/crates/archon/src/serve.rs b/crates/archon/src/serve.rs index 00bbb177..03023ad5 100644 --- a/crates/archon/src/serve.rs +++ b/crates/archon/src/serve.rs @@ -454,6 +454,7 @@ async fn subtitle_target( /// that Syntaxis expects for dispatching downloads. struct SessionEngine { session: Arc, + extraction_limits: ergasia::ExtractionLimits, } impl ergasia::DownloadEngine for SessionEngine { @@ -505,12 +506,12 @@ impl ergasia::DownloadEngine for SessionEngine { }) } - fn extract( + async fn extract( &self, download_path: &std::path::Path, output_dir: &std::path::Path, ) -> Result, ergasia::ErgasiaError> { - ergasia::extract_archives(download_path, output_dir, 3) + ergasia::extract_archives(download_path, output_dir, self.extraction_limits).await } } @@ -677,12 +678,12 @@ pub async fn run_serve(args: ServeArgs, out: &mut impl Write) -> Result<(), Host .await .context(DownloadEngineSnafu)?, ); - ergasia_session.reconcile_persisted_torrents(); info!("ergasia (download engine) initialized"); // Layer 2: Syntaxis (queue orchestration, depends on ergasia) let engine_adapter = Arc::new(SessionEngine { session: Arc::clone(&ergasia_session), + extraction_limits: ergasia::ExtractionLimits::from(&config.ergasia), }); let syntaxis_svc = Arc::new( DownloadQueue::new( diff --git a/crates/archon/tests/acquisition_integration.rs b/crates/archon/tests/acquisition_integration.rs index d9b60442..0d2f3c9d 100644 --- a/crates/archon/tests/acquisition_integration.rs +++ b/crates/archon/tests/acquisition_integration.rs @@ -89,7 +89,7 @@ impl ergasia::DownloadEngine for MockEngine { }) } - fn extract( + async fn extract( &self, _download_path: &std::path::Path, _output_dir: &std::path::Path, diff --git a/crates/ergasia/Cargo.toml b/crates/ergasia/Cargo.toml index 32f09b05..a1d77a09 100644 --- a/crates/ergasia/Cargo.toml +++ b/crates/ergasia/Cargo.toml @@ -14,16 +14,17 @@ tokio.workspace = true tokio-util.workspace = true dashmap.workspace = true serde.workspace = true +serde_json.workspace = true bytes.workspace = true librqbit.workspace = true zip.workspace = true unrar = "0.5" sevenz-rust2.workspace = true regex.workspace = true +rustix = { version = "1", features = ["fs"] } [dev-dependencies] rstest.workspace = true -serde_json.workspace = true tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } tempfile = "3" diff --git a/crates/ergasia/src/error.rs b/crates/ergasia/src/error.rs index 177d6f3c..65e99b78 100644 --- a/crates/ergasia/src/error.rs +++ b/crates/ergasia/src/error.rs @@ -77,6 +77,51 @@ pub enum ErgasiaError { location: snafu::Location, }, + #[snafu(display("failed to query available disk space for {}", path.display()))] + DiskSpaceQuery { + path: PathBuf, + error: String, + #[snafu(implicit)] + location: snafu::Location, + }, + + #[snafu(display("unsafe archive entry {entry:?} in {}: {reason}", archive.display()))] + UnsafeArchiveEntry { + archive: PathBuf, + entry: String, + reason: String, + #[snafu(implicit)] + location: snafu::Location, + }, + + #[snafu(display( + "archive {} declares {declared_uncompressed} bytes uncompressed from {compressed} bytes compressed, exceeding the {max_ratio}x decompression ratio limit", + archive.display() + ))] + DecompressionRatioExceeded { + archive: PathBuf, + compressed: u64, + declared_uncompressed: u64, + max_ratio: f64, + #[snafu(implicit)] + location: snafu::Location, + }, + + #[snafu(display("archive extraction task failed to complete"))] + ExtractionJoin { + source: tokio::task::JoinError, + #[snafu(implicit)] + location: snafu::Location, + }, + + #[snafu(display("failed to persist torrent map at {}", path.display()))] + TorrentMapPersistence { + path: PathBuf, + error: String, + #[snafu(implicit)] + location: snafu::Location, + }, + #[snafu(display("unsupported archive format at {}: magic bytes {magic_bytes:02X?}", path.display()))] UnsupportedFormat { path: PathBuf, diff --git a/crates/ergasia/src/extract/fs_walk.rs b/crates/ergasia/src/extract/fs_walk.rs new file mode 100644 index 00000000..9d3235de --- /dev/null +++ b/crates/ergasia/src/extract/fs_walk.rs @@ -0,0 +1,49 @@ +// Filesystem-walk inventory: derives ExtractedFile lists from what is actually on disk. + +use std::collections::HashSet; +use std::path::{Path, PathBuf}; + +use crate::extract::pipeline::ExtractedFile; + +pub(crate) fn snapshot_paths(dir: &Path) -> HashSet { + let mut paths = HashSet::new(); + visit(dir, &mut |path, _| { + paths.insert(path.to_path_buf()); + }); + paths +} + +pub(crate) fn collect_files_excluding( + dir: &Path, + exclude: &HashSet, + files: &mut Vec, +) { + visit(dir, &mut |path, size| { + if !exclude.contains(path) { + files.push(ExtractedFile { + path: path.to_path_buf(), + size_bytes: size, + }); + } + }); +} + +fn visit(dir: &Path, on_file: &mut impl FnMut(&Path, u64)) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + + for entry in entries.flatten() { + let path = entry.path(); + // SAFETY: DirEntry::file_type does not follow symlinks, so a symlinked + // directory is inventoried as a file instead of being traversed. + let Ok(file_type) = entry.file_type() else { + continue; + }; + if file_type.is_dir() { + visit(&path, on_file); + } else if let Ok(meta) = path.symlink_metadata() { + on_file(&path, meta.len()); + } + } +} diff --git a/crates/ergasia/src/extract/mod.rs b/crates/ergasia/src/extract/mod.rs index 05a61e21..188a6d6d 100644 --- a/crates/ergasia/src/extract/mod.rs +++ b/crates/ergasia/src/extract/mod.rs @@ -1,9 +1,10 @@ mod detect; +mod fs_walk; mod pipeline; mod rar; mod seven_zip; mod zip_extract; pub use detect::{ArchiveFormat, detect_archive_format}; -pub use pipeline::{ExtractedFile, ExtractionResult, extract_archives}; +pub use pipeline::{ExtractedFile, ExtractionLimits, ExtractionResult, extract_archives}; pub use rar::find_rar_first_volume; diff --git a/crates/ergasia/src/extract/pipeline.rs b/crates/ergasia/src/extract/pipeline.rs index 01f5217f..c1d7a1e4 100644 --- a/crates/ergasia/src/extract/pipeline.rs +++ b/crates/ergasia/src/extract/pipeline.rs @@ -1,13 +1,18 @@ use std::path::{Path, PathBuf}; +use horismos::ErgasiaConfig; use serde::{Deserialize, Serialize}; -use snafu::ensure; +use snafu::{ResultExt, ensure}; -use crate::error::{ErgasiaError, InsufficientDiskSpaceSnafu, NestingDepthExceededSnafu}; +use crate::error::{ + DecompressionRatioExceededSnafu, DiskSpaceQuerySnafu, ErgasiaError, ExtractionJoinSnafu, + InsufficientDiskSpaceSnafu, NestingDepthExceededSnafu, +}; use crate::extract::detect::{ArchiveFormat, detect_archive_format, detect_by_magic_bytes}; use crate::extract::rar::{extract_rar, find_rar_first_volume}; use crate::extract::seven_zip::extract_7z; use crate::extract::zip_extract::extract_zip; +use crate::extract::{fs_walk, rar, seven_zip, zip_extract}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ExtractionResult { @@ -23,10 +28,43 @@ pub struct ExtractedFile { pub size_bytes: u64, } -pub fn extract_archives( +#[derive(Debug, Clone, Copy)] +pub struct ExtractionLimits { + pub max_depth: u8, + pub max_decompression_ratio: f64, +} + +impl From<&ErgasiaConfig> for ExtractionLimits { + fn from(config: &ErgasiaConfig) -> Self { + Self { + max_depth: config.max_extraction_depth, + max_decompression_ratio: config.max_decompression_ratio, + } + } +} + +// WHY: the whole pipeline (magic-byte reads, archive decompression, directory +// walks) is blocking I/O; one spawn_blocking boundary here keeps every format +// backend off the async executor instead of sprinkling wrappers per call site. +pub async fn extract_archives( download_path: &Path, output_dir: &Path, - max_depth: u8, + limits: ExtractionLimits, +) -> Result, ErgasiaError> { + let download_path = download_path.to_path_buf(); + let output_dir = output_dir.to_path_buf(); + + tokio::task::spawn_blocking(move || { + extract_archives_blocking(&download_path, &output_dir, limits) + }) + .await + .context(ExtractionJoinSnafu)? +} + +fn extract_archives_blocking( + download_path: &Path, + output_dir: &Path, + limits: ExtractionLimits, ) -> Result, ErgasiaError> { let archives = find_archives_in_dir(download_path); if archives.is_empty() { @@ -41,7 +79,7 @@ pub fn extract_archives( .build() })?; - check_disk_space(download_path, output_dir)?; + preflight_archives(&archives, output_dir, limits.max_decompression_ratio)?; let Some((_, first_format)) = archives.first() else { return Ok(None); @@ -54,7 +92,7 @@ pub fn extract_archives( all_files.extend(files); } - let nested_levels = handle_nested(output_dir, 1, max_depth, &mut all_files)?; + let nested_levels = handle_nested(output_dir, 1, limits, &mut all_files)?; Ok(Some(ExtractionResult { extracted_path: output_dir.to_path_buf(), @@ -96,22 +134,31 @@ fn find_archives_in_dir(dir: &Path) -> Vec<(PathBuf, ArchiveFormat)> { archives } +// WHY: inventory is diffed against a pre-extraction snapshot so reported paths +// always match the sanitized on-disk write locations, and files from earlier +// archives sharing the output dir are never double-counted. fn extract_single( archive_path: &Path, output_dir: &Path, format: ArchiveFormat, ) -> Result, ErgasiaError> { + let before = fs_walk::snapshot_paths(output_dir); + match format { - ArchiveFormat::Rar => extract_rar(archive_path, output_dir), - ArchiveFormat::Zip => extract_zip(archive_path, output_dir), - ArchiveFormat::SevenZip => extract_7z(archive_path, output_dir), + ArchiveFormat::Rar => extract_rar(archive_path, output_dir)?, + ArchiveFormat::Zip => extract_zip(archive_path, output_dir)?, + ArchiveFormat::SevenZip => extract_7z(archive_path, output_dir)?, } + + let mut files = Vec::new(); + fs_walk::collect_files_excluding(output_dir, &before, &mut files); + Ok(files) } fn handle_nested( dir: &Path, current_depth: u8, - max_depth: u8, + limits: ExtractionLimits, all_files: &mut Vec, ) -> Result { let nested_archives = find_nested_archives(dir); @@ -120,10 +167,10 @@ fn handle_nested( } ensure!( - current_depth < max_depth, + current_depth < limits.max_depth, NestingDepthExceededSnafu { depth: current_depth, - max: max_depth, + max: limits.max_depth, } ); @@ -136,12 +183,18 @@ fn handle_nested( .build() })?; + preflight_archives( + &nested_archives, + &nested_output, + limits.max_decompression_ratio, + )?; + for (archive_path, format) in &nested_archives { let files = extract_single(archive_path, &nested_output, *format)?; all_files.extend(files); } - handle_nested(&nested_output, current_depth + 1, max_depth, all_files) + handle_nested(&nested_output, current_depth + 1, limits, all_files) } fn find_nested_archives(dir: &Path) -> Vec<(PathBuf, ArchiveFormat)> { @@ -164,11 +217,22 @@ fn find_nested_archives(dir: &Path) -> Vec<(PathBuf, ArchiveFormat)> { archives } -fn check_disk_space(download_path: &Path, output_dir: &Path) -> Result<(), ErgasiaError> { - let archive_size = calculate_archive_size(download_path); - let needed = (archive_size as f64 * 1.1) as u64; +// WHY: both guards run before any extraction write. The ratio guard is a policy +// check distinct from disk-space sufficiency: a decompression bomb can pass the +// space check on a large disk and still be hostile. +fn preflight_archives( + archives: &[(PathBuf, ArchiveFormat)], + output_dir: &Path, + max_ratio: f64, +) -> Result<(), ErgasiaError> { + let mut total_declared: u64 = 0; + for (archive_path, format) in archives { + let declared = enforce_decompression_ratio(archive_path, *format, max_ratio)?; + total_declared = total_declared.saturating_add(declared); + } - let available = get_available_space(output_dir); + let needed = needed_with_headroom(total_declared); + let available = get_available_space(output_dir)?; ensure!( available >= needed, @@ -178,46 +242,69 @@ fn check_disk_space(download_path: &Path, output_dir: &Path) -> Result<(), Ergas Ok(()) } -fn calculate_archive_size(dir: &Path) -> u64 { - let Ok(entries) = std::fs::read_dir(dir) else { - return 0; - }; +fn enforce_decompression_ratio( + archive_path: &Path, + format: ArchiveFormat, + max_ratio: f64, +) -> Result { + let declared = declared_uncompressed_size(archive_path, format)?; + let compressed = compressed_size_on_disk(archive_path, format)?; - entries - .flatten() - .filter_map(|e| { - let path = e.path(); - if path.is_file() && detect_archive_format(&path).is_some() { - path.metadata().ok().map(|m| m.len()) - } else { - None - } - }) - .sum() + ensure!( + declared as f64 <= compressed as f64 * max_ratio, + DecompressionRatioExceededSnafu { + archive: archive_path.to_path_buf(), + compressed, + declared_uncompressed: declared, + max_ratio, + } + ); + + Ok(declared) } -fn get_available_space(path: &Path) -> u64 { - let output = match std::process::Command::new("df") - .arg("--output=avail") - .arg("-B1") - .arg(path) - .output() - { - Ok(output) => output, - Err(e) => { - tracing::warn!(error = %e, path = %path.display(), "failed to query available disk space"); - return u64::MAX; +fn declared_uncompressed_size( + archive_path: &Path, + format: ArchiveFormat, +) -> Result { + match format { + ArchiveFormat::Rar => rar::declared_uncompressed_size(archive_path), + ArchiveFormat::Zip => zip_extract::declared_uncompressed_size(archive_path), + ArchiveFormat::SevenZip => seven_zip::declared_uncompressed_size(archive_path), + } +} + +fn compressed_size_on_disk( + archive_path: &Path, + format: ArchiveFormat, +) -> Result { + if format == ArchiveFormat::Rar { + return Ok(rar::volume_set_size(archive_path)); + } + + archive_path.metadata().map(|m| m.len()).map_err(|e| { + crate::error::OpenArchiveSnafu { + path: archive_path.to_path_buf(), + error: e.to_string(), } - }; + .build() + }) +} + +fn needed_with_headroom(total_declared: u64) -> u64 { + total_declared.saturating_add(total_declared / 10) +} + +fn get_available_space(path: &Path) -> Result { + let stat = rustix::fs::statvfs(path).map_err(|e| { + DiskSpaceQuerySnafu { + path: path.to_path_buf(), + error: e.to_string(), + } + .build() + })?; - String::from_utf8(output.stdout) - .ok() - .and_then(|s| { - s.lines() - .nth(1) - .and_then(|line| line.trim().parse::().ok()) - }) - .unwrap_or(u64::MAX) + Ok(stat.f_bavail.saturating_mul(stat.f_frsize)) } #[cfg(test)] @@ -227,6 +314,11 @@ mod tests { use super::*; use crate::error::InsufficientDiskSpaceSnafu; + const TEST_LIMITS: ExtractionLimits = ExtractionLimits { + max_depth: 3, + max_decompression_ratio: 100.0, + }; + fn create_test_zip(dir: &Path, name: &str, contents: &[(&str, &[u8])]) -> PathBuf { let zip_path = dir.join(name); let file = std::fs::File::create(&zip_path).unwrap(); @@ -242,8 +334,20 @@ mod tests { zip_path } - #[test] - fn extract_zip_archive_via_pipeline() { + fn create_deflated_bomb_zip(dir: &Path, name: &str, uncompressed_len: usize) -> PathBuf { + let zip_path = dir.join(name); + let file = std::fs::File::create(&zip_path).unwrap(); + let mut writer = zip::ZipWriter::new(file); + let options = zip::write::SimpleFileOptions::default() + .compression_method(zip::CompressionMethod::Deflated); + writer.start_file("bomb.bin", options).unwrap(); + writer.write_all(&vec![0u8; uncompressed_len]).unwrap(); + writer.finish().unwrap(); + zip_path + } + + #[tokio::test] + async fn extract_zip_archive_via_pipeline() { let dir = tempfile::tempdir().unwrap(); let download_dir = dir.path().join("download"); let output_dir = dir.path().join("output"); @@ -255,7 +359,8 @@ mod tests { &[("hello.txt", b"Hello!"), ("world.txt", b"World!")], ); - let result = extract_archives(&download_dir, &output_dir, 3) + let result = extract_archives(&download_dir, &output_dir, TEST_LIMITS) + .await .unwrap() .unwrap(); assert_eq!(result.archive_format, ArchiveFormat::Zip); @@ -264,20 +369,22 @@ mod tests { assert!(output_dir.join("world.txt").exists()); } - #[test] - fn no_archives_returns_none() { + #[tokio::test] + async fn no_archives_returns_none() { let dir = tempfile::tempdir().unwrap(); let download_dir = dir.path().join("download"); let output_dir = dir.path().join("output"); std::fs::create_dir_all(&download_dir).unwrap(); std::fs::write(download_dir.join("readme.txt"), b"just a text file").unwrap(); - let result = extract_archives(&download_dir, &output_dir, 3).unwrap(); + let result = extract_archives(&download_dir, &output_dir, TEST_LIMITS) + .await + .unwrap(); assert!(result.is_none()); } - #[test] - fn nested_zip_extraction() { + #[tokio::test] + async fn nested_zip_extraction() { let dir = tempfile::tempdir().unwrap(); let download_dir = dir.path().join("download"); let output_dir = dir.path().join("output"); @@ -303,7 +410,8 @@ mod tests { writer.finish().unwrap(); } - let result = extract_archives(&download_dir, &output_dir, 3) + let result = extract_archives(&download_dir, &output_dir, TEST_LIMITS) + .await .unwrap() .unwrap(); assert!(result.nested_levels >= 1); @@ -315,8 +423,8 @@ mod tests { ); } - #[test] - fn nesting_depth_exceeded() { + #[tokio::test] + async fn nesting_depth_exceeded() { let dir = tempfile::tempdir().unwrap(); let download_dir = dir.path().join("download"); let output_dir = dir.path().join("output"); @@ -354,7 +462,11 @@ mod tests { writer.finish().unwrap(); } - let result = extract_archives(&download_dir, &output_dir, 2); + let limits = ExtractionLimits { + max_depth: 2, + max_decompression_ratio: 100.0, + }; + let result = extract_archives(&download_dir, &output_dir, limits).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!( @@ -363,6 +475,215 @@ mod tests { ); } + #[tokio::test] + async fn reject_archive_exceeding_decompression_ratio() { + let dir = tempfile::tempdir().unwrap(); + let download_dir = dir.path().join("download"); + let output_dir = dir.path().join("output"); + std::fs::create_dir_all(&download_dir).unwrap(); + + // 1 MiB of zeros deflates to a few KiB: the declared/compressed ratio + // far exceeds the 10x test limit. + create_deflated_bomb_zip(&download_dir, "bomb.zip", 1024 * 1024); + + let limits = ExtractionLimits { + max_depth: 3, + max_decompression_ratio: 10.0, + }; + let err = extract_archives(&download_dir, &output_dir, limits) + .await + .unwrap_err(); + assert!( + matches!(err, ErgasiaError::DecompressionRatioExceeded { .. }), + "expected DecompressionRatioExceeded, got: {err}" + ); + let leftovers: Vec<_> = std::fs::read_dir(&output_dir).unwrap().flatten().collect(); + assert!( + leftovers.is_empty(), + "expected no extraction output, found {leftovers:?}" + ); + } + + #[tokio::test] + async fn reject_nested_bomb() { + let dir = tempfile::tempdir().unwrap(); + let download_dir = dir.path().join("download"); + let output_dir = dir.path().join("output"); + std::fs::create_dir_all(&download_dir).unwrap(); + + // The outer zip stores the bomb uncompressed (ratio ~1x), so only the + // nested pre-flight can catch the inner high-ratio archive. + let staging = dir.path().join("staging"); + std::fs::create_dir_all(&staging).unwrap(); + let bomb = create_deflated_bomb_zip(&staging, "inner.zip", 1024 * 1024); + let bomb_bytes = std::fs::read(&bomb).unwrap(); + + let outer_path = download_dir.join("outer.zip"); + { + let file = std::fs::File::create(&outer_path).unwrap(); + let mut writer = zip::ZipWriter::new(file); + let options = zip::write::SimpleFileOptions::default() + .compression_method(zip::CompressionMethod::Stored); + writer.start_file("inner.zip", options).unwrap(); + writer.write_all(&bomb_bytes).unwrap(); + writer.finish().unwrap(); + } + + let limits = ExtractionLimits { + max_depth: 3, + max_decompression_ratio: 10.0, + }; + let err = extract_archives(&download_dir, &output_dir, limits) + .await + .unwrap_err(); + assert!( + matches!(err, ErgasiaError::DecompressionRatioExceeded { .. }), + "expected DecompressionRatioExceeded for nested bomb, got: {err}" + ); + assert!( + !output_dir.join(".nested").join("bomb.bin").exists(), + "nested bomb payload must not be extracted" + ); + } + + #[tokio::test] + async fn multi_archive_inventory_not_double_counted() { + let dir = tempfile::tempdir().unwrap(); + let download_dir = dir.path().join("download"); + let output_dir = dir.path().join("output"); + std::fs::create_dir_all(&download_dir).unwrap(); + + create_test_zip(&download_dir, "a.zip", &[("first.txt", b"one")]); + create_test_zip(&download_dir, "b.zip", &[("second.txt", b"two")]); + + let result = extract_archives(&download_dir, &output_dir, TEST_LIMITS) + .await + .unwrap() + .unwrap(); + assert_eq!( + result.files.len(), + 2, + "each extracted file must be inventoried exactly once: {:?}", + result.files + ); + } + + #[tokio::test] + async fn inventory_paths_match_filesystem() { + let dir = tempfile::tempdir().unwrap(); + let download_dir = dir.path().join("download"); + let output_dir = dir.path().join("output"); + std::fs::create_dir_all(&download_dir).unwrap(); + + // Entry names with redundant components that extraction normalizes. + create_test_zip( + &download_dir, + "messy.zip", + &[("a/./b.txt", b"dot component"), ("c//d.txt", b"empty seg")], + ); + + let result = extract_archives(&download_dir, &output_dir, TEST_LIMITS) + .await + .unwrap() + .unwrap(); + + assert_eq!(result.files.len(), 2); + for file in &result.files { + let meta = file.path.symlink_metadata().unwrap_or_else(|_| { + panic!( + "inventory path does not exist on disk: {}", + file.path.display() + ) + }); + assert_eq!( + meta.len(), + file.size_bytes, + "size mismatch for {}", + file.path.display() + ); + } + } + + // WHY: with a single worker thread, a synchronous extraction inside the + // async fn would never yield between the counter snapshot and completion, + // freezing the ticker at zero progress; the spawn_blocking boundary yields + // to the executor, so the ticker must advance. + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + async fn extract_does_not_block_executor() { + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + + let dir = tempfile::tempdir().unwrap(); + let download_dir = dir.path().join("download"); + let output_dir = dir.path().join("output"); + std::fs::create_dir_all(&download_dir).unwrap(); + let entries: Vec<(String, Vec)> = (0..200) + .map(|i| (format!("file_{i}.txt"), vec![b'x'; 4096])) + .collect(); + let entry_refs: Vec<(&str, &[u8])> = entries + .iter() + .map(|(name, data)| (name.as_str(), data.as_slice())) + .collect(); + create_test_zip(&download_dir, "many.zip", &entry_refs); + + let counter = Arc::new(AtomicU64::new(0)); + let stop = Arc::new(AtomicBool::new(false)); + let ticker_counter = Arc::clone(&counter); + let ticker_stop = Arc::clone(&stop); + let ticker = tokio::spawn(async move { + while !ticker_stop.load(Ordering::Relaxed) { + ticker_counter.fetch_add(1, Ordering::Relaxed); + tokio::task::yield_now().await; + } + }); + + let before = counter.load(Ordering::Relaxed); + let result = extract_archives(&download_dir, &output_dir, TEST_LIMITS).await; + let after = counter.load(Ordering::Relaxed); + stop.store(true, Ordering::Relaxed); + ticker.await.unwrap(); + + assert!(result.unwrap().is_some()); + assert!( + after > before, + "executor made no progress while extraction ran: before={before} after={after}" + ); + } + + #[test] + fn disk_space_query_failure_propagates() { + let err = + get_available_space(Path::new("/nonexistent-harmonia-test-path/child")).unwrap_err(); + assert!( + matches!(err, ErgasiaError::DiskSpaceQuery { .. }), + "expected DiskSpaceQuery, got: {err}" + ); + } + + #[test] + fn preflight_propagates_disk_space_query_failure() { + let dir = tempfile::tempdir().unwrap(); + let zip_path = create_test_zip(dir.path(), "test.zip", &[("a.txt", b"data")]); + + let err = preflight_archives( + &[(zip_path, ArchiveFormat::Zip)], + Path::new("/nonexistent-harmonia-test-path/child"), + 100.0, + ) + .unwrap_err(); + assert!( + matches!(err, ErgasiaError::DiskSpaceQuery { .. }), + "expected DiskSpaceQuery, got: {err}" + ); + } + + #[test] + fn needed_with_headroom_saturates() { + assert_eq!(needed_with_headroom(0), 0); + assert_eq!(needed_with_headroom(100), 110); + assert_eq!(needed_with_headroom(u64::MAX), u64::MAX); + } + #[test] fn insufficient_disk_space_detected() { let err: ErgasiaError = InsufficientDiskSpaceSnafu { diff --git a/crates/ergasia/src/extract/rar.rs b/crates/ergasia/src/extract/rar.rs index 65a65512..033b2cbf 100644 --- a/crates/ergasia/src/extract/rar.rs +++ b/crates/ergasia/src/extract/rar.rs @@ -4,7 +4,6 @@ use std::sync::LazyLock; use regex::Regex; use crate::error::ErgasiaError; -use crate::extract::pipeline::ExtractedFile; static MODERN_RAR_RE: LazyLock = LazyLock::new(|| { Regex::new(r"\.part(\d+)\.rar$") @@ -63,10 +62,7 @@ pub fn find_rar_first_volume(dir: &Path) -> Option { Some(first_rar.clone()) } -pub fn extract_rar( - archive_path: &Path, - output_dir: &Path, -) -> Result, ErgasiaError> { +pub fn extract_rar(archive_path: &Path, output_dir: &Path) -> Result<(), ErgasiaError> { let archive = unrar::Archive::new(archive_path) .open_for_processing() .map_err(|e| { @@ -77,7 +73,6 @@ pub fn extract_rar( .build() })?; - let mut files = Vec::new(); let mut cursor = archive; loop { @@ -93,11 +88,6 @@ pub fn extract_rar( break; }; - let entry = header.entry(); - let is_dir = entry.is_directory(); - let filename = entry.filename.clone(); - let size = entry.unpacked_size; - let next = header.extract_with_base(output_dir).map_err(|e| { crate::error::ExtractFileSnafu { path: archive_path.to_path_buf(), @@ -106,17 +96,68 @@ pub fn extract_rar( .build() })?; - if !is_dir { - files.push(ExtractedFile { - path: output_dir.join(&filename), - size_bytes: size, - }); - } - cursor = next; } - Ok(files) + Ok(()) +} + +pub(crate) fn declared_uncompressed_size(archive_path: &Path) -> Result { + let archive = unrar::Archive::new(archive_path) + .open_for_listing() + .map_err(|e| { + crate::error::OpenArchiveSnafu { + path: archive_path.to_path_buf(), + error: e.to_string(), + } + .build() + })?; + + let mut total: u64 = 0; + for header in archive { + let entry = header.map_err(|e| { + crate::error::OpenArchiveSnafu { + path: archive_path.to_path_buf(), + error: e.to_string(), + } + .build() + })?; + total = total.saturating_add(entry.unpacked_size); + } + Ok(total) +} + +// WHY: a multi-volume RAR declares the unpacked size of the whole set, so the +// ratio guard must compare against the on-disk size of every volume, not just +// the first one. +pub(crate) fn volume_set_size(first_volume: &Path) -> u64 { + let Some(dir) = first_volume.parent() else { + return first_volume.metadata().map(|m| m.len()).unwrap_or(0); + }; + + let Ok(entries) = std::fs::read_dir(dir) else { + return first_volume.metadata().map(|m| m.len()).unwrap_or(0); + }; + + entries + .flatten() + .map(|e| e.path()) + .filter(|p| p.is_file() && is_rar_volume(p)) + .filter_map(|p| p.metadata().ok().map(|m| m.len())) + .fold(0u64, |total, len| total.saturating_add(len)) +} + +fn is_rar_volume(path: &Path) -> bool { + path.extension() + .and_then(|e| e.to_str()) + .map(|ext| { + let ext = ext.to_ascii_lowercase(); + ext == "rar" + || (ext.len() == 3 + && ext.starts_with('r') + && ext.chars().skip(1).all(|c| c.is_ascii_digit())) + }) + .unwrap_or(false) } #[cfg(test)] diff --git a/crates/ergasia/src/extract/seven_zip.rs b/crates/ergasia/src/extract/seven_zip.rs index 270378ca..dd0b0e35 100644 --- a/crates/ergasia/src/extract/seven_zip.rs +++ b/crates/ergasia/src/extract/seven_zip.rs @@ -1,12 +1,8 @@ use std::path::Path; use crate::error::ErgasiaError; -use crate::extract::pipeline::ExtractedFile; -pub fn extract_7z( - archive_path: &Path, - output_dir: &Path, -) -> Result, ErgasiaError> { +pub fn extract_7z(archive_path: &Path, output_dir: &Path) -> Result<(), ErgasiaError> { sevenz_rust2::decompress_file(archive_path, output_dir).map_err(|e| { crate::error::ExtractFileSnafu { path: archive_path.to_path_buf(), @@ -15,25 +11,20 @@ pub fn extract_7z( .build() })?; - let mut files = Vec::new(); - collect_files(output_dir, &mut files); - Ok(files) + Ok(()) } -fn collect_files(dir: &Path, files: &mut Vec) { - let Ok(entries) = std::fs::read_dir(dir) else { - return; - }; - - for entry in entries.flatten() { - let path = entry.path(); - if path.is_dir() { - collect_files(&path, files); - } else if let Ok(meta) = path.metadata() { - files.push(ExtractedFile { - path, - size_bytes: meta.len(), - }); +pub(crate) fn declared_uncompressed_size(archive_path: &Path) -> Result { + let archive = sevenz_rust2::Archive::open(archive_path).map_err(|e| { + crate::error::OpenArchiveSnafu { + path: archive_path.to_path_buf(), + error: e.to_string(), } - } + .build() + })?; + + Ok(archive + .files + .iter() + .fold(0u64, |total, entry| total.saturating_add(entry.size))) } diff --git a/crates/ergasia/src/extract/zip_extract.rs b/crates/ergasia/src/extract/zip_extract.rs index d3b5ab2e..b1417afa 100644 --- a/crates/ergasia/src/extract/zip_extract.rs +++ b/crates/ergasia/src/extract/zip_extract.rs @@ -1,22 +1,55 @@ use std::fs::File; use std::path::Path; -use crate::error::ErgasiaError; -use crate::extract::pipeline::ExtractedFile; +use snafu::ensure; +use zip::ZipArchive; -pub fn extract_zip( - archive_path: &Path, - output_dir: &Path, -) -> Result, ErgasiaError> { - let file = File::open(archive_path).map_err(|e| { - crate::error::OpenArchiveSnafu { +use crate::error::{ErgasiaError, UnsafeArchiveEntrySnafu}; + +const S_IFMT: u32 = 0o170000; +const S_IFLNK: u32 = 0o120000; + +pub fn extract_zip(archive_path: &Path, output_dir: &Path) -> Result<(), ErgasiaError> { + let mut archive = open_zip(archive_path)?; + + ensure_safe_entries(&mut archive, archive_path)?; + + archive.extract(output_dir).map_err(|e| { + crate::error::ExtractFileSnafu { path: archive_path.to_path_buf(), error: e.to_string(), } .build() })?; - let mut archive = zip::ZipArchive::new(file).map_err(|e| { + Ok(()) +} + +pub(crate) fn declared_uncompressed_size(archive_path: &Path) -> Result { + let mut archive = open_zip(archive_path)?; + + let mut total: u64 = 0; + for i in 0..archive.len() { + let entry = archive.by_index_raw(i).map_err(|e| { + crate::error::OpenArchiveSnafu { + path: archive_path.to_path_buf(), + error: e.to_string(), + } + .build() + })?; + total = total.saturating_add(entry.size()); + } + Ok(total) +} + +// NOTE: entry names use zip's cross-platform path semantics, so both Unix and +// Windows-style roots (including drive-letter prefixes) count as absolute. +fn is_absolute_entry_name(name: &str) -> bool { + name.starts_with('/') || name.starts_with('\\') || name.get(1..2) == Some(":") +} + +fn open_zip(archive_path: &Path) -> Result, ErgasiaError> { + let file = File::open(archive_path).map_err(|e| { crate::error::OpenArchiveSnafu { path: archive_path.to_path_buf(), error: e.to_string(), @@ -24,35 +57,65 @@ pub fn extract_zip( .build() })?; - archive.extract(output_dir).map_err(|e| { - crate::error::ExtractFileSnafu { + ZipArchive::new(file).map_err(|e| { + crate::error::OpenArchiveSnafu { path: archive_path.to_path_buf(), error: e.to_string(), } .build() - })?; + }) +} - let mut files = Vec::new(); +// SAFETY: rejects the whole archive before any write occurs, so a hostile +// archive is refused atomically instead of partially extracted. +fn ensure_safe_entries( + archive: &mut ZipArchive, + archive_path: &Path, +) -> Result<(), ErgasiaError> { for i in 0..archive.len() { - let entry = archive.by_index(i).map_err(|e| { - crate::error::ExtractFileSnafu { + let entry = archive.by_index_raw(i).map_err(|e| { + crate::error::OpenArchiveSnafu { path: archive_path.to_path_buf(), error: e.to_string(), } .build() })?; - if !entry.is_dir() { - let name = entry.name().to_string(); - let size = entry.size(); - files.push(ExtractedFile { - path: output_dir.join(name), - size_bytes: size, - }); - } - } + let is_symlink = entry + .unix_mode() + .map(|mode| mode & S_IFMT == S_IFLNK) + .unwrap_or(false); + ensure!( + !is_symlink, + UnsafeArchiveEntrySnafu { + archive: archive_path.to_path_buf(), + entry: entry.name().to_string(), + reason: "symlink entry".to_string(), + } + ); + + // WHY: zip's enclosed_name() neutralizes a leading root by stripping it + // rather than rejecting the entry, so absolute names need an explicit + // check to refuse the archive outright. + ensure!( + !is_absolute_entry_name(entry.name()), + UnsafeArchiveEntrySnafu { + archive: archive_path.to_path_buf(), + entry: entry.name().to_string(), + reason: "absolute entry name".to_string(), + } + ); - Ok(files) + ensure!( + entry.enclosed_name().is_some(), + UnsafeArchiveEntrySnafu { + archive: archive_path.to_path_buf(), + entry: entry.name().to_string(), + reason: "path escapes the extraction root".to_string(), + } + ); + } + Ok(()) } #[cfg(test)] @@ -61,6 +124,34 @@ mod tests { use super::*; + fn zip_options() -> zip::write::SimpleFileOptions { + zip::write::SimpleFileOptions::default().compression_method(zip::CompressionMethod::Stored) + } + + fn replace_bytes(haystack: &mut [u8], needle: &[u8], replacement: &[u8]) { + assert_eq!(needle.len(), replacement.len()); + let mut replaced = false; + let mut i = 0; + while i + needle.len() <= haystack.len() { + if &haystack[i..i + needle.len()] == needle { + haystack[i..i + needle.len()].copy_from_slice(replacement); + replaced = true; + i += needle.len(); + } else { + i += 1; + } + } + assert!(replaced, "needle not found in archive bytes"); + } + + fn assert_no_entries(dir: &Path) { + let entries: Vec<_> = std::fs::read_dir(dir).unwrap().flatten().collect(); + assert!( + entries.is_empty(), + "expected empty output dir, found {entries:?}" + ); + } + #[test] fn extract_zip_archive() { let dir = tempfile::tempdir().unwrap(); @@ -71,17 +162,16 @@ mod tests { { let file = File::create(&zip_path).unwrap(); let mut writer = zip::ZipWriter::new(file); - let options = zip::write::SimpleFileOptions::default() - .compression_method(zip::CompressionMethod::Stored); - writer.start_file("hello.txt", options).unwrap(); + writer.start_file("hello.txt", zip_options()).unwrap(); writer.write_all(b"Hello, World!").unwrap(); - writer.start_file("subdir/nested.txt", options).unwrap(); + writer + .start_file("subdir/nested.txt", zip_options()) + .unwrap(); writer.write_all(b"Nested content").unwrap(); writer.finish().unwrap(); } - let files = extract_zip(&zip_path, &output_dir).unwrap(); - assert_eq!(files.len(), 2); + extract_zip(&zip_path, &output_dir).unwrap(); let extracted_hello = output_dir.join("hello.txt"); assert!(extracted_hello.exists()); @@ -89,5 +179,103 @@ mod tests { std::fs::read_to_string(&extracted_hello).unwrap(), "Hello, World!" ); + assert!(output_dir.join("subdir/nested.txt").exists()); + } + + #[test] + fn declared_size_sums_entries() { + let dir = tempfile::tempdir().unwrap(); + let zip_path = dir.path().join("test.zip"); + + { + let file = File::create(&zip_path).unwrap(); + let mut writer = zip::ZipWriter::new(file); + writer.start_file("a.txt", zip_options()).unwrap(); + writer.write_all(&[0u8; 100]).unwrap(); + writer.start_file("b.txt", zip_options()).unwrap(); + writer.write_all(&[0u8; 50]).unwrap(); + writer.finish().unwrap(); + } + + assert_eq!(declared_uncompressed_size(&zip_path).unwrap(), 150); + } + + #[test] + fn reject_symlink_entry() { + let dir = tempfile::tempdir().unwrap(); + let zip_path = dir.path().join("evil.zip"); + let output_dir = dir.path().join("output"); + std::fs::create_dir_all(&output_dir).unwrap(); + + { + let file = File::create(&zip_path).unwrap(); + let mut writer = zip::ZipWriter::new(file); + writer.start_file("benign.txt", zip_options()).unwrap(); + writer.write_all(b"decoy").unwrap(); + writer + .add_symlink("link", "../../../etc/passwd", zip_options()) + .unwrap(); + writer.finish().unwrap(); + } + + let err = extract_zip(&zip_path, &output_dir).unwrap_err(); + assert!( + matches!(err, ErgasiaError::UnsafeArchiveEntry { .. }), + "expected UnsafeArchiveEntry, got: {err}" + ); + assert_no_entries(&output_dir); + } + + #[test] + fn reject_absolute_path_entry() { + let dir = tempfile::tempdir().unwrap(); + let zip_path = dir.path().join("evil.zip"); + let output_dir = dir.path().join("output"); + std::fs::create_dir_all(&output_dir).unwrap(); + + { + let file = File::create(&zip_path).unwrap(); + let mut writer = zip::ZipWriter::new(file); + writer.start_file("Xetc/evil.txt", zip_options()).unwrap(); + writer.write_all(b"absolute").unwrap(); + writer.finish().unwrap(); + } + + // The zip writer sanitizes names, so patch the placeholder into a + // genuinely absolute entry name (same byte length keeps offsets valid). + let mut bytes = std::fs::read(&zip_path).unwrap(); + replace_bytes(&mut bytes, b"Xetc/evil.txt", b"/etc/evil.txt"); + std::fs::write(&zip_path, &bytes).unwrap(); + + let err = extract_zip(&zip_path, &output_dir).unwrap_err(); + assert!( + matches!(err, ErgasiaError::UnsafeArchiveEntry { .. }), + "expected UnsafeArchiveEntry, got: {err}" + ); + assert_no_entries(&output_dir); + } + + #[test] + fn reject_parent_traversal_entry() { + let dir = tempfile::tempdir().unwrap(); + let zip_path = dir.path().join("evil.zip"); + let output_dir = dir.path().join("output"); + std::fs::create_dir_all(&output_dir).unwrap(); + + { + let file = File::create(&zip_path).unwrap(); + let mut writer = zip::ZipWriter::new(file); + writer.start_file("../escape.txt", zip_options()).unwrap(); + writer.write_all(b"traversal").unwrap(); + writer.finish().unwrap(); + } + + let err = extract_zip(&zip_path, &output_dir).unwrap_err(); + assert!( + matches!(err, ErgasiaError::UnsafeArchiveEntry { .. }), + "expected UnsafeArchiveEntry, got: {err}" + ); + assert_no_entries(&output_dir); + assert!(!dir.path().join("escape.txt").exists()); } } diff --git a/crates/ergasia/src/lib.rs b/crates/ergasia/src/lib.rs index aa6068c9..3b8aecc5 100644 --- a/crates/ergasia/src/lib.rs +++ b/crates/ergasia/src/lib.rs @@ -5,10 +5,13 @@ pub mod seeding; pub mod session; pub mod state; +use std::future::Future; use std::path::Path; pub use error::ErgasiaError; -pub use extract::{ArchiveFormat, ExtractedFile, ExtractionResult, extract_archives}; +pub use extract::{ + ArchiveFormat, ExtractedFile, ExtractionLimits, ExtractionResult, extract_archives, +}; pub use progress::DownloadProgress; pub use seeding::{SeedingPolicy, TrackerSeedPolicy}; pub use session::TorrentSession; @@ -48,7 +51,5 @@ pub trait DownloadEngine: Send + Sync { &self, download_path: &Path, output_dir: &Path, - ) -> Result, ErgasiaError>; + ) -> impl Future, ErgasiaError>> + Send; } - -use std::future::Future; diff --git a/crates/ergasia/src/session.rs b/crates/ergasia/src/session.rs index 94f7261b..dbd20e01 100644 --- a/crates/ergasia/src/session.rs +++ b/crates/ergasia/src/session.rs @@ -10,24 +10,45 @@ use librqbit::{ AddTorrent, AddTorrentOptions, AddTorrentResponse, ManagedTorrent, Session, SessionOptions, SessionPersistenceConfig, TorrentStats, }; +use serde::{Deserialize, Serialize}; use themelion::ids::DownloadId; use tokio_util::sync::CancellationToken; use tracing::instrument; use crate::error::{ - AddTorrentSnafu, ErgasiaError, PauseActionSnafu, SessionInitSnafu, TorrentNotFoundSnafu, + AddTorrentSnafu, ErgasiaError, PauseActionSnafu, SessionInitSnafu, TorrentMapPersistenceSnafu, + TorrentNotFoundSnafu, }; use crate::seeding::SeedingPolicy; +const TORRENT_MAP_FILE: &str = "harmonia-torrent-map.json"; + pub struct SeedHandle { pub cancel: CancellationToken, } +// WHY: librqbit persists its own session state but knows nothing about +// harmonia's DownloadId, so the download_id <-> librqbit id mapping is +// persisted in a side-table colocated with the session state and reloaded on +// startup — otherwise every persisted torrent is unmanageable after a restart. +#[derive(Debug, Serialize, Deserialize)] +struct PersistedTorrentMap { + torrents: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +struct PersistedTorrentEntry { + download_id: DownloadId, + torrent_id: usize, +} + pub struct TorrentSession { session: Arc, pub policy: SeedingPolicy, pub seed_tracker: Arc>, torrent_map: DashMap, + map_path: PathBuf, + persist_lock: tokio::sync::Mutex<()>, } impl TorrentSession { @@ -78,12 +99,20 @@ impl TorrentSession { time_threshold: Duration::from_secs(config.seed_time_threshold_hours * 3600), }; - Ok(Self { + let torrent_session = Self { session, policy, seed_tracker: Arc::new(DashMap::new()), torrent_map: DashMap::new(), - }) + map_path: PathBuf::from(&config.session_state_path).join(TORRENT_MAP_FILE), + persist_lock: tokio::sync::Mutex::new(()), + }; + + // INVARIANT: the session is not handed out until torrent_map reflects + // every torrent librqbit restored from persisted state. + torrent_session.reconcile_persisted_torrents().await?; + + Ok(torrent_session) } #[instrument(skip(self, magnet_uri), fields(download_id = %download_id))] @@ -129,6 +158,15 @@ impl TorrentSession { AddTorrentResponse::Added(id, handle) | AddTorrentResponse::AlreadyManaged(id, handle) => { self.torrent_map.insert(download_id, id); + if let Err(persist_err) = self.persist_torrent_map().await { + // WHY: fail loudly but non-destructively — the mapping is + // rolled back so the caller sees a consistent failure, while + // the torrent stays in librqbit (an AlreadyManaged torrent + // may belong to another download, so deleting it here could + // destroy live data). + self.torrent_map.remove(&download_id); + return Err(persist_err); + } Ok((id, handle)) } AddTorrentResponse::ListOnly(_) => Err(AddTorrentSnafu { @@ -189,11 +227,260 @@ impl TorrentSession { })?; self.torrent_map.remove(&download_id); + self.persist_torrent_map().await?; + Ok(()) + } + + async fn reconcile_persisted_torrents(&self) -> Result<(), ErgasiaError> { + let persisted = self.load_torrent_map().await?; + + let mut restored = 0usize; + let mut dropped = 0usize; + for entry in persisted { + if self + .session + .get(TorrentIdOrHash::Id(entry.torrent_id)) + .is_some() + { + self.torrent_map.insert(entry.download_id, entry.torrent_id); + restored += 1; + } else { + tracing::warn!( + download_id = %entry.download_id, + torrent_id = entry.torrent_id, + "dropping torrent map entry no longer present in the session" + ); + dropped += 1; + } + } + + if dropped > 0 { + self.persist_torrent_map().await?; + } + + let live = self.session.with_torrents(|torrents| torrents.count()); + tracing::info!(restored, dropped, live, "reconciled persisted torrents"); + Ok(()) + } + + async fn load_torrent_map(&self) -> Result, ErgasiaError> { + let bytes = match tokio::fs::read(&self.map_path).await { + Ok(bytes) => bytes, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(e) => { + return Err(TorrentMapPersistenceSnafu { + path: self.map_path.clone(), + error: e.to_string(), + } + .build()); + } + }; + + match serde_json::from_slice::(&bytes) { + Ok(map) => Ok(map.torrents), + Err(e) => { + // WHY: a corrupt side-table must not brick startup — librqbit's + // own session state is intact. Quarantine the file for forensics + // and continue with an empty map (same managability as before + // the side-table existed). + let quarantine = self.map_path.with_extension("json.corrupt"); + tokio::fs::rename(&self.map_path, &quarantine).await.ok(); + tracing::warn!( + path = %self.map_path.display(), + quarantine = %quarantine.display(), + error = %e, + "torrent map file is corrupt; quarantined and starting with an empty map" + ); + Ok(Vec::new()) + } + } + } + + async fn persist_torrent_map(&self) -> Result<(), ErgasiaError> { + let _guard = self.persist_lock.lock().await; + + let torrents: Vec = self + .torrent_map + .iter() + .map(|kv| PersistedTorrentEntry { + download_id: *kv.key(), + torrent_id: *kv.value(), + }) + .collect(); + + let payload = + serde_json::to_vec_pretty(&PersistedTorrentMap { torrents }).map_err(|e| { + TorrentMapPersistenceSnafu { + path: self.map_path.clone(), + error: e.to_string(), + } + .build() + })?; + + if let Some(parent) = self.map_path.parent() { + tokio::fs::create_dir_all(parent).await.map_err(|e| { + TorrentMapPersistenceSnafu { + path: self.map_path.clone(), + error: e.to_string(), + } + .build() + })?; + } + + // WHY: write-then-rename keeps the side-table atomic — a crash mid-write + // leaves the previous map intact instead of a truncated file. + let tmp_path = self.map_path.with_extension("json.tmp"); + tokio::fs::write(&tmp_path, &payload).await.map_err(|e| { + TorrentMapPersistenceSnafu { + path: tmp_path.clone(), + error: e.to_string(), + } + .build() + })?; + tokio::fs::rename(&tmp_path, &self.map_path) + .await + .map_err(|e| { + TorrentMapPersistenceSnafu { + path: self.map_path.clone(), + error: e.to_string(), + } + .build() + })?; + Ok(()) } +} + +#[cfg(test)] +mod tests { + use std::path::Path; + use std::sync::LazyLock; + + use super::*; + + // WHY: each session binds listen ports and starts a DHT; serializing the + // session tests avoids port/DHT-persistence races inside one test binary. + static SESSION_TEST_LOCK: LazyLock> = + LazyLock::new(|| tokio::sync::Mutex::new(())); + + fn test_config(root: &Path, port_base: u16) -> ErgasiaConfig { + ErgasiaConfig { + download_dir: root.join("downloads"), + session_state_path: root.join("state"), + listen_port_range: [port_base, port_base + 8], + ..ErgasiaConfig::default() + } + } - pub fn reconcile_persisted_torrents(&self) { - let count = self.session.with_torrents(|torrents| torrents.count()); - tracing::info!(count, "reconciled persisted torrents"); + fn minimal_torrent_bytes(name: &str) -> bytes::Bytes { + let mut buf = Vec::new(); + buf.extend_from_slice(b"d4:infod6:lengthi11e4:name"); + buf.extend_from_slice(format!("{}:{}", name.len(), name).as_bytes()); + buf.extend_from_slice(b"12:piece lengthi16384e6:pieces20:"); + buf.extend_from_slice(&[0xAA; 20]); + buf.extend_from_slice(b"ee"); + bytes::Bytes::from(buf) + } + + async fn wait_for_session_state(state_dir: &Path) { + for _ in 0..100 { + let has_state = std::fs::read_dir(state_dir) + .map(|entries| { + entries.flatten().any(|e| { + e.path() + .extension() + .map(|ext| ext == "json") + .unwrap_or(false) + && e.metadata().map(|m| m.len() > 2).unwrap_or(false) + }) + }) + .unwrap_or(false); + if has_state { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!("librqbit session state was never persisted to {state_dir:?}"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn torrent_map_rebuilt_after_restart() { + let _guard = SESSION_TEST_LOCK.lock().await; + let dir = tempfile::tempdir().unwrap(); + let config = test_config(dir.path(), 24101); + let download_id = DownloadId::new(); + + { + let session = TorrentSession::new(&config).await.unwrap(); + session + .add_torrent_from_bytes(download_id, minimal_torrent_bytes("restart.bin")) + .await + .unwrap(); + assert!(session.get_stats(download_id).is_ok()); + wait_for_session_state(&config.session_state_path).await; + session.session.stop().await; + } + + let session = TorrentSession::new(&config).await.unwrap(); + assert!( + session.get_stats(download_id).is_ok(), + "download must be manageable after restart" + ); + session.session.stop().await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn reconcile_drops_stale_entries() { + let _guard = SESSION_TEST_LOCK.lock().await; + let dir = tempfile::tempdir().unwrap(); + let config = test_config(dir.path(), 24201); + let stale_id = DownloadId::new(); + + std::fs::create_dir_all(&config.session_state_path).unwrap(); + let map_path = config.session_state_path.join(TORRENT_MAP_FILE); + let stale = PersistedTorrentMap { + torrents: vec![PersistedTorrentEntry { + download_id: stale_id, + torrent_id: 4242, + }], + }; + std::fs::write(&map_path, serde_json::to_vec(&stale).unwrap()).unwrap(); + + let session = TorrentSession::new(&config).await.unwrap(); + assert!( + matches!( + session.get_stats(stale_id), + Err(ErgasiaError::TorrentNotFound { .. }) + ), + "stale entry must be dropped, not resurrected" + ); + + let rewritten: PersistedTorrentMap = + serde_json::from_slice(&std::fs::read(&map_path).unwrap()).unwrap(); + assert!( + rewritten.torrents.is_empty(), + "stale entry must be pruned from the persisted map" + ); + session.session.stop().await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn corrupt_torrent_map_is_quarantined() { + let _guard = SESSION_TEST_LOCK.lock().await; + let dir = tempfile::tempdir().unwrap(); + let config = test_config(dir.path(), 24301); + + std::fs::create_dir_all(&config.session_state_path).unwrap(); + let map_path = config.session_state_path.join(TORRENT_MAP_FILE); + std::fs::write(&map_path, b"{ not json").unwrap(); + + let session = TorrentSession::new(&config) + .await + .expect("corrupt side-table must not brick startup"); + assert!( + map_path.with_extension("json.corrupt").exists(), + "corrupt map must be quarantined for forensics" + ); + session.session.stop().await; } } diff --git a/crates/ergasia/src/state.rs b/crates/ergasia/src/state.rs index db69e18a..e24a97e1 100644 --- a/crates/ergasia/src/state.rs +++ b/crates/ergasia/src/state.rs @@ -32,7 +32,9 @@ impl DownloadState { | (Downloading, Completed) | (Downloading, Failed) | (Completed, Seeding) + | (Completed, Deleted) | (Seeding, SeedPolicySatisfied) + | (Seeding, Failed) | (SeedPolicySatisfied, Deleted) | (Queued, Failed) ) @@ -105,7 +107,9 @@ mod tests { (DownloadState::Downloading, DownloadState::Completed), (DownloadState::Downloading, DownloadState::Failed), (DownloadState::Completed, DownloadState::Seeding), + (DownloadState::Completed, DownloadState::Deleted), (DownloadState::Seeding, DownloadState::SeedPolicySatisfied), + (DownloadState::Seeding, DownloadState::Failed), (DownloadState::SeedPolicySatisfied, DownloadState::Deleted), (DownloadState::Queued, DownloadState::Failed), ]; diff --git a/crates/horismos/src/subsystems.rs b/crates/horismos/src/subsystems.rs index e3a52cbf..13dbf52d 100644 --- a/crates/horismos/src/subsystems.rs +++ b/crates/horismos/src/subsystems.rs @@ -283,6 +283,7 @@ pub struct ErgasiaConfig { pub max_connections_per_torrent: u32, pub magnet_resolve_timeout_seconds: u64, pub max_extraction_depth: u8, + pub max_decompression_ratio: f64, pub extraction_cleanup_hours: u64, } @@ -302,6 +303,7 @@ impl Default for ErgasiaConfig { max_connections_per_torrent: 0, magnet_resolve_timeout_seconds: 120, max_extraction_depth: 3, + max_decompression_ratio: 100.0, extraction_cleanup_hours: 48, } } diff --git a/crates/syntaxis/src/pipeline.rs b/crates/syntaxis/src/pipeline.rs index c1693fde..e2e9bf9c 100644 --- a/crates/syntaxis/src/pipeline.rs +++ b/crates/syntaxis/src/pipeline.rs @@ -75,7 +75,7 @@ pub(crate) async fn run_pipeline( let source_path = download_path.to_path_buf(); // Step 1: try extraction. On failure, mark failed and return immediately. - let working_path = match engine.extract(download_path, download_path) { + let working_path = match engine.extract(download_path, download_path).await { Ok(Some(result)) => { info!(extracted_path = %result.extracted_path.display(), "extracted archives"); result.extracted_path @@ -197,7 +197,7 @@ mod tests { async fn get_progress(&self, _id: DownloadId) -> Result { unimplemented!() } - fn extract( + async fn extract( &self, _path: &Path, _out: &Path, @@ -220,7 +220,7 @@ mod tests { async fn get_progress(&self, _id: DownloadId) -> Result { unimplemented!() } - fn extract( + async fn extract( &self, _path: &Path, _out: &Path, diff --git a/docs/download/archive.md b/docs/download/archive.md index 81f61221..a56deac9 100644 --- a/docs/download/archive.md +++ b/docs/download/archive.md @@ -255,6 +255,10 @@ extraction_temp_dir = "/data/downloads/.extraction" # Maximum nested archive depth. Prevents zip bombs and infinite recursion. max_extraction_depth = 3 +# Maximum declared-uncompressed to compressed size ratio. Archives declaring +# more than this ratio are rejected before any extraction write occurs. +max_decompression_ratio = 100.0 + # How long to retain failed extraction directories before cleanup (hours). extraction_cleanup_hours = 48 ```