/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2023 Dyne.org foundation * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as * published by the Free Software Foundation, either version 3 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ use std::net::SocketAddr; use async_std::sync::{Arc, Mutex}; use futures::{ io::{BufReader, ReadHalf, WriteHalf}, AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, FutureExt, }; use log::{debug, error, info, warn}; use darkfi::{event_graph::model::Event, system::Subscription, Error, Result}; use crate::{ crypto::{decrypt_privmsg, decrypt_target, encrypt_privmsg}, settings, settings::RPL, ChannelInfo, PrivMsgEvent, }; use super::{ClientSubMsg, IrcConfig, NotifierMsg}; pub struct IrcClient { // network stream write_stream: WriteHalf, read_stream: BufReader>, pub address: SocketAddr, // irc config irc_config: IrcConfig, server_notifier: smol::channel::Sender<(NotifierMsg, u64)>, subscription: Subscription, missed_events: Arc>>>, } impl IrcClient { pub fn new( write_stream: WriteHalf, read_stream: BufReader>, address: SocketAddr, irc_config: IrcConfig, server_notifier: smol::channel::Sender<(NotifierMsg, u64)>, subscription: Subscription, missed_events: Arc>>>, ) -> Self { Self { write_stream, read_stream, address, irc_config, subscription, server_notifier, missed_events, } } /// Start listening for messages came from View or irc client pub async fn listen(&mut self) { loop { let mut line = String::new(); futures::select! { // Process msg from View or other client connnected to the same irc server msg = self.subscription.receive().fuse() => { match msg { ClientSubMsg::Privmsg(mut m) => { if let Err(e) = self.process_msg(&mut m).await { error!("[CLIENT {}] Process msg: {}", self.address, e); break } } ClientSubMsg::Config(c) => self.update_config(c).await, } } // Process msg from IRC client err = self.read_stream.read_line(&mut line).fuse() => { if let Err(e) = err { error!("[CLIENT {}] Read line error: {}", self.address, e); break } if let Err(e) = self.process_line(line).await { error!("[CLIENT {}] Process line failed: {}", self.address, e); break } } } } warn!("[CLIENT {}] Close connection", self.address); self.subscription.unsubscribe().await; } pub async fn update_config(&mut self, new_config: IrcConfig) { self.irc_config.channels.extend(new_config.channels); self.irc_config.contacts.extend(new_config.contacts); self.irc_config.nickname = new_config.nickname; self.irc_config.private_key = new_config.private_key; self.irc_config.password = new_config.password; if self.on_receive_join(self.irc_config.channels.keys().cloned().collect()).await.is_err() { warn!("Error to join updated channels"); } } pub async fn process_msg(&mut self, msg: &mut PrivMsgEvent) -> Result<()> { info!("[CLIENT {}] msg from View: {:?}", self.address, msg.to_string()); decrypt_target( msg, &self.irc_config.channels, &self.irc_config.contacts, &self.irc_config.private_key, ); if msg.target.starts_with('#') { // Try to potentially decrypt the incoming message. if !self.irc_config.channels.contains_key(&msg.target) { return Ok(()) } let chan_info = self.irc_config.channels.get_mut(&msg.target).unwrap(); if !chan_info.joined { return Ok(()) } if let Some(salt_box) = &chan_info.salt_box(&msg.target) { decrypt_privmsg(salt_box, msg); info!( "[CLIENT {}] Decrypted received message: {:?}", self.address, msg.to_string() ); } // add the nickname to the channel's names if !chan_info.names.contains(&msg.nick) { chan_info.names.push(msg.nick.clone()); } self.reply(&msg.to_string()).await?; } else if self.irc_config.private_key.is_some() { if let Some(contact_info) = self.irc_config.contacts.get(&msg.target) { let salt_box = &contact_info .salt_box(self.irc_config.private_key.as_ref().unwrap(), &msg.target); if salt_box.is_none() { return Ok(()) } decrypt_privmsg(salt_box.as_ref().unwrap(), msg); info!( "[CLIENT {}] Decrypted received message: {:?}", self.address, msg.to_string() ); self.reply(&msg.to_string()).await?; } } Ok(()) } pub async fn process_line(&mut self, line: String) -> Result<()> { let irc_msg = match clean_input_line(line) { Ok(msg) => msg, Err(e) => { warn!("[CLIENT {}] Connection error: {}", self.address, e); return Err(Error::ChannelStopped) } }; info!("[CLIENT {}] Process msg: {}", self.address, irc_msg); if let Err(e) = self.update(irc_msg).await { warn!("[CLIENT {}] Connection error: {}", self.address, e); return Err(Error::ChannelStopped) } Ok(()) } async fn update(&mut self, line: String) -> Result<()> { if line.len() > settings::MAXIMUM_LENGTH_OF_MESSAGE { return Err(Error::MalformedPacket) } let (command, value) = parse_line(&line)?; let (command, value) = (command.as_str(), value.as_str()); 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!("[CLIENT {}] Unimplemented `{}` command", self.address, command), } self.registre().await?; Ok(()) } async fn registre(&mut self) -> Result<()> { if !self.irc_config.is_pass_init && self.irc_config.password.is_empty() { self.irc_config.is_pass_init = true } if !self.irc_config.is_registered && self.irc_config.is_cap_end && self.irc_config.is_nick_init && self.irc_config.is_user_init { debug!("Initializing peer connection"); let register_reply = format!(":darkfi 001 {} :Let there be dark\r\n", self.irc_config.nickname); self.reply(®ister_reply).await?; self.irc_config.is_registered = true; // join all channels self.on_receive_join(self.irc_config.channels.keys().cloned().collect()).await?; if *self.irc_config.capabilities.get("no-history").unwrap() { return Ok(()) } } Ok(()) } async fn reply(&mut self, message: &str) -> Result<()> { self.write_stream.write_all(message.as_bytes()).await?; debug!("Sent {}", message.trim_end()); Ok(()) } fn on_quit(&self) -> Result<()> { // Close the connection Err(Error::NetworkServiceStopped) } async fn on_receive_user(&mut self) -> Result<()> { // We can stuff any extra things like public keys in here. // Ignore it for now. if self.irc_config.is_pass_init { self.irc_config.is_user_init = true; } else { // Close the connection warn!("[CLIENT {}] Password is required", self.address); return self.on_quit() } Ok(()) } async fn on_receive_pass(&mut self, password: &str) -> Result<()> { if self.irc_config.password == password { self.irc_config.is_pass_init = true } else { // Close the connection warn!("[CLIENT {}] Password is not correct!", self.address); return self.on_quit() } Ok(()) } async fn on_receive_nick(&mut self, nickname: &str) -> Result<()> { if nickname.len() > settings::MAXIMUM_LENGTH_OF_NICKNAME { return Ok(()) } self.irc_config.is_nick_init = true; let old_nick = std::mem::replace(&mut self.irc_config.nickname, nickname.to_string()); let nick_reply = format!(":{}!anon@dark.fi NICK {}\r\n", old_nick, self.irc_config.nickname); self.reply(&nick_reply).await } async fn on_receive_part(&mut self, channels: Vec) -> Result<()> { for chan in channels.iter() { let part_reply = format!(":{}!anon@dark.fi PART {}\r\n", self.irc_config.nickname, chan); self.reply(&part_reply).await?; if self.irc_config.channels.contains_key(chan) { let chan_info = self.irc_config.channels.get_mut(chan).unwrap(); chan_info.joined = false; } } Ok(()) } async fn on_receive_topic(&mut self, line: &str, channel: &str) -> Result<()> { if let Some(substr_idx) = line.find(':') { // Client is setting the topic if substr_idx >= line.len() { return Err(Error::MalformedPacket) } let topic = &line[substr_idx + 1..]; let chan_info = self.irc_config.channels.get_mut(channel).unwrap(); chan_info.topic = Some(topic.to_string()); let topic_reply = format!( ":{}!anon@dark.fi TOPIC {} :{}\r\n", self.irc_config.nickname, channel, topic ); self.reply(&topic_reply).await?; } else { // Client is asking or the topic let chan_info = self.irc_config.channels.get(channel).unwrap(); let topic_reply = if let Some(topic) = &chan_info.topic { format!( "{} {} {} :{}\r\n", RPL::Topic as u32, self.irc_config.nickname, channel, topic ) } else { const TOPIC: &str = "No topic is set"; format!( "{} {} {} :{}\r\n", RPL::NoTopic as u32, self.irc_config.nickname, channel, TOPIC ) }; self.reply(&topic_reply).await?; } Ok(()) } async fn on_ping(&mut self, value: &str) -> Result<()> { let pong = format!("PONG {}\r\n", value); self.reply(&pong).await } async fn on_receive_cap(&mut self, line: &str, subcommand: &str) -> Result<()> { self.irc_config.is_cap_end = false; let capabilities_keys: Vec = self.irc_config.capabilities.keys().cloned().collect(); match subcommand { "LS" => { let cap_ls_reply = format!( ":{}!anon@dark.fi CAP * LS :{}\r\n", self.irc_config.nickname, capabilities_keys.join(" ") ); self.reply(&cap_ls_reply).await?; } "REQ" => { let substr_idx = line.find(':').ok_or(Error::MalformedPacket)?; if substr_idx >= line.len() { return Err(Error::MalformedPacket) } let cap: Vec<&str> = line[substr_idx + 1..].split(' ').collect(); let mut ack_list = vec![]; let mut nak_list = vec![]; for c in cap { if self.irc_config.capabilities.contains_key(c) { self.irc_config.capabilities.insert(c.to_string(), true); ack_list.push(c); } else { nak_list.push(c); } } let cap_ack_reply = format!( ":{}!anon@dark.fi CAP * ACK :{}\r\n", self.irc_config.nickname, ack_list.join(" ") ); let cap_nak_reply = format!( ":{}!anon@dark.fi CAP * NAK :{}\r\n", self.irc_config.nickname, nak_list.join(" ") ); self.reply(&cap_ack_reply).await?; self.reply(&cap_nak_reply).await?; } "LIST" => { let enabled_capabilities: Vec = self .irc_config .capabilities .clone() .into_iter() .filter(|(_, v)| *v) .map(|(k, _)| k) .collect(); let cap_list_reply = format!( ":{}!anon@dark.fi CAP * LIST :{}\r\n", self.irc_config.nickname, enabled_capabilities.join(" ") ); self.reply(&cap_list_reply).await?; } "END" => { self.irc_config.is_cap_end = true; } _ => {} } Ok(()) } async fn on_receive_names(&mut self, channels: Vec) -> Result<()> { for chan in channels.iter() { if !chan.starts_with('#') { continue } if self.irc_config.channels.contains_key(chan) { let chan_info = self.irc_config.channels.get(chan).unwrap(); if chan_info.names.is_empty() { return Ok(()) } let names_reply = format!( ":{}!anon@dark.fi {} = {} : {}\r\n", self.irc_config.nickname, RPL::NameReply as u32, chan, chan_info.names.join(" ") ); self.reply(&names_reply).await?; let end_of_names = format!( ":DarkFi {:03} {} {} :End of NAMES list\r\n", RPL::EndOfNames as u32, self.irc_config.nickname, chan ); self.reply(&end_of_names).await?; } } Ok(()) } async fn on_receive_privmsg(&mut self, line: &str, target: &str) -> Result<()> { let substr_idx = line.find(':').ok_or(Error::MalformedPacket)?; if substr_idx >= line.len() { return Err(Error::MalformedPacket) } let message = line[substr_idx + 1..].to_string(); info!("[CLIENT {}] (Plain) PRIVMSG {} :{}", self.address, target, message,); let mut privmsg = PrivMsgEvent { nick: self.irc_config.nickname.clone(), target: target.to_string(), msg: message, }; if target.starts_with('#') { if !self.irc_config.channels.contains_key(target) { return Ok(()) } let channel_info = self.irc_config.channels.get(target).unwrap(); if !channel_info.joined { return Ok(()) } if let Some(salt_box) = &channel_info.salt_box(target) { encrypt_privmsg(salt_box, &mut privmsg); info!("[CLIENT {}] (Encrypted) PRIVMSG: {:?}", self.address, privmsg.to_string()); } } else if self.irc_config.private_key.is_some() { if let Some(contact_info) = self.irc_config.contacts.get(target) { if let Some(salt_box) = &contact_info.salt_box(self.irc_config.private_key.as_ref().unwrap(), target) { encrypt_privmsg(salt_box, &mut privmsg); info!( "[CLIENT {}] (Encrypted) PRIVMSG: {:?}", self.address, privmsg.to_string() ); } } } self.server_notifier .send((NotifierMsg::Privmsg(privmsg), self.subscription.get_id())) .await?; Ok(()) } async fn on_receive_join(&mut self, channels: Vec) -> Result<()> { for chan in channels.iter() { if !chan.starts_with('#') { continue } if !self.irc_config.channels.contains_key(chan) { let mut chan_info = ChannelInfo::new(); chan_info.topic = Some("n/a".to_string()); self.irc_config.channels.insert(chan.to_string(), chan_info); } let chan_info = self.irc_config.channels.get_mut(chan).unwrap(); if chan_info.joined { return Ok(()) } chan_info.joined = true; let topic = if let Some(topic) = chan_info.topic.clone() { topic } else { "n/a".to_string() }; chan_info.topic = Some(topic.to_string()); { let j = format!(":{}!anon@dark.fi JOIN {}\r\n", self.irc_config.nickname, chan); let t = format!(":DarkFi TOPIC {} :{}\r\n", chan, topic); self.reply(&j).await?; self.reply(&t).await?; } } // Process missed messages if any (sorted by event's timestamp) let mut hash_vec = self.missed_events.lock().await.clone(); hash_vec.sort_by(|a, b| a.timestamp.cmp(&b.timestamp)); for event in hash_vec { let mut action = event.action.clone(); if let Err(e) = self.process_msg(&mut action).await { error!("[CLIENT {}] Process msg: {}", self.address, e); continue } } Ok(()) } } // // Helper functions // fn clean_input_line(mut line: String) -> Result { if line.is_empty() { return Err(Error::ChannelStopped) } if line == "\n" || line == "\r\n" { 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 { 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, value.to_owned())) }