(fix): HTTP error handling in VC API

common/warp_utils/src/task.rs

@@ -11,7 +11,6 @@ where
         .await
         .unwrap_or_else(|_| Err(warp::reject::reject()))
 }
-
 /// A convenience wrapper around `blocking_task` that returns a `warp::reply::Response`.
 ///
 /// Using this method consistently makes it possible to simplify types using `.unify()` or `.uor()`.

validator_client/src/http_api/mod.rs

@@ -39,6 +39,7 @@ use task_executor::TaskExecutor;
 use tokio_stream::{wrappers::BroadcastStream, StreamExt};
 use types::{ChainSpec, ConfigAndPreset, EthSpec};
 use validator_dir::Builder as ValidatorDirBuilder;
+use warp::reply::Reply;
 use warp::{
     http::{
         header::{HeaderValue, CONTENT_TYPE},
@@ -265,7 +266,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path("version"))
         .and(warp::path::end())
         .and(signer.clone())
-        .and_then(|signer| {
+        .then(|signer| {
             blocking_signed_json_task(signer, move || {
                 Ok(api_types::GenericResponse::from(api_types::VersionData {
                     version: version_with_platform(),
@@ -278,7 +279,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path("health"))
         .and(warp::path::end())
         .and(signer.clone())
-        .and_then(|signer| {
+        .then(|signer| {
             blocking_signed_json_task(signer, move || {
                 eth2::lighthouse::Health::observe()
                     .map(api_types::GenericResponse::from)
@@ -292,7 +293,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(spec_filter.clone())
         .and(signer.clone())
-        .and_then(|spec: Arc<_>, signer| {
+        .then(|spec: Arc<_>, signer| {
             blocking_signed_json_task(signer, move || {
                 let config = ConfigAndPreset::from_chain_spec::<E>(&spec, None);
                 Ok(api_types::GenericResponse::from(config))
@@ -305,7 +306,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(validator_store_filter.clone())
         .and(signer.clone())
-        .and_then(|validator_store: Arc<ValidatorStore<T, E>>, signer| {
+        .then(|validator_store: Arc<ValidatorStore<T, E>>, signer| {
             blocking_signed_json_task(signer, move || {
                 let validators = validator_store
                     .initialized_validators()
@@ -330,7 +331,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(validator_store_filter.clone())
         .and(signer.clone())
-        .and_then(
+        .then(
             |validator_pubkey: PublicKey, validator_store: Arc<ValidatorStore<T, E>>, signer| {
                 blocking_signed_json_task(signer, move || {
                     let validator = validator_store
@@ -365,7 +366,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(app_start_filter)
         .and(validator_dir_filter.clone())
         .and(signer.clone())
-        .and_then(|sysinfo, app_start: std::time::Instant, val_dir, signer| {
+        .then(|sysinfo, app_start: std::time::Instant, val_dir, signer| {
             blocking_signed_json_task(signer, move || {
                 let app_uptime = app_start.elapsed().as_secs();
                 Ok(api_types::GenericResponse::from(observe_system_health_vc(
@@ -383,7 +384,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(graffiti_flag_filter)
         .and(signer.clone())
         .and(log_filter.clone())
-        .and_then(
+        .then(
             |validator_store: Arc<ValidatorStore<T, E>>,
              graffiti_file: Option<GraffitiFile>,
              graffiti_flag: Option<Graffiti>,
@@ -421,7 +422,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(spec_filter.clone())
         .and(signer.clone())
         .and(task_executor_filter.clone())
-        .and_then(
+        .then(
             move |body: Vec<api_types::ValidatorRequest>,
                   validator_dir: PathBuf,
                   secrets_dir: PathBuf,
@@ -468,7 +469,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(spec_filter)
         .and(signer.clone())
         .and(task_executor_filter.clone())
-        .and_then(
+        .then(
             move |body: api_types::CreateValidatorsMnemonicRequest,
                   validator_dir: PathBuf,
                   secrets_dir: PathBuf,
@@ -517,7 +518,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(validator_store_filter.clone())
         .and(signer.clone())
         .and(task_executor_filter.clone())
-        .and_then(
+        .then(
             move |body: api_types::KeystoreValidatorsPostRequest,
                   validator_dir: PathBuf,
                   secrets_dir: PathBuf,
@@ -603,7 +604,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(validator_store_filter.clone())
         .and(signer.clone())
         .and(task_executor_filter.clone())
-        .and_then(
+        .then(
             |body: Vec<api_types::Web3SignerValidatorRequest>,
              validator_store: Arc<ValidatorStore<T, E>>,
              signer,
@@ -656,7 +657,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(graffiti_file_filter)
         .and(signer.clone())
         .and(task_executor_filter.clone())
-        .and_then(
+        .then(
             |validator_pubkey: PublicKey,
              body: api_types::ValidatorPatchRequest,
              validator_store: Arc<ValidatorStore<T, E>>,
@@ -726,7 +727,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
     let get_auth = get_auth
         .and(signer.clone())
         .and(api_token_path_filter)
-        .and_then(|signer, token_path: PathBuf| {
+        .then(|signer, token_path: PathBuf| {
             blocking_signed_json_task(signer, move || {
                 Ok(AuthResponse {
                     token_path: token_path.display().to_string(),
@@ -743,7 +744,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(validator_store_filter.clone())
         .and(task_executor_filter.clone())
         .and(log_filter.clone())
-        .and_then(
+        .then(
             move |request, signer, validator_store, task_executor, log| {
                 blocking_signed_json_task(signer, move || {
                     if allow_keystore_export {
@@ -770,7 +771,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(validator_store_filter.clone())
         .and(signer.clone())
-        .and_then(
+        .then(
             |validator_pubkey: PublicKey, validator_store: Arc<ValidatorStore<T, E>>, signer| {
                 blocking_signed_json_task(signer, move || {
                     if validator_store
@@ -810,7 +811,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(validator_store_filter.clone())
         .and(signer.clone())
-        .and_then(
+        .then(
             |validator_pubkey: PublicKey,
              request: api_types::UpdateFeeRecipientRequest,
              validator_store: Arc<ValidatorStore<T, E>>,
@@ -850,7 +851,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(validator_store_filter.clone())
         .and(signer.clone())
-        .and_then(
+        .then(
             |validator_pubkey: PublicKey, validator_store: Arc<ValidatorStore<T, E>>, signer| {
                 blocking_signed_json_task(signer, move || {
                     if validator_store
@@ -887,7 +888,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(validator_store_filter.clone())
         .and(signer.clone())
-        .and_then(
+        .then(
             |validator_pubkey: PublicKey, validator_store: Arc<ValidatorStore<T, E>>, signer| {
                 blocking_signed_json_task(signer, move || {
                     if validator_store
@@ -919,7 +920,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(validator_store_filter.clone())
         .and(signer.clone())
-        .and_then(
+        .then(
             |validator_pubkey: PublicKey,
              request: api_types::UpdateGasLimitRequest,
              validator_store: Arc<ValidatorStore<T, E>>,
@@ -959,7 +960,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(warp::path::end())
         .and(validator_store_filter.clone())
         .and(signer.clone())
-        .and_then(
+        .then(
             |validator_pubkey: PublicKey, validator_store: Arc<ValidatorStore<T, E>>, signer| {
                 blocking_signed_json_task(signer, move || {
                     if validator_store
@@ -1000,7 +1001,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(log_filter.clone())
         .and(signer.clone())
         .and(task_executor_filter.clone())
-        .and_then(
+        .then(
             |pubkey: PublicKey,
              query: api_types::VoluntaryExitQuery,
              validator_store: Arc<ValidatorStore<T, E>>,
@@ -1032,7 +1033,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
     let get_std_keystores = std_keystores
         .and(signer.clone())
         .and(validator_store_filter.clone())
-        .and_then(|signer, validator_store: Arc<ValidatorStore<T, E>>| {
+        .then(|signer, validator_store: Arc<ValidatorStore<T, E>>| {
             blocking_signed_json_task(signer, move || Ok(keystores::list(validator_store)))
         });
 
@@ -1045,7 +1046,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(validator_store_filter.clone())
         .and(task_executor_filter.clone())
         .and(log_filter.clone())
-        .and_then(
+        .then(
             move |request,
                   signer,
                   validator_dir,
@@ -1074,7 +1075,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(validator_store_filter.clone())
         .and(task_executor_filter.clone())
         .and(log_filter.clone())
-        .and_then(|request, signer, validator_store, task_executor, log| {
+        .then(|request, signer, validator_store, task_executor, log| {
             blocking_signed_json_task(signer, move || {
                 keystores::delete(request, validator_store, task_executor, log)
             })
@@ -1084,7 +1085,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
     let get_std_remotekeys = std_remotekeys
         .and(signer.clone())
         .and(validator_store_filter.clone())
-        .and_then(|signer, validator_store: Arc<ValidatorStore<T, E>>| {
+        .then(|signer, validator_store: Arc<ValidatorStore<T, E>>| {
             blocking_signed_json_task(signer, move || Ok(remotekeys::list(validator_store)))
         });
 
@@ -1095,7 +1096,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(validator_store_filter.clone())
         .and(task_executor_filter.clone())
         .and(log_filter.clone())
-        .and_then(|request, signer, validator_store, task_executor, log| {
+        .then(|request, signer, validator_store, task_executor, log| {
             blocking_signed_json_task(signer, move || {
                 remotekeys::import(request, validator_store, task_executor, log)
             })
@@ -1108,7 +1109,7 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
         .and(validator_store_filter)
         .and(task_executor_filter)
         .and(log_filter.clone())
-        .and_then(|request, signer, validator_store, task_executor, log| {
+        .then(|request, signer, validator_store, task_executor, log| {
             blocking_signed_json_task(signer, move || {
                 remotekeys::delete(request, validator_store, task_executor, log)
             })
@@ -1227,22 +1228,18 @@ pub fn serve<T: 'static + SlotClock + Clone, E: EthSpec>(
     Ok((listening_socket, server))
 }
 
-/// Executes `func` in blocking tokio task (i.e., where long-running tasks are permitted).
-/// JSON-encodes the return value of `func`, using the `signer` function to produce a signature of
-/// those bytes.
-pub async fn blocking_signed_json_task<S, F, T>(
-    signer: S,
-    func: F,
-) -> Result<impl warp::Reply, warp::Rejection>
+// Convert a warp `Rejection` into a `Response`.
+///
+/// This function should *always* be used to convert rejections into responses. This prevents warp
+/// from trying to backtrack in strange ways. See: https://github.com/sigp/lighthouse/issues/3404
+
+pub async fn convert_rejection<T>(res: Result<T, warp::Rejection>) -> Response<Vec<u8>>
 where
-    S: Fn(&[u8]) -> String,
-    F: FnOnce() -> Result<T, warp::Rejection> + Send + 'static,
     T: Serialize + Send + 'static,
 {
-    warp_utils::task::blocking_task(func)
-        .await
-        .map(|func_output| {
-            let mut response = match serde_json::to_vec(&func_output) {
+    match res {
+        Ok(func_output) => {
+            let response = match serde_json::to_vec(&func_output) {
                 Ok(body) => {
                     let mut res = Response::new(body);
                     res.headers_mut()
@@ -1254,14 +1251,43 @@ where
                     .body(vec![])
                     .expect("can produce simple response from static values"),
             };
-
-            let body: &Vec<u8> = response.body();
-            let signature = signer(body);
-            let header_value =
-                HeaderValue::from_str(&signature).expect("hash can be encoded as header");
-
-            response.headers_mut().append("Signature", header_value);
-
             response
-        })
+        }
+        Err(e) => {
+            let resp = warp_utils::reject::handle_rejection(e).await;
+            //Building the response with the Rejection
+            match resp {
+                Ok(reply) => {
+                    let response = reply.into_response();
+                    let (_parts, body) = response.into_parts();
+                    let body_bytes = hyper::body::to_bytes(body).await.unwrap();
+                    Response::builder()
+                        .status(StatusCode::BAD_REQUEST)
+                        .body(body_bytes.to_vec())
+                        .expect("can't produce response from rejection")
+                }
+                Err(_) => Response::builder()
+                    .status(StatusCode::INTERNAL_SERVER_ERROR)
+                    .body(vec![])
+                    .expect("unhandled error in blocking task"),
+            }
+        }
+    }
+}
+/// Executes `func` in blocking tokio task (i.e., where long-running tasks are permitted).
+/// JSON-encodes the return value of `func`, using the `signer` function to produce a signature of
+/// those bytes.
+pub async fn blocking_signed_json_task<S, F, T>(signer: S, func: F) -> Response<Vec<u8>>
+where
+    S: Fn(&[u8]) -> String,
+    F: FnOnce() -> Result<T, warp::Rejection> + Send + 'static,
+    T: Serialize + Send + 'static,
+{
+    let response = warp_utils::task::blocking_task(func).await;
+    let mut conv_res = convert_rejection(response).await;
+    let body: &Vec<u8> = conv_res.body();
+    let signature = signer(body);
+    let header_value = HeaderValue::from_str(&signature).expect("hash can be encoded as header");
+    conv_res.headers_mut().append("Signature", header_value);
+    conv_res
 }