v2.1: Clean up rpc filter verification (backport of #5305) (#5306)

Clean up rpc filter verification (#5305)

Co-authored-by: Kirill Fomichev <[email protected]>
(cherry picked from commit 22a98ce)

Co-authored-by: Will Hickey <[email protected]>

Author: mergify[bot]

rpc/src/rpc.rs

@@ -2412,7 +2412,7 @@ impl JsonRpcRequestProcessor {
     }
 }
 
-fn optimize_filters(filters: &mut [RpcFilterType]) {
+pub(crate) fn optimize_filters(filters: &mut [RpcFilterType]) {
     filters.iter_mut().for_each(|filter_type| {
         if let RpcFilterType::Memcmp(compare) = filter_type {
             if let Err(err) = compare.convert_to_raw_bytes() {
@@ -2443,6 +2443,18 @@ fn verify_transaction(
     Ok(())
 }
 
+pub(crate) fn verify_filters(filters: &[RpcFilterType]) -> Result<()> {
+    if filters.len() > MAX_GET_PROGRAM_ACCOUNT_FILTERS {
+        return Err(Error::invalid_params(format!(
+            "Too many filters provided; max {MAX_GET_PROGRAM_ACCOUNT_FILTERS}"
+        )));
+    }
+    for filter in filters {
+        verify_filter(filter)?;
+    }
+    Ok(())
+}
+
 fn verify_filter(input: &RpcFilterType) -> Result<()> {
     input
         .verify()
@@ -3391,14 +3403,7 @@ pub mod rpc_accounts_scan {
                 } else {
                     (None, vec![], false, true)
                 };
-                if filters.len() > MAX_GET_PROGRAM_ACCOUNT_FILTERS {
-                    return Err(Error::invalid_params(format!(
-                        "Too many filters provided; max {MAX_GET_PROGRAM_ACCOUNT_FILTERS}"
-                    )));
-                }
-                for filter in &filters {
-                    verify_filter(filter)?;
-                }
+                verify_filters(&filters)?;
                 meta.get_program_accounts(program_id, config, filters, with_context, sort_results)
                     .await
             }

rpc/src/rpc_pubsub.rs

@@ -3,7 +3,7 @@
 use crate::{rpc_pubsub_service, rpc_subscriptions::RpcSubscriptions};
 use {
     crate::{
-        rpc::check_is_at_least_confirmed,
+        rpc::{check_is_at_least_confirmed, optimize_filters, verify_filters},
         rpc_pubsub_service::PubSubConfig,
         rpc_subscription_tracker::{
             AccountSubscriptionParams, BlockSubscriptionKind, BlockSubscriptionParams,
@@ -450,9 +450,18 @@ impl RpcSolPubSubInternal for RpcSolPubSubImpl {
         config: Option<RpcProgramAccountsConfig>,
     ) -> Result<SubscriptionId> {
         let config = config.unwrap_or_default();
+        let mut filters = config.filters.unwrap_or_default();
+        if let Err(error) = verify_filters(&filters) {
+            return Err(Error {
+                code: ErrorCode::InvalidParams,
+                message: error.to_string(),
+                data: None,
+            });
+        }
+        optimize_filters(&mut filters);
         let params = ProgramSubscriptionParams {
             pubkey: param::<Pubkey>(&pubkey_str, "pubkey")?,
-            filters: config.filters.unwrap_or_default(),
+            filters,
             encoding: config
                 .account_config
                 .encoding

rpc/src/rpc_pubsub_service.rs

@@ -375,7 +375,10 @@ async fn handle_connection(
         protocol: None,
     };
     server.send_response(&accept).await?;
-    let (mut sender, mut receiver) = server.into_builder().finish();
+    let mut builder = server.into_builder();
+    builder.set_max_message_size(4_096);
+    builder.set_max_frame_size(4_096);
+    let (mut sender, mut receiver) = builder.finish();
 
     let mut broadcast_receiver = subscription_control.broadcast_receiver();
     let mut data = Vec::new();