Problem: L2 network cannot filter transactions (#47)

Co-authored-by: Thomas Nguy <[email protected]>

core/bin/zksync_server/src/main.rs

@@ -44,7 +44,7 @@ struct Cli {
     /// Comma-separated list of components to launch.
     #[arg(
         long,
-        default_value = "api,tree,eth,state_keeper,housekeeper,tee_verifier_input_producer,commitment_generator,da_dispatcher"
+        default_value = "api,tree,eth,state_keeper,housekeeper,tee_verifier_input_producer,commitment_generator,da_dispatcher,deny_list"
     )]
     components: ComponentsToRun,
     /// Path to the yaml config. If set, it will be used instead of env vars.

core/bin/zksync_server/src/node_builder.rs

@@ -16,7 +16,7 @@ use zksync_default_da_clients::{
 };
 use zksync_metadata_calculator::MetadataCalculatorConfig;
 use zksync_node_api_server::{
-    tx_sender::{ApiContracts, TxSenderConfig},
+    tx_sender::{deny_list_pool_sink::DenyListPoolSink, ApiContracts, TxSenderConfig},
     web3::{state::InternalApiConfig, Namespace},
 };
 use zksync_node_framework::{
@@ -63,7 +63,7 @@ use zksync_node_framework::{
             server::{Web3ServerLayer, Web3ServerOptionalConfig},
             tree_api_client::TreeApiClientLayer,
             tx_sender::{PostgresStorageCachesConfig, TxSenderLayer},
-            tx_sink::MasterPoolSinkLayer,
+            tx_sink::{DenyListPoolSinkLayer, MasterPoolSinkLayer},
         },
     },
     service::{ZkStackService, ZkStackServiceBuilder},
@@ -279,7 +279,7 @@ impl MainNodeBuilder {
         Ok(self)
     }
 
