use std::net::SocketAddr; use futures::{io::WriteHalf, AsyncRead, AsyncWrite, AsyncWriteExt}; use fxhash::FxHashMap; use log::{debug, info, warn}; use darkfi::{net::P2pPtr, system::SubscriberPtr, Error, Result}; use crate::{ buffers::{ArcPrivmsgsBuffer, SeenMsgIds}, crypto::{decrypt_privmsg, decrypt_target}, ChannelInfo, Privmsg, MAXIMUM_LENGTH_OF_MESSAGE, }; mod command; pub struct IrcServerConnection { // server stream write_stream: WriteHalf, peer_address: SocketAddr, // msg ids seen_msg_ids: SeenMsgIds, privmsgs_buffer: ArcPrivmsgsBuffer, // user & channels is_nick_init: bool, is_user_init: bool, is_registered: bool, is_cap_end: bool, is_pass_init: bool, nickname: String, auto_channels: Vec, pub configured_chans: FxHashMap, pub configured_contacts: FxHashMap, capabilities: FxHashMap, // p2p p2p: P2pPtr, senders: SubscriberPtr, subscriber_id: u64, password: String, } impl IrcServerConnection { #[allow(clippy::too_many_arguments)] pub fn new( write_stream: WriteHalf, peer_address: SocketAddr, seen_msg_ids: SeenMsgIds, privmsgs_buffer: ArcPrivmsgsBuffer, auto_channels: Vec, password: String, configured_chans: FxHashMap, configured_contacts: FxHashMap, p2p: P2pPtr, senders: SubscriberPtr, subscriber_id: u64, ) -> Self { let mut capabilities = FxHashMap::default(); capabilities.insert("no-history".to_string(), false); Self { write_stream, peer_address, seen_msg_ids, privmsgs_buffer, is_nick_init: false, is_user_init: false, is_registered: false, is_cap_end: true, is_pass_init: false, nickname: "anon".to_string(), auto_channels, password, configured_chans, configured_contacts, capabilities, p2p, senders, subscriber_id, } } pub async fn process_msg_from_p2p(&mut self, msg: &Privmsg) -> Result<()> { info!("Received msg from P2p network: {:?}", msg); let mut msg = msg.clone(); decrypt_target(&mut msg, self.configured_chans.clone(), self.configured_contacts.clone()); if msg.target.starts_with('#') { // Try to potentially decrypt the incoming message. if !self.configured_chans.contains_key(&msg.target) { return Ok(()) } let chan_info = self.configured_chans.get_mut(&msg.target).unwrap(); if !chan_info.joined { return Ok(()) } if let Some(salt_box) = &chan_info.salt_box { decrypt_privmsg(salt_box, &mut msg); info!("Decrypted received message: {:?}", msg); } // add the nickname to the channel's names if !chan_info.names.contains(&msg.nickname) { chan_info.names.push(msg.nickname.clone()); } self.reply(&msg.to_string()).await?; return Ok(()) } else if self.is_cap_end && self.is_nick_init && self.nickname == msg.target { if self.configured_contacts.contains_key(&msg.target) { let salt_box = self.configured_contacts.get(&msg.target).unwrap(); decrypt_privmsg(salt_box, &mut msg); info!("Decrypted received message: {:?}", msg); } self.reply(&msg.to_string()).await?; } Ok(()) } pub async fn process_line_from_client( &mut self, err: std::result::Result, line: String, ) -> Result<()> { if let Err(e) = err { warn!("Read line error {}: {}", self.peer_address, e); return Err(Error::ChannelStopped) } info!("Received msg from IRC client: {:?}", line); let irc_msg = clean_input_line(line, &self.peer_address)?; if let Err(e) = self.update(irc_msg).await { warn!("Connection error: {} for {}", e, self.peer_address); return Err(Error::ChannelStopped) } Ok(()) } async fn update(&mut self, line: String) -> Result<()> { if line.len() > MAXIMUM_LENGTH_OF_MESSAGE { return Err(Error::MalformedPacket) } if self.password.is_empty() { self.is_pass_init = true } let (command, value) = parse_line(&line)?; let (command, value) = (command.as_str(), value.as_str()); info!("IRC server received command: {}", command); match command { "PASS" => self.on_receive_pass(value).await?, "USER" => self.on_receive_user().await?, "NAMES" => self.on_receive_names(value.split(',').map(String::from).collect()).await?, "NICK" => self.on_receive_nick(value).await?, "JOIN" => self.on_receive_join(value.split(',').map(String::from).collect()).await?, "PART" => self.on_receive_part(value.split(',').map(String::from).collect()).await?, "TOPIC" => self.on_receive_topic(&line, value).await?, "PING" => self.on_ping(value).await?, "PRIVMSG" => self.on_receive_privmsg(&line, value).await?, "CAP" => self.on_receive_cap(&line, &value.to_uppercase()).await?, "QUIT" => self.on_quit()?, _ => warn!("Unimplemented `{}` command", command), } self.registre().await?; Ok(()) } async fn registre(&mut self) -> Result<()> { if !self.is_registered && self.is_cap_end && self.is_nick_init && self.is_user_init { debug!("Initializing peer connection"); let register_reply = format!(":darkfi 001 {} :Let there be dark\r\n", self.nickname); self.reply(®ister_reply).await?; self.is_registered = true; self.on_receive_join(self.auto_channels.clone()).await?; if *self.capabilities.get("no-history").unwrap() { return Ok(()) } // Send dm messages in buffer let mut privmsgs_buffer = self.privmsgs_buffer.lock().await; privmsgs_buffer.update(); for msg in privmsgs_buffer.iter() { let is_dm = msg.target == self.nickname || (msg.nickname == self.nickname && !msg.target.starts_with('#')); if is_dm { self.senders.notify_by_id(msg.clone(), self.subscriber_id).await; } } drop(privmsgs_buffer); } Ok(()) } async fn reply(&mut self, message: &str) -> Result<()> { self.write_stream.write_all(message.as_bytes()).await?; debug!("Sent {}", message); Ok(()) } } // // Helper functions // fn clean_input_line(mut line: String, peer_address: &SocketAddr) -> Result { if line.is_empty() { warn!("Received empty line from {}. ", peer_address); warn!("Closing connection."); return Err(Error::ChannelStopped) } if line == "\n" || line == "\r\n" { warn!("Closing connection."); return Err(Error::ChannelStopped) } if &line[(line.len() - 2)..] == "\r\n" { // Remove CRLF line.pop(); line.pop(); } else if &line[(line.len() - 1)..] == "\n" { line.pop(); } else { warn!("Closing connection."); return Err(Error::ChannelStopped) } Ok(line.clone()) } fn parse_line(line: &str) -> Result<(String, String)> { let mut tokens = line.split_ascii_whitespace(); // Commands can begin with :garbage but we will reject clients doing // that for now to keep the protocol simple and focused. let command = tokens.next().ok_or(Error::MalformedPacket)?.to_uppercase(); let value = tokens.next().ok_or(Error::MalformedPacket)?; Ok((command.to_owned(), value.to_owned())) }