_ Git - rust-lightning/blob - lightning-block-sync/src/rpc.rs

   1 //! Simple RPC client implementation which implements [`BlockSource`] against a Bitcoin Core RPC
   2 //! endpoint.
   3
   4 use crate::{BlockData, BlockHeaderData, BlockSource, AsyncBlockSourceResult};
   5 use crate::http::{HttpClient, HttpEndpoint, HttpError, JsonResponse};
   6 use crate::gossip::UtxoSource;
   7
   8 use bitcoin::hash_types::BlockHash;
   9 use bitcoin::OutPoint;
  10
  11 use std::sync::Mutex;
  12
  13 use serde_json;
  14
  15 use std::convert::TryFrom;
  16 use std::convert::TryInto;
  17 use std::error::Error;
  18 use std::fmt;
  19 use std::sync::atomic::{AtomicUsize, Ordering};
  20
  21 /// An error returned by the RPC server.
  22 #[derive(Debug)]
  23 pub struct RpcError {
  24         /// The error code.
  25         pub code: i64,
  26         /// The error message.
  27         pub message: String,
  28 }
  29
  30 impl fmt::Display for RpcError {
  31     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
  32         write!(f, "RPC error {}: {}", self.code, self.message)
  33     }
  34 }
  35
  36 impl Error for RpcError {}
  37
  38 /// A simple RPC client for calling methods using HTTP `POST`.
  39 ///
  40 /// Implements [`BlockSource`] and may return an `Err` containing [`RpcError`]. See
  41 /// [`RpcClient::call_method`] for details.
  42 pub struct RpcClient {
  43         basic_auth: String,
  44         endpoint: HttpEndpoint,
  45         client: Mutex<Option<HttpClient>>,
  46         id: AtomicUsize,
  47 }
  48
  49 impl RpcClient {
  50         /// Creates a new RPC client connected to the given endpoint with the provided credentials. The
  51         /// credentials should be a base64 encoding of a user name and password joined by a colon, as is
  52         /// required for HTTP basic access authentication.
  53         pub fn new(credentials: &str, endpoint: HttpEndpoint) -> std::io::Result<Self> {
  54                 Ok(Self {
  55                         basic_auth: "Basic ".to_string() + credentials,
  56                         endpoint,
  57                         client: Mutex::new(None),
  58                         id: AtomicUsize::new(0),
  59                 })
  60         }
  61
  62         /// Calls a method with the response encoded in JSON format and interpreted as type `T`.
  63         ///
  64         /// When an `Err` is returned, [`std::io::Error::into_inner`] may contain an [`RpcError`] if
  65         /// [`std::io::Error::kind`] is [`std::io::ErrorKind::Other`].
  66         pub async fn call_method<T>(&self, method: &str, params: &[serde_json::Value]) -> std::io::Result<T>
  67         where JsonResponse: TryFrom<Vec<u8>, Error = std::io::Error> + TryInto<T, Error = std::io::Error> {
  68                 let host = format!("{}:{}", self.endpoint.host(), self.endpoint.port());
  69                 let uri = self.endpoint.path();
  70                 let content = serde_json::json!({
  71                         "method": method,
  72                         "params": params,
  73                         "id": &self.id.fetch_add(1, Ordering::AcqRel).to_string()
  74                 });
  75
  76                 let mut client = if let Some(client) = self.client.lock().unwrap().take() { client }
  77                         else { HttpClient::connect(&self.endpoint)? };
  78                 let http_response = client.post::<JsonResponse>(&uri, &host, &self.basic_auth, content).await;
  79                 *self.client.lock().unwrap() = Some(client);
  80
  81                 let mut response = match http_response {
  82                         Ok(JsonResponse(response)) => response,
  83                         Err(e) if e.kind() == std::io::ErrorKind::Other => {
  84                                 match e.get_ref().unwrap().downcast_ref::<HttpError>() {
  85                                         Some(http_error) => match JsonResponse::try_from(http_error.contents.clone()) {
  86                                                 Ok(JsonResponse(response)) => response,
  87                                                 Err(_) => Err(e)?,
  88                                         },
  89                                         None => Err(e)?,
  90                                 }
  91                         },
  92                         Err(e) => Err(e)?,
  93                 };
  94
  95                 if !response.is_object() {
  96                         return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "expected JSON object"));
  97                 }
  98
  99                 let error = &response["error"];
 100                 if !error.is_null() {
 101                         // TODO: Examine error code for a more precise std::io::ErrorKind.
 102                         let rpc_error = RpcError {
 103                                 code: error["code"].as_i64().unwrap_or(-1),
 104                                 message: error["message"].as_str().unwrap_or("unknown error").to_string()
 105                         };
 106                         return Err(std::io::Error::new(std::io::ErrorKind::Other, rpc_error));
 107                 }
 108
 109                 let result = match response.get_mut("result") {
 110                         Some(result) => result.take(),
 111                         None =>
 112                                 return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "expected JSON result")),
 113                 };
 114
 115                 JsonResponse(result).try_into()
 116         }
 117 }
 118
 119 impl BlockSource for RpcClient {
 120         fn get_header<'a>(&'a self, header_hash: &'a BlockHash, _height: Option<u32>) -> AsyncBlockSourceResult<'a, BlockHeaderData> {
 121                 Box::pin(async move {
 122                         let header_hash = serde_json::json!(header_hash.to_string());
 123                         Ok(self.call_method("getblockheader", &[header_hash]).await?)
 124                 })
 125         }
 126
 127         fn get_block<'a>(&'a self, header_hash: &'a BlockHash) -> AsyncBlockSourceResult<'a, BlockData> {
 128                 Box::pin(async move {
 129                         let header_hash = serde_json::json!(header_hash.to_string());
 130                         let verbosity = serde_json::json!(0);
 131                         Ok(BlockData::FullBlock(self.call_method("getblock", &[header_hash, verbosity]).await?))
 132                 })
 133         }
 134
 135         fn get_best_block<'a>(&'a self) -> AsyncBlockSourceResult<'a, (BlockHash, Option<u32>)> {
 136                 Box::pin(async move {
 137                         Ok(self.call_method("getblockchaininfo", &[]).await?)
 138                 })
 139         }
 140 }
 141
 142 impl UtxoSource for RpcClient {
 143         fn get_block_hash_by_height<'a>(&'a self, block_height: u32) -> AsyncBlockSourceResult<'a, BlockHash> {
 144                 Box::pin(async move {
 145                         let height_param = serde_json::json!(block_height);
 146                         Ok(self.call_method("getblockhash", &[height_param]).await?)
 147                 })
 148         }
 149
 150         fn is_output_unspent<'a>(&'a self, outpoint: OutPoint) -> AsyncBlockSourceResult<'a, bool> {
 151                 Box::pin(async move {
 152                         let txid_param = serde_json::json!(outpoint.txid.to_string());
 153                         let vout_param = serde_json::json!(outpoint.vout);
 154                         let include_mempool = serde_json::json!(false);
 155                         let utxo_opt: serde_json::Value = self.call_method(
 156                                 "gettxout", &[txid_param, vout_param, include_mempool]).await?;
 157                         Ok(!utxo_opt.is_null())
 158                 })
 159         }
 160 }
 161
 162 #[cfg(test)]
 163 mod tests {
 164         use super::*;
 165         use crate::http::client_tests::{HttpServer, MessageBody};
 166
 167         use bitcoin::hashes::Hash;
 168
 169         /// Credentials encoded in base64.
 170         const CREDENTIALS: &'static str = "dXNlcjpwYXNzd29yZA==";
 171
 172         /// Converts a JSON value into `u64`.
 173         impl TryInto<u64> for JsonResponse {
 174                 type Error = std::io::Error;
 175
 176                 fn try_into(self) -> std::io::Result<u64> {
 177                         match self.0.as_u64() {
 178                                 None => Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "not a number")),
 179                                 Some(n) => Ok(n),
 180                         }
 181                 }
 182         }
 183
 184         #[tokio::test]
 185         async fn call_method_returning_unknown_response() {
 186                 let server = HttpServer::responding_with_not_found();
 187                 let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap();
 188
 189                 match client.call_method::<u64>("getblockcount", &[]).await {
 190                         Err(e) => assert_eq!(e.kind(), std::io::ErrorKind::Other),
 191                         Ok(_) => panic!("Expected error"),
 192                 }
 193         }
 194
 195         #[tokio::test]
 196         async fn call_method_returning_malfomred_response() {
 197                 let response = serde_json::json!("foo");
 198                 let server = HttpServer::responding_with_ok(MessageBody::Content(response));
 199                 let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap();
 200
 201                 match client.call_method::<u64>("getblockcount", &[]).await {
 202                         Err(e) => {
 203                                 assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
 204                                 assert_eq!(e.get_ref().unwrap().to_string(), "expected JSON object");
 205                         },
 206                         Ok(_) => panic!("Expected error"),
 207                 }
 208         }
 209
 210         #[tokio::test]
 211         async fn call_method_returning_error() {
 212                 let response = serde_json::json!({
 213                         "error": { "code": -8, "message": "invalid parameter" },
 214                 });
 215                 let server = HttpServer::responding_with_server_error(response);
 216                 let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap();
 217
 218                 let invalid_block_hash = serde_json::json!("foo");
 219                 match client.call_method::<u64>("getblock", &[invalid_block_hash]).await {
 220                         Err(e) => {
 221                                 assert_eq!(e.kind(), std::io::ErrorKind::Other);
 222                                 let rpc_error: Box<RpcError> = e.into_inner().unwrap().downcast().unwrap();
 223                                 assert_eq!(rpc_error.code, -8);
 224                                 assert_eq!(rpc_error.message, "invalid parameter");
 225                         },
 226                         Ok(_) => panic!("Expected error"),
 227                 }
 228         }
 229
 230         #[tokio::test]
 231         async fn call_method_returning_missing_result() {
 232                 let response = serde_json::json!({  });
 233                 let server = HttpServer::responding_with_ok(MessageBody::Content(response));
 234                 let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap();
 235
 236                 match client.call_method::<u64>("getblockcount", &[]).await {
 237                         Err(e) => {
 238                                 assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
 239                                 assert_eq!(e.get_ref().unwrap().to_string(), "expected JSON result");
 240                         },
 241                         Ok(_) => panic!("Expected error"),
 242                 }
 243         }
 244
 245         #[tokio::test]
 246         async fn call_method_returning_malformed_result() {
 247                 let response = serde_json::json!({ "result": "foo" });
 248                 let server = HttpServer::responding_with_ok(MessageBody::Content(response));
 249                 let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap();
 250
 251                 match client.call_method::<u64>("getblockcount", &[]).await {
 252                         Err(e) => {
 253                                 assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
 254                                 assert_eq!(e.get_ref().unwrap().to_string(), "not a number");
 255                         },
 256                         Ok(_) => panic!("Expected error"),
 257                 }
 258         }
 259
 260         #[tokio::test]
 261         async fn call_method_returning_valid_result() {
 262                 let response = serde_json::json!({ "result": 654470 });
 263                 let server = HttpServer::responding_with_ok(MessageBody::Content(response));
 264                 let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap();
 265
 266                 match client.call_method::<u64>("getblockcount", &[]).await {
 267                         Err(e) => panic!("Unexpected error: {:?}", e),
 268                         Ok(count) => assert_eq!(count, 654470),
 269                 }
 270         }
 271
 272         #[tokio::test]
 273         async fn fails_to_fetch_spent_utxo() {
 274                 let response = serde_json::json!({ "result": null });
 275                 let server = HttpServer::responding_with_ok(MessageBody::Content(response));
 276                 let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap();
 277                 let outpoint = OutPoint::new(bitcoin::Txid::from_byte_array([0; 32]), 0);
 278                 let unspent_output = client.is_output_unspent(outpoint).await.unwrap();
 279                 assert_eq!(unspent_output, false);
 280         }
 281
 282         #[tokio::test]
 283         async fn fetches_utxo() {
 284                 let response = serde_json::json!({ "result": {"bestblock": 1, "confirmations": 42}});
 285                 let server = HttpServer::responding_with_ok(MessageBody::Content(response));
 286                 let client = RpcClient::new(CREDENTIALS, server.endpoint()).unwrap();
 287                 let outpoint = OutPoint::new(bitcoin::Txid::from_byte_array([0; 32]), 0);
 288                 let unspent_output = client.is_output_unspent(outpoint).await.unwrap();
 289                 assert_eq!(unspent_output, true);
 290         }
 291 }