/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2025 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},
settings::{RpcSettings, RpcSettingsOpt},
},
system::{sleep, StoppableTask, StoppableTaskPtr, Subscription},
util::path::{expand_path, get_config_path},
Error, Result,
};
use darkfi_sdk::crypto::pasta_prelude::PrimeField;
use rand::rngs::OsRng;
use settings::list_configured_contacts;
use sled_overlay::sled;
use smol::{fs, lock::Mutex, stream::StreamExt, Executor};
use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
use tracing::{debug, error, info};
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::RlnIdentity};
/// JSON-RPC methods
mod rpc;
/// Settings utilities
mod settings;
fn panic_hook(panic_info: &std::panic::PanicHookInfo) {
error!("panic occurred: {panic_info}");
error!("{}", std::backtrace::Backtrace::force_capture());
std::process::abort()
}
#[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
#[serde(default)]
#[structopt(
name = "darkirc",
about = cli_desc!(),
version = concat!(env!("CARGO_PKG_VERSION"), "-", env!("COMMITISH"))
)]
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: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/share/darkfi/darkirc_db")]
/// Datastore (DB) path
datastore: String,
#[structopt(short, long, default_value = "~/.local/share/darkfi/replayed_darkirc_db")]
/// Replay logs (DB) path
replay_datastore: String,
#[structopt(long)]
/// Flag to store Sled DB instructions
replay_mode: bool,
#[structopt(long)]
/// Generate a new NaCl keypair and exit
gen_chacha_keypair: bool,
#[structopt(long)]
/// Generate a new encrypted channel NaCl secret and exit
gen_channel_secret: bool,
#[structopt(long = "get-chacha-pubkey")]
/// Recover NaCl public key from a secret key
chacha_secret: Option,
#[structopt(long)]
/// Generate a new RLN identity
gen_rln_identity: bool,
#[structopt(long)]
/// Flag to skip syncing the DAG (no history)
skip_dag_sync: bool,
#[structopt(long)]
/// IRC Password (Encrypted with bcrypt-2b)
password: Option,
#[structopt(long)]
/// Encrypt a given password for the IRC server connection
encrypt_password: bool,
#[structopt(long)]
/// List configured contacts.
list_contacts: bool,
#[structopt(flatten)]
/// P2P network settings
net: SettingsOpt,
#[structopt(flatten)]
/// JSON-RPC settings
rpc: RpcSettingsOpt,
}
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<()> {
// Abort the application on panic right away
std::panic::set_hook(Box::new(panic_hook));
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 under your contact, you can reuse this keypair for multiple contacts\n"
);
println!("[contact.\"satoshi\"]");
println!("dm_chacha_public = \"YOUR_CONTACT_PUBLIC_KEY\"");
println!("my_dm_chacha_secret = \"{secret}\"");
println!("#my_dm_chacha_public = \"{public}\"");
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 args.gen_rln_identity {
let identity = RlnIdentity::new(&mut OsRng);
let nullifier = bs58::encode(identity.nullifier.to_repr()).into_string();
let trapdoor = bs58::encode(identity.trapdoor.to_repr()).into_string();
println!("Place this in your config file:\n");
println!("[rln]");
println!("nullifier = \"{nullifier}\"");
println!("trapdoor = \"{trapdoor}\"");
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.list_contacts {
let config_path = match get_config_path(args.config, CONFIG_FILE) {
Ok(path) => path,
Err(e) => {
error!("Unable to get config path: {e}");
return Err(e);
}
};
let contents = match fs::read_to_string(&config_path).await {
Ok(c) => c,
Err(e) => {
error!("Unable read path `{config_path:?}`: {e}");
return Err(e.into());
}
};
let contents = match toml::from_str(&contents) {
Ok(v) => v,
Err(e) => {
error!("Failed parsing TOML config: {e}");
return Err(Error::ParseFailed("Failed parsing TOML config"));
}
};
// Parse configured contacts
let contacts = match list_configured_contacts(&contents) {
Ok(c) => c,
Err(e) => {
error!("List contacts failed `{config_path:?}`: {e}");
return Err(e);
}
};
for (name, (public_key, my_secret_key)) in contacts {
let public_key = bs58::encode(public_key.to_bytes()).into_string();
let my_public_key = my_secret_key.public_key();
let my_secret_key = bs58::encode(my_secret_key.to_bytes()).into_string();
let my_public_key = bs58::encode(my_public_key.to_bytes()).into_string();
println!("{name}: {public_key} using key {my_secret_key}({my_public_key})")
}
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 = match expand_path(&args.datastore) {
Ok(v) => v,
Err(e) => {
error!("Bad datastore path `{}`: {e}", args.datastore);
return Err(e);
}
};
if let Err(e) = fs::create_dir_all(&datastore).await {
error!("Failed to create data store path `{datastore:?}`: {e}");
return Err(e.into());
}
let replay_datastore = match expand_path(&args.replay_datastore) {
Ok(v) => v,
Err(e) => {
error!("Bad replay datastore path `{}`: {e}", args.replay_datastore);
return Err(e);
}
};
let replay_mode = args.replay_mode;
info!("Instantiating event DAG");
let sled_db = match sled::open(datastore.clone()) {
Ok(v) => v,
Err(e) => {
error!("Failed to open datastore database `{datastore:?}`: {e}");
return Err(e.into());
}
};
let p2p_settings: darkfi::net::Settings =
(env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION"), args.net).try_into()?;
let p2p = match P2p::new(p2p_settings, ex.clone()).await {
Ok(p2p) => p2p,
Err(e) => {
error!("Unable to create P2P network: {e}");
return Err(e);
}
};
let event_graph = match EventGraph::new(
p2p.clone(),
sled_db.clone(),
replay_datastore.clone(),
replay_mode,
"darkirc_dag",
1,
ex.clone(),
)
.await
{
Ok(v) => v,
Err(e) => {
error!("Event graph failed to start: {e}");
return Err(e);
}
};
let prune_task = event_graph.prune_task.get().unwrap();
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 rpc_settings: RpcSettings = args.rpc.into();
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(rpc_settings, 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 = match get_config_path(args.config.clone(), CONFIG_FILE) {
Ok(v) => v,
Err(e) => {
error!("Cannot get config path `{:?}`: {e}", args.config);
return Err(e);
}
};
let irc_server = match IrcServer::new(
darkirc.clone(),
args.irc_listen,
args.irc_tls_cert,
args.irc_tls_secret,
config_path,
password,
)
.await
{
Ok(v) => v,
Err(e) => {
error!("Unable to create IRC server: {e}");
return Err(e);
}
};
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");
if let Err(e) = p2p.clone().start().await {
error!("P2P failed to start: {e}");
return Err(e);
}
// Initial DAG sync
if let Err(e) = sync_task(&p2p, &event_graph, args.skip_dag_sync).await {
error!("DAG sync task failed to start: {e}");
return Err(e);
};
// Stoppable task to monitor network and resync on disconnect.
let sync_mon_task = StoppableTask::new();
sync_mon_task.clone().start(
sync_and_monitor(p2p.clone(), event_graph.clone(), args.skip_dag_sync),
|res| async move {
match res {
Ok(()) | Err(Error::DetachedTaskStopped) => { /* TODO: */ }
Err(e) => error!("Failed sync task: {e}"),
}
},
Error::DetachedTaskStopped,
ex.clone(),
);
// 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;
prune_task.stop().await;
info!("Flushing sled database...");
let flushed_bytes = sled_db.flush_async().await?;
info!("Flushed {flushed_bytes} bytes");
info!("Shut down successfully");
Ok(())
}
/// Async task to monitor network disconnections.
async fn monitor_network(subscription: &Subscription) -> Result<()> {
Err(subscription.receive().await)
}
/// Async task to endlessly try to sync DAG, returns Ok if done.
async fn sync_task(p2p: &P2pPtr, event_graph: &EventGraphPtr, skip_dag_sync: bool) -> Result<()> {
let comms_timeout = p2p.settings().read_arc().await.outbound_connect_timeout_max();
loop {
if p2p.is_connected() {
info!("Got peer connection");
// We'll attempt to sync for ever
if !skip_dag_sync {
info!("Syncing event DAG");
match event_graph.dag_sync().await {
Ok(()) => break,
Err(e) => {
// 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 ({e}), retrying in {comms_timeout}s...");
sleep(comms_timeout).await;
}
}
} else {
*event_graph.synced.write().await = true;
break;
}
} else {
info!("Waiting for some P2P connections...");
sleep(comms_timeout).await;
}
}
Ok(())
}
/// Async task to monitor the network and force resync on disconnections
async fn sync_and_monitor(
p2p: P2pPtr,
event_graph: EventGraphPtr,
skip_dag_sync: bool,
) -> Result<()> {
loop {
let net_subscription = p2p.hosts().subscribe_disconnect().await;
let result = monitor_network(&net_subscription).await;
net_subscription.unsubscribe().await;
match result {
Ok(_) => return Ok(()),
Err(Error::NetworkNotConnected) => {
// Sync node again
info!("Network disconnection detected, resyncing...");
*event_graph.synced.write().await = false;
sync_task(&p2p, &event_graph, skip_dag_sync).await?;
}
Err(e) => return Err(e),
}
}
}