bin/ircd2: WIP implement raft for ircd

Cargo.lock

@@ -2174,6 +2174,27 @@ dependencies = [
  "smol",
 ]
 
+[[package]]
+name = "ircd2"
+version = "0.3.0"
+dependencies = [
+ "async-channel",
+ "async-executor",
+ "async-std",
+ "async-trait",
+ "clap 3.1.8",
+ "ctrlc-async",
+ "darkfi",
+ "easy-parallel",
+ "futures",
+ "fxhash",
+ "log",
+ "rand 0.8.5",
+ "serde_json",
+ "simplelog",
+ "smol",
+]
+
 [[package]]
 name = "itertools"
 version = "0.10.3"

Cargo.toml

@@ -26,6 +26,7 @@ members = [
     "bin/faucetd",
     #"bin/gatewayd",
     "bin/ircd",
+    "bin/ircd2",
     "bin/dnetview",
     "bin/daod",
     "bin/dao-cli",

bin/ircd2/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "ircd2"
+version = "0.3.0"
+homepage = "https://dark.fi"
+description = "P2P IRC daemon"
+authors = ["darkfi <dev@dark.fi>"]
+repository = "https://github.com/darkrenaissance/darkfi"
+license = "AGPL-3.0-only"
+edition = "2021"
+
+[dependencies]
+darkfi = {path = "../../", features = ["net", "rpc", "raft"]}
+# Async
+smol = "1.2.5"
+futures = "0.3.21"
+async-std = "1.11.0"
+async-trait = "0.1.53"
+async-channel = "1.6.1"
+async-executor = "1.4.1"
+easy-parallel = "3.2.0"
+
+# Crypto
+rand = "0.8.5"
+
+# Misc
+clap = {version = "3.1.8", features = ["derive"]}
+log = "0.4.16"
+simplelog = "0.12.0-alpha1"
+fxhash = "0.2.1"
+ctrlc-async = {version= "3.2.2", features = ["termination"]}
+
+# Encoding and parsing
+serde_json = "1.0.79"

bin/ircd2/README.md

@@ -0,0 +1,65 @@
+# p2p IRC
+
+This is a local daemon which can be attached to with any IRC frontend.
+It uses the darkfi p2p engine to synchronize chats between hosts.
+
+## Local Deployment
+
+### Seed Node
+
+First you must run a seed node. The seed node is a static host which nodes can
+connect to when they first connect to the network. The `seed_session` simply
+connects to a seed node and runs `protocol_seed`, which requests a list of
+addresses from the seed node and disconnects straight after receiving them.
+
+    LOG_TARGETS=net cargo run -- -vv --accept 0.0.0.0:9999 --irc 127.0.0.1:6688
+
+Note that the above command doesn't specify an external address since the
+seed node shouldn't be advertised in the list of connectable nodes. The seed
+node does not participate as a normal node in the p2p network. It simply allows
+new nodes to discover other nodes in the network during the bootstrapping phase.
+
+### Inbound Node
+
+This is a node accepting inbound connections on the network but which is not
+making any outbound connections.
+
+The external address is important and must be correct.
+
+    LOG_TARGETS=net cargo run -- -vv --accept 0.0.0.0:11004 --external $LOCAL_IP:11004 --seeds $SEED_IP:9999 --irc 127.0.0.1:6667
+
+### Outbound Node
+
+This is a node which has 8 outbound connection slots and no inbound connections.
+This means the node has 8 slots which will actively search for unique nodes to
+connect to in the p2p network.
+
+    LOG_TARGETS=net cargo run -- -vv --slots 5 --seeds $SEED_IP:9999 --irc 127.0.0.1:6668
+
+### Attaching the IRC Frontend
+
+Assuming you have run the above 3 commands to create a small model testnet,
+and both inbound and outbound nodes above are connected, you can test them
+out using weechat.
+
+To create separate weechat instances, use the `--dir` command:
+
+    weechat --dir /tmp/a/
+    weechat --dir /tmp/b/
+
+Then in both clients, you must set the option to connect to temporary servers:
+
+    /set irc.look.temporary_servers on
+
+Finally you can attach to the local IRCd instances:
+
+    /connect localhost/6667
+    /connect localhost/6668
+
+And send messages to yourself.
+
+### Running a Fullnode
+
+See the script `script/run_node.sh` for an example of how to deploy a full node which
+does seed session synchronization, and accepts both inbound and outbound
+connections.

bin/ircd2/script/run_node.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+
+# Change this value to the hostname of the seed server
+SEED_HOSTNAME=XXX
+
+LOCAL_IP=$(ip route get 8.8.8.8 | head -1 | awk '{print $7}')
+SEED_IP=$(getent hosts $SEED_HOSTNAME.local | awk '{print $1}' | head -n 1)
+cargo run -- --accept 0.0.0.0:11004 --slots 5 --external $LOCAL_IP:11004 --seeds $SEED_IP:9999 --irc 127.0.0.1:6667
+

bin/ircd2/script/run_seed_node.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+LOCAL_IP=$(ip route get 8.8.8.8 | head -1 | awk '{print $7}')
+cargo run -- --accept 0.0.0.0:9999 --irc 127.0.0.1:6688
+

bin/ircd2/script/tmux_session.sh

@@ -0,0 +1,6 @@
+#!/bin/sh
+
+tmux new-session -d 'LOG_TARGETS=net ../../../target/release/ircd -vv --accept 127.0.0.1:9999 --irc 127.0.0.1:6688'
+tmux split-window -v 'LOG_TARGETS=net ../../../target/release/ircd -vv --accept 127.0.0.1:11004 --external 127.0.0.1:11004 --seeds 127.0.0.1:9999 --irc 127.0.0.1:6667'
+tmux split-window -h 'LOG_TARGETS=net ../../../target/release/ircd -vv --slots 5 --seeds 127.0.0.1:9999 --irc 127.0.0.1:6668'
+tmux attach 

bin/ircd2/src/main.rs

@@ -0,0 +1,224 @@
+use async_std::net::{TcpListener, TcpStream};
+use std::{net::SocketAddr, sync::Arc};
+
+use async_channel::Receiver;
+use async_executor::Executor;
+use clap::Parser;
+use easy_parallel::Parallel;
+use futures::{io::BufReader, AsyncBufReadExt, AsyncReadExt, FutureExt};
+use log::{debug, error, info, warn};
+use simplelog::{ColorChoice, TermLogger, TerminalMode};
+
+use darkfi::{
+    cli_desc, net,
+    raft::Raft,
+    rpc::rpcserver::{listen_and_serve, RpcServerConfig},
+    util::cli::log_config,
+    Error, Result,
+};
+
+pub(crate) mod privmsg;
+pub(crate) mod rpc;
+pub(crate) mod server;
+
+use crate::{privmsg::Privmsg, rpc::JsonRpcInterface, server::IrcServerConnection};
+
+#[derive(Parser)]
+#[clap(name = "ircd", about = cli_desc!(), version)]
+struct Args {
+    /// Accept address
+    #[clap(short, long)]
+    accept: Option<SocketAddr>,
+
+    /// Seed node (repeatable)
+    #[clap(short, long)]
+    seed: Vec<SocketAddr>,
+
+    /// Manual connection (repeatable)
+    #[clap(short, long)]
+    connect: Vec<SocketAddr>,
+
+    /// Connection slots
+    #[clap(long, default_value_t = 0)]
+    slots: u32,
+
+    /// External address
+    #[clap(short, long)]
+    external: Option<SocketAddr>,
+
+    /// IRC listen address
+    #[clap(short = 'r', long, default_value = "127.0.0.1:6667")]
+    irc: SocketAddr,
+
+    /// RPC listen address
+    #[clap(long, default_value = "127.0.0.1:8000")]
+    rpc: SocketAddr,
+
+    /// Verbosity level
+    #[clap(short, parse(from_occurrences))]
+    verbose: u8,
+}
+
+async fn process_user_input(
+    mut line: String,
+    peer_addr: SocketAddr,
+    conn: &mut IrcServerConnection,
+    sender: async_channel::Sender<Privmsg>,
+) -> Result<()> {
+    if line.is_empty() {
+        warn!("Received empty line from {}. Closing connection.", peer_addr);
+        return Err(Error::ChannelStopped)
+    }
+
+    assert!(&line[(line.len() - 2)..] == "\r\n");
+    // Remove CRLF
+    line.pop();
+    line.pop();
+
+    debug!("Received '{}' from {}", line, peer_addr);
+
+    if let Err(e) = conn.update(line, sender).await {
+        warn!("Connection error: {} for {}", e, peer_addr);
+        return Err(Error::ChannelStopped)
+    }
+
+    Ok(())
+}
+
+async fn process(
+    receiver: Receiver<Privmsg>,
+    stream: TcpStream,
+    peer_addr: SocketAddr,
+    sender: async_channel::Sender<Privmsg>,
+) -> Result<()> {
+    let (reader, writer) = stream.split();
+
+    let mut reader = BufReader::new(reader);
+    let mut conn = IrcServerConnection::new(writer);
+
+    loop {
+        let mut line = String::new();
+        futures::select! {
+            privmsg = receiver.recv().fuse() => {
+                let msg = privmsg.expect("internal message queue error");
+                debug!("ABOUT TO SEND: {:?}", msg);
+                let irc_msg = format!(":{}!anon@dark.fi PRIVMSG {} :{}\r\n",
+                                      msg.nickname,
+                                      msg.channel,
+                                      msg.message,
+                                      );
+
+                conn.reply(&irc_msg).await?;
+            }
+
+            err = reader.read_line(&mut line).fuse() => {
+                if let Err(e) = err {
+                    warn!("Read line error. Closing stream for {}: {}", peer_addr, e);
+                    return Ok(())
+                }
+
+                process_user_input(line, peer_addr, &mut conn, sender.clone()).await?;
+            }
+        };
+    }
+}
+
+async fn start(executor: Arc<Executor<'_>>, args: Args, net_settings: net::Settings) -> Result<()> {
+    let listener = TcpListener::bind(args.irc).await?;
+    let local_addr = listener.local_addr()?;
+    info!("Listening on {}", local_addr);
+
+    //
+    // Raft
+    //
+    let mut raft = Raft::<Privmsg>::new(net_settings.inbound, std::path::PathBuf::from("msgs.db"))?;
+
+    let raft_sender = raft.get_broadcast();
+    let commits = raft.get_commits();
+
+    //
+    // RPC interface
+
+    let rpc_config = RpcServerConfig {
+        socket_addr: args.rpc,
+        // TODO: Use net/transport:
+        use_tls: false,
+        identity_path: Default::default(),
+        identity_pass: Default::default(),
+    };
+    let executor_cloned = executor.clone();
+    let rpc_interface = Arc::new(JsonRpcInterface { addr: args.rpc });
+    let rpc_task = executor.spawn(async move {
+        listen_and_serve(rpc_config, rpc_interface, executor_cloned.clone()).await
+    });
+
+    //
+    // IRC instance
+    //
+    let executor_cloned = executor.clone();
+    let irc_task: smol::Task<Result<()>> = executor.spawn(async move {
+        loop {
+            let (stream, peer_addr) = match listener.accept().await {
+                Ok((s, a)) => (s, a),
+                Err(e) => {
+                    error!("Failed listening for connections: {}", e);
+                    return Err(Error::ServiceStopped)
+                }
+            };
+
+            info!("Accepted client: {}", peer_addr);
+
+            executor_cloned
+                .spawn(process(commits.clone(), stream, peer_addr, raft_sender.clone()))
+                .detach();
+        }
+    });
+
+    let stop_signal = async_channel::bounded::<()>(10);
+
+    ctrlc_async::set_async_handler(async move {
+        warn!(target: "ircd", "ircd start() Exit Signal");
+        // cleaning up tasks running in the background
+        stop_signal.0.send(()).await.expect("send exit signal to raft");
+        rpc_task.cancel().await;
+        irc_task.cancel().await;
+    })
+    .expect("handle exit signal");
+
+    // blocking
+    raft.start(net_settings.clone(), executor.clone(), stop_signal.1.clone()).await?;
+
+    Ok(())
+}
+
+fn main() -> Result<()> {
+    let args = Args::parse();
+
+    let (lvl, conf) = log_config(args.verbose.into())?;
+    TermLogger::init(lvl, conf, TerminalMode::Mixed, ColorChoice::Auto)?;
+
+    let net_settings = net::Settings {
+        inbound: args.accept,
+        outbound_connections: args.slots,
+        external_addr: args.external,
+        peers: args.connect.clone(),
+        seeds: args.seed.clone(),
+        ..Default::default()
+    };
+
+    let ex = Arc::new(Executor::new());
+    let ex_clone = ex.clone();
+    let (signal, shutdown) = async_channel::unbounded::<()>();
+    let (_, result) = Parallel::new()
+        .each(0..4, |_| smol::future::block_on(ex.run(shutdown.recv())))
+        // Run the main future on the current thread.
+        .finish(|| {
+            smol::future::block_on(async move {
+                start(ex_clone.clone(), args, net_settings).await?;
+                drop(signal);
+                Ok::<(), darkfi::Error>(())
+            })
+        });
+
+    result
+}

bin/ircd2/src/privmsg.rs

@@ -0,0 +1,11 @@
+use darkfi::util::serial::{SerialDecodable, SerialEncodable};
+
+pub type PrivmsgId = u32;
+
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct Privmsg {
+    pub id: PrivmsgId,
+    pub nickname: String,
+    pub channel: String,
+    pub message: String,
+}

bin/ircd2/src/rpc.rs

@@ -0,0 +1,53 @@
+use std::{net::SocketAddr, sync::Arc};
+
+use async_executor::Executor;
+use async_trait::async_trait;
+use log::debug;
+use serde_json::{json, Value};
+
+use darkfi::rpc::{
+    jsonrpc,
+    jsonrpc::{ErrorCode, JsonRequest, JsonResult},
+    rpcserver::RequestHandler,
+};
+
+pub struct JsonRpcInterface {
+    pub addr: SocketAddr,
+}
+
+#[async_trait]
+impl RequestHandler for JsonRpcInterface {
+    async fn handle_request(&self, req: JsonRequest, _executor: Arc<Executor<'_>>) -> JsonResult {
+        if req.params.as_array().is_none() {
+            return jsonrpc::error(ErrorCode::InvalidRequest, None, req.id).into()
+        }
+
+        debug!(target: "RPC", "--> {}", serde_json::to_string(&req).unwrap());
+
+        match req.method.as_str() {
+            Some("ping") => self.pong(req.id, req.params).await,
+            //Some("get_info") => self.get_info(req.id, req.params).await,
+            Some(_) | None => jsonrpc::error(ErrorCode::MethodNotFound, None, req.id).into(),
+        }
+    }
+}
+
+impl JsonRpcInterface {
+    // RPCAPI:
+    // Replies to a ping method.
+    // --> {"jsonrpc": "2.0", "method": "ping", "params": [], "id": 42}
+    // <-- {"jsonrpc": "2.0", "result": "pong", "id": 42}
+    async fn pong(&self, id: Value, _params: Value) -> JsonResult {
+        jsonrpc::response(json!("pong"), id).into()
+    }
+
+    // TODO
+    // RPCAPI:
+    // Retrieves P2P network information.
+    // --> {"jsonrpc": "2.0", "method": "get_info", "params": [], "id": 42}
+    // <-- {"jsonrpc": "2.0", result": {"nodeID": [], "nodeinfo": [], "id": 42}
+    //async fn get_info(&self, id: Value, _params: Value) -> JsonResult {
+    //    let resp = self.p2p.get_info().await;
+    //    jsonrpc::response(resp, id).into()
+    //}
+}

bin/ircd2/src/server.rs

@@ -0,0 +1,135 @@
+use async_std::net::TcpStream;
+use futures::{io::WriteHalf, AsyncWriteExt};
+use log::{debug, info, warn};
+use rand::{rngs::OsRng, RngCore};
+
+use darkfi::{Error, Result};
+
+use crate::privmsg::Privmsg;
+
+pub struct IrcServerConnection {
+    write_stream: WriteHalf<TcpStream>,
+    is_nick_init: bool,
+    is_user_init: bool,
+    is_registered: bool,
+    nickname: String,
+    _channels: Vec<String>,
+}
+
+impl IrcServerConnection {
+    pub fn new(write_stream: WriteHalf<TcpStream>) -> Self {
+        Self {
+            write_stream,
+            is_nick_init: false,
+            is_user_init: false,
+            is_registered: false,
+            nickname: "".to_string(),
+            _channels: vec![],
+        }
+    }
+
+    pub async fn update(
+        &mut self,
+        line: String,
+        sender: async_channel::Sender<Privmsg>,
+    ) -> Result<()> {
+        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)?;
+
+        debug!("Received command: {}", command);
+
+        match command {
+            "USER" => {
+                // We can stuff any extra things like public keys in here.
+                // Ignore it for now.
+                self.is_user_init = true;
+            }
+            "NICK" => {
+                let nickname = tokens.next().ok_or(Error::MalformedPacket)?;
+                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?;
+            }
+            "JOIN" => {
+                // Ignore since channels are all autojoin
+                // let channel = tokens.next().ok_or(Error::MalformedPacket)?;
+                // self.channels.push(channel.to_string());
+
+                // let join_reply = format!(":{}!anon@dark.fi JOIN {}\r\n", self.nickname, channel);
+                // self.reply(&join_reply).await?;
+
+                // self.write_stream.write_all(b":f00!f00@127.0.01 PRIVMSG #dev :y0\r\n").await?;
+            }
+            "PING" => {
+                let line_clone = line.clone();
+                let split_line: Vec<&str> = line_clone.split_whitespace().collect();
+                if split_line.len() > 1 && split_line[0] == "PING" {
+                    let pong = format!("PONG {}\r\n", split_line[1]);
+                    self.reply(&pong).await?;
+                }
+            }
+            "PRIVMSG" => {
+                let channel = tokens.next().ok_or(Error::MalformedPacket)?;
+                let substr_idx = line.find(':').ok_or(Error::MalformedPacket)?;
+
+                if substr_idx >= line.len() {
+                    return Err(Error::MalformedPacket)
+                }
+
+                let message = &line[substr_idx + 1..];
+                info!("Message {}: {}", channel, message);
+
+                let random_id = OsRng.next_u32();
+
+                let protocol_msg = Privmsg {
+                    id: random_id,
+                    nickname: self.nickname.clone(),
+                    channel: channel.to_string(),
+                    message: message.to_string(),
+                };
+
+                sender.send(protocol_msg).await?;
+            }
+            "QUIT" => {
+                // Close the connection
+                return Err(Error::ServiceStopped)
+            }
+            _ => {
+                warn!("Unimplemented `{}` command", command);
+            }
+        }
+
+        if !self.is_registered && 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;
+
+            // Auto-joins
+            macro_rules! autojoin {
+                ($channel:expr,$topic:expr) => {
+                    let j = format!(":{}!anon@dark.fi JOIN {}\r\n", self.nickname, $channel);
+                    let t = format!(":DarkFi TOPIC {} :{}\r\n", $channel, $topic);
+                    self.reply(&j).await?;
+                    self.reply(&t).await?;
+                };
+            }
+
+            autojoin!("#dev", "Development of DarkFi");
+            autojoin!("#markets", "Markets, trading, DeFi, algo, biz, finance, and economics");
+            autojoin!("#memes", "Memetic engineering");
+        }
+
+        Ok(())
+    }
+
+    pub async fn reply(&mut self, message: &str) -> Result<()> {
+        self.write_stream.write_all(message.as_bytes()).await?;
+        debug!("Sent {}", message);
+        Ok(())
+    }
+}