use std::{env, str, time::Duration}; use async_std::io::timeout; use futures::{select, AsyncReadExt, AsyncWriteExt, FutureExt}; use log::error; use rand::Rng; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; use url::Url; use crate::{ net::{TcpTransport, TorTransport, Transport, TransportName, TransportStream, UnixTransport}, Error, Result, }; #[derive(Debug, Clone)] pub enum ErrorCode { ParseError, InvalidRequest, MethodNotFound, InvalidParams, InternalError, KeyGenError, GetAddressesError, ImportAndExportFile, SetDefaultAddress, InvalidAmountParam, InvalidNetworkParam, InvalidTokenIdParam, InvalidAddressParam, InvalidSymbolParam, InvalidId, ServerError(i64), } impl ErrorCode { pub fn code(&self) -> i64 { match *self { ErrorCode::ParseError => -32700, ErrorCode::InvalidRequest => -32600, ErrorCode::MethodNotFound => -32601, ErrorCode::InvalidParams => -32602, ErrorCode::InternalError => -32603, ErrorCode::KeyGenError => -32002, ErrorCode::GetAddressesError => -32003, ErrorCode::ImportAndExportFile => -32004, ErrorCode::SetDefaultAddress => -32005, ErrorCode::InvalidAmountParam => -32010, ErrorCode::InvalidNetworkParam => -32011, ErrorCode::InvalidTokenIdParam => -32012, ErrorCode::InvalidAddressParam => -32013, ErrorCode::InvalidSymbolParam => -32014, ErrorCode::InvalidId => -32030, ErrorCode::ServerError(c) => c, } } pub fn description(&self) -> String { let desc = match *self { ErrorCode::ParseError => "Parse error", ErrorCode::InvalidRequest => "Invalid request", ErrorCode::MethodNotFound => "Method not found", ErrorCode::InvalidParams => "Invalid params", ErrorCode::InternalError => "Internal error", ErrorCode::KeyGenError => "Key gen error", ErrorCode::GetAddressesError => "get addresses error", ErrorCode::ImportAndExportFile => "error import/export a file", ErrorCode::SetDefaultAddress => "error set default address", ErrorCode::InvalidAmountParam => "Invalid amount param", ErrorCode::InvalidNetworkParam => "Invalid network param", ErrorCode::InvalidTokenIdParam => "Invalid token id param", ErrorCode::InvalidAddressParam => "Invalid address param", ErrorCode::InvalidSymbolParam => "Invalid symbol param", ErrorCode::InvalidId => "Invalid Id", ErrorCode::ServerError(_) => "Server error", }; desc.to_string() } } #[derive(Serialize, Deserialize, Clone, Debug)] #[serde(untagged)] pub enum JsonResult { Resp(JsonResponse), Err(JsonError), Notif(JsonNotification), } impl From for JsonResult { fn from(resp: JsonResponse) -> Self { Self::Resp(resp) } } impl From for JsonResult { fn from(err: JsonError) -> Self { Self::Err(err) } } impl From for JsonResult { fn from(notif: JsonNotification) -> Self { Self::Notif(notif) } } #[derive(Serialize, Deserialize, Clone, Debug)] pub struct JsonRequest { pub jsonrpc: Value, pub method: Value, pub params: Value, pub id: Value, } #[derive(Serialize, Deserialize, Clone, Debug)] pub struct JsonErrorVal { pub code: Value, pub message: Value, } #[derive(Serialize, Deserialize, Clone, Debug)] pub struct JsonError { pub jsonrpc: Value, pub error: JsonErrorVal, pub id: Value, } #[derive(Serialize, Deserialize, Clone, Debug)] pub struct JsonResponse { pub jsonrpc: Value, pub result: Value, pub id: Value, } #[derive(Serialize, Deserialize, Clone, Debug)] pub struct JsonNotification { pub jsonrpc: Value, pub method: Value, pub params: Value, } pub fn request(m: Value, p: Value) -> JsonRequest { let mut rng = rand::thread_rng(); JsonRequest { jsonrpc: json!("2.0"), method: m, params: p, id: json!(rng.gen::()) } } pub fn response(r: Value, i: Value) -> JsonResponse { JsonResponse { jsonrpc: json!("2.0"), result: r, id: i } } pub fn error(c: ErrorCode, m: Option, i: Value) -> JsonError { let ev = JsonErrorVal { code: json!(c.code()), message: if m.is_none() { json!(c.description()) } else { json!(Some(m)) }, }; JsonError { jsonrpc: json!("2.0"), error: ev, id: i } } pub fn notification(m: Value, p: Value) -> JsonNotification { JsonNotification { jsonrpc: json!("2.0"), method: m, params: p } } async fn reqrep_loop( mut stream: T, result_sender: async_channel::Sender, data_receiver: async_channel::Receiver, stop_receiver: async_channel::Receiver<()>, ) -> Result<()> { // If we don't get a reply after 30 seconds, we'll fail. let read_timeout = Duration::from_secs(30); loop { let mut buf = [0; 8192]; select! { data = data_receiver.recv().fuse() => { let data_str = serde_json::to_string(&data?)?; stream.write_all(data_str.as_bytes()).await?; let bytes_read = timeout(read_timeout, async { stream.read(&mut buf[..]).await }).await?; let reply: JsonResult = serde_json::from_slice(&buf[0..bytes_read])?; result_sender.send(reply).await?; } _ = stop_receiver.recv().fuse() => break } } Ok(()) } pub async fn open_channels( uri: &Url, ) -> Result<( async_channel::Sender, async_channel::Receiver, async_channel::Sender<()>, )> { let (data_sender, data_receiver) = async_channel::unbounded(); let (result_sender, result_receiver) = async_channel::unbounded(); let (stop_sender, stop_receiver) = async_channel::unbounded(); let transport_name = TransportName::try_from(uri.clone())?; macro_rules! reqrep { ($stream:expr, $transport:expr, $upgrade:expr) => {{ if let Err(err) = $stream { error!("RPC Setup for {} failed: {}", uri, err); return Err(Error::ConnectFailed) } let stream = $stream?.await; if let Err(err) = stream { error!("RPC Connection to {} failed: {}", uri, err); return Err(Error::ConnectFailed) } let stream = stream?; match $upgrade { None => { smol::spawn(reqrep_loop(stream, result_sender, data_receiver, stop_receiver)) .detach(); } Some(u) if u == "tls" => { let stream = $transport.upgrade_dialer(stream)?.await?; smol::spawn(reqrep_loop(stream, result_sender, data_receiver, stop_receiver)) .detach(); } Some(u) => return Err(Error::UnsupportedTransportUpgrade(u)), } }}; } match transport_name { TransportName::Tcp(upgrade) => { let transport = TcpTransport::new(None, 1024); let stream = transport.dial(uri.clone()); reqrep!(stream, transport, upgrade); } TransportName::Tor(upgrade) => { let socks5_url = Url::parse( &env::var("DARKFI_TOR_SOCKS5_URL") .unwrap_or_else(|_| "socks5://127.0.0.1:9050".to_string()), )?; let transport = TorTransport::new(socks5_url, None)?; let stream = transport.clone().dial(uri.clone()); reqrep!(stream, transport, upgrade); } TransportName::Unix => { let transport = UnixTransport::new(); let stream = transport.dial(uri.clone()).await; if let Err(err) = stream { error!("RPC Connection to {} failed: {}", uri, err); return Err(Error::ConnectFailed) } smol::spawn(reqrep_loop(stream?, result_sender, data_receiver, stop_receiver)).detach(); } _ => unimplemented!(), } Ok((data_sender, result_receiver, stop_sender)) } pub async fn send_request(uri: &Url, data: Value) -> Result { let data_str = serde_json::to_string(&data)?; let transport_name = TransportName::try_from(uri.clone())?; macro_rules! reply { ($stream:expr, $transport:expr, $upgrade:expr) => {{ if let Err(err) = $stream { error!("RPC Setup for {} failed: {}", uri, err); return Err(Error::ConnectFailed) } let stream = $stream?.await; if let Err(err) = stream { error!("RPC Connection to {} failed: {}", uri, err); return Err(Error::ConnectFailed) } let stream = stream?; match $upgrade { None => get_reply(stream, data_str).await, Some(u) if u == "tls" => { let stream = $transport.upgrade_dialer(stream)?.await?; get_reply(stream, data_str).await } Some(u) => Err(Error::UnsupportedTransportUpgrade(u)), } }}; } match transport_name { TransportName::Tcp(upgrade) => { let transport = TcpTransport::new(None, 1024); let stream = transport.dial(uri.clone()); reply!(stream, transport, upgrade) } TransportName::Tor(upgrade) => { let socks5_url = Url::parse( &env::var("DARKFI_TOR_SOCKS5_URL") .unwrap_or_else(|_| "socks5://127.0.0.1:9050".to_string()), )?; let transport = TorTransport::new(socks5_url, None)?; let stream = transport.clone().dial(uri.clone()); reply!(stream, transport, upgrade) } TransportName::Unix => { let transport = UnixTransport::new(); let stream = transport.dial(uri.clone()).await; if let Err(err) = stream { error!("RPC Connection to {} failed: {}", uri, err); return Err(Error::ConnectFailed) } get_reply(stream?, data_str).await } _ => unimplemented!(), } } async fn get_reply(mut stream: T, data_str: String) -> Result { // If we don't get a reply after 30 seconds, we'll fail. let read_timeout = Duration::from_secs(30); let mut buf = [0; 8192]; stream.write_all(data_str.as_bytes()).await?; let bytes_read = timeout(read_timeout, async { stream.read(&mut buf[..]).await }).await?; let reply: JsonResult = serde_json::from_slice(&buf[0..bytes_read])?; Ok(reply) } // Utils to quickly handle errors pub type ValueResult = std::result::Result; pub fn from_result(res: ValueResult, id: Value) -> JsonResult { match res { Ok(v) => JsonResult::Resp(response(v, id)), Err(e) => error(e, None, id).into(), } }