bin/ircd: move the irc server functionalities from main file

bin/ircd/src/irc/client.rs

@@ -0,0 +1,600 @@
+use std::net::SocketAddr;
+
+use futures::{
+    io::{BufReader, ReadHalf, WriteHalf},
+    AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, FutureExt,
+};
+use fxhash::FxHashMap;
+use log::{debug, error, info, warn};
+
+use darkfi::{
+    net::P2pPtr,
+    system::{SubscriberPtr, Subscription},
+    Error, Result,
+};
+
+use crate::{
+    buffers::{ArcPrivmsgsBuffer, SeenIds},
+    crypto::{decrypt_privmsg, decrypt_target, encrypt_privmsg},
+    privmsg::{MAXIMUM_LENGTH_OF_MESSAGE, MAXIMUM_LENGTH_OF_NICKNAME},
+    ChannelInfo, ContactInfo, Privmsg,
+};
+
+const RPL_NOTOPIC: u32 = 331;
+const RPL_TOPIC: u32 = 332;
+const RPL_NAMEREPLY: u32 = 353;
+const RPL_ENDOFNAMES: u32 = 366;
+
+pub struct IrcClient<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> {
+    // network stream
+    write_stream: WriteHalf<C>,
+    pub address: SocketAddr,
+
+    // msgs buffer
+    privmsgs_buffer: ArcPrivmsgsBuffer,
+    seen_msg_ids: SeenIds,
+
+    // init bool
+    is_nick_init: bool,
+    is_user_init: bool,
+    is_registered: bool,
+    is_cap_end: bool,
+    is_pass_init: bool,
+
+    // user config
+    nickname: String,
+    password: String,
+    capabilities: FxHashMap<String, bool>,
+
+    // channels and contacts
+    auto_channels: Vec<String>,
+    pub configured_chans: FxHashMap<String, ChannelInfo>,
+    pub configured_contacts: FxHashMap<String, ContactInfo>,
+
+    // p2p
+    p2p: P2pPtr,
+    p2p_notifiers: SubscriberPtr<Privmsg>,
+    subscription: Subscription<Privmsg>,
+}
+
+impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        write_stream: WriteHalf<C>,
+        address: SocketAddr,
+        privmsgs_buffer: ArcPrivmsgsBuffer,
+        seen_msg_ids: SeenIds,
+        password: String,
+        auto_channels: Vec<String>,
+        configured_chans: FxHashMap<String, ChannelInfo>,
+        configured_contacts: FxHashMap<String, ContactInfo>,
+        p2p: P2pPtr,
+        p2p_notifiers: SubscriberPtr<Privmsg>,
+        subscription: Subscription<Privmsg>,
+    ) -> Self {
+        let mut capabilities = FxHashMap::default();
+        capabilities.insert("no-history".to_string(), false);
+        Self {
+            write_stream,
+            address,
+            privmsgs_buffer,
+            seen_msg_ids,
+            is_nick_init: false,
+            is_user_init: false,
+            is_registered: false,
+            is_cap_end: true,
+            is_pass_init: false,
+            nickname: "anon".to_string(),
+            password,
+            auto_channels,
+            configured_chans,
+            configured_contacts,
+            capabilities,
+            p2p,
+            p2p_notifiers,
+            subscription,
+        }
+    }
+
+    /// Start listening for messages came from p2p network or irc client
+    pub async fn listen(&mut self, mut reader: BufReader<ReadHalf<C>>) {
+        loop {
+            let mut line = String::new();
+
+            futures::select! {
+                msg = self.subscription.receive().fuse() => {
+                    if let Err(e) = self.process_msg_from_p2p(&msg).await {
+                        error!("[CLIENT {}] Process msg from p2p: {}",  self.address, e);
+                        break
+                    }
+                }
+                err = reader.read_line(&mut line).fuse() => {
+                    if let Err(e) = self.process_line(err, line).await {
+                        error!("[CLIENT {}] Process line failed: {}",  self.address, e);
+                        break
+                    }
+                }
+            }
+        }
+
+        warn!("[CLIENT {}] Close connection", self.address);
+        self.subscription.unsubscribe().await;
+    }
+
+    pub async fn process_msg_from_p2p(&mut self, msg: &Privmsg) -> Result<()> {
+        info!("[P2P] Received: {}", msg.to_string().trim());
+
+        let mut msg = msg.clone();
+        let mut contact = String::new();
+        decrypt_target(
+            &mut contact,
+            &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 {
+            if !self.configured_contacts.contains_key(&contact) {
+                return Ok(())
+            }
+
+            let contact_info = self.configured_contacts.get(&contact).unwrap();
+            if let Some(salt_box) = &contact_info.salt_box {
+                decrypt_privmsg(salt_box, &mut msg);
+                // This is for /query
+                msg.nickname = contact;
+                info!("[P2P] Decrypted received message: {:?}", msg);
+            }
+
+            self.reply(&msg.to_string()).await?;
+        }
+
+        Ok(())
+    }
+
+    pub async fn process_line(
+        &mut self,
+        err: std::result::Result<usize, std::io::Error>,
+        line: String,
+    ) -> Result<()> {
+        if let Err(e) = err {
+            warn!("[CLIENT {}] Read line error: {}", self.address, e);
+            return Err(Error::ChannelStopped)
+        }
+
+        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 {}] 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() > 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());
+
+        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.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(&register_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 privmsgs_buffer = self.privmsgs_buffer.lock().await;
+            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.p2p_notifiers.notify_by_id(msg.clone(), self.subscription.get_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(())
+    }
+
+    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.is_pass_init {
+            self.is_user_init = true;
+        } else {
+            // Close the connection
+            warn!("[IRC SERVER] Password is required");
+            return self.on_quit()
+        }
+        Ok(())
+    }
+
+    async fn on_receive_pass(&mut self, password: &str) -> Result<()> {
+        if self.password == password {
+            self.is_pass_init = true
+        } else {
+            // Close the connection
+            warn!("[IRC SERVER] Password is not correct!");
+            return self.on_quit()
+        }
+        Ok(())
+    }
+
+    async fn on_receive_nick(&mut self, nickname: &str) -> Result<()> {
+        if nickname.len() > MAXIMUM_LENGTH_OF_NICKNAME {
+            return Ok(())
+        }
+
+        self.is_nick_init = true;
+        let old_nick = std::mem::replace(&mut self.nickname, nickname.to_string());
+
+        let nick_reply = format!(":{}!anon@dark.fi NICK {}\r\n", old_nick, self.nickname);
+        self.reply(&nick_reply).await
+    }
+
+    async fn on_receive_part(&mut self, channels: Vec<String>) -> Result<()> {
+        for chan in channels.iter() {
+            let part_reply = format!(":{}!anon@dark.fi PART {}\r\n", self.nickname, chan);
+            self.reply(&part_reply).await?;
+            if self.configured_chans.contains_key(chan) {
+                let chan_info = self.configured_chans.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.configured_chans.get_mut(channel).unwrap();
+            chan_info.topic = Some(topic.to_string());
+
+            let topic_reply =
+                format!(":{}!anon@dark.fi TOPIC {} :{}\r\n", self.nickname, channel, topic);
+            self.reply(&topic_reply).await?;
+        } else {
+            // Client is asking or the topic
+            let chan_info = self.configured_chans.get(channel).unwrap();
+            let topic_reply = if let Some(topic) = &chan_info.topic {
+                format!("{} {} {} :{}\r\n", RPL_TOPIC, self.nickname, channel, topic)
+            } else {
+                const TOPIC: &str = "No topic is set";
+                format!("{} {} {} :{}\r\n", RPL_NOTOPIC, self.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.is_cap_end = false;
+
+        let capabilities_keys: Vec<String> = self.capabilities.keys().cloned().collect();
+
+        match subcommand {
+            "LS" => {
+                let cap_ls_reply = format!(
+                    ":{}!anon@dark.fi CAP * LS :{}\r\n",
+                    self.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.capabilities.contains_key(c) {
+                        self.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.nickname,
+                    ack_list.join(" ")
+                );
+
+                let cap_nak_reply = format!(
+                    ":{}!anon@dark.fi CAP * NAK :{}\r\n",
+                    self.nickname,
+                    nak_list.join(" ")
+                );
+
+                self.reply(&cap_ack_reply).await?;
+                self.reply(&cap_nak_reply).await?;
+            }
+
+            "LIST" => {
+                let enabled_capabilities: Vec<String> = self
+                    .capabilities
+                    .clone()
+                    .into_iter()
+                    .filter(|(_, v)| *v)
+                    .map(|(k, _)| k)
+                    .collect();
+
+                let cap_list_reply = format!(
+                    ":{}!anon@dark.fi CAP * LIST :{}\r\n",
+                    self.nickname,
+                    enabled_capabilities.join(" ")
+                );
+                self.reply(&cap_list_reply).await?;
+            }
+
+            "END" => {
+                self.is_cap_end = true;
+            }
+            _ => {}
+        }
+        Ok(())
+    }
+
+    async fn on_receive_names(&mut self, channels: Vec<String>) -> Result<()> {
+        for chan in channels.iter() {
+            if !chan.starts_with('#') {
+                continue
+            }
+            if self.configured_chans.contains_key(chan) {
+                let chan_info = self.configured_chans.get(chan).unwrap();
+
+                if chan_info.names.is_empty() {
+                    return Ok(())
+                }
+
+                let names_reply = format!(
+                    ":{}!anon@dark.fi {} = {} : {}\r\n",
+                    self.nickname,
+                    RPL_NAMEREPLY,
+                    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, self.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 privmsgs_buffer = self.privmsgs_buffer.lock().await;
+        let last_term = privmsgs_buffer.last_term() + 1;
+        drop(privmsgs_buffer);
+
+        let mut privmsg = Privmsg::new(&self.nickname, target, &message, last_term);
+
+        if target.starts_with('#') {
+            if !self.configured_chans.contains_key(target) {
+                return Ok(())
+            }
+
+            let channel_info = self.configured_chans.get(target).unwrap();
+
+            if !channel_info.joined {
+                return Ok(())
+            }
+
+            if let Some(salt_box) = &channel_info.salt_box {
+                encrypt_privmsg(salt_box, &mut privmsg);
+                info!("[CLIENT {}] (Encrypted) PRIVMSG: {:?}", self.address, privmsg);
+            }
+        } else {
+            if !self.configured_contacts.contains_key(target) {
+                return Ok(())
+            }
+
+            let contact_info = self.configured_contacts.get(target).unwrap();
+            if let Some(salt_box) = &contact_info.salt_box {
+                encrypt_privmsg(salt_box, &mut privmsg);
+                info!("[CLIENT {}] (Encrypted) PRIVMSG: {:?}", self.address, privmsg);
+            }
+        }
+
+        {
+            (*self.seen_msg_ids.lock().await).push(privmsg.id);
+            (*self.privmsgs_buffer.lock().await).push(&privmsg)
+        }
+
+        self.p2p_notifiers
+            .notify_with_exclude(privmsg.clone(), &[self.subscription.get_id()])
+            .await;
+
+        debug!(target: "ircd", "PRIVMSG to be sent: {:?}", privmsg);
+        self.p2p.broadcast(privmsg).await?;
+
+        Ok(())
+    }
+
+    async fn on_receive_join(&mut self, channels: Vec<String>) -> Result<()> {
+        for chan in channels.iter() {
+            if !chan.starts_with('#') {
+                continue
+            }
+            if !self.configured_chans.contains_key(chan) {
+                let mut chan_info = ChannelInfo::new()?;
+                chan_info.topic = Some("n/a".to_string());
+                self.configured_chans.insert(chan.to_string(), chan_info);
+            }
+
+            let chan_info = self.configured_chans.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.nickname, chan);
+                let t = format!(":DarkFi TOPIC {} :{}\r\n", chan, topic);
+                self.reply(&j).await?;
+                self.reply(&t).await?;
+            }
+
+            // Send messages in buffer
+            if !self.capabilities.get("no-history").unwrap() {
+                for msg in self.privmsgs_buffer.lock().await.iter() {
+                    if msg.target == *chan {
+                        self.p2p_notifiers
+                            .notify_by_id(msg.clone(), self.subscription.get_id())
+                            .await;
+                    }
+                }
+            }
+        }
+        self.on_receive_names(channels).await?;
+        Ok(())
+    }
+}
+
+//
+// Helper functions
+//
+fn clean_input_line(mut line: String) -> Result<String> {
+    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()))
+}

bin/ircd/src/irc/mod.rs

@@ -0,0 +1,181 @@
+use async_std::{net::TcpListener, sync::Arc};
+use std::{fs::File, net::SocketAddr};
+
+use async_executor::Executor;
+use futures::{io::BufReader, AsyncRead, AsyncReadExt, AsyncWrite};
+use futures_rustls::{rustls, TlsAcceptor};
+use fxhash::FxHashMap;
+use log::{error, info};
+
+use darkfi::{net::P2pPtr, system::SubscriberPtr, util::expand_path, Error, Result};
+
+use crate::{
+    buffers::{ArcPrivmsgsBuffer, SeenIds},
+    settings::Args,
+    ChannelInfo, ContactInfo, Privmsg,
+};
+
+mod client;
+
+pub use client::IrcClient;
+
+pub struct IrcServer {
+    settings: Args,
+    privmsgs_buffer: ArcPrivmsgsBuffer,
+    seen_msg_ids: SeenIds,
+    auto_channels: Vec<String>,
+    password: String,
+    configured_chans: FxHashMap<String, ChannelInfo>,
+    configured_contacts: FxHashMap<String, ContactInfo>,
+    p2p: P2pPtr,
+    p2p_notifiers: SubscriberPtr<Privmsg>,
+}
+
+impl IrcServer {
+    pub async fn new(
+        settings: Args,
+        privmsgs_buffer: ArcPrivmsgsBuffer,
+        seen_msg_ids: SeenIds,
+        auto_channels: Vec<String>,
+        password: String,
+        configured_chans: FxHashMap<String, ChannelInfo>,
+        configured_contacts: FxHashMap<String, ContactInfo>,
+        p2p: P2pPtr,
+        p2p_notifiers: SubscriberPtr<Privmsg>,
+    ) -> Result<Self> {
+        Ok(Self {
+            settings,
+            privmsgs_buffer,
+            seen_msg_ids,
+            auto_channels,
+            password,
+            configured_chans,
+            configured_contacts,
+            p2p,
+            p2p_notifiers,
+        })
+    }
+
+    /// Start listening to new irc clients connecting to the irc server address
+    /// then spawn new connections
+    pub async fn start(&self, executor: Arc<Executor<'_>>) -> Result<()> {
+        let (listener, acceptor) = self.setup_listener().await?;
+        info!("[IRC SERVER] listening on {}", self.settings.irc_listen);
+        loop {
+            let (stream, peer_addr) = match listener.accept().await {
+                Ok((s, a)) => (s, a),
+                Err(e) => {
+                    error!("[IRC SERVER] Failed accepting new connections: {}", e);
+                    continue
+                }
+            };
+
+            let result = if let Some(acceptor) = acceptor.clone() {
+                let stream = match acceptor.accept(stream).await {
+                    Ok(s) => s,
+                    Err(e) => {
+                        error!("[IRC SERVER] Failed accepting TLS connection: {}", e);
+                        continue
+                    }
+                };
+                self.process_connection(executor.clone(), stream, peer_addr).await
+            } else {
+                self.process_connection(executor.clone(), stream, peer_addr).await
+            };
+
+            if let Err(e) = result {
+                error!("[IRC SERVER] Failed processing connection {}: {}", peer_addr, e);
+                continue
+            };
+
+            info!("[IRC SERVER] Accept new connection: {}", peer_addr);
+        }
+    }
+
+    /// On every new connection create new IrcClient which will process the messages
+    async fn process_connection<C: AsyncRead + AsyncWrite + Send + Unpin + 'static>(
+        &self,
+        executor: Arc<Executor<'_>>,
+        stream: C,
+        peer_addr: SocketAddr,
+    ) -> Result<()> {
+        let (reader, writer) = stream.split();
+
+        let reader = BufReader::new(reader);
+
+        // New subscription
+        let p2p_subscription = self.p2p_notifiers.clone().subscribe().await;
+
+        let privmsgs_buffer = self.privmsgs_buffer.clone();
+        let seen_msg_ids = self.seen_msg_ids.clone();
+        let auto_channels = self.auto_channels.clone();
+        let password = self.password.clone();
+        let configured_chans = self.configured_chans.clone();
+        let configured_contacts = self.configured_contacts.clone();
+        let p2p = self.p2p.clone();
+        let p2p_notifiers = self.p2p_notifiers.clone();
+
+        executor
+            .spawn(async move {
+                // New irc connection
+                let mut client = IrcClient::new(
+                    writer,
+                    peer_addr,
+                    privmsgs_buffer,
+                    seen_msg_ids,
+                    password,
+                    auto_channels,
+                    configured_chans,
+                    configured_contacts,
+                    p2p,
+                    p2p_notifiers,
+                    p2p_subscription,
+                );
+
+                client.listen(reader).await;
+            })
+            .detach();
+
+        Ok(())
+    }
+
+    /// Setup a listener for irc server
+    async fn setup_listener(&self) -> Result<(TcpListener, Option<TlsAcceptor>)> {
+        let listenaddr = self.settings.irc_listen.socket_addrs(|| None)?[0];
+        let listener = TcpListener::bind(listenaddr).await?;
+
+        let acceptor = match self.settings.irc_listen.scheme() {
+            "tls" => {
+                // openssl genpkey -algorithm ED25519 > example.com.key
+                // openssl req -new -out example.com.csr -key example.com.key
+                // openssl x509 -req -days 700 -in example.com.csr -signkey example.com.key -out example.com.crt
+
+                if self.settings.irc_tls_secret.is_none() || self.settings.irc_tls_cert.is_none() {
+                    error!("[IRC SERVER] To listen using TLS, please set irc_tls_secret and irc_tls_cert in your config file.");
+                    return Err(Error::KeypairPathNotFound)
+                }
+
+                let file =
+                    File::open(expand_path(self.settings.irc_tls_secret.as_ref().unwrap())?)?;
+                let mut reader = std::io::BufReader::new(file);
+                let secret = &rustls_pemfile::pkcs8_private_keys(&mut reader)?[0];
+                let secret = rustls::PrivateKey(secret.clone());
+
+                let file = File::open(expand_path(self.settings.irc_tls_cert.as_ref().unwrap())?)?;
+                let mut reader = std::io::BufReader::new(file);
+                let certificate = &rustls_pemfile::certs(&mut reader)?[0];
+                let certificate = rustls::Certificate(certificate.clone());
+
+                let config = rustls::ServerConfig::builder()
+                    .with_safe_defaults()
+                    .with_no_client_auth()
+                    .with_single_cert(vec![certificate], secret)?;
+
+                let acceptor = TlsAcceptor::from(Arc::new(config));
+                Some(acceptor)
+            }
+            _ => None,
+        };
+        Ok((listener, acceptor))
+    }
+}

bin/ircd/src/irc_server/command.rs

@@ -1,313 +0,0 @@
-use futures::{AsyncRead, AsyncWrite};
-use log::{debug, info, warn};
-
-use darkfi::{Error, Result};
-
-use crate::{
-    crypto::encrypt_privmsg,
-    privmsg::{Privmsg, MAXIMUM_LENGTH_OF_NICKNAME},
-    ChannelInfo,
-};
-
-use super::IrcServerConnection;
-
-const RPL_NOTOPIC: u32 = 331;
-const RPL_TOPIC: u32 = 332;
-const RPL_NAMEREPLY: u32 = 353;
-const RPL_ENDOFNAMES: u32 = 366;
-
-impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C> {
-    pub(super) fn on_quit(&self) -> Result<()> {
-        // Close the connection
-        Err(Error::NetworkServiceStopped)
-    }
-
-    pub(super) 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.is_pass_init {
-            self.is_user_init = true;
-        } else {
-            // Close the connection
-            warn!("[IRC SERVER] Password is required");
-            return self.on_quit()
-        }
-        Ok(())
-    }
-
-    pub(super) async fn on_receive_pass(&mut self, password: &str) -> Result<()> {
-        if self.password == password {
-            self.is_pass_init = true
-        } else {
-            // Close the connection
-            warn!("[IRC SERVER] Password is not correct!");
-            return self.on_quit()
-        }
-        Ok(())
-    }
-
-    pub(super) async fn on_receive_nick(&mut self, nickname: &str) -> Result<()> {
-        if nickname.len() > MAXIMUM_LENGTH_OF_NICKNAME {
-            return Ok(())
-        }
-
-        self.is_nick_init = true;
-        let old_nick = std::mem::replace(&mut self.nickname, nickname.to_string());
-
-        let nick_reply = format!(":{}!anon@dark.fi NICK {}\r\n", old_nick, self.nickname);
-        self.reply(&nick_reply).await
-    }
-
-    pub(super) async fn on_receive_part(&mut self, channels: Vec<String>) -> Result<()> {
-        for chan in channels.iter() {
-            let part_reply = format!(":{}!anon@dark.fi PART {}\r\n", self.nickname, chan);
-            self.reply(&part_reply).await?;
-            if self.configured_chans.contains_key(chan) {
-                let chan_info = self.configured_chans.get_mut(chan).unwrap();
-                chan_info.joined = false;
-            }
-        }
-        Ok(())
-    }
-
-    pub(super) 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.configured_chans.get_mut(channel).unwrap();
-            chan_info.topic = Some(topic.to_string());
-
-            let topic_reply =
-                format!(":{}!anon@dark.fi TOPIC {} :{}\r\n", self.nickname, channel, topic);
-            self.reply(&topic_reply).await?;
-        } else {
-            // Client is asking or the topic
-            let chan_info = self.configured_chans.get(channel).unwrap();
-            let topic_reply = if let Some(topic) = &chan_info.topic {
-                format!("{} {} {} :{}\r\n", RPL_TOPIC, self.nickname, channel, topic)
-            } else {
-                const TOPIC: &str = "No topic is set";
-                format!("{} {} {} :{}\r\n", RPL_NOTOPIC, self.nickname, channel, TOPIC)
-            };
-            self.reply(&topic_reply).await?;
-        }
-        Ok(())
-    }
-
-    pub(super) async fn on_ping(&mut self, value: &str) -> Result<()> {
-        let pong = format!("PONG {}\r\n", value);
-        self.reply(&pong).await
-    }
-
-    pub(super) async fn on_receive_cap(&mut self, line: &str, subcommand: &str) -> Result<()> {
-        self.is_cap_end = false;
-
-        let capabilities_keys: Vec<String> = self.capabilities.keys().cloned().collect();
-
-        match subcommand {
-            "LS" => {
-                let cap_ls_reply = format!(
-                    ":{}!anon@dark.fi CAP * LS :{}\r\n",
-                    self.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.capabilities.contains_key(c) {
-                        self.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.nickname,
-                    ack_list.join(" ")
-                );
-
-                let cap_nak_reply = format!(
-                    ":{}!anon@dark.fi CAP * NAK :{}\r\n",
-                    self.nickname,
-                    nak_list.join(" ")
-                );
-
-                self.reply(&cap_ack_reply).await?;
-                self.reply(&cap_nak_reply).await?;
-            }
-
-            "LIST" => {
-                let enabled_capabilities: Vec<String> = self
-                    .capabilities
-                    .clone()
-                    .into_iter()
-                    .filter(|(_, v)| *v)
-                    .map(|(k, _)| k)
-                    .collect();
-
-                let cap_list_reply = format!(
-                    ":{}!anon@dark.fi CAP * LIST :{}\r\n",
-                    self.nickname,
-                    enabled_capabilities.join(" ")
-                );
-                self.reply(&cap_list_reply).await?;
-            }
-
-            "END" => {
-                self.is_cap_end = true;
-            }
-            _ => {}
-        }
-        Ok(())
-    }
-
-    pub(super) async fn on_receive_names(&mut self, channels: Vec<String>) -> Result<()> {
-        for chan in channels.iter() {
-            if !chan.starts_with('#') {
-                continue
-            }
-            if self.configured_chans.contains_key(chan) {
-                let chan_info = self.configured_chans.get(chan).unwrap();
-
-                if chan_info.names.is_empty() {
-                    return Ok(())
-                }
-
-                let names_reply = format!(
-                    ":{}!anon@dark.fi {} = {} : {}\r\n",
-                    self.nickname,
-                    RPL_NAMEREPLY,
-                    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, self.nickname, chan
-                );
-
-                self.reply(&end_of_names).await?;
-            }
-        }
-        Ok(())
-    }
-
-    pub(super) 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.peer_address, target, message,);
-
-        let privmsgs_buffer = self.privmsgs_buffer.lock().await;
-        let last_term = privmsgs_buffer.last_term() + 1;
-        drop(privmsgs_buffer);
-
-        let mut privmsg = Privmsg::new(&self.nickname, target, &message, last_term);
-
-        if target.starts_with('#') {
-            if !self.configured_chans.contains_key(target) {
-                return Ok(())
-            }
-
-            let channel_info = self.configured_chans.get(target).unwrap();
-
-            if !channel_info.joined {
-                return Ok(())
-            }
-
-            if let Some(salt_box) = &channel_info.salt_box {
-                encrypt_privmsg(salt_box, &mut privmsg);
-                info!("[CLIENT {}] (Encrypted) PRIVMSG: {:?}", self.peer_address, privmsg);
-            }
-        } else {
-            if !self.configured_contacts.contains_key(target) {
-                return Ok(())
-            }
-
-            let contact_info = self.configured_contacts.get(target).unwrap();
-            if let Some(salt_box) = &contact_info.salt_box {
-                encrypt_privmsg(salt_box, &mut privmsg);
-                info!("[CLIENT {}] (Encrypted) PRIVMSG: {:?}", self.peer_address, privmsg);
-            }
-        }
-
-        {
-            (*self.seen_msg_ids.lock().await).push(privmsg.id);
-            (*self.privmsgs_buffer.lock().await).push(&privmsg)
-        }
-
-        self.senders.notify_with_exclude(privmsg.clone(), &[self.subscriber_id]).await;
-
-        debug!(target: "ircd", "PRIVMSG to be sent: {:?}", privmsg);
-        self.p2p.broadcast(privmsg).await?;
-
-        Ok(())
-    }
-
-    pub(super) async fn on_receive_join(&mut self, channels: Vec<String>) -> Result<()> {
-        for chan in channels.iter() {
-            if !chan.starts_with('#') {
-                continue
-            }
-            if !self.configured_chans.contains_key(chan) {
-                let mut chan_info = ChannelInfo::new()?;
-                chan_info.topic = Some("n/a".to_string());
-                self.configured_chans.insert(chan.to_string(), chan_info);
-            }
-
-            let chan_info = self.configured_chans.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.nickname, chan);
-                let t = format!(":DarkFi TOPIC {} :{}\r\n", chan, topic);
-                self.reply(&j).await?;
-                self.reply(&t).await?;
-            }
-
-            // Send messages in buffer
-            if !self.capabilities.get("no-history").unwrap() {
-                for msg in self.privmsgs_buffer.lock().await.iter() {
-                    if msg.target == *chan {
-                        self.senders.notify_by_id(msg.clone(), self.subscriber_id).await;
-                    }
-                }
-            }
-        }
-        self.on_receive_names(channels).await?;
-        Ok(())
-    }
-}

