/* 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::{
collections::{HashMap, HashSet},
process::exit,
sync::Arc,
time::UNIX_EPOCH,
};
use async_trait::async_trait;
use semver::Version;
use smol::{
lock::{Mutex, MutexGuard},
stream::StreamExt,
Executor,
};
use structopt::StructOpt;
use structopt_toml::StructOptToml;
use tinyjson::JsonValue;
use toml::Value;
use tracing::{debug, error, info, warn};
use url::Url;
use darkfi::{
async_daemonize, cli_desc,
net::{
self,
acceptor::InboundListenerHealth,
hosts::HostColor,
settings::{BanPolicy, MagicBytes, NetworkProfile},
P2p, P2pPtr,
},
rpc::{
jsonrpc::*,
server::{listen_and_serve, RequestHandler},
settings::{RpcSettings, RpcSettingsOpt},
},
system::{sleep, StoppableTask, StoppableTaskPtr},
util::path::get_config_path,
Error, Result,
};
const CONFIG_FILE: &str = "lilith_config.toml";
const CONFIG_FILE_CONTENTS: &str = include_str!("../lilith_config.toml");
#[derive(Clone, Debug, serde::Deserialize, StructOpt, StructOptToml)]
#[serde(default)]
#[structopt(name = "lilith", about = cli_desc!())]
struct Args {
#[structopt(flatten)]
/// JSON-RPC settings
rpc: RpcSettingsOpt,
#[structopt(short, long)]
/// Configuration file to use
config: Option,
#[structopt(short, long)]
/// Set log file to ouput into
log: Option,
#[structopt(short, parse(from_occurrences))]
/// Increase verbosity (-vvv supported)
verbose: u8,
#[structopt(long, default_value = "120")]
/// Interval after which to check whitelist peers
whitelist_refinery_interval: u64,
}
/// Struct representing a spawned P2P network
struct Spawn {
/// String identifier,
pub name: String,
/// P2P pointer
pub p2p: P2pPtr,
}
impl Spawn {
async fn get_whitelist(&self) -> Vec {
self.p2p
.hosts()
.container
.fetch_all(HostColor::White)
.iter()
.map(|(addr, _url)| JsonValue::String(addr.to_string()))
.collect()
}
async fn get_greylist(&self) -> Vec {
self.p2p
.hosts()
.container
.fetch_all(HostColor::Grey)
.iter()
.map(|(addr, _url)| JsonValue::String(addr.to_string()))
.collect()
}
async fn get_goldlist(&self) -> Vec {
self.p2p
.hosts()
.container
.fetch_all(HostColor::Gold)
.iter()
.map(|(addr, _url)| JsonValue::String(addr.to_string()))
.collect()
}
async fn info(&self) -> JsonValue {
let mut addr_vec = vec![];
for addr in &self.p2p.settings().read().await.inbound_addrs {
addr_vec.push(JsonValue::String(addr.as_ref().to_string()));
}
let listeners = self
.p2p
.session_inbound()
.listener_health()
.await
.iter()
.map(listener_health_info)
.collect();
JsonValue::Object(HashMap::from([
("name".to_string(), JsonValue::String(self.name.clone())),
("urls".to_string(), JsonValue::Array(addr_vec)),
("listeners".to_string(), JsonValue::Array(listeners)),
("whitelist".to_string(), JsonValue::Array(self.get_whitelist().await)),
("greylist".to_string(), JsonValue::Array(self.get_greylist().await)),
("goldlist".to_string(), JsonValue::Array(self.get_goldlist().await)),
]))
}
}
fn listener_health_info(health: &InboundListenerHealth) -> JsonValue {
JsonValue::Object(HashMap::from([
("url".to_string(), JsonValue::String(health.url.to_string())),
("running".to_string(), JsonValue::Boolean(health.running)),
("active".to_string(), JsonValue::Number(health.active as f64)),
("negotiating".to_string(), JsonValue::Number(health.negotiating as f64)),
("limit".to_string(), JsonValue::Number(health.limit as f64)),
("saturated".to_string(), JsonValue::Boolean(health.saturated())),
("accept_backoff".to_string(), JsonValue::Boolean(health.accept_backoff)),
]))
}
/// Defines the network-specific settings
#[derive(Clone)]
struct NetInfo {
/// Accept addresses the network will use
pub accept_addrs: Vec,
/// Other seeds to connect to
pub seeds: Vec,
/// Manual peers to connect to
pub peers: Vec,
/// Supported network version
pub version: Version,
/// App Identifier for the app running on the network
pub app_name: String,
/// Enable localnet hosts
pub localnet: bool,
/// Path to P2P datastore
pub datastore: String,
/// Path to hostlist
pub hostlist: String,
/// Magic bytes used to distinguish the p2p network
pub magic_bytes: MagicBytes,
}
/// Struct representing the daemon
struct Lilith {
/// Spawned networks
pub networks: Vec,
/// JSON-RPC connection tracker
pub rpc_connections: Mutex>,
}
impl Lilith {
/// Since `Lilith` does not make outbound connections, if a peer is
/// upgraded to whitelist it will remain on the whitelist even if the
/// give peer is no longer online.
///
/// To protect `Lilith` from sharing potentially offline nodes,
/// `whitelist_refinery` periodically ping nodes on the whitelist. If they
/// are reachable, we update their last seen field. Otherwise, we downgrade
/// them to the greylist.
///
/// Note: if `Lilith` loses connectivity this method will delete peers from
/// the whitelist, meaning `Lilith` will need to rebuild its hostlist when
/// it comes back online.
async fn whitelist_refinery(
network_name: String,
p2p: P2pPtr,
refinery_interval: u64,
) -> Result<()> {
debug!(target: "net::refinery::whitelist_refinery", "Starting whitelist refinery for \"{network_name}\"");
let hosts = p2p.hosts();
loop {
sleep(refinery_interval).await;
match hosts.container.fetch_last(HostColor::White) {
Some(entry) => {
let url = &entry.0;
let last_seen = &entry.1;
if !hosts.refinable(url) {
debug!(target: "net::refinery::whitelist_refinery", "Addr={} not available!",
url.clone());
continue
}
if !p2p.session_refine().handshake_node(url.clone(), p2p.clone()).await {
debug!(target: "net::refinery:::whitelist_refinery",
"Host {url} is not responsive. Downgrading from whitelist");
if let Err(e) = hosts.greylist_host(url, *last_seen).await {
error!(target: "net::refinery::whitelist_refinery", "Could not send {url} to the greylist: {e}");
}
continue
}
debug!(target: "net::refinery::whitelist_refinery",
"Peer {url} is responsive. Updating last_seen");
// This node is active. Update the last seen field.
let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
if let Err(e) = hosts.whitelist_host(url, last_seen).await {
error!(target: "net::refinery::whitelist_refinery", "Could not send {url} to the whitelist: {e}");
}
}
None => {
debug!(target: "net::refinery::whitelist_refinery",
"Whitelist is empty! Cannot start refinery process");
continue
}
}
}
}
// RPCAPI:
// Returns all spawned networks names with their node addresses.
// --> {"jsonrpc": "2.0", "method": "spawns", "params": [], "id": 42}
// <-- {"jsonrpc": "2.0", "result": {"spawns": spawns_info}, "id": 42}
async fn spawns(&self, id: i64, _params: JsonValue) -> JsonResult {
let mut spawns = vec![];
for spawn in &self.networks {
spawns.push(spawn.info().await);
}
let json =
JsonValue::Object(HashMap::from([("spawns".to_string(), JsonValue::Array(spawns))]));
JsonResponse::new(json, id).into()
}
}
#[async_trait]
impl RequestHandler<()> for Lilith {
async fn handle_request(&self, req: JsonRequest) -> JsonResult {
return match req.method.as_str() {
"ping" => self.pong(req.id, req.params).await,
"spawns" => self.spawns(req.id, req.params).await,
_ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
}
}
async fn connections_mut(&self) -> MutexGuard<'life0, HashSet> {
self.rpc_connections.lock().await
}
}
/// Parse a TOML string for any configured network and return a map containing
/// said configurations.
fn parse_configured_networks(data: &str) -> Result> {
let mut ret = HashMap::new();
if let Value::Table(map) = toml::from_str(data)? {
if map.contains_key("network") && map["network"].is_table() {
for net in map["network"].as_table().unwrap() {
info!(target: "lilith", "Found configuration for network: {}", net.0);
let table = net.1.as_table().unwrap();
if !table.contains_key("accept_addrs") {
warn!(target: "lilith", "Network accept addrs are mandatory, skipping network.");
continue
}
if !table.contains_key("hostlist") {
error!(target: "lilith", "Hostlist path is mandatory! Configure and try again.");
exit(1)
}
let name = net.0.to_string();
let accept_addrs: Vec = table["accept_addrs"]
.as_array()
.unwrap()
.iter()
.map(|x| Url::parse(x.as_str().unwrap()).unwrap())
.collect();
let mut seeds = vec![];
if table.contains_key("seeds") {
if let Some(s) = table["seeds"].as_array() {
for seed in s {
if let Some(u) = seed.as_str() {
if let Ok(url) = Url::parse(u) {
seeds.push(url);
}
}
}
}
}
let mut peers = vec![];
if table.contains_key("peers") {
if let Some(p) = table["peers"].as_array() {
for peer in p {
if let Some(u) = peer.as_str() {
if let Ok(url) = Url::parse(u) {
peers.push(url);
}
}
}
}
}
let localnet = if table.contains_key("localnet") {
table["localnet"].as_bool().unwrap()
} else {
false
};
let version = if table.contains_key("version") {
semver::Version::parse(table["version"].as_str().unwrap())?
} else {
semver::Version::parse(option_env!("CARGO_PKG_VERSION").unwrap_or("0.0.0"))?
};
let app_name = if table.contains_key("app_name") {
table["app_name"].as_str().unwrap().to_string()
} else {
String::new()
};
let datastore: String = table["datastore"].as_str().unwrap().to_string();
let hostlist: String = table["hostlist"].as_str().unwrap().to_string();
let magic_bytes: [u8; 4] = table["magic_bytes"]
.as_array()
.unwrap()
.iter()
.map(|v| v.as_integer().unwrap() as u8)
.collect::>()
.try_into()
.expect("Wrong magic bytes value");
let net_info = NetInfo {
accept_addrs,
seeds,
peers,
version,
app_name,
localnet,
datastore,
hostlist,
magic_bytes: MagicBytes(magic_bytes),
};
ret.insert(name, net_info);
}
}
}
Ok(ret)
}
async fn spawn_net(name: String, info: &NetInfo, ex: Arc>) -> Result {
let mut listen_urls = vec![];
// Configure listen addrs for this network
for url in &info.accept_addrs {
listen_urls.push(url.clone());
}
let (active_profiles, profiles) = supported_network_profiles();
// P2P network settings
let settings = net::Settings {
magic_bytes: info.magic_bytes.clone(),
inbound_addrs: listen_urls.clone(),
seeds: info.seeds.clone(),
peers: info.peers.clone(),
outbound_connections: 0,
inbound_connections: 512,
app_version: info.version.clone(),
app_name: info.app_name.clone(),
localnet: info.localnet,
p2p_datastore: Some(info.datastore.clone()),
hostlist: Some(info.hostlist.clone()),
active_profiles,
ban_policy: BanPolicy::Relaxed,
profiles,
..Default::default()
};
// Create P2P instance
let p2p = P2p::new(settings, ex.clone()).await?;
let addrs_str: Vec<&str> = listen_urls.iter().map(|x| x.as_str()).collect();
info!(target: "lilith", "Starting seed network node for \"{name}\" on {addrs_str:?}");
p2p.clone().start().await?;
let spawn = Spawn { name, p2p };
Ok(spawn)
}
fn supported_network_profiles() -> (Vec, HashMap) {
let definitions = [
("tcp", NetworkProfile::default()),
("tcp+tls", NetworkProfile::default()),
("tor", NetworkProfile::tor_default()),
("tor+tls", NetworkProfile::tor_default()),
("i2p", NetworkProfile::tor_default()),
("i2p+tls", NetworkProfile::tor_default()),
];
let active_profiles = definitions.iter().map(|(name, _)| (*name).to_string()).collect();
let profiles =
definitions.into_iter().map(|(name, profile)| (name.to_string(), profile)).collect();
(active_profiles, profiles)
}
async_daemonize!(realmain);
async fn realmain(args: Args, ex: Arc>) -> Result<()> {
// Pick up network settings from the TOML config
let cfg_path = get_config_path(args.config, CONFIG_FILE)?;
let toml_contents = std::fs::read_to_string(cfg_path)?;
let configured_nets = parse_configured_networks(&toml_contents)?;
if configured_nets.is_empty() {
error!(target: "lilith", "No networks are enabled in config");
exit(1);
}
// Spawn configured networks
let mut networks = vec![];
for (name, info) in &configured_nets {
match spawn_net(name.to_string(), info, ex.clone()).await {
Ok(spawn) => networks.push(spawn),
Err(e) => {
error!(target: "lilith", "Failed to start P2P network seed for \"{name}\": {e}");
exit(1);
}
}
}
// Set up main daemon and background refinery_tasks
let lilith = Arc::new(Lilith { networks, rpc_connections: Mutex::new(HashSet::new()) });
let mut refinery_tasks = HashMap::new();
for network in &lilith.networks {
let name = network.name.clone();
let task = StoppableTask::new();
task.clone().start(
Lilith::whitelist_refinery(name.clone(), network.p2p.clone(), args.whitelist_refinery_interval),
|res| async move {
match res {
Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
Err(e) => error!(target: "lilith", "Failed starting refinery task for \"{name}\": {e}"),
}
},
Error::DetachedTaskStopped,
ex.clone(),
);
refinery_tasks.insert(network.name.clone(), task);
}
// JSON-RPC server
let rpc_settings: RpcSettings = args.rpc.into();
info!(target: "lilith", "Starting JSON-RPC server on {}", rpc_settings.listen);
let lilith_ = lilith.clone();
let rpc_task = StoppableTask::new();
rpc_task.clone().start(
listen_and_serve(rpc_settings, lilith.clone(), None, ex.clone()),
|res| async move {
match res {
Ok(()) | Err(Error::RpcServerStopped) => lilith_.stop_connections().await,
Err(e) => error!(target: "lilith", "Failed starting JSON-RPC server: {e}"),
}
},
Error::RpcServerStopped,
ex.clone(),
);
// Signal handling for graceful termination.
let (signals_handler, signals_task) = SignalHandler::new(ex)?;
signals_handler.wait_termination(signals_task).await?;
info!(target: "lilith", "Caught termination signal, cleaning up and exiting...");
info!(target: "lilith", "Stopping JSON-RPC server...");
rpc_task.stop().await;
// Cleanly stop p2p networks
for spawn in &lilith.networks {
info!(target: "lilith", "Stopping \"{}\" task", spawn.name);
refinery_tasks.get(&spawn.name).unwrap().stop().await;
info!(target: "lilith", "Stopping \"{}\" P2P", spawn.name);
spawn.p2p.stop().await;
}
info!(target: "lilith", "Bye!");
Ok(())
}
#[cfg(test)]
mod tests {
use darkfi::net::{acceptor::InboundListenerHealth, hosts::HostContainer};
use tinyjson::JsonValue;
use url::Url;
use super::{listener_health_info, supported_network_profiles};
#[test]
fn test_listener_health_info_reports_saturation() {
let health = InboundListenerHealth {
url: Url::parse("tor+tls://example.onion:9000").unwrap(),
running: true,
active: 2,
negotiating: 1,
limit: 3,
accept_backoff: false,
};
let info = listener_health_info(&health);
assert_eq!(info["url"], JsonValue::String(health.url.to_string()));
assert_eq!(info["running"], JsonValue::Boolean(true));
assert_eq!(info["active"], JsonValue::Number(2.0));
assert_eq!(info["negotiating"], JsonValue::Number(1.0));
assert_eq!(info["limit"], JsonValue::Number(3.0));
assert_eq!(info["saturated"], JsonValue::Boolean(true));
assert_eq!(info["accept_backoff"], JsonValue::Boolean(false));
}
#[test]
fn test_supported_network_profiles_are_consistent() {
let (active_profiles, profiles) = supported_network_profiles();
assert_eq!(active_profiles, ["tcp", "tcp+tls", "tor", "tor+tls", "i2p", "i2p+tls"]);
assert_eq!(profiles.len(), active_profiles.len());
assert!(active_profiles.iter().all(|profile| profiles.contains_key(profile)));
assert!(!active_profiles.iter().any(|profile| profile.starts_with("nym")));
let shareable = HostContainer::shareable_schemes(&active_profiles, &[], &None, &None);
assert_eq!(shareable, active_profiles);
}
}