diff options
| -rw-r--r-- | mumctl/src/main.rs | 34 | ||||
| -rw-r--r-- | mumd/src/command.rs | 57 | ||||
| -rw-r--r-- | mumd/src/main.rs | 11 | ||||
| -rw-r--r-- | mumd/src/network/tcp.rs | 6 | ||||
| -rw-r--r-- | mumd/src/network/udp.rs | 49 | ||||
| -rw-r--r-- | mumd/src/state.rs | 55 | ||||
| -rw-r--r-- | mumlib/src/command.rs | 22 | ||||
| -rw-r--r-- | mumlib/src/config.rs | 14 | ||||
| -rw-r--r-- | mumlib/src/lib.rs | 1 |
9 files changed, 209 insertions, 40 deletions
diff --git a/mumctl/src/main.rs b/mumctl/src/main.rs index cbad4c6..11c51e6 100644 --- a/mumctl/src/main.rs +++ b/mumctl/src/main.rs @@ -74,7 +74,8 @@ fn main() { ) .subcommand( SubCommand::with_name("remove").arg(Arg::with_name("name").required(true)), - ), + ) + .subcommand(SubCommand::with_name("list")), ) .subcommand( SubCommand::with_name("channel") @@ -126,6 +127,35 @@ fn main() { match_server_remove(matches, &mut config); } else if let Some(matches) = matches.subcommand_matches("add") { match_server_add(matches, &mut config); + } else if let Some(_) = matches.subcommand_matches("list") { + let servers = config + .as_ref() + .map(|e| e.servers.as_ref().map(|e| e.clone()).unwrap_or(Vec::new())) + .unwrap_or(Vec::new()); + if servers.len() == 0 { + println!("{} No servers in config", "warning:".yellow()); + } + for (server, response) in servers + .into_iter() + .map(|e| { + let response = send_command(Command::ServerStatus { + host: e.host.clone(), + port: e.port.unwrap_or(mumlib::DEFAULT_PORT), + }); + (e, response) + }) + .filter(|e| e.1.is_ok()) + .map(|e| (e.0, e.1.unwrap().unwrap())) + { + if let CommandResponse::ServerStatus { + users, max_users, .. + } = response + { + println!("{} [{}/{}]", server.name, users, max_users) + } else { + unreachable!() + } + } } } else if let Some(matches) = matches.subcommand_matches("channel") { if let Some(_matches) = matches.subcommand_matches("list") { @@ -224,7 +254,7 @@ fn match_server_connect(matches: &clap::ArgMatches<'_>, config: &mumlib::config: let host = matches.value_of("host").unwrap(); let username = matches.value_of("username"); let port = match matches.value_of("port").map(|e| e.parse()) { - None => Some(64738), + None => Some(mumlib::DEFAULT_PORT), Some(Err(_)) => None, Some(Ok(v)) => Some(v), }; diff --git a/mumd/src/command.rs b/mumd/src/command.rs index d4b25d0..330e3fc 100644 --- a/mumd/src/command.rs +++ b/mumd/src/command.rs @@ -1,9 +1,11 @@ -use crate::state::State; +use crate::state::{ExecutionContext, State}; use crate::network::tcp::{TcpEvent, TcpEventCallback}; use ipc_channel::ipc::IpcSender; use log::*; +use mumble_protocol::ping::PongPacket; use mumlib::command::{Command, CommandResponse}; +use std::net::SocketAddr; use std::sync::{Arc, Mutex}; use tokio::sync::{mpsc, oneshot}; @@ -14,28 +16,51 @@ pub async fn handle( IpcSender<mumlib::error::Result<Option<CommandResponse>>>, )>, tcp_event_register_sender: mpsc::UnboundedSender<(TcpEvent, TcpEventCallback)>, + ping_request_sender: mpsc::UnboundedSender<(u64, SocketAddr, Box<dyn FnOnce(PongPacket)>)>, ) { debug!("Begin listening for commands"); while let Some((command, response_sender)) = command_receiver.recv().await { debug!("Received command {:?}", command); let mut state = state.lock().unwrap(); - let (event, generator) = state.handle_command(command); + let event = state.handle_command(command); drop(state); - if let Some(event) = event { - let (tx, rx) = oneshot::channel(); - //TODO handle this error - let _ = tcp_event_register_sender.send(( - event, - Box::new(move |e| { - let response = generator(Some(e)); - response_sender.send(response).unwrap(); - tx.send(()).unwrap(); - }), - )); + match event { + ExecutionContext::TcpEvent(event, generator) => { + let (tx, rx) = oneshot::channel(); + //TODO handle this error + let _ = tcp_event_register_sender.send(( + event, + Box::new(move |e| { + let response = generator(e); + response_sender.send(response).unwrap(); + tx.send(()).unwrap(); + }), + )); - rx.await.unwrap(); - } else { - response_sender.send(generator(None)).unwrap(); + rx.await.unwrap(); + } + ExecutionContext::Now(generator) => { + response_sender.send(generator()).unwrap(); + } + ExecutionContext::Ping(generator, converter) => { + match generator() { + Ok(addr) => { + let res = ping_request_sender.send(( + 0, + addr, + Box::new(move |packet| { + response_sender.send(converter(packet)).unwrap(); + }), + )); + if res.is_err() { + panic!(); + } + } + Err(e) => { + response_sender.send(Err(e)).unwrap(); + } + }; + } } } } diff --git a/mumd/src/main.rs b/mumd/src/main.rs index 37ff0dd..b83299f 100644 --- a/mumd/src/main.rs +++ b/mumd/src/main.rs @@ -36,11 +36,12 @@ async fn main() { let (connection_info_sender, connection_info_receiver) = watch::channel::<Option<ConnectionInfo>>(None); let (response_sender, response_receiver) = mpsc::unbounded_channel(); + let (ping_request_sender, ping_request_receiver) = mpsc::unbounded_channel(); let state = State::new(packet_sender, connection_info_sender); let state = Arc::new(Mutex::new(state)); - let (_, _, _, e) = join!( + let (_, _, _, e, _) = join!( network::tcp::handle( Arc::clone(&state), connection_info_receiver.clone(), @@ -53,11 +54,17 @@ async fn main() { connection_info_receiver.clone(), crypt_state_receiver, ), - command::handle(state, command_receiver, response_sender), + command::handle( + state, + command_receiver, + response_sender, + ping_request_sender, + ), spawn_blocking(move || { // IpcSender is blocking receive_oneshot_commands(command_sender); }), + network::udp::handle_pings(ping_request_receiver), ); e.unwrap(); } diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs index cd11690..131f066 100644 --- a/mumd/src/network/tcp.rs +++ b/mumd/src/network/tcp.rs @@ -27,7 +27,7 @@ type TcpSender = SplitSink< type TcpReceiver = SplitStream<Framed<TlsStream<TcpStream>, ControlCodec<Serverbound, Clientbound>>>; -pub(crate) type TcpEventCallback = Box<dyn FnOnce(&TcpEventData)>; +pub(crate) type TcpEventCallback = Box<dyn FnOnce(TcpEventData)>; #[derive(Debug, Clone, Hash, Eq, PartialEq)] pub enum TcpEvent { @@ -228,7 +228,7 @@ async fn listen( if let Some(vec) = event_queue.lock().unwrap().get_mut(&TcpEvent::Connected) { let old = std::mem::take(vec); for handler in old { - handler(&TcpEventData::Connected(&msg)); + handler(TcpEventData::Connected(&msg)); } } let mut state = state.lock().unwrap(); @@ -282,7 +282,7 @@ async fn listen( if let Some(vec) = event_queue.lock().unwrap().get_mut(&TcpEvent::Disconnected) { let old = std::mem::take(vec); for handler in old { - handler(&TcpEventData::Disconnected); + handler(TcpEventData::Disconnected); } } }, diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index 4f96c4c..f97807d 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -6,9 +6,13 @@ use bytes::Bytes; use futures::{join, pin_mut, select, FutureExt, SinkExt, StreamExt}; use futures_util::stream::{SplitSink, SplitStream}; use mumble_protocol::crypt::ClientCryptState; +use mumble_protocol::ping::{PingPacket, PongPacket}; use mumble_protocol::voice::{VoicePacket, VoicePacketPayload}; use mumble_protocol::Serverbound; +use std::collections::HashMap; +use std::convert::TryFrom; use std::net::{Ipv6Addr, SocketAddr}; +use std::rc::Rc; use std::sync::{Arc, Mutex}; use tokio::net::UdpSocket; use tokio::sync::{mpsc, oneshot, watch}; @@ -225,3 +229,48 @@ async fn send_voice( debug!("UDP sender process killed"); } + +pub async fn handle_pings( + mut ping_request_receiver: mpsc::UnboundedReceiver<( + u64, + SocketAddr, + Box<dyn FnOnce(PongPacket)>, + )>, +) { + let udp_socket = UdpSocket::bind((Ipv6Addr::from(0u128), 0u16)) + .await + .expect("Failed to bind UDP socket"); + + let (mut receiver, mut sender) = udp_socket.split(); + + let pending = Rc::new(Mutex::new(HashMap::new())); + + let sender_handle = async { + while let Some((id, socket_addr, handle)) = ping_request_receiver.recv().await { + let packet = PingPacket { id }; + let packet: [u8; 12] = packet.into(); + sender.send_to(&packet, &socket_addr).await.unwrap(); + pending.lock().unwrap().insert(id, handle); + } + }; + + let receiver_handle = async { + let mut buf = vec![0; 24]; + while let Ok(read) = receiver.recv(&mut buf).await { + assert_eq!(read, 24); + + let packet = match PongPacket::try_from(buf.as_slice()) { + Ok(v) => v, + Err(_) => panic!(), + }; + + if let Some(handler) = pending.lock().unwrap().remove(&packet.id) { + handler(packet); + } + } + }; + + debug!("Waiting for ping requests"); + + join!(sender_handle, receiver_handle); +} diff --git a/mumd/src/state.rs b/mumd/src/state.rs index 4d8c139..56ce030 100644 --- a/mumd/src/state.rs +++ b/mumd/src/state.rs @@ -11,26 +11,40 @@ use crate::network::tcp::{TcpEvent, TcpEventData}; use log::*; use mumble_protocol::control::msgs; use mumble_protocol::control::ControlPacket; +use mumble_protocol::ping::PongPacket; use mumble_protocol::voice::Serverbound; use mumlib::command::{Command, CommandResponse}; use mumlib::config::Config; use mumlib::error::{ChannelIdentifierError, Error}; use mumlib::state::UserDiff; -use std::net::ToSocketAddrs; +use std::net::{SocketAddr, ToSocketAddrs}; use tokio::sync::{mpsc, watch}; macro_rules! at { ($event:expr, $generator:expr) => { - (Some($event), Box::new($generator)) + ExecutionContext::TcpEvent($event, Box::new($generator)) }; } macro_rules! now { ($data:expr) => { - (None, Box::new(move |_| $data)) + ExecutionContext::Now(Box::new(move || $data)) }; } +//TODO give me a better name +pub enum ExecutionContext { + TcpEvent( + TcpEvent, + Box<dyn FnOnce(TcpEventData) -> mumlib::error::Result<Option<CommandResponse>>>, + ), + Now(Box<dyn FnOnce() -> mumlib::error::Result<Option<CommandResponse>>>), + Ping( + Box<dyn FnOnce() -> mumlib::error::Result<SocketAddr>>, + Box<dyn FnOnce(PongPacket) -> mumlib::error::Result<Option<CommandResponse>>>, + ), +} + #[derive(Clone, Debug, Eq, PartialEq)] pub enum StatePhase { Disconnected, @@ -68,13 +82,7 @@ impl State { } //TODO? move bool inside Result - pub fn handle_command( - &mut self, - command: Command, - ) -> ( - Option<TcpEvent>, - Box<dyn FnOnce(Option<&TcpEventData>) -> mumlib::error::Result<Option<CommandResponse>>>, - ) { + pub fn handle_command(&mut self, command: Command) -> ExecutionContext { match command { Command::ChannelJoin { channel_identifier } => { if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) { @@ -128,7 +136,7 @@ impl State { } Command::ChannelList => { if !matches!(*self.phase_receiver().borrow(), StatePhase::Connected) { - return (None, Box::new(|_| Err(Error::DisconnectedError))); + return now!(Err(Error::DisconnectedError)); } let list = channel::into_channel( self.server.as_ref().unwrap().channels(), @@ -173,7 +181,7 @@ impl State { .unwrap(); at!(TcpEvent::Connected, |e| { //runs the closure when the client is connected - if let Some(TcpEventData::Connected(msg)) = e { + if let TcpEventData::Connected(msg) = e { Ok(Some(CommandResponse::ServerConnect { welcome_message: if msg.has_welcome_text() { Some(msg.get_welcome_text().to_string()) @@ -240,6 +248,25 @@ impl State { self.audio.set_user_volume(user_id, volume); now!(Ok(None)) } + Command::ServerStatus { host, port } => ExecutionContext::Ping( + Box::new(move || { + match (host.as_str(), port) + .to_socket_addrs() + .map(|mut e| e.next()) + { + Ok(Some(v)) => Ok(v), + _ => Err(mumlib::error::Error::InvalidServerAddrError(host, port)), + } + }), + Box::new(move |pong| { + Ok(Some(CommandResponse::ServerStatus { + version: pong.version, + users: pong.users, + max_users: pong.max_users, + bandwidth: pong.bandwidth, + })) + }), + ), } } @@ -252,9 +279,9 @@ impl State { // check if this is initial state if !self.server().unwrap().users().contains_key(&session) { self.parse_initial_user_state(session, msg); - return None; + None } else { - return Some(self.parse_updated_user_state(session, msg)); + Some(self.parse_updated_user_state(session, msg)) } } diff --git a/mumlib/src/command.rs b/mumlib/src/command.rs index d2d5d53..2fed4f1 100644 --- a/mumlib/src/command.rs +++ b/mumlib/src/command.rs @@ -20,11 +20,27 @@ pub enum Command { }, ServerDisconnect, Status, + ServerStatus { + host: String, + port: u16, + }, } #[derive(Debug, Deserialize, Serialize)] pub enum CommandResponse { - ChannelList { channels: Channel }, - ServerConnect { welcome_message: Option<String> }, - Status { server_state: Server }, + ChannelList { + channels: Channel, + }, + ServerConnect { + welcome_message: Option<String>, + }, + Status { + server_state: Server, + }, + ServerStatus { + version: u32, + users: u32, + max_users: u32, + bandwidth: u32, + }, } diff --git a/mumlib/src/config.rs b/mumlib/src/config.rs index 2eca104..3a2fa27 100644 --- a/mumlib/src/config.rs +++ b/mumlib/src/config.rs @@ -1,6 +1,8 @@ +use crate::DEFAULT_PORT; use serde::{Deserialize, Serialize}; use std::convert::TryFrom; use std::fs; +use std::net::{SocketAddr, ToSocketAddrs}; use std::path::Path; use toml::value::Array; use toml::Value; @@ -59,6 +61,18 @@ pub struct ServerConfig { pub password: Option<String>, } +impl ServerConfig { + pub fn to_socket_addr(&self) -> Option<SocketAddr> { + match (self.host.as_str(), self.port.unwrap_or(DEFAULT_PORT)) + .to_socket_addrs() + .map(|mut e| e.next()) + { + Ok(Some(addr)) => Some(addr), + _ => None, + } + } +} + pub fn get_cfg_path() -> String { if let Ok(var) = std::env::var("XDG_CONFIG_HOME") { let path = format!("{}/mumdrc", var); diff --git a/mumlib/src/lib.rs b/mumlib/src/lib.rs index a54990e..439efa9 100644 --- a/mumlib/src/lib.rs +++ b/mumlib/src/lib.rs @@ -7,6 +7,7 @@ use colored::*; use log::*; pub const SOCKET_PATH: &str = "/var/tmp/mumd"; +pub const DEFAULT_PORT: u16 = 64738; pub fn setup_logger<T: Into<fern::Output>>(target: T, color: bool) { fern::Dispatch::new() |