diff --git a/src/raw_client.rs b/src/raw_client.rs index bb4d6b1..7e3a583 100644 --- a/src/raw_client.rs +++ b/src/raw_client.rs @@ -596,6 +596,7 @@ enum ChannelMessage { Response(serde_json::Value), WakeUp, Error(Arc), + InvalidResponse(serde_json::Value), } impl RawClient { @@ -713,7 +714,7 @@ impl RawClient { } trace!("<== {}", raw_resp); - let resp: serde_json::Value = serde_json::from_str(&raw_resp)?; + let mut resp: serde_json::Value = serde_json::from_str(&raw_resp)?; // Normally there is and id, but it's missing for spontaneous notifications // from the server @@ -763,12 +764,46 @@ impl RawClient { } None => { // No id, that's probably a notification. - let mut resp = resp; - - if let Some(method) = resp["method"].take().as_str() { - self.handle_notification(method, resp["params"].take())?; - } else { + if resp.is_object() { + // Never index mutably into a value we haven't proven to be an + // object: doing so panics on scalars and arrays. + let method = resp + .get("method") + .and_then(serde_json::Value::as_str) + .map(str::to_owned); + + if let Some(method) = method { + let params = resp + .get_mut("params") + .map(serde_json::Value::take) + .unwrap_or(serde_json::Value::Null); + self.handle_notification(&method, params)?; + } else { + warn!("Unexpected response: {:?}", resp); + } + } else if resp.is_null() { + // A null value has historically been ignored, since it carries + // no notification payload. Keep accepting it to preserve behavior. warn!("Unexpected response: {:?}", resp); + } else { + // Only JSON objects can be notifications. Anything else without an + // id is a protocol violation: reject it instead of panicking on + // mutable indexing, and wake every blocked caller so none of them + // can deadlock waiting for a response that will never arrive. + warn!("Unexpected non-object response: {:?}", resp); + + let mut waiting_map = self.waiting_map.lock()?; + for (_, sender) in waiting_map.drain() { + if sender + .send(ChannelMessage::InvalidResponse(resp.clone())) + .is_err() + { + warn!( + "Receiver disconnected while reporting an invalid response" + ); + } + } + return Err(Error::InvalidResponse(resp)); } } } @@ -853,6 +888,11 @@ impl RawClient { break Err(Error::SharedIOError(e)); } + ChannelMessage::InvalidResponse(resp) => { + warn!("Received ChannelMessage::InvalidResponse"); + + break Err(Error::InvalidResponse(resp)); + } } } e @ Err(_) => break e, @@ -1546,6 +1586,89 @@ mod test { } } + #[test] + fn non_object_notifications_are_rejected_without_panicking() { + // Id-less scalar and array values used to panic on mutable indexing while looking up the + // notification method. They must now surface as an ordinary protocol error. + for payload in ["5", "[1, 2]", "\"scalar\"", "true"] { + let expected: serde_json::Value = serde_json::from_str(payload).unwrap(); + let stream = + MockStream::new(format!("{payload}\n{}", server_version_response(0, "1.4"))); + let client = RawClient::from(stream); + + match client.request_server_version() { + Err(Error::InvalidResponse(value)) => assert_eq!(value, expected), + other => panic!("expected InvalidResponse for `{payload}`, got {other:?}"), + } + } + } + + #[test] + fn invalid_notification_wakes_all_waiting_requests() { + // A malformed notification must not leave other callers blocked forever waiting for a + // response that will never be read. + let client = RawClient::from(MockStream::new(b"5\n".to_vec())); + let (disconnected_sender, disconnected_receiver) = mpsc::channel(); + drop(disconnected_receiver); + let (first_sender, first_receiver) = mpsc::channel(); + let (second_sender, second_receiver) = mpsc::channel(); + { + let mut waiting_map = client.waiting_map.lock().unwrap(); + waiting_map.insert(0, disconnected_sender); + waiting_map.insert(1, first_sender); + waiting_map.insert(2, second_sender); + } + + let reader_error = client._reader_thread(Some(1)).unwrap_err(); + assert!(matches!( + reader_error, + Error::InvalidResponse(value) if value == serde_json::json!(5) + )); + assert!(client.waiting_map.lock().unwrap().is_empty()); + + for receiver in [first_receiver, second_receiver] { + let message = receiver + .recv_timeout(Duration::from_secs(1)) + .expect("invalid notification must wake every waiting request"); + assert!(matches!( + message, + ChannelMessage::InvalidResponse(value) if value == serde_json::json!(5) + )); + } + } + + #[test] + fn null_notification_is_ignored_and_request_completes() { + // Negative control from the report: an id-less null is still ignored and the in-flight + // request is served from the following line. + let stream = MockStream::new(format!("null\n{}", server_version_response(0, "1.4"))); + let client = RawClient::from(stream); + + let response = client.request_server_version().unwrap(); + assert_eq!(response.protocol_version, "1.4"); + } + + #[test] + fn valid_notification_is_processed_before_the_response() { + let notification = + r#"{"method":"blockchain.headers.subscribe","params":[{"height":42,"hex":"00"}]}"#; + let stream = MockStream::new(format!( + "{notification}\n{}", + server_version_response(0, "1.4") + )); + let client = RawClient::from(stream); + + let response = client.request_server_version().unwrap(); + assert_eq!(response.protocol_version, "1.4"); + + let notification = client + .block_headers_pop_raw() + .unwrap() + .expect("the valid notification should have been queued"); + assert_eq!(notification.height, 42); + assert_eq!(notification.header, vec![0]); + } + fn get_test_auth_client( authorization_provider: Option, ) -> RawClient {