anvil_server/
handler.rs

1use crate::RpcHandler;
2use anvil_rpc::{
3    error::RpcError,
4    request::{Id, Request, RpcCall, RpcMethodCall, Version},
5    response::{Response, RpcResponse},
6};
7use axum::{
8    Json,
9    extract::{State, rejection::JsonRejection},
10    http::StatusCode,
11    response::{IntoResponse, Response as AxumResponse},
12};
13use futures::{FutureExt, future};
14
15/// Handles incoming JSON-RPC Request.
16// NOTE: `handler` must come first because the `request` extractor consumes the request body.
17pub async fn handle<Http: RpcHandler, Ws>(
18    State((handler, _)): State<(Http, Ws)>,
19    request: Result<Json<Request>, JsonRejection>,
20) -> AxumResponse {
21    match request {
22        Ok(Json(req)) => handle_request(req, handler)
23            .await
24            .map(Json)
25            .map(IntoResponse::into_response)
26            .unwrap_or_else(|| StatusCode::NO_CONTENT.into_response()),
27        Err(JsonRejection::JsonSyntaxError(err)) => {
28            warn!(target: "rpc", ?err, "invalid request");
29            Json(Response::error(RpcError::parse_error())).into_response()
30        }
31        Err(err) => {
32            warn!(target: "rpc", ?err, "invalid request");
33            Json(Response::error(RpcError::invalid_request())).into_response()
34        }
35    }
36}
37
38/// Handle the JSON-RPC [Request]
39///
40/// This will try to deserialize the payload into the request type of the handler and if successful
41/// invoke the handler
42pub async fn handle_request<Handler: RpcHandler>(
43    req: Request,
44    handler: Handler,
45) -> Option<Response> {
46    /// processes batch calls
47    fn responses_as_batch(outs: Vec<Option<RpcResponse>>) -> Option<Response> {
48        let batch: Vec<_> = outs.into_iter().flatten().collect();
49        (!batch.is_empty()).then_some(Response::Batch(batch))
50    }
51
52    match req {
53        Request::Single(call) => handle_call(call, handler).await.map(Response::Single),
54        Request::Batch(calls) => {
55            if calls.is_empty() {
56                return Some(Response::error(RpcError::invalid_request()));
57            }
58            future::join_all(calls.into_iter().map(move |call| handle_call(call, handler.clone())))
59                .map(responses_as_batch)
60                .await
61        }
62    }
63}
64
65/// handle a single RPC method call
66async fn handle_call<Handler: RpcHandler>(call: RpcCall, handler: Handler) -> Option<RpcResponse> {
67    match call {
68        RpcCall::MethodCall(call) => {
69            // Detailed logging with params is done in on_call
70            Some(handler.on_call(call).await)
71        }
72        RpcCall::Notification(notification) => {
73            let call = RpcMethodCall {
74                jsonrpc: notification.jsonrpc.unwrap_or(Version::V2),
75                method: notification.method,
76                params: notification.params,
77                id: Id::Null,
78            };
79            trace!(target: "rpc", method = ?call.method, "handling rpc notification");
80            drop(handler.on_call(call).await);
81            None
82        }
83        RpcCall::Invalid { id } => {
84            warn!(target: "rpc", ?id,  "invalid rpc call");
85            Some(RpcResponse::invalid_request(id))
86        }
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93    use crate::{ServerConfig, http_router};
94    use anvil_rpc::{
95        request::{RequestParams, RpcNotification},
96        response::ResponseResult,
97    };
98    use axum::{
99        body::{Body, to_bytes},
100        http::{Request as HttpRequest, header},
101    };
102    use std::{
103        pin::pin,
104        sync::{
105            Arc, Mutex,
106            atomic::{AtomicUsize, Ordering},
107        },
108        task::{Context, Poll, Waker},
109    };
110    use tower::ServiceExt;
111
112    #[derive(Clone, Default)]
113    struct TestHandler {
114        requests: Arc<AtomicUsize>,
115    }
116
117    #[derive(Clone, Debug, serde::Deserialize)]
118    #[serde(tag = "method", content = "params")]
119    enum TypedRequest {
120        #[serde(rename = "known", alias = "known_alias")]
121        Known(Vec<TestParam>),
122    }
123
124    #[derive(Clone, Debug, serde::Deserialize)]
125    enum TestParam {
126        #[serde(rename = "allowed")]
127        Allowed,
128    }
129
130    #[derive(Clone, Default)]
131    struct ErrorHandler {
132        errors: Arc<Mutex<Vec<RecordedError>>>,
133    }
134
135    type RecordedError = (String, Id, serde_json::Value, RpcError);
136
137    #[derive(Clone, Default)]
138    struct CredentialHandler {
139        observed: Arc<Mutex<Vec<serde_json::Value>>>,
140    }
141
142    impl RpcHandler for TestHandler {
143        type Request = serde_json::Value;
144
145        async fn on_request(&self, _request: Self::Request) -> ResponseResult {
146            self.requests.fetch_add(1, Ordering::Relaxed);
147            ResponseResult::success(())
148        }
149    }
150
151    impl RpcHandler for ErrorHandler {
152        type Request = TypedRequest;
153
154        async fn on_request(&self, request: Self::Request) -> ResponseResult {
155            let TypedRequest::Known(params) = request;
156            drop(params);
157            ResponseResult::Error(RpcError::internal_error())
158        }
159
160        fn on_request_error(
161            &self,
162            method: &str,
163            id: &Id,
164            params: &serde_json::Value,
165            error: &RpcError,
166        ) {
167            self.errors.lock().unwrap().push((
168                method.to_owned(),
169                id.clone(),
170                params.clone(),
171                error.clone(),
172            ));
173        }
174    }
175
176    impl RpcHandler for CredentialHandler {
177        type Request = serde_json::Value;
178
179        async fn on_request(&self, request: Self::Request) -> ResponseResult {
180            self.observed.lock().unwrap().push(request);
181            ResponseResult::Error(RpcError::internal_error())
182        }
183
184        fn on_request_error(
185            &self,
186            _method: &str,
187            _id: &Id,
188            params: &serde_json::Value,
189            _error: &RpcError,
190        ) {
191            self.observed.lock().unwrap().push(params.clone());
192        }
193    }
194
195    fn notification() -> RpcCall {
196        RpcCall::Notification(RpcNotification {
197            jsonrpc: Some(Version::V2),
198            method: "record".to_owned(),
199            params: RequestParams::None,
200        })
201    }
202
203    fn method_call(id: i64) -> RpcCall {
204        RpcCall::MethodCall(RpcMethodCall {
205            jsonrpc: Version::V2,
206            method: "record".to_owned(),
207            params: RequestParams::None,
208            id: Id::Number(id),
209        })
210    }
211
212    fn typed_call(method: &str) -> RpcMethodCall {
213        RpcMethodCall {
214            jsonrpc: Version::V2,
215            method: method.to_owned(),
216            params: RequestParams::Array(vec![serde_json::json!("bogus")]),
217            id: Id::Number(1),
218        }
219    }
220
221    fn run_ready<F: Future>(future: F) -> F::Output {
222        let waker = Waker::noop();
223        let mut cx = Context::from_waker(waker);
224        let mut future = pin!(future);
225        match future.as_mut().poll(&mut cx) {
226            Poll::Ready(output) => output,
227            Poll::Pending => panic!("future unexpectedly pending"),
228        }
229    }
230
231    #[test]
232    fn empty_batch_returns_invalid_request() {
233        let response = run_ready(handle_request(Request::Batch(vec![]), TestHandler::default()));
234
235        assert_eq!(response, Some(Response::error(RpcError::invalid_request())));
236    }
237
238    #[test]
239    fn distinguishes_deserialization_errors_without_invoking_error_hook() {
240        let handler = ErrorHandler::default();
241        let invalid_params = run_ready(handler.on_call(typed_call("known")));
242        let unknown_method = run_ready(handler.on_call(typed_call("unknown")));
243
244        assert_eq!(
245            invalid_params,
246            RpcResponse::new(
247                Id::Number(1),
248                RpcError::invalid_params("unknown variant `bogus`, expected `allowed`")
249            )
250        );
251        assert_eq!(unknown_method, RpcResponse::new(Id::Number(1), RpcError::method_not_found()));
252        assert!(handler.errors.lock().unwrap().is_empty());
253    }
254
255    #[test]
256    fn notification_only_batch_executes_without_response() {
257        let handler = TestHandler::default();
258        let response = run_ready(handle_request(
259            Request::Batch(vec![notification(), notification()]),
260            handler.clone(),
261        ));
262
263        assert_eq!(response, None);
264        assert_eq!(handler.requests.load(Ordering::Relaxed), 2);
265    }
266
267    #[test]
268    fn mixed_batch_executes_notification_without_including_response() {
269        let handler = TestHandler::default();
270        let response = run_ready(handle_request(
271            Request::Batch(vec![notification(), method_call(1)]),
272            handler.clone(),
273        ));
274
275        assert_eq!(
276            response,
277            Some(Response::Batch(vec![RpcResponse::new(
278                Id::Number(1),
279                ResponseResult::success(())
280            )]))
281        );
282        assert_eq!(handler.requests.load(Ordering::Relaxed), 2);
283    }
284
285    #[test]
286    fn http_notification_returns_no_content_after_execution() {
287        let handler = TestHandler::default();
288        let response = run_ready(handle(
289            State((handler.clone(), ())),
290            Ok(Json(Request::Single(notification()))),
291        ));
292
293        assert_eq!(response.status(), StatusCode::NO_CONTENT);
294        assert!(run_ready(to_bytes(response.into_body(), usize::MAX)).unwrap().is_empty());
295        assert_eq!(handler.requests.load(Ordering::Relaxed), 1);
296    }
297
298    #[test]
299    fn malformed_json_returns_parse_error() {
300        let request = HttpRequest::post("/")
301            .header(header::CONTENT_TYPE, "application/json")
302            .body(Body::from("{"))
303            .unwrap();
304        let response = run_ready(
305            http_router(ServerConfig::default(), TestHandler::default()).oneshot(request),
306        )
307        .unwrap();
308        let body = run_ready(to_bytes(response.into_body(), usize::MAX)).unwrap();
309
310        assert_eq!(
311            serde_json::from_slice::<serde_json::Value>(&body).unwrap(),
312            serde_json::to_value(Response::error(RpcError::parse_error())).unwrap()
313        );
314    }
315
316    #[test]
317    fn request_error_hook_preserves_method_alias_id_and_params() {
318        let handler = ErrorHandler::default();
319        let id = Id::String("request-7".to_owned());
320        let call = RpcMethodCall {
321            params: RequestParams::Array(vec![serde_json::json!("allowed")]),
322            id: id.clone(),
323            ..typed_call("known_alias")
324        };
325
326        let response = run_ready(handler.on_call(call));
327
328        assert_eq!(response, RpcResponse::new(id.clone(), RpcError::internal_error()));
329        assert_eq!(
330            *handler.errors.lock().unwrap(),
331            vec![(
332                "known_alias".to_owned(),
333                id,
334                serde_json::json!(["allowed"]),
335                RpcError::internal_error()
336            )]
337        );
338    }
339
340    #[test]
341    fn request_error_hook_handles_notifications_without_response() {
342        let handler = ErrorHandler::default();
343        let call = RpcCall::Notification(RpcNotification {
344            jsonrpc: Some(Version::V2),
345            method: "known".to_owned(),
346            params: RequestParams::Array(vec![serde_json::json!("allowed")]),
347        });
348
349        let response = run_ready(handle_request(Request::Single(call), handler.clone()));
350
351        assert_eq!(response, None);
352        assert_eq!(
353            *handler.errors.lock().unwrap(),
354            vec![(
355                "known".to_owned(),
356                Id::Null,
357                serde_json::json!(["allowed"]),
358                RpcError::internal_error()
359            )]
360        );
361    }
362
363    #[test]
364    fn request_error_hook_redacts_credentials_after_execution() {
365        let handler = CredentialHandler::default();
366        let params = serde_json::json!({"params": {"forking": {
367            "jsonRpcUrl": "https://user:pass@example.invalid/key?token=x#secret",
368            "blockNumber": 42
369        }}});
370        let call = serde_json::from_value(serde_json::json!({
371            "jsonrpc": "2.0",
372            "method": "hardhat_reset",
373            "params": params,
374            "id": "reset"
375        }))
376        .unwrap();
377
378        let response = run_ready(handler.on_call(call));
379
380        assert_eq!(
381            response,
382            RpcResponse::new(Id::String("reset".to_owned()), RpcError::internal_error())
383        );
384        assert_eq!(
385            *handler.observed.lock().unwrap(),
386            vec![
387                serde_json::json!({"method": "hardhat_reset", "params": params}),
388                serde_json::json!({"params": {"forking": {
389                    "jsonRpcUrl": "https://example.invalid/",
390                    "blockNumber": 42
391                }}})
392            ]
393        );
394    }
395}
anvil_server/handler.rs

anvil_server/
handler.rs