bin/ircd/src/irc_server/mod.rs

@@ -1,261 +0,0 @@
-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, SeenIds},
-    crypto::{decrypt_privmsg, decrypt_target},
-    privmsg::MAXIMUM_LENGTH_OF_MESSAGE,
-    ChannelInfo, ContactInfo, Privmsg,
-};
-
-mod command;
-
-pub struct IrcServerConnection<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> {
-    // server stream
-    write_stream: WriteHalf<C>,
-    pub peer_address: SocketAddr,
-    // msg ids
-    seen_msg_ids: SeenIds,
-    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<String>,
-    pub configured_chans: FxHashMap<String, ChannelInfo>,
-    pub configured_contacts: FxHashMap<String, ContactInfo>,
-    capabilities: FxHashMap<String, bool>,
-    // p2p
-    p2p: P2pPtr,
-    senders: SubscriberPtr<Privmsg>,
-    subscriber_id: u64,
-    password: String,
-}
-
-impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C> {
-    #[allow(clippy::too_many_arguments)]
-    pub fn new(
-        write_stream: WriteHalf<C>,
-        peer_address: SocketAddr,
-        seen_msg_ids: SeenIds,
-        privmsgs_buffer: ArcPrivmsgsBuffer,
-        auto_channels: Vec<String>,
-        password: String,
-        configured_chans: FxHashMap<String, ChannelInfo>,
-        configured_contacts: FxHashMap<String, ContactInfo>,
-        p2p: P2pPtr,
-        senders: SubscriberPtr<Privmsg>,
-        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!("[P2P] Received: {}", msg.to_string().trim());
-
-        let mut msg = msg.clone();
-        let mut contact = String::new();
-        decrypt_target(
-            &mut contact,
-            &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 {
-            if !self.configured_contacts.contains_key(&contact) {
-                return Ok(())
-            }
-
-            let contact_info = self.configured_contacts.get(&contact).unwrap();
-            if let Some(salt_box) = &contact_info.salt_box {
-                decrypt_privmsg(salt_box, &mut msg);
-                // This is for /query
-                msg.nickname = contact;
-                info!("[P2P] Decrypted received message: {:?}", msg);
-            }
-
-            self.reply(&msg.to_string()).await?;
-        }
-
-        Ok(())
-    }
-
-    pub async fn process_line_from_client(
-        &mut self,
-        err: std::result::Result<usize, std::io::Error>,
-        line: String,
-    ) -> Result<()> {
-        if let Err(e) = err {
-            warn!("[CLIENT {}] Read line error: {}", self.peer_address, e);
-            return Err(Error::ChannelStopped)
-        }
-
-        let irc_msg = match clean_input_line(line) {
-            Ok(msg) => msg,
-            Err(e) => {
-                warn!("[CLIENT {}] Connection error: {}", self.peer_address, e);
-                return Err(Error::ChannelStopped)
-            }
-        };
-
-        info!("[CLIENT {}] Msg: {}", self.peer_address, irc_msg);
-
-        if let Err(e) = self.update(irc_msg).await {
-            warn!("[CLIENT {}] Connection error: {}", self.peer_address, e);
-            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());
-
-        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.peer_address, 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(&register_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 privmsgs_buffer = self.privmsgs_buffer.lock().await;
-            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) -> Result<String> {
-    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()))
-}

bin/ircd/src/main.rs

@@ -1,18 +1,11 @@
-use async_std::{
-    net::TcpListener,
-    sync::{Arc, Mutex},
-};
-use std::{fmt, fs::File, net::SocketAddr};
+use async_std::sync::{Arc, Mutex};
+use std::fmt;
 
 use async_channel::Receiver;
 use async_executor::Executor;
-use futures::{
-    io::{BufReader, ReadHalf},
-    AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, FutureExt,
-};
-use futures_rustls::{rustls, TlsAcceptor};
+
 use fxhash::FxHashMap;
-use log::{error, info, warn};
+use log::{info, warn};
 use rand::rngs::OsRng;
 use smol::future;
 use structopt_toml::StructOptToml;
@@ -20,19 +13,19 @@ use structopt_toml::StructOptToml;
 use darkfi::{
     async_daemonize, net,
     rpc::server::listen_and_serve,
-    system::{Subscriber, SubscriberPtr, Subscription},
+    system::{Subscriber, SubscriberPtr},
     util::{
         cli::{get_log_config, get_log_level, spawn_config},
         expand_path,
         file::save_json_file,
         path::get_config_path,
     },
-    Error, Result,
+    Result,
 };
 
 pub mod buffers;
 pub mod crypto;
-pub mod irc_server;
+pub mod irc;
 pub mod privmsg;
 pub mod protocol_privmsg;
 pub mod rpc;
@@ -40,7 +33,7 @@ pub mod settings;
 
 use crate::{
     buffers::{ArcPrivmsgsBuffer, PrivmsgsBuffer, RingBuffer, SeenIds, SIZE_OF_MSG_IDSS_BUFFER},
-    irc_server::IrcServerConnection,
+    irc::IrcServer,
     privmsg::Privmsg,
     protocol_privmsg::ProtocolPrivmsg,
     rpc::JsonRpcInterface,
@@ -64,186 +57,72 @@ impl fmt::Display for KeyPair {
 
 pub type UnreadMsgs = Arc<Mutex<FxHashMap<String, Privmsg>>>;
 
-async fn setup_listener(settings: Args) -> Result<(TcpListener, Option<TlsAcceptor>)> {
-
-    let listenaddr = settings.irc_listen.socket_addrs(|| None)?[0];
-    let listener = TcpListener::bind(listenaddr).await?;
-
-    let acceptor = match settings.irc_listen.scheme() {
-        "tls" => {
-            // openssl genpkey -algorithm ED25519 > example.com.key
-            // openssl req -new -out example.com.csr -key example.com.key
-            // openssl x509 -req -days 700 -in example.com.csr -signkey example.com.key -out example.com.crt
-
-            if settings.irc_tls_secret.is_none() || settings.irc_tls_cert.is_none() {
-                error!("To listen using TLS, please set irc_tls_secret and irc_tls_cert in your config file.");
-                return Err(Error::KeypairPathNotFound)
-            }
-
-            let file = File::open(expand_path(&settings.irc_tls_secret.unwrap())?)?;
-            let mut reader = std::io::BufReader::new(file);
-            let secret = &rustls_pemfile::pkcs8_private_keys(&mut reader)?[0];
-            let secret = rustls::PrivateKey(secret.clone());
-
-            let file = File::open(expand_path(&settings.irc_tls_cert.unwrap())?)?;
-            let mut reader = std::io::BufReader::new(file);
-            let certificate = &rustls_pemfile::certs(&mut reader)?[0];
-            let certificate = rustls::Certificate(certificate.clone());
-
-            let config = rustls::ServerConfig::builder()
-                .with_safe_defaults()
-                .with_no_client_auth()
-                .with_single_cert(vec![certificate], secret)?;
-
-            let acceptor = TlsAcceptor::from(Arc::new(config));
-            Some(acceptor)
-        }
-        _ => None,
-    };
-    Ok((listener, acceptor))
-}
-
-async fn start_listening(ircd: Ircd, executor: Arc<Executor<'_>>, settings: Args) -> Result<()> {
-    let (listener, acceptor) = setup_listener(settings.clone()).await?;
-    info!("[IRC SERVER] listening on {}", settings.irc_listen);
-    loop {
-        let (stream, peer_addr) = match listener.accept().await {
-            Ok((s, a)) => (s, a),
-            Err(e) => {
-                error!("[IRC SERVER] Failed accepting new connections: {}", e);
-                continue
-            }
-        };
-
-        let result = if let Some(acceptor) = acceptor.clone() {
-            let stream = match acceptor.accept(stream).await {
-                Ok(s) => s,
-                Err(e) => {
-                    error!("[IRC SERVER] Failed accepting TLS connection: {}", e);
-                    continue
-                }
-            };
-            ircd.process_new_connection(executor.clone(), stream, peer_addr).await
-        } else {
-            ircd.process_new_connection(executor.clone(), stream, peer_addr).await
-        };
-
-        if let Err(e) = result {
-            error!("[IRC SERVER] Failed processing connection {}: {}", peer_addr, e);
-            continue
-        };
-
-        info!("[IRC SERVER] Accept new connection: {}", peer_addr);
-    }
-}
-
 struct Ircd {
     // msgs
-    seen_msg_ids: SeenIds,
     privmsgs_buffer: ArcPrivmsgsBuffer,
+    seen_msg_ids: SeenIds,
     // channels
     autojoin_chans: Vec<String>,
     configured_chans: FxHashMap<String, ChannelInfo>,
     configured_contacts: FxHashMap<String, ContactInfo>,
     // p2p
     p2p: net::P2pPtr,
-    senders: SubscriberPtr<Privmsg>,
+    p2p_notifiers: SubscriberPtr<Privmsg>,
     password: String,
 }
 
 impl Ircd {
     fn new(
-        seen_msg_ids: SeenIds,
         privmsgs_buffer: ArcPrivmsgsBuffer,
+        seen_msg_ids: SeenIds,
         autojoin_chans: Vec<String>,
         password: String,
         configured_chans: FxHashMap<String, ChannelInfo>,
         configured_contacts: FxHashMap<String, ContactInfo>,
         p2p: net::P2pPtr,
     ) -> Self {
-        let senders = Subscriber::new();
+        let p2p_notifiers = Subscriber::new();
         Self {
-            seen_msg_ids,
             privmsgs_buffer,
+            seen_msg_ids,
             autojoin_chans,
             password,
             configured_chans,
             configured_contacts,
             p2p,
-            senders,
+            p2p_notifiers,
         }
     }
 
-    fn start_p2p_receive_loop(&self, executor: Arc<Executor<'_>>, p2p_receiver: Receiver<Privmsg>) {
-        let senders = self.senders.clone();
+    async fn start(
+        &self,
+        settings: &Args,
+        executor: Arc<Executor<'_>>,
+        p2p_receiver: Receiver<Privmsg>,
+    ) -> Result<()> {
+        let p2p_notifiers = self.p2p_notifiers.clone();
         executor
             .spawn(async move {
                 while let Ok(msg) = p2p_receiver.recv().await {
-                    senders.notify(msg).await;
+                    p2p_notifiers.notify(msg).await;
                 }
             })
             .detach();
-    }
 
-    async fn process_new_connection<C: AsyncRead + AsyncWrite + Send + Unpin + 'static>(
-        &self,
-        executor: Arc<Executor<'_>>,
-        stream: C,
-        peer_addr: SocketAddr,
-    ) -> Result<()> {
-        let (reader, writer) = stream.split();
-
-        let reader = BufReader::new(reader);
-
-        // New subscriber
-        let receiver = self.senders.clone().subscribe().await;
-
-        // New irc connection
-        let conn = IrcServerConnection::new(
-            writer,
-            peer_addr,
-            self.seen_msg_ids.clone(),
+        let irc_server = IrcServer::new(
+            settings.clone(),
             self.privmsgs_buffer.clone(),
+            self.seen_msg_ids.clone(),
             self.autojoin_chans.clone(),
             self.password.clone(),
             self.configured_chans.clone(),
             self.configured_contacts.clone(),
             self.p2p.clone(),
-            self.senders.clone(),
-            receiver.get_id(),
-        );
-
-        executor.spawn(Self::listen(conn, reader, receiver)).detach();
+            self.p2p_notifiers.clone(),
+        )
+        .await?;
 
-        Ok(())
-    }
-
-    async fn listen<C: AsyncRead + AsyncWrite + Send + Unpin + 'static>(
-        mut conn: IrcServerConnection<C>,
-        mut reader: BufReader<ReadHalf<C>>,
-        receiver: Subscription<Privmsg>,
-    ) -> Result<()> {
-        loop {
-            let mut line = String::new();
-
-            futures::select! {
-                msg = receiver.receive().fuse() => {
-                    if let Err(e) = conn.process_msg_from_p2p(&msg).await {
-                        error!("Process msg from p2p failed {}: {}",  conn.peer_address, e);
-                        break
-                    }
-                }
-                err = reader.read_line(&mut line).fuse() => {
-                    if let Err(e) = conn.process_line_from_client(err, line).await {
-                        error!("Process line from client failed {}: {}", conn.peer_address, e);
-                        break
-                    }
-                }
-            }
-        }
-
-        warn!("Close connection for: {}", conn.peer_address);
-        receiver.unsubscribe().await;
+        irc_server.start(executor).await?;
         Ok(())
     }
 }
@@ -346,8 +225,8 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     //
 
     let ircd = Ircd::new(
-        seen_msg_ids.clone(),
         privmsgs_buffer.clone(),
+        seen_msg_ids.clone(),
         settings.autojoin.clone(),
         password.clone(),
         configured_chans.clone(),
@@ -355,8 +234,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         p2p.clone(),
     );
 
-    ircd.start_p2p_receive_loop(executor.clone(), p2p_recv_channel);
-    executor.spawn(start_listening(ircd, executor.clone(), settings.clone())).detach();
+    ircd.start(&settings, executor.clone(), p2p_recv_channel).await?;
 
     // Run once receive exit signal
     let (signal, shutdown) = async_channel::bounded::<()>(1);

bin/ircd/src/settings.rs

@@ -138,7 +138,6 @@ fn parse_priv_key(data: &str) -> Result<String> {
     Ok(pk)
 }
 
-
 /// Parse a TOML string for any configured contact list and return
 /// a map containing said configurations.
 ///