fix auth layer

src/daemon/context.rs

@@ -25,10 +25,10 @@ use crate::{
 use anyhow::Context;
 use dialoguer::console::Term;
 use fvm_shared4::address::Network;
+use parking_lot::RwLock;
 use std::cell::RefCell;
 use std::path::PathBuf;
 use std::sync::Arc;
-use tokio::sync::RwLock;
 use tracing::{info, warn};
 
 pub struct AppContext {

src/rpc/auth_layer.rs

@@ -5,17 +5,20 @@ use crate::auth::{JWT_IDENTIFIER, verify_token};
 use crate::key_management::KeyStore;
 use crate::rpc::{CANCEL_METHOD_NAME, Permission, RpcMethod as _, chain};
 use ahash::{HashMap, HashMapExt as _};
+use futures::future::Either;
 use http::{
     HeaderMap,
     header::{AUTHORIZATION, HeaderValue},
 };
+use itertools::Itertools as _;
 use jsonrpsee::MethodResponse;
-use jsonrpsee::core::middleware::{Batch, Notification};
+use jsonrpsee::core::middleware::{Batch, BatchEntry, BatchEntryErr, Notification};
 use jsonrpsee::server::middleware::rpc::RpcServiceT;
+use jsonrpsee::types::Id;
 use jsonrpsee::types::{ErrorObject, error::ErrorCode};
 use once_cell::sync::Lazy;
+use parking_lot::RwLock;
 use std::sync::Arc;
-use tokio::sync::RwLock;
 use tower::Layer;
 use tracing::debug;
 
@@ -74,9 +77,30 @@ pub struct Auth<S> {
     service: S,
 }
 
+impl<S> Auth<S> {
+    fn check_permissions<'a>(&self, method_name: &str) -> Result<(), ErrorObject<'a>> {
+        match check_permissions(&self.keystore, self.headers.get(AUTHORIZATION), method_name) {
+            Ok(true) => Ok(()),
+            Ok(false) => Err(ErrorObject::borrowed(
+                http::StatusCode::UNAUTHORIZED.as_u16() as _,
+                "Unauthorized",
+                None,
+            )),
+            Err(code) => Err(ErrorObject::from(code)),
+        }
+    }
+}
+
 impl<S> RpcServiceT for Auth<S>
 where
-    S: RpcServiceT<MethodResponse = MethodResponse> + Send + Sync + Clone + 'static,
+    S: RpcServiceT<
+            MethodResponse = MethodResponse,
+            NotificationResponse = MethodResponse,
+            BatchResponse = MethodResponse,
+        > + Send
+        + Sync
+        + Clone
+        + 'static,
 {
     type MethodResponse = S::MethodResponse;
     type NotificationResponse = S::NotificationResponse;
@@ -86,51 +110,62 @@ where
         &self,
         req: jsonrpsee::types::Request<'a>,
     ) -> impl Future<Output = Self::MethodResponse> + Send + 'a {
-        let method_name = req.method_name().to_owned();
-        let auth_header = self.headers.get(AUTHORIZATION).cloned();
-        let keystore = self.keystore.clone();
-        let service = self.service.clone();
-        async move {
-            match check_permissions(&keystore, auth_header, &method_name).await {
-                Ok(true) => service.call(req).await,
-                Ok(false) => MethodResponse::error(
-                    req.id(),
-                    ErrorObject::borrowed(
-                        http::StatusCode::UNAUTHORIZED.as_u16() as _,
-                        "Unauthorized",
-                        None,
-                    ),
-                ),
-                Err(code) => MethodResponse::error(req.id(), ErrorObject::from(code)),
-            }
+        match self.check_permissions(req.method_name()) {
+            Ok(()) => Either::Left(self.service.call(req)),
+            Err(e) => Either::Right(async move { MethodResponse::error(req.id(), e) }),
         }
     }
 
     fn batch<'a>(&self, batch: Batch<'a>) -> impl Future<Output = Self::BatchResponse> + Send + 'a {
-        self.service.batch(batch)
+        let entries = batch
+            .into_iter()
+            .filter_map(|entry| match entry {
+                Ok(BatchEntry::Call(req)) => {
+                    Some(match self.check_permissions(req.method_name()) {
+                        Ok(()) => Ok(BatchEntry::Call(req)),
+                        Err(e) => Err(BatchEntryErr::new(req.id(), e)),
+                    })
+                }
+                Ok(BatchEntry::Notification(notif)) => {
+                    match self.check_permissions(notif.method_name()) {
+                        Ok(_) => Some(Ok(BatchEntry::Notification(notif))),
+                        Err(_) => {
+                            // Just filter out the notification if the auth fails
+                            // because notifications are not expected to return a response.
+                            None
+                        }
+                    }
+                }
+                // Errors which could happen such as invalid JSON-RPC call
+                // or invalid JSON are just passed through.
+                Err(err) => Some(Err(err)),
+            })
+            .collect_vec();
+        self.service.batch(Batch::from(entries))
     }
 
     fn notification<'a>(
         &self,
         n: Notification<'a>,
     ) -> impl Future<Output = Self::NotificationResponse> + Send + 'a {
