From 00ee87b286dc5af62d2d7e8f6e4174f1c860fc8e Mon Sep 17 00:00:00 2001 From: Nehliin Date: Sat, 29 Aug 2026 12:47:36 +0200 Subject: [PATCH 1/7] Support banning peers --- bittorrent/src/connection_manager.rs | 19 ++++++++++++++++++- bittorrent/src/event_loop.rs | 4 ++++ bittorrent/src/torrent.rs | 2 ++ 3 files changed, 24 insertions(+), 1 deletion(-) diff --git a/bittorrent/src/connection_manager.rs b/bittorrent/src/connection_manager.rs index 4125677..be5ba28 100644 --- a/bittorrent/src/connection_manager.rs +++ b/bittorrent/src/connection_manager.rs @@ -3,6 +3,7 @@ use std::net::SocketAddr; use std::ops::{Index, IndexMut}; use std::os::fd::{AsRawFd, RawFd}; +use ahash::HashSet; use bytes::BufMut; use heapless::spsc::Producer; use rayon::Scope; @@ -68,6 +69,7 @@ impl ConnectionState { pub struct ConnectionManager { connections: SlotMap, + ban_list: HashSet, max_connections: usize, our_id: PeerId, } @@ -77,6 +79,7 @@ impl ConnectionManager { Self { connections: SlotMap::with_capacity_and_key(max_connections), max_connections, + ban_list: Default::default(), our_id, } } @@ -84,7 +87,7 @@ impl ConnectionManager { // The address scan is linear but since it's bound by max_connections, which // is expected to be in the hundreds, it should be fast fn can_accept_new(&self, peer: SocketAddr) -> bool { - if self.connections.len() >= self.max_connections { + if self.connections.len() >= self.max_connections || self.ban_list.contains(&peer) { log::trace!( "Ignoring peer, max connections ({}) reached", self.max_connections @@ -94,6 +97,20 @@ impl ConnectionManager { !self.connections.values().any(|state| state.addr() == peer) } + pub fn ban_peer<'state, Q: SubmissionQueue>( + &mut self, + peer: SocketAddr, + io: &mut Io, + state: &mut StateRef<'state>, + ) { + self.ban_list.insert(peer); + for (conn_id, conn) in self.connections.iter_mut() { + if conn.addr() == peer { + Self::disconnect_entry(conn, conn_id, io, state); + } + } + } + /// Attempt to open an outgoing connection to `peer`. No-op if the peer is /// already tracked or the connection cap has been reached. pub fn maybe_connect_to_peer(&mut self, peer: SockAddr, io: &mut Io) { diff --git a/bittorrent/src/event_loop.rs b/bittorrent/src/event_loop.rs index adab2df..aa48d50 100644 --- a/bittorrent/src/event_loop.rs +++ b/bittorrent/src/event_loop.rs @@ -702,6 +702,10 @@ impl<'scope, 'state: 'scope> EventLoop { ); } } + Command::BanPeer { peer_addr } => { + log::debug!("Banning a peer with the addr: {peer_addr}"); + connection_manager.ban_peer(peer_addr, io, state_ref); + } Command::Stop => { if !matches!(self.state, EventLoopState::ShuttingDown { .. }) { log::info!("Shutdown requested, closing all connections"); diff --git a/bittorrent/src/torrent.rs b/bittorrent/src/torrent.rs index c8d568c..339dfe8 100644 --- a/bittorrent/src/torrent.rs +++ b/bittorrent/src/torrent.rs @@ -175,6 +175,8 @@ pub enum Command { /// Use the `TorrentEvent::Paused` event to determine when the torrent successfully entered /// the paused state. Resume, + /// Ban a specific peer + BanPeer { peer_addr: SocketAddr }, } /// Metrics for a given peer From e6cb97b3f11996d9199545c0a4363910ad260499 Mon Sep 17 00:00:00 2001 From: Nehliin Date: Sat, 29 Aug 2026 13:12:56 +0200 Subject: [PATCH 2/7] Add tests --- bittorrent/src/peer_comm/tests.rs | 118 ++++++++++- bittorrent/tests/ban_peer.rs | 338 ++++++++++++++++++++++++++++++ 2 files changed, 454 insertions(+), 2 deletions(-) create mode 100644 bittorrent/tests/ban_peer.rs diff --git a/bittorrent/src/peer_comm/tests.rs b/bittorrent/src/peer_comm/tests.rs index b1f24ba..2f7b9e3 100644 --- a/bittorrent/src/peer_comm/tests.rs +++ b/bittorrent/src/peer_comm/tests.rs @@ -4371,14 +4371,20 @@ fn bad_peer_addrs(event_rx: &mut Consumer<'_, TorrentEvent>) -> Vec addrs } +#[track_caller] +fn insert_established_at(connections: &mut ConnectionManager, addr: SocketAddr) -> ConnectionId { + let conn_id = connections.insert_established_with_key(|k| generate_peer(true, k)); + connections[conn_id].peer_addr = addr; + conn_id +} + #[track_caller] fn mark_for_disconnect( connections: &mut ConnectionManager, addr: &str, reason: DisconnectReason, ) -> ConnectionId { - let conn_id = connections.insert_established_with_key(|k| generate_peer(true, k)); - connections[conn_id].peer_addr = addr.parse().unwrap(); + let conn_id = insert_established_at(connections, addr.parse().unwrap()); connections[conn_id].pending_disconnect = Some(reason); conn_id } @@ -4704,6 +4710,114 @@ fn rejected_incoming_connection_is_closed_without_being_tracked() { assert_eq!(scheduled_closes(&io), vec![None]); } +// Banning a peer tears down the connection it currently has and keeps it out +// for good, even after the slot of the closed connection has been freed +#[test] +fn banned_peer_is_disconnected_and_kept_out() { + let mut download_state = setup_test(); + + rayon::in_place_scope(|_scope| { + let mut state_ref = download_state.as_ref(); + let mut connections = ConnectionManager::new(PeerId::generate(), 128); + let mut io = Io::new( + BackloggedSubmissionQueue::new(MockSubmissionQueue), + &torrent::Config::default(), + ); + + let banned: SocketAddrV4 = "127.0.0.1:6881".parse().unwrap(); + let conn_id = insert_established_at(&mut connections, SocketAddr::V4(banned)); + + connections.ban_peer(SocketAddr::V4(banned), &mut io, &mut state_ref); + + // The established connection is torn down like any other disconnect + assert_eq!(scheduled_closes(&io), vec![Some(conn_id)]); + assert!(connections.established_mut(conn_id).is_none()); + assert_eq!(connections.num_established(), 0); + + // Freeing the slot doesn't make the peer connectable again + connections.remove_closed(conn_id); + assert!(connections.is_empty()); + connections.maybe_connect_to_peer(banned.into(), &mut io); + assert!( + scheduled_connects(&io).is_empty(), + "a banned peer should never be connected to again" + ); + assert!(connections.is_empty()); + }); +} + +// The ban applies to connections the peer initiates as well, not only to the +// ones we initiate +#[test] +fn incoming_connection_from_a_banned_peer_is_rejected() { + let mut download_state = setup_test(); + + rayon::in_place_scope(|_scope| { + let mut state_ref = download_state.as_ref(); + let info_hash = *state_ref.info_hash(); + let mut connections = ConnectionManager::new(PeerId::generate(), 128); + let mut io = Io::new( + BackloggedSubmissionQueue::new(MockSubmissionQueue), + &torrent::Config::default(), + ); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + connections.ban_peer(client.local_addr().unwrap(), &mut io, &mut state_ref); + + let (accepted, _) = listener.accept().unwrap(); + connections + .on_accepted(accepted.into(), info_hash, &mut io) + .unwrap(); + + // Rejected before being tracked, so the close carries no connection id + assert!(connections.is_empty()); + assert_eq!(scheduled_closes(&io), vec![None]); + }); +} + +// Only the banned peer is affected, connections to everyone else are untouched +#[test] +fn banning_a_peer_leaves_other_connections_alone() { + let mut download_state = setup_test(); + + rayon::in_place_scope(|_scope| { + let mut state_ref = download_state.as_ref(); + let mut connections = ConnectionManager::new(PeerId::generate(), 128); + let mut io = Io::new( + BackloggedSubmissionQueue::new(MockSubmissionQueue), + &torrent::Config::default(), + ); + + // Both peers are connected to before either of them is banned + let banned: SocketAddrV4 = "127.0.0.1:6881".parse().unwrap(); + let other: SocketAddrV4 = "127.0.0.1:6882".parse().unwrap(); + connections.maybe_connect_to_peer(banned.into(), &mut io); + let [banned_conn] = scheduled_connects(&io)[..] else { + panic!("the peer about to be banned should have a connect scheduled"); + }; + connections.maybe_connect_to_peer(other.into(), &mut io); + let other_connects: Vec<_> = scheduled_connects(&io) + .into_iter() + .filter(|conn_id| *conn_id != banned_conn) + .collect(); + let [other_conn] = other_connects[..] else { + panic!("the other peer should have a connect scheduled"); + }; + assert_eq!(connections.total_connections(), 2); + + connections.ban_peer(SocketAddr::V4(banned), &mut io, &mut state_ref); + + // Only the banned peer is torn down + assert_eq!(scheduled_closes(&io), vec![Some(banned_conn)]); + assert!(connections.fd(banned_conn).is_none()); + // while the other one keeps its socket and is left to finish connecting + assert!(connections.fd(other_conn).is_some()); + connections.remove_closed(banned_conn); + assert_eq!(connections.total_connections(), 1); + }); +} + // A handshake as it would arrive from a peer, advertising the fast extension // (BEP 6) or not. The extension protocol is deliberately left unadvertised so // that the only message queued up in response is the initial bitfield one. diff --git a/bittorrent/tests/ban_peer.rs b/bittorrent/tests/ban_peer.rs new file mode 100644 index 0000000..babf7eb --- /dev/null +++ b/bittorrent/tests/ban_peer.rs @@ -0,0 +1,338 @@ +mod common; + +use std::{ + collections::HashMap, + net::{SocketAddr, SocketAddrV4, TcpListener}, + sync::{ + Arc, + atomic::{AtomicBool, AtomicU16, Ordering}, + }, + time::{Duration, Instant}, +}; + +use heapless::spsc; +use vortex_bittorrent::{Command, Config, PeerId, State, Torrent, TorrentEvent}; + +use common::{TempDir, create_test_torrent, init_test_environment}; + +const TIMEOUT: u64 = 120; +/// How long the banned peer is given to actually go away after the ban was sent +const DISCONNECT_TIMEOUT: Duration = Duration::from_secs(15); +/// A piece whose last subpiece arrived just before the connection was closed may +/// still finish hashing afterwards, so the download is only required to have +/// stopped making progress once it has been given this long to settle +const SETTLE_TIME: Duration = Duration::from_secs(3); +/// How long the download is watched after the disconnect to ensure it stays aborted +const ABORTED_OBSERVATION: Duration = Duration::from_secs(8); + +/// What the downloader is currently waiting for +#[derive(Clone, Copy)] +enum Phase { + /// Downloading from the seeder, waiting for enough progress to ban it mid download + Downloading, + /// The ban has been sent, waiting for the seeder to be disconnected + Banned { since: Instant }, + /// The seeder is gone, making sure the download stays aborted + Aborted { + since: Instant, + /// The completed piece count once the inflight pieces were given time to + /// finish, `None` until the download has been left alone for [`SETTLE_TIME`] + settled_progress: Option, + }, +} + +/// Banning the only peer we are downloading from mid download disconnects it and +/// aborts the download instead of finishing it +#[test] +fn banned_peer_aborts_download() { + init_test_environment(); + + // Enough pieces that the download is guaranteed to still be in progress by + // the time the first completed piece is reported + let test_files: HashMap> = [ + ( + "file2.txt".to_string(), + b"BitTorrent Test Data!".repeat(200), + ), + ("subdir/file3.txt".to_string(), vec![42u8; 16384 * 5_000]), + ] + .into_iter() + .collect(); + + let torrent_name = format!("test_ban_peer_{}", rand::random::()); + + let (seeder_dir, metadata) = create_test_torrent(&test_files, &torrent_name, 16384); + let num_pieces = metadata.pieces.len(); + let tmp_dir_path = seeder_dir.path().clone(); + + let downloader_dir = TempDir::new(&format!("{}_downloader", torrent_name)); + let downloader_path = downloader_dir.path().clone(); + + // TODO: Fix this in a better way + let prev_hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |info| { + prev_hook(info); + // hacky clean up if a panic happens + let _ = std::fs::remove_dir_all(&tmp_dir_path); + let _ = std::fs::remove_dir_all(&downloader_path); + std::process::abort(); + })); + + // Set up seeder with the completed files + let seeder_id = PeerId::generate(); + let seeder_state = State::from_metadata_and_root( + seeder_id, + metadata.clone(), + seeder_dir.path().clone(), + Config::default(), + ) + .expect("Failed to create seeder state"); + let mut seeder_torrent = Torrent::new(seeder_state); + + // Set up downloader (empty) + let downloader_id = PeerId::generate(); + let downloader_state = State::from_metadata_and_root( + downloader_id, + metadata.clone(), + downloader_dir.path().clone(), + Config::default(), + ) + .expect("Failed to create downloader state"); + let mut downloader_torrent = Torrent::new(downloader_state); + + let (downloader_command_tx, downloader_command_rc) = std::sync::mpsc::sync_channel(64); + let downloader_command_tx_for_seeder = downloader_command_tx.clone(); + let (seeder_command_tx, seeder_command_rc) = std::sync::mpsc::sync_channel(64); + + let test_time = Instant::now(); + + let seeder_shutting_down = Arc::new(AtomicBool::new(false)); + let seeder_shutting_down_clone = seeder_shutting_down.clone(); + + // Shared so the downloader knows which address to ban + let seeder_port = Arc::new(AtomicU16::new(0)); + let seeder_port_for_seeder = seeder_port.clone(); + + let mut seeder_event_q: spsc::Queue = spsc::Queue::new(); + let (seeder_event_tx, mut seeder_event_rc) = seeder_event_q.split(); + + std::thread::scope(|s| { + // Seeder event loop thread + s.spawn(move || { + let seeder_listener = TcpListener::bind("127.0.0.1:0").unwrap(); + seeder_torrent + .start(seeder_event_tx, seeder_command_rc, seeder_listener) + .unwrap(); + }); + + // Seeder event handler thread, it observes the ban from the other side of + // the connection: the downloader should be gone and stay gone + let seeder_handle = s.spawn(move || { + let mut saw_downloader = false; + let mut saw_disconnect = false; + + loop { + if test_time.elapsed() >= Duration::from_secs(TIMEOUT) { + panic!("Test timeout in seeder event handler"); + } + + while let Some(event) = seeder_event_rc.dequeue() { + match event { + TorrentEvent::Running { port } => { + log::info!("Seeder running on port {}", port); + seeder_port_for_seeder.store(port, Ordering::Release); + downloader_command_tx_for_seeder + .send(Command::ConnectToPeers(vec![ + format!("127.0.0.1:{}", port).parse().unwrap(), + ])) + .unwrap(); + } + TorrentEvent::TorrentMetrics { peer_metrics, .. } => { + if peer_metrics.is_empty() { + if saw_downloader { + saw_disconnect = true; + } + } else { + assert!( + !saw_disconnect, + "The banned downloader reconnected to the seeder" + ); + saw_downloader = true; + } + } + TorrentEvent::Paused => panic!("Seeder should never pause"), + _ => {} + } + } + + if seeder_shutting_down.load(Ordering::Acquire) { + break; + } + } + (saw_downloader, saw_disconnect) + }); + + // Downloader thread + let downloader_handle = s.spawn(move || { + let mut downloader_event_q: spsc::Queue = spsc::Queue::new(); + let (downloader_event_tx, mut downloader_event_rc) = downloader_event_q.split(); + + let downloader_listener = TcpListener::bind("127.0.0.1:0").unwrap(); + + std::thread::scope(|downloader_scope| { + downloader_scope.spawn(move || { + downloader_torrent + .start( + downloader_event_tx, + downloader_command_rc, + downloader_listener, + ) + .unwrap(); + }); + + let mut phase = Phase::Downloading; + + loop { + if test_time.elapsed() >= Duration::from_secs(TIMEOUT) { + panic!("Test timeout - the download never reached the point of the ban"); + } + + while let Some(event) = downloader_event_rc.dequeue() { + match event { + TorrentEvent::TorrentComplete => match phase { + Phase::Downloading => panic!( + "The download completed before the seeder could be banned mid download" + ), + _ => panic!("The ban should have aborted the download"), + }, + TorrentEvent::TorrentMetrics { + progress, + peer_metrics, + .. + } => { + let completed = progress.as_ref().map_or(0, |p| p.total_completed()); + let connected_peers = peer_metrics.len(); + log::debug!( + "Downloader progress: {completed}/{num_pieces} pieces, {connected_peers} connected peers" + ); + match phase { + // Ban the seeder as soon as the download is + // demonstrably underway but far from done + Phase::Downloading if connected_peers > 0 && completed > 0 => { + assert!( + completed < num_pieces, + "The seeder should be banned mid download, not after it finished" + ); + let port = seeder_port.load(Ordering::Acquire); + assert_ne!(port, 0, "The seeder port should be known by now"); + let seeder_addr: SocketAddrV4 = + format!("127.0.0.1:{port}").parse().unwrap(); + log::info!( + "Banning the seeder at {seeder_addr} after {completed}/{num_pieces} pieces" + ); + downloader_command_tx + .send(Command::BanPeer { + peer_addr: SocketAddr::V4(seeder_addr), + }) + .unwrap(); + phase = Phase::Banned { + since: Instant::now(), + }; + } + Phase::Downloading => {} + Phase::Banned { since } => { + if connected_peers == 0 { + log::info!("The banned seeder was disconnected"); + // The ban has to outlive the disconnect, so + // reconnecting to it must be refused as well + let port = seeder_port.load(Ordering::Acquire); + downloader_command_tx + .send(Command::ConnectToPeers(vec![ + format!("127.0.0.1:{port}").parse().unwrap(), + ])) + .unwrap(); + phase = Phase::Aborted { + since: Instant::now(), + settled_progress: None, + }; + } else { + assert!( + since.elapsed() < DISCONNECT_TIMEOUT, + "The banned seeder is still connected {}s after the ban", + since.elapsed().as_secs() + ); + } + } + Phase::Aborted { + since, + settled_progress, + } => { + assert_eq!( + connected_peers, 0, + "The banned seeder must not be connected to again" + ); + assert!( + completed < num_pieces, + "The download must not complete without the banned seeder" + ); + match settled_progress { + Some(settled) => assert_eq!( + completed, settled, + "The download must not make progress after the only peer was banned" + ), + None if since.elapsed() >= SETTLE_TIME => { + phase = Phase::Aborted { + since, + settled_progress: Some(completed), + } + } + None => {} + } + } + } + } + TorrentEvent::MetadataComplete(_) => { + log::info!("Downloader: Metadata complete"); + } + TorrentEvent::Paused => panic!("Downloader should never pause"), + _ => {} + } + } + + if let Phase::Aborted { + since, + settled_progress, + } = phase + && since.elapsed() >= ABORTED_OBSERVATION + { + let settled_progress = settled_progress + .expect("The download should have reported metrics while aborted"); + let _ = downloader_command_tx.send(Command::Stop); + let _ = seeder_command_tx.send(Command::Stop); + seeder_shutting_down_clone.store(true, Ordering::Release); + return settled_progress; + } + } + }) + }); + + let final_progress = downloader_handle.join().unwrap(); + let (seeder_saw_downloader, seeder_saw_disconnect) = seeder_handle.join().unwrap(); + + assert!( + final_progress < num_pieces, + "The download should have been left incomplete, got {final_progress}/{num_pieces} pieces" + ); + assert!( + seeder_saw_downloader, + "The seeder never saw the downloader connect" + ); + assert!( + seeder_saw_disconnect, + "The seeder never saw the banned downloader disconnect" + ); + log::info!( + "Download aborted at {final_progress}/{num_pieces} pieces after banning the seeder" + ); + }); +} From 96ab8cfbcb55012a444d88be82518f9dce3083bd Mon Sep 17 00:00:00 2001 From: Nehliin Date: Sat, 29 Aug 2026 13:30:06 +0200 Subject: [PATCH 3/7] fix incorrect logline --- bittorrent/src/connection_manager.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/bittorrent/src/connection_manager.rs b/bittorrent/src/connection_manager.rs index be5ba28..a237d6e 100644 --- a/bittorrent/src/connection_manager.rs +++ b/bittorrent/src/connection_manager.rs @@ -87,7 +87,11 @@ impl ConnectionManager { // The address scan is linear but since it's bound by max_connections, which // is expected to be in the hundreds, it should be fast fn can_accept_new(&self, peer: SocketAddr) -> bool { - if self.connections.len() >= self.max_connections || self.ban_list.contains(&peer) { + if self.ban_list.contains(&peer) { + log::trace!("Ignoring peer that has already been banned ({peer})",); + return false; + } + if self.connections.len() >= self.max_connections { log::trace!( "Ignoring peer, max connections ({}) reached", self.max_connections From c7204e93cbddd58cb903a19b388ca3518cc35f91 Mon Sep 17 00:00:00 2001 From: Nehliin Date: Sat, 29 Aug 2026 16:07:15 +0200 Subject: [PATCH 4/7] Ban ips instead --- bittorrent/src/connection_manager.rs | 11 +++++++---- bittorrent/src/event_loop.rs | 6 +++--- bittorrent/src/torrent.rs | 7 ++++--- bittorrent/tests/ban_peer.rs | 4 ++-- 4 files changed, 16 insertions(+), 12 deletions(-) diff --git a/bittorrent/src/connection_manager.rs b/bittorrent/src/connection_manager.rs index a237d6e..f20e4a0 100644 --- a/bittorrent/src/connection_manager.rs +++ b/bittorrent/src/connection_manager.rs @@ -1,5 +1,5 @@ use std::io; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; use std::ops::{Index, IndexMut}; use std::os::fd::{AsRawFd, RawFd}; @@ -69,7 +69,7 @@ impl ConnectionState { pub struct ConnectionManager { connections: SlotMap, - ban_list: HashSet, + ban_list: HashSet, max_connections: usize, our_id: PeerId, } @@ -87,7 +87,7 @@ impl ConnectionManager { // The address scan is linear but since it's bound by max_connections, which // is expected to be in the hundreds, it should be fast fn can_accept_new(&self, peer: SocketAddr) -> bool { - if self.ban_list.contains(&peer) { + if self.ban_list.contains(&peer.ip()) { log::trace!("Ignoring peer that has already been banned ({peer})",); return false; } @@ -101,9 +101,12 @@ impl ConnectionManager { !self.connections.values().any(|state| state.addr() == peer) } + /// Ban all peers that have the given IP, disconnects any exiting peers. + /// ip is used instead of SocketAddr since incoming source ports are controlled + /// by the peer pub fn ban_peer<'state, Q: SubmissionQueue>( &mut self, - peer: SocketAddr, + peer_ip: IpAddr, io: &mut Io, state: &mut StateRef<'state>, ) { diff --git a/bittorrent/src/event_loop.rs b/bittorrent/src/event_loop.rs index aa48d50..144ef21 100644 --- a/bittorrent/src/event_loop.rs +++ b/bittorrent/src/event_loop.rs @@ -702,9 +702,9 @@ impl<'scope, 'state: 'scope> EventLoop { ); } } - Command::BanPeer { peer_addr } => { - log::debug!("Banning a peer with the addr: {peer_addr}"); - connection_manager.ban_peer(peer_addr, io, state_ref); + Command::BanPeer { peer_ip } => { + log::debug!("Banning peer(s): {peer_ip}"); + connection_manager.ban_peer(peer_ip, io, state_ref); } Command::Stop => { if !matches!(self.state, EventLoopState::ShuttingDown { .. }) { diff --git a/bittorrent/src/torrent.rs b/bittorrent/src/torrent.rs index 339dfe8..e516ef1 100644 --- a/bittorrent/src/torrent.rs +++ b/bittorrent/src/torrent.rs @@ -2,7 +2,7 @@ use std::{ cell::OnceCell, collections::VecDeque, io::{self}, - net::{SocketAddr, SocketAddrV4, TcpListener}, + net::{IpAddr, SocketAddr, SocketAddrV4, TcpListener}, path::{Path, PathBuf}, sync::mpsc::{Receiver, Sender}, time::{Duration, Instant}, @@ -175,8 +175,9 @@ pub enum Command { /// Use the `TorrentEvent::Paused` event to determine when the torrent successfully entered /// the paused state. Resume, - /// Ban a specific peer - BanPeer { peer_addr: SocketAddr }, + /// Ban any peer(s) for the remainder of this run: its current connection, if any, + /// is disconnected and it is neither connected to nor accepted again. + BanPeer { peer_ip: IpAddr }, } /// Metrics for a given peer diff --git a/bittorrent/tests/ban_peer.rs b/bittorrent/tests/ban_peer.rs index babf7eb..99d6895 100644 --- a/bittorrent/tests/ban_peer.rs +++ b/bittorrent/tests/ban_peer.rs @@ -2,7 +2,7 @@ mod common; use std::{ collections::HashMap, - net::{SocketAddr, SocketAddrV4, TcpListener}, + net::{IpAddr, SocketAddrV4, TcpListener}, sync::{ Arc, atomic::{AtomicBool, AtomicU16, Ordering}, @@ -232,7 +232,7 @@ fn banned_peer_aborts_download() { ); downloader_command_tx .send(Command::BanPeer { - peer_addr: SocketAddr::V4(seeder_addr), + peer_ip: IpAddr::V4(*seeder_addr.ip()), }) .unwrap(); phase = Phase::Banned { From 57d47cf7c8e8a4b27062817cf83d32a1c7ee8470 Mon Sep 17 00:00:00 2001 From: Nehliin Date: Sat, 29 Aug 2026 16:07:22 +0200 Subject: [PATCH 5/7] Add metrics for banned peers --- bittorrent/src/connection_manager.rs | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/bittorrent/src/connection_manager.rs b/bittorrent/src/connection_manager.rs index f20e4a0..23db5e4 100644 --- a/bittorrent/src/connection_manager.rs +++ b/bittorrent/src/connection_manager.rs @@ -110,11 +110,17 @@ impl ConnectionManager { io: &mut Io, state: &mut StateRef<'state>, ) { - self.ban_list.insert(peer); - for (conn_id, conn) in self.connections.iter_mut() { - if conn.addr() == peer { - Self::disconnect_entry(conn, conn_id, io, state); + self.ban_list.insert(peer_ip); + for (conn_id, entry) in self.connections.iter_mut() { + if entry.addr().ip() != peer_ip { + continue; + } + #[cfg(feature = "metrics")] + if let ConnectionState::Established { .. } = &*entry { + let counter = metrics::counter!("disconnects", "reason" => "banned"); + counter.increment(1); } + Self::disconnect_entry(entry, conn_id, io, state); } } From 5f5bd4d0ff4133d8c384970fc26f89d4d39ae1e7 Mon Sep 17 00:00:00 2001 From: Nehliin Date: Sat, 29 Aug 2026 16:07:26 +0200 Subject: [PATCH 6/7] update tests --- bittorrent/src/peer_comm/tests.rs | 88 ++++++++++++++++++++++++++++--- 1 file changed, 80 insertions(+), 8 deletions(-) diff --git a/bittorrent/src/peer_comm/tests.rs b/bittorrent/src/peer_comm/tests.rs index 2f7b9e3..737ba87 100644 --- a/bittorrent/src/peer_comm/tests.rs +++ b/bittorrent/src/peer_comm/tests.rs @@ -1,5 +1,5 @@ use std::{ - net::{SocketAddr, SocketAddrV4, TcpListener, TcpStream}, + net::{IpAddr, SocketAddr, SocketAddrV4, TcpListener, TcpStream}, rc::Rc, time::{Duration, Instant}, }; @@ -4727,7 +4727,7 @@ fn banned_peer_is_disconnected_and_kept_out() { let banned: SocketAddrV4 = "127.0.0.1:6881".parse().unwrap(); let conn_id = insert_established_at(&mut connections, SocketAddr::V4(banned)); - connections.ban_peer(SocketAddr::V4(banned), &mut io, &mut state_ref); + connections.ban_peer(IpAddr::V4(*banned.ip()), &mut io, &mut state_ref); // The established connection is torn down like any other disconnect assert_eq!(scheduled_closes(&io), vec![Some(conn_id)]); @@ -4738,6 +4738,10 @@ fn banned_peer_is_disconnected_and_kept_out() { connections.remove_closed(conn_id); assert!(connections.is_empty()); connections.maybe_connect_to_peer(banned.into(), &mut io); + // Neither does showing up on another port, the ban is on the peer and + // not on the address it happened to be reported on + let same_peer_other_port: SocketAddrV4 = "127.0.0.1:51413".parse().unwrap(); + connections.maybe_connect_to_peer(same_peer_other_port.into(), &mut io); assert!( scheduled_connects(&io).is_empty(), "a banned peer should never be connected to again" @@ -4746,8 +4750,9 @@ fn banned_peer_is_disconnected_and_kept_out() { }); } -// The ban applies to connections the peer initiates as well, not only to the -// ones we initiate +// The ban applies to connections the peer initiates as well, which is the case +// that matters most since those are the ones that can't be filtered out before +// a socket exists #[test] fn incoming_connection_from_a_banned_peer_is_rejected() { let mut download_state = setup_test(); @@ -4761,10 +4766,18 @@ fn incoming_connection_from_a_banned_peer_is_rejected() { &torrent::Config::default(), ); + // The peer is banned by the address it listens on, while the connection + // it opens to us comes from an ephemeral port that is different every time + let listening_addr: SocketAddrV4 = "127.0.0.1:6881".parse().unwrap(); + connections.ban_peer(IpAddr::V4(*listening_addr.ip()), &mut io, &mut state_ref); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let client = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); - connections.ban_peer(client.local_addr().unwrap(), &mut io, &mut state_ref); - + assert_ne!( + client.local_addr().unwrap().port(), + listening_addr.port(), + "the connection should come from an ephemeral port" + ); let (accepted, _) = listener.accept().unwrap(); connections .on_accepted(accepted.into(), info_hash, &mut io) @@ -4791,7 +4804,7 @@ fn banning_a_peer_leaves_other_connections_alone() { // Both peers are connected to before either of them is banned let banned: SocketAddrV4 = "127.0.0.1:6881".parse().unwrap(); - let other: SocketAddrV4 = "127.0.0.1:6882".parse().unwrap(); + let other: SocketAddrV4 = "127.0.0.2:6881".parse().unwrap(); connections.maybe_connect_to_peer(banned.into(), &mut io); let [banned_conn] = scheduled_connects(&io)[..] else { panic!("the peer about to be banned should have a connect scheduled"); @@ -4806,7 +4819,7 @@ fn banning_a_peer_leaves_other_connections_alone() { }; assert_eq!(connections.total_connections(), 2); - connections.ban_peer(SocketAddr::V4(banned), &mut io, &mut state_ref); + connections.ban_peer(IpAddr::V4(*banned.ip()), &mut io, &mut state_ref); // Only the banned peer is torn down assert_eq!(scheduled_closes(&io), vec![Some(banned_conn)]); @@ -4815,7 +4828,66 @@ fn banning_a_peer_leaves_other_connections_alone() { assert!(connections.fd(other_conn).is_some()); connections.remove_closed(banned_conn); assert_eq!(connections.total_connections(), 1); + // and everyone that isn't banned can still be connected to. The banned + // peer's connect event is still counted here, it lives until its cancel + // completes, which never happens against a mocked submission queue + let not_banned: SocketAddrV4 = "127.0.0.3:6881".parse().unwrap(); + let connects_before = scheduled_connects(&io).len(); + connections.maybe_connect_to_peer(not_banned.into(), &mut io); + assert_eq!( + scheduled_connects(&io).len(), + connects_before + 1, + "a peer that isn't banned should still be connected to" + ); + }); +} + +// Bans show up in the disconnect metrics like every other teardown, and a +// connection that is already on its way out isn't counted a second time +#[cfg(feature = "metrics")] +#[test] +fn banning_a_peer_records_a_banned_disconnect() { + use metrics::{Key, Label}; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use metrics_util::{CompositeKey, MetricKind}; + + let mut download_state = setup_test(); + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + rayon::in_place_scope(|_scope| { + let mut state_ref = download_state.as_ref(); + let mut connections = ConnectionManager::new(PeerId::generate(), 128); + let mut io = Io::new( + BackloggedSubmissionQueue::new(MockSubmissionQueue), + &torrent::Config::default(), + ); + + let banned: SocketAddrV4 = "127.0.0.1:6881".parse().unwrap(); + let conn_id = insert_established_at(&mut connections, SocketAddr::V4(banned)); + + metrics::with_local_recorder(&recorder, || { + connections.ban_peer(IpAddr::V4(*banned.ip()), &mut io, &mut state_ref); + // The second ban finds the connection closing, its teardown has + // already been counted + connections.ban_peer(IpAddr::V4(*banned.ip()), &mut io, &mut state_ref); + }); + + assert!(connections.established_mut(conn_id).is_none()); + assert_eq!(scheduled_closes(&io), vec![Some(conn_id)]); }); + + let snapshot = snapshotter.snapshot(); + #[allow(clippy::mutable_key_type)] + let metrics = snapshot.into_hashmap(); + let banned_disconnects = CompositeKey::new( + MetricKind::Counter, + Key::from_parts("disconnects", vec![Label::new("reason", "banned")]), + ); + let Some((_, _, DebugValue::Counter(disconnects))) = metrics.get(&banned_disconnects) else { + panic!("the ban should have been counted as a disconnect: {metrics:?}"); + }; + assert_eq!(*disconnects, 1); } // A handshake as it would arrive from a peer, advertising the fast extension From 8ea874f4e0ed6928fc20c6e233cb409f400a3afb Mon Sep 17 00:00:00 2001 From: Nehliin Date: Sat, 29 Aug 2026 16:21:54 +0200 Subject: [PATCH 7/7] fix typo --- bittorrent/src/connection_manager.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/bittorrent/src/connection_manager.rs b/bittorrent/src/connection_manager.rs index 23db5e4..6a5e69a 100644 --- a/bittorrent/src/connection_manager.rs +++ b/bittorrent/src/connection_manager.rs @@ -101,7 +101,7 @@ impl ConnectionManager { !self.connections.values().any(|state| state.addr() == peer) } - /// Ban all peers that have the given IP, disconnects any exiting peers. + /// Ban all peers that have the given IP, disconnects any existing peers. /// ip is used instead of SocketAddr since incoming source ports are controlled /// by the peer pub fn ban_peer<'state, Q: SubmissionQueue>( @@ -190,7 +190,7 @@ impl ConnectionManager { let addr = addr.as_socket().expect("must be AF_INET"); if !self.can_accept_new(addr) { log::debug!( - "Rejecting incoming connection from already tracked or excess peer: {addr:?}" + "Rejecting incoming connection from already tracked, banned or excess peer: {addr:?}" ); io.close_socket(socket, None); return Ok(());