/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2022 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, }; use async_std::sync::Arc; use async_trait::async_trait; use log::{error, info, warn}; use serde_json::{json, Value}; use structopt_toml::StructOptToml; use url::Url; use darkfi::{ async_daemonize, net, net::P2pPtr, rpc::{ jsonrpc::{ ErrorCode::{InvalidParams, MethodNotFound}, JsonError, JsonNotification, JsonRequest, JsonResponse, JsonResult, JsonSubscriber, }, server::{listen_and_serve, RequestHandler}, }, system::{Subscriber, SubscriberPtr}, util::{ async_util::sleep, file::{load_file, save_file}, path::{expand_path, get_config_path}, }, Result, }; mod config; use config::{parse_configured_networks, Args, NetInfo}; const CONFIG_FILE: &str = "lilith_config.toml"; const CONFIG_FILE_CONTENTS: &str = include_str!("../lilith_config.toml"); /// Struct representing a spawned p2p network. struct Spawn { name: String, p2p: P2pPtr, } impl Spawn { async fn addresses(&self) -> Vec { self.p2p.hosts().load_all().await.iter().map(|addr| addr.to_string()).collect() } pub async fn info(&self) -> serde_json::Value { // Building addr_vec string let mut addr_vec = vec![]; for addr in &self.p2p.settings().inbound { addr_vec.push(addr.as_ref().to_string()); } json!({ "name": self.name.clone(), "urls": addr_vec, "hosts": self.addresses().await, }) } } /// Struct representing the daemon. struct Lilith { /// Configured urls urls: Vec, /// Spawned networks spawns: Vec, // TODO: Subscriber should come from ValidatorState or something subscriber: SubscriberPtr, } impl Lilith { async fn spawns_hosts(&self) -> HashMap> { // Building urls string let mut spawns = HashMap::new(); for spawn in &self.spawns { spawns.insert(spawn.name.clone(), spawn.addresses().await); } spawns } // RPCAPI: // Returns all spawned networks names with their node addresses. // --> {"jsonrpc": "2.0", "method": "spawns", "params": [], "id": 42} // <-- {"jsonrpc": "2.0", "result": "{spawns}", "id": 42} async fn spawns(&self, id: Value, _params: &[Value]) -> JsonResult { // Building urls string let mut urls_vec = vec![]; for url in &self.urls { urls_vec.push(url.as_ref().to_string()); } // Gathering spawns info let mut spawns = vec![]; for spawn in &self.spawns { spawns.push(spawn.info().await); } // Generating json let json = json!({ "urls": urls_vec, "spawns": spawns, }); JsonResponse::new(json, id).into() } // RPCAPI: // Replies to a ping method. // --> {"jsonrpc": "2.0", "method": "ping", "params": [], "id": 42} // <-- {"jsonrpc": "2.0", "result": "pong", "id": 42} async fn pong(&self, id: Value, _params: &[Value]) -> JsonResult { JsonResponse::new(json!("pong"), id).into() } // RPCAPI: // Create a new subscriber for new blocks to notify connected peer. // // --> {"jsonrpc": "2.0", "method": "blockchain.notify_blocks", "params": [], "id": 1} // <-- {"jsonrpc": "2.0", "result": {...}, "id": 1} pub async fn blockchain_notify_blocks(&self, id: Value, params: &[Value]) -> JsonResult { if !params.is_empty() { return JsonError::new(InvalidParams, None, id).into() } JsonSubscriber::new(self.subscriber.clone()).into() } } // TODO: remove this async fn simulate_blocks(subscriber: SubscriberPtr) { // Notifications simulation let message = JsonNotification::new("blockchain.notify_blocks", Value::from("New Block created!")); loop { subscriber.notify(message.clone()).await; sleep(10).await; } } #[async_trait] impl RequestHandler for Lilith { async fn handle_request(&self, req: JsonRequest) -> JsonResult { if !req.params.is_array() { return JsonError::new(InvalidParams, None, req.id).into() } let params = req.params.as_array().unwrap(); match req.method.as_str() { Some("spawns") => return self.spawns(req.id, params).await, Some("ping") => return self.pong(req.id, params).await, Some("blockchain.notify_blocks") => { return self.blockchain_notify_blocks(req.id, params).await } Some(_) | None => return JsonError::new(MethodNotFound, None, req.id).into(), } } } async fn spawn_network( name: &str, info: NetInfo, urls: Vec, saved_hosts: Option<&HashSet>, ex: Arc>, ) -> Result { let mut full_urls = Vec::new(); for url in &urls { let mut url = url.clone(); url.set_port(Some(info.port))?; full_urls.push(url); } let network_settings = net::Settings { inbound: full_urls.clone(), seeds: info.seeds, peers: info.peers, outbound_connections: 0, localnet: info.localnet, channel_log: info.channel_log, app_version: None, ..Default::default() }; let p2p = net::P2p::new(network_settings).await; // Setting saved hosts match saved_hosts { Some(hosts) => { // Converting hashet to vec let mut vec = vec![]; for url in hosts { vec.push(url.clone()); } p2p.hosts().store(vec).await; } None => info!("No saved hosts found for {}", name), } // Building ext_addr_vec string let mut urls_vec = vec![]; for url in &full_urls { urls_vec.push(url.as_ref().to_string()); } info!("Starting seed network node for {} at: {:?}", name, urls_vec); p2p.clone().start(ex.clone()).await?; let _ex = ex.clone(); let _p2p = p2p.clone(); ex.spawn(async move { if let Err(e) = _p2p.run(_ex).await { error!("Failed starting P2P network seed: {}", e); } }) .detach(); let spawn = Spawn { name: name.to_string(), p2p }; Ok(spawn) } /// Retrieve saved hosts for provided networks fn load_hosts(path: &Path, networks: &[&str]) -> HashMap> { let mut saved_hosts = HashMap::new(); info!("Retrieving saved hosts from: {:?}", path); let contents = load_file(path); if let Err(e) = contents { warn!("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!("Skipping malformed url: {} ({})", data[1], e); continue } }; hosts.insert(url); saved_hosts.insert(data[0].to_string(), hosts); } } saved_hosts } /// Save spawns current hosts fn save_hosts(path: &Path, spawns: HashMap>) { let mut string = "".to_string(); for (name, urls) in spawns { for url in urls { string.push_str(&name); string.push('\t'); string.push_str(&url); string.push('\n'); } } if !string.eq("") { info!("Saving current hosts of spawnned networks to: {:?}", path); if let Err(e) = save_file(path, &string) { error!("Failed saving hosts: {}", e); } } } async_daemonize!(realmain); async fn realmain(args: Args, ex: Arc>) -> Result<()> { // We use this handler to block this function after detaching all // tasks, and to catch a shutdown signal, where we can clean up and // exit gracefully. let (signal, shutdown) = smol::channel::bounded::<()>(1); ctrlc::set_handler(move || { async_std::task::block_on(signal.send(())).unwrap(); }) .unwrap(); // Pick up network settings from the TOML configuration 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)?; // Verify any daemon network is enabled if configured_nets.is_empty() { info!("No daemon network is enabled!"); return Ok(()) } // Setting urls let mut urls = args.urls.clone(); if urls.is_empty() { info!("Urls are not provided, will use: tcp://127.0.0.1"); let url = Url::parse("tcp://127.0.0.1")?; urls.push(url); } // Retrieve saved hosts for configured networks let full_path = expand_path(&args.hosts_file)?; let nets: Vec<&str> = configured_nets.keys().map(|x| x.as_str()).collect(); let saved_hosts = load_hosts(&full_path, &nets); // Spawn configured networks let mut spawns = vec![]; for (name, info) in &configured_nets { match spawn_network(name, info.clone(), urls.clone(), saved_hosts.get(name), ex.clone()) .await { Ok(spawn) => spawns.push(spawn), Err(e) => error!("Failed starting {} P2P network seed: {}", name, e), } } // TODO: Subscriber should come from ValidatorState or something let subscriber: SubscriberPtr = Subscriber::new(); let lilith = Lilith { urls, spawns, subscriber: subscriber.clone() }; let lilith = Arc::new(lilith); // JSON-RPC server info!("Starting JSON-RPC server"); let _ex = ex.clone(); ex.spawn(listen_and_serve(args.rpc_listen, lilith.clone(), _ex)).detach(); // JSON-RPC notifications simulation ex.spawn(simulate_blocks(subscriber)).detach(); // Wait for SIGINT shutdown.recv().await?; print!("\r"); info!("Caught termination signal, cleaning up and exiting..."); // Save spawns current hosts save_hosts(&full_path, lilith.spawns_hosts().await); Ok(()) }