From 0e12dd1b13a4567cdaa472245eb2436ed87c27b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Sat, 2 Dec 2023 12:24:41 +0100 Subject: [PATCH] use zerocopy in udp protocol, easy running transfer CI locally --- .../actions/test-file-transfers/Dockerfile | 10 - .../actions/test-file-transfers/entrypoint.sh | 5 +- Cargo.lock | 26 + crates/peer_id/Cargo.toml | 3 +- crates/peer_id/src/lib.rs | 18 +- crates/udp/src/common.rs | 37 +- crates/udp/src/workers/socket/mio.rs | 23 +- crates/udp/src/workers/socket/storage.rs | 10 +- crates/udp/src/workers/socket/uring/mod.rs | 2 +- crates/udp/src/workers/socket/validator.rs | 4 +- crates/udp/src/workers/swarm/mod.rs | 35 +- crates/udp/src/workers/swarm/storage.rs | 15 +- crates/udp/tests/common/mod.rs | 27 +- crates/udp/tests/invalid_connection_id.rs | 25 +- crates/udp/tests/requests_responses.rs | 20 +- crates/udp_bench/src/announce.rs | 25 +- crates/udp_bench/src/scrape.rs | 8 +- crates/udp_load_test/src/utils.rs | 2 +- .../udp_load_test/src/worker/request_gen.rs | 23 +- crates/udp_protocol/Cargo.toml | 2 + crates/udp_protocol/src/common.rs | 171 +++++- crates/udp_protocol/src/request.rs | 499 ++++++++---------- crates/udp_protocol/src/response.rs | 427 +++++++-------- docker/ci.Dockerfile | 18 + 24 files changed, 783 insertions(+), 652 deletions(-) delete mode 100644 .github/actions/test-file-transfers/Dockerfile create mode 100644 docker/ci.Dockerfile diff --git a/.github/actions/test-file-transfers/Dockerfile b/.github/actions/test-file-transfers/Dockerfile deleted file mode 100644 index aaf8299..0000000 --- a/.github/actions/test-file-transfers/Dockerfile +++ /dev/null @@ -1,10 +0,0 @@ -# Not used by Github action, but can be used to run test locally: -# 1. docker build -t aquatic ./path/to/Dockerfile -# 2. docker run aquatic -# 3. On failure, run `docker rmi aquatic -f` and go back to step 1 - -FROM rust:bullseye - -COPY entrypoint.sh /entrypoint.sh - -ENTRYPOINT ["/entrypoint.sh"] \ No newline at end of file diff --git a/.github/actions/test-file-transfers/entrypoint.sh b/.github/actions/test-file-transfers/entrypoint.sh index 9672b86..ccac05c 100755 --- a/.github/actions/test-file-transfers/entrypoint.sh +++ b/.github/actions/test-file-transfers/entrypoint.sh @@ -68,7 +68,10 @@ tls_private_key_path = './key.pk8' " > tls.toml ./target/debug/aquatic http -c tls.toml > "$HOME/tls.log" 2>&1 & -echo "[network] +echo " +log_level = 'trace' + +[network] address = '127.0.0.1:3000'" > udp.toml ./target/debug/aquatic udp -c udp.toml > "$HOME/udp.log" 2>&1 & diff --git a/Cargo.lock b/Cargo.lock index 63fbd46..2b80164 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -199,6 +199,7 @@ dependencies = [ "quickcheck", "regex", "serde", + "zerocopy", ] [[package]] @@ -303,8 +304,10 @@ dependencies = [ "aquatic_peer_id", "byteorder", "either", + "pretty_assertions", "quickcheck", "quickcheck_macros", + "zerocopy", ] [[package]] @@ -875,6 +878,12 @@ dependencies = [ "powerfmt", ] +[[package]] +name = "diff" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" + [[package]] name = "digest" version = "0.10.7" @@ -2039,6 +2048,16 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de" +[[package]] +name = "pretty_assertions" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af7cee1a6c8a5b9208b3cb1061f10c0cb689087b3d8ce85fb9d2dd7a29b6ba66" +dependencies = [ + "diff", + "yansi", +] + [[package]] name = "privdrop" version = "0.5.4" @@ -3135,12 +3154,19 @@ version = "0.48.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" +[[package]] +name = "yansi" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09041cd90cf85f7f8b2df60c646f853b7f535ce68f85244eb6731cf89fa498ec" + [[package]] name = "zerocopy" version = "0.7.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e97e415490559a91254a2979b4829267a57d2fcd741a98eee8b722fb57289aa0" dependencies = [ + "byteorder", "zerocopy-derive", ] diff --git a/crates/peer_id/Cargo.toml b/crates/peer_id/Cargo.toml index 5563c1d..4acc099 100644 --- a/crates/peer_id/Cargo.toml +++ b/crates/peer_id/Cargo.toml @@ -20,4 +20,5 @@ compact_str = "0.7" hex = "0.4" regex = "1" serde = { version = "1", features = ["derive"] } -quickcheck = { version = "1", optional = true } \ No newline at end of file +quickcheck = { version = "1", optional = true } +zerocopy = { version = "0.7", features = ["derive"] } \ No newline at end of file diff --git a/crates/peer_id/src/lib.rs b/crates/peer_id/src/lib.rs index ce1a463..dbc7fcf 100644 --- a/crates/peer_id/src/lib.rs +++ b/crates/peer_id/src/lib.rs @@ -3,8 +3,24 @@ use std::{borrow::Cow, fmt::Display, sync::OnceLock}; use compact_str::{format_compact, CompactString}; use regex::bytes::Regex; use serde::{Deserialize, Serialize}; +use zerocopy::{AsBytes, FromBytes, FromZeroes}; -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + PartialOrd, + Ord, + Hash, + Serialize, + Deserialize, + AsBytes, + FromBytes, + FromZeroes, +)] +#[repr(transparent)] pub struct PeerId(pub [u8; 20]); impl PeerId { diff --git a/crates/udp/src/common.rs b/crates/udp/src/common.rs index 5e64376..1fe08d1 100644 --- a/crates/udp/src/common.rs +++ b/crates/udp/src/common.rs @@ -1,6 +1,5 @@ use std::collections::BTreeMap; use std::hash::Hash; -use std::net::{Ipv4Addr, Ipv6Addr}; use std::sync::atomic::AtomicUsize; use std::sync::Arc; @@ -35,8 +34,8 @@ pub enum ConnectedRequest { #[derive(Debug)] pub enum ConnectedResponse { - AnnounceIpv4(AnnounceResponse), - AnnounceIpv6(AnnounceResponse), + AnnounceIpv4(AnnounceResponse), + AnnounceIpv6(AnnounceResponse), Scrape(PendingScrapeResponse), } @@ -125,7 +124,7 @@ impl PeerStatus { pub fn from_event_and_bytes_left(event: AnnounceEvent, bytes_left: NumberOfBytes) -> Self { if event == AnnounceEvent::Stopped { Self::Stopped - } else if bytes_left.0 == 0 { + } else if bytes_left.0.get() == 0 { Self::Seeding } else { Self::Leeching @@ -207,17 +206,17 @@ mod tests { let f = PeerStatus::from_event_and_bytes_left; - assert_eq!(Stopped, f(AnnounceEvent::Stopped, NumberOfBytes(0))); - assert_eq!(Stopped, f(AnnounceEvent::Stopped, NumberOfBytes(1))); + assert_eq!(Stopped, f(AnnounceEvent::Stopped, NumberOfBytes::new(0))); + assert_eq!(Stopped, f(AnnounceEvent::Stopped, NumberOfBytes::new(1))); - assert_eq!(Seeding, f(AnnounceEvent::Started, NumberOfBytes(0))); - assert_eq!(Leeching, f(AnnounceEvent::Started, NumberOfBytes(1))); + assert_eq!(Seeding, f(AnnounceEvent::Started, NumberOfBytes::new(0))); + assert_eq!(Leeching, f(AnnounceEvent::Started, NumberOfBytes::new(1))); - assert_eq!(Seeding, f(AnnounceEvent::Completed, NumberOfBytes(0))); - assert_eq!(Leeching, f(AnnounceEvent::Completed, NumberOfBytes(1))); + assert_eq!(Seeding, f(AnnounceEvent::Completed, NumberOfBytes::new(0))); + assert_eq!(Leeching, f(AnnounceEvent::Completed, NumberOfBytes::new(1))); - assert_eq!(Seeding, f(AnnounceEvent::None, NumberOfBytes(0))); - assert_eq!(Leeching, f(AnnounceEvent::None, NumberOfBytes(1))); + assert_eq!(Seeding, f(AnnounceEvent::None, NumberOfBytes::new(0))); + assert_eq!(Leeching, f(AnnounceEvent::None, NumberOfBytes::new(1))); } // Assumes that announce response with maximum amount of ipv6 peers will @@ -229,17 +228,19 @@ mod tests { let config = Config::default(); let peers = ::std::iter::repeat(ResponsePeer { - ip_address: Ipv6Addr::new(1, 1, 1, 1, 1, 1, 1, 1), - port: Port(1), + ip_address: Ipv6AddrBytes(Ipv6Addr::new(1, 1, 1, 1, 1, 1, 1, 1).octets()), + port: Port::new(1), }) .take(config.protocol.max_response_peers) .collect(); let response = Response::AnnounceIpv6(AnnounceResponse { - transaction_id: TransactionId(1), - announce_interval: AnnounceInterval(1), - seeders: NumberOfPeers(1), - leechers: NumberOfPeers(1), + fixed: AnnounceResponseFixedData { + transaction_id: TransactionId::new(1), + announce_interval: AnnounceInterval::new(1), + seeders: NumberOfPeers::new(1), + leechers: NumberOfPeers::new(1), + }, peers, }); diff --git a/crates/udp/src/workers/socket/mio.rs b/crates/udp/src/workers/socket/mio.rs index c248b3c..1d254f3 100644 --- a/crates/udp/src/workers/socket/mio.rs +++ b/crates/udp/src/workers/socket/mio.rs @@ -195,6 +195,8 @@ impl SocketWorker { let src = CanonicalSocketAddr::new(src); + ::log::trace!("received request bytes: {}", hex_slice(&self.buffer[..bytes_read])); + let request_parsable = match Request::from_bytes( &self.buffer[..bytes_read], self.config.protocol.max_scrape_torrents, @@ -221,7 +223,7 @@ impl SocketWorker { if self.validator.connection_id_valid(src, connection_id) { let response = ErrorResponse { transaction_id, - message: err.right_or("Parse error").into(), + message: err.into(), }; local_responses.push((response.into(), src)); @@ -285,6 +287,8 @@ impl SocketWorker { match request { Request::Connect(request) => { + ::log::trace!("received {:?} from {:?}", request, src); + let connection_id = self.validator.create_connection_id(src); let response = Response::Connect(ConnectResponse { @@ -295,6 +299,8 @@ impl SocketWorker { local_responses.push((response, src)) } Request::Announce(request) => { + ::log::trace!("received {:?} from {:?}", request, src); + if self .validator .connection_id_valid(src, request.connection_id) @@ -323,6 +329,8 @@ impl SocketWorker { } } Request::Scrape(request) => { + ::log::trace!("received {:?} from {:?}", request, src); + if self .validator .connection_id_valid(src, request.connection_id) @@ -372,6 +380,8 @@ impl SocketWorker { canonical_addr.get_ipv6_mapped() }; + ::log::trace!("sending {:?} to {}, bytes: {}", response, addr, hex_slice(&cursor.get_ref()[..bytes_written])); + match socket.send_to(&cursor.get_ref()[..bytes_written], addr) { Ok(amt) if config.statistics.active() => { let stats = if canonical_addr.is_ipv4() { @@ -430,3 +440,14 @@ impl SocketWorker { } } } + +fn hex_slice(bytes: &[u8]) -> String { + let mut output = String::with_capacity(bytes.len() * 3); + + for chunk in bytes.chunks(4) { + output.push_str(&hex::encode(chunk)); + output.push(' '); + } + + output +} \ No newline at end of file diff --git a/crates/udp/src/workers/socket/storage.rs b/crates/udp/src/workers/socket/storage.rs index 8aed47c..b9e0952 100644 --- a/crates/udp/src/workers/socket/storage.rs +++ b/crates/udp/src/workers/socket/storage.rs @@ -156,8 +156,8 @@ mod tests { } let request = ScrapeRequest { - transaction_id: TransactionId(t), - connection_id: ConnectionId(c), + transaction_id: TransactionId::new(t), + connection_id: ConnectionId::new(c), info_hashes, }; @@ -192,9 +192,9 @@ mod tests { ( i, TorrentScrapeStatistics { - seeders: NumberOfPeers((info_hash.0[0]) as i32), - leechers: NumberOfPeers(0), - completed: NumberOfDownloads(0), + seeders: NumberOfPeers::new((info_hash.0[0]) as i32), + leechers: NumberOfPeers::new(0), + completed: NumberOfDownloads::new(0), }, ) }) diff --git a/crates/udp/src/workers/socket/uring/mod.rs b/crates/udp/src/workers/socket/uring/mod.rs index 4a9b54e..c9b7a21 100644 --- a/crates/udp/src/workers/socket/uring/mod.rs +++ b/crates/udp/src/workers/socket/uring/mod.rs @@ -393,7 +393,7 @@ impl SocketWorker { if self.validator.connection_id_valid(addr, connection_id) { let response = ErrorResponse { transaction_id, - message: err.right_or("Parse error").into(), + message: err.into(), }; self.local_responses.push_back((response.into(), addr)); diff --git a/crates/udp/src/workers/socket/validator.rs b/crates/udp/src/workers/socket/validator.rs index 74e2f76..9cd25fb 100644 --- a/crates/udp/src/workers/socket/validator.rs +++ b/crates/udp/src/workers/socket/validator.rs @@ -59,7 +59,7 @@ impl ConnectionValidator { (&mut connection_id_bytes[..4]).copy_from_slice(&valid_until); (&mut connection_id_bytes[4..]).copy_from_slice(&hash); - ConnectionId(i64::from_ne_bytes(connection_id_bytes)) + ConnectionId::new(i64::from_ne_bytes(connection_id_bytes)) } pub fn connection_id_valid( @@ -67,7 +67,7 @@ impl ConnectionValidator { source_addr: CanonicalSocketAddr, connection_id: ConnectionId, ) -> bool { - let bytes = connection_id.0.to_ne_bytes(); + let bytes = connection_id.0.get().to_ne_bytes(); let (valid_until, hash) = bytes.split_at(4); let valid_until: [u8; 4] = valid_until.try_into().unwrap(); diff --git a/crates/udp/src/workers/swarm/mod.rs b/crates/udp/src/workers/swarm/mod.rs index d4bb3c5..6284d42 100644 --- a/crates/udp/src/workers/swarm/mod.rs +++ b/crates/udp/src/workers/swarm/mod.rs @@ -53,7 +53,7 @@ pub fn run_swarm_worker( &statistics_sender, &mut torrents.ipv4, request, - ip, + ip.into(), peer_valid_until, ); @@ -66,7 +66,7 @@ pub fn run_swarm_worker( &statistics_sender, &mut torrents.ipv6, request, - ip, + ip.into(), peer_valid_until, ); @@ -126,18 +126,19 @@ fn handle_announce_request( peer_ip: I, peer_valid_until: ValidUntil, ) -> AnnounceResponse { - let max_num_peers_to_take: usize = if request.peers_wanted.0 <= 0 { + let max_num_peers_to_take: usize = if request.peers_wanted.0.get() <= 0 { config.protocol.max_response_peers } else { ::std::cmp::min( config.protocol.max_response_peers, - request.peers_wanted.0.try_into().unwrap(), + request.peers_wanted.0.get().try_into().unwrap(), ) }; let torrent_data = torrents.0.entry(request.info_hash).or_default(); - let peer_status = PeerStatus::from_event_and_bytes_left(request.event, request.bytes_left); + let peer_status = + PeerStatus::from_event_and_bytes_left(request.event.into(), request.bytes_left); torrent_data.update_peer( config, @@ -156,10 +157,14 @@ fn handle_announce_request( }; AnnounceResponse { - transaction_id: request.transaction_id, - announce_interval: AnnounceInterval(config.protocol.peer_announce_interval), - leechers: NumberOfPeers(torrent_data.num_leechers().try_into().unwrap_or(i32::MAX)), - seeders: NumberOfPeers(torrent_data.num_seeders().try_into().unwrap_or(i32::MAX)), + fixed: AnnounceResponseFixedData { + transaction_id: request.transaction_id, + announce_interval: AnnounceInterval::new(config.protocol.peer_announce_interval), + leechers: NumberOfPeers::new( + torrent_data.num_leechers().try_into().unwrap_or(i32::MAX), + ), + seeders: NumberOfPeers::new(torrent_data.num_seeders().try_into().unwrap_or(i32::MAX)), + }, peers: response_peers, } } @@ -168,8 +173,6 @@ fn handle_scrape_request( torrents: &mut TorrentMap, request: PendingScrapeRequest, ) -> PendingScrapeResponse { - const EMPTY_STATS: TorrentScrapeStatistics = create_torrent_scrape_statistics(0, 0); - let torrent_stats = request .info_hashes .into_iter() @@ -178,7 +181,7 @@ fn handle_scrape_request( .0 .get(&info_hash) .map(|torrent_data| torrent_data.scrape_statistics()) - .unwrap_or(EMPTY_STATS); + .unwrap_or_else(|| create_torrent_scrape_statistics(0, 0)); (i, stats) }) @@ -191,10 +194,10 @@ fn handle_scrape_request( } #[inline(always)] -const fn create_torrent_scrape_statistics(seeders: i32, leechers: i32) -> TorrentScrapeStatistics { +fn create_torrent_scrape_statistics(seeders: i32, leechers: i32) -> TorrentScrapeStatistics { TorrentScrapeStatistics { - seeders: NumberOfPeers(seeders), - completed: NumberOfDownloads(0), // No implementation planned - leechers: NumberOfPeers(leechers), + seeders: NumberOfPeers::new(seeders), + completed: NumberOfDownloads::new(0), // No implementation planned + leechers: NumberOfPeers::new(leechers), } } diff --git a/crates/udp/src/workers/swarm/storage.rs b/crates/udp/src/workers/swarm/storage.rs index 567cc80..676ad91 100644 --- a/crates/udp/src/workers/swarm/storage.rs +++ b/crates/udp/src/workers/swarm/storage.rs @@ -1,5 +1,3 @@ -use std::net::Ipv4Addr; -use std::net::Ipv6Addr; use std::sync::atomic::Ordering; use std::sync::Arc; @@ -256,8 +254,8 @@ impl TorrentMap { } pub struct TorrentMaps { - pub ipv4: TorrentMap, - pub ipv6: TorrentMap, + pub ipv4: TorrentMap, + pub ipv6: TorrentMap, } impl Default for TorrentMaps { @@ -312,7 +310,6 @@ impl TorrentMaps { #[cfg(test)] mod tests { use std::collections::HashSet; - use std::net::Ipv4Addr; use quickcheck::{quickcheck, TestResult}; use rand::thread_rng; @@ -326,10 +323,10 @@ mod tests { peer_id } - fn gen_peer(i: u32) -> Peer { + fn gen_peer(i: u32) -> Peer { Peer { - ip_address: Ipv4Addr::from(i.to_be_bytes()), - port: Port(1), + ip_address: Ipv4AddrBytes(i.to_be_bytes()), + port: Port::new(1), is_seeder: false, valid_until: ValidUntil::new(ServerStartInstant::new(), 0), } @@ -341,7 +338,7 @@ mod tests { let gen_num_peers = data.0 as u32; let req_num_peers = data.1 as usize; - let mut peer_map: PeerMap = Default::default(); + let mut peer_map: PeerMap = Default::default(); let mut opt_sender_key = None; let mut opt_sender_peer = None; diff --git a/crates/udp/tests/common/mod.rs b/crates/udp/tests/common/mod.rs index f981e56..832d027 100644 --- a/crates/udp/tests/common/mod.rs +++ b/crates/udp/tests/common/mod.rs @@ -10,8 +10,8 @@ use anyhow::Context; use aquatic_udp::{common::BUFFER_SIZE, config::Config}; use aquatic_udp_protocol::{ common::PeerId, AnnounceEvent, AnnounceRequest, ConnectRequest, ConnectionId, InfoHash, - NumberOfBytes, NumberOfPeers, PeerKey, Port, Request, Response, ScrapeRequest, ScrapeResponse, - TransactionId, + Ipv4AddrBytes, NumberOfBytes, NumberOfPeers, PeerKey, Port, Request, Response, ScrapeRequest, + ScrapeResponse, TransactionId, }; // FIXME: should ideally try different ports and use sync primitives to find @@ -26,7 +26,7 @@ pub fn run_tracker(config: Config) { pub fn connect(socket: &UdpSocket, tracker_addr: SocketAddr) -> anyhow::Result { let request = Request::Connect(ConnectRequest { - transaction_id: TransactionId(0), + transaction_id: TransactionId::new(0), }); let response = request_and_response(&socket, tracker_addr, request)?; @@ -55,17 +55,18 @@ pub fn announce( let request = Request::Announce(AnnounceRequest { connection_id, - transaction_id: TransactionId(0), + action_placeholder: Default::default(), + transaction_id: TransactionId::new(0), info_hash, peer_id, - bytes_downloaded: NumberOfBytes(0), - bytes_uploaded: NumberOfBytes(0), - bytes_left: NumberOfBytes(if seeder { 0 } else { 1 }), - event: AnnounceEvent::Started, - ip_address: None, - key: PeerKey(0), - peers_wanted: NumberOfPeers(peers_wanted as i32), - port: Port(peer_port), + bytes_downloaded: NumberOfBytes::new(0), + bytes_uploaded: NumberOfBytes::new(0), + bytes_left: NumberOfBytes::new(if seeder { 0 } else { 1 }), + event: AnnounceEvent::Started.into(), + ip_address: Ipv4AddrBytes([0; 4]), + key: PeerKey::new(0), + peers_wanted: NumberOfPeers::new(peers_wanted as i32), + port: Port::new(peer_port), }); Ok(request_and_response(&socket, tracker_addr, request)?) @@ -79,7 +80,7 @@ pub fn scrape( ) -> anyhow::Result { let request = Request::Scrape(ScrapeRequest { connection_id, - transaction_id: TransactionId(0), + transaction_id: TransactionId::new(0), info_hashes, }); diff --git a/crates/udp/tests/invalid_connection_id.rs b/crates/udp/tests/invalid_connection_id.rs index ce4b049..11579a3 100644 --- a/crates/udp/tests/invalid_connection_id.rs +++ b/crates/udp/tests/invalid_connection_id.rs @@ -11,8 +11,8 @@ use std::{ use anyhow::Context; use aquatic_udp::{common::BUFFER_SIZE, config::Config}; use aquatic_udp_protocol::{ - common::PeerId, AnnounceEvent, AnnounceRequest, ConnectionId, InfoHash, NumberOfBytes, - NumberOfPeers, PeerKey, Port, Request, ScrapeRequest, TransactionId, + common::PeerId, AnnounceEvent, AnnounceRequest, ConnectionId, InfoHash, Ipv4AddrBytes, + NumberOfBytes, NumberOfPeers, PeerKey, Port, Request, ScrapeRequest, TransactionId, }; #[test] @@ -40,22 +40,23 @@ fn test_invalid_connection_id() -> anyhow::Result<()> { let announce_request = Request::Announce(AnnounceRequest { connection_id: invalid_connection_id, - transaction_id: TransactionId(0), + action_placeholder: Default::default(), + transaction_id: TransactionId::new(0), info_hash: InfoHash([0; 20]), peer_id: PeerId([0; 20]), - bytes_downloaded: NumberOfBytes(0), - bytes_uploaded: NumberOfBytes(0), - bytes_left: NumberOfBytes(0), - event: AnnounceEvent::Started, - ip_address: None, - key: PeerKey(0), - peers_wanted: NumberOfPeers(10), - port: Port(1), + bytes_downloaded: NumberOfBytes::new(0), + bytes_uploaded: NumberOfBytes::new(0), + bytes_left: NumberOfBytes::new(0), + event: AnnounceEvent::Started.into(), + ip_address: Ipv4AddrBytes([0; 4]), + key: PeerKey::new(0), + peers_wanted: NumberOfPeers::new(10), + port: Port::new(1), }); let scrape_request = Request::Scrape(ScrapeRequest { connection_id: invalid_connection_id, - transaction_id: TransactionId(0), + transaction_id: TransactionId::new(0), info_hashes: vec![InfoHash([0; 20])], }); diff --git a/crates/udp/tests/requests_responses.rs b/crates/udp/tests/requests_responses.rs index c84f0f8..ee7abc2 100644 --- a/crates/udp/tests/requests_responses.rs +++ b/crates/udp/tests/requests_responses.rs @@ -67,11 +67,11 @@ fn test_multiple_connect_announce_scrape() -> anyhow::Result<()> { assert_eq!(announce_response.peers.len(), i.min(PEERS_WANTED)); - assert_eq!(announce_response.seeders.0, num_seeders); - assert_eq!(announce_response.leechers.0, num_leechers); + assert_eq!(announce_response.fixed.seeders.0.get(), num_seeders); + assert_eq!(announce_response.fixed.leechers.0.get(), num_leechers); let response_peer_ports: HashSet = - HashSet::from_iter(announce_response.peers.iter().map(|p| p.port.0)); + HashSet::from_iter(announce_response.peers.iter().map(|p| p.port.0.get())); let expected_peer_ports: HashSet = HashSet::from_iter((0..i).map(|i| PEER_PORT_START + i as u16)); @@ -89,10 +89,16 @@ fn test_multiple_connect_announce_scrape() -> anyhow::Result<()> { ) .with_context(|| "scrape")?; - assert_eq!(scrape_response.torrent_stats[0].seeders.0, num_seeders); - assert_eq!(scrape_response.torrent_stats[0].leechers.0, num_leechers); - assert_eq!(scrape_response.torrent_stats[1].seeders.0, 0); - assert_eq!(scrape_response.torrent_stats[1].leechers.0, 0); + assert_eq!( + scrape_response.torrent_stats[0].seeders.0.get(), + num_seeders + ); + assert_eq!( + scrape_response.torrent_stats[0].leechers.0.get(), + num_leechers + ); + assert_eq!(scrape_response.torrent_stats[1].seeders.0.get(), 0); + assert_eq!(scrape_response.torrent_stats[1].leechers.0.get(), 0); } Ok(()) diff --git a/crates/udp_bench/src/announce.rs b/crates/udp_bench/src/announce.rs index 9302764..3b94f14 100644 --- a/crates/udp_bench/src/announce.rs +++ b/crates/udp_bench/src/announce.rs @@ -49,7 +49,7 @@ pub fn bench_announce_handler( num_responses += 1; if let Some(last_peer) = r.peers.last() { - dummy ^= last_peer.port.0; + dummy ^= last_peer.port.0.get(); } } } @@ -61,7 +61,7 @@ pub fn bench_announce_handler( num_responses += 1; if let Some(last_peer) = r.peers.last() { - dummy ^= last_peer.port.0; + dummy ^= last_peer.port.0.get(); } } } @@ -91,18 +91,19 @@ pub fn create_requests( let info_hash_index = gamma_usize(rng, gamma, max_index); let request = AnnounceRequest { - connection_id: ConnectionId(0), - transaction_id: TransactionId(rng.gen()), + connection_id: ConnectionId::new(0), + action_placeholder: Default::default(), + transaction_id: TransactionId::new(rng.gen()), info_hash: info_hashes[info_hash_index], peer_id: PeerId(rng.gen()), - bytes_downloaded: NumberOfBytes(rng.gen()), - bytes_uploaded: NumberOfBytes(rng.gen()), - bytes_left: NumberOfBytes(rng.gen()), - event: AnnounceEvent::Started, - ip_address: None, - key: PeerKey(rng.gen()), - peers_wanted: NumberOfPeers(rng.gen()), - port: Port(rng.gen()), + bytes_downloaded: NumberOfBytes::new(rng.gen()), + bytes_uploaded: NumberOfBytes::new(rng.gen()), + bytes_left: NumberOfBytes::new(rng.gen()), + event: AnnounceEvent::Started.into(), + ip_address: Ipv4AddrBytes([0; 4]), + key: PeerKey::new(rng.gen()), + peers_wanted: NumberOfPeers::new(rng.gen()), + port: Port::new(rng.gen()), }; requests.push(( diff --git a/crates/udp_bench/src/scrape.rs b/crates/udp_bench/src/scrape.rs index 0a1229f..c110086 100644 --- a/crates/udp_bench/src/scrape.rs +++ b/crates/udp_bench/src/scrape.rs @@ -60,7 +60,7 @@ pub fn bench_scrape_handler( num_responses += 1; if let Some(stat) = response.torrent_stats.values().last() { - dummy ^= stat.leechers.0; + dummy ^= stat.leechers.0.get(); } } } @@ -72,7 +72,7 @@ pub fn bench_scrape_handler( num_responses += 1; if let Some(stat) = response.torrent_stats.values().last() { - dummy ^= stat.leechers.0; + dummy ^= stat.leechers.0.get(); } } } @@ -108,8 +108,8 @@ pub fn create_requests( } let request = ScrapeRequest { - connection_id: ConnectionId(0), - transaction_id: TransactionId(rng.gen()), + connection_id: ConnectionId::new(0), + transaction_id: TransactionId::new(rng.gen()), info_hashes: request_info_hashes, }; diff --git a/crates/udp_load_test/src/utils.rs b/crates/udp_load_test/src/utils.rs index fd32443..6b7f748 100644 --- a/crates/udp_load_test/src/utils.rs +++ b/crates/udp_load_test/src/utils.rs @@ -19,7 +19,7 @@ pub fn generate_info_hash() -> InfoHash { } pub fn generate_transaction_id(rng: &mut impl Rng) -> TransactionId { - TransactionId(rng.gen()) + TransactionId::new(rng.gen()) } pub fn create_connect_request(transaction_id: TransactionId) -> Request { diff --git a/crates/udp_load_test/src/worker/request_gen.rs b/crates/udp_load_test/src/worker/request_gen.rs index 8f504fd..a918341 100644 --- a/crates/udp_load_test/src/worker/request_gen.rs +++ b/crates/udp_load_test/src/worker/request_gen.rs @@ -49,14 +49,14 @@ pub fn process_response( rng, info_hashes, torrent_peers, - r.transaction_id, + r.fixed.transaction_id, ), Response::AnnounceIpv6(r) => if_torrent_peer_move_and_create_random_request( config, rng, info_hashes, torrent_peers, - r.transaction_id, + r.fixed.transaction_id, ), Response::Scrape(r) => if_torrent_peer_move_and_create_random_request( config, @@ -143,24 +143,25 @@ fn create_announce_request( ) -> Request { let (event, bytes_left) = { if rng.gen_bool(config.requests.peer_seeder_probability) { - (AnnounceEvent::Completed, NumberOfBytes(0)) + (AnnounceEvent::Completed, NumberOfBytes::new(0)) } else { - (AnnounceEvent::Started, NumberOfBytes(50)) + (AnnounceEvent::Started, NumberOfBytes::new(50)) } }; (AnnounceRequest { connection_id: torrent_peer.connection_id, + action_placeholder: Default::default(), transaction_id, info_hash: torrent_peer.info_hash, peer_id: torrent_peer.peer_id, - bytes_downloaded: NumberOfBytes(50), - bytes_uploaded: NumberOfBytes(50), + bytes_downloaded: NumberOfBytes::new(50), + bytes_uploaded: NumberOfBytes::new(50), bytes_left, - event, - ip_address: None, - key: PeerKey(12345), - peers_wanted: NumberOfPeers(100), + event: event.into(), + ip_address: Ipv4AddrBytes([0; 4]), + key: PeerKey::new(12345), + peers_wanted: NumberOfPeers::new(100), port: torrent_peer.port, }) .into() @@ -209,7 +210,7 @@ fn create_torrent_peer( scrape_hash_indeces, connection_id, peer_id: generate_peer_id(), - port: Port(rng.gen()), + port: Port::new(rng.gen()), } } diff --git a/crates/udp_protocol/Cargo.toml b/crates/udp_protocol/Cargo.toml index 033b65f..cd4614d 100644 --- a/crates/udp_protocol/Cargo.toml +++ b/crates/udp_protocol/Cargo.toml @@ -15,7 +15,9 @@ aquatic_peer_id.workspace = true byteorder = "1" either = "1" +zerocopy = { version = "0.7", features = ["derive"] } [dev-dependencies] +pretty_assertions = "1" quickcheck = "1" quickcheck_macros = "1" diff --git a/crates/udp_protocol/src/common.rs b/crates/udp_protocol/src/common.rs index c99a0cb..829cd55 100644 --- a/crates/udp_protocol/src/common.rs +++ b/crates/udp_protocol/src/common.rs @@ -2,45 +2,174 @@ use std::fmt::Debug; use std::net::{Ipv4Addr, Ipv6Addr}; pub use aquatic_peer_id::{PeerClient, PeerId}; +use zerocopy::network_endian::{I32, I64, U16, U32}; +use zerocopy::{AsBytes, FromBytes, FromZeroes}; -pub trait Ip: Clone + Copy + Debug + PartialEq + Eq {} +pub trait Ip: Clone + Copy + Debug + PartialEq + Eq + AsBytes {} -impl Ip for Ipv4Addr {} -impl Ip for Ipv6Addr {} +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct AnnounceInterval(pub I32); -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct AnnounceInterval(pub i32); +impl AnnounceInterval { + pub fn new(v: i32) -> Self { + Self(I32::new(v)) + } +} -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] pub struct InfoHash(pub [u8; 20]); -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct ConnectionId(pub i64); +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct ConnectionId(pub I64); -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct TransactionId(pub i32); +impl ConnectionId { + pub fn new(v: i64) -> Self { + Self(I64::new(v)) + } +} -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct NumberOfBytes(pub i64); +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct TransactionId(pub I32); -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct NumberOfPeers(pub i32); +impl TransactionId { + pub fn new(v: i32) -> Self { + Self(I32::new(v)) + } +} -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct NumberOfDownloads(pub i32); +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct NumberOfBytes(pub I64); -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct Port(pub u16); +impl NumberOfBytes { + pub fn new(v: i64) -> Self { + Self(I64::new(v)) + } +} -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub struct PeerKey(pub u32); +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct NumberOfPeers(pub I32); -#[derive(PartialEq, Eq, Clone, Debug)] +impl NumberOfPeers { + pub fn new(v: i32) -> Self { + Self(I32::new(v)) + } +} + +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct NumberOfDownloads(pub I32); + +impl NumberOfDownloads { + pub fn new(v: i32) -> Self { + Self(I32::new(v)) + } +} + +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct Port(pub U16); + +impl Port { + pub fn new(v: u16) -> Self { + Self(U16::new(v)) + } +} + +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct PeerKey(pub I32); + +impl PeerKey { + pub fn new(v: i32) -> Self { + Self(I32::new(v)) + } +} + +#[derive(PartialEq, Eq, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(C, packed)] pub struct ResponsePeer { pub ip_address: I, pub port: Port, } +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct Ipv4AddrBytes(pub [u8; 4]); + +impl Ip for Ipv4AddrBytes {} + +impl Into for Ipv4AddrBytes { + fn into(self) -> Ipv4Addr { + Ipv4Addr::from(self.0) + } +} + +impl Into for Ipv4Addr { + fn into(self) -> Ipv4AddrBytes { + Ipv4AddrBytes(self.octets()) + } +} + +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct Ipv6AddrBytes(pub [u8; 16]); + +impl Ip for Ipv6AddrBytes {} + +impl Into for Ipv6AddrBytes { + fn into(self) -> Ipv6Addr { + Ipv6Addr::from(self.0) + } +} + +impl Into for Ipv6Addr { + fn into(self) -> Ipv6AddrBytes { + Ipv6AddrBytes(self.octets()) + } +} + +pub fn read_i32_ne(bytes: &mut impl ::std::io::Read) -> ::std::io::Result { + let mut tmp = [0u8; 4]; + + bytes.read_exact(&mut tmp)?; + + Ok(I32::from_bytes(tmp)) +} + +pub fn read_i64_ne(bytes: &mut impl ::std::io::Read) -> ::std::io::Result { + let mut tmp = [0u8; 8]; + + bytes.read_exact(&mut tmp)?; + + Ok(I64::from_bytes(tmp)) +} + +pub fn read_u16_ne(bytes: &mut impl ::std::io::Read) -> ::std::io::Result { + let mut tmp = [0u8; 2]; + + bytes.read_exact(&mut tmp)?; + + Ok(U16::from_bytes(tmp)) +} + +pub fn read_u32_ne(bytes: &mut impl ::std::io::Read) -> ::std::io::Result { + let mut tmp = [0u8; 4]; + + bytes.read_exact(&mut tmp)?; + + Ok(U32::from_bytes(tmp)) +} + +pub fn invalid_data() -> ::std::io::Error { + ::std::io::Error::new(::std::io::ErrorKind::InvalidData, "invalid data") +} + #[cfg(test)] impl quickcheck::Arbitrary for InfoHash { fn arbitrary(g: &mut quickcheck::Gen) -> Self { diff --git a/crates/udp_protocol/src/request.rs b/crates/udp_protocol/src/request.rs index 0ed591c..cd2963e 100644 --- a/crates/udp_protocol/src/request.rs +++ b/crates/udp_protocol/src/request.rs @@ -1,9 +1,9 @@ -use std::convert::TryInto; -use std::io::{self, Cursor, Read, Write}; -use std::net::Ipv4Addr; +use std::io::{self, Cursor, Write}; -use byteorder::{NetworkEndian, ReadBytesExt, WriteBytesExt}; +use byteorder::{NetworkEndian, WriteBytesExt}; use either::Either; +use zerocopy::FromZeroes; +use zerocopy::{byteorder::network_endian::I32, AsBytes, FromBytes}; use aquatic_peer_id::PeerId; @@ -11,103 +11,6 @@ use super::common::*; const PROTOCOL_IDENTIFIER: i64 = 4_497_486_125_440; -#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] -pub enum AnnounceEvent { - Started, - Stopped, - Completed, - None, -} - -impl AnnounceEvent { - #[inline] - pub fn from_i32(i: i32) -> Self { - match i { - 1 => Self::Completed, - 2 => Self::Started, - 3 => Self::Stopped, - _ => Self::None, - } - } - - #[inline] - pub fn to_i32(&self) -> i32 { - match self { - AnnounceEvent::None => 0, - AnnounceEvent::Completed => 1, - AnnounceEvent::Started => 2, - AnnounceEvent::Stopped => 3, - } - } -} - -#[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, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct ScrapeRequest { - pub connection_id: ConnectionId, - pub transaction_id: TransactionId, - pub info_hashes: Vec, -} - -#[derive(Debug)] -pub enum RequestParseError { - Sendable { - connection_id: ConnectionId, - transaction_id: TransactionId, - err: Either, - }, - Unsendable { - err: Either, - }, -} - -impl RequestParseError { - pub fn sendable_io(err: io::Error, connection_id: i64, transaction_id: i32) -> Self { - Self::Sendable { - connection_id: ConnectionId(connection_id), - transaction_id: TransactionId(transaction_id), - err: Either::Left(err), - } - } - pub fn sendable_text(text: &'static str, connection_id: i64, transaction_id: i32) -> Self { - Self::Sendable { - connection_id: ConnectionId(connection_id), - transaction_id: TransactionId(transaction_id), - err: Either::Right(text), - } - } - pub fn unsendable_io(err: io::Error) -> Self { - Self::Unsendable { - err: Either::Left(err), - } - } - pub fn unsendable_text(text: &'static str) -> Self { - Self::Unsendable { - err: Either::Right(text), - } - } -} - #[derive(PartialEq, Eq, Clone, Debug)] pub enum Request { Connect(ConnectRequest), @@ -115,6 +18,115 @@ pub enum Request { Scrape(ScrapeRequest), } +impl Request { + pub fn write(self, bytes: &mut impl Write) -> Result<(), io::Error> { + match self { + Request::Connect(r) => { + bytes.write_i64::(PROTOCOL_IDENTIFIER)?; + bytes.write_i32::(0)?; + bytes.write_all(r.transaction_id.as_bytes())?; + } + + Request::Announce(r) => { + bytes.write_all(r.as_bytes())?; + } + + Request::Scrape(r) => { + bytes.write_all(r.connection_id.as_bytes())?; + bytes.write_i32::(2)?; + bytes.write_all(r.transaction_id.as_bytes())?; + bytes.write_all((*r.info_hashes.as_slice()).as_bytes())?; + } + } + + Ok(()) + } + + pub fn from_bytes(bytes: &[u8], max_scrape_torrents: u8) -> Result { + let action = bytes + .get(8..12) + .map(|bytes| I32::from_bytes(bytes.try_into().unwrap())) + .ok_or_else(|| RequestParseError::unsendable_text("Couldn't parse action"))?; + + match action.get() { + // Connect + 0 => { + let mut bytes = Cursor::new(bytes); + + let protocol_identifier = + read_i64_ne(&mut bytes).map_err(RequestParseError::unsendable_io)?; + let _action = read_i32_ne(&mut bytes).map_err(RequestParseError::unsendable_io)?; + let transaction_id = read_i32_ne(&mut bytes) + .map(TransactionId) + .map_err(RequestParseError::unsendable_io)?; + + if protocol_identifier.get() == PROTOCOL_IDENTIFIER { + Ok((ConnectRequest { transaction_id }).into()) + } else { + Err(RequestParseError::unsendable_text( + "Protocol identifier missing", + )) + } + } + // Announce + 1 => { + let request = AnnounceRequest::read_from_prefix(bytes) + .ok_or_else(|| RequestParseError::unsendable_text("invalid data"))?; + + // Make sure not to create AnnounceEventBytes with invalid value + if matches!(request.event.0.get(), (0..=3)) { + Ok(Request::Announce(request)) + } else { + Err(RequestParseError::sendable_text( + "Invalid announce event", + request.connection_id, + request.transaction_id, + )) + } + } + // Scrape + 2 => { + let mut bytes = Cursor::new(bytes); + + let connection_id = read_i64_ne(&mut bytes) + .map(ConnectionId) + .map_err(RequestParseError::unsendable_io)?; + let _action = read_i32_ne(&mut bytes).map_err(RequestParseError::unsendable_io)?; + let transaction_id = read_i32_ne(&mut bytes) + .map(TransactionId) + .map_err(RequestParseError::unsendable_io)?; + + let remaining_bytes = { + let position = bytes.position() as usize; + let inner = bytes.into_inner(); + &inner[position..] + }; + + let info_hashes = FromBytes::slice_from(remaining_bytes).ok_or_else(|| { + RequestParseError::sendable_text( + "Invalid info hash list. Note that full scrapes are not allowed", + connection_id, + transaction_id, + ) + })?; + + let info_hashes = Vec::from( + &info_hashes[..(max_scrape_torrents as usize).min(info_hashes.len())], + ); + + Ok((ScrapeRequest { + connection_id, + transaction_id, + info_hashes, + }) + .into()) + } + + _ => Err(RequestParseError::unsendable_text("Invalid action")), + } + } +} + impl From for Request { fn from(r: ConnectRequest) -> Self { Self::Connect(r) @@ -133,173 +145,116 @@ impl From for Request { } } -impl Request { - pub fn write(self, bytes: &mut impl Write) -> Result<(), io::Error> { - match self { - Request::Connect(r) => { - bytes.write_i64::(PROTOCOL_IDENTIFIER)?; - bytes.write_i32::(0)?; - bytes.write_i32::(r.transaction_id.0)?; - } +#[derive(PartialEq, Eq, Clone, Debug)] +pub struct ConnectRequest { + pub transaction_id: TransactionId, +} - Request::Announce(r) => { - bytes.write_i64::(r.connection_id.0)?; - bytes.write_i32::(1)?; - bytes.write_i32::(r.transaction_id.0)?; +#[derive(PartialEq, Eq, Clone, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(C, packed)] +pub struct AnnounceRequest { + pub connection_id: ConnectionId, + /// This field is only present to enable zero-copy serialization and + /// deserialization. + pub action_placeholder: AnnounceActionPlaceholder, + pub transaction_id: TransactionId, + pub info_hash: InfoHash, + pub peer_id: PeerId, + pub bytes_downloaded: NumberOfBytes, + pub bytes_left: NumberOfBytes, + pub bytes_uploaded: NumberOfBytes, + pub event: AnnounceEventBytes, + pub ip_address: Ipv4AddrBytes, + pub key: PeerKey, + pub peers_wanted: NumberOfPeers, + pub port: Port, +} - bytes.write_all(&r.info_hash.0)?; - bytes.write_all(&r.peer_id.0)?; +/// Note: Request::from_bytes only creates this struct with value 1 +#[derive(PartialEq, Eq, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct AnnounceActionPlaceholder(I32); - bytes.write_i64::(r.bytes_downloaded.0)?; - bytes.write_i64::(r.bytes_left.0)?; - bytes.write_i64::(r.bytes_uploaded.0)?; - - bytes.write_i32::(r.event.to_i32())?; - - bytes.write_all(&r.ip_address.map_or([0; 4], |ip| ip.octets()))?; - - bytes.write_u32::(r.key.0)?; - bytes.write_i32::(r.peers_wanted.0)?; - bytes.write_u16::(r.port.0)?; - } - - Request::Scrape(r) => { - bytes.write_i64::(r.connection_id.0)?; - bytes.write_i32::(2)?; - bytes.write_i32::(r.transaction_id.0)?; - - for info_hash in r.info_hashes { - bytes.write_all(&info_hash.0)?; - } - } - } - - Ok(()) +impl Default for AnnounceActionPlaceholder { + fn default() -> Self { + Self(I32::new(1)) } +} - pub fn from_bytes(bytes: &[u8], max_scrape_torrents: u8) -> Result { - let mut cursor = Cursor::new(bytes); +/// Note: Request::from_bytes only creates this struct with values 0..=3 +#[derive(PartialEq, Eq, Clone, Copy, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(transparent)] +pub struct AnnounceEventBytes(I32); - let connection_id = cursor - .read_i64::() - .map_err(RequestParseError::unsendable_io)?; - let action = cursor - .read_i32::() - .map_err(RequestParseError::unsendable_io)?; - let transaction_id = cursor - .read_i32::() - .map_err(RequestParseError::unsendable_io)?; +impl From for AnnounceEventBytes { + fn from(value: AnnounceEvent) -> Self { + Self(I32::new(match value { + AnnounceEvent::None => 0, + AnnounceEvent::Completed => 1, + AnnounceEvent::Started => 2, + AnnounceEvent::Stopped => 3, + })) + } +} - match action { - // Connect - 0 => { - if connection_id == PROTOCOL_IDENTIFIER { - Ok((ConnectRequest { - transaction_id: TransactionId(transaction_id), - }) - .into()) - } else { - Err(RequestParseError::unsendable_text( - "Protocol identifier missing", - )) - } - } +#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)] +pub enum AnnounceEvent { + Started, + Stopped, + Completed, + None, +} - // Announce - 1 => { - let mut info_hash = [0; 20]; - let mut peer_id = [0; 20]; - let mut ip = [0; 4]; +impl From for AnnounceEvent { + fn from(value: AnnounceEventBytes) -> Self { + match value.0.get() { + 1 => Self::Completed, + 2 => Self::Started, + 3 => Self::Stopped, + _ => Self::None, + } + } +} - cursor.read_exact(&mut info_hash).map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; - cursor.read_exact(&mut peer_id).map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; +#[derive(PartialEq, Eq, Clone, Debug)] +pub struct ScrapeRequest { + pub connection_id: ConnectionId, + pub transaction_id: TransactionId, + pub info_hashes: Vec, +} - let bytes_downloaded = cursor.read_i64::().map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; - let bytes_left = cursor.read_i64::().map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; - let bytes_uploaded = cursor.read_i64::().map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; - let event = cursor.read_i32::().map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; +#[derive(Debug)] +pub enum RequestParseError { + Sendable { + connection_id: ConnectionId, + transaction_id: TransactionId, + err: &'static str, + }, + Unsendable { + err: Either, + }, +} - cursor.read_exact(&mut ip).map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; - - let key = cursor.read_u32::().map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; - let peers_wanted = cursor.read_i32::().map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; - let port = cursor.read_u16::().map_err(|err| { - RequestParseError::sendable_io(err, connection_id, transaction_id) - })?; - - 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), - }) - .into()) - } - - // Scrape - 2 => { - let position = cursor.position() as usize; - let inner = cursor.into_inner(); - - let info_hashes: Vec = (&inner[position..]) - .chunks_exact(20) - .take(max_scrape_torrents as usize) - .map(|chunk| InfoHash(chunk.try_into().unwrap())) - .collect(); - - if info_hashes.is_empty() { - Err(RequestParseError::sendable_text( - "Full scrapes are not allowed", - connection_id, - transaction_id, - )) - } else { - Ok((ScrapeRequest { - connection_id: ConnectionId(connection_id), - transaction_id: TransactionId(transaction_id), - info_hashes, - }) - .into()) - } - } - - _ => Err(RequestParseError::sendable_text( - "Invalid action", - connection_id, - transaction_id, - )), +impl RequestParseError { + pub fn sendable_text( + text: &'static str, + connection_id: ConnectionId, + transaction_id: TransactionId, + ) -> Self { + Self::Sendable { + connection_id, + transaction_id, + err: text, + } + } + pub fn unsendable_io(err: io::Error) -> Self { + Self::Unsendable { + err: Either::Left(err), + } + } + pub fn unsendable_text(text: &'static str) -> Self { + Self::Unsendable { + err: Either::Right(text), } } } @@ -308,6 +263,7 @@ impl Request { mod tests { use quickcheck::TestResult; use quickcheck_macros::quickcheck; + use zerocopy::network_endian::{I32, I64, U16}; use super::*; @@ -325,7 +281,7 @@ mod tests { impl quickcheck::Arbitrary for ConnectRequest { fn arbitrary(g: &mut quickcheck::Gen) -> Self { Self { - transaction_id: TransactionId(i32::arbitrary(g)), + transaction_id: TransactionId(I32::new(i32::arbitrary(g))), } } } @@ -333,18 +289,19 @@ mod tests { impl quickcheck::Arbitrary for AnnounceRequest { fn arbitrary(g: &mut quickcheck::Gen) -> Self { Self { - connection_id: ConnectionId(i64::arbitrary(g)), - transaction_id: TransactionId(i32::arbitrary(g)), + connection_id: ConnectionId(I64::new(i64::arbitrary(g))), + action_placeholder: AnnounceActionPlaceholder::default(), + transaction_id: TransactionId(I32::new(i32::arbitrary(g))), info_hash: InfoHash::arbitrary(g), peer_id: PeerId::arbitrary(g), - bytes_downloaded: NumberOfBytes(i64::arbitrary(g)), - bytes_uploaded: NumberOfBytes(i64::arbitrary(g)), - bytes_left: NumberOfBytes(i64::arbitrary(g)), - event: AnnounceEvent::arbitrary(g), - ip_address: None, - key: PeerKey(u32::arbitrary(g)), - peers_wanted: NumberOfPeers(i32::arbitrary(g)), - port: Port(u16::arbitrary(g)), + bytes_downloaded: NumberOfBytes(I64::new(i64::arbitrary(g))), + bytes_uploaded: NumberOfBytes(I64::new(i64::arbitrary(g))), + bytes_left: NumberOfBytes(I64::new(i64::arbitrary(g))), + event: AnnounceEvent::arbitrary(g).into(), + ip_address: Ipv4AddrBytes::arbitrary(g), + key: PeerKey::new(i32::arbitrary(g)), + peers_wanted: NumberOfPeers(I32::new(i32::arbitrary(g))), + port: Port(U16::new(u16::arbitrary(g))), } } } @@ -356,8 +313,8 @@ mod tests { .collect(); Self { - connection_id: ConnectionId(i64::arbitrary(g)), - transaction_id: TransactionId(i32::arbitrary(g)), + connection_id: ConnectionId(I64::new(i64::arbitrary(g))), + transaction_id: TransactionId(I32::new(i32::arbitrary(g))), info_hashes, } } @@ -372,7 +329,7 @@ mod tests { let success = request == r2; if !success { - println!("before: {:#?}\nafter: {:#?}", request, r2); + ::pretty_assertions::assert_eq!(request, r2); } success diff --git a/crates/udp_protocol/src/response.rs b/crates/udp_protocol/src/response.rs index 634724f..45d45b8 100644 --- a/crates/udp_protocol/src/response.rs +++ b/crates/udp_protocol/src/response.rs @@ -1,69 +1,138 @@ use std::borrow::Cow; -use std::convert::TryInto; -use std::io::{self, Cursor, Write}; -use std::net::{Ipv4Addr, Ipv6Addr}; +use std::io::{self, Write}; +use std::mem::size_of; -use byteorder::{NetworkEndian, ReadBytesExt, WriteBytesExt}; +use byteorder::{NetworkEndian, WriteBytesExt}; +use zerocopy::{AsBytes, FromBytes, FromZeroes}; use super::common::*; -#[derive(PartialEq, Eq, Debug, Copy, Clone)] -pub struct TorrentScrapeStatistics { - pub seeders: NumberOfPeers, - pub completed: NumberOfDownloads, - pub leechers: NumberOfPeers, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct ConnectResponse { - pub connection_id: ConnectionId, - pub transaction_id: TransactionId, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct AnnounceResponse { - pub transaction_id: TransactionId, - pub announce_interval: AnnounceInterval, - pub leechers: NumberOfPeers, - pub seeders: NumberOfPeers, - pub peers: Vec>, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct ScrapeResponse { - pub transaction_id: TransactionId, - pub torrent_stats: Vec, -} - -#[derive(PartialEq, Eq, Clone, Debug)] -pub struct ErrorResponse { - pub transaction_id: TransactionId, - pub message: Cow<'static, str>, -} - #[derive(PartialEq, Eq, Clone, Debug)] pub enum Response { Connect(ConnectResponse), - AnnounceIpv4(AnnounceResponse), - AnnounceIpv6(AnnounceResponse), + AnnounceIpv4(AnnounceResponse), + AnnounceIpv6(AnnounceResponse), Scrape(ScrapeResponse), Error(ErrorResponse), } +impl Response { + #[inline] + pub fn write(&self, bytes: &mut impl Write) -> Result<(), io::Error> { + match self { + Response::Connect(r) => { + bytes.write_i32::(0)?; + bytes.write_all(r.as_bytes())?; + } + Response::AnnounceIpv4(r) => { + bytes.write_i32::(1)?; + bytes.write_all(r.fixed.as_bytes())?; + bytes.write_all((*r.peers.as_slice()).as_bytes())?; + } + Response::AnnounceIpv6(r) => { + bytes.write_i32::(1)?; + bytes.write_all(r.fixed.as_bytes())?; + bytes.write_all((*r.peers.as_slice()).as_bytes())?; + } + Response::Scrape(r) => { + bytes.write_i32::(2)?; + bytes.write_all(r.transaction_id.as_bytes())?; + bytes.write_all((*r.torrent_stats.as_slice()).as_bytes())?; + } + Response::Error(r) => { + bytes.write_i32::(3)?; + bytes.write_all(r.transaction_id.as_bytes())?; + bytes.write_all(r.message.as_bytes())?; + } + } + + Ok(()) + } + + #[inline] + pub fn from_bytes(mut bytes: &[u8], ipv4: bool) -> Result { + let action = read_i32_ne(&mut bytes)?; + + match action.get() { + // Connect + 0 => Ok(Response::Connect( + ConnectResponse::read_from_prefix(bytes).ok_or_else(invalid_data)?, + )), + // Announce + 1 if ipv4 => { + let fixed = + AnnounceResponseFixedData::read_from_prefix(bytes).ok_or_else(invalid_data)?; + + let peers = if let Some(bytes) = bytes.get(size_of::()..) + { + Vec::from( + ResponsePeer::::slice_from(bytes) + .ok_or_else(invalid_data)?, + ) + } else { + Vec::new() + }; + + Ok(Response::AnnounceIpv4(AnnounceResponse { fixed, peers })) + } + 1 if !ipv4 => { + let fixed = + AnnounceResponseFixedData::read_from_prefix(bytes).ok_or_else(invalid_data)?; + + let peers = if let Some(bytes) = bytes.get(size_of::()..) + { + Vec::from( + ResponsePeer::::slice_from(bytes) + .ok_or_else(invalid_data)?, + ) + } else { + Vec::new() + }; + + Ok(Response::AnnounceIpv6(AnnounceResponse { fixed, peers })) + } + // Scrape + 2 => { + let transaction_id = read_i32_ne(&mut bytes).map(TransactionId)?; + let torrent_stats = + Vec::from(TorrentScrapeStatistics::slice_from(bytes).ok_or_else(invalid_data)?); + + Ok((ScrapeResponse { + transaction_id, + torrent_stats, + }) + .into()) + } + // Error + 3 => { + let transaction_id = read_i32_ne(&mut bytes).map(TransactionId)?; + let message = String::from_utf8_lossy(&bytes).into_owned().into(); + + Ok((ErrorResponse { + transaction_id, + message, + }) + .into()) + } + _ => Err(invalid_data()), + } + } +} + impl From for Response { fn from(r: ConnectResponse) -> Self { Self::Connect(r) } } -impl From> for Response { - fn from(r: AnnounceResponse) -> Self { +impl From> for Response { + fn from(r: AnnounceResponse) -> Self { Self::AnnounceIpv4(r) } } -impl From> for Response { - fn from(r: AnnounceResponse) -> Self { +impl From> for Response { + fn from(r: AnnounceResponse) -> Self { Self::AnnounceIpv6(r) } } @@ -80,203 +149,85 @@ impl From for Response { } } -impl Response { - #[inline] - pub fn write(&self, bytes: &mut impl Write) -> Result<(), io::Error> { - match self { - Response::Connect(r) => { - bytes.write_i32::(0)?; - bytes.write_i32::(r.transaction_id.0)?; - bytes.write_i64::(r.connection_id.0)?; - } - Response::AnnounceIpv4(r) => { - bytes.write_i32::(1)?; - bytes.write_i32::(r.transaction_id.0)?; - bytes.write_i32::(r.announce_interval.0)?; - bytes.write_i32::(r.leechers.0)?; - bytes.write_i32::(r.seeders.0)?; +#[derive(PartialEq, Eq, Clone, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(C, packed)] +pub struct ConnectResponse { + pub transaction_id: TransactionId, + pub connection_id: ConnectionId, +} - for peer in r.peers.iter() { - bytes.write_all(&peer.ip_address.octets())?; - bytes.write_u16::(peer.port.0)?; - } - } - Response::AnnounceIpv6(r) => { - bytes.write_i32::(1)?; - bytes.write_i32::(r.transaction_id.0)?; - bytes.write_i32::(r.announce_interval.0)?; - bytes.write_i32::(r.leechers.0)?; - bytes.write_i32::(r.seeders.0)?; +#[derive(PartialEq, Eq, Clone, Debug)] +pub struct AnnounceResponse { + pub fixed: AnnounceResponseFixedData, + pub peers: Vec>, +} - for peer in r.peers.iter() { - bytes.write_all(&peer.ip_address.octets())?; - bytes.write_u16::(peer.port.0)?; - } - } - Response::Scrape(r) => { - bytes.write_i32::(2)?; - bytes.write_i32::(r.transaction_id.0)?; +#[derive(PartialEq, Eq, Clone, Debug, AsBytes, FromBytes, FromZeroes)] +#[repr(C, packed)] +pub struct AnnounceResponseFixedData { + pub transaction_id: TransactionId, + pub announce_interval: AnnounceInterval, + pub leechers: NumberOfPeers, + pub seeders: NumberOfPeers, +} - for torrent_stat in r.torrent_stats.iter() { - bytes.write_i32::(torrent_stat.seeders.0)?; - bytes.write_i32::(torrent_stat.completed.0)?; - bytes.write_i32::(torrent_stat.leechers.0)?; - } - } - Response::Error(r) => { - bytes.write_i32::(3)?; - bytes.write_i32::(r.transaction_id.0)?; +#[derive(PartialEq, Eq, Clone, Debug)] +pub struct ScrapeResponse { + pub transaction_id: TransactionId, + pub torrent_stats: Vec, +} - bytes.write_all(r.message.as_bytes())?; - } - } +#[derive(PartialEq, Eq, Debug, Copy, Clone, AsBytes, FromBytes, FromZeroes)] +#[repr(C, packed)] +pub struct TorrentScrapeStatistics { + pub seeders: NumberOfPeers, + pub completed: NumberOfDownloads, + pub leechers: NumberOfPeers, +} - Ok(()) - } - - #[inline] - pub fn from_bytes(bytes: &[u8], ipv4: bool) -> Result { - let mut cursor = Cursor::new(bytes); - - let action = cursor.read_i32::()?; - let transaction_id = cursor.read_i32::()?; - - match action { - // Connect - 0 => { - let connection_id = cursor.read_i64::()?; - - Ok((ConnectResponse { - connection_id: ConnectionId(connection_id), - transaction_id: TransactionId(transaction_id), - }) - .into()) - } - // Announce - 1 if ipv4 => { - let announce_interval = cursor.read_i32::()?; - let leechers = cursor.read_i32::()?; - let seeders = cursor.read_i32::()?; - - let position = cursor.position() as usize; - let inner = cursor.into_inner(); - - let peers = inner[position..] - .chunks_exact(6) - .map(|chunk| { - let ip_bytes: [u8; 4] = (&chunk[..4]).try_into().unwrap(); - let ip_address = Ipv4Addr::from(ip_bytes); - let port = (&chunk[4..]).read_u16::().unwrap(); - - ResponsePeer { - ip_address, - port: Port(port), - } - }) - .collect(); - - Ok((AnnounceResponse { - transaction_id: TransactionId(transaction_id), - announce_interval: AnnounceInterval(announce_interval), - leechers: NumberOfPeers(leechers), - seeders: NumberOfPeers(seeders), - peers, - }) - .into()) - } - 1 if !ipv4 => { - let announce_interval = cursor.read_i32::()?; - let leechers = cursor.read_i32::()?; - let seeders = cursor.read_i32::()?; - - let position = cursor.position() as usize; - let inner = cursor.into_inner(); - - let peers = inner[position..] - .chunks_exact(18) - .map(|chunk| { - let ip_bytes: [u8; 16] = (&chunk[..16]).try_into().unwrap(); - let ip_address = Ipv6Addr::from(ip_bytes); - let port = (&chunk[16..]).read_u16::().unwrap(); - - ResponsePeer { - ip_address, - port: Port(port), - } - }) - .collect(); - - Ok((AnnounceResponse { - transaction_id: TransactionId(transaction_id), - announce_interval: AnnounceInterval(announce_interval), - leechers: NumberOfPeers(leechers), - seeders: NumberOfPeers(seeders), - peers, - }) - .into()) - } - // Scrape - 2 => { - let position = cursor.position() as usize; - let inner = cursor.into_inner(); - - let stats = inner[position..] - .chunks_exact(12) - .map(|chunk| { - let mut cursor: Cursor<&[u8]> = Cursor::new(&chunk[..]); - - let seeders = cursor.read_i32::().unwrap(); - let downloads = cursor.read_i32::().unwrap(); - let leechers = cursor.read_i32::().unwrap(); - - TorrentScrapeStatistics { - seeders: NumberOfPeers(seeders), - completed: NumberOfDownloads(downloads), - leechers: NumberOfPeers(leechers), - } - }) - .collect(); - - Ok((ScrapeResponse { - transaction_id: TransactionId(transaction_id), - torrent_stats: stats, - }) - .into()) - } - // Error - 3 => { - let position = cursor.position() as usize; - let inner = cursor.into_inner(); - - Ok((ErrorResponse { - transaction_id: TransactionId(transaction_id), - message: String::from_utf8_lossy(&inner[position..]) - .into_owned() - .into(), - }) - .into()) - } - _ => Ok((ErrorResponse { - transaction_id: TransactionId(transaction_id), - message: "Invalid action".into(), - }) - .into()), - } - } +#[derive(PartialEq, Eq, Clone, Debug)] +pub struct ErrorResponse { + pub transaction_id: TransactionId, + pub message: Cow<'static, str>, } #[cfg(test)] mod tests { use quickcheck_macros::quickcheck; + use zerocopy::network_endian::I32; + use zerocopy::network_endian::I64; use super::*; + impl quickcheck::Arbitrary for Ipv4AddrBytes { + fn arbitrary(g: &mut quickcheck::Gen) -> Self { + Self([ + u8::arbitrary(g), + u8::arbitrary(g), + u8::arbitrary(g), + u8::arbitrary(g), + ]) + } + } + + impl quickcheck::Arbitrary for Ipv6AddrBytes { + fn arbitrary(g: &mut quickcheck::Gen) -> Self { + let mut bytes = [0; 16]; + + for byte in bytes.iter_mut() { + *byte = u8::arbitrary(g) + } + + Self(bytes) + } + } + impl quickcheck::Arbitrary for TorrentScrapeStatistics { fn arbitrary(g: &mut quickcheck::Gen) -> Self { Self { - seeders: NumberOfPeers(i32::arbitrary(g)), - completed: NumberOfDownloads(i32::arbitrary(g)), - leechers: NumberOfPeers(i32::arbitrary(g)), + seeders: NumberOfPeers(I32::new(i32::arbitrary(g))), + completed: NumberOfDownloads(I32::new(i32::arbitrary(g))), + leechers: NumberOfPeers(I32::new(i32::arbitrary(g))), } } } @@ -284,8 +235,8 @@ mod tests { impl quickcheck::Arbitrary for ConnectResponse { fn arbitrary(g: &mut quickcheck::Gen) -> Self { Self { - connection_id: ConnectionId(i64::arbitrary(g)), - transaction_id: TransactionId(i32::arbitrary(g)), + connection_id: ConnectionId(I64::new(i64::arbitrary(g))), + transaction_id: TransactionId(I32::new(i32::arbitrary(g))), } } } @@ -297,10 +248,12 @@ mod tests { .collect(); Self { - transaction_id: TransactionId(i32::arbitrary(g)), - announce_interval: AnnounceInterval(i32::arbitrary(g)), - leechers: NumberOfPeers(i32::arbitrary(g)), - seeders: NumberOfPeers(i32::arbitrary(g)), + fixed: AnnounceResponseFixedData { + transaction_id: TransactionId(I32::new(i32::arbitrary(g))), + announce_interval: AnnounceInterval(I32::new(i32::arbitrary(g))), + leechers: NumberOfPeers(I32::new(i32::arbitrary(g))), + seeders: NumberOfPeers(I32::new(i32::arbitrary(g))), + }, peers, } } @@ -313,7 +266,7 @@ mod tests { .collect(); Self { - transaction_id: TransactionId(i32::arbitrary(g)), + transaction_id: TransactionId(I32::new(i32::arbitrary(g))), torrent_stats, } } @@ -328,7 +281,7 @@ mod tests { let success = response == r2; if !success { - println!("before: {:#?}\nafter: {:#?}", response, r2); + ::pretty_assertions::assert_eq!(response, r2); } success @@ -340,12 +293,16 @@ mod tests { } #[quickcheck] - fn test_announce_response_ipv4_convert_identity(response: AnnounceResponse) -> bool { + fn test_announce_response_ipv4_convert_identity( + response: AnnounceResponse, + ) -> bool { same_after_conversion(response.into(), true) } #[quickcheck] - fn test_announce_response_ipv6_convert_identity(response: AnnounceResponse) -> bool { + fn test_announce_response_ipv6_convert_identity( + response: AnnounceResponse, + ) -> bool { same_after_conversion(response.into(), false) } diff --git a/docker/ci.Dockerfile b/docker/ci.Dockerfile new file mode 100644 index 0000000..91890d5 --- /dev/null +++ b/docker/ci.Dockerfile @@ -0,0 +1,18 @@ +# Can be used to run file transfer CI test locally. Usage: +# 1. docker build -t aquatic -f ./docker/ci.Dockerfile . +# 2. docker run aquatic +# 3. On failure, run `docker rmi aquatic -f` and go back to step 1 + +FROM rust:bullseye + +RUN mkdir "/opt/aquatic" + +ENV "GITHUB_WORKSPACE" "/opt/aquatic" + +WORKDIR "/opt/aquatic" + +COPY ./.github/actions/test-file-transfers/entrypoint.sh entrypoint.sh +COPY Cargo.toml Cargo.lock ./ +COPY crates crates + +ENTRYPOINT ["./entrypoint.sh"] \ No newline at end of file