diff --git a/Cargo.lock b/Cargo.lock index 5e90db008..af57a3768 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,6 +11,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "aquatic_udp_protocol" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84c3026324bc5073042edfbc47699cc60c1a9cf24849f5b104c18f98d2ad4175" +dependencies = [ + "byteorder", +] + [[package]] name = "arrayvec" version = "0.5.2" @@ -1733,8 +1742,9 @@ dependencies = [ [[package]] name = "torrust-tracker" -version = "2.0.1" +version = "2.1.0" dependencies = [ + "aquatic_udp_protocol", "binascii", "byteorder", "chrono", diff --git a/Cargo.toml b/Cargo.toml index db9b89856..6e78211ea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "torrust-tracker" -version = "2.0.1" +version = "2.1.0" authors = ["Mick van Dijke ", "Naim A. "] description = "A feature rich BitTorrent tracker." edition = "2018" @@ -30,3 +30,5 @@ rand = "0.8.4" env_logger = "0.9.0" config = "0.11" derive_more = "0.99" + +aquatic_udp_protocol = "0.1.0" diff --git a/src/common.rs b/src/common.rs index 82ea19ab8..1ee1190ab 100644 --- a/src/common.rs +++ b/src/common.rs @@ -1,4 +1,5 @@ use serde::{Deserialize, Serialize}; +use aquatic_udp_protocol::{AnnounceEvent, NumberOfBytes}; pub const MAX_PACKET_SIZE: usize = 0xffff; pub const MAX_SCRAPE_TORRENTS: u8 = 74; @@ -14,42 +15,25 @@ pub enum Actions { Error = 3, } -#[repr(u32)] -#[derive(Serialize, Deserialize, Clone, Copy)] -pub enum Events { - None = 0, - Complete = 1, - Started = 2, - Stopped = 3, -} - -#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub enum AnnounceEvent { - None, - Completed, +#[derive(Serialize, Deserialize)] +#[serde(remote = "AnnounceEvent")] +pub enum AnnounceEventDef { Started, Stopped, + Completed, + None } -impl AnnounceEvent { - #[inline] - pub fn from_i32(i: i32) -> Self { - match i { - 0 => Self::None, - 1 => Self::Completed, - 2 => Self::Started, - 3 => Self::Stopped, - _ => Self::None, - } - } -} - -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct AnnounceInterval(pub i32); +#[derive(Serialize, Deserialize)] +#[serde(remote = "NumberOfBytes")] +pub struct NumberOfBytesDef(pub i64); #[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, Ord)] pub struct InfoHash(pub [u8; 20]); +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, PartialOrd, Ord)] +pub struct PeerId(pub [u8; 20]); + impl InfoHash { pub fn to_string(&self) -> String { let mut buffer = [0u8; 40]; @@ -146,31 +130,6 @@ impl<'v> serde::de::Visitor<'v> for InfoHashVisitor { } } -#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct ConnectionId(pub i64); - -#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct TransactionId(pub i32); - -#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct NumberOfBytes(pub i64); - -#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct NumberOfPeers(pub i32); - -#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct NumberOfDownloads(pub i32); - -#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct Port(pub u16); - -#[repr(transparent)] -#[derive(Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug, PartialOrd, Ord)] -pub struct PeerId(pub [u8; 20]); - -#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct PeerKey(pub u32); - impl PeerId { pub fn get_client_name(&self) -> Option<&'static str> { if self.0[0] == b'M' { diff --git a/src/http_api_server.rs b/src/http_api_server.rs index 5f7339036..d33fa3ada 100644 --- a/src/http_api_server.rs +++ b/src/http_api_server.rs @@ -4,6 +4,7 @@ use std::cmp::min; use std::collections::{HashMap, HashSet}; use std::sync::Arc; use warp::{filters, reply, reply::Reply, serve, Filter, Server}; +use crate::TorrentPeer; use super::common::*; #[derive(Deserialize, Debug)] @@ -15,14 +16,11 @@ struct TorrentInfoQuery { #[derive(Serialize)] struct Torrent<'a> { info_hash: &'a InfoHash, - #[serde(flatten)] - data: &'a crate::tracker::TorrentEntry, seeders: u32, completed: u32, leechers: u32, - #[serde(skip_serializing_if = "Option::is_none")] - peers: Option>, + peers: Option>, } #[derive(Serialize, Debug)] @@ -87,7 +85,6 @@ pub fn build_server(tracker: Arc) -> Server) -> Server) -> Server = torrent_entry - .get_peers_iter() - .take(1000) - .map(|(peer_id, peer_info)| (peer_id.clone(), peer_info.clone())) - .collect(); + let peers = torrent_entry.get_peers(None); Ok(reply::json(&Torrent { info_hash: &info_hash, - data: torrent_entry, seeders, completed, leechers, diff --git a/src/http_server.rs b/src/http_server.rs index bf1f7f88d..f54f36eeb 100644 --- a/src/http_server.rs +++ b/src/http_server.rs @@ -17,13 +17,13 @@ use super::common::*; #[derive(Deserialize, Debug)] pub struct AnnounceRequest { - pub downloaded: NumberOfBytes, - pub uploaded: NumberOfBytes, + pub downloaded: u32, + pub uploaded: u32, pub key: String, pub peer_id: String, pub port: u16, pub info_hash: String, - pub left: NumberOfBytes, + pub left: u32, pub event: Option, pub compact: Option, } @@ -283,7 +283,7 @@ impl HttpServer { Some(v) => AuthKey::from_string(&v) }; - if let Err(e) = self.tracker.authenticate_request(&info_hash.unwrap(), &auth_key).await { + if let Err(e) = self.tracker.authenticate_request(&info_hash.unwrap(), auth_key).await { return match e { TorrentError::TorrentNotWhitelisted => { debug!("Info_hash not whitelisted."); @@ -297,8 +297,13 @@ impl HttpServer { debug!("Peer not authenticated."); Some(HttpServer::send_error("peer not authenticated")) } + _ => { + debug!("Unhandled HTTP error."); + Some(HttpServer::send_error("oops")) + } } } + None } diff --git a/src/lib.rs b/src/lib.rs index 375c0f903..d040d3719 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,7 +4,6 @@ pub mod http_server; pub mod tracker; pub mod http_api_server; pub mod common; -pub mod response; pub mod utils; pub mod database; pub mod key_manager; @@ -16,4 +15,3 @@ pub use self::http_server::*; pub use self::tracker::*; pub use self::http_api_server::*; pub use self::common::*; -pub use self::response::*; diff --git a/src/response.rs b/src/response.rs deleted file mode 100644 index 9734e3769..000000000 --- a/src/response.rs +++ /dev/null @@ -1,126 +0,0 @@ -use std; -use std::io::{Write}; -use std::net::{SocketAddr}; -use byteorder::{NetworkEndian, WriteBytesExt}; -use super::common::*; -use std::io; -use crate::TorrentPeer; - -#[derive(PartialEq, Eq, Clone, Debug)] -pub enum UdpResponse { - Connect(UdpConnectionResponse), - Announce(UdpAnnounceResponse), - Scrape(UdpScrapeResponse), - Error(UdpErrorResponse), -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct UdpConnectionResponse { - pub action: Actions, - pub transaction_id: TransactionId, - pub connection_id: ConnectionId, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct UdpAnnounceResponse { - pub action: Actions, - pub transaction_id: TransactionId, - pub interval: u32, - pub leechers: u32, - pub seeders: u32, - pub peers: Vec, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct UdpScrapeResponse { - pub action: Actions, - pub transaction_id: TransactionId, - pub torrent_stats: Vec, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct UdpScrapeResponseEntry { - pub seeders: i32, - pub completed: i32, - pub leechers: i32, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct UdpErrorResponse { - pub action: Actions, - pub transaction_id: TransactionId, - pub message: String, -} - -impl From for UdpResponse { - fn from(r: UdpConnectionResponse) -> Self { - Self::Connect(r) - } -} - -impl From for UdpResponse { - fn from(r: UdpAnnounceResponse) -> Self { - Self::Announce(r) - } -} - -impl From for UdpResponse { - fn from(r: UdpScrapeResponse) -> Self { - Self::Scrape(r) - } -} - -impl From for UdpResponse { - fn from(r: UdpErrorResponse) -> Self { - Self::Error(r) - } -} - -impl UdpResponse { - pub fn write_to_bytes(self, bytes: &mut impl Write) -> Result<(), io::Error> { - match self { - UdpResponse::Connect(r) => { - bytes.write_i32::(0)?; // 0 = connect - bytes.write_i32::(r.transaction_id.0)?; - bytes.write_i64::(r.connection_id.0)?; - }, - UdpResponse::Announce(r) => { - bytes.write_i32::(1)?; // 1 = announce - bytes.write_i32::(r.transaction_id.0)?; - bytes.write_u32::(r.interval)?; - bytes.write_u32::(r.leechers)?; - bytes.write_u32::(r.seeders)?; - - for peer in r.peers { - match peer.peer_addr { - SocketAddr::V4(socket_addr) => { - bytes.write_all(&socket_addr.ip().octets())?; - bytes.write_u16::(socket_addr.port())?; - } - SocketAddr::V6(socket_addr) => { - bytes.write_all(&socket_addr.ip().octets())?; - bytes.write_u16::(socket_addr.port())?; - } - } - } - }, - UdpResponse::Scrape(r) => { - bytes.write_i32::(2)?; // 2 = scrape - bytes.write_i32::(r.transaction_id.0)?; - - for torrent_stat in r.torrent_stats { - bytes.write_i32::(torrent_stat.seeders)?; - bytes.write_i32::(torrent_stat.completed)?; - bytes.write_i32::(torrent_stat.leechers)?; - } - }, - UdpResponse::Error(r) => { - bytes.write_i32::(3)?; - bytes.write_i32::(r.transaction_id.0)?; - bytes.write_all(r.message.as_bytes())?; - }, - } - - Ok(()) - } -} diff --git a/src/tracker.rs b/src/tracker.rs index 3e6bcca3e..768a4d03f 100644 --- a/src/tracker.rs +++ b/src/tracker.rs @@ -1,14 +1,16 @@ use serde::{Deserialize, Serialize}; +use serde; use std::borrow::Cow; use std::collections::BTreeMap; use tokio::sync::RwLock; -use crate::common::{NumberOfBytes, InfoHash}; +use crate::common::{InfoHash}; use super::common::*; use std::net::{SocketAddr, IpAddr}; -use crate::{Configuration, http_server, key_manager, udp_server}; +use crate::{Configuration, http_server, key_manager}; use std::collections::btree_map::Entry; use crate::database::SqliteDatabase; use std::sync::Arc; +use aquatic_udp_protocol::{AnnounceEvent, NumberOfBytes}; use log::debug; use crate::key_manager::{AuthKey}; use r2d2_sqlite::rusqlite; @@ -37,20 +39,22 @@ pub enum TrackerMode { #[derive(PartialEq, Eq, Debug, Clone, Serialize)] pub struct TorrentPeer { - #[serde(skip)] pub peer_id: PeerId, - #[serde(rename = "ip")] pub peer_addr: SocketAddr, #[serde(serialize_with = "ser_instant")] pub updated: std::time::Instant, + #[serde(with = "NumberOfBytesDef")] pub uploaded: NumberOfBytes, + #[serde(with = "NumberOfBytesDef")] pub downloaded: NumberOfBytes, + #[serde(with = "NumberOfBytesDef")] pub left: NumberOfBytes, + #[serde(with = "AnnounceEventDef")] pub event: AnnounceEvent, } impl TorrentPeer { - pub fn from_udp_announce_request(announce_request: &udp_server::AnnounceRequest, remote_addr: SocketAddr, peer_addr: Option) -> Self { + pub fn from_udp_announce_request(announce_request: &aquatic_udp_protocol::AnnounceRequest, remote_addr: SocketAddr, peer_addr: Option) -> Self { // Potentially substitute localhost IP with external IP let peer_addr = match peer_addr { None => SocketAddr::new(IpAddr::from(remote_addr.ip()), announce_request.port.0), @@ -64,7 +68,7 @@ impl TorrentPeer { }; TorrentPeer { - peer_id: announce_request.peer_id, + peer_id: PeerId(announce_request.peer_id.0), peer_addr, updated: std::time::Instant::now(), uploaded: announce_request.bytes_uploaded, @@ -102,9 +106,9 @@ impl TorrentPeer { peer_id: PeerId::from(announce_request.peer_id.as_bytes()), peer_addr, updated: std::time::Instant::now(), - uploaded: announce_request.uploaded, - downloaded: announce_request.downloaded, - left: announce_request.left, + uploaded: NumberOfBytes(announce_request.uploaded as i64), + downloaded: NumberOfBytes(announce_request.downloaded as i64), + left: NumberOfBytes(announce_request.left as i64), event } } @@ -151,7 +155,7 @@ impl TorrentEntry { } } - pub fn get_peers(&self, remote_addr: &std::net::SocketAddr) -> Vec { + pub fn get_peers(&self, remote_addr: Option<&std::net::SocketAddr>) -> Vec { let mut list = Vec::new(); for (_, peer) in self .peers @@ -161,8 +165,10 @@ impl TorrentEntry { { // skip ip address of client - if peer.peer_addr == *remote_addr { - //continue; + if let Some(remote_addr) = remote_addr { + if peer.peer_addr == *remote_addr { + continue; + } } list.push(peer.clone()); @@ -170,10 +176,6 @@ impl TorrentEntry { list } - pub fn get_peers_iter(&self) -> impl Iterator { - self.peers.iter() - } - pub fn update_torrent_stats_with_peer(&mut self, peer: &TorrentPeer, peer_old: Option) { match peer_old { None => { @@ -243,6 +245,8 @@ pub enum TorrentError { TorrentNotWhitelisted, PeerNotAuthenticated, PeerKeyNotValid, + NoPeersFound, + CouldNotSendResponse } pub struct TorrentTracker { @@ -282,7 +286,7 @@ impl TorrentTracker { key_manager::verify_auth_key(&db_key) } - pub async fn authenticate_request(&self, info_hash: &InfoHash, key: &Option) -> Result<(), TorrentError> { + pub async fn authenticate_request(&self, info_hash: &InfoHash, key: Option) -> Result<(), TorrentError> { match self.config.mode { TrackerMode::PublicMode => Ok(()), TrackerMode::ListedMode => { @@ -295,7 +299,7 @@ impl TorrentTracker { TrackerMode::PrivateMode => { match key { Some(key) => { - if self.verify_auth_key(key).await.is_err() { + if self.verify_auth_key(&key).await.is_err() { return Err(TorrentError::PeerKeyNotValid) } @@ -309,7 +313,7 @@ impl TorrentTracker { TrackerMode::PrivateListedMode => { match key { Some(key) => { - if self.verify_auth_key(key).await.is_err() { + if self.verify_auth_key(&key).await.is_err() { return Err(TorrentError::PeerKeyNotValid) } @@ -356,7 +360,7 @@ impl TorrentTracker { None } Some(entry) => { - Some(entry.get_peers(peer_addr)) + Some(entry.get_peers(Some(peer_addr))) } } } diff --git a/src/udp_server.rs b/src/udp_server.rs index 079ff67a0..cf0474f7c 100644 --- a/src/udp_server.rs +++ b/src/udp_server.rs @@ -1,236 +1,31 @@ use log::{debug}; use std; -use std::convert::TryInto; -use std::io; -use std::net::{Ipv4Addr, SocketAddr}; +use std::net::{SocketAddr}; use std::sync::Arc; -use std::io::{Cursor, Read}; +use std::io::{Cursor}; +use aquatic_udp_protocol::{AnnounceInterval, AnnounceRequest, AnnounceResponse, ConnectRequest, ConnectResponse, ErrorResponse, IpVersion, NumberOfDownloads, NumberOfPeers, Port, Request, Response, ResponsePeer, ScrapeRequest, ScrapeResponse, TorrentScrapeStatistics, TransactionId}; use tokio::net::UdpSocket; -use byteorder::{NetworkEndian, ReadBytesExt}; use super::common::*; -use crate::response::*; use crate::utils::get_connection_id; use crate::tracker::TorrentTracker; -use crate::{TorrentPeer, TrackerMode, TorrentError}; -use crate::key_manager::AuthKey; - -#[derive(PartialEq, Eq, Clone, Debug)] -pub enum Request { - Connect(ConnectRequest), - Announce(AnnounceRequest), - Scrape(ScrapeRequest), -} +use crate::{TorrentPeer, TorrentError}; -impl From for Request { - fn from(r: ConnectRequest) -> Self { - Self::Connect(r) - } +struct RequestError { + error: TorrentError, + transaction_id: TransactionId } -impl From for Request { - fn from(r: AnnounceRequest) -> Self { - Self::Announce(r) - } +struct AnnounceRequestWrapper { + announce_request: AnnounceRequest, + info_hash: super::common::InfoHash, } -impl From for Request { - fn from(r: ScrapeRequest) -> Self { - Self::Scrape(r) - } -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct ConnectRequest { - pub transaction_id: TransactionId, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct AnnounceRequest { - pub connection_id: ConnectionId, - pub transaction_id: TransactionId, - pub info_hash: InfoHash, - pub peer_id: PeerId, - pub bytes_downloaded: NumberOfBytes, - pub bytes_uploaded: NumberOfBytes, - pub bytes_left: NumberOfBytes, - pub event: AnnounceEvent, - pub ip_address: Option, - pub key: PeerKey, - pub peers_wanted: NumberOfPeers, - pub port: Port, - pub auth_key: Option, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct ScrapeRequest { - pub connection_id: ConnectionId, - pub transaction_id: TransactionId, - pub info_hashes: Vec, -} - -#[derive(Debug)] -pub struct RequestParseError { - pub transaction_id: Option, - pub message: Option, - pub error: Option, -} - -impl RequestParseError { - pub fn new(err: io::Error, transaction_id: i32) -> Self { - Self { - transaction_id: Some(TransactionId(transaction_id)), - message: None, - error: Some(err), - } - } - pub fn io(err: io::Error) -> Self { - Self { - transaction_id: None, - message: None, - error: Some(err), - } - } - pub fn text(transaction_id: i32, message: &str) -> Self { - Self { - transaction_id: Some(TransactionId(transaction_id)), - message: Some(message.to_string()), - error: None, - } - } -} - -impl Request { - pub fn from_bytes(bytes: &[u8]) -> Result { - let mut cursor = Cursor::new(bytes); - - let connection_id = cursor - .read_i64::() - .map_err(RequestParseError::io)?; - let action = cursor - .read_i32::() - .map_err(RequestParseError::io)?; - let transaction_id = cursor - .read_i32::() - .map_err(RequestParseError::io)?; - - - - match action { - // Connect - 0 => { - if connection_id == PROTOCOL_ID { - Ok((ConnectRequest { - transaction_id: TransactionId(transaction_id), - }) - .into()) - } else { - Err(RequestParseError::text( - transaction_id, - "Protocol identifier missing", - )) - } - } - - // Announce - 1 => { - let mut info_hash = [0; 20]; - let mut peer_id = [0; 20]; - let mut ip = [0; 4]; - - cursor - .read_exact(&mut info_hash) - .map_err(|err| RequestParseError::new(err, transaction_id))?; - cursor - .read_exact(&mut peer_id) - .map_err(|err| RequestParseError::new(err, transaction_id))?; - - let bytes_downloaded = cursor - .read_i64::() - .map_err(|err| RequestParseError::new(err, transaction_id))?; - let bytes_left = cursor - .read_i64::() - .map_err(|err| RequestParseError::new(err, transaction_id))?; - let bytes_uploaded = cursor - .read_i64::() - .map_err(|err| RequestParseError::new(err, transaction_id))?; - let event = cursor - .read_i32::() - .map_err(|err| RequestParseError::new(err, transaction_id))?; - - cursor - .read_exact(&mut ip) - .map_err(|err| RequestParseError::new(err, transaction_id))?; - - let key = cursor - .read_u32::() - .map_err(|err| RequestParseError::new(err, transaction_id))?; - let peers_wanted = cursor - .read_i32::() - .map_err(|err| RequestParseError::new(err, transaction_id))?; - let port = cursor - .read_u16::() - .map_err(|err| RequestParseError::new(err, transaction_id))?; - - // BEP 41: add auth key if available - let auth_key: Option = if bytes.len() > 98 + AUTH_KEY_LENGTH { - let mut key_buffer = [0; AUTH_KEY_LENGTH]; - // key should be the last bytes - cursor.set_position((bytes.len() - AUTH_KEY_LENGTH) as u64); - if cursor.read_exact(&mut key_buffer).is_ok() { - debug!("AuthKey buffer: {:?}", key_buffer); - AuthKey::from_buffer(key_buffer) - } else { - None - } - } else { - None - }; - - let opt_ip = if ip == [0; 4] { - None - } else { - Some(Ipv4Addr::from(ip)) - }; - - Ok((AnnounceRequest { - connection_id: ConnectionId(connection_id), - transaction_id: TransactionId(transaction_id), - info_hash: InfoHash(info_hash), - peer_id: PeerId(peer_id), - bytes_downloaded: NumberOfBytes(bytes_downloaded), - bytes_uploaded: NumberOfBytes(bytes_uploaded), - bytes_left: NumberOfBytes(bytes_left), - event: AnnounceEvent::from_i32(event), - ip_address: opt_ip, - key: PeerKey(key), - peers_wanted: NumberOfPeers(peers_wanted), - port: Port(port), - auth_key, - }) - .into()) - } - - // Scrape - 2 => { - let position = cursor.position() as usize; - let inner = cursor.into_inner(); - - let info_hashes = (&inner[position..]) - .chunks_exact(20) - .take(MAX_SCRAPE_TORRENTS as usize) - .map(|chunk| InfoHash(chunk.try_into().unwrap())) - .collect(); - - Ok((ScrapeRequest { - connection_id: ConnectionId(connection_id), - transaction_id: TransactionId(transaction_id), - info_hashes, - }) - .into()) - } - - _ => Err(RequestParseError::text(transaction_id, "Invalid action")), +impl AnnounceRequestWrapper { + pub fn new(announce_request: AnnounceRequest) -> Self { + AnnounceRequestWrapper { + announce_request: announce_request.clone(), + info_hash: InfoHash(announce_request.info_hash.0) } } } @@ -250,51 +45,6 @@ impl UdpServer { }) } - pub async fn authenticate_announce_request(&self, announce_request: &AnnounceRequest) -> Result<(), TorrentError> { - match self.tracker.config.mode { - TrackerMode::PublicMode => Ok(()), - TrackerMode::ListedMode => { - if !self.tracker.is_info_hash_whitelisted(&announce_request.info_hash).await { - return Err(TorrentError::TorrentNotWhitelisted) - } - - Ok(()) - } - TrackerMode::PrivateMode => { - match &announce_request.auth_key { - Some(auth_key) => { - if self.tracker.verify_auth_key(auth_key).await.is_err() { - return Err(TorrentError::PeerKeyNotValid) - } - - Ok(()) - } - None => { - return Err(TorrentError::PeerNotAuthenticated) - } - } - } - TrackerMode::PrivateListedMode => { - match &announce_request.auth_key { - Some(auth_key) => { - if self.tracker.verify_auth_key(auth_key).await.is_err() { - return Err(TorrentError::PeerKeyNotValid) - } - - if !self.tracker.is_info_hash_whitelisted(&announce_request.info_hash).await { - return Err(TorrentError::TorrentNotWhitelisted) - } - - Ok(()) - } - None => { - return Err(TorrentError::PeerNotAuthenticated) - } - } - } - } - } - pub async fn accept_packets(self) -> Result<(), std::io::Error> { let tracker = Arc::new(self); @@ -311,38 +61,12 @@ impl UdpServer { } async fn handle_packet(&self, remote_addr: SocketAddr, payload: &[u8]) { - let request = Request::from_bytes(&payload[..payload.len()]); + let request = Request::from_bytes(&payload[..payload.len()], MAX_SCRAPE_TORRENTS); match request { Ok(request) => { debug!("New request: {:?}", request); - - // todo: check for expired connection_id - match request { - Request::Connect(r) => self.handle_connect(remote_addr, r).await, - Request::Announce(r) => { - match self.tracker.authenticate_request(&r.info_hash, &r.auth_key).await { - Ok(()) => self.handle_announce(remote_addr, r).await, - Err(e) => { - match e { - TorrentError::TorrentNotWhitelisted => { - debug!("Info_hash not whitelisted."); - self.send_error(remote_addr, &r.transaction_id, "torrent not whitelisted").await; - } - TorrentError::PeerKeyNotValid => { - debug!("Peer key not valid."); - self.send_error(remote_addr, &r.transaction_id, "peer key not valid").await; - } - TorrentError::PeerNotAuthenticated => { - debug!("Peer not authenticated."); - self.send_error(remote_addr, &r.transaction_id, "peer not authenticated").await; - } - } - } - } - }, - Request::Scrape(r) => self.handle_scrape(remote_addr, r).await - } + self.handle_request(request, remote_addr).await; } Err(err) => { debug!("request_from_bytes error: {:?}", err); @@ -350,94 +74,148 @@ impl UdpServer { } } - async fn handle_connect(&self, remote_addr: SocketAddr, request: ConnectRequest) { + async fn handle_request(&self, request: Request, remote_addr: SocketAddr) { + // todo: check for expired connection_id + let request_result = match request { + Request::Connect(connect_request) => { + self.handle_connect(remote_addr, &connect_request).await + .map_err(|error| RequestError { error, transaction_id: connect_request.transaction_id }) + } + Request::Announce(announce_request) => { + self.handle_announce(remote_addr, &announce_request).await + .map_err(|error| RequestError { error, transaction_id: announce_request.transaction_id }) + } + Request::Scrape(scrape_request) => { + self.handle_scrape(&scrape_request).await + .map_err(|error| RequestError { error, transaction_id: scrape_request.transaction_id }) + } + }; + + match request_result { + Ok(response) => { + let _ = self.send_response(remote_addr, response).await; + } + Err(request_error) => { + let _ = self.handle_error(request_error.error, remote_addr, request_error.transaction_id).await; + } + } + } + + async fn handle_connect(&self, remote_addr: SocketAddr, request: &ConnectRequest) -> Result { let connection_id = get_connection_id(&remote_addr); - let response = UdpResponse::from(UdpConnectionResponse { - action: Actions::Connect, + let response = Response::from(ConnectResponse { transaction_id: request.transaction_id, connection_id, }); - let _ = self.send_response(remote_addr, response).await; + Ok(response) } - async fn handle_announce(&self, remote_addr: SocketAddr, request: AnnounceRequest) { - let peer = TorrentPeer::from_udp_announce_request(&request, remote_addr, self.tracker.config.get_ext_ip()); + async fn handle_announce(&self, remote_addr: SocketAddr, announce_request: &AnnounceRequest) -> Result { + let wrapped_announce_request = AnnounceRequestWrapper::new(announce_request.clone()); + self.tracker.authenticate_request(&wrapped_announce_request.info_hash, None).await?; + + let peer = TorrentPeer::from_udp_announce_request(&wrapped_announce_request.announce_request, remote_addr, self.tracker.config.get_ext_ip()); - match self.tracker.update_torrent_with_peer_and_get_stats(&request.info_hash, &peer).await { + return match self.tracker.update_torrent_with_peer_and_get_stats(&wrapped_announce_request.info_hash, &peer).await { Ok(torrent_stats) => { // get all peers excluding the client_addr - let peers = match self.tracker.get_torrent_peers(&request.info_hash, &peer.peer_addr).await { + let peers = match self.tracker.get_torrent_peers(&wrapped_announce_request.info_hash, &peer.peer_addr).await { Some(v) => v, None => { - debug!("announce: No peers found."); - return; + return Err(TorrentError::NoPeersFound); } }; - let response = UdpResponse::from(UdpAnnounceResponse { - action: Actions::Announce, - transaction_id: request.transaction_id, - interval: self.tracker.config.udp_tracker.announce_interval, - leechers: torrent_stats.leechers, - seeders: torrent_stats.seeders, - peers, + let response = Response::from(AnnounceResponse { + transaction_id: wrapped_announce_request.announce_request.transaction_id, + announce_interval: AnnounceInterval(self.tracker.config.udp_tracker.announce_interval as i32), + leechers: NumberOfPeers(torrent_stats.leechers as i32), + seeders: NumberOfPeers(torrent_stats.seeders as i32), + peers: peers.iter().map(|peer| + ResponsePeer { + ip_address: peer.peer_addr.ip(), + port: Port(peer.peer_addr.port()) + }).collect() }); - let _ = self.send_response(remote_addr, response).await; - } - Err(e) => { - debug!("{:?}", e); - self.send_error(remote_addr, &request.transaction_id, "error adding torrent").await; + Ok(response) } + Err(e) => Err(e) } } - async fn handle_scrape(&self, remote_addr: SocketAddr, request: ScrapeRequest) { - let mut scrape_response = UdpScrapeResponse { - action: Actions::Scrape, - transaction_id: request.transaction_id, - torrent_stats: Vec::new(), - }; - + async fn handle_scrape(&self, request: &ScrapeRequest) -> Result { let db = self.tracker.get_torrents().await; + let mut torrent_stats: Vec = Vec::new(); + for info_hash in request.info_hashes.iter() { + let info_hash = InfoHash(info_hash.0); let scrape_entry = match db.get(&info_hash) { Some(torrent_info) => { let (seeders, completed, leechers) = torrent_info.get_stats(); - UdpScrapeResponseEntry { - seeders: seeders as i32, - completed: completed as i32, - leechers: leechers as i32, + TorrentScrapeStatistics { + seeders: NumberOfPeers(seeders as i32), + completed: NumberOfDownloads(completed as i32), + leechers: NumberOfPeers(leechers as i32), } } None => { - UdpScrapeResponseEntry { - seeders: 0, - completed: 0, - leechers: 0, + TorrentScrapeStatistics { + seeders: NumberOfPeers(0), + completed: NumberOfDownloads(0), + leechers: NumberOfPeers(0), } } }; - scrape_response.torrent_stats.push(scrape_entry); + torrent_stats.push(scrape_entry); } - let response = UdpResponse::from(scrape_response); + let response = Response::from(ScrapeResponse { + transaction_id: request.transaction_id, + torrent_stats + }); + + Ok(response) + } + + async fn handle_error(&self, e: TorrentError, remote_addr: SocketAddr, tx_id: TransactionId) { + let mut err_msg = "oops"; - let _ = self.send_response(remote_addr, response).await; + match e { + TorrentError::TorrentNotWhitelisted => { + debug!("Info_hash not whitelisted."); + err_msg = "info hash not whitelisted"; + } + TorrentError::PeerKeyNotValid => { + debug!("Peer key not valid."); + err_msg = "peer key not valid"; + } + TorrentError::PeerNotAuthenticated => { + debug!("Peer not authenticated."); + err_msg = "peer not authenticated"; + } + TorrentError::NoPeersFound => { + debug!("No peers found."); + err_msg = "no peers found"; + } + _ => {} + } + + self.send_error(remote_addr, tx_id, err_msg).await; } - async fn send_response(&self, remote_addr: SocketAddr, response: UdpResponse) -> Result { + async fn send_response(&self, remote_addr: SocketAddr, response: Response) -> Result { debug!("sending response to: {:?}", &remote_addr); let buffer = vec![0u8; MAX_PACKET_SIZE]; let mut cursor = Cursor::new(buffer); - match response.write_to_bytes(&mut cursor) { + match response.write(&mut cursor, IpVersion::IPv4) { Ok(_) => { let position = cursor.position() as usize; let inner = cursor.get_ref(); @@ -458,6 +236,15 @@ impl UdpServer { } } + async fn send_error(&self, remote_addr: SocketAddr, transaction_id: TransactionId, error_msg: &str) { + let response = Response::from(ErrorResponse { + transaction_id, + message: error_msg.to_string(), + }); + + let _ = self.send_response(remote_addr, response).await; + } + async fn send_packet(&self, remote_addr: &SocketAddr, payload: &[u8]) -> Result { match self.socket.send_to(payload, remote_addr).await { Err(err) => { @@ -467,16 +254,4 @@ impl UdpServer { Ok(sz) => Ok(sz), } } - - async fn send_error(&self, remote_addr: SocketAddr, transaction_id: &TransactionId, error_msg: &str) { - let error_response = UdpErrorResponse { - action: Actions::Error, - transaction_id: transaction_id.clone(), - message: error_msg.to_string(), - }; - - let response = UdpResponse::from(error_response); - - let _ = self.send_response(remote_addr, response).await; - } } diff --git a/src/utils.rs b/src/utils.rs index 11c61e4fb..5790e6067 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,8 +1,8 @@ use std::net::SocketAddr; -use crate::common::*; use std::time::SystemTime; use std::error::Error; use std::fmt::Write; +use aquatic_udp_protocol::ConnectionId; pub fn get_connection_id(remote_address: &SocketAddr) -> ConnectionId { match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {