Add a torrent from-link subcommand

It is now possible to create a torrent file from a magnet link. The
search is performed with minimal concurrency, and so may take a while to
complete.

type: added
fixes: casey#255

Cargo.toml

@@ -33,6 +33,7 @@ rand              = "0.7.3"
 open              = "1.4.0"
 pretty_assertions = "0.6.0"
 pretty_env_logger = "0.4.0"
+rayon             = "<=1.6.0"
 regex             = "1.0.0"
 serde-hex         = "0.1.0"
 serde_bytes       = "0.11.0"

bin/gen/config.yaml

@@ -19,6 +19,10 @@ examples:
   text:    "Intermodal can be used to create `.torrent` files:"
   code:    "imdl torrent create --input foo"
 
+- command: imdl torrent from-link
+  text:    "Intermodal can be used to create a `.torrent` file from a magnet link:"
+  code:    "imdl torrent from-link magnet:?foo"
+
 - command: imdl torrent show
   text:    "Print information about existing `.torrent` files:"
   code:    "imdl torrent show --input foo.torrent"

src/common.rs

@@ -19,7 +19,7 @@ pub(crate) use std::{
   process::ExitStatus,
   str::{self, FromStr},
   string::FromUtf8Error,
-  sync::Once,
+  sync::{mpsc::channel, Once},
   time::{Duration, SystemTime, SystemTimeError},
   usize,
 };
@@ -94,6 +94,7 @@ mod test {
     ops::{Deref, DerefMut},
     process::Command,
     rc::Rc,
+    thread,
   };
 
   // test dependencies

src/error.rs