-        self.service.notification(n)
+        match self.check_permissions(n.method_name()) {
+            Ok(()) => Either::Left(self.service.notification(n)),
+            Err(e) => Either::Right(async move { MethodResponse::error(Id::Null, e) }),
+        }
     }
 }
 
 /// Verify JWT Token and return the token's permissions.
-async fn auth_verify(token: &str, keystore: &RwLock<KeyStore>) -> anyhow::Result<Vec<String>> {
-    let ks = keystore.read().await;
-    let ki = ks.get(JWT_IDENTIFIER)?;
-    let perms = verify_token(token, ki.private_key())?;
-    Ok(perms)
+fn auth_verify(token: &str, keystore: &RwLock<KeyStore>) -> anyhow::Result<Vec<String>> {
+    let key_info = keystore.read().get(JWT_IDENTIFIER)?;
+    Ok(verify_token(token, key_info.private_key())?)
 }
 
-async fn check_permissions(
+fn check_permissions(
     keystore: &RwLock<KeyStore>,
-    auth_header: Option<HeaderValue>,
+    auth_header: Option<&HeaderValue>,
     method: &str,
-) -> anyhow::Result<bool, ErrorCode> {
+) -> Result<bool, ErrorCode> {
     let claims = match auth_header {
         Some(token) => {
             let token = token
@@ -140,9 +175,7 @@ async fn check_permissions(
 
             debug!("JWT from HTTP Header: {}", token);
 
-            auth_verify(token, keystore)
-                .await
-                .map_err(|_| ErrorCode::InvalidRequest)?
+            auth_verify(token, keystore).map_err(|_| ErrorCode::InvalidRequest)?
         }
         // If no token is passed, assume read behavior
         None => vec!["read".to_owned()],
@@ -162,39 +195,39 @@ mod tests {
     use crate::rpc::wallet;
     use chrono::Duration;
 
-    #[tokio::test]
-    async fn check_permissions_no_header() {
+    #[test]
+    fn check_permissions_no_header() {
         let keystore = Arc::new(RwLock::new(
             KeyStore::new(crate::KeyStoreConfig::Memory).unwrap(),
         ));
 
-        let res = check_permissions(&keystore, None, ChainHead::NAME).await;
+        let res = check_permissions(&keystore, None, ChainHead::NAME);
         assert_eq!(res, Ok(true));
 
-        let res = check_permissions(&keystore, None, "Cthulhu.InvokeElderGods").await;
+        let res = check_permissions(&keystore, None, "Cthulhu.InvokeElderGods");
         assert_eq!(res.unwrap_err(), ErrorCode::MethodNotFound);
 
-        let res = check_permissions(&keystore, None, wallet::WalletNew::NAME).await;
+        let res = check_permissions(&keystore, None, wallet::WalletNew::NAME);
         assert_eq!(res, Ok(false));
     }
 
-    #[tokio::test]
-    async fn check_permissions_invalid_header() {
+    #[test]
+    fn check_permissions_invalid_header() {
         let keystore = Arc::new(RwLock::new(
             KeyStore::new(crate::KeyStoreConfig::Memory).unwrap(),
         ));
 
         let auth_header = HeaderValue::from_static("Bearer Azathoth");
-        let res = check_permissions(&keystore, Some(auth_header), ChainHead::NAME).await;
+        let res = check_permissions(&keystore, Some(&auth_header), ChainHead::NAME);
         assert_eq!(res.unwrap_err(), ErrorCode::InvalidRequest);
 
         let auth_header = HeaderValue::from_static("Cthulhu");
-        let res = check_permissions(&keystore, Some(auth_header), ChainHead::NAME).await;
+        let res = check_permissions(&keystore, Some(&auth_header), ChainHead::NAME);
         assert_eq!(res.unwrap_err(), ErrorCode::InvalidRequest);
     }
 
-    #[tokio::test]
-    async fn check_permissions_valid_header() {
+    #[test]
+    fn check_permissions_valid_header() {
         use crate::auth::*;
         let keystore = Arc::new(RwLock::new(
             KeyStore::new(crate::KeyStoreConfig::Memory).unwrap(),
@@ -204,7 +237,6 @@ mod tests {
         let key_info = generate_priv_key();
         keystore
             .write()
-            .await
             .put(JWT_IDENTIFIER, key_info.clone())
             .unwrap();
         let token_exp = Duration::hours(1);
@@ -217,20 +249,15 @@ mod tests {
 
         // Should work with the `Bearer` prefix
         let auth_header = HeaderValue::from_str(&format!("Bearer {token}")).unwrap();
-        let res = check_permissions(&keystore, Some(auth_header.clone()), ChainHead::NAME).await;
+        let res = check_permissions(&keystore, Some(&auth_header), ChainHead::NAME);
         assert_eq!(res, Ok(true));
 
-        let res = check_permissions(
-            &keystore,
-            Some(auth_header.clone()),
-            wallet::WalletNew::NAME,
-        )
-        .await;
+        let res = check_permissions(&keystore, Some(&auth_header), wallet::WalletNew::NAME);
         assert_eq!(res, Ok(true));
 
         // Should work without the `Bearer` prefix
         let auth_header = HeaderValue::from_str(&token).unwrap();
-        let res = check_permissions(&keystore, Some(auth_header), wallet::WalletNew::NAME).await;
+        let res = check_permissions(&keystore, Some(&auth_header), wallet::WalletNew::NAME);
         assert_eq!(res, Ok(true));
     }
 }

src/rpc/methods/auth.rs

@@ -27,7 +27,7 @@ impl RpcMethod<2> for AuthNew {
         ctx: Ctx<impl Blockstore>,
         (permissions, expiration_secs): Self::Params,
     ) -> Result<Self::Ok, ServerError> {
-        let ks = ctx.keystore.read().await;
+        let ks = ctx.keystore.read();
         let ki = ks.get(JWT_IDENTIFIER)?;
         let token = create_token(
             permissions,
@@ -51,7 +51,7 @@ impl RpcMethod<1> for AuthVerify {
         ctx: Ctx<impl Blockstore>,
         (header_raw,): Self::Params,
     ) -> Result<Self::Ok, ServerError> {
-        let ks = ctx.keystore.read().await;
+        let ks = ctx.keystore.read();
         let token = header_raw.trim_start_matches("Bearer ");
         let ki = ks.get(JWT_IDENTIFIER)?;
         let perms = verify_token(token, ki.private_key())?;

src/rpc/methods/miner.rs

@@ -31,15 +31,13 @@ use bls_signatures::Serialize as _;
 use cid::Cid;
 use fil_actors_shared::fvm_ipld_amt::Amtv0 as Amt;
 use fvm_ipld_blockstore::Blockstore;
-
 use fvm_shared2::crypto::signature::BLS_SIG_LEN;
 use group::prime::PrimeCurveAffine as _;
 use itertools::Itertools;
-
+use parking_lot::RwLock;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use serde_tuple::Serialize_tuple;
-use tokio::sync::RwLock;
 
 use std::sync::Arc;
 
@@ -213,9 +211,7 @@ impl RpcMethod<1> for MinerCreateBlock {
             parent_base_fee,
         };
 
-        block_header.signature = sign_block_header(&block_header, &worker, ctx.keystore.clone())
-            .await?
-            .into();
+        block_header.signature = sign_block_header(&block_header, &worker, &ctx.keystore)?.into();
 
         Ok(BlockMessage {
             header: CachingBlockHeader::from(block_header),
@@ -225,22 +221,23 @@ impl RpcMethod<1> for MinerCreateBlock {
     }
 }
 
-async fn sign_block_header(
+fn sign_block_header(
     block_header: &RawBlockHeader,
     worker: &Address,
-    keystore: Arc<RwLock<KeyStore>>,
+    keystore: &RwLock<KeyStore>,
 ) -> Result<Signature> {
     let signing_bytes = block_header.signing_bytes();
 
-    let mut keystore = keystore.write().await;
-    let key = match crate::key_management::find_key(worker, &keystore) {
-        Ok(key) => key,
-        Err(_) => {
-            let key_info = crate::key_management::try_find(worker, &mut keystore)?;
-            Key::try_from(key_info)?
+    let key = {
+        let mut keystore = keystore.write();
+        match crate::key_management::find_key(worker, &keystore) {
+            Ok(key) => key,
+            Err(_) => {
+                let key_info = crate::key_management::try_find(worker, &mut keystore)?;
+                Key::try_from(key_info)?
+            }
         }
     };
-    drop(keystore);
 
     let sig = crate::key_management::sign(
         *key.key_info.key_type(),

src/rpc/methods/mpool.rs

@@ -245,7 +245,6 @@ impl RpcMethod<2> for MpoolPushMessage {
     ) -> Result<Self::Ok, ServerError> {
         let from = message.from;
 
-        let mut keystore = ctx.keystore.as_ref().write().await;
         let heaviest_tipset = ctx.chain_store().heaviest_tipset();
         let key_addr = ctx
             .state_manager
@@ -274,7 +273,7 @@ impl RpcMethod<2> for MpoolPushMessage {
         message.sequence = nonce;
         let key = crate::key_management::Key::try_from(crate::key_management::try_find(
             &key_addr,
-            &mut keystore,
+            &mut ctx.keystore.as_ref().write(),
         )?)?;
         let sig = crate::key_management::sign(
             *key.key_info.key_type(),

src/rpc/methods/sync.rs

@@ -161,8 +161,9 @@ mod tests {
     use crate::shim::address::Address;
     use crate::state_manager::StateManager;
     use crate::utils::encoding::from_slice_with_fallback;
+    use parking_lot::RwLock;
     use tokio::sync::mpsc;
-    use tokio::{sync::RwLock, task::JoinSet};
+    use tokio::task::JoinSet;
 
     const TEST_NET_NAME: &str = "test";
 
@@ -233,7 +234,7 @@ mod tests {
             mpool: Arc::new(pool),
             bad_blocks: Default::default(),
             msgs_in_tipset: Default::default(),
-            sync_status: Arc::new(parking_lot::RwLock::new(SyncStatusReport::default())),
+            sync_status: Arc::new(RwLock::new(SyncStatusReport::default())),
             eth_event_handler: Arc::new(EthEventHandler::new()),
             sync_network_context,
             network_name: TEST_NET_NAME.to_owned(),