feat(lib): answer FolderContentsRequest with the folder's listing
"Download folder" in SoulseekQt and Nicotine+ asks the sharer for that one folder (peer code 36) instead of the whole share, and a sharer that never answers leaves the folder download hanging. The requested folder now comes back as a FolderContentsResponse (code 37), which carries a directory listing in the same form a browse does, so the two share one codec. The browse and folder replies also share one reply helper.
This commit is contained in:
@@ -2,10 +2,10 @@ use crate::actor::{Actor, ActorHandle, ConnectionState};
|
|||||||
use crate::client::ClientOperation;
|
use crate::client::ClientOperation;
|
||||||
use crate::dispatcher::MessageDispatcher;
|
use crate::dispatcher::MessageDispatcher;
|
||||||
use crate::message::peer::{
|
use crate::message::peer::{
|
||||||
FileSearchResponse, GetShareFileList, PeerInit, PlaceInQueueRequest,
|
FileSearchResponse, FolderContentsRequest, GetShareFileList, PeerInit,
|
||||||
PlaceInQueueResponse, QueueUploadHandler, SharedDirectory,
|
PlaceInQueueRequest, PlaceInQueueResponse, QueueUploadHandler,
|
||||||
SharedFileListResponseHandler, TransferRequest, TransferResponse,
|
SharedDirectory, SharedFileListResponseHandler, TransferRequest,
|
||||||
UploadDeniedHandler, UploadFailedHandler,
|
TransferResponse, UploadDeniedHandler, UploadFailedHandler,
|
||||||
};
|
};
|
||||||
use crate::message::server::MessageFactory;
|
use crate::message::server::MessageFactory;
|
||||||
use crate::message::{Handlers, Message, MessageReader, MessageType};
|
use crate::message::{Handlers, Message, MessageReader, MessageType};
|
||||||
@@ -46,6 +46,11 @@ pub enum PeerMessage {
|
|||||||
ShareListRequested,
|
ShareListRequested,
|
||||||
/// A peer we are browsing sent us their shared-file listing (code 5).
|
/// A peer we are browsing sent us their shared-file listing (code 5).
|
||||||
ShareListReceived(Vec<SharedDirectory>),
|
ShareListReceived(Vec<SharedDirectory>),
|
||||||
|
/// A peer asked for one folder of our shares (they sent us code 36).
|
||||||
|
FolderContentsRequested {
|
||||||
|
token: u32,
|
||||||
|
folder: String,
|
||||||
|
},
|
||||||
/// Offer the queued file to that peer: send an upload TransferRequest.
|
/// Offer the queued file to that peer: send an upload TransferRequest.
|
||||||
ServeUpload {
|
ServeUpload {
|
||||||
token: u32,
|
token: u32,
|
||||||
@@ -178,6 +183,7 @@ impl PeerActor {
|
|||||||
handlers.register_handler(TransferRequest);
|
handlers.register_handler(TransferRequest);
|
||||||
handlers.register_handler(TransferResponse);
|
handlers.register_handler(TransferResponse);
|
||||||
handlers.register_handler(GetShareFileList);
|
handlers.register_handler(GetShareFileList);
|
||||||
|
handlers.register_handler(FolderContentsRequest);
|
||||||
handlers.register_handler(UploadDeniedHandler);
|
handlers.register_handler(UploadDeniedHandler);
|
||||||
handlers.register_handler(UploadFailedHandler);
|
handlers.register_handler(UploadFailedHandler);
|
||||||
handlers.register_handler(PlaceInQueueRequest);
|
handlers.register_handler(PlaceInQueueRequest);
|
||||||
@@ -273,6 +279,9 @@ impl PeerActor {
|
|||||||
PeerMessage::ShareListReceived(directories) => {
|
PeerMessage::ShareListReceived(directories) => {
|
||||||
self.handle_share_list_received(directories);
|
self.handle_share_list_received(directories);
|
||||||
}
|
}
|
||||||
|
PeerMessage::FolderContentsRequested { token, folder } => {
|
||||||
|
self.handle_folder_contents_requested(token, folder);
|
||||||
|
}
|
||||||
PeerMessage::RequestTransfer(download) => {
|
PeerMessage::RequestTransfer(download) => {
|
||||||
let message = MessageFactory::build_transfer_request_message(
|
let message = MessageFactory::build_transfer_request_message(
|
||||||
&download.filename,
|
&download.filename,
|
||||||
@@ -441,6 +450,20 @@ impl PeerActor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn handle_folder_contents_requested(&self, token: u32, folder: String) {
|
||||||
|
let requester_key = self.peer_username();
|
||||||
|
if let Err(e) =
|
||||||
|
self.client_channel
|
||||||
|
.send(ClientOperation::FolderContentsRequested {
|
||||||
|
requester_key,
|
||||||
|
token,
|
||||||
|
folder,
|
||||||
|
})
|
||||||
|
{
|
||||||
|
error!("[peer_actor] forward FolderContentsRequested: {}", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn handle_share_list_received(&self, directories: Vec<SharedDirectory>) {
|
fn handle_share_list_received(&self, directories: Vec<SharedDirectory>) {
|
||||||
let username = self.peer_username();
|
let username = self.peer_username();
|
||||||
if let Err(e) =
|
if let Err(e) =
|
||||||
|
|||||||
@@ -292,6 +292,12 @@ pub enum ClientOperation {
|
|||||||
ShareListRequested {
|
ShareListRequested {
|
||||||
requester_key: String,
|
requester_key: String,
|
||||||
},
|
},
|
||||||
|
/// A peer asked for one folder of our shares; send a FolderContentsResponse.
|
||||||
|
FolderContentsRequested {
|
||||||
|
requester_key: String,
|
||||||
|
token: u32,
|
||||||
|
folder: String,
|
||||||
|
},
|
||||||
/// A peer we are browsing returned their shared-file listing.
|
/// A peer we are browsing returned their shared-file listing.
|
||||||
BrowseResult {
|
BrowseResult {
|
||||||
username: String,
|
username: String,
|
||||||
|
|||||||
@@ -756,35 +756,36 @@ impl Client {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
ClientOperation::ShareListRequested { requester_key } => {
|
ClientOperation::ShareListRequested { requester_key } => {
|
||||||
// Reply with our full shared-file listing.
|
Self::reply_to_peer(
|
||||||
let (registry, message) = match client_context
|
&client_context,
|
||||||
.read_safe()
|
&requester_key,
|
||||||
{
|
|ctx| {
|
||||||
Ok(ctx) => {
|
crate::message::peer::build_shared_file_list(
|
||||||
let dirs = ctx
|
&shared_directories(ctx),
|
||||||
.shares
|
|
||||||
.directories()
|
|
||||||
.into_iter()
|
|
||||||
.map(|(name, files)| {
|
|
||||||
crate::message::peer::SharedDirectory {
|
|
||||||
name,
|
|
||||||
files,
|
|
||||||
}
|
|
||||||
})
|
|
||||||
.collect::<Vec<_>>();
|
|
||||||
(
|
|
||||||
ctx.peer_registry.clone(),
|
|
||||||
crate::message::peer::build_shared_file_list(&dirs),
|
|
||||||
)
|
)
|
||||||
}
|
},
|
||||||
Err(_) => continue,
|
);
|
||||||
};
|
}
|
||||||
if let Some(registry) = registry {
|
ClientOperation::FolderContentsRequested {
|
||||||
let _ = registry.send_to_peer(
|
requester_key,
|
||||||
&requester_key,
|
token,
|
||||||
PeerMessage::SendMessage(message),
|
folder,
|
||||||
);
|
} => {
|
||||||
}
|
// ponytail: walks the whole share for one folder, as a
|
||||||
|
// browse does; index by directory if it ever shows up.
|
||||||
|
Self::reply_to_peer(
|
||||||
|
&client_context,
|
||||||
|
&requester_key,
|
||||||
|
|ctx| {
|
||||||
|
let dirs: Vec<_> = shared_directories(ctx)
|
||||||
|
.into_iter()
|
||||||
|
.filter(|dir| dir.name == folder)
|
||||||
|
.collect();
|
||||||
|
crate::message::peer::build_folder_contents(
|
||||||
|
token, &folder, &dirs,
|
||||||
|
)
|
||||||
|
},
|
||||||
|
);
|
||||||
}
|
}
|
||||||
ClientOperation::BrowseResult {
|
ClientOperation::BrowseResult {
|
||||||
username,
|
username,
|
||||||
@@ -845,6 +846,22 @@ impl Client {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Answer a peer with the message `build` derives from the client state.
|
||||||
|
fn reply_to_peer(
|
||||||
|
client_context: &Arc<RwLock<ClientContext>>,
|
||||||
|
requester_key: &str,
|
||||||
|
build: impl FnOnce(&ClientContext) -> crate::message::Message,
|
||||||
|
) {
|
||||||
|
let (registry, message) = match client_context.read_safe() {
|
||||||
|
Ok(ctx) => (ctx.peer_registry.clone(), build(&ctx)),
|
||||||
|
Err(_) => return,
|
||||||
|
};
|
||||||
|
if let Some(registry) = registry {
|
||||||
|
let _ = registry
|
||||||
|
.send_to_peer(requester_key, PeerMessage::SendMessage(message));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn sweep_expired_connects(client_context: &Arc<RwLock<ClientContext>>) {
|
fn sweep_expired_connects(client_context: &Arc<RwLock<ClientContext>>) {
|
||||||
let expired = match client_context.write_safe() {
|
let expired = match client_context.write_safe() {
|
||||||
Ok(mut ctx) => ctx.take_expired_connects(Instant::now()),
|
Ok(mut ctx) => ctx.take_expired_connects(Instant::now()),
|
||||||
@@ -855,3 +872,17 @@ impl Client {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Our shares in the wire form a browse or folder reply carries.
|
||||||
|
fn shared_directories(
|
||||||
|
ctx: &ClientContext,
|
||||||
|
) -> Vec<crate::message::peer::SharedDirectory> {
|
||||||
|
ctx.shares
|
||||||
|
.directories()
|
||||||
|
.into_iter()
|
||||||
|
.map(|(name, files)| crate::message::peer::SharedDirectory {
|
||||||
|
name,
|
||||||
|
files,
|
||||||
|
})
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,68 @@
|
|||||||
|
//! `FolderContentsRequest` (peer code 36) and its reply (code 37): one folder
|
||||||
|
//! of a share in the listing form code 5 uses, which is what "download
|
||||||
|
//! folder" in other clients asks for.
|
||||||
|
|
||||||
|
use crate::message::peer::SharedDirectory;
|
||||||
|
use crate::message::peer::shared_file_list::{
|
||||||
|
decompress_body, read_directories, write_directories,
|
||||||
|
};
|
||||||
|
use crate::message::{Message, MessageHandler};
|
||||||
|
use crate::peer::PeerMessage;
|
||||||
|
use crate::utils::zlib::compress_stored;
|
||||||
|
use std::sync::mpsc::Sender;
|
||||||
|
|
||||||
|
pub struct FolderContentsRequest;
|
||||||
|
impl MessageHandler<PeerMessage> for FolderContentsRequest {
|
||||||
|
fn get_code(&self) -> u32 {
|
||||||
|
36
|
||||||
|
}
|
||||||
|
fn handle(&self, message: &mut Message, sender: Sender<PeerMessage>) {
|
||||||
|
let token = message.read_int32();
|
||||||
|
let folder = message.read_string();
|
||||||
|
let _ =
|
||||||
|
sender.send(PeerMessage::FolderContentsRequested { token, folder });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build a `FolderContentsResponse` (peer code 37).
|
||||||
|
#[must_use]
|
||||||
|
pub fn build_folder_contents(
|
||||||
|
token: u32,
|
||||||
|
folder: &str,
|
||||||
|
dirs: &[SharedDirectory],
|
||||||
|
) -> Message {
|
||||||
|
let mut payload = Message::new();
|
||||||
|
payload.write_int32(token).write_string(folder);
|
||||||
|
write_directories(&mut payload, dirs);
|
||||||
|
Message::new()
|
||||||
|
.write_int32(37)
|
||||||
|
.write_raw_bytes(compress_stored(&payload.get_data()))
|
||||||
|
.clone()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Parse a `FolderContentsResponse` positioned at its compressed blob, or
|
||||||
|
/// `None` when it does not inflate.
|
||||||
|
#[must_use]
|
||||||
|
pub fn parse_folder_contents(
|
||||||
|
message: &mut Message,
|
||||||
|
) -> Option<(u32, String, Vec<SharedDirectory>)> {
|
||||||
|
let mut body = decompress_body(message)?;
|
||||||
|
let token = body.read_int32();
|
||||||
|
let folder = body.read_string();
|
||||||
|
Some((token, folder, read_directories(&mut body)))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn a_folder_listing_round_trips() {
|
||||||
|
let dirs = vec![SharedDirectory {
|
||||||
|
name: "music\\album".to_string(),
|
||||||
|
files: vec![("one.flac".to_string(), 40), ("two.flac".to_string(), 50)],
|
||||||
|
}];
|
||||||
|
let built = build_folder_contents(7, "music\\album", &dirs);
|
||||||
|
let mut message = Message::new_with_data(built.get_data());
|
||||||
|
message.set_pointer(4);
|
||||||
|
assert_eq!(
|
||||||
|
parse_folder_contents(&mut message),
|
||||||
|
Some((7, "music\\album".to_string(), dirs))
|
||||||
|
);
|
||||||
|
}
|
||||||
@@ -1,4 +1,5 @@
|
|||||||
mod file_search_response;
|
mod file_search_response;
|
||||||
|
mod folder_contents;
|
||||||
mod get_share_file_list;
|
mod get_share_file_list;
|
||||||
mod peer_init;
|
mod peer_init;
|
||||||
mod place_in_queue_request;
|
mod place_in_queue_request;
|
||||||
@@ -14,6 +15,9 @@ mod upload_failed;
|
|||||||
pub use file_search_response::{
|
pub use file_search_response::{
|
||||||
FileEntry, FileSearchResponse, build_file_search_response,
|
FileEntry, FileSearchResponse, build_file_search_response,
|
||||||
};
|
};
|
||||||
|
pub use folder_contents::{
|
||||||
|
FolderContentsRequest, build_folder_contents, parse_folder_contents,
|
||||||
|
};
|
||||||
pub use get_share_file_list::GetShareFileList;
|
pub use get_share_file_list::GetShareFileList;
|
||||||
pub use peer_init::PeerInit;
|
pub use peer_init::PeerInit;
|
||||||
pub use place_in_queue_request::PlaceInQueueRequest;
|
pub use place_in_queue_request::PlaceInQueueRequest;
|
||||||
|
|||||||
@@ -30,20 +30,7 @@ impl MessageHandler<PeerMessage> for SharedFileListResponseHandler {
|
|||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn build_shared_file_list(dirs: &[SharedDirectory]) -> Message {
|
pub fn build_shared_file_list(dirs: &[SharedDirectory]) -> Message {
|
||||||
let mut payload = Message::new();
|
let mut payload = Message::new();
|
||||||
payload.write_int32(dirs.len() as u32);
|
write_directories(&mut payload, dirs);
|
||||||
for dir in dirs {
|
|
||||||
payload
|
|
||||||
.write_string(&dir.name)
|
|
||||||
.write_int32(dir.files.len() as u32);
|
|
||||||
for (name, size) in &dir.files {
|
|
||||||
payload
|
|
||||||
.write_int8(1)
|
|
||||||
.write_string(name)
|
|
||||||
.write_int64(*size)
|
|
||||||
.write_string("") // extension
|
|
||||||
.write_int32(0); // attribute count
|
|
||||||
}
|
|
||||||
}
|
|
||||||
payload.write_int32(0); // unknown
|
payload.write_int32(0); // unknown
|
||||||
payload.write_int32(0); // number of private directories
|
payload.write_int32(0); // number of private directories
|
||||||
|
|
||||||
@@ -60,19 +47,35 @@ pub fn build_shared_file_list(dirs: &[SharedDirectory]) -> Message {
|
|||||||
/// Returns an empty listing if the payload is malformed.
|
/// Returns an empty listing if the payload is malformed.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn parse_shared_file_list(message: &mut Message) -> Vec<SharedDirectory> {
|
pub fn parse_shared_file_list(message: &mut Message) -> Vec<SharedDirectory> {
|
||||||
let pointer = message.get_pointer();
|
decompress_body(message)
|
||||||
let size = message.get_size();
|
.map_or_else(Vec::new, |mut body| read_directories(&mut body))
|
||||||
let compressed = message.get_slice(pointer, size);
|
}
|
||||||
let Ok(data) = deflate(&compressed) else {
|
|
||||||
return Vec::new();
|
|
||||||
};
|
|
||||||
|
|
||||||
let mut body = Message::new_with_data(data);
|
/// Write a directory listing in the form both code 5 and code 37 carry.
|
||||||
|
pub fn write_directories(payload: &mut Message, dirs: &[SharedDirectory]) {
|
||||||
|
payload.write_int32(dirs.len() as u32);
|
||||||
|
for dir in dirs {
|
||||||
|
payload
|
||||||
|
.write_string(&dir.name)
|
||||||
|
.write_int32(dir.files.len() as u32);
|
||||||
|
for (name, size) in &dir.files {
|
||||||
|
payload
|
||||||
|
.write_int8(1)
|
||||||
|
.write_string(name)
|
||||||
|
.write_int64(*size)
|
||||||
|
.write_string("") // extension
|
||||||
|
.write_int32(0); // attribute count
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Read back what [`write_directories`] wrote, stopping early when a hostile
|
||||||
|
/// count outruns the payload so a bogus length cannot spin into a huge
|
||||||
|
/// allocation loop.
|
||||||
|
pub fn read_directories(body: &mut Message) -> Vec<SharedDirectory> {
|
||||||
let dir_count = body.read_int32();
|
let dir_count = body.read_int32();
|
||||||
let mut dirs = Vec::new();
|
let mut dirs = Vec::new();
|
||||||
for _ in 0..dir_count {
|
for _ in 0..dir_count {
|
||||||
// Stop if a hostile count outruns the (decompressed) payload, so a
|
|
||||||
// bogus length can't spin us into a huge allocation loop.
|
|
||||||
if body.get_pointer() >= body.get_size() {
|
if body.get_pointer() >= body.get_size() {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
@@ -104,6 +107,15 @@ pub fn parse_shared_file_list(message: &mut Message) -> Vec<SharedDirectory> {
|
|||||||
dirs
|
dirs
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Inflate a compressed peer payload positioned at the blob.
|
||||||
|
pub fn decompress_body(message: &mut Message) -> Option<Message> {
|
||||||
|
let pointer = message.get_pointer();
|
||||||
|
let size = message.get_size();
|
||||||
|
deflate(&message.get_slice(pointer, size))
|
||||||
|
.ok()
|
||||||
|
.map(Message::new_with_data)
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn hostile_dir_count_does_not_hang() {
|
fn hostile_dir_count_does_not_hang() {
|
||||||
// A compressed body claiming ~4 billion directories with no data must
|
// A compressed body claiming ~4 billion directories with no data must
|
||||||
|
|||||||
@@ -3180,3 +3180,70 @@ fn a_search_reply_reports_no_free_slot_while_the_only_slot_is_taken() {
|
|||||||
|
|
||||||
let _ = std::fs::remove_dir_all(share);
|
let _ = std::fs::remove_dir_all(share);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// "Download folder" in SoulseekQt and Nicotine+ asks the sharer for that one
|
||||||
|
// folder (peer code 36) instead of the whole share; a sharer that never
|
||||||
|
// answers leaves the folder download hanging.
|
||||||
|
#[test]
|
||||||
|
fn a_third_party_client_fetches_one_folder_of_our_shares() {
|
||||||
|
let server = server_or_skip!();
|
||||||
|
|
||||||
|
let share_dir = unique_download_dir();
|
||||||
|
std::fs::create_dir_all(share_dir.join("album")).unwrap();
|
||||||
|
std::fs::create_dir_all(share_dir.join("other")).unwrap();
|
||||||
|
std::fs::write(share_dir.join("album").join("one.flac"), b"xxxx").unwrap();
|
||||||
|
std::fs::write(share_dir.join("album").join("two.flac"), b"yy").unwrap();
|
||||||
|
std::fs::write(share_dir.join("other").join("three.flac"), b"z").unwrap();
|
||||||
|
let root = share_dir.file_name().unwrap().to_str().unwrap().to_string();
|
||||||
|
|
||||||
|
let sharer_port = free_port().expect("sharer port");
|
||||||
|
let mut sharer = Client::with_settings(ClientSettings {
|
||||||
|
shared_directories: vec![share_dir.display().to_string()],
|
||||||
|
..server.listening_settings("e2e_folder_sharer", "pw", sharer_port)
|
||||||
|
});
|
||||||
|
sharer.connect().expect("sharer connect");
|
||||||
|
assert!(sharer.login().expect("sharer login"));
|
||||||
|
let server_addr = format!("{}:{}", server.host, server.port);
|
||||||
|
let _qt = login_raw(&server_addr, "e2e_folder_browser", "pw")
|
||||||
|
.expect("third-party client logs in");
|
||||||
|
|
||||||
|
let folder = format!("{root}\\album");
|
||||||
|
let mut p = connect_retry(
|
||||||
|
&format!("127.0.0.1:{sharer_port}"),
|
||||||
|
Duration::from_secs(5),
|
||||||
|
)
|
||||||
|
.expect("dial the sharer");
|
||||||
|
p.set_read_timeout(Some(Duration::from_secs(15))).unwrap();
|
||||||
|
p.write_all(&peer_init_bytes("e2e_folder_browser", "P", 0))
|
||||||
|
.unwrap();
|
||||||
|
let mut request = Message::new();
|
||||||
|
request
|
||||||
|
.write_int32(36)
|
||||||
|
.write_int32(77)
|
||||||
|
.write_string(&folder);
|
||||||
|
p.write_all(&request.get_buffer()).unwrap();
|
||||||
|
p.flush().unwrap();
|
||||||
|
|
||||||
|
let mut response = expect_code(&mut p, 37, Duration::from_secs(15))
|
||||||
|
.expect("the sharer answers with a FolderContentsResponse");
|
||||||
|
response.set_pointer(8);
|
||||||
|
let (token, echoed, directories) =
|
||||||
|
soulseek_rs::message::peer::parse_folder_contents(&mut response)
|
||||||
|
.expect("a well-formed folder listing");
|
||||||
|
assert_eq!(token, 77);
|
||||||
|
assert_eq!(echoed, folder);
|
||||||
|
let mut names: Vec<(String, Vec<String>)> = directories
|
||||||
|
.into_iter()
|
||||||
|
.map(|d| (d.name, d.files.into_iter().map(|(name, _)| name).collect()))
|
||||||
|
.collect();
|
||||||
|
for (_, files) in &mut names {
|
||||||
|
files.sort();
|
||||||
|
}
|
||||||
|
assert_eq!(
|
||||||
|
names,
|
||||||
|
[(folder, vec!["one.flac".to_string(), "two.flac".to_string()])],
|
||||||
|
"only the requested folder, with its files"
|
||||||
|
);
|
||||||
|
|
||||||
|
let _ = std::fs::remove_dir_all(share_dir);
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user