@@ -32,6 +32,8 @@ pub(crate) enum Error {
   Filesystem { source: io::Error, path: PathBuf },
   #[snafu(display("Error searching for files: {}", source))]
   FileSearch { source: ignore::Error },
+  #[snafu(display("Failed to fetch infodict from accessible peers"))]
+  FromLinkNoInfo,
   #[snafu(display("Invalid glob: {}", source))]
   GlobParse { source: globset::Error },
   #[snafu(display("Failed to serialize torrent info dictionary: {}", source))]

src/magnet_link.rs

@@ -1,12 +1,12 @@
 use crate::common::*;
 
-#[derive(Debug, PartialEq)]
+#[derive(Clone, Debug, PartialEq)]
 pub(crate) struct MagnetLink {
-  infohash: Infohash,
-  name: Option<String>,
-  peers: Vec<HostPort>,
-  trackers: Vec<Url>,
-  indices: BTreeSet<u64>,
+  pub(crate) infohash: Infohash,
+  pub(crate) name: Option<String>,
+  pub(crate) peers: Vec<HostPort>,
+  pub(crate) trackers: Vec<Url>,
+  pub(crate) indices: BTreeSet<u64>,
 }
 
 impl MagnetLink {

src/peer/client.rs

@@ -16,15 +16,15 @@ pub(crate) enum State {
 }
 
 impl Client {
-  pub(crate) fn connect(addr: &SocketAddr, infohash: Infohash) -> Result<Self> {
-    let conn = Connection::new(addr, infohash)?;
+  pub(crate) fn connect(addr: &SocketAddr, infohash: &Infohash) -> Result<Self> {
+    let conn = Connection::new(addr, *infohash)?;
 
     if !conn.supports_extension_protocol() {
       return Err(Error::PeerUtMetadataNotSupported);
     }
 
     Ok(Client {
-      infohash,
+      infohash: *infohash,
       conn,
       state: State::Idle,
       extension_handshake: None,
@@ -207,6 +207,76 @@ impl Client {
       info: None,
     })
   }
+
+  #[cfg(test)]
+  pub(crate) fn new_info_dict_seeder(info: &Info) -> (thread::JoinHandle<()>, SocketAddr) {
+    let info_dict = bendy::serde::ser::to_bytes(info).unwrap();
+    let infohash = Infohash::from_bencoded_info_dict(&info_dict);
+    let listener = TcpListener::bind((Ipv4Addr::UNSPECIFIED, 0)).unwrap();
+    let addr = (Ipv4Addr::LOCALHOST, listener.local_addr().unwrap().port()).into();
+
+    let handle = thread::spawn(move || {
+      let mut probe = Client::listen(&listener, infohash).unwrap();
+      // Send extended handshake with metadata size set.
+      let handshake = msg::extended::Handshake {
+        metadata_size: Some(info_dict.len()),
+        ..msg::extended::Handshake::default()
+      };
+      let msg = peer::Message::new_extended(
+        msg::extended::Id::Handshake,
+        &bendy::serde::ser::to_bytes(&handshake).unwrap(),
+      );
+      probe.conn.send(&msg).unwrap();
+
+      // The first message from the fetcher is an extension handshake.
+      let msg = probe.conn.next().unwrap();
+      assert_eq!(msg.id, msg::Id::Extended);
+      let (id, _) = msg.parse_extended_payload().unwrap();
+      assert_eq!(id, msg::extended::Id::Handshake);
+      probe.handle_msg(&msg).unwrap();
+
+      let mut pieces = info_dict.len() / msg::extended::UtMetadata::PIECE_LENGTH;
+      if info_dict.len() % msg::extended::UtMetadata::PIECE_LENGTH > 0 {
+        pieces += 1;
+      }
+
+      // Respond to any servicable ut_metadata request. Ignore errors.
+      loop {
+        let msg = match probe.conn.next() {
+          Ok(msg) => msg,
+          Err(_) => continue,
+        };
+
+        let payload = match msg.parse_extended_payload() {
+          Ok((_, payload)) => payload,
+          Err(_) => continue,
+        };
+
+        let req: msg::extended::UtMetadata = match bendy::serde::de::from_bytes(payload) {
+          Ok(req) => req,
+          Err(_) => continue,
+        };
+        if req.piece > pieces {
+          continue;
+        }
+
+        let range = std::ops::Range {
+          start: msg::extended::UtMetadata::PIECE_LENGTH * req.piece,
+          end: if pieces == 1 {
+            info_dict.len()
+          } else {
+            msg::extended::UtMetadata::PIECE_LENGTH * (req.piece + 1)
+          },
+        };
+
+        probe
+          .send_ut_metadata_data(req.piece, info_dict.len(), &info_dict[range])
+          .unwrap();
+      }
+    });
+
+    (handle, addr)
+  }
 }
 
 #[cfg(test)]

src/subcommand/torrent.rs

@@ -2,6 +2,7 @@ use crate::common::*;
 
 mod announce;
 mod create;
+mod from_link;
 mod link;
 mod piece_length;
 mod show;
@@ -17,6 +18,7 @@ mod verify;
 pub(crate) enum Torrent {
   Announce(announce::Announce),
   Create(create::Create),
+  FromLink(from_link::FromLink),
   Link(link::Link),
   #[structopt(alias = "piece-size")]
   PieceLength(piece_length::PieceLength),
@@ -30,6 +32,7 @@ impl Torrent {
     match self {
       Self::Announce(announce) => announce.run(env),
       Self::Create(create) => create.run(env, options),
+      Self::FromLink(from_link) => from_link.run(env, options),
       Self::Link(link) => link.run(env),
       Self::PieceLength(piece_length) => piece_length.run(env),
       Self::Show(show) => show.run(env),

src/subcommand/torrent/announce.rs

@@ -60,7 +60,7 @@ impl Announce {
         }
       };
 
-      let client = match tracker::Client::from_url(tracker_url) {
+      let client = match tracker::Client::from_url(&tracker_url) {
         Ok(client) => client,
         Err(err) => {
           errln!(env, "Couldn't build tracker client. {}", err)?;
@@ -69,7 +69,7 @@ impl Announce {
       };
 
       usable_trackers += 1;
-      match client.announce_exchange(infohash) {
+      match client.announce_exchange(&infohash) {
         Ok(peer_list) => peers.extend(peer_list),
         Err(err) => errln!(env, "Announce failed: {}", err)?,
       }