diff --git a/crates/ipx/src/scan.rs b/crates/ipx/src/scan.rs index 5f7d472..f58c00f 100644 --- a/crates/ipx/src/scan.rs +++ b/crates/ipx/src/scan.rs @@ -1060,6 +1060,39 @@ mod tests { assert_eq!(ctx.db.feed_summary("iconic").await.unwrap().image, Some(icon)); } + #[tokio::test] + async fn a_torrent_enclosure_goes_to_the_torrent_worker_not_the_file_downloader() { + let body = b"TOneg1\ + "; + let scan = |ctx: Arc| async move { + let got = feed::Fetched::Body { bytes: body.to_vec(), etag: None, last_modified: None }; + let cfg = config::Feed { url: "http://127.0.0.1:9/feed".into(), ..feed() }; + match scan_one(&ctx, "t", &cfg, &db::HttpState::default(), true, Some((Ok((got, None)), None))) + .await + .unwrap() + { + Outcome::Feed(s) => (s.downloaded, s.torrents), + _ => panic!("a feed"), + } + }; + let state = |ctx: Arc| async move { ctx.db.strings_for_test("SELECT state FROM enclosures").await }; + + // Detached, as the daemon runs them: no slot free, so it waits its turn as 'downloading'. + let mut ctx = test_ctx(config::Config::default()).await; + ctx.detach_torrents = true; + ctx.torrent_slots = Arc::new(tokio::sync::Semaphore::new(0)); + let ctx = Arc::new(ctx); + assert_eq!(scan(ctx.clone()).await, (0, 1), "counted as a torrent, not a download"); + assert_eq!(state(ctx).await, ["downloading"], "so the next scan does not queue it again"); + + // Turned off, it is skipped and still counted. + let mut cfg = config::Config::default(); + cfg.torrent.enabled = false; + let ctx = Arc::new(test_ctx(cfg).await); + assert_eq!(scan(ctx.clone()).await, (0, 1)); + assert_eq!(state(ctx).await, ["skipped"]); + } + #[tokio::test] async fn nothing_from_a_feed_that_never_posted_is_an_error_and_from_one_that_has_is_not() { let base = empty_or_feed_server().await; diff --git a/crates/ipx/src/torrent.rs b/crates/ipx/src/torrent.rs index f44ad40..880e16c 100644 --- a/crates/ipx/src/torrent.rs +++ b/crates/ipx/src/torrent.rs @@ -29,6 +29,9 @@ impl Torrents { let opts = SessionOptions { listen: Some(ListenerOptions { listen_addr, ..Default::default() }), client_name_and_version: Some(concat!("ipx ", env!("CARGO_PKG_VERSION")).into()), + // A test's swarm is two sessions and a tracker on 127.0.0.1, not the public DHT or the LAN. + dht: if cfg!(test) { None } else { Some(Default::default()) }, + disable_local_service_discovery: cfg!(test), ..Default::default() }; @@ -184,6 +187,136 @@ pub fn ratio(uploaded: u64, total: u64) -> f64 { mod tests { use super::*; + fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port() + } + + /// A swarm on 127.0.0.1: a seeding session holding `file`, and a web server answering for the + /// .torrent (`/ep.torrent`) and as its tracker, naming the seed as the one peer. The server's + /// address, and the seeding session, which stops seeding when dropped. + async fn swarm(file: &Path) -> (SocketAddr, Arc) { + let web = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let at = web.local_addr().unwrap(); + let seed_port = free_port(); + let made = librqbit::create_torrent( + file, + librqbit::CreateTorrentOptions { + name: None, + trackers: vec![format!("http://{at}/announce")], + piece_length: Some(16 * 1024), + }, + &librqbit::spawn_utils::BlockingSpawner::new(2), + ) + .await + .unwrap(); + let torrent = made.as_bytes().unwrap(); + // Compact peers: the address's four bytes, then the port's two. + let mut peers = b"d8:intervali60e5:peers6:".to_vec(); + peers.extend_from_slice(&[127, 0, 0, 1]); + peers.extend_from_slice(&seed_port.to_be_bytes()); + peers.push(b'e'); + let app = axum::Router::new() + .route("/ep.torrent", axum::routing::get(move || std::future::ready(torrent.clone()))) + .route("/announce", axum::routing::get(move || std::future::ready(peers.clone()))); + tokio::spawn(async move { axum::serve(web, app).await }); + + let seed = Session::new_with_opts( + file.parent().unwrap().to_owned(), + SessionOptions { + dht: None, + disable_local_service_discovery: true, + listen: Some(ListenerOptions { + listen_addr: format!("127.0.0.1:{seed_port}").parse().unwrap(), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + .unwrap(); + seed.add_torrent( + AddTorrent::from_bytes(made.as_bytes().unwrap()), + Some(AddTorrentOptions { + output_folder: Some(file.parent().unwrap().display().to_string()), + overwrite: true, + ..Default::default() + }), + ) + .await + .unwrap(); + (at, seed) + } + + #[tokio::test(flavor = "multi_thread")] + async fn a_detached_torrent_lands_in_its_own_folder_is_recorded_and_frees_its_slot() { + let root = std::env::temp_dir().join(format!("ipx-torrent-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let (seeded, downloads) = (root.join("seed"), root.join("downloads")); + std::fs::create_dir_all(&seeded).unwrap(); + let episode: Vec = (0..200_000u32).map(|i| (i * 7 % 251) as u8).collect(); + std::fs::write(seeded.join("episode.mp3"), &episode).unwrap(); + let (at, _seed) = swarm(&seeded.join("episode.mp3")).await; + + let mut cfg = Config::default(); + cfg.general.download_dir = downloads.clone(); + cfg.torrent.enabled = true; + let port = free_port(); + cfg.torrent.port_range = format!("{port}-{port}"); + cfg.torrent.seed_ratio = 0.0; + let mut ctx = crate::test_util::test_ctx(cfg.clone()).await; + ctx.detach_torrents = true; + let ctx = Arc::new(ctx); + let url = format!("http://{at}/ep.torrent"); + ctx.db + .exec_for_test(&format!( + "INSERT INTO enclosures (id, feed_id, guid, url, state) VALUES (9, 'f', 'g', '{url}', 'downloading')" + )) + .await + .unwrap(); + + crate::scan::spawn_torrent(&ctx, "f".into(), 9, url.clone(), downloads.join("Show")); + let row = tokio::time::timeout(Duration::from_secs(90), async { + loop { + let row = ctx.db.enclosure(9).await.unwrap().unwrap(); + if row.state != "downloading" { + return row; + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + }) + .await + .expect("the torrent finished within 90 s"); + let folder = own_folder(&downloads.join("Show"), 9); + assert_eq!((row.state.as_str(), row.last_error.as_deref()), ("done", None)); + assert_eq!(row.path.as_deref(), Some(folder.display().to_string().as_str()), "the folder is ours (#318)"); + assert_eq!(std::fs::read(folder.join("episode.mp3")).unwrap(), episode); + // The slot is let go once the row is written; give the task its moment. + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(ctx.torrent_slots.available_permits(), 2, "the slot is free again"); + let _ = std::fs::remove_dir_all(&root); + } + + #[tokio::test] + async fn a_torrent_file_from_an_internal_address_is_refused_unless_allowed() { + let app = axum::Router::new().route("/ep.torrent", axum::routing::get(|| async { "d4:infode" })); + let web = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let at = web.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(web, app).await }); + let guarded = |allowed: &[&str]| { + let allowed = + crate::feed::guard::Allowed::parse(&allowed.iter().map(|s| s.to_string()).collect::>()).unwrap(); + reqwest::Client::builder().connector_layer(crate::feed::guard::Guard::new(allowed)).build().unwrap() + }; + let url = format!("http://{at}/ep.torrent"); + let err = torrent_file(&guarded(&[]), &url).await.unwrap_err(); + assert!(format!("{err:#}").contains("fetching the .torrent"), "{err:#}"); + assert_eq!( + torrent_file(&guarded(&["127.0.0.1"]), &url).await.unwrap(), + b"d4:infode", + "fetch_internal allows it" + ); + } + #[test] fn ratio_handles_the_empty_torrent() { assert_eq!(ratio(0, 100), 0.0);