From 0d55d6c45599c2a4996f10c5066041369c06d9b1 Mon Sep 17 00:00:00 2001 From: Antoine Poinsot Date: Fri, 5 Aug 2022 18:43:19 +0200 Subject: [PATCH] jsonrpc: a simple JSONRPC2 server --- src/bin/cli.rs | 10 +- src/bin/daemon.rs | 4 +- src/config.rs | 9 + src/jsonrpc/api.rs | 17 ++ src/jsonrpc/mod.rs | 161 ++++++++++++++++++ src/jsonrpc/server.rs | 380 ++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 44 ++++- 7 files changed, 615 insertions(+), 10 deletions(-) create mode 100644 src/jsonrpc/api.rs create mode 100644 src/jsonrpc/mod.rs create mode 100644 src/jsonrpc/server.rs diff --git a/src/bin/cli.rs b/src/bin/cli.rs index c80b154e..870bc271 100644 --- a/src/bin/cli.rs +++ b/src/bin/cli.rs @@ -14,7 +14,7 @@ use std::os::unix::net::UnixStream; // Exits with error fn show_usage() { eprintln!("Usage:"); - eprintln!(" revault-cli [--conf conf_path] [--raw] [ ...]"); + eprintln!(" minisafe-cli [--conf conf_path] [--raw] [ ...]"); process::exit(1); } @@ -73,7 +73,7 @@ fn rpc_request(method: String, params: Vec) -> Json { object.insert("jsonrpc".to_string(), Json::String("2.0".to_string())); object.insert( "id".to_string(), - Json::String(format!("revault-cli-{}", process::id())), + Json::String(format!("minisafe-cli-{}", process::id())), ); object.insert("method".to_string(), method); object.insert("params".to_string(), params); @@ -94,7 +94,7 @@ fn socket_file(conf_file: Option) -> PathBuf { [ data_dir, config.bitcoind_config.network.to_string().as_str(), - "revaultd_rpc", + "minisafed_rpc", ] .iter() .collect() @@ -127,7 +127,7 @@ fn main() { process::exit(1); }); socket - .write_all(request.to_string().as_bytes()) + .write_all(&[request.to_string().as_bytes(), b"\n"].concat()) .unwrap_or_else(|e| { eprintln!("Writing to {:?}: '{}'", &socket_file, e); process::exit(1); @@ -160,7 +160,7 @@ fn main() { println!("{:#}", serde_json::json!({ "error": e })); } else { log::warn!( - "revaultd response doesn't contain result or error: '{}'", + "minisafed response doesn't contain result or error: '{}'", response ); println!("{:#}", response); diff --git a/src/bin/daemon.rs b/src/bin/daemon.rs index fe654182..734cf40d 100644 --- a/src/bin/daemon.rs +++ b/src/bin/daemon.rs @@ -63,7 +63,9 @@ fn main() { // The panic hook will log::error panic!("Starting Minisafe daemon: {}", e); }); - daemon.shutdown(); + daemon + .rpc_server() + .expect("JSONRPC server must terminate cleanly"); // We are always logging to stdout, should it be then piped to the log file (if self) or // not. So just make sure that all messages were actually written. diff --git a/src/config.rs b/src/config.rs index 425d559f..ec41369c 100644 --- a/src/config.rs +++ b/src/config.rs @@ -94,6 +94,15 @@ pub struct Config { pub bitcoind_config: BitcoindConfig, } +impl Config { + pub fn data_dir(&self) -> Option { + self.data_dir + .as_ref() + .map(Clone::clone) + .or_else(config_folder_path) + } +} + #[derive(PartialEq, Eq, Debug)] pub enum ConfigError { DatadirNotFound, diff --git a/src/jsonrpc/api.rs b/src/jsonrpc/api.rs new file mode 100644 index 00000000..f3a7f9a1 --- /dev/null +++ b/src/jsonrpc/api.rs @@ -0,0 +1,17 @@ +use crate::{ + jsonrpc::{Error, Request, Response}, + DaemonControl, +}; + +/// Handle an incoming JSONRPC2 request. +pub fn handle_request(control: &DaemonControl, req: Request) -> Result { + let result = match req.method.as_str() { + "getinfo" => serde_json::json!(&control.get_info()), + "getnewaddress" => serde_json::json!(&control.get_new_address()), + _ => { + return Err(Error::method_not_found()); + } + }; + + Ok(Response::success(req.id, result)) +} diff --git a/src/jsonrpc/mod.rs b/src/jsonrpc/mod.rs new file mode 100644 index 00000000..a771e666 --- /dev/null +++ b/src/jsonrpc/mod.rs @@ -0,0 +1,161 @@ +mod api; +pub mod server; + +use std::{error, fmt}; + +use serde::{self, Deserialize, Deserializer, Serialize, Serializer}; + +#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +#[serde(untagged)] +pub enum Params { + Array(Vec), + Map(serde_json::Map), +} + +#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +#[serde(untagged)] +pub enum ReqId { + Num(u64), + Str(String), +} + +/// A JSONRPC2 request. See https://www.jsonrpc.org/specification#request_object. +#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Request { + /// Version. Must be "2.0". + pub jsonrpc: String, + /// Command name. + pub method: String, + /// Command parameters. + pub params: Option, + /// Request identifier. + pub id: ReqId, +} + +/// JSONRPC2 error codes. See https://www.jsonrpc.org/specification#error_object. +#[derive(Debug, PartialEq, Clone)] +pub enum ErrorCode { + /// The method does not exist / is not available. + MethodNotFound, + /// Invalid method parameter(s). + InvalidParams, + /// Reserved for implementation-defined server-errors. + ServerError(i64), +} + +impl Into for &ErrorCode { + fn into(self) -> i64 { + match self { + ErrorCode::MethodNotFound => -32601, + ErrorCode::InvalidParams => -32602, + ErrorCode::ServerError(code) => *code, + } + } +} + +impl From for ErrorCode { + fn from(code: i64) -> ErrorCode { + match code { + -32601 => ErrorCode::MethodNotFound, + -32602 => ErrorCode::InvalidParams, + code => ErrorCode::ServerError(code), + } + } +} + +impl<'a> Deserialize<'a> for ErrorCode { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'a>, + { + let code: i64 = Deserialize::deserialize(deserializer)?; + Ok(code.into()) + } +} + +impl Serialize for ErrorCode { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_i64(self.into()) + } +} + +/// JSONRPC2 error response. See https://www.jsonrpc.org/specification#error_object. +#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Error { + pub code: ErrorCode, + pub message: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, +} + +impl Error { + pub fn new(code: ErrorCode, message: impl Into) -> Error { + Error { + message: message.into(), + code, + data: None, + } + } + + pub fn method_not_found() -> Error { + Error::new(ErrorCode::MethodNotFound, "Method not found") + } + + pub fn invalid_params(message: impl Into) -> Error { + Error::new( + ErrorCode::InvalidParams, + format!("Invalid params: {}", message.into()), + ) + } +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + let code: i64 = (&self.code).into(); + write!(f, "{}: {}", code, self.message) + } +} + +impl error::Error for Error {} + +/// JSONRPC2 response. See https://www.jsonrpc.org/specification#response_object. +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Response { + /// Version. Must be "2.0". + jsonrpc: String, + /// Required on success. Must not exist on error. + #[serde(skip_serializing_if = "Option::is_none")] + result: Option, + /// Required on error. Must not exist on success. + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + /// Request identifier. + id: ReqId, +} + +impl Response { + fn new(id: ReqId, result: Option, error: Option) -> Response { + Response { + jsonrpc: "2.0".to_string(), + result, + error, + id, + } + } + + pub fn success(id: ReqId, result: serde_json::Value) -> Response { + Response::new(id, Some(result), None) + } + + pub fn error(id: ReqId, error: Error) -> Response { + Response::new(id, None, Some(error)) + } +} diff --git a/src/jsonrpc/server.rs b/src/jsonrpc/server.rs new file mode 100644 index 00000000..a0dad02d --- /dev/null +++ b/src/jsonrpc/server.rs @@ -0,0 +1,380 @@ +//! JSONRPC2 server +//! +//! This module implements the connections and streams handling logic for receiving +//! JSONRPC2 requests on a Unix Domain Socket. + +use crate::{ + jsonrpc::{api, Request, Response}, + DaemonControl, +}; + +use std::{ + io, + os::unix::net, + path, + sync::{self, atomic}, + thread, time, +}; + +// Maximum number of concurrent RPC connections we may accept. +const MAX_CONNECTIONS: u32 = 16; + +// Read a command from the stream. +// +// In order to both treat commands separately (respond as soon as we read one), and support +// multiple commands in a single read or in multiple parts, we are given the context as writable +// arguments: +// - `buf` is the buffer used to read from the socket. It will be extended as needed. It must be +// initialized. +// - `end`: The index of the end of the data read from the stream. Since `buf` needs to be +// initialized with dummy values, it can be very different from `buf.len()`. Used to not check +// for the separator character in the parts of the buffer with dummy values. +// - `cursor`: The index at which we checked for the separator character (`\n`). Used to not +// check twice for it on the same buffer chunk. +fn read_command( + stream: &mut dyn io::Read, + buf: &mut Vec, + end: &mut usize, + cursor: &mut usize, +) -> Result, io::Error> { + assert!(!buf.is_empty()); + + loop { + // First off, check if there are no existing commands in the buffer. + let pos = buf[*cursor..*end].iter().position(|byt| byt == &b'\n'); + log::trace!( + "pos: {:?}, buf[cur..end]: {:?}", + pos, + String::from_utf8_lossy(&buf[*cursor..*end]) + ); + if let Some(pos) = pos { + log::trace!( + "Parsing Request from: {:?}", + String::from_utf8_lossy(&buf[..*cursor + pos]) + ); + // TODO: don't return an io::Error here, instead try to parse a Request. Failing that, + // try to parse a serde_json::Value. Then return accordingly a JSONRPC "malformed + // request" or "invalid JSON" error. + let req: Request = serde_json::from_slice(&buf[..*cursor + pos])?; + *buf = buf[pos + 1..].to_vec(); // FIXME: can we avoid reallocating here? + *cursor = 0; + *end -= pos + 1; + + return Ok(Some(req)); + } + + // If nothing can be gathered from the buffer, continue reading. + let new_read = stream.read(&mut buf[*end..])?; + if new_read == 0 { + return Ok(None); + } + + // If we filled the buffer, increase its size and try again. + *end += new_read; + let buffer_filled = *end == buf.len(); + if buffer_filled { + buf.resize(buf.len() * 2, 0); + continue; + } + } +} + +// Handle all messages from this connection. +fn connection_handler( + control: sync::Arc>, + mut stream: net::UnixStream, +) -> Result<(), io::Error> { + let mut buf = vec![0; 2048]; + let mut end = 0; + let mut cursor = 0; + + loop { + let req = match read_command(&mut stream, &mut buf, &mut end, &mut cursor)? { + Some(req) => req, + None => { + // Connection closed. + return Ok(()); + } + }; + + // TODO: respond in case of invalid JSON or invalid JSONRPC request. + let req_id = req.id.clone(); + + log::trace!("JSONRPC request: {:?}", serde_json::to_string(&req)); + let response = api::handle_request(&control.lock().unwrap(), req) + .unwrap_or_else(|e| Response::error(req_id, e)); + log::trace!("JSONRPC response: {:?}", serde_json::to_string(&response)); + if let Err(e) = serde_json::to_writer(&stream, &response) { + log::error!("Error writing response: '{}'", e); + return Ok(()); + } + } +} + +// FIXME: have a decent way to share the DaemonControl between connections. Maybe make it Clone? +/// The main event loop. Wait for connections, and treat requests sent through them. +pub fn rpcserver_loop( + listener: net::UnixListener, + daemon_control: sync::Arc>, +) -> Result<(), io::Error> { + // Keep it simple. We don't need great performances so just treat each connection in + // its thread, with a given maximum number of connections. + let connections_counter = sync::Arc::from(atomic::AtomicU32::new(0)); + + loop { + let (connection, _) = match listener.accept() { + Ok(c) => c, + Err(e) => { + log::error!("Accepting new connection: '{}'", e); + continue; + } + }; + log::trace!("New JSONRPC connection"); + + while connections_counter.load(atomic::Ordering::Relaxed) >= MAX_CONNECTIONS { + thread::sleep(time::Duration::from_millis(50)); + } + connections_counter.fetch_add(1, atomic::Ordering::Relaxed); + + let handler_id = connections_counter.load(atomic::Ordering::Relaxed); + thread::Builder::new() + .name(format!("minisafe-jsonrpc-{}", handler_id)) + .spawn({ + let control = daemon_control.clone(); + let counter = connections_counter.clone(); + + move || { + if let Err(e) = connection_handler(control, connection) { + log::error!("Error while handling connection {}: '{}'", handler_id, e); + } else { + log::trace!("Connection {} terminated without error.", handler_id); + } + counter.fetch_sub(1, atomic::Ordering::Relaxed); + } + })?; + } +} + +// Tries to bind to the socket, if we are told it's already in use try to connect +// to check there is actually someone listening and it's not a leftover from a +// crash. +fn bind(socket_path: &path::Path) -> Result { + match net::UnixListener::bind(socket_path) { + Ok(l) => Ok(l), + Err(e) => { + if e.kind() == io::ErrorKind::AddrInUse { + return match net::UnixStream::connect(socket_path) { + Ok(_) => Err(e), + Err(_) => { + // Ok, no one's here. Just delete the socket and bind. + log::debug!("Removing leftover rpc socket."); + std::fs::remove_file(socket_path)?; + net::UnixListener::bind(socket_path) + } + }; + } + + Err(e) + } + } +} + +/// Bind to the UDS at `socket_path` +pub fn rpcserver_setup(socket_path: &path::Path) -> Result { + log::debug!("Binding socket at {}", socket_path.display()); + // Create the socket with RW permissions only for the user + let old_umask = unsafe { libc::umask(0o177) }; + let listener = bind(&socket_path); + unsafe { + libc::umask(old_umask); + } + + listener +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::jsonrpc::{Params, ReqId}; + + use std::{env, fs, io::Write, process}; + + fn read_one_command(socket_path: &path::Path) -> thread::JoinHandle> { + let listener = rpcserver_setup(socket_path).unwrap(); + thread::spawn(move || { + let (mut conn, _) = listener.accept().unwrap(); + let mut buf = vec![0; 32]; + let mut end = 0; + let mut cursor = 0; + read_command(&mut conn, &mut buf, &mut end, &mut cursor).unwrap() + }) + } + + fn read_all_commands(socket_path: &path::Path) -> thread::JoinHandle> { + let listener = rpcserver_setup(socket_path).unwrap(); + thread::spawn(move || { + let (mut conn, _) = listener.accept().unwrap(); + let mut buf = vec![0; 32]; + let mut end = 0; + let mut cursor = 0; + let mut reqs = Vec::new(); + + loop { + match read_command(&mut conn, &mut buf, &mut end, &mut cursor).unwrap() { + Some(req) => { + reqs.push(req); + } + None => return reqs, + } + } + }) + } + + fn write_messages(socket_path: &path::Path, messages: &[&[u8]]) { + let mut client = net::UnixStream::connect(&socket_path).unwrap(); + for mess in messages { + client.write_all(&mess).unwrap(); + // Simulate throttling, this mimics real conditions and actually triggered a crash. + thread::sleep(time::Duration::from_millis(50)); + } + } + + #[test] + fn command_read_single() { + let socket_path = env::temp_dir().join(format!( + "minisafed-jsonrpc-socket-{}-{:?}", + process::id(), + thread::current().id() + )); + + // A simple dummy request + let t = read_all_commands(&socket_path); + let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": {"a": "b"}}"#; + let parsed_req: Request = serde_json::from_slice(req).unwrap(); + write_messages(&socket_path, &[req, b"\n"]); + let read_req = t.join().unwrap(); + assert_eq!(parsed_req, read_req[0]); + + // Same, but with params as a list and a string id + let t = read_one_command(&socket_path); + let req = br#"{"jsonrpc": "2.0", "id": "987-abc", "method": "test", "params": ["a", 10]}"#; + let parsed_req: Request = serde_json::from_slice(req).unwrap(); + write_messages(&socket_path, &[req, b"\n"]); + let read_req = t.join().unwrap().unwrap(); + assert_eq!(parsed_req, read_req); + + fs::remove_file(&socket_path).unwrap(); + } + + #[test] + fn command_read_parts() { + let socket_path = env::temp_dir().join(format!( + "minisafed-jsonrpc-socket-{}-{:?}", + process::id(), + thread::current().id() + )); + + // A single request written in two parts + let t = read_one_command(&socket_path); + let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": ["a", 10]}"#; + let parsed_req: Request = serde_json::from_slice(req).unwrap(); + write_messages( + &socket_path, + &[&req[..req.len() / 2], &req[req.len() / 2..], b"\n"], + ); + let read_req = t.join().unwrap().unwrap(); + assert_eq!(parsed_req, read_req); + + // A single request written in many parts + let t = read_one_command(&socket_path); + let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": ["a", 10]}"#; + let parsed_req: Request = serde_json::from_slice(req).unwrap(); + let tmp: Vec> = req.into_iter().map(|c| vec![*c]).collect(); + let mut to_send: Vec<&[u8]> = tmp.iter().map(|v| v.as_slice()).collect(); + to_send.push(b"\n"); + write_messages(&socket_path, &to_send); + let read_req = t.join().unwrap().unwrap(); + assert_eq!(parsed_req, read_req); + + fs::remove_file(&socket_path).unwrap(); + } + + #[test] + fn command_read_multiple() { + let socket_path = env::temp_dir().join(format!( + "minisafed-jsonrpc-socket-{}-{:?}", + process::id(), + thread::current().id() + )); + + // Multiple requests, in parts + let t = read_all_commands(&socket_path); + let reqs = [ + &br#"{"jsonrpc": "2.0", "id": 20478, "me"#[..], + br#"thod": "test", "params": ["a", 10]}"#, + b"\n", + br#"{"jsonrpc": "2.0", "id": 20479, "method": "testADZ", "params": {}}"#, + b"\n", + br#"{"jsonrpc": "2.0", "id": 20499, "method": "t"#, + br#"e_edzA", "params": {"ttt": 980}}"#, + b"\n", + ]; + let parsed_reqs: Vec = vec![ + serde_json::from_slice(&[reqs[0], reqs[1]].concat()).unwrap(), + serde_json::from_slice(reqs[3]).unwrap(), + serde_json::from_slice(&[reqs[5], reqs[6]].concat()).unwrap(), + ]; + write_messages(&socket_path, &reqs); + let read_reqs = t.join().unwrap(); + assert_eq!(parsed_reqs, read_reqs); + + // The same requests, sent at once. + let t = read_all_commands(&socket_path); + let req_parts = [ + &br#"{"jsonrpc": "2.0", "id": 20478, "method": "test", "params": ["a", 10]}"#[..], + b"\n", + br#"{"jsonrpc": "2.0", "id": 20479, "method": "testADZ", "params": {}}"#, + b"\n", + br#"{"jsonrpc": "2.0", "id": 20499, "method": "te_edzA", "params": {"ttt": 980}}"#, + b"\n", + ] + .concat(); + write_messages(&socket_path, &[req_parts.as_slice()]); + let read_reqs = t.join().unwrap(); + assert_eq!(parsed_reqs, read_reqs); + + fs::remove_file(&socket_path).unwrap(); + } + + #[test] + fn command_read_linebreak() { + let socket_path = env::temp_dir().join(format!( + "minisafed-jsonrpc-socket-{}-{:?}", + process::id(), + thread::current().id() + )); + + // Multiple requests, in parts + let t = read_one_command(&socket_path); + let mut params = serde_json::map::Map::new(); + params.insert( + "dummy param".to_string(), + "dummy value + with line + breaks" + .to_string() + .into(), + ); + let req = Request { + jsonrpc: "2.0".to_string(), + method: "dummy".to_string(), + params: Some(Params::Map(params)), + id: ReqId::Num(0), + }; + write_messages(&socket_path, &[&serde_json::to_vec(&req).unwrap(), b"\n"]); + let read_req = t.join().unwrap().unwrap(); + assert_eq!(req, read_req); + + fs::remove_file(&socket_path).unwrap(); + } +} diff --git a/src/lib.rs b/src/lib.rs index 9207173a..526ffeb2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,15 +5,19 @@ pub mod config; mod daemonize; mod database; pub mod descriptors; +#[cfg(feature = "jsonrpc_server")] +mod jsonrpc; pub use miniscript; +#[cfg(feature = "jsonrpc_server")] +use crate::jsonrpc::server::{rpcserver_loop, rpcserver_setup}; use crate::{ bitcoin::{ d::{BitcoinD, BitcoindError}, poller, BitcoinInterface, }, - config::{config_folder_path, Config}, + config::Config, database::{ sqlite::{FreshDbOptions, SqliteDb, SqliteDbError}, DatabaseInterface, @@ -183,9 +187,8 @@ impl DaemonHandle { // First, check the data directory let mut data_dir = config - .data_dir - .clone() - .unwrap_or(config_folder_path().ok_or(StartupError::DefaultDataDirNotFound)?); + .data_dir() + .ok_or(StartupError::DefaultDataDirNotFound)?; data_dir.push(config.bitcoind_config.network.to_string()); let fresh_data_dir = !data_dir.as_path().exists(); if fresh_data_dir { @@ -259,6 +262,39 @@ impl DaemonHandle { }) } + /// Start the JSONRPC server and listen for incoming commands until we die. + /// Like DaemonHandle::shutdown(), this stops the Bitcoin poller at teardown. + #[cfg(feature = "jsonrpc_server")] + pub fn rpc_server(self) -> Result<(), io::Error> { + let DaemonHandle { + control, + bitcoin_poller: poller, + } = self; + + let rpc_socket: path::PathBuf = [ + control + .config + .data_dir() + .expect("Didn't fail at startup, must not now") + .as_path(), + path::Path::new(&control.config.bitcoind_config.network.to_string()), + path::Path::new("minisafed_rpc"), + ] + .iter() + .collect(); + let listener = rpcserver_setup(&rpc_socket)?; + log::info!("JSONRPC server started."); + + // FIXME: don't use a Mutex... + let control = sync::Arc::from(sync::Mutex::from(control)); + rpcserver_loop(listener, control)?; + log::info!("JSONRPC server stopped."); + + poller.stop(); + + Ok(()) + } + // NOTE: this moves out the data as it should not be reused after shutdown /// Shut down the Minisafe daemon. pub fn shutdown(self) {