lightning_block_sync/
rpc.rs

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

lightning_block_sync/
rpc.rs
