bin/ircd: major fix on protocol privmsg & more clear log messages

bin/ircd/src/irc_server/command.rs

@@ -29,7 +29,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
             self.is_user_init = true;
         } else {
             // Close the connection
-            warn!("Password is required");
+            warn!("[IRC SERVER] Password is required");
             return self.on_quit()
         }
         Ok(())
@@ -40,7 +40,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
             self.is_pass_init = true
         } else {
             // Close the connection
-            warn!("Password is not correct!");
+            warn!("[IRC SERVER] Password is not correct!");
             return self.on_quit()
         }
         Ok(())
@@ -222,7 +222,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
 
         let message = line[substr_idx + 1..].to_string();
 
-        info!("(Plain) PRIVMSG {} :{}", target, message);
+        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;
@@ -243,7 +243,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
 
             if let Some(salt_box) = &channel_info.salt_box {
                 encrypt_privmsg(salt_box, &mut privmsg);
-                info!("(Encrypted) PRIVMSG: {:?}", privmsg);
+                info!("[CLIENT {}] (Encrypted) PRIVMSG: {:?}", self.peer_address, privmsg);
             }
         } else {
             if !self.configured_contacts.contains_key(target) {
@@ -253,7 +253,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
             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!("(Encrypted) PRIVMSG: {:?}", privmsg);
+                info!("[CLIENT {}] (Encrypted) PRIVMSG: {:?}", self.peer_address, privmsg);
             }
         }
 

bin/ircd/src/irc_server/mod.rs

@@ -80,7 +80,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
     }
 
     pub async fn process_msg_from_p2p(&mut self, msg: &Privmsg) -> Result<()> {
-        info!("Received msg from P2p network: {:?}", msg);
+        info!("[P2P] Received: {}", msg.to_string().trim());
 
         let mut msg = msg.clone();
         let mut contact = String::new();
@@ -123,7 +123,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
                 decrypt_privmsg(salt_box, &mut msg);
                 // This is for /query
                 msg.nickname = contact;
-                info!("Decrypted received message: {:?}", msg);
+                info!("[P2P] Decrypted received message: {:?}", msg);
             }
 
             self.reply(&msg.to_string()).await?;
@@ -138,15 +138,22 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
         line: String,
     ) -> Result<()> {
         if let Err(e) = err {
-            warn!("Read line error {}: {}", self.peer_address, e);
+            warn!("[CLIENT {}] 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)?;
+        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!("Connection error: {} for {}", e, self.peer_address);
+            warn!("[CLIENT {}] Connection error: {}", self.peer_address, e);
             return Err(Error::ChannelStopped)
         }
         Ok(())
@@ -163,7 +170,6 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
 
         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?,
@@ -177,7 +183,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
             "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),
+            _ => warn!("[CLIENT {}] Unimplemented `{}` command", self.peer_address, command),
         }
 
         self.registre().await?;
@@ -223,15 +229,12 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
 // Helper functions
 //
 
-fn clean_input_line(mut line: String, peer_address: &SocketAddr) -> Result<String> {
+fn clean_input_line(mut line: String) -> Result<String> {
     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)
     }
 
@@ -242,7 +245,6 @@ fn clean_input_line(mut line: String, peer_address: &SocketAddr) -> Result<Strin
     } else if &line[(line.len() - 1)..] == "\n" {
         line.pop();
     } else {
-        warn!("Closing connection.");
         return Err(Error::ChannelStopped)
     }
 

bin/ircd/src/main.rs

@@ -108,12 +108,12 @@ async fn setup_listener(settings: Args) -> Result<(TcpListener, Option<TlsAccept
 
 async fn start_listening(ircd: Ircd, executor: Arc<Executor<'_>>, settings: Args) -> Result<()> {
     let (listener, acceptor) = setup_listener(settings.clone()).await?;
-    info!("IRC listening on {}", settings.irc_listen);
+    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!("failed accepting new connections: {}", e);
+                error!("[IRC SERVER] Failed accepting new connections: {}", e);
                 continue
             }
         };
@@ -122,7 +122,7 @@ async fn start_listening(ircd: Ircd, executor: Arc<Executor<'_>>, settings: Args
             let stream = match acceptor.accept(stream).await {
                 Ok(s) => s,
                 Err(e) => {
-                    error!("Failed accepting TLS connection: {}", e);
+                    error!("[IRC SERVER] Failed accepting TLS connection: {}", e);
                     continue
                 }
             };
@@ -132,11 +132,11 @@ async fn start_listening(ircd: Ircd, executor: Arc<Executor<'_>>, settings: Args
         };
 
         if let Err(e) = result {
-            error!("Failed processing connection {}: {}", peer_addr, e);
+            error!("[IRC SERVER] Failed processing connection {}: {}", peer_addr, e);
             continue
         };
 
-        info!("IRC Accepted new client: {}", peer_addr);
+        info!("[IRC SERVER] Accept new connection: {}", peer_addr);
     }
 }
 
@@ -245,9 +245,8 @@ impl Ircd {
             }
         }
 
-        warn!("Close connection for clinet {}", conn.peer_address);
+        warn!("Close connection for: {}", conn.peer_address);
         receiver.unsubscribe().await;
-
         Ok(())
     }
 }
@@ -255,6 +254,7 @@ impl Ircd {
 async_daemonize!(realmain);
 async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let seen_msg_ids = Arc::new(Mutex::new(RingBuffer::new(SIZE_OF_MSG_IDSS_BUFFER)));
+    let seen_inv_ids = Arc::new(Mutex::new(RingBuffer::new(SIZE_OF_MSG_IDSS_BUFFER)));
     let privmsgs_buffer = PrivmsgsBuffer::new();
     let unread_msgs = Arc::new(Mutex::new(FxHashMap::default()));
 
@@ -304,12 +304,14 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let registry = p2p.protocol_registry();
 
     let seen_msg_ids_cloned = seen_msg_ids.clone();
+    let seen_inv_ids_cloned = seen_inv_ids.clone();
     let privmsgs_buffer_cloned = privmsgs_buffer.clone();
     let unread_msgs_cloned = unread_msgs.clone();
     registry
         .register(net::SESSION_ALL, move |channel, p2p| {
             let sender = p2p_send_channel.clone();
             let seen_msg_ids_cloned = seen_msg_ids_cloned.clone();
+            let seen_inv_ids_cloned = seen_inv_ids_cloned.clone();
             let privmsgs_buffer_cloned = privmsgs_buffer_cloned.clone();
             let unread_msgs_cloned = unread_msgs_cloned.clone();
             async move {
@@ -318,6 +320,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
                     sender,
                     p2p,
                     seen_msg_ids_cloned,
+                    seen_inv_ids_cloned,
                     privmsgs_buffer_cloned,
                     unread_msgs_cloned,
                 )

bin/ircd/src/protocol_privmsg.rs

@@ -4,6 +4,7 @@ use async_executor::Executor;
 use async_trait::async_trait;
 use chrono::Utc;
 use log::debug;
+use rand::{rngs::OsRng, RngCore};
 use ripemd::{Digest, Ripemd160};
 
 use darkfi::{
@@ -24,18 +25,20 @@ const MAX_CONFIRM: u8 = 4;
 const SLEEP_TIME_FOR_RESEND: u64 = 1200;
 const UNREAD_MSG_EXPIRE_TIME: i64 = 259200;
 
-#[derive(SerialDecodable, SerialEncodable, Clone)]
+#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
 struct Inv {
+    id: u64,
     invs: Vec<InvObject>,
 }
 
 impl Inv {
     fn new(invs: Vec<InvObject>) -> Self {
-        Self { invs }
+        let id = OsRng.next_u64();
+        Self { id, invs }
     }
 }
 
-#[derive(SerialDecodable, SerialEncodable, Clone)]
+#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
 struct GetData {
     invs: Vec<InvObject>,
 }
@@ -46,7 +49,7 @@ impl GetData {
     }
 }
 
-#[derive(SerialDecodable, SerialEncodable, Clone)]
+#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
 struct InvObject(String);
 
 pub struct ProtocolPrivmsg {
@@ -57,6 +60,7 @@ pub struct ProtocolPrivmsg {
     getdata_sub: net::MessageSubscription<GetData>,
     p2p: net::P2pPtr,
     msg_ids: SeenIds,
+    inv_ids: SeenIds,
     msgs: ArcPrivmsgsBuffer,
     unread_msgs: UnreadMsgs,
     channel: net::ChannelPtr,
@@ -68,28 +72,32 @@ impl ProtocolPrivmsg {
         notify: async_channel::Sender<Privmsg>,
         p2p: net::P2pPtr,
         msg_ids: SeenIds,
+        inv_ids: SeenIds,
         msgs: ArcPrivmsgsBuffer,
         unread_msgs: UnreadMsgs,
     ) -> net::ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
         message_subsytem.add_dispatch::<Privmsg>().await;
+        message_subsytem.add_dispatch::<Inv>().await;
+        message_subsytem.add_dispatch::<GetData>().await;
 
         let msg_sub =
-            channel.subscribe_msg::<Privmsg>().await.expect("Missing Privmsg dispatcher!");
-
-        let inv_sub = channel.subscribe_msg::<Inv>().await.expect("Missing Inv dispatcher!");
+            channel.clone().subscribe_msg::<Privmsg>().await.expect("Missing Privmsg dispatcher!");
 
         let getdata_sub =
-            channel.subscribe_msg::<GetData>().await.expect("Missing Inv dispatcher!");
+            channel.clone().subscribe_msg::<GetData>().await.expect("Missing GetData dispatcher!");
+
+        let inv_sub = channel.subscribe_msg::<Inv>().await.expect("Missing Inv dispatcher!");
 
         Arc::new(Self {
             notify,
             msg_sub,
             inv_sub,
+            getdata_sub,
             jobsman: net::ProtocolJobsManager::new("ProtocolPrivmsg", channel.clone()),
             p2p,
             msg_ids,
-            getdata_sub,
+            inv_ids,
             msgs,
             unread_msgs,
             channel,
@@ -98,10 +106,18 @@ impl ProtocolPrivmsg {
 
     async fn handle_receive_inv(self: Arc<Self>) -> Result<()> {
         debug!(target: "ircd", "ProtocolPrivmsg::handle_receive_inv() [START]");
+        let exclude_list = vec![self.channel.address()];
         loop {
             let inv = self.inv_sub.receive().await?;
             let inv = (*inv).to_owned();
 
+            let mut inv_ids = self.inv_ids.lock().await;
+            if inv_ids.contains(&inv.id) {
+                continue
+            }
+            inv_ids.push(inv.id);
+            drop(inv_ids);
+
             let mut inv_requested = vec![];
             for inv_object in inv.invs.iter() {
                 let mut msgs = self.unread_msgs.lock().await;
@@ -117,6 +133,8 @@ impl ProtocolPrivmsg {
             }
 
             self.update_unread_msgs().await?;
+
+            self.p2p.broadcast_with_exclude(inv, &exclude_list).await?;
         }
     }
 
@@ -125,7 +143,7 @@ impl ProtocolPrivmsg {
         let exclude_list = vec![self.channel.address()];
         loop {
             let msg = self.msg_sub.receive().await?;
-            let msg = (*msg).to_owned();
+            let mut msg = (*msg).to_owned();
 
             let mut msg_ids = self.msg_ids.lock().await;
             if msg_ids.contains(&msg.id) {
@@ -134,13 +152,15 @@ impl ProtocolPrivmsg {
             msg_ids.push(msg.id);
             drop(msg_ids);
 
-            if msg.read_confirms > MAX_CONFIRM {
+            if msg.read_confirms >= MAX_CONFIRM {
                 self.add_to_msgs(&msg).await?;
             } else {
+                msg.read_confirms += 1;
                 let hash = self.add_to_unread_msgs(&msg).await;
-                self.channel.send(Inv::new(vec![InvObject(hash)])).await?;
+                self.p2p.broadcast(Inv::new(vec![InvObject(hash)])).await?;
             }
 
+            self.update_unread_msgs().await?;
             self.p2p.broadcast_with_exclude(msg, &exclude_list).await?;
         }
     }
@@ -176,7 +196,7 @@ impl ProtocolPrivmsg {
                 msgs.remove(&hash);
                 continue
             }
-            if msg.read_confirms > MAX_CONFIRM {
+            if msg.read_confirms >= MAX_CONFIRM {
                 self.add_to_msgs(&msg).await?;
                 msgs.remove(&hash);
             }

bin/ircd/src/settings.rs

@@ -121,7 +121,11 @@ fn parse_priv_key(data: &str) -> Result<String> {
         _ => return Ok(pk),
     };
 
-    if !map.contains_key("private_key") && !map["private_key"].is_table() {
+    if !map.contains_key("private_key") {
+        return Ok(pk)
+    }
+
+    if !map["private_key"].is_table() {
         return Ok(pk)
     }
 
@@ -150,7 +154,11 @@ pub fn parse_configured_contacts(data: &str) -> Result<FxHashMap<String, Contact
         _ => return Ok(ret),
     };
 
-    if !map.contains_key("contact") && !map["contact"].is_table() {
+    if !map.contains_key("contact") {
+        return Ok(ret)
+    }
+
+    if !map["contact"].is_table() {
         return Ok(ret)
     }
 
@@ -198,7 +206,11 @@ pub fn parse_configured_channels(data: &str) -> Result<FxHashMap<String, Channel
         _ => return Ok(ret),
     };
 
-    if !map.contains_key("channel") && !map["channel"].is_table() {
+    if !map.contains_key("channel") {
+        return Ok(ret)
+    }
+
+    if !map["channel"].is_table() {
         return Ok(ret)
     }