1use crate::errors::{Error, Result};
4use bytes::Bytes;
5use http_body_util::{BodyExt, Full, Limited};
6use hyper::header::{CONTENT_LENGTH, CONTENT_TYPE, HeaderValue};
7use hyper::{Method, Request};
8use hyper_util::client::legacy::Client;
9use hyper_util::rt::TokioExecutor;
10use serde_json::{Value, json};
11use std::sync::Arc;
12use std::sync::atomic::{AtomicU64, Ordering};
13use std::time::Duration;
14use tokio::sync::OnceCell;
15
16type HyperClient = Client<
17 hyper_rustls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>,
18 Full<Bytes>,
19>;
20
21pub const MAX_JSONRPC_RESPONSE_BYTES: usize = 1024 * 1024;
23
24static HTTP: OnceCell<HyperClient> = OnceCell::const_new();
25
26async fn http_client() -> Result<&'static HyperClient> {
27 HTTP.get_or_try_init(|| async {
28 static INIT: std::sync::Once = std::sync::Once::new();
29 INIT.call_once(|| {
30 let _ = rustls::crypto::ring::default_provider().install_default();
31 });
32 let mut roots = connectrpc::rustls::RootCertStore::empty();
33 roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
34 let tls = connectrpc::rustls::ClientConfig::builder()
35 .with_root_certificates(roots)
36 .with_no_client_auth();
37 let https = hyper_rustls::HttpsConnectorBuilder::new()
38 .with_tls_config(tls)
39 .https_or_http()
40 .enable_http1()
41 .build();
42 Ok::<_, Error>(
43 Client::builder(TokioExecutor::new())
44 .pool_idle_timeout(Duration::from_secs(30))
45 .build(https),
46 )
47 })
48 .await
49}
50
51#[derive(Debug)]
53pub struct JsonRpcClient {
54 url: String,
55 timeout: Duration,
56 next_id: Arc<AtomicU64>,
57}
58
59impl Clone for JsonRpcClient {
60 fn clone(&self) -> Self {
61 Self {
62 url: self.url.clone(),
63 timeout: self.timeout,
64 next_id: self.next_id.clone(),
65 }
66 }
67}
68
69impl JsonRpcClient {
70 pub fn new(url: impl Into<String>, timeout: Duration) -> Self {
71 Self {
72 url: url.into(),
73 timeout,
74 next_id: Arc::new(AtomicU64::new(0)),
75 }
76 }
77
78 pub async fn request(&self, method: &str, params: Value) -> Result<Value> {
79 let id = self.next_id.fetch_add(1, Ordering::Relaxed) + 1;
80 let payload = json!({
81 "jsonrpc": "2.0",
82 "id": id,
83 "method": method,
84 "params": params,
85 });
86 let body = serde_json::to_vec(&payload)
87 .map_err(|e| Error::transport(format!("jsonrpc encode: {e}")))?;
88
89 let uri: hyper::Uri = self
90 .url
91 .parse()
92 .map_err(|e| Error::transport(format!("jsonrpc invalid url: {e}")))?;
93 let req = Request::builder()
94 .method(Method::POST)
95 .uri(uri)
96 .header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
97 .body(Full::new(Bytes::from(body)))
98 .map_err(|e| Error::transport(format!("jsonrpc request build: {e}")))?;
99
100 let client = http_client().await?;
101 let timeout = self.timeout;
102 let (status, bytes) = tokio::time::timeout(timeout, async {
103 let resp = client
104 .request(req)
105 .await
106 .map_err(|e| Error::transport(format!("jsonrpc HTTP request failed: {e}")))?;
107 let status = resp.status();
108 if content_length_exceeds_limit(resp.headers(), MAX_JSONRPC_RESPONSE_BYTES) {
109 return Err(Error::transport(format!(
110 "jsonrpc response exceeds {MAX_JSONRPC_RESPONSE_BYTES} bytes"
111 )));
112 }
113 let bytes = Limited::new(resp.into_body(), MAX_JSONRPC_RESPONSE_BYTES)
114 .collect()
115 .await
116 .map_err(|e| Error::transport(format!("jsonrpc read body: {e}")))?
117 .to_bytes();
118 Ok::<_, Error>((status, bytes))
119 })
120 .await
121 .map_err(|_| Error::transport(format!("jsonrpc timeout calling {method}")))??;
122
123 if !status.is_success() {
124 return Err(Error::transport(format!(
125 "jsonrpc HTTP {}: {}",
126 status.as_u16(),
127 String::from_utf8_lossy(&bytes)
128 )));
129 }
130
131 let body: Value = serde_json::from_slice(&bytes)
132 .map_err(|e| Error::transport(format!("jsonrpc invalid JSON: {e}")))?;
133 parse_jsonrpc_result(&body, id, method)
134 }
135}
136
137fn content_length_exceeds_limit(headers: &http::HeaderMap, max_bytes: usize) -> bool {
138 headers
139 .get(CONTENT_LENGTH)
140 .and_then(|value| value.to_str().ok())
141 .and_then(|value| value.parse::<usize>().ok())
142 .is_some_and(|length| length > max_bytes)
143}
144
145pub fn parse_jsonrpc_result(body: &Value, expected_id: u64, method: &str) -> Result<Value> {
147 let obj = body
148 .as_object()
149 .ok_or_else(|| Error::transport("jsonrpc response must be a JSON object".to_owned()))?;
150
151 match obj.get("jsonrpc").and_then(|v| v.as_str()) {
152 Some("2.0") => {}
153 Some(other) => {
154 return Err(Error::transport(format!(
155 "jsonrpc unsupported version: {other}"
156 )));
157 }
158 None => {
159 return Err(Error::transport(
160 "jsonrpc response missing jsonrpc version".to_owned(),
161 ));
162 }
163 }
164
165 let id_ok = match obj.get("id") {
166 Some(Value::Number(n)) => n.as_u64() == Some(expected_id),
167 Some(Value::String(s)) => s.parse::<u64>().ok() == Some(expected_id),
168 _ => false,
169 };
170 if !id_ok {
171 return Err(Error::transport(format!(
172 "jsonrpc response id mismatch (expected {expected_id})"
173 )));
174 }
175
176 let has_result = obj.contains_key("result");
177 let error = obj.get("error").filter(|e| !e.is_null());
178 match (has_result, error) {
179 (true, None) => Ok(obj.get("result").cloned().unwrap_or(Value::Null)),
180 (false, Some(err)) => {
181 if !err.is_object() {
182 return Err(Error::transport(format!(
183 "jsonrpc {method}: error must be an object"
184 )));
185 }
186 let message = err
187 .get("message")
188 .and_then(|m| m.as_str())
189 .unwrap_or("jsonrpc error");
190 Err(Error::transport(format!("jsonrpc {method}: {message}")))
191 }
192 (true, Some(_)) => Err(Error::transport(format!(
193 "jsonrpc {method}: response must not include both result and error"
194 ))),
195 (false, None) => Err(Error::transport(format!(
196 "jsonrpc {method}: response must include result or error"
197 ))),
198 }
199}
200
201#[cfg(test)]
202mod tests {
203 use super::*;
204
205 #[test]
206 fn envelope_rejects_both_result_and_error() {
207 let body = json!({
208 "jsonrpc": "2.0",
209 "id": 1,
210 "result": 1,
211 "error": {"code": -1, "message": "nope"},
212 });
213 let err = parse_jsonrpc_result(&body, 1, "eth_call").unwrap_err();
214 assert!(err.to_string().contains("both result and error"));
215 }
216
217 #[test]
218 fn envelope_rejects_neither_result_nor_error() {
219 let body = json!({"jsonrpc": "2.0", "id": 1});
220 assert!(parse_jsonrpc_result(&body, 1, "eth_call").is_err());
221 }
222
223 #[test]
224 fn envelope_rejects_id_mismatch() {
225 let body = json!({"jsonrpc": "2.0", "id": 9, "result": true});
226 assert!(parse_jsonrpc_result(&body, 1, "eth_call").is_err());
227 }
228
229 #[test]
230 fn envelope_accepts_null_result() {
231 let body = json!({"jsonrpc": "2.0", "id": 1, "result": null});
232 assert_eq!(
233 parse_jsonrpc_result(&body, 1, "eth_call").unwrap(),
234 Value::Null
235 );
236 }
237}