test(daemon): route the suite at its chokepoint and cover the protocol directly
Only 25 of the 83 CLI e2e tests actually reached a daemon: the helpers that build a command bypassed the one place that routed. Routing at TestServer::args instead brings that to 39 and, more importantly, puts the streaming tests through a daemon — event.room and event.message delivery to a remote client's mirror had no end-to-end coverage at all. Adds what the command surface structurally cannot reach: - Two clients against one daemon at once, which is the reason the hub exists and was proven nowhere in CI. - tests/daemon_protocol_e2e.rs, speaking the socket by hand: TCP token required and socket token not, pre-auth refusal, error ids, notification silence, the line cap, the handshake timeout, and that a call shaped by the published contract is accepted. - daemon status/stop/token as commands. - A TUI unit test for conversation seeding, the part of the attach path that had actually been broken. The protocol suite caught a regression in yesterday's hardening on its first run: the socket was shut down before the writer flushed a failed handshake's refusal, so a client with a wrong token saw the connection vanish instead of being told why.
This commit is contained in:
Michel de Graaf
@@ -108,6 +108,14 @@ jobs:
|
||||
SOULSEEK_E2E_REQUIRED: "1"
|
||||
run: cargo test -p soulseek-rs --test cli_e2e -- --nocapture
|
||||
|
||||
# The control protocol as a third party meets it: the TCP transport, the
|
||||
# token, and the gate. None of it is reachable through the command
|
||||
# surface, so the CLI suite cannot cover it in either mode.
|
||||
- name: Run daemon protocol e2e tests (required against soulfind)
|
||||
env:
|
||||
SOULSEEK_E2E_REQUIRED: "1"
|
||||
run: cargo test -p soulseek-rs --test daemon_protocol_e2e -- --nocapture --test-threads=1
|
||||
|
||||
# The same suite again, with every session served by a daemon instead of
|
||||
# the command's own login. Identical records and exit codes both ways is
|
||||
# what makes daemon mode a routing decision rather than a second
|
||||
|
||||
@@ -181,9 +181,13 @@ fn handle(stream: Stream, daemon: &Daemon) -> std::io::Result<()> {
|
||||
let writer = std::thread::spawn(move || write_loop(stream, &outbox));
|
||||
|
||||
if !authenticate(&mut reader, daemon, trusted, &out) {
|
||||
closer.shutdown();
|
||||
// Let the writer flush the refusal before closing: shutting the socket
|
||||
// first discards the reply, and the caller learns only that the
|
||||
// connection vanished — which is indistinguishable from the daemon
|
||||
// having crashed.
|
||||
drop(out);
|
||||
let _ = writer.join();
|
||||
closer.shutdown();
|
||||
return Ok(());
|
||||
}
|
||||
// An established client is expected to idle, holding the connection open
|
||||
@@ -237,10 +241,10 @@ fn handle(stream: Stream, daemon: &Daemon) -> std::io::Result<()> {
|
||||
}
|
||||
|
||||
daemon.hub.unsubscribe(subscription);
|
||||
closer.shutdown();
|
||||
drop(out);
|
||||
let _ = pump.join();
|
||||
let _ = writer.join();
|
||||
closer.shutdown();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -264,3 +264,227 @@ pub fn launch_main_tui(
|
||||
);
|
||||
tui.run(terminal)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::daemon::proto::ChatMessageDto;
|
||||
use crate::models::MessageDirection;
|
||||
|
||||
/// A session that knows a conversation and nothing else — what attaching
|
||||
/// to a daemon looks like from the TUI's side.
|
||||
struct TalkativeSession(Vec<ChatMessageDto>);
|
||||
|
||||
impl SessionApi for TalkativeSession {
|
||||
fn message_history(&self) -> Vec<ChatMessageDto> {
|
||||
self.0.clone()
|
||||
}
|
||||
fn username(&self) -> String {
|
||||
"tester".to_string()
|
||||
}
|
||||
fn listen_port(&self) -> Option<u16> {
|
||||
None
|
||||
}
|
||||
fn session_loss(&self) -> Option<soulseek_rs::SessionLoss> {
|
||||
None
|
||||
}
|
||||
fn search_with_cancel(
|
||||
&self,
|
||||
_query: &str,
|
||||
_timeout: Duration,
|
||||
_cancel: Option<Arc<std::sync::atomic::AtomicBool>>,
|
||||
) -> soulseek_rs::Result<Vec<soulseek_rs::SearchResult>> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
fn get_search_results(
|
||||
&self,
|
||||
_key: &str,
|
||||
) -> Vec<soulseek_rs::SearchResult> {
|
||||
Vec::new()
|
||||
}
|
||||
fn get_search_results_count(&self, _key: &str) -> usize {
|
||||
0
|
||||
}
|
||||
fn try_get_search_results(
|
||||
&self,
|
||||
_key: &str,
|
||||
) -> Option<Vec<soulseek_rs::SearchResult>> {
|
||||
None
|
||||
}
|
||||
fn start_wishlist_search(
|
||||
&self,
|
||||
_query: &str,
|
||||
) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn wishlist_interval(&self) -> Duration {
|
||||
Duration::from_mins(12)
|
||||
}
|
||||
fn download(
|
||||
&self,
|
||||
_filename: String,
|
||||
_username: String,
|
||||
_size: u64,
|
||||
_directory: String,
|
||||
) -> soulseek_rs::Result<(
|
||||
soulseek_rs::types::Download,
|
||||
std::sync::mpsc::Receiver<soulseek_rs::DownloadStatus>,
|
||||
)> {
|
||||
Err(soulseek_rs::SoulseekRs::NotConnected)
|
||||
}
|
||||
fn download_with_metadata(
|
||||
&self,
|
||||
_filename: String,
|
||||
_username: String,
|
||||
_size: u64,
|
||||
_directory: String,
|
||||
_metadata: soulseek_rs::types::DownloadMetadata,
|
||||
) -> soulseek_rs::Result<(
|
||||
soulseek_rs::types::Download,
|
||||
std::sync::mpsc::Receiver<soulseek_rs::DownloadStatus>,
|
||||
)> {
|
||||
Err(soulseek_rs::SoulseekRs::NotConnected)
|
||||
}
|
||||
fn get_all_downloads(&self) -> Vec<soulseek_rs::types::Download> {
|
||||
Vec::new()
|
||||
}
|
||||
fn pause_download(&self, _u: &str, _f: &str) -> bool {
|
||||
false
|
||||
}
|
||||
fn resume_download(&self, _u: &str, _f: &str) -> bool {
|
||||
false
|
||||
}
|
||||
fn remove_queued_download(&self, _u: &str, _f: &str) -> bool {
|
||||
false
|
||||
}
|
||||
fn remove_download(&self, _u: &str, _f: &str) -> bool {
|
||||
false
|
||||
}
|
||||
fn uploads(&self) -> Vec<soulseek_rs::UploadInfo> {
|
||||
Vec::new()
|
||||
}
|
||||
fn take_upload_events(&self) -> Vec<soulseek_rs::UploadInfo> {
|
||||
Vec::new()
|
||||
}
|
||||
fn cancel_upload(&self, _u: &str, _f: &str) -> bool {
|
||||
false
|
||||
}
|
||||
fn set_upload_slots(&self, _slots: usize) {}
|
||||
fn check_privileges(&self) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn own_privilege_seconds(&self) -> Option<u32> {
|
||||
None
|
||||
}
|
||||
fn request_room_list(&self) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn join_room(&self, _room: &str) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn leave_room(&self, _room: &str) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn say_in_room(
|
||||
&self,
|
||||
_room: &str,
|
||||
_message: &str,
|
||||
) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn room_members(&self, _room: &str) -> Vec<String> {
|
||||
Vec::new()
|
||||
}
|
||||
fn take_room_events(&self) -> Vec<soulseek_rs::RoomEvent> {
|
||||
Vec::new()
|
||||
}
|
||||
fn send_private_message(
|
||||
&self,
|
||||
_u: &str,
|
||||
_m: &str,
|
||||
) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn take_private_messages(&self) -> Vec<soulseek_rs::UserMessage> {
|
||||
Vec::new()
|
||||
}
|
||||
fn browse_user(&self, _username: &str) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn take_browse_result(
|
||||
&self,
|
||||
_username: &str,
|
||||
) -> Option<Vec<soulseek_rs::SharedDirectory>> {
|
||||
None
|
||||
}
|
||||
fn request_user_info(
|
||||
&self,
|
||||
_username: &str,
|
||||
) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn user_info(&self, _username: &str) -> Option<soulseek_rs::UserInfo> {
|
||||
None
|
||||
}
|
||||
fn shared_counts(&self) -> (u32, u32) {
|
||||
(0, 0)
|
||||
}
|
||||
fn shared_directories(&self) -> Vec<String> {
|
||||
Vec::new()
|
||||
}
|
||||
fn set_shared_directories(
|
||||
&self,
|
||||
_directories: Vec<String>,
|
||||
) -> soulseek_rs::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn tui(history: Vec<ChatMessageDto>) -> MainTui {
|
||||
MainTui::new(
|
||||
Arc::new(TalkativeSession(history)),
|
||||
"/tmp".to_string(),
|
||||
1,
|
||||
Duration::from_secs(1),
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_attached_session_opens_with_the_conversation_already_in_it() {
|
||||
// A TUI attached to a daemon has no state file of its own; the
|
||||
// conversation has to come from the session, or it starts blank while
|
||||
// the daemon has been collecting messages for hours.
|
||||
let tui = tui(vec![
|
||||
ChatMessageDto {
|
||||
peer: "bob".into(),
|
||||
outgoing: false,
|
||||
text: "hi".into(),
|
||||
at: 1,
|
||||
},
|
||||
ChatMessageDto {
|
||||
peer: "bob".into(),
|
||||
outgoing: true,
|
||||
text: "hello".into(),
|
||||
at: 2,
|
||||
},
|
||||
]);
|
||||
|
||||
assert_eq!(tui.state.messages.len(), 2);
|
||||
assert_eq!(tui.state.messages[0].peer, "bob");
|
||||
assert_eq!(tui.state.messages[0].direction, MessageDirection::Incoming);
|
||||
assert_eq!(
|
||||
tui.state.messages[1].direction,
|
||||
MessageDirection::Outgoing,
|
||||
"both halves of the conversation, not just what arrived"
|
||||
);
|
||||
assert_eq!(tui.state.chat_peers(), ["bob"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_local_session_starts_from_its_own_snapshot_instead() {
|
||||
// Locally the session knows no history and the state file is the whole
|
||||
// record, so nothing is invented.
|
||||
assert!(tui(Vec::new()).state.messages.is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
+185
-11
@@ -701,13 +701,34 @@ impl TestServer {
|
||||
format!("{}:{}", self.host, self.port)
|
||||
}
|
||||
|
||||
/// Arguments that point the binary at this server as `user`.
|
||||
/// Arguments that point the binary at this server as `user` — or, in
|
||||
/// daemon mode, at a daemon logged in as `user`.
|
||||
///
|
||||
/// Every helper that builds a command goes through here, so routing at
|
||||
/// this one point is what puts the streaming and wishlist tests through a
|
||||
/// daemon too, rather than only the ones that happen to use [`cli`].
|
||||
///
|
||||
/// The listener stays on: a Soulseek transfer is delivered over a
|
||||
/// connection the *uploader* opens back to us, so a downloader without a
|
||||
/// listener never receives its file.
|
||||
fn args(&self, user: &str) -> Vec<String> {
|
||||
self.args_on(user, free_port().expect("listener port"))
|
||||
self.args_with(user, &[])
|
||||
}
|
||||
|
||||
/// The same, with flags that configure the *session* rather than the
|
||||
/// command — shares, the listener, where downloads land. Locally they go
|
||||
/// to the run itself; in daemon mode they configure the daemon, because
|
||||
/// that is whose session they now describe.
|
||||
fn args_with(&self, user: &str, session_flags: &[String]) -> Vec<String> {
|
||||
match mode() {
|
||||
Mode::Local => {
|
||||
let mut args =
|
||||
self.args_on(user, free_port().expect("listener port"));
|
||||
args.extend(session_flags.iter().cloned());
|
||||
args
|
||||
}
|
||||
Mode::Daemon => self.daemon_args(user, session_flags),
|
||||
}
|
||||
}
|
||||
|
||||
/// The same arguments on a chosen listener port, for the tests about what
|
||||
@@ -746,6 +767,19 @@ impl TestServer {
|
||||
]
|
||||
}
|
||||
|
||||
/// The state directory of the daemon serving `user`, where its socket and
|
||||
/// token live.
|
||||
fn daemon_state(&self, user: &str) -> PathBuf {
|
||||
self.daemon_for(user, &[]);
|
||||
let key = format!("{}|{user}", self.address());
|
||||
let daemons = self.daemons.lock().expect("daemon registry");
|
||||
daemons
|
||||
.get(&key)
|
||||
.expect("the daemon was just started")
|
||||
.state
|
||||
.clone()
|
||||
}
|
||||
|
||||
fn daemon_for(&self, user: &str, session_flags: &[String]) -> PathBuf {
|
||||
let key = format!("{}|{user}", self.address());
|
||||
let mut daemons = self.daemons.lock().expect("daemon registry");
|
||||
@@ -893,17 +927,14 @@ fn probe_bytes() -> Vec<u8> {
|
||||
/// the whole point: if the records and exit code come out the same either way,
|
||||
/// daemon mode is equivalent.
|
||||
fn cli(server: &TestServer, user: &str, args: &[&str]) -> Output {
|
||||
let mut all = match mode() {
|
||||
Mode::Local => server.args(user),
|
||||
Mode::Daemon => {
|
||||
let (session_flags, _) = split_session_flags(args);
|
||||
server.daemon_args(user, &session_flags)
|
||||
let (session_flags, command_args) = match mode() {
|
||||
// Locally a flag is a flag: the run configures its own session.
|
||||
Mode::Local => {
|
||||
(Vec::new(), args.iter().map(|a| (*a).to_string()).collect())
|
||||
}
|
||||
Mode::Daemon => split_session_flags(args),
|
||||
};
|
||||
let command_args: Vec<String> = match mode() {
|
||||
Mode::Local => args.iter().map(|a| (*a).to_string()).collect(),
|
||||
Mode::Daemon => split_session_flags(args).1,
|
||||
};
|
||||
let mut all = server.args_with(user, &session_flags);
|
||||
all.extend(command_args);
|
||||
let refs: Vec<&str> = all.iter().map(String::as_str).collect();
|
||||
run(&refs)
|
||||
@@ -3221,3 +3252,146 @@ fn concurrent_downloads_sharing_a_listener_port_all_land() {
|
||||
assert_eq!(written, probe_bytes(), "run {i} wrote the wrong bytes");
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Daemon mode
|
||||
//
|
||||
// Everything above proves a command behaves the same whichever kind of session
|
||||
// serves it. These prove the things that only exist *because* there is a
|
||||
// daemon: one session shared by several clients at once, the remote transport
|
||||
// and its token, and the gate that decides who may talk to it at all.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Two clients, one session, at the same time.
|
||||
///
|
||||
/// This is the whole reason the daemon exists: the server allows one login per
|
||||
/// account, so without it a second command would displace the first and both
|
||||
/// would start reporting a lost session.
|
||||
#[test]
|
||||
fn two_clients_share_one_daemon_session_at_once() {
|
||||
if mode() != Mode::Daemon {
|
||||
println!("skipped: this is a daemon-mode property");
|
||||
return;
|
||||
}
|
||||
let server = server_or_skip!();
|
||||
let share = Scratch::new("share");
|
||||
std::fs::write(share.path().join("cli_probe_shared.bin"), probe_bytes())
|
||||
.expect("share file");
|
||||
let _sharer =
|
||||
server.client("cli_e2e_two_client_peer", vec![share.display()]);
|
||||
settle();
|
||||
|
||||
// One daemon, because both runs use the same account.
|
||||
let args = server.args("cli_e2e_two_clients");
|
||||
let spawn = |extra: &[&str]| {
|
||||
let mut all = args.clone();
|
||||
all.extend(extra.iter().map(|a| (*a).to_string()));
|
||||
let refs: Vec<&str> = all.iter().map(String::as_str).collect();
|
||||
command(&refs)
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()
|
||||
.expect("the binary should start")
|
||||
};
|
||||
|
||||
let first = spawn(&["--json", "--search-timeout", "8", "search", "shared"]);
|
||||
let second = spawn(&["--json", "browse", "cli_e2e_two_client_peer"]);
|
||||
|
||||
let first = first.wait_with_output().expect("wait");
|
||||
let second = second.wait_with_output().expect("wait");
|
||||
|
||||
assert_eq!(
|
||||
code(&first),
|
||||
EXIT_OK,
|
||||
"the search failed while another client was attached: {}",
|
||||
stderr(&first)
|
||||
);
|
||||
assert_eq!(
|
||||
code(&second),
|
||||
EXIT_OK,
|
||||
"the browse failed while another client was attached: {}",
|
||||
stderr(&second)
|
||||
);
|
||||
assert!(
|
||||
!records(&first).is_empty() && !records(&second).is_empty(),
|
||||
"both clients should have got answers"
|
||||
);
|
||||
// Neither may report the session being taken over: that is exactly what a
|
||||
// second *login* would have caused, and what borrowing one avoids.
|
||||
for output in [&first, &second] {
|
||||
assert_ne!(code(output), EXIT_SESSION_LOST, "{}", stderr(output));
|
||||
}
|
||||
}
|
||||
|
||||
/// A daemon reports itself, and stops when asked.
|
||||
#[test]
|
||||
fn a_daemon_reports_its_session_and_stops_on_request() {
|
||||
if mode() != Mode::Daemon {
|
||||
println!("skipped: there is no daemon to ask in local mode");
|
||||
return;
|
||||
}
|
||||
let server = server_or_skip!();
|
||||
// A daemon of its own: this test stops it, and the shared fixtures are
|
||||
// still serving the rest of the suite.
|
||||
let user = "cli_e2e_daemon_status";
|
||||
let state = server.daemon_state(user);
|
||||
let socket = state.join("daemon.sock");
|
||||
let at = socket.display().to_string();
|
||||
|
||||
let status =
|
||||
run(&["--no-config", "--json", "--daemon", &at, "daemon", "status"]);
|
||||
assert_eq!(code(&status), EXIT_OK, "stderr: {}", stderr(&status));
|
||||
let record: serde_json::Value =
|
||||
serde_json::from_str(records(&status).first().expect("one record"))
|
||||
.expect("valid JSON");
|
||||
assert_eq!(record["user"], user);
|
||||
assert_eq!(record["server"], server.address());
|
||||
assert!(
|
||||
record["clients"].as_u64().is_some_and(|n| n >= 1),
|
||||
"the asking client counts as attached: {record}"
|
||||
);
|
||||
|
||||
let stopped = run(&["--no-config", "--daemon", &at, "daemon", "stop"]);
|
||||
assert_eq!(code(&stopped), EXIT_OK, "stderr: {}", stderr(&stopped));
|
||||
|
||||
// The socket goes away, and a command that finds nothing there says so
|
||||
// rather than hanging.
|
||||
let deadline = Instant::now() + Duration::from_secs(20);
|
||||
while socket.exists() && Instant::now() < deadline {
|
||||
std::thread::sleep(Duration::from_millis(200));
|
||||
}
|
||||
let after = run(&["--no-config", "--daemon", &at, "daemon", "status"]);
|
||||
assert_eq!(
|
||||
code(&after),
|
||||
EXIT_CONNECTION,
|
||||
"a stopped daemon is a connection error, not a hang: {}",
|
||||
stderr(&after)
|
||||
);
|
||||
}
|
||||
|
||||
/// `daemon token` prints a secret, and the same one every time.
|
||||
#[test]
|
||||
fn the_daemon_token_is_stable_across_reads() {
|
||||
let state = Scratch::new("token");
|
||||
let read = || {
|
||||
let mut command = Command::new(BIN);
|
||||
for name in CLI_ENV_VARS {
|
||||
command.env_remove(name);
|
||||
}
|
||||
command.env("SOULSEEK_STATE_DIR", state.path());
|
||||
command.current_dir(std::env::temp_dir());
|
||||
command.args(["--no-config", "daemon", "token"]);
|
||||
command.output().expect("the binary should run")
|
||||
};
|
||||
|
||||
let first = read();
|
||||
assert_eq!(code(&first), EXIT_OK, "stderr: {}", stderr(&first));
|
||||
let token = records(&first).first().expect("a token").clone();
|
||||
assert_eq!(token.len(), 64, "256 bits, hex encoded");
|
||||
assert!(token.chars().all(|c| c.is_ascii_hexdigit()));
|
||||
assert_eq!(
|
||||
records(&read()).first().expect("a token"),
|
||||
&token,
|
||||
"a restart must not lock existing clients out"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,494 @@
|
||||
//! The control protocol as a third party meets it: a socket, JSON-RPC lines,
|
||||
//! and the gate that decides who gets past `auth`.
|
||||
//!
|
||||
//! The CLI suite drives the daemon through this binary, which means it only
|
||||
//! ever sends well-formed requests over a Unix socket. Everything here is the
|
||||
//! other half — the TCP transport and its token, and what happens to a caller
|
||||
//! who is hostile, broken, or simply not authenticated. None of it is
|
||||
//! reachable through the command surface, and all of it is load-bearing: on
|
||||
//! the socket, connecting *is* authenticating.
|
||||
//!
|
||||
//! Like the rest of the e2e suite this needs a soulfind server and skips
|
||||
//! without one; `SOULSEEK_E2E_REQUIRED=1` turns that skip into a failure.
|
||||
|
||||
use std::io::{BufRead, BufReader, Write};
|
||||
use std::net::TcpStream;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
const BIN: &str = env!("CARGO_BIN_EXE_soulseek-rs");
|
||||
|
||||
/// Every environment variable the binary reads, cleared so a developer's shell
|
||||
/// cannot change what a test observes — and so no run finds a real daemon.
|
||||
const CLI_ENV_VARS: [&str; 15] = [
|
||||
"SOULSEEK_CONFIG_DIR",
|
||||
"SOULSEEK_STATE_DIR",
|
||||
"SOULSEEK_DAEMON",
|
||||
"SOULSEEK_DAEMON_TOKEN",
|
||||
"SOULSEEK_USERNAME",
|
||||
"SOULSEEK_PASSWORD",
|
||||
"SOULSEEK_PASSWORD_CMD",
|
||||
"SOULSEEK_SERVER",
|
||||
"SOULSEEK_NO_LISTENER",
|
||||
"SOULSEEK_LISTENER_PORT",
|
||||
"SOULSEEK_DOWNLOAD_DIR",
|
||||
"SOULSEEK_SHARED_DIR",
|
||||
"SOULSEEK_MAX_CONCURRENT_DOWNLOADS",
|
||||
"SOULSEEK_SEARCH_TIMEOUT",
|
||||
"SOULSEEK_CONFIG",
|
||||
];
|
||||
|
||||
/// A daemon bound to loopback TCP as well as its socket, so both transports
|
||||
/// can be exercised against one session.
|
||||
struct Daemon {
|
||||
child: Child,
|
||||
state: PathBuf,
|
||||
port: u16,
|
||||
_server: Soulfind,
|
||||
}
|
||||
|
||||
impl Daemon {
|
||||
fn start() -> Option<Self> {
|
||||
let server = Soulfind::start()?;
|
||||
let state = scratch("daemon-proto");
|
||||
let port = free_port()?;
|
||||
|
||||
let mut command = Command::new(BIN);
|
||||
for name in CLI_ENV_VARS {
|
||||
command.env_remove(name);
|
||||
}
|
||||
command.env("SOULSEEK_STATE_DIR", &state);
|
||||
command.current_dir(std::env::temp_dir());
|
||||
command.args([
|
||||
"--no-config",
|
||||
"--server",
|
||||
&server.address(),
|
||||
"--username",
|
||||
"cli_e2e_proto",
|
||||
"--password",
|
||||
"pw",
|
||||
"--listener-port",
|
||||
&free_port()?.to_string(),
|
||||
"daemon",
|
||||
"--bind",
|
||||
&format!("127.0.0.1:{port}"),
|
||||
]);
|
||||
command.stdout(Stdio::null()).stderr(Stdio::null());
|
||||
let child = command.spawn().ok()?;
|
||||
|
||||
let daemon = Self {
|
||||
child,
|
||||
state,
|
||||
port,
|
||||
_server: server,
|
||||
};
|
||||
daemon.wait_until_ready()?;
|
||||
Some(daemon)
|
||||
}
|
||||
|
||||
/// Ready means logged in, not merely bound: the daemon binds before it
|
||||
/// logs in, and a client that connects between the two would be asking
|
||||
/// about a session that is not there yet.
|
||||
fn wait_until_ready(&self) -> Option<()> {
|
||||
let deadline = Instant::now() + Duration::from_mins(1);
|
||||
while Instant::now() < deadline {
|
||||
if let Some(mut connection) = self.connect_tcp()
|
||||
&& let Some(reply) = connection.call(&format!(
|
||||
r#"{{"jsonrpc":"2.0","id":1,"method":"auth","params":{{"protocol":1,"token":"{}"}}}}"#,
|
||||
self.token()
|
||||
))
|
||||
&& reply.contains("\"username\"")
|
||||
{
|
||||
return Some(());
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(250));
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn token(&self) -> String {
|
||||
std::fs::read_to_string(self.state.join("daemon.token"))
|
||||
.map(|token| token.trim().to_string())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn socket_path(&self) -> PathBuf {
|
||||
self.state.join("daemon.sock")
|
||||
}
|
||||
|
||||
fn connect_tcp(&self) -> Option<Connection> {
|
||||
TcpStream::connect(("127.0.0.1", self.port))
|
||||
.ok()
|
||||
.map(|stream| {
|
||||
let read = stream.try_clone().expect("a second handle");
|
||||
Connection::from_parts(Box::new(read), Box::new(stream))
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn connect_socket(&self) -> Option<Connection> {
|
||||
let stream =
|
||||
std::os::unix::net::UnixStream::connect(self.socket_path()).ok()?;
|
||||
let read = stream.try_clone().expect("a second handle");
|
||||
Some(Connection::from_parts(Box::new(read), Box::new(stream)))
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Daemon {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.child.kill();
|
||||
let _ = self.child.wait();
|
||||
let _ = std::fs::remove_dir_all(&self.state);
|
||||
}
|
||||
}
|
||||
|
||||
/// One control connection, spoken to by hand.
|
||||
struct Connection {
|
||||
reader: BufReader<Box<dyn std::io::Read + Send>>,
|
||||
writer: Box<dyn Write + Send>,
|
||||
}
|
||||
|
||||
impl Connection {
|
||||
/// Boxed halves, so one type covers both transports.
|
||||
fn from_parts(
|
||||
read: Box<dyn std::io::Read + Send>,
|
||||
write: Box<dyn Write + Send>,
|
||||
) -> Self {
|
||||
Self {
|
||||
reader: BufReader::new(read),
|
||||
writer: write,
|
||||
}
|
||||
}
|
||||
|
||||
/// Send one line and read the reply, `None` if the daemon hung up first.
|
||||
fn call(&mut self, line: &str) -> Option<String> {
|
||||
self.send(line)?;
|
||||
self.read()
|
||||
}
|
||||
|
||||
fn send(&mut self, line: &str) -> Option<()> {
|
||||
writeln!(self.writer, "{line}").ok()?;
|
||||
self.writer.flush().ok()
|
||||
}
|
||||
|
||||
fn read(&mut self) -> Option<String> {
|
||||
let mut reply = String::new();
|
||||
match self.reader.read_line(&mut reply) {
|
||||
Ok(0) | Err(_) => None,
|
||||
Ok(_) => Some(reply),
|
||||
}
|
||||
}
|
||||
|
||||
fn authenticate(&mut self, token: Option<&str>) -> Option<String> {
|
||||
let params = match token {
|
||||
Some(token) => format!(r#"{{"protocol":1,"token":"{token}"}}"#),
|
||||
None => r#"{"protocol":1}"#.to_string(),
|
||||
};
|
||||
self.call(&format!(
|
||||
r#"{{"jsonrpc":"2.0","id":1,"method":"auth","params":{params}}}"#
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
macro_rules! daemon_or_skip {
|
||||
() => {
|
||||
match Daemon::start() {
|
||||
Some(daemon) => daemon,
|
||||
None => {
|
||||
let required = std::env::var("SOULSEEK_E2E_REQUIRED")
|
||||
.is_ok_and(|v| v != "0" && !v.is_empty());
|
||||
assert!(
|
||||
!required,
|
||||
"SOULSEEK_E2E_REQUIRED is set but no daemon could be \
|
||||
started (set SOULFIND_BIN)"
|
||||
);
|
||||
println!(
|
||||
"daemon protocol e2e skipped: no soulfind server (set \
|
||||
SOULFIND_BIN to run)"
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tcp_needs_the_token_and_the_socket_does_not() {
|
||||
let daemon = daemon_or_skip!();
|
||||
|
||||
let mut anonymous = daemon.connect_tcp().expect("connect");
|
||||
let refused = anonymous.authenticate(None).expect("an answer");
|
||||
assert!(
|
||||
refused.contains("\"error\""),
|
||||
"TCP without a token must be refused: {refused}"
|
||||
);
|
||||
|
||||
let mut wrong = daemon.connect_tcp().expect("connect");
|
||||
let bad = wrong
|
||||
.authenticate(Some(&"0".repeat(64)))
|
||||
.expect("an answer");
|
||||
assert!(
|
||||
bad.contains("\"error\""),
|
||||
"a wrong token must be refused: {bad}"
|
||||
);
|
||||
|
||||
let mut good = daemon.connect_tcp().expect("connect");
|
||||
let accepted = good.authenticate(Some(&daemon.token())).expect("an answer");
|
||||
assert!(
|
||||
accepted.contains("\"result\"") && accepted.contains("cli_e2e_proto"),
|
||||
"the right token must be accepted: {accepted}"
|
||||
);
|
||||
|
||||
#[cfg(unix)]
|
||||
{
|
||||
// The socket's permissions have already answered the question.
|
||||
let mut local = daemon.connect_socket().expect("connect");
|
||||
let reply = local.authenticate(None).expect("an answer");
|
||||
assert!(
|
||||
reply.contains("\"result\""),
|
||||
"a Unix socket needs no token: {reply}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nothing_can_be_asked_before_authenticating() {
|
||||
let daemon = daemon_or_skip!();
|
||||
let mut connection = daemon.connect_tcp().expect("connect");
|
||||
|
||||
let reply = connection
|
||||
.call(r#"{"jsonrpc":"2.0","id":1,"method":"daemon.status"}"#)
|
||||
.expect("an answer");
|
||||
assert!(
|
||||
reply.contains("\"error\""),
|
||||
"a request before auth must be refused: {reply}"
|
||||
);
|
||||
assert!(
|
||||
!reply.contains("cli_e2e_proto"),
|
||||
"an unauthenticated caller must learn nothing about the session: \
|
||||
{reply}"
|
||||
);
|
||||
assert!(
|
||||
connection
|
||||
.call(r#"{"jsonrpc":"2.0","id":2,"method":"daemon.status"}"#)
|
||||
.is_none(),
|
||||
"the connection must be dropped, not left open to try again"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_error_carries_an_id_so_a_client_can_match_it() {
|
||||
let daemon = daemon_or_skip!();
|
||||
let mut connection = daemon.connect_tcp().expect("connect");
|
||||
connection
|
||||
.authenticate(Some(&daemon.token()))
|
||||
.expect("authenticated");
|
||||
|
||||
let reply = connection
|
||||
.call(r#"{"jsonrpc":"2.0","id":7,"method":"no.such.method"}"#)
|
||||
.expect("an answer");
|
||||
assert!(reply.contains("\"id\":7"), "{reply}");
|
||||
assert!(reply.contains("-32601"), "{reply}");
|
||||
|
||||
// A malformed line still gets an id — null — rather than looking like a
|
||||
// notification the client should ignore.
|
||||
let malformed = connection.call("{ this is not json").expect("an answer");
|
||||
assert!(malformed.contains("\"id\":null"), "{malformed}");
|
||||
assert!(malformed.contains("-32700"), "{malformed}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_notification_is_answered_with_silence() {
|
||||
let daemon = daemon_or_skip!();
|
||||
let mut connection = daemon.connect_tcp().expect("connect");
|
||||
connection
|
||||
.authenticate(Some(&daemon.token()))
|
||||
.expect("authenticated");
|
||||
|
||||
// No id, so no reply — the next line read must be the answer to the
|
||||
// request that follows it, not a reply to this one.
|
||||
connection
|
||||
.send(r#"{"jsonrpc":"2.0","method":"no.such.method"}"#)
|
||||
.expect("sent");
|
||||
let reply = connection
|
||||
.call(r#"{"jsonrpc":"2.0","id":9,"method":"daemon.status"}"#)
|
||||
.expect("an answer");
|
||||
assert!(
|
||||
reply.contains("\"id\":9"),
|
||||
"a notification must not produce a reply of its own: {reply}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_oversized_line_cannot_make_the_daemon_allocate() {
|
||||
let daemon = daemon_or_skip!();
|
||||
let mut connection = daemon.connect_tcp().expect("connect");
|
||||
|
||||
// Over the 1 MiB cap, and never terminated: an unauthenticated peer must
|
||||
// not be able to make the daemon buffer without bound.
|
||||
let flood = "x".repeat(2 * 1024 * 1024);
|
||||
let _ = connection.send(&flood);
|
||||
assert!(
|
||||
connection.read().is_none(),
|
||||
"an oversized request must end the connection"
|
||||
);
|
||||
|
||||
// The daemon is still serving everyone else.
|
||||
let mut healthy = daemon.connect_tcp().expect("connect");
|
||||
let reply = healthy
|
||||
.authenticate(Some(&daemon.token()))
|
||||
.expect("an answer");
|
||||
assert!(reply.contains("\"result\""), "{reply}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_silent_connection_is_closed_rather_than_held_open() {
|
||||
let daemon = daemon_or_skip!();
|
||||
let mut connection = daemon.connect_tcp().expect("connect");
|
||||
|
||||
// Says nothing at all. The handshake timeout is 30s; well inside a minute.
|
||||
let deadline = Instant::now() + Duration::from_mins(1);
|
||||
let mut closed = false;
|
||||
while Instant::now() < deadline {
|
||||
if connection.read().is_none() {
|
||||
closed = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert!(
|
||||
closed,
|
||||
"a peer that connects and never authenticates must not hold a thread \
|
||||
forever"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn discover_serves_a_contract_whose_parameters_match_what_is_accepted() {
|
||||
let daemon = daemon_or_skip!();
|
||||
let mut connection = daemon.connect_tcp().expect("connect");
|
||||
connection
|
||||
.authenticate(Some(&daemon.token()))
|
||||
.expect("authenticated");
|
||||
|
||||
let reply = connection
|
||||
.call(r#"{"jsonrpc":"2.0","id":1,"method":"rpc.discover"}"#)
|
||||
.expect("an answer");
|
||||
let document: serde_json::Value =
|
||||
serde_json::from_str(&reply).expect("valid JSON");
|
||||
let methods = document["result"]["methods"]
|
||||
.as_array()
|
||||
.expect("the contract lists methods");
|
||||
|
||||
// The published parameter names have to be the keys the daemon reads.
|
||||
// Published as one descriptor named `params`, a generated client would
|
||||
// send {"params":{"params":{…}}} and every call would fail -32602 — which
|
||||
// is exactly what this asserts cannot happen.
|
||||
let join = methods
|
||||
.iter()
|
||||
.find(|m| m["name"] == "room.join")
|
||||
.expect("room.join is published");
|
||||
let names: Vec<&str> = join["params"]
|
||||
.as_array()
|
||||
.expect("params is an array")
|
||||
.iter()
|
||||
.map(|p| p["name"].as_str().expect("a name"))
|
||||
.collect();
|
||||
assert_eq!(names, ["room"], "{join}");
|
||||
|
||||
// And a call built from that description is accepted.
|
||||
let accepted = connection
|
||||
.call(
|
||||
r#"{"jsonrpc":"2.0","id":2,"method":"room.join","params":{"room":"cli_e2e_proto_room"}}"#,
|
||||
)
|
||||
.expect("an answer");
|
||||
assert!(
|
||||
accepted.contains("\"result\""),
|
||||
"a call shaped by the published contract must be accepted: {accepted}"
|
||||
);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Fixtures
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
struct Soulfind {
|
||||
child: Option<Child>,
|
||||
db: PathBuf,
|
||||
port: u16,
|
||||
}
|
||||
|
||||
impl Soulfind {
|
||||
fn start() -> Option<Self> {
|
||||
let bin = soulfind_binary()?;
|
||||
let port = free_port()?;
|
||||
let db = std::env::temp_dir().join(format!("soulfind-proto-{port}.db"));
|
||||
let _ = std::fs::remove_file(&db);
|
||||
|
||||
let child = Command::new(bin)
|
||||
.args(["-d", db.to_str()?, "-p", &port.to_string()])
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::null())
|
||||
.spawn()
|
||||
.ok()?;
|
||||
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
while Instant::now() < deadline {
|
||||
if TcpStream::connect(("127.0.0.1", port)).is_ok() {
|
||||
return Some(Self {
|
||||
child: Some(child),
|
||||
db,
|
||||
port,
|
||||
});
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
}
|
||||
let mut child = child;
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
None
|
||||
}
|
||||
|
||||
fn address(&self) -> String {
|
||||
format!("127.0.0.1:{}", self.port)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Soulfind {
|
||||
fn drop(&mut self) {
|
||||
if let Some(child) = self.child.as_mut() {
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
}
|
||||
let _ = std::fs::remove_file(&self.db);
|
||||
}
|
||||
}
|
||||
|
||||
fn soulfind_binary() -> Option<PathBuf> {
|
||||
if let Ok(path) = std::env::var("SOULFIND_BIN") {
|
||||
let path = PathBuf::from(path);
|
||||
return path.exists().then_some(path);
|
||||
}
|
||||
Path::new(env!("CARGO_MANIFEST_DIR"))
|
||||
.ancestors()
|
||||
.map(|dir| dir.join("soulfind/bin/soulfind"))
|
||||
.find(|candidate| candidate.exists())
|
||||
}
|
||||
|
||||
fn free_port() -> Option<u16> {
|
||||
std::net::TcpListener::bind("127.0.0.1:0")
|
||||
.ok()?
|
||||
.local_addr()
|
||||
.ok()
|
||||
.map(|addr| addr.port())
|
||||
}
|
||||
|
||||
fn scratch(label: &str) -> PathBuf {
|
||||
static COUNTER: std::sync::atomic::AtomicU32 =
|
||||
std::sync::atomic::AtomicU32::new(0);
|
||||
let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
let path = std::env::temp_dir()
|
||||
.join(format!("soulseek-{label}-{}-{n}", std::process::id()));
|
||||
std::fs::create_dir_all(&path).expect("scratch directory");
|
||||
path
|
||||
}
|
||||
Reference in New Issue
Block a user