/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2026 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::{ io::Write, sync::{atomic::Ordering, Arc}, }; use darkfi::{ async_daemonize, cli_desc, event_graph::{proto::ProtocolEventGraph, EventGraph, EventGraphConfig, EventGraphPtr}, net::{session::SESSION_DEFAULT, settings::SettingsOpt, P2p, P2pPtr}, rpc::{ jsonrpc::JsonSubscriber, server::{listen_and_serve, RequestHandler}, settings::{RpcSettings, RpcSettingsOpt}, util::JsonValue, }, system::{sleep, StoppableTask, Subscription}, util::path::{expand_path, get_config_path}, Error, Result, }; use darkfi_sdk::crypto::pasta_prelude::PrimeField; use irc2::{ crypto::{bcrypt::bcrypt_hash_password, rln::RlnIdentity}, genesis_commits, irc::server::IrcServer, rpc, settings::list_configured_contacts, DarkIrc, }; use rand::rngs::OsRng; use sled_overlay::sled; use smol::{fs, 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"); // ===================================================================== // DarkIRC consensus parameters. // // These define the EventGraph configuration that EVERY DarkIRC node // in the network must agree on. Changing any of them is a hard fork. // They are passed verbatim to `EventGraph::new` at startup. // ===================================================================== /// Epoch origin for DAG rotation (UTC midnight, 1 March 2025). /// Rotation boundaries are computed as offsets from this point. const DARKIRC_INITIAL_GENESIS: u64 = 1_740_787_200_000; /// DAG rotation period, in hours. const DARKIRC_HOURS_ROTATION: u64 = 1; /// Genesis payload. Two protocols MUST use distinct values; this /// also feeds into `RlnAppId::from_genesis` so RLN signals from one /// deployment never appear valid on another. const DARKIRC_GENESIS_CONTENTS: &[u8] = b"darkirc-v1"; /// How many rotation periods to keep in the rolling DAG window. /// With `hours_rotation = 1` and `max_dags = 24`, this gives a /// 24-hour history window. Older events are evicted from sled. const DARKIRC_MAX_DAGS: usize = 24; 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, /// How many DAGs to sync. #[structopt(long, default_value = "8")] dags_count: usize, #[structopt(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)] // Whether to sync headers only or full sync fast_mode: 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, } #[global_allocator] static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; #[allow(non_upper_case_globals)] #[export_name = "malloc_conf"] pub static malloc_conf: &[u8] = b"dirty_decay_ms:1000,muzzy_decay_ms:1000\0"; async_daemonize!(realmain); async fn realmain(args: Args, ex: Arc>) -> Result<()> { if args.fast_mode { info!("fast mode enabled"); } // 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(); // Default per-epoch budget for fresh identities. let user_msg_limit: u64 = 10; println!("Generated a fresh RLN identity.\n"); println!("To register on the network, paste this into your IRC client:\n"); println!( " /msg NickServ REGISTER {nullifier} {trapdoor} {user_msg_limit}\n" ); println!( "Replace with any local label you like (\"alice\", \"throwaway\", etc)." ); println!( "Keep the nullifier and trapdoor secret - they ARE the identity. \ A `darkirc --gen-rln-identity` run is NOT idempotent; treat the \ output like a freshly-minted password." ); 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); } }; // Consensus config. Every node must use exactly these values. let eg_config = EventGraphConfig { initial_genesis: DARKIRC_INITIAL_GENESIS, hours_rotation: DARKIRC_HOURS_ROTATION, genesis_contents: DARKIRC_GENESIS_CONTENTS.to_vec(), pregenerated_identity_commitments: genesis_commits::pregenerated_identity_commitments(), max_dags: Some(DARKIRC_MAX_DAGS), }; let event_graph = match EventGraph::new( p2p.clone(), sled_db.clone(), replay_datastore.clone(), replay_mode, eg_config, ex.clone(), ) .await { Ok(v) => v, Err(e) => { error!("Event graph failed to start: {e}"); return Err(e); } }; // The prune task is only spawned when `hours_rotation > 0`. We // require rotation here, so the unwrap is safe. 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:?}"); let json = deg_event_to_json(&event); deg_sub_.notify(vec![json].into()).await; } }, |res| async { match res { Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ } Err(e) => panic!("{e}"), } }, Error::DetachedTaskStopped, ex.clone(), ); info!("Starting Gource subs task"); let gource_sub = JsonSubscriber::new("gource.subscribe_events"); let gource_sub_ = gource_sub.clone(); let event_graph_gource = event_graph.clone(); let gource_task = StoppableTask::new(); gource_task.clone().start( async move { let event_pub = event_graph_gource.event_pub.clone().subscribe().await; loop { let ev = event_pub.receive().await; if let Some(json) = rpc::privmsg_event_to_gource(&ev).await { gource_sub_.notify(vec![json].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, gource_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, args.fast_mode, args.dags_count).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, args.fast_mode, args.dags_count, ), |res| async move { match res { Ok(()) | Err(Error::DetachedTaskStopped) => { /* TODO: */ } Err(e) => error!("Failed sync task: {e}"), } }, Error::DetachedTaskStopped, ex.clone(), ); // Drain pending static broadcasts whenever the EG transitions // from unsynced to synced. // // NickServ REGISTER while the local DAG is unsynced will queue // the (event, blob) pair on `IrcServer::pending_static_broadcasts` // instead of broadcasting (a pre-sync broadcast goes nowhere - // peers gate `handle_static_put` AND `handle_tip_req` on their // own is_synced state). This task watches for the rising edge // of `is_synced()` and re-issues the queued broadcasts. let drain_task = StoppableTask::new(); let irc_server_for_drain = irc_server.clone(); let event_graph_for_drain = event_graph.clone(); drain_task.clone().start( async move { let mut last_state = event_graph_for_drain.is_synced(); loop { sleep(1).await; let now_state = event_graph_for_drain.is_synced(); // Rising edge: unsynced -> synced. if now_state && !last_state { match irc_server_for_drain.drain_pending_static_broadcasts().await { Ok(0) => { /* nothing pending; common case */ } Ok(n) => { info!("Drained {n} pending static broadcasts after sync"); } Err(e) => { error!("Failed to drain pending broadcasts: {e}"); } } } last_state = now_state; } }, |res| async move { match res { Ok(()) | Err(Error::DetachedTaskStopped) => { /* normal shutdown */ } Err(e) => error!("Drain task failed: {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; gource_task.stop().await; info!("Stopping IRC server"); irc_task.stop().await; drain_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, fast_mode: bool, dags_count: usize, ) -> Result<()> { // skip_dag_sync means "this node opts out of syncing entirely". if skip_dag_sync { event_graph.synced.store(true, Ordering::Release); info!("DAG sync skipped; marking synced immediately"); return Ok(()) } let comms_timeout = p2p.settings().read_arc().await.outbound_connect_timeout_max(); loop { if p2p.is_connected() { info!("Got peer connection"); info!("Syncing static DAG"); match event_graph.static_sync().await { Ok(()) => { info!("Static synced successfully") } Err(e) => { error!("Failed syncing static graph: {e}"); p2p.stop().await; return Err(Error::StaticDagSyncFailed) } } info!("Syncing event DAG"); // Sync mode is now per-call: full sync replays // every event (heavy, used by archival nodes), fast // sync only fetches headers (light, used by clients // that don't need to re-verify history). let sync_result = if fast_mode { event_graph.sync_selected_headers(dags_count).await } else { event_graph.sync_selected(dags_count).await }; match sync_result { Ok(()) => { info!( "Event DAG synced successfully ({} mode, {} dag(s))", if fast_mode { "fast" } else { "full" }, dags_count, ); 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 { 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, fast_mode: bool, dags_count: usize, ) -> Result<()> { // If sync is skipped entirely there's nothing to monitor. if skip_dag_sync { return Ok(()) } 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.store(false, Ordering::Release); sync_task(&p2p, &event_graph, skip_dag_sync, fast_mode, dags_count).await?; } Err(e) => return Err(e), } } } fn deg_event_to_json(ev: &darkfi::event_graph::deg::DegEvent) -> JsonValue { use darkfi::{ event_graph::deg::{DegEvent, MessageInfo}, rpc::util::json_map, }; fn info_to_json(direction: &str, info: &MessageInfo) -> JsonValue { let info_arr: Vec = info.info.iter().cloned().map(JsonValue::String).collect(); json_map([ ("direction", JsonValue::String(direction.into())), ("cmd", JsonValue::String(info.cmd.clone())), // NanoTimestamp's Display is the human-readable form; // emit it as a string to avoid losing precision through // the JSON number type (f64 can't hold nanos cleanly). ("time", JsonValue::String(format!("{}", info.time))), ("info", JsonValue::Array(info_arr)), ]) } match ev { DegEvent::SendMessage(info) => info_to_json("send", info), DegEvent::RecvMessage(info) => info_to_json("recv", info), } }