/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2023 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}, path::Path, process::exit, sync::Arc, }; use async_trait::async_trait; use futures::future::join_all; use log::{debug, error, info, warn}; use semver::Version; use smol::{stream::StreamExt, Executor}; use structopt::StructOpt; use structopt_toml::StructOptToml; use tinyjson::JsonValue; use toml::Value; use url::Url; use darkfi::{ async_daemonize, cli_desc, net::{self, connector::Connector, protocol::ProtocolVersion, session::Session, P2p, P2pPtr}, rpc::{ jsonrpc::*, server::{listen_and_serve, RequestHandler}, }, system::{sleep, StoppableTask}, util::{ file::{load_file, save_file}, path::{expand_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(long, default_value = "tcp://127.0.0.1:18927")] /// JSON-RPC listen URL pub rpc_listen: Url, #[structopt(long)] /// Accept addresses (URL without port) pub accept_addrs: Vec, #[structopt(short, long)] /// Configuration file to use pub config: Option, #[structopt(long, default_value = "~/.config/darkfi/lilith_hosts.tsv")] /// Hosts .tsv file to use pub hosts_file: String, #[structopt(short, long)] /// Set log file to ouput into log: Option, #[structopt(short, parse(from_occurrences))] /// Increase verbosity (-vvv supported) pub verbose: u8, } /// Struct representing a spawned P2P network struct Spawn { /// String identifier, pub name: String, /// P2P pointer pub p2p: P2pPtr, } impl Spawn { async fn addresses(&self) -> Vec { self.p2p .hosts() .fetch_all() .await .iter() .map(|addr| JsonValue::String(addr.to_string())) .collect() } async fn info(&self) -> JsonValue { let mut addr_vec = vec![]; for addr in &self.p2p.settings().inbound_addrs { addr_vec.push(JsonValue::String(addr.as_ref().to_string())); } JsonValue::Object(HashMap::from([ ("name".to_string(), JsonValue::String(self.name.clone())), ("urls".to_string(), JsonValue::Array(addr_vec)), ("hosts".to_string(), JsonValue::Array(self.addresses().await)), ])) } } /// Defines the network-specific settings #[derive(Clone)] struct NetInfo { /// Specific port the network will use pub port: u16, /// Other seeds to connect to pub seeds: Vec, /// Manual peers to connect to pub peers: Vec, /// Supported network version pub version: Version, /// Enable localnet hosts pub localnet: bool, } /// Struct representing the daemon struct Lilith { /// Spawned networks pub networks: Vec, } impl Lilith { /// Internal task to run a periodic purge of unreachable hosts /// for a specific P2P network. async fn periodic_purge(name: String, p2p: P2pPtr, ex: Arc>) -> Result<()> { info!(target: "lilith", "Starting periodic host purge task for \"{}\"", name); loop { // We'll pick up to 10 hosts every minute and try to connect to // them. If we can't reach them, we'll remove them from our set. sleep(60).await; debug!(target: "lilith", "[{}] Picking random hosts from db", name); let lottery_winners = p2p.clone().hosts().fetch_n_random(10).await; let win_str: Vec<&str> = lottery_winners.iter().map(|x| x.as_str()).collect(); debug!(target: "lilith", "[{}] Got: {:?}", name, win_str); let mut tasks = vec![]; for host in &lottery_winners { let p2p_ = p2p.clone(); let ex_ = ex.clone(); tasks.push(async move { let session_out = p2p_.session_outbound().await; let session_weak = Arc::downgrade(&p2p_.session_outbound().await); let connector = Connector::new(p2p_.settings(), session_weak); debug!(target: "lilith", "Connecting to {}", host); match connector.connect(host).await { Ok((_url, channel)) => { debug!(target: "lilith", "Connected successfully!"); let proto_ver = ProtocolVersion::new( channel.clone(), p2p_.settings().clone(), p2p_.hosts().clone(), ) .await; let handshake_task = session_out.perform_handshake_protocols( proto_ver, channel.clone(), ex_.clone(), ); channel.clone().start(ex_.clone()); match handshake_task.await { Ok(()) => { debug!(target: "lilith", "Handshake success! Stopping channel."); channel.stop().await; } Err(e) => { debug!(target: "lilith", "Handshake failure! {}", e); } } } Err(e) => { debug!(target: "lilith", "Failed to connect to {}, removing from set ({})", host, e); // Remove from hosts set p2p_.hosts().remove(host).await; } } }); } join_all(tasks).await; } } // 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: u16, _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 { match req.method.as_str() { "ping" => return self.pong(req.id, req.params).await, "spawns" => return self.spawns(req.id, req.params).await, _ => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(), } } } /// Attempt to read existing hosts tsv fn load_hosts(path: &Path, networks: &[&str]) -> HashMap> { let mut saved_hosts = HashMap::new(); let contents = load_file(path); if let Err(e) = contents { warn!(target: "lilith", "Failed retrieving saved hosts: {}", e); return saved_hosts } for line in contents.unwrap().lines() { let data: Vec<&str> = line.split('\t').collect(); if networks.contains(&data[0]) { let mut hosts = match saved_hosts.get(data[0]) { Some(hosts) => hosts.clone(), None => HashSet::new(), }; let url = match Url::parse(data[1]) { Ok(u) => u, Err(e) => { warn!(target: "lilith", "Skipping malformed url: {} ({})", data[1], e); continue } }; hosts.insert(url); saved_hosts.insert(data[0].to_string(), hosts); } } saved_hosts } async fn save_hosts(path: &Path, networks: &[Spawn]) { let mut tsv = String::new(); for spawn in networks { for host in spawn.p2p.hosts().fetch_all().await { tsv.push_str(&format!("{}\t{}\n", spawn.name, host.as_str())); } } if !tsv.eq("") { info!(target: "lilith", "Saving current hosts of spawned networks to: {:?}", path); if let Err(e) = save_file(path, &tsv) { error!(target: "lilith", "Failed saving hosts: {}", e); } } } /// 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("port") { warn!(target: "lilith", "Network port is mandatory, skipping network."); continue } let name = net.0.to_string(); let port = table["port"].as_integer().unwrap().try_into().unwrap(); 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 net_info = NetInfo { port, seeds, peers, version, localnet }; ret.insert(name, net_info); } } } Ok(ret) } async fn spawn_net( name: String, info: &NetInfo, accept_addrs: &[Url], saved_hosts: &HashSet, ex: Arc>, ) -> Result { let mut listen_urls = vec![]; // Configure listen addrs for this network for url in accept_addrs { let mut url = url.clone(); url.set_port(Some(info.port))?; listen_urls.push(url); } // P2P network settings let settings = net::Settings { inbound_addrs: listen_urls.clone(), seeds: info.seeds.clone(), peers: info.peers.clone(), outbound_connections: 0, outbound_connect_timeout: 30, app_version: info.version.clone(), localnet: info.localnet, allowed_transports: vec![ "tcp".to_string(), "tcp+tls".to_string(), "tor".to_string(), "tor+tls".to_string(), "nym".to_string(), "nym+tls".to_string(), ], ..Default::default() }; // Create P2P instance let p2p = P2p::new(settings, ex.clone()).await; // Fill db with cached hosts let hosts: Vec = saved_hosts.iter().cloned().collect(); p2p.hosts().store(&hosts).await; let addrs_str: Vec<&str> = listen_urls.iter().map(|x| x.as_str()).collect(); info!(target: "lilith", "Starting seed network node for \"{}\" on {:?}", name, addrs_str); p2p.clone().start().await?; let spawn = Spawn { name, p2p }; Ok(spawn) } 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); } // Retrieve any saved hosts for configured networks let net_names: Vec<&str> = configured_nets.keys().map(|x| x.as_str()).collect(); let saved_hosts = load_hosts(&expand_path(&args.hosts_file)?, &net_names); // Spawn configured networks let mut networks = vec![]; for (name, info) in &configured_nets { // TODO: Here we could actually differentiate between network versions // e.g. p2p_v3, p2p_v4, etc. Therefore we can spawn multiple networks // and they would all be version-checked, so we avoid mismatches when // seeding peers. match spawn_net( name.to_string(), info, &args.accept_addrs, saved_hosts.get(name).unwrap_or(&HashSet::new()), 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 tasks let lilith = Arc::new(Lilith { networks }); let mut periodic_tasks = HashMap::new(); for network in &lilith.networks { let name = network.name.clone(); let task = StoppableTask::new(); task.clone().start( Lilith::periodic_purge(name.clone(), network.p2p.clone(), ex.clone()), |res| async move { match res { Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ } Err(e) => error!(target: "lilith", "Failed starting periodic task for \"{}\": {}", name, e), } }, Error::DetachedTaskStopped, ex.clone(), ); periodic_tasks.insert(network.name.clone(), task); } // JSON-RPC server info!(target: "lilith", "Starting JSON-RPC server on {}", args.rpc_listen); let rpc_task = StoppableTask::new(); rpc_task.clone().start( listen_and_serve(args.rpc_listen, lilith.clone(), ex.clone()), |res| async { match res { Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ } 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..."); // Save in-memory hosts to tsv file save_hosts(&expand_path(&args.hosts_file)?, &lilith.networks).await; info!(target: "lilith", "Stopping JSON-RPC server..."); rpc_task.stop().await; // Cleanly stop p2p networks for spawn in &lilith.networks { info!(target: "lilith", "Stopping \"{}\" periodic task", spawn.name); periodic_tasks.get(&spawn.name).unwrap().stop().await; info!(target: "lilith", "Stopping \"{}\" P2P", spawn.name); spawn.p2p.stop().await; } info!(target: "lilith", "Bye!"); Ok(()) }