/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2024 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::{collections::HashSet, io::Write, path::PathBuf, sync::Arc};
use darkfi::{
async_daemonize, cli_desc,
event_graph::{proto::ProtocolEventGraph, EventGraph, EventGraphPtr},
net::{session::SESSION_DEFAULT, settings::SettingsOpt, P2p, P2pPtr},
rpc::{
jsonrpc::JsonSubscriber,
server::{listen_and_serve, RequestHandler},
},
system::{sleep, StoppableTask, StoppableTaskPtr},
util::path::{expand_path, get_config_path},
Error, Result,
};
use log::{debug, error, info};
use rand::rngs::OsRng;
use smol::{fs, lock::Mutex, stream::StreamExt, Executor};
use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
use url::Url;
const CONFIG_FILE: &str = "darkirc_config.toml";
const CONFIG_FILE_CONTENTS: &str = include_str!("../darkirc_config.toml");
/// IRC server and client handler implementation
mod irc;
use irc::server::IrcServer;
/// Cryptography utilities
mod crypto;
use crypto::bcrypt::bcrypt_hash_password;
// RLN
//mod rln;
/// JSON-RPC methods
mod rpc;
/// Settings utilities
mod settings;
#[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
#[serde(default)]
#[structopt(name = "darkirc", about = cli_desc!())]
struct Args {
#[structopt(short, parse(from_occurrences))]
/// Increase verbosity (-vvv supported)
verbose: u8,
#[structopt(short, long)]
/// Configuration file to use
config: Option,
#[structopt(long)]
/// Set log file output
log: Option,
#[structopt(long, default_value = "tcp://127.0.0.1:26660")]
/// RPC server listen address
rpc_listen: Url,
#[structopt(long, default_value = "tcp://127.0.0.1:6667")]
/// IRC server listen address
irc_listen: Url,
/// Optional TLS certificate file path if `irc_listen` uses TLS
irc_tls_cert: Option,
/// Optional TLS certificate key file path if `irc_listen` uses TLS
irc_tls_secret: Option,
#[structopt(short, long, default_value = "~/.local/darkfi/darkirc_db")]
/// Datastore (DB) path
datastore: String,
#[structopt(short, long, default_value = "~/.local/darkfi/replayed_darkirc_db")]
/// Replay logs (DB) path
replay_datastore: String,
/// Flag to store Sled DB instructions
#[structopt(long)]
replay_mode: bool,
/// Generate a new NaCl keypair and exit
#[structopt(long)]
gen_chacha_keypair: bool,
/// Generate a new encrypted channel NaCl secret and exit
#[structopt(long)]
gen_channel_secret: bool,
/// Recover NaCl public key from a secret key
#[structopt(long = "get-chacha-pubkey")]
chacha_secret: Option,
/// Flag to skip syncing the DAG (no history).
#[structopt(long)]
skip_dag_sync: bool,
/// Number of attempts to sync the DAG.
#[structopt(long, default_value = "5")]
sync_attempts: u8,
/// Number of seconds to wait before trying again if sync fails.
#[structopt(long, default_value = "10")]
sync_timeout: u8,
/// IRC Password (Encrypted with bcrypt-2b)
#[structopt(long)]
pub password: Option,
/// Encrypt a given password for the IRC server connection
#[structopt(long)]
encrypt_password: bool,
/// P2P network settings
#[structopt(flatten)]
net: SettingsOpt,
}
pub struct DarkIrc {
/// P2P network pointer
p2p: P2pPtr,
/// Sled DB (also used in event_graph and for RLN)
sled: sled::Db,
/// Event Graph instance
event_graph: EventGraphPtr,
/// JSON-RPC connection tracker
rpc_connections: Mutex>,
/// dnet JSON-RPC subscriber
dnet_sub: JsonSubscriber,
/// deg JSON-RPC subscriber
deg_sub: JsonSubscriber,
/// Replay logs (DB) path
replay_datastore: PathBuf,
}
impl DarkIrc {
fn new(
p2p: P2pPtr,
sled: sled::Db,
event_graph: EventGraphPtr,
dnet_sub: JsonSubscriber,
deg_sub: JsonSubscriber,
replay_datastore: PathBuf,
) -> Self {
Self {
p2p,
sled,
event_graph,
rpc_connections: Mutex::new(HashSet::new()),
dnet_sub,
deg_sub,
replay_datastore,
}
}
}
async_daemonize!(realmain);
async fn realmain(args: Args, ex: Arc>) -> Result<()> {
if args.gen_chacha_keypair {
let secret = crypto_box::SecretKey::generate(&mut OsRng);
let public = secret.public_key();
let secret = bs58::encode(secret.to_bytes()).into_string();
let public = bs58::encode(public.to_bytes()).into_string();
println!("Place this in your config file:\n");
println!("[crypto]");
println!("#dm_chacha_public = \"{}\"", public);
println!("dm_chacha_secret = \"{}\"", secret);
return Ok(())
}
if args.gen_channel_secret {
let secret = crypto_box::SecretKey::generate(&mut OsRng);
let secret = bs58::encode(secret.to_bytes()).into_string();
println!("Place this in your config file:\n");
println!("[channel.\"#yourchannelname\"]");
println!("secret = \"{}\"", secret);
return Ok(())
}
if let Some(chacha_secret) = args.chacha_secret {
let bytes = match bs58::decode(chacha_secret).into_vec() {
Ok(v) => v,
Err(e) => {
println!("Error: {}", e);
return Err(Error::ParseFailed("Secret key parsing failed"))
}
};
if bytes.len() != 32 {
return Err(Error::ParseFailed("Decoded base58 is not 32 bytes long"))
}
let secret: [u8; 32] = bytes.try_into().unwrap();
let secret = crypto_box::SecretKey::from(secret);
println!("{}", bs58::encode(secret.public_key().to_bytes()).into_string());
return Ok(())
}
if args.encrypt_password {
let mut pw = String::new();
print!("Enter password: ");
std::io::stdout().flush()?;
std::io::stdin().read_line(&mut pw)?;
if let Some('\n') = pw.chars().next_back() {
pw.pop();
}
if let Some('\r') = pw.chars().next_back() {
pw.pop();
}
println!("{}", bcrypt_hash_password(pw));
std::io::stdout().flush()?;
return Ok(())
}
info!("Initializing DarkIRC node");
// Create datastore path if not there already.
let datastore = expand_path(&args.datastore)?;
fs::create_dir_all(&datastore).await?;
let replay_datastore = expand_path(&args.replay_datastore)?;
let replay_mode = args.replay_mode;
info!("Instantiating event DAG");
let sled_db = sled::open(datastore)?;
let mut p2p_settings: darkfi::net::Settings = args.net.into();
p2p_settings.app_version = semver::Version::parse(env!("CARGO_PKG_VERSION")).unwrap();
let p2p = P2p::new(p2p_settings, ex.clone()).await;
let event_graph = EventGraph::new(
p2p.clone(),
sled_db.clone(),
replay_datastore.clone(),
replay_mode,
"darkirc_dag",
1,
ex.clone(),
)
.await?;
info!("Registering EventGraph P2P protocol");
let event_graph_ = Arc::clone(&event_graph);
let registry = p2p.protocol_registry();
registry
.register(SESSION_DEFAULT, move |channel, _| {
let event_graph_ = event_graph_.clone();
async move { ProtocolEventGraph::init(event_graph_, channel).await.unwrap() }
})
.await;
info!("Starting dnet subs task");
let dnet_sub = JsonSubscriber::new("dnet.subscribe_events");
let dnet_sub_ = dnet_sub.clone();
let p2p_ = p2p.clone();
let dnet_task = StoppableTask::new();
dnet_task.clone().start(
async move {
let dnet_sub = p2p_.dnet_subscribe().await;
loop {
let event = dnet_sub.receive().await;
debug!("Got dnet event: {:?}", event);
dnet_sub_.notify(vec![event.into()].into()).await;
}
},
|res| async {
match res {
Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
Err(e) => panic!("{}", e),
}
},
Error::DetachedTaskStopped,
ex.clone(),
);
info!("Starting deg subs task");
let deg_sub = JsonSubscriber::new("deg.subscribe_events");
let deg_sub_ = deg_sub.clone();
let event_graph_ = event_graph.clone();
let deg_task = StoppableTask::new();
deg_task.clone().start(
async move {
let deg_sub = event_graph_.deg_subscribe().await;
loop {
let event = deg_sub.receive().await;
debug!("Got deg event: {:?}", event);
deg_sub_.notify(vec![event.into()].into()).await;
}
},
|res| async {
match res {
Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
Err(e) => panic!("{}", e),
}
},
Error::DetachedTaskStopped,
ex.clone(),
);
info!("Starting JSON-RPC server");
let darkirc = Arc::new(DarkIrc::new(
p2p.clone(),
sled_db.clone(),
event_graph.clone(),
dnet_sub,
deg_sub,
replay_datastore.clone(),
));
let darkirc_ = Arc::clone(&darkirc);
let rpc_task = StoppableTask::new();
rpc_task.clone().start(
listen_and_serve(args.rpc_listen, darkirc.clone(), None, ex.clone()),
|res| async move {
match res {
Ok(()) | Err(Error::RpcServerStopped) => darkirc_.stop_connections().await,
Err(e) => error!("Failed stopping JSON-RPC server: {}", e),
}
},
Error::RpcServerStopped,
ex.clone(),
);
info!("Starting IRC server");
let password = args.password.unwrap_or_default();
let config_path = get_config_path(args.config, CONFIG_FILE)?;
let irc_server = IrcServer::new(
darkirc.clone(),
args.irc_listen,
args.irc_tls_cert,
args.irc_tls_secret,
config_path,
password,
)
.await?;
let irc_task = StoppableTask::new();
let ex_ = ex.clone();
irc_task.clone().start(
irc_server.clone().listen(ex_),
|res| async move {
match res {
Ok(()) | Err(Error::DetachedTaskStopped) => { /* TODO: */ }
Err(e) => error!("Failed stopping IRC server: {}", e),
}
},
Error::DetachedTaskStopped,
ex.clone(),
);
info!("Starting P2P network");
p2p.clone().start().await?;
info!("Waiting for some P2P connections...");
sleep(5).await;
// We'll attempt to sync {sync_attempts} times
if !args.skip_dag_sync {
for i in 1..=args.sync_attempts {
info!("Syncing event DAG (attempt #{})", i);
match event_graph.dag_sync().await {
Ok(()) => break,
Err(e) => {
if i == args.sync_attempts {
error!("Failed syncing DAG. Exiting.");
p2p.stop().await;
return Err(Error::DagSyncFailed)
} else {
// TODO: Maybe at this point we should prune or something?
// TODO: Or maybe just tell the user to delete the DAG from FS.
error!("Failed syncing DAG ({}), retrying in {}s...", e, args.sync_timeout);
sleep(args.sync_timeout.into()).await;
}
}
}
}
} else {
*event_graph.synced.write().await = true;
}
// Signal handling for graceful termination.
let (signals_handler, signals_task) = SignalHandler::new(ex)?;
signals_handler.wait_termination(signals_task).await?;
info!("Caught termination signal, cleaning up and exiting...");
info!("Stopping P2P network");
p2p.stop().await;
info!("Stopping JSON-RPC server");
rpc_task.stop().await;
dnet_task.stop().await;
deg_task.stop().await;
info!("Stopping IRC server");
irc_task.stop().await;
info!("Flushing sled database...");
let flushed_bytes = sled_db.flush_async().await?;
info!("Flushed {} bytes", flushed_bytes);
info!("Shut down successfully");
Ok(())
}