-    fn add_tx_sender_layer(mut self) -> anyhow::Result<Self> {
+    fn add_tx_sender_layer(mut self, denylist: bool) -> anyhow::Result<Self> {
         let sk_config = try_load_config!(self.configs.state_keeper_config);
         let rpc_config = try_load_config!(self.configs.api_config).web3_json_rpc;
         let postgres_storage_caches_config = PostgresStorageCachesConfig {
@@ -289,7 +289,14 @@ impl MainNodeBuilder {
         };
 
         // On main node we always use master pool sink.
-        self.node.add_layer(MasterPoolSinkLayer);
+        if denylist {
+            let txsink_config = try_load_config!(self.configs.api_config).tx_sink;
+            self.node
+                .add_layer(DenyListPoolSinkLayer(txsink_config.denylist));
+        } else {
+            self.node.add_layer(MasterPoolSinkLayer);
+        }
+
         self.node.add_layer(TxSenderLayer::new(
             TxSenderConfig::new(
                 &sk_config,
@@ -634,6 +641,10 @@ impl MainNodeBuilder {
                         .add_storage_initialization_layer(LayerKind::Task)?
                         .add_state_keeper_layer()?;
                 }
+                Component::TxSinkDenyList => {
+                    // txsinkdenylist has higher priority than HttpApi and WsApi server, therefore the txsink will be added first.
+                    self = self.add_tx_sender_layer(true)?;
+                }
                 Component::HttpApi => {
                     self = self
                         .add_tx_sender_layer()?

core/lib/config/src/configs/api.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     fmt,
     net::SocketAddr,
     num::{NonZeroU32, NonZeroUsize},
@@ -24,6 +24,8 @@ pub struct ApiConfig {
     pub healthcheck: HealthCheckConfig,
     /// Configuration options for Merkle tree API.
     pub merkle_tree: MerkleTreeApiConfig,
+    /// Configuration options for the transactions sink.
+    pub tx_sink: TxSinkConfig,
 }
 
 /// Response size limits for specific RPC methods.
@@ -407,6 +409,21 @@ impl MerkleTreeApiConfig {
     }
 }
 
+#[derive(Debug, Deserialize, Clone, PartialEq)]
+pub struct TxSinkConfig {
+    pub deny_list: Option<String>,
+}
+
+impl TxSinkConfig {
+    pub fn deny_list(&self) -> Option<HashSet<Address>> {
+        self.deny_list.as_ref().map(|list| {
+            list.split(',')
+                .map(|element| Address::from_str(element).unwrap())
+                .collect()
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

core/lib/config/src/testonly.rs

@@ -50,6 +50,7 @@ impl Distribution<configs::ApiConfig> for EncodeDist {
             prometheus: self.sample(rng),
             healthcheck: self.sample(rng),
             merkle_tree: self.sample(rng),
+            tx_sink: self.sample(rng),
         }
     }
 }
@@ -132,6 +133,14 @@ impl Distribution<configs::api::MerkleTreeApiConfig> for EncodeDist {
     }
 }
 
+impl Distribution<configs::api::TxSinkConfig> for EncodeDist {
+    fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::api::TxSinkConfig {
+        configs::api::TxSinkConfig {
+            deny_list: self.sample(rng),
+        }
+    }
+}
+
 impl Distribution<configs::PrometheusConfig> for EncodeDist {
     fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::PrometheusConfig {
         configs::PrometheusConfig {

core/lib/env_config/src/api.rs

@@ -1,7 +1,8 @@
 use anyhow::Context as _;
 use zksync_config::configs::{
     api::{
-        ContractVerificationApiConfig, HealthCheckConfig, MerkleTreeApiConfig, Web3JsonRpcConfig,
+        ContractVerificationApiConfig, HealthCheckConfig, MerkleTreeApiConfig, TxSinkConfig,
+        Web3JsonRpcConfig,
     },
     ApiConfig, PrometheusConfig,
 };
@@ -15,6 +16,7 @@ impl FromEnv for ApiConfig {
             prometheus: PrometheusConfig::from_env().context("PrometheusConfig")?,
             healthcheck: HealthCheckConfig::from_env().context("HealthCheckConfig")?,
             merkle_tree: MerkleTreeApiConfig::from_env().context("MerkleTreeApiConfig")?,
+            tx_sink: TxSinkConfig::from_env().context("TxSinkConfig")?,
         })
     }
 }
@@ -44,6 +46,13 @@ impl FromEnv for MerkleTreeApiConfig {
     }
 }
 
+impl FromEnv for TxSinkConfig {
+    /// Loads configuration from env variables.
+    fn from_env() -> anyhow::Result<Self> {
+        envy_load("tx_sink", "TX_SINK_")
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::num::{NonZeroU32, NonZeroUsize};
@@ -112,6 +121,9 @@ mod tests {
                 hard_time_limit_ms: Some(2_000),
             },
             merkle_tree: MerkleTreeApiConfig { port: 8082 },
+            tx_sink: TxSinkConfig {
+                deny_list: vec![addr("0x1234567890abcdef")],
+            },
         }
     }
 
@@ -158,6 +170,7 @@ mod tests {
             API_HEALTHCHECK_SLOW_TIME_LIMIT_MS=250
             API_HEALTHCHECK_HARD_TIME_LIMIT_MS=2000
             API_MERKLE_TREE_PORT=8082
+            API_TX_SINK_DENY_LIST="0x1234567890abcdef"
         "#;
         lock.set_env(config);
 

core/lib/protobuf_config/src/api.rs

@@ -17,6 +17,7 @@ impl ProtoRepr for proto::Api {
             prometheus: read_required_repr(&self.prometheus).context("prometheus")?,
             healthcheck: read_required_repr(&self.healthcheck).context("healthcheck")?,
             merkle_tree: read_required_repr(&self.merkle_tree).context("merkle_tree")?,
+            tx_sink: read_required_repr(&self.tx_sink).context("tx_sink")?,
         })
     }
 
@@ -26,6 +27,7 @@ impl ProtoRepr for proto::Api {
             prometheus: Some(ProtoRepr::build(&this.prometheus)),
             healthcheck: Some(ProtoRepr::build(&this.healthcheck)),
             merkle_tree: Some(ProtoRepr::build(&this.merkle_tree)),
+            tx_sink: Some(ProtoRepr::build(&this.tx_sink)),
         }
     }
 }
@@ -271,3 +273,18 @@ impl ProtoRepr for proto::MerkleTreeApi {
         }
     }
 }
+
+impl ProtoRepr for proto::TxSink {
+    type Type = api::TxSinkConfig;
+    fn read(&self) -> anyhow::Result<Self::Type> {
+        Ok(Self::Type {
+            deny_list: self.deny_list.clone(),
+        })
+    }
+
+    fn build(this: &Self::Type) -> Self {
+        Self {
+            deny_list: this.deny_list.clone(),
+        }
+    }
+}

core/lib/protobuf_config/src/proto/config/api.proto

@@ -42,11 +42,10 @@ message Web3JsonRpc {
   repeated MaxResponseSizeOverride max_response_body_size_overrides = 31;
   repeated string api_namespaces = 32; // Optional, if empty all namespaces are available
   optional bool extended_api_tracing = 33; // optional, default false
-  reserved 15; reserved "l1_to_l2_transactions_compatibility_mode";
+  reserved 15;
+  reserved "l1_to_l2_transactions_compatibility_mode";
 }
 
-
-
 message HealthCheck {
   optional uint32 port = 1; // required; u16
   optional uint64 slow_time_limit_ms = 2; // optional; ms
@@ -57,9 +56,14 @@ message MerkleTreeApi {
   optional uint32 port = 1; // required; u16
 }
 
+message TxSink {
+  optional string deny_list = 1; // optional
+}
+
 message Api {
   optional Web3JsonRpc web3_json_rpc = 1; // required
   optional utils.Prometheus prometheus = 3; // required
   optional HealthCheck healthcheck = 4; // required
   optional MerkleTreeApi merkle_tree = 5; // required
+  optional TxSink tx_sink = 6; // optional
 }

core/lib/protobuf_config/src/proto/config/general.proto

@@ -2,25 +2,25 @@ syntax = "proto3";
 
 package zksync.config.general;
 
-import "zksync/config/prover.proto";
 import "zksync/config/api.proto";
+import "zksync/config/base_token_adjuster.proto";
 import "zksync/config/chain.proto";
+import "zksync/config/circuit_breaker.proto";
+import "zksync/config/commitment_generator.proto";
 import "zksync/config/contract_verifier.proto";
+import "zksync/config/da_dispatcher.proto";
 import "zksync/config/database.proto";
-import "zksync/config/circuit_breaker.proto";
 import "zksync/config/eth_sender.proto";
+import "zksync/config/external_price_api_client.proto";
 import "zksync/config/house_keeper.proto";
+import "zksync/config/object_store.proto";
 import "zksync/config/observability.proto";
+import "zksync/config/prover.proto";
+import "zksync/config/pruning.proto";
+import "zksync/config/snapshot_recovery.proto";
 import "zksync/config/snapshots_creator.proto";
 import "zksync/config/utils.proto";
-import "zksync/config/da_dispatcher.proto";
 import "zksync/config/vm_runner.proto";
-import "zksync/config/commitment_generator.proto";
-import "zksync/config/snapshot_recovery.proto";
-import "zksync/config/pruning.proto";
-import "zksync/config/object_store.proto";
-import "zksync/config/base_token_adjuster.proto";
-import "zksync/config/external_price_api_client.proto";
 
 message GeneralConfig {
   optional config.database.Postgres postgres = 1;

core/lib/protobuf_config/src/tests.rs

@@ -11,6 +11,7 @@ fn test_encoding() {
     test_encode_all_formats::<ReprConv<proto::api::Web3JsonRpc>>(rng);
     test_encode_all_formats::<ReprConv<proto::api::HealthCheck>>(rng);
     test_encode_all_formats::<ReprConv<proto::api::MerkleTreeApi>>(rng);
+    test_encode_all_formats::<ReprConv<proto::api::TxSink>>(rng);
     test_encode_all_formats::<ReprConv<proto::api::Api>>(rng);
     test_encode_all_formats::<ReprConv<proto::utils::Prometheus>>(rng);
     test_encode_all_formats::<ReprConv<proto::chain::StateKeeper>>(rng);

core/lib/zksync_core_leftovers/src/lib.rs

@@ -62,6 +62,8 @@ pub enum Component {
     BaseTokenRatioPersister,
     /// VM runner-based component that saves VM execution data for basic witness generation.
     VmRunnerBwip,
+    /// Component for filtering L2 transactions by denylist
+    TxSinkDenyList,
 }
 
 #[derive(Debug)]
@@ -106,6 +108,7 @@ impl FromStr for Components {
                 Ok(Components(vec![Component::BaseTokenRatioPersister]))
             }
             "vm_runner_bwip" => Ok(Components(vec![Component::VmRunnerBwip])),
+            "deny_list" => Ok(Components(vec![Component::TxSinkDenyList])),
             other => Err(format!("{} is not a valid component name", other)),
         }
     }

core/node/api_server/src/tx_sender/deny_list_pool_sink.rs

@@ -0,0 +1,39 @@
+use std::collections::HashSet;
+
+use zksync_dal::transactions_dal::L2TxSubmissionResult;
+use zksync_types::{fee::TransactionExecutionMetrics, l2::L2Tx, Address};
+
+use super::{tx_sink::TxSink, SubmitTxError};
+use crate::tx_sender::MasterPoolSink;
+
+/// Wrapper for the master DB pool that allows to submit transactions to the mempool.
+#[derive(Debug)]
+pub struct DenyListPoolSink {
+    deny_list: HashSet<Address>,
+    master_pool_sync: MasterPoolSink,
+}
+
+impl DenyListPoolSink {
+    pub fn new(master_pool_sync: MasterPoolSink, deny_list: HashSet<Address>) -> Self {
+        Self {
+            master_pool_sync,
+            deny_list,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl TxSink for DenyListPoolSink {
+    async fn submit_tx(
+        &self,
+        tx: &L2Tx,
+        execution_metrics: TransactionExecutionMetrics,
+    ) -> Result<L2TxSubmissionResult, SubmitTxError> {
+        let address_and_nonce = (tx.initiator_account(), tx.nonce());
+        if self.deny_list.contains(&address_and_nonce.0) {
+            return Err(SubmitTxError::SenderInDenyList(tx.initiator_account()));
+        }
+
+        self.master_pool_sync.submit_tx(tx, execution_metrics).await
+    }
+}

core/node/api_server/src/tx_sender/mod.rs

@@ -46,6 +46,7 @@ use crate::{
     tx_sender::result::ApiCallResult,
 };
 
+pub mod deny_list_pool_sink;
 pub mod master_pool_sink;
 pub mod proxy;
 mod result;

core/node/api_server/src/tx_sender/result.rs

@@ -1,6 +1,6 @@
 use thiserror::Error;
 use zksync_multivm::interface::{ExecutionResult, VmExecutionResultAndLogs};
-use zksync_types::{l2::error::TxCheckError, U256};
+use zksync_types::{l2::error::TxCheckError, Address, U256};
 use zksync_web3_decl::error::EnrichedClientError;
 
 use crate::execution_sandbox::{SandboxExecutionError, ValidationError};
@@ -75,6 +75,8 @@ pub enum SubmitTxError {
     /// Catch-all internal error (e.g., database error) that should not be exposed to the caller.
     #[error("internal error")]
     Internal(#[from] anyhow::Error),
+    #[error("sender address {0} is in deny list")]
+    SenderInDenyList(Address),
 }
 
 impl SubmitTxError {
@@ -108,6 +110,7 @@ impl SubmitTxError {
             Self::ProxyError(_) => "proxy-error",
             Self::FailedToPublishCompressedBytecodes => "failed-to-publish-compressed-bytecodes",
             Self::Internal(_) => "internal",
+            Self::SenderInDenyList(_) => "sender-in-deny-list",
         }
